Skip to content

refactor(engine): manage child process lifecycles #2160

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jrusso1020 wants to merge 1 commit into
base: 07-10-refactor_engine_add_browser_leases
Choose a base branch
from
Open

refactor(engine): manage child process lifecycles #2160

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions packages/engine/src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
10 changes: 8 additions & 2 deletions packages/engine/src/services/chunkEncoder.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down
205 changes: 53 additions & 152 deletions packages/engine/src/services/chunkEncoder.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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";
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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 {
Expand Down
2 changes: 1 addition & 1 deletion packages/engine/src/services/streamingEncoder.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
Loading
Loading