diff --git a/.changes/data-channel-buffer-wait b/.changes/data-channel-buffer-wait new file mode 100644 index 000000000..d0388efaf --- /dev/null +++ b/.changes/data-channel-buffer-wait @@ -0,0 +1 @@ +patch type="fixed" "Fix data stream writes that could hang after a full reconnect, carried on after the room disconnected, or leaked a listener while waiting for the send buffer to drain. Writes now also wait for a reconnect to finish" diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 118cb247b..af32afa34 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -41,6 +41,7 @@ import '../publication/local.dart'; import '../support/disposable.dart'; import '../support/platform.dart' show lkPlatformIsTest, lkPlatformIs, PlatformType; import '../support/region_url_provider.dart'; +import '../support/reusable_completer.dart'; import '../support/websocket.dart'; import '../track/local/local.dart'; import '../track/local/video.dart'; @@ -178,6 +179,15 @@ class Engine extends Disposable with EventsEmittable { Reliability.lossy: false, }; + // Completed whenever something a buffer wait depends on changes: a buffer status, the reconnect + // state or closure. A wait checks its condition again each time it wakes. + final ReusableCompleter _sendStateChanged = ReusableCompleter(); + + void _setBufferStatus(Reliability kind, bool isLow) { + _dcBufferStatus[kind] = isLow; + _sendStateChanged.complete(); + } + List? _enabledPublishCodecs; List? get enabledPublishCodecs => _enabledPublishCodecs; @@ -197,7 +207,12 @@ class Engine extends Disposable with EventsEmittable { maxPacketCount: 1000, // max 1000 packets ); final TTLMap _reliableReceivedState = TTLMap(30000); - bool _isReconnecting = false; + bool _reconnecting = false; + bool get _isReconnecting => _reconnecting; + set _isReconnecting(bool value) { + _reconnecting = value; + _sendStateChanged.complete(); + } Completer? _publisherConnectionCompleter; @@ -422,35 +437,21 @@ class Engine extends Disposable with EventsEmittable { return null; } - Future waitForBufferStatusLow(Reliability kind) async { - // A wait started after closure cannot observe the closing event. - if (_isClosed || isDisposed) { - throw UnexpectedConnectionState('Engine disconnected'); - } - - final Completer completer = Completer(); - - if (isBufferStatusLow(kind) == true) { - completer.complete(); - } else { - onClosing() { - if (!completer.isCompleted) { - completer.completeError(UnexpectedConnectionState('Engine disconnected')); - } + // A full reconnect creates its empty channel before it finishes, and a send made in between + // is parked for a resume that never comes, so the wait also holds until the reconnect is over. + Future waitForBufferStatusLow(Reliability kind) => + _waitUntil(() => !_isReconnecting && (_dcBufferStatus[kind]! || isBufferStatusLow(kind) == true)); + + // Resolves once [isReady] holds, checking it again each time [_sendStateChanged] fires. Fails + // as soon as the engine closes, even when [isReady] already holds. + Future _waitUntil(bool Function() isReady) async { + while (true) { + if (_isClosed || isDisposed) { + throw UnexpectedConnectionState('Engine disconnected'); } - - events.once((e) => onClosing()); - - while (!completer.isCompleted && !_dcBufferStatus[kind]!) { - await Future.delayed(const Duration(milliseconds: 10)); - } - if (completer.isCompleted) { - return; - } - completer.complete(); + if (isReady()) return; + await _sendStateChanged.future; } - - return completer.future; } Future _resendReliableMessagesForResume(int lastMessageSeq) async { @@ -569,7 +570,7 @@ class Engine extends Disposable with EventsEmittable { logger.fine('sendDataPacket(label:${channel.label}, sequence:${packet.sequence})'); await channel.send(message); - _dcBufferStatus[reliability] = await channel.getBufferedAmount() <= channel.bufferedAmountLowThreshold!; + _setBufferStatus(reliability, await channel.getBufferedAmount() <= channel.bufferedAmountLowThreshold!); // Align buffer with WebRTC buffer for reliable packets if (reliability == Reliability.reliable) { @@ -882,8 +883,10 @@ class Engine extends Disposable with EventsEmittable { ); // _onDCStateUpdated(Reliability.lossy, state) _lossyDCPub?.bufferedAmountLowThreshold = 2 * 1024 * 1024; + // A new channel starts empty, whatever the one it replaces had buffered. + _setBufferStatus(Reliability.lossy, true); _lossyDCPub?.onBufferedAmountLow = (_) { - _dcBufferStatus[Reliability.lossy] = (_lossyDCPub!.bufferedAmount! <= _lossyDCPub!.bufferedAmountLowThreshold!); + _setBufferStatus(Reliability.lossy, _lossyDCPub!.bufferedAmount! <= _lossyDCPub!.bufferedAmountLowThreshold!); }; } catch (err) { logger.severe('[$objectId] createDataChannel() did throw $err'); @@ -905,9 +908,12 @@ class Engine extends Disposable with EventsEmittable { ), ); _reliableDCPub?.bufferedAmountLowThreshold = 2 * 1024 * 1024; + _setBufferStatus(Reliability.reliable, true); _reliableDCPub?.onBufferedAmountLow = (_) { - _dcBufferStatus[Reliability.reliable] = - (_reliableDCPub!.bufferedAmount! <= _reliableDCPub!.bufferedAmountLowThreshold!); + _setBufferStatus( + Reliability.reliable, + _reliableDCPub!.bufferedAmount! <= _reliableDCPub!.bufferedAmountLowThreshold!, + ); }; } catch (err) { logger.severe('[$objectId] createDataChannel() did throw $err'); @@ -1695,6 +1701,7 @@ class Engine extends Disposable with EventsEmittable { void markClosed() { if (_isClosed) return; _isClosed = true; + _sendStateChanged.complete(); events.emit(EngineClosingEvent()); } diff --git a/test/core/buffer_status_wait_test.dart b/test/core/buffer_status_wait_test.dart new file mode 100644 index 000000000..8defae964 --- /dev/null +++ b/test/core/buffer_status_wait_test.dart @@ -0,0 +1,155 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +import 'package:flutter_test/flutter_test.dart'; + +import 'package:livekit_client/livekit_client.dart'; +import 'package:livekit_client/src/proto/livekit_models.pb.dart' as lk_models; +import '../mock/e2e_container.dart'; +import '../mock/peerconnection_mock.dart'; +import 'signal_client_test.dart' show joinResponse, offerResponse; + +const _full = 10 * 1024 * 1024; + +void main() { + late E2EContainer container; + + setUp(() async { + resetMockDataChannels(); + container = E2EContainer(); + await container.connectRoom(); + }); + + tearDown(() async { + await container.dispose(); + }); + + void setBufferedAmount(int? amount) { + for (final dc in mockDataChannels) { + dc.fakeBufferedAmount = amount; + } + } + + /// Starts a full reconnect and answers its join, leaving it to wait for the subscriber offer. + /// The new, empty publisher channels exist from here on, but the reconnect has not finished. + Future startFullReconnect() async { + final ws = container.wsConnector; + final previousHandlers = ws.handlers; + container.engine.fullReconnectOnNext = true; + ws.onDispose(); + for (var i = 0; i < 200 && identical(ws.handlers, previousHandlers); i++) { + await Future.delayed(const Duration(milliseconds: 10)); + } + await Future.delayed(const Duration(milliseconds: 1)); + ws.onData(joinResponse.writeToBuffer()); + } + + void finishFullReconnect() => container.wsConnector.onData(offerResponse.writeToBuffer()); + + /// Fills the reliable channel and sends once, so the engine's cached status also says it is full. + Future fillReliableChannel() async { + // The mock loops published data back into this room as if a remote sent it, which the room + // rejects. These tests only need the send to refresh the engine's cached status. + for (final dc in mockDataChannels) { + dc.onMessageSend = null; + } + setBufferedAmount(_full); + await container.room.localParticipant!.publishData([1], reliable: true); + expect(container.engine.isBufferStatusLow(Reliability.reliable), isFalse); + } + + test('a wait ends when a full reconnect replaces the full channel', () async { + await fillReliableChannel(); + final wait = container.engine.waitForBufferStatusLow(Reliability.reliable); + + // The replaced channel never reports a drain, only the new one is empty. + await startFullReconnect(); + finishFullReconnect(); + + await wait.timeout(const Duration(seconds: 1)); + }); + + test('a wait wakes as soon as the channel reports a drain', () async { + await fillReliableChannel(); + var done = false; + final wait = container.engine.waitForBufferStatusLow(Reliability.reliable).whenComplete(() => done = true); + await pumpEventQueue(); + expect(done, isFalse); + + setBufferedAmount(null); + for (final dc in mockDataChannels) { + dc.onBufferedAmountLow?.call(0); + } + + // Only zero-length timers run here, so a wait that polls on a timer would not have woken yet. + await pumpEventQueue(); + expect(done, isTrue); + await wait; + }); + + test('a wait does not leave an engine listener behind', () async { + final before = container.engine.events.listeners.length; + await fillReliableChannel(); + final wait = container.engine.waitForBufferStatusLow(Reliability.reliable); + + setBufferedAmount(null); + for (final dc in mockDataChannels) { + dc.onBufferedAmountLow?.call(0); + } + await wait.timeout(const Duration(seconds: 1)); + + expect(container.engine.events.listeners.length, before); + }); + + test('a wait still fails when the engine closes', () async { + await fillReliableChannel(); + final wait = container.engine.waitForBufferStatusLow(Reliability.reliable); + final outcome = wait.then((_) => 'resolved', onError: (Object e) => e); + + container.engine.markClosed(); + + expect(await outcome.timeout(const Duration(seconds: 2)), isNot('resolved')); + }); + + test('a wait holds until a full reconnect finishes', () async { + final engine = container.engine; + await fillReliableChannel(); + + final marker = lk_models.DataPacket( + kind: lk_models.DataPacket_Kind.RELIABLE, + user: lk_models.UserPacket(payload: [42]), + ); + var markerSent = false; + onMockReliableChannelCreated = (dc) => dc.onMessageSend = (message) { + final packet = lk_models.DataPacket.fromBuffer(message.binary); + if (packet.hasUser() && packet.user.payload.length == 1 && packet.user.payload.first == 42) { + markerSent = true; + } + }; + + // What a stream writer does with each chunk. + final write = engine + .waitForBufferStatusLow(Reliability.reliable) + .then((_) => engine.sendDataPacket(marker, reliability: Reliability.reliable)); + + // The new channel exists and is empty, but the reconnect waits for the subscriber transport. + await startFullReconnect(); + await Future.delayed(const Duration(milliseconds: 300)); + expect(markerSent, isFalse, reason: 'the write went out before the reconnect finished'); + + finishFullReconnect(); + await write.timeout(const Duration(seconds: 2)); + expect(markerSent, isTrue, reason: 'the write never reached the new channel'); + }); +} 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; diff --git a/test/mock/peerconnection_mock.dart b/test/mock/peerconnection_mock.dart index 65f3e41d4..5e31ad29e 100644 --- a/test/mock/peerconnection_mock.dart +++ b/test/mock/peerconnection_mock.dart @@ -35,12 +35,17 @@ MockDataChannel? findMockDataChannelByLabel(String label, {bool requireListener /// publisher↔subscriber loopback. List get mockDataChannels => _dataChannels; +/// Called with every reliable channel the SDK creates, so a test can watch the +/// channel a full reconnect creates from the moment it exists. +void Function(MockDataChannel dc)? onMockReliableChannelCreated; + /// Reset the global mock-channel list. Tests that create a fresh `E2EContainer` /// after an earlier one finished must call this before re-connecting, otherwise /// the publisher↔subscriber pairing (which only fires when the list reaches /// length 2) will not happen on the new channels. void resetMockDataChannels() { _dataChannels.clear(); + onMockReliableChannelCreated = null; } class MockPeerConnection extends RTCPeerConnection { @@ -239,6 +244,7 @@ a=rtpmap:32 MPV/90000 final dc = MockDataChannel(dataChannelDict.id, label); if (label == '_reliable') { _dataChannels.add(dc); + onMockReliableChannelCreated?.call(dc); } return dc; }