diff --git a/apps/macos/Tests/OpenClawIPCTests/GatewayChannelConnectTests.swift b/apps/macos/Tests/OpenClawIPCTests/GatewayChannelConnectTests.swift index f337203d45db..6241eec67eef 100644 --- a/apps/macos/Tests/OpenClawIPCTests/GatewayChannelConnectTests.swift +++ b/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] = [] @@ -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() }) diff --git a/apps/shared/OpenClawKit/Sources/OpenClawKit/GatewayChannel+Testing.swift b/apps/shared/OpenClawKit/Sources/OpenClawKit/GatewayChannel+Testing.swift new file mode 100644 index 000000000000..c2a2b0922c94 --- /dev/null +++ b/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 + } +} diff --git a/apps/shared/OpenClawKit/Sources/OpenClawKit/GatewayChannel.swift b/apps/shared/OpenClawKit/Sources/OpenClawKit/GatewayChannel.swift index c569fda36bf0..146bfa37f47f 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawKit/GatewayChannel.swift +++ b/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] = [:] + var pending: [String: CheckedContinuation] = [:] private var connected = false private var connectAttemptTask: Task? /// 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] = [:] + var connectWaiters: [UUID: CheckedContinuation] = [:] 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) diff --git a/apps/shared/OpenClawKit/Sources/OpenClawKit/GatewayConnectFailureBackoff.swift b/apps/shared/OpenClawKit/Sources/OpenClawKit/GatewayConnectFailureBackoff.swift new file mode 100644 index 000000000000..147b5dba70ba --- /dev/null +++ b/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) + } +}