Room dependency stages

.changes/connection-scoped-data-channel-crypto

@@ -0,0 +1 @@
+patch type="fixed" "Data channel crypto is now installed uniformly and scoped to the connection: rooms using legacy E2EEOptions can decrypt incoming encrypted packets, a reused Room no longer inherits the previous connection's released crypto manager, and the negotiated data channel max-message-size resets between sessions"

Sources/LiveKit/Core/DataChannelPair.swift

@@ -516,6 +516,8 @@ class DataChannelPair: NSObject, @unchecked Sendable, Loggable {
             $0.reliableReceivedState.removeAll()
             $0.lossy = nil
             $0.wasReset = true
+            // Negotiated per session (from the SDP answer); the next session must not inherit it.
+            $0.maxMessageSize = Self.defaultMaxMessageSize
             return result
         }
 

Sources/LiveKit/Core/Room+Engine.swift

@@ -46,10 +46,12 @@ extension Room {
 
         await _state.transport?.close()
 
-        // Reset publish state
-        _state.mutate {
-            $0.transport = nil
+        // Retire the join only after the transports are down: teardown gates (e.g. the data-track
+        // publish gate) re-arm off the closing channels, so the "no transport" window that
+        // in-flight callers observe must not open before they do.
+        _ = _state.mutate {
             $0.hasPublished = false
+            return $0.stage.retireJoin()
         }
     }
 
@@ -123,7 +125,6 @@ extension Room {
 // MARK: - Internal
 
 extension Room {
-    // swiftlint:disable:next function_body_length
     func configureTransports(connectResponse: SignalClient.ConnectResponse, singlePeerConnection: Bool) async throws {
         func makeConfiguration() -> LKRTCConfiguration {
             let connectOptions = _state.connectOptions
@@ -155,66 +156,22 @@ extension Room {
         if case let .join(joinResponse) = connectResponse {
             log("Configuring transports with JOIN response...")
 
-            guard _state.transport == nil else {
-                log("Transports are already configured")
-                return
-            }
-
-            let isSinglePC = singlePeerConnection
-            let isSubscriberPrimary = isSinglePC ? false : joinResponse.subscriberPrimary
-            log("subscriberPrimary: \(isSubscriberPrimary), singlePeerConnection: \(isSinglePC)")
-
-            // Publisher always created; is primary in single PC mode
-            let publisher = try Transport(config: rtcConfiguration,
-                                          target: .publisher,
-                                          primary: isSinglePC || !isSubscriberPrimary,
-                                          singlePCMode: isSinglePC,
-                                          delegate: self)
-
-            await publisher.set { [weak self] offer, offerId in
-                guard let self else { return }
-                log("Publisher onOffer with offerId: \(offerId), sdp: \(offer.sdp)")
-                try await signalClient.send(offer: offer, offerId: offerId)
-                connectSpan?.record("offer_sent")
+            guard case let .connecting(connection) = _state.stage else {
+                log("Received JOIN without a connecting stage", .error)
+                throw LiveKitError(.invalidState, message: "Received JOIN without a connecting stage")
             }
 
-            // data over pub channel for backwards compatibility
-
-            let reliableDataChannel = await publisher.dataChannel(for: LKRTCDataChannel.Labels.reliable,
-                                                                  configuration: RTC.createDataChannelConfiguration())
-
-            let lossyDataChannel = await publisher.dataChannel(for: LKRTCDataChannel.Labels.lossy,
-                                                               configuration: RTC.createDataChannelConfiguration(ordered: false, maxRetransmits: 0))
-
-            publisherDataChannel.set(reliable: reliableDataChannel)
-            publisherDataChannel.set(lossy: lossyDataChannel)
-
-            // Data track channel (unordered, unreliable — DTP handles its own sequencing). Hand it
-            // to the session-scoped data track subsystem (created at connect; persists reconnects).
-            let dataTrackChannel = await publisher.dataChannel(for: LKRTCDataChannel.Labels.dataTrack,
-                                                               configuration: RTC.createDataChannelConfiguration(ordered: false, maxRetransmits: 0))
-            if let dataTrackChannel { dataTracks?.setPublisherChannel(dataTrackChannel) }
-
-            log("dataChannel.\(String(describing: reliableDataChannel?.label)) : \(String(describing: reliableDataChannel?.channelId))")
-            log("dataChannel.\(String(describing: lossyDataChannel?.label)) : \(String(describing: lossyDataChannel?.channelId))")
-            log("dataChannel.\(String(describing: dataTrackChannel?.label)) : \(String(describing: dataTrackChannel?.channelId))")
-
-            let subscriber = isSinglePC ? nil : try Transport(config: rtcConfiguration,
-                                                              target: .subscriber,
-                                                              primary: isSubscriberPrimary,
-                                                              delegate: self)
-
-            let transport: TransportMode = if let subscriber, isSubscriberPrimary {
-                .subscriberPrimary(publisher: publisher, subscriber: subscriber)
-            } else if let subscriber {
-                .publisherPrimary(publisher: publisher, subscriber: subscriber)
-            } else {
-                .publisherOnly(publisher: publisher)
+            let join = try await JoinDependencies.make(room: self,
+                                                       connection: connection,
+                                                       joinResponse: joinResponse,
+                                                       rtcConfiguration: rtcConfiguration,
+                                                       singlePeerConnection: singlePeerConnection)
+            do {
+                try _state.mutate { try $0.stage.join(join) }
+            } catch {
+                await join.transport.close()
+                throw error
             }
-            _state.mutate { $0.transport = transport }
-
-            log("[Connect] Fast publish enabled: \(joinResponse.fastPublish ? "true" : "false")")
-
         } else if case let .reconnect(reconnectResponse) = connectResponse {
             log("[Connect] Configuring transports with RECONNECT response...")
             try await _state.transport?.set(configuration: rtcConfiguration)

Sources/LiveKit/Core/Room+EngineDelegate.swift

@@ -56,8 +56,6 @@ extension Room {
                 // Re-connecting
                 delegates.notify { $0.roomIsReconnecting?(self) }
             } else if case .disconnected = state.connectionState {
-                // Clear out e2eeManager instance
-                e2eeManager = nil
                 // Disconnected
                 if case .connecting = oldState.connectionState {
                     delegates.notify { $0.room?(self, didFailToConnectWithError: oldState.disconnectError) }

Sources/LiveKit/Core/Room.swift

@@ -93,11 +93,15 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable {
 
     // MARK: - Internal
 
-    private let _e2eeManager = StateSync<E2EEManager?>(nil)
-
+    /// The E2EE manager for the current connection.
+    ///
+    /// Derived from ``RoomOptions/e2eeOptions`` or ``RoomOptions/encryptionOptions`` during
+    /// ``connect(url:token:connectOptions:roomOptions:)``, carried across reconnects, and
+    /// released when the connection ends. Setting it takes effect only while a connection
+    /// exists, and lasts until that connection ends.
     public var e2eeManager: E2EEManager? {
-        get { _e2eeManager.copy() }
-        set { _e2eeManager.mutate { $0 = newValue } }
+        get { _state.stage.connection?.e2ee.copy() }
+        set { _state.stage.connection?.e2ee.mutate { $0 = newValue } }
     }
 
     /// Enables or disables end-to-end encryption on this Room.
@@ -139,10 +143,7 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable {
 
     // MARK: - Data Tracks
 
-    // The data track subsystem (managers, channels, signal/packet routing) behind one reference,
-    // created on connect and reset atomically on teardown — keeps it off the Room's surface.
-    let _dataTracks = StateSync<DataTracks?>(nil)
-    var dataTracks: DataTracks? { _dataTracks.copy() }
+    var dataTracks: DataTracks? { _state.stage.connection?.dataTracks }
 
     // MARK: - PreConnect
 
@@ -197,7 +198,11 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable {
         var disconnectError: LiveKitError?
         var hasPublished: Bool = false
 
-        var transport: TransportMode?
+        // Dependency plane: which per-connection / per-join subsystems exist. Payloads are staged
+        // and retired only through its transitions (connect / configureTransports / cleanUp).
+        var stage: DependencyStage = .idle
+
+        var transport: TransportMode? { stage.join?.transport }
 
         // Timing
         var connectSpan: Span?
@@ -406,35 +411,26 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable {
         // dropped in the gap between init and connect. Idempotent across reconnects.
         await setupRpc()
 
-        // enable E2EE
-        if let e2eeOptions = state.roomOptions.e2eeOptions {
-            e2eeManager = E2EEManager(e2eeOptions: e2eeOptions)
-            e2eeManager!.setup(room: self)
-        } else if let encryptionOptions = state.roomOptions.encryptionOptions {
-            e2eeManager = E2EEManager(options: encryptionOptions)
-            e2eeManager!.setup(room: self)
-
-            subscriberDataChannel.set(e2eeManager: e2eeManager)
-            publisherDataChannel.set(e2eeManager: e2eeManager)
-        } else {
-            e2eeManager = nil
-
-            subscriberDataChannel.set(e2eeManager: nil)
-            publisherDataChannel.set(e2eeManager: nil)
-        }
-
-        // Create the data track subsystem once per session, before transports. It's session-scoped
-        // — kept across reconnects (only its channels are swapped) so published tracks can be
-        // republished — and torn down on real disconnect (see cleanUp).
-        setupDataTracks()
+        // Connection-scoped subsystems (data tracks, the E2EE manager derived from the room
+        // options): carried across full reconnects, released on disconnect.
+        let dependencies = ConnectionDependencies(room: self, roomOptions: state.roomOptions)
 
-        _state.mutate {
+        try _state.mutate {
+            try $0.stage.begin(dependencies)
             $0.connectSpan = sharedTracing.beginSpan("connect")
             $0.providedUrl = providedUrl
             $0.token = token
             $0.connectionState = .connecting
         }
 
+        // E2EE wiring: delegate registration and cryptors, then the pairs' manager — installed
+        // for every options branch: sending consults the manager's data-channel toggle, but
+        // receiving needs the manager regardless — a frame-only (legacy e2eeOptions) room must
+        // still decrypt packets from participants publishing with data-channel encryption enabled.
+        e2eeManager?.setup(room: self)
+        subscriberDataChannel.set(e2eeManager: e2eeManager)
+        publisherDataChannel.set(e2eeManager: e2eeManager)
+
         var nextUrl = providedUrl
         var nextRegion: RegionInfo?
         let regionManager = await regionManager(for: providedUrl)
@@ -617,12 +613,31 @@ extension Room {
         // stats collection from accessing destroyed WebRTC channels.
         cancelTimers()
         await cleanUpRTC(withError: disconnectError)
-        cleanUpDataTracks(isFullReconnect: isFullReconnect)
+        var retiredConnection: ConnectionDependencies?
+        if isFullReconnect {
+            // Data tracks are connection-scoped: across a full reconnect only their transport
+            // channels are swapped so published tracks can be republished.
+            dataTracks?.handleTransportsTeardown()
+            // Remote data tracks outlive the reconnect — the subsystem re-attaches them to the
+            // recreated participants. Detach them here, before `cleanUpParticipants` reports an
+            // unpublish for tracks that were never unpublished.
+            for participant in _state.remoteParticipants.values {
+                participant.detachDataTracks()
+            }
+        } else {
+            retiredConnection = _state.mutate { $0.stage.end() }.connection
+        }
         await cleanUpParticipants(isFullReconnect: isFullReconnect)
 
-        // Cleanup for E2EE
-        if let e2eeManager {
-            e2eeManager.cleanUp(isFullReconnect: isFullReconnect)
+        // E2EE: a full reconnect keeps the manager and releases only the frame cryptors (they
+        // rebuild through the publish/subscribe delegate hooks as tracks return); connection
+        // teardown releases the retired manager and detaches the pairs' reference.
+        if isFullReconnect {
+            e2eeManager?.cleanUp(isFullReconnect: true)
+        } else {
+            retiredConnection?.tearDown()
+            subscriberDataChannel.set(e2eeManager: nil)
+            publisherDataChannel.set(e2eeManager: nil)
         }
 
         // Reset state
@@ -638,6 +653,7 @@ extension Room {
                 isReconnectingWithMode: $0.isReconnectingWithMode,
                 connectionState: $0.connectionState,
                 reconnectTask: $0.reconnectTask,
+                stage: $0.stage,
             ) : State(
                 connectOptions: $0.connectOptions,
                 roomOptions: $0.roomOptions,