diff --git a/doc/bin/gstreamer.md b/doc/bin/gstreamer.md index 99bc063da5..c3695b214d 100644 --- a/doc/bin/gstreamer.md +++ b/doc/bin/gstreamer.md @@ -77,7 +77,8 @@ is at least 1, and a rate is the delta over any window you sample. Unlike Set `encoder=true` on audio and video pads a local encoder feeds (`x264enc`, `opusenc`, ...). The pad then measures how late each frame reaches the sink behind its running time and raises the catalog `jitter` by the spread, -so players buffer for an encoder that delivers irregularly. Leave it off, the +and `delay` by how far it trails the earliest such pad, so players buffer for an +encoder that delivers irregularly or behind the others. Leave it off, the default, for file, demuxed, and network media: their arrival reflects the disk or the network, not the original encoder, and a GStreamer segment cannot tell the two apart. Text and opaque pads refuse it. diff --git a/doc/bin/obs.md b/doc/bin/obs.md index e22801d357..a9733f27d1 100644 --- a/doc/bin/obs.md +++ b/doc/bin/obs.md @@ -35,7 +35,8 @@ OBS Studio install. OBS reports each locally encoded packet's handoff to libmoq against the shared broadcast media clock. Each track's catalog `jitter` is the largest measured -delay above that track's own recent minimum, rounded up to milliseconds. +delay above that track's own recent minimum, and its `delay` is how far that +minimum trails the earliest track, both rounded up to milliseconds. ## Source quality and moq-transcode diff --git a/doc/concept/audio-jitter.md b/doc/concept/audio-jitter.md index eb4c490763..f300399fba 100644 --- a/doc/concept/audio-jitter.md +++ b/doc/concept/audio-jitter.md @@ -271,6 +271,39 @@ is the one thing a measurement cannot be. Note that `js/watch` today adds the two rather than taking the maximum, which over-buffers a bursty publisher by its own flush span. +## Across renditions + +Every track plays from one clock, anchored on the earliest lateness any track +has shown. The estimator measures each track against its own fastest frame, so a +constant offset between tracks cancels out of `measured`: a video encoder that +flushes 200 ms behind the audio encoder produces the same video target as one +that does not, and its frames would all arrive 200 ms late for the shared clock. + +The catalog `delay` field carries that offset: how far a rendition's minimum +flush lateness trails the broadcast's earliest rendition, measured by the +publisher and never lowered. It is an addend, not a floor, because nothing on +this page contains it. The shared playout is the largest requirement among the +renditions actually subscribed: + +``` +playout = max over subscribed renditions of (delay + target) +``` + +A receiver **must not** subtract one rendition's `delay` from another's. Each is +a lifetime maximum taken against a sliding baseline, so two values need not +share an origin: if the earliest rendition later drifts behind another, their +difference understates the real spread. The maximum never under-buffers, +because the subscribed renditions' spread is measured from an earliest baseline +no earlier than the broadcast's, so it never exceeds the largest `delay` among +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. + ## Rise, fall, startup, and the ceiling There is **no separate rise or fall limiter**. The histogram is the only diff --git a/doc/concept/hang.md b/doc/concept/hang.md index c95fa3f2e4..20f3bfc0ed 100644 --- a/doc/concept/hang.md +++ b/doc/concept/hang.md @@ -55,6 +55,7 @@ A few things the catalog can express beyond decoder config: - **Labels.** Any rendition may carry a human-readable `label` for a track picker. The map key stays the track name used to subscribe, so labels need not be unique and renaming one doesn't rename the track. - **Renditions in another broadcast.** A rendition may point at a relative broadcast path, so a transcoder can publish a ladder that adds low rungs and references the source's original rendition without re-publishing its bytes. The path resolves against where the consumer found the catalog, so a reference that escapes above the root names nothing and the catalog is rejected. - **Jitter.** A rendition can say how far its frames fell behind the media clock before the publisher flushed them, in whole milliseconds rounded up. Encoders report the spread of lateness above each rendition's own recent minimum, so a constant encoder delay is not jitter; container imports estimate batch spans without counting ingest delay. It describes the publisher, never the network, only grows over the life of a stream, and a player sizes its buffer to at least this much. A `0` is read as absent. +- **Delay.** A rendition can also say how far its frames reach the transport behind the broadcast's earliest rendition, measured the same way from each rendition's minimum lateness, so a video encoder running 200 ms behind audio advertises `delay: 200` on video. It follows the same rules as jitter. A player holds the largest `delay + jitter` among the renditions it subscribes to, and never subtracts one rendition's `delay` from another's. - **Stalled renditions.** A publisher can flag a rendition as temporarily bad so players prefer another one without the track disappearing. First-party video publishers set this flag after more than three frame intervals of source silence or encoding lag while subscribed, and clear it after three on-time completed frames or when idle. Browser and native capture poll while waiting; FLV and MPEG-TS importers observe video silence as container data arrives. The shared detector is `hang::catalog::stalled::Detector` in Rust and `Catalog.Stalled.Detector` in JavaScript. It is a playback diagnostic, not an authorization or routing signal. - **Archive.** A broadcast may advertise an `archive` entry naming its timeline track (a small index of each complete aligned segment) and, if recorded, the replay MoQ path, object-store URL, and format version. The timeline is what lets the [HLS gateway](/bin/hls) build playlists without subscribing to media. - **Clock.** The optional root `clock` maps PTS zero to wall time so every media track and the archive index share one fixed epoch after timescale conversion. It is independent of `archive`, so a live-only publisher can expose wall-clock timing without creating a segment index. diff --git a/doc/lib/rs/moq-mux.md b/doc/lib/rs/moq-mux.md index 5fe596a4fd..7be69723be 100644 --- a/doc/lib/rs/moq-mux.md +++ b/doc/lib/rs/moq-mux.md @@ -41,9 +41,11 @@ retires the entry. Calling `modify` before the first `set` returns `Error::NotPublished`. Container writes measure bitrate; importers can also measure batch span or reorder delay for jitter. Locally encoded frames call `container::Producer::flush(timestamp, Instant::now())`; jitter is the spread -above that track's own recent minimum lateness, published as soon as it rises. -Generic imports remain clock-free. Invalid or decreasing jitter is rejected -before the edit is retained, including while the initial catalog is reserved. +above that track's own recent minimum lateness, and delay is how far that +minimum trails the earliest track on the same catalog. Both are published as +soon as they rise. Generic imports remain clock-free. Invalid or decreasing +jitter or delay is rejected before the edit is retained, including while the +initial catalog is reserved. Codec importers propagate catalog and media errors through their configuration and frame-writing methods. diff --git a/drafts/draft-lcurley-moq-hang.md b/drafts/draft-lcurley-moq-hang.md index 30d711e3c9..04fffe9264 100644 --- a/drafts/draft-lcurley-moq-hang.md +++ b/drafts/draft-lcurley-moq-hang.md @@ -481,6 +481,7 @@ type CommonExtensions = { "label": string | undefined, "container": Container, "jitter": number | undefined, + "delay": number | undefined, } ~~~ @@ -532,6 +533,19 @@ For example: - A fragment or packet batch contributes the media span between its earliest timestamp and flush point. - Reordered frames contribute the delay they were held before flushing, without treating a decode-order presentation timestamp gap as delay by itself. +### delay {#field-delay} +The maximum amount, in milliseconds, by which a rendition's minimum flush lateness ({{field-jitter}}) has trailed the smallest minimum among the broadcast's renditions that measure it. +If absent, a consumer SHOULD assume the rendition does not trail the others. + +A publisher measures each rendition's minimum over the same recent window it uses for `jitter`, from one clock shared by every rendition, and advertises the largest difference observed. +A container importer does not measure `delay`. +The rounding, `0`, and never-lower rules of `jitter` apply unchanged. + +A consumer SHOULD hold at least the largest `delay` plus `jitter` among the renditions it plays together. +A consumer MUST NOT subtract one rendition's `delay` from another's: each is a maximum over the life of the stream, so two values need not share an origin. + +For example, a video encoder that flushes 200 milliseconds after the audio encoder for the same media time advertises a video `delay` of 200 and no audio `delay`. + # Container {#container} Audio, video, and text tracks use a container to encapsulate the media payload. A rendition declares its container via the `container` field of its catalog entry ({{common}}): @@ -1085,6 +1099,7 @@ A publisher MAY estimate an unknown final duration from the frame cadence, but M - An audio endpoint bounds only the terminal packets that follow it in its own group. - Replaced the archive timeline `wall` field with a root `clock` section (`wall` plus `timescale`): one fixed broadcast mapping every track and the archive index convert into, independent of any archive. Zero timescales and walls past the JSON-safe integer range are refused. - Added optional `bitrate` and `jitter` fields to `json` and `binary` track entries. +- Added the optional `delay` rendition field: how far a rendition's minimum flush lateness trails the broadcast's earliest rendition, never lowered once advertised and never subtracted across renditions. - Recommended namespaced keys for application root sections. # Acknowledgments diff --git a/js/hang/src/catalog/audio.ts b/js/hang/src/catalog/audio.ts index ec7039eedb..f897c9125a 100644 --- a/js/hang/src/catalog/audio.ts +++ b/js/hang/src/catalog/audio.ts @@ -59,6 +59,17 @@ export const AudioConfigSchema = z.object({ z.transform((value) => (value === 0 ? undefined : value)), ), ), + + // How far this rendition's frames reach the transport behind the broadcast's earliest + // rendition, in whole milliseconds rounded up. A player holds `delay + jitter` for it and never + // subtracts one rendition's `delay` from another's. Absent on the earliest rendition. It only + // ever grows over the life of a stream. + delay: z.optional( + z.pipe( + u53Schema, + z.transform((value) => (value === 0 ? undefined : value)), + ), + ), }); /** Schema for the catalog audio section: a map of track name to rendition config. */ diff --git a/js/hang/src/catalog/root.test.ts b/js/hang/src/catalog/root.test.ts index a62dbe1181..c37154805c 100644 --- a/js/hang/src/catalog/root.test.ts +++ b/js/hang/src/catalog/root.test.ts @@ -1,6 +1,7 @@ import { expect, test } from "bun:test"; import * as z from "@zod/mini"; import { ARCHIVE_VERSION } from "./archive.ts"; +import { u53 } from "./integers.ts"; import type { RelativeBroadcast } from "./path.ts"; import { RootSchema } from "./root.ts"; @@ -95,6 +96,29 @@ test("legacy zero jitter is absent for audio and video", () => { expect(JSON.stringify(parsed)).not.toContain('"jitter"'); }); +test("delay parses beside jitter and zero is absent", () => { + const parsed = RootSchema.parse({ + audio: { + renditions: { + audio: { + codec: "opus", + container: { kind: "legacy" }, + sampleRate: 48000, + numberOfChannels: 2, + delay: 0, + }, + }, + }, + video: { + renditions: { video: { codec: "avc1.64001f", container: { kind: "legacy" }, jitter: 34, delay: 200 } }, + }, + text: { renditions: { captions: { format: "vtt", container: { kind: "legacy" }, delay: 120 } } }, + }); + expect(parsed.audio?.renditions.audio?.delay).toBeUndefined(); + expect(parsed.video?.renditions.video?.delay).toBe(u53(200)); + expect(parsed.text?.renditions.captions?.delay).toBe(u53(120)); +}); + test("clock round-trips at the root", () => { const parsed = RootSchema.parse({ clock: { wall: 1_751_846_400_000_000, timescale: 1_000_000 }, diff --git a/js/hang/src/catalog/text.ts b/js/hang/src/catalog/text.ts index 3bd4e3dc21..5be0bee11b 100644 --- a/js/hang/src/catalog/text.ts +++ b/js/hang/src/catalog/text.ts @@ -58,6 +58,17 @@ export const TextConfigSchema = z.object({ z.transform((value) => (value === 0 ? undefined : value)), ), ), + + // How far this rendition's frames reach the transport behind the broadcast's earliest + // rendition, in whole milliseconds rounded up. A player holds `delay + jitter` for it and never + // subtracts one rendition's `delay` from another's. Absent on the earliest rendition. It only + // ever grows over the life of a stream. + delay: z.optional( + z.pipe( + u53Schema, + z.transform((value) => (value === 0 ? undefined : value)), + ), + ), }); /** Schema for the catalog text section: a map of track name to rendition config. */ diff --git a/js/hang/src/catalog/video.ts b/js/hang/src/catalog/video.ts index 2d173e1758..53c9946a7f 100644 --- a/js/hang/src/catalog/video.ts +++ b/js/hang/src/catalog/video.ts @@ -73,6 +73,17 @@ export const VideoConfigSchema = z.object({ z.transform((value) => (value === 0 ? undefined : value)), ), ), + + // How far this rendition's frames reach the transport behind the broadcast's earliest + // rendition, in whole milliseconds rounded up. A player holds `delay + jitter` for it and never + // subtracts one rendition's `delay` from another's. Absent on the earliest rendition. It only + // ever grows over the life of a stream. + delay: z.optional( + z.pipe( + u53Schema, + z.transform((value) => (value === 0 ? undefined : value)), + ), + ), }); /** diff --git a/js/msf/src/catalog.test.ts b/js/msf/src/catalog.test.ts index 749b18819d..a6e16a222b 100644 --- a/js/msf/src/catalog.test.ts +++ b/js/msf/src/catalog.test.ts @@ -185,6 +185,30 @@ test("preserves SAP fields through decode and encode", () => { expect(wireTracks[0].jitter).toBe(15); }); +test("preserves delay through decode and encode", () => { + const catalog = decode( + encodeJson({ + version: "draft-01", + tracks: [ + { + name: "video0", + packaging: "loc", + isLive: true, + role: "video", + codec: "avc1.640028", + delay: 200, + }, + ], + }), + ); + + expect(catalog.tracks[0].delay).toBe(200); + + const wire = decodeJson(encode(catalog)); + const wireTracks = wire.tracks as { delay?: number }[]; + expect(wireTracks[0].delay).toBe(200); +}); + test.each([ ["omitted", undefined], ["false", false], diff --git a/js/msf/src/catalog.ts b/js/msf/src/catalog.ts index a11cbca452..9596adbee0 100644 --- a/js/msf/src/catalog.ts +++ b/js/msf/src/catalog.ts @@ -48,6 +48,10 @@ const trackShape = { // The player's buffer must be at least this large to avoid underruns. // Mirrors the `jitter` field in the hang catalog. jitter: z.optional(z.number()), + + // 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()), }; /** Zod schema describing a single track entry in an MSF catalog. */ diff --git a/js/publish/src/catalog.test.ts b/js/publish/src/catalog.test.ts index cd76a88781..215da4f89c 100644 --- a/js/publish/src/catalog.test.ts +++ b/js/publish/src/catalog.test.ts @@ -117,10 +117,37 @@ test("a reconnecting subscriber is seeded with the full current catalog", async effect.close(); }); -test("catalog producer refuses zero jitter before retaining an edit", () => { - const catalog = new CatalogProducer(); +for (const field of ["jitter", "delay"] as const) { + test(`catalog producer refuses zero ${field} before retaining an edit`, () => { + const catalog = new CatalogProducer(); + for (const section of ["audio", "video", "text"] as const) { + expect(() => + catalog.mutate((value) => { + Object.assign(value, { + [section]: { + renditions: { + media: { + codec: "opus", + container: { kind: "legacy" }, + sampleRate: 48000, + numberOfChannels: 2, + [field]: 0, + }, + }, + }, + }); + }), + ).toThrow(`omit ${field}`); + } + catalog.mutate((value) => { + expect(value.audio).toBeUndefined(); + expect(value.video).toBeUndefined(); + }); + }); + for (const section of ["audio", "video", "text"] as const) { - expect(() => + test(`catalog refuses ${section} ${field} decreases without retaining them`, () => { + const catalog = new CatalogProducer(); catalog.mutate((value) => { Object.assign(value, { [section]: { @@ -130,70 +157,45 @@ test("catalog producer refuses zero jitter before retaining an edit", () => { container: { kind: "legacy" }, sampleRate: 48000, numberOfChannels: 2, - jitter: 0, + [field]: 100, }, }, }, }); - }), - ).toThrow("omit jitter"); - } - catalog.mutate((value) => { - expect(value.audio).toBeUndefined(); - expect(value.video).toBeUndefined(); - }); -}); - -for (const section of ["audio", "video", "text"] as const) { - test(`catalog refuses ${section} jitter decreases without retaining them`, () => { - const catalog = new CatalogProducer(); - catalog.mutate((value) => { - Object.assign(value, { - [section]: { - renditions: { - media: { - codec: "opus", - container: { kind: "legacy" }, - sampleRate: 48000, - numberOfChannels: 2, - jitter: 100, - }, - }, - }, }); - }); - // The section is optional on the loose root type, so re-read it through a guard. - const retained = (value: Catalog.Root) => { - const sectionValue = value[section]; - if (!sectionValue) throw new Error(`expected a retained ${section} section`); - return sectionValue; - }; - for (const jitter of [Catalog.u53(50), undefined]) { - expect(() => + // The section is optional on the loose root type, so re-read it through a guard. + const retained = (value: Catalog.Root) => { + const sectionValue = value[section]; + if (!sectionValue) throw new Error(`expected a retained ${section} section`); + return sectionValue; + }; + for (const estimate of [Catalog.u53(50), undefined]) { + expect(() => + catalog.mutate((value) => { + retained(value).renditions.media[field] = estimate; + }), + ).toThrow(`${field} cannot decrease`); catalog.mutate((value) => { - retained(value).renditions.media.jitter = jitter; - }), - ).toThrow("jitter cannot decrease"); + expect(retained(value).renditions.media[field]).toBe(Catalog.u53(100)); + }); + } catalog.mutate((value) => { - expect(retained(value).renditions.media.jitter).toBe(Catalog.u53(100)); + delete retained(value).renditions.media; }); - } - catalog.mutate((value) => { - delete retained(value).renditions.media; - }); - catalog.mutate((value) => { - Object.assign(retained(value).renditions, { - media: { - codec: "opus", - container: { kind: "legacy" }, - sampleRate: 48000, - numberOfChannels: 2, - jitter: 50, - }, + catalog.mutate((value) => { + Object.assign(retained(value).renditions, { + media: { + codec: "opus", + container: { kind: "legacy" }, + sampleRate: 48000, + numberOfChannels: 2, + [field]: 50, + }, + }); }); }); - }); + } } for (const section of ["json", "binary"] as const) { diff --git a/js/publish/src/catalog.ts b/js/publish/src/catalog.ts index 3a40e3d1ae..225a4e4715 100644 --- a/js/publish/src/catalog.ts +++ b/js/publish/src/catalog.ts @@ -24,13 +24,15 @@ export class CatalogProducer { mutate(fn: (catalog: Catalog.Root) => void): void { const value = structuredClone(this.#value); fn(value); - for (const [section, next] of Object.entries(jitters(value))) { - const previous = jitters(this.#value)[section]; - for (const [name, jitter] of Object.entries(next)) { - if (jitter === 0) throw new Error("omit jitter for a track flushed immediately"); - const before = previous?.[name]; - if (before !== undefined && (jitter === undefined || jitter < before)) { - throw new Error("jitter cannot decrease for an existing track"); + for (const field of ["jitter", "delay"] as const) { + const previous = advertised(this.#value, field); + for (const [section, next] of Object.entries(advertised(value, field))) { + for (const [name, estimate] of Object.entries(next)) { + if (estimate === 0) throw new Error(`omit ${field} rather than advertising 0`); + const before = previous[section]?.[name]; + if (before !== undefined && (estimate === undefined || estimate < before)) { + throw new Error(`${field} cannot decrease for an existing track`); + } } } } @@ -60,10 +62,13 @@ export class CatalogProducer { } } -/** Every track's advertised jitter, by section and then track name. */ -function jitters(catalog: Catalog.Root): Record> { - const pick = (tracks: Record | undefined) => - Object.fromEntries(Object.entries(tracks ?? {}).map(([name, config]) => [name, config.jitter])); +/** Every track's advertised `field`, by section and then track name. */ +function advertised( + catalog: Catalog.Root, + field: "jitter" | "delay", +): Record> { + const pick = (tracks: Record | undefined) => + Object.fromEntries(Object.entries(tracks ?? {}).map(([name, config]) => [name, config[field]])); return { audio: pick(catalog.audio?.renditions), video: pick(catalog.video?.renditions), diff --git a/js/watch/src/audio/config.test.ts b/js/watch/src/audio/config.test.ts index 36ee450983..ca5dee3a23 100644 --- a/js/watch/src/audio/config.test.ts +++ b/js/watch/src/audio/config.test.ts @@ -68,6 +68,11 @@ test("an advertised jitter of zero falls back to the codec frame duration", () = expect(playbackJitter(config({ jitter: 60 }))).toBe(Time.Milli(63)); }); +test("a rendition's delay adds to its jitter", () => { + expect(playbackJitter(config({ delay: 200 }))).toBe(Time.Milli(223)); + expect(playbackJitter(config({ delay: 200, jitter: 60 }))).toBe(Time.Milli(263)); +}); + test("AAC and MP3 jitter follows their codec frame sizes", () => { expect(playbackJitter(config({ codec: "mp4a.40.2", sampleRate: 48000 }))).toBe(Time.Milli(25)); expect(playbackJitter(config({ codec: "mp4a.40.2", sampleRate: 24000 }))).toBe(Time.Milli(49)); diff --git a/js/watch/src/audio/config.ts b/js/watch/src/audio/config.ts index e5ae6d5fdc..2665ce83d8 100644 --- a/js/watch/src/audio/config.ts +++ b/js/watch/src/audio/config.ts @@ -40,7 +40,10 @@ export function playbackIdentity(config: Catalog.AudioConfig): PlaybackIdentity }; } -/** The jitter to add to the sync buffer for a rendition, in milliseconds. */ +/** + * The sync buffer a rendition needs, in milliseconds: its catalog `delay` behind the broadcast's + * earliest rendition plus its own jitter. + */ export function playbackJitter(config: Catalog.AudioConfig): Time.Milli { // A publisher advertising 0 is claiming frames are never delayed, which no encoder can do, so // fall back to the codec's frame duration the same way an absent field does. @@ -48,7 +51,7 @@ export function playbackJitter(config: Catalog.AudioConfig): Time.Milli { // Add the worklet render quantum so the ring buffer has margin between frame arrivals. const overhead = Math.ceil((WORKLET_QUANTUM / config.sampleRate) * 1000); - return Time.Milli(codecJitter + overhead); + return Time.Milli((config.delay ?? 0) + codecJitter + overhead); } // Estimate the minimum jitter (frame duration) based on the audio codec. diff --git a/js/watch/src/msf.test.ts b/js/watch/src/msf.test.ts index dd2e930259..2a58e4d815 100644 --- a/js/watch/src/msf.test.ts +++ b/js/watch/src/msf.test.ts @@ -1,7 +1,67 @@ import { expect, test } from "bun:test"; +import { u53 } from "@moq/hang/catalog"; import type * as Msf from "@moq/msf"; import { toHang } from "./msf"; +test("copies delay onto the hang rendition", () => { + const catalog: Msf.Catalog = { + tracks: [ + { + name: "video", + packaging: "loc", + role: "video", + codec: "vp09.00.10.08", + delay: 200, + jitter: 40, + }, + { + name: "audio", + packaging: "loc", + role: "audio", + codec: "opus", + delay: 80, + }, + ], + }; + + expect(toHang(catalog).video?.renditions.video?.delay).toBe(u53(200)); + expect(toHang(catalog).video?.renditions.video?.jitter).toBe(u53(40)); + expect(toHang(catalog).audio?.renditions.audio?.delay).toBe(u53(80)); +}); + +test("rounds a fractional MSF delay up, and drops zero", () => { + const catalog: Msf.Catalog = { + tracks: [ + { + name: "video", + packaging: "loc", + role: "video", + codec: "vp09.00.10.08", + delay: 200.2, + }, + { + name: "audio", + packaging: "loc", + role: "audio", + codec: "opus", + delay: 0.2, + }, + { + name: "early", + packaging: "loc", + role: "audio", + codec: "opus", + delay: 0, + }, + ], + }; + + const hang = toHang(catalog); + 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(); +}); + test("preserves stalled video renditions", () => { const catalog: Msf.Catalog = { tracks: [ diff --git a/js/watch/src/msf.ts b/js/watch/src/msf.ts index cc7271ef66..a78a396ec5 100644 --- a/js/watch/src/msf.ts +++ b/js/watch/src/msf.ts @@ -46,6 +46,13 @@ 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. +function delayMillis(value: number): ReturnType | undefined { + if (value <= 0) return undefined; + return u53(Math.ceil(value)); +} + function toVideoConfig(track: Msf.Track): Catalog.VideoConfig | undefined { if (!track.codec) return undefined; @@ -62,6 +69,7 @@ function toVideoConfig(track: Msf.Track): Catalog.VideoConfig | undefined { bitrate: track.bitrate != null ? u53(track.bitrate) : undefined, stalled: track.stalled, jitter: track.jitter != null ? u53(track.jitter) : undefined, + delay: track.delay != null ? delayMillis(track.delay) : undefined, }; } @@ -85,6 +93,7 @@ function toAudioConfig(track: Msf.Track): Catalog.AudioConfig | undefined { numberOfChannels: u53(channels), bitrate: track.bitrate != null ? u53(track.bitrate) : undefined, jitter: track.jitter != null ? u53(track.jitter) : undefined, + delay: track.delay != null ? delayMillis(track.delay) : undefined, }; } diff --git a/js/watch/src/sync.test.ts b/js/watch/src/sync.test.ts index d8d18764b8..f0c068ebdd 100644 --- a/js/watch/src/sync.test.ts +++ b/js/watch/src/sync.test.ts @@ -1,7 +1,10 @@ import { describe, expect, it } from "bun:test"; +import type * as Catalog from "@moq/hang/catalog"; import { Time } from "@moq/net"; import { Signal } from "@moq/signals"; +import { playbackJitter } from "./audio/config"; import { type Delay, Sync } from "./sync"; +import { renditionJitter } from "./video/playhead"; // Effects in @moq/signals flush on a microtask, so let pending updates drain before asserting. const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); @@ -80,6 +83,23 @@ describe("delay and buffer", () => { sync.close(); }); + it("holds the largest delay plus jitter among subscribed renditions", async () => { + const audio = { codec: "opus", container: { kind: "legacy" }, sampleRate: 48000, numberOfChannels: 2 }; + const video = { codec: "avc1.640028", container: { kind: "legacy" }, jitter: 34, delay: 200 }; + const sync = new Sync({ delay: 100 as Time.Milli }); + sync.register(new Signal(playbackJitter(audio as Catalog.AudioConfig))); + const unsubscribe = sync.register(new Signal(renditionJitter(video as Catalog.VideoConfig))); + await flush(); + // 100 ms of network jitter over the video's 200 ms delay and 34 ms spread. + expect(sync.out.delay.peek()).toBe(334 as Time.Milli); + + // Dropping the slow rendition lowers latency to the audio's own 20 ms frame and 3 ms quantum. + unsubscribe(); + await flush(); + expect(sync.out.delay.peek()).toBe(123 as Time.Milli); + sync.close(); + }); + it("unregisters duplicate jitter inputs independently", async () => { const media = new Signal(20 as Time.Milli); const sync = new Sync({ delay: 100 as Time.Milli }); diff --git a/js/watch/src/video/decoder.test.ts b/js/watch/src/video/decoder.test.ts new file mode 100644 index 0000000000..fac3a7bbde --- /dev/null +++ b/js/watch/src/video/decoder.test.ts @@ -0,0 +1,149 @@ +import { describe, expect, it } from "bun:test"; +import * as Catalog from "@moq/hang/catalog"; +import type * as Moq from "@moq/net"; +import { Time } from "@moq/net"; +import { Signal } from "@moq/signals"; +import type { Broadcast } from "../broadcast"; +import { Sync } from "../sync"; +import { Decoder } from "./decoder"; +import { Source } from "./source"; + +const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); + +async function settle(): Promise { + for (let i = 0; i < 8; i++) await flush(); +} + +function config(fields: Record): Catalog.VideoConfig { + return Catalog.VideoConfigSchema.parse({ + codec: "avc1.640028", + container: { kind: "legacy" }, + ...fields, + }); +} + +// The subscription is already closed, so the track never reaches VideoDecoder. +function closedConsumer(): Moq.Broadcast.Consumer { + return { closed: new Signal("ended") } as unknown as Moq.Broadcast.Consumer; +} + +function watchBroadcast(catalog: Signal): Broadcast { + return { + out: { catalog }, + relativeBroadcast: () => closedConsumer(), + } as unknown as Broadcast; +} + +async function play(catalog: Signal): Promise<{ + broadcast: Signal; + source: Source; + sync: Sync; + decoder: Decoder; +}> { + const broadcast = new Signal(watchBroadcast(catalog)); + const source = new Source({ + broadcast, + supported: async () => true, + }); + const sync = new Sync({ delay: Time.Milli(0) }); + await settle(); + const decoder = new Decoder({ source, sync }); + await settle(); + return { broadcast, source, sync, decoder }; +} + +describe("Decoder jitter across a source switch", () => { + it("keeps the outgoing floor when the next source reuses the track name", async () => { + const outgoing = new Signal({ + video: { renditions: { video: config({ delay: 200, jitter: 60 }) } }, + }); + const { broadcast, source, sync, decoder } = await play(outgoing); + try { + expect(decoder.out.jitter.peek()).toBe(Time.Milli(260)); + + // A rise on the catalog this rendition subscribed to still resizes the floor. + outgoing.set({ + video: { renditions: { video: config({ delay: 300, jitter: 60 }) } }, + }); + await settle(); + expect(decoder.out.jitter.peek()).toBe(Time.Milli(360)); + + broadcast.set( + watchBroadcast( + new Signal({ + video: { renditions: { video: config({ jitter: 20 }) } }, + }), + ), + ); + await settle(); + // The old frames are still on screen. The new catalog's same name is not their floor. + expect(decoder.out.jitter.peek()).toBe(Time.Milli(360)); + } finally { + decoder.close(); + source.close(); + sync.close(); + } + }); + + it("follows a newer pending catalog when the track name stays the same", async () => { + const outgoing = new Signal({ + video: { renditions: { video: config({ jitter: 20 }) } }, + }); + const { broadcast, source, sync, decoder } = await play(outgoing); + try { + expect(decoder.out.jitter.peek()).toBe(Time.Milli(20)); + + broadcast.set( + watchBroadcast( + new Signal({ + video: { renditions: { video: config({ delay: 200 }) } }, + }), + ), + ); + await settle(); + expect(decoder.out.jitter.peek()).toBe(Time.Milli(200)); + + const newest = new Signal({ + video: { renditions: { video: config({ delay: 500 }) } }, + }); + broadcast.set(watchBroadcast(newest)); + await settle(); + // The first switch is still pending, so the name did not change. The new catalog still has to win. + expect(decoder.out.jitter.peek()).toBe(Time.Milli(500)); + + newest.set({ + video: { renditions: { video: config({ delay: 700 }) } }, + }); + await settle(); + expect(decoder.out.jitter.peek()).toBe(Time.Milli(700)); + } finally { + decoder.close(); + source.close(); + sync.close(); + } + }); + + it("keeps the outgoing floor when the next source uses a different name", async () => { + const outgoing = new Signal({ + video: { renditions: { hd: config({ delay: 200, jitter: 60 }) } }, + }); + const { broadcast, source, sync, decoder } = await play(outgoing); + try { + expect(decoder.out.jitter.peek()).toBe(Time.Milli(260)); + + broadcast.set( + watchBroadcast( + new Signal({ + video: { renditions: { sd: config({ jitter: 20 }) } }, + }), + ), + ); + await settle(); + expect(decoder.out.jitter.peek()).toBe(Time.Milli(260)); + } finally { + decoder.close(); + source.close(); + sync.close(); + } + }); +}); diff --git a/js/watch/src/video/decoder.ts b/js/watch/src/video/decoder.ts index c6319ff5e9..592eb898d3 100644 --- a/js/watch/src/video/decoder.ts +++ b/js/watch/src/video/decoder.ts @@ -92,7 +92,10 @@ export class Decoder { // The current track running, held so we can cancel it when the new track is ready. #active = new Signal(undefined); - #pendingJitter = new Signal(undefined); + // The rendition preparing to replace the active one, with the catalog of the broadcast it + // subscribed to. One value so a later source that reuses the track name still notifies: + // the name alone would compare equal, and the jitter effect would keep the previous catalog. + #pending = new Signal<{ track: string; catalog: Getter } | undefined>(undefined); readonly #identity: Computed; #signals = new Effect(); @@ -125,9 +128,20 @@ export class Decoder { this.#signals.run(this.#runBuffering.bind(this)); } + // Read from the full catalog entry, which the decoder config omits, so a rising entry resizes Sync. + // Each in-flight rendition stays tied to the catalog it subscribed to: during a source switch the + // shared catalog already describes the next broadcast, and the same track name there is a + // different rendition. #runJitter(effect: Effect): void { - const active = effect.get(this.#active)?.jitter; - const pending = effect.get(this.#pendingJitter); + const floor = (track: string | undefined, catalog: Getter | undefined) => { + if (track === undefined || catalog === undefined) return undefined; + const config = effect.get(catalog)?.video?.renditions?.[track]; + return config && renditionJitter(config); + }; + const activeTrack = effect.get(this.#active); + const active = floor(activeTrack?.track, activeTrack?.catalog); + const pendingTrack = effect.get(this.#pending); + const pending = floor(pendingTrack?.track, pendingTrack?.catalog); effect.set(this.#out.jitter, switchJitter({ active, pending })); } @@ -164,8 +178,9 @@ export class Decoder { track, config: identity.decoder, stats: this.#out.stats, + catalog: broadcast.out.catalog, }); - effect.set(this.#pendingJitter, pending.jitter); + effect.set(this.#pending, { track, catalog: broadcast.out.catalog }); effect.cleanup(() => pending?.close()); @@ -185,7 +200,7 @@ export class Decoder { // Upgrade the pending track to active. // #runActive will be in charge of it now. this.#active.set(pending); - this.#pendingJitter.set(undefined); + this.#pending.set(undefined); pending = undefined; // This effect is done; close it to avoid a useless re-run. @@ -267,6 +282,9 @@ interface DecoderTrackProps { broadcast: Moq.Broadcast.Consumer; track: string; config: DecoderConfig; + // The broadcast catalog this subscription started from. Updates on that broadcast still apply; + // a later source does not. + catalog: Getter; stats: Signal; } @@ -276,8 +294,8 @@ class DecoderTrack { broadcast: Moq.Broadcast.Consumer; track: string; config: DecoderConfig; + catalog: Getter; stats: Signal; - jitter: Time.Milli | undefined; timestamp = new Signal(undefined); frame = new Signal(undefined); @@ -299,8 +317,8 @@ class DecoderTrack { this.broadcast = props.broadcast; this.track = props.track; this.config = props.config; + this.catalog = props.catalog; this.stats = props.stats; - this.jitter = renditionJitter(props.config); this.#signals.run(this.#run.bind(this)); } diff --git a/js/watch/src/video/playhead.test.ts b/js/watch/src/video/playhead.test.ts index 7c701a46a0..08c5fef605 100644 --- a/js/watch/src/video/playhead.test.ts +++ b/js/watch/src/video/playhead.test.ts @@ -4,7 +4,7 @@ import type { Time } from "@moq/net"; import { caughtUp, renditionJitter, switchJitter } from "./playhead"; // `jitter` is a branded u53 in the catalog schema, so build the config through a cast. -function config(props: { jitter?: number; framerate?: number }): Catalog.VideoConfig { +function config(props: { jitter?: number; delay?: number; framerate?: number }): Catalog.VideoConfig { return { codec: "avc1.640028", container: { kind: "legacy" }, ...props } as Catalog.VideoConfig; } @@ -26,6 +26,12 @@ describe("renditionJitter", () => { it("is undefined when the catalog declares neither", () => { expect(renditionJitter(config({}))).toBeUndefined(); }); + + it("adds the delay behind the earliest rendition", () => { + expect(renditionJitter(config({ delay: 200, jitter: 60 }))).toBe(ms(260)); + expect(renditionJitter(config({ delay: 200, framerate: 30 }))).toBe(ms(234)); + expect(renditionJitter(config({ delay: 200 }))).toBe(ms(200)); + }); }); describe("caughtUp", () => { diff --git a/js/watch/src/video/playhead.ts b/js/watch/src/video/playhead.ts index 7f341d3c36..851f645190 100644 --- a/js/watch/src/video/playhead.ts +++ b/js/watch/src/video/playhead.ts @@ -9,15 +9,19 @@ const SLACK = Time.Milli(100); /** * How far behind live a rendition's playhead can sit while still being at its own live edge. * - * Frames arrive one group at a time, so this is the rendition's group cadence: the sync buffer - * has to cover it or playback starves between groups. The catalog value wins. Otherwise assume - * the publisher flushes each frame as it's encoded, so a frame interval is the longest we wait. - * Undefined when the catalog declares neither. + * Its catalog `delay` behind the broadcast's earliest rendition, plus its own spread: frames arrive + * one group at a time, so the sync buffer has to cover the group cadence or playback starves + * between groups. The catalog `jitter` wins. Otherwise assume the publisher flushes each frame as + * it's encoded, so a frame interval is the longest we wait. Undefined when the catalog declares + * none of them. */ export function renditionJitter(config: Catalog.VideoConfig): Time.Milli | undefined { - if (config.jitter !== undefined) return Time.Milli(config.jitter); - if (config.framerate) return Time.Milli(Math.ceil(1000 / config.framerate)); - return undefined; + let spread: Time.Milli | undefined; + if (config.jitter !== undefined) spread = Time.Milli(config.jitter); + else if (config.framerate) spread = Time.Milli(Math.ceil(1000 / config.framerate)); + + if (config.delay === undefined) return spread; + return Time.Milli.add(Time.Milli(config.delay), spread ?? Time.Milli.zero); } /** The playheads involved in promoting a new rendition. */ diff --git a/quest/m0/audio-jitter-target/README.md b/quest/m0/audio-jitter-target/README.md index cf59623b4b..9f9600f3e1 100644 --- a/quest/m0/audio-jitter-target/README.md +++ b/quest/m0/audio-jitter-target/README.md @@ -71,7 +71,6 @@ buffer against uneven arrivals. ## Related -- [Jitter clock](/quest/m1/jitter-flush-clock.md) - the advertised jitter (#3513 landed the flush span), which `doc/concept/audio-jitter.md` settles as a floor on the measured target - [Audio quality harness](/quest/m1/audio-quality-harness/README.md) - the automated proof, built on its own schedule - [Time stretch](/quest/m1/watch-audio-time-stretch.md) - inaudible convergence, on top of this - [Plan: A/V clock](/quest/m0/plan-av-clock.md) - the clock this target eventually feeds diff --git a/quest/m1/README.md b/quest/m1/README.md index 5aef7064c5..b5fb56fec6 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -39,7 +39,7 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Optional max age](/quest/m1/ietf-max-age.md) - max age is optional, set only by the publisher, and crosses moq-transport as MAX_CACHE_DURATION - [IETF announce count](/quest/m1/ietf-announce-count.md) - an opt-in moq-transport extension carries the replay count, so IETF announce consumers go live without a timer -- [Jitter clock](/quest/m1/jitter-flush-clock.md) - renditions advertise `delay` (lag behind the earliest track) and `jitter` (spread), measured at encoder flush, never lowered; js/watch sizes playout over what it subscribes +- [Publish delay](/quest/m1/publish-delay.md) - js/publish encoders advertise `delay` behind the earliest rendition, like moq-mux - [Import discontinuity](/quest/m1/import-discontinuity.md) - a seek or pause resets the flush jitter baseline, from moqsink, libmoq, and moq-ffi - [Data jitter](/quest/m1/data-jitter.md) - JSON and binary tracks with a capture time advertise a detected `delay` and `jitter` - [Play tune-in backpressure](/quest/m1/play-tunein-backpressure.md) - moq play: a tune-in burst larger than the video queue parks the decoder, so the clock never reaches live at a wide `--delay` diff --git a/quest/m1/data-jitter.md b/quest/m1/data-jitter.md index e21e11ca3b..3ccf0386a4 100644 --- a/quest/m1/data-jitter.md +++ b/quest/m1/data-jitter.md @@ -5,8 +5,8 @@ A JSON or binary track whose payloads carry a capture time on the publisher's own clock (a UDP datagram's arrival, a sensor read) advertises a detected `delay` and `jitter`: how late the publisher hands payloads to the transport -relative to that time, the same measurement -[jitter clock](/quest/m1/jitter-flush-clock.md) defines for encoders. A +relative to that time, the same measurement `moq_mux::catalog::Estimator` +makes for encoders. A telemetry source slower than the video it accompanies shows up as `delay`. Tracks written without a capture time advertise neither rather than a meaningless zero. @@ -45,6 +45,3 @@ Public API: additive on `hang`, `moq-binary`, `moq-json`, `moq-mux`, `@moq/hang`, `@moq/binary`, and `@moq/json`. Wire: one optional field on data entries. -## Required - -- [Jitter clock](/quest/m1/jitter-flush-clock.md) - defines the flush-lateness measurement and `delay` diff --git a/quest/m1/jitter-flush-clock.md b/quest/m1/jitter-flush-clock.md deleted file mode 100644 index e83e361955..0000000000 --- a/quest/m1/jitter-flush-clock.md +++ /dev/null @@ -1,113 +0,0 @@ -# [L] moq-mux: catalog delay and jitter measure how far behind the media clock an encoder flushes - -## Goal - -An original publisher advertises two numbers per rendition, both measured -from the gap between a frame's media timestamp and the wall-clock moment it -handed that frame to the transport (its lateness), the same lateness -`js/watch` `sync.ts` computes on receive, measured one hop earlier: - -- `delay`: the rendition's minimum lateness behind the broadcast's earliest - rendition. A video encoder running 200 ms behind the audio encoder - advertises `delay: 200` on video and none on audio. -- `jitter`: the spread of the rendition's lateness above its own minimum. A - fragment flushed at its end, a B-frame held for reordering, and a - latency-buffered batch all raise it without being special cases. - -`delay + jitter` is the worst-case lateness, and neither value is ever -lowered once advertised. `js/watch` sizes playout over the renditions it -subscribes to as `max(delay + jitter) - min(delay)` plus network jitter, and -resizes when that set changes, so dropping a slow track lowers latency. - -Only encoders feed the clock; file and pipe imports and the RTMP, SRT, and TS -gateways never do, so an unstable ingest link cannot inflate the catalog. TS -and fragmented MP4 imports retain their clock-free jitter estimates from the -media span of each emitted batch and advertise no `delay`. - -## Plan - -- **Landed (#3940):** `jitter` measured at encoder flush. `Estimator::flush`, - `container::Producer::flush`, and codec importer forwarding; each rendition - keeps its own 10 s sliding minimum and advertises the lifetime maximum spread - above it. The `moq-video` and `moq-audio` encoders (so `moq import capture`), - libmoq (so OBS), moq-ffi and its wrappers, and the `js/publish` encoders call - it. The provisional PTS-gap floor is gone, and `moq_mux::Error::JitterDecreased` - plus zero-as-absent text jitter enforce never-lower in Rust and JS. - `moq-gst` pads opt in with `encoder=true`; imports stay clock-free. - What remains below is `delay` and the player. A seek or pause resetting the - baseline from moqsink and the bindings is - [Import discontinuity](/quest/m1/import-discontinuity.md), including a - `moq-gst` encoder pad across a `PLAYING -> PAUSED -> PLAYING` cycle (running - time stops, the wall clock does not) and a flushing seek. -- **Measurement.** Lateness is `now - timestamp`, observed by the existing - `flush` calls, so no call site changes. Each rendition keeps its own - baseline, the minimum lateness over a sliding window (about 10 s), so a media - clock that drifts slower than wall time does not ratchet forever. The - broadcast baseline, on `catalog::Producer` and shared by every rendition - that flushes, is the minimum of those; the per-rendition `Baseline` epochs - must become one shared epoch for the subtraction to mean anything. `delay` is - the rendition baseline minus the broadcast baseline, reported as its lifetime - maximum. Renditions that never flush advertise no `delay`. -- **Open:** lifetime maxima taken against a sliding baseline stop sharing an - origin when the earliest rendition changes. If A starts at 0 and B at - 200 ms, B keeps `delay: 200`; if A then drifts to 500 ms, A advertises 300 - and `Sync` computes `300 - 200 = 100` while the tracks are 300 ms apart. - Options: - - *No subtraction (recommended).* Keep the sliding baselines and never-lower, - and change the player rule to `max(delay + jitter)` over the subscribed - renditions, dropping `- min(delay)`. The subscribed renditions' true spread - is measured from an earliest subscribed baseline no earlier than the - broadcast baseline, so it never exceeds the largest advertised `delay`, - and the catalog alone can never under-buffer. The cost is over-buffering by - `min(delay)` when the broadcast's earliest rendition is not subscribed - (a video-only viewer of a broadcast whose audio leads by 200 ms pays - 200 ms). Dropping a slow track still lowers latency. The draft says a - consumer MUST NOT subtract `delay` values across renditions. - - *Fixed common origin.* Measure every `delay` from the broadcast's first - lateness. Exact subtraction, but common drift raises every rendition - together, so values grow without bound and the catalog republishes for the - life of the broadcast; the problem the sliding window exists to avoid. - - *Coordinated rebasing.* Advertise each `delay` as its current value against - the current broadcast baseline and let it fall. Exact, but `delay` gives up - never-lower, the catalog churns as baselines move, and the player must - shrink safely. - - *No `delay` field.* The player measures each subscribed track's own - arrival baseline and sizes by their spread. No wire change, and it also - covers gateway and ingest offsets, but a track's offset is unknown until its - first frames arrive, so subscribing to a slower rendition glitches once. - Every option also needs the player's reference to follow the earliest - subscribed track's current arrival rather than its lifetime minimum, since - `Sync.received` only ever lowers it; that is receiver-side and belongs with - the arrival minimum the [audio jitter target](/quest/m0/audio-jitter-target/README.md) - already expires. -- A faster-than-real-time source flushes early; each frame becomes the new - minimum and both stay at zero, which is correct for something that is not - live. -- **Wire.** Add optional `delay` beside `jitter` on video, audio, and text - renditions in `rs/hang`, `js/hang`, and `drafts/draft-lcurley-moq-hang.md`, - serialized with `MillisCeil` and zero-as-absent like `jitter`. Additive: - today's `jitter` never included a cross-track offset. Extend the draft's - never-lower rule to `delay` (unless rebasing wins), along with - `moq_mux::Error::JitterDecreased` and the `js/publish/src/catalog.ts` check. -- **Player.** `Sync` registers each subscribed rendition's `delay` and - `jitter` and recomputes when a rendition registers, unregisters, or its - catalog entry rises. Rename `Sync`'s own `delay` output (the resolved - playout total) so it does not collide with the field. #3954 on the - [audio jitter target](/quest/m0/audio-jitter-target/README.md) line changes - what `register()` takes, so build on whichever lands first. -- **Docs.** Update `doc/concept/audio-jitter.md`, the normative playout page, - with the cross-track rule and the catalog floor it reads. -- **Tests** inject the clock. Cover: a constant offset reports as `delay` on - the slower track and nothing on the faster, a slow common drift stays - bounded, the earliest rendition changing does not under-buffer, a decrease is - refused on every site, and `Sync` resizes on a subscription change. - -Public API: additive `delay` field in `hang` and `@moq/hang`, the -decreased-estimate error extended to `delay`, `Sync` output rename in -`@moq/watch`. Wire: one optional catalog field. - -## Related - -- [Audio jitter target](/quest/m0/audio-jitter-target/README.md) - reads `jitter` as its floor, which is now the spread alone -- [Audio time-stretch](/quest/m1/watch-audio-time-stretch.md) - converges audio to a resized target without skipping -- [Data jitter](/quest/m1/data-jitter.md) - feeds the same measurement from JSON and binary tracks diff --git a/quest/m1/publish-delay.md b/quest/m1/publish-delay.md new file mode 100644 index 0000000000..aeb4209d69 --- /dev/null +++ b/quest/m1/publish-delay.md @@ -0,0 +1,27 @@ +# [S] js/publish: encoders advertise catalog delay + +## Goal + +The browser publisher advertises `delay` the way `moq-mux` does: each rendition +reports how far its minimum flush lateness trails the broadcast's earliest +rendition, as a lifetime maximum that is never lowered. A browser video encoder +running behind its audio encoder advertises that offset on video, so +`js/watch` holds it instead of playing video late. + +## Plan + +- `js/publish/src/jitter.ts` already measures each rendition's lateness on + `performance.now()`, and every js/publish timestamp shares that clock, so a + broadcast-wide sliding minimum beside the per-rendition one is enough. Mirror + `moq_mux::catalog::Estimator`: one window per rendition, one shared by the + broadcast, and `delay` is the gap between their minima. +- The shared minimum belongs to the `Broadcast` the encoders register on, not to + the encoders, so a swapped broadcast starts fresh. Keep it off the public API. +- The draft's rule stands: a consumer never subtracts `delay` across + renditions and holds the largest `delay + jitter` it subscribes to, so the + publisher reports its sliding-baseline maximum as is. +- `CatalogProducer` already refuses a lowered or zero `delay`. + +## Related + +- [Data jitter](/quest/m1/data-jitter.md) - the same measurement for JSON and binary tracks in Rust diff --git a/quest/m2/watch-data-sync.md b/quest/m2/watch-data-sync.md index e0d3d6ae8b..19b0b1f440 100644 --- a/quest/m2/watch-data-sync.md +++ b/quest/m2/watch-data-sync.md @@ -21,7 +21,6 @@ releases it. ## Required -- [Jitter clock](/quest/m1/jitter-flush-clock.md) - the `delay` field and `Sync` sizing this registers into - [Data jitter](/quest/m1/data-jitter.md) - data tracks advertise the `delay` and `jitter` this reads ## Related diff --git a/rs/hang/src/catalog/audio/mod.rs b/rs/hang/src/catalog/audio/mod.rs index 6afa7cef18..7c013c94a9 100644 --- a/rs/hang/src/catalog/audio/mod.rs +++ b/rs/hang/src/catalog/audio/mod.rs @@ -124,6 +124,17 @@ pub struct AudioConfig { #[serde_as(as = "MillisCeil")] #[serde(default)] pub jitter: Option, + + /// How far this rendition's frames reach the transport behind the broadcast's earliest + /// rendition, measured at the publisher from each rendition's minimum flush lateness. + /// Absent on the earliest rendition and on any rendition the publisher did not measure. + /// + /// A consumer holds `delay + jitter` for this rendition and MUST NOT subtract one rendition's + /// `delay` from another's: each is a lifetime maximum, so two need not share an origin. It only + /// ever grows over the life of a stream, and is serialized like [`jitter`](Self::jitter). + #[serde_as(as = "MillisCeil")] + #[serde(default)] + pub delay: Option, } impl AudioConfig { @@ -144,6 +155,7 @@ impl AudioConfig { description: None, container: Container::default(), jitter: None, + delay: None, } } } diff --git a/rs/hang/src/catalog/root.rs b/rs/hang/src/catalog/root.rs index 21c1e11e85..8941e3835f 100644 --- a/rs/hang/src/catalog/root.rs +++ b/rs/hang/src/catalog/root.rs @@ -301,18 +301,19 @@ mod test { assert_eq!(encoded, output, "wrong encoded output"); } - /// Lock in the on-wire shape of the jitter field: a bare integer number + /// Lock in the on-wire shape of the jitter and delay fields: a bare integer number /// of milliseconds. If `Option` ever loses the `duration_millis` /// serde adapter, this regresses to serde's default `{secs, nanos}` shape. #[test] - fn jitter_serialized_as_millis() { + fn jitter_and_delay_serialized_as_millis() { let mut encoded = r#"{ "video": { "renditions": { "video": { "codec": "avc1.64001f", "container": {"kind": "legacy"}, - "jitter": 100 + "jitter": 100, + "delay": 200 } } }, @@ -355,6 +356,7 @@ mod test { optimize_for_latency: None, container: Container::Legacy, jitter: Some(std::time::Duration::from_millis(100)), + delay: Some(std::time::Duration::from_millis(200)), }, ); @@ -371,6 +373,7 @@ mod test { description: None, container: Container::Legacy, jitter: Some(std::time::Duration::from_millis(40)), + delay: None, }, ); diff --git a/rs/hang/src/catalog/text/mod.rs b/rs/hang/src/catalog/text/mod.rs index 29e089e590..496b016af2 100644 --- a/rs/hang/src/catalog/text/mod.rs +++ b/rs/hang/src/catalog/text/mod.rs @@ -107,6 +107,17 @@ pub struct TextConfig { #[serde_as(as = "MillisCeil")] #[serde(default)] pub jitter: Option, + + /// How far this rendition's frames reach the transport behind the broadcast's earliest + /// rendition, measured at the publisher from each rendition's minimum flush lateness. + /// Absent on the earliest rendition and on any rendition the publisher did not measure. + /// + /// A consumer holds `delay + jitter` for this rendition and MUST NOT subtract one rendition's + /// `delay` from another's: each is a lifetime maximum, so two need not share an origin. It only + /// ever grows over the life of a stream, and is serialized like [`jitter`](Self::jitter). + #[serde_as(as = "MillisCeil")] + #[serde(default)] + pub delay: Option, } impl TextConfig { @@ -125,6 +136,7 @@ impl TextConfig { label: None, container: Container::default(), jitter: None, + delay: None, } } } diff --git a/rs/hang/src/catalog/video/mod.rs b/rs/hang/src/catalog/video/mod.rs index 1c75126d0e..db43d6c3eb 100644 --- a/rs/hang/src/catalog/video/mod.rs +++ b/rs/hang/src/catalog/video/mod.rs @@ -235,6 +235,17 @@ pub struct VideoConfig { #[serde_as(as = "MillisCeil")] #[serde(default)] pub jitter: Option, + + /// How far this rendition's frames reach the transport behind the broadcast's earliest + /// rendition, measured at the publisher from each rendition's minimum flush lateness. + /// Absent on the earliest rendition and on any rendition the publisher did not measure. + /// + /// A consumer holds `delay + jitter` for this rendition and MUST NOT subtract one rendition's + /// `delay` from another's: each is a lifetime maximum, so two need not share an origin. It only + /// ever grows over the life of a stream, and is serialized like [`jitter`](Self::jitter). + #[serde_as(as = "MillisCeil")] + #[serde(default)] + pub delay: Option, } impl VideoConfig { @@ -260,6 +271,7 @@ impl VideoConfig { optimize_for_latency: None, container: Container::default(), jitter: None, + delay: None, } } } diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index f1f1a03691..5832ae9375 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -190,6 +190,7 @@ fn normalize_video(config: &VideoConfig) -> VideoConfig { let mut config = config.clone(); config.bitrate = None; config.jitter = None; + config.delay = None; config.label = None; config.stalled = None; // The muxer ignores a non-finite framerate, and NaN never equals itself, so a catalog @@ -203,6 +204,7 @@ fn normalize_audio(config: &AudioConfig) -> AudioConfig { let mut config = config.clone(); config.bitrate = None; config.jitter = None; + config.delay = None; config.label = None; config } diff --git a/rs/moq-msf/src/lib.rs b/rs/moq-msf/src/lib.rs index 200aa204f2..83087ca201 100644 --- a/rs/moq-msf/src/lib.rs +++ b/rs/moq-msf/src/lib.rs @@ -184,6 +184,15 @@ pub struct Track { /// Serialized as a JSON number of milliseconds, matching the hang catalog. #[serde_as(as = "Option")] pub jitter: Option, + + /// How far this rendition trails the broadcast's earliest rendition (non-standard + /// extension; not in the MSF/CMSF drafts). + /// + /// Serialized as a JSON number of milliseconds, matching [`jitter`](Self::jitter). Absent on + /// the earliest rendition. A consumer holds `delay + jitter` and does not subtract one + /// rendition's `delay` from another's. + #[serde_as(as = "Option")] + pub delay: Option, } impl Catalog<()> { @@ -480,6 +489,7 @@ impl Track { max_grp_sap_starting_type: None, max_obj_sap_starting_type: None, jitter: None, + delay: None, } } } @@ -631,6 +641,7 @@ mod test { max_grp_sap_starting_type: None, max_obj_sap_starting_type: None, jitter: None, + delay: None, } } @@ -655,6 +666,7 @@ mod test { max_grp_sap_starting_type: None, max_obj_sap_starting_type: None, jitter: None, + delay: None, } } @@ -679,6 +691,7 @@ mod test { max_grp_sap_starting_type: Some(1), max_obj_sap_starting_type: Some(2), jitter: Some(Duration::from_millis(15)), + delay: None, } } @@ -700,6 +713,7 @@ mod test { assert!(track.get("maxGrpSapStartingType").is_none()); assert!(track.get("maxObjSapStartingType").is_none()); assert!(track.get("jitter").is_none()); + assert!(track.get("delay").is_none()); assert_eq!(track["stalled"], true); } @@ -823,6 +837,7 @@ mod test { assert_eq!(track.max_grp_sap_starting_type, None); assert_eq!(track.max_obj_sap_starting_type, None); assert_eq!(track.jitter, None); + assert_eq!(track.delay, None); } #[test] @@ -858,6 +873,20 @@ mod test { assert_eq!(value["tracks"][0]["jitter"].as_f64(), Some(15.0)); } + #[test] + fn delay_roundtrips() { + let mut track = track_with_sap_and_jitter(); + track.delay = Some(Duration::from_millis(200)); + let original = Catalog::new(vec![track]); + + let json = original.to_json().unwrap(); + let parsed = Catalog::<()>::from_str(&json).unwrap(); + assert_eq!(parsed.tracks[0].delay, Some(Duration::from_millis(200))); + + let value: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(value["tracks"][0]["delay"].as_f64(), Some(200.0)); + } + #[test] fn serialize_emits_draft01_version() { // Callers never set a version; we always emit the newest draft string. diff --git a/rs/moq-mux/src/catalog/estimate.rs b/rs/moq-mux/src/catalog/estimate.rs index 8ac0378131..9455626547 100644 --- a/rs/moq-mux/src/catalog/estimate.rs +++ b/rs/moq-mux/src/catalog/estimate.rs @@ -1,4 +1,5 @@ use std::collections::VecDeque; +use std::sync::{Arc, Mutex, PoisonError}; use std::time::{Duration, Instant}; use moq_net::Timestamp; @@ -19,6 +20,8 @@ pub struct Estimate { pub jitter: Option, /// The maximum bitrate in bits per second. pub bitrate: Option, + /// The most this track's minimum flush lateness has trailed the broadcast's earliest track. + pub delay: Option, } impl Estimate { @@ -33,6 +36,12 @@ impl Estimate { self.bitrate = bitrate.into(); self } + + /// Set the delay behind the broadcast's earliest track (or clear it with `None`). + pub fn with_delay(mut self, delay: impl Into>) -> Self { + self.delay = delay.into(); + self + } } /// Measures the catalog jitter and bitrate of one track from the frames written to it. @@ -76,7 +85,13 @@ impl Estimate { pub struct Estimator { jitter: Jitter, bitrate: Bitrate, - baseline: Baseline, + /// This track's recent minimum flush lateness. + baseline: Window, + /// The recent minimum across every track sharing it. A standalone estimator's is its own, so + /// it never measures a delay. + broadcast: Baseline, + /// The largest amount [`baseline`](Self::baseline) has trailed [`broadcast`](Self::broadcast). + delay: Duration, } impl Estimator { @@ -85,6 +100,14 @@ impl Estimator { Self::default() } + /// Measure `delay` against the tracks sharing `broadcast`. + pub(crate) fn with_broadcast(broadcast: Baseline) -> Self { + Self { + broadcast, + ..Self::default() + } + } + /// Observe a frame of `bytes` encoded bytes at presentation time `timestamp`, as written by /// [`container::Producer::write`](crate::container::Producer::write). pub fn write(&mut self, timestamp: Timestamp, bytes: usize) { @@ -94,10 +117,13 @@ impl Estimator { /// Measure when an encoder handed a frame to the transport. Only locally encoded frames should /// call this; imports keep clock-free batch and reorder estimates. Jitter is the spread above - /// this track's own recent minimum lateness, so a constant encoder delay is not jitter. + /// this track's own recent minimum lateness, so a constant encoder delay is not jitter. Delay + /// is how far that minimum trails the earliest track on the same catalog. pub fn flush(&mut self, timestamp: Timestamp, now: Instant) { - let spread = self.baseline.observe(timestamp, now); - self.jitter.max = self.jitter.max.max(spread); + let (lateness, earliest) = self.broadcast.observe(timestamp, now); + let minimum = self.baseline.observe(now, lateness); + self.jitter.max = self.jitter.max.max(nanos_between(minimum, lateness)); + self.delay = self.delay.max(nanos_between(earliest, minimum)); } /// Close the current span at `end`, as [`container::Producer::cut`](crate::container::Producer::cut) @@ -110,11 +136,15 @@ impl Estimator { self.bitrate.cut(end.map(nanos)); } - /// Discard the open bitrate span and the flush baseline, so nothing is measured across a break + /// Discard the open bitrate span and the flush baselines, so nothing is measured across a break /// in the timeline. See [`container::Producer::discontinuity`](crate::container::Producer::discontinuity). + /// + /// The broadcast baseline is cleared too: a pause usually stops every track, and the others' + /// pre-pause minimum would otherwise read as a delay until it left the window. pub fn discontinuity(&mut self) { self.bitrate.discontinuity(); - self.baseline = Baseline::default(); + self.baseline = Window::default(); + self.broadcast.discontinuity(); } /// Observe a frame's reorder delay (`PTS - DTS`), which raises the jitter to the decode buffer a @@ -137,6 +167,7 @@ impl Estimator { Estimate { jitter: self.jitter.current(), bitrate: self.bitrate.current(), + delay: (!self.delay.is_zero()).then_some(self.delay), } } } @@ -258,14 +289,43 @@ impl Jitter { } } -/// One track's minimum encode lateness over the recent window. +/// The flush clock shared by every track on one catalog: a common epoch, so lateness compares +/// across tracks, and the minimum lateness any of them flushed with over the recent window. /// /// An `Instant` has no public mapping to the broadcast's media epoch. The first observation -/// chooses a local origin; its unknown offset cancels when subtracting the recent minimum. A -/// monotonic deque makes insertion and expiry amortized O(1). +/// chooses a local origin; its unknown offset cancels when subtracting one minimum from another. +#[derive(Clone, Default)] +pub(crate) struct Baseline(Arc>); + #[derive(Default)] -struct Baseline { +struct Shared { epoch: Option, + window: Window, +} + +impl Baseline { + /// Return the frame's lateness on the shared epoch and the broadcast's minimum including it. + fn observe(&self, timestamp: Timestamp, now: Instant) -> (i128, i128) { + let mut shared = self.0.lock().unwrap_or_else(PoisonError::into_inner); + let epoch = *shared.epoch.get_or_insert(now); + let elapsed = match now.checked_duration_since(epoch) { + Some(duration) => duration.as_nanos() as i128, + None => -(epoch.duration_since(now).as_nanos() as i128), + }; + let lateness = elapsed - timestamp.as_nanos() as i128; + (lateness, shared.window.observe(now, lateness)) + } + + fn discontinuity(&self) { + self.0.lock().unwrap_or_else(PoisonError::into_inner).window = Window::default(); + } +} + +/// The minimum lateness over the last [`JITTER_WINDOW`], so a media clock drifting against the +/// wall clock does not ratchet forever. A monotonic deque makes insertion and expiry amortized +/// O(1). +#[derive(Default)] +struct Window { samples: VecDeque, } @@ -274,32 +334,29 @@ struct Sample { lateness: i128, } -impl Baseline { - fn observe(&mut self, timestamp: Timestamp, now: Instant) -> Duration { - let epoch = *self.epoch.get_or_insert(now); +impl Window { + /// Insert a lateness observed at `now` and return the window's minimum. + fn observe(&mut self, now: Instant, lateness: i128) -> i128 { while self .samples .front() - .is_some_and(|sample| now.duration_since(sample.at) > JITTER_WINDOW) + .is_some_and(|sample| now.saturating_duration_since(sample.at) > JITTER_WINDOW) { self.samples.pop_front(); } - - let elapsed = match now.checked_duration_since(epoch) { - Some(duration) => duration.as_nanos() as i128, - None => -(epoch.duration_since(now).as_nanos() as i128), - }; - let lateness = elapsed - timestamp.as_nanos() as i128; while self.samples.back().is_some_and(|sample| sample.lateness >= lateness) { self.samples.pop_back(); } self.samples.push_back(Sample { at: now, lateness }); - let minimum = self.samples.front().expect("the current sample was inserted").lateness; - let spread = u64::try_from(lateness - minimum).unwrap_or(u64::MAX); - Duration::from_nanos(spread) + self.samples.front().expect("the current sample was inserted").lateness } } +/// `to - from` as a duration, clamped at zero. +fn nanos_between(from: i128, to: i128) -> Duration { + Duration::from_nanos(u64::try_from(to - from).unwrap_or(if to > from { u64::MAX } else { 0 })) +} + #[cfg(test)] mod tests { use super::*; @@ -355,34 +412,94 @@ mod tests { #[test] fn baseline_expires_drift() { + let mut estimator = Estimator::new(); let anchor = Instant::now(); - - let mut drift = Baseline::default(); - let maximum = (0..100u128) - .map(|second| { - drift.observe( - micros((second * 1_000_000) as u64), - anchor + Duration::from_millis((second * 1_001) as u64), - ) - }) - .max() - .unwrap(); - assert!(maximum <= Duration::from_millis(10), "{maximum:?}"); + for second in 0..100u64 { + estimator.flush( + micros(second * 1_000_000), + anchor + Duration::from_millis(second * 1_001), + ); + } + let jitter = estimator.estimate().jitter.unwrap(); + assert!(jitter <= Duration::from_millis(10), "{jitter:?}"); } #[test] fn early_flush_keeps_lowering_the_baseline() { - let mut baseline = Baseline::default(); + let mut estimator = Estimator::new(); let anchor = Instant::now(); - for second in 0..100u128 { - assert_eq!( - baseline.observe( - micros((second * 2_000_000) as u64), - anchor + Duration::from_secs(second as u64) - ), - Duration::ZERO - ); + for second in 0..100u64 { + estimator.flush(micros(second * 2_000_000), anchor + Duration::from_secs(second)); } + assert_eq!(estimator.estimate(), Estimate::default()); + } + + /// Two tracks on one broadcast baseline, flushed every 100 ms for `seconds` with the lateness + /// each closure returns (in ms) at a given frame's media time. + fn pair(seconds: u64, fast: impl Fn(u64) -> u64, slow: impl Fn(u64) -> u64) -> (Estimate, Estimate) { + let broadcast = Baseline::default(); + let mut estimators = [ + Estimator::with_broadcast(broadcast.clone()), + Estimator::with_broadcast(broadcast), + ]; + let anchor = Instant::now(); + for frame in 0..seconds * 10 { + let pts = frame * 100; + for (estimator, lateness) in estimators.iter_mut().zip([fast(pts), slow(pts)]) { + estimator.flush(micros(pts * 1_000), anchor + Duration::from_millis(pts + lateness)); + } + } + let [fast, slow] = estimators.map(|estimator| estimator.estimate()); + (fast, slow) + } + + #[test] + fn a_constant_offset_is_delay_on_the_slower_track() { + let (fast, slow) = pair(5, |_| 0, |_| 200); + assert_eq!(fast, Estimate::default()); + assert_eq!(slow, Estimate::default().with_delay(Duration::from_millis(200))); + } + + #[test] + fn a_common_drift_stays_bounded() { + // Both media clocks run 0.1% slow for 100 s, so each lateness climbs 100 ms together. + let (fast, slow) = pair(100, |pts| pts / 1_000, |pts| 200 + pts / 1_000); + assert_eq!(fast.delay, None); + assert_eq!(slow.delay, Some(Duration::from_millis(200))); + for jitter in [fast.jitter, slow.jitter] { + assert!(jitter.unwrap() <= Duration::from_millis(10), "{jitter:?}"); + } + } + + /// The earliest track starts at 0 and the other at 200 ms, then the first drifts to 500 ms. The + /// delays no longer share an origin, but the drifting track's own `delay` covers the new offset, + /// so the largest `delay + jitter` never falls below the real spread. + #[test] + fn the_earliest_track_changing_does_not_under_buffer() { + let drift = |pts: u64| pts.saturating_sub(10_000).min(50_000) / 100; + let (drifted, steady) = pair(80, drift, |_| 200); + assert_eq!(steady.delay, Some(Duration::from_millis(200))); + assert_eq!(drifted.delay, Some(Duration::from_millis(300))); + } + + #[test] + fn a_discontinuity_clears_the_broadcast_baseline() { + let broadcast = Baseline::default(); + let mut audio = Estimator::with_broadcast(broadcast.clone()); + let mut video = Estimator::with_broadcast(broadcast); + let anchor = Instant::now(); + audio.flush(micros(0), anchor); + video.flush(micros(0), anchor + Duration::from_millis(50)); + assert_eq!(video.estimate().delay, Some(Duration::from_millis(50))); + + // Both resume at the previous live edge after a 2 s pause: the pause is not a delay. + audio.discontinuity(); + video.discontinuity(); + video.flush(micros(40_000), anchor + Duration::from_millis(2_090)); + audio.flush(micros(40_000), anchor + Duration::from_millis(2_040)); + video.flush(micros(80_000), anchor + Duration::from_millis(2_130)); + assert_eq!(video.estimate().delay, Some(Duration::from_millis(50))); + assert_eq!(audio.estimate().delay, None); } #[test] diff --git a/rs/moq-mux/src/catalog/msf/consumer.rs b/rs/moq-mux/src/catalog/msf/consumer.rs index 2d49b61510..c8e2a256a0 100644 --- a/rs/moq-mux/src/catalog/msf/consumer.rs +++ b/rs/moq-mux/src/catalog/msf/consumer.rs @@ -237,6 +237,7 @@ fn video_config_from_msf(track: &moq_msf::Track) -> Result> config.framerate = track.framerate; config.container = container; config.jitter = track.jitter; + config.delay = track.delay; Ok(Some(config)) } @@ -276,6 +277,7 @@ fn audio_config_from_msf(track: &moq_msf::Track) -> Result> config.description = legacy_description(track)?; config.container = container; config.jitter = track.jitter; + config.delay = track.delay; Ok(Some(config)) } @@ -446,6 +448,30 @@ mod test { assert_eq!(video.stalled, Some(true)); } + #[test] + fn delay_round_trips_onto_hang() { + let mut video = video_track("video0", moq_msf::Packaging::Legacy, None); + video.jitter = Some(std::time::Duration::from_millis(40)); + video.delay = Some(std::time::Duration::from_millis(200)); + + let mut audio = audio_track("audio0", moq_msf::Packaging::Loc); + audio.delay = Some(std::time::Duration::from_millis(80)); + + let catalog = from_msf::<()>(&moq_msf::Catalog::new(vec![video, audio])).expect("delay should convert"); + assert_eq!( + catalog.video.renditions["video0"].delay, + Some(std::time::Duration::from_millis(200)) + ); + assert_eq!( + catalog.video.renditions["video0"].jitter, + Some(std::time::Duration::from_millis(40)) + ); + assert_eq!( + catalog.audio.renditions["audio0"].delay, + Some(std::time::Duration::from_millis(80)) + ); + } + #[test] fn loc_audio_yields_loc_container() { let msf = moq_msf::Catalog::new(vec![audio_track("audio0", moq_msf::Packaging::Loc)]); diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index 0d0958a123..0811704b1b 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -182,6 +182,9 @@ pub struct Producer { /// Connection allocator passthrough tracks claim their peak-hold bitrate on. /// See [`Config::with_bandwidth`]. bandwidth: moq_net::bandwidth::Allocator, + /// The minimum flush lateness across this catalog's renditions, which each rendition's + /// advertised `delay` is measured against. + baseline: super::estimate::Baseline, } // Manual Clone so a producer is cheaply clonable regardless of whether `E` is. @@ -194,6 +197,7 @@ impl Clone for Producer { timeline: self.timeline.clone(), max_age: self.max_age, bandwidth: self.bandwidth.clone(), + baseline: self.baseline.clone(), } } } @@ -342,6 +346,7 @@ impl Producer { timeline, max_age: config.max_age, bandwidth: config.bandwidth, + baseline: Default::default(), }) } @@ -615,6 +620,11 @@ impl Producer { .with_bandwidth(self.bandwidth.clone())) } + /// A fresh estimator whose `delay` is measured against this catalog's other renditions. + pub(crate) fn estimator(&self) -> super::Estimator { + super::Estimator::with_broadcast(self.baseline.clone()) + } + /// The allocator passthrough tracks claim on. fMP4 writes groups by hand, so it reads this itself. pub(crate) fn bandwidth(&self) -> moq_net::bandwidth::Allocator { self.bandwidth.clone() @@ -941,6 +951,7 @@ fn to_msf_media(catalog: &hang::Catalog) -> moq_msf::Catalog { track.max_grp_sap_starting_type = sap_type; track.max_obj_sap_starting_type = sap_type; track.jitter = config.jitter; + track.delay = config.delay; tracks.push(track); } @@ -971,6 +982,7 @@ fn to_msf_media(catalog: &hang::Catalog) -> moq_msf::Catalog { track.max_grp_sap_starting_type = Some(1); track.max_obj_sap_starting_type = Some(1); track.jitter = config.jitter; + track.delay = config.delay; tracks.push(track); } @@ -1608,6 +1620,7 @@ mod test { video_config.framerate = Some(30.0); video_config.container = Container::Legacy; video_config.jitter = Some(std::time::Duration::from_millis(100)); + video_config.delay = Some(std::time::Duration::from_millis(200)); let mut video_renditions = BTreeMap::new(); video_renditions.insert("video0".to_string(), video_config); @@ -1615,6 +1628,7 @@ mod test { let mut audio_config = AudioConfig::new(AudioCodec::Opus, 48_000, 2); audio_config.container = Container::Legacy; audio_config.jitter = Some(std::time::Duration::from_millis(40)); + audio_config.delay = Some(std::time::Duration::from_millis(80)); let mut audio_renditions = BTreeMap::new(); audio_renditions.insert("audio0".to_string(), audio_config); @@ -1631,12 +1645,14 @@ mod test { assert_eq!(video.max_grp_sap_starting_type, Some(2)); assert_eq!(video.max_obj_sap_starting_type, Some(2)); assert_eq!(video.jitter, Some(std::time::Duration::from_millis(100))); + assert_eq!(video.delay, Some(std::time::Duration::from_millis(200))); let audio = &msf.tracks[1]; assert_eq!(audio.role, Some(moq_msf::Role::Audio)); assert_eq!(audio.max_grp_sap_starting_type, Some(1)); assert_eq!(audio.max_obj_sap_starting_type, Some(1)); assert_eq!(audio.jitter, Some(std::time::Duration::from_millis(40))); + assert_eq!(audio.delay, Some(std::time::Duration::from_millis(80))); } #[test] diff --git a/rs/moq-mux/src/catalog/tracks.rs b/rs/moq-mux/src/catalog/tracks.rs index 6c2c349fdb..91c0670fcc 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -190,6 +190,8 @@ pub struct VideoHint { pub optimize_for_latency: Option, /// The maximum jitter before the next frame is emitted. pub jitter: Option, + /// How far this rendition trails the broadcast's earliest rendition. + pub delay: Option, /// The container wrapping each frame on the wire. /// /// Unlike the other fields this is a choice, not a hint: the bitstream never reveals a @@ -224,6 +226,7 @@ impl From for VideoHint { framerate: config.framerate, optimize_for_latency: config.optimize_for_latency, jitter: config.jitter, + delay: config.delay, container: config.container, } } @@ -247,6 +250,7 @@ impl VideoHint { fill(&mut config.framerate, self.framerate); fill(&mut config.optimize_for_latency, self.optimize_for_latency); fill(&mut config.jitter, self.jitter); + fill(&mut config.delay, self.delay); config.container = self.container.clone(); } @@ -276,11 +280,15 @@ impl RenditionConfig for hang::catalog::VideoConfig { } fn estimate(&self) -> Estimate { - Estimate::default().with_jitter(self.jitter).with_bitrate(self.bitrate) + Estimate::default() + .with_jitter(self.jitter) + .with_bitrate(self.bitrate) + .with_delay(self.delay) } fn set_estimate(&mut self, estimate: Estimate) { self.jitter = estimate.jitter; self.bitrate = estimate.bitrate; + self.delay = estimate.delay; } } @@ -300,11 +308,15 @@ impl RenditionConfig for hang::catalog::AudioConfig { } fn estimate(&self) -> Estimate { - Estimate::default().with_jitter(self.jitter).with_bitrate(self.bitrate) + Estimate::default() + .with_jitter(self.jitter) + .with_bitrate(self.bitrate) + .with_delay(self.delay) } fn set_estimate(&mut self, estimate: Estimate) { self.jitter = estimate.jitter; self.bitrate = estimate.bitrate; + self.delay = estimate.delay; } } @@ -319,10 +331,11 @@ impl RenditionConfig for hang::catalog::TextConfig { catalog.text.renditions.remove(name); } fn estimate(&self) -> Estimate { - Estimate::default().with_jitter(self.jitter) + Estimate::default().with_jitter(self.jitter).with_delay(self.delay) } fn set_estimate(&mut self, estimate: Estimate) { self.jitter = estimate.jitter; + self.delay = estimate.delay; } } @@ -518,6 +531,11 @@ impl> Rendition { &self.name } + /// A fresh estimator measuring `delay` against the catalog's other renditions. + pub(crate) fn estimator(&self) -> super::Estimator { + self.catalog.estimator() + } + /// Resolve a timestamp on the broadcast's shared clock (see [`Producer::timestamp`]). pub fn timestamp(&self, hint: Option) -> crate::Result { self.catalog.timestamp(hint) @@ -533,7 +551,7 @@ impl> Rendition { pub(crate) fn set(&mut self, mut config: C) -> crate::Result<()> { let supplied = config.estimate(); let resolved = Self::resolved(&supplied, &self.detected); - self.check_jitter(&resolved)?; + self.check_decrease(&resolved)?; config.set_estimate(resolved.clone()); { let mut guard = self.catalog.modify()?; @@ -560,6 +578,9 @@ impl> Rendition { if estimate.bitrate.is_none() { estimate.bitrate = detected.bitrate; } + if estimate.delay.is_none() { + estimate.delay = detected.delay; + } estimate } @@ -581,10 +602,11 @@ impl> Rendition { return Ok(()); } let mut resolved = Self::resolved(&self.supplied, &estimate); - // A measurement never lowers the published jitter, including one raised through `modify`. - if let Some(published) = self.config()?.estimate().jitter { - resolved.jitter = Some(resolved.jitter.map_or(published, |jitter| jitter.max(published))); - } + // A measurement never lowers the published jitter or delay, including one raised through + // `modify`. + let published = self.config()?.estimate(); + resolved.jitter = resolved.jitter.max(published.jitter); + resolved.delay = resolved.delay.max(published.delay); self.detected = estimate; if self.published.as_ref() != Some(&resolved) { let mut config = self.config()?; @@ -595,13 +617,18 @@ impl> Rendition { Ok(()) } - fn check_jitter(&self, next: &Estimate) -> crate::Result<()> { - if self.present - && let Some(previous) = self.config()?.estimate().jitter - && next.jitter.is_none_or(|jitter| jitter < previous) - { + /// Refuse a config that lowers the jitter or delay already advertised to subscribers. + fn check_decrease(&self, next: &Estimate) -> crate::Result<()> { + if !self.present { + return Ok(()); + } + let previous = self.config()?.estimate(); + if next.jitter < previous.jitter { return Err(crate::Error::JitterDecreased); } + if next.delay < previous.delay { + return Err(crate::Error::DelayDecreased); + } Ok(()) } @@ -620,7 +647,7 @@ impl> Rendition { if !self.present { return Err(crate::Error::NotPublished); } - self.check_jitter(&config.estimate())?; + self.check_decrease(&config.estimate())?; let mut guard = self.catalog.modify()?; let mut next = (*guard).clone(); config.insert(&mut next, &self.name); @@ -723,6 +750,45 @@ mod tests { ); } + #[test] + fn published_delay_never_decreases() { + let (_broadcast, catalog, mut rendition) = video_track(); + let delayed = |delay| { + let mut config = config(None, None); + config.delay = delay; + config + }; + rendition.set(delayed(Some(Duration::from_millis(200)))).unwrap(); + for smaller in [Some(Duration::from_millis(100)), None] { + assert!(matches!( + rendition.set(delayed(smaller)), + Err(crate::Error::DelayDecreased) + )); + assert!(matches!( + rendition.replace(delayed(smaller)), + Err(crate::Error::DelayDecreased) + )); + } + assert_eq!( + catalog.snapshot().video.renditions["v"].delay, + Some(Duration::from_millis(200)) + ); + + let (_broadcast, catalog, mut detected) = video_track(); + detected.set(config(None, None)).unwrap(); + detected + .estimate(Estimate::default().with_delay(Duration::from_millis(200))) + .unwrap(); + // A lower measurement holds the published value rather than failing the write path. + detected + .estimate(Estimate::default().with_delay(Duration::from_millis(100))) + .unwrap(); + assert_eq!( + catalog.snapshot().video.renditions["v"].delay, + Some(Duration::from_millis(200)) + ); + } + #[test] fn measurement_keeps_a_jitter_raised_by_modify() { let (_broadcast, catalog, mut rendition) = video_track(); diff --git a/rs/moq-mux/src/container/producer.rs b/rs/moq-mux/src/container/producer.rs index fa18792c30..abdd85fd72 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -187,7 +187,7 @@ where previous_timestamp: None, cadence: None, reordered: false, - estimator: crate::catalog::Estimator::new(), + estimator: rendition.estimator(), bandwidth: None, rendition: Some(Box::new(rendition)), } @@ -202,7 +202,7 @@ where /// /// Estimate fields the config left to detection at [`set`](Self::set) stay owned by detection: /// an edit to them here is published but replaced by the next measurement, except that jitter - /// never drops below the published value. Call `set` with the field filled in to pin it. + /// and delay never drop below the published value. Call `set` with the field filled in to pin it. pub fn modify(&mut self) -> crate::Result> { let rendition = self.rendition.as_mut().ok_or(crate::Error::NotPublished)?; let config = rendition.config()?; @@ -794,7 +794,7 @@ mod tests { } #[test] - fn catalog_flush_measures_each_rendition_against_its_own_minimum() { + fn catalog_flush_measures_delay_across_renditions_and_jitter_within_each() { let mut broadcast = moq_net::broadcast::Info::new().produce(); let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); let mut tracks = Vec::new(); @@ -814,9 +814,11 @@ mod tests { let ms = std::time::Duration::from_millis; let pts = |millis: u64| Timestamp::from_micros(millis * 1_000).unwrap(); tracks[0].flush(pts(0), anchor).unwrap(); - // A constant 200ms offset behind the other rendition is not jitter. + // A constant 200ms offset behind the other rendition is delay, not jitter. tracks[1].flush(pts(0), anchor + ms(200)).unwrap(); assert_eq!(catalog.snapshot().video.renditions["slow"].jitter, None); + assert_eq!(catalog.snapshot().video.renditions["slow"].delay, Some(ms(200))); + assert_eq!(catalog.snapshot().video.renditions["fast"].delay, None); // A frame flushed 60ms later than the slow rendition's own minimum is. tracks[1].flush(pts(40), anchor + ms(300)).unwrap(); diff --git a/rs/moq-mux/src/error.rs b/rs/moq-mux/src/error.rs index a0a50989da..9603dfa5e5 100644 --- a/rs/moq-mux/src/error.rs +++ b/rs/moq-mux/src/error.rs @@ -241,6 +241,10 @@ pub enum Error { /// A rendition tried to lower jitter already advertised to subscribers. #[error("catalog jitter cannot decrease for a published rendition")] JitterDecreased, + + /// A rendition tried to lower delay already advertised to subscribers. + #[error("catalog delay cannot decrease for a published rendition")] + DelayDecreased, } impl Error { diff --git a/rs/moq-video/src/encode/producer.rs b/rs/moq-video/src/encode/producer.rs index dd76386bcd..998e7dbf1c 100644 --- a/rs/moq-video/src/encode/producer.rs +++ b/rs/moq-video/src/encode/producer.rs @@ -663,10 +663,10 @@ mod tests { producer.publish(&encoder.finish().unwrap()).unwrap(); let (name, resolved) = rendition(&catalog).expect("the importer should have registered a video rendition"); - // Jitter aside, which is measured from the frames rather than declared by either. + // Jitter and delay aside, which are measured from the frames rather than declared by either. let (mut before, mut after) = (advertised, resolved.clone()); - before.jitter = None; - after.jitter = None; + (before.jitter, before.delay) = (None, None); + (after.jitter, after.delay) = (None, None); assert_eq!( before, after, "the first keyframe should confirm the advertised rendition, not correct it"