From 8740c7a11792a892f625e183bdd72d4d1ee93c43 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 23:13:42 -0700 Subject: [PATCH 1/5] quest: claim quest/m0/broadcast-epoch/apps From 6a0bdf1b9b82599d2691d2149974771871a15a8c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 23:46:41 -0700 Subject: [PATCH 2/5] feat(apps): follow a republished broadcast and log publisher epochs moq play restarts on a new publisher instance when the old one ends with Unroutable, and @moq/watch switches video at once and re-anchors the shared clock on a republish. Publishers in moq-cli and moq-boy log the epoch they announce. Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 13 ++-- doc/lib/js/watch.md | 5 ++ js/net/src/lite/publisher.ts | 2 +- js/net/src/lite/subscriber.ts | 2 +- js/watch/src/audio/decoder.test.ts | 5 +- js/watch/src/audio/decoder.ts | 10 +-- js/watch/src/video/decoder.test.ts | 41 +++++++++-- js/watch/src/video/decoder.ts | 15 +++- rs/moq-boy/src/main.rs | 4 +- rs/moq-cli/Cargo.toml | 4 +- rs/moq-cli/src/archive.rs | 4 +- rs/moq-cli/src/hls.rs | 4 +- rs/moq-cli/src/moq.rs | 12 +++- rs/moq-cli/src/play/media.rs | 111 +++++++++++++++++++++++++++-- rs/moq-cli/src/play/source.rs | 7 +- rs/moq-cli/src/publish.rs | 4 +- rs/moq-cli/src/rtc.rs | 4 +- rs/moq-cli/src/transcode.rs | 4 +- 18 files changed, 207 insertions(+), 44 deletions(-) diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 5949ca85ca..608c95254f 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -226,7 +226,9 @@ the newest cached group, including when a rendition is reopened, so playback does not replay the retained backlog. A publisher that retires the rendition being played ends that track and the role picks a replacement. A retired audio rendition plays out what the speaker holds while its replacement fills, so the -switch does not cost a delay of silence. Playback is +switch does not cost a delay of silence. A restarted publisher's broadcast +replaces the one playing, and playback starts over on it with a fresh catalog +and decoders; the log names each broadcast's epoch. Playback is behind the `play` feature, since it pulls in windowing and audio-device dependencies: @@ -391,9 +393,12 @@ moq --connect https://relay.example.com/anon \ Two publishers of the same broadcast name are interchangeable sources: relays hold both routes and fail over between them mid-group. They must -produce identical tracks with aligned groups. A restarted encoder is the same -broadcast too, so one whose groups restart from 0 must publish under a new -name, or viewers wait for its sequence to catch up. +produce identical tracks with aligned groups. Each run announces a fresh +[epoch](/concept/moq-lite#publisher-epochs), logged at startup. Over +moq-lite 07 (opt-in while its wire is in progress) the epoch reaches the relay, +so a restarted publisher replaces the old broadcast and viewers switch to it at +once. Older versions carry no epoch, so viewers stay on the old route until it +goes. ## Cluster diff --git a/doc/lib/js/watch.md b/doc/lib/js/watch.md index ded3988fe7..ff6acd0a98 100644 --- a/doc/lib/js/watch.md +++ b/doc/lib/js/watch.md @@ -44,6 +44,11 @@ Video holds its last picture while paused, out of view, or waiting for a resumed rendition's first frame. Its reported timestamp stays with that picture. Going offline or closing the player clears it. +A restarted publisher announces a new [epoch](/concept/moq-lite#publisher-epochs) +at the same name, and the player follows it: a fresh catalog, and video, audio, +and the clock start over on its timeline instead of waiting for it to catch up +to the old one. + The overlay adds play/pause, volume, fullscreen, a quality selector, a buffering indicator, an unsupported-codec warning, and a stats panel. `` shows what the browser can play. diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index 09a67a90a4..5e3d3e5247 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -430,7 +430,7 @@ export class Publisher { const onWire = (route: Route): Route => (hasRouteCost(this.version) ? route : { ...route, cost: Cost.zero }); const announce = async (suffix: Path.Valid, route: Route) => { - console.debug(`announce: broadcast=${suffix} active=true`); + console.debug(`announce: broadcast=${suffix} active=true epoch=${route.epoch}`); if (hasAnnounceId(this.version)) announceIds.set(suffix, nextAnnounceId++); await encodeAnnounceBroadcast( stream.writer, diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index eb2bae2d21..a3789b31ce 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -458,7 +458,7 @@ export class Subscriber { advertised.set(path, { live: true, route, captures }); if (epoch) this.#epochs.set(path, epoch); - console.debug(`announced: broadcast=${path} active=true`); + console.debug(`announced: broadcast=${path} active=true epoch=${epoch}`); announced.append({ prefix: path, captures, kind: "start", route }); } diff --git a/js/watch/src/audio/decoder.test.ts b/js/watch/src/audio/decoder.test.ts index a3858d2f05..86025f8412 100644 --- a/js/watch/src/audio/decoder.test.ts +++ b/js/watch/src/audio/decoder.test.ts @@ -362,12 +362,15 @@ describe("Decoder across a rendition's absence", () => { playback.remove(); await microtasks(); + const reanchor = spyOn(Sync.prototype, "reset"); playback.replace(); playback.restore(); await microtasks(); await playback.play(12); - // The ring re-anchors on the new broadcast rather than discarding it as already played. + // The ring and the shared clock re-anchor on the new broadcast rather than discarding it as + // already played. + expect(reanchor).toHaveBeenCalledTimes(1); expect(drain(ring)).toBeGreaterThan(0); expect(contexts).toHaveLength(1); expect(playback.rings()).toEqual([ring]); diff --git a/js/watch/src/audio/decoder.ts b/js/watch/src/audio/decoder.ts index cfcf455365..ce6f53d106 100644 --- a/js/watch/src/audio/decoder.ts +++ b/js/watch/src/audio/decoder.ts @@ -331,13 +331,15 @@ export class Decoder { const active = broadcast.relativeBroadcast(effect, identity.broadcast); if (!active) return; - // Another broadcast (a new name, or a republish) brings its own timeline, which a ring anchored - // on the previous one would discard as already played. A return to the same instance keeps the - // anchor, so the relay's redelivered last group is still dropped as old. Every handle to one - // instance shares `closed`, so it identifies the instance where the handle itself does not. + // Another broadcast (a new name, or a republish) brings its own timeline, which a ring and a + // clock anchored on the previous one would discard as already played. A return to the same + // instance keeps the anchor, so the relay's redelivered last group is still dropped as old. + // Every handle to one instance shares `closed`, so it identifies the instance where the + // handle itself does not. if (this.#instance !== undefined && this.#instance !== active.closed) { this.#ring.peek()?.reset(); this.#decodeBuffered.set([]); + this.sync.reset(); } this.#instance = active.closed; diff --git a/js/watch/src/video/decoder.test.ts b/js/watch/src/video/decoder.test.ts index b1565a597d..3f971193a4 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; } @@ -213,12 +213,12 @@ async function guardedPlayback(kind: "legacy" | "cmaf", reads: Read[]) { const video = config({ codedWidth: 16, codedHeight: 16, description: "01640028ffe100046764002801000268ee" }); if (kind === "cmaf") video.container = { kind, init: Buffer.from(Container.Cmaf.createVideoInitSegment(video)).toString("base64") }; - const producer = new Moq.Broadcast.Producer(); - const consumer = producer.consume(); + let producer = new Moq.Broadcast.Producer(); + const active = new Signal(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" }); @@ -229,11 +229,19 @@ async function guardedPlayback(kind: "legacy" | "cmaf", reads: Read[]) { decoder, enabled, submitted, + sync, + // A restarted publisher takes the name: another instance of the same broadcast. + republish() { + active.peek().close(); + producer.close(); + producer = new Moq.Broadcast.Producer(); + active.set(producer.consume()); + }, close() { decoder.close(); source.close(); sync.close(); - consumer.close(); + active.peek().close(); producer.close(); next.mockRestore(); for (const [name, original] of [ @@ -265,6 +273,25 @@ it("promoting from no active track holds its picture and timestamp until a frame } }); +it("switches to a republished broadcast at once, its timeline starting over", async () => { + const reads = [sample(10, 10_000_000)]; + const playback = await guardedPlayback("legacy", reads); + try { + expect(playback.decoder.out.timestamp.peek()).toBe(Time.Milli(10_000)); + const reset = spyOn(playback.sync, "reset"); + + reads.push(sample(0, 1000)); + playback.republish(); + await microtasks(); + + // Waiting to catch up to the old picture would hold it until the new timeline passed 10s. + expect(playback.decoder.out.timestamp.peek()).toBe(Time.Milli(1)); + expect(reset).toHaveBeenCalledTimes(1); + } finally { + playback.close(); + } +}); + // A late marker from older history raises a discontinuity without making older groups newer. function marker(group: number, end: number, discontinuity: number): Read { return { group, discontinuity, continuous: false, frame: undefined, end: Time.Micro(end) }; diff --git a/js/watch/src/video/decoder.ts b/js/watch/src/video/decoder.ts index 56134653c2..70dc6a64df 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(); @@ -185,6 +189,15 @@ export class Decoder { return; } + // A republish at the same name is another instance, whose timeline starts over. Waiting for + // it to catch up to the old picture would freeze video until it passed the old timestamps, + // so switch at once and re-anchor the clock. A rendition switch within one instance, or a + // switch to another name, still waits. + const closed = effect.get(broadcast.out.active)?.closed; + const republished = this.#instance?.broadcast === broadcast && this.#instance.closed !== closed; + this.#instance = { broadcast, closed }; + if (republished) this.sync.reset(); + // Start a new pending effect. let pending: DecoderTrack | undefined = new DecoderTrack({ sync: this.sync, @@ -202,7 +215,7 @@ export class Decoder { if (!pending) return; const current = effect.get(this.#active); - if (current) { + if (current && !republished) { // A zero timestamp is a rendered frame, not a missing one. const pendingTimestamp = effect.get(pending.timestamp); if (pendingTimestamp === undefined) return; 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/Cargo.toml b/rs/moq-cli/Cargo.toml index 5e667ff564..c0fbe6a38e 100644 --- a/rs/moq-cli/Cargo.toml +++ b/rs/moq-cli/Cargo.toml @@ -57,7 +57,7 @@ transcode = ["dep:moq-transcode", "dep:moq-video"] # Native playback (`moq ... play`): decode into a wgpu window and a cpal # speaker. Off by default because it pulls both graphics and audio device stacks # into a binary that may only be routing containers or serving gateways. -play = ["dep:moq-video", "moq-video/render", "dep:moq-audio", "moq-audio/playback", "dep:pollster", "dep:winit"] +play = ["dep:moq-video", "moq-video/render", "dep:moq-audio", "dep:moq-json", "moq-audio/playback", "dep:pollster", "dep:winit"] # NVIDIA hardware codecs (NVENC + NVDEC) on Linux, on by default. Just turns on # the matching moq-video feature, so it only bites when something else pulls in # moq-video: `capture`, `transcode`, or `play`. For a build with no CUDA @@ -107,6 +107,8 @@ moq-audio = { workspace = true, optional = true, features = ["aac"] } moq-auth = { workspace = true, features = ["serve"] } # `server` enables the HTTP egress server for `moq export hls`; the importer is always available. moq-hls = { workspace = true, features = ["server"] } +# Sees through its transparent transport errors when `play` follows a republish. +moq-json = { workspace = true, optional = true } # Names the MSF catalog track `export archive` records. moq-msf = { workspace = true } moq-mux = { workspace = true } diff --git a/rs/moq-cli/src/archive.rs b/rs/moq-cli/src/archive.rs index 6aa0653993..70dba1318f 100644 --- a/rs/moq-cli/src/archive.rs +++ b/rs/moq-cli/src/archive.rs @@ -118,9 +118,7 @@ pub async fn import(origin: moq_net::origin::Producer, name: String, args: Impor .await .with_context(|| format!("no readable recording at {}", args.store))?; let serve = reader.serve(); - broadcast - .announce(moq_tokio::moq_net::origin::Route::default().with_epoch(moq_tokio::moq_net::Epoch::mint())) - .context("failed to announce broadcast")?; + crate::moq::announce(&broadcast).context("failed to announce broadcast")?; tracing::info!(%name, store = %args.store, "replaying"); notify_ready(); diff --git a/rs/moq-cli/src/hls.rs b/rs/moq-cli/src/hls.rs index 9d754ef295..4fbfc511fe 100644 --- a/rs/moq-cli/src/hls.rs +++ b/rs/moq-cli/src/hls.rs @@ -59,9 +59,7 @@ pub async fn import(target: ImportTarget, playlist: String) -> anyhow::Result<() .with_max_age(max_age) .with_bandwidth(bandwidth); let catalog = moq_mux::catalog::Producer::new(&mut producer, config)?; - producer - .announce(moq_tokio::moq_net::origin::Route::default().with_epoch(moq_tokio::moq_net::Epoch::mint())) - .context("failed to announce broadcast")?; + crate::moq::announce(&producer).context("failed to announce broadcast")?; let playlist = playlist_url(&playlist)?; let mut importer = moq_hls::import::Import::new(producer, catalog, moq_hls::import::Config::new(playlist))?; diff --git a/rs/moq-cli/src/moq.rs b/rs/moq-cli/src/moq.rs index 116c46258e..f128d7c9a6 100644 --- a/rs/moq-cli/src/moq.rs +++ b/rs/moq-cli/src/moq.rs @@ -2,7 +2,8 @@ //! //! The dial and accept loops live in `moq-tokio` (`Client::publish`/`consume` //! and `Server::serve_publish`/`serve_consume`); this module carries the systemd -//! readiness notification used by every endpoint plus the MoQ side of an import. +//! readiness notification used by every endpoint, the MoQ side of an import, and +//! how a publisher announces itself. use hang::moq_net; @@ -35,3 +36,12 @@ pub fn notify_ready() { #[cfg(unix)] let _ = sd_notify::notify(&[sd_notify::NotifyState::Ready]); } + +/// Advertise `broadcast` as a fresh publisher instance, so a restart replaces the +/// previous run's broadcast rather than resuming into it, and log the epoch. +pub fn announce(broadcast: &moq_net::broadcast::Producer) -> Result<(), moq_net::Error> { + let epoch = moq_net::Epoch::mint(); + broadcast.announce(moq_net::origin::Route::default().with_epoch(epoch.clone()))?; + tracing::info!(broadcast = %broadcast.info().path, %epoch, "announced"); + Ok(()) +} diff --git a/rs/moq-cli/src/play/media.rs b/rs/moq-cli/src/play/media.rs index 31a3c943bc..60100176ea 100644 --- a/rs/moq-cli/src/play/media.rs +++ b/rs/moq-cli/src/play/media.rs @@ -63,13 +63,38 @@ impl Media { output.send(event); } + /// Play whatever the path resolves to, following it onto a new broadcast when + /// a restarted publisher replaces the one playing. async fn play(self) -> anyhow::Result<()> { - let source = subscribe(self.origin.clone(), &self.broadcast).await?; - let broadcast = source - .broadcast() - .await - .context("failed to subscribe to the broadcast")?; - let catalog = catalog::Consumer::<()>::new(&broadcast, self.args.catalog_format(&self.broadcast)) + loop { + let source = subscribe(self.origin.clone(), &self.broadcast).await?; + let broadcast = source + .broadcast() + .await + .context("failed to subscribe to the broadcast")?; + let err = match self.play_broadcast(&source, &broadcast).await { + Err(err) if unroutable(&err) => err, + result => return result, + }; + + // A newer publisher instance won the path, or the route went away. The + // handle stays routed until it closes, so wait for that before asking + // again, then start over: the next broadcast's timeline is its own. + tracing::info!(%err, "broadcast went away, waiting for it to return"); + broadcast.closed().await; + *self.presentation.lock().unwrap() = Presentation::new(self.args.video_delay()); + self.video.lock().unwrap().clear(); + self.drained.notify_one(); + } + } + + /// Play one broadcast until its catalog and every track it started end. + async fn play_broadcast( + &self, + source: &moq_mux::Source, + broadcast: &moq_net::broadcast::Consumer, + ) -> anyhow::Result<()> { + let catalog = catalog::Consumer::<()>::new(broadcast, self.args.catalog_format(&self.broadcast)) .await .context("failed to subscribe to the catalog")?; let mut catalogs = catalog.select(self.args.select.selection(None)); @@ -277,6 +302,30 @@ impl Media { } } +/// Whether playback ended because a newer publisher instance won the path, +/// which ends the old one's subscriptions with `Unroutable`: the signal to ask +/// for the path again, whichever layer it surfaced through. +fn unroutable(err: &anyhow::Error) -> bool { + err.chain().any(|cause| { + // These wrappers are transparent, which hides the transport error from the chain. + let net = cause + .downcast_ref::() + .or_else(|| match cause.downcast_ref::()? { + moq_json::Error::Net(err) => Some(err), + _ => None, + }) + .or_else(|| match cause.downcast_ref::()? { + moq_audio::Error::Net(err) => Some(err), + _ => None, + }) + .or_else(|| match cause.downcast_ref::()? { + moq_video::Error::Net(err) => Some(err), + _ => None, + }); + matches!(net, Some(moq_net::Error::Unroutable)) + }) +} + struct AudioPlayback { changed: Arc, speaker: O::Speaker, @@ -555,6 +604,56 @@ 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)); } + /// A restarted publisher announces a fresh epoch while its old route still + /// stands. The newest epoch wins the path and ends the old broadcast, so + /// playback follows the new one, timeline starting over, instead of failing on + /// the one it replaced. + #[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 publish = || { + 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(moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint())) + .unwrap(); + (broadcast, catalog, audio) + }; + + let (_old_broadcast, _old_catalog, mut old) = publish(); + let recorder = Recorder::default(); + let player = tokio::spawn(media(&origin, Duration::from_millis(50), recorder.clone()).run()); + for index in 0..10 { + old.write(packet(index, OLD)).unwrap(); + tokio::time::sleep(PACKET_DURATION).await; + } + + let (_new_broadcast, mut new_catalog, mut new) = publish(); + for index in 0..10 { + new.write(packet(index, NEW)).unwrap(); + tokio::time::sleep(PACKET_DURATION).await; + } + new.finish().unwrap(); + drop(new); + 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"), + } + let played = recorder.played(); + assert!(played.iter().any(|p| p.sample == OLD), "the old broadcast never played"); + let new: Duration = played.iter().filter(|p| p.sample == NEW).map(|p| p.to - p.from).sum(); + assert_eq!(new, PACKET_DURATION * 10, "the new broadcast did not play in full"); + } + #[tokio::test] async fn a_finite_audio_track_plays_its_final_samples() { tokio::time::pause(); diff --git a/rs/moq-cli/src/play/source.rs b/rs/moq-cli/src/play/source.rs index 5696665246..4f1f4d0fe3 100644 --- a/rs/moq-cli/src/play/source.rs +++ b/rs/moq-cli/src/play/source.rs @@ -11,10 +11,15 @@ use hang::moq_net; /// 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 + let route = origin .routed(broadcast) .await .with_context(|| format!("origin closed before broadcast `{broadcast}` was announced"))?; + tracing::info!( + broadcast, + epoch = route.epoch.as_ref().map(tracing::field::display), + "announced" + ); Ok(moq_mux::Source::new(origin, broadcast)) } diff --git a/rs/moq-cli/src/publish.rs b/rs/moq-cli/src/publish.rs index c18499aaff..c92b742b1e 100644 --- a/rs/moq-cli/src/publish.rs +++ b/rs/moq-cli/src/publish.rs @@ -399,9 +399,7 @@ impl Publish { let Some(broadcast) = &self.broadcast else { return Ok(()); }; - broadcast - .announce(moq_tokio::moq_net::origin::Route::default().with_epoch(moq_tokio::moq_net::Epoch::mint())) - .context("failed to announce broadcast") + crate::moq::announce(broadcast).context("failed to announce broadcast") } /// Drive the source until stdin EOF (or the capture devices stop). diff --git a/rs/moq-cli/src/rtc.rs b/rs/moq-cli/src/rtc.rs index d9e55666e0..37c08f85f7 100644 --- a/rs/moq-cli/src/rtc.rs +++ b/rs/moq-cli/src/rtc.rs @@ -96,9 +96,7 @@ pub async fn connect_import(target: ImportTarget, url: Url) -> anyhow::Result<() .context("failed to create broadcast")?; // The WHEP pull fills the tracks as they arrive; announce up front so viewers // can discover the broadcast while it connects. - producer - .announce(moq_tokio::moq_net::origin::Route::default().with_epoch(moq_tokio::moq_net::Epoch::mint())) - .context("failed to announce broadcast")?; + crate::moq::announce(&producer).context("failed to announce broadcast")?; tracing::info!(url = %RedactedUrl::new(&url), %name, "WHEP client pulling"); notify_ready(); diff --git a/rs/moq-cli/src/transcode.rs b/rs/moq-cli/src/transcode.rs index dc25e4cabe..06f593290e 100644 --- a/rs/moq-cli/src/transcode.rs +++ b/rs/moq-cli/src/transcode.rs @@ -165,9 +165,7 @@ pub async fn run(moq: MoqSide, args: Args, net: Net) -> anyhow::Result<()> { let output = publish .create_broadcast(&output_path) .context("failed to create the derivative broadcast")?; - output - .announce(moq_tokio::moq_net::origin::Route::default().with_epoch(moq_tokio::moq_net::Epoch::mint())) - .context("failed to announce the derivative broadcast")?; + crate::moq::announce(&output).context("failed to announce the derivative broadcast")?; tracing::info!(source = %source_path, output = %output_path, "transcoding"); tokio::select! { From 922b72c0a3a5f345be592f2465807398729075d4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:01:11 -0700 Subject: [PATCH 3/5] quest: finish broadcast-epoch/apps Co-Authored-By: Claude Opus 5.5 --- quest/m0/broadcast-epoch/README.md | 5 ++--- quest/m0/broadcast-epoch/apps.md | 21 --------------------- 2 files changed, 2 insertions(+), 24 deletions(-) delete mode 100644 quest/m0/broadcast-epoch/apps.md diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index e8d4b4cd75..eef5ed1690 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -42,8 +42,8 @@ Decided: - 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, - moq-boy, and moq-room through [Apps](/quest/m0/broadcast-epoch/apps.md), the - ingest gateways, moqsink, and the bindings below. moq-stats mints one per group announcement + moq-boy, and moq-room (done), the ingest gateways, moqsink, and the + bindings below. moq-stats mints one per group announcement through [Stats epochs](/quest/m0/broadcast-epoch/stats-epoch.md), which also gates the release (decided 2026-10-04). - The m1 quests gating this line moved under it in the 2026-10-05 audit, and @@ -68,7 +68,6 @@ timeout, and killing the newest epoch falls back to a still-live older one. ## Required -- [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 - [Gateways](/quest/m0/broadcast-epoch/gateways.md) - RTMP, SRT, and WHIP ingest mint an epoch per incoming connection, so an encoder reconnect is a clean takeover - [TS restart](/quest/m0/broadcast-epoch/ts-restart.md) - a signalled backward TS discontinuity finishes the broadcast and continues the same input under a fresh epoch - [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 3c758d3b03..0000000000 --- a/quest/m0/broadcast-epoch/apps.md +++ /dev/null @@ -1,21 +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`, -moq-boy, and moq-room 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. Test a republish mid-playback in the browser and native players. -- Update `doc/bin/cli.md` and every example invocation that shows a published - path. - From 0093590404d98d5d8989226ed5caf69310e34ea9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 09:11:36 -0700 Subject: [PATCH 4/5] feat(apps): follow the broadcast's announcement as its online signal moq play starts on an announcement, starts over on a new one (a new epoch, or any change to a route without one), plays out what it holds when the announcement ends, and waits for it to return instead of exiting. @moq/watch re-requests its broadcast when a route without an epoch changes. Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 23 ++- doc/lib/js/watch.md | 13 +- js/watch/src/broadcast.test.ts | 69 ++++++++ js/watch/src/broadcast.ts | 43 ++++- rs/moq-cli/Cargo.toml | 4 +- rs/moq-cli/src/play/media.rs | 309 ++++++++++++++++++++++----------- rs/moq-cli/src/play/source.rs | 66 ++----- rs/moq-cli/src/play/window.rs | 3 +- 8 files changed, 353 insertions(+), 177 deletions(-) diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 52e382c8f6..726040a625 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -227,11 +227,19 @@ the newest cached group, including when a rendition is reopened, so playback does not replay the retained backlog. A publisher that retires the rendition being played ends that track and the role picks a replacement. A retired audio rendition plays out what the speaker holds while its replacement fills, so the -switch does not cost a delay of silence. A restarted publisher's broadcast -replaces the one playing, and playback starts over on it with a fresh catalog -and decoders; the log names each broadcast's epoch. Playback is -behind the `play` feature, since it pulls in windowing and audio-device -dependencies: +switch does not cost a delay of silence. + +The broadcast's announcement is its online signal. Playback starts when the +name is announced and starts over, with a fresh catalog and decoders, on each +new announcement: a restarted publisher's new +[epoch](/concept/moq-lite#publisher-epochs), or over moq-lite 06, which carries +no epoch, any change to the route. When the announcement ends, playback plays +out what it holds and waits for the broadcast to come back, for as long as the +window stays open. The log names each announced broadcast's epoch. A catalog +with nothing this build can play still fails at once. + +Playback is behind the `play` feature, since it pulls in windowing and +audio-device dependencies: ```bash cargo install moq-cli --no-default-features --features "iroh,noq,websocket,play" @@ -398,8 +406,9 @@ produce identical tracks with aligned groups. Each run announces a fresh [epoch](/concept/moq-lite#publisher-epochs), logged at startup. Over moq-lite 07 (opt-in while its wire is in progress) the epoch reaches the relay, so a restarted publisher replaces the old broadcast and viewers switch to it at -once. Older versions carry no epoch, so viewers stay on the old route until it -goes. +once. Older versions carry no epoch, so a restart that overlaps the old route +reaches viewers only once that route goes: a clean stop retracts it at once, +while a crashed publisher's route lasts until its session times out. ## Cluster diff --git a/doc/lib/js/watch.md b/doc/lib/js/watch.md index ff6acd0a98..53e4f837cc 100644 --- a/doc/lib/js/watch.md +++ b/doc/lib/js/watch.md @@ -44,10 +44,14 @@ Video holds its last picture while paused, out of view, or waiting for a resumed rendition's first frame. Its reported timestamp stays with that picture. Going offline or closing the player clears it. -A restarted publisher announces a new [epoch](/concept/moq-lite#publisher-epochs) +With `announced` on, the broadcast's announcement is its online signal. The +player starts when the name is announced, goes `offline` when the announcement +ends, and keeps waiting for it to come back for as long as it is mounted. A +restarted publisher announces a new [epoch](/concept/moq-lite#publisher-epochs) at the same name, and the player follows it: a fresh catalog, and video, audio, and the clock start over on its timeline instead of waiting for it to catch up -to the old one. +to the old one. moq-lite 06 carries no epoch, so any change to the route +covering the name may be a restart, and the player starts over on it too. The overlay adds play/pause, volume, fullscreen, a quality selector, a buffering indicator, an unsupported-codec warning, and a stats panel. @@ -57,8 +61,9 @@ buffering indicator, an unsupported-codec warning, and a stats panel. `error` when the origin refuses the broadcast, such as a `dynamic()` handler rejecting the request, and `el.broadcast.out.error` then holds the refusal's `Error`; the overlay shows its message in place of the offline notice. A -refusal is final: only a new `name`, a new origin, a changed `announced`, or -re-enabling asks again, which clears both back to `offline`. +refusal is final: only a new `name`, a new origin, a changed `announced`, +re-enabling, or a change to a route without an epoch asks again, which clears +both back to `offline`. ## Binding from a framework diff --git a/js/watch/src/broadcast.test.ts b/js/watch/src/broadcast.test.ts index 6762fc8c28..22f6cda149 100644 --- a/js/watch/src/broadcast.test.ts +++ b/js/watch/src/broadcast.test.ts @@ -501,3 +501,72 @@ describe("manual catalog", () => { } }); }); + +// A broadcast's announcement is its online/offline signal. moq-lite 06 carries no epoch, so a +// restarted publisher shows up either as a retraction and a fresh announcement or, when it +// takes over a route still standing, as that route changing in place. +describe("online and offline", () => { + it("follows a broadcast that ends and is announced again", async () => { + const owner = new Origin.Producer(); + const name = Path.from("live.hang"); + const source = new Broadcast({ origin: owner, name, enabled: true, announced: true, catalogFormat: "manual" }); + try { + const first = publish(owner, name); + await settle(); + const old = source.out.active.peek(); + expect(old).toBeDefined(); + + first.close(); + await settle(); + expect(source.out.active.peek()).toBeUndefined(); + + const second = publish(owner, name); + await settle(); + const active = source.out.active.peek(); + expect(active).toBeDefined(); + expect(active).not.toBe(old); + expect(active?.closed.peek()).toBeUndefined(); + second.close(); + } finally { + source.close(); + owner.close(); + await settle(); + } + }); + + it("starts over when a route without an epoch changes in place", async () => { + const owner = new Origin.Producer(); + const name = Path.from("live.hang"); + const route = owner.dynamic(name); + const requests = route.requested(); + const source = new Broadcast({ origin: owner, name, enabled: true, announced: true, catalogFormat: "manual" }); + try { + const first = await requests.next(); + first.value?.accept(new Moq.Broadcast.Producer()); + await settle(); + const old = source.out.active.peek(); + expect(old).toBeDefined(); + + // Nothing says the changed route serves the same content, so the path is asked for + // again and everything reading the handle starts over on a fresh one. + route.update({ cost: 4n }); + await settle(); + const active = source.out.active.peek(); + expect(active).toBeDefined(); + expect(active).not.toBe(old); + + // An epoch says the content is the same, so a re-price keeps the handle. + route.update({ epoch: Moq.Epoch.mint(), cost: 4n }); + await settle(); + const minted = source.out.active.peek(); + route.update({ ...route.route, cost: 2n }); + await settle(); + expect(source.out.active.peek()).toBe(minted); + } finally { + source.close(); + route.close(); + owner.close(); + await settle(); + } + }); +}); diff --git a/js/watch/src/broadcast.ts b/js/watch/src/broadcast.ts index 157138e9e2..64cfbfeba2 100644 --- a/js/watch/src/broadcast.ts +++ b/js/watch/src/broadcast.ts @@ -85,8 +85,9 @@ type BroadcastOutput = { status: Signal; active: Signal; - // Why the origin refused the broadcast, while `status` is "error". A refusal is final: - // only a fresh request (a new `name`, `origin`, or `announced`, or re-enabling) clears it and asks again. + // Why the origin refused the broadcast, while `status` is "error". A refusal is final: only a + // fresh request (a new `name`, `origin`, or `announced`, re-enabling, or a route without an + // epoch changing) clears it and asks again. error: Signal; // The effective catalog: the fetched one, or a copy of input.catalog in manual mode, minus @@ -230,16 +231,40 @@ export class Broadcast { if (!origin) return; const name = effect.get(this.in.name); - const request = origin.request(name, { announced: effect.get(this.in.announced) }); - effect.cleanup(() => request.close()); + const announced = effect.get(this.in.announced); + + // An epoch names the publisher instance, and a new one already swaps the handle. A route + // without one (moq-lite 06) cannot say whether a change in place is a restarted publisher, + // so ask for the path again on any change, which starts everything downstream over. + const routes = origin.broadcasts(undefined, { hidden: true }); + const instance = announced + ? effect.computed((nested) => { + let route: Moq.Origin.Route | undefined; + let length = -1; + for (const [prefix, candidate] of nested.get(routes)) { + if (prefix.length > length && Path.hasPrefix(prefix, name)) { + route = candidate; + length = prefix.length; + } + } + return route?.epoch ?? route; + }) + : undefined; effect.run((nested) => { - nested.set(this.#out.active, nested.get(request.active), undefined); - }); + if (instance) nested.get(instance); - effect.run((nested) => { - const closed = nested.get(request.closed); - if (closed) nested.set(this.#out.error, closed, undefined); + const request = origin.request(name, { announced }); + nested.cleanup(() => request.close()); + + nested.run((inner) => { + inner.set(this.#out.active, inner.get(request.active), undefined); + }); + + nested.run((inner) => { + const closed = inner.get(request.closed); + if (closed) inner.set(this.#out.error, closed, undefined); + }); }); } diff --git a/rs/moq-cli/Cargo.toml b/rs/moq-cli/Cargo.toml index c0fbe6a38e..5e667ff564 100644 --- a/rs/moq-cli/Cargo.toml +++ b/rs/moq-cli/Cargo.toml @@ -57,7 +57,7 @@ transcode = ["dep:moq-transcode", "dep:moq-video"] # Native playback (`moq ... play`): decode into a wgpu window and a cpal # speaker. Off by default because it pulls both graphics and audio device stacks # into a binary that may only be routing containers or serving gateways. -play = ["dep:moq-video", "moq-video/render", "dep:moq-audio", "dep:moq-json", "moq-audio/playback", "dep:pollster", "dep:winit"] +play = ["dep:moq-video", "moq-video/render", "dep:moq-audio", "moq-audio/playback", "dep:pollster", "dep:winit"] # NVIDIA hardware codecs (NVENC + NVDEC) on Linux, on by default. Just turns on # the matching moq-video feature, so it only bites when something else pulls in # moq-video: `capture`, `transcode`, or `play`. For a build with no CUDA @@ -107,8 +107,6 @@ moq-audio = { workspace = true, optional = true, features = ["aac"] } moq-auth = { workspace = true, features = ["serve"] } # `server` enables the HTTP egress server for `moq export hls`; the importer is always available. moq-hls = { workspace = true, features = ["server"] } -# Sees through its transparent transport errors when `play` follows a republish. -moq-json = { workspace = true, optional = true } # Names the MSF catalog track `export archive` records. moq-msf = { workspace = true } moq-mux = { workspace = true } diff --git a/rs/moq-cli/src/play/media.rs b/rs/moq-cli/src/play/media.rs index 60100176ea..c4924c5623 100644 --- a/rs/moq-cli/src/play/media.rs +++ b/rs/moq-cli/src/play/media.rs @@ -7,6 +7,7 @@ use std::time::Duration; use anyhow::Context; use hang::moq_net; +use moq_net::announce::Event as AnnounceEvent; use moq_mux::catalog::{self, Stream}; // tokio's clock, which is the wall clock unless a test pauses it to drive the // playout clock itself. @@ -15,7 +16,7 @@ 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::source::announced; use super::timeline::{AudioTimeline, Presentation, fit, timestamp}; use super::video::Video; use super::window::Event; @@ -63,38 +64,74 @@ impl Media { output.send(event); } - /// Play whatever the path resolves to, following it onto a new broadcast when - /// a restarted publisher replaces the one playing. + /// Follow the path's announcements for as long as the origin lasts. An + /// announcement is the broadcast coming online, so each new one starts + /// playback over on the broadcast it names. A retraction is the broadcast going + /// offline: whatever is playing plays out, and the player waits for it to come + /// back. async fn play(self) -> anyhow::Result<()> { + let path = moq_net::Path::new(&self.broadcast); + let mut announced = announced(&self.origin, &self.broadcast)?; + let mut playing = None; + loop { - let source = subscribe(self.origin.clone(), &self.broadcast).await?; - let broadcast = source - .broadcast() - .await - .context("failed to subscribe to the broadcast")?; - let err = match self.play_broadcast(&source, &broadcast).await { - Err(err) if unroutable(&err) => err, - result => return result, - }; + tokio::select! { + event = announced.next() => { + let Some(event) = event else { + // The origin is gone, and nothing can announce the path again. + return Ok(()); + }; + let (announce, online) = match event { + AnnounceEvent::Start(announce) => (announce, true), + // An update to a route carrying an epoch is the same publisher + // instance, re-priced or re-routed, and the front follows it on its + // own. Without an epoch nothing says the content is the same, so + // any change may be another publisher. + AnnounceEvent::Update(announce) if announce.route.epoch.is_none() => (announce, true), + AnnounceEvent::Update(_) => continue, + AnnounceEvent::End(announce) => (announce, false), + }; + if !path.has_prefix(&announce.prefix) { + continue; + } + if !online { + tracing::info!(broadcast = %self.broadcast, "offline, waiting for it to return"); + continue; + } - // A newer publisher instance won the path, or the route went away. The - // handle stays routed until it closes, so wait for that before asking - // again, then start over: the next broadcast's timeline is its own. - tracing::info!(%err, "broadcast went away, waiting for it to return"); - broadcast.closed().await; - *self.presentation.lock().unwrap() = Presentation::new(self.args.video_delay()); - self.video.lock().unwrap().clear(); - self.drained.notify_one(); + tracing::info!( + broadcast = %self.broadcast, + epoch = announce.route.epoch.as_ref().map(tracing::field::display), + "announced" + ); + // The next broadcast's timeline is its own. + drop(playing.take()); + *self.presentation.lock().unwrap() = Presentation::new(self.args.video_delay()); + self.video.lock().unwrap().clear(); + self.drained.notify_one(); + playing = Some(Box::pin(self.play_broadcast())); + } + 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 one broadcast until its catalog and every track it started end. - async fn play_broadcast( - &self, - source: &moq_mux::Source, - broadcast: &moq_net::broadcast::Consumer, - ) -> anyhow::Result<()> { - let catalog = catalog::Consumer::<()>::new(broadcast, self.args.catalog_format(&self.broadcast)) + /// Play the broadcast at the path until its catalog and every track it + /// started end. + async fn play_broadcast(&self) -> anyhow::Result<()> { + let source = moq_mux::Source::new(self.origin.clone(), &self.broadcast); + let broadcast = source + .broadcast() + .await + .context("failed to subscribe to the broadcast")?; + let catalog = catalog::Consumer::<()>::new(&broadcast, self.args.catalog_format(&self.broadcast)) .await .context("failed to subscribe to the catalog")?; let mut catalogs = catalog.select(self.args.select.selection(None)); @@ -151,7 +188,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; } } @@ -293,39 +332,26 @@ 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()); + } } } } -/// Whether playback ended because a newer publisher instance won the path, -/// which ends the old one's subscriptions with `Unroutable`: the signal to ask -/// for the path again, whichever layer it surfaced through. -fn unroutable(err: &anyhow::Error) -> bool { - err.chain().any(|cause| { - // These wrappers are transparent, which hides the transport error from the chain. - let net = cause - .downcast_ref::() - .or_else(|| match cause.downcast_ref::()? { - moq_json::Error::Net(err) => Some(err), - _ => None, - }) - .or_else(|| match cause.downcast_ref::()? { - moq_audio::Error::Net(err) => Some(err), - _ => None, - }) - .or_else(|| match cause.downcast_ref::()? { - moq_video::Error::Net(err) => Some(err), - _ => None, - }); - matches!(net, Some(moq_net::Error::Unroutable)) - }) +/// A broadcast this build cannot play. Fatal, unlike a broadcast that ends or +/// fails on the wire, which the player waits out. +#[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,6 +564,57 @@ mod tests { } } + /// 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. + 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. + fn played(recorder: &Recorder, sample: f32) -> Duration { + recorder + .played() + .iter() + .filter(|p| p.sample == sample) + .map(|p| p.to - p.from) + .sum() + } + + /// The player keeps watching 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 watching the path"), + Event::Wake => {} + } + } + assert!(!player.is_finished(), "the player stopped watching the path"); + player.abort(); + } + /// A publisher retires an audio rendition by naming its replacement and then /// finishing the old track. The retired sink still holds a delay of audio, and /// the replacement's sink holds its own before its first sample sounds, so @@ -587,12 +664,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(); @@ -605,9 +677,9 @@ mod tests { assert_eq!(old_duration, Duration::from_secs(1)); } /// A restarted publisher announces a fresh epoch while its old route still - /// stands. The newest epoch wins the path and ends the old broadcast, so - /// playback follows the new one, timeline starting over, instead of failing on - /// the one it replaced. + /// stands. The newest epoch wins the path, which announces it as another + /// broadcast, so playback starts over on the new one instead of failing on the + /// one it replaced. #[tokio::test] async fn a_republish_plays_the_new_broadcast() { tokio::time::pause(); @@ -615,43 +687,81 @@ mod tests { const OLD: f32 = 0.25; const NEW: f32 = 0.5; let origin = moq_tokio::origin::spawn(); - let publish = || { - 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(moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint())) - .unwrap(); - (broadcast, catalog, audio) - }; + let epoch = || moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); - let (_old_broadcast, _old_catalog, mut old) = publish(); + 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()); - for index in 0..10 { - old.write(packet(index, OLD)).unwrap(); - tokio::time::sleep(PACKET_DURATION).await; - } + write(&mut old, 10, OLD).await; - let (_new_broadcast, mut new_catalog, mut new) = publish(); - for index in 0..10 { - new.write(packet(index, NEW)).unwrap(); - tokio::time::sleep(PACKET_DURATION).await; - } - new.finish().unwrap(); - drop(new); - new_catalog.finish().unwrap(); + let (_new_broadcast, _new_catalog, mut new) = publish(&origin, epoch()); + write(&mut new, 10, NEW).await; - player.await.unwrap(); - match recorder.events().pop() { - Some(Event::Ended) => {} - Some(Event::Failed(err)) => panic!("playback failed: {err}"), - _ => panic!("playback never ended"), - } - let played = recorder.played(); - assert!(played.iter().any(|p| p.sample == OLD), "the old broadcast never played"); - let new: Duration = played.iter().filter(|p| p.sample == NEW).map(|p| p.to - p.from).sum(); - assert_eq!(new, PACKET_DURATION * 10, "the new broadcast did not play in full"); + settle(player, &recorder).await; + assert!(played(&recorder, OLD) > Duration::ZERO, "the old broadcast never played"); + assert_eq!(played(&recorder, NEW), PACKET_DURATION * 10, "the new broadcast did not play in full"); + } + + /// Without epochs (moq-lite 06), a publisher that stops cleanly retracts the + /// path, and its restart announces 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 watching 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), PACKET_DURATION * 10, "the old broadcast did not play in full"); + assert_eq!(played(&recorder, NEW), PACKET_DURATION * 10, "the new broadcast did not play in full"); + } + + /// Without epochs (moq-lite 06), a publisher can restart while its old route + /// still stands, say until a crashed session times out. The old route keeps the + /// path until it goes, and then the winning route changes in place. Nothing + /// says the new route serves the same content, so the player starts over on it. + #[tokio::test] + async fn an_overlapping_restart_plays_the_new_broadcast() { + tokio::time::pause(); + + const OLD: f32 = 0.25; + const NEW: f32 = 0.5; + let origin = moq_tokio::origin::spawn(); + let route = |cost: u64| moq_net::origin::Route::default().with_cost(cost); + + let (old_broadcast, old_catalog, mut old) = publish(&origin, route(1)); + 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, route(9)); + tokio::time::sleep(Duration::from_secs(1)).await; + drop((old_broadcast, old_catalog, old)); + // The origin holds a changed route back before announcing it, 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) > Duration::ZERO, "the old broadcast never played"); + assert_eq!(played(&recorder, NEW), PACKET_DURATION * 10, "the new broadcast did not play in full"); } #[tokio::test] @@ -669,12 +779,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/source.rs b/rs/moq-cli/src/play/source.rs index 4f1f4d0fe3..722572d3f8 100644 --- a/rs/moq-cli/src/play/source.rs +++ b/rs/moq-cli/src/play/source.rs @@ -1,56 +1,20 @@ -//! Resolving the broadcast to play. +//! Watching 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 announcements of the routes on `origin` that overlap `broadcast`. /// -/// 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 { - let route = origin - .routed(broadcast) - .await - .with_context(|| format!("origin closed before broadcast `{broadcast}` was announced"))?; - tracing::info!( - broadcast, - epoch = route.epoch.as_ref().map(tracing::field::display), - "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(); - } +/// Only those covering the path matter to the player; the scope just keeps the +/// rest of the origin from waking it. A path at the depth limit cannot be spelled +/// as a subtree, so it watches the whole origin instead. +pub(super) fn announced( + origin: &moq_net::origin::Consumer, + broadcast: &str, +) -> anyhow::Result { + let origin = match moq_net::Pattern::subtree(broadcast) { + Ok(subtree) => origin.scope("", &moq_net::Patterns::from(subtree))?, + Err(moq_net::InvalidPattern::TooManySegments) => origin.clone(), + Err(err) => return Err(err.into()), + }; + Ok(origin.announced()) } 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), } From 3bfa0374246c96f51f9edc39a517fd2bc63fec1d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 09:17:38 -0700 Subject: [PATCH 5/5] quest: record the online/offline decision for broadcast epochs Co-Authored-By: Claude Opus 5.5 --- quest/m0/broadcast-epoch/README.md | 5 ++++ rs/moq-cli/src/play/media.rs | 40 ++++++++++++++++++++++++------ 2 files changed, 37 insertions(+), 8 deletions(-) diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index 1d81a17944..0059499bc7 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -53,6 +53,11 @@ Decided: epoch, so a cluster GOAWAY redial or a standby takeover ends subscriptions instead of resuming them (accepted 2026-10-06 over negotiating the field on lite-06). A release that needs seamless failover promotes lite-07 first. +- Restarts do not promote lite-07 (decided 2026-10-07). A broadcast's + announcement is its online signal: players start when it is announced, start + over on a new announcement (a new epoch, or any change to a route without + one), stop when it ends, and wait for it to return indefinitely. Recovery + never hinges on a subscription error such as `Unroutable`. Decided 2026-10-06: older versions and moq-transport lose cross-route resume, since their routes have no epoch, but their mid-group start handling stays. diff --git a/rs/moq-cli/src/play/media.rs b/rs/moq-cli/src/play/media.rs index c4924c5623..f9365a3c14 100644 --- a/rs/moq-cli/src/play/media.rs +++ b/rs/moq-cli/src/play/media.rs @@ -7,8 +7,8 @@ use std::time::Duration; use anyhow::Context; use hang::moq_net; -use moq_net::announce::Event as AnnounceEvent; 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; @@ -333,7 +333,9 @@ impl Media { // Renditions on offer and not one of them playable, with nothing // already running to fall back on. if tasks.is_empty() && !rejected.is_empty() { - return Err(Unplayable(format!("no playable rendition in the catalog: {}", rejected.join("; "))).into()); + return Err( + Unplayable(format!("no playable rendition in the catalog: {}", rejected.join("; "))).into(), + ); } } } @@ -698,8 +700,15 @@ mod tests { write(&mut new, 10, NEW).await; settle(player, &recorder).await; - assert!(played(&recorder, OLD) > Duration::ZERO, "the old broadcast never played"); - assert_eq!(played(&recorder, NEW), PACKET_DURATION * 10, "the new broadcast did not play in full"); + assert!( + played(&recorder, OLD) > Duration::ZERO, + "the old broadcast never played" + ); + assert_eq!( + played(&recorder, NEW), + PACKET_DURATION * 10, + "the new broadcast did not play in full" + ); } /// Without epochs (moq-lite 06), a publisher that stops cleanly retracts the @@ -729,8 +738,16 @@ mod tests { write(&mut new, 10, NEW).await; settle(player, &recorder).await; - assert_eq!(played(&recorder, OLD), PACKET_DURATION * 10, "the old broadcast did not play in full"); - assert_eq!(played(&recorder, NEW), PACKET_DURATION * 10, "the new broadcast did not play in full"); + assert_eq!( + played(&recorder, OLD), + PACKET_DURATION * 10, + "the old broadcast did not play in full" + ); + assert_eq!( + played(&recorder, NEW), + PACKET_DURATION * 10, + "the new broadcast did not play in full" + ); } /// Without epochs (moq-lite 06), a publisher can restart while its old route @@ -760,8 +777,15 @@ mod tests { write(&mut new, 10, NEW).await; settle(player, &recorder).await; - assert!(played(&recorder, OLD) > Duration::ZERO, "the old broadcast never played"); - assert_eq!(played(&recorder, NEW), PACKET_DURATION * 10, "the new broadcast did not play in full"); + assert!( + played(&recorder, OLD) > Duration::ZERO, + "the old broadcast never played" + ); + assert_eq!( + played(&recorder, NEW), + PACKET_DURATION * 10, + "the new broadcast did not play in full" + ); } #[tokio::test]