diff --git a/packages/engine/src/services/videoFrameExtractor.ts b/packages/engine/src/services/videoFrameExtractor.ts index ec0cedf2f..953b3433f 100644 --- a/packages/engine/src/services/videoFrameExtractor.ts +++ b/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)) { diff --git a/packages/engine/src/utils/urlDownloader.test.ts b/packages/engine/src/utils/urlDownloader.test.ts index 22fae4f93..b544ee8ca 100644 --- a/packages/engine/src/utils/urlDownloader.test.ts +++ b/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); + 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({ + 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); + 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); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("removes a partial body after a network reset before retrying", async () => { + const resetBody = new ReadableStream({ + 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({ + 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({ + 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); + 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((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({ + 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((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); + 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((resolve) => { + setTimeout(() => resolve(new Response("complete")), 10); + }), + ) + .mockImplementationOnce(async (_url: string, init: RequestInit) => { + const body = new ReadableStream({ + 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); + const path = await first; + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); +}); diff --git a/packages/engine/src/utils/urlDownloader.ts b/packages/engine/src/utils/urlDownloader.ts index 3dbd65321..894e69876 100644 --- a/packages/engine/src/utils/urlDownloader.ts +++ b/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(); const inFlightDownloads = new Map>(); +const signalScopes = new WeakMap(); +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 { + 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 { + 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 { + 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 { + // 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 { + 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 { // 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 {