From ecad35a2254c98b84c9410072908c902eb3f7241 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miguel=20=C3=81ngel?= Date: Wed, 28 Oct 2026 10:27:00 +0000 Subject: [PATCH] fix(cli): resume a model download that stalls partway instead of failing (#6649) * fix(cli): resume a model download that stalls partway instead of failing * fix(cli): pin the idle-only timeout; one-line doc for the resume * fix(cli): resume from the bytes on disk and check the size before caching * refactor(cli): split the download response handling into named steps --- packages/cli/src/utils/download.test.ts | 204 +++++++++++++++++++++ packages/cli/src/utils/download.ts | 229 ++++++++++++++++-------- 2 files changed, 360 insertions(+), 73 deletions(-) diff --git a/packages/cli/src/utils/download.test.ts b/packages/cli/src/utils/download.test.ts index 2aca1fcc18..8148c2866a 100644 --- a/packages/cli/src/utils/download.test.ts +++ b/packages/cli/src/utils/download.test.ts @@ -212,6 +212,210 @@ describe("downloadFile", () => { }); }); +describe("stalled downloads", () => { + // A host whose every connection stops after `cut` bytes until the idle timeout fires. + function stallingHost( + content: Buffer, + cut: number, + honourRange: boolean, + ranges: (string | undefined)[], + ) { + return ((_url: string, optionsOrCallback: unknown, maybeCallback?: unknown) => { + const callback = ( + typeof optionsOrCallback === "function" ? optionsOrCallback : maybeCallback + ) as (response: IncomingMessage) => void; + const range = (optionsOrCallback as { headers?: { Range?: string } }).headers?.Range; + ranges.push(range); + const start = honourRange && range ? Number(/bytes=(\d+)-/.exec(range)?.[1]) : 0; + const response = new PassThrough() as PassThrough & { + statusCode: number; + headers: Record; + }; + response.statusCode = start > 0 ? 206 : 200; + response.headers = { + "content-length": String(content.length - start), + ...(start > 0 + ? { "content-range": `bytes ${start}-${content.length - 1}/${content.length}` } + : {}), + }; + class FakeRequest extends EventEmitter { + private timeout: ReturnType | undefined; + setTimeout(ms: number, onTimeout: () => void): this { + this.timeout = setTimeout(onTimeout, ms); + return this; + } + destroy(error: Error): void { + clearTimeout(this.timeout); + response.destroy(error); + this.emit("error", error); + } + } + const request = new FakeRequest(); + callback(response as unknown as IncomingMessage); + const end = Math.min(start + cut, content.length); + queueMicrotask(() => { + response.write(content.subarray(start, end)); + if (end === content.length) { + clearTimeout((request as unknown as { timeout: ReturnType }).timeout); + response.end(); + } + }); + return request as unknown as ClientRequest; + }) as typeof httpsGet; + } + + it("resumes a download that stalls at the same byte count on every connection", async () => { + const content = Buffer.from(Array.from({ length: 1000 }, (_, i) => i % 251)); + const ranges: (string | undefined)[] = []; + mockGet.mockImplementation(stallingHost(content, 300, true, ranges)); + const dir = mkdtempSync(join(tmpdir(), "hyperframes-download-")); + tempDirs.push(dir); + const dest = join(dir, "model.onnx"); + const progress: number[] = []; + + await downloadFile("https://example.test/model.onnx", dest, { + timeoutMs: 10, + onProgress: (received) => progress.push(received), + }); + + expect(readFileSync(dest).equals(content)).toBe(true); + expect(ranges).toEqual([undefined, "bytes=300-", "bytes=600-", "bytes=900-"]); + expect(progress.at(-1)).toBe(1000); + expect(readdirSync(dir)).toEqual(["model.onnx"]); + }); + + it("resumes from the bytes on disk when a reset drops chunks the file had not written", async () => { + const chunk = 64 * 1024; + const content = Buffer.from(Array.from({ length: 64 * chunk }, (_, i) => (i * 7) % 256)); + const ranges: (string | undefined)[] = []; + const dir = mkdtempSync(join(tmpdir(), "hyperframes-download-")); + tempDirs.push(dir); + const dest = join(dir, "model.onnx"); + mockGet.mockImplementation((( + _url: string, + optionsOrCallback: unknown, + maybeCallback?: unknown, + ) => { + const callback = ( + typeof optionsOrCallback === "function" ? optionsOrCallback : maybeCallback + ) as (response: IncomingMessage) => void; + const range = (optionsOrCallback as { headers?: { Range?: string } }).headers?.Range; + ranges.push(range); + const start = range ? Number(/bytes=(\d+)-/.exec(range)?.[1]) : 0; + const response = new PassThrough({ highWaterMark: chunk }) as PassThrough & { + statusCode: number; + headers: Record; + }; + response.statusCode = start > 0 ? 206 : 200; + response.headers = { + "content-length": String(content.length - start), + ...(start > 0 + ? { "content-range": `bytes ${start}-${content.length - 1}/${content.length}` } + : {}), + }; + const request = new EventEmitter() as ClientRequest; + request.setTimeout = vi.fn().mockReturnValue(request); + callback(response as unknown as IncomingMessage); + queueMicrotask(() => { + if (start > 0) { + response.end(content.subarray(start)); + return; + } + // A quarter lands on disk; then a burst arrives and the connection resets before the file writes it. + const quarter = content.length / 4; + for (let at = 0; at < quarter; at += chunk) + response.write(content.subarray(at, at + chunk)); + const written = setInterval(() => { + const partial = readdirSync(dir).find((name) => name.endsWith(".tmp")); + if (!partial || fs.statSync(join(dir, partial)).size < quarter) return; + clearInterval(written); + for (let at = quarter; at < 2 * quarter; at += chunk) + response.write(content.subarray(at, at + chunk)); + response.destroy(new Error("socket hang up")); + }, 1); + }); + return request; + }) as typeof httpsGet); + + await downloadFile("https://example.test/model.onnx", dest, { timeoutMs: 1000 }); + + expect(ranges).toHaveLength(2); + expect(readFileSync(dest).equals(content)).toBe(true); + }); + + it("keeps a slow download going past the timeout while bytes still arrive", async () => { + const content = Buffer.alloc(200, 3); + mockGet.mockImplementation(((_url: string, callback: (response: IncomingMessage) => void) => { + const response = new PassThrough() as PassThrough & { + statusCode: number; + headers: Record; + }; + response.statusCode = 200; + response.headers = { "content-length": "200" }; + class FakeRequest extends EventEmitter { + private timeout: ReturnType | undefined; + private onTimeout = () => {}; + private ms = 0; + // Like a socket idle timeout: each received chunk re-arms it. + setTimeout(ms: number, onTimeout: () => void): this { + this.ms = ms; + this.onTimeout = onTimeout; + this.rearm(); + return this; + } + rearm(): void { + clearTimeout(this.timeout); + this.timeout = setTimeout(this.onTimeout, this.ms); + } + destroy(error: Error): void { + clearTimeout(this.timeout); + clearInterval(drip); + response.destroy(error); + this.emit("error", error); + } + } + const request = new FakeRequest(); + callback(response as unknown as IncomingMessage); + let sent = 0; + const drip = setInterval(() => { + response.write(content.subarray(sent, sent + 10)); + sent += 10; + request.rearm(); + if (sent >= content.length) { + clearInterval(drip); + request.destroy = () => {}; + response.end(); + } + }, 5); + return request as unknown as ClientRequest; + }) as typeof httpsGet); + const dir = mkdtempSync(join(tmpdir(), "hyperframes-download-")); + tempDirs.push(dir); + const dest = join(dir, "model.onnx"); + const started = Date.now(); + + await downloadFile("https://example.test/model.onnx", dest, { timeoutMs: 25 }); + + expect(Date.now() - started).toBeGreaterThan(25 * 2); + expect(readFileSync(dest).equals(content)).toBe(true); + }); + + it("gives up when a retry brings no new bytes", async () => { + const content = Buffer.alloc(1000, 7); + const ranges: (string | undefined)[] = []; + mockGet.mockImplementation(stallingHost(content, 300, false, ranges)); + const dir = mkdtempSync(join(tmpdir(), "hyperframes-download-")); + tempDirs.push(dir); + const dest = join(dir, "model.onnx"); + + await expect( + downloadFile("https://example.test/model.onnx", dest, { timeoutMs: 10 }), + ).rejects.toThrow("Download timed out after 10ms"); + expect(ranges).toEqual([undefined, "bytes=300-"]); + expect(readdirSync(dir)).toEqual([]); + }); +}); + describe("redirect handling", () => { it("closes each redirect response before opening the next request", async () => { const responses: IncomingMessage[] = []; diff --git a/packages/cli/src/utils/download.ts b/packages/cli/src/utils/download.ts index 60df467445..c3bfa0d35c 100644 --- a/packages/cli/src/utils/download.ts +++ b/packages/cli/src/utils/download.ts @@ -1,6 +1,7 @@ import { randomUUID } from "node:crypto"; -import { createWriteStream, renameSync, unlinkSync } from "node:fs"; +import { createWriteStream, renameSync, statSync, unlinkSync } from "node:fs"; import { get as httpsGet } from "node:https"; +import { once } from "node:events"; import type { ClientRequest, IncomingMessage } from "node:http"; import { Transform } from "node:stream"; import { pipeline } from "node:stream/promises"; @@ -43,12 +44,10 @@ function removePartialFile(path: string): void { } } -function measureDownload(options: DownloadOptions, contentLength: string | undefined) { - const length = Number(contentLength); - const totalBytes = Number.isSafeInteger(length) && length >= 0 ? length : null; - let receivedBytes = 0; +function measureDownload(options: DownloadOptions, startBytes: number, totalBytes: number | null) { + let receivedBytes = startBytes; let reportedAt = Date.now(); - options.onProgress?.(0, totalBytes); + options.onProgress?.(receivedBytes, totalBytes); const stream = new Transform({ transform(chunk: Buffer, _encoding, callback) { receivedBytes += chunk.byteLength; @@ -64,94 +63,150 @@ function measureDownload(options: DownloadOptions, contentLength: string | undef callback(null, chunk); }, }); - return { stream, complete: () => options.onProgress?.(receivedBytes, totalBytes) }; + return { + stream, + received: () => receivedBytes, + complete: () => options.onProgress?.(receivedBytes, totalBytes), + }; } function getUrl( url: string, signal: AbortSignal | undefined, + offset: number, onResponse: (res: IncomingMessage) => void, ): ClientRequest { - return signal ? httpsGet(url, { signal }, onResponse) : httpsGet(url, onResponse); + const headers = offset > 0 ? { Range: `bytes=${offset}-` } : undefined; + if (!signal && !headers) return httpsGet(url, onResponse); + return httpsGet( + url, + { ...(signal ? { signal } : {}), ...(headers ? { headers } : {}) }, + onResponse, + ); } -/** - * Download a file from a URL, following redirects. - * Uses atomic write (download to .tmp, rename on success) to prevent - * corrupt partial files from persisting in the cache on interruption. - * - * Location headers are resolved against the URL that sent them, because they - * are frequently relative: a host answering 307 with `/api/resolve-cache/...` - * is normal, and passing that string back as a request target is not. - * - * The timeout is per request, so it re-arms on each hop of a redirect chain - * rather than budgeting the whole chain. A stalled socket is what it exists to - * catch, and without it the CLI waits forever. - */ -export function downloadFile( +function fileSize(path: string): number { + try { + return statSync(path).size; + } catch { + return 0; + } +} + +function byteCount(value: string | undefined): number | null { + const n = Number(value); + return value !== undefined && Number.isSafeInteger(n) && n >= 0 ? n : null; +} + +/** Where a 206 body starts and the whole file's size, from `Content-Range: bytes start-end/total`. */ +function contentRange(header: string | undefined): { start: number; total: number | null } | null { + const match = /^bytes (\d+)-\d+\/(\d+|\*)$/.exec(header ?? ""); + return match + ? { start: Number(match[1]), total: match[2] === "*" ? null : Number(match[2]) } + : null; +} + +type ResponsePlan = + | { kind: "redirect"; location: string } + | { kind: "body"; start: number; total: number | null; append: boolean } + | { kind: "refuse"; status: number | undefined }; + +/** What to do with a response to a request for the bytes from `offset` on. */ +function planResponse(res: IncomingMessage, offset: number): ResponsePlan { + const status = res.statusCode; + const location = res.headers.location; + if (status && REDIRECT_CODES.has(status) && location) return { kind: "redirect", location }; + const range = status === 206 ? contentRange(res.headers["content-range"]) : null; + if (offset > 0 && range?.start === offset) { + return { kind: "body", start: offset, total: range.total, append: true }; + } + // A 200, including one to a range request, is the whole file from byte 0. + if (status === 200) { + return { + kind: "body", + start: 0, + total: byteCount(res.headers["content-length"]), + append: false, + }; + } + return { kind: "refuse", status }; +} + +/** Writes a response body to `tmp`, leaving `state.received` at the bytes that reached disk. */ +async function writeBody( + res: IncomingMessage, + tmp: string, + plan: Extract, + state: { received: number; total: number | null }, + options: DownloadOptions, +): Promise { + state.received = plan.start; + state.total = plan.total; + const meter = measureDownload(options, plan.start, plan.total); + const file = createWriteStream(tmp, { flags: plan.append ? "a" : "w" }); + try { + await pipeline(res, meter.stream, file); + } catch (error) { + // Chunks still buffered in the destroyed file stream never reach disk. + if (!file.closed) await once(file, "close").catch(() => {}); + state.received = fileSize(tmp); + throw error; + } + state.received = meter.received(); + meter.complete(); +} + +/** One request chain; `state.received` holds the bytes in `tmp` even when it rejects. */ +function downloadAttempt( url: string, - dest: string, - options: DownloadOptions = {}, + tmp: string, + state: { received: number; total: number | null }, + options: DownloadOptions, ): Promise { - const tmp = `${dest}.${process.pid}.${randomUUID()}.tmp`; const timeoutMs = options.timeoutMs ?? DEFAULT_DOWNLOAD_TIMEOUT_MS; + const offset = state.received; return new Promise((resolve, reject) => { const follow = (u: string, hops = 0) => { let activeResponse: IncomingMessage | undefined; let responseDiscarded = false; let responsePipelineStarted = false; let requestError: Error | undefined; - let request: ClientRequest; - try { - request = getUrl(u, options.signal, (res) => { - activeResponse = res; - if (res.statusCode && REDIRECT_CODES.has(res.statusCode)) { - const location = res.headers.location; - if (location) { - responseDiscarded = true; - res.destroy(); - if (hops >= MAX_REDIRECTS) { - removePartialFile(tmp); - reject( - new Error( - `Download failed: more than ${MAX_REDIRECTS} redirects from ${url.split("?")[0]}`, - ), - ); - return; - } - try { - follow(redirectTarget(location, u), hops + 1); - } catch (error) { - removePartialFile(tmp); - reject(error); - } - return; + const discard = (res: IncomingMessage) => { + responseDiscarded = true; + res.destroy(); + }; + const onResponse = (res: IncomingMessage) => { + activeResponse = res; + const plan = planResponse(res, offset); + if (plan.kind === "refuse") { + discard(res); + reject(new Error(`Download failed: HTTP ${plan.status}`)); + } else if (plan.kind === "redirect") { + discard(res); + if (hops >= MAX_REDIRECTS) { + reject( + new Error( + `Download failed: more than ${MAX_REDIRECTS} redirects from ${url.split("?")[0]}`, + ), + ); + } else { + try { + follow(redirectTarget(plan.location, u), hops + 1); + } catch (error) { + reject(error); } } - if (res.statusCode !== 200) { - responseDiscarded = true; - res.destroy(); - removePartialFile(tmp); - reject(new Error(`Download failed: HTTP ${res.statusCode}`)); - return; - } - const meter = measureDownload(options, res.headers["content-length"]); - const file = createWriteStream(tmp); + } else { responsePipelineStarted = true; - const transfer = pipeline(res, meter.stream, file); - transfer - .then(() => { - renameSync(tmp, dest); - meter.complete(); - resolve(); - }) - .catch((err) => { - removePartialFile(tmp); - reject(requestError ?? err); - }); - }); + writeBody(res, tmp, plan, state, options).then(resolve, (err) => + reject(requestError ?? err), + ); + } + }; + let request: ClientRequest; + try { + request = getUrl(u, options.signal, offset, onResponse); } catch (error) { - removePartialFile(tmp); reject(error); return; } @@ -165,10 +220,38 @@ export function downloadFile( activeResponse?.destroy(err); return; } - removePartialFile(tmp); reject(err); }); }; follow(url); }); } + +/** Some networks drop every connection at the same byte, so a retry resumes from the partial file. */ +export async function downloadFile( + url: string, + dest: string, + options: DownloadOptions = {}, +): Promise { + const tmp = `${dest}.${process.pid}.${randomUUID()}.tmp`; + const state: { received: number; total: number | null } = { received: 0, total: null }; + for (;;) { + const before = state.received; + try { + await downloadAttempt(url, tmp, state, options); + const size = fileSize(tmp); + if (state.total !== null && size !== state.total) { + state.received = size; + throw new Error(`Download incomplete: ${size} of ${state.total} bytes`); + } + renameSync(tmp, dest); + return; + } catch (error) { + const overLimit = error instanceof Error && error.message.startsWith("Download exceeded"); + if (options.signal?.aborted || overLimit || state.received <= before) { + removePartialFile(tmp); + throw error; + } + } + } +}