fix(render): publish artifacts atomically (#2154)

packages/producer/src/services/render/artifactTransaction.test.ts

@@ -0,0 +1,202 @@
+import { describe, expect, it } from "vitest";
+import {
+  existsSync,
+  lstatSync,
+  mkdirSync,
+  mkdtempSync,
+  readFileSync,
+  readdirSync,
+  renameSync as renamePathSync,
+  rmSync,
+  writeFileSync,
+} from "node:fs";
+import { dirname, join } from "node:path";
+import { tmpdir } from "node:os";
+import { ArtifactTransaction } from "./artifactTransaction.js";
+
+function tempDir(): string {
+  return mkdtempSync(join(tmpdir(), "hf-artifact-transaction-"));
+}
+
+function transactionDirectories(dir: string): string[] {
+  return readdirSync(dir).filter((name) => name.includes(".hf-transaction-"));
+}
+
+describe("ArtifactTransaction", () => {
+  it("uses a private, atomically reserved sibling directory", () => {
+    const dir = tempDir();
+    const destination = join(dir, "render.mp4");
+    const transaction = new ArtifactTransaction(destination, "file");
+    const transactionDir = dirname(transaction.stagingPath);
+
+    expect(dirname(transactionDir)).toBe(dir);
+    expect(transaction.stagingPath).toBe(join(transactionDir, "render.mp4"));
+    if (process.platform !== "win32") {
+      expect(lstatSync(transactionDir).mode & 0o777).toBe(0o700);
+    }
+
+    transaction.rollback();
+    expect(transactionDirectories(dir)).toEqual([]);
+  });
+
+  it("atomically replaces a file only after validation", () => {
+    const dir = tempDir();
+    const destination = join(dir, "render.mp4");
+    writeFileSync(destination, "existing");
+    const transaction = new ArtifactTransaction(destination, "file");
+    writeFileSync(transaction.stagingPath, "new-render");
+
+    transaction.commit();
+
+    expect(readFileSync(destination, "utf8")).toBe("new-render");
+    expect(existsSync(transaction.stagingPath)).toBe(false);
+    expect(transactionDirectories(dir)).toEqual([]);
+  });
+
+  it("leaves an existing file byte-identical when validation fails", () => {
+    const dir = tempDir();
+    const destination = join(dir, "render.gif");
+    const existing = Buffer.from([0, 1, 2, 3, 255]);
+    writeFileSync(destination, existing);
+    const transaction = new ArtifactTransaction(destination, "file");
+    writeFileSync(transaction.stagingPath, "");
+
+    expect(() => transaction.commit()).toThrow("not a non-empty file");
+    transaction.rollback();
+
+    expect(readFileSync(destination)).toEqual(existing);
+    expect(existsSync(transaction.stagingPath)).toBe(false);
+    expect(transactionDirectories(dir)).toEqual([]);
+  });
+
+  it("keeps the existing file addressable when atomic promotion fails", () => {
+    const dir = tempDir();
+    const destination = join(dir, "render.mp4");
+    writeFileSync(destination, "existing");
+    let replacementCalls = 0;
+    const transaction = new ArtifactTransaction(destination, "file", {
+      existsSync,
+      renameSync(source, target) {
+        replacementCalls += 1;
+        expect(source).toBe(transaction.stagingPath);
+        expect(target).toBe(destination);
+        expect(readFileSync(destination, "utf8")).toBe("existing");
+        throw new Error("injected replacement failure");
+      },
+      rmSync,
+    });
+    writeFileSync(transaction.stagingPath, "new-render");
+
+    expect(() => transaction.commit()).toThrow("injected replacement failure");
+    expect(replacementCalls).toBe(1);
+    expect(readFileSync(destination, "utf8")).toBe("existing");
+
+    transaction.rollback();
+    expect(existsSync(transaction.stagingPath)).toBe(false);
+    expect(transactionDirectories(dir)).toEqual([]);
+  });
+
+  it("removes cancelled staging output without touching the destination", () => {
+    const dir = tempDir();
+    const destination = join(dir, "render.mp4");
+    writeFileSync(destination, "keep-me");
+    const transaction = new ArtifactTransaction(destination, "file");
+    writeFileSync(transaction.stagingPath, "partial-render");
+
+    transaction.rollback();
+
+    expect(readFileSync(destination, "utf8")).toBe("keep-me");
+    expect(existsSync(transaction.stagingPath)).toBe(false);
+    expect(transactionDirectories(dir)).toEqual([]);
+  });
+
+  it("promotes a validated PNG sequence as one directory artifact", () => {
+    const dir = tempDir();
+    const destination = join(dir, "frames");
+    mkdirSync(destination);
+    writeFileSync(join(destination, "frame_000001.png"), "old");
+    const transaction = new ArtifactTransaction(destination, "directory");
+    mkdirSync(transaction.stagingPath);
+    writeFileSync(join(transaction.stagingPath, "frame_000001.png"), "png-1");
+    writeFileSync(join(transaction.stagingPath, "frame_000002.png"), "png-2");
+
+    transaction.commit();
+
+    expect(readdirSync(destination).sort()).toEqual(["frame_000001.png", "frame_000002.png"]);
+    expect(readFileSync(join(destination, "frame_000001.png"), "utf8")).toBe("png-1");
+    expect(transactionDirectories(dir)).toEqual([]);
+  });
+
+  it("restores the previous PNG sequence when directory promotion fails", () => {
+    const dir = tempDir();
+    const destination = join(dir, "frames");
+    mkdirSync(destination);
+    writeFileSync(join(destination, "frame_000001.png"), "existing-frame");
+    let transaction: ArtifactTransaction;
+    transaction = new ArtifactTransaction(destination, "directory", {
+      existsSync,
+      renameSync(source, target) {
+        if (source === transaction.stagingPath && target === destination) {
+          throw new Error("injected directory promotion failure");
+        }
+        renamePathSync(source, target);
+      },
+      rmSync,
+    });
+    mkdirSync(transaction.stagingPath);
+    writeFileSync(join(transaction.stagingPath, "frame_000001.png"), "new-frame");
+
+    expect(() => transaction.commit()).toThrow("injected directory promotion failure");
+
+    expect(readFileSync(join(destination, "frame_000001.png"), "utf8")).toBe("existing-frame");
+    expect(transactionDirectories(dir)).toEqual([]);
+  });
+
+  it("does not delete a concurrent PNG sequence published during recovery", () => {
+    const dir = tempDir();
+    const destination = join(dir, "frames");
+    mkdirSync(destination);
+    writeFileSync(join(destination, "frame_000001.png"), "existing-frame");
+    const concurrentTransaction = new ArtifactTransaction(destination, "directory");
+    mkdirSync(concurrentTransaction.stagingPath);
+    writeFileSync(join(concurrentTransaction.stagingPath, "frame_000001.png"), "concurrent-frame");
+
+    let firstTransaction: ArtifactTransaction;
+    firstTransaction = new ArtifactTransaction(destination, "directory", {
+      existsSync,
+      renameSync(source, target) {
+        if (source === firstTransaction.stagingPath && target === destination) {
+          concurrentTransaction.commit();
+          expect(readFileSync(join(destination, "frame_000001.png"), "utf8")).toBe(
+            "concurrent-frame",
+          );
+          throw new Error("injected first promotion failure");
+        }
+        renamePathSync(source, target);
+      },
+      rmSync,
+    });
+    mkdirSync(firstTransaction.stagingPath);
+    writeFileSync(join(firstTransaction.stagingPath, "frame_000001.png"), "first-frame");
+
+    expect(() => firstTransaction.commit()).toThrow("injected first promotion failure");
+
+    expect(readFileSync(join(destination, "frame_000001.png"), "utf8")).toBe("concurrent-frame");
+    expect(transactionDirectories(dir)).toEqual([]);
+  });
+
+  it("rejects an empty PNG sequence and preserves the existing directory", () => {
+    const dir = tempDir();
+    const destination = join(dir, "frames");
+    mkdirSync(destination);
+    writeFileSync(join(destination, "frame_000001.png"), "existing-frame");
+    const transaction = new ArtifactTransaction(destination, "directory");
+    mkdirSync(transaction.stagingPath);
+
+    expect(() => transaction.commit()).toThrow("directory is empty");
+    transaction.rollback();
+
+    expect(readFileSync(join(destination, "frame_000001.png"), "utf8")).toBe("existing-frame");
+    expect(transactionDirectories(dir)).toEqual([]);
+  });
+});

packages/producer/src/services/render/artifactTransaction.ts

@@ -0,0 +1,191 @@
+import {
+  closeSync,
+  existsSync,
+  fstatSync,
+  mkdtempSync,
+  openSync,
+  readSync,
+  readdirSync,
+  renameSync,
+  rmSync,
+} from "node:fs";
+import { basename, dirname, extname, join, resolve } from "node:path";
+
+export type ArtifactKind = "file" | "directory";
+
+export interface ArtifactTransactionFileSystem {
+  existsSync(path: string): boolean;
+  renameSync(source: string, destination: string): void;
+  rmSync(path: string, options: { recursive: true; force: true }): void;
+}
+
+const defaultFileSystem: ArtifactTransactionFileSystem = {
+  existsSync,
+  renameSync,
+  rmSync,
+};
+
+function createSiblingTransactionDirectory(destination: string): string {
+  const parent = dirname(destination);
+  const extension = extname(destination);
+  const stem = extension ? basename(destination, extension) : basename(destination);
+  // mkdtemp reserves the directory atomically and creates it with private
+  // permissions. Keeping it beside the destination preserves same-filesystem
+  // rename semantics without exposing predictable files in a shared temp dir.
+  return mkdtempSync(join(parent, `.${stem}.hf-transaction-`));
+}
+
+function assertReadableNonEmptyFile(path: string): void {
+  // Validate the opened object, not a pathname checked before opening it.
+  // The descriptor pins the file across validation and closes the TOCTOU gap
+  // between lstat(path) and open(path).
+  const fd = openSync(path, "r");
+  try {
+    const stat = fstatSync(fd);
+    if (!stat.isFile() || stat.size <= 0) {
+      throw new Error(`Render artifact is not a non-empty file: ${path}`);
+    }
+    readSync(fd, Buffer.allocUnsafe(1), 0, 1, 0);
+  } finally {
+    closeSync(fd);
+  }
+}
+
+function collectDirectoryFiles(root: string): string[] {
+  const files: string[] = [];
+  const visit = (directory: string): void => {
+    for (const entry of readdirSync(directory, { withFileTypes: true })) {
+      const path = join(directory, entry.name);
+      if (entry.isDirectory()) visit(path);
+      else if (entry.isFile()) files.push(path);
+    }
+  };
+  visit(root);
+  return files;
+}
+
+/**
+ * Stages a render beside its final destination and promotes only a validated
+ * artifact. File promotion uses one atomic replacement rename, so an existing
+ * file remains addressable until the new file replaces it. Replacing a
+ * non-empty directory cannot be expressed as one portable rename; that case
+ * uses a recoverable backup handoff while preserving the previous contents on
+ * ordinary failures.
+ */
+export class ArtifactTransaction {
+  readonly destinationPath: string;
+  readonly stagingPath: string;
+  private readonly transactionDirectory: string;
+  private readonly backupPath: string;
+  private state: "active" | "committed" | "rolled-back" = "active";
+
+  constructor(
+    destinationPath: string,
+    private readonly kind: ArtifactKind,
+    private readonly fileSystem: ArtifactTransactionFileSystem = defaultFileSystem,
+  ) {
+    this.destinationPath = resolve(destinationPath);
+    this.transactionDirectory = createSiblingTransactionDirectory(this.destinationPath);
+    this.stagingPath = join(this.transactionDirectory, basename(this.destinationPath));
+    this.backupPath = join(this.transactionDirectory, "backup");
+  }
+
+  validate(): void {
+    if (this.kind === "file") {
+      assertReadableNonEmptyFile(this.stagingPath);
+      return;
+    }
+    let files: string[];
+    try {
+      files = collectDirectoryFiles(this.stagingPath);
+    } catch (error) {
+      throw new Error(`Render artifact is not a readable directory: ${this.stagingPath}`, {
+        cause: error,
+      });
+    }
+    if (files.length === 0) {
+      throw new Error(`Render artifact directory is empty: ${this.stagingPath}`);
+    }
+    for (const file of files) assertReadableNonEmptyFile(file);
+  }
+
+  commit(): void {
+    if (this.state !== "active") {
+      throw new Error(`Cannot commit an artifact transaction in state ${this.state}`);
+    }
+    this.validate();
+    const hadDestination = this.fileSystem.existsSync(this.destinationPath);
+
+    // Both files and new directories publish with one rename. In particular,
+    // do not move an existing file out of the way first: rename replaces it
+    // atomically, so readers see either the complete old file or the complete
+    // new file and a failed rename leaves the old file untouched.
+    if (this.kind === "file" || !hadDestination) {
+      this.fileSystem.renameSync(this.stagingPath, this.destinationPath);
+      this.state = "committed";
+      this.cleanupTransactionDirectory();
+      return;
+    }
+
+    // Portable Node filesystem APIs cannot atomically replace a non-empty
+    // directory. Keep the backup handoff for an existing PNG-sequence output;
+    // it provides recovery on ordinary errors, but not atomic visibility or
+    // crash recovery. Callers should publish directory outputs to a fresh path
+    // when they require the same visibility guarantee as file artifacts.
+    this.fileSystem.renameSync(this.destinationPath, this.backupPath);
+    try {
+      this.fileSystem.renameSync(this.stagingPath, this.destinationPath);
+      this.state = "committed";
+    } catch (error) {
+      // A competing transaction may have published while this destination was
+      // temporarily vacant. Never remove that caller-visible directory: only
+      // restore our backup while the destination remains unclaimed.
+      this.restoreBackupIfDestinationUnclaimed();
+      this.state = "rolled-back";
+      this.cleanupTransactionDirectory();
+      throw error;
+    }
+    // Promotion succeeded. Cleanup is best-effort: a stale private transaction
+    // directory is safer than reporting failure after publishing the artifact.
+    this.cleanupTransactionDirectory();
+  }
+
+  rollback(): void {
+    if (this.state !== "active") return;
+    this.fileSystem.rmSync(this.stagingPath, { recursive: true, force: true });
+    if (
+      this.fileSystem.existsSync(this.backupPath) &&
+      !this.fileSystem.existsSync(this.destinationPath)
+    ) {
+      this.fileSystem.renameSync(this.backupPath, this.destinationPath);
+    }
+    this.cleanupTransactionDirectory();
+    this.state = "rolled-back";
+  }
+
+  private restoreBackupIfDestinationUnclaimed(): void {
+    if (
+      !this.fileSystem.existsSync(this.backupPath) ||
+      this.fileSystem.existsSync(this.destinationPath)
+    ) {
+      return;
+    }
+    try {
+      this.fileSystem.renameSync(this.backupPath, this.destinationPath);
+    } catch (error) {
+      // Losing the rename race to a concurrent publisher is successful
+      // recovery: its complete artifact owns the destination. Propagate other
+      // restore failures so the caller can retry rollback with the backup kept.
+      if (!this.fileSystem.existsSync(this.destinationPath)) throw error;
+    }
+  }
+
+  private cleanupTransactionDirectory(): void {
+    try {
+      this.fileSystem.rmSync(this.transactionDirectory, { recursive: true, force: true });
+    } catch {
+      // Never turn a successfully published artifact into a failed render just
+      // because best-effort cleanup of its private transaction directory failed.
+    }
+  }
+}

packages/producer/src/services/renderOrchestrator.ts

@@ -95,6 +95,7 @@ import {
   OrderedRenderEventPublisher,
   publishRenderFailure,
 } from "./render/renderEventPublisher.js";
+import { ArtifactTransaction } from "./render/artifactTransaction.js";
 import { normalizeErrorMessage } from "../utils/errorMessage.js";
 import { formatCaptureFrameName } from "../utils/paths.js";
 import { resolveEffectiveHdrMode } from "./render/hdrMode.js";
@@ -1537,6 +1538,11 @@ export async function executeRenderJob(
   const isMov = outputFormat === "mov";
   const isPngSequence = outputFormat === "png-sequence";
   const isGif = outputFormat === "gif";
+  const artifactTransaction = new ArtifactTransaction(
+    outputPath,
+    isPngSequence ? "directory" : "file",
+  );
+  const stagedOutputPath = artifactTransaction.stagingPath;
   const needsAlpha = isWebm || isMov || isPngSequence;
   // `forceScreenshot` is resolved exactly once inside `compileStage` (alpha
   // output + composition `renderModeHints` are folded together there) and
@@ -2901,7 +2907,7 @@ export async function executeRenderJob(
             runEncodeStage({
               job,
               log,
-              outputPath,
+              outputPath: stagedOutputPath,
               framesDir,
               videoOnlyPath,
               width,
@@ -2959,7 +2965,7 @@ export async function executeRenderJob(
             job,
             videoOnlyPath,
             audioOutputPath,
-            outputPath,
+            outputPath: stagedOutputPath,
             hasAudio,
             abortSignal,
             assertNotAborted,
@@ -2971,10 +2977,7 @@ export async function executeRenderJob(
       observability.checkpoint("assemble", `skipped for ${outputFormat}`);
     }
 
-    // ── Complete ─────────────────────────────────────────────────────────
-    job.outputPath = outputPath;
-    updateJobStatus(job, "complete", "Render complete", 100, onProgress);
-    await eventPublisher.flush();
+    artifactTransaction.validate();
 
     const totalElapsed = Date.now() - pipelineStart;
 
@@ -2982,7 +2985,7 @@ export async function executeRenderJob(
     // Record transient-tab-death retry burn (recovered case) so it's visible on
     // dashboard 1783183, not just logs. The catch mirrors this for the failed case.
     recordTransientRetryObservability();
-    observability.checkpoint("pipeline", "completed", { totalElapsedMs: totalElapsed });
+    observability.checkpoint("pipeline", "artifact validated", { totalElapsedMs: totalElapsed });
     const observabilitySummary = observability.summary({
       lastBrowserConsole,
       capture: captureObservability,
@@ -3043,9 +3046,9 @@ export async function executeRenderJob(
       // easy access. Skipped for png-sequence: outputPath is a directory, not
       // a single file — the captured frames already live in `framesDir` under
       // workDir during a debug run anyway.
-      if (!isPngSequence && existsSync(outputPath)) {
+      if (!isPngSequence && existsSync(stagedOutputPath)) {
         const debugOutput = join(workDir, `output${videoExt}`);
-        copyFileSync(outputPath, debugOutput);
+        copyFileSync(stagedOutputPath, debugOutput);
       }
     } else if (process.env.KEEP_TEMP === "1") {
       log.info("KEEP_TEMP=1 — leaving workDir on disk for inspection", { workDir });
@@ -3058,7 +3061,13 @@ export async function executeRenderJob(
         log,
       );
     }
+
+    artifactTransaction.commit();
+    job.outputPath = outputPath;
+    updateJobStatus(job, "complete", "Render complete", 100, onProgress);
+    await eventPublisher.flush();
   } catch (error) {
+    await safeCleanup("rollback staged artifact", () => artifactTransaction.rollback(), log);
     if (error instanceof RenderCancelledError || abortSignal?.aborted) {
       job.error = error instanceof Error ? error.message : "render_cancelled";
       updateJobStatus(job, "cancelled", "Render cancelled", job.progress, onProgress);