feat(js/net): list a broadcast's tracks over lite-07 TRACKS

doc/concept/moq-lite.md

@@ -56,8 +56,8 @@ always a prefix. A service that serves only some of the paths beneath it
 refuses the rest as they are requested. Announcements are hints; the request is
 what decides.
 
-Inside a broadcast, a reader can list its tracks (moq-lite-07, Rust only so
-far): each track with its fixed properties, a synced mark, then tracks as they
+Inside a broadcast, a reader can list its tracks (moq-lite-07): each track
+with its fixed properties, a synced mark, then tracks as they
 come and go. Listing is discovery, not delivery: subscribe to each track you
 want, and its groups still cached upstream reach you late. A broadcast lists
 the tracks its publisher creates, never one served only on request, and a path

doc/lib/js/net.md

@@ -42,11 +42,12 @@ broadcast.announce();
 - **Connections share by URL and versions.** `new Connection({ url })` pools and reconnects with backoff, which the elements use. An ordered `versions` list, spelled like `moq-lite-06` as in Rust, applies to both WebTransport and WebSocket; equivalent lists share, while a different order or an omitted list gets its own pool entry. 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. 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.
+- **List a broadcast's tracks.** `for await (const event of broadcast.tracks())` reports each track the broadcast lists (`added`, with its `Track.Info`), one `synced`, then `added` and `removed` as tracks come and go, until the broadcast ends; leaving the loop stops listing. Over a session it needs `moq-lite-07-wip`, syncs at once with what this side holds, and the peer's tracks follow as `added`. A track served only on request is never listed. See [discovery](/concept/moq-lite#discovery).
 - **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 (opt in with `versions: ["moq-lite-07-wip", "moq-lite-06"]`, listing every fallback you still want) 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 (opt in with `versions: ["moq-lite-07-wip", "moq-lite-06"]`, listing every fallback you still want) 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 or track listing 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/net/bench/broadcasts.ts

@@ -105,6 +105,36 @@ for (const trackCount of [8, 32, 128]) {
 	}
 }
 
+// One created and removed track reaches every listing, however many tracks the broadcast holds.
+const listed = 256;
+console.log("tracks,listings,create_remove_and_read_us");
+for (const trackCount of [8, 128, 1024]) {
+	for (const listingCount of [1, 8, 32]) {
+		const broadcast = new BroadcastProducer();
+		for (let index = 0; index < trackCount; index++) broadcast.createTrack(`held-${index}`);
+		const listings = Array.from({ length: listingCount }, () => broadcast.consume().tracks());
+		// Past each listing's initial set and SYNCED.
+		for (const listing of listings) {
+			for (let index = 0; index <= trackCount; index++) await listing.next();
+		}
+		const start = performance.now();
+		for (let index = 0; index < listed; index++) {
+			broadcast.createTrack(`new-${index}`);
+			for (const listing of listings) {
+				const event = await listing.next();
+				if (event?.kind !== "added" || event.name !== `new-${index}`) throw new Error("listing missed a track");
+			}
+			broadcast.removeTrack(`new-${index}`);
+			for (const listing of listings) {
+				if ((await listing.next())?.kind !== "removed") throw new Error("listing missed a removal");
+				checksum++;
+			}
+		}
+		console.log(`${trackCount},${listingCount},${(((performance.now() - start) * 1000) / listed).toFixed(1)}`);
+		broadcast.close();
+	}
+}
+
 // Resolve one touched path while both the table and that path's readers grow.
 console.log("request,routes,readers,update_and_read_us");
 const epoch = Epoch.parse("01900000-0000-7000-8000-000000000001");

js/net/src/auth.test.ts

@@ -623,6 +623,40 @@ test("a fetch abandoned while its token waits never reaches the wire", async ()
 	}
 });
 
+// The peer refuses a listing its token has not yet authorized, so the request waits for it.
+test("a listing waits for its token before reaching the wire", async () => {
+	const origin = new OriginProducer();
+	const broadcast = origin.createBroadcast(Path.from("room/x"));
+	broadcast.announce();
+	broadcast.createTrack("video");
+	const { client, server, transport } = await connect({ serverPublish: origin, protocol: Lite.ALPN_07_WIP });
+	const requests = server.auth.requests();
+	const setup = await requests.next();
+	if (!setup) throw new Error("missing setup token");
+	const initial = setup.accept(grant([], []));
+	await waitFor(client.auth.grant, (g) => g !== undefined);
+	const adding = client.auth.add("first");
+	try {
+		const unanswered = await requests.next();
+		if (!unanswered) throw new Error("missing token");
+		const opened = transport.sendStreams.bidi.length;
+		const tracks = wireOf(client).consume(Path.from("room/x")).tracks();
+		for (let i = 0; i < 10; i++) await nextTask();
+		expect(transport.sendStreams.bidi.length).toBe(opened);
+		const issued = unanswered.accept(grant([], ["room"]));
+		await adding;
+		expect(await tracks.next()).toEqual({ kind: "synced" });
+		expect(await tracks.next()).toMatchObject({ kind: "added", name: "video" });
+		tracks.close();
+		issued.close();
+	} finally {
+		initial.close();
+		client.abort();
+		server.abort();
+		await adding.catch(() => void 0);
+	}
+});
+
 // The window between the wait seeing the caller leave and acting on it spans a few
 // microtasks, so sweep where the new caller lands.
 test.each(Array.from({ length: 17 }, (_, gap) => gap))(

js/net/src/broadcast.ts

@@ -10,11 +10,14 @@ import { NotFound } from "./error.ts";
 import type { Consumer as GroupConsumer } from "./group.ts";
 import { Route } from "./hop.ts";
 import { hooks, TrackSequences } from "./internal.ts";
+import { Lister, Listing, listTracks, type Tracks } from "./listing.ts";
 import * as Path from "./path.ts";
 import * as track from "./track.ts";
 import { untilAborted } from "./util/abort.ts";
 import { registerWire, trackOf } from "./wire.ts";
 
+export { Tracks, type TracksEvent } from "./listing.ts";
+
 /** The origin callback a created broadcast uses to advertise its exact path. @internal */
 export interface Announcer {
 	/** Advertise or re-price this broadcast's path. */
@@ -48,6 +51,8 @@ class BroadcastState {
 	requested = new Signal<track.Request[]>([]);
 	pending = new Set<track.Request>();
 	closed = new Once<null>();
+	// The tracks the broadcast lists, and the readers listing them.
+	listing = new Listing(this.closed);
 	// The tracks the application inserted. A finished one stays and keeps serving its cache; only an
 	// abort or removeTrack drops it, the way only a dropped producer leaves a Rust broadcast.
 	tracks = new Map<string, track.Producer>();
@@ -141,6 +146,7 @@ async function requested(state: BroadcastState): Promise<track.Request | undefin
 function closeState(state: BroadcastState) {
 	if (state.closed.peek() !== undefined) return;
 	state.closed.set(null);
+	state.listing.detach(null);
 	for (const cleanup of state.demands.values()) cleanup();
 	for (const request of state.pending) request.reject();
 	state.requested.mutate((requests) => {
@@ -297,18 +303,19 @@ function takeQueued(state: BroadcastState, name: string): track.Request | undefi
 	return queued;
 }
 
-// Serve `producer` as the broadcast's track for its name.
-function adopt(state: BroadcastState, producer: track.Producer): void {
+// Serve `producer` as the broadcast's track for its name, listed with `info`.
+function adopt(state: BroadcastState, producer: track.Producer, info: track.Info): void {
 	const lookups = state.lookups.get(producer) ?? new Lookups();
 	state.lookups.set(producer, lookups);
 	watchDemand(state, producer, lookups);
 	state.tracks.set(producer.name, producer);
+	state.listing.listOwn(producer.name, producer, info);
 
 	// A finished track keeps serving its cache, so only an abort evicts it.
 	void producer.closed.then((closed) => {
-		if (closed instanceof Error && state.tracks.get(producer.name) === producer) {
-			state.tracks.delete(producer.name);
-		}
+		if (!(closed instanceof Error)) return;
+		if (state.tracks.get(producer.name) === producer) state.tracks.delete(producer.name);
+		state.listing.unlistOwn(producer.name, producer);
 	});
 }
 
@@ -362,6 +369,8 @@ export class Producer {
 			largest: (name) => largest(this.#state, name),
 			fetchGroup: (name, sequence, options) => fetchGroup(this.#state, name, sequence, options),
 			requested: () => requested(this.#state),
+			tracks: (owner) => listTracks(this.#state.listing, owner),
+			lister: () => new Lister(this.#state.listing),
 		});
 	}
 
@@ -420,7 +429,7 @@ export class Producer {
 			const lookups = this.#state.lookups.get(producer);
 			if (lookups) lookups.opened = false;
 			this.#state.requests.delete(name);
-			adopt(this.#state, producer);
+			adopt(this.#state, producer, track.infoDefaults(info));
 			return producer;
 		}
 
@@ -433,20 +442,21 @@ export class Producer {
 
 		const producer = new track.Producer(name).accept(info);
 		hooks.bindSequence(producer, this.#state.sequences.get(name));
-		adopt(this.#state, producer);
+		adopt(this.#state, producer, track.infoDefaults(info));
 		return producer;
 	}
 
 	/**
-	 * Remove a statically inserted track, including a finished cached one, and stop counting it
-	 * toward {@link demand} at once. Throws once the broadcast is finished.
+	 * Remove a statically inserted track, including a finished cached one: stop counting it
+	 * toward {@link demand} and listing it at once. Throws once the broadcast is finished.
 	 */
 	removeTrack(name: string): void {
 		this.#assertUnfinished();
 		const track = this.#state.tracks.get(name);
 		if (!track) return;
 		this.#state.tracks.delete(name);
 		this.#state.demands.get(track)?.();
+		this.#state.listing.unlistOwn(name, track);
 	}
 
 	/** A lazy read handle for a track on this broadcast. */
@@ -588,6 +598,7 @@ interface Shared {
 	state: BroadcastState;
 	path: Path.Valid;
 	epoch?: Epoch.Valid;
+	unlisted?: boolean;
 }
 
 // Constructs a Consumer from within this module without exposing a public constructor
@@ -606,6 +617,8 @@ export class Consumer {
 	#state: BroadcastState;
 	#path: Path.Valid;
 	#epoch?: Epoch.Valid;
+	// Served on request through a route claiming a shorter prefix, which names no broadcast to list.
+	#unlisted: boolean;
 
 	// Guards against a double close() on this handle over-decrementing the consumer count.
 	#closed = false;
@@ -614,6 +627,7 @@ export class Consumer {
 	protected constructor(shared?: Shared) {
 		this.#path = shared?.path ?? Path.empty();
 		this.#epoch = shared?.epoch;
+		this.#unlisted = shared?.unlisted ?? false;
 		if (shared) {
 			this.#state = shared.state;
 		} else {
@@ -628,6 +642,8 @@ export class Consumer {
 			largest: (name) => largest(this.#state, name),
 			fetchGroup: (name, sequence, options) => fetchGroup(this.#state, name, sequence, options),
 			requested: () => requested(this.#state),
+			tracks: (owner) => this.#tracks(owner),
+			lister: () => new Lister(this.#state.listing),
 		});
 	}
 
@@ -640,6 +656,9 @@ export class Consumer {
 		hooks.stampEpoch = (target, epoch) => {
 			target.#epoch = epoch;
 		};
+		hooks.stampUnlisted = (target) => {
+			target.#unlisted = true;
+		};
 	}
 
 	/**
@@ -691,14 +710,39 @@ export class Consumer {
 	// Hand this consumer's backing state and path to a clone. Opaque (`never`) so the state type
 	// stays unexported; a subclass passes it straight back into its own `super(...)`.
 	protected shareState(): never {
-		return { state: this.#state, path: this.#path, epoch: this.#epoch } satisfies Shared as never;
+		return {
+			state: this.#state,
+			path: this.#path,
+			epoch: this.#epoch,
+			unlisted: this.#unlisted,
+		} satisfies Shared as never;
 	}
 
 	/** Get a lazy handle for a track on this broadcast. Repeat subscriptions dedupe onto one upstream subscription. */
 	track(name: string): track.Consumer {
 		return trackOf(name, this);
 	}
 
+	/**
+	 * List the broadcast's tracks: each one with its info, then `synced`, then changes as they
+	 * happen, until the broadcast ends.
+	 *
+	 * Lists the tracks its producer created (until they abort or are removed) and, for a
+	 * broadcast consumed from a session, the tracks the peer lists. Through a session the
+	 * listing syncs at once, with what this side holds, and the peer's tracks follow as live
+	 * adds; every listing of one broadcast shares one request to the peer. A track served only
+	 * on request is never listed, and listing never requests one. Throws once the broadcast
+	 * has closed.
+	 */
+	tracks(): Tracks {
+		return this.#tracks();
+	}
+
+	#tracks(owner?: object): Tracks {
+		if (this.#unlisted) throw new NotFound("a route claiming a prefix lists no tracks");
+		return listTracks(this.#state.listing, owner);
+	}
+
 	/**
 	 * Release this handle. The broadcast is closed once this was the last live handle;
 	 * while other {@link clone}s remain open it stays live.

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

@@ -867,6 +867,18 @@ test("a PUBLISH_NAMESPACE update that changes the publisher applies in place", a
 	expect(await announced.next()).toMatchObject({ prefix: Path.from("theirs"), kind: "end" });
 });
 
+test("a consumed broadcast refuses to list its tracks rather than claim it has none", async () => {
+	const pair = createMockTransportPair(ALPN.DRAFT_19);
+	const session = new NativeSession(pair.server, VERSION, true);
+	const subscriber = new Subscriber({ session, cluster: { self: SELF, peer: PEER } });
+	const consumer = subscriber.consume(Path.from("theirs"));
+	await expect(consumer.tracks().next()).rejects.toThrow("track listing requires moq-lite-07");
+	// A later reader asks again, and is refused again.
+	await expect(consumer.tracks().next()).rejects.toThrow("track listing requires moq-lite-07");
+	consumer.close();
+	subscriber.close();
+});
+
 /**
  * A second PUBLISH_NAMESPACE on the stream that already carries one is no longer an
  * update: it is the base draft's duplicate request, a protocol violation.

js/net/src/ietf/subscriber.ts

@@ -29,6 +29,7 @@ import {
 	scopeHead,
 	scopeOverlaps,
 } from "../internal.ts";
+import type { Lister } from "../listing.ts";
 import * as Path from "../path.ts";
 import { type Cursor, Reader, type Stream } from "../stream.ts";
 import { Tail } from "../tail.ts";
@@ -726,6 +727,8 @@ export class Subscriber {
 	#createConsume(path: Path.Valid, instance: ReturnType<BroadcastCache["instance"]>): broadcast.Consumer {
 		// 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 (async () => {
 			for (;;) {
 				const request = await wireOf(consumer).requested();
@@ -738,6 +741,22 @@ export class Subscriber {
 		return this.#consumes.insert(path, instance, consumer);
 	}
 
+	// moq-transport lists no tracks here: refuse each listing until the broadcast closes, since an
+	// empty answer would claim the peer lists nothing.
+	async #refuseListing(lister: Lister): Promise<void> {
+		try {
+			for (;;) {
+				while (!lister.used.peek()) {
+					if (lister.ended.peek() !== undefined) return;
+					await race([Signal.race(lister.used, lister.ended)]);
+				}
+				lister.abort(new Error("track listing requires moq-lite-07 or newer"));
+			}
+		} finally {
+			lister.close();
+		}
+	}
+
 	// The adapter is gone. If the transport has already closed, that close is the
 	// error. A still-open transport, such as a GOAWAY drain, has no peer code yet,
 	// so this does not wait for it. Either way the route failed, so a resumed track moves on.