fix(device): time out stalled artifact downloads (#240)

Author: riglar

src/utils/device-artifacts.ts

@@ -1,5 +1,5 @@
 import { createWriteStream } from 'node:fs';
-import { mkdir } from 'node:fs/promises';
+import { mkdir, rm } from 'node:fs/promises';
 import * as path from 'node:path';
 import { Readable } from 'node:stream';
 import { pipeline } from 'node:stream/promises';
@@ -113,15 +113,84 @@ export function planArtifactFiles(pages: DeviceArtifactsPage[]): DeviceArtifactF
 
 const CONCURRENCY = 4;
 
+/** Timeouts and retries for one artifact download. */
+export interface DownloadArtifactOptions {
+  /** Give up on an attempt that receives no data for this long. */
+  idleTimeoutMs?: number;
+  /** Give up on an attempt that takes longer than this in total. */
+  totalTimeoutMs?: number;
+  /** How many times to try each file before reporting it failed. */
+  attempts?: number;
+}
+
+const DEFAULT_IDLE_TIMEOUT_MS = 30_000;
+const DEFAULT_TOTAL_TIMEOUT_MS = 10 * 60_000;
+const DEFAULT_ATTEMPTS = 2;
+
+function seconds(ms: number): string {
+  return ms >= 1000 ? `${Math.round(ms / 1000)}s` : `${ms}ms`;
+}
+
 /**
- * Download every file with a link into `dir`, four at a time. Returns the
- * files that could not be fetched, with why, rather than failing the rest.
+ * Fetch one URL to `target`. Aborts when no data arrives for `idleMs`
+ * (including while waiting for the response headers) or when the whole
+ * download takes longer than `totalMs`. A CDN stream can stall without
+ * closing, and without these the command would wait for it forever.
+ */
+async function downloadOnce(url: string, target: string, idleMs: number, totalMs: number): Promise<void> {
+  const controller = new AbortController();
+  const abort = (reason: string) => controller.abort(new Error(reason));
+  let idle = setTimeout(() => abort(`timed out: no data for ${seconds(idleMs)}`), idleMs);
+  const total = setTimeout(() => abort(`timed out: took longer than ${seconds(totalMs)}`), totalMs);
+  const resetIdle = () => {
+    clearTimeout(idle);
+    idle = setTimeout(() => abort(`timed out: no data for ${seconds(idleMs)}`), idleMs);
+  };
+
+  try {
+    const res = await fetch(url, { signal: controller.signal });
+    if (!res.ok || !res.body) throw new Error(`HTTP ${res.status}`);
+    resetIdle();
+    const body = Readable.fromWeb(res.body as Parameters<typeof Readable.fromWeb>[0]);
+    await pipeline(
+      body,
+      async function* (source: AsyncIterable<Buffer>) {
+        for await (const chunk of source) {
+          resetIdle();
+          yield chunk;
+        }
+      },
+      createWriteStream(target),
+      { signal: controller.signal },
+    );
+  } catch (error) {
+    // Report the timeout, not the AbortError it surfaces as.
+    if (controller.signal.aborted && controller.signal.reason instanceof Error) {
+      throw controller.signal.reason;
+    }
+
+    throw error;
+  } finally {
+    clearTimeout(idle);
+    clearTimeout(total);
+  }
+}
+
+/**
+ * Download every file with a link into `dir`, four at a time. Each file is
+ * tried twice (a stalled CDN response usually succeeds on a second
+ * request). Returns the files that could not be fetched, with why, rather
+ * than failing the rest.
  */
 export async function downloadArtifactFiles(
   dir: string,
   files: DeviceArtifactFile[],
   onFile: (file: DeviceArtifactFile) => void = () => {},
+  options: DownloadArtifactOptions = {},
 ): Promise<{ file: string; reason: string }[]> {
+  const idleMs = options.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_MS;
+  const totalMs = options.totalTimeoutMs ?? DEFAULT_TOTAL_TIMEOUT_MS;
+  const attempts = Math.max(1, options.attempts ?? DEFAULT_ATTEMPTS);
   const failed: { file: string; reason: string }[] = [];
   const queue = [...files];
   async function worker(): Promise<void> {
@@ -132,17 +201,24 @@ export async function downloadArtifactFiles(
       }
 
       const target = path.join(dir, next.file);
-      try {
-        await mkdir(path.dirname(target), { recursive: true });
-        const res = await fetch(next.url);
-        if (!res.ok || !res.body) throw new Error(`HTTP ${res.status}`);
-        await pipeline(
-          Readable.fromWeb(res.body as Parameters<typeof Readable.fromWeb>[0]),
-          createWriteStream(target),
-        );
+      let reason = '';
+      for (let attempt = 1; attempt <= attempts; attempt++) {
+        try {
+          await mkdir(path.dirname(target), { recursive: true });
+          await downloadOnce(next.url, target, idleMs, totalMs);
+          reason = '';
+          break;
+        } catch (error) {
+          reason = error instanceof Error ? error.message : String(error);
+        }
+      }
+
+      if (reason) {
+        // Don't leave a half-written file that looks like a real one.
+        await rm(target, { force: true }).catch(() => {});
+        failed.push({ file: next.file, reason: attempts > 1 ? `${reason} (after ${attempts} tries)` : reason });
+      } else {
         onFile(next);
-      } catch (error) {
-        failed.push({ file: next.file, reason: error instanceof Error ? error.message : String(error) });
       }
     }
   }

test/unit/device-artifacts.test.ts

@@ -104,4 +104,73 @@ describe('dcd device artifacts', () => {
       [path.join('actions', '001-screenshot.jpg'), 'device.log'].sort(),
     );
   });
+  /** A response whose body never sends a byte, and never closes. */
+  const stalled = () => new Response(new ReadableStream({ start() {} }));
+
+  it('reports a stalled download as timed out instead of hanging', async () => {
+    let calls = 0;
+    globalThis.fetch = (async () => {
+      calls++;
+      return stalled();
+    }) as typeof fetch;
+
+    const started = Date.now();
+    const failed = await downloadArtifactFiles(
+      tempDir,
+      [{ kind: 'logs', file: 'device.log', url: 'https://stalls', bytes: 3, created_at: 't' }],
+      undefined,
+      { idleTimeoutMs: 50 },
+    );
+    expect(Date.now() - started).to.be.lessThan(2000);
+    expect(calls).to.equal(2);
+    expect(failed).to.have.length(1);
+    expect(failed[0].file).to.equal('device.log');
+    expect(failed[0].reason).to.match(/timed out: no data for 50ms/);
+    expect(fs.existsSync(path.join(tempDir, 'device.log'))).to.equal(false);
+  });
+
+  it('times out a download that is slow overall', async () => {
+    let tick: NodeJS.Timeout | undefined;
+    globalThis.fetch = (async () =>
+      new Response(
+        new ReadableStream({
+          // A trickle: data keeps arriving, but the file never finishes.
+          start(controller) {
+            tick = setInterval(() => controller.enqueue(new Uint8Array([1])), 10);
+          },
+          cancel() {
+            clearInterval(tick);
+          },
+        }),
+      )) as typeof fetch;
+
+    const failed = await downloadArtifactFiles(
+      tempDir,
+      [{ kind: 'logs', file: 'device.log', url: 'https://trickle', bytes: 3, created_at: 't' }],
+      undefined,
+      { attempts: 1, idleTimeoutMs: 500, totalTimeoutMs: 100 },
+    );
+    clearInterval(tick);
+    expect(failed.map((f) => f.reason)).to.deep.equal(['timed out: took longer than 100ms']);
+  });
+
+  it('retries a failed download once', async () => {
+    const calls: string[] = [];
+    globalThis.fetch = (async (input: unknown) => {
+      calls.push(String(input));
+      return calls.length === 1 ? stalled() : new Response('log line');
+    }) as typeof fetch;
+
+    const saved: string[] = [];
+    const failed = await downloadArtifactFiles(
+      tempDir,
+      [{ kind: 'logs', file: 'device.log', url: 'https://flaky', bytes: 8, created_at: 't' }],
+      (f) => saved.push(f.file),
+      { idleTimeoutMs: 50 },
+    );
+    expect(failed).to.deep.equal([]);
+    expect(calls).to.have.length(2);
+    expect(saved).to.deep.equal(['device.log']);
+    expect(fs.readFileSync(path.join(tempDir, 'device.log'), 'utf8')).to.equal('log line');
+  });
 });