fix(net): refuse new requests from retired sources

doc/concept/moq-lite.md

@@ -87,6 +87,8 @@ authorization and hidden names work exactly as without one.
   winning the path, is announced as a restart. Subscriptions already open stay
   on the old instance until dropped or its route goes, while new requests get
   the new one, so a player follows the restart by requesting the path again.
+  Held handles can still read their old cached content, but cannot open new
+  epochless requests after retirement. Request the path again for new content.
   That is how a restarted encoder, whose group numbers start over, takes the
   name instead of stalling viewers on the old sequence. A relay drops its copy
   of the old instance, so a new subscriber never sees its cached groups.

js/net/bench/broadcasts.ts

@@ -327,3 +327,41 @@ for (const routeCount of routeCounts) {
 		source.close();
 	}
 }
+
+// Retiring a covering route cancels its pending descendants, not unrelated consumers.
+console.log("retirement,routes,consumers,retire_us,refused");
+for (const routeCount of routeCounts) {
+	for (const consumerCount of observerCounts) {
+		let elapsed = 0;
+		let refused = 0;
+		for (let round = 0; round < 256; round++) {
+			const cache = new BroadcastCache();
+			const source = new BroadcastProducer();
+			const owners = Array.from({ length: routeCount }, () => ({}));
+			const held = [];
+			for (let route = 0; route < routeCount; route++) {
+				const prefix = Path.from(`route/${route}`);
+				cache.announce(prefix, owners[route], Route.default);
+				for (let consumer = 0; consumer < (route === 0 ? consumerCount : 1); consumer++) {
+					const path = Path.from(`${prefix}/job/${consumer}`);
+					const instance = cache.instance(path);
+					if (route === 0)
+						instance.validity?.signal.addEventListener(
+							"abort",
+							() => {
+								refused++;
+							},
+							{ once: true },
+						);
+					held.push(cache.insert(path, instance, source.consume()));
+				}
+			}
+			const start = performance.now();
+			cache.withdraw(Path.from("route/0"), owners[0]);
+			elapsed += performance.now() - start;
+			for (const consumer of held) consumer.close();
+			source.close();
+		}
+		console.log(`retirement,${routeCount},${consumerCount},${((elapsed * 1000) / 256).toFixed(2)},${refused}`);
+	}
+}

js/net/src/consume.test.ts

@@ -99,3 +99,75 @@ test.each([false, true])("route alternation reuses every held instance (epoch: %
 	a.close();
 	b.close();
 });
+
+test("request validity cancels one selection period and permits the held fallback again", () => {
+	const cache = new BroadcastCache();
+	const path = Path.from("pool/job");
+	const broad = Path.from("pool");
+	const a = {};
+	const b = {};
+	cache.announce(broad, a, Route.default);
+	const sourceA = new Producer();
+	const instanceA = cache.instance(path);
+	const heldA = cache.insert(path, instanceA, sourceA.consume());
+	const first = instanceA.validity?.signal;
+	expect(first?.aborted).toBe(false);
+	cache.announce(path, b, Route.default);
+	expect(first?.aborted).toBe(true);
+	const sourceB = new Producer();
+	const instanceB = cache.instance(path);
+	const heldB = cache.insert(path, instanceB, sourceB.consume());
+	expect(instanceB.validity?.signal.aborted).toBe(false);
+	cache.withdraw(path, b);
+	expect(instanceB.validity?.signal.aborted).toBe(true);
+	expect(instanceA.validity?.signal).not.toBe(first);
+	expect(instanceA.validity?.signal.aborted).toBe(false);
+	expect(first?.aborted).toBe(true);
+	expect(heldA.closed.peek()).toBeUndefined();
+	expect(heldB.closed.peek()).toBeUndefined();
+	heldA.close();
+	heldB.close();
+	sourceA.close();
+	sourceB.close();
+});
+
+test("request validity follows the selected interest and ignores unrelated withdrawals", () => {
+	const cache = new BroadcastCache();
+	const path = Path.from("pool/job");
+	const broad = {};
+	const first = {};
+	const second = {};
+	cache.announce(Path.from("pool"), broad, Route.default);
+	cache.announce(path, first, Route.default);
+	cache.announce(path, second, Route.default);
+	const source = new Producer();
+	const instance = cache.instance(path);
+	const held = cache.insert(path, instance, source.consume());
+	const signal = instance.validity?.signal;
+	cache.withdraw(Path.from("pool"), broad);
+	cache.withdraw(path, second);
+	cache.announce(path, first, { ...Route.default, cost: 1n });
+	expect(instance.validity?.signal).toBe(signal);
+	expect(signal?.aborted).toBe(false);
+	cache.withdraw(path, first);
+	expect(signal?.aborted).toBe(true);
+	held.close();
+	source.close();
+});
+
+test("a blind selection retires on discovery while explicit epochs need no path gate", () => {
+	const cache = new BroadcastCache();
+	const path = Path.from("pool/job");
+	const blind = cache.instance(path);
+	const source = new Producer();
+	const held = cache.insert(path, blind, source.consume());
+	const owner = {};
+	cache.announce(Path.from("pool"), owner, { ...Route.default, epoch: Epoch.mint() });
+	expect(blind.validity?.signal.aborted).toBe(true);
+	expect(cache.instance(path).validity).toBeUndefined();
+	cache.withdraw(Path.from("pool"), owner);
+	expect(blind.validity?.signal.aborted).toBe(true);
+	expect(cache.instance(path).identity).not.toBe(blind.identity);
+	held.close();
+	source.close();
+});

js/net/src/consume.ts

@@ -3,19 +3,39 @@ import type * as Epoch from "./epoch.ts";
 import type { Route } from "./hop.ts";
 import { compareRouteCandidates, coveringPrefixes } from "./internal.ts";
 import type * as Path from "./path.ts";
-import type { Instance } from "./wire.ts";
+import { type Instance, sameInstance } from "./wire.ts";
 
-class Blind implements Instance {
+class Validity {
+	#current = new AbortController();
+
+	get signal(): AbortSignal {
+		return this.#current.signal;
+	}
+
+	update(allowed: boolean): void {
+		if (allowed) {
+			if (this.#current.signal.aborted) this.#current = new AbortController();
+		} else if (!this.#current.signal.aborted) {
+			this.#current.abort(new Error("broadcast instance retired"));
+		}
+	}
+}
+
+type Selection = Instance & { readonly validity?: Validity };
+
+class Blind implements Selection {
 	readonly identity = {};
 	readonly route = {};
+	readonly validity = new Validity();
 }
 
 /** Consumed broadcasts shared only within the selected serving instance. @internal */
 export class BroadcastCache {
-	#cache = new Map<Path.Valid, Map<Epoch.Valid | object, broadcast.Consumer>>();
+	#cache = new Map<Path.Valid, Map<Epoch.Valid | object, { consumer: broadcast.Consumer; selection: Selection }>>();
 	#announced = new Map<Path.Valid, Map<object, { identity: object; route: Route }>>();
 	#blind = new Map<Path.Valid, Blind>();
 	#blindPrefixes = new Map<Path.Valid, Set<Path.Valid>>();
+	#paths = new Map<Path.Valid, Set<Path.Valid>>();
 
 	/** Record this interest's live advertisement, preserving identity across route updates. */
 	announce(path: Path.Valid, owner: object, route: Route, restart = false): void {
@@ -28,44 +48,66 @@ export class BroadcastCache {
 		}
 		const identity = (!restart && entries.get(owner)?.identity) || {};
 		entries.set(owner, { identity, route });
+		this.#update(path);
 	}
 
 	/** Withdraw this interest's advertisement without disturbing another interest's routes. */
 	withdraw(path: Path.Valid, owner: object): void {
 		const entries = this.#announced.get(path);
 		entries?.delete(owner);
 		if (entries?.size === 0) this.#announced.delete(path);
+		this.#update(path);
 	}
 
 	/** Resolve the same most-specific-prefix and route preference as the origin. */
-	instance(path: Path.Valid): Instance {
+	instance(path: Path.Valid): Selection {
 		for (const prefix of coveringPrefixes(path)) {
 			const entries = this.#announced.get(prefix);
 			if (!entries) continue;
 			let best: { identity: object; route: Route } | undefined;
 			for (const entry of entries.values()) {
 				if (!best || compareRouteCandidates(path, entry.route, best.route) < 0) best = entry;
 			}
-			if (best) return best;
+			if (best) {
+				const cached = this.#cache.get(path)?.get(best.route.epoch ?? best.identity);
+				return {
+					...best,
+					validity:
+						best.route.epoch === undefined
+							? ((cached?.consumer.closed.peek() === undefined
+									? cached?.selection.validity
+									: undefined) ?? new Validity())
+							: undefined,
+				};
+			}
 		}
 		return this.#blind.get(path) ?? new Blind();
 	}
 
 	/** Clone the live cached consumer only when it serves the selected instance. */
 	get(path: Path.Valid, instance: Instance): broadcast.Consumer | undefined {
 		const cached = this.#cache.get(path)?.get(instance.route.epoch ?? instance.identity);
-		return cached?.closed.peek() === undefined ? cached?.clone() : undefined;
+		return cached?.consumer.closed.peek() === undefined ? cached?.consumer.clone() : undefined;
 	}
 
 	/** Cache each live instance so a temporary route winner never displaces a held fallback. */
-	insert(path: Path.Valid, instance: Instance, consumer: broadcast.Consumer): broadcast.Consumer {
+	insert(path: Path.Valid, instance: Selection, consumer: broadcast.Consumer): broadcast.Consumer {
 		let entries = this.#cache.get(path);
 		if (!entries) {
 			entries = new Map();
 			this.#cache.set(path, entries);
 		}
 		const key = instance.route.epoch ?? instance.identity;
-		entries.set(key, consumer);
+		const entry = { consumer, selection: instance };
+		entries.set(key, entry);
+		for (const prefix of coveringPrefixes(path)) {
+			let paths = this.#paths.get(prefix);
+			if (!paths) {
+				paths = new Set();
+				this.#paths.set(prefix, paths);
+			}
+			paths.add(path);
+		}
 		if (instance instanceof Blind) {
 			this.#blind.set(path, instance);
 			for (const prefix of coveringPrefixes(path)) {
@@ -78,9 +120,10 @@ export class BroadcastCache {
 			}
 		}
 		void consumer.closed.then(() => {
-			if (entries.get(key) !== consumer) return;
+			instance.validity?.update(false);
+			if (entries.get(key) !== entry) return;
 			entries.delete(key);
-			if (entries.size === 0 && this.#cache.get(path) === entries) this.#cache.delete(path);
+			if (entries.size === 0 && this.#cache.get(path) === entries) this.#forgetPath(path);
 			if (this.#blind.get(path) === instance) this.#forgetBlind(path);
 		});
 		return consumer;
@@ -89,14 +132,33 @@ export class BroadcastCache {
 	#forgetBlind(path: Path.Valid): void {
 		const blind = this.#blind.get(path);
 		if (!blind) return;
+		blind.validity.update(false);
 		this.#blind.delete(path);
 		const entries = this.#cache.get(path);
 		entries?.delete(blind.identity);
-		if (entries?.size === 0) this.#cache.delete(path);
+		if (entries?.size === 0) this.#forgetPath(path);
 		for (const prefix of coveringPrefixes(path)) {
 			const paths = this.#blindPrefixes.get(prefix);
 			paths?.delete(path);
 			if (paths?.size === 0) this.#blindPrefixes.delete(prefix);
 		}
 	}
+
+	#forgetPath(path: Path.Valid): void {
+		this.#cache.delete(path);
+		for (const prefix of coveringPrefixes(path)) {
+			const paths = this.#paths.get(prefix);
+			paths?.delete(path);
+			if (paths?.size === 0) this.#paths.delete(prefix);
+		}
+	}
+
+	#update(prefix: Path.Valid): void {
+		for (const path of this.#paths.get(prefix) ?? []) {
+			const selected = this.instance(path);
+			for (const entry of this.#cache.get(path)?.values() ?? []) {
+				entry.selection.validity?.update(sameInstance(entry.selection, selected));
+			}
+		}
+	}
 }

js/net/src/ietf/adapter.test.ts

@@ -542,3 +542,30 @@ test("more requests than the setup window complete on one session", async () =>
 		await reading;
 	}
 });
+
+// Canceling the wait must remove the allocation itself, not just stop awaiting it.
+test("a canceled request ID wait never spends a later grant", async () => {
+	for (const [version, alpn] of WINDOW_DRAFTS) {
+		const pair = createMockTransportPair(alpn);
+		const control = await Stream.open(pair.server, { version });
+		const adapter = new ControlStreamAdapter(pair.server, control, version, 0n, true);
+		const running = adapter.run().catch(() => undefined);
+		const peer = await Stream.accept(pair.client, version);
+		if (!peer) throw new Error("no control stream");
+		try {
+			const signal = new AbortController();
+			const canceled = adapter.nextRequestId(signal.signal);
+			signal.abort(new Error("retired"));
+			await expect(canceled).rejects.toThrow("retired");
+			await peer.writer.u53(MaxRequestId.id);
+			await new MaxRequestId({ requestId: 4n }).encode(peer.writer, version);
+			expect(await adapter.nextRequestId()).toBe(0n);
+			await expect(adapter.nextRequestId(signal.signal)).rejects.toThrow("retired");
+			expect(await adapter.nextRequestId()).toBe(2n);
+		} finally {
+			peer.close();
+			adapter.close();
+			void running;
+		}
+	}
+});