From 971d225b325e5ee7a3cb6e385f565c2ca64d7fed Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 23 Oct 2026 19:27:20 +0100 Subject: [PATCH 1/3] perf(media-server): parallelise recording verification behind opt-in env settings --- ...recording-verification.integration.test.ts | 371 ++++++++++ .../src/lib/recording-packet-proof.ts | 83 ++- .../src/lib/recording-verification.ts | 642 +++++++++++++++++- apps/media-server/src/routes/video.ts | 10 +- 4 files changed, 1068 insertions(+), 38 deletions(-) diff --git a/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts b/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts index 129c807bdad..5a29ced547a 100644 --- a/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts @@ -2444,6 +2444,377 @@ function objectResponse( return new Response(bytes, { headers }); } +describe("chunked packet-proven verification", () => { + async function segmentedRecording(name: string, gop: number) { + const fragments = await mkdtemp(join(directory, `${name}-`)); + await run([ + "ffmpeg", + "-v", + "error", + "-f", + "lavfi", + "-i", + "testsrc2=size=320x180:rate=30", + "-frames:v", + "120", + "-c:v", + "libx264", + "-preset", + "ultrafast", + "-g", + String(gop), + "-keyint_min", + String(gop), + "-sc_threshold", + "0", + "-pix_fmt", + "yuv420p", + "-f", + "dash", + "-seg_duration", + "0.5", + "-use_template", + "1", + "-use_timeline", + "1", + "-init_seg_name", + "init.mp4", + "-media_seg_name", + "fragment-$Number%05d$.m4s", + join(fragments, "manifest.mpd"), + ]); + const init = join(fragments, "init.mp4"); + const segments = (await readdir(fragments)) + .filter((file) => /^fragment-\d+\.m4s$/.test(file)) + .sort() + .map((file) => join(fragments, file)); + const combined = join(directory, `${name}-combined.mp4`); + await writeFile( + combined, + Buffer.concat( + await Promise.all([init, ...segments].map((path) => readFile(path))), + ), + ); + const output = join(directory, `${name}-output.mp4`); + await muxMediaTracksToMp4(combined, silent, output); + return { init, segments, combined, output }; + } + + const previous = process.env.MEDIA_SERVER_VERIFY_DECODE_CHUNKS; + afterAll(() => { + if (previous === undefined) + delete process.env.MEDIA_SERVER_VERIFY_DECODE_CHUNKS; + else process.env.MEDIA_SERVER_VERIFY_DECODE_CHUNKS = previous; + }); + + async function both( + recording: Awaited>, + ) { + delete process.env.MEDIA_SERVER_VERIFY_DECODE_CHUNKS; + const single = await verifyRemuxedRecording( + recording.combined, + silent, + recording.output, + { requireAudio: false }, + ); + process.env.MEDIA_SERVER_VERIFY_DECODE_CHUNKS = "3"; + const warn = spyOn(console, "warn"); + try { + const chunked = await verifyRemuxedRecording( + recording.combined, + silent, + recording.output, + { requireAudio: false }, + { init: recording.init, segments: recording.segments }, + ); + const fallbacks = warn.mock.calls.filter((call) => + String(call[0]).includes("recording_chunked_decode_fallback"), + ); + return { single, chunked, fallbacks }; + } finally { + warn.mockRestore(); + delete process.env.MEDIA_SERVER_VERIFY_DECODE_CHUNKS; + } + } + + test("returns the same evidence as one full decode", async () => { + const recording = await segmentedRecording("chunk-parity", 15); + expect(recording.segments.length).toBeGreaterThanOrEqual(6); + const { single, chunked, fallbacks } = await both(recording); + expect(fallbacks).toHaveLength(0); + expect(chunked).toEqual(single); + expect(chunked.video.frameCount).toBe(120); + expect(chunked.audio).not.toBeNull(); + }); + + async function misalignedRecording(name: string) { + const fragmented = join(directory, `${name}-fragmented.mp4`); + await run([ + "ffmpeg", + "-v", + "error", + "-f", + "lavfi", + "-i", + "testsrc2=size=320x180:rate=30", + "-frames:v", + "120", + "-c:v", + "libx264", + "-preset", + "ultrafast", + "-g", + "45", + "-keyint_min", + "45", + "-sc_threshold", + "0", + "-pix_fmt", + "yuv420p", + "-movflags", + "empty_moov+default_base_moof+separate_moof", + "-frag_duration", + "400000", + fragmented, + ]); + const bytes = await readFile(fragmented); + const boxes: { type: string; start: number; end: number }[] = []; + for (let offset = 0; offset < bytes.length; ) { + const size = bytes.readUInt32BE(offset); + boxes.push({ + type: bytes.toString("ascii", offset + 4, offset + 8), + start: offset, + end: offset + size, + }); + offset += size; + } + const firstFragment = boxes.findIndex((box) => box.type === "moof"); + const init = join(directory, `${name}-init.mp4`); + await writeFile(init, bytes.subarray(0, boxes[firstFragment].start)); + const segments: string[] = []; + for (let index = firstFragment; index < boxes.length; index += 2) { + const segment = join( + directory, + `${name}-fragment-${segments.length}.m4s`, + ); + await writeFile( + segment, + bytes.subarray( + boxes[index].start, + boxes[index + 1]?.end ?? bytes.length, + ), + ); + segments.push(segment); + } + const output = join(directory, `${name}-output.mp4`); + await muxMediaTracksToMp4(fragmented, silent, output); + return { init, segments, combined: fragmented, output }; + } + + test("falls back to one full decode when fragments do not start on keyframes", async () => { + const recording = await misalignedRecording("chunk-misaligned"); + expect(recording.segments.length).toBeGreaterThanOrEqual(6); + const { single, chunked } = await both(recording); + expect(chunked).toEqual(single); + expect(chunked.video.frameCount).toBe(120); + }); + + test("still rejects a corrupt recording through the full decode", async () => { + const recording = await segmentedRecording("chunk-corrupt", 15); + const target = recording.segments[3]; + const bytes = await readFile(target); + bytes.fill( + 0, + Math.floor(bytes.length / 2), + Math.floor(bytes.length / 2) + 2048, + ); + await writeFile(target, bytes); + const combined = Buffer.concat( + await Promise.all( + [recording.init, ...recording.segments].map((path) => readFile(path)), + ), + ); + await writeFile(recording.combined, combined); + await muxMediaTracksToMp4(recording.combined, silent, recording.output); + 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(); + } finally { + delete process.env.MEDIA_SERVER_VERIFY_DECODE_CHUNKS; + } + }); +}); + +describe("ranged remote byte verification", () => { + const payload = new Uint8Array(3 * 1024 * 1024 + 12_345).map( + (_, index) => (index * 31 + 7) % 251, + ); + const sha256 = createHash("sha256").update(payload).digest("hex"); + const identity = '"ranged-object"'; + + function rangedServer( + respond?: (start: number, end: number, attempt: number) => Response | null, + ) { + const ranges: string[] = []; + const attempts = new Map(); + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request) { + if (request.headers.get("if-match") !== identity) + return new Response("Object changed", { status: 412 }); + const range = request.headers.get("range"); + const match = range?.match(/^bytes=(\d+)-(\d+)$/); + if (!match) + return new Response(payload, { headers: { ETag: identity } }); + const start = Number(match[1]); + const end = Number(match[2]); + if (range !== "bytes=0-0") ranges.push(range ?? ""); + const attempt = (attempts.get(range ?? "") ?? 0) + 1; + attempts.set(range ?? "", attempt); + const custom = + range === "bytes=0-0" ? null : respond?.(start, end, attempt); + if (custom) return custom; + return new Response(payload.slice(start, end + 1), { + status: 206, + headers: { + ETag: identity, + "Content-Range": `bytes ${start}-${end}/${payload.length}`, + }, + }); + }, + }); + return { server, ranges }; + } + + async function verify(port: number) { + return verifyRemoteRecordingBytes( + `http://127.0.0.1:${port}/recording.mp4`, + { + expectedObjectIdentity: identity, + expectedSha256: sha256, + expectedFileSize: payload.length, + }, + ); + } + + const env = { ...process.env }; + beforeAll(() => { + process.env.MEDIA_SERVER_VERIFY_DOWNLOAD_CONCURRENCY = "3"; + process.env.MEDIA_SERVER_VERIFY_DOWNLOAD_PART_MB = "1"; + }); + afterAll(() => { + process.env.MEDIA_SERVER_VERIFY_DOWNLOAD_CONCURRENCY = + env.MEDIA_SERVER_VERIFY_DOWNLOAD_CONCURRENCY; + process.env.MEDIA_SERVER_VERIFY_DOWNLOAD_PART_MB = + env.MEDIA_SERVER_VERIFY_DOWNLOAD_PART_MB; + if (env.MEDIA_SERVER_VERIFY_DOWNLOAD_CONCURRENCY === undefined) + delete process.env.MEDIA_SERVER_VERIFY_DOWNLOAD_CONCURRENCY; + if (env.MEDIA_SERVER_VERIFY_DOWNLOAD_PART_MB === undefined) + delete process.env.MEDIA_SERVER_VERIFY_DOWNLOAD_PART_MB; + }); + + test("hashes every pinned range in order", async () => { + const { server, ranges } = rangedServer(); + try { + const result = await verify(server.port); + expect(result).toEqual({ + objectIdentity: identity, + fileSize: payload.length, + remoteSha256: sha256, + }); + const mb = 1024 * 1024; + expect([...ranges].sort()).toEqual( + [0, 1, 2, 3] + .map( + (index) => + `bytes=${index * mb}-${Math.min(payload.length, (index + 1) * mb) - 1}`, + ) + .sort(), + ); + } finally { + await server.stop(true); + } + }); + + test("rejects a range served from a different object", async () => { + const { server } = rangedServer((start, end) => + start === 2 * 1024 * 1024 + ? new Response(payload.slice(start, end + 1), { + status: 206, + headers: { + ETag: '"replaced-object"', + "Content-Range": `bytes ${start}-${end}/${payload.length}`, + }, + }) + : null, + ); + try { + await expect(verify(server.port)).rejects.toThrow( + "Recording object changed during byte verification", + ); + } finally { + await server.stop(true); + } + }); + + test("rejects a range that does not match its requested bytes", async () => { + const { server } = rangedServer((start, end) => + start === 1024 * 1024 + ? new Response(payload.slice(start, end + 1), { + status: 206, + headers: { + ETag: identity, + "Content-Range": `bytes ${start}-${end - 1}/${payload.length}`, + }, + }) + : null, + ); + try { + await expect(verify(server.port)).rejects.toThrow( + "Recording byte resume range does not match the verified object", + ); + } finally { + await server.stop(true); + } + }); + + test("retries a range after a transient storage error", async () => { + const { server, ranges } = rangedServer((start, _end, attempt) => + start === 0 && attempt === 1 + ? new Response("busy", { status: 503 }) + : null, + ); + try { + expect((await verify(server.port)).remoteSha256).toBe(sha256); + expect( + ranges.filter((range) => range.startsWith("bytes=0-")), + ).toHaveLength(2); + } finally { + await server.stop(true); + } + }); + + test("falls back to one sequential read when storage ignores ranges", async () => { + const { server } = rangedServer( + () => new Response(payload, { headers: { ETag: identity } }), + ); + try { + expect((await verify(server.port)).remoteSha256).toBe(sha256); + } finally { + await server.stop(true); + } + }); +}); + describe("remote recording object identity", () => { test("resumes interrupted byte verification without rereading the verified prefix", async () => { const identity = '"resume-object"'; diff --git a/apps/media-server/src/lib/recording-packet-proof.ts b/apps/media-server/src/lib/recording-packet-proof.ts index 59cf3806084..67cf1ace444 100644 --- a/apps/media-server/src/lib/recording-packet-proof.ts +++ b/apps/media-server/src/lib/recording-packet-proof.ts @@ -381,41 +381,60 @@ export async function proveRecordingPackets( const before = await Promise.all( paths.map((path) => lstat(path, { bigint: true })), ); - const sourceTiming = await readRecordingVideoTiming(videoPath, { - abortSignal: signal, - timeoutMs: 45 * 60_000, - }); - if (sourceTiming.terminalPacketCount !== 1) - throw new Error("Tied terminal samples require decoded source evidence"); - const sourceVideo = await readTrack(videoPath, "video", signal); - const outputVideo = await readTrack(outputPath, "video", signal); - if (JSON.stringify(sourceVideo) !== JSON.stringify(outputVideo)) - throw new RecordingPacketMismatchError( - "Recording encoded video does not preserve the source", - ); - if (audioPath) { - const { tail: sourceTail, ...sourceAudio } = await readTrack( - audioPath, - "audio", - signal, - ); - const { tail: outputTail, ...outputAudio } = await readTrack( - outputPath, - "audio", - signal, - ); - if (JSON.stringify(sourceAudio) !== JSON.stringify(outputAudio)) + // Every read below is independent and the files are pinned by `before`, so + // run them together and still check their results in the original order. + const reads = new AbortController(); + const readSignal = AbortSignal.any([signal, reads.signal]); + const settled = (promise: Promise) => { + promise.catch(() => {}); + return promise; + }; + const timingOptions = { abortSignal: readSignal, timeoutMs: 45 * 60_000 }; + const sourceTimingRead = settled( + readRecordingVideoTiming(videoPath, timingOptions), + ); + const sourceVideoRead = settled(readTrack(videoPath, "video", readSignal)); + const outputVideoRead = settled(readTrack(outputPath, "video", readSignal)); + const sourceAudioRead = audioPath + ? settled(readTrack(audioPath, "audio", readSignal)) + : null; + const outputAudioRead = audioPath + ? settled(readTrack(outputPath, "audio", readSignal)) + : null; + const outputTimingRead = settled( + readRecordingVideoTiming(outputPath, timingOptions), + ); + let sourceTiming: Awaited; + let outputTiming: Awaited; + let sourceVideo: Awaited; + let outputVideo: Awaited; + try { + sourceTiming = await sourceTimingRead; + if (sourceTiming.terminalPacketCount !== 1) + throw new Error("Tied terminal samples require decoded source evidence"); + sourceVideo = await sourceVideoRead; + outputVideo = await outputVideoRead; + if (JSON.stringify(sourceVideo) !== JSON.stringify(outputVideo)) throw new RecordingPacketMismatchError( - "Recording encoded audio does not preserve the source", + "Recording encoded video does not preserve the source", ); - // MP4 edit-list rounding can change the reported AAC tail without changing decoded samples. - if (sourceTail?.duration !== outputTail?.duration) - throw new Error("Recording audio tail requires decoded source evidence"); + if (sourceAudioRead && outputAudioRead) { + const { tail: sourceTail, ...sourceAudio } = await sourceAudioRead; + const { tail: outputTail, ...outputAudio } = await outputAudioRead; + if (JSON.stringify(sourceAudio) !== JSON.stringify(outputAudio)) + throw new RecordingPacketMismatchError( + "Recording encoded audio does not preserve the source", + ); + // MP4 edit-list rounding can change the reported AAC tail without changing decoded samples. + if (sourceTail?.duration !== outputTail?.duration) + throw new Error( + "Recording audio tail requires decoded source evidence", + ); + } + outputTiming = await outputTimingRead; + } finally { + reads.abort(); } - const outputTiming = await readRecordingVideoTiming(outputPath, { - abortSignal: signal, - timeoutMs: 45 * 60_000, - }); const normalizeTiming = (timing: typeof sourceTiming) => ({ packets: timing.packetTimelineSha256, terminalPackets: timing.terminalPacketCount, diff --git a/apps/media-server/src/lib/recording-verification.ts b/apps/media-server/src/lib/recording-verification.ts index 59b6b8547d2..97e472a8ebd 100644 --- a/apps/media-server/src/lib/recording-verification.ts +++ b/apps/media-server/src/lib/recording-verification.ts @@ -1,9 +1,11 @@ import { spawn } from "node:child_process"; import { createHash, type Hash } from "node:crypto"; -import { createReadStream } from "node:fs"; -import { lstat } from "node:fs/promises"; -import { isAbsolute } from "node:path"; +import { createReadStream, createWriteStream } from "node:fs"; +import { lstat, rm } from "node:fs/promises"; +import { availableParallelism } from "node:os"; +import { dirname, isAbsolute, join } from "node:path"; import { Readable } from "node:stream"; +import { pipeline } from "node:stream/promises"; import { setTimeout as sleep } from "node:timers/promises"; import { decodeThreadsFromEnv, PROCESS_TIMEOUT_MS } from "./media-common"; import { @@ -668,6 +670,454 @@ export async function inspectRecordingSources( return { ...result, integrity: result.integrity }; } +export interface RecordingVideoChunkSource { + init: string; + segments: string[]; +} + +export function verifyDecodeChunks( + env: Record = process.env, +): number { + const value = Number(env.MEDIA_SERVER_VERIFY_DECODE_CHUNKS?.trim()); + return Number.isInteger(value) && value > 1 ? Math.min(value, 16) : 1; +} + +class ChunkedDecodeUnavailable extends Error {} + +const DECODED_STREAM_METADATA = + /^#(tb|media_type|sample_rate|dimensions|sar|channel_layout_name) (\d+):/; + +async function collectDecoderLines( + stream: AsyncIterable, + lines: string[], + onLine: () => void, +): Promise { + const decoder = new TextDecoder(); + let pending = ""; + const take = (line: string) => { + const trimmed = line.trim(); + if (!trimmed) return; + lines.push(trimmed); + if (!trimmed.startsWith("#")) onLine(); + }; + for await (const chunk of stream) { + pending += decoder.decode(chunk, { stream: true }); + let newline = pending.indexOf("\n"); + while (newline !== -1) { + if (newline > MAX_OUTPUT_LINE_LENGTH) { + throw new Error("Recording decoder output exceeded its limit"); + } + take(pending.slice(0, newline)); + pending = pending.slice(newline + 1); + newline = pending.indexOf("\n"); + } + if (pending.length > MAX_OUTPUT_LINE_LENGTH) { + throw new Error("Recording decoder output exceeded its limit"); + } + } + take(pending + decoder.decode()); +} + +async function concatFiles(paths: string[], target: string): Promise { + const output = createWriteStream(target); + try { + for (const path of paths) { + await pipeline(createReadStream(path), output, { end: false }); + } + } finally { + await new Promise((resolve, reject) => + output.end((error?: Error | null) => (error ? reject(error) : resolve())), + ); + } +} + +const KEYFRAME_SEARCH_SEGMENTS = 8; + +async function countOutputLines( + command: string, + args: string[], + pattern: RegExp, + signal?: AbortSignal, +): Promise { + const child = spawn(command, args, { + stdio: ["ignore", "pipe", "ignore"], + signal, + timeout: 60_000, + }); + let count = 0; + let pending = ""; + child.stdout.setEncoding("utf8"); + child.stdout.on("data", (text: string) => { + pending += text; + const lines = pending.split("\n"); + pending = lines.pop() ?? ""; + for (const line of lines) if (pattern.test(line)) count++; + }); + const code = await new Promise((resolve) => { + child.once("error", () => resolve(null)); + child.once("close", resolve); + }); + if (pattern.test(pending)) count++; + return code === 0 ? count : null; +} + +// A fragment can open with a sync sample the decoder still cannot start from +// cleanly, so prove it: decoding it alone must yield a frame for every packet. +async function decodesCleanlyFromStart( + init: string, + segment: string, + signal?: AbortSignal, +): Promise { + const input = `concat:${init}|${segment}`; + const [packets, frames] = await Promise.all([ + countOutputLines( + "ffprobe", + [ + "-v", + "error", + "-protocol_whitelist", + "file,concat", + "-select_streams", + "v:0", + "-show_entries", + "packet=flags", + "-of", + "csv=p=0", + input, + ], + /^\S/, + signal, + ), + countOutputLines( + "ffmpeg", + [ + "-hide_banner", + "-nostdin", + "-copyts", + "-v", + "error", + "-xerror", + "-err_detect", + "explode+crccheck", + "-threads", + "1", + "-protocol_whitelist", + "file,concat", + "-f", + "mov", + "-i", + input, + "-map", + "0:v:0", + "-fps_mode", + "passthrough", + "-enc_time_base:v", + "-1", + "-c:v", + "wrapped_avframe", + "-bsf:v", + "setts=pts=N:dts=N:duration=1", + "-f", + "framecrc", + "pipe:1", + ], + /^0,/, + signal, + ), + ]); + return packets !== null && packets > 0 && frames === packets; +} + +// Splits only where a fragment decodes cleanly on its own, searching a few +// fragments past each even split point and merging runs when none does. +async function keyframeAlignedGroups( + source: RecordingVideoChunkSource, + chunkCount: number, + signal?: AbortSignal, +): Promise { + const total = source.segments.length; + const per = Math.ceil(total / chunkCount); + const targets = Array.from( + { length: chunkCount - 1 }, + (_, index) => (index + 1) * per, + ).filter((target) => target < total); + const found = await Promise.all( + targets.map(async (target) => { + for ( + let index = target; + index < Math.min(total, target + KEYFRAME_SEARCH_SEGMENTS); + index++ + ) { + if ( + await decodesCleanlyFromStart( + source.init, + source.segments[index], + signal, + ) + ) + return index; + } + return undefined; + }), + ); + const boundaries = [0]; + for (const index of found) { + if (index !== undefined && index > boundaries[boundaries.length - 1]) + boundaries.push(index); + } + return boundaries.map((start, position) => + source.segments.slice(start, boundaries[position + 1] ?? total), + ); +} + +// Decodes the packet-proven video as independent runs of whole fragments in +// parallel, then replays their frame lines through the same parser and evidence +// checks as one decode. Any decoder complaint falls back to the single decode, +// which stays the judge of whether a recording is corrupt. +async function decodePacketProvenChunks( + outputPath: string, + source: RecordingVideoChunkSource, + chunkCount: number, + options: RecordingVerificationOptions, + packetProof: Awaited>, +): Promise { + const groups = await keyframeAlignedGroups( + source, + chunkCount, + options.abortSignal, + ); + if (groups.length < 2) + throw new ChunkedDecodeUnavailable("no cleanly decodable fragment runs"); + const directory = dirname(outputPath); + const chunkPaths = groups.map((_, index) => + join(directory, `verify-chunk-${index}.mp4`), + ); + const configuredThreads = Number( + decodeThreadsFromEnv("MEDIA_SERVER_VERIFY_DECODE_THREADS", 0), + ); + const threads = String( + Math.max( + 1, + Math.floor( + (configuredThreads > 0 ? configuredThreads : availableParallelism()) / + groups.length, + ), + ), + ); + const base = [ + "-hide_banner", + "-nostdin", + "-copyts", + "-v", + "error", + "-xerror", + "-err_detect", + "explode+crccheck", + "-filter_threads", + "1", + "-filter_complex_threads", + "1", + ]; + const decoders: Awaited>[] = []; + const outputs: string[][] = []; + const readers: Promise[] = []; + let failure: Error | undefined; + let decodedFrames = 0; + let lastAdvance = performance.now(); + let lastReport = 0; + const stopAll = () => { + for (const decoder of decoders) { + if (decoder.exitCode === null) { + try { + decoder.kill("SIGKILL"); + } catch { + failure ??= new Error("Recording decoder termination failed"); + } + } + } + }; + const cancel = () => { + failure ??= new Error("Recording verification was cancelled"); + stopAll(); + }; + options.abortSignal?.addEventListener("abort", cancel, { once: true }); + const timeout = setTimeout(() => { + failure ??= new RecordingVerificationError( + "Recording verification timed out", + true, + ); + stopAll(); + }, options.timeoutMs ?? MAX_LOCAL_VERIFICATION_MS); + const stall = setInterval(() => { + if (performance.now() - lastAdvance > DECODER_STALL_MS) { + failure ??= new RecordingVerificationError( + "Recording decoder stopped making progress", + true, + ); + stopAll(); + } + }, 1000); + const onFrame = () => { + decodedFrames++; + lastAdvance = performance.now(); + if (performance.now() - lastReport >= 1000) { + lastReport = performance.now(); + options.onProgress?.({ + frames: decodedFrames, + totalFrames: packetProof.videoPackets, + }); + } + }; + try { + await Promise.all( + groups.map((group, index) => + concatFiles([source.init, ...group], chunkPaths[index]), + ), + ); + const jobs = [ + ...chunkPaths.map((path) => [ + ...base, + "-threads", + threads, + "-protocol_whitelist", + "file", + "-f", + "mov", + "-i", + path, + "-map", + "0:v:0", + "-fps_mode", + "passthrough", + "-enc_time_base:v", + "-1", + "-c:v", + "wrapped_avframe", + "-bsf:v", + "setts=pts=N:dts=N:duration=1", + "-max_interleave_delta", + "1", + "-f", + "framecrc", + "pipe:1", + ]), + ...(packetProof.hasAudio + ? [ + [ + ...base, + "-threads", + "1", + "-protocol_whitelist", + "file", + "-f", + "mov", + "-i", + outputPath, + "-map", + "0:a:0", + "-c:a", + "pcm_f64le", + "-threads", + "1", + "-max_interleave_delta", + "1", + "-f", + "framecrc", + "pipe:1", + ], + ] + : []), + ]; + const errors: Promise<{ error: string }>[] = []; + for (const args of jobs) { + options.abortSignal?.throwIfAborted(); + const decoder = registerSubprocess(await spawnRecordingDecoder(args, 0)); + decoders.push(decoder); + const lines: string[] = []; + outputs.push(lines); + const frames = collectDecoderLines(decoder.stdout, lines, onFrame); + const stderr = readDecoderErrors(decoder.stderr, outputPath); + frames.catch(stopAll); + stderr.catch(stopAll); + readers.push(frames, stderr); + errors.push(stderr); + } + const exits = await Promise.all(decoders.map((decoder) => decoder.exited)); + await Promise.all(readers); + if (failure) throw failure; + const stderr = await Promise.all(errors); + for (const [index, decoder] of decoders.entries()) { + if (exits[index] !== 0 || stderr[index].error || decoder.signalCode) + throw new ChunkedDecodeUnavailable( + stderr[index].error || `decoder exited with ${exits[index]}`, + ); + } + const streams = new Map(); + const videoOutputs = outputs.slice(0, chunkPaths.length); + const metadataOf = (lines: string[]) => + lines.filter((line) => DECODED_STREAM_METADATA.test(line)).join("\n"); + const videoMetadata = metadataOf(videoOutputs[0]); + let offset = 0; + for (const [index, lines] of videoOutputs.entries()) { + if (index > 0 && metadataOf(lines) !== videoMetadata) + throw new ChunkedDecodeUnavailable("chunk stream parameters differ"); + let frames = 0; + for (const line of lines) { + if (line.startsWith("#")) { + if (index === 0) decodeLine(line, streams, true); + continue; + } + const columns = line.split(",").map((value) => value.trim()); + if (columns[0] !== "0") + throw new ChunkedDecodeUnavailable("unexpected chunk stream"); + const dts = Number(columns[1]); + const pts = Number(columns[2]); + if (dts !== frames || pts !== frames) + throw new ChunkedDecodeUnavailable("chunk frames are not contiguous"); + columns[1] = String(dts + offset); + columns[2] = String(pts + offset); + decodeLine(columns.join(", "), streams, true); + frames++; + } + offset += frames; + } + // A decoder silently skips frames until a keyframe, so a short count is a split problem, not proof. + if (offset !== packetProof.videoPackets) + throw new ChunkedDecodeUnavailable("chunk frame count differs"); + if (packetProof.hasAudio) { + for (const line of outputs[chunkPaths.length]) { + if (line.startsWith("#")) { + const metadata = line.match(DECODED_STREAM_METADATA); + if (metadata?.[2] === "0") + decodeLine(line.replace(/^(#\w+) 0:/, "$1 1:"), streams, true); + continue; + } + if (!line.startsWith("0,")) + throw new ChunkedDecodeUnavailable("unexpected audio stream"); + decodeLine(`1${line.slice(1)}`, streams, true); + } + } + return validateEvidence( + streams, + options, + true, + packetProof.outputTiming, + false, + ); + } finally { + clearTimeout(timeout); + clearInterval(stall); + options.abortSignal?.removeEventListener("abort", cancel); + stopAll(); + await Promise.allSettled([ + ...decoders.map((decoder) => decoder.exited), + ...readers, + ]); + for (const decoder of decoders) unregisterSubprocess(decoder); + await Promise.all(chunkPaths.map((path) => rm(path, { force: true }))); + } +} + export async function verifyRemuxedRecording( videoPath: string, audioPath: string | null, @@ -676,6 +1126,7 @@ export async function verifyRemuxedRecording( RecordingVerificationOptions, "requireAudio" | "abortSignal" | "timeoutMs" | "onProgress" >, + videoChunks?: RecordingVideoChunkSource, ): Promise { const signal = AbortSignal.any([ ...(options.abortSignal ? [options.abortSignal] : []), @@ -708,9 +1159,45 @@ export async function verifyRemuxedRecording( throw new Error("Recording source preservation was not verified"); return { ...result, sourcePreserved: true }; } - const result = await decodeRecording( + const decodeOptions = { + ...options, + timeoutMs: options.timeoutMs ?? MAX_LOCAL_VERIFICATION_MS, + }; + const chunkCount = verifyDecodeChunks(); + let result: RecordingVerificationResult | undefined; + if ( + videoChunks && + chunkCount > 1 && + videoChunks.segments.length >= chunkCount * 2 + ) { + try { + result = await decodePacketProvenChunks( + outputPath, + videoChunks, + chunkCount, + decodeOptions, + proof, + ); + } catch (error) { + signal.throwIfAborted(); + if ( + error instanceof RecordingVerificationError && + error.message === "Recording verification timed out" + ) + throw error; + console.warn( + JSON.stringify({ + event: "recording_chunked_decode_fallback", + reason: (error instanceof Error ? error.message : String(error)) + .replace(/https?:\/\/\S+/g, "") + .slice(0, 300), + }), + ); + } + } + result ??= await decodeRecording( outputPath, - { ...options, timeoutMs: options.timeoutMs ?? MAX_LOCAL_VERIFICATION_MS }, + decodeOptions, undefined, undefined, proof, @@ -1188,12 +1675,157 @@ async function readRemoteIdentity( } } +const DEFAULT_RANGED_VERIFY_PART_MB = 16; + +export function verifyDownloadConcurrency( + env: Record = process.env, +): number { + const value = Number(env.MEDIA_SERVER_VERIFY_DOWNLOAD_CONCURRENCY?.trim()); + return Number.isInteger(value) && value > 1 ? Math.min(value, 16) : 1; +} + +export function verifyDownloadPartBytes( + env: Record = process.env, +): number { + const value = Number(env.MEDIA_SERVER_VERIFY_DOWNLOAD_PART_MB?.trim()); + const megabytes = + Number.isInteger(value) && value >= 1 && value <= 64 + ? value + : DEFAULT_RANGED_VERIFY_PART_MB; + return megabytes * 1024 * 1024; +} + +class RangedVerificationUnsupported extends Error {} + +// Same proof as the sequential read (every byte, hashed in order, pinned to one +// ETag), fetched as concurrent ranges so large uploads re-read at network speed. +async function hashRemoteRecordingRanged( + input: string, + objectIdentity: string, + fileSize: number, + abortSignal: AbortSignal, + concurrency: number, + partBytes: number, +): Promise { + const reads = new AbortController(); + const signal = AbortSignal.any([abortSignal, reads.signal]); + const partCount = Math.ceil(fileSize / partBytes); + const readPart = async (index: number): Promise => { + const start = index * partBytes; + const end = Math.min(fileSize, start + partBytes) - 1; + for (let attempt = 0; ; attempt++) { + signal.throwIfAborted(); + let response: Response | undefined; + try { + try { + response = await fetchMedia(input, { + headers: { + "If-Match": objectIdentity, + "X-Cap-Recording-Verification": "1", + Range: `bytes=${start}-${end}`, + }, + signal, + }); + } catch (error) { + if (error instanceof MediaTransferBudgetError) throw error; + throw new RecordingVerificationError( + "Recording bytes could not be read", + true, + ); + } + if (response.status === 200) throw new RangedVerificationUnsupported(); + if (response.status !== 206) { + throw new RecordingVerificationError( + `Recording byte verification failed: ${response.status}`, + response.status === 408 || + response.status === 429 || + response.status >= 500, + ); + } + if (response.headers.get("etag") !== objectIdentity || !response.body) + throw new Error("Recording object changed during byte verification"); + if ( + response.headers.get("content-range") !== + `bytes ${start}-${end}/${fileSize}` + ) + throw new Error( + "Recording byte resume range does not match the verified object", + ); + let bytes: Uint8Array; + try { + bytes = new Uint8Array(await response.arrayBuffer()); + } catch { + throw new RecordingVerificationError( + "Recording bytes could not be read completely", + true, + ); + } + if (bytes.byteLength !== end - start + 1) + throw new RecordingVerificationError( + "Recording object size changed during byte verification", + false, + ); + return bytes; + } catch (error) { + signal.throwIfAborted(); + if ( + !(error instanceof RecordingVerificationError) || + !error.retryable || + attempt >= 3 + ) + throw error; + } finally { + if (!response?.body?.locked) + await response?.body?.cancel().catch(() => {}); + } + await sleep(250 * 2 ** attempt, undefined, { signal }); + } + }; + const hash = createHash("sha256"); + const inFlight = new Map>(); + let requested = 0; + try { + for (let index = 0; index < partCount; index++) { + while (requested < partCount && requested < index + concurrency) { + const part = readPart(requested); + part.catch(() => {}); + inFlight.set(requested, part); + requested++; + } + const bytes = await inFlight.get(index); + inFlight.delete(index); + if (!bytes) throw new Error("Recording byte range was not requested"); + hash.update(bytes); + } + return hash.digest("hex"); + } finally { + reads.abort(); + await Promise.allSettled(inFlight.values()); + } +} + async function hashRemoteRecording( input: string, objectIdentity: string, fileSize: number, abortSignal: AbortSignal, ): Promise { + const concurrency = verifyDownloadConcurrency(); + const partBytes = verifyDownloadPartBytes(); + if (concurrency > 1 && fileSize > partBytes) { + try { + return await hashRemoteRecordingRanged( + input, + objectIdentity, + fileSize, + abortSignal, + concurrency, + partBytes, + ); + } catch (error) { + if (!(error instanceof RangedVerificationUnsupported)) throw error; + } + } const hash = createHash("sha256"); let bytesRead = 0; for (let attempt = 0; ; attempt++) { diff --git a/apps/media-server/src/routes/video.ts b/apps/media-server/src/routes/video.ts index 0d7f78dbfbd..30c23e42117 100644 --- a/apps/media-server/src/routes/video.ts +++ b/apps/media-server/src/routes/video.ts @@ -84,6 +84,7 @@ import { RecordingTimingError } from "../lib/recording-timing"; import { hashRecordingFile, isRetryableRecordingVerificationError, + verifyDecodeChunks, verifyRecording, verifyRemoteRecording, verifyRemoteRecordingBytes, @@ -3005,7 +3006,12 @@ async function muxSegmentsAsync( abortController.signal, ), ); - await rm(videoDir, { recursive: true, force: true }); + // Parallel verification decodes runs of these fragments; otherwise free them now. + const videoChunks = + verifyDecodeChunks() > 1 + ? { init: videoInitPath, segments: videoSegmentFiles } + : undefined; + if (!videoChunks) await rm(videoDir, { recursive: true, force: true }); let combinedAudioPath: string | null = null; @@ -3081,9 +3087,11 @@ async function muxSegmentsAsync( }); }, }, + videoChunks, ), ), ); + if (videoChunks) await rm(videoDir, { recursive: true, force: true }); logVideoEvent("video_mux_verification_complete", { jobId, videoId, From 087fa98ec06c766498490dd6df7a7499d9be0f65 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 23 Oct 2026 19:34:25 +0100 Subject: [PATCH 2/3] fix(media-server): bound ranged readback bodies and retry short ranges --- ...recording-verification.integration.test.ts | 43 +++++++++++++++++++ .../src/lib/recording-verification.ts | 23 +++++++--- 2 files changed, 60 insertions(+), 6 deletions(-) diff --git a/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts b/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts index 5a29ced547a..cdac180b263 100644 --- a/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts @@ -2803,6 +2803,49 @@ describe("ranged remote byte verification", () => { } }); + test("rejects an oversized range body without buffering it whole", async () => { + const { server } = rangedServer((start, end) => + start === 1024 * 1024 + ? new Response(payload, { + status: 206, + headers: { + ETag: identity, + "Content-Range": `bytes ${start}-${end}/${payload.length}`, + }, + }) + : null, + ); + try { + await expect(verify(server.port)).rejects.toThrow( + "Recording object size changed during byte verification", + ); + } finally { + await server.stop(true); + } + }); + + test("retries a range whose body ends early", async () => { + const { server, ranges } = rangedServer((start, end, attempt) => + start === 2 * 1024 * 1024 && attempt === 1 + ? new Response(payload.slice(start, start + 1000), { + status: 206, + headers: { + ETag: identity, + "Content-Range": `bytes ${start}-${end}/${payload.length}`, + }, + }) + : null, + ); + try { + expect((await verify(server.port)).remoteSha256).toBe(sha256); + expect( + ranges.filter((range) => range.startsWith(`bytes=${2 * 1024 * 1024}-`)), + ).toHaveLength(2); + } finally { + await server.stop(true); + } + }); + test("falls back to one sequential read when storage ignores ranges", async () => { const { server } = rangedServer( () => new Response(payload, { headers: { ETag: identity } }), diff --git a/apps/media-server/src/lib/recording-verification.ts b/apps/media-server/src/lib/recording-verification.ts index 97e472a8ebd..7941d5fec54 100644 --- a/apps/media-server/src/lib/recording-verification.ts +++ b/apps/media-server/src/lib/recording-verification.ts @@ -1751,19 +1751,30 @@ async function hashRemoteRecordingRanged( throw new Error( "Recording byte resume range does not match the verified object", ); - let bytes: Uint8Array; + const expected = end - start + 1; + const bytes = new Uint8Array(expected); + let received = 0; try { - bytes = new Uint8Array(await response.arrayBuffer()); - } catch { + for await (const chunk of response.body) { + if (received + chunk.byteLength > expected) + throw new RecordingVerificationError( + "Recording object size changed during byte verification", + false, + ); + bytes.set(chunk, received); + received += chunk.byteLength; + } + } catch (error) { + if (error instanceof RecordingVerificationError) throw error; throw new RecordingVerificationError( "Recording bytes could not be read completely", true, ); } - if (bytes.byteLength !== end - start + 1) + if (received !== expected) throw new RecordingVerificationError( - "Recording object size changed during byte verification", - false, + "Recording bytes could not be read completely", + true, ); return bytes; } catch (error) { From 237b874d5e3d397df6b085f3787efa78f8a5e9a9 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 23 Oct 2026 19:52:26 +0100 Subject: [PATCH 3/3] fix(media-server): stop chunk copies on cancellation and assert the verification fallback --- .../recording-verification.integration.test.ts | 5 +++-- .../src/lib/recording-verification.ts | 18 +++++++++++++++--- 2 files changed, 18 insertions(+), 5 deletions(-) diff --git a/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts b/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts index cdac180b263..618cfa77956 100644 --- a/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts @@ -2614,7 +2614,8 @@ describe("chunked packet-proven verification", () => { test("falls back to one full decode when fragments do not start on keyframes", async () => { const recording = await misalignedRecording("chunk-misaligned"); expect(recording.segments.length).toBeGreaterThanOrEqual(6); - const { single, chunked } = await both(recording); + const { single, chunked, fallbacks } = await both(recording); + expect(fallbacks).toHaveLength(1); expect(chunked).toEqual(single); expect(chunked.video.frameCount).toBe(120); }); @@ -2695,7 +2696,7 @@ describe("ranged remote byte verification", () => { return { server, ranges }; } - async function verify(port: number) { + async function verify(port: number | undefined) { return verifyRemoteRecordingBytes( `http://127.0.0.1:${port}/recording.mp4`, { diff --git a/apps/media-server/src/lib/recording-verification.ts b/apps/media-server/src/lib/recording-verification.ts index 7941d5fec54..dcd7c5346f4 100644 --- a/apps/media-server/src/lib/recording-verification.ts +++ b/apps/media-server/src/lib/recording-verification.ts @@ -718,11 +718,19 @@ async function collectDecoderLines( take(pending + decoder.decode()); } -async function concatFiles(paths: string[], target: string): Promise { +async function concatFiles( + paths: string[], + target: string, + signal?: AbortSignal, +): Promise { const output = createWriteStream(target); try { for (const path of paths) { - await pipeline(createReadStream(path), output, { end: false }); + signal?.throwIfAborted(); + await pipeline(createReadStream(path), output, { + end: false, + ...(signal ? { signal } : {}), + }); } } finally { await new Promise((resolve, reject) => @@ -971,7 +979,11 @@ async function decodePacketProvenChunks( try { await Promise.all( groups.map((group, index) => - concatFiles([source.init, ...group], chunkPaths[index]), + concatFiles( + [source.init, ...group], + chunkPaths[index], + options.abortSignal, + ), ), ); const jobs = [