diff --git a/src/main/ssh/ssh-connection.test.ts b/src/main/ssh/ssh-connection.test.ts index e8c5d71234c..3274d3f5daa 100644 --- a/src/main/ssh/ssh-connection.test.ts +++ b/src/main/ssh/ssh-connection.test.ts @@ -23,6 +23,7 @@ type MockSshClient = { lastExecCommand?: string lastConnectConfig?: unknown exec: (cmd: string, cb: (err: Error | undefined, channel: unknown) => void) => void + sftp: (cb: (err: Error | undefined, channel: unknown) => void) => void } let clientInstances: MockSshClient[] = [] @@ -690,15 +691,15 @@ describe('SshConnection', () => { vi.useFakeTimers() try { - const outcomePromise = conn - .exec('printf ready') - .then(() => 'opened') - .catch((error: Error) => error.message) + const outcomePromise = conn.exec('printf ready').catch((error: Error) => error) await vi.advanceTimersByTimeAsync(30_000) const outcome = await Promise.race([outcomePromise, Promise.resolve('pending')]) - expect(outcome).toBe('SSH exec channel timed out') + expect(outcome).toMatchObject({ + message: 'SSH exec channel timed out', + sshChannelCloseConfirmed: false + }) } finally { vi.useRealTimers() } @@ -784,15 +785,54 @@ describe('SshConnection', () => { try { const outcomePromise = conn .exec('printf ready', { signal: controller.signal }) - .then(() => 'opened') - .catch((error: Error) => error.name) + .catch((error: Error) => error) controller.abort() // Why: a hung socket must not pin the aborted caller for the full 30s // connect timeout — the abort settles at the 5s grace bound instead. await vi.advanceTimersByTimeAsync(5_000) - await expect(outcomePromise).resolves.toBe('AbortError') + await expect(outcomePromise).resolves.toMatchObject({ + name: 'AbortError', + sshChannelCloseConfirmed: false + }) + } finally { + vi.useRealTimers() + } + }) + + it('drains an exec channel that opens after the abort grace has settled', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + await conn.connect() + execBehavior = 'pending' + const controller = new AbortController() + const lateChannel = Object.assign(new EventEmitter(), { + close: vi.fn(), + resume: vi.fn(), + stderr: { resume: vi.fn() } + }) + + vi.useFakeTimers() + try { + const outcomePromise = conn + .exec('printf ready', { signal: controller.signal }) + .catch((error: Error) => error) + + controller.abort() + await vi.advanceTimersByTimeAsync(5_000) + const outcome = await outcomePromise + expect(outcome).toMatchObject({ + name: 'AbortError', + sshChannelCloseConfirmed: false + }) + + pendingExecCallback?.(undefined, lateChannel) + + expect(lateChannel.resume).toHaveBeenCalledTimes(1) + expect(lateChannel.stderr.resume).toHaveBeenCalledTimes(1) + expect(lateChannel.close).toHaveBeenCalledTimes(1) + lateChannel.emit('close') + expect(outcome).toMatchObject({ sshChannelCloseConfirmed: true }) } finally { vi.useRealTimers() } @@ -837,7 +877,7 @@ describe('SshConnection', () => { const lateChannel = Object.assign(new EventEmitter(), { close: vi.fn(), resume: vi.fn(), - stderr: { resume: vi.fn() } + stderr: Object.assign(new EventEmitter(), { resume: vi.fn() }) }) const outcomePromise = conn @@ -855,11 +895,41 @@ describe('SshConnection', () => { expect(early).toBe('pending') expect(lateChannel.close).toHaveBeenCalledTimes(1) expect(lateChannel.resume).toHaveBeenCalled() + expect(() => lateChannel.emit('error', new Error('late channel teardown'))).not.toThrow() + expect(() => lateChannel.stderr.emit('error', new Error('late stderr teardown'))).not.toThrow() lateChannel.emit('close') await expect(outcomePromise).resolves.toBe('AbortError') }) + it('removes the late-channel close listener when abort grace expires', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + await conn.connect() + vi.useFakeTimers() + try { + sftpBehavior = 'pending' + const controller = new AbortController() + const lateSftp = Object.assign(new EventEmitter(), { end: vi.fn() }) + + const outcomePromise = conn + .sftp(controller.signal) + .then(() => 'opened') + .catch((error: Error) => error.name) + await vi.advanceTimersByTimeAsync(0) + controller.abort() + pendingSftpCallback?.(undefined, lateSftp) + expect(lateSftp.listenerCount('close')).toBe(1) + expect(() => lateSftp.emit('error', new Error('late SFTP teardown'))).not.toThrow() + + await vi.advanceTimersByTimeAsync(5_000) + + await expect(outcomePromise).resolves.toBe('AbortError') + expect(lateSftp.listenerCount('close')).toBe(0) + } finally { + vi.useRealTimers() + } + }) + it('times out when ssh2 never opens an SFTP channel', async () => { const conn = new SshConnection(createTarget(), createCallbacks()) await conn.connect() @@ -867,15 +937,13 @@ describe('SshConnection', () => { vi.useFakeTimers() try { - const outcomePromise = conn - .sftp() - .then(() => 'opened') - .catch((error: Error) => error.message) + const outcomePromise = conn.sftp().catch((error: Error) => error) await vi.advanceTimersByTimeAsync(30_000) const outcome = await Promise.race([outcomePromise, Promise.resolve('pending')]) - expect(outcome).toBe('SSH SFTP channel timed out') + expect(outcome).toMatchObject({ message: 'SSH SFTP channel timed out' }) + expect(outcome).not.toHaveProperty('sshChannelCloseConfirmed') } finally { vi.useRealTimers() } @@ -1368,6 +1436,61 @@ describe('SshConnection', () => { ) }) + it('composes a caller abort into system SSH relay uploads', async () => { + vi.mocked(resolveWithSshG).mockResolvedValueOnce(createResolvedConfig()) + const conn = new SshConnection(createTarget({ configHost: 'fdpass-host' }), createCallbacks()) + const controller = new AbortController() + let transferSignal: AbortSignal | undefined + vi.mocked(uploadDirectoryViaSystemSsh).mockImplementationOnce( + (_target, _localDir, _remoteDir, options) => { + transferSignal = options?.signal + return new Promise((_resolve, reject) => { + transferSignal?.addEventListener('abort', () => reject(transferSignal?.reason), { + once: true + }) + }) + } + ) + + await conn.connect() + const upload = conn.uploadDirectory('/tmp/local-relay', '/remote/relay', { + signal: controller.signal + }) + await vi.waitFor(() => expect(transferSignal).toBeDefined()) + controller.abort() + + await expect(upload).rejects.toMatchObject({ name: 'AbortError' }) + expect(transferSignal?.aborted).toBe(true) + }) + + it('keeps connection disconnect cancellation linked to caller-scoped relay writes', async () => { + vi.mocked(resolveWithSshG).mockResolvedValueOnce(createResolvedConfig()) + const conn = new SshConnection(createTarget({ configHost: 'fdpass-host' }), createCallbacks()) + const controller = new AbortController() + let transferSignal: AbortSignal | undefined + vi.mocked(writeFileViaSystemSsh).mockImplementationOnce( + (_target, _remotePath, _contents, options) => { + transferSignal = options?.signal + return new Promise((_resolve, reject) => { + transferSignal?.addEventListener('abort', () => reject(transferSignal?.reason), { + once: true + }) + }) + } + ) + + await conn.connect() + const write = conn.writeFile('/remote/relay/.version', '0.1.0', { + signal: controller.signal + }) + await vi.waitFor(() => expect(transferSignal).toBeDefined()) + await conn.disconnect() + + await expect(write).rejects.toMatchObject({ name: 'AbortError' }) + expect(controller.signal.aborted).toBe(false) + expect(transferSignal?.aborted).toBe(true) + }) + it('keeps an upload session cancelled after the connection disconnects', async () => { const conn = new SshConnection( createTarget({ proxyCommand: 'ssh -W %h:%p bastion.example.com' }), diff --git a/src/main/ssh/ssh-connection.ts b/src/main/ssh/ssh-connection.ts index 2856d33b2f0..b8bd0cbd57f 100644 --- a/src/main/ssh/ssh-connection.ts +++ b/src/main/ssh/ssh-connection.ts @@ -41,6 +41,10 @@ import { import type { RemoteHostPlatform } from './ssh-remote-platform' import type { FileUploadSession } from '../providers/types' import { isSshSessionLimitError } from './ssh-session-limit-error' +import { + createLinkedSshFileTransferSignal, + raceSftpFileTransferWithAbort +} from './ssh-file-transfer-abort' export type { SshConnectionCallbacks } from './ssh-connection-utils' type SshRemoteFileOptions = { @@ -157,13 +161,14 @@ export class SshConnection { 'SSH exec channel timed out', (callback) => client.exec(remoteCommand, callback), (channel) => channel.close(), - options?.signal + options?.signal, + true ), options?.signal ) } - async sftp(): Promise { + async sftp(signal?: AbortSignal): Promise { if (this.useSystemSshTransport) { throw new Error('SFTP is not available when using system SSH transport') } @@ -171,12 +176,15 @@ export class SshConnection { throw new Error('Not connected') } const client = this.client - return this.openSessionChannelWithRetry(() => - this.waitForSshCallback( - 'SSH SFTP channel timed out', - (callback) => client.sftp(callback), - (sftp) => sftp.end() - ) + return this.openSessionChannelWithRetry( + () => + this.waitForSshCallback( + 'SSH SFTP channel timed out', + (callback) => client.sftp(callback), + (sftp) => sftp.end(), + signal + ), + signal ) } @@ -219,10 +227,20 @@ export class SshConnection { timeoutMessage: string, register: (callback: (error: Error | undefined, value: T) => void) => void, cleanupLateValue?: (value: T) => void, - signal?: AbortSignal + signal?: AbortSignal, + trackRemoteCommandTermination = false ): Promise { return new Promise((resolve, reject) => { + type ChannelOpenTerminationError = Error & { sshChannelCloseConfirmed: boolean } let settled = false + let unconfirmedOpenError: ChannelOpenTerminationError | null = null + const markOpenUnconfirmed = (error: Error): Error => { + if (!trackRemoteCommandTermination) { + return error + } + unconfirmedOpenError = Object.assign(error, { sshChannelCloseConfirmed: false }) + return unconfirmedOpenError + } // Why: rejecting the instant the signal aborts lets the caller proceed // while the in-flight channel open completes in the background and holds // a server-side session slot (MaxSessions). Mark the abort and settle @@ -241,16 +259,41 @@ export class SshConnection { abortDeadlineTimer = setTimeout(() => { settled = true cleanup() - reject(createSshOperationAbortError()) + reject(markOpenUnconfirmed(createSshOperationAbortError())) }, ABORTED_CHANNEL_CLOSE_GRACE_MS) } const timer = setTimeout(() => { settled = true cleanup() - reject(abortRequested ? createSshOperationAbortError() : new Error(timeoutMessage)) + reject( + markOpenUnconfirmed( + abortRequested ? createSshOperationAbortError() : new Error(timeoutMessage) + ) + ) }, CONNECT_TIMEOUT_MS) + const discardLateValue = (value: T, onClose?: () => void): void => { + const emitter = value as Partial & { + resume?: () => void + stderr?: Partial & { resume?: () => void } + } + const swallowLateError = (): void => {} + emitter.on?.('error', swallowLateError) + emitter.stderr?.on?.('error', swallowLateError) + if (onClose) { + emitter.once?.('close', onClose) + } + // Why: ssh2 can withhold CHANNEL_CLOSE while discarded exec streams + // remain unread, and teardown errors have no other owner. + emitter.resume?.() + emitter.stderr?.resume?.() + try { + cleanupLateValue?.(value) + } catch { + /* best effort */ + } + } const rejectAfterClose = (value: T): void => { - const abortError = createSshOperationAbortError() + const abortError = markOpenUnconfirmed(createSshOperationAbortError()) const emitter = value as Partial & { resume?: () => void stderr?: { resume?: () => void } @@ -262,23 +305,24 @@ export class SshConnection { } finished = true clearTimeout(closeGraceTimer) + emitter.removeListener?.('close', confirmAndDone) reject(abortError) } + const confirmAndDone = (): void => { + if (unconfirmedOpenError === abortError) { + unconfirmedOpenError.sshChannelCloseConfirmed = true + } + done() + } // Why: bounded — a remote that never confirms the close must not hang // the aborted operation forever. const closeGraceTimer = setTimeout(done, ABORTED_CHANNEL_CLOSE_GRACE_MS) if (typeof emitter.once === 'function') { - emitter.once('close', done) + emitter.once('close', confirmAndDone) } // Why: ssh2 withholds the 'close' event until the channel's streams // are drained; nobody else will ever read this discarded channel. - emitter.resume?.() - emitter.stderr?.resume?.() - try { - cleanupLateValue?.(value) - } catch { - /* best effort */ - } + discardLateValue(value) if (typeof emitter.once !== 'function') { done() } @@ -289,11 +333,11 @@ export class SshConnection { // rejected. Close that late resource so the remote channel is not // left open with no owner. if (!error && value !== undefined) { - try { - cleanupLateValue?.(value) - } catch { - /* best effort */ - } + discardLateValue(value, () => { + if (unconfirmedOpenError) { + unconfirmedOpenError.sshChannelCloseConfirmed = true + } + }) } return } @@ -334,23 +378,51 @@ export class SshConnection { async uploadDirectory( localDir: string, remoteDir: string, - options?: SshRemoteFileOptions + options?: SshRemoteFileOptions & { signal?: AbortSignal } ): Promise { - if (!this.useSystemSshTransport) { - const sftp = await this.sftp() - try { - const { uploadDirectory } = await import('./ssh-relay-deploy-helpers') - await uploadDirectory(sftp, localDir, remoteDir) - } finally { - sftp.end() + // Why: relay deployment timeout and connection teardown are independent; + // either owner must stop a transfer that can otherwise outlive its lock. + const linkedSignal = createLinkedSshFileTransferSignal( + [this.systemOperationAbortController.signal, options?.signal].filter( + (signal): signal is AbortSignal => signal !== undefined + ) + ) + try { + if (!this.useSystemSshTransport) { + const sftp = await this.sftp(linkedSignal.signal) + const swallowLateSftpError = (): void => {} + let sftpEndRequested = false + const endSftp = (): void => { + if (!sftpEndRequested) { + sftpEndRequested = true + sftp.end() + } + } + sftp.on('error', swallowLateSftpError) + sftp.once('close', () => sftp.removeListener('error', swallowLateSftpError)) + try { + const { uploadDirectory } = await import('./ssh-relay-deploy-helpers') + await raceSftpFileTransferWithAbort( + uploadDirectory(sftp, localDir, remoteDir), + linkedSignal.signal, + (onClose) => { + sftp.once('close', onClose) + endSftp() + } + ) + } finally { + endSftp() + } + return } - return + await uploadDirectoryViaSystemSsh(this.target, localDir, remoteDir, { + signal: linkedSignal.signal, + hostPlatform: options?.hostPlatform, + ...this.getSystemSshBuildArgsOptions() + }) + } finally { + linkedSignal.dispose() } - await uploadDirectoryViaSystemSsh(this.target, localDir, remoteDir, { - signal: this.systemOperationAbortController.signal, - hostPlatform: options?.hostPlatform, - ...this.getSystemSshBuildArgsOptions() - }) } async downloadFile( @@ -404,55 +476,74 @@ export class SshConnection { async writeFile( remotePath: string, contents: string, - options?: SshRemoteFileOptions + options?: SshRemoteFileOptions & { signal?: AbortSignal } ): Promise { - if (!this.useSystemSshTransport) { - const sftp = await this.sftp() - const swallowLateSftpError = (): void => {} - sftp.on('error', swallowLateSftpError) - try { - await new Promise((resolve, reject) => { - const ws = sftp.createWriteStream(remotePath) - let settled = false - const cleanup = (): void => { - ws.removeListener('close', onClose) - ws.removeListener('error', onError) + // Keep package/version writes under the same dual cancellation contract as uploads. + const linkedSignal = createLinkedSshFileTransferSignal( + [this.systemOperationAbortController.signal, options?.signal].filter( + (signal): signal is AbortSignal => signal !== undefined + ) + ) + try { + if (!this.useSystemSshTransport) { + const sftp = await this.sftp(linkedSignal.signal) + const swallowLateSftpError = (): void => {} + let sftpEndRequested = false + const endSftp = (): void => { + if (!sftpEndRequested) { + sftpEndRequested = true + sftp.end() } - const onClose = (): void => { - if (settled) { - return + } + sftp.on('error', swallowLateSftpError) + sftp.once('close', () => sftp.removeListener('error', swallowLateSftpError)) + try { + const write = new Promise((resolve, reject) => { + const ws = sftp.createWriteStream(remotePath) + let settled = false + const cleanup = (): void => { + sftp.removeListener('error', onError) + ws.removeListener('close', onClose) + ws.removeListener('error', onError) } - settled = true - cleanup() - resolve() - } - const onError = (err: Error): void => { - sftp.removeListener('error', onError) - if (settled) { - return + const onClose = (): void => { + if (settled) { + return + } + settled = true + cleanup() + resolve() } - settled = true - cleanup() - reject(err) - } - sftp.prependOnceListener('error', onError) - ws.once('close', onClose) - ws.once('error', onError) - ws.end(contents) - }) - } finally { - sftp.end() - setImmediate(() => { - sftp.removeListener('error', swallowLateSftpError) - }) + const onError = (err: Error): void => { + if (settled) { + return + } + settled = true + cleanup() + reject(err) + } + sftp.prependOnceListener('error', onError) + ws.once('close', onClose) + ws.once('error', onError) + ws.end(contents) + }) + await raceSftpFileTransferWithAbort(write, linkedSignal.signal, (onClose) => { + sftp.once('close', onClose) + endSftp() + }) + } finally { + endSftp() + } + return } - return + await writeFileViaSystemSsh(this.target, remotePath, contents, { + signal: linkedSignal.signal, + hostPlatform: options?.hostPlatform, + ...this.getSystemSshBuildArgsOptions() + }) + } finally { + linkedSignal.dispose() } - await writeFileViaSystemSsh(this.target, remotePath, contents, { - signal: this.systemOperationAbortController.signal, - hostPlatform: options?.hostPlatform, - ...this.getSystemSshBuildArgsOptions() - }) } async writeBuffer( diff --git a/src/main/ssh/ssh-file-transfer-abort.test.ts b/src/main/ssh/ssh-file-transfer-abort.test.ts new file mode 100644 index 00000000000..49abe0a1839 --- /dev/null +++ b/src/main/ssh/ssh-file-transfer-abort.test.ts @@ -0,0 +1,55 @@ +import { describe, expect, it, vi } from 'vitest' +import { raceSftpFileTransferWithAbort } from './ssh-file-transfer-abort' + +describe('raceSftpFileTransferWithAbort', () => { + it('waits for confirmed SFTP close before rejecting an abort', async () => { + const controller = new AbortController() + let confirmClose: () => void = () => {} + const promise = raceSftpFileTransferWithAbort( + new Promise(() => {}), + controller.signal, + (onClose) => { + confirmClose = onClose + } + ) + + controller.abort() + const pending = await Promise.race([ + promise.then( + () => 'settled', + () => 'settled' + ), + Promise.resolve('pending') + ]) + expect(pending).toBe('pending') + + confirmClose() + await expect(promise).rejects.toMatchObject({ + name: 'AbortError', + sshChannelCloseConfirmed: true + }) + }) + + it('marks teardown unconfirmed when SFTP never closes', async () => { + vi.useFakeTimers() + try { + const controller = new AbortController() + const promise = raceSftpFileTransferWithAbort( + new Promise(() => {}), + controller.signal, + () => {} + ) + const outcome = promise.catch((error: Error) => error) + + controller.abort() + await vi.advanceTimersByTimeAsync(5_000) + + await expect(outcome).resolves.toMatchObject({ + name: 'AbortError', + sshChannelCloseConfirmed: false + }) + } finally { + vi.useRealTimers() + } + }) +}) diff --git a/src/main/ssh/ssh-file-transfer-abort.ts b/src/main/ssh/ssh-file-transfer-abort.ts new file mode 100644 index 00000000000..a7b6cce2e49 --- /dev/null +++ b/src/main/ssh/ssh-file-transfer-abort.ts @@ -0,0 +1,76 @@ +import { createSshOperationAbortError } from './ssh-connection-utils' + +export function createLinkedSshFileTransferSignal(signals: readonly AbortSignal[]): { + signal: AbortSignal + dispose: () => void +} { + const controller = new AbortController() + const listeners = signals.map((signal) => { + const listener = (): void => controller.abort(signal.reason) + if (signal.aborted) { + listener() + } else { + signal.addEventListener('abort', listener, { once: true }) + } + return { signal, listener } + }) + return { + signal: controller.signal, + dispose: () => { + for (const { signal, listener } of listeners) { + signal.removeEventListener('abort', listener) + } + } + } +} + +export function raceSftpFileTransferWithAbort( + operation: Promise, + signal: AbortSignal, + closeSftp: (onClose: () => void) => void +): Promise { + return new Promise((resolve, reject) => { + let settled = false + let abortError: (Error & { sshChannelCloseConfirmed: boolean }) | null = null + let closeGraceTimer: ReturnType | null = null + const settle = (fn: typeof resolve | typeof reject, value: T | Error): void => { + if (settled) { + return + } + settled = true + if (closeGraceTimer) { + clearTimeout(closeGraceTimer) + } + signal.removeEventListener('abort', onAbort) + fn(value as never) + } + const onAbort = (): void => { + // Why: rejecting the caller is insufficient; ending SFTP stops the + // abandoned transfer from mutating a successor relay install. + abortError = Object.assign(createSshOperationAbortError(), { + sshChannelCloseConfirmed: false + }) + closeGraceTimer = setTimeout(() => settle(reject, abortError!), 5_000) + closeSftp(() => { + abortError!.sshChannelCloseConfirmed = true + settle(reject, abortError!) + }) + } + signal.addEventListener('abort', onAbort, { once: true }) + void operation.then( + (value) => { + if (!abortError) { + settle(resolve, value) + } + }, + (error: unknown) => { + if (!abortError) { + settle(reject, error instanceof Error ? error : new Error(String(error))) + } + } + ) + if (signal.aborted) { + onAbort() + } + }) +} diff --git a/src/main/ssh/ssh-relay-build-toolchain.ts b/src/main/ssh/ssh-relay-build-toolchain.ts index cdbdb676441..208deceb6e0 100644 --- a/src/main/ssh/ssh-relay-build-toolchain.ts +++ b/src/main/ssh/ssh-relay-build-toolchain.ts @@ -141,15 +141,20 @@ export function formatMissingToolchainError( // callers fall back to the original install error in those cases. export async function probeBuildToolchain( conn: SshConnection, - hostPlatform: RemoteHostPlatform + hostPlatform: RemoteHostPlatform, + signal?: AbortSignal ): Promise { if (isWindowsRemoteHost(hostPlatform)) { return null } try { - const output = await execCommand(conn, buildToolchainProbeCommand(), { wrapCommand: true }) + const output = await execCommand(conn, buildToolchainProbeCommand(), { + wrapCommand: true, + signal + }) return parseBuildToolchainProbe(output) } catch { + signal?.throwIfAborted() return null } } 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 a29801c3632..da132775545 100644 --- a/src/main/ssh/ssh-relay-cross-version-isolation.test.ts +++ b/src/main/ssh/ssh-relay-cross-version-isolation.test.ts @@ -33,6 +33,9 @@ vi.mock('./ssh-relay-deploy-helpers', () => ({ onData: vi.fn(), onClose: vi.fn() }), + isUnconfirmedSshCommandTermination: (error: unknown) => + error instanceof Error && + (error as Error & { sshChannelCloseConfirmed?: boolean }).sshChannelCloseConfirmed === false, execCommand: vi.fn() })) diff --git a/src/main/ssh/ssh-relay-deploy-helpers.test.ts b/src/main/ssh/ssh-relay-deploy-helpers.test.ts index cb5477091d6..4df9c8ad28a 100644 --- a/src/main/ssh/ssh-relay-deploy-helpers.test.ts +++ b/src/main/ssh/ssh-relay-deploy-helpers.test.ts @@ -11,9 +11,10 @@ import { function createMockChannel(): ClientChannel { return Object.assign(new EventEmitter(), { - stderr: new EventEmitter(), + stderr: Object.assign(new EventEmitter(), { resume: vi.fn() }), stdin: { write: vi.fn() }, - close: vi.fn() + close: vi.fn(), + resume: vi.fn() }) as unknown as ClientChannel } @@ -29,6 +30,44 @@ async function execCommandRejection(promise: Promise): Promise { } describe('waitForSentinel', () => { + it('closes and rejects with AbortError while clearing startup resources on abort', async () => { + vi.useFakeTimers() + try { + const channel = createMockChannel() + const controller = new AbortController() + const removeListener = vi.spyOn(controller.signal, 'removeEventListener') + const transportPromise = waitForSentinel(channel, controller.signal) + + expect(vi.getTimerCount()).toBe(1) + controller.abort() + + await expect(transportPromise).rejects.toMatchObject({ name: 'AbortError' }) + expect(channel.close).toHaveBeenCalledTimes(1) + expect(removeListener).toHaveBeenCalledWith('abort', expect.any(Function)) + expect(vi.getTimerCount()).toBe(0) + } finally { + vi.useRealTimers() + } + }) + + it('removes the abort listener and timers when startup fails before the sentinel', async () => { + vi.useFakeTimers() + try { + const channel = createMockChannel() + const controller = new AbortController() + const removeListener = vi.spyOn(controller.signal, 'removeEventListener') + const transportPromise = waitForSentinel(channel, controller.signal) + + channel.emit('error', new Error('remote host rebooted')) + + await expect(transportPromise).rejects.toThrow('remote host rebooted') + expect(removeListener).toHaveBeenCalledWith('abort', expect.any(Function)) + expect(vi.getTimerCount()).toBe(0) + } finally { + vi.useRealTimers() + } + }) + it('buffers post-sentinel chunks until the transport subscribes', async () => { const channel = createMockChannel() const transportPromise = waitForSentinel(channel) @@ -168,6 +207,32 @@ describe('waitForSentinel', () => { }) describe('execCommand', () => { + it('waits for channel close before rejecting a timed-out remote command', async () => { + vi.useFakeTimers() + try { + const channel = createMockChannel() + const conn = { exec: vi.fn().mockResolvedValue(channel) } + const commandPromise = execCommand(conn as never, 'npm rebuild', { timeoutMs: 1_000 }) + + await vi.advanceTimersByTimeAsync(1_000) + expect(channel.close).toHaveBeenCalledTimes(1) + expect( + await Promise.race([ + commandPromise.then( + () => 'settled', + () => 'settled' + ), + Promise.resolve('pending') + ]) + ).toBe('pending') + + channel.emit('close', 0) + await expect(commandPromise).rejects.toThrow('timed out after 1s') + } finally { + vi.useRealTimers() + } + }) + it('rejects on command channel errors instead of emitting uncaught errors', async () => { const channel = createMockChannel() const conn = { @@ -177,6 +242,8 @@ describe('execCommand', () => { await Promise.resolve() expect(() => channel.emit('error', new Error('remote host rebooted'))).not.toThrow() + expect(channel.close).toHaveBeenCalledOnce() + channel.emit('close', 1) await expect(commandPromise).rejects.toThrow('remote host rebooted') expect(channel.listenerCount('error')).toBe(0) expect(channel.listenerCount('data')).toBe(0) @@ -224,6 +291,22 @@ describe('execCommand', () => { expect(error.message).toContain('gnu++20') }) + it('bounds command output while preserving the actionable tail', async () => { + const channel = createMockChannel() + const conn = { exec: vi.fn().mockResolvedValue(channel) } + const commandPromise = execCommand(conn as never, 'npm rebuild') + + await Promise.resolve() + channel.emit('data', Buffer.from(`old-prefix-${'x'.repeat(1024 * 1024)}`)) + channel.emit('data', Buffer.from('gyp ERR! tail diagnosis')) + channel.emit('close', 1) + + const error = await execCommandRejection(commandPromise) + expect(error.message).not.toContain('old-prefix') + expect(error.message).toContain('gyp ERR! tail diagnosis') + expect(error.message.length).toBeLessThan(1024 * 1024 + 200) + }) + it('keeps the merged error message greppable by the build-toolchain probe', async () => { const channel = createMockChannel() const conn = { @@ -260,14 +343,26 @@ describe('execCommand', () => { await Promise.resolve() const rejection = expect(commandPromise).rejects.toThrow('timed out') await vi.advanceTimersByTimeAsync(30_000) + expect(channel.close).toHaveBeenCalledOnce() + await vi.advanceTimersByTimeAsync(5_000) await rejection expect(channel.close).toHaveBeenCalledOnce() - expect(channel.listenerCount('error')).toBe(0) + expect(channel.resume).toHaveBeenCalledOnce() + expect(channel.stderr.resume).toHaveBeenCalledOnce() + expect(() => channel.emit('error', new Error('late channel error'))).not.toThrow() + expect(() => channel.stderr.emit('error', new Error('late stderr error'))).not.toThrow() + channel.emit('data', Buffer.from('late stdout')) + channel.stderr.emit('data', Buffer.from('late stderr')) + expect(channel.listenerCount('error')).toBe(1) expect(channel.listenerCount('data')).toBe(0) + expect(channel.listenerCount('close')).toBe(1) + expect(channel.stderr.listenerCount('error')).toBe(1) + expect(channel.stderr.listenerCount('data')).toBe(0) + channel.emit('close', 0) + expect(channel.listenerCount('error')).toBe(0) expect(channel.listenerCount('close')).toBe(0) expect(channel.stderr.listenerCount('error')).toBe(0) - expect(channel.stderr.listenerCount('data')).toBe(0) } finally { vi.useRealTimers() } @@ -309,6 +404,48 @@ describe('execCommand', () => { expect(channel.stderr.listenerCount('data')).toBe(0) }) + it('keeps system-SSH command termination unconfirmed after the local child closes', async () => { + const channel = createMockChannel() + const controller = new AbortController() + const conn = { + exec: vi.fn().mockResolvedValue(channel), + usesSystemSshTransport: vi.fn().mockReturnValueOnce(true).mockReturnValue(false) + } + const commandPromise = execCommand(conn as never, 'npm install', { + signal: controller.signal + }) + + await Promise.resolve() + controller.abort() + channel.emit('close', 0) + + await expect(commandPromise).rejects.toMatchObject({ + name: 'AbortError', + sshChannelCloseConfirmed: false + }) + }) + + it('tags an externally closed system-SSH command as unconfirmed', async () => { + const channel = createMockChannel() as ClientChannel & { _closeRequested?: boolean } + channel.close = vi.fn(() => { + channel._closeRequested = true + }) + const conn = { + exec: vi.fn().mockResolvedValue(channel), + usesSystemSshTransport: vi.fn().mockReturnValue(true) + } + const commandPromise = execCommand(conn as never, 'npm install') + + await Promise.resolve() + channel.close() + channel.emit('close', 1) + + await expect(commandPromise).rejects.toMatchObject({ + name: 'AbortError', + sshChannelCloseConfirmed: false + }) + }) + it('handles aborts that happen while the SSH exec channel is still opening', async () => { const channel = createMockChannel() const controller = new AbortController() @@ -357,6 +494,8 @@ describe('execCommand', () => { expect(conn.exec).toHaveBeenCalledWith('npm install', { wrapCommand: false }) const rejection = expect(commandPromise).rejects.toThrow('timed out after 240s') await vi.advanceTimersByTimeAsync(240_000) + expect(channel.close).toHaveBeenCalledOnce() + await vi.advanceTimersByTimeAsync(5_000) await rejection expect(channel.close).toHaveBeenCalledOnce() diff --git a/src/main/ssh/ssh-relay-deploy-helpers.ts b/src/main/ssh/ssh-relay-deploy-helpers.ts index 1b1cdd27074..80d2b78ced2 100644 --- a/src/main/ssh/ssh-relay-deploy-helpers.ts +++ b/src/main/ssh/ssh-relay-deploy-helpers.ts @@ -1,18 +1,21 @@ import type { ClientChannel } from 'ssh2' -import type { SshConnection } from './ssh-connection' -import { createSshOperationAbortError, type SshExecOptions } from './ssh-connection-utils' +import { createSshOperationAbortError } from './ssh-connection-utils' import { RELAY_SENTINEL, RELAY_SENTINEL_TIMEOUT_MS } from './relay-protocol' import type { MultiplexerTransport } from './ssh-channel-multiplexer' import { buildRelayVersionMismatchError } from './ssh-relay-handshake-mismatch' export { uploadFile, uploadDirectory, mkdirSftp } from './sftp-upload' +export { execCommand, isUnconfirmedSshCommandTermination } from './ssh-relay-exec-command' // ── Sentinel detection ──────────────────────────────────────────────── const MAX_RELAY_STARTUP_BUFFER_BYTES = 64 * 1024 const RELAY_SENTINEL_BUFFER = Buffer.from(RELAY_SENTINEL, 'utf-8') -export function waitForSentinel(channel: ClientChannel): Promise { +export function waitForSentinel( + channel: ClientChannel, + signal?: AbortSignal +): Promise { return new Promise((resolve, reject) => { let sentinelReceived = false let settled = false @@ -39,17 +42,14 @@ export function waitForSentinel(channel: ClientChannel): Promise { 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()}` : ''}` - ) + rejectStartup( + new Error( + `Relay failed to start within ${RELAY_SENTINEL_TIMEOUT_MS / 1000}s.${stderrOutput ? ` stderr: ${stderrOutput.trim()}` : ''}` ) - } + ) }, TIMEOUT_GRACE_MS) + channel.close() }, RELAY_SENTINEL_TIMEOUT_MS) const cancelTimers = (): void => { @@ -59,6 +59,31 @@ export function waitForSentinel(channel: ClientChannel): Promise { + cancelTimers() + signal?.removeEventListener('abort', onAbort) + } + const rejectStartup = (err: Error): void => { + if (settled) { + return + } + settled = true + cleanupStartup() + reject(err) + } + const onAbort = (): void => { + if (settled) { + return + } + settled = true + cleanupStartup() + channel.close() + reject(createSshOperationAbortError()) + } + signal?.addEventListener('abort', onAbort, { once: true }) + if (signal?.aborted) { + onAbort() + } channel.on('exit', (code: number | null) => { if (typeof code === 'number') { @@ -87,12 +112,8 @@ export function waitForSentinel(channel: ClientChannel): Promise { - cancelTimers() if (!sentinelReceived) { - if (!settled) { - settled = true - reject(err) - } + rejectStartup(err) return } notifyClosed() @@ -113,9 +134,7 @@ export function waitForSentinel(channel: ClientChannel): Promise { if (!sentinelReceived) { - cancelTimers() if (!settled) { - settled = true // Why: a wire-handshake mismatch on the daemon side closes the // socket; --connect prints the mismatch detail to stderr and exits // with code 42 BEFORE writing the sentinel. Translate that into a @@ -126,13 +145,13 @@ export function waitForSentinel(channel: ClientChannel): Promise 0) { pendingAfterSentinel = afterSentinel } - settled = true - const transport: MultiplexerTransport = { write: (buf: Buffer) => channel.stdin.write(buf), onData: (cb) => { @@ -233,97 +251,3 @@ export function waitForSentinel(channel: ClientChannel): Promise { - const { timeoutMs = EXEC_TIMEOUT_MS, ...execOptions } = options ?? {} - const signal = options?.signal - if (signal?.aborted) { - throw createSshOperationAbortError() - } - const channel = await conn.exec(command, execOptions) - return new Promise((resolve, reject) => { - let stdout = '' - let stderr = '' - let settled = false - - const cleanup = (): void => { - clearTimeout(timeout) - signal?.removeEventListener('abort', onAbort) - channel.off('error', fail) - channel.stderr.off('error', fail) - channel.off('data', onStdoutData) - channel.stderr.off('data', onStderrData) - channel.off('close', onClose) - } - const settle = (fn: typeof resolve | typeof reject, val: string | Error): void => { - if (settled) { - return - } - settled = true - cleanup() - fn(val as never) - } - const fail = (err: Error): void => { - settle(reject, abortRequested ? createSshOperationAbortError() : err) - } - // Why: sshd counts the session against MaxSessions until CHANNEL_CLOSE - // completes. Settling on abort before the channel actually closes lets the - // concurrent-bootstrap sequential fallback reissue an exec while the slot - // is still held, so it gets refused again. Close and settle from onClose. - let abortRequested = false - const onAbort = (): void => { - abortRequested = true - channel.close() - } - const onStdoutData = (data: Buffer): void => { - stdout += data.toString('utf-8') - } - const onStderrData = (data: Buffer): void => { - stderr += data.toString('utf-8') - } - const onClose = (code: number): void => { - if (abortRequested) { - settle(reject, createSshOperationAbortError()) - } else if (code !== 0) { - // Why: on the system-ssh transport channel.stderr carries local OpenSSH - // client noise; preferring it masks the real failure in stdout (2>&1). - const output = [stderr.trim(), stdout.trim()].filter(Boolean).join('\n') - settle(reject, new Error(`Command "${command}" failed (exit ${code}): ${output}`)) - } else { - settle(resolve, stdout) - } - } - const timeout = setTimeout(() => { - channel.close() - settle( - reject, - abortRequested - ? createSshOperationAbortError() - : new Error(`Command "${command}" timed out after ${timeoutMs / 1000}s`) - ) - }, timeoutMs) - - // Why: remote reboot tears down exec channels with stream errors. Without - // scoped listeners, Node treats those as uncaught exceptions. - signal?.addEventListener('abort', onAbort, { once: true }) - channel.on('error', fail) - channel.stderr.on('error', fail) - channel.on('data', onStdoutData) - channel.stderr.on('data', onStderrData) - channel.on('close', onClose) - if (signal?.aborted) { - onAbort() - } - }) -} diff --git a/src/main/ssh/ssh-relay-deploy-timing.ts b/src/main/ssh/ssh-relay-deploy-timing.ts index 0bd2ca95a17..5dc24f937f8 100644 --- a/src/main/ssh/ssh-relay-deploy-timing.ts +++ b/src/main/ssh/ssh-relay-deploy-timing.ts @@ -5,9 +5,9 @@ export const NATIVE_DEPS_COMMAND_TIMEOUT_MS = 240_000 // Why: a missing binding can require both install and rebuild while the same // install lock is held. Concurrent first installs must wait for that valid // holder instead of failing halfway through its bounded work. -export const NATIVE_DEPS_REPAIR_BUDGET_MS = 2 * NATIVE_DEPS_COMMAND_TIMEOUT_MS +const NATIVE_DEPS_REPAIR_BUDGET_MS = 2 * NATIVE_DEPS_COMMAND_TIMEOUT_MS -// Why: individual exec commands have their own timeouts, but the full deploy -// pipeline has no overall bound. Leave headroom after the native-deps budget -// for platform detection, upload, probes, and launch. -export const RELAY_DEPLOY_TIMEOUT_MS = NATIVE_DEPS_REPAIR_BUDGET_MS + 60_000 +// Why: native repair also runs separately bounded chmod/probe/diagnostic/ +// finalize commands around install + rebuild. Keep the outer bound above that +// valid worst path while remaining below the 20-minute stale-lock threshold. +export const RELAY_DEPLOY_TIMEOUT_MS = NATIVE_DEPS_REPAIR_BUDGET_MS + 7 * 60_000 diff --git a/src/main/ssh/ssh-relay-deploy.test.ts b/src/main/ssh/ssh-relay-deploy.test.ts index 4eb5bd57364..70c253934b0 100644 --- a/src/main/ssh/ssh-relay-deploy.test.ts +++ b/src/main/ssh/ssh-relay-deploy.test.ts @@ -42,6 +42,9 @@ vi.mock('./ssh-relay-deploy-helpers', () => ({ onData: vi.fn(), onClose: vi.fn() }), + isUnconfirmedSshCommandTermination: (error: unknown) => + error instanceof Error && + (error as Error & { sshChannelCloseConfirmed?: boolean }).sshChannelCloseConfirmed === false, execCommand: vi.fn().mockResolvedValue('__ORCA_REMOTE_PLATFORM__ Linux x86_64') })) @@ -49,31 +52,38 @@ vi.mock('./ssh-remote-node-resolution', () => ({ resolveRemoteNodePath: vi.fn().mockResolvedValue('/usr/bin/node') })) -// Why: the versioned-install module shells out to the remote for install -// state, lock acquisition, and GC. Tests stub these to no-ops so the deploy -// happy-path is exercised without a real SSH connection. +// Why: the versioned-install modules shell out for install state, locking, +// and GC. Stub them so deploy tests need no real SSH connection. vi.mock('./ssh-relay-versioned-install', () => ({ readLocalFullVersion: vi.fn().mockReturnValue('0.1.0+abcdef012345'), computeRemoteRelayDir: (home: string, v: string) => `${home}/.orca-remote/relay-${v}`, isRelayAlreadyInstalled: vi.fn().mockResolvedValue(true), - acquireInstallLock: vi.fn().mockResolvedValue(undefined), finalizeInstall: vi.fn().mockResolvedValue(undefined), abandonInstall: vi.fn().mockResolvedValue(undefined), gcOldRelayVersions: vi.fn().mockResolvedValue(undefined) })) +vi.mock('./ssh-relay-install-lock', () => ({ + acquireInstallLock: vi.fn().mockResolvedValue(undefined) +})) + vi.mock('./ssh-relay-repair-lock', () => ({ tryAcquireRelayRepairLock: vi.fn().mockResolvedValue(true) })) vi.mock('./ssh-connection-utils', () => ({ - shellEscape: (s: string) => `'${s}'` + shellEscape: (s: string) => `'${s}'`, + createSshOperationAbortError: () => + Object.assign(new Error('SSH operation was cancelled'), { + name: 'AbortError' + }) })) import { deployAndLaunchRelay } from './ssh-relay-deploy' import { execCommand, waitForSentinel } from './ssh-relay-deploy-helpers' import { resolveRemoteNodePath } from './ssh-remote-node-resolution' import { isRelayAlreadyInstalled } from './ssh-relay-versioned-install' +import { acquireInstallLock } from './ssh-relay-install-lock' import type { SshConnection } from './ssh-connection' import type * as SshRemoteNodeResolution from './ssh-remote-node-resolution' import { @@ -137,7 +147,8 @@ describe('deployAndLaunchRelay', () => { expect(mockExecCommand).toHaveBeenCalledWith( conn, - "printf '\\n%s ' '__ORCA_REMOTE_PLATFORM__'; uname -sm" + "printf '\\n%s ' '__ORCA_REMOTE_PLATFORM__'; uname -sm", + expect.objectContaining({ signal: expect.any(AbortSignal) }) ) }) @@ -157,6 +168,26 @@ describe('deployAndLaunchRelay', () => { expect(progress).toContain('Starting relay...') }) + it('does not launch fresh after unconfirmed stale-socket cleanup', async () => { + const conn = makeMockConnection() + const unconfirmedCleanup = Object.assign(new Error('socket cleanup still running'), { + sshChannelCloseConfirmed: false + }) + vi.mocked(waitForSentinel).mockRejectedValueOnce(new Error('stale relay reconnect failed')) + vi.mocked(execCommand) + .mockResolvedValueOnce('__ORCA_REMOTE_PLATFORM__ Linux x86_64') + .mockResolvedValueOnce('/home/user') + .mockResolvedValueOnce('ORCA-NATIVE-DEPS-OK') + .mockResolvedValueOnce('ALIVE') + .mockRejectedValueOnce(unconfirmedCleanup) + + await expect(deployAndLaunchRelay(conn)).rejects.toBe(unconfirmedCleanup) + + const commands = vi.mocked(conn.exec).mock.calls.map(([command]) => command) + expect(commands).toHaveLength(1) + expect(commands.some((command) => command.includes('--detached'))).toBe(false) + }) + it('resolves the remote node path once per deploy', async () => { const conn = makeMockConnection() const mockExecCommand = vi.mocked(execCommand) @@ -322,7 +353,10 @@ describe('deployAndLaunchRelay', () => { expect(vi.mocked(isRelayAlreadyInstalled).mock.calls[0]?.[3]).toMatchObject({ rethrowSessionLimitErrors: true }) - expect(vi.mocked(isRelayAlreadyInstalled).mock.calls[1]?.[3]).toBeUndefined() + expect(vi.mocked(isRelayAlreadyInstalled).mock.calls[1]?.[3]).toMatchObject({ + rethrowSessionLimitErrors: undefined, + signal: expect.any(AbortSignal) + }) expect(resolveRemoteNodePath).toHaveBeenCalledTimes(2) }) @@ -522,15 +556,136 @@ describe('deployAndLaunchRelay', () => { await vi.advanceTimersByTimeAsync(301_000) expect(await Promise.race([promise, Promise.resolve('pending')])).toBe('pending') - await vi.advanceTimersByTimeAsync(240_000) + await vi.advanceTimersByTimeAsync(600_000) const result = await promise expect(result).toBeInstanceOf(Error) - expect((result as Error).message).toBe('Relay deployment timed out after 540s') + expect((result as Error).message).toBe('Relay deployment timed out after 900s') vi.useRealTimers() }) + it('aborts a contended install-lock wait at the overall deploy timeout', async () => { + vi.useFakeTimers() + try { + const conn = makeMockConnection() + vi.mocked(execCommand) + .mockResolvedValueOnce('__ORCA_REMOTE_PLATFORM__ Linux x86_64') + .mockResolvedValueOnce('/home/user') + vi.mocked(isRelayAlreadyInstalled).mockResolvedValueOnce(false) + let lockSignal: AbortSignal | undefined + vi.mocked(acquireInstallLock).mockImplementationOnce((_conn, _dir, _host, options) => { + lockSignal = options?.signal + return new Promise((_resolve, reject) => { + lockSignal?.addEventListener('abort', () => reject(lockSignal?.reason), { once: true }) + }) + }) + + const promise = deployAndLaunchRelay(conn).catch((err: Error) => err) + await vi.advanceTimersByTimeAsync(0) + expect(acquireInstallLock).toHaveBeenCalledTimes(1) + + await vi.advanceTimersByTimeAsync(900_000) + + const result = await promise + expect(result).toBeInstanceOf(Error) + expect((result as Error).message).toBe('Relay deployment timed out after 900s') + expect(lockSignal?.aborted).toBe(true) + } finally { + vi.useRealTimers() + } + }) + + it('aborts an in-progress relay upload at the overall deploy timeout', async () => { + vi.useFakeTimers() + try { + const conn = makeMockConnection() + vi.mocked(execCommand) + .mockResolvedValueOnce('__ORCA_REMOTE_PLATFORM__ Linux x86_64') + .mockResolvedValueOnce('/home/user') + .mockResolvedValueOnce('') // mkdir remote relay dir + vi.mocked(isRelayAlreadyInstalled).mockResolvedValueOnce(false).mockResolvedValueOnce(false) + let uploadSignal: AbortSignal | undefined + conn.uploadDirectory = vi.fn((_localDir, _remoteDir, options) => { + uploadSignal = options?.signal + return new Promise((_resolve, reject) => { + uploadSignal?.addEventListener('abort', () => reject(uploadSignal?.reason), { + once: true + }) + }) + }) + + const promise = deployAndLaunchRelay(conn).catch((err: Error) => err) + await vi.advanceTimersByTimeAsync(0) + expect(conn.uploadDirectory).toHaveBeenCalledTimes(1) + + await vi.advanceTimersByTimeAsync(900_000) + + const result = await promise + expect(result).toBeInstanceOf(Error) + expect((result as Error).message).toBe('Relay deployment timed out after 900s') + expect(uploadSignal?.aborted).toBe(true) + } finally { + vi.useRealTimers() + } + }) + + it('aborts a launch started near the deploy deadline and closes its channel once', async () => { + vi.useFakeTimers() + try { + const launchChannel = { + on: vi.fn(), + stderr: { on: vi.fn() }, + stdin: {}, + stdout: { on: vi.fn() }, + close: vi.fn() + } + const conn = makeMockConnection() + vi.mocked(conn.exec).mockResolvedValue(launchChannel as never) + const mockExecCommand = vi.mocked(execCommand) + mockExecCommand + .mockResolvedValueOnce('__ORCA_REMOTE_PLATFORM__ Linux x86_64') + .mockResolvedValueOnce('/home/user') + .mockImplementationOnce( + () => + new Promise((resolve) => + setTimeout(() => resolve('ORCA-NATIVE-DEPS-OK'), 899_900) + ) + ) + .mockResolvedValueOnce('DEAD') + .mockImplementationOnce((_conn, _command, options) => { + return new Promise((_resolve, reject) => { + options?.signal?.addEventListener( + 'abort', + () => { + const error = new Error('SSH operation was cancelled') + error.name = 'AbortError' + reject(error) + }, + { once: true } + ) + }) + }) + + const promise = deployAndLaunchRelay(conn).catch((err: Error) => err) + await vi.advanceTimersByTimeAsync(899_900) + expect(conn.exec).toHaveBeenCalledTimes(1) + expect(launchChannel.close).not.toHaveBeenCalled() + + await vi.advanceTimersByTimeAsync(100) + + const result = await promise + expect(result).toBeInstanceOf(Error) + expect((result as Error).message).toBe('Relay deployment timed out after 900s') + expect(launchChannel.close).toHaveBeenCalledTimes(1) + expect(mockExecCommand).toHaveBeenCalledTimes(5) + await vi.advanceTimersByTimeAsync(10_000) + expect(mockExecCommand).toHaveBeenCalledTimes(5) + } finally { + vi.useRealTimers() + } + }) + it('uses distinct target-specific relay socket paths', async () => { const connA = makeMockConnection() const connB = makeMockConnection() @@ -607,6 +762,20 @@ describe('deployAndLaunchRelay', () => { expect(launchScript).not.toContain('\\\\.\\pipe\\agent-hooks') const waitScript = decodedScripts.find((script) => script.includes('deadline=Date.now()')) ?? '' expect(waitScript).toContain('setTimeout(attempt,intervalMs)') + const windowsLaunchCalls = mockExecCommand.mock.calls.filter(([, command]) => { + const script = decodePowerShellCommand(command) + return ( + script?.includes('.windows-active-pipe') || + script?.includes('Invoke-CimMethod') || + script?.includes('deadline=Date.now()') + ) + }) + expect(windowsLaunchCalls.length).toBeGreaterThan(0) + expect( + windowsLaunchCalls.every(([, , options]) => options?.signal instanceof AbortSignal) + ).toBe(true) + expect(vi.mocked(conn.exec).mock.calls[0]?.[1]?.signal).toBeInstanceOf(AbortSignal) + expect(vi.mocked(waitForSentinel).mock.calls[0]?.[1]).toBeInstanceOf(AbortSignal) }) it('relaunches Windows remotes on a fallback pipe when reconnecting the occupied pipe fails', async () => { diff --git a/src/main/ssh/ssh-relay-deploy.ts b/src/main/ssh/ssh-relay-deploy.ts index fb0bad68c8f..35161c15ccc 100644 --- a/src/main/ssh/ssh-relay-deploy.ts +++ b/src/main/ssh/ssh-relay-deploy.ts @@ -8,20 +8,25 @@ import { app } from 'electron' import type { SshConnection } from './ssh-connection' import type { RelayPlatform } from './relay-protocol' import type { MultiplexerTransport } from './ssh-channel-multiplexer' -import { uploadDirectory, waitForSentinel, execCommand } from './ssh-relay-deploy-helpers' +import { + uploadDirectory, + waitForSentinel, + execCommand, + isUnconfirmedSshCommandTermination +} from './ssh-relay-deploy-helpers' import { resolveRemoteNodePath } from './ssh-remote-node-resolution' import { readLocalFullVersion, computeRemoteRelayDir, isRelayAlreadyInstalled, - acquireInstallLock, finalizeInstall, abandonInstall, gcOldRelayVersions } from './ssh-relay-versioned-install' +import { acquireInstallLock } from './ssh-relay-install-lock' import { tryAcquireRelayRepairLock } from './ssh-relay-repair-lock' import { NATIVE_DEPS_COMMAND_TIMEOUT_MS, RELAY_DEPLOY_TIMEOUT_MS } from './ssh-relay-deploy-timing' -import { shellEscape } from './ssh-connection-utils' +import { createSshOperationAbortError, shellEscape } from './ssh-connection-utils' import { probeBuildToolchain, formatMissingToolchainError, @@ -99,15 +104,25 @@ export async function deployAndLaunchRelay( relayInstanceId?: string ): Promise { let timeoutHandle: ReturnType + // Why: Promise.race does not cancel its loser. Stop a contended install-lock + // waiter so it cannot acquire and mutate the relay after this call times out. + const deployAbortController = new AbortController() const timeoutPromise = new Promise((_resolve, reject) => { timeoutHandle = setTimeout(() => { + deployAbortController.abort() reject(new Error(`Relay deployment timed out after ${RELAY_DEPLOY_TIMEOUT_MS / 1000}s`)) }, RELAY_DEPLOY_TIMEOUT_MS) }) try { return await Promise.race([ - deployAndLaunchRelayInner(conn, onProgress, graceTimeSeconds, relayInstanceId), + deployAndLaunchRelayInner( + conn, + onProgress, + graceTimeSeconds, + relayInstanceId, + deployAbortController.signal + ), timeoutPromise ]) } finally { @@ -170,22 +185,29 @@ type RelayBootstrapState = { async function resolveRelayBootstrapStateSequentially( conn: SshConnection, hostPlatform: RemoteHostPlatform, - fullVersion: string + fullVersion: string, + signal?: AbortSignal ): Promise { - const installState = await resolveRemoteInstallState(conn, hostPlatform, fullVersion) - const nodePath = await resolveRemoteNodePath(conn, hostPlatform) + const installState = await resolveRemoteInstallState(conn, hostPlatform, fullVersion, { signal }) + const nodePath = await resolveRemoteNodePath(conn, hostPlatform, { signal }) return { ...installState, nodePath } } async function resolveRelayBootstrapState( conn: SshConnection, hostPlatform: RemoteHostPlatform, - fullVersion: string + fullVersion: string, + signal?: AbortSignal ): Promise { if (!conn.canRunConcurrentExecCommands()) { - return resolveRelayBootstrapStateSequentially(conn, hostPlatform, fullVersion) + return resolveRelayBootstrapStateSequentially(conn, hostPlatform, fullVersion, signal) } const abortController = new AbortController() + const abortForDeploy = (): void => abortController.abort() + signal?.addEventListener('abort', abortForDeploy, { once: true }) + if (signal?.aborted) { + abortForDeploy() + } const installStatePromise = resolveRemoteInstallState(conn, hostPlatform, fullVersion, { rethrowSessionLimitErrors: true, signal: abortController.signal @@ -196,10 +218,12 @@ async function resolveRelayBootstrapState( }) try { const [installState, nodePath] = await Promise.all([installStatePromise, nodePathPromise]) + signal?.throwIfAborted() return { ...installState, nodePath } } catch (err) { abortController.abort() const settled = await Promise.allSettled([installStatePromise, nodePathPromise]) + signal?.throwIfAborted() if (!isSshSessionLimitError(err)) { throw err } @@ -215,7 +239,9 @@ async function resolveRelayBootstrapState( console.warn( '[ssh-relay] Concurrent bootstrap probes hit the remote SSH session limit; retrying sequentially.' ) - return resolveRelayBootstrapStateSequentially(conn, hostPlatform, fullVersion) + return resolveRelayBootstrapStateSequentially(conn, hostPlatform, fullVersion, signal) + } finally { + signal?.removeEventListener('abort', abortForDeploy) } } @@ -232,11 +258,12 @@ async function deployAndLaunchRelayInner( conn: SshConnection, onProgress?: (status: string) => void, graceTimeSeconds?: number, - relayInstanceId?: string + relayInstanceId?: string, + deploySignal?: AbortSignal ): Promise { onProgress?.('Detecting remote platform...') console.log('[ssh-relay] Detecting remote platform...') - const hostPlatform = await detectRemoteHostPlatform(conn) + const hostPlatform = await detectRemoteHostPlatform(conn, { signal: deploySignal }) if (!hostPlatform) { throw new Error( 'Unsupported remote platform. Orca relay supports: linux-x64, linux-arm64, darwin-x64, darwin-arm64, win32-x64, win32-arm64.' @@ -266,47 +293,70 @@ async function deployAndLaunchRelayInner( // fail-fast; remotes that reject overlapping session channels retry the old // sequential order so restrictive SSH servers keep working. const { remoteHome, remoteRelayDir, alreadyInstalled, nodePath } = - await resolveRelayBootstrapState(conn, hostPlatform, fullVersion) + await resolveRelayBootstrapState(conn, hostPlatform, fullVersion, deploySignal) console.log(`[ssh-relay] Remote dir: ${remoteRelayDir}`) console.log(`[ssh-relay] Already installed at ${fullVersion}: ${alreadyInstalled}`) if (alreadyInstalled) { - await repairInstalledNativeDeps(conn, remoteRelayDir, platform, hostPlatform, nodePath) + await repairInstalledNativeDeps( + conn, + remoteRelayDir, + platform, + hostPlatform, + nodePath, + deploySignal + ) + deploySignal?.throwIfAborted() } else { // Why: serialize concurrent first-installs of the same version against // each other via an atomic mkdir lock. The losing caller polls and either // re-checks `alreadyInstalled` (now true) or steals a stale lock. - await acquireInstallLock(conn, remoteRelayDir, hostPlatform) + await acquireInstallLock(conn, remoteRelayDir, hostPlatform, { signal: deploySignal }) try { // Re-probe after acquiring the lock — a sibling installer may have // finished while we were waiting. - if (!(await isRelayAlreadyInstalled(conn, remoteRelayDir, hostPlatform))) { + if ( + !(await isRelayAlreadyInstalled(conn, remoteRelayDir, hostPlatform, { + signal: deploySignal + })) + ) { onProgress?.('Uploading relay...') console.log('[ssh-relay] Uploading relay...') - await uploadRelay(conn, platform, remoteRelayDir, fullVersion, hostPlatform) + await uploadRelay(conn, platform, remoteRelayDir, fullVersion, hostPlatform, deploySignal) console.log('[ssh-relay] Upload complete') onProgress?.('Installing native dependencies...') console.log('[ssh-relay] Installing native dependencies...') - await installNativeDeps(conn, remoteRelayDir, platform, hostPlatform, nodePath) + await installNativeDeps( + conn, + remoteRelayDir, + platform, + hostPlatform, + nodePath, + deploySignal + ) console.log('[ssh-relay] Native deps installed') // Why: write `.install-complete` BEFORE releasing the lock so a // sibling never observes the dir as "complete but locked", which // would lead GC to skip a recoverable dir indefinitely. - await finalizeInstall(conn, remoteRelayDir, hostPlatform) + await finalizeInstall(conn, remoteRelayDir, hostPlatform, { signal: deploySignal }) } else { await abandonInstall(conn, remoteRelayDir, hostPlatform) } } catch (err) { // Why: leave a partial install dir in place (no `.install-complete`) // so the next deploy detects the partial and re-runs upload + install. - // Just release the lock so a concurrent caller can retry. - await abandonInstall(conn, remoteRelayDir, hostPlatform) + // Keep the lock if remote command termination was not confirmed; stale + // recovery is safer than overlapping a still-running npm process. + if (!isUnconfirmedSshCommandTermination(err)) { + await abandonInstall(conn, remoteRelayDir, hostPlatform) + } throw err } } + deploySignal?.throwIfAborted() onProgress?.('Starting relay...') console.log('[ssh-relay] Launching relay...') const launched = await launchRelay( @@ -315,7 +365,8 @@ async function deployAndLaunchRelayInner( hostPlatform, nodePath, graceTimeSeconds, - relayInstanceId + relayInstanceId, + deploySignal ) console.log('[ssh-relay] Relay started successfully') @@ -343,7 +394,8 @@ async function uploadRelay( platform: RelayPlatform, remoteDir: string, fullVersion: string, - hostPlatform: RemoteHostPlatform + hostPlatform: RemoteHostPlatform, + signal?: AbortSignal ): Promise { const localRelayDir = getLocalRelayPath(platform) if (!localRelayDir || !existsSync(localRelayDir)) { @@ -354,16 +406,19 @@ async function uploadRelay( } // Create remote directory - await execHostCommand(conn, hostPlatform, makeRemoteDirectoryCommand(hostPlatform, remoteDir)) + await execHostCommand(conn, hostPlatform, makeRemoteDirectoryCommand(hostPlatform, remoteDir), { + signal + }) - await uploadDirectoryForConnection(conn, localRelayDir, remoteDir, hostPlatform) + await uploadDirectoryForConnection(conn, localRelayDir, remoteDir, hostPlatform, signal) // Make the node binary executable if (!isWindowsRemoteHost(hostPlatform)) { await execHostCommand( conn, hostPlatform, - makeRemoteExecutableCommand(hostPlatform, joinRemotePath(hostPlatform, remoteDir, 'node')) + makeRemoteExecutableCommand(hostPlatform, joinRemotePath(hostPlatform, remoteDir, 'node')), + { signal } ) } @@ -374,7 +429,8 @@ async function uploadRelay( conn, hostPlatform, joinRemotePath(hostPlatform, remoteDir, '.version'), - fullVersion + fullVersion, + signal ) } @@ -382,10 +438,11 @@ async function uploadDirectoryForConnection( conn: SshConnection, localRelayDir: string, remoteDir: string, - hostPlatform: RemoteHostPlatform + hostPlatform: RemoteHostPlatform, + signal?: AbortSignal ): Promise { if (typeof conn.uploadDirectory === 'function') { - await conn.uploadDirectory(localRelayDir, remoteDir, { hostPlatform }) + await conn.uploadDirectory(localRelayDir, remoteDir, { hostPlatform, signal }) return } @@ -401,10 +458,11 @@ async function writeRemoteFile( conn: SshConnection, hostPlatform: RemoteHostPlatform, remotePath: string, - contents: string + contents: string, + signal?: AbortSignal ): Promise { if (typeof conn.writeFile === 'function') { - await conn.writeFile(remotePath, contents, { hostPlatform }) + await conn.writeFile(remotePath, contents, { hostPlatform, signal }) return } @@ -441,7 +499,8 @@ async function hasRequiredNativeDeps( conn: SshConnection, remoteDir: string, hostPlatform: RemoteHostPlatform, - nodePath: string + nodePath: string, + signal?: AbortSignal ): Promise { const escapedNode = shellEscape(nodePath) try { @@ -458,9 +517,10 @@ async function hasRequiredNativeDeps( remoteDir, `(${escapedNode} -e ${shellEscape(`${LOAD_RELAY_NATIVE_DEPS}; console.log("ORCA-NATIVE-DEPS-OK")`)} 2>/dev/null || echo MISSING)` ) - const probe = await execHostCommand(conn, hostPlatform, command) + const probe = await execHostCommand(conn, hostPlatform, command, { signal }) return probe.includes('ORCA-NATIVE-DEPS-OK') } catch { + signal?.throwIfAborted() return false } } @@ -470,9 +530,10 @@ async function repairInstalledNativeDeps( remoteDir: string, platform: RelayPlatform, hostPlatform: RemoteHostPlatform, - nodePath: string + nodePath: string, + signal?: AbortSignal ): Promise { - if (await hasRequiredNativeDeps(conn, remoteDir, hostPlatform, nodePath)) { + if (await hasRequiredNativeDeps(conn, remoteDir, hostPlatform, nodePath, signal)) { return } @@ -482,23 +543,25 @@ async function repairInstalledNativeDeps( // this path used require.resolve (which passed for a present-but-unloadable // binding), so a fatal repair here would be a straight regression. console.warn(`[ssh-relay] Repairing missing native deps at ${remoteDir}`) - if (!(await tryAcquireRelayRepairLock(conn, remoteDir, hostPlatform))) { + if (!(await tryAcquireRelayRepairLock(conn, remoteDir, hostPlatform, { signal }))) { console.warn(`[ssh-relay] Native-deps repair lock is busy at ${remoteDir}; launching degraded`) return } try { // Why: older complete relay dirs were created before @parcel/watcher was // installed. Re-probe under the lock so only one reconnect mutates the dir. - if (!(await hasRequiredNativeDeps(conn, remoteDir, hostPlatform, nodePath))) { - await installNativeDeps(conn, remoteDir, platform, hostPlatform, nodePath) - await finalizeInstall(conn, remoteDir, hostPlatform) + if (!(await hasRequiredNativeDeps(conn, remoteDir, hostPlatform, nodePath, signal))) { + await installNativeDeps(conn, remoteDir, platform, hostPlatform, nodePath, signal) + await finalizeInstall(conn, remoteDir, hostPlatform, { signal }) } else { await abandonInstall(conn, remoteDir, hostPlatform) } } catch (err) { // Release the lock and fall through to a degraded launch. The next reconnect // retries, so a transient failure (offline registry) self-heals. - await abandonInstall(conn, remoteDir, hostPlatform) + if (!isUnconfirmedSshCommandTermination(err)) { + await abandonInstall(conn, remoteDir, hostPlatform) + } console.warn( `[ssh-relay] Native deps repair failed at ${remoteDir}; launching degraded: ${ err instanceof Error ? err.message : String(err) @@ -520,7 +583,8 @@ async function installNativeDeps( remoteDir: string, platform: RelayPlatform, hostPlatform: RemoteHostPlatform, - nodePath: string + nodePath: string, + signal?: AbortSignal ): Promise { // Why: commandWithNodePath puts node's bin directory in PATH for npm's child processes. // npm install runs node-pty's prebuild script (`node scripts/prebuild.js`) @@ -542,7 +606,8 @@ async function installNativeDeps( conn, hostPlatform, joinRemotePath(hostPlatform, remoteDir, 'package.json'), - pkgJson + pkgJson, + signal ) try { @@ -558,7 +623,7 @@ async function installNativeDeps( RELAY_NATIVE_DEPS ) .map(([dep, version]) => powerShellLiteral(`${dep}@${version}`)) - .join(' ')}` + .join(' ')}; if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }` ) : commandWithNodePath( hostPlatform, @@ -567,9 +632,14 @@ async function installNativeDeps( `npm install --ignore-scripts=false --omit=dev --no-audit --no-fund ${installArgs} 2>&1` ) await execHostCommand(conn, hostPlatform, command, { - timeoutMs: NATIVE_DEPS_COMMAND_TIMEOUT_MS + timeoutMs: NATIVE_DEPS_COMMAND_TIMEOUT_MS, + signal }) } catch (err) { + if (isUnconfirmedSshCommandTermination(err)) { + throw err + } + signal?.throwIfAborted() // Don't write .install-complete on hard fail; reconnect retries on a // partial install. Greppable token so user bug reports paste something // searchable. @@ -582,7 +652,7 @@ async function installNativeDeps( // remote and replace node-gyp's opaque `not found: make` with an actionable // install hint instead of leaking the raw npm output to the user. if (platform.startsWith('linux') && shouldProbeBuildToolchainAfterNativeDepsFailure(msg)) { - const toolchain = await probeBuildToolchain(conn, hostPlatform) + const toolchain = await probeBuildToolchain(conn, hostPlatform, signal) if (toolchain?.toolchainMissing) { throw new Error(formatMissingToolchainError(toolchain, msg)) } @@ -590,25 +660,30 @@ async function installNativeDeps( throw err } - await makeNodePtySpawnHelperExecutable(conn, remoteDir, hostPlatform) + await makeNodePtySpawnHelperExecutable(conn, remoteDir, hostPlatform, signal) - let probe = await probeInstalledNativeDeps(conn, remoteDir, hostPlatform, nodePath) + let probe = await probeInstalledNativeDeps(conn, remoteDir, hostPlatform, nodePath, signal) if (!probe.available) { // Why: npm treats an already-present package as up to date, so enabling // lifecycle scripts on install cannot repair a binding skipped earlier. console.warn(`[ssh-relay] Rebuilding unloadable native deps at ${remoteDir}`) let rebuilt = false try { - await rebuildNativeDeps(conn, remoteDir, hostPlatform, nodePath) + await rebuildNativeDeps(conn, remoteDir, hostPlatform, nodePath, signal) rebuilt = true } catch (err) { + if (isUnconfirmedSshCommandTermination(err)) { + throw err + } + signal?.throwIfAborted() console.warn( `[ssh-relay][NATIVE-DEPS-REBUILD-FAIL] npm rebuild native deps failed at ${remoteDir} (${platform}): ${(err as Error).message}` ) } + signal?.throwIfAborted() if (rebuilt) { - await makeNodePtySpawnHelperExecutable(conn, remoteDir, hostPlatform) - probe = await probeInstalledNativeDeps(conn, remoteDir, hostPlatform, nodePath) + await makeNodePtySpawnHelperExecutable(conn, remoteDir, hostPlatform, signal) + probe = await probeInstalledNativeDeps(conn, remoteDir, hostPlatform, nodePath, signal) } } @@ -625,7 +700,8 @@ async function rebuildNativeDeps( conn: SshConnection, remoteDir: string, hostPlatform: RemoteHostPlatform, - nodePath: string + nodePath: string, + signal?: AbortSignal ): Promise { const depNames = Object.keys(RELAY_NATIVE_DEPS) const command = isWindowsRemoteHost(hostPlatform) @@ -633,7 +709,7 @@ async function rebuildNativeDeps( hostPlatform, nodePath, remoteDir, - `npm rebuild --ignore-scripts=false ${depNames.map(powerShellLiteral).join(' ')}` + `npm rebuild --ignore-scripts=false ${depNames.map(powerShellLiteral).join(' ')}; if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }` ) : commandWithNodePath( hostPlatform, @@ -642,14 +718,16 @@ async function rebuildNativeDeps( `npm rebuild --ignore-scripts=false ${depNames.map(shellEscape).join(' ')} 2>&1` ) await execHostCommand(conn, hostPlatform, command, { - timeoutMs: NATIVE_DEPS_COMMAND_TIMEOUT_MS + timeoutMs: NATIVE_DEPS_COMMAND_TIMEOUT_MS, + signal }) } async function makeNodePtySpawnHelperExecutable( conn: SshConnection, remoteDir: string, - hostPlatform: RemoteHostPlatform + hostPlatform: RemoteHostPlatform, + signal?: AbortSignal ): Promise { if (isWindowsRemoteHost(hostPlatform)) { return @@ -659,7 +737,8 @@ async function makeNodePtySpawnHelperExecutable( await execHostCommand( conn, hostPlatform, - `find ${shellEscape(joinRemotePath(hostPlatform, remoteDir, 'node_modules/node-pty/prebuilds'))} -name spawn-helper -exec chmod +x {} + 2>/dev/null; true` + `find ${shellEscape(joinRemotePath(hostPlatform, remoteDir, 'node_modules/node-pty/prebuilds'))} -name spawn-helper -exec chmod +x {} + 2>/dev/null; true`, + { signal } ) } @@ -667,7 +746,8 @@ async function probeInstalledNativeDeps( conn: SshConnection, remoteDir: string, hostPlatform: RemoteHostPlatform, - nodePath: string + nodePath: string, + signal?: AbortSignal ): Promise<{ available: boolean; output: string; stderr: string }> { // require() catches unloadable installs (wrong arch, missing prebuild, or a // skipped lifecycle script) that require.resolve() and test -d both miss. @@ -688,18 +768,21 @@ async function probeInstalledNativeDeps( remoteDir, `(${shellEscape(nodePath)} -e ${shellEscape(probeJs)} ${shellEscape(PROBE_OK)} 2>${escapedStderr} || echo MISSING)` ) - const probeOutput = await execHostCommand(conn, hostPlatform, probeCommand) + const probeOutput = await execHostCommand(conn, hostPlatform, probeCommand, { signal }) const remoteStderr = probeOutput.includes(PROBE_OK) || isWindowsRemoteHost(hostPlatform) ? '' - : await execHostCommand(conn, hostPlatform, `cat ${escapedStderr} 2>/dev/null; true`).catch( - () => '' - ) - await execHostCommand( - conn, - hostPlatform, - removeRemoteFileCommand(hostPlatform, stderrFile) - ).catch(() => {}) + : await execHostCommand(conn, hostPlatform, `cat ${escapedStderr} 2>/dev/null; true`, { + signal + }).catch(() => '') + signal?.throwIfAborted() + if (!isWindowsRemoteHost(hostPlatform)) { + // The POSIX probe redirects stderr to this file; the Windows probe does not. + await execHostCommand(conn, hostPlatform, removeRemoteFileCommand(hostPlatform, stderrFile), { + signal + }).catch(() => {}) + signal?.throwIfAborted() + } return { available: probeOutput.includes(PROBE_OK), output: probeOutput, @@ -744,7 +827,8 @@ async function launchRelay( hostPlatform: RemoteHostPlatform, nodePath: string, graceTimeSeconds?: number, - relayInstanceId?: string + relayInstanceId?: string, + signal?: AbortSignal ): Promise<{ transport: MultiplexerTransport; nodePath: string; sockPath: string }> { // Why: Phase 1 of the plan requires Node.js on the remote. We use the // system `node` rather than bundling a node binary, keeping the relay @@ -776,21 +860,27 @@ async function launchRelay( conn, hostPlatform, remoteDir, - activePipeMarkerPath + activePipeMarkerPath, + signal )) ?? { sockPath: sockFile, endpointDir } const fallbackEndpoint = buildWindowsRelayFallbackEndpoint(hostPlatform, remoteDir, sockName) - return launchWindowsRelay(conn, hostPlatform, { - remoteDir, - nodePath, - sockPath: activeEndpoint.sockPath, - endpointDir: activeEndpoint.endpointDir, - graceTime, - activePipeMarkerPath, - reconnectFallback: fallbackEndpoint - }) + return launchWindowsRelay( + conn, + hostPlatform, + { + remoteDir, + nodePath, + sockPath: activeEndpoint.sockPath, + endpointDir: activeEndpoint.endpointDir, + graceTime, + activePipeMarkerPath, + reconnectFallback: fallbackEndpoint + }, + signal + ) } // Why: after an app restart a relay may still be running in its grace @@ -800,29 +890,43 @@ async function launchRelay( try { const probeOutput = await execCommand( conn, - `test -S ${shellEscape(sockFile)} && echo ALIVE || echo DEAD` + `test -S ${shellEscape(sockFile)} && echo ALIVE || echo DEAD`, + { signal } ) console.warn(`[ssh-relay] Socket probe result: "${probeOutput.trim()}"`) if (probeOutput.trim() === 'ALIVE') { console.log('[ssh-relay] Existing relay socket found, attempting reconnect...') try { const channel = await conn.exec( - `cd ${escapedDir} && ${escapedNode} relay.js --connect --sock-path ${shellEscape(sockFile)}` + `cd ${escapedDir} && ${escapedNode} relay.js --connect --sock-path ${shellEscape(sockFile)}`, + { signal } ) - const transport = await waitForSentinel(channel) + const transport = await waitForSentinel(channel, signal) console.log('[ssh-relay] Reconnected to existing relay via socket') return { transport, nodePath, sockPath: sockFile } } catch (err) { + signal?.throwIfAborted() console.warn( '[ssh-relay] Socket reconnect failed, launching fresh relay:', err instanceof Error ? err.message : String(err) ) // Why: stale socket from a crashed relay — remove it so the // fresh launch can bind a new socket at the same path. - await execCommand(conn, `rm -f ${shellEscape(sockFile)}`).catch(() => {}) + await execCommand(conn, `rm -f ${shellEscape(sockFile)}`, { signal }).catch( + (cleanupErr) => { + if (isUnconfirmedSshCommandTermination(cleanupErr)) { + throw cleanupErr + } + } + ) + signal?.throwIfAborted() } } - } catch { + } catch (err) { + if (isUnconfirmedSshCommandTermination(err)) { + throw err + } + signal?.throwIfAborted() // Probe failed — fall through to fresh launch } @@ -843,7 +947,7 @@ async function launchRelay( // subsequent logging (it wraps process.stderr/stdout), so the current log // stays at relay.log for the `tail relay.log` diagnostics workflow. const launchCmd = `cd ${escapedDir} && nohup ${escapedNode} relay.js --detached --grace-time ${graceTime} --sock-path ${shellEscape(sockFile)} --log-file ${shellEscape(logFile)} > ${shellEscape(logFile)} 2>&1 {}) launchChannel.on('error', () => {}) launchChannel.stderr.on('data', () => {}) @@ -865,36 +969,39 @@ async function launchRelay( const POLL_TIMEOUT_MS = 10_000 const pollStart = Date.now() let socketReady = false - while (Date.now() - pollStart < POLL_TIMEOUT_MS) { - try { - // Why: node is guaranteed to exist on the remote (we just deployed - // the relay with it). Using it to probe the socket is more portable - // than python3/socat/perl which may not be installed. The socket - // path is passed as argv[1] to avoid shell quoting issues with -e. - const result = await execCommand( - conn, - `${escapedNode} -e 'var s=require("net").connect(process.argv[1]);s.on("connect",function(){s.destroy();process.stdout.write("READY")});s.on("error",function(){process.stdout.write("WAITING")})' ${shellEscape(sockFile)} 2>/dev/null || (test -S ${shellEscape(sockFile)} && echo READY || echo WAITING)` - ) - if (result.trim() === 'READY') { - socketReady = true - break + try { + while (Date.now() - pollStart < POLL_TIMEOUT_MS) { + try { + // Why: node is guaranteed to exist on the remote (we just deployed + // the relay with it). Using it to probe the socket is more portable + // than python3/socat/perl which may not be installed. The socket + // path is passed as argv[1] to avoid shell quoting issues with -e. + const result = await execCommand( + conn, + `${escapedNode} -e 'var s=require("net").connect(process.argv[1]);s.on("connect",function(){s.destroy();process.stdout.write("READY")});s.on("error",function(){process.stdout.write("WAITING")})' ${shellEscape(sockFile)} 2>/dev/null || (test -S ${shellEscape(sockFile)} && echo READY || echo WAITING)`, + { signal } + ) + if (result.trim() === 'READY') { + socketReady = true + break + } + } catch { + signal?.throwIfAborted() + /* exec failed, retry */ } - } catch { - /* exec failed, retry */ + await waitForRelayPoll(POLL_INTERVAL_MS, signal) } - await new Promise((r) => setTimeout(r, POLL_INTERVAL_MS)) + } finally { + launchChannel.close() } - // Why: close the fire-and-forget launch channel now that the relay's - // socket is either ready or the poll timed out. Leaving it open leaks - // an SSH channel per relay restart. - launchChannel.close() - if (!socketReady) { const logOutput = await execCommand( conn, - `tail -20 ${shellEscape(logFile)} 2>/dev/null || echo "(no log)"` + `tail -20 ${shellEscape(logFile)} 2>/dev/null || echo "(no log)"`, + { signal } ).catch(() => '(could not read log)') + signal?.throwIfAborted() throw new Error(`Relay failed to start within ${POLL_TIMEOUT_MS / 1000}s. Log:\n${logOutput}`) } @@ -903,9 +1010,28 @@ async function launchRelay( // stdin/stdout to the relay's Unix socket — same path used for reconnect // after app restart. const channel = await conn.exec( - `cd ${escapedDir} && ${escapedNode} relay.js --connect --sock-path ${shellEscape(sockFile)}` + `cd ${escapedDir} && ${escapedNode} relay.js --connect --sock-path ${shellEscape(sockFile)}`, + { signal } ) - return { transport: await waitForSentinel(channel), nodePath, sockPath: sockFile } + return { transport: await waitForSentinel(channel, signal), nodePath, sockPath: sockFile } +} + +function waitForRelayPoll(delayMs: number, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + const onAbort = (): void => { + clearTimeout(timeout) + signal?.removeEventListener('abort', onAbort) + reject(createSshOperationAbortError()) + } + const timeout = setTimeout(() => { + signal?.removeEventListener('abort', onAbort) + resolve() + }, delayMs) + signal?.addEventListener('abort', onAbort, { once: true }) + if (signal?.aborted) { + onAbort() + } + }) } function buildWindowsRelayFallbackEndpoint( @@ -925,15 +1051,20 @@ async function readWindowsActiveRelayEndpoint( conn: SshConnection, hostPlatform: RemoteHostPlatform, remoteDir: string, - markerPath: string + markerPath: string, + signal?: AbortSignal ): Promise { const output = await execHostCommand( conn, hostPlatform, powerShellCommand( `if (Test-Path -LiteralPath ${powerShellLiteral(markerPath)} -PathType Leaf) { Get-Content -LiteralPath ${powerShellLiteral(markerPath)} -Raw -ErrorAction SilentlyContinue }` - ) - ).catch(() => '') + ), + { signal } + ).catch(() => { + signal?.throwIfAborted() + return '' + }) const sockPath = output.trim() if (!isWindowsRelayPipePath(sockPath)) { return null @@ -948,15 +1079,18 @@ async function rememberWindowsActiveRelayEndpoint( conn: SshConnection, hostPlatform: RemoteHostPlatform, markerPath: string, - sockPath: string + sockPath: string, + signal?: AbortSignal ): Promise { await execHostCommand( conn, hostPlatform, powerShellCommand( `Set-Content -LiteralPath ${powerShellLiteral(markerPath)} -Value ${powerShellLiteral(sockPath)} -NoNewline` - ) + ), + { signal } ).catch((err) => { + signal?.throwIfAborted() // Why: fallback pipe names are deterministic, so losing this marker does // not force the next deploy to orphan an undiscoverable relay. console.warn( @@ -982,17 +1116,19 @@ type WindowsRelayLaunchOptions = { async function launchWindowsRelay( conn: SshConnection, hostPlatform: RemoteHostPlatform, - opts: WindowsRelayLaunchOptions + opts: WindowsRelayLaunchOptions, + signal?: AbortSignal ): Promise<{ transport: MultiplexerTransport; nodePath: string; sockPath: string }> { let launchOpts = opts - if ((await probeWindowsRelayPipe(conn, hostPlatform, opts)) === 'READY') { + if ((await probeWindowsRelayPipe(conn, hostPlatform, opts, signal)) === 'READY') { try { - const transport = await connectWindowsRelay(conn, hostPlatform, opts) + const transport = await connectWindowsRelay(conn, hostPlatform, opts, signal) await rememberWindowsActiveRelayEndpoint( conn, hostPlatform, opts.activePipeMarkerPath, - opts.sockPath + opts.sockPath, + signal ) return { transport, @@ -1000,6 +1136,7 @@ async function launchWindowsRelay( sockPath: opts.sockPath } } catch (err) { + signal?.throwIfAborted() console.warn( '[ssh-relay] Windows named pipe reconnect failed, launching fresh relay:', err instanceof Error ? err.message : String(err) @@ -1017,15 +1154,16 @@ async function launchWindowsRelay( if ( launchOpts !== opts && - (await probeWindowsRelayPipe(conn, hostPlatform, launchOpts)) === 'READY' + (await probeWindowsRelayPipe(conn, hostPlatform, launchOpts, signal)) === 'READY' ) { try { - const transport = await connectWindowsRelay(conn, hostPlatform, launchOpts) + const transport = await connectWindowsRelay(conn, hostPlatform, launchOpts, signal) await rememberWindowsActiveRelayEndpoint( conn, hostPlatform, launchOpts.activePipeMarkerPath, - launchOpts.sockPath + launchOpts.sockPath, + signal ) return { transport, @@ -1033,6 +1171,7 @@ async function launchWindowsRelay( sockPath: launchOpts.sockPath } } catch (err) { + signal?.throwIfAborted() console.warn( '[ssh-relay] Windows fallback pipe reconnect failed, relaunching relay:', err instanceof Error ? err.message : String(err) @@ -1054,20 +1193,29 @@ async function launchWindowsRelay( launchOpts.graceTime, logFile, errFile - ) + ), + { signal } ) const POLL_INTERVAL_MS = 200 const POLL_TIMEOUT_MS = 10_000 if ( - await waitForWindowsRelayPipe(conn, hostPlatform, launchOpts, POLL_TIMEOUT_MS, POLL_INTERVAL_MS) + await waitForWindowsRelayPipe( + conn, + hostPlatform, + launchOpts, + POLL_TIMEOUT_MS, + POLL_INTERVAL_MS, + signal + ) ) { - const transport = await connectWindowsRelay(conn, hostPlatform, launchOpts) + const transport = await connectWindowsRelay(conn, hostPlatform, launchOpts, signal) await rememberWindowsActiveRelayEndpoint( conn, hostPlatform, launchOpts.activePipeMarkerPath, - launchOpts.sockPath + launchOpts.sockPath, + signal ) return { transport, @@ -1079,8 +1227,12 @@ async function launchWindowsRelay( const logOutput = await execHostCommand( conn, hostPlatform, - windowsRelayTailLogCommand(logFile, errFile) - ).catch(() => '(could not read log)') + windowsRelayTailLogCommand(logFile, errFile), + { signal } + ).catch(() => { + signal?.throwIfAborted() + return '(could not read log)' + }) throw new Error(`Relay failed to start within ${POLL_TIMEOUT_MS / 1000}s. Log:\n${logOutput}`) } @@ -1091,13 +1243,14 @@ async function connectWindowsRelay( remoteDir: string nodePath: string sockPath: string - } + }, + signal?: AbortSignal ): Promise { const channel = await conn.exec( windowsRelayConnectCommand(hostPlatform, opts.nodePath, opts.remoteDir, opts.sockPath), - { wrapCommand: false } + { wrapCommand: false, signal } ) - return waitForSentinel(channel) + return waitForSentinel(channel, signal) } function windowsRelayConnectCommand( @@ -1164,12 +1317,14 @@ async function probeWindowsRelayPipe( remoteDir: string nodePath: string sockPath: string - } + }, + signal?: AbortSignal ): Promise<'READY' | 'WAITING'> { const result = await execHostCommand( conn, hostPlatform, - windowsRelayProbeCommand(hostPlatform, opts.nodePath, opts.remoteDir, opts.sockPath) + windowsRelayProbeCommand(hostPlatform, opts.nodePath, opts.remoteDir, opts.sockPath), + { signal } ) return result.trim() === 'READY' ? 'READY' : 'WAITING' } @@ -1183,7 +1338,8 @@ async function waitForWindowsRelayPipe( sockPath: string }, timeoutMs: number, - intervalMs: number + intervalMs: number, + signal?: AbortSignal ): Promise { try { const result = await execHostCommand( @@ -1192,10 +1348,12 @@ async function waitForWindowsRelayPipe( windowsRelayWaitCommand(hostPlatform, opts.nodePath, opts.remoteDir, opts.sockPath, { timeoutMs, intervalMs - }) + }), + { signal } ) return result.trim() === 'READY' } catch { + signal?.throwIfAborted() return false } } diff --git a/src/main/ssh/ssh-relay-exec-command.ts b/src/main/ssh/ssh-relay-exec-command.ts new file mode 100644 index 00000000000..e4bfd8bc042 --- /dev/null +++ b/src/main/ssh/ssh-relay-exec-command.ts @@ -0,0 +1,156 @@ +import type { SshConnection } from './ssh-connection' +import { createSshOperationAbortError, type SshExecOptions } from './ssh-connection-utils' +import type { SystemSshCommandChannel } from './system-ssh-command' + +const EXEC_TIMEOUT_MS = 30_000 +const COMMAND_CLOSE_GRACE_MS = 5_000 +const MAX_EXEC_OUTPUT_CHARS = 1024 * 1024 + +type ExecCommandOptions = SshExecOptions & { + timeoutMs?: number +} + +type SshCommandTerminationError = Error & { + sshChannelCloseConfirmed: boolean +} + +export function isUnconfirmedSshCommandTermination( + error: unknown +): error is SshCommandTerminationError { + return ( + error instanceof Error && + (error as Partial).sshChannelCloseConfirmed === false + ) +} + +export async function execCommand( + conn: SshConnection, + command: string, + options?: ExecCommandOptions +): Promise { + const { timeoutMs = EXEC_TIMEOUT_MS, ...execOptions } = options ?? {} + const signal = options?.signal + if (signal?.aborted) { + throw createSshOperationAbortError() + } + // Why: reconnect/disconnect can flip the connection back to ssh2 before a + // killed local OpenSSH child emits close; the channel's transport is immutable. + const openedWithSystemSsh = conn.usesSystemSshTransport?.() === true + const channel = await conn.exec(command, execOptions) + return new Promise((resolve, reject) => { + let stdout = '' + let stderr = '' + let settled = false + let terminationError: SshCommandTerminationError | null = null + let closeGraceTimer: ReturnType | null = null + + const cleanup = (): void => { + clearTimeout(timeout) + if (closeGraceTimer) { + clearTimeout(closeGraceTimer) + } + signal?.removeEventListener('abort', onAbort) + channel.off('error', fail) + channel.stderr.off('error', fail) + channel.off('data', onStdoutData) + channel.stderr.off('data', onStderrData) + channel.off('close', onClose) + } + const settle = (fn: typeof resolve | typeof reject, val: string | Error): void => { + if (settled) { + return + } + settled = true + cleanup() + fn(val as never) + } + const guardUnconfirmedTeardown = (): void => { + const swallowLateError = (): void => {} + const cleanupGuards = (): void => { + channel.off('error', swallowLateError) + channel.stderr.off('error', swallowLateError) + channel.off('close', cleanupGuards) + } + channel.on('error', swallowLateError) + channel.stderr.on('error', swallowLateError) + channel.once('close', cleanupGuards) + // Why: an unconfirmed close can arrive after the caller's bounded wait; + // keep draining discarded streams so ssh2 can finish CHANNEL_CLOSE. + channel.resume() + channel.stderr.resume() + } + // Why: sshd counts the session against MaxSessions until CHANNEL_CLOSE + // completes. Settling on abort before the channel actually closes lets the + // concurrent-bootstrap sequential fallback reissue an exec while the slot + // is still held, so it gets refused again. Close and settle from onClose. + const requestTermination = (error: Error): void => { + if (terminationError) { + return + } + terminationError = Object.assign(error, { sshChannelCloseConfirmed: false }) + clearTimeout(timeout) + // Why: callers must not release an install lock while its remote npm + // process can still mutate node_modules. Prefer confirmed channel close, + // but bound a broken transport's teardown wait. + closeGraceTimer = setTimeout(() => { + guardUnconfirmedTeardown() + settle(reject, error) + }, COMMAND_CLOSE_GRACE_MS) + channel.close() + } + const fail = (err: Error): void => requestTermination(err) + const onAbort = (): void => requestTermination(createSshOperationAbortError()) + const onStdoutData = (data: Buffer): void => { + stdout = appendExecOutputTail(stdout, data.toString('utf-8')) + } + const onStderrData = (data: Buffer): void => { + stderr = appendExecOutputTail(stderr, data.toString('utf-8')) + } + const onClose = (code: number): void => { + if ( + !terminationError && + openedWithSystemSsh && + (channel as SystemSshCommandChannel)._closeRequested + ) { + terminationError = Object.assign(createSshOperationAbortError(), { + sshChannelCloseConfirmed: false + }) + } + if (terminationError) { + // Why: a system-SSH channel closes when the local OpenSSH child exits; + // that does not prove the remote command stopped, especially with a ControlMaster. + if (!openedWithSystemSsh) { + terminationError.sshChannelCloseConfirmed = true + } + settle(reject, terminationError) + } else if (code !== 0) { + // Why: on the system-ssh transport channel.stderr carries local OpenSSH + // client noise; preferring it masks the real failure in stdout (2>&1). + const output = [stderr.trim(), stdout.trim()].filter(Boolean).join('\n') + settle(reject, new Error(`Command "${command}" failed (exit ${code}): ${output}`)) + } else { + settle(resolve, stdout) + } + } + const timeout = setTimeout(() => { + requestTermination(new Error(`Command "${command}" timed out after ${timeoutMs / 1000}s`)) + }, timeoutMs) + + // Why: remote reboot tears down exec channels with stream errors. Without + // scoped listeners, Node treats those as uncaught exceptions. + signal?.addEventListener('abort', onAbort, { once: true }) + channel.on('error', fail) + channel.stderr.on('error', fail) + channel.on('data', onStdoutData) + channel.stderr.on('data', onStderrData) + channel.on('close', onClose) + if (signal?.aborted) { + onAbort() + } + }) +} + +function appendExecOutputTail(existing: string, chunk: string): string { + const combined = existing + chunk + return combined.length > MAX_EXEC_OUTPUT_CHARS ? combined.slice(-MAX_EXEC_OUTPUT_CHARS) : combined +} diff --git a/src/main/ssh/ssh-relay-install-lock-commands.ts b/src/main/ssh/ssh-relay-install-lock-commands.ts index dd341161ecc..20ca1b7ec53 100644 --- a/src/main/ssh/ssh-relay-install-lock-commands.ts +++ b/src/main/ssh/ssh-relay-install-lock-commands.ts @@ -2,8 +2,6 @@ import { shellEscape } from './ssh-connection-utils' import { powerShellCommand, powerShellLiteral } from './ssh-remote-powershell' import { isWindowsRemoteHost, type RemoteHostPlatform } from './ssh-remote-platform' -const STALE_STEAL_LOCK_SECONDS = 120 - export function acquireInstallLockParentCommand( host: RemoteHostPlatform, remoteRelayDir: string @@ -57,25 +55,24 @@ function posixStealInstallLockCommand(lockDir: string, staleAfterSeconds: number return [ `${posixLockIdentityAssignment(lockDir, 'lock_key')} && mtime=\${lock_key%%:*} && now=$(date +%s) && age=$((now - mtime)) || age=0;`, `if [ "\${age:-0}" -le ${staleAfterSeconds} ] 2>/dev/null; then echo BUSY; else`, - `steal=${escapedStealLockPrefix}.$mtime;`, + `steal_root=${escapedStealLockPrefix};`, + 'steal_generation=0;', + 'steal="$steal_root.$steal_generation";', 'owns_steal=0;', - 'owned_steal=;', 'lock_tombstone=;', - 'attempts=0;', - `while [ "$attempts" -lt 8 ] && [ "$owns_steal" != 1 ]; do`, - 'if mkdir "$steal" 2>/dev/null; then owns_steal=1; owned_steal=$steal; break; fi;', - `steal_key=$(stat -c %Y:%i "$steal" 2>/dev/null || stat -f %m:%i "$steal" 2>/dev/null) && steal_mtime=\${steal_key%%:*} && steal_now=$(date +%s) && steal_age=$((steal_now - steal_mtime)) || break;`, - `if [ "\${steal_age:-0}" -le ${STALE_STEAL_LOCK_SECONDS} ] 2>/dev/null; then break; fi;`, - `steal_token=$(printf '%s' "$steal_key" | tr ':/' '__');`, - `steal="$steal.next.$steal_token";`, - 'attempts=$((attempts + 1));', + 'while [ "$owns_steal" != 1 ]; do', + 'if mkdir "$steal" 2>/dev/null; then owns_steal=1; break; fi;', + `steal_mtime=$(stat -c %Y "$steal" 2>/dev/null || stat -f %m "$steal" 2>/dev/null) && steal_now=$(date +%s) && steal_age=$((steal_now - steal_mtime)) || break;`, + 'if [ "${steal_age:-0}" -le 120 ] 2>/dev/null; then break; fi;', + 'steal_generation=$((steal_generation + 1));', + 'steal="$steal_root.$steal_generation";', 'done;', 'if [ "$owns_steal" = 1 ]; then', - `trap 'rmdir "$owned_steal" 2>/dev/null || true; rm -rf "$lock_tombstone" 2>/dev/null || true' EXIT;`, + `trap 'rm -rf "$steal_root".* 2>/dev/null || true; rm -rf "$lock_tombstone" 2>/dev/null || true' EXIT;`, `${posixLockIdentityAssignment(lockDir, 'current_key')} && current_mtime=\${current_key%%:*} && current_now=$(date +%s) && current_age=$((current_now - current_mtime)) || current_age=0;`, `if [ "$current_key" = "$lock_key" ] && [ "\${current_age:-0}" -gt ${staleAfterSeconds} ] 2>/dev/null; then`, `lock_tombstone=${escapedLockDir}.tombstone.$$.$(date +%s);`, - `if mv ${escapedLockDir} "$lock_tombstone" 2>/dev/null; then mkdir ${escapedLockDir} 2>&1 && echo OK || echo BUSY; else echo BUSY; fi;`, + `if [ ! -e "$lock_tombstone" ] && mv ${escapedLockDir} "$lock_tombstone" 2>/dev/null; then mkdir ${escapedLockDir} 2>&1 && echo OK || echo BUSY; else echo BUSY; fi;`, 'else echo BUSY; fi;', 'else echo BUSY; fi; fi' ].join(' ') @@ -91,25 +88,25 @@ function windowsStealInstallLockCommand(lockDir: string, staleAfterSeconds: numb '$lockIdentity = "${mtime}:$($item.CreationTimeUtc.Ticks)"', '$now = [DateTimeOffset]::UtcNow.ToUnixTimeSeconds()', `if (($now - $mtime) -le ${staleAfterSeconds}) { 'BUSY' } else {`, - '$steal = "$lock.steal.$mtime"', + '$stealRoot = "$lock.steal"', + '$stealGeneration = 0', + '$steal = "$stealRoot.$stealGeneration"', '$ownsSteal = $false', - '$ownedSteal = $null', '$lockTombstone = $null', 'try {', - 'for ($attempt = 0; ($attempt -lt 8) -and (-not $ownsSteal); $attempt++) {', + 'while (-not $ownsSteal) {', 'try {', '$null = New-Item -ItemType Directory -Path $steal -ErrorAction Stop', '$ownsSteal = $true', - '$ownedSteal = $steal', 'break', '} catch {', 'try {', '$stealItem = Get-Item -LiteralPath $steal -ErrorAction Stop', '$stealMtime = ([DateTimeOffset]$stealItem.LastWriteTimeUtc).ToUnixTimeSeconds()', '$stealAge = [DateTimeOffset]::UtcNow.ToUnixTimeSeconds() - $stealMtime', - `if ($stealAge -le ${STALE_STEAL_LOCK_SECONDS}) { break }`, - '$stealIdentity = "$stealMtime-$($stealItem.CreationTimeUtc.Ticks)"', - '$steal = "$steal.next.$stealIdentity"', + 'if ($stealAge -le 120) { break }', + '$stealGeneration++', + '$steal = "$stealRoot.$stealGeneration"', '} catch { break }', '}', '}', @@ -124,7 +121,14 @@ function windowsStealInstallLockCommand(lockDir: string, staleAfterSeconds: numb "try { $null = New-Item -ItemType Directory -Path $lock -ErrorAction Stop; 'OK' } catch { 'BUSY' }", "} else { 'BUSY' }", '}', - "} catch { 'BUSY' } finally { if ($null -ne $ownedSteal) { Remove-Item -LiteralPath $ownedSteal -Recurse -Force -ErrorAction SilentlyContinue }; if ($null -ne $lockTombstone) { Remove-Item -LiteralPath $lockTombstone -Recurse -Force -ErrorAction SilentlyContinue } }", + "} catch { 'BUSY' } finally {", + 'if ($ownsSteal) {', + '$stealParent = Split-Path -Parent $stealRoot', + '$stealLeaf = Split-Path -Leaf $stealRoot', + 'Get-ChildItem -LiteralPath $stealParent -Force -ErrorAction SilentlyContinue | Where-Object { $_.Name.StartsWith($stealLeaf + ".", [StringComparison]::Ordinal) } | Remove-Item -Recurse -Force -ErrorAction SilentlyContinue', + '}', + 'if ($null -ne $lockTombstone) { Remove-Item -LiteralPath $lockTombstone -Recurse -Force -ErrorAction SilentlyContinue }', + '}', '}', "} catch { 'BUSY' }" ].join('; ') diff --git a/src/main/ssh/ssh-relay-install-lock.ts b/src/main/ssh/ssh-relay-install-lock.ts new file mode 100644 index 00000000000..47dab9d00e1 --- /dev/null +++ b/src/main/ssh/ssh-relay-install-lock.ts @@ -0,0 +1,116 @@ +import type { SshConnection } from './ssh-connection' +import { execCommand } from './ssh-relay-deploy-helpers' +import { RELAY_DEPLOY_TIMEOUT_MS } from './ssh-relay-deploy-timing' +import { + acquireInstallLockParentCommand, + tryCreateInstallLockCommand, + tryStealInstallLockCommand +} from './ssh-relay-install-lock-commands' +import { + getRemoteHostPlatform, + joinRemotePath, + type RemoteHostPlatform +} from './ssh-remote-platform' + +export const RELAY_INSTALL_LOCK_NAME = '.install-lock' + +const INSTALL_LOCK_POLL_MS = 1_000 +// Why: a fresh lock can cross the stale threshold during our bounded wait. +// Recheck infrequently so it becomes recoverable without adding an exec per poll. +const INSTALL_LOCK_STALE_RECHECK_MS = 60_000 +// Why: the lock holder can legitimately use the full deploy bound for upload, +// native install/rebuild, probes, and finalization. A concurrent first install +// must not fail earlier; completed-relay repair uses a separate one-shot path. +const INSTALL_LOCK_TIMEOUT_MS = RELAY_DEPLOY_TIMEOUT_MS +// Why: native-deps repair can keep running after the deploy backstop wins its +// Promise.race, so stale takeover must leave room for that bounded work. +export const INSTALL_LOCK_STALE_MS = 20 * 60_000 +export const INSTALL_LOCK_STALE_SECONDS = INSTALL_LOCK_STALE_MS / 1000 +const DEFAULT_REMOTE_HOST = getRemoteHostPlatform('linux-x64') + +function execHostCommand( + conn: SshConnection, + host: RemoteHostPlatform, + command: string, + options?: { signal?: AbortSignal } +): Promise { + return execCommand(conn, command, { + wrapCommand: host.commandDialect !== 'powershell', + signal: options?.signal + }) +} + +/** + * Acquire the per-version install lock via atomic `mkdir`. + * Why: `mkdir` gives every supported remote shell one winner without polling state locally. + */ +export async function acquireInstallLock( + conn: SshConnection, + remoteRelayDir: string, + host: RemoteHostPlatform = DEFAULT_REMOTE_HOST, + options?: { signal?: AbortSignal } +): Promise { + const lockDir = joinRemotePath(host, remoteRelayDir, RELAY_INSTALL_LOCK_NAME) + await execHostCommand(conn, host, acquireInstallLockParentCommand(host, remoteRelayDir), { + signal: options?.signal + }) + + const start = Date.now() + let lastStaleCheckAt = Number.NEGATIVE_INFINITY + while (true) { + try { + const result = await execHostCommand(conn, host, tryCreateInstallLockCommand(host, lockDir), { + signal: options?.signal + }) + if (result.trim().endsWith('OK')) { + return + } + } catch { + options?.signal?.throwIfAborted() + // A failed mkdir is lock contention; keep the connection-specific error + // out of the user path until the bounded wait expires. + } + if (Date.now() - lastStaleCheckAt >= INSTALL_LOCK_STALE_RECHECK_MS) { + lastStaleCheckAt = Date.now() + // Why: recover an already-stale lock immediately, then keep checking in + // case a fresh holder crosses the stale threshold while we are waiting. + const steal = await execHostCommand( + conn, + host, + tryStealInstallLockCommand(host, lockDir, INSTALL_LOCK_STALE_SECONDS), + { signal: options?.signal } + ).catch(() => 'BUSY') + options?.signal?.throwIfAborted() + if (steal.trim().endsWith('OK')) { + console.warn(`[ssh-relay] Stealing stale install lock at ${lockDir}`) + return + } + } + if (Date.now() - start >= INSTALL_LOCK_TIMEOUT_MS) { + throw new Error( + `Could not acquire relay install lock at ${lockDir} after ${ + INSTALL_LOCK_TIMEOUT_MS / 1000 + }s; another install is still in progress.` + ) + } + await waitForInstallLockPoll(options?.signal) + } +} + +function waitForInstallLockPoll(signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + const handleAbort = (): void => { + clearTimeout(timeout) + signal?.removeEventListener('abort', handleAbort) + reject(signal?.reason) + } + const timeout = setTimeout(() => { + signal?.removeEventListener('abort', handleAbort) + resolve() + }, INSTALL_LOCK_POLL_MS) + signal?.addEventListener('abort', handleAbort, { once: true }) + if (signal?.aborted) { + handleAbort() + } + }) +} diff --git a/src/main/ssh/ssh-relay-native-deps-install.test.ts b/src/main/ssh/ssh-relay-native-deps-install.test.ts index 5dd1b4905ca..49211526f6b 100644 --- a/src/main/ssh/ssh-relay-native-deps-install.test.ts +++ b/src/main/ssh/ssh-relay-native-deps-install.test.ts @@ -30,6 +30,9 @@ vi.mock('./ssh-relay-deploy-helpers', () => ({ onData: vi.fn(), onClose: vi.fn() }), + isUnconfirmedSshCommandTermination: (error: unknown) => + error instanceof Error && + (error as Error & { sshChannelCloseConfirmed?: boolean }).sshChannelCloseConfirmed === false, execCommand: vi.fn() })) @@ -41,12 +44,15 @@ vi.mock('./ssh-relay-versioned-install', () => ({ readLocalFullVersion: vi.fn().mockReturnValue('0.1.0+testhash'), computeRemoteRelayDir: (home: string, v: string) => `${home}/.orca-remote/relay-${v}`, isRelayAlreadyInstalled: vi.fn().mockResolvedValue(false), - acquireInstallLock: vi.fn().mockResolvedValue(undefined), finalizeInstall: vi.fn().mockResolvedValue(undefined), abandonInstall: vi.fn().mockResolvedValue(undefined), gcOldRelayVersions: vi.fn().mockResolvedValue(undefined) })) +vi.mock('./ssh-relay-install-lock', () => ({ + acquireInstallLock: vi.fn().mockResolvedValue(undefined) +})) + vi.mock('./ssh-relay-repair-lock', () => ({ tryAcquireRelayRepairLock: vi.fn().mockResolvedValue(true) })) @@ -56,15 +62,16 @@ vi.mock('./ssh-connection-utils', () => ({ })) import { deployAndLaunchRelay } from './ssh-relay-deploy' -import { execCommand } from './ssh-relay-deploy-helpers' +import { execCommand, uploadDirectory } from './ssh-relay-deploy-helpers' +import { RELAY_DEPLOY_TIMEOUT_MS } from './ssh-relay-deploy-timing' import { parseUnameToRelayPlatform } from './relay-protocol' import { resolveRemoteNodePath } from './ssh-remote-node-resolution' import { - acquireInstallLock, abandonInstall, finalizeInstall, isRelayAlreadyInstalled } from './ssh-relay-versioned-install' +import { acquireInstallLock } from './ssh-relay-install-lock' import { tryAcquireRelayRepairLock } from './ssh-relay-repair-lock' import type { SshConnection } from './ssh-connection' @@ -226,6 +233,7 @@ describe('installNativeDeps (via deployAndLaunchRelay)', () => { // a leaked response. clearAllMocks doesn't drop the queue (it only clears // .mock.calls), so we explicitly mockReset. vi.mocked(execCommand).mockReset() + vi.mocked(uploadDirectory).mockResolvedValue(undefined) sftpCapture.paths.length = 0 for (const k of Object.keys(sftpCapture.contents)) { delete sftpCapture.contents[k] @@ -453,6 +461,45 @@ describe('installNativeDeps (via deployAndLaunchRelay)', () => { expect(vi.mocked(abandonInstall)).toHaveBeenCalledTimes(1) }) + it('aborts an in-progress native install and releases its lock at deploy timeout', async () => { + vi.useFakeTimers() + try { + const conn = makeMockConnection(sftpCapture) + feed([ + '__ORCA_REMOTE_PLATFORM__ Linux x86_64', + '/home/u', + '', // mkdir remoteDir + '' // chmod +x node + ]) + let installSignal: AbortSignal | undefined + vi.mocked(execCommand).mockImplementationOnce((_conn, command, options) => { + expect(command).toContain('npm install') + installSignal = options?.signal + return new Promise((_resolve, reject) => { + installSignal?.addEventListener('abort', () => reject(installSignal?.reason), { + once: true + }) + }) + }) + + const promise = deployAndLaunchRelay(conn).catch((err: Error) => err) + await vi.waitFor(() => expect(installSignal).toBeDefined()) + + await vi.advanceTimersByTimeAsync(RELAY_DEPLOY_TIMEOUT_MS) + + const result = await promise + expect(result).toBeInstanceOf(Error) + expect((result as Error).message).toBe( + `Relay deployment timed out after ${RELAY_DEPLOY_TIMEOUT_MS / 1000}s` + ) + expect(installSignal?.aborted).toBe(true) + expect(vi.mocked(abandonInstall)).toHaveBeenCalledTimes(1) + expect(vi.mocked(finalizeInstall)).not.toHaveBeenCalled() + } finally { + vi.useRealTimers() + } + }) + it('lets a probe SSH-channel failure bubble up rather than silently mapping to MISSING', async () => { const conn = makeMockConnection(sftpCapture) feed( @@ -609,10 +656,8 @@ describe('installNativeDeps (via deployAndLaunchRelay)', () => { '', // mkdir remoteDir '', // npm install native deps 'MISSING\n', // native process exit normalized by PowerShell command - '', // remove probe stderr file '', // npm rebuild native deps 'MISSING\n', // rebuilt native process still cannot load - '', // remove second probe stderr file '', // no persisted active pipe marker 'WAITING', '', // WMI relay launch @@ -638,6 +683,16 @@ describe('installNativeDeps (via deployAndLaunchRelay)', () => { .filter((script) => script.includes('npm install') || script.includes('npm rebuild')) expect(npmScripts).toHaveLength(2) expect(npmScripts.every((script) => script.includes('--ignore-scripts=false'))).toBe(true) + expect( + npmScripts.every((script) => + script.includes('if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }') + ) + ).toBe(true) + expect( + vi + .mocked(execCommand) + .mock.calls.some(([, command]) => command.includes('.npty-probe.stderr')) + ).toBe(false) const warnMessages = warnSpy.mock.calls.map((args) => String(args[0] ?? '')) expect(warnMessages.some((m) => m.includes('[ssh-relay][NPTY-MISSING]'))).toBe(true) @@ -704,6 +759,9 @@ describe('installNativeDeps (via deployAndLaunchRelay)', () => { await deployAndLaunchRelay(conn) expect(vi.mocked(tryAcquireRelayRepairLock)).toHaveBeenCalledTimes(1) + expect(vi.mocked(tryAcquireRelayRepairLock).mock.calls[0]?.[3]?.signal).toBeInstanceOf( + AbortSignal + ) expect(vi.mocked(finalizeInstall)).toHaveBeenCalledTimes(1) const execCalls = vi.mocked(execCommand).mock.calls.map(([, c]) => c) expect( @@ -739,6 +797,157 @@ describe('installNativeDeps (via deployAndLaunchRelay)', () => { expect(warnMessages.some((m) => m.includes('launching degraded'))).toBe(true) }) + it('retains the repair lock when remote command termination is unconfirmed', async () => { + vi.mocked(isRelayAlreadyInstalled).mockResolvedValue(true) + const conn = makeMockConnection(sftpCapture) + vi.mocked(execCommand) + .mockResolvedValueOnce('__ORCA_REMOTE_PLATFORM__ Linux x86_64') + .mockResolvedValueOnce('/home/u') + .mockResolvedValueOnce('MISSING') + .mockResolvedValueOnce('MISSING') + .mockRejectedValueOnce( + Object.assign(new Error('npm termination was not confirmed'), { + sshChannelCloseConfirmed: false + }) + ) + .mockResolvedValueOnce('DEAD') + .mockResolvedValueOnce('READY') + + await deployAndLaunchRelay(conn) + + expect(vi.mocked(abandonInstall)).not.toHaveBeenCalled() + const warnMessages = warnSpy.mock.calls.map((args) => String(args[0] ?? '')) + expect(warnMessages.some((message) => message.includes('launching degraded'))).toBe(true) + }) + + it('retains the first-install lock when an aborted npm install has unconfirmed teardown', async () => { + vi.useFakeTimers() + try { + const conn = makeMockConnection(sftpCapture) + feed([ + '__ORCA_REMOTE_PLATFORM__ Linux x86_64', + '/home/u', + '', // mkdir remoteDir + '' // chmod +x node + ]) + let installSignal: AbortSignal | undefined + vi.mocked(execCommand).mockImplementationOnce((_conn, command, options) => { + expect(command).toContain('npm install') + installSignal = options?.signal + return new Promise((_resolve, reject) => { + installSignal?.addEventListener( + 'abort', + () => { + // Mirrors execCommand's bounded close grace when ssh2 never + // confirms that the remote npm process stopped. + setTimeout( + () => + reject( + Object.assign(new Error('npm teardown remained unconfirmed'), { + sshChannelCloseConfirmed: false + }) + ), + 5_000 + ) + }, + { once: true } + ) + }) + }) + + const deploy = deployAndLaunchRelay(conn).catch((err: Error) => err) + await vi.waitFor(() => expect(installSignal).toBeDefined()) + await vi.advanceTimersByTimeAsync(RELAY_DEPLOY_TIMEOUT_MS) + const result = await deploy + expect(result).toBeInstanceOf(Error) + expect((result as Error).message).toContain('Relay deployment timed out') + await vi.advanceTimersByTimeAsync(5_000) + + expect(vi.mocked(abandonInstall)).not.toHaveBeenCalled() + expect(vi.mocked(finalizeInstall)).not.toHaveBeenCalled() + } finally { + vi.useRealTimers() + } + }) + + it('does not finalize or release a first-install lock after unconfirmed rebuild teardown', async () => { + const conn = makeMockConnection(sftpCapture) + vi.mocked(execCommand) + .mockResolvedValueOnce('__ORCA_REMOTE_PLATFORM__ Linux x86_64') + .mockResolvedValueOnce('/home/u') + .mockResolvedValueOnce('') + .mockResolvedValueOnce('') + .mockResolvedValueOnce('') + .mockResolvedValueOnce('') + .mockResolvedValueOnce('MISSING') + .mockResolvedValueOnce('rebuild diagnostics') + .mockResolvedValueOnce('') + .mockRejectedValueOnce( + Object.assign(new Error('rebuild termination was not confirmed'), { + sshChannelCloseConfirmed: false + }) + ) + + await expect(deployAndLaunchRelay(conn)).rejects.toThrow( + 'rebuild termination was not confirmed' + ) + expect(vi.mocked(finalizeInstall)).not.toHaveBeenCalled() + expect(vi.mocked(abandonInstall)).not.toHaveBeenCalled() + }) + + it('retains the first-install lock when an aborted rebuild has unconfirmed teardown', async () => { + vi.useFakeTimers() + try { + const conn = makeMockConnection(sftpCapture) + feed([ + '__ORCA_REMOTE_PLATFORM__ Linux x86_64', + '/home/u', + '', // mkdir remoteDir + '', // chmod +x node + '', // npm install + '', // chmod prebuilds + 'MISSING', + 'rebuild diagnostics', + '' // remove probe diagnostics + ]) + let rebuildSignal: AbortSignal | undefined + vi.mocked(execCommand).mockImplementationOnce((_conn, command, options) => { + expect(command).toContain('npm rebuild') + rebuildSignal = options?.signal + return new Promise((_resolve, reject) => { + rebuildSignal?.addEventListener( + 'abort', + () => { + setTimeout( + () => + reject( + Object.assign(new Error('rebuild teardown remained unconfirmed'), { + sshChannelCloseConfirmed: false + }) + ), + 5_000 + ) + }, + { once: true } + ) + }) + }) + + const deploy = deployAndLaunchRelay(conn).catch((err: Error) => err) + await vi.waitFor(() => expect(rebuildSignal).toBeDefined()) + await vi.advanceTimersByTimeAsync(RELAY_DEPLOY_TIMEOUT_MS) + const result = await deploy + expect(result).toBeInstanceOf(Error) + expect((result as Error).message).toContain('Relay deployment timed out') + await vi.advanceTimersByTimeAsync(5_000) + + expect(vi.mocked(abandonInstall)).not.toHaveBeenCalled() + expect(vi.mocked(finalizeInstall)).not.toHaveBeenCalled() + } finally { + vi.useRealTimers() + } + }) + it('launches degraded when the repair install lock cannot be acquired', async () => { // Why: lock contention/wedge must not block a completed relay from launching // in degraded mode — repair is best-effort and we hold no lock to release. diff --git a/src/main/ssh/ssh-relay-repair-lock.ts b/src/main/ssh/ssh-relay-repair-lock.ts index ebcac8c09ab..f7cd3e1e8b8 100644 --- a/src/main/ssh/ssh-relay-repair-lock.ts +++ b/src/main/ssh/ssh-relay-repair-lock.ts @@ -11,16 +11,17 @@ import { joinRemotePath, type RemoteHostPlatform } from './ssh-remote-platform' -import { INSTALL_LOCK_STALE_SECONDS, RELAY_INSTALL_LOCK_NAME } from './ssh-relay-versioned-install' +import { INSTALL_LOCK_STALE_SECONDS, RELAY_INSTALL_LOCK_NAME } from './ssh-relay-install-lock' const DEFAULT_REMOTE_HOST = getRemoteHostPlatform('linux-x64') function execHostCommand( conn: SshConnection, host: RemoteHostPlatform, - command: string + command: string, + signal?: AbortSignal ): Promise { - return execCommand(conn, command, { wrapCommand: !isWindowsRemoteHost(host) }) + return execCommand(conn, command, { wrapCommand: !isWindowsRemoteHost(host), signal }) } /** @@ -32,15 +33,22 @@ function execHostCommand( export async function tryAcquireRelayRepairLock( conn: SshConnection, remoteRelayDir: string, - host: RemoteHostPlatform = DEFAULT_REMOTE_HOST + host: RemoteHostPlatform = DEFAULT_REMOTE_HOST, + options?: { signal?: AbortSignal } ): Promise { const lockDir = joinRemotePath(host, remoteRelayDir, RELAY_INSTALL_LOCK_NAME) try { - await execHostCommand(conn, host, acquireInstallLockParentCommand(host, remoteRelayDir)) + await execHostCommand( + conn, + host, + acquireInstallLockParentCommand(host, remoteRelayDir), + options?.signal + ) const firstAttempt = await execHostCommand( conn, host, - tryCreateInstallLockCommand(host, lockDir) + tryCreateInstallLockCommand(host, lockDir), + options?.signal ) if (firstAttempt.trim().endsWith('OK')) { return true @@ -48,7 +56,8 @@ export async function tryAcquireRelayRepairLock( const steal = await execHostCommand( conn, host, - tryStealInstallLockCommand(host, lockDir, INSTALL_LOCK_STALE_SECONDS) + tryStealInstallLockCommand(host, lockDir, INSTALL_LOCK_STALE_SECONDS), + options?.signal ) if (steal.trim().endsWith('OK')) { console.warn(`[ssh-relay] Stealing stale install lock at ${lockDir}`) @@ -56,6 +65,7 @@ export async function tryAcquireRelayRepairLock( } return false } catch { + options?.signal?.throwIfAborted() return false } } diff --git a/src/main/ssh/ssh-relay-versioned-install.test.ts b/src/main/ssh/ssh-relay-versioned-install.test.ts index e7e96ad7c2f..2453b5a8cd7 100644 --- a/src/main/ssh/ssh-relay-versioned-install.test.ts +++ b/src/main/ssh/ssh-relay-versioned-install.test.ts @@ -18,11 +18,11 @@ import { readLocalFullVersion, computeRemoteRelayDir, isRelayAlreadyInstalled, - acquireInstallLock, finalizeInstall, abandonInstall, gcOldRelayVersions } from './ssh-relay-versioned-install' +import { acquireInstallLock } from './ssh-relay-install-lock' import { tryAcquireRelayRepairLock } from './ssh-relay-repair-lock' import { execCommand } from './ssh-relay-deploy-helpers' import { getRemoteHostPlatform } from './ssh-remote-platform' @@ -89,6 +89,18 @@ describe('isRelayAlreadyInstalled', () => { expect(await isRelayAlreadyInstalled(conn, '/r')).toBe(false) }) + it('does not convert an aborted install probe into a missing install', async () => { + const controller = new AbortController() + mockExec.mockImplementationOnce(async () => { + controller.abort() + throw Object.assign(new Error('cancelled'), { name: 'AbortError' }) + }) + + await expect( + isRelayAlreadyInstalled(conn, '/r', undefined, { signal: controller.signal }) + ).rejects.toMatchObject({ name: 'AbortError' }) + }) + it('keeps default probe failures as not installed for SSH session-limit-shaped errors', async () => { const sessionLimitError = Object.assign(new Error('(SSH) Channel open failure: open failed'), { reason: 4 @@ -132,17 +144,7 @@ describe('acquireInstallLock', () => { expect(mockExec).toHaveBeenCalledTimes(2) }) - it('returns immediately when a best-effort repair lock is busy', async () => { - mockExec.mockResolvedValueOnce('').mockResolvedValueOnce('BUSY').mockResolvedValueOnce('BUSY') - - await expect(tryAcquireRelayRepairLock(conn, '/r')).resolves.toBe(false) - - const commands = mockExec.mock.calls.map(([, command]) => command) - expect(commands.some((command) => command.includes('.install-lock.steal'))).toBe(true) - expect(commands.filter((command) => command.startsWith('rm -rf'))).toHaveLength(0) - }) - - it('does not steal a live repair lock that exceeds the deploy backstop', async () => { + it('returns immediately without deleting a live repair lock', async () => { // Why: repair can run npm install plus rebuild under the same lock; a // second reconnect must launch degraded rather than corrupt node_modules. mockExec.mockResolvedValueOnce('').mockResolvedValueOnce('BUSY').mockResolvedValueOnce('BUSY') @@ -150,7 +152,7 @@ describe('acquireInstallLock', () => { await expect(tryAcquireRelayRepairLock(conn, '/r')).resolves.toBe(false) const commands = mockExec.mock.calls.map(([, command]) => command) - expect(commands.some((command) => command.includes('.install-lock.steal'))).toBe(true) + expect(commands.some((command) => command.includes('lock_tombstone'))).toBe(true) expect(commands.filter((command) => command.startsWith('rm -rf'))).toHaveLength(0) }) @@ -160,8 +162,22 @@ describe('acquireInstallLock', () => { await expect(tryAcquireRelayRepairLock(conn, '/r')).resolves.toBe(true) const commands = mockExec.mock.calls.map(([, command]) => command) - expect(commands.some((command) => command.includes('.install-lock.steal'))).toBe(true) - expect(commands.filter((command) => command.includes('.install-lock.steal'))).toHaveLength(1) + expect(commands.some((command) => command.includes('lock_tombstone'))).toBe(true) + expect(commands.filter((command) => command.includes('lock_tombstone'))).toHaveLength(1) + }) + + it('propagates cancellation through best-effort repair lock commands', async () => { + const abortController = new AbortController() + const abortError = Object.assign(new Error('cancelled'), { name: 'AbortError' }) + mockExec.mockRejectedValueOnce(abortError) + + const promise = tryAcquireRelayRepairLock(conn, '/r', undefined, { + signal: abortController.signal + }) + abortController.abort() + + await expect(promise).rejects.toMatchObject({ name: 'AbortError' }) + expect(mockExec.mock.calls[0]?.[2]?.signal).toBe(abortController.signal) }) it('polls until the lock becomes available (concurrent installer wins, then we acquire)', async () => { @@ -176,6 +192,7 @@ describe('acquireInstallLock', () => { .mockResolvedValueOnce('') .mockResolvedValueOnce('BUSY') .mockResolvedValueOnce('BUSY') + .mockResolvedValueOnce('BUSY') .mockResolvedValueOnce('OK') const promise = acquireInstallLock(conn, '/r') @@ -192,15 +209,26 @@ describe('acquireInstallLock', () => { } }) - it('steals a stale lock with a serialized remote steal command', async () => { + it('immediately tries to steal an already-stale lock', async () => { + mockExec.mockResolvedValueOnce('').mockResolvedValueOnce('BUSY').mockResolvedValueOnce('OK') + + await expect(acquireInstallLock(conn, '/r')).resolves.toBeUndefined() + + const cmds = mockExec.mock.calls.map(([, c]) => c) + expect(cmds).toHaveLength(3) + expect(cmds[2]).toContain('lock_tombstone') + }) + + it('retries stale takeover when a fresh lock ages out during the wait', async () => { vi.useFakeTimers({ now: 1_700_000_000_000 }) try { + const recoverableAt = Date.now() + 60_000 mockExec.mockImplementation(async (_conn: unknown, cmd: string) => { if (cmd.startsWith('mkdir -p')) { return '' } - if (cmd.includes('.install-lock.steal')) { - return 'OK' + if (cmd.includes('lock_tombstone')) { + return Date.now() >= recoverableAt ? 'OK' : 'BUSY' } if (cmd.includes('mkdir') && cmd.includes('.install-lock')) { return 'BUSY' @@ -209,12 +237,12 @@ describe('acquireInstallLock', () => { }) const promise = acquireInstallLock(conn, '/r') - // Drive through the full deploy bound so stale recovery fires. - await vi.advanceTimersByTimeAsync(545_000) - await promise + await vi.advanceTimersByTimeAsync(61_000) - const cmds = mockExec.mock.calls.map(([, c]) => c) - expect(cmds.some((c) => c.includes('.install-lock.steal'))).toBe(true) + await expect(promise).resolves.toBeUndefined() + expect(mockExec.mock.calls.filter(([, cmd]) => cmd.includes('lock_tombstone'))).toHaveLength( + 2 + ) } finally { vi.useRealTimers() } @@ -227,7 +255,7 @@ describe('acquireInstallLock', () => { if (cmd.startsWith('mkdir -p')) { return '' } - if (cmd.includes('.install-lock.steal')) { + if (cmd.includes('lock_tombstone')) { return 'BUSY' } if (cmd.includes('mkdir') && cmd.includes('.install-lock')) { @@ -239,14 +267,37 @@ describe('acquireInstallLock', () => { return '' }) - const rejection = expect(acquireInstallLock(conn, '/r')).rejects.toThrow(/not yet stale/i) - await vi.advanceTimersByTimeAsync(545_000) + const rejection = expect(acquireInstallLock(conn, '/r')).rejects.toThrow( + /another install is still in progress/i + ) + await vi.advanceTimersByTimeAsync(905_000) await rejection } finally { vi.useRealTimers() } }) + it('stops polling when the caller aborts the lock wait', async () => { + vi.useFakeTimers() + try { + const abortController = new AbortController() + mockExec.mockResolvedValueOnce('').mockResolvedValue('BUSY') + + const promise = acquireInstallLock(conn, '/r', undefined, { + signal: abortController.signal + }) + await vi.advanceTimersByTimeAsync(1_000) + abortController.abort() + await expect(promise).rejects.toMatchObject({ name: 'AbortError' }) + + const callCountAfterAbort = mockExec.mock.calls.length + await vi.advanceTimersByTimeAsync(10_000) + expect(mockExec).toHaveBeenCalledTimes(callCountAfterAbort) + } finally { + vi.useRealTimers() + } + }) + it('keeps waiting while a slow first installer is within the deploy bound', async () => { vi.useFakeTimers({ now: 1_700_000_000_000 }) try { @@ -265,7 +316,9 @@ describe('acquireInstallLock', () => { await vi.advanceTimersByTimeAsync(501_000) await expect(promise).resolves.toBeUndefined() - expect(mockExec.mock.calls.some(([, cmd]) => cmd.includes('.install-lock.steal'))).toBe(false) + expect( + mockExec.mock.calls.filter(([, cmd]) => cmd.includes('lock_tombstone')).length + ).toBeGreaterThan(1) } finally { vi.useRealTimers() } diff --git a/src/main/ssh/ssh-relay-versioned-install.ts b/src/main/ssh/ssh-relay-versioned-install.ts index 4028bf4233b..88df794cd02 100644 --- a/src/main/ssh/ssh-relay-versioned-install.ts +++ b/src/main/ssh/ssh-relay-versioned-install.ts @@ -15,13 +15,8 @@ import { existsSync, readFileSync } from 'node:fs' import type { SshConnection } from './ssh-connection' import { RELAY_REMOTE_DIR } from './relay-protocol' import { execCommand } from './ssh-relay-deploy-helpers' -import { RELAY_DEPLOY_TIMEOUT_MS } from './ssh-relay-deploy-timing' -import { - acquireInstallLockParentCommand, - lockAgeSecondsCommand, - tryCreateInstallLockCommand, - tryStealInstallLockCommand -} from './ssh-relay-install-lock-commands' +import { lockAgeSecondsCommand } from './ssh-relay-install-lock-commands' +import { INSTALL_LOCK_STALE_MS, RELAY_INSTALL_LOCK_NAME } from './ssh-relay-install-lock' import { listRelayBaseDirsCommand, probeDirectoryExistsCommand, @@ -55,21 +50,7 @@ const RELAY_VERSION_DIR_REGEX = /^relay-(v?\d+\.\d+\.\d+(\+[0-9a-f]+)?)$/ // living on remote disks forever. const LEGACY_RELAY_DIR_REGEX = /^relay-v\d+\.\d+\.\d+$/ -export const RELAY_INSTALL_LOCK_NAME = '.install-lock' const INSTALL_COMPLETE_NAME = '.install-complete' - -const INSTALL_LOCK_POLL_MS = 1_000 -// Why: the lock holder can legitimately use the full deploy bound for upload, -// native install/rebuild, probes, and finalization. A concurrent first install -// must not fail earlier; completed-relay repair uses a separate one-shot path. -const INSTALL_LOCK_TIMEOUT_MS = RELAY_DEPLOY_TIMEOUT_MS -// Why: a stale lock dir from a crashed installer must be recoverable without -// user intervention. After the timeout we check the lock's mtime; if it's -// older than this window the previous installer is assumed dead and we steal -// the lock. Native-deps repair can keep running after the deploy backstop wins -// Promise.race, so leave enough room for its bounded under-lock exec sequence. -const INSTALL_LOCK_STALE_MS = 20 * 60_000 -export const INSTALL_LOCK_STALE_SECONDS = INSTALL_LOCK_STALE_MS / 1000 const DEFAULT_REMOTE_HOST = getRemoteHostPlatform('linux-x64') type RelayInstalledProbeOptions = { @@ -152,6 +133,7 @@ export async function isRelayAlreadyInstalled( ) return probe.trim() === 'OK' } catch (err) { + options?.signal?.throwIfAborted() if (options?.rethrowSessionLimitErrors && isSshSessionLimitError(err)) { throw err } @@ -159,58 +141,7 @@ export async function isRelayAlreadyInstalled( } } -/** - * Acquire the per-version install lock via atomic `mkdir`. Returns when the - * caller owns the lock; throws if the lock could not be acquired within - * INSTALL_LOCK_TIMEOUT_MS even after one stale-lock recovery attempt. - * - * Why mkdir: POSIX `mkdir` is atomic and fails with EEXIST if the dir already - * exists, giving us a free mutex. A second concurrent caller polls and - * eventually either acquires the lock or steals it after the stale window. - */ -export async function acquireInstallLock( - conn: SshConnection, - remoteRelayDir: string, - host: RemoteHostPlatform = DEFAULT_REMOTE_HOST -): Promise { - const lockDir = joinRemotePath(host, remoteRelayDir, RELAY_INSTALL_LOCK_NAME) - // Why: the parent dir may not exist yet on a first install. mkdir -p is - // safe to run multiple times — it's a no-op if the dir already exists. - await execHostCommand(conn, host, acquireInstallLockParentCommand(host, remoteRelayDir)) - - let start = Date.now() - while (true) { - try { - const result = await execHostCommand(conn, host, tryCreateInstallLockCommand(host, lockDir)) - if (result.trim().endsWith('OK')) { - return - } - } catch { - /* mkdir failed with non-zero — fall through to BUSY treatment */ - } - if (Date.now() - start >= INSTALL_LOCK_TIMEOUT_MS) { - // Stale-lock recovery is one remote command so two observers cannot both - // delete a lock that the other one just recreated. - const steal = await execHostCommand( - conn, - host, - tryStealInstallLockCommand(host, lockDir, INSTALL_LOCK_STALE_SECONDS) - ).catch(() => 'BUSY') - if (steal.trim().endsWith('OK')) { - console.warn(`[ssh-relay] Stealing stale install lock at ${lockDir}`) - return - } - throw new Error( - `Could not acquire relay install lock at ${lockDir} after ${ - INSTALL_LOCK_TIMEOUT_MS / 1000 - }s and the lock is not yet stale.` - ) - } - await new Promise((r) => setTimeout(r, INSTALL_LOCK_POLL_MS)) - } -} - -export async function isRelayInstallLockStale( +async function isRelayInstallLockStale( conn: SshConnection, lockDir: string, host: RemoteHostPlatform = DEFAULT_REMOTE_HOST @@ -238,12 +169,18 @@ export async function isRelayInstallLockStale( export async function finalizeInstall( conn: SshConnection, remoteRelayDir: string, - host: RemoteHostPlatform = DEFAULT_REMOTE_HOST + host: RemoteHostPlatform = DEFAULT_REMOTE_HOST, + options?: { signal?: AbortSignal } ): Promise { const sentinel = joinRemotePath(host, remoteRelayDir, INSTALL_COMPLETE_NAME) const lock = joinRemotePath(host, remoteRelayDir, RELAY_INSTALL_LOCK_NAME) - await execHostCommand(conn, host, writeRemoteEmptyFileCommand(host, sentinel)) - await execHostCommand(conn, host, removeRemoteTreeCommand(host, lock)).catch(() => {}) + await execHostCommand(conn, host, writeRemoteEmptyFileCommand(host, sentinel), { + signal: options?.signal + }) + await execHostCommand(conn, host, removeRemoteTreeCommand(host, lock), { + signal: options?.signal + }).catch(() => {}) + options?.signal?.throwIfAborted() } /** diff --git a/src/main/ssh/ssh-remote-commands.test.ts b/src/main/ssh/ssh-remote-commands.test.ts index ae62a46de7c..460fb4651f6 100644 --- a/src/main/ssh/ssh-remote-commands.test.ts +++ b/src/main/ssh/ssh-remote-commands.test.ts @@ -163,7 +163,7 @@ describe('ssh remote command builders', () => { expect(windowsScript).toContain('Write-Output ($now - $mtime)') }) - it('serializes stale install-lock recovery under a sibling steal mutex', () => { + it('serializes stale recovery with unbounded numbered sibling claims', () => { const posixCommand = tryStealInstallLockCommand( posix, '/home/me/.orca-remote/relay/.install-lock', @@ -173,19 +173,20 @@ describe('ssh remote command builders', () => { tryStealInstallLockCommand(windows, 'C:/Users/me/.orca-remote/relay/.install-lock', 20 * 60) ) + expect(posixCommand).toContain('.install-lock') expect(posixCommand).toContain('.install-lock.steal') - expect(posixCommand).toContain('$mtime') + expect(posixCommand).toContain('steal_generation') + expect(posixCommand).toContain('mtime=${lock_key%%:*}') expect(posixCommand).toContain('-gt 1200') expect(posixCommand).toContain('current_mtime') - expect(posixCommand).toContain('steal_age') - expect(posixCommand).toContain('steal="$steal.next.$steal_token"') + expect(posixCommand).toContain('steal_generation + 1') + expect(posixCommand).not.toContain('.next.') expect(posixCommand).toContain('trap') - expect(windowsScript).toContain('$lock.steal.$mtime') + expect(windowsScript).toContain('$lock.steal') + expect(windowsScript).toContain('$stealGeneration++') expect(windowsScript).toContain('-gt 1200') expect(windowsScript).toContain('$currentIdentity -eq $lockIdentity') - expect(windowsScript).toContain('$stealAge -le 120') - expect(windowsScript).toContain('$steal = "$steal.next.$stealIdentity"') - expect(windowsScript).toContain('New-Item -ItemType Directory -Path $steal') + expect(windowsScript).not.toContain('.next.') expect(windowsScript).toContain('finally') }) @@ -208,7 +209,7 @@ describe('ssh remote command builders', () => { }) it.runIf(process.platform !== 'win32')( - 'recovers a stale POSIX steal mutex left by a crashed stealer', + 'recovers and cleans more than eight orphaned numbered steal claims', () => { const root = mkdtempSync(join(tmpdir(), 'orca-install-lock-')) try { @@ -217,17 +218,18 @@ describe('ssh remote command builders', () => { const staleDate = new Date(Date.now() - 60 * 60_000) utimesSync(lockDir, staleDate, staleDate) const lockMtimeSeconds = Math.floor(statSync(lockDir).mtimeMs / 1000) - const stealDir = `${lockDir}.steal.${lockMtimeSeconds}` - mkdirSync(stealDir) - utimesSync(stealDir, staleDate, staleDate) + for (let i = 0; i < 12; i++) { + const orphan = `${lockDir}.steal.${i}` + mkdirSync(orphan) + utimesSync(orphan, staleDate, staleDate) + } const command = tryStealInstallLockCommand(posix, lockDir, 20 * 60) const output = execFileSync('/bin/sh', ['-c', command], { encoding: 'utf8' }) expect(output.trim()).toBe('OK') expect(existsSync(lockDir)).toBe(true) - expect(existsSync(stealDir)).toBe(true) - expect(readdirSync(root).some((name) => name.includes('.next.'))).toBe(false) + expect(readdirSync(root).filter((name) => name.includes('.steal.'))).toHaveLength(0) expect(Math.floor(statSync(lockDir).mtimeMs / 1000)).toBeGreaterThan(lockMtimeSeconds) } finally { rmSync(root, { recursive: true, force: true }) @@ -236,7 +238,7 @@ describe('ssh remote command builders', () => { ) it.runIf(process.platform !== 'win32')( - 'lets only one POSIX stealer take over a preexisting stale steal mutex', + 'lets only one POSIX caller move and recreate a stale install lock', async () => { const root = mkdtempSync(join(tmpdir(), 'orca-install-lock-race-')) try { @@ -244,11 +246,6 @@ describe('ssh remote command builders', () => { mkdirSync(lockDir) const staleDate = new Date(Date.now() - 60 * 60_000) utimesSync(lockDir, staleDate, staleDate) - const lockMtimeSeconds = Math.floor(statSync(lockDir).mtimeMs / 1000) - const stealDir = `${lockDir}.steal.${lockMtimeSeconds}` - mkdirSync(stealDir) - utimesSync(stealDir, staleDate, staleDate) - const command = tryStealInstallLockCommand(posix, lockDir, 20 * 60) const outputs = await Promise.all( Array.from({ length: 64 }, () => runShellCommand(command)) @@ -257,8 +254,6 @@ describe('ssh remote command builders', () => { expect(okCount).toBe(1) expect(existsSync(lockDir)).toBe(true) - expect(existsSync(stealDir)).toBe(true) - expect(readdirSync(root).some((name) => name.includes('.next.'))).toBe(false) expect(readdirSync(root).some((name) => name.includes('.tombstone'))).toBe(false) } finally { rmSync(root, { recursive: true, force: true }) diff --git a/src/main/ssh/ssh-remote-node-resolution.ts b/src/main/ssh/ssh-remote-node-resolution.ts index 01d6a0c7d48..a5041ed158d 100644 --- a/src/main/ssh/ssh-remote-node-resolution.ts +++ b/src/main/ssh/ssh-remote-node-resolution.ts @@ -122,6 +122,7 @@ true if (options?.rethrowSessionLimitErrors && isSshSessionLimitError(err)) { throw err } + throwIfAborted(options) // Fall through to login shell. } return null @@ -167,6 +168,7 @@ async function tryResolveViaLoginShell( if (options?.rethrowSessionLimitErrors && isSshSessionLimitError(err)) { throw err } + throwIfAborted(options) // Fall through. } return null @@ -191,6 +193,7 @@ async function nodeMeetsVersionRequirement( if (options?.rethrowSessionLimitErrors && isSshSessionLimitError(err)) { throw err } + throwIfAborted(options) // Binary missing or fails to run — not usable. return false } @@ -240,6 +243,7 @@ async function resolveRemoteWindowsNodePath( if (options?.rethrowSessionLimitErrors && isSshSessionLimitError(err)) { throw err } + throwIfAborted(options) // Fall through to the shared error below. } @@ -262,6 +266,7 @@ async function windowsNodeMeetsVersionRequirement( if (options?.rethrowSessionLimitErrors && isSshSessionLimitError(err)) { throw err } + throwIfAborted(options) return false } } diff --git a/src/main/ssh/ssh-remote-platform-detection.ts b/src/main/ssh/ssh-remote-platform-detection.ts index f72b5384335..44c31e617e4 100644 --- a/src/main/ssh/ssh-remote-platform-detection.ts +++ b/src/main/ssh/ssh-remote-platform-detection.ts @@ -11,30 +11,38 @@ import { powerShellCommand } from './ssh-remote-powershell' const PLATFORM_PROBE_MARKER = '__ORCA_REMOTE_PLATFORM__' export async function detectRemoteHostPlatform( - conn: SshConnection + conn: SshConnection, + options?: { signal?: AbortSignal } ): Promise { - const unamePlatform = await detectUnamePlatform(conn) + const unamePlatform = await detectUnamePlatform(conn, options?.signal) if (unamePlatform) { return getRemoteHostPlatform(unamePlatform) } - const windowsPlatform = await detectWindowsPlatform(conn) + const windowsPlatform = await detectWindowsPlatform(conn, options?.signal) return windowsPlatform ? getRemoteHostPlatform(windowsPlatform) : null } -async function detectUnamePlatform(conn: SshConnection): Promise { +async function detectUnamePlatform( + conn: SshConnection, + signal?: AbortSignal +): Promise { try { - const output = await execCommand( - conn, - // Why: Remote startup output may omit its trailing newline and must not absorb the marker. - `printf '\\n%s ' '${PLATFORM_PROBE_MARKER}'; uname -sm` - ) + // Why: Remote startup output may omit its trailing newline and must not absorb the marker. + const command = `printf '\\n%s ' '${PLATFORM_PROBE_MARKER}'; uname -sm` + const output = signal + ? await execCommand(conn, command, { signal }) + : await execCommand(conn, command) return parseRemotePlatformOutput(output) } catch { + signal?.throwIfAborted() return null } } -async function detectWindowsPlatform(conn: SshConnection): Promise { +async function detectWindowsPlatform( + conn: SshConnection, + signal?: AbortSignal +): Promise { try { const script = [ '$arch = $env:PROCESSOR_ARCHITECTURE', @@ -43,9 +51,13 @@ async function detectWindowsPlatform(conn: SshConnection): Promise { expect(mockProc.stdin.end).toHaveBeenCalledWith('contents') }) + it('marks a system command channel when local teardown is requested', () => { + const channel = spawnSystemSshCommand(createTarget(), 'npm install') + + channel.close() + + expect(channel._closeRequested).toBe(true) + expect(mockProc.kill).toHaveBeenCalledWith('SIGTERM') + }) + it('removes wrapped process listeners after command close', () => { const proc = createEventedProcess() spawnMock.mockReturnValue(proc) diff --git a/src/main/ssh/system-ssh-command.ts b/src/main/ssh/system-ssh-command.ts index 1f0ce5ed6df..eab96245ba5 100644 --- a/src/main/ssh/system-ssh-command.ts +++ b/src/main/ssh/system-ssh-command.ts @@ -17,6 +17,7 @@ export type SystemSshProcess = { export type SystemSshCommandChannel = ClientChannel & { _process?: ChildProcess + _closeRequested?: boolean } type SystemSshCommandOptions = SshExecOptions & SystemSshBuildArgsOptions @@ -103,12 +104,14 @@ function wrapCommandProcess(proc: ChildProcess): SystemSshCommandChannel { stdin: NodeJS.WritableStream stderr: NodeJS.ReadableStream _process?: ChildProcess + _closeRequested?: boolean close: () => void } mutableChannel.stdin = proc.stdin! mutableChannel.stderr = proc.stderr! mutableChannel._process = proc mutableChannel.close = () => { + mutableChannel._closeRequested = true try { proc.kill('SIGTERM') } catch { diff --git a/src/main/ssh/system-ssh-operation-lifecycle.ts b/src/main/ssh/system-ssh-operation-lifecycle.ts index 76a2ae3fb12..c9c4424325b 100644 --- a/src/main/ssh/system-ssh-operation-lifecycle.ts +++ b/src/main/ssh/system-ssh-operation-lifecycle.ts @@ -100,7 +100,13 @@ export async function awaitWithSystemSshAbort( // children to emit close after we've already signaled them. abortChildren() suppressLateOperationError = true - abortReject?.(createAbortError()) + abortReject?.( + Object.assign(createAbortError(), { + // The child was signaled, but this fast abort path intentionally does + // not wait for process exit; callers holding remote locks must retain them. + sshChannelCloseConfirmed: false + }) + ) } signal.addEventListener('abort', abort, { once: true }) if (signal.aborted) {