fix(macos): back off failed Gateway connects (#116114)

* fix(macos): back off failed gateway connects

* fix(macos): skip backoff for transport cancellation

apps/macos/Tests/OpenClawIPCTests/GatewayChannelConnectTests.swift

@@ -6,7 +6,7 @@ import Testing
 
 @Suite(.serialized)
 struct GatewayChannelConnectTests {
-    private actor NonCooperativeChallengeGate {
+    private actor NonCooperativeGate {
         private var isOpen = false
         private var didStart = false
         private var startWaiters: [CheckedContinuation<Void, Never>] = []
@@ -61,10 +61,10 @@ struct GatewayChannelConnectTests {
 
     private final class FirstChallengeTaskPlan: @unchecked Sendable {
         private let lock = NSLock()
-        private let gate: NonCooperativeChallengeGate
+        private let gate: NonCooperativeGate
         private var taskCount = 0
 
-        init(gate: NonCooperativeChallengeGate) {
+        init(gate: NonCooperativeGate) {
             self.gate = gate
         }
 
@@ -299,8 +299,86 @@ struct GatewayChannelConnectTests {
         #expect(session.snapshotMakeCount() == 1)
     }
 
+    @Test func `failed connect coalesces callers behind backoff`() async throws {
+        let gate = NonCooperativeGate()
+        let session = self.makeSession(response: .invalid(delayMs: 0))
+        let channel = try GatewayChannelActor(
+            url: #require(URL(string: "ws://example.invalid")),
+            token: nil,
+            session: WebSocketSessionBox(session: session))
+
+        await #expect(throws: (any Error).self) {
+            try await channel.connect()
+        }
+        await channel._test_setConnectFailureBackoffWaitHandler {
+            await gate.wait()
+        }
+
+        let retries = (0..<5).map { _ in
+            Task { try await channel.connect() }
+        }
+        await gate.waitUntilStarted()
+        try await AsyncTimeout.withTimeout(
+            seconds: 2,
+            onTimeout: {
+                NSError(
+                    domain: "GatewayChannelConnectTests",
+                    code: 1,
+                    userInfo: [NSLocalizedDescriptionKey: "retry callers did not join the shared connect attempt"])
+            },
+            operation: {
+                while await channel._test_connectWaiterCount() < retries.count {
+                    await Task.yield()
+                }
+            })
+        #expect(session.snapshotMakeCount() == 1)
+        await gate.open()
+
+        for retry in retries {
+            if case .success = await retry.result {
+                Issue.record("retry unexpectedly succeeded")
+            }
+        }
+        await channel._test_setConnectFailureBackoffWaitHandler(nil)
+        await channel.shutdown()
+
+        #expect(session.snapshotMakeCount() == 2)
+    }
+
+    @Test func `shutdown during connect backoff does not create a socket`() async throws {
+        let gate = NonCooperativeGate()
+        let completion = ConnectAttemptCompletionProbe()
+        let session = self.makeSession(response: .invalid(delayMs: 0))
+        let channel = try GatewayChannelActor(
+            url: #require(URL(string: "ws://example.invalid")),
+            token: nil,
+            session: WebSocketSessionBox(session: session))
+
+        await #expect(throws: (any Error).self) {
+            try await channel.connect()
+        }
+        await channel._test_setConnectFailureBackoffWaitHandler {
+            await gate.wait()
+        }
+        await channel._test_setConnectRunFinishedHandler {
+            Task { await completion.record() }
+        }
+
+        let retry = Task { try await channel.connect() }
+        await gate.waitUntilStarted()
+        await channel.shutdown()
+        await gate.open()
+        await completion.wait(for: 1)
+        if case .success = await retry.result {
+            Issue.record("retry unexpectedly succeeded after shutdown")
+        }
+
+        await channel._test_setConnectRunFinishedHandler(nil)
+        #expect(session.snapshotMakeCount() == 1)
+    }
+
     @Test func `timed out connect cannot use retry socket after late challenge`() async throws {
-        let gate = NonCooperativeChallengeGate()
+        let gate = NonCooperativeGate()
         let completion = ConnectAttemptCompletionProbe()
         let plan = FirstChallengeTaskPlan(gate: gate)
         let session = GatewayTestWebSocketSession(taskFactory: { plan.makeTask() })

apps/shared/OpenClawKit/Sources/OpenClawKit/GatewayChannel+Testing.swift

@@ -0,0 +1,33 @@
+import Foundation
+
+extension GatewayChannelActor {
+    func _test_setConnectTimeoutSeconds(_ seconds: Double) {
+        self.connectTimeoutSeconds = seconds
+    }
+
+    func _test_setConnectAttemptFinishedHandler(_ handler: (@Sendable (UUID) -> Void)?) {
+        self.testConnectAttemptFinishedHandler = handler
+    }
+
+    #if DEBUG
+    func _test_setConnectRunFinishedHandler(_ handler: (@Sendable () -> Void)?) {
+        self.testConnectRunFinishedHandler = handler
+    }
+
+    func _test_setConnectFailureBackoffWaitHandler(_ handler: (@Sendable () async throws -> Void)?) {
+        self.testConnectFailureBackoffWaitHandler = handler
+    }
+
+    func _test_setRequestResumedHandler(_ handler: (@Sendable () async -> Void)?) {
+        self.testRequestResumedHandler = handler
+    }
+    #endif
+
+    func _test_pendingRequestCount() -> Int {
+        self.pending.count
+    }
+
+    func _test_connectWaiterCount() -> Int {
+        self.connectWaiters.count
+    }
+}

apps/shared/OpenClawKit/Sources/OpenClawKit/GatewayChannel.swift

@@ -50,7 +50,7 @@ public actor GatewayChannelActor {
     private let logger = Logger(subsystem: "ai.openclaw", category: "gateway")
     private var task: WebSocketTaskBox?
     private var activeConnectAttemptID: UUID?
-    private var pending: [String: CheckedContinuation<GatewayFrame, Error>] = [:]
+    var pending: [String: CheckedContinuation<GatewayFrame, Error>] = [:]
     private var connected = false
     private var connectAttemptTask: Task<Void, Never>?
     /// Socket ownership epoch. Every callback and send stays bound to the task
@@ -59,7 +59,7 @@ public actor GatewayChannelActor {
     private var disconnectedConnectionGeneration: UInt64?
     private var disconnectNotificationInProgress = false
     private var automaticReconnectRequested = false
-    private var connectWaiters: [UUID: CheckedContinuation<Void, Error>] = [:]
+    var connectWaiters: [UUID: CheckedContinuation<Void, Error>] = [:]
     private var url: URL
     private var token: String?
     private var bootstrapToken: String?
@@ -67,6 +67,7 @@ public actor GatewayChannelActor {
     private let authBindingKey: SymmetricKey?
     private let session: WebSocketSessioning
     private var backoffMs: Double = 500
+    var connectFailureBackoff = GatewayConnectFailureBackoff()
     private var shouldReconnect = true
     private var lastSeq: Int?
     private var lastTick: Date?
@@ -77,10 +78,12 @@ public actor GatewayChannelActor {
     private let encoder = JSONEncoder()
     // Remote gateways (tailscale/wan) can take longer to deliver connect.challenge.
     // Connect now requires this nonce before we send device-auth.
-    private var connectTimeoutSeconds: Double = 30
-    private var testConnectAttemptFinishedHandler: (@Sendable (UUID) -> Void)?
+    var connectTimeoutSeconds: Double = 30
+    var testConnectAttemptFinishedHandler: (@Sendable (UUID) -> Void)?
     #if DEBUG
-    private var testRequestResumedHandler: (@Sendable () async -> Void)?
+    var testConnectRunFinishedHandler: (@Sendable () -> Void)?
+    var testConnectFailureBackoffWaitHandler: (@Sendable () async throws -> Void)?
+    var testRequestResumedHandler: (@Sendable () async -> Void)?
     #endif
     private let connectChallengeTimeoutSeconds: Double = 6.0
     // Some networks will silently drop idle TCP/TLS flows around ~30s. The gateway tick is server->client,
@@ -143,28 +146,6 @@ public actor GatewayChannelActor {
         return self.lastAuthBinding?.binding
     }
 
-    func _test_setConnectTimeoutSeconds(_ seconds: Double) {
-        self.connectTimeoutSeconds = seconds
-    }
-
-    func _test_setConnectAttemptFinishedHandler(_ handler: (@Sendable (UUID) -> Void)?) {
-        self.testConnectAttemptFinishedHandler = handler
-    }
-
-    #if DEBUG
-    func _test_setRequestResumedHandler(_ handler: (@Sendable () async -> Void)?) {
-        self.testRequestResumedHandler = handler
-    }
-    #endif
-
-    func _test_pendingRequestCount() -> Int {
-        self.pending.count
-    }
-
-    func _test_connectWaiterCount() -> Int {
-        self.connectWaiters.count
-    }
-
     public func shutdown() async {
         self.shouldReconnect = false
         self.connected = false
@@ -284,6 +265,9 @@ public actor GatewayChannelActor {
         } catch {
             self.finishConnectAttempt(error: error)
         }
+        #if DEBUG
+        self.testConnectRunFinishedHandler?()
+        #endif
     }
 
     private func waitForConnectAttempt() async throws {
@@ -322,6 +306,10 @@ public actor GatewayChannelActor {
     }
 
     private func performConnectAttempt() async throws {
+        guard self.shouldReconnect else { throw CancellationError() }
+        guard !self.disconnectNotificationInProgress else { throw CancellationError() }
+        try await self.waitForConnectFailureBackoff()
+        try Task.checkCancellation()
         guard self.shouldReconnect else { throw CancellationError() }
         guard !self.disconnectNotificationInProgress else { throw CancellationError() }
         if self.connected {
@@ -375,6 +363,7 @@ public actor GatewayChannelActor {
             } else {
                 self.wrap(error, context: "connect to gateway @ \(self.url.absoluteString)")
             }
+            self.connectFailureBackoff.record(error: error, pendingDeviceTokenRetry: self.pendingDeviceTokenRetry)
             await self.transitionToDisconnected(
                 reason: "connect failed: \(wrapped.localizedDescription)",
                 error: wrapped,
@@ -392,6 +381,7 @@ public actor GatewayChannelActor {
         self.automaticReconnectRequested = false
         self.reconnectPausedForAuthFailure = false
         self.backoffMs = 500
+        self.connectFailureBackoff.reset()
         self.lastSeq = nil
         self.listen(connectionGeneration: connectionGeneration)
         self.startTickWatchdog(connectionGeneration: connectionGeneration)

apps/shared/OpenClawKit/Sources/OpenClawKit/GatewayConnectFailureBackoff.swift

@@ -0,0 +1,56 @@
+import Foundation
+
+struct GatewayConnectFailureBackoff {
+    private var milliseconds: Double = 500
+    private var retryNotBefore: ContinuousClock.Instant?
+
+    var deadline: ContinuousClock.Instant? {
+        self.retryNotBefore
+    }
+
+    mutating func clear(deadline: ContinuousClock.Instant) {
+        if self.retryNotBefore == deadline {
+            self.retryNotBefore = nil
+        }
+    }
+
+    mutating func record(error: Error, pendingDeviceTokenRetry: Bool) {
+        guard !Self.isCancellation(error) else { return }
+        let delayMs = pendingDeviceTokenRetry ? min(self.milliseconds, 250) : self.milliseconds
+        let clock = ContinuousClock()
+        self.retryNotBefore = clock.now.advanced(by: .milliseconds(Int64(delayMs.rounded(.up))))
+        self.milliseconds = min(self.milliseconds * 2, 30000)
+    }
+
+    mutating func reset() {
+        self.milliseconds = 500
+        self.retryNotBefore = nil
+    }
+
+    private static func isCancellation(_ error: Error) -> Bool {
+        if error is CancellationError { return true }
+        let nsError = error as NSError
+        return nsError.domain == NSURLErrorDomain &&
+            nsError.code == URLError.Code.cancelled.rawValue
+    }
+}
+
+extension GatewayChannelActor {
+    func waitForConnectFailureBackoff() async throws {
+        guard let deadline = self.connectFailureBackoff.deadline else { return }
+        // Delay inside the shared connect attempt so callers coalesce before a
+        // socket is created instead of starting independent retry bursts.
+        #if DEBUG
+        if let testConnectFailureBackoffWaitHandler {
+            try await testConnectFailureBackoffWaitHandler()
+            self.connectFailureBackoff.clear(deadline: deadline)
+            return
+        }
+        #endif
+        let clock = ContinuousClock()
+        if clock.now < deadline {
+            try await clock.sleep(until: deadline)
+        }
+        self.connectFailureBackoff.clear(deadline: deadline)
+    }
+}