feat(net): checkpoint public request answering

js/net/src/answering.test.ts

@@ -0,0 +1,88 @@
+import { afterEach, beforeEach, expect, jest, test } from "bun:test";
+import { Broadcast, Origin, Path } from "./index.ts";
+
+beforeEach(() => jest.useFakeTimers());
+afterEach(() => jest.useRealTimers());
+
+async function settle() {
+	for (let i = 0; i < 64; i++) await Promise.resolve();
+}
+
+test("public track answering is armed before demand and stops without closing accepted tracks", async () => {
+	const source = new Broadcast.Producer();
+	const answering = source.dynamic();
+	const consumer = source.consume();
+	const subscribed = consumer.track("video").subscribe();
+	const request = await answering.requestedTrack();
+	expect(request?.name).toBe("video");
+	const track = request?.accept();
+	if (!track) throw new Error("expected track");
+	const pending = answering.requestedTrack();
+	answering.close();
+	expect(await pending).toBeUndefined();
+	expect(track.closed.peek()).toBeUndefined();
+	const absent = consumer.track("other").subscribe();
+	expect(absent.closed.peek()).toBeInstanceOf(Error);
+	absent.close();
+	subscribed.close();
+	consumer.close();
+	source.close();
+});
+
+test("a fallback is scoped, never advertised, and its answered requests end with it", async () => {
+	const origin = new Origin.Producer();
+	const allowed = new Path.Patterns([Path.Pattern.subtree(Path.from("room"))]);
+	const fallback = origin.scope(Path.empty(), allowed).fallback();
+	const iterator = fallback.requested();
+	const unseen = origin.request(Path.from("elsewhere"));
+	const wanted = origin.request(Path.from("room/alice"));
+	const next = await iterator.next();
+	expect(next.value?.path).toBe(Path.from("room/alice"));
+	const source = new Broadcast.Producer();
+	next.value?.accept(source);
+	await settle();
+	expect(wanted.active.peek()).toBeDefined();
+	expect(unseen.active.peek()).toBeUndefined();
+	const announcements = origin.announced();
+	let advertised = false;
+	const event = announcements.next().then((value) => {
+		advertised = value !== undefined;
+	});
+	await settle();
+	expect(advertised).toBe(false);
+	fallback.close();
+	await settle();
+	expect(wanted.closed.peek()).toBeInstanceOf(Error);
+	wanted.close();
+	unseen.close();
+	announcements.close();
+	await event;
+	source.close();
+	origin.close();
+	await iterator.return?.(undefined);
+});
+
+test("an announced route retires a blind answer rather than stitching its request", async () => {
+	const origin = new Origin.Producer();
+	const fallback = origin.fallback();
+	const iterator = fallback.requested();
+	const wanted = origin.request(Path.from("room"));
+	const next = await iterator.next();
+	const blind = new Broadcast.Producer();
+	next.value?.accept(blind);
+	await settle();
+	expect(wanted.active.peek()).toBeDefined();
+	const announced = origin.createBroadcast(Path.from("room"));
+	announced.announce();
+	await settle();
+	expect(wanted.closed.peek()).toBeInstanceOf(Error);
+	const fresh = origin.request(Path.from("room"));
+	expect(fresh.active.peek()).toBeDefined();
+	fresh.close();
+	wanted.close();
+	fallback.close();
+	announced.close();
+	blind.close();
+	origin.close();
+	await iterator.return?.(undefined);
+});

js/net/src/broadcast.ts

@@ -63,6 +63,7 @@ class BroadcastState {
 	// publisher that has started pulling them. The counterpart of a live Rust `broadcast::Dynamic`;
 	// without one, a track nobody publishes is `NotFound` rather than a request nobody will answer.
 	served = false;
+	handlers = 0;
 	// Each name's sequence namespace, shared by every producer of the name, created or requested.
 	sequences = new TrackSequences();
 	// Live consumer handles sharing this state (see {@link Consumer.clone}). The broadcast
@@ -121,19 +122,20 @@ function dequeueRequest(state: BroadcastState): track.Request | undefined {
 }
 
 // The next on-demand track request, or undefined once the broadcast closes.
-async function requested(state: BroadcastState): Promise<track.Request | undefined> {
+async function requested(state: BroadcastState, stopped?: GetPromise<null>): Promise<track.Request | undefined> {
 	// Pulling requests is what makes a broadcast serve tracks on demand, and it latches for the
 	// broadcast's lifetime since JS has no drop to mark the handler gone. A subscribe before the first
 	// pull is answered NotFound, so an on-demand publisher starts pulling before it publishes and keeps
 	// pulling until the broadcast closes.
-	state.served = true;
+	if (stopped === undefined) state.served = true;
 	for (;;) {
+		if (stopped !== undefined && stopped.peek() !== undefined) return undefined;
 		const request = dequeueRequest(state);
 		if (request) return request;
 
 		if (state.closed.peek() !== undefined) return undefined;
 
-		await Signal.race(state.requested, state.closed);
+		await Signal.race(state.requested, state.closed, ...(stopped !== undefined ? [stopped] : []));
 	}
 }
 
@@ -349,6 +351,44 @@ export class Demand {
 	}
 }
 
+let makeDynamic: (state: BroadcastState) => Dynamic;
+
+/** Answer on-demand track requests and listings for a broadcast. */
+export class Dynamic {
+	#state: BroadcastState;
+	#stopped = new Once<null>();
+
+	private constructor(state: BroadcastState) {
+		this.#state = state;
+		state.handlers++;
+		state.served = true;
+	}
+
+	static {
+		makeDynamic = (state) => new Dynamic(state);
+	}
+
+	/** Wait for the next requested track, or undefined once this handler or its broadcast closes. */
+	requestedTrack(): Promise<track.Request | undefined> {
+		return requested(this.#state, this.#stopped);
+	}
+
+	/** Answer the broadcast's track listings, refusing a second simultaneous lister. */
+	lister(): Lister {
+		return new Lister(this.#state.listing);
+	}
+
+	/** Stop handling new tracks, rejecting queued requests once no handler remains. */
+	close(): void {
+		if (this.#stopped.peek() !== undefined) return;
+		this.#stopped.set(null);
+		if (--this.#state.handlers !== 0) return;
+		this.#state.served = false;
+		for (const request of this.#state.requested.peek()) request.reject(new NotFound("no track handler"));
+		this.#state.requested.set([]);
+	}
+}
+
 /**
  * The write side of a broadcast.
  *
@@ -402,6 +442,11 @@ export class Producer {
 		return this.#demand;
 	}
 
+	/** Handle on-demand track requests before handing consumers the broadcast. */
+	dynamic(): Dynamic {
+		return makeDynamic(this.#state);
+	}
+
 	/** A read handle for this broadcast, named by the path the origin created it at. */
 	consume(): Consumer {
 		return makeConsumer({ state: this.#state, path: this.#path });
@@ -647,6 +692,14 @@ export class Consumer {
 		});
 	}
 
+	#answering?: Dynamic;
+
+	/** Give a custom session its source's on-demand track requests and listings. */
+	protected dynamic(): Dynamic {
+		this.#answering ??= makeDynamic(this.#state);
+		return this.#answering;
+	}
+
 	static {
 		makeConsumer = (shared) => new Consumer(shared as never);
 		hooks.stampPath = (target, path) => {

js/net/src/connection/forward.ts

@@ -5,11 +5,10 @@
  */
 
 import { Milli, Nano } from "@moq/clock";
-import { type Dispose, race } from "@moq/signals";
 import type * as Announce from "../announced.ts";
 import { type Route, routesEqual } from "../hop.ts";
 import { coveringPrefixes } from "../internal.ts";
-import type { Dynamic, Producer as OriginProducer, RequestSlot } from "../origin.ts";
+import type { Dynamic, Producer as OriginProducer } from "../origin.ts";
 import * as Path from "../path.ts";
 import { after } from "../util/timer.ts";
 import { wireOf } from "../wire.ts";
@@ -295,49 +294,12 @@ async function drive(handle: Dynamic, conn: Established): Promise<void> {
  * any blind answer, so answering one would only park a handle nothing reads.
  */
 async function serveRequests(conn: Established, origin: OriginProducer): Promise<void> {
-	const table = wireOf(origin);
+	const fallback = origin.fallback();
 	const session = wireOf(conn);
-	// The withdraws for the answers this session provided, so a dead session only takes
-	// back its own. Keyed by path but remembering the slot, because a path outlives its
-	// slot: the last handle closing tears the slot down and a new request installs a fresh
-	// one, and those two writes coalesce into a single wakeup. Matching on the path alone
-	// would read the new slot as already answered and leave it unanswered forever.
-	const answered = new Map<Path.Valid, { slot: RequestSlot; withdraw: Dispose }>();
-
-	let dead = false;
-	const closed = conn.closed.then(() => {
-		dead = true;
-	});
-
-	for (;;) {
-		const map = table.requests.peek();
-		if (!map || dead) break;
-
-		for (const [path, slot] of map) {
-			if (!table.blind(slot)) continue;
-			if (answered.get(path)?.slot === slot || slot.answer !== undefined) continue;
-			if (table.routes(path)) continue;
-			const withdraw = table.answer(path, session.consume(path));
-			if (withdraw) answered.set(path, { slot, withdraw });
-		}
-
-		// A withdrawn request already released the answer; just forget our claim on the path.
-		// A replaced slot counts as withdrawn: the answer we hold belongs to the slot that
-		// went away, not to whatever now occupies the path.
-		for (const [path, entry] of [...answered]) {
-			if (map.get(path) === entry.slot && table.blind(entry.slot)) continue;
-			answered.delete(path);
-			entry.withdraw();
-		}
-
-		// Woken by the table too, not just the requests: a path that stops being routed needs
-		// the blind answer this loop skipped while it was.
-		await race([table.changed(), closed]);
-	}
-
-	// Session gone: withdraw our answers, waking a standby session to provide fresh ones.
-	for (const { withdraw } of answered.values()) {
-		withdraw();
+	void conn.closed.then(() => fallback.close());
+	try {
+		for await (const request of fallback.requested()) request.accept(session.consume(request.path));
+	} finally {
+		fallback.close();
 	}
-	answered.clear();
 }

js/net/src/goaway-requests.test.ts

@@ -321,13 +321,13 @@ async function routeDeath(kind: "lite" | "ietf", after: boolean): Promise<void>
 	]);
 
 	const watching = consume.announced();
+	expect((await watching.next())?.kind).toBe("start");
 	const request = consume.request(Path.from("room"));
 	let routes: ReturnType<typeof consume.announced> | undefined;
 	let replacement: Awaited<ReturnType<typeof connect>> | undefined;
 	let replacementServer: Awaited<ReturnType<typeof accept>> | undefined;
 
 	try {
-		expect((await watching.next())?.kind).toBe("start");
 		await waitUntil(() => request.active.peek() !== undefined);
 		const front = request.active.peek();
 		if (!front) throw new Error("no route");

js/net/src/ietf/subscriber.ts

@@ -39,7 +39,7 @@ import { untilAborted } from "../util/abort.ts";
 import { Deadline, requestOptions } from "../util/deadline.ts";
 import { abandoned as demandAbandoned, isAbandoned, unwanted } from "../util/demand.ts";
 import { LIMITS, type Limits, SlotMap, Slots } from "../util/limits.ts";
-import { overrideBroadcastWire, wireOf } from "../wire.ts";
+import { overrideBroadcastWire } from "../wire.ts";
 import type { Session } from "./adapter.ts";
 import { DuplicateTrackAlias, RetiredTrackAlias, TrackAliases } from "./aliases.ts";
 import * as Cluster from "./cluster.ts";
@@ -730,10 +730,10 @@ export class Subscriber {
 		// moq-transport resolves metadata through SUBSCRIBE and has no one-shot group fetch here.
 		const consumer = new ConsumeBroadcast();
 		// Registered before anyone can list the broadcast, so no reader syncs on an empty answer.
-		void this.#refuseListing(wireOf(consumer).lister());
+		void this.#refuseListing(consumer.answering.lister());
 		void (async () => {
 			for (;;) {
-				const request = await wireOf(consumer).requested();
+				const request = await consumer.answering.requestedTrack();
 				if (!request) break;
 				// Every request names the instance the consume resolved, so the peer refuses
 				// rather than answers from another.
@@ -1759,6 +1759,10 @@ export class Subscriber {
  * group fetch, so `track.Consumer.fetchGroup()` is rejected.
  */
 class ConsumeBroadcast extends broadcast.Consumer {
+	get answering(): broadcast.Dynamic {
+		return this.dynamic();
+	}
+
 	constructor(state?: never) {
 		super(state);
 		overrideBroadcastWire(this, {

js/net/src/lite/subscriber.ts

@@ -36,7 +36,7 @@ import { Deadline, REQUEST_TIMEOUT_MS, requestOptions } from "../util/deadline.t
 import { abandoned, isAbandoned, unwanted } from "../util/demand.ts";
 import { LIMITS, type Limits, SlotMap, Slots } from "../util/limits.ts";
 import { Budget } from "../util/yield.ts";
-import { overrideBroadcastWire, wireOf } from "../wire.ts";
+import { overrideBroadcastWire } from "../wire.ts";
 import {
 	AnnounceHistory,
 	AnnounceInit,
@@ -648,11 +648,11 @@ export class Subscriber {
 		};
 		const consumer = new ConsumeBroadcast(this, path, consumed);
 		// Registered before anyone can list the broadcast, so no reader syncs without the peer.
-		void this.#runListing(path, consumed, wireOf(consumer).lister());
+		void this.#runListing(path, consumed, consumer.answering.lister());
 
 		void (async () => {
 			for (;;) {
-				const request = await wireOf(consumer).requested();
+				const request = await consumer.answering.requestedTrack();
 				if (!request) break;
 				void this.#runSubscribe(path, consumed, request);
 			}
@@ -1848,6 +1848,10 @@ async function untilAbandoned<T>(group: netGroup.Producer, step: Promise<T>): Pr
  * session. Live subscribes still flow through the inherited requested() queue.
  */
 class ConsumeBroadcast extends broadcast.Consumer {
+	get answering(): broadcast.Dynamic {
+		return this.dynamic();
+	}
+
 	#subscriber: Subscriber;
 	#path: Path.Valid;
 	#consumed: Consumed;