Reconnect errors (#845)

pblazej · web-flow · commit 5b53abf3a43b · 2026-01-09T16:31:30.000+01:00
Resolves #844 - Handles `offerId` (symptoms) - Fixes double offer on ICE restart (root cause) - Fixes deadlock, probably introduced by strict ordering in #804 (the fix is still valid though) - Handles `leaveAction` param vs legacy `canReconnect` (requires protocol v16 introduced elsewhere)
diff --git a/.changes/reconnect-errors b/.changes/reconnect-errors
@@ -0,0 +1 @@
+patch type="fixed" "Reconnect sequence stuck in failed state"
diff --git a/Sources/LiveKit/Core/Room+Engine.swift b/Sources/LiveKit/Core/Room+Engine.swift
@@ -164,10 +164,10 @@ extension Room {
                                           primary: !isSubscriberPrimary,
                                           delegate: self)
 
-            await publisher.set { [weak self] offer in
+            await publisher.set { [weak self] offer, offerId in
                 guard let self else { return }
-                log("Publisher onOffer \(offer.sdp)")
-                try await signalClient.send(offer: offer)
+                log("Publisher onOffer with offerId: \(offerId), sdp: \(offer.sdp)")
+                try await signalClient.send(offer: offer, offerId: offerId)
             }
 
             // data over pub channel for backwards compatibility
@@ -322,7 +322,13 @@ extension Room {
 
             log("[Connect] Waiting for subscriber to connect...")
             // Wait for primary transport to connect (if not already)
-            try await primaryTransportConnectedCompleter.wait(timeout: _state.connectOptions.primaryTransportConnectTimeout)
+            do {
+                try await primaryTransportConnectedCompleter.wait(timeout: _state.connectOptions.primaryTransportConnectTimeout)
+                log("[Connect] Subscriber transport connected")
+            } catch {
+                log("[Connect] Subscriber transport failed to connect, error: \(error)", .error)
+                throw error
+            }
             try Task.checkCancellation()
 
             // send SyncState before offer
@@ -334,7 +340,13 @@ extension Room {
                 // Only if published, wait for publisher to connect...
                 log("[Connect] Waiting for publisher to connect...")
                 try await publisher.createAndSendOffer(iceRestart: true)
-                try await publisherTransportConnectedCompleter.wait(timeout: _state.connectOptions.publisherTransportConnectTimeout)
+                do {
+                    try await publisherTransportConnectedCompleter.wait(timeout: _state.connectOptions.publisherTransportConnectTimeout)
+                    log("[Connect] Publisher transport connected")
+                } catch {
+                    log("[Connect] Publisher transport failed to connect, error: \(error)", .error)
+                    throw error
+                }
             }
         }
 
@@ -469,8 +481,8 @@ extension Room {
             $0.subscribe = !autoSubscribe
         }
 
-        try await signalClient.sendSyncState(answer: previousAnswer?.toPBType(),
-                                             offer: previousOffer?.toPBType(),
+        try await signalClient.sendSyncState(answer: previousAnswer?.toPBType(offerId: 0),
+                                             offer: previousOffer?.toPBType(offerId: 0),
                                              subscription: subscription,
                                              publishTracks: localParticipant.publishedTracksInfo(),
                                              dataChannels: publisherDataChannel.infos(),
diff --git a/Sources/LiveKit/Core/Room+SignalClientDelegate.swift b/Sources/LiveKit/Core/Room+SignalClientDelegate.swift
@@ -34,23 +34,32 @@ extension Room: SignalClientDelegate {
            // engine is currently connected state
            case .connected = _state.connectionState
         {
-            do {
-                try await startReconnect(reason: .websocket)
-            } catch {
-                log("Failed calling startReconnect, error: \(error)", .error)
+            Task {
+                do {
+                    try await startReconnect(reason: .websocket)
+                } catch {
+                    log("Failed calling startReconnect, error: \(error)", .error)
+                }
             }
         }
     }
 
-    func signalClient(_: SignalClient, didReceiveLeave canReconnect: Bool, reason: Livekit_DisconnectReason) async {
-        log("canReconnect: \(canReconnect), reason: \(reason)")
+    func signalClient(_: SignalClient, didReceiveLeave action: Livekit_LeaveRequest.Action, reason: Livekit_DisconnectReason) async {
+        log("action: \(action), reason: \(reason)")
 
-        if canReconnect {
-            // force .full for next reconnect
+        let error = LiveKitError.from(reason: reason)
+        switch action {
+        case .reconnect:
+            // Force .full for next reconnect
             _state.mutate { $0.nextReconnectMode = .full }
-        } else {
-            // Server indicates it's not recoverable
-            await cleanUp(withError: LiveKitError.from(reason: reason))
+            fallthrough
+        case .resume:
+            // Abort current attempt
+            await signalClient.cleanUp(withError: error)
+        case .disconnect:
+            await cleanUp(withError: error)
+        default:
+            log("Unknown leave action: \(action), ignoring", .warning)
         }
     }
 
@@ -341,17 +350,19 @@ extension Room: SignalClientDelegate {
         }
     }
 
-    func signalClient(_: SignalClient, didReceiveAnswer answer: LKRTCSessionDescription) async {
+    func signalClient(_: SignalClient, didReceiveAnswer answer: LKRTCSessionDescription, offerId: UInt32) async {
+        log("Received answer for offerId: \(offerId)")
+
         do {
             let publisher = try requirePublisher()
-            try await publisher.set(remoteDescription: answer)
+            try await publisher.set(remoteDescription: answer, offerId: offerId)
         } catch {
-            log("Failed to set remote description, error: \(error)", .error)
+            log("Failed to set remote description with offerId: \(offerId), error: \(error)", .error)
         }
     }
 
-    func signalClient(_ signalClient: SignalClient, didReceiveOffer offer: LKRTCSessionDescription) async {
-        log("Received offer, creating & sending answer...")
+    func signalClient(_ signalClient: SignalClient, didReceiveOffer offer: LKRTCSessionDescription, offerId: UInt32) async {
+        log("Received offer with offerId: \(offerId), creating & sending answer...")
 
         guard let subscriber = _state.subscriber else {
             log("Failed to send answer, subscriber is nil", .error)
@@ -362,9 +373,9 @@ extension Room: SignalClientDelegate {
             try await subscriber.set(remoteDescription: offer)
             let answer = try await subscriber.createAnswer()
             try await subscriber.set(localDescription: answer)
-            try await signalClient.send(answer: answer)
+            try await signalClient.send(answer: answer, offerId: offerId)
         } catch {
-            log("Failed to send answer with error: \(error)", .error)
+            log("Failed to send answer for offerId: \(offerId), error: \(error)", .error)
         }
     }
 
diff --git a/Sources/LiveKit/Core/SignalClient.swift b/Sources/LiveKit/Core/SignalClient.swift
@@ -129,13 +129,15 @@ actor SignalClient: Loggable {
                                      participantSid: participantSid,
                                      adaptiveStream: adaptiveStream)
 
-        if reconnectMode != nil {
-            log("[Connect] with url: \(url)")
+        let isReconnect = reconnectMode != nil
+
+        if isReconnect {
+            log("Reconnecting with url: \(url)")
         } else {
             log("Connecting with url: \(url)")
         }
 
-        _state.mutate { $0.connectionState = (reconnectMode != nil ? .reconnecting : .connecting) }
+        _state.mutate { $0.connectionState = (isReconnect ? .reconnecting : .connecting) }
 
         do {
             let socket = try await WebSocket(url: url,
@@ -293,10 +295,12 @@ private extension SignalClient {
             await _restartPingTimer()
 
         case let .answer(sd):
-            _delegate.notifyDetached { await $0.signalClient(self, didReceiveAnswer: sd.toRTCType()) }
+            let (rtcDescription, offerId) = sd.toRTCType()
+            _delegate.notifyDetached { await $0.signalClient(self, didReceiveAnswer: rtcDescription, offerId: offerId) }
 
         case let .offer(sd):
-            _delegate.notifyDetached { await $0.signalClient(self, didReceiveOffer: sd.toRTCType()) }
+            let (rtcDescription, offerId) = sd.toRTCType()
+            _delegate.notifyDetached { await $0.signalClient(self, didReceiveOffer: rtcDescription, offerId: offerId) }
 
         case let .trickle(trickle):
             guard let rtcCandidate = try? RTC.createIceCandidate(fromJsonString: trickle.candidateInit) else {
@@ -332,7 +336,7 @@ private extension SignalClient {
             _delegate.notifyDetached { await $0.signalClient(self, didUpdateRemoteMute: Track.Sid(from: mute.sid), muted: mute.muted) }
 
         case let .leave(leave):
-            _delegate.notifyDetached { await $0.signalClient(self, didReceiveLeave: leave.canReconnect, reason: leave.reason) }
+            _delegate.notifyDetached { await $0.signalClient(self, didReceiveLeave: leave.action, reason: leave.reason) }
 
         case let .streamStateUpdate(states):
             _delegate.notifyDetached { await $0.signalClient(self, didUpdateTrackStreamStates: states.streamStates) }
@@ -375,17 +379,17 @@ extension SignalClient {
 // MARK: - Send methods
 
 extension SignalClient {
-    func send(offer: LKRTCSessionDescription) async throws {
+    func send(offer: LKRTCSessionDescription, offerId: UInt32) async throws {
         let r = Livekit_SignalRequest.with {
-            $0.offer = offer.toPBType()
+            $0.offer = offer.toPBType(offerId: offerId)
         }
 
         try await _sendRequest(r)
     }
 
-    func send(answer: LKRTCSessionDescription) async throws {
+    func send(answer: LKRTCSessionDescription, offerId: UInt32) async throws {
         let r = Livekit_SignalRequest.with {
-            $0.answer = answer.toPBType()
+            $0.answer = answer.toPBType(offerId: offerId)
         }
 
         try await _sendRequest(r)
diff --git a/Sources/LiveKit/Core/Transport.swift b/Sources/LiveKit/Core/Transport.swift
@@ -21,7 +21,7 @@ internal import LiveKitWebRTC
 actor Transport: NSObject, Loggable {
     // MARK: - Types
 
-    typealias OnOfferBlock = @Sendable (LKRTCSessionDescription) async throws -> Void
+    typealias OnOfferBlock = @Sendable (LKRTCSessionDescription, UInt32) async throws -> Void
 
     // MARK: - Public
 
@@ -56,6 +56,7 @@ actor Transport: NSObject, Loggable {
     private var _reNegotiate: Bool = false
     private var _onOffer: OnOfferBlock?
     private var _isRestartingIce: Bool = false
+    private var _latestOfferId: UInt32 = 0
 
     // forbid direct access to PeerConnection
     private let _pc: LKRTCPeerConnection
@@ -110,6 +111,20 @@ actor Transport: NSObject, Loggable {
         await _iceCandidatesQueue.process(candidate, if: remoteDescription != nil && !_isRestartingIce)
     }
 
+    func set(remoteDescription sd: LKRTCSessionDescription, offerId: UInt32) async throws {
+        if signalingState != .haveLocalOffer {
+            log("Received answer with unexpected signaling state: \(signalingState), expected .haveLocalOffer", .warning)
+        }
+
+        if offerId == 0 {
+            log("Skipping validation for legacy server (missing offerId), latestOfferId: \(_latestOfferId)", .warning)
+        } else if offerId != _latestOfferId {
+            throw LiveKitError(.invalidState, message: "OfferId mismatch, expected \(_latestOfferId) but got \(offerId)")
+        }
+
+        try await set(remoteDescription: sd)
+    }
+
     func set(remoteDescription sd: LKRTCSessionDescription) async throws {
         try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, Error>) in
             _pc.setRemoteDescription(sd) { error in
@@ -157,12 +172,14 @@ actor Transport: NSObject, Loggable {
 
         // Actually negotiate
         func _negotiateSequence() async throws {
+            _latestOfferId += 1
             let offer = try await createOffer(for: constraints)
             try await set(localDescription: offer)
-            try await _onOffer(offer)
+            try await _onOffer(offer, _latestOfferId)
         }
 
         if signalingState == .haveLocalOffer, iceRestart, let sd = remoteDescription {
+            _reNegotiate = false // Clear flag to prevent double offer
             try await set(remoteDescription: sd)
             return try await _negotiateSequence()
         }
diff --git a/Sources/LiveKit/Protocols/SignalClientDelegate.swift b/Sources/LiveKit/Protocols/SignalClientDelegate.swift
@@ -21,8 +21,8 @@ internal import LiveKitWebRTC
 protocol SignalClientDelegate: AnyObject, Sendable {
     func signalClient(_ signalClient: SignalClient, didUpdateConnectionState newState: ConnectionState, oldState: ConnectionState, disconnectError: LiveKitError?) async
     func signalClient(_ signalClient: SignalClient, didReceiveConnectResponse connectResponse: SignalClient.ConnectResponse) async
-    func signalClient(_ signalClient: SignalClient, didReceiveAnswer answer: LKRTCSessionDescription) async
-    func signalClient(_ signalClient: SignalClient, didReceiveOffer offer: LKRTCSessionDescription) async
+    func signalClient(_ signalClient: SignalClient, didReceiveAnswer answer: LKRTCSessionDescription, offerId: UInt32) async
+    func signalClient(_ signalClient: SignalClient, didReceiveOffer offer: LKRTCSessionDescription, offerId: UInt32) async
     func signalClient(_ signalClient: SignalClient, didReceiveIceCandidate iceCandidate: IceCandidate, target: Livekit_SignalTarget) async
     func signalClient(_ signalClient: SignalClient, didUnpublishLocalTrack localTrack: Livekit_TrackUnpublishedResponse) async
     func signalClient(_ signalClient: SignalClient, didUpdateParticipants participants: [Livekit_ParticipantInfo]) async
@@ -35,6 +35,6 @@ protocol SignalClientDelegate: AnyObject, Sendable {
     func signalClient(_ signalClient: SignalClient, didReceiveRoomMoved response: Livekit_RoomMovedResponse) async
     func signalClient(_ signalClient: SignalClient, didUpdateSubscriptionPermission permission: Livekit_SubscriptionPermissionUpdate) async
     func signalClient(_ signalClient: SignalClient, didUpdateToken token: String) async
-    func signalClient(_ signalClient: SignalClient, didReceiveLeave canReconnect: Bool, reason: Livekit_DisconnectReason) async
+    func signalClient(_ signalClient: SignalClient, didReceiveLeave action: Livekit_LeaveRequest.Action, reason: Livekit_DisconnectReason) async
     func signalClient(_ signalClient: SignalClient, didSubscribeTrack trackSid: Track.Sid) async
 }
diff --git a/Sources/LiveKit/Types/SessionDescription.swift b/Sources/LiveKit/Types/SessionDescription.swift
@@ -17,9 +17,10 @@
 internal import LiveKitWebRTC
 
 extension LKRTCSessionDescription {
-    func toPBType() -> Livekit_SessionDescription {
+    func toPBType(offerId: UInt32) -> Livekit_SessionDescription {
         var sd = Livekit_SessionDescription()
         sd.sdp = sdp
+        sd.id = offerId
 
         switch type {
         case .answer: sd.type = "answer"
@@ -33,7 +34,7 @@ extension LKRTCSessionDescription {
 }
 
 extension Livekit_SessionDescription {
-    func toRTCType() -> LKRTCSessionDescription {
+    func toRTCType() -> (LKRTCSessionDescription, UInt32) {
         var sdpType: LKRTCSdpType
         switch type {
         case "answer": sdpType = .answer
@@ -42,6 +43,6 @@ extension Livekit_SessionDescription {
         default: fatalError("Unknown state \(type)") // This should never happen
         }
 
-        return RTC.createSessionDescription(type: sdpType, sdp: sdp)
+        return (RTC.createSessionDescription(type: sdpType, sdp: sdp), id)
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+patch type="fixed" "Reconnect sequence stuck in failed state"`
Original file line number	Diff line number	Diff line change
`@@ -17,9 +17,10 @@`
`17`	`17`	`internal import LiveKitWebRTC`
`18`	`18`
`19`	`19`	`extension LKRTCSessionDescription {`
`20`		`- func toPBType() -> Livekit_SessionDescription {`
	`20`	`+ func toPBType(offerId: UInt32) -> Livekit_SessionDescription {`
`21`	`21`	`var sd = Livekit_SessionDescription()`
`22`	`22`	`sd.sdp = sdp`
	`23`	`+ sd.id = offerId`
`23`	`24`
`24`	`25`	`switch type {`
`25`	`26`	`case .answer: sd.type = "answer"`
`@@ -33,7 +34,7 @@ extension LKRTCSessionDescription {`
`33`	`34`	`}`
`34`	`35`
`35`	`36`	`extension Livekit_SessionDescription {`
`36`		`- func toRTCType() -> LKRTCSessionDescription {`
	`37`	`+ func toRTCType() -> (LKRTCSessionDescription, UInt32) {`
`37`	`38`	`var sdpType: LKRTCSdpType`
`38`	`39`	`switch type {`
`39`	`40`	`case "answer": sdpType = .answer`
`@@ -42,6 +43,6 @@ extension Livekit_SessionDescription {`
`42`	`43`	`default: fatalError("Unknown state \(type)") // This should never happen`
`43`	`44`	`}`
`44`	`45`
`45`		`- return RTC.createSessionDescription(type: sdpType, sdp: sdp)`
	`46`	`+ return (RTC.createSessionDescription(type: sdpType, sdp: sdp), id)`
`46`	`47`	`}`
`47`	`48`	`}`