hiroshihorie · Oct 29, 2026 · Oct 29, 2026 · Oct 29, 2026 · Oct 29, 2026 · Oct 29, 2026
@@ -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"
@@ -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<EngineEvent> {
     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<void> _sendStateChanged = ReusableCompleter<void>();
+
+  void _setBufferStatus(Reliability kind, bool isLow) {
+    _dcBufferStatus[kind] = isLow;
+    _sendStateChanged.complete();
+  }
+
   List<lk_models.Codec>? _enabledPublishCodecs;
 
   List<lk_models.Codec>? get enabledPublishCodecs => _enabledPublishCodecs;
@@ -197,7 +207,12 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
     maxPacketCount: 1000, // max 1000 packets
   );
   final TTLMap<String, int> _reliableReceivedState = TTLMap<String, int>(30000);
-  bool _isReconnecting = false;
+  bool _reconnecting = false;
+  bool get _isReconnecting => _reconnecting;
+  set _isReconnecting(bool value) {
+    _reconnecting = value;
+    _sendStateChanged.complete();
+  }
 
   Completer<void>? _publisherConnectionCompleter;
 
@@ -422,35 +437,21 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
     return null;
   }
 
-  Future<void> waitForBufferStatusLow(Reliability kind) async {
-    // A wait started after closure cannot observe the closing event.
-    if (_isClosed || isDisposed) {
-      throw UnexpectedConnectionState('Engine disconnected');
-    }
-
-    final Completer<void> 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<void> 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<void> _waitUntil(bool Function() isReady) async {
+    while (true) {
+      if (_isClosed || isDisposed) {
+        throw UnexpectedConnectionState('Engine disconnected');
       }
-
-      events.once<EngineClosingEvent>((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<void> _resendReliableMessagesForResume(int lastMessageSeq) async {
@@ -569,7 +570,7 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
     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<EngineEvent> {
       );
       // _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<EngineEvent> {
         ),
       );
       _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<EngineEvent> {
   void markClosed() {
     if (_isClosed) return;
     _isClosed = true;
+    _sendStateChanged.complete();
     events.emit(EngineClosingEvent());
   }
 

@@ -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<void> 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<void>.delayed(const Duration(milliseconds: 10));
+    }
+    await Future<void>.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<void> 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<Object?>((_) => '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<void>.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');
+  });
+}
@@ -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<int> getBufferedAmount() => Future.value(bufferedAmountLowThreshold);
+  Future<int> getBufferedAmount() => Future.value(bufferedAmount);
 
   @override
   int? get id => _id;

@@ -35,12 +35,17 @@ MockDataChannel? findMockDataChannelByLabel(String label, {bool requireListener
 /// publisher↔subscriber loopback.
 List<MockDataChannel> 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;
   }
Original file line number	Diff line number	Diff line change
		@@ -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"