fix(engine): make remote video downloads atomic (#2774)

## Summary - stage remote video downloads in private per-attempt directories and atomically publish only complete non-empty files - keep the deadline active through response-body streaming and retry one bounded transient failure - preserve render cancellation without sharing abort ownership across independent callers - manually follow at most five redirects, validating HTTPS/public-host policy before every hop - remove stale zero-byte finals and avoid a permanent render-scoped cache map ## Root cause The previous downloader wrote directly to the final cache path and cleared its timeout as soon as response headers arrived. A body timeout or mid-stream socket reset could therefore leave a truncated file at a path that later extraction treated as complete. The downstream symptom was zero extracted frames followed by a generic video coverage failure. ## Retry policy Exactly one retry is allowed for 408, 429, 5xx, timeouts, empty successful bodies, and network/socket failures including nested Undici errors. Cancellation, 404/410, other 4xx, URL/redirect validation errors, and filesystem errors are not retried. ## Security and portability - `mkdtempSync` creates an unguessable same-filesystem staging directory; exclusive writes plus atomic rename prevent symlink planting and partial publication - redirects use `redirect: "manual"` and every resolved `Location` is revalidated before the next request, blocking redirect-to-private/IMDS bypasses - the partial file is opened read/write for `fsync`, which preserves flush semantics and avoids Windows `EPERM` ## Rollout safety This does not change render-plan schema, Plan v1 artifacts, chunk routing, or distributed rendering semantics. It is suitable for the candidate sidecar lane first; stable can remain pinned while we compare video extraction and coverage failures. ## Validation - focused urlDownloader suite: 28 passed - full engine suite: 1,169 passed, 3 skipped - engine typecheck passed - oxlint and oxfmt passed - fallow audit passed - independent code review: approved, no blockers
2026-09-02 20:18:35 +00:00 · 2026-07-26 15:45:50 -04:00
parent a8ee81f3d8 2e84faeb28
commit 814f9cd977
3 changed files with 812 additions and 78 deletions
@@ -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)) {
@@ -1,5 +1,38 @@
-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,
+  symlinkSync,
+  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", () => {
@@ -58,8 +91,450 @@ describe("assertPublicHttpsUrl — SSRF guard", () => {
    expect(() => assertPublicHttpsUrl("https://[::1]/secret")).toThrow("private/reserved");
  });

+  it("rejects normalized reserved IPv6 and IPv4-mapped forms", () => {
+    for (const url of [
+      "https://[::]/secret",
+      "https://[fe80::1]/secret",
+      "https://[fc00::1]/secret",
+      "https://[::ffff:127.0.0.1]/secret",
+      "https://[::ffff:169.254.169.254]/latest/meta-data/",
+    ]) {
+      expect(() => assertPublicHttpsUrl(url), url).toThrow("private/reserved");
+    }
+  });
+
+  it("rejects CGNAT and other non-public IPv4 ranges", () => {
+    for (const url of [
+      "https://100.64.0.1/secret",
+      "https://198.18.0.1/secret",
+      "https://192.0.2.1/secret",
+      "https://224.0.0.1/secret",
+      "https://240.0.0.1/secret",
+      "https://255.255.255.255/secret",
+    ]) {
+      expect(() => assertPublicHttpsUrl(url), url).toThrow("private/reserved");
+    }
+  });
+
  it("rejects invalid URLs", () => {
    expect(() => assertPublicHttpsUrl("not-a-url")).toThrow("Invalid URL");
    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("rejects an IPv4-mapped IMDS redirect before issuing the second request", async () => {
+    const fetchMock = vi.fn().mockResolvedValueOnce(
+      new Response(null, {
+        status: 302,
+        headers: { location: "https://[::ffff:169.254.169.254]/latest/meta-data/" },
+      }),
+    );
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+
+    await expect(
+      downloadToTemp("https://cdn.example/mapped-private-redirect.mp4", dir, 1_000),
+    ).rejects.toMatchObject({
+      kind: "http_rejected",
+      retryable: false,
+    } satisfies Partial<UrlDownloadError>);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+
+  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 trust a nonempty symlink at the final cache path", async () => {
+    const fetchMock = vi.fn().mockResolvedValue(new Response("downloaded"));
+    vi.stubGlobal("fetch", fetchMock);
+    const dir = makeTempDir();
+    const target = join(dir, "attacker-controlled.mp4");
+    const cachePath = join(dir, "download_eda0de5dc5a3.mp4");
+    writeFileSync(target, "not-the-download");
+    symlinkSync(target, cachePath);
+
+    const path = await downloadToTemp("https://cdn.example/stale-empty.mp4", dir, 1_000);
+
+    expect(path).toBe(cachePath);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(readFileSync(path, "utf8")).toBe("downloaded");
+    expect(readFileSync(target, "utf8")).toBe("not-the-download");
+  });
+
+  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([]);
+  });
+});
@@ -1,42 +1,145 @@
-import { createWriteStream, existsSync, mkdirSync } from "fs";
+import {
+  closeSync,
+  createWriteStream,
+  existsSync,
+  fsyncSync,
+  mkdtempSync,
+  mkdirSync,
+  lstatSync,
+  openSync,
+  renameSync,
+  rmSync,
+  statSync,
+} from "fs";
 import { createHash } from "crypto";
-import { join, extname } from "path";
+import { BlockList, isIP } from "node:net";
+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;

-// SSRF guard: these prefixes identify non-public address space that
-// compositions (customer-supplied) must never be able to reach via the
-// download path. Blocks AWS IMDS (169.254.169.254), loopback, RFC1918,
-// and unspecified addresses. All comparisons are on the raw hostname
-// string; DNS resolution is NOT performed here, so DNS-rebinding bypasses
-// are not closed by this check — that gap is acceptable for the risk level.
-const BLOCKED_HOST_PREFIXES = [
-  "169.254.", // link-local / AWS IMDS
-  "127.", // loopback IPv4
-  "10.", // RFC1918
-  "192.168.", // RFC1918
-  "0.", // unspecified
-  "[::1]", // loopback IPv6
-  "[fc", // RFC4193 unique-local IPv6
-  "[fd", // RFC4193 unique-local IPv6
-];
-// 172.16.0.0 – 172.31.255.255 (RFC1918)
-const BLOCKED_172_RANGE = { min: 16, max: 31 };
+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 =
+      typeof current === "object" && current !== null && "cause" in current
+        ? current.cause
+        : undefined;
+  }
+  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 =
+    typeof error === "object" && error !== null && "code" in error && typeof error.code === "string"
+      ? error.code
+      : "";
+  return (
+    RETRYABLE_NETWORK_CODES.has(code) ||
+    code.startsWith("UND_ERR_") ||
+    /fetch failed|network|socket|connection reset|terminated/i.test(message)
+  );
+}
+
+const NON_PUBLIC_IPV4_ADDRESSES = new BlockList();
+for (const [network, prefix] of [
+  ["0.0.0.0", 8],
+  ["10.0.0.0", 8],
+  ["100.64.0.0", 10],
+  ["127.0.0.0", 8],
+  ["169.254.0.0", 16],
+  ["172.16.0.0", 12],
+  ["192.0.0.0", 24],
+  ["192.0.2.0", 24],
+  ["192.88.99.0", 24],
+  ["192.168.0.0", 16],
+  ["198.18.0.0", 15],
+  ["198.51.100.0", 24],
+  ["203.0.113.0", 24],
+  ["224.0.0.0", 4],
+  ["240.0.0.0", 4],
+] as const) {
+  NON_PUBLIC_IPV4_ADDRESSES.addSubnet(network, prefix, "ipv4");
+}
+const NON_PUBLIC_IPV6_ADDRESSES = new BlockList();
+for (const [network, prefix] of [
+  ["::", 128],
+  ["::1", 128],
+  ["::ffff:0:0", 96],
+  ["fc00::", 7],
+  ["fe80::", 10],
+  ["fec0::", 10],
+  ["ff00::", 8],
+  ["2001:db8::", 32],
+] as const) {
+  NON_PUBLIC_IPV6_ADDRESSES.addSubnet(network, prefix, "ipv6");
+}

 function isBlockedHost(hostname: string): boolean {
-  const h = hostname.toLowerCase();
+  const h = hostname.toLowerCase().replace(/^\[|\]$/g, "");
  if (h === "localhost") return true;
-  if (BLOCKED_HOST_PREFIXES.some((p) => h.startsWith(p))) return true;
-  // 172.16–172.31
-  const m = h.match(/^172\.(\d{1,3})\./);
-  if (m) {
-    const octet = parseInt(m[1] ?? "0", 10);
-    if (octet >= BLOCKED_172_RANGE.min && octet <= BLOCKED_172_RANGE.max) return true;
-  }
-  return false;
+  const addressType = isIP(h);
+  if (addressType === 0) return false;
+  return addressType === 4
+    ? NON_PUBLIC_IPV4_ADDRESSES.check(h, "ipv4")
+    : NON_PUBLIC_IPV6_ADDRESSES.check(h, "ipv6");
 }

 /**
@@ -70,21 +173,209 @@ function getFilenameFromUrl(url: string): string {
  return `download_${hash}${ext}`;
 }

+function hasCompleteFile(path: string): boolean {
+  if (!existsSync(path)) return false;
+  const entry = lstatSync(path);
+  if (entry.isFile() && entry.size > 0) return true;
+  // Old versions could leave an empty file behind. Never trust that stale
+  // cache entry—or a symlink/special entry planted at the final path.
+  rmSync(path, { recursive: entry.isDirectory(), 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 +387,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 {