Fix waiting for the data channel send buffer to drain

.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"

lib/src/core/engine.dart

@@ -401,34 +401,14 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
   }
 
   Future<void> 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<void>.delayed(const Duration(milliseconds: 10));
+    }
     if (_isClosed || isDisposed) {
       throw 'Engine disconnected';
     }
-
-    final Completer<void> completer = Completer();
-
-    if (isBufferStatusLow(kind) == true) {
-      completer.complete();
-    } else {
-      onClosing() {
-        if (!completer.isCompleted) {
-          completer.completeError('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();
-    }
-
-    return completer.future;
   }
 
   Future<void> _resendReliableMessagesForResume(int lastMessageSeq) async {
@@ -860,6 +840,8 @@ 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.
+      _dcBufferStatus[Reliability.lossy] = true;
       _lossyDCPub?.onBufferedAmountLow = (_) {
         _dcBufferStatus[Reliability.lossy] = (_lossyDCPub!.bufferedAmount! <= _lossyDCPub!.bufferedAmountLowThreshold!);
       };
@@ -883,6 +865,7 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
         ),
       );
       _reliableDCPub?.bufferedAmountLowThreshold = 2 * 1024 * 1024;
+      _dcBufferStatus[Reliability.reliable] = true;
       _reliableDCPub?.onBufferedAmountLow = (_) {
         _dcBufferStatus[Reliability.reliable] =
             (_reliableDCPub!.bufferedAmount! <= _reliableDCPub!.bufferedAmountLowThreshold!);

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<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 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<Object?>((_) => 'resolved', onError: (Object e) => e);
+
+    container.engine.markClosed();
+
+    expect(await outcome.timeout(const Duration(seconds: 2)), isNot('resolved'));
+  });
+}

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<int> getBufferedAmount() => Future.value(bufferedAmountLowThreshold);
+  Future<int> getBufferedAmount() => Future.value(bufferedAmount);
 
   @override
   int? get id => _id;