fix(watch): subscribe auto audio at the estimator's ceiling, replay through the estimator

doc/concept/audio-jitter.md

@@ -47,7 +47,17 @@ at least as long as the estimate's ceiling, which is why `moq play` waits 2 s on
 a stalled group under `--delay auto`. The observation is also only as honest as
 the caller is prompt: `read()` is timed when it resolves, so a player drains it
 as packets come and lets the playback buffer hold the target, rather than
-pacing reads to the speaker. A frame dropped by the relay is never observed, which is correct.
+pacing reads to the speaker.
+
+A frame dropped by the relay is never observed, which is correct only while the
+drop is not the receiver's own doing. The subscription's max age drops groups
+before any container sees them, at the relay and in the receiver's own transport
+subscriber, so a subscription cut to the target hides every frame later than
+the target: the estimate creeps up one bucket at a time while the frames it
+should have measured go unplayed. In automatic mode both implementations
+subscribe with at least the ceiling, 2 s: `moq play --delay auto` through its
+age budget, and the browser through the audio subscription alone, since its
+container consumer observes each frame before applying the local budget.
 A frame the container will later discard *is* observed, which is also correct:
 it arrived, and when it arrived is the measurement.
 

js/watch/src/audio/decoder.ts

@@ -27,7 +27,7 @@ import {
 	playbackIdentity,
 } from "./config";
 import { Handover } from "./handover";
-import { reanchor, ringSamples, target } from "./latency";
+import { AUTO_MAX_AGE, reanchor, ringSamples, target } from "./latency";
 // Compiled and inlined as a blob URL via vite-plugin-worklet.
 import RenderWorklet from "./render-worklet.ts?worklet";
 import type { Source } from "./source";
@@ -123,6 +123,9 @@ export class Decoder {
 	// The container consumer's arrival estimate, unset while nothing is subscribed.
 	#measured = new Signal<Time.Milli | undefined>(undefined);
 
+	// The subscription's max age: the shared budget, raised to the estimator's ceiling in "auto".
+	#subscribeMaxAge = 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);
 
@@ -165,13 +168,23 @@ export class Decoder {
 			return config ? decoderConfig(config) : undefined;
 		});
 
+		this.#signals.run(this.#runSubscribeMaxAge.bind(this));
 		this.#signals.run(this.#runWorklet.bind(this));
 		this.#signals.run(this.#runEnabled.bind(this));
 		this.#signals.run(this.#runLatency.bind(this));
 		this.#signals.run(this.#runLatencyReanchor.bind(this));
 		this.#signals.run(this.#runDecoder.bind(this));
 	}
 
+	// A group the relay expires is never observed, so a subscription cut to the target would cap the
+	// estimate at the target it already holds. The container consumer keeps the shared budget as its
+	// local skip, since it observes each frame before applying it. See doc/concept/audio-jitter.md.
+	#runSubscribeMaxAge(effect: Effect): void {
+		const maxAge = effect.get(this.sync.out.maxAge);
+		const auto = effect.get(this.sync.in.delay) === "auto";
+		this.#subscribeMaxAge.set(auto ? Time.Milli.max(maxAge, AUTO_MAX_AGE) : maxAge);
+	}
+
 	#runWorklet(effect: Effect): void {
 		// It takes a second or so to initialize the AudioContext/AudioWorklet, so do it even if disabled.
 		// This is less efficient for video-only playback but makes muting/unmuting instant.
@@ -349,7 +362,7 @@ export class Decoder {
 			broadcast: active,
 			track,
 			priority: Catalog.PRIORITY.audio,
-			maxAge: this.sync.out.maxAge,
+			maxAge: this.#subscribeMaxAge,
 		});
 		if (!sub) return;
 

js/watch/src/audio/latency.ts

@@ -21,6 +21,13 @@ export function target(props: Target): Time.Milli {
 	return Time.Milli.add(floored, props.frame ?? Time.Milli.zero);
 }
 
+/**
+ * The least subscription max age an "auto" track asks for: the estimator's ceiling.
+ *
+ * The measured term saturates at 100 buckets of 20 ms, so a frame later than this adds nothing.
+ */
+export const AUTO_MAX_AGE = 2000 as Time.Milli;
+
 /** Whether a deeper target re-stalls the ring, and the baseline the next target is compared against. */
 export interface Reanchor {
 	stall: boolean;

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

@@ -1,12 +1,15 @@
-import { describe, expect, it } from "bun:test";
-import { Time } from "@moq/net";
+import { describe, expect, it, spyOn } from "bun:test";
+import * as Container from "@moq/hang/container";
+import { Group, Time, Track, Varint } from "@moq/net";
+import { AUTO_MAX_AGE, target } from "./latency";
 import { AudioRingBuffer } from "./ring-buffer";
 import { allocSharedRingBuffer, SharedRingBuffer } from "./shared-ring-buffer";
 
 // Replay recorded-shape arrival traces through both rings and count what a listener would hear:
 // quanta rendered short (an underrun) and samples the ring threw away (a skip). Both rings are
 // driven through the same harness, since the postMessage fallback is the path every page without
-// cross-origin isolation takes.
+// cross-origin isolation takes. Each ring is sized from the target the estimator settles on, measured
+// through the transport subscription and container consumer the decoder reads.
 
 const RATE = 48000;
 const QUANTUM = 128; // an AudioWorklet render quantum
@@ -51,17 +54,78 @@ function trace(frames: number, burst: number, spread: number, seed = 7): Arrival
 	return out;
 }
 
+/** A legacy container frame: the media timestamp, then a payload the consumer never decodes. */
+function encode(media: Time.Micro): Uint8Array {
+	const timestamp = Varint.encode(media);
+	const frame = new Uint8Array(timestamp.byteLength + 1);
+	frame.set(timestamp, 0);
+	return frame;
+}
+
+interface Measured {
+	/** The playout target after each distinct arrival instant. */
+	targets: Time.Milli[];
+	/** Frames that reached the consumer's reader rather than being dropped for age. */
+	delivered: number;
+}
+
 /**
- * Roughly what the estimator converges on for a steady trace: the p95 arrival delay plus one frame.
- * The estimator itself is graded against the conformance corpus in @moq/hang; this only sizes the
- * rings under test.
+ * Feed `t` through the path the decoder measures on: a transport subscription into a container
+ * consumer, on a stubbed monotonic clock, one group per frame. After each arrival `budget` turns the
+ * current target into the subscription's max age, the way the decoder does.
  */
-function target(t: Arrival[]): number {
-	let min = Number.POSITIVE_INFINITY;
-	for (const { media, arrival } of t) min = Math.min(min, arrival - media);
-	const spreads = t.map(({ media, arrival }) => arrival - media - min).sort((a, b) => a - b);
-	const p95 = Math.ceil(spreads[Math.floor(spreads.length * 0.95)]);
-	return p95 + CHUNK_MS;
+async function measure(t: Arrival[], budget: (target: Time.Milli) => Time.Milli): Promise<Measured> {
+	let clock = 0;
+	const now = spyOn(performance, "now").mockImplementation(() => clock);
+	const track = new Track.Producer("audio");
+	const subscriber = track.subscribe({ maxAge: budget(Time.Milli.zero) });
+	const consumer = new Container.Consumer(subscriber, {
+		format: new Container.Legacy.Format("audio"),
+		maxAge: AUTO_MAX_AGE,
+	});
+
+	const result: Measured = { targets: [], delivered: 0 };
+	const reader = (async () => {
+		for (;;) {
+			const next = await consumer.next();
+			if (!next) return;
+			if (next.frame) result.delivered++;
+		}
+	})();
+
+	try {
+		let next = 0;
+		while (next < t.length) {
+			clock = t[next].arrival;
+			for (; next < t.length && t[next].arrival === clock; next++) {
+				const media = Time.Micro.fromMilli(t[next].media as Time.Milli);
+				const group = new Group.Producer(next);
+				group.writeFrame({ payload: encode(media), timestamp: Time.Timestamp.fromMicros(media) });
+				group.close();
+				track.writeGroup(group);
+			}
+
+			// A macrotask boundary, so the group readers have observed everything written above.
+			await new Promise((resolve) => setImmediate(resolve));
+
+			const current = target({ measured: consumer.spread.peek(), frame: CHUNK_MS as Time.Milli });
+			result.targets.push(current);
+			subscriber.update({ maxAge: budget(current) });
+		}
+	} finally {
+		track.close();
+		consumer.close();
+		await reader;
+		now.mockRestore();
+	}
+
+	return result;
+}
+
+/** The target the estimator settles on by the end of `t`, given the subscription "auto" asks for. */
+async function settled(t: Arrival[]): Promise<number> {
+	const { targets } = await measure(t, (target) => Time.Milli.max(target, AUTO_MAX_AGE));
+	return targets[targets.length - 1];
 }
 
 /** The two rings behind one interface, since the harness drives them identically. */
@@ -140,34 +204,25 @@ const RINGS: Array<[string, (latencyMs: number) => Ring]> = [
 ];
 
 describe.each(RINGS)("%s ring replay", (_name, build) => {
-	it("plays an evenly paced sender without underruns or skips", () => {
+	it("plays an evenly paced sender without underruns or skips", async () => {
 		const t = trace(600, 1, 10);
-		expect(replay(build(target(t)), t, 2000)).toEqual({ underruns: 0, skipped: 0 });
+		expect(replay(build(await settled(t)), t, 2000)).toEqual({ underruns: 0, skipped: 0 });
 	});
 
-	it("plays a bursty sender at the measured target", () => {
+	it("plays a bursty sender at the measured target", async () => {
 		// Three frames flushed at once, which is what an importer packing PES payloads produces.
 		const t = trace(600, 3, 5);
-		expect(target(t)).toBeGreaterThan(2 * CHUNK_MS);
-		const result = replay(build(target(t)), t, 2000);
-		expect(result.underruns).toBeLessThanOrEqual(2);
-		expect(result.skipped).toBe(0);
+		const measured = await settled(t);
+		expect(measured).toBeGreaterThan(2 * CHUNK_MS);
+		expect(replay(build(measured), t, 2000)).toEqual({ underruns: 0, skipped: 0 });
 	});
 
-	it("keeps a five frame flush span playing", () => {
+	it("keeps a five frame flush span playing", async () => {
 		// A hundred milliseconds of arrivals landing at once. The target covers the trough and the
 		// slack above it absorbs the peak, so the flush plays instead of being cut down to the
 		// target on arrival.
-		//
-		// A 95th percentile target sits by construction at the edge of what arrives, so the tail
-		// beyond it can still land a couple of times across the ten second window; closing that
-		// without a deeper buffer is what the time-stretch quest is for. Two orders of magnitude
-		// below the round-trip target below, and the postMessage ring bounds its refill at capacity
-		// rather than in read(), so it can also land one flush past the band.
 		const t = trace(600, 5, 5);
-		const result = replay(build(target(t)), t, 2000);
-		expect(result.underruns).toBeLessThanOrEqual(2);
-		expect(result.skipped).toBeLessThanOrEqual(2 * CHUNK);
+		expect(replay(build(await settled(t)), t, 2000)).toEqual({ underruns: 0, skipped: 0 });
 	});
 
 	it("underruns constantly when the target ignores the arrival spread", () => {
@@ -179,3 +234,38 @@ describe.each(RINGS)("%s ring replay", (_name, build) => {
 		expect(result.skipped).toBeGreaterThan(10 * CHUNK);
 	});
 });
+
+describe("subscription budget", () => {
+	// A minute of evenly paced audio, long enough for the startup ramp to hand over to the steady
+	// forget factor and the target to settle at its floor, then six seconds of 100ms flushes.
+	const PACED = 3000;
+	const t: Arrival[] = [];
+	for (let i = 0; i < PACED; i++) t.push({ media: i * CHUNK_MS, arrival: i * CHUNK_MS + 50 });
+	t.push(
+		...trace(300, 6, 0).map(({ media, arrival }) => ({
+			media: media + PACED * CHUNK_MS,
+			arrival: arrival + PACED * CHUNK_MS,
+		})),
+	);
+
+	// The targets across the first twenty flushes, 2.4s of them.
+	function onset(targets: Time.Milli[]): Time.Milli[] {
+		return targets.slice(PACED, PACED + 20);
+	}
+
+	it("reaches a flush that starts after the target settled", async () => {
+		const { targets, delivered } = await measure(t, (target) => Time.Milli.max(target, AUTO_MAX_AGE));
+		expect(targets[PACED - 1]).toBe((2 * CHUNK_MS) as Time.Milli);
+		expect(delivered).toBe(t.length);
+		// Every flush lands whole, so its 100ms reaches the 95th percentile within a few intervals.
+		expect(Math.max(...onset(targets))).toBeGreaterThanOrEqual(120);
+	});
+
+	it("drops the flush it should measure when the budget follows the target", async () => {
+		// The transport skips a group older than the subscription's max age, before the consumer
+		// can observe it, so the estimate only ever sees what the target it already holds allows.
+		const { targets, delivered } = await measure(t, (target) => target);
+		expect(delivered).toBeLessThan(t.length);
+		expect(Math.max(...onset(targets))).toBeLessThan(120);
+	});
+});

quest/m0/audio-jitter-target/watch.md

@@ -13,29 +13,30 @@ it, and fixes only what the measurement shows.
 
 ## Plan
 
+`replay.test.ts` measures each trace through a transport subscription into
+`Container.Consumer` on a stubbed clock and sizes both rings from the target
+that settles, so a recorded trace drops in as an `Arrival[]`. In `"auto"` the
+audio subscription asks for at least the estimator's 2 s ceiling, as native
+does; a subscription cut to the target hid every frame later than it.
+`demo/web` sets no delay and `js/watch` stores none, so a fresh tile runs auto.
+
 - Replay the recorded traces from #3477 rather than synthetic ones of the same
-  shape. They are on the reporter's fork (`fperex/moq`, branch
-  `debug/rt-audio`) with the raw ndjson attached to release
-  `rt-audio-traces-2026-09-06`. Trim a copy into the repository and replay it
-  through both rings in `js/watch/src/audio/replay.test.ts`, which today sizes
-  the rings from a p95 approximation rather than from the estimator.
+  shape. The raw ndjson was attached to release `rt-audio-traces-2026-09-06` on
+  the reporter's fork (`fperex/moq`), which no longer exists; the
+  `debug/rt-audio` branch keeps only per-run summaries and event logs. Ask the
+  reporter to re-attach them, or record fresh ones with the
+  [browser harness](/quest/m1/audio-quality-harness/browser.md), then trim a
+  copy into the repository and replay it in `js/watch/src/audio/replay.test.ts`.
 - Manual run against the public relay on Chrome and Safari, the two rows the
-  issue measured. Watch the stats panel's audio underrun counter and the
-  latency tab's auto readout. Measure the publisher's audio encoder
-  input-to-output lag in the same run using the reporter's instrumented harness;
-  #3518 fixed the known cause, so the 7.35 s lag and the 88 to 275 ms/s drift
-  the issue reported stand unconfirmed. If drift survives, the suspects are
-  `writeFrame` opening a group per audio frame under WebTransport stream credit
-  and the main-thread task queue delivering encoder output. Turn that into its
-  own quest rather than fixing it here.
-- `demo/web` no longer pins each tile to a fixed 100 ms delay, which is why a
-  fresh session came up on the 100 ms chip. Confirm auto is what the demo now
-  exercises.
-- In auto mode the subscription's `maxAge` follows the target, so a group
-  staler than the current target is dropped before `Container.Consumer`
-  observes it, and the estimate cannot rise past what it already allows.
-  Native gives auto a separate 2 s budget (`AUTO_MAX_AGE` in `moq play`).
-  Decide from the run whether the browser needs the same.
+  issue measured, with a real microphone and 40 ms or more of added RTT. Watch
+  the stats panel's audio underrun counter and the latency tab's auto readout,
+  on both the isolated and the postMessage ring paths. Measure the publisher's
+  audio encoder input-to-output lag in the same run using the reporter's
+  instrumented harness; #3518 fixed the known cause, so the 7.35 s lag and the
+  88 to 275 ms/s drift the issue reported stand unconfirmed. If drift survives,
+  the suspects are `writeFrame` opening a group per audio frame under
+  WebTransport stream credit and the main-thread task queue delivering encoder
+  output. Turn that into its own quest rather than fixing it here.
 
 ## Related