test(net): check refusal state cheaply and benchmark its scaling

quest/m0/ci/README.md

@@ -163,5 +163,4 @@ The historical samples locate candidates; they are not controlled benchmarks.
 - [Halve the slowest tests' CPU](/quest/m0/ci/slow-tests.md) - the slowest 1% of tests take under half their 289 CPU-seconds, with coverage kept
 - [Release credentials off main](/quest/m0/ci/publish-release-only.md) - `publish` admits only `release` and release tags once its `workflow_run` publishers run on their tag
 - [The queue host keeps the queue's disk](/quest/m0/ci/queue-host-disk.md) - agent sessions can't starve the queue runner of disk, and a crashed runner restarts
-- [The next test-CPU cuts](/quest/m0/ci/test-cost-followups.md) - the slow-test leftovers from #6987, cut at their source
 - [Biome checks every file it can parse](/quest/m0/ci/biome-whole-repo.md) - Biome runs over the whole repository and main is clean under it

rs/moq-net/benches/track.rs

@@ -114,6 +114,10 @@
 //! group filled must cost each reader the same, so a walk over the ranges, the groups of each
 //! fill, or the audience shows up as a slope.
 //!
+//! `track_fill_refusal` refuses a wide fill one group at a time, swept over the number of
+//! refusals and readers. Each reader consumes each refusal before the next, so the cost per
+//! refusal must not grow with the history of answers.
+//!
 //! Run with `cargo bench -p moq-net --bench track`.
 
 #![expect(
@@ -2074,6 +2078,98 @@ fn bench_range_filling(c: &mut Criterion) {
 	group.finish();
 }
 
+/// A shared fill refused newest first, with each reader taking each answer as it arrives.
+struct RefusedFill {
+	_broadcast: broadcast::Producer,
+	_track: track::Producer,
+	_dynamic: track::Requests,
+	fill: track::Fill,
+	readers: Vec<track::Subscriber>,
+}
+
+impl RefusedFill {
+	const START: u64 = 1 << 40;
+
+	fn new(readers: usize) -> Self {
+		let broadcast = broadcast::Info::default().produce();
+		let mut track = broadcast.create_track("bench", timed()).unwrap();
+		let dynamic = track.requests();
+		track.start_at(Self::START).unwrap();
+		let mut group = track.create_group(group::Info { sequence: Self::START }).unwrap();
+		group.write_frame(Timestamp::ZERO, Bytes::from_static(b"x")).unwrap();
+		group.finish().unwrap();
+		let waiter = kio::Waiter::noop();
+		let readers = (0..readers)
+			.map(|_| {
+				let mut reader = track.subscribe(
+					track::Subscription::default()
+						.with_range(0..)
+						.with_max_delay(AGED_BUDGET),
+				);
+				let Poll::Ready(Ok(Some(group))) = reader.poll_recv_group(&waiter) else {
+					panic!("the live group is cached");
+				};
+				assert_eq!(group.sequence, Self::START);
+				assert!(reader.poll_recv_group(&waiter).is_pending());
+				reader
+			})
+			.collect();
+		let Poll::Ready(Ok(fill)) = dynamic.poll_requested_fill(&waiter) else {
+			panic!("the miss below the start is asked");
+		};
+		assert!(
+			dynamic.poll_requested_fill(&waiter).is_pending(),
+			"readers share the fill"
+		);
+		Self {
+			_broadcast: broadcast,
+			_track: track,
+			_dynamic: dynamic,
+			fill,
+			readers,
+		}
+	}
+
+	fn run(&mut self, refusals: u64) {
+		let waiter = kio::Waiter::noop();
+		for refused in 1..=refusals {
+			let sequence = self.fill.wanted().unwrap();
+			self.fill.drop_groups(sequence..sequence + 1);
+			for reader in &mut self.readers {
+				assert!(reader.poll_recv_group(&waiter).is_pending());
+			}
+			assert_eq!(self.fill.wanted(), Some(Self::START - refused - 1));
+		}
+	}
+}
+
+fn bench_fill_refusal(c: &mut Criterion) {
+	let mut group = c.benchmark_group("track_fill_refusal");
+	group.sample_size(10);
+	for readers in [1usize, 8, 64] {
+		for refusals in [16u64, 256, 4096] {
+			group.throughput(Throughput::Elements(refusals * readers as u64));
+			group.bench_function(BenchmarkId::new(format!("{readers}r"), refusals), |b| {
+				b.iter_batched_ref(
+					|| RefusedFill::new(readers),
+					|fill| fill.run(refusals),
+					BatchSize::PerIteration,
+				);
+			});
+		}
+	}
+	// Keep the long single-reader history that exposes even a cheap per-refusal copy.
+	group.throughput(Throughput::Elements(100_000));
+	group.bench_function("1r/100000", |b| {
+		b.iter_batched_ref(
+			|| RefusedFill::new(1),
+			|fill| fill.run(100_000),
+			BatchSize::PerIteration,
+		);
+	});
+	group.finish();
+}
+
 criterion_group!(
 	benches,
 	bench_range_widening,
@@ -2106,6 +2202,7 @@ criterion_group!(
 	bench_empty_run,
 	bench_time_end,
 	bench_fill,
-	bench_range_filling
+	bench_range_filling,
+	bench_fill_refusal
 );
 criterion_main!(benches);

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

@@ -19848,12 +19848,13 @@ mod test {
 	}
 
 	/// A handler refusing a wide fill one group at a time, as a FETCH-capable session does,
-	/// pays for each refusal rather than for every refusal before it.
+	/// advances after each refusal and coalesces the resolved run, forgetting consumed answers.
+	/// `track_fill_refusal` benchmarks the cost over refusals and readers.
 	#[test]
 	#[cfg(feature = "async")]
 	fn a_fill_refused_group_by_group_stays_linear() {
 		const WIDE: u64 = 1 << 40;
-		const REFUSED: u64 = 100_000;
+		const REFUSED: u64 = 32;
 		let mut producer = track_producer("test", Info::timed());
 		let dynamic = producer.requests();
 		let mut subscriber = producer.subscribe(replay().with_range(0..));
@@ -19864,12 +19865,18 @@ mod test {
 		let Poll::Ready(Ok(fill)) = dynamic.poll_requested_fill(&kio::Waiter::noop()) else {
 			panic!("the miss below the start is asked");
 		};
-		for _ in 0..REFUSED {
+		for refused in 1..=REFUSED {
 			let sequence = fill.wanted().unwrap();
 			fill.drop_groups(sequence..sequence + 1);
 			assert!(subscriber.poll_recv_group(&kio::Waiter::noop()).is_pending());
+			assert_eq!(fill.wanted(), Some(WIDE - refused - 1));
+			assert_eq!(fill.answers(), 0, "the reader consumed the refusal");
+			assert_eq!(
+				fill.resolved_runs(),
+				vec![WIDE - refused..WIDE],
+				"adjacent refusals remain one resolved run"
+			);
 		}
-		assert_eq!(fill.wanted(), Some(WIDE - REFUSED - 1));
 	}
 
 	/// A descending reader joining a queued fill makes it newest first, as any descending

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

@@ -786,6 +786,12 @@ impl Fill {
 		outcome.refused.iter().count() + outcome.accepted.iter().count()
 	}
 
+	/// The runs answered so far, even once their readers consumed the answers.
+	#[cfg(all(test, feature = "async"))]
+	pub(super) fn resolved_runs(&self) -> Vec<ops::Range<u64>> {
+		self.inner.result.read().resolved.iter().cloned().collect()
+	}
+
 	/// The next group still owed in the fill's order: a reader still reads it, and it was
 	/// neither answered nor delivered live since it was asked. `None` once none is owed or
 	/// no reader waits on the fill, so stop serving it.