fix(net): keep benchmark delivery batches readable

quest/m0/nightly/README.md

@@ -30,13 +30,16 @@ This README's own work, after the children:
   macOS `receive_credit` timeout has no owner since its quest was deleted in
   [moq-dev/moq#6732](https://github.com/moq-dev/moq/pull/6732), so plan one under
   [test flakes](/quest/m0/test-flakes-2/README.md) if it recurs, the paired bench
-  threshold to [bench CI](/quest/m1/bench-ci.md), and the moq-net bench
-  panics in "Rust bench targets" to
-  [moq-net's benches run to completion](/quest/m1/bench-test-mode.md).
+  threshold to [bench CI](/quest/m1/bench-ci.md). The moq-net benchmark
+  fixtures were repaired in [moq-dev/moq#7940](https://github.com/moq-dev/moq/pull/7940).
   The receive-windows benchmark's overcommitted cells, skipped since
   [moq-dev/moq#7838](https://github.com/moq-dev/moq/pull/7838), moved to
   [m1](/quest/m1/receive-windows-evicted.md) on 2026-11-04 (maintainer).
 
+Decided 2026-11-05 (maintainer, /quest-plan): catch the next moq-net bench
+break in Nightly's measured coverage only. Do not add a test-mode bench run
+to the PR gate; the full run costs about 17 minutes per PR.
+
 Public API: none. Wire: none.
 
 ## Required

quest/m1/README.md

@@ -174,7 +174,6 @@ their end report, so an embedder's drain alert pages on real cuts.
 - [Main is no slower than release](/quest/m1/release-perf/README.md) - every benchmark where main is slower than release is fixed or accepted with a reason, checked by a final paired run before the cut
 - [Calibrate the paired nightly benchmarks](/quest/m1/bench-ci.md) - A/A and deliberately slowed Nightly runs set the paired comparison's threshold, so its regression alerts can be trusted
 - [moq-json's table/deltas bench measures the same work every iteration](/quest/m1/bench-json-deltas.md) - the `table/deltas/1%` cases stop flipping between snapshot and delta timings, so their trend history is usable
-- [moq-net's benches run to completion](/quest/m1/bench-test-mode.md) - the `origin`, `session` and `memory` benches stop panicking in test mode and Nightly's measured run, and the maintainer picks where CI catches the next break
 - [Publish moq-bench to crates.io](/quest/m1/bench-publish.md) - `cargo install moq-bench` installs one binary, with the host sampler as `moq-bench host`, first as 0.1.0
 - [Catalog contributions become includes on the root's timeline](/quest/m1/catalog-include.md) - rename the unreleased `contributions` map before the cut and state that included broadcasts share the root's timeline
 - [Unbounded groups](/quest/m1/unbounded-groups/README.md) - groups and frames lose their size caps; a group keeps a 32 MB tail, and readers behind it replay through paced flights or refresh

rs/moq-net/benches/memory.rs

@@ -100,6 +100,8 @@ fn now() -> moq_net::time::Instant {
 
 /// One publisher and one subscriber session over an in-memory pair.
 struct Link {
+	pools: Vec<cache::Pool>,
+	cache_now: moq_net::time::Instant,
 	tracks: Vec<track::Producer>,
 	subscribers: Vec<track::Subscriber>,
 	payload: Bytes,
@@ -110,12 +112,12 @@ struct Link {
 
 impl Link {
 	async fn new(streams: usize, size: usize) -> Self {
+		let mut pools = Vec::new();
 		let version: Version = VERSION.parse().unwrap();
-		let origin = |hop: u64| {
+		let mut origin = |hop: u64| {
 			let mut config = origin::Config::new(Hop::new(hop).unwrap());
-			// Hold one round's groups, with room for the handover, and no more.
-			config.pool =
-				cache::Pool::new(cache::Config::default().with_capacity((4 * streams * FRAMES * size) as u64));
+			config.pool = cache::Pool::new(cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY));
+			pools.push(config.pool.clone());
 			let (producer, driver) = origin::Producer::new(config);
 			tokio::spawn(moq_net::time::run(driver));
 			producer
@@ -171,6 +173,8 @@ impl Link {
 		}
 
 		let mut link = Self {
+			pools,
+			cache_now: now(),
 			tracks,
 			subscribers,
 			payload: Bytes::from(vec![0; size]),
@@ -180,6 +184,7 @@ impl Link {
 		};
 		// Warm every stream so the timed rounds skip first-group setup.
 		for _ in 0..4 {
+			link.tick();
 			link.round().await;
 		}
 		link
@@ -206,13 +211,31 @@ impl Link {
 		bytes
 	}
 
+	// Advance the cache/model clock between rounds to expire read history.
+	fn tick(&mut self) {
+		self.cache_now += cache::DEFAULT_EXPIRY;
+		for pool in &self.pools {
+			pool.poll(self.cache_now, &kio::Waiter::noop());
+		}
+	}
+
 	async fn measure(&mut self, iters: u64) -> Duration {
 		let expected = self.tracks.len() * FRAMES * self.payload.len();
-		let start = Instant::now();
+		let counting = std::env::var_os("MEMORY_ALLOCS").is_some();
+		let mut elapsed = Duration::ZERO;
 		for _ in 0..iters {
+			self.tick();
+			if counting {
+				COUNTING.store(true, Ordering::Relaxed);
+			}
+			let start = Instant::now();
 			assert_eq!(self.round().await, expected);
+			elapsed += start.elapsed();
+			if counting {
+				COUNTING.store(false, Ordering::Relaxed);
+			}
 		}
-		start.elapsed()
+		elapsed
 	}
 }
 
@@ -225,9 +248,7 @@ fn allocations() {
 		let mut link = rt.block_on(Link::new(streams, size));
 		ALLOCATIONS.store(0, Ordering::Relaxed);
 		ALLOCATED.store(0, Ordering::Relaxed);
-		COUNTING.store(true, Ordering::Relaxed);
 		rt.block_on(link.measure(ROUNDS as u64));
-		COUNTING.store(false, Ordering::Relaxed);
 
 		let frames = (ROUNDS * streams * FRAMES) as f64;
 		let allocations = ALLOCATIONS.load(Ordering::Relaxed) as f64 / frames;

rs/moq-net/benches/origin.rs

@@ -564,6 +564,7 @@ fn bench_reprice_duplicate(c: &mut Criterion) {
 							)
 							.unwrap()
 							.with_hidden(true)
+							.covering()
 							.routes()
 					})
 					.collect();

rs/moq-net/benches/session/mod.rs

@@ -34,9 +34,11 @@
 //! one datagram per track per round across publisher and viewer axes, then larger payloads
 //! and cadences, on the lite versions (moq-transport carries no datagram tracks). Throughput
 //! counts delivered datagrams, each checked for its sequence and payload, so a lost one fails.
-//! `SESSION_EXPIRY_ONLY=1` uses the standalone 30s-expiry pool and reports groups
-//! written plus charged cache bytes every 10s during long profiling runs. Each round
-//! advances the cache clock by a second, so the expiry window spans 30 rounds.
+//! Delivery uses expiry-only pools: history expires between rounds outside timing,
+//! so byte pressure cannot shed unread frames during a batch.
+//! `SESSION_EXPIRY_ONLY=1` advances the cache clock by a second instead of the 30s
+//! expiry window, retaining 30 rounds for profiling. It reports groups written and
+//! charged cache bytes every 10s during long runs.
 //!
 //! `session_join_*` times a new viewer connecting, resolving a broadcast, and
 //! receiving its latest group, swept over what the relays already announce.
@@ -123,9 +125,7 @@ const DASH_VERSIONS: [&str; 2] = ["moq-lite-06", "moq-transport-22"];
 /// A stats frame: a small JSON snapshot.
 const DASH_FRAME: usize = 128;
 
-/// Each endpoint gets its own bounded pool, as a relay configures one per
-/// process: without a byte target, every hop keeps each group for the whole
-/// expiry window and memory grows with the run length instead of the shape.
+/// Initial budgets for join and churn fixtures. Delivery rooms use expiry-only pools.
 const RELAY_CACHE: u64 = 16 * 1024 * 1024;
 const CLIENT_CACHE: u64 = 1024 * 1024;
 
@@ -517,7 +517,7 @@ pub(crate) trait Backend {
 	type Relay: Clone;
 	const PREFIX: &'static str;
 
-	fn relay(id: u64, capacity: u64) -> (Self::Relay, cache::Pool);
+	fn relay(id: u64, capacity: Option<u64>) -> (Self::Relay, cache::Pool);
 	fn origin(relay: &Self::Relay) -> &origin::Producer;
 	async fn connect(options: MockConnectOptions, relay: &Self::Relay, hidden: bool) -> Pair;
 }
@@ -529,7 +529,7 @@ impl Backend for Model {
 	type Relay = origin::Producer;
 	const PREFIX: &'static str = "";
 
-	fn relay(id: u64, capacity: u64) -> (Self::Relay, cache::Pool) {
+	fn relay(id: u64, capacity: Option<u64>) -> (Self::Relay, cache::Pool) {
 		make_origin(id, capacity)
 	}
 
@@ -566,9 +566,10 @@ impl Drop for Pair {
 	}
 }
 
-fn make_origin(id: u64, capacity: u64) -> (origin::Producer, cache::Pool) {
+fn make_origin(id: u64, capacity: Option<u64>) -> (origin::Producer, cache::Pool) {
 	let mut config = origin::Config::new(Hop::new(id).unwrap());
-	let pool = cache::Pool::new(if std::env::var_os("SESSION_EXPIRY_ONLY").is_some() {
+	let profiling = std::env::var_os("SESSION_EXPIRY_ONLY").is_some();
+	let pool = cache::Pool::new(if capacity.is_none() || profiling {
 		cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY)
 	} else {
 		cache::Config::default().with_capacity(capacity)
@@ -582,6 +583,7 @@ fn make_origin(id: u64, capacity: u64) -> (origin::Producer, cache::Pool) {
 /// Relays and publishers, before any viewer joins. Holds every handle that
 /// keeps a session or broadcast alive.
 struct Cluster<B: Backend> {
+	expiry: bool,
 	version: Version,
 	window: usize,
 	credit: Option<usize>,
@@ -601,8 +603,9 @@ struct Cluster<B: Backend> {
 }
 
 impl<B: Backend> Cluster<B> {
-	async fn new(version: &str, shape: Shape) -> Self {
+	async fn new(version: &str, shape: Shape, expiry: bool) -> Self {
 		let mut this = Self {
+			expiry,
 			version: version.parse().unwrap(),
 			window: shape.window,
 			credit: shape.credit,
@@ -620,7 +623,7 @@ impl<B: Backend> Cluster<B> {
 
 		for _ in 0..shape.relays {
 			this.next_hop += 1;
-			let (relay, pool) = B::relay(this.next_hop, RELAY_CACHE);
+			let (relay, pool) = B::relay(this.next_hop, (!expiry).then_some(RELAY_CACHE));
 			this.relays.push(relay);
 			this.pools.push(pool);
 		}
@@ -678,7 +681,7 @@ impl<B: Backend> Cluster<B> {
 
 	fn origin(&mut self, capacity: u64) -> (origin::Producer, cache::Pool) {
 		self.next_hop += 1;
-		make_origin(self.next_hop, capacity)
+		make_origin(self.next_hop, (!self.expiry).then_some(capacity))
 	}
 
 	/// Connect a viewer session to `relay`.
@@ -810,7 +813,7 @@ impl<B: Backend> Room<B> {
 		);
 		assert!(!shape.paced || shape.burst == 1, "a paced round writes one frame");
 		assert!(!shape.fetch || !shape.paced, "fetch rounds read finished groups");
-		let mut cluster = Cluster::<B>::new(version, shape).await;
+		let mut cluster = Cluster::<B>::new(version, shape, true).await;
 
 		let mut viewers = Vec::new();
 		let mut watched = vec![false; shape.total()];
@@ -879,12 +882,6 @@ impl<B: Backend> Room<B> {
 		room.writing.resize_with(room.watched.len(), || None);
 		room.reading.resize_with(shape.viewers * shape.watch, || None);
 
-		// First measure the complete batch without eviction. The delivery benchmark must
-		// fit its working set; cache-pressure loss is a different workload.
-		let capacities: Vec<_> = room.pools().map(cache::Pool::capacity).collect();
-		for pool in room.pools() {
-			pool.resize(None);
-		}
 		let blocked_before: usize = room
 			.viewers
 			.iter()
@@ -898,11 +895,6 @@ impl<B: Backend> Room<B> {
 			room.tick();
 			room.round().await;
 		}
-		for (pool, capacity) in room.pools().zip(capacities) {
-			// Keep the old and current batch resident during the handover, including
-			// bookkeeping rather than estimating a group from payload bytes alone.
-			pool.resize(capacity.map(|capacity| capacity.max(pool.used() * 2)));
-		}
 		// Exercise turnover as well as the empty cache, including Criterion's one-pass
 		// nightly smoke mode.
 		for _ in 0..4 {
@@ -945,11 +937,15 @@ impl<B: Backend> Room<B> {
 		self.cluster.pools.iter().chain(viewers)
 	}
 
-	/// Unpaced rounds otherwise collapse many complete batches into one cache tick,
-	/// where yesterday's read outranks today's unread write. Sample a new recency
-	/// epoch per round without sleeping or changing protocol time.
+	/// Advance the cache/model clock outside timing to expire old batches.
+	/// Each round completes its reads before this step; pending read deadlines
+	/// share this clock too.
 	fn tick(&mut self) {
-		self.cache_now += Duration::from_secs(1);
+		self.cache_now += if std::env::var_os("SESSION_EXPIRY_ONLY").is_some() {
+			Duration::from_secs(1)
+		} else {
+			cache::DEFAULT_EXPIRY
+		};
 		for pool in self.pools() {
 			pool.poll(self.cache_now, &kio::Waiter::noop());
 		}
@@ -1315,6 +1311,7 @@ pub(crate) fn range_updates<B: Backend>(c: &mut Criterion) {
 							publishers: 1,
 							..Shape::BASE
 						},
+						false,
 					)
 					.await;
 					let mut viewers = Vec::new();
@@ -1407,7 +1404,7 @@ fn join<B: Backend>(c: &mut Criterion, name: &str, shapes: impl IntoIterator<Ite
 			group.bench_function(BenchmarkId::from_parameter(shape.id(version)), |b| {
 				let cluster = cluster.get_or_insert_with(|| {
 					rt.block_on(async {
-						let cluster = Cluster::<B>::new(version, *shape).await;
+						let cluster = Cluster::<B>::new(version, *shape, false).await;
 						let payload = Bytes::from(vec![0; shape.frame]);
 						for track in &cluster.tracks {
 							for sequence in 0..shape.history {
@@ -1503,6 +1500,7 @@ impl<B: Backend> DashRoom<B> {
 				viewers: 0,
 				..Shape::BASE
 			},
+			false,
 		)
 		.await;
 
@@ -1666,6 +1664,7 @@ impl<B: Backend> ChurnRoom<B> {
 				viewers: 0,
 				..Shape::BASE
 			},
+			false,
 		)
 		.await;
 		let published = &cluster._broadcasts[0];
@@ -1881,7 +1880,7 @@ pub(crate) fn withdrawal<B: Backend>(c: &mut Criterion) {
 								publishers: 0,
 								..Shape::BASE
 							};
-							let cluster = Cluster::<B>::new("moq-lite-06", shape).await;
+							let cluster = Cluster::<B>::new("moq-lite-06", shape, false).await;
 							let mut announced = B::origin(cluster.relays.last().unwrap()).consume().routes();
 							let live: Vec<_> = (0..broadcasts)
 								.map(|i| {

rs/moq-relay/benches/session.rs

@@ -68,10 +68,10 @@ impl Backend for Relay {
 	type Relay = Node;
 	const PREFIX: &'static str = "relay_";
 
-	fn relay(id: u64, capacity: u64) -> (Node, cache::Pool) {
+	fn relay(id: u64, capacity: Option<u64>) -> (Node, cache::Pool) {
 		let mut config = moq_relay::cache::Config::default();
 		if std::env::var_os("SESSION_EXPIRY_ONLY").is_none() {
-			config.capacity = Some(capacity.to_string());
+			config.capacity = capacity.map(|capacity| capacity.to_string());
 		}
 		let cache = config.init().unwrap();
 		let pool = cache.pool.clone();