feat(js/net)!: announce streams yield a live marker once caught up

demo/web/src/index.ts

@@ -176,8 +176,8 @@ function createTile(name: string): WatchTile {
 // ---------------------------------------------------------------------------
 //
 // Subscribe to announcements under the prefix and keep a live set of active broadcasts.
-// `announced.next()` drains the update stream, so we track membership ourselves: active=true adds the
-// path, active=false removes it. `Connection.announced()` spans reconnects (it retracts everything on
+// `announced.next()` drains the event stream, so we track membership ourselves: an announcement adds the
+// path, a retraction removes it. `Connection.announced()` spans reconnects (it retracts everything on
 // disconnect and re-announces on reconnect), so the set self-heals without any extra wiring here.
 const discovery = new Signals.Effect();
 discovery.run((effect) => {
@@ -190,12 +190,13 @@ discovery.run((effect) => {
 		for (;;) {
 			const entry = await Promise.race([effect.cancel, announced.next()]);
 			if (!entry) break;
+			if (entry.kind === "live") continue;
 			const path = entry.prefix;
 			// Only catalog-backed broadcasts are watchable streams; this skips the relay's
 			// `.stats` broadcast (see the stats dashboard demo for that one).
 			if (!path.endsWith(".hang") && !path.endsWith(".msf")) continue;
-			if (Net.Announce.isActive(entry.kind)) live.add(path);
-			else live.delete(path);
+			if (entry.kind === "retracted") live.delete(path);
+			else live.add(path);
 			broadcasts.set([...live].sort());
 		}
 	});

demo/web/src/stats.ts

@@ -131,11 +131,12 @@ discovery.run((effect) => {
 		for (;;) {
 			const entry = await Promise.race([effect.cancel, announced.next()]);
 			if (!entry) break;
+			if (entry.kind === "live") continue;
 			const path = entry.prefix;
 			const node = Net.Path.stripPrefix(prefix, path);
 			if (!node) continue;
 
-			if (Net.Announce.isActive(entry.kind)) {
+			if (entry.kind !== "retracted") {
 				if (subs.has(node)) continue;
 				const ne = new Signals.Effect();
 				subs.set(node, ne);

doc/concept/moq-lite.md

@@ -139,16 +139,16 @@ root. The pattern scope filters which prefixes are visible without changing a
 route's prefix. Announce events carry the covered path, captures, and what
 happened to it: Rust `announce::Event::{Announced, Updated, Retracted}`, each
 holding an `announce::Announce { prefix, captures: Option<Vec<Pattern>>, route }`,
-and TypeScript `Announce.Update { prefix, captures, route, kind }`, where the kind
-is announced, updated (a reprice in place), or retracted. Captures are present
-when the announced prefix pins every wildcard in the most-specific matching
-scope member. The Rust consumer is a `Stream` and the TypeScript one an async
-iterable. The Rust consumer, and every binding over moq-ffi or moq-c, also
-yields one `announce::Event::Live` once the
-routes live at subscribe time have all been delivered, including those a peer
-session was still sending: moq-lite-05+ counts them in `ANNOUNCE_OK`,
-moq-lite-01/02 send them in `ANNOUNCE_INIT`, IETF draft-16+ sessions count them
-in `REQUEST_OK` when both sides speak
+and TypeScript `Announce.Event`, whose `kind` is `"announced"`, `"updated"`, or
+`"retracted"` alongside the same `Announce.Announce` fields. An update is a
+reprice in place. Captures are present when the announced prefix pins every
+wildcard in the most-specific matching scope member. The Rust consumer is a
+`Stream` and the TypeScript one an async iterable. Both, and every binding over
+moq-ffi or moq-c, also yield one `Live` marker (TypeScript `{ kind: "live" }`)
+once the routes live at subscribe time have all been delivered, including those
+a peer session was still sending: moq-lite-05+ counts them in `ANNOUNCE_OK`,
+moq-lite-01/02 send them in `ANNOUNCE_INIT`, Rust IETF draft-16+ sessions count
+them in `REQUEST_OK` when both sides speak
 [active-count](/draft/moq-active-count), and anything else waits for the
 stream to go quiet.
 

doc/lib/js/net.md

@@ -50,7 +50,7 @@ for (;;) {
 - **Origins** hold the broadcasts, not the connection: closing a session unannounces them but leaves them created for the next one. `origin.request(path)` resolves an announced local broadcast with no round trip, so a page that watches what it publishes reads its own copy, unless a cheaper route announces the same path. A refusal from the most specific route ends the request: `request.closed` settles with the handler's error and no broader route is asked. Create, populate, then `announce()` for an exact path; use `dynamic(prefix, route)` when the set of paths is not known: an exact-path subscribe before the tracks exist is refused, and nobody, local or remote, can see or reach a broadcast until it announces.
 - **Connections** race WebTransport against WebSocket. `new Connection({ url })` pools one connection per relay URL and reconnects with backoff, which the elements use. Supplying WebTransport/WebSocket options, discovery, delay, or a caller-owned origin selects a private loop; explicit `share: true` refuses those options. `closed` settles when the handle is released (`null` on a clean close); the failure that stopped retrying the current URL is `error`, and a new URL recovers the same handle. A connection owns one send-rate sampler and one `Bandwidth.Allocator`; publishers reserve against it so their encoder targets sum to the estimate instead of each matching it.
 - **Bandwidth** (`Bandwidth.Allocator`) divides the connection's send-rate estimate by track priority, max-min fair within a tier. An idle track claims nothing. The receive side is untouched.
-- **Discovery** by any pattern scope (`origin.announced(scope)`, such as `room/*/chat`; default everything). Each event's `prefix` is the covered prefix relative to the origin, `captures` reports what the scope's wildcards matched when the prefix pins them, and `kind` says whether it was announced, updated, or retracted. The consumer is an async iterable. `origin.broadcasts(scope)` is a live `Getter<ReadonlyMap<Path.Valid, Route>>` of the same covered prefixes for UIs that need the current set. A borrowed `Connection.origin` also exposes `dynamic(prefix, route)` for serving paths on demand.
+- **Discovery** by any pattern scope (`origin.announced(scope)`, such as `room/*/chat`; default everything). A route event's `kind` says whether it was announced, updated, or retracted; its `prefix` is the covered prefix relative to the origin, and `captures` reports what the scope's wildcards matched when the prefix pins them. One `{ kind: "live" }` marker, with no prefix, follows the routes live at subscribe time. The consumer is an async iterable. `origin.broadcasts(scope)` is a live `Getter<ReadonlyMap<Path.Valid, Route>>` of the same covered prefixes for UIs that need the current set. A borrowed `Connection.origin` also exposes `dynamic(prefix, route)` for serving paths on demand.
 - **Publisher retention** is optional `Track.Info.maxAge`: omission sets no limit, while zero keeps the live edge. No implicit five-second window is applied.
 - **Subscriptions** carry a priority, a `Time.Milli` max age, and optional `groups` bounds. Groups arrive out of order and are read frame by frame, with `Error.TooFarBehind` when a reader asks for a frame the group never held and `Error.GroupTooLarge` when a write exceeds the cache budget and aborts the group.
 - **Track ends**: `close()` ends a track at its live edge, while `finishAt(n)` declares the exclusive end ahead of it and still accepts the groups below. A subscriber reads the end with `final()` or awaits `finished()`. A remote track ends only once every group below its end has arrived or was dropped; one reset before its header arrived is skipped after the subscription's max age on moq-lite (one second without one), or after one second on IETF.
@@ -100,11 +100,14 @@ Three operations, on an origin:
   route is always a prefix on every wire.
 
 A route is a capability, not an inventory. `origin.announced(scope)` yields
-`Announce.Update` values: `prefix` is the covered prefix relative to the origin,
-`captures` is one pattern per scope wildcard when the prefix pins a complete
-match (otherwise `undefined`), `kind` is `"announced"`, `"updated"` (a
-reprice in place), or `"retracted"`, and `route` carries hops and cost (on a
-retraction, its last values). The consumer is an async iterable. A prefix is
+`Announce.Event` values. `kind` is `"announced"`, `"updated"` (a reprice in
+place), or `"retracted"`, each also carrying an `Announce.Announce`: `prefix`
+is the covered prefix relative to the origin, `captures` is one pattern per
+scope wildcard when the prefix pins a complete match (otherwise `undefined`),
+and `route` carries hops and cost (on a retraction, its last values). A single
+`{ kind: "live" }` follows the routes live at subscribe time, including every
+route a connected peer was still sending, so a caller listing what is live
+stops there. The consumer is an async iterable. A prefix is
 not a broadcast name; the scope filters locally while sessions request its
 literal head on the wire. Paths with a `.`-prefixed segment below that head
 are [hidden](/concept/moq-lite#hidden-broadcasts) unless `announced(scope, { hidden: true })` opts in;

js/moq-boy/src/element.tsx

@@ -137,13 +137,14 @@ export default class MoqBoy extends HTMLElement {
 			for (;;) {
 				const entry = await effect.race(announced.next());
 				if (!entry) break;
+				if (entry.kind === "live") continue;
 
 				// A broad route that cannot pin the game id names nothing to open.
 				const capture = entry.captures?.[0];
 				if (!capture?.isLiteral) continue;
 
 				const id = capture.text;
-				if (Moq.Announce.isActive(entry.kind) && !this.#sessions.has(id)) {
+				if (entry.kind !== "retracted" && !this.#sessions.has(id)) {
 					const config: GameConfig = {
 						sessionId: id,
 						connection: this.connection,

js/net/examples/discovery.ts

@@ -10,6 +10,10 @@ async function main() {
 
 	// Discover broadcasts announced by the server
 	for await (const announcement of announced) {
+		if (announcement.kind === "live") {
+			console.log("Caught up: everything live has been listed");
+			continue;
+		}
 		if (announcement.kind === "retracted") continue;
 		console.log("New stream available:", announcement.prefix);
 

js/net/src/announce.ts

@@ -3,4 +3,4 @@
  *
  * @module
  */
-export { Consumer, isActive, type Kind, type Update } from "./announced.ts";
+export { type Announce, Consumer, type Event } from "./announced.ts";

js/net/src/announced.test.ts

@@ -57,6 +57,22 @@ test("a same-name re-announce is a distinct update", async () => {
 	expect(await consumer.next()).toEqual({ prefix: p("a"), captures: undefined, kind: "announced", route });
 });
 
+test("the live marker is delivered once", async () => {
+	const producer = new Announce.Producer();
+	const consumer = producer.consume();
+
+	// A stream spanning sessions forwards each one's marker; only the first is news.
+	const route = Route.default;
+	producer.append({ kind: "live" });
+	producer.append({ prefix: p("a"), captures: undefined, kind: "announced", route });
+	producer.append({ kind: "live" });
+	producer.append({ prefix: p("b"), captures: undefined, kind: "announced", route });
+
+	expect(await consumer.next()).toEqual({ kind: "live" });
+	expect(await consumer.next()).toMatchObject({ prefix: p("a") });
+	expect(await consumer.next()).toMatchObject({ prefix: p("b") });
+});
+
 test("closing resolves next with undefined", async () => {
 	const producer = new Announce.Producer();
 	const consumer = producer.consume();

js/net/src/announced.ts

@@ -8,14 +8,7 @@ import type { Route } from "./hop.js";
 import type * as Path from "./path.js";
 
 /**
- * What an {@link Update} reports about its prefix.
- *
- * @public
- */
-export type Kind = "announced" | "updated" | "retracted";
-
-/**
- * A route announcement, update, or retraction.
+ * A route over a prefix, delivered inside an {@link Event}.
  *
  * An announcement is always a prefix, never a broadcast: a route claims that
  * {@link prefix} and every path beneath it can be served. By convention a publisher
@@ -25,19 +18,30 @@ export type Kind = "announced" | "updated" | "retracted";
  *
  * @public
  */
-export interface Update {
+export interface Announce {
 	/**
 	 * The prefix the route covers, relative to the origin (for a session, its URL path).
 	 */
 	prefix: Path.Valid;
 	/** What the filter's wildcards stood for, when this prefix pins all of them. */
 	captures: Path.Pattern[] | undefined;
-	/** Whether the prefix was announced, re-priced, or retracted. */
-	kind: Kind;
 	/** Hops and cost of the route; on a retraction, its last advertised values. */
 	route: Route;
 }
 
+/**
+ * What an announcement stream yields.
+ *
+ * `announced`: a route now covers the prefix. `updated`: the route covering it changed
+ * hops or cost, in place. `retracted`: no route covers it any more. `live`: every route
+ * live at subscribe time has been delivered, including those a connected peer was still
+ * sending, so what follows is live changes; yielded at most once, and a caller listing
+ * what is live stops there.
+ *
+ * @public
+ */
+export type Event = ({ kind: "announced" | "updated" | "retracted" } & Announce) | { kind: "live" };
+
 /**
  * Options for an announcement stream.
  *
@@ -53,14 +57,11 @@ export interface Options {
 	hidden?: boolean;
 }
 
-/** Whether a route covers the path after an update of this {@link Kind}. */
-export function isActive(kind: Kind): boolean {
-	return kind !== "retracted";
-}
-
 /** Reactive backing state shared by announcement producers and consumers. */
 class AnnounceState {
-	queue = new Signal<Update[]>([]);
+	queue = new Signal<Event[]>([]);
+	// Whether the live marker was appended, so a repeat (from a stream spanning sessions) is dropped.
+	live = false;
 	closed = new Once<Error | null>();
 }
 
@@ -94,11 +95,15 @@ export class Producer {
 		return makeConsumer(this.#state);
 	}
 
-	/** Writes an announcement to the queue. */
-	append(update: Update) {
+	/** Writes an event to the queue. The `live` marker is written once; later ones are dropped. */
+	append(event: Event) {
 		if (this.#state.closed.peek() !== undefined) throw new Error("announcements are closed");
+		if (event.kind === "live") {
+			if (this.#state.live) return;
+			this.#state.live = true;
+		}
 		this.#state.queue.mutate((queue) => {
-			queue.push(update);
+			queue.push(event);
 		});
 	}
 
@@ -136,17 +141,17 @@ export class Consumer {
 		makeConsumer = (state) => new Consumer(state);
 	}
 
-	/** The announcements as they arrive, until the stream closes. */
-	async *[Symbol.asyncIterator](): AsyncGenerator<Update, void, undefined> {
+	/** The events as they arrive, until the stream closes. */
+	async *[Symbol.asyncIterator](): AsyncGenerator<Event, void, undefined> {
 		for (;;) {
-			const update = await this.next();
-			if (!update) return;
-			yield update;
+			const event = await this.next();
+			if (!event) return;
+			yield event;
 		}
 	}
 
-	/** Returns the next announcement. */
-	async next(): Promise<Update | undefined> {
+	/** Returns the next event, or undefined once the stream closes. */
+	async next(): Promise<Event | undefined> {
 		for (;;) {
 			const announce = this.#state.queue.peek().shift();
 			if (announce) return announce;
@@ -164,3 +169,46 @@ export class Consumer {
 		closeState(this.#state, abort);
 	}
 }
+
+/**
+ * When an initial set has landed, for wires that never say where it ends (lite-03/04,
+ * moq-transport): once its stream goes quiet.
+ *
+ * A peer writes its whole set back to back, so the first announcement gets a round trip's
+ * grace and each one after it only has to beat its siblings. Mirrors `Quiet` in `rs/moq-net`.
+ *
+ * @internal
+ */
+export class Quiet {
+	/** How long the stream may stay silent before its first announcement, in ms. */
+	static readonly FIRST = 500;
+	/** How long the stream may stay silent between announcements, in ms. */
+	static readonly GAP = 30;
+
+	#landed: () => void;
+	#timer: ReturnType<typeof setTimeout> | undefined;
+
+	/** Start counting from now; `landed` runs once the stream goes quiet. */
+	constructor(landed: () => void) {
+		this.#landed = landed;
+		this.#timer = setTimeout(() => this.#land(), Quiet.FIRST);
+	}
+
+	/** An announcement arrived: the set is still landing. */
+	heard(): void {
+		if (this.#timer === undefined) return;
+		clearTimeout(this.#timer);
+		this.#timer = setTimeout(() => this.#land(), Quiet.GAP);
+	}
+
+	/** Stop counting without landing, once the stream is gone. Idempotent. */
+	close(): void {
+		clearTimeout(this.#timer);
+		this.#timer = undefined;
+	}
+
+	#land(): void {
+		this.#timer = undefined;
+		this.#landed();
+	}
+}

js/net/src/connection/established.ts

@@ -37,7 +37,8 @@ export interface Established {
 	 * Subscribe to broadcast announcements matching `scope`, any pattern (`foo/**`
 	 * for a subtree, `room/* /chat` for each room's chat, default `**`). Paths are
 	 * relative to the session; captures report what the scope's wildcards stood for.
-	 * Hidden routes are left out unless `options.hidden` opts in.
+	 * Hidden routes are left out unless `options.hidden` opts in. The `live` marker follows
+	 * once the peer's initial set has landed.
 	 */
 	announced(scope?: Path.Pattern, options?: announce.Options): announce.Consumer;
 

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

@@ -189,6 +189,55 @@ test("a request replaced across one coalesced wakeup still gets answered", async
 	origin.close();
 });
 
+/** Whether `next` is still waiting once everything queued has run. */
+async function pending(next: Promise<unknown>): Promise<boolean> {
+	const waiting = Symbol("waiting");
+	const result = await Promise.race([next, settle().then(() => waiting)]);
+	return result === waiting;
+}
+
+test("the origin's live marker waits for the session's initial set", async () => {
+	const origin = new OriginProducer();
+	const session = new FakeSession();
+	forwardAnnounced(session.session, origin);
+
+	const announced = origin.announced();
+	session.announces.append({ prefix: Path.from("a"), captures: undefined, kind: "announced", route: Route.default });
+	expect(await announced.next()).toMatchObject({ prefix: Path.from("a"), kind: "announced" });
+
+	// The peer has not said its initial set is complete.
+	const next = announced.next();
+	expect(await pending(next)).toBe(true);
+
+	session.announces.append({ kind: "live" });
+	expect(await next).toEqual({ kind: "live" });
+
+	// A stream opened after the set landed does not wait on it.
+	const later = origin.announced();
+	expect(await later.next()).toMatchObject({ prefix: Path.from("a"), kind: "announced" });
+	expect(await later.next()).toEqual({ kind: "live" });
+
+	announced.close();
+	later.close();
+	origin.close();
+});
+
+test("a session dying before its initial set lands releases the marker", async () => {
+	const origin = new OriginProducer();
+	const session = new FakeSession();
+	forwardAnnounced(session.session, origin);
+
+	const announced = origin.announced();
+	const next = announced.next();
+	expect(await pending(next)).toBe(true);
+
+	session.die();
+	expect(await next).toEqual({ kind: "live" });
+
+	announced.close();
+	origin.close();
+});
+
 test("scoped origins subscribe to disjoint literal heads with hidden routes", () => {
 	const origin = new OriginProducer().scope(
 		Path.empty(),