fix(net): bind JS consumes to the serving announcement

js/net/bench/broadcasts.ts

@@ -1,7 +1,9 @@
 /** Sweep observers against routes for one route update, and against tracks for one demand edge. */
 
 import { Producer as BroadcastProducer } from "../src/broadcast.ts";
+import { BroadcastCache } from "../src/consume.ts";
 import * as Epoch from "../src/epoch.ts";
+import { Route } from "../src/hop.ts";
 import { Producer } from "../src/origin.ts";
 import * as Path from "../src/path.ts";
 
@@ -120,3 +122,99 @@ for (const pinned of [false, true]) {
 		}
 	}
 }
+
+// Only a touched announcement and path should cost work, regardless of unrelated routes or consumes.
+console.log("subscriber,announcements,consumers,update_and_consume_us");
+for (const routeCount of routeCounts) {
+	for (const consumerCount of [8, 32, 128]) {
+		const cache = new BroadcastCache();
+		const owners = Array.from({ length: routeCount }, () => ({}));
+		for (let i = 0; i < routeCount; i++) cache.announce(Path.from(`pool/${i}`), owners[i], Route.default);
+		const sources = Array.from({ length: consumerCount }, () => new BroadcastProducer());
+		const held = sources.map((source, i) => {
+			const path = Path.from(`pool/${i % routeCount}/job/${i}`);
+			return cache.insert(path, cache.instance(path), source.consume());
+		});
+		const touched = Path.from("pool/0/job/0");
+		const start = performance.now();
+		for (let i = 0; i < 4096; i++) {
+			cache.announce(Path.from("pool/0"), owners[0], { ...Route.default, cost: BigInt(i) });
+			const shared = cache.get(touched, cache.instance(touched));
+			if (!shared || shared.closed !== held[0].closed) throw new Error("update replaced the source");
+			shared.close();
+		}
+		console.log(`cache,${routeCount},${consumerCount},${(((performance.now() - start) * 1000) / 4096).toFixed(2)}`);
+		for (const consumer of held) consumer.close();
+		for (const source of sources) source.close();
+	}
+}
+
+// Switching a more-specific route on and off must recover all held fallback consumers.
+console.log("alternation,announcements,consumers,select_us,duplicates");
+for (const routeCount of routeCounts) {
+	for (const consumerCount of [1, 8, 32]) {
+		const cache = new BroadcastCache();
+		const owners = Array.from({ length: routeCount }, () => ({}));
+		for (let i = 0; i < routeCount; i++) cache.announce(Path.from(`pool/${i}`), owners[i], Route.default);
+		const sources = Array.from({ length: consumerCount }, () => new BroadcastProducer());
+		const paths = sources.map((_, i) => Path.from(`pool/0/job/${i}`));
+		const held = sources.map((source, i) => cache.insert(paths[i], cache.instance(paths[i]), source.consume()));
+		const specific = {};
+		const prefix = Path.from("pool/0/job");
+		const epoch = Epoch.mint();
+		let duplicates = 0;
+		const start = performance.now();
+		for (let round = 0; round < 256; round++) {
+			if (round % 2 === 0) cache.announce(prefix, specific, { ...Route.default, epoch });
+			else cache.withdraw(prefix, specific);
+			for (let i = 0; i < consumerCount; i++) {
+				const instance = cache.instance(paths[i]);
+				const shared = cache.get(paths[i], instance);
+				if (shared) shared.close();
+				else {
+					if (round > 0) duplicates++;
+					held.push(cache.insert(paths[i], instance, sources[i].consume()));
+				}
+			}
+		}
+		console.log(
+			`alternation,${routeCount},${consumerCount},${(((performance.now() - start) * 1000) / 256).toFixed(2)},${duplicates}`,
+		);
+		for (const consumer of held) consumer.close();
+		for (const source of sources) source.close();
+	}
+}
+
+// Blind invalidation touches only the held paths covered by the new announcement.
+console.log("blind,announcements,consumers,invalidate_us,stale");
+for (const routeCount of routeCounts) {
+	for (const consumerCount of [1, 8, 32]) {
+		const cache = new BroadcastCache();
+		for (let i = 0; i < routeCount; i++) cache.announce(Path.from(`other/${i}`), {}, Route.default);
+		const source = new BroadcastProducer();
+		const keeper = source.consume();
+		const paths = Array.from({ length: consumerCount }, (_, i) => Path.from(`pool/job/${i}`));
+		const prefix = Path.from("pool");
+		const owner = {};
+		let elapsed = 0;
+		let stale = 0;
+		for (let round = 0; round < 256; round++) {
+			const held = paths.map((path) => cache.insert(path, cache.instance(path), source.consume()));
+			const start = performance.now();
+			cache.announce(prefix, owner, Route.default);
+			cache.withdraw(prefix, owner);
+			elapsed += performance.now() - start;
+			for (const path of paths) {
+				const shared = cache.get(path, cache.instance(path));
+				if (shared) {
+					stale++;
+					shared.close();
+				}
+			}
+			for (const consumer of held) consumer.close();
+		}
+		console.log(`blind,${routeCount},${consumerCount},${((elapsed * 1000) / 256).toFixed(2)},${stale}`);
+		keeper.close();
+		source.close();
+	}
+}

js/net/src/consume.test.ts

@@ -0,0 +1,101 @@
+import { expect, test } from "bun:test";
+import { Producer } from "./broadcast.ts";
+import { BroadcastCache } from "./consume.ts";
+import * as Epoch from "./epoch.ts";
+import { Route } from "./hop.ts";
+import * as Path from "./path.ts";
+
+test("cache identity follows the serving announcement, preserving held tracks", async () => {
+	const cache = new BroadcastCache();
+	const pool = Path.from("pool");
+	const job = Path.from("pool/job");
+	const broad = {};
+	const specific = {};
+	const epoch = Epoch.mint();
+	cache.announce(pool, broad, { ...Route.default, epoch });
+	const source = new Producer();
+	const track = source.createTrack("video");
+	const held = cache.insert(job, cache.instance(job), source.consume());
+	const ordered = held.track("video").subscribe().ordered();
+	cache.announce(job, specific, { ...Route.default, epoch });
+	const same = cache.get(job, cache.instance(job));
+	expect(same?.closed).toBe(held.closed);
+
+	cache.announce(job, specific, Route.default, true);
+	expect(cache.instance(job).route.epoch).toBeUndefined();
+	expect(cache.get(job, cache.instance(job))).toBeUndefined();
+	const replacement = new Producer();
+	const fresh = cache.insert(job, cache.instance(job), replacement.consume());
+	cache.withdraw(pool, broad);
+	const shared = cache.get(job, cache.instance(job));
+	expect(shared?.closed).toBe(fresh.closed);
+	// Cache replacement never transfers or closes the track an old handle is reading.
+	track.appendGroup().close();
+	expect((await ordered.nextGroup())?.sequence).toBe(0);
+	cache.withdraw(job, specific);
+	expect(cache.get(job, cache.instance(job))).toBeUndefined();
+	for (const consumer of [held, same, fresh, shared]) consumer?.close();
+	ordered.close();
+	source.close();
+	replacement.close();
+});
+
+test("withdrawing one interest keeps another interest's explicit identity", () => {
+	const cache = new BroadcastCache();
+	const path = Path.from("room");
+	const first = {};
+	const second = {};
+	const route = { ...Route.default, epoch: Epoch.mint() };
+	cache.announce(path, first, route);
+	cache.announce(path, second, route);
+	const source = new Producer();
+	const held = cache.insert(path, cache.instance(path), source.consume());
+	cache.withdraw(path, first);
+	const same = cache.get(path, cache.instance(path));
+	expect(same?.closed).toBe(held.closed);
+	cache.withdraw(path, second);
+	expect(cache.get(path, cache.instance(path))).toBeUndefined();
+	held.close();
+	same?.close();
+	source.close();
+});
+
+test("an announcement lifecycle retires a blind cache without disturbing unrelated paths", () => {
+	const cache = new BroadcastCache();
+	const path = Path.from("pool/job");
+	const other = Path.from("other/job");
+	const source = new Producer();
+	const unrelated = new Producer();
+	const held = cache.insert(path, cache.instance(path), source.consume());
+	const otherHeld = cache.insert(other, cache.instance(other), unrelated.consume());
+	const owner = {};
+	cache.announce(Path.from("pool"), owner, Route.default);
+	cache.withdraw(Path.from("pool"), owner);
+	expect(cache.get(path, cache.instance(path))).toBeUndefined();
+	const same = cache.get(other, cache.instance(other));
+	expect(same?.closed).toBe(otherHeld.closed);
+	expect(held.closed.peek()).toBeUndefined();
+	for (const consumer of [held, otherHeld, same]) consumer?.close();
+	source.close();
+	unrelated.close();
+});
+
+test.each([false, true])("route alternation reuses every held instance (epoch: %s)", (explicit) => {
+	const cache = new BroadcastCache();
+	const path = Path.from("pool/job");
+	const broad = Path.from("pool");
+	const ownerA = {};
+	const ownerB = {};
+	cache.announce(broad, ownerA, { ...Route.default, epoch: explicit ? Epoch.mint() : undefined });
+	const a = new Producer();
+	const heldA = cache.insert(path, cache.instance(path), a.consume());
+	cache.announce(path, ownerB, { ...Route.default, epoch: explicit ? Epoch.mint() : undefined });
+	const b = new Producer();
+	const heldB = cache.insert(path, cache.instance(path), b.consume());
+	cache.withdraw(path, ownerB);
+	const again = cache.get(path, cache.instance(path));
+	expect(again?.closed).toBe(heldA.closed);
+	for (const consumer of [heldA, heldB, again]) consumer?.close();
+	a.close();
+	b.close();
+});

js/net/src/consume.ts

@@ -1,68 +1,102 @@
 import type * as broadcast from "./broadcast.ts";
-import * as Path from "./path.ts";
+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";
 
-/**
- * Per-path dedup cache for consumed broadcasts, shared by the moq-lite and moq-ietf
- * subscribers.
- *
- * Resolving the same path must not mint a fresh subscription per call: repeat requests
- * for the same path (e.g. several renditions referencing one `broadcast: "./source"`) should
- * share a single upstream subscription. This mirrors the Rust `origin::Consumer` weak-cache:
- * a still-live path resolves to a shared {@link broadcast.Consumer.clone}, a closed one is
- * re-consumed on the next request. Each handle is reference-counted, so the shared broadcast
- * closes once every caller has closed its handle.
- *
- * @internal
- */
+class Blind implements Instance {
+	readonly identity = {};
+	readonly route = {};
+}
+
+/** Consumed broadcasts shared only within the selected serving instance. @internal */
 export class BroadcastCache {
-	// The base handle per path; callers get reference-counted clones of it.
-	#cache = new Map<Path.Valid, broadcast.Consumer>();
+	#cache = new Map<Path.Valid, Map<Epoch.Valid | object, broadcast.Consumer>>();
+	#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>>();
+
+	/** Record this interest's live advertisement, preserving identity across route updates. */
+	announce(path: Path.Valid, owner: object, route: Route, restart = false): void {
+		// Only unannounced consumers covered by this advertisement lose their blind identity.
+		for (const covered of this.#blindPrefixes.get(path) ?? []) this.#forgetBlind(covered);
+		let entries = this.#announced.get(path);
+		if (!entries) {
+			entries = new Map();
+			this.#announced.set(path, entries);
+		}
+		const identity = (!restart && entries.get(owner)?.identity) || {};
+		entries.set(owner, { identity, route });
+	}
+
+	/** 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);
+	}
 
-	/** A shared handle to the live broadcast cached for `path`, or `undefined` on a miss. */
-	get(path: Path.Valid): broadcast.Consumer | undefined {
-		const base = this.#cache.get(path);
-		if (base && base.closed.peek() === undefined) return base.clone();
-		return undefined;
+	/** Resolve the same most-specific-prefix and route preference as the origin. */
+	instance(path: Path.Valid): Instance {
+		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;
+		}
+		return this.#blind.get(path) ?? new Blind();
 	}
 
-	/**
-	 * Cache `consumer` as the base handle for `path` (evicting it once it closes) and return it.
-	 * Call on a {@link get} miss, after wiring up the fresh consumer's subscribe loop.
-	 */
-	insert(path: Path.Valid, consumer: broadcast.Consumer): broadcast.Consumer {
-		this.#cache.set(path, consumer);
+	/** 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;
+	}
 
-		// Drop the entry once the broadcast closes (every handle released), so the next request
-		// re-consumes rather than cloning a dead handle. Guard against a newer entry for the path.
+	/** 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 {
+		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);
+		if (instance instanceof Blind) {
+			this.#blind.set(path, instance);
+			for (const prefix of coveringPrefixes(path)) {
+				let paths = this.#blindPrefixes.get(prefix);
+				if (!paths) {
+					paths = new Set();
+					this.#blindPrefixes.set(prefix, paths);
+				}
+				paths.add(path);
+			}
+		}
 		void consumer.closed.then(() => {
-			if (this.#cache.get(path) === consumer) this.#cache.delete(path);
+			if (entries.get(key) !== consumer) return;
+			entries.delete(key);
+			if (entries.size === 0 && this.#cache.get(path) === entries) this.#cache.delete(path);
+			if (this.#blind.get(path) === instance) this.#forgetBlind(path);
 		});
-
 		return consumer;
 	}
 
-	/**
-	 * Stop sharing the broadcasts cached at or beneath `path`, so the next request subscribes fresh.
-	 *
-	 * Call when the advertisement for `path` goes away or is replaced. It covered every path
-	 * beneath it, so those go too, even one a more specific advertisement also covers: that costs
-	 * a duplicate subscription, never a stale one. A handle only leaves the cache on its own
-	 * once *every* holder has closed it, so one holder outliving the publisher (a second
-	 * watcher, or a caller consuming the path directly) would otherwise keep the dead
-	 * generation's cached tracks alive and hand them to whoever consumes the path next.
-	 * Existing handles are left alone: they belong to their holders, and the wire resets
-	 * whatever they still have open.
-	 *
-	 * Eviction is unconditional, which costs a dedup miss when two announcement streams watch
-	 * one path: the second stream's retraction can arrive after the first has already seen the
-	 * replacement, dropping the fresh entry so the next request subscribes again instead of
-	 * sharing. Telling that stale retraction from a live one needs a generation id on the
-	 * advertisement (moq-lite's `Epoch`, not yet on the wire), so until then this errs toward a
-	 * duplicate subscription rather than risk handing out a dead one.
-	 */
-	evict(path: Path.Valid): void {
-		for (const cached of this.#cache.keys()) {
-			if (Path.hasPrefix(path, cached)) this.#cache.delete(cached);
+	#forgetBlind(path: Path.Valid): void {
+		const blind = this.#blind.get(path);
+		if (!blind) return;
+		this.#blind.delete(path);
+		const entries = this.#cache.get(path);
+		entries?.delete(blind.identity);
+		if (entries?.size === 0) this.#cache.delete(path);
+		for (const prefix of coveringPrefixes(path)) {
+			const paths = this.#blindPrefixes.get(prefix);
+			paths?.delete(path);
+			if (paths?.size === 0) this.#blindPrefixes.delete(prefix);
 		}
 	}
 }