From f3dc687a3ec84a38d95d1a22855b9b8b7a2e217c Mon Sep 17 00:00:00 2001 From: cloudwebrtc Date: Thu, 29 Oct 2026 14:05:19 +0800 Subject: [PATCH] Apply backpressure to native data stream sends The native outbound pump drained every packet batch from the Rust core straight into an unbounded Dart future chain (`_sendChain`). Nothing waited for the data channel to drain, because the old `waitForBufferStatusLow` flow control was lost in the move to the core. The core's polled `OutgoingPacketQueue` acks a send as soon as the packets are buffered, so `sendText`/`sendBytes`/`sendFile` resolved before any data was on the wire. Transport errors were only logged, so callers saw success for data that was never delivered. Fix: - The pump now sends one packet at a time, awaiting `engine.waitForBufferStatusLow(Reliability.reliable)` before each send, and pulls the next batch only once the current one is out. The backlog stays queued in the core instead of piling up in Dart. - One-shot sends get a Dart-generated stream id that is registered before the core call. The pump completes it when the stream's last packet (its trailer, or an inline header) reaches the engine, and fails it with `DataStreamError(AbnormalEnd)` if a packet fails to send. `send*` now resolves only when the data is out, and send failures reach the caller. Measured p1 -> p2 through a local SFU (64 MB `sendBytes`, uncompressed, both clients in one macOS process): before after sendBytes resolves 532 ms 2815 ms bytes on wire by then ~0 ~64 MB receiver handler invoked 29.7 s 3.7 s receiver complete 36.4 s 4.2 s throughput 1.76 MB/s 15.3 MB/s payload integrity OK OK Tests: two regression tests in data_stream_v2_test.dart, which fail on the previous implementation. One checks that a send pauses while the channel reports a full buffer and resolves only after its trailer is sent. The other checks that a transport failure fails the send. MockDataChannel gains `fakeBufferedAmount` to simulate an undrained channel. Not addressed here: peak sender memory. The core's outgoing queue is unbounded, so the whole payload is still buffered on the Rust side, and the generated livekit_ffi bindings leak lowered/lifted buffers. Both need fixes in rust-sdks. Co-Authored-By: Claude Opus 5.5 --- lib/src/data_stream/data_streams_native.dart | 98 ++++++++++++++++---- test/core/data_stream_v2_test.dart | 46 +++++++++ test/mock/datachannel_mock.dart | 8 +- 3 files changed, 130 insertions(+), 22 deletions(-) diff --git a/lib/src/data_stream/data_streams_native.dart b/lib/src/data_stream/data_streams_native.dart index b3ecc7495..ff0f293c2 100644 --- a/lib/src/data_stream/data_streams_native.dart +++ b/lib/src/data_stream/data_streams_native.dart @@ -80,9 +80,11 @@ class NativeDataStreams implements DataStreams { final Map _textStreamHandlers = {}; final Map _byteStreamHandlers = {}; - /// Serializes outbound sends so packet order survives the hop from the pump into the engine's - /// async send. - Future _sendChain = Future.value(); + /// One per in-flight `send*` call, keyed by its stream id, completed by the pump once the stream's + /// last packet has gone to the engine. The core acknowledges a send as soon as its packets are + /// queued, so this is what holds `sendText`/`sendBytes`/`sendFile` until the data is actually out, + /// and lets a transport failure reach the caller. + final Map> _pendingSends = {}; bool _disposed = false; @@ -112,14 +114,17 @@ class NativeDataStreams implements DataStreams { // own byte stream referenced by `attachedStreamIds` in the text header. final attachments = options?.attachments ?? const []; final attachmentIds = [for (var i = 0; i < attachments.length; i++) const Uuid().v4()]; + final id = const Uuid().v4(); - final info = await mappingFfiErrors( + final info = await _sendTracked( + id, () => _outgoing.sendText( text: text, options: ffi.StreamTextOptions( topic: options?.topic ?? '', attributes: options?.attributes ?? const {}, destinationIdentities: options?.destinationIdentities ?? const [], + id: id, attachedStreamIds: attachmentIds, compress: options?.compress, ), @@ -147,13 +152,16 @@ class NativeDataStreams implements DataStreams { @override Future sendBytes(List bytes, SendBytesOptions? options) async { - final info = await mappingFfiErrors( + final id = const Uuid().v4(); + final info = await _sendTracked( + id, () => _outgoing.sendBytes( data: Uint8List.fromList(bytes), options: ffi.StreamByteOptions( topic: options?.topic ?? '', attributes: options?.attributes ?? const {}, destinationIdentities: options?.destinationIdentities ?? const [], + id: id, name: options?.name, mimeType: options?.mimeType, compress: options?.compress, @@ -174,7 +182,8 @@ class NativeDataStreams implements DataStreams { } Future _sendFileWithId(String id, File file, SendFileOptions options) async { - await mappingFfiErrors( + await _sendTracked( + id, () => _outgoing.sendFile( // The core streams the file from disk rather than buffering it. path: file.path, @@ -243,8 +252,32 @@ class NativeDataStreams implements DataStreams { ); } + /// Runs a one-shot core send for stream [id] and resolves once its last packet has been handed to + /// the engine, rather than when the core has merely queued it. + /// + /// Registered before the core call: the pump can put the whole stream out before that call + /// returns. + Future _sendTracked(String id, Future Function() send) async { + final sent = Completer(); + // The pump may fail this before we get to await it. + sent.future.ignore(); + _pendingSends[id] = sent; + try { + final result = await mappingFfiErrors(send); + await sent.future; + return result; + } finally { + _pendingSends.remove(id); + } + } + /// Drains outbound packets from the core and puts them on the wire, in order. /// + /// Each packet waits for the reliable data channel's buffer to drain below its low threshold, and + /// the next batch is pulled only once this one is out. The core acknowledges sends as soon as it + /// has queued them, so this is the only flow control on the path: without it a large send is + /// drained into Dart all at once and pushed at the channel faster than it can empty. + /// /// The pump owns the queue's lifetime and is the only thing that may dispose it — freeing it /// while a `nextPackets` is in flight is a use-after-free. [dispose] wakes us by closing the /// queue rather than releasing it. @@ -254,32 +287,57 @@ class NativeDataStreams implements DataStreams { final batch = await _outgoingPackets.nextPackets(); if (batch == null) break; // closed or shutting down for (final encoded in batch) { - _enqueueSend(encoded); + await _sendOutbound(lk_models.DataPacket.fromBuffer(encoded)); } } } catch (e) { logger.warning('[DataStreams] outgoing pump failed: $e'); } finally { _outgoingPackets.dispose(); + for (final sent in _pendingSends.values.toList()) { + if (!sent.isCompleted) { + sent.completeError( + DataStreamError( + reason: DataStreamErrorReason.AbnormalEnd, + message: 'Data streams were shut down before the stream was sent', + ), + ); + } + } } } - void _enqueueSend(Uint8List encoded) { - _sendChain = _sendChain.then((_) async { + Future _sendOutbound(lk_models.DataPacket packet) async { + final streamId = switch (packet.whichValue()) { + lk_models.DataPacket_Value.streamHeader => packet.streamHeader.streamId, + lk_models.DataPacket_Value.streamChunk => packet.streamChunk.streamId, + lk_models.DataPacket_Value.streamTrailer => packet.streamTrailer.streamId, + _ => null, + }; + // A stream ends with its trailer, or with its only packet when the content was inlined. + final isLast = packet.hasStreamTrailer() || (packet.hasStreamHeader() && packet.streamHeader.hasInlineContent()); + try { final room = _room.target; - if (room == null || _disposed) return; - try { - // Back through the engine rather than the data channel directly, so E2EE wrapping, - // reliable sequencing and resume-resend all still apply. - await room.engine.sendDataPacket( - lk_models.DataPacket.fromBuffer(encoded), - reliability: Reliability.reliable, + if (room == null || _disposed) { + throw StateError('the room was disposed'); + } + await room.engine.waitForBufferStatusLow(Reliability.reliable); + // Back through the engine rather than the data channel directly, so E2EE wrapping, + // reliable sequencing and resume-resend all still apply. + await room.engine.sendDataPacket(packet, reliability: Reliability.reliable); + if (isLast) _pendingSends[streamId]?.complete(); + } catch (e) { + logger.warning('[DataStreams] failed to send outbound packet: $e'); + final sent = _pendingSends[streamId]; + if (sent != null && !sent.isCompleted) { + sent.completeError( + DataStreamError( + reason: DataStreamErrorReason.AbnormalEnd, + message: 'Failed to send data stream packet: $e', + ), ); - } catch (e) { - // The core acknowledges sends unconditionally, so there is nobody to propagate this to. - logger.warning('[DataStreams] failed to send outbound packet: $e'); } - }); + } } // MARK: - Receive diff --git a/test/core/data_stream_v2_test.dart b/test/core/data_stream_v2_test.dart index b9f66ddbe..8611b8a00 100644 --- a/test/core/data_stream_v2_test.dart +++ b/test/core/data_stream_v2_test.dart @@ -185,6 +185,52 @@ void main() { expect(info.name, equals('unknown')); expect(info.mimeType, equals('application/octet-stream')); }); + + test('a send waits for the data channel to drain and resolves once the stream is out', () async { + await container.simulateRemoteParticipantJoin('legacy', clientProtocol: 0); + container.capturedDataPackets.clear(); + final dc = mockDataChannels.firstWhere((dc) => dc.label == '_reliable' && dc.onBufferedAmountLow != null); + dc.fakeBufferedAmount = dc.bufferedAmountLowThreshold! + 1; + + var resolved = false; + final send = room.localParticipant! + .sendBytes( + Uint8List(64 * 1024), + options: SendBytesOptions(topic: 'files', destinationIdentities: ['legacy']), + ) + .then((info) { + resolved = true; + return info; + }); + + await Future.delayed(const Duration(milliseconds: 200)); + expect(streamPackets(), hasLength(1), reason: 'only the header goes out before the buffer reports full'); + expect(resolved, isFalse, reason: 'the core has queued everything, but it is not on the wire yet'); + + dc.fakeBufferedAmount = null; + dc.onBufferedAmountLow!(0); + final info = await send; + + final packets = streamPackets(); + expect(packets.length, greaterThan(2)); + expect(packets.last.hasStreamTrailer(), isTrue); + expect(packets.last.streamTrailer.streamId, info.id); + }); + + test('a transport failure fails the send', () async { + await container.simulateRemoteParticipantJoin('alice', clientProtocol: 2, capabilities: _v2WithCompression); + final dc = mockDataChannels.firstWhere((dc) => dc.label == '_reliable' && dc.onBufferedAmountLow != null); + final original = dc.onMessageSend; + dc.onMessageSend = (_) => throw StateError('data channel closed'); + + await expectLater( + room.localParticipant!.sendText('hello', options: SendTextOptions(topic: 'chat')), + throwsA(isA().having((e) => e.reason, 'reason', DataStreamErrorReason.AbnormalEnd)), + ); + + dc.onMessageSend = original; + await room.localParticipant!.sendText('hello again', options: SendTextOptions(topic: 'chat')); + }); }); group('receive side', () { diff --git a/test/mock/datachannel_mock.dart b/test/mock/datachannel_mock.dart index e63121f83..e015b8630 100644 --- a/test/mock/datachannel_mock.dart +++ b/test/mock/datachannel_mock.dart @@ -50,11 +50,15 @@ class MockDataChannel extends RTCDataChannel { // ignore: overridden_fields int? bufferedAmountLowThreshold = 0; + /// Overrides the reported buffered amount, to simulate a channel that hasn't drained. Defaults to + /// the low threshold, i.e. never above it. + int? fakeBufferedAmount; + @override - int? get bufferedAmount => bufferedAmountLowThreshold; + int? get bufferedAmount => fakeBufferedAmount ?? bufferedAmountLowThreshold; @override - Future getBufferedAmount() => Future.value(bufferedAmountLowThreshold); + Future getBufferedAmount() => Future.value(bufferedAmount); @override int? get id => _id;