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(