fix(net): sample a closing subscription's lag once more

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

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

rs/moq-net/src/model/cache.rs

@@ -493,8 +493,9 @@ pub(crate) struct Track {
 	sweep: OnceLock<usize>,
 
 	// 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<crate::stats::Production>,
 }
 
 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<crate::stats::Production> {
 		&self.production
 	}
 

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()
 			});
 		}
 		{
@@ -1829,7 +1830,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<cache::Track>,
+	// 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<Production>,
 	sampled: u64,
 }
 
@@ -1859,20 +1864,16 @@ impl Frontier {
 	pub(crate) fn watch(&self, track: &Arc<cache::Track>) {
 		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 +1883,14 @@ impl Frontier {
 	pub(crate) fn unwatch(&self, track: &Arc<cache::Track>) {
 		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 +1914,10 @@ impl FrontierInner {
 		let mut newest = None;
 		let mut first: Option<Duration> = 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 +1926,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
@@ -1954,6 +1950,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 +3069,99 @@ 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(stats: &Registry) -> u64 {
+		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.stats), GROUP_BYTES);
+	}
+
+	#[test]
+	fn a_subscription_closed_mid_interval_adds_only_the_bytes_since_its_last_sample() {
+		let mut h = Harness::new();
+		let sub = h.subscribe();
+		let mut last = h.tick();
+		h.produce();
+		last = {
+			let now = h.tick();
+			assert_eq!(grew(&last, &now), vec![(4, GROUP_BYTES)]);
+			now
+		};
+		h.produce();
+		h.produce();
+		drop(sub);
+
+		let now = h.tick();
+		assert_eq!(
+			grew(&last, &now).iter().map(|(_, bytes)| bytes).sum::<u64>(),
+			2 * GROUP_BYTES
+		);
+	}
+
+	#[test]
+	fn an_in_flight_delivery_defers_the_final_sample_to_its_own_drop() {
+		let mut h = Harness::new();
+		let sub = h.subscribe();
+		let delivery = Harness::write(&sub, &h.produce());
+		drop(sub);
+		// The delivery still holds the frontier, so it has not closed yet.
+		assert_eq!(recorded(&h.stats), 0);
+
+		drop(delivery);
+		assert_eq!(recorded(&h.stats), GROUP_BYTES);
+	}
+
+	#[test]
+	fn the_last_subscriber_samples_a_track_its_own_drop_releases() {
+		let mut h = Harness::new();
+		let sub = h.subscribe();
+		h.produce();
+		// The producer is gone, so the subscriber holds the last reference to the track.
+		drop(h.track);
+		drop(h.broadcast);
+
+		drop(sub);
+		assert_eq!(recorded(&h.stats), GROUP_BYTES);
+	}
+
+	#[test]
+	fn an_in_flight_delivery_samples_a_track_that_already_closed() {
+		let mut h = Harness::new();
+		let sub = h.subscribe();
+		let delivery = Harness::write(&sub, &h.produce());
+		drop(h.track);
+		drop(h.broadcast);
+		drop(sub);
+		assert_eq!(recorded(&h.stats), 0);
+
+		// The track died with the subscriber; the delivery still owes its bytes.
+		drop(delivery);
+		assert_eq!(recorded(&h.stats), GROUP_BYTES);
+	}
+
 	#[test]
 	fn a_closed_subscription_leaves_the_sampler() {
 		let mut h = Harness::new();