diff --git a/.changes/data-channel-buffer-wait b/.changes/data-channel-buffer-wait new file mode 100644 index 000000000..aab8dd362 --- /dev/null +++ b/.changes/data-channel-buffer-wait @@ -0,0 +1 @@ +patch type="fixed" "Fix data channel sends that could hang after a full reconnect, keep waiting after the engine closed, or leak a listener while waiting for the send buffer to drain" diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 198775eeb..57ce8ed4b 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -401,34 +401,14 @@ class Engine extends Disposable with EventsEmittable { } Future waitForBufferStatusLow(Reliability kind) async { - // A wait started after closure cannot observe the closing event. + // The cached status only changes on a send or on the channel's low-buffer event. A channel + // that a full reconnect replaced fires neither, so the live channel is checked as well. + while (!_isClosed && !isDisposed && !_dcBufferStatus[kind]! && isBufferStatusLow(kind) != true) { + await Future.delayed(const Duration(milliseconds: 10)); + } if (_isClosed || isDisposed) { throw 'Engine disconnected'; } - - final Completer completer = Completer(); - - if (isBufferStatusLow(kind) == true) { - completer.complete(); - } else { - onClosing() { - if (!completer.isCompleted) { - completer.completeError('Engine disconnected'); - } - } - - events.once((e) => onClosing()); - - while (!completer.isCompleted && !_dcBufferStatus[kind]!) { - await Future.delayed(const Duration(milliseconds: 10)); - } - if (completer.isCompleted) { - return; - } - completer.complete(); - } - - return completer.future; } Future _resendReliableMessagesForResume(int lastMessageSeq) async { @@ -860,6 +840,8 @@ 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. + _dcBufferStatus[Reliability.lossy] = true; _lossyDCPub?.onBufferedAmountLow = (_) { _dcBufferStatus[Reliability.lossy] = (_lossyDCPub!.bufferedAmount! <= _lossyDCPub!.bufferedAmountLowThreshold!); }; @@ -883,6 +865,7 @@ class Engine extends Disposable with EventsEmittable { ), ); _reliableDCPub?.bufferedAmountLowThreshold = 2 * 1024 * 1024; + _dcBufferStatus[Reliability.reliable] = true; _reliableDCPub?.onBufferedAmountLow = (_) { _dcBufferStatus[Reliability.reliable] = (_reliableDCPub!.bufferedAmount! <= _reliableDCPub!.bufferedAmountLowThreshold!); diff --git a/test/core/buffer_status_wait_test.dart b/test/core/buffer_status_wait_test.dart new file mode 100644 index 000000000..c965047f1 --- /dev/null +++ b/test/core/buffer_status_wait_test.dart @@ -0,0 +1,88 @@ +// 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 '../mock/e2e_container.dart'; +import '../mock/peerconnection_mock.dart'; + +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; + } + } + + /// 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 once the channel is empty, even without a low-buffer event', () async { + await fillReliableChannel(); + final wait = container.engine.waitForBufferStatusLow(Reliability.reliable); + + // An empty channel that never reports the drain, as when a full reconnect replaces the + // channel the wait started on. + setBufferedAmount(null); + + await wait.timeout(const Duration(seconds: 1)); + }); + + 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')); + }); +} 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;