diff --git a/doc/concept/audio-jitter.md b/doc/concept/audio-jitter.md index c613a91c54b..9c44549f6ef 100644 --- a/doc/concept/audio-jitter.md +++ b/doc/concept/audio-jitter.md @@ -383,10 +383,10 @@ them. The cost is over-buffering by the smallest subscribed `delay` when the broadcast's earliest rendition is not subscribed. The playout is recomputed when a rendition is subscribed, unsubscribed, or its -catalog entry rises, so dropping a slow rendition lowers latency. The shared -reference itself only ever moves earlier, so once the earliest subscribed track -leaves, the clock stays anchored to it until playback re-anchors; that errs on -the safe side. +catalog entry rises, so dropping a slow rendition lowers latency. Each track's +own arrivals only ever move its part of the shared reference earlier. A track +that leaves, or whose timeline restarts, releases its part: the earliest +remaining track's anchor takes over, or the next frame re-anchors. A track that replaces another in the same slot and on the same broadcast path, such as the new name a codec change publishes under, continues the outgoing diff --git a/doc/lib/js/watch.md b/doc/lib/js/watch.md index 12b7250b591..aba5df261ab 100644 --- a/doc/lib/js/watch.md +++ b/doc/lib/js/watch.md @@ -49,6 +49,14 @@ suspends the element until corrected or removed. The The overlay adds play/pause, volume, fullscreen, a quality selector, a buffering indicator, and a stats panel. +Audio, video, and captions play from one clock on local time, anchored at the +earliest arrival (with `buffer`, at the earliest retained frame once startup +fills). A congested track falls behind on its own instead of stalling the +others; in `auto` delay, its rising playout target still raises the delay +every track plays at. A discontinuity restarts only its own track's timeline, +and a track that is disabled, deselected, or fails releases its hold on the +clock. + `el.audio.out.error` and `el.video.out.error` report a fatal decode error. The failing track closes its subscription and stops decoding. Select another source or rendition to clear the failure; the player does not resubscribe to diff --git a/js/msf/src/catalog.test.ts b/js/msf/src/catalog.test.ts index 426192a9f52..d352acda37e 100644 --- a/js/msf/src/catalog.test.ts +++ b/js/msf/src/catalog.test.ts @@ -239,6 +239,21 @@ test("preserves delay through decode and encode", () => { expect(wireTracks[0].delay).toBe(200); }); +test("rejects a negative delay", () => { + const raw = encodeJson({ + version: "draft-01", + tracks: [{ name: "video0", packaging: "loc", isLive: true, role: "video", codec: "avc1.640028", delay: -1 }], + }); + expect(() => decode(raw)).toThrow(); + expect(() => + encode({ + tracks: [ + { name: "video0", packaging: "loc", isLive: true, role: "video", codec: "avc1.640028", delay: -1 }, + ], + }), + ).toThrow(); +}); + test("carries extension root members through decode and encode", () => { // An extension section (here `mpegts`) rides the catalog root untouched, and a // section this build has never heard of survives the same way. diff --git a/js/msf/src/catalog.ts b/js/msf/src/catalog.ts index bb9eb2ff6cb..dc79ac70bc3 100644 --- a/js/msf/src/catalog.ts +++ b/js/msf/src/catalog.ts @@ -56,7 +56,8 @@ const trackShape = { // Non-standard: how far this rendition trails the broadcast's earliest, in milliseconds. // Mirrors hang `delay`. A player holds `delay + jitter` and does not subtract across renditions. - delay: z.optional(z.number()), + // A rendition cannot trail the earliest by a negative amount, so that is malformed. + delay: z.optional(z.number().check(z.nonnegative())), }; /** Zod schema describing a single track entry in an MSF catalog. */ @@ -127,6 +128,9 @@ const WireCatalogSchema = z.object({ /** Serialize a catalog to its JSON wire representation (draft-01). */ export function encode(catalog: Catalog): Uint8Array { + // Refuse what `decode` would reject, so a publisher fails here instead of its subscribers. + CatalogSchema.parse(catalog); + // Hoist inline init payloads into a shared, deduplicated initDataList and // reference each from its track via initRef (the draft-01 wire shape). const initDataList: z.infer[] = []; diff --git a/js/watch/src/audio/decoder.test.ts b/js/watch/src/audio/decoder.test.ts index e6bbc94b16d..9f4646adef9 100644 --- a/js/watch/src/audio/decoder.test.ts +++ b/js/watch/src/audio/decoder.test.ts @@ -1302,7 +1302,7 @@ it("keeps audio's opening buffered while an enabled video track fills", async () const installed = clock.install(); const playback = await play(Milli(100), true); const video = new Signal<{ start: Milli; end: Milli } | null>(null); - const remove = playback.sync.register(new Signal(Milli(100)), video); + const remove = playback.sync.track({ target: new Signal(Milli(100)), startup: video }); try { await playback.play(8); await clock.advance(Nano.zero); @@ -1317,7 +1317,7 @@ it("keeps audio's opening buffered while an enabled video track fills", async () expect(ring.read([new Float32Array(960), new Float32Array(960)])).toBe(960); expect(ring.ring?.timestamp).toBe(Micro((opening ?? 0) + 20_000)); } finally { - remove(); + remove.close(); playback.close(); await microtasks(); installed.close(); @@ -1329,7 +1329,7 @@ it("starts above the decoded audio floor without expanding PCM storage", async ( const installed = clock.install(); const playback = await play(Milli(1000), true); const video = new Signal<{ start: Milli; end: Milli } | null>(null); - const remove = playback.sync.register(new Signal(Milli(1000)), video); + const remove = playback.sync.track({ target: new Signal(Milli(1000)), startup: video }); try { await playback.play(80); await clock.advance(Nano.zero); @@ -1346,7 +1346,7 @@ it("starts above the decoded audio floor without expanding PCM storage", async ( expect(ring.read([new Float32Array(960), new Float32Array(960)])).toBe(960); expect(ring.ring?.timestamp).toBe(Micro((opening ?? 0) + 20_000)); } finally { - remove(); + remove.close(); playback.close(); await microtasks(); installed.close(); @@ -1397,6 +1397,7 @@ for (const cmaf of [false, true]) { const installed = clock.install(); const playback = await play(Milli(100), true, cmaf); const reset = spyOn(playback.sync, "reset"); + const video = playback.sync.track({ target: new Signal(undefined) }); const prime = spyOn(AudioDecoder.prototype, "reset"); try { await playback.play(7); @@ -1408,7 +1409,7 @@ for (const cmaf of [false, true]) { const held = ring.length; const reference = playback.sync.out.reference.peek(); const resets = playback.resets(); - const parked = playback.sync.wait(Milli(1000)); + const parked = video.wait(Milli(1000)); expect(held).toBeGreaterThan(0); playback.push({ kind: "gap", timestamp: Micro(160_000), duration: Micro(20_000) }); await microtasks(); @@ -1438,7 +1439,7 @@ it.each([100, 1000])( async (delay) => { const playback = await play(Milli(delay), true); const video = new Signal<{ start: Milli; end: Milli } | null>(null); - const remove = playback.sync.register(new Signal(Milli(delay)), video); + const remove = playback.sync.track({ target: new Signal(Milli(delay)), startup: video }); const disconnect = spyOn(FakeWorklet.prototype, "disconnect"); const closeTrack = spyOn(Moq.Track.Subscriber.prototype, "close"); try { @@ -1457,7 +1458,7 @@ it.each([100, 1000])( await microtasks(); expect(playback.inserted()).toHaveLength(decoded); } finally { - remove(); + remove.close(); playback.close(); await microtasks(); disconnect.mockRestore(); @@ -1468,7 +1469,10 @@ it.each([100, 1000])( it("a rejected audio support probe does not leave buffered video waiting on a pending track", async () => { const sync = new Sync({ delay: Milli(100), buffer: Milli(1000) }); - const remove = sync.register(new Signal(Milli(100)), new Signal({ start: Milli(0), end: Milli(100) })); + const remove = sync.track({ + target: new Signal(Milli(100)), + startup: new Signal({ start: Milli(0), end: Milli(100) }), + }); let reject!: (error: Error) => void; const failure = new Error("audio support probe failed"); const logged = spyOn(console, "warn").mockImplementation(() => {}); @@ -1498,7 +1502,7 @@ it("a rejected audio support probe does not leave buffered video waiting on a pe const decoder = new Decoder({ source, sync }); try { await microtasks(); - sync.received(Milli(0)); + remove.received(Milli(0)); await microtasks(); expect(sync.playing.peek()).toBe(false); reject(failure); @@ -1510,7 +1514,7 @@ it("a rejected audio support probe does not leave buffered video waiting on a pe } finally { decoder.close(); source.close(); - remove(); + remove.close(); sync.close(); logged.mockRestore(); } @@ -1521,7 +1525,7 @@ it.each([false, true])("releases shared startup across a hole in queued audio (C const installed = clock.install(); const playback = await play(Milli(1500), true, cmaf); const video = new Signal<{ start: Milli; end: Milli } | null>({ start: Milli(20), end: Milli(1520) }); - const remove = playback.sync.register(new Signal(Milli(1500)), video); + const remove = playback.sync.track({ target: new Signal(Milli(1500)), startup: video }); try { await playback.play(25); // The audio after the hole counts, the hole itself does not. @@ -1547,7 +1551,7 @@ it.each([false, true])("releases shared startup across a hole in queued audio (C expect(playback.sync.playing.peek()).toBe(true); expect(ring.read([new Float32Array(960), new Float32Array(960)])).toBe(960); } finally { - remove(); + remove.close(); playback.close(); await microtasks(); installed.close(); diff --git a/js/watch/src/audio/decoder.ts b/js/watch/src/audio/decoder.ts index 0a5f972efa4..98eec787190 100644 --- a/js/watch/src/audio/decoder.ts +++ b/js/watch/src/audio/decoder.ts @@ -20,7 +20,7 @@ import { base64ToBytes } from "../base64"; import { arrivedAt, type Counter, type Key, type Received, Tally } from "../feedback"; import { type Failure, failureMoved, nextMedia, subscribeMedia } from "../media"; import { supportCacheKey } from "../support-probe"; -import type { Sync } from "../sync"; +import type { Sync, Track } from "../sync"; import { AudioBuffer } from "./buffer"; import { type DecoderConfig, @@ -127,6 +127,8 @@ export class Decoder { readonly source: Source; /** The clock that schedules playback. */ readonly sync: Sync; + // Audio's hold on the clock: its target, startup span, and arrivals. + readonly #track: Track; readonly #out: DecoderOutput = { error: new Signal(undefined), @@ -232,7 +234,8 @@ export class Decoder { const hold = Milli.fromMicro(ringHold(effect.get(this.sync.out.delay))); return { start: span.start, end: Milli(span.end + Math.min(queued, hold)) }; }); - this.#signals.cleanup(this.sync.register(playout, startup)); + this.#track = this.sync.track({ target: playout, startup }); + this.#signals.cleanup(() => this.#track.close()); this.#identity = this.#signals.computed((effect) => { const config = effect.get(this.source.out.config); return config ? playbackIdentity(config) : undefined; @@ -614,7 +617,7 @@ export class Decoder { // Classify ingress before decoder backpressure changes the delivery clock. const timestamp = Milli.fromMicro(frame.timestamp as Micro); - const late = this.sync.received(timestamp, frame.arrival); + const late = this.#track.received(timestamp, frame.arrival); this.#arrivals.write({ kind: "frame", timestamp, arrival: frame.arrival, late }); const duration = packetDuration(config.codec, frame); @@ -700,7 +703,7 @@ export class Decoder { const { frame } = next; const timestamp = Milli.fromMicro(frame.timestamp); - const late = this.sync.received(timestamp, frame.arrival); + const late = this.#track.received(timestamp, frame.arrival); this.#arrivals.write({ kind: "frame", timestamp, arrival: frame.arrival, late }); const duration = packetDuration(config.codec, frame); @@ -841,6 +844,8 @@ export class Decoder { #retire(): void { this.#consumer = undefined; this.#measured.set(undefined); + // Its arrivals stop holding the clock's reference. + this.#track.reset(); } // The ring, once the worklet has loaded. Decoding waits for it, so the frames that arrive first diff --git a/js/watch/src/audio/replay.ts b/js/watch/src/audio/replay.ts index 9e8f366ab34..d8a6766cfc9 100644 --- a/js/watch/src/audio/replay.ts +++ b/js/watch/src/audio/replay.ts @@ -150,7 +150,8 @@ export async function* replay(trace: Arrival[], options: Options): AsyncGenerato delay: config.delay !== undefined ? Milli(config.delay) : undefined, }), ); - signals.cleanup(sync.register(playout)); + const audio = sync.track({ target: playout }); + signals.cleanup(() => audio.close()); await settle(); let delay = sync.out.delay.peek(); @@ -190,7 +191,7 @@ export async function* replay(trace: Arrival[], options: Options): AsyncGenerato const { timestamp, arrival } = next.frame; read.set(timestamp); const media = Milli.fromMicro(timestamp); - const late = sync.received(media, arrival); + const late = audio.received(media, arrival); arrivals.push({ kind: "frame", timestamp: media, arrival, late }); const entry = recorded.get(timestamp); if (!entry) throw new Error(`the consumer delivered an unrecorded frame at ${timestamp} us`); diff --git a/js/watch/src/index.ts b/js/watch/src/index.ts index c3b1679de15..4ee7d5cc566 100644 --- a/js/watch/src/index.ts +++ b/js/watch/src/index.ts @@ -22,7 +22,7 @@ export { assets } from "./assets"; export * as Audio from "./audio"; export * from "./broadcast"; export * from "./player"; -export * from "./sync"; +export { type Delay, Sync, type SyncInput, type SyncOutput } from "./sync"; /** Caption and subtitle playback. */ export * as Text from "./text"; /** Video rendition selection, decoding, and rendering. */ diff --git a/js/watch/src/msf.test.ts b/js/watch/src/msf.test.ts index 3303c7564d1..97c797be4c5 100644 --- a/js/watch/src/msf.test.ts +++ b/js/watch/src/msf.test.ts @@ -50,7 +50,7 @@ test.each(["mp4a.40.2", "flac", "future-audio"])("keeps MSF decoder warmup absen expect(audio?.warmup).toBeUndefined(); }); -test("rounds a fractional MSF delay up, and drops zero", () => { +test("rounds a fractional MSF delay up, drops zero, and rejects a negative one", () => { const catalog: Msf.Catalog = { tracks: [ { @@ -81,6 +81,14 @@ test("rounds a fractional MSF delay up, and drops zero", () => { expect(hang.video?.renditions.video?.delay).toBe(u53(201)); expect(hang.audio?.renditions.audio?.delay).toBe(u53(1)); expect(hang.audio?.renditions.early?.delay).toBeUndefined(); + + // A rendition cannot trail the earliest by a negative amount, even between -1 and 0. + for (const delay of [-1, -0.5]) { + const negative: Msf.Catalog = { + tracks: [{ name: "audio", packaging: "loc", role: "audio", codec: "opus", delay }], + }; + expect(() => toHang(negative)).toThrow(); + } }); test("rounds a fractional MSF jitter up, keeps zero, and rejects a negative one", () => { diff --git a/js/watch/src/msf.ts b/js/watch/src/msf.ts index 9d7219cdc4c..d87c0611099 100644 --- a/js/watch/src/msf.ts +++ b/js/watch/src/msf.ts @@ -49,10 +49,11 @@ function toContainer(track: Msf.Track): ContainerInfo | undefined { } // Hang stores delay as whole milliseconds rounded up, and zero means the rendition is not behind. -// MSF writes a fractional millisecond, which u53 rejects. +// MSF writes a fractional millisecond, which u53 rejects. A negative value goes to u53 unrounded, so +// it is rejected like a negative jitter rather than read as absent. function delayMillis(value: number): ReturnType | undefined { - if (value <= 0) return undefined; - return u53(Math.ceil(value)); + if (value === 0) return undefined; + return u53(value < 0 ? value : Math.ceil(value)); } // Hang stores jitter as whole milliseconds; round MSF's fractional value up. A negative value goes diff --git a/js/watch/src/retention.test.ts b/js/watch/src/retention.test.ts index 1a0b823030e..85a3d072822 100644 --- a/js/watch/src/retention.test.ts +++ b/js/watch/src/retention.test.ts @@ -59,14 +59,15 @@ async function pendingListeners(run: () => Promise): Promise { // piles it up for the life of the player. test("waits on a stable clock leave nothing behind", async () => { const sync = new Sync({ delay: Milli(10) }); + const video = sync.track({ target: new Signal(undefined) }); // Let the delay effect run before anchoring. await yieldNow(); - sync.received(Nano.toMilli(Nano(now()))); + video.received(Nano.toMilli(Nano(now()))); const wait = async () => { for (let round = 0; round < 10; round++) { const at = Nano.toMilli(Nano(now())); - await Promise.all(Array.from({ length: 100 }, () => sync.wait(at))); + await Promise.all(Array.from({ length: 100 }, () => video.wait(at))); } }; expect(await pendingReactions(wait)).toBe(0); @@ -87,6 +88,7 @@ test("a long subscription through the player path leaves nothing behind", async const producer = new Container.Legacy.Producer(track, format); const sync = new Sync({ delay: Milli(10) }); + const audio = sync.track({ target: new Signal(undefined) }); const effect = new Effect(); const sub = subscribeMedia(effect, { broadcast: broadcast.consume(), @@ -111,8 +113,8 @@ test("a long subscription through the player path leaves nothing behind", async if (!next) throw new Error("the track ended"); const timestamp = Milli.fromMicro(next.frame.timestamp); - sync.received(timestamp); - const presented = effect.race(sync.wait(timestamp)); + audio.received(timestamp); + const presented = effect.race(audio.wait(timestamp)); presenting.add(presented); void presented.finally(() => presenting.delete(presented)); } diff --git a/js/watch/src/sync.test.ts b/js/watch/src/sync.test.ts index 5dc3869b326..2ba372cb264 100644 --- a/js/watch/src/sync.test.ts +++ b/js/watch/src/sync.test.ts @@ -77,8 +77,8 @@ describe("auto delay", () => { const audio = new Signal(120 as Milli); const video = new Signal(40 as Milli); const sync = new Sync(); - const unregisterAudio = sync.register(audio); - sync.register(video); + const unregisterAudio = sync.track({ target: audio }); + sync.track({ target: video }); await flush(); expect(sync.out.delay.peek()).toBe(120 as Milli); @@ -87,7 +87,7 @@ describe("auto delay", () => { await flush(); expect(sync.out.delay.peek()).toBe(270 as Milli); - unregisterAudio(); + unregisterAudio.close(); await flush(); expect(sync.out.delay.peek()).toBe(40 as Milli); sync.close(); @@ -96,14 +96,14 @@ describe("auto delay", () => { it("unregisters duplicate targets independently", async () => { const media = new Signal(20 as Milli); const sync = new Sync(); - const unregisterFirst = sync.register(media); - const unregisterSecond = sync.register(media); + const unregisterFirst = sync.track({ target: media }); + const unregisterSecond = sync.track({ target: media }); - unregisterFirst(); + unregisterFirst.close(); await flush(); expect(sync.out.delay.peek()).toBe(20 as Milli); - unregisterSecond(); + unregisterSecond.close(); await flush(); expect(sync.out.delay.peek()).toBe(0 as Milli); sync.close(); @@ -112,7 +112,7 @@ describe("auto delay", () => { it("takes a fixed delay literally, ignoring the measured targets", async () => { const media = new Signal(250 as Milli); const sync = new Sync({ delay: 100 as Milli }); - sync.register(media); + sync.track({ target: media }); await flush(); expect(sync.out.delay.peek()).toBe(100 as Milli); sync.close(); @@ -133,24 +133,27 @@ describe("wait", () => { it("presents at once when constructed instant, before effects flush", async () => { const sync = new Sync({ delay: "instant" }); - sync.received(nowMs()); - expect(await sync.wait(Milli(nowMs() + 10_000))).toBe(true); + const video = sync.track({ target: new Signal(undefined) }); + video.received(nowMs()); + expect(await video.wait(Milli(nowMs() + 10_000))).toBe(true); sync.close(); }); it("drops an unanchored frame", async () => { const sync = new Sync({ delay: "instant" }); - expect(await sync.wait(nowMs())).toBe(false); + const video = sync.track({ target: new Signal(undefined) }); + expect(await video.wait(nowMs())).toBe(false); sync.close(); }); it("presents a frame once its deadline comes due", async () => { const sync = new Sync({ delay: 100 as Milli }); + const video = sync.track({ target: new Signal(undefined) }); await flush(); const timestamp = nowMs(); - sync.received(timestamp); + video.received(timestamp); - const waiting = sync.wait(timestamp); + const waiting = video.wait(timestamp); await flush(); expect(Bun.peek.status(waiting)).toBe("pending"); @@ -162,10 +165,11 @@ describe("wait", () => { it("wakes a sleeping wait to present when the delay switches to instant", async () => { const delay = new Signal(10_000 as Milli); const sync = new Sync({ delay }); + const video = sync.track({ target: new Signal(undefined) }); await flush(); - sync.received(nowMs()); + video.received(nowMs()); - const waiting = sync.wait(nowMs()); + const waiting = video.wait(nowMs()); await flush(); expect(Bun.peek.status(waiting)).toBe("pending"); @@ -176,45 +180,48 @@ describe("wait", () => { it("drops a frame a skip passed over, before and after it parks", async () => { const sync = new Sync({ delay: 100 as Milli }); + const video = sync.track({ target: new Signal(undefined) }); await flush(); const timestamp = nowMs(); - sync.received(timestamp); + video.received(timestamp); - const parked = sync.wait(timestamp); + const parked = video.wait(timestamp); await flush(); expect(Bun.peek.status(parked)).toBe("pending"); // A frame 10s newer, with no buffer, skips the playhead past the parked one. - sync.received(Milli(timestamp + 10_000)); + video.received(Milli(timestamp + 10_000)); expect(await parked).toBe(false); - expect(await sync.wait(timestamp)).toBe(false); + expect(await video.wait(timestamp)).toBe(false); sync.close(); }); it("drops a sleeping wait at a reset, even when the next anchor restores the reference", async () => { const sync = new Sync({ delay: 10_000 as Milli }); + const video = sync.track({ target: new Signal(undefined) }); await flush(); const timestamp = nowMs(); - sync.received(timestamp); + video.received(timestamp); - const waiting = sync.wait(timestamp); + const waiting = video.wait(timestamp); await flush(); // In one turn, so the reference never visibly changes. sync.reset(); - sync.received(timestamp); + video.received(timestamp); expect(await waiting).toBe(false); sync.close(); }); it("drops a sleeping wait when its signal aborts", async () => { const sync = new Sync({ delay: 10_000 as Milli }); + const video = sync.track({ target: new Signal(undefined) }); await flush(); const timestamp = nowMs(); - sync.received(timestamp); + video.received(timestamp); const controller = new AbortController(); - const waiting = sync.wait(timestamp, controller.signal); + const waiting = video.wait(timestamp, controller.signal); await flush(); controller.abort(); expect(await waiting).toBe(false); @@ -223,11 +230,12 @@ describe("wait", () => { it("drops a sleeping wait when closed", async () => { const sync = new Sync({ delay: 10_000 as Milli }); + const video = sync.track({ target: new Signal(undefined) }); await flush(); const timestamp = nowMs(); - sync.received(timestamp); + video.received(timestamp); - const waiting = sync.wait(timestamp); + const waiting = video.wait(timestamp); await flush(); sync.close(); expect(await waiting).toBe(false); @@ -240,12 +248,13 @@ describe("recorded ingress", () => { clock.jump(Nano.fromMilli(Milli(10_000))); const installed = clock.install(); const sync = new Sync({ delay: Milli(100) }); + const video = sync.track({ target: new Signal(undefined) }); try { for (let i = 0; i < 20; i++) await Promise.resolve(); - expect(sync.received(Milli(0), Milli(1000))).toBe(false); + expect(video.received(Milli(0), Milli(1000))).toBe(false); expect(sync.out.reference.peek()).toBe(1000); - expect(sync.received(Milli(20), Milli(1020))).toBe(false); - expect(sync.received(Milli(40), Milli(1200))).toBe(true); + expect(video.received(Milli(20), Milli(1020))).toBe(false); + expect(video.received(Milli(40), Milli(1200))).toBe(true); } finally { installed.close(); sync.close(); @@ -266,63 +275,67 @@ describe("skip floor", () => { it("a skip passes over only frames behind the playhead it jumped to", async () => { const sync = new Sync({ delay: Milli(100) }); + const video = sync.track({ target: new Signal(undefined) }); await flush(); - sync.received(Milli(1000), Milli(5000)); - expect(sync.presentable(Milli(0))).toBe(true); + video.received(Milli(1000), Milli(5000)); + expect(video.presentable(Milli(0))).toBe(true); // A startup burst with no buffer: audio 50ms newer skips the playhead to 950ms. - sync.received(Milli(1050), Milli(5000)); - expect(sync.presentable(Milli(1000))).toBe(true); - expect(sync.presentable(Milli(949))).toBe(false); + video.received(Milli(1050), Milli(5000)); + expect(video.presentable(Milli(1000))).toBe(true); + expect(video.presentable(Milli(949))).toBe(false); // A large skip passes over the earlier frame. - sync.received(Milli(10_000), Milli(5000)); - expect(sync.presentable(Milli(1000))).toBe(false); - expect(sync.presentable(Milli(9900))).toBe(true); + video.received(Milli(10_000), Milli(5000)); + expect(video.presentable(Milli(1000))).toBe(false); + expect(video.presentable(Milli(9900))).toBe(true); sync.close(); }); it("a skip never lowers the floor and a reset clears it", async () => { const delay = new Signal(Milli(100)); const sync = new Sync({ delay }); + const video = sync.track({ target: new Signal(undefined) }); await flush(); - sync.received(Milli(0), Milli(5000)); - sync.received(Milli(10_000), Milli(5000)); - expect(sync.presentable(Milli(9000))).toBe(false); + video.received(Milli(0), Milli(5000)); + video.received(Milli(10_000), Milli(5000)); + expect(video.presentable(Milli(9000))).toBe(false); // A larger cap re-anchors to a lower playhead, but cannot revive what was passed over. delay.set(Milli(5000)); await flush(); - sync.received(Milli(11_000), Milli(5000)); - expect(sync.presentable(Milli(9000))).toBe(false); + video.received(Milli(11_000), Milli(5000)); + expect(video.presentable(Milli(9000))).toBe(false); // A new timeline may start its timestamps over, once it is anchored. sync.reset(); - expect(sync.presentable(Milli(9000))).toBe(false); - sync.received(Milli(0), Milli(5000)); - expect(sync.presentable(Milli(9000))).toBe(true); + expect(video.presentable(Milli(9000))).toBe(false); + video.received(Milli(0), Milli(5000)); + expect(video.presentable(Milli(9000))).toBe(true); sync.close(); }); it("a skip counts how long the skipping frame waited after ingress", async () => { const sync = new Sync({ delay: Milli(100) }); + const video = sync.track({ target: new Signal(undefined) }); await flush(); - sync.received(Milli(0), Milli(5000)); + video.received(Milli(0), Milli(5000)); // Reordering held this audio for 200ms, so the new reference puts the playhead at 10.1s now. clock.jump(Nano.fromMilli(Milli(200))); - sync.received(Milli(10_000), Milli(5000)); - expect(sync.presentable(Milli(10_050))).toBe(false); - expect(sync.presentable(Milli(10_100))).toBe(true); + video.received(Milli(10_000), Milli(5000)); + expect(video.presentable(Milli(10_050))).toBe(false); + expect(video.presentable(Milli(10_100))).toBe(true); sync.close(); }); it("instant skips pass over nothing", async () => { const sync = new Sync({ delay: "instant" }); + const video = sync.track({ target: new Signal(undefined) }); await flush(); - sync.received(Milli(0), Milli(5000)); - sync.received(Milli(10_000), Milli(5000)); - expect(sync.presentable(Milli(0))).toBe(true); + video.received(Milli(0), Milli(5000)); + video.received(Milli(10_000), Milli(5000)); + expect(video.presentable(Milli(0))).toBe(true); sync.close(); }); }); @@ -337,14 +350,17 @@ describe("buffered startup", () => { audio: new Signal<{ start: Milli; end: Milli } | null>(null), video: new Signal<{ start: Milli; end: Milli } | null>(null), }; - const dispose = Object.values(spans).map((span) => sync.register(new Signal(Milli(100)), span)); + const tracks = Object.values(spans).map((startup) => + sync.track({ target: new Signal(Milli(100)), startup }), + ); + const [video] = tracks; try { await clock.advance(Nano.zero); - sync.received(Milli(500)); + video.received(Milli(500)); spans[first].set({ start: Milli(500), end: Milli(600) }); await clock.advance(Nano.zero); let delivered = false; - const waiting = sync.wait(Milli(500)).then((ok) => { + const waiting = video.wait(Milli(500)).then((ok) => { delivered = ok; }); await clock.advance(Nano.fromMilli(Milli(1000))); @@ -359,7 +375,7 @@ describe("buffered startup", () => { expect(delivered).toBe(true); expect(sync.now()).toBe(Milli(500)); } finally { - for (const fn of dispose) fn(); + for (const track of tracks) track.close(); sync.close(); installed.close(); } @@ -369,24 +385,24 @@ describe("buffered startup", () => { const clock = new Manual(); const installed = clock.install(); const sync = new Sync({ delay: Milli(100), buffer: Milli(1000) }); - const dispose = sync.register(new Signal(Milli(100)), new Signal(null)); + const video = sync.track({ target: new Signal(Milli(100)), startup: new Signal(null) }); try { await clock.advance(Nano.zero); - sync.received(Milli(0)); + video.received(Milli(0)); await clock.advance(Nano.zero); const abort = new AbortController(); - const cancelled = sync.wait(Milli(0), abort.signal); + const cancelled = video.wait(Milli(0), abort.signal); abort.abort(); expect(await cancelled).toBe(false); - const reset = sync.wait(Milli(0)); + const reset = video.wait(Milli(0)); sync.reset(); - sync.received(Milli(0)); + video.received(Milli(0)); expect(await reset).toBe(false); - const closed = sync.wait(Milli(0)); + const closed = video.wait(Milli(0)); sync.close(); expect(await closed).toBe(false); } finally { - dispose(); + video.close(); sync.close(); installed.close(); } @@ -397,29 +413,182 @@ describe("buffered startup", () => { const buffer = new Signal(Milli(1000)); const sync = new Sync({ delay: Milli(100), buffer }); const absent = new Signal<{ start: Milli; end: Milli } | null | undefined>(null); - const remove = sync.register(new Signal(Milli(100)), absent); - const video = sync.register(new Signal(Milli(100)), new Signal({ start: Milli(0), end: Milli(100) })); + const remove = sync.track({ target: new Signal(Milli(100)), startup: absent }); + const video = sync.track({ + target: new Signal(Milli(100)), + startup: new Signal({ start: Milli(0), end: Milli(100) }), + }); try { await clock.advance(Nano.zero); - sync.received(Milli(0)); + video.received(Milli(0)); await clock.advance(Nano.zero); expect(sync.playing.peek()).toBe(false); absent.set(undefined); await clock.advance(Nano.zero); - expect(await sync.wait(Milli(0))).toBe(true); + expect(await video.wait(Milli(0))).toBe(true); buffer.set(Milli.zero); sync.reset(); absent.set(null); await clock.advance(Nano.zero); - sync.received(Milli(0)); - const waiting = sync.wait(Milli(0)); + video.received(Milli(0)); + const waiting = video.wait(Milli(0)); await clock.advance(Nano.fromMilli(Milli(100))); expect(await waiting).toBe(true); } finally { - remove(); - video(); + remove.close(); + video.close(); sync.close(); installed.close(); } }); }); + +describe("tracks", () => { + // Local time is pinned, so every playhead and deadline below is exact. + let clock: Manual; + let installed: Installed; + beforeEach(() => { + clock = new Manual(); + clock.jump(Nano.fromMilli(Milli(5000))); + installed = clock.install(); + }); + afterEach(() => installed.close()); + + const target = () => new Signal(undefined); + + it("a video-only discontinuity leaves audio's timeline alone", async () => { + const sync = new Sync({ delay: Milli(100) }); + const audio = sync.track({ target: target() }); + const video = sync.track({ target: target() }); + await flush(); + // Video arrived earliest and anchors the clock. + audio.received(Milli(1000), Milli(5000)); + video.received(Milli(1010), Milli(5000)); + expect(sync.now()).toBe(Milli(910)); + + // Video's parked frame belongs to the timeline that ended; audio's arrivals take the reference + // and its parked frame stays on its own timeline. + const parked = video.wait(Milli(950)); + const playing = audio.wait(Milli(950)); + video.reset(); + expect(await parked).toBe(false); + expect(sync.now()).toBe(Milli(900)); + await clock.advance(Nano.fromMilli(Milli(50))); + expect(await playing).toBe(true); + + // An audio break likewise leaves video's parked frames alone. + video.received(Milli(1100), Milli(5050)); + const picture = video.wait(Milli(1040)); + audio.reset(); + await clock.advance(Nano.fromMilli(Milli(50))); + expect(await picture).toBe(true); + sync.close(); + }); + + it("releases the reference of the earliest track when it leaves", async () => { + const sync = new Sync({ delay: Milli(100) }); + const audio = sync.track({ target: target() }); + const video = sync.track({ target: target() }); + await flush(); + // Video trails audio by 300ms, so audio anchors the clock and video is late against it. + audio.received(Milli(1000), Milli(5000)); + expect(video.received(Milli(1000), Milli(5300))).toBe(true); + + // Once audio stops arriving, video plays against its own arrivals instead. + audio.reset(); + expect(sync.out.reference.peek()).toBe(4300); + expect(video.received(Milli(1033), Milli(5333))).toBe(false); + + // Closing the last track leaves nothing anchored. + video.close(); + expect(sync.now()).toBeUndefined(); + sync.close(); + }); + + it("forgets a departed track's skip once the reference moves back to the survivor's", async () => { + const sync = new Sync({ delay: Milli(100) }); + const audio = sync.track({ target: target() }); + const video = sync.track({ target: target() }); + await flush(); + // Audio's burst skips the playhead to 1900; video's own arrivals put it at 900. + video.received(Milli(1000), Milli(5000)); + audio.received(Milli(2000), Milli(5000)); + expect(video.presentable(Milli(1100))).toBe(false); + + audio.reset(); + expect(sync.now()).toBe(Milli(900)); + const waiting = video.wait(Milli(1100)); + await clock.advance(Nano.fromMilli(Milli(200))); + expect(await waiting).toBe(true); + sync.close(); + }); + + it("keeps the survivor's own skip when the reference moves back to it", async () => { + const sync = new Sync({ delay: Milli(100) }); + const audio = sync.track({ target: target() }); + const video = sync.track({ target: target() }); + await flush(); + // Video skips its own playhead from 900 to 1400, then audio's burst skips it to 2900. + video.received(Milli(1000), Milli(5000)); + video.received(Milli(1500), Milli(5000)); + audio.received(Milli(3000), Milli(5000)); + + audio.reset(); + expect(sync.now()).toBe(Milli(1400)); + expect(video.presentable(Milli(1300))).toBe(false); + expect(video.presentable(Milli(1450))).toBe(true); + sync.close(); + }); + + it("keeps the playhead where buffered startup put it when the owner leaves", async () => { + const sync = new Sync({ delay: Milli(100), buffer: Milli(1000) }); + const audioSpan = new Signal<{ start: Milli; end: Milli } | null>(null); + const videoSpan = new Signal<{ start: Milli; end: Milli } | null>(null); + const audio = sync.track({ target: target(), startup: audioSpan }); + const video = sync.track({ target: target(), startup: videoSpan }); + await flush(); + audio.received(Milli(500), Milli(5000)); + video.received(Milli(500), Milli(5000)); + + // Startup releases a second later, anchoring at the retained opening rather than its arrival. + await clock.advance(Nano.fromMilli(Milli(1000))); + audioSpan.set({ start: Milli(500), end: Milli(700) }); + videoSpan.set({ start: Milli(500), end: Milli(700) }); + await flush(); + expect(sync.now()).toBe(Milli(500)); + + audio.reset(); + expect(sync.now()).toBe(Milli(500)); + sync.close(); + }); + + it("drops every track's parked frames at a reset", async () => { + const sync = new Sync({ delay: Milli(100) }); + const audio = sync.track({ target: target() }); + const video = sync.track({ target: target() }); + await flush(); + audio.received(Milli(1000), Milli(5000)); + const sound = audio.wait(Milli(1500)); + const picture = video.wait(Milli(1500)); + sync.reset(); + expect(await sound).toBe(false); + expect(await picture).toBe(false); + expect(sync.now()).toBeUndefined(); + + // The next frame anchors a new timeline, ignoring the anchors before the reset. + video.received(Milli(3000), Milli(5000)); + expect(sync.now()).toBe(Milli(2900)); + sync.close(); + }); + + it("reads zero until media time zero's turn", async () => { + const sync = new Sync({ delay: Milli(100) }); + const video = sync.track({ target: target() }); + await flush(); + video.received(Milli(0), Milli(5000)); + expect(sync.now()).toBe(Milli.zero); + await clock.advance(Nano.fromMilli(Milli(110))); + expect(sync.now()).toBe(Milli(10)); + sync.close(); + }); +}); diff --git a/js/watch/src/sync.ts b/js/watch/src/sync.ts index 7abc4f18fbe..06398a92528 100644 --- a/js/watch/src/sync.ts +++ b/js/watch/src/sync.ts @@ -1,14 +1,5 @@ import { Milli, Nano, now, sleep } from "@moq/clock"; -import { - type Dispose, - Effect, - type Getter, - getter, - type Inputs, - type Readonlys, - readonlys, - Signal, -} from "@moq/signals"; +import { Effect, type Getter, getter, type Inputs, type Readonlys, readonlys, Signal } from "@moq/signals"; /** * How far playback trails the live edge. @@ -51,6 +42,7 @@ export type SyncOutput = { // The earliest time we've received a frame, relative to its timestamp: arrival minus timestamp // in milliseconds, an offset that is negative when the timestamps run ahead of the local clock. // This will keep being updated as we catch up to the live playhead then will be relatively static. + // Each track holds its own part of it; see `Track.reset()`. reference: Signal; // The resolved delay from the live edge to the playhead: the configured number, or in "auto" the @@ -70,12 +62,34 @@ export type SyncOutput = { maxDelay: Signal; }; -/** The playback clock: decides when each frame plays so audio and video stay together at the configured delay. */ -export class Sync { - /** The wired inputs, read-only. */ - readonly in: Readonlys; +/** A decoded span from the opening of a track's timeline. @internal */ +export type Span = { start: Milli; end: Milli }; - readonly #out: SyncOutput = { +/** What a track holds in the shared clock. @internal */ +export type TrackProps = { + /** The track's "auto" playout target, unset until it has one. */ + target: Getter; + /** For buffered startup: undefined is inactive, null is selected but has no playable span yet. */ + startup?: Getter; +}; + +/** What one track holds in the clock. @internal */ +export type Part = { + props: TrackProps; + // Bumped whenever this track's timeline ends, so its parked frames drop. + epoch: Signal; + // The reference this track's own arrivals hold on its current timeline. + anchor?: number; + // The furthest playhead this track's own skips jumped to, restored when it takes the reference. + floor?: number; +}; + +const wall = (): Milli => Nano.toMilli(Nano(now())); + +/** The state a {@link Sync} and its track handles share. @internal */ +export class Clock { + readonly in: Readonlys; + readonly out: SyncOutput = { instant: new Signal(false), reference: new Signal(undefined), delay: new Signal(Milli.zero), @@ -83,251 +97,184 @@ export class Sync { buffered: new Signal(false), maxDelay: new Signal(Milli.zero), }; - /** The derived outputs, read-only. */ - readonly out = readonlys(this.#out); - - #media = new Signal< - { - target: Getter; - // undefined is inactive, null is selected but has no playable first span yet. - startup?: Getter<{ start: Milli; end: Milli } | null | undefined>; - }[] - >([]); - #started = false; - #playing = new Signal(true); - /** Whether buffered startup has released the media consumers. @internal */ - readonly playing: Getter = this.#playing; + + readonly parts = new Signal([]); + readonly playing = new Signal(true); + started = false; // Bumped by every `reset()`. A frame that waited across a bump belongs to the previous timeline, // even when the next frame restores the same reference value. - #epoch = new Signal(0); + readonly epoch = new Signal(0); // The furthest playhead a skip has jumped to on this timeline. A frame below it was passed over. - #floor: number | undefined; + floor: number | undefined; // Releases parked waits on `close()`. - #closed = new Signal(false); + readonly closed = new Signal(false); - #signals = new Effect(); + // The track whose arrivals set the reference, which releases it when its timeline ends. + owner?: Part; constructor(props?: Inputs) { this.in = { delay: getter(props?.delay ?? ("auto" as Delay)), buffer: getter(props?.buffer ?? Milli.zero), }; - - this.#signals.run(this.#runInstant.bind(this)); - this.#signals.run(this.#runDelay.bind(this)); - this.#signals.run(this.#runMaxDelay.bind(this)); - this.#signals.run(this.#runStartup.bind(this)); } - /** - * Hold a decoder's "auto" playout target in the shared playback clock until disposed. - * @internal - */ - register( - target: Getter, - startup?: Getter<{ start: Milli; end: Milli } | null | undefined>, - ): Dispose { - const registered = { target, startup }; - this.#media.update((media) => [...media, registered]); - return () => this.#media.update((media) => media.filter((candidate) => candidate !== registered)); + // Fold an arrival into a reference: keep it while the frame adds lookahead within `maxDelay`, and + // lower it (skip ahead) only as far as that budget requires. + fold(current: number | undefined, ref: number): number { + if (current === undefined || ref >= current) return current ?? ref; + const sleep = current - ref + this.out.delay.peek(); + const cap = this.out.maxDelay.peek(); + if (sleep <= cap) return current; + return ref + cap - this.out.delay.peek(); } - // Startup is data-driven: time passing cannot fill a missing track. The first picture still - // travels through the ordinary renderer while the consumers wait for this shared gate. - #runStartup(effect: Effect): void { - effect.get(this.#epoch); - const buffered = effect.get(this.#out.buffered); - const delay = effect.get(this.#out.delay); - const reference = effect.get(this.#out.reference); - const spans = effect.get(this.#media).flatMap(({ startup }) => { - const span = startup && effect.get(startup); - return span === undefined ? [] : [span]; - }); - if (!buffered || spans.length === 0) { - this.#playing.set(true); - return; - } - if (this.#started) return; - const ready = spans.filter((span) => span !== null); - if ( - reference === undefined || - ready.length !== spans.length || - ready.some((span) => span.end - span.start < delay) - ) { - this.#playing.set(false); - return; - } - const first = Math.min(...ready.map((span) => span.start)); - this.#started = true; - // Anchor at the retained opening, not at its arrival: startup may have waited on another track. - this.#out.reference.set(Nano.toMilli(Nano(now())) - first - delay); - this.#playing.set(true); - } - - #runInstant(effect: Effect): void { - this.#out.instant.set(effect.get(this.in.delay) === "instant"); - } - - // Derive `buffered` / `maxDelay` from the resolved delay and the configured lookahead. - #runMaxDelay(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.#out.instant) ? Milli.zero : effect.get(this.in.buffer); - - this.#out.buffered.set(buffer > 0); - this.#out.maxDelay.set(Milli(delay + buffer)); - } - - // 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 mode = effect.get(this.in.delay); - - let delay = Milli.zero; - if (typeof mode === "number") { - delay = mode; - } else if (mode === "auto") { - for (const registered of effect.get(this.#media)) { - delay = Milli(Math.max(delay, effect.get(registered.target) ?? Milli.zero)); - } - } - - this.#out.delay.set(delay); - } - - /** - * Anchor playback from ingress time and report whether the frame arrived late. - * @internal - */ - // Fold a newly received frame into the reference. The reference anchors playback to the - // wall clock; we lower it (skip ahead) only when keeping it would push the lookahead past `maxDelay`. - received(timestamp: Milli, arrival = Nano.toMilli(Nano(now()))): boolean { - this.#out.timestamp.update((current) => (current === undefined || timestamp > current ? timestamp : current)); + received(part: Part, timestamp: Milli, arrival: Milli): boolean { + this.out.timestamp.update((current) => (current === undefined || timestamp > current ? timestamp : current)); const ref = arrival - timestamp; - const currentRef = this.#out.reference.peek(); + const current = this.out.reference.peek(); + part.anchor = this.fold(part.anchor, ref); // First frame anchors the reference. - if (currentRef === undefined) { - this.#out.reference.set(ref); + if (current === undefined) { + this.out.reference.set(ref); + this.owner = part; return false; } // Retain the opening while another track fills; the ordinary max-delay bounds still own // storage, but a later arrival must not advance the shared playhead before startup. - if (!this.#playing.peek()) return false; + if (!this.playing.peek()) return false; // Check if `wait()` would not sleep at all. // NOTE: We check here instead of in `wait()` so we can identify when frames are received late. // Otherwise, chained `wait()` calls would cause a false-positive during CPU starvation. - const delay = this.#out.delay.peek(); - const sleep = currentRef - ref + delay; - const late = sleep < 0; + const delay = this.out.delay.peek(); + const late = current - ref + delay < 0; - // Frame isn't earlier than the anchor: it can't add lookahead, so keep the reference. - if (ref >= currentRef) return late; - - // Frame is earlier (more lookahead). `sleep` is how far ahead of the playhead keeping the - // anchor would put it. - const cap = this.#out.maxDelay.peek(); - if (sleep <= cap) return late; // within budget: let the buffer grow instead of skipping ahead - - // Over the cap: re-anchor down so the resulting lookahead is exactly the cap. - const reference = ref + cap - delay; - this.#out.reference.set(reference); + const reference = this.fold(current, ref); + if (reference === current) return late; + this.out.reference.set(reference); + this.owner = part; // The playhead now sits at `timestamp - cap`, plus however long this frame waited between // ingress and here. "instant" has no cap, so every newer frame would pass over whatever is // still decoding; it presents frames as they exist instead. if (this.in.delay.peek() !== "instant") { - const floor = Nano.toMilli(Nano(now())) - reference - delay; - if (this.#floor === undefined || floor > this.#floor) this.#floor = floor; + const floor = wall() - reference - delay; + if (this.floor === undefined || floor > this.floor) this.floor = floor; + if (part.floor === undefined || floor > part.floor) part.floor = floor; } return late; } - /** - * Re-anchor playback to the next frame received. Call this at an utterance boundary - * in buffered mode (typically alongside flushing the audio buffer) so the new content - * plays from its own first frame instead of inheriting the previous reference. - */ + // End one track's timeline: its parked frames drop and it releases its part of the reference. + end(part: Part): void { + part.epoch.update((epoch) => epoch + 1); + part.anchor = undefined; + part.floor = undefined; + if (this.owner !== part) return; + + // The earliest remaining track holds the reference, else the next frame anchors a new one. + let earliest: Part | undefined; + for (const other of this.parts.peek()) { + if (other.anchor === undefined) continue; + if (earliest?.anchor === undefined || other.anchor < earliest.anchor) earliest = other; + } + this.owner = earliest; + if (earliest?.anchor === undefined) { + this.unanchor(); + return; + } + const current = this.out.reference.peek(); + const reference = earliest.anchor; + this.out.reference.set(reference); + + // The floor marked frames a skip on the departed track's reference passed over. Moving back + // to the survivor's puts its frames ahead of the playhead again, except those its own skips + // passed over; moving forward is a skip. + this.floor = earliest.floor; + if (current !== undefined && reference < current && this.in.delay.peek() !== "instant") { + this.floor = wall() - reference - this.out.delay.peek(); + } + } + + // Clear the reference so the next frame anchors a new timeline. Every parked frame belongs to + // the old one. + unanchor(): void { + this.epoch.update((epoch) => epoch + 1); + this.floor = undefined; + this.started = false; + this.owner = undefined; + this.playing.set(!this.out.buffered.peek()); + this.out.reference.set(undefined); + } + reset(): void { - this.#epoch.update((epoch) => epoch + 1); - this.#floor = undefined; - this.#started = false; - this.#playing.set(!this.#out.buffered.peek()); - this.#out.reference.set(undefined); + for (const part of this.parts.peek()) { + part.anchor = undefined; + part.floor = undefined; + } + this.unanchor(); } - /** - * Whether a frame at `timestamp` is anchored on the current timeline and ahead of any skip, so - * it may still present. Advisory, for showing a frame early; {@link Sync.wait} decides on its own. - * @internal - */ presentable(timestamp: Milli): boolean { - return this.#out.reference.peek() !== undefined && !this.#skipped(timestamp); + return this.out.reference.peek() !== undefined && !this.skipped(timestamp); } // Whether a playhead skip on this timeline passed over `timestamp`. - #skipped(timestamp: Milli): boolean { - return this.#floor !== undefined && timestamp < this.#floor; + skipped(timestamp: Milli): boolean { + return this.floor !== undefined && timestamp < this.floor; } - /** - * The PTS that should be rendering right now, derived from the reference + delay; negative - * before the first frame's turn. Undefined before ingress or while buffered startup is filling. - */ now(): Milli | undefined { - if (!this.#playing.peek()) return undefined; - const reference = this.#out.reference.peek(); + if (!this.playing.peek()) return undefined; + const reference = this.out.reference.peek(); if (reference === undefined) return undefined; - return Milli(Nano.toMilli(Nano(now())) - reference - this.#out.delay.peek()); + // Media time is nonnegative, so a playhead before the first frame's turn reads as zero. + return Milli(Math.max(0, wall() - reference - this.out.delay.peek())); } - /** - * Sleep until it's time to render this frame, then resolve whether to present it. - * - * `true` presents the frame now. `false` drops it: it was unanchored, a skip passed over it, a - * `reset()` ended its timeline, or `close()` or `signal` released the wait. - * @internal - */ - async wait(timestamp: Milli, signal?: AbortSignal): Promise { - const epoch = this.#epoch.peek(); + async wait(part: Part, timestamp: Milli, signal?: AbortSignal): Promise { + const epoch = this.epoch.peek(); + const own = part.epoch.peek(); for (;;) { // Closing or aborting wakes the sleep below, so frames parked here leave. - if (this.#closed.peek() || signal?.aborted) return false; + if (this.closed.peek() || signal?.aborted) return false; // A reset clears the reference, so an unanchored frame predates the current timeline. - const reference = this.#out.reference.peek(); - if (reference === undefined || this.#epoch.peek() !== epoch || this.#skipped(timestamp)) return false; + const reference = this.out.reference.peek(); + if (reference === undefined || this.epoch.peek() !== epoch || this.skipped(timestamp)) return false; + if (part.epoch.peek() !== own) return false; // 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. // Peek the input, not `out.instant`, which lags a same-turn change until effects flush. if (this.in.delay.peek() === "instant") return true; - if (!this.#playing.peek()) { - await this.#sleep(undefined, signal); + if (!this.playing.peek()) { + await this.sleep(part, undefined, signal); continue; } // NOTE: This function runs in parallel for each frame. - const sleep = reference + timestamp - Nano.toMilli(Nano(now())) + this.#out.delay.peek(); + const sleep = reference + timestamp - wall() + this.out.delay.peek(); // Skip the timer for small sleeps; the host's timer resolution (~4ms) would overshoot. if (sleep < 5) return true; - await this.#sleep(sleep, signal); + await this.sleep(part, sleep, signal); } } // Sleeps for `ms`, or wakes early once anything the sleep was computed from changes or `signal` // aborts. Releases every listener either way: frames sleep once each, so a listener left on a // signal that never changes would pile up for the life of the player. - #sleep(ms: number | undefined, signal?: AbortSignal): Promise { + sleep(part: Part, ms: number | undefined, signal?: AbortSignal): Promise { return new Promise((resolve) => { const timer = new AbortController(); const wake = () => { @@ -338,11 +285,12 @@ export class Sync { if (ms !== undefined) void sleep(Nano.fromMilli(Milli(ms)), { signal: timer.signal }).then(wake, () => {}); const disposes = [ this.in.delay.changed(wake), - this.#out.delay.changed(wake), - this.#out.reference.changed(wake), - this.#epoch.changed(wake), - this.#closed.changed(wake), - this.#playing.changed(wake), + this.out.delay.changed(wake), + this.out.reference.changed(wake), + this.epoch.changed(wake), + this.closed.changed(wake), + this.playing.changed(wake), + part.epoch.changed(wake), ]; if (signal) { signal.addEventListener("abort", wake, { once: true }); @@ -350,10 +298,189 @@ export class Sync { } }); } +} + +/** + * One track's hold on the shared clock: its playout target and startup span, its arrivals, and the + * waits that pace its frames. Close it when the track goes away. + * @internal + */ +export class Track { + readonly #clock: Clock; + readonly #part: Part; + + /** @internal */ + constructor(clock: Clock, props: TrackProps) { + this.#clock = clock; + this.#part = { props, epoch: new Signal(0) }; + clock.parts.update((parts) => [...parts, this.#part]); + } + + /** Fold a received frame into the clock and report whether it arrived too late for its turn. */ + received(timestamp: Milli, arrival = wall()): boolean { + return this.#clock.received(this.#part, timestamp, arrival); + } + + /** + * Whether a frame at `timestamp` is anchored on the current timeline and ahead of any skip, so it + * may still present. Advisory, for showing a frame early; {@link Track.wait} decides on its own. + */ + presentable(timestamp: Milli): boolean { + return this.#clock.presentable(timestamp); + } + + /** + * Sleep until it's time to render this frame, then resolve whether to present it. + * + * `true` presents the frame now. `false` drops it: it was unanchored, a skip passed over it, a + * reset ended its timeline, or `close()` or `signal` released the wait. + */ + wait(timestamp: Milli, signal?: AbortSignal): Promise { + return this.#clock.wait(this.#part, timestamp, signal); + } + + /** + * End this track's timeline, at a discontinuity or when it stops receiving. Its parked frames + * drop, and its arrivals stop holding the reference; the other tracks' timelines are untouched. + */ + reset(): void { + this.#clock.end(this.#part); + } + + /** Release this track's hold on the clock. */ + close(): void { + this.#clock.end(this.#part); + this.#clock.parts.update((parts) => parts.filter((part) => part !== this.#part)); + } +} + +/** The playback clock: decides when each frame plays so audio and video stay together at the configured delay. */ +export class Sync { + /** The wired inputs, read-only. */ + readonly in: Readonlys; + /** The derived outputs, read-only. */ + readonly out: Readonlys; + + readonly #clock: Clock; + /** Whether buffered startup has released the media consumers. @internal */ + readonly playing: Getter; + + #signals = new Effect(); + + constructor(props?: Inputs) { + this.#clock = new Clock(props); + this.in = this.#clock.in; + this.out = readonlys(this.#clock.out); + this.playing = this.#clock.playing; + + this.#signals.run(this.#runInstant.bind(this)); + this.#signals.run(this.#runDelay.bind(this)); + this.#signals.run(this.#runMaxDelay.bind(this)); + this.#signals.run(this.#runStartup.bind(this)); + } + + /** + * Open a track's hold on the clock: its playout target, startup span, arrivals, and paced frames. + * @internal + */ + track(props: TrackProps): Track { + return new Track(this.#clock, props); + } + + // Startup is data-driven: time passing cannot fill a missing track. The first picture still + // travels through the ordinary renderer while the consumers wait for this shared gate. + #runStartup(effect: Effect): void { + const clock = this.#clock; + effect.get(clock.epoch); + const buffered = effect.get(clock.out.buffered); + const delay = effect.get(clock.out.delay); + const reference = effect.get(clock.out.reference); + const spans = effect.get(clock.parts).flatMap((part) => { + const span = part.props.startup && effect.get(part.props.startup); + return span === undefined ? [] : [{ part, span }]; + }); + if (!buffered || spans.length === 0) { + clock.playing.set(true); + return; + } + if (clock.started) return; + const ready = spans.flatMap(({ part, span }) => (span ? [{ part, span }] : [])); + if ( + reference === undefined || + ready.length !== spans.length || + ready.some(({ span }) => span.end - span.start < delay) + ) { + clock.playing.set(false); + return; + } + // The track with the earliest opening holds the reference it anchors. + const earliest = ready.reduce((a, b) => (b.span.start < a.span.start ? b : a)); + const first = earliest.span.start; + clock.started = true; + clock.owner = earliest.part; + // Anchor at the retained opening, not at its arrival: startup may have waited on another track. + // Every track's own anchor moves by the same offset, keeping the gaps between tracks that + // playing would have held, so a later handoff moves the playhead only by that gap. + const anchored = wall() - first - delay; + for (const part of clock.parts.peek()) { + if (part.anchor !== undefined) part.anchor += anchored - reference; + } + clock.out.reference.set(anchored); + clock.playing.set(true); + } + + #runInstant(effect: Effect): void { + this.#clock.out.instant.set(effect.get(this.in.delay) === "instant"); + } + + // Derive `buffered` / `maxDelay` from the resolved delay and the configured lookahead. + #runMaxDelay(effect: Effect): void { + const out = this.#clock.out; + const delay = effect.get(out.delay); + // "instant" holds nothing, so a configured lookahead doesn't apply. + const buffer = effect.get(out.instant) ? Milli.zero : effect.get(this.in.buffer); + + out.buffered.set(buffer > 0); + out.maxDelay.set(Milli(delay + buffer)); + } + + // 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 mode = effect.get(this.in.delay); + + let delay = Milli.zero; + if (typeof mode === "number") { + delay = mode; + } else if (mode === "auto") { + for (const { props } of effect.get(this.#clock.parts)) { + delay = Milli(Math.max(delay, effect.get(props.target) ?? Milli.zero)); + } + } + + this.#clock.out.delay.set(delay); + } + + /** + * Re-anchor playback to the next frame received. Call this at an utterance boundary + * in buffered mode (typically alongside flushing the audio buffer) so the new content + * plays from its own first frame instead of inheriting the previous reference. + */ + reset(): void { + this.#clock.reset(); + } + + /** + * The PTS that should be rendering right now, derived from the reference and delay; zero until + * media time zero's turn. Undefined before ingress or while buffered startup is filling. + */ + now(): Milli | undefined { + return this.#clock.now(); + } /** Stop and release everything this holds. */ close() { - this.#closed.set(true); + this.#clock.closed.set(true); this.#signals.close(); } } diff --git a/js/watch/src/text/renderer.test.ts b/js/watch/src/text/renderer.test.ts index c54a8164989..13c27a18d85 100644 --- a/js/watch/src/text/renderer.test.ts +++ b/js/watch/src/text/renderer.test.ts @@ -1,5 +1,9 @@ import { beforeAll, describe, expect, test } from "bun:test"; +import { Milli, yieldNow } from "@moq/clock"; +import type * as Catalog from "@moq/hang/catalog"; +import { Signal } from "@moq/signals"; import { parseText, VTTCue } from "media-captions"; +import { Sync } from "../sync"; import { type CueStore, clearCue, @@ -7,9 +11,11 @@ import { insertCues, MAX_RETAINED, pruneCues, + Renderer, rollUp, sanitizeCue, } from "./renderer"; +import type { Source } from "./source"; function store(cues: VTTCue[] = []): CueStore { return { cues, clears: [] }; @@ -285,3 +291,40 @@ describe("sanitizeCue", () => { expect(html(sanitizeCue("[BUNNY SNORES]"))).toBe("[BUNNY SNORES]"); }); }); + +describe("Renderer", () => { + test("holds the selected rendition's catalog delay in the shared clock while it can render", async () => { + const config = new Signal({ delay: 300 } as Catalog.TextConfig); + const source = { + in: { broadcast: new Signal(undefined) }, + out: { config, track: new Signal(undefined) }, + } as unknown as Source; + const enabled = new Signal(true); + const sync = new Sync(); + // Never touched: with no track selected, nothing subscribes or renders. + const container = new Signal({} as HTMLElement); + const renderer = new Renderer({ source, sync, enabled, container }); + try { + await yieldNow(); + expect(sync.out.delay.peek()).toBe(Milli(300)); + + // Captions that are off hold nothing. + enabled.set(false); + await yieldNow(); + expect(sync.out.delay.peek()).toBe(Milli.zero); + enabled.set(true); + container.set(undefined); + await yieldNow(); + expect(sync.out.delay.peek()).toBe(Milli.zero); + container.set({} as HTMLElement); + await yieldNow(); + expect(sync.out.delay.peek()).toBe(Milli(300)); + config.set(undefined); + await yieldNow(); + expect(sync.out.delay.peek()).toBe(Milli.zero); + } finally { + renderer.close(); + sync.close(); + } + }); +}); diff --git a/js/watch/src/text/renderer.ts b/js/watch/src/text/renderer.ts index 049caca2b23..2e405267c5f 100644 --- a/js/watch/src/text/renderer.ts +++ b/js/watch/src/text/renderer.ts @@ -5,7 +5,7 @@ import { Error as NetError } from "@moq/net"; import { Derived, Effect, type Getter, getter, type Inputs, type Readonlys } from "@moq/signals"; import type { VTTCue, VTTRegion } from "media-captions"; import { subscribeMedia } from "../media"; -import type { Sync } from "../sync"; +import type { Sync, Track } from "../sync"; import type { Source } from "./source"; // Drop cues whose end is more than this far behind the playhead, to bound memory on a long stream. @@ -227,6 +227,10 @@ export class Renderer { #signals = new Effect(); + // The captions' hold on the clock: the selected rendition's catalog delay, so cues from a + // rendition that trails the others still line up with them. + readonly #clock: Track; + // Whether the clock-skew warning has already fired for the current track, so a misconfigured // publisher costs one line instead of one per cue. #skewWarned = false; @@ -239,6 +243,15 @@ export class Renderer { enabled: getter(props?.enabled ?? true), }; + const target = this.#signals.computed((effect) => { + // Captions with nowhere to render hold nothing. + if (!effect.get(this.in.enabled) || !effect.get(this.in.container)) return undefined; + const delay = effect.get(this.source.out.config)?.delay; + return delay !== undefined ? Milli(delay) : undefined; + }); + this.#clock = this.sync.track({ target }); + this.#signals.cleanup(() => this.#clock.close()); + this.#signals.run(this.#run.bind(this)); } diff --git a/js/watch/src/video/decoder.test.ts b/js/watch/src/video/decoder.test.ts index fa400a5cacf..6f643ff4aa0 100644 --- a/js/watch/src/video/decoder.test.ts +++ b/js/watch/src/video/decoder.test.ts @@ -8,7 +8,7 @@ import { Effect, Signal } from "@moq/signals"; import { ringSamples } from "../audio/latency"; import { AudioRingBuffer } from "../audio/ring-buffer"; import type { Broadcast } from "../broadcast"; -import { type Delay, Sync } from "../sync"; +import { type Delay, Sync, Track } from "../sync"; import { Decoder, insertRecoveryPoint } from "./decoder"; import { Source } from "./source"; @@ -380,9 +380,12 @@ async function guardedPlayback( ? mediaProducer.createTrack("video", { timescale: Moq.Time.Timescale.MILLI, maxAge: Container.MAX_AGE }) : undefined; const decoder = new Decoder({ source, sync, enabled }); + // Another track on the same clock, whose arrivals the tests drive by hand. + const audio = sync.track({ target: new Signal(undefined) }); await microtasks(); return { decoder, + audio, delay, fail: (error: DOMException, index = -1) => errors.at(index)?.(error), emit: (timestamp: number, index = -1) => @@ -644,7 +647,7 @@ for (const kind of ["legacy", "cmaf"] as const) { expect(playback.outputs).toEqual(stage === "queued" ? [] : [picture]); const before = playback.decoder.out.frame.peek(); - playback.sync.received(Milli(skipTo), Milli(5000)); + playback.audio.received(Milli(skipTo), Milli(5000)); playback.release(); await microtasks(); const due = playback.decoder.out.frame.peek(); @@ -680,7 +683,7 @@ for (const kind of ["legacy", "cmaf"] as const) { expect(playback.outputs).toEqual([]); // "instant" has no cap, so every newer frame skips the playhead without passing over anything. - playback.sync.received(Milli(10_000), Nano.toMilli(Nano(now()))); + playback.audio.received(Milli(10_000), Nano.toMilli(Nano(now()))); playback.release(); await microtasks(); @@ -782,14 +785,18 @@ for (const kind of ["legacy", "cmaf"] as const) { it("releases a failed video subscription and recovers only on a new source", async () => { const close = spyOn(Moq.Track.Subscriber.prototype, "close"); + const reset = spyOn(Track.prototype, "reset"); const playback = await guardedPlayback("legacy", []); try { const before = close.mock.calls.length; + const resets = reset.mock.calls.length; const failure = new DOMException("invalid encoded video", "EncodingError"); playback.fail(failure); await microtasks(); expect(playback.decoder.out.error.peek()).toBe(failure); expect(close.mock.calls.length).toBeGreaterThan(before); + // Video's arrivals stop holding the shared clock. + expect(reset.mock.calls.length).toBeGreaterThan(resets); playback.enabled.set(false); await microtasks(); playback.enabled.set(true); @@ -801,6 +808,7 @@ it("releases a failed video subscription and recovers only on a new source", asy } finally { playback.close(); close.mockRestore(); + reset.mockRestore(); } }); @@ -932,7 +940,7 @@ it("a shared audio clock reset cannot present a parked video frame", async () => await microtasks(); expect(playback.decoder.out.timestamp.peek()).toBe(Milli(0)); // The next utterance anchors the clock past the parked frame's deadline. - playback.sync.received(Milli(2000)); + playback.audio.received(Milli(2000)); await microtasks(); expect(playback.decoder.out.timestamp.peek()).toBe(Milli(0)); } finally { @@ -947,9 +955,9 @@ it("closing the track releases a parked video wait", async () => { const reads: Read[] = []; const playback = await guardedPlayback("legacy", reads); const waits: Promise[] = []; - const original = playback.sync.wait.bind(playback.sync); - const wait = spyOn(playback.sync, "wait").mockImplementation((timestamp, signal) => { - const waiting = original(timestamp, signal); + const original = Track.prototype.wait; + const wait = spyOn(Track.prototype, "wait").mockImplementation(function (this: Track, timestamp, signal) { + const waiting = original.call(this, timestamp, signal); waits.push(waiting); return waiting; }); @@ -988,7 +996,7 @@ it("a reset drops a parked video frame even when the next anchor restores the re expect(playback.submitted).toEqual([0, 100_000]); // In one turn, so the reference never visibly changes and the parked sleep keeps its timer. playback.sync.reset(); - playback.sync.received(Milli(0)); + playback.audio.received(Milli(0)); await microtasks(); // The parked sleep comes due; its frame belongs to the timeline the reset ended. await clock.advance(1000); @@ -1210,7 +1218,7 @@ for (const kind of ["legacy", "cmaf"] as const) { const reads: Read[] = []; const playback = await guardedPlayback(kind, reads, [], false, 1000); const audio = new Signal<{ start: Milli; end: Milli } | null>(null); - const remove = playback.sync.register(new Signal(Milli(100)), audio); + const remove = playback.sync.track({ target: new Signal(Milli(100)), startup: audio }); const reset = spyOn(playback.sync, "reset"); try { playback.enabled.set(false); @@ -1238,7 +1246,7 @@ for (const kind of ["legacy", "cmaf"] as const) { expect(playback.decoder.out.timestamp.peek()).toBe(Milli(940)); expect(reset).not.toHaveBeenCalled(); } finally { - remove(); + remove.close(); reset.mockRestore(); playback.close(); await microtasks(); @@ -1258,7 +1266,7 @@ for (const kind of ["legacy", "cmaf"] as const) { reads.push(sample(group * 2n, Number(group) * 40_000)); } const playback = await guardedPlayback(kind, reads, [], false, 1000); - const remove = playback.sync.register(new Signal(Milli(100)), new Signal(null)); + const remove = playback.sync.track({ target: new Signal(Milli(100)), startup: new Signal(null) }); try { await microtasks(); await clock.advance(0); @@ -1267,7 +1275,7 @@ for (const kind of ["legacy", "cmaf"] as const) { expect(playback.submitted.at(-1)).toBe(1_120_000); expect(playback.sync.playing.peek()).toBe(false); } finally { - remove(); + remove.close(); playback.close(); await microtasks(); clock.installed.close(); @@ -1278,7 +1286,7 @@ for (const kind of ["legacy", "cmaf"] as const) { it("bounds startup decoding while another track is missing and releases it on close", async () => { const clock = manualAt(0); const playback = await guardedPlayback("legacy", [], [], true, 1000); - const remove = playback.sync.register(new Signal(Milli(100)), new Signal(null)); + const remove = playback.sync.track({ target: new Signal(Milli(100)), startup: new Signal(null) }); if (!playback.track) throw new Error("missing video track"); const group = playback.track.appendGroup(); try { @@ -1291,7 +1299,7 @@ it("bounds startup decoding while another track is missing and releases it on cl await microtasks(); expect(playback.pictures.every((frame) => frame.closed)).toBe(true); } finally { - remove(); + remove.close(); group.finish(); playback.close(); await microtasks(); @@ -1304,7 +1312,7 @@ it("resumes decoding the moment shared startup releases", async () => { const clock = manualAt(0); const playback = await guardedPlayback("legacy", [], [], true, 1000); const audio = new Signal<{ start: Milli; end: Milli } | null>(null); - const remove = playback.sync.register(new Signal(Milli(100)), audio); + const remove = playback.sync.track({ target: new Signal(Milli(100)), startup: audio }); if (!playback.track) throw new Error("missing video track"); const group = playback.track.appendGroup(); try { @@ -1317,7 +1325,7 @@ it("resumes decoding the moment shared startup releases", async () => { await clock.advance(0); expect(playback.submitted).toHaveLength(8); } finally { - remove(); + remove.close(); group.finish(); playback.close(); await microtasks(); @@ -1352,7 +1360,7 @@ it("continues filling when the shared floor rises while video waits on audio", a const clock = manualAt(0); const playback = await guardedPlayback("legacy", [], [], true, 1000); const audio = new Signal<{ start: Milli; end: Milli } | null>(null); - const remove = playback.sync.register(new Signal(Milli(100)), audio); + const remove = playback.sync.track({ target: new Signal(Milli(100)), startup: audio }); if (!playback.track) throw new Error("missing video track"); const group = playback.track.appendGroup(); try { @@ -1366,7 +1374,7 @@ it("continues filling when the shared floor rises while video waits on audio", a await clock.advance(240); expect(playback.decoder.out.frame.peek()?.timestamp).toBeGreaterThanOrEqual(200_000); } finally { - remove(); + remove.close(); group.finish(); playback.close(); await microtasks(); @@ -1397,7 +1405,7 @@ for (const kind of ["legacy", "cmaf"] as const) { const clock = manualAt(0); const playback = await guardedPlayback(kind, [sample(0n, 0), sample(0n, 100_000)], [0, 100_000], false, 1000); const other = new Signal<{ start: Milli; end: Milli } | null>(null); - const remove = playback.sync.register(new Signal(Milli(100)), other); + const remove = playback.sync.track({ target: new Signal(Milli(100)), startup: other }); try { playback.release(); await microtasks(); @@ -1419,7 +1427,7 @@ for (const kind of ["legacy", "cmaf"] as const) { await microtasks(); expect(playback.pictures.filter((frame) => !frame.closed)).toEqual([]); } finally { - remove(); + remove.close(); playback.close(); await microtasks(); clock.installed.close(); @@ -1577,16 +1585,16 @@ describe("Decoder with a warmup rendition", () => { it("moves the shared clock only from the min-delay decoder", async () => { const run = await refresh(); - const received = spyOn(run.playback.sync, "received"); - const reset = spyOn(run.playback.sync, "reset"); + const received = spyOn(Track.prototype, "received"); + const reset = spyOn(Track.prototype, "reset"); try { await run.push(MIN, at(10n, 1000, true), at(10n, 1033)); await run.push(MAX, at(9n, 700, true), at(9n, 733), at(10n, 1000, true), at(10n, 1033)); expect(received.mock.calls.map(([timestamp]) => timestamp)).toEqual([Milli(1000), Milli(1033)]); expect(run.current()).toEqual([1033, MAX]); - // The publisher restarts its timeline. The min-delay decoder resets the clock and shows the - // new timeline; the max-delay picture still on the old one never wins. + // The publisher restarts its timeline. The min-delay decoder restarts video's timeline on the + // clock and shows the new one; the max-delay picture still on the old one never wins. await run.push(MIN, marker(10n, 1_066_000), at(11n, 0, true)); expect(reset).toHaveBeenCalledTimes(1); expect(run.current()).toEqual([0, MIN]); diff --git a/js/watch/src/video/decoder.ts b/js/watch/src/video/decoder.ts index 38503bb77bc..d26a1dbfc63 100644 --- a/js/watch/src/video/decoder.ts +++ b/js/watch/src/video/decoder.ts @@ -20,7 +20,7 @@ import { base64ToBytes } from "../base64"; import { arrivedAt, type Counter, type Key, type Received, Tally } from "../feedback"; import { type Failure, failureMoved, nextMedia, subscribeMedia } from "../media"; import { supportCacheKey } from "../support-probe"; -import type { Sync } from "../sync"; +import type { Sync, Track } from "../sync"; import { type DecoderConfig, decoderConfigKey, type PlaybackIdentity, playbackIdentity } from "./config"; import { caughtUp, renditionJitter, switchJitter } from "./playhead"; import { Recovery } from "./recovery"; @@ -111,6 +111,8 @@ export class Decoder { readonly source: Source; /** The clock that schedules playback. */ readonly sync: Sync; + // Video's hold on the clock: its target, startup span, arrivals, and paced frames. + readonly #clock: Track; readonly #out: DecoderOutput = { error: new Signal(undefined), @@ -188,7 +190,8 @@ export class Decoder { const ranges = effect.get(active.buffered); return startupSpan(ranges, timestamp); }); - this.#signals.cleanup(this.sync.register(target, startup)); + this.#clock = this.sync.track({ target, startup }); + this.#signals.cleanup(() => this.#clock.close()); this.#identity = this.#signals.computed((effect) => { const config = effect.get(this.source.out.config); return config ? playbackIdentity(config) : undefined; @@ -256,6 +259,8 @@ export class Decoder { // Close the active track when disabled (e.g. paused or not visible). // The pending cleanup won't do this because it was already promoted to #active. this.#active.set(undefined); + // Nothing arrives until it resumes, so its arrivals stop holding the clock. + this.#clock.reset(); // Pause and visibility keep the picture while the broadcast is active. Disconnecting // the broadcast or removing every rendition releases it. if (!broadcast || !effect.get(broadcast.out.active) || (enabled && !track)) { @@ -268,6 +273,7 @@ export class Decoder { if (!active || !key) { // Going offline should clear the last rendered frame. this.#active.set(undefined); + this.#clock.reset(); this.#clearCurrentFrame(); this.#out.buffered.set([]); return; @@ -293,6 +299,7 @@ export class Decoder { // Start a new pending effect. let pending: Rendition | undefined = new Rendition({ sync: this.sync, + clock: this.#clock, broadcast: active, track, name: Catalog.track(track, identity), @@ -316,6 +323,8 @@ export class Decoder { pending.close(); pending = undefined; this.#pending.set(undefined); + // With no active rendition left, nothing holds video's timeline. + if (!this.#active.peek()) this.#clock.reset(); }); effect.set(this.#pending, { track, catalog: broadcast.out.catalog }); @@ -364,6 +373,8 @@ export class Decoder { if (!error) return; this.#recordError(active, error); this.#active.set(undefined); + // A pending replacement carries video's timeline on; its parked frames must stay. + if (!this.#pending.peek()) this.#clock.reset(); this.#clearCurrentFrame(); }); @@ -474,6 +485,8 @@ export class Decoder { interface DecoderTrackProps { sync: Sync; + // Video's hold on the clock, shared by every rendition: their frames are one timeline. + clock: Track; // Set for the max-delay decoder of a rendition with a catalog `warmup`: its start and its // lateness budget reach this much further, and it leaves the shared clock and stats alone. warmup?: Getter; @@ -504,6 +517,7 @@ interface DecoderTrackProps { class DecoderTrack { error = new Signal(undefined); sync: Sync; + clock: Track; broadcast: Moq.Broadcast.Consumer; track: string; name: string; @@ -574,6 +588,7 @@ class DecoderTrack { constructor(props: DecoderTrackProps) { this.sync = props.sync; + this.clock = props.clock; this.broadcast = props.broadcast; this.track = props.track; this.name = props.name; @@ -608,7 +623,7 @@ class DecoderTrack { #receive(frame: Container.ReceivedFrame, accepted: boolean): void { if (!this.#drives) return; const timestamp = Milli.fromMicro(frame.timestamp as Micro); - const late = accepted && this.sync.received(timestamp, frame.arrival); + const late = accepted && this.clock.received(timestamp, frame.arrival); const bytes = frame.payload.byteLength; const arrived = arrivedAt(frame.arrival); this.tally?.arrive(this.key, timestamp, arrived); @@ -676,14 +691,14 @@ class DecoderTrack { // Render something while we wait for the sync to catch up, unless the frame is // already off the timeline: WebCodecs can hold a picture past a reset or a skip. // Only the min-delay decoder posts one: the max-delay picture shows on the clock. - if (this.#drives && this.frame.peek() === undefined && this.sync.presentable(timestamp)) { + if (this.#drives && this.frame.peek() === undefined && this.clock.presentable(timestamp)) { this.group = group; this.frame.set(frame.clone()); } // Resolves at once when the delay is "instant". Closing the track releases it. // Check the captured signal again in case cancellation followed the resolved wait. - if (!(await this.sync.wait(timestamp, signal)) || signal.aborted) return; + if (!(await this.clock.wait(timestamp, signal)) || signal.aborted) return; // The max-delay decoder restarts its timeline after the clock did, so the clock's own // epoch cannot tell its leftover pictures apart. if (this.#generation !== generation) return; @@ -903,7 +918,7 @@ class DecoderTrack { try { // Wait on the first picture, which is due the moment startup releases, so decoding // resumes with the whole floor still ahead instead of once it has played out. - await effect.race(refill, this.sync.wait(first, waiting.abort)); + await effect.race(refill, this.clock.wait(first, waiting.abort)); } finally { waiting.close(); } @@ -930,8 +945,9 @@ class DecoderTrack { this.#buffered.set([]); this.#generation++; this.start.set(group); - // The min-delay decoder restarts the shared clock. The max-delay one reads the same end later. - if (this.#drives) this.sync.reset(); + // The min-delay decoder restarts video's timeline, leaving audio's. The max-delay one reads the + // same end later. + if (this.#drives) this.clock.reset(); this.#resetPending = true; } diff --git a/quest/m1/README.md b/quest/m1/README.md index 050aaea2152..360bbed1f94 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -160,7 +160,6 @@ their end report, so an embedder's drain alert pages on real cuts. - [Sockets](/quest/m1/socket/README.md) - a generic dial/accept `Socket` in moq-net, Rust, JS, and Python, with tracks as streams; moq-boy and CARLA move onto it - [A smaller, safer API than release](/quest/m1/api-audit/README.md) - every surface added since release has a consumer or is gone, overlapping operations are collapsed, names mirror across Rust, JS, and the bindings, and every break says how to migrate - [Audio jitter target](/quest/m1/audio-jitter-target/README.md) - the playout target is default in both languages; time-stretched browser playback without steady-state underruns and a native trace replay remain -- [A/V clock](/quest/m1/av-clock.md) - the audio playhead drives Sync.reference while audio plays, through per-track sync handles - [Container reads drop backfill](/quest/m1/read-backfill.md) - the JS container consumer drops groups below its read cursor like Rust - [Buffered playback subscribes in ascending order](/quest/m1/buffered-ascending.md) - a buffered `` asks for the oldest group first, so a voice reply's opening arrives first, and audio's head hold shrinks to the floor - [Push-to-talk loses no speech](/quest/m1/push-to-talk/README.md) - an unmute is heard from its first frame with no catalog round trip: mute moves from holding a disabled rendition to suppressed silence with comfort noise, after a route spike; capture delivers frames on a steady cadence diff --git a/quest/m1/api-audit/README.md b/quest/m1/api-audit/README.md index c02205b885e..4656fc70329 100644 --- a/quest/m1/api-audit/README.md +++ b/quest/m1/api-audit/README.md @@ -12,7 +12,7 @@ Non-goals: redesigning independently scoped surfaces ([subscribe ranges](/quest/m1/subscribe-ranges/README.md), [archives](/doc/concept/archives.md), [compressed tracks](/doc/concept/compression.md), -[A/V clock](/quest/m1/av-clock.md)), and inherited debt that a later quest +and `Sync`'s per-track clock handles), and inherited debt that a later quest already reshapes. ## Plan diff --git a/quest/m1/audio-jitter-target/README.md b/quest/m1/audio-jitter-target/README.md index d2d108b062b..05bc0f3b2dc 100644 --- a/quest/m1/audio-jitter-target/README.md +++ b/quest/m1/audio-jitter-target/README.md @@ -13,8 +13,8 @@ the same arrival trace. Boundaries: converging by playing slightly faster or slower is the browser's [Watch](/quest/m1/audio-jitter-target/watch.md) quest. No packet loss concealment. -Video keeps its own target; making the audio playhead the clock is [A/V -clock](/quest/m1/av-clock.md). +Video keeps its own target; the shared clock stays on local time, and +[Watch](/quest/m1/audio-jitter-target/watch.md) converges the audio ring on it. ## Plan diff --git a/quest/m1/audio-jitter-target/watch.md b/quest/m1/audio-jitter-target/watch.md index 74f5dc24c53..939cde0ba6c 100644 --- a/quest/m1/audio-jitter-target/watch.md +++ b/quest/m1/audio-jitter-target/watch.md @@ -150,6 +150,16 @@ path, so the 20 ms default stays. Decided 2026-10-15: keep this proof in m1 with the existing playback and A/V clock work. +Decided 2026-11-05 (maintainer, /quest-plan): steady-state lip sync comes +from this stretch converging the ring's playhead on `Sync.now()`, the shared +clock on local time anchored at the earliest arrival. The clock never follows +audio: one slow track must not hold the others, so audio and video stay in sync +in the steady state and drift apart only while a track is congested. +[moq-dev/moq#7941](https://github.com/moq-dev/moq/pull/7941) dropped an +audio-driven clock for this. Recommendation: measure the gap against the sample +reaching the output, through `AudioContext.getOutputTimestamp()`, so output +latency, Bluetooth included, is part of the lip sync. + ## Waiting - Maintainer: Safari with 40 ms or more of added RTT, the baseline row the @@ -159,10 +169,6 @@ A/V clock work. panel open for two minutes, and record its audio Underruns after 30 s and the latency tab's auto delay. -## Required - -- [A/V clock](/quest/m1/av-clock.md) - the clock the stretch converges toward - ## Related - [Capped budgets](/quest/m1/audio-jitter-target/capped-budgets.md) - re-measures the AAC browser budgets after the publisher change, no gate on this proof diff --git a/quest/m1/av-clock.md b/quest/m1/av-clock.md deleted file mode 100644 index 7a7726ec1fd..00000000000 --- a/quest/m1/av-clock.md +++ /dev/null @@ -1,72 +0,0 @@ -# [M] The audio playhead drives Sync.reference while audio plays - -## Goal - -`js/watch` keeps audio and video in sync from one clock. While audio plays, -`Sync.reference` derives from the audio playhead; muted or video-only playback -falls back to the wall clock. Today the ring is depth-driven and free-runs on -the AudioContext clock while video paces against a wall clock anchored at the -earliest arrival, so a ring that re-buffers or skips drifts against video until -the next re-anchor. - -Acceptance includes the 2026-10-07 audit's skip case: a video-only latency -skip (`js/watch/src/video/decoder.ts`, `sync.reset()` on a video -discontinuity) must not move audio's timeline. Today it re-anchors the shared -`Sync` while the audio ring stays on the old one. Folded in here rather than -fixed separately, since audio driving the clock removes the shared anchor. - -## Plan - -Moved from m0 to m1 in the 2026-09-30 audit: it waits on the whole jitter -line and is a published `@moq/watch` break, so it is not in flight. - -Settled: per-track handles, and this quest lands them. `sync.track("audio")` -and `sync.track("video")` each report their advertised delay and measured -spread, and one is nominated as the clock source. `SyncInput` -(`js/watch/src/sync.ts`, today `delay` and `buffer`) breaks once, -and a third track joins without another pair of inputs; `Sync.register` -already keeps one jitter entry per track and grows into the handles. The -measured target per track comes from the [Audio jitter -target](/quest/m1/audio-jitter-target/README.md) line: each decoder registers -its own target through `Sync.register`. This quest folds the registrations -into the handles. `SyncInput` is a published `@moq/watch` shape, so this is a -break. The decoder-facing calls (`received`, `wait`, `register`) are already -internal, with `wait` resolving whether to present or drop, so the handles build -on that. - -Recommendations for the implementation: - -- Playhead source. The worklet's ring (`js/watch/src/audio/ring-buffer.ts`) - posts an estimate and the main thread extrapolates between posts. -- Video reads a locally extrapolated clock, re-synced once per audio quantum, - so the per-frame `sync.wait()` (`js/watch/src/video/decoder.ts`) never - crosses a thread. -- Transitions. On mute or audio track end the reference falls back to the - wall clock at the last audio-derived value, so video does not jump. A ring - re-stall reads as the playhead pausing, and the reference pauses with it. -- Reset coupling stays: `Player.reset()` (`js/watch/src/player.ts`) already - flushes the audio ring alongside `sync.reset()`. -- The text renderer is the third track: it reads `sync.now()` - (`js/watch/src/text/renderer.ts`) to drive the cue clock and prune cues. -- Close the player gaps [#4170](https://github.com/moq-dev/moq/pull/4170) - left, since the handles own them. `Sync.received` only ever lowers its - reference, so after the earliest subscribed track leaves, playback stays - anchored to it; a track's handle going away must release its part of the - reference, the same expiry the jitter target's arrival minimum has. Text - renditions register no floor with `Sync`, so their catalog `delay` is - ignored; the text handle registers one like audio and video. The MSF - catalog schema (`js/msf/src/catalog.ts`) accepts a negative `delay` and - folds it into absent; refuse it on decode instead. - -Decided 2026-10-27 (maintainer): this waits only on the estimator's browser -budgets being re-measured, not the whole -[audio jitter target](/quest/m1/audio-jitter-target/README.md) epic, whose -remaining manual browser proof does not gate reshaping the clock. -[moq-dev/moq#7898](https://github.com/moq-dev/moq/pull/7898) re-measured every -browser row. What [capped budgets](/quest/m1/audio-jitter-target/capped-budgets.md) -still owes, the AAC rows after a publisher change, does not gate it either. - -## Related - -- [Watch](/quest/m1/audio-jitter-target/watch.md) - time-stretches the audio ring toward the clock this produces -- [Watch worker](/quest/m2/watch-worker.md) - moves `Sync` into a worker afterwards; keep the handles free of main-thread assumptions diff --git a/quest/m2/audio-bufferbloat-av.md b/quest/m2/audio-bufferbloat-av.md index daf7929ad38..12caea622a3 100644 --- a/quest/m2/audio-bufferbloat-av.md +++ b/quest/m2/audio-bufferbloat-av.md @@ -4,16 +4,15 @@ A nightly test sends audio and video together through a bufferbloated link and proves audio holds up better than video: audio stays within its underrun -and skip budgets while video may freeze or skip, and every video frame that -renders is in sync with audio. Today no test publishes both through the -shaper, and the audio-quality profiles leave rate limits off. +and skip budgets while video may freeze, skip, or lag audio. Today no test +publishes both through the shaper, and the audio-quality profiles leave rate +limits off. ## Plan Decided in planning (2026-10-20): -- Behavior asserted is today's design: continuity plus sync. Audio is sent - first (players subscribe to audio at a higher priority than video) and its +- Behavior asserted is audio continuity. Audio is sent first (players subscribe to audio at a higher priority than video) and its ring plays through a video stall. In `"auto"` delay the shared target is the deepest track's, so late video raises audio latency too; that is expected, not a failure. Making audio latency independent of video would be a player change, not this @@ -21,10 +20,11 @@ Decided in planning (2026-10-20): - Extend `test/audio-quality` rather than routing media QA through the shaper: add rows whose publisher sends the media QA pattern (frame counter plus stepped tone), and reuse the tap worklet, budgets, captures, and - replays. Add media QA's sync probe to grade rendered frames against the - same sync bound `just test media` holds the browser to. Video freezes and - skips are reported, not gated: the point is that audio does not pay for - them. + replays. Add media QA's sync probe to report each rendered frame's gap to + audio. Video freezes, skips, and lag are reported, not gated: the point is + that audio does not pay for them. Since 2026-11-05 the shared clock never + follows audio, so congested video lags it by design, and steady-state lip + sync belongs to [the Watch stretch](/quest/m1/audio-jitter-target/watch.md). - Folded in: the native `pulse` rows become enforced. They report only because an intermittent two-second device stall voids about one AAC native row in seven (see `test/audio-quality/README.md`). Reproduce and fix that @@ -55,5 +55,4 @@ Public API: none. Wire: none. ## Related - [Audio jitter target](/quest/m1/audio-jitter-target/README.md) - the estimator whose budgets these rows extend -- [A/V clock](/quest/m1/av-clock.md) - audio driving the clock should make these rows easier, not harder - [ViSQOL grading](/quest/m2/audio-visqol.md) - adds a quality score to the same harness diff --git a/quest/m2/moqsrc-av-sync.md b/quest/m2/moqsrc-av-sync.md index 6c19324b67c..ced9ec5e39b 100644 --- a/quest/m2/moqsrc-av-sync.md +++ b/quest/m2/moqsrc-av-sync.md @@ -26,4 +26,3 @@ Public API: none. Wire: none. ## Related - [moqsrc reconnect](/doc/bin/gstreamer.md#moqsrc) - redials and resumes on the same pads, a natural second scenario for this check -- [A/V clock](/quest/m1/av-clock.md) - the browser side of A/V sync diff --git a/quest/m2/plan-watch-worker.md b/quest/m2/plan-watch-worker.md index cfb378c446c..b11b2aaaff9 100644 --- a/quest/m2/plan-watch-worker.md +++ b/quest/m2/plan-watch-worker.md @@ -72,4 +72,3 @@ follow-up wires the capture worklet to the encoder worker with a ## Related - [Watch worker](/quest/m2/watch-worker.md) - the implementation this rewrites -- [A/V clock](/quest/m1/av-clock.md) - the `Sync` shape the implementation moves; the prototype can pace against today's diff --git a/quest/m2/watch-data-sync.md b/quest/m2/watch-data-sync.md index 5921c62196e..c5f207c6953 100644 --- a/quest/m2/watch-data-sync.md +++ b/quest/m2/watch-data-sync.md @@ -28,5 +28,4 @@ releases it. ## Required -- [A/V clock](/quest/m1/av-clock.md) - the playhead the reader releases against - [Watch worker](/quest/m2/watch-worker.md) - moves `Sync` into a worker, where the reader registers diff --git a/quest/m2/watch-decode-gate.md b/quest/m2/watch-decode-gate.md index ce3d8012d58..3fdb50d2d5d 100644 --- a/quest/m2/watch-decode-gate.md +++ b/quest/m2/watch-decode-gate.md @@ -26,7 +26,3 @@ Public API: none. Wire: none. ## Required - [Watch worker](/quest/m2/watch-worker.md) - moves decode and `Sync` into a worker - -## Related - -- [A/V clock](/quest/m1/av-clock.md) - changes what `Sync` waits on diff --git a/quest/m2/watch-worker.md b/quest/m2/watch-worker.md index 1b81e2469fb..08e74f12a6b 100644 --- a/quest/m2/watch-worker.md +++ b/quest/m2/watch-worker.md @@ -31,7 +31,6 @@ Gate on the plan's jank harness and N-player sweep, both nightly. ## Required - [Plan: watch worker](/quest/m2/plan-watch-worker.md) - picks the worker model and rewrites this quest -- [A/V clock](/quest/m1/av-clock.md) - reshapes `Sync` and the worklet playhead, so the move to the worker happens once ## Related diff --git a/quest/m2/webcodecs-priming.md b/quest/m2/webcodecs-priming.md index f2f50a709ca..050390794bd 100644 --- a/quest/m2/webcodecs-priming.md +++ b/quest/m2/webcodecs-priming.md @@ -18,7 +18,7 @@ gap ([w3c/webcodecs#626](https://github.com/w3c/webcodecs/issues/626)). Out of scope: Rust, where `moq-audio` already declares Opus lookahead as pre-skip and folds the AAC delay into timestamps (`folded_delay` in `rs/moq-audio/src/encode/encoder.rs`); and playback-side sync, which -[A/V clock](/quest/m1/av-clock.md) owns. +[Watch](/quest/m1/audio-jitter-target/watch.md) converges by time-stretching. ## Plan diff --git a/test/video/src/captions.ts b/test/video/src/captions.ts index df6cbc31935..0a46bf4fc7f 100644 --- a/test/video/src/captions.ts +++ b/test/video/src/captions.ts @@ -30,7 +30,9 @@ export function captions() { const target = new Signal(undefined); const source = new Source({ broadcast, target }); const sync = new Sync({ delay: Milli.zero }); - sync.received(Milli(1000)); + // Stands in for the media that anchors the clock the cues show against. + const media = sync.track({ target: new Signal(undefined) }); + media.received(Milli(1000)); const container = document.createElement("div"); container.id = "captions"; container.style.cssText = "position:relative;width:640px;height:360px"; @@ -44,6 +46,7 @@ export function captions() { renderer.close(); source.close(); broadcast.close(); + media.close(); sync.close(); producer.close(); origin.close();