diff --git a/demo/web/src/index.ts b/demo/web/src/index.ts index c70b95d171..83b9bb7e0a 100644 --- a/demo/web/src/index.ts +++ b/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); diff --git a/demo/web/src/viz.ts b/demo/web/src/viz.ts index abe1a5a496..827b83eda4 100644 --- a/demo/web/src/viz.ts +++ b/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); diff --git a/doc/concept/audio-jitter.md b/doc/concept/audio-jitter.md index 79cca21684..f1ced5e263 100644 --- a/doc/concept/audio-jitter.md +++ b/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 diff --git a/doc/concept/audio-jitter/corpus.test.ts b/doc/concept/audio-jitter/corpus.test.ts index dd24b2b887..93f7a19e36 100644 --- a/doc/concept/audio-jitter/corpus.test.ts +++ b/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"; diff --git a/doc/lib/js/watch.md b/doc/lib/js/watch.md index 6a5a79fee3..04a1cbdb97 100644 --- a/doc/lib/js/watch.md +++ b/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"`. | diff --git a/js/hang/src/container/consumer.test.ts b/js/hang/src/container/consumer.test.ts index 5f369b3d4f..061fee5b9b 100644 --- a/js/hang/src/container/consumer.test.ts +++ b/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 () => { diff --git a/js/hang/src/container/consumer.ts b/js/hang/src/container/consumer.ts index 6540d02445..e7f24d32a7 100644 --- a/js/hang/src/container/consumer.ts +++ b/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 = this.#buffered; + #jitter = new Jitter(); + #spread = new Signal(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 = this.#spread; + #signals = new Effect(); #closed = new Once(); @@ -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, diff --git a/js/hang/src/container/jitter.test.ts b/js/hang/src/container/jitter.test.ts new file mode 100644 index 0000000000..4eb28c140a --- /dev/null +++ b/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); + }); +}); diff --git a/js/hang/src/container/jitter.ts b/js/hang/src/container/jitter.ts new file mode 100644 index 0000000000..6cbb9d0a70 --- /dev/null +++ b/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; + } +} diff --git a/js/hang/src/util/opus.test.ts b/js/hang/src/util/opus.test.ts index d86001567b..a72283f7de 100644 --- a/js/hang/src/util/opus.test.ts +++ b/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(); + }); +}); diff --git a/js/hang/src/util/opus.ts b/js/hang/src/util/opus.ts index 76461e4d49..b5bfad282b 100644 --- a/js/hang/src/util/opus.ts +++ b/js/hang/src/util/opus.ts @@ -83,3 +83,38 @@ export function toDOps(description: Uint8Array): Uint8Array { output[10] = input.getUint8(10); return output; } + +/** + * Number of 48 kHz samples in an Opus packet, read from its TOC byte (RFC 6716 §3.1). + * + * Mirrors `packet_samples` in rs/moq-mux. Opus timing is always reckoned at 48 kHz regardless of the + * encoder's internal bandwidth. Undefined for an empty packet or a code-3 packet missing its + * frame-count byte. + */ +export function packetSamples(packet: Uint8Array): number | undefined { + const toc = packet.at(0); + if (toc === undefined) return undefined; + + let frames: number; + const code = toc & 0b11; + if (code === 0) frames = 1; + else if (code !== 3) frames = 2; + else { + // Code 3: the frame count is the low 6 bits of the following byte. + const count = packet.at(1); + if (count === undefined) return undefined; + frames = count & 0b11_1111; + } + + return configSamples(toc >> 3) * frames; +} + +// 48 kHz samples per frame for a TOC config index (0..=31), per RFC 6716 Table 1. +function configSamples(config: number): number { + // SILK NB/MB/WB: 10, 20, 40, 60 ms. + if (config < 12) return [480, 960, 1920, 2880][config % 4]; + // Hybrid SWB/FB: 10, 20 ms. + if (config < 16) return [480, 960][config % 2]; + // CELT NB/WB/SWB/FB: 2.5, 5, 10, 20 ms. + return [120, 240, 480, 960][config % 4]; +} diff --git a/js/moq-boy/src/game.ts b/js/moq-boy/src/game.ts index c9bec3f4d5..0da8cb3602 100644 --- a/js/moq-boy/src/game.ts +++ b/js/moq-boy/src/game.ts @@ -136,10 +136,7 @@ export class Game { }); this.#signals.cleanup(() => this.audioSource.close()); - this.sync = new Watch.Sync({ - delay: this.delay, - probe: connection.probe, - }); + this.sync = new Watch.Sync({ delay: this.delay }); this.#signals.cleanup(() => this.sync.close()); this.#signals.run(this.#runPixelBudget.bind(this)); diff --git a/js/watch/src/audio/buffer.ts b/js/watch/src/audio/buffer.ts index 4c3a0263e5..504cd5ee14 100644 --- a/js/watch/src/audio/buffer.ts +++ b/js/watch/src/audio/buffer.ts @@ -1,6 +1,6 @@ import { Time } from "@moq/net"; import { Effect, type Getter, Signal } from "@moq/signals"; -import type { Data, InitPost, InitShared, Latency, Reset, State, Truncate } from "./render"; +import type { Data, InitPost, InitShared, Latency, Reset, Stall, State, Truncate } from "./render"; import { allocSharedRingBuffer, SharedRingBuffer } from "./shared-ring-buffer"; /** @@ -21,7 +21,7 @@ class Backpressure { this.#headroom = headroom; } - // Move the gate as the floor changes (e.g. "real-time" jitter tracking RTT). + // Move the gate as the floor changes (e.g. "auto" tracking the measured arrival spread). setHeadroom(headroom: Time.Micro): void { this.#headroom = headroom; } @@ -77,6 +77,14 @@ export interface AudioBuffer { /** Flush buffered samples and re-stall, ready to anchor the next utterance (buffered mode). */ reset(): void; + /** + * Hold playback until the ring holds the target again, keeping everything buffered. + * + * Used when the target deepens: `setLatency` alone only raises the bar a future refill has to + * clear, so a ring already playing keeps draining at its old depth. + */ + stall(): void; + /** * Drop buffered samples at or after `timestamp`, keeping what is already due. * @@ -100,6 +108,9 @@ export interface AudioBuffer { /** Whether the buffer is stalled (waiting to fill). */ readonly stalled: Getter; + /** How many times the ring has run dry mid-playback, cumulative. */ + readonly underruns: Getter; + /** Release any resources (event listeners, intervals, etc.). */ close(): void; } @@ -148,6 +159,9 @@ class SharedAudioBuffer implements AudioBuffer { readonly #stalled = new Signal(true); readonly stalled: Getter = this.#stalled; + readonly #underruns = new Signal(0); + readonly underruns: Getter = this.#underruns; + #backpressure: Backpressure; #signals = new Effect(); @@ -174,6 +188,7 @@ class SharedAudioBuffer implements AudioBuffer { const stalled = this.#ring.stalled; this.#timestamp.set(this.#ring.timestamp); this.#stalled.set(stalled); + this.#underruns.set(this.#ring.underruns); // While stalled the playhead is parked, so release the decode loop to refill the floor; // once playing, hold it to ~the floor ahead. if (stalled) this.#backpressure.flush(); @@ -210,8 +225,16 @@ class SharedAudioBuffer implements AudioBuffer { this.#backpressure.flush(); // the old timeline is gone; let the decode loop re-anchor } + stall(): void { + this.#ring.stall(); + this.#backpressure.flush(); // let the decode loop fill the deeper target + } + wait(timestamp: Time.Micro): Promise { - // Stalled = still filling the floor (bootstrap or underflow): let frames through to refill. + // Stalled = still filling the floor (bootstrap, an underrun the reader re-stalled on, or an + // explicit reset): let frames through so the ring refills to the floor. This is the single + // re-buffer path; the ring un-stalls itself on the insert that reaches the floor, so the + // gate closes again within one frame rather than draining the whole lookahead. if (this.#ring.stalled) return Promise.resolve(); return this.#backpressure.wait(timestamp, this.#ring.timestamp); } @@ -234,6 +257,9 @@ class PostAudioBuffer implements AudioBuffer { readonly #stalled = new Signal(true); readonly stalled: Getter = this.#stalled; + readonly #underruns = new Signal(0); + readonly underruns: Getter = this.#underruns; + // Backpressure runs off the playhead the worklet reports in its state messages. #backpressure: Backpressure; @@ -256,6 +282,7 @@ class PostAudioBuffer implements AudioBuffer { if (data?.type === "state") { this.#timestamp.set(data.timestamp); this.#stalled.set(data.stalled); + this.#underruns.set(data.underruns); // While stalled the playhead is parked, so release the decode loop to refill the floor; // once playing, hold it to ~the floor ahead. if (data.stalled) this.#backpressure.flush(); @@ -295,8 +322,18 @@ class PostAudioBuffer implements AudioBuffer { this.#backpressure.flush(); // the old timeline is gone; let the decode loop re-anchor } + stall(): void { + const msg: Stall = { type: "stall" }; + this.#worklet.port.postMessage(msg); + // Mirror it locally rather than waiting for the worklet's next state message, so `wait()` + // releases the decode loop now. + this.#stalled.set(true); + this.#backpressure.flush(); // let the decode loop fill the deeper target + } + wait(timestamp: Time.Micro): Promise { - // Stalled = still filling the floor (bootstrap or underflow): let frames through to refill. + // Stalled = still filling the floor (bootstrap, an underrun the reader re-stalled on, or an + // explicit reset): let frames through so the ring refills to the floor. See SharedAudioBuffer. if (this.#stalled.peek()) return Promise.resolve(); // Uses the worklet-reported playhead, which lags by a state-message interval; the floor's // headroom covers that. The worklet still drops the oldest if a frame slips through. diff --git a/js/watch/src/audio/config.test.ts b/js/watch/src/audio/config.test.ts index 36ee450983..6922de15d6 100644 --- a/js/watch/src/audio/config.test.ts +++ b/js/watch/src/audio/config.test.ts @@ -2,7 +2,7 @@ import { expect, test } from "bun:test"; import * as Catalog from "@moq/hang/catalog"; import { Time } from "@moq/net"; import { Effect, Signal } from "@moq/signals"; -import { decoderConfig, playbackIdentity, playbackJitter } from "./config"; +import { decoderConfig, frameDuration, packetDuration, playbackIdentity } from "./config"; const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); @@ -60,21 +60,27 @@ test("routing and decoder inputs change the playback identity", () => { expect(playbackIdentity(config({ numberOfChannels: 1 }))).not.toEqual(base); }); -test("an advertised jitter of zero falls back to the codec frame duration", () => { - // 48kHz stereo Opus: 20ms frames plus the 128 sample worklet quantum (3ms). - const floor = playbackJitter(config()); - expect(floor).toBe(Time.Milli(23)); - expect(playbackJitter(config({ jitter: 0 }))).toBe(floor); - expect(playbackJitter(config({ jitter: 60 }))).toBe(Time.Milli(63)); +test("AAC and MP3 frame durations follow their codec frame sizes", () => { + expect(frameDuration(config({ codec: "mp4a.40.2", sampleRate: 48000 }))).toBeCloseTo(21.333, 3); + expect(frameDuration(config({ codec: "mp4a.40.5", sampleRate: 48000 }))).toBeCloseTo(42.667, 3); + expect(frameDuration(config({ codec: "mp3", sampleRate: 48000 }))).toBe(Time.Milli(24)); + expect(frameDuration(config({ codec: "mp3", sampleRate: 24000 }))).toBe(Time.Milli(24)); }); -test("AAC and MP3 jitter follows their codec frame sizes", () => { - expect(playbackJitter(config({ codec: "mp4a.40.2", sampleRate: 48000 }))).toBe(Time.Milli(25)); - expect(playbackJitter(config({ codec: "mp4a.40.2", sampleRate: 24000 }))).toBe(Time.Milli(49)); - expect(playbackJitter(config({ codec: "mp3", sampleRate: 48000 }))).toBe(Time.Milli(27)); - expect(playbackJitter(config({ codec: "mp3", sampleRate: 24000 }))).toBe(Time.Milli(30)); +test("Opus and unknown codecs have no constant frame duration", () => { + // Opus states its duration per packet, in the TOC byte. + expect(frameDuration(config())).toBeUndefined(); + expect(frameDuration(config({ codec: "flac" }))).toBeUndefined(); }); -test("an unknown codec without advertised jitter only reserves the worklet quantum", () => { - expect(playbackJitter(config({ codec: "flac", jitter: 0 }))).toBe(Time.Milli(3)); +test("an implicit CMAF duration falls through to the Opus TOC", () => { + // TOC 0x78: config 15 (hybrid fullband, 20 ms), one frame. + const frame = { + timestamp: Time.Micro(0), + keyframe: true, + payload: new Uint8Array([0x78, 0]), + duration: Time.Micro(0), + }; + expect(packetDuration("opus", frame)).toBe(Time.Milli(20)); + expect(packetDuration("mp4a.40.2", frame)).toBeUndefined(); }); diff --git a/js/watch/src/audio/config.ts b/js/watch/src/audio/config.ts index e5ae6d5fdc..a24fed8ff8 100644 --- a/js/watch/src/audio/config.ts +++ b/js/watch/src/audio/config.ts @@ -1,10 +1,11 @@ import type * as Catalog from "@moq/hang/catalog"; +import type * as Container from "@moq/hang/container"; +import * as Util from "@moq/hang/util"; import { Time } from "@moq/net"; -// AudioWorklet always renders in 128-sample quanta. -const WORKLET_QUANTUM = 128; -const OPUS_FRAME_DURATION_MS = 20; +const OPUS_RATE = 48_000; const AAC_LC_FRAME_SAMPLES = 1024; +const AAC_HE_FRAME_SAMPLES = 2048; const MP3_MPEG1_FRAME_SAMPLES = 1152; const MP3_MPEG2_FRAME_SAMPLES = 576; const MP3_MPEG1_MIN_SAMPLE_RATE = 32000; @@ -40,36 +41,37 @@ export function playbackIdentity(config: Catalog.AudioConfig): PlaybackIdentity }; } -/** The jitter to add to the sync buffer for a rendition, in milliseconds. */ -export function playbackJitter(config: Catalog.AudioConfig): Time.Milli { - // A publisher advertising 0 is claiming frames are never delayed, which no encoder can do, so - // fall back to the codec's frame duration the same way an absent field does. - const codecJitter = (config.jitter || defaultJitter(config)) ?? 0; - - // Add the worklet render quantum so the ring buffer has margin between frame arrivals. - const overhead = Math.ceil((WORKLET_QUANTUM / config.sampleRate) * 1000); - return Time.Milli(codecJitter + overhead); +/** + * A frame's duration as a constant over the catalog's sample rate, for codecs that fix it. + * + * Undefined for Opus, whose packets state their own duration (see `Util.Opus.packetSamples`), and + * for codecs with no constant. Never learned from observed timestamps. + */ +export function frameDuration(config: Pick): Time.Milli | undefined { + const samples = frameSamples(config); + if (samples === undefined) return undefined; + return Time.Milli((samples * 1000) / config.sampleRate); } -// Estimate the minimum jitter (frame duration) based on the audio codec. -// TODO these are defaults; the actual frame duration depends on encoder config. -function defaultJitter(config: Catalog.AudioConfig): number | undefined { - if (config.codec.startsWith("opus")) { - // Opus supports 2.5–60ms but 20ms is the real-time default. - return OPUS_FRAME_DURATION_MS; - } - - if (config.codec.startsWith("mp4a")) { - // 1024 samples for LC-AAC; HE-AAC/AAC-LD use different sizes. - return Math.ceil((AAC_LC_FRAME_SAMPLES / config.sampleRate) * 1000); - } +/** + * A frame's own duration, when it states one: the container's per-sample duration (CMAF), else the + * duration an Opus packet declares in its TOC byte (RFC 6716 §3.1), always reckoned at 48 kHz. + * CMAF reports an implicit duration as zero, which states nothing. + */ +export function packetDuration(codec: string, frame: Container.Frame): Time.Milli | undefined { + if (frame.duration) return Time.Milli.fromMicro(frame.duration); + if (!codec.startsWith("opus")) return undefined; + const samples = Util.Opus.packetSamples(frame.payload); + return samples === undefined ? undefined : Time.Milli((samples * 1000) / OPUS_RATE); +} +function frameSamples(config: Pick): number | undefined { + // HE-AAC (v1 and v2) doubles the output rate of its AAC-LC core. + if (config.codec === "mp4a.40.5" || config.codec === "mp4a.40.29") return AAC_HE_FRAME_SAMPLES; + if (config.codec.startsWith("mp4a")) return AAC_LC_FRAME_SAMPLES; if (config.codec === "mp3") { - // 1152 samples per frame for MPEG-1 Layer III; MPEG-2/2.5 use 576. - const samples = - config.sampleRate >= MP3_MPEG1_MIN_SAMPLE_RATE ? MP3_MPEG1_FRAME_SAMPLES : MP3_MPEG2_FRAME_SAMPLES; - return Math.ceil((samples / config.sampleRate) * 1000); + // MPEG-1 Layer III has 1152 samples per frame; MPEG-2/2.5 have 576. + return config.sampleRate >= MP3_MPEG1_MIN_SAMPLE_RATE ? MP3_MPEG1_FRAME_SAMPLES : MP3_MPEG2_FRAME_SAMPLES; } - return undefined; } diff --git a/js/watch/src/audio/decoder.ts b/js/watch/src/audio/decoder.ts index 54fa26490e..aecca17b80 100644 --- a/js/watch/src/audio/decoder.ts +++ b/js/watch/src/audio/decoder.ts @@ -18,9 +18,16 @@ import { nextMedia, subscribeMedia } from "../media"; import type { Sync } from "../sync"; import { type AudioBuffer, createAudioBuffer } from "./buffer"; -import { type DecoderConfig, decoderConfig, type PlaybackIdentity, playbackIdentity } from "./config"; +import { + type DecoderConfig, + decoderConfig, + frameDuration, + type PlaybackIdentity, + packetDuration, + playbackIdentity, +} from "./config"; import { Handover } from "./handover"; -import { reanchorFloor, ringSamples } from "./latency"; +import { 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"; @@ -63,6 +70,9 @@ type DecoderOutput = { // Whether the audio buffer is stalled (waiting to fill) stalled: Signal; + // How many times the ring ran dry mid-playback, so the UI can show that the target is too low. + underruns: Signal; + // Combined buffered ranges (network jitter + decode buffer) buffered: Signal; }; @@ -90,6 +100,7 @@ export class Decoder { stats: new Signal(undefined), timestamp: new Signal(undefined), stalled: new Signal(true), + underruns: new Signal(0), buffered: new Signal([]), }; readonly out = readonlys(this.#out); @@ -109,9 +120,15 @@ export class Decoder { // Ordered discontinuity and endpoint state from the container consumer. #terminal = new Terminal(); - // The latency floor as of the last settled change, to detect a floor *increase* (needs a deeper - // cushion) versus a decrease or a real-time RTT wiggle. See #runLatencyReanchor. - #prevFloor?: Time.Milli; + // The container consumer's arrival estimate, unset while nothing is subscribed. + #measured = new Signal(undefined); + + // The codec's frame duration: the catalog constant, refined by each frame's own duration. + #frame = new Signal(undefined); + + // The derived target as of the last settled change, to detect a *deepening* (which needs the + // ring to refill) versus a decrease. See #runLatencyReanchor. + #prevTarget?: Time.Milli; // Which subscription the ring's buffered samples came from. See #runDecoder. #handover = new Handover(); @@ -132,7 +149,13 @@ export class Decoder { this.source = props.source; this.sync = props.sync; - this.#signals.cleanup(this.sync.register(this.source.out.jitter)); + // The "auto" playout target this track needs, per doc/concept/audio-jitter.md. + const playout = this.#signals.computed((effect) => { + const measured = effect.get(this.#measured); + if (measured === undefined) return undefined; + return target({ measured, advertised: effect.get(this.source.out.jitter), frame: effect.get(this.#frame) }); + }); + this.#signals.cleanup(this.sync.register(playout)); this.#identity = this.#signals.computed((effect) => { const config = effect.get(this.source.out.config); return config ? playbackIdentity(config) : undefined; @@ -220,6 +243,9 @@ export class Decoder { effect.run((inner) => { this.#out.stalled.set(inner.get(ring.stalled)); }); + effect.run((inner) => { + this.#out.underruns.set(inner.get(ring.underruns)); + }); effect.set(this.#out.root, worklet); }); @@ -255,32 +281,38 @@ export class Decoder { ring.setLatency(ringSamples(ring.rate, delay)); } - // Re-anchor when the delay floor *increases*. A larger floor needs a deeper cushion: video - // rebuilds it implicitly (its per-frame sync.wait() reads the live buffer, so it just holds - // longer), but the audio ring keeps draining at its old depth -- resize() (via setLatency) only - // re-stalls an *empty* ring, so a mid-playback ring never refills to the new floor and audio runs - // ahead of video (the "raise latency, only video re-buffers" desync). reset() re-stalls the ring - // so it refills to the new floor. Watch the latency target and media delay, excluding adaptive - // RTT jitter, and debounce so a slider drag coalesces into one re-anchor. Decreases are left to - // natural catch-up. + // Park playback when the target *deepens*, so the ring refills to it. Video rebuilds a deeper + // cushion implicitly (its per-frame sync.wait() reads the live buffer, so it just holds longer), + // but the audio ring keeps draining at its old depth: setLatency only raises the bar a future + // refill has to clear, so a ring already playing never gets deeper and audio runs ahead of video + // (the "raise latency, only video re-buffers" desync). Stalling spends the deficit as silence, + // once, instead of leaving it to the underrun that the shallow buffer eventually causes anyway. + // + // The derived delay, not the user's setting, since the arrival estimator moves it too. Only a + // deepening worth more than a frame counts, so the estimator's small refinements ride through, + // and the debounce coalesces a slider drag or a converging estimate into a single stall. + // Decreases are left to natural catch-up. #runLatencyReanchor(effect: Effect): void { - const delay = effect.get(this.sync.out.delay); - const jitter = effect.get(this.sync.out.jitter); - const floor = reanchorFloor({ - delay: effect.get(this.sync.in.delay), - media: Time.Milli.sub(delay, jitter), - }); - if (this.#prevFloor === undefined) { + const target = effect.get(this.sync.out.delay); + const step = effect.get(this.#frame) ?? Time.Milli.zero; + if (this.#prevTarget === undefined) { // Startup: the initial fill already builds the cushion; just record the baseline. - this.#prevFloor = floor; + this.#prevTarget = target; + return; + } + // A decrease lands at once: `#runLatency` has already let the ring shrink to it, so a rise + // that follows within the debounce has to be measured from there. + if (target < this.#prevTarget) { + this.#prevTarget = target; return; } - // When the timer fires, the floor read above is still current: any change would have rerun + // When the timer fires, the target read above is still current: any change would have rerun // this effect (tearing down the timer), so compare it against the pre-change baseline directly. - const baseline = this.#prevFloor; + const baseline = this.#prevTarget; effect.timer(() => { - if (floor > baseline) this.reset(); - this.#prevFloor = floor; + const next = reanchor(baseline, target, step); + if (next.stall) this.#ring?.stall(); + this.#prevTarget = next.baseline; }, LATENCY_REANCHOR_DEBOUNCE_MS); } @@ -299,6 +331,7 @@ export class Decoder { if (!identity) return; const config = identity.decoder; + this.#frame.set(frameDuration(config)); // Honor a per-rendition `broadcast` override: subscribe on the resolved source // broadcast instead of the catalog's own broadcast. @@ -348,6 +381,11 @@ export class Decoder { this.#out.buffered.update(() => Container.mergeBufferedRanges(network, decode)); }); + // Feed the arrival estimate into the playout target. Cleared on teardown so a departed track + // stops holding the buffer open. + effect.run((inner) => this.#measured.set(inner.get(consumer.spread))); + effect.cleanup(() => this.#measured.set(undefined)); + effect.spawn(async () => { const loaded = await Util.Libav.polyfill(); if (!loaded) return; // cancelled @@ -403,6 +441,9 @@ export class Decoder { const timestamp = Time.Milli.fromMicro(frame.timestamp as Time.Micro); this.sync.received(timestamp, "audio"); + const duration = packetDuration(config.codec, frame); + if (duration !== undefined) this.#frame.set(duration); + this.#out.stats.update((stats) => ({ bytesReceived: (stats?.bytesReceived ?? 0) + frame.payload.byteLength, })); @@ -455,6 +496,11 @@ export class Decoder { this.#out.buffered.update(() => Container.mergeBufferedRanges(network, decode)); }); + // Feed the arrival estimate into the playout target. Cleared on teardown so a departed track + // stops holding the buffer open. + effect.run((inner) => this.#measured.set(inner.get(consumer.spread))); + effect.cleanup(() => this.#measured.set(undefined)); + effect.spawn(async () => { const loaded = await Util.Libav.polyfill(); if (!loaded) return; // cancelled @@ -492,6 +538,9 @@ export class Decoder { const timestamp = Time.Milli.fromMicro(frame.timestamp); this.sync.received(timestamp, "audio"); + const duration = packetDuration(config.codec, frame); + if (duration !== undefined) this.#frame.set(duration); + this.#out.stats.update((stats) => ({ bytesReceived: (stats?.bytesReceived ?? 0) + frame.payload.byteLength, })); diff --git a/js/watch/src/audio/latency.test.ts b/js/watch/src/audio/latency.test.ts index ecc7c1f4d2..f5b35444f1 100644 --- a/js/watch/src/audio/latency.test.ts +++ b/js/watch/src/audio/latency.test.ts @@ -1,17 +1,21 @@ import { describe, expect, it } from "bun:test"; import type { Time } from "@moq/net"; -import { reanchorFloor, ringSamples } from "./latency"; +import { reanchor, ringSamples, target } from "./latency"; const ms = (value: number) => value as Time.Milli; -describe("reanchorFloor", () => { - it("includes the fixed delay and the largest media delay", () => { - expect(reanchorFloor({ delay: ms(100), media: ms(80) })).toBe(ms(180)); +describe("target", () => { + it("adds one frame to the measured term", () => { + expect(target({ measured: ms(40), frame: ms(20) })).toBe(ms(60)); }); - it("tracks rendition delay without adaptive RTT jitter", () => { - expect(reanchorFloor({ delay: "auto", media: ms(80) })).toBe(ms(80)); - expect(reanchorFloor({ delay: "auto", media: ms(200) })).toBe(ms(200)); + it("floors the measured term at the advertised span instead of adding it", () => { + expect(target({ measured: ms(40), advertised: ms(200), frame: ms(20) })).toBe(ms(220)); + expect(target({ measured: ms(300), advertised: ms(200), frame: ms(20) })).toBe(ms(320)); + }); + + it("is the measured term alone when nothing else is known", () => { + expect(target({ measured: ms(100) })).toBe(ms(100)); }); }); @@ -31,3 +35,21 @@ describe("ringSamples", () => { expect(ringSamples(48_000, ms(100))).toBe(4_800); }); }); + +describe("reanchor", () => { + it("re-stalls once a run of one-frame rises adds up to more than a frame", () => { + let baseline = ms(40); + const stalls = [60, 80, 100].map((next) => { + const result = reanchor(baseline, ms(next), ms(20)); + baseline = result.baseline; + return result.stall; + }); + expect(stalls).toEqual([false, true, false]); + expect(baseline).toBe(ms(80)); + }); + + it("follows a fall down, so the next rise is measured from there", () => { + expect(reanchor(ms(100), ms(60), ms(20))).toEqual({ stall: false, baseline: ms(60) }); + expect(reanchor(ms(60), ms(100), ms(20))).toEqual({ stall: true, baseline: ms(100) }); + }); +}); diff --git a/js/watch/src/audio/latency.ts b/js/watch/src/audio/latency.ts index 1766d2b8d5..b9dd78becd 100644 --- a/js/watch/src/audio/latency.ts +++ b/js/watch/src/audio/latency.ts @@ -1,21 +1,42 @@ import { Time } from "@moq/net"; -import type { Delay } from "../sync"; -/** The inputs that determine whether audio needs a deeper playback cushion. */ -export interface ReanchorFloor { - /** How far playback trails the live edge. */ - delay: Delay; +/** The terms of the audio playout target, all in milliseconds. */ +export interface Target { + /** The arrival estimate from the container consumer. */ + measured: Time.Milli; + /** The flush span the rendition advertises, if any. */ + advertised?: Time.Milli; + /** The codec's frame duration, if known. */ + frame?: Time.Milli; +} - /** Largest additional delay required by the registered media decoders. */ - media?: Time.Milli; +/** + * The "auto" playout target: the measured term floored by the advertised span, plus one frame. + * + * A floor rather than a sum, because the receiver's measurement already contains the publisher's + * flush delay. See doc/concept/audio-jitter.md. + */ +export function target(props: Target): Time.Milli { + const floored = Time.Milli.max(props.measured, props.advertised ?? Time.Milli.zero); + return Time.Milli.add(floored, props.frame ?? Time.Milli.zero); } -/** The stable delay floor whose increase requires the audio ring to refill. */ -export function reanchorFloor(props: ReanchorFloor): Time.Milli { - // "auto" and "instant" contribute nothing: the adaptive RTT component is deliberately excluded - // so an RTT wiggle doesn't re-anchor, and "instant" holds nothing at all. - const target = typeof props.delay === "number" ? props.delay : Time.Milli.zero; - return Time.Milli.add(target, props.media ?? Time.Milli.zero); +/** Whether a deeper target re-stalls the ring, and the baseline the next target is compared against. */ +export interface Reanchor { + stall: boolean; + baseline: Time.Milli; +} + +/** + * Compare a new target against the depth the ring last filled to. + * + * A rise of more than `step` re-stalls. A smaller one rides through but keeps the old baseline, so a + * run of one-bucket rises still re-stalls once they add up. A fall lowers the baseline, since the + * ring's latency skip follows the target down on its own. + */ +export function reanchor(baseline: Time.Milli, target: Time.Milli, step: Time.Milli): Reanchor { + if (target - baseline > step) return { stall: true, baseline: target }; + return { stall: false, baseline: Time.Milli.min(baseline, target) }; } // An AudioWorkletProcessor renders in fixed 128-sample quanta, so a ring shallower than one can diff --git a/js/watch/src/audio/render-worklet.ts b/js/watch/src/audio/render-worklet.ts index 723dfcfa99..313cb67f31 100644 --- a/js/watch/src/audio/render-worklet.ts +++ b/js/watch/src/audio/render-worklet.ts @@ -2,11 +2,17 @@ import type { Message, State } from "./render"; import { AudioRingBuffer } from "./ring-buffer"; import { SharedRingBuffer } from "./shared-ring-buffer"; +// Samples to fade over when playback stops or resumes short of a full quantum. A shortfall +// otherwise steps straight to silence and back, which is an audible click on every underrun. +const RAMP = 64; + class Render extends AudioWorkletProcessor { // Set after init, depending on which path the main thread chose. #backend?: SharedRingBuffer | AudioRingBuffer; #underflow = 0; #stateCounter = 0; + // Whether the previous quantum ended short, so the next one fades back in. + #short = false; constructor() { super(); @@ -18,10 +24,12 @@ class Render extends AudioWorkletProcessor { const previous = this.#backend instanceof SharedRingBuffer ? this.#backend : undefined; this.#backend = new SharedRingBuffer(msg, previous); this.#underflow = 0; + this.#short = false; } else if (msg.type === "init-post") { console.log("[audio-worklet] init-post: using postMessage path"); this.#backend = new AudioRingBuffer(msg); this.#underflow = 0; + this.#short = false; } else if (msg.type === "data") { // Only meaningful in post mode. if (this.#backend instanceof AudioRingBuffer) this.#backend.write(msg.timestamp, msg.data); @@ -34,6 +42,9 @@ class Render extends AudioWorkletProcessor { } else if (msg.type === "reset") { // Only meaningful in post mode; shared mode resets via the control array. if (this.#backend instanceof AudioRingBuffer) this.#backend.reset(); + } else if (msg.type === "stall") { + // Only meaningful in post mode; shared mode stalls via the control array. + if (this.#backend instanceof AudioRingBuffer) this.#backend.stall(); } }; } @@ -44,10 +55,29 @@ class Render extends AudioWorkletProcessor { const samplesRead = backend?.read(output) ?? 0; if (samplesRead < output[0].length) { + // Fade the tail of what we did read down to the silence that follows it. + for (const channel of output) { + const ramp = Math.min(RAMP, samplesRead); + for (let i = 0; i < ramp; i++) { + channel[samplesRead - ramp + i] *= 1 - (i + 1) / ramp; + } + } this.#underflow += output[0].length - samplesRead; - } else if (this.#underflow > 0 && backend) { - console.debug(`audio underflow: ${Math.round((1000 * this.#underflow) / backend.rate)}ms`); - this.#underflow = 0; + this.#short = true; + } else { + if (this.#short) { + // Fade back in from the silence the shortfall left behind. + for (const channel of output) { + for (let i = 0; i < Math.min(RAMP, samplesRead); i++) { + channel[i] *= (i + 1) / RAMP; + } + } + this.#short = false; + } + if (this.#underflow > 0 && backend) { + console.debug(`audio underflow: ${Math.round((1000 * this.#underflow) / backend.rate)}ms`); + this.#underflow = 0; + } } // In post mode the main thread can't read worklet state directly, so we @@ -61,6 +91,7 @@ class Render extends AudioWorkletProcessor { type: "state", timestamp: backend.timestamp, stalled: backend.stalled, + underruns: backend.underruns, }; this.port.postMessage(state); } diff --git a/js/watch/src/audio/render.ts b/js/watch/src/audio/render.ts index e0c8252468..a537ac1bbc 100644 --- a/js/watch/src/audio/render.ts +++ b/js/watch/src/audio/render.ts @@ -2,7 +2,7 @@ import type { Time } from "@moq/net"; import type { SharedRingBufferInit } from "./shared-ring-buffer"; /** Everything the main thread sends the render worklet over its port. */ -export type Message = InitShared | InitPost | Data | Latency | Reset | Truncate; +export type Message = InitShared | InitPost | Data | Latency | Reset | Stall | Truncate; export type ToMain = State; /** Init message when SharedArrayBuffer is available. */ @@ -26,6 +26,14 @@ export interface Reset { type: "reset"; } +/** + * Hold playback until the buffer holds the target again, keeping what is buffered (fallback path + * only; the shared path stalls via Atomics). + */ +export interface Stall { + type: "stall"; +} + /** * Drop buffered samples at or after `timestamp`, keeping what is already due (fallback path only; * the shared path truncates via Atomics). @@ -53,4 +61,6 @@ export interface State { type: "state"; timestamp: Time.Micro; stalled: boolean; + // How many times the ring has run dry mid-playback, cumulative. + underruns: number; } diff --git a/js/watch/src/audio/replay.test.ts b/js/watch/src/audio/replay.test.ts new file mode 100644 index 0000000000..6cc04d7770 --- /dev/null +++ b/js/watch/src/audio/replay.test.ts @@ -0,0 +1,181 @@ +import { describe, expect, it } from "bun:test"; +import { Time } from "@moq/net"; +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. + +const RATE = 48000; +const QUANTUM = 128; // an AudioWorklet render quantum +const CHUNK_MS = 20; // one Opus frame +const CHUNK = (RATE * CHUNK_MS) / 1000; + +/** Deterministic PRNG, so a trace is the same on every machine. */ +function rng(seed: number): () => number { + let a = seed >>> 0; + return () => { + a = (a + 0x6d2b79f5) >>> 0; + let t = Math.imul(a ^ (a >>> 15), 1 | a); + t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t; + return ((t ^ (t >>> 14)) >>> 0) / 4294967296; + }; +} + +/** One frame of a trace: when it was captured, and the wall time it reached us. */ +interface Arrival { + media: number; + arrival: number; +} + +/** + * A sender emitting a frame every `CHUNK_MS`, holding `burst - 1` of them back and flushing the + * run at once, plus `spread` ms of per-flush network jitter. `burst` of 1 is an evenly paced + * sender; anything above it is the PES packing an importer produces. + */ +function trace(frames: number, burst: number, spread: number, seed = 7): Arrival[] { + const rand = rng(seed); + const out: Arrival[] = []; + let held: number[] = []; + for (let i = 0; i < frames; i++) { + const media = i * CHUNK_MS; + held.push(media); + if (held.length < burst) continue; + // Every frame keeps its own capture time; the whole run shares one arrival. + const arrival = media + 50 + rand() * spread; + for (const m of held) out.push({ media: m, arrival }); + held = []; + } + return out; +} + +/** + * 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. + */ +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; +} + +/** The two rings behind one interface, since the harness drives them identically. */ +interface Ring { + insert(timestamp: Time.Micro, data: Float32Array[]): void; + read(output: Float32Array[]): number; + readonly length: number; +} + +function shared(latencyMs: number): Ring { + const samples = Math.ceil((RATE * latencyMs) / 1000); + const ring = new SharedRingBuffer(allocSharedRingBuffer(1, Math.max(RATE, samples * 2), RATE)); + ring.setLatency(samples); + return ring; +} + +function post(latencyMs: number): Ring { + const ring = new AudioRingBuffer({ rate: RATE, channels: 1, latency: latencyMs as Time.Milli }); + return { + insert: (timestamp, data) => ring.write(timestamp, data), + read: (output) => ring.read(output), + get length() { + return ring.length; + }, + }; +} + +interface Result { + /** Quanta rendered short of a full block after the warmup, i.e. audible underruns. */ + underruns: number; + /** Samples inserted after the warmup that were never played: skipped or overflow-dropped. */ + skipped: number; +} + +/** + * Play `t` through `ring` in real time: insert every frame the moment it arrives, and pull one + * render quantum every quantum's worth of wall time, which is what the AudioWorklet does. + */ +function replay(ring: Ring, t: Arrival[], warmupMs: number): Result { + const step = (QUANTUM / RATE) * 1000; + const output = [new Float32Array(QUANTUM)]; + const end = t[t.length - 1].arrival; + + let next = 0; + let played = 0; + let inserted = 0; + let short = 0; + let started = false; + let mark: { played: number; inserted: number; buffered: number; short: number } | undefined; + + for (let now = 0; now < end; now += step) { + while (next < t.length && t[next].arrival <= now) { + ring.insert(Time.Micro.fromMilli(t[next].media as Time.Milli), [new Float32Array(CHUNK).fill(0.5)]); + inserted += CHUNK; + next++; + } + + const count = ring.read(output); + played += count; + if (started && count < QUANTUM) short++; + if (count > 0) started = true; + + if (now >= warmupMs && !mark) mark = { played, inserted, buffered: ring.length, short }; + } + + const from = mark ?? { played, inserted, buffered: ring.length, short }; + return { + underruns: short - from.short, + skipped: inserted - from.inserted - (played - from.played) - (ring.length - from.buffered), + }; +} + +const RINGS: Array<[string, (latencyMs: number) => Ring]> = [ + ["shared", shared], + ["post", post], +]; + +describe.each(RINGS)("%s ring replay", (_name, build) => { + it("plays an evenly paced sender without underruns or skips", () => { + const t = trace(600, 1, 10); + expect(replay(build(target(t)), t, 2000)).toEqual({ underruns: 0, skipped: 0 }); + }); + + it("plays a bursty sender at the measured target", () => { + // 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); + }); + + it("keeps a five frame flush span playing", () => { + // 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); + }); + + it("underruns constantly when the target ignores the arrival spread", () => { + // 46ms is what the round-trip formula produced on the connection this was measured on: it + // describes the network and says nothing about a sender that flushes five frames at once. + const t = trace(600, 5, 5); + const result = replay(build(46), t, 2000); + expect(result.underruns).toBeGreaterThan(100); + expect(result.skipped).toBeGreaterThan(10 * CHUNK); + }); +}); diff --git a/js/watch/src/audio/ring-buffer.test.ts b/js/watch/src/audio/ring-buffer.test.ts index 5d1fb2ceba..701dfe78c3 100644 --- a/js/watch/src/audio/ring-buffer.test.ts +++ b/js/watch/src/audio/ring-buffer.test.ts @@ -35,7 +35,9 @@ describe("initialization", () => { it("should initialize with valid parameters", () => { const buffer = new AudioRingBuffer({ rate: 48000, channels: 2, latency: 100 as Time.Milli }); - expect(buffer.capacity).toBe(4800); // 48000 * 0.1 + // Capacity is twice the target: the ring has to be able to hold the slack the overflow band + // tolerates, or every chunk arriving on a full ring would drop the oldest samples. + expect(buffer.capacity).toBe(9600); // 48000 * 0.1 * 2 expect(buffer.length).toBe(0); }); @@ -281,9 +283,10 @@ describe("stall behavior", () => { describe("ring buffer wrapping", () => { it("should wrap around when buffer is full", () => { + // 100 sample target, so 200 samples of capacity. const buffer = new AudioRingBuffer({ rate: 1000, channels: 1, latency: 100 as Time.Milli }); - // Fill the buffer + // Fill to the target write(buffer, 0 as Time.Milli, 100, { channels: 1, value: 1.0 }); expect(buffer.stalled).toBe(false); @@ -297,22 +300,23 @@ describe("ring buffer wrapping", () => { // Now we have 100 samples available (50-149) expect(buffer.length).toBe(100); - // Write 50 more samples at timestamp 150, this will wrap around + // One chunk above the target is tolerated rather than dropped (50-199). write(buffer, 150 as Time.Milli, 50, { channels: 1, value: 3.0 }); + expect(buffer.length).toBe(150); - // Should still have 100 samples (buffer is at capacity) + // Past the band: drop back to the target, wrapping abs 200-249 onto slots 0-49. + write(buffer, 200 as Time.Milli, 50, { channels: 1, value: 4.0 }); expect(buffer.length).toBe(100); - // Read all 100 samples + // Read all 100 samples: abs 150-199 = 3.0, abs 200-249 = 4.0. const output2 = read(buffer, 100, 1); expect(output2[0].length).toBe(100); - // First 50 should be 2.0, next 50 should be 3.0 for (let i = 0; i < 50; i++) { - expect(output2[0][i]).toBe(2.0); + expect(output2[0][i]).toBe(3.0); } for (let i = 50; i < 100; i++) { - expect(output2[0][i]).toBe(3.0); + expect(output2[0][i]).toBe(4.0); } }); }); @@ -390,7 +394,7 @@ describe("edge cases", () => { describe("resize", () => { it("should resize to a larger buffer", () => { const buffer = new AudioRingBuffer({ rate: 1000, channels: 1, latency: 100 as Time.Milli }); - expect(buffer.capacity).toBe(100); + expect(buffer.capacity).toBe(200); // Write 50 samples write(buffer, 0 as Time.Milli, 50, { channels: 1, value: 1.0 }); @@ -399,27 +403,27 @@ describe("resize", () => { // Resize to larger buffer (200ms = 200 samples) buffer.resize(200 as Time.Milli); - expect(buffer.capacity).toBe(200); + expect(buffer.capacity).toBe(400); expect(buffer.length).toBe(50); // Samples preserved expect(buffer.stalled).toBe(true); // Should trigger stall }); it("should resize to a smaller buffer and keep the most recent samples", () => { const buffer = new AudioRingBuffer({ rate: 1000, channels: 1, latency: 100 as Time.Milli }); - expect(buffer.capacity).toBe(100); + expect(buffer.capacity).toBe(200); // Write 80 samples: first 40 with value 1.0, next 40 with value 2.0 write(buffer, 0 as Time.Milli, 40, { channels: 1, value: 1.0 }); write(buffer, 40 as Time.Milli, 40, { channels: 1, value: 2.0 }); expect(buffer.length).toBe(80); - // Resize to smaller buffer (50ms = 50 samples) - // Should keep the most recent 50 samples (samples 30-79) - buffer.resize(50 as Time.Milli); + // Resize to a 30 sample target (60 samples of capacity). + // Should keep the most recent 60 samples (samples 20-79) + buffer.resize(30 as Time.Milli); - expect(buffer.capacity).toBe(50); - expect(buffer.length).toBe(50); // Truncated to new capacity - expect(buffer.stalled).toBe(true); // Should trigger stall + expect(buffer.capacity).toBe(60); + expect(buffer.length).toBe(60); // Truncated to new capacity + expect(buffer.stalled).toBe(true); // Never reached the old 100 sample target }); it("should be a no-op when capacity is unchanged", () => { @@ -434,7 +438,7 @@ describe("resize", () => { // Should still not be stalled (no-op) expect(buffer.stalled).toBe(false); - expect(buffer.capacity).toBe(100); + expect(buffer.capacity).toBe(200); }); it("should throw on zero latency", () => { @@ -456,8 +460,8 @@ describe("resize", () => { // Resize to smaller buffer buffer.resize(50 as Time.Milli); - expect(buffer.capacity).toBe(50); - expect(buffer.length).toBe(50); + expect(buffer.capacity).toBe(100); + expect(buffer.length).toBe(60); expect(buffer.stalled).toBe(true); }); @@ -468,7 +472,7 @@ describe("resize", () => { // Resize empty buffer buffer.resize(200 as Time.Milli); - expect(buffer.capacity).toBe(200); + expect(buffer.capacity).toBe(400); expect(buffer.length).toBe(0); expect(buffer.stalled).toBe(true); }); @@ -488,10 +492,10 @@ describe("resize", () => { write(buffer, 100 as Time.Milli, 30, { channels: 1, value: 2.0 }); expect(buffer.length).toBe(70); // 130 - 60 - // Resize to 50 samples - should keep most recent 50 (samples 80-129) + // Resize to a 50 sample target - 100 of capacity, so all 70 survive (samples 60-129) buffer.resize(50 as Time.Milli); - expect(buffer.capacity).toBe(50); - expect(buffer.length).toBe(50); + expect(buffer.capacity).toBe(100); + expect(buffer.length).toBe(70); expect(buffer.stalled).toBe(false); }); @@ -505,22 +509,26 @@ describe("resize", () => { buffer.resize(50 as Time.Milli); expect(buffer.stalled).toBe(true); - // Write new data to fill the buffer and exit stall - // The overflow will discard preserved samples and advance readIndex + // Write new data to reach the new target and exit stall. Nothing is dropped: the ring has + // room for the preserved samples plus a chunk above the target. write(buffer, 50 as Time.Milli, 50, { channels: 1, value: 2.0 }); expect(buffer.stalled).toBe(false); + expect(buffer.length).toBe(100); - // Read should return the new data (value 2.0) - const output = read(buffer, 50, 1); - expect(output[0].length).toBe(50); + // Read plays the preserved samples first, then the new ones. + const output = read(buffer, 100, 1); + expect(output[0].length).toBe(100); for (let i = 0; i < 50; i++) { + expect(output[0][i]).toBe(1.0); + } + for (let i = 50; i < 100; i++) { expect(output[0][i]).toBe(2.0); } }); it("should preserve samples correctly when resizing larger then filling", () => { const buffer = new AudioRingBuffer({ rate: 1000, channels: 1, latency: 50 as Time.Milli }); - expect(buffer.capacity).toBe(50); + expect(buffer.capacity).toBe(100); // Write 30 samples write(buffer, 0 as Time.Milli, 30, { channels: 1, value: 1.0 }); @@ -528,7 +536,7 @@ describe("resize", () => { // Resize to larger buffer (100ms = 100 samples) buffer.resize(100 as Time.Milli); - expect(buffer.capacity).toBe(100); + expect(buffer.capacity).toBe(200); expect(buffer.length).toBe(30); // All samples preserved expect(buffer.stalled).toBe(true); @@ -569,26 +577,26 @@ describe("resize", () => { // abs 110-119 → slots 10-19, value 3.0. write(buffer, 110 as Time.Milli, 10, { channels: 1, value: 3.0 }); - // Preserved range after resize: most-recent 50 samples = abs 70..119. - // copyStart = 120 - 50 = 70, and 70 % 50 = 20 (non-zero → triggers the bug). + // Preserved range after resize: all 100 buffered samples = abs 20..119. + // copyStart = 120 - 100 = 20, and 20 % 100 = 20 (non-zero → triggers the bug). buffer.resize(50 as Time.Milli); - expect(buffer.capacity).toBe(50); - expect(buffer.length).toBe(50); + expect(buffer.capacity).toBe(100); + expect(buffer.length).toBe(100); expect(buffer.stalled).toBe(false); // Read the preserved samples. Expected layout by absolute index: - // abs 70..99 = 1.0 (30 samples, from the initial fill) + // abs 20..99 = 1.0 (80 samples, from the initial fill) // abs 100..109 = 2.0 (10 samples) // abs 110..119 = 3.0 (10 samples) - const output = read(buffer, 50, 1); - expect(output[0].length).toBe(50); - for (let i = 0; i < 30; i++) { + const output = read(buffer, 100, 1); + expect(output[0].length).toBe(100); + for (let i = 0; i < 80; i++) { expect(output[0][i]).toBe(1.0); } - for (let i = 30; i < 40; i++) { + for (let i = 80; i < 90; i++) { expect(output[0][i]).toBe(2.0); } - for (let i = 40; i < 50; i++) { + for (let i = 90; i < 100; i++) { expect(output[0][i]).toBe(3.0); } }); @@ -607,16 +615,16 @@ describe("resize", () => { // Wrap the writer over slots 0-29 with value 2.0 (abs 100-129). write(buffer, 100 as Time.Milli, 30, { channels: 1, value: 2.0 }); - // Resize smaller. Preserved = last 50 samples = abs 80..129. + // Resize smaller. All 90 buffered samples fit the new 100 sample capacity: abs 40..129. buffer.resize(50 as Time.Milli); - expect(buffer.length).toBe(50); + expect(buffer.length).toBe(90); expect(buffer.stalled).toBe(false); // Write the next 10 samples at their real timestamp. This should append, // not create a gap or be discarded as "too old". write(buffer, 130 as Time.Milli, 10, { channels: 1, value: 3.0 }); - // Buffer capacity is 50, so the oldest 10 samples (abs 80..89) drop out. + // That put the ring past the band, so it drops back to the 50 sample target. // Remaining: abs 90..99 = 1.0 (10), abs 100..129 = 2.0 (30), abs 130..139 = 3.0 (10). expect(buffer.length).toBe(50); @@ -667,7 +675,7 @@ describe("live anchoring", () => { // with zeros, un-stalled on that overflow, and left the playhead a floor before the frame. it("anchors a live ring to the first frame instead of gap-filling from zero", () => { const buffer = new AudioRingBuffer({ rate: 1000, channels: 1, latency: 100 as Time.Milli }); - expect(buffer.capacity).toBe(100); + expect(buffer.capacity).toBe(200); write(buffer, 2000 as Time.Milli, 40, { channels: 1, value: 0.5 }); // Still short of the floor, so nothing plays yet and the ring holds only real samples. @@ -685,6 +693,102 @@ describe("live anchoring", () => { }); }); +describe("re-buffer", () => { + it("re-stalls and refills to the target after running dry", () => { + const buffer = new AudioRingBuffer({ rate: 1000, channels: 1, latency: 40 as Time.Milli }); + + write(buffer, 0 as Time.Milli, 20, { channels: 1, value: 1.0 }); + write(buffer, 20 as Time.Milli, 20, { channels: 1, value: 1.0 }); + expect(buffer.stalled).toBe(false); + + // Drain it. + expect(read(buffer, 40, 1)[0].length).toBe(40); + expect(buffer.length).toBe(0); + + // The next read finds nothing and parks playback rather than handing the worklet a quantum + // of silence and trying again on the next one. + expect(read(buffer, 20, 1)[0].length).toBe(0); + expect(buffer.stalled).toBe(true); + expect(buffer.underruns).toBe(1); + + // A single chunk is not the target, so playback stays parked. + write(buffer, 40 as Time.Milli, 20, { channels: 1, value: 2.0 }); + expect(buffer.stalled).toBe(true); + expect(read(buffer, 20, 1)[0].length).toBe(0); + + // Reaching the target resumes it, and nothing buffered in the meantime was lost. + write(buffer, 60 as Time.Milli, 20, { channels: 1, value: 2.0 }); + expect(buffer.stalled).toBe(false); + expect(read(buffer, 40, 1)[0].length).toBe(40); + }); + + it("counts a quantum it could only partly fill as an underrun", () => { + const buffer = new AudioRingBuffer({ rate: 1000, channels: 1, latency: 40 as Time.Milli }); + + write(buffer, 0 as Time.Milli, 20, { channels: 1, value: 1.0 }); + write(buffer, 20 as Time.Milli, 20, { channels: 1, value: 1.0 }); + expect(read(buffer, 30, 1)[0].length).toBe(30); + + // Ten samples remain against a quantum of twenty: the rest of the quantum is silence, which + // is an underrun whether or not a chunk lands before the next read. It is not a stall: a + // refill would cost the whole target to cover a gap shorter than one quantum. + expect(read(buffer, 20, 1)[0].length).toBe(10); + expect(buffer.stalled).toBe(false); + expect(buffer.underruns).toBe(1); + + // A chunk landing before the next read keeps playback going. + write(buffer, 40 as Time.Milli, 20, { channels: 1, value: 2.0 }); + expect(read(buffer, 20, 1)[0].length).toBe(20); + expect(buffer.underruns).toBe(1); + }); + + it("stall() parks playback without discarding what is buffered", () => { + const buffer = new AudioRingBuffer({ rate: 1000, channels: 1, latency: 40 as Time.Milli }); + + write(buffer, 0 as Time.Milli, 20, { channels: 1, value: 1.0 }); + write(buffer, 20 as Time.Milli, 20, { channels: 1, value: 1.0 }); + expect(read(buffer, 20, 1)[0].length).toBe(20); + + // The target deepens, so playback parks until the ring holds the new depth. + buffer.resize(60 as Time.Milli); + buffer.stall(); + expect(buffer.stalled).toBe(true); + expect(read(buffer, 20, 1)[0].length).toBe(0); + expect(buffer.length).toBe(20); // nothing was thrown away, unlike reset() + + write(buffer, 40 as Time.Milli, 20, { channels: 1, value: 2.0 }); + expect(buffer.stalled).toBe(true); // 40 buffered, target is 60 + + write(buffer, 60 as Time.Milli, 20, { channels: 1, value: 2.0 }); + expect(buffer.stalled).toBe(false); + + // Playback resumes on the same timeline: the samples buffered before the stall play first. + const output = read(buffer, 20, 1); + expect(output[0].length).toBe(20); + expect(output[0][0]).toBe(1.0); + expect(buffer.underruns).toBe(0); + }); + + it("stall() is a no-op once the ring already holds the target", () => { + const buffer = new AudioRingBuffer({ rate: 1000, channels: 1, latency: 40 as Time.Milli }); + write(buffer, 0 as Time.Milli, 20, { channels: 1, value: 1.0 }); + write(buffer, 20 as Time.Milli, 20, { channels: 1, value: 1.0 }); + write(buffer, 40 as Time.Milli, 20, { channels: 1, value: 1.0 }); + + buffer.resize(60 as Time.Milli); + buffer.stall(); + expect(buffer.stalled).toBe(false); + expect(read(buffer, 20, 1)[0].length).toBe(20); + }); + + it("does not count an underrun while parked", () => { + const buffer = new AudioRingBuffer({ rate: 1000, channels: 1, latency: 40 as Time.Milli }); + read(buffer, 20, 1); + read(buffer, 20, 1); + expect(buffer.underruns).toBe(0); + }); +}); + describe("buffered mode", () => { function createBuffered(latency: number) { return new AudioRingBuffer({ rate: 1000, channels: 1, latency: latency as Time.Milli, buffered: true }); diff --git a/js/watch/src/audio/ring-buffer.ts b/js/watch/src/audio/ring-buffer.ts index fb94f03692..a9154a7cd1 100644 --- a/js/watch/src/audio/ring-buffer.ts +++ b/js/watch/src/audio/ring-buffer.ts @@ -8,6 +8,13 @@ export class AudioRingBuffer { readonly rate: number; readonly channels: number; #stalled = true; + #underruns = 0; + + // Samples in the most recent write, i.e. one decoded chunk. The overflow band tolerates this + // much above the target before dropping, so a ring sitting exactly on the target doesn't drop + // audio the moment the next chunk lands. The most recent write rather than a running maximum: + // one oversized decode would otherwise widen the band for the life of the ring. + #chunk = 0; // Buffered mode: play through everything buffered without skipping ahead. readonly #buffered: boolean; @@ -33,9 +40,11 @@ export class AudioRingBuffer { this.channels = props.channels; this.#buffered = props.buffered ?? false; - // The ring holds the latency floor as PCM. Buffered mode gives it headroom above the floor - // so the backpressure-paced decode loop (on the main thread) doesn't overflow-drop; the rest - // of the lookahead stays encoded upstream. + // The ring holds the latency floor as PCM, with headroom above it. Sizing capacity to the + // floor exactly makes the ring physically incapable of holding the slack the overflow band + // wants, so every chunk that arrives while the ring is on target drops the oldest samples. + // In buffered mode the headroom is also what keeps the backpressure-paced decode loop (on + // the main thread) from overflow-dropping; the rest of the lookahead stays encoded upstream. const capacity = this.#capacityFor(this.#latencySamples); this.#buffer = []; @@ -45,13 +54,18 @@ export class AudioRingBuffer { } #capacityFor(latencySamples: number): number { - return this.#buffered ? latencySamples * 2 : latencySamples; + return latencySamples * 2; } get stalled(): boolean { return this.#stalled; } + /** How many times the reader has run dry mid-playback. */ + get underruns(): number { + return this.#underruns; + } + get timestamp(): Time.Micro { return Time.Micro.fromSecond((this.#readIndex / this.rate) as Time.Second); } @@ -127,13 +141,18 @@ export class AudioRingBuffer { } const end = start + samples; - - // Check if we need to discard old samples to prevent overflow - const overflow = end - this.#readIndex - this.#buffer[0].length; - if (overflow >= 0) { - // Discard old samples and exit stalled mode - this.#stalled = false; - this.#readIndex += overflow; + this.#chunk = data[0].length; + + // Bound the ring. While playing, drop the oldest once it holds a whole chunk more than the + // target and land back on the target: frames arrive one chunk at a time, so a ring sitting + // exactly on the target is a chunk above it the moment the next one lands, and dropping on + // that overshoot discards audio on every single write. While stalled the reader is not + // consuming, so only the hard capacity applies; the band would throw away the very audio the + // refill is accumulating. Buffered mode plays through everything, so it is capacity-bound too. + const playing = !this.#stalled && !this.#buffered; + const limit = playing ? Math.min(this.#latencySamples + this.#chunk, this.capacity) : this.capacity; + if (end - this.#readIndex > limit) { + this.#readIndex = end - (playing ? Math.min(this.#latencySamples, this.capacity) : this.capacity); } // Fill gaps with zeros if there's a discontinuity @@ -171,9 +190,10 @@ export class AudioRingBuffer { this.#writeIndex = end; } - // Start playback once we've buffered the latency target. In buffered mode the cap - // is large, so we usually un-stall here rather than via the overflow path above. - if (this.#buffered && this.length >= this.#latencySamples) { + // Start playback once we've buffered the latency target. This is the only way out of a + // stall, so an underrun mid-playback refills to the target before resuming rather than + // playing the next chunk on an empty cushion. + if (this.length >= this.#latencySamples) { this.#stalled = false; } } @@ -191,6 +211,19 @@ export class AudioRingBuffer { this.#writeIndex = Math.max(target, this.#readIndex); } + /** + * Hold playback until the ring holds the target again, keeping everything buffered. + * + * Used when the target deepens: `resize` alone only raises the bar a future refill has to clear, + * so a ring already playing keeps draining at its old depth and audio runs that much ahead of + * video. Parking the playhead spends exactly the deficit as silence and resumes on the same + * timeline, where `reset` would throw the buffer away and re-anchor. A ring that already holds + * the target has no deficit, and parking it would only wait on an insert that may never come. + */ + stall(): void { + if (this.length < this.#latencySamples) this.#stalled = true; + } + // Flush all buffered samples and re-stall, ready to anchor the next utterance. reset(): void { this.#readIndex = 0; @@ -204,7 +237,12 @@ export class AudioRingBuffer { if (this.#stalled) return 0; const samples = Math.min(this.#writeIndex - this.#readIndex, output[0].length); - if (samples === 0) return 0; + if (samples <= 0) { + // Ran dry mid-playback: re-stall so write() refills to the target before resuming. + this.#stalled = true; + this.#underruns++; + return 0; + } for (let channel = 0; channel < this.channels; channel++) { const dst = output[channel]; @@ -219,6 +257,12 @@ export class AudioRingBuffer { } this.#readIndex += samples; + + // A short quantum ends in silence, so it counts as an underrun even if a chunk lands before + // the next read. It does not park playback: the shortfall is under one quantum, and a + // refill would spend the whole target as silence to cover it. + if (samples < output[0].length) this.#underruns++; + return samples; } } diff --git a/js/watch/src/audio/shared-ring-buffer.test.ts b/js/watch/src/audio/shared-ring-buffer.test.ts index 2647ea3cc7..d7fd8a4f13 100644 --- a/js/watch/src/audio/shared-ring-buffer.test.ts +++ b/js/watch/src/audio/shared-ring-buffer.test.ts @@ -32,6 +32,24 @@ function insert( buffer.insert(Time.Micro.fromMilli(timestampMs as Time.Milli), data); } +/** + * Insert `samples` as a run of `chunk`-sized inserts starting at `timestampMs`. + * + * The skip band is one chunk wide, so a test that wants a skip has to arrive the way a decoder + * does: many small chunks, not one big one. The rate is 1000 in these tests, so a sample is a ms. + */ +function insertChunks( + buffer: SharedRingBuffer, + timestampMs: number, + samples: number, + chunk: number, + opts?: { channels?: number; value?: number }, +): void { + for (let i = 0; i < samples; i += chunk) { + insert(buffer, timestampMs + i, Math.min(chunk, samples - i), opts); + } +} + function read(buffer: SharedRingBuffer, samples: number, channelCount?: number): Float32Array[] { const ch = channelCount ?? buffer.channels; const output: Float32Array[] = []; @@ -52,7 +70,7 @@ describe("initialization", () => { expect(init.capacity).toBe(128); expect(init.rate).toBe(1000); expect(init.samples.byteLength).toBe(2 * 128 * 4); // 2 channels * 128 samples * Float32 - expect(init.control.byteLength).toBe(4 * 4); // 4 control slots * Int32 + expect(init.control.byteLength).toBe(6 * 4); // 6 control slots * Int32 expect(init.state.byteLength).toBe(8); // packed epoch + read cursor }); @@ -407,11 +425,11 @@ describe("overflow", () => { }); describe("latency skip", () => { - it("should skip READ when buffered exceeds LATENCY", () => { + it("should skip READ when buffered exceeds LATENCY plus a chunk", () => { const buffer = create({ rate: 1000, channels: 1, capacity: 100, latency: 20 }); - // Fill 60 samples — exceeds LATENCY of 20 - insert(buffer, 0, 60, { channels: 1, value: 1.0 }); + // Fill 60 samples in 10-sample chunks — a whole chunk past the 20 sample LATENCY. + insertChunks(buffer, 0, 60, 10, { channels: 1, value: 1.0 }); expect(buffer.stalled).toBe(false); // Read should skip ahead to maintain LATENCY distance from WRITE @@ -421,6 +439,18 @@ describe("latency skip", () => { expect(output[0].length).toBe(20); }); + it("should tolerate one chunk above LATENCY", () => { + // A ring sitting on the target is a chunk above it the moment the next chunk lands, so + // skipping on that overshoot would discard audio on every single insert. + const buffer = create({ rate: 1000, channels: 1, capacity: 100, latency: 20 }); + + insertChunks(buffer, 0, 40, 20, { channels: 1, value: 1.0 }); + expect(buffer.stalled).toBe(false); + + const output = read(buffer, 128, 1); + expect(output[0].length).toBe(40); + }); + it("should not skip when buffered is within LATENCY", () => { const buffer = create({ rate: 1000, channels: 1, capacity: 100, latency: 50 }); @@ -458,8 +488,8 @@ describe("setLatency", () => { it("should dynamically change latency affecting skip behavior", () => { const buffer = create({ rate: 1000, channels: 1, capacity: 100, latency: 50 }); - // Fill 80 samples - insert(buffer, 0, 80, { channels: 1, value: 1.0 }); + // Fill 80 samples, in chunks so the skip band stays narrow. + insertChunks(buffer, 0, 80, 10, { channels: 1, value: 1.0 }); expect(buffer.stalled).toBe(false); // With LATENCY=50, reading should skip to 30 (80-50) @@ -467,7 +497,7 @@ describe("setLatency", () => { expect(output1[0].length).toBe(50); // Write more - insert(buffer, 80, 80, { channels: 1, value: 2.0 }); + insertChunks(buffer, 80, 80, 10, { channels: 1, value: 2.0 }); // Change latency to 20 buffer.setLatency(20); @@ -478,6 +508,97 @@ describe("setLatency", () => { }); }); +describe("re-buffer", () => { + it("re-stalls and refills to the target after running dry", () => { + const buffer = create({ rate: 1000, channels: 1, capacity: 256, latency: 40 }); + + insertChunks(buffer, 0, 40, 20, { channels: 1, value: 1.0 }); + expect(buffer.stalled).toBe(false); + + // Drain it. + expect(read(buffer, 40, 1)[0].length).toBe(40); + expect(buffer.length).toBe(0); + + // The next read finds nothing and parks playback rather than handing the worklet a quantum + // of silence and trying again on the next one. + expect(read(buffer, 20, 1)[0].length).toBe(0); + expect(buffer.stalled).toBe(true); + expect(buffer.underruns).toBe(1); + + // A single chunk is not the target, so playback stays parked. + insert(buffer, 40, 20, { channels: 1, value: 2.0 }); + expect(buffer.stalled).toBe(true); + expect(read(buffer, 20, 1)[0].length).toBe(0); + + // Reaching the target resumes it, and nothing buffered in the meantime was lost. + insert(buffer, 60, 20, { channels: 1, value: 2.0 }); + expect(buffer.stalled).toBe(false); + expect(read(buffer, 40, 1)[0].length).toBe(40); + }); + + it("counts a quantum it could only partly fill as an underrun", () => { + const buffer = create({ rate: 1000, channels: 1, capacity: 256, latency: 40 }); + + insertChunks(buffer, 0, 40, 20, { channels: 1, value: 1.0 }); + expect(read(buffer, 30, 1)[0].length).toBe(30); + + // Ten samples remain against a quantum of twenty: the rest of the quantum is silence, which + // is an underrun whether or not a chunk lands before the next read. It is not a stall: a + // refill would cost the whole target to cover a gap shorter than one quantum. + expect(read(buffer, 20, 1)[0].length).toBe(10); + expect(buffer.stalled).toBe(false); + expect(buffer.underruns).toBe(1); + + // A chunk landing before the next read keeps playback going. + insert(buffer, 40, 20, { channels: 1, value: 2.0 }); + expect(read(buffer, 20, 1)[0].length).toBe(20); + expect(buffer.underruns).toBe(1); + }); + + it("stall() parks playback without discarding what is buffered", () => { + const buffer = create({ rate: 1000, channels: 1, capacity: 256, latency: 40 }); + + insertChunks(buffer, 0, 40, 20, { channels: 1, value: 1.0 }); + expect(read(buffer, 20, 1)[0].length).toBe(20); + + // The target deepens, so playback parks until the ring holds the new depth. + buffer.setLatency(60); + buffer.stall(); + expect(buffer.stalled).toBe(true); + expect(read(buffer, 20, 1)[0].length).toBe(0); + expect(buffer.length).toBe(20); // nothing was thrown away, unlike reset() + + insert(buffer, 40, 20, { channels: 1, value: 2.0 }); + expect(buffer.stalled).toBe(true); // 40 buffered, target is 60 + + insert(buffer, 60, 20, { channels: 1, value: 2.0 }); + expect(buffer.stalled).toBe(false); + + // Playback resumes on the same timeline: the samples buffered before the stall play first. + const output = read(buffer, 20, 1); + expect(output[0].length).toBe(20); + expect(output[0][0]).toBe(1.0); + expect(buffer.underruns).toBe(0); + }); + + it("stall() is a no-op once the ring already holds the target", () => { + const buffer = create({ rate: 1000, channels: 1, capacity: 256, latency: 40 }); + insertChunks(buffer, 0, 60, 20, { channels: 1, value: 1.0 }); + + buffer.setLatency(60); + buffer.stall(); + expect(buffer.stalled).toBe(false); + expect(read(buffer, 20, 1)[0].length).toBe(20); + }); + + it("does not count an underrun while parked", () => { + const buffer = create({ rate: 1000, channels: 1, capacity: 256, latency: 40 }); + read(buffer, 20, 1); + read(buffer, 20, 1); + expect(buffer.underruns).toBe(0); + }); +}); + describe("stalled getter", () => { it("should reflect STALLED flag", () => { const buffer = create({ rate: 1000, channels: 1, capacity: 100, latency: 50 }); diff --git a/js/watch/src/audio/shared-ring-buffer.ts b/js/watch/src/audio/shared-ring-buffer.ts index ce79094854..0972598446 100644 --- a/js/watch/src/audio/shared-ring-buffer.ts +++ b/js/watch/src/audio/shared-ring-buffer.ts @@ -3,10 +3,24 @@ import { Time } from "@moq/net"; // Control array slot indices. The playhead is not here: see `state`. const WRITE = 0; const LATENCY = 1; +/** + * Whether playback is held while the ring refills. The one slot both threads write. + * + * The writer clears it once the ring holds the target, the reader raises it when the ring runs + * dry. Neither needs a compare-exchange. The reader only raises it having observed an empty ring, + * which means it already drained everything the writer published, and the writer only clears it + * having published enough to cover the target. A raise landing after a clear costs one quantum of + * silence and the next insert undoes it, since every insert re-checks. A clear landing after a + * raise is correct on its face: the samples are there. + */ const STALLED = 2; // Timeline identity changes only on re-anchor, independently of the packed mutation epoch. const TIMELINE = 3; -const CONTROL_SLOTS = 4; +// The size of the most recent insert, i.e. one decoded chunk. See `read`. +const CHUNK = 4; +// How many times the reader ran dry mid-playback. Diagnostics only. +const UNDERRUN = 5; +const CONTROL_SLOTS = 6; /** * The playhead and its mutation epoch, packed into one 64-bit word: epoch in the high @@ -297,6 +311,15 @@ export class SharedRingBuffer { } } + // Publish the chunk size for the reader's skip band before the samples it describes become + // visible. The two stores are not one transaction, so the reader can land between them: + // this way it sees the wider band with the old cursor, which only makes it skip less, where + // the other order would have it measure a longer span against a narrower band and cut audio + // that is not actually late. The most recent insert, not a running maximum: one oversized + // decode would otherwise widen the band for the life of the ring, and a publisher that + // changes its frame duration is described within one insert. + Atomics.store(this.#control, CHUNK, originalLength); + // Advance WRITE (only forward) Atomics.store(this.#control, WRITE, i32Max(Atomics.load(this.#control, WRITE), end)); @@ -329,17 +352,29 @@ export class SharedRingBuffer { const write = Atomics.load(this.#control, WRITE); const latency = Atomics.load(this.#control, LATENCY); - // Latency skip: if buffered data exceeds LATENCY, skip ahead. + // Latency skip: skip ahead only once the ring holds a whole chunk more than the target, + // landing back on the target. Frames arrive one chunk at a time, so a ring sitting exactly + // on the target is a chunk above it the moment the next one lands; skipping on that + // overshoot discards audio on every single insert. The chunk of slack is the difference + // between tolerating normal arrival and cutting it. // CAS ensures we never step backward relative to a concurrent writer advance. // Disabled in buffered mode, where we deliberately play through the whole buffer. const buffered = (write - read) | 0; - if (!this.buffered && latency > 0 && buffered > latency) { + const slack = Atomics.load(this.#control, CHUNK); + if (!this.buffered && latency > 0 && buffered > ((latency + slack) | 0)) { const skipTo = (write - latency) | 0; if (((skipTo - read) | 0) > 0) read = skipTo; } const available = (write - read) | 0; const count = Math.min(available, output[0].length); + if (available <= 0) { + // Ran dry mid-playback. Re-stall so `insert` refills to the target before playback + // resumes: the alternative is playing the next chunk on an empty cushion, which + // underruns again on the very next quantum. See STALLED for the write discipline. + Atomics.store(this.#control, STALLED, 1); + Atomics.add(this.#control, UNDERRUN, 1); + } if (count <= 0) { // A latency skip still has to be published, and still only if nothing moved. if (((read - readOf(state)) | 0) > 0) { @@ -368,6 +403,11 @@ export class SharedRingBuffer { return 0; } + // A short quantum ends in silence, so it counts as an underrun even if a chunk lands before + // the next read. It does not park playback: the shortfall is under one quantum, and a + // refill would spend the whole target as silence to cover it. + if (count < output[0].length) Atomics.add(this.#control, UNDERRUN, 1); + return count; } @@ -404,6 +444,23 @@ export class SharedRingBuffer { this.#step(); } + /** + * Hold playback until the ring holds the target again, keeping everything buffered. + * Main thread only. + * + * Used when the target deepens: `setLatency` alone only raises the bar a future refill has to + * clear, so a ring already playing keeps draining at its old depth and audio runs that much + * ahead of video. Parking the playhead spends exactly the deficit as silence and resumes on the + * same timeline, where `reset` would throw the buffer away and re-anchor. A ring that already + * holds the target has no deficit, and parking it would only wait on an insert that may never + * come. Main thread only, like `insert`, so WRITE is exact and a concurrent read only shrinks + * what is buffered. + */ + stall(): void { + const buffered = (Atomics.load(this.#control, WRITE) - readOf(Atomics.load(this.#state, 0))) | 0; + if (buffered < Atomics.load(this.#control, LATENCY)) Atomics.store(this.#control, STALLED, 1); + } + /** * Flush buffered samples and re-stall, ready to anchor the next utterance (buffered mode). * Main thread only. The worklet reader sees STALLED and stops until the next insert. @@ -458,6 +515,8 @@ export class SharedRingBuffer { Atomics.store(dst.#control, WRITE, write); Atomics.store(dst.#control, LATENCY, latency); Atomics.store(dst.#control, STALLED, stalled); + Atomics.store(dst.#control, CHUNK, Atomics.load(this.#control, CHUNK)); + Atomics.store(dst.#control, UNDERRUN, Atomics.load(this.#control, UNDERRUN)); // Carry the unwrapped playhead over, rebased onto dst's READ. Fold the same `read` // snapshot the copy used so both sides agree on one observation; `copyStart` is at or @@ -498,6 +557,11 @@ export class SharedRingBuffer { return Atomics.load(this.#control, STALLED) === 1; } + /** How many times the reader has run dry mid-playback. */ + get underruns(): number { + return Atomics.load(this.#control, UNDERRUN); + } + /** * Number of buffered samples (WRITE - READ). * diff --git a/js/watch/src/audio/source.ts b/js/watch/src/audio/source.ts index 7961836df5..64ea421b21 100644 --- a/js/watch/src/audio/source.ts +++ b/js/watch/src/audio/source.ts @@ -1,8 +1,7 @@ import type * as Catalog from "@moq/hang/catalog"; -import type * as Moq from "@moq/net"; +import * as Moq from "@moq/net"; import { Effect, type Getter, getter, type Inputs, type Readonlys, readonlys, Signal } from "@moq/signals"; import type { Broadcast } from "../broadcast"; -import { playbackJitter } from "./config"; export type Target = { // Optional manual override for the selected rendition name. @@ -33,7 +32,7 @@ type SourceOutput = { track: Signal; config: Signal; - // The per-rendition jitter (ms) to add to the sync buffer. Wired into Sync by the parent. + // The jitter the selected rendition advertises (ms): a floor on the decoder's playout target. jitter: Signal; }; @@ -127,7 +126,8 @@ export class Source { effect.set(this.#out.track, selected.track); effect.set(this.#out.config, selected.config); - effect.set(this.#out.jitter, playbackJitter(selected.config)); + const jitter = selected.config.jitter; + effect.set(this.#out.jitter, jitter !== undefined ? Moq.Time.Milli(jitter) : undefined); } /** diff --git a/js/watch/src/element.ts b/js/watch/src/element.ts index f9d1eb2657..17aef0a2c9 100644 --- a/js/watch/src/element.ts +++ b/js/watch/src/element.ts @@ -457,7 +457,7 @@ export default class MoqWatch extends HTMLElement { /** * How far playback trails the live edge, in milliseconds. See {@link Delay}. * - * `"auto"` (the default) sizes the jitter buffer from the connection RTT. `"instant"` drops the + * `"auto"` (the default) sizes the jitter buffer from how late frames arrive. `"instant"` drops the * clock instead: video paints the moment it decodes and audio is disabled. */ get delay(): Delay { diff --git a/js/watch/src/player.ts b/js/watch/src/player.ts index 997f04c819..45979dfa7a 100644 --- a/js/watch/src/player.ts +++ b/js/watch/src/player.ts @@ -22,7 +22,7 @@ export type PlayerInput = { catalogFormat: Getter; /** Catalog supplied when the format is `manual`. */ catalog: Getter; - /** Connection probe used for rendition selection and delay. */ + /** Connection probe used for rendition selection. */ probe: Getter; /** Canvas to paint video into. */ canvas: Getter; @@ -117,7 +117,7 @@ export class Player { this.text = new Text.Source({ broadcast: this.broadcast, target: this.in.captions }); this.#signals.cleanup(() => this.text.close()); - this.sync = new Sync({ delay: this.in.delay, buffer: this.in.buffer, probe: this.in.probe }); + this.sync = new Sync({ delay: this.in.delay, buffer: this.in.buffer }); this.#signals.cleanup(() => this.sync.close()); this.video = new Video.Decoder({ source: videoSource, sync: this.sync, enabled: this.#videoEnabled }); diff --git a/js/watch/src/sync.test.ts b/js/watch/src/sync.test.ts index ed2411a8ec..892814b535 100644 --- a/js/watch/src/sync.test.ts +++ b/js/watch/src/sync.test.ts @@ -62,37 +62,60 @@ describe("delay and buffer", () => { expect(sync.out.maxAge.peek()).toBe(0 as Time.Milli); sync.close(); }); +}); - it("includes registered decoder jitter until the decoder unregisters", async () => { - const media = new Signal(20 as Time.Milli); - const sync = new Sync({ delay: 100 as Time.Milli }); - const unregister = sync.register(media); +describe("auto delay", () => { + it("holds nothing until a decoder registers", async () => { + const sync = new Sync(); + await flush(); + expect(sync.out.delay.peek()).toBe(0 as Time.Milli); + sync.close(); + }); + + it("follows the deepest registered target until it unregisters", async () => { + const audio = new Signal(120 as Time.Milli); + const video = new Signal(40 as Time.Milli); + const sync = new Sync(); + const unregisterAudio = sync.register(audio); + sync.register(video); await flush(); expect(sync.out.delay.peek()).toBe(120 as Time.Milli); + expect(sync.out.jitter.peek()).toBe(120 as Time.Milli); - media.set(80 as Time.Milli); + // A publisher flushing 250ms at once needs 250ms of buffer, whatever the round trip is. + audio.set(270 as Time.Milli); await flush(); - expect(sync.out.delay.peek()).toBe(180 as Time.Milli); + expect(sync.out.delay.peek()).toBe(270 as Time.Milli); - unregister(); + unregisterAudio(); await flush(); - expect(sync.out.delay.peek()).toBe(100 as Time.Milli); + expect(sync.out.delay.peek()).toBe(40 as Time.Milli); sync.close(); }); - it("unregisters duplicate jitter inputs independently", async () => { + it("unregisters duplicate targets independently", async () => { const media = new Signal(20 as Time.Milli); - const sync = new Sync({ delay: 100 as Time.Milli }); + const sync = new Sync(); const unregisterFirst = sync.register(media); const unregisterSecond = sync.register(media); unregisterFirst(); await flush(); - expect(sync.out.delay.peek()).toBe(120 as Time.Milli); + expect(sync.out.delay.peek()).toBe(20 as Time.Milli); unregisterSecond(); await flush(); + expect(sync.out.delay.peek()).toBe(0 as Time.Milli); + sync.close(); + }); + + it("takes a fixed delay literally, ignoring the measured targets", async () => { + const media = new Signal(250 as Time.Milli); + const sync = new Sync({ delay: 100 as Time.Milli }); + sync.register(media); + await flush(); expect(sync.out.delay.peek()).toBe(100 as Time.Milli); + expect(sync.out.jitter.peek()).toBe(100 as Time.Milli); sync.close(); }); }); diff --git a/js/watch/src/sync.ts b/js/watch/src/sync.ts index 07a5da3d12..e1dfe01bfc 100644 --- a/js/watch/src/sync.ts +++ b/js/watch/src/sync.ts @@ -14,7 +14,9 @@ import { /** * How far playback trails the live edge. * - * `"auto"` (the default) sizes the jitter buffer from the connection RTT; a `Time.Milli` fixes it. + * `"auto"` (the default) holds the largest playout target the registered decoders measure from how + * late their frames arrive (see doc/concept/audio-jitter.md). A `Time.Milli` fixes the delay at + * exactly that number and disables adaptation. * * `"instant"` drops the buffer and the pacing together: nothing is held, and {@link Sync.wait} * returns without sleeping, so a frame presents as soon as it exists. It also overrides @@ -24,9 +26,6 @@ import { */ export type Delay = "instant" | "auto" | Time.Milli; -const MIN_JITTER = Time.Milli(20); -const FALLBACK_JITTER = Time.Milli(100); - export type SyncInput = { /** How far playback trails the live edge. See {@link Delay}. */ delay: Getter; @@ -41,10 +40,7 @@ export type SyncInput = { */ buffer: Getter; - /** - * The connection's PROBE estimates, whose RTT drives "auto" jitter. Usually wired - * from a `Connection`'s `probe`. - */ + /** @internal */ probe: Getter; }; @@ -53,12 +49,11 @@ type SyncOutput = { // This will keep being updated as we catch up to the live playhead then will be relatively static. reference: Signal; - // The resolved delay from the live edge to the playhead: jitter + max(audio, video). + // The resolved delay from the live edge to the playhead: the configured number, or in "auto" the + // largest registered playout target. Zero for "instant". delay: Signal; - // The jitter component of `delay` (always numeric). - // In "auto" mode this is updated automatically from RTT. - // When the delay is a number, jitter equals that number. + // The jitter buffer, always equal to `delay`. jitter: Signal; // The media timestamp of the most recently received frame. @@ -80,7 +75,7 @@ export class Sync { readonly #out: SyncOutput = { reference: new Signal(undefined), delay: new Signal(Time.Milli.zero), - jitter: new Signal(FALLBACK_JITTER), + jitter: new Signal(Time.Milli.zero), timestamp: new Signal(undefined), buffered: new Signal(false), maxAge: new Signal(Time.Milli.zero), @@ -94,10 +89,7 @@ export class Sync { // Per-label late-frame tracking: accumulate count and max lateness, flush on recovery. #late = new Map(); - // Minimum RTT seen, used as the baseline for jitter calculation. - // Avoids inflating jitter due to bufferbloat. - #minRtt: number | undefined; - #media = new Signal<{ jitter: Getter }[]>([]); + #media = new Signal<{ target: Getter }[]>([]); #signals = new Effect(); @@ -110,14 +102,13 @@ export class Sync { this.#update = Promise.withResolvers(); - this.#signals.run(this.#runJitter.bind(this)); this.#signals.run(this.#runDelay.bind(this)); this.#signals.run(this.#runMaxAge.bind(this)); } - /** Include a decoder's rendition delay in the shared playback clock until disposed. */ - register(jitter: Getter): Dispose { - const registered = { jitter }; + /** Hold a decoder's "auto" playout target in the shared playback clock until disposed. */ + register(target: Getter): Dispose { + const registered = { target }; this.#media.update((media) => [...media, registered]); return () => this.#media.update((media) => media.filter((candidate) => candidate !== registered)); } @@ -132,51 +123,21 @@ export class Sync { this.#out.maxAge.set(Time.Milli.add(delay, buffer)); } - #runJitter(effect: Effect): void { - const delay = effect.get(this.in.delay); - - if (delay === "instant") { - // Holds nothing at all. - this.#minRtt = undefined; - this.#out.jitter.set(Time.Milli.zero); - return; - } - - if (typeof delay === "number") { - // Fixed mode: the configured delay is the jitter. - this.#minRtt = undefined; - this.#out.jitter.set(delay); - return; - } - - // "auto" mode: compute jitter from the connection's RTT estimate. - const rtt = effect.get(this.in.probe)?.rtt; - if (rtt !== undefined) { - // Track minimum RTT as baseline, ignoring bufferbloat. - this.#minRtt = this.#minRtt !== undefined ? Math.min(this.#minRtt, rtt) : rtt; - - // Buffer enough for a retransmit (1 RTT for ACK + retransmit). - const jitter = Time.Milli(Math.max(MIN_JITTER, this.#minRtt * 1.25)); - this.#out.jitter.set(jitter); - return; - } - - // No RTT available: fall back to static default. - this.#minRtt = undefined; - this.#out.jitter.set(FALLBACK_JITTER); - } - + // A fixed delay is taken literally and "auto" holds the deepest track's target, so every track + // plays from one clock. See doc/concept/audio-jitter.md for how a target is measured. #runDelay(effect: Effect): void { - const jitter = effect.get(this.#out.jitter); - let media = Time.Milli.zero; - for (const registered of effect.get(this.#media)) { - media = Time.Milli.max(media, effect.get(registered.jitter) ?? Time.Milli.zero); + const mode = effect.get(this.in.delay); + + let delay = Time.Milli.zero; + if (typeof mode === "number") { + delay = mode; + } else if (mode === "auto") { + for (const registered of effect.get(this.#media)) { + delay = Time.Milli.max(delay, effect.get(registered.target) ?? Time.Milli.zero); + } } - // A zero delay still holds the rendition's own delay, which is a frame interval at 60fps. - // "instant" holds nothing at all. - const instant = effect.get(this.in.delay) === "instant"; - const delay = instant ? Time.Milli.zero : Time.Milli.add(media, jitter); + this.#out.jitter.set(delay); this.#out.delay.set(delay); this.#update.resolve(); diff --git a/js/watch/src/ui/components/buffering-indicator.ts b/js/watch/src/ui/components/buffering-indicator.ts index 432b0e4b37..a55ab7bf25 100644 --- a/js/watch/src/ui/components/buffering-indicator.ts +++ b/js/watch/src/ui/components/buffering-indicator.ts @@ -9,7 +9,15 @@ export function bufferingIndicator(parent: Effect, watch: MoqWatch): HTMLElement container.appendChild(spinner); parent.run((effect) => { - const buffering = effect.get(watch.video.out.stalled); + // Audio re-buffers on its own once the ring runs dry, which is just as much a stall as a + // video one. Gate it on audio actually being downloaded: the ring reports stalled from + // construction and drains once the download stops, so a video-only broadcast or a muted + // player would otherwise show the spinner forever. "instant" never downloads audio at all. + const audio = + effect.get(watch.audio.in.enabled) && + effect.get(watch.audio.source.out.config) !== undefined && + effect.get(watch.controls.delay) !== "instant"; + const buffering = effect.get(watch.video.out.stalled) || (audio && effect.get(watch.audio.out.stalled)); const offline = effect.get(watch.broadcast.out.status) === "offline"; const unsupported = effect.get(watch.video.source.out.error) === "unsupported"; container.style.display = buffering && !offline && !unsupported ? "" : "none"; diff --git a/js/watch/src/ui/components/latency.ts b/js/watch/src/ui/components/latency.ts index 57428989c6..43d5c00fa4 100644 --- a/js/watch/src/ui/components/latency.ts +++ b/js/watch/src/ui/components/latency.ts @@ -63,7 +63,7 @@ export function latencyTab(parent: Effect, watch: MoqWatch): HTMLElement { const hint = DOM.create( "div", { className: "tab-hint" }, - "A larger delay smooths over network jitter at the cost of latency. Auto tracks the connection RTT. Drag the timeline to fine-tune.", + "A larger delay smooths over network jitter at the cost of latency. Auto follows how late frames actually arrive. Drag the timeline to fine-tune.", ); container.append(chips, timeline, readout, hint); diff --git a/js/watch/src/ui/stats.ts b/js/watch/src/ui/stats.ts index bbb18199ba..2e80e25f86 100644 --- a/js/watch/src/ui/stats.ts +++ b/js/watch/src/ui/stats.ts @@ -86,6 +86,7 @@ export function statsTab(parent: Effect, watch: MoqWatch): HTMLElement { const aRate2 = line(audioCard.grid, "Sample rate"); const aChannels = line(audioCard.grid, "Channels"); const aBitrate = line(audioCard.grid, "Bitrate"); + const aUnderruns = line(audioCard.grid, "Underruns"); track(parent, audioCard, { catalog: watch.audio.source.out.catalog, flag: watch.controls.muted, @@ -137,6 +138,10 @@ export function statsTab(parent: Effect, watch: MoqWatch): HTMLElement { aBitrate.textContent = aBitrate2 !== undefined ? formatBitrate(aBitrate2) : "—"; if (aStats) aPrev = { bytes: aStats.bytesReceived, when: now }; + // A non-zero count means the target is below what arrivals actually need. + const aUnder = watch.audio.out.underruns.peek(); + aUnderruns.textContent = watch.audio.out.stalled.peek() ? `${aUnder} (buffering)` : `${aUnder}`; + // Network. "Estimated max" is the congestion controller / PROBE estimate; // "Actual" is the goodput we measure from the video + audio byte counters. const probe = watch.connection.probe.peek(); diff --git a/js/watch/src/video/decoder.ts b/js/watch/src/video/decoder.ts index d8bab4511a..5a8907da43 100644 --- a/js/watch/src/video/decoder.ts +++ b/js/watch/src/video/decoder.ts @@ -93,6 +93,8 @@ export class Decoder { // The current track running, held so we can cancel it when the new track is ready. #active = new Signal(undefined); #pendingJitter = new Signal(undefined); + // The active track's arrival estimate from its container consumer. + #spread = new Signal(undefined); readonly #identity: Computed; #signals = new Effect(); @@ -112,7 +114,15 @@ export class Decoder { this.source = props.source; this.sync = props.sync; - this.#signals.cleanup(this.sync.register(this.out.jitter)); + // "auto" holds the arrival estimate, floored at the rendition's own delay. The audio frame + // term has no video counterpart: the rendition delay already covers a frame interval. + const target = this.#signals.computed((effect) => { + const jitter = effect.get(this.out.jitter); + const spread = effect.get(this.#spread); + if (spread === undefined) return jitter; + return Time.Milli.max(spread, jitter ?? Time.Milli.zero); + }); + this.#signals.cleanup(this.sync.register(target)); this.#identity = this.#signals.computed((effect) => { const config = effect.get(this.source.out.config); return config ? playbackIdentity(config) : undefined; @@ -198,6 +208,7 @@ export class Decoder { if (!active) { // Clear stale data when disabled (e.g. paused or not visible). this.#out.buffered.set([]); + this.#spread.set(undefined); return; } @@ -214,6 +225,7 @@ export class Decoder { }); effect.proxy(this.#out.timestamp, active.timestamp); effect.proxy(this.#out.buffered, active.buffered); + effect.proxy(this.#spread, active.spread); } #runDisplay(effect: Effect): void { @@ -285,6 +297,9 @@ class DecoderTrack { // Network jitter + decode buffer. buffered = new Signal([]); + // How late frames arrive relative to the earliest one, from the container consumer. + spread = new Signal(undefined); + // Decoded frames waiting to be rendered. #buffered = new Signal([]); @@ -400,6 +415,9 @@ class DecoderTrack { this.buffered.update(() => Container.mergeBufferedRanges(network, decode)); }); + // Publish the arrival estimate for the "auto" target. + effect.run((inner) => this.spread.set(inner.get(consumer.spread))); + decoder.configure({ codec: this.config.codec, description: this.config.description ? Util.Hex.toBytes(this.config.description) : undefined, @@ -476,6 +494,9 @@ class DecoderTrack { this.buffered.update(() => Container.mergeBufferedRanges(network, decode)); }); + // Publish the arrival estimate for the "auto" target. + effect.run((inner) => this.spread.set(inner.get(consumer.spread))); + // Configure decoder with description from catalog decoder.configure({ codec: this.config.codec, diff --git a/quest/m0/audio-jitter-target/README.md b/quest/m0/audio-jitter-target/README.md index b84e7ddfa6..5b68c70405 100644 --- a/quest/m0/audio-jitter-target/README.md +++ b/quest/m0/audio-jitter-target/README.md @@ -28,31 +28,10 @@ new `spread` observation. The algorithm is written down at `doc/concept/audio-jitter.md`, with a conformance corpus beside it that both implementations will read. -Neither `main` nor `dev` has a measured estimator. `js/watch/src/sync.ts:159` -still computes `max(MIN_JITTER, minRtt * 1.25)` from the connection's PROBE, -and `js/watch/src/audio/latency.ts` still exists. `sync.ts` also adds the -advertised jitter to that term, where the document settles on a maximum. - -The prior art is the branch of PR #3517, -`origin/quest/m0/3477-watch-auto-latency`, two commits ahead of `dev`. The PR -is closed and never merged; the watch quest starts from the branch rather than -from `dev`. It already deletes the RTT term -(`MIN_JITTER`, `FALLBACK_JITTER`, `#minRtt`, and the `probe` input are gone -from `sync.ts`; `latency.ts` survives, minus `reanchorFloor`) and plumbs a -per-track arrival `spread` through `Container.Consumer`, measured at container -frame arrival and before the age budget can skip a group, which is the right -observation point. Its estimator, `js/hang/src/container/jitter.ts`, is a -decaying histogram of 5 ms buckets, 200 of them so the percentile saturates at -one second, read at the 95th percentile plus one frame, with the arrival -minimum expiring over two 30 s windows, `reanchor()` on a discontinuity, and -the step down bounded to one frame per second. `jitter.test.ts` and -`js/watch/src/audio/replay.test.ts` cover it. What it gets wrong is the extra -frame, learned from the first gap between observed timestamps, and a rise that -is immediate and unclamped, so a tune-in across a stale group sets the target -to seconds. - -Note that `sync.ts` has since been refactored on `main` to a `register(jitter)` -list, so the branch does not rebase cleanly. +The browser implementation has landed on this line: `js/hang/src/container/jitter.ts` +observes each frame in `Container.Consumer` and passes the corpus, `js/watch` +composes each track's target and `Sync` holds the deepest one in `"auto"`. What +remains of the watch quest is proving it in a real browser. Native is done: `rs/moq-audio` estimates the target behind `decode::Options::delay` and `decode::Consumer::delay`, passes the corpus both @@ -63,7 +42,7 @@ the browser's `replay.test.ts` does. ## Quests -- [Watch](/quest/m0/audio-jitter-target/watch.md) - js/watch and js/hang bring the #3517 branch's estimator into conformance +- [Watch](/quest/m0/audio-jitter-target/watch.md) - the browser's measured target, proven on Chrome and Safari against the public relay ## Closes diff --git a/quest/m0/audio-jitter-target/watch.md b/quest/m0/audio-jitter-target/watch.md index 188de90dc9..b6d9989c5d 100644 --- a/quest/m0/audio-jitter-target/watch.md +++ b/quest/m0/audio-jitter-target/watch.md @@ -1,100 +1,42 @@ -# [M] js/watch: the playout target follows the spec, proven in a real browser +# [S] js/watch: prove the measured playout target in a real browser ## Goal -`js/watch` computes its auto target with the algorithm in -`doc/concept/audio-jitter.md` and passes the conformance corpus beside it. The -"Real-time" preset plays clean audio on a LAN and against the -public relay: zero underruns after convergence and no skip-aheads in steady -state, on both the isolated and the postMessage ring paths, confirmed by a -manual run on Chrome and Safari with a real microphone and 40 ms or more of -added RTT. +The "Real-time" preset plays clean audio on a LAN and against the public relay: +zero underruns after convergence and no skip-aheads in steady state, on both the +isolated and the postMessage ring paths, confirmed by a manual run on Chrome and +Safari with a real microphone and 40 ms or more of added RTT. -Boundaries: the ring's slack, re-stall and underrun counter are mechanical and -already correct on the #3517 branch; this quest changes the estimator above -them. +Boundaries: the estimator already conforms to `doc/concept/audio-jitter.md` and +passes the corpus in `js/hang/src/container/jitter.test.ts`. This quest measures +it, and fixes only what the measurement shows. ## Plan -Start from `origin/quest/m0/3477-watch-auto-latency`, the branch of the closed -PR #3517, two commits ahead of `dev`. Rebase it onto the base branch first, and -expect conflicts: `sync.ts` on `main` has since replaced the `audio`/`video` -inputs with a `register(jitter)` list. Nothing on `main` or `dev` has any of -it: `js/watch/src/sync.ts:159` still sizes auto from `minRtt * 1.25`. - -What the branch already has, and this quest keeps: - -- The RTT term is gone. `MIN_JITTER`, `FALLBACK_JITTER` and `#minRtt` are gone - from `sync.ts` and auto is the larger of the per-track spreads. - `js/watch/src/audio/latency.ts` survives, minus `reanchorFloor`; `ringSamples` - and `RENDER_QUANTUM` stay, and `RENDER_QUANTUM` belongs to the ring's slack - rather than to the target. -- The measurement point: `Container.Consumer` observes each frame at arrival, - before the age budget can skip a group, and exposes it as a `spread` getter. -- The estimator in `js/hang/src/container/jitter.ts`: a histogram of 5 ms - buckets, 200 of them, decayed per arrival by `0.9993`, read at the 95th - percentile and clamped to `BUCKETS * BUCKET` (one second) and to the largest - spread seen; the arrival minimum expires over two 30 s windows; `reanchor()` - drops the baseline on a discontinuity; the step down is bounded to one frame - per second. `jitter.test.ts` covers it. -- The ring work: one chunk of slack above the target, landing back on the - target when skipping, re-stalling on empty, and the underrun counter reaching - the stats panel and the buffering indicator. `replay.test.ts` drives both - rings from a recorded arrival trace. - -What remains is conformance to `doc/concept/audio-jitter.md`. The -document's estimator is a different enough shape that rewriting `jitter.ts` -against it is likely cheaper than patching the branch's: - -- The "plus one frame" term. `jitter.ts` learns it as the running minimum of - positive gaps between consecutive observed timestamps, so the first gap sets - it with nothing to validate against, `#publish` raises the target to - `percentile + step` immediately, and `reanchor()` clears `#latest` but not the - learned spacing. That is the 14.56 s reading, and it still reproduces on the - branch head. Write the regression first: two frames whose media timestamps are - far apart, then paced audio with zero real jitter, asserting the target stays - near the frame duration. Then take the frame duration from the codec, which - for Opus means parsing the TOC byte at the container layer, since `js/` has no - parser today and `js/watch/src/audio/config.ts` guesses 20 ms. -- Delay measured against the fastest frame in a 2 s media-time window, not - against a 30 s arrival minimum. Pruning by media time is what makes a - timestamp jump self-correcting, so `reanchor()` goes away. -- The histogram: 20 ms buckets, 100 of them, the underrun forget factor decayed - once per resampled 500 ms maximum rather than per arrival, the cold-start - prior, and the startup ramp. No rise or fall limiter; the document explains - why the branch needed one and this does not. -- Reordered arrivals excluded from the reference and the histogram. -- The advertised jitter floors the target instead of being added to it. - `sync.ts` sums them today, which over-buffers a bursty publisher by its own - flush span. -- Run the conformance corpus in `jitter.test.ts` alongside the existing cases. - `doc/concept/audio-jitter/*.json` is checked in and read directly. -- Check what the viewer's saved preset does on load. The element defaults - `delay` to `"auto"` and nothing in `demo/` overrides it, yet a fresh session - came up on the 100 ms chip, so something is restoring or overriding it. Pin - that down: a stored preference silently winning over the default is its own - bug, and it also means auto gets far less real exposure than it looks like. - 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 `replay.test.ts`. + 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. - Manual run against the public relay on Chrome and Safari, the two rows the - issue measured. 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. - -The branch also replaces `probe` in `SyncInput` with per-track `audioSpread` -and `videoSpread` inputs, which breaks the published `@moq/watch` type. This -quest lands on `main`, so it adds the spread inputs beside `probe` and stops -reading `probe`; removing it is part of the `SyncInput` reshape in -[Plan: A/V clock](/quest/m0/plan-av-clock.md). Land the estimator so -that quest can adopt it without a second estimator change. + 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. ## Related -- [Plan: A/V clock](/quest/m0/plan-av-clock.md) - reshapes `SyncInput` around the per-track spread this quest produces +- [Plan: A/V clock](/quest/m0/plan-av-clock.md) - reshapes `SyncInput` around the per-track target this line produces