refactor(engine): manage child process lifecycles

Author: jrusso1020

packages/engine/src/index.ts

@@ -230,6 +230,12 @@ export {
   type RunFfmpegOptions,
   type RunFfmpegResult,
 } from "./utils/runFfmpeg.js";
+export {
+  ManagedChildProcess,
+  type ManagedChildProcessOptions,
+  type ManagedChildProcessOutcome,
+  type ManagedProcessTerminationReason,
+} from "./utils/managedChildProcess.js";
 export {
   assertConfiguredFfmpegBinariesExist,
   getFfmpegBinary,

packages/engine/src/services/chunkEncoder.test.ts

@@ -94,6 +94,12 @@ async function flushMuxCodecResolution(): Promise<void> {
   await Promise.resolve();
 }
 
+async function flushManagedProcessResolution(): Promise<void> {
+  await Promise.resolve();
+  await Promise.resolve();
+  await Promise.resolve();
+}
+
 describe("ENCODER_PRESETS", () => {
   it("has draft, standard, and high presets", () => {
     expect(ENCODER_PRESETS).toHaveProperty("draft");
@@ -310,7 +316,7 @@ describe("encodeFramesChunkedConcat ffmpegEncodeTimeout", () => {
 
     expect(calls).toHaveLength(1);
     emitClose(calls[0]!.proc, 0);
-    await Promise.resolve();
+    await flushManagedProcessResolution();
 
     expect(calls).toHaveLength(2);
     const concatProc = calls[1]!.proc;
@@ -353,7 +359,7 @@ describe("encodeFramesChunkedConcat ffmpegEncodeTimeout", () => {
     expect(chunkProc.kill).not.toHaveBeenCalled();
 
     emitClose(chunkProc, 0);
-    await Promise.resolve();
+    await flushManagedProcessResolution();
 
     expect(calls).toHaveLength(2);
     const concatProc = calls[1]!.proc;

packages/engine/src/services/chunkEncoder.ts

@@ -6,10 +6,8 @@
  * Supports CPU (libx264) and GPU encoding.
  */
 
-import { spawn } from "child_process";
 import { copyFileSync, existsSync, mkdirSync, readdirSync, statSync, writeFileSync } from "fs";
 import { join, dirname, extname } from "path";
-import { trackChildProcess } from "../utils/processTracker.js";
 import { DEFAULT_CONFIG, type EngineConfig } from "../config.js";
 import {
   type GpuEncoder,
@@ -20,7 +18,6 @@ import {
 import { type HdrTransfer, getHdrEncoderColorParams } from "../utils/hdr.js";
 import { withEvenDimensionPad } from "../utils/evenDimensions.js";
 import { formatFfmpegError, runFfmpeg } from "../utils/runFfmpeg.js";
-import { getFfmpegBinary } from "../utils/ffmpegBinaries.js";
 import { extractAudioMetadata } from "../utils/ffprobe.js";
 import { type Fps, fpsToFfmpegArg } from "@hyperframes/core";
 import type { EncoderOptions, EncodeResult, MuxResult } from "./chunkEncoder.types.js";
@@ -484,82 +481,40 @@ export async function encodeFramesFromDir(
   const inputPath = join(framesDir, framePattern);
   const inputArgs = ["-framerate", fpsToFfmpegArg(options.fps), "-i", inputPath];
   const args = buildEncoderArgs(options, inputArgs, outputPath, gpuEncoder);
-
-  return new Promise((resolve) => {
-    const ffmpeg = spawn(getFfmpegBinary(), args);
-    trackChildProcess(ffmpeg);
-    let stderr = "";
-    const onAbort = () => {
-      ffmpeg.kill("SIGTERM");
+  const encodeTimeout = config?.ffmpegEncodeTimeout ?? DEFAULT_CONFIG.ffmpegEncodeTimeout;
+  const result = await runFfmpeg(args, { signal, timeout: encodeTimeout });
+  if (result.terminationReason === "abort") {
+    return {
+      success: false,
+      outputPath,
+      durationMs: result.durationMs,
+      framesEncoded: 0,
+      fileSize: 0,
+      error: "FFmpeg encode cancelled",
     };
-    if (signal) {
-      if (signal.aborted) {
-        ffmpeg.kill("SIGTERM");
-      } else {
-        signal.addEventListener("abort", onAbort, { once: true });
-      }
-    }
-
-    const encodeTimeout = config?.ffmpegEncodeTimeout ?? DEFAULT_CONFIG.ffmpegEncodeTimeout;
-    let timedOut = false;
-    const timer = setTimeout(() => {
-      timedOut = true;
-      ffmpeg.kill("SIGTERM");
-    }, encodeTimeout);
-
-    ffmpeg.stderr.on("data", (data) => {
-      stderr += data.toString();
-    });
-
-    ffmpeg.on("close", (code) => {
-      clearTimeout(timer);
-      if (signal) signal.removeEventListener("abort", onAbort);
-      const durationMs = Date.now() - startTime;
-      if (signal?.aborted && !timedOut) {
-        resolve({
-          success: false,
-          outputPath,
-          durationMs,
-          framesEncoded: 0,
-          fileSize: 0,
-          error: "FFmpeg encode cancelled",
-        });
-        return;
-      }
-
-      if (code !== 0 || timedOut) {
-        resolve({
-          success: false,
-          outputPath,
-          durationMs,
-          framesEncoded: 0,
-          fileSize: 0,
-          error: appendEncodeTimeoutMessage(
-            formatFfmpegError(code, stderr),
-            timedOut,
-            encodeTimeout,
-          ),
-        });
-        return;
-      }
-
-      const fileSize = existsSync(outputPath) ? statSync(outputPath).size : 0;
-      resolve({ success: true, outputPath, durationMs, framesEncoded: frameCount, fileSize });
-    });
-
-    ffmpeg.on("error", (err) => {
-      clearTimeout(timer);
-      if (signal) signal.removeEventListener("abort", onAbort);
-      resolve({
-        success: false,
-        outputPath,
-        durationMs: Date.now() - startTime,
-        framesEncoded: 0,
-        fileSize: 0,
-        error: appendEncodeTimeoutMessage(`[FFmpeg] ${err.message}`, timedOut, encodeTimeout),
-      });
-    });
-  });
+  }
+  if (!result.success) {
+    return {
+      success: false,
+      outputPath,
+      durationMs: result.durationMs,
+      framesEncoded: 0,
+      fileSize: 0,
+      error: appendEncodeTimeoutMessage(
+        formatFfmpegError(result.exitCode, result.stderr),
+        result.terminationReason === "deadline",
+        encodeTimeout,
+      ),
+    };
+  }
+  const fileSize = existsSync(outputPath) ? statSync(outputPath).size : 0;
+  return {
+    success: true,
+    outputPath,
+    durationMs: Date.now() - startTime,
+    framesEncoded: frameCount,
+    fileSize,
+  };
 }
 
 export async function encodeFramesChunkedConcat(
@@ -624,45 +579,18 @@ export async function encodeFramesChunkedConcat(
     let gpuEncoder: GpuEncoder = null;
     if (options.useGpu) gpuEncoder = await getCachedGpuEncoder();
     const args = buildEncoderArgs(options, inputArgs, chunkPath, gpuEncoder);
-    const chunkResult = await new Promise<{ success: boolean; error?: string }>((resolve) => {
-      const ffmpeg = spawn(getFfmpegBinary(), args);
-      trackChildProcess(ffmpeg);
-      let stderr = "";
-      const encodeTimeout = config?.ffmpegEncodeTimeout ?? DEFAULT_CONFIG.ffmpegEncodeTimeout;
-      let timedOut = false;
-      const timer = setTimeout(() => {
-        timedOut = true;
-        ffmpeg.kill("SIGTERM");
-      }, encodeTimeout);
-      ffmpeg.stderr.on("data", (d) => {
-        stderr += d.toString();
-      });
-      ffmpeg.on("close", (code) => {
-        clearTimeout(timer);
-        if (code === 0 && !timedOut) resolve({ success: true });
-        else {
-          resolve({
-            success: false,
-            error: appendEncodeTimeoutMessage(
-              `Chunk ${i} encode failed: ${stderr.slice(-400)}`,
-              timedOut,
-              encodeTimeout,
-            ),
-          });
-        }
-      });
-      ffmpeg.on("error", (err) => {
-        clearTimeout(timer);
-        resolve({
-          success: false,
-          error: appendEncodeTimeoutMessage(
-            `Chunk ${i} encode error: ${err.message}`,
-            timedOut,
+    const encodeTimeout = config?.ffmpegEncodeTimeout ?? DEFAULT_CONFIG.ffmpegEncodeTimeout;
+    const processResult = await runFfmpeg(args, { signal, timeout: encodeTimeout });
+    const chunkResult = {
+      success: processResult.success,
+      error: processResult.success
+        ? undefined
+        : appendEncodeTimeoutMessage(
+            `Chunk ${i} encode failed: ${processResult.stderr.slice(-400)}`,
+            processResult.terminationReason === "deadline",
             encodeTimeout,
           ),
-        });
-      });
-    });
+    };
     if (!chunkResult.success) {
       return {
         success: false,
@@ -692,45 +620,18 @@ export async function encodeFramesChunkedConcat(
     "-y",
     outputPath,
   ];
-  const concatResult = await new Promise<{ success: boolean; error?: string }>((resolve) => {
-    const ffmpeg = spawn(getFfmpegBinary(), concatArgs);
-    trackChildProcess(ffmpeg);
-    let stderr = "";
-    const encodeTimeout = config?.ffmpegEncodeTimeout ?? DEFAULT_CONFIG.ffmpegEncodeTimeout;
-    let timedOut = false;
-    const timer = setTimeout(() => {
-      timedOut = true;
-      ffmpeg.kill("SIGTERM");
-    }, encodeTimeout);
-    ffmpeg.stderr.on("data", (d) => {
-      stderr += d.toString();
-    });
-    ffmpeg.on("close", (code) => {
-      clearTimeout(timer);
-      if (code === 0 && !timedOut) resolve({ success: true });
-      else {
-        resolve({
-          success: false,
-          error: appendEncodeTimeoutMessage(
-            `Chunk concat failed: ${stderr.slice(-400)}`,
-            timedOut,
-            encodeTimeout,
-          ),
-        });
-      }
-    });
-    ffmpeg.on("error", (err) => {
-      clearTimeout(timer);
-      resolve({
-        success: false,
-        error: appendEncodeTimeoutMessage(
-          `Chunk concat error: ${err.message}`,
-          timedOut,
+  const encodeTimeout = config?.ffmpegEncodeTimeout ?? DEFAULT_CONFIG.ffmpegEncodeTimeout;
+  const concatProcessResult = await runFfmpeg(concatArgs, { signal, timeout: encodeTimeout });
+  const concatResult = {
+    success: concatProcessResult.success,
+    error: concatProcessResult.success
+      ? undefined
+      : appendEncodeTimeoutMessage(
+          `Chunk concat failed: ${concatProcessResult.stderr.slice(-400)}`,
+          concatProcessResult.terminationReason === "deadline",
           encodeTimeout,
         ),
-      });
-    });
-  });
+  };
 
   if (!concatResult.success) {
     return {

packages/engine/src/services/streamingEncoder.test.ts

@@ -818,7 +818,7 @@ describe("spawnStreamingEncoder lifecycle and cleanup", () => {
     await expect(resolveWithin(writePromise)).resolves.toBe(false);
     expect(encoder.getExitStatus()).toBe("error");
     expect(proc.stdin.listenerCount("drain")).toBe(0);
-    expect(proc.listenerCount("close")).toBe(baselineCloseListeners);
+    expect(proc.listenerCount("close")).toBeLessThanOrEqual(baselineCloseListeners);
 
     const result = await encoder.close();
     expect(result.success).toBe(false);