fix(render): publish artifacts atomically

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.
+    }
+  }
+}