diff --git a/doc/concept/hang.md b/doc/concept/hang.md index 22699656f8..c95fa3f2e4 100644 --- a/doc/concept/hang.md +++ b/doc/concept/hang.md @@ -104,7 +104,8 @@ document would silently discard everything but the last payload: The rest is descriptive: `compression` (`deflate`, the same group-scoped `deflate-raw` the catalog uses), `schema` on a JSON track, `mime` on a binary -one, plus the optional `broadcast` reference. A +one, `bitrate` and `jitter` with the same meaning as for media, plus the +optional `broadcast` reference. A consumer that doesn't recognize a `mode` or `compression` ignores that track and round-trips it verbatim. @@ -116,6 +117,12 @@ or `catalog.binary.tracks`, then pair its name and config with `moq_publish_binary_*` do the same, retracting on `_finish`. In the browser, read the same map, subscribe by name, and hand the track to `@moq/json` or `@moq/binary`. +An application with its own per-track fields can list a data track in its own +root section instead, flattening the JSON or binary entry beside those fields +so there is one entry per track. Name the section with a namespaced key such as +`com.example.mavlink`. A generic consumer only finds tracks in `json` and +`binary`. + ## Container The `container.kind` on each rendition says how frames are framed: diff --git a/doc/lib/rs/moq-mux.md b/doc/lib/rs/moq-mux.md index 613f9e2b52..a3fd8d42e8 100644 --- a/doc/lib/rs/moq-mux.md +++ b/doc/lib/rs/moq-mux.md @@ -47,6 +47,39 @@ before the edit is retained, including while the initial catalog is reserved. Codec importers propagate catalog and media errors through their configuration and frame-writing methods. +Data tracks go through the catalog too. `catalog.json_stream(track, config)` +(or `json_snapshot`, `binary_snapshot`, `binary_stream`) writes the track's +`json` or `binary` entry, measures an absent `bitrate` from the writes, and +retires the entry when the producer drops. To list the track in your own +section beside application fields, pass that section's entry instead of a +`json::Config` or `binary::Config`: any `RenditionConfig` that embeds the data +config through `AsMut`. + +```rust +#[derive(Serialize, Deserialize, Clone)] +struct Mavlink { + #[serde(flatten)] + binary: hang::catalog::BinaryConfig, // mode, compression, bitrate, ... + sysid: u8, +} + +impl AsMut for Mavlink { + fn as_mut(&mut self) -> &mut hang::catalog::BinaryConfig { + &mut self.binary + } +} + +// Plus `RenditionConfig` writing to `catalog.ext.mavlink`, a map +// serialized under the `com.example.mavlink` root key. +let binary = hang::catalog::BinaryConfig::new(hang::catalog::Mode::Stream); +let mut telemetry = catalog.binary_stream(track, Mavlink { binary, sysid: 1 })?; +telemetry.append(packet)?; +``` + +The producer sets the entry's `mode` and encodes the track with its +`compression`. Read it back from `Catalog` and subscribe with +`catalog::Entry::new(name, &entry.binary)`. + ```bash cargo add moq-mux ``` diff --git a/drafts/draft-lcurley-moq-hang.md b/drafts/draft-lcurley-moq-hang.md index ed250cf3fa..30d711e3c9 100644 --- a/drafts/draft-lcurley-moq-hang.md +++ b/drafts/draft-lcurley-moq-hang.md @@ -134,6 +134,7 @@ type Catalog = { ~~~ Additional fields MAY be added based on the application. +An application SHOULD name its own root sections with a namespaced key, such as a reverse-DNS name (`com.example.telemetry`), so they cannot collide with a section a later version of this specification defines. The catalog SHOULD be mostly static, delegating any dynamic content to other tracks. For example, a chat entry should name a chat track, not carry individual chat messages. @@ -388,6 +389,8 @@ type JsonSchema = { "compression": Compression | undefined, "schema": string | undefined, "broadcast": string | undefined, + "bitrate": number | undefined, + "jitter": number | undefined, } ~~~ @@ -401,6 +404,8 @@ type BinarySchema = { "compression": Compression | undefined, "mime": string | undefined, "broadcast": string | undefined, + "bitrate": number | undefined, + "jitter": number | undefined, } ~~~ @@ -456,6 +461,10 @@ A `snapshot` group covers a single value (plus any deltas), so its window spans ### broadcast {#data-shared} The `broadcast` field carries the same meaning here as it does for a media rendition ({{field-broadcast}}). +### bitrate and jitter {#data-estimates} +The optional `bitrate` field is the track's maximum bitrate in bits per second. +The optional `jitter` field carries the same meaning and rules as it does for a media rendition ({{field-jitter}}), with a payload in place of a frame. + ## Binary Fields {#binary} A decoder config field carrying raw bytes, notably `description` (an `AllowSharedBufferSource` in WebCodecs), is carried in the catalog as a hex string ({{!RFC4648, Section 8}}). A publisher SHOULD emit lowercase hexadecimal characters and MUST NOT emit a `0x` prefix or any separators. @@ -1075,6 +1084,8 @@ A publisher MAY estimate an unknown final duration from the frame cadence, but M - A publisher that stops producing and may resume on the same track SHOULD publish a discontinuity marker when it stops. - An audio endpoint bounds only the terminal packets that follow it in its own group. - Replaced the archive timeline `wall` field with a root `clock` section (`wall` plus `timescale`): one fixed broadcast mapping every track and the archive index convert into, independent of any archive. Zero timescales and walls past the JSON-safe integer range are refused. +- Added optional `bitrate` and `jitter` fields to `json` and `binary` track entries. +- Recommended namespaced keys for application root sections. # Acknowledgments {:numbered="false"} diff --git a/js/hang/src/catalog/binary.ts b/js/hang/src/catalog/binary.ts index 2c209e1413..593eaf24d5 100644 --- a/js/hang/src/catalog/binary.ts +++ b/js/hang/src/catalog/binary.ts @@ -1,5 +1,6 @@ import * as z from "zod/mini"; import { CompressionSchema } from "./compression"; +import { u53Schema } from "./integers"; import { ModeSchema } from "./mode"; import { RelativeBroadcastSchema } from "./path"; @@ -28,6 +29,18 @@ export const BinaryConfigSchema = z.looseObject({ // An optional media type for each payload (e.g. "image/jpeg"). Purely descriptive: // a consumer that doesn't recognize it can still read the track. mime: z.optional(z.string()), + + // The maximum bitrate of the track in bits per second, if known. + bitrate: z.optional(u53Schema), + + // The maximum delay between a payload being ready and the publisher flushing it, in whole + // milliseconds rounded up, with the same meaning as a video rendition's `jitter`. + jitter: z.optional( + z.pipe( + u53Schema, + z.transform((value) => (value === 0 ? undefined : value)), + ), + ), }); /** diff --git a/js/hang/src/catalog/json.ts b/js/hang/src/catalog/json.ts index 6ee22e78a0..c5900559d7 100644 --- a/js/hang/src/catalog/json.ts +++ b/js/hang/src/catalog/json.ts @@ -1,5 +1,6 @@ import * as z from "zod/mini"; import { CompressionSchema } from "./compression"; +import { u53Schema } from "./integers"; import { ModeSchema } from "./mode"; import { RelativeBroadcastSchema } from "./path"; @@ -27,6 +28,18 @@ export const JsonConfigSchema = z.looseObject({ // An optional identifier for the shape of each value, typically a JSON Schema URL. // Purely descriptive: a consumer that doesn't recognize it can still read the track. schema: z.optional(z.string()), + + // The maximum bitrate of the track in bits per second, if known. + bitrate: z.optional(u53Schema), + + // The maximum delay between a payload being ready and the publisher flushing it, in whole + // milliseconds rounded up, with the same meaning as a video rendition's `jitter`. + jitter: z.optional( + z.pipe( + u53Schema, + z.transform((value) => (value === 0 ? undefined : value)), + ), + ), }); /** diff --git a/js/publish/src/catalog.test.ts b/js/publish/src/catalog.test.ts index 58873544ec..cd76a88781 100644 --- a/js/publish/src/catalog.test.ts +++ b/js/publish/src/catalog.test.ts @@ -195,3 +195,45 @@ for (const section of ["audio", "video", "text"] as const) { }); }); } + +for (const section of ["json", "binary"] as const) { + test(`catalog refuses zero or decreasing ${section} jitter without retaining it`, () => { + const catalog = new CatalogProducer(); + const tracks = (value: Catalog.Root) => { + const sectionValue = value[section]; + if (!sectionValue) throw new Error(`expected a retained ${section} section`); + return sectionValue.tracks; + }; + + expect(() => + catalog.mutate((value) => { + value[section] = { tracks: { data: { mode: "stream", jitter: Catalog.u53(0) } } }; + }), + ).toThrow("omit jitter"); + catalog.mutate((value) => { + expect(value[section]).toBeUndefined(); + }); + + catalog.mutate((value) => { + value[section] = { tracks: { data: { mode: "stream", jitter: Catalog.u53(100) } } }; + }); + for (const jitter of [Catalog.u53(50), undefined]) { + expect(() => + catalog.mutate((value) => { + tracks(value).data.jitter = jitter; + }), + ).toThrow("jitter cannot decrease"); + catalog.mutate((value) => { + expect(tracks(value).data.jitter).toBe(Catalog.u53(100)); + }); + } + + // A new track under the same name, after the old one is gone, starts over. + catalog.mutate((value) => { + delete tracks(value).data; + }); + catalog.mutate((value) => { + tracks(value).data = { mode: "stream", jitter: Catalog.u53(50) }; + }); + }); +} diff --git a/js/publish/src/catalog.ts b/js/publish/src/catalog.ts index 7c45ec7962..3a40e3d1ae 100644 --- a/js/publish/src/catalog.ts +++ b/js/publish/src/catalog.ts @@ -24,12 +24,13 @@ export class CatalogProducer { mutate(fn: (catalog: Catalog.Root) => void): void { const value = structuredClone(this.#value); fn(value); - for (const section of ["audio", "video", "text"] as const) { - for (const [name, config] of Object.entries(value[section]?.renditions ?? {})) { - if (config.jitter === 0) throw new Error("omit jitter for a track flushed immediately"); - const previous = this.#value[section]?.renditions[name]?.jitter; - if (previous !== undefined && (config.jitter === undefined || config.jitter < previous)) { - throw new Error("jitter cannot decrease for an existing rendition"); + for (const [section, next] of Object.entries(jitters(value))) { + const previous = jitters(this.#value)[section]; + for (const [name, jitter] of Object.entries(next)) { + if (jitter === 0) throw new Error("omit jitter for a track flushed immediately"); + const before = previous?.[name]; + if (before !== undefined && (jitter === undefined || jitter < before)) { + throw new Error("jitter cannot decrease for an existing track"); } } } @@ -59,6 +60,19 @@ export class CatalogProducer { } } +/** Every track's advertised jitter, by section and then track name. */ +function jitters(catalog: Catalog.Root): Record> { + const pick = (tracks: Record | undefined) => + Object.fromEntries(Object.entries(tracks ?? {}).map(([name, config]) => [name, config.jitter])); + return { + audio: pick(catalog.audio?.renditions), + video: pick(catalog.video?.renditions), + text: pick(catalog.text?.renditions), + json: pick(catalog.json?.tracks), + binary: pick(catalog.binary?.tracks), + }; +} + // The wall time of `performance.now() === 0`, the zero every js/publish timestamp counts from. function pageClock(): Catalog.Clock { const wall = Math.round((performance.timeOrigin - Catalog.MOQ_EPOCH_UNIX_MILLIS) * 1000); diff --git a/quest/m1/README.md b/quest/m1/README.md index 3aff7303c9..f2c3be7a91 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -30,7 +30,6 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Binding surface](/quest/m1/binding-surface.md) - moq-ffi, libmoq, and every wrapper expose the decode delay, route source, and connection timing - [FFI shape](/quest/m1/ffi-shape/README.md) - the bindings mirror Rust's layers: net at the root, then media, json, audio, and video namespaces built from the handle below - [Track demand](/quest/m1/track-demand.md) - Rust and JS watch a track's subscribers through `demand()` alone -- [Data sections](/quest/m1/data-sections.md) - an application lists JSON and binary tracks in its own catalog section with its own per-track fields, published in one moq-mux call; data entries gain `bitrate` and `jitter` - [Broadcast close](/quest/m1/broadcast-close/README.md) - `close()` is the one way to end a broadcast in every language, a permanent retraction that leaves in-flight tracks alone - [Relay peer set](/quest/m1/relay-peer-set.md) - a wire consumer tells a client hop from a peer hop, and every mesh credential can mark a peer - [CLI import clock](/quest/m1/cli-import-clock.md) - fMP4, TS, and FLV imports publish on the shared broadcast clock across restarts diff --git a/quest/m1/data-jitter.md b/quest/m1/data-jitter.md index 76cee93321..e21e11ca3b 100644 --- a/quest/m1/data-jitter.md +++ b/quest/m1/data-jitter.md @@ -27,14 +27,17 @@ meaningless zero. broadcast-wide baseline, distort every other track; it stays in the payload. Reject a timestamp ahead of the clock's `now` rather than clamp it. - Add optional `delay` to `JsonConfig` and `BinaryConfig` in `rs/hang`, - `js/hang`, and the draft, beside the `jitter` - [data sections](/quest/m1/data-sections.md) adds. + `js/hang`, and the draft, beside `jitter`. - Feed the catalog's flush clock from the `moq-mux` data producers when a capture time is present and a frame was actually emitted. A `moq-json` snapshot `update` with an unchanged value succeeds without writing one, so the lower producer reports whether it emitted, and a repeated value must not move the baseline (cover it in tests). Publish the result through the embedded config's - `Estimate`, as [data sections](/quest/m1/data-sections.md) does for bitrate. + `Estimate`, as the data producers already do for bitrate. +- Have the lower producers also report each emitted frame's encoded size, and + measure bitrate from that instead of the pre-compression payload or + serialized value: today an unchanged snapshot `update` still counts, and + DEFLATE can slightly expand an incompressible payload. - Mirror the capture timestamp in the published `js/binary` and `js/json` producers, so browser publishers can produce the same timed tracks. @@ -45,4 +48,3 @@ entries. ## Required - [Jitter clock](/quest/m1/jitter-flush-clock.md) - defines the flush-lateness measurement and `delay` -- [Data sections](/quest/m1/data-sections.md) - adds the `jitter` field and the producers this feeds diff --git a/quest/m1/data-sections.md b/quest/m1/data-sections.md deleted file mode 100644 index df8802602b..0000000000 --- a/quest/m1/data-sections.md +++ /dev/null @@ -1,66 +0,0 @@ -# [M] moq-mux: publish data tracks into an application's own catalog section - -## Goal - -An application lists a JSON or binary track in its own root section, with -its own per-track fields next to the reading rules, and publishes it with one -`moq-mux` call that handles framing, compression, the entry's lifecycle, and -a detected `bitrate`. JSON and binary entries gain optional `bitrate` and -`jitter`, matching video and audio. - -The motivating case is MAVLink telemetry: the application's -`catalog["com.example.mavlink"].tracks[name]` flattens a `BinaryConfig` -(mode, compression, broadcast) beside `sysid`, `compids`, and `dialect`. One -entry per track, so nothing has to be kept in sync with a second listing. - -Non-goals: hang defines no MAVLink (or other application) section, and -`BinaryConfig`/`JsonConfig` gain no generic or opaque extension field. A -single `tracks` map holds every data track, so one type parameter would force -every application track kind into one enum. Generic data-track tooling does -not list tracks in an application section; that is the trade for one source of -truth. - -## Plan - -- **Producers.** `Producer::binary_snapshot`/`binary_stream` and - `json_snapshot`/`json_stream` accept any entry that implements - `RenditionConfig` and embeds a `BinaryConfig`/`JsonConfig`, exposed - through a small trait (a `&mut` accessor to the embedded config). The - producer fixes `mode`, applies `compression`, and writes the entry into - whichever section the `RenditionConfig` names; dropping the handle removes - it. The existing `binary::Config`/`json::Config` builders keep working - unchanged, so this generalizes a concrete parameter (RFC 1105 minor) and - lands on `main`. Name the trait by role; `catalog::Entry` is already the - consumer-side pairing of a name and config. -- **Consumers** need nothing new: `Entry::new(name, &mavlink.binary)` already - subscribes through `binary::Consumer`. -- **Bitrate/jitter.** Add optional `bitrate` (bits per second) and `jitter` - (`MillisCeil`, same meaning as video and audio) to `BinaryConfig` and - `JsonConfig` in `rs/hang`, `js/hang`, and the Binary and JSON sections of - `drafts/draft-lcurley-moq-hang.md`. Opt the data producers into bitrate - detection through the embedded config's `Estimate`. Extend `js/publish`'s - `CatalogProducer.mutate` check (nonzero, never lowered; - `js/publish/src/catalog.ts`), today audio and video only, to the `json` and - `binary` sections, with tests. `jitter` is set by the - publisher only here; detection is [data jitter](/quest/m1/data-jitter.md), - because flush lateness needs a source timestamp the producers do not take - yet. -- **Draft.** One line: application root sections SHOULD use a namespaced key - (e.g. reverse-DNS), since `text`, `json`, and `binary` already needed - lenient decoding for keys applications used first. -- **Docs.** A custom-section example in `doc/lib/rs/moq-mux.md` beside the - existing data-track one, and the `RenditionConfig` doc example switched to a - section that embeds a `BinaryConfig`. No new page. -- **Tests.** A custom section round-trips through a producer and a - `Catalog` consumer, drop removes the entry, a duplicate name is refused, - and bitrate is detected. The existing `binary::Config` path stays covered. - -Public API: additive in `moq-mux` (generalized data producer parameter, one -new trait) and `hang`/`@moq/hang` (two optional fields). Wire: two optional -catalog fields, additive. - -## Related - -- [Data jitter](/quest/m1/data-jitter.md) - detects the `jitter` this quest adds -- [MAVLink bridge](/quest/m2/teleop/mavlink.md) - the in-repo consumer of the same shape -- [Robot teleoperation primitive](/quest/m2/teleop/robot.md) - its telemetry section embeds data-track configs this way diff --git a/quest/m2/teleop/robot.md b/quest/m2/teleop/robot.md index 6cd731e4c5..42614ceb11 100644 --- a/quest/m2/teleop/robot.md +++ b/quest/m2/teleop/robot.md @@ -64,7 +64,7 @@ The framing is where the guarantee lives, not the subscription flags: - The catalog section, through `moq-mux`'s `CatalogExt` and `RenditionConfig`: a namespaced root section whose entries embed a `JsonConfig` or `BinaryConfig` beside the robot's own fields, published - through the data producers ([data sections](/quest/m1/data-sections.md)). + through the `moq-mux` data producers. No hang schema change. - Announce-prefix fan-in, generalised from `rs/moq-boy/src/input.rs`. - The two delivery classes, as `moq-json`'s snapshot and stream modes with @@ -81,10 +81,6 @@ Port `moq-boy` onto the crate in the same change, as the no-arbitration case. It is the only existing consumer, and if the abstraction cannot express crowd control then it is the wrong abstraction. -## Required - -- [Data sections](/quest/m1/data-sections.md) - publishes the telemetry section's data tracks - ## Related - [arbitration](/quest/m2/teleop/arbitration.md) - which controller is obeyed diff --git a/rs/hang/src/catalog/binary.rs b/rs/hang/src/catalog/binary.rs index 7a89406785..7c4ec811eb 100644 --- a/rs/hang/src/catalog/binary.rs +++ b/rs/hang/src/catalog/binary.rs @@ -2,6 +2,7 @@ use std::collections::{BTreeMap, btree_map}; use serde::{Deserialize, Serialize}; +use crate::catalog::millis::MillisCeil; use crate::catalog::{Compression, Mode}; /// The binary tracks a broadcast publishes, keyed by track name. @@ -86,6 +87,16 @@ pub struct BinaryConfig { #[serde(default)] pub mime: Option, + /// The maximum bitrate of the track in bits per second, if known. + #[serde(default)] + pub bitrate: Option, + + /// The maximum delay between a payload being ready and the publisher flushing it, with the same + /// meaning and whole-millisecond encoding as [`VideoConfig::jitter`](crate::catalog::VideoConfig::jitter). + #[serde_as(as = "MillisCeil")] + #[serde(default)] + pub jitter: Option, + /// Fields this build doesn't recognize, kept so the entry round-trips. /// /// A future [`Mode`] or [`Compression`] almost certainly comes with fields describing it, and @@ -104,7 +115,16 @@ impl BinaryConfig { mode, compression: None, mime: None, + bitrate: None, + jitter: None, extra: Default::default(), } } } + +/// The config itself, so a data producer takes it wherever it takes an entry embedding one. +impl AsMut for BinaryConfig { + fn as_mut(&mut self) -> &mut Self { + self + } +} diff --git a/rs/hang/src/catalog/json.rs b/rs/hang/src/catalog/json.rs index cdc85a0397..9fa0d355c2 100644 --- a/rs/hang/src/catalog/json.rs +++ b/rs/hang/src/catalog/json.rs @@ -2,6 +2,7 @@ use std::collections::{BTreeMap, btree_map}; use serde::{Deserialize, Serialize}; +use crate::catalog::millis::MillisCeil; use crate::catalog::{Compression, Mode}; /// The JSON tracks a broadcast publishes, keyed by track name. @@ -84,6 +85,16 @@ pub struct JsonConfig { #[serde(default)] pub schema: Option, + /// The maximum bitrate of the track in bits per second, if known. + #[serde(default)] + pub bitrate: Option, + + /// The maximum delay between a payload being ready and the publisher flushing it, with the same + /// meaning and whole-millisecond encoding as [`VideoConfig::jitter`](crate::catalog::VideoConfig::jitter). + #[serde_as(as = "MillisCeil")] + #[serde(default)] + pub jitter: Option, + /// Fields this build doesn't recognize, kept so the entry round-trips. /// /// A future [`Mode`] or [`Compression`] almost certainly comes with fields describing it, and @@ -102,7 +113,16 @@ impl JsonConfig { mode, compression: None, schema: None, + bitrate: None, + jitter: None, extra: Default::default(), } } } + +/// The config itself, so a data producer takes it wherever it takes an entry embedding one. +impl AsMut for JsonConfig { + fn as_mut(&mut self) -> &mut Self { + self + } +} diff --git a/rs/hang/src/catalog/root.rs b/rs/hang/src/catalog/root.rs index baad3caae8..21c1e11e85 100644 --- a/rs/hang/src/catalog/root.rs +++ b/rs/hang/src/catalog/root.rs @@ -780,6 +780,68 @@ mod test { assert_eq!(output, encoded, "encode mismatch"); } + /// Data tracks carry the same optional `bitrate` and whole-millisecond `jitter` as media. + #[test] + fn data_track_bitrate_and_jitter() { + let encoded = r#"{"video":{"renditions":{}},"audio":{"renditions":{}},"json":{"tracks":{"gps":{"mode":"stream","bitrate":8000,"jitter":100}}},"binary":{"tracks":{"frames":{"mode":"snapshot","bitrate":64000,"jitter":34}}}}"#; + + let mut gps = JsonConfig::new(Mode::Stream); + gps.bitrate = Some(8_000); + gps.jitter = Some(std::time::Duration::from_millis(100)); + + let mut frames = BinaryConfig::new(Mode::Snapshot); + frames.bitrate = Some(64_000); + frames.jitter = Some(std::time::Duration::from_micros(33_334)); + + let mut catalog = Catalog::<()>::default(); + catalog.json.insert("gps", gps).unwrap(); + catalog.binary.insert("frames", frames).unwrap(); + + assert_eq!( + catalog.to_json().unwrap(), + encoded, + "jitter rounds up to whole milliseconds" + ); + + let decoded = Catalog::<()>::from_str(encoded).unwrap(); + assert_eq!( + decoded.binary.tracks["frames"].jitter, + Some(std::time::Duration::from_millis(34)) + ); + assert_eq!(decoded.json.tracks["gps"].bitrate, Some(8_000)); + } + + /// An application lists a data track in its own section by flattening a data config beside its + /// own fields, so the reading rules and the application's fields share one entry. + #[test] + fn a_data_config_flattens_into_an_application_entry() { + #[derive(Serialize, Deserialize, PartialEq, Debug, Clone)] + struct Mavlink { + #[serde(flatten)] + binary: BinaryConfig, + sysid: u8, + } + + #[derive(Serialize, Deserialize, PartialEq, Debug, Default, Clone)] + struct Ext { + #[serde(rename = "com.example.mavlink", default)] + mavlink: BTreeMap, + } + + let encoded = r#"{"video":{"renditions":{}},"audio":{"renditions":{}},"com.example.mavlink":{"telemetry":{"mode":"stream","compression":"deflate","sysid":1}}}"#; + + let catalog = Catalog::::from_str(encoded).unwrap(); + let entry = &catalog.ext.mavlink["telemetry"]; + assert_eq!(entry.sysid, 1); + assert_eq!(entry.binary.mode, Mode::Stream); + assert_eq!(entry.binary.compression, Some(Compression::Deflate)); + assert!( + entry.binary.extra.is_empty(), + "the application's own fields are not unknown data-track fields" + ); + assert_eq!(catalog.to_json().unwrap(), encoded); + } + /// A track using a future mode or compression must survive a reparse-and-republish intact, so a /// relay doesn't corrupt what it can't read. Its siblings stay readable. #[test] diff --git a/rs/moq-mux/src/binary.rs b/rs/moq-mux/src/binary.rs index 4f5ab5f4dc..d0ad13a996 100644 --- a/rs/moq-mux/src/binary.rs +++ b/rs/moq-mux/src/binary.rs @@ -47,17 +47,20 @@ //! # } //! ``` +use std::marker::PhantomData; + use bytes::Bytes; use hang::catalog::{BinaryConfig, Compression, Mode}; -use crate::catalog::Rendition; use crate::catalog::hang::CatalogExt; +use crate::catalog::{IntoRendition, Listing, RenditionConfig}; /// Everything a binary track declares about itself, beyond its mode and name. /// /// Start from [`default`](Default::default) and chain the setters. The mode is not in here: it is -/// fixed by which producer you create. +/// fixed by which producer you create. To list the track in an application's own catalog section +/// instead of `binary`, pass that section's entry (see [`IntoRendition`]). #[derive(Debug, Clone, Default)] #[non_exhaustive] pub struct Config { @@ -83,16 +86,33 @@ impl Config { self.mime = Some(mime.into()); self } +} - /// The catalog entry describing a track published under this config in `mode`. - pub(crate) fn entry(&self, mode: Mode) -> BinaryConfig { - let mut entry = BinaryConfig::new(mode); +impl IntoRendition for Config { + type Config = BinaryConfig; + + fn into_rendition(self) -> BinaryConfig { + // The producer overwrites the mode with the one it publishes in. + let mut entry = BinaryConfig::new(Mode::Snapshot); entry.compression = self.compression.then_some(Compression::Deflate); - entry.mime = self.mime.clone(); + entry.mime = self.mime; entry } } +/// Fix `config`'s mode and return whether its frames are compressed. +/// +/// Errors on a compression this build can't write, rather than advertising one the frames don't use, +/// and on a `broadcast` reference, which would point consumers away from the track this publishes. +fn prepare(config: &mut impl AsMut, mode: Mode) -> crate::Result { + let binary = config.as_mut(); + if binary.broadcast.is_some() { + return Err(crate::Error::ForeignBroadcast); + } + binary.mode = mode; + crate::compression(binary.compression.as_ref()) +} + /// Publishes a latest-value binary track, advertised in the catalog for as long as this handle /// lives. /// @@ -100,27 +120,33 @@ impl Config { /// For a log where every payload survives, use [`Stream`]. pub struct Snapshot { inner: moq_binary::snapshot::Producer, - rendition: Rendition, + listing: Listing, + /// Which catalog the entry lives in. The entry's own type is erased by `Listing`. + _catalog: PhantomData E>, } impl Snapshot { - pub(crate) fn new( + pub(crate) fn new + AsMut>( track: moq_net::track::Producer, - mut rendition: Rendition, - config: &Config, + rendition: crate::catalog::Rendition, + mut config: C, ) -> crate::Result { let mut binary = moq_binary::snapshot::Config::default(); - if config.compression { + if prepare(&mut config, Mode::Snapshot)? { binary.compression = moq_binary::Compression::Deflate; } let inner = moq_binary::snapshot::Producer::new(track, binary); - rendition.set(config.entry(Mode::Snapshot))?; - Ok(Self { inner, rendition }) + let listing = Listing::new(rendition, config)?; + Ok(Self { + inner, + listing, + _catalog: PhantomData, + }) } /// The track name, which is also the catalog key. pub fn name(&self) -> &str { - self.rendition.name() + self.listing.name() } /// Create a subscriber for the underlying track. @@ -130,7 +156,10 @@ impl Snapshot { /// Publish a new payload, superseding the previous one. pub fn update(&mut self, payload: impl Into) -> crate::Result<()> { - Ok(self.inner.update(payload)?) + let payload = payload.into(); + let len = payload.len(); + self.inner.update(payload)?; + self.listing.record(|| len) } /// Finish the track and retire its catalog entry. @@ -154,25 +183,28 @@ pub struct Stream { /// Cleared when a terminal failure ends the track, which retires the catalog entry with it. An /// entry advertising a track that can no longer accept records only misleads a consumer that /// discovers it afterwards. - rendition: Option>, + listing: Option, + /// Which catalog the entry lives in. The entry's own type is erased by `Listing`. + _catalog: PhantomData E>, } impl Stream { - pub(crate) fn new( + pub(crate) fn new + AsMut>( track: moq_net::track::Producer, - mut rendition: Rendition, - config: &Config, + rendition: crate::catalog::Rendition, + mut config: C, ) -> crate::Result { let mut binary = moq_binary::stream::Config::default(); - if config.compression { + if prepare(&mut config, Mode::Stream)? { binary.compression = moq_binary::Compression::Deflate; } let inner = moq_binary::stream::Producer::new(track, binary); - rendition.set(config.entry(Mode::Stream))?; + let listing = Listing::new(rendition, config)?; Ok(Self { inner, - name: rendition.name().to_string(), - rendition: Some(rendition), + name: listing.name().to_string(), + listing: Some(listing), + _catalog: PhantomData, }) } @@ -192,18 +224,25 @@ impl Stream { /// Append one payload to the log. /// /// A payload that cannot be written ends the track (see - /// [`moq_binary::stream::Producer::append`]) and retires the catalog entry with it. + /// [`moq_binary::stream::Producer::append`]) and retires the catalog entry with it. A catalog + /// error publishing the measured bitrate is returned after the payload was written, so the track + /// stays open and a retry would duplicate it. pub fn append(&mut self, payload: impl Into) -> crate::Result<()> { - let Err(err) = self.inner.append(payload) else { - return Ok(()); - }; - - // The inner producer has already closed the track. Dropping the rendition retires the catalog - // entry too: waiting for the handle to drop would keep advertising a track that can no longer - // accept records, so a consumer discovering it now would subscribe to an already-ended log. - self.rendition = None; + let payload = payload.into(); + let len = payload.len(); + if let Err(err) = self.inner.append(payload) { + // The inner producer has already closed the track. Dropping the listing retires the + // catalog entry too: waiting for the handle to drop would keep advertising a track that + // can no longer accept records, so a consumer discovering it now would subscribe to an + // already-ended log. + self.listing = None; + return Err(err.into()); + } - Err(err.into()) + match &mut self.listing { + Some(listing) => listing.record(|| len), + None => Ok(()), + } } /// Finish the track and retire its catalog entry. @@ -425,4 +464,198 @@ mod test { assert!(broadcast.create_track("data", None).is_err()); } + + /// A data track listed in an application's own section, beside its own per-track fields. + mod section { + use std::collections::BTreeMap; + + use serde::{Deserialize, Serialize}; + + use super::*; + use crate::catalog::Estimate; + use crate::catalog::hang::Catalog; + + #[derive(Serialize, Deserialize, Clone, Default, Debug, PartialEq)] + struct Ext { + #[serde(rename = "com.example.mavlink", default, skip_serializing_if = "BTreeMap::is_empty")] + mavlink: BTreeMap, + } + + impl CatalogExt for Ext {} + + #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] + struct Mavlink { + #[serde(flatten)] + binary: BinaryConfig, + sysid: u8, + } + + impl AsMut for Mavlink { + fn as_mut(&mut self) -> &mut BinaryConfig { + &mut self.binary + } + } + + impl RenditionConfig for Mavlink { + fn insert(self, catalog: &mut Catalog, name: &str) { + catalog.ext.mavlink.insert(name.to_string(), self); + } + fn get_mut<'a>(catalog: &'a mut Catalog, name: &str) -> Option<&'a mut Self> { + catalog.ext.mavlink.get_mut(name) + } + fn remove(catalog: &mut Catalog, name: &str) { + catalog.ext.mavlink.remove(name); + } + + fn detects() -> bool { + true + } + fn estimate(&self) -> Estimate { + Estimate::default() + .with_bitrate(self.binary.bitrate) + .with_jitter(self.binary.jitter) + } + fn set_estimate(&mut self, estimate: Estimate) { + self.binary.bitrate = estimate.bitrate; + self.binary.jitter = estimate.jitter; + } + } + + fn mavlink(sysid: u8) -> Mavlink { + let mut binary = BinaryConfig::new(Mode::Snapshot); + binary.compression = Some(Compression::Deflate); + Mavlink { binary, sysid } + } + + fn catalog() -> (moq_net::broadcast::Producer, crate::catalog::Producer) { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let config = crate::catalog::Config::default().with_catalog(Catalog::::default()); + let catalog = crate::catalog::Producer::new(&mut broadcast, config).unwrap(); + (broadcast, catalog) + } + + /// The producer fixes the mode and keeps the application's fields; a `Catalog` consumer + /// reads the entry back and subscribes through its embedded config alone. + #[tokio::test] + async fn roundtrips_through_a_catalog_consumer() { + let (mut broadcast, catalog) = catalog(); + let source = crate::source::announced(&broadcast.consume()); + let mut consumer = catalog.consume().unwrap(); + + let mut telemetry = catalog + .binary_stream(track(&mut broadcast, "telemetry"), mavlink(7)) + .unwrap(); + telemetry.append(&b"heartbeat"[..]).unwrap(); + + let published = consumer.next().await.unwrap().expect("catalog published"); + let entry = published.ext.mavlink.get("telemetry").expect("missing entry"); + assert_eq!(entry.sysid, 7, "the application's fields survive"); + assert_eq!(entry.binary.mode, Mode::Stream, "the producer fixes the mode"); + assert_eq!(entry.binary.compression, Some(Compression::Deflate)); + assert!(published.binary.tracks.is_empty(), "not listed in the binary section"); + + let mut reader = crate::catalog::Entry::new("telemetry", &entry.binary) + .subscribe(&source) + .await + .unwrap(); + telemetry.finish().unwrap(); + assert_eq!(reader.next().await.unwrap(), Some(Bytes::from_static(b"heartbeat"))); + assert_eq!(reader.next().await.unwrap(), None); + } + + #[test] + fn dropping_the_producer_retires_the_entry() { + let (mut broadcast, catalog) = catalog(); + let telemetry = catalog + .binary_snapshot(track(&mut broadcast, "telemetry"), mavlink(1)) + .unwrap(); + assert!(catalog.snapshot().ext.mavlink.contains_key("telemetry")); + + drop(telemetry); + assert!(!catalog.snapshot().ext.mavlink.contains_key("telemetry")); + } + + /// The name is owned per section, so an entry already in the application's section refuses + /// a second producer without touching the first. + #[test] + fn a_duplicate_name_is_refused() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let mut seed = Catalog::::default(); + seed.ext.mavlink.insert("telemetry".to_string(), mavlink(1)); + let config = crate::catalog::Config::default().with_catalog(seed); + let catalog = crate::catalog::Producer::new(&mut broadcast, config).unwrap(); + + assert!(matches!( + catalog.binary_stream(track(&mut broadcast, "telemetry"), mavlink(2)), + Err(crate::Error::Hang(hang::Error::Duplicate(_))) + )); + assert_eq!(catalog.snapshot().ext.mavlink["telemetry"].sysid, 1); + } + + /// A compression this build can't write is refused rather than advertised over plain frames. + #[test] + fn an_unknown_compression_is_refused() { + let (mut broadcast, catalog) = catalog(); + let mut entry = mavlink(1); + entry.binary.compression = Some(Compression::Unknown("zstd".to_string())); + + assert!(matches!( + catalog.binary_stream(track(&mut broadcast, "telemetry"), entry), + Err(crate::Error::UnsupportedCompression(_)) + )); + assert!(catalog.snapshot().ext.mavlink.is_empty()); + } + + /// The producer publishes locally, so an entry pointing at another broadcast is refused rather + /// than advertised: consumers would resolve it there and never reach the published payloads. + #[test] + fn a_broadcast_reference_is_refused() { + let (mut broadcast, catalog) = catalog(); + let mut entry = mavlink(1); + entry.binary.broadcast = Some(moq_net::path::RelativeOwned::new("source")); + + assert!(matches!( + catalog.binary_stream(track(&mut broadcast, "telemetry"), entry), + Err(crate::Error::ForeignBroadcast) + )); + assert!(catalog.snapshot().ext.mavlink.is_empty()); + } + + /// Writes fill an absent bitrate, through the entry's embedded config. + #[test] + fn detects_bitrate() { + let (mut broadcast, catalog) = catalog(); + let mut telemetry = catalog + .binary_stream(track(&mut broadcast, "telemetry"), mavlink(1)) + .unwrap(); + + // 40ms payloads of 5 kB: 1 Mbps, over more than the bitrate window. + for i in 0..60u64 { + let now = moq_net::Timestamp::from_micros(i * 40_000).unwrap(); + telemetry.listing.as_mut().unwrap().record_at(now, 5_000).unwrap(); + } + + let entry = &catalog.snapshot().ext.mavlink["telemetry"]; + assert_eq!(entry.binary.bitrate, Some(1_000_000)); + assert_eq!(entry.binary.jitter, None, "write spacing is not a flush delay"); + } + + /// A supplied bitrate is authoritative, so writes aren't measured at all: for JSON that would + /// be a second serialization per write, for nothing. + #[test] + fn a_supplied_bitrate_skips_measurement() { + let (mut broadcast, catalog) = catalog(); + let mut entry = mavlink(1); + entry.binary.bitrate = Some(64_000); + let mut telemetry = catalog + .binary_stream(track(&mut broadcast, "telemetry"), entry) + .unwrap(); + + let listing = telemetry.listing.as_mut().unwrap(); + listing + .record(|| panic!("measured a write despite a supplied bitrate")) + .unwrap(); + assert_eq!(catalog.snapshot().ext.mavlink["telemetry"].binary.bitrate, Some(64_000)); + } + } } diff --git a/rs/moq-mux/src/catalog/data.rs b/rs/moq-mux/src/catalog/data.rs new file mode 100644 index 0000000000..10f73b0a0b --- /dev/null +++ b/rs/moq-mux/src/catalog/data.rs @@ -0,0 +1,128 @@ +use super::hang::CatalogExt; +use super::{Estimator, Rendition, RenditionConfig}; + +/// The config a data producer ([`Producer::json_stream`](super::Producer::json_stream) and the +/// like) publishes as its catalog entry. +/// +/// Implemented for the [`json::Config`](crate::json::Config) and +/// [`binary::Config`](crate::binary::Config) builders, which list the track in the `json` or +/// `binary` section. Also implemented for any [`RenditionConfig`] that embeds the data config `D` +/// ([`JsonConfig`](hang::catalog::JsonConfig) or [`BinaryConfig`](hang::catalog::BinaryConfig)) +/// through [`AsMut`], which is how an application lists a data track in its own section beside its +/// own per-track fields; see the [`RenditionConfig`] example. +/// +/// The producer sets the embedded config's `mode`, encodes the track with its `compression`, and +/// owns the entry: it is written when the producer is created and removed when it drops. +pub trait IntoRendition { + /// The catalog entry, embedding `D`. + type Config: RenditionConfig + AsMut; + + /// Build the catalog entry. + fn into_rendition(self) -> Self::Config; +} + +impl + AsMut> IntoRendition for C { + type Config = C; + + fn into_rendition(self) -> C { + self + } +} + +/// A data track's catalog entry, owned for the life of its producer and kept current with the +/// bitrate its writes measure. +/// +/// Erases the entry's type, so a data producer's own type doesn't depend on which section lists it. +pub(crate) struct Listing { + rendition: Box, + estimator: Estimator, + /// Whether writes are measured: only when the entry detects its estimate and the publisher + /// didn't supply a bitrate, which detection never overrides. + measures: bool, +} + +/// The parts of a [`Rendition`] a [`Listing`] uses, without its config type. +trait Owned: Send + Sync { + fn name(&self) -> &str; + fn timestamp(&self) -> crate::Result; + fn estimate(&mut self, estimate: super::Estimate) -> crate::Result<()>; +} + +impl> Owned for Rendition { + fn name(&self) -> &str { + Rendition::name(self) + } + fn timestamp(&self) -> crate::Result { + Rendition::timestamp(self, None) + } + fn estimate(&mut self, estimate: super::Estimate) -> crate::Result<()> { + Rendition::estimate(self, estimate) + } +} + +impl Listing { + /// Publish `config` as the entry `rendition` reserved. + pub(crate) fn new>( + mut rendition: Rendition, + config: C, + ) -> crate::Result { + let measures = C::detects() && config.estimate().bitrate.is_none(); + rendition.set(config)?; + Ok(Self { + rendition: Box::new(rendition), + estimator: Estimator::new(), + measures, + }) + } + + /// The track name, which is also the catalog key. + pub(crate) fn name(&self) -> &str { + self.rendition.name() + } + + /// Measure a write of `bytes`, stamped on the broadcast clock. + /// + /// `bytes` is only evaluated for an entry that measures its bitrate, since measuring can cost a + /// second serialization. + pub(crate) fn record(&mut self, bytes: impl FnOnce() -> usize) -> crate::Result<()> { + if !self.measures { + return Ok(()); + } + let now = self.rendition.timestamp()?; + self.record_at(now, bytes()) + } + + /// [`record`](Self::record) at a chosen time, which a test needs since the broadcast clock only + /// moves in real time. + pub(crate) fn record_at(&mut self, now: moq_net::Timestamp, bytes: usize) -> crate::Result<()> { + // Each write is its own span, closed by the next one. + self.estimator.cut(Some(now)); + self.estimator.write(now, bytes); + + // The spacing between writes is the application's cadence, not a flush delay, so only the + // bitrate is measured. A publisher that knows its jitter sets it on the entry. + let estimate = self.estimator.estimate().with_jitter(None); + self.rendition.estimate(estimate) + } +} + +/// The serialized size of `value`, as an upper bound on what a JSON write puts on the wire: +/// compression and deltas only shrink it. +pub(crate) fn json_len(value: &T) -> usize { + struct Count(usize); + + impl std::io::Write for Count { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0 += buf.len(); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + let mut count = Count(0); + // Only reached after the producer serialized the same value, so this cannot fail. + let _ = serde_json::to_writer(&mut count, value); + count.0 +} diff --git a/rs/moq-mux/src/catalog/mod.rs b/rs/moq-mux/src/catalog/mod.rs index c18738f8b2..88ef14f672 100644 --- a/rs/moq-mux/src/catalog/mod.rs +++ b/rs/moq-mux/src/catalog/mod.rs @@ -25,6 +25,7 @@ pub mod msf; mod claim; mod consumer; +mod data; mod entry; mod estimate; mod format; @@ -35,6 +36,8 @@ pub(crate) mod tracks; pub(crate) use claim::Claim; pub use consumer::Consumer; +pub use data::IntoRendition; +pub(crate) use data::{Listing, json_len}; pub use entry::Entry; pub use estimate::{Estimate, Estimator}; pub use format::*; diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index 9931f815d5..0d0958a123 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -653,20 +653,28 @@ impl Producer { /// Publish `track` as a latest-value JSON track, advertising it in the catalog. /// /// The caller creates the track on the broadcast, as it does for a media track; this writes its - /// catalog entry and removes the - /// entry when the returned handle drops. The catalog key is [`track.name()`](moq_net::track::Producer::name) - /// verbatim, with no `.z` suffix even when compressed, since the entry's compression flag is - /// what a consumer reads. + /// catalog entry and removes the entry when the returned handle drops. The catalog key is + /// [`track.name()`](moq_net::track::Producer::name) verbatim, with no `.z` suffix even when + /// compressed, since the entry's compression flag is what a consumer reads. /// - /// Errors if the catalog already carries an entry under that name, for example one seeded - /// through [`Config::with_catalog`] or one pointing at a sibling broadcast. + /// `config` is a [`json::Config`](crate::json::Config) for the `json` section, or an + /// application's own entry embedding a [`JsonConfig`](hang::catalog::JsonConfig) (see + /// [`IntoRendition`](super::IntoRendition)). The producer sets its `mode`, encodes the track + /// with its `compression`, and fills an absent `bitrate` from what it writes. A + /// [`delta_ratio`](crate::json::Config::delta_ratio) on `json::Config` selects the snapshot + /// encoder; it is not written into the catalog entry. The config is `'static` so that ratio + /// can be read off the builder. + /// + /// Errors if the entry's section already carries that name, for example an entry seeded + /// through [`Config::with_catalog`] or one pointing at a sibling broadcast, or if the entry + /// declares a compression this build can't write or references another broadcast. pub fn json_snapshot( &self, track: moq_net::track::Producer, - config: crate::json::Config, + config: impl super::IntoRendition + 'static, ) -> crate::Result> { let rendition = self.data_entry(track.name())?; - crate::json::Snapshot::new(track, rendition, &config) + crate::json::Snapshot::new(track, rendition, config) } /// Publish `track` as an append-log JSON track, advertising it in the catalog. @@ -676,36 +684,37 @@ impl Producer { pub fn json_stream( &self, track: moq_net::track::Producer, - config: crate::json::Config, + config: impl super::IntoRendition, ) -> crate::Result> { let rendition = self.data_entry(track.name())?; - crate::json::Stream::new(track, rendition, &config) + crate::json::Stream::new(track, rendition, config.into_rendition()) } /// Publish `track` as a latest-value binary track, advertising it in the catalog. /// /// See [`json_snapshot`](Self::json_snapshot) for the lifecycle; this differs only in that the - /// payloads are opaque bytes. + /// payloads are opaque bytes, and `config` is a [`binary::Config`](crate::binary::Config) or an + /// entry embedding a [`BinaryConfig`](hang::catalog::BinaryConfig). pub fn binary_snapshot( &self, track: moq_net::track::Producer, - config: crate::binary::Config, + config: impl super::IntoRendition, ) -> crate::Result> { let rendition = self.data_entry(track.name())?; - crate::binary::Snapshot::new(track, rendition, &config) + crate::binary::Snapshot::new(track, rendition, config.into_rendition()) } /// Publish `track` as an append-log binary track, advertising it in the catalog. /// - /// See [`json_snapshot`](Self::json_snapshot) for the lifecycle; this differs only in that the - /// payloads are opaque bytes and every one is preserved rather than superseded. + /// See [`binary_snapshot`](Self::binary_snapshot); this differs only in that every payload is + /// preserved rather than superseded. pub fn binary_stream( &self, track: moq_net::track::Producer, - config: crate::binary::Config, + config: impl super::IntoRendition, ) -> crate::Result> { let rendition = self.data_entry(track.name())?; - crate::binary::Stream::new(track, rendition, &config) + crate::binary::Stream::new(track, rendition, config.into_rendition()) } /// Reserve the catalog entry a data producer owns, keyed by its track name. diff --git a/rs/moq-mux/src/catalog/tracks.rs b/rs/moq-mux/src/catalog/tracks.rs index 95f9f78196..6c2c349fdb 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -7,52 +7,74 @@ use super::hang::{Catalog, CatalogExt}; /// A catalog config that can be published as a named rendition. /// -/// Implement it on your own config type to get the full catalog lifecycle through -/// [`Reserved::track`]: reservation gating, removal on drop, and optional jitter/bitrate detection. -/// [`VideoConfig`](hang::catalog::VideoConfig) and [`AudioConfig`](hang::catalog::AudioConfig) -/// implement it for every extension; a custom config implements it for the one [`CatalogExt`] that -/// holds it: +/// Implement it on your own config type to get the full catalog lifecycle: reservation gating, +/// removal on drop, and optional jitter/bitrate detection. [`VideoConfig`](hang::catalog::VideoConfig) +/// and [`AudioConfig`](hang::catalog::AudioConfig) implement it for every extension; a custom +/// config implements it for the one [`CatalogExt`] that holds it. Publish a media track under it +/// with [`Reserved::track`], or a data track with [`Producer::binary_stream`] and the like when it +/// embeds a data config (see [`IntoRendition`](super::IntoRendition)): /// /// ``` +/// # use std::collections::BTreeMap; +/// # use hang::catalog::{BinaryConfig, Mode}; /// # use moq_mux::catalog::{Estimate, RenditionConfig}; /// # use moq_mux::catalog::hang::{Catalog, CatalogExt}; /// # use serde::{Deserialize, Serialize}; -/// # use std::collections::BTreeMap; /// #[derive(Serialize, Deserialize, Clone, Default)] -/// struct MyExt { -/// telemetry: BTreeMap, +/// struct Ext { +/// #[serde(rename = "com.example.mavlink", default)] +/// mavlink: BTreeMap, /// } -/// impl CatalogExt for MyExt {} +/// impl CatalogExt for Ext {} /// -/// #[derive(Serialize, Deserialize, Clone, Default)] -/// struct Telemetry { -/// schema: String, -/// bitrate: Option, +/// #[derive(Serialize, Deserialize, Clone)] +/// struct Mavlink { +/// #[serde(flatten)] +/// binary: BinaryConfig, +/// sysid: u8, /// } /// -/// impl RenditionConfig for Telemetry { -/// fn detects() -> bool { -/// true +/// impl AsMut for Mavlink { +/// fn as_mut(&mut self) -> &mut BinaryConfig { +/// &mut self.binary /// } +/// } /// -/// fn insert(self, catalog: &mut Catalog, name: &str) { -/// catalog.ext.telemetry.insert(name.to_string(), self); +/// impl RenditionConfig for Mavlink { +/// fn insert(self, catalog: &mut Catalog, name: &str) { +/// catalog.ext.mavlink.insert(name.to_string(), self); /// } -/// fn get_mut<'a>(catalog: &'a mut Catalog, name: &str) -> Option<&'a mut Self> { -/// catalog.ext.telemetry.get_mut(name) +/// fn get_mut<'a>(catalog: &'a mut Catalog, name: &str) -> Option<&'a mut Self> { +/// catalog.ext.mavlink.get_mut(name) /// } -/// fn remove(catalog: &mut Catalog, name: &str) { -/// catalog.ext.telemetry.remove(name); +/// fn remove(catalog: &mut Catalog, name: &str) { +/// catalog.ext.mavlink.remove(name); /// } /// -/// // Opt into bitrate detection; jitter is left undetected. +/// // Opt into bitrate detection through the embedded config. +/// fn detects() -> bool { +/// true +/// } /// fn estimate(&self) -> Estimate { -/// Estimate::default().with_bitrate(self.bitrate) +/// Estimate::default().with_bitrate(self.binary.bitrate).with_jitter(self.binary.jitter) /// } /// fn set_estimate(&mut self, estimate: Estimate) { -/// self.bitrate = estimate.bitrate; +/// self.binary.bitrate = estimate.bitrate; +/// self.binary.jitter = estimate.jitter; /// } /// } +/// +/// # fn example( +/// # broadcast: &mut moq_net::broadcast::Producer, +/// # catalog: &moq_mux::catalog::Producer, +/// # ) -> moq_mux::Result<()> { +/// let track = broadcast.create_track("telemetry", None)?; +/// // The producer fixes the mode, so the one passed here is only a placeholder. +/// let entry = Mavlink { binary: BinaryConfig::new(Mode::Stream), sysid: 1 }; +/// let mut telemetry = catalog.binary_stream(track, entry)?; +/// telemetry.append(&b"\xfd..."[..])?; +/// # Ok(()) +/// # } /// ``` /// /// Note that `insert` takes the whole [`Catalog`], not just the extension, so the built-in media @@ -62,7 +84,7 @@ use super::hang::{Catalog, CatalogExt}; /// [`Reserved::track`] and [`Producer::track`](super::Producer::track) enroll the track in the /// broadcast timeline, measure it, and keep its estimate current automatically. pub trait RenditionConfig: Clone + Send + 'static { - /// Whether container writes should update this config's estimate fields. + /// Whether container or data-track writes should update this config's estimate fields. fn detects() -> bool { false } @@ -95,6 +117,17 @@ impl RenditionConfig for hang::catalog::JsonConfig { fn remove(catalog: &mut Catalog, name: &str) { catalog.json.tracks.remove(name); } + + fn detects() -> bool { + true + } + fn estimate(&self) -> Estimate { + Estimate::default().with_jitter(self.jitter).with_bitrate(self.bitrate) + } + fn set_estimate(&mut self, estimate: Estimate) { + self.jitter = estimate.jitter; + self.bitrate = estimate.bitrate; + } } impl RenditionConfig for hang::catalog::BinaryConfig { @@ -107,6 +140,17 @@ impl RenditionConfig for hang::catalog::BinaryConfig { fn remove(catalog: &mut Catalog, name: &str) { catalog.binary.tracks.remove(name); } + + fn detects() -> bool { + true + } + fn estimate(&self) -> Estimate { + Estimate::default().with_jitter(self.jitter).with_bitrate(self.bitrate) + } + fn set_estimate(&mut self, estimate: Estimate) { + self.jitter = estimate.jitter; + self.bitrate = estimate.bitrate; + } } /// Caller-provided catalog fields for a video track: a starting point for what the importer detects. diff --git a/rs/moq-mux/src/error.rs b/rs/moq-mux/src/error.rs index 5a1637f7a0..a0a50989da 100644 --- a/rs/moq-mux/src/error.rs +++ b/rs/moq-mux/src/error.rs @@ -48,6 +48,10 @@ pub enum Error { #[error("unsupported track compression: {0}")] UnsupportedCompression(String), + /// A locally published track's catalog entry points at another broadcast. + #[error("a locally published track can't reference another broadcast")] + ForeignBroadcast, + /// Error parsing or building CMAF moof+mdat fragments. #[error("cmaf: {0}")] Cmaf(#[from] crate::container::fmp4::Error), diff --git a/rs/moq-mux/src/json.rs b/rs/moq-mux/src/json.rs index 26692fa745..e08e610acd 100644 --- a/rs/moq-mux/src/json.rs +++ b/rs/moq-mux/src/json.rs @@ -51,18 +51,21 @@ //! # } //! ``` +use std::marker::PhantomData; + use serde::Serialize; use serde::de::DeserializeOwned; use hang::catalog::{Compression, JsonConfig, Mode}; -use crate::catalog::Rendition; use crate::catalog::hang::CatalogExt; +use crate::catalog::{IntoRendition, Listing, RenditionConfig}; /// Everything a JSON track declares about itself, beyond its mode and name. /// /// Start from [`default`](Default::default) and chain the setters. The mode is not in here: it is -/// fixed by which producer you create. +/// fixed by which producer you create. To list the track in an application's own catalog section +/// instead of `json`, pass that section's entry (see [`IntoRendition`]). #[derive(Debug, Clone, Default)] #[non_exhaustive] pub struct Config { @@ -101,46 +104,86 @@ impl Config { self.delta_ratio = Some(delta_ratio); self } +} + +impl IntoRendition for Config { + type Config = JsonConfig; - /// The catalog entry describing a track published under this config in `mode`. - pub(crate) fn entry(&self, mode: Mode) -> JsonConfig { - let mut entry = JsonConfig::new(mode); + fn into_rendition(self) -> JsonConfig { + // The producer overwrites the mode with the one it publishes in. + let mut entry = JsonConfig::new(Mode::Snapshot); entry.compression = self.compression.then_some(Compression::Deflate); - entry.schema = self.schema.clone(); + entry.schema = self.schema; entry } } +/// Fix `config`'s mode and return whether its frames are compressed. +/// +/// Errors on a compression this build can't write, rather than advertising one the frames don't use, +/// and on a `broadcast` reference, which would point consumers away from the track this publishes. +fn prepare(config: &mut impl AsMut, mode: Mode) -> crate::Result { + let json = config.as_mut(); + if json.broadcast.is_some() { + return Err(crate::Error::ForeignBroadcast); + } + json.mode = mode; + crate::compression(json.compression.as_ref()) +} + +/// The snapshot encoder ratio on a [`Config`] builder, if `config` is one. +/// +/// [`IntoRendition`] only returns the catalog entry, and this ratio is not a catalog field. +/// Downcast keeps it on the existing builder instead of a new trait method. +fn delta_ratio_of(config: &C) -> Option { + (config as &dyn std::any::Any) + .downcast_ref::() + .and_then(|config| config.delta_ratio) +} + /// Publishes a latest-value JSON track, advertised in the catalog for as long as this handle lives. /// /// Every [`update`](Self::update) supersedes the last, so a consumer reads only the newest value. /// For a log where every record survives, use [`Stream`]. pub struct Snapshot { inner: moq_json::snapshot::Producer, - rendition: Rendition, + listing: Listing, + /// Which catalog the entry lives in. The entry's own type is erased by `Listing`. + _catalog: PhantomData E>, } impl Snapshot { - pub(crate) fn new( + pub(crate) fn new( track: moq_net::track::Producer, - mut rendition: Rendition, - config: &Config, - ) -> crate::Result { + rendition: crate::catalog::Rendition, + config: C, + ) -> crate::Result + where + C: IntoRendition + std::any::Any, + { + // Read before `into_rendition` consumes the builder. Only [`Config`] carries a ratio; + // a custom section entry has none, and the default encoder ratio applies. + let delta_ratio = delta_ratio_of(&config); + let mut config = config.into_rendition(); let mut json = moq_json::snapshot::Config::default(); - if config.compression { + if prepare(&mut config, Mode::Snapshot)? { json.compression = moq_json::Compression::Deflate; } - if let Some(delta_ratio) = config.delta_ratio { + if let Some(delta_ratio) = delta_ratio { json.delta_ratio = delta_ratio; } let inner = moq_json::snapshot::Producer::new(track, json); - rendition.set(config.entry(Mode::Snapshot))?; - Ok(Self { inner, rendition }) + let listing = Listing::new(rendition, config)?; + Ok(Self { + inner, + listing, + _catalog: PhantomData, + }) } /// The track name, which is also the catalog key. pub fn name(&self) -> &str { - self.rendition.name() + self.listing.name() } /// Create a subscriber for the underlying track. @@ -150,7 +193,8 @@ impl Snapshot { /// Publish a new value, superseding the previous one. pub fn update(&mut self, value: &T) -> crate::Result<()> { - Ok(self.inner.update(value)?) + self.inner.update(value)?; + self.listing.record(|| crate::catalog::json_len(value)) } /// Finish the track and retire its catalog entry. @@ -174,25 +218,28 @@ pub struct Stream { /// Cleared when a terminal failure ends the track, which retires the catalog entry with it. An /// entry advertising a track that can no longer accept records only misleads a consumer that /// discovers it afterwards. - rendition: Option>, + listing: Option, + /// Which catalog the entry lives in. The entry's own type is erased by `Listing`. + _catalog: PhantomData E>, } impl Stream { - pub(crate) fn new( + pub(crate) fn new + AsMut>( track: moq_net::track::Producer, - mut rendition: Rendition, - config: &Config, + rendition: crate::catalog::Rendition, + mut config: C, ) -> crate::Result { let mut json = moq_json::stream::Config::default(); - if config.compression { + if prepare(&mut config, Mode::Stream)? { json.compression = moq_json::Compression::Deflate; } let inner = moq_json::stream::Producer::new(track, json); - rendition.set(config.entry(Mode::Stream))?; + let listing = Listing::new(rendition, config)?; Ok(Self { inner, - name: rendition.name().to_string(), - rendition: Some(rendition), + name: listing.name().to_string(), + listing: Some(listing), + _catalog: PhantomData, }) } @@ -211,19 +258,22 @@ impl Stream { /// Append one record to the log. /// - /// Any failure ends the track (see [`moq_json::stream::Producer::append`]) and retires the - /// catalog entry with it. + /// A record that cannot be written ends the track (see [`moq_json::stream::Producer::append`]) + /// and retires the catalog entry with it. A catalog error publishing the measured bitrate is + /// returned after the record was written, so the track stays open and a retry would duplicate it. pub fn append(&mut self, value: &T) -> crate::Result<()> { - let Err(err) = self.inner.append(value) else { - return Ok(()); - }; - - // The inner producer has already ended the track. Dropping the rendition retires the catalog - // entry: waiting for the handle to drop would keep advertising a track that can no longer - // accept records, so a consumer discovering it now would subscribe to an already-ended log. - self.rendition = None; + if let Err(err) = self.inner.append(value) { + // The inner producer has already ended the track. Dropping the listing retires the catalog + // entry: waiting for the handle to drop would keep advertising a track that can no longer + // accept records, so a consumer discovering it now would subscribe to an already-ended log. + self.listing = None; + return Err(err.into()); + } - Err(err.into()) + match &mut self.listing { + Some(listing) => listing.record(|| crate::catalog::json_len(value)), + None => Ok(()), + } } /// Finish the track and retire its catalog entry. @@ -414,6 +464,46 @@ mod test { assert_eq!(drain(consumer), vec![json!({ "live": true })]); } + /// `delta_ratio` is an encoder setting on [`Config`], not a catalog field. A ratio of 0 + /// publishes each value as its own group; a positive ratio keeps the next value in that group. + #[test] + fn a_config_delta_ratio_reaches_the_encoder() { + let (mut broadcast, catalog) = catalog(); + let mut full = catalog + .json_snapshot::(track(&mut broadcast, "full"), Config::default().with_delta_ratio(0)) + .unwrap(); + let mut delta = catalog + .json_snapshot::(track(&mut broadcast, "delta"), Config::default().with_delta_ratio(100)) + .unwrap(); + + let mut full_track = full.consume(); + let mut delta_track = delta.consume(); + for value in [json!({ "n": 1 }), json!({ "n": 2 })] { + full.update(&value).unwrap(); + delta.update(&value).unwrap(); + } + full.finish().unwrap(); + delta.finish().unwrap(); + + // The default subscription budget keeps only the latest group. Ratio 0 rolled a new + // group for the second value, so that group holds one frame. A positive ratio appends + // the second value to the same group. + assert_eq!(ready_groups(&mut full_track), vec![1]); + assert_eq!(ready_groups(&mut delta_track), vec![2]); + } + + fn ready_groups(subscriber: &mut moq_net::track::Subscriber) -> Vec { + let waiter = kio::Waiter::noop(); + let mut counts = Vec::new(); + loop { + match subscriber.poll_recv_group(&waiter) { + Poll::Ready(Ok(Some(group))) => counts.push(group.frame_count()), + Poll::Ready(Ok(None)) | Poll::Pending => return counts, + Poll::Ready(Err(err)) => panic!("group ended in error: {err}"), + } + } + } + #[test] fn the_entry_describes_how_to_read_the_track() { let (mut broadcast, catalog) = catalog(); @@ -486,6 +576,35 @@ mod test { assert_eq!(catalog.snapshot().json.tracks.get("chat"), Some(&existing)); } + /// Writes fill an absent bitrate; one the publisher supplied is left alone. + #[test] + fn writes_fill_an_absent_bitrate() { + let (mut broadcast, catalog) = catalog(); + let mut gps = catalog + .json_stream::(track(&mut broadcast, "gps"), Config::default()) + .unwrap(); + let mut supplied = JsonConfig::new(Mode::Stream); + supplied.bitrate = Some(4_200); + let mut status = catalog + .json_snapshot::(track(&mut broadcast, "status"), supplied) + .unwrap(); + + // 40ms records of 500 bytes: 100 kbps, over more than the bitrate window. + for i in 0..60u64 { + let now = moq_net::Timestamp::from_micros(i * 40_000).unwrap(); + gps.listing.as_mut().unwrap().record_at(now, 500).unwrap(); + status.listing.record_at(now, 500).unwrap(); + } + + assert_eq!(entry(&catalog, "gps").bitrate, Some(100_000)); + assert_eq!(entry(&catalog, "status").bitrate, Some(4_200)); + assert_eq!( + entry(&catalog, "gps").jitter, + None, + "write spacing is not a flush delay" + ); + } + /// The catalog is the only thing that announces a data track, so walking it is the discovery /// path. Each entry carries its own name, so nothing has to be threaded alongside it. #[test]