fix: preserve audio timing failure and cancellation semantics

Author: richiemcilroy

apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts

@@ -49,7 +49,12 @@ const sourceReads: {
 const uploadConditions: (string | null)[] = [];
 const multipartCallbacks: { action: string; payload: unknown }[] = [];
 let rejectMultipartSigning = false;
-let sourceFault: "changed" | "missing" | "corrupt" | undefined;
+let sourceFault:
+	| "changed"
+	| "missing"
+	| "corrupt"
+	| "corrupt-audio"
+	| undefined;
 let corruptRecordingReadback = false;
 let transientFixtureFailures = 0;
 let permanentFixtureFailures = 0;
@@ -295,7 +300,11 @@ beforeAll(async () => {
 					ifMatch: request.headers.get("if-match"),
 					verification: request.headers.get("x-cap-recording-verification"),
 				});
-				const affected = url.pathname.endsWith("video-segment.m4s");
+				const affected = url.pathname.endsWith(
+					sourceFault === "corrupt-audio"
+						? "audio-segment.m4s"
+						: "video-segment.m4s",
+				);
 				if (affected && sourceFault === "missing")
 					return new Response(null, { status: 404 });
 				if (
@@ -306,7 +315,8 @@ beforeAll(async () => {
 					return new Response(null, { status: 412 });
 				return new Response(
 					Uint8Array.from(
-						affected && sourceFault === "corrupt"
+						affected &&
+							(sourceFault === "corrupt" || sourceFault === "corrupt-audio")
 							? new Uint8Array(source.byteLength)
 							: source,
 					).buffer,
@@ -707,7 +717,7 @@ describe("media routes real-world integration tests", () => {
 		30_000,
 	);
 
-	test.each(["changed", "missing", "corrupt"] as const)(
+	test.each(["changed", "missing", "corrupt", "corrupt-audio"] as const)(
 		"withholds upload and proof after a pinned source is %s",
 		async (fault) => {
 			sourceFault = fault;
@@ -721,7 +731,7 @@ describe("media routes real-world integration tests", () => {
 				const job = await waitForTerminalJob(jobId);
 				expect(job.phase).toBe("error");
 				expect(job.errorCode).toBe(
-					fault === "corrupt" ? "source-invalid" : `source-${fault}`,
+					fault.startsWith("corrupt") ? "source-invalid" : `source-${fault}`,
 				);
 				expect(job.recordingVerification).toBeUndefined();
 				expect(uploadConditions).toHaveLength(0);

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

@@ -14,7 +14,10 @@ import { join } from "node:path";
 import { Readable } from "node:stream";
 import { EncodedPacketSink, FilePathSource, Input, MP4 } from "mediabunny";
 import { muxMediaTracksToMp4 } from "../../lib/media-video";
-import { proveRecordingPackets } from "../../lib/recording-packet-proof";
+import {
+	proveRecordingPackets,
+	readRecordingAudioTail,
+} from "../../lib/recording-packet-proof";
 import {
 	RecordingTimingError,
 	readRecordingVideoTiming,
@@ -489,6 +492,33 @@ afterAll(async () => {
 });
 
 describe("encoded recording preservation", () => {
+	test("retains the processing deadline reason when muxing is already cancelled", async () => {
+		const reason = new Error("Recording processing timed out");
+		await expect(
+			muxMediaTracksToMp4(
+				silent,
+				silent,
+				join(directory, "cancelled-mux.mp4"),
+				AbortSignal.abort(reason),
+			),
+		).rejects.toBe(reason);
+	});
+	test("distinguishes malformed audio from unavailable local timing reads", async () => {
+		await expect(
+			readRecordingAudioTail(
+				join(directory, "absent-audio.mp4"),
+				AbortSignal.timeout(5000),
+			),
+		).rejects.toMatchObject({ retryable: true });
+		await expect(
+			readRecordingAudioTail(silent, AbortSignal.abort()),
+		).rejects.toMatchObject({ retryable: true });
+		const malformed = join(directory, "malformed-audio.mp4");
+		await writeFile(malformed, Buffer.alloc(32));
+		await expect(
+			readRecordingAudioTail(malformed, AbortSignal.timeout(5000)),
+		).rejects.toMatchObject({ retryable: false });
+	});
 	test("uses decoded source evidence for tied terminal video samples", async () => {
 		const input = await tiedTimestampSource("packet-tied-terminal.mp4", 2);
 		const output = join(directory, "packet-tied-terminal-output.mp4");

apps/media-server/src/lib/media-video.ts

@@ -2474,7 +2474,7 @@ export async function muxMediaTracksToMp4(
 	outputPath: string,
 	abortSignal?: AbortSignal,
 ): Promise<void> {
-	if (abortSignal?.aborted) throw new Error("Recording mux was cancelled");
+	abortSignal?.throwIfAborted();
 	abortSignal = AbortSignal.any([
 		...(abortSignal ? [abortSignal] : []),
 		AbortSignal.timeout(PROCESS_TIMEOUT_MS),
@@ -2484,7 +2484,7 @@ export async function muxMediaTracksToMp4(
 		abortSignal,
 		timeoutMs: PROCESS_TIMEOUT_MS,
 	});
-	if (abortSignal?.aborted) throw new Error("Recording mux was cancelled");
+	abortSignal?.throwIfAborted();
 	const lastTimestamp = timing.lastTimestampTicks - timing.firstTimestampTicks;
 	// FFmpeg 7 can discard a fragmented MP4's stored final sample duration.
 	const videoTimingFilter = `setts=pts=PTS:dts=DTS:duration=if(eq(PTS-STARTPTS\\,${lastTimestamp})\\,${timing.lastDurationTicks}\\,DURATION)`;

apps/media-server/src/lib/recording-packet-proof.ts

@@ -2,6 +2,7 @@ import { spawn } from "node:child_process";
 import { createHash } from "node:crypto";
 import { lstat } from "node:fs/promises";
 import { isAbsolute } from "node:path";
+import { setTimeout as yieldToEvents } from "node:timers/promises";
 import { EncodedPacketSink, FilePathSource, Input, MP4 } from "mediabunny";
 import {
 	RecordingTimingError,
@@ -121,12 +122,16 @@ export async function readRecordingAudioTail(
 	signal: AbortSignal,
 	countPackets = false,
 ) {
-	const input = new Input({ formats: [MP4], source: new FilePathSource(path) });
-	const dispose = () => input.dispose();
-	signal.addEventListener("abort", dispose, { once: true });
+	const source = new FilePathSource(path).ref();
+	const input = new Input({ formats: [MP4], source });
+	let formatReady = false;
 	try {
+		signal.throwIfAborted();
+		await input.getFormat();
+		formatReady = true;
 		signal.throwIfAborted();
 		const tracks = await input.getAudioTracks();
+		signal.throwIfAborted();
 		if (tracks.length !== 1)
 			throw new Error("Recording audio tracks are ambiguous");
 		const sink = new EncodedPacketSink(tracks[0]);
@@ -159,14 +164,19 @@ export async function readRecordingAudioTail(
 			packet.sequenceNumber < 0
 		)
 			throw new Error("Recording audio duration is not exact");
-		const packetCount = countPackets
-			? (
-					await tracks[0].computePacketStats(Number.POSITIVE_INFINITY, {
-						metadataOnly: true,
-						skipLiveWait: true,
-					})
-				).packetCount
-			: undefined;
+		let packetCount: number | undefined;
+		if (countPackets) {
+			packetCount = 0;
+			for await (const _packet of sink.packets(undefined, undefined, {
+				metadataOnly: true,
+				skipLiveWait: true,
+			})) {
+				signal.throwIfAborted();
+				packetCount++;
+				if (packetCount % 1024 === 0)
+					await yieldToEvents(0, undefined, { signal });
+			}
+		}
 		if (
 			packetCount !== undefined &&
 			(!Number.isSafeInteger(packetCount) || packetCount <= 0)
@@ -181,9 +191,25 @@ export async function readRecordingAudioTail(
 			duration: time(String(ticks), `1/${scale}`),
 			hash: `SHA256:${createHash("sha256").update(packet.data).digest("hex")}`,
 		};
+	} catch (error) {
+		const retryable =
+			signal.aborted ||
+			(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 {
-		signal.removeEventListener("abort", dispose);
-		dispose();
+		// Mediabunny 1.45 disposal leaks rejections after failed format detection and can strand pending reads.
+		if (formatReady) input.dispose();
+		else source.free();
 	}
 }
 

apps/media-server/src/routes/video.ts

@@ -1722,7 +1722,7 @@ function classifySourceError(error: unknown): RecordingErrorCode {
 	if (/HTTP error 404|Server returned 404/.test(error.message))
 		return "source-missing";
 	if (
-		/Decoded recording|Invalid decoded recording|Recording has no decoded|Recording video packets are missing|Recording timing has no video track|does not match the completed local file|Verified recording size/.test(
+		/Decoded recording|Invalid decoded recording|Recording has no decoded|Recording video packets are missing|Recording audio timing is invalid|Recording timing has no video track|does not match the completed local file|Verified recording size/.test(
 			error.message,
 		)
 	)