From d3bde7f9b2ff2fdc24e017589dc0585ed80ddec4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 4 Oct 2026 20:42:34 -0700 Subject: [PATCH 1/4] quest: claim watch-delay-rebuild Co-Authored-By: Claude Opus 5.5 From 7279c5717164d7648f21c57a05996b8c1d790a31 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 4 Oct 2026 20:48:56 -0700 Subject: [PATCH 2/4] fix(watch): keep the audio decoder across a numeric delay change The audio decoder gated on `sync.in.delay === "instant"` by tracking the whole delay signal, so every numeric step (100 to 120 ms, or "auto" to a number) resubscribed, rebuilt the decoder, and truncated the ring on the handover: an audible gap. Sync now derives `out.instant`, and the decoder and Sync's own instant checks read that, so only a switch to or from "instant" reruns them. Numeric changes still reach the ring through `setLatency`. Co-Authored-By: Claude Opus 5.5 --- js/watch/src/audio/decoder.test.ts | 243 ++++++++++++++++++++++++++ js/watch/src/audio/decoder.ts | 4 +- js/watch/src/sync.test.ts | 1 + js/watch/src/sync.ts | 23 ++- quest/m0/README.md | 1 - quest/m0/audio-jitter-target/watch.md | 5 - quest/m0/watch-delay-rebuild.md | 37 ---- quest/m1/watch-audio-graph.md | 1 - 8 files changed, 262 insertions(+), 53 deletions(-) create mode 100644 js/watch/src/audio/decoder.test.ts delete mode 100644 quest/m0/watch-delay-rebuild.md diff --git a/js/watch/src/audio/decoder.test.ts b/js/watch/src/audio/decoder.test.ts new file mode 100644 index 0000000000..d36b33dbec --- /dev/null +++ b/js/watch/src/audio/decoder.test.ts @@ -0,0 +1,243 @@ +import { afterEach, beforeEach, describe, expect, it, mock, spyOn } from "bun:test"; +import * as Catalog from "@moq/hang/catalog"; +import * as Container from "@moq/hang/container"; +import * as Moq from "@moq/net"; +import { Time } from "@moq/net"; +import { Signal } from "@moq/signals"; +import type { Broadcast } from "../broadcast"; +import { type Delay, Sync } from "../sync"; +import { SharedRingBuffer } from "./shared-ring-buffer"; +import { Source } from "./source"; + +// Bun cannot load the blob-URL worklet import. +mock.module("./render-worklet.ts?worklet", () => ({ default: async () => "blob:fake-render" })); +const { Decoder } = await import("./decoder"); + +// Drain reactive work without advancing playback time. +async function microtasks() { + for (let i = 0; i < 100; i++) await Promise.resolve(); +} + +class FakeContext extends EventTarget { + readonly state = "running"; + readonly sampleRate: number; + readonly audioWorklet = { addModule: async () => {} }; + constructor(options: AudioContextOptions) { + super(); + this.sampleRate = options.sampleRate ?? 48_000; + } + resume = async () => {}; + close = async () => {}; +} + +class FakeWorklet { + readonly port = Object.assign(new EventTarget(), { postMessage() {}, start() {} }); + disconnect() {} +} + +class FakeData { + readonly sampleRate = 48_000; + readonly numberOfChannels = 2; + readonly numberOfFrames = 960; + readonly timestamp: number; + constructor(timestamp: number) { + this.timestamp = timestamp; + } + copyTo() {} + close() {} +} + +type Read = NonNullable>>; + +let frameTimestamp = 0; +function frame(): Read { + frameTimestamp += 20_000; + return { + group: 0, + discontinuity: 0, + continuous: true, + frame: { timestamp: Time.Micro(frameTimestamp), payload: new Uint8Array([1]), keyframe: true }, + }; +} + +const globals = ["AudioContext", "AudioWorkletNode", "AudioDecoder", "AudioEncoder", "EncodedAudioChunk"] as const; +const originals = new Map(); +let codecs = 0; + +beforeEach(() => { + frameTimestamp = 0; + codecs = 0; + for (const name of globals) originals.set(name, Object.getOwnPropertyDescriptor(globalThis, name)); + + class Codec { + state = "configured"; + readonly init: AudioDecoderInit; + constructor(init: AudioDecoderInit) { + this.init = init; + codecs++; + } + configure() {} + reset() {} + decode(chunk: { timestamp: number }) { + this.init.output(new FakeData(chunk.timestamp) as unknown as AudioData); + } + close() { + this.state = "closed"; + } + } + const chunk = class { + readonly timestamp: number; + constructor(init: EncodedAudioChunkInit) { + this.timestamp = init.timestamp; + } + }; + + const values = { + AudioContext: FakeContext, + AudioWorkletNode: FakeWorklet, + AudioDecoder: Codec, + AudioEncoder: class {}, + EncodedAudioChunk: chunk, + }; + for (const name of globals) Object.defineProperty(globalThis, name, { configurable: true, value: values[name] }); +}); + +afterEach(() => { + for (const name of globals) { + const original = originals.get(name); + if (original) Object.defineProperty(globalThis, name, original); + else Reflect.deleteProperty(globalThis, name); + } + mock.restore(); +}); + +// Hands frames to whichever container consumer is reading, and ends a consumer's reads on close. +function feed() { + const closed = new WeakSet(); + const queue: Read[] = []; + let waiter: { consumer: Container.Consumer; resolve: (read: Read | undefined) => void } | undefined; + + spyOn(Container.Consumer.prototype, "next").mockImplementation(function (this: Container.Consumer) { + if (closed.has(this)) return Promise.resolve(undefined); + const read = queue.shift(); + if (read) return Promise.resolve(read); + return new Promise((resolve) => { + waiter = { consumer: this, resolve }; + }); + }); + + const close = Container.Consumer.prototype.close; + spyOn(Container.Consumer.prototype, "close").mockImplementation(function (this: Container.Consumer) { + closed.add(this); + if (waiter?.consumer === this) { + waiter.resolve(undefined); + waiter = undefined; + } + close.call(this); + }); + + return (read: Read) => { + const current = waiter; + waiter = undefined; + if (current) current.resolve(read); + else queue.push(read); + }; +} + +async function play(initial: Delay) { + const push = feed(); + const truncate = spyOn(SharedRingBuffer.prototype, "truncate"); + const reset = spyOn(SharedRingBuffer.prototype, "reset"); + + const producer = new Moq.Broadcast.Producer(); + const consumer = producer.consume(); + const relativeBroadcast = mock(() => consumer); + const audio = Catalog.AudioConfigSchema.parse({ + codec: "opus", + container: { kind: "legacy" }, + sampleRate: 48_000, + numberOfChannels: 2, + }); + const catalog = new Signal({ audio: { renditions: { audio } } }); + const broadcast = new Signal({ + out: { catalog }, + relativeBroadcast, + } as unknown as Broadcast); + + const delay = new Signal(initial); + const source = new Source({ broadcast, supported: async () => true }); + const sync = new Sync({ delay }); + const decoder = new Decoder({ source, sync }); + await microtasks(); + + // Enough frames to get past the legacy decoder's warm-up, so a handover would truncate. + const play = async () => { + for (let i = 0; i < 6; i++) { + push(frame()); + await microtasks(); + } + }; + + return { + delay, + play, + // Each subscription resolves the rendition's broadcast once. + subscriptions: () => relativeBroadcast.mock.calls.length, + codecs: () => codecs, + truncates: () => truncate.mock.calls.length, + resets: () => reset.mock.calls.length, + close() { + decoder.close(); + sync.close(); + source.close(); + consumer.close(); + producer.close(); + }, + }; +} + +describe("Decoder across a delay change", () => { + for (const initial of [Time.Milli(100), "auto"] as const) { + it(`keeps the subscription and ring when ${initial} becomes a number`, async () => { + const playback = await play(initial); + try { + await playback.play(); + expect(playback.subscriptions()).toBe(1); + expect(playback.codecs()).toBe(1); + + playback.delay.set(Time.Milli(120)); + await microtasks(); + await playback.play(); + + expect(playback.subscriptions()).toBe(1); + expect(playback.codecs()).toBe(1); + expect(playback.truncates()).toBe(0); + expect(playback.resets()).toBe(0); + } finally { + playback.close(); + } + }); + } + + it("rebuilds on a switch to and from instant", async () => { + const playback = await play(Time.Milli(100)); + try { + await playback.play(); + + playback.delay.set("instant"); + await microtasks(); + expect(playback.resets()).toBe(1); + + playback.delay.set(Time.Milli(100)); + await microtasks(); + await playback.play(); + + expect(playback.subscriptions()).toBe(2); + expect(playback.codecs()).toBe(2); + // The replacement subscription drops whatever the ring still held from the first. + expect(playback.truncates()).toBe(1); + } finally { + playback.close(); + } + }); +}); diff --git a/js/watch/src/audio/decoder.ts b/js/watch/src/audio/decoder.ts index 5c86a1a250..11aad68392 100644 --- a/js/watch/src/audio/decoder.ts +++ b/js/watch/src/audio/decoder.ts @@ -221,7 +221,7 @@ export class Decoder { #runEnabled(effect: Effect): void { const enabled = effect.get(this.in.enabled); if (!enabled) return; - if (effect.get(this.sync.in.delay) === "instant") { + if (effect.get(this.sync.out.instant)) { this.reset(); return; } @@ -253,7 +253,7 @@ export class Decoder { #runDecoder(effect: Effect): void { const enabled = effect.get(this.in.enabled); if (!enabled) return; - if (effect.get(this.sync.in.delay) === "instant") return; + if (effect.get(this.sync.out.instant)) return; const broadcast = effect.get(this.source.in.broadcast); if (!broadcast) return; diff --git a/js/watch/src/sync.test.ts b/js/watch/src/sync.test.ts index f0c068ebdd..16953b0d86 100644 --- a/js/watch/src/sync.test.ts +++ b/js/watch/src/sync.test.ts @@ -60,6 +60,7 @@ describe("delay and buffer", () => { it("holds nothing when instant, whatever the buffer says", async () => { const sync = new Sync({ delay: "instant", buffer: 30_000 as Time.Milli }); await flush(); + expect(sync.out.instant.peek()).toBe(true); expect(sync.out.buffered.peek()).toBe(false); expect(sync.out.delay.peek()).toBe(0 as Time.Milli); expect(sync.out.maxAge.peek()).toBe(0 as Time.Milli); diff --git a/js/watch/src/sync.ts b/js/watch/src/sync.ts index c6d3404be7..2e7872bcbc 100644 --- a/js/watch/src/sync.ts +++ b/js/watch/src/sync.ts @@ -19,8 +19,7 @@ import { * `"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 * {@link SyncInput.buffer}, since holding a lookahead would contradict holding nothing. Audio is - * the owner's business: a ring with no depth underruns, so whoever wants this mode is expected to - * turn audio off. + * off: a ring with no depth underruns, so the audio decoder unsubscribes and flushes its ring. */ export type Delay = "instant" | "auto" | Time.Milli; @@ -49,6 +48,10 @@ export type SyncInput = { }; type SyncOutput = { + // Derived: whether the delay is "instant". Numeric and "auto" delays share `false`, so readers + // that only care about the mode don't rerun when the number changes. + instant: Signal; + // The earliest time we've received a frame, relative to its timestamp. // This will keep being updated as we catch up to the live playhead then will be relatively static. reference: Signal; @@ -78,6 +81,7 @@ export class Sync { readonly in: Readonlys; readonly #out: SyncOutput = { + instant: new Signal(false), reference: new Signal(undefined), delay: new Signal(Time.Milli.zero), jitter: new Signal(FALLBACK_JITTER), @@ -104,6 +108,7 @@ export class Sync { probe: getter(props?.probe), }; + this.#signals.run(this.#runInstant.bind(this)); this.#signals.run(this.#runJitter.bind(this)); this.#signals.run(this.#runDelay.bind(this)); this.#signals.run(this.#runMaxAge.bind(this)); @@ -116,11 +121,15 @@ export class Sync { return () => this.#media.update((media) => media.filter((candidate) => candidate !== registered)); } + #runInstant(effect: Effect): void { + this.#out.instant.set(effect.get(this.in.delay) === "instant"); + } + // Derive `buffered` / `maxAge` from the resolved delay and the configured lookahead. #runMaxAge(effect: Effect): void { const delay = effect.get(this.#out.delay); // "instant" holds nothing, so a configured lookahead doesn't apply. - const buffer = effect.get(this.in.delay) === "instant" ? Time.Milli.zero : effect.get(this.in.buffer); + const buffer = effect.get(this.#out.instant) ? Time.Milli.zero : effect.get(this.in.buffer); this.#out.buffered.set(buffer > 0); this.#out.maxAge.set(Time.Milli.add(delay, buffer)); @@ -169,7 +178,7 @@ export class Sync { // 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 instant = effect.get(this.#out.instant); const delay = instant ? Time.Milli.zero : Time.Milli.add(media, jitter); this.#out.delay.set(delay); } @@ -243,7 +252,7 @@ export class Sync { async wait(timestamp: Time.Milli): Promise { // A zero delay still sleeps: the sleep comes from the reference, which holds an early frame // until its timestamp comes up. "instant" is the only thing that skips the wait itself. - if (this.in.delay.peek() === "instant") return; + if (this.#out.instant.peek()) return; const reference = this.#out.reference.peek(); if (reference === undefined) { @@ -252,7 +261,7 @@ export class Sync { for (;;) { // Switching to "instant" wakes the sleep below, so frames parked here leave. - if (this.in.delay.peek() === "instant") return; + if (this.#out.instant.peek()) return; // Sleep until it's time to decode the next frame. // NOTE: This function runs in parallel for each frame. @@ -284,7 +293,7 @@ export class Sync { }; const timer = setTimeout(() => wake(true), ms); const disposes = [ - this.in.delay.changed(() => wake(false)), + this.#out.instant.changed(() => wake(false)), this.#out.delay.changed(() => wake(false)), this.#out.reference.changed(() => wake(false)), ]; diff --git a/quest/m0/README.md b/quest/m0/README.md index c708a364b4..688dba7cb3 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -56,7 +56,6 @@ a published `@moq/watch` break. - [Broadcast epochs](/quest/m0/broadcast-epoch/README.md) - every first-party publisher that can restart mints a fresh `@` epoch, viewers follow the newest live one, and bare names still resolve on every version - [Audio jitter target](/quest/m0/audio-jitter-target/README.md) - the audio playout target is a measured estimate of arrival timing in both languages, not a round-trip guess - [Opus DTX](/quest/m0/opus-dtx.md) - voice audio publishes without DTX, so its timeline follows the capture clock through silence -- [Delay rebuild](/quest/m0/watch-delay-rebuild.md) - a numeric audio delay change keeps the decoder and its ring ## Related diff --git a/quest/m0/audio-jitter-target/watch.md b/quest/m0/audio-jitter-target/watch.md index feafda41e7..c5c27846eb 100644 --- a/quest/m0/audio-jitter-target/watch.md +++ b/quest/m0/audio-jitter-target/watch.md @@ -104,8 +104,3 @@ 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/m1/av-clock.md). Land the estimator so that quest can adopt it without a second estimator change. - - -## Related - -- [Delay rebuild](/quest/m0/watch-delay-rebuild.md) - lands first, so preset steps stop rebuilding the decoder diff --git a/quest/m0/watch-delay-rebuild.md b/quest/m0/watch-delay-rebuild.md deleted file mode 100644 index 99dbf0de11..0000000000 --- a/quest/m0/watch-delay-rebuild.md +++ /dev/null @@ -1,37 +0,0 @@ -# [XS] A numeric delay change keeps the audio decoder - -## Goal - -Changing `@moq/watch`'s numeric audio delay (for example 100 to 120 ms) moves -the playout target without resubscribing, rebuilding the decoder, or -truncating the ring, so a delay step is never an audible gap. - -## Plan - -`js/watch/src/audio/decoder.ts` reads `effect.get(this.sync.in.delay) === -"instant"` in both `#runDecoder` and the worklet effect. The effect tracks the -whole signal, so every numeric change reruns both, and the rerun's handover -truncates the ring. `"auto"` never changes, so only numeric callers are hit. - -Decided (2026-10-04): `Sync` exposes `out.instant`, one owner for the -predicate, and the decoder and Sync's own instant checks read it. Signals -compare by value, so a numeric change no longer reruns the effects; numeric -changes already reach the ring through `#runLatency`'s `setLatency`. A later -`SyncInput` reshape in [A/V clock](/quest/m1/av-clock.md) may split the type, -which this does not block. Fix the stale `Delay` doc in -`js/watch/src/sync.ts`, which still says the owner turns audio off. It lands -ahead of the jitter target because that quest's preset testing steps the -delay and would count these gaps as underruns. - -Tests: a decoder fed frames keeps its subscription and ring contents across a -numeric change and across `"auto"` to a number, and still rebuilds on a -switch to or from `"instant"`. - -## Closes - -- [#4777](https://github.com/moq-dev/moq/issues/4777) - close this issue when the quest finishes - -## Related - -- [Watch jitter target](/quest/m0/audio-jitter-target/watch.md) - edits the same decoder; its delay presets hit this bug -- [Audio graph lifetime](/quest/m1/watch-audio-graph.md) - the other needless audio teardown in this file diff --git a/quest/m1/watch-audio-graph.md b/quest/m1/watch-audio-graph.md index 067f001d05..2b4e29d474 100644 --- a/quest/m1/watch-audio-graph.md +++ b/quest/m1/watch-audio-graph.md @@ -39,5 +39,4 @@ first frame after return. ## Related - [Enabled flag](/quest/m1/catalog-enabled.md) - a publisher mute disables the rendition instead of removing it -- [Delay rebuild](/quest/m0/watch-delay-rebuild.md) - the other needless audio teardown in this file - [Audio warmup](/quest/m1/audio-warmup.md) - the Opus pre-roll trim on the same path From aa4039141f66ebbab41055edaf878dc3b4381b84 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 4 Oct 2026 21:39:47 -0700 Subject: [PATCH 3/4] fix(watch): keep Sync.wait's instant check synchronous out.instant updates on a microtask, so a same-turn switch to "instant" (or a Sync constructed instant) left wait reading a stale predicate. Peek the input in wait instead; the decoder effects keep the deduplicated signal. Co-Authored-By: Claude Opus 5.5 --- js/watch/src/sync.test.ts | 6 ++++++ js/watch/src/sync.ts | 5 +++-- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/js/watch/src/sync.test.ts b/js/watch/src/sync.test.ts index 16953b0d86..c095fdcdc3 100644 --- a/js/watch/src/sync.test.ts +++ b/js/watch/src/sync.test.ts @@ -119,6 +119,12 @@ describe("delay and buffer", () => { }); describe("wait", () => { + it("returns at once when constructed instant, before effects flush", async () => { + const sync = new Sync({ delay: "instant" }); + await sync.wait(Time.Milli.zero); + sync.close(); + }); + it("wakes a sleeping wait when the delay switches to instant", async () => { const delay = new Signal(10_000 as Time.Milli); const sync = new Sync({ delay }); diff --git a/js/watch/src/sync.ts b/js/watch/src/sync.ts index 2e7872bcbc..e57c000540 100644 --- a/js/watch/src/sync.ts +++ b/js/watch/src/sync.ts @@ -252,7 +252,8 @@ export class Sync { async wait(timestamp: Time.Milli): Promise { // A zero delay still sleeps: the sleep comes from the reference, which holds an early frame // until its timestamp comes up. "instant" is the only thing that skips the wait itself. - if (this.#out.instant.peek()) return; + // Peek the input, not `out.instant`, which lags a same-turn change until effects flush. + if (this.in.delay.peek() === "instant") return; const reference = this.#out.reference.peek(); if (reference === undefined) { @@ -261,7 +262,7 @@ export class Sync { for (;;) { // Switching to "instant" wakes the sleep below, so frames parked here leave. - if (this.#out.instant.peek()) return; + if (this.in.delay.peek() === "instant") return; // Sleep until it's time to decode the next frame. // NOTE: This function runs in parallel for each frame. From 145ad2c63aec5ba7a9da050dbe440b3c0c84e026 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 09:26:51 -0700 Subject: [PATCH 4/4] fix(watch): wake Sync.wait on the same predicate it checks Co-Authored-By: Claude Opus 5.5 --- js/watch/src/sync.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/js/watch/src/sync.ts b/js/watch/src/sync.ts index e57c000540..c6eff3d25d 100644 --- a/js/watch/src/sync.ts +++ b/js/watch/src/sync.ts @@ -294,7 +294,7 @@ export class Sync { }; const timer = setTimeout(() => wake(true), ms); const disposes = [ - this.#out.instant.changed(() => wake(false)), + this.in.delay.changed(() => wake(false)), this.#out.delay.changed(() => wake(false)), this.#out.reference.changed(() => wake(false)), ];