fix: terminate stalled audio timing inspection at its deadline

Author: richiemcilroy

apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts

@@ -492,6 +492,55 @@ afterAll(async () => {
 });
 
 describe("encoded recording preservation", () => {
+	test.skipIf(process.platform === "win32")(
+		"kills and joins a stalled audio inspector on cancellation and timeout",
+		async () => {
+			for (const mode of ["cancel", "timeout"]) {
+				const path = join(directory, `stalled-audio-${mode}`);
+				await run(["mkfifo", path]);
+				const controller = new AbortController();
+				const pending = readRecordingAudioTail(path, controller.signal);
+				pending.catch(() => {});
+				let pids: number[] = [];
+				let timer: ReturnType<typeof setTimeout> | undefined;
+				try {
+					for (let attempt = 0; attempt < 100 && !pids.length; attempt++) {
+						pids = await decoderPids(path);
+						if (!pids.length) await Bun.sleep(10);
+					}
+					expect(pids).toHaveLength(1);
+					await Bun.sleep(100);
+					const started = performance.now();
+					if (mode === "timeout")
+						timer = setTimeout(
+							() =>
+								controller.abort(new DOMException("Timed out", "TimeoutError")),
+							50,
+						);
+					else controller.abort();
+					await expect(
+						Promise.race([
+							pending,
+							Bun.sleep(2000).then(() => {
+								throw new Error("Audio inspection ignored cancellation");
+							}),
+						]),
+					).rejects.toMatchObject({ retryable: true });
+					expect(performance.now() - started).toBeLessThan(2000);
+					expect(await decoderPids(path)).toEqual([]);
+				} finally {
+					if (timer) clearTimeout(timer);
+					controller.abort();
+					for (const pid of pids) {
+						try {
+							process.kill(pid, "SIGKILL");
+						} catch {}
+					}
+					await pending.catch(() => {});
+				}
+			}
+		},
+	);
 	test("retains the processing deadline reason when muxing is already cancelled", async () => {
 		const reason = new Error("Recording processing timed out");
 		await expect(
@@ -1859,7 +1908,10 @@ async function decoderPids(input: string): Promise<number[]> {
 			processes.map(async (pid) => {
 				try {
 					const command = await readFile(`/proc/${pid}/cmdline`, "utf8");
-					return command.includes("ffmpeg") && command.includes(input)
+					return (command.includes("ffmpeg") ||
+						command.includes("ffprobe") ||
+						command.includes("recording-audio-timing.ts")) &&
+						command.includes(input)
 						? Number(pid)
 						: null;
 				} catch (error) {
@@ -1878,7 +1930,13 @@ async function decoderPids(input: string): Promise<number[]> {
 	const output = await run(["ps", "-axo", "pid=,command="]);
 	return output
 		.split("\n")
-		.filter((line) => line.includes("ffmpeg") && line.includes(input))
+		.filter(
+			(line) =>
+				(line.includes("ffmpeg") ||
+					line.includes("ffprobe") ||
+					line.includes("recording-audio-timing.ts")) &&
+				line.includes(input),
+		)
 		.map((line) => Number.parseInt(line.trim(), 10));
 }
 

apps/media-server/src/lib/recording-audio-timing.ts

@@ -0,0 +1,106 @@
+import { createHash } from "node:crypto";
+import { EncodedPacketSink, FilePathSource, Input, MP4 } from "mediabunny";
+import { RecordingTimingError } from "./recording-timing";
+
+async function inspectAudioTail(path: string, countPackets = false) {
+	const source = new FilePathSource(path).ref();
+	const input = new Input({ formats: [MP4], source });
+	let formatReady = false;
+	try {
+		await input.getFormat();
+		formatReady = true;
+		const tracks = await input.getAudioTracks();
+		if (tracks.length !== 1)
+			throw new Error("Recording audio tracks are ambiguous");
+		const sink = new EncodedPacketSink(tracks[0]);
+		const packet = await sink.getPacket(Number.POSITIVE_INFINITY, {
+			skipLiveWait: true,
+		});
+		if (
+			!packet ||
+			(await sink.getNextPacket(packet, {
+				metadataOnly: true,
+				skipLiveWait: true,
+			}))
+		)
+			throw new Error("Recording audio tail is ambiguous");
+		const scale = await tracks[0].getTimeResolution();
+		const scaled = packet.duration * scale;
+		const ticks = Math.round(scaled);
+		const tolerance = Math.max(
+			0.0000001,
+			Math.abs(scaled) * Number.EPSILON * 4,
+		);
+		if (
+			!Number.isSafeInteger(scale) ||
+			scale <= 0 ||
+			!Number.isSafeInteger(ticks) ||
+			ticks <= 0 ||
+			tolerance >= 0.25 ||
+			Math.abs(scaled - ticks) > tolerance ||
+			!Number.isSafeInteger(packet.sequenceNumber) ||
+			packet.sequenceNumber < 0
+		)
+			throw new Error("Recording audio duration is not exact");
+		let packetCount: number | undefined;
+		if (countPackets) {
+			packetCount = 0;
+			for await (const _packet of sink.packets(undefined, undefined, {
+				metadataOnly: true,
+				skipLiveWait: true,
+			})) {
+				packetCount++;
+			}
+		}
+		if (
+			packetCount !== undefined &&
+			(!Number.isSafeInteger(packetCount) || packetCount <= 0)
+		)
+			throw new Error("Recording audio packet count is invalid");
+		return {
+			packetCount,
+			durationTicks: ticks,
+			timeScale: scale,
+			size: packet.byteLength,
+			hash: `SHA256:${createHash("sha256").update(packet.data).digest("hex")}`,
+		};
+	} catch (error) {
+		const retryable =
+			typeof error === "object" &&
+			error !== null &&
+			"code" in error &&
+			typeof error.code === "string";
+		const failure = new RecordingTimingError(
+			retryable
+				? "Recording audio timing inspection was interrupted"
+				: "Recording audio timing is invalid",
+			retryable,
+		);
+		failure.cause = error;
+		throw failure;
+	} finally {
+		// Mediabunny 1.45 disposal leaks rejections after failed format detection and can strand pending reads.
+		if (formatReady) input.dispose();
+		else source.free();
+	}
+}
+
+if (import.meta.main) {
+	try {
+		const path = process.argv[2];
+		if (!path) throw new Error("Missing audio timing path");
+		console.log(
+			JSON.stringify(await inspectAudioTail(path, process.argv[3] === "count")),
+		);
+	} catch (error) {
+		console.log(
+			JSON.stringify({
+				error:
+					error instanceof RecordingTimingError && !error.retryable
+						? "invalid"
+						: "unavailable",
+			}),
+		);
+		process.exitCode = 1;
+	}
+}