feat(net): add announcedBroadcast, and resolve lite restarts by publisher identity

js/net/README.md

@@ -60,6 +60,7 @@ await quicheLoaded; //This is a promise, connect after it resolves
 - **[Publishing](examples/publish.ts)** - Publish data to a broadcast
 - **[Subscribing](examples/subscribe.ts)** - Subscribe to and receive broadcast data
 - **[Discovery](examples/discovery.ts)** - Discover broadcasts announced by the server
+- **[Waiting](examples/wait.ts)** - Wait for one known broadcast to come online, and follow it
 - **[Server side usage](https://github.com/sb2702/webcodecs-examples/tree/main/src/moq-server)** - Publish from browser to a server
 
 ## License

js/net/examples/wait.ts

@@ -0,0 +1,48 @@
+import * as Moq from "@moq/net";
+
+// @moq/net re-exports the reactive primitives, so an app only needs the one dependency.
+const { Effect } = Moq.Signals;
+
+async function main() {
+	const url = new URL("https://cdn.moq.dev/anon");
+	const connection = new Moq.Connection.Reload({ url, enabled: true });
+
+	// Wait for a broadcast that may not exist yet. `consume` would subscribe blind and get reset
+	// if nobody is publishing the path; this waits for the announcement instead.
+	const broadcast = connection.announcedBroadcast(Moq.Path.from("my-broadcast"));
+
+	const effect = new Effect();
+	effect.run((effect) => {
+		// Re-runs every time the broadcast comes online or goes away, including across reconnects
+		// and same-name republishes.
+		const active = effect.get(broadcast.active);
+		if (!active) {
+			console.log("broadcast is offline");
+			return;
+		}
+
+		console.log("broadcast is live");
+		const track = active.track("chat").subscribe({ priority: 0 });
+		effect.cleanup(() => track.close());
+
+		effect.spawn(async () => {
+			for (;;) {
+				const group = await Promise.race([effect.cancel, track.recvGroup()]);
+				if (!group) break;
+				console.log("received:", await group.readString());
+			}
+		});
+	});
+
+	// Run until interrupted. `closed` rejects if the reconnect loop gives up, so release
+	// everything on the way out either way.
+	try {
+		await connection.closed;
+	} finally {
+		effect.close();
+		broadcast.close();
+		connection.close();
+	}
+}
+
+main().catch(console.error);

js/net/src/announced.test.ts

@@ -19,9 +19,9 @@ test("a same-name re-announce is a distinct update", async () => {
 	const producer = new Announce.Producer();
 	const consumer = producer.consume();
 
-	// A republish (a lite-06 RESTART, or an unannounce+announce that coalesces) arrives as a
-	// redundant active:true. The stream carries it as its own update, which is what lets a watcher
-	// notice the new instance even though membership never observably flipped.
+	// The stream is a log, not a set: it carries a redundant active:true as its own update rather
+	// than collapsing it. Deciding what a repeat means belongs to the session layer, which resolves
+	// a restart into either nothing (a route change) or an end + start (a new publisher).
 	producer.append({ path: p("a"), active: true });
 	producer.append({ path: p("a"), active: true });
 

js/net/src/announced.ts

@@ -3,7 +3,9 @@
  *
  * @module
  */
-import { type GetPromise, Once, Signal } from "@moq/signals";
+import { Effect, type GetPromise, type Getter, type GetterInit, getter, Once, Signal } from "@moq/signals";
+import type * as broadcast from "./broadcast.js";
+import type { Established } from "./connection/established.js";
 import * as Path from "./path.js";
 
 /**
@@ -126,3 +128,158 @@ export class Consumer {
 		closeState(this.#state, abort);
 	}
 }
+
+// Connections already warned about missing broadcast discovery, so the fallback logs at most
+// once per connection instead of once per watched path.
+const warnedNoDiscovery = new WeakSet<Established>();
+
+/**
+ * What to watch, for {@link Broadcast}.
+ *
+ * @public
+ */
+export interface BroadcastProps {
+	/**
+	 * The connection to watch on. Accepts a live {@link Established} session, or a reactive one
+	 * (a `Connection.Reload`'s `established`), which is how the handle survives reconnects.
+	 */
+	connection: GetterInit<Established | undefined>;
+
+	/** The broadcast path to watch. */
+	path: Path.Valid;
+}
+
+/**
+ * A reactive handle to a single broadcast: {@link Broadcast.active} holds a live
+ * {@link broadcast.Consumer} while the path is announced and `undefined` while nobody
+ * publishes it.
+ *
+ * Use this instead of {@link Established.consume} whenever the broadcast may not exist yet.
+ * Subscribing to a path nobody publishes gets the stream reset, so a consumer that races the
+ * publisher stays silent forever unless it retries; this waits for the announcement instead.
+ *
+ * A same-name republish re-consumes, so the handle attaches to the new instance rather than
+ * clinging to the dead one. A relay failover that keeps the same publisher does *not*: the
+ * subscription resumes across the new route, so `active` holds the same consumer throughout and
+ * never goes offline. Only a change of publisher produces an offline/online transition.
+ *
+ * Built from a reconnecting `Connection.Reload`, the handle also spans reconnects: the broadcast
+ * drops to `undefined` while disconnected and resolves again once the new connection announces it.
+ *
+ * Falls back to consuming blind (and warns once) on a relay without
+ * {@link Established.discovery}, where there is no announcement to wait for. `active` then
+ * means *assumed present* rather than known live: nothing reports whether the path exists, so
+ * a subscribe to a missing broadcast is how a caller finds out. The handle stays usable either
+ * way, and because it is scoped to the path rather than to one publisher, a subscribe made
+ * after a publisher finally appears succeeds.
+ *
+ * If discovery fails on a live session (the announcement stream is reset, or the relay
+ * refuses it) the handle goes offline and stays there: nothing reopens the stream on that
+ * connection. Build it from a `Connection.Reload` if you need it to recover, since a new
+ * connection starts a new stream.
+ *
+ * Close it to release the announcement stream and the current broadcast.
+ *
+ * @public
+ */
+export class Broadcast {
+	/** The broadcast path this handle watches. */
+	readonly path: Path.Valid;
+
+	/** The live broadcast, or `undefined` while it is offline. */
+	readonly active: Getter<broadcast.Consumer | undefined>;
+
+	#active = new Signal<broadcast.Consumer | undefined>(undefined);
+	#signals = new Effect();
+
+	/**
+	 * Watch a path on a connection.
+	 *
+	 * Prefer `announcedBroadcast(path)` on the connection itself. Reach for this when the
+	 * session you want to follow isn't either connection type, e.g. your own
+	 * `Getter<Established | undefined>`.
+	 */
+	constructor({ connection, path }: BroadcastProps) {
+		this.path = path;
+		this.active = this.#active;
+
+		const source = getter(connection);
+		this.#signals.run((effect) => {
+			const conn = effect.get(source);
+			if (!conn) return;
+
+			// Without discovery no announcement ever arrives, so waiting would hang forever.
+			if (!conn.discovery) {
+				if (!warnedNoDiscovery.has(conn)) {
+					warnedNoDiscovery.add(conn);
+					console.warn("relay does not support broadcast discovery; consuming without waiting.");
+				}
+
+				const blind = conn.consume(path);
+				effect.cleanup(() => blind.close());
+				effect.set(this.#active, blind, undefined);
+
+				// The announcement-gated path below goes offline when the stream ends with the
+				// session; without discovery there is no stream, so watch the session itself.
+				// A consumed broadcast is a path-scoped handle, not a subscription, so its own
+				// `closed` says nothing about whether the path exists or the session is alive.
+				// Raced against the run's teardown so a closed handle isn't retained until the
+				// session ends; the cleanup above has already cleared `active` in that case.
+				effect.spawn(async () => {
+					await Promise.race([effect.cancel, conn.closed]);
+					if (this.#active.peek() === blind) this.#active.set(undefined);
+				});
+				return;
+			}
+
+			const announced = conn.announced(path);
+			effect.cleanup(() => announced.close());
+
+			let current: broadcast.Consumer | undefined;
+			const offline = () => {
+				const mine = current;
+				current?.close();
+				current = undefined;
+				// Only clear what this run put there. A spawn task that resumes after its run was
+				// torn down would otherwise wipe the consumer a newer run already installed.
+				if (this.#active.peek() === mine) this.#active.set(undefined);
+			};
+			effect.cleanup(offline);
+
+			effect.spawn(async () => {
+				try {
+					for (;;) {
+						const event = await Promise.race([effect.cancel, announced.next()]);
+						if (!event) break;
+
+						// Scoped to `path`, so the exact broadcast arrives with an empty suffix; ignore children.
+						if (event.path !== Path.empty()) continue;
+
+						if (event.active) {
+							// A live subscription survives a redundant (re-)announce; only replace a dead one.
+							if (current && current.closed.peek() === undefined) continue;
+							current?.close();
+							current = conn.consume(path);
+							this.#active.set(current);
+						} else {
+							offline();
+						}
+					}
+				} catch (err) {
+					// Discovery failed: the session died under the stream, or the relay refused
+					// to answer. Nothing reopens it on this connection, so say so out loud.
+					console.warn("broadcast discovery failed", err);
+				}
+
+				// The stream ended, or this run was torn down (its cleanup already ran). Either
+				// way there is nothing left announcing the path, so don't hold a dead broadcast.
+				offline();
+			});
+		});
+	}
+
+	/** Closes the handle and the broadcast it currently holds. Idempotent. */
+	close() {
+		this.#signals.close();
+	}
+}

js/net/src/connection/established.ts

@@ -35,9 +35,23 @@ export interface Established {
 	/** Publish a broadcast at the given path. */
 	publish(path: Path.Valid, broadcast: broadcast.Producer): void;
 
-	/** Consume the broadcast at the given path. */
+	/**
+	 * Consume the broadcast at the given path, immediately.
+	 *
+	 * The subscription is reset if nobody publishes the path, so use
+	 * {@link announcedBroadcast} instead when the broadcast may not be online yet.
+	 */
 	consume(path: Path.Valid): broadcast.Consumer;
 
+	/**
+	 * A reactive handle to the broadcast at the given path, live only while it is announced.
+	 *
+	 * The announcement-gated counterpart to {@link consume}: it waits for the broadcast to come
+	 * online instead of resetting, and drops back to `undefined` when it goes away. See
+	 * {@link announce.Broadcast}. Close the handle when done.
+	 */
+	announcedBroadcast(path: Path.Valid): announce.Broadcast;
+
 	/**
 	 * Snapshot the transport's counters, querying it fresh on each call.
 	 *

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

@@ -1,6 +1,8 @@
 import { expect, test } from "bun:test";
+import { Producer as BroadcastProducer } from "../broadcast.ts";
 import * as Lite from "../lite/index.ts";
 import { createMockTransportPair } from "../mock.ts";
+import * as Path from "../path.ts";
 import { accept } from "./index.ts";
 import { Reload, type ReloadProps } from "./reload.ts";
 
@@ -115,3 +117,60 @@ test("a peer that severs immediately keeps escalating the backoff", async () =>
 		globalThis.WebTransport = original;
 	}
 });
+
+// Polls until `pred` holds, so a regression fails the test instead of hanging it.
+async function waitUntil(pred: () => boolean): Promise<void> {
+	for (let i = 0; i < 500; i++) {
+		if (pred()) return;
+		await settle();
+	}
+	throw new Error("timed out waiting for condition");
+}
+
+test("announcedBroadcast follows the reconnect loop", async () => {
+	const original = globalThis.WebTransport;
+	const url = new URL("https://example.com/");
+
+	// Every connect attempt gets a fresh session, whose server publishes the path once the
+	// handshake finishes. The client therefore always asks before the broadcast exists.
+	const sessions: { close: () => void }[] = [];
+	const published: BroadcastProducer[] = [];
+	const stub = function StubWebTransport() {
+		const pair = createMockTransportPair(Lite.ALPN_06_WIP);
+		void accept(pair.server, url).then((server) => {
+			sessions.push(server);
+			const broadcast = new BroadcastProducer();
+			published.push(broadcast);
+			server.publish(Path.from("late"), broadcast);
+		});
+		return pair.client;
+	};
+	globalThis.WebTransport = stub as unknown as typeof WebTransport;
+
+	const reload = new Reload({
+		enabled: true,
+		url,
+		websocket: { enabled: false },
+		delay: { initial: 10, multiplier: 1, max: 10 },
+	});
+	const watched = reload.announcedBroadcast(Path.from("late"));
+
+	try {
+		await waitUntil(() => watched.active.peek() !== undefined);
+		const first = watched.active.peek();
+
+		// The session dies: the handle drops the broadcast rather than clinging to a dead one.
+		sessions[0]?.close();
+		await waitUntil(() => watched.active.peek() === undefined);
+
+		// The reconnect re-announces it, and the handle re-consumes on the new session.
+		await waitUntil(() => watched.active.peek() !== undefined);
+		expect(watched.active.peek()).not.toBe(first);
+	} finally {
+		watched.close();
+		reload.close();
+		for (const broadcast of published) broadcast.close();
+		for (const session of sessions) session.close();
+		globalThis.WebTransport = original;
+	}
+});

js/net/src/connection/reload.ts

@@ -293,6 +293,21 @@ export class Reload {
 		return consumer;
 	}
 
+	/**
+	 * A reactive handle to one broadcast, spanning reconnects.
+	 *
+	 * The same {@link Announce.Broadcast} as {@link Established.announcedBroadcast}, but it
+	 * follows the reconnect loop: the broadcast drops to `undefined` when the connection dies
+	 * and resolves again once the new connection announces the path. Use it instead of
+	 * consuming off {@link Reload.established} whenever the broadcast may come online after you
+	 * do, which is exactly the case a blind `consume` loses.
+	 *
+	 * Close the handle when done; {@link Reload.close} only drops it to `undefined`.
+	 */
+	announcedBroadcast(path: Path.Valid): Announce.Broadcast {
+		return new Announce.Broadcast({ connection: this.established, path });
+	}
+
 	/**
 	 * Snapshot the live connection's transport counters, or undefined while disconnected.
 	 * See {@link Established.stats}.

js/net/src/consume.ts

@@ -40,4 +40,25 @@ export class BroadcastCache {
 
 		return consumer;
 	}
+
+	/**
+	 * Stop sharing the broadcast cached for `path`, so the next request subscribes fresh.
+	 *
+	 * Call when the path's advertisement goes away. 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 {
+		this.#cache.delete(path);
+	}
 }