From c5b2b84634804713dc5341766dd02aba3bae28ad Mon Sep 17 00:00:00 2001 From: t0ms Date: Fri, 16 Oct 2026 22:23:31 +0100 Subject: [PATCH] feat(cli): report import stats and export feedback `moq import --stats` names a stats track in the catalog the import builds and fills it with what each rendition has written: frames, payload bytes, and video keyframes, with the connection's transport. `moq export --echo ` publishes the feedback a catalog solicits, counting what the export reads and its newest arrival. `moq export stats` and `moq export echo` print either as JSON lines. moq-mux counts at the container producer and the source, so every container import and export shares the counters. Capture encoders, gateways, `archive`, and `ts --program all` refuse `--stats`; MSF catalogs refuse `--echo`. Co-authored-by: Claude Opus 5.5 --- doc/bin/cli.md | 31 ++ doc/concept/stats.md | 9 +- quest/m2/stats/README.md | 4 +- quest/m2/stats/rust.md | 67 ++-- quest/m3/msf-publish-tracks.md | 4 +- rs/moq-cli/Cargo.toml | 4 +- rs/moq-cli/src/args.rs | 140 ++++++- rs/moq-cli/src/main.rs | 76 +++- rs/moq-cli/src/publish.rs | 66 ++- rs/moq-cli/src/report.rs | 443 +++++++++++++++++++++ rs/moq-cli/src/subscribe.rs | 5 + rs/moq-cli/tests/stats.rs | 271 +++++++++++++ rs/moq-mux/src/catalog/producer.rs | 28 +- rs/moq-mux/src/container/fmp4/import.rs | 8 + rs/moq-mux/src/container/producer.rs | 22 + rs/moq-mux/src/container/source.rs | 6 + rs/moq-mux/src/container/ts/export_test.rs | 47 +++ rs/moq-mux/src/lib.rs | 1 + rs/moq-mux/src/source.rs | 24 +- rs/moq-mux/src/stats.rs | 168 ++++++++ 20 files changed, 1345 insertions(+), 79 deletions(-) create mode 100644 rs/moq-cli/src/report.rs create mode 100644 rs/moq-cli/tests/stats.rs create mode 100644 rs/moq-mux/src/stats.rs diff --git a/doc/bin/cli.md b/doc/bin/cli.md index d92d2accf6c..4901362671e 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -397,6 +397,37 @@ is a replacement. An export that fails while the broadcast stays up, such as on a codec TS cannot carry, exits 1 without lingering. Only `ts` can mark a break, so the other formats refuse `--linger`. +## Stats and feedback + +`import --stats` names a stats track in the catalog and reports what each +rendition has written, about once a second while someone subscribes, with the +connection's RTT, estimated send rate, and loss. It covers the stdin imports; +capture, the gateways, `archive`, and `ts --program all` refuse it. + +`export --echo ` answers a publisher that solicits feedback: under the +catalog's `echo` path it publishes `.echo`, reporting what each rendition +the export reads has received and its newest arrival, until the export ends. +The name is one path segment. It covers the stdout container exports and reads +the solicitation from the hang catalog, so `--catalog-format msf` refuses it. + +The `stats` and `echo` sinks print reports as JSON lines. `export stats` +follows one broadcast's stats track until the broadcast ends; `export echo` +prints `{"broadcast", "feedback"}` for every `.echo` broadcast under +`--broadcast` as they come and go. + +```bash +moq ... --broadcast event import --stats ts < feed.ts +moq ... --broadcast event export stats +moq ... --broadcast event export --echo alice ts > out.ts +moq ... --broadcast viewers export echo +``` + +Counters are cumulative for the life of the process: a rendition that ends +keeps its final counts, and one that returns under the same ID continues them. +An fMP4 import counts samples as frames. Encoder counters (`skipped`, +`bitrate`) and playback feedback (`decoded`, `late`, `stalls`) are not reported +yet. [Stats](/concept/stats#media-reports) has the schema. + ## Debugging `RUST_LOG=debug` prints the negotiated version and every subscription. diff --git a/doc/concept/stats.md b/doc/concept/stats.md index 9f9fd4525ac..d1d9c914484 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -32,9 +32,12 @@ 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. +`moq import --stats` and `moq export --echo` report them from the +[CLI](/bin/cli#stats-and-feedback). An import counts the frames and payload +bytes it writes to each rendition, and video keyframes; an export counts what +it reads and records its newest arrival. See the +[hang draft](/draft/moq-hang#media-stats) for track framing, field units, and +the feedback convention. ## Broadcasts diff --git a/quest/m2/stats/README.md b/quest/m2/stats/README.md index 45195f92ee9..afec56dcf89 100644 --- a/quest/m2/stats/README.md +++ b/quest/m2/stats/README.md @@ -111,8 +111,8 @@ connection view and the dashboard flow. ## Required -- [Rust reporters](/quest/m2/stats/rust.md) - the CLI, players, encoders, and - moq-mux remuxes publish stats and feedback +- [Rust reporters](/quest/m2/stats/rust.md) - the player, encoders, and + gateways publish stats and feedback, as the CLI remuxes do - [Browser reporters](/quest/m2/stats/js.md) - `` publishes stats and `` publishes feedback - [Client health](/quest/m2/stats/health.md) - two snapshots become a diff --git a/quest/m2/stats/rust.md b/quest/m2/stats/rust.md index bc9e18e0bd5..985ade5a447 100644 --- a/quest/m2/stats/rust.md +++ b/quest/m2/stats/rust.md @@ -1,52 +1,39 @@ -# [M] The Rust CLI, players, encoders, and remuxes report stats and feedback +# [M] The Rust player, encoders, and gateways report stats and feedback ## Goal -- `moq import --stats` and `moq publish --stats` announce a stats track in - the catalog and fill it with what they send. -- `moq play` and `moq export`, given `--echo `, publish an `.echo` - broadcast under each watched catalog's echo path and fill its feedback - track. -- `moq export stats ` prints a broadcast's stats snapshots, and - `moq export echo ` prints the feedback of every `.echo` broadcast - under a prefix, as JSON lines. +- `moq import capture --stats` announces a stats track in the catalog and + fills it with what its encoders send, `skipped` and `bitrate` included. +- `moq play`, given `--echo `, publishes an `.echo` broadcast under each + watched catalog's echo path and fills its feedback track with what it + received, decoded, and played. +- The gateway imports (`hls`, `rtmp`, `srt`, `rtc`) and `archive` take + `--stats`. ## Plan +- `moq import --stats` for the stdin imports, `moq export --echo`, and the + `export stats` and `export echo` sinks already report what the `moq-mux` + importers write and exporters read (`rs/moq-mux/src/stats.rs`, + `rs/moq-cli/src/report.rs`). This quest adds the counters they lack and + reuses the rest; `import --stats` refuses capture and the gateways until it + lands. - Counters come from the source, per track, as cumulative `stats()` snapshots: no callback, no per-frame lock. - - Publisher: the `moq-video` and `moq-audio` encode producers count frames, - bytes, keyframes, and drops. The `moq-mux` importers count the frames and - bytes they write per track. A remux never encodes, so without them the - CLI would have nothing to report for `moq import`. + - Publisher: the `moq-video` and `moq-audio` encode producers create their + tracks directly on the broadcast, so they count frames, bytes, keyframes, + and drops themselves, and expose the target bitrate. - Viewer: the decode and render paths count received, decoded, late, - stalled, underruns, and errors. The `moq-mux` exporters count the frames - and bytes they receive per track, so `moq export --echo` reports a row - for each rendition it remuxes. `underruns` in + stalled, underruns, and errors. `underruns` in `rs/moq-audio/src/playback` is already counted privately. -- `moq-cli` folds the counters into `hang::stats::Snapshot` on the stats - interval and writes it through `moq_json::snapshot`, with the `.z` sibling. - `transport` comes from the connection's `ConnectionStats`. - Decided in the quest audit: this quest owns generic per-rendition reporting; - [TS health stats](/quest/m2/ts-health-stats.md) owns serializing and flattening - TS container counters under `mpegts` in publisher stats and export `.echo`. - That integration stays in m2 so this quest does not duplicate its work. -- Feedback: after reading a catalog with an `echo` section, the player - resolves the echo path against the broadcast, appends `.echo`, and - publishes there with the fixed feedback - track, keyed by the catalog's rendition IDs. Each catalog update - 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. -- The media test publishes with `--stats` and plays with `--echo` against a - publisher that solicits feedback. It asserts that the publisher's frame - count matches what was sent and that the viewer's newest arrival advances. +- The gateways build their catalogs in their own crates, so each needs a way + to name the stats track before its first catalog. +- `moq play --echo` reconciles each watched catalog's `echo` section as + `export --echo` does, and refuses `--catalog-format msf` the same way: the + MSF catalog does not carry an `echo` solicitation. +- The media test captures with `--stats` and plays with `--echo` against a + publisher that solicits feedback. It asserts the encoder's frame count and + that the player's newest arrival advances. - Decided 2026-10-08: no E2EE refusal here. The CLI has no encrypted broadcast yet, so [Encrypted native CLI](/quest/m2/e2ee/cli.md) refuses `--stats` and `--echo` when it adds one: they would publish rendition IDs and per-track @@ -54,5 +41,5 @@ ## Related -- [TS health stats](/quest/m2/ts-health-stats.md) - adds TS-specific counters after this generic reporting plumbing lands +- [TS health stats](/quest/m2/ts-health-stats.md) - adds TS-specific counters to the reports - [Encrypted native CLI](/quest/m2/e2ee/cli.md) - refuses stats and feedback on an encrypted broadcast diff --git a/quest/m3/msf-publish-tracks.md b/quest/m3/msf-publish-tracks.md index 17ff695c5a3..7b31c2fdc43 100644 --- a/quest/m3/msf-publish-tracks.md +++ b/quest/m3/msf-publish-tracks.md @@ -11,8 +11,8 @@ consumer asks. Raised in [#5360](https://github.com/moq-dev/moq/issues/5360). Until then the [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 +the MSF catalog. `moq export --echo` refuses `--catalog-format msf`, and the +[Rust reporters](/quest/m2/stats/rust.md) quest's `moq play --echo` will (decided 2026-10-12). Open: how the entry names the hang payload, since MSF fixes its packaging diff --git a/rs/moq-cli/Cargo.toml b/rs/moq-cli/Cargo.toml index 14401ce2245..72f116c6710 100644 --- a/rs/moq-cli/Cargo.toml +++ b/rs/moq-cli/Cargo.toml @@ -108,6 +108,8 @@ moq-audio = { workspace = true, optional = true, features = ["aac"] } moq-auth = { workspace = true, features = ["serve"] } # `server` enables the HTTP egress server for `moq export hls`; the importer is always available. moq-hls = { workspace = true, features = ["server"] } +# Writes and reads the stats and feedback snapshots. +moq-json = { workspace = true } # Names the MSF catalog track `export archive` records. moq-msf = { workspace = true } moq-mux = { workspace = true } @@ -139,8 +141,6 @@ winit = { version = "0.30.13", optional = true } sd-notify = { workspace = true } [dev-dependencies] -# Decodes the timeline the archive round-trip test checks. -moq-json = { path = "../moq-json" } # `test_relay` stands up a real relay for the `fetch` and `announced` tests to read through. moq-relay = { path = "../moq-relay", default-features = false, features = ["test-support"] } # Writes the PAT and PMTs, CRCs included, of the two-program input `import ts --program all` splits. diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index d89b157ba2b..e295e436819 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -339,9 +339,38 @@ impl Invocation { } for command in &self.stages { + if let Command::Import(import) = command + && import.stats + && let Some(reason) = import.source.stats_refusal() + { + anyhow::bail!("--stats names a track in the catalog the import builds, but {reason}"); + } let Command::Export(export) = command else { continue; }; + if let Some(sink) = export.sink.report() { + anyhow::ensure!( + export.catalog_format.is_none(), + "`export {sink}` reads no media, so it takes no --catalog-format" + ); + anyhow::ensure!( + export.echo.is_none(), + "`export {sink}` reads no media to report on; remove --echo" + ); + } + if export.echo.is_some() { + if export.sink.stdout().is_none() + && let Some(sink) = export.sink.ignores_selection() + { + anyhow::bail!( + "--echo reports what a stdout container export reads, and `export {sink}` is not one" + ); + } + anyhow::ensure!( + !matches!(export.catalog_format, Some(CatalogFormatArg::Msf)), + "--echo follows the hang catalog's `echo` section, which the MSF catalog does not carry" + ); + } if let Some(stdout) = export.sink.stdout() { anyhow::ensure!( stdout.linger.is_zero() || matches!(stdout.format, SubscribeFormat::Ts), @@ -745,6 +774,11 @@ pub struct Import { #[usage(long = "latency-max", hide = true)] latency_max: Option, + /// Name a stats track in the catalog and fill it with what each track has written, about + /// once a second while subscribed. The stdin imports only. + #[usage(long)] + pub stats: bool, + /// The single source feeding the Origin. #[usage(subcommand)] pub source: ImportSource, @@ -827,6 +861,22 @@ impl ImportSource { Self::Ts(_) | Self::Fmp4 | Self::Flv | Self::Hls(_) => None, } } + + /// Why `--stats` can't report this source, or `None` when it can: the CLI builds the one + /// catalog of a stdin import, whose tracks `moq-mux` counts. + fn stats_refusal(&self) -> Option<&'static str> { + match self { + Self::Ts(args) if args.program == Some(TsProgram::All) => { + Some("`import ts --program all` builds a catalog per program") + } + Self::Hls(_) | Self::Rtmp(_) | Self::Srt(_) | Self::Rtc(_) | Self::Archive(_) => { + Some("the gateway and archive imports build their own catalogs") + } + #[cfg(feature = "capture")] + Self::Capture(_) => Some("`import capture` encodes, and its encoders do not count yet"), + _ => None, + } + } } /// The MPEG-TS stdin container: which programs of a multiplex to publish. @@ -930,6 +980,12 @@ pub struct Export { #[usage(flatten)] pub select: crate::subscribe::SelectArgs, + /// Publish feedback where the broadcast's catalog solicits it: a `.echo` broadcast + /// under the catalog's `echo` path, reporting what each rendition this export reads has + /// received. One path segment. The stdout container exports only. + #[usage(long)] + pub echo: Option, + /// The single sink draining the Origin. #[usage(subcommand)] pub sink: ExportSink, @@ -981,6 +1037,10 @@ pub enum ExportSink { Rtc(crate::rtc::Args), /// Record the broadcast into an object store until it ends. Archive(crate::archive::ExportArgs), + /// Write the broadcast's stats snapshots to stdout as JSON lines. + Stats, + /// Write the feedback of every `.echo` broadcast under `--broadcast` to stdout as JSON lines. + Echo, } impl ExportSink { @@ -994,12 +1054,23 @@ impl ExportSink { Self::Srt(_) => "srt", Self::Rtc(_) => "rtc", Self::Archive(_) => "archive", + Self::Stats => "stats", + Self::Echo => "echo", }) } - /// Whether this sink writes to stdout (the container formats). + /// The report sink's name: these read no media, so they skip catalog discovery. + fn report(&self) -> Option<&'static str> { + match self { + Self::Stats => Some("stats"), + Self::Echo => Some("echo"), + _ => None, + } + } + + /// Whether this sink writes to stdout (the container formats and the report sinks). pub fn is_stdout(&self) -> bool { - self.stdout().is_some() + self.stdout().is_some() || self.report().is_some() } /// The stdout container format and its options, when this sink writes to @@ -1395,6 +1466,71 @@ mod tests { } } + /// `--stats` goes where the CLI builds the one catalog it names a track in. + #[test] + fn stats_needs_a_catalog_the_cli_builds() { + let import = |source: &[&str]| { + let argv = [ + "moq", + "--connect", + "http://relay", + "--broadcast", + "room", + "import", + "--stats", + ]; + Invocation::try_parse_from(argv.iter().chain(source).copied()).unwrap() + }; + for source in [ + ["ts"].as_slice(), + &["fmp4"], + &["flv"], + &["avc3"], + &["ts", "--program", "1"], + ] { + import(source).validate().unwrap(); + } + for source in [ + ["ts", "--program", "all"].as_slice(), + &["hls", "http://origin/master.m3u8"], + &["rtmp", "--listen", "127.0.0.1:1935"], + ] { + let err = import(source).validate().unwrap_err().to_string(); + assert!(err.contains("--stats"), "{source:?}: {err}"); + } + } + + /// `--echo` reports what a stdout container export reads from a hang catalog, under one + /// path segment, and the report sinks read no media to report on. + #[test] + fn echo_needs_a_container_export_and_a_hang_catalog() { + for sink in ["ts", "fmp4", "mkv", "flv", "h264", "h265"] { + export(&["--echo", "alice", sink]).unwrap().validate().unwrap(); + } + for sink in [["hls", "--listen", "127.0.0.1:8080"].as_slice(), &["stats"], &["echo"]] { + let err = export(&[["--echo", "alice"].as_slice(), sink].concat()) + .unwrap() + .validate() + .unwrap_err(); + assert!(err.to_string().contains("--echo"), "{sink:?}: {err}"); + } + let err = export(&["--echo", "alice", "--catalog-format", "msf", "fmp4"]) + .unwrap() + .validate() + .unwrap_err(); + assert!(err.to_string().contains("MSF"), "{err}"); + for name in ["", "a/b", ".", "..", "*"] { + assert!(export(&["--echo", name, "fmp4"]).is_err(), "`{name}` must be refused"); + } + for sink in ["stats", "echo"] { + let err = export(&["--catalog-format", "hang", sink]) + .unwrap() + .validate() + .unwrap_err(); + assert!(err.to_string().contains("--catalog-format"), "{sink}: {err}"); + } + } + #[test] fn import_ts_takes_a_program_number_or_all() { // `None` when the command line is refused. diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index e6d53e2f90f..d8787c396a6 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -17,6 +17,7 @@ mod hls; mod moq; mod play; mod publish; +mod report; mod rtc; mod rtmp; mod srt; @@ -387,7 +388,7 @@ async fn run() -> anyhow::Result<()> { /// Which directions the stages need on the shared MoQ attachment. #[derive(Clone, Copy, Default)] pub struct Directions { - /// Any `import`: the Origin is published outward. + /// Any `import`, or an `export --echo`: the Origin is published outward. pub publish: bool, /// Any `export` or `play`: the Origin is filled from the network. pub consume: bool, @@ -396,8 +397,13 @@ pub struct Directions { impl Directions { /// The union of what the stages need, so one attachment serves them all. fn of(stages: &[Command]) -> Self { + let publishes = |stage: &Command| match stage { + Command::Import(_) => true, + Command::Export(export) => export.echo.is_some(), + _ => false, + }; Self { - publish: stages.iter().any(|stage| matches!(stage, Command::Import(_))), + publish: stages.iter().any(publishes), consume: stages.iter().any(|stage| matches!(stage, Command::Export(_))), } } @@ -563,9 +569,16 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res if import.source.stdin_format().is_some() { claim("stdin", &mut stdin, &name)?; } - if let Some(publish) = - spawn_import(&origin, import, name, epoch.clone(), bandwidth.clone(), &mut tasks)? - { + let monitor = connection.as_ref().map(moq_tokio::Connection::monitor); + if let Some(publish) = spawn_import( + &origin, + import, + name, + epoch.clone(), + bandwidth.clone(), + monitor, + &mut tasks, + )? { locals.push(publish); } } @@ -573,7 +586,8 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res if export.sink.is_stdout() { claim("stdout", &mut stdout, &name)?; } - spawn_export(&origin, export, name, &mut tasks)?; + let monitor = connection.as_ref().map(moq_tokio::Connection::monitor); + spawn_export(&origin, export, name, monitor, &mut tasks)?; } other => unreachable!("`{}` is not a stage", other.name()), } @@ -644,13 +658,15 @@ fn display_name(name: &str) -> &str { /// `ts --program all` mint their own per connection or program. /// /// Returns the pipeline that has to run on the caller's thread, for the sources -/// that have one (the stdin containers and capture). +/// that have one (the stdin containers and capture). `connection` supplies a +/// `--stats` report's transport measurements. fn spawn_import( origin: &moq_net::origin::Producer, import: Import, name: String, epoch: moq_net::Epoch, bandwidth: moq_net::bandwidth::Allocator, + connection: Option, tasks: &mut JoinSet>, ) -> anyhow::Result> { if let ImportSource::Rtc(rtc) = &import.source @@ -687,6 +703,11 @@ fn spawn_import( let broadcast = origin.create_broadcast(&name).context("failed to create broadcast")?; Publish::new(broadcast, &format, config)? }; + let publish = if import.stats { + publish.with_stats(connection)? + } else { + publish + }; // Media stays servable at EOF until it goes stale, the same max-age relays retain it for. let mut publish = publish.linger(max_age.unwrap_or(hang::container::MAX_AGE)); publish.announce(epoch)?; @@ -752,10 +773,12 @@ fn spawn_import( } /// Route the shared Origin OUT to one stage's sink, filling it from the MoQ network. +/// `connection` supplies an `--echo` report's transport measurements. fn spawn_export( origin: &moq_net::origin::Producer, export: Export, name: String, + connection: Option, tasks: &mut JoinSet>, ) -> anyhow::Result<()> { if let ExportSink::Rtc(rtc) = &export.sink @@ -777,7 +800,8 @@ fn spawn_export( select: export.select, }; let consumer = origin.consume(); - tasks.spawn(async move { run_stdout(consumer, name, args).await }); + let echo = export.echo.map(|echo| (origin.clone(), echo, connection)); + tasks.spawn(async move { run_stdout(consumer, name, args, echo).await }); } else { match export.sink { ExportSink::Hls(args) => { @@ -826,6 +850,13 @@ fn spawn_export( .unwrap_or_default(); tasks.spawn(archive::export(origin.consume(), name, format, args)); } + ExportSink::Stats => { + let name = require_broadcast(name, "export stats")?; + tasks.spawn(report::print_stats(origin.consume(), name)); + } + ExportSink::Echo => { + tasks.spawn(report::print_echo(origin.consume(), name)); + } _ => unreachable!("container formats are handled by stdout_format above"), } } @@ -833,8 +864,21 @@ fn spawn_export( Ok(()) } -/// Subscribe to `name` from the Origin and write it to stdout. -async fn run_stdout(consumer: moq_net::origin::Consumer, name: String, args: SubscribeArgs) -> anyhow::Result<()> { +/// Where an `export --echo` publishes its feedback, under what name, and over which connection. +type EchoTarget = ( + moq_net::origin::Producer, + report::EchoName, + Option, +); + +/// Subscribe to `name` from the Origin and write it to stdout, publishing feedback beside it +/// where the catalog solicits it when `echo` is set. +async fn run_stdout( + consumer: moq_net::origin::Consumer, + name: String, + args: SubscribeArgs, + echo: Option, +) -> anyhow::Result<()> { let catalog = args.catalog_format(&name); // Confirm the broadcast is reachable and wait for it to be announced; `Subscribe` then @@ -845,7 +889,17 @@ async fn run_stdout(consumer: moq_net::origin::Consumer, name: String, args: Sub .await .ok_or_else(|| anyhow::anyhow!("origin closed before broadcast `{name}` was announced"))?; - Subscribe::new(consumer, &name, catalog, args).await?.run().await + let subscribe = Subscribe::new(consumer, &name, catalog, args).await?; + let Some((origin, echo, connection)) = echo else { + return subscribe.run().await; + }; + let mut echo = report::Echo::new(origin, subscribe.source().clone(), catalog, echo, connection); + let result = tokio::select! { + result = subscribe.run() => result, + never = echo.run() => match never {}, + }; + echo.finish(); + result } /// Run until an endpoint ends, keeping all tasks alive while a signalled listener drains. diff --git a/rs/moq-cli/src/publish.rs b/rs/moq-cli/src/publish.rs index e937997e532..c67f511b06b 100644 --- a/rs/moq-cli/src/publish.rs +++ b/rs/moq-cli/src/publish.rs @@ -303,6 +303,8 @@ pub struct Publish { linger: std::time::Duration, // A stdin import's route, announced once its catalog lists the tracks; see [`Self::announce`]. route: Option, + // The `--stats` report, written beside the decode loop; see [`Self::with_stats`]. + stats: Option, } /// A broadcast's announcement, held until its catalog first goes out. @@ -362,6 +364,7 @@ impl Publish { broadcast: Some(broadcast), linger: Default::default(), route: None, + stats: None, }); } @@ -404,6 +407,7 @@ impl Publish { broadcast: Some(broadcast), linger: Default::default(), route: None, + stats: None, }) } @@ -422,6 +426,7 @@ impl Publish { broadcast: None, linger: Default::default(), route: None, + stats: None, } } @@ -462,9 +467,27 @@ impl Publish { broadcast: Some(broadcast), linger: Default::default(), route: None, + stats: None, }) } + /// Name a stats track in the catalog and write it beside the decode loop: what each track + /// has written, and the transport of `connection` when there is one. + /// + /// Call it before the importer declares its tracks, so the section rides the first catalog. + pub fn with_stats(mut self, connection: Option) -> anyhow::Result { + let (Some(broadcast), Source::Stream { catalog, .. }) = (self.broadcast.as_mut(), &self.source) else { + anyhow::bail!("--stats reports a stdin import's one catalog"); + }; + let stats = match catalog { + PublishCatalog::Media(catalog) => crate::report::Stats::new(broadcast, catalog, connection)?, + PublishCatalog::Ts(catalog) => crate::report::Stats::new(broadcast, catalog, connection)?, + PublishCatalog::TsPrograms => anyhow::bail!("`import ts --program all` builds a catalog per program"), + }; + self.stats = Some(stats); + Ok(self) + } + /// Keep a finished stdin import up for `linger` after EOF, so a subscriber that learns the /// final catalog late can still request and drain its tracks. Pass the media max-age: the /// content is servable until it goes stale. SIGINT or SIGTERM ends it early. @@ -496,7 +519,7 @@ impl Publish { (Some(broadcast), Some(route)) => Some(Pending::new(broadcast.clone(), route)?), _ => None, }; - decode(decoder, catalog, tokio::io::stdin(), self.linger, pending).await + decode(decoder, catalog, tokio::io::stdin(), self.linger, pending, self.stats).await } #[cfg(feature = "capture")] Source::Capture { @@ -556,12 +579,16 @@ impl Publish { /// renditions retire from the catalog as the decoder drops, which a subscriber would read /// as removed tracks, and a catalog dropped unfinished reads as a publisher that vanished. /// The finished tracks then stay servable for `linger`. +/// +/// `stats` reports beside the read loop, so a stalled input keeps reporting, and ends with its +/// final counts when the media tracks do. async fn decode( mut decoder: PublishDecoder, mut catalog: PublishCatalog, mut input: impl tokio::io::AsyncRead + Unpin, linger: std::time::Duration, mut pending: Option, + mut stats: Option, ) -> anyhow::Result<()> { // An empty BytesMut grows to only 64 bytes and clearing it keeps that capacity. // Stdin dispatches every read to a blocking thread, so those tiny reads can @@ -577,7 +604,7 @@ async fn decode( // Run the read/decode loop so an error surfaces here rather than // dropping the decoder (and its tracks) with a bare Error::Dropped. - let result: anyhow::Result<()> = async { + let read = async { loop { buffer.clear(); let n = tokio::io::AsyncReadExt::read_buf(&mut input, &mut buffer).await?; @@ -598,8 +625,14 @@ async fn decode( } } } - } - .await; + }; + let result: anyhow::Result<()> = match stats.as_mut() { + Some(stats) => tokio::select! { + result = read => result, + never = stats.run() => match never {}, + }, + None => read.await, + }; // Flush on a clean EOF; on any error (read, decode, or the flush // itself) abort with the real cause so subscribers see it instead of @@ -618,10 +651,13 @@ async fn decode( } match outcome { Ok(()) => { + if let Some(stats) = &mut stats { + stats.finish(); + } catalog.finish()?; - // The decoder and catalog own the tracks, so holding them keeps the finished - // broadcast servable: a subscriber that saw the final catalog only at EOF still has - // to request its media. + // The decoder, catalog, and stats report own the tracks, so holding them keeps the + // finished broadcast servable: a subscriber that saw the final catalog only at EOF + // still has to request its media. tracing::info!( ?linger, "input ended; serving the finished broadcast until its media is stale" @@ -1375,7 +1411,7 @@ pub(crate) mod tests { remaining: BBB, reads: 0, }; - decode(decoder, catalog, &mut input, Duration::ZERO, None) + decode(decoder, catalog, &mut input, Duration::ZERO, None, None) .await .unwrap(); assert_eq!( @@ -1410,7 +1446,9 @@ pub(crate) mod tests { panic!("expected a stream source"); }; let start = std::time::Instant::now(); - decode(decoder, catalog, input, Duration::ZERO, None).await.unwrap(); + decode(decoder, catalog, input, Duration::ZERO, None, None) + .await + .unwrap(); let elapsed = start.elapsed(); let bytes = INPUT.len() * copies; println!( @@ -1433,7 +1471,7 @@ pub(crate) mod tests { let Source::Stream { decoder, catalog } = publish.source else { panic!("expected a stream source"); }; - decode(decoder, catalog, BBB, Duration::ZERO, None).await.unwrap(); + decode(decoder, catalog, BBB, Duration::ZERO, None, None).await.unwrap(); drop(publish.broadcast); let mut last = None; @@ -1466,7 +1504,7 @@ pub(crate) mod tests { panic!("expected a stream source"); }; let started = tokio::time::Instant::now(); - let mut import = std::pin::pin!(decode(decoder, catalog, BBB, LINGER, None)); + let mut import = std::pin::pin!(decode(decoder, catalog, BBB, LINGER, None, None)); let name = tokio::select! { res = &mut import => panic!("the import returned before its linger: {res:?}"), name = async { @@ -1520,7 +1558,7 @@ pub(crate) mod tests { let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); let pending = Pending::new(broadcast, route).unwrap(); let (mut stdin, input_end) = tokio::io::duplex(input.len()); - let mut import = std::pin::pin!(decode(decoder, catalog, input_end, Duration::ZERO, Some(pending))); + let mut import = std::pin::pin!(decode(decoder, catalog, input_end, Duration::ZERO, Some(pending), None)); tokio::select! { res = &mut import => panic!("the import returned before its input ended: {res:?}"), @@ -1584,7 +1622,7 @@ pub(crate) mod tests { let Source::Stream { decoder, catalog } = publish.source else { panic!("stream source"); }; - let err = decode(decoder, catalog, Broken, Duration::ZERO, None) + let err = decode(decoder, catalog, Broken, Duration::ZERO, None, None) .await .unwrap_err(); assert_eq!(err.to_string(), "stdin failed"); @@ -1611,7 +1649,7 @@ pub(crate) mod tests { panic!("expected a stream source"); }; let pending = Pending::new(broadcast, Default::default()).unwrap(); - let err = decode(decoder, catalog, &[][..], Duration::ZERO, Some(pending)) + let err = decode(decoder, catalog, &[][..], Duration::ZERO, Some(pending), None) .await .unwrap_err(); assert!(err.to_string().contains("declared"), "{err:#}"); diff --git a/rs/moq-cli/src/report.rs b/rs/moq-cli/src/report.rs new file mode 100644 index 00000000000..7324cec862d --- /dev/null +++ b/rs/moq-cli/src/report.rs @@ -0,0 +1,443 @@ +//! Media reports: the stats track `import --stats` names in its catalog, the feedback +//! `export --echo` publishes where a catalog solicits it, and the `export stats` and +//! `export echo` sinks that print them. +//! +//! Each report is one snapshot track and its compressed `.z` sibling, written about once a +//! second while someone subscribes. See the hang draft's media statistics section for the wire. + +use std::collections::{BTreeMap, HashMap}; +use std::convert::Infallible; +use std::time::Duration; + +use anyhow::Context; +use hang::moq_net; +use moq_mux::catalog::{CatalogFormat, Stream}; +use moq_tokio::connection::Monitor; +use tokio::io::AsyncWriteExt; + +/// How often a report is written: about once a second, as the hang draft asks of a producer. +const INTERVAL: Duration = Duration::from_secs(1); + +/// The publisher snapshot track an import names in its catalog's `stats` section. +pub const STATS_TRACK: &str = "stats.json"; + +/// The suffix naming a snapshot track's compressed sibling. +const COMPRESSED: &str = ".z"; + +/// Below every media track, so a report never delays the media it describes. +const PRIORITY: u8 = 0; + +/// A snapshot track and its compressed `.z` sibling, written together. +struct Pair { + plain: moq_json::snapshot::Producer, + compressed: moq_json::snapshot::Producer, +} + +impl Pair { + /// Create `name` and `name.z` on `broadcast`. + fn create(broadcast: &mut moq_net::broadcast::Producer, name: &str) -> anyhow::Result { + let info = moq_net::track::Info::default().with_priority(PRIORITY); + let plain = broadcast.create_track(name, info.clone())?; + let compressed = broadcast.create_track(format!("{name}{COMPRESSED}").as_str(), info)?; + let mut deflate = moq_json::snapshot::Config::default(); + deflate.compression = moq_json::Compression::Deflate; + Ok(Self { + plain: moq_json::snapshot::Producer::new(plain, moq_json::snapshot::Config::default().with_delta_ratio(0)), + compressed: moq_json::snapshot::Producer::new(compressed, deflate), + }) + } + + /// Publish `value` on both tracks, unless nobody subscribes to either. + fn update(&mut self, value: &T) { + if self.plain.demand().is_used() || self.compressed.demand().is_used() { + self.write(value); + } + } + + /// Publish `value` on both tracks. + fn write(&mut self, value: &T) { + for track in [&mut self.plain, &mut self.compressed] { + if let Err(err) = track.update(value) { + tracing::debug!(%err, "failed to write a report"); + } + } + } + + /// End both tracks cleanly, keeping the last snapshot. + fn finish(&mut self) { + for track in [&mut self.plain, &mut self.compressed] { + if let Err(err) = track.finish() { + tracing::debug!(%err, "failed to finish a report"); + } + } + } +} + +/// Which way the reported media crosses the connection, which picks its rate estimate. +#[derive(Clone, Copy)] +enum Direction { + Send, + Receive, +} + +/// The connection's transport measurements now, or none without a dialed connection or while +/// it reconnects. +fn transport(connection: Option<&Monitor>, direction: Direction) -> hang::stats::Transport { + let mut transport = hang::stats::Transport::default(); + if let Some(stats) = connection.and_then(Monitor::stats) { + let rate = match direction { + Direction::Send => stats.estimated_send_rate, + Direction::Receive => stats.estimated_recv_rate, + }; + transport.rtt = stats.rtt; + transport.rate = rate.map(moq_net::bandwidth::Rate::as_bps); + transport.lost_bytes = stats.bytes_lost; + transport.lost_packets = stats.packets_lost; + } + transport +} + +/// An import's stats track: what each of its tracks has written, from its catalog's counters, +/// and the connection's transport. +pub struct Stats { + tracks: Pair, + renditions: Box BTreeMap + Send>, + connection: Option, +} + +impl Stats { + /// Create the stats track on `broadcast` and name it in `catalog`'s `stats` section. + /// + /// Call it while an importer still holds its reservation, so the section rides the first + /// catalog rather than announcing a catalog with no tracks. + pub fn new( + broadcast: &mut moq_net::broadcast::Producer, + catalog: &moq_mux::catalog::Producer, + connection: Option, + ) -> anyhow::Result { + let tracks = Pair::create(broadcast, STATS_TRACK)?; + let mut catalog = catalog.clone(); + let mut guard = catalog.modify()?; + guard.stats = Some(hang::catalog::Stats::new(STATS_TRACK)); + guard.commit()?; + Ok(Self { + tracks, + renditions: Box::new(move || catalog.stats()), + connection, + }) + } + + fn snapshot(&self) -> hang::stats::Snapshot { + let mut snapshot = hang::stats::Snapshot::default(); + snapshot.transport = transport(self.connection.as_ref(), Direction::Send); + snapshot.renditions = (self.renditions)(); + snapshot + } + + fn write(&mut self) { + let snapshot = self.snapshot(); + self.tracks.update(&snapshot); + } + + /// Write a snapshot every interval, for as long as the import runs beside it. + pub async fn run(&mut self) -> Infallible { + let mut interval = tokio::time::interval(INTERVAL); + loop { + interval.tick().await; + self.write(); + } + } + + /// Write the final counts, subscribed or not, and end the track. Hold on to it while the + /// import lingers, so a late subscriber still reads them. + pub fn finish(&mut self) { + let last = self.snapshot(); + self.tracks.write(&last); + self.tracks.finish(); + } +} + +/// An `export --echo` name: the one path segment a feedback broadcast is published at, before +/// the `.echo` suffix. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct EchoName(String); + +impl std::str::FromStr for EchoName { + type Err = String; + + fn from_str(name: &str) -> Result { + if name.is_empty() || name.contains('/') || matches!(name, "." | ".." | "*" | "**") { + return Err(format!("`{name}` is not a single path segment")); + } + Ok(Self(name.to_string())) + } +} + +/// The feedback broadcast in force: where it is published and its snapshot tracks. +struct Feedback { + path: moq_net::PathOwned, + // Held for the route: dropping it retracts the announcement. + _broadcast: moq_net::broadcast::Producer, + tracks: Pair, +} + +/// An export's solicited feedback: once its catalog carries an `echo` section, a +/// `/.echo` broadcast whose `feedback.json` reports what each rendition the export +/// reads has received, and the connection's transport. Each catalog update reconciles it: a +/// removed section or a changed path retracts the broadcast it replaces. It follows the catalog +/// of the instance the export first reads, and ends when the export does. +pub struct Echo { + origin: moq_net::origin::Producer, + source: moq_mux::Source, + format: CatalogFormat, + name: EchoName, + connection: Option, + feedback: Option, +} + +impl Echo { + /// Feedback for the export reading `source`, published into `origin`. + pub fn new( + origin: moq_net::origin::Producer, + source: moq_mux::Source, + format: CatalogFormat, + name: EchoName, + connection: Option, + ) -> Self { + Self { + origin, + source, + format, + name, + connection, + feedback: None, + } + } + + /// Follow the catalog's `echo` section and write a snapshot every interval, for as long as the + /// export runs beside it. + pub async fn run(&mut self) -> Infallible { + let mut catalog = match self.source.catalog::<()>(self.format).await { + Ok(catalog) => Some(catalog), + Err(err) => { + tracing::warn!(%err, "no catalog to read a feedback solicitation from"); + None + } + }; + let mut interval = tokio::time::interval(INTERVAL); + loop { + tokio::select! { + update = next(&mut catalog) => match update { + Ok(Some(update)) => self.reconcile(update.echo.as_ref()), + // The export still writes what it has read once the catalog ends, so the + // feedback in force stays up until the export finishes it. + Ok(None) => catalog = None, + Err(err) => { + tracing::warn!(%err, "stopped reading the catalog's feedback solicitation"); + catalog = None; + } + }, + _ = interval.tick() => self.write(), + } + } + } + + /// Publish at the path `echo` resolves to, retracting a broadcast at any other. + fn reconcile(&mut self, echo: Option<&hang::catalog::Echo>) { + let target = echo.and_then(|echo| { + let prefix = self.source.resolve_reference(Some(&echo.path)); + if prefix.is_none() { + tracing::warn!(path = %echo.path.as_str(), "the catalog solicits feedback above the connection root"); + } + Some(prefix?.join(format!("{}{}", self.name.0, hang::echo::SUFFIX).as_str())) + }); + if self.feedback.as_ref().map(|feedback| &feedback.path) == target.as_ref() { + return; + } + self.retire(); + let Some(path) = target else { + return; + }; + match self.publish(&path) { + Ok(feedback) => { + tracing::info!(path = %path.as_str(), "publishing feedback"); + self.feedback = Some(feedback); + self.write(); + } + Err(err) => tracing::warn!(%err, path = %path.as_str(), "failed to publish feedback"), + } + } + + fn publish(&self, path: &moq_net::PathOwned) -> anyhow::Result { + let mut broadcast = self.origin.create_broadcast(path)?; + let tracks = Pair::create(&mut broadcast, hang::echo::TRACK)?; + broadcast.announce(moq_net::origin::Route::default())?; + Ok(Feedback { + path: path.clone(), + _broadcast: broadcast, + tracks, + }) + } + + fn write(&mut self) { + let Some(feedback) = &mut self.feedback else { + return; + }; + let mut snapshot = hang::echo::Snapshot::default(); + snapshot.transport = transport(self.connection.as_ref(), Direction::Receive); + snapshot.renditions = self.source.stats(); + feedback.tracks.update(&snapshot); + } + + /// End the feedback broadcast in force, if any. + fn retire(&mut self) { + if let Some(mut feedback) = self.feedback.take() { + feedback.tracks.finish(); + } + } + + /// Write the final counts and retract the feedback broadcast. + pub fn finish(mut self) { + self.write(); + self.retire(); + } +} + +/// The next catalog update, or never once there is no catalog to read. +async fn next(catalog: &mut Option) -> moq_mux::Result>> { + match catalog { + Some(catalog) => catalog.next().await, + None => std::future::pending().await, + } +} + +/// The next snapshot of a report track, or never while there is none. +async fn snapshot( + track: &mut Option>, +) -> moq_json::Result> { + match track { + Some(track) => track.next().await, + None => std::future::pending().await, + } +} + +/// `export stats`: write the broadcast's stats snapshots to stdout as JSON lines, following the +/// track its catalog's `stats` section names, until the catalog and that track have both ended. +pub async fn print_stats(origin: moq_net::origin::Consumer, path: String) -> anyhow::Result<()> { + let mut stdout = tokio::io::stdout(); + origin + .routed(&path) + .await + .ok_or_else(|| anyhow::anyhow!("origin closed before broadcast `{path}` was announced"))?; + let broadcast = origin + .request_broadcast(path.as_str(), None) + .await + .with_context(|| format!("failed to resolve broadcast `{path}`"))?; + let source = moq_mux::Source::new(origin, broadcast.clone()); + let mut catalog = Some(source.catalog::<()>(CatalogFormat::Hang).await?); + + let mut named = None; + let mut track = None; + loop { + tokio::select! { + update = next(&mut catalog) => { + // A finished import ends its stats track just before its catalog: read it out. + let Some(update) = update? else { + if track.is_none() { + return Ok(()); + } + catalog = None; + continue; + }; + let stats = update.stats.map(|stats| stats.track); + if stats != named { + track = match &stats { + Some(name) => { + let subscriber = broadcast.track(name)?.subscribe(None).await?; + Some(moq_json::snapshot::Consumer::new(subscriber, Default::default())) + } + None => None, + }; + named = stats; + } + } + value = snapshot(&mut track) => match value? { + Some(value) => { + stdout.write_all(format!("{value}\n").as_bytes()).await?; + stdout.flush().await?; + } + None if catalog.is_none() => return Ok(()), + None => track = None, + }, + } + } +} + +/// One line of `export echo`: the feedback broadcast and its latest snapshot. +#[derive(serde::Serialize)] +struct EchoLine<'a> { + broadcast: &'a str, + feedback: serde_json::Value, +} + +/// `export echo`: write the feedback of every `.echo` broadcast under `prefix` to stdout as JSON +/// lines, following them as they start and end. +pub async fn print_echo(origin: moq_net::origin::Consumer, prefix: String) -> anyhow::Result<()> { + let mut stdout = tokio::io::stdout(); + let prefix = moq_net::Path::new(&prefix).to_owned(); + let mut announced = origin.announced(); + let (lines, mut written) = tokio::sync::mpsc::unbounded_channel::(); + let mut readers = HashMap::new(); + loop { + tokio::select! { + event = announced.next() => { + let event = event.context("announcements ended")?; + let (announce, active) = match event { + moq_net::announce::Event::Start(announce) => (announce, true), + moq_net::announce::Event::End(announce) => (announce, false), + _ => continue, + }; + let path = announce.prefix; + if !path.has_prefix(&prefix) || !hang::echo::is_echo(&path) { + continue; + } + if active { + let reader = tokio::spawn(read_echo(origin.clone(), path.clone(), lines.clone())); + readers.insert(path, reader); + } else if let Some(reader) = readers.remove(&path) { + reader.abort(); + } + } + Some(line) = written.recv() => { + stdout.write_all(line.as_bytes()).await?; + stdout.flush().await?; + } + } + } +} + +/// Send a line for each feedback snapshot the broadcast at `path` publishes, until it ends. +async fn read_echo( + origin: moq_net::origin::Consumer, + path: moq_net::PathOwned, + lines: tokio::sync::mpsc::UnboundedSender, +) { + let read = async { + let broadcast = origin.request_broadcast(&path, None).await?; + let subscriber = broadcast.track(hang::echo::TRACK)?.subscribe(None).await?; + let mut track = moq_json::snapshot::Consumer::::new(subscriber, Default::default()); + while let Some(feedback) = track.next().await? { + let line = EchoLine { + broadcast: path.as_str(), + feedback, + }; + if lines.send(format!("{}\n", serde_json::to_string(&line)?)).is_err() { + break; + } + } + anyhow::Ok(()) + }; + if let Err(err) = read.await { + tracing::debug!(%err, path = %path.as_str(), "stopped reading feedback"); + } +} diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index 4d0fcd780ee..4b72048eac8 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -279,6 +279,11 @@ impl Subscribe { }) } + /// The resolved broadcast, whose [`stats`](moq_mux::Source::stats) count what the export reads. + pub fn source(&self) -> &moq_mux::Source { + &self.source + } + /// Build the catalog stream, narrowed by the rendition selection flags. The /// catalog source honors the requested format (e.g. compressed `HangZ` or `Msf`). async fn stream(&self) -> anyhow::Result> { diff --git a/rs/moq-cli/tests/stats.rs b/rs/moq-cli/tests/stats.rs new file mode 100644 index 00000000000..37e33067b56 --- /dev/null +++ b/rs/moq-cli/tests/stats.rs @@ -0,0 +1,271 @@ +//! `moq import --stats` and `moq export --echo` over a real relay, read back through the +//! `export stats` and `export echo` sinks: each report counts what its side carried. +#![cfg(unix)] + +use std::collections::BTreeMap; +use std::process::Stdio; +use std::time::Duration; + +use hang::moq_net; +use moq_mux::container::ts; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines}; +use tokio::process::{Child, ChildStdout}; + +const TIMEOUT: Duration = Duration::from_secs(30); +/// Five seconds of HEVC and Opus at a constant 400 kb/s. +const CLIP: &[u8] = include_bytes!("../../moq-mux/src/container/ts/test_data/bbb5s_cbr.ts"); +/// Bytes fed every [`PACE`], which keeps [`CLIP`] near real time. +const CHUNK: usize = 188 * 40; +const PACE: Duration = Duration::from_millis(150); +/// Past the two seconds of PCR a live TS catalog waits for, so the broadcast is up by then. +const HALF: usize = CLIP.len() / 2; + +/// Feed `clip` to `write` at about real time. +async fn feed(clip: &[u8], mut write: impl AsyncFnMut(&[u8])) { + let mut pace = tokio::time::interval(PACE); + for chunk in clip.chunks(CHUNK) { + pace.tick().await; + write(chunk).await; + } +} + +struct Relay { + url: String, + fingerprint: String, + origin: moq_net::origin::Producer, +} + +async fn relay() -> Relay { + let _ = moq_tokio::crypto::install_default(); + let fixture = moq_relay::test_relay().await.expect("test relay"); + let origin = fixture.relay.cluster().origin.clone(); + let ready = fixture.relay.ready(); + tokio::spawn(fixture.relay.run()); + ready.wait().await.expect("relay ready"); + Relay { + url: fixture.url.to_string(), + fingerprint: fixture.fingerprint, + origin, + } +} + +fn moq(relay: &Relay, broadcast: &str, args: &[&str]) -> tokio::process::Command { + let mut command = tokio::process::Command::new(env!("CARGO_BIN_EXE_moq")); + // A `MOQ_*` variable in the developer's shell must not reconfigure the child. + for (name, _) in std::env::vars_os() { + if name.to_string_lossy().starts_with("MOQ_") { + command.env_remove(name); + } + } + command + .args(["--connect", &relay.url, "--connect-tls-fingerprint", &relay.fingerprint]) + .args(["--broadcast", broadcast]) + .args(args) + .kill_on_drop(true); + command +} + +/// Start `moq export ` on `broadcast`, reading the JSON lines it prints. +fn sink(relay: &Relay, broadcast: &str, sink: &str) -> (Child, Lines>) { + let mut child = moq(relay, broadcast, &["export", sink]) + .stdout(Stdio::piped()) + .spawn() + .expect("spawn the sink"); + let lines = BufReader::new(child.stdout.take().expect("stdout")).lines(); + (child, lines) +} + +/// The next line the sink prints, or `None` once it exits. +async fn next(lines: &mut Lines>) -> Option { + let line = tokio::time::timeout(TIMEOUT, lines.next_line()) + .await + .expect("the sink printed nothing") + .expect("read the sink"); + line.map(|line| serde_json::from_str(&line).unwrap_or_else(|err| panic!("`{line}` is not a report: {err}"))) +} + +async fn wait(child: &mut Child) -> std::process::ExitStatus { + tokio::time::timeout(TIMEOUT, child.wait()) + .await + .expect("moq never exited") + .expect("wait for moq") +} + +/// What a TS import of `clip` writes to each rendition, from an in-process import. +fn written(clip: &[u8]) -> BTreeMap { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let config = + moq_mux::catalog::Config::default().with_catalog(moq_mux::catalog::hang::Catalog::::default()); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config).unwrap(); + let mut import = ts::Import::new(broadcast, catalog.reserve()); + import.decode(clip).unwrap(); + import.finish().unwrap(); + catalog.stats() +} + +/// The stats an import names in its catalog count each rendition while it runs, and its last +/// report holds what the import wrote in full. +#[tokio::test] +async fn import_stats_count_what_each_rendition_wrote() { + let relay = relay().await; + let expected = written(CLIP); + assert_eq!(expected.len(), 2, "the clip carries video and audio: {expected:?}"); + + let mut import = moq(&relay, "demo", &["import", "--stats", "--max-age", "2s", "ts"]) + .stdin(Stdio::piped()) + .spawn() + .expect("spawn the import"); + let mut stdin = import.stdin.take().expect("stdin"); + let (mut sink, mut lines) = sink(&relay, "demo", "stats"); + + // Half the clip, then a report before the rest: the import reports while it runs, and a + // stalled input keeps reporting until the sink has joined. + feed(&CLIP[..HALF], async |chunk| { + stdin.write_all(chunk).await.expect("write stdin") + }) + .await; + let mut reports: Vec = vec![next(&mut lines).await.expect("a report mid-import")]; + feed(&CLIP[HALF..], async |chunk| { + stdin.write_all(chunk).await.expect("write stdin") + }) + .await; + drop(stdin); + while let Some(report) = next(&mut lines).await { + reports.push(report); + } + assert!(wait(&mut import).await.success()); + assert!(wait(&mut sink).await.success(), "the sink ends with the broadcast"); + + let last = reports.last().unwrap(); + assert_eq!(last.renditions, expected, "the last report holds everything written"); + let (first, written) = (&reports[0].renditions, &last.renditions); + assert!( + !first.is_empty() + && first + .iter() + .all(|(name, rendition)| rendition.frames < written[name].frames), + "the first report counts half the clip: {first:?}" + ); + assert!( + reports.iter().any(|report| report.transport.rtt.is_some()), + "the import reports its connection" + ); +} + +/// A sink that joins once the import has finished, while it lingers, still reads its final +/// counts. +#[tokio::test] +async fn a_late_sink_reads_the_final_stats() { + let relay = relay().await; + let expected = written(CLIP); + + let mut import = moq(&relay, "demo", &["import", "--stats", "--max-age", "10s", "ts"]) + .stdin(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn the import"); + let mut stdin = import.stdin.take().expect("stdin"); + stdin.write_all(CLIP).await.expect("write stdin"); + drop(stdin); + let mut stderr = BufReader::new(import.stderr.take().expect("stderr")).lines(); + let ended = tokio::time::timeout(TIMEOUT, async { + while let Some(line) = stderr.next_line().await.expect("read the import's stderr") { + if line.contains("input ended") { + return; + } + } + panic!("the import exited before its input ended"); + }); + ended.await.expect("the import never reached the end of its input"); + + let (mut sink, mut lines) = sink(&relay, "demo", "stats"); + let report: hang::stats::Snapshot = next(&mut lines).await.expect("the final report"); + assert_eq!(report.renditions, expected); + sink.kill().await.unwrap(); + import.kill().await.unwrap(); +} + +/// One line of `export echo`. +#[derive(serde::Deserialize)] +struct Feedback { + broadcast: String, + feedback: hang::echo::Snapshot, +} + +/// An export asked for feedback publishes it where the catalog solicits it, counting what it +/// reads as it reads it, and `export echo` finds it under that prefix. +#[tokio::test] +async fn export_echo_reports_what_the_export_reads() { + let relay = relay().await; + let expected = written(CLIP); + + // A publisher soliciting feedback under `viewers`, a sibling of its broadcast. + let mut broadcast = relay.origin.create_broadcast("demo").expect("create the broadcast"); + let config = moq_mux::catalog::Config::default() + .with_max_age(Some(Duration::from_secs(1))) + .with_catalog(moq_mux::catalog::hang::Catalog::::default()); + let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast, config).unwrap(); + let mut import = ts::Import::new(broadcast.clone(), catalog.reserve()); + let mut solicit = catalog.modify().unwrap(); + solicit.echo = Some(hang::catalog::Echo::new("viewers".to_string())); + solicit.commit().unwrap(); + broadcast.announce(moq_net::origin::Route::default()).unwrap(); + + let (mut sink, mut lines) = sink(&relay, "viewers", "echo"); + let mut export = moq(&relay, "demo", &["export", "--echo", "alice", "ts"]) + .stdout(Stdio::piped()) + .spawn() + .expect("spawn the export"); + let mut output = export.stdout.take().expect("stdout"); + let draining = tokio::spawn(async move { tokio::io::copy(&mut output, &mut tokio::io::sink()).await }); + + // Half the clip, then feedback before the rest, so the export reports while it reads. + feed(&CLIP[..HALF], async |chunk| import.decode(chunk).unwrap()).await; + let mut feedback: Vec = vec![next(&mut lines).await.expect("feedback mid-export")]; + feed(&CLIP[HALF..], async |chunk| import.decode(chunk).unwrap()).await; + import.finish().unwrap(); + catalog.finish().unwrap(); + + let status = loop { + tokio::select! { + line = next(&mut lines) => feedback.push(line.expect("the echo sink runs until stopped")), + status = export.wait() => break status.expect("wait for the export"), + } + }; + assert!(status.success(), "the export finishes cleanly, got {status}"); + // The export writes its final counts as it exits. + while let Ok(Some(line)) = tokio::time::timeout(Duration::from_secs(2), next(&mut lines)).await { + feedback.push(line); + } + draining.await.unwrap().expect("drain the export"); + sink.kill().await.unwrap(); + drop(broadcast); + + assert!( + feedback.iter().all(|line| line.broadcast == "viewers/alice.echo"), + "the feedback is published at the solicited path" + ); + let first = &feedback.first().expect("the sink printed feedback").feedback; + let last = &feedback.last().unwrap().feedback; + assert!(!last.renditions.is_empty(), "the export reports what it reads"); + for (name, read) in &last.renditions { + assert!( + read.frames > 0 && read.frames <= expected[name].frames, + "{name}: read {} of {} frames", + read.frames, + expected[name].frames + ); + let newest = read.arrival.as_ref().expect("an arrival").timestamp; + let earliest = first.renditions.get(name).and_then(|read| read.arrival.as_ref()); + assert!( + earliest.is_none_or(|earliest| earliest.timestamp < newest), + "{name}: the newest arrival advances over {} lines: {earliest:?} then {:?}", + feedback.len(), + read.arrival + ); + } + assert!( + feedback.iter().any(|line| line.feedback.transport.rtt.is_some()), + "the export reports its connection" + ); +} diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index f68fb6be486..9304eb523bd 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -250,6 +250,8 @@ pub struct Producer { /// The minimum flush lateness across this catalog's renditions, which each rendition's /// advertised `delay` is measured against. baseline: super::estimate::Baseline, + /// What each media track minted under this catalog has written. See [`stats`](Self::stats). + sent: Arc>, } // Manual Clone so a producer is cheaply clonable regardless of whether `E` is. @@ -263,6 +265,7 @@ impl Clone for Producer { max_age: self.max_age, bandwidth: self.bandwidth.clone(), baseline: self.baseline.clone(), + sent: self.sent.clone(), } } } @@ -429,6 +432,7 @@ impl Producer { max_age: config.max_age, bandwidth: config.bandwidth, baseline: Default::default(), + sent: Default::default(), }) } @@ -752,9 +756,11 @@ impl Producer { { let mut catalog = self.clone(); let recorder = catalog.enroll(track.name())?; + let sent = self.sent(rendition.name()); let mut producer = crate::container::Producer::with_rendition(track, container, rendition) .with_recorder(recorder) - .with_bandwidth(self.bandwidth.clone()); + .with_bandwidth(self.bandwidth.clone()) + .with_sent(sent); if let Some(config) = config { producer.set(config)?; } @@ -773,9 +779,27 @@ impl Producer { { let mut catalog = self.clone(); let recorder = catalog.enroll(track.name())?; + let sent = self.sent(track.name()); Ok(crate::container::Producer::new(track, container) .with_recorder(recorder) - .with_bandwidth(self.bandwidth.clone())) + .with_bandwidth(self.bandwidth.clone()) + .with_sent(sent)) + } + + /// The counters a track writing under catalog key `name` bumps, for one that writes its groups + /// by hand rather than through a [`container::Producer`](crate::container::Producer). + pub(crate) fn sent(&self, name: &str) -> Arc { + self.sent.track(name) + } + + /// What each media track minted under this catalog has written, keyed by its catalog key: the + /// rendition ID, or the track name for an `m2ts` or verbatim `mpegts` track. + /// + /// Counters are cumulative for the life of the catalog, so a track that ended keeps its final + /// counts and a later track under the same key continues them. Fold this into a + /// [`hang::stats::Snapshot`] on the stats interval. + pub fn stats(&self) -> std::collections::BTreeMap { + self.sent.sent() } /// A fresh estimator whose `delay` is measured against this catalog's other renditions. diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index 6f63d9383ea..bd2d61d49e8 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -134,6 +134,9 @@ struct Fmp4Track { /// hand, so this sits beside the estimator the same way it does on a /// [`container::Producer`](crate::container::Producer). claim: crate::catalog::Claim, + + /// The samples and sample bytes the catalog reports this track has written. + sent: std::sync::Arc, } impl Fmp4Track { @@ -339,6 +342,7 @@ impl Import { } }; + let sent = self.catalog.sent(track.name()); self.tracks.insert( track_id, Fmp4Track { @@ -354,6 +358,7 @@ impl Import { pending_sequence: None, estimator: Estimator::new(), claim: crate::catalog::Claim::new(self.catalog.bandwidth()), + sent, }, ); } @@ -777,6 +782,7 @@ impl Import { let mut min_timestamp = None; let mut max_end = None; let mut contains_keyframe = false; + let mut keyframes = 0u64; let total_samples: usize = traf.trun.iter().map(|t| t.entries.len()).sum(); let mut sample_index = 0usize; @@ -842,6 +848,7 @@ impl Import { }; contains_keyframe |= keyframe; + keyframes += u64::from(keyframe && track.kind == Kind::Video); if min_timestamp.is_none_or(|min| timestamp <= min) { min_timestamp = Some(timestamp); @@ -1004,6 +1011,7 @@ impl Import { track.group = Some(g); track.estimator.write(timestamp, fragment_len); + track.sent.add(total_samples as u64, track_mdat_data.len(), keyframes); let end = max_end.ok_or(Error::MissingTrun)?; if let Some(recorder) = track.recorder.as_mut() { recorder.end(end); diff --git a/rs/moq-mux/src/container/producer.rs b/rs/moq-mux/src/container/producer.rs index f2afb8b9991..4d74e57b606 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -97,6 +97,9 @@ pub struct Producer { /// The catalog rendition this media producer owns, when it was created through a catalog. rendition: Option>>, + + /// The counters the catalog reports for this track, when it was created through a catalog. + sent: Option>, } trait Rendition: Send { @@ -164,6 +167,7 @@ impl Producer { estimator: crate::catalog::Estimator::new(), bandwidth: None, rendition: None, + sent: None, } } } @@ -200,6 +204,7 @@ where estimator: rendition.estimator(), bandwidth: None, rendition: Some(Box::new(rendition)), + sent: None, } } @@ -370,6 +375,21 @@ where self } + /// Count every frame this track writes into `sent`, which its catalog reports. + pub(crate) fn with_sent(mut self, sent: std::sync::Arc) -> Self { + self.sent = Some(sent); + self + } + + /// Count a written frame. Only video keyframes count as keyframes: elsewhere the bit only + /// marks where a group starts. + fn count(&self, bytes: usize, keyframe: bool) { + if let Some(sent) = &self.sent { + let keyframe = keyframe && self.container.kind() == Kind::Video; + sent.add(1, bytes, keyframe.into()); + } + } + /// Raise the standing claim when the catalog estimate has a new peak. fn claim(&mut self) { let Some(bandwidth) = self.bandwidth.as_mut() else { @@ -466,11 +486,13 @@ where // group, a timestamp that won't convert) leaves the producer usable, and the estimate's // extrema never fall, so counting one would inflate the catalog for good. self.estimator.write(timestamp, bytes); + self.count(bytes, keyframe); self.observe_end(timestamp, duration); } else { // Buffered frames are measured on the way in instead. The flush that eventually writes // them takes the track down with it when it fails, so there is nothing to unwind. self.estimator.write(frame.timestamp, frame.payload.len()); + self.count(frame.payload.len(), frame.keyframe); self.observe_end(frame.timestamp, frame.duration); self.buffer.push(frame); diff --git a/rs/moq-mux/src/container/source.rs b/rs/moq-mux/src/container/source.rs index 09c4a701c2c..06025f01d81 100644 --- a/rs/moq-mux/src/container/source.rs +++ b/rs/moq-mux/src/container/source.rs @@ -96,6 +96,8 @@ pub(crate) struct ExportSource { cuts: u64, // A stopped encoder may still emit final packets in this group on the old timeline. ends: Option<(u64, u64)>, + /// The counters the export source reports for this rendition, as received, before any transform. + received: std::sync::Arc, } impl ExportSource { @@ -182,6 +184,7 @@ impl ExportSource { open: false, cuts: 0, ends: None, + received: source.received(name), }; source.resolve_video_dimensions(&[])?; Ok(Some(source)) @@ -219,6 +222,7 @@ impl ExportSource { open: false, cuts: 0, ends: None, + received: source.received(name), })) } @@ -249,6 +253,7 @@ impl ExportSource { open: false, cuts: 0, ends: None, + received: source.received(name), }) } @@ -413,6 +418,7 @@ impl ExportSource { self.finish_ends(); } self.open = true; + self.received.frame(frame.timestamp, frame.payload.len()); frame } Some(Item::Read(Read::Gap { timestamp, duration })) => { diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 97cb158f29a..c7e44954b33 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -2114,6 +2114,53 @@ async fn read_frames(consumer: &moq_net::broadcast::Consumer, name: &str, kind: frames } +/// The import counts what each rendition writes and the export what it reads: both agree with +/// the frames a subscriber reads back, and only video counts keyframes. +#[tokio::test(start_paused = true)] +async fn stats_count_what_the_import_writes_and_the_export_reads() { + let data = include_bytes!("test_data/bbb_cbr.ts"); + + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve()); + import.decode(&BytesMut::from(&data[..])).unwrap(); + import.finish().unwrap(); + + let snapshot = catalog.snapshot(); + let sent = catalog.stats(); + let video = snapshot.video.renditions.keys().map(|name| (name, Kind::Video)); + let audio = snapshot.audio.renditions.keys().map(|name| (name, Kind::Audio)); + let renditions: Vec<_> = video.chain(audio).collect(); + assert_eq!( + renditions.len(), + 2, + "the clip carries one video and one audio rendition" + ); + assert_eq!(sent.len(), renditions.len(), "a report per rendition: {sent:?}"); + for (name, kind) in renditions { + let frames = read_frames(&consumer, name, kind).await; + let report = &sent[name]; + assert!(!frames.is_empty(), "{name} carries frames"); + assert_eq!(report.frames, frames.len() as u64, "{name} frames"); + assert_eq!( + report.bytes, + frames.iter().map(Vec::len).sum::() as u64, + "{name} bytes" + ); + assert_eq!(report.keyframes > 0, kind == Kind::Video, "{name} keyframes"); + } + + let source = crate::source::announced(&consumer).await; + drain_with(open_export(source.clone()).await.unwrap()).await; + let received = source.stats(); + for (name, report) in &sent { + assert_eq!(received[name].frames, report.frames, "{name} frames read"); + assert_eq!(received[name].bytes, report.bytes, "{name} bytes read"); + assert!(received[name].arrival.is_some(), "{name} records its newest arrival"); + } +} + /// Both real Kyrion MP2 programs must survive TS -> MoQ -> TS byte-for-byte, and /// the PMT must re-announce them as MPEG-1 audio (0x03): the capture is 48 kHz, /// so the half-rate type (0x04) would be unfaithful. This capture is a dirty start diff --git a/rs/moq-mux/src/lib.rs b/rs/moq-mux/src/lib.rs index 7fb00707feb..67630fc8fd2 100644 --- a/rs/moq-mux/src/lib.rs +++ b/rs/moq-mux/src/lib.rs @@ -39,6 +39,7 @@ mod pace; pub mod rate; pub mod select; mod source; +mod stats; pub mod timeline; pub use clock::Clock; diff --git a/rs/moq-mux/src/source.rs b/rs/moq-mux/src/source.rs index 5cb73cb4188..c8f6c9437b1 100644 --- a/rs/moq-mux/src/source.rs +++ b/rs/moq-mux/src/source.rs @@ -34,6 +34,9 @@ pub struct Source { origin: moq_net::origin::Consumer, path: moq_net::PathOwned, pins: Pins, + /// What each rendition an export reads through this source has received. See + /// [`stats`](Self::stats). + received: Arc>, } impl Source { @@ -52,20 +55,39 @@ impl Source { origin, path, pins: Arc::new(Mutex::new(pins)), + received: Default::default(), } } /// This source holding nothing but `broadcast` at its own path: an export switching to - /// another instance starts over from the broadcast it switched to. + /// another instance starts over from the broadcast it switched to. Its counters carry on. pub(crate) fn switched(&self, broadcast: &moq_net::broadcast::Consumer) -> Self { let pins = HashMap::from([(self.path.clone(), broadcast.clone())]); Self { origin: self.origin.clone(), path: self.path.clone(), pins: Arc::new(Mutex::new(pins)), + received: self.received.clone(), } } + /// The counters an export reading rendition `name` bumps. + pub(crate) fn received(&self, name: &str) -> Arc { + self.received.track(name) + } + + /// What each rendition exported through this source (and its clones) has received, keyed by + /// rendition ID in the catalog that listed it, including renditions referenced from another + /// broadcast. + /// + /// Counters are cumulative for the life of the source, across resubscriptions and a + /// [`Follower`](crate::container::ts::Follower) switching instances, so a rendition no export + /// reads any more keeps its final counts. Fold this into an + /// [`echo::Snapshot`](hang::echo::Snapshot) on the feedback interval. + pub fn stats(&self) -> std::collections::BTreeMap { + self.received.received() + } + /// Serve this source's own path from `broadcast` from now on, as when the same instance /// returns on a new route and the old broadcast's routes are gone. pub(crate) fn pin(&self, broadcast: &moq_net::broadcast::Consumer) { diff --git a/rs/moq-mux/src/stats.rs b/rs/moq-mux/src/stats.rs new file mode 100644 index 00000000000..f877a495f49 --- /dev/null +++ b/rs/moq-mux/src/stats.rs @@ -0,0 +1,168 @@ +//! Cumulative per-rendition counters behind a publisher's stats and a viewer's feedback. +//! +//! Each track bumps its own atomics as it writes or reads a frame, so counting takes no lock. A +//! reporter reads the whole set about once a second through the [`Registry`] the tracks enrolled in: +//! the catalog [`Producer`](crate::catalog::Producer) for what a publisher writes, the export +//! [`Source`](crate::Source) for what a viewer reads. + +use std::collections::BTreeMap; +use std::sync::atomic::{AtomicU64, Ordering::Relaxed}; +use std::sync::{Arc, Mutex, PoisonError}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +/// What one published track has written. +#[derive(Default)] +pub(crate) struct Sent { + frames: AtomicU64, + bytes: AtomicU64, + keyframes: AtomicU64, +} + +impl Sent { + /// Count `frames` media frames carrying `bytes` of payload, `keyframes` of them keyframes. + pub(crate) fn add(&self, frames: u64, bytes: usize, keyframes: u64) { + self.frames.fetch_add(frames, Relaxed); + self.bytes.fetch_add(bytes as u64, Relaxed); + self.keyframes.fetch_add(keyframes, Relaxed); + } + + fn report(&self) -> hang::stats::Rendition { + let mut report = hang::stats::Rendition::default(); + report.frames = self.frames.load(Relaxed); + report.bytes = self.bytes.load(Relaxed); + report.keyframes = self.keyframes.load(Relaxed); + report + } +} + +/// What one exported track has read. +#[derive(Default)] +pub(crate) struct Received { + frames: AtomicU64, + bytes: AtomicU64, + /// The latest frame's media timestamp, in microseconds. + timestamp: AtomicU64, + /// When the latest frame arrived, in milliseconds since the Unix epoch, or zero before the first. + wall: AtomicU64, +} + +impl Received { + /// Count one frame of `bytes` payload at media `timestamp`, arriving now. + pub(crate) fn frame(&self, timestamp: moq_net::Timestamp, bytes: usize) { + self.frames.fetch_add(1, Relaxed); + self.bytes.fetch_add(bytes as u64, Relaxed); + // A clock before the MoQ epoch has no wire encoding, so such an arrival is never recorded. + let wall = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |since| u64::try_from(since.as_millis()).unwrap_or(0)); + if wall < hang::catalog::MOQ_EPOCH_UNIX_MILLIS { + return; + } + let micros = u64::try_from(timestamp.as_micros()).unwrap_or(u64::MAX); + self.timestamp.store(micros, Relaxed); + self.wall.store(wall, Relaxed); + } + + fn report(&self) -> hang::echo::Rendition { + let mut report = hang::echo::Rendition::default(); + report.frames = self.frames.load(Relaxed); + report.bytes = self.bytes.load(Relaxed); + // The pair is read without a lock, so it may straddle two consecutive frames: off by one + // frame interval at most, which no latency reading resolves. + let wall = self.wall.load(Relaxed); + if wall != 0 + && let Ok(timestamp) = moq_net::Timestamp::from_micros(self.timestamp.load(Relaxed)) + { + let wall = UNIX_EPOCH + Duration::from_millis(wall); + report.arrival = Some(hang::echo::Arrival::new(timestamp, wall)); + } + report + } +} + +/// Counters keyed by rendition ID, shared by the tracks that bump them and the reporter that reads +/// them. +pub(crate) struct Registry(Mutex>>); + +impl Default for Registry { + fn default() -> Self { + Self(Mutex::default()) + } +} + +impl Registry { + /// The counters for `name`, continuing those of an earlier track under that name. + pub(crate) fn track(&self, name: &str) -> Arc { + let mut entries = self.0.lock().unwrap_or_else(PoisonError::into_inner); + entries.entry(name.to_string()).or_default().clone() + } + + /// Read every entry. A track that ended keeps its final counts for the registry's life, as the + /// catalog a finished import leaves up still lists it. + fn report(&self, read: impl Fn(&T) -> R) -> BTreeMap { + let entries = self.0.lock().unwrap_or_else(PoisonError::into_inner); + entries + .iter() + .map(|(name, counters)| (name.clone(), read(counters))) + .collect() + } +} + +impl Registry { + /// What each enrolled track has written. + pub(crate) fn sent(&self) -> BTreeMap { + self.report(Sent::report) + } +} + +impl Registry { + /// What each enrolled track has read. + pub(crate) fn received(&self) -> BTreeMap { + self.report(Received::report) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn an_ended_track_keeps_its_final_counts() { + let registry = Registry::::default(); + let track = registry.track("video"); + track.add(3, 300, 1); + drop(track); + + for _ in 0..2 { + let report = registry.sent(); + assert_eq!(report["video"].frames, 3); + assert_eq!(report["video"].bytes, 300); + assert_eq!(report["video"].keyframes, 1); + } + } + + #[test] + fn a_returning_track_continues_its_counts() { + let registry = Registry::::default(); + registry.track("video").add(2, 20, 0); + registry.track("video").add(1, 10, 0); + assert_eq!(registry.sent()["video"].frames, 3); + assert_eq!(registry.sent()["video"].bytes, 30); + } + + #[test] + fn arrival_pairs_the_latest_timestamp_with_its_wall_time() { + let registry = Registry::::default(); + let track = registry.track("hd"); + assert_eq!(registry.received()["hd"].arrival, None); + + let before = SystemTime::now(); + track.frame(moq_net::Timestamp::from_millis(40).unwrap(), 100); + track.frame(moq_net::Timestamp::from_millis(80).unwrap(), 50); + let report = ®istry.received()["hd"]; + assert_eq!((report.frames, report.bytes), (2, 150)); + let arrival = report.arrival.as_ref().unwrap(); + assert_eq!(arrival.timestamp.as_millis(), 80); + assert!(arrival.wall + Duration::from_millis(1) >= before); + } +}