feat(apps): players follow announce Start, Restart, and End

demo/web/src/stats.ts

@@ -136,22 +136,26 @@ discovery.run((effect) => {
 			const node = Net.Path.stripPrefix(prefix, path);
 			if (!node) continue;
 
-			if (entry.kind !== "end") {
-				if (subs.has(node)) continue;
-				const ne = new Signals.Effect();
-				subs.set(node, ne);
-				subscribeNode(ne, origin, path, node);
-			} else {
+			// The same node re-priced or failed over: its counters carry on.
+			if (entry.kind === "update") continue;
+			if (entry.kind === "start" && subs.has(node)) continue;
+
+			// An end, or a restart: another instance at the path counts from zero, so neither
+			// series may splice across it.
+			if (entry.kind !== "start") {
 				subs.get(node)?.close();
 				subs.delete(node);
 				nodeStats.mutate((s) => {
 					delete s[node];
 				});
-				// A restart ends the old epoch and starts a new one at the same path,
-				// counting from zero, so neither series may splice across it.
 				history.delete(node);
 				clusterMembership = "";
 			}
+			if (entry.kind === "end") continue;
+
+			const ne = new Signals.Effect();
+			subs.set(node, ne);
+			subscribeNode(ne, origin, path, node);
 		}
 	});
 });

doc/bin/cli.md

@@ -127,8 +127,17 @@ default, `auto`, sizes the buffer to how unevenly audio arrives, with the same
 [algorithm](/concept/audio-jitter) as the browser player. A duration fixes the
 delay and is also how long a stalled group is waited on before it is skipped.
 Each role follows the catalog, switching rendition when the publisher retires
-the one playing. Playback is behind the `play` feature, since it pulls in
-windowing and audio-device dependencies:
+the one playing.
+
+The broadcast's announcement is its online signal, followed until the window
+closes. Playback starts when the name is announced. A
+[restarted publisher](#publisher-runs) starts it over on the new run, with a
+fresh catalog, decoders, and clock. When the announcement ends, what is playing
+finishes and the player waits for the name to return. The log names each run's
+epoch. A catalog with nothing this build can play still ends the player.
+
+Playback is behind the `play` feature, since it pulls in windowing and
+audio-device dependencies:
 
 ```bash
 cargo install moq-cli --no-default-features --features "iroh,noq,websocket,play"

doc/lib/js/net.md

@@ -35,7 +35,7 @@ broadcast.announce();
 
 - **The origin holds the broadcasts, not the connection.** Closing a session unannounces them and leaves them created for the next one. A broadcast is invisible, locally and remotely, until `announce()`. `dynamic(prefix, route)` claims a prefix and yields each requested path to accept or reject. `accept` throws on a broadcast a session delivered, since JS apps do not proxy; copy its tracks into a broadcast you produce instead.
 - **Requests can pin an epoch.** `consumer.request(path, { epoch })` resolves only through a route announcing that [publisher epoch](/concept/moq-lite#publisher-epochs) and reports `unroutable` otherwise. The resolved broadcast's `epoch` names the route it came through, pinned or not.
-- **Follow restarts.** An announce event of kind `"restart"` says another publisher instance now serves the prefix: request the path again. A request already resolved stays on the old instance, and ends with an `unroutable` error once that instance stops serving: it never moves onto another instance, since without a shared epoch nothing says the new one holds the same bytes. A blind answer (a session without discovery) ends with its session for the same reason, so only announcements can trigger following a restart.
+- **Follow restarts.** `origin.follow(path)` reduces the routes covering a path to the one a request resolves (the most specific) and streams its `start`, `update`, `restart`, and `end`. A `"restart"` says another publisher instance now serves the path: request it again. A request already resolved stays on the old instance, and ends with an `unroutable` error once that instance stops serving: it never moves onto another instance, since without a shared epoch nothing says the new one holds the same bytes. A blind answer (a session without discovery) ends with its session for the same reason, so only announcements can trigger following a restart.
 - **One connection per URL, unless you opt out.** `new Connection({ url })` pools and reconnects with backoff, which the elements use. Supplying your own transport options, discovery, or origin selects a private loop.
 - **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: a request on the old session ends when that session closes, so request the path again. 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.

doc/lib/js/watch.md

@@ -47,6 +47,14 @@ buffering indicator, and a stats panel.
 `error` the origin refused the broadcast, `el.broadcast.out.error` says why,
 and the player does not ask again until it is pointed elsewhere or re-enabled.
 
+The broadcast's announcement is its online signal: the player starts when a
+route serving the name is announced, goes `offline` when none is left, and
+waits for one to return for as long as it is mounted. When the exact route
+ends while a covering prefix still serves the name, it moves to the prefix. A
+name outside the origin's scope is an `error`. A restarted publisher at the same name is followed at
+once, with a fresh catalog, and video, audio, and the clock start over on its
+timeline instead of waiting for it to catch up to the old one.
+
 ## Binding from a framework
 
 `import "@moq/watch/element"` registers `<moq-watch>` when the module

doc/lib/rs/moq-net.md

@@ -17,7 +17,7 @@ What you use it for, beyond what the concept page already describes:
 
 - **An origin outlives the session.** Broadcasts are created on the origin, and a reconnect announces them again. Closing the session does not delete them.
 - **Requests can pin an epoch.** `consumer.request_broadcast(path, Some(epoch))` resolves only through a route announcing that [publisher epoch](/concept/moq-lite#publisher-epochs); `None` takes whichever route wins. The resolved consumer names its epoch in `info().epoch`, pinned or not.
-- **Follow restarts.** `AnnounceEvent::Restart` says another publisher instance now serves the prefix: request the path again. A broadcast already resolved stays on the old instance until dropped or its route goes.
+- **Follow restarts.** `consumer.follow(path)` reduces the routes covering a path to the one a request resolves (the most specific) and yields its start, update, restart, and end. A restart says another publisher instance now serves the path: request it again. A broadcast already resolved stays on the old instance until dropped or its route goes.
 - **One track per name.** Concurrent subscriptions share one request and producer. A producer that replaces an ended one continues the name's group and datagram sequences; only a new broadcast restarts them. Readers of an aborted track resume onto the replacement only when the broadcast announces an epoch; otherwise their subscription ends with the error.
 - **Publish only while someone is watching.** `demand()` on a track, group, or broadcast says whether a subscriber is attached, which is how capture and transcode skip work nobody asked for. Holding a broadcast consumer is not demand. Holding a track consumer is, so drop one you are not reading. A shared fetch stays up until its last reader leaves.
 - **You drive the session, or `moq-tokio` does.** `connect` and `accept` return a session plus a driver that never reads the clock itself. `moq_net::time::run` polls it on tokio or in the browser. `moq-tokio` and `moq-wasm` hide that. A custom transport implements `moq_net::transport::poll`.

js/net/src/announced.ts

@@ -5,7 +5,7 @@
  */
 import { type GetPromise, Once, Signal } from "@moq/signals";
 import type { Route } from "./hop.js";
-import type * as Path from "./path.js";
+import * as Path from "./path.js";
 
 /**
  * A route over a prefix, delivered inside an {@link Event}.
@@ -163,3 +163,64 @@ export class Consumer {
 		closeState(this.#state, abort);
 	}
 }
+
+/**
+ * A reducer from the announcements of the routes covering `path` to those of the one route
+ * serving it: the most specific. Another route taking over is a `restart`, or an `update` when
+ * both carry the same epoch, since those serve the same bytes. Routes beneath `path` serve
+ * other broadcasts and are skipped, as is any change no request sees.
+ *
+ * While nothing serves the path, the rest of a batch (one change to the table) folds into one
+ * `start`, so the replay a late follower begins with (a covering prefix, then the exact path
+ * beneath it) starts on the route serving the path rather than starting and restarting. Once
+ * something serves it, each change comes through on its own: an `end` followed by a `start` is
+ * a gap that ended any request on the old route, even when both routes carry one epoch.
+ * Synchronous, so a followed stream delivers in step with an announced one. Mirrors
+ * `announce::Follow` in rs/moq-net.
+ *
+ * @internal
+ */
+export function follower(path: Path.Valid): (events: Event[]) => Event[] {
+	// Every route standing over the path, by prefix.
+	const covering = new Map<Path.Valid, Announce>();
+	const serving = () => {
+		let best: Announce | undefined;
+		for (const announce of covering.values()) {
+			if (!best || announce.prefix.length > best.prefix.length) best = announce;
+		}
+		return best;
+	};
+
+	// Apply one route's event, returning what it means for the path alone.
+	const fold = ({ kind, ...announce }: Event): Event | undefined => {
+		if (!Path.hasPrefix(announce.prefix, path)) return undefined;
+		const before = serving();
+		if (kind === "end") covering.delete(announce.prefix);
+		else covering.set(announce.prefix, announce);
+		const after = serving();
+
+		if (!before) return after && { kind: "start", ...after };
+		if (!after) return { kind: "end", ...before };
+		if (before.prefix !== after.prefix) {
+			const same = before.route.epoch !== undefined && before.route.epoch === after.route.epoch;
+			return { kind: same ? "update" : "restart", ...after };
+		}
+		// A route less specific than the serving one changing is skipped: no request sees it.
+		if (after.prefix !== announce.prefix) return undefined;
+		return { kind: kind === "restart" ? "restart" : "update", ...after };
+	};
+
+	return (events) => {
+		const reduced: Event[] = [];
+		let idle = !serving();
+		for (const event of events) {
+			const change = fold(event);
+			if (idle || !change) continue;
+			reduced.push(change);
+			idle = change.kind === "end";
+		}
+		const after = serving();
+		if (idle && after) reduced.push({ kind: "start", ...after });
+		return reduced;
+	};
+}

js/net/src/lite/publisher.ts

@@ -459,7 +459,7 @@ export class Publisher {
 		const onWire = (route: Route): Route => (hasRouteCost(this.version) ? route : { ...route, cost: Cost.zero });
 
 		const announce = async (suffix: Path.Valid, route: Route) => {
-			console.debug(`announce: broadcast=${suffix} active=true`);
+			console.debug(`announce: broadcast=${suffix} active=true epoch=${route.epoch}`);
 			if (hasAnnounceId(this.version)) announceIds.set(suffix, nextAnnounceId++);
 			await encodeAnnounceBroadcast(
 				stream.writer,
@@ -495,7 +495,7 @@ export class Publisher {
 				await announce(suffix, route);
 				return;
 			}
-			console.debug(`announce: broadcast=${suffix} restart=true`);
+			console.debug(`announce: broadcast=${suffix} restart=true epoch=${route.epoch}`);
 			await encodeAnnounceBroadcast(
 				stream.writer,
 				{ status: "restart", id, epoch: route.epoch, hops: wireHops(route), cost: route.cost },

js/net/src/lite/subscriber.ts

@@ -510,7 +510,7 @@ export class Subscriber {
 					advertised.set(path, { live: true, route, captures });
 					if (epoch) this.#epochs.set(path, epoch);
 					else this.#epochs.delete(path);
-					console.debug(`announced: broadcast=${path} restart=true`);
+					console.debug(`announced: broadcast=${path} restart=true epoch=${epoch}`);
 					announced.append({ prefix: path, captures, kind: "restart", route });
 					continue;
 				}
@@ -533,7 +533,7 @@ export class Subscriber {
 				if (epoch) this.#epochs.set(path, epoch);
 				else this.#epochs.delete(path);
 
-				console.debug(`announced: broadcast=${path} active=true`);
+				console.debug(`announced: broadcast=${path} active=true epoch=${epoch}`);
 				announced.append({ prefix: path, captures, kind: "start", route });
 			}
 

js/net/src/origin.test.ts

@@ -331,6 +331,138 @@ test("announced streams the table under a scope with origin-relative paths", asy
 	a.close();
 });
 
+test("follow reports the route serving the path", async () => {
+	const origin = new Producer();
+	const follow = origin.consume().follow(Path.from("pool/job"));
+	const epoch = Epoch.mint();
+	const next = async () => {
+		const event = await follow.next();
+		return event && { kind: event.kind, prefix: event.prefix };
+	};
+
+	// A prefix above the path covers it.
+	const pool = origin.dynamic(Path.from("pool"), { epoch });
+	expect(await next()).toEqual({ kind: "start", prefix: Path.from("pool") });
+
+	// A route beneath the path serves another broadcast, so the next event is the exact path's.
+	const beneath = publish(origin, Path.from("pool/job/thumbnail"));
+
+	// The path itself, from the same instance, takes over without a restart.
+	const exact = origin.createBroadcast(Path.from("pool/job"));
+	exact.announce({ epoch });
+	expect(await next()).toEqual({ kind: "update", prefix: Path.from("pool/job") });
+
+	// The covering prefix no longer serves the path, so its re-price is not seen: the next
+	// event is another instance at the path.
+	pool.update({ ...pool.route, cost: 5n });
+	exact.announce({ epoch: Epoch.mint() });
+	expect(await next()).toEqual({ kind: "restart", prefix: Path.from("pool/job") });
+
+	// It goes, so the prefix serves the path again: another instance.
+	exact.close();
+	expect(await next()).toEqual({ kind: "restart", prefix: Path.from("pool") });
+
+	pool.update({ ...pool.route, cost: 9n });
+	expect(await next()).toEqual({ kind: "update", prefix: Path.from("pool") });
+
+	pool.close();
+	expect(await next()).toEqual({ kind: "end", prefix: Path.from("pool") });
+
+	// The stream ends with the origin.
+	origin.close();
+	expect(await follow.next()).toBeUndefined();
+	beneath.close();
+});
+
+test("follow restarts onto a more specific route without an epoch", async () => {
+	const origin = new Producer();
+	const follow = origin.consume().follow(Path.from("pool/job"));
+
+	const pool = origin.dynamic(Path.from("pool"));
+	expect(await follow.next()).toMatchObject({ kind: "start", prefix: Path.from("pool") });
+
+	const exact = publish(origin, Path.from("pool/job"));
+	expect(await follow.next()).toMatchObject({ kind: "restart", prefix: Path.from("pool/job") });
+
+	follow.close();
+	exact.close();
+	pool.close();
+	origin.close();
+});
+
+test("follow ignores a route scoped beneath the path", async () => {
+	const origin = new Producer();
+	// A cheaper dynamic at the same prefix that claims only `room/*/chat`, so it never serves the path.
+	const chat = origin.scope(Path.empty(), new Path.Patterns([Path.Pattern.parse("room/*/chat")]));
+	const dynamic = chat.dynamic(Path.from("room/alice"));
+	const follow = origin.consume().follow(Path.from("room/alice"));
+
+	const first = origin.createBroadcast(Path.from("room/alice"));
+	first.announce({ cost: 5n });
+	expect(await follow.next()).toMatchObject({ kind: "start", route: { cost: 5n } });
+
+	// A republish is another instance of the route that serves the path.
+	const second = origin.createBroadcast(Path.from("room/alice"));
+	second.announce({ cost: 5n });
+	first.close();
+	expect(await follow.next()).toMatchObject({ kind: "restart", route: { cost: 5n } });
+
+	follow.close();
+	second.close();
+	dynamic.close();
+	origin.close();
+});
+
+test("follow starts on the serving route when joining late", async () => {
+	const origin = new Producer();
+	const pool = origin.dynamic(Path.from("pool"));
+	const exact = publish(origin, Path.from("pool/job"));
+
+	const follow = origin.consume().follow(Path.from("pool/job"));
+	expect(await follow.next()).toMatchObject({ kind: "start", prefix: Path.from("pool/job") });
+
+	// Nothing else is queued: the next event is the exact route ending, a restart onto the prefix.
+	exact.close();
+	expect(await follow.next()).toMatchObject({ kind: "restart", prefix: Path.from("pool") });
+
+	follow.close();
+	pool.close();
+	origin.close();
+});
+
+test("follow keeps a gap between routes of one epoch", async () => {
+	const origin = new Producer();
+	const epoch = Epoch.mint();
+	const exact = origin.createBroadcast(Path.from("pool/job"));
+	exact.announce({ epoch });
+	const follow = origin.consume().follow(Path.from("pool/job"));
+	expect(await follow.next()).toMatchObject({ kind: "start", prefix: Path.from("pool/job") });
+
+	// Both changes land before the table is read again, but the path went unserved in between,
+	// which ended any request on it: a start of its own, not an update.
+	exact.close();
+	const pool = origin.dynamic(Path.from("pool"), { epoch });
+	expect(await follow.next()).toMatchObject({ kind: "end", prefix: Path.from("pool/job") });
+	expect(await follow.next()).toMatchObject({ kind: "start", prefix: Path.from("pool") });
+
+	follow.close();
+	pool.close();
+	origin.close();
+});
+
+test("follow refuses a path no pattern can spell", () => {
+	const origin = new Producer();
+	expect(() => origin.consume().follow(Path.from("camera*main"))).toThrow();
+	origin.close();
+});
+
+test("follow refuses a path outside the scope", () => {
+	const origin = new Producer();
+	const scoped = origin.scope(Path.empty(), new Path.Patterns([Path.Pattern.subtree(Path.from("pool/job/cam"))]));
+	expect(() => scoped.consume().follow(Path.from("pool/job"))).toThrow();
+	origin.close();
+});
+
 test("hidden paths need an opt-in or a scope naming the dot segment", async () => {
 	const origin = new Producer();
 	const consumer = origin.consume();