From 4e86382e2af72f93d2a9e4b103ce508fdbd260e5 Mon Sep 17 00:00:00 2001 From: James Date: Fri, 10 Jul 2026 12:37:30 -0700 Subject: [PATCH] refactor(engine): manage child process lifecycles --- packages/engine/src/index.ts | 6 + .../engine/src/services/chunkEncoder.test.ts | 10 +- packages/engine/src/services/chunkEncoder.ts | 205 +++++------------- .../src/services/streamingEncoder.test.ts | 2 +- .../engine/src/services/streamingEncoder.ts | 88 +++----- .../src/services/videoFrameExtractor.ts | 115 ++++------ packages/engine/src/utils/ffprobe.ts | 88 ++++---- packages/engine/src/utils/gpuEncoder.ts | 88 +++----- .../src/utils/managedChildProcess.test.ts | 102 +++++++++ .../engine/src/utils/managedChildProcess.ts | 178 +++++++++++++++ packages/engine/src/utils/runFfmpeg.ts | 75 ++----- .../producer/src/services/animatedGifPrep.ts | 37 +--- .../src/services/distributed/assemble.ts | 1 + .../src/services/render/audioPadTrim.test.ts | 20 +- .../src/services/render/audioPadTrim.ts | 182 ++++++++-------- 15 files changed, 638 insertions(+), 559 deletions(-) create mode 100644 packages/engine/src/utils/managedChildProcess.test.ts create mode 100644 packages/engine/src/utils/managedChildProcess.ts diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index b00b991637..dc5f0a4243 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -228,6 +228,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, diff --git a/packages/engine/src/services/chunkEncoder.test.ts b/packages/engine/src/services/chunkEncoder.test.ts index 4014afd386..a602388c11 100644 --- a/packages/engine/src/services/chunkEncoder.test.ts +++ b/packages/engine/src/services/chunkEncoder.test.ts @@ -94,6 +94,12 @@ async function flushMuxCodecResolution(): Promise { await Promise.resolve(); } +async function flushManagedProcessResolution(): Promise { + 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; diff --git a/packages/engine/src/services/chunkEncoder.ts b/packages/engine/src/services/chunkEncoder.ts index 44f3e68ab3..6254fe1509 100644 --- a/packages/engine/src/services/chunkEncoder.ts +++ b/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"; @@ -476,82 +473,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( @@ -616,45 +571,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, @@ -684,45 +612,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 { diff --git a/packages/engine/src/services/streamingEncoder.test.ts b/packages/engine/src/services/streamingEncoder.test.ts index cd99c7554d..01a6cf79e1 100644 --- a/packages/engine/src/services/streamingEncoder.test.ts +++ b/packages/engine/src/services/streamingEncoder.test.ts @@ -805,7 +805,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); diff --git a/packages/engine/src/services/streamingEncoder.ts b/packages/engine/src/services/streamingEncoder.ts index 109cf4041b..b08f55e4b1 100644 --- a/packages/engine/src/services/streamingEncoder.ts +++ b/packages/engine/src/services/streamingEncoder.ts @@ -16,6 +16,10 @@ import { spawn, type ChildProcess } from "child_process"; import { once } from "events"; import { trackChildProcess } from "../utils/processTracker.js"; +import { + ManagedChildProcess, + type ManagedProcessTerminationReason, +} from "../utils/managedChildProcess.js"; import { existsSync, mkdirSync, statSync } from "fs"; import { dirname } from "path"; @@ -439,7 +443,6 @@ export async function spawnStreamingEncoder( const args = buildStreamingArgs(options, outputPath, gpuEncoder); - const startTime = Date.now(); const ffmpeg: ChildProcess = spawn(getFfmpegBinary(), args, { stdio: ["pipe", "pipe", "pipe"], }); @@ -448,43 +451,11 @@ export async function spawnStreamingEncoder( let exitStatus: "running" | "success" | "error" = "running"; let stderr = ""; let exitCode: number | null = null; - let exitPromiseResolve: ((value: void) => void) | null = null; - const exitPromise = new Promise((resolve) => (exitPromiseResolve = resolve)); - - // Track stderr for progress and error messages - ffmpeg.stderr?.on("data", (data: Buffer) => { - stderr += data.toString(); - }); - - ffmpeg.on("close", (code: number | null) => { - exitCode = code; - exitStatus = code === 0 ? "success" : "error"; - exitPromiseResolve?.(); - }); - - ffmpeg.on("error", (err: Error) => { - exitStatus = "error"; - stderr += `\nProcess error: ${err.message}`; - exitPromiseResolve?.(); - }); + let terminationReason: ManagedProcessTerminationReason = "exit"; ffmpeg.stdin?.on("error", () => {}); ffmpeg.stdout?.on("error", () => {}); - // Handle abort signal - const onAbort = () => { - if (exitStatus === "running") { - ffmpeg.kill("SIGTERM"); - } - }; - if (signal) { - if (signal.aborted) { - ffmpeg.kill("SIGTERM"); - } else { - signal.addEventListener("abort", onAbort, { once: true }); - } - } - // Inactivity timeout: fires only when no frame has been written for // `ffmpegStreamingTimeout` ms. A slow-but-progressing capture (e.g. a CI // runner under load) keeps resetting the timer on each writeFrame, so total @@ -495,16 +466,17 @@ export async function spawnStreamingEncoder( // libx264 printed its summary and exited 255, observable as // "Streaming encode failed: FFmpeg exited with code 255" with audio:0kB). const streamingTimeout = config?.ffmpegStreamingTimeout ?? DEFAULT_CONFIG.ffmpegStreamingTimeout; - let timer: NodeJS.Timeout | null = null; - const resetTimer = () => { - if (timer) clearTimeout(timer); - timer = setTimeout(() => { - if (exitStatus === "running") { - ffmpeg.kill("SIGTERM"); - } - }, streamingTimeout); - }; - resetTimer(); + const managed = new ManagedChildProcess(ffmpeg, { + signal, + inactivityTimeoutMs: streamingTimeout, + }); + const exitPromise = managed.wait().then((outcome) => { + exitCode = outcome.exitCode; + stderr = outcome.stderr; + terminationReason = outcome.reason; + exitStatus = outcome.reason === "exit" && outcome.exitCode === 0 ? "success" : "error"; + return outcome; + }); const waitForDrainOrExit = async ( stdin: NonNullable, @@ -531,7 +503,7 @@ export async function spawnStreamingEncoder( throw err; }); - if (exitStatus !== "running") { + if (managed.isSettled || exitStatus !== "running") { return "exit"; } @@ -565,7 +537,7 @@ export async function spawnStreamingEncoder( // before draining, waitForDrainOrExit returns "exit", removes its // one-shot listeners, and callers see `false` instead of hanging. if (accepted) { - resetTimer(); + managed.markActivity(); return true; } @@ -573,7 +545,7 @@ export async function spawnStreamingEncoder( if (drainResult !== "drain" || exitStatus !== "running") { return false; } - resetTimer(); + managed.markActivity(); return true; }, @@ -582,9 +554,6 @@ export async function spawnStreamingEncoder( // path tracks an `encoderClosed` flag and may still re-call close() in // the outer finally if the inner cleanup raised before the flag flipped. // Each step here must be safe to repeat: - // - clearTimeout: safe to call on an already-cleared/fired timer - // - removeEventListener: no-op if the listener was already removed - // (and {once: true} would have removed it on the first abort anyway) // - stdin.end gated on !destroyed: skipped on the second call // - exitPromise: a single shared Promise; awaiting an already-resolved // Promise resolves immediately with the same captured exitCode @@ -592,12 +561,6 @@ export async function spawnStreamingEncoder( // repeated calls. If you change this method, preserve idempotency or // a regression here will silently double-close ffmpeg and produce // harder-to-trace errors at the orchestrator layer. - if (timer) { - clearTimeout(timer); - timer = null; - } - if (signal) signal.removeEventListener("abort", onAbort); - const stdin = ffmpeg.stdin; if (stdin && !stdin.destroyed) { await new Promise((resolve) => { @@ -605,11 +568,10 @@ export async function spawnStreamingEncoder( }); } - await exitPromise; - - const durationMs = Date.now() - startTime; + const outcome = await exitPromise; + const durationMs = outcome.durationMs; - if (signal?.aborted) { + if (terminationReason === "abort") { return { success: false, durationMs, @@ -619,11 +581,15 @@ export async function spawnStreamingEncoder( } if (exitCode !== 0) { + const inactivitySuffix = + terminationReason === "inactivity" + ? `\nFFmpeg stopped after ${streamingTimeout} ms without consuming a frame.` + : ""; return { success: false, durationMs, fileSize: 0, - error: formatFfmpegError(exitCode, stderr), + error: `${formatFfmpegError(exitCode, stderr)}${inactivitySuffix}`, }; } diff --git a/packages/engine/src/services/videoFrameExtractor.ts b/packages/engine/src/services/videoFrameExtractor.ts index 33c7473b71..a44dd1b294 100644 --- a/packages/engine/src/services/videoFrameExtractor.ts +++ b/packages/engine/src/services/videoFrameExtractor.ts @@ -6,12 +6,10 @@ * Videos are replaced with elements during capture. */ -import { spawn } from "child_process"; import { copyFileSync, existsSync, linkSync, mkdirSync, readdirSync, rmSync } from "fs"; import { isAbsolute, join, posix, resolve, sep } from "path"; import { parseHTML } from "linkedom"; import { decodeUrlPathVariants, MEDIA_DURATION_CLAMP_EPSILON_SECONDS } from "@hyperframes/core"; -import { trackChildProcess } from "../utils/processTracker.js"; import { resolveReferencedStart, type RefResolverEl } from "./referenceResolver.js"; import { extractMediaMetadata, type VideoMetadata } from "../utils/ffprobe.js"; import { @@ -20,7 +18,7 @@ import { type HdrTransfer, } from "../utils/hdr.js"; import { downloadToTemp, isHttpUrl } from "../utils/urlDownloader.js"; -import { getFfmpegBinary } from "../utils/ffmpegBinaries.js"; +import { runFfmpeg } from "../utils/runFfmpeg.js"; import { DEFAULT_CONFIG, type EngineConfig } from "../config.js"; import { unwrapTemplate } from "../utils/htmlTemplate.js"; import { @@ -328,78 +326,51 @@ export async function extractVideoFramesRange( if (format === "png") args.push("-compression_level", "1"); args.push("-y", outputPattern); - return new Promise((resolve, reject) => { - const ffmpeg = spawn(getFfmpegBinary(), args); - trackChildProcess(ffmpeg); - let stderr = ""; - const onAbort = () => { - ffmpeg.kill("SIGTERM"); - }; - if (signal) { - if (signal.aborted) { - ffmpeg.kill("SIGTERM"); - } else { - signal.addEventListener("abort", onAbort, { once: true }); - } + const processResult = await runFfmpeg(args, { signal, timeout: ffmpegProcessTimeout }); + if (processResult.terminationReason === "abort") { + throw new Error("Video frame extraction cancelled"); + } + if (processResult.terminationReason === "spawn_error") { + if ((processResult.error as NodeJS.ErrnoException | undefined)?.code === "ENOENT") { + throw new Error("[FFmpeg] ffmpeg not found"); } + throw processResult.error ?? new Error(processResult.stderr); + } + if (!processResult.success) { + // With the SDR-to-HDR remap folded into this pass, a filter failure + // (e.g. an ffmpeg built without the colorspace filter) would otherwise + // surface as a generic extract error and the operator has to grep the + // filter chain to learn it was the HDR conversion. Attribute it. + const hdrPrefix = options.sdrToHdrTransfer + ? `SDR→HDR conversion failed (colorspace filter in extract pass, target ${options.sdrToHdrTransfer}): ` + : ""; + const timeoutSuffix = + processResult.terminationReason === "deadline" + ? ` (timed out after ${ffmpegProcessTimeout} ms)` + : ""; + throw new Error( + `${hdrPrefix}FFmpeg exited with code ${processResult.exitCode}${timeoutSuffix}: ${processResult.stderr.slice(-500)}`, + ); + } - const timer = setTimeout(() => { - ffmpeg.kill("SIGTERM"); - }, ffmpegProcessTimeout); - - ffmpeg.stderr.on("data", (data) => { - stderr += data.toString(); - }); - - ffmpeg.on("close", (code) => { - clearTimeout(timer); - if (signal) signal.removeEventListener("abort", onAbort); - if (signal?.aborted) { - reject(new Error("Video frame extraction cancelled")); - return; - } - if (code !== 0) { - // With the SDR-to-HDR remap folded into this pass, a filter failure - // (e.g. an ffmpeg built without the colorspace filter) would otherwise - // surface as a generic extract error and the operator has to grep the - // filter chain to learn it was the HDR conversion. Attribute it. - const hdrPrefix = options.sdrToHdrTransfer - ? `SDR→HDR conversion failed (colorspace filter in extract pass, target ${options.sdrToHdrTransfer}): ` - : ""; - reject(new Error(`${hdrPrefix}FFmpeg exited with code ${code}: ${stderr.slice(-500)}`)); - return; - } - - const framePaths = new Map(); - const files = readdirSync(videoOutputDir) - .filter((f) => f.startsWith(FRAME_FILENAME_PREFIX) && f.endsWith(`.${format}`)) - .sort(); - files.forEach((file, index) => { - framePaths.set(index, join(videoOutputDir, file)); - }); - - resolve({ - videoId, - srcPath: videoPath, - outputDir: videoOutputDir, - framePattern, - fps, - totalFrames: framePaths.size, - metadata, - framePaths, - }); - }); - - ffmpeg.on("error", (err) => { - clearTimeout(timer); - if (signal) signal.removeEventListener("abort", onAbort); - if ((err as NodeJS.ErrnoException).code === "ENOENT") { - reject(new Error("[FFmpeg] ffmpeg not found")); - } else { - reject(err); - } - }); + const framePaths = new Map(); + const files = readdirSync(videoOutputDir) + .filter((f) => f.startsWith(FRAME_FILENAME_PREFIX) && f.endsWith(`.${format}`)) + .sort(); + files.forEach((file, index) => { + framePaths.set(index, join(videoOutputDir, file)); }); + + return { + videoId, + srcPath: videoPath, + outputDir: videoOutputDir, + framePattern, + fps, + totalFrames: framePaths.size, + metadata, + framePaths, + }; } /** diff --git a/packages/engine/src/utils/ffprobe.ts b/packages/engine/src/utils/ffprobe.ts index 2fbaf42259..503e4e3b06 100644 --- a/packages/engine/src/utils/ffprobe.ts +++ b/packages/engine/src/utils/ffprobe.ts @@ -3,42 +3,40 @@ import { spawn } from "child_process"; import { readFileSync } from "fs"; import { extname } from "path"; import { FFPROBE_PATH_ENV, getFfprobeBinary } from "./ffmpegBinaries.js"; +import { ManagedChildProcess } from "./managedChildProcess.js"; +import { trackChildProcess } from "./processTracker.js"; /** Spawn ffprobe with given args, return stdout. Throws on non-zero exit or missing binary. */ -function runFfprobe(args: string[]): Promise { - return new Promise((resolve, reject) => { - const command = getFfprobeBinary(); - const proc = spawn(command, args); - let stdout = ""; - let stderr = ""; - proc.stdout.on("data", (data) => { - stdout += data.toString(); - }); - proc.stderr.on("data", (data) => { - stderr += data.toString(); - }); - proc.on("close", (code) => { - if (code !== 0) { - reject(new Error(`[FFmpeg] ffprobe exited with code ${code}: ${stderr}`)); - } else { - resolve(stdout); - } - }); - proc.on("error", (err) => { - if ((err as NodeJS.ErrnoException).code === "ENOENT") { - const configured = process.env[FFPROBE_PATH_ENV]?.trim(); - reject( - new Error( - configured - ? `[FFmpeg] ffprobe not found at ${FFPROBE_PATH_ENV}="${configured}". Please install FFmpeg.` - : "[FFmpeg] ffprobe not found. Please install FFmpeg.", - ), - ); - } else { - reject(err); - } - }); +async function runFfprobe(args: string[], signal?: AbortSignal): Promise { + const command = getFfprobeBinary(); + const proc = spawn(command, args); + trackChildProcess(proc); + let stdout = ""; + proc.stdout.on("data", (data) => { + 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") { + const configured = process.env[FFPROBE_PATH_ENV]?.trim(); + throw new Error( + configured + ? `[FFmpeg] ffprobe not found at ${FFPROBE_PATH_ENV}="${configured}". Please install FFmpeg.` + : "[FFmpeg] ffprobe not found. Please install FFmpeg.", + ); + } + throw outcome.error ?? new Error(outcome.stderr); + } + if (outcome.reason !== "exit" || outcome.exitCode !== 0) { + throw new Error( + `[FFmpeg] ffprobe ${outcome.reason} with code ${outcome.exitCode}: ${outcome.stderr}`, + ); + } + return stdout; } function parseProbeJson(stdout: string): FFProbeOutput { @@ -348,20 +346,21 @@ export async function extractMediaMetadata(filePath: string): Promise { - const cached = audioMetadataCache.get(filePath); +export async function extractAudioMetadata( + filePath: string, + options?: { signal?: AbortSignal }, +): Promise { + // A caller-owned abort signal cannot safely share a cached in-flight probe: + // cancelling one consumer would also cancel unrelated consumers. Signal-bound + // probes therefore bypass the process-promise cache. + const cached = options?.signal ? undefined : audioMetadataCache.get(filePath); if (cached) return cached; const probePromise = (async (): Promise => { - const stdout = await runFfprobe([ - "-v", - "quiet", - "-print_format", - "json", - "-show_format", - "-show_streams", - filePath, - ]); + const stdout = await runFfprobe( + ["-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", filePath], + options?.signal, + ); const output = parseProbeJson(stdout); const audioStream = output.streams.find((s) => s.codec_type === "audio"); if (!audioStream) throw new Error("[FFmpeg] No audio stream found"); @@ -379,6 +378,7 @@ export async function extractAudioMetadata(filePath: string): Promise { if (audioMetadataCache.get(filePath) === probePromise) { diff --git a/packages/engine/src/utils/gpuEncoder.ts b/packages/engine/src/utils/gpuEncoder.ts index fcdb7fa570..bf5c3a5d30 100644 --- a/packages/engine/src/utils/gpuEncoder.ts +++ b/packages/engine/src/utils/gpuEncoder.ts @@ -8,6 +8,8 @@ import { spawn } from "child_process"; import { getFfmpegBinary } from "./ffmpegBinaries.js"; +import { ManagedChildProcess } from "./managedChildProcess.js"; +import { trackChildProcess } from "./processTracker.js"; export type ConcreteGpuEncoder = "nvenc" | "videotoolbox" | "vaapi" | "qsv" | "amf"; export type GpuEncoder = ConcreteGpuEncoder | null; @@ -60,25 +62,20 @@ export async function selectUsableGpuEncoder( } export async function detectGpuEncoder(): Promise { - return new Promise((resolve) => { - const ffmpeg = spawn(getFfmpegBinary(), ["-encoders"], { - stdio: ["pipe", "pipe", "pipe"], - }); - let stdout = ""; - - ffmpeg.stdout.on("data", (data) => { - stdout += data.toString(); - }); - - ffmpeg.on("close", () => { - const candidates = getCompiledGpuEncoders(stdout); - void selectUsableGpuEncoder(candidates, canUseGpuEncoder) - .then(resolve) - .catch(() => resolve(null)); - }); - - ffmpeg.on("error", () => resolve(null)); + const ffmpeg = spawn(getFfmpegBinary(), ["-encoders"], { + stdio: ["pipe", "pipe", "pipe"], }); + trackChildProcess(ffmpeg); + let stdout = ""; + ffmpeg.stdout.on("data", (data) => { + stdout += data.toString(); + }); + const outcome = await new ManagedChildProcess(ffmpeg, { + deadlineAtMs: Date.now() + 30_000, + }).wait(); + if (outcome.reason !== "exit" || outcome.exitCode !== 0) return null; + const candidates = getCompiledGpuEncoders(stdout); + return selectUsableGpuEncoder(candidates, canUseGpuEncoder).catch(() => null); } let cachedGpuEncoder: GpuEncoder | undefined = undefined; @@ -147,46 +144,23 @@ export function getProbeArgs(encoder: ConcreteGpuEncoder): string[] { } async function canUseGpuEncoder(encoder: ConcreteGpuEncoder): Promise { - return new Promise((resolve) => { - let settled = false; - let timedOut = false; - let killTimer: ReturnType | undefined; - let stderr = ""; - const finish = (usable: boolean) => { - if (settled) return; - settled = true; - clearTimeout(timer); - if (killTimer) clearTimeout(killTimer); - resolve(usable); - }; - const ffmpeg = spawn(getFfmpegBinary(), getProbeArgs(encoder), { - stdio: ["ignore", "ignore", "pipe"], - }); - - ffmpeg.stderr?.on("data", (data) => { - stderr += data.toString(); - }); - - const timer = setTimeout(() => { - timedOut = true; - ffmpeg.kill("SIGTERM"); - killTimer = setTimeout(() => { - ffmpeg.kill("SIGKILL"); - finish(false); - }, GPU_PROBE_KILL_GRACE_MS); - }, GPU_PROBE_TIMEOUT_MS); - - ffmpeg.on("close", (code, signal) => { - const usable = code === 0; - logGpuProbeFailure(encoder, { code, signal, stderr, timedOut }); - finish(usable); - }); - - ffmpeg.on("error", (error) => { - logGpuProbeFailure(encoder, { error, timedOut }); - finish(false); - }); + const ffmpeg = spawn(getFfmpegBinary(), getProbeArgs(encoder), { + stdio: ["ignore", "ignore", "pipe"], + }); + trackChildProcess(ffmpeg); + const outcome = await new ManagedChildProcess(ffmpeg, { + deadlineAtMs: Date.now() + GPU_PROBE_TIMEOUT_MS, + terminationGraceMs: GPU_PROBE_KILL_GRACE_MS, + }).wait(); + const usable = outcome.reason === "exit" && outcome.exitCode === 0; + logGpuProbeFailure(encoder, { + code: outcome.exitCode, + signal: outcome.signal, + stderr: outcome.stderr, + error: outcome.error, + timedOut: outcome.reason === "deadline", }); + return usable; } function logGpuProbeFailure( diff --git a/packages/engine/src/utils/managedChildProcess.test.ts b/packages/engine/src/utils/managedChildProcess.test.ts new file mode 100644 index 0000000000..cbd8536fb2 --- /dev/null +++ b/packages/engine/src/utils/managedChildProcess.test.ts @@ -0,0 +1,102 @@ +import { EventEmitter } from "node:events"; +import type { ChildProcess } from "node:child_process"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { ManagedChildProcess } from "./managedChildProcess.js"; + +function childProcess() { + const child = new EventEmitter() as EventEmitter & { + stderr: EventEmitter; + kill: ReturnType; + }; + child.stderr = new EventEmitter(); + child.kill = vi.fn().mockReturnValue(true); + return child as unknown as ChildProcess & { + stderr: EventEmitter; + kill: ReturnType; + }; +} + +describe("ManagedChildProcess", () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it("returns a typed natural exit and bounded stderr tail", async () => { + const child = childProcess(); + const managed = new ManagedChildProcess(child, { stderrMaxBytes: 5 }); + child.stderr.emit("data", Buffer.from("123456789")); + child.emit("close", 0, null); + + await expect(managed.wait()).resolves.toMatchObject({ + reason: "exit", + exitCode: 0, + stderr: "56789", + }); + }); + + it("escalates abort from SIGTERM to SIGKILL and resolves only after close", async () => { + vi.useFakeTimers(); + const child = childProcess(); + const controller = new AbortController(); + const managed = new ManagedChildProcess(child, { + signal: controller.signal, + terminationGraceMs: 50, + }); + + controller.abort(); + expect(child.kill).toHaveBeenCalledWith("SIGTERM"); + await vi.advanceTimersByTimeAsync(50); + expect(child.kill).toHaveBeenCalledWith("SIGKILL"); + + let reaped = false; + void managed.wait().then(() => { + reaped = true; + }); + await Promise.resolve(); + expect(reaped).toBe(false); + child.emit("close", null, "SIGKILL"); + await expect(managed.wait()).resolves.toMatchObject({ reason: "abort", signal: "SIGKILL" }); + }); + + it("distinguishes deadline from inactivity and refreshes activity", async () => { + vi.useFakeTimers(); + const deadlineChild = childProcess(); + const deadline = new ManagedChildProcess(deadlineChild, { + deadlineAtMs: Date.now() + 100, + terminationGraceMs: 1_000, + }); + await vi.advanceTimersByTimeAsync(100); + expect(deadlineChild.kill).toHaveBeenCalledWith("SIGTERM"); + deadlineChild.emit("close", null, "SIGTERM"); + await expect(deadline.wait()).resolves.toMatchObject({ reason: "deadline" }); + + const inactiveChild = childProcess(); + const inactive = new ManagedChildProcess(inactiveChild, { + inactivityTimeoutMs: 100, + terminationGraceMs: 1_000, + }); + await vi.advanceTimersByTimeAsync(75); + inactive.markActivity(); + await vi.advanceTimersByTimeAsync(75); + expect(inactiveChild.kill).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(25); + expect(inactiveChild.kill).toHaveBeenCalledWith("SIGTERM"); + inactiveChild.emit("close", null, "SIGTERM"); + await expect(inactive.wait()).resolves.toMatchObject({ reason: "inactivity" }); + }); + + it("settles a spawn failure and removes cancellation listeners", async () => { + const child = childProcess(); + const controller = new AbortController(); + const managed = new ManagedChildProcess(child, { signal: controller.signal }); + child.emit("error", new Error("spawn ENOENT")); + controller.abort(); + + await expect(managed.wait()).resolves.toMatchObject({ + reason: "spawn_error", + exitCode: null, + stderr: "spawn ENOENT", + }); + expect(child.kill).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/engine/src/utils/managedChildProcess.ts b/packages/engine/src/utils/managedChildProcess.ts new file mode 100644 index 0000000000..9dfeb70a32 --- /dev/null +++ b/packages/engine/src/utils/managedChildProcess.ts @@ -0,0 +1,178 @@ +import type { ChildProcess } from "node:child_process"; + +export type ManagedProcessTerminationReason = + | "exit" + | "abort" + | "deadline" + | "inactivity" + | "spawn_error"; + +export interface ManagedChildProcessOutcome { + reason: ManagedProcessTerminationReason; + exitCode: number | null; + signal: NodeJS.Signals | null; + stderr: string; + durationMs: number; + error?: Error; +} + +export interface ManagedChildProcessOptions { + signal?: AbortSignal; + deadlineAtMs?: number; + inactivityTimeoutMs?: number; + terminationGraceMs?: number; + stderrMaxBytes?: number; + onStderr?: (chunk: string) => void; + now?: () => number; +} + +const DEFAULT_TERMINATION_GRACE_MS = 2_000; +const DEFAULT_STDERR_MAX_BYTES = 64 * 1024; + +/** Owns cancellation, escalation, stderr and reaping for one child process. */ +export class ManagedChildProcess { + private readonly startedAtMs: number; + private readonly now: () => number; + private readonly outcomePromise: Promise; + private resolveOutcome!: (outcome: ManagedChildProcessOutcome) => void; + private requestedReason: Exclude | null = + null; + private stderrTail = Buffer.alloc(0); + private settled = false; + private deadlineTimer: NodeJS.Timeout | null = null; + private inactivityTimer: NodeJS.Timeout | null = null; + private escalationTimer: NodeJS.Timeout | null = null; + + constructor( + readonly child: ChildProcess, + private readonly options: ManagedChildProcessOptions = {}, + ) { + this.now = options.now ?? Date.now; + this.startedAtMs = this.now(); + this.outcomePromise = new Promise((resolve) => { + this.resolveOutcome = resolve; + }); + + child.stderr?.on("data", this.onStderr); + child.once("close", this.onClose); + child.once("error", this.onError); + this.installAbort(); + this.installDeadline(); + this.markActivity(); + } + + wait(): Promise { + return this.outcomePromise; + } + + get isSettled(): boolean { + return this.settled; + } + + markActivity(): void { + if (this.settled || this.options.inactivityTimeoutMs === undefined) return; + if (this.inactivityTimer) clearTimeout(this.inactivityTimer); + this.inactivityTimer = setTimeout( + () => this.requestTermination("inactivity"), + Math.max(0, this.options.inactivityTimeoutMs), + ); + this.inactivityTimer.unref?.(); + } + + private readonly onStderr = (data: Buffer | string): void => { + const chunk = Buffer.isBuffer(data) ? data : Buffer.from(data); + const maxBytes = this.options.stderrMaxBytes ?? DEFAULT_STDERR_MAX_BYTES; + this.stderrTail = Buffer.concat([this.stderrTail, chunk]); + if (this.stderrTail.byteLength > maxBytes) { + this.stderrTail = this.stderrTail.subarray(this.stderrTail.byteLength - maxBytes); + } + this.options.onStderr?.(chunk.toString()); + }; + + private readonly onClose = (exitCode: number | null, signal: NodeJS.Signals | null): void => { + this.settle({ + reason: this.requestedReason ?? "exit", + exitCode, + signal, + stderr: this.stderrTail.toString(), + durationMs: this.now() - this.startedAtMs, + }); + }; + + private readonly onError = (error: Error): void => { + this.settle({ + reason: "spawn_error", + exitCode: null, + signal: null, + stderr: this.stderrTail.length > 0 ? this.stderrTail.toString() : error.message, + durationMs: this.now() - this.startedAtMs, + error, + }); + }; + + private installAbort(): void { + const signal = this.options.signal; + if (!signal) return; + if (signal.aborted) { + this.requestTermination("abort"); + return; + } + signal.addEventListener("abort", this.onAbort, { once: true }); + } + + private readonly onAbort = (): void => { + this.requestTermination("abort"); + }; + + private installDeadline(): void { + if (this.options.deadlineAtMs === undefined) return; + const remainingMs = Math.max(0, this.options.deadlineAtMs - this.now()); + this.deadlineTimer = setTimeout(() => this.requestTermination("deadline"), remainingMs); + this.deadlineTimer.unref?.(); + } + + private requestTermination( + reason: Exclude, + ): void { + if (this.settled || this.requestedReason) return; + this.requestedReason = reason; + try { + this.child.kill("SIGTERM"); + } catch { + // A close/error event owns settlement; escalation remains the backstop. + } + const graceMs = this.options.terminationGraceMs ?? DEFAULT_TERMINATION_GRACE_MS; + this.escalationTimer = setTimeout( + () => { + if (this.settled) return; + try { + this.child.kill("SIGKILL"); + } catch { + // The child may have exited between the settled check and kill. + } + }, + Math.max(0, graceMs), + ); + this.escalationTimer.unref?.(); + } + + private settle(outcome: ManagedChildProcessOutcome): void { + if (this.settled) return; + this.settled = true; + this.clearTimers(); + this.options.signal?.removeEventListener("abort", this.onAbort); + this.child.stderr?.off("data", this.onStderr); + this.child.off("close", this.onClose); + this.child.off("error", this.onError); + this.resolveOutcome(outcome); + } + + private clearTimers(): void { + if (this.deadlineTimer) clearTimeout(this.deadlineTimer); + if (this.inactivityTimer) clearTimeout(this.inactivityTimer); + if (this.escalationTimer) clearTimeout(this.escalationTimer); + this.deadlineTimer = null; + this.inactivityTimer = null; + this.escalationTimer = null; + } +} diff --git a/packages/engine/src/utils/runFfmpeg.ts b/packages/engine/src/utils/runFfmpeg.ts index 418422f999..5140b78900 100644 --- a/packages/engine/src/utils/runFfmpeg.ts +++ b/packages/engine/src/utils/runFfmpeg.ts @@ -9,6 +9,10 @@ import { spawn } from "child_process"; import { getFfmpegBinary } from "./ffmpegBinaries.js"; import { trackChildProcess } from "./processTracker.js"; +import { + ManagedChildProcess, + type ManagedProcessTerminationReason, +} from "./managedChildProcess.js"; export interface RunFfmpegOptions { signal?: AbortSignal; @@ -21,6 +25,8 @@ export interface RunFfmpegResult { exitCode: number | null; stderr: string; durationMs: number; + terminationReason: ManagedProcessTerminationReason; + error?: Error; } const DEFAULT_TIMEOUT = 300_000; @@ -77,60 +83,21 @@ export function formatFfmpegError( } export async function runFfmpeg(args: string[], opts?: RunFfmpegOptions): Promise { - const startMs = Date.now(); - const signal = opts?.signal; const timeout = opts?.timeout ?? DEFAULT_TIMEOUT; - const onStderr = opts?.onStderr; - - return new Promise((resolve) => { - const ffmpeg = spawn(getFfmpegBinary(), args); - trackChildProcess(ffmpeg); - let stderr = ""; - - const onAbort = () => { - ffmpeg.kill("SIGTERM"); - }; - - if (signal) { - if (signal.aborted) { - ffmpeg.kill("SIGTERM"); - } else { - signal.addEventListener("abort", onAbort, { once: true }); - } - } - - const timer = setTimeout(() => { - ffmpeg.kill("SIGTERM"); - }, timeout); - - ffmpeg.stderr.on("data", (data: Buffer) => { - const chunk = data.toString(); - stderr += chunk; - if (onStderr) { - onStderr(chunk); - } - }); - - ffmpeg.on("close", (code) => { - clearTimeout(timer); - if (signal) signal.removeEventListener("abort", onAbort); - resolve({ - success: !signal?.aborted && code === 0, - exitCode: code, - stderr, - durationMs: Date.now() - startMs, - }); - }); - - ffmpeg.on("error", (err) => { - clearTimeout(timer); - if (signal) signal.removeEventListener("abort", onAbort); - resolve({ - success: false, - exitCode: null, - stderr: err.message, - durationMs: Date.now() - startMs, - }); - }); + const ffmpeg = spawn(getFfmpegBinary(), args); + trackChildProcess(ffmpeg); + const managed = new ManagedChildProcess(ffmpeg, { + signal: opts?.signal, + deadlineAtMs: Date.now() + timeout, + onStderr: opts?.onStderr, }); + const outcome = await managed.wait(); + return { + success: outcome.reason === "exit" && outcome.exitCode === 0, + exitCode: outcome.exitCode, + stderr: outcome.stderr, + durationMs: outcome.durationMs, + terminationReason: outcome.reason, + error: outcome.error, + }; } diff --git a/packages/producer/src/services/animatedGifPrep.ts b/packages/producer/src/services/animatedGifPrep.ts index 9eb626ded0..be4a1f101a 100644 --- a/packages/producer/src/services/animatedGifPrep.ts +++ b/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 { - await new Promise((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: { diff --git a/packages/producer/src/services/distributed/assemble.ts b/packages/producer/src/services/distributed/assemble.ts index 747240f034..3b59b91b3e 100644 --- a/packages/producer/src/services/distributed/assemble.ts +++ b/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}`); diff --git a/packages/producer/src/services/render/audioPadTrim.test.ts b/packages/producer/src/services/render/audioPadTrim.test.ts index cbf62e9623..5226f38884 100644 --- a/packages/producer/src/services/render/audioPadTrim.test.ts +++ b/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, @@ -138,6 +138,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 }, diff --git a/packages/producer/src/services/render/audioPadTrim.ts b/packages/producer/src/services/render/audioPadTrim.ts index b089b09a81..659eb95caf 100644 --- a/packages/producer/src/services/render/audioPadTrim.ts +++ b/packages/producer/src/services/render/audioPadTrim.ts @@ -26,7 +26,9 @@ import { formatFfmpegError, getFfmpegBinary, getFfprobeBinary, + ManagedChildProcess, runFfmpeg, + trackChildProcess, type AudioMetadata, } from "@hyperframes/engine"; @@ -64,12 +66,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; - probeAudioInfo?: (audioPath: string) => Promise; + probeAudioInfo?: (audioPath: string, signal?: AbortSignal) => Promise; runFfmpeg?: ( args: string[], options?: { stdin?: string }, @@ -242,15 +245,20 @@ function concatFileLine(path: string): string { export async function padOrTrimAudioToVideoFrameCount( input: PadTrimAudioInput, ): Promise { - 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[], options?: { stdin?: string }) => + defaultRunFfmpeg(args, { ...options, signal: 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") { @@ -363,39 +371,48 @@ interface FfprobeOutput { streams?: FfprobeStreamInfo[]; } -async function defaultProbeVideoFrameInfo(videoPath: string): Promise { +async function defaultProbeVideoFrameInfo( + videoPath: string, + signal?: AbortSignal, +): Promise { // 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([ - "-v", - "error", - "-select_streams", - "v:0", - "-show_entries", - "stream=nb_frames,r_frame_rate", - "-of", - "json", - videoPath, - ]); + const fastInfo = await runFfprobeJson( + [ + "-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([ - "-v", - "error", - "-select_streams", - "v:0", - "-count_packets", - "-show_entries", - "stream=nb_read_packets,r_frame_rate", - "-of", - "json", - videoPath, - ]); + const slowInfo = await runFfprobeJson( + [ + "-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); @@ -415,9 +432,12 @@ function parseFrameRate(rate: string): { fpsNum: number; fpsDen: number } { return { fpsNum, fpsDen }; } -async function defaultProbeAudioInfo(audioPath: string): Promise { +async function defaultProbeAudioInfo( + audioPath: string, + signal?: AbortSignal, +): Promise { // extractAudioMetadata is the shared ffprobe wrapper (caches results). - const metadata: AudioMetadata = await extractAudioMetadata(audioPath); + const metadata: AudioMetadata = await extractAudioMetadata(audioPath, { signal }); return { durationSeconds: metadata.durationSeconds, sampleRate: metadata.sampleRate, @@ -428,11 +448,13 @@ async function defaultProbeAudioInfo(audioPath: string): Promise async function defaultRunFfmpeg( args: string[], - options?: { stdin?: string }, + options?: { stdin?: string; signal?: AbortSignal }, ): Promise<{ success: boolean; error?: string }> { - if (options?.stdin !== undefined) return runFfmpegWithStdin(args, options.stdin); + if (options?.stdin !== undefined) { + return runFfmpegWithStdin(args, options.stdin, options.signal); + } - const result = await runFfmpeg(args); + const result = await runFfmpeg(args, { signal: options?.signal }); if (result.success) return { success: true }; return { success: false, @@ -443,67 +465,49 @@ async function defaultRunFfmpeg( async function runFfmpegWithStdin( args: string[], stdin: string, + signal?: AbortSignal, ): Promise<{ success: boolean; error?: string }> { - return new Promise((resolve) => { - const proc = spawn(getFfmpegBinary(), args); - let stderr = ""; - - proc.stderr.on("data", (data: Buffer) => { - stderr += data.toString(); - }); - - proc.on("error", (err) => { - resolve({ - success: false, - error: `[audioPadTrim] ${err instanceof Error ? err.message : String(err)}`, - }); - }); - - proc.on("close", (code) => { - if (code === 0) { - resolve({ success: true }); - return; - } - resolve({ - success: false, - error: `[audioPadTrim] ${formatFfmpegError(code, stderr)}`, - }); - }); - - proc.stdin.end(stdin); + const proc = spawn(getFfmpegBinary(), args); + trackChildProcess(proc); + const managed = new ManagedChildProcess(proc, { + signal, + deadlineAtMs: Date.now() + 300_000, }); + proc.stdin.end(stdin); + const outcome = await managed.wait(); + if (outcome.reason === "exit" && outcome.exitCode === 0) return { success: true }; + return { + success: false, + error: `[audioPadTrim] ${formatFfmpegError(outcome.exitCode, outcome.stderr)}`, + }; } // ── ffprobe JSON runner (shared between fast/slow video probe paths) ───── -function runFfprobeJson(args: string[]): Promise { - 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(args: string[], signal?: AbortSignal): Promise { + 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}`); + } }