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;