diff --git a/doc/concept/hang.md b/doc/concept/hang.md index ed97d7200e4..27fe6ee8626 100644 --- a/doc/concept/hang.md +++ b/doc/concept/hang.md @@ -72,6 +72,22 @@ A few things the catalog can express beyond decoder config: - **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 timeline. Publish independent sources as sibling broadcasts under one prefix, such as `robot/front.hang` and `robot/rear.hang`; each broadcast's audio and video entries are renditions of that source. In one Rust publisher, pass the same `moq_mux::Clock` to each `catalog::Config::with_clock`; JS publishers on one page already share the page clock. Consumers discover the siblings by prefix and align their timestamps through the advertised wall mapping; `@moq/watch` plays one broadcast at a time. - **Extensions.** The root is a loose object. Applications add their own sections (`scte35`, for example) next to the ones hang defines, optionally naming a track that carries the data. Every library exposes a way to write your section without clobbering the built-in ones, and readers ignore what they don't know. +## Media reports + +A catalog may name a publisher snapshot track with `stats: { track }` and +solicit viewer feedback with `echo: { path }`. Applications can use hang's +shared Rust and JS snapshot schemas to exchange per-rendition delivery and +playout measurements, alongside transport measurements. The echo path is +relative to the catalog broadcast; an application assigns one `.echo` name +per viewer and soliciting catalog, and controls the token rights under that +prefix. Echo broadcasts carry `feedback.json` without a media catalog, so +readers can filter them from playable content. + +These are schema and discovery primitives. The publishers and players do not +automatically produce reports. MSF catalogs omit both sections. See +[media stats](/concept/stats#media-reports) for the roles and the +[hang draft](/draft/moq-hang#media-stats) for their contract. + ## Text Captions and subtitles are their own tracks in a `text` section, not part of diff --git a/doc/concept/stats.md b/doc/concept/stats.md index 58e3cbef7fe..9f9fd4525ac 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -12,6 +12,30 @@ read them in any language. [`moq-stats`](https://docs.rs/moq-stats) is the Rust producer and consumer, and the relay's [`[stats]`](/bin/relay/config#stats) section turns it on. +## Media reports + +Media reports use hang's catalog discovery, separately from a relay's +traffic broadcasts below. `stats: { track }` names a publisher's snapshot; +`echo: { path }` solicits a viewer's `feedback.json` snapshot on an +application-named `.echo` broadcast. Both snapshots map the catalog's +rendition IDs to their counters and carry optional transport measurements. +The publisher reports only renditions it writes; a viewer reports a +referenced rendition under its ID in the catalog that listed it. + +The shared schemas are `hang::{stats, echo}::Snapshot` and +`Hang.Stats.SnapshotSchema` / `Hang.Echo.SnapshotSchema` in `@moq/hang`. +Container-specific sections can sit beside `transport` and `renditions`. +Counters are cumulative; readers derive rates over an interval and treat a +replacement or decrease as a reset. Bitrate, latency, RTT, estimated rate, +and sample age are gauges. Loss is cumulative lost bytes and packets, +with unavailable transport measurements absent. Missing rendition counters +mean zero, and JavaScript preserves omission. A known-zero gauge or transport +measurement stays explicit, so it remains distinct from unavailable. + +These types do not enable automatic reporting in a publisher or player. +See the [hang draft](/draft/moq-hang#media-stats) for track framing, +field units, and the feedback convention. + ## Broadcasts Each node publishes under a prefix, `.stats` by default, which diff --git a/drafts/draft-lcurley-moq-hang.md b/drafts/draft-lcurley-moq-hang.md index 38897d09fe7..a73e3204dd6 100644 --- a/drafts/draft-lcurley-moq-hang.md +++ b/drafts/draft-lcurley-moq-hang.md @@ -127,6 +127,8 @@ type Catalog = { "video": VideoSchema | undefined, "archive": ArchiveSchema | undefined, "clock": ClockSchema | undefined, + "stats": { "track": string } | undefined, + "echo": { "path": string } | undefined, "text": TextSchema | undefined, "json": JsonTracks | undefined, "binary": BinaryTracks | undefined, @@ -142,7 +144,7 @@ For example, a chat entry should name a chat track, not carry individual chat me This way catalog updates are rare and a client MAY choose to not subscribe. 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}}. +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, the application data tracks in {{data}}, and optional publisher statistics and solicited viewer feedback ({{media-stats}}). ## Clock {#clock} The catalog's root `clock` field is the broadcast's one continuous clock: @@ -602,6 +604,80 @@ If absent, `enabled` defaults to true; a publisher SHOULD only write it when fal Earlier versions defined a video `stalled` field instead. A publisher MUST NOT write `stalled`, and a consumer MUST ignore it. +# Media Statistics and Viewer Feedback {#media-stats} + +The optional catalog `stats` section names the publisher's snapshot track in the catalog's own broadcast: `{ "track": "stats.json" }`. +A publisher snapshot covers only renditions that publisher writes, omitting references to another broadcast. + +The optional catalog `echo` section solicits feedback: `{ "path": "viewers" }`. +The path is relative to the broadcast serving the catalog, with the same URL-style resolution and root containment as {{field-broadcast}}. +For example, from `room/live`, `viewers` resolves to `room/viewers`, and `live/viewers` resolves to `room/live/viewers`. +A consumer MUST reject a catalog whose echo path resolves above its connection root. +After reading the soliciting catalog, a viewer MAY announce one broadcast under that prefix at an application-assigned name plus `.echo`, for example `room/viewers/alice.echo`. +Each such broadcast belongs to one soliciting catalog and carries the fixed `feedback.json` track; it has no catalog of its own. +A reader recognizes feedback by the `.echo` suffix on the broadcast's final path segment, so it does not list feedback as playable media. +The application controls names and publish/subscribe authorization; a report is not authenticated beyond those rights. +Without `echo`, no feedback is solicited. + +An application may already carry its own `stats` or `echo` root key, so a reader MUST treat a `stats` value without a string `track`, or an `echo` value without a string `path`, as absent rather than refuse the catalog. + +Both tracks carry JSON snapshots, optionally with the `.z` sibling encoding of JSON snapshots. +The uncompressed track carries a complete snapshot in a group's single UTF-8 JSON frame. +The compressed sibling starts each group with a full snapshot and continues with RFC 7396 merge patches in the DEFLATE stream defined in {{compression}}. +A producer SHOULD update about once per second while observed; a report need not leave the process when unsubscribed. +A reader takes the newest complete snapshot and treats the reporting broadcast's replacement or decreasing cumulative counters as a reset. + +~~~ +type PublisherSnapshot = { + transport?: Transport, + renditions?: { [renditionId: string]: PublisherRendition }, + // ... any container or application sections ... +} +type ViewerSnapshot = { + transport?: Transport, + renditions?: { [renditionId: string]: ViewerRendition }, + // ... any container or application sections ... +} +type PublisherRendition = { + frames?: number, bytes?: number, keyframes?: number, skipped?: number, + bitrate?: number, +} +type ViewerRendition = { + frames?: number, bytes?: number, decoded?: number, late?: number, + errors?: number, stalls?: number, stalled?: number, underruns?: number, + arrival?: { timestamp: number, wall: number }, latency?: number, +} +type Transport = { + rtt?: number, rate?: number, lostBytes?: number, lostPackets?: number, age?: number, +} +~~~ + +Every number is a nonnegative integer no larger than 2^53 - 1; malformed or out-of-range known fields MUST be refused. +Unknown fields MUST be ignored; application and container sections MAY be flattened beside `transport` and `renditions`. +Absent rendition counters mean zero, and zero counters MAY be omitted; JavaScript preserves omission instead of synthesizing fields. +Unavailable gauges and transport measurements MUST remain absent, and a known zero MUST remain explicit. +A reader treats JSON `null` in an optional `stats` or `echo` catalog section, gauge, transport measurement, or `arrival` as absent; publishers omit unavailable fields. +An `arrival` MUST carry both `timestamp` and `wall`; a reader MUST refuse one missing either. +Rendition IDs are the unique keys from the soliciting catalog, independent of media kind and wire track name. +A viewer reports referenced renditions under the IDs in the catalog it read, even when their media comes from another broadcast. +An empty rendition entry is valid and represents zero cumulative counters; its ID MUST NOT be removed merely because its counters are zero. + +`frames` and `bytes` count cumulative media payload frames and bytes sent or received, according to the report's role. +Publisher `keyframes` counts sent keyframes; `skipped` counts frames skipped before sending. +Viewer `decoded`, `late`, and `errors` count decoded frames, frames too late for playout, and decode errors. +`stalls` counts playout stalls, `stalled` is cumulative stalled duration, and `underruns` counts audio buffer underruns. +`bitrate` is the latest target encoder bitrate, `latency` the latest playout latency, and `arrival` pairs the newest received media timestamp with its arrival wall time. +Durations and media timestamps are milliseconds; `arrival.wall` is milliseconds since the MoQ epoch (2020-01-01T00:00:00Z). +Sub-millisecond durations and timestamps are floored when encoded. +Rates are bits per second. + +`transport` describes the reporting connection: latest round-trip time `rtt`, estimated bitrate `rate`, cumulative loss counters `lostBytes` and `lostPackets`, and sample `age` at reporting time. +All transport measurements are optional; a browser may know only its PROBE round-trip time. +Loss is a pair of cumulative counters, never a gauge or fraction: a reader derives loss over the same interval as rendition rates and handles resets. +Gauges MUST NOT be summed across reports. + +The MSF catalog conversion carries neither `stats` nor `echo`; their transport through MSF `publishTracks` is not defined here. + # 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}}): @@ -1136,6 +1212,7 @@ This document has no IANA actions. {:numbered="false"} - Derived HLS and DASH exports hand disabled references to enabled renditions at the first record numbered past the advertised ones, keeping the record `sequence` as the HLS media sequence and starting a DASH period. +- Added optional `stats` and `echo` catalog sections, publisher/viewer snapshots keyed by rendition ID, and the `.echo` feedback broadcast convention. - Fixed a track's identity for its name ({{identity}}): codec, description, container, display aspect, sample rate, and channel count never change, while every other field, including size, bitrate, and framerate, updates in place. A change of identity takes a new track name, which a rendition may name with `track` while keeping its ID, or a new Epoch. - Specified MSF text metadata preservation and refusal of unrepresentable text packaging during gateway conversion. diff --git a/js/hang/src/catalog/consumer.test.ts b/js/hang/src/catalog/consumer.test.ts index 1a8a9a34f9d..1883e87981a 100644 --- a/js/hang/src/catalog/consumer.test.ts +++ b/js/hang/src/catalog/consumer.test.ts @@ -127,3 +127,29 @@ test("watch subscribes and yields typed catalog updates", async () => { producer.finish(); broadcast.close(); }); + +test("echo feedback prefix refuses root escape with the same URL resolution as renditions", () => { + for (const [base, reference, expected] of [ + ["room/live", "viewers", "room/viewers"], + ["room/live", "live/viewers", "room/live/viewers"], + ["room/live", "../viewers", "viewers"], + ["room/live", "../../viewers", undefined], + ["live", "../viewers", undefined], + ] as const) { + const root = { echo: { path: Moq.Path.normalizeRelative(reference) } }; + const path = Moq.Path.from(base); + if (expected === undefined) expect(() => checkResolvable(root, path)).toThrow(EscapingBroadcast); + else { + expect(checkResolvable(root, path)).toEqual(root); + expect(Moq.Path.tryResolve(path, root.echo.path)).toBe(Moq.Path.from(expected)); + } + } +}); + +test("watch refuses an echo prefix escaping the requested broadcast root", async () => { + await expect( + watchRequested({ echo: { path: Moq.Path.normalizeRelative("../../viewers") } }), + ).rejects.toBeInstanceOf(EscapingBroadcast); + const root = { echo: { path: Moq.Path.normalizeRelative("viewers") } }; + expect(await watchRequested(root)).toMatchObject({ value: root }); +}); diff --git a/js/hang/src/catalog/echo.ts b/js/hang/src/catalog/echo.ts new file mode 100644 index 00000000000..ebe05c39032 --- /dev/null +++ b/js/hang/src/catalog/echo.ts @@ -0,0 +1,11 @@ +import * as z from "@zod/mini"; +import { RelativeBroadcastSchema } from "./path"; + +/** The relative prefix under which viewers may publish feedback. */ +export const EchoSchema = z.object({ + // The prefix relative to the broadcast serving this catalog. + path: RelativeBroadcastSchema, +}); + +/** The relative prefix soliciting viewer feedback. */ +export type Echo = z.infer; diff --git a/js/hang/src/catalog/index.ts b/js/hang/src/catalog/index.ts index e744b765588..27e1e00c5a1 100644 --- a/js/hang/src/catalog/index.ts +++ b/js/hang/src/catalog/index.ts @@ -1,7 +1,7 @@ /** * JSON catalog schema and types describing a broadcast's tracks: audio/video decoder configs for * WebCodecs, container format, and rendition layout, the `archive` entry and root `clock`, - * plus the `json` and `binary` sections listing application data tracks that aren't media. + * plus the `json` and `binary` data sections and optional `stats` and `echo` discovery. * * @module */ @@ -25,6 +25,7 @@ export * from "./binary"; export * from "./clock"; export * from "./compression"; export * from "./container"; +export * from "./echo"; export * from "./format"; export * from "./hex"; export * from "./integers"; @@ -33,6 +34,7 @@ export * from "./mode"; export * from "./path"; export * from "./priority"; export * from "./root"; +export * from "./stats"; export * from "./text"; export * from "./track"; export type { VideoConfig } from "./video"; diff --git a/js/hang/src/catalog/root.test.ts b/js/hang/src/catalog/root.test.ts index f06098efe0e..26d84712e98 100644 --- a/js/hang/src/catalog/root.test.ts +++ b/js/hang/src/catalog/root.test.ts @@ -257,3 +257,24 @@ test("an ID used by both video and audio is refused", () => { const parsed = RootSchema.parse({ video: { renditions: { main: avc } }, audio: { renditions: { main: opus } } }); expect(() => checkRenditions(parsed)).toThrow(DuplicateRendition); }); + +test("shared stats and echo discovery round trips without changing old catalogs", async () => { + const fixture = await Bun.file( + new URL("../../../../rs/hang/fixtures/catalog-stats-echo.json", import.meta.url), + ).json(); + expect(RootSchema.parse(fixture)).toEqual(fixture); + expect(RootSchema.parse({})).toEqual({}); +}); + +test("a foreign stats or echo value decodes as absent instead of failing the catalog", () => { + for (const foreign of [{}, 5, "on", { track: 5 }, { path: ["a"] }, null]) { + for (const key of ["stats", "echo"]) { + const parsed = RootSchema.parse({ video: { renditions: {} }, [key]: foreign }); + expect(parsed.stats).toBeUndefined(); + expect(parsed.echo).toBeUndefined(); + } + } + const parsed = RootSchema.parse({ stats: { track: "stats.json" }, echo: { path: "viewers" } }); + expect(parsed.stats).toEqual({ track: "stats.json" }); + expect(parsed.echo?.path).toBe("viewers" as RelativeBroadcast); +}); diff --git a/js/hang/src/catalog/root.ts b/js/hang/src/catalog/root.ts index d8e7c459ee9..15367259265 100644 --- a/js/hang/src/catalog/root.ts +++ b/js/hang/src/catalog/root.ts @@ -6,11 +6,13 @@ import { ArchiveSchema } from "./archive"; import { AudioSchema } from "./audio"; import { BinarySchema } from "./binary"; import { ClockSchema } from "./clock"; +import { EchoSchema } from "./echo"; import { TRACK } from "./format"; import { JsonSchema } from "./json"; import type { RelativeBroadcast } from "./path"; import { PRIORITY } from "./priority"; -import { section } from "./section"; +import { named, section } from "./section"; +import { StatsSchema } from "./stats"; import { TextSchema } from "./text"; import { VideoSchema } from "./video"; @@ -20,7 +22,8 @@ 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. `stats` names the publisher's media snapshot track, and `echo` + * solicits viewer feedback under a relative broadcast prefix. * * 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 @@ -35,6 +38,10 @@ export const RootSchema = z.looseObject({ // The broadcast's one continuous clock, if the publisher exposes one. Independent of // `archive`: a live-only publisher exposes its mapping without creating a segment index. clock: z.optional(ClockSchema), + // Decoded leniently like `text` below: a value without a string `track` or `path` is someone + // else's and decodes as absent. See `named`. + stats: named(StatsSchema, "track"), + echo: named(EchoSchema, "path"), // `text`, `json`, and `binary` are generic enough keys that an application could have been // carrying its own before these sections were reserved, so a value that isn't a section decodes // as absent rather than failing the whole catalog. A value that IS a section but carries a @@ -114,6 +121,7 @@ export function checkResolvable(root: Root, base: Moq.Path.Valid): Root { if (broadcast && Path.tryResolve(base, broadcast) === undefined) throw new EscapingBroadcast(broadcast); } } + if (root.echo && Path.tryResolve(base, root.echo.path) === undefined) throw new EscapingBroadcast(root.echo.path); return root; } diff --git a/js/hang/src/catalog/section.ts b/js/hang/src/catalog/section.ts index 29511413ebe..1aa0b26aca5 100644 --- a/js/hang/src/catalog/section.ts +++ b/js/hang/src/catalog/section.ts @@ -32,3 +32,27 @@ export function section(schema: T, map: string) { ]), ); } + +/** + * Like {@link section}, for a section identified by one string field (`stats.track`, `echo.path`) + * rather than a map. Mirrors the Rust `deserialize_named`. + * + * @param schema - the section's own schema, applied when the field is a string. + * @param key - the section's required string field. + */ +export function named(schema: T, key: string) { + return z.optional( + z.union([ + schema, + z.pipe( + z.custom( + (value) => + typeof value !== "object" || + value === null || + typeof (value as Record)[key] !== "string", + ), + z.transform(() => undefined), + ), + ]), + ); +} diff --git a/js/hang/src/catalog/stats.ts b/js/hang/src/catalog/stats.ts new file mode 100644 index 00000000000..65098d6d2c9 --- /dev/null +++ b/js/hang/src/catalog/stats.ts @@ -0,0 +1,10 @@ +import * as z from "@zod/mini"; + +/** The publisher's optional media snapshot track in this broadcast. */ +export const StatsSchema = z.object({ + // The publisher's snapshot track name. + track: z.string(), +}); + +/** The publisher's media snapshot track in this broadcast. */ +export type Stats = z.infer; diff --git a/js/hang/src/echo.ts b/js/hang/src/echo.ts new file mode 100644 index 00000000000..fb22f7f5cae --- /dev/null +++ b/js/hang/src/echo.ts @@ -0,0 +1,70 @@ +/** + * Media report schemas with cumulative counters and optional gauges. + * Missing counters mean zero; parsing preserves omission. + * @module + */ + +import type { Path } from "@moq/net"; +import * as z from "@zod/mini"; +import { u53Schema } from "./catalog/integers"; +import { measurementSchema } from "./snapshot"; +import { TransportSchema } from "./stats"; + +/** The newest received media timestamp paired with its arrival wall time. */ +export const ArrivalSchema = z.object({ + // Media timestamp in milliseconds. + timestamp: u53Schema, + // Arrival milliseconds since the MoQ epoch. + wall: u53Schema, +}); +/** The newest received media timestamp and its wall-clock arrival. */ +export type Arrival = z.infer; + +/** Cumulative counters and latest playout measurements for one rendition. */ +export const RenditionSchema = z.object({ + // Cumulative media frames received. + frames: z.optional(u53Schema), + // Cumulative media payload bytes received. + bytes: z.optional(u53Schema), + // Cumulative media frames decoded. + decoded: z.optional(u53Schema), + // Cumulative frames received too late for playout. + late: z.optional(u53Schema), + // Cumulative decode errors. + errors: z.optional(u53Schema), + // Cumulative playout stalls. + stalls: z.optional(u53Schema), + // Cumulative stalled duration in milliseconds. + stalled: z.optional(u53Schema), + // Cumulative audio buffer underruns. + underruns: z.optional(u53Schema), + // The newest received media timestamp and its arrival wall time. + arrival: z.optional( + z.pipe( + z.nullable(ArrivalSchema), + z.transform((value) => value ?? undefined), + ), + ), + // Latest playout latency in milliseconds; absent when unavailable. + latency: measurementSchema, +}); +/** One rendition's delivery counters and latest playout measurements. */ +export type Rendition = z.infer; +/** One cumulative viewer report, keyed by the catalog's rendition IDs. */ +export const SnapshotSchema = z.looseObject({ + // Connection measurements; unavailable fields remain absent. + transport: z.optional(TransportSchema), + // Empty entries retain a rendition ID whose counters are all zero. + renditions: z.optional(z.record(z.string(), RenditionSchema)), +}); +/** One viewer report, with application or container sections flattened beside it. */ +export type Snapshot = z.infer; + +/** The suffix identifying viewer feedback broadcasts at announce time. */ +export const SUFFIX = ".echo"; +/** The snapshot track carried by an echo broadcast. */ +export const TRACK = "feedback.json"; +/** Whether a broadcast path identifies feedback rather than playable media. */ +export function isEcho(path: Path.Valid): boolean { + return path.endsWith(SUFFIX); +} diff --git a/js/hang/src/index.ts b/js/hang/src/index.ts index 83a4850789a..1035d7caf02 100644 --- a/js/hang/src/index.ts +++ b/js/hang/src/index.ts @@ -9,4 +9,9 @@ export * as Net from "@moq/net"; export * as Signals from "@moq/signals"; export * as Catalog from "./catalog"; export * as Container from "./container"; +/** Solicited viewer feedback schemas and broadcast names. */ +export * as Echo from "./echo"; + +/** Publisher media and transport snapshot schemas. */ +export * as Stats from "./stats"; export * as Timeline from "./timeline"; diff --git a/js/hang/src/snapshot.ts b/js/hang/src/snapshot.ts new file mode 100644 index 00000000000..9508136b797 --- /dev/null +++ b/js/hang/src/snapshot.ts @@ -0,0 +1,10 @@ +import * as z from "@zod/mini"; +import { u53Schema } from "./catalog/integers"; + +/** An optional measurement, with JSON null decoded as unavailable like Rust Option. */ +export const measurementSchema = z.optional( + z.pipe( + z.nullable(u53Schema), + z.transform((value) => value ?? undefined), + ), +); diff --git a/js/hang/src/stats.test.ts b/js/hang/src/stats.test.ts new file mode 100644 index 00000000000..221fff8b9bb --- /dev/null +++ b/js/hang/src/stats.test.ts @@ -0,0 +1,90 @@ +import { expect, test } from "bun:test"; +import { Path } from "@moq/net"; +import * as z from "@zod/mini"; +import * as Echo from "./echo"; +import * as Stats from "./stats"; + +for (const [name, schema] of [ + ["../../../rs/hang/fixtures/stats-snapshot.json", Stats.SnapshotSchema], + ["../../../rs/hang/fixtures/echo-snapshot.json", Echo.SnapshotSchema], + ["../../../rs/hang/fixtures/stats-zero.json", Stats.SnapshotSchema], + ["../../../rs/hang/fixtures/echo-zero.json", Echo.SnapshotSchema], +] as const) { + test(`shared Rust fixture ${name} round trips through JS`, async () => { + const input = await Bun.file(new URL(name, import.meta.url)).json(); + expect(JSON.parse(JSON.stringify(schema.parse(input)))).toEqual(input); + }); +} + +test("missing rendition counters mean zero while parsing preserves omission", () => { + const input = { renditions: { reset: {} } }; + const stats = Stats.SnapshotSchema.parse(input); + const echo = Echo.SnapshotSchema.parse(input); + expect(stats).toEqual(input); + expect(echo).toEqual(input); + expect(stats.renditions?.reset.frames ?? 0).toBe(0); + expect(echo.renditions?.reset.stalled ?? 0).toBe(0); + expect(stats.renditions?.reset.bitrate).toBeUndefined(); + expect(echo.renditions?.reset.latency).toBeUndefined(); + expect(stats.transport).toBeUndefined(); + expect(Object.hasOwn(stats, "transport")).toBe(false); + expect(Object.hasOwn(stats.renditions?.reset ?? {}, "frames")).toBe(false); + expect(Object.hasOwn(stats.renditions?.reset ?? {}, "bitrate")).toBe(false); + expect(Object.hasOwn(echo.renditions?.reset ?? {}, "arrival")).toBe(false); + expect(Object.hasOwn(echo.renditions?.reset ?? {}, "latency")).toBe(false); + expect(Stats.SnapshotSchema.parse({})).toEqual({}); + expect(Echo.SnapshotSchema.parse({})).toEqual({}); +}); + +test("optional null measurements mean unavailable rather than measured zero", () => { + expect( + JSON.parse( + JSON.stringify( + Stats.SnapshotSchema.parse({ + transport: { rtt: null, rate: null, lostBytes: null, lostPackets: null, age: null }, + renditions: { hd: { bitrate: null } }, + }), + ), + ), + ).toEqual({ transport: {}, renditions: { hd: {} } }); + expect(JSON.parse(JSON.stringify(Echo.RenditionSchema.parse({ latency: null, arrival: null })))).toEqual({}); +}); + +test("unknown track fields are ignored and flattened snapshot extensions survive", () => { + expect(Stats.RenditionSchema.parse({ future: 42 })).toEqual({}); + expect(Echo.RenditionSchema.parse({ future: 42 })).toEqual({}); + const input = { mpegts: { packets: 42 } }; + expect(Stats.SnapshotSchema.parse(input)).toEqual(input); + expect(Echo.SnapshotSchema.parse(input)).toEqual(input); + const extended = z.extend(Echo.SnapshotSchema, { mpegts: z.object({ packets: z.number() }) }); + expect(extended.parse(input)).toEqual(input); +}); + +test("shared invalid counters refuse instead of rounding", async () => { + const cases = (await Bun.file( + new URL("../../../rs/hang/fixtures/snapshot-invalid.json", import.meta.url), + ).json()) as unknown[]; + for (const value of cases) expect(() => Stats.RenditionSchema.parse(value)).toThrow(); + for (const value of [ + { stalled: -1 }, + { latency: 0.5 }, + { arrival: { timestamp: Number.MAX_SAFE_INTEGER + 1, wall: 0 } }, + { arrival: { timestamp: 0, wall: -1 } }, + { arrival: {} }, + { arrival: { timestamp: 0 } }, + { arrival: { wall: 0 } }, + { frames: null }, + { frames: Number.NaN }, + ]) + expect(() => Echo.RenditionSchema.parse(value)).toThrow(); + for (const value of [{ rtt: -1 }, { rate: 0.5 }, { lostBytes: Number.MAX_SAFE_INTEGER + 1 }]) + expect(() => Stats.TransportSchema.parse(value)).toThrow(); +}); + +test("feedback discovery agrees with the Rust fixed track and final path suffix", () => { + expect(Echo.TRACK).toBe("feedback.json"); + expect(Echo.SUFFIX).toBe(".echo"); + expect(Echo.isEcho(Path.from("room/alice.echo"))).toBe(true); + expect(Echo.isEcho(Path.from("room.echo/alice"))).toBe(false); + expect(Echo.isEcho(Path.from("room/alice.echoes"))).toBe(false); +}); diff --git a/js/hang/src/stats.ts b/js/hang/src/stats.ts new file mode 100644 index 00000000000..fefc20add26 --- /dev/null +++ b/js/hang/src/stats.ts @@ -0,0 +1,47 @@ +/** + * Media report schemas with cumulative counters and optional gauges. + * Missing counters mean zero; parsing preserves omission. + * @module + */ +import * as z from "@zod/mini"; +import { u53Schema } from "./catalog/integers"; +import { measurementSchema } from "./snapshot"; +/** Transport gauges and cumulative loss counters known to the observer. */ +export const TransportSchema = z.object({ + // Latest round-trip time in milliseconds. + rtt: measurementSchema, + // Latest estimated connection bitrate in bits per second. + rate: measurementSchema, + // Cumulative bytes reported lost by the transport. + lostBytes: measurementSchema, + // Cumulative packets reported lost by the transport. + lostPackets: measurementSchema, + // Transport sample age in milliseconds at reporting time. + age: measurementSchema, +}); +/** Transport measurements, absent when unavailable; a measured zero remains zero. */ +export type Transport = z.infer; +/** Cumulative counters and target bitrate for one locally published rendition. */ +export const RenditionSchema = z.object({ + // Cumulative media frames sent. + frames: z.optional(u53Schema), + // Cumulative media payload bytes sent. + bytes: z.optional(u53Schema), + // Cumulative keyframes sent. + keyframes: z.optional(u53Schema), + // Cumulative frames skipped before sending. + skipped: z.optional(u53Schema), + // Latest target encoder bitrate in bits per second; absent when unavailable. + bitrate: measurementSchema, +}); +/** One locally published rendition's counters and optional target bitrate. */ +export type Rendition = z.infer; +/** One cumulative publisher report, keyed by the catalog's rendition IDs. */ +export const SnapshotSchema = z.looseObject({ + // Connection measurements; unavailable fields remain absent. + transport: z.optional(TransportSchema), + // Empty entries retain a rendition ID whose counters are all zero. + renditions: z.optional(z.record(z.string(), RenditionSchema)), +}); +/** One publisher report, with application or container sections flattened beside it. */ +export type Snapshot = z.infer; diff --git a/quest/m2/latency-ledger.md b/quest/m2/latency-ledger.md index 3a034041873..7248e8e3ba8 100644 --- a/quest/m2/latency-ledger.md +++ b/quest/m2/latency-ledger.md @@ -33,10 +33,6 @@ probes, which was the right trade to get it running. This quest promotes them. so the sum-to-end-to-end identity (with its named `unaccounted` remainder) can be added once a spanning clock exists. -## Required - -- [Media stats schema](/quest/m2/stats/schema.md) - adds the stats and feedback snapshots this extends - ## Related - [QoS](/quest/m2/qos/README.md) - relay-side health, the same idea from the other end diff --git a/quest/m2/qos-publisher-timeliness.md b/quest/m2/qos-publisher-timeliness.md index eda75132bd7..4faf97ab37f 100644 --- a/quest/m2/qos-publisher-timeliness.md +++ b/quest/m2/qos-publisher-timeliness.md @@ -57,8 +57,3 @@ excluded. - [Viewer lag histogram](/quest/m2/qos/lag-histogram.md) - the egress histogram whose type and edges this reuses - [Stats prefix tracks](/quest/m2/stats-prefix-tracks.md) - the prefix tracks the drift histogram lands on, beside the totals - -## Related - -- [Media stats](/quest/m2/stats/schema.md) - the publisher's own view - of the same uplink, in the transport section of its stats track diff --git a/quest/m2/stats/README.md b/quest/m2/stats/README.md index d8b7a08ba6f..45195f92ee9 100644 --- a/quest/m2/stats/README.md +++ b/quest/m2/stats/README.md @@ -111,8 +111,6 @@ connection view and the dashboard flow. ## Required -- [Schema](/quest/m2/stats/schema.md) - hang defines the `stats` and - `echo` catalog sections, their snapshot types, and the draft text - [Rust reporters](/quest/m2/stats/rust.md) - the CLI, players, encoders, and moq-mux remuxes publish stats and feedback - [Browser reporters](/quest/m2/stats/js.md) - `` publishes diff --git a/quest/m2/stats/health.md b/quest/m2/stats/health.md index 78918b6f527..99511272a48 100644 --- a/quest/m2/stats/health.md +++ b/quest/m2/stats/health.md @@ -31,7 +31,7 @@ broadcast, which #4510 replaced with these tracks (the relay's own Guidance, to be settled while building: -- Inputs are the snapshot types from [the schema](/quest/m2/stats/schema.md): +- Inputs are the snapshot types from [the schema](/doc/concept/stats.md#media-reports): `transport` (rtt, rate, cumulative bytes and packets lost, sample age) and the per-rendition counters. Decided in the 2026-10-10 audit: rendition rates and transport loss use deltas of cumulative counters over the two @@ -61,10 +61,6 @@ change, from shared fixtures in both languages. Docs: the media section of `doc/ Public API: new health types in `hang` and `@moq/hang`. Wire: none. -## Required - -- [Schema](/quest/m2/stats/schema.md) - the snapshot types the model reads - ## Related - [QoS](/quest/m2/qos/README.md) - the relay's delivery counters, which a combined downstream verdict also reads diff --git a/quest/m2/stats/js.md b/quest/m2/stats/js.md index 2a90a395fad..d64cd020dd7 100644 --- a/quest/m2/stats/js.md +++ b/quest/m2/stats/js.md @@ -37,10 +37,6 @@ echo path of a catalog soliciting feedback. Both use the `@moq/hang` schemas. stats track and the `echo` attribute when it adds one: they would publish rendition IDs and per-track counters in plaintext beside it. -## Required - -- [Schema](/quest/m2/stats/schema.md) - the zod schemas this fills - ## Related - [Encrypted browser components](/quest/m2/e2ee/browser.md) - refuses stats and feedback on an encrypted broadcast diff --git a/quest/m2/stats/rust.md b/quest/m2/stats/rust.md index 10d333fbfc0..bc9e18e0bd5 100644 --- a/quest/m2/stats/rust.md +++ b/quest/m2/stats/rust.md @@ -38,6 +38,9 @@ reconciles it: a removed `echo` section, a changed path, or no longer watching unannounces the old broadcast. A name that is not a single path segment is refused at parse time. +- A viewer given `--echo` with `--catalog-format msf` refuses: the MSF + catalog does not carry an `echo` solicitation. Imports always emit the + MSF catalog beside hang, omitting `stats` and `echo` from MSF. - `moq export stats` and `moq export echo` are sinks that skip `.hang` media discovery, so they route around `catalog_format`. - `doc/bin/cli.md` documents the flags and sinks. @@ -49,10 +52,6 @@ `--echo` when it adds one: they would publish rendition IDs and per-track counters in plaintext beside it. -## Required - -- [Schema](/quest/m2/stats/schema.md) - the sections and snapshot types - ## Related - [TS health stats](/quest/m2/ts-health-stats.md) - adds TS-specific counters after this generic reporting plumbing lands diff --git a/quest/m2/stats/schema.md b/quest/m2/stats/schema.md deleted file mode 100644 index d9206c5e473..00000000000 --- a/quest/m2/stats/schema.md +++ /dev/null @@ -1,74 +0,0 @@ -# [M] hang defines the stats and echo sections and their snapshots - -## Goal - -`hang::Catalog` carries two optional root sections, `stats: { track }` and -`echo: { path }`. hang defines the snapshot types those tracks carry: the -publisher's per-rendition and transport counters, and a viewer's -per-rendition and transport feedback. Rust and JS parse the same fixtures, -and the hang draft specs both. Nothing produces them yet. - -## Plan - -- `rs/hang/src/catalog`: `Stats { track }` and `Echo { path }`, both - `#[non_exhaustive]`, as `Option` fields on `Catalog` that are omitted from - the wire when absent. `path` is relative to the broadcast serving the - catalog and resolves through `try_resolve`, like a rendition's - `broadcast`; `.echo` is appended to the result. A path that escapes - the root refuses the catalog. Shared fixtures in `rs/hang/fixtures` (beside - `catalog-clock.json`, which JS mirrors) pin both in both languages. - Additive on main. -- `rs/hang/src/stats.rs`: the publisher snapshot, - `Snapshot { transport, renditions: BTreeMap, #[serde(flatten)] ext: E }`. - The generic lets moq-mux flatten `{ mpegts: ts::stats::Snapshot }` in beside it, the - way `Catalog` takes `ts::Ext`. `Track` holds sent frames, sent bytes, - keyframes, skipped frames, and the target bitrate as a gauge. -- `rs/hang/src/echo.rs`: - `Snapshot { transport, renditions: BTreeMap, #[serde(flatten)] ext: E }`, - the same flatten as `stats::Snapshot`, so a TS export folds its - `mpegts` egress counters in (decided 2026-10-12, from - [#5360](https://github.com/moq-dev/moq/issues/5360)). `Track` - holds: - - received frames and bytes, decoded, late, and decode errors; - - stalls, stalled duration, and underruns; - - the newest media timestamp received, with the wall time it arrived; - - the playout latency, as a gauge. -- Both snapshots key by the catalog's rendition ID. Feedback covers a - rendition that references another broadcast under its ID in the catalog - that lists it; the publisher snapshot omits referenced renditions (decided - in the [README](/quest/m2/stats/README.md)). -- `Transport` is shared: rtt, estimated rate, cumulative bytes and packets - lost, and sample age. Decided in the 2026-10-10 audit: loss is cumulative - counters, not a reported gauge, so health derives loss over the same - interval as rendition rates and handles resets. RTT and estimated rate - remain gauges. Every field is optional, because a browser has only PROBE rtt. -- Every field is defaulted, zero and `None` are omitted, unknown fields are - ignored, and each type is `#[non_exhaustive]`. Durations are milliseconds - and rates bits per second, as in `moq-stats`. -- hang names the `.echo` suffix and the feedback track (a helper such as - `hang::echo::is_echo(path)`), so a reader filters at announce time and a - publisher subscribes by a known name. -- `js/hang`: zod schemas mirroring both sections and all three snapshot - types, field for field. Fixtures are shared through `js/test`. -- `drafts/draft-lcurley-moq-hang.md` specs the two sections, the snapshot - schemas, and the `.echo` convention: one broadcast per soliciting catalog, - under the catalog's echo path, announced after the viewer reads the - catalog. Validate with `just drafts check`. -- **MSF carries neither section yet** (decided 2026-10-12). The MSF - conversion in `rs/moq-mux` copies only the media sections and `ext`. A - viewer given `--echo` with `--catalog-format msf` refuses, since it would - never see an `echo` section to answer. An import has no catalog-format - flag and always emits the MSF catalog beside the hang one, so that catalog - omits `stats` and `echo` the way it omits `json` and `binary`; hang - readers still get them. Carrying them through MSF `publishTracks` is - [MSF feedback tracks](/quest/m3/msf-publish-tracks.md). -- Docs: `doc/concept/hang.md`, and a media section in `doc/concept/stats.md`. -- Tests: - - fixtures round-trip in both languages; - - an old catalog without the sections parses unchanged; - - a snapshot with an unknown field parses. - -## Related - -- [#4145](https://github.com/moq-dev/moq/pull/4145) - the closed moq-stats - extension design; salvage its field docs and serde helpers diff --git a/quest/m2/ts-health-stats.md b/quest/m2/ts-health-stats.md index b3ddf2d14c8..d72d74cb681 100644 --- a/quest/m2/ts-health-stats.md +++ b/quest/m2/ts-health-stats.md @@ -13,11 +13,11 @@ counter rates and PID quiet times. Decided while planning [#1838](https://github.com/moq-dev/moq/issues/1838): The quest audit confirmed this stays in m2 and owns TS counter serialization -and integration. Rust reporters in m1 supplies only the generic per-rendition +and integration. Rust reporters supplies only the generic per-rendition plumbing; it does not implement this quest's `mpegts` payloads. - **Surface through the stats plumbing**, as the issue settled. The - [media stats schema](/quest/m2/stats/schema.md) settles the shape: an + [media stats schema](/doc/concept/stats.md#media-reports) settles the shape: an import flattens `ts::stats::Snapshot` into the publisher's stats snapshot under the catalog's `mpegts` key, the counters stay owned by moq-mux beside `ts::Ext`, and `hang` stays TS-free. @@ -25,11 +25,21 @@ plumbing; it does not implement this quest's `mpegts` payloads. [#5360](https://github.com/moq-dev/moq/issues/5360)). An export is a viewer, so `moq export ts --echo ` folds `ts::stats::Snapshot` into its solicited `.echo` snapshot under `mpegts`, through the `E` flatten the - [media stats schema](/quest/m2/stats/schema.md) gives `echo::Snapshot`. + [media stats schema](/doc/concept/stats.md#media-reports) gives `echo::Snapshot`. The egress is where an operator otherwise cannot see per-PID outages: a 57 s video hole passed every TR 101 290 P1 check in the reporter's lab, and only per-PID access-unit counts caught it. There is no export `--stats` flag. +- **Re-timing gets its own section beside `mpegts`** (decided 2026-10-16, + on [#5643](https://github.com/moq-dev/moq/pull/5643)). An export that + re-times onto an output clock reports what `ts::stats::Snapshot` cannot + hold: `ts::stats::Export`'s stream-wide `dropped`, `trimmed`, `gaps`, + `drift`, and `out_of_tolerance`, which its conversion to `Snapshot` + discards, plus the release margin and the output clock's offset from the + source. They describe the export's own output + clock rather than viewer playout, so they sit flat beside `transport` and + `renditions` now instead of waiting for the + [latency ledger](/quest/m2/latency-ledger.md)'s `.echo` shape. - **Counters sum, gauges do not.** Every check is a cumulative counter, which keeps `.z` deltas small and lets an aggregate sum; the PID's `quiet` is a gauge and merges newest-wins. @@ -46,5 +56,4 @@ plumbing; it does not implement this quest's `mpegts` payloads. ## Required -- [Media stats schema](/quest/m2/stats/schema.md) - the snapshot the counters flatten into - [Rust reporters](/quest/m2/stats/rust.md) - `moq import --stats` and the stats interval diff --git a/quest/m3/msf-publish-tracks.md b/quest/m3/msf-publish-tracks.md index 1afa898b441..17ff695c5a3 100644 --- a/quest/m3/msf-publish-tracks.md +++ b/quest/m3/msf-publish-tracks.md @@ -10,15 +10,12 @@ consumer asks. ## Plan Raised in [#5360](https://github.com/moq-dev/moq/issues/5360). Until then the -[media stats schema](/quest/m2/stats/schema.md) leaves both sections out of -the MSF catalog and refuses a viewer's `--echo` under `--catalog-format msf` +[media stats schema](/doc/concept/stats.md#media-reports) leaves both sections out of +the MSF catalog. The [Rust reporters](/quest/m2/stats/rust.md) quest refuses a +viewer's `--echo` under `--catalog-format msf` when it adds that flag (decided 2026-10-12). Open: how the entry names the hang payload, since MSF fixes its packaging values in the draft; and whether to add a CMCD v2 Event Mode encoding (draft-wilaw-moq-cmcd-event-timeline) of the overlapping fields (stalls, dropped frames, latency, throughput, errors) for consumers that ingest CMCD. - -## Required - -- [Media stats schema](/quest/m2/stats/schema.md) - the `stats` and `echo` sections this maps diff --git a/quest/m3/stats-encoder-feedback.md b/quest/m3/stats-encoder-feedback.md index 7927095ca8a..943af7ff149 100644 --- a/quest/m3/stats-encoder-feedback.md +++ b/quest/m3/stats-encoder-feedback.md @@ -86,7 +86,6 @@ Open, to settle before starting (moved from the ## Required -- [Schema](/quest/m2/stats/schema.md) - the `echo` section and feedback snapshot - [Rust reporters](/quest/m2/stats/rust.md) - the viewers that report and the CLI it wires diff --git a/quest/m3/teleop-robot.md b/quest/m3/teleop-robot.md index 16d5e2e3695..9244529adb6 100644 --- a/quest/m3/teleop-robot.md +++ b/quest/m3/teleop-robot.md @@ -79,7 +79,7 @@ The framing is where the guarantee lives, not the subscription flags: opaque-bytes ones for binary frames (`moq-flate`). - Per-stage timestamp instrumentation, generalised from moq-boy's `status` track. Check it against the publisher-reported stats track - ([media stats](/quest/m2/stats/schema.md), moq#2734) before adding a + ([media stats](/doc/concept/stats.md#media-reports), moq#2734) before adding a second stats surface. Capability only: publishing a competitive benchmark is out of scope, because Transitive's breakdown puts camera plus USB at roughly 100 ms of a 130 ms glass-to-glass total, so we would mostly be diff --git a/rs/hang/fixtures/catalog-stats-echo.json b/rs/hang/fixtures/catalog-stats-echo.json new file mode 100644 index 00000000000..f7e41a4d73c --- /dev/null +++ b/rs/hang/fixtures/catalog-stats-echo.json @@ -0,0 +1,18 @@ +{ + "video": { + "renditions": { + "hd": { + "codec": "avc1.64001f", + "container": { + "kind": "legacy" + } + } + } + }, + "stats": { + "track": "stats.json" + }, + "echo": { + "path": "viewers" + } +} diff --git a/rs/hang/fixtures/echo-snapshot.json b/rs/hang/fixtures/echo-snapshot.json new file mode 100644 index 00000000000..d749c98d868 --- /dev/null +++ b/rs/hang/fixtures/echo-snapshot.json @@ -0,0 +1,27 @@ +{ + "transport": { + "rtt": 24, + "rate": 7000000, + "lostBytes": 150, + "lostPackets": 1, + "age": 6 + }, + "renditions": { + "hd": { + "frames": 890, + "bytes": 1030000, + "decoded": 880, + "late": 3, + "errors": 2, + "stalls": 1, + "stalled": 120, + "underruns": 4, + "arrival": { + "timestamp": 30000, + "wall": 175184640000 + }, + "latency": 200 + }, + "reset": {} + } +} diff --git a/rs/hang/fixtures/echo-zero.json b/rs/hang/fixtures/echo-zero.json new file mode 100644 index 00000000000..4d27c8e7bb2 --- /dev/null +++ b/rs/hang/fixtures/echo-zero.json @@ -0,0 +1,19 @@ +{ + "transport": { + "rtt": 0, + "rate": 0, + "lostBytes": 0, + "lostPackets": 0, + "age": 0 + }, + "renditions": { + "reset": {}, + "hd": { + "arrival": { + "timestamp": 0, + "wall": 0 + }, + "latency": 0 + } + } +} diff --git a/rs/hang/fixtures/snapshot-invalid.json b/rs/hang/fixtures/snapshot-invalid.json new file mode 100644 index 00000000000..230a1c61cf4 --- /dev/null +++ b/rs/hang/fixtures/snapshot-invalid.json @@ -0,0 +1,17 @@ +[ + { + "frames": -1 + }, + { + "bytes": 0.5 + }, + { + "keyframes": 9007199254740992 + }, + { + "skipped": null + }, + { + "bitrate": -1 + } +] diff --git a/rs/hang/fixtures/stats-snapshot.json b/rs/hang/fixtures/stats-snapshot.json new file mode 100644 index 00000000000..54f5510c7a5 --- /dev/null +++ b/rs/hang/fixtures/stats-snapshot.json @@ -0,0 +1,19 @@ +{ + "transport": { + "rtt": 20, + "rate": 8000000, + "lostBytes": 300, + "lostPackets": 2, + "age": 5 + }, + "renditions": { + "hd": { + "frames": 900, + "bytes": 1048576, + "keyframes": 30, + "skipped": 4, + "bitrate": 6000000 + }, + "reset": {} + } +} diff --git a/rs/hang/fixtures/stats-zero.json b/rs/hang/fixtures/stats-zero.json new file mode 100644 index 00000000000..5ce6af14b2c --- /dev/null +++ b/rs/hang/fixtures/stats-zero.json @@ -0,0 +1,15 @@ +{ + "transport": { + "rtt": 0, + "rate": 0, + "lostBytes": 0, + "lostPackets": 0, + "age": 0 + }, + "renditions": { + "reset": {}, + "hd": { + "bitrate": 0 + } + } +} diff --git a/rs/hang/src/catalog/consumer.rs b/rs/hang/src/catalog/consumer.rs index aa67dd7c70a..c2cd3f7a356 100644 --- a/rs/hang/src/catalog/consumer.rs +++ b/rs/hang/src/catalog/consumer.rs @@ -73,6 +73,11 @@ fn check_resolvable(base: &PathOwned, catalog: &Catalog) -> Result<()> { } } + if let Some(echo) = &catalog.echo + && base.try_resolve(&echo.path).is_none() + { + return Err(crate::Error::EscapingBroadcast(echo.path.to_string())); + } Ok(()) } @@ -80,6 +85,44 @@ fn check_resolvable(base: &PathOwned, catalog: &Catalog) -> Result<()> { mod test { use super::*; + #[test] + fn echo_prefix_cannot_escape_the_catalog_root() { + let base = moq_net::PathOwned::from("room/live"); + let mut catalog = Catalog::<()>::from_str(include_str!("../../fixtures/catalog-stats-echo.json")).unwrap(); + assert!(check_resolvable(&base, &catalog).is_ok()); + let echo = catalog.echo.as_mut().unwrap(); + echo.path = moq_net::path::RelativeOwned::from("../viewers"); + assert!(check_resolvable(&base, &catalog).is_ok()); + catalog.echo.as_mut().unwrap().path = moq_net::path::RelativeOwned::from("../../viewers"); + assert!(matches!( + check_resolvable(&base, &catalog), + Err(crate::Error::EscapingBroadcast(_)) + )); + } + + #[tokio::test] + async fn subscribe_validates_echo_prefix_before_yielding() { + for (reference, accepted) in [("viewers", true), ("../../viewers", false)] { + let mut info = moq_net::broadcast::Info::new(); + info.path = "room/live".into(); + let broadcast = info.produce(); + let track = broadcast + .create_track(Catalog::DEFAULT_NAME, Catalog::default_track_info()) + .unwrap(); + let mut consumer = Catalog::<()>::subscribe(&broadcast.consume()).await.unwrap(); + let published = serde_json::json!({ "echo": { "path": reference } }).to_string(); + let mut group = track.append_group().unwrap(); + group.write_frame(moq_net::Timestamp::ZERO, published).unwrap(); + group.finish().unwrap(); + let result = consumer.next().await; + if accepted { + assert!(result.unwrap().unwrap().echo.is_some()); + } else { + assert!(matches!(result, Err(crate::Error::EscapingBroadcast(_)))); + } + } + } + #[tokio::test] async fn subscribe_reads_a_typed_root() { let broadcast = moq_net::broadcast::Info::new().produce(); diff --git a/rs/hang/src/catalog/echo.rs b/rs/hang/src/catalog/echo.rs new file mode 100644 index 00000000000..a56863c40a9 --- /dev/null +++ b/rs/hang/src/catalog/echo.rs @@ -0,0 +1,16 @@ +use serde::{Deserialize, Serialize}; + +/// The relative broadcast prefix under which viewers may publish feedback. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub struct Echo { + /// The feedback prefix relative to the broadcast serving this catalog. + pub path: moq_net::path::RelativeOwned, +} + +impl Echo { + /// Solicit feedback under a prefix relative to the catalog's broadcast. + pub fn new(path: impl Into) -> Self { + Self { path: path.into() } + } +} diff --git a/rs/hang/src/catalog/mod.rs b/rs/hang/src/catalog/mod.rs index 1dfa04f7410..c6106892571 100644 --- a/rs/hang/src/catalog/mod.rs +++ b/rs/hang/src/catalog/mod.rs @@ -12,6 +12,7 @@ mod clock; mod compression; mod consumer; mod container; +mod echo; mod hex; mod json; mod m2ts; @@ -20,6 +21,7 @@ mod mode; mod priority; mod reference; mod root; +mod stats; mod text; mod video; @@ -30,12 +32,14 @@ pub use clock::*; pub use compression::*; pub use consumer::*; pub use container::*; +pub use echo::*; pub use json::*; pub use m2ts::*; pub use mode::*; pub use priority::*; pub use reference::*; pub use root::*; +pub use stats::*; pub use text::*; pub use video::*; @@ -95,3 +99,32 @@ where serde_json::from_value(value).map_err(serde::de::Error::custom) } + +/// Decode an optional section identified by one string field (`stats.track`, `echo.path`). +/// +/// A value without that string field is someone else's key and decodes as absent, for the same +/// reason as [`deserialize_section`]; a value with it is ours and must decode. +fn deserialize_named<'de, D, T>(deserializer: D, key: &str) -> Result, D::Error> +where + D: serde::Deserializer<'de>, + T: serde::de::DeserializeOwned, +{ + use serde::Deserialize; + let value = serde_json::Value::deserialize(deserializer)?; + + if !value.get(key).is_some_and(serde_json::Value::is_string) { + return Ok(None); + } + + serde_json::from_value(value) + .map(Some) + .map_err(serde::de::Error::custom) +} + +pub(crate) fn deserialize_stats<'de, D: serde::Deserializer<'de>>(deserializer: D) -> Result, D::Error> { + deserialize_named(deserializer, "track") +} + +pub(crate) fn deserialize_echo<'de, D: serde::Deserializer<'de>>(deserializer: D) -> Result, D::Error> { + deserialize_named(deserializer, "path") +} diff --git a/rs/hang/src/catalog/root.rs b/rs/hang/src/catalog/root.rs index 11144461964..0805de65656 100644 --- a/rs/hang/src/catalog/root.rs +++ b/rs/hang/src/catalog/root.rs @@ -13,7 +13,8 @@ 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), plus `stats` naming the publisher's media snapshot track and +/// `echo` soliciting viewer feedback under a relative broadcast prefix. /// 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. @@ -107,6 +108,19 @@ pub struct Catalog { #[serde(default, skip_serializing_if = "Option::is_none")] pub m2ts: Option, + /// The publisher's optional media snapshot track. + /// + /// A `stats` value without a string `track` decodes as absent rather than failing the + /// catalog, for the same reason as [`json`](Self::json). + #[serde(default, deserialize_with = "crate::catalog::deserialize_stats")] + pub stats: Option, + + /// The optional prefix soliciting viewer feedback broadcasts. + /// + /// An `echo` value without a string `path` decodes as absent, the same as [`stats`](Self::stats). + #[serde(default, deserialize_with = "crate::catalog::deserialize_echo")] + pub echo: Option, + /// Application-defined root sections, flattened alongside the base catalog sections. #[serde(flatten)] pub ext: E, @@ -751,6 +765,26 @@ mod test { } } + /// `stats` and `echo` are as generic as `json`: a value without their string field is someone + /// else's and decodes as absent, while one with it is ours. + #[test] + fn a_foreign_stats_or_echo_value_does_not_fail_the_catalog() { + for foreign in ["{}", "5", r#""on""#, r#"{"track":5}"#, r#"{"path":["a"]}"#, "null"] { + for section in ["stats", "echo"] { + let wire = format!(r#"{{"video":{{"renditions":{{}}}},"{section}":{foreign}}}"#); + let catalog = Catalog::<()>::from_str(&wire) + .unwrap_or_else(|err| panic!("{section}: {foreign} took the catalog down: {err}")); + assert!( + catalog.stats.is_none() && catalog.echo.is_none(), + "{section}: {foreign}" + ); + } + } + let catalog = Catalog::<()>::from_str(r#"{"stats":{"track":"stats.json"},"echo":{"path":"viewers"}}"#).unwrap(); + assert_eq!(catalog.stats, Some(crate::catalog::Stats::new("stats.json"))); + assert_eq!(catalog.echo, Some(crate::catalog::Echo::new("viewers"))); + } + #[test] fn data_sections_stay_off_the_wire_when_empty() { // Empty sections stay off the wire, video and audio included. diff --git a/rs/hang/src/catalog/stats.rs b/rs/hang/src/catalog/stats.rs new file mode 100644 index 00000000000..c6cadb445c2 --- /dev/null +++ b/rs/hang/src/catalog/stats.rs @@ -0,0 +1,16 @@ +use serde::{Deserialize, Serialize}; + +/// The publisher's media snapshot track in this broadcast. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub struct Stats { + /// The name of the publisher snapshot track. + pub track: String, +} + +impl Stats { + /// Name the publisher snapshot track. + pub fn new(track: impl Into) -> Self { + Self { track: track.into() } + } +} diff --git a/rs/hang/src/echo.rs b/rs/hang/src/echo.rs new file mode 100644 index 00000000000..6c7b13cc613 --- /dev/null +++ b/rs/hang/src/echo.rs @@ -0,0 +1,97 @@ +//! Solicited viewer feedback, carried on an echo broadcast. +use crate::snapshot::{Millis, Number, is_zero}; +use crate::stats::Transport; +use serde::{Deserialize, Serialize}; +use std::{collections::BTreeMap, time::Duration}; +/// One cumulative viewer report, keyed by the catalog's rendition IDs. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[serde(default, rename_all = "camelCase")] +#[serde(bound(serialize = "E: Serialize", deserialize = "E: serde::de::DeserializeOwned + Default"))] +#[non_exhaustive] +pub struct Snapshot { + /// The reporting connection's transport measurements. + #[serde(skip_serializing_if = "Transport::is_empty")] + pub transport: Transport, + /// Per-rendition counters, including entries with all counters reset to zero. + #[serde(skip_serializing_if = "BTreeMap::is_empty")] + pub renditions: BTreeMap, + /// Application-defined root sections flattened beside the media report. + #[serde(flatten)] + pub ext: E, +} + +/// Cumulative counters and latest playout measurements for one rendition. +#[serde_with::serde_as] +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[serde(default, rename_all = "camelCase")] +#[non_exhaustive] +pub struct Rendition { + /// Cumulative media frames received. + #[serde_as(as = "Number")] + #[serde(skip_serializing_if = "is_zero")] + pub frames: u64, + /// Cumulative media payload bytes received. + #[serde_as(as = "Number")] + #[serde(skip_serializing_if = "is_zero")] + pub bytes: u64, + /// Cumulative frames decoded. + #[serde_as(as = "Number")] + #[serde(skip_serializing_if = "is_zero")] + pub decoded: u64, + /// Cumulative frames received too late for playout. + #[serde_as(as = "Number")] + #[serde(skip_serializing_if = "is_zero")] + pub late: u64, + /// Cumulative decode errors. + #[serde_as(as = "Number")] + #[serde(skip_serializing_if = "is_zero")] + pub errors: u64, + /// Cumulative playout stalls. + #[serde_as(as = "Number")] + #[serde(skip_serializing_if = "is_zero")] + pub stalls: u64, + /// Cumulative audio buffer underruns. + #[serde_as(as = "Number")] + #[serde(skip_serializing_if = "is_zero")] + pub underruns: u64, + /// Cumulative time spent stalled. + #[serde_as(as = "Millis")] + #[serde(skip_serializing_if = "Duration::is_zero")] + pub stalled: Duration, + /// The newest media timestamp received and its arrival wall time. + #[serde(skip_serializing_if = "Option::is_none")] + pub arrival: Option, + /// Latest playout latency. + #[serde_as(as = "Option")] + #[serde(skip_serializing_if = "Option::is_none")] + pub latency: Option, +} + +/// The newest received media timestamp paired with its arrival wall time. +#[serde_with::serde_as] +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] +#[non_exhaustive] +pub struct Arrival { + /// The media timestamp, encoded in milliseconds. + #[serde_as(as = "Millis")] + pub timestamp: moq_net::Timestamp, + /// The arrival wall time, encoded in milliseconds since the MoQ epoch. + #[serde_as(as = "crate::snapshot::Wall")] + pub wall: std::time::SystemTime, +} + +impl Arrival { + /// Pair a received media timestamp with the wall time it arrived. + pub fn new(timestamp: moq_net::Timestamp, wall: std::time::SystemTime) -> Self { + Self { timestamp, wall } + } +} + +/// The suffix identifying viewer feedback broadcasts at announce time. +pub const SUFFIX: &str = ".echo"; +/// The snapshot track carried by an echo broadcast. +pub const TRACK: &str = "feedback.json"; +/// Whether a broadcast path identifies viewer feedback rather than playable media. +pub fn is_echo(path: &moq_net::Path) -> bool { + path.as_str().ends_with(SUFFIX) +} diff --git a/rs/hang/src/lib.rs b/rs/hang/src/lib.rs index cfe65b273ed..adc5354ec1a 100644 --- a/rs/hang/src/lib.rs +++ b/rs/hang/src/lib.rs @@ -15,6 +15,13 @@ //! //! See the [moq-mux](https://crates.io/crates/moq-mux) crate for importing existing media formats into hang broadcasts. mod error; +mod snapshot; + +/// Publisher media and transport snapshots. +pub mod stats; + +/// Solicited viewer feedback and echo broadcast names. +pub mod echo; /// The catalog is used to describe the available media tracks and codecs. pub mod catalog; diff --git a/rs/hang/src/snapshot.rs b/rs/hang/src/snapshot.rs new file mode 100644 index 00000000000..c8b1f99cf93 --- /dev/null +++ b/rs/hang/src/snapshot.rs @@ -0,0 +1,197 @@ +//! Shared JSON adapters for media snapshots. +use serde::{Deserialize, Serialize}; +use serde_with::{DeserializeAs, SerializeAs}; +use std::time::Duration; + +pub(crate) fn is_zero(value: &u64) -> bool { + *value == 0 +} +pub(crate) struct Number; +impl SerializeAs for Number { + fn serialize_as(value: &u64, serializer: S) -> Result { + if *value > crate::catalog::MAX_SAFE_INTEGER { + return Err(serde::ser::Error::custom("snapshot integer exceeds JSON safe range")); + } + value.serialize(serializer) + } +} +impl<'de> DeserializeAs<'de, u64> for Number { + fn deserialize_as>(deserializer: D) -> Result { + let value = u64::deserialize(deserializer)?; + if value > crate::catalog::MAX_SAFE_INTEGER { + return Err(serde::de::Error::custom("snapshot integer exceeds JSON safe range")); + } + Ok(value) + } +} + +pub(crate) struct Millis; +impl SerializeAs for Millis { + fn serialize_as(value: &Duration, serializer: S) -> Result { + let value = u64::try_from(value.as_millis()).map_err(serde::ser::Error::custom)?; + Number::serialize_as(&value, serializer) + } +} +impl<'de> DeserializeAs<'de, Duration> for Millis { + fn deserialize_as>(deserializer: D) -> Result { + Number::deserialize_as(deserializer).map(Duration::from_millis) + } +} +impl SerializeAs for Millis { + fn serialize_as(value: &moq_net::Timestamp, serializer: S) -> Result { + let value = u64::try_from(value.as_millis()).map_err(serde::ser::Error::custom)?; + Number::serialize_as(&value, serializer) + } +} +impl<'de> DeserializeAs<'de, moq_net::Timestamp> for Millis { + fn deserialize_as>(deserializer: D) -> Result { + moq_net::Timestamp::from_millis(Number::deserialize_as(deserializer)?).map_err(serde::de::Error::custom) + } +} + +pub(crate) struct Wall; +fn epoch() -> std::time::SystemTime { + std::time::UNIX_EPOCH + Duration::from_millis(crate::catalog::MOQ_EPOCH_UNIX_MILLIS) +} +impl SerializeAs for Wall { + fn serialize_as(value: &std::time::SystemTime, serializer: S) -> Result { + let duration = value.duration_since(epoch()).map_err(serde::ser::Error::custom)?; + Millis::serialize_as(&duration, serializer) + } +} +impl<'de> DeserializeAs<'de, std::time::SystemTime> for Wall { + fn deserialize_as>(deserializer: D) -> Result { + epoch() + .checked_add(Millis::deserialize_as(deserializer)?) + .ok_or_else(|| serde::de::Error::custom("snapshot wall clock overflow")) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{Catalog, echo, stats}; + + fn roundtrip(input: &str) + where + T: serde::de::DeserializeOwned + Serialize, + { + let value: serde_json::Value = serde_json::from_str(input).unwrap(); + let decoded: T = serde_json::from_value(value.clone()).unwrap(); + assert_eq!(serde_json::to_value(decoded).unwrap(), value); + } + + #[test] + fn shared_snapshots_roundtrip() { + roundtrip::(include_str!("../fixtures/stats-snapshot.json")); + roundtrip::(include_str!("../fixtures/echo-snapshot.json")); + roundtrip::(include_str!("../fixtures/catalog-stats-echo.json")); + roundtrip::(include_str!("../fixtures/stats-zero.json")); + roundtrip::(include_str!("../fixtures/echo-zero.json")); + } + + #[test] + fn defaults_and_unknown_fields() { + assert_eq!( + serde_json::to_value(stats::Snapshot::<()>::default()).unwrap(), + serde_json::json!({}) + ); + assert_eq!( + serde_json::to_value(echo::Snapshot::<()>::default()).unwrap(), + serde_json::json!({}) + ); + let rendition: stats::Rendition = serde_json::from_str(r#"{"future":17}"#).unwrap(); + assert_eq!(rendition, stats::Rendition::default()); + assert_eq!(serde_json::to_value(rendition).unwrap(), serde_json::json!({})); + let catalog = Catalog::<()>::from_str("{}").unwrap(); + assert!(catalog.stats.is_none() && catalog.echo.is_none()); + assert_eq!(catalog.to_json().unwrap(), "{}"); + } + + #[test] + fn missing_counters_mean_zero_but_unknown_measurements_stay_absent() { + let stats: stats::Snapshot = serde_json::from_str(r#"{"renditions":{"reset":{}}}"#).unwrap(); + assert_eq!(stats.renditions["reset"].frames, 0); + assert_eq!(stats.renditions["reset"].bitrate, None); + assert_eq!(stats.transport.rtt, None); + assert_eq!(stats.transport.lost_packets, None); + let echo: echo::Snapshot = serde_json::from_str(r#"{"renditions":{"reset":{}}}"#).unwrap(); + assert_eq!(echo.renditions["reset"].stalled, Duration::ZERO); + assert_eq!(echo.renditions["reset"].latency, None); + } + + #[test] + fn optional_null_measurements_are_unavailable() { + roundtrip::(r#"{"renditions":{"hd":{}}}"#); + let stats: stats::Snapshot = + serde_json::from_str(r#"{"transport":{"rtt":null},"renditions":{"hd":{"bitrate":null}}}"#).unwrap(); + assert_eq!(stats.transport.rtt, None); + assert_eq!(stats.renditions["hd"].bitrate, None); + let echo: echo::Snapshot = + serde_json::from_str(r#"{"renditions":{"hd":{"latency":null,"arrival":null}}}"#).unwrap(); + assert_eq!(echo.renditions["hd"].latency, None); + assert_eq!(echo.renditions["hd"].arrival, None); + } + + #[test] + fn extensions_roundtrip_at_both_roles() { + #[derive(Serialize, Deserialize, Debug, Default, PartialEq)] + struct Ext { + mpegts: serde_json::Value, + } + let input = r#"{"mpegts":{"packets":42}}"#; + roundtrip::>(input); + roundtrip::>(input); + let base: stats::Snapshot = serde_json::from_str(input).unwrap(); + assert!(base.renditions.is_empty()); + } + + #[test] + fn malformed_counters_refuse_instead_of_rounding() { + let cases: Vec = + serde_json::from_str(include_str!("../fixtures/snapshot-invalid.json")).unwrap(); + for case in cases { + assert!( + serde_json::from_value::(case.clone()).is_err(), + "{case}" + ); + } + for input in [ + r#"{"stalled":-1}"#, + r#"{"latency":0.5}"#, + r#"{"arrival":{"timestamp":9007199254740992,"wall":0}}"#, + r#"{"arrival":{"timestamp":0,"wall":-1}}"#, + r#"{"arrival":{}}"#, + r#"{"arrival":{"timestamp":0}}"#, + r#"{"arrival":{"wall":0}}"#, + ] { + assert!(serde_json::from_str::(input).is_err(), "{input}"); + } + let rendition = stats::Rendition { + frames: crate::catalog::MAX_SAFE_INTEGER + 1, + ..Default::default() + }; + assert!(serde_json::to_value(rendition).is_err()); + } + + #[test] + fn echo_suffix_filters_only_the_broadcast_name() { + assert!(echo::is_echo(&moq_net::Path::new("room/alice.echo"))); + assert!(!echo::is_echo(&moq_net::Path::new("room.echo/alice"))); + assert!(!echo::is_echo(&moq_net::Path::new("room/alice.echoes"))); + assert_eq!(echo::TRACK, "feedback.json"); + } + + #[test] + fn media_and_wall_units_are_milliseconds() { + let report: echo::Snapshot = serde_json::from_str(include_str!("../fixtures/echo-snapshot.json")).unwrap(); + let arrival = report.renditions["hd"].arrival.as_ref().unwrap(); + assert_eq!(arrival.timestamp, moq_net::Timestamp::from_millis(30_000).unwrap()); + assert_eq!( + arrival.wall.duration_since(epoch()).unwrap(), + Duration::from_millis(175_184_640_000) + ); + assert_eq!(report.renditions["hd"].stalled, Duration::from_millis(120)); + assert_eq!(report.transport.rtt, Some(Duration::from_millis(24))); + } +} diff --git a/rs/hang/src/stats.rs b/rs/hang/src/stats.rs new file mode 100644 index 00000000000..fceeff10402 --- /dev/null +++ b/rs/hang/src/stats.rs @@ -0,0 +1,85 @@ +//! Publisher media and transport snapshots. +use crate::snapshot::{Millis, Number, is_zero}; +use serde::{Deserialize, Serialize}; +use std::{collections::BTreeMap, time::Duration}; +/// One cumulative publisher report, keyed by the catalog's rendition IDs. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[serde(default, rename_all = "camelCase")] +#[serde(bound(serialize = "E: Serialize", deserialize = "E: serde::de::DeserializeOwned + Default"))] +#[non_exhaustive] +pub struct Snapshot { + /// The reporting connection's transport measurements. + #[serde(skip_serializing_if = "Transport::is_empty")] + pub transport: Transport, + /// Per-rendition counters, including entries with all counters reset to zero. + #[serde(skip_serializing_if = "BTreeMap::is_empty")] + pub renditions: BTreeMap, + /// Application-defined root sections flattened beside the media report. + #[serde(flatten)] + pub ext: E, +} + +/// Cumulative counters for one locally published rendition. +#[serde_with::serde_as] +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[serde(default, rename_all = "camelCase")] +#[non_exhaustive] +pub struct Rendition { + /// Cumulative frames sent. + #[serde_as(as = "Number")] + #[serde(skip_serializing_if = "is_zero")] + pub frames: u64, + /// Cumulative media payload bytes sent. + #[serde_as(as = "Number")] + #[serde(skip_serializing_if = "is_zero")] + pub bytes: u64, + /// Cumulative keyframes sent. + #[serde_as(as = "Number")] + #[serde(skip_serializing_if = "is_zero")] + pub keyframes: u64, + /// Cumulative frames skipped before sending. + #[serde_as(as = "Number")] + #[serde(skip_serializing_if = "is_zero")] + pub skipped: u64, + /// Latest target encoder bitrate, in bits per second. + #[serde_as(as = "Option")] + #[serde(skip_serializing_if = "Option::is_none")] + pub bitrate: Option, +} +/// Transport gauges and cumulative loss counters known to the observer. +#[serde_with::serde_as] +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[serde(default, rename_all = "camelCase")] +#[non_exhaustive] +pub struct Transport { + /// Latest round-trip time. + #[serde_as(as = "Option")] + #[serde(skip_serializing_if = "Option::is_none")] + pub rtt: Option, + /// Latest estimated connection bitrate, in bits per second. + #[serde_as(as = "Option")] + #[serde(skip_serializing_if = "Option::is_none")] + pub rate: Option, + /// Cumulative bytes reported lost by the transport. + #[serde_as(as = "Option")] + #[serde(skip_serializing_if = "Option::is_none")] + pub lost_bytes: Option, + /// Cumulative packets reported lost by the transport. + #[serde_as(as = "Option")] + #[serde(skip_serializing_if = "Option::is_none")] + pub lost_packets: Option, + /// Age of the transport sample when the snapshot was taken. + #[serde_as(as = "Option")] + #[serde(skip_serializing_if = "Option::is_none")] + pub age: Option, +} +impl Transport { + /// Whether the observer has no transport measurements to encode. + pub(crate) fn is_empty(&self) -> bool { + self.rtt.is_none() + && self.rate.is_none() + && self.lost_bytes.is_none() + && self.lost_packets.is_none() + && self.age.is_none() + } +} diff --git a/rs/moq-mux/src/catalog/hang/consumer.rs b/rs/moq-mux/src/catalog/hang/consumer.rs index 0c7465b79e5..35f5a4ddf1e 100644 --- a/rs/moq-mux/src/catalog/hang/consumer.rs +++ b/rs/moq-mux/src/catalog/hang/consumer.rs @@ -20,8 +20,8 @@ use crate::catalog::CatalogFormat; pub struct Consumer { inner: moq_json::snapshot::Consumer>, format: CatalogFormat, - /// Path of the broadcast serving this catalog, which every rendition's `broadcast` - /// reference resolves against. + /// Path of the broadcast serving this catalog, which rendition references and the + /// echo prefix resolve against. base: PathOwned, } @@ -55,7 +55,7 @@ impl Consumer { /// Poll for the next catalog update. /// /// Fails with [`Error::EscapingBroadcast`](crate::Error::EscapingBroadcast) if any rendition's - /// `broadcast` reference walks above the root. Such a reference names no broadcast, and the + /// `broadcast` reference or echo prefix walks above the root. Such a reference names no broadcast, and the /// root is the consumer's authorized subtree, so it is an attempt to reference content the /// consumer cannot name. The whole catalog is rejected rather than the one rendition: a /// publisher that emits one has a bug (or is hostile), and silently serving the rest hides @@ -88,10 +88,10 @@ impl From for Consumer { } } -/// Reject the catalog if any track's `broadcast` reference walks above the root. +/// Reject the catalog if a track reference or echo prefix walks above the root. /// -/// Every section carrying a `broadcast` reference must be chained here; a section left out -/// silently exempts its tracks from the containment check. +/// Every section carrying a relative broadcast path must be included here; a section left out +/// silently exempts its paths from the containment check. fn check_resolvable(base: &PathOwned, catalog: &Catalog) -> Result<()> { let video = catalog.video.renditions.iter(); let audio = catalog.audio.renditions.iter(); @@ -116,6 +116,11 @@ fn check_resolvable(base: &PathOwned, catalog: &Catalog) -> Re } } + if let Some(echo) = &catalog.echo + && base.try_resolve(&echo.path).is_none() + { + return Err(crate::Error::EscapingBroadcast(echo.path.to_string())); + } Ok(()) } @@ -256,6 +261,42 @@ mod test { publish_catalog_served(published).await } + fn soliciting(reference: &str) -> Catalog { + let (mut catalog, _) = catalog_payload("here"); + catalog.echo = Some(hang::catalog::Echo::new( + moq_net::path::Relative::new(reference).into_owned(), + )); + catalog + } + + #[tokio::test] + async fn echo_prefix_is_checked_before_the_native_reader_yields() { + for reference in ["viewers", "../viewers", ""] { + assert!(publish_catalog(soliciting(reference)).unwrap().unwrap().echo.is_some()); + } + for reference in ["../../viewers", "../../../elsewhere"] { + let result = publish_catalog(soliciting(reference)); + assert!( + matches!(result, Err(crate::Error::EscapingBroadcast(_))), + "{reference}: {result:?}" + ); + } + } + + #[tokio::test] + async fn echo_prefix_is_checked_against_the_dynamic_served_path() { + assert!( + publish_catalog_served(soliciting("../viewers")) + .await + .unwrap() + .unwrap() + .echo + .is_some() + ); + let result = publish_catalog_served(soliciting("../../../elsewhere")).await; + assert!(matches!(result, Err(crate::Error::EscapingBroadcast(_))), "{result:?}"); + } + #[test] fn refuses_oversized_catalog_before_reconciling_renditions() { let mut published = Catalog::<()>::default(); diff --git a/rs/moq-mux/src/catalog/hang/ext.rs b/rs/moq-mux/src/catalog/hang/ext.rs index 90957aa130a..49fe88c5c23 100644 --- a/rs/moq-mux/src/catalog/hang/ext.rs +++ b/rs/moq-mux/src/catalog/hang/ext.rs @@ -39,8 +39,8 @@ impl CatalogExt for () {} /// cross. Publish/consume a [`hang::Catalog`] and use [`set`](Self::set)/[`get`](Self::get). /// The default extension stays `()` (unknown sections dropped); opt into `Extra` explicitly. /// -/// `video`, `audio`, `text`, `archive`, `clock`, `json`, `binary`, and the retired `timeline` -/// key are reserved, so decoding and [`set`](Self::set) reject them to keep the wire JSON free of +/// `video`, `audio`, `text`, `archive`, `clock`, `json`, `binary`, `stats`, `echo`, and the +/// retired `timeline` key are reserved, so decoding and [`set`](Self::set) reject them to keep the wire JSON free of /// duplicate or retired keys. #[derive(Serialize, Clone, Default, Debug, PartialEq)] #[serde(transparent)] @@ -61,7 +61,7 @@ impl<'de> Deserialize<'de> for Extra { fn reserved(name: &str) -> bool { matches!( name, - "video" | "audio" | "text" | "archive" | "clock" | "json" | "binary" | "timeline" + "video" | "audio" | "text" | "archive" | "clock" | "json" | "binary" | "stats" | "echo" | "timeline" ) || moq_msf::reserved_root(name) } @@ -87,7 +87,8 @@ impl Extra { } /// Set (or replace) a section. Errors if `name` collides with a reserved member of either - /// catalog: hang's base sections (`video`/`audio`/`text`/`archive`/`clock`/`json`/`binary`), + /// catalog: hang's base sections (`video`/`audio`/`text`/`archive`/`clock`/`json`/`binary`/ + /// `stats`/`echo`), /// the retired `timeline` key that the wire format forbids beside `archive`, or a root member /// MSF defines ([`moq_msf::reserved_root`]). /// @@ -214,10 +215,15 @@ mod test { .set_section("timeline", serde_json::json!({})), Err(crate::Error::ReservedSection(_)) )); - assert!(matches!( - producer.modify().unwrap().set_section("clock", serde_json::json!({})), - Err(crate::Error::ReservedSection(_)) - )); + for name in ["clock", "stats", "echo"] { + assert!( + matches!( + producer.modify().unwrap().set_section(name, serde_json::json!({})), + Err(crate::Error::ReservedSection(_)) + ), + "{name} must be refused", + ); + } // Nor can a member MSF defines: the same sections ride the MSF track, where a // collision would serialize as a duplicate JSON key rather than an error. diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index 5524175255c..f68fb6be486 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -1328,6 +1328,17 @@ mod test { } } + #[test] + fn msf_omits_media_report_discovery() { + let catalog = + hang::Catalog::<()>::from_str(include_str!("../../../hang/fixtures/catalog-stats-echo.json")).unwrap(); + let msf = to_msf::<()>(&catalog, &moq_net::Path::empty()); + let value = serde_json::to_value(msf).unwrap(); + assert!(value.get("stats").is_none()); + assert!(value.get("echo").is_none()); + assert!(!value["tracks"].as_array().unwrap().is_empty()); + } + #[tokio::test(start_paused = true)] async fn abort_preserves_the_cause_on_every_catalog_format_and_clone() { let mut broadcast = moq_net::broadcast::Info::default().produce();