fix(output): write atomically so a failed re-run cannot destroy a good file

src/codecs/image.ts

@@ -1,7 +1,8 @@
 import sharp from "sharp";
-import { writeFile, mkdir } from "node:fs/promises";
-import { dirname, extname } from "node:path";
+import { writeFile } from "node:fs/promises";
+import { extname } from "node:path";
 import { CompressorError } from "../core/errors.js";
+import { withAtomicOutput } from "../core/atomic-output.js";
 import { sniffFile } from "../core/sniff.js";
 import { isInputOnlyImageFormat } from "../types/image-formats.js";
 import {
@@ -108,8 +109,10 @@ export async function encodeImage(
   return {
     bytes: buffer.byteLength,
     write: async () => {
-      await mkdir(dirname(outputPath), { recursive: true });
-      await writeFile(outputPath, buffer);
+      await withAtomicOutput(outputPath, async (temporaryPath) => {
+        await writeFile(temporaryPath, buffer);
+        return { value: undefined, replace: true };
+      });
     },
   };
 }

src/codecs/video.ts

@@ -1,5 +1,4 @@
-import { mkdir, stat, rm } from "node:fs/promises";
-import { dirname } from "node:path";
+import { stat } from "node:fs/promises";
 import { runFfmpeg, type FfmpegTools } from "./ffmpeg.js";
 import { CompressorError } from "../core/errors.js";
 import {
@@ -388,27 +387,19 @@ export interface EncodeVideoResult {
   readonly bytes: number;
 }
 
-/** Run one encode, cleaning up the partial file if anything goes wrong. */
+/** Run one encode into the caller-owned staging path. */
 export async function encodeVideo(
   params: EncodeVideoParams,
 ): Promise<EncodeVideoResult> {
   const { tools, outputPath } = params;
 
-  await mkdir(dirname(outputPath), { recursive: true });
-
-  try {
-    await runFfmpeg({
-      ffmpeg: tools.ffmpeg,
-      args: params.args,
-      durationSeconds: params.durationSeconds ?? null,
-      ...(params.onProgress ? { onProgress: params.onProgress } : {}),
-      ...(params.signal ? { signal: params.signal } : {}),
-    });
-  } catch (err) {
-    // A half-written file is worse than none: it looks like a successful run.
-    await rm(outputPath, { force: true }).catch(() => undefined);
-    throw err;
-  }
+  await runFfmpeg({
+    ffmpeg: tools.ffmpeg,
+    args: params.args,
+    durationSeconds: params.durationSeconds ?? null,
+    ...(params.onProgress ? { onProgress: params.onProgress } : {}),
+    ...(params.signal ? { signal: params.signal } : {}),
+  });
 
   const info = await stat(outputPath);
   return { bytes: info.size };

src/core/atomic-output.ts

@@ -0,0 +1,37 @@
+import { randomUUID } from "node:crypto";
+import { mkdir, rename, rm } from "node:fs/promises";
+import { basename, dirname, extname, join } from "node:path";
+
+export interface AtomicOutputDecision<T> {
+  readonly value: T;
+  readonly replace: boolean;
+}
+
+/**
+ * Produce a file beside its destination and replace the destination atomically.
+ *
+ * The leading dot keeps discovery from treating an in-flight encode as input
+ * during a concurrent or later run. Keeping the real extension last also lets
+ * tools such as ffmpeg infer the correct output format.
+ */
+export function temporaryOutputPath(outputPath: string): string {
+  const extension = extname(outputPath);
+  const stem = basename(outputPath, extension);
+  return join(dirname(outputPath), `.${stem}-${randomUUID()}.tmp${extension}`);
+}
+
+export async function withAtomicOutput<T>(
+  outputPath: string,
+  produce: (temporaryPath: string) => Promise<AtomicOutputDecision<T>>,
+): Promise<T> {
+  await mkdir(dirname(outputPath), { recursive: true });
+  const temporaryPath = temporaryOutputPath(outputPath);
+
+  try {
+    const decision = await produce(temporaryPath);
+    if (decision.replace) await rename(temporaryPath, outputPath);
+    return decision.value;
+  } finally {
+    await rm(temporaryPath, { force: true }).catch(() => undefined);
+  }
+}

src/core/compress.ts

@@ -1,9 +1,10 @@
-import { access, stat, rm } from "node:fs/promises";
+import { access, stat } from "node:fs/promises";
 import { join, relative, resolve, basename, extname, dirname } from "node:path";
 
 import { discoverFiles, type DiscoveredFile } from "./discover.js";
 import { mapWithConcurrency, defaultConcurrency } from "./pool.js";
 import { CompressorError, toFailure } from "./errors.js";
+import { withAtomicOutput } from "./atomic-output.js";
 import { encodeImage, resolveImageTarget } from "../codecs/image.js";
 import {
   encodeVideo,
@@ -393,7 +394,8 @@ async function runVideoJob(
   job: CompressionJob,
   ctx: ExecuteContext,
 ): Promise<JobOutput> {
-  if (!ctx.tools) {
+  const tools = ctx.tools;
+  if (!tools) {
     throw new CompressorError(
       "FFMPEG_NOT_FOUND",
       "ffmpeg is required to compress video.",
@@ -412,30 +414,37 @@ async function runVideoJob(
     ? planStreams(extension, probe)
     : { plan: openStreamPlan(probe), dropped: [] as string[] };
 
-  const args = isVideoContainer(extension)
-    ? curatedArgsFor(extension, job, ctx, probe, plan.plan)
-    : await openArgsFor(extension, job, ctx, probe, plan.plan);
-
-  const encoded = await encodeVideo({
-    tools: ctx.tools,
-    inputPath: job.inputPath,
-    outputPath: job.outputPath,
-    args,
-    durationSeconds: probe.durationSeconds,
-    ...(ctx.options.onProgress
-      ? {
-          onProgress: (ratio: number) =>
-            ctx.options.onProgress?.({ type: "job-progress", job, ratio }),
-        }
-      : {}),
-    ...(ctx.options.signal ? { signal: ctx.options.signal } : {}),
+  return withAtomicOutput<JobOutput>(job.outputPath, async (temporaryPath) => {
+    const args = isVideoContainer(extension)
+      ? curatedArgsFor(extension, job, ctx, probe, plan.plan, temporaryPath)
+      : await openArgsFor(extension, job, ctx, probe, plan.plan, temporaryPath);
+
+    const encoded = await encodeVideo({
+      tools,
+      inputPath: job.inputPath,
+      outputPath: temporaryPath,
+      args,
+      durationSeconds: probe.durationSeconds,
+      ...(ctx.options.onProgress
+        ? {
+            onProgress: (ratio: number) =>
+              ctx.options.onProgress?.({ type: "job-progress", job, ratio }),
+          }
+        : {}),
+      ...(ctx.options.signal ? { signal: ctx.options.signal } : {}),
+    });
+
+    if (ctx.skipLarger && encoded.bytes >= job.inputBytes) {
+      return {
+        value: { bytes: null, warnings: plan.dropped },
+        replace: false,
+      };
+    }
+    return {
+      value: { bytes: encoded.bytes, warnings: plan.dropped },
+      replace: true,
+    };
   });
-
-  if (ctx.skipLarger && encoded.bytes >= job.inputBytes) {
-    await rm(job.outputPath, { force: true }).catch(() => undefined);
-    return { bytes: null, warnings: plan.dropped };
-  }
-  return { bytes: encoded.bytes, warnings: plan.dropped };
 }
 
 const EMPTY_PROBE: MediaProbe = {
@@ -523,13 +532,14 @@ function curatedArgsFor(
   ctx: ExecuteContext,
   probe: MediaProbe,
   plan: StreamPlan | null,
+  outputPath: string,
 ): string[] {
   const videoCodec = resolveVideoCodec(container, ctx.options.videoCodec);
   const audioCodec = resolveAudioCodec(container, ctx.options.audioCodec, probe.audio);
 
   return curatedArgs({
     inputPath: job.inputPath,
-    outputPath: job.outputPath,
+    outputPath,
     container,
     // Checked against the container's own matrix by the resolvers above.
     videoCodec: videoCodec,
@@ -557,6 +567,7 @@ async function openArgsFor(
   ctx: ExecuteContext,
   probe: MediaProbe,
   plan: StreamPlan | null,
+  outputPath: string,
 ): Promise<string[]> {
   const ffmpeg = ctx.tools?.ffmpeg ?? "ffmpeg";
   const muxer = extension.slice(1);
@@ -586,7 +597,7 @@ async function openArgsFor(
 
   return buildOpenVideoArgs({
     inputPath: job.inputPath,
-    outputPath: job.outputPath,
+    outputPath,
     extension,
     // null means "let ffmpeg decide", which is always a legal choice.
     videoCodec: requested ?? null,

src/core/pool.ts

@@ -34,7 +34,15 @@ export async function mapWithConcurrency<T, R>(
     }
   }
 
-  await Promise.all(Array.from({ length: workers }, () => worker()));
+  // An aborted sibling may still be unwinding an encoder. Waiting for every
+  // active worker ensures its output cleanup finishes before the run rejects.
+  const settled = await Promise.allSettled(
+    Array.from({ length: workers }, () => worker()),
+  );
+  const rejected = settled.find(
+    (result): result is PromiseRejectedResult => result.status === "rejected",
+  );
+  if (rejected) throw rejected.reason;
   return results;
 }
 

test/atomic-output.test.ts

@@ -0,0 +1,86 @@
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { readFile, readdir, writeFile } from "node:fs/promises";
+import { basename, dirname, join } from "node:path";
+
+import { tempDir } from "./helpers.js";
+import { temporaryOutputPath, withAtomicOutput } from "../src/core/atomic-output.js";
+
+describe("atomic output lifecycle", () => {
+  let dir: string;
+  let cleanup: () => Promise<void>;
+
+  beforeAll(async () => {
+    ({ dir, cleanup } = await tempDir());
+  });
+  afterAll(() => cleanup());
+
+  it("uses unique hidden names that keep the destination extension last", () => {
+    const output = join(dir, "hero.mp4");
+    const first = temporaryOutputPath(output);
+    const second = temporaryOutputPath(output);
+
+    expect(dirname(first)).toBe(dir);
+    expect(basename(first)).toMatch(/^\.hero-.+\.tmp\.mp4$/);
+    expect(first).not.toBe(second);
+  });
+
+  it("atomically replaces the destination after a successful write", async () => {
+    const output = join(dir, "success", "hero.webp");
+    await withAtomicOutput(output, async (temporaryPath) => {
+      await writeFile(temporaryPath, "new bytes");
+      return { value: undefined, replace: true };
+    });
+
+    expect(await readFile(output, "utf8")).toBe("new bytes");
+    expect(await readdir(dirname(output))).toEqual(["hero.webp"]);
+  });
+
+  it("preserves the destination and removes the temp file after failure", async () => {
+    const output = join(dir, "failure", "hero.webp");
+    await withAtomicOutput(output, async (temporaryPath) => {
+      await writeFile(temporaryPath, "old bytes");
+      return { value: undefined, replace: true };
+    });
+
+    await expect(
+      withAtomicOutput(output, async (temporaryPath) => {
+        await writeFile(temporaryPath, "partial bytes");
+        throw new Error("encode failed");
+      }),
+    ).rejects.toThrow("encode failed");
+
+    expect(await readFile(output, "utf8")).toBe("old bytes");
+    expect(await readdir(dirname(output))).toEqual(["hero.webp"]);
+  });
+
+  it("preserves the destination when the caller declines replacement", async () => {
+    const output = join(dir, "skipped", "hero.webp");
+    await withAtomicOutput(output, async (temporaryPath) => {
+      await writeFile(temporaryPath, "old bytes");
+      return { value: undefined, replace: true };
+    });
+
+    await withAtomicOutput(output, async (temporaryPath) => {
+      await writeFile(temporaryPath, "larger bytes");
+      return { value: undefined, replace: false };
+    });
+
+    expect(await readFile(output, "utf8")).toBe("old bytes");
+    expect(await readdir(dirname(output))).toEqual(["hero.webp"]);
+  });
+
+  it("removes the temp file when the operation is aborted", async () => {
+    const output = join(dir, "aborted", "hero.webp");
+    const abortError = new Error("aborted");
+    abortError.name = "AbortError";
+
+    await expect(
+      withAtomicOutput(output, async (temporaryPath) => {
+        await writeFile(temporaryPath, "partial bytes");
+        throw abortError;
+      }),
+    ).rejects.toBe(abortError);
+
+    expect(await readdir(dirname(output))).toEqual([]);
+  });
+});