fix(net): judge sparse misses from changed runs

quest/m1/release-perf/README.md

@@ -1,4 +1,4 @@
-# Main is no slower than release
+# [M] Main is no slower than release
 
 ## Goal
 
@@ -216,20 +216,17 @@ Decided 2026-10-25 (maintainer), in
 [moq-dev/moq#6784](https://github.com/moq-dev/moq/pull/6784):
 
 - The sparse read slope from
-  [dynamic range requests](https://github.com/moq-dev/moq/pull/6171) is the
-  child below, and it blocks the cut.
+  [dynamic range requests](https://github.com/moq-dev/moq/pull/6171) blocks
+  the cut; [moq-dev/moq#6921](https://github.com/moq-dev/moq/pull/6921)
+  flattened it.
 - The TS schedule's growth is left to the
   [performance audit](/quest/m2/audit-perf.md); it does not block the cut.
-- Once the child lands, re-run the relay fanout pairing on a quiet host,
-  recording `/proc/loadavg`, with
+- Now that the sparse fix has landed, re-run the relay fanout pairing on a
+  quiet host, recording `/proc/loadavg`, with
   `BENCH_ARTIFACTS=<scratch> just bench origin/release` (it also re-checks
   the sparse rows). Fix it if the relay
   CPU still rises with consistent sign; otherwise this epic is done.
 
-## Required
-
-- [Sparse reads judge misses incrementally](/quest/m1/release-perf/sparse-misses.md) - `track_sparse_read` stays flat across gaps, as before range requests
-
 ## Related
 
 - [Performance audit](/quest/m2/audit-perf.md) - compares nightly to nightly; this epic compares main to release

rs/moq-net/benches/track.rs

@@ -26,6 +26,8 @@
 //! `track_sparse_read` hands out groups from a feed with every other sequence missing,
 //! holding a fixed number of gaps the reader still remembers: each hand-out must cost the
 //! same however many gaps it holds, so a walk over them shows up as a slope.
+//! `track_sparse_poll` repeats a parked read without changing the feed, so revisiting
+//! deferred gaps on an otherwise unchanged poll shows up independently of delivery.
 //!
 //! `track_sparse_expire` opens one new gap while one old gap expires, so a full expiry
 //! rescan on each arrival shows up as a slope over the gaps still within budget.
@@ -1292,6 +1294,26 @@ fn bench_sparse_read(c: &mut Criterion) {
 	group.finish();
 }
 
+fn bench_sparse_poll(c: &mut Criterion) {
+	let mut group = c.benchmark_group("track_sparse_poll");
+	for (name, info) in [("untimed", track::Info::default()), ("timed", timed())] {
+		for gaps in [16u64, 256, 4096] {
+			group.bench_function(BenchmarkId::new(name, gaps), |b| {
+				b.iter_custom(|iterations| {
+					let mut sparse = Sparse::new(info.clone(), gaps);
+					let waiter = kio::Waiter::noop();
+					let start = WallInstant::now();
+					for _ in 0..iterations {
+						assert!(black_box(sparse.sub.poll_recv_group(&waiter)).is_pending());
+					}
+					start.elapsed()
+				});
+			});
+		}
+	}
+	group.finish();
+}
+
 /// A sparse feed that opens one gap per millisecond, keeping `gaps` within its budget.
 struct Expiring {
 	sparse: Sparse,
@@ -1940,6 +1962,7 @@ criterion_group!(
 	bench_datagram_pressure,
 	bench_parked_read,
 	bench_sparse_read,
+	bench_sparse_poll,
 	bench_sparse_expire,
 	bench_gap_flush,
 	bench_wall,

rs/moq-net/src/ledger.rs

@@ -47,6 +47,44 @@ pub(crate) struct Ledger {
 	// Index both: one due gap must not make every arrival scan all the younger ones.
 	ages: BTreeSet<(Instant, u64)>,
 	grace: Duration,
+	changes: Changes,
+}
+
+/// Revisions of the current runs, indexed in both orders. Coalescing or removing a run
+/// removes its revision too, so an idle reader never retains a history of mutations.
+#[derive(Debug, Default)]
+pub(crate) struct Changes {
+	runs: BTreeMap<u64, (u64, u64)>,
+	changed: BTreeMap<u64, u64>,
+	revision: u64,
+}
+
+impl Changes {
+	pub fn revision(&self) -> u64 {
+		self.revision
+	}
+
+	pub fn remove(&mut self, start: u64) {
+		if let Some((_, revision)) = self.runs.remove(&start) {
+			self.changed.remove(&revision);
+		}
+	}
+
+	pub fn insert(&mut self, run: Range<u64>) {
+		self.remove(run.start);
+		self.revision = self.revision.checked_add(1).expect("run revisions exhausted");
+		self.runs.insert(run.start, (run.end, self.revision));
+		self.changed.insert(self.revision, run.start);
+	}
+
+	pub fn since(&self, revision: u64) -> impl Iterator<Item = Range<u64>> + '_ {
+		self.changed
+			.range((std::ops::Bound::Excluded(revision), std::ops::Bound::Unbounded))
+			.map(|(_, start)| {
+				let (end, _) = self.runs.get(start).expect("a changed run stays indexed");
+				*start..*end
+			})
+	}
 }
 
 impl Default for Ledger {
@@ -62,6 +100,7 @@ impl Ledger {
 			runs: BTreeMap::new(),
 			ages: BTreeSet::new(),
 			grace,
+			changes: Changes::default(),
 		}
 	}
 
@@ -71,6 +110,7 @@ impl Ledger {
 	}
 
 	fn remove(&mut self, start: u64) -> Run {
+		self.changes.remove(start);
 		let run = self.runs.remove(&start).expect("an indexed run");
 		self.ages.remove(&(run.since, start));
 		run
@@ -112,6 +152,20 @@ impl Ledger {
 				self.runs.insert(first, Run { end, since });
 			}
 		}
+		self.changes.insert(first..end);
+	}
+
+	/// Account a live arrival, refreshing a cached incarnation even if its sequence was
+	/// accounted before. Its gap ages stay unchanged, but readers must revisit a refusal.
+	pub fn arrive(&mut self, sequence: u64, now: Instant) {
+		let revision = self.changes.revision();
+		self.account(sequence..sequence.saturating_add(1), now);
+		if self.changes.revision() == revision
+			&& let Some((&start, run)) = self.runs.range(..=sequence).next_back()
+			&& sequence < run.end
+		{
+			self.changes.insert(start..run.end);
+		}
 	}
 
 	/// Fold every gap older than the grace into the runs around it.
@@ -126,12 +180,9 @@ impl Ledger {
 				break;
 			}
 			let run = self.remove(start);
-			self.runs
-				.range_mut(..start)
-				.next_back()
-				.expect("not the first run")
-				.1
-				.end = run.end;
+			let (&lower, previous) = self.runs.range_mut(..start).next_back().expect("not the first run");
+			previous.end = run.end;
+			self.changes.insert(lower..run.end);
 		}
 	}
 
@@ -174,6 +225,10 @@ impl Ledger {
 			.map(move |(&start, run)| start.max(groups.start)..run.end.min(groups.end))
 	}
 
+	pub fn changes(&self) -> &Changes {
+		&self.changes
+	}
+
 	/// Whether every group in `groups` is accounted for.
 	pub fn covers(&self, groups: Range<u64>) -> bool {
 		groups.is_empty() || self.run_end(groups.start).is_some_and(|end| groups.end <= end)
@@ -199,6 +254,7 @@ impl Ledger {
 				self.ages.remove(&(run.since, start));
 				run.since = now;
 				self.ages.insert((now, start));
+				self.changes.insert(start..run.end);
 			}
 			below = run.end;
 		}
@@ -254,6 +310,27 @@ impl Ledger {
 mod tests {
 	use super::*;
 
+	#[test]
+	fn changes_keep_only_current_runs_across_coalescing_and_expiry() {
+		let now = Instant::ZERO;
+		let mut ledger = Ledger::new(GRACE);
+		ledger.account(0..1, now);
+		ledger.account(4..5, now);
+		ledger.account(8..9, now);
+		let revision = ledger.changes().revision();
+		ledger.account(2..3, now);
+		assert_eq!(ledger.changes().since(revision).collect::<Vec<_>>(), vec![2..3]);
+		let revision = ledger.changes().revision();
+		ledger.account(1..4, now);
+		assert_eq!(ledger.changes().since(revision).collect::<Vec<_>>(), vec![0..5]);
+		ledger.expire(now + GRACE);
+		assert_eq!(ledger.changes().since(0).collect::<Vec<_>>(), vec![0..9]);
+		assert_eq!(ledger.changes.changed.len(), ledger.runs.len());
+		let revision = ledger.changes().revision();
+		ledger.account(2..7, now + GRACE);
+		assert_eq!(ledger.changes().revision(), revision, "already accounted is unchanged");
+	}
+
 	/// Account `groups` and fold what aged out, as each arrival does.
 	fn arrive(ledger: &mut Ledger, groups: Range<u64>, now: Instant) {
 		ledger.account(groups, now);

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

@@ -446,6 +446,7 @@ pub(crate) struct TrackState {
 	drops: RangeMap<u64, StreamError>,
 	// Bumped on every insert into `drops`, so a cursor rescans it only after a change.
 	drops_revision: u64,
+	drops_changes: ledger::Changes,
 
 	// Whether the cache reflects the live feed: readers get nothing from it while it does
 	// not, since how stale it is cannot be told; fetches still do. A track is live from
@@ -886,7 +887,26 @@ impl TrackState {
 			return;
 		}
 		self.cache.wakes().arrived(groups.clone());
-		self.drops.insert(groups, err);
+		self.change_drops(groups, Some(err));
+	}
+
+	/// Keep the change index aligned with the range map, including the neighbours an
+	/// insertion coalesces and the two pieces a removal can leave.
+	fn change_drops(&mut self, groups: ops::Range<u64>, err: Option<StreamError>) {
+		let mut affected = groups.start.saturating_sub(1)..groups.end.saturating_add(1);
+		let old: Vec<_> = self.drops.overlapping(&affected).map(|(run, _)| run.clone()).collect();
+		for run in old {
+			self.drops_changes.remove(run.start);
+			affected.start = affected.start.min(run.start);
+			affected.end = affected.end.max(run.end);
+		}
+		match err {
+			Some(err) => self.drops.insert(groups, err),
+			None => self.drops.remove(groups),
+		}
+		for (run, _) in self.drops.overlapping(&affected) {
+			self.drops_changes.insert(run.clone());
+		}
 		self.drops_revision += 1;
 	}
 
@@ -1775,7 +1795,7 @@ impl TrackState {
 
 	/// Record that the live feed delivered `sequence`, then fold the gaps that aged out.
 	fn record_arrival(&mut self, sequence: u64, now: crate::time::Instant) {
-		self.arrivals.account(sequence..sequence.saturating_add(1), now);
+		self.arrivals.arrive(sequence, now);
 		// A feed still delivering keeps the gap above its last arrival open.
 		if self.tail.is_some() {
 			self.tail = Some(now);
@@ -1797,8 +1817,7 @@ impl TrackState {
 	/// How long a missing live group is waited for, from the max delay the track's
 	/// subscribers ask for.
 	fn arrival_grace(&self) -> Duration {
-		let max_delay = snapshot_subscription(&self.subscriptions, self.max_age_bound())
-			.map_or(Duration::ZERO, |subscription| subscription.budget());
+		let max_delay = clamp_max_delay(self.subscriptions.lock().max_delay(), self.max_age_bound());
 		ledger::grace(max_delay)
 	}
 
@@ -1908,8 +1927,7 @@ impl TrackState {
 			&& self.drops.contains_key(&sequence)
 			&& let Some(end) = sequence.checked_add(1)
 		{
-			self.drops.remove(sequence..end);
-			self.drops_revision += 1;
+			self.change_drops(sequence..end, None);
 		}
 		let now = self.cache.pool().now_wall();
 		if visible {
@@ -6384,7 +6402,17 @@ impl Cursor {
 			misses.judge(
 				&state,
 				(from, revision, cap),
-				|| below(window.missable(from), cap),
+				|span| {
+					let end = cap.map_or(*span.end(), |cap| (*span.end()).min(cap.saturating_sub(1)));
+					if cap == Some(0) || *span.start() > end {
+						return RangeInclusiveSet::new();
+					}
+					window
+						.missable(*span.start())
+						.take_while(|run| *run.start() <= end)
+						.map(|run| *run.start()..=(*run.end()).min(end))
+						.collect()
+				},
 				drift,
 				&mut given_up,
 				waiter,
@@ -8075,6 +8103,35 @@ mod test {
 		Producer::new(Arc::new(broadcast::Info::default()), name, info)
 	}
 
+	#[test]
+	fn drop_changes_follow_split_coalesced_and_recreated_runs() {
+		let producer = track_producer("test", Info::default());
+		let indexed = || {
+			let state = producer.state.read();
+			let mut changed: Vec<_> = state.drops_changes.since(0).collect();
+			changed.sort_by_key(|run| run.start);
+			let current: Vec<_> = state.drops.iter().map(|(run, _)| run.clone()).collect();
+			assert_eq!(changed, current, "one change entry per current drop run");
+		};
+		producer.drop_groups(10..20, StreamError::Old);
+		producer.drop_groups(20..30, StreamError::Old);
+		indexed();
+		producer.drop_groups(15..25, StreamError::Cancel);
+		indexed();
+		producer
+			.create_group(group::Info { sequence: 17 })
+			.unwrap()
+			.finish()
+			.unwrap();
+		indexed();
+		producer.drop_groups(14..26, StreamError::Old);
+		indexed();
+		assert_eq!(
+			producer.state.read().drops_changes.since(0).collect::<Vec<_>>(),
+			vec![10..30]
+		);
+	}
+
 	/// Let `duration` pass on the cache pool `producer`'s groups charge into.
 	fn elapse(producer: &Producer, duration: Duration) {
 		producer.broadcast.pool.step(duration);