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,41 @@
+import * as Moq from "@moq/net";
+import { Effect } from "@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, then release the handle and the connection.
+	await connection.closed;
+	effect.close();
+	broadcast.close();
+}
+
+main().catch(console.error);

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,116 @@ 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>();
+
+/**
+ * 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.
+ *
+ * The handle re-consumes on every (re-)announce, so a same-name republish (a new publisher, or
+ * a relay-failover RESTART) re-attaches to the new instance rather than clinging to the dead
+ * one. Built from a reconnecting `Connection.Reload`, it 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.
+ *
+ * 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 `path` on a connection. Accepts a live {@link Established} session or a reactive
+	 * one (a `Connection.Reload`'s `established`), which is how the handle survives reconnects.
+	 *
+	 * Obtain one from `announcedBroadcast(path)` on either connection type rather than
+	 * constructing it directly.
+	 */
+	constructor(connection: GetterInit<Established | undefined>, path: Path.Valid) {
+		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);
+				return;
+			}
+
+			const announced = conn.announced(path);
+			effect.cleanup(() => announced.close());
+
+			let current: broadcast.Consumer | undefined;
+			const offline = () => {
+				current?.close();
+				current = undefined;
+				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) {
+					// The stream was reset, which means the session died under it.
+					console.debug("announcement stream reset", 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(this.established, path);
+	}
+
 	/**
 	 * Snapshot the live connection's transport counters, or undefined while disconnected.
 	 * See {@link Established.stats}.

js/net/src/ietf/connection.ts

@@ -1,5 +1,5 @@
 import { type Getter, Signal } from "@moq/signals";
-import type * as announce from "../announced.ts";
+import * as announce from "../announced.ts";
 import type * as broadcast from "../broadcast.ts";
 import type { Established } from "../connection/established.ts";
 import { type Probe, type Stats, transportStats } from "../connection/stats.ts";
@@ -174,6 +174,16 @@ export class Connection implements Established {
 		return this.#subscriber.consume(path);
 	}
 
+	/**
+	 * Watches a broadcast, live only while it is announced.
+	 *
+	 * @param path - The path of the broadcast to watch
+	 * @returns A reactive handle to the broadcast
+	 */
+	announcedBroadcast(path: Path.Valid): announce.Broadcast {
+		return new announce.Broadcast(this, path);
+	}
+
 	/**
 	 * Accepts bidi streams (virtual for v14-v16, real for v17) and dispatches.
 	 */