From 2a8049378acea63021b9e04904566020a00b9156 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 13:47:41 -0700 Subject: [PATCH 1/6] quest(qos/starvation): claim Co-Authored-By: Claude Opus 5.5 From 18cf2b2a1a2fabe88b677999593945fec9b69af5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 14:14:20 -0700 Subject: [PATCH 2/6] feat(stats): per-broadcast viewer lag and dropped media on egress rows Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + doc/bin/relay/config.md | 9 + doc/concept/stats.md | 40 +- quest/m1/qos/README.md | 3 - quest/m1/qos/publisher-timeliness.md | 11 +- quest/m1/qos/starvation-frames.md | 22 +- quest/m1/qos/starvation.md | 89 --- quest/m1/qos/stats/schema.md | 5 - quest/m1/quic/README.md | 2 - quest/m1/track-priority-scope.md | 5 - rs/moq-net/Cargo.toml | 1 + rs/moq-net/benches/stats.rs | 70 ++- rs/moq-net/src/ietf/publisher.rs | 76 +++ rs/moq-net/src/lite/publisher.rs | 110 +++- rs/moq-net/src/model/cache.rs | 10 + rs/moq-net/src/model/group.rs | 8 +- rs/moq-net/src/model/resume.rs | 65 +++ rs/moq-net/src/model/track.rs | 36 +- rs/moq-net/src/stats.rs | 772 ++++++++++++++++++++++++++- rs/moq-stats/src/aggregate.rs | 32 ++ rs/moq-stats/src/lib.rs | 2 +- 21 files changed, 1222 insertions(+), 147 deletions(-) delete mode 100644 quest/m1/qos/starvation.md diff --git a/Cargo.lock b/Cargo.lock index 08d5838edf..006f115fdc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4503,6 +4503,7 @@ dependencies = [ "rand 0.10.3", "serde", "serde_json", + "serde_with", "slab", "thiserror 2.0.21", "tokio", diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 6f8677ac36..a22c93e37a 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -208,6 +208,15 @@ and auth root, split by a **tier** label chosen by the auth server's grant or possible. [Stats](/concept/stats) describes the paths, tracks, and encodings; read them with the [`moq-stats`](https://docs.rs/moq-stats) crate. +`publisher.json` also reports how far behind each broadcast's viewers are. Every +`interval` the relay samples each subscription's lag, the newest frame produced +minus the newest frame the viewer acknowledged, into a byte-weighted `lag` +histogram with buckets at 50 ms, 100 ms, 250 ms, 500 ms, 1 s, 2 s, 5 s, and above. +`dropped_duration`, `dropped_bytes`, and `dropped_groups` count media written to a +viewer on a group stream that ended before the viewer acknowledged it. Both +resolve one group at a time, so they overstate lag and drops for long groups; +see [Lag](/concept/stats#lag). + ## \[iroh] ```toml diff --git a/doc/concept/stats.md b/doc/concept/stats.md index 7b6cb321ca..5968aba1d0 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -84,6 +84,8 @@ return later as a new broadcast. "fetches": 0, "bytes": 1048576, "frames": 900, "groups": 30, "datagrams": 0, "stale": { "bytes": 0, "frames": 0, "groups": 0, "datagrams": 0 }, + "lag": { "50ms": 1040000, "1s": 8576 }, + "dropped_duration": 966.5, "dropped_bytes": 8576, "dropped_groups": 1, "announced": 1, "announced_closed": 0, "broadcasts": 3, "broadcasts_closed": 1, "subscriptions": 6, "subscriptions_closed": 2 @@ -101,11 +103,44 @@ return later as a new broadcast. | `bytes` / `frames` / `groups` | Payload delivered. | | `datagrams` | Groups delivered as an unreliable datagram. A subset of `groups`. | | `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` | `publisher.json` only: how far behind the viewers' acknowledged media is, as bytes per bucket of media time. See [Lag](#lag). Omitted while empty. | +| `dropped_duration` / `dropped_bytes` / `dropped_groups` | `publisher.json` only: media written to a viewer on a group stream that was reset or abandoned before the viewer acknowledged it. The duration is fractional milliseconds. Omitted while zero. | -The last six fields are legacy spellings of the `*_started` and `*_ended` +The last six fields in the example are legacy spellings of the `*_started` and `*_ended` counters, still written so an older consumer reads a newer relay. A reader should prefer the canonical name and fall back to the legacy one. +### Lag + +Each track subscription has an acknowledged **frontier**: the newest frame +timestamp the viewer is known to have received. It starts at the track's newest +frame when the subscription opens, so a backlog the viewer deliberately starts +behind never counts, and advances when the viewer acknowledges the end (FIN) of +a group stream. Every stats interval the relay samples each subscription's lag, +the newest frame the track has produced minus the frontier, or the wall-clock +time the frontier has stood still while newer media exists if that is larger. +It adds the bytes the track produced during the interval to the bucket that lag +falls in, and the broadcast's subscriptions sum into one entry. + +| Bucket | Lag | +| --- | --- | +| `50ms` | under 50 ms | +| `100ms` | 50 ms to 100 ms | +| `250ms` | 100 ms to 250 ms | +| `500ms` | 250 ms to 500 ms | +| `1s` | 500 ms to 1 s | +| `2s` | 1 s to 2 s | +| `5s` | 2 s to 5 s | +| `inf` | 5 s and above | + +Each bucket is a cumulative byte counter, so the difference of two frames is +the byte-weighted lag distribution over that interval, from which a dashboard +reads percentiles or a mean. A paused broadcast produces nothing and adds +nothing. The frontier moves once per group, so a healthy viewer of a stream +with two-second groups still spreads across the buckets below `2s`; the dropped +counters likewise count a whole group's written span, including frames the +viewer did receive. + ### Presence ```json @@ -119,7 +154,8 @@ ends on the old one and starts on the new. ### Counters -Every counter is a cumulative, monotonic unsigned integer. A rate is the +Every counter is cumulative and monotonic: an unsigned integer, apart from +`dropped_duration` in fractional milliseconds. A rate is the difference between two frames divided by the time between them, and a live count is started minus ended. A frame never shows ended above started. diff --git a/quest/m1/qos/README.md b/quest/m1/qos/README.md index b6c84121b6..1ac265ffa5 100644 --- a/quest/m1/qos/README.md +++ b/quest/m1/qos/README.md @@ -33,9 +33,6 @@ preflight, consumes them downstream. ## Quests -- [Starvation](/quest/m1/qos/starvation.md) - per broadcast, how far behind - the acknowledged frontier of its subscriptions is, in media time, plus the - media dropped before it was acknowledged - [Starvation at frame granularity](/quest/m1/qos/starvation-frames.md) - the acknowledged frontier moves at every frame boundary through `poll_acked`, with a delivery-delay histogram for jitter diff --git a/quest/m1/qos/publisher-timeliness.md b/quest/m1/qos/publisher-timeliness.md index 8fd16bf480..2c82cec89e 100644 --- a/quest/m1/qos/publisher-timeliness.md +++ b/quest/m1/qos/publisher-timeliness.md @@ -10,8 +10,8 @@ cooperation. ## Plan -This is the ingress mirror of [starvation](/quest/m1/qos/starvation.md) and -needs no acknowledgments: the relay is the receiver, and a frame is measured +This is the ingress mirror of the egress `lag` histogram on `stats::Traffic` +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 @@ -27,8 +27,9 @@ milliseconds, and the re-anchor happens on its own as old minima expire. Document the window length beside the bucket edges. Aggregate as a byte-weighted cumulative histogram of drift on the -`Role::Subscriber` (ingress) rows, with the same bucket edges and the same -monotonic contract as the starvation histogram. Beside it keep two cumulative +`Role::Subscriber` (ingress) rows, reusing `stats::Histogram` (its +`EDGES` and cumulative contract) and its crate-private atomic backing, as the +egress `lag` field does. Beside it keep two cumulative counters: `timestamp_regressions`, frames whose timestamp is below the previous group's newest timestamp, and `stalls`, gaps between complete frames longer than a documented threshold. Compare only across group boundaries for @@ -51,7 +52,5 @@ excluded. ## Related -- [Starvation](/quest/m1/qos/starvation.md) - the egress half, same - histogram shape - [Client stats](/quest/m1/qos/stats/schema.md) - the publisher's own view of the same uplink, in the transport section of its stats broadcast diff --git a/quest/m1/qos/starvation-frames.md b/quest/m1/qos/starvation-frames.md index 34ab0a1adf..874c69528d 100644 --- a/quest/m1/qos/starvation-frames.md +++ b/quest/m1/qos/starvation-frames.md @@ -3,17 +3,19 @@ ## Goal Each subscription's acknowledged frontier moves at every frame boundary -instead of once per group, so the starvation histogram and dropped-media -counters from the group slice read a frontier at most one frame stale, and a -second histogram reports per-frame delivery delay so viewer jitter is visible -per broadcast. The wire shape from the group slice is unchanged apart from -the new histogram. +instead of once per group, so the `lag` histogram and `dropped_*` counters +on `stats::Traffic` read a frontier at most one frame stale, and a second +histogram reports per-frame delivery delay so viewer jitter is visible per +broadcast. The wire shape is unchanged apart from the new histogram. ## Plan A frame is atomic to the viewer: it is useful only once fully delivered. The -sampler from the parent quest is unchanged; what changes is how often the -frontier it reads moves. `Writer` in `rs/moq-net/src/coding/writer.rs` gains +sampler in `Registry::report` (`rs/moq-net/src/stats.rs`) is unchanged; what +changes is how often the frontier it reads moves. Today each group serve task +holds a `stats::Delivery` that advances the frontier when `Writer::poll_close` +reports the FIN acknowledged, and counts the group's whole written span as +dropped otherwise. `Writer` in `rs/moq-net/src/coding/writer.rs` gains a monotonic written-offset counter, since every write funnels through it, so at each frame end the per-group serve task knows the stream offset and the frame timestamp. Record `(offset, timestamp, bytes, written_at)` and await @@ -24,7 +26,7 @@ acknowledged in order for the purpose of this metric, so poll the oldest 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 parent quest read a frontier that +timestamp, so the interval samples 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 @@ -38,7 +40,7 @@ the return one-way delay and that a backend without the hook leaves the histogram absent, never zero. Backends that return unsupported from `poll_acked` keep the group-granularity -sampling from the parent quest, so the histogram never disappears over qmux or +`Delivery`, so the histogram never disappears over qmux or a browser transport; document which resolution a node offers. Tests: the frontier tracking frame ends under a peer that acknowledges in @@ -50,7 +52,5 @@ unsupported. ## Required -- [Starvation](/quest/m1/qos/starvation.md) - fixes the wire shape and the - group-granularity fallback - [poll_acked in web-transport](/quest/m1/quic/ack-hook.md) - the released hook this samples through diff --git a/quest/m1/qos/starvation.md b/quest/m1/qos/starvation.md deleted file mode 100644 index 829af00b10..0000000000 --- a/quest/m1/qos/starvation.md +++ /dev/null @@ -1,89 +0,0 @@ -# [M] Starvation: how far behind viewers are, from the relay - -## Goal - -The relay's `moq-stats` egress rows report, per broadcast, how far behind the -acknowledged frontier of its subscriptions is in media time, and how much -media was abandoned before the peer acknowledged it. A dashboard can read the -relative health of a broadcast's viewers from the relay's point of view -without any per-viewer row on the wire. - -## Plan - -This is the DELIVERY half of a health verdict and closes the byte-backlog -design in #2733: media-time lag is the primary signal, bytes are the weights. - -Define the frontier and the sample separately. Each subscription has an -acknowledged frontier: the newest frame timestamp the peer is known to have -received. At activation, before any frame is chosen or acknowledged, it is -the track's newest produced timestamp at that moment, so a subscription -blocked before its first write is still sampled, its lag is exactly the -media produced since it started, and the backlog it deliberately starts -behind (a latency floor, an ordered start) never reads as starvation. In this slice the frontier advances when a group -stream's FIN is acknowledged, which `Writer::close()` already awaits in -`Subscription::serve_group` in `rs/moq-net/src/lite/publisher.rs` and in the -IETF publisher, and the group's last frame timestamp becomes the frontier. - -The sample is taken on the stats producer's interval tick, not on any ACK -and not in the write path. Each tick walks the active subscriptions and -records `lag = newest frame timestamp the track has produced - the -subscription's frontier`, weighted by the bytes the track produced during -the interval. Both are timestamps on the same track, so the missing epoch -does not matter. The write path cannot be the sampler: a flow-controlled -subscription's serve tasks block inside `write`, which is precisely the -stalled viewer the metric exists to show, and sampling only on ACK would let -that viewer emit nothing and vanish from the snapshot deltas. A periodic -sample keeps it climbing the buckets while its frontier stands still. While -the source is paused nothing is produced, so a tick carries no weight and -the histogram does not move; a viewer's lag stays whatever it was until its -frontier catches up, and reappears in the buckets once production resumes. This slice moves the frontier once per group; -the [frame-granularity quest](/quest/m1/qos/starvation-frames.md) moves it -per frame without changing the wire shape or the sampler, so fix both here. - -Aggregate as a byte-weighted cumulative histogram on `Traffic`, on the -`Role::Publisher` (egress) rows of the existing `(tier, broadcast, role)` -key: fixed log-spaced buckets of media time, each a monotonic byte counter -(bytes produced for a subscription during intervals when its frontier lag -fell in that bucket). Propose buckets at 50 ms, -100 ms, 250 ms, 500 ms, 1 s, 2 s, 5 s, and above, and document the edges in -the stats section of `doc/bin/relay/config.md`. A consumer diffs two snapshots -for a byte-weighted distribution, percentiles, or mean over any interval. -This keeps the crate's cumulative-monotonic contract and the `.z` merge-patch -deltas intact; do not add gauges. - -Account for media that is never acknowledged. When a group stream is reset by -skip, expiry, subscriber stop, or session close before its FIN is -acknowledged, add the media duration the peer never got to a cumulative -`dropped_duration`, with `dropped_bytes` and `dropped_groups` beside it. The -duration is the group's newest written timestamp minus its newest acknowledged -timestamp; at group granularity nothing inside the group is known to be -acknowledged, so the whole written span counts, which over-reports until the -frame slice lands. Say so in the docs. Sum tracks per broadcast: the histogram -is in media time, so mixing audio and video is coherent, and byte weighting -already lets video dominate. - -Mechanics. The track's newest produced timestamp and interval byte count -live beside the existing per-frame bump points on the crate-private -`stats::Meter` in `rs/moq-net/src/stats.rs`; the frontier is a shared cell on -the `stats::Subscription` guard that the serve task advances on ACK and the -sampler reads. The serve task records the newest timestamp it has written to -each group so a drop can be sized. `moq-stats` sums histograms across nodes -in its aggregate consumer the same way it sums counters today. Update the stats -section of `doc/bin/relay/config.md` with the new fields and their meaning. - -Tests: a subscriber that reads at media rate lands in the lowest bucket; a -flow-controlled subscriber walks up the buckets as backlog grows, including -one that stops acknowledging entirely; a skipped group adds its span to -`dropped_duration` and nothing to the histogram beyond the interval samples -already taken while it was outstanding; a -session close mid-group counts as dropped; two subscriptions of one broadcast -sum into one row; the aggregate consumer sums two nodes bucket by bucket. - -## Related - -- [Starvation at frame granularity](/quest/m1/qos/starvation-frames.md) - - moves the frontier at every frame end once `poll_acked` is released -- [Publisher timeliness](/quest/m1/qos/publisher-timeliness.md) - the ingress - mirror on the `Role::Subscriber` rows -- [Client stats](/quest/m1/qos/stats/README.md) - receiver-side evidence - for the same lag, reported by the viewers themselves diff --git a/quest/m1/qos/stats/schema.md b/quest/m1/qos/stats/schema.md index 2a60d265ec..f6e696709b 100644 --- a/quest/m1/qos/stats/schema.md +++ b/quest/m1/qos/stats/schema.md @@ -64,8 +64,3 @@ what a subscriber received and played, per audio and video. The relay is `Traffic` and `Presence` serializers once a moq-stats release carrying both has shipped and the demo dashboard (and any moq.pro consumer) reads the `*_started` / `*_ended` names. Deserialize keeps accepting them. Wire only. - -## Related - -- [Starvation](/quest/m1/qos/starvation.md) - the relay's delivery counters, - the other half of a health verdict diff --git a/quest/m1/quic/README.md b/quest/m1/quic/README.md index f3234590a4..7fc8d4720f 100644 --- a/quest/m1/quic/README.md +++ b/quest/m1/quic/README.md @@ -90,8 +90,6 @@ This is a transport API change, not a MoQ wire change. - [Scope track priority](/quest/m1/track-priority-scope.md) - the per-broadcast fairness policy on cluster sessions -- [Starvation](/quest/m1/qos/starvation.md) - the first consumer of ACK - progress: how far behind viewers are, from the relay's point of view - [Receive timestamps](/quest/m2/quic-receive-ts.md) - per-packet arrival times for GCC and deadlines - [GCC egress experiment](/quest/m2/quic-gcc.md) - a measured verdict on diff --git a/quest/m1/track-priority-scope.md b/quest/m1/track-priority-scope.md index 4e4552e432..37bb1a6aa5 100644 --- a/quest/m1/track-priority-scope.md +++ b/quest/m1/track-priority-scope.md @@ -72,8 +72,3 @@ change. - [Hierarchical stream scheduling](/quest/m1/quic/scheduler.md) - the fair tier the per-broadcast send groups ride on - -## Related - -- [Starvation](/quest/m1/qos/starvation.md) - the relay-side signal that - shows a starved subscription diff --git a/rs/moq-net/Cargo.toml b/rs/moq-net/Cargo.toml index b65a6d5402..69d3208761 100644 --- a/rs/moq-net/Cargo.toml +++ b/rs/moq-net/Cargo.toml @@ -29,6 +29,7 @@ moq-pattern = { workspace = true, features = ["serde"] } num_enum = { workspace = true } rand = { workspace = true } serde = { workspace = true } +serde_with = { workspace = true } slab = { workspace = true } thiserror = { workspace = true } tracing = { workspace = true } diff --git a/rs/moq-net/benches/stats.rs b/rs/moq-net/benches/stats.rs index 299b7fe9d7..7afbb61861 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 and tiers so a per-drain cost that -//! grows faster than the entries it reports shows up as a slope. +//! grows faster than the entries it reports shows up as a slope, and over +//! broadcasts and 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, criterion_group, criterion_main}; use futures::FutureExt; -use moq_net::{announce, broadcast, origin, stats}; +use moq_net::{Timestamp, announce, broadcast, origin, stats, track}; /// `(broadcasts, tiers)` shapes; every broadcast records under every tier. const SHAPES: [(usize, usize); 4] = [(100, 1), (1_000, 1), (100, 10), (1_000, 10)]; @@ -64,5 +66,67 @@ fn report(c: &mut Criterion) { group.finish(); } -criterion_group!(benches, report); +/// `(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, each track holding one frame. +struct Subscribed { + registry: stats::Registry, + _driver: origin::Driver, + _publishers: Vec<(broadcast::Producer, track::Producer)>, + _subscribers: Vec, +} + +fn subscribed(broadcasts: usize, subscriptions: usize) -> Subscribed { + let registry = stats::Registry::new(stats::Config::new()); + let (producer, mut driver) = origin::Producer::new(origin::Config::default()); + let waiter = kio::Waiter::noop(); + 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", None).unwrap(); + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, vec![0u8; 100]).unwrap(); + group.finish().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()); + 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: publishers, + _subscribers: subscribers, + } +} + +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.registry.report(&mut report); + let subscribed: u64 = report.traffic.iter().map(|e| e.publisher.subscriptions_started).sum(); + assert_eq!(subscribed, (broadcasts * subscriptions) as u64, "every subscription is tagged"); + + group.bench_with_input( + BenchmarkId::from_parameter(format!("{broadcasts}x{subscriptions}")), + &fixture, + |b, fixture| b.iter(|| fixture.registry.report(&mut report)), + ); + } + group.finish(); +} + +criterion_group!(benches, report, sample); criterion_main!(benches); diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 1b4d7f3af6..d7c8c3a04b 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -2014,6 +2014,7 @@ impl TrackServe { self.timescale, self.version, slice, + self.track.delivery(), )); } Poll::Ready(Ok(None)) => { @@ -2040,6 +2041,9 @@ struct GroupServe { timescale: Option, version: Version, object_delta: 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, } @@ -2084,6 +2088,7 @@ impl GroupServe { timescale: Option, version: Version, slice: GroupSlice, + delivery: crate::stats::Delivery, ) -> Self { group.skip_to(slice.skip); group.end_at(slice.until.map_or(Bound::Unbounded, Bound::Excluded)); @@ -2096,6 +2101,7 @@ impl GroupServe { timescale, version, object_delta, + delivery, state: GroupState::Open, } } @@ -2199,6 +2205,7 @@ impl GroupServe { ) { break 'serve Err(err); } + self.delivery.wrote(batched.timestamp, batched.payload.len() as u64); let payload = std::mem::take(&mut batched.payload); if !payload.is_empty() { *chunk = Some(payload); @@ -2228,6 +2235,7 @@ impl GroupServe { ) { break 'serve Err(err); } + self.delivery.wrote(next.timestamp, next.size); // An empty object has no payload to stream. if next.size > 0 { *frame = Some(next); @@ -2263,6 +2271,7 @@ impl GroupServe { let sequence = self.msg.group_id; self.state = GroupState::Done; return Poll::Ready(res.map(|()| { + std::mem::take(&mut self.delivery).acked(); tracing::debug!(sequence, "finished group"); })); } @@ -2368,6 +2377,7 @@ mod group_priority_test { Some(Timescale::default()), Version::Draft14, GroupSlice::default(), + Default::default(), ); kio::wait(|waiter| serve.poll_serve(waiter)).await.unwrap(); @@ -2430,6 +2440,69 @@ mod group_priority_test { 60 } + /// An acknowledged FIN advances the subscription's frontier; a subgroup abandoned + /// mid-stream counts as dropped instead. + #[tokio::test] + async fn subgroups_report_to_the_frontier() { + use futures::FutureExt; + + let registry = crate::stats::Registry::new(crate::stats::Config::new()); + let session = registry.tier(crate::stats::Tier::default()).session("root"); + let broadcast = crate::broadcast::Info::new().produce(); + let track = broadcast.create_track("test", None).unwrap(); + let sub = broadcast + .consume() + .with_stats(session.egress("demo")) + .track("test") + .unwrap() + .subscribe(None) + .now_or_never() + .unwrap() + .unwrap(); + let serve = |group: group::Consumer| { + GroupServe::new( + SinkSession::new(Default::default()), + ietf::GroupHeader { + track_alias: 0, + group_id: group.sequence, + sub_group_id: 0, + publisher_priority: 0, + flags: Default::default(), + }, + 0, + group, + Some(Timescale::default()), + Version::Draft14, + GroupSlice::default(), + sub.delivery(), + ) + }; + let ms = |ms| crate::Timestamp::from_millis(ms).unwrap(); + + 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(group.consume()); + kio::wait(|waiter| done.poll_serve(waiter)).await.unwrap(); + + let mut group = track.create_group(group::Info { sequence: 1 }).unwrap(); + group.write_frame(ms(1000), b"hi".as_slice()).unwrap(); + group.write_frame(ms(1250), b"yo".as_slice()).unwrap(); + let mut abandoned = serve(group.consume()); + assert!(kio::wait(|waiter| abandoned.poll_serve(waiter)).now_or_never().is_none()); + drop(abandoned); + + let mut report = crate::stats::Report::default(); + registry.report(&mut report); + let row = report.traffic[0].publisher; + assert_eq!(row.dropped_groups, 1); + assert_eq!(row.dropped_bytes, 4); + assert_eq!(row.dropped_duration, Duration::from_millis(250)); + // 1250ms produced against a 500ms frontier: the [500ms, 1s) bucket. + assert_eq!(row.lag.buckets(), &[0, 0, 0, 0, 6, 0, 0, 0]); + } + /// A subgroup waiting for stream credit keeps its subscription expiry armed. #[tokio::test] async fn group_waiting_for_stream_credit_expires() { @@ -2458,6 +2531,7 @@ mod group_priority_test { Some(Timescale::default()), Version::Draft19, GroupSlice::default(), + Default::default(), ); let mut serving = std::pin::pin!(kio::wait(|waiter| serve.poll_serve(waiter))); assert!( @@ -2506,6 +2580,7 @@ mod group_priority_test { Some(Timescale::default()), Version::Draft19, GroupSlice::default(), + Default::default(), ); let mut serving = std::pin::pin!(kio::wait(|waiter| serve.poll_serve(waiter))); // Let it run until it blocks on the rest of the frame. @@ -3458,6 +3533,7 @@ mod serve_tests { Some(Timescale::default()), Version::Draft20, slice, + Default::default(), ); kio::wait(|waiter| serve.poll_serve(waiter)).await.unwrap(); diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index a841eefa34..5ddf433b1e 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -2183,8 +2183,9 @@ impl Subscription { frame_start: u64, priority: PriorityHandle, group: group::Consumer, + delivery: crate::stats::Delivery, ) -> Result<(), Error> { - let mut serve = Box::new(GroupServe::new(self, sequence, frame_start, priority, group)); + let mut serve = Box::new(GroupServe::new(self, sequence, frame_start, priority, group, delivery)); kio::wait(move |waiter| serve.poll_serve(waiter)).await } } @@ -2341,8 +2342,15 @@ impl TrackRun { .ctx .priority .insert(Priority::new(current_priority, self.ctx.id, sequence)); - self.children - .push(GroupServe::new(self.ctx.clone(), sequence, frame_start, handle, group)); + let delivery = self.track.delivery(); + self.children.push(GroupServe::new( + self.ctx.clone(), + sequence, + frame_start, + handle, + group, + delivery, + )); } Recv::Datagram(datagram) => self.ctx.serve_datagram(datagram), Recv::Boundary(group) => { @@ -2376,6 +2384,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, } @@ -2409,6 +2420,7 @@ impl GroupServe { frame_start: u64, priority: PriorityHandle, group: group::Consumer, + delivery: crate::stats::Delivery, ) -> Self { Self { ctx, @@ -2417,6 +2429,7 @@ impl GroupServe { sequence, frame_start, prev_ts: 0, + delivery, state: GroupState::Open, } } @@ -2549,6 +2562,7 @@ impl GroupServe { if let Err(err) = buffered { break 'serve Err(err); } + self.delivery.wrote(batched.timestamp, batched.payload.len() as u64); let payload = std::mem::take(&mut batched.payload); if !payload.is_empty() { *chunk = Some(payload); @@ -2578,6 +2592,7 @@ impl GroupServe { if let Err(err) = buffered { break 'serve Err(err); } + self.delivery.wrote(next.timestamp, next.size); *frame = Some(next); } Poll::Ready(Ok(None)) => break 'serve Ok(()), @@ -2617,6 +2632,7 @@ impl GroupServe { let res = ready!(writer.poll_close(&mut cx)); self.state = GroupState::Done; return Poll::Ready(res.map(|()| { + std::mem::take(&mut self.delivery).acked(); tracing::debug!(sequence = self.sequence, "finished group"); })); } @@ -2777,7 +2793,7 @@ mod serve_group_test { .unwrap(); let handle = subscription.priority.insert(Priority::new(0, 0, 0)); - let mut serve = std::pin::pin!(subscription.serve_group(0, 0, handle, group.consume())); + let mut serve = std::pin::pin!(subscription.serve_group(0, 0, handle, group.consume(), Default::default())); // Drain the frame, leaving the task parked awaiting the next one. assert!(futures::poll!(serve.as_mut()).is_pending()); @@ -2818,7 +2834,7 @@ mod serve_group_test { let group = subscriber.recv_group().await.unwrap().expect("old group"); let handle = subscription.priority.insert(Priority::new(0, 0, 0)); - let mut serve = std::pin::pin!(subscription.serve_group(0, 0, handle, group)); + let mut serve = std::pin::pin!(subscription.serve_group(0, 0, handle, group, Default::default())); assert!( futures::poll!(serve.as_mut()).is_pending(), "transport write is blocked" @@ -2867,7 +2883,7 @@ mod serve_group_test { let group = subscriber.recv_group().await.unwrap().expect("old group"); let handle = subscription.priority.insert(Priority::new(0, 0, 0)); - let mut serve = std::pin::pin!(subscription.serve_group(0, 0, handle, group)); + let mut serve = std::pin::pin!(subscription.serve_group(0, 0, handle, group, Default::default())); assert!( futures::poll!(serve.as_mut()).is_pending(), "waiting for the final byte" @@ -2924,7 +2940,7 @@ mod serve_group_test { let group = subscriber.recv_group().await.unwrap().expect("old group"); let handle = subscription.priority.insert(Priority::new(0, 0, 0)); - let mut serve = std::pin::pin!(subscription.serve_group(0, 0, handle, group)); + let mut serve = std::pin::pin!(subscription.serve_group(0, 0, handle, group, Default::default())); assert!( futures::poll!(serve.as_mut()).is_pending(), "stream credit is exhausted" @@ -2964,6 +2980,84 @@ mod serve_group_test { ); } + /// An acknowledged FIN advances the subscription's frontier; a group reset or + /// abandoned mid-stream (the session closing drops its serve task) counts as + /// dropped instead. + #[tokio::test] + async fn group_streams_report_to_the_frontier() { + let registry = crate::stats::Registry::new(crate::stats::Config::new()); + let session = registry.tier(crate::stats::Tier::default()).session("root"); + let broadcast = broadcast::Info::new().produce(); + let track = broadcast.create_track("test", None).unwrap(); + let sub = broadcast + .consume() + .with_stats(session.egress("demo")) + .track("test") + .unwrap() + .subscribe(None) + .now_or_never() + .unwrap() + .unwrap(); + let row = || { + let mut report = crate::stats::Report::default(); + registry.report(&mut report); + report.traffic.iter().find(|e| e.path.as_str() == "demo").unwrap().publisher + }; + let subscription = || Subscription { + session: SinkSession::new(Log::default()), + id: 0, + track_name: "test".into(), + priority: PriorityQueue::default(), + track_priority: kio::Producer::new(0u8).consume(), + track_priority_seen: 0, + version: Version::Lite06, + timescale: Some(crate::Timescale::default()), + }; + let ms = |ms| Timestamp::from_millis(ms).unwrap(); + + // Served to completion: 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 ctx = subscription(); + let handle = ctx.priority.insert(Priority::new(0, 0, 0)); + ctx.serve_group(0, 0, handle, group.consume(), sub.delivery()) + .await + .unwrap(); + let before = row(); + + // Aborted mid-stream after one frame. + let mut group = track.create_group(group::Info { sequence: 1 }).unwrap(); + group.write_frame(ms(1000), b"hello".as_slice()).unwrap(); + let ctx = subscription(); + let handle = ctx.priority.insert(Priority::new(0, 0, 1)); + let mut serve = std::pin::pin!(ctx.serve_group(1, 0, handle, group.consume(), sub.delivery())); + assert!(futures::poll!(serve.as_mut()).is_pending()); + group.abort(Error::Old).unwrap(); + assert!(matches!(serve.await, Err(Error::Old))); + + // Abandoned mid-stream: the session closed and dropped the serve task. + let mut group = track.create_group(group::Info { sequence: 2 }).unwrap(); + group.write_frame(ms(2000), b"hi".as_slice()).unwrap(); + group.write_frame(ms(2100), b"yo".as_slice()).unwrap(); + let ctx = subscription(); + let handle = ctx.priority.insert(Priority::new(0, 0, 2)); + let mut serve = Box::pin(ctx.serve_group(2, 0, handle, group.consume(), sub.delivery())); + assert!(futures::poll!(serve.as_mut()).is_pending()); + drop(serve); + + let after = row(); + assert_eq!(before.dropped_groups, 0); + assert_eq!(after.dropped_groups, 2); + assert_eq!(after.dropped_bytes, 5 + 4); + assert_eq!(after.dropped_duration, Duration::from_millis(100)); + // The newest frame is at 2100ms and the frontier at 500ms: 1.6s behind, for + // the 9 bytes produced since the last sample. + assert_eq!(before.lag.buckets(), &[2, 0, 0, 0, 0, 0, 0, 0]); + assert_eq!(after.lag.buckets(), &[2, 0, 0, 0, 0, 9, 0, 0]); + } + /// A group that completes cleanly must not reset at all. The completion path /// releases the stream via `poll_close`; leaving the writer to drop after /// `finish()` would fire the Drop fallback and tack a spurious Cancel reset @@ -2994,7 +3088,7 @@ mod serve_group_test { group.finish().unwrap(); let handle = subscription.priority.insert(Priority::new(0, 0, 0)); - subscription.serve_group(0, 0, handle, consumer).await.unwrap(); + subscription.serve_group(0, 0, handle, consumer, Default::default()).await.unwrap(); assert_eq!(log.resets(), Vec::::new(), "clean completion must not reset"); diff --git a/rs/moq-net/src/model/cache.rs b/rs/moq-net/src/model/cache.rs index 6bdb7cfe23..2aead96582 100644 --- a/rs/moq-net/src/model/cache.rs +++ b/rs/moq-net/src/model/cache.rs @@ -490,6 +490,10 @@ pub(crate) struct Track { // This account's slot in the pool's sweep registry, absent when the pool has no // expiry window (nothing is registered) or for the detached default account. sweep: OnceLock, + + // What the track's groups have produced, for the egress lag sampler. Kept on the + // account because it is already the one per-track handle every group holds. + production: crate::stats::Production, } impl Track { @@ -502,6 +506,7 @@ impl Track { expiry_cursor: AtomicUsize::new(0), state, sweep: OnceLock::new(), + production: Default::default(), }); if let Some(key) = track.pool.register(&track) { let _ = track.sweep.set(key); @@ -514,6 +519,11 @@ impl Track { &self.pool } + /// What this track's groups have produced. + pub(crate) fn production(&self) -> &crate::stats::Production { + &self.production + } + /// Charge a new group's fixed overhead, returning its [`Charge`]. pub(crate) fn charge(self: &Arc) -> Charge { self.pool.add(ENTRY_OVERHEAD); diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 0205ae10b4..58e554a165 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -519,6 +519,7 @@ impl Producer { // Ingress payload: one whole frame written. self.stats.frames(1); self.stats.bytes(size); + self.cache.production().record(timestamp, size); Ok(()) } @@ -571,6 +572,7 @@ impl Producer { state.cache += size; now = state.charge.add(size); state.stamp(frame.timestamp); + self.cache.production().record(frame.timestamp, size); state.frames.push_back(frame); } state.next_index = next_index; @@ -632,8 +634,9 @@ impl Producer { self.cache.settle(now); // Ingress payload: one frame opened; its bytes are counted per chunk as the - // frame::Producer writes them. + // frame::Producer writes them. Production counts the declared size up front. self.stats.frames(1); + self.cache.production().record(timestamp, frame.size); let meter = self.stats.clone(); let info = frame::Info { @@ -688,8 +691,9 @@ impl Producer { self.cache.settle(now); // Ingress payload: one frame opened; its bytes are counted per chunk as the - // producer writes them. + // producer writes them. Production counts the declared size up front. self.stats.frames(1); + self.cache.production().record(timestamp, frame.size); let meter = self.stats.clone(); let info = frame::Info { diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index e2227a17f1..dbaa231210 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -630,6 +630,7 @@ impl Consumer { end_sequence: None, outer: Anchor::default(), drift_anchor: kio::Producer::new(Anchor::default()), + frontier: Default::default(), } } @@ -1381,6 +1382,9 @@ struct SegmentSub { /// Set while this is a warm segment (see [`Producer::park`]) that has not been /// cleared for live reads: the copy spliced after it, once there is one. warm: Option, + /// The logical subscription's acknowledged frontier, told about this segment's + /// track once its cursor activates. + frontier: crate::stats::Frontier, } /// A warm segment waiting on the copy spliced after it; see [`Subscriber::poll_activate`]. @@ -1497,6 +1501,22 @@ pub struct Subscriber { /// with the newest edge across the segments), shared with groups that outlive this /// cursor poll. Refreshed on every [`Self::poll_sync`]. drift_anchor: kio::Producer, + /// The acknowledged frontier of the logical subscription, told about every + /// segment's track so the lag sampler weighs what the splice produces. + frontier: crate::stats::Frontier, +} + +impl Subscriber { + /// Report every segment's track, current and future, to `frontier`. + pub(crate) fn watch(&mut self, frontier: &crate::stats::Frontier) { + self.frontier = frontier.clone(); + for seg in &mut self.segments { + seg.frontier = frontier.clone(); + if let SubState::Active(sub) = &mut seg.sub { + sub.watch(frontier); + } + } + } } impl Subscriber { @@ -1667,6 +1687,7 @@ impl Subscriber { edge: segment.track.latest(), next, }), + frontier: self.frontier.clone(), }); } } @@ -1845,6 +1866,7 @@ impl Subscriber { sub.raise_start_to(seg.first_group().max(min_sequence)); sub.set_anchor(seg.anchor.clone()); let _ = sub.update(slice(prefs, seg.ask, seg.end)); + sub.watch(&seg.frontier); seg.sub = SubState::Active(Box::new(sub)); } // The underlying track was rejected or closed: stall, not error. @@ -2479,6 +2501,49 @@ mod test { recv_pending(&mut sub); } + /// A spliced subscription's lag weighs what every segment's track produces, so a + /// route-fed broadcast (every relay hop) is sampled across failovers. + #[tokio::test] + async fn lag_weighs_every_segment() { + use crate::stats; + + let registry = stats::Registry::new(stats::Config::new()); + let session = registry.tier(stats::Tier::default()).session("root"); + let weight = || { + let mut report = stats::Report::default(); + registry.report(&mut report); + report.traffic[0].publisher.lag.total() + }; + let (mut track_a, consumer_a) = track_pair("a"); + let (mut track_b, consumer_b) = track_pair("b"); + let mut producer = Producer::new(); + producer.switch(&consumer_a, None).unwrap(); + + let mut sub = track::Consumer::spliced("video".into(), Arc::new(broadcast::Info::default()), producer.consume()) + .with_stats(session.egress("demo")) + .subscribe(replay()) + .now_or_never() + .unwrap() + .unwrap(); + let pending = |sub: &mut track::Subscriber| { + assert!( + kio::wait(|waiter| sub.poll_recv_group(waiter)) + .now_or_never() + .is_none() + ); + }; + // Polling activates the segment's cursor, which names its track to the frontier. + pending(&mut sub); + write_group_at(&mut track_a, 0, "a0", Duration::ZERO); + assert_eq!(weight(), 2); + + producer.switch(&consumer_b, Position::group(1)).unwrap(); + let _ = kio::wait(|waiter| sub.poll_recv_group(waiter)).now_or_never(); + pending(&mut sub); + write_group_at(&mut track_b, 1, "b1", Duration::from_secs(1)); + assert_eq!(weight(), 4); + } + #[tokio::test] async fn demand_reflects_boundaries() { let (track_a, consumer_a) = track_pair("a"); diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index ba83c78728..25902cd8d7 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1632,7 +1632,7 @@ impl Producer { }), // 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(), } } @@ -2761,6 +2761,9 @@ impl Subscribing { let drift_anchor = kio::Producer::new(Anchor::default()); let min_sequence = floor_of(&self.subscription.read()); + let stats_sub = self.stats.subscribe(); + let cache = state.read().cache.clone(); + stats_sub.frontier().watch(&cache); Poll::Ready(Ok(Subscriber { name: self.name.clone(), broadcast: self.broadcast.clone(), @@ -2780,7 +2783,7 @@ impl Subscribing { seek_pending: BTreeMap::new(), }), stats: self.stats.clone(), - _stats_sub: self.stats.subscribe(), + stats_sub, })) } SubscribingKind::Spliced(resume) => { @@ -2788,13 +2791,16 @@ impl Subscribing { // window is applied to each per-session aggregate, not here. let info = ready!(resume.poll_info(waiter))?; + let stats_sub = self.stats.subscribe(); + let mut spliced = resume.subscribe_shared(self.subscription.clone()); + spliced.watch(stats_sub.frontier()); Poll::Ready(Ok(Subscriber { name: self.name.clone(), broadcast: self.broadcast.clone(), info, - inner: SubscriberKind::Spliced(Box::new(resume.subscribe_shared(self.subscription.clone()))), + inner: SubscriberKind::Spliced(Box::new(spliced)), stats: self.stats.clone(), - _stats_sub: self.stats.subscribe(), + stats_sub, })) } } @@ -3053,8 +3059,9 @@ 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 carries the acknowledged frontier the + // lag sampler reads. Empty (no-op) for an untagged track. + stats_sub: stats::Subscription, } enum SubscriberKind { @@ -4017,6 +4024,23 @@ impl Subscriber { Ok(()) } + /// Track one group stream written toward the peer against this subscription's + /// acknowledged frontier. A no-op for an untagged subscriber. + pub(crate) fn delivery(&self) -> stats::Delivery { + self.stats_sub.frontier().delivery() + } + + /// Report the tracks feeding this cursor to `frontier`: its own, or each segment's. + pub(crate) fn watch(&mut self, frontier: &stats::Frontier) { + match &mut self.inner { + SubscriberKind::Plain(plain) => { + let cache = plain.state.read().cache.clone(); + frontier.watch(&cache); + } + SubscriberKind::Spliced(spliced) => spliced.watch(frontier), + } + } + /// Return the latest sequence number in the track. pub fn latest(&self) -> Option { match &self.inner { diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index 7f84c93a37..e8a0792d62 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -59,6 +59,20 @@ //! Driven by [`Handle::session`] (the [`Session`] context); a //! [`Session::set_tier`] ends the session on the old tier and starts it on //! the new one. +//! * `lag` (egress only): a byte-weighted [`Histogram`] of how far behind each +//! subscription's acknowledged frontier is, in media time. 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 walks the live subscriptions and records `max(newest produced - +//! frontier, wall time the frontier has stood still while behind)` weighted +//! by the bytes the track produced since the previous call. A paused source +//! produces nothing, so it adds no weight. +//! * `dropped_duration` / `dropped_bytes` / `dropped_groups` (egress only): +//! media written toward a peer on a group stream that was reset or abandoned +//! before its FIN was acknowledged. The duration is the span between the +//! group's oldest and newest written frame timestamps; nothing inside a group +//! is known to be acknowledged, so the whole written span counts. //! //! Counters are strictly monotonic (only `fetch_add`); a counter going //! backwards across reads means the underlying entry was garbage collected @@ -111,15 +125,17 @@ use std::{ collections::HashMap, fmt, sync::{ - Arc, Mutex, + Arc, Mutex, Weak, atomic::{AtomicU64, Ordering}, }, + time::Duration, }; use kio::Lock; use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use serde_with::{DurationMilliSecondsWithFrac, serde_as}; -use crate::{AsPath, PathOwned, Pattern, Patterns}; +use crate::{AsPath, PathOwned, Pattern, Patterns, Timestamp, cache, runtime::Instant}; /// Cumulative atomic counters for a single `(tier, role)` on a broadcast. /// @@ -159,6 +175,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: media written on a group stream that ended unacknowledged. + dropped_duration_nanos: AtomicU64, + dropped_bytes: AtomicU64, + dropped_groups: AtomicU64, + // Egress only: the live subscriptions `sample` walks. Pruned there once dead. + frontiers: Mutex>>, } /// Atomic backing for one [`Content`] readout. @@ -210,6 +234,10 @@ impl Counters { let groups = self.groups.load(Ordering::Relaxed); let datagrams = self.datagrams.load(Ordering::Relaxed); let stale = self.stale.snapshot(); + let lag = self.lag.snapshot(); + let dropped_duration = Duration::from_nanos(self.dropped_duration_nanos.load(Ordering::Relaxed)); + let dropped_bytes = self.dropped_bytes.load(Ordering::Relaxed); + let dropped_groups = self.dropped_groups.load(Ordering::Relaxed); Traffic { announces_started, announces_ended, @@ -224,6 +252,152 @@ impl Counters { groups, datagrams, stale, + lag, + dropped_duration, + dropped_bytes, + dropped_groups, + } + } + + /// Take one lag sample from every live egress subscription, pruning the dead. + fn sample(&self, now: Instant) { + let mut frontiers = self.frontiers.lock().expect("stats frontiers poisoned"); + frontiers.retain(|frontier| { + let Some(frontier) = frontier.upgrade() else { + return false; + }; + if let Some((lag, weight)) = frontier.sample(now) { + self.lag.record(lag, weight); + } + 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. + pub fn total(&self) -> u64 { + self.buckets.iter().sum() + } + + /// True when nothing has been recorded. + pub fn is_empty(&self) -> bool { + self.buckets.iter().all(|b| *b == 0) + } + + /// Fold another readout into this one, bucket by bucket. + pub fn add(&mut self, other: &Histogram) { + for (bucket, other) in self.buckets.iter_mut().zip(other.buckets) { + *bucket += 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)), } } } @@ -330,6 +504,18 @@ 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, in media + /// time, weighted by the bytes the track produced while it was. Sampled on each + /// [`Registry::report`]; see the module docs. + pub lag: Histogram, + /// Egress only: media written on group streams that were reset or abandoned + /// before the peer acknowledged them. Group granularity counts each group's + /// whole written span, so this over-reports media that did arrive. + pub dropped_duration: Duration, + /// Egress only: bytes written on those abandoned group streams. + pub dropped_bytes: u64, + /// Egress only: group streams abandoned after writing at least one frame. + pub dropped_groups: u64, } /// One spelling of a counter edge on the wire: absent, or a present integer. @@ -351,6 +537,7 @@ fn counter_edge(canonical: Edge, legacy: Edge) -> u64 { canonical.0.or(legacy.0).unwrap_or(0) } +#[serde_as] #[derive(Serialize)] struct TrafficSer { announces_started: u64, @@ -372,6 +559,15 @@ struct TrafficSer { groups: u64, datagrams: u64, stale: Content, + #[serde(skip_serializing_if = "Histogram::is_empty")] + lag: Histogram, + #[serde_as(as = "DurationMilliSecondsWithFrac")] + #[serde(skip_serializing_if = "Duration::is_zero")] + dropped_duration: Duration, + #[serde(skip_serializing_if = "is_zero")] + dropped_bytes: u64, + #[serde(skip_serializing_if = "is_zero")] + dropped_groups: u64, } impl From for TrafficSer { @@ -396,10 +592,15 @@ impl From for TrafficSer { groups: t.groups, datagrams: t.datagrams, stale: t.stale, + lag: t.lag, + dropped_duration: t.dropped_duration, + dropped_bytes: t.dropped_bytes, + dropped_groups: t.dropped_groups, } } } +#[serde_as] #[derive(Default, Deserialize)] #[serde(default)] struct TrafficDe { @@ -422,6 +623,11 @@ struct TrafficDe { groups: u64, datagrams: u64, stale: Content, + lag: Histogram, + #[serde_as(as = "DurationMilliSecondsWithFrac")] + dropped_duration: Duration, + dropped_bytes: u64, + dropped_groups: u64, } impl From for Traffic { @@ -440,6 +646,10 @@ impl From for Traffic { groups: d.groups, datagrams: d.datagrams, stale: d.stale, + lag: d.lag, + dropped_duration: d.dropped_duration, + dropped_bytes: d.dropped_bytes, + dropped_groups: d.dropped_groups, } } } @@ -472,6 +682,10 @@ impl Traffic { self.groups += other.groups; self.datagrams += other.datagrams; self.stale.add(other.stale); + self.lag.add(&other.lag); + self.dropped_duration += other.dropped_duration; + self.dropped_bytes += other.dropped_bytes; + self.dropped_groups += other.dropped_groups; } /// True while the broadcast is announced (an announce guard is open). @@ -953,6 +1167,10 @@ impl Registry { /// Refill `report` with the per-broadcast detail and prune dead entries. /// + /// Each call is also the lag sampler's tick: every live egress subscription + /// records one [`Traffic::lag`] sample weighted by what its track produced + /// since the previous call, so call it on a steady interval. + /// /// Clears `report`, keeping its capacity so a caller draining on an /// interval reuses one report instead of allocating per drain, then fills /// every `(broadcast, tier)` traffic readout and every `(tier, root)` @@ -967,11 +1185,14 @@ impl Registry { let Some(shared) = self.shared.as_ref() else { return; }; + // Every report is a lag sample; see `Traffic::lag`. + let now = crate::model::clock::now(); { let mut entries = shared.entries.lock(); for (path, entry) in entries.iter() { let tiers = entry.tiers.lock().expect("stats tiers poisoned"); for (tier, counters) in tiers.iter() { + counters.publisher.sample(now); report.traffic.push(TrafficEntry { path: path.clone(), tier: tier.clone(), @@ -1401,22 +1622,25 @@ 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. + // 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, + frontier, } } @@ -1456,6 +1680,16 @@ pub(crate) struct Subscription { side: Side, /// `Some((session, path))` on the egress side, to release the viewer refcount. viewer: Option<(Session, PathOwned)>, + /// The acknowledged frontier the lag sampler reads; empty off the egress side. + frontier: Frontier, +} + +impl Subscription { + /// The subscription's acknowledged frontier, for the model to name the tracks + /// feeding it and the wire to advance it. + pub(crate) fn frontier(&self) -> &Frontier { + &self.frontier + } } impl Drop for Subscription { @@ -1492,6 +1726,217 @@ impl Drop for Announce { } } +/// A media timestamp as a plain duration from zero, the unit every lag comparison +/// uses. Saturates past `u64` nanoseconds (about 584 years). +fn media_time(timestamp: Timestamp) -> Duration { + Duration::from_nanos(u64::try_from(timestamp.as_nanos()).unwrap_or(u64::MAX)) +} + +/// What one track has produced: the lag sampler's weight and the edge it measures a +/// subscription's frontier against. Bumped beside the ingress [`Meter`] on every +/// frame write, whether or not the track is tagged, since its readers may be. +#[derive(Default, Debug)] +pub(crate) struct Production { + // Media-time nanoseconds plus one, so zero means nothing produced yet. + first: AtomicU64, + newest: AtomicU64, + // Cumulative frame payload bytes. + bytes: AtomicU64, +} + +impl Production { + /// Record one frame of `bytes` at `timestamp`. + pub(crate) fn record(&self, timestamp: Timestamp, bytes: u64) { + let at = u64::try_from(media_time(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); + self.bytes.fetch_add(bytes, Ordering::Relaxed); + } + + 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, +} + +struct FrontierState { + /// The newest timestamp the peer acknowledged, seeded with the track's newest + /// when the subscription opened, so a backlog it deliberately starts behind never + /// reads as lag. `None` until either exists. + acked: Option, + /// When `acked` last moved, or when the subscription opened. + advanced: Instant, + /// When a sample first found the subscription behind since `advanced`. + behind: Option, + /// The tracks feeding the subscription: one, or one per segment of a splice. + sources: Vec, +} + +/// A track feeding a [`Frontier`], with the produced bytes already sampled. +struct Source { + track: Weak, + sampled: u64, +} + +impl Frontier { + fn open(counters: &Arc) -> Self { + let inner = Arc::new(FrontierInner { + counters: counters.clone(), + state: Mutex::new(FrontierState { + acked: None, + advanced: crate::model::clock::now(), + behind: None, + sources: Vec::new(), + }), + }); + counters + .publisher + .frontiers + .lock() + .expect("stats frontiers poisoned") + .push(Arc::downgrade(&inner)); + Self(Some(inner)) + } + + /// Add a track feeding this subscription. Idempotent. The first track seeds the + /// frontier with its newest frame. + pub(crate) fn watch(&self, track: &Arc) { + let Some(inner) = &self.0 else { return }; + let mut state = inner.state.lock().expect("stats frontier poisoned"); + if state.sources.iter().any(|s| std::ptr::eq(s.track.as_ptr(), Arc::as_ptr(track))) { + return; + } + let production = track.production(); + if state.acked.is_none() { + state.acked = Production::load(&production.newest); + } + state.sources.push(Source { + track: Arc::downgrade(track), + sampled: production.bytes.load(Ordering::Relaxed), + }); + } + + /// Start tracking one group stream written toward the peer. + pub(crate) fn delivery(&self) -> Delivery { + Delivery { + frontier: self.0.clone(), + ..Default::default() + } + } +} + +impl FrontierInner { + /// One sample: the lag and the bytes produced since the last one, or `None` when + /// nothing was produced (a paused source moves nothing). + fn sample(&self, now: Instant) -> Option<(Duration, u64)> { + let mut state = self.state.lock().expect("stats frontier poisoned"); + let mut weight = 0; + let mut newest = None; + let mut first: Option = None; + state.sources.retain_mut(|source| { + // Only this Weak outliving the track means it can never produce again. + let Some(track) = source.track.upgrade() else { + return false; + }; + let production = track.production(); + let bytes = production.bytes.load(Ordering::Relaxed); + weight += bytes.saturating_sub(source.sampled); + source.sampled = bytes; + newest = newest.max(Production::load(&production.newest)); + if let Some(at) = Production::load(&production.first) { + first = Some(first.map_or(at, |first| first.min(at))); + } + true + }); + let newest = newest?; + // Opened before the track produced anything: its first frame is where the + // subscription started. + let acked = *state.acked.get_or_insert(first?); + + // Tracked even without weight, so a pause the peer caught up in doesn't read + // as a stalled frontier once production resumes. + let behind = newest.saturating_sub(acked); + let lag = if behind.is_zero() { + state.behind = None; + Duration::ZERO + } else { + // Media time alone trusts the publisher's timestamps; a frontier standing + // still on the wall clock while there is newer media is lag too. + let since = state.behind.unwrap_or(now).max(state.advanced); + state.behind = Some(since); + behind.max(now.duration_since(since)) + }; + (weight > 0).then_some((lag, weight)) + } +} + +/// One group stream written toward a peer, reported to its subscription's +/// [`Frontier`]: an acknowledged FIN advances the frontier, anything else that ends +/// it counts the written media as dropped on drop. Empty (no-op) off the egress side. +#[derive(Default)] +#[must_use = "an unacknowledged delivery counts as dropped on drop"] +pub(crate) struct Delivery { + frontier: Option>, + oldest: Option, + newest: Option, + bytes: u64, +} + +impl Delivery { + /// Record one frame of `bytes` at `timestamp` written to the stream. + pub(crate) fn wrote(&mut self, timestamp: Timestamp, bytes: u64) { + if self.frontier.is_none() { + return; + } + let at = media_time(timestamp); + self.oldest = Some(self.oldest.map_or(at, |oldest| oldest.min(at))); + self.newest = self.newest.max(Some(at)); + self.bytes += bytes; + } + + /// The peer acknowledged the stream's FIN: advance the frontier to its newest frame. + pub(crate) fn acked(mut self) { + let (Some(inner), Some(newest)) = (self.frontier.take(), self.newest) else { + return; + }; + let mut state = inner.state.lock().expect("stats frontier poisoned"); + if state.acked.is_none_or(|acked| newest > acked) { + state.acked = Some(newest); + state.advanced = crate::model::clock::now(); + } + } +} + +impl Drop for Delivery { + fn drop(&mut self) { + let (Some(inner), Some(oldest), Some(newest)) = (self.frontier.take(), self.oldest, self.newest) else { + return; + }; + let counters = &inner.counters.publisher; + let span = u64::try_from((newest - oldest).as_nanos()).unwrap_or(u64::MAX); + counters.dropped_duration_nanos.fetch_add(span, Ordering::Relaxed); + counters.dropped_bytes.fetch_add(self.bytes, Ordering::Relaxed); + counters.dropped_groups.fetch_add(1, Ordering::Relaxed); + } +} + #[cfg(test)] mod tests { use std::sync::{Arc, atomic::Ordering::Relaxed}; @@ -2177,3 +2622,322 @@ 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, on the frozen test clock. +#[cfg(all(test, not(loom)))] +mod lag_tests { + use futures::FutureExt; + + use super::*; + use crate::{broadcast, model::clock, 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, + } + + 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", None).unwrap(); + Self { + stats, + session, + broadcast, + track, + next: Duration::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, advancing the clock + /// by a 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 default 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; + clock::advance(TICK); + written + } + + /// Write a group toward the peer through `sub`'s frontier. + fn write(sub: &track::Subscriber, frames: &[Timestamp]) -> Delivery { + let mut delivery = sub.delivery(); + for ts in frames { + delivery.wrote(*ts, 100); + } + delivery + } + + /// One sampler tick; the egress row for the broadcast. + fn tick(&self) -> Traffic { + let mut report = Report::default(); + self.stats.report(&mut report); + 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() + } + + #[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!(buckets.windows(2).all(|w| w[0] <= w[1]), "{buckets:?}"); + assert!(buckets.last() > buckets.first(), "{buckets:?}"); + } + + #[test] + fn wall_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; its frontier standing + // still on the wall clock says otherwise. + let 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])); + clock::advance(TICK); + let now = h.tick(); + let grew = grew(&last, &now); + assert_eq!(grew.len(), 1, "{grew:?}"); + buckets.push(grew[0].0); + last = now; + } + // Behind by 0s, then 10ms, then 1s .. 5s of wall clock. + assert_eq!(buckets, vec![0, 0, 5, 6, 6, 6, 7]); + } + + #[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 { + clock::advance(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. + assert_eq!(grew(&last, &now), vec![(5, 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.delivery()); + + let after = h.tick(); + assert_eq!(after.dropped_groups, 1); + assert_eq!(after.dropped_bytes, 15 * 100); + assert_eq!(after.dropped_duration, Duration::from_millis(14 * 33)); + // 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 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` nearly three; both weigh the same three groups. + assert_eq!(grew(&last, &now), vec![(5, 3 * GROUP_BYTES), (6, 3 * 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 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_duration = Duration::from_micros(1500); + traffic.dropped_bytes = 3; + 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_duration":1.5"#), "{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_duration, Duration::from_millis(3)); + } +} diff --git a/rs/moq-stats/src/aggregate.rs b/rs/moq-stats/src/aggregate.rs index cc76e8c60d..d97c8110c6 100644 --- a/rs/moq-stats/src/aggregate.rs +++ b/rs/moq-stats/src/aggregate.rs @@ -656,6 +656,38 @@ mod tests { assert_eq!(snap.broadcasts_started, 2, "one viewer per node"); } + #[tokio::test(start_paused = true)] + async fn merges_lag_bucket_by_bucket() { + // Each node's lag histogram and dropped media sum into the merged entry, + // bucket by bucket, like every other cumulative counter. + let origin = produce_origin(); + let mut node_a = NodeBroadcast::new(&origin, "acme", "a"); + let mut node_b = NodeBroadcast::new(&origin, "acme", "b"); + let entry = |lag: &str, dropped: &str| { + let json = format!(r#"{{"bytes":1,"lag":{lag},{dropped}}}"#); + serde_json::from_str::(&json).expect("traffic") + }; + node_a.frame.insert( + "acme/room".to_string(), + entry(r#"{"50ms":10,"1s":3}"#, r#""dropped_duration":1.5,"dropped_bytes":7,"dropped_groups":1"#), + ); + node_a.traffic.update(&node_a.frame).expect("publish"); + node_b.frame.insert( + "acme/room".to_string(), + entry(r#"{"50ms":5,"inf":2}"#, r#""dropped_duration":2,"dropped_bytes":3,"dropped_groups":2"#), + ); + node_b.traffic.update(&node_b.frame).expect("publish"); + + let agg = Consumer::new(origin.consume(), Config::new().with_depth(1)); + let mut traffic = agg.traffic(&Tier::default(), Role::Publisher); + let frame = read_until_bytes(&mut traffic, "acme/room", 2).await; + let merged = frame.get("acme/room").expect("entry"); + assert_eq!(merged.lag.buckets(), &[15, 0, 0, 0, 3, 0, 0, 2]); + assert_eq!(merged.dropped_duration, Duration::from_micros(3500)); + assert_eq!(merged.dropped_bytes, 10); + assert_eq!(merged.dropped_groups, 3); + } + #[tokio::test(start_paused = true)] async fn node_drop_keeps_the_traffic_total() { // Dropping a node unannounces its broadcast, but traffic is sticky: its diff --git a/rs/moq-stats/src/lib.rs b/rs/moq-stats/src/lib.rs index 99b2411232..c4f34a1b2e 100644 --- a/rs/moq-stats/src/lib.rs +++ b/rs/moq-stats/src/lib.rs @@ -35,7 +35,7 @@ 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::{Handle, Presence, Registry, Role, Tier, Traffic}; +pub use moq_net::stats::{Handle, Histogram, Presence, Registry, Role, Tier, Traffic}; use moq_net::{AsPath, Path, PathOwned}; From 1480bc4474d169b7f1c9957af1ee4783cc3e6633 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 14:34:36 -0700 Subject: [PATCH 3/6] fix(stats): builder for IETF delivery, match live sources only Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/benches/stats.rs | 6 +++++- rs/moq-net/src/ietf/publisher.rs | 25 +++++++++++++++---------- rs/moq-net/src/lite/publisher.rs | 14 +++++++++++--- rs/moq-net/src/model/resume.rs | 19 ++++++++----------- rs/moq-net/src/stats.rs | 13 ++++++++----- rs/moq-stats/src/aggregate.rs | 10 ++++++++-- 6 files changed, 55 insertions(+), 32 deletions(-) diff --git a/rs/moq-net/benches/stats.rs b/rs/moq-net/benches/stats.rs index 7afbb61861..e7b2aa7288 100644 --- a/rs/moq-net/benches/stats.rs +++ b/rs/moq-net/benches/stats.rs @@ -117,7 +117,11 @@ fn sample(c: &mut Criterion) { let mut report = stats::Report::default(); fixture.registry.report(&mut report); let subscribed: u64 = report.traffic.iter().map(|e| e.publisher.subscriptions_started).sum(); - assert_eq!(subscribed, (broadcasts * subscriptions) as u64, "every subscription is tagged"); + assert_eq!( + subscribed, + (broadcasts * subscriptions) as u64, + "every subscription is tagged" + ); group.bench_with_input( BenchmarkId::from_parameter(format!("{broadcasts}x{subscriptions}")), diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 6246310abd..88be323a7b 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -2088,9 +2088,9 @@ impl TrackServe { self.timescale, self.version, slice, - self.track.delivery(), ) - .counted(self.opened.clone()), + .counted(self.opened.clone()) + .delivered(self.track.delivery()), ); } Poll::Ready(Ok(None)) => { @@ -2169,7 +2169,6 @@ impl GroupServe { timescale: Option, version: Version, slice: GroupSlice, - delivery: crate::stats::Delivery, ) -> Self { group.skip_to(slice.skip); group.end_at(slice.until.map_or(Bound::Unbounded, Bound::Excluded)); @@ -2183,7 +2182,7 @@ impl GroupServe { timescale, version, object_delta, - delivery, + delivery: Default::default(), state: GroupState::Open, } } @@ -2194,6 +2193,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 + } + fn poll_serve(&mut self, waiter: &kio::Waiter) -> Poll> { let mut cx = std::task::Context::from_waker(waiter.waker()); loop { @@ -2469,7 +2474,6 @@ mod group_priority_test { Some(Timescale::default()), Version::Draft14, GroupSlice::default(), - Default::default(), ); kio::wait(|waiter| serve.poll_serve(waiter)).await.unwrap(); @@ -2566,8 +2570,8 @@ mod group_priority_test { Some(Timescale::default()), Version::Draft14, GroupSlice::default(), - sub.delivery(), ) + .delivered(sub.delivery()) }; let ms = |ms| crate::Timestamp::from_millis(ms).unwrap(); @@ -2582,7 +2586,11 @@ mod group_priority_test { group.write_frame(ms(1000), b"hi".as_slice()).unwrap(); group.write_frame(ms(1250), b"yo".as_slice()).unwrap(); let mut abandoned = serve(group.consume()); - assert!(kio::wait(|waiter| abandoned.poll_serve(waiter)).now_or_never().is_none()); + assert!( + kio::wait(|waiter| abandoned.poll_serve(waiter)) + .now_or_never() + .is_none() + ); drop(abandoned); let mut report = crate::stats::Report::default(); @@ -2623,7 +2631,6 @@ mod group_priority_test { Some(Timescale::default()), Version::Draft19, GroupSlice::default(), - Default::default(), ); let mut serving = std::pin::pin!(kio::wait(|waiter| serve.poll_serve(waiter))); assert!( @@ -2672,7 +2679,6 @@ mod group_priority_test { Some(Timescale::default()), Version::Draft19, GroupSlice::default(), - Default::default(), ); let mut serving = std::pin::pin!(kio::wait(|waiter| serve.poll_serve(waiter))); // Let it run until it blocks on the rest of the frame. @@ -3625,7 +3631,6 @@ mod serve_tests { Some(Timescale::default()), Version::Draft20, slice, - Default::default(), ); kio::wait(|waiter| serve.poll_serve(waiter)).await.unwrap(); diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 3ad1ee86fb..192e105796 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -3046,7 +3046,12 @@ mod serve_group_test { let row = || { let mut report = crate::stats::Report::default(); registry.report(&mut report); - report.traffic.iter().find(|e| e.path.as_str() == "demo").unwrap().publisher + report + .traffic + .iter() + .find(|e| e.path.as_str() == "demo") + .unwrap() + .publisher }; let subscription = || Subscription { session: SinkSession::new(Log::default()), @@ -3057,7 +3062,7 @@ mod serve_group_test { track_priority_seen: 0, version: Version::Lite06, timescale: Some(crate::Timescale::default()), - opens: Default::default(), + opens: Default::default(), }; let ms = |ms| Timestamp::from_millis(ms).unwrap(); @@ -3135,7 +3140,10 @@ mod serve_group_test { group.finish().unwrap(); let handle = subscription.priority.insert(Priority::new(0, 0, 0)); - subscription.serve_group(0, 0, handle, consumer, Default::default()).await.unwrap(); + subscription + .serve_group(0, 0, handle, consumer, Default::default()) + .await + .unwrap(); assert_eq!(log.resets(), Vec::::new(), "clean completion must not reset"); diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 7fc8b53ff7..87e6a5008a 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -2523,18 +2523,15 @@ mod test { let mut producer = Producer::new(); producer.switch(&consumer_a, None).unwrap(); - let mut sub = track::Consumer::spliced("video".into(), Arc::new(broadcast::Info::default()), producer.consume()) - .with_stats(session.egress("demo")) - .subscribe(replay()) - .now_or_never() - .unwrap() - .unwrap(); + let mut sub = + track::Consumer::spliced("video".into(), Arc::new(broadcast::Info::default()), producer.consume()) + .with_stats(session.egress("demo")) + .subscribe(replay()) + .now_or_never() + .unwrap() + .unwrap(); let pending = |sub: &mut track::Subscriber| { - assert!( - kio::wait(|waiter| sub.poll_recv_group(waiter)) - .now_or_never() - .is_none() - ); + assert!(kio::wait(|waiter| sub.poll_recv_group(waiter)).now_or_never().is_none()); }; // Polling activates the segment's cursor, which names its track to the frontier. pending(&mut sub); diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index 18384b5bc2..94cda0c151 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -325,7 +325,7 @@ impl Histogram { } /// Fold another readout into this one, bucket by bucket. - pub fn add(&mut self, other: &Histogram) { + pub(crate) fn add(&mut self, other: &Histogram) { for (bucket, other) in self.buckets.iter_mut().zip(other.buckets) { *bucket += other; } @@ -1784,9 +1784,7 @@ impl Production { } fn load(cell: &AtomicU64) -> Option { - cell.load(Ordering::Relaxed) - .checked_sub(1) - .map(Duration::from_nanos) + cell.load(Ordering::Relaxed).checked_sub(1).map(Duration::from_nanos) } } @@ -1845,7 +1843,12 @@ impl Frontier { pub(crate) fn watch(&self, track: &Arc) { let Some(inner) = &self.0 else { return }; let mut state = inner.state.lock().expect("stats frontier poisoned"); - if state.sources.iter().any(|s| std::ptr::eq(s.track.as_ptr(), Arc::as_ptr(track))) { + // Only a live source can match: a dead one's address may be reused. + if state + .sources + .iter() + .any(|s| s.track.upgrade().is_some_and(|live| Arc::ptr_eq(&live, track))) + { return; } let production = track.production(); diff --git a/rs/moq-stats/src/aggregate.rs b/rs/moq-stats/src/aggregate.rs index d97c8110c6..b3fc28fe64 100644 --- a/rs/moq-stats/src/aggregate.rs +++ b/rs/moq-stats/src/aggregate.rs @@ -669,12 +669,18 @@ mod tests { }; node_a.frame.insert( "acme/room".to_string(), - entry(r#"{"50ms":10,"1s":3}"#, r#""dropped_duration":1.5,"dropped_bytes":7,"dropped_groups":1"#), + entry( + r#"{"50ms":10,"1s":3}"#, + r#""dropped_duration":1.5,"dropped_bytes":7,"dropped_groups":1"#, + ), ); node_a.traffic.update(&node_a.frame).expect("publish"); node_b.frame.insert( "acme/room".to_string(), - entry(r#"{"50ms":5,"inf":2}"#, r#""dropped_duration":2,"dropped_bytes":3,"dropped_groups":2"#), + entry( + r#"{"50ms":5,"inf":2}"#, + r#""dropped_duration":2,"dropped_bytes":3,"dropped_groups":2"#, + ), ); node_b.traffic.update(&node_b.frame).expect("publish"); From 0d7533189b52038c927bf675625964c50560a312 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 15:02:12 -0700 Subject: [PATCH 4/6] refactor(stats): nest dropped media as Traffic::dropped Co-Authored-By: Claude Opus 5.5 --- doc/bin/relay/config.md | 2 +- doc/concept/stats.md | 6 +- quest/m1/qos/starvation-frames.md | 6 +- rs/moq-net/src/ietf/publisher.rs | 6 +- rs/moq-net/src/lite/publisher.rs | 8 +-- rs/moq-net/src/stats.rs | 107 +++++++++++++++++------------- rs/moq-stats/src/aggregate.rs | 10 +-- rs/moq-stats/src/lib.rs | 2 +- 8 files changed, 81 insertions(+), 66 deletions(-) diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index f3f3b3013d..dc53a950fb 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -215,7 +215,7 @@ read them with the [`moq-stats`](https://docs.rs/moq-stats) crate. `interval` the relay samples each subscription's lag, the newest frame produced minus the newest frame the viewer acknowledged, into a byte-weighted `lag` histogram with buckets at 50 ms, 100 ms, 250 ms, 500 ms, 1 s, 2 s, 5 s, and above. -`dropped_duration`, `dropped_bytes`, and `dropped_groups` count media written to a +`dropped` counts the duration, bytes, and groups of media written to a viewer on a group stream that ended before the viewer acknowledged it. Both resolve one group at a time, so they overstate lag and drops for long groups; see [Lag](/concept/stats#lag). diff --git a/doc/concept/stats.md b/doc/concept/stats.md index 5968aba1d0..a41814a2c2 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -85,7 +85,7 @@ return later as a new broadcast. "bytes": 1048576, "frames": 900, "groups": 30, "datagrams": 0, "stale": { "bytes": 0, "frames": 0, "groups": 0, "datagrams": 0 }, "lag": { "50ms": 1040000, "1s": 8576 }, - "dropped_duration": 966.5, "dropped_bytes": 8576, "dropped_groups": 1, + "dropped": { "duration": 966.5, "bytes": 8576, "groups": 1 }, "announced": 1, "announced_closed": 0, "broadcasts": 3, "broadcasts_closed": 1, "subscriptions": 6, "subscriptions_closed": 2 @@ -104,7 +104,7 @@ return later as a new broadcast. | `datagrams` | Groups delivered as an unreliable datagram. A subset of `groups`. | | `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` | `publisher.json` only: how far behind the viewers' acknowledged media is, as bytes per bucket of media time. See [Lag](#lag). Omitted while empty. | -| `dropped_duration` / `dropped_bytes` / `dropped_groups` | `publisher.json` only: media written to a viewer on a group stream that was reset or abandoned before the viewer acknowledged it. The duration is fractional milliseconds. Omitted while zero. | +| `dropped` | `publisher.json` only: media written to a viewer on a group stream that was reset or abandoned before the viewer acknowledged it, as `duration` (fractional milliseconds), `bytes`, and `groups`. Omitted while empty. | The last six fields in the example are legacy spellings of the `*_started` and `*_ended` counters, still written so an older consumer reads a newer relay. A reader @@ -155,7 +155,7 @@ ends on the old one and starts on the new. ### Counters Every counter is cumulative and monotonic: an unsigned integer, apart from -`dropped_duration` in fractional milliseconds. A rate is the +`dropped.duration` in fractional milliseconds. A rate is the difference between two frames divided by the time between them, and a live count is started minus ended. A frame never shows ended above started. diff --git a/quest/m1/qos/starvation-frames.md b/quest/m1/qos/starvation-frames.md index 874c69528d..76a5266fef 100644 --- a/quest/m1/qos/starvation-frames.md +++ b/quest/m1/qos/starvation-frames.md @@ -3,7 +3,7 @@ ## Goal Each subscription's acknowledged frontier moves at every frame boundary -instead of once per group, so the `lag` histogram and `dropped_*` counters +instead of once per group, so the `lag` histogram and `dropped` counters on `stats::Traffic` read a frontier at most one frame stale, and a second histogram reports per-frame delivery delay so viewer jitter is visible per broadcast. The wire shape is unchanged apart from the new histogram. @@ -28,7 +28,7 @@ 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 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 +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. @@ -46,7 +46,7 @@ a browser transport; document which resolution a node offers. 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 under an injected ACK delay staying inside one bucket of the +unacknowledged span to `dropped.duration`; delivery delay under an injected ACK delay staying inside one bucket of the true RTT; and fallback to group-granularity frontier tracking when the hook is unsupported. diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 88be323a7b..1e4d56bc46 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -2596,9 +2596,9 @@ mod group_priority_test { let mut report = crate::stats::Report::default(); registry.report(&mut report); let row = report.traffic[0].publisher; - assert_eq!(row.dropped_groups, 1); - assert_eq!(row.dropped_bytes, 4); - assert_eq!(row.dropped_duration, Duration::from_millis(250)); + assert_eq!(row.dropped.groups, 1); + assert_eq!(row.dropped.bytes, 4); + assert_eq!(row.dropped.duration, Duration::from_millis(250)); // 1250ms produced against a 500ms frontier: the [500ms, 1s) bucket. assert_eq!(row.lag.buckets(), &[0, 0, 0, 0, 6, 0, 0, 0]); } diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 192e105796..d2f08ff9ee 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -3099,10 +3099,10 @@ mod serve_group_test { drop(serve); let after = row(); - assert_eq!(before.dropped_groups, 0); - assert_eq!(after.dropped_groups, 2); - assert_eq!(after.dropped_bytes, 5 + 4); - assert_eq!(after.dropped_duration, Duration::from_millis(100)); + assert_eq!(before.dropped.groups, 0); + assert_eq!(after.dropped.groups, 2); + assert_eq!(after.dropped.bytes, 5 + 4); + assert_eq!(after.dropped.duration, Duration::from_millis(100)); // The newest frame is at 2100ms and the frontier at 500ms: 1.6s behind, for // the 9 bytes produced since the last sample. assert_eq!(before.lag.buckets(), &[2, 0, 0, 0, 0, 0, 0, 0]); diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index 94cda0c151..77a5c3db1a 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -68,8 +68,7 @@ //! frontier, wall time the frontier has stood still while behind)` weighted //! by the bytes the track produced since the previous call. A paused source //! produces nothing, so it adds no weight. -//! * `dropped_duration` / `dropped_bytes` / `dropped_groups` (egress only): -//! media written toward a peer on a group stream that was reset or abandoned +//! * `dropped` (egress only, a [`Dropped`]): media written toward a peer on a group stream that was reset or abandoned //! before its FIN was acknowledged. The duration is the span between the //! group's oldest and newest written frame timestamps; nothing inside a group //! is known to be acknowledged, so the whole written span counts. @@ -235,9 +234,11 @@ impl Counters { let datagrams = self.datagrams.load(Ordering::Relaxed); let stale = self.stale.snapshot(); let lag = self.lag.snapshot(); - let dropped_duration = Duration::from_nanos(self.dropped_duration_nanos.load(Ordering::Relaxed)); - let dropped_bytes = self.dropped_bytes.load(Ordering::Relaxed); - let dropped_groups = self.dropped_groups.load(Ordering::Relaxed); + let dropped = Dropped { + duration: Duration::from_nanos(self.dropped_duration_nanos.load(Ordering::Relaxed)), + bytes: self.dropped_bytes.load(Ordering::Relaxed), + groups: self.dropped_groups.load(Ordering::Relaxed), + }; Traffic { announces_started, announces_ended, @@ -253,9 +254,7 @@ impl Counters { datagrams, stale, lag, - dropped_duration, - dropped_bytes, - dropped_groups, + dropped, } } @@ -431,6 +430,40 @@ impl Content { } } +/// Media written toward a peer on group streams that were reset or abandoned +/// before the peer acknowledged them. The nested shape of [`Traffic::dropped`]. +/// +/// Group granularity counts each group's whole written span, so this over-reports +/// media that did arrive. +#[serde_as] +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default)] +#[non_exhaustive] +pub struct Dropped { + /// Cumulative media time: each group's oldest to newest written frame. + /// Fractional milliseconds on the wire. + #[serde_as(as = "DurationMilliSecondsWithFrac")] + pub duration: Duration, + /// Cumulative bytes written on those streams. + pub bytes: u64, + /// Cumulative streams abandoned after writing at least one frame. + pub groups: u64, +} + +impl Dropped { + /// True when nothing has been dropped. + pub fn is_empty(&self) -> bool { + self.duration.is_zero() && self.bytes == 0 && self.groups == 0 + } + + /// Fold another readout into this one, counter by counter. + pub(crate) fn add(&mut self, other: Self) { + self.duration += other.duration; + self.bytes += other.bytes; + self.groups += other.groups; + } +} + /// Per-(tier, root) session gauge. One of these is shared (via `Arc`) by every /// [`Session`] guard for the same auth root on the same tier: `sessions_started` /// bumps on connect, `sessions_ended` on disconnect. @@ -509,13 +542,8 @@ pub struct Traffic { /// [`Registry::report`]; see the module docs. pub lag: Histogram, /// Egress only: media written on group streams that were reset or abandoned - /// before the peer acknowledged them. Group granularity counts each group's - /// whole written span, so this over-reports media that did arrive. - pub dropped_duration: Duration, - /// Egress only: bytes written on those abandoned group streams. - pub dropped_bytes: u64, - /// Egress only: group streams abandoned after writing at least one frame. - pub dropped_groups: u64, + /// before the peer acknowledged them. + pub dropped: Dropped, } /// One spelling of a counter edge on the wire: absent, or a present integer. @@ -537,7 +565,6 @@ fn counter_edge(canonical: Edge, legacy: Edge) -> u64 { canonical.0.or(legacy.0).unwrap_or(0) } -#[serde_as] #[derive(Serialize)] struct TrafficSer { announces_started: u64, @@ -561,13 +588,8 @@ struct TrafficSer { stale: Content, #[serde(skip_serializing_if = "Histogram::is_empty")] lag: Histogram, - #[serde_as(as = "DurationMilliSecondsWithFrac")] - #[serde(skip_serializing_if = "Duration::is_zero")] - dropped_duration: Duration, - #[serde(skip_serializing_if = "is_zero")] - dropped_bytes: u64, - #[serde(skip_serializing_if = "is_zero")] - dropped_groups: u64, + #[serde(skip_serializing_if = "Dropped::is_empty")] + dropped: Dropped, } impl From for TrafficSer { @@ -593,14 +615,11 @@ impl From for TrafficSer { datagrams: t.datagrams, stale: t.stale, lag: t.lag, - dropped_duration: t.dropped_duration, - dropped_bytes: t.dropped_bytes, - dropped_groups: t.dropped_groups, + dropped: t.dropped, } } } -#[serde_as] #[derive(Default, Deserialize)] #[serde(default)] struct TrafficDe { @@ -624,10 +643,7 @@ struct TrafficDe { datagrams: u64, stale: Content, lag: Histogram, - #[serde_as(as = "DurationMilliSecondsWithFrac")] - dropped_duration: Duration, - dropped_bytes: u64, - dropped_groups: u64, + dropped: Dropped, } impl From for Traffic { @@ -647,9 +663,7 @@ impl From for Traffic { datagrams: d.datagrams, stale: d.stale, lag: d.lag, - dropped_duration: d.dropped_duration, - dropped_bytes: d.dropped_bytes, - dropped_groups: d.dropped_groups, + dropped: d.dropped, } } } @@ -683,9 +697,7 @@ impl Traffic { self.datagrams += other.datagrams; self.stale.add(other.stale); self.lag.add(&other.lag); - self.dropped_duration += other.dropped_duration; - self.dropped_bytes += other.dropped_bytes; - self.dropped_groups += other.dropped_groups; + self.dropped.add(other.dropped); } /// True while the broadcast is announced (an announce guard is open). @@ -2869,7 +2881,7 @@ mod lag_tests { assert_eq!(grew(&last, &now), vec![(0, GROUP_BYTES)]); last = now; } - assert_eq!(last.dropped_groups, 0); + assert_eq!(last.dropped.groups, 0); } #[test] @@ -2997,9 +3009,9 @@ mod lag_tests { drop(sub.delivery()); let after = h.tick(); - assert_eq!(after.dropped_groups, 1); - assert_eq!(after.dropped_bytes, 15 * 100); - assert_eq!(after.dropped_duration, Duration::from_millis(14 * 33)); + assert_eq!(after.dropped.groups, 1); + assert_eq!(after.dropped.bytes, 15 * 100); + assert_eq!(after.dropped.duration, Duration::from_millis(14 * 33)); // The drop itself adds nothing to the histogram beyond the tick's own sample. assert_eq!(after.lag.total() - before.lag.total(), GROUP_BYTES); } @@ -3042,17 +3054,20 @@ mod lag_tests { let mut traffic = Traffic::default(); traffic.lag.buckets[0] = 7; traffic.lag.buckets[7] = 9; - traffic.dropped_duration = Duration::from_micros(1500); - traffic.dropped_bytes = 3; - traffic.dropped_groups = 1; + traffic.dropped.duration = Duration::from_micros(1500); + traffic.dropped.bytes = 3; + 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_duration":1.5"#), "{json}"); + assert!( + json.contains(r#""dropped":{"duration":1.5,"bytes":3,"groups":1}"#), + "{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_duration, Duration::from_millis(3)); + assert_eq!(sum.dropped.duration, Duration::from_millis(3)); } } diff --git a/rs/moq-stats/src/aggregate.rs b/rs/moq-stats/src/aggregate.rs index b3fc28fe64..11dc744914 100644 --- a/rs/moq-stats/src/aggregate.rs +++ b/rs/moq-stats/src/aggregate.rs @@ -671,7 +671,7 @@ mod tests { "acme/room".to_string(), entry( r#"{"50ms":10,"1s":3}"#, - r#""dropped_duration":1.5,"dropped_bytes":7,"dropped_groups":1"#, + r#""dropped":{"duration":1.5,"bytes":7,"groups":1}"#, ), ); node_a.traffic.update(&node_a.frame).expect("publish"); @@ -679,7 +679,7 @@ mod tests { "acme/room".to_string(), entry( r#"{"50ms":5,"inf":2}"#, - r#""dropped_duration":2,"dropped_bytes":3,"dropped_groups":2"#, + r#""dropped":{"duration":2,"bytes":3,"groups":2}"#, ), ); node_b.traffic.update(&node_b.frame).expect("publish"); @@ -689,9 +689,9 @@ mod tests { let frame = read_until_bytes(&mut traffic, "acme/room", 2).await; let merged = frame.get("acme/room").expect("entry"); assert_eq!(merged.lag.buckets(), &[15, 0, 0, 0, 3, 0, 0, 2]); - assert_eq!(merged.dropped_duration, Duration::from_micros(3500)); - assert_eq!(merged.dropped_bytes, 10); - assert_eq!(merged.dropped_groups, 3); + assert_eq!(merged.dropped.duration, Duration::from_micros(3500)); + assert_eq!(merged.dropped.bytes, 10); + assert_eq!(merged.dropped.groups, 3); } #[tokio::test(start_paused = true)] diff --git a/rs/moq-stats/src/lib.rs b/rs/moq-stats/src/lib.rs index c4f34a1b2e..a9ad409e4a 100644 --- a/rs/moq-stats/src/lib.rs +++ b/rs/moq-stats/src/lib.rs @@ -35,7 +35,7 @@ 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::{Handle, Histogram, Presence, Registry, Role, Tier, Traffic}; +pub use moq_net::stats::{Dropped, Handle, Histogram, Presence, Registry, Role, Tier, Traffic}; use moq_net::{AsPath, Path, PathOwned}; From c4eaffea474c8e80ee70c789588993b7a83d67ff Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 18:22:31 -0700 Subject: [PATCH 5/6] fix(stats): stop weighing a spliced segment's track once a switch caps it A capped route can keep publishing past the boundary; its production never reaches this reader, so it no longer counts toward lag weight or the newest edge. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/resume.rs | 32 +++++++++++++++++++++++++++++--- rs/moq-net/src/model/track.rs | 11 +++++++++++ rs/moq-net/src/stats.rs | 10 ++++++++++ 3 files changed, 50 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 87e6a5008a..9edefa9dbf 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -1520,6 +1520,17 @@ impl Subscriber { } } } + + /// Stop reporting any segment's track to `frontier`, current or future. + pub(crate) fn unwatch(&mut self, frontier: &crate::stats::Frontier) { + self.frontier = Default::default(); + for seg in &mut self.segments { + seg.frontier = Default::default(); + if let SubState::Active(sub) = &mut seg.sub { + sub.unwatch(frontier); + } + } + } } impl Subscriber { @@ -1656,6 +1667,13 @@ impl Subscriber { warm.next = next; } if existing.end != segment.end { + // A capped segment's track may live on past the boundary; what it + // produces there never reaches this reader, so it stops weighing lag. + if existing.end.is_none() + && let SubState::Active(sub) = &mut existing.sub + { + sub.unwatch(&existing.frontier); + } // The boundary bounds the drift anchor as well as the demand; the // anchor follows in `refresh_anchor`. existing.end = segment.end; @@ -1870,7 +1888,10 @@ impl Subscriber { sub.raise_start_to(seg.first_group().max(min_sequence)); sub.set_anchor(seg.anchor.clone()); let _ = sub.update(slice(prefs, seg.ask, seg.end)); - sub.watch(&seg.frontier); + // Only the open-ended segment weighs lag; see `apply`. + if seg.end.is_none() { + sub.watch(&seg.frontier); + } seg.sub = SubState::Active(Box::new(sub)); } // The underlying track was rejected or closed: stall, not error. @@ -2505,8 +2526,9 @@ mod test { recv_pending(&mut sub); } - /// A spliced subscription's lag weighs what every segment's track produces, so a - /// route-fed broadcast (every relay hop) is sampled across failovers. + /// A spliced subscription's lag weighs what the open-ended segment's track + /// produces, so a route-fed broadcast (every relay hop) is sampled across + /// failovers, and a capped route that lives on stops counting. #[tokio::test] async fn lag_weighs_every_segment() { use crate::stats; @@ -2543,6 +2565,10 @@ mod test { pending(&mut sub); write_group_at(&mut track_b, 1, "b1", Duration::from_secs(1)); assert_eq!(weight(), 4); + + // The capped route keeps publishing past the boundary: not this reader's media. + write_group_at(&mut track_a, 2, "a2", Duration::from_secs(2)); + assert_eq!(weight(), 4); } #[tokio::test] diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 158e2fd08d..196a236985 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -4077,6 +4077,17 @@ impl Subscriber { } } + /// Stop reporting the tracks feeding this cursor to `frontier`. + pub(crate) fn unwatch(&mut self, frontier: &stats::Frontier) { + match &mut self.inner { + SubscriberKind::Plain(plain) => { + let cache = plain.state.read().cache.clone(); + frontier.unwatch(&cache); + } + SubscriberKind::Spliced(spliced) => spliced.unwatch(frontier), + } + } + /// Return the latest sequence number in the track. pub fn latest(&self) -> Option { match &self.inner { diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index 77a5c3db1a..5c8a5f30a4 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -1873,6 +1873,16 @@ impl Frontier { }); } + /// Remove a track that no longer feeds this subscription, such as a splice + /// segment capped by a switch, so what it produces past the cap stops counting. + pub(crate) fn unwatch(&self, track: &Arc) { + let Some(inner) = &self.0 else { return }; + let mut state = inner.state.lock().expect("stats frontier poisoned"); + state + .sources + .retain(|s| s.track.upgrade().is_some_and(|live| !Arc::ptr_eq(&live, track))); + } + /// Start tracking one group stream written toward the peer. pub(crate) fn delivery(&self) -> Delivery { Delivery { From bee6fdfbe73e21fa54ce3089c4e3be24c5b25a56 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 19:25:43 -0700 Subject: [PATCH 6/6] fix(stats): keep an unwatched source's pending bytes, order production edges Unwatching a capped splice segment carried away the bytes it produced since the last sample; carry them into the next one. Publish production bytes with release and read them with acquire so the sampler never weighs bytes against edges that predate them. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/resume.rs | 14 ++++++++------ rs/moq-net/src/stats.rs | 29 +++++++++++++++++++++++------ 2 files changed, 31 insertions(+), 12 deletions(-) diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 9edefa9dbf..83dfe74369 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -2560,15 +2560,17 @@ mod test { write_group_at(&mut track_a, 0, "a0", Duration::ZERO); assert_eq!(weight(), 2); - producer.switch(&consumer_b, Position::group(1)).unwrap(); - let _ = kio::wait(|waiter| sub.poll_recv_group(waiter)).now_or_never(); + // Produced before the switch but not sampled yet: still this reader's media. + write_group_at(&mut track_a, 1, "a1", Duration::from_millis(500)); + producer.switch(&consumer_b, Position::group(2)).unwrap(); + while kio::wait(|waiter| sub.poll_recv_group(waiter)).now_or_never().is_some() {} pending(&mut sub); - write_group_at(&mut track_b, 1, "b1", Duration::from_secs(1)); - assert_eq!(weight(), 4); + write_group_at(&mut track_b, 2, "b2", Duration::from_secs(1)); + assert_eq!(weight(), 6); // The capped route keeps publishing past the boundary: not this reader's media. - write_group_at(&mut track_a, 2, "a2", Duration::from_secs(2)); - assert_eq!(weight(), 4); + write_group_at(&mut track_a, 3, "a3", Duration::from_secs(2)); + assert_eq!(weight(), 6); } #[tokio::test] diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index 5c8a5f30a4..e10f87832b 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -1792,7 +1792,8 @@ impl Production { } // A max, not a store: B-frames present out of order within a group. self.newest.fetch_max(at, Ordering::Relaxed); - self.bytes.fetch_add(bytes, 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 { @@ -1822,6 +1823,8 @@ struct FrontierState { behind: Option, /// The tracks feeding the subscription: one, or one per segment of a splice. sources: Vec, + /// Bytes an unwatched source produced since the last sample, still owed to it. + unsampled: u64, } /// A track feeding a [`Frontier`], with the produced bytes already sampled. @@ -1839,6 +1842,7 @@ impl Frontier { advanced: crate::model::clock::now(), behind: None, sources: Vec::new(), + unsampled: 0, }), }); counters @@ -1878,9 +1882,21 @@ impl Frontier { pub(crate) fn unwatch(&self, track: &Arc) { let Some(inner) = &self.0 else { return }; let mut state = inner.state.lock().expect("stats frontier poisoned"); - state - .sources - .retain(|s| s.track.upgrade().is_some_and(|live| !Arc::ptr_eq(&live, track))); + let mut unsampled = 0; + state.sources.retain(|s| { + let Some(live) = s.track.upgrade() else { return false }; + if !Arc::ptr_eq(&live, track) { + return true; + } + // What it produced before the cap reached this reader: keep it for the next sample. + unsampled += live + .production() + .bytes + .load(Ordering::Relaxed) + .saturating_sub(s.sampled); + false + }); + state.unsampled += unsampled; } /// Start tracking one group stream written toward the peer. @@ -1897,7 +1913,7 @@ impl FrontierInner { /// nothing was produced (a paused source moves nothing). fn sample(&self, now: Instant) -> Option<(Duration, u64)> { let mut state = self.state.lock().expect("stats frontier poisoned"); - let mut weight = 0; + let mut weight = std::mem::take(&mut state.unsampled); let mut newest = None; let mut first: Option = None; state.sources.retain_mut(|source| { @@ -1906,7 +1922,8 @@ impl FrontierInner { return false; }; let production = track.production(); - let bytes = production.bytes.load(Ordering::Relaxed); + // Acquire pairs with `Production::record`: the edges read below cover these bytes. + let bytes = production.bytes.load(Ordering::Acquire); weight += bytes.saturating_sub(source.sampled); source.sampled = bytes; newest = newest.max(Production::load(&production.newest));