fix(ffmpeg): make process shutdown race-safe

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<null, Readable, Readable>;
+    readonly process: ChildProcessByStdio<null, Readable, Readable>;
-    ready: Promise<void>;
+    readonly ready: Promise<void>;
     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<null, Readable, Readable>): Promise<void> {
+function waitForFfmpegOutput(
+    childProcess: ChildProcessByStdio<null, Readable, Readable>,
+    signal: AbortSignal,
+): Promise<void> {
+    const { promise, resolve, reject } = Promise.withResolvers<void>();
     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 settle = (complete: () => void): void => {
-        const cleanup = (): void => {
+        if (settled) {
-            childProcess.off('error', onError);
+            return;
-            childProcess.off('exit', onExit);
+        }
-            childProcess.stdout.off('readable', onReadable);
-        };
 
-        const fail = (message: string, cause?: unknown): void => {
+        settled = true;
-            cleanup();
+        cleanup();
-            reject(new FfmpegProcessError(addStderrTail(message, stderrTail), cause));
+        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 => {
+    const onExit = (code: number | null, exitSignal: NodeJS.Signals | null): void => {
-            fail(`ffmpeg exited before producing audio. Exit code: ${code ?? 'none'}, signal: ${signal ?? 'none'}.`);
+        fail(`ffmpeg exited before producing audio. Exit code: ${code ?? 'none'}, signal: ${exitSignal ?? 'none'}.`);
-        };
+    };
 
-        const onReadable = (): void => {
+    const onReadable = (): void => {
-            cleanup();
+        settle(() => resolve());
-            resolve();
+    };
-        };
 
-        childProcess.once('error', onError);
+    const onAbort = (): void => {
-        childProcess.once('exit', onExit);
+        const reason: unknown = signal.reason;
-        childProcess.stdout.once('readable', onReadable);
+        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<null, Readable, Readable>): void {
+function stopProcess(
+    childProcess: ChildProcessByStdio<null, Readable, Readable>,
+    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();
 }

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.toMatchObject({
-        await expect(handle.ready).rejects.toThrow('spawn ENOENT');
+            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();
     });
 });