feat(watch): size the auto delay from the measured arrival estimator

demo/web/src/index.ts

@@ -119,9 +119,6 @@ function createTile(name: string): WatchTile {
 	const watch = document.createElement("moq-watch") as MoqWatch;
 	watch.name = name;
 	watch.muted = true; // unmuted only while active (see below)
-	// Default to a fixed 100ms jitter buffer (instead of adaptive "auto") so
-	// the delay visualization has something to show. Drag it in the panel.
-	watch.setAttribute("delay", "100ms");
 	const canvas = document.createElement("canvas");
 	canvas.style.cssText = "width: 100%; height: auto;";
 	watch.appendChild(canvas);

demo/web/src/viz.ts

@@ -307,7 +307,7 @@ export function bufferBars(parent: Signals.Effect, watch: MoqWatch): HTMLElement
 		const timestamp = watch.sync.now() as number | undefined;
 		const stalled = watch.video.out.stalled.peek();
 		drawRanges(video.canvas, watch.video.out.buffered.peek(), timestamp, stalled);
-		drawRanges(audio.canvas, watch.audio.out.buffered.peek(), timestamp, false);
+		drawRanges(audio.canvas, watch.audio.out.buffered.peek(), timestamp, watch.audio.out.stalled.peek());
 		parent.animate(draw);
 	};
 	parent.animate(draw);

doc/concept/audio-jitter.md

@@ -228,10 +228,8 @@ carry a number the payload already states.
 
 ### What is not in the frame duration
 
-The browser's AudioWorklet renders in fixed 128-sample blocks, and
-`js/watch/src/audio/config.ts` currently adds that block to the per-codec frame
-duration, so 48 kHz Opus reads as 23 ms rather than 20 ms. **That block is not
-part of the target.** It is a property of the render backend, not of the
+The browser's AudioWorklet renders in fixed 128-sample blocks. **That block is
+not part of the target.** It is a property of the render backend, not of the
 network, and native has no worklet at all. If one language's step carried it and
 the other's did not, the two would produce different target series from the same
 trace and the corpus could not hold.
@@ -278,9 +276,6 @@ point strictly contains the publisher's. The floor earns its place by being
 correct *immediately*, before the estimator has seen a full flush cycle, which
 is the one thing a measurement cannot be.
 
-Note that `js/watch` today adds the two rather than taking the maximum, which
-over-buffers a bursty publisher by its own flush span.
-
 ## Rise, fall, startup, and the ceiling
 
 There is **no separate rise or fall limiter**. The histogram is the only

doc/concept/audio-jitter/corpus.test.ts

@@ -2,7 +2,7 @@
 // has to catch is the corpus drifting from the algorithm it claims to encode: a checked-in
 // file edited by hand to make a failing implementation pass would otherwise go unnoticed.
 //
-// Grading an implementation is each language's own test suite, since neither exists yet.
+// Grading an implementation is each language's own test suite, e.g. js/hang's jitter.test.ts.
 
 import { describe, expect, test } from "bun:test";
 import { readFileSync } from "node:fs";

doc/lib/js/watch.md

@@ -30,7 +30,7 @@ in sync at the latency you ask for.
 | --- | --- |
 | `url`, `name` | Relay URL (with `?jwt=` if needed) and broadcast name. |
 | `paused`, `muted`, `volume` | The usual player controls, mirrored as reactive properties. |
-| `delay` | How far playback trails the live edge: `"auto"` (derived from RTT, the default), a duration like `"300ms"`, or `"instant"` to paint frames as they decode with no pacing at all. |
+| `delay` | How far playback trails the live edge: `"auto"` (the default, sized from how late frames actually arrive; see [audio jitter](/concept/audio-jitter)), a duration like `"300ms"`, or `"instant"` to paint frames as they decode with no pacing at all. |
 | `buffer` | Future-dated media held beyond the live edge before playback skips ahead, e.g. `"30s"`. Defaults to none. |
 | `captions` | The caption track to show, or absent for off. `el.text.out.available` lists the renditions for a picker. |
 | `visible` | Only subscribe to video while the element is on screen: a margin (`"20%"` default, `"200px"`), `"always"`, or `"never"`. |

js/hang/src/container/consumer.test.ts

@@ -515,6 +515,25 @@ test("Consumer skips groups via PTS-span when over the max age", async () => {
 	consumer.close();
 });
 
+test("Consumer measures arrivals as they land", async () => {
+	const track = new Track.Producer("test");
+	const consumer = new Consumer(track.subscribe(), { format: new LegacyFormat("audio"), maxAge: 5000 as Time.Milli });
+
+	// Nothing observed yet: the cold-start estimate.
+	expect(consumer.spread.peek()).toBe(100 as Time.Milli);
+
+	// Groups arriving exactly as fast as their media, across two resample intervals. Nothing is
+	// late, so the evidence pulls the estimate down from the cold start.
+	for (let i = 0; i < 3; i++) {
+		writeGroupWithLegacyFrames(track, i, [(i * 520_000) as Time.Micro]);
+		await settle(520);
+	}
+	expect(consumer.spread.peek()).toBeLessThan(100 as Time.Milli);
+
+	track.close();
+	consumer.close();
+});
+
 // --- Ordering ---
 
 test("Consumer delivers groups in sequence order regardless of arrival order", async () => {

js/hang/src/container/consumer.ts

@@ -3,6 +3,7 @@ import * as Moq from "@moq/net";
 import { Effect, type Getter, type GetterInit, getter, Once, Signal } from "@moq/signals";
 
 import type { Format } from "./format";
+import { Jitter } from "./jitter";
 import type { BufferedRanges, Frame } from "./types";
 
 /** Options for constructing a {@link Consumer}. */
@@ -100,6 +101,16 @@ export class Consumer {
 	/** The time ranges currently buffered and ready to play. */
 	readonly buffered: Getter<BufferedRanges> = this.#buffered;
 
+	#jitter = new Jitter();
+	#spread = new Signal<Time.Milli>(this.#jitter.measured);
+	/**
+	 * How much buffer late arrivals need, measured as frames come off the transport.
+	 *
+	 * The measured term of doc/concept/audio-jitter.md, excluding the codec frame and any advertised
+	 * floor. Starts at the cold-start estimate before any frame arrives.
+	 */
+	readonly spread: Getter<Time.Milli> = this.#spread;
+
 	#signals = new Effect();
 	#closed = new Once<Error | null>();
 
@@ -172,11 +183,19 @@ export class Consumer {
 				const next = await group.consumer.readFrame();
 				if (!next) break;
 				group.empty = false;
+				const arrival = Moq.Time.Milli.now();
 
 				const decoded = this.#format.decode(next.payload);
 
 				for (const sample of decoded) {
 					const marker = this.#format.end?.(sample) !== undefined;
+
+					// Observed here, before any gap handling or age budget: a target derived from
+					// what survives the budget would only ever confirm the budget it was cut to.
+					if (!marker) {
+						this.#jitter.observe(arrival, Moq.Time.Milli.fromMicro(sample.timestamp));
+						this.#spread.set(this.#jitter.measured);
+					}
 					const frame: Frame = {
 						payload: sample.payload,
 						timestamp: sample.timestamp,

js/hang/src/container/jitter.test.ts

@@ -0,0 +1,95 @@
+import { describe, expect, it } from "bun:test";
+import { readdirSync, readFileSync } from "node:fs";
+import { dirname, join } from "node:path";
+import { fileURLToPath } from "node:url";
+import type { Time } from "@moq/net";
+import { Jitter } from "./jitter";
+
+// The conformance corpus beside doc/concept/audio-jitter.md, read directly so this is graded
+// against the document rather than against another implementation.
+const CORPUS = join(dirname(fileURLToPath(import.meta.url)), "../../../../doc/concept/audio-jitter");
+
+interface Case {
+	name: string;
+	frame: number;
+	advertised?: number;
+	delay?: number;
+	arrival: number[];
+	media: number[];
+	target: [number, number][];
+}
+
+const cases: Case[] = readdirSync(CORPUS)
+	.filter((file) => file.endsWith(".json"))
+	.map((file) => JSON.parse(readFileSync(join(CORPUS, file), "utf8")));
+
+// The composition the document specifies on top of the measured term. `@moq/watch` owns it in
+// production (`audio/latency.ts`); it is repeated here so the whole target series is checked.
+function target(jitter: Jitter, c: Case): number {
+	if (c.delay !== undefined) return c.delay;
+	return Math.max(jitter.measured, c.advertised ?? 0) + c.frame;
+}
+
+describe("audio jitter conformance", () => {
+	it("reads the whole corpus", () => {
+		expect(cases.map((c) => c.name).sort()).toEqual([
+			"advertised",
+			"buildup",
+			"fixed",
+			"flush",
+			"idle",
+			"paced",
+			"reorder",
+			"spike",
+			"tunein",
+		]);
+	});
+
+	for (const c of cases) {
+		it(c.name, () => {
+			const jitter = new Jitter();
+			const expected = new Map(c.target);
+			let current: number | undefined;
+			for (let i = 0; i < c.arrival.length; i++) {
+				jitter.observe(c.arrival[i] as Time.Milli, c.media[i] as Time.Milli);
+				const value = target(jitter, c);
+				const listed = expected.get(i);
+				if (listed !== undefined) {
+					expect(value).toBeCloseTo(listed, 6);
+				} else if (current !== undefined) {
+					// Unlisted frames hold the previous target.
+					expect(value).toBeCloseTo(current, 6);
+				}
+				current = value;
+			}
+		});
+	}
+});
+
+describe("Jitter", () => {
+	it("starts at the prior's 95th percentile", () => {
+		expect(new Jitter().measured).toBe(100 as Time.Milli);
+	});
+
+	it("is not moved by a stale frame ahead of the live edge", () => {
+		// The #3517 regression: one frame from a stale group, then the live edge 14.5s of media later,
+		// then evenly paced audio with no real jitter. The jump must not read as a delay or a frame.
+		const jitter = new Jitter();
+		jitter.observe(0 as Time.Milli, 0 as Time.Milli);
+		for (let i = 0; i < 500; i++) {
+			jitter.observe((10 + i * 20) as Time.Milli, (14_500 + i * 20) as Time.Milli);
+		}
+		expect(jitter.measured).toBe(20 as Time.Milli);
+	});
+
+	it("does not follow the absolute delay", () => {
+		// Only the spread of arrivals matters, not how far away the sender is.
+		const near = new Jitter();
+		const far = new Jitter();
+		for (let i = 0; i < 500; i++) {
+			near.observe((10 + i * 20) as Time.Milli, (i * 20) as Time.Milli);
+			far.observe((2000 + i * 20) as Time.Milli, (i * 20) as Time.Milli);
+		}
+		expect(far.measured).toBe(near.measured);
+	});
+});

js/hang/src/container/jitter.ts

@@ -0,0 +1,125 @@
+import type { Time } from "@moq/net";
+
+// Implements the estimator in doc/concept/audio-jitter.md, which is normative. The design and
+// every constant come from WebRTC NetEq's underrun optimizer; this is written from the document,
+// not from NetEq's source.
+
+/** Milliseconds of media time the reference search looks back over. */
+const WINDOW = 2000;
+
+/** Milliseconds of delay per histogram bucket. */
+const BUCKET = 20;
+
+/** Histogram buckets, so the measured term saturates at `BUCKET * BUCKETS`. */
+const BUCKETS = 100;
+
+/** The quantile of the delay distribution the estimate is read at. */
+const QUANTILE = 0.95;
+
+/** Weight retained per folded observation, a ~20s half-life at one fold per `RESAMPLE`. */
+const FORGET = 0.983;
+
+/** Milliseconds of arrival time per resampled observation. */
+const RESAMPLE = 500;
+
+/** Holds the forget factor below `FORGET` early on, so evidence overwrites the prior. */
+const RAMP = 2;
+
+/** Idle intervals folded at once, purely to bound the work. */
+const IDLE_MAX = 1024;
+
+/** The cold-start distribution, which is also what an idle interval relaxes back to. */
+const PRIOR = Float64Array.from({ length: BUCKETS }, (_, i) => 0.5 ** (i + 1));
+
+/**
+ * How much buffer late arrivals need, measured from arrival timing alone.
+ *
+ * Each frame's delay is measured against the fastest frame in the last `WINDOW` of media time, the
+ * largest delay per `RESAMPLE` interval lands in a decaying histogram, and the estimate is its 95th
+ * percentile. Excludes the codec frame and any advertised floor, which the caller composes on top.
+ */
+export class Jitter {
+	/** Arrivals still inside the reference window, in media order. */
+	#history: { arrival: number; media: number }[] = [];
+	/** The largest media timestamp observed, so a reordered frame is recognized. */
+	#newest = Number.NEGATIVE_INFINITY;
+
+	/** The arrival resample intervals are counted from, unset until the first observation. */
+	#origin?: number;
+	/** The interval currently accumulating, counted from `#origin`. */
+	#interval = 0;
+	/** The largest delay seen in that interval. */
+	#peak = 0;
+
+	#histogram = Float64Array.from(PRIOR);
+	/** Observations folded in, which drives the cold-start ramp. */
+	#count = 0;
+
+	/** Fold one frame in: `arrival` on the receiver's monotonic clock, `media` its timestamp. */
+	observe(arrival: Time.Milli, media: Time.Milli): void {
+		// A reordered frame is excluded rather than measured: the media-time gap back to the frame
+		// that overtook it would otherwise be added to the delay.
+		if (media <= this.#newest) return;
+		this.#newest = media;
+
+		// Pruning by media time is what makes a timestamp jump self-correcting: everything before it
+		// leaves the window at once and the new frame becomes its own reference.
+		let stale = 0;
+		while (stale < this.#history.length && this.#history[stale].media < media - WINDOW) stale++;
+		if (stale > 0) this.#history.splice(0, stale);
+		this.#history.push({ arrival, media });
+
+		// Delay against the fastest frame in the window, not the previous one, so a queue that
+		// fills a little per frame reads as the build-up it is.
+		let reference = this.#history[0];
+		for (const entry of this.#history) {
+			if (entry.arrival - entry.media < reference.arrival - reference.media) reference = entry;
+		}
+		const delay = Math.max(0, arrival - reference.arrival - (media - reference.media));
+
+		if (this.#origin === undefined) {
+			this.#origin = arrival;
+			this.#peak = delay;
+			return;
+		}
+
+		const interval = Math.floor((arrival - this.#origin) / RESAMPLE);
+		if (interval <= this.#interval) {
+			this.#peak = Math.max(this.#peak, delay);
+			return;
+		}
+
+		this.#fold(Math.min(BUCKETS - 1, Math.floor(this.#peak / BUCKET)));
+
+		// An interval with no arrival relaxes the estimate toward the prior at the same rate.
+		const idle = Math.min(interval - this.#interval - 1, IDLE_MAX);
+		for (let i = 0; i < idle; i++) this.#fold(undefined);
+
+		this.#interval = interval;
+		this.#peak = delay;
+	}
+
+	/** The measured term: the upper edge of the bucket holding the 95th percentile. */
+	get measured(): Time.Milli {
+		let cumulative = 0;
+		for (let i = 0; i < BUCKETS; i++) {
+			cumulative += this.#histogram[i];
+			if (cumulative > QUANTILE) return ((i + 1) * BUCKET) as Time.Milli;
+		}
+		return (BUCKETS * BUCKET) as Time.Milli;
+	}
+
+	// `h = f*h + (1-f)*x`, where x is all weight on the observed bucket, or the prior when the
+	// interval held no arrival. The total stays at 1 without renormalizing.
+	#fold(bucket: number | undefined): void {
+		this.#count += 1;
+		const forget = Math.min(FORGET, this.#count / (this.#count + RAMP));
+		const fresh = 1 - forget;
+
+		for (let i = 0; i < BUCKETS; i++) {
+			this.#histogram[i] *= forget;
+			if (bucket === undefined) this.#histogram[i] += fresh * PRIOR[i];
+		}
+		if (bucket !== undefined) this.#histogram[bucket] += fresh;
+	}
+}

js/hang/src/util/opus.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from "bun:test";
-import { pickRate, preSkip, SAMPLE_RATES, supportsRate, toDOps } from "./opus";
+import { packetSamples, pickRate, preSkip, SAMPLE_RATES, supportsRate, toDOps } from "./opus";
 
 describe("pickRate", () => {
 	// Matches the pick_opus_rate tests in rs/moq-audio/src/codec.rs.
@@ -80,3 +80,24 @@ describe("preSkip", () => {
 		expect(preSkip(dops)).toBe(312);
 	});
 });
+
+describe("packetSamples", () => {
+	// Matches packet_samples_reads_toc in rs/moq-mux/src/codec/opus/mod.rs.
+	it("reads the TOC byte", () => {
+		// config 16 (CELT NB 2.5 ms), code 0 (1 frame).
+		expect(packetSamples(new Uint8Array([16 << 3]))).toBe(120);
+		// config 3 (SILK NB 60 ms), code 0.
+		expect(packetSamples(new Uint8Array([3 << 3]))).toBe(2880);
+		// config 1 (SILK NB 20 ms), code 1 (2 frames).
+		expect(packetSamples(new Uint8Array([(1 << 3) | 1]))).toBe(1920);
+		// config 1, code 3 with 4 frames.
+		expect(packetSamples(new Uint8Array([(1 << 3) | 3, 4]))).toBe(3840);
+		// config 31 (CELT FB 20 ms), the libopus real-time default.
+		expect(packetSamples(new Uint8Array([31 << 3]))).toBe(960);
+	});
+
+	it("is undefined for a truncated packet", () => {
+		expect(packetSamples(new Uint8Array([]))).toBeUndefined();
+		expect(packetSamples(new Uint8Array([(1 << 3) | 3]))).toBeUndefined();
+	});
+});