feat(apps): follow a broadcast's announcement as its online signal and log epochs

doc/bin/cli.md

@@ -227,9 +227,19 @@ the newest cached group, including when a rendition is reopened, so playback
 does not replay the retained backlog. A publisher that retires the rendition
 being played ends that track and the role picks a replacement. A retired audio
 rendition plays out what the speaker holds while its replacement fills, so the
-switch does not cost a delay of silence. Playback is
-behind the `play` feature, since it pulls in windowing and audio-device
-dependencies:
+switch does not cost a delay of silence.
+
+The broadcast's announcement is its online signal. Playback starts when the
+name is announced and starts over, with a fresh catalog and decoders, on each
+new announcement: a restarted publisher's new
+[epoch](/concept/moq-lite#publisher-epochs), or over moq-lite 06, which carries
+no epoch, any change to the route. When the announcement ends, playback plays
+out what it holds and waits for the broadcast to come back, for as long as the
+window stays open. The log names each announced broadcast's epoch. A catalog
+with nothing this build can play still fails at once.
+
+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"
@@ -392,9 +402,13 @@ moq --connect https://relay.example.com/anon \
 
 Two publishers of the same broadcast name are interchangeable sources:
 relays hold both routes and fail over between them mid-group. They must
-produce identical tracks with aligned groups. A restarted encoder is the same
-broadcast too, so one whose groups restart from 0 must publish under a new
-name, or viewers wait for its sequence to catch up.
+produce identical tracks with aligned groups. Each run announces a fresh
+[epoch](/concept/moq-lite#publisher-epochs), logged at startup. Over
+moq-lite 07 (opt-in while its wire is in progress) the epoch reaches the relay,
+so a restarted publisher replaces the old broadcast and viewers switch to it at
+once. Older versions carry no epoch, so a restart that overlaps the old route
+reaches viewers only once that route goes: a clean stop retracts it at once,
+while a crashed publisher's route lasts until its session times out.
 
 ## Cluster
 

doc/lib/js/watch.md

@@ -44,6 +44,15 @@ Video holds its last picture while paused, out of view, or waiting for a
 resumed rendition's first frame. Its reported timestamp stays with that picture.
 Going offline or closing the player clears it.
 
+With `announced` on, the broadcast's announcement is its online signal. The
+player starts when the name is announced, goes `offline` when the announcement
+ends, and keeps waiting for it to come back for as long as it is mounted. A
+restarted publisher announces a new [epoch](/concept/moq-lite#publisher-epochs)
+at the same name, and the player follows it: 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. moq-lite 06 carries no epoch, so any change to the route
+covering the name may be a restart, and the player starts over on it too.
+
 The overlay adds play/pause, volume, fullscreen, a quality selector, a
 buffering indicator, an unsupported-codec warning, and a stats panel.
 `<moq-watch-support>` shows what the browser can play.
@@ -52,8 +61,9 @@ buffering indicator, an unsupported-codec warning, and a stats panel.
 `error` when the origin refuses the broadcast, such as a `dynamic()` handler
 rejecting the request, and `el.broadcast.out.error` then holds the refusal's
 `Error`; the overlay shows its message in place of the offline notice. A
-refusal is final: only a new `name`, a new origin, a changed `announced`, or
-re-enabling asks again, which clears both back to `offline`.
+refusal is final: only a new `name`, a new origin, a changed `announced`,
+re-enabling, or a change to a route without an epoch asks again, which clears
+both back to `offline`.
 
 ## Binding from a framework
 

js/net/src/lite/publisher.ts

@@ -430,7 +430,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,

js/net/src/lite/subscriber.ts

@@ -458,7 +458,7 @@ export class Subscriber {
 				advertised.set(path, { live: true, route, captures });
 				if (epoch) this.#epochs.set(path, epoch);
 
-				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/watch/src/audio/decoder.test.ts

@@ -362,12 +362,15 @@ describe("Decoder across a rendition's absence", () => {
 
 			playback.remove();
 			await microtasks();
+			const reanchor = spyOn(Sync.prototype, "reset");
 			playback.replace();
 			playback.restore();
 			await microtasks();
 			await playback.play(12);
 
-			// The ring re-anchors on the new broadcast rather than discarding it as already played.
+			// The ring and the shared clock re-anchor on the new broadcast rather than discarding it as
+			// already played.
+			expect(reanchor).toHaveBeenCalledTimes(1);
 			expect(drain(ring)).toBeGreaterThan(0);
 			expect(contexts).toHaveLength(1);
 			expect(playback.rings()).toEqual([ring]);

js/watch/src/audio/decoder.ts

@@ -331,13 +331,15 @@ export class Decoder {
 		const active = broadcast.relativeBroadcast(effect, identity.broadcast);
 		if (!active) return;
 
-		// Another broadcast (a new name, or a republish) brings its own timeline, which a ring anchored
-		// on the previous one would discard as already played. A return to the same instance keeps the
-		// anchor, so the relay's redelivered last group is still dropped as old. Every handle to one
-		// instance shares `closed`, so it identifies the instance where the handle itself does not.
+		// Another broadcast (a new name, or a republish) brings its own timeline, which a ring and a
+		// clock anchored on the previous one would discard as already played. A return to the same
+		// instance keeps the anchor, so the relay's redelivered last group is still dropped as old.
+		// Every handle to one instance shares `closed`, so it identifies the instance where the
+		// handle itself does not.
 		if (this.#instance !== undefined && this.#instance !== active.closed) {
 			this.#ring.peek()?.reset();
 			this.#decodeBuffered.set([]);
+			this.sync.reset();
 		}
 		this.#instance = active.closed;
 

js/watch/src/broadcast.test.ts

@@ -501,3 +501,72 @@ describe("manual catalog", () => {
 		}
 	});
 });
+
+// A broadcast's announcement is its online/offline signal. moq-lite 06 carries no epoch, so a
+// restarted publisher shows up either as a retraction and a fresh announcement or, when it
+// takes over a route still standing, as that route changing in place.
+describe("online and offline", () => {
+	it("follows a broadcast that ends and is announced again", async () => {
+		const owner = new Origin.Producer();
+		const name = Path.from("live.hang");
+		const source = new Broadcast({ origin: owner, name, enabled: true, announced: true, catalogFormat: "manual" });
+		try {
+			const first = publish(owner, name);
+			await settle();
+			const old = source.out.active.peek();
+			expect(old).toBeDefined();
+
+			first.close();
+			await settle();
+			expect(source.out.active.peek()).toBeUndefined();
+
+			const second = publish(owner, name);
+			await settle();
+			const active = source.out.active.peek();
+			expect(active).toBeDefined();
+			expect(active).not.toBe(old);
+			expect(active?.closed.peek()).toBeUndefined();
+			second.close();
+		} finally {
+			source.close();
+			owner.close();
+			await settle();
+		}
+	});
+
+	it("starts over when a route without an epoch changes in place", async () => {
+		const owner = new Origin.Producer();
+		const name = Path.from("live.hang");
+		const route = owner.dynamic(name);
+		const requests = route.requested();
+		const source = new Broadcast({ origin: owner, name, enabled: true, announced: true, catalogFormat: "manual" });
+		try {
+			const first = await requests.next();
+			first.value?.accept(new Moq.Broadcast.Producer());
+			await settle();
+			const old = source.out.active.peek();
+			expect(old).toBeDefined();
+
+			// Nothing says the changed route serves the same content, so the path is asked for
+			// again and everything reading the handle starts over on a fresh one.
+			route.update({ cost: 4n });
+			await settle();
+			const active = source.out.active.peek();
+			expect(active).toBeDefined();
+			expect(active).not.toBe(old);
+
+			// An epoch says the content is the same, so a re-price keeps the handle.
+			route.update({ epoch: Moq.Epoch.mint(), cost: 4n });
+			await settle();
+			const minted = source.out.active.peek();
+			route.update({ ...route.route, cost: 2n });
+			await settle();
+			expect(source.out.active.peek()).toBe(minted);
+		} finally {
+			source.close();
+			route.close();
+			owner.close();
+			await settle();
+		}
+	});
+});

js/watch/src/broadcast.ts

@@ -85,8 +85,9 @@ type BroadcastOutput = {
 	status: Signal<Status>;
 	active: Signal<Moq.Broadcast.Consumer | undefined>;
 
-	// Why the origin refused the broadcast, while `status` is "error". A refusal is final:
-	// only a fresh request (a new `name`, `origin`, or `announced`, or re-enabling) clears it and asks again.
+	// Why the origin refused the broadcast, while `status` is "error". A refusal is final: only a
+	// fresh request (a new `name`, `origin`, or `announced`, re-enabling, or a route without an
+	// epoch changing) clears it and asks again.
 	error: Signal<Error | undefined>;
 
 	// The effective catalog: the fetched one, or a copy of input.catalog in manual mode, minus
@@ -230,16 +231,40 @@ export class Broadcast {
 		if (!origin) return;
 
 		const name = effect.get(this.in.name);
-		const request = origin.request(name, { announced: effect.get(this.in.announced) });
-		effect.cleanup(() => request.close());
+		const announced = effect.get(this.in.announced);
+
+		// An epoch names the publisher instance, and a new one already swaps the handle. A route
+		// without one (moq-lite 06) cannot say whether a change in place is a restarted publisher,
+		// so ask for the path again on any change, which starts everything downstream over.
+		const routes = origin.broadcasts(undefined, { hidden: true });
+		const instance = announced
+			? effect.computed((nested) => {
+					let route: Moq.Origin.Route | undefined;
+					let length = -1;
+					for (const [prefix, candidate] of nested.get(routes)) {
+						if (prefix.length > length && Path.hasPrefix(prefix, name)) {
+							route = candidate;
+							length = prefix.length;
+						}
+					}
+					return route?.epoch ?? route;
+				})
+			: undefined;
 
 		effect.run((nested) => {
-			nested.set(this.#out.active, nested.get(request.active), undefined);
-		});
+			if (instance) nested.get(instance);
 
-		effect.run((nested) => {
-			const closed = nested.get(request.closed);
-			if (closed) nested.set(this.#out.error, closed, undefined);
+			const request = origin.request(name, { announced });
+			nested.cleanup(() => request.close());
+
+			nested.run((inner) => {
+				inner.set(this.#out.active, inner.get(request.active), undefined);
+			});
+
+			nested.run((inner) => {
+				const closed = inner.get(request.closed);
+				if (closed) inner.set(this.#out.error, closed, undefined);
+			});
 		});
 	}
 

js/watch/src/video/decoder.test.ts

@@ -3,7 +3,7 @@ import * as Catalog from "@moq/hang/catalog";
 import * as Container from "@moq/hang/container";
 import * as Moq from "@moq/net";
 import { Time } from "@moq/net";
-import { Signal } from "@moq/signals";
+import { type Effect, Signal } from "@moq/signals";
 import type { Broadcast } from "../broadcast";
 import { Sync } from "../sync";
 import { Decoder } from "./decoder";
@@ -30,7 +30,7 @@ function closedConsumer(): Moq.Broadcast.Consumer {
 
 function watchBroadcast(catalog: Signal<Catalog.Root>): Broadcast {
 	return {
-		out: { catalog },
+		out: { catalog, active: new Signal(undefined) },
 		relativeBroadcast: () => closedConsumer(),
 	} as unknown as Broadcast;
 }
@@ -213,12 +213,12 @@ async function guardedPlayback(kind: "legacy" | "cmaf", reads: Read[]) {
 	const video = config({ codedWidth: 16, codedHeight: 16, description: "01640028ffe100046764002801000268ee" });
 	if (kind === "cmaf")
 		video.container = { kind, init: Buffer.from(Container.Cmaf.createVideoInitSegment(video)).toString("base64") };
-	const producer = new Moq.Broadcast.Producer();
-	const consumer = producer.consume();
+	let producer = new Moq.Broadcast.Producer();
+	const active = new Signal<Moq.Broadcast.Consumer>(producer.consume());
 	const catalog = new Signal<Catalog.Root>({ video: { renditions: { video } } });
 	const broadcast = new Signal<Broadcast | undefined>({
-		out: { catalog },
-		relativeBroadcast: () => consumer,
+		out: { catalog, active },
+		relativeBroadcast: (effect: Effect) => effect.get(active),
 	} as unknown as Broadcast);
 	const source = new Source({ broadcast, supported: async () => true });
 	const sync = new Sync({ delay: "instant" });
@@ -229,11 +229,19 @@ async function guardedPlayback(kind: "legacy" | "cmaf", reads: Read[]) {
 		decoder,
 		enabled,
 		submitted,
+		sync,
+		// A restarted publisher takes the name: another instance of the same broadcast.
+		republish() {
+			active.peek().close();
+			producer.close();
+			producer = new Moq.Broadcast.Producer();
+			active.set(producer.consume());
+		},
 		close() {
 			decoder.close();
 			source.close();
 			sync.close();
-			consumer.close();
+			active.peek().close();
 			producer.close();
 			next.mockRestore();
 			for (const [name, original] of [
@@ -265,6 +273,25 @@ it("promoting from no active track holds its picture and timestamp until a frame
 	}
 });
 
+it("switches to a republished broadcast at once, its timeline starting over", async () => {
+	const reads = [sample(10, 10_000_000)];
+	const playback = await guardedPlayback("legacy", reads);
+	try {
+		expect(playback.decoder.out.timestamp.peek()).toBe(Time.Milli(10_000));
+		const reset = spyOn(playback.sync, "reset");
+
+		reads.push(sample(0, 1000));
+		playback.republish();
+		await microtasks();
+
+		// Waiting to catch up to the old picture would hold it until the new timeline passed 10s.
+		expect(playback.decoder.out.timestamp.peek()).toBe(Time.Milli(1));
+		expect(reset).toHaveBeenCalledTimes(1);
+	} finally {
+		playback.close();
+	}
+});
+
 // A late marker from older history raises a discontinuity without making older groups newer.
 function marker(group: number, end: number, discontinuity: number): Read {
 	return { group, discontinuity, continuous: false, frame: undefined, end: Time.Micro(end) };

js/watch/src/video/decoder.ts

@@ -14,6 +14,7 @@ import {
 	Signal,
 } from "@moq/signals";
 import { base64ToBytes } from "../base64";
+import type { Broadcast } from "../broadcast";
 import { nextMedia, subscribeMedia } from "../media";
 
 import type { Sync } from "../sync";
@@ -98,6 +99,9 @@ export class Decoder {
 	#pending = new Signal<{ track: string; catalog: Getter<Catalog.Root | undefined> } | undefined>(undefined);
 	// The active track's arrival estimate from its container consumer.
 	#spread = new Signal<Time.Milli | undefined>(undefined);
+	// The catalog broadcast the last rendition subscribed under, and its instance. Every handle to
+	// one instance shares `closed`, so it identifies the instance where the handle itself does not.
+	#instance?: { broadcast: Broadcast; closed: unknown };
 	readonly #identity: Computed<PlaybackIdentity | undefined>;
 
 	#signals = new Effect();
@@ -185,6 +189,15 @@ export class Decoder {
 			return;
 		}
 
+		// A republish at the same name is another instance, whose timeline starts over. Waiting for
+		// it to catch up to the old picture would freeze video until it passed the old timestamps,
+		// so switch at once and re-anchor the clock. A rendition switch within one instance, or a
+		// switch to another name, still waits.
+		const closed = effect.get(broadcast.out.active)?.closed;
+		const republished = this.#instance?.broadcast === broadcast && this.#instance.closed !== closed;
+		this.#instance = { broadcast, closed };
+		if (republished) this.sync.reset();
+
 		// Start a new pending effect.
 		let pending: DecoderTrack | undefined = new DecoderTrack({
 			sync: this.sync,
@@ -202,7 +215,7 @@ export class Decoder {
 			if (!pending) return;
 
 			const current = effect.get(this.#active);
-			if (current) {
+			if (current && !republished) {
 				// A zero timestamp is a rendered frame, not a missing one.
 				const pendingTimestamp = effect.get(pending.timestamp);
 				if (pendingTimestamp === undefined) return;