diff --git a/src/main/ssh/ssh-relay-cross-version-isolation.test.ts b/src/main/ssh/ssh-relay-cross-version-isolation.test.ts index cf203aa26c5..001d30a35d4 100644 --- a/src/main/ssh/ssh-relay-cross-version-isolation.test.ts +++ b/src/main/ssh/ssh-relay-cross-version-isolation.test.ts @@ -137,5 +137,21 @@ describe('cross-version isolation', () => { (c) => c.includes('rm -rf') && c.includes('relay-0.1.0+v1hash') ) expect(v1RemoveCmds).toHaveLength(0) + + // (d) blanket isolation: every command that mentions v1hash MUST be a + // GC liveness probe (`ls`, `test -d`, `test -f`, or `for f in .../*.sock`) + // — never a write, mkdir, chmod, touch, rm, node launch, or socket poll. + // This prevents a future refactor that accidentally writes to the v1 dir + // (e.g. shared install-complete, upload over symlink) from passing. + const v1Refs = allCmds.filter((c) => c.includes('relay-0.1.0+v1hash')) + for (const cmd of v1Refs) { + const isReadOnlyProbe = + /^\s*ls\b/.test(cmd) || + /\btest -d\b/.test(cmd) || + /\btest -f\b/.test(cmd) || + /\btest -S\b/.test(cmd) || + /\bfor f in .*\.sock\b/.test(cmd) + expect(isReadOnlyProbe, `unexpected v1 reference: ${cmd}`).toBe(true) + } }) }) diff --git a/src/main/ssh/ssh-relay-deploy-helpers.test.ts b/src/main/ssh/ssh-relay-deploy-helpers.test.ts index fd66ecb1f4f..3c30d23318b 100644 --- a/src/main/ssh/ssh-relay-deploy-helpers.test.ts +++ b/src/main/ssh/ssh-relay-deploy-helpers.test.ts @@ -3,6 +3,10 @@ import { describe, expect, it, vi } from 'vitest' import type { ClientChannel } from 'ssh2' import { execCommand, waitForSentinel } from './ssh-relay-deploy-helpers' import { RELAY_SENTINEL } from './relay-protocol' +import { + RelayVersionMismatchError, + RELAY_EXIT_CODE_VERSION_MISMATCH +} from './ssh-relay-version-mismatch-error' function createMockChannel(): ClientChannel { return Object.assign(new EventEmitter(), { @@ -60,6 +64,51 @@ describe('waitForSentinel', () => { expect(() => channel.emit('error', new Error('remote host rebooted'))).not.toThrow() expect(onClose).toHaveBeenCalledTimes(1) }) + + it('translates a pre-sentinel exit-42 + close into RelayVersionMismatchError', async () => { + const channel = createMockChannel() + const transportPromise = waitForSentinel(channel) + + channel.stderr.emit( + 'data', + Buffer.from( + '[relay-connect] Handshake mismatch: expected=0.1.0+aaa, daemon=0.1.0+bbb; exiting 42\n' + ) + ) + channel.emit('exit', RELAY_EXIT_CODE_VERSION_MISMATCH) + channel.emit('close') + + await expect(transportPromise).rejects.toBeInstanceOf(RelayVersionMismatchError) + await transportPromise.catch((err: RelayVersionMismatchError) => { + expect(err.expected).toBe('0.1.0+aaa') + expect(err.got).toBe('0.1.0+bbb') + }) + }) + + it('translates a pre-sentinel exit-42 even when the version detail is missing', async () => { + const channel = createMockChannel() + const transportPromise = waitForSentinel(channel) + + channel.stderr.emit('data', Buffer.from('boom\n')) + channel.emit('exit', RELAY_EXIT_CODE_VERSION_MISMATCH) + channel.emit('close') + + await expect(transportPromise).rejects.toBeInstanceOf(RelayVersionMismatchError) + }) + + it('rejects with a generic error (not RelayVersionMismatchError) on a non-42 exit code', async () => { + const channel = createMockChannel() + const transportPromise = waitForSentinel(channel) + + channel.stderr.emit('data', Buffer.from('node: bad bytecode\n')) + channel.emit('exit', 1) + channel.emit('close') + + await expect(transportPromise).rejects.toThrow(/Relay process exited before ready/) + await transportPromise.catch((err: unknown) => { + expect(err).not.toBeInstanceOf(RelayVersionMismatchError) + }) + }) }) describe('execCommand', () => { diff --git a/src/main/ssh/ssh-relay-deploy-helpers.ts b/src/main/ssh/ssh-relay-deploy-helpers.ts index f1cbbcbf730..4b3146a0cd1 100644 --- a/src/main/ssh/ssh-relay-deploy-helpers.ts +++ b/src/main/ssh/ssh-relay-deploy-helpers.ts @@ -24,18 +24,41 @@ export function waitForSentinel(channel: ClientChannel): Promise | null = null + const timeout = setTimeout(() => { - if (!settled) { - settled = true - channel.close() - reject( - new Error( - `Relay failed to start within ${RELAY_SENTINEL_TIMEOUT_MS / 1000}s.${stderrOutput ? ` stderr: ${stderrOutput.trim()}` : ''}` + timeoutFired = true + channel.close() + timeoutGraceTimer = setTimeout(() => { + if (!settled) { + settled = true + reject( + new Error( + `Relay failed to start within ${RELAY_SENTINEL_TIMEOUT_MS / 1000}s.${stderrOutput ? ` stderr: ${stderrOutput.trim()}` : ''}` + ) ) - ) - } + } + }, TIMEOUT_GRACE_MS) }, RELAY_SENTINEL_TIMEOUT_MS) + const cancelTimers = (): void => { + clearTimeout(timeout) + if (timeoutGraceTimer) { + clearTimeout(timeoutGraceTimer) + timeoutGraceTimer = null + } + } + channel.on('exit', (code: number | null) => { if (typeof code === 'number') { lastExitCode = code @@ -64,7 +87,7 @@ export function waitForSentinel(channel: ClientChannel): Promise { - clearTimeout(timeout) + cancelTimers() if (!sentinelReceived) { if (!settled) { settled = true @@ -83,7 +106,7 @@ export function waitForSentinel(channel: ClientChannel): Promise { if (!sentinelReceived) { - clearTimeout(timeout) + cancelTimers() if (!settled) { settled = true // Why: a wire-handshake mismatch on the daemon side closes the @@ -91,15 +114,20 @@ export function waitForSentinel(channel: ClientChannel): Promise ({ + deployAndLaunchRelay: vi.fn() +})) + +vi.mock('./ssh-channel-multiplexer', () => { + return { + SshChannelMultiplexer: class MockSshChannelMultiplexer { + notify = vi.fn() + request = vi.fn().mockResolvedValue([]) + onNotification = vi.fn().mockReturnValue(() => {}) + onDispose = vi.fn().mockReturnValue(() => {}) + dispose = vi.fn() + isDisposed = vi.fn().mockReturnValue(false) + } + } +}) + +vi.mock('../providers/ssh-pty-provider', () => ({ + SshPtyProvider: class MockSshPtyProvider { + onData = vi.fn().mockReturnValue(() => {}) + onReplay = vi.fn().mockReturnValue(() => {}) + onExit = vi.fn().mockReturnValue(() => {}) + attach = vi.fn().mockResolvedValue(undefined) + dispose = vi.fn() + } +})) + +vi.mock('../providers/ssh-filesystem-provider', () => ({ + SshFilesystemProvider: class MockSshFilesystemProvider { + dispose = vi.fn() + } +})) + +vi.mock('../providers/ssh-git-provider', () => ({ + SshGitProvider: class MockSshGitProvider {} +})) + +vi.mock('../ipc/pty', () => ({ + registerSshPtyProvider: vi.fn(), + unregisterSshPtyProvider: vi.fn(), + getSshPtyProvider: vi.fn().mockReturnValue({ + dispose: vi.fn(), + attach: vi.fn().mockResolvedValue(undefined) + }), + getPtyIdsForConnection: vi.fn().mockReturnValue([]), + clearPtyOwnershipForConnection: vi.fn(), + clearProviderPtyState: vi.fn(), + deletePtyOwnership: vi.fn() +})) + +vi.mock('../providers/ssh-filesystem-dispatch', () => ({ + registerSshFilesystemProvider: vi.fn(), + unregisterSshFilesystemProvider: vi.fn(), + getSshFilesystemProvider: vi.fn().mockReturnValue({ dispose: vi.fn() }) +})) + +vi.mock('../providers/ssh-git-dispatch', () => ({ + registerSshGitProvider: vi.fn(), + unregisterSshGitProvider: vi.fn() +})) + +const { deployAndLaunchRelay } = await import('./ssh-relay-deploy') + +function createMockDeps(): { + mockConn: SshConnection + mockStore: Store + mockPortForward: SshPortForwardManager + getMainWindow: () => BrowserWindow | null +} { + const mockConn = {} as SshConnection + const mockStore = { + getRepos: vi.fn().mockReturnValue([]) + } as unknown as Store + const mockPortForward = { + removeAllForwards: vi.fn() + } as unknown as SshPortForwardManager + const mockWindow = { + isDestroyed: (): boolean => false, + webContents: { send: vi.fn() } + } as unknown as BrowserWindow + const getMainWindow = vi.fn().mockReturnValue(mockWindow) as unknown as () => BrowserWindow | null + return { mockConn, mockStore, mockPortForward, getMainWindow } +} + +function mockDeploySuccess(): void { + const mockTransport = { + write: vi.fn(), + onData: vi.fn(), + onClose: vi.fn() + } + vi.mocked(deployAndLaunchRelay).mockResolvedValue({ + transport: mockTransport, + platform: 'linux-x64' + }) +} + +describe('SshRelaySession terminal relay error (RelayVersionMismatchError)', () => { + beforeEach(() => { + vi.clearAllMocks() + mockDeploySuccess() + }) + + it('fires onTerminalRelayError on initial establish() and rethrows', async () => { + const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps() + const session = new SshRelaySession('target-1', getMainWindow, mockStore, mockPortForward) + const onTerminal = vi.fn() + const onLost = vi.fn() + session.setOnTerminalRelayError(onTerminal) + session.setOnRelayLost(onLost) + + const mismatchErr = new RelayVersionMismatchError( + '0.1.0+aaa', + '0.1.0+bbb', + '[relay-connect] Handshake mismatch...' + ) + vi.mocked(deployAndLaunchRelay).mockRejectedValueOnce(mismatchErr) + + await expect(session.establish(mockConn)).rejects.toBe(mismatchErr) + expect(onTerminal).toHaveBeenCalledTimes(1) + expect(onTerminal).toHaveBeenCalledWith('target-1', mismatchErr) + expect(onLost).not.toHaveBeenCalled() + expect(session.getState()).toBe('idle') + }) + + it('does NOT fire onTerminalRelayError on a generic establish failure', async () => { + const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps() + const session = new SshRelaySession('target-1', getMainWindow, mockStore, mockPortForward) + const onTerminal = vi.fn() + session.setOnTerminalRelayError(onTerminal) + + vi.mocked(deployAndLaunchRelay).mockRejectedValueOnce(new Error('boom')) + + await expect(session.establish(mockConn)).rejects.toThrow('boom') + expect(onTerminal).not.toHaveBeenCalled() + expect(session.getState()).toBe('idle') + }) + + it('fires onTerminalRelayError on reconnect() when deploy throws RelayVersionMismatchError', async () => { + const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps() + const session = new SshRelaySession('target-1', getMainWindow, mockStore, mockPortForward) + const onTerminal = vi.fn() + session.setOnTerminalRelayError(onTerminal) + + await session.establish(mockConn) + expect(session.getState()).toBe('ready') + + const mismatchErr = new RelayVersionMismatchError('0.1.0+old', '0.1.0+new', '') + vi.mocked(deployAndLaunchRelay).mockRejectedValueOnce(mismatchErr) + + await session.reconnect(mockConn) + expect(onTerminal).toHaveBeenCalledTimes(1) + expect(onTerminal).toHaveBeenCalledWith('target-1', mismatchErr) + }) +}) diff --git a/src/main/ssh/ssh-relay-session.ts b/src/main/ssh/ssh-relay-session.ts index 95df7d86fbb..4401053f777 100644 --- a/src/main/ssh/ssh-relay-session.ts +++ b/src/main/ssh/ssh-relay-session.ts @@ -173,6 +173,20 @@ export class SshRelaySession { this.teardownProviders('shutdown') this._state = 'idle' } + // Why: a wire-handshake mismatch on the FIRST connect is also terminal + // — the deployed relay binary on disk does not match a still-running + // daemon (typically because a legacy daemon from before the + // versioned-dir change is still alive). Notify the terminal-error + // callback so ssh.ts surfaces an actionable message and the caller's + // catch path doesn't conflate this with a transient deploy failure. + // We still rethrow so doConnect's existing failure path runs (clean up + // the SSH connection); ssh.ts's handler is idempotent. + if (isRelayVersionMismatchError(err)) { + console.warn( + `[ssh-relay-session] Terminal relay version mismatch on initial connect for ${this.targetId}: ${err.message}` + ) + this._onTerminalRelayError?.(this.targetId, err) + } throw err } } diff --git a/src/main/ssh/ssh-relay-versioned-install.test.ts b/src/main/ssh/ssh-relay-versioned-install.test.ts index a5b40dc768a..a7bae19e9cf 100644 --- a/src/main/ssh/ssh-relay-versioned-install.test.ts +++ b/src/main/ssh/ssh-relay-versioned-install.test.ts @@ -104,6 +104,93 @@ describe('acquireInstallLock', () => { expect(mockExec).toHaveBeenCalledTimes(2) }) + it('polls until the lock becomes available (concurrent installer wins, then we acquire)', async () => { + vi.useFakeTimers() + try { + // Sequence: + // 1. mkdir -p (parent dir prep) + // 2. mkdir lockDir → BUSY (someone else holds it) + // 3. mkdir lockDir → BUSY again + // 4. mkdir lockDir → OK (concurrent installer released) + mockExec + .mockResolvedValueOnce('') + .mockResolvedValueOnce('BUSY') + .mockResolvedValueOnce('BUSY') + .mockResolvedValueOnce('OK') + + const promise = acquireInstallLock(conn, '/r') + // Drive the polling loop: each iteration awaits a 1s timer. + for (let i = 0; i < 5; i++) { + await vi.advanceTimersByTimeAsync(1_000) + } + await promise + const cmds = mockExec.mock.calls.map(([, c]) => c) + const mkdirAttempts = cmds.filter((c) => c.includes('mkdir') && c.includes('.install-lock')) + expect(mkdirAttempts.length).toBeGreaterThanOrEqual(3) + } finally { + vi.useRealTimers() + } + }) + + it('steals a stale lock and retries with a reset timeout window', async () => { + vi.useFakeTimers({ now: 1_700_000_000_000 }) + try { + let mkdirCalls = 0 + mockExec.mockImplementation(async (_conn: unknown, cmd: string) => { + if (cmd.startsWith('mkdir -p')) { + return '' + } + if (cmd.includes('mkdir') && cmd.includes('.install-lock')) { + mkdirCalls++ + return mkdirCalls > 200 ? 'OK' : 'BUSY' + } + if (cmd.includes('stat')) { + return `${Math.floor((Date.now() - 10 * 60 * 1000) / 1000)}\n` + } + if (cmd.startsWith('rm -rf')) { + mkdirCalls = 1000 + return '' + } + return '' + }) + + const promise = acquireInstallLock(conn, '/r') + // Drive through the full timeout (120s) so the stale-recovery branch + // fires, then drive a few more seconds for the post-recovery retry. + await vi.advanceTimersByTimeAsync(125_000) + await promise + + const cmds = mockExec.mock.calls.map(([, c]) => c) + expect(cmds.some((c) => c.includes('rm -rf') && c.includes('.install-lock'))).toBe(true) + } finally { + vi.useRealTimers() + } + }) + + it('throws if the timeout elapses and the lock is fresh', async () => { + vi.useFakeTimers({ now: 1_700_000_000_000 }) + try { + mockExec.mockImplementation(async (_conn: unknown, cmd: string) => { + if (cmd.startsWith('mkdir -p')) { + return '' + } + if (cmd.includes('mkdir') && cmd.includes('.install-lock')) { + return 'BUSY' + } + if (cmd.includes('stat')) { + return `${Math.floor(Date.now() / 1000)}\n` + } + return '' + }) + + const rejection = expect(acquireInstallLock(conn, '/r')).rejects.toThrow(/not yet stale/i) + await vi.advanceTimersByTimeAsync(125_000) + await rejection + } finally { + vi.useRealTimers() + } + }) + it('finalizeInstall writes .install-complete then removes the lock', async () => { mockExec.mockResolvedValueOnce('').mockResolvedValueOnce('') await finalizeInstall(conn, '/r') @@ -156,9 +243,47 @@ describe('gcOldRelayVersions', () => { expect(cmds.some((c) => c.includes('rm -rf'))).toBe(false) }) - it('skips siblings whose .install-lock is held', async () => { + it('skips siblings whose .install-lock is held and fresh', async () => { mockExec.mockResolvedValueOnce('relay-0.1.0+aaa\n') mockExec.mockResolvedValueOnce('LOCKED') + // isLockStale: mtime ~now → not stale. + mockExec.mockResolvedValueOnce(`${Math.floor(Date.now() / 1000)}\n`) + await gcOldRelayVersions(conn, '/home/u', '/home/u/.orca-remote/relay-0.1.0+bbb') + const cmds = mockExec.mock.calls.map(([, c]) => c) + expect(cmds.some((c) => c.includes('rm -rf'))).toBe(false) + }) + + it('removes a sibling with a stale lock + .install-complete (rm-lock failed mid-finalize)', async () => { + mockExec.mockResolvedValueOnce('relay-0.1.0+aaa\n') + mockExec.mockResolvedValueOnce('LOCKED') + // isLockStale: mtime well in the past → stale. + const staleSec = Math.floor((Date.now() - 10 * 60 * 1000) / 1000) + mockExec.mockResolvedValueOnce(`${staleSec}\n`) + mockExec.mockResolvedValueOnce('COMPLETE') // .install-complete present + mockExec.mockResolvedValueOnce('') // socket probe → no ALIVE + mockExec.mockResolvedValueOnce('') // rm -rf + await gcOldRelayVersions(conn, '/home/u', '/home/u/.orca-remote/relay-0.1.0+bbb') + const lastCmd = mockExec.mock.calls.at(-1)?.[1] ?? '' + expect(lastCmd).toContain('rm -rf') + expect(lastCmd).toContain('relay-0.1.0+aaa') + }) + + it('GCs a legacy relay-v0.1.0 dir whose daemon is dead (no .install-complete required)', async () => { + mockExec.mockResolvedValueOnce('relay-v0.1.0\n') + mockExec.mockResolvedValueOnce('OPEN') // not locked + mockExec.mockResolvedValueOnce('') // socket probe → no ALIVE (no completeProbe — legacy) + mockExec.mockResolvedValueOnce('') // rm -rf + await gcOldRelayVersions(conn, '/home/u', '/home/u/.orca-remote/relay-0.1.0+bbb') + const cmds = mockExec.mock.calls.map(([, c]) => c) + expect(cmds.some((c) => c.includes('rm -rf') && c.includes('relay-v0.1.0'))).toBe(true) + // critically: no .install-complete probe on legacy dirs + expect(cmds.some((c) => c.includes('.install-complete'))).toBe(false) + }) + + it('keeps a legacy relay-v0.1.0 dir whose daemon is still serving', async () => { + mockExec.mockResolvedValueOnce('relay-v0.1.0\n') + mockExec.mockResolvedValueOnce('OPEN') + mockExec.mockResolvedValueOnce('ALIVE') // socket alive → keep await gcOldRelayVersions(conn, '/home/u', '/home/u/.orca-remote/relay-0.1.0+bbb') const cmds = mockExec.mock.calls.map(([, c]) => c) expect(cmds.some((c) => c.includes('rm -rf'))).toBe(false) diff --git a/src/main/ssh/ssh-relay-versioned-install.ts b/src/main/ssh/ssh-relay-versioned-install.ts index 30bced1e06d..07aaa28c2d0 100644 --- a/src/main/ssh/ssh-relay-versioned-install.ts +++ b/src/main/ssh/ssh-relay-versioned-install.ts @@ -23,6 +23,13 @@ import { shellEscape } from './ssh-connection-utils' // so the GC eventually drains the old layout once its daemons idle out. const RELAY_VERSION_DIR_REGEX = /^relay-(v?\d+\.\d+\.\d+(\+[0-9a-f]+)?)$/ +// Why: legacy dirs from before `.install-complete` was introduced (i.e. the +// `relay-v0.1.0` shape with no content-hash suffix). They are missing the +// install-complete sentinel by definition and need a separate liveness-only +// GC check so they actually drain after the legacy daemon dies, instead of +// living on remote disks forever. +const LEGACY_RELAY_DIR_REGEX = /^relay-v\d+\.\d+\.\d+$/ + const INSTALL_LOCK_NAME = '.install-lock' const INSTALL_COMPLETE_NAME = '.install-complete' @@ -113,7 +120,7 @@ export async function acquireInstallLock( // safe to run multiple times — it's a no-op if the dir already exists. await execCommand(conn, `mkdir -p ${shellEscape(remoteRelayDir)}`) - const start = Date.now() + let start = Date.now() let recoveredOnce = false while (true) { try { @@ -136,12 +143,15 @@ export async function acquireInstallLock( ) } // Stale-lock recovery: if the lock dir's mtime is older than the stale - // window, the previous installer crashed. Steal it and retry once. + // window, the previous installer crashed. Steal it and retry once, + // resetting the timeout window so a single post-recovery race doesn't + // immediately exhaust the budget. const ageOk = await isLockStale(conn, lockDir) if (ageOk) { console.warn(`[ssh-relay] Stealing stale install lock at ${lockDir}`) await execCommand(conn, `rm -rf ${shellEscape(lockDir)}`).catch(() => {}) recoveredOnce = true + start = Date.now() continue } throw new Error( @@ -239,7 +249,7 @@ export async function gcOldRelayVersions( for (const name of candidates) { const dir = `${baseDir}/${name}` try { - const safe = await isCandidateSafeToRemove(conn, dir) + const safe = await isCandidateSafeToRemove(conn, dir, name) if (!safe) { kept.push(name) continue @@ -262,33 +272,46 @@ export async function gcOldRelayVersions( } } -async function isCandidateSafeToRemove(conn: SshConnection, dir: string): Promise { - // Why: skip mid-install or crashed-install partial dirs. A locked dir is - // unsafe because removing it would corrupt a concurrent installer; a dir - // missing .install-complete is either locked (handled above) or a crashed - // partial that the next deploy will recover. +async function isCandidateSafeToRemove( + conn: SshConnection, + dir: string, + name: string +): Promise { + const isLegacy = LEGACY_RELAY_DIR_REGEX.test(name) + const lockProbe = await execCommand( conn, `test -d ${shellEscape(`${dir}/${INSTALL_LOCK_NAME}`)} && echo LOCKED || echo OPEN` ).catch(() => 'OPEN') - if (lockProbe.trim() === 'LOCKED') { - return false + const locked = lockProbe.trim() === 'LOCKED' + + if (locked) { + // Why: a locked dir is normally unsafe to remove — but a STALE lock + // (mtime older than INSTALL_LOCK_STALE_MS) means the previous installer + // crashed and is never coming back. If the dir also has the + // .install-complete sentinel (touch succeeded but the rm-lock at the + // end of finalizeInstall failed), removing the dir is safe — no + // installer is racing us, and the daemon (if any) keeps running off + // its already-loaded code regardless of disk state. + const lockDir = `${dir}/${INSTALL_LOCK_NAME}` + if (!(await isLockStale(conn, lockDir))) { + return false + } + process.stderr.write?.(`[ssh-relay] GC: lock at ${lockDir} is stale; treating as recoverable\n`) } - const completeProbe = await execCommand( - conn, - `test -f ${shellEscape(`${dir}/${INSTALL_COMPLETE_NAME}`)} && echo COMPLETE || echo PARTIAL` - ).catch(() => 'PARTIAL') - if (completeProbe.trim() !== 'COMPLETE') { - // Why: legacy dirs from before `.install-complete` was introduced are - // missing the sentinel. Treat them as recoverable partials (the new - // deploy targeting their version dir would re-run install). The legacy - // dir name (`relay-v0.1.0`) is NOT considered safe to remove here unless - // the legacy daemon has died — and the dead-daemon check below also - // gates on `.install-complete`. Net effect: legacy dirs persist until a - // future migration explicitly drains them. Acceptable: no daemon there - // is reachable by the new client, just disk usage. - return false + // Legacy dirs (relay-v0.1.0) predate .install-complete. Skip the sentinel + // check for them and rely solely on the live-socket probe — that's the + // only signal we have that a legacy daemon is still serving clients. + if (!isLegacy) { + const completeProbe = await execCommand( + conn, + `test -f ${shellEscape(`${dir}/${INSTALL_COMPLETE_NAME}`)} && echo COMPLETE || echo PARTIAL` + ).catch(() => 'PARTIAL') + if (completeProbe.trim() !== 'COMPLETE') { + // Crashed-install partial; leave for the next deploy to recover. + return false + } } const sockAlive = await hasLiveRelaySocket(conn, dir) diff --git a/src/relay/protocol.ts b/src/relay/protocol.ts index a430e43d9d0..e555289bca1 100644 --- a/src/relay/protocol.ts +++ b/src/relay/protocol.ts @@ -153,6 +153,16 @@ export class FrameDecoder { reset(): void { this.buffer = Buffer.alloc(0) } + + // Why: at the handshake → dispatcher transition, the next consumer must + // pick up any bytes that arrived in the same TCP chunk as the handshake + // frame. This returns and clears the decoder's internal residue so the + // caller can hand it to the dispatcher (or stdout pipe) without loss. + drain(): Buffer { + const out = this.buffer + this.buffer = Buffer.alloc(0) + return out + } } export function parseJsonRpcMessage(payload: Buffer): JsonRpcMessage { diff --git a/src/relay/relay-handshake-roundtrip.test.ts b/src/relay/relay-handshake-roundtrip.test.ts new file mode 100644 index 00000000000..28d884940b0 --- /dev/null +++ b/src/relay/relay-handshake-roundtrip.test.ts @@ -0,0 +1,227 @@ +import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest' +import { createServer, connect, type Server, type Socket } from 'net' +import { mkdtempSync, rmSync } from 'fs' +import { tmpdir } from 'os' +import { join } from 'path' + +import { + setupDaemonHandshake, + runConnectHandshake, + EXIT_CODE_VERSION_MISMATCH +} from './relay-handshake' +import { + encodeHandshakeFrame, + encodeJsonRpcFrame, + FrameDecoder, + type DecodedFrame, + MessageType +} from './protocol' + +// Why: --connect normally calls process.exit on mismatch / fatal handshake +// errors. Stub it for tests so the harness sees a thrown sentinel error +// rather than tearing down the test runner. +class ExitCalled extends Error { + code: number + constructor(code: number) { + super(`process.exit(${code})`) + this.code = code + } +} + +describe('handshake round-trip over a real Socket pair', () => { + let server: Server + let sockPath: string + let tmpDir: string + let exitSpy: ReturnType + + let uncaughtHandler: (err: Error) => void + + beforeEach(() => { + tmpDir = mkdtempSync(join(tmpdir(), 'orca-handshake-test-')) + sockPath = join(tmpDir, 'relay.sock') + exitSpy = vi.spyOn(process, 'exit').mockImplementation(((code?: number) => { + throw new ExitCalled(code ?? 0) + }) as never) + // Why: process.exit is called from inside async callbacks + // (process.stderr.write flush callback) which would otherwise surface + // as an uncaughtException after the test resolves and tear down the + // runner. We swallow ExitCalled — exitSpy still records the call so + // assertions hold. + uncaughtHandler = (err: Error): void => { + if (err instanceof ExitCalled) { + return + } + throw err + } + process.on('uncaughtException', uncaughtHandler) + }) + + afterEach(async () => { + process.off('uncaughtException', uncaughtHandler) + exitSpy.mockRestore() + for (const s of liveServerSockets) { + s.destroy() + } + liveServerSockets.length = 0 + if (server) { + await new Promise((r) => server.close(() => r())) + } + rmSync(tmpDir, { recursive: true, force: true }) + }) + + const liveServerSockets: Socket[] = [] + function trackServerSocket(s: Socket): Socket { + liveServerSockets.push(s) + return s + } + + function startDaemon(version: string): Promise<{ + accepted: Promise<{ sock: Socket; leftover: Buffer }> + }> { + return new Promise((resolve) => { + const acceptedDeferred: { + promise: Promise<{ sock: Socket; leftover: Buffer }> + resolve: (v: { sock: Socket; leftover: Buffer }) => void + } = (() => { + let _resolve: (v: { sock: Socket; leftover: Buffer }) => void = () => {} + const promise = new Promise<{ sock: Socket; leftover: Buffer }>((r) => { + _resolve = r + }) + return { promise, resolve: _resolve } + })() + + server = createServer((sock) => { + trackServerSocket(sock) + setupDaemonHandshake(sock, { + launchVersion: version, + onAccepted: (s, leftover) => acceptedDeferred.resolve({ sock: s, leftover }) + }) + }) + server.listen(sockPath, () => resolve({ accepted: acceptedDeferred.promise })) + }) + } + + it('accepts a matching version and delivers no leftover when the bridge sent only the handshake', async () => { + const { accepted } = await startDaemon('0.1.0+match') + + const bridgeSock = connect(sockPath) + await new Promise((r) => bridgeSock.once('connect', () => r())) + + const acceptedCb = vi.fn<(leftover: Buffer) => void>() + runConnectHandshake(bridgeSock, '0.1.0+match', { onAccepted: acceptedCb }) + + const { leftover } = await accepted + expect(leftover.length).toBe(0) + + await vi.waitFor(() => expect(acceptedCb).toHaveBeenCalledTimes(1)) + expect(acceptedCb.mock.calls[0][0].length).toBe(0) + + bridgeSock.destroy() + }) + + it('preserves leftover bytes on the daemon side when an extra frame is coalesced after the handshake', async () => { + // Why: simulate an aggressive client that pipelines a frame immediately + // after the handshake. We bypass runConnectHandshake here and write the + // raw bytes directly so we control the coalescing behaviour. + const { accepted } = await startDaemon('0.1.0+match') + + const bridgeSock = connect(sockPath) + await new Promise((r) => bridgeSock.once('connect', () => r())) + + const handshakeFrame = encodeHandshakeFrame({ + type: 'orca-relay-handshake', + version: '0.1.0+match' + }) + const trailingPayload = encodeJsonRpcFrame({ jsonrpc: '2.0', method: 'noop', params: {} }, 1, 0) + bridgeSock.write(Buffer.concat([handshakeFrame, trailingPayload])) + + const { leftover } = await accepted + + const seen: DecodedFrame[] = [] + const dec = new FrameDecoder((f) => seen.push(f)) + dec.feed(leftover) + expect(seen).toHaveLength(1) + expect(seen[0].type).toBe(MessageType.Regular) + + bridgeSock.destroy() + }) + + it('preserves leftover bytes on the bridge side when the daemon coalesces handshake-ok + a JSON-RPC frame', async () => { + let serverHandshakeSeen = false + server = createServer((sock) => { + trackServerSocket(sock) + const decoder = new FrameDecoder((frame) => { + if (frame.type !== MessageType.Handshake || serverHandshakeSeen) { + return + } + serverHandshakeSeen = true + const ok = encodeHandshakeFrame({ + type: 'orca-relay-handshake-ok', + version: '0.1.0+match' + }) + const trailing = encodeJsonRpcFrame( + { jsonrpc: '2.0', method: 'pty.event', params: { evt: 'data' } }, + 7, + 1 + ) + sock.write(Buffer.concat([ok, trailing])) + }) + sock.on('data', (chunk: Buffer) => decoder.feed(chunk)) + }) + await new Promise((r) => server.listen(sockPath, () => r())) + + const bridgeSock = connect(sockPath) + await new Promise((r) => bridgeSock.once('connect', () => r())) + + const acceptedCb = vi.fn<(leftover: Buffer) => void>() + runConnectHandshake(bridgeSock, '0.1.0+match', { onAccepted: acceptedCb }) + + await vi.waitFor(() => expect(acceptedCb).toHaveBeenCalledTimes(1)) + const leftover = acceptedCb.mock.calls[0][0] + + const seen: DecodedFrame[] = [] + const dec = new FrameDecoder((f) => seen.push(f)) + dec.feed(leftover) + expect(seen).toHaveLength(1) + expect(seen[0].type).toBe(MessageType.Regular) + + bridgeSock.destroy() + }) + + it('exits with EXIT_CODE_VERSION_MISMATCH when the daemon reports a mismatch', async () => { + await startDaemon('0.1.0+server-version') + + const bridgeSock = connect(sockPath) + await new Promise((r) => bridgeSock.once('connect', () => r())) + + const acceptedCb = vi.fn() + runConnectHandshake(bridgeSock, '0.1.0+different', { onAccepted: acceptedCb }) + + await vi.waitFor(() => expect(exitSpy).toHaveBeenCalled()) + expect(exitSpy).toHaveBeenCalledWith(EXIT_CODE_VERSION_MISMATCH) + expect(acceptedCb).not.toHaveBeenCalled() + + bridgeSock.destroy() + }) + + it('does not call onAccepted before any handshake-ok frame arrives', async () => { + // Why: silent server that never replies. acceptedCb must stay + // un-invoked even though the bridge has flushed its handshake frame. + server = createServer((sock) => { + trackServerSocket(sock) + /* swallow */ + }) + await new Promise((r) => server.listen(sockPath, () => r())) + + const bridgeSock = connect(sockPath) + await new Promise((r) => bridgeSock.once('connect', () => r())) + + const acceptedCb = vi.fn() + runConnectHandshake(bridgeSock, '0.1.0+match', { onAccepted: acceptedCb }) + + await new Promise((r) => setTimeout(r, 100)) + expect(acceptedCb).not.toHaveBeenCalled() + + bridgeSock.destroy() + }) +}) diff --git a/src/relay/relay-handshake.ts b/src/relay/relay-handshake.ts index 251fd859d56..653e03ddb92 100644 --- a/src/relay/relay-handshake.ts +++ b/src/relay/relay-handshake.ts @@ -7,7 +7,7 @@ // independently unit-testable. import { dirname, join } from 'path' -import { existsSync, readFileSync } from 'fs' +import { existsSync, readFileSync, realpathSync } from 'fs' import type { Socket } from 'net' import { RELAY_VERSION, @@ -27,13 +27,27 @@ export const EXIT_CODE_VERSION_MISMATCH = 42 // Why: the deploy step writes a content-hashed version marker (e.g. // "0.1.0+0a5fe134d020") into ${remoteDir}/.version next to relay.js. Read it // from the directory the running script lives in (NOT process.cwd()) so test -// spawns from arbitrary working dirs still report a coherent version. Falls -// back to bare RELAY_VERSION if the marker is missing — the wire handshake -// will then refuse a fresh client whose .version differs. +// spawns from arbitrary working dirs still report a coherent version. We +// resolve symlinks via realpathSync so a daemon launched indirectly (e.g. +// `node /tmp/symlink-to-relay.js`) still finds .version next to the real +// script. Falls back to bare RELAY_VERSION only if the file truly cannot be +// read; the wire handshake then refuses a fresh content-hashed client and +// the user gets a clean typed error rather than a silent stale-daemon loop. export function readLaunchVersion(): string { try { const entry = process.argv[1] - const dir = entry ? dirname(entry) : process.cwd() + let dir: string + if (entry) { + let resolved = entry + try { + resolved = realpathSync(entry) + } catch { + /* fall back to the unresolved path */ + } + dir = dirname(resolved) + } else { + dir = process.cwd() + } const versionFile = join(dir, '.version') if (existsSync(versionFile)) { const v = readFileSync(versionFile, 'utf-8').trim() @@ -50,7 +64,12 @@ export function readLaunchVersion(): string { // ── Daemon side ───────────────────────────────────────────────────── export type DaemonHandshakeCallbacks = { - onAccepted: (sock: Socket) => void + // Why: leftover is any bytes the FrameDecoder buffered AFTER the handshake + // frame (e.g. the bridge wrote handshake + a JSON-RPC frame in the same + // TCP send). The caller MUST feed leftover into the dispatcher before + // attaching the new 'data' listener, otherwise those bytes are silently + // lost. + onAccepted: (sock: Socket, leftover: Buffer) => void launchVersion: string } @@ -61,9 +80,19 @@ export type DaemonHandshakeCallbacks = { // the socket so the bridge exits 42 and the client surfaces a typed error // instead of looping over the dispatcher. export function setupDaemonHandshake(sock: Socket, cb: DaemonHandshakeCallbacks): void { - const decoder = new FrameDecoder( + let handshakeResolved = false + const decoder: FrameDecoder = new FrameDecoder( (frame: DecodedFrame) => { - handleDaemonHandshakeFrame(sock, frame, cb) + if (handshakeResolved) { + return + } + const accepted = handleDaemonHandshakeFrame(sock, frame, cb.launchVersion) + if (accepted) { + handshakeResolved = true + const leftover = decoder.drain() + detachHandshakeListener(sock) + cb.onAccepted(sock, leftover) + } }, (err) => { process.stderr.write(`[relay] Handshake decode error: ${err.message}\n`) @@ -90,14 +119,14 @@ export function detachHandshakeListener(sock: Socket): void { function handleDaemonHandshakeFrame( sock: Socket, frame: DecodedFrame, - cb: DaemonHandshakeCallbacks -): void { + launchVersion: string +): boolean { if (frame.type !== MessageType.Handshake) { process.stderr.write( `[relay] Protocol violation pre-handshake: type=${frame.type}; closing socket\n` ) sock.destroy() - return + return false } let msg: ReturnType try { @@ -107,24 +136,24 @@ function handleDaemonHandshakeFrame( `[relay] Could not parse handshake: ${(err as Error).message}; closing socket\n` ) sock.destroy() - return + return false } if (msg.type !== 'orca-relay-handshake') { process.stderr.write( `[relay] Unexpected handshake type from client: ${msg.type}; closing socket\n` ) sock.destroy() - return + return false } - if (msg.version !== cb.launchVersion) { + if (msg.version !== launchVersion) { process.stderr.write( - `[relay] Handshake mismatch: own=${cb.launchVersion}, client=${msg.version}; closing socket\n` + `[relay] Handshake mismatch: own=${launchVersion}, client=${msg.version}; closing socket\n` ) try { sock.write( encodeHandshakeFrame({ type: 'orca-relay-handshake-mismatch', - expected: cb.launchVersion, + expected: launchVersion, got: msg.version }) ) @@ -132,18 +161,22 @@ function handleDaemonHandshakeFrame( /* best-effort — close+exit-42 still wins */ } sock.end() - return + return false } process.stderr.write(`[relay] Handshake OK from version=${msg.version}\n`) - sock.write(encodeHandshakeFrame({ type: 'orca-relay-handshake-ok', version: cb.launchVersion })) - detachHandshakeListener(sock) - cb.onAccepted(sock) + sock.write(encodeHandshakeFrame({ type: 'orca-relay-handshake-ok', version: launchVersion })) + return true } // ── --connect side ────────────────────────────────────────────────── export type ConnectHandshakeCallbacks = { - onAccepted: () => void + // Why: leftover is any bytes the FrameDecoder buffered AFTER the + // handshake-ok frame. The caller MUST forward leftover to process.stdout + // (the SSH stdout pipe) before attaching the raw bridge, otherwise daemon + // bytes coalesced into the same TCP send as handshake-ok are silently + // dropped. + onAccepted: (leftover: Buffer) => void } // Why: the wire-level version handshake from the bridge side. Before we attach @@ -161,7 +194,7 @@ export function runConnectHandshake( ): void { let handshakeDone = false - const decoder = new FrameDecoder( + const decoder: FrameDecoder = new FrameDecoder( (frame: DecodedFrame) => { if (handshakeDone) { return @@ -186,16 +219,24 @@ export function runConnectHandshake( if (msg.type === 'orca-relay-handshake-ok') { process.stderr.write(`[relay-connect] Handshake OK at version=${msg.version}\n`) handshakeDone = true + const leftover = decoder.drain() sock.removeAllListeners('data') - cb.onAccepted() + cb.onAccepted(leftover) return } if (msg.type === 'orca-relay-handshake-mismatch') { + // Why: explicit stderr flush + exit so the diagnostic line is + // delivered to the client BEFORE the process exits. Without this, + // process.stderr writes can be buffered/async on pipe transports + // and parseHandshakeMismatchStderr loses the version detail. process.stderr.write( - `[relay-connect] Handshake mismatch: expected=${msg.expected}, daemon=${msg.got}; exiting ${EXIT_CODE_VERSION_MISMATCH}\n` + `[relay-connect] Handshake mismatch: expected=${msg.expected}, daemon=${msg.got}; exiting ${EXIT_CODE_VERSION_MISMATCH}\n`, + () => { + sock.destroy() + process.exit(EXIT_CODE_VERSION_MISMATCH) + } ) - sock.destroy() - process.exit(EXIT_CODE_VERSION_MISMATCH) + return } process.stderr.write(`[relay-connect] Unexpected handshake type: ${msg.type}\n`) sock.destroy() diff --git a/src/relay/relay.ts b/src/relay/relay.ts index 2d4bcf44238..3df1555ee9b 100644 --- a/src/relay/relay.ts +++ b/src/relay/relay.ts @@ -81,7 +81,7 @@ function runConnectMode(sockPath: string): void { sock.on('connect', () => { clearTimeout(connectTimeout) runConnectHandshake(sock, myVersion, { - onAccepted: () => { + onAccepted: (leftover: Buffer) => { // Why: RELAY_SENTINEL must be written AFTER the handshake passes; if it // were written earlier, waitForSentinel on the client would resolve // and start sending JSON-RPC over a socket the daemon was about to @@ -89,6 +89,14 @@ function runConnectMode(sockPath: string): void { // re-entering the backoff loop. Sequencing it post-handshake makes // mismatch a clean exit-42 path with no false-positive sentinel. process.stdout.write(RELAY_SENTINEL) + // Why: bytes that arrived in the same TCP send as the handshake-ok + // frame were buffered inside the handshake's FrameDecoder. Forward + // them to stdout BEFORE attaching sock.pipe(process.stdout), so the + // multiplexer downstream sees them in order and no daemon frames + // are silently dropped at the transition. + if (leftover.length > 0) { + process.stdout.write(leftover) + } process.stdin.pipe(sock) sock.pipe(process.stdout) } @@ -220,7 +228,7 @@ function main(): void { let socketServer: Server | null = null const launchVersion = readLaunchVersion() - function attachAcceptedSocket(sock: Socket): void { + function attachAcceptedSocket(sock: Socket, leftover: Buffer): void { // Why: only one client at a time. If a second reconnect arrives (e.g. // user restarts again quickly), close the stale bridge so the new one // takes over cleanly. We null activeSocket BEFORE destroying so the old @@ -249,6 +257,14 @@ function main(): void { } }) + // Why: bytes that arrived in the same TCP send as the handshake frame + // were buffered inside the handshake's FrameDecoder. Feed them into the + // dispatcher BEFORE wiring sock.on('data'), so frame ordering is + // preserved and no client data is silently dropped at the transition. + if (leftover.length > 0) { + dispatcher.feed(leftover) + } + sock.on('data', (chunk: Buffer) => { if (activeSocket !== sock) { return