diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 7659f43353..5661434c54 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -367,7 +367,11 @@ its linger restarts its counters under a new epoch without an announce. The node's totals count everything since its epoch began, so billing reads those; a group's per-path traffic and per-label sessions and HTTP requests are for dashboards. Session counters report how close sessions come to the -per-session limits, past which one is closed with `TOO_MANY_REQUESTS`. +per-session limits, past which one is closed with `TOO_MANY_REQUESTS`. The +totals and prefix tracks also carry how far viewers are behind +([`lag`](/concept/stats#traffic)) and what was dropped before they acknowledged +it, sampled once per `interval`: while the broadcast produces, a viewer that +acknowledges nothing for a whole interval reads at least that far behind. [Stats](/concept/stats) describes the paths, tracks, and encodings; read them with the [`moq-stats`](https://docs.rs/moq-stats) crate. diff --git a/doc/concept/stats.md b/doc/concept/stats.md index 9c03f08540..451ab8565d 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -299,7 +299,8 @@ reader that misses every frame from the group's last change until it is unlisted loses that instance's tail, so a meter reads the node's totals. `paths` holds each broadcast path's egress (`publisher`) and ingress -(`subscriber`) [Traffic](#traffic). A path appears from its first nonzero +(`subscriber`) [Traffic](#traffic), without `lag` and `dropped`: the group's +own totals, the node's totals, and the prefix tracks carry those. A path appears from its first nonzero counter until the relay drops its counters, even while idle: the last viewer can leave while the publisher still holds the path, and a returning viewer resumes the same counters. So a path missing from a frame had no counters, and @@ -367,6 +368,8 @@ stale. "bytes": 1048576, "frames": 900, "groups": 30, "datagrams": 0, "inflated": 0, "stale": { "bytes": 0, "frames": 0, "groups": 0, "datagrams": 0 }, + "lag": { "50ms": 1040000, "1s": 8576 }, + "dropped": { "bytes": 1500, "frames": 15, "groups": 1, "datagrams": 0 }, "announced": 1, "announced_closed": 0, "broadcasts": 3, "broadcasts_closed": 1, "subscriptions": 6, "subscriptions_closed": 2 @@ -385,6 +388,18 @@ stale. | `datagrams` | Groups delivered as an unreliable datagram. A subset of `groups`. | | `inflated` | Bytes inflated from compressed groups for plain readers, counted once per group however many share the copy, plus received compressed datagrams, which inflate on arrival. Not part of `bytes`, which counts payload as stored or sent. | | `stale` | Payload skipped because it aged past a subscriber's latency budget, with the same four fields. Disjoint from the top-level payload counters. | +| `lag` | Egress only: how far each subscription's peer is behind, weighted by bytes. Every stats interval, the bytes a subscription's track produced since the last are counted in the bucket of its lag then. Keys are each bucket's exclusive upper edge (`50ms`, `100ms`, `250ms`, `500ms`, `1s`, `2s`, `5s`, then `inf`), and an empty bucket is left out. | +| `dropped` | Egress only: content written on group streams that ended before the peer acknowledged them (a skip, an expiry, or a closed session), with the same four fields as `stale`. `groups` counts those that carried a frame, `frames` the frames they carried, and `bytes` the payload the transport accepted on them. A subset of the top-level payload counters, which count it as handed over. `datagrams` stays zero, since a datagram has no acknowledgement to miss. | + +A subscription's lag is the newest media its track produced minus the newest +the peer acknowledged, counting a group stream as acknowledged once its FIN +is. It starts level with the newest media when the subscription opens, so a +backlog it starts behind is not lag. A peer that acknowledges nothing for a +whole interval in which the track produced reads at least that interval behind, +and every interval after adds to it until an acknowledgement arrives, so a +media clock that crawls or stops cannot hide a stalled viewer. A subscription +that closes counts its last bytes at its lag when it closed. Diff two frames +for one interval's distribution, and sum nodes bucket by bucket. The last six fields are legacy spellings of the `*_started` and `*_ended` counters, still written so an older consumer reads a newer relay. A reader diff --git a/quest/m2/README.md b/quest/m2/README.md index a24c3b2545..23006dc3d6 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -30,7 +30,6 @@ lite-07; bitrate enforcement is not part of this release. - [Generated C bindings](/quest/m2/c/README.md) - C generated from moq-ffi ships as `moq-c` 0.8.0 and replaces the hand-written crate - [Benchmark sources](/quest/m2/bench-sources.md) - every measuring nightly benchmark feeds the regression table, and none is a smoke run - [Poll-native async](/quest/m2/ffi-poll-migrate.md) - moq-ffi's pure-kio async exports poll their state directly instead of going through `ffi::Task` -- [QoS](/quest/m2/qos/README.md) - the relay's egress `lag` histogram and `dropped` counters on the stats totals and prefix tracks - [Lazy retained HLS instances](/quest/m2/hls-lazy-instances.md) - a replaced HLS instance starts its watchers only when a request reaches it - [HLS and DASH browser playback](/quest/m2/hls-player-test.md) - hls.js and dash.js play, reload, switch, and survive reconfiguration nightly - [moq-hls export on kio](/quest/m2/hls-export-kio.md) - export watchers and the init build are owned kio tasks instead of tokio::spawn diff --git a/quest/m2/latency-ledger.md b/quest/m2/latency-ledger.md index 2b77623727..4e3f052120 100644 --- a/quest/m2/latency-ledger.md +++ b/quest/m2/latency-ledger.md @@ -35,5 +35,4 @@ probes, which was the right trade to get it running. This quest promotes them. ## Related -- [QoS](/quest/m2/qos/README.md) - relay-side health, the same idea from the other end - [Media stats](/quest/m2/stats/README.md) - the hang stats and `.echo` snapshots these stages join diff --git a/quest/m2/qos-lag-dashboard.md b/quest/m2/qos-lag-dashboard.md index 39979adf29..daf95bf759 100644 --- a/quest/m2/qos-lag-dashboard.md +++ b/quest/m2/qos-lag-dashboard.md @@ -4,19 +4,17 @@ The demo stats dashboard (`demo/web/src/stats.ts`) shows the relay's viewer lag and dropped media: percentiles of the egress `lag` histogram over the -chart window, and `dropped` duration, bytes, and groups as rates, cluster-wide +chart window, and `dropped` bytes, frames, and groups as rates, cluster-wide and per node like the existing counters. ## Plan -- Read both from the stats totals and prefix tracks - (`/prefix.json`) rather than the group tracks' per-path rows they sit on today (decided in the - 2026-10-06 audit; the rows stay for existing readers, 2026-10-12); [viewer lag histogram](/quest/m2/qos/lag-histogram.md) - lands them on the egress side of its per-group totals and prefix tracks - (decided 2026-10-08). `lag` is a cumulative byte count per bucket keyed +- Read both from the egress side of the stats totals, a group's totals, and + the prefix tracks (`/prefix.json`); a group's per-path `paths` rows + leave them out (decided in the 2026-10-06 audit). `lag` is a cumulative byte count per bucket keyed by its upper edge (`"50ms"` to `"5s"`, then `"inf"`, empty buckets - omitted); `dropped` is `{ duration, bytes, groups }` with the duration in - fractional milliseconds. `doc/concept/stats.md` documents them. Diff two + omitted); `dropped` has the same four fields as `stale` (`bytes`, `frames`, + `groups`, `datagrams`). `doc/concept/stats.md` documents them. Diff two samples within one subscription for an interval's distribution, as the dashboard already does to turn cumulative bytes into rates, and sum nodes bucket by bucket. @@ -30,7 +28,3 @@ and per node like the existing counters. package is planned. Public API: none. Wire: none. - -## Required - -- [Viewer lag histogram](/quest/m2/qos/lag-histogram.md) - the `lag` histogram and `dropped` counters the dashboard reads diff --git a/quest/m2/qos-publisher-timeliness.md b/quest/m2/qos-publisher-timeliness.md index f6f4d34021..41bb2c736d 100644 --- a/quest/m2/qos-publisher-timeliness.md +++ b/quest/m2/qos-publisher-timeliness.md @@ -10,9 +10,8 @@ cooperation. ## Plan -This is the ingress mirror of the egress `lag` histogram ([viewer lag -histogram](/quest/m2/qos/lag-histogram.md), which lands first so this reuses -its histogram type, decided 2026-10-08) and needs no acknowledgments: the relay is the receiver, and a frame is measured +This is the ingress mirror of the egress `lag` histogram (reuse its +`moq_net::stats::Histogram` type, decided 2026-10-08) and needs no acknowledgments: the relay is the receiver, and a frame is measured when its last byte arrives, because a partial frame is not useful. Frame timestamps are relative and jittered with no epoch, so measure each @@ -51,7 +50,3 @@ simulated hour; a paused publisher records a stall and no drift for frames it never sent; a regression across groups is counted while intra-group reordering is not; a track without a timescale is excluded. - -## Required - -- [Viewer lag histogram](/quest/m2/qos/lag-histogram.md) - the egress histogram whose type and edges this reuses diff --git a/quest/m2/qos/README.md b/quest/m2/qos/README.md deleted file mode 100644 index ef385e5d8a..0000000000 --- a/quest/m2/qos/README.md +++ /dev/null @@ -1,67 +0,0 @@ -# QoS: broadcast health and congestion - -## Goal - -Publish the relay QoS lag and dropped-counter API after the next release. -The histogram lands on totals and prefix tracks beside the existing maps. -Publisher timeliness, client media health, and dashboard presentation remain -separate additive work in m2. - -## Plan - -Two layers, and the split is what makes this several quests rather than one. -`moq-lite` knows about DELIVERY, what was written toward a peer and what the -peer acknowledged, per subscription, and `hang` knows about MEDIA, which is -what a health verdict has to mean. Neither is useful alone: delivery without -media context cannot tell a slow viewer from a keyframe burst, and media -without delivery cannot tell congestion from a publisher that stopped. - -Everything the relay reports is distilled per broadcast on the existing -`moq-stats` keys. No per-subscriber or per-session row reaches the wire from -the relay: many subscriptions collapse into byte-weighted cumulative -histograms, which stay monotonic and merge-patch friendly, and which any -consumer can diff into a distribution. Clients report for themselves through -hang stats and `.echo` feedback tracks, planned in their own line on `main`. - -Only the lag counters land here. Client health is -[graded in hang](/doc/concept/stats.md#health), and preflight is -[preflight](/quest/m2/stats/preflight.md) in the media stats line. The -moq.pro (downstream) dashboard work, including the health badge and the -connection-health drill-down, consumes both. - -The per-broadcast verdict combining client reports, the relay's starvation, -and publisher timeliness lives downstream. This line only reports the -counters. - -Decided (2026-10-06 audit): stats prefix tracks (`/prefix.json`) -landed first in #5811, since they are a prerequisite of the lag API; the lag -histogram was deferred to m2 by the maintainer on 2026-10-15 and lands on the -stats totals and -prefix tracks rather than the group tracks' per-path -`paths` rows, which stay unchanged for existing readers -(2026-10-12). -Rejected: the histogram first, which would make prefix tracks carry the lag -histogram and dropped counters across. - -Decided (2026-10-08): the egress histogram is its own quest, -[viewer lag histogram](/quest/m2/qos/lag-histogram.md), since every other -child reads it. The line's PR, #4133, closed without landing (2026-10-09); -its code is kept on `wip/4133-lag-histogram`. - -Decided (2026-09-28): the line's moq-stats changes break the published -crate. Client stats left the line (2026-09-29) -when media stats moved out of moq-stats onto hang tracks, which are additive -on `main`. - -## Required - -- [Viewer lag histogram](/quest/m2/qos/lag-histogram.md) - the egress `lag` - histogram and `dropped` counters land on the stats totals and prefix - tracks - -## Related - -- [Media stats](/quest/m2/stats/README.md) - publishers and viewers report - their own media, transport, and playback health, the media half of a verdict -- [Loss delay](/quest/m3/cut-through/loss-delay.md) - an ingress counter of - bytes a loss held back by at least one RTT, on the same rows diff --git a/quest/m2/qos/lag-histogram.md b/quest/m2/qos/lag-histogram.md deleted file mode 100644 index 596e617f6c..0000000000 --- a/quest/m2/qos/lag-histogram.md +++ /dev/null @@ -1,58 +0,0 @@ -# [M] Viewer lag histogram - -## Goal - -The relay's `moq-stats` egress side reports, per broadcast, how far behind -its viewers are according to what the network has acknowledged: a -byte-weighted cumulative `lag` histogram and `dropped` counters (duration, -bytes, and groups dropped before they were acknowledged), on the stats totals -and prefix tracks (`/prefix.json`). A spliced subscription -stays exact across a route switch, and a closing subscription records its -last partial interval. - -## Plan - -Decided 2026-10-08: split out of the QoS line so the dashboard and publisher -timeliness can require a histogram that exists on `main`. This quest lands it -on `main` now that prefix tracks have landed -([#5811](https://github.com/moq-dev/moq/pull/5811)), moved from the -group tracks' per-path `paths` rows it writes today -onto the totals and prefix tracks. - -Decided 2026-10-09: the QoS line -([#4133](https://github.com/moq-dev/moq/pull/4133)) closed without landing, -since every code change it had over `main` is this histogram. Its code is kept -on `wip/4133-lag-histogram` as this quest's starting point: the sampler -(#4298, `rs/moq-net/src/stats.rs`), the splice fixes (#5009), and the final -sample (#4451). Porting it is not mechanical: - -- `main` moved to caller-supplied time (#4437) and removed the frozen test - clock (`model::clock::advance`). The sampler reads wall time in - `Registry::report`, when a `Frontier` opens, on `Delivery` acks, and in the - final sample in `Drop for FrontierInner`, which has no caller to supply a - time. Redesign those to take time from their callers, and give the final - sample an explicit close path. -- Untimed tracks (#4822): record `Production` only for timed frames. -- `max_age` became `max_delay` on subscriptions (#4917). - -Open review findings on #4133 to fix while porting: - -- A timestamp regression hides a stalled viewer - ([3](https://github.com/moq-dev/moq/pull/4133#issuecomment-6053929743)). -- A copy served across a no-route gap stays watched - ([B](https://github.com/moq-dev/moq/pull/4133#issuecomment-6053929743)). -- A partial write is charged the frame's full declared size - ([P2](https://github.com/moq-dev/moq/pull/4133#pullrequestreview-5452586850)). -- `Dropped::add` should saturate instead of overflowing - ([comment](https://github.com/moq-dev/moq/pull/4133#discussion_r4215691174)). -- Bound the frontier list when `Registry::report` is not called - ([comment](https://github.com/moq-dev/moq/pull/4133#discussion_r4215691184)). - -Keep the histogram contract the dashboard reads: buckets keyed by upper edge -(`"50ms"` to `"5s"`, then `"inf"`, empty buckets omitted), cumulative and -monotonic so readers diff two samples and sum nodes bucket by bucket. Update -`doc/concept/stats.md` and the stats section of `doc/bin/relay/config.md`. - -Public API: breaking in `moq-stats`, as the line already decided. Wire: the -stats JSON gains the egress fields. - diff --git a/quest/m2/starvation-frames.md b/quest/m2/starvation-frames.md index 753a13974e..1b9d2fad61 100644 --- a/quest/m2/starvation-frames.md +++ b/quest/m2/starvation-frames.md @@ -26,9 +26,9 @@ pending frame and retire everything at or below the acknowledged offset. Each acknowledged frame moves the subscription's frontier to that frame's timestamp, so the interval samples from the lag histogram read a frontier that is at most one frame stale instead of one group. When a stream -is reset before a frame is acknowledged, `dropped_duration` now grows by the -span from the newest acknowledged frame to the newest written one, which is -the exact media the viewer lost. +is reset before a frame is acknowledged, `dropped` now counts only the frames +written after the newest acknowledged one, which is the exact media the viewer +lost. Add a second byte-weighted cumulative histogram of delivery delay: `acked_at - written_at` per frame, where `acked_at` is when the serve task is @@ -42,19 +42,15 @@ Backends that return unsupported from `poll_acked` keep the group-granularity sampling from the lag histogram, so the histogram never disappears over qmux or a browser transport; document which resolution a node offers. -Deferred to m2 with the parent QoS sampler. Group-granularity lag lands -first after the next release; frame-granularity sampling then adds the ACK -hook without making either resolution a release gate. - -Decided 2026-10-08: the group-granularity sampler is the -[viewer lag histogram](/quest/m2/qos/lag-histogram.md), not yet on main (in -progress as [moq-dev/moq#6791](https://github.com/moq-dev/moq/pull/6791)), so -that quest is Required. +The group-granularity sampler is `Registry::report` in +`rs/moq-net/src/stats.rs`, reading the frontier that `stats::Delivery` +advances on each acknowledged group FIN. Frame-granularity sampling adds the +ACK hook without making either resolution a release gate. Tests: the frontier tracking frame ends under a peer that acknowledges in bursts, with interval samples landing one bucket lower than group-granularity tracking of the same run; a reset mid-group attributing only the -unacknowledged span to `dropped_duration`; delivery delay staying inside one bucket of the true RTT plus the peer's +unacknowledged frames to `dropped`; delivery delay staying inside one bucket of the true RTT plus the peer's `max_ack_delay`; and fallback to group-granularity frontier tracking when the hook is unsupported. @@ -62,5 +58,3 @@ unsupported. - [poll_acked on moq-net's send stream](/quest/m2/quic-ack-hook.md) - the hook this samples through -- [Egress lag histogram](/quest/m2/qos/lag-histogram.md) - the - group-granularity sampler and starvation histogram this refines diff --git a/quest/m2/stats/README.md b/quest/m2/stats/README.md index 81d4fc0110..7dc431557a 100644 --- a/quest/m2/stats/README.md +++ b/quest/m2/stats/README.md @@ -89,9 +89,8 @@ kind. removes `.z` siblings; follow the delta policy it retains on the single track. An unsubscribed stats track never leaves the process. Gauges are carried but never summed. - **On main.** Every change is additive: optional sections on - `#[non_exhaustive]` catalog types, and new types. The line left the - [QoS](/quest/m2/qos/README.md) line, which keeps the relay's moq-stats - changes. + `#[non_exhaustive]` catalog types, and new types. The relay's moq-stats + changes stay out of this line. - **No `@moq/stats` package.** Media types live in `@moq/hang`, and the demo dashboard's relay-stats reader stays where it is. - Docs stay inline: `doc/concept/hang.md` documents both catalog sections, @@ -121,8 +120,6 @@ connection view and the dashboard flow. ## Related -- [QoS](/quest/m2/qos/README.md) - the relay's delivery counters; the - combined per-broadcast verdict reading both lives downstream - [Encoder feedback](/quest/m3/stats-encoder-feedback.md) - a Rust encoder adapts its bitrate to what its viewers report - [Latency ledger](/quest/m2/latency-ledger.md) - adds the viewer's latency diff --git a/quest/m3/cut-through/README.md b/quest/m3/cut-through/README.md index f5b658ff20..338e1134af 100644 --- a/quest/m3/cut-through/README.md +++ b/quest/m3/cut-through/README.md @@ -75,5 +75,4 @@ stats field, which its quest documents inline. ## Related -- [QoS](/quest/m2/qos/README.md) - the loss-delay counter follows its per-broadcast ingress row conventions - [Frame slot accounting](https://github.com/moq-dev/moq/pull/5239) - landed: the cache charges frame slot growth, which covers the in-flight frames a cut-through build would hold diff --git a/quest/m3/cut-through/loss-delay.md b/quest/m3/cut-through/loss-delay.md index 2b2ca92d01..f1b747051f 100644 --- a/quest/m3/cut-through/loss-delay.md +++ b/quest/m3/cut-through/loss-delay.md @@ -38,7 +38,3 @@ an RTT counts nothing. Public API: an additive trait accessor. Wire: a new `moq-stats` field, additive on `main` since `Traffic` is `#[non_exhaustive]` and readers default missing fields. - -## Related - -- [QoS](/quest/m2/qos/README.md) - the per-broadcast ingress conventions this follows diff --git a/rs/moq-net/benches/stats.rs b/rs/moq-net/benches/stats.rs index cc6e1300db..9973caa4e2 100644 --- a/rs/moq-net/benches/stats.rs +++ b/rs/moq-net/benches/stats.rs @@ -1,12 +1,14 @@ //! Drain benchmark for the stats registry: [`stats::Registry::report`] refilling //! one reused report, swept over broadcasts, tiers, labels, sessions and connections -//! so a cost that grows faster than the entries it reports shows up as a slope. +//! so a cost that grows faster than the entries it reports shows up as a slope, and +//! over egress subscriptions per broadcast, since every drain also samples each +//! subscription's lag. //! //! Run with `cargo bench -p moq-net --bench stats`. use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; use futures::FutureExt; -use moq_net::{announce, fuzz, origin, stats, transport}; +use moq_net::{Timescale, Timestamp, announce, fuzz, origin, stats, track, transport}; /// `(broadcasts, tiers)` shapes; every broadcast records under every tier. const SHAPES: [(usize, usize); 4] = [(100, 1), (1_000, 1), (100, 10), (1_000, 10)]; @@ -48,7 +50,7 @@ fn report(c: &mut Criterion) { for (broadcasts, tiers) in SHAPES { let fixture = registry(broadcasts, tiers); let mut report = stats::Report::default(); - fixture.registry.report(&mut report); + fixture.registry.report(&mut report, moq_net::time::Instant::ZERO); assert_eq!( report.traffic.len(), broadcasts * tiers, @@ -58,7 +60,7 @@ fn report(c: &mut Criterion) { group.bench_with_input( BenchmarkId::from_parameter(format!("{broadcasts}x{tiers}")), &fixture, - |b, fixture| b.iter(|| fixture.registry.report(&mut report)), + |b, fixture| b.iter(|| fixture.registry.report(&mut report, moq_net::time::Instant::ZERO)), ); } group.finish(); @@ -122,7 +124,7 @@ fn session_report(c: &mut Criterion) { }) .collect(); let mut report = stats::Report::default(); - registry.report(&mut report); + registry.report(&mut report, moq_net::time::Instant::ZERO); assert_eq!(report.sessions.len(), labels); for row in &report.sessions { assert_eq!(row.presence.sessions_started, sessions as u64); @@ -132,7 +134,7 @@ fn session_report(c: &mut Criterion) { group.throughput(Throughput::Elements((labels * sessions) as u64)); group.bench_function( BenchmarkId::from_parameter(format!("{labels}x{sessions}/connections={connections}")), - |b| b.iter(|| registry.report(&mut report)), + |b| b.iter(|| registry.report(&mut report, moq_net::time::Instant::ZERO)), ); std::hint::black_box((contexts, wires)); } @@ -174,5 +176,111 @@ fn retag(c: &mut Criterion) { group.finish(); } -criterion_group!(benches, report, request, write, session_report, retag); +/// `(broadcasts, subscriptions per broadcast)` shapes for the lag sampler. +const SUBSCRIBED: [(usize, usize); 4] = [(100, 1), (1_000, 1), (100, 10), (1_000, 10)]; + +/// A registry with `subscriptions` tagged egress subscriptions on each of +/// `broadcasts` live broadcasts. The cache has a byte target, as a relay's does, so +/// producing a group per report evicts old ones rather than growing. +struct Subscribed { + registry: stats::Registry, + _driver: origin::Driver, + publishers: Vec<(origin::Publication, track::Producer)>, + _subscribers: Vec, + /// Milliseconds of media produced so far. + produced: std::cell::Cell, +} + +impl Subscribed { + /// One second of media on every track: a group of one 100 byte frame. + fn produce(&self) { + let at = Timestamp::from_millis(self.produced.get()).unwrap(); + for (_, track) in &self.publishers { + let mut group = track.append_group().unwrap(); + group.write_frame(at, vec![0u8; 100]).unwrap(); + group.finish().unwrap(); + } + self.produced.set(self.produced.get() + 1000); + } +} + +fn subscribed(broadcasts: usize, subscriptions: usize) -> Subscribed { + let registry = stats::Registry::new(stats::Config::new()); + let mut config = origin::Config::default(); + config.pool = moq_net::cache::Pool::new(moq_net::cache::Config::default().with_capacity(16 << 20)); + let (producer, mut driver) = origin::Producer::new(config); + let waiter = kio::Waiter::noop(); + let timed = track::Info::default().with_timescale(Timescale::MILLI); + let mut publishers = Vec::new(); + let mut subscribers = Vec::new(); + for i in 0..broadcasts { + let path = format!("room/{i}"); + let broadcast = producer.publish(path.as_str(), origin::Route::default()).unwrap(); + let track = broadcast.create_track("video", timed.clone()).unwrap(); + for _ in 0..subscriptions { + let session = registry.tier(stats::Tier::default()).session("root"); + let pending = producer + .consume() + .with_stats(session) + .request_broadcast(path.as_str(), None); + driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); + let consumer = pending.now_or_never().expect("resolves once driven").unwrap(); + let pending = consumer.track("video").unwrap().subscribe(None); + driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); + subscribers.push(pending.now_or_never().expect("subscribes once driven").unwrap()); + } + publishers.push((broadcast, track)); + } + Subscribed { + registry, + _driver: driver, + publishers, + _subscribers: subscribers, + produced: std::cell::Cell::new(0), + } +} + +/// A report on a live relay: every subscription's track produced since the last, so +/// each one's bytes are weighed. Only the report is timed. +fn sample(c: &mut Criterion) { + let mut group = c.benchmark_group("stats/sample"); + for (broadcasts, subscriptions) in SUBSCRIBED { + let fixture = subscribed(broadcasts, subscriptions); + let mut report = stats::Report::default(); + fixture.produce(); + fixture.registry.report(&mut report, moq_net::time::Instant::now()); + let subscribed: u64 = report.traffic.iter().map(|e| e.publisher.subscriptions_started).sum(); + assert_eq!( + subscribed, + (broadcasts * subscriptions) as u64, + "every subscription is tagged" + ); + let weighed: u64 = report.traffic.iter().map(|e| e.publisher.lag.total()).sum(); + assert_eq!( + weighed, + (broadcasts * subscriptions * 100) as u64, + "every subscription is weighed" + ); + + group.bench_with_input( + BenchmarkId::from_parameter(format!("{broadcasts}x{subscriptions}")), + &fixture, + |b, fixture| { + b.iter_custom(|iters| { + let mut spent = std::time::Duration::ZERO; + for _ in 0..iters { + fixture.produce(); + let now = moq_net::time::Instant::now(); + fixture.registry.report(&mut report, now); + spent += now.elapsed(); + } + spent + }) + }, + ); + } + group.finish(); +} + +criterion_group!(benches, report, request, write, session_report, retag, sample); criterion_main!(benches); diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index e9c8284803..a881638040 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -1481,8 +1481,8 @@ mod tests { bump_publisher_bytes(®istry, "room/1", &tier, 5); let mut report = crate::stats::Report::default(); - registry.report(&mut report); - registry.report(&mut report); + registry.report(&mut report, crate::time::Instant::ZERO); + registry.report(&mut report, crate::time::Instant::ZERO); let entry = report .traffic .iter() diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index ef756f7101..2e9513c5f0 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -5095,7 +5095,8 @@ impl TrackServe { slice, ) .counted(self.opened.clone()) - .terminal(self.terminal.clone()), + .terminal(self.terminal.clone()) + .delivered(self.track.deliver()), ); } Poll::Ready(Ok(None)) => { @@ -5492,6 +5493,9 @@ struct GroupServe { ended: bool, terminal: Option>, marked: bool, + /// What this stream wrote, for the subscription's acknowledged frontier: an + /// acknowledged FIN advances it, anything else counts as dropped on drop. + delivery: crate::stats::Delivery, state: GroupState, /// A long cached group drains a slice per poll, so its siblings still run. budget: kio::coop::Budget, @@ -5564,6 +5568,7 @@ impl GroupServe { ended: false, terminal: None, marked: false, + delivery: Default::default(), state: GroupState::Open, budget: kio::coop::Budget::new(32), } @@ -5574,6 +5579,12 @@ impl GroupServe { self } + /// Report what this stream writes to its subscription's acknowledged frontier. + fn delivered(mut self, delivery: crate::stats::Delivery) -> Self { + self.delivery = delivery; + self + } + /// Count this group's stream into `opened` once it opens. fn counted(mut self, opened: Arc) -> Self { self.opened = opened; @@ -5665,7 +5676,8 @@ impl GroupServe { } if let Some(pending) = chunk { match writer.poll_write(&mut cx, pending) { - Poll::Ready(Ok(_)) => { + Poll::Ready(Ok(written)) => { + self.delivery.wrote(written); if !bytes::Buf::has_remaining(pending) { *chunk = None; } @@ -5727,6 +5739,7 @@ impl GroupServe { }, )?; self.prior_object = Some(id); + self.delivery.frame(batched.timestamp); let payload = std::mem::take(&mut batched.frame.payload); if !payload.is_empty() { *chunk = Some(payload); @@ -5775,6 +5788,7 @@ impl GroupServe { }, )?; self.prior_object = Some(id); + self.delivery.frame(next.timestamp); // An empty object has no payload to stream. if next.size > 0 { *frame = Some(next); @@ -5877,6 +5891,7 @@ impl GroupServe { terminal.sent.store(true, Ordering::Relaxed); } return Poll::Ready(res.map(|()| { + std::mem::take(&mut self.delivery).acked(); tracing::debug!(sequence, "finished group"); })); } @@ -7011,6 +7026,88 @@ mod group_priority_test { assert!(matches!(serving.await, Err(Error::Old))); } + + /// Subgroup streams report to their subscription's frontier: an acknowledged FIN + /// moves it, and a stream abandoned mid-frame drops only what the transport took. + #[moq_clock::test] + async fn subgroups_report_to_the_frontier() { + use futures::FutureExt; + + let registry = crate::stats::Registry::new(crate::stats::Config::new()); + let stats = registry.tier(crate::stats::Tier::default()).session("root"); + let broadcast = crate::broadcast::Info::new().produce(); + let track = broadcast.create_track("test", crate::track::Info::timed()).unwrap(); + let sub = broadcast + .consume() + .with_stats(stats.egress("demo")) + .track("test") + .unwrap() + .subscribe(None) + .now_or_never() + .unwrap() + .unwrap(); + let serve = |session: SinkSession, group: group::Consumer| { + GroupServe::new( + crate::driver::Transport::new(session), + ietf::GroupHeader { + track_alias: 0, + group_id: group.sequence, + sub_group_id: 0, + publisher_priority: 0, + flags: Default::default(), + }, + Rank::solo(kio::Producer::new(0).consume()), + group, + Some(Timescale::MILLI), + Version::Draft18, + GroupSlice::default(), + ) + .delivered(sub.deliver()) + }; + let ms = |ms| crate::Timestamp::from_millis(ms).unwrap(); + + // Acknowledged: the frontier moves to its newest frame, 500ms. + let mut group = track.create_group(group::Info { sequence: 0 }).unwrap(); + group.write_frame(ms(0), b"a".as_slice()).unwrap(); + group.write_frame(ms(500), b"b".as_slice()).unwrap(); + group.finish().unwrap(); + let mut done = serve(SinkSession::new(Default::default()), group.consume()); + kio::wait(|waiter| done.poll_serve(waiter)).await.unwrap(); + + // The transport takes the headers and part of the payload, then holds the rest + // until the stream is abandoned. + let mut group = track.create_group(group::Info { sequence: 1 }).unwrap(); + group.write_frame(ms(1000), vec![0u8; 1000]).unwrap(); + let gate = kio::Producer::new(false); + let session = SinkSession::gated_uni_prefix(100, gate.consume()); + let log = session.log.clone(); + let mut abandoned = serve(session, group.consume()); + assert!( + kio::wait(|waiter| abandoned.poll_serve(waiter)) + .now_or_never() + .is_none() + ); + drop(abandoned); + let payload = log + .trail() + .into_iter() + .rev() + .find_map(|(_, sent)| match sent { + crate::lite::test_transport::Sent::Write(bytes) => Some(bytes), + _ => None, + }) + .expect("a payload write"); + assert!((1..1000).contains(&payload.len()), "{}", payload.len()); + + let mut report = crate::stats::Report::default(); + registry.report(&mut report, crate::time::Instant::ZERO); + let row = report.traffic[0].publisher; + assert_eq!(row.dropped.groups, 1); + assert_eq!(row.dropped.bytes, payload.len() as u64); + assert_eq!(row.dropped.frames, 1); + // 1000ms produced against a 500ms frontier: the [500ms, 1s) bucket. + assert_eq!(row.lag.buckets(), &[0, 0, 0, 0, 1002, 0, 0, 0]); + } } #[cfg(test)] diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index e2cde332c6..d669aa3750 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -4333,14 +4333,10 @@ impl TrackRun { ), }; debug_assert!(self.opening.is_none(), "one group waits for stream credit at a time"); - self.opening = Some(GroupServe::new( - self.ctx.group_context(), - sequence, - frame_start, - capped, - rank, - group, - )); + self.opening = Some( + GroupServe::new(self.ctx.group_context(), sequence, frame_start, capped, rank, group) + .delivered(self.track.deliver()), + ); } } @@ -4382,6 +4378,9 @@ struct GroupServe { // frame's delta is absolute (against an implicit prev value of 0), every // subsequent delta is signed against the previous frame. prev_ts: u64, + // What this stream wrote, for the subscription's acknowledged frontier: an + // acknowledged FIN advances it, anything else counts as dropped on drop. + delivery: crate::stats::Delivery, state: GroupState, // A long cached group drains a slice per poll, so its siblings still run. budget: kio::coop::Budget, @@ -4438,11 +4437,18 @@ impl GroupServe { frame_start, capped, prev_ts: 0, + delivery: Default::default(), state: GroupState::Open, budget: kio::coop::Budget::new(32), } } + /// Report what this stream writes to its subscription's acknowledged frontier. + fn delivered(mut self, delivery: crate::stats::Delivery) -> Self { + self.delivery = delivery; + self + } + /// Count opened streams for the scheduling tests. fn settle_open(&mut self, _opened: bool) { #[cfg(test)] @@ -4600,7 +4606,8 @@ impl GroupServe { } if let Some(pending) = chunk { match writer.poll_write(&mut cx, pending) { - Poll::Ready(Ok(_)) => { + Poll::Ready(Ok(written)) => { + self.delivery.wrote(written); if !bytes::Buf::has_remaining(pending) { *chunk = None; } @@ -4654,6 +4661,7 @@ impl GroupServe { if let Err(err) = buffered { break 'serve Err(err); } + self.delivery.frame(batched.frame.timestamp); let payload = std::mem::take(&mut batched.frame.payload); if !payload.is_empty() { *chunk = Some(payload); @@ -4708,6 +4716,7 @@ impl GroupServe { if let Err(err) = buffered { break 'serve Err(err); } + self.delivery.frame(next.timestamp); *frame = Some(next); } Poll::Ready(Ok(None)) => match records.take() { @@ -4766,6 +4775,7 @@ impl GroupServe { }; self.state = GroupState::Done; return Poll::Ready(res.map(|()| { + std::mem::take(&mut self.delivery).acked(); tracing::debug!(sequence = self.sequence, "finished group"); })); } @@ -7097,6 +7107,98 @@ mod serve_group_test { run.await.unwrap(); assert_eq!(*log.writes.lock().unwrap(), [0, 3, 0, 1, 0, 2, 3, 0, 0, 0x34, 1, 1, 2]); } + + /// Group streams report to their subscription's frontier: an acknowledged FIN moves + /// it, and a stream abandoned mid-frame drops only what the transport took. + #[moq_clock::test] + async fn group_streams_report_to_the_frontier() { + let registry = crate::stats::Registry::new(crate::stats::Config::new()); + let stats = registry.tier(crate::stats::Tier::default()).session("root"); + let broadcast = broadcast::Info::new().produce(); + let track = broadcast.create_track("test", crate::track::Info::timed()).unwrap(); + let sub = broadcast + .consume() + .with_stats(stats.egress("demo")) + .track("test") + .unwrap() + .subscribe(None) + .now_or_never() + .unwrap() + .unwrap(); + let subscription = |session: SinkSession| Subscription { + session: crate::driver::Transport::new(session), + id: 0, + track_name: "test".into(), + priority: PriorityQueue::default(), + publisher: 0, + group: None, + steer: kio::Producer::new(Steer::default()).consume(), + steer_seen: Steer::default(), + version: Version::Lite06, + stamp: Stamp::new(Version::Lite06, Some(crate::Timescale::MILLI)), + runtime: crate::time::sim(), + outcomes: Default::default(), + ietf: false, + }; + let serve = |mut ctx: Subscription, sequence: u64, group: group::Consumer| { + let handle = ctx.priority.insert(Priority::new(0, 0, sequence)); + let mut serve = GroupServe::new( + ctx.group_context(), + sequence, + 0, + false, + super::Rank::Queue(handle), + group, + ) + .delivered(sub.deliver()); + async move { + kio::wait(|waiter| serve.poll_open(&mut ctx.session, ctx.group.as_ref(), waiter)) + .await + .unwrap(); + kio::wait(move |waiter| serve.poll_serve(waiter)).await + } + }; + let ms = |ms| Timestamp::from_millis(ms).unwrap(); + + // Acknowledged: the frontier moves to its newest frame, 500ms. + let mut group = track.create_group(group::Info { sequence: 0 }).unwrap(); + group.write_frame(ms(0), b"a".as_slice()).unwrap(); + group.write_frame(ms(500), b"b".as_slice()).unwrap(); + group.finish().unwrap(); + serve(subscription(SinkSession::new(Log::default())), 0, group.consume()) + .await + .unwrap(); + + // The transport takes the headers and part of the payload, then holds the rest + // until the stream is abandoned. + let mut group = track.create_group(group::Info { sequence: 1 }).unwrap(); + group.write_frame(ms(1000), vec![0u8; 1000]).unwrap(); + let gate = kio::Producer::new(false); + let session = SinkSession::gated_uni_prefix(100, gate.consume()); + let log = session.log.clone(); + let mut abandoned = Box::pin(serve(subscription(session), 1, group.consume())); + assert!(futures::poll!(abandoned.as_mut()).is_pending()); + drop(abandoned); + let payload = log + .trail() + .into_iter() + .rev() + .find_map(|(_, sent)| match sent { + Sent::Write(bytes) => Some(bytes), + _ => None, + }) + .expect("a payload write"); + assert!((1..1000).contains(&payload.len()), "{}", payload.len()); + + let mut report = crate::stats::Report::default(); + registry.report(&mut report, crate::time::Instant::ZERO); + let row = report.traffic[0].publisher; + assert_eq!(row.dropped.groups, 1); + assert_eq!(row.dropped.bytes, payload.len() as u64); + assert_eq!(row.dropped.frames, 1); + // 1000ms produced against a 500ms frontier: the [500ms, 1s) bucket. + assert_eq!(row.lag.buckets(), &[0, 0, 0, 0, 1002, 0, 0, 0]); + } } #[cfg(test)] diff --git a/rs/moq-net/src/lite/test_transport.rs b/rs/moq-net/src/lite/test_transport.rs index 81d78ccbcf..bb3c0b5b42 100644 --- a/rs/moq-net/src/lite/test_transport.rs +++ b/rs/moq-net/src/lite/test_transport.rs @@ -607,6 +607,8 @@ pub struct SinkSession { bi_open_park: kio::Park, /// Set by [`Self::gated_uni`] to hold unidirectional stream writes. uni_gate: Option>, + /// Set by [`Self::gated_uni_prefix`]. + uni_prefix: usize, /// Set by [`Self::gated_open_uni`] to withhold unidirectional stream credit. uni_open_gate: Option>, uni_open_park: kio::Park, @@ -717,6 +719,15 @@ impl SinkSession { } } + /// Write this prefix on each unidirectional stream before holding its remaining + /// bytes, so the write that crosses it is accepted only in part. + pub fn gated_uni_prefix(prefix: usize, gate: kio::Consumer) -> Self { + Self { + uni_prefix: prefix, + ..Self::gated_uni(gate) + } + } + /// Hold unidirectional stream opens until `gate` grants stream credit. pub fn gated_open_uni(gate: kio::Consumer) -> Self { Self::default().with_open_uni_gate(gate) @@ -802,6 +813,7 @@ impl poll::Session for SinkSession { Some(gate) => SinkSend::gated(self.log.clone(), gate.clone()), None => SinkSend::new(self.log.clone()), }; + send.prefix = self.uni_prefix; send.unacked_fin = self.unacked_fin; Poll::Ready(Ok(send.opened(options))) } diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index 12474e48c4..6a86a6bbdd 100644 --- a/rs/moq-net/src/model/broadcast.rs +++ b/rs/moq-net/src/model/broadcast.rs @@ -859,7 +859,7 @@ impl Clone for Consumer { impl Consumer { /// Attach an egress stats scope, inherited by the tracks subscribed through this /// handle. Set by a tagged `origin::Consumer` at the broadcast handoff. - pub(super) fn with_stats(mut self, scope: stats::Scope) -> Self { + pub(crate) fn with_stats(mut self, scope: stats::Scope) -> Self { self.stats = scope; self } diff --git a/rs/moq-net/src/model/cache.rs b/rs/moq-net/src/model/cache.rs index 17c54841b8..7337281cc4 100644 --- a/rs/moq-net/src/model/cache.rs +++ b/rs/moq-net/src/model/cache.rs @@ -714,6 +714,11 @@ pub(super) struct Track { // What the frames this track's groups are still receiving may allocate up front, // shared with every track minted from the same origin handle. frames: super::frame::Budget, + + // What the track's groups have produced, for the egress lag sampler. Here because + // this account is the one per-track handle every group's frame writes already + // follow. Shared so a sampler still reads the last bytes after the track is gone. + produced: Arc, } /// The greatest location a track has written or its source reported, followed by the track @@ -819,6 +824,7 @@ impl Track { wakes: Default::default(), aborts: OnceLock::new(), largest: Default::default(), + produced: Default::default(), }); if let Some(key) = track.pool.register(&track) { let _ = track.sweep.set(key); @@ -831,6 +837,11 @@ impl Track { &self.pool } + /// What this track's groups have produced. + pub(super) fn produced(&self) -> &Arc { + &self.produced + } + /// What this track's frames may allocate up front. pub(super) fn frames(&self) -> &super::frame::Budget { &self.frames diff --git a/rs/moq-net/src/model/frame.rs b/rs/moq-net/src/model/frame.rs index d14efe8e8c..1ad300f8e1 100644 --- a/rs/moq-net/src/model/frame.rs +++ b/rs/moq-net/src/model/frame.rs @@ -749,8 +749,9 @@ pub struct Producer { // A receive handler's cancellation cleanup, carried across polls. cancel_on_drop: bool, // Ingress payload meter, inherited from the parent group. Counts each written - // chunk's bytes. Empty (no-op) for an untagged group. - stats: stats::Meter, + // chunk's bytes, and what the track produced. Empty (no-op) for an untagged group; + // `None` for a frame its compressed group deflates on commit, which counts both then. + stats: Option, // The frame's share of the origin's [`Budget`], if it got one, held until it ends. _reserved: Option, } @@ -771,13 +772,13 @@ impl Producer { info, done: false, cancel_on_drop: false, - stats: stats::Meter::default(), + stats: Some(stats::Meter::default()), _reserved: reserved, } } /// Attach the parent group's ingress meter, so written chunks bump `bytes`. - pub(super) fn with_meter(mut self, meter: stats::Meter) -> Self { + pub(super) fn with_meter(mut self, meter: Option) -> Self { self.stats = meter; self } @@ -833,7 +834,10 @@ impl Producer { return Err(Error::WrongSize); } // Ingress payload: count the chunk's bytes as they're written. - self.stats.bytes(len as u64); + if let Some(stats) = &self.stats { + stats.bytes(len as u64); + self.group().produced(self.info.timestamp, len as u64); + } // Fast path: a single whole-frame write keeps the caller's allocation. if len == self.buf.size() && self.buf.written(Ordering::Acquire) == 0 { match self.buf.try_set_bytes(chunk.into_bytes()) { diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index d4a703a212..ec7e9d8ccd 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -1292,9 +1292,16 @@ impl Producer { // Ingress payload: one whole frame written. self.stats.frames(1); self.stats.bytes(size); + self.produced(timestamp, size); Ok(()) } + /// Record `bytes` of a frame at `timestamp` toward the lag sampler, whether or not + /// this group is tagged; see [`crate::stats::Produced`]. + pub(super) fn produced(&self, timestamp: Option, bytes: u64) { + self.handle.cache.produced().record(timestamp, bytes); + } + /// Write a whole batch of frames at once, draining `frames`. /// /// One lock covers the batch, so an ingest with several frames in hand pays the @@ -1372,6 +1379,7 @@ impl Producer { state.cache += size; state.stamp(frame.timestamp); latest = frame.timestamp; + self.handle.cache.produced().record(frame.timestamp, size); let slots = state.push(Entry::plain(frame.into(), inflated)); now = state.charge.add(size + slots); } @@ -1440,13 +1448,13 @@ impl Producer { Ok(frame::Producer::new(self, buf, info, reserved).with_meter(meter)) } - /// The meter a streamed frame's chunks bump: none for a plain frame a compressed group - /// compresses on commit, which counts its stored slice then, as `bytes` counts payload - /// as stored. - fn chunk_meter(&self, inflated: Option) -> stats::Meter { + /// The meter a streamed frame's chunks bump, with what the track produced: none for a + /// plain frame a compressed group compresses on commit, which counts its stored slice + /// then, as both count payload as stored. + fn chunk_meter(&self, inflated: Option) -> Option { match self.handle.track.compress && inflated.is_none() { - true => stats::Meter::default(), - false => self.stats.clone(), + true => None, + false => Some(self.stats.clone()), } } @@ -1538,8 +1546,10 @@ impl Producer { } // Ingress payload: one frame opened; its bytes are counted per chunk as the - // producer writes them. + // producer writes them. What the track produced too: the header places the frame in time, and + // its payload weighs only as it arrives. self.stats.frames(1); + self.produced(timestamp, 0); Ok(frame::Info { size: frame.size, @@ -1578,7 +1588,7 @@ impl Producer { let partial = state.partial.take(); let mut frame = frame; let mut inflated = 0; - if partial.as_ref().is_some_and(|partial| partial.scratch) { + if let Some(partial) = partial.as_ref().filter(|partial| partial.scratch) { let flate = state.flate.as_mut().expect("scratch only opens on a compressed group"); inflated = frame.payload.len() as u64; frame.payload = flate.deflate(&frame.payload.contiguous(0)).into(); @@ -1591,6 +1601,7 @@ impl Producer { state.cache += size; now = state.charge.add(size); self.stats.bytes(size); + self.produced(partial.timestamp, size); } else if let Some(size) = partial.as_ref().and_then(|partial| partial.inflated) { // Admitted against the budget when it opened. inflated = size; diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 96393b568c..9693dc696c 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -7878,7 +7878,7 @@ mod tests { project.request_broadcast(".svc/foo", None).await.expect("resolves"); let mut report = stats::Report::default(); - registry.report(&mut report); + registry.report(&mut report, crate::time::Instant::ZERO); let paths: Vec<_> = report .traffic .iter() diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 33bb63def2..bf3e0e7722 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -267,8 +267,9 @@ impl Consumer { } } - /// Subscribe to the logical track with the reader's own preferences. - pub(super) fn subscribe(&self, subscription: track::Preferences) -> Subscriber { + /// Subscribe to the logical track with the reader's own preferences, weighing what + /// each copy may still deliver against `frontier`. + pub(super) fn subscribe(&self, subscription: track::Preferences, frontier: stats::Frontier) -> Subscriber { let mirrored = subscription.read().clone(); let window = Window::new(&mirrored.ranges, None, &Unheld).expect("an unheld hop refuses nothing"); // An absolute floor holds from the start, as on a plain cursor; a relative one @@ -292,6 +293,7 @@ impl Consumer { peeked: None, declared: None, refused: false, + frontier, }) }); // Subscribed to the serving copy at once, so groups it has now are the reader's @@ -414,6 +416,12 @@ struct Copy { /// Held by every group handed out from this copy, so a replaced copy stays subscribed /// while one is still read from it. lease: Arc<()>, + /// What the track instance the copy subscribed to produces. + produced: Arc, + /// Weighs [`Self::produced`] against the reader's frontier while the copy may still + /// deliver a new group: from its subscription until [`Self::until`] bounds it. Bound + /// to the copy, so one let go stops weighing however it leaves. + feed: stats::Feed, } // Boxed: a subscriber is twice a pending answer, and a copy holds one for its whole life. @@ -857,6 +865,8 @@ struct Reader { /// Runs the serving copy gave up that the reader has yet to report; see /// [`Reader::poll_drop`]. drops: RangeMap, + /// The subscription's acknowledged frontier, which each copy's [`Copy::feed`] feeds. + frontier: stats::Frontier, } impl Reader { @@ -997,6 +1007,7 @@ impl Reader { copy.generation = generation; copy.replaced = false; copy.until = None; + copy.feed = self.frontier.feed(©.produced); copy.since = None; copy.done = None; copy.datagrams_done = None; @@ -1018,9 +1029,13 @@ impl Reader { // The wire request carries the floor from the first message: some sessions read // the demand only once. let wire = self.mirrored.clamped(floor, None, &resolved); + let sub = track.subscribe(wire); + let produced = sub.produced(); self.copies.push(Copy { generation, - sub: Sub::Pending(track.subscribe(wire)), + sub: Sub::Pending(sub), + feed: self.frontier.feed(&produced), + produced, track, floor, from: self.mirrored.ranges.clone(), @@ -1164,6 +1179,7 @@ impl Reader { for copy in &mut self.copies { if copy.replaced && copy.until.is_none() { copy.until = Some(until); + copy.feed = stats::Feed::default(); copy.since = Some(copy.track.now()); copy.update(mirrored); } @@ -1184,6 +1200,10 @@ impl Reader { for copy in &mut self.copies[..index] { if copy.replaced && copy.until.is_none() { copy.until = Some(bound); + // What the copy produces past the bound never reaches this reader. Frames + // still written into a group it already handed out go unweighed too: an + // approximation that keeps per-group accounting off the frame path. + copy.feed = stats::Feed::default(); copy.since = Some(copy.track.now()); copy.update(&self.mirrored); } @@ -1686,9 +1706,13 @@ impl Reader { let resolved = self.window().resolved(); let floor = Some(lowered.start); let wire = self.mirrored.clamped(floor, Some(lowered.end), &resolved); + let sub = track.subscribe(wire); let copy = Copy { generation, - sub: Sub::Pending(track.subscribe(wire)), + produced: sub.produced(), + sub: Sub::Pending(sub), + // Bounded from the start, so never weighed, like any bounded copy. + feed: stats::Feed::default(), floor, from: self.mirrored.ranges.clone(), resolved, @@ -4188,7 +4212,7 @@ mod test { routes.serve(a.consume()); let sub = routes .consume() - .subscribe(kio::Producer::new(Subscription::default()).into()); + .subscribe(kio::Producer::new(Subscription::default()).into(), Default::default()); let mut reader = sub.reader.lock().unwrap(); let copy = &mut reader.copies[0]; copy.floor = Some(0); @@ -4349,7 +4373,9 @@ mod test { let subscription = Subscription::default() .with_max_delay(Duration::from_secs(30)) .with_ranges([range]); - let sub = routes.consume().subscribe(kio::Producer::new(subscription).into()); + let sub = routes + .consume() + .subscribe(kio::Producer::new(subscription).into(), Default::default()); let mut reader = sub.reader.lock().unwrap(); let (groups, mirrored) = (reader.groups, reader.mirrored.clone()); let old = &mut reader.copies[0]; @@ -4380,7 +4406,7 @@ mod test { let a = copy(); routes.serve(a.consume()); let subscription = kio::Producer::new(Subscription::default()); - let sub = routes.consume().subscribe(subscription.into()); + let sub = routes.consume().subscribe(subscription.into(), Default::default()); let mut reader = sub.reader.lock().unwrap(); let old = &mut reader.copies[0]; old.until = Some(last); @@ -6116,4 +6142,168 @@ mod test { ], ); } + + /// A stats-tagged reader of `logical`, as a session serving a viewer holds one. + fn tagged(logical: &track::Producer, session: &stats::Session) -> track::Subscriber { + logical + .consume() + .with_stats(session.egress("demo")) + .subscribe(Subscription::default().with_max_delay(Duration::from_secs(10))) + .now_or_never() + .unwrap() + .unwrap() + } + + /// One group of one frame at `sequence`, half a second of media apart. + fn publish(track: &track::Producer, sequence: u64, payload: &[u8]) { + let mut group = track.create_group(group::Info { sequence }).unwrap(); + group.write_frame(ts(sequence * 500), payload).unwrap(); + group.finish().unwrap(); + } + + /// The egress lag one report reads. Every report samples at the same instant, so + /// only weight and media lag move. + fn lag(registry: &stats::Registry) -> stats::Histogram { + let mut report = stats::Report::default(); + registry.report(&mut report, crate::time::Instant::ZERO); + report.traffic[0].publisher.lag + } + + /// A route-fed track's lag weighs what its copies may still deliver: a replaced copy + /// until the new route hands out its first group, then only the new route. + #[test] + fn lag_weighs_the_serving_copy() { + let registry = stats::Registry::new(stats::Config::new()); + let session = registry.tier(stats::Tier::default()).session("root"); + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = tagged(&logical, &session); + drain(&mut sub, false); + publish(&a, 0, b"a0"); + assert_eq!(lag(®istry).total(), 2); + + // B has nothing yet, so A may still deliver: its group counts, 500ms behind the + // frame the subscription started from. + let b = copy(); + routes.serve(b.consume()); + publish(&a, 1, b"a1"); + drain(&mut sub, false); + let pending = lag(®istry); + assert_eq!(pending.total(), 4); + assert_eq!(pending.buckets()[4], 2); + + // B's first group bounds A, whose later groups never reach this reader. + publish(&b, 2, b"b2"); + drain(&mut sub, false); + assert_eq!(lag(®istry).total(), 6); + publish(&a, 3, b"a3"); + assert_eq!(lag(®istry).total(), 6); + } + + /// A copy served across a no-route gap stops counting once the new route delivers, + /// even while a group the reader holds keeps it alive. + #[test] + fn lag_stops_weighing_a_copy_replaced_across_a_gap() { + let registry = stats::Registry::new(stats::Config::new()); + let session = registry.tier(stats::Tier::default()).session("root"); + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = tagged(&logical, &session); + drain(&mut sub, false); + publish(&a, 0, b"aa"); + let held = recv(&mut sub); + + routes.park(); + drain(&mut sub, false); + let b = copy(); + routes.serve(b.consume()); + drain(&mut sub, false); + publish(&b, 1, b"bbbb"); + drain(&mut sub, false); + assert_eq!(lag(®istry).total(), 6); + + publish(&a, 2, b"cccccc"); + assert_eq!(lag(®istry).total(), 6); + drop(held); + } + + /// A replaced copy bounded at the serving copy's clean end stops counting, even while + /// it may still deliver a group below that end. + #[test] + fn lag_stops_weighing_a_copy_bounded_at_a_clean_end() { + let registry = stats::Registry::new(stats::Config::new()); + let session = registry.tier(stats::Tier::default()).session("root"); + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = tagged(&logical, &session); + drain(&mut sub, false); + publish(&a, 0, b"a"); + publish(&a, 2, b"a"); + assert_eq!(drain(&mut sub, false), [0, 2]); + + // The new route replays only the group already handed out, and ends there. + let b = copy(); + publish(&b, 2, b"b"); + b.finish().unwrap(); + routes.serve(b.consume()); + assert!(drain(&mut sub, false).is_empty(), "group 1 may still land"); + assert_eq!(copies(&routes), 2); + assert_eq!(lag(®istry).total(), 2); + + publish(&a, 3, b"cccccc"); + assert_eq!(lag(®istry).total(), 2); + } + + /// A replaced copy let go before anything bounds it stops counting as it goes. + #[test] + fn lag_stops_weighing_a_copy_let_go_unbounded() { + let registry = stats::Registry::new(stats::Config::new()); + let session = registry.tier(stats::Tier::default()).session("root"); + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = tagged(&logical, &session); + publish(&a, 0, b"aa"); + assert_eq!(lag(®istry).total(), 2); + + // A reader that only reads datagrams keeps no replaced copy. + routes.serve(copy().consume()); + assert!(sub.poll_recv_datagram(&kio::Waiter::noop()).is_pending()); + publish(&a, 1, b"bbbb"); + assert_eq!(lag(®istry).total(), 2); + } + + /// A route that comes back after another delivered is weighed again. + #[test] + fn lag_weighs_a_route_that_comes_back() { + let registry = stats::Registry::new(stats::Config::new()); + let session = registry.tier(stats::Tier::default()).session("root"); + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = tagged(&logical, &session); + drain(&mut sub, false); + publish(&a, 0, b"aa"); + let held = recv(&mut sub); + + let b = copy(); + routes.serve(b.consume()); + publish(&b, 1, b"bb"); + drain(&mut sub, false); + assert_eq!(lag(®istry).total(), 4); + + routes.serve(a.consume()); + drain(&mut sub, false); + publish(&a, 2, b"cccc"); + assert_eq!(lag(®istry).total(), 8); + drop(held); + } } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 6162e2610e..fd6805defb 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -3350,7 +3350,7 @@ impl Producer { delivered: Delivered::default(), // A producer-side (in-process) subscribe is not egress: stay untagged. stats: stats::Scope::default(), - _stats_sub: stats::Subscription::default(), + stats_sub: stats::Subscription::default(), largest: Reported::default(), } } @@ -4753,6 +4753,11 @@ pub struct Subscribing { } impl Subscribing { + /// What the subscribed track instance produces, which the lag sampler weighs. + pub(super) fn produced(&self) -> Arc { + self.state.read().cache.produced().clone() + } + /// Query the same track with this subscription already registered. pub(super) fn query(&self) -> Querying { Querying { @@ -4771,13 +4776,18 @@ impl Subscribing { .map_err(|e| e.abort.clone().unwrap_or(Error::Dropped))??; refuse_untimed(&info, &self.subscription.read())?; + let mut stats_sub = self.stats.subscribe(); let resume = self.state.read().routes.clone(); + // A route-fed track weighs whichever copy serves it; a plain one, its own. let inner = match resume { Some(resume) => Inner::Resume( - Box::new(resume.subscribe(self.subscription.clone())), + Box::new(resume.subscribe(self.subscription.clone(), stats_sub.frontier().clone())), self.state.clone(), ), - None => Inner::Plain(Cursor::new(self.state.clone(), self.subscription.clone())), + None => { + stats_sub.feed(&self.produced()); + Inner::Plain(Cursor::new(self.state.clone(), self.subscription.clone())) + } }; Poll::Ready(Ok(Subscriber { name: self.name.clone(), @@ -4786,7 +4796,7 @@ impl Subscribing { inner, delivered: Delivered::default(), stats: self.stats.clone(), - _stats_sub: self.stats.subscribe(), + stats_sub, largest: Reported::default(), })) } @@ -5216,8 +5226,10 @@ pub struct Subscriber { // (no-op) for an untagged track. stats: stats::Scope, // The subscription guard: bumps `subscriptions` (and the egress viewer refcount) - // while held, closing them on drop. Empty (no-op) for an untagged track. - _stats_sub: stats::Subscription, + // while held, closing them on drop, and holds the egress lag frontier with a plain + // track's feed into it. Empty (no-op) for an untagged track. + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + stats_sub: stats::Subscription, largest: Reported, } @@ -6739,6 +6751,13 @@ impl Subscriber { } } + /// Track one group stream written toward the peer against this subscription's + /// acknowledged frontier. A no-op for an untagged subscriber. + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + pub(crate) fn deliver(&self) -> stats::Delivery { + self.stats_sub.deliver() + } + /// Create a handle for updating this subscriber's delivery preferences. pub fn control(&self) -> Control { let timed = self.info.timescale.is_some(); diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index d283a20e9b..2851f16131 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -62,6 +62,22 @@ //! once, on the side of the reader whose read inflated it. A received compressed //! datagram inflates on arrival, counted on the receiving side. Kept separate from //! `bytes`, which counts payload as stored or sent. +//! * `lag` (egress only): a byte-weighted [`Histogram`] of how far behind each +//! subscription's acknowledged frontier is. The frontier is the newest frame +//! timestamp the peer acknowledged: it starts at the track's newest frame when the +//! subscription opens and advances when a group stream's FIN is acknowledged. +//! [`Registry::report`] is the sampler: each call weighs the bytes the subscription's +//! tracks produced since the previous call at `max(newest produced - frontier, +//! stall)`. The stall is the time, measured between the instants the calls are +//! given, that the tracks kept producing while the peer acknowledged no FIN; any +//! acknowledgement resets it. A paused source produces nothing, so it adds no +//! weight. A subscription that closes is weighed once more, at its lag then and its +//! stall as of the last call. +//! * `dropped` (egress only, a [`Content`]): content written toward a peer on a group +//! stream that was reset or abandoned before its FIN was acknowledged: the frames +//! it carried, the payload bytes the transport accepted, and one group. Nothing +//! inside a group is known to be acknowledged, so everything written counts. A +//! subset of `bytes` / `frames` / `groups`, which count it as handed over. //! * `requests` ([`Presence`]): cumulative request/response admissions under the //! label and tier at admission. [`Handle::request`] meters their traffic //! without starting or ending a connected session. @@ -139,12 +155,13 @@ use std::{ Arc, Mutex, MutexGuard, PoisonError, Weak, atomic::{AtomicU64, AtomicUsize, Ordering, fence}, }, + time::Duration, }; use kio::Lock; use serde::{Deserialize, Deserializer, Serialize, Serializer}; -use crate::{AsPath, PathOwned, Pattern, Patterns, util::SendCell}; +use crate::{AsPath, PathOwned, Pattern, Patterns, Timestamp, time::Instant, util::SendCell}; mod outcomes; pub use outcomes::{Failure, Failures, RefusalKind, Refusals}; @@ -190,6 +207,14 @@ pub(crate) struct Counters { // Content the drift budget gave up on before delivery. Disjoint from the // top-level payload counters, which count only what was handed over. stale: ContentCounters, + // Egress only: byte-weighted frontier lag, recorded by `sample`. + lag: HistogramCounters, + // Egress only: content written on a group stream that ended unacknowledged. A + // subset of the top-level payload counters, which count it as handed over. + dropped: ContentCounters, + // Egress only: the subscriptions `sample` walks, swept of closed ones there and + // whenever the list fills. + frontiers: Mutex>>, } /// Atomic backing for one [`Content`] readout. @@ -242,6 +267,8 @@ impl Counters { let datagrams = self.datagrams.load(Ordering::Relaxed); let inflated = self.inflated.load(Ordering::Relaxed); let stale = self.stale.snapshot(); + let lag = self.lag.snapshot(); + let dropped = self.dropped.snapshot(); Traffic { announces_started, announces_ended, @@ -257,13 +284,156 @@ impl Counters { datagrams, inflated, stale, + lag, + dropped, + } + } + + /// Take one lag sample at `now` from every live egress subscription, dropping the closed. + fn sample(&self, now: Instant) { + let mut frontiers = self.frontiers.lock().unwrap_or_else(PoisonError::into_inner); + frontiers.retain(|frontier| { + let Some(frontier) = frontier.upgrade() else { + return false; + }; + frontier.sample(Some(now)); + true + }); + } +} + +/// Number of [`Histogram`] buckets: one below each edge, plus the open top bucket. +const BUCKETS: usize = Histogram::EDGES.len() + 1; + +/// A byte-weighted histogram over media time, cumulative like every [`Traffic`] counter. +/// +/// Each bucket counts the bytes whose sampled value fell below its edge in +/// [`Self::EDGES`] and at or above the edge before it; the last bucket is open +/// above the top edge. Buckets only grow, so the difference of two readouts is the +/// byte-weighted distribution over that interval. +/// +/// On the wire it is an object keyed by each bucket's upper edge (`"50ms"` .. +/// `"5s"`, then `"inf"`), with empty buckets omitted. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(from = "HistogramWire", into = "HistogramWire")] +pub struct Histogram { + buckets: [u64; BUCKETS], +} + +impl Histogram { + /// The exclusive upper edge of every bucket but the last. + pub const EDGES: [Duration; 7] = [ + Duration::from_millis(50), + Duration::from_millis(100), + Duration::from_millis(250), + Duration::from_millis(500), + Duration::from_secs(1), + Duration::from_secs(2), + Duration::from_secs(5), + ]; + + /// The bucket `value` falls in. + fn bucket(value: Duration) -> usize { + Self::EDGES.partition_point(|edge| *edge <= value) + } + + /// Bytes per bucket, lowest first; the last is open above the top edge. + pub fn buckets(&self) -> &[u64] { + &self.buckets + } + + /// Total bytes across every bucket, saturating. + pub fn total(&self) -> u64 { + self.buckets.iter().fold(0, |sum, bucket| sum.saturating_add(*bucket)) + } + + /// True when nothing has been recorded. + pub fn is_empty(&self) -> bool { + self.buckets.iter().all(|bucket| *bucket == 0) + } + + /// Fold another readout into this one, bucket by bucket, saturating: a merged + /// readout may come from a peer. + pub(crate) fn add(&mut self, other: &Histogram) { + for (bucket, other) in self.buckets.iter_mut().zip(other.buckets) { + *bucket = bucket.saturating_add(other); + } + } +} + +/// The wire shape of a [`Histogram`]: one key per bucket, named by its upper edge. +#[derive(Default, Serialize, Deserialize)] +#[serde(default)] +struct HistogramWire { + #[serde(rename = "50ms", skip_serializing_if = "is_zero")] + b50ms: u64, + #[serde(rename = "100ms", skip_serializing_if = "is_zero")] + b100ms: u64, + #[serde(rename = "250ms", skip_serializing_if = "is_zero")] + b250ms: u64, + #[serde(rename = "500ms", skip_serializing_if = "is_zero")] + b500ms: u64, + #[serde(rename = "1s", skip_serializing_if = "is_zero")] + b1s: u64, + #[serde(rename = "2s", skip_serializing_if = "is_zero")] + b2s: u64, + #[serde(rename = "5s", skip_serializing_if = "is_zero")] + b5s: u64, + #[serde(rename = "inf", skip_serializing_if = "is_zero")] + inf: u64, +} + +fn is_zero(value: &u64) -> bool { + *value == 0 +} + +impl From for Histogram { + fn from(w: HistogramWire) -> Self { + Self { + buckets: [w.b50ms, w.b100ms, w.b250ms, w.b500ms, w.b1s, w.b2s, w.b5s, w.inf], + } + } +} + +impl From for HistogramWire { + fn from(h: Histogram) -> Self { + let [b50ms, b100ms, b250ms, b500ms, b1s, b2s, b5s, inf] = h.buckets; + Self { + b50ms, + b100ms, + b250ms, + b500ms, + b1s, + b2s, + b5s, + inf, + } + } +} + +/// Atomic backing for one [`Histogram`] readout. +#[derive(Default, Debug)] +struct HistogramCounters { + buckets: [AtomicU64; BUCKETS], +} + +impl HistogramCounters { + fn record(&self, value: Duration, weight: u64) { + if weight > 0 { + self.buckets[Histogram::bucket(value)].fetch_add(weight, Ordering::Relaxed); + } + } + + fn snapshot(&self) -> Histogram { + Histogram { + buckets: std::array::from_fn(|i| self.buckets[i].load(Ordering::Relaxed)), } } } /// Payload-volume counters for content with the same delivery outcome. /// -/// This is the nested shape used by [`Traffic::stale`]. The successfully +/// This is the nested shape used by [`Traffic::stale`] and [`Traffic::dropped`]. The successfully /// delivered equivalents remain as top-level [`Traffic`] fields for wire /// compatibility with existing stats consumers. #[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] @@ -288,6 +458,11 @@ impl Content { self.groups += other.groups; self.datagrams += other.datagrams; } + + /// True when nothing has been counted. + fn is_empty(&self) -> bool { + *self == Self::default() + } } /// Per-(tier, label) session gauge. One of these is shared (via `Arc`) by every @@ -492,6 +667,15 @@ pub struct Traffic { /// counters: skipped content is never handed over. A steady rate here means /// subscribers are consistently behind the live edge. pub stale: Content, + /// Egress only: how far behind the peer's acknowledged frontier was, weighted by + /// the bytes the track produced while it was. Sampled on each + /// [`Registry::report`]; see the module docs. + pub lag: Histogram, + /// Egress only: content written on group streams that were reset or abandoned + /// before the peer acknowledged them. A subset of the top-level payload counters, + /// which count it as handed over. A datagram has no acknowledgement to miss, so + /// `datagrams` stays zero. + pub dropped: Content, } /// One spelling of a counter edge on the wire: absent, or a present integer. @@ -535,6 +719,10 @@ struct TrafficSer { datagrams: u64, inflated: u64, stale: Content, + #[serde(skip_serializing_if = "Histogram::is_empty")] + lag: Histogram, + #[serde(skip_serializing_if = "Content::is_empty")] + dropped: Content, } impl From for TrafficSer { @@ -560,6 +748,8 @@ impl From for TrafficSer { datagrams: t.datagrams, inflated: t.inflated, stale: t.stale, + lag: t.lag, + dropped: t.dropped, } } } @@ -587,6 +777,8 @@ struct TrafficDe { datagrams: u64, inflated: u64, stale: Content, + lag: Histogram, + dropped: Content, } impl From for Traffic { @@ -606,6 +798,8 @@ impl From for Traffic { datagrams: d.datagrams, inflated: d.inflated, stale: d.stale, + lag: d.lag, + dropped: d.dropped, } } } @@ -639,6 +833,8 @@ impl Traffic { self.datagrams += other.datagrams; self.inflated += other.inflated; self.stale.add(other.stale); + self.lag.add(&other.lag); + self.dropped.add(other.dropped); } /// Distinct sessions currently subscribed (viewers on the egress side). @@ -1289,7 +1485,11 @@ impl Registry { /// an interval sees the closing readout exactly once. A pruned path that /// sees traffic again restarts from zero; see the module docs on counter /// resets. Leaves the report empty for a disabled registry. - pub fn report(&self, report: &mut Report) { + /// + /// Each call is also one lag sample of every egress subscription, taken at + /// `now`: the interval since the previous call's instant is the one a stalled + /// frontier is charged; see the module docs. + pub fn report(&self, report: &mut Report, now: Instant) { self.collect_wire(); report.traffic.clear(); report.sessions.clear(); @@ -1305,7 +1505,10 @@ impl Registry { let free = Arc::strong_count(entry) == 1; let mut tiers = entry.tiers.lock().unwrap_or_else(PoisonError::into_inner); tiers.retain(|tier, counters| { + // A live frontier holds its counters, so a closed entry has none left to + // sample: each recorded its last weight as it closed. let closed = free && Arc::strong_count(counters) == 1; + counters.publisher.sample(now); let publisher = counters.publisher.snapshot(); let subscriber = counters.subscriber.snapshot(); report.traffic.push(TrafficEntry { @@ -2004,22 +2207,26 @@ impl Scope { pub(crate) fn subscribe(&self) -> Subscription { let counters = self.counters(); let mut viewer = None; + let mut frontier = Frontier::default(); if let Some(counters) = &counters { self.side .counters(counters) .subscriptions_started .fetch_add(1, Ordering::Relaxed); - // Viewer refcount is egress-only: `broadcasts_started` counts distinct sessions - // watching a broadcast. + // The viewer refcount and lag are egress-only: `broadcasts_started` counts + // distinct sessions watching a broadcast. if matches!(self.side, Side::Publisher) { self.session.viewer_open(&self.path, counters); viewer = Some((self.session.clone(), self.path.clone())); + frontier = Frontier::open(counters); } } Subscription { counters, side: self.side, viewer, + feed: Feed::default(), + frontier, } } @@ -2059,6 +2266,32 @@ pub(crate) struct Subscription { side: Side, /// `Some((session, path))` on the egress side, to release the viewer refcount. viewer: Option<(Session, PathOwned)>, + /// The plain track this subscription reads, weighed against [`Self::frontier`]. + /// Declared first so it folds into the frontier before the frontier's last sample. + feed: Feed, + /// The peer's acknowledged frontier on the egress side, empty otherwise. + frontier: Frontier, +} + +impl Subscription { + /// This subscription's acknowledged frontier, empty (no-op) off the egress side. + pub(crate) fn frontier(&self) -> &Frontier { + &self.frontier + } + + /// Weigh what `produced` produces against this subscription's frontier for as long + /// as the guard lives: the plain track it reads. A route-fed track's copies hold + /// their own [`Feed`]s instead. + pub(crate) fn feed(&mut self, produced: &Arc) { + self.feed = self.frontier.feed(produced); + } + + /// Track one group stream written toward the peer against this subscription's + /// frontier. A no-op off the egress side. + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + pub(crate) fn deliver(&self) -> Delivery { + self.frontier.deliver() + } } impl Drop for Subscription { @@ -2095,6 +2328,362 @@ impl Drop for Announce { } } +/// What one track has produced: the lag sampler's weight and the edge it measures a +/// subscription's frontier against. Recorded beside the ingress meter on every frame +/// write, whether or not the track is tagged, since its readers may be. An untimed +/// frame has no place in media time, so it records nothing. +#[derive(Default, Debug)] +pub(crate) struct Produced { + // Media-time nanoseconds plus one, so zero means nothing produced yet. + first: AtomicU64, + newest: AtomicU64, + // Cumulative payload bytes of timed frames. + bytes: AtomicU64, +} + +impl Produced { + /// Record `bytes` of a frame at `timestamp`, nothing for an untimed one. + pub(crate) fn record(&self, timestamp: Option, bytes: u64) { + let Some(timestamp) = timestamp else { + return; + }; + let at = u64::try_from(timestamp.as_nanos()) + .unwrap_or(u64::MAX) + .saturating_add(1); + if self.first.load(Ordering::Relaxed) == 0 { + let _ = self.first.compare_exchange(0, at, Ordering::Relaxed, Ordering::Relaxed); + } + // A max, not a store: B-frames present out of order within a group. + self.newest.fetch_max(at, Ordering::Relaxed); + // Release: a sampler that acquires these bytes also sees the edges above. + self.bytes.fetch_add(bytes, Ordering::Release); + } + + fn load(cell: &AtomicU64) -> Option { + cell.load(Ordering::Relaxed).checked_sub(1).map(Duration::from_nanos) + } +} + +/// One egress subscription's acknowledged frontier, shared by the wire's group serve +/// tasks (which advance it through [`Delivery`]) and the lag sampler (which reads it on +/// every [`Registry::report`]). Empty (no-op) off the egress side. +#[derive(Clone, Default)] +pub(crate) struct Frontier(Option>); + +struct FrontierInner { + counters: Arc, + state: Mutex, +} + +#[derive(Default)] +struct FrontierState { + /// The newest timestamp the peer acknowledged, seeded with the newest frame of the + /// first track fed, so a backlog the subscription deliberately starts behind + /// never reads as lag. `None` until either exists. + acked: Option, + /// The peer acknowledged a FIN since the last sample. + moved: bool, + /// How long the sources kept producing while the peer acknowledged nothing, as of + /// the last sample. + stalled: Duration, + /// The last sample's instant, `None` before the first. + sampled: Option, + /// The tracks feeding the subscription: its own, or each route copy that may still + /// deliver to it. + sources: Vec, + /// What retired sources produced since their last sample, each at the edge it had, + /// so a sample records it even before a replacement has a frame, and never at the + /// replacement's lag. At most [`MAX_FOLDED`] entries. + folded: Vec, +} + +/// The most retired sources a frontier keeps apart between samples. A peer can retire +/// copies as fast as it moves its route, so past this the latest share one entry. +const MAX_FOLDED: usize = 8; + +/// Bytes a retired source produced before it stopped feeding the subscription, +/// weighed at the edge it had then. +struct Folded { + weight: u64, + newest: Duration, + first: Duration, +} + +impl Folded { + /// Merge `other` in at the later edge, so a shared entry's lag reads high, never low. + fn absorb(&mut self, other: Folded) { + self.weight = self.weight.saturating_add(other.weight); + self.newest = self.newest.max(other.newest); + self.first = self.first.min(other.first); + } +} + +/// A track feeding a [`Frontier`], with the produced bytes already sampled. +struct Source { + produced: Arc, + sampled: u64, + /// The [`Feed`] guards naming it; it leaves with the last one. + feeds: usize, +} + +impl Frontier { + fn open(counters: &Arc) -> Self { + let inner = Arc::new(FrontierInner { + counters: counters.clone(), + state: Default::default(), + }); + let mut frontiers = counters + .publisher + .frontiers + .lock() + .unwrap_or_else(PoisonError::into_inner); + // Sweep the closed once the list fills, so a registry reported rarely or never + // still tracks the live count. + if frontiers.len() == frontiers.capacity() { + frontiers.retain(|frontier| frontier.strong_count() > 0); + let live = frontiers.len(); + frontiers.reserve(live.max(1)); + } + frontiers.push(Arc::downgrade(&inner)); + drop(frontiers); + Self(Some(inner)) + } + + /// Weigh what `produced` produces against this frontier until the returned guard + /// drops. The first track fed seeds the frontier with its newest frame. + pub(crate) fn feed(&self, produced: &Arc) -> Feed { + let Some(inner) = &self.0 else { + return Feed::default(); + }; + let mut state = inner.state(); + match state + .sources + .iter_mut() + .find(|source| Arc::ptr_eq(&source.produced, produced)) + { + Some(source) => source.feeds += 1, + None => { + if state.acked.is_none() { + state.acked = Produced::load(&produced.newest); + } + state.sources.push(Source { + produced: produced.clone(), + sampled: produced.bytes.load(Ordering::Acquire), + feeds: 1, + }); + } + } + Feed { + frontier: Arc::downgrade(inner), + produced: Some(produced.clone()), + } + } + + /// Start tracking one group stream written toward the peer. + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + pub(crate) fn deliver(&self) -> Delivery { + Delivery { + frontier: self.0.clone(), + ..Default::default() + } + } +} + +impl FrontierInner { + fn state(&self) -> MutexGuard<'_, FrontierState> { + self.state.lock().unwrap_or_else(PoisonError::into_inner) + } + + /// Drop one [`Feed`] on `produced`. The last one retires the source, keeping + /// what it produced since its last sample at the edge it has now. + fn unfeed(&self, produced: &Arc) { + let mut state = self.state(); + let Some(index) = state + .sources + .iter() + .position(|source| Arc::ptr_eq(&source.produced, produced)) + else { + return; + }; + let source = &mut state.sources[index]; + source.feeds -= 1; + if source.feeds > 0 { + return; + } + let source = state.sources.swap_remove(index); + // Acquire pairs with `Produced::record`, so the edges cover these bytes. + let weight = produced.bytes.load(Ordering::Acquire).saturating_sub(source.sampled); + if weight > 0 + && let Some(newest) = Produced::load(&produced.newest) + && let Some(first) = Produced::load(&produced.first) + { + let item = Folded { weight, newest, first }; + if state.folded.len() < MAX_FOLDED { + state.folded.push(item); + } else if let Some(last) = state.folded.last_mut() { + last.absorb(item); + } + } + } + + /// Weigh what the sources produced since the last sample. `now` is the sampler's + /// instant, or `None` for the final sample as the subscription closes, which + /// charges no further stall: nothing supplies the time it closed at. + fn sample(&self, now: Option) { + let mut state = self.state(); + let folded = std::mem::take(&mut state.folded); + + // Any acknowledgement resets the stall, and an interval in which the sources + // produced while none arrived adds to it. Judged by bytes rather than by + // timestamps, so a publisher whose timestamps step back cannot hide a viewer + // that stopped acknowledging. + let produced = !folded.is_empty() + || (state.sources.iter()).any(|source| source.produced.bytes.load(Ordering::Relaxed) > source.sampled); + if state.moved { + state.stalled = Duration::ZERO; + } + if let Some(now) = now { + if !state.moved + && produced && let Some(at) = state.sampled + { + state.stalled = state.stalled.saturating_add(now.saturating_duration_since(at)); + } + state.moved = false; + state.sampled = Some(now); + } + + // Opened before anything produced: the earliest frame is where it started. + if state.acked.is_none() { + let first = (state.sources.iter()).filter_map(|source| Produced::load(&source.produced.first)); + state.acked = folded.iter().map(|item| item.first).chain(first).min(); + } + // Nothing timed produced yet: whatever lands from here is weighed next time. + let Some(acked) = state.acked else { + return; + }; + let stalled = state.stalled; + let histogram = &self.counters.publisher.lag; + let weigh = |edge: Duration, weight: u64| histogram.record(edge.saturating_sub(acked).max(stalled), weight); + // One source's bytes are not another's, a retired copy's nor a live one's: each + // is weighed at its own edge. + for item in &folded { + weigh(item.newest, item.weight); + } + for source in &mut state.sources { + let produced = &source.produced; + // Acquire pairs with `Produced::record`: the edge read below covers these bytes. + let bytes = produced.bytes.load(Ordering::Acquire); + let weight = bytes.saturating_sub(source.sampled); + source.sampled = bytes; + if weight > 0 + && let Some(newest) = Produced::load(&produced.newest) + { + weigh(newest, weight); + } + } + } +} + +impl Drop for FrontierInner { + /// The last strong reference (the subscription guard or an in-flight [`Delivery`]) + /// is gone, so no sample will find this frontier again: weigh what the sources + /// produced since the last one now. + fn drop(&mut self) { + self.sample(None); + } +} + +/// Keeps one track weighed against a [`Frontier`]; see [`Frontier::feed`]. Dropping it +/// keeps what the track produced so far, at the edge it had, and stops weighing the rest. +#[derive(Default)] +#[must_use = "dropping the guard stops weighing the track"] +pub(crate) struct Feed { + frontier: Weak, + produced: Option>, +} + +impl Drop for Feed { + fn drop(&mut self) { + if let Some(produced) = self.produced.take() + && let Some(inner) = self.frontier.upgrade() + { + inner.unfeed(&produced); + } + } +} + +/// One group stream written toward a peer, reported to its subscription's +/// [`Frontier`]: an acknowledged FIN advances the frontier, and anything else that ends +/// the stream counts what it wrote as dropped. Empty (no-op) off the egress side. +#[derive(Default)] +#[must_use = "an unacknowledged delivery counts as dropped on drop"] +#[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] +pub(crate) struct Delivery { + frontier: Option>, + /// Frames the stream carries. + frames: u64, + /// The newest frame timestamp the stream carries, where an acknowledged FIN moves + /// the frontier. + newest: Option, + /// Payload bytes the transport accepted. + bytes: u64, +} + +#[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] +impl Delivery { + /// Record one frame written to the stream, at `timestamp` when timed. Its payload + /// counts only as the transport accepts it; see [`Self::wrote`]. + pub(crate) fn frame(&mut self, timestamp: Option) { + if self.frontier.is_none() { + return; + } + self.frames += 1; + self.newest = self.newest.max(timestamp.map(Duration::from)); + } + + /// Record `bytes` of payload the transport accepted. + pub(crate) fn wrote(&mut self, bytes: usize) { + if self.frontier.is_some() { + self.bytes = self.bytes.saturating_add(bytes as u64); + } + } + + /// The peer acknowledged the stream's FIN: the frontier moved, to the stream's + /// newest frame if that is newer. + pub(crate) fn acked(mut self) { + let Some(inner) = self.frontier.take() else { + return; + }; + if self.frames == 0 { + return; + } + let mut state = inner.state(); + state.moved = true; + if let Some(newest) = self.newest + && state.acked.is_none_or(|acked| newest > acked) + { + state.acked = Some(newest); + } + } +} + +impl Drop for Delivery { + fn drop(&mut self) { + let Some(inner) = self.frontier.take() else { + return; + }; + if self.frames == 0 { + return; + } + inner.counters.publisher.dropped.add(Content { + bytes: self.bytes, + frames: self.frames, + groups: 1, + datagrams: 0, + }); + } +} + #[cfg(test)] mod tests { use std::sync::{Arc, atomic::Ordering::Relaxed}; @@ -2428,7 +3017,7 @@ mod tests { fn drain(stats: &Registry) -> Report { let mut report = Report::default(); - stats.report(&mut report); + stats.report(&mut report, Instant::ZERO); report } @@ -2440,12 +3029,12 @@ mod tests { let _scopes: Vec<_> = (0..8).map(|i| ctx.egress(format!("b/{i}").as_str())).collect(); let mut report = Report::default(); - stats.report(&mut report); + stats.report(&mut report, Instant::ZERO); assert_eq!(report.traffic.len(), 8); assert_eq!(report.sessions.len(), 1); let (traffic, sessions) = (report.traffic.as_ptr(), report.sessions.as_ptr()); - stats.report(&mut report); + stats.report(&mut report, Instant::ZERO); assert_eq!(report.traffic.len(), 8, "refilled, not appended"); assert_eq!(report.sessions.len(), 1); assert_eq!(report.traffic.as_ptr(), traffic, "traffic buffer reused"); @@ -2571,7 +3160,7 @@ mod tests { ]; assert_eq!(registry.snapshot().control(), expected); drop(context); - registry.report(&mut Report::default()); + registry.report(&mut Report::default(), Instant::ZERO); assert_eq!(registry.snapshot().control(), expected); assert_eq!(registry.snapshot().control(), expected, "scrapes do not drain totals"); } @@ -2902,12 +3491,12 @@ mod tests { drop(session); let mut report = Report::default(); - stats.report(&mut report); + stats.report(&mut report, Instant::ZERO); assert!(!report.sessions[0].closed, "a live meter keeps its label's row open"); meter.datagram(4); drop(meter); - stats.report(&mut report); + stats.report(&mut report, Instant::ZERO); assert_eq!(report.sessions[0].presence.sent_bytes, 7); assert!(report.sessions[0].closed); assert_eq!( @@ -3023,7 +3612,7 @@ mod tests { drop(request); drop(clone); let mut report = Report::default(); - stats.report(&mut report); + stats.report(&mut report, Instant::ZERO); for (label, requests, bytes) in [("old", 1, 7), ("new", 0, 11)] { let tier = Tier::new(label); let presence = report.sessions.iter().find(|entry| entry.tier == tier).unwrap(); @@ -3045,7 +3634,7 @@ mod tests { bytes ); } - stats.report(&mut report); + stats.report(&mut report, Instant::ZERO); assert!(report.sessions.is_empty()); let totals = stats.snapshot().sessions(); assert_eq!(totals.iter().map(|(_, p)| p.requests).sum::(), 1); @@ -3242,3 +3831,587 @@ mod tests { assert!(serde_json::from_str::(r#"{"sessions":null,"sessions_closed":1}"#).is_err()); } } + +/// The lag sampler and dropped-media counters, driven through the model the way the +/// wire's group serve tasks drive them, at report instants the test supplies. +#[cfg(all(test, not(loom)))] +mod lag_tests { + use futures::FutureExt; + + use super::*; + use crate::{broadcast, track}; + + const PATH: &str = "demo"; + const TICK: Duration = Duration::from_secs(1); + /// Bytes per produced group: the weight one tick of media carries. + const GROUP_BYTES: u64 = 30 * 100; + + struct Harness { + stats: Registry, + session: Session, + broadcast: broadcast::Producer, + track: track::Producer, + /// Media time of the next group's first frame. + next: Duration, + /// The instant the next report samples at. + now: Instant, + } + + impl Harness { + fn new() -> Self { + let stats = Registry::new(Config::new()); + let session = stats.tier(Tier::default()).session("root"); + let broadcast = broadcast::Info::new().produce(); + let track = broadcast.create_track("video", track::Info::timed()).unwrap(); + Self { + stats, + session, + broadcast, + track, + next: Duration::ZERO, + now: Instant::ZERO, + } + } + + /// A tagged egress subscription, as a session serving a viewer holds one. + fn subscribe(&self) -> track::Subscriber { + self.subscribe_as(&self.session) + } + + fn subscribe_as(&self, session: &Session) -> track::Subscriber { + self.broadcast + .consume() + .with_stats(session.egress(PATH)) + .track("video") + .unwrap() + .subscribe(None) + .now_or_never() + .expect("track is live") + .expect("subscribed") + } + + /// Produce one second of media as one group of 30 frames over one tick, and + /// return the frame timestamps written. + fn produce(&mut self) -> Vec { + let mut group = self.track.append_group().unwrap(); + let mut written = Vec::new(); + for i in 0..30u64 { + // Whole milliseconds, the track's scale, so the timestamps written + // toward the peer match what the track stored. + let at = self.next + Duration::from_millis(i * 33); + let ts = Timestamp::try_from(at).unwrap(); + group.write_frame(ts, vec![0u8; 100]).unwrap(); + written.push(ts); + } + group.finish().unwrap(); + self.next += TICK; + self.now += TICK; + written + } + + /// Write a group toward the peer through `sub`'s frontier, every byte accepted. + fn write(sub: &track::Subscriber, frames: &[Timestamp]) -> Delivery { + let mut delivery = sub.deliver(); + for ts in frames { + delivery.frame(Some(*ts)); + delivery.wrote(100); + } + delivery + } + + /// One report; the egress row for the broadcast. + fn tick(&self) -> Traffic { + let mut report = Report::default(); + self.stats.report(&mut report, self.now); + report + .traffic + .iter() + .find(|e| e.path.as_str() == PATH) + .map(|e| e.publisher) + .expect("broadcast row") + } + } + + /// The buckets that grew between two readouts. + fn grew(before: &Traffic, after: &Traffic) -> Vec<(usize, u64)> { + before + .lag + .buckets() + .iter() + .zip(after.lag.buckets()) + .enumerate() + .filter(|(_, (b, a))| a > b) + .map(|(i, (b, a))| (i, a - b)) + .collect() + } + + /// The egress lag bytes recorded so far, without a report sampling. + fn recorded(stats: &Registry) -> u64 { + stats + .snapshot() + .traffic() + .into_iter() + .find(|(_, role, _)| matches!(role, Role::Publisher)) + .map_or(0, |(_, _, traffic)| traffic.lag.total()) + } + + #[test] + fn bucket_edges_are_upper_exclusive() { + assert_eq!(Histogram::bucket(Duration::ZERO), 0); + assert_eq!(Histogram::bucket(Duration::from_millis(49)), 0); + assert_eq!(Histogram::bucket(Duration::from_millis(50)), 1); + assert_eq!(Histogram::bucket(Duration::from_millis(4999)), 6); + assert_eq!(Histogram::bucket(Duration::from_secs(5)), 7); + assert_eq!(Histogram::bucket(Duration::MAX), 7); + } + + #[test] + fn media_rate_reader_stays_in_the_lowest_bucket() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let mut last = h.tick(); + for _ in 0..5 { + let frames = h.produce(); + Harness::write(&sub, &frames).acked(); + let now = h.tick(); + assert_eq!(grew(&last, &now), vec![(0, GROUP_BYTES)]); + last = now; + } + assert_eq!(last.dropped.groups, 0); + } + + #[test] + fn stalled_reader_walks_up_the_buckets() { + let mut h = Harness::new(); + // Media already produced before the subscription opens is a deliberate start + // point, not lag. + h.produce(); + let sub = h.subscribe(); + let mut last = h.tick(); + assert!(last.lag.is_empty()); + + // Never acknowledges anything: the frontier stands at the newest frame the + // track held when it opened, and each second of media adds a second of lag. + let mut groups = Vec::new(); + let mut buckets = Vec::new(); + for _ in 0..6 { + let frames = h.produce(); + groups.push(Harness::write(&sub, &frames)); + let now = h.tick(); + let grew = grew(&last, &now); + assert_eq!(grew.len(), 1, "one sample per tick: {grew:?}"); + assert_eq!(grew[0].1, GROUP_BYTES); + buckets.push(grew[0].0); + last = now; + } + // 1s, 2s, 3s, 4s, 5s, 6s behind. + assert_eq!(buckets, vec![5, 6, 6, 6, 7, 7]); + } + + #[test] + fn slow_reader_backlog_grows_into_higher_buckets() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let mut last = h.tick(); + let mut pending = std::collections::VecDeque::new(); + let mut buckets = Vec::new(); + for tick in 0..8 { + pending.push_back(Harness::write(&sub, &h.produce())); + // Acknowledges one group every other tick: half the media rate. + if tick % 2 == 1 { + pending.pop_front().unwrap().acked(); + } + let now = h.tick(); + buckets.push(grew(&last, &now)[0].0); + last = now; + } + assert_eq!(buckets, vec![5, 5, 6, 6, 6, 6, 6, 6]); + } + + #[test] + fn the_report_clock_bounds_a_slow_timeline() { + // The publisher's clock crawls: 10ms of media per second. Media time alone + // says a viewer that never acknowledges is barely behind; each report + // interval it produced through without an acknowledgement says otherwise. + let mut h = Harness::new(); + let sub = h.subscribe(); + let mut stalled = Vec::new(); + let mut last = h.tick(); + let mut buckets = Vec::new(); + for i in 0..7u64 { + let ts = Timestamp::from_millis(i * 10).unwrap(); + let mut group = h.track.append_group().unwrap(); + group.write_frame(ts, vec![0u8; 100]).unwrap(); + group.finish().unwrap(); + stalled.push(Harness::write(&sub, &[ts])); + h.now += TICK; + let now = h.tick(); + let grew = grew(&last, &now); + assert_eq!(grew.len(), 1, "{grew:?}"); + buckets.push(grew[0].0); + last = now; + } + // Stalled 1s .. 7s, against at most 60ms of media. + assert_eq!(buckets, vec![5, 6, 6, 6, 7, 7, 7]); + } + + #[test] + fn timestamps_that_stop_advancing_do_not_hide_a_stalled_viewer() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let mut last = h.tick(); + for _ in 0..2 { + let frames = h.produce(); + Harness::write(&sub, &frames).acked(); + } + let now = h.tick(); + assert_eq!(grew(&last, &now), vec![(0, 2 * GROUP_BYTES)]); + last = now; + + // The publisher's clock stops: each group repeats the last one's timestamps, an + // equal start the track admits, so nothing is newer than what was acknowledged. + // The viewer stops acknowledging. Media time reads the frontier as level with the + // newest frame; the intervals produced through without an acknowledgement still + // count. + let mut stalled = Vec::new(); + let mut buckets = Vec::new(); + for _ in 0..3 { + h.next = TICK; + let frames = h.produce(); + stalled.push(Harness::write(&sub, &frames)); + let now = h.tick(); + buckets.push(grew(&last, &now)[0].0); + last = now; + } + assert_eq!(buckets, vec![5, 6, 6]); + } + + #[test] + fn a_pause_adds_no_weight_and_no_lag() { + let mut h = Harness::new(); + let sub = h.subscribe(); + Harness::write(&sub, &h.produce()).acked(); + let mut last = h.tick(); + + // Nothing produced for a minute: the histogram does not move. + for _ in 0..60 { + h.now += TICK; + let now = h.tick(); + assert_eq!(now.lag, last.lag); + last = now; + } + + // Resuming does not read the pause as a stalled frontier: the group still in + // flight is one second of media behind, not the minute the source was away. + let _in_flight = Harness::write(&sub, &h.produce()); + let now = h.tick(); + assert_eq!(grew(&last, &now), vec![(5, GROUP_BYTES)]); + } + + #[test] + fn a_subscription_opened_before_any_media_measures_from_the_first_frame() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let last = h.tick(); + let _stalled = Harness::write(&sub, &h.produce()); + h.produce(); + let now = h.tick(); + // Two seconds produced since it opened, none acknowledged: 1.96s from the + // first frame to the newest, over two seconds stalled. + assert_eq!(grew(&last, &now), vec![(6, 2 * GROUP_BYTES)]); + } + + #[test] + fn an_unacknowledged_group_counts_as_dropped() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let before = h.tick(); + + let frames = h.produce(); + // Half the group written, then the stream is reset (skip, expiry, stop, or + // session close all drop the serve task). + drop(Harness::write(&sub, &frames[..15])); + // Nothing written at all is not a dropped group. + drop(sub.deliver()); + + let after = h.tick(); + assert_eq!(after.dropped.groups, 1); + assert_eq!(after.dropped.frames, 15); + assert_eq!(after.dropped.bytes, 15 * 100); + // The drop itself adds nothing to the histogram beyond the tick's own sample. + assert_eq!(after.lag.total() - before.lag.total(), GROUP_BYTES); + } + + #[test] + fn a_partly_accepted_frame_drops_only_what_the_transport_took() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let frames = h.produce(); + let mut delivery = sub.deliver(); + delivery.frame(Some(frames[0])); + delivery.wrote(40); + delivery.frame(Some(frames[1])); + drop(delivery); + let dropped = h.tick().dropped; + assert_eq!((dropped.groups, dropped.bytes), (1, 40)); + } + + #[test] + fn a_frame_written_in_chunks_weighs_what_arrived() { + let h = Harness::new(); + let _sub = h.subscribe(); + let last = h.tick(); + let group = h.track.append_group().unwrap(); + let mut frame = group + .create_frame(crate::frame::Info { + size: 1000, + timestamp: Some(Timestamp::ZERO), + }) + .unwrap(); + frame.write(vec![0u8; 400]).unwrap(); + let now = h.tick(); + assert_eq!(grew(&last, &now), vec![(0, 400)]); + } + + /// A compressed track's frame weighs the slice it stores, whether written whole or + /// streamed in chunks of plain payload. + #[test] + fn a_compressed_frame_weighs_its_stored_size_however_written() { + let h = Harness::new(); + let info = track::Info::timed().with_compress(true); + let track = h.broadcast.create_track("captions", info).unwrap(); + let _sub = (h.broadcast.consume().with_stats(h.session.egress(PATH))) + .track("captions") + .unwrap() + .subscribe(None) + .now_or_never() + .expect("track is live") + .expect("subscribed"); + let payload = vec![7u8; 4000]; + let mut last = h.tick(); + let mut weights = Vec::new(); + for (index, chunked) in [false, true].into_iter().enumerate() { + let mut group = track.append_group().unwrap(); + let timestamp = Timestamp::from_millis(1000 * index as u64).unwrap(); + if chunked { + let info = crate::frame::Info { + size: payload.len() as u64, + timestamp: Some(timestamp), + }; + let mut frame = group.create_frame(info).unwrap(); + for chunk in payload.chunks(1000) { + frame.write(chunk.to_vec()).unwrap(); + } + group = frame.finish().unwrap(); + } else { + group.write_frame(timestamp, payload.clone()).unwrap(); + } + group.finish().unwrap(); + let now = h.tick(); + weights.push(now.lag.total() - last.lag.total()); + last = now; + } + assert!(weights[0] < payload.len() as u64, "the frame compressed: {weights:?}"); + assert_eq!(weights[0], weights[1]); + } + + #[test] + fn subscriptions_of_one_broadcast_sum_into_one_row() { + let mut h = Harness::new(); + let other = h.stats.tier(Tier::default()).session("root"); + let a = h.subscribe(); + let b = h.subscribe_as(&other); + let last = h.tick(); + let mut stalled = Vec::new(); + for _ in 0..2 { + let frames = h.produce(); + Harness::write(&a, &frames).acked(); + stalled.push(Harness::write(&b, &frames)); + } + h.produce(); + let now = h.tick(); + // `a` is one second behind, `b` three; both weigh the same three groups. + assert_eq!(grew(&last, &now), vec![(5, 3 * GROUP_BYTES), (6, 3 * GROUP_BYTES)]); + } + + #[test] + fn a_subscription_closed_between_reports_is_sampled_once() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let last = h.tick(); + h.produce(); + drop(sub); + + // The report that follows finds no frontier, yet the group still lands, at the + // 957ms it was behind when the guard dropped: the close charges no stall. + let now = h.tick(); + assert_eq!(grew(&last, &now), vec![(4, GROUP_BYTES)]); + + // Sampled exactly once: the closing report pruned the row, and later media adds + // nothing to the retired totals. + h.produce(); + h.stats.report(&mut Report::default(), h.now); + assert_eq!(recorded(&h.stats), GROUP_BYTES); + } + + #[test] + fn a_subscription_closed_mid_interval_adds_only_the_bytes_since_its_last_sample() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let mut last = h.tick(); + h.produce(); + last = { + let now = h.tick(); + assert_eq!(grew(&last, &now), vec![(5, GROUP_BYTES)]); + now + }; + h.produce(); + h.produce(); + drop(sub); + + let now = h.tick(); + assert_eq!( + grew(&last, &now).iter().map(|(_, bytes)| bytes).sum::(), + 2 * GROUP_BYTES + ); + } + + #[test] + fn an_in_flight_delivery_defers_the_final_sample_to_its_own_drop() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let delivery = Harness::write(&sub, &h.produce()); + drop(sub); + // The delivery still holds the frontier, so it has not closed yet. + assert_eq!(recorded(&h.stats), 0); + + drop(delivery); + assert_eq!(recorded(&h.stats), GROUP_BYTES); + } + + #[test] + fn the_last_subscriber_samples_a_track_its_own_drop_releases() { + let mut h = Harness::new(); + let sub = h.subscribe(); + h.produce(); + // The producer is gone, so the subscriber holds the last reference to the track. + drop(h.track); + drop(h.broadcast); + + drop(sub); + assert_eq!(recorded(&h.stats), GROUP_BYTES); + } + + #[test] + fn an_in_flight_delivery_samples_a_track_that_already_closed() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let delivery = Harness::write(&sub, &h.produce()); + drop(h.track); + drop(h.broadcast); + drop(sub); + assert_eq!(recorded(&h.stats), 0); + + // The track died with the subscriber; the delivery still owes its bytes. + drop(delivery); + assert_eq!(recorded(&h.stats), GROUP_BYTES); + } + + #[test] + fn a_closed_subscription_leaves_the_sampler() { + let mut h = Harness::new(); + let sub = h.subscribe(); + h.tick(); + drop(sub); + h.produce(); + let row = h.tick(); + assert!(row.lag.is_empty()); + } + + #[test] + fn the_frontier_list_stays_bounded_without_reports() { + let h = Harness::new(); + let scope = h.session.egress(PATH); + for _ in 0..1000 { + drop(scope.subscribe()); + } + let counters = h.stats.shared().entries.lock()[&PathOwned::from(PATH.to_string())].tier(&Tier::default()); + let held = counters.publisher.frontiers.lock().unwrap().len(); + assert!(held <= 8, "{held} frontiers held with none live"); + } + + /// Two copies live at once, as a resumed subscription's old and new are until the + /// old one drains, each weigh at their own edge: the one far ahead does not drag + /// the other's bytes up the buckets. + #[test] + fn live_copies_weigh_at_their_own_edges() { + let h = Harness::new(); + let subscription = h.session.egress(PATH).subscribe(); + let (near, far) = (Arc::new(Produced::default()), Arc::new(Produced::default())); + let _feeds = (subscription.frontier().feed(&near), subscription.frontier().feed(&far)); + let at = |millis| Some(Timestamp::from_millis(millis).unwrap()); + near.record(at(0), 10); + far.record(at(0), 10); + let last = h.tick(); + near.record(at(100), 10); + far.record(at(3000), 20); + assert_eq!(grew(&last, &h.tick()), vec![(2, 10), (6, 20)]); + } + + /// A route moved over and over between reports keeps only a few retired copies + /// apart, and still weighs every byte they produced. + #[test] + fn retired_copies_stay_bounded_between_reports() { + let h = Harness::new(); + let subscription = h.session.egress(PATH).subscribe(); + let frontier = subscription.frontier(); + for _ in 0..1000 { + let copy = Arc::new(Produced::default()); + let feed = frontier.feed(©); + copy.record(Some(Timestamp::ZERO), 10); + drop(feed); + } + let held = frontier.0.as_ref().expect("egress").state().folded.len(); + assert!(held <= MAX_FOLDED, "{held} retired copies held"); + assert_eq!(h.tick().lag.total(), 10_000); + } + + #[test] + fn lag_and_drops_round_trip_and_stay_off_the_wire_when_empty() { + let empty = serde_json::to_string(&Traffic::default()).unwrap(); + assert!(!empty.contains("lag") && !empty.contains("dropped"), "{empty}"); + + let mut traffic = Traffic::default(); + traffic.lag.buckets[0] = 7; + traffic.lag.buckets[7] = 9; + traffic.dropped.bytes = 3; + traffic.dropped.frames = 2; + traffic.dropped.groups = 1; + let json = serde_json::to_string(&traffic).unwrap(); + assert!(json.contains(r#""lag":{"50ms":7,"inf":9}"#), "{json}"); + assert!( + json.contains(r#""dropped":{"bytes":3,"frames":2,"groups":1,"datagrams":0}"#), + "{json}" + ); + assert_eq!(serde_json::from_str::(&json).unwrap(), traffic); + + let mut sum = traffic; + sum.add(traffic); + assert_eq!(sum.lag.buckets(), &[14, 0, 0, 0, 0, 0, 0, 18]); + assert_eq!(sum.dropped.bytes, 6); + } + + #[test] + fn merging_peer_readouts_saturates() { + let lag = Histogram { + buckets: [u64::MAX; BUCKETS], + }; + let mut traffic = Traffic { + lag, + ..Default::default() + }; + traffic.add(traffic); + assert_eq!(traffic.lag, lag); + assert_eq!(traffic.lag.total(), u64::MAX); + } +} diff --git a/rs/moq-net/tests/integration/idle_session.rs b/rs/moq-net/tests/integration/idle_session.rs index 593ee2ead4..f563102875 100644 --- a/rs/moq-net/tests/integration/idle_session.rs +++ b/rs/moq-net/tests/integration/idle_session.rs @@ -189,7 +189,7 @@ async fn wire_collection_never_arms_the_idle_sampler() { pair.server_transport.wire().set_sent_bytes(tick); pair.server_transport.wire().set_received_bytes(tick * 2); let mut report = moq_net::stats::Report::default(); - registry.report(&mut report); + registry.report(&mut report, moq_net::time::Instant::ZERO); assert_eq!( (report.sessions[0].presence.wire_tx, report.sessions[0].presence.wire_rx), (tick, tick * 2) diff --git a/rs/moq-relay/src/auth/admission_tests.rs b/rs/moq-relay/src/auth/admission_tests.rs index 1bec4316f4..06c9fe943c 100644 --- a/rs/moq-relay/src/auth/admission_tests.rs +++ b/rs/moq-relay/src/auth/admission_tests.rs @@ -172,7 +172,7 @@ async fn gateway_admission_refuses_once_in_the_shared_stats_registry() { assert_eq!(values[0].1.sessions_refused.refused, 1); assert_eq!(values[0].1.sessions_started, 0); let mut report = moq_net::stats::Report::default(); - registry.report(&mut report); + registry.report(&mut report, moq_net::time::Instant::ZERO); assert_eq!( report.sessions[0].label.as_str(), "", @@ -210,7 +210,7 @@ async fn embedded_expired_refusals_keep_verified_root_and_tier() { assert_eq!(actual, expected); } let mut report = moq_net::stats::Report::default(); - registry.report(&mut report); + registry.report(&mut report, moq_net::time::Instant::ZERO); assert_eq!(report.sessions.len(), 2); for entry in report.sessions { assert!(["", "edge"].contains(&entry.tier.as_str())); diff --git a/rs/moq-relay/src/internal.rs b/rs/moq-relay/src/internal.rs index 06d46a7926..2c2decfb28 100644 --- a/rs/moq-relay/src/internal.rs +++ b/rs/moq-relay/src/internal.rs @@ -1284,7 +1284,7 @@ mod tests { check(); drop(guard); drop(session); - registry.report(&mut Default::default()); + registry.report(&mut Default::default(), moq_net::time::Instant::ZERO); check(); } } @@ -1311,7 +1311,7 @@ mod tests { drop(broadcast); drop(producer); let mut report = moq_net::stats::Report::default(); - stats.report(&mut report); + stats.report(&mut report, moq_net::time::Instant::ZERO); let body = render_metrics(&stats.snapshot(), &[], &[]); assert!( diff --git a/rs/moq-relay/src/refusals.rs b/rs/moq-relay/src/refusals.rs index 52debeb4cc..592b1256b7 100644 --- a/rs/moq-relay/src/refusals.rs +++ b/rs/moq-relay/src/refusals.rs @@ -88,7 +88,7 @@ mod tests { } } let mut report = moq_net::stats::Report::default(); - registry.report(&mut report); + registry.report(&mut report, moq_net::time::Instant::ZERO); assert_eq!(report.sessions.len(), 4); for entry in report.sessions { assert!(["", "tenant"].contains(&entry.label.as_str())); diff --git a/rs/moq-relay/src/web.rs b/rs/moq-relay/src/web.rs index 5b51d48d85..7d19ad4310 100644 --- a/rs/moq-relay/src/web.rs +++ b/rs/moq-relay/src/web.rs @@ -1333,7 +1333,7 @@ mod tests { decider.await.unwrap(); let mut report = Report::default(); - registry.report(&mut report); + registry.report(&mut report, moq_net::time::Instant::ZERO); let [entry] = report.sessions.as_slice() else { panic!("expected one attributed entry: {:?}", report.sessions); }; @@ -1436,7 +1436,7 @@ mod tests { } let _leases = decider.await.unwrap(); let mut report = Report::default(); - state.cluster.stats.report(&mut report); + state.cluster.stats.report(&mut report, moq_net::time::Instant::ZERO); assert_eq!(report.traffic.len(), 2); assert_eq!(report.sessions.len(), 2); for label in ["first", "second"] { diff --git a/rs/moq-relay/tests/integration/auth_lifetime.rs b/rs/moq-relay/tests/integration/auth_lifetime.rs index df617c0b4a..0b35fae740 100644 --- a/rs/moq-relay/tests/integration/auth_lifetime.rs +++ b/rs/moq-relay/tests/integration/auth_lifetime.rs @@ -614,7 +614,7 @@ async fn a_moved_tier_retags_the_live_session() { // Each label's (active sessions, sent bytes, received bytes), from one report. let labels = || { let mut report = stats::Report::default(); - registry.report(&mut report); + registry.report(&mut report, moq_net::time::Instant::ZERO); report .sessions .into_iter() diff --git a/rs/moq-relay/tests/integration/refusals.rs b/rs/moq-relay/tests/integration/refusals.rs index 5ec64a027f..3ff613ec87 100644 --- a/rs/moq-relay/tests/integration/refusals.rs +++ b/rs/moq-relay/tests/integration/refusals.rs @@ -402,7 +402,7 @@ async fn expired_verified_tokens_count_under_the_token_root_and_auth_tier() { assert_eq!(relay.refused("expired").await, 1); assert_eq!(relay.refused("refused").await, 0); let mut report = moq_net::stats::Report::default(); - relay.stats.report(&mut report); + relay.stats.report(&mut report, moq_net::time::Instant::ZERO); assert_eq!(report.sessions.len(), 1); let entry = &report.sessions[0]; assert_eq!(entry.tier.as_str(), tier.unwrap_or("")); diff --git a/rs/moq-stats/benches/producer_tick.rs b/rs/moq-stats/benches/producer_tick.rs index 613cbc7129..721807c452 100644 --- a/rs/moq-stats/benches/producer_tick.rs +++ b/rs/moq-stats/benches/producer_tick.rs @@ -138,7 +138,7 @@ impl Relay { .collect(); let mut report = stats::Report::default(); - registry.report(&mut report); + registry.report(&mut report, moq_net::time::Instant::ZERO); assert_eq!(report.traffic.len(), paths * tiers, "one held entry per path and tier"); let depth = if groups.is_some() { 1 } else { 0 }; diff --git a/rs/moq-stats/src/aggregate.rs b/rs/moq-stats/src/aggregate.rs index 5362551d7f..6be54be86e 100644 --- a/rs/moq-stats/src/aggregate.rs +++ b/rs/moq-stats/src/aggregate.rs @@ -329,6 +329,16 @@ impl Mergeable for Traffic { || self.stale.frames < prev.stale.frames || self.stale.groups < prev.stale.groups || self.stale.datagrams < prev.stale.datagrams + || self + .lag + .buckets() + .iter() + .zip(prev.lag.buckets()) + .any(|(now, prev)| now < prev) + || self.dropped.bytes < prev.dropped.bytes + || self.dropped.frames < prev.dropped.frames + || self.dropped.groups < prev.dropped.groups + || self.dropped.datagrams < prev.dropped.datagrams } } @@ -2278,6 +2288,32 @@ mod tests { (merged, (path, None)) } + /// Nodes' lag histograms and dropped media merge bucket by bucket, and each is a + /// counter a reset shows in. + #[test] + fn lag_and_drops_merge_and_regress_like_every_counter() { + let traffic = |json: &str| serde_json::from_str::(json).expect("traffic"); + let node = traffic(r#"{"lag":{"50ms":5,"inf":2},"dropped":{"bytes":3,"frames":2,"groups":1}}"#); + let mut merged = node; + Traffic::merge( + &mut merged, + traffic(r#"{"lag":{"1s":4},"dropped":{"bytes":1,"frames":1,"groups":1}}"#), + ); + assert_eq!(merged.lag.buckets(), &[5, 0, 0, 0, 4, 0, 0, 2]); + let dropped = merged.dropped; + assert_eq!((dropped.bytes, dropped.frames, dropped.groups), (4, 3, 2)); + + assert!(!node.regressed(&node)); + for lower in [ + r#"{"lag":{"50ms":4,"inf":2},"dropped":{"bytes":3,"frames":2,"groups":1}}"#, + r#"{"lag":{"50ms":5,"inf":2},"dropped":{"bytes":2,"frames":2,"groups":1}}"#, + r#"{"lag":{"50ms":5,"inf":2},"dropped":{"bytes":3,"frames":1,"groups":1}}"#, + r#"{"lag":{"50ms":5,"inf":2},"dropped":{"bytes":3,"frames":2,"groups":0}}"#, + ] { + assert!(traffic(lower).regressed(&node), "{lower}"); + } + } + #[test] fn outcomes_survive_node_departure_resume_and_counter_only_reset() { let (mut merged, key) = hand_merged::(); diff --git a/rs/moq-stats/src/lib.rs b/rs/moq-stats/src/lib.rs index 3e094a7294..324b36e2b3 100644 --- a/rs/moq-stats/src/lib.rs +++ b/rs/moq-stats/src/lib.rs @@ -47,7 +47,9 @@ use std::collections::BTreeMap; /// Counter collection, re-exported from [`moq_net::stats`] so stats consumers /// can depend on this crate alone. -pub use moq_net::stats::{Failure, Failures, Handle, Presence, RefusalKind, Refusals, Registry, Role, Tier, Traffic}; +pub use moq_net::stats::{ + Failure, Failures, Handle, Histogram, Presence, RefusalKind, Refusals, Registry, Role, Tier, Traffic, +}; use moq_net::{AsPath, Epoch, Path, PathOwned}; diff --git a/rs/moq-stats/src/produce.rs b/rs/moq-stats/src/produce.rs index e587d4562d..37318d5b45 100644 --- a/rs/moq-stats/src/produce.rs +++ b/rs/moq-stats/src/produce.rs @@ -239,7 +239,8 @@ impl Task { loop { // Scheduled instants are at least an interval apart, even after a late tick, - // so each drain gets an interval of its own to open its groups in. + // so each drain gets an interval of its own to open its groups in. The lag + // sample takes the time it actually ran: a viewer stalls across the delay too. let scheduled = ticker.tick().await; if weak.upgrade().is_none() { @@ -247,7 +248,7 @@ impl Task { return; } - drain.collect(); + drain.collect(moq_clock::now()); drain.publish(scheduled); } } @@ -303,10 +304,10 @@ impl Drain { } } - /// Drain the registry and sort each entry into its pending frames, - /// creating group tracks on first sight. - fn collect(&mut self) { - self.task.registry.report(&mut self.report); + /// Drain the registry at `now`, its lag sample's instant, and sort each entry + /// into its pending frames, creating group tracks on first sight. + fn collect(&mut self, now: Instant) { + self.task.registry.report(&mut self.report, now); self.tick += 1; self.node.collect(&self.report, self.tick, self.task.depth); } @@ -855,6 +856,15 @@ struct PathUsage { subscriber: Option, } +/// `traffic` as a group track's per-path row carries it: without the egress lag +/// histogram and dropped media, which the totals and prefix tracks report, so a +/// row per path stays small. +fn path_row(mut traffic: Traffic) -> Traffic { + traffic.lag = Default::default(); + traffic.dropped = Default::default(); + traffic +} + /// Group tracks are timed like every stats track, and keep merge-patch deltas /// inside the track's own DEFLATE window, so a joiner asks for a fresh group /// once the deltas outweigh eight snapshots. @@ -919,8 +929,8 @@ impl GroupTrack { tier.paths.entries.push(( entry.path.clone(), PathUsage { - publisher: publisher.then_some(entry.publisher), - subscriber: subscriber.then_some(entry.subscriber), + publisher: publisher.then_some(path_row(entry.publisher)), + subscriber: subscriber.then_some(path_row(entry.subscriber)), }, )); } @@ -1428,7 +1438,7 @@ mod tests { async fn pumping_at(drain: &mut Drain, now: Instant, future: impl std::future::Future) -> T { let mut future = std::pin::pin!(future); for _ in 0..1000 { - drain.collect(); + drain.collect(now); drain.publish(now); if let Some(result) = futures::FutureExt::now_or_never(future.as_mut()) { return result; @@ -1500,7 +1510,7 @@ mod tests { counters.set_received_bytes(n * 2); }; store(40); - drain.collect(); + drain.collect(Instant::now()); drain.publish(Instant::now()); let consumer = edge.consume().with_hidden(true); let mut events = consumer.announced(); @@ -1534,7 +1544,7 @@ mod tests { "a cache exposed an earlier hold's bytes" ); store(reading + delta); - drain.collect(); + drain.collect(Instant::now()); drain.publish(Instant::now()); let next = pumping(&mut drain, reader.next()).await.unwrap().unwrap().value; assert_eq!((next[""].wire_tx, next[""].wire_rx), (delta, delta * 2)); @@ -1564,7 +1574,7 @@ mod tests { drop(reader); for _ in 0..100 { tokio::task::yield_now().await; - drain.collect(); + drain.collect(Instant::now()); drain.publish(Instant::now()); if drain.node.presence.is_empty() { break; @@ -2367,7 +2377,7 @@ mod tests { }) .expect("drain"); let mut tick = || { - drain.collect(); + drain.collect(moq_clock::now()); drain.publish(moq_clock::now()); drain.node.presence.len() }; @@ -2521,6 +2531,63 @@ mod tests { assert!(matches!(subscribing.await, Err(moq_net::Error::NotFound))); } + /// The egress lag histogram reaches the totals, the prefix tracks, and a group's + /// totals, while the group's per-path rows leave it out. + #[tokio::test(start_paused = true)] + async fn lag_reaches_totals_and_prefixes_but_not_path_rows() { + let (producer, origin) = grouped_producer(DEFAULT_LINGER); + let registry = producer.registry(); + let _live = feed(registry, Tier::default(), "acme/live", true, 1, 10).await; + drive_tick().await; + let (_, stats) = announced(&origin).await; + + assert_eq!(default_totals(&stats).await.publisher.lag.total(), 10); + let mut track = hold_prefix(&stats, "acme").await; + let frame = last_prefix(&mut track).expect("answered at once"); + assert_eq!(frame[""].publisher.lag.total(), 10); + assert_eq!(frame[""].children["live"].publisher.lag.total(), 10); + let group = read_group(&stats, "acme").await; + let tier = &group.tiers[""]; + assert_eq!(tier.publisher.lag.total(), 10); + assert!(tier.paths["acme/live"].publisher.lag.is_empty()); + } + + /// A drain that runs late samples lag at the time it ran, not the tick it was due + /// at: a viewer that acknowledged nothing is behind across the delay too. + #[tokio::test(start_paused = true)] + async fn a_late_drain_charges_the_stall_up_to_when_it_ran() { + let (producer, origin) = test_producer(None); + let registry = producer.registry(); + let data = produce_origin(); + let source = data.publish("acme/live", origin::Route::default()).expect("publish"); + let video = source.create_track("video", timed()).expect("create_track"); + let egress = data + .consume() + .with_stats(registry.tier(Tier::default()).session("viewer")); + let broadcast = egress.request_broadcast("acme/live", None).await.expect("resolve"); + let _sub = broadcast + .track("video") + .expect("track") + .subscribe(None) + .await + .expect("subscribe"); + // An idle drain with the subscription open, which the stall is measured from. + drive_tick().await; + + let mut group = video.append_group().expect("group"); + group.write_frame(Timestamp::ZERO, vec![0u8; 100]).expect("write"); + group.finish().expect("finish"); + // The next drain, due an interval from now, runs 3s from now. + tokio::time::advance(Duration::from_secs(3)).await; + for _ in 0..4 { + tokio::task::yield_now().await; + } + + let (_, stats) = announced(&origin).await; + let lag = default_totals(&stats).await.publisher.lag; + assert_eq!(lag.buckets(), &[0, 0, 0, 0, 0, 0, 100, 0], "stalled 3s, under 5s"); + } + /// Counters follow presence: a prefix held while 300 subscriptions are /// live counts all 300 as active, and every one leaving reads as ended. /// The child leaves the map after its closing readout; its tail stays in @@ -2624,7 +2691,7 @@ mod tests { }) .expect("drain"); let mut tick = || { - drain.collect(); + drain.collect(moq_clock::now()); drain.publish(moq_clock::now()); drain.node.prefixes.len() }; @@ -2845,7 +2912,7 @@ mod tests { // Warm up: create the groups and tracks and grow every buffer. for _ in 0..3 { - drain.collect(); + drain.collect(moq_clock::now()); drain.publish(moq_clock::now()); } // Hold a prefix with a child per tier, as a reader would. @@ -2861,11 +2928,11 @@ mod tests { ); drain.node.prefixes["room0"].pair.plain.track.subscribe(None) }); - drain.collect(); + drain.collect(moq_clock::now()); drain.publish(moq_clock::now()); let before = counting::allocs(); - drain.collect(); + drain.collect(moq_clock::now()); let allocs = counting::allocs() - before; drain.publish(moq_clock::now()); @@ -2935,7 +3002,7 @@ mod tests { let start = Instant::now(); let acme = registry.tier(Tier::default()).session("acme/live"); - drain.collect(); + drain.collect(start); drain.publish(start); let consumer = edge.consume().with_hidden(true); let mut announced = consumer.announced(); diff --git a/rs/moq-stats/src/produce_bench.rs b/rs/moq-stats/src/produce_bench.rs index 43a2506c8a..cae5763c53 100644 --- a/rs/moq-stats/src/produce_bench.rs +++ b/rs/moq-stats/src/produce_bench.rs @@ -52,7 +52,7 @@ impl Driver { /// Drain the registry and encode every track once, one interval after the last tick. pub fn tick(&mut self) -> FrameBytes { self.now += self.drain.task.interval; - self.drain.collect(); + self.drain.collect(self.now); self.drain.publish(self.now); self.drain.group_frame_bytes() }