feat(moq-mux): publish data tracks into an application's own catalog section

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:

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<hang::catalog::BinaryConfig> for Mavlink {
+    fn as_mut(&mut self) -> &mut hang::catalog::BinaryConfig {
+        &mut self.binary
+    }
+}
+
+// Plus `RenditionConfig<Ext>` 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<Ext>` and subscribe with
+`catalog::Entry::new(name, &entry.binary)`.
+
 ```bash
 cargo add moq-mux
 ```

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"}

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)),
+		),
+	),
 });
 
 /**

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)),
+		),
+	),
 });
 
 /**

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) };
+		});
+	});
+}

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<string, Record<string, number | undefined>> {
+	const pick = (tracks: Record<string, { jitter?: number }> | 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);

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

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

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<E>`: 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