bug: fixing macos app gateway hang
This commit is contained in:
Olivia Buzek
parent
961b4adc1c
commit
ed14efe470
@ -7,44 +7,78 @@ final class ConnectionModeCoordinator {
|
||||
|
||||
private let logger = Logger(subsystem: "com.clawdbot", category: "connection")
|
||||
private var lastMode: AppState.ConnectionMode?
|
||||
private var applyTask: Task<Void, Never>?
|
||||
|
||||
/// Apply the requested connection mode by starting/stopping local gateway,
|
||||
/// managing the control-channel SSH tunnel, and cleaning up chat windows/panels.
|
||||
func apply(mode: AppState.ConnectionMode, paused: Bool) async {
|
||||
if let lastMode = self.lastMode, lastMode != mode {
|
||||
GatewayProcessManager.shared.clearLastFailure()
|
||||
NodesStore.shared.lastError = nil
|
||||
/// This spawns a background task to avoid blocking the UI during gateway startup.
|
||||
/// Cancels any in-flight mode transition to prevent concurrent state changes.
|
||||
func apply(mode: AppState.ConnectionMode, paused: Bool) {
|
||||
// Cancel previous mode transition if still running
|
||||
applyTask?.cancel()
|
||||
|
||||
applyTask = Task.detached(priority: .userInitiated) { [weak self] in
|
||||
await self?.applyAsync(mode: mode, paused: paused)
|
||||
}
|
||||
self.lastMode = mode
|
||||
}
|
||||
|
||||
private func applyAsync(mode: AppState.ConnectionMode, paused: Bool) async {
|
||||
// Check for cancellation before starting work
|
||||
guard !Task.isCancelled else { return }
|
||||
|
||||
let lastMode = await MainActor.run { self.lastMode }
|
||||
if let lastMode, lastMode != mode {
|
||||
await MainActor.run {
|
||||
GatewayProcessManager.shared.clearLastFailure()
|
||||
NodesStore.shared.lastError = nil
|
||||
}
|
||||
}
|
||||
await MainActor.run { self.lastMode = mode }
|
||||
|
||||
// Check for cancellation before mode-specific logic
|
||||
guard !Task.isCancelled else { return }
|
||||
|
||||
switch mode {
|
||||
case .unconfigured:
|
||||
_ = await NodeServiceManager.stop()
|
||||
NodesStore.shared.lastError = nil
|
||||
guard !Task.isCancelled else { return }
|
||||
await MainActor.run { NodesStore.shared.lastError = nil }
|
||||
await RemoteTunnelManager.shared.stopAll()
|
||||
WebChatManager.shared.resetTunnels()
|
||||
GatewayProcessManager.shared.stop()
|
||||
guard !Task.isCancelled else { return }
|
||||
await MainActor.run {
|
||||
WebChatManager.shared.resetTunnels()
|
||||
GatewayProcessManager.shared.stop()
|
||||
}
|
||||
await GatewayConnection.shared.shutdown()
|
||||
guard !Task.isCancelled else { return }
|
||||
await ControlChannel.shared.disconnect()
|
||||
guard !Task.isCancelled else { return }
|
||||
Task.detached { await PortGuardian.shared.sweep(mode: .unconfigured) }
|
||||
|
||||
case .local:
|
||||
_ = await NodeServiceManager.stop()
|
||||
NodesStore.shared.lastError = nil
|
||||
guard !Task.isCancelled else { return }
|
||||
await MainActor.run { NodesStore.shared.lastError = nil }
|
||||
await RemoteTunnelManager.shared.stopAll()
|
||||
WebChatManager.shared.resetTunnels()
|
||||
guard !Task.isCancelled else { return }
|
||||
await MainActor.run { WebChatManager.shared.resetTunnels() }
|
||||
let shouldStart = GatewayAutostartPolicy.shouldStartGateway(mode: .local, paused: paused)
|
||||
if shouldStart {
|
||||
GatewayProcessManager.shared.setActive(true)
|
||||
guard !Task.isCancelled else { return }
|
||||
await MainActor.run { GatewayProcessManager.shared.setActive(true) }
|
||||
if GatewayAutostartPolicy.shouldEnsureLaunchAgent(
|
||||
mode: .local,
|
||||
paused: paused)
|
||||
{
|
||||
Task { await GatewayProcessManager.shared.ensureLaunchAgentEnabledIfNeeded() }
|
||||
}
|
||||
_ = await GatewayProcessManager.shared.waitForGatewayReady()
|
||||
// Note: GatewayProcessManager.startIfNeeded() includes its own readiness polling
|
||||
// via enableLaunchdGateway() (which waits up to 6 seconds for the gateway to accept
|
||||
// connections), so no additional wait is needed here.
|
||||
} else {
|
||||
GatewayProcessManager.shared.stop()
|
||||
await MainActor.run { GatewayProcessManager.shared.stop() }
|
||||
}
|
||||
guard !Task.isCancelled else { return }
|
||||
do {
|
||||
try await ControlChannel.shared.configure(mode: .local)
|
||||
} catch {
|
||||
@ -52,19 +86,24 @@ final class ConnectionModeCoordinator {
|
||||
self.logger.error(
|
||||
"control channel local configure failed: \(error.localizedDescription, privacy: .public)")
|
||||
}
|
||||
guard !Task.isCancelled else { return }
|
||||
Task.detached { await PortGuardian.shared.sweep(mode: .local) }
|
||||
|
||||
case .remote:
|
||||
// Never run a local gateway in remote mode.
|
||||
GatewayProcessManager.shared.stop()
|
||||
WebChatManager.shared.resetTunnels()
|
||||
await MainActor.run { GatewayProcessManager.shared.stop() }
|
||||
guard !Task.isCancelled else { return }
|
||||
await MainActor.run { WebChatManager.shared.resetTunnels() }
|
||||
|
||||
do {
|
||||
NodesStore.shared.lastError = nil
|
||||
await MainActor.run { NodesStore.shared.lastError = nil }
|
||||
guard !Task.isCancelled else { return }
|
||||
if let error = await NodeServiceManager.start() {
|
||||
NodesStore.shared.lastError = "Node service start failed: \(error)"
|
||||
await MainActor.run { NodesStore.shared.lastError = "Node service start failed: \(error)" }
|
||||
}
|
||||
guard !Task.isCancelled else { return }
|
||||
_ = try await GatewayEndpointStore.shared.ensureRemoteControlTunnel()
|
||||
guard !Task.isCancelled else { return }
|
||||
let settings = CommandResolver.connectionSettings()
|
||||
try await ControlChannel.shared.configure(mode: .remote(
|
||||
target: settings.target,
|
||||
@ -72,6 +111,7 @@ final class ConnectionModeCoordinator {
|
||||
} catch {
|
||||
self.logger.error("remote tunnel/configure failed: \(error.localizedDescription, privacy: .public)")
|
||||
}
|
||||
guard !Task.isCancelled else { return }
|
||||
|
||||
Task.detached { await PortGuardian.shared.sweep(mode: .remote) }
|
||||
}
|
||||
|
||||
@ -138,10 +138,18 @@ actor GatewayConnection {
|
||||
let mode = await MainActor.run { AppStateStore.shared.connectionMode }
|
||||
switch mode {
|
||||
case .local:
|
||||
// Trigger gateway start. This suspends (non-blocking) until the MainActor schedules it.
|
||||
await MainActor.run { GatewayProcessManager.shared.setActive(true) }
|
||||
|
||||
var lastError: Error = error
|
||||
for delayMs in [150, 400, 900] {
|
||||
// Short retry delays (150ms total) for Canvas responsiveness. Gateway startup happens
|
||||
// asynchronously in the background, so these fast retries typically succeed on the
|
||||
// 2nd or 3rd attempt once the gateway accepts connections.
|
||||
//
|
||||
// Tradeoff: Cold-start scenarios (first launch, after sleep) may take 2-3 seconds,
|
||||
// but Canvas operations prioritize fast failure feedback over waiting. If retries
|
||||
// fail, the user sees an error immediately rather than a frozen UI.
|
||||
for delayMs in [50, 100] {
|
||||
try await Task.sleep(nanoseconds: UInt64(delayMs) * 1_000_000)
|
||||
do {
|
||||
return try await client.request(method: method, params: params, timeoutMs: timeoutMs)
|
||||
@ -155,7 +163,9 @@ actor GatewayConnection {
|
||||
let fallback = await GatewayEndpointStore.shared.maybeFallbackToTailnet(from: cfg.url)
|
||||
{
|
||||
await self.configure(url: fallback.url, token: fallback.token, password: fallback.password)
|
||||
for delayMs in [150, 400, 900] {
|
||||
// Fast retries for Tailnet fallback (150ms total). Tailnet connections should be
|
||||
// quick since the gateway is already running; short delays provide fast feedback.
|
||||
for delayMs in [50, 100] {
|
||||
try await Task.sleep(nanoseconds: UInt64(delayMs) * 1_000_000)
|
||||
do {
|
||||
guard let client = self.client else {
|
||||
@ -184,7 +194,9 @@ actor GatewayConnection {
|
||||
lastError = error
|
||||
}
|
||||
|
||||
for delayMs in [150, 400, 900] {
|
||||
// Fast retries for remote tunnel recovery (150ms total). SSH tunnel setup is relatively
|
||||
// quick; short delays allow fast recovery or fast failure feedback.
|
||||
for delayMs in [50, 100] {
|
||||
try await Task.sleep(nanoseconds: UInt64(delayMs) * 1_000_000)
|
||||
do {
|
||||
let cfg = try await self.configProvider()
|
||||
|
||||
@ -6,17 +6,43 @@ import Observation
|
||||
final class GatewayProcessManager {
|
||||
static let shared = GatewayProcessManager()
|
||||
|
||||
/// Represents the lifecycle state of the local gateway process.
|
||||
///
|
||||
/// Valid state transitions:
|
||||
/// - `.stopped` → `.starting(phase)` → (`.running` | `.attachedExisting` | `.failed`)
|
||||
/// - `.failed` → `.starting(phase)` (user retry)
|
||||
/// - any state → `.stopped` (manual stop)
|
||||
enum Status: Equatable {
|
||||
/// Gateway is not running and not attempting to start.
|
||||
case stopped
|
||||
case starting
|
||||
|
||||
/// Gateway startup is in progress.
|
||||
/// - Parameter phase: Optional startup phase for user feedback:
|
||||
/// - `"checking"`: Initial state, determining what to do
|
||||
/// - `"attaching"`: Attempting to connect to existing gateway process
|
||||
/// - `"launching"`: Launching new gateway via launchd
|
||||
/// - `"waiting for ready"`: Polling for gateway to accept connections
|
||||
case starting(phase: String?)
|
||||
|
||||
/// Gateway was freshly launched by this app and is now accepting connections.
|
||||
/// - Parameter details: Optional process details (e.g., "pid 12345")
|
||||
case running(details: String?)
|
||||
|
||||
/// Gateway was already running (launched externally or by previous app instance)
|
||||
/// and this app successfully attached to it.
|
||||
/// - Parameter details: Optional process and health details
|
||||
case attachedExisting(details: String?)
|
||||
|
||||
/// Gateway failed to start or attach.
|
||||
/// - Parameter message: Human-readable failure reason shown to user
|
||||
case failed(String)
|
||||
|
||||
var label: String {
|
||||
switch self {
|
||||
case .stopped: return "Stopped"
|
||||
case .starting: return "Starting…"
|
||||
case let .starting(phase):
|
||||
if let phase, !phase.isEmpty { return "Starting (\(phase))…" }
|
||||
return "Starting…"
|
||||
case let .running(details):
|
||||
if let details, !details.isEmpty { return "Running (\(details))" }
|
||||
return "Running"
|
||||
@ -110,15 +136,17 @@ final class GatewayProcessManager {
|
||||
case .stopped, .failed:
|
||||
break
|
||||
}
|
||||
self.status = .starting
|
||||
self.status = .starting(phase: "checking")
|
||||
self.logger.debug("gateway start requested")
|
||||
|
||||
// First try to latch onto an already-running gateway to avoid spawning a duplicate.
|
||||
Task { [weak self] in
|
||||
guard let self else { return }
|
||||
await MainActor.run { self.status = .starting(phase: "attaching") }
|
||||
if await self.attachExistingGatewayIfAvailable() {
|
||||
return
|
||||
}
|
||||
await MainActor.run { self.status = .starting(phase: "launching") }
|
||||
await self.enableLaunchdGateway()
|
||||
}
|
||||
}
|
||||
@ -205,10 +233,12 @@ final class GatewayProcessManager {
|
||||
let data = try await attemptAttach()
|
||||
let snap = decodeHealthSnapshot(from: data)
|
||||
let details = self.describe(details: instanceText, port: port, snap: snap)
|
||||
self.existingGatewayDetails = details
|
||||
self.clearLastFailure()
|
||||
self.status = .attachedExisting(details: details)
|
||||
self.appendLog("[gateway] using existing instance: \(details)\n")
|
||||
await MainActor.run {
|
||||
self.existingGatewayDetails = details
|
||||
self.clearLastFailure()
|
||||
self.status = .attachedExisting(details: details)
|
||||
self.appendLog("[gateway] using existing instance: \(details)\n")
|
||||
}
|
||||
self.logger.info("gateway using existing instance details=\(details)")
|
||||
self.refreshControlChannelIfNeeded(reason: "attach existing")
|
||||
self.refreshLog()
|
||||
@ -221,21 +251,23 @@ final class GatewayProcessManager {
|
||||
|
||||
if hasListener {
|
||||
let reason = self.describeAttachFailure(error, port: port, instance: instance)
|
||||
self.existingGatewayDetails = instanceText
|
||||
self.status = .failed(reason)
|
||||
self.lastFailureReason = reason
|
||||
self.appendLog("[gateway] existing listener on port \(port) but attach failed: \(reason)\n")
|
||||
await MainActor.run {
|
||||
self.existingGatewayDetails = instanceText
|
||||
self.status = .failed(reason)
|
||||
self.lastFailureReason = reason
|
||||
self.appendLog("[gateway] existing listener on port \(port) but attach failed: \(reason)\n")
|
||||
}
|
||||
self.logger.warning("gateway attach failed reason=\(reason)")
|
||||
return true
|
||||
}
|
||||
|
||||
// No reachable gateway (and no listener) — fall through to spawn.
|
||||
self.existingGatewayDetails = nil
|
||||
await MainActor.run { self.existingGatewayDetails = nil }
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
self.existingGatewayDetails = nil
|
||||
await MainActor.run { self.existingGatewayDetails = nil }
|
||||
return false
|
||||
}
|
||||
|
||||
@ -298,7 +330,7 @@ final class GatewayProcessManager {
|
||||
}
|
||||
|
||||
private func enableLaunchdGateway() async {
|
||||
self.existingGatewayDetails = nil
|
||||
await MainActor.run { self.existingGatewayDetails = nil }
|
||||
let resolution = await Task.detached(priority: .utility) {
|
||||
GatewayEnvironment.resolveGatewayCommand()
|
||||
}.value
|
||||
@ -313,26 +345,31 @@ final class GatewayProcessManager {
|
||||
|
||||
if GatewayLaunchAgentManager.isLaunchAgentWriteDisabled() {
|
||||
let message = "Launchd disabled; start the Gateway manually or disable attach-only."
|
||||
self.status = .failed(message)
|
||||
self.lastFailureReason = "launchd disabled"
|
||||
self.appendLog("[gateway] launchd disabled; skipping auto-start\n")
|
||||
await MainActor.run {
|
||||
self.status = .failed(message)
|
||||
self.lastFailureReason = "launchd disabled"
|
||||
self.appendLog("[gateway] launchd disabled; skipping auto-start\n")
|
||||
}
|
||||
self.logger.info("gateway launchd enable skipped (disable marker set)")
|
||||
return
|
||||
}
|
||||
|
||||
let bundlePath = Bundle.main.bundleURL.path
|
||||
let port = GatewayEnvironment.gatewayPort()
|
||||
self.appendLog("[gateway] enabling launchd job (\(gatewayLaunchdLabel)) on port \(port)\n")
|
||||
await MainActor.run { self.appendLog("[gateway] enabling launchd job (\(gatewayLaunchdLabel)) on port \(port)\n") }
|
||||
self.logger.info("gateway enabling launchd port=\(port)")
|
||||
let err = await GatewayLaunchAgentManager.set(enabled: true, bundlePath: bundlePath, port: port)
|
||||
if let err {
|
||||
self.status = .failed(err)
|
||||
self.lastFailureReason = err
|
||||
await MainActor.run {
|
||||
self.status = .failed(err)
|
||||
self.lastFailureReason = err
|
||||
}
|
||||
self.logger.error("gateway launchd enable failed: \(err)")
|
||||
return
|
||||
}
|
||||
|
||||
// Best-effort: wait for the gateway to accept connections.
|
||||
await MainActor.run { self.status = .starting(phase: "waiting for ready") }
|
||||
let deadline = Date().addingTimeInterval(6)
|
||||
while Date() < deadline {
|
||||
if !self.desiredActive { return }
|
||||
@ -340,8 +377,10 @@ final class GatewayProcessManager {
|
||||
_ = try await self.connection.requestRaw(method: .health, timeoutMs: 1500)
|
||||
let instance = await PortGuardian.shared.describe(port: port)
|
||||
let details = instance.map { "pid \($0.pid)" }
|
||||
self.clearLastFailure()
|
||||
self.status = .running(details: details)
|
||||
await MainActor.run {
|
||||
self.clearLastFailure()
|
||||
self.status = .running(details: details)
|
||||
}
|
||||
self.logger.info("gateway started details=\(details ?? "ok")")
|
||||
self.refreshControlChannelIfNeeded(reason: "gateway started")
|
||||
self.refreshLog()
|
||||
@ -351,8 +390,10 @@ final class GatewayProcessManager {
|
||||
}
|
||||
}
|
||||
|
||||
self.status = .failed("Gateway did not start in time")
|
||||
self.lastFailureReason = "launchd start timeout"
|
||||
await MainActor.run {
|
||||
self.status = .failed("Gateway did not start in time")
|
||||
self.lastFailureReason = "launchd start timeout"
|
||||
}
|
||||
self.logger.warning("gateway start timed out")
|
||||
}
|
||||
|
||||
|
||||
@ -73,7 +73,7 @@ struct ClawdbotApp: App {
|
||||
self.applyStatusItemAppearance(paused: self.state.isPaused, sleeping: self.isGatewaySleeping)
|
||||
}
|
||||
.onChange(of: self.state.connectionMode) { _, mode in
|
||||
Task { await ConnectionModeCoordinator.shared.apply(mode: mode, paused: self.state.isPaused) }
|
||||
ConnectionModeCoordinator.shared.apply(mode: mode, paused: self.state.isPaused)
|
||||
CLIInstallPrompter.shared.checkAndPromptIfNeeded(reason: "connection-mode")
|
||||
}
|
||||
|
||||
@ -272,7 +272,7 @@ final class AppDelegate: NSObject, NSApplicationDelegate {
|
||||
self.state = AppStateStore.shared
|
||||
AppActivationPolicy.apply(showDockIcon: self.state?.showDockIcon ?? false)
|
||||
if let state {
|
||||
Task { await ConnectionModeCoordinator.shared.apply(mode: state.connectionMode, paused: state.isPaused) }
|
||||
ConnectionModeCoordinator.shared.apply(mode: state.connectionMode, paused: state.isPaused)
|
||||
}
|
||||
TerminationSignalWatcher.shared.start()
|
||||
NodePairingApprovalPrompter.shared.start()
|
||||
|
||||
Loading…
Reference in New Issue
Block a user