diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 2ad9a9022a..726040a625 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -227,9 +227,19 @@ the newest cached group, including when a rendition is reopened, so playback does not replay the retained backlog. A publisher that retires the rendition being played ends that track and the role picks a replacement. A retired audio rendition plays out what the speaker holds while its replacement fills, so the -switch does not cost a delay of silence. Playback is -behind the `play` feature, since it pulls in windowing and audio-device -dependencies: +switch does not cost a delay of silence. + +The broadcast's announcement is its online signal. Playback starts when the +name is announced and starts over, with a fresh catalog and decoders, on each +new announcement: a restarted publisher's new +[epoch](/concept/moq-lite#publisher-epochs), or over moq-lite 06, which carries +no epoch, any change to the route. When the announcement ends, playback plays +out what it holds and waits for the broadcast to come back, for as long as the +window stays open. The log names each announced broadcast's epoch. A catalog +with nothing this build can play still fails at once. + +Playback is behind the `play` feature, since it pulls in windowing and +audio-device dependencies: ```bash cargo install moq-cli --no-default-features --features "iroh,noq,websocket,play" @@ -392,9 +402,13 @@ moq --connect https://relay.example.com/anon \ Two publishers of the same broadcast name are interchangeable sources: relays hold both routes and fail over between them mid-group. They must -produce identical tracks with aligned groups. A restarted encoder is the same -broadcast too, so one whose groups restart from 0 must publish under a new -name, or viewers wait for its sequence to catch up. +produce identical tracks with aligned groups. Each run announces a fresh +[epoch](/concept/moq-lite#publisher-epochs), logged at startup. Over +moq-lite 07 (opt-in while its wire is in progress) the epoch reaches the relay, +so a restarted publisher replaces the old broadcast and viewers switch to it at +once. Older versions carry no epoch, so a restart that overlaps the old route +reaches viewers only once that route goes: a clean stop retracts it at once, +while a crashed publisher's route lasts until its session times out. ## Cluster diff --git a/doc/lib/js/watch.md b/doc/lib/js/watch.md index ded3988fe7..53e4f837cc 100644 --- a/doc/lib/js/watch.md +++ b/doc/lib/js/watch.md @@ -44,6 +44,15 @@ Video holds its last picture while paused, out of view, or waiting for a resumed rendition's first frame. Its reported timestamp stays with that picture. Going offline or closing the player clears it. +With `announced` on, the broadcast's announcement is its online signal. The +player starts when the name is announced, goes `offline` when the announcement +ends, and keeps waiting for it to come back for as long as it is mounted. A +restarted publisher announces a new [epoch](/concept/moq-lite#publisher-epochs) +at the same name, and the player follows it: a fresh catalog, and video, audio, +and the clock start over on its timeline instead of waiting for it to catch up +to the old one. moq-lite 06 carries no epoch, so any change to the route +covering the name may be a restart, and the player starts over on it too. + The overlay adds play/pause, volume, fullscreen, a quality selector, a buffering indicator, an unsupported-codec warning, and a stats panel. `` shows what the browser can play. @@ -52,8 +61,9 @@ buffering indicator, an unsupported-codec warning, and a stats panel. `error` when the origin refuses the broadcast, such as a `dynamic()` handler rejecting the request, and `el.broadcast.out.error` then holds the refusal's `Error`; the overlay shows its message in place of the offline notice. A -refusal is final: only a new `name`, a new origin, a changed `announced`, or -re-enabling asks again, which clears both back to `offline`. +refusal is final: only a new `name`, a new origin, a changed `announced`, +re-enabling, or a change to a route without an epoch asks again, which clears +both back to `offline`. ## Binding from a framework 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 dbf1fc2263..9d1fcc63fc 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/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/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/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index b6bc106787..0059499bc7 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 @@ -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. @@ -68,7 +73,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. - 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/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..f9365a3c14 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,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,8 +64,69 @@ impl Media { output.send(event); } + /// 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 source = subscribe(self.origin.clone(), &self.broadcast).await?; + let path = moq_net::Path::new(&self.broadcast); + let mut announced = announced(&self.origin, &self.broadcast)?; + let mut playing = None; + + loop { + 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; + } + + 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 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 @@ -126,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; } } @@ -268,15 +332,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. 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, @@ -489,6 +566,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 @@ -538,12 +666,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 +678,116 @@ 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, 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(); + + 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) > 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] async fn a_finite_audio_track_plays_its_final_samples() { tokio::time::pause(); @@ -570,12 +803,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 5696665246..722572d3f8 100644 --- a/rs/moq-cli/src/play/source.rs +++ b/rs/moq-cli/src/play/source.rs @@ -1,51 +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 { - 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(); - } +/// 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), } 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! {