From 7f26f551df60e491ac099c01465a36f021e173b9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 15:26:37 -0700 Subject: [PATCH 01/11] quest: claim quest/m0/broadcast-epoch/apps From 108054d8bc878c68ef218ffb36c59818b724ff5e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 17:39:03 -0700 Subject: [PATCH 02/11] feat(apps): players follow announce Start, Restart, and End moq play follows its path through a new moq_mux::Source::follow: it plays on Start, starts over with a fresh catalog, decoders, and clock on Restart, lets what is playing finish on End and waits for the next Start, and ignores Update. @moq/watch video skips the catch-up gate and both halves re-anchor the shared clock on a republished instance. demo/web stats resubscribes on a restart. Publishers and the lite announce logs name the epoch. Co-Authored-By: Claude Opus 5.5 --- demo/web/src/stats.ts | 20 ++- doc/bin/cli.md | 13 +- doc/lib/js/watch.md | 6 + js/net/src/lite/publisher.ts | 4 +- js/net/src/lite/subscriber.ts | 4 +- js/watch/src/audio/decoder.test.ts | 6 +- js/watch/src/audio/decoder.ts | 10 +- js/watch/src/video/decoder.test.ts | 41 ++++- js/watch/src/video/decoder.ts | 15 +- quest/m0/broadcast-epoch/README.md | 7 +- quest/m0/broadcast-epoch/apps.md | 25 --- quest/m0/broadcast-epoch/moqsrc.md | 5 +- rs/moq-boy/src/main.rs | 4 +- rs/moq-cli/src/main.rs | 1 + rs/moq-cli/src/play/media.rs | 277 ++++++++++++++++++++++++++--- rs/moq-cli/src/play/mod.rs | 1 - rs/moq-cli/src/play/source.rs | 51 ------ rs/moq-cli/src/play/window.rs | 3 +- rs/moq-mux/src/lib.rs | 2 +- rs/moq-mux/src/source.rs | 192 ++++++++++++++++++++ 20 files changed, 553 insertions(+), 134 deletions(-) delete mode 100644 quest/m0/broadcast-epoch/apps.md delete mode 100644 rs/moq-cli/src/play/source.rs diff --git a/demo/web/src/stats.ts b/demo/web/src/stats.ts index 954e554629..924efff05b 100644 --- a/demo/web/src/stats.ts +++ b/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); } }); }); diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 35384e504e..652af7df57 100644 --- a/doc/bin/cli.md +++ b/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" diff --git a/doc/lib/js/watch.md b/doc/lib/js/watch.md index 1c6753e70d..1453925d0a 100644 --- a/doc/lib/js/watch.md +++ b/doc/lib/js/watch.md @@ -46,6 +46,12 @@ 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 the +name is announced, goes `offline` when it ends, and waits for it to return for +as long as it is mounted. 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 `` when the module diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index 45fbe8b65f..e087b6a18f 100644 --- a/js/net/src/lite/publisher.ts +++ b/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 }, diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index e1c5991d3f..afbab2cca2 100644 --- a/js/net/src/lite/subscriber.ts +++ b/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 }); } diff --git a/js/watch/src/audio/decoder.test.ts b/js/watch/src/audio/decoder.test.ts index 73276c30f9..c74243259f 100644 --- a/js/watch/src/audio/decoder.test.ts +++ b/js/watch/src/audio/decoder.test.ts @@ -420,12 +420,16 @@ 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); + reanchor.mockRestore(); expect(drain(ring)).toBeGreaterThan(0); expect(contexts).toHaveLength(1); expect(playback.rings()).toEqual([ring]); diff --git a/js/watch/src/audio/decoder.ts b/js/watch/src/audio/decoder.ts index e5aa18b8c4..83cc96b270 100644 --- a/js/watch/src/audio/decoder.ts +++ b/js/watch/src/audio/decoder.ts @@ -334,13 +334,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; diff --git a/js/watch/src/video/decoder.test.ts b/js/watch/src/video/decoder.test.ts index 2df963a73e..641fe2383f 100644 --- a/js/watch/src/video/decoder.test.ts +++ b/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): Broadcast { return { - out: { catalog }, + out: { catalog, active: new Signal(undefined) }, relativeBroadcast: () => closedConsumer(), } as unknown as Broadcast; } @@ -245,12 +245,12 @@ async function guardedPlayback(kind: "legacy" | "cmaf", reads: Read[], hold: rea 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(producer.consume()); const catalog = new Signal({ video: { renditions: { video } } }); const broadcast = new Signal({ - 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" }); @@ -263,14 +263,22 @@ async function guardedPlayback(kind: "legacy" | "cmaf", reads: Read[], hold: rea catalog, submitted, outputs, + sync, get resets() { return resets; }, + // 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 [ @@ -302,6 +310,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(); + } +}); + it("clears the picture once no rendition is selectable", async () => { const playback = await guardedPlayback("legacy", [sample(10, 1000)]); try { diff --git a/js/watch/src/video/decoder.ts b/js/watch/src/video/decoder.ts index fdccf2d5a8..2cd8279d3b 100644 --- a/js/watch/src/video/decoder.ts +++ b/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 } | undefined>(undefined); // The active track's arrival estimate from its container consumer. #spread = new Signal(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; #signals = new Effect(); @@ -188,6 +192,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, @@ -205,7 +218,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; diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index 956eb80694..690212692a 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -59,9 +59,9 @@ Decided: to m1. - A catalog `broadcast` reference by name follows the newest epoch, since a path cannot name one. -- Every first-party publisher that can restart mints its own: the apps and - moq-boy through [Apps](/quest/m0/broadcast-epoch/apps.md), the - ingest gateways, moqsink, and the bindings below. moq-stats mints one per +- Every first-party publisher that can restart mints its own: the apps, + moq-boy, the ingest gateways, moqsink, and the bindings below. Players + (`moq play`, `@moq/watch`, demo/web) follow the announce `Restart`. moq-stats mints one per [group announcement](/doc/concept/stats.md#broadcasts), which also gated the release (decided 2026-10-04). - The m1 quests gating this line moved under it in the 2026-10-05 audit, and @@ -94,7 +94,6 @@ then. MoQ Pro's VOD `storage.json` moves from the same seed to its own epoch. ## Required - [No stitch](/quest/m0/broadcast-epoch/no-stitch.md) - an epochless track ends on an upstream error instead of re-splicing onto whatever instance now wins upstream -- [Apps](/quest/m0/broadcast-epoch/apps.md) - moq-cli, the browser publish and watch components, and demo/web restart into a new epoch and reset on the switch - [Export ts](/quest/m0/broadcast-epoch/export-ts.md) - `export ts` and SRT egress linger only for the same epoch, and switch to a replacement only with `--stitch`, as a full program switch - [Publish catalog restart](/quest/m0/broadcast-epoch/publish-catalog-restart.md) - `@moq/publish` never reuses catalog group numbers under one name and epoch after a re-announce - [Bindings](/quest/m0/broadcast-epoch/bindings.md) - moq-ffi and every wrapper expose the epoch and let a publisher announce one diff --git a/quest/m0/broadcast-epoch/apps.md b/quest/m0/broadcast-epoch/apps.md deleted file mode 100644 index 4d5ad02b4d..0000000000 --- a/quest/m0/broadcast-epoch/apps.md +++ /dev/null @@ -1,25 +0,0 @@ -# [M] Apps restart into a new epoch and reset on the switch - -## Goal - -`moq-cli` publish and play, `@moq/publish`, `@moq/watch`, `demo/web`, and -moq-boy announce a freshly minted epoch per run. Each publish -run is a new epoch, so a restart while the -old route lingers is a new broadcast rather than a resume into the old one, -which stalls viewers until the new run's group sequence catches up. A viewer -switches to a republish within an RTT. Logs show the epoch. - -## Plan - -- Publish sides mint per run as of #4942: moq-cli publish, HLS import, - archive replay, WHEP import, and transcode output; moq-boy; `@moq/publish`, - the clock, and moq-boy's viewer feedback. What remains is below. -- Watch sides handle "the broadcast changed" as a fresh catalog and decoder - reset. Players (`moq play`, `@moq/watch`, demo/web) start on `Start`, - restart on `Restart`, and stop on `End`, so a re-price or a same-epoch - failover (an `Update`) never restarts playback. #4970, which drove them - from announcements, closed unmerged. Test a republish mid-playback in the - browser and native players. -- moq-room is out (decided 2026-10-08): it only discovers participants, and - the caller that publishes into a room mints the epoch. - diff --git a/quest/m0/broadcast-epoch/moqsrc.md b/quest/m0/broadcast-epoch/moqsrc.md index 23304cd899..cb1037d2a5 100644 --- a/quest/m0/broadcast-epoch/moqsrc.md +++ b/quest/m0/broadcast-epoch/moqsrc.md @@ -61,9 +61,8 @@ Decided in planning (2026-10-06, after #4955 found that no quest covered still drains to EOS. No timer and no property: announcements already say when the source is back, and a timer cannot tell a slow restart from a dead source. -- **Where.** The logic stays in `rs/moq-gst/src/source`. If - [Apps](/quest/m0/broadcast-epoch/apps.md) lands a shared helper for - following announcements first, use it instead. +- **Where.** The logic stays in `rs/moq-gst/src/source`, following the path + through `moq_mux::Source::follow`, as `moq play` does. - **Test.** An in-process test republishes the path under a newer epoch while the old publisher stays up: the new run's buffers render through a synced sink, a pad linked by name keeps flowing, and the bus carries no error. diff --git a/rs/moq-boy/src/main.rs b/rs/moq-boy/src/main.rs index 94e2bf3aa4..e251c20119 100644 --- a/rs/moq-boy/src/main.rs +++ b/rs/moq-boy/src/main.rs @@ -236,9 +236,11 @@ async fn run(config: &Config) -> Result<()> { let mut broadcast = publish_origin .create_broadcast(&broadcast_path) .context("failed to create broadcast")?; + let epoch = moq_net::Epoch::mint(); broadcast - .announce(moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint())) + .announce(moq_net::origin::Route::default().with_epoch(epoch.clone())) .context("failed to announce broadcast")?; + tracing::info!(broadcast = %broadcast_path, %epoch, "announced"); // Consume origin: viewer broadcasts under the viewer prefix. // JS publishes viewer feedback at "{viewer_prefix}/{name}/{viewerId}" diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 68178ea6e3..03c45e6e15 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -529,6 +529,7 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res // One publisher instance per run, unless redundant publishers named a shared one. let epoch = moq.epoch.clone().unwrap_or_else(moq_net::Epoch::mint); + tracing::info!(%epoch, "publisher instance"); for stage in stages { let name = stage.broadcast(&moq); diff --git a/rs/moq-cli/src/play/media.rs b/rs/moq-cli/src/play/media.rs index 62b70aa946..e4eb074eb0 100644 --- a/rs/moq-cli/src/play/media.rs +++ b/rs/moq-cli/src/play/media.rs @@ -8,6 +8,7 @@ use std::time::Duration; use anyhow::Context; use hang::moq_net; use moq_mux::catalog::{self, Stream}; +use moq_net::announce::Event as AnnounceEvent; // tokio's clock, which is the wall clock unless a test pauses it to drive the // playout clock itself. use tokio::time::Instant; @@ -15,7 +16,6 @@ use tokio::time::Instant; use super::args::Args; use super::output::{Output, Sink, Speaker}; use super::playback::{Kind, Playback, joined}; -use super::source::subscribe; use super::timeline::{AudioTimeline, Presentation, fit, timestamp}; use super::video::Video; use super::window::Event; @@ -63,8 +63,57 @@ impl Media { output.send(event); } + /// Follow the path's announcements for as long as the origin lasts: a start + /// plays the broadcast, a restart drops it for the instance that replaced it, + /// and an end lets what is playing finish while waiting for the next start. An + /// update is the same instance, which its subscriptions already ride out. async fn play(self) -> anyhow::Result<()> { - let source = subscribe(self.origin.clone(), &self.broadcast).await?; + let source = moq_mux::Source::new(self.origin.clone(), &self.broadcast); + let mut follow = source.follow()?; + let mut playing = None; + + loop { + tokio::select! { + event = follow.next() => { + // The origin is gone, so nothing can announce the path again. + let Some(event) = event else { return Ok(()) }; + let announce = match event { + AnnounceEvent::Start(announce) | AnnounceEvent::Restart(announce) => announce, + AnnounceEvent::Update(_) => continue, + AnnounceEvent::End(_) => { + tracing::info!(broadcast = %self.broadcast, "offline, waiting for it to return"); + continue; + } + }; + tracing::info!( + broadcast = %self.broadcast, + epoch = announce.route.epoch.as_ref().map(tracing::field::display), + "online" + ); + + // Another instance's timeline is its own, so nothing of the old one carries over. + drop(playing.take()); + *self.presentation.lock().unwrap() = Presentation::new(self.args.video_delay()); + self.video.lock().unwrap().clear(); + self.drained.notify_one(); + self.output.send(Event::Wake); + playing = Some(Box::pin(self.play_broadcast(source.clone()))); + } + result = async { playing.as_mut().expect("guarded").await }, if playing.is_some() => { + playing = None; + match result { + Ok(()) => tracing::info!(broadcast = %self.broadcast, "broadcast ended"), + Err(err) if err.is::() => return Err(err), + Err(err) => tracing::warn!(broadcast = %self.broadcast, err = format!("{err:#}"), "broadcast ended"), + } + } + } + } + } + + /// Play the broadcast at the path until its catalog and every track it + /// started end. + async fn play_broadcast(&self, source: moq_mux::Source) -> anyhow::Result<()> { let broadcast = source .broadcast() .await @@ -126,7 +175,9 @@ impl Media { match snapshot.context("failed to read the catalog")? { Some(snapshot) => playback.received(snapshot), None => { - anyhow::ensure!(playback.played, "the catalog contains no playable audio or video renditions"); + if !playback.played { + return Err(Unplayable("the catalog contains no playable audio or video renditions".into()).into()); + } playback.catalog_ended = true; } } @@ -268,15 +319,28 @@ impl Media { // Renditions on offer and not one of them playable, with nothing // already running to fall back on. - anyhow::ensure!( - !tasks.is_empty() || rejected.is_empty(), - "no playable rendition in the catalog: {}", - rejected.join("; ") - ); + if tasks.is_empty() && !rejected.is_empty() { + return Err( + Unplayable(format!("no playable rendition in the catalog: {}", rejected.join("; "))).into(), + ); + } } } } +/// A broadcast this build cannot play, which ends the player. A broadcast that +/// ends or fails on the wire does not: the next announcement plays again. +#[derive(Debug)] +struct Unplayable(String); + +impl std::fmt::Display for Unplayable { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0) + } +} + +impl std::error::Error for Unplayable {} + struct AudioPlayback { changed: Arc, speaker: O::Speaker, @@ -538,12 +602,7 @@ mod tests { drop(new); catalog.finish().unwrap(); - player.await.unwrap(); - match recorder.events().pop() { - Some(Event::Ended) => {} - Some(Event::Failed(err)) => panic!("playback failed: {err}"), - _ => panic!("playback never ended"), - } + settle(player, &recorder).await; let played = recorder.played(); let old_end = played.iter().filter(|p| p.sample == OLD).map(|p| p.to).max().unwrap(); @@ -555,6 +614,189 @@ mod tests { let old_duration: Duration = played.iter().filter(|p| p.sample == OLD).map(|p| p.to - p.from).sum(); assert_eq!(old_duration, Duration::from_secs(1)); } + + /// Publish an audio-only broadcast at `room` under `route`, returning what keeps + /// it up and its one rendition. + fn publish( + origin: &moq_net::origin::Producer, + route: moq_net::origin::Route, + ) -> ( + moq_net::broadcast::Producer, + catalog::Producer, + moq_mux::container::Producer, + ) { + let mut broadcast = origin.create_broadcast("room").unwrap(); + let catalog = catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let audio = rendition(&broadcast, &catalog, "audio"); + broadcast.announce(route).unwrap(); + (broadcast, catalog, audio) + } + + /// Write `count` packets of `sample` in real time, its timeline starting at 0. + async fn write(audio: &mut moq_mux::container::Producer, count: u64, sample: f32) { + for index in 0..count { + audio.write(packet(index, sample)).unwrap(); + tokio::time::sleep(PACKET_DURATION).await; + } + } + + /// How long `sample` played for, and through how many sinks. + fn played(recorder: &Recorder, sample: f32) -> (Duration, usize) { + let played: Vec<_> = recorder.played().into_iter().filter(|p| p.sample == sample).collect(); + let mut sinks: Vec<_> = played.iter().map(|p| p.sink).collect(); + sinks.dedup(); + (played.iter().map(|p| p.to - p.from).sum(), sinks.len()) + } + + /// The player follows the path for as long as the origin lasts, so it never + /// ends on its own: give it long enough to play out what it was handed, then + /// stop it. + async fn settle(player: tokio::task::JoinHandle<()>, recorder: &Recorder) { + tokio::time::sleep(Duration::from_secs(5)).await; + for event in recorder.events() { + match event { + Event::Failed(err) => panic!("playback failed: {err}"), + Event::Ended | Event::Finished => panic!("the player stopped following the path"), + Event::Wake => {} + } + } + assert!(!player.is_finished(), "the player stopped following the path"); + player.abort(); + } + + /// A restarted publisher announces a fresh epoch while its old run still + /// stands. The restart drops the old broadcast and plays the new one, whose + /// timeline starts over, instead of waiting for it to catch up. + #[tokio::test] + async fn a_republish_plays_the_new_broadcast() { + tokio::time::pause(); + + const OLD: f32 = 0.25; + const NEW: f32 = 0.5; + let origin = moq_tokio::origin::spawn(); + let epoch = || moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + + let (_old_broadcast, _old_catalog, mut old) = publish(&origin, epoch()); + let recorder = Recorder::default(); + let player = tokio::spawn(media(&origin, Duration::from_millis(50), recorder.clone()).run()); + write(&mut old, 10, OLD).await; + + let (_new_broadcast, _new_catalog, mut new) = publish(&origin, epoch()); + write(&mut new, 10, NEW).await; + + settle(player, &recorder).await; + assert!( + played(&recorder, OLD).0 > Duration::ZERO, + "the old broadcast never played" + ); + assert_eq!( + played(&recorder, NEW).0, + PACKET_DURATION * 10, + "the new broadcast did not play in full" + ); + } + + /// Without epochs (moq-lite 06), a restart that overlaps the old route still + /// replaces it: the newest announcement wins the path. + #[tokio::test] + async fn an_epochless_republish_plays_the_new_broadcast() { + tokio::time::pause(); + + const OLD: f32 = 0.25; + const NEW: f32 = 0.5; + let origin = moq_tokio::origin::spawn(); + + let (_old_broadcast, _old_catalog, mut old) = publish(&origin, Default::default()); + let recorder = Recorder::default(); + let player = tokio::spawn(media(&origin, Duration::from_millis(50), recorder.clone()).run()); + write(&mut old, 10, OLD).await; + + let (_new_broadcast, _new_catalog, mut new) = publish(&origin, Default::default()); + // A restart without an epoch waits out the origin's update hold, since a + // withdrawal wave's stale route looks just like one, and the player joins the + // new broadcast at its live edge. + tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD).await; + write(&mut new, 10, NEW).await; + + settle(player, &recorder).await; + assert!( + played(&recorder, OLD).0 > Duration::ZERO, + "the old broadcast never played" + ); + assert_eq!( + played(&recorder, NEW).0, + PACKET_DURATION * 10, + "the new broadcast did not play in full" + ); + } + + /// A publisher that stops cleanly ends the path, and its restart starts it + /// afresh. The player waits through the gap and plays the restarted broadcast. + #[tokio::test] + async fn a_broadcast_that_returns_plays_again() { + tokio::time::pause(); + + const OLD: f32 = 0.25; + const NEW: f32 = 0.5; + let origin = moq_tokio::origin::spawn(); + + let (old_broadcast, mut old_catalog, mut old) = publish(&origin, Default::default()); + let recorder = Recorder::default(); + let player = tokio::spawn(media(&origin, Duration::from_millis(50), recorder.clone()).run()); + write(&mut old, 10, OLD).await; + old.finish().unwrap(); + old_catalog.finish().unwrap(); + drop((old_broadcast, old_catalog, old)); + + // Offline for longer than any timeout the player might give up on. + tokio::time::sleep(Duration::from_secs(60)).await; + assert!(!player.is_finished(), "the player stopped following the path"); + + let (_new_broadcast, _new_catalog, mut new) = publish(&origin, Default::default()); + write(&mut new, 10, NEW).await; + + settle(player, &recorder).await; + assert_eq!( + played(&recorder, OLD).0, + PACKET_DURATION * 10, + "the old broadcast lost its tail" + ); + assert_eq!( + played(&recorder, NEW).0, + PACKET_DURATION * 10, + "the new broadcast did not play in full" + ); + } + + /// Re-pricing the same instance is an update, which playback rides through on + /// one sink rather than starting over. + #[tokio::test] + async fn a_reprice_keeps_playing() { + tokio::time::pause(); + + const SAMPLE: f32 = 0.25; + let origin = moq_tokio::origin::spawn(); + let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + + let (broadcast, _catalog, mut audio) = publish(&origin, route.clone()); + let recorder = Recorder::default(); + let player = tokio::spawn(media(&origin, Duration::from_millis(50), recorder.clone()).run()); + for index in 0..20 { + if index == 10 { + broadcast.announce(route.clone().with_cost(5)).unwrap(); + } + audio.write(packet(index, SAMPLE)).unwrap(); + tokio::time::sleep(PACKET_DURATION).await; + } + + settle(player, &recorder).await; + assert_eq!( + played(&recorder, SAMPLE), + (PACKET_DURATION * 20, 1), + "the update restarted playback" + ); + } + #[tokio::test] async fn a_finite_audio_track_plays_its_final_samples() { tokio::time::pause(); @@ -570,12 +812,7 @@ mod tests { audio.finish().unwrap(); drop(audio); catalog.finish().unwrap(); - player.await.unwrap(); - match recorder.events().pop() { - Some(Event::Ended) => {} - Some(Event::Failed(err)) => panic!("playback failed: {err}"), - _ => panic!("playback never ended"), - } + settle(player, &recorder).await; let played: Duration = recorder.played().iter().map(|p| p.to - p.from).sum(); assert_eq!(played, PACKET_DURATION, "the finished track lost its tail"); } diff --git a/rs/moq-cli/src/play/mod.rs b/rs/moq-cli/src/play/mod.rs index c455dd1e8d..8f2af57b9c 100644 --- a/rs/moq-cli/src/play/mod.rs +++ b/rs/moq-cli/src/play/mod.rs @@ -18,7 +18,6 @@ mod args; mod buffer; mod layout; mod playback; -mod source; mod timeline; #[cfg(all(test, feature = "play"))] diff --git a/rs/moq-cli/src/play/source.rs b/rs/moq-cli/src/play/source.rs deleted file mode 100644 index 5696665246..0000000000 --- a/rs/moq-cli/src/play/source.rs +++ /dev/null @@ -1,51 +0,0 @@ -//! Resolving the broadcast to play. - -use anyhow::Context; -use hang::moq_net; - -/// Wait for a route to cover `broadcast` on `origin`, then subscribe to it. -/// -/// The wait is the whole point. Subscribing goes through -/// `origin::Consumer::request_broadcast`, which resolves `Unroutable` on the -/// spot when no session has registered a handler yet rather than waiting for -/// one, and the media task starts well before the first handshake lands. The -/// window is already up, so this shows as a black frame rather than as a hang. -pub(super) async fn subscribe(origin: moq_net::origin::Consumer, broadcast: &str) -> anyhow::Result { - origin - .routed(broadcast) - .await - .with_context(|| format!("origin closed before broadcast `{broadcast}` was announced"))?; - - Ok(moq_mux::Source::new(origin, broadcast)) -} - -#[cfg(test)] -mod tests { - use super::*; - use std::time::Duration; - - /// Subscribing before the announcement lands doesn't wait, it fails: with no - /// session yet there is no handler registered on the origin, and - /// `request_broadcast` resolves `Unroutable` immediately. The media task is - /// spawned right after the reconnect loop starts, so it gets there first. - #[tokio::test] - async fn subscribe_waits_for_the_announcement() { - tokio::time::pause(); - - let origin = moq_tokio::origin::spawn(); - let consumer = origin.consume(); - - // Direct resolution has no route before the announcement. - let unannounced = moq_mux::Source::new(consumer.clone(), "room.hang").broadcast().await; - assert!(unannounced.is_err(), "expected an unroutable broadcast"); - - // Waiting first parks instead, for as long as it takes. - let mut waiting = std::pin::pin!(subscribe(consumer, "room.hang")); - let parked = tokio::time::timeout(Duration::from_secs(60), &mut waiting).await; - assert!(parked.is_err(), "expected to still be waiting on the announcement"); - - let _broadcast = origin.create_broadcast("room.hang").unwrap(); - _broadcast.announce(Default::default()).unwrap(); - waiting.await.unwrap(); - } -} diff --git a/rs/moq-cli/src/play/window.rs b/rs/moq-cli/src/play/window.rs index 850b3774f4..b516778123 100644 --- a/rs/moq-cli/src/play/window.rs +++ b/rs/moq-cli/src/play/window.rs @@ -37,7 +37,8 @@ pub(super) enum Event { Wake, /// Stop now: Ctrl-C, or the transport is gone and there is nothing more coming. Finished, - /// Every track reached its end. Present what is still queued, then stop. + /// The origin closed, so the broadcast can never come back. Present what is + /// still queued, then stop. Ended, Failed(String), } diff --git a/rs/moq-mux/src/lib.rs b/rs/moq-mux/src/lib.rs index 7fb00707fe..f0f055b389 100644 --- a/rs/moq-mux/src/lib.rs +++ b/rs/moq-mux/src/lib.rs @@ -44,7 +44,7 @@ pub mod timeline; pub use clock::Clock; pub use error::*; pub use pace::Pacer; -pub use source::{Binding, Source}; +pub use source::{Binding, Follow, Source}; /// Translate a catalog entry's declared compression into the flag the codecs take. /// diff --git a/rs/moq-mux/src/source.rs b/rs/moq-mux/src/source.rs index 85d45d4f4b..3add6b9cd1 100644 --- a/rs/moq-mux/src/source.rs +++ b/rs/moq-mux/src/source.rs @@ -68,6 +68,36 @@ impl Source { Ok(self.origin.routed_broadcast(&self.path).await?) } + /// Follow the announcements of the routes covering this source's path. + /// + /// This is how a player follows a broadcast across publisher restarts: play on a + /// [`Start`](moq_net::announce::Event::Start), drop everything and resolve the path afresh + /// on a [`Restart`](moq_net::announce::Event::Restart), and stop on an + /// [`End`](moq_net::announce::Event::End). An + /// [`Update`](moq_net::announce::Event::Update) is the same publisher instance re-priced or + /// failed over, which subscriptions already ride out. + /// + /// Fails with [`Unauthorized`](moq_net::Error::Unauthorized) when the origin's scope can + /// never cover the path. + pub fn follow(&self) -> crate::Result { + // Scoped to the path's subtree so the rest of the origin never wakes it. A path at the + // depth limit cannot be spelled as a subtree, so that one watches everything. + let origin = match moq_net::Pattern::subtree(self.path.as_str()) { + Ok(subtree) => self.origin.scope("", &subtree.into())?, + Err(_) => self.origin.clone(), + }; + // A narrower scope than the whole path would never see a route covering it. + if !origin.allowed().matches(self.path.as_str()) { + return Err(moq_net::Error::Unauthorized.into()); + } + Ok(Follow { + // Hiding narrows discovery, not lookup: a hidden path plays like any other. + announced: origin.with_hidden(true).announced(), + path: self.path.clone(), + covering: Vec::new(), + }) + } + /// Begin resolving the catalog broadcast (the one at this source's path). pub(crate) fn request_catalog(&self) -> kio::Pending { self.origin.request_broadcast(&self.path, None) @@ -218,6 +248,93 @@ impl Source { } } +/// The announcements covering a [`Source`]'s path, as one broadcast coming online, being +/// replaced, and going offline. Build one with [`Source::follow`]. +/// +/// Several routes can cover the path at once, such as the path itself and a prefix above it, +/// and a request resolves through the most specific one, so only that route's events come +/// through. Another route taking over is a [`Restart`](moq_net::announce::Event::Restart), +/// unless both carry the same epoch: those serve the same bytes, so it is an +/// [`Update`](moq_net::announce::Event::Update). +pub struct Follow { + announced: moq_net::announce::Consumer, + path: moq_net::PathOwned, + /// Every route standing over the path. + covering: Vec, +} + +impl Follow { + /// The next change to the route serving the path, or `None` once the origin closes. + pub async fn next(&mut self) -> Option { + kio::wait(|waiter| self.poll_next(waiter)).await + } + + /// Poll for the next change, registering `waiter` when there is none yet. + pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll> { + loop { + let Some(event) = ready!(self.announced.poll_next(waiter)) else { + return Poll::Ready(None); + }; + if let Some(event) = self.fold(event) { + return Poll::Ready(Some(event)); + } + } + } + + /// The most specific route covering the path, which is the one a request resolves. + fn serving(&self) -> Option<&moq_net::announce::Announce> { + self.covering + .iter() + .max_by_key(|announce| announce.prefix.as_str().len()) + } + + /// Apply one route's event, returning what it means for the path, if anything. + fn fold(&mut self, event: moq_net::announce::Event) -> Option { + use moq_net::announce::Event; + + let (Event::Start(announce) | Event::Update(announce) | Event::Restart(announce) | Event::End(announce)) = + &event; + // The subtree also carries the routes beneath the path, which serve other broadcasts. + if !self.path.has_prefix(&announce.prefix) { + return None; + } + let prefix = announce.prefix.clone(); + let before = self.serving().cloned(); + + self.covering.retain(|standing| standing.prefix != prefix); + let restart = match event { + Event::End(_) => false, + Event::Start(announce) | Event::Update(announce) => { + self.covering.push(announce); + false + } + Event::Restart(announce) => { + self.covering.push(announce); + true + } + }; + + let after = self.serving().cloned(); + match (before, after) { + (None, None) => None, + (None, Some(after)) => Some(Event::Start(after)), + (Some(before), None) => Some(Event::End(before)), + (Some(before), Some(after)) if before.prefix != after.prefix => { + let same = before.route.epoch.is_some() && before.route.epoch == after.route.epoch; + Some(if same { + Event::Update(after) + } else { + Event::Restart(after) + }) + } + // A route less specific than the serving one changed, which no request sees. + (Some(_), Some(after)) if after.prefix != prefix => None, + (Some(_), Some(after)) if restart => Some(Event::Restart(after)), + (Some(_), Some(after)) => Some(Event::Update(after)), + } + } +} + /// A held broadcast or the retained result of one eagerly issued broadcast request. /// /// Reading the binding never looks up its path again, even after a failure or publisher @@ -339,6 +456,81 @@ mod tests { } } + /// The follower's next event, which must arrive within the origin's update hold. + async fn followed(follow: &mut Follow) -> (&'static str, String) { + use moq_net::announce::Event; + let event = tokio::time::timeout(std::time::Duration::from_secs(1), follow.next()) + .await + .expect("no event") + .expect("the origin closed"); + match event { + Event::Start(announce) => ("start", announce.prefix.to_string()), + Event::Update(announce) => ("update", announce.prefix.to_string()), + Event::Restart(announce) => ("restart", announce.prefix.to_string()), + Event::End(announce) => ("end", announce.prefix.to_string()), + } + } + + /// The follower stays quiet past the origin's update hold. + async fn quiet(follow: &mut Follow) { + if let Ok(event) = tokio::time::timeout(std::time::Duration::from_secs(1), follow.next()).await { + panic!("expected nothing, got {event:?}"); + } + } + + #[tokio::test] + async fn follow_reports_the_route_serving_the_path() { + tokio::time::pause(); + let origin = produce_origin(); + let mut follow = Source::new(origin.consume(), "pool/job").follow().unwrap(); + let epoch = moq_net::Epoch::mint(); + let route = |epoch: &moq_net::Epoch| moq_net::origin::Route::default().with_epoch(epoch.clone()); + + // A prefix above the path covers it. + let pool = origin.dynamic("pool", route(&epoch)).unwrap(); + assert_eq!(followed(&mut follow).await, ("start", "pool".into())); + + // A route beneath the path serves another broadcast. + let _beneath = origin.publish("pool/job/thumbnail", Default::default()).unwrap(); + quiet(&mut follow).await; + + // The path itself, from the same instance, takes over without a restart. + let exact = origin.publish("pool/job", route(&epoch)).unwrap(); + assert_eq!(followed(&mut follow).await, ("update", "pool/job".into())); + + // The covering prefix no longer serves the path, so its changes are not seen. + pool.update(pool.route().with_cost(5)).unwrap(); + quiet(&mut follow).await; + + // Another instance at the path. + exact.announce(route(&moq_net::Epoch::mint())).unwrap(); + assert_eq!(followed(&mut follow).await, ("restart", "pool/job".into())); + + // It goes, so the prefix serves the path again: another instance. + drop(exact); + assert_eq!(followed(&mut follow).await, ("restart", "pool".into())); + + pool.update(pool.route().with_cost(9)).unwrap(); + assert_eq!(followed(&mut follow).await, ("update", "pool".into())); + + drop(pool); + assert_eq!(followed(&mut follow).await, ("end", "pool".into())); + } + + /// Without an epoch nothing says two routes serve the same bytes. + #[tokio::test] + async fn follow_restarts_onto_a_more_specific_route_without_an_epoch() { + tokio::time::pause(); + let origin = produce_origin(); + let mut follow = Source::new(origin.consume(), "pool/job").follow().unwrap(); + + let _pool = origin.dynamic("pool", Default::default()).unwrap(); + assert_eq!(followed(&mut follow).await, ("start", "pool".into())); + + let _exact = origin.publish("pool/job", Default::default()).unwrap(); + assert_eq!(followed(&mut follow).await, ("restart", "pool/job".into())); + } + #[tokio::test] async fn binding_retains_a_pending_request_across_cancelled_and_repeated_reads() { let origin = produce_origin(); From ab9296e8191643c020776e6b74302375d48398b7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 18:22:19 -0700 Subject: [PATCH 03/11] refactor(net): move the path follower into moq-net and @moq/net `origin::Consumer::follow(path)` returns `announce::Follow`, which reduces the announcements of every route covering a path to the Start, Update, Restart, and End of the one serving it. `@moq/net` mirrors it as `Origin.Consumer.follow(path)` (and on `Producer` and `Table`), returning an `Announce.Consumer`. `routed` is now the follower's first event. `moq play` builds on it, and `moq_mux::Source::follow` / `moq_mux::Follow` are gone. Co-Authored-By: Claude Opus 5.5 --- doc/lib/js/net.md | 2 +- doc/lib/rs/moq-net.md | 2 +- js/net/src/announced.ts | 55 ++++++++- js/net/src/origin.test.ts | 66 ++++++++++ js/net/src/origin.ts | 24 ++++ quest/m0/broadcast-epoch/moqsrc.md | 2 +- rs/moq-cli/src/play/media.rs | 2 +- rs/moq-mux/src/lib.rs | 2 +- rs/moq-mux/src/source.rs | 192 ----------------------------- rs/moq-net/src/model/follow.rs | 191 ++++++++++++++++++++++++++++ rs/moq-net/src/model/mod.rs | 3 + rs/moq-net/src/model/origin.rs | 54 ++++---- 12 files changed, 374 insertions(+), 221 deletions(-) create mode 100644 rs/moq-net/src/model/follow.rs diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index db24d9c9ee..e3c41b51c2 100644 --- a/doc/lib/js/net.md +++ b/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. diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 18d8989ed4..be5e2c8a55 100644 --- a/doc/lib/rs/moq-net.md +++ b/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. - **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`. diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index f85d83911d..41308afa78 100644 --- a/js/net/src/announced.ts +++ b/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,56 @@ export class Consumer { closeState(this.#state, abort); } } + +/** + * Reduce `source`, 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. Closing the result closes `source`. + * + * Mirrors `announce::Follow` in rs/moq-net. + * + * @internal + */ +export function follow(path: Path.Valid, source: Consumer): Consumer { + const producer = new Producer(); + void producer.closed.then(() => source.close()); + void (async () => { + // Every route standing over the path, by prefix. + const covering = new Map(); + const serving = () => { + let best: Announce | undefined; + for (const announce of covering.values()) { + if (!best || announce.prefix.length > best.prefix.length) best = announce; + } + return best; + }; + + try { + for await (const { kind, ...announce } of source) { + if (!Path.hasPrefix(announce.prefix, path)) continue; + const before = serving(); + if (kind === "end") covering.delete(announce.prefix); + else covering.set(announce.prefix, announce); + const after = serving(); + + if (!before) { + if (after) producer.append({ kind: "start", ...after }); + } else if (!after) { + producer.append({ kind: "end", ...before }); + } else if (before.prefix !== after.prefix) { + const same = before.route.epoch !== undefined && before.route.epoch === after.route.epoch; + producer.append({ kind: same ? "update" : "restart", ...after }); + } else if (after.prefix === announce.prefix) { + // A route less specific than the serving one changing is skipped: no request sees it. + producer.append({ kind: kind === "restart" ? "restart" : "update", ...after }); + } + } + producer.close(); + } catch (err) { + // Also reached when the reader closed between events; closing again is a no-op. + producer.close(err instanceof Error ? err : new Error(String(err))); + } + })(); + return producer.consume(); +} diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index 6c6ede3fbd..ea3ea11ef7 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -330,6 +330,72 @@ 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 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(); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index d1c837520d..db3a56e64f 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -852,6 +852,9 @@ export interface Table { /** The available broadcasts under `scope`, as a live stream; see {@link Consumer.announced}. */ announced(scope?: Path.Pattern, options?: announce.Options): announce.Consumer; + /** The announcements of the route serving `path`, as a live stream; see {@link Consumer.follow}. */ + follow(path: Path.Valid): announce.Consumer; + /** Advertise a prefix and serve requests under it; see {@link Producer.dynamic}. */ dynamic( prefix: Path.Valid, @@ -1216,6 +1219,11 @@ export class Producer implements Table { return this.#reader.announced(scope, options); } + /** The announcements of the route serving `path`, as a live stream; see {@link Consumer.follow}. */ + follow(path: Path.Valid): announce.Consumer { + return this.#reader.follow(path); + } + /** Close the origin, every broadcast it still routes, and its announcement streams. Idempotent. */ close(abort?: Error) { if (this.#state.closed.peek() !== undefined) return; @@ -1613,6 +1621,22 @@ export class Consumer { return producer.consume(); } + /** + * The announcements of the routes covering `path`, reduced to the one serving it: the most + * specific, which is the one a {@link request} resolves. + * + * This is how a player follows a broadcast across publisher restarts: play on `start`, drop + * everything and request the path afresh on `restart`, and stop on `end`. An `update` is + * the same publisher instance re-priced or failed over, which subscriptions already ride + * out. Another route taking over is a `restart`, or an `update` when both carry the same + * epoch. Throws when the path is outside this consumer's scope. Close it when done. + */ + follow(path: Path.Valid): announce.Consumer { + this.#scope.path(path); + // Hiding narrows discovery, not lookup, so a hidden path follows like any other. + return announce.follow(path, this.announced(Path.Pattern.subtree(path), { hidden: true })); + } + /** One snapshot shared by map readers and announcement-stream diffing. */ #listed(patterns: Path.Patterns, hidden: boolean): Map { const next = new Map(); diff --git a/quest/m0/broadcast-epoch/moqsrc.md b/quest/m0/broadcast-epoch/moqsrc.md index cb1037d2a5..4a2759e99e 100644 --- a/quest/m0/broadcast-epoch/moqsrc.md +++ b/quest/m0/broadcast-epoch/moqsrc.md @@ -62,7 +62,7 @@ Decided in planning (2026-10-06, after #4955 found that no quest covered when the source is back, and a timer cannot tell a slow restart from a dead source. - **Where.** The logic stays in `rs/moq-gst/src/source`, following the path - through `moq_mux::Source::follow`, as `moq play` does. + through moq-net's `origin::Consumer::follow`, as `moq play` does. - **Test.** An in-process test republishes the path under a newer epoch while the old publisher stays up: the new run's buffers render through a synced sink, a pad linked by name keeps flowing, and the bus carries no error. diff --git a/rs/moq-cli/src/play/media.rs b/rs/moq-cli/src/play/media.rs index e4eb074eb0..9c26d65bbb 100644 --- a/rs/moq-cli/src/play/media.rs +++ b/rs/moq-cli/src/play/media.rs @@ -68,8 +68,8 @@ impl Media { /// and an end lets what is playing finish while waiting for the next start. An /// update is the same instance, which its subscriptions already ride out. async fn play(self) -> anyhow::Result<()> { + let mut follow = self.origin.follow(&self.broadcast)?; let source = moq_mux::Source::new(self.origin.clone(), &self.broadcast); - let mut follow = source.follow()?; let mut playing = None; loop { diff --git a/rs/moq-mux/src/lib.rs b/rs/moq-mux/src/lib.rs index f0f055b389..7fb00707fe 100644 --- a/rs/moq-mux/src/lib.rs +++ b/rs/moq-mux/src/lib.rs @@ -44,7 +44,7 @@ pub mod timeline; pub use clock::Clock; pub use error::*; pub use pace::Pacer; -pub use source::{Binding, Follow, Source}; +pub use source::{Binding, Source}; /// Translate a catalog entry's declared compression into the flag the codecs take. /// diff --git a/rs/moq-mux/src/source.rs b/rs/moq-mux/src/source.rs index 3add6b9cd1..85d45d4f4b 100644 --- a/rs/moq-mux/src/source.rs +++ b/rs/moq-mux/src/source.rs @@ -68,36 +68,6 @@ impl Source { Ok(self.origin.routed_broadcast(&self.path).await?) } - /// Follow the announcements of the routes covering this source's path. - /// - /// This is how a player follows a broadcast across publisher restarts: play on a - /// [`Start`](moq_net::announce::Event::Start), drop everything and resolve the path afresh - /// on a [`Restart`](moq_net::announce::Event::Restart), and stop on an - /// [`End`](moq_net::announce::Event::End). An - /// [`Update`](moq_net::announce::Event::Update) is the same publisher instance re-priced or - /// failed over, which subscriptions already ride out. - /// - /// Fails with [`Unauthorized`](moq_net::Error::Unauthorized) when the origin's scope can - /// never cover the path. - pub fn follow(&self) -> crate::Result { - // Scoped to the path's subtree so the rest of the origin never wakes it. A path at the - // depth limit cannot be spelled as a subtree, so that one watches everything. - let origin = match moq_net::Pattern::subtree(self.path.as_str()) { - Ok(subtree) => self.origin.scope("", &subtree.into())?, - Err(_) => self.origin.clone(), - }; - // A narrower scope than the whole path would never see a route covering it. - if !origin.allowed().matches(self.path.as_str()) { - return Err(moq_net::Error::Unauthorized.into()); - } - Ok(Follow { - // Hiding narrows discovery, not lookup: a hidden path plays like any other. - announced: origin.with_hidden(true).announced(), - path: self.path.clone(), - covering: Vec::new(), - }) - } - /// Begin resolving the catalog broadcast (the one at this source's path). pub(crate) fn request_catalog(&self) -> kio::Pending { self.origin.request_broadcast(&self.path, None) @@ -248,93 +218,6 @@ impl Source { } } -/// The announcements covering a [`Source`]'s path, as one broadcast coming online, being -/// replaced, and going offline. Build one with [`Source::follow`]. -/// -/// Several routes can cover the path at once, such as the path itself and a prefix above it, -/// and a request resolves through the most specific one, so only that route's events come -/// through. Another route taking over is a [`Restart`](moq_net::announce::Event::Restart), -/// unless both carry the same epoch: those serve the same bytes, so it is an -/// [`Update`](moq_net::announce::Event::Update). -pub struct Follow { - announced: moq_net::announce::Consumer, - path: moq_net::PathOwned, - /// Every route standing over the path. - covering: Vec, -} - -impl Follow { - /// The next change to the route serving the path, or `None` once the origin closes. - pub async fn next(&mut self) -> Option { - kio::wait(|waiter| self.poll_next(waiter)).await - } - - /// Poll for the next change, registering `waiter` when there is none yet. - pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll> { - loop { - let Some(event) = ready!(self.announced.poll_next(waiter)) else { - return Poll::Ready(None); - }; - if let Some(event) = self.fold(event) { - return Poll::Ready(Some(event)); - } - } - } - - /// The most specific route covering the path, which is the one a request resolves. - fn serving(&self) -> Option<&moq_net::announce::Announce> { - self.covering - .iter() - .max_by_key(|announce| announce.prefix.as_str().len()) - } - - /// Apply one route's event, returning what it means for the path, if anything. - fn fold(&mut self, event: moq_net::announce::Event) -> Option { - use moq_net::announce::Event; - - let (Event::Start(announce) | Event::Update(announce) | Event::Restart(announce) | Event::End(announce)) = - &event; - // The subtree also carries the routes beneath the path, which serve other broadcasts. - if !self.path.has_prefix(&announce.prefix) { - return None; - } - let prefix = announce.prefix.clone(); - let before = self.serving().cloned(); - - self.covering.retain(|standing| standing.prefix != prefix); - let restart = match event { - Event::End(_) => false, - Event::Start(announce) | Event::Update(announce) => { - self.covering.push(announce); - false - } - Event::Restart(announce) => { - self.covering.push(announce); - true - } - }; - - let after = self.serving().cloned(); - match (before, after) { - (None, None) => None, - (None, Some(after)) => Some(Event::Start(after)), - (Some(before), None) => Some(Event::End(before)), - (Some(before), Some(after)) if before.prefix != after.prefix => { - let same = before.route.epoch.is_some() && before.route.epoch == after.route.epoch; - Some(if same { - Event::Update(after) - } else { - Event::Restart(after) - }) - } - // A route less specific than the serving one changed, which no request sees. - (Some(_), Some(after)) if after.prefix != prefix => None, - (Some(_), Some(after)) if restart => Some(Event::Restart(after)), - (Some(_), Some(after)) => Some(Event::Update(after)), - } - } -} - /// A held broadcast or the retained result of one eagerly issued broadcast request. /// /// Reading the binding never looks up its path again, even after a failure or publisher @@ -456,81 +339,6 @@ mod tests { } } - /// The follower's next event, which must arrive within the origin's update hold. - async fn followed(follow: &mut Follow) -> (&'static str, String) { - use moq_net::announce::Event; - let event = tokio::time::timeout(std::time::Duration::from_secs(1), follow.next()) - .await - .expect("no event") - .expect("the origin closed"); - match event { - Event::Start(announce) => ("start", announce.prefix.to_string()), - Event::Update(announce) => ("update", announce.prefix.to_string()), - Event::Restart(announce) => ("restart", announce.prefix.to_string()), - Event::End(announce) => ("end", announce.prefix.to_string()), - } - } - - /// The follower stays quiet past the origin's update hold. - async fn quiet(follow: &mut Follow) { - if let Ok(event) = tokio::time::timeout(std::time::Duration::from_secs(1), follow.next()).await { - panic!("expected nothing, got {event:?}"); - } - } - - #[tokio::test] - async fn follow_reports_the_route_serving_the_path() { - tokio::time::pause(); - let origin = produce_origin(); - let mut follow = Source::new(origin.consume(), "pool/job").follow().unwrap(); - let epoch = moq_net::Epoch::mint(); - let route = |epoch: &moq_net::Epoch| moq_net::origin::Route::default().with_epoch(epoch.clone()); - - // A prefix above the path covers it. - let pool = origin.dynamic("pool", route(&epoch)).unwrap(); - assert_eq!(followed(&mut follow).await, ("start", "pool".into())); - - // A route beneath the path serves another broadcast. - let _beneath = origin.publish("pool/job/thumbnail", Default::default()).unwrap(); - quiet(&mut follow).await; - - // The path itself, from the same instance, takes over without a restart. - let exact = origin.publish("pool/job", route(&epoch)).unwrap(); - assert_eq!(followed(&mut follow).await, ("update", "pool/job".into())); - - // The covering prefix no longer serves the path, so its changes are not seen. - pool.update(pool.route().with_cost(5)).unwrap(); - quiet(&mut follow).await; - - // Another instance at the path. - exact.announce(route(&moq_net::Epoch::mint())).unwrap(); - assert_eq!(followed(&mut follow).await, ("restart", "pool/job".into())); - - // It goes, so the prefix serves the path again: another instance. - drop(exact); - assert_eq!(followed(&mut follow).await, ("restart", "pool".into())); - - pool.update(pool.route().with_cost(9)).unwrap(); - assert_eq!(followed(&mut follow).await, ("update", "pool".into())); - - drop(pool); - assert_eq!(followed(&mut follow).await, ("end", "pool".into())); - } - - /// Without an epoch nothing says two routes serve the same bytes. - #[tokio::test] - async fn follow_restarts_onto_a_more_specific_route_without_an_epoch() { - tokio::time::pause(); - let origin = produce_origin(); - let mut follow = Source::new(origin.consume(), "pool/job").follow().unwrap(); - - let _pool = origin.dynamic("pool", Default::default()).unwrap(); - assert_eq!(followed(&mut follow).await, ("start", "pool".into())); - - let _exact = origin.publish("pool/job", Default::default()).unwrap(); - assert_eq!(followed(&mut follow).await, ("restart", "pool/job".into())); - } - #[tokio::test] async fn binding_retains_a_pending_request_across_cancelled_and_repeated_reads() { let origin = produce_origin(); diff --git a/rs/moq-net/src/model/follow.rs b/rs/moq-net/src/model/follow.rs new file mode 100644 index 0000000000..ddd4b1c7db --- /dev/null +++ b/rs/moq-net/src/model/follow.rs @@ -0,0 +1,191 @@ +use std::task::{Poll, ready}; + +use super::origin_impl::{Announce, AnnounceConsumer, AnnounceEvent}; +use crate::PathOwned; + +/// The announcements covering one path, as one broadcast coming online, being replaced, and +/// going offline. Created by [`Consumer::follow`](crate::origin::Consumer::follow). +/// +/// Several routes can cover the path at once, such as the path itself and a prefix above it, +/// and a request resolves through the most specific one, so only that route's events come +/// through. Another route taking over is a [`Restart`](AnnounceEvent::Restart), unless both +/// carry the same epoch: those serve the same bytes, so it is an +/// [`Update`](AnnounceEvent::Update). Routes beneath the path serve other broadcasts and are +/// ignored. +pub struct Follow { + announced: AnnounceConsumer, + path: PathOwned, + /// Every route standing over the path. + covering: Vec, +} + +impl Follow { + pub(super) fn new(announced: AnnounceConsumer, path: PathOwned) -> Self { + Self { + announced, + path, + covering: Vec::new(), + } + } + + /// The next change to the route serving the path, or `None` once the origin closes. + pub async fn next(&mut self) -> Option { + kio::wait(|waiter| self.poll_next(waiter)).await + } + + /// Poll for the next change, registering `waiter` when there is none yet. + pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll> { + loop { + let Some(event) = ready!(self.announced.poll_next(waiter)) else { + return Poll::Ready(None); + }; + if let Some(event) = self.fold(event) { + return Poll::Ready(Some(event)); + } + } + } + + /// The most specific route covering the path, which is the one a request resolves. + fn serving(&self) -> Option<&Announce> { + self.covering + .iter() + .max_by_key(|announce| announce.prefix.as_str().len()) + } + + /// Apply one route's event, returning what it means for the path, if anything. + fn fold(&mut self, event: AnnounceEvent) -> Option { + let (AnnounceEvent::Start(announce) + | AnnounceEvent::Update(announce) + | AnnounceEvent::Restart(announce) + | AnnounceEvent::End(announce)) = &event; + if !self.path.has_prefix(&announce.prefix) { + return None; + } + let prefix = announce.prefix.clone(); + let before = self.serving().cloned(); + + self.covering.retain(|standing| standing.prefix != prefix); + let restart = match event { + AnnounceEvent::End(_) => false, + AnnounceEvent::Start(announce) | AnnounceEvent::Update(announce) => { + self.covering.push(announce); + false + } + AnnounceEvent::Restart(announce) => { + self.covering.push(announce); + true + } + }; + + let after = self.serving().cloned(); + match (before, after) { + (None, None) => None, + (None, Some(after)) => Some(AnnounceEvent::Start(after)), + (Some(before), None) => Some(AnnounceEvent::End(before)), + (Some(before), Some(after)) if before.prefix != after.prefix => { + let same = before.route.epoch.is_some() && before.route.epoch == after.route.epoch; + Some(match same { + true => AnnounceEvent::Update(after), + false => AnnounceEvent::Restart(after), + }) + } + // A route less specific than the serving one changed, which no request sees. + (Some(_), Some(after)) if after.prefix != prefix => None, + (Some(_), Some(after)) if restart => Some(AnnounceEvent::Restart(after)), + (Some(_), Some(after)) => Some(AnnounceEvent::Update(after)), + } + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::super::ProduceTest; + use super::*; + use crate::{Epoch, Error, Hop, Pattern, Patterns, origin::Route}; + + /// The follower's next event, which must arrive within the origin's update hold. + async fn followed(follow: &mut Follow) -> (&'static str, String) { + let event = moq_net_sim::timeout(Duration::from_secs(1), follow.next()) + .await + .expect("no event") + .expect("the origin closed"); + match event { + AnnounceEvent::Start(announce) => ("start", announce.prefix.to_string()), + AnnounceEvent::Update(announce) => ("update", announce.prefix.to_string()), + AnnounceEvent::Restart(announce) => ("restart", announce.prefix.to_string()), + AnnounceEvent::End(announce) => ("end", announce.prefix.to_string()), + } + } + + /// The follower stays quiet past the origin's update hold. + async fn quiet(follow: &mut Follow) { + if let Ok(event) = moq_net_sim::timeout(Duration::from_secs(1), follow.next()).await { + panic!("expected nothing, got {event:?}"); + } + } + + fn route(epoch: &Epoch) -> Route { + Route::default().with_epoch(epoch.clone()) + } + + #[moq_net_sim::test] + async fn follow_reports_the_route_serving_the_path() { + let origin = Hop::new(1).unwrap().produce(); + let mut follow = origin.consume().follow("pool/job").unwrap(); + let epoch = Epoch::mint(); + + // A prefix above the path covers it. + let pool = origin.dynamic("pool", route(&epoch)).unwrap(); + assert_eq!(followed(&mut follow).await, ("start", "pool".into())); + + // A route beneath the path serves another broadcast. + let _beneath = origin.publish("pool/job/thumbnail", Route::default()).unwrap(); + quiet(&mut follow).await; + + // The path itself, from the same instance, takes over without a restart. + let exact = origin.publish("pool/job", route(&epoch)).unwrap(); + assert_eq!(followed(&mut follow).await, ("update", "pool/job".into())); + + // The covering prefix no longer serves the path, so its changes are not seen. + pool.update(pool.route().with_cost(5)).unwrap(); + quiet(&mut follow).await; + + // Another instance at the path. + exact.announce(route(&Epoch::mint())).unwrap(); + assert_eq!(followed(&mut follow).await, ("restart", "pool/job".into())); + + // It goes, so the prefix serves the path again: another instance. + drop(exact); + assert_eq!(followed(&mut follow).await, ("restart", "pool".into())); + + pool.update(pool.route().with_cost(9)).unwrap(); + assert_eq!(followed(&mut follow).await, ("update", "pool".into())); + + drop(pool); + assert_eq!(followed(&mut follow).await, ("end", "pool".into())); + } + + /// Without an epoch nothing says two routes serve the same bytes. + #[moq_net_sim::test] + async fn follow_restarts_onto_a_more_specific_route_without_an_epoch() { + let origin = Hop::new(1).unwrap().produce(); + let mut follow = origin.consume().follow("pool/job").unwrap(); + + let _pool = origin.dynamic("pool", Route::default()).unwrap(); + assert_eq!(followed(&mut follow).await, ("start", "pool".into())); + + let _exact = origin.publish("pool/job", Route::default()).unwrap(); + assert_eq!(followed(&mut follow).await, ("restart", "pool/job".into())); + } + + /// A path the consumer's scope can never cover fails at once instead of waiting forever. + #[test] + fn follow_refuses_a_path_outside_the_scope() { + let origin = Hop::new(1).unwrap().produce(); + let scope = Patterns::from(Pattern::subtree("pool/job/cam").unwrap()); + let scoped = origin.consume().scope("", &scope).unwrap(); + assert!(matches!(scoped.follow("pool/job"), Err(Error::Unauthorized))); + } +} diff --git a/rs/moq-net/src/model/mod.rs b/rs/moq-net/src/model/mod.rs index eba650c029..6940c846c2 100644 --- a/rs/moq-net/src/model/mod.rs +++ b/rs/moq-net/src/model/mod.rs @@ -10,6 +10,8 @@ pub mod track; // modules so neither leaks the other's plumbing. #[path = "origin.rs"] mod origin_impl; +// Reduces the routes covering one path to the one serving it. +mod follow; // The failover state machine origin fronts run; pure, so its transitions are // tested without a runtime. mod front; @@ -48,6 +50,7 @@ pub mod origin { /// Subscribing to route (un)announcements from an origin. pub mod announce { + pub use super::follow::Follow; pub use super::origin_impl::{Announce, AnnounceConsumer as Consumer, AnnounceEvent as Event}; } diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index dfc64587d2..4977356dbf 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -4392,37 +4392,45 @@ impl Consumer { /// [`Self::routed_broadcast`]: pairing this with [`Self::request_broadcast`] /// leaves a gap where the covering route can retract. pub async fn routed(&self, path: impl AsPath) -> Option { + // A follower's first event is always a start. + match self.follow(path).ok()?.next().await? { + AnnounceEvent::Start(announce) | AnnounceEvent::Update(announce) | AnnounceEvent::Restart(announce) => { + Some(announce.route) + } + AnnounceEvent::End(_) => None, + } + } + + /// Follow the announcements of the routes covering `path`, reduced to the one serving it. + /// + /// This is how a player follows a broadcast across publisher restarts: play on a + /// [`Start`](AnnounceEvent::Start), drop everything and request the path afresh on a + /// [`Restart`](AnnounceEvent::Restart), and stop on an [`End`](AnnounceEvent::End). An + /// [`Update`](AnnounceEvent::Update) is the same publisher instance re-priced or failed + /// over, which subscriptions already ride out. Fails with [`Error::Unauthorized`] when + /// this consumer's scope can never cover the path. + pub fn follow(&self, path: impl AsPath) -> Result { let path = path.as_path(); - // Scope a fresh consumer down to this path's subtree, so we only wake for - // announcements that overlap the requested path. - // A max-depth path cannot be spelled as `path/**` (`**` would be a 33rd - // segment), so watch the existing stream and match covering claims instead. + // Scope down to the path's subtree, so the rest of the origin never wakes it. A path + // no pattern can spell (a segment containing `*`) watches everything instead, and the + // follower skips the routes that do not cover it. let consumer = match Pattern::subtree(path.as_str()) { - Ok(subtree) => self.scope("", &Patterns::from(subtree)).ok()?, - Err(InvalidPattern::TooManySegments) => self.clone(), - Err(_) => return None, + Ok(subtree) => self.scope("", &Patterns::from(subtree))?, + Err(_) => self.clone(), }; - // `scope` keeps narrower permissions intact: if we ask for `foo` on a - // consumer limited to `foo/specific`, `foo` itself is unauthorized. Bail - // rather than loop forever. + // `scope` keeps narrower permissions intact: on a consumer limited to + // `foo/specific`, no route can ever cover `foo`. if !consumer.allowed().matches(path.as_str()) { - return None; + return Err(Error::Unauthorized); } - // Use an untagged stream: this is a lookup, not egress announce - // forwarding, so it must not drive the announce guards. Hiding narrows - // discovery, not lookup, so a hidden path resolves like any other. - let mut announced = consumer.untagged().with_hidden(true).announced(); - loop { - if let AnnounceEvent::Start(announce) | AnnounceEvent::Update(announce) | AnnounceEvent::Restart(announce) = - announced.next().await? - && path.has_prefix(&announce.prefix) - { - return Some(announce.route); - } - } + // Untagged: this is a lookup, not egress announce forwarding, so it must not drive + // the announce guards. Hiding narrows discovery, not lookup, so a hidden path + // follows like any other. + let announced = consumer.untagged().with_hidden(true).announced(); + Ok(crate::announce::Follow::new(announced, path.to_owned())) } /// Block until `path` resolves to a broadcast: [`Self::request_broadcast`], From 8d70dd54075b9412ddfee3d72b7c479bc96a6f38 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 19:19:48 -0700 Subject: [PATCH 04/11] feat(watch): follow the route serving the name `Broadcast` follows its name through `Origin.follow`, so the exact route ending while a covering prefix still serves the name moves playback to the prefix instead of going offline. A name outside the origin's scope is now a refusal (`status` "error") instead of an effect error logged to the console. `@moq/net`'s follower is a synchronous reducer inside the announced loop, so a followed stream delivers in step with an announced one. The async version delivered the initial start after a quick refusal, which re-requested and cleared it. Co-Authored-By: Claude Opus 5.5 --- doc/lib/js/watch.md | 8 ++-- js/net/src/announced.ts | 72 ++++++++++++++-------------------- js/net/src/origin.ts | 25 +++++++++--- js/watch/src/broadcast.test.ts | 57 ++++++++++++++++++++++++++- js/watch/src/broadcast.ts | 27 ++++++++----- 5 files changed, 128 insertions(+), 61 deletions(-) diff --git a/doc/lib/js/watch.md b/doc/lib/js/watch.md index 1453925d0a..f90c414428 100644 --- a/doc/lib/js/watch.md +++ b/doc/lib/js/watch.md @@ -46,9 +46,11 @@ 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 the -name is announced, goes `offline` when it ends, and waits for it to return for -as long as it is mounted. A restarted publisher at the same name is followed at +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. diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index 41308afa78..b043f4047a 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -165,54 +165,42 @@ export class Consumer { } /** - * Reduce `source`, the announcements of the routes covering `path`, to those of the one route + * 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. Closing the result closes `source`. + * other broadcasts and are skipped, as is any change no request sees (`undefined`). * + * Synchronous, so a followed stream delivers in step with an announced one. * Mirrors `announce::Follow` in rs/moq-net. * * @internal */ -export function follow(path: Path.Valid, source: Consumer): Consumer { - const producer = new Producer(); - void producer.closed.then(() => source.close()); - void (async () => { - // Every route standing over the path, by prefix. - const covering = new Map(); - const serving = () => { - let best: Announce | undefined; - for (const announce of covering.values()) { - if (!best || announce.prefix.length > best.prefix.length) best = announce; - } - return best; - }; - - try { - for await (const { kind, ...announce } of source) { - if (!Path.hasPrefix(announce.prefix, path)) continue; - const before = serving(); - if (kind === "end") covering.delete(announce.prefix); - else covering.set(announce.prefix, announce); - const after = serving(); - - if (!before) { - if (after) producer.append({ kind: "start", ...after }); - } else if (!after) { - producer.append({ kind: "end", ...before }); - } else if (before.prefix !== after.prefix) { - const same = before.route.epoch !== undefined && before.route.epoch === after.route.epoch; - producer.append({ kind: same ? "update" : "restart", ...after }); - } else if (after.prefix === announce.prefix) { - // A route less specific than the serving one changing is skipped: no request sees it. - producer.append({ kind: kind === "restart" ? "restart" : "update", ...after }); - } - } - producer.close(); - } catch (err) { - // Also reached when the reader closed between events; closing again is a no-op. - producer.close(err instanceof Error ? err : new Error(String(err))); +export function follower(path: Path.Valid): (event: Event) => Event | undefined { + // Every route standing over the path, by prefix. + const covering = new Map(); + const serving = () => { + let best: Announce | undefined; + for (const announce of covering.values()) { + if (!best || announce.prefix.length > best.prefix.length) best = announce; } - })(); - return producer.consume(); + return best; + }; + + return ({ kind, ...announce }) => { + 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 }; + }; } diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index db3a56e64f..5248da6eaa 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -1633,8 +1633,11 @@ export class Consumer { */ follow(path: Path.Valid): announce.Consumer { this.#scope.path(path); + const producer = new announce.Producer(); + const patterns = this.#scope.patterns(Path.Pattern.subtree(path)); // Hiding narrows discovery, not lookup, so a hidden path follows like any other. - return announce.follow(path, this.announced(Path.Pattern.subtree(path), { hidden: true })); + void this.#runAnnounced(producer, patterns, true, announce.follower(path)); + return producer.consume(); } /** One snapshot shared by map readers and announcement-stream diffing. */ @@ -1671,7 +1674,17 @@ export class Consumer { return next; } - async #runAnnounced(producer: announce.Producer, patterns: Path.Patterns, hidden: boolean): Promise { + async #runAnnounced( + producer: announce.Producer, + patterns: Path.Patterns, + hidden: boolean, + reduce: (event: announce.Event) => announce.Event | undefined = (event) => event, + ): Promise { + const append = (event: announce.Event) => { + const reduced = reduce(event); + if (reduced) producer.append(reduced); + }; + // Keyed by the presented path (from the origin, not the scope), valued by identity // plus route. Diffing the instance rather than mere presence means a republish emits a // restart; a re-price of the same instance emits an update. @@ -1688,7 +1701,7 @@ export class Consumer { for (const [path, snap] of active) { if (!next.has(path)) - producer.append({ + append({ prefix: path, captures: snap.captures, kind: "end", @@ -1698,11 +1711,11 @@ export class Consumer { for (const [path, snap] of next) { const prev = active.get(path); if (!prev) { - producer.append({ prefix: path, captures: snap.captures, kind: "start", route: snap.route }); + append({ prefix: path, captures: snap.captures, kind: "start", route: snap.route }); } else if (!sameInstance(prev, snap)) { - producer.append({ prefix: path, captures: snap.captures, kind: "restart", route: snap.route }); + append({ prefix: path, captures: snap.captures, kind: "restart", route: snap.route }); } else if (!routesEqual(prev.route, snap.route)) { - producer.append({ prefix: path, captures: snap.captures, kind: "update", route: snap.route }); + append({ prefix: path, captures: snap.captures, kind: "update", route: snap.route }); } } active = next; diff --git a/js/watch/src/broadcast.test.ts b/js/watch/src/broadcast.test.ts index 594ca34421..bbb48e9b2c 100644 --- a/js/watch/src/broadcast.test.ts +++ b/js/watch/src/broadcast.test.ts @@ -313,7 +313,7 @@ describe("restart", () => { const old = source.out.active.peek(); expect(old).toBeDefined(); - // A newer publisher announces the exact path: a start, not a restart, for this name. + // A newer publisher announces the exact path, which now serves the name: a restart. const exact = owner.createBroadcast(name); exact.announce({ epoch: Moq.Epoch.mint() }); await settle(); @@ -333,9 +333,64 @@ describe("restart", () => { owner.close(); await settle(); }); + + it("falls back to a covering prefix when the exact route ends", async () => { + const owner = new Origin.Producer(); + const name = Path.from("pool/job.hang"); + const pool = owner.dynamic(Path.from("pool")); + const requests = pool.requested(); + const upstream = new Moq.Broadcast.Producer(); + void (async () => { + for await (const request of requests) request.accept(upstream); + })(); + const exact = publish(owner, name); + const source = new Broadcast({ origin: owner, name, enabled: true, announced: true, catalogFormat: "manual" }); + await settle(); + const old = source.out.active.peek(); + expect(old).toBeDefined(); + + // The exact route goes while the prefix still covers the name: a restart onto the prefix. + exact.close(); + await settle(); + const fallback = source.out.active.peek(); + expect(fallback).toBeDefined(); + expect(fallback).not.toBe(old); + expect(source.out.error.peek()).toBeUndefined(); + + // Nothing covers the name any more: offline, not an error. + pool.close(); + await settle(); + expect(source.out.active.peek()).toBeUndefined(); + expect(source.out.error.peek()).toBeUndefined(); + + upstream.close(); + source.close(); + owner.close(); + await settle(); + }); }); describe("refusal", () => { + it("refuses a name outside the origin's scope", async () => { + const owner = new Origin.Producer(); + const scoped = owner.scope(Path.empty(), new Path.Patterns([Path.Pattern.subtree(Path.from("room"))])); + const source = new Broadcast({ + origin: scoped, + name: Path.from("lobby/live.hang"), + enabled: true, + announced: true, + catalogFormat: "manual", + }); + await settle(); + + expect(source.out.error.peek()?.message).toBe("path is outside the origin scope"); + expect(source.out.status.peek()).toBe("error"); + + source.close(); + owner.close(); + await settle(); + }); + for (const announced of [true, false]) { it(`reports a refusal until a fresh request (announced: ${announced})`, async () => { const owner = new Origin.Producer(); diff --git a/js/watch/src/broadcast.ts b/js/watch/src/broadcast.ts index b5e4993d0e..e82da627aa 100644 --- a/js/watch/src/broadcast.ts +++ b/js/watch/src/broadcast.ts @@ -260,9 +260,10 @@ export class Broadcast { // Subscribe to the broadcast, by default waiting for its announcement so we never race a // publisher that comes online after us. A request stays on the publisher instance it resolved - // and ends once that stops serving, so a republish or reconnect is followed by requesting - // again whenever the path is announced (a start or a restart); the request ending is never - // the trigger. Mirror its handle into `active`, and a refusal into `error`. + // and ends once that stops serving, so a republish, a reconnect, or a covering prefix taking + // over is followed by requesting again whenever the route serving the path starts or restarts; + // the request ending is never the trigger. Mirror its handle into `active`, and a refusal + // into `error`. #runBroadcast(effect: Effect): void { const enabled = effect.get(this.in.enabled); if (!enabled) return; @@ -273,21 +274,29 @@ export class Broadcast { const name = effect.get(this.in.name); const announced = effect.get(this.in.announced); + // Observed whether or not the first request waits for an announcement: announcements are + // the only restart signal. Each start or restart of the route serving the name requests + // afresh, and the fresh request replaces the current one unless both resolve the same + // broadcast. Following first means a name outside the origin's scope is refused here, + // before any request, and reported like any other refusal. + let stream: Moq.Announce.Consumer; + try { + stream = origin.follow(name); + } catch (err) { + effect.set(this.#out.error, err instanceof Error ? err : new Error(String(err)), undefined); + return; + } + effect.cleanup(() => stream.close()); + const current = new Signal(origin.request(name, { announced })); effect.cleanup(() => current.peek().close()); - // Observed whether or not the first request waits for an announcement: announcements are - // the only restart signal. Each start or restart covering the name requests afresh, and - // the fresh request replaces the current one unless both resolve the same broadcast. - const stream = origin.announced(Path.Pattern.literal(name), { hidden: true }); - effect.cleanup(() => stream.close()); effect.spawn(async () => { for (;;) { const entry = await effect.race(stream.next()); // An event that settled just before teardown still resumes here: open nothing then. if (!entry || effect.abort.aborted) break; if (entry.kind === "end" || entry.kind === "update") continue; - if (!Path.hasPrefix(entry.prefix, name)) continue; const previous = current.peek(); const fresh = origin.request(name, { announced }); const was = previous.active.peek(); From 69801c089cc511c594e792d7ed6a9d3d92c37ffe Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 21:36:33 -0700 Subject: [PATCH 05/11] quest: drop Required links to quests that merged (#5138, #5141) `quest check` fails on main: both blockers landed and their quests were deleted. Co-Authored-By: Claude Opus 5.5 --- quest/m1/js-restart-keeps-request.md | 4 ---- quest/m1/webgpu-safari.md | 4 ---- 2 files changed, 8 deletions(-) diff --git a/quest/m1/js-restart-keeps-request.md b/quest/m1/js-restart-keeps-request.md index 2587b3a409..6992b22e9d 100644 --- a/quest/m1/js-restart-keeps-request.md +++ b/quest/m1/js-restart-keeps-request.md @@ -26,7 +26,3 @@ both languages already cover (`a_newer_epoch_leaves_the_broadcast_in_flight`). Add the test to `js/net/src/origin.test.ts`, failing before the fix. Public API: behavior only. Wire: none. - -## Required - -- [Dynamic epoch](/quest/m0/broadcast-epoch/dynamic-epoch.md) - #5141, which corrects the Rust `Dynamic::update` doc this quest aligns JS to diff --git a/quest/m1/webgpu-safari.md b/quest/m1/webgpu-safari.md index a37c812c4c..0089eeb11b 100644 --- a/quest/m1/webgpu-safari.md +++ b/quest/m1/webgpu-safari.md @@ -27,10 +27,6 @@ instead of guessing. Public API: possibly a new `RendererError` value. Wire: none. -## Required - -- [WebGPU renderer](/quest/m1/webgpu-renderer.md) - the renderer and `@moq/video` package (#5138) this verifies - ## Related - [WebGPU HDR](/quest/m1/webgpu-hdr.md) - builds on the same renderer From b8b0bf5b4d3769f838ca65f36c76131375037f4d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 21:54:03 -0700 Subject: [PATCH 06/11] fix(net): follow scopes to the path itself, not its subtree A route claiming only paths beneath the followed one (a scoped dynamic at the same prefix) could win that prefix on cost and mask the route that serves the path, so a republish of the real one produced no restart. Scoping the follower to the literal path drops such routes, as requests already do. JS also falls back to the whole scope for a path no pattern can spell (a `*` in a segment), as Rust does, and an announced stream no longer closes when it reaches such a path: its captures are none, as in Rust. Co-Authored-By: Claude Opus 5.5 --- js/net/src/internal.ts | 13 +++++++++-- js/net/src/origin.test.ts | 36 +++++++++++++++++++++++++++++ js/net/src/origin.ts | 12 +++++++++- rs/moq-net/src/model/follow.rs | 41 +++++++++++++++++++++++++++++++++- rs/moq-net/src/model/origin.rs | 12 +++++----- 5 files changed, 105 insertions(+), 9 deletions(-) diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 0fa1f7f737..b04f4c6526 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -83,9 +83,18 @@ export function scopeOverlaps(scope: Path.Pattern, prefix: Path.Valid): boolean return scope.overlaps(Path.Pattern.subtree(prefix)); } -/** What `scope` captures from an exact announced prefix, if it pins every wildcard. */ +/** + * What `scope` captures from an exact announced prefix, if it pins every wildcard. A prefix no + * pattern can spell (a segment containing `*`) captures nothing, as in Rust. + */ export function scopeCaptures(scope: Path.Pattern, prefix: Path.Valid): Path.Pattern[] | undefined { - return scope.captures(Path.Pattern.literal(prefix)); + let literal: Path.Pattern; + try { + literal = Path.Pattern.literal(prefix); + } catch { + return undefined; + } + return scope.captures(literal); } /** diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index 1ec78e1c48..7d1c5a12da 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -390,6 +390,42 @@ test("follow restarts onto a more specific route without an epoch", async () => 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 accepts a path no pattern can spell", async () => { + const origin = new Producer(); + const path = Path.from("camera*main"); + const follow = origin.consume().follow(path); + + const broadcast = publish(origin, path); + expect(await follow.next()).toMatchObject({ kind: "start", prefix: path }); + + follow.close(); + broadcast.close(); + 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"))])); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index bf7acbad2d..c5532863c1 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -1640,7 +1640,17 @@ export class Consumer { follow(path: Path.Valid): announce.Consumer { this.#scope.path(path); const producer = new announce.Producer(); - const patterns = this.#scope.patterns(Path.Pattern.subtree(path)); + // Scoped to the path itself, which still sees every route whose claim covers it, so a + // route claiming only paths beneath it never masks the one that serves it, just as a + // request skips it. A path no pattern can spell (a segment containing `*`) watches the + // whole scope instead, and the follower skips what does not cover it. + let scope: Path.Pattern; + try { + scope = Path.Pattern.literal(path); + } catch { + scope = Path.Pattern.all(); + } + const patterns = this.#scope.patterns(scope); // Hiding narrows discovery, not lookup, so a hidden path follows like any other. void this.#runAnnounced(producer, patterns, true, announce.follower(path)); return producer.consume(); diff --git a/rs/moq-net/src/model/follow.rs b/rs/moq-net/src/model/follow.rs index ddd4b1c7db..7b13603677 100644 --- a/rs/moq-net/src/model/follow.rs +++ b/rs/moq-net/src/model/follow.rs @@ -103,7 +103,10 @@ mod tests { use super::super::ProduceTest; use super::*; - use crate::{Epoch, Error, Hop, Pattern, Patterns, origin::Route}; + use crate::{ + Epoch, Error, Hop, Pattern, Patterns, + origin::{Cost, Route}, + }; /// The follower's next event, which must arrive within the origin's update hold. async fn followed(follow: &mut Follow) -> (&'static str, String) { @@ -180,6 +183,42 @@ mod tests { assert_eq!(followed(&mut follow).await, ("restart", "pool/job".into())); } + /// A route whose claim covers only paths beneath the followed one never serves it, so it + /// cannot mask the route that does, even when it would win the prefix on cost. + #[moq_net_sim::test] + async fn follow_ignores_a_route_scoped_beneath_the_path() { + let origin = Hop::new(1).unwrap().produce(); + let chat = Patterns::from("room/*/chat".parse::().unwrap()); + let _chat = origin + .scope("", &chat) + .unwrap() + .dynamic("room/alice", Route::default()) + .unwrap(); + let mut follow = origin.consume().follow("room/alice").unwrap(); + + let first = origin.publish("room/alice", Route::default().with_cost(5)).unwrap(); + let event = moq_net_sim::timeout(Duration::from_secs(1), follow.next()) + .await + .unwrap() + .unwrap(); + assert!( + matches!(&event, AnnounceEvent::Start(announce) if announce.route.cost == Cost::new(5)), + "{event:?}" + ); + + // A republish is another instance of the route that serves the path. + let _second = origin.publish("room/alice", Route::default().with_cost(5)).unwrap(); + drop(first); + let event = moq_net_sim::timeout(Duration::from_secs(1), follow.next()) + .await + .unwrap() + .unwrap(); + assert!( + matches!(&event, AnnounceEvent::Restart(announce) if announce.route.cost == Cost::new(5)), + "{event:?}" + ); + } + /// A path the consumer's scope can never cover fails at once instead of waiting forever. #[test] fn follow_refuses_a_path_outside_the_scope() { diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 6c2ea939b0..44ca122164 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -4472,11 +4472,13 @@ impl Consumer { pub fn follow(&self, path: impl AsPath) -> Result { let path = path.as_path(); - // Scope down to the path's subtree, so the rest of the origin never wakes it. A path - // no pattern can spell (a segment containing `*`) watches everything instead, and the - // follower skips the routes that do not cover it. - let consumer = match Pattern::subtree(path.as_str()) { - Ok(subtree) => self.scope("", &Patterns::from(subtree))?, + // Scope down to the path itself, which still sees every route whose claim covers it: + // the rest of the origin never wakes it, and a route claiming only paths beneath it + // (a scoped dynamic at the same prefix) never masks the one that serves it, just as a + // request skips it. A path no pattern can spell (a segment containing `*`) watches + // everything instead, and the follower skips the routes that do not cover it. + let consumer = match Pattern::literal(path.as_str()) { + Ok(literal) => self.scope("", &Patterns::from(literal))?, Err(_) => self.clone(), }; From 08164d7d7a01dcfe714059639289a5616490ebd4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 21:54:03 -0700 Subject: [PATCH 07/11] fix(watch): key a video restart on the media the rendition reads The restart check read the catalog broadcast's instance, so a rendition reading another broadcast through a `broadcast` override missed that one's republish and froze, while a catalog-only republish reset the clock with the media unchanged. Key it on the resolved media's path and instance. Co-Authored-By: Claude Opus 5.5 --- js/watch/src/video/decoder.test.ts | 73 +++++++++++++++++++++++++++--- js/watch/src/video/decoder.ts | 17 ++++--- 2 files changed, 74 insertions(+), 16 deletions(-) diff --git a/js/watch/src/video/decoder.test.ts b/js/watch/src/video/decoder.test.ts index 641fe2383f..71ba45a02c 100644 --- a/js/watch/src/video/decoder.test.ts +++ b/js/watch/src/video/decoder.test.ts @@ -24,14 +24,17 @@ function config(fields: Record): Catalog.VideoConfig { } // The subscription is already closed, so the track never reaches VideoDecoder. -function closedConsumer(): Moq.Broadcast.Consumer { - return { closed: new Signal("ended") } as unknown as Moq.Broadcast.Consumer; +function closedConsumer(path: string): Moq.Broadcast.Consumer { + return { closed: new Signal("ended"), path } as unknown as Moq.Broadcast.Consumer; } +// Each source is another broadcast, at its own path. +let sources = 0; function watchBroadcast(catalog: Signal): Broadcast { + const media = closedConsumer(`source-${sources++}`); return { out: { catalog, active: new Signal(undefined) }, - relativeBroadcast: () => closedConsumer(), + relativeBroadcast: () => media, } as unknown as Broadcast; } @@ -245,13 +248,23 @@ async function guardedPlayback(kind: "legacy" | "cmaf", reads: Read[], hold: rea const video = config({ codedWidth: 16, codedHeight: 16, description: "01640028ffe100046764002801000268ee" }); if (kind === "cmaf") video.container = { kind, init: Buffer.from(Container.Cmaf.createVideoInitSegment(video)).toString("base64") }; + // The catalog's own broadcast, and the one the rendition reads. They differ once only one of + // them is republished, as with a rendition's `broadcast` override. let producer = new Moq.Broadcast.Producer(); + let mediaProducer = producer; const active = new Signal(producer.consume()); + const media = new Signal(producer.consume()); const catalog = new Signal({ video: { renditions: { video } } }); const broadcast = new Signal({ out: { catalog, active }, - relativeBroadcast: (effect: Effect) => effect.get(active), + relativeBroadcast: (effect: Effect) => effect.get(media), } as unknown as Broadcast); + const closeAll = () => { + active.peek().close(); + media.peek().close(); + producer.close(); + mediaProducer.close(); + }; const source = new Source({ broadcast, supported: async () => true }); const sync = new Sync({ delay: "instant" }); const enabled = new Signal(true); @@ -269,17 +282,29 @@ async function guardedPlayback(kind: "legacy" | "cmaf", reads: Read[], hold: rea }, // A restarted publisher takes the name: another instance of the same broadcast. republish() { + closeAll(); + producer = new Moq.Broadcast.Producer(); + mediaProducer = producer; + active.set(producer.consume()); + media.set(producer.consume()); + }, + // Only the catalog's broadcast restarts; the media the rendition reads stays up. + republishCatalog() { active.peek().close(); - producer.close(); producer = new Moq.Broadcast.Producer(); active.set(producer.consume()); }, + // Only the media the rendition reads restarts, under the same catalog. + republishMedia() { + media.peek().close(); + mediaProducer = new Moq.Broadcast.Producer(); + media.set(mediaProducer.consume()); + }, close() { decoder.close(); source.close(); sync.close(); - active.peek().close(); - producer.close(); + closeAll(); next.mockRestore(); for (const [name, original] of [ ["VideoDecoder", originalDecoder], @@ -329,6 +354,40 @@ it("switches to a republished broadcast at once, its timeline starting over", as } }); +it("switches at once when only the media a rendition reads is republished", async () => { + const reads = [sample(10, 10_000_000)]; + const playback = await guardedPlayback("legacy", reads); + try { + const reset = spyOn(playback.sync, "reset"); + + reads.push(sample(0, 1000)); + playback.republishMedia(); + await microtasks(); + + expect(playback.decoder.out.timestamp.peek()).toBe(Time.Milli(1)); + expect(reset).toHaveBeenCalledTimes(1); + } finally { + playback.close(); + } +}); + +it("keeps the clock when only the catalog's broadcast is republished", async () => { + const reads = [sample(10, 10_000_000)]; + const playback = await guardedPlayback("legacy", reads); + try { + const reset = spyOn(playback.sync, "reset"); + + playback.republishCatalog(); + await microtasks(); + + // The media is the same instance, so its timeline did not start over. + expect(playback.decoder.out.timestamp.peek()).toBe(Time.Milli(10_000)); + expect(reset).not.toHaveBeenCalled(); + } finally { + playback.close(); + } +}); + it("clears the picture once no rendition is selectable", async () => { const playback = await guardedPlayback("legacy", [sample(10, 1000)]); try { diff --git a/js/watch/src/video/decoder.ts b/js/watch/src/video/decoder.ts index 94555fe0ec..9a4986e54d 100644 --- a/js/watch/src/video/decoder.ts +++ b/js/watch/src/video/decoder.ts @@ -15,7 +15,6 @@ import { } from "@moq/signals"; import { rotateVideoDimensions } from "@moq/video"; import { base64ToBytes } from "../base64"; -import type { Broadcast } from "../broadcast"; import { nextMedia, subscribeMedia } from "../media"; import type { Sync } from "../sync"; @@ -99,9 +98,9 @@ export class Decoder { #pending = new Signal<{ track: string; catalog: Getter } | undefined>(undefined); // The active track's arrival estimate from its container consumer. #spread = new Signal(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 }; + // The path of the media the last rendition read, and its instance. Every handle to one instance + // shares `closed`, so it identifies the instance where the handle itself does not. + #instance?: { path: Moq.Path.Valid; closed: unknown }; readonly #identity: Computed; #signals = new Effect(); @@ -192,13 +191,13 @@ export class Decoder { return; } - // A republish at the same name is another instance, whose timeline starts over. Waiting for + // A republish of the media this rendition reads (the catalog's own broadcast, or the one a + // `broadcast` override names) 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 }; + // switch to another path, still waits. + const republished = this.#instance?.path === active.path && this.#instance.closed !== active.closed; + this.#instance = { path: active.path, closed: active.closed }; if (republished) this.sync.reset(); // Start a new pending effect. From e0ec0552208ae4f14dca294c6c89b06dc8944516 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 22:06:34 -0700 Subject: [PATCH 08/11] fix(net): follow refuses a path no pattern can spell Falling back to the whole scope for a path with a `*` segment lets a route that never serves it win its prefix, and JS announcements already stop at such a prefix. Refuse it instead (`Error::InvalidPath`, a throw in JS), as `routed` did before it was built on `follow`, and drop the partial JS captures guard. Co-Authored-By: Claude Opus 5.5 --- js/net/src/internal.ts | 13 ++----------- js/net/src/origin.test.ts | 11 ++--------- js/net/src/origin.ts | 14 ++++---------- rs/moq-net/src/model/follow.rs | 11 +++++++++++ rs/moq-net/src/model/origin.rs | 11 ++++------- 5 files changed, 23 insertions(+), 37 deletions(-) diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index b04f4c6526..0fa1f7f737 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -83,18 +83,9 @@ export function scopeOverlaps(scope: Path.Pattern, prefix: Path.Valid): boolean return scope.overlaps(Path.Pattern.subtree(prefix)); } -/** - * What `scope` captures from an exact announced prefix, if it pins every wildcard. A prefix no - * pattern can spell (a segment containing `*`) captures nothing, as in Rust. - */ +/** What `scope` captures from an exact announced prefix, if it pins every wildcard. */ export function scopeCaptures(scope: Path.Pattern, prefix: Path.Valid): Path.Pattern[] | undefined { - let literal: Path.Pattern; - try { - literal = Path.Pattern.literal(prefix); - } catch { - return undefined; - } - return scope.captures(literal); + return scope.captures(Path.Pattern.literal(prefix)); } /** diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index 7d1c5a12da..0c7bda70de 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -413,16 +413,9 @@ test("follow ignores a route scoped beneath the path", async () => { origin.close(); }); -test("follow accepts a path no pattern can spell", async () => { +test("follow refuses a path no pattern can spell", () => { const origin = new Producer(); - const path = Path.from("camera*main"); - const follow = origin.consume().follow(path); - - const broadcast = publish(origin, path); - expect(await follow.next()).toMatchObject({ kind: "start", prefix: path }); - - follow.close(); - broadcast.close(); + expect(() => origin.consume().follow(Path.from("camera*main"))).toThrow(); origin.close(); }); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index c5532863c1..7c973aee05 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -1635,22 +1635,16 @@ export class Consumer { * everything and request the path afresh on `restart`, and stop on `end`. An `update` is * the same publisher instance re-priced or failed over, which subscriptions already ride * out. Another route taking over is a `restart`, or an `update` when both carry the same - * epoch. Throws when the path is outside this consumer's scope. Close it when done. + * epoch. Throws when the path is outside this consumer's scope, or when no pattern can spell + * it (a segment containing `*`). Close it when done. */ follow(path: Path.Valid): announce.Consumer { this.#scope.path(path); const producer = new announce.Producer(); // Scoped to the path itself, which still sees every route whose claim covers it, so a // route claiming only paths beneath it never masks the one that serves it, just as a - // request skips it. A path no pattern can spell (a segment containing `*`) watches the - // whole scope instead, and the follower skips what does not cover it. - let scope: Path.Pattern; - try { - scope = Path.Pattern.literal(path); - } catch { - scope = Path.Pattern.all(); - } - const patterns = this.#scope.patterns(scope); + // request skips it. A path no pattern can spell (a segment containing `*`) throws here. + const patterns = this.#scope.patterns(Path.Pattern.literal(path)); // Hiding narrows discovery, not lookup, so a hidden path follows like any other. void this.#runAnnounced(producer, patterns, true, announce.follower(path)); return producer.consume(); diff --git a/rs/moq-net/src/model/follow.rs b/rs/moq-net/src/model/follow.rs index 7b13603677..c466ce0645 100644 --- a/rs/moq-net/src/model/follow.rs +++ b/rs/moq-net/src/model/follow.rs @@ -219,6 +219,17 @@ mod tests { ); } + /// A path no pattern can spell is refused rather than followed through the whole scope, + /// where a route that never serves it could win its prefix. + #[test] + fn follow_refuses_a_path_no_pattern_can_spell() { + let origin = Hop::new(1).unwrap().produce(); + assert!(matches!( + origin.consume().follow("camera*main"), + Err(Error::InvalidPath(_)) + )); + } + /// A path the consumer's scope can never cover fails at once instead of waiting forever. #[test] fn follow_refuses_a_path_outside_the_scope() { diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 44ca122164..e349210637 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -4468,19 +4468,16 @@ impl Consumer { /// [`Restart`](AnnounceEvent::Restart), and stop on an [`End`](AnnounceEvent::End). An /// [`Update`](AnnounceEvent::Update) is the same publisher instance re-priced or failed /// over, which subscriptions already ride out. Fails with [`Error::Unauthorized`] when - /// this consumer's scope can never cover the path. + /// this consumer's scope can never cover the path, and with [`Error::InvalidPath`] when no + /// pattern can spell it (a segment containing `*`). pub fn follow(&self, path: impl AsPath) -> Result { let path = path.as_path(); // Scope down to the path itself, which still sees every route whose claim covers it: // the rest of the origin never wakes it, and a route claiming only paths beneath it // (a scoped dynamic at the same prefix) never masks the one that serves it, just as a - // request skips it. A path no pattern can spell (a segment containing `*`) watches - // everything instead, and the follower skips the routes that do not cover it. - let consumer = match Pattern::literal(path.as_str()) { - Ok(literal) => self.scope("", &Patterns::from(literal))?, - Err(_) => self.clone(), - }; + // request skips it. + let consumer = self.scope("", &Patterns::from(Pattern::literal(path.as_str())?))?; // `scope` keeps narrower permissions intact: on a consumer limited to // `foo/specific`, no route can ever cover `foo`. From 8b175d145d0adf4181a744df2c8997518010ee2c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 22:41:11 -0700 Subject: [PATCH 09/11] fix(net): a late follower starts on the serving route The replay a follower starts with names a covering prefix before the exact path beneath it, which came out as a start on the prefix and then a restart onto the exact path, so a late-joining player requested the broadcast twice. Fold every announcement on hand (one table change in JS) into one event. Co-Authored-By: Claude Opus 5.5 --- js/net/src/announced.ts | 41 ++++++++++++++++++------ js/net/src/origin.test.ts | 17 ++++++++++ js/net/src/origin.ts | 8 +++-- rs/moq-net/src/model/follow.rs | 58 ++++++++++++++++++++++++++++++---- 4 files changed, 105 insertions(+), 19 deletions(-) diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index b043f4047a..2bdb9433b9 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -168,14 +168,16 @@ export class Consumer { * 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 (`undefined`). + * other broadcasts and are skipped, as is any change no request sees. * - * Synchronous, so a followed stream delivers in step with an announced one. - * Mirrors `announce::Follow` in rs/moq-net. + * Each batch (one change to the table) reduces to at most one event, so the replay a late + * follower starts with (a covering prefix, then the exact path beneath it) is one `start` on + * the route serving the path rather than a `start` and a `restart`. 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): (event: Event) => Event | undefined { +export function follower(path: Path.Valid): (events: Event[]) => Event[] { // Every route standing over the path, by prefix. const covering = new Map(); const serving = () => { @@ -186,21 +188,42 @@ export function follower(path: Path.Valid): (event: Event) => Event | undefined return best; }; - return ({ kind, ...announce }) => { + // Apply one route's event, returning what it means for the path alone. + const fold = ({ kind, ...announce }: Event): Event["kind"] | 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) return after && "start"; + if (!after) return "end"; if (before.prefix !== after.prefix) { const same = before.route.epoch !== undefined && before.route.epoch === after.route.epoch; - return { kind: same ? "update" : "restart", ...after }; + return same ? "update" : "restart"; } // 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 kind === "restart" ? "restart" : "update"; + }; + + return (events) => { + const before = serving(); + // Whether any event changed the serving route, and whether one changed its instance. + let changed = false; + let replaced = false; + for (const event of events) { + const kind = fold(event); + if (kind) changed = true; + if (kind && kind !== "update") replaced = true; + } + + const after = serving(); + if (!before) return after ? [{ kind: "start", ...after }] : []; + if (!after) return [{ kind: "end", ...before }]; + // Routes with one epoch serve the same bytes, whatever happened in between. + const same = before.route.epoch !== undefined && before.route.epoch === after.route.epoch; + if (replaced && !same) return [{ kind: "restart", ...after }]; + return changed ? [{ kind: "update", ...after }] : []; }; } diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index 0c7bda70de..4e90bdfc97 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -413,6 +413,23 @@ test("follow ignores a route scoped beneath the path", async () => { 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 refuses a path no pattern can spell", () => { const origin = new Producer(); expect(() => origin.consume().follow(Path.from("camera*main"))).toThrow(); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index 7c973aee05..f67bd9fdd5 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -1688,11 +1688,12 @@ export class Consumer { producer: announce.Producer, patterns: Path.Patterns, hidden: boolean, - reduce: (event: announce.Event) => announce.Event | undefined = (event) => event, + reduce: (events: announce.Event[]) => announce.Event[] = (events) => events, ): Promise { + // Each pass's changes, reduced together so a follower sees one table change as one event. + const batch: announce.Event[] = []; const append = (event: announce.Event) => { - const reduced = reduce(event); - if (reduced) producer.append(reduced); + batch.push(event); }; // Keyed by the presented path (from the origin, not the scope), valued by identity @@ -1729,6 +1730,7 @@ export class Consumer { } } active = next; + for (const event of reduce(batch.splice(0))) producer.append(event); await Signal.race(this.#state.local, this.#state.advertisedLocal, this.#state.routes, producer.closed); if (producer.closed.peek() !== undefined) return; diff --git a/rs/moq-net/src/model/follow.rs b/rs/moq-net/src/model/follow.rs index c466ce0645..8e818adbde 100644 --- a/rs/moq-net/src/model/follow.rs +++ b/rs/moq-net/src/model/follow.rs @@ -1,4 +1,4 @@ -use std::task::{Poll, ready}; +use std::task::Poll; use super::origin_impl::{Announce, AnnounceConsumer, AnnounceEvent}; use crate::PathOwned; @@ -34,14 +34,45 @@ impl Follow { } /// Poll for the next change, registering `waiter` when there is none yet. + /// + /// Every announcement already on hand is folded into one change, so the replay a late + /// follower starts with (a covering prefix, then the exact path beneath it) is one start on + /// the route serving the path rather than a start and a restart. pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll> { - loop { - let Some(event) = ready!(self.announced.poll_next(waiter)) else { - return Poll::Ready(None); - }; - if let Some(event) = self.fold(event) { - return Poll::Ready(Some(event)); + let before = self.serving().cloned(); + // Whether any announcement on hand changed the serving route, and whether one changed + // which instance it is. + let (mut changed, mut replaced) = (false, false); + let ended = loop { + match self.announced.poll_next(waiter) { + Poll::Ready(Some(event)) => match self.fold(event) { + None => {} + Some(AnnounceEvent::Update(_)) => changed = true, + Some(_) => (changed, replaced) = (true, true), + }, + Poll::Ready(None) => break true, + Poll::Pending => break false, } + }; + + let change = match (before, self.serving().cloned()) { + (None, None) => None, + (None, Some(after)) => Some(AnnounceEvent::Start(after)), + (Some(before), None) => Some(AnnounceEvent::End(before)), + (Some(before), Some(after)) => { + // Routes with one epoch serve the same bytes, whatever happened in between. + let same = before.route.epoch.is_some() && before.route.epoch == after.route.epoch; + match (replaced && !same, changed) { + (true, _) => Some(AnnounceEvent::Restart(after)), + (false, true) => Some(AnnounceEvent::Update(after)), + (false, false) => None, + } + } + }; + match (change, ended) { + (Some(change), _) => Poll::Ready(Some(change)), + (None, true) => Poll::Ready(None), + (None, false) => Poll::Pending, } } @@ -170,6 +201,19 @@ mod tests { assert_eq!(followed(&mut follow).await, ("end", "pool".into())); } + /// A follower that joins late starts on the route serving the path, not on whichever + /// covering route the replay happened to reach first. + #[moq_net_sim::test] + async fn follow_starts_on_the_serving_route_when_joining_late() { + let origin = Hop::new(1).unwrap().produce(); + let _pool = origin.dynamic("pool", Route::default()).unwrap(); + let _exact = origin.publish("pool/job", Route::default()).unwrap(); + + let mut follow = origin.consume().follow("pool/job").unwrap(); + assert_eq!(followed(&mut follow).await, ("start", "pool/job".into())); + quiet(&mut follow).await; + } + /// Without an epoch nothing says two routes serve the same bytes. #[moq_net_sim::test] async fn follow_restarts_onto_a_more_specific_route_without_an_epoch() { From 99ad89aa6e8d15a09ae40ce9775169f3ab9c1b22 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 23:13:50 -0700 Subject: [PATCH 10/11] fix(net): coalesce a follower's changes only while nothing serves the path Folding every change on hand into one event turned an end followed by a start under the same epoch into an update, which hid a gap that had already ended the request on the old route, so @moq/watch never asked again. Fold only while the path is unserved (the late-join replay), and pass each change through once something serves it. Co-Authored-By: Claude Opus 5.5 --- js/net/src/announced.ts | 43 +++++++++--------- js/net/src/origin.test.ts | 20 +++++++++ js/watch/src/broadcast.test.ts | 33 ++++++++++++++ rs/moq-net/src/model/follow.rs | 81 +++++++++++++++++++--------------- 4 files changed, 118 insertions(+), 59 deletions(-) diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index 2bdb9433b9..7efdf2f73c 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -170,10 +170,13 @@ export class Consumer { * 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. * - * Each batch (one change to the table) reduces to at most one event, so the replay a late - * follower starts with (a covering prefix, then the exact path beneath it) is one `start` on - * the route serving the path rather than a `start` and a `restart`. Synchronous, so a followed - * stream delivers in step with an announced one. Mirrors `announce::Follow` in rs/moq-net. + * 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 */ @@ -189,41 +192,35 @@ export function follower(path: Path.Valid): (events: Event[]) => Event[] { }; // Apply one route's event, returning what it means for the path alone. - const fold = ({ kind, ...announce }: Event): Event["kind"] | undefined => { + 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 && "start"; - if (!after) return "end"; + 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 same ? "update" : "restart"; + 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 === "restart" ? "restart" : "update"; + return { kind: kind === "restart" ? "restart" : "update", ...after }; }; return (events) => { - const before = serving(); - // Whether any event changed the serving route, and whether one changed its instance. - let changed = false; - let replaced = false; + const reduced: Event[] = []; + let idle = !serving(); for (const event of events) { - const kind = fold(event); - if (kind) changed = true; - if (kind && kind !== "update") replaced = true; + const change = fold(event); + if (idle || !change) continue; + reduced.push(change); + idle = change.kind === "end"; } - const after = serving(); - if (!before) return after ? [{ kind: "start", ...after }] : []; - if (!after) return [{ kind: "end", ...before }]; - // Routes with one epoch serve the same bytes, whatever happened in between. - const same = before.route.epoch !== undefined && before.route.epoch === after.route.epoch; - if (replaced && !same) return [{ kind: "restart", ...after }]; - return changed ? [{ kind: "update", ...after }] : []; + if (idle && after) reduced.push({ kind: "start", ...after }); + return reduced; }; } diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index 4e90bdfc97..f790460e9d 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -430,6 +430,26 @@ test("follow starts on the serving route when joining late", async () => { 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(); diff --git a/js/watch/src/broadcast.test.ts b/js/watch/src/broadcast.test.ts index bbb48e9b2c..d5d3b7be68 100644 --- a/js/watch/src/broadcast.test.ts +++ b/js/watch/src/broadcast.test.ts @@ -334,6 +334,39 @@ describe("restart", () => { await settle(); }); + it("resumes on a covering prefix of the same epoch after a gap", async () => { + const owner = new Origin.Producer(); + const name = Path.from("pool/job.hang"); + const epoch = Moq.Epoch.mint(); + const exact = owner.createBroadcast(name); + exact.announce({ epoch }); + const source = new Broadcast({ origin: owner, name, enabled: true, announced: true, catalogFormat: "manual" }); + await settle(); + const old = source.out.active.peek(); + expect(old).toBeDefined(); + + // The exact route goes, which ends the request on it, and the prefix arrives under the same + // epoch before the table is read again. The player still has to ask again. + exact.close(); + const pool = owner.dynamic(Path.from("pool"), { epoch }); + const requests = pool.requested(); + const upstream = new Moq.Broadcast.Producer(); + void (async () => { + for await (const request of requests) request.accept(upstream); + })(); + await settle(); + const resumed = source.out.active.peek(); + expect(resumed).toBeDefined(); + expect(resumed).not.toBe(old); + expect(source.out.error.peek()).toBeUndefined(); + + upstream.close(); + pool.close(); + source.close(); + owner.close(); + await settle(); + }); + it("falls back to a covering prefix when the exact route ends", async () => { const owner = new Origin.Producer(); const name = Path.from("pool/job.hang"); diff --git a/rs/moq-net/src/model/follow.rs b/rs/moq-net/src/model/follow.rs index 8e818adbde..f8dd57ddc4 100644 --- a/rs/moq-net/src/model/follow.rs +++ b/rs/moq-net/src/model/follow.rs @@ -1,4 +1,4 @@ -use std::task::Poll; +use std::task::{Poll, ready}; use super::origin_impl::{Announce, AnnounceConsumer, AnnounceEvent}; use crate::PathOwned; @@ -35,44 +35,37 @@ impl Follow { /// Poll for the next change, registering `waiter` when there is none yet. /// - /// Every announcement already on hand is folded into one change, so the replay a late - /// follower starts with (a covering prefix, then the exact path beneath it) is one start on - /// the route serving the path rather than a start and a restart. + /// While nothing serves the path, every announcement already on hand is folded 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. pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll> { - let before = self.serving().cloned(); - // Whether any announcement on hand changed the serving route, and whether one changed - // which instance it is. - let (mut changed, mut replaced) = (false, false); - let ended = loop { - match self.announced.poll_next(waiter) { - Poll::Ready(Some(event)) => match self.fold(event) { - None => {} - Some(AnnounceEvent::Update(_)) => changed = true, - Some(_) => (changed, replaced) = (true, true), - }, - Poll::Ready(None) => break true, - Poll::Pending => break false, - } - }; - - let change = match (before, self.serving().cloned()) { - (None, None) => None, - (None, Some(after)) => Some(AnnounceEvent::Start(after)), - (Some(before), None) => Some(AnnounceEvent::End(before)), - (Some(before), Some(after)) => { - // Routes with one epoch serve the same bytes, whatever happened in between. - let same = before.route.epoch.is_some() && before.route.epoch == after.route.epoch; - match (replaced && !same, changed) { - (true, _) => Some(AnnounceEvent::Restart(after)), - (false, true) => Some(AnnounceEvent::Update(after)), - (false, false) => None, + if self.serving().is_none() { + let ended = loop { + match self.announced.poll_next(waiter) { + Poll::Ready(Some(event)) => { + self.fold(event); + } + Poll::Ready(None) => break true, + Poll::Pending => break false, } + }; + return match (self.serving().cloned(), ended) { + (Some(serving), _) => Poll::Ready(Some(AnnounceEvent::Start(serving))), + (None, true) => Poll::Ready(None), + (None, false) => Poll::Pending, + }; + } + + loop { + let Some(event) = ready!(self.announced.poll_next(waiter)) else { + return Poll::Ready(None); + }; + if let Some(event) = self.fold(event) { + return Poll::Ready(Some(event)); } - }; - match (change, ended) { - (Some(change), _) => Poll::Ready(Some(change)), - (None, true) => Poll::Ready(None), - (None, false) => Poll::Pending, } } @@ -214,6 +207,22 @@ mod tests { quiet(&mut follow).await; } + /// A gap with nothing serving the path ends any request on it, so a covering route that + /// arrives after it is a start of its own, even under the same epoch. + #[moq_net_sim::test] + async fn follow_keeps_a_gap_between_routes_of_one_epoch() { + let origin = Hop::new(1).unwrap().produce(); + let epoch = Epoch::mint(); + let exact = origin.publish("pool/job", route(&epoch)).unwrap(); + let mut follow = origin.consume().follow("pool/job").unwrap(); + assert_eq!(followed(&mut follow).await, ("start", "pool/job".into())); + + drop(exact); + assert_eq!(followed(&mut follow).await, ("end", "pool/job".into())); + let _pool = origin.dynamic("pool", route(&epoch)).unwrap(); + assert_eq!(followed(&mut follow).await, ("start", "pool".into())); + } + /// Without an epoch nothing says two routes serve the same bytes. #[moq_net_sim::test] async fn follow_restarts_onto_a_more_specific_route_without_an_epoch() { From b3aed0198e79149cb7610d80d14921df8bc1e205 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 23:28:25 -0700 Subject: [PATCH 11/11] fix(moq-cli): an idle player plays an update A covering route of one epoch arriving as the exact route goes, both before the follower is polled, reaches it as an update: the announce cursor delivers by prefix, so the path never shows unserved. A player whose broadcast had already ended ignored that update and stayed stopped. With nothing playing, an update now plays the path. Co-Authored-By: Claude Opus 5.5 --- rs/moq-cli/src/play/media.rs | 49 ++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/rs/moq-cli/src/play/media.rs b/rs/moq-cli/src/play/media.rs index 9c26d65bbb..fe9d4f406b 100644 --- a/rs/moq-cli/src/play/media.rs +++ b/rs/moq-cli/src/play/media.rs @@ -79,6 +79,11 @@ impl Media { let Some(event) = event else { return Ok(()) }; let announce = match event { AnnounceEvent::Start(announce) | AnnounceEvent::Restart(announce) => announce, + // The same instance, which what is playing rides out. With nothing + // playing, it is the path served again after a gap the announcements + // folded away (a covering route of one epoch arriving as the exact one + // went), so play it. + AnnounceEvent::Update(announce) if playing.is_none() => announce, AnnounceEvent::Update(_) => continue, AnnounceEvent::End(_) => { tracing::info!(broadcast = %self.broadcast, "offline, waiting for it to return"); @@ -768,6 +773,50 @@ mod tests { ); } + /// A covering route of the same epoch that arrives as the exact route goes, both + /// before the player looks, reaches it as an update: the announcements never show + /// the path unserved. A player with nothing playing still plays it. + #[tokio::test] + async fn an_update_while_idle_plays_the_route() { + tokio::time::pause(); + + const OLD: f32 = 0.25; + const NEW: f32 = 0.5; + let origin = moq_tokio::origin::spawn(); + let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + + let (exact, mut exact_catalog, mut old) = publish(&origin, route.clone()); + let recorder = Recorder::default(); + let player = tokio::spawn(media(&origin, Duration::from_millis(50), recorder.clone()).run()); + write(&mut old, 10, OLD).await; + old.finish().unwrap(); + exact_catalog.finish().unwrap(); + drop((exact_catalog, old)); + // The broadcast ends while its route stands, so the player goes idle. + tokio::time::sleep(Duration::from_secs(60)).await; + + let mut served = moq_net::broadcast::Info::new().produce(); + let served_catalog = catalog::Producer::new(&mut served, Default::default()).unwrap(); + let mut new = rendition(&served, &served_catalog, "audio"); + let pool = origin.dynamic("", route).unwrap(); + drop(exact); + let consumer = served.consume(); + tokio::spawn(async move { + while let Ok(request) = pool.requested_broadcast().await { + request.accept(consumer.clone()); + } + }); + write(&mut new, 10, NEW).await; + + settle(player, &recorder).await; + assert_eq!( + played(&recorder, NEW).0, + PACKET_DURATION * 10, + "the idle player never played the covering route" + ); + drop((served, served_catalog)); + } + /// Re-pricing the same instance is an update, which playback rides through on /// one sink rather than starting over. #[tokio::test]