fix(cloud): preserve existing output when downloads fail (#3840)

* fix(cloud): preserve existing output when downloads fail

* test(cloud): pin same-directory download staging

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<Uint8Array>({
+      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"]);
+    },
+  );
 });

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<Uint8Array>) {
-      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<Uint8Array>) {
+          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<void> {
-  return new Promise<void>((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<void> {
-  return new Promise<void>((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}`);
 }

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)}`);