fix(engine): make video downloads atomic and retry transient failures

packages/engine/src/services/videoFrameExtractor.ts

@@ -730,7 +730,7 @@ export async function extractAllVideoFrames(
       if (isHttpUrl(videoPath)) {
         const downloadDir = join(options.outputDir, "_downloads");
         mkdirSync(downloadDir, { recursive: true });
-        videoPath = await downloadToTemp(videoPath, downloadDir);
+        videoPath = await downloadToTemp(videoPath, downloadDir, undefined, signal);
       }
 
       if (!existsSync(videoPath)) {

packages/engine/src/utils/urlDownloader.test.ts

@@ -1,5 +1,30 @@
-import { describe, expect, it } from "vitest";
-import { assertPublicHttpsUrl } from "./urlDownloader.js";
+// fallow-ignore-file code-duplication
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { existsSync, mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { assertPublicHttpsUrl, downloadToTemp, UrlDownloadError } from "./urlDownloader.js";
+
+const tempDirs: string[] = [];
+
+function makeTempDir(): string {
+  const dir = mkdtempSync(join(tmpdir(), "hf-url-download-"));
+  tempDirs.push(dir);
+  return dir;
+}
+
+function temporaryDownloadEntries(dir: string): string[] {
+  return readdirSync(dir).filter(
+    (name) => name.includes(".partial-") || name.startsWith(".hf-download-"),
+  );
+}
+
+afterEach(() => {
+  vi.unstubAllGlobals();
+  for (const dir of tempDirs.splice(0)) {
+    rmSync(dir, { recursive: true, force: true });
+  }
+});
 
 describe("assertPublicHttpsUrl — SSRF guard", () => {
   it("accepts public HTTPS URLs", () => {
@@ -63,3 +88,384 @@ describe("assertPublicHttpsUrl — SSRF guard", () => {
     expect(() => assertPublicHttpsUrl("")).toThrow("Invalid URL");
   });
 });
+
+describe("downloadToTemp atomic publication and bounded retry", () => {
+  it("follows a bounded redirect only after validating the next public HTTPS hop", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce(
+        new Response(null, {
+          status: 302,
+          headers: { location: "https://media.example/final.mp4" },
+        }),
+      )
+      .mockResolvedValueOnce(new Response("complete"));
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    const path = await downloadToTemp("https://cdn.example/redirect.mp4", dir, 1_000);
+
+    expect(fetchMock).toHaveBeenNthCalledWith(
+      2,
+      "https://media.example/final.mp4",
+      expect.objectContaining({ redirect: "manual" }),
+    );
+    expect(readFileSync(path, "utf8")).toBe("complete");
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+
+  it("rejects a public redirect to a private host before issuing the second request", async () => {
+    const fetchMock = vi.fn().mockResolvedValueOnce(
+      new Response(null, {
+        status: 302,
+        headers: { location: "http://169.254.169.254/latest/meta-data/" },
+      }),
+    );
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    await expect(
+      downloadToTemp("https://cdn.example/private-redirect.mp4", dir, 1_000),
+    ).rejects.toMatchObject({
+      kind: "http_rejected",
+      retryable: false,
+    } satisfies Partial<UrlDownloadError>);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+
+  it("retries one HTTP 503 and publishes only the complete final file", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce(new Response(null, { status: 503, statusText: "Service Unavailable" }))
+      .mockResolvedValueOnce(new Response("complete"));
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    const path = await downloadToTemp("https://cdn.example/retry-503.mp4", dir, 1_000);
+
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(readFileSync(path, "utf8")).toBe("complete");
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+
+  it("cancels a streaming HTTP error body before retrying", async () => {
+    let errorBodyCancelled = false;
+    const errorBody = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode("error details"));
+      },
+      cancel() {
+        errorBodyCancelled = true;
+      },
+    });
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce(
+        new Response(errorBody, { status: 503, statusText: "Service Unavailable" }),
+      )
+      .mockResolvedValueOnce(new Response("complete"));
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    const path = await downloadToTemp("https://cdn.example/streaming-503.mp4", dir, 1_000);
+
+    expect(errorBodyCancelled).toBe(true);
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(readFileSync(path, "utf8")).toBe("complete");
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+
+  it("does not retry a deterministic 404", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValue(new Response(null, { status: 404, statusText: "Not Found" }));
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    await expect(
+      downloadToTemp("https://cdn.example/missing-404.mp4", dir, 1_000),
+    ).rejects.toMatchObject({
+      kind: "http_not_found",
+      retryable: false,
+      status: 404,
+    } satisfies Partial<UrlDownloadError>);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+
+  it("exhausts the transient retry budget after exactly one retry", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValue(new Response(null, { status: 503, statusText: "Service Unavailable" }));
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    await expect(
+      downloadToTemp("https://cdn.example/always-503.mp4", dir, 1_000),
+    ).rejects.toMatchObject({
+      kind: "http_transient",
+      retryable: true,
+      status: 503,
+    } satisfies Partial<UrlDownloadError>);
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+
+  it("removes a partial body after a network reset before retrying", async () => {
+    const resetBody = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode("partial"));
+        const error = Object.assign(new Error("socket reset"), { code: "ECONNRESET" });
+        controller.error(error);
+      },
+    });
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce(new Response(resetBody))
+      .mockResolvedValueOnce(new Response("complete"));
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    const path = await downloadToTemp("https://cdn.example/reset-once.mp4", dir, 1_000);
+
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(readFileSync(path, "utf8")).toBe("complete");
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+
+  it("retries an Undici mid-body disconnect reported through a nested cause", async () => {
+    const disconnectedBody = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode("partial"));
+        const cause = Object.assign(new Error("other side closed"), {
+          code: "UND_ERR_SOCKET",
+        });
+        controller.error(new TypeError("terminated", { cause }));
+      },
+    });
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce(new Response(disconnectedBody))
+      .mockResolvedValueOnce(new Response("complete"));
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    const path = await downloadToTemp("https://cdn.example/undici-reset-once.mp4", dir, 1_000);
+
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(readFileSync(path, "utf8")).toBe("complete");
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+
+  it("keeps the deadline active through a stalled response body", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockImplementationOnce(async (_url: string, init: RequestInit) => {
+        const stalledBody = new ReadableStream<Uint8Array>({
+          start(controller) {
+            controller.enqueue(new TextEncoder().encode("partial"));
+            init.signal?.addEventListener(
+              "abort",
+              () => controller.error(new DOMException("aborted", "AbortError")),
+              { once: true },
+            );
+          },
+        });
+        return new Response(stalledBody);
+      })
+      .mockResolvedValueOnce(new Response("complete"));
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    const path = await downloadToTemp("https://cdn.example/stalled-body.mp4", dir, 20);
+
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(readFileSync(path, "utf8")).toBe("complete");
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+
+  it("retries a zero-byte 200 response without publishing it", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce(new Response(""))
+      .mockResolvedValueOnce(new Response("complete"));
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    const path = await downloadToTemp("https://cdn.example/empty-once.mp4", dir, 1_000);
+
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(readFileSync(path, "utf8")).toBe("complete");
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+
+  it("retries a 200 response with no body", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce(new Response(null))
+      .mockResolvedValueOnce(new Response("complete"));
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    const path = await downloadToTemp("https://cdn.example/null-body-once.mp4", dir, 1_000);
+
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(readFileSync(path, "utf8")).toBe("complete");
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+
+  it("removes a stale zero-byte final file before downloading", async () => {
+    const fetchMock = vi.fn().mockResolvedValue(new Response("complete"));
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+    const stalePath = join(dir, "download_eda0de5dc5a3.mp4");
+    writeFileSync(stalePath, "");
+
+    const path = await downloadToTemp("https://cdn.example/stale-empty.mp4", dir, 1_000);
+
+    expect(path).toBe(stalePath);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(readFileSync(path, "utf8")).toBe("complete");
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+
+  it("does not retry caller cancellation", async () => {
+    const fetchMock = vi.fn();
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+    const controller = new AbortController();
+    controller.abort();
+
+    await expect(
+      downloadToTemp("https://cdn.example/cancelled.mp4", dir, 1_000, controller.signal),
+    ).rejects.toMatchObject({
+      kind: "cancelled",
+      retryable: false,
+    } satisfies Partial<UrlDownloadError>);
+    expect(fetchMock).not.toHaveBeenCalled();
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+
+  it("deduplicates concurrent callers for the same URL and destination", async () => {
+    const fetchMock = vi.fn().mockImplementation(
+      () =>
+        new Promise<Response>((resolve) => {
+          setTimeout(() => resolve(new Response("complete")), 10);
+        }),
+    );
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    const [first, second] = await Promise.all([
+      downloadToTemp("https://cdn.example/concurrent.mp4", dir, 1_000),
+      downloadToTemp("https://cdn.example/concurrent.mp4", dir, 1_000),
+    ]);
+
+    expect(first).toBe(second);
+    expect(existsSync(first)).toBe(true);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("does not let one caller cancellation abort another caller", async () => {
+    const firstController = new AbortController();
+    const secondController = new AbortController();
+    const fetchMock = vi
+      .fn()
+      .mockImplementationOnce(async (_url: string, init: RequestInit) => {
+        const body = new ReadableStream<Uint8Array>({
+          start(controller) {
+            controller.enqueue(new TextEncoder().encode("partial"));
+            init.signal?.addEventListener(
+              "abort",
+              () => controller.error(new DOMException("aborted", "AbortError")),
+              { once: true },
+            );
+          },
+        });
+        return new Response(body);
+      })
+      .mockImplementationOnce(
+        () =>
+          new Promise<Response>((resolve) => {
+            setTimeout(() => resolve(new Response("complete")), 10);
+          }),
+      );
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    const first = downloadToTemp(
+      "https://cdn.example/cancellation-isolation-a.mp4",
+      dir,
+      1_000,
+      firstController.signal,
+    );
+    const second = downloadToTemp(
+      "https://cdn.example/cancellation-isolation-a.mp4",
+      dir,
+      1_000,
+      secondController.signal,
+    );
+    firstController.abort();
+
+    await expect(first).rejects.toMatchObject({
+      kind: "cancelled",
+      retryable: false,
+    } satisfies Partial<UrlDownloadError>);
+    const path = await second;
+    expect(readFileSync(path, "utf8")).toBe("complete");
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+
+  it("does not let a later caller cancellation abort the first caller", async () => {
+    const firstController = new AbortController();
+    const secondController = new AbortController();
+    const fetchMock = vi
+      .fn()
+      .mockImplementationOnce(
+        () =>
+          new Promise<Response>((resolve) => {
+            setTimeout(() => resolve(new Response("complete")), 10);
+          }),
+      )
+      .mockImplementationOnce(async (_url: string, init: RequestInit) => {
+        const body = new ReadableStream<Uint8Array>({
+          start(controller) {
+            controller.enqueue(new TextEncoder().encode("partial"));
+            init.signal?.addEventListener(
+              "abort",
+              () => controller.error(new DOMException("aborted", "AbortError")),
+              { once: true },
+            );
+          },
+        });
+        return new Response(body);
+      });
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    const first = downloadToTemp(
+      "https://cdn.example/cancellation-isolation-b.mp4",
+      dir,
+      1_000,
+      firstController.signal,
+    );
+    const second = downloadToTemp(
+      "https://cdn.example/cancellation-isolation-b.mp4",
+      dir,
+      1_000,
+      secondController.signal,
+    );
+    secondController.abort();
+
+    await expect(second).rejects.toMatchObject({
+      kind: "cancelled",
+      retryable: false,
+    } satisfies Partial<UrlDownloadError>);
+    const path = await first;
+    expect(readFileSync(path, "utf8")).toBe("complete");
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(temporaryDownloadEntries(dir)).toEqual([]);
+  });
+});

packages/engine/src/utils/urlDownloader.ts

@@ -1,11 +1,94 @@
-import { createWriteStream, existsSync, mkdirSync } from "fs";
+import {
+  closeSync,
+  createWriteStream,
+  existsSync,
+  fsyncSync,
+  mkdtempSync,
+  mkdirSync,
+  openSync,
+  renameSync,
+  rmSync,
+  statSync,
+} from "fs";
 import { createHash } from "crypto";
-import { join, extname } from "path";
+import { dirname, extname, join } from "path";
 import { Readable } from "stream";
-import { finished } from "stream/promises";
+import { pipeline } from "stream/promises";
 
-const downloadPathCache = new Map<string, string>();
 const inFlightDownloads = new Map<string, Promise<string>>();
+const signalScopes = new WeakMap<AbortSignal, number>();
+let nextSignalScope = 1;
+
+function signalScopeKey(signal: AbortSignal | undefined): string {
+  if (!signal) return "none";
+  let scope = signalScopes.get(signal);
+  if (scope === undefined) {
+    scope = nextSignalScope;
+    nextSignalScope += 1;
+    signalScopes.set(signal, scope);
+  }
+  return String(scope);
+}
+
+export type UrlDownloadFailureKind =
+  | "cancelled"
+  | "timeout"
+  | "http_not_found"
+  | "http_rejected"
+  | "http_transient"
+  | "network"
+  | "empty_body"
+  | "filesystem";
+
+export class UrlDownloadError extends Error {
+  constructor(
+    readonly kind: UrlDownloadFailureKind,
+    readonly retryable: boolean,
+    message: string,
+    readonly status?: number,
+  ) {
+    super(message);
+    this.name = "UrlDownloadError";
+  }
+}
+
+function classifyHttpFailure(status: number, statusText: string): UrlDownloadError {
+  const message = `HTTP ${status}: ${statusText}`;
+  if (status === 404 || status === 410) {
+    return new UrlDownloadError("http_not_found", false, message, status);
+  }
+  if (status === 408 || status === 429 || status >= 500) {
+    return new UrlDownloadError("http_transient", true, message, status);
+  }
+  return new UrlDownloadError("http_rejected", false, message, status);
+}
+
+function classifyDownloadFailure(error: unknown): UrlDownloadError {
+  if (error instanceof UrlDownloadError) return error;
+  const message = error instanceof Error ? error.message : String(error);
+  let current: unknown = error;
+  // Undici often wraps a mid-body socket failure as `TypeError: terminated`
+  // with the actionable `UND_ERR_*` code on `cause`.
+  for (let depth = 0; current && depth < 4; depth += 1) {
+    if (isRetryableNetworkCause(current)) {
+      return new UrlDownloadError("network", true, `Download failed: ${message}`);
+    }
+    current = (current as Error & { cause?: unknown }).cause;
+  }
+  return new UrlDownloadError("filesystem", false, `Download failed: ${message}`);
+}
+
+const RETRYABLE_NETWORK_CODES = new Set(["ECONNRESET", "ECONNREFUSED", "ETIMEDOUT", "EAI_AGAIN"]);
+
+function isRetryableNetworkCause(error: unknown): boolean {
+  const message = error instanceof Error ? error.message : String(error);
+  const code = (error as NodeJS.ErrnoException | undefined)?.code ?? "";
+  return (
+    RETRYABLE_NETWORK_CODES.has(code) ||
+    code.startsWith("UND_ERR_") ||
+    /fetch failed|network|socket|connection reset|terminated/i.test(message)
+  );
+}
 
 // SSRF guard: these prefixes identify non-public address space that
 // compositions (customer-supplied) must never be able to reach via the
@@ -70,21 +153,208 @@ function getFilenameFromUrl(url: string): string {
   return `download_${hash}${ext}`;
 }
 
+function hasCompleteFile(path: string): boolean {
+  if (!existsSync(path)) return false;
+  if (statSync(path).size > 0) return true;
+  // Old versions wrote directly to the final path and could leave an empty
+  // file behind. Never trust that stale cache entry.
+  rmSync(path, { force: true });
+  return false;
+}
+
+const REDIRECT_STATUSES = new Set([301, 302, 303, 307, 308]);
+const MAX_REDIRECTS = 5;
+
+function assertAllowedDownloadUrl(url: string, redirect: boolean): void {
+  try {
+    assertPublicHttpsUrl(url);
+  } catch {
+    throw new UrlDownloadError(
+      "http_rejected",
+      false,
+      redirect ? "Download redirect target is not permitted" : "Download URL is not permitted",
+    );
+  }
+}
+
+async function cancelResponseBody(response: Response): Promise<void> {
+  try {
+    await response.body?.cancel();
+  } catch {
+    // Redirect validation remains authoritative if teardown also fails.
+  }
+}
+
+function resolveRedirectUrl(response: Response, currentUrl: string, redirects: number): string {
+  if (redirects >= MAX_REDIRECTS) {
+    throw new UrlDownloadError("http_rejected", false, "Download exceeded redirect limit");
+  }
+  const location = response.headers.get("location");
+  if (!location) {
+    throw new UrlDownloadError(
+      "http_rejected",
+      false,
+      "Download redirect omitted a Location header",
+    );
+  }
+  try {
+    return new URL(location, currentUrl).toString();
+  } catch {
+    throw new UrlDownloadError("http_rejected", false, "Download redirect Location is invalid");
+  }
+}
+
+async function fetchWithValidatedRedirects(
+  initialUrl: string,
+  controller: AbortController,
+): Promise<Response> {
+  let currentUrl = initialUrl;
+  for (let redirects = 0; ; redirects += 1) {
+    assertAllowedDownloadUrl(currentUrl, redirects > 0);
+
+    // lgtm[js/file-access-to-http] — every redirect hop is fetched manually
+    // only after the HTTPS/private-host guard above; automatic redirect
+    // following is disabled so an allowed host cannot bounce into IMDS.
+    const response = await fetch(currentUrl, {
+      signal: controller.signal,
+      redirect: "manual",
+    });
+    if (!REDIRECT_STATUSES.has(response.status)) return response;
+
+    await cancelResponseBody(response);
+    currentUrl = resolveRedirectUrl(response, currentUrl, redirects);
+  }
+}
+
+async function fetchToPartial(
+  url: string,
+  partialPath: string,
+  controller: AbortController,
+): Promise<void> {
+  const response = await fetchWithValidatedRedirects(url, controller);
+  if (!response.ok) {
+    // Do not leave a streaming error response holding an Undici connection
+    // while the bounded retry starts.
+    try {
+      await response.body?.cancel();
+    } catch {
+      // The HTTP status remains the useful failure if teardown also fails.
+    }
+    throw classifyHttpFailure(response.status, response.statusText);
+  }
+  if (!response.body) {
+    throw new UrlDownloadError("empty_body", true, "Download response body is empty");
+  }
+
+  const fileStream = createWriteStream(partialPath, { flags: "wx" });
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  const readableStream = Readable.fromWeb(response.body as any);
+  await pipeline(readableStream, fileStream);
+  if (statSync(partialPath).size === 0) {
+    throw new UrlDownloadError("empty_body", true, "Download response body contained zero bytes");
+  }
+}
+
+function syncAndPublishPartial(partialPath: string, localPath: string): void {
+  // Windows rejects fsync on a read-only handle (EPERM); the partial is ours
+  // and writable, so r+ preserves the same flush semantics cross-platform.
+  const fd = openSync(partialPath, "r+");
+  try {
+    fsyncSync(fd);
+  } finally {
+    closeSync(fd);
+  }
+
+  // Different cancellation scopes intentionally do not share a physical
+  // request. If another complete attempt won the final-path race, reuse it.
+  if (hasCompleteFile(localPath)) return;
+  try {
+    renameSync(partialPath, localPath);
+  } catch (error) {
+    if (!hasCompleteFile(localPath)) throw error;
+  }
+}
+
+async function runDownloadAttempt(
+  url: string,
+  localPath: string,
+  timeoutMs: number,
+  signal?: AbortSignal,
+): Promise<string> {
+  // A private, unguessable directory prevents symlink planting and keeps the
+  // partial on the destination filesystem so the final rename stays atomic.
+  const attemptDir = mkdtempSync(join(dirname(localPath), ".hf-download-"));
+  const partialPath = join(attemptDir, "payload");
+  const controller = new AbortController();
+  let timedOut = false;
+  let callerAborted = signal?.aborted ?? false;
+  const onCallerAbort = (): void => {
+    callerAborted = true;
+    controller.abort();
+  };
+  signal?.addEventListener("abort", onCallerAbort, { once: true });
+  const timeoutId = setTimeout(() => {
+    timedOut = true;
+    controller.abort();
+  }, timeoutMs);
+
+  try {
+    if (callerAborted) {
+      throw new UrlDownloadError("cancelled", false, "Download cancelled");
+    }
+    await fetchToPartial(url, partialPath, controller);
+    syncAndPublishPartial(partialPath, localPath);
+    return localPath;
+  } catch (error) {
+    if (callerAborted) {
+      throw new UrlDownloadError("cancelled", false, "Download cancelled");
+    }
+    if (timedOut) {
+      throw new UrlDownloadError("timeout", true, `Download timeout after ${timeoutMs / 1000}s`);
+    }
+    throw classifyDownloadFailure(error);
+  } finally {
+    clearTimeout(timeoutId);
+    signal?.removeEventListener("abort", onCallerAbort);
+    controller.abort();
+    rmSync(attemptDir, { recursive: true, force: true });
+  }
+}
+
+async function downloadWithRetry(
+  url: string,
+  localPath: string,
+  timeoutMs: number,
+  signal?: AbortSignal,
+): Promise<string> {
+  const maxTransientRetries = 1;
+  for (let attempt = 0; ; attempt += 1) {
+    try {
+      return await runDownloadAttempt(url, localPath, timeoutMs, signal);
+    } catch (error) {
+      const classified = classifyDownloadFailure(error);
+      if (!classified.retryable || attempt >= maxTransientRetries) throw classified;
+    }
+  }
+}
+
 export async function downloadToTemp(
   url: string,
   destDir: string,
   timeoutMs: number = 300000,
+  signal?: AbortSignal,
 ): Promise<string> {
   // Reject non-HTTPS URLs and private/reserved address ranges before
   // touching the cache or filesystem — customer-supplied compositions must
   // not be able to trigger outbound fetches to internal infrastructure.
   assertPublicHttpsUrl(url);
 
-  const cachedPath = downloadPathCache.get(url);
-  if (cachedPath && existsSync(cachedPath)) {
-    return cachedPath;
-  }
-  const inFlight = inFlightDownloads.get(url);
+  const cacheKey = `${url}\0${destDir}`;
+  // The physical request may be shared only by callers with the same
+  // cancellation scope and deadline. Otherwise the first caller's abort or
+  // timeout would incorrectly own every waiter.
+  const inFlightKey = `${cacheKey}\0${timeoutMs}\0${signalScopeKey(signal)}`;
+  const inFlight = inFlightDownloads.get(inFlightKey);
   if (inFlight) {
     return inFlight;
   }
@@ -96,46 +366,14 @@ export async function downloadToTemp(
   const filename = getFilenameFromUrl(url);
   const localPath = join(destDir, filename);
 
-  if (existsSync(localPath)) {
-    downloadPathCache.set(url, localPath);
-    return localPath;
-  }
+  if (hasCompleteFile(localPath)) return localPath;
 
-  const downloadPromise = (async () => {
-    try {
-      const controller = new AbortController();
-      const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
-
-      const response = await fetch(url, { signal: controller.signal });
-      clearTimeout(timeoutId);
-
-      if (!response.ok) {
-        throw new Error(`HTTP ${response.status}: ${response.statusText}`);
-      }
-
-      if (!response.body) {
-        throw new Error("Response body is empty");
-      }
-
-      const fileStream = createWriteStream(localPath);
-      // eslint-disable-next-line @typescript-eslint/no-explicit-any
-      const readableStream = Readable.fromWeb(response.body as any);
-      await finished(readableStream.pipe(fileStream));
-
-      downloadPathCache.set(url, localPath);
-      return localPath;
-    } catch (err) {
-      const message = err instanceof Error ? err.message : String(err);
-      if (message.includes("aborted")) {
-        throw new Error(`[URLDownloader] Download timeout after ${timeoutMs / 1000}s: ${url}`);
-      }
-      throw new Error(`[URLDownloader] Download failed: ${message}`);
-    } finally {
-      inFlightDownloads.delete(url);
-    }
-  })();
-  inFlightDownloads.set(url, downloadPromise);
-  return downloadPromise;
+  const downloadPromise = downloadWithRetry(url, localPath, timeoutMs, signal);
+  const trackedDownload = downloadPromise.finally(() => {
+    inFlightDownloads.delete(inFlightKey);
+  });
+  inFlightDownloads.set(inFlightKey, trackedDownload);
+  return trackedDownload;
 }
 
 export function isHttpUrl(path: string): boolean {