From 94a566f971fe592fa16899d645bdc242c9c1738c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 28 Sep 2026 19:36:48 -0700 Subject: [PATCH 1/4] chore(quest): claim final-lag-sample Co-Authored-By: Claude Sonnet 5.5 From cff4c75e3d7f2cabbe3311db20832629ca5267cc Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 28 Sep 2026 19:44:52 -0700 Subject: [PATCH 2/4] fix(net): sample a closing subscription's lag once more Co-Authored-By: Claude Sonnet 5.5 --- doc/concept/stats.md | 5 ++- quest/m1/qos/README.md | 2 - quest/m1/qos/final-lag-sample.md | 34 -------------- rs/moq-net/src/stats.rs | 76 ++++++++++++++++++++++++++++++++ 4 files changed, 80 insertions(+), 37 deletions(-) delete mode 100644 quest/m1/qos/final-lag-sample.md diff --git a/doc/concept/stats.md b/doc/concept/stats.md index 2ca7da9015..af17282898 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -122,7 +122,10 @@ 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. +falls in, and the broadcast's subscriptions sum into one entry. A subscription +that ends between two intervals takes one last sample when it closes, so the +bytes since its previous sample are not lost, even if it opened and closed +between two ticks. | Bucket | Lag | | --- | --- | diff --git a/quest/m1/qos/README.md b/quest/m1/qos/README.md index 2028ac97d0..e7ec0b899a 100644 --- a/quest/m1/qos/README.md +++ b/quest/m1/qos/README.md @@ -39,8 +39,6 @@ on `dev`. ## Required -- [Final lag sample](/quest/m1/qos/final-lag-sample.md) - a closing - subscription records its last partial interval instead of losing it - [Lag dashboard](/quest/m1/qos/lag-dashboard.md) - the demo stats dashboard shows viewer lag percentiles and dropped media - [Starvation at frame granularity](/quest/m1/qos/starvation-frames.md) - the diff --git a/quest/m1/qos/final-lag-sample.md b/quest/m1/qos/final-lag-sample.md deleted file mode 100644 index 94086997d9..0000000000 --- a/quest/m1/qos/final-lag-sample.md +++ /dev/null @@ -1,34 +0,0 @@ -# [XS] A closing subscription takes its last lag sample - -## Goal - -When an egress subscription ends, the bytes its track produced since the last -stats tick still land in the `lag` histogram at its final lag, instead of -being lost. A subscription that opens and closes between two ticks is sampled -at least once. - -## Plan - -Lag is sampled only on `Registry::report` ticks: `Counters::sample` in -`rs/moq-net/src/stats.rs` walks weak `FrontierInner` references and prunes -the dead ones without sampling them, so a subscription's last partial -interval disappears with it. - -- The subscription guard and each in-flight group `Delivery` hold a strong - reference, so the frontier dies only once the last of them does. Its drop is - the natural place for one final `sample(now)` into the same counters. -- No double counting: `sample` advances each source's `sampled` bytes under the - frontier lock, so a tick racing the drop leaves nothing for it to count again. -- Test: a subscription that opens and closes between ticks while its track - produces lands those bytes in the histogram; one that closes mid-interval - adds exactly the bytes since its last sample. Note the behaviour in - `doc/concept/stats.md`. - -Public API: none. Wire: none, only the histogram's values change. - -Overlaps the lag-splice quest on the line branch -([#4381](https://github.com/moq-dev/moq/pull/4381)), which also changes how -`FrontierInner::sample` holds `unsampled` weight and when a spliced segment's -source stops being sampled. Land them in sequence on the same sampler, not in -parallel, and make the final drop-time sample keep any weight lag-splice -defers. diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index e10f87832b..eb203bdb9f 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -1954,6 +1954,17 @@ impl FrontierInner { } } +impl Drop for FrontierInner { + /// The last strong reference (the subscription guard or an in-flight [`Delivery`]) + /// is gone, so no tick will find this frontier again: record the bytes produced + /// since the last sample now, at the lag they last had. + fn drop(&mut self) { + if let Some((lag, weight)) = self.sample(crate::model::clock::now()) { + self.counters.publisher.lag.record(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. @@ -3062,6 +3073,71 @@ mod lag_tests { assert_eq!(grew(&last, &now), vec![(5, 3 * GROUP_BYTES), (6, 3 * GROUP_BYTES)]); } + /// The egress lag bytes recorded so far, without ticking the sampler. + fn recorded(h: &Harness) -> u64 { + h.stats + .snapshot() + .traffic() + .into_iter() + .find(|(_, role, _)| matches!(role, Role::Publisher)) + .map_or(0, |(_, _, traffic)| traffic.lag.total()) + } + + #[test] + fn a_subscription_closed_between_ticks_is_sampled_once() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let last = h.tick(); + h.produce(); + drop(sub); + + // The tick that follows finds no frontier, yet the group still lands, at the + // 957ms it was behind when the guard dropped. + let now = h.tick(); + assert_eq!(grew(&last, &now), vec![(4, GROUP_BYTES)]); + + // Sampled exactly once: the closing tick pruned the row, and later media adds + // nothing to the retired totals. + h.produce(); + h.stats.report(&mut Report::default()); + assert_eq!(recorded(&h), GROUP_BYTES); + } + + #[test] + fn a_subscription_closed_mid_interval_adds_only_the_bytes_since_its_last_sample() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let mut last = h.tick(); + h.produce(); + last = { + let now = h.tick(); + assert_eq!(grew(&last, &now), vec![(4, GROUP_BYTES)]); + now + }; + h.produce(); + h.produce(); + drop(sub); + + let now = h.tick(); + assert_eq!( + grew(&last, &now).iter().map(|(_, bytes)| bytes).sum::(), + 2 * GROUP_BYTES + ); + } + + #[test] + fn an_in_flight_delivery_defers_the_final_sample_to_its_own_drop() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let delivery = Harness::write(&sub, &h.produce()); + drop(sub); + // The delivery still holds the frontier, so it has not closed yet. + assert_eq!(recorded(&h), 0); + + drop(delivery); + assert_eq!(recorded(&h), GROUP_BYTES); + } + #[test] fn a_closed_subscription_leaves_the_sampler() { let mut h = Harness::new(); From 5b98e2bdb9d1622a575f268913909129df98155a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 29 Sep 2026 10:49:17 -0700 Subject: [PATCH 3/4] fix(net): read a dead track's last bytes in the final lag sample The frontier held each source only by a Weak to its track, so when the track died first (the subscriber's state dropping before its stats guard, or a group serve dropping its group before its Delivery), the final sample failed the upgrade and discarded the bytes since the last tick. Sources now hold the track's Production strongly and use the Weak only for liveness. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/cache.rs | 7 ++-- rs/moq-net/src/stats.rs | 73 +++++++++++++++++++++++------------ 2 files changed, 52 insertions(+), 28 deletions(-) diff --git a/rs/moq-net/src/model/cache.rs b/rs/moq-net/src/model/cache.rs index e5ea169a7b..c16c1579ee 100644 --- a/rs/moq-net/src/model/cache.rs +++ b/rs/moq-net/src/model/cache.rs @@ -493,8 +493,9 @@ pub(crate) struct Track { 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, + // account because it is already the one per-track handle every group holds. Shared + // so a sampler can still read the last bytes after the track is gone. + production: Arc, } impl Track { @@ -521,7 +522,7 @@ impl Track { } /// What this track's groups have produced. - pub(crate) fn production(&self) -> &crate::stats::Production { + pub(crate) fn production(&self) -> &Arc { &self.production } diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index eb203bdb9f..ae4aeed840 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -1829,7 +1829,11 @@ struct FrontierState { /// A track feeding a [`Frontier`], with the produced bytes already sampled. struct Source { + // Only liveness: once dead, the track can never produce again. track: Weak, + // Strong, so the bytes a track produced before it died are still read, however + // its last handles and this frontier's last reference race to drop. + production: Arc, sampled: u64, } @@ -1859,20 +1863,16 @@ 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"); - // 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))) - { + let production = track.production(); + if state.sources.iter().any(|s| Arc::ptr_eq(&s.production, production)) { return; } - let production = track.production(); if state.acked.is_none() { state.acked = Production::load(&production.newest); } state.sources.push(Source { track: Arc::downgrade(track), + production: production.clone(), sampled: production.bytes.load(Ordering::Relaxed), }); } @@ -1882,18 +1882,14 @@ 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"); + let production = track.production(); let mut unsampled = 0; state.sources.retain(|s| { - let Some(live) = s.track.upgrade() else { return false }; - if !Arc::ptr_eq(&live, track) { + if !Arc::ptr_eq(&s.production, production) { 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); + unsampled += production.bytes.load(Ordering::Relaxed).saturating_sub(s.sampled); false }); state.unsampled += unsampled; @@ -1917,11 +1913,10 @@ impl FrontierInner { 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(); + // Checked before reading: every writer holds the track, so a dead one has + // already recorded everything it ever will, and this read is its last. + let live = source.track.strong_count() > 0; + let production = &source.production; // 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); @@ -1930,7 +1925,7 @@ impl FrontierInner { if let Some(at) = Production::load(&production.first) { first = Some(first.map_or(at, |first| first.min(at))); } - true + live }); let newest = newest?; // Opened before the track produced anything: its first frame is where the @@ -3074,8 +3069,8 @@ mod lag_tests { } /// The egress lag bytes recorded so far, without ticking the sampler. - fn recorded(h: &Harness) -> u64 { - h.stats + fn recorded(stats: &Registry) -> u64 { + stats .snapshot() .traffic() .into_iter() @@ -3100,7 +3095,7 @@ mod lag_tests { // nothing to the retired totals. h.produce(); h.stats.report(&mut Report::default()); - assert_eq!(recorded(&h), GROUP_BYTES); + assert_eq!(recorded(&h.stats), GROUP_BYTES); } #[test] @@ -3132,10 +3127,38 @@ mod lag_tests { let delivery = Harness::write(&sub, &h.produce()); drop(sub); // The delivery still holds the frontier, so it has not closed yet. - assert_eq!(recorded(&h), 0); + assert_eq!(recorded(&h.stats), 0); + + drop(delivery); + assert_eq!(recorded(&h.stats), GROUP_BYTES); + } + + #[test] + fn the_last_subscriber_samples_a_track_its_own_drop_releases() { + let mut h = Harness::new(); + let sub = h.subscribe(); + h.produce(); + // The producer is gone, so the subscriber holds the last reference to the track. + drop(h.track); + drop(h.broadcast); + + drop(sub); + assert_eq!(recorded(&h.stats), GROUP_BYTES); + } + + #[test] + fn an_in_flight_delivery_samples_a_track_that_already_closed() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let delivery = Harness::write(&sub, &h.produce()); + drop(h.track); + drop(h.broadcast); + drop(sub); + assert_eq!(recorded(&h.stats), 0); + // The track died with the subscriber; the delivery still owes its bytes. drop(delivery); - assert_eq!(recorded(&h), GROUP_BYTES); + assert_eq!(recorded(&h.stats), GROUP_BYTES); } #[test] From 13facea2aaab8befc62ed87be3cc5c7617cd7ccd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 29 Sep 2026 11:53:37 -0700 Subject: [PATCH 4/4] fix(net): decide a stats entry is retired before reading it out Registry::report read every entry out, then pruned those only the map still held. A guard whose last bump (such as a frontier's final lag sample) landed between the two was retired without that bump reaching any per-broadcast report. Each tier now decides it is done before its readout, so a racing bump lands in this readout or the next. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/stats.rs | 51 +++++++++++++++++++++-------------------- 1 file changed, 26 insertions(+), 25 deletions(-) diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index ae4aeed840..10fcbbf977 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -1208,37 +1208,38 @@ impl Registry { let mut retired = shared.retired.lock(); { 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() { + entries.retain(|path, entry| { + // With only the map's Arc left, and the map locked, no guard can reach + // the entry again. + let orphan = Arc::strong_count(entry) == 1; + let mut tiers = entry.tiers.lock().expect("stats tiers poisoned"); + tiers.retain(|tier, counters| { + // Decided before the readout, so a last bump racing this report (such + // as a frontier's final lag sample) lands in this readout or the next, + // never after the one that retires the counters. + let done = orphan && Arc::strong_count(counters) == 1; + if done { + // Pairs with the last holder's releasing decrement, so the readout + // below sees every bump it made. + std::sync::atomic::fence(Ordering::Acquire); + } counters.publisher.sample(now); + let publisher = counters.publisher.snapshot(); + let subscriber = counters.subscriber.snapshot(); + if done { + let totals = retired.traffic.entry(tier.clone()).or_default(); + totals[Role::Publisher.idx()].add(publisher); + totals[Role::Subscriber.idx()].add(subscriber); + } report.traffic.push(TrafficEntry { path: path.clone(), tier: tier.clone(), - publisher: counters.publisher.snapshot(), - subscriber: counters.subscriber.snapshot(), + publisher, + subscriber, }); - } - } - // Prune entries no guard holds anymore: with only the map's Arc - // left, no future bump can land, so the entry is done. (A guard - // created after the readout above still holds the Arc and keeps - // its entry alive.) - entries.retain(|_, entry| { - if Arc::strong_count(entry) > 1 { - return true; - } - let mut tiers = entry.tiers.lock().expect("stats tiers poisoned"); - tiers.retain(|tier, counters| { - if Arc::strong_count(counters) > 1 { - return true; - } - let totals = retired.traffic.entry(tier.clone()).or_default(); - totals[Role::Publisher.idx()].add(counters.publisher.snapshot()); - totals[Role::Subscriber.idx()].add(counters.subscriber.snapshot()); - false + !done }); - !tiers.is_empty() + !orphan || !tiers.is_empty() }); } {