diff --git a/example/pubspec.yaml b/example/pubspec.yaml
index c855990e1..e1bfc2a9a 100644
--- a/example/pubspec.yaml
+++ b/example/pubspec.yaml
@@ -46,3 +46,9 @@ flutter:
   uses-material-design: true
   assets:
     - images/logo-dark.svg
+
+# TEMPORARY: the polled outgoing queue's delivery acknowledgement (OutgoingPacketBatch) is not in a
+# published livekit_ffi yet. Replace with a version bump once livekit-ffi releases it.
+dependency_overrides:
+  livekit_ffi:
+    path: ../../rust-sdks/.claude/worktrees/polled-delivery-ack/livekit-ffi/packages/dart
diff --git a/lib/src/data_stream/data_streams_native.dart b/lib/src/data_stream/data_streams_native.dart
index b3ecc7495..fb46b420e 100644
--- a/lib/src/data_stream/data_streams_native.dart
+++ b/lib/src/data_stream/data_streams_native.dart
@@ -20,6 +20,7 @@ import 'dart:typed_data';
 import 'package:fixnum/fixnum.dart';
 import 'package:livekit_ffi/livekit_ffi.dart' as ffi;
 import 'package:meta/meta.dart';
+import 'package:mime_type/mime_type.dart';
 import 'package:path/path.dart' show basename;
 import 'package:uuid/uuid.dart';
 
@@ -80,10 +81,6 @@ class NativeDataStreams implements DataStreams {
   final Map<String, TextStreamHandler> _textStreamHandlers = {};
   final Map<String, ByteStreamHandler> _byteStreamHandlers = {};
 
-  /// Serializes outbound sends so packet order survives the hop from the pump into the engine's
-  /// async send.
-  Future<void> _sendChain = Future.value();
-
   bool _disposed = false;
 
   @override
@@ -183,7 +180,8 @@ class NativeDataStreams implements DataStreams {
           attributes: const {},
           destinationIdentities: options.destinationIdentities,
           id: id,
-          mimeType: options.mimeType,
+          // The core falls back to application/octet-stream and does no inference of its own.
+          mimeType: options.mimeType ?? mime(basename(file.path)),
           name: basename(file.path),
         ),
       ),
@@ -245,6 +243,15 @@ class NativeDataStreams implements DataStreams {
 
   /// Drains outbound packets from the core and puts them on the wire, in order.
   ///
+  /// Each batch is sent, one packet at a time and each only once the reliable data channel's
+  /// buffer is below its low-water mark, before the next batch is pulled. That keeps order
+  /// without a separate send chain and paces the wire as `WritableStream` did on main.
+  ///
+  /// Every batch is then settled: the core holds the send that produced it until we do, so a
+  /// `sendText`/`sendBytes`/`write` resolves only once its packets are on the data channel, and
+  /// fails with a [DataStreamError] when they could not be put there. The core produces no further
+  /// batch meanwhile, which is what stops a `sendFile` from reading ahead of the wire.
+  ///
   /// The pump owns the queue's lifetime and is the only thing that may dispose it — freeing it
   /// while a `nextPackets` is in flight is a use-after-free. [dispose] wakes us by closing the
   /// queue rather than releasing it.
@@ -253,8 +260,16 @@ class NativeDataStreams implements DataStreams {
       while (true) {
         final batch = await _outgoingPackets.nextPackets();
         if (batch == null) break; // closed or shutting down
-        for (final encoded in batch) {
-          _enqueueSend(encoded);
+        try {
+          for (final encoded in batch.packets()) {
+            await _send(encoded);
+          }
+          batch.complete();
+        } catch (e) {
+          logger.warning('[DataStreams] failed to send outbound packets: $e');
+          batch.fail(reason: '$e');
+        } finally {
+          batch.dispose();
         }
       }
     } catch (e) {
@@ -264,22 +279,16 @@ class NativeDataStreams implements DataStreams {
     }
   }
 
-  void _enqueueSend(Uint8List encoded) {
-    _sendChain = _sendChain.then((_) async {
-      final room = _room.target;
-      if (room == null || _disposed) return;
-      try {
-        // Back through the engine rather than the data channel directly, so E2EE wrapping,
-        // reliable sequencing and resume-resend all still apply.
-        await room.engine.sendDataPacket(
-          lk_models.DataPacket.fromBuffer(encoded),
-          reliability: Reliability.reliable,
-        );
-      } catch (e) {
-        // The core acknowledges sends unconditionally, so there is nobody to propagate this to.
-        logger.warning('[DataStreams] failed to send outbound packet: $e');
-      }
-    });
+  Future<void> _send(Uint8List encoded) async {
+    final room = _room.target;
+    if (room == null || _disposed) throw StateError('data streams were disposed');
+    await room.engine.waitForBufferStatusLow(Reliability.reliable);
+    // Back through the engine rather than the data channel directly, so E2EE wrapping, reliable
+    // sequencing and resume-resend all still apply.
+    await room.engine.sendDataPacket(
+      lk_models.DataPacket.fromBuffer(encoded),
+      reliability: Reliability.reliable,
+    );
   }
 
   // MARK: - Receive
@@ -375,6 +384,8 @@ class NativeDataStreams implements DataStreams {
         next: textReader.next,
         toBytes: (piece) => Uint8List.fromList(utf8.encode(piece)),
         streamId: info.id,
+        latestAttributes: () => textReader.info().attributes,
+        close: textReader.close,
         dispose: textReader.dispose,
       );
       handler(TextStreamReader(info, controller, info.size), identity);
@@ -399,6 +410,8 @@ class NativeDataStreams implements DataStreams {
         next: byteReader.next,
         toBytes: (piece) => piece,
         streamId: info.id,
+        latestAttributes: () => byteReader.info().attributes,
+        close: byteReader.close,
         dispose: byteReader.dispose,
       );
       handler(ByteStreamReader(info, controller, info.size), identity);
@@ -411,14 +424,22 @@ class NativeDataStreams implements DataStreams {
   /// stops while the subscription is paused, so the core's backpressure is preserved rather than
   /// buffering the whole stream into Dart.
   ///
+  /// [info] is a snapshot taken when the stream opened, but a trailer can carry more attributes.
+  /// The core merges those into its own info before the read that ends the stream returns, so they
+  /// are copied over from [latestAttributes] then, before the stream closes, as the v1 path does.
+  ///
   /// The pump owns the reader's lifetime and is the only thing that may dispose it. Disposing from
   /// `onCancel` instead would free the Rust handle while a `next()` is still in flight — a
-  /// use-after-free that shows up as a SIGBUS, not a Dart exception.
+  /// use-after-free that shows up as a SIGBUS, not a Dart exception. `onCancel` calls [close]
+  /// instead: it returns at once, ends the stream in the core even if the sender goes quiet, and
+  /// fails a pending `next()` with `AlreadyClosed`, which wakes the pump so it can dispose.
   DataStreamController<lk_models.DataStream_Chunk> _controllerFor<T extends Object>({
     required BaseStreamInfo info,
     required Future<T?> Function() next,
     required Uint8List Function(T piece) toBytes,
     required String streamId,
+    required Map<String, String> Function() latestAttributes,
+    required void Function() close,
     required void Function() dispose,
   }) {
     late final StreamController<lk_models.DataStream_Chunk> controller;
@@ -440,7 +461,10 @@ class NativeDataStreams implements DataStreams {
       try {
         while (!cancelled && !controller.isClosed && !controller.isPaused) {
           final piece = await next();
-          if (piece == null) break;
+          if (piece == null) {
+            info.attributes = {...info.attributes, ...latestAttributes()};
+            break;
+          }
           if (cancelled || controller.isClosed) break;
           wrapper.write(
             lk_models.DataStream_Chunk(
@@ -456,7 +480,9 @@ class NativeDataStreams implements DataStreams {
           disposeOnce();
         }
       } on ffi.DataStreamException catch (e) {
-        wrapper.error(toLKError(e));
+        // A cancelled subscription closed the reader; the AlreadyClosed it causes is expected and
+        // there is nobody listening to report it to.
+        if (!cancelled) wrapper.error(toLKError(e));
         await wrapper.close();
         disposeOnce();
       } catch (e) {
@@ -476,10 +502,11 @@ class NativeDataStreams implements DataStreams {
     controller = StreamController<lk_models.DataStream_Chunk>(
       onListen: () => unawaited(pump()),
       onResume: () => unawaited(pump()),
-      // Only flag it: the pump disposes once it has stopped touching the reader. If it is blocked
-      // in `next()` the reader stays alive until that resolves, which is the safe order.
+      // Close, don't dispose: the pump disposes once it has stopped touching the reader. Closing
+      // fails a `next()` it is blocked in, so it gets there without waiting on the sender.
       onCancel: () {
         cancelled = true;
+        if (!disposed) close();
         if (!running) disposeOnce();
       },
     );
diff --git a/lib/src/data_stream/data_streams_web.dart b/lib/src/data_stream/data_streams_web.dart
index 9e2c2a19a..3705ff8e4 100644
--- a/lib/src/data_stream/data_streams_web.dart
+++ b/lib/src/data_stream/data_streams_web.dart
@@ -371,7 +371,7 @@ class WebDataStreams implements DataStreams {
         streamId: streamId,
         totalSize: totalLength,
         name: basename(file.path),
-        mimeType: options.mimeType,
+        mimeType: options.mimeType ?? mime(basename(file.path)),
         topic: options.topic,
         destinationIdentities: options.destinationIdentities,
         encryptionType: options.encryptionType,
diff --git a/pubspec.lock b/pubspec.lock
index bd89e4b8a..227daaf02 100644
--- a/pubspec.lock
+++ b/pubspec.lock
@@ -5,18 +5,18 @@ packages:
     dependency: transitive
     description:
       name: _fe_analyzer_shared
-      sha256: cd6add6f846f35fb79f3c315296703c1a24f3cfd7f4739d91a74961c1c7e9f1b
+      sha256: "8d7ff3948166b8ec5da0fbb5962000926b8e02f2ed9b3e51d1738905fbd4c98d"
       url: "https://pub.dev"
     source: hosted
-    version: "100.0.0"
+    version: "93.0.0"
   analyzer:
     dependency: transitive
     description:
       name: analyzer
-      sha256: "6ba98576948803398b69e3a444df24eacdbe12ed699c7014e120ea38552debbf"
+      sha256: de7148ed2fcec579b19f122c1800933dfa028f6d9fd38a152b04b1516cec120b
       url: "https://pub.dev"
     source: hosted
-    version: "13.0.0"
+    version: "10.0.1"
   archive:
     dependency: transitive
     description:
@@ -197,10 +197,10 @@ packages:
     dependency: transitive
     description:
       name: dart_style
-      sha256: "59d53ef8eaed9d288ed9767618e2b31c4fa0383a127db59d5eb2e737a7638a60"
+      sha256: "29f7ecc274a86d32920b1d9cfc7502fa87220da41ec60b55f329559d5732e2b2"
       url: "https://pub.dev"
     source: hosted
-    version: "3.1.9"
+    version: "3.1.7"
   dart_webrtc:
     dependency: "direct main"
     description:
@@ -443,11 +443,10 @@ packages:
   livekit_ffi:
     dependency: "direct main"
     description:
-      name: livekit_ffi
-      sha256: "38ff93b4cb83239d2a4f5a57fe2ea68bd7ff845d1682c4f2c14713b6af04850d"
-      url: "https://pub.dev"
-    source: hosted
-    version: "0.13.2"
+      path: "../rust-sdks/.claude/worktrees/polled-delivery-ack/livekit-ffi/packages/dart"
+      relative: true
+    source: path
+    version: "0.13.3"
   logger:
     dependency: transitive
     description:
@@ -468,10 +467,10 @@ packages:
     dependency: transitive
     description:
       name: matcher
-      sha256: "31bd099b47c10cd1aeb55146a2d46ce0277630ecef3f7dae54ad7873f36696cd"
+      sha256: dc0b7dc7651697ea4ff3e69ef44b0407ea32c487a39fff6a4004fa585e901861
       url: "https://pub.dev"
     source: hosted
-    version: "0.12.20"
+    version: "0.12.19"
   material_color_utilities:
     dependency: transitive
     description:
@@ -484,10 +483,10 @@ packages:
     dependency: "direct main"
     description:
       name: meta
-      sha256: "307249ce4ff29d58a18e97f6345f539382eb9c9c29ecda628900f31de0443dd9"
+      sha256: "23f08335362185a5ea2ad3a4e597f1375e78bce8a040df5c600c8d3552ef2394"
       url: "https://pub.dev"
     source: hosted
-    version: "1.19.0"
+    version: "1.17.0"
   mime:
     dependency: transitive
     description:
@@ -508,10 +507,10 @@ packages:
     dependency: "direct dev"
     description:
       name: mockito
-      sha256: c8d040d754367108fbe482dcb79dd72b8fe60ac6727abd15b4783c5560297ee6
+      sha256: eff30d002f0c8bf073b6f929df4483b543133fcafce056870163587b03f1d422
       url: "https://pub.dev"
     source: hosted
-    version: "5.7.0"
+    version: "5.6.4"
   nm:
     dependency: transitive
     description:
@@ -761,10 +760,10 @@ packages:
     dependency: "direct main"
     description:
       name: synchronized
-      sha256: "61894a1956de6b4fc1aefd0892e109514a1a706cbece3ac59decd90ff5a7a423"
+      sha256: "63896c27e81b28f8cb4e69ead0d3e8f03f1d1e5fc531a3e579cabed6a2c7c9e5"
       url: "https://pub.dev"
     source: hosted
-    version: "3.4.1+1"
+    version: "3.4.0+1"
   term_glyph:
     dependency: transitive
     description:
@@ -777,10 +776,10 @@ packages:
     dependency: transitive
     description:
       name: test_api
-      sha256: "2a122cbe059f8b610d3a5415f42e255b6c17b1f21eee1d960f31080237fb4f11"
+      sha256: "8161c84903fd860b26bfdefb7963b3f0b68fee7adea0f59ef805ecca346f0c7a"
       url: "https://pub.dev"
     source: hosted
-    version: "0.7.12"
+    version: "0.7.10"
   tint:
     dependency: transitive
     description:
@@ -809,10 +808,10 @@ packages:
     dependency: transitive
     description:
       name: vector_math
-      sha256: f36f9f3be64c6198714492bb455c11056e33e2f85d9a0b676a48301e44fdcf47
+      sha256: d530bd74fea330e6e364cda7a85019c434070188383e1cd8d9777ee586914c5b
       url: "https://pub.dev"
     source: hosted
-    version: "2.4.2"
+    version: "2.2.0"
   vm_service:
     dependency: transitive
     description:
@@ -902,5 +901,5 @@ packages:
     source: hosted
     version: "3.1.3"
 sdks:
-  dart: ">=3.12.0 <4.0.0"
+  dart: ">=3.11.0 <4.0.0"
   flutter: ">=3.38.4"
diff --git a/pubspec.yaml b/pubspec.yaml
index 884b44740..6ed1ffe72 100644
--- a/pubspec.yaml
+++ b/pubspec.yaml
@@ -59,7 +59,7 @@ dependencies:
 
   # Rust core (livekit-ffi), delivered as a bundled cdylib via Native Assets.
   # Native platforms only, see lib/src/ffi/.
-  livekit_ffi: ^0.13.2
+  livekit_ffi: ^0.13.3
 
 dev_dependencies:
   flutter_test:
@@ -102,3 +102,9 @@ topics:
   - video
   - livestream
   - conference
+
+# TEMPORARY: the polled outgoing queue's delivery acknowledgement (OutgoingPacketBatch) is not in a
+# published livekit_ffi yet. Replace with a version bump once livekit-ffi releases it.
+dependency_overrides:
+  livekit_ffi:
+    path: ../rust-sdks/.claude/worktrees/polled-delivery-ack/livekit-ffi/packages/dart
diff --git a/test/core/data_stream_v2_test.dart b/test/core/data_stream_v2_test.dart
index b9f66ddbe..a040c8a0a 100644
--- a/test/core/data_stream_v2_test.dart
+++ b/test/core/data_stream_v2_test.dart
@@ -23,6 +23,7 @@ library;
 
 import 'dart:async';
 import 'dart:convert';
+import 'dart:io';
 import 'dart:typed_data';
 
 import 'package:fixnum/fixnum.dart';
@@ -185,6 +186,98 @@ void main() {
       expect(info.name, equals('unknown'));
       expect(info.mimeType, equals('application/octet-stream'));
     });
+
+    test('outbound packets wait for the reliable data channel buffer to drain', () async {
+      await container.simulateRemoteParticipantJoin('alice', clientProtocol: 2, capabilities: _v2WithCompression);
+      final reliable = mockDataChannels.where((dc) => dc.label == '_reliable').toList();
+      for (final dc in reliable) {
+        dc.bufferedAmountOverride = 1 << 30;
+      }
+      Iterable<String> sentTopics() => streamPackets().map((p) => p.streamHeader.topic);
+
+      // The engine samples the buffer after each send, so the first packet goes out and leaves
+      // the channel marked as above its low-water mark.
+      await room.localParticipant!.sendText('first', options: SendTextOptions(topic: 'first'));
+      await _waitUntil(() => sentTopics().contains('first'));
+
+      unawaited(room.localParticipant!.sendText('second', options: SendTextOptions(topic: 'second')));
+      await Future<void>.delayed(const Duration(milliseconds: 200));
+      expect(sentTopics(), isNot(contains('second')), reason: 'held while the buffer is full');
+
+      for (final dc in reliable) {
+        dc.bufferedAmountOverride = null;
+        dc.onBufferedAmountLow?.call(0);
+      }
+      await _waitUntil(() => sentTopics().contains('second'));
+    });
+
+    test('a send resolves only once its packets are on the data channel', () async {
+      await container.simulateRemoteParticipantJoin('alice', clientProtocol: 2, capabilities: _v2WithCompression);
+      final reliable = mockDataChannels.where((dc) => dc.label == '_reliable').toList();
+      for (final dc in reliable) {
+        dc.bufferedAmountOverride = 1 << 30;
+      }
+      // Leaves the channel marked as above its low-water mark; see the backpressure test.
+      await room.localParticipant!.sendText('first', options: SendTextOptions(topic: 'first'));
+
+      var resolved = false;
+      final second = room.localParticipant!
+          .sendText('second', options: SendTextOptions(topic: 'second'))
+          .whenComplete(() => resolved = true);
+      await Future<void>.delayed(const Duration(milliseconds: 200));
+      expect(resolved, isFalse, reason: 'nothing has been sent yet');
+
+      for (final dc in reliable) {
+        dc.bufferedAmountOverride = null;
+        dc.onBufferedAmountLow?.call(0);
+      }
+      await second;
+      expect(streamPackets().map((p) => p.streamHeader.topic), contains('second'));
+    });
+
+    test('a send fails when its packets cannot be put on the data channel', () async {
+      await container.simulateRemoteParticipantJoin('alice', clientProtocol: 2, capabilities: _v2WithCompression);
+      for (final dc in mockDataChannels.where((dc) => dc.label == '_reliable')) {
+        dc.onMessageSend = (_) => throw StateError('data channel is gone');
+      }
+
+      await expectLater(
+        room.localParticipant!.sendText('lost', options: SendTextOptions(topic: 'lost')),
+        throwsA(isA<DataStreamError>()),
+      );
+    });
+
+    group('MIME type inference', () {
+      late Directory dir;
+      setUp(() async => dir = await Directory.systemTemp.createTemp('lk_mime'));
+      tearDown(() async => dir.delete(recursive: true));
+
+      Future<File> file(String name) async => File('${dir.path}/$name')..writeAsBytesSync([1, 2, 3, 4]);
+
+      test('sendFile infers the MIME type from the file name', () async {
+        await room.localParticipant!.sendFile(await file('hello.png'), options: SendFileOptions(topic: 'files'));
+        expect(streamPackets().first.streamHeader.mimeType, equals('image/png'));
+      });
+
+      test('an explicit MIME type wins over inference', () async {
+        await room.localParticipant!.sendFile(
+          await file('hello.png'),
+          options: SendFileOptions(topic: 'files', mimeType: 'application/x-custom'),
+        );
+        expect(streamPackets().first.streamHeader.mimeType, equals('application/x-custom'));
+      });
+
+      test('sendText attachments infer their MIME type', () async {
+        await room.localParticipant!.sendText(
+          'see attached',
+          options: SendTextOptions(topic: 'chat', attachments: [await file('notes.pdf')]),
+        );
+        final byteHeaders = streamPackets()
+            .where((p) => p.hasStreamHeader() && p.streamHeader.hasByteHeader())
+            .map((p) => p.streamHeader);
+        expect(byteHeaders.single.mimeType, equals('application/pdf'));
+      });
+    });
   });
 
   group('receive side', () {
@@ -266,6 +359,51 @@ void main() {
       expect(await received.future, equals(text));
     });
 
+    test('trailer attributes are merged into the reader info by the time the stream ends', () async {
+      final opened = Completer<void>();
+      final done = Completer<TextStreamInfo>();
+      room.registerTextStreamHandler('trailered', (reader, identity) async {
+        opened.complete();
+        expect(await reader.readAll(), equals('hello'));
+        done.complete(reader.info!);
+      });
+
+      lk_models.DataPacket packet() =>
+          lk_models.DataPacket(kind: lk_models.DataPacket_Kind.RELIABLE, participantIdentity: 'alice');
+      container.deliverInboundDataPacket(
+        packet()
+          ..streamHeader = lk_models.DataStream_Header(
+            streamId: 'trailered-1',
+            topic: 'trailered',
+            mimeType: 'text/plain',
+            timestamp: Int64(DateTime.timestamp().millisecondsSinceEpoch),
+            attributes: {'a': '1'}.entries,
+            textHeader: lk_models.DataStream_TextHeader(),
+          ),
+      );
+      // The handler has to see the stream before the trailer lands, or the info it is handed would
+      // already include the trailer's attributes and prove nothing.
+      await opened.future.timeout(const Duration(seconds: 5));
+      container.deliverInboundDataPacket(
+        packet()
+          ..streamChunk = lk_models.DataStream_Chunk(
+            streamId: 'trailered-1',
+            chunkIndex: Int64(0),
+            content: Uint8List.fromList(utf8.encode('hello')),
+          ),
+      );
+      container.deliverInboundDataPacket(
+        packet()
+          ..streamTrailer = lk_models.DataStream_Trailer(
+            streamId: 'trailered-1',
+            attributes: {'b': '2'}.entries,
+          ),
+      );
+
+      final info = await done.future.timeout(const Duration(seconds: 5));
+      expect(info.attributes, equals({'a': '1', 'b': '2'}));
+    });
+
     test('a stream on an unregistered topic is ignored', () async {
       var fired = false;
       room.registerTextStreamHandler('registered', (reader, identity) async {
@@ -358,6 +496,50 @@ void main() {
   });
 
   group('open stream accounting', () {
+    test('cancelling the subscription ends the stream in the core', () async {
+      final ds = room.dataStreams as NativeDataStreams;
+      final cancelled = Completer<void>();
+      room.registerTextStreamHandler('cancelled', (reader, identity) async {
+        late final StreamSubscription<lk_models.DataStream_Chunk> sub;
+        sub = reader.listen((_) async {
+          await sub.cancel();
+          cancelled.complete();
+        });
+      });
+
+      lk_models.DataPacket packet() =>
+          lk_models.DataPacket(kind: lk_models.DataPacket_Kind.RELIABLE, participantIdentity: 'alice');
+      room.dataStreams.handleIncomingPacket(
+        packet()
+          ..streamHeader = lk_models.DataStream_Header(
+            streamId: 'cancelled-1',
+            topic: 'cancelled',
+            mimeType: 'text/plain',
+            timestamp: Int64(DateTime.timestamp().millisecondsSinceEpoch),
+            textHeader: lk_models.DataStream_TextHeader(),
+          ),
+        EncryptionType.kNone,
+      );
+      room.dataStreams.handleIncomingPacket(
+        packet()
+          ..streamChunk = lk_models.DataStream_Chunk(
+            streamId: 'cancelled-1',
+            chunkIndex: Int64(0),
+            content: Uint8List.fromList(utf8.encode('first')),
+          ),
+        EncryptionType.kNone,
+      );
+
+      await cancelled.future.timeout(const Duration(seconds: 5));
+      // The pump is now parked in next() with the sender silent; only closing the reader frees
+      // the stream. The close reaches the core's loop asynchronously, so give it a moment.
+      final deadline = DateTime.now().add(const Duration(seconds: 2));
+      while (await ds.debugOpenStreamCount() != 0 && DateTime.now().isBefore(deadline)) {
+        await Future<void>.delayed(const Duration(milliseconds: 10));
+      }
+      expect(await ds.debugOpenStreamCount(), 0);
+    });
+
     test('the count follows headers, and reset discards the manager', () async {
       room.registerTextStreamHandler('counted', (reader, identity) async {});
       final ds = room.dataStreams as NativeDataStreams;
@@ -473,3 +655,12 @@ void main() {
     });
   });
 }
+
+/// Polls [condition] until it holds, failing after [timeout].
+Future<void> _waitUntil(bool Function() condition, {Duration timeout = const Duration(seconds: 5)}) async {
+  final deadline = DateTime.now().add(timeout);
+  while (!condition()) {
+    if (DateTime.now().isAfter(deadline)) fail('condition not met within $timeout');
+    await Future<void>.delayed(const Duration(milliseconds: 10));
+  }
+}
diff --git a/test/mock/datachannel_mock.dart b/test/mock/datachannel_mock.dart
index e63121f83..6434204e4 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;
 
+  /// When set, reported as the buffered amount instead of the threshold, so a test can hold the
+  /// channel above its low-water mark. Clear it and fire [onBufferedAmountLow] to drain.
+  int? bufferedAmountOverride;
+
   @override
-  int? get bufferedAmount => bufferedAmountLowThreshold;
+  int? get bufferedAmount => bufferedAmountOverride ?? bufferedAmountLowThreshold;
 
   @override
-  Future<int> getBufferedAmount() => Future.value(bufferedAmountLowThreshold);
+  Future<int> getBufferedAmount() => Future.value(bufferedAmount);
 
   @override
   int? get id => _id;