fix(audio): prime Opus decoders by advertised warmup

js/hang/src/util/opus.ts

@@ -6,6 +6,9 @@
 // 44100 is never a valid decoder config. Chrome hides this by ignoring the configured rate; Safari
 // trusts it and fails every decode.
 
+/** The decoder pre-roll in milliseconds a joining Opus viewer discards, per RFC 7845 section 4.6. */
+export const WARMUP = 80;
+
 /** Full-band Opus: its highest rate, and the one to use when the source rate is unknown. */
 export const DEFAULT_SAMPLE_RATE = 48_000;
 

js/justfile

@@ -101,3 +101,8 @@ arrivals-bench:
 clock-bench:
     bun install --frozen-lockfile
     bun clock/bench/timers.ts
+
+# Sweep decoder count against PCM callback size during audio warmup.
+audio-warmup-bench:
+    bun install --frozen-lockfile
+    bun watch/bench/warmup.ts

js/publish/src/audio/encoder.test.ts

@@ -25,6 +25,7 @@ describe("resolve", () => {
 	test("defaults Opus to 20ms", () => {
 		const resolved = resolve(captured, "opus");
 		expect(resolved.frameDuration).toBe(Time.Micro(20_000));
+		expect(Number(resolved.catalog.warmup)).toBe(80);
 		expect(resolved.catalog.jitter).toBeUndefined();
 	});
 
@@ -54,6 +55,7 @@ describe("resolve", () => {
 	test("leaves AAC without a frame duration", () => {
 		const resolved = resolve(captured, "aac");
 		expect(resolved.frameDuration).toBeUndefined();
+		expect(resolved.catalog.warmup).toBeUndefined();
 		expect(resolved.catalog.jitter).toBeUndefined();
 	});
 });

js/publish/src/audio/encoder.ts

@@ -813,6 +813,7 @@ export function resolve(captured: Format, selected: Codec): Resolved {
 	return {
 		catalog: {
 			codec: "opus",
+			warmup: Catalog.u53(Util.Opus.WARMUP),
 			sampleRate,
 			numberOfChannels,
 			bitrate: Catalog.u53(codec.bitrate ?? captured.channelCount * OPUS_BITRATE_PER_CHANNEL),

js/watch/bench/warmup.ts

@@ -0,0 +1,37 @@
+import { Time } from "@moq/net";
+import { Terminal } from "../src/audio/terminal";
+
+// Decoder count × callback size checks that trimming costs a span calculation,
+// independent of the PCM payload size. Periodic gaps exercise the priming path.
+const CALLBACKS = 4096;
+console.log("decoders,frames,warmup_us,ns_per_callback");
+let retained = 0;
+for (const count of [1, 16, 64]) {
+	for (const frames of [120, 960, 2880]) {
+		for (const warmup of [0, 80_000]) {
+			const timings = [];
+			for (let repeat = 0; repeat < 3; repeat++) {
+				const decoders = Array.from({ length: count }, () => {
+					const decoder = new Terminal();
+					decoder.clear(0, Time.Micro(warmup));
+					return decoder;
+				});
+				const start = performance.now();
+				for (let callback = 0; callback < CALLBACKS; callback++) {
+					const timestamp = Time.Micro((callback * frames * 1_000_000) / 48_000);
+					const read = { kind: "frame" as const, group: callback, frame: { timestamp } };
+					const sample = { timestamp, sampleRate: 48_000, numberOfFrames: frames };
+					for (const decoder of decoders) {
+						if (callback % 64 === 0) decoder.update({ kind: "gap" });
+						decoder.update(read);
+						retained += decoder.span(sample).frames;
+					}
+				}
+				timings.push(((performance.now() - start) * 1e6) / (CALLBACKS * count));
+			}
+			timings.sort((a, b) => a - b);
+			console.log(`${count},${frames},${warmup},${timings[1]?.toFixed(1)}`);
+		}
+	}
+}
+if (retained === 0) throw new Error("warmup discarded the whole stream");

js/watch/src/audio/config.ts

@@ -13,7 +13,7 @@ const MP3_MPEG1_MIN_SAMPLE_RATE = 32000;
 /** The catalog fields that determine the demuxer and WebCodecs decoder instance. */
 export type DecoderConfig = Pick<
 	Catalog.AudioConfig,
-	"codec" | "container" | "description" | "sampleRate" | "numberOfChannels"
+	"codec" | "container" | "description" | "sampleRate" | "numberOfChannels" | "warmup"
 >;
 
 /** The routing and decoder state whose changes require a replacement subscription. */
@@ -31,6 +31,7 @@ export function decoderConfig(config: Catalog.AudioConfig): DecoderConfig {
 		description: config.description,
 		sampleRate: config.sampleRate,
 		numberOfChannels: config.numberOfChannels,
+		warmup: config.warmup,
 	};
 }
 

js/watch/src/audio/decoder.test.ts

@@ -439,13 +439,12 @@ describe("Decoder across a rendition's absence", () => {
 			const before = playback.inserted().length;
 			const timestamps = await playback.play(6);
 
-			// One graph throughout, and the return decodes into the same ring from its first frame past
-			// the legacy warm-up.
+			// One graph throughout; a rendition without warmup keeps its entire return.
 			expect(contexts).toHaveLength(1);
 			expect(context.calls.close).toBe(0);
 			expect(playback.subscriptions()).toBe(2);
 			expect(playback.rings()).toEqual([ring]);
-			expect(playback.inserted().slice(before)).toEqual(timestamps.slice(3));
+			expect(playback.inserted().slice(before)).toEqual(timestamps);
 		} finally {
 			playback.close();
 		}
@@ -563,8 +562,8 @@ describe("Decoder across a rendition's absence", () => {
 			load();
 			await microtasks();
 
-			// Everything past the legacy warm-up reaches the ring, none of it dropped for want of one.
-			expect(playback.inserted()).toEqual(timestamps.slice(3));
+			// Every frame reaches the ring once the worklet loads.
+			expect(playback.inserted()).toEqual(timestamps);
 		} finally {
 			playback.close();
 		}

js/watch/src/audio/decoder.ts

@@ -26,10 +26,7 @@ import { AUTO_MAX_DELAY, ringSamples, target } from "./latency";
 // A blob: URL, or a hosted file when assets() is set; see vite-plugin-worklet.
 import RenderWorklet from "./render-worklet.ts?worklet";
 import type { Source } from "./source";
-import { type DecodedSpan, Terminal } from "./terminal";
-import { Warmup } from "./warmup";
-
-const LEGACY_WARMUP_CALLBACKS = 3;
+import { Terminal } from "./terminal";
 
 export type DecoderInput = {
 	// Whether to download the audio track. Defaults to true.
@@ -128,6 +125,7 @@ export class Decoder {
 
 	// The subscription's max delay: the shared budget, raised to the estimator's ceiling in "auto".
 	#subscribeMaxDelay = new Signal<Time.Milli>(Time.Milli.zero);
+	#maxDelay = new Signal<Time.Milli>(Time.Milli.zero);
 
 	// The codec's frame duration: the catalog constant, refined by each frame's own duration.
 	#frame = new Signal<Time.Milli | undefined>(undefined);
@@ -189,7 +187,9 @@ export class Decoder {
 	#runSubscribeMaxDelay(effect: Effect): void {
 		const maxDelay = effect.get(this.sync.out.maxDelay);
 		const auto = effect.get(this.sync.in.delay) === "auto";
-		this.#subscribeMaxDelay.set(auto ? Time.Milli.max(maxDelay, AUTO_MAX_DELAY) : maxDelay);
+		const warmup = Time.Milli(effect.get(this.source.out.config)?.warmup ?? 0);
+		this.#maxDelay.set(Time.Milli.add(maxDelay, warmup));
+		this.#subscribeMaxDelay.set(Time.Milli.add(auto ? Time.Milli.max(maxDelay, AUTO_MAX_DELAY) : maxDelay, warmup));
 	}
 
 	#runShape(effect: Effect): void {
@@ -412,14 +412,14 @@ export class Decoder {
 	): void {
 		const preSkip =
 			config.codec === "opus" && config.description ? Util.Opus.preSkip(Util.Hex.toBytes(config.description)) : 0;
-		this.#terminal.clear(preSkip);
+		this.#terminal.clear(preSkip, Time.Micro.fromMilli(Time.Milli(config.warmup ?? 0)));
 		const format =
 			config.container.kind === "loc" ? new Container.Loc.Format("audio") : new Container.Legacy.Format(config);
 		// Create consumer with slightly less latency than the render worklet to avoid underflowing.
 		// TODO include JITTER_UNDERHEAD
 		const consumer = new Container.Consumer(sub, {
 			format,
-			maxDelay: this.sync.out.maxDelay,
+			maxDelay: this.#maxDelay,
 		});
 		effect.cleanup(() => consumer.close());
 
@@ -439,18 +439,8 @@ export class Decoder {
 			const loaded = await Util.Libav.polyfill();
 			if (!loaded) return; // cancelled
 
-			const warmup = new Warmup(LEGACY_WARMUP_CALLBACKS);
-
 			const decoder = new AudioDecoder({
-				output: (data) => {
-					const decoded = this.#terminal.span(data);
-					if (warmup.drop()) {
-						// Drop initial callbacks to prime the decoder.
-						data.close();
-						return;
-					}
-					this.#emit(data, decoded);
-				},
+				output: (data) => this.#emit(data),
 				error: (error) => failure.set(error),
 			});
 			effect.cleanup(() => {
@@ -529,7 +519,7 @@ export class Decoder {
 		const init = Container.Cmaf.decodeInitSegment(initSegment);
 		const opusDescription = config.description ? Util.Hex.toBytes(config.description) : init.description;
 		const preSkip = config.codec === "opus" && opusDescription ? Util.Opus.preSkip(opusDescription) : 0;
-		this.#terminal.clear(preSkip);
+		this.#terminal.clear(preSkip, Time.Micro.fromMilli(Time.Milli(config.warmup ?? 0)));
 		// Opus in CMAF uses raw packets (not OGG-wrapped), so description must be omitted.
 		// The dOps box from the init segment is not a valid OGG Identification Header.
 		const description =
@@ -541,7 +531,7 @@ export class Decoder {
 
 		const consumer = new Container.Consumer(sub, {
 			format: new Container.Cmaf.Format(init),
-			maxDelay: this.sync.out.maxDelay,
+			maxDelay: this.#maxDelay,
 		});
 		effect.cleanup(() => consumer.close());
 
@@ -634,8 +624,8 @@ export class Decoder {
 		}
 	}
 
-	#emit(sample: AudioData, decoded: DecodedSpan = this.#terminal.span(sample)) {
-		const { timestamp, frameOffset, frames } = decoded;
+	#emit(sample: AudioData) {
+		const { timestamp, frameOffset, frames } = this.#terminal.span(sample);
 		const timestampMilli = Time.Milli.fromMicro(timestamp);
 		if (frames === 0) {
 			sample.close();

js/watch/src/audio/terminal.test.ts

@@ -10,6 +10,27 @@ const frame = (group: number, timestamp: number) => ({
 const end = (group: number, timestamp: number) => ({ kind: "end" as const, group, end: timestamp as Time.Micro });
 const gap = { kind: "gap" as const };
 
+test("warmup trims by timestamp across callback sizes, not ordinary groups or declared returns", () => {
+	const terminal = new Terminal();
+	terminal.clear(0, 80_000 as Time.Micro);
+	terminal.update(frame(10, 1_000_000));
+	expect(terminal.span({ timestamp: 1_000_000, sampleRate: 48_000, numberOfFrames: 2_880 }).frames).toBe(0);
+	terminal.update(frame(11, 1_060_000));
+	expect(terminal.span({ timestamp: 1_060_000, sampleRate: 48_000, numberOfFrames: 2_880 })).toEqual({
+		timestamp: 1_080_000 as Time.Micro,
+		frameOffset: 960,
+		frames: 1_920,
+	});
+	terminal.update(frame(12, 1_120_000));
+	expect(terminal.span({ timestamp: 1_120_000, sampleRate: 48_000, numberOfFrames: 960 }).frames).toBe(960);
+	terminal.update(end(13, 1_140_000));
+	terminal.update(frame(14, 2_000_000));
+	expect(terminal.span({ timestamp: 2_000_000, sampleRate: 48_000, numberOfFrames: 960 }).frames).toBe(960);
+	terminal.update(gap);
+	terminal.update(frame(16, 3_000_000));
+	expect(terminal.span({ timestamp: 3_000_000, sampleRate: 48_000, numberOfFrames: 960 }).frames).toBe(0);
+});
+
 test("Terminal trims a partial packet to the source endpoint", () => {
 	const terminal = new Terminal();
 	terminal.update(end(0, 17_916));

js/watch/src/audio/terminal.ts

@@ -32,24 +32,33 @@ export class Terminal {
 	#epoch?: Time.Micro;
 	#preSkip = 0;
 	#preSkipRemaining?: number;
+	#warmup = 0 as Time.Micro;
+	#prime = true;
+	#before?: Time.Micro;
 
 	/** The exclusive source endpoint most recently received. */
 	get end(): Time.Micro | undefined {
 		return this.#end;
 	}
 
-	/** Reset state for a new container consumer with Opus pre-skip in 48 kHz frames. */
-	clear(preSkip = 0): void {
+	/** Reset with Opus pre-skip in 48 kHz frames and warmup in microseconds. */
+	clear(preSkip = 0, warmup = 0 as Time.Micro): void {
 		this.#started = false;
 		this.#end = undefined;
 		this.#endGroup = undefined;
 		this.#preSkip = preSkip;
+		this.#warmup = warmup;
+		this.#prime = true;
+		this.#before = undefined;
 		this.#resetEpoch();
 	}
 
 	/** Apply one ordered consumer read and report whether a codec epoch ended before it. */
 	update(read: Update): boolean {
-		if (read.kind === "gap") return this.#restart();
+		if (read.kind === "gap") {
+			this.#prime = true;
+			return this.#restart();
+		}
 		if (read.kind === "end") {
 			this.#end = read.end;
 			this.#endGroup = read.group;
@@ -60,6 +69,10 @@ export class Terminal {
 		const reset = this.#end !== undefined && read.group !== this.#endGroup && this.#restart();
 		this.#started = true;
 		this.#epoch ??= read.frame.timestamp;
+		if (this.#prime) {
+			this.#before = (read.frame.timestamp + this.#warmup) as Time.Micro;
+			this.#prime = false;
+		}
 		return reset;
 	}
 
@@ -69,6 +82,7 @@ export class Terminal {
 		this.#started = false;
 		this.#end = undefined;
 		this.#endGroup = undefined;
+		this.#before = undefined;
 		this.#resetEpoch();
 		return started;
 	}
@@ -80,12 +94,18 @@ export class Terminal {
 
 		const delayFrames = Math.floor((this.#preSkip * sample.sampleRate) / 48_000);
 		const remaining = this.#preSkipRemaining ?? delayFrames;
-		const frameOffset = Math.min(remaining, sample.numberOfFrames);
+		let frameOffset = Math.min(remaining, sample.numberOfFrames);
 		this.#preSkipRemaining = remaining - frameOffset;
 
 		const delay = Math.round((delayFrames * 1_000_000) / sample.sampleRate);
-		const timestamp = Math.max(epoch, sample.timestamp - delay) as Time.Micro;
+		let timestamp = Math.max(epoch, sample.timestamp - delay) as Time.Micro;
 		let frames = sample.numberOfFrames - frameOffset;
+		if (this.#before !== undefined && timestamp < this.#before) {
+			const trim = Math.min(frames, Math.ceil(((this.#before - timestamp) * sample.sampleRate) / 1_000_000));
+			frameOffset += trim;
+			frames -= trim;
+			timestamp = (timestamp + Math.round((trim * 1_000_000) / sample.sampleRate)) as Time.Micro;
+		}
 		if (this.#end !== undefined) {
 			if (this.#end <= timestamp) {
 				frames = 0;

js/watch/src/msf.test.ts

@@ -30,6 +30,26 @@ test("copies delay onto the hang rendition", () => {
 	expect(toHang(catalog).audio?.renditions.audio?.delay).toBe(u53(80));
 });
 
+test.each(["loc", "legacy"])("derives 80 ms Opus decoder warmup from %s MSF audio", (packaging) => {
+	for (const samplerate of [8000, 48000]) {
+		const catalog: Msf.Catalog = {
+			tracks: [{ name: "audio", packaging, role: "audio", codec: "opus", samplerate, delay: 3, jitter: 4 }],
+		};
+		const audio = toHang(catalog).audio?.renditions.audio;
+		expect(audio?.warmup).toBe(u53(80));
+		expect(audio?.sampleRate).toBe(u53(samplerate));
+		expect(audio?.delay).toBe(u53(3));
+		expect(audio?.jitter).toBe(u53(4));
+	}
+});
+
+test.each(["mp4a.40.2", "flac", "future-audio"])("keeps MSF decoder warmup absent for %s", (codec) => {
+	const audio = toHang({ tracks: [{ name: "audio", packaging: "loc", role: "audio", codec }] }).audio?.renditions
+		.audio;
+	expect(audio?.codec).toBe(codec);
+	expect(audio?.warmup).toBeUndefined();
+});
+
 test("rounds a fractional MSF delay up, and drops zero", () => {
 	const catalog: Msf.Catalog = {
 		tracks: [