From bdb7d58c4bee4fc720d9eed54965059d90d4d131 Mon Sep 17 00:00:00 2001 From: Jinjing <6427696+AmethystLiang@users.noreply.github.com> Date: Sat, 1 Aug 2026 22:21:37 -0700 Subject: [PATCH] fix(P1-B): recover SSH targets and remote file watchers after a network drop (#12032) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(P1-B): recover system-SSH targets after a network drop Two defects stopped a remote workspace auto-recovering after a blip. runReconnectAttempt classified failures with isTransientError, which only matches ETIMEDOUT/ECONNREFUSED/ECONNRESET by errno code or literal substring. The system-SSH transport — the only transport FIDO2 and ProxyUseFdpass targets can use — reports network failures as OpenSSH prose ("System SSH connection timed out"), so the ladder published a permanent 'error' on the first timeout and the target never came back without a manual reconnect. isTransientReconnectError adds a network-shaped prose table on top of isTransientError and is used only on the reconnect path: connect() keeps the narrow classifier so an unreachable host still fails fast instead of burning five 30s attempts and five security-key touch prompts. Auth and passphrase failures stay permanent on both paths. runReconnectAttempt also had no generation fence, so a superseded attempt published its cancellation as a permanent error over the winner's live connection — reachable when a system-transport proc.onExit schedules a reconnect while an attempt is still in flight. Cancellation now carries a stable error name, and both connect() and runReconnectAttempt claim their connectGeneration and stay silent when a newer attempt owns the state. * fix(P1-B): retry a dropped watcher overflow marker on real capacity emitWatcherOverflowToClient published the {kind:'overflow'} resync marker with controlOverflow:'reject'. A full control queue rejects at admission with no settlement callback, so the marker was silently discarded and the remote File Explorer stayed stale until some later watcher event happened to produce another one — for a quiet tree, possibly never. The emitter now retains a rejected marker per (client, root) and republishes it when the sink actually frees up. The existing onLegacyPtyCapacity signal cannot drive that: it is gated on producer retention, so it stays silent exactly under the dual-queue pressure that caused the rejection. RelayDispatcher.onClientCapacity is an ungated per-client capacity signal that fires on every writer settlement and drain. It lives on the dispatcher rather than the writer so a retained marker survives setWrite() replacing the primary sink, and setWrite notifies capacity once afterwards so the marker does not wait on traffic that may never arrive. Retention is bounded to one marker per (client, root), released on settlement and purged on client detach. * fix(P1-B): address all review findings on SSH network recovery Fix four issues from code review: 1. **Bug — admitted overflow markers lost on setWrite**: Retain markers when settlement fails `ok: false`, not just on admission rejection. Prevents desynced filesystem trees after SSH sink replacement. 2. **SSH error classification expanded**: Add missing OpenSSH patterns (`ssh_exchange_identification`, `connection closed by remote`) and new `isDefiniteSystemSshHostFailure()` classifier. 3. **ControlMaster retry optimization**: Skip second probe when first failure is already definite host-level (network timeout, refused, unreachable). Saves ~30s per reconnect ladder step. 4. **Overflow flush under dual-queue pressure**: Gate pending marker retries on control-lane headroom instead of re-attempting on every capacity notification. Reduces thrash proportional to producer traffic. Add regression tests for marker republish on sink replacement and validate auth error detection against live OpenSSH credential rejection messages. * rm random doc * fix(P1-B): skip credential-failure retries and recover watcher markers o - Auth and passphrase errors fail immediately without retry attempts - Bare "System SSH probe failed (exit 255)" is transient only for reconnect - Watcher markers survive client invalidation when switching SSH connections - Add network error patterns: "lost connection", "remote end closed" (cherry picked from commit ce5b639e03ba49fd83563488986721615f0f069b) --- src/main/ipc/ssh.ts | 21 +- .../ssh-connect-attempt-cancellation.test.ts | 28 ++ .../ssh/ssh-connect-attempt-cancellation.ts | 16 ++ src/main/ssh/ssh-connection-utils.test.ts | 12 + src/main/ssh/ssh-connection-utils.ts | 1 + src/main/ssh/ssh-connection.test.ts | 219 +++++++++++++++ src/main/ssh/ssh-connection.ts | 63 ++++- ...ssh-reconnect-error-classification.test.ts | 94 +++++++ .../ssh/ssh-reconnect-error-classification.ts | 83 ++++++ src/relay/dispatcher-client-writer.ts | 4 + src/relay/dispatcher-writer-admission.ts | 12 +- src/relay/dispatcher.ts | 80 +++++- .../relay-filesystem-watch-registry.test.ts | 5 +- src/relay/relay-watcher-event-emitter.test.ts | 261 +++++++++++++++++- src/relay/relay-watcher-event-emitter.ts | 153 ++++++++-- 15 files changed, 996 insertions(+), 56 deletions(-) create mode 100644 src/main/ssh/ssh-connect-attempt-cancellation.test.ts create mode 100644 src/main/ssh/ssh-connect-attempt-cancellation.ts create mode 100644 src/main/ssh/ssh-reconnect-error-classification.test.ts create mode 100644 src/main/ssh/ssh-reconnect-error-classification.ts diff --git a/src/main/ipc/ssh.ts b/src/main/ipc/ssh.ts index 0227b84f36a..91234031aa3 100644 --- a/src/main/ipc/ssh.ts +++ b/src/main/ipc/ssh.ts @@ -21,6 +21,7 @@ import type { import { SSH_TERMINATE_RECONNECT_REQUIRED } from '../../shared/constants' import { isRuntimeOwnedSshTargetId } from '../../shared/execution-host' import { isAuthError } from '../ssh/ssh-connection-utils' +import { createCancelledConnectAttemptError } from '../ssh/ssh-connect-attempt-cancellation' import { forceStopRelayForTarget } from '../ssh/ssh-relay-reset' import { isSshPtyNotFoundError } from '../providers/ssh-pty-errors' import { toAppSshPtyId, toRelaySshPtyId } from '../providers/ssh-pty-id' @@ -272,10 +273,6 @@ function isCurrentConnectAttempt(targetId: string, authority: DirectSshAuthority return authority.targetId === targetId && isCurrentSshProviderAuthority(authority) } -function connectCancelledError(): Error { - return new Error('SSH connection attempt was cancelled') -} - // Why: publish reset's teardown/force-stop/disconnect lifecycle so new connects and duplicate resets can't race it. const resetRelayInFlight = new Map>() @@ -974,7 +971,7 @@ export function registerSshHandlers( } } if (!isCurrentConnectAttempt(targetId, admissionAuthority)) { - throw connectCancelledError() + throw createCancelledConnectAttemptError() } const observedAuthority = admissionAuthority if (existing) { @@ -984,7 +981,7 @@ export function registerSshHandlers( } } if (!isCurrentSshProviderAuthority(observedAuthority)) { - throw connectCancelledError() + throw createCancelledConnectAttemptError() } pendingTransportReconnects.delete(targetId) @@ -1047,7 +1044,7 @@ export function registerSshHandlers( // Why: await port teardown before disposing, else the new session's restorePortForwards can hit EADDRINUSE on not-yet-released ports. await portForwardManager!.removeAllForwards(targetId) if (!isCurrentConnectAttempt(targetId, authority)) { - throw connectCancelledError() + throw createCancelledConnectAttemptError() } existingSession.detach() if (activeSessions.get(targetId) === existingSession) { @@ -1063,7 +1060,7 @@ export function registerSshHandlers( throw disconnectResult.error } if (!isCurrentConnectAttempt(targetId, authority)) { - throw connectCancelledError() + throw createCancelledConnectAttemptError() } } @@ -1084,14 +1081,14 @@ export function registerSshHandlers( try { conn = await connectionManager!.connect(target) if (!ownsSession()) { - throw connectCancelledError() + throw createCancelledConnectAttemptError() } } catch (err) { // Why: connect()'s internal state may not have reached the renderer; broadcast explicitly so the UI leaves 'connecting'. const errObj = err instanceof Error ? err : new Error(String(err)) const status: SshConnectionStatus = isAuthError(errObj) ? 'auth-failed' : 'error' if (!ownsSession()) { - throw connectCancelledError() + throw createCancelledConnectAttemptError() } // Why: clear this failed connect's flag so a later non-prompting connect isn't deferred. credentialRequestedForTarget.delete(targetId) @@ -1117,7 +1114,7 @@ export function registerSshHandlers( await session.establish(conn, relayGracePeriodForTarget(target)) if (!ownsSession()) { - throw connectCancelledError() + throw createCancelledConnectAttemptError() } // Why: we manually pushed `deploying-relay`, so send `connected` straight to the renderer — routing through onStateChange would trigger reconnect logic. @@ -1131,7 +1128,7 @@ export function registerSshHandlers( }) } catch (err) { if (!ownsSession()) { - throw connectCancelledError() + throw createCancelledConnectAttemptError() } activeSessions.delete(targetId) clearRelayLostBackoff(targetId) diff --git a/src/main/ssh/ssh-connect-attempt-cancellation.test.ts b/src/main/ssh/ssh-connect-attempt-cancellation.test.ts new file mode 100644 index 00000000000..b98b54762ad --- /dev/null +++ b/src/main/ssh/ssh-connect-attempt-cancellation.test.ts @@ -0,0 +1,28 @@ +import { describe, it, expect } from 'vitest' +import { + createCancelledConnectAttemptError, + isCancelledConnectAttemptError +} from './ssh-connect-attempt-cancellation' +import { isAuthError } from './ssh-connection-utils' +import { isTransientReconnectError } from './ssh-reconnect-error-classification' + +describe('createCancelledConnectAttemptError', () => { + it('is recognisable by identity, not just by message text', () => { + const error = createCancelledConnectAttemptError() + expect(isCancelledConnectAttemptError(error)).toBe(true) + // A cancelled attempt must never look like a permanent failure to the reconnect ladder. + expect(isTransientReconnectError(error)).toBe(false) + expect(isAuthError(error)).toBe(false) + }) + + it('does not classify unrelated errors as cancellations', () => { + expect(isCancelledConnectAttemptError(new Error('connect ETIMEDOUT'))).toBe(false) + }) + + it('ignores a lookalike message from a producer that skipped the factory', () => { + // Every real producer imports the factory, so message text alone must not qualify. + expect( + isCancelledConnectAttemptError(new Error(createCancelledConnectAttemptError().message)) + ).toBe(false) + }) +}) diff --git a/src/main/ssh/ssh-connect-attempt-cancellation.ts b/src/main/ssh/ssh-connect-attempt-cancellation.ts new file mode 100644 index 00000000000..0229f827ae0 --- /dev/null +++ b/src/main/ssh/ssh-connect-attempt-cancellation.ts @@ -0,0 +1,16 @@ +const CANCELLED_CONNECT_ATTEMPT_MESSAGE = 'SSH connection attempt was cancelled' +const CANCELLED_CONNECT_ATTEMPT_NAME = 'SshConnectAttemptCancelledError' + +// Why: cancellation means a newer attempt (or disconnect) already owns the connection, so it must be +// recognisable by identity — publishing its message as a permanent error hides the live attempt. +export function createCancelledConnectAttemptError(): Error { + const error = new Error(CANCELLED_CONNECT_ATTEMPT_MESSAGE) + error.name = CANCELLED_CONNECT_ATTEMPT_NAME + return error +} + +// Identity only: every producer goes through createCancelledConnectAttemptError, so a message match +// would add no reach and would misclassify an unrelated error that happens to share the wording. +export function isCancelledConnectAttemptError(err: Error): boolean { + return err.name === CANCELLED_CONNECT_ATTEMPT_NAME +} diff --git a/src/main/ssh/ssh-connection-utils.test.ts b/src/main/ssh/ssh-connection-utils.test.ts index 5e06a9e7386..eba0de23087 100644 --- a/src/main/ssh/ssh-connection-utils.test.ts +++ b/src/main/ssh/ssh-connection-utils.test.ts @@ -233,9 +233,21 @@ describe('isAuthError', () => { ) }) + it.each([ + 'Permission denied (publickey).', + 'Permission denied (publickey,password).', + 'Permission denied, please try again.' + ])('detects OpenSSH credential rejection: %s', (message) => { + expect(isAuthError(new Error(message))).toBe(true) + }) + it('returns false for transient errors', () => { expect(isAuthError(new Error('connect ETIMEDOUT'))).toBe(false) }) + + it('does not classify a local filesystem permission failure as authentication', () => { + expect(isAuthError(new Error('Permission denied (os error 13)'))).toBe(false) + }) }) // ── isAgentFallbackError ──────────────────────────────────────────── diff --git a/src/main/ssh/ssh-connection-utils.ts b/src/main/ssh/ssh-connection-utils.ts index bce298039be..77cb5ab07a5 100644 --- a/src/main/ssh/ssh-connection-utils.ts +++ b/src/main/ssh/ssh-connection-utils.ts @@ -52,6 +52,7 @@ export function isAuthError(err: Error): boolean { msg.includes('all configured authentication methods failed') || msg.includes('authentication failed') || msg.includes('too many authentication failures') || + /permission denied(?:, please try again\.?| \([^)]*(?:publickey|password|keyboard-interactive|gssapi|hostbased)[^)]*\))/.test(msg) || (err as { level?: string }).level === 'client-authentication' ) } diff --git a/src/main/ssh/ssh-connection.test.ts b/src/main/ssh/ssh-connection.test.ts index 25bcaf447af..dc8cdc42250 100644 --- a/src/main/ssh/ssh-connection.test.ts +++ b/src/main/ssh/ssh-connection.test.ts @@ -275,6 +275,15 @@ function createFailingSystemCommandChannel( return channel } +// A probe that never answers — the shape a FIDO2/system-transport host takes when the network drops. +function createHangingSystemCommandChannel(): ReturnType { + const channel = new EventEmitter() as ReturnType + channel.stdin = { end: vi.fn(), write: vi.fn() } + channel.stderr = new EventEmitter() + channel.close = vi.fn() + return channel +} + function createPendingSystemSshProcess() { const stdout = new EventEmitter() return { @@ -652,6 +661,92 @@ describe('SshConnection', () => { } }) + it('keeps a system-transport target on the ladder after a probe timeout', async () => { + vi.useFakeTimers() + try { + vi.mocked(resolveWithSshG).mockResolvedValue(createResolvedConfig()) + const statuses: string[] = [] + const conn = new SshConnection( + createTarget(), + createCallbacks({ + onStateChange: vi.fn((_id, state) => statuses.push(state.status)) + }) + ) + await conn.connect() + expect(conn.usesSystemSshTransport()).toBe(true) + + // The probe times out with OpenSSH prose, not an errno the transient code table matches. + spawnSystemSshCommandMock.mockImplementation(() => createHangingSystemCommandChannel()) + statuses.length = 0 + const reconnected = conn.reconnect() + await vi.advanceTimersByTimeAsync(CONNECT_TIMEOUT_MS) + await reconnected + + // Shipped published 'error' here, stranding FIDO2 hosts with no path back short of a restart. + expect(statuses).not.toContain('error') + expect(conn.getState().status).toBe('reconnecting') + + const probesBefore = spawnSystemSshCommandMock.mock.calls.length + await vi.advanceTimersByTimeAsync(RECONNECT_BACKOFF_MS[0]) + expect(spawnSystemSshCommandMock.mock.calls.length).toBeGreaterThan(probesBefore) + } finally { + vi.useRealTimers() + } + }) + + it('publishes no cancellation error when a connect supersedes a parked ladder attempt', async () => { + const published: { status: string; error?: string }[] = [] + let releaseParked: (() => void) | null = null + vi.mocked(resolveWithSshG).mockImplementationOnce( + () => + new Promise((resolve) => { + releaseParked = () => resolve(null) + }) + ) + const conn = new SshConnection( + createTarget(), + createCallbacks({ + onStateChange: vi.fn((_id, state) => published.push({ status: state.status, ...state })) + }) + ) + + // The ladder attempt parks inside ssh -G while the user presses Connect. + const parked = conn.reconnect() + await conn.connect() + releaseParked!() + await parked + + expect(conn.getState().status).toBe('connected') + // Shipped published error:'SSH connection attempt was cancelled' over a live connection. + expect(published.filter((entry) => entry.status === 'error')).toEqual([]) + expect(published.map((entry) => entry.error).filter(Boolean)).toEqual([]) + }) + + it('rejects a superseded connect without publishing a permanent error', async () => { + const published: string[] = [] + let releaseParked: (() => void) | null = null + vi.mocked(resolveWithSshG).mockImplementationOnce( + () => + new Promise((resolve) => { + releaseParked = () => resolve(null) + }) + ) + const conn = new SshConnection( + createTarget(), + createCallbacks({ + onStateChange: vi.fn((_id, state) => published.push(state.status)) + }) + ) + + const superseded = conn.connect() + await conn.connect() + releaseParked!() + + await expect(superseded).rejects.toThrow('SSH connection attempt was cancelled') + expect(conn.getState().status).toBe('connected') + expect(published).not.toContain('error') + }) + it('transitions through connecting → connected states', async () => { const states: string[] = [] const callbacks = createCallbacks({ @@ -1686,6 +1781,101 @@ describe('SshConnection', () => { expect(conn.canRunConcurrentExecCommands()).toBe(false) }) + it('retries a generic system SSH probe timeout without ControlMaster', async () => { + vi.useFakeTimers() + try { + getOrcaControlSocketPathMock.mockImplementation( + (_target: SshTarget, options?: { disableControlMaster?: boolean }) => + options?.disableControlMaster ? null : '/tmp/orca-ssh-501/stale-socket' + ) + spawnSystemSshCommandMock + .mockImplementationOnce(() => createHangingSystemCommandChannel()) + .mockImplementation(() => createSystemCommandChannel()) + vi.mocked(resolveWithSshG).mockResolvedValue(createResolvedConfig()) + const conn = new SshConnection(createTarget({ configHost: 'fdpass-host' }), createCallbacks()) + + const settled = conn.connect() + await vi.advanceTimersByTimeAsync(CONNECT_TIMEOUT_MS) + + await expect(settled).resolves.toBeUndefined() + expect(spawnSystemSshCommandMock).toHaveBeenCalledTimes(2) + expect(removeControlSocketPathMock).toHaveBeenCalledWith( + '/tmp/orca-ssh-501/stale-socket' + ) + expect(spawnSystemSshCommandMock).toHaveBeenNthCalledWith( + 2, + expect.anything(), + 'echo ORCA-SYSTEM-SSH-OK', + expect.objectContaining({ disableControlMaster: true }) + ) + } finally { + vi.useRealTimers() + } + }) + + it('skips a second probe for a definite host failure', async () => { + getOrcaControlSocketPathMock.mockReturnValue('/tmp/orca-ssh-501/stale-socket') + spawnSystemSshCommandMock.mockImplementation(() => + createFailingSystemCommandChannel( + 255, + 'ssh: connect to host box port 22: No route to host' + ) + ) + vi.mocked(resolveWithSshG).mockResolvedValue(createResolvedConfig()) + const conn = new SshConnection(createTarget({ configHost: 'fdpass-host' }), createCallbacks()) + + await expect(conn.connect()).rejects.toThrow('No route to host') + expect(spawnSystemSshCommandMock).toHaveBeenCalledTimes(1) + expect(removeControlSocketPathMock).toHaveBeenCalledWith('/tmp/orca-ssh-501/stale-socket') + }) + + it.each([ + 'Permission denied (publickey,password).', + 'Encrypted private key detected, but no passphrase given' + ])('skips a second probe for terminal credential failures: %s', async (stderr) => { + getOrcaControlSocketPathMock.mockReturnValue('/tmp/orca-ssh-501/stale-socket') + spawnSystemSshCommandMock.mockImplementation(() => + createFailingSystemCommandChannel(255, stderr) + ) + vi.mocked(resolveWithSshG).mockResolvedValue(createResolvedConfig()) + const conn = new SshConnection(createTarget({ configHost: 'fdpass-host' }), createCallbacks()) + + await expect(conn.connect()).rejects.toThrow(stderr) + expect(spawnSystemSshCommandMock).toHaveBeenCalledTimes(1) + expect(removeControlSocketPathMock).toHaveBeenCalledWith('/tmp/orca-ssh-501/stale-socket') + }) + + it('retries a generic direct system SSH timeout without ControlMaster', async () => { + vi.useFakeTimers() + try { + getOrcaControlSocketPathMock.mockImplementation( + (_target: SshTarget, options?: { disableControlMaster?: boolean }) => + options?.disableControlMaster ? null : '/tmp/orca-ssh-501/stale-socket' + ) + spawnSystemSshMock + .mockImplementationOnce(() => createPendingSystemSshProcess()) + .mockImplementation(() => createSystemSshProcess()) + vi.mocked(resolveWithSshG).mockResolvedValue(createResolvedConfig()) + const conn = new SshConnection(createTarget({ configHost: 'fdpass-host' }), createCallbacks()) + + const settled = conn.connectViaSystemSsh() + await vi.advanceTimersByTimeAsync(CONNECT_TIMEOUT_MS) + + await expect(settled).resolves.toBeDefined() + expect(spawnSystemSshMock).toHaveBeenCalledTimes(2) + expect(removeControlSocketPathMock).toHaveBeenCalledWith( + '/tmp/orca-ssh-501/stale-socket' + ) + expect(spawnSystemSshMock).toHaveBeenNthCalledWith( + 2, + expect.anything(), + expect.objectContaining({ disableControlMaster: true }) + ) + } finally { + vi.useRealTimers() + } + }) + it('uses system SSH transport for ProxyCommand targets before ssh2 auth', async () => { const conn = new SshConnection( createTarget({ proxyCommand: 'ssh -W %h:%p bastion.example.com' }), @@ -1985,6 +2175,19 @@ describe('SshConnection', () => { expect(spawnSystemSshCommandMock).not.toHaveBeenCalled() }) + it('publishes auth-failed when OpenSSH denies reconnect credentials', async () => { + vi.mocked(resolveWithSshG).mockResolvedValue(createResolvedConfig()) + const conn = new SshConnection(createTarget({ configHost: 'fdpass-host' }), createCallbacks()) + await conn.connect() + + spawnSystemSshCommandMock.mockImplementation(() => + createFailingSystemCommandChannel(255, 'Permission denied (publickey,password).') + ) + await conn.reconnect() + + expect(conn.getState().status).toBe('auth-failed') + }) + it('clears system SSH transport when the GSSAPI-first probe throws synchronously', async () => { // Why: no system ssh binary makes spawnSystemSshCommand throw before the // probe's try/catch, so the ssh2 fall-through must still reset the flag — @@ -2275,6 +2478,22 @@ describe('SshConnection', () => { expect(conn.canRunConcurrentExecCommands()).toBe(false) }) + it.each([ + 'Permission denied (publickey,password).', + 'Encrypted private key detected, but no passphrase given' + ])('skips a direct retry for terminal credential failures: %s', async (message) => { + getOrcaControlSocketPathMock.mockReturnValue('/tmp/orca-ssh-501/stale-socket') + spawnSystemSshMock.mockImplementation(() => { + throw new Error(message) + }) + vi.mocked(resolveWithSshG).mockResolvedValueOnce(createResolvedConfig()) + const conn = new SshConnection(createTarget({ configHost: 'fdpass-host' }), createCallbacks()) + + await expect(conn.connectViaSystemSsh()).rejects.toThrow(message) + expect(spawnSystemSshMock).toHaveBeenCalledTimes(1) + expect(removeControlSocketPathMock).toHaveBeenCalledWith('/tmp/orca-ssh-501/stale-socket') + }) + it('kills delayed direct system SSH startup on disconnect and ignores late stdout', async () => { const proc = createPendingSystemSshProcess() spawnSystemSshMock.mockReturnValueOnce(proc) diff --git a/src/main/ssh/ssh-connection.ts b/src/main/ssh/ssh-connection.ts index f183a704975..0016491e2b0 100644 --- a/src/main/ssh/ssh-connection.ts +++ b/src/main/ssh/ssh-connection.ts @@ -40,6 +40,14 @@ import { type SshExecOptions, type SshConnectionCallbacks } from './ssh-connection-utils' +import { + createCancelledConnectAttemptError, + isCancelledConnectAttemptError +} from './ssh-connect-attempt-cancellation' +import { + isDefiniteSystemSshHostFailure, + isTransientReconnectError +} from './ssh-reconnect-error-classification' import { SshReconnectLadder } from './ssh-reconnect-ladder' import { getPassphrasePrivateKeyPath } from './ssh-private-key-authentication' import { @@ -601,8 +609,9 @@ export class SshConnection { let lastError: Error | null = null for (let attempt = 0; attempt < INITIAL_RETRY_ATTEMPTS; attempt++) { + const connectGeneration = ++this.connectGeneration try { - await this.attemptConnect() + await this.attemptConnect(connectGeneration) this.reconnectLadder.reset() this.reconnectLadder.markConnected(Date.now()) return @@ -614,6 +623,14 @@ export class SshConnection { throw lastError } + // Why: a newer attempt owns the connection now and publishes its own outcome; retrying here would fight it. + if ( + !this.isCurrentConnectAttempt(connectGeneration) || + isCancelledConnectAttemptError(lastError) + ) { + throw lastError + } + if (isAuthError(lastError) || isPassphraseError(lastError)) { this.setState('auth-failed', lastError.message) throw lastError @@ -635,11 +652,11 @@ export class SshConnection { throw finalError } - private async attemptConnect(): Promise { + // Why: callers claim the generation so their catch can tell their own failure from a superseded one. + private async attemptConnect(connectGeneration = ++this.connectGeneration): Promise { this.setState('connecting') this.proxyProcess?.kill() this.proxyProcess = null - const connectGeneration = ++this.connectGeneration const resolved = await resolveWithSshG(this.target.configHost || this.target.label).catch( () => null @@ -904,7 +921,7 @@ export class SshConnection { const onClose = (code: number | null): void => { settle(() => { if (this.disposed || connectGeneration !== this.connectGeneration) { - reject(new Error('SSH connection attempt was cancelled')) + reject(this.createCancelledConnectAttemptError()) return } if ( @@ -961,10 +978,21 @@ export class SshConnection { try { await this.doSystemSshProbe(connectGeneration) } catch (err) { - if (!controlPath || this.disposed || connectGeneration !== this.connectGeneration) { + if ( + !controlPath || + this.disposed || + connectGeneration !== this.connectGeneration + ) { throw err } removeControlSocketPath(controlPath) + // Why: a timeout retry doubles the ladder step, but a stuck master can cause the first timeout. + if ( + isDefiniteSystemSshHostFailure(err) || + (err instanceof Error && (isAuthError(err) || isPassphraseError(err))) + ) { + throw err + } this.systemSshResolvedConfig = cloneResolvedConfig(resolved) this.systemSshControlMasterDisabledForSession = true try { @@ -990,6 +1018,12 @@ export class SshConnection { throw err } removeControlSocketPath(controlPath) + if ( + isDefiniteSystemSshHostFailure(err) || + (err instanceof Error && (isAuthError(err) || isPassphraseError(err))) + ) { + throw err + } this.systemSshControlMasterDisabledForSession = true if (!this.isCurrentConnectAttempt(connectGeneration)) { throw this.createCancelledConnectAttemptError() @@ -1087,7 +1121,7 @@ export class SshConnection { } private createCancelledConnectAttemptError(): Error { - return new Error('SSH connection attempt was cancelled') + return createCancelledConnectAttemptError() } private spawnTrackedSystemSshCommand(command: string, options?: SshExecOptions): ClientChannel { @@ -1182,7 +1216,7 @@ export class SshConnection { cleanupStartupListeners() client.end() client.destroy() - reject(new Error('SSH connection attempt was cancelled')) + reject(this.createCancelledConnectAttemptError()) return } settled = true @@ -1264,22 +1298,31 @@ export class SshConnection { } private async runReconnectAttempt(): Promise { + const connectGeneration = ++this.connectGeneration try { // Why: reset before connecting so the 'connected' broadcast carries reconnectAttempt=0, which ssh.ts uses to trigger relay re-establishment. this.state.reconnectAttempt = 0 - await this.attemptConnect() + await this.attemptConnect(connectGeneration) + if (!this.isCurrentConnectAttempt(connectGeneration)) { + return + } // Why: attemptConnect resolves only after a real handshake on either transport; the system-ssh probe's 'connected' must not clear the failure streak. this.reconnectLadder.markConnected(Date.now()) } catch (err) { - if (this.disposed) { + // Why: a superseded attempt has no outcome to publish — the attempt that claimed the generation owns the state, and cancellation is that supersession. + if (this.disposed || !this.isCurrentConnectAttempt(connectGeneration)) { return } const error = err instanceof Error ? err : new Error(String(err)) + if (isCancelledConnectAttemptError(error)) { + return + } if (isAuthError(error) || isPassphraseError(error)) { this.setState('auth-failed', error.message) return } - if (!isTransientError(error)) { + // Why: the system-SSH transport reports timeouts as prose, so the narrow code table would strand FIDO2 targets in 'error' forever. + if (!isTransientReconnectError(error)) { this.setState('error', error.message) return } diff --git a/src/main/ssh/ssh-reconnect-error-classification.test.ts b/src/main/ssh/ssh-reconnect-error-classification.test.ts new file mode 100644 index 00000000000..cf8319db03c --- /dev/null +++ b/src/main/ssh/ssh-reconnect-error-classification.test.ts @@ -0,0 +1,94 @@ +import { describe, it, expect } from 'vitest' +import { isTransientError } from './ssh-connection-utils' +import { + isDefiniteSystemSshHostFailure, + isTransientReconnectError +} from './ssh-reconnect-error-classification' + +describe('isTransientReconnectError', () => { + it('treats the system SSH connect timeout as recoverable', () => { + const err = new Error('System SSH connection timed out') + // Guards the split: widening isTransientError would spend 5 connect() attempts on this. + expect(isTransientError(err)).toBe(false) + expect(isTransientReconnectError(err)).toBe(true) + }) + + it.each([ + 'System SSH probe failed (exit 255).', + 'System SSH probe failed (exit 255). stderr: ssh: connect to host box port 22: Connection refused', + 'System SSH probe failed (exit 255). stderr: ssh: connect to host box port 22: No route to host', + 'System SSH probe failed (exit 255). stderr: ssh: connect to host box port 22: Network is unreachable', + 'System SSH probe failed (exit 255). stderr: kex_exchange_identification: read: Connection reset by peer', + 'System SSH probe failed (exit 255). stderr: ssh: Could not resolve hostname box: Name or service not known', + 'System SSH probe failed (exit 255). stderr: ssh_exchange_identification: read: Connection reset by peer', + 'System SSH probe failed (exit 255). stderr: ssh_exchange_identification: Connection closed by remote host', + 'System SSH probe failed (exit 255). stderr: Connection closed by remote host', + 'System SSH probe failed (exit 255). stderr: Lost connection', + 'System SSH probe failed (exit 255). stderr: Remote end closed connection' + ])('treats OpenSSH network prose as recoverable: %s', (message) => { + expect(isTransientReconnectError(new Error(message))).toBe(true) + }) + + it('keeps the bare probe failure narrow to the reconnect classifier', () => { + const err = new Error('System SSH probe failed (exit 255).') + expect(isTransientError(err)).toBe(false) + expect(isTransientReconnectError(err)).toBe(true) + }) + + // A server-side rejection prints the same verb without "remote"; retrying it forever would hide a + // real misconfiguration behind the ladder. + it('keeps a bare "Connection closed by port " permanent', () => { + expect( + isTransientReconnectError( + new Error( + 'System SSH probe failed (exit 255). stderr: Connection closed by 10.0.0.4 port 22' + ) + ) + ).toBe(false) + }) + + it('keeps credential failures permanent', () => { + expect( + isTransientReconnectError(new Error('All configured authentication methods failed')) + ).toBe(false) + expect( + isTransientReconnectError( + new Error('System SSH probe failed (exit 255). stderr: Permission denied (publickey).') + ) + ).toBe(false) + expect( + isTransientReconnectError( + new Error('Encrypted private OpenSSH key detected, but no passphrase given') + ) + ).toBe(false) + }) + + it.each([ + 'ssh: connect to host box port 22: Connection refused', + 'ssh: connect to host box port 22: Connection reset by peer', + 'ssh: connect to host box port 22: Connection timed out', + 'ssh: connect to host box port 22: No route to host', + 'ssh: Could not resolve hostname box: Name or service not known' + ])('recognizes definite host failures for the ControlMaster retry gate: %s', (message) => { + expect(isDefiniteSystemSshHostFailure(new Error(message))).toBe(true) + }) + + it.each([ + 'System SSH connection timed out', + 'Connection refused', + 'Connection reset by peer', + 'mux client failed: master is unresponsive' + ])('keeps potentially mux-shaped failures eligible for direct retry: %s', (message) => { + expect(isDefiniteSystemSshHostFailure(new Error(message))).toBe(false) + }) + + it('still covers the errno codes isTransientError already matched', () => { + const err = new Error('refused') as NodeJS.ErrnoException + err.code = 'ECONNREFUSED' + expect(isTransientReconnectError(err)).toBe(true) + }) + + it('keeps unrelated failures permanent', () => { + expect(isTransientReconnectError(new Error('something went wrong'))).toBe(false) + }) +}) diff --git a/src/main/ssh/ssh-reconnect-error-classification.ts b/src/main/ssh/ssh-reconnect-error-classification.ts new file mode 100644 index 00000000000..dce71dc6dae --- /dev/null +++ b/src/main/ssh/ssh-reconnect-error-classification.ts @@ -0,0 +1,83 @@ +import { isAuthError, isPassphraseError, isTransientError } from './ssh-connection-utils' + +// Why: the system-SSH transport reports network failures as OpenSSH prose, not errno codes, so its +// probe timeouts and connect failures never match isTransientError's code table. +const NETWORK_LIKE_ERROR_FRAGMENTS = [ + 'system ssh connection timed out', + 'timed out while waiting for handshake', + 'connection timed out', + 'operation timed out', + 'connection refused', + 'connection reset', + 'no route to host', + 'network is unreachable', + 'network is down', + 'host is down', + 'broken pipe', + 'temporary failure in name resolution', + 'name or service not known', + 'nodename nor servname', + 'could not resolve hostname', + 'kex_exchange_identification', + // Pre-7.x OpenSSH wording for the same banner-exchange failure. + 'ssh_exchange_identification', + 'lost connection', + 'remote end closed', + // Deliberately not the bare 'connection closed by': OpenSSH prints "Connection closed by port 22" + // for server-side rejections (MaxStartups, DenyUsers) too, and those must stay permanent. + 'connection closed by remote' +] + +const DEFINITE_HOST_FAILURE_FRAGMENTS = [ + 'no route to host', + 'network is unreachable', + 'network is down', + 'host is down', + 'temporary failure in name resolution', + 'name or service not known', + 'nodename nor servname', + 'could not resolve hostname' +] + +const DEFINITE_HOST_FAILURE_CODES = new Set(['EHOSTUNREACH', 'ENETUNREACH', 'EAI_AGAIN']) + +const OPENSSH_HOST_CONNECT_FAILURE = + /\bconnect to host .+? port \d+: (?:connection (?:timed out|refused|reset(?: by peer)?)|operation timed out)\b/i + +const EMPTY_SYSTEM_SSH_PROBE_EXIT_255 = /^system ssh probe failed \(exit 255\)\.$/ + +/** + * Why a second classifier: connect() spends isTransientError on INITIAL_RETRY_ATTEMPTS, so widening + * it would turn one 30s system-SSH timeout into ~160s of silence and five security-key touch + * prompts. The reconnect ladder has its own backoff and give-up bound, so it can afford to treat a + * network-shaped failure as recoverable. Auth stays permanent on both paths. + */ +export function isTransientReconnectError(err: Error): boolean { + if (isAuthError(err) || isPassphraseError(err)) { + return false + } + if (isTransientError(err)) { + return true + } + const message = err.message.toLowerCase() + return ( + EMPTY_SYSTEM_SSH_PROBE_EXIT_255.test(message) || + NETWORK_LIKE_ERROR_FRAGMENTS.some((fragment) => message.includes(fragment)) + ) +} + +/** Definite host failures where a ControlMaster-free retry cannot help. */ +export function isDefiniteSystemSshHostFailure(err: unknown): boolean { + if (!(err instanceof Error)) { + return false + } + const code = (err as NodeJS.ErrnoException).code + if (code && DEFINITE_HOST_FAILURE_CODES.has(code)) { + return true + } + const message = err.message.toLowerCase() + return ( + DEFINITE_HOST_FAILURE_FRAGMENTS.some((fragment) => message.includes(fragment)) || + OPENSSH_HOST_CONNECT_FAILURE.test(err.message) + ) +} diff --git a/src/relay/dispatcher-client-writer.ts b/src/relay/dispatcher-client-writer.ts index 58be59173cf..dd9a2e3b245 100644 --- a/src/relay/dispatcher-client-writer.ts +++ b/src/relay/dispatcher-client-writer.ts @@ -59,6 +59,10 @@ export class DispatcherClientWriter { return this.sink.producerFrameCapacity } + canEnqueueControl(bytes: number): boolean { + return !this.closed && this.admission.canAdmitControl(bytes) + } + get fixedFrameCapacity(): number { return this.sink.frameCapacity('fixed-bulk', this.saturated) } diff --git a/src/relay/dispatcher-writer-admission.ts b/src/relay/dispatcher-writer-admission.ts index 07db61c12e8..27504697f9f 100644 --- a/src/relay/dispatcher-writer-admission.ts +++ b/src/relay/dispatcher-writer-admission.ts @@ -70,6 +70,13 @@ export class DispatcherWriterAdmission { return this.producerBytes } + canAdmitControl(bytes: number): boolean { + return ( + this.controlFrames < DISPATCHER_CONTROL_QUEUE_MAX_FRAMES && + this.controlBytes + bytes <= DISPATCHER_CONTROL_QUEUE_MAX_BYTES + ) + } + get queuedEntries(): number { return Object.values(this.queues).reduce((total, queue) => total + queue.length, 0) } @@ -147,10 +154,7 @@ export class DispatcherWriterAdmission { private admitControl(entry: DispatcherWriterEntry): AdmissionResult { // Why: best-effort controls may fail soft without weakening protocol-critical admission. const overflowIsNonFatal = entry.overflowIsNonFatal === true - if ( - this.controlFrames >= DISPATCHER_CONTROL_QUEUE_MAX_FRAMES || - this.controlBytes + entry.estimatedBytes > DISPATCHER_CONTROL_QUEUE_MAX_BYTES - ) { + if (!this.canAdmitControl(entry.estimatedBytes)) { if (overflowIsNonFatal) { return { accepted: false } } diff --git a/src/relay/dispatcher.ts b/src/relay/dispatcher.ts index e8766dc347d..98e8959547f 100644 --- a/src/relay/dispatcher.ts +++ b/src/relay/dispatcher.ts @@ -104,6 +104,7 @@ export class RelayDispatcher { private clientDetachListeners = new Set<(clientId: number) => void>() private disposeListeners = new Set<() => void>() private legacyCapacityListeners = new Set<() => void>() + private clientCapacityListeners = new Map void>>() private publicationTransactionDepth = 0 private deferredLegacyCapacity = false private deferredForcedLegacyCapacity = false @@ -131,6 +132,8 @@ export class RelayDispatcher { this.primaryClient.writer.close(new Error('Relay primary sink replaced')) this.resetClient(this.primaryClient) this.primaryClient.writer = this.createWriter(this.primaryClient, write, sinkOptions) + // Why: a frame retained against the replaced sink would otherwise wait for traffic that may never come. + this.notifyClientCapacity(this.primaryClient.id) } // Why: mark in-flight requests stale on disconnect so a late pty.spawn/fs.watch can't create unowned remote state. @@ -189,6 +192,48 @@ export class RelayDispatcher { return () => this.legacyCapacityListeners.delete(listener) } + /** + * Ungated per-client writer capacity, for a frame that lost control-lane admission and must retry. + * onLegacyPtyCapacity cannot serve that: it is gated on producer retention, so it stays silent + * exactly under the dual-queue pressure that rejected the frame. Registered on the dispatcher, not + * the writer, so a retry survives setWrite() replacing the primary sink. + * Returns null only when the client is gone for good. A merely closed client still counts: setWrite + * marks the primary closed before replacing its sink, and that replacement is exactly when a frame + * stranded by the old writer must be armed to retry. + */ + onClientCapacity(clientId: number, listener: () => void): (() => void) | null { + const client = this.clients.get(clientId) + if (this.disposed || !client) { + return null + } + const listeners = this.clientCapacityListeners.get(clientId) ?? new Set<() => void>() + listeners.add(listener) + this.clientCapacityListeners.set(clientId, listeners) + return () => { + const current = this.clientCapacityListeners.get(clientId) + if (!current?.delete(listener) || current.size > 0) { + return + } + this.clientCapacityListeners.delete(clientId) + } + } + + /** + * Whether the id still names a client. A detach notification does not always mean it stopped: + * invalidateClient() detaches the primary without removing it, and setWrite() revives that same id. + */ + isClientAttached(clientId: number): boolean { + return !this.disposed && this.clients.has(clientId) + } + + canAdmitControlFrame(clientId: number, estimatedBytes: number): boolean { + const client = this.clients.get(clientId) + if (this.disposed || !client || client.closed) { + return false + } + return client.writer.canEnqueueControl(estimatedBytes) + } + get legacyRetentionBelowLowWater(): boolean { return this.publicationLedger.belowLowWater(this.activeClientKeys()) } @@ -563,7 +608,10 @@ export class RelayDispatcher { method: string, params?: Record, onSettled: (result: SinkWriteSettlement) => void = () => {}, - options: { controlOverflow?: 'close-client' | 'reject' } = {} + options: { + controlOverflow?: 'close-client' | 'reject' + estimatedBytes?: number + } = {} ): boolean { if (this.disposed) { onSettled({ ok: false, error: new Error('Relay dispatcher is disposed') }) @@ -583,7 +631,7 @@ export class RelayDispatcher { }, 'control', onSettled, - undefined, + options.estimatedBytes, options.controlOverflow ) } @@ -727,6 +775,7 @@ export class RelayDispatcher { listener() } this.legacyCapacityListeners.clear() + this.clientCapacityListeners.clear() for (const listener of Array.from(this.disposeListeners)) { listener() } @@ -1187,10 +1236,29 @@ export class RelayDispatcher { const writer = new DispatcherClientWriter(write, sinkOptions, (error) => { this.closeClient(client, error, client !== this.primaryClient) }) - writer.onCapacity(() => this.notifyLegacyCapacityIfLow()) + writer.onCapacity(() => { + this.notifyLegacyCapacityIfLow() + this.notifyClientCapacity(client.id) + }) return writer } + private notifyClientCapacity(clientId: number): void { + const listeners = this.clientCapacityListeners.get(clientId) + if (!listeners?.size) { + return + } + for (const listener of Array.from(listeners)) { + try { + listener() + } catch (err) { + process.stderr.write( + `[relay] Client capacity listener failed: ${err instanceof Error ? err.message : String(err)}\n` + ) + } + } + } + private closeClient(client: RelayClient, error: Error, remove: boolean): void { if (client.closed) { return @@ -1203,6 +1271,12 @@ export class RelayDispatcher { this.clients.delete(client.id) } this.notifyClientDetached(client.id) + if (remove) { + // Only for a client that is gone for good: an invalidated primary is revived by setWrite, and a + // frame stranded by its retired sink must stay armed to retry. After the detach fan-out, so a + // listener that unsubscribes on detach is not left holding a stale slot. + this.clientCapacityListeners.delete(client.id) + } this.notifyLegacyCapacity(true) if (!/^Relay (?:primary client invalidated|client detached)$/.test(error.message)) { process.stderr.write(`[relay] Client write closed: ${error.message}\n`) diff --git a/src/relay/relay-filesystem-watch-registry.test.ts b/src/relay/relay-filesystem-watch-registry.test.ts index 0c6109a55ea..de924230f0f 100644 --- a/src/relay/relay-filesystem-watch-registry.test.ts +++ b/src/relay/relay-filesystem-watch-registry.test.ts @@ -96,7 +96,10 @@ function createDispatcher() { onClientDetached: vi.fn((listener: (clientId: number) => void) => { detached.add(listener) return () => detached.delete(listener) - }) + }), + // Detach here always retires the id, so the emitter's marker cleanup runs. + isClientAttached: vi.fn(() => false), + notificationFrameBytes: vi.fn(() => 64) } } diff --git a/src/relay/relay-watcher-event-emitter.test.ts b/src/relay/relay-watcher-event-emitter.test.ts index 35b448e8588..acf734acd71 100644 --- a/src/relay/relay-watcher-event-emitter.test.ts +++ b/src/relay/relay-watcher-event-emitter.test.ts @@ -4,6 +4,7 @@ import { describe, expect, it, vi } from 'vitest' import type { WatcherProcessEvent } from '../main/ipc/parcel-watcher-process-protocol' import { RelayDispatcher } from './dispatcher' import { + DISPATCHER_CONTROL_QUEUE_MAX_BYTES, DEFAULT_PRODUCER_QUEUE_MAX_BYTES, DISPATCHER_CONTROL_QUEUE_MAX_FRAMES } from './dispatcher-writer-admission' @@ -78,6 +79,28 @@ function createRecordingSink(highWaterMark?: number): { } } +// A sink that acknowledges writes on demand, so frames can be held in flight one lane at a time. +function createCallbackSink(): { + frames: Buffer[] + settle: (index: number) => void + write: RelayClientWrite + options: RelayClientSinkOptions +} { + const frames: Buffer[] = [] + const settlements: ((result: { ok: true }) => void)[] = [] + const write: RelayClientWrite = (frame, onSettled) => { + frames.push(Buffer.from(frame)) + settlements.push(onSettled) + return true + } + return { + frames, + settle: (index) => settlements[index]?.({ ok: true }), + write, + options: { supportsWriteCallback: true, close: () => {} } + } +} + function watcherBatch(count: number): { type: 'create'; path: string; isDirectory: false }[] { return Array.from({ length: count }, (_unused, index) => ({ type: 'create' as const, @@ -207,22 +230,79 @@ describe('relay watcher overflow suppression key', () => { } }) - it('releases the outstanding key when notification admission rejects without settlement', () => { - const sink = createRecordingSink(65536) - const dispatcher = new RelayDispatcher(sink.write, sink.options) - const notify = vi.spyOn(dispatcher, 'tryNotifyClient').mockReturnValue(false) + it('drops the marker instead of retaining it when the client is already gone', () => { + const primary = createRecordingSink(65536) + const secondary = createRecordingSink(65536) + const dispatcher = new RelayDispatcher(primary.write, primary.options) + const secondaryId = dispatcher.attachClient(secondary.write, secondary.options) + dispatcher.detachClient(secondaryId) + // The race the emitter has to survive: the id list was sampled before the client went away. + const activeIds = vi.spyOn(dispatcher, 'activeClientIds').mockReturnValue([secondaryId]) + const notify = vi.spyOn(dispatcher, 'tryNotifyClient') try { emitRelayWatcherOverflow(dispatcher, '/workspace', false) + // A second attempt proves nothing was retained: a removed client has no tree left to resync. emitRelayWatcherOverflow(dispatcher, '/workspace', false) expect(notify).toHaveBeenCalledTimes(2) + expect(watcherFrames(secondary.frames)).toHaveLength(0) } finally { notify.mockRestore() + activeIds.mockRestore() dispatcher.dispose() } }) - it('fails soft and releases the outstanding key when the control queue is full', () => { + it('republishes an admitted marker the replaced sink never wrote', () => { + const stalled = createRecordingSink(65536) + const replacement = createRecordingSink(65536) + stalled.blockNextWrite() + const dispatcher = new RelayDispatcher(stalled.write, stalled.options) + + try { + const clientId = dispatcher.activeClientIds()[0] + // Saturates the sink without filling the control queue, so the marker is admitted and then parked. + dispatcher.notifyClient(clientId, 'control.warmup') + emitRelayWatcherOverflow(dispatcher, '/workspace', false) + expect(watcherFrames(stalled.frames)).toHaveLength(0) + + // setWrite fails every queued frame; an admitted-but-unwritten marker is as lost as a rejected one. + dispatcher.setWrite(replacement.write, replacement.options) + + expect(watcherFrames(replacement.frames).flatMap(frameEvents)).toEqual([ + { kind: 'overflow', absolutePath: '/workspace' } + ]) + } finally { + dispatcher.dispose() + } + }) + + it('republishes an admitted marker when invalidateClient precedes the sink replacement', () => { + const stalled = createRecordingSink(65536) + const replacement = createRecordingSink(65536) + stalled.blockNextWrite() + const dispatcher = new RelayDispatcher(stalled.write, stalled.options) + + try { + const clientId = dispatcher.activeClientIds()[0] + dispatcher.notifyClient(clientId, 'control.warmup') + emitRelayWatcherOverflow(dispatcher, '/workspace', false) + expect(watcherFrames(stalled.frames)).toHaveLength(0) + + // A dropped SSH channel invalidates the primary — a detach that keeps the id, so the marker the + // failed settlement retains must survive it and ride out on the reconnected sink. + dispatcher.invalidateClient() + dispatcher.setWrite(replacement.write, replacement.options) + + expect(watcherFrames(replacement.frames).flatMap(frameEvents)).toEqual([ + { kind: 'overflow', absolutePath: '/workspace' } + ]) + } finally { + dispatcher.dispose() + } + }) + + it('retries a marker rejected by a full control queue once capacity returns', () => { const sink = createRecordingSink(65536) sink.blockNextWrite() const dispatcher = new RelayDispatcher(sink.write, sink.options) @@ -237,8 +317,8 @@ describe('relay watcher overflow suppression key', () => { expect(watcherFrames(sink.frames)).toHaveLength(0) expect(sink.closes()).toBe(0) + // No second emit: the watcher has no further trigger, so the retry must come from the retained marker. sink.drainWaiters.shift()?.() - emitRelayWatcherOverflow(dispatcher, '/workspace', false) expect(watcherFrames(sink.frames).flatMap(frameEvents)).toEqual([ { kind: 'overflow', absolutePath: '/workspace' } ]) @@ -247,6 +327,175 @@ describe('relay watcher overflow suppression key', () => { dispatcher.dispose() } }) + + it('leaves a retained marker alone until the control lane actually frees a slot', () => { + const sink = createCallbackSink() + const dispatcher = new RelayDispatcher(sink.write, sink.options) + + try { + const clientId = dispatcher.activeClientIds()[0] + dispatcher.tryNotifyPtyData({ paneId: 'pane', data: 'x' }) + for (let index = 0; index < DISPATCHER_CONTROL_QUEUE_MAX_FRAMES; index += 1) { + dispatcher.notifyClient(clientId, `control.${index}`) + } + emitRelayWatcherOverflow(dispatcher, '/workspace', false) + expect(watcherFrames(sink.frames)).toHaveLength(0) + + const notify = vi.spyOn(dispatcher, 'tryNotifyClient') + // Capacity fires for every lane: a producer settlement frees no control slot, so re-encoding + // the marker here could only be rejected again. + sink.settle(sink.frames.findIndex((frame) => frameMethod(frame) === 'pty.data')) + expect(notify).not.toHaveBeenCalled() + + // A control settlement does free one, and the retained marker goes out on it. + sink.settle(sink.frames.findIndex((frame) => frameMethod(frame) === 'control.0')) + expect(watcherFrames(sink.frames).flatMap(frameEvents)).toEqual([ + { kind: 'overflow', absolutePath: '/workspace' } + ]) + notify.mockRestore() + } finally { + dispatcher.dispose() + } + }) + + it('waits until the control byte budget can admit the retained marker', () => { + const sink = createCallbackSink() + const dispatcher = new RelayDispatcher(sink.write, sink.options) + + try { + const clientId = dispatcher.activeClientIds()[0] + const markerParams = { + events: [{ kind: 'overflow', absolutePath: '/workspace' }] + } + const markerBytes = dispatcher.notificationFrameBytes('fs.changed', markerParams) + const emptyFillBytes = dispatcher.notificationFrameBytes('control.byte-fill', { data: '' }) + const fillDataBytes = DISPATCHER_CONTROL_QUEUE_MAX_BYTES - markerBytes + 1 - emptyFillBytes + + dispatcher.tryNotifyPtyData({ paneId: 'pane', data: 'x' }) + dispatcher.notifyClient(clientId, 'control.byte-fill', { data: 'x'.repeat(fillDataBytes) }) + emitRelayWatcherOverflow(dispatcher, '/workspace', false) + expect(watcherFrames(sink.frames)).toHaveLength(0) + + const notify = vi.spyOn(dispatcher, 'tryNotifyClient') + sink.settle(sink.frames.findIndex((frame) => frameMethod(frame) === 'pty.data')) + expect(notify).not.toHaveBeenCalled() + + sink.settle(sink.frames.findIndex((frame) => frameMethod(frame) === 'control.byte-fill')) + expect(watcherFrames(sink.frames).flatMap(frameEvents)).toEqual([ + { kind: 'overflow', absolutePath: '/workspace' } + ]) + notify.mockRestore() + } finally { + dispatcher.dispose() + } + }) + + it('keeps a retained marker per root and sends each exactly once', () => { + const sink = createRecordingSink(65536) + sink.blockNextWrite() + const dispatcher = new RelayDispatcher(sink.write, sink.options) + + try { + const clientId = dispatcher.activeClientIds()[0] + for (let index = 0; index < DISPATCHER_CONTROL_QUEUE_MAX_FRAMES; index += 1) { + dispatcher.notifyClient(clientId, `control.${index}`) + } + + for (let index = 0; index < 3; index += 1) { + emitRelayWatcherOverflow(dispatcher, '/workspace', false) + emitRelayWatcherOverflow(dispatcher, '/other', false) + } + sink.drainWaiters.shift()?.() + + // A retained root must neither duplicate nor suppress a peer root's resync. + expect(watcherFrames(sink.frames).flatMap(frameEvents)).toEqual([ + { kind: 'overflow', absolutePath: '/workspace' }, + { kind: 'overflow', absolutePath: '/other' } + ]) + expect(sink.closes()).toBe(0) + } finally { + dispatcher.dispose() + } + }) + + it('republishes a retained marker against a replaced primary sink', () => { + const stalled = createRecordingSink(65536) + const replacement = createRecordingSink(65536) + stalled.blockNextWrite() + const dispatcher = new RelayDispatcher(stalled.write, stalled.options) + + try { + const clientId = dispatcher.activeClientIds()[0] + for (let index = 0; index < DISPATCHER_CONTROL_QUEUE_MAX_FRAMES; index += 1) { + dispatcher.notifyClient(clientId, `control.${index}`) + } + emitRelayWatcherOverflow(dispatcher, '/workspace', false) + expect(watcherFrames(stalled.frames)).toHaveLength(0) + + // setWrite replaces the sink without a detach, so the marker must follow the client, not the writer. + dispatcher.setWrite(replacement.write, replacement.options) + + expect(watcherFrames(replacement.frames).flatMap(frameEvents)).toEqual([ + { kind: 'overflow', absolutePath: '/workspace' } + ]) + } finally { + dispatcher.dispose() + } + }) + + it('republishes a retained marker when invalidateClient precedes the sink replacement', () => { + const stalled = createRecordingSink(65536) + const replacement = createRecordingSink(65536) + stalled.blockNextWrite() + const dispatcher = new RelayDispatcher(stalled.write, stalled.options) + + try { + const clientId = dispatcher.activeClientIds()[0] + for (let index = 0; index < DISPATCHER_CONTROL_QUEUE_MAX_FRAMES; index += 1) { + dispatcher.notifyClient(clientId, `control.${index}`) + } + emitRelayWatcherOverflow(dispatcher, '/workspace', false) + expect(watcherFrames(stalled.frames)).toHaveLength(0) + + // The SSH reconnect order: invalidateClient detaches the primary without retiring the id, so the + // pending marker must survive that detach and ride out on the replacement sink. + dispatcher.invalidateClient() + dispatcher.setWrite(replacement.write, replacement.options) + + expect(watcherFrames(replacement.frames).flatMap(frameEvents)).toEqual([ + { kind: 'overflow', absolutePath: '/workspace' } + ]) + } finally { + dispatcher.dispose() + } + }) + + it('forgets a retained marker when the client detaches', () => { + const primary = createRecordingSink(65536) + const secondary = createRecordingSink(65536) + secondary.blockNextWrite() + const dispatcher = new RelayDispatcher(primary.write, primary.options) + + try { + const secondaryId = dispatcher.attachClient(secondary.write, secondary.options) + for (let index = 0; index < DISPATCHER_CONTROL_QUEUE_MAX_FRAMES; index += 1) { + dispatcher.notifyClient(secondaryId, `control.${index}`) + } + emitRelayWatcherOverflow(dispatcher, '/workspace', false) + expect(watcherFrames(secondary.frames)).toHaveLength(0) + + dispatcher.detachClient(secondaryId) + secondary.drainWaiters.shift()?.() + + expect(watcherFrames(secondary.frames)).toHaveLength(0) + // The primary's own marker went out on its first attempt and stays unaffected. + expect(watcherFrames(primary.frames).flatMap(frameEvents)).toEqual([ + { kind: 'overflow', absolutePath: '/workspace' } + ]) + } finally { + dispatcher.dispose() + } + }) }) describe('relay watcher batch chunking', () => { diff --git a/src/relay/relay-watcher-event-emitter.ts b/src/relay/relay-watcher-event-emitter.ts index 54aa6af8ca2..6bfef6a97f0 100644 --- a/src/relay/relay-watcher-event-emitter.ts +++ b/src/relay/relay-watcher-event-emitter.ts @@ -13,8 +13,15 @@ type WatcherBatchSizing = { batchBytes: number } -// Why: one outstanding marker per (client, root) keeps sustained backpressure bounded. -const outstandingOverflowMarkers = new WeakMap>() +type OverflowMarkerState = { + // Why: one outstanding marker per (client, root) keeps sustained backpressure bounded. + inFlight: Set + // Rejected markers, retained per client so they can be republished when the control lane frees up. + pending: Map> + capacityUnsubscribes: Map void> +} + +const overflowMarkerStates = new WeakMap() export function emitRelayWatcherEvents( dispatcher: RelayDispatcher, @@ -157,6 +164,40 @@ function publishWatcherBatchToClient( } } +function overflowMarkerState(dispatcher: RelayDispatcher): OverflowMarkerState { + const existing = overflowMarkerStates.get(dispatcher) + if (existing) { + return existing + } + const state: OverflowMarkerState = { + inFlight: new Set(), + pending: new Map(), + capacityUnsubscribes: new Map() + } + overflowMarkerStates.set(dispatcher, state) + // In-flight keys need no sweep here: closing a client settles every queued and written frame first. + dispatcher.onClientDetached((clientId) => { + // Not every detach retires the id: invalidateClient() detaches the primary without removing it and + // setWrite() revives it, so dropping the markers here would desync the tree the reconnect restores. + if (dispatcher.isClientAttached(clientId)) { + return + } + forgetClientMarkers(state, clientId) + }) + return state +} + +function forgetClientMarkers(state: OverflowMarkerState, clientId: number): void { + // Unsubscribe first so no re-entrant flush can observe a half-cleared client. + state.capacityUnsubscribes.get(clientId)?.() + state.capacityUnsubscribes.delete(clientId) + state.pending.delete(clientId) +} + +function overflowMarkerParams(rootPath: string): Record { + return { events: [{ kind: 'overflow', absolutePath: rootPath }] } +} + // Why: the control lane — on the producer lane the marker would hit the same full queue that just // rejected the batch and be dropped, silently desyncing the remote file tree. function emitWatcherOverflowToClient( @@ -164,34 +205,106 @@ function emitWatcherOverflowToClient( clientId: number, rootPath: string ): void { + const state = overflowMarkerState(dispatcher) // Per root, never per client alone: an outstanding marker for one tree must not suppress another's resync. - const key = `${clientId} ${rootPath}` - const outstanding = outstandingOverflowMarkers.get(dispatcher) ?? new Set() - if (outstanding.has(key)) { + if ( + state.inFlight.has(`${clientId} ${rootPath}`) || + state.pending.get(clientId)?.has(rootPath) === true + ) { return } - outstanding.add(key) - outstandingOverflowMarkers.set(dispatcher, outstanding) - const release = (): void => { - outstanding.delete(key) - if (outstanding.size === 0) { - outstandingOverflowMarkers.delete(dispatcher) - } - } + publishOverflowMarker(dispatcher, state, clientId, rootPath) +} + +function publishOverflowMarker( + dispatcher: RelayDispatcher, + state: OverflowMarkerState, + clientId: number, + rootPath: string, + estimatedBytes?: number +): void { + const key = `${clientId} ${rootPath}` + const params = overflowMarkerParams(rootPath) + const frameBytes = estimatedBytes ?? dispatcher.notificationFrameBytes('fs.changed', params) + state.inFlight.add(key) + let settled = false const accepted = dispatcher.tryNotifyClient( clientId, 'fs.changed', - { events: [{ kind: 'overflow', absolutePath: rootPath }] }, - () => { + params, + (result) => { // Settles on write, drop, or client close, so the slot can never leak. - release() + settled = true + state.inFlight.delete(key) + if (result.ok) { + return + } + // A frame the sink never wrote leaves the tree just as desynced as a rejected one — setWrite + // fails every queued and in-flight frame this way. Retain unconditionally: a real detach clears + // it through onClientDetached, which fires after this settlement. + retainOverflowMarker(dispatcher, state, clientId, rootPath, frameBytes) }, - { controlOverflow: 'reject' } + { controlOverflow: 'reject', estimatedBytes: frameBytes } ) - // Admission rejection has no settlement callback, so release the slot directly. - if (!accepted) { - release() + if (accepted || settled) { + return } + // Admission rejection has no settlement callback: retain the marker instead of desyncing the tree. + state.inFlight.delete(key) + retainOverflowMarker(dispatcher, state, clientId, rootPath, frameBytes) +} + +function retainOverflowMarker( + dispatcher: RelayDispatcher, + state: OverflowMarkerState, + clientId: number, + rootPath: string, + estimatedBytes: number +): void { + if (!state.capacityUnsubscribes.has(clientId)) { + const unsubscribe = dispatcher.onClientCapacity(clientId, () => + flushPendingOverflowMarkers(dispatcher, state, clientId) + ) + if (!unsubscribe) { + // The client went away between admission and arming, so there is nothing left to resync. + return + } + state.capacityUnsubscribes.set(clientId, unsubscribe) + } + const roots = state.pending.get(clientId) + if (roots) { + roots.set(rootPath, estimatedBytes) + return + } + state.pending.set(clientId, new Map([[rootPath, estimatedBytes]])) +} + +function flushPendingOverflowMarkers( + dispatcher: RelayDispatcher, + state: OverflowMarkerState, + clientId: number +): void { + const roots = state.pending.get(clientId) + if (!roots) { + return + } + for (const [rootPath, estimatedBytes] of Array.from(roots)) { + // Capacity fires on every lane; retry only frames the control queue can admit now. + if (!dispatcher.canAdmitControlFrame(clientId, estimatedBytes)) { + continue + } + // Drop before republishing so a synchronous settlement cannot see the marker as still pending — + // and skip roots a re-entrant flush already took, which would otherwise send the marker twice. + if (!roots.delete(rootPath)) { + continue + } + publishOverflowMarker(dispatcher, state, clientId, rootPath, estimatedBytes) + } + // Identity check: a re-entrant flush may have retired this set and armed a fresh one to keep. + if (roots.size > 0 || state.pending.get(clientId) !== roots) { + return + } + forgetClientMarkers(state, clientId) } export function emitRelayWatcherOverflow(