Data streams v2: fix review findings in the native shim

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

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();
       },
     );

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,

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"

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