fix(net): hand each group sequence out once per reader

quest/m0/broadcast-epoch/README.md

@@ -123,7 +123,6 @@ End+Start. The child quests own their focused regressions.
 
 ## Required
 
-- [Resume duplicates](/quest/m0/broadcast-epoch/resume-duplicates.md) - a delayed group is delivered once even beyond the old 1024-entry history
 - [Catalog references](/quest/m0/broadcast-epoch/catalog-references.md) - players and exporters pin siblings consistently, with optional reference epochs and documented legacy best effort
 
 ## Related

quest/m0/parked-drain.md

@@ -39,7 +39,3 @@ against capped subscribers per track: the parked axis flattens, and a
 per-abort cost that grows with subscribers shows up as a slope.
 
 Public API: none. Wire: none.
-
-## Required
-
-- [Resume duplicates](/quest/m0/broadcast-epoch/resume-duplicates.md) - reshapes `track::Subscriber`'s delivery record in the same code

quest/m1/js-group-handover.md

@@ -29,6 +29,17 @@ missing frame and byte. It does not copy through the earlier single-writer
 pump or immediately cancel every old copy. Decided in the 2026-10-10 audit:
 mirror the current identity, reader, recovery and lifetime contract, including
 the late-duplicate correction, rather than porting the obsolete pump.
+Decided 2026-10-13: every track reader, direct or resumed, hands out each group
+sequence at most once, including a group aborted and re-created and one a
+second route delivers, so no consumer ever dedupes itself. The record is a
+per-reader interval set of delivered sequences on the track subscriber, shared
+by direct and resumed reads, with no fixed-size history. Once a group is past
+the reader's budget (max delay, clamped by max age), the reader stops expecting
+every group below it and counts them as delivered, so only gaps within the
+budget cost memory. A genuinely unseen late group within the budget is still
+delivered. Group timestamps on a track never rewind; which late groups are
+delivered after a rewind is unspecified, so the floor is judged by the
+delivered group above a gap rather than the gap itself.
 Things to settle along the way:
 
 - Whether the request's `active` broadcast stays the same object across a swap,
@@ -64,7 +75,6 @@ subscriptions across a swap. Report it in the PR.
 ## Required
 
 - [One max_age meaning](/quest/m1/cache-max-age.md) - the rule that gives up a resumed group no route continues
-- [Resume duplicates](/quest/m0/broadcast-epoch/resume-duplicates.md) - the corrected once-only delivery contract for reordered copies
 
 ## Related
 

quest/m1/resume-reorder.md

@@ -34,5 +34,4 @@ Public API: behavior only. Wire: none.
 
 ## Related
 
-- [Resume duplicates](/quest/m0/broadcast-epoch/resume-duplicates.md) - retain once-only delivery without excluding genuinely unseen late groups
 - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - explicit terminal dispositions end recovery without waiting for a newer group

rs/moq-net/benches/origin.rs

@@ -19,6 +19,8 @@
 //!
 //! 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
+//! route delivered, and the routes each reader still holds a copy of.
 //!
 //! Run with `cargo bench -p moq-net --bench origin`.
 
@@ -1125,6 +1127,129 @@ fn bench_copy_walk(c: &mut Criterion) {
 	group.finish();
 }
 
+/// `(readers, retained, routes)` for [`bench_resume_overlap`]: readers of one track,
+/// the groups every route holds, and the routes it swapped through.
+const RESUME_OVERLAP: [(usize, usize, usize); 9] = [
+	(1, 1, 2),
+	(1, 64, 2),
+	(1, 1024, 2),
+	(32, 64, 2),
+	(256, 64, 2),
+	(32, 1024, 2),
+	(1, 64, 8),
+	(1, 64, 32),
+	(32, 64, 8),
+];
+
+/// A track that swapped through its routes while its readers' local cap held every
+/// group back, so each reader keeps a copy per route and every copy holds the same
+/// groups, each replacement in reverse arrival order.
+struct ResumeOverlap {
+	_producer: origin::Producer,
+	_generations: Vec<(broadcast::Producer, track::Producer)>,
+	readers: Vec<track::Subscriber>,
+}
+
+async fn resume_overlap(readers: usize, retained: usize, routes: usize) -> ResumeOverlap {
+	let (producer, driver) = origin::Producer::new(origin::Config::default());
+	tokio::spawn(moq_net::time::run(driver));
+	let epoch = Epoch::mint();
+	let generation = |order: &mut dyn Iterator<Item = usize>| {
+		let broadcast = producer.create_broadcast(COPY_PATH).unwrap();
+		let track = broadcast.create_track("0", timed()).unwrap();
+		for sequence in order {
+			let mut group = track
+				.create_group(moq_net::group::Info {
+					sequence: sequence as u64,
+				})
+				.unwrap();
+			group.write_frame(Timestamp::ZERO, b"f".as_ref()).unwrap();
+			group.finish().unwrap();
+		}
+		broadcast
+			.announce(origin::Route::default().with_epoch(epoch.clone()))
+			.unwrap();
+		(broadcast, track)
+	};
+	let mut generations = vec![generation(&mut (0..retained))];
+	let resolved = tokio::time::timeout(COPY_WAIT, producer.consume().request_broadcast(COPY_PATH, None))
+		.await
+		.expect("front did not resolve")
+		.expect("front refused");
+	let track = resolved.track("0").unwrap();
+	// A budget covering every group, so none is written off as late.
+	let subscription = track::Subscription::default().with_max_delay(Duration::from_secs(3600));
+	let mut subscribers = Vec::with_capacity(readers);
+	for _ in 0..readers {
+		let mut subscriber = tokio::time::timeout(COPY_WAIT, track.subscribe(subscription.clone()))
+			.await
+			.expect("track did not splice")
+			.expect("subscribe failed");
+		// Held back until the drain, so every route's copy is kept for its unread groups.
+		subscriber.set_groups(..0);
+		subscribers.push(subscriber);
+	}
+	let waiter = kio::Waiter::noop();
+	for _ in 1..routes {
+		generations.push(generation(&mut (0..retained).rev()));
+		let (_, latest) = generations.last().unwrap();
+		tokio::time::timeout(COPY_WAIT, latest.demand().used())
+			.await
+			.expect("route swap did not splice")
+			.expect("replacement track closed");
+		// Each reader follows the swap, subscribing to the new copy.
+		for subscriber in &mut subscribers {
+			assert!(
+				subscriber.poll_recv_group(&waiter).is_pending(),
+				"the cap holds every group"
+			);
+		}
+	}
+	for subscriber in &mut subscribers {
+		subscriber.set_groups(..);
+	}
+	ResumeOverlap {
+		_producer: producer,
+		_generations: generations,
+		readers: subscribers,
+	}
+}
+
+/// Readers draining a track whose routes all delivered every group: each group is
+/// handed out once, and every other copy's duplicate is skipped against what the reader
+/// handed out. The slopes over retained groups and routes are that skipping's cost.
+fn bench_resume_overlap(c: &mut Criterion) {
+	let mut group = c.benchmark_group("origin/resume_overlap");
+	group.sample_size(10);
+	for (readers, retained, routes) in RESUME_OVERLAP {
+		group.throughput(Throughput::Elements((readers * retained) as u64));
+		group.bench_function(
+			BenchmarkId::from_parameter(format!("{readers}r_{retained}g_{routes}c")),
+			|b| {
+				let runtime = tokio::runtime::Builder::new_current_thread()
+					.enable_all()
+					.build()
+					.unwrap();
+				let waiter = kio::Waiter::noop();
+				b.iter_batched_ref(
+					|| runtime.block_on(resume_overlap(readers, retained, routes)),
+					|rig| {
+						for reader in &mut rig.readers {
+							let mut delivered = 0;
+							while let Poll::Ready(Ok(Some(_))) = reader.poll_recv_group(&waiter) {
+								delivered += 1;
+							}
+							assert_eq!(delivered, retained, "every group once");
+						}
+					},
+					BatchSize::PerIteration,
+				);
+			},
+		);
+	}
+	group.finish();
+}
+
 criterion_group!(
 	benches,
 	bench_announce,
@@ -1145,6 +1270,7 @@ criterion_group!(
 	bench_narrow,
 	bench_relay,
 	bench_parked,
-	bench_copy_walk
+	bench_copy_walk,
+	bench_resume_overlap
 );
 criterion_main!(benches);