Skip to content
Merged
Original file line number Diff line number Diff line change
Expand Up @@ -118,6 +118,10 @@ public nonisolated struct ActionInvocation: Sendable, Hashable {
origin == .user || focusRequested || arguments["focus"]?.boolValue == true
}

/// A person asked for this run (a click, the palette, a key, a `cmux` command): only such a
/// run may take over a remote session another daemon holds (HQ #4877 G5).
public var isPersonsRequest: Bool { origin == .user || origin == .cli }

public subscript(_ name: String) -> ActionValue? {
arguments[name]
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -151,7 +151,7 @@ extension AgentTabStore {
tabs.cloudMachines.wire(services)
tabs.remoteControl = services.iphone.remoteControl
// The composer strip's machine dot: the sidebar header's same link state; using a machine
// that is down runs its header's Reconnect.
// that is down runs its header's Reconnect, as `.page`: never a takeover (HQ #4877 G5).
tabs.machineLink.state = { [weak tabs, weak services] view in
guard let tabs, let services, let key = tabs.views.first(where: { $0.value === view })?.key,
let machine = tabs.remoteMachineID(of: key) else { return nil }
Expand All @@ -161,7 +161,7 @@ extension AgentTabStore {
services.map { SidebarBridge.paneUse(machine, in: $0.machines) } ?? .stuck
}
tabs.machineLink.reconnect = { [weak tabs] machine in
_ = tabs?.actionRegistry?.perform("machine.reconnect", invocation: ActionInvocation(target: ActionTargetRef(kind: .machine, id: machine), origin: .user))
_ = tabs?.actionRegistry?.perform("machine.reconnect", invocation: ActionInvocation(target: ActionTargetRef(kind: .machine, id: machine), origin: .page))
}
// The composer chips and shell commands chat pages held at the last quit (every relaunch's quit).
tabs.relaunchWake = services.relaunchWake
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ enum RemoteGroupHandlers {
// `cmux workspace group create --remote HOST --name N` names the group's header.
let name = invocation["name"]?.stringValue?.trimmingCharacters(in: .whitespacesAndNewlines)
let make = { (text: String) throws in
let session = try ssh.connect(destination: text, session: nil, offerInstall: interactive)
let session = try ssh.connect(destination: text, session: nil, offerInstall: interactive, byPerson: invocation.isPersonsRequest)
markWhenNamed(session, name: name)
if interactive { RemoteHandlers.showWhenConnected(session, context) }
}
Expand Down Expand Up @@ -71,7 +71,7 @@ enum RemoteGroupHandlers {
switch route(machine, machines) {
case .wakeServer: machines.server(machine)?.wake()
case .retryCloud: machines.session(machine)?.retry()
case .ssh: _ = registry.perform("remote.reconnect", invocation: ActionInvocation(target: ActionTargetRef(kind: .machine, id: machine)))
case .ssh: _ = registry.perform("remote.reconnect", invocation: ActionInvocation(target: ActionTargetRef(kind: .machine, id: machine), origin: invocation.origin))
case .none: throw ActionFailure.invalidTarget(WorkspaceVerbStrings.noMachine)
}
})
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ enum RemoteHandlers {
let connect = { (text: String) in
let session = try ssh.connect(destination: text, session: invocation["session"]?.stringValue,
binary: invocation["binary"]?.stringValue, stateDir: invocation["stateDir"]?.stringValue,
offerInstall: interactive)
offerInstall: interactive, byPerson: invocation.isPersonsRequest)
if interactive { showWhenConnected(session, context) }
}
if let text = invocation["destination"]?.stringValue {
Expand Down Expand Up @@ -106,7 +106,7 @@ enum RemoteHandlers {
registry.track(work)
}
CloudHandlers.bind("remote.reconnect", registry, reason: hasMachine) { invocation in
ssh.reconnect(try machine(invocation, context))
ssh.reconnect(try machine(invocation, context), byPerson: invocation.isPersonsRequest)
}
CloudHandlers.bind("remote.disconnect", registry, reason: hasMachine) { invocation in
ssh.disconnect(try machine(invocation, context))
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ enum SSHTargetHandlers {
static func bind(into registry: ActionRegistry, context: AppActionContext) {
let ssh = context.services.ssh
let available: @MainActor () -> String? = { ssh.unavailableReason }
CloudHandlers.bind("reconnectWorkspace", registry, reason: available) { ssh.reconnect(try workspaceMachine($0, context)) }
CloudHandlers.bind("reconnectWorkspace", registry, reason: available) { ssh.reconnect(try workspaceMachine($0, context), byPerson: $0.isPersonsRequest) }
CloudHandlers.bind("disconnectWorkspace", registry, reason: available) { ssh.disconnect(try workspaceMachine($0, context)) }
CloudHandlers.bind("copyWorkspaceSSHError", registry, reason: available) { invocation in
let session = try workspaceMachine(invocation, context)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -61,12 +61,16 @@ final class SSHMachineSession {
@ObservationIgnored var offersInstall = false
@ObservationIgnored var onStatusChange: ((SSHMachineSession, SSHConnectionMachine.Status) -> Void)?
@ObservationIgnored var connectRecovery: SSHConnectRecovery?
/// In use by another cmux daemon: only a person's Reconnect takes it over (G5). Retries spent: Offline (G4).
var waitsForUser: Bool { link.waitsForUser }
var retriesSpent: Bool { daemon.store.retriesSpent }
/// Changes at the start of every user or automatic connection attempt, even when the
/// transport and daemon report the same failure again.
private(set) var connectionAttemptID: UInt64 = 0
@ObservationIgnored var connectPresentationTask: Task<Void, Never>?
@ObservationIgnored private var statusTask: Task<Void, Never>?
@ObservationIgnored private var startupTask: Task<Void, Never>?
@ObservationIgnored private var restartTask: Task<Void, Never>?

/// - Parameter mainWork: The app's main-thread scheduler (the local daemon's), for the machine's event batches.
init(host: SSHHost, binary: URL, paths: SSHPaths, environment: @escaping @Sendable () async -> [String: String],
Expand Down Expand Up @@ -185,9 +189,24 @@ final class SSHMachineSession {
// mirror moves now, so the new attempt never shows the old failure. The link
// then reports `.connecting` (already mirrored) and the retry's own result.
if linkStatus.needsRecovery { linkStatusChanged(.connecting) }
daemon.store.noteRetriesSpent(false)
daemon.beginAttempt()
}

/// Disconnect then Connect, with a new retry budget: Reconnect once the retries are spent.
func restart() {
daemon.shutdownConnection()
beginConnectionAttempt()
linkStatus = .connecting
// task-owner: teardown hop; cancelled by the next restart and by close(), so a closed session never connects
restartTask?.cancel()
restartTask = Task { [weak self, link] in
await link.stop()
guard !Task.isCancelled, let self, self.autoConnect else { return }
self.connect()
}
}

/// Reopens a daemon attempt without disturbing an already-connected SSH transport.
func reconnectDaemon() {
beginConnectionAttempt()
Expand All @@ -207,6 +226,7 @@ final class SSHMachineSession {

func disconnect(keepAutoConnect: Bool = false) {
if !keepAutoConnect { autoConnect = false }
restartTask?.cancel()
daemon.shutdownConnection()
linkStatus = .offline
let link = link
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -149,10 +149,10 @@ final class SSHService {

// MARK: Connect, disconnect, forget

/// Connects to `destination` (or reconnects the saved machine it names).
/// Connects to `destination` (or reconnects the saved machine it names); `byPerson` as ``reconnect(_:byPerson:)``.
@discardableResult
func connect(destination text: String, session name: String?, binary: String? = nil, stateDir: String? = nil,
offerInstall: Bool) throws -> SSHMachineSession {
offerInstall: Bool, byPerson: Bool = true) throws -> SSHMachineSession {
if let reason = unavailableReason { throw ActionFailure(message: reason) }
let destination: SSHDestination
do { destination = try SSHDestination(parsing: text) } catch { throw ActionFailure(message: RemoteStrings.invalidDestination(text, error)) }
Expand All @@ -163,7 +163,7 @@ final class SSHService {
rememberHost(saved)
saved.offersInstall = offerInstall
if offerInstall { offerInstallWhenOld(saved) }
reconnect(saved)
reconnect(saved, byPerson: byPerson)
logger.info("connect \(destination.description, privacy: .public): reusing \(saved.machineID, privacy: .public)")
return saved
}
Expand All @@ -188,6 +188,7 @@ final class SSHService {
}
let session = machines.sshSession(host.machineID) ?? makeSession(host)
guard let session else { throw ActionFailure(message: RemoteStrings.noClient) }
if !byPerson, session.waitsForUser { return session }
// Saved now, before the first connect succeeds (no session id yet).
session.autoConnect = true
rememberHost(session)
Expand Down Expand Up @@ -219,13 +220,15 @@ final class SSHService {
return text
}

func reconnect(_ session: SSHMachineSession) {
guard !policyDisabled else { return }
func reconnect(_ session: SSHMachineSession, byPerson: Bool = true) {
guard !policyDisabled, byPerson || !session.waitsForUser else { return } // only a person takes over a session in use (G5)
// A retry gets a fresh error from the new attempt. Do not let the
// recovery sheet copy the previous attempt while the link is working.
session.lastError = nil
switch session.linkStatus.reconnectStep(daemonConnected: session.daemon.startup == .connected, daemonFailed: session.daemonFailure != nil) {
switch session.linkStatus.reconnectStep(daemonConnected: session.daemon.startup == .connected, daemonFailed: session.daemonFailure != nil,
retriesSpent: session.retriesSpent) {
case .connect: session.connect()
case .restart: session.restart()
case .reconnectDaemon: session.reconnectDaemon()
case .wake: session.wake(.user)
case .newAttempt:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -37,9 +37,9 @@ extension SidebarBridge {
}

/// What a chat that needs machine `machineID` finds its link doing: only a link that dropped
/// or never came up reconnects on use (``MachineLinkState/reconnectsOnUse(_:)``).
/// or never came up reconnects on use (``MachineLinkState/reconnectsOnUse(_:)``), never an SSH session in use (G5).
static func paneUse(_ machineID: String, in machines: MachineRegistry) -> AgentPaneMachineLink.Use {
guard let status = machineHeader(machineID, in: machines)?.status else { return .stuck }
guard machines.sshSession(machineID)?.waitsForUser != true, let status = machineHeader(machineID, in: machines)?.status else { return .stuck }
switch status.linkState {
case .connected: return .up
case .connecting: return .coming
Expand All @@ -58,18 +58,19 @@ extension SidebarBridge {

static func sshStatus(_ session: SSHMachineSession, compatibility: DaemonCompatibility?) -> SidebarMachine.Status {
let connected = if case .connected = session.daemon.store.connectionState { true } else { false }
let status = sshStatus(link: session.linkStatus, startupFailed: session.daemonFailure != nil, daemonConnected: connected, compatibility: compatibility)
let status = sshStatus(link: session.linkStatus, startupFailed: session.daemonFailure != nil, daemonConnected: connected, compatibility: compatibility, retriesSpent: session.retriesSpent)
return status == .connected && session.updateOffered ? .updateAvailable : status
}

/// `startupFailed`: the daemon gave up after SSH reached the machine
/// (`SSHMachineSession.daemonFailure`): Failed to start, not Connecting (cx-zdh8).
/// `retriesSpent`: a later reconnect gave up: Offline, not Connecting forever (HQ #4877 G4).
static func sshStatus(link: SSHConnectionMachine.Status, startupFailed: Bool, daemonConnected: Bool,
compatibility: DaemonCompatibility?) -> SidebarMachine.Status {
compatibility: DaemonCompatibility?, retriesSpent: Bool = false) -> SidebarMachine.Status {
switch link {
case .offline: return .offline
case .installing: return .installing
case .reconnecting: return .reconnecting
case .reconnecting: return daemonConnected || !retriesSpent ? .reconnecting : .offline
case .authFailed, .hostKeyUntrusted: return .authFailed
case .unreachable: return .unreachable
case .needsInstall(.protocolMismatch), .needsInstall(.refused(.protocolOlder)): return .updateRequired
Expand All @@ -79,7 +80,7 @@ extension SidebarBridge {
case .connecting, .connected:
guard daemonConnected else {
if let compatibility, compatibility.level == .incompatible, startupFailed { return .updateRequired }
return startupFailed ? .failed : .connecting
return startupFailed ? .failed : retriesSpent ? .offline : .connecting
}
if compatibility?.level == .limited { return .updateAvailable }
return .connected
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,8 @@ public actor DaemonConnection {
var handoffHoldTimer: DemandTimer?
/// Shows a request refused because it names a pane still opening (``PaneNotReady``); see ``observePaneNotReady(_:)``.
private var paneNotReadyObserver: (@Sendable (PaneNotReady) -> Void)?
/// Told each time a reconnect fails with the timed retries spent; see ``observeRetriesSpent(_:)``.
private var retriesSpentObserver: (@Sendable () -> Void)?
/// The launch's first snapshot, fetched right after the handshake (``prefetchSnapshot()``).
var prefetchedSnapshot: (serial: UInt64, fetch: Task<(tree: DaemonTree, barrier: UInt64), any Error>)?

Expand Down Expand Up @@ -237,6 +239,13 @@ public actor DaemonConnection {
paneNotReadyObserver = observer
}

/// Sets what is told when the reconnect loop's timed retries are spent and it waits for an
/// event only (a remote machine's header then reads offline, not connecting, HQ #4877 G4).
/// Runs on this connection's executor, after each failed attempt past the budget.
public func observeRetriesSpent(_ observer: (@Sendable () -> Void)?) {
retriesSpentObserver = observer
}

/// The event sequence this connection has routed so far, or nil when
/// not connected. Taken after a command's reply, it is a write barrier:
/// once `DaemonStore.appliedSequence` reaches it, the store reflects
Expand Down Expand Up @@ -417,6 +426,8 @@ public actor DaemonConnection {
}
if pacer.isExhausted {
logger.info("cmux-tui reconnect: timed retries spent; waiting for the daemon socket or another event")
// A busy daemon keeps its timed retry: still reconnecting, not offline.
if pacer.waitsForEventOnly, !Task.isCancelled, !isClosedPhase { retriesSpentObserver?() }
}
}
reconnectTask = nil
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,10 @@ public struct RetryPacer: Sendable {
/// The timed-retry budget is spent: wait for an event only.
public var isExhausted: Bool { failures >= policy.timedRetries }

/// The next wait arms no timer: the budget is spent and the last attempt was not a busy
/// refusal (``refusedBusy()``), whose daemon is alive and keeps a timed retry.
public var waitsForEventOnly: Bool { isExhausted && !busyRefusal }

/// A socket change may wake the next wait: false in a crash loop.
public var wakesOnSocket: Bool { socketFailures < policy.socketRetries }

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ enum DaemonConnectionChange {
store.sideEvents.noteControl(envelope)
store.session.connected(servesStateResources: identity.supports(DaemonCapability.stateResources.rawValue))
store.connectionState = .connected(identity)
store.noteRetriesSpent(false)
store.noteHandshake(identity)
return .resync
case .disconnected(let reason):
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -378,6 +378,16 @@ public final class DaemonStore: StateResourceQueries {
connectionState = .failed(message)
}

/// A remote machine's reconnect spent its timed retries and waits for an event only
/// (`DaemonConnection.observeRetriesSpent`): its header reads offline, not connecting (HQ
/// #4877 G4). Cleared by the next connect and by a new attempt.
public private(set) var retriesSpent = false

public func noteRetriesSpent(_ spent: Bool) {
if spent, case .connected = connectionState { return }
if retriesSpent != spent { retriesSpent = spent }
}

// MARK: Derived state

/// Indexes the whole tree again, primes every record and flattens the
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
public import CmuxNextDaemon

extension DaemonService {
/// A remote machine's connection, as the service starts using it: requests refused for a pane
/// still opening show their notice, and a reconnect that spent its timed retries marks the
/// store (``DaemonStore/retriesSpent``), so its header reads offline instead of connecting
/// forever (HQ #4877 G4). A hop that lands after the link came back
/// (``DaemonStore/noteRetriesSpent(_:)``) or after this connection was replaced or shut down
/// (a Reconnect's restart) changes nothing.
func observeRemote(_ connection: DaemonConnection) async {
await connection.observePaneNotReady(paneNotReadyReporter)
await connection.observeRetriesSpent { [weak self, weak connection] in
// task-owner: one hop to main per failed attempt past the budget; nothing to cancel
Task { @MainActor in
guard let self, let connection, self.connection === connection else { return }
self.store.noteRetriesSpent(true)
}
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -346,7 +346,7 @@ public final class DaemonService {
await connection.closeAndWait()
return
}
await connection.observePaneNotReady(self.paneNotReadyReporter)
await self.observeRemote(connection)
self.didConnect(connection, identity: identity)
logger.info("\(machineID, privacy: .public): cmux-tui \(identity.version, privacy: .public) session \(identity.session, privacy: .public)")
await store.run(connection: connection, scheduler: drainScheduler)
Expand Down
Loading
Loading