From f48424e198d11050e5db2f93c08c254e19fb3d33 Mon Sep 17 00:00:00 2001 From: Julian Lechner Date: Mon, 24 Aug 2026 14:12:40 +0200 Subject: [PATCH] fix(ffmpeg): make process shutdown race-safe --- src/ffmpeg.ts | 105 +++++++++++++++++++++++++++++-------------- tests/ffmpeg.test.ts | 39 +++++++++++----- 2 files changed, 100 insertions(+), 44 deletions(-) diff --git a/src/ffmpeg.ts b/src/ffmpeg.ts index e207920..5eb7b37 100644 --- a/src/ffmpeg.ts +++ b/src/ffmpeg.ts @@ -14,8 +14,8 @@ const FORCE_KILL_TIMEOUT_MS = 2_000; const STDERR_TAIL_BYTES = 4_096; export type FfmpegProcessHandle = { - process: ChildProcessByStdio; - ready: Promise; + readonly process: ChildProcessByStdio; + readonly ready: Promise; stop(): void; }; @@ -36,7 +36,8 @@ export function resolveFfmpegExecutable(options: FfmpegOptions = {}): string { } } catch (error) { throw new AudioManagerConfigError( - `Unable to resolve ffmpeg-static. Install it or pass ffmpeg.executablePath. Cause: ${String(error)}`, + 'Unable to resolve ffmpeg-static. Install it or pass ffmpeg.executablePath.', + { cause: error }, ); } @@ -52,7 +53,8 @@ export function startFfmpeg(input: string, options: FfmpegOptions = {}): FfmpegP ...(options.outputArgs ?? DEFAULT_OUTPUT_ARGS), ]; const childProcess = spawn(executable, args, { stdio: ['ignore', 'pipe', 'pipe'] }); - const ready = waitForFfmpegOutput(childProcess); + const abortController = new AbortController(); + const ready = waitForFfmpegOutput(childProcess, abortController.signal); childProcess.stderr.resume(); @@ -60,49 +62,76 @@ export function startFfmpeg(input: string, options: FfmpegOptions = {}): FfmpegP process: childProcess, ready, stop: (): void => { - stopProcess(childProcess); + stopProcess(childProcess, abortController); }, }; } -function waitForFfmpegOutput(childProcess: ChildProcessByStdio): Promise { +function waitForFfmpegOutput( + childProcess: ChildProcessByStdio, + signal: AbortSignal, +): Promise { + const { promise, resolve, reject } = Promise.withResolvers(); let stderrTail = ''; + let settled = false; const appendStderr = (chunk: Buffer | string): void => { stderrTail = (stderrTail + String(chunk)).slice(-STDERR_TAIL_BYTES); }; - childProcess.stderr.on('data', appendStderr); + const cleanup = (): void => { + childProcess.off('exit', onExit); + childProcess.stdout.off('readable', onReadable); + childProcess.stderr.off('data', appendStderr); + signal.removeEventListener('abort', onAbort); + }; - return new Promise((resolve, reject) => { - const cleanup = (): void => { - childProcess.off('error', onError); - childProcess.off('exit', onExit); - childProcess.stdout.off('readable', onReadable); - }; + const settle = (complete: () => void): void => { + if (settled) { + return; + } - const fail = (message: string, cause?: unknown): void => { - cleanup(); - reject(new FfmpegProcessError(addStderrTail(message, stderrTail), cause)); - }; + settled = true; + cleanup(); + complete(); + }; - const onError = (error: Error): void => { + const fail = (message: string, cause?: unknown): void => { + settle(() => reject(new FfmpegProcessError(addStderrTail(message, stderrTail), cause))); + }; + + const onError = (error: Error): void => { + if (!settled) { fail(`Unable to start ffmpeg. Cause: ${error.message}`, error); - }; + } + }; - const onExit = (code: number | null, signal: NodeJS.Signals | null): void => { - fail(`ffmpeg exited before producing audio. Exit code: ${code ?? 'none'}, signal: ${signal ?? 'none'}.`); - }; + const onExit = (code: number | null, exitSignal: NodeJS.Signals | null): void => { + fail(`ffmpeg exited before producing audio. Exit code: ${code ?? 'none'}, signal: ${exitSignal ?? 'none'}.`); + }; - const onReadable = (): void => { - cleanup(); - resolve(); - }; + const onReadable = (): void => { + settle(() => resolve()); + }; - childProcess.once('error', onError); - childProcess.once('exit', onExit); - childProcess.stdout.once('readable', onReadable); - }); + const onAbort = (): void => { + const reason: unknown = signal.reason; + settle(() => + reject( + reason instanceof Error + ? reason + : new FfmpegProcessError('ffmpeg was stopped before producing audio.', reason), + ), + ); + }; + + childProcess.stderr.on('data', appendStderr); + childProcess.on('error', onError); + childProcess.once('exit', onExit); + childProcess.stdout.once('readable', onReadable); + signal.addEventListener('abort', onAbort, { once: true }); + + return promise; } function addStderrTail(message: string, stderrTail: string): string { @@ -111,22 +140,30 @@ function addStderrTail(message: string, stderrTail: string): string { return trimmedTail ? `${message} stderr: ${trimmedTail}` : message; } -function stopProcess(childProcess: ChildProcessByStdio): void { +function stopProcess( + childProcess: ChildProcessByStdio, + abortController: AbortController, +): void { + if (abortController.signal.aborted) { + return; + } + + abortController.abort(new FfmpegProcessError('ffmpeg was stopped before producing audio.')); childProcess.stdout.destroy(); childProcess.stderr.destroy(); - childProcess.removeAllListeners(); - if (childProcess.killed || childProcess.exitCode !== null || childProcess.signalCode !== null) { + if (childProcess.exitCode !== null || childProcess.signalCode !== null) { return; } childProcess.kill('SIGTERM'); const forceKillTimeout = setTimeout(() => { - if (!childProcess.killed && childProcess.exitCode === null && childProcess.signalCode === null) { + if (childProcess.exitCode === null && childProcess.signalCode === null) { childProcess.kill('SIGKILL'); } }, FORCE_KILL_TIMEOUT_MS); + childProcess.once('close', () => clearTimeout(forceKillTimeout)); forceKillTimeout.unref(); } diff --git a/tests/ffmpeg.test.ts b/tests/ffmpeg.test.ts index 9db9d11..8db54c0 100644 --- a/tests/ffmpeg.test.ts +++ b/tests/ffmpeg.test.ts @@ -20,11 +20,12 @@ type MockChildProcess = { stderr: { destroy: jest.Mock; on: jest.Mock; + off: jest.Mock; resume: jest.Mock; }; + on: jest.Mock; once: jest.Mock; off: jest.Mock; - removeAllListeners: jest.Mock; kill: jest.Mock; killed: boolean; exitCode: number | null; @@ -41,11 +42,12 @@ function createMockChildProcess(): MockChildProcess { stderr: { destroy: jest.fn(), on: jest.fn(), + off: jest.fn(), resume: jest.fn(), }, + on: jest.fn(), once: jest.fn(), off: jest.fn(), - removeAllListeners: jest.fn(), kill: jest.fn(), killed: false, exitCode: null, @@ -61,6 +63,10 @@ function getProcessHandler(childProcess: MockChildProcess, eventName: string): ( return childProcess.once.mock.calls.find(([event]) => event === eventName)?.[1] as (...args: unknown[]) => void; } +function getPersistentProcessHandler(childProcess: MockChildProcess, eventName: string): (...args: unknown[]) => void { + return childProcess.on.mock.calls.find(([event]) => event === eventName)?.[1] as (...args: unknown[]) => void; +} + function getStdoutHandler(childProcess: MockChildProcess, eventName: string): (...args: unknown[]) => void { return childProcess.stdout.once.mock.calls.find(([event]) => event === eventName)?.[1] as ( ...args: unknown[] @@ -117,7 +123,7 @@ describe('ffmpeg helpers', () => { ], { stdio: ['ignore', 'pipe', 'pipe'] }, ); - expect(childProcess.once).toHaveBeenCalledWith('error', expect.any(Function)); + expect(childProcess.on).toHaveBeenCalledWith('error', expect.any(Function)); expect(childProcess.once).toHaveBeenCalledWith('exit', expect.any(Function)); expect(childProcess.stdout.once).toHaveBeenCalledWith('readable', expect.any(Function)); expect(childProcess.stderr.on).toHaveBeenCalledWith('data', expect.any(Function)); @@ -129,12 +135,16 @@ describe('ffmpeg helpers', () => { mockSpawnReturn(childProcess); const handle = startFfmpeg('tests/audio.mp3'); - const errorHandler = getProcessHandler(childProcess, 'error'); + const errorHandler = getPersistentProcessHandler(childProcess, 'error'); + const spawnError = new Error('spawn ENOENT'); - errorHandler(new Error('spawn ENOENT')); + errorHandler(spawnError); - await expect(handle.ready).rejects.toThrow(FfmpegProcessError); - await expect(handle.ready).rejects.toThrow('spawn ENOENT'); + await expect(handle.ready).rejects.toMatchObject({ + cause: spawnError, + message: expect.stringContaining('spawn ENOENT'), + name: FfmpegProcessError.name, + }); }); it('includes stderr when ffmpeg exits before producing audio', async () => { @@ -172,8 +182,10 @@ describe('ffmpeg helpers', () => { const handle = startFfmpeg('tests/audio.mp3'); const readableHandler = getStdoutHandler(childProcess, 'readable'); const exitHandler = getProcessHandler(childProcess, 'exit'); + const errorHandler = getPersistentProcessHandler(childProcess, 'error'); readableHandler(); + errorHandler(new Error('late process error')); exitHandler(1, null); await expect(handle.ready).resolves.toBeUndefined(); @@ -196,17 +208,22 @@ describe('ffmpeg helpers', () => { ); }); - it('stops a running child process and schedules a force kill fallback', () => { + it('stops a running child process and schedules a force kill fallback', async () => { jest.useFakeTimers(); const childProcess = createMockChildProcess(); + childProcess.kill.mockImplementation(() => { + childProcess.killed = true; + return true; + }); mockSpawnReturn(childProcess); const handle = startFfmpeg('tests/audio.mp3'); + const readyExpectation = expect(handle.ready).rejects.toThrow('stopped before producing audio'); handle.stop(); + await readyExpectation; expect(childProcess.stdout.destroy).toHaveBeenCalledTimes(1); expect(childProcess.stderr.destroy).toHaveBeenCalledTimes(1); - expect(childProcess.removeAllListeners).toHaveBeenCalledTimes(1); expect(childProcess.kill).toHaveBeenCalledWith('SIGTERM'); jest.advanceTimersByTime(2_000); @@ -214,14 +231,16 @@ describe('ffmpeg helpers', () => { expect(childProcess.kill).toHaveBeenNthCalledWith(2, 'SIGKILL'); }); - it('does not signal a process that already exited', () => { + it('does not signal a process that already exited', async () => { const childProcess = createMockChildProcess(); childProcess.exitCode = 0; mockSpawnReturn(childProcess); const handle = startFfmpeg('tests/audio.mp3'); + const readyExpectation = expect(handle.ready).rejects.toThrow('stopped before producing audio'); handle.stop(); + await readyExpectation; expect(childProcess.kill).not.toHaveBeenCalled(); }); });