diff --git a/doc/concept/audio-jitter.md b/doc/concept/audio-jitter.md index f1ced5e263..c098ec764a 100644 --- a/doc/concept/audio-jitter.md +++ b/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. diff --git a/js/watch/src/audio/decoder.ts b/js/watch/src/audio/decoder.ts index aecca17b80..5b9f5c9a40 100644 --- a/js/watch/src/audio/decoder.ts +++ b/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(undefined); + // The subscription's max age: the shared budget, raised to the estimator's ceiling in "auto". + #subscribeMaxAge = new Signal(Time.Milli.zero); + // The codec's frame duration: the catalog constant, refined by each frame's own duration. #frame = new Signal(undefined); @@ -165,6 +168,7 @@ 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)); @@ -172,6 +176,15 @@ export class Decoder { 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; diff --git a/js/watch/src/audio/latency.ts b/js/watch/src/audio/latency.ts index b9dd78becd..4637da8bf0 100644 --- a/js/watch/src/audio/latency.ts +++ b/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; diff --git a/js/watch/src/audio/replay.test.ts b/js/watch/src/audio/replay.test.ts index 6cc04d7770..4722bcbb5a 100644 --- a/js/watch/src/audio/replay.test.ts +++ b/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 { + 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 { + 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); + }); +}); diff --git a/quest/m0/audio-jitter-target/watch.md b/quest/m0/audio-jitter-target/watch.md index b6d9989c5d..4d1151c954 100644 --- a/quest/m0/audio-jitter-target/watch.md +++ b/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