perf: faster Instant processing and early playback for everyone

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

@@ -909,10 +909,10 @@ describe("media routes real-world integration tests", () => {
 		let hashSettled = false;
 		const hashing = spyOn(
 			recordingVerification,
-			"hashRecordingFile",
+			"hashRecordingUpload",
 		).mockImplementation(
-			(_path, signal) =>
-				new Promise<string>((_resolve, reject) => {
+			(_path, _upload, signal) =>
+				new Promise<{ sha256: string }>((_resolve, reject) => {
 					signal?.addEventListener("abort", () => {
 						hashAborted = true;
 						setTimeout(() => {
@@ -948,12 +948,12 @@ describe("media routes real-world integration tests", () => {
 
 	// Hashing now overlaps the decode check, so whichever check reads the changed bytes first refuses it.
 	test("refuses a local output changed while it is decoded and hashed", async () => {
-		const originalHash = recordingVerification.hashRecordingFile;
+		const originalHash = recordingVerification.hashRecordingUpload;
 		const hashing = spyOn(
 			recordingVerification,
-			"hashRecordingFile",
-		).mockImplementation(async (path, signal) => {
-			const digest = await originalHash(path, signal);
+			"hashRecordingUpload",
+		).mockImplementation(async (path, upload, signal) => {
+			const digest = await originalHash(path, upload, signal);
 			const bytes = await readFile(path);
 			bytes[bytes.length - 1] ^= 1;
 			await writeFile(path, bytes);

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

@@ -26,6 +26,7 @@ import {
 } from "../../lib/recording-timing";
 import {
 	decodePreverifiedRun,
+	hashRecordingUpload,
 	inspectRecordingSources,
 	isRetryableRecordingVerificationError,
 	type PreverifiedVideoRun,
@@ -34,6 +35,7 @@ import {
 	verifyRemoteRecordingBytes,
 	verifyRemuxedRecording,
 } from "../../lib/recording-verification";
+import { getActiveSubprocessCount } from "../../lib/subprocess";
 
 const FIXTURES = join(import.meta.dir, "..", "fixtures");
 let directory: string;
@@ -2974,6 +2976,107 @@ describe("chunked packet-proven verification", () => {
 		}
 	});
 
+	test("a packet mismatch wins over the fragment decode it runs beside", async () => {
+		const recording = await segmentedRecording("chunk-mismatch", 15);
+		const bytes = await readFile(recording.output);
+		const mdat = bytes.indexOf(Buffer.from("mdat"));
+		if (mdat < 0) throw new Error("Fixture has no media data");
+		bytes[mdat + 64] ^= 0xff;
+		await writeFile(recording.output, bytes);
+		const active = getActiveSubprocessCount();
+		process.env.MEDIA_SERVER_VERIFY_DECODE_CHUNKS = "3";
+		try {
+			await expect(
+				verifyRemuxedRecording(
+					recording.combined,
+					silent,
+					recording.output,
+					{ requireAudio: false },
+					{ init: recording.init, segments: recording.segments },
+				),
+			).rejects.toThrow(
+				/video does not preserve the source \(first difference at packet \d+: data differs\)/,
+			);
+			expect(getActiveSubprocessCount()).toBe(active);
+		} finally {
+			delete process.env.MEDIA_SERVER_VERIFY_DECODE_CHUNKS;
+		}
+	});
+
+	test("decodes the fragments while the output is still being written", async () => {
+		const recording = await segmentedRecording("chunk-early", 15);
+		process.env.MEDIA_SERVER_VERIFY_DECODE_CHUNKS = "3";
+		try {
+			const expected = await verifyRemuxedRecording(
+				recording.combined,
+				silent,
+				recording.output,
+				{ requireAudio: false },
+				{ init: recording.init, segments: recording.segments },
+			);
+			const pending = join(directory, "chunk-early-pending.mp4");
+			let written!: () => void;
+			const outputReady = new Promise<void>((resolve) => {
+				written = resolve;
+			});
+			const early = verifyRemuxedRecording(
+				recording.combined,
+				silent,
+				pending,
+				{ requireAudio: false },
+				{ init: recording.init, segments: recording.segments },
+				[],
+				outputReady,
+			);
+			const warn = spyOn(console, "warn");
+			let result: Awaited<typeof early>;
+			try {
+				await new Promise((resolve) => setTimeout(resolve, 200));
+				await muxMediaTracksToMp4(recording.combined, silent, pending);
+				written();
+				result = await early;
+				expect(
+					warn.mock.calls.some((call) =>
+						String(call[0]).includes("recording_chunked_decode_fallback"),
+					),
+				).toBe(false);
+			} finally {
+				warn.mockRestore();
+			}
+			expect(result.sourcePreserved).toBe(true);
+			expect(result.video.frameCount).toBe(expected.video.frameCount);
+			expect(result.video.duration).toBe(expected.video.duration);
+			expect(Boolean(result.audio)).toBe(Boolean(expected.audio));
+		} finally {
+			delete process.env.MEDIA_SERVER_VERIFY_DECODE_CHUNKS;
+		}
+	});
+
+	test("a failed output write fails verification as itself and stops the decoders", async () => {
+		const recording = await segmentedRecording("chunk-write-failed", 15);
+		const active = getActiveSubprocessCount();
+		process.env.MEDIA_SERVER_VERIFY_DECODE_CHUNKS = "3";
+		try {
+			const failure = new Error("Remux failed");
+			await expect(
+				verifyRemuxedRecording(
+					recording.combined,
+					silent,
+					join(directory, "chunk-write-failed-missing.mp4"),
+					{ requireAudio: false },
+					{ init: recording.init, segments: recording.segments },
+					[],
+					new Promise((_resolve, reject) =>
+						setTimeout(() => reject(failure), 100),
+					),
+				),
+			).rejects.toBe(failure);
+			expect(getActiveSubprocessCount()).toBe(active);
+		} finally {
+			delete process.env.MEDIA_SERVER_VERIFY_DECODE_CHUNKS;
+		}
+	});
+
 	async function preverify(
 		recording: { init: string; segments: string[] },
 		from: number,
@@ -3153,6 +3256,45 @@ describe("chunked packet-proven verification", () => {
 	});
 });
 
+describe("upload content tags", () => {
+	const md5 = (bytes: Uint8Array) => createHash("md5").update(bytes).digest();
+	const multipartTag = (bytes: Uint8Array, partSize: number) => {
+		const parts: Buffer[] = [];
+		for (let start = 0; start < bytes.length; start += partSize)
+			parts.push(md5(bytes.subarray(start, start + partSize)));
+		return `"${createHash("md5").update(Buffer.concat(parts)).digest("hex")}-${parts.length}"`;
+	};
+
+	test.each([
+		["uneven", 3 * 1024 * 1024 + 12_345],
+		["part aligned", 2 * 1024 * 1024],
+		["one partial part", 777],
+	])("matches the S3 ETags of a %s file", async (_, size) => {
+		const bytes = new Uint8Array(size).map(
+			(_, index) => (index * 13 + 5) % 253,
+		);
+		const work = await mkdtemp(join(tmpdir(), "upload-tag-"));
+		try {
+			const path = join(work, "result.mp4");
+			await writeFile(path, bytes);
+			const sha256 = createHash("sha256").update(bytes).digest("hex");
+			expect(await hashRecordingUpload(path, {})).toEqual({
+				sha256,
+				contentIdentity: `"${md5(bytes).toString("hex")}"`,
+			});
+			expect(
+				await hashRecordingUpload(path, { partSize: 1024 * 1024 }),
+			).toEqual({
+				sha256,
+				contentIdentity: multipartTag(bytes, 1024 * 1024),
+			});
+			expect(await hashRecordingUpload(path, undefined)).toEqual({ sha256 });
+		} finally {
+			await rm(work, { recursive: true, force: true });
+		}
+	});
+});
+
 describe("ranged remote byte verification", () => {
 	const payload = new Uint8Array(3 * 1024 * 1024 + 12_345).map(
 		(_, index) => (index * 31 + 7) % 251,
@@ -3245,6 +3387,99 @@ describe("ranged remote byte verification", () => {
 		}
 	});
 
+	test("trusts an ETag that is the uploaded file's own content tag", async () => {
+		const contentIdentity = `"${createHash("md5").update(payload).digest("hex")}"`;
+		const requests: string[] = [];
+		const server = Bun.serve({
+			hostname: "127.0.0.1",
+			port: 0,
+			fetch(request) {
+				requests.push(request.headers.get("range") ?? "full");
+				if (request.headers.get("if-match") !== contentIdentity)
+					return new Response("Object changed", { status: 412 });
+				return new Response(payload.slice(0, 1), {
+					status: 206,
+					headers: {
+						ETag: contentIdentity,
+						"Content-Range": `bytes 0-0/${payload.length}`,
+					},
+				});
+			},
+		});
+		try {
+			const result = await verifyRemoteRecordingBytes(
+				`http://127.0.0.1:${server.port}/recording.mp4`,
+				{
+					expectedObjectIdentity: contentIdentity,
+					expectedSha256: sha256,
+					expectedFileSize: payload.length,
+					uploadedContentIdentity: contentIdentity,
+				},
+			);
+			expect(result).toEqual({
+				objectIdentity: contentIdentity,
+				fileSize: payload.length,
+				remoteSha256: sha256,
+			});
+			expect(requests).toEqual(["bytes=0-0"]);
+		} finally {
+			await server.stop(true);
+		}
+	});
+
+	test("re-reads every byte when the ETag is not the uploaded content tag", async () => {
+		const { server, ranges } = rangedServer();
+		try {
+			const result = await verifyRemoteRecordingBytes(
+				`http://127.0.0.1:${server.port}/recording.mp4`,
+				{
+					expectedObjectIdentity: identity,
+					expectedSha256: sha256,
+					expectedFileSize: payload.length,
+					uploadedContentIdentity: `"${createHash("md5").update(payload).digest("hex")}"`,
+				},
+			);
+			expect(result.remoteSha256).toBe(sha256);
+			expect(ranges).toHaveLength(4);
+		} finally {
+			await server.stop(true);
+		}
+	});
+
+	test("rejects a content-tagged object whose size differs", async () => {
+		const contentIdentity = `"${createHash("md5").update(payload).digest("hex")}"`;
+		const server = Bun.serve({
+			hostname: "127.0.0.1",
+			port: 0,
+			fetch() {
+				return new Response(payload.slice(0, 1), {
+					status: 206,
+					headers: {
+						ETag: contentIdentity,
+						"Content-Range": `bytes 0-0/${payload.length + 1}`,
+					},
+				});
+			},
+		});
+		try {
+			await expect(
+				verifyRemoteRecordingBytes(
+					`http://127.0.0.1:${server.port}/recording.mp4`,
+					{
+						expectedObjectIdentity: contentIdentity,
+						expectedSha256: sha256,
+						expectedFileSize: payload.length,
+						uploadedContentIdentity: contentIdentity,
+					},
+				),
+			).rejects.toThrow(
+				"Uploaded recording size does not match the muxed file",
+			);
+		} finally {
+			await server.stop(true);
+		}
+	});
+
 	test("rejects a range served from a different object", async () => {
 		const { server } = rangedServer((start, end) =>
 			start === 2 * 1024 * 1024

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

@@ -22,7 +22,7 @@ interface PacketStream {
 const STREAM_FIELDS =
 	"index,start_pts,codec_type,codec_name,profile,level,codec_tag_string,width,height,sample_aspect_ratio,pix_fmt,color_range,color_space,color_transfer,color_primaries,chroma_location,field_order,refs,sample_fmt,sample_rate,channels,channel_layout,time_base,extradata_hash";
 
-async function runInspector(
+export async function runInspector(
 	command: string,
 	args: string[],
 	signal: AbortSignal,
@@ -245,7 +245,11 @@ class PacketLog {
 		this.pts[this.length] = seconds(pts);
 		this.dts[this.length] = seconds(dts);
 		this.sizes[this.length] = Number(size);
-		this.hashes[this.length] = Number.parseInt(dataHash.slice(7, 15), 16);
+		const digest = dataHash.indexOf(":") + 1;
+		this.hashes[this.length] = Number.parseInt(
+			dataHash.slice(digest, digest + 8),
+			16,
+		);
 		this.keyframes[this.length] =
 			(flags?.includes("K") ? 1 : 0) | (flags?.includes("D") ? 2 : 0);
 		this.length++;
@@ -516,14 +520,21 @@ async function readTrack(
 		);
 		hash.update("\n");
 	};
+	// Source and output packets are hashed by the same ffprobe and only compared with each other, to catch a
+	// remux that changed a packet; FFmpeg's SHA-256 made this read the slowest part of verifying a long video.
+	// Audio keeps SHA-256 because its tail is checked against a hash computed outside ffprobe.
+	const packetHash =
+		kind === "video"
+			? { name: "murmur3", pattern: /^murmur3:[a-f0-9]{32}$/ }
+			: { name: "sha256", pattern: /^SHA256:[a-f0-9]{64}$/ };
 	await probe(
 		path,
 		[
 			"-select_streams",
 			selection,
 			"-show_packets",
 			"-show_data_hash",
-			"sha256",
+			packetHash.name,
 			"-show_entries",
 			"packet=stream_index,pts,dts,duration,size,flags,data_hash:packet_side_data",
 			"-of",
@@ -547,7 +558,7 @@ async function readTrack(
 			if (
 				Number(fields.stream_index) !== index ||
 				!/^[1-9]\d*$/.test(fields.size ?? "") ||
-				!/^SHA256:[a-f0-9]{64}$/.test(fields.data_hash ?? "")
+				!packetHash.pattern.test(fields.data_hash ?? "")
 			)
 				throw new Error("Recording packet content is incomplete");
 			if (pendingPacket) commitPacket(pendingPacket, fields.dts);