mirror of
https://github.com/heygen-com/hyperframes.git
synced 2026-10-30 09:16:55 +00:00
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
This commit is contained in:
Miguel Ángel
@@ -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<string, string>;
|
||||
};
|
||||
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<typeof setTimeout> | 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<typeof setTimeout> }).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<string, string>;
|
||||
};
|
||||
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<string, string>;
|
||||
};
|
||||
response.statusCode = 200;
|
||||
response.headers = { "content-length": "200" };
|
||||
class FakeRequest extends EventEmitter {
|
||||
private timeout: ReturnType<typeof setTimeout> | 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[] = [];
|
||||
|
||||
@@ -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<ResponsePlan, { kind: "body" }>,
|
||||
state: { received: number; total: number | null },
|
||||
options: DownloadOptions,
|
||||
): Promise<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user