From eccdb1186ba822cf2a4d13ea918408f3ca40a745 Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Thu, 29 Oct 2026 15:45:57 +0800 Subject: [PATCH 1/2] Fix waiting for the data channel send buffer to drain waitForBufferStatusLow had three problems: - It polled only the cached buffer status, which changes on a send or on the channel's low-buffer event. When a full reconnect replaced a backed up channel, the new one fired neither, so a waiter could hang until disconnect. The cache is now reset when the channels are created, and the wait also checks the live channel. - When the engine closed during a wait, the wait resolved normally and the caller went on to send. It now fails. - Every wait that blocked left an EngineClosingEvent listener behind until the engine closed. The listener is now cancelled. MockDataChannel gains fakeBufferedAmount to simulate a channel that has not drained. --- .changes/data-channel-buffer-wait | 1 + lib/src/core/engine.dart | 36 +++++------ test/core/buffer_status_wait_test.dart | 89 ++++++++++++++++++++++++++ test/mock/datachannel_mock.dart | 8 ++- 4 files changed, 112 insertions(+), 22 deletions(-) create mode 100644 .changes/data-channel-buffer-wait create mode 100644 test/core/buffer_status_wait_test.dart 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 0f46a92f8..c129e5491 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -395,29 +395,22 @@ class Engine extends Disposable with EventsEmittable { } Future waitForBufferStatusLow(Reliability kind) async { - final Completer completer = Completer(); + if (isBufferStatusLow(kind) == true) return; - 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; + var closing = false; + final cancelClosingListener = events.once((_) => closing = true); + try { + // 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 (!closing && !_dcBufferStatus[kind]! && isBufferStatusLow(kind) != true) { + await Future.delayed(const Duration(milliseconds: 10)); } - completer.complete(); + } finally { + await cancelClosingListener?.call(); + } + if (closing) { + return Future.error('Engine disconnected'); } - - return completer.future; } Future _resendReliableMessagesForResume(int lastMessageSeq) async { @@ -849,6 +842,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!); }; @@ -872,6 +867,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..50b82ff8a --- /dev/null +++ b/test/core/buffer_status_wait_test.dart @@ -0,0 +1,89 @@ +// 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/internal/events.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 its closing 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.events.emit(EngineClosingEvent()); + + 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; From bb1c5b8c2b81adff031f83395da23ec69ec04d75 Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Fri, 30 Oct 2026 04:05:53 +0800 Subject: [PATCH 2/2] Wait on the engine's closed state instead of a closing listener Every terminal path now marks the engine closed synchronously, so the wait can check that state directly. This drops the listener the wait had to remove again, and a wait started after closure fails the same way. --- lib/src/core/engine.dart | 23 +++++------------------ test/core/buffer_status_wait_test.dart | 5 ++--- 2 files changed, 7 insertions(+), 21 deletions(-) diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index f2a52eac5..57ce8ed4b 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -401,27 +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'; } - - if (isBufferStatusLow(kind) == true) return; - - var closing = false; - final cancelClosingListener = events.once((_) => closing = true); - try { - // 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 (!closing && !_dcBufferStatus[kind]! && isBufferStatusLow(kind) != true) { - await Future.delayed(const Duration(milliseconds: 10)); - } - } finally { - await cancelClosingListener?.call(); - } - if (closing) { - return Future.error('Engine disconnected'); - } } Future _resendReliableMessagesForResume(int lastMessageSeq) async { diff --git a/test/core/buffer_status_wait_test.dart b/test/core/buffer_status_wait_test.dart index 50b82ff8a..c965047f1 100644 --- a/test/core/buffer_status_wait_test.dart +++ b/test/core/buffer_status_wait_test.dart @@ -15,7 +15,6 @@ import 'package:flutter_test/flutter_test.dart'; import 'package:livekit_client/livekit_client.dart'; -import 'package:livekit_client/src/internal/events.dart'; import '../mock/e2e_container.dart'; import '../mock/peerconnection_mock.dart'; @@ -63,7 +62,7 @@ void main() { await wait.timeout(const Duration(seconds: 1)); }); - test('a wait does not leave its closing listener behind', () async { + 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); @@ -82,7 +81,7 @@ void main() { final wait = container.engine.waitForBufferStatusLow(Reliability.reliable); final outcome = wait.then((_) => 'resolved', onError: (Object e) => e); - container.engine.events.emit(EngineClosingEvent()); + container.engine.markClosed(); expect(await outcome.timeout(const Duration(seconds: 2)), isNot('resolved')); });