fix(net): retain resume bounds across route copies

js/net/bench/resume.ts

@@ -137,6 +137,57 @@ for (const swaps of swapCounts) {
 }
 if (checksum === 0) throw new Error("benchmark did no work");
 
+// Two routes keep the same live group while taking turns. Sweep audience and flap
+// count: the cost per swap must not grow with redundant subscriptions to those routes.
+async function measureFlaps(readers: number, flaps: number): Promise<number> {
+	const front = new Front();
+	const sources = [new BroadcastProducer(), new BroadcastProducer()];
+	const handles = sources.map((source) => source.consume());
+	const tracks = sources.map((source) => source.createTrack("seq"));
+	const groups = tracks.map((track) => {
+		const group = track.appendGroup();
+		group.writeFrame({ payload });
+		return group;
+	});
+	front.serve(handles[0]);
+	const subscribers = Array.from({ length: readers }, () =>
+		front.consumer.track("seq").subscribe({ maxDelay: Milli(3_600_000) }),
+	);
+	const held = subscribers.map((subscriber) => {
+		const group = subscriber.tryRecvGroup();
+		if (!group) throw new Error("missing held group");
+		return group;
+	});
+	await flush();
+	const start = performance.now();
+	for (let flap = 0; flap < flaps; flap++) {
+		front.serve(handles[(flap + 1) % 2]);
+		await flush();
+		for (const subscriber of subscribers) {
+			if (subscriber.tryRecvGroup()) throw new Error("repeated held group");
+		}
+	}
+	const elapsed = performance.now() - start;
+	for (const group of held) group.close();
+	for (const subscriber of subscribers) subscriber.close();
+	front.consumer.close();
+	for (const group of groups) group.close();
+	for (const handle of handles) handle.close();
+	for (const source of sources) source.close();
+	await flush();
+	return (elapsed * 1000) / readers / flaps;
+}
+
+await measureFlaps(4, 16);
+console.log("readers,flaps,flap_us");
+for (const readers of readerCounts) {
+	for (const flaps of [1, 16, 64]) {
+		const samples: number[] = [];
+		for (let rep = 0; rep < reps; rep++) samples.push(await measureFlaps(readers, flaps));
+		console.log(`${readers},${flaps},${Math.min(...samples).toFixed(2)}`);
+	}
+}
+
 // An adopted copy may repeat a large prefix before it reaches unread content. Time readiness
 // separately from constructing the copies, sweeping both the replay and independent readers.
 async function measureReplay(readers: number, replayed: number): Promise<number> {

js/net/src/resume.test.ts

@@ -1,4 +1,4 @@
-import { expect, test } from "bun:test";
+import { expect, spyOn, test } from "bun:test";
 import { Manual, Milli, Nano } from "@moq/clock";
 import { Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts";
 import * as Epoch from "./epoch.ts";
@@ -104,6 +104,87 @@ async function setup() {
 	};
 }
 
+test("returning to a route reuses its subscription and held group", async () => {
+	const clock = new Manual();
+	const restore = clock.install();
+	const front = new Front();
+	const sources = [new BroadcastProducer(), new BroadcastProducer()];
+	const handles = sources.map((source) => source.consume());
+	const tracks = sources.map(copy);
+	const open = tracks[0].write(0, 1, false);
+	const mirrored = spyOn(open, "mirror");
+	front.serve(handles[0]);
+	const sub = front.consumer.track("seq").subscribe({ maxDelay: Milli(3_600_000) });
+	try {
+		await clock.advance(Nano.zero);
+		const held = sub.tryRecvGroup();
+		if (!held) throw new Error("expected the held group");
+		expect(dec.decode((await held.readFrame())?.payload)).toBe("0.0");
+		for (let round = 0; round < 32; round++) {
+			front.serve(handles[round % 2]);
+			await clock.advance(Nano.zero);
+			expect(sub.tryRecvGroup()).toBeUndefined();
+			expect(mirrored.mock.calls.length).toBe(1);
+		}
+		tracks[1].write(1);
+		expect(drain(sub)).toEqual([1]);
+		front.serve(handles[0]);
+		await clock.advance(Nano.zero);
+		open.writeFrame({ payload: enc.encode("0.1") });
+		expect(dec.decode((await held.readFrame())?.payload)).toBe("0.1");
+		tracks[0].write(2);
+		expect(drain(sub)).toEqual([2]);
+		held.close();
+	} finally {
+		mirrored.mockRestore();
+		open.close();
+		sub.close();
+		front.consumer.close();
+		for (const handle of handles) handle.close();
+		for (const source of sources) source.close();
+		restore();
+	}
+});
+
+test("a returning route resumes where a fresh copy would", async () => {
+	const clock = new Manual();
+	const restore = clock.install();
+	const front = new Front();
+	const sources = [new BroadcastProducer(), new BroadcastProducer()];
+	const handles = sources.map((source) => source.consume());
+	const [ta, tb] = sources.map(copy);
+	const open = ta.write(0, 1, false);
+	const mirrored = spyOn(open, "mirror");
+	front.serve(handles[0]);
+	const sub = front.consumer.track("seq").subscribe({ maxDelay: Milli(3_600_000) });
+	try {
+		await clock.advance(Nano.zero);
+		const held = sub.tryRecvGroup();
+		if (!held) throw new Error("expected the held group");
+		front.serve(handles[1]);
+		await clock.advance(Nano.zero);
+		tb.write(10);
+		tb.write(20);
+		expect(drain(sub)).toEqual([10, 20]);
+		ta.write(15);
+		front.serve(handles[0]);
+		await clock.advance(Nano.zero);
+		expect(mirrored.mock.calls.length).toBe(1);
+		expect(drain(sub)).toEqual([]);
+		ta.write(26);
+		expect(drain(sub)).toEqual([26]);
+		held.close();
+	} finally {
+		mirrored.mockRestore();
+		open.close();
+		sub.close();
+		front.consumer.close();
+		for (const handle of handles) handle.close();
+		for (const source of sources) source.close();
+		restore();
+	}
+});
+
 test("a routed source start follows the serving copy and its subscriber floor", async () => {
 	const a = new BroadcastProducer();
 	const b = new BroadcastProducer();

js/net/src/resume.ts

@@ -347,6 +347,29 @@ class Reader implements ResumeSource {
 		// The replaced copies keep delivering until the new one delivers its first group: a
 		// group they get meanwhile may be one the new route never has. See #observe.
 		for (const copy of this.#copies) copy.replaced = true;
+		// Cloned handles share closed, so a returning route reuses its subscription and
+		// leases rather than replaying its cache and retaining another copy per flap.
+		const index = this.#copies.findIndex((copy) => copy.source.from.closed === source.from.closed);
+		if (index !== -1) {
+			const [copy] = this.#copies.splice(index, 1);
+			source.retain();
+			copy.source.release();
+			copy.source = source;
+			copy.generation = route.generation;
+			copy.replaced = false;
+			copy.until = undefined;
+			copy.since = undefined;
+			copy.done = undefined;
+			copy.datagramsDone = undefined;
+			// Resume where a fresh copy would: a group it buffered while replaced, below the newest
+			// another route handed out, is not handed out.
+			if (this.#newest !== undefined) copy.floor = Math.max(copy.floor ?? 0, this.#newest);
+			this.#update(copy);
+			this.#bound(copy);
+			this.#copies.push(copy);
+			this.wake();
+			return;
+		}
 		const floor = this.#newest;
 		this.#add({
 			generation: route.generation,

quest/m1/README.md

@@ -128,7 +128,6 @@ tracks, so per-group stats stop costing cluster announces.
 - [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 holds a declared end](/quest/m1/resume-rust-end.md) - a resumed Rust track keeps a declared end across copies, dedupes flapping copies, and judges retention inside the reader's window
 - [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

quest/m1/resume-compress.md

@@ -25,7 +25,3 @@ not `compress`. JS's `compatible()` (`js/net/src/resume.ts`) compares all four.
   B's groups decoded under A's metadata. Matching compression keeps resuming.
 
 Public API: behavior only. Wire: none.
-
-## Related
-
-- [Rust resume holds a declared end](/quest/m1/resume-rust-end.md) - the other Rust copy-compatibility check

quest/m1/resume-handed-out.md

@@ -26,11 +26,4 @@ read group 1; A's partial group 0 fails. B was dropped, so group 0 stalls.
   single copy, so the per-group record's cost shows up.
 - Regression test on a mocked clock for the reproduction above.
 
-Lands after [#6224](https://github.com/moq-dev/moq/pull/6224), which edits the
-same reader, so it does not widen a PR under review.
-
 Public API: behavior only. Wire: none.
-
-## Required
-
-- [Rust resume holds a declared end](/quest/m1/resume-rust-end.md) - lands first; it edits the same reader

rs/moq-net/benches/origin.rs

@@ -1310,7 +1310,7 @@ fn bench_copy_walk(c: &mut Criterion) {
 
 /// `(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); 15] = [
+const RESUME_OVERLAP: [(usize, usize, usize); 17] = [
 	(1, 1, 1),
 	(1, 64, 1),
 	(32, 64, 1),
@@ -1326,6 +1326,8 @@ const RESUME_OVERLAP: [(usize, usize, usize); 15] = [
 	(1, 64, 8),
 	(1, 64, 32),
 	(32, 64, 8),
+	(32, 64, 32),
+	(256, 64, 8),
 ];
 
 /// A track that swapped through its routes while its readers' local cap held every

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

@@ -2761,7 +2761,7 @@ impl Consumer {
 		// A front's reader may continue the group from another route's copy when it reads.
 		if self
 			.recover
-			.as_ref()
+			.as_mut()
 			.is_some_and(|recover| recover.wants(Some(&aborted), waiter))
 		{
 			return None;