refactor(engine): manage child process lifecycles (#2160)

* refactor(engine): manage child process lifecycles

* fix(engine): preserve child reaping after runtime errors

* fix(engine): untrack child processes on exit

packages/producer/src/services/animatedGifPrep.ts

@@ -1,6 +1,5 @@
 // fallow-ignore-file complexity
 import { createHash } from "node:crypto";
-import { spawn } from "node:child_process";
 import {
   copyFileSync,
   existsSync,
@@ -13,7 +12,7 @@ import {
 import { dirname, isAbsolute, join, resolve } from "node:path";
 import { parseHTML } from "linkedom";
 import { parseAnimatedGifMetadata, type AnimatedGifMetadata } from "@hyperframes/core";
-import { DEFAULT_VP9_CPU_USED, getFfmpegBinary } from "@hyperframes/engine";
+import { DEFAULT_VP9_CPU_USED, runFfmpeg } from "@hyperframes/engine";
 import { isHttpUrl } from "../utils/urlDownloader.js";
 
 const PREPARED_GIF_SUBDIR = "_animated_gif";
@@ -247,30 +246,16 @@ export function buildAnimatedGifTranscodeArgs(input: {
 }
 
 async function runAnimatedGifTranscode(request: AnimatedGifTranscodeRequest): Promise<void> {
-  await new Promise<void>((resolvePromise, reject) => {
-    const proc = spawn(getFfmpegBinary(), request.args);
-    let stderr = "";
-    const timeout = request.timeoutMs ?? 300_000;
-    const timer = setTimeout(() => {
-      proc.kill("SIGTERM");
-      reject(new Error(`Animated GIF transcode timed out after ${timeout}ms`));
-    }, timeout);
-    proc.stderr.on("data", (chunk: Buffer) => {
-      stderr += chunk.toString();
-    });
-    proc.on("close", (code) => {
-      clearTimeout(timer);
-      if (code === 0) {
-        resolvePromise();
-        return;
-      }
-      reject(new Error(`Animated GIF transcode failed (${code}): ${stderr.slice(-500)}`));
-    });
-    proc.on("error", (error) => {
-      clearTimeout(timer);
-      reject(error);
-    });
-  });
+  const timeout = request.timeoutMs ?? 300_000;
+  const result = await runFfmpeg(request.args, { timeout });
+  if (result.success) return;
+  if (result.terminationReason === "deadline") {
+    throw new Error(`Animated GIF transcode timed out after ${timeout}ms`);
+  }
+  throw (
+    result.error ??
+    new Error(`Animated GIF transcode failed (${result.exitCode}): ${result.stderr.slice(-500)}`)
+  );
 }
 
 async function ensurePreparedWebm(input: {

packages/producer/src/services/distributed/assemble.ts

@@ -296,6 +296,7 @@ export async function assemble(
         videoPath: postConcatPath,
         audioPath,
         outputPath: paddedAudioPath,
+        signal: abortSignal,
       });
       if (!padTrimResult.success) {
         throw new Error(`[assemble] audio pad/trim failed: ${padTrimResult.error}`);

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

@@ -13,7 +13,7 @@
  * No real ffmpeg/ffprobe runs in these tests.
  */
 
-import { describe, expect, it } from "bun:test";
+import { describe, expect, it, mock } from "bun:test";
 import {
   buildPadTrimAudioArgs,
   buildPadTrimAudioPlan,
@@ -198,6 +198,24 @@ describe("padOrTrimAudioToVideoFrameCount", () => {
     return { input, captured };
   }
 
+  it("passes the render abort signal to the audio metadata probe", async () => {
+    const controller = new AbortController();
+    const probeVideoFrameInfo = mock(async () => ({ frameCount: 30, fpsNum: 30, fpsDen: 1 }));
+    const probeAudioInfo = mock(async () => ({ durationSeconds: 1 }));
+
+    await padOrTrimAudioToVideoFrameCount({
+      videoPath: "/tmp/v.mp4",
+      audioPath: "/tmp/a.aac",
+      outputPath: "/tmp/o.aac",
+      signal: controller.signal,
+      probeVideoFrameInfo,
+      probeAudioInfo,
+      runFfmpeg: mock(async () => ({ success: true })),
+    });
+
+    expect(probeAudioInfo).toHaveBeenCalledWith("/tmp/a.aac", controller.signal);
+  });
+
   it("pads a video of N=180 frames at 30/1 fps with shorter audio", async () => {
     const { input, captured } = harness({
       video: { frameCount: 180, fpsNum: 30, fpsDen: 1 },

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

@@ -24,7 +24,9 @@ import {
   extractAudioMetadata,
   formatFfmpegError,
   getFfprobeBinary,
+  ManagedChildProcess,
   runFfmpeg,
+  trackChildProcess,
   type AudioMetadata,
 } from "@hyperframes/engine";
 
@@ -62,12 +64,13 @@ export interface PadTrimAudioInput {
   audioPath: string;
   /** Path the helper writes the duration-corrected audio to. */
   outputPath: string;
+  signal?: AbortSignal;
   /**
    * Optional injectables for unit tests. Production callers omit them and
    * get the real `ffprobe`/`ffmpeg`-backed implementations.
    */
   probeVideoFrameInfo?: (videoPath: string) => Promise<ProbeVideoFrameInfo>;
-  probeAudioInfo?: (audioPath: string) => Promise<AudioProbeInfo>;
+  probeAudioInfo?: (audioPath: string, signal?: AbortSignal) => Promise<AudioProbeInfo>;
   runFfmpeg?: (args: string[]) => Promise<{ success: boolean; error?: string }>;
 }
 
@@ -265,15 +268,17 @@ function concatFileLine(path: string): string {
 export async function padOrTrimAudioToVideoFrameCount(
   input: PadTrimAudioInput,
 ): Promise<PadTrimAudioResult> {
-  const probeVideo = input.probeVideoFrameInfo ?? defaultProbeVideoFrameInfo;
+  const probeVideo =
+    input.probeVideoFrameInfo ??
+    ((videoPath: string) => defaultProbeVideoFrameInfo(videoPath, input.signal));
   const probeAudio = input.probeAudioInfo ?? defaultProbeAudioInfo;
-  const runner = input.runFfmpeg ?? defaultRunFfmpeg;
+  const runner = input.runFfmpeg ?? ((args: string[]) => defaultRunFfmpeg(args, input.signal));
 
   // Probe video and audio in parallel — the two ffprobe invocations are
   // independent and account for most of this function's wall-clock time.
   const [videoResult, audioResult] = await Promise.allSettled([
     probeVideo(input.videoPath),
-    probeAudio(input.audioPath),
+    probeAudio(input.audioPath, input.signal),
   ]);
 
   if (videoResult.status === "rejected") {
@@ -393,39 +398,48 @@ interface FfprobeOutput {
   streams?: FfprobeStreamInfo[];
 }
 
-async function defaultProbeVideoFrameInfo(videoPath: string): Promise<ProbeVideoFrameInfo> {
+async function defaultProbeVideoFrameInfo(
+  videoPath: string,
+  signal?: AbortSignal,
+): Promise<ProbeVideoFrameInfo> {
   // Try the container header (`nb_frames`) first — single moov atom read,
   // no decode. Closed-GOP, B-frame-free streams (the only ones we'll ever
   // ask to pad/trim) reliably set it. Fall back to `-count_packets` which
   // walks the packet stream when the header doesn't carry the count.
-  const fastInfo = await runFfprobeJson<FfprobeOutput>([
-    "-v",
-    "error",
-    "-select_streams",
-    "v:0",
-    "-show_entries",
-    "stream=nb_frames,r_frame_rate",
-    "-of",
-    "json",
-    videoPath,
-  ]);
+  const fastInfo = await runFfprobeJson<FfprobeOutput>(
+    [
+      "-v",
+      "error",
+      "-select_streams",
+      "v:0",
+      "-show_entries",
+      "stream=nb_frames,r_frame_rate",
+      "-of",
+      "json",
+      videoPath,
+    ],
+    signal,
+  );
   let stream = fastInfo.streams?.[0];
   const fastCount = Number(stream?.nb_frames);
   if (stream && Number.isFinite(fastCount) && fastCount > 0) {
     return { frameCount: fastCount, ...parseFrameRate(stream.r_frame_rate ?? "") };
   }
-  const slowInfo = await runFfprobeJson<FfprobeOutput>([
-    "-v",
-    "error",
-    "-select_streams",
-    "v:0",
-    "-count_packets",
-    "-show_entries",
-    "stream=nb_read_packets,r_frame_rate",
-    "-of",
-    "json",
-    videoPath,
-  ]);
+  const slowInfo = await runFfprobeJson<FfprobeOutput>(
+    [
+      "-v",
+      "error",
+      "-select_streams",
+      "v:0",
+      "-count_packets",
+      "-show_entries",
+      "stream=nb_read_packets,r_frame_rate",
+      "-of",
+      "json",
+      videoPath,
+    ],
+    signal,
+  );
   stream = slowInfo.streams?.[0];
   if (!stream) throw new Error(`ffprobe found no video stream in ${videoPath}`);
   const slowCount = Number(stream.nb_read_packets);
@@ -445,10 +459,13 @@ function parseFrameRate(rate: string): { fpsNum: number; fpsDen: number } {
   return { fpsNum, fpsDen };
 }
 
-async function defaultProbeAudioInfo(audioPath: string): Promise<AudioProbeInfo> {
+async function defaultProbeAudioInfo(
+  audioPath: string,
+  signal?: AbortSignal,
+): Promise<AudioProbeInfo> {
   // The shared ffprobe wrapper derives AAC-LC duration from packet count so
-  // every consumer sees the same VBR-safe metadata.
-  const metadata: AudioMetadata = await extractAudioMetadata(audioPath);
+  // every consumer sees the same VBR-safe metadata while preserving cancellation.
+  const metadata: AudioMetadata = await extractAudioMetadata(audioPath, { signal });
   return {
     durationSeconds: metadata.durationSeconds,
     sampleRate: metadata.sampleRate,
@@ -457,8 +474,11 @@ async function defaultProbeAudioInfo(audioPath: string): Promise<AudioProbeInfo>
   };
 }
 
-async function defaultRunFfmpeg(args: string[]): Promise<{ success: boolean; error?: string }> {
-  const result = await runFfmpeg(args);
+async function defaultRunFfmpeg(
+  args: string[],
+  signal?: AbortSignal,
+): Promise<{ success: boolean; error?: string }> {
+  const result = await runFfmpeg(args, { signal });
   if (result.success) return { success: true };
   return {
     success: false,
@@ -468,34 +488,30 @@ async function defaultRunFfmpeg(args: string[]): Promise<{ success: boolean; err
 
 // ── ffprobe JSON runner (shared between fast/slow video probe paths) ─────
 
-function runFfprobeJson<T>(args: string[]): Promise<T> {
-  return new Promise((resolve, reject) => {
-    const proc = spawn(getFfprobeBinary(), args);
-    let stdout = "";
-    let stderr = "";
-    proc.stdout.on("data", (data: Buffer) => {
-      stdout += data.toString();
-    });
-    proc.stderr.on("data", (data: Buffer) => {
-      stderr += data.toString();
-    });
-    proc.on("error", (err) => {
-      if ((err as NodeJS.ErrnoException).code === "ENOENT") {
-        reject(new Error("[audioPadTrim] ffprobe not found. Please install FFmpeg."));
-      } else {
-        reject(err);
-      }
-    });
-    proc.on("close", (code) => {
-      if (code !== 0) {
-        reject(new Error(`ffprobe exited ${code}: ${stderr}`));
-        return;
-      }
-      try {
-        resolve(JSON.parse(stdout) as T);
-      } catch (err) {
-        reject(new Error(`Failed to parse ffprobe output: ${(err as Error).message}`));
-      }
-    });
+async function runFfprobeJson<T>(args: string[], signal?: AbortSignal): Promise<T> {
+  const proc = spawn(getFfprobeBinary(), args);
+  trackChildProcess(proc);
+  let stdout = "";
+  proc.stdout.on("data", (data: Buffer) => {
+    stdout += data.toString();
   });
+  const managed = new ManagedChildProcess(proc, {
+    signal,
+    deadlineAtMs: Date.now() + 30_000,
+  });
+  const outcome = await managed.wait();
+  if (outcome.reason === "spawn_error") {
+    if ((outcome.error as NodeJS.ErrnoException | undefined)?.code === "ENOENT") {
+      throw new Error("[audioPadTrim] ffprobe not found. Please install FFmpeg.");
+    }
+    throw outcome.error ?? new Error(outcome.stderr);
+  }
+  if (outcome.reason !== "exit" || outcome.exitCode !== 0) {
+    throw new Error(`ffprobe ${outcome.reason}: ${outcome.stderr}`);
+  }
+  try {
+    return JSON.parse(stdout) as T;
+  } catch (err) {
+    throw new Error(`Failed to parse ffprobe output: ${(err as Error).message}`);
+  }
 }