Abort failed runtime downloads after cleanup (#26269)

* fix(ssh): abort runtime download requests after failure

* Keep region correction fixtures past initial control grace

---------

Co-authored-by: m4air <m4air@m4airs-MacBook-Air.local>

src/main/ssh/pinned-runtime-materializer-node.test.ts

@@ -1,5 +1,5 @@
 import { createHash } from 'node:crypto'
-import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
+import { mkdir, mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
@@ -19,8 +19,13 @@ function sha256(bytes: Uint8Array): string {
   return createHash('sha256').update(bytes).digest('hex')
 }
 
-function fetcherFor(body: Uint8Array): typeof fetch {
-  return vi.fn<typeof fetch>(async () => new Response(Buffer.from(body), { status: 200 }))
+function fetcherFor(body: Uint8Array, signals: AbortSignal[] = []): typeof fetch {
+  return vi.fn<typeof fetch>(async (_url, options) => {
+    if (options?.signal) {
+      signals.push(options.signal)
+    }
+    return new Response(Buffer.from(body), { status: 200 })
+  })
 }
 
 /** A real .tar.gz laid out like the official one, so extraction runs the host's tar. */
@@ -53,7 +58,8 @@ describe.skipIf(process.platform === 'win32')('pinned Node runtime materializer'
     const archive = await nodeDistArchiveFixture(new TextEncoder().encode('node'))
     Object.assign(NODE_RUNTIME_ASSETS[TARGET], { archiveSha256: sha256(archive) })
     const cacheRoot = join(root, 'cache')
-    const fetcher = fetcherFor(archive)
+    const signals: AbortSignal[] = []
+    const fetcher = fetcherFor(archive, signals)
 
     const cached = await materializeNodeRuntimeArchive(TARGET, cacheRoot, { fetcher })
     expect(cached.endsWith(originalAsset.archive)).toBe(true)
@@ -68,6 +74,8 @@ describe.skipIf(process.platform === 'win32')('pinned Node runtime materializer'
     const repaired = await materializeNodeRuntimeArchive(TARGET, cacheRoot, { fetcher })
     expect(repaired).not.toBe(cached)
     expect(sha256(new Uint8Array(await readFile(repaired)))).toBe(sha256(archive))
+    expect(signals).toHaveLength(2)
+    expect(signals.every((signal) => !signal.aborted)).toBe(true)
   })
 
   it('refuses an archive that does not match the pin before caching it', async () => {
@@ -77,3 +85,69 @@ describe.skipIf(process.platform === 'win32')('pinned Node runtime materializer'
     ).rejects.toThrow('Node archive checksum mismatch')
   })
 })
+
+describe('pinned runtime archive request ownership', () => {
+  it.each(['http-error', 'oversized-declaration', 'staging-collision'] as const)(
+    'retires the owned %s request without cancelling its caller',
+    async (failure) => {
+      const cacheRoot = join(root, 'cache')
+      const archiveRoot = join(cacheRoot, 'node', 'archives')
+      const parent = new AbortController()
+      const signals: AbortSignal[] = []
+      const cancel = vi.fn()
+      const fetcher = vi.fn<typeof fetch>(async (_url, options) => {
+        if (options?.signal) {
+          signals.push(options.signal)
+        }
+        if (failure === 'staging-collision') {
+          const stage = (await readdir(archiveRoot)).find((name) => name.startsWith('.download-'))
+          if (!stage) {
+            throw new Error('Missing download staging directory')
+          }
+          await mkdir(join(archiveRoot, stage, originalAsset.archive))
+        }
+        return new Response(new ReadableStream({ cancel }), {
+          status: failure === 'http-error' ? 404 : 200,
+          statusText: 'Unavailable',
+          headers:
+            failure === 'oversized-declaration'
+              ? { 'content-length': String(200 * 1024 * 1024 + 1) }
+              : {}
+        })
+      })
+      const pending = materializeNodeRuntimeArchive(TARGET, cacheRoot, {
+        fetcher,
+        signal: parent.signal
+      })
+      await (failure === 'staging-collision'
+        ? expect(pending).rejects.toMatchObject({ code: 'EEXIST' })
+        : expect(pending).rejects.toThrow(
+            failure === 'http-error'
+              ? 'Node download failed: 404 Unavailable'
+              : 'Node download exceeded the archive size limit'
+          ))
+      expect(cancel).toHaveBeenCalledOnce()
+      expect(signals).toHaveLength(1)
+      expect(signals[0]?.aborted).toBe(true)
+      expect(parent.signal.aborted).toBe(false)
+      expect(await readdir(archiveRoot)).toEqual([])
+    }
+  )
+
+  it('keeps the caller cancellation reason and removes the incomplete stage', async () => {
+    const cacheRoot = join(root, 'cache')
+    const parent = new AbortController()
+    const reason = new Error('deployment cancelled')
+    const cancel = vi.fn()
+    const fetcher = vi.fn<typeof fetch>(async () => {
+      queueMicrotask(() => parent.abort(reason))
+      return new Response(new ReadableStream({ cancel }))
+    })
+    await expect(
+      materializeNodeRuntimeArchive(TARGET, cacheRoot, { fetcher, signal: parent.signal })
+    ).rejects.toBe(reason)
+    expect(parent.signal.reason).toBe(reason)
+    expect(cancel).toHaveBeenCalledOnce()
+    expect(await readdir(join(cacheRoot, 'node', 'archives'))).toEqual([])
+  })
+})

src/main/ssh/runtime-archive-download.test.ts

@@ -21,6 +21,13 @@ function archive(body: Uint8Array = BODY): PinnedRuntimeArchive {
 
 const ok = (body: Uint8Array = BODY): Response => new Response(Buffer.from(body), { status: 200 })
 
+function requestSignal(options: RequestInit | undefined): AbortSignal {
+  if (!options?.signal) {
+    throw new Error('Expected an owned archive request signal')
+  }
+  return options.signal
+}
+
 /** A body that delivers its first half, then fails as Chromium does when the network drops. */
 function droppedMidway(): Response {
   const stream = new ReadableStream<Uint8Array>({
@@ -52,29 +59,56 @@ describe('downloading a pinned runtime archive over a flaky network', () => {
     ['a server error', () => Promise.resolve(new Response('busy', { status: 503 }))],
     ['a reset partway through the body', () => Promise.resolve(droppedMidway())]
   ])('retries after %s and still verifies the checksum', async (_name, first) => {
-    const fetcher = vi.fn<typeof fetch>().mockImplementationOnce(first).mockResolvedValue(ok())
+    const signals: AbortSignal[] = []
+    let retiredBeforeRetry = false
+    const fetcher = vi
+      .fn<typeof fetch>()
+      .mockImplementationOnce(async (_url, options) => {
+        signals.push(requestSignal(options))
+        return first()
+      })
+      .mockImplementation(async (_url, options) => {
+        retiredBeforeRetry = signals[0]?.aborted === true
+        signals.push(requestSignal(options))
+        return ok()
+      })
     await downloadVerifiedArchive(archive(), destination, fetcher, undefined, NO_DELAY)
     expect(fetcher).toHaveBeenCalledTimes(2)
     expect(new Uint8Array(await readFile(destination))).toEqual(BODY)
+    expect(retiredBeforeRetry).toBe(true)
+    expect(signals[0]?.aborted).toBe(true)
+    expect(signals[1]?.aborted).toBe(false)
   })
 
   it('gives up after its bounded attempts with the last error', async () => {
-    const fetcher = vi.fn<typeof fetch>().mockRejectedValue(new Error('net::ERR_NETWORK_CHANGED'))
+    const signals: AbortSignal[] = []
+    const failure = new Error('net::ERR_NETWORK_CHANGED')
+    const fetcher = vi.fn<typeof fetch>(async (_url, options) => {
+      signals.push(requestSignal(options))
+      throw failure
+    })
     await expect(
       downloadVerifiedArchive(archive(), destination, fetcher, undefined, NO_DELAY)
-    ).rejects.toThrow('net::ERR_NETWORK_CHANGED')
+    ).rejects.toBe(failure)
     expect(fetcher).toHaveBeenCalledTimes(NO_DELAY.length + 1)
+    expect(signals).toHaveLength(NO_DELAY.length + 1)
+    expect(signals.every((signal) => signal.aborted)).toBe(true)
   })
 
   it.each([
     ['a missing asset', () => new Response('gone', { status: 404 })],
     ['a checksum mismatch', () => ok(new Uint8Array([9, 9]))]
   ])('never retries %s', async (_name, response) => {
-    const fetcher = vi.fn<typeof fetch>(async () => response())
+    const signals: AbortSignal[] = []
+    const fetcher = vi.fn<typeof fetch>(async (_url, options) => {
+      signals.push(requestSignal(options))
+      return response()
+    })
     await expect(
       downloadVerifiedArchive(archive(), destination, fetcher, undefined, NO_DELAY)
     ).rejects.toThrow()
     expect(fetcher).toHaveBeenCalledTimes(1)
+    expect(signals[0]?.aborted).toBe(true)
   })
 
   it('stops waiting to retry once the caller cancels', async () => {

src/main/ssh/runtime-archive-download.ts

@@ -151,7 +151,9 @@ async function downloadVerifiedArchiveOnce(
     }
   } catch (error) {
     // Why: the fetch may reject a stalled abort with its own AbortError rather than our reason.
-    throw stall.signal.aborted && !signal?.aborted ? stalled : error
+    const downloadError = stall.signal.aborted && !signal?.aborted ? stalled : error
+    stall.abort()
+    throw downloadError
   } finally {
     clearTimeout(stallTimer)
   }