feat: add ConnectOptions.dataStream with maxPayloadByteLength

The Rust core has always accepted a payload cap — it is the guard against
a sender making a receiver allocate arbitrary memory, including via a
compressed stream that inflates far past its wire size — but nothing
exposed it, so `null` was hardcoded and every room ran with the core's
5 GB default.

Placed on `ConnectOptions` rather than `RoomOptions`, which is where
Swift puts it. It fits the lifecycle better here: the incoming manager is
created lazily on the first inbound packet precisely so a connect-time
value is in effect by the time it is read. Worth knowing it diverges from
Swift if cross-SDK consistency matters more than that.

Also enforced on web, which otherwise would have accepted the option and
silently done nothing with it — the same trap as the ignored
`connect(roomOptions:)` argument. The Dart path tracks accumulated
content bytes per stream so an unknown-length stream is capped too, not
just one that declares an oversized `totalLength`.

Both paths follow the core's semantics, which are subtler than they look:
the stream-opened event fires *before* the cap is applied, so the topic
handler still runs and it is the reader that fails with
`LengthExceeded`. A consumer is told the stream died rather than watching
it never arrive. My first pass had web refusing the stream outright and
the doc comment describing that; both are corrected here.

Author: 1egoman

.changes/data-stream-options

@@ -0,0 +1 @@
+minor type="added" "ConnectOptions.dataStream with maxPayloadByteLength, bounding the payload a single incoming data stream may deliver"

lib/src/data_stream/data_streams_native.dart

@@ -71,8 +71,8 @@ class NativeDataStreams implements DataStreams {
   late final ffi.OutgoingDataStreamManager _outgoing;
   late final ffi.OutgoingPacketQueue _outgoingPackets;
 
-  /// Created on the first inbound packet rather than here, so a `maxPayloadSize` supplied at
-  /// connect time is picked up.
+  /// Created on the first inbound packet rather than here, so a
+  /// [DataStreamOptions.maxPayloadByteLength] supplied at connect time is picked up.
   ffi.IncomingDataStreamManager? _incoming;
   ffi.IncomingStreamQueue? _incomingStreams;
 
@@ -286,7 +286,11 @@ class NativeDataStreams implements DataStreams {
   ffi.IncomingDataStreamManager _incomingManager() {
     final existing = _incoming;
     if (existing != null) return existing;
-    final incoming = ffi.polledIncomingDataStreamManager(maxPayloadByteLength: null);
+    // Read now rather than at construction: this runs on the first inbound packet, i.e. after
+    // connect, so a cap supplied via `connect(connectOptions:)` is in effect by this point.
+    final incoming = ffi.polledIncomingDataStreamManager(
+      maxPayloadByteLength: _room.target?.connectOptions.dataStream.maxPayloadByteLength,
+    );
     _incoming = incoming.manager;
     _incomingStreams = incoming.streams;
     unawaited(_pumpIncoming(incoming.streams));

lib/src/data_stream/data_streams_web.dart

@@ -26,6 +26,7 @@ import '../core/room.dart';
 import '../e2ee/options.dart';
 import '../internal/events.dart';
 import '../logger.dart';
+import '../options.dart';
 import '../proto/livekit_models.pb.dart' as lk_models;
 import '../types/data_stream.dart';
 import '../types/other.dart';
@@ -59,6 +60,32 @@ class WebDataStreams implements DataStreams {
   final Map<String, DataStreamController<lk_models.DataStream_Chunk>> _byteStreamControllers = {};
   final Map<String, DataStreamController<lk_models.DataStream_Chunk>> _textStreamControllers = {};
 
+  /// Content bytes delivered so far per stream id, checked against
+  /// [DataStreamOptions.maxPayloadByteLength].
+  final Map<String, int> _receivedBytes = {};
+
+  int get _maxPayloadByteLength => _room.connectOptions.dataStream.maxPayloadByteLength ?? kDefaultMaxPayloadByteLength;
+
+  /// Whether a stream declaring [totalLength] is over the payload cap. Streams of unknown length
+  /// pass here and are capped as their chunks arrive instead.
+  bool _declaresOverCap(int? totalLength) => totalLength != null && totalLength > _maxPayloadByteLength;
+
+  /// Fails an oversized stream's reader, after its handler has been given it.
+  ///
+  /// Matches the Rust core, which emits the stream-opened event before applying the cap: the
+  /// consumer is told the stream failed rather than never hearing about it.
+  Future<void> _failOverCap(
+    DataStreamController<lk_models.DataStream_Chunk> controller,
+    String streamId,
+  ) async {
+    logger.warning(
+      'incoming stream $streamId exceeds the maxPayloadByteLength of $_maxPayloadByteLength',
+    );
+    controller.error(_payloadTooLarge());
+    await controller.close();
+    _forgetStream(streamId);
+  }
+
   @override
   void registerTextStreamHandler(String topic, TextStreamHandler callback) => textStreamHandlers[topic] = callback;
 
@@ -124,6 +151,9 @@ class WebDataStreams implements DataStreams {
       _byteStreamControllers[streamHeader.streamId] = controller;
 
       streamHandlerCallback(ByteStreamReader(info, controller, info.size), participantIdentity);
+      if (_declaresOverCap(streamHeader.hasTotalLength() ? info.size : null)) {
+        await _failOverCap(controller, streamHeader.streamId);
+      }
       return;
     }
 
@@ -166,6 +196,9 @@ class WebDataStreams implements DataStreams {
       _textStreamControllers[streamHeader.streamId] = controller;
 
       streamHandlerCallback(TextStreamReader(info, controller, info.size), participantIdentity);
+      if (_declaresOverCap(streamHeader.hasTotalLength() ? info.size : null)) {
+        await _failOverCap(controller, streamHeader.streamId);
+      }
     }
   }
 
@@ -174,8 +207,14 @@ class WebDataStreams implements DataStreams {
     if (textController != null) {
       if (textController.info.encryptionType != encryptionType) {
         textController.error(_encryptionMismatch());
-        _textStreamControllers.remove(chunk.streamId);
+        _forgetStream(chunk.streamId);
       } else if (chunk.content.isNotEmpty) {
+        if (_exceedsPayloadCap(chunk)) {
+          textController.error(_payloadTooLarge());
+          unawaited(textController.close());
+          _forgetStream(chunk.streamId);
+          return;
+        }
         textController.write(chunk);
       }
     }
@@ -184,36 +223,61 @@ class WebDataStreams implements DataStreams {
     if (byteController != null) {
       if (byteController.info.encryptionType != encryptionType) {
         byteController.error(_encryptionMismatch());
-        _byteStreamControllers.remove(chunk.streamId);
+        _forgetStream(chunk.streamId);
       } else if (chunk.content.isNotEmpty) {
+        if (_exceedsPayloadCap(chunk)) {
+          byteController.error(_payloadTooLarge());
+          unawaited(byteController.close());
+          _forgetStream(chunk.streamId);
+          return;
+        }
         byteController.write(chunk);
       }
     }
   }
 
+  /// Accumulates this chunk against the stream's running total, returning true once the payload
+  /// cap is passed.
+  bool _exceedsPayloadCap(lk_models.DataStream_Chunk chunk) {
+    final total = (_receivedBytes[chunk.streamId] ?? 0) + chunk.content.length;
+    _receivedBytes[chunk.streamId] = total;
+    return total > _maxPayloadByteLength;
+  }
+
+  DataStreamError _payloadTooLarge() => DataStreamError(
+    message: 'Stream payload exceeds the maxPayloadByteLength of $_maxPayloadByteLength',
+    reason: DataStreamErrorReason.LengthExceeded,
+  );
+
+  void _forgetStream(String streamId) {
+    _textStreamControllers.remove(streamId);
+    _byteStreamControllers.remove(streamId);
+    _receivedBytes.remove(streamId);
+  }
+
   Future<void> _handleStreamTrailer(lk_models.DataStream_Trailer trailer, EncryptionType encryptionType) async {
     final textController = _textStreamControllers[trailer.streamId];
     if (textController != null) {
       if (textController.info.encryptionType != encryptionType) {
         textController.error(_encryptionMismatch());
-        _textStreamControllers.remove(trailer.streamId);
+        _forgetStream(trailer.streamId);
         return;
       }
       textController.info.attributes = {...textController.info.attributes, ...trailer.attributes};
       await textController.close();
-      _textStreamControllers.remove(trailer.streamId);
+      _forgetStream(trailer.streamId);
     }
 
     final byteController = _byteStreamControllers[trailer.streamId];
     if (byteController != null) {
       if (byteController.info.encryptionType != encryptionType) {
         byteController.error(_encryptionMismatch());
-        _byteStreamControllers.remove(trailer.streamId);
+        _forgetStream(trailer.streamId);
         return;
       }
       byteController.info.attributes = {...byteController.info.attributes, ...trailer.attributes};
       await byteController.close();
-      _byteStreamControllers.remove(trailer.streamId);
+      _forgetStream(trailer.streamId);
     }
   }
 
@@ -461,12 +525,12 @@ class WebDataStreams implements DataStreams {
     for (final controller in bytes) {
       controller.error(abnormalEndError);
       await controller.close();
-      _byteStreamControllers.remove(controller.info.id);
+      _forgetStream(controller.info.id);
     }
     for (final controller in texts) {
       controller.error(abnormalEndError);
       await controller.close();
-      _textStreamControllers.remove(controller.info.id);
+      _forgetStream(controller.info.id);
     }
   }
 
@@ -477,6 +541,7 @@ class WebDataStreams implements DataStreams {
     }
     _textStreamControllers.clear();
     _byteStreamControllers.clear();
+    _receivedBytes.clear();
   }
 
   @override

lib/src/options.dart

@@ -219,15 +219,38 @@ class ConnectOptions {
 
   final Timeouts timeouts;
 
+  /// Tuning for incoming data streams.
+  final DataStreamOptions dataStream;
+
   const ConnectOptions({
     this.autoSubscribe = true,
     this.rtcConfiguration = const RTCConfiguration(),
     this.protocolVersion = ProtocolVersion.v16,
     this.clientProtocolVersion = ClientProtocolVersion.current,
     this.timeouts = Timeouts.defaultTimeouts,
+    this.dataStream = const DataStreamOptions(),
   });
 }
 
+/// Options for receiving data streams.
+/// {@category Room}
+class DataStreamOptions {
+  /// Largest payload, in bytes, that a single incoming stream may deliver.
+  ///
+  /// Bounds the memory one sender can make a receiver allocate, including via a compressed stream
+  /// that inflates far beyond its wire size. The topic handler is still invoked for an oversized
+  /// stream — it is the *reader* that fails, with [DataStreamErrorReason.LengthExceeded] — so
+  /// consumers find out rather than seeing a stream silently vanish.
+  ///
+  /// `null` uses [kDefaultMaxPayloadByteLength].
+  final int? maxPayloadByteLength;
+
+  const DataStreamOptions({this.maxPayloadByteLength});
+}
+
+/// Default for [DataStreamOptions.maxPayloadByteLength]; matches the Rust core's own default.
+const int kDefaultMaxPayloadByteLength = 5000000000;
+
 /// Options used to modify the behavior of the [Room].
 /// {@category Room}
 class RoomOptions {

test/core/data_stream_v2_test.dart

@@ -282,4 +282,90 @@ void main() {
       expect(fired, isFalse);
     });
   });
+
+  group('maxPayloadByteLength', () {
+    test('a stream declaring more than the cap fails its reader', () async {
+      // A fresh container so the cap is set at connect time, which is when the native manager
+      // reads it.
+      resetMockDataChannels();
+      final capped = E2EContainer();
+      addTearDown(capped.dispose);
+      await capped.connectRoom(
+        connectOptions: const ConnectOptions(
+          dataStream: DataStreamOptions(maxPayloadByteLength: 16),
+        ),
+      );
+
+      // The handler is still invoked — the core reports the stream opened before applying the
+      // cap — and it is the read that fails.
+      final outcome = Completer<Object?>();
+      capped.room.registerTextStreamHandler('capped', (reader, identity) async {
+        try {
+          await reader.readAll();
+          outcome.complete(null);
+        } catch (e) {
+          outcome.complete(e);
+        }
+      });
+
+      capped.deliverInboundDataPacket(
+        lk_models.DataPacket(
+          kind: lk_models.DataPacket_Kind.RELIABLE,
+          participantIdentity: 'alice',
+          streamHeader: lk_models.DataStream_Header(
+            streamId: 'too-big',
+            topic: 'capped',
+            mimeType: 'text/plain',
+            timestamp: Int64(DateTime.timestamp().millisecondsSinceEpoch),
+            totalLength: Int64(1000),
+            inlineContent: Uint8List.fromList(utf8.encode('x' * 1000)),
+            textHeader: lk_models.DataStream_TextHeader(),
+          ),
+        ),
+      );
+
+      final error = await outcome.future.timeout(const Duration(seconds: 5));
+      expect(error, isA<DataStreamError>());
+      expect(
+        (error as DataStreamError).reason,
+        DataStreamErrorReason.LengthExceeded,
+        reason: 'the payload exceeds maxPayloadByteLength',
+      );
+    });
+
+    test('a stream within the cap is delivered', () async {
+      resetMockDataChannels();
+      final capped = E2EContainer();
+      addTearDown(capped.dispose);
+      await capped.connectRoom(
+        connectOptions: const ConnectOptions(
+          dataStream: DataStreamOptions(maxPayloadByteLength: 1000),
+        ),
+      );
+
+      const text = 'small enough';
+      final received = Completer<String>();
+      capped.room.registerTextStreamHandler('capped', (reader, identity) async {
+        received.complete(await reader.readAll());
+      });
+
+      capped.deliverInboundDataPacket(
+        lk_models.DataPacket(
+          kind: lk_models.DataPacket_Kind.RELIABLE,
+          participantIdentity: 'alice',
+          streamHeader: lk_models.DataStream_Header(
+            streamId: 'small',
+            topic: 'capped',
+            mimeType: 'text/plain',
+            timestamp: Int64(DateTime.timestamp().millisecondsSinceEpoch),
+            totalLength: Int64(utf8.encode(text).length),
+            inlineContent: Uint8List.fromList(utf8.encode(text)),
+            textHeader: lk_models.DataStream_TextHeader(),
+          ),
+        ),
+      );
+
+      expect(await received.future, equals(text));
+    });
+  });
 }