fix: settle system ssh aborts (#3842)

src/main/ssh/ssh-system-fallback.test.ts

@@ -1,4 +1,5 @@
-import { EventEmitter } from 'events'
+import { EventEmitter } from 'node:events'
+import { PassThrough } from 'node:stream'
 import { describe, expect, it, vi, beforeEach } from 'vitest'
 
 const { existsSyncMock, spawnMock } = vi.hoisted(() => ({
@@ -19,6 +20,7 @@ import {
   findSystemSsh,
   spawnSystemSsh,
   spawnSystemSshCommand,
+  uploadDirectoryViaSystemSsh,
   writeFileViaSystemSsh
 } from './ssh-system-fallback'
 import type { SshTarget } from '../../shared/ssh-types'
@@ -62,6 +64,37 @@ function createEventedProcess(): EventedProcess {
   return proc
 }
 
+function createMockChildProcess(): EventEmitter & {
+  stdin: PassThrough
+  stdout: PassThrough
+  stderr: PassThrough
+  pid: number
+  kill: ReturnType<typeof vi.fn>
+  killed: boolean
+  exitCode: number | null
+} {
+  const child = new EventEmitter() as EventEmitter & {
+    stdin: PassThrough
+    stdout: PassThrough
+    stderr: PassThrough
+    pid: number
+    kill: ReturnType<typeof vi.fn>
+    killed: boolean
+    exitCode: number | null
+  }
+  child.stdin = new PassThrough()
+  child.stdout = new PassThrough()
+  child.stderr = new PassThrough()
+  child.pid = 12345
+  child.killed = false
+  child.exitCode = null
+  child.kill = vi.fn(() => {
+    child.killed = true
+    return true
+  })
+  return child
+}
+
 describe('findSystemSsh', () => {
   beforeEach(() => {
     existsSyncMock.mockReset()
@@ -256,3 +289,60 @@ describe('spawnSystemSsh', () => {
     expect(typeof result.onExit).toBe('function')
   })
 })
+
+describe('system SSH operation aborts', () => {
+  beforeEach(() => {
+    existsSyncMock.mockReset()
+    spawnMock.mockReset()
+    existsSyncMock.mockImplementation((p: string) => p === '/usr/bin/ssh')
+  })
+
+  it('rejects directory uploads when aborted even if child processes do not close', async () => {
+    const tarCreate = createMockChildProcess()
+    const sshExtract = createMockChildProcess()
+    spawnMock.mockReturnValueOnce(tarCreate).mockReturnValueOnce(sshExtract)
+    const controller = new AbortController()
+
+    const uploadPromise = uploadDirectoryViaSystemSsh(
+      createTarget(),
+      '/tmp/local-relay',
+      '/tmp/remote-relay',
+      { signal: controller.signal }
+    )
+    controller.abort()
+
+    const outcome = await Promise.race([
+      uploadPromise.then(
+        () => 'resolved',
+        (error: Error) => error.name
+      ),
+      new Promise<string>((resolve) => setTimeout(() => resolve('pending'), 0))
+    ])
+
+    expect(outcome).toBe('AbortError')
+    expect(tarCreate.kill).toHaveBeenCalledTimes(1)
+    expect(sshExtract.kill).toHaveBeenCalledTimes(1)
+  })
+
+  it('rejects remote file writes when aborted even if ssh never closes', async () => {
+    const sshProcess = createMockChildProcess()
+    spawnMock.mockReturnValueOnce(sshProcess)
+    const controller = new AbortController()
+
+    const writePromise = writeFileViaSystemSsh(createTarget(), '/tmp/remote-file', 'contents', {
+      signal: controller.signal
+    })
+    controller.abort()
+
+    const outcome = await Promise.race([
+      writePromise.then(
+        () => 'resolved',
+        (error: Error) => error.name
+      ),
+      new Promise<string>((resolve) => setTimeout(() => resolve('pending'), 0))
+    ])
+
+    expect(outcome).toBe('AbortError')
+    expect(sshProcess.kill).toHaveBeenCalledTimes(1)
+  })
+})

src/main/ssh/ssh-system-fallback.ts

@@ -109,25 +109,25 @@ export async function uploadDirectoryViaSystemSsh(
     }
   )
 
-  const abort = (): void => {
-    killProcess(tarCreate)
-    killProcess(sshExtract)
-  }
-  options?.signal?.addEventListener('abort', abort, { once: true })
   let tarResult: ProcessResult | null = null
   let sshResult: ProcessResult | null = null
   try {
-    ;[tarResult, sshResult] = await Promise.all([
-      waitForProcess(tarCreate, 'local tar relay upload'),
-      waitForProcess(sshExtract, 'system ssh relay upload'),
-      pipeline(tarCreate.stdout!, sshExtract.stdin!)
-    ]).then(([tar, ssh]) => [tar, ssh])
+    ;[tarResult, sshResult] = await awaitWithSystemSshAbort(
+      options?.signal,
+      () => {
+        killProcess(tarCreate)
+        killProcess(sshExtract)
+      },
+      Promise.all([
+        waitForProcess(tarCreate, 'local tar relay upload'),
+        waitForProcess(sshExtract, 'system ssh relay upload'),
+        pipeline(tarCreate.stdout!, sshExtract.stdin!)
+      ]).then(([tar, ssh]) => [tar, ssh] as const)
+    )
   } catch (err) {
     killProcess(tarCreate)
     killProcess(sshExtract)
     throw err
-  } finally {
-    options?.signal?.removeEventListener('abort', abort)
   }
 
   if (tarResult?.stderr.trim()) {
@@ -146,17 +146,15 @@ export async function writeFileViaSystemSsh(
 ): Promise<void> {
   throwIfAborted(options?.signal)
   const channel = spawnSystemSshCommand(target, `cat > ${shellEscape(remotePath)}`)
-  const abort = (): void => {
-    channel.close()
-  }
-  options?.signal?.addEventListener('abort', abort, { once: true })
-  const closePromise = waitForChannelClose(channel, `write ${remotePath}`)
-  channel.stdin.end(contents)
-  try {
-    await closePromise
-  } finally {
-    options?.signal?.removeEventListener('abort', abort)
+  const closePromise = awaitWithSystemSshAbort(
+    options?.signal,
+    () => channel.close(),
+    waitForChannelClose(channel, `write ${remotePath}`)
+  )
+  if (!options?.signal?.aborted) {
+    channel.stdin.end(contents)
   }
+  await closePromise
 }
 
 export function buildSshArgs(target: SshTarget): string[] {
@@ -368,11 +366,54 @@ function killProcess(proc: ChildProcess): void {
   }
 }
 
+async function awaitWithSystemSshAbort<T>(
+  signal: AbortSignal | undefined,
+  abortChildren: () => void,
+  operation: Promise<T>
+): Promise<T> {
+  if (!signal) {
+    return operation
+  }
+  let abortReject: ((error: Error) => void) | null = null
+  let suppressLateOperationError = false
+  const abortPromise = new Promise<never>((_resolve, reject) => {
+    abortReject = reject
+  })
+  const abort = (): void => {
+    // Why: abort is connection teardown; do not wait for stubborn system ssh/tar
+    // children to emit close after we've already signaled them.
+    abortChildren()
+    suppressLateOperationError = true
+    abortReject?.(createAbortError())
+  }
+  signal.addEventListener('abort', abort, { once: true })
+  if (signal.aborted) {
+    abort()
+  }
+  try {
+    return await Promise.race([
+      operation.catch((error: unknown) => {
+        if (suppressLateOperationError) {
+          return new Promise<never>(() => {})
+        }
+        throw error
+      }),
+      abortPromise
+    ])
+  } finally {
+    signal.removeEventListener('abort', abort)
+  }
+}
+
 function throwIfAborted(signal: AbortSignal | undefined): void {
   if (!signal?.aborted) {
     return
   }
+  throw createAbortError()
+}
+
+function createAbortError(): Error & { name: string } {
   const error = new Error('System SSH operation was cancelled') as Error & { name: string }
   error.name = 'AbortError'
-  throw error
+  return error
 }