diff --git a/doc/concept/hang.md b/doc/concept/hang.md index e430d27b7c..c48092fcc6 100644 --- a/doc/concept/hang.md +++ b/doc/concept/hang.md @@ -54,6 +54,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. +- **Whole contributions.** A catalog can list relative broadcast references without fetching their catalogs. Explicit selection opens a contribution, whose rendition names remain separate from the source. `@moq/watch` exposes this as a scoped child broadcast; listing contributions alone never starts their media producers. - **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. - **Disabled renditions.** An audio or video rendition with `enabled: false` has no frames coming, and a player must not select it. Pausing a rendition in `@moq/publish` keeps it in the catalog this way rather than removing it, so a mute is a one-field update; a disabled video rendition encodes one black keyframe first, so a player that predates the field shows black instead of a frozen picture. The field is written only when false. The older video `stalled` field is no longer written, and readers ignore it. diff --git a/drafts/draft-lcurley-moq-hang.md b/drafts/draft-lcurley-moq-hang.md index 7f95bc3083..4c54f2721f 100644 --- a/drafts/draft-lcurley-moq-hang.md +++ b/drafts/draft-lcurley-moq-hang.md @@ -122,6 +122,7 @@ The root of the catalog is a JSON document with the following schema: ~~~ type Catalog = { + "contributions": Record | undefined, "audio": AudioSchema | undefined, "video": VideoSchema | undefined, "archive": ArchiveSchema | undefined, @@ -143,6 +144,16 @@ The `json` and `binary` sections ({{data}}) are how a track like that is listed. This specification defines audio, video, and text media tracks, plus an optional `archive` entry ({{archive-catalog}}) naming the timeline track ({{timeline}}) that indexes each track, an optional root `clock` entry ({{clock}}) mapping content time to wall time, and the application data tracks in {{data}}. +## Contributions {#contributions} +The optional `contributions` map names whole broadcasts whose catalogs describe their renditions. +Each entry MUST contain a string `broadcast`, resolved relative to this catalog's broadcast using {{field-broadcast}}. +A malformed entry or a reference escaping the root MUST cause the catalog to be rejected. + +Listing an entry MUST NOT subscribe to the contribution; a consumer opens its catalog only upon explicit selection. +A contribution's rendition identifiers belong to that contribution and MUST NOT replace identifiers in the referring catalog. +An unavailable contribution does not invalidate the referring catalog. +References are not recursively expanded: each further contribution requires an explicit selection. + ## Clock {#clock} The catalog's root `clock` field is the broadcast's one continuous clock: @@ -1074,6 +1085,8 @@ This document has no IANA actions. ## moq-hang-04 {:numbered="false"} +- Added optional whole-broadcast `contributions` with explicit lazy selection and scoped rendition identifiers. + - Defined encoder `jitter` as flush lateness above the rendition's own recent minimum, replacing fixed frame-duration hints; container batches retain media-span estimates. - Timed CMAF samples from the frame timestamp, which is the fragment's earliest presentation time; `tfdt` is only relative within the fragment, and places the samples of an untimed frame. A timed CMAF track counts in its `mdhd` ticks. - Allowed a DVR to delete timeline objects no checkpoint recovery needs, oldest first. diff --git a/js/hang/src/catalog/root.ts b/js/hang/src/catalog/root.ts index 67f955caa5..fd138f53be 100644 --- a/js/hang/src/catalog/root.ts +++ b/js/hang/src/catalog/root.ts @@ -8,7 +8,7 @@ import { BinarySchema } from "./binary"; import { ClockSchema } from "./clock"; import { TRACK } from "./format"; import { JsonSchema } from "./json"; -import type { RelativeBroadcast } from "./path"; +import { type RelativeBroadcast, RelativeBroadcastSchema } from "./path"; import { PRIORITY } from "./priority"; import { section } from "./section"; import { TextSchema } from "./text"; @@ -20,7 +20,7 @@ import { VideoSchema } from "./video"; * The media sections are `video`, `audio`, and `text`, alongside the broadcast's `archive` * (the segment index and any durable recording) and its `clock` (the one wall-clock mapping); * `json` and `binary` list application data tracks that aren't media. A section is omitted - * when it holds no tracks. + * when it holds no tracks. `contributions` lists whole broadcasts without opening them. * * This is a *loose* object: unknown root sections pass through validation untouched, so an * application can add its own sections (e.g. `scte35`) without modifying hang. A base consumer @@ -28,6 +28,8 @@ import { VideoSchema } from "./video"; * built via `z.extend(RootSchema, { ... })`. */ export const RootSchema = z.looseObject({ + // Whole broadcasts are listed without opening their catalogs. IDs stay scoped to each broadcast. + contributions: z.optional(z.record(z.string(), z.object({ broadcast: RelativeBroadcastSchema }))), video: z.optional(VideoSchema), audio: z.optional(AudioSchema), // The broadcast's segment index and any durable archive, if the publisher offers one. @@ -94,6 +96,7 @@ export function checkResolvable(root: Root, base: Moq.Path.Valid): Root { root.text?.renditions, root.json?.tracks, root.binary?.tracks, + root.contributions, ]; for (const section of sections) { for (const { broadcast } of Object.values(section ?? {})) { diff --git a/js/watch/src/broadcast.ts b/js/watch/src/broadcast.ts index b394800e81..600a485fb5 100644 --- a/js/watch/src/broadcast.ts +++ b/js/watch/src/broadcast.ts @@ -3,7 +3,7 @@ import * as Json from "@moq/json"; import * as Msf from "@moq/msf"; import type * as Moq from "@moq/net"; import { Error as NetError, Path, StreamCode } from "@moq/net"; -import { Effect, type Getter, getter, type Inputs, type Readonlys, readonlys, Signal } from "@moq/signals"; +import { Derived, Effect, type Getter, getter, type Inputs, type Readonlys, readonlys, Signal } from "@moq/signals"; import { toHang } from "./msf"; @@ -357,7 +357,15 @@ export class Broadcast { const catalog = effect.get(this.in.catalog); let accepted: Catalog.Root | undefined; try { - accepted = catalog && Catalog.checkResolvable(Catalog.checkRenditions(catalog), name); + accepted = + catalog && + Catalog.checkResolvable( + Catalog.checkRenditions({ + ...catalog, + contributions: Catalog.RootSchema.shape.contributions.parse(catalog.contributions), + }), + name, + ); } catch (err) { console.error("rejecting catalog", name, err); } @@ -478,6 +486,42 @@ export class Broadcast { return this.#requestBroadcast(effect, origin, target.path); } + /** + * Select a whole contribution, opening its catalog only while the caller owns this effect. + * An absent entry leaves the child disabled; adding or replacing it updates the child. + * Wire the returned broadcast into a media source inside a nested effect. Its rendition + * IDs stay local to that broadcast, so they never replace the parent catalog's IDs. + * + * ```ts + * effects.run((effect) => { + * const child = broadcast.contribution(effect, "extra"); + * const source = new Video.Source({ broadcast: child, supported: Video.Decoder.supported }); + * effect.cleanup(() => source.close()); + * const decoder = new Video.Decoder({ source, sync }); + * effect.cleanup(() => decoder.close()); + * // Feed decoder.out.frame into your renderer. + * }); + * ``` + */ + contribution(effect: Effect, id: string): Broadcast { + const selected = effect.computed((run) => { + const catalog = run.get(this.out.catalog); + const entry = catalog?.contributions?.[id]; + if (!entry || !Object.hasOwn(catalog?.contributions ?? {}, id)) return undefined; + const name = Path.tryResolve(run.get(this.in.name), entry.broadcast); + if (name === undefined) throw new Catalog.EscapingBroadcast(entry.broadcast); + return name; + }); + const child = new Broadcast({ + origin: this.in.origin, + name: new Derived([selected], (name) => name ?? Path.empty()), + enabled: new Derived([selected, this.in.enabled], (name, enabled) => name !== undefined && enabled), + announced: this.in.announced, + }); + effect.cleanup(() => child.close()); + return child; + } + close() { this.#signals.close(); for (const entry of this.#references.values()) entry.request.close(); diff --git a/js/watch/src/contribution.test.ts b/js/watch/src/contribution.test.ts new file mode 100644 index 0000000000..34d72a2e82 --- /dev/null +++ b/js/watch/src/contribution.test.ts @@ -0,0 +1,160 @@ +import { expect, it } from "bun:test"; +import * as Catalog from "@moq/hang/catalog"; +import { Origin, Path } from "@moq/net"; +import { Effect, Signal } from "@moq/signals"; +import { Broadcast } from "./broadcast"; +import { Source } from "./video/source"; + +it("lists contributions without demand and keeps selected rendition IDs in their own broadcast", async () => { + const origin = new Origin.Producer(); + const parent = origin.createBroadcast(Path.from("room/main.hang")); + const contribution = origin.createBroadcast(Path.from("room/extra.hang")); + parent.announce(); + contribution.announce(); + const track = contribution.createTrack(Catalog.TRACK); + const catalog = Catalog.RootSchema.parse({ + video: { renditions: { video: { codec: "avc1.42001f", container: { kind: "legacy" } } } }, + contributions: { extra: { broadcast: "./extra.hang" }, missing: { broadcast: "./absent.hang" } }, + }); + const source = new Broadcast({ + origin, + name: Path.from("room/main.hang"), + catalogFormat: "manual", + catalog, + announced: false, + }); + const selection = new Effect(); + for (const [publisher, content] of [ + [parent, "source frames"], + [contribution, "contribution frames"], + ] as const) { + publisher.createTrack("video").writeFrame({ payload: new TextEncoder().encode(content) }); + } + try { + await source.out.catalog.changed(); + expect(source.out.catalog.peek()?.contributions).toEqual(catalog.contributions); + // Listing the reference does not subscribe to its existing catalog track. + expect(contribution.demand().used.peek()).toBe(false); + const child = source.contribution(selection, "extra"); + expect(child).toBeDefined(); + if (!child) throw new Error("missing selected contribution"); + await contribution.demand().used.changed(); + expect(contribution.demand().used.peek()).toBe(true); + track.writeFrame({ + timestamp: undefined, + payload: new TextEncoder().encode( + JSON.stringify({ + video: { renditions: { video: { codec: "vp8", container: { kind: "legacy" } } } }, + }), + ), + }); + await child.out.catalog.changed(); + expect(child.out.catalog.peek()?.video?.renditions.video.codec).toBe("vp8"); + expect(source.out.catalog.peek()?.video?.renditions.video.codec).toBe("avc1.42001f"); + const media = new Source({ broadcast: child, target: { name: "video" }, supported: async () => true }); + selection.cleanup(() => media.close()); + await media.out.config.changed(); + expect(media.out.track.peek()).toBe("video"); + const selected = child.relativeBroadcast(selection, media.out.config.peek()?.broadcast); + const frames = selected + ?.track(media.out.track.peek() ?? "") + .subscribe() + .ordered(); + const original = source.relativeBroadcast(selection, undefined)?.track("video").subscribe().ordered(); + try { + expect(await frames?.readString()).toBe("contribution frames"); + expect(await original?.readString()).toBe("source frames"); + } finally { + frames?.close(); + original?.close(); + } + expect(source.contribution(selection, "unknown").in.enabled.peek()).toBe(false); + const absent = origin.dynamic(Path.from("room/absent.hang")); + selection.cleanup(() => absent.close()); + const requests = absent.requested(); + const missing = source.contribution(selection, "missing"); + if (!missing) throw new Error("missing contribution selection"); + const { value: request } = await requests.next(); + expect(request?.path).toBe(Path.from("room/absent.hang")); + request?.reject(new Error("contribution unavailable")); + await missing.out.error.changed(); + expect(missing.out.active.peek()).toBeUndefined(); + expect(missing.out.error.peek()?.message).toBe("contribution unavailable"); + expect(source.out.catalog.peek()?.video?.renditions.video.codec).toBe("avc1.42001f"); + selection.close(); + await contribution.demand().unused(); + expect(contribution.demand().used.peek()).toBe(false); + } finally { + selection.close(); + source.close(); + origin.close(); + } +}); + +it("refuses malformed and escaping contribution entries", () => { + for (const entry of [null, {}, { broadcast: 42 }]) { + expect(() => Catalog.RootSchema.parse({ contributions: { extra: entry } })).toThrow(); + } + const catalog = Catalog.RootSchema.parse({ contributions: { extra: { broadcast: "../../escape" } } }); + expect(() => Catalog.checkResolvable(catalog, Path.from("room/main.hang"))).toThrow(Catalog.EscapingBroadcast); +}); + +it("keeps a reactive selection through unrelated catalog updates and follows replacement and removal", async () => { + const origin = new Origin.Producer(); + const main = origin.createBroadcast(Path.from("room/main.hang")); + main.announce(); + const first = origin.createBroadcast(Path.from("room/first.hang")); + const second = origin.createBroadcast(Path.from("room/second.hang")); + for (const publisher of [first, second]) { + publisher.announce(); + publisher.createTrack(Catalog.TRACK).writeFrame({ payload: new TextEncoder().encode("{}") }); + } + const catalog = new Signal(Catalog.RootSchema.parse({ contributions: { extra: { broadcast: "./first.hang" } } })); + const parent = new Broadcast({ + origin, + name: Path.from("room/main.hang"), + catalogFormat: "manual", + catalog, + announced: false, + }); + const selected = new Signal(undefined); + let selections = 0; + const owner = new Effect((effect) => { + selections += 1; + effect.set(selected, parent.contribution(effect, "extra")); + }); + try { + const child = await selected.changed(); + if (!child) throw new Error("selection did not create a child"); + await child.out.catalog.changed(); + const active = child.out.active.peek(); + let withdrawals = 0; + const dispose = first.demand().used.subscribe((used) => { + if (!used) withdrawals += 1; + }); + owner.cleanup(dispose); + catalog.set( + Catalog.RootSchema.parse({ + contributions: { extra: { broadcast: "./first.hang" }, other: { broadcast: "./other.hang" } }, + metadata: "changed", + }), + ); + await parent.out.catalog.changed(); + // The next selected-path update fences the unrelated update without a timer or flush count. + catalog.set(Catalog.RootSchema.parse({ contributions: { extra: { broadcast: "./second.hang" } } })); + await second.demand().used.changed(); + expect(selections).toBe(1); + expect(selected.peek()).toBe(child); + expect(active?.path).toBe(Path.from("room/first.hang")); + expect(child.out.active.peek()?.path).toBe(Path.from("room/second.hang")); + expect(withdrawals).toBe(1); + catalog.set(Catalog.RootSchema.parse({})); + await second.demand().unused(); + expect(child.in.enabled.peek()).toBe(false); + expect(selections).toBe(1); + } finally { + owner.close(); + parent.close(); + origin.close(); + } +}); diff --git a/rs/hang/src/catalog/consumer.rs b/rs/hang/src/catalog/consumer.rs index aa67dd7c70..53d82cc163 100644 --- a/rs/hang/src/catalog/consumer.rs +++ b/rs/hang/src/catalog/consumer.rs @@ -66,6 +66,7 @@ fn check_resolvable(base: &PathOwned, catalog: &Catalog) -> Result<()> { .chain(text.map(|(_, config)| config.broadcast.as_ref())) .chain(json.map(|(_, config)| config.broadcast.as_ref())) .chain(binary.map(|(_, config)| config.broadcast.as_ref())) + .chain(catalog.contributions.values().map(|entry| Some(&entry.broadcast))) .flatten() { if !rel.is_empty() && base.try_resolve(rel).is_none() { @@ -80,6 +81,12 @@ fn check_resolvable(base: &PathOwned, catalog: &Catalog) -> Result<()> { mod test { use super::*; + #[test] + fn contributions_cannot_escape_the_root() { + let catalog = Catalog::<()>::from_str(r#"{"contributions":{"extra":{"broadcast":"../../escape"}}}"#).unwrap(); + assert!(check_resolvable(&PathOwned::from("room/main.hang"), &catalog).is_err()); + } + #[tokio::test] async fn subscribe_reads_a_typed_root() { let broadcast = moq_net::broadcast::Info::new().produce(); diff --git a/rs/hang/src/catalog/root.rs b/rs/hang/src/catalog/root.rs index f8fd16d0be..cefc4c1752 100644 --- a/rs/hang/src/catalog/root.rs +++ b/rs/hang/src/catalog/root.rs @@ -13,7 +13,7 @@ use serde::{Deserialize, Serialize}; /// `archive` (each track's timeline and any durable recording), the optional `clock` (the /// broadcast's one wall-clock mapping), the data sections /// (`json`, `binary`) for application tracks that aren't media, and the optional `m2ts` (a -/// transport stream carried whole). +/// transport stream carried whole). `contributions` names whole broadcasts for explicit selection. /// Applications extend it with their own root sections (e.g. `scte35`) through `E`. The catalog /// does not deny unknown fields, so a base consumer ignores the extra sections and an extended /// catalog stays wire-compatible. See the `extension_roundtrip` test. @@ -29,6 +29,10 @@ use serde::{Deserialize, Serialize}; #[serde(bound(serialize = "E: Serialize", deserialize = "E: DeserializeOwned + Default"))] #[non_exhaustive] pub struct Catalog { + /// Whole broadcasts offered for explicit, lazy selection, keyed by contribution ID. + #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")] + pub contributions: std::collections::BTreeMap, + /// Video track information with multiple renditions. /// /// Contains a map of video track renditions that the viewer can choose from @@ -110,6 +114,13 @@ pub struct Catalog { pub ext: E, } +/// A whole broadcast whose catalog describes its own renditions. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] +pub struct Contribution { + /// Reference resolved relative to the catalog broadcast. + pub broadcast: moq_net::path::RelativeOwned, +} + impl Catalog<()> { /// The default name for the catalog track. pub const DEFAULT_NAME: &str = "catalog.json"; @@ -153,6 +164,7 @@ impl Catalog { /// Copy the media sections into a catalog with no application extension. pub fn media(&self) -> Catalog<()> { Catalog { + contributions: self.contributions.clone(), video: self.video.clone(), audio: self.audio.clone(), archive: self.archive.clone(), @@ -245,6 +257,21 @@ mod test { )); } + #[test] + fn contributions_round_trip_and_refuse_malformed_entries() { + let json = r#"{"contributions":{"extra":{"broadcast":"./extra.hang"}}}"#; + let catalog = Catalog::<()>::from_str(json).unwrap(); + assert_eq!(catalog.contributions["extra"].broadcast.as_str(), "extra.hang"); + assert_eq!(Catalog::<()>::from_str(&catalog.to_json().unwrap()).unwrap(), catalog); + for json in [ + r#"{"contributions":{"extra":null}}"#, + r#"{"contributions":{"extra":{}}}"#, + r#"{"contributions":{"extra":{"broadcast":42}}}"#, + ] { + assert!(Catalog::<()>::from_str(json).is_err()); + } + } + #[test] fn simple() { let mut encoded = r#"{