fix(net): preserve owed groups across history resumes

quest/m1/README.md

@@ -196,7 +196,6 @@ per-frame CPU growth since #5926.
 - [Ordered reads can skip owed holes, and snapshot latest() does](/quest/m1/ordered-owed-holes.md) - moq-net's `Ordered` gains a read that skips past an owed hole to the newest buffered group, while keeping the reader's bounds
 - [A Rust relay never replays from group 0 to a lite-03/04 subscriber](/quest/m1/relay-inflight-replay.md) - A Rust relay or publisher serving a lite-03 or lite-04 SUBSCRIBE from a track copy that is still resolving starts at the newest group once the copy resolves
 - [The resume model passes its long run again](/quest/m1/resume-model-7542.md) - fix #7542's regression that drops groups a long-budget reader can still get
-- [A resume more than one group behind never hangs](/quest/m1/ietf-resume-behind.md) - a resumed read gets every group or a drop on every wire version
 - [Stats name the route instance's epoch with a type](/quest/m1/stats-epoch-type.md) - `GroupFrame::epoch` gets a dedicated type and tests stop mislabelling the instance
 
 ## Related

rs/moq-net/benches/track.rs

@@ -1915,7 +1915,7 @@ struct Fills {
 }
 
 impl Fills {
-	fn new(readers: usize, runs: u64) -> Self {
+	fn new(readers: usize, runs: u64, budget: Option<Duration>) -> Self {
 		let broadcast = broadcast::Info::default().produce();
 		let mut track = broadcast.create_track("bench", timed()).unwrap();
 		let dynamic = track.requests();
@@ -1933,7 +1933,7 @@ impl Fills {
 			.chain([track::Range::groups(start..)]);
 		let subscription = track::Subscription::default()
 			.with_ranges(ranges)
-			.with_delay(track::Delay::default().with_max(AGED_BUDGET));
+			.with_delay(track::Delay::default().with_max(budget));
 		let readers = (0..readers).map(|_| track.subscribe(subscription.clone())).collect();
 		Self {
 			_broadcast: broadcast,
@@ -1975,17 +1975,23 @@ impl Fills {
 /// swept over the readers and the runs. Each reader must pay the same per run, so a cost
 /// that grows with the audience or the runs shows up as a slope.
 fn bench_fill(c: &mut Criterion) {
-	let mut group = c.benchmark_group("track_fill");
-	group.sample_size(10);
-	for readers in [1usize, 8, 64] {
-		for runs in [1u64, 64, 1024] {
-			group.throughput(Throughput::Elements(runs * readers as u64));
-			group.bench_function(BenchmarkId::new(format!("{readers}r"), runs), |b| {
-				b.iter_batched_ref(|| Fills::new(readers, runs), Fills::run, BatchSize::PerIteration);
-			});
+	for (name, budget) in [("track_fill", Some(AGED_BUDGET)), ("track_fill_unlimited", None)] {
+		let mut group = c.benchmark_group(name);
+		group.sample_size(10);
+		for readers in [1usize, 8, 64] {
+			for runs in [1u64, 64, 1024] {
+				group.throughput(Throughput::Elements(runs * readers as u64));
+				group.bench_function(BenchmarkId::new(format!("{readers}r"), runs), |b| {
+					b.iter_batched_ref(
+						|| Fills::new(readers, runs, budget),
+						Fills::run,
+						BatchSize::PerIteration,
+					);
+				});
+			}
 		}
+		group.finish();
 	}
-	group.finish();
 }
 
 /// Readers whose disjoint ranges all fall below where the feed starts, and the handler that

rs/moq-net/src/ietf/subscriber.rs

@@ -15194,6 +15194,60 @@ mod stitch_tests {
 		assert!(run.track.consume().peek_group(SEQUENCE).is_none(), "nothing is cached");
 	}
 
+	/// A refused multi-group FETCH names every owed group dropped, without retrying
+	/// each group or ending the live track. This exercises the wire refusal decoder.
+	#[kio::test]
+	async fn a_refused_range_fetch_reports_every_owed_group_dropped() {
+		for version in [Version::Draft18, Version::Draft20, Version::Draft22] {
+			let mut refusal = Vec::new();
+			let mut writer = Encoder::new(&mut refusal, version.into());
+			writer.varint(ietf::RequestError::ID).unwrap();
+			ietf::RequestError {
+				error_code: request::NOT_SUPPORTED,
+				reason_phrase: "FETCH spans more than one group".into(),
+				retry_interval: 0,
+			}
+			.encode(&mut writer, version)
+			.unwrap();
+			let session = ScriptedSession::per_stream_eof(vec![vec![], refusal]);
+			let mut run = GroupFetchRun::with_session(version, session.clone()).await;
+			let now = kio::clock::now();
+			run.track.broadcast().pool.poll(now, &kio::Waiter::noop());
+			run.track.received();
+			run.track.start_at(3).unwrap();
+			run.track.create_group(3u64.into()).unwrap().finish().unwrap();
+			let requests = run.track.requests();
+			let mut reader = run.track.subscribe(
+				track::Subscription::default()
+					.with_range(1..3)
+					.with_delay(track::Delay::default().with_max(None)),
+			);
+			let waiter = kio::Waiter::noop();
+			assert!(reader.poll_drop(&waiter).is_pending());
+			assert!(reader.poll_recv_group(&waiter).is_pending());
+			let Poll::Ready(Ok(fill)) = requests.poll_requested_fill(&waiter) else {
+				panic!("{version}: missing history was not requested");
+			};
+			assert_eq!((fill.start().group, fill.end()), (1, 3));
+			run.subscriber
+				.clone()
+				.run_group_fetch(Path::from_static("broadcast"), "video".into(), fill, None)
+				.await;
+			// The peer may still deliver a late subgroup stream during the arrival grace.
+			run.track.broadcast().pool.poll(now + crate::ledger::GRACE, &waiter);
+			assert_eq!(
+				reader.poll_drop(&waiter),
+				Poll::Ready((1..3, crate::StreamError::NotFound)),
+				"{version}"
+			);
+			assert!(reader.poll_drop(&waiter).is_pending(), "each drop is reported once");
+			assert!(matches!(reader.poll_recv_group(&waiter), Poll::Ready(Ok(None))));
+			assert!(!run.track.is_closed(), "the live track survives");
+			// The harness opened its unused data stream, then the one range FETCH.
+			assert_eq!(session.log.bi_opens(), 2, "no per-group fallback");
+		}
+	}
+
 	/// A third-party publisher may answer a range even though ours refuses it. The subscriber
 	/// caches the supplied groups and reports the gaps between them as misses.
 	#[kio::test]

rs/moq-net/src/lite/subscriber.rs

@@ -2555,6 +2555,35 @@ mod tests {
 		);
 	}
 
+	/// A bounded feed goes live at its answer alone only when its resolved ranges stop short
+	/// of Largest. One that serves Largest waits for it, so a finite read judges an old
+	/// cached group against that edge rather than against itself.
+	#[kio::test]
+	async fn a_bounded_feed_goes_live_ahead_only_without_largest() {
+		let tail = kio::Producer::new(Tail::default());
+		let Sub::Active(mut active) = feed_sub(&tail).await else {
+			unreachable!("feed_sub is active")
+		};
+		let largest = Some(Position::group(10));
+		active.ranges = vec![lite::Range::span(Some(0), 0, Some(10), None)];
+		active.served = Some(0);
+		active.starts = vec![Some(0)];
+		assert_eq!(active.never_serves(largest), None, "group 10 is in 0..11");
+
+		active.ranges = vec![lite::Range::span(Some(0), 0, Some(3), None)];
+		assert_eq!(active.never_serves(largest), largest, "0..4 never reaches group 10");
+
+		active.ranges.push(lite::Range::span(Some(10), 0, Some(12), None));
+		active.starts = vec![Some(0), Some(10)];
+		assert_eq!(active.never_serves(largest), None, "a later range serves it");
+		active.starts[1] = None;
+		assert_eq!(active.never_serves(largest), largest, "that range served nothing");
+
+		active.ranges[1].end = None;
+		assert_eq!(active.never_serves(largest), None, "an open feed reaches it");
+		assert_eq!(active.never_serves(None), None, "nothing to go live at");
+	}
+
 	/// QUIC does not order streams, so group 2's header can reach the relay before group 1's.
 	/// A fetch held for group 1 keeps waiting through the grace and is served when it lands.
 	#[kio::test]
@@ -7578,6 +7607,19 @@ impl SubStream {
 		Some(start..end)
 	}
 
+	/// `largest`, when this bounded subscription's ranges, as SUBSCRIBE_START resolved them,
+	/// stop short of it, so the feed never delivers it. The range and arrival ledger then wait
+	/// for the groups it owes instead of an edge outside the requested history. A feed that
+	/// serves Largest waits for it, so an old cache is judged against that edge.
+	fn never_serves(&self, largest: Option<Position>) -> Option<Position> {
+		largest.filter(|largest| {
+			bounded(&self.ranges)
+				&& self
+					.covered()
+					.is_none_or(|spans| !spans.iter().any(|span| span.contains(&largest.group)))
+		})
+	}
+
 	/// The groups the publisher owed once it FINs a covered bounded lite-07 subscription
 	/// without SUBSCRIBE_END: each range from its absolute start, or else from where
 	/// SUBSCRIBE_START resolved delivery, up to its end. Groups below SUBSCRIBE_START were
@@ -9585,9 +9627,16 @@ impl ServeLoop {
 										}
 										// No range starting anywhere leaves nothing owed.
 										let served = start.starts.iter().flatten().min().copied();
+										active.served = served;
+										active.starts = start.starts.clone();
 										// Where the live feed is, on versions whose answer says.
 										if serve.subscriber.version.has_largest() {
-											self.serving.set_live(start.largest, served);
+											// A bounded feed that never delivers Largest goes live at the
+											// answer alone; one that does waits for it to show.
+											match active.never_serves(start.largest) {
+												Some(largest) => self.serving.set_live_ahead(largest, served),
+												None => self.serving.set_live(start.largest, served),
+											}
 											if let Some(largest) = start.largest {
 												self.serving.report_largest(largest);
 											}
@@ -9608,8 +9657,6 @@ impl ServeLoop {
 												_ => self.serving.start_at(served),
 											};
 										}
-										active.served = served;
-										active.starts = start.starts.clone();
 										// The groups the SUBSCRIBE asked for below it are not
 										// waited for, whatever the demand asks later. One that
 										// still arrives is delivered, so they are not dropped.