From 65a0698ace1d3f7e4e9c9844187577a97f9891ce Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?= <86720177+pblazej@users.noreply.github.com> Date: Mon, 24 Aug 2026 15:56:52 +0200 Subject: [PATCH 1/4] refactor(room): stage connection- and join-scoped dependencies in a typed plane Room's per-connection and per-join subsystems were spread across three conventions: the optional State.transport field, the _dataTracks side-slot, and setup/teardown extensions ordered by comments and an isFullReconnect flag. They now live in stage payloads inside Room.State: idle -> connecting(ConnectionDependencies) -> connected(JoinDependencies) - ConnectionDependencies (data tracks) is created by connect() and carried across a full reconnect; JoinDependencies (transports, their channels) is built from a JOIN response by one factory and retired on teardown. A join is constructible only from its connection, so cross-tier construction order is compiler-checked, and stage-gated storage replaces the optional fields: transports exist iff the stage is .connected. - Payloads are staged and retired only through the stage transitions, which return the retired payload for teardown. Because the stage lives beside the data fields in the one StateSync, a transition and its state reset are a single atomic mutation. - configureTransports' silent "transports are already configured" no-op becomes a failed transition: a duplicate JOIN now throws instead of being swallowed. Read sites are unchanged (State.transport and Room.dataTracks forward into the stage). Verified with RoomTests, RoomStateTests, DataTrackLifecycleTests, and the new DependencyStageTests against a local server. Co-Authored-By: Claude Fable 5 --- .../Core/Room+DataTrackLifecycle.swift | 39 ---- Sources/LiveKit/Core/Room+Engine.swift | 79 ++----- Sources/LiveKit/Core/Room.swift | 35 +++- Sources/LiveKit/Core/RoomDependencies.swift | 197 ++++++++++++++++++ .../Room/DependencyStageTests.swift | 87 ++++++++ 5 files changed, 326 insertions(+), 111 deletions(-) delete mode 100644 Sources/LiveKit/Core/Room+DataTrackLifecycle.swift create mode 100644 Sources/LiveKit/Core/RoomDependencies.swift create mode 100644 Tests/LiveKitCoreTests/Room/DependencyStageTests.swift diff --git a/Sources/LiveKit/Core/Room+DataTrackLifecycle.swift b/Sources/LiveKit/Core/Room+DataTrackLifecycle.swift deleted file mode 100644 index b5119d9af..000000000 --- a/Sources/LiveKit/Core/Room+DataTrackLifecycle.swift +++ /dev/null @@ -1,39 +0,0 @@ -/* - * 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. - */ - -import Foundation - -extension Room { - func setupDataTracks() { - _dataTracks.mutate { $0 = DataTracks(room: self) } - } - - func cleanUpDataTracks(isFullReconnect: Bool = false) { - // Session-scoped: keep the subsystem across a full reconnect so its managers can republish; - // tear it down only on a real disconnect. - guard !isFullReconnect else { - 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() - } - return - } - _dataTracks.mutate { $0 = nil } - } -} diff --git a/Sources/LiveKit/Core/Room+Engine.swift b/Sources/LiveKit/Core/Room+Engine.swift index 55b8084c1..16b9db58d 100644 --- a/Sources/LiveKit/Core/Room+Engine.swift +++ b/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) diff --git a/Sources/LiveKit/Core/Room.swift b/Sources/LiveKit/Core/Room.swift index 7b91dc44a..16405a14d 100644 --- a/Sources/LiveKit/Core/Room.swift +++ b/Sources/LiveKit/Core/Room.swift @@ -139,10 +139,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(nil) - var dataTracks: DataTracks? { _dataTracks.copy() } + var dataTracks: DataTracks? { _state.stage.connection?.dataTracks } // MARK: - PreConnect @@ -197,7 +194,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? @@ -423,12 +424,11 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable { 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: carried across full reconnects, released on disconnect. + let dependencies = ConnectionDependencies(room: self) - _state.mutate { + try _state.mutate { + try $0.stage.begin(dependencies) $0.connectSpan = sharedTracing.beginSpan("connect") $0.providedUrl = providedUrl $0.token = token @@ -617,7 +617,19 @@ extension Room { // stats collection from accessing destroyed WebRTC channels. cancelTimers() await cleanUpRTC(withError: disconnectError) - cleanUpDataTracks(isFullReconnect: isFullReconnect) + 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 { + _ = _state.mutate { $0.stage.end() } + } await cleanUpParticipants(isFullReconnect: isFullReconnect) // Cleanup for E2EE @@ -638,6 +650,7 @@ extension Room { isReconnectingWithMode: $0.isReconnectingWithMode, connectionState: $0.connectionState, reconnectTask: $0.reconnectTask, + stage: $0.stage, ) : State( connectOptions: $0.connectOptions, roomOptions: $0.roomOptions, diff --git a/Sources/LiveKit/Core/RoomDependencies.swift b/Sources/LiveKit/Core/RoomDependencies.swift new file mode 100644 index 000000000..462a45200 --- /dev/null +++ b/Sources/LiveKit/Core/RoomDependencies.swift @@ -0,0 +1,197 @@ +/* + * 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. + */ + +import Foundation + +internal import LiveKitWebRTC + +// MARK: - Connection dependencies + +/// Subsystems scoped to one connection: created by `Room.connect()`, carried across a full +/// reconnect, released on disconnect. +/// +/// Membership in this type is the survival policy: a full reconnect retires the +/// ``JoinDependencies`` built on top of it and keeps this value; a disconnect retires both. +final class ConnectionDependencies: Sendable { + /// The data track subsystem (managers, channels, signal/packet routing) behind one reference. + /// Across a full reconnect only its transport channels are swapped so published tracks can be + /// republished. + let dataTracks: DataTracks + + init(room: Room) { + dataTracks = DataTracks(room: room) + } +} + +extension ConnectionDependencies: Equatable { + static func == (lhs: ConnectionDependencies, rhs: ConnectionDependencies) -> Bool { lhs === rhs } +} + +// MARK: - Join dependencies + +/// Subsystems scoped to one server JOIN: created from a JOIN response, retired on full reconnect +/// or disconnect, carried across a quick (resume) reconnect. +/// +/// Constructible only through ``make(room:connection:joinResponse:rtcConfiguration:singlePeerConnection:)`` +/// from its predecessor — a join without a connection cannot be represented, and a re-join re-runs +/// the factory by construction. +final class JoinDependencies: Sendable { + /// The connection this join was established on. Join implies connection, by construction. + let connection: ConnectionDependencies + let transport: TransportMode + + private init(connection: ConnectionDependencies, transport: TransportMode) { + self.connection = connection + self.transport = transport + } + + /// Builds the join-tier graph from a JOIN response: transports, their data channels, and the + /// hand-off of the data track channel to the connection-scoped subsystem. + static func make(room: Room, + connection: ConnectionDependencies, + joinResponse: Livekit_JoinResponse, + rtcConfiguration: LKRTCConfiguration, + singlePeerConnection: Bool) async throws -> JoinDependencies + { + let isSinglePC = singlePeerConnection + let isSubscriberPrimary = isSinglePC ? false : joinResponse.subscriberPrimary + room.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: room) + + await publisher.set { [weak room] offer, offerId in + guard let room else { return } + room.log("Publisher onOffer with offerId: \(offerId), sdp: \(offer.sdp)") + try await room.signalClient.send(offer: offer, offerId: offerId) + room.connectSpan?.record("offer_sent") + } + + // 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)) + + room.publisherDataChannel.set(reliable: reliableDataChannel) + room.publisherDataChannel.set(lossy: lossyDataChannel) + + // Data track channel (unordered, unreliable — DTP handles its own sequencing). + let dataTrackChannel = await publisher.dataChannel(for: LKRTCDataChannel.Labels.dataTrack, + configuration: RTC.createDataChannelConfiguration(ordered: false, maxRetransmits: 0)) + if let dataTrackChannel { connection.dataTracks.setPublisherChannel(dataTrackChannel) } + + room.log("dataChannel.\(String(describing: reliableDataChannel?.label)) : \(String(describing: reliableDataChannel?.channelId))") + room.log("dataChannel.\(String(describing: lossyDataChannel?.label)) : \(String(describing: lossyDataChannel?.channelId))") + room.log("dataChannel.\(String(describing: dataTrackChannel?.label)) : \(String(describing: dataTrackChannel?.channelId))") + + let subscriber = isSinglePC ? nil : try Transport(config: rtcConfiguration, + target: .subscriber, + primary: isSubscriberPrimary, + delegate: room) + + 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) + } + + room.log("[Connect] Fast publish enabled: \(joinResponse.fastPublish ? "true" : "false")") + + return JoinDependencies(connection: connection, transport: transport) + } +} + +extension JoinDependencies: Equatable { + static func == (lhs: JoinDependencies, rhs: JoinDependencies) -> Bool { lhs === rhs } +} + +// MARK: - Stage + +/// The dependency plane of a ``Room``: which per-connection / per-join subsystems currently exist. +/// +/// Stage payloads form an initializer chain (a join is constructible only from a connection), so +/// construction order across tiers is a compile-time property, and stage-gated storage replaces +/// optional fields: transports exist if and only if the stage is `.connected`. +/// +/// Lives inside `Room.State`, so a stage transition and its data-tier reset are one atomic +/// mutation under the same lock. +enum DependencyStage: Equatable { + case idle + case connecting(ConnectionDependencies) + case connected(JoinDependencies) +} + +extension DependencyStage { + var connection: ConnectionDependencies? { + switch self { + case .idle: nil + case let .connecting(connection): connection + case let .connected(join): join.connection + } + } + + var join: JoinDependencies? { + if case let .connected(join) = self { + join + } else { nil } + } + + // MARK: - Transitions (the only paths on which stage payloads are staged or retired) + + /// `connect()`: idle → connecting. Throws if a previous session is still staged — + /// `cleanUp()` must have retired it first. + mutating func begin(_ connection: ConnectionDependencies) throws { + guard case .idle = self else { + throw LiveKitError(.invalidState, message: "Cannot begin a connection while one is staged") + } + self = .connecting(connection) + } + + /// JOIN response applied: connecting → connected. Throws on a duplicate JOIN or a join built + /// against a retired connection. + mutating func join(_ join: JoinDependencies) throws { + guard case let .connecting(connection) = self, join.connection === connection else { + throw LiveKitError(.invalidState, message: "JOIN is only valid for the currently connecting connection") + } + self = .connected(join) + } + + /// Full reconnect (and the join half of a disconnect): retires the join, keeps the connection. + /// The caller owns closing the returned join's resources — detach, then destroy. + mutating func retireJoin() -> JoinDependencies? { + guard case let .connected(join) = self else { return nil } + self = .connecting(join.connection) + return join + } + + /// Disconnect: retires everything. The join is expected to have been retired and closed + /// already (`cleanUpRTC`); it is returned here so no path can drop one silently. + mutating func end() -> (join: JoinDependencies?, connection: ConnectionDependencies?) { + let join = retireJoin() + guard case let .connecting(connection) = self else { return (join, nil) } + self = .idle + return (join, connection) + } +} diff --git a/Tests/LiveKitCoreTests/Room/DependencyStageTests.swift b/Tests/LiveKitCoreTests/Room/DependencyStageTests.swift new file mode 100644 index 000000000..61a38f2c7 --- /dev/null +++ b/Tests/LiveKitCoreTests/Room/DependencyStageTests.swift @@ -0,0 +1,87 @@ +/* + * 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. + */ + +import Foundation +@testable import LiveKit +import Testing + +/// Verifies the transition contract of the dependency stage: payloads are staged and retired +/// only through the transitions, a connection survives a join retirement, and ending the stage +/// hands every payload back exactly once. +struct DependencyStageTests { + @Test func beginStagesConnection() throws { + let room = Room() + let connection = ConnectionDependencies(room: room) + + var stage = DependencyStage.idle + try stage.begin(connection) + + #expect(stage.connection === connection) + #expect(stage.join == nil) + } + + @Test func beginWhileStagedThrows() throws { + let room = Room() + + var stage = DependencyStage.idle + try stage.begin(ConnectionDependencies(room: room)) + + #expect(throws: LiveKitError.self) { + try stage.begin(ConnectionDependencies(room: room)) + } + } + + @Test func retireJoinWithoutJoinKeepsConnection() throws { + let room = Room() + let connection = ConnectionDependencies(room: room) + + var stage = DependencyStage.idle + try stage.begin(connection) + + #expect(stage.retireJoin() == nil) + #expect(stage.connection === connection) + } + + @Test func endRetiresConnection() throws { + let room = Room() + let connection = ConnectionDependencies(room: room) + + var stage = DependencyStage.idle + try stage.begin(connection) + + let retired = stage.end() + #expect(retired.connection === connection) + #expect(retired.join == nil) + #expect(stage == .idle) + #expect(stage.connection == nil) + } + + @Test func endFromIdleRetiresNothing() { + var stage = DependencyStage.idle + + let retired = stage.end() + #expect(retired.connection == nil) + #expect(retired.join == nil) + #expect(stage == .idle) + } + + @Test func freshRoomStartsIdle() { + let room = Room() + #expect(room._state.stage == .idle) + #expect(room.dataTracks == nil) + #expect(room._state.transport == nil) + } +} From 047a6e96e9328beb1785cb35a8ec097cf09e27d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?= <86720177+pblazej@users.noreply.github.com> Date: Mon, 24 Aug 2026 16:02:14 +0200 Subject: [PATCH 2/4] fix(e2ee): scope the pairs' data channel crypto to the connection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The data channel pairs' E2EE wiring had three connection-boundary leaks: - connect()'s legacy e2eeOptions branch never installed the manager on the pairs, so a frame-only encrypted room could not decrypt incoming packets from participants publishing with data-channel encryption enabled — and, worse, the pairs silently kept whatever manager a previous connection had installed. The manager is now installed uniformly in all three options branches; sending still consults isDataChannelEncryptionEnabled, which stays false for legacy options. - Nothing cleared the pairs' manager reference on disconnect; a reused Room inherited the released manager of the previous connection. cleanUp now detaches it at connection end (kept across a full reconnect, where the manager survives by design). - The negotiated max-message-size survived reset() into the next session, applying a stale ceiling until the new SDP answer arrived. reset() restores the default. Regression-tested by EncryptedDataChannelTests/encryptionSurvivesFullReconnect (encrypted send after a full reconnect) and DataChannelPairTests/resetRestoresDefaultMaxMessageSize. Co-Authored-By: Claude Fable 5 --- .../connection-scoped-data-channel-crypto | 1 + Sources/LiveKit/Core/DataChannelPair.swift | 2 + Sources/LiveKit/Core/Room.swift | 22 ++++++--- .../DataChannel/DataChannelPairTests.swift | 25 ++++++++++ .../EncryptedDataChannelTests.swift | 49 +++++++++++++++++++ 5 files changed, 91 insertions(+), 8 deletions(-) create mode 100644 .changes/connection-scoped-data-channel-crypto diff --git a/.changes/connection-scoped-data-channel-crypto b/.changes/connection-scoped-data-channel-crypto new file mode 100644 index 000000000..5c4e9d277 --- /dev/null +++ b/.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" diff --git a/Sources/LiveKit/Core/DataChannelPair.swift b/Sources/LiveKit/Core/DataChannelPair.swift index 0f34a779e..5c7828092 100644 --- a/Sources/LiveKit/Core/DataChannelPair.swift +++ b/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 } diff --git a/Sources/LiveKit/Core/Room.swift b/Sources/LiveKit/Core/Room.swift index 16405a14d..b837e9c6c 100644 --- a/Sources/LiveKit/Core/Room.swift +++ b/Sources/LiveKit/Core/Room.swift @@ -410,19 +410,19 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable { // 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) } + e2eeManager?.setup(room: self) + + // Install on the pairs for every 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. + subscriberDataChannel.set(e2eeManager: e2eeManager) + publisherDataChannel.set(e2eeManager: e2eeManager) // Connection-scoped subsystems: carried across full reconnects, released on disconnect. let dependencies = ConnectionDependencies(room: self) @@ -636,6 +636,12 @@ extension Room { if let e2eeManager { e2eeManager.cleanUp(isFullReconnect: isFullReconnect) } + if !isFullReconnect { + // The pairs' crypto reference is connection-scoped: a reused Room must not inherit + // a released manager from the previous connection. + subscriberDataChannel.set(e2eeManager: nil) + publisherDataChannel.set(e2eeManager: nil) + } // Reset state _state.mutate { diff --git a/Tests/LiveKitCoreTests/DataChannel/DataChannelPairTests.swift b/Tests/LiveKitCoreTests/DataChannel/DataChannelPairTests.swift index 7e502e7b0..484859860 100644 --- a/Tests/LiveKitCoreTests/DataChannel/DataChannelPairTests.swift +++ b/Tests/LiveKitCoreTests/DataChannel/DataChannelPairTests.swift @@ -47,6 +47,31 @@ struct DataChannelPairTests { } throws: { ($0 as? LiveKitError)?.type == .invalidState } } + @Test func resetRestoresDefaultMaxMessageSize() async throws { + let pair = DataChannelPair() + pair.set(maxMessageSize: 16) + + // The size check runs against the serialized packet at send time, before any channel exists. + let oversized = Livekit_UserPacket.with { $0.payload = Data(repeating: 0, count: 64) } + await #expect { + try await pair.send(userPacket: oversized, kind: .reliable) + } throws: { ($0 as? LiveKitError)?.type == .invalidParameter } + + pair.reset(throwing: nil) + + // The negotiated ceiling is session-scoped: after reset the same packet clears the + // restored default and parks awaiting channels, instead of failing the stale size check. + let sendTask = Task { + try await pair.send(userPacket: oversized, kind: .reliable) + } + try await Task.sleep(nanoseconds: 50_000_000) + + pair.reset(throwing: LiveKitError(.invalidState, message: "drain")) + await #expect { + try await sendTask.value + } throws: { ($0 as? LiveKitError)?.type == .invalidState } + } + @Test func resetWithNilErrorFailsParkedSendsAsCancelled() async throws { let pair = DataChannelPair() diff --git a/Tests/LiveKitCoreTests/DataChannel/EncryptedDataChannelTests.swift b/Tests/LiveKitCoreTests/DataChannel/EncryptedDataChannelTests.swift index 43d5eb5ea..57e78b111 100644 --- a/Tests/LiveKitCoreTests/DataChannel/EncryptedDataChannelTests.swift +++ b/Tests/LiveKitCoreTests/DataChannel/EncryptedDataChannelTests.swift @@ -217,6 +217,55 @@ import LiveKitWebRTC } } + /// Regression: the data-packet cryptor is connection-tier and must + /// survive a full reconnect. It used to be destroyed by full-reconnect + /// cleanup and re-created only inside connect() — which a full reconnect + /// never re-enters — leaving every subsequent encrypted send throwing + /// "Cryptor is nil". + @Test func encryptionSurvivesFullReconnect() async throws { + let testMessage = "Hello, encrypted world, after a full reconnect!" + let testData = try #require(testMessage.data(using: .utf8)) + + let sharedKey = "shared-key-\(UUID().uuidString)" + let options = KeyProviderOptions(sharedKey: true, keyDerivationAlgorithm: .hkdf) + let senderKeyProvider = BaseKeyProvider(options: options) + let receiverKeyProvider = BaseKeyProvider(options: options) + senderKeyProvider.setKey(key: sharedKey) + receiverKeyProvider.setKey(key: sharedKey) + + try await confirmation("Encrypted data received after full reconnect") { confirm in + self._receivedData.mutate { $0 = Data() } + self._onDataReceived.mutate { $0 = { confirm() } } + + try await TestEnvironment.withRooms(encryptedRooms(sender: senderKeyProvider, receiver: receiverKeyProvider)) { rooms in + let sender = rooms[0] + + try await sender.debug_simulate(scenario: .fullReconnect) + #expect(sender.connectionState == .connected) + + // The new session repopulates the roster from its own join response. + var attempts = 0 + while sender.remoteParticipants.isEmpty, attempts < 100 { + try await Task.sleep(nanoseconds: 100_000_000) + attempts += 1 + } + let remoteIdentity = try #require(sender.remoteParticipants.keys.first) + + let userPacket = Livekit_UserPacket.with { + $0.payload = testData + $0.destinationIdentities = [remoteIdentity.stringValue] + } + + try await sender.send(userPacket: userPacket, kind: .reliable) + + await self.awaitEvent() + } + + let receivedMessage = String(data: self._receivedData.copy(), encoding: .utf8) + #expect(receivedMessage == testMessage, "Encrypted message should still decrypt after the sender's full reconnect") + } + } + @Test(arguments: [KeyDerivationAlgorithm.pbkdf2, .hkdf]) func keyRatcheting(algorithm: KeyDerivationAlgorithm) async throws { let testMessage = "Hello with automatic ratcheting!" From 7a4f8f6f1319cd1ce32ce34a42065cc8079b85a2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?= <86720177+pblazej@users.noreply.github.com> Date: Mon, 24 Aug 2026 16:17:59 +0200 Subject: [PATCH 3/4] refactor(e2ee): move the E2EE manager into the connection dependencies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ConnectionDependencies now derives the manager from the room options and owns it for the connection's lifetime; the public Room.e2eeManager facade (declaration unchanged) reads and writes through the stage, so it is nil before connect and after disconnect, and the setter takes effect only while a connection exists. - Retirement is structural: stage.end() dropping the connection payload replaces the engine-delegate branch that nil'd the manager on .disconnected — a branch that also mutated a second lock from inside Room's state callback. - Connection teardown lives on the retired payload (ConnectionDependencies.tearDown), called from cleanUp at the same point the manager cleanup ran before, after participant cleanup. Co-Authored-By: Claude Fable 5 --- .../LiveKit/Core/Room+EngineDelegate.swift | 2 - Sources/LiveKit/Core/Room.swift | 59 +++++++++---------- Sources/LiveKit/Core/RoomDependencies.swift | 20 ++++++- .../Room/DependencyStageTests.swift | 32 ++++++++-- 4 files changed, 74 insertions(+), 39 deletions(-) diff --git a/Sources/LiveKit/Core/Room+EngineDelegate.swift b/Sources/LiveKit/Core/Room+EngineDelegate.swift index 9a8e91204..53409e7b8 100644 --- a/Sources/LiveKit/Core/Room+EngineDelegate.swift +++ b/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) } diff --git a/Sources/LiveKit/Core/Room.swift b/Sources/LiveKit/Core/Room.swift index b837e9c6c..0b0304fa4 100644 --- a/Sources/LiveKit/Core/Room.swift +++ b/Sources/LiveKit/Core/Room.swift @@ -93,11 +93,15 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable { // MARK: - Internal - private let _e2eeManager = StateSync(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. @@ -407,25 +411,9 @@ 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) - } else if let encryptionOptions = state.roomOptions.encryptionOptions { - e2eeManager = E2EEManager(options: encryptionOptions) - } else { - e2eeManager = nil - } - e2eeManager?.setup(room: self) - - // Install on the pairs for every 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. - subscriberDataChannel.set(e2eeManager: e2eeManager) - publisherDataChannel.set(e2eeManager: e2eeManager) - - // Connection-scoped subsystems: carried across full reconnects, released on disconnect. - let dependencies = ConnectionDependencies(room: self) + // 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) try _state.mutate { try $0.stage.begin(dependencies) @@ -435,6 +423,14 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable { $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,6 +613,7 @@ extension Room { // stats collection from accessing destroyed WebRTC channels. cancelTimers() await cleanUpRTC(withError: disconnectError) + 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. @@ -628,17 +625,17 @@ extension Room { participant.detachDataTracks() } } else { - _ = _state.mutate { $0.stage.end() } + retiredConnection = _state.mutate { $0.stage.end() }.connection } await cleanUpParticipants(isFullReconnect: isFullReconnect) - // Cleanup for E2EE - if let e2eeManager { - e2eeManager.cleanUp(isFullReconnect: isFullReconnect) - } - if !isFullReconnect { - // The pairs' crypto reference is connection-scoped: a reused Room must not inherit - // a released manager from the previous connection. + // 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) } diff --git a/Sources/LiveKit/Core/RoomDependencies.swift b/Sources/LiveKit/Core/RoomDependencies.swift index 462a45200..ca296f7bc 100644 --- a/Sources/LiveKit/Core/RoomDependencies.swift +++ b/Sources/LiveKit/Core/RoomDependencies.swift @@ -31,8 +31,26 @@ final class ConnectionDependencies: Sendable { /// republished. let dataTracks: DataTracks - init(room: Room) { + /// The E2EE manager, derived from the room options. A synchronized cell rather than a `let`: + /// the public ``Room/e2eeManager`` setter writes through it, so the value stays swappable + /// while its lifetime is the connection's. + let e2ee: StateSync + + init(room: Room, roomOptions: RoomOptions) { dataTracks = DataTracks(room: room) + let manager: E2EEManager? = if let e2eeOptions = roomOptions.e2eeOptions { + E2EEManager(e2eeOptions: e2eeOptions) + } else if let encryptionOptions = roomOptions.encryptionOptions { + E2EEManager(options: encryptionOptions) + } else { + nil + } + e2ee = StateSync(manager) + } + + /// Ordered teardown, called with the retired value after ``DependencyStage/end()``. + func tearDown() { + e2ee.copy()?.cleanUp(isFullReconnect: false) } } diff --git a/Tests/LiveKitCoreTests/Room/DependencyStageTests.swift b/Tests/LiveKitCoreTests/Room/DependencyStageTests.swift index 61a38f2c7..71f8cd2b3 100644 --- a/Tests/LiveKitCoreTests/Room/DependencyStageTests.swift +++ b/Tests/LiveKitCoreTests/Room/DependencyStageTests.swift @@ -24,7 +24,7 @@ import Testing struct DependencyStageTests { @Test func beginStagesConnection() throws { let room = Room() - let connection = ConnectionDependencies(room: room) + let connection = ConnectionDependencies(room: room, roomOptions: RoomOptions()) var stage = DependencyStage.idle try stage.begin(connection) @@ -37,16 +37,16 @@ struct DependencyStageTests { let room = Room() var stage = DependencyStage.idle - try stage.begin(ConnectionDependencies(room: room)) + try stage.begin(ConnectionDependencies(room: room, roomOptions: RoomOptions())) #expect(throws: LiveKitError.self) { - try stage.begin(ConnectionDependencies(room: room)) + try stage.begin(ConnectionDependencies(room: room, roomOptions: RoomOptions())) } } @Test func retireJoinWithoutJoinKeepsConnection() throws { let room = Room() - let connection = ConnectionDependencies(room: room) + let connection = ConnectionDependencies(room: room, roomOptions: RoomOptions()) var stage = DependencyStage.idle try stage.begin(connection) @@ -57,7 +57,7 @@ struct DependencyStageTests { @Test func endRetiresConnection() throws { let room = Room() - let connection = ConnectionDependencies(room: room) + let connection = ConnectionDependencies(room: room, roomOptions: RoomOptions()) var stage = DependencyStage.idle try stage.begin(connection) @@ -83,5 +83,27 @@ struct DependencyStageTests { #expect(room._state.stage == .idle) #expect(room.dataTracks == nil) #expect(room._state.transport == nil) + #expect(room.e2eeManager == nil) + } + + @Test func e2eeManagerScopesToConnection() throws { + let encryptionOptions = EncryptionOptions(keyProvider: BaseKeyProvider(isSharedKey: true, sharedKey: "stage-test")) + let room = Room() + + // Idle: the facade has no connection to read or write through. + room.e2eeManager = E2EEManager(options: encryptionOptions) + #expect(room.e2eeManager == nil) + + // Built outside the lock, like connect() — its construction touches Room state. + let dependencies = ConnectionDependencies(room: room, roomOptions: RoomOptions(encryptionOptions: encryptionOptions)) + try room._state.mutate { try $0.stage.begin(dependencies) } + #expect(room.e2eeManager != nil, "Staging a connection derives the manager from the room options") + + let replacement = E2EEManager(options: encryptionOptions) + room.e2eeManager = replacement + #expect(room.e2eeManager === replacement, "The setter writes through to the connection's cell") + + _ = room._state.mutate { $0.stage.end() } + #expect(room.e2eeManager == nil, "The manager is released with its connection") } } From 5ea1faa79fb95abfcb7bbdc8b31757396dd062fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?= <86720177+pblazej@users.noreply.github.com> Date: Mon, 24 Aug 2026 16:23:20 +0200 Subject: [PATCH 4/4] test(e2ee): move the reconnect regression test into a suite extension EncryptedDataChannelTests exceeded SwiftLint's type_body_length after gaining encryptionSurvivesFullReconnect; the test moves verbatim into an extension, matching the file's existing RoomDelegate extension. Co-Authored-By: Claude Fable 5 --- .../EncryptedDataChannelTests.swift | 102 +++++++++--------- 1 file changed, 53 insertions(+), 49 deletions(-) diff --git a/Tests/LiveKitCoreTests/DataChannel/EncryptedDataChannelTests.swift b/Tests/LiveKitCoreTests/DataChannel/EncryptedDataChannelTests.swift index 57e78b111..155fb7928 100644 --- a/Tests/LiveKitCoreTests/DataChannel/EncryptedDataChannelTests.swift +++ b/Tests/LiveKitCoreTests/DataChannel/EncryptedDataChannelTests.swift @@ -217,55 +217,6 @@ import LiveKitWebRTC } } - /// Regression: the data-packet cryptor is connection-tier and must - /// survive a full reconnect. It used to be destroyed by full-reconnect - /// cleanup and re-created only inside connect() — which a full reconnect - /// never re-enters — leaving every subsequent encrypted send throwing - /// "Cryptor is nil". - @Test func encryptionSurvivesFullReconnect() async throws { - let testMessage = "Hello, encrypted world, after a full reconnect!" - let testData = try #require(testMessage.data(using: .utf8)) - - let sharedKey = "shared-key-\(UUID().uuidString)" - let options = KeyProviderOptions(sharedKey: true, keyDerivationAlgorithm: .hkdf) - let senderKeyProvider = BaseKeyProvider(options: options) - let receiverKeyProvider = BaseKeyProvider(options: options) - senderKeyProvider.setKey(key: sharedKey) - receiverKeyProvider.setKey(key: sharedKey) - - try await confirmation("Encrypted data received after full reconnect") { confirm in - self._receivedData.mutate { $0 = Data() } - self._onDataReceived.mutate { $0 = { confirm() } } - - try await TestEnvironment.withRooms(encryptedRooms(sender: senderKeyProvider, receiver: receiverKeyProvider)) { rooms in - let sender = rooms[0] - - try await sender.debug_simulate(scenario: .fullReconnect) - #expect(sender.connectionState == .connected) - - // The new session repopulates the roster from its own join response. - var attempts = 0 - while sender.remoteParticipants.isEmpty, attempts < 100 { - try await Task.sleep(nanoseconds: 100_000_000) - attempts += 1 - } - let remoteIdentity = try #require(sender.remoteParticipants.keys.first) - - let userPacket = Livekit_UserPacket.with { - $0.payload = testData - $0.destinationIdentities = [remoteIdentity.stringValue] - } - - try await sender.send(userPacket: userPacket, kind: .reliable) - - await self.awaitEvent() - } - - let receivedMessage = String(data: self._receivedData.copy(), encoding: .utf8) - #expect(receivedMessage == testMessage, "Encrypted message should still decrypt after the sender's full reconnect") - } - } - @Test(arguments: [KeyDerivationAlgorithm.pbkdf2, .hkdf]) func keyRatcheting(algorithm: KeyDerivationAlgorithm) async throws { let testMessage = "Hello with automatic ratcheting!" @@ -359,6 +310,59 @@ import LiveKitWebRTC } } +// MARK: - Reconnects + +extension EncryptedDataChannelTests { + /// Regression: the data-packet cryptor is connection-tier and must + /// survive a full reconnect. It used to be destroyed by full-reconnect + /// cleanup and re-created only inside connect() — which a full reconnect + /// never re-enters — leaving every subsequent encrypted send throwing + /// "Cryptor is nil". + @Test func encryptionSurvivesFullReconnect() async throws { + let testMessage = "Hello, encrypted world, after a full reconnect!" + let testData = try #require(testMessage.data(using: .utf8)) + + let sharedKey = "shared-key-\(UUID().uuidString)" + let options = KeyProviderOptions(sharedKey: true, keyDerivationAlgorithm: .hkdf) + let senderKeyProvider = BaseKeyProvider(options: options) + let receiverKeyProvider = BaseKeyProvider(options: options) + senderKeyProvider.setKey(key: sharedKey) + receiverKeyProvider.setKey(key: sharedKey) + + try await confirmation("Encrypted data received after full reconnect") { confirm in + self._receivedData.mutate { $0 = Data() } + self._onDataReceived.mutate { $0 = { confirm() } } + + try await TestEnvironment.withRooms(encryptedRooms(sender: senderKeyProvider, receiver: receiverKeyProvider)) { rooms in + let sender = rooms[0] + + try await sender.debug_simulate(scenario: .fullReconnect) + #expect(sender.connectionState == .connected) + + // The new session repopulates the roster from its own join response. + var attempts = 0 + while sender.remoteParticipants.isEmpty, attempts < 100 { + try await Task.sleep(nanoseconds: 100_000_000) + attempts += 1 + } + let remoteIdentity = try #require(sender.remoteParticipants.keys.first) + + let userPacket = Livekit_UserPacket.with { + $0.payload = testData + $0.destinationIdentities = [remoteIdentity.stringValue] + } + + try await sender.send(userPacket: userPacket, kind: .reliable) + + await self.awaitEvent() + } + + let receivedMessage = String(data: self._receivedData.copy(), encoding: .utf8) + #expect(receivedMessage == testMessage, "Encrypted message should still decrypt after the sender's full reconnect") + } + } +} + // MARK: - RoomDelegate extension EncryptedDataChannelTests: RoomDelegate {