diff --git a/packages/producer/src/services/render/artifactTransaction.test.ts b/packages/producer/src/services/render/artifactTransaction.test.ts new file mode 100644 index 0000000000..943da4e9f2 --- /dev/null +++ b/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([]); + }); +}); diff --git a/packages/producer/src/services/render/artifactTransaction.ts b/packages/producer/src/services/render/artifactTransaction.ts new file mode 100644 index 0000000000..4213479a6c --- /dev/null +++ b/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. + } + } +} diff --git a/packages/producer/src/services/renderOrchestrator.ts b/packages/producer/src/services/renderOrchestrator.ts index 51ed3c7b3d..e6c2b2813c 100644 --- a/packages/producer/src/services/renderOrchestrator.ts +++ b/packages/producer/src/services/renderOrchestrator.ts @@ -94,6 +94,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"; @@ -1536,6 +1537,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 @@ -2879,7 +2885,7 @@ export async function executeRenderJob( runEncodeStage({ job, log, - outputPath, + outputPath: stagedOutputPath, framesDir, videoOnlyPath, width, @@ -2937,7 +2943,7 @@ export async function executeRenderJob( job, videoOnlyPath, audioOutputPath, - outputPath, + outputPath: stagedOutputPath, hasAudio, abortSignal, assertNotAborted, @@ -2949,10 +2955,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; @@ -2960,7 +2963,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, @@ -3021,9 +3024,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 }); @@ -3036,7 +3039,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);