From 47c62cc4cff67c25c4857a220c5e384d42bea397 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 17:41:08 -0700 Subject: [PATCH 1/8] chore: claim hang stats schema quest Co-Authored-By: GPT-6 From b9da83c8e68d99bc84a3f8ab5e7205341529f852 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 18:10:38 -0700 Subject: [PATCH 2/8] feat(hang): define media stats and viewer feedback schemas Co-Authored-By: GPT-6 --- doc/concept/hang.md | 16 ++ doc/concept/stats.md | 24 +++ drafts/draft-lcurley-moq-hang.md | 78 ++++++++- js/hang/src/catalog/consumer.test.ts | 26 +++ js/hang/src/catalog/echo.ts | 12 ++ js/hang/src/catalog/index.ts | 4 +- js/hang/src/catalog/root.test.ts | 8 + js/hang/src/catalog/root.ts | 18 ++- js/hang/src/catalog/stats.ts | 10 ++ js/hang/src/echo.ts | 70 ++++++++ js/hang/src/index.ts | 5 + js/hang/src/snapshot.ts | 10 ++ js/hang/src/stats.test.ts | 87 ++++++++++ js/hang/src/stats.ts | 47 ++++++ quest/m2/latency-ledger.md | 4 - quest/m2/qos-publisher-timeliness.md | 5 - quest/m2/stats/README.md | 2 - quest/m2/stats/health.md | 6 +- quest/m2/stats/js.md | 4 - quest/m2/stats/rust.md | 7 +- quest/m2/stats/schema.md | 74 --------- quest/m2/teleop/README.md | 3 - quest/m2/ts-health-stats.md | 7 +- quest/m3/msf-publish-tracks.md | 9 +- quest/m3/stats-encoder-feedback.md | 1 - quest/m3/teleop-robot.md | 2 +- rs/hang/fixtures/catalog-stats-echo.json | 18 +++ rs/hang/fixtures/echo-snapshot.json | 27 ++++ rs/hang/fixtures/echo-zero.json | 16 ++ rs/hang/fixtures/snapshot-invalid.json | 17 ++ rs/hang/fixtures/stats-snapshot.json | 19 +++ rs/hang/fixtures/stats-zero.json | 15 ++ rs/hang/src/catalog/consumer.rs | 43 +++++ rs/hang/src/catalog/echo.rs | 10 ++ rs/hang/src/catalog/mod.rs | 4 + rs/hang/src/catalog/root.rs | 9 +- rs/hang/src/catalog/stats.rs | 10 ++ rs/hang/src/echo.rs | 105 ++++++++++++ rs/hang/src/lib.rs | 7 + rs/hang/src/snapshot.rs | 197 +++++++++++++++++++++++ rs/hang/src/stats.rs | 85 ++++++++++ rs/moq-mux/src/catalog/producer.rs | 11 ++ 42 files changed, 1015 insertions(+), 117 deletions(-) create mode 100644 js/hang/src/catalog/echo.ts create mode 100644 js/hang/src/catalog/stats.ts create mode 100644 js/hang/src/echo.ts create mode 100644 js/hang/src/snapshot.ts create mode 100644 js/hang/src/stats.test.ts create mode 100644 js/hang/src/stats.ts delete mode 100644 quest/m2/stats/schema.md create mode 100644 rs/hang/fixtures/catalog-stats-echo.json create mode 100644 rs/hang/fixtures/echo-snapshot.json create mode 100644 rs/hang/fixtures/echo-zero.json create mode 100644 rs/hang/fixtures/snapshot-invalid.json create mode 100644 rs/hang/fixtures/stats-snapshot.json create mode 100644 rs/hang/fixtures/stats-zero.json create mode 100644 rs/hang/src/catalog/echo.rs create mode 100644 rs/hang/src/catalog/stats.rs create mode 100644 rs/hang/src/echo.rs create mode 100644 rs/hang/src/snapshot.rs create mode 100644 rs/hang/src/stats.rs diff --git a/doc/concept/hang.md b/doc/concept/hang.md index e924764cf93..80c1c2f9d7d 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 398f5365004..3c2ca6a44b8 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 1bf41110859..5a9a307a403 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,78 @@ 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. + +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]: PublisherTrack }, + // ... any container or application sections ... +} +type ViewerSnapshot = { + transport?: Transport, + renditions?: { [renditionId: string]: ViewerTrack }, + // ... any container or application sections ... +} +type PublisherTrack = { + frames?: number, bytes?: number, keyframes?: number, skipped?: number, + bitrate?: number, +} +type ViewerTrack = { + 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 absent `arrival.timestamp` means zero and an absent `arrival.wall` means the MoQ epoch. +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}}): @@ -1123,6 +1197,8 @@ This document has no IANA actions. ## moq-hang-04 {:numbered="false"} +- 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. - Specified conversion between relative broadcast references and MSF track namespaces without changing track identity, reading a namespace aligned with the catalog path's trailing segments as the catalog itself. 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..2d49603169a --- /dev/null +++ b/js/hang/src/catalog/echo.ts @@ -0,0 +1,12 @@ +import { Path } from "@moq/net"; +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: z._default(RelativeBroadcastSchema, Path.normalizeRelative("")), +}); + +/** 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..d3e161a9539 100644 --- a/js/hang/src/catalog/root.test.ts +++ b/js/hang/src/catalog/root.test.ts @@ -257,3 +257,11 @@ 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({}); +}); diff --git a/js/hang/src/catalog/root.ts b/js/hang/src/catalog/root.ts index d8e7c459ee9..49d2796d191 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 { 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,18 @@ 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), + stats: z.optional( + z.pipe( + z.nullable(StatsSchema), + z.transform((value) => value ?? undefined), + ), + ), + echo: z.optional( + z.pipe( + z.nullable(EchoSchema), + z.transform((value) => value ?? undefined), + ), + ), // `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 +129,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/stats.ts b/js/hang/src/catalog/stats.ts new file mode 100644 index 00000000000..8717c4ce2ea --- /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._default(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..8693f1a33e5 --- /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; omitted means zero. + timestamp: z.optional(u53Schema), + // Arrival milliseconds since the MoQ epoch; omitted means that epoch. + wall: z.optional(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 TrackSchema = 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 Track = 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(), TrackSchema)), +}); +/** 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..5fd45a66462 --- /dev/null +++ b/js/hang/src/stats.test.ts @@ -0,0 +1,87 @@ +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"; + +const fixtures = new URL("../../../rs/hang/fixtures/", import.meta.url); + +for (const [name, schema] of [ + ["stats-snapshot.json", Stats.SnapshotSchema], + ["echo-snapshot.json", Echo.SnapshotSchema], + ["stats-zero.json", Stats.SnapshotSchema], + ["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, fixtures)).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.TrackSchema.parse({ latency: null, arrival: null })))).toEqual({}); +}); + +test("unknown track fields are ignored and flattened snapshot extensions survive", () => { + expect(Stats.TrackSchema.parse({ future: 42 })).toEqual({}); + expect(Echo.TrackSchema.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("snapshot-invalid.json", fixtures)).json()) as unknown[]; + for (const value of cases) expect(() => Stats.TrackSchema.parse(value)).toThrow(); + for (const value of [ + { stalled: -1 }, + { latency: 0.5 }, + { arrival: { timestamp: Number.MAX_SAFE_INTEGER + 1 } }, + { arrival: { wall: -1 } }, + { frames: null }, + { frames: Number.NaN }, + ]) + expect(() => Echo.TrackSchema.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..2b5174a0be1 --- /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 TrackSchema = 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 Track = 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(), TrackSchema)), +}); +/** 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 0dccc383e36..84723ad9b30 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/m1/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 5425497a314..f0c75b77dd6 100644 --- a/quest/m2/qos-publisher-timeliness.md +++ b/quest/m2/qos-publisher-timeliness.md @@ -57,8 +57,3 @@ excluded. - [Viewer lag histogram](/quest/m1/qos/lag-histogram.md) - the egress histogram whose type and edges this reuses - [Stats prefix tracks](/quest/m1/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 000355cf3ab..12e1b78f422 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 c487b1fd406..0f77fe2d244 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/m1/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/teleop/README.md b/quest/m2/teleop/README.md index dfa8b609057..bc113ded981 100644 --- a/quest/m2/teleop/README.md +++ b/quest/m2/teleop/README.md @@ -95,9 +95,6 @@ stating plainly because it is what a builder is comparing against. controller commands a vehicle at a time, with explicit handoff and a stated authorization boundary, parked in m3 - [e2ee](/quest/m2/e2ee/README.md) - the answer for a protected control link -- [Media stats](/quest/m2/stats/schema.md) - publisher-reported stats - on a catalog-announced track (moq#2734); teleop's latency instrumentation - extends those types rather than adding a second stats surface - [Video hardware validation](/quest/m3/video-hardware.md) - the VAAPI run that covers Intel ground robots and NUC companions - [CLI packaging](/quest/m2/cli-packaging.md) - ships the V4L2-M2M encoder the diff --git a/quest/m2/ts-health-stats.md b/quest/m2/ts-health-stats.md index b3ddf2d14c8..a6c2613e6c9 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,7 +25,7 @@ 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 @@ -46,5 +46,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..b176e009631 --- /dev/null +++ b/rs/hang/fixtures/echo-zero.json @@ -0,0 +1,16 @@ +{ + "transport": { + "rtt": 0, + "rate": 0, + "lostBytes": 0, + "lostPackets": 0, + "age": 0 + }, + "renditions": { + "reset": {}, + "hd": { + "arrival": {}, + "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..3b83ce964c5 --- /dev/null +++ b/rs/hang/src/catalog/echo.rs @@ -0,0 +1,10 @@ +use serde::{Deserialize, Serialize}; + +/// The relative broadcast prefix under which viewers may publish feedback. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[serde(default)] +#[non_exhaustive] +pub struct Echo { + /// The feedback prefix relative to the broadcast serving this catalog. + pub path: moq_net::path::RelativeOwned, +} diff --git a/rs/hang/src/catalog/mod.rs b/rs/hang/src/catalog/mod.rs index 1dfa04f7410..b7e4eb1c868 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::*; diff --git a/rs/hang/src/catalog/root.rs b/rs/hang/src/catalog/root.rs index 11144461964..5b77107dafa 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,12 @@ pub struct Catalog { #[serde(default, skip_serializing_if = "Option::is_none")] pub m2ts: Option, + /// The publisher's optional media snapshot track. + pub stats: Option, + + /// The optional prefix soliciting viewer feedback broadcasts. + pub echo: Option, + /// Application-defined root sections, flattened alongside the base catalog sections. #[serde(flatten)] pub ext: E, diff --git a/rs/hang/src/catalog/stats.rs b/rs/hang/src/catalog/stats.rs new file mode 100644 index 00000000000..29ad857bc44 --- /dev/null +++ b/rs/hang/src/catalog/stats.rs @@ -0,0 +1,10 @@ +use serde::{Deserialize, Serialize}; + +/// The publisher's media snapshot track in this broadcast. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[serde(default)] +#[non_exhaustive] +pub struct Stats { + /// The name of the publisher snapshot track. + pub track: String, +} diff --git a/rs/hang/src/echo.rs b/rs/hang/src/echo.rs new file mode 100644 index 00000000000..93e7388217a --- /dev/null +++ b/rs/hang/src/echo.rs @@ -0,0 +1,105 @@ +//! 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 Track { + /// 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)] +#[serde(default, rename_all = "camelCase")] +#[non_exhaustive] +pub struct Arrival { + /// The media timestamp, encoded in milliseconds. + #[serde_as(as = "Millis")] + #[serde(skip_serializing_if = "timestamp_zero")] + pub timestamp: moq_net::Timestamp, + /// The arrival wall time, encoded in milliseconds since the MoQ epoch. + #[serde_as(as = "crate::snapshot::Wall")] + #[serde(skip_serializing_if = "crate::snapshot::wall_zero")] + pub wall: std::time::SystemTime, +} +impl Default for Arrival { + fn default() -> Self { + Self { + timestamp: moq_net::Timestamp::ZERO, + wall: crate::snapshot::wall_default(), + } + } +} + +/// 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) +} + +fn timestamp_zero(timestamp: &moq_net::Timestamp) -> bool { + timestamp.is_zero() +} 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..cc183d0f4cd --- /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) +} +pub(crate) fn wall_default() -> std::time::SystemTime { + epoch() +} +pub(crate) fn wall_zero(value: &std::time::SystemTime) -> bool { + *value == epoch() +} +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 track: stats::Track = serde_json::from_str(r#"{"future":17}"#).unwrap(); + assert_eq!(track, stats::Track::default()); + assert_eq!(serde_json::to_value(track).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}}"#, + r#"{"arrival":{"wall":-1}}"#, + ] { + assert!(serde_json::from_str::(input).is_err(), "{input}"); + } + let track = stats::Track { + frames: crate::catalog::MAX_SAFE_INTEGER + 1, + ..Default::default() + }; + assert!(serde_json::to_value(track).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(wall_default()).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..bec49525c4d --- /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 Track { + /// 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/producer.rs b/rs/moq-mux/src/catalog/producer.rs index 0b80c54512e..fe604960e86 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -1313,6 +1313,17 @@ mod test { /// sitting above media can't starve it. Regression: the MSF and timeline tracks /// were minted with no `Info` at all, so one broadcast advertised its hang catalog /// at 100 and the same catalog in MSF at 0. + #[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] async fn non_media_tracks_rank_above_media() { use hang::catalog::PRIORITY; From 095355578e3d9a41f7b577be0e1ab625af5ae920 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 18:15:10 -0700 Subject: [PATCH 3/8] test(hang): declare shared snapshot fixture inputs explicitly Co-Authored-By: GPT-6 --- js/hang/src/stats.test.ts | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/js/hang/src/stats.test.ts b/js/hang/src/stats.test.ts index 5fd45a66462..6ded25efcce 100644 --- a/js/hang/src/stats.test.ts +++ b/js/hang/src/stats.test.ts @@ -4,16 +4,14 @@ import * as z from "@zod/mini"; import * as Echo from "./echo"; import * as Stats from "./stats"; -const fixtures = new URL("../../../rs/hang/fixtures/", import.meta.url); - for (const [name, schema] of [ - ["stats-snapshot.json", Stats.SnapshotSchema], - ["echo-snapshot.json", Echo.SnapshotSchema], - ["stats-zero.json", Stats.SnapshotSchema], - ["echo-zero.json", Echo.SnapshotSchema], + ["../../../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, fixtures)).json(); + const input = await Bun.file(new URL(name, import.meta.url)).json(); expect(JSON.parse(JSON.stringify(schema.parse(input)))).toEqual(input); }); } @@ -63,7 +61,9 @@ test("unknown track fields are ignored and flattened snapshot extensions survive }); test("shared invalid counters refuse instead of rounding", async () => { - const cases = (await Bun.file(new URL("snapshot-invalid.json", fixtures)).json()) as unknown[]; + 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.TrackSchema.parse(value)).toThrow(); for (const value of [ { stalled: -1 }, From 716edc756f1905a8bd32ec35899ce5b2e2ddd294 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 18:53:03 -0700 Subject: [PATCH 4/8] fix(mux): reject echo prefixes escaping the catalog root Co-Authored-By: GPT-6 --- rs/moq-mux/src/catalog/hang/consumer.rs | 53 ++++++++++++++++++++++--- 1 file changed, 47 insertions(+), 6 deletions(-) diff --git a/rs/moq-mux/src/catalog/hang/consumer.rs b/rs/moq-mux/src/catalog/hang/consumer.rs index 0c7465b79e5..55d392cc177 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"); + let mut echo = hang::catalog::Echo::default(); + echo.path = moq_net::path::Relative::new(reference).into_owned(); + catalog.echo = Some(echo); + 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(); From 83aff9b5de2eae4987e81db4047039a9372605f5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 23:23:35 -0700 Subject: [PATCH 5/8] fix(hang): require stats track and echo path as the draft specifies The draft types `stats.track` and `echo.path` as required, but both readers defaulted them, so `{"stats":{}}` named an empty track and `{"echo":{}}` silently solicited feedback at the parent prefix. Refuse both in Rust and JS. Also reattach a moq-mux test doc comment that the new MSF test had split from its test. Co-Authored-By: Claude Opus 5.5 --- js/hang/src/catalog/echo.ts | 3 +-- js/hang/src/catalog/root.test.ts | 2 ++ js/hang/src/catalog/stats.ts | 2 +- rs/hang/src/catalog/echo.rs | 1 - rs/hang/src/catalog/stats.rs | 1 - rs/hang/src/snapshot.rs | 2 ++ rs/moq-mux/src/catalog/producer.rs | 10 +++++----- 7 files changed, 11 insertions(+), 10 deletions(-) diff --git a/js/hang/src/catalog/echo.ts b/js/hang/src/catalog/echo.ts index 2d49603169a..ebe05c39032 100644 --- a/js/hang/src/catalog/echo.ts +++ b/js/hang/src/catalog/echo.ts @@ -1,11 +1,10 @@ -import { Path } from "@moq/net"; 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: z._default(RelativeBroadcastSchema, Path.normalizeRelative("")), + path: RelativeBroadcastSchema, }); /** The relative prefix soliciting viewer feedback. */ diff --git a/js/hang/src/catalog/root.test.ts b/js/hang/src/catalog/root.test.ts index d3e161a9539..a2c5170b600 100644 --- a/js/hang/src/catalog/root.test.ts +++ b/js/hang/src/catalog/root.test.ts @@ -264,4 +264,6 @@ test("shared stats and echo discovery round trips without changing old catalogs" ).json(); expect(RootSchema.parse(fixture)).toEqual(fixture); expect(RootSchema.parse({})).toEqual({}); + expect(() => RootSchema.parse({ stats: {} })).toThrow(); + expect(() => RootSchema.parse({ echo: {} })).toThrow(); }); diff --git a/js/hang/src/catalog/stats.ts b/js/hang/src/catalog/stats.ts index 8717c4ce2ea..65098d6d2c9 100644 --- a/js/hang/src/catalog/stats.ts +++ b/js/hang/src/catalog/stats.ts @@ -3,7 +3,7 @@ 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._default(z.string(), ""), + track: z.string(), }); /** The publisher's media snapshot track in this broadcast. */ diff --git a/rs/hang/src/catalog/echo.rs b/rs/hang/src/catalog/echo.rs index 3b83ce964c5..7f840c9fa5e 100644 --- a/rs/hang/src/catalog/echo.rs +++ b/rs/hang/src/catalog/echo.rs @@ -2,7 +2,6 @@ use serde::{Deserialize, Serialize}; /// The relative broadcast prefix under which viewers may publish feedback. #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] -#[serde(default)] #[non_exhaustive] pub struct Echo { /// The feedback prefix relative to the broadcast serving this catalog. diff --git a/rs/hang/src/catalog/stats.rs b/rs/hang/src/catalog/stats.rs index 29ad857bc44..12383d5e843 100644 --- a/rs/hang/src/catalog/stats.rs +++ b/rs/hang/src/catalog/stats.rs @@ -2,7 +2,6 @@ use serde::{Deserialize, Serialize}; /// The publisher's media snapshot track in this broadcast. #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] -#[serde(default)] #[non_exhaustive] pub struct Stats { /// The name of the publisher snapshot track. diff --git a/rs/hang/src/snapshot.rs b/rs/hang/src/snapshot.rs index cc183d0f4cd..7599109d236 100644 --- a/rs/hang/src/snapshot.rs +++ b/rs/hang/src/snapshot.rs @@ -112,6 +112,8 @@ mod tests { let catalog = Catalog::<()>::from_str("{}").unwrap(); assert!(catalog.stats.is_none() && catalog.echo.is_none()); assert_eq!(catalog.to_json().unwrap(), "{}"); + assert!(Catalog::<()>::from_str(r#"{"stats":{}}"#).is_err()); + assert!(Catalog::<()>::from_str(r#"{"echo":{}}"#).is_err()); } #[test] diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index fe604960e86..c56653c8a41 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -1308,11 +1308,6 @@ mod test { } } - /// The catalog, its MSF twin, and the timeline all rank above media. They're the - /// index a player reads before any media is useful, and they're small enough that - /// sitting above media can't starve it. Regression: the MSF and timeline tracks - /// were minted with no `Info` at all, so one broadcast advertised its hang catalog - /// at 100 and the same catalog in MSF at 0. #[test] fn msf_omits_media_report_discovery() { let catalog = @@ -1324,6 +1319,11 @@ mod test { assert!(!value["tracks"].as_array().unwrap().is_empty()); } + /// The catalog, its MSF twin, and the timeline all rank above media. They're the + /// index a player reads before any media is useful, and they're small enough that + /// sitting above media can't starve it. Regression: the MSF and timeline tracks + /// were minted with no `Info` at all, so one broadcast advertised its hang catalog + /// at 100 and the same catalog in MSF at 0. #[tokio::test] async fn non_media_tracks_rank_above_media() { use hang::catalog::PRIORITY; From 923b296653635365af10c5cb38c8176c98d50029 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 23:38:43 -0700 Subject: [PATCH 6/8] refactor(hang): construct catalog stats and echo with new, like M2ts Both sections now have a required field, so a `Default` built an unreadable value. Follow the `M2ts::new(track)` convention instead. Co-Authored-By: Claude Opus 5.5 --- rs/hang/src/catalog/echo.rs | 9 ++++++++- rs/hang/src/catalog/stats.rs | 9 ++++++++- rs/moq-mux/src/catalog/hang/consumer.rs | 6 +++--- 3 files changed, 19 insertions(+), 5 deletions(-) diff --git a/rs/hang/src/catalog/echo.rs b/rs/hang/src/catalog/echo.rs index 7f840c9fa5e..a56863c40a9 100644 --- a/rs/hang/src/catalog/echo.rs +++ b/rs/hang/src/catalog/echo.rs @@ -1,9 +1,16 @@ use serde::{Deserialize, Serialize}; /// The relative broadcast prefix under which viewers may publish feedback. -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[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/stats.rs b/rs/hang/src/catalog/stats.rs index 12383d5e843..c6cadb445c2 100644 --- a/rs/hang/src/catalog/stats.rs +++ b/rs/hang/src/catalog/stats.rs @@ -1,9 +1,16 @@ use serde::{Deserialize, Serialize}; /// The publisher's media snapshot track in this broadcast. -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[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/moq-mux/src/catalog/hang/consumer.rs b/rs/moq-mux/src/catalog/hang/consumer.rs index 55d392cc177..35f5a4ddf1e 100644 --- a/rs/moq-mux/src/catalog/hang/consumer.rs +++ b/rs/moq-mux/src/catalog/hang/consumer.rs @@ -263,9 +263,9 @@ mod test { fn soliciting(reference: &str) -> Catalog { let (mut catalog, _) = catalog_payload("here"); - let mut echo = hang::catalog::Echo::default(); - echo.path = moq_net::path::Relative::new(reference).into_owned(); - catalog.echo = Some(echo); + catalog.echo = Some(hang::catalog::Echo::new( + moq_net::path::Relative::new(reference).into_owned(), + )); catalog } From c538fd88695cabb01c97f3fef3183c548e09f638 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 12:16:09 -0700 Subject: [PATCH 7/8] feat(hang)!: name per-rendition reports Rendition and require arrival pairs Apply the maintainer's schema decisions: - Rename stats::Track and echo::Track to Rendition (RenditionSchema in JS, PublisherRendition/ViewerRendition in the draft), matching the renditions map and the catalog's rendition IDs. - Require both arrival.timestamp and arrival.wall; a half pair was silently filled with zero or the MoQ epoch. Arrival::new builds one. - Decode a stats value without a string track, or an echo value without a string path, as absent, like text/json/binary: an application may already use either generic key. Co-Authored-By: Claude Opus 5.5 --- drafts/draft-lcurley-moq-hang.md | 14 +++++++------ js/hang/src/catalog/root.test.ts | 15 ++++++++++++-- js/hang/src/catalog/root.ts | 18 +++++------------ js/hang/src/catalog/section.ts | 24 ++++++++++++++++++++++ js/hang/src/echo.ts | 14 ++++++------- js/hang/src/stats.test.ts | 17 +++++++++------- js/hang/src/stats.ts | 6 +++--- rs/hang/fixtures/echo-zero.json | 5 ++++- rs/hang/src/catalog/mod.rs | 29 +++++++++++++++++++++++++++ rs/hang/src/catalog/root.rs | 27 +++++++++++++++++++++++++ rs/hang/src/echo.rs | 22 +++++++-------------- rs/hang/src/snapshot.rs | 34 +++++++++++++++----------------- rs/hang/src/stats.rs | 4 ++-- 13 files changed, 155 insertions(+), 74 deletions(-) diff --git a/drafts/draft-lcurley-moq-hang.md b/drafts/draft-lcurley-moq-hang.md index 4888f821e3e..a73e3204dd6 100644 --- a/drafts/draft-lcurley-moq-hang.md +++ b/drafts/draft-lcurley-moq-hang.md @@ -619,6 +619,8 @@ A reader recognizes feedback by the `.echo` suffix on the broadcast's final path 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}}. @@ -628,22 +630,22 @@ A reader takes the newest complete snapshot and treats the reporting broadcast's ~~~ type PublisherSnapshot = { transport?: Transport, - renditions?: { [renditionId: string]: PublisherTrack }, + renditions?: { [renditionId: string]: PublisherRendition }, // ... any container or application sections ... } type ViewerSnapshot = { transport?: Transport, - renditions?: { [renditionId: string]: ViewerTrack }, + renditions?: { [renditionId: string]: ViewerRendition }, // ... any container or application sections ... } -type PublisherTrack = { +type PublisherRendition = { frames?: number, bytes?: number, keyframes?: number, skipped?: number, bitrate?: number, } -type ViewerTrack = { +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, + arrival?: { timestamp: number, wall: number }, latency?: number, } type Transport = { rtt?: number, rate?: number, lostBytes?: number, lostPackets?: number, age?: number, @@ -655,7 +657,7 @@ Unknown fields MUST be ignored; application and container sections MAY be flatte 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 absent `arrival.timestamp` means zero and an absent `arrival.wall` means the MoQ epoch. +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. diff --git a/js/hang/src/catalog/root.test.ts b/js/hang/src/catalog/root.test.ts index a2c5170b600..26d84712e98 100644 --- a/js/hang/src/catalog/root.test.ts +++ b/js/hang/src/catalog/root.test.ts @@ -264,6 +264,17 @@ test("shared stats and echo discovery round trips without changing old catalogs" ).json(); expect(RootSchema.parse(fixture)).toEqual(fixture); expect(RootSchema.parse({})).toEqual({}); - expect(() => RootSchema.parse({ stats: {} })).toThrow(); - expect(() => RootSchema.parse({ echo: {} })).toThrow(); +}); + +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 49d2796d191..15367259265 100644 --- a/js/hang/src/catalog/root.ts +++ b/js/hang/src/catalog/root.ts @@ -11,7 +11,7 @@ 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"; @@ -38,18 +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), - stats: z.optional( - z.pipe( - z.nullable(StatsSchema), - z.transform((value) => value ?? undefined), - ), - ), - echo: z.optional( - z.pipe( - z.nullable(EchoSchema), - z.transform((value) => value ?? undefined), - ), - ), + // 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 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/echo.ts b/js/hang/src/echo.ts index 8693f1a33e5..fb22f7f5cae 100644 --- a/js/hang/src/echo.ts +++ b/js/hang/src/echo.ts @@ -12,16 +12,16 @@ import { TransportSchema } from "./stats"; /** The newest received media timestamp paired with its arrival wall time. */ export const ArrivalSchema = z.object({ - // Media timestamp in milliseconds; omitted means zero. - timestamp: z.optional(u53Schema), - // Arrival milliseconds since the MoQ epoch; omitted means that epoch. - wall: z.optional(u53Schema), + // 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 TrackSchema = z.object({ +export const RenditionSchema = z.object({ // Cumulative media frames received. frames: z.optional(u53Schema), // Cumulative media payload bytes received. @@ -49,13 +49,13 @@ export const TrackSchema = z.object({ latency: measurementSchema, }); /** One rendition's delivery counters and latest playout measurements. */ -export type Track = z.infer; +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(), TrackSchema)), + renditions: z.optional(z.record(z.string(), RenditionSchema)), }); /** One viewer report, with application or container sections flattened beside it. */ export type Snapshot = z.infer; diff --git a/js/hang/src/stats.test.ts b/js/hang/src/stats.test.ts index 6ded25efcce..221fff8b9bb 100644 --- a/js/hang/src/stats.test.ts +++ b/js/hang/src/stats.test.ts @@ -47,12 +47,12 @@ test("optional null measurements mean unavailable rather than measured zero", () ), ), ).toEqual({ transport: {}, renditions: { hd: {} } }); - expect(JSON.parse(JSON.stringify(Echo.TrackSchema.parse({ latency: null, arrival: null })))).toEqual({}); + 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.TrackSchema.parse({ future: 42 })).toEqual({}); - expect(Echo.TrackSchema.parse({ future: 42 })).toEqual({}); + 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); @@ -64,16 +64,19 @@ 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.TrackSchema.parse(value)).toThrow(); + 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 } }, - { arrival: { wall: -1 } }, + { 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.TrackSchema.parse(value)).toThrow(); + 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(); }); diff --git a/js/hang/src/stats.ts b/js/hang/src/stats.ts index 2b5174a0be1..fefc20add26 100644 --- a/js/hang/src/stats.ts +++ b/js/hang/src/stats.ts @@ -22,7 +22,7 @@ export const TransportSchema = z.object({ /** 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 TrackSchema = z.object({ +export const RenditionSchema = z.object({ // Cumulative media frames sent. frames: z.optional(u53Schema), // Cumulative media payload bytes sent. @@ -35,13 +35,13 @@ export const TrackSchema = z.object({ bitrate: measurementSchema, }); /** One locally published rendition's counters and optional target bitrate. */ -export type Track = z.infer; +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(), TrackSchema)), + 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/rs/hang/fixtures/echo-zero.json b/rs/hang/fixtures/echo-zero.json index b176e009631..4d27c8e7bb2 100644 --- a/rs/hang/fixtures/echo-zero.json +++ b/rs/hang/fixtures/echo-zero.json @@ -9,7 +9,10 @@ "renditions": { "reset": {}, "hd": { - "arrival": {}, + "arrival": { + "timestamp": 0, + "wall": 0 + }, "latency": 0 } } diff --git a/rs/hang/src/catalog/mod.rs b/rs/hang/src/catalog/mod.rs index b7e4eb1c868..c6106892571 100644 --- a/rs/hang/src/catalog/mod.rs +++ b/rs/hang/src/catalog/mod.rs @@ -99,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 5b77107dafa..0805de65656 100644 --- a/rs/hang/src/catalog/root.rs +++ b/rs/hang/src/catalog/root.rs @@ -109,9 +109,16 @@ pub struct Catalog { 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. @@ -758,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/echo.rs b/rs/hang/src/echo.rs index 93e7388217a..6c7b13cc613 100644 --- a/rs/hang/src/echo.rs +++ b/rs/hang/src/echo.rs @@ -14,7 +14,7 @@ pub struct Snapshot { pub transport: Transport, /// Per-rendition counters, including entries with all counters reset to zero. #[serde(skip_serializing_if = "BTreeMap::is_empty")] - pub renditions: BTreeMap, + pub renditions: BTreeMap, /// Application-defined root sections flattened beside the media report. #[serde(flatten)] pub ext: E, @@ -25,7 +25,7 @@ pub struct Snapshot { #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] #[serde(default, rename_all = "camelCase")] #[non_exhaustive] -pub struct Track { +pub struct Rendition { /// Cumulative media frames received. #[serde_as(as = "Number")] #[serde(skip_serializing_if = "is_zero")] @@ -70,24 +70,20 @@ pub struct Track { /// The newest received media timestamp paired with its arrival wall time. #[serde_with::serde_as] #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] -#[serde(default, rename_all = "camelCase")] #[non_exhaustive] pub struct Arrival { /// The media timestamp, encoded in milliseconds. #[serde_as(as = "Millis")] - #[serde(skip_serializing_if = "timestamp_zero")] pub timestamp: moq_net::Timestamp, /// The arrival wall time, encoded in milliseconds since the MoQ epoch. #[serde_as(as = "crate::snapshot::Wall")] - #[serde(skip_serializing_if = "crate::snapshot::wall_zero")] pub wall: std::time::SystemTime, } -impl Default for Arrival { - fn default() -> Self { - Self { - timestamp: moq_net::Timestamp::ZERO, - wall: crate::snapshot::wall_default(), - } + +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 } } } @@ -99,7 +95,3 @@ pub const TRACK: &str = "feedback.json"; pub fn is_echo(path: &moq_net::Path) -> bool { path.as_str().ends_with(SUFFIX) } - -fn timestamp_zero(timestamp: &moq_net::Timestamp) -> bool { - timestamp.is_zero() -} diff --git a/rs/hang/src/snapshot.rs b/rs/hang/src/snapshot.rs index 7599109d236..c8b1f99cf93 100644 --- a/rs/hang/src/snapshot.rs +++ b/rs/hang/src/snapshot.rs @@ -53,12 +53,6 @@ pub(crate) struct Wall; fn epoch() -> std::time::SystemTime { std::time::UNIX_EPOCH + Duration::from_millis(crate::catalog::MOQ_EPOCH_UNIX_MILLIS) } -pub(crate) fn wall_default() -> std::time::SystemTime { - epoch() -} -pub(crate) fn wall_zero(value: &std::time::SystemTime) -> bool { - *value == epoch() -} 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)?; @@ -106,14 +100,12 @@ mod tests { serde_json::to_value(echo::Snapshot::<()>::default()).unwrap(), serde_json::json!({}) ); - let track: stats::Track = serde_json::from_str(r#"{"future":17}"#).unwrap(); - assert_eq!(track, stats::Track::default()); - assert_eq!(serde_json::to_value(track).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(), "{}"); - assert!(Catalog::<()>::from_str(r#"{"stats":{}}"#).is_err()); - assert!(Catalog::<()>::from_str(r#"{"echo":{}}"#).is_err()); } #[test] @@ -159,21 +151,27 @@ mod tests { 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}"); + assert!( + serde_json::from_value::(case.clone()).is_err(), + "{case}" + ); } for input in [ r#"{"stalled":-1}"#, r#"{"latency":0.5}"#, - r#"{"arrival":{"timestamp":9007199254740992}}"#, - r#"{"arrival":{"wall":-1}}"#, + 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}"); + assert!(serde_json::from_str::(input).is_err(), "{input}"); } - let track = stats::Track { + let rendition = stats::Rendition { frames: crate::catalog::MAX_SAFE_INTEGER + 1, ..Default::default() }; - assert!(serde_json::to_value(track).is_err()); + assert!(serde_json::to_value(rendition).is_err()); } #[test] @@ -190,7 +188,7 @@ mod tests { 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(wall_default()).unwrap(), + arrival.wall.duration_since(epoch()).unwrap(), Duration::from_millis(175_184_640_000) ); assert_eq!(report.renditions["hd"].stalled, Duration::from_millis(120)); diff --git a/rs/hang/src/stats.rs b/rs/hang/src/stats.rs index bec49525c4d..fceeff10402 100644 --- a/rs/hang/src/stats.rs +++ b/rs/hang/src/stats.rs @@ -13,7 +13,7 @@ pub struct Snapshot { pub transport: Transport, /// Per-rendition counters, including entries with all counters reset to zero. #[serde(skip_serializing_if = "BTreeMap::is_empty")] - pub renditions: BTreeMap, + pub renditions: BTreeMap, /// Application-defined root sections flattened beside the media report. #[serde(flatten)] pub ext: E, @@ -24,7 +24,7 @@ pub struct Snapshot { #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] #[serde(default, rename_all = "camelCase")] #[non_exhaustive] -pub struct Track { +pub struct Rendition { /// Cumulative frames sent. #[serde_as(as = "Number")] #[serde(skip_serializing_if = "is_zero")] From a7d93a25560890985ec11785c42cabc5b3484a12 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 12:27:32 -0700 Subject: [PATCH 8/8] fix(mux): reserve stats and echo in by-name catalog extensions Extra::set accepted both names, so a caller could emit a duplicate root key beside the built-in field. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/catalog/hang/ext.rs | 22 ++++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) 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.