Improve disconnect vs reconnect reliability (#761)

Resolves #757 

- Adds cancellation to reconnect, making sure that it cannot continue
after `disconnect()` and bring the room back to life
- ⚠️ Optionally, we can introduce a breaking `.disconnecting` state to
prevent new tasks from being started (vs cancelling ongoing tasks - a
little _less elegant_)
- Theoretically it should not affect other state transitions, as
`disconnect()` is basically "destructive" - the only valid path leads to
`.disconnected` during `cleanup()`, however that's not statically
enforced in any way
- On the other hand, it does not provide any added value for other parts
of the SDK, just covering this tiny time window

Author: pblazej

.changes/disconnect-reconnect

@@ -0,0 +1 @@
+patch type="fixed" "Fixed race condition between reconnect and disconnect leading to failed disconnects"

.changes/disconnected

@@ -0,0 +1 @@
+patch type="added" "Added .disconnected connection state"

Sources/LiveKit/Core/Room+Engine.swift

@@ -357,16 +357,16 @@ extension Room {
         }
 
         do {
-            try await Task.retrying(totalAttempts: _state.connectOptions.reconnectAttempts,
-                                    retryDelay: { @Sendable attempt in
-                                        let delay = TimeInterval.computeReconnectDelay(forAttempt: attempt,
-                                                                                       baseDelay: self._state.connectOptions.reconnectAttemptDelay,
-                                                                                       maxDelay: self._state.connectOptions.reconnectMaxDelay,
-                                                                                       totalAttempts: self._state.connectOptions.reconnectAttempts,
-                                                                                       addJitter: true)
-                                        self.log("[Connect] Retry cycle waiting for \(String(format: "%.2f", delay)) seconds before attempt \(attempt + 1)")
-                                        return delay
-                                    }) { currentAttempt, totalAttempts in
+            let reconnectTask = Task.retrying(totalAttempts: _state.connectOptions.reconnectAttempts,
+                                              retryDelay: { @Sendable attempt in
+                                                  let delay = TimeInterval.computeReconnectDelay(forAttempt: attempt,
+                                                                                                 baseDelay: self._state.connectOptions.reconnectAttemptDelay,
+                                                                                                 maxDelay: self._state.connectOptions.reconnectMaxDelay,
+                                                                                                 totalAttempts: self._state.connectOptions.reconnectAttempts,
+                                                                                                 addJitter: true)
+                                                  self.log("[Connect] Retry cycle waiting for \(String(format: "%.2f", delay)) seconds before attempt \(attempt + 1)")
+                                                  return delay
+                                              }) { currentAttempt, totalAttempts in
                 // Not reconnecting state anymore
                 guard let currentMode = self._state.isReconnectingWithMode else {
                     self.log("[Connect] Not in reconnect state anymore, exiting retry cycle.")
@@ -403,12 +403,19 @@ extension Room {
                     // Re-throw
                     throw error
                 }
-            }.value
+            }
+
+            _state.mutate {
+                $0.reconnectTask = reconnectTask
+            }
+
+            try await reconnectTask.value
 
             // Re-connect sequence successful
             log("[Connect] Sequence completed")
             _state.mutate {
                 $0.connectionState = .connected
+                $0.reconnectTask = nil
                 $0.isReconnectingWithMode = nil
                 $0.nextReconnectMode = nil
             }

Sources/LiveKit/Core/Room.swift

@@ -167,6 +167,7 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable {
         var nextReconnectMode: ReconnectMode?
         var isReconnectingWithMode: ReconnectMode?
         var connectionState: ConnectionState = .disconnected
+        var reconnectTask: Task<Void, Error>?
         var disconnectError: LiveKitError?
         var connectStopwatch = Stopwatch(label: "connect")
         var hasPublished: Bool = false
@@ -410,16 +411,38 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable {
 
     @objc
     public func disconnect() async {
-        // Return if already disconnected state
-        if case .disconnected = connectionState { return }
+        let shouldDisconnect = _state.mutate {
+            switch $0.connectionState {
+            case .disconnecting, .disconnected:
+                return false
+            default:
+                $0.connectionState = .disconnecting
+                return true
+            }
+        }
+        guard shouldDisconnect else { return }
+
+        cancelReconnect()
 
         do {
             try await signalClient.sendLeave()
         } catch {
             log("Failed to send leave with error: \(error)")
         }
 
+        cancelReconnect()
+
         await cleanUp()
+
+        cancelReconnect()
+    }
+
+    private func cancelReconnect() {
+        _state.mutate {
+            log("Cancelling reconnect task: \(String(describing: $0.reconnectTask))")
+            $0.reconnectTask?.cancel()
+            $0.reconnectTask = nil
+        }
     }
 }
 
@@ -430,6 +453,7 @@ extension Room {
     func cleanUp(withError disconnectError: Error? = nil,
                  isFullReconnect: Bool = false) async
     {
+        guard !Task.isCancelled else { return }
         log("withError: \(String(describing: disconnectError)), isFullReconnect: \(isFullReconnect)")
 
         // Reset completers
@@ -456,11 +480,13 @@ extension Room {
                 token: $0.token,
                 nextReconnectMode: $0.nextReconnectMode,
                 isReconnectingWithMode: $0.isReconnectingWithMode,
-                connectionState: $0.connectionState
+                connectionState: $0.connectionState,
+                reconnectTask: $0.reconnectTask
             ) : State(
                 connectOptions: $0.connectOptions,
                 roomOptions: $0.roomOptions,
                 connectionState: .disconnected,
+                reconnectTask: $0.reconnectTask,
                 disconnectError: LiveKitError.from(error: disconnectError)
             )
         }

Sources/LiveKit/Extensions/CustomStringConvertible.swift

@@ -134,6 +134,7 @@ extension ConnectionState: CustomStringConvertible {
         case .connecting: ".connecting"
         case .reconnecting: ".reconnecting"
         case .connected: ".connected"
+        case .disconnecting: ".disconnecting"
         }
     }
 }

Sources/LiveKit/Types/ConnectionState.swift

@@ -35,6 +35,7 @@ public enum ConnectionState: Int, Sendable {
     case connecting
     case reconnecting
     case connected
+    case disconnecting
 }
 
 extension ConnectionState: Identifiable {