fix(net): keep a backup for a group read from a single route

quest/m1/README.md

@@ -126,7 +126,6 @@ tracks, so per-group stats stop costing cluster announces.
 - [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - both modes have explicit profiles, automated validation, loss tests, and a dedicated guide
 - [Replica resume](/quest/m1/replica-resume.md) - a lagging same-epoch replica resumes with every frame exactly once and in order, on lite-07 and on moq-transport
 - [Rust resume checks compression](/quest/m1/resume-compress.md) - a same-epoch route copy with a different `compress` is refused instead of corrupting framing
-- [Rust resume backs single-route groups](/quest/m1/resume-handed-out.md) - a group first read from one route keeps the later copy that holds it, with a passthrough benchmark
 - [WebGPU on Safari](/quest/m1/webgpu-safari.md) - the WebGPU renderer is verified on Safari 26 for macOS and iOS
 - [Exporters follow a rendition to a new track](/quest/m1/export-rendition-rebind.md) - the fMP4, TS, MKV and FLV exporters rebuild a rendition whose resolved `track` or `broadcast` changes under the same ID
 - [HLS cursor failure](/quest/m1/hls-segments-failure.md) - the export segment cursor returns a failed timeline as an error after draining, so a truncated recording never looks complete

rs/moq-net/benches/origin.rs

@@ -17,6 +17,9 @@
 //! An equal-cost pool is swept the same way, over its members and the paths it
 //! already serves, and over those paths and the cursors watching the pool.
 //!
+//! One-frame groups through a front with a single route are swept over readers and the
+//! groups each holds open, which is what a reader records per group it hands out.
+//!
 //! A route swap on one front is swept over its tracks and the copies each track
 //! still holds from earlier routes, and a front retiring over the fronts around it.
 //! Readers draining a swapped track are swept over their count, the groups every
@@ -1113,6 +1116,65 @@ fn bench_relay(c: &mut Criterion) {
 	group.finish();
 }
 
+/// `(readers, open)` shapes for [`bench_passthrough`]: readers of one track, and the
+/// groups each holds open at once.
+const PASSTHROUGH: [(usize, usize); 4] = [(1, 1), (1, 64), (100, 1), (100, 64)];
+
+/// One-frame groups read through a front that only ever had one route, so each handed-out
+/// group's per-reader record (kept for a later route to back it) is a visible share of the
+/// cost. Each reader holds `open` groups at once, the live count that record tracks.
+fn bench_passthrough(c: &mut Criterion) {
+	let mut group = c.benchmark_group("origin/passthrough");
+	for (readers, open) in PASSTHROUGH {
+		group.throughput(Throughput::Elements((readers * open) as u64));
+		group.bench_function(BenchmarkId::from_parameter(format!("{readers}r_{open}g")), |b| {
+			let runtime = tokio::runtime::Builder::new_current_thread()
+				.enable_all()
+				.build()
+				.unwrap();
+			let (producer, driver) = origin::Producer::new(origin::Config::default());
+			runtime.spawn(moq_net::time::run(driver));
+			let broadcast = producer.publish("room/live", origin::Route::default()).unwrap();
+			let source = broadcast.create_track("0", timed()).unwrap();
+			let mut subscriptions = runtime.block_on(async {
+				let resolved = producer.consume().request_broadcast("room/live", None).await.unwrap();
+				let track = resolved.track("0").unwrap();
+				let mut subscriptions = Vec::with_capacity(readers);
+				for _ in 0..readers {
+					subscriptions.push(track.subscribe(None).await.unwrap());
+				}
+				subscriptions
+			});
+			let mut held: Vec<Vec<moq_net::group::Consumer>> = (0..readers).map(|_| Vec::with_capacity(open)).collect();
+
+			b.iter_custom(|iterations| {
+				runtime.block_on(async {
+					let started = std::time::Instant::now();
+					for _ in 0..iterations {
+						// Received as each is written: a reader behind the live edge skips ahead.
+						for _ in 0..open {
+							let mut group = source.append_group().unwrap();
+							group.write_frame(Timestamp::ZERO, b"frame".as_ref()).unwrap();
+							group.finish().unwrap();
+							for (subscription, held) in subscriptions.iter_mut().zip(&mut held) {
+								held.push(subscription.recv_group().await.unwrap().expect("group"));
+							}
+						}
+						for held in &mut held {
+							for group in held.iter_mut() {
+								group.read_frame().await.unwrap().expect("frame");
+							}
+							held.clear();
+						}
+					}
+					started.elapsed()
+				})
+			});
+		});
+	}
+	group.finish();
+}
+
 /// Re-poll stalled groups through a front, including their subscription budget.
 /// Sweep tracks and readers separately to expose unrelated table scans.
 fn bench_parked(c: &mut Criterion) {
@@ -1525,6 +1587,7 @@ criterion_group!(
 	bench_narrow_delivery,
 	bench_narrow,
 	bench_relay,
+	bench_passthrough,
 	bench_parked,
 	bench_copy_walk,
 	bench_resume_overlap

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

@@ -833,7 +833,7 @@ struct Reader {
 	groups: (u64, Cap),
 	/// Whether groups were ever asked for; until then the reader keeps no replaced copy.
 	groups_read: bool,
-	/// The groups handed out while copies overlap, by sequence, held while each is read.
+	/// The groups handed out, by sequence, held while each is read.
 	out: Vec<(u64, Weak<()>)>,
 }
 
@@ -1181,18 +1181,15 @@ impl Reader {
 		let sequence = group.sequence;
 		self.newest = self.newest.max(Some(sequence));
 		// Another copy holding the group, now or once it lands there, is where a failure
-		// continues it, so it stays while the group is read; see [`Copy::retains`]. Only
-		// while copies overlap: a passthrough read has none.
-		let out = (self.copies.len() > 1).then(|| {
-			// Sweep the groups no longer read once the list fills, so it tracks the live count.
-			if self.out.len() == self.out.capacity() {
-				self.out.retain(|(_, out)| out.strong_count() > 0);
-				self.out.reserve(self.out.len().max(1));
-			}
-			let out = Arc::new(());
-			self.out.push((sequence, Arc::downgrade(&out)));
-			out
-		});
+		// continues it, so it stays while the group is read; see [`Copy::retains`]. Recorded
+		// even with one copy, since a later route's copy may come to hold it.
+		// Sweep the groups no longer read once the list fills, so it tracks the live count.
+		if self.out.len() == self.out.capacity() {
+			self.out.retain(|(_, out)| out.strong_count() > 0);
+			self.out.reserve(self.out.len().max(1));
+		}
+		let out = Arc::new(());
+		self.out.push((sequence, Arc::downgrade(&out)));
 		let copy = &self.copies[index];
 		let handoff = group.handoff();
 		group.with_recover(Recover {
@@ -1204,7 +1201,7 @@ impl Reader {
 			copy: copy.track.clone(),
 			handoff,
 			lease: Some(copy.lease.clone()),
-			out,
+			out: Some(out),
 			logical: None,
 			fetch: None,
 		})
@@ -2842,6 +2839,50 @@ mod test {
 		);
 	}
 
+	/// A group first read while only one copy existed is still backed by a later copy that
+	/// holds it, so a failure continues from that copy instead of waiting out the budget.
+	#[test]
+	#[cfg(feature = "async")]
+	fn a_group_read_from_a_single_route_keeps_its_backup() {
+		let routes = Producer::new();
+		let logical = logical(&routes);
+		let a = copy();
+		routes.serve(a.consume());
+		let mut sub = subscribe(&logical, Duration::from_secs(10));
+
+		let mut open = a.create_group(group::Info { sequence: 0 }).unwrap();
+		open.write_frame(ts(0), b"a".as_ref()).unwrap();
+		let mut reading = recv(&mut sub);
+		assert_eq!(read(&mut reading), Some(b"a".to_vec()));
+
+		// B holds the whole group, then its route fails, keeping it.
+		let b = copy();
+		let mut whole = b.create_group(group::Info { sequence: 0 }).unwrap();
+		whole.write_frame(ts(0), b"a".as_ref()).unwrap();
+		whole.write_frame(ts(1), b"b".as_ref()).unwrap();
+		whole.finish().unwrap();
+		routes.serve(b.consume());
+		assert!(
+			sub.poll_recv_group(&kio::Waiter::noop()).is_pending(),
+			"B replays only group 0"
+		);
+		b.abort(Error::SessionClosed).unwrap();
+
+		// C lacks the group; reading a later one from it lets go of what nothing needs.
+		let c = copy();
+		let mut later = c.create_group(group::Info { sequence: 1 }).unwrap();
+		later.write_frame(ts(2), b"c".as_ref()).unwrap();
+		later.finish().unwrap();
+		routes.serve(c.consume());
+		assert_eq!(recv(&mut sub).sequence, 1);
+		assert!(sub.poll_recv_group(&kio::Waiter::noop()).is_pending());
+
+		open.abort(Error::SessionClosed).unwrap();
+		a.abort(Error::SessionClosed).unwrap();
+		assert_eq!(read(&mut reading), Some(b"b".to_vec()));
+		assert_eq!(read(&mut reading), None);
+	}
+
 	#[test]
 	#[cfg(feature = "async")]
 	fn a_group_both_routes_deliver_is_handed_out_once() {