diff --git a/packages/cli/src/cloud/download.test.ts b/packages/cli/src/cloud/download.test.ts index 3fd731ee39..4c5b4908e8 100644 --- a/packages/cli/src/cloud/download.test.ts +++ b/packages/cli/src/cloud/download.test.ts @@ -1,4 +1,14 @@ -import { mkdtempSync, readFileSync, rmSync, statSync } from "node:fs"; +import { + chmodSync, + lstatSync, + mkdtempSync, + readFileSync, + readdirSync, + rmSync, + statSync, + symlinkSync, + writeFileSync, +} from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; @@ -109,4 +119,88 @@ describe("cloud/download", () => { ).rejects.toThrow(); expect(() => statSync(dest)).toThrow(); }); + + async function expectFailedReplacement(dest: string, fetchImpl: typeof fetch, error: RegExp) { + await expect(downloadToFile("https://example/x", dest, { fetchImpl })).rejects.toThrow(error); + expect(readFileSync(dest, "utf8")).toBe("previous render"); + expect(readdirSync(dir)).toEqual(["out.mp4"]); + } + + it("preserves an existing output when the response is truncated", async () => { + const dest = join(dir, "out.mp4"); + writeFileSync(dest, "previous render"); + await expectFailedReplacement( + dest, + makeBytesFetch(new Uint8Array([1]), { "content-length": "2" }), + /Truncated download/, + ); + }); + + it("keeps the old output visible until successful replacement and preserves its mode", async () => { + const dest = join(dir, "out.mp4"); + writeFileSync(dest, "previous render"); + chmodSync(dest, 0o640); + await downloadToFile("https://example/x", dest, { + fetchImpl: makeBytesFetch(new Uint8Array([42])), + onProgress: () => { + expect(readFileSync(dest, "utf8")).toBe("previous render"); + expect(readdirSync(dir).filter((name) => name.startsWith(".hf-download-"))).toHaveLength(1); + }, + }); + expect(readFileSync(dest)).toEqual(Buffer.from([42])); + if (process.platform !== "win32") expect(statSync(dest).mode & 0o777).toBe(0o640); + expect(readdirSync(dir)).toEqual(["out.mp4"]); + }); + + it("preserves the previous output when cancelled during progress", async () => { + const dest = join(dir, "out.mp4"); + writeFileSync(dest, "previous render"); + const controller = new AbortController(); + await expect( + downloadToFile("https://example/x", dest, { + fetchImpl: makeBytesFetch(new Uint8Array(1024 * 1024)), + signal: controller.signal, + onProgress: () => controller.abort(new Error("user cancelled")), + }), + ).rejects.toThrow(/user cancelled/); + expect(readFileSync(dest, "utf8")).toBe("previous render"); + expect(readdirSync(dir)).toEqual(["out.mp4"]); + }); + + it("preserves the previous output when the response stream errors", async () => { + const dest = join(dir, "out.mp4"); + writeFileSync(dest, "previous render"); + let sent = false; + const body = new ReadableStream({ + pull(controller) { + if (sent) controller.error(new Error("network interrupted")); + else { + sent = true; + controller.enqueue(new Uint8Array([42])); + } + }, + }); + await expectFailedReplacement( + dest, + (async () => new Response(body)) as typeof fetch, + /network interrupted/, + ); + }); + + it.skipIf(process.platform === "win32")( + "preserves existing and dangling output symlinks", + async () => { + const target = join(dir, "target.mp4"); + const alias = join(dir, "alias.mp4"); + symlinkSync("target.mp4", alias); + for (const value of [1, 2]) { + await downloadToFile("https://example/x", alias, { + fetchImpl: makeBytesFetch(new Uint8Array([value])), + }); + expect(lstatSync(alias).isSymbolicLink()).toBe(true); + expect(readFileSync(target)).toEqual(Buffer.from([value])); + } + expect(readdirSync(dir).sort()).toEqual(["alias.mp4", "target.mp4"]); + }, + ); }); diff --git a/packages/cli/src/cloud/download.ts b/packages/cli/src/cloud/download.ts index d450f8729e..6bf3a5e646 100644 --- a/packages/cli/src/cloud/download.ts +++ b/packages/cli/src/cloud/download.ts @@ -9,12 +9,23 @@ * Failure behavior is "all or nothing": on any error we (1) listen for * stream errors / aborts so awaits resolve promptly instead of hanging, * (2) verify the final byte count matches `content-length` when the - * server supplied one, and (3) `unlinkSync` the partial output so a - * subsequent retry doesn't pick up a corrupted file. + * server supplied one, and (3) discard the private staged output. An + * existing destination is replaced only after the download succeeds. */ -import { createWriteStream, mkdirSync, unlinkSync } from "node:fs"; -import { dirname } from "node:path"; +import { + chmodSync, + createWriteStream, + lstatSync, + mkdirSync, + mkdtempSync, + readlinkSync, + renameSync, + rmSync, + statSync, +} from "node:fs"; +import { dirname, join, resolve } from "node:path"; +import { pipeline } from "node:stream/promises"; export interface DownloadOptions { signal?: AbortSignal; @@ -31,9 +42,11 @@ export interface DownloadResult { /** * Stream `url` into `destPath`. Creates the parent directory if needed, - * truncates any existing file at the destination, and deletes the - * partial output on any error so the caller never observes a corrupt - * file at the returned path. + * replaces an existing file only after the complete response is written, + * and removes staged bytes on failure while preserving the old output. + * Atomic replacement requires a writable parent directory and creates a new + * inode: mode is retained, owner/group are not, and hard links keep old bytes. + * A read-only file can be replaced when its parent permits the rename. */ // fallow-ignore-next-line complexity export async function downloadToFile( @@ -56,96 +69,48 @@ export async function downloadToFile( const total = totalHeader ? Number.parseInt(totalHeader, 10) : undefined; const totalOpt = total !== undefined && Number.isFinite(total) ? total : undefined; - const file = createWriteStream(destPath); + const destination = resolveDownloadDestination(destPath); + const previous = statSync(destination, { throwIfNoEntry: false }); + const stage = mkdtempSync(join(dirname(destination), ".hf-download-")); + const stagedFile = join(stage, "download"); let bytes = 0; - let errored = false; try { - for await (const chunk of res.body as unknown as AsyncIterable) { - if (options.signal?.aborted) { - throw options.signal.reason instanceof Error - ? options.signal.reason - : new Error("Download aborted"); - } - bytes += chunk.byteLength; - options.onProgress?.(bytes, totalOpt); - if (!file.write(chunk)) { - await waitForDrain(file, options.signal); - } - } - if (totalOpt !== undefined && bytes !== totalOpt) { - throw new Error( - `Truncated download: got ${bytes} bytes, expected ${totalOpt} (content-length). ` + - `The presigned URL may have expired mid-transfer — refetch via \`hyperframes cloud get\`.`, - ); - } - } catch (err) { - errored = true; - throw err; + await pipeline( + async function* () { + for await (const chunk of res.body as unknown as AsyncIterable) { + bytes += chunk.byteLength; + options.onProgress?.(bytes, totalOpt); + yield chunk; + } + if (totalOpt !== undefined && bytes !== totalOpt) { + throw new Error( + `Truncated download: got ${bytes} bytes, expected ${totalOpt} (content-length). ` + + `The presigned URL may have expired mid-transfer — refetch via \`hyperframes cloud get\`.`, + ); + } + }, + createWriteStream(stagedFile, { flags: "wx" }), + { signal: options.signal }, + ); + options.signal?.throwIfAborted(); + if (previous?.isFile()) chmodSync(stagedFile, previous.mode & 0o777); + renameSync(stagedFile, destination); + } catch (error) { + const reason = options.signal?.reason; + if (options.signal?.aborted && reason instanceof Error) throw reason; + throw error; } finally { - await closeFile(file); - if (errored) { - // Don't let a partial file pose as the final artifact. Best- - // effort unlink — if it fails (already gone, permission), we - // re-throw the original error. - try { - unlinkSync(destPath); - } catch { - /* swallow */ - } - } + rmSync(stage, { recursive: true, force: true }); } return { path: destPath, bytes }; } -/** - * Resolve when the write stream emits `drain`, or reject on `error` / - * `close` / signal abort — avoids the hang from awaiting a one-shot - * `drain` event that never fires because the stream tore down first. - */ -function waitForDrain(file: NodeJS.WritableStream, signal?: AbortSignal): Promise { - return new Promise((resolve, reject) => { - const cleanup = () => { - file.off("drain", onDrain); - file.off("error", onError); - file.off("close", onClose); - signal?.removeEventListener("abort", onAbort); - }; - const onDrain = (): void => { - cleanup(); - resolve(); - }; - const onError = (err: Error): void => { - cleanup(); - reject(err); - }; - const onClose = (): void => { - cleanup(); - reject(new Error("write stream closed before drain")); - }; - const onAbort = (): void => { - cleanup(); - const reason = signal?.reason; - reject(reason instanceof Error ? reason : new Error("Download aborted")); - }; - file.once("drain", onDrain); - file.once("error", onError); - file.once("close", onClose); - signal?.addEventListener("abort", onAbort, { once: true }); - }); -} - -function closeFile(file: NodeJS.WritableStream): Promise { - return new Promise((resolve) => { - // Best-effort cleanup: any underlying failure has already been - // surfaced as the original throw from the for-await loop. We - // listen for `error` so a failing close (bad fd, late ENOSPC on - // flush) doesn't leak an unhandled 'error' onto the stream, and - // resolve either way so the finally block proceeds to unlinkSync. - const done = (): void => { - file.off("error", done); - resolve(); - }; - file.once("error", done); - file.end(() => done()); - }); +/** Preserve the existing behavior of writing through a caller-selected symlink. */ +function resolveDownloadDestination(destPath: string): string { + let destination = resolve(destPath); + for (let hops = 0; hops < 40; hops++) { + if (!lstatSync(destination, { throwIfNoEntry: false })?.isSymbolicLink()) return destination; + destination = resolve(dirname(destination), readlinkSync(destination)); + } + throw new Error(`Too many symbolic links in download destination: ${destPath}`); } diff --git a/packages/cli/src/commands/cloud/render.ts b/packages/cli/src/commands/cloud/render.ts index 27284071e5..b2087a715e 100644 --- a/packages/cli/src/commands/cloud/render.ts +++ b/packages/cli/src/commands/cloud/render.ts @@ -852,8 +852,7 @@ async function streamVideo( destPath: string, asJson: boolean, ): Promise<{ bytes: number }> { - // `downloadToFile` already creates the parent directory and cleans - // up the partial file on error — no pre-mkdir needed here. + // `downloadToFile` creates the parent directory and preserves existing output on failure. if (!asJson) { console.log(""); console.log(`${c.accent("◆")} Downloading to ${c.accent(destPath)}`);