test(net): cover lagging replicas on epoch wires

js/net/src/resume.test.ts

@@ -328,6 +328,145 @@ test("a resumed fork shares an on-demand track request", async () => {
 	expect(await extra).toBeUndefined();
 });
 
+test("a lagging same-epoch replica resumes from one publisher's bytes exactly once", async () => {
+	const { clock, restore } = manual();
+	const publisher = new Producer();
+	const original = publisher.createBroadcast(path);
+	original.announce({ epoch: EPOCH });
+	const source = original.createTrack("seq");
+	const origin = new Producer();
+
+	// Both replicas forward this one source. Only the incumbent pauses mid-group.
+	function replica(lagging: boolean) {
+		const broadcast = new BroadcastProducer();
+		const output = broadcast.createTrack("seq");
+		const input = source.subscribe({ maxDelay: Milli(60_000) });
+		const release = Promise.withResolvers<void>();
+		let stopped = false;
+		const done = (async () => {
+			try {
+				for (;;) {
+					const group = await input.recvGroup();
+					if (!group) return;
+					const copy = new GroupProducer(group.sequence);
+					output.writeGroup(copy);
+					try {
+						for (let index = 0; ; index++) {
+							const frame = await group.readFrame();
+							if (!frame) break;
+							if (lagging && group.sequence === 1 && index === 2) await release.promise;
+							copy.writeFrame(frame);
+						}
+						copy.close();
+					} finally {
+						group.close();
+					}
+				}
+			} catch (err) {
+				if (stopped) return;
+				output.close(err instanceof Error ? err : new Error(String(err)));
+				broadcast.close();
+				return err;
+			}
+		})();
+		return {
+			broadcast,
+			done,
+			stop(err?: Error) {
+				stopped = true;
+				input.close();
+				release.resolve();
+				output.close(err);
+				broadcast.close();
+			},
+		};
+	}
+
+	const a = replica(true);
+	const b = replica(false);
+	const disposeA = serve(origin, a.broadcast, 5n);
+	const request = origin.request(path);
+	const standby = b.broadcast.consume();
+	const warm = standby.track("seq").subscribe({ maxDelay: Milli(60_000) });
+	let sub: track.Subscriber | undefined;
+	let disposeB: (() => void) | undefined;
+	try {
+		await clock.advance(Nano(0));
+		const active = request.active.peek();
+		if (!active) throw new Error("expected the incumbent route");
+		sub = active.track("seq").subscribe({ maxDelay: Milli(60_000) });
+		const first = source.appendGroup();
+		for (let frame = 0; frame < 4; frame++) first.writeFrame({ payload: enc.encode(`0.${frame}`) });
+		first.close();
+		const seen: string[] = [];
+		const initial = await sub.recvGroup();
+		if (!initial) throw new Error("expected group 0");
+		expect(initial.sequence).toBe(0);
+		for (let frame = 0; frame < 4; frame++) seen.push(dec.decode((await initial.readFrame())?.payload));
+		expect(await initial.readFrame()).toBeUndefined();
+		initial.close();
+
+		const open = source.appendGroup();
+		for (let frame = 0; frame < 2; frame++) open.writeFrame({ payload: enc.encode(`1.${frame}`) });
+		const interrupted = await sub.recvGroup();
+		if (!interrupted) throw new Error("expected group 1");
+		expect(interrupted.sequence).toBe(1);
+		for (let frame = 0; frame < 2; frame++) seen.push(dec.decode((await interrupted.readFrame())?.payload));
+		for (let frame = 2; frame < 4; frame++) open.writeFrame({ payload: enc.encode(`1.${frame}`) });
+		open.close();
+		const last = source.appendGroup();
+		for (let frame = 0; frame < 4; frame++) last.writeFrame({ payload: enc.encode(`2.${frame}`) });
+		last.close();
+
+		// The standby holds the complete original while the incumbent cannot pass 1.1.
+		for (let sequence = 0; sequence < 3; sequence++) {
+			const group = await warm.recvGroup();
+			if (!group) throw new Error("expected a warmed group");
+			expect(group.sequence).toBe(sequence);
+			for (let frame = 0; frame < 4; frame++) {
+				expect(dec.decode((await group.readFrame())?.payload)).toBe(`${sequence}.${frame}`);
+			}
+			expect(await group.readFrame()).toBeUndefined();
+			group.close();
+		}
+		disposeB = serve(origin, b.broadcast, 9n);
+		await clock.advance(Nano(0));
+		expect(request.active.peek()).toBe(active);
+		disposeA();
+		a.stop(new StreamError(StreamCode.SessionClosed));
+		await clock.advance(Nano(0));
+		for (let frame = 2; frame < 4; frame++) seen.push(dec.decode((await interrupted.readFrame())?.payload));
+		expect(await interrupted.readFrame()).toBeUndefined();
+		interrupted.close();
+		const successor = await sub.recvGroup();
+		if (!successor) throw new Error("expected group 2");
+		expect(successor.sequence).toBe(2);
+		for (let frame = 0; frame < 4; frame++) seen.push(dec.decode((await successor.readFrame())?.payload));
+		expect(await successor.readFrame()).toBeUndefined();
+		successor.close();
+		expect(seen).toEqual(Array.from({ length: 12 }, (_, index) => `${Math.floor(index / 4)}.${index % 4}`));
+		await clock.advance(Nano(0));
+		expect(sub.tryRecvGroup()).toBeUndefined();
+		// The source is still open, so the resumed subscription must stay live.
+		expect(sub.closed.peek()).toBeUndefined();
+	} finally {
+		sub?.close();
+		warm.close();
+		standby.close();
+		request.close();
+		disposeA();
+		disposeB?.();
+		a.stop();
+		b.stop();
+		original.close();
+		publisher.close();
+		origin.close();
+		const outcomes = await Promise.all([a.done, b.done]);
+		restore();
+		expect(outcomes).toEqual([undefined, undefined]);
+	}
+});
+
 test("a subscription reads on across a same-epoch route swap, each group once", async () => {
 	const { a, b, active, request, swap, close } = await setup();
 	const ta = copy(a);

quest/m1/README.md

@@ -125,7 +125,6 @@ tracks, so per-group stats stop costing cluster announces.
 - [JS release preflight](/quest/m1/js-publish-preflight.md) - `release.ts` refuses a package whose `@moq/*` dependency is not on npm, so no dependent publishes uninstallable
 - [Stream sockets disable Nagle](/quest/m1/tcp-nodelay.md) - every qmux TCP, TLS, and WebSocket socket, accepted or dialed, sets `TCP_NODELAY`
 - [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

rs/moq-net/tests/integration/route_change.rs

@@ -520,7 +520,7 @@ pinned_tests! {
 /// partway through group 1: `A` has handed on (1, 0) and (1, 1) while `P` has already
 /// moved on to group 2. `B` (warmed by another reader on `W`) holds everything, so `R`
 /// resumes there from (1, 2).
-async fn lagging_route_dies(version: Version) -> mpsc::UnboundedReceiver<Delivery> {
+async fn lagging_route_dies(version: Version) {
 	async fn lagged(version: Version, from: &origin::Producer, to: &origin::Producer, lag: Duration) -> MockPair {
 		let mut options = MockConnectOptions::new(version);
 		options.server_publish = Some(from.consume());
@@ -580,27 +580,46 @@ async fn lagging_route_dies(version: Version) -> mpsc::UnboundedReceiver<Deliver
 	write(&mut group, 1, 3);
 	group.finish().unwrap();
 	let mut group = track.append_group().unwrap();
-	write(&mut group, 2, 0);
+	for frame in 0..FRAMES {
+		write(&mut group, 2, frame);
+	}
+	group.finish().unwrap();
 
 	a_r.server.abort(Error::Cancel);
 	a_r.client.abort(Error::Cancel);
 	settle().await;
-	moq_clock::sim::spawn(async move {
-		let _keep = (broadcast, track, group, p_a, p_b, b_w, b_r, warm, a_r, p, a, b, r, w);
-		std::future::pending::<()>().await
-	});
-	rx
+	for (sequence, frame) in (2..FRAMES)
+		.map(|frame| (1, frame))
+		.chain((0..FRAMES).map(|frame| (2, frame)))
+	{
+		let timestamp = Timestamp::from_millis(1_000 + sequence * 100 + frame).unwrap();
+		assert_eq!(
+			next_timed(&mut rx).await,
+			(sequence, Some(timestamp), payload(sequence, frame)),
+			"{version}"
+		);
+	}
+	settle().await;
+	assert!(
+		matches!(rx.try_recv(), Err(mpsc::TryRecvError::Empty)),
+		"{version}: trailing delivery, or the subscription ended"
+	);
+	let _keep = (broadcast, track, group, p_a, p_b, b_w, b_r, warm, a_r, p, a, b, r, w);
 }
 
-/// Lite carries the resume point upstream, so the rest of the group and the next one
-/// arrive in order.
 #[moq_clock::test]
 async fn lite_resumes_after_a_lagging_route_dies() {
-	let version: Version = "moq-lite-07-wip".parse().unwrap();
-	let mut rx = lagging_route_dies(version).await;
-	assert_eq!(next(&mut rx).await, (1, payload(1, 2)));
-	assert_eq!(next(&mut rx).await, (1, payload(1, 3)));
-	assert_eq!(next(&mut rx).await, (2, payload(2, 0)));
+	lagging_route_dies("moq-lite-07-wip".parse().unwrap()).await;
+}
+
+#[moq_clock::test]
+async fn ietf_17_resumes_after_a_lagging_route_dies() {
+	lagging_route_dies("moq-transport-17".parse().unwrap()).await;
+}
+
+#[moq_clock::test]
+async fn ietf_22_resumes_after_a_lagging_route_dies() {
+	lagging_route_dies("moq-transport-22".parse().unwrap()).await;
 }
 
 /// `P` publishes through the relay `O`, which `M1` and `M2` both pull from, and `E`