fix(data-streams): honor connect-time maxPayloadSize

The incoming manager was built at Room.init with the payload cap read
  from the initial room options, so a maxPayloadSize supplied at connect
  time was ignored. Build the incoming manager lazily on the first inbound
  packet (post-connect), reading the room's current options then. Guarded
  by StateSync so it's constructed exactly once. reset()/closeStreams(from:)
  no-op when no packets have arrived.

Author: 1egoman

Sources/LiveKit/Core/Room.swift

@@ -280,7 +280,7 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable {
 
         super.init()
 
-        dataStreams = DataStreams(room: self, maxPayloadSize: _state.roomOptions.dataStreamOptions.maxPayloadSize)
+        dataStreams = DataStreams(room: self)
 
         // log sdk & os versions
         log("sdk: \(LiveKitSDK.version), ffi: \(LiveKitSDK.ffiVersion), os: \(String(describing: Utils.os()))(\(Utils.osVersionString())), modelId: \(String(describing: Utils.modelIdentifier() ?? "unknown"))")

Sources/LiveKit/DataStream/DataStreams.swift

@@ -34,9 +34,14 @@ internal import LiveKitUniFFI
 /// delegates are immutable after init. Not an actor — the UniFFI delegate callbacks are synchronous
 /// and can't `await`.
 final class DataStreams: NSObject, @unchecked Sendable, Loggable {
-    private let incoming: LiveKitUniFFI.IncomingDataStreamManager
     private let outgoing: LiveKitUniFFI.OutgoingDataStreamManager
 
+    // Created lazily on the first inbound packet, not at init: the incoming manager's payload cap
+    // comes from the room's options, which aren't finalized until `connect` — after this coordinator
+    // is built at `Room.init`. Deferring lets it pick up a `maxPayloadSize` passed at connect time.
+    // StateSync-guarded so it's constructed exactly once even if packets race in.
+    private let _incoming = StateSync<LiveKitUniFFI.IncomingDataStreamManager?>(nil)
+
     // Held weakly: the Room owns this coordinator, so the back-reference must not retain it. Used
     // for the room-level encryption type stamped onto stream info, and for logging.
     private weak var room: Room?
@@ -58,22 +63,31 @@ final class DataStreams: NSObject, @unchecked Sendable, Loggable {
     private let orderedTopics = StateSync<Set<String>>([])
     private let orderedTails = StateSync<[String: [String: Task<Void, Never>]]>([:])
 
-    init(room: Room, maxPayloadSize: Int? = nil) {
+    init(room: Room) {
         self.room = room
-        let incomingDelegate = IncomingDelegate()
         let outgoingDelegate = OutgoingDelegate(room: room)
         let registry = Registry(room: room)
-        // `maxPayloadSize` caps the reassembled size of an incoming stream (nil → the core's default
-        // cap). Topic routing (incl. the `lk.rpc` guard) is handled Swift-side in `Room+DataStream`,
-        // matching the previous pure-Swift implementation.
-        incoming = LiveKitUniFFI.IncomingDataStreamManager(
-            delegate: incomingDelegate,
-            maxPayloadByteLength: maxPayloadSize.map { UInt64($0) },
-        )
         outgoing = LiveKitUniFFI.OutgoingDataStreamManager(delegate: outgoingDelegate, registry: registry)
         super.init()
-        // The FFI manager retains its delegate strongly, so the delegate points back here weakly.
-        incomingDelegate.coordinator = self
+    }
+
+    /// The incoming manager, created on first use with the room's current payload cap. Topic routing
+    /// (incl. the `lk.rpc` guard) is handled Swift-side in `Room+DataStream`.
+    private func incomingManager() -> LiveKitUniFFI.IncomingDataStreamManager {
+        _incoming.mutate { existing in
+            if let existing { return existing }
+            let delegate = IncomingDelegate()
+            delegate.coordinator = self
+            // `nil` → the core's default cap. Read now (first packet, i.e. post-connect) so a
+            // `maxPayloadSize` supplied via `connect(roomOptions:)` is honored.
+            let maxPayloadSize = room?._state.roomOptions.dataStreamOptions.maxPayloadSize
+            let manager = LiveKitUniFFI.IncomingDataStreamManager(
+                delegate: delegate,
+                maxPayloadByteLength: maxPayloadSize.map { UInt64($0) },
+            )
+            existing = manager
+            return manager
+        }
     }
 
     // Room-level encryption type, stamped onto every stream info as it crosses the FFI boundary.
@@ -178,7 +192,7 @@ final class DataStreams: NSObject, @unchecked Sendable, Loggable {
     /// the incoming manager. The FFI re-decodes the serialized `DataPacket` itself.
     func handleIncoming(_ dataPacket: Livekit_DataPacket) {
         guard let data = try? dataPacket.serializedData() else { return }
-        incoming.handlePacketReceived(packet: data)
+        incomingManager().handlePacketReceived(packet: data)
     }
 
     // MARK: - Stream lifecycle
@@ -187,13 +201,14 @@ final class DataStreams: NSObject, @unchecked Sendable, Loggable {
     /// on a reader that will never finish would otherwise stall its topic's ordered queue. Handler
     /// registrations survive, so streams arriving after a reconnect are still handled.
     func reset() {
-        incoming.abortAllStreams()
+        // No-op if the incoming manager was never created (no packets received): nothing is open.
+        _incoming.copy()?.abortAllStreams()
     }
 
     /// Fails open incoming streams sent by `identity` (they disconnected mid-send), so their readers
     /// throw and their handlers return instead of hanging.
     func closeStreams(from identity: Participant.Identity) {
-        incoming.abortStreamsFrom(identity: identity.stringValue)
+        _incoming.copy()?.abortStreamsFrom(identity: identity.stringValue)
     }
 
     // MARK: - Stream open dispatch (called from the incoming delegate)

Tests/LiveKitCoreTests/DataStream/StreamOptionsTests.swift

@@ -59,5 +59,9 @@ struct StreamOptionsTests {
         #expect(DataStreamOptions(maxPayloadSize: 1000).maxPayloadSize == 1000)
         #expect(RoomOptions().dataStreamOptions.maxPayloadSize == nil)
         #expect(RoomOptions(dataStreamOptions: DataStreamOptions(maxPayloadSize: 42)).dataStreamOptions.maxPayloadSize == 42)
+        // Objective-C accessor mirrors the Swift `Int?`.
+        #expect(DataStreamOptions().maxPayloadSizeNumber == nil)
+        #expect(DataStreamOptions(maxPayloadSize: 1000).maxPayloadSizeNumber == 1000)
+        #expect(DataStreamOptions(maxPayloadSizeNumber: 1000).maxPayloadSize == 1000)
     }
 }

Tests/LiveKitObjCTests/DataStreamOptionsObjCTests.m

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2026 LiveKit
+ *
+ * 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.
+ */
+
+// Cross-ref: Tests/LiveKitCoreTests/DataStream/StreamOptionsTests.swift
+//
+// Guards that DataStreamOptions stays usable from Objective-C (it's a class, not a struct, so it
+// bridges) and that adding it to RoomOptions did not drop RoomOptions's Objective-C initializer.
+
+@import XCTest;
+@import LiveKit;
+
+@interface DataStreamOptionsObjCTests : XCTestCase
+@end
+
+@implementation DataStreamOptionsObjCTests
+
+- (void)testDataStreamOptionsConstructibleFromObjC {
+    DataStreamOptions *defaults = [[DataStreamOptions alloc] init];
+    XCTAssertNil(defaults.maxPayloadSizeNumber);
+
+    DataStreamOptions *capped = [[DataStreamOptions alloc] initWithMaxPayloadSizeNumber:@1024];
+    XCTAssertEqualObjects(capped.maxPayloadSizeNumber, @1024);
+
+    XCTAssertEqualObjects(capped, [[DataStreamOptions alloc] initWithMaxPayloadSizeNumber:@1024]);
+    XCTAssertNotEqualObjects(capped, defaults);
+}
+
+- (void)testRoomOptionsRetainsObjCInitializerWithDataStreamOptions {
+    // Adding a value type here would have dropped RoomOptions's Objective-C initializer entirely
+    // (Swift won't export an init that takes an ObjC-unrepresentable parameter). Because
+    // DataStreamOptions is a class, the designated initializer — which takes `dataStreamOptions:` —
+    // must remain callable from Objective-C. The sub-option parameters have their own
+    // ObjC-unavailable initializers (pre-existing), so we assert the selector exists rather than
+    // invoking it.
+    SEL initializer = @selector(initWithDefaultCameraCaptureOptions:
+                                defaultScreenShareCaptureOptions:
+                                defaultAudioCaptureOptions:
+                                defaultVideoPublishOptions:
+                                defaultAudioPublishOptions:
+                                defaultDataPublishOptions:
+                                dataStreamOptions:
+                                adaptiveStream:
+                                dynacast:
+                                stopLocalTrackOnUnpublish:
+                                suspendLocalVideoTracksInBackground:
+                                e2eeOptions:
+                                encryptionOptions:
+                                reportRemoteTrackStatistics:
+                                singlePeerConnection:);
+    XCTAssertTrue([RoomOptions instancesRespondToSelector:initializer],
+                  @"RoomOptions must keep its Objective-C initializer accepting dataStreamOptions:");
+}
+
+@end