feat(net): resume JS tracks across same-epoch route swaps; keep a replaced route feeding until its successor delivers

doc/lib/js/net.md

@@ -36,16 +36,16 @@ broadcast.announce();
 - **The origin holds the broadcasts, not the connection.** Closing a session unannounces them and leaves them created for the next one. A broadcast is invisible, locally and remotely, until `announce()`. `dynamic(prefix, route)` claims a prefix and yields each requested path to accept or reject. `accept` throws on a broadcast a session delivered, since JS apps do not proxy; copy its tracks into a broadcast you produce instead.
 - **Requests can pin an epoch.** `consumer.request(path, { epoch })` resolves only through a route announcing that [publisher epoch](/concept/moq-lite#publisher-epochs) and reports `unroutable` otherwise. The resolved broadcast's `epoch` names the route it came through, pinned or not.
 - **Rank several CDNs.** Connect to each with its own `consume: origin.withPreference(Origin.Preference(rank))` (higher wins; 0 is neutral and a negative rank demotes), all on one origin. Requests use the preferred link that has a route and fail over to the next; see [multiple CDNs](/concept/moq-lite#multiple-cdns).
-- **Follow restarts.** `origin.request(path, { announced: true })` follows whichever publisher instance serves the path: its `active` changes to a fresh broadcast on a restart, clears on an end, and resolves again on the next start, so read it reactively and close the request when done. Instances are never stitched, so subscribe again to each new broadcast. A refusal ends the request. Add `epoch` to pin one instance; a plain `request(path)` keeps its accepted instance, can resolve blindly, and ends once that instance stops serving. To list what is live, iterate `announced(scope)`.
+- **Follow restarts.** `origin.request(path, { announced: true })` follows whichever publisher instance serves the path: its `active` changes to a fresh broadcast on a restart, clears on an end, and resolves again on the next start, so read it reactively and close the request when done. Instances are never stitched, so subscribe again to each new broadcast. Another route of the same epoch taking over (epochs travel only on `moq-lite-07-wip`) keeps `active` as is, and each track subscription resumes there from the first group it lacks. A refusal ends the request. Add `epoch` to pin one instance; a plain `request(path)` keeps its accepted instance, can resolve blindly, and ends once that instance stops serving. To list what is live, iterate `announced(scope)`.
 - **Discovery follows what you watch.** A `consume` origin asks the relay for announcements only under the prefixes its readers watch with `announced(scope)` or `request(path, { announced: true })`, stops asking about a prefix 2s after nothing at or beneath it is watched, and asks again on reconnect. A plain `request(path)` asks nothing and subscribes blind, but once it resolves through a discovered route it watches its path until closed, so it outlives the listing that found it. A relay that refuses an announce request turns that session's announced requests blind too.
 - **One connection per URL, unless you opt out.** `new Connection({ url })` pools and reconnects with backoff, which the elements use. Supplying your own transport options or origin selects a private loop. Receive estimates are on demand: observe `connection.probe` reactively or subscribe while you need them; `peek()` alone opens no PROBE stream.
-- **GOAWAY moves the session.** The connection dials the replacement at once while the old session keeps serving its groups, up to `goaway.handover` (default 10s, or the relay's deadline when that is sooner). New requests keep opening on the old session until the replacement's route outranks it, which is a restart without an epoch: an unpinned announced request follows the replacement, while a blind or epoch-pinned request stays on its accepted instance. An empty GOAWAY redials the same URL. A redirect stays on the same host unless `goaway.redirect` is `"follow"`.
+- **GOAWAY moves the session.** The connection dials the replacement at once while the old session keeps serving its groups, up to `goaway.handover` (default 10s, or the relay's deadline when that is sooner). New requests keep opening on the old session until the replacement's route outranks it, which is a restart without an epoch: an unpinned announced request follows the replacement, while a blind or epoch-pinned request stays on its accepted instance. With an epoch, which travels only when `moq-lite-07-wip` is offered, the replacement serves the same broadcast, so open subscriptions move across without missing a group. An empty GOAWAY redials the same URL. A redirect stays on the same host unless `goaway.redirect` is `"follow"`.
 - **One send estimate per connection.** `Bandwidth.Allocator` divides it by track priority, max-min fair within a tier. An idle track claims nothing. Publishers reserve against it so their targets sum to the estimate.
 - **One track per name.** Concurrent subscriptions share one request and producer. `createTrack` on a name with a request still queued answers it. The application owns sequence numbers: a track that replaces an ended one, created or requested, throws on an append that would reuse a sequence the name already wrote, so it writes its first group with an explicit sequence. Only a new broadcast starts over.
 - **Timing is per track, with no default.** A track that declares a `timescale` carries a `timestamp` on every frame; one without is [untimed](/concept/moq-lite#subscriptions) and its frames carry none. Omitting it means untimed, as in Rust. Nothing stamps a frame for you: the publisher passes the timestamp, and the `Timed` type (`{ value, at }`) carries one into the `@moq/json` and `@moq/flate` producers.
 - **Subscriber staleness is `maxDelay`.** It counts from a group's successor, by media time and by wall clock, whichever reaches it first. Passing the old `maxAge` key throws a `TypeError` naming `maxDelay`. Publisher retention is the separate `Track.Info.maxAge`.
 - **Hidden paths** stay out of discovery unless the announce request opts in. See [hidden broadcasts](/concept/moq-lite#hidden-broadcasts).
-- **Authorization is in band.** On moq-lite 07 (offered only when `moq-lite-07-wip` is listed in `webtransport.protocols`; the WebSocket fallback cannot offer it) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), `auth.grant` watches what the relay lets this side publish and subscribe to, and `auth.add(token)` presents another token without reconnecting. Publishing outside the grant closes the connection with `SessionCode.Unauthorized`, naming the path. On moq-lite, a subscription the grant stops covering resets with `StreamCode.Unauthorized` and the session stays up. On moq-transport, `auth.add(token)` keeps the token even without the extension, and a broadcast announced with the `Auth.Auth` it returns (`broadcast.announce({ auth })`) carries it on that announcement's `PUBLISH_NAMESPACE`; announcing again with another renews the request. An `Auth.Auth` belongs to the session that added it, so after a reconnect, add the token again and re-announce.
+- **Authorization is in band.** On moq-lite 07 (offered only when `moq-lite-07-wip` is listed in `webtransport.protocols` or `websocket.protocols`) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), `auth.grant` watches what the relay lets this side publish and subscribe to, and `auth.add(token)` presents another token without reconnecting. Publishing outside the grant closes the connection with `SessionCode.Unauthorized`, naming the path. On moq-lite, a subscription the grant stops covering resets with `StreamCode.Unauthorized` and the session stays up. On moq-transport, `auth.add(token)` keeps the token even without the extension, and a broadcast announced with the `Auth.Auth` it returns (`broadcast.announce({ auth })`) carries it on that announcement's `PUBLISH_NAMESPACE`; announcing again with another renews the request. An `Auth.Auth` belongs to the session that added it, so after a reconnect, add the token again and re-announce.
 - **A graceful close waits.** `await connection.close()` withdraws announcements and drains for up to one second. Application delivery requires explicit `moq-lite-07-wip`; older Lite versions wait only for transport ACKs. IETF media streams are not drained. `abort()` ends immediately.
 
 Examples:

js/justfile

@@ -47,7 +47,7 @@ test:
 test-changed list:
     python3 ../sh/js/workspace.py test '{{ list }}'
 
-# Run one net benchmark: broadcasts, forward, requests, reader, frames, track, interests, observed, varint, lite-varint, priority, or request-ids.
+# Run one net benchmark: broadcasts, forward, requests, reader, frames, track, resume, interests, observed, varint, lite-varint, priority, or request-ids.
 bench name:
     bun install --frozen-lockfile
     bun net/bench/{{ name }}.ts

js/net/bench/resume.ts

@@ -0,0 +1,125 @@
+/**
+ * Sweep readers and route swaps for groups read through a logical front, against the same
+ * groups read straight from the track.
+ *
+ * A front reads its serving route's copy directly, so with one route it should cost about what a
+ * direct read does, and each swap should leave nothing behind once its groups are read.
+ */
+
+import { Producer as BroadcastProducer } from "../src/broadcast.ts";
+import { Producer as GroupProducer } from "../src/group.ts";
+import { Front } from "../src/resume.ts";
+import type { Subscriber } from "../src/track.ts";
+
+const readerCounts = [1, 4, 16];
+const swapCounts = [0, 1, 16];
+const groups = 400;
+const reps = 7;
+// A reader through a front must not cost more per group as readers or swaps grow. The margin is
+// loose because the machine is noisy.
+const maxSlope = 3;
+const payload = new Uint8Array(80);
+let checksum = 0;
+
+// Let every woken reader and pending splice run.
+const flush = () => new Promise<void>((resolve) => setImmediate(resolve));
+
+function read(subscribers: Subscriber[]) {
+	for (const subscriber of subscribers) {
+		const group = subscriber.tryRecvGroup();
+		if (!group) throw new Error("group was not delivered");
+		const frame = group.tryReadFrame();
+		if (!frame) throw new Error("frame was not delivered");
+		checksum += frame.payload.byteLength;
+		group.close();
+		// Read until idle, as a reader parked on the next group does.
+		if (subscriber.tryRecvGroup()) throw new Error("group delivered twice");
+	}
+}
+
+/** Microseconds per group per reader, reading through a front after `swaps` route swaps, or directly. */
+async function measure(readers: number, swaps: number | undefined): Promise<number> {
+	let sequence = 0;
+	const sources: { broadcast: BroadcastProducer; write: () => void }[] = [];
+	const source = () => {
+		const broadcast = new BroadcastProducer();
+		const track = broadcast.createTrack("seq");
+		// Every route's copy holds the same groups under the same sequences.
+		const write = () => {
+			const group = new GroupProducer(sequence);
+			track.writeGroup(group);
+			group.writeFrame({ payload });
+			group.close();
+		};
+		sources.push({ broadcast, write });
+		return broadcast;
+	};
+	const publish = () => {
+		for (const { write } of sources) write();
+		sequence++;
+	};
+
+	let subscribers: Subscriber[];
+	let front: Front | undefined;
+	if (swaps === undefined) {
+		const broadcast = source().consume();
+		subscribers = Array.from({ length: readers }, () => broadcast.track("seq").subscribe());
+	} else {
+		front = new Front();
+		front.serve(source().consume());
+		const consumer = front.consumer;
+		subscribers = Array.from({ length: readers }, () => consumer.track("seq").subscribe());
+		for (let swap = 0; swap < swaps; swap++) {
+			publish();
+			read(subscribers);
+			// Only the serving route publishes from here, as a replaced one would stop.
+			sources.splice(0, sources.length - 1);
+			front.serve(source().consume());
+			await flush();
+		}
+	}
+	publish();
+	read(subscribers);
+	await flush();
+
+	const start = performance.now();
+	for (let index = 0; index < groups; index++) {
+		publish();
+		read(subscribers);
+	}
+	const elapsed = performance.now() - start;
+
+	for (const subscriber of subscribers) subscriber.close();
+	front?.consumer.close();
+	return (elapsed * 1000) / groups / readers;
+}
+
+// Warm the JIT so the first row isn't the slowest.
+await measure(4, 1);
+await measure(4, undefined);
+
+console.log("readers,swaps,direct_us,front_us");
+let baseline: number | undefined;
+for (const swaps of swapCounts) {
+	for (const readers of readerCounts) {
+		const direct: number[] = [];
+		const front: number[] = [];
+		// Interleaved, so a noisy stretch of the machine lands on both columns alike.
+		for (let rep = 0; rep < reps; rep++) {
+			direct.push(await measure(readers, undefined));
+			front.push(await measure(readers, swaps));
+		}
+		// The fastest run is the one least disturbed by the rest of the machine.
+		const directUs = Math.min(...direct);
+		const frontUs = Math.min(...front);
+		console.log(`${readers},${swaps},${directUs.toFixed(2)},${frontUs.toFixed(2)}`);
+
+		baseline ??= frontUs;
+		if (frontUs > baseline * maxSlope) {
+			throw new Error(
+				`${readers} readers after ${swaps} swaps cost ${frontUs.toFixed(2)} us/group, over ${maxSlope}x ${baseline.toFixed(2)}`,
+			);
+		}
+	}
+}
+if (checksum === 0) throw new Error("benchmark did no work");

js/net/src/connection/connect.test.ts

@@ -292,3 +292,56 @@ test("SETUP preserves a delayed session close code", async () => {
 		jest.useRealTimers();
 	}
 });
+
+// The WebSocket subprotocols one connect() attempt asks a server for.
+async function websocketOffer(props: Omit<ConnectProps, "url">): Promise<string[]> {
+	const original = globalThis.WebTransport;
+	// No WebTransport, so the WebSocket attempt goes at once.
+	globalThis.WebTransport = undefined as unknown as typeof WebTransport;
+	let offer: string[] = [];
+	const server = Bun.serve({
+		port: 0,
+		fetch(request) {
+			offer = (request.headers.get("sec-websocket-protocol") ?? "").split(",").map((protocol) => protocol.trim());
+			return new Response("no", { status: 400 });
+		},
+	});
+	const controller = new AbortController();
+	try {
+		const attempt = connect(new URL(`http://127.0.0.1:${server.port}/`), {
+			signal: controller.signal,
+			...props,
+		}).catch(() => undefined);
+		for (let i = 0; i < 100 && offer.length === 0; i++) await settle();
+		controller.abort();
+		await attempt;
+		return offer;
+	} finally {
+		globalThis.WebTransport = original;
+		server.stop(true);
+	}
+}
+
+test("the WebSocket fallback offers lite-07 only when asked", async () => {
+	const defaults = await websocketOffer({});
+	expect(defaults.some((protocol) => protocol.endsWith(ALPN_06))).toBe(true);
+	expect(defaults.some((protocol) => protocol.endsWith(ALPN_07_WIP))).toBe(false);
+
+	const asked = await websocketOffer({ websocket: { protocols: [ALPN_07_WIP, ALPN_06] } });
+	expect(asked[0]?.endsWith(ALPN_07_WIP)).toBe(true);
+	expect(asked.some((protocol) => protocol.endsWith(ALPN_05))).toBe(false);
+});
+
+test("the WebSocket fallback refuses an unsupported protocol", async () => {
+	const original = globalThis.WebTransport;
+	globalThis.WebTransport = undefined as unknown as typeof WebTransport;
+	try {
+		for (const protocol of ["moq-lite-99", "toString"]) {
+			await expect(connect(url, { websocket: { protocols: [protocol] } })).rejects.toThrow(
+				"unsupported WebSocket protocol",
+			);
+		}
+	} finally {
+		globalThis.WebTransport = original;
+	}
+});