From 89c351e7ea584f4e9abdb357f38a6aa0de213e67 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 19:58:52 -0800 Subject: [PATCH 1/5] chore(quest): claim quest/m0/failover.md Quest-Claim: 17a7ad7e1abfe629 From 52929cb3cd865d8e76187fc1ea9e923b79576f70 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 20:30:28 -0800 Subject: [PATCH 2/5] feat(studio): publish to a failover relay under the same epoch --failover publishes the program to a second relay under the one publisher epoch, so a viewer linked to both resumes mid-group when the first fails. Both relays must speak one --connect-version with epochs, or the studio refuses to start. Broadcast sources fail over too, and the Health panel shows each relay's state. Co-Authored-By: Claude Opus 5.5 --- README.md | 31 ++++-- quest/m0/README.md | 1 - quest/m0/failover.md | 18 ---- rs/moq-studio/src/engine.rs | 169 ++++++++++++++++++++++++++----- rs/moq-studio/src/health.rs | 195 ++++++++++++++++++++++++++++-------- rs/moq-studio/src/main.rs | 77 +++++++++++++- rs/moq-studio/src/ui.rs | 47 ++++++--- 7 files changed, 423 insertions(+), 115 deletions(-) delete mode 100644 quest/m0/failover.md diff --git a/README.md b/README.md index 7201d55a..03cb84ce 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,13 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ changes are saved to it a second after you stop editing, with no save step. - Without `--connect`, the studio composes and previews but publishes nothing, and broadcast sources stay offline. +- `--failover ` publishes the same broadcast to a second relay under the + same publisher epoch, so a viewer linked to both resumes mid-group from it + when the first fails. Both relays must speak the one `--connect-version`, + which must carry publisher epochs (`moq-lite-07-wip` today, which a relay + accepts only when configured to), or the studio refuses to start. Its + credential rides its URL's `?jwt=`; `--connect-token` goes to the first + relay only. - `capture` adds camera, display, window, microphone, and system-audio sources. On Linux it links ALSA; add `pipewire` for PipeWire cameras. `openh264` is the software H.264 fallback for a machine without a hardware encoder. @@ -40,7 +47,7 @@ Every panel is a tab you can drag, split, or stack. | Filters | Add, order, and set the filters of the selected source or item, and copy or paste them. | | Audio | Set each audio source's gain and mute, and watch its meter and the program's. | | Output | See the broadcast path, whether the program is encoding, and the encoder's work. | -| Health | See how the stream reaches viewers, track by track, and the cause of each problem. | +| Health | See how the stream reaches viewers, relay by relay and track by track, and the cause of each problem. | A new source, from the Sources or Devices panel, is placed over whatever the Items panel shows. The **top layer** is drawn over every scene, in preview and @@ -62,7 +69,8 @@ comes along, renamed if its name is taken. Copied filters paste onto the selected source or item, after its own. The clipboard holds text in the scene file's format. -A broadcast source subscribes through the same relay connection, plays the +A broadcast source subscribes through the same relay connections, from the +first relay while it has the broadcast and the failover only while it does not, plays the best video and audio its catalog lists, switches when the catalog changes, and comes back on its own when the publisher restarts. Audio from every source is mixed into the program whether or not its scene is live; mute a source to leave it out. @@ -185,16 +193,19 @@ compositor uploads. The Health panel grades the stream every second on the thresholds moq's `hang::health` shares with the rest of the stack, and lists each problem -with its cause in plain words: the relay is unreachable, the uplink is +with its cause in plain words: a relay is unreachable, the uplink is congested or losing packets, the relay is far away, the encoder is behind or stuck, a camera or broadcast source on program stalled, or viewers are late, -stalling, or cannot decode a track. Below that it shows: - -- **Uplink**: the round trip to the relay, the bandwidth the connection - estimates it can carry, the media every track writes, and the share lost and - resent. Loss and headroom are graded only while someone watches, since the - relay pulls nothing otherwise. -- **Watching**: the sessions subscribed to the broadcast, from the relay's +stalling, or cannot decode a track. A relay lost while the other serves is +degraded, since viewers have nothing left to fall back to; with none serving +it is unhealthy. Below that it shows: + +- **Relays**: each relay's state, and the protocol its session negotiated. +- **Uplink**: the round trip to the first connected relay, the bandwidth the + connection estimates it can carry, the media every track writes, and the + share lost and resent. Loss and headroom are graded only while someone + watches, since the relay pulls nothing otherwise. +- **Watching**: the sessions subscribed to the broadcast, from the relays' `.stats` broadcasts. A relay shares those only with a connection allowed to read them, so most show "unknown". In a cluster, a relay forwarding the stream to another counts as one session. diff --git a/quest/m0/README.md b/quest/m0/README.md index 2faecf13..cdad56dc 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -86,7 +86,6 @@ repository. - [Broadcast sources play in sync](/quest/m0/broadcast-sync.md) - subscribed broadcasts and guests play audio and video on the publisher's timestamps, with `moq play`'s delay - [Wayland display and window sources](/quest/m0/wayland-sources.md) - portal-picked sources that reopen without the picker - [Live encode control](/quest/m0/encode-control.md) - simulcast rungs that follow the bandwidth grant, live bitrate and resolution, keyframes on request -- [Relay failover](/quest/m0/failover.md) - the same epoch on two relays, so a relay failure does not drop the stream - [Instant replay and DVR](/quest/m0/replay.md) - replays and clips from held groups, and viewer scrubbing through the archive - [HTML/CSS overlays](/quest/m0/overlays.md) - script-free HTML and CSS graphics through Blitz natively and the DOM in the browser - [Animation and transitions](/quest/m0/animation.md) - move, slide, wipe, and stinger transitions, and keyframed item properties diff --git a/quest/m0/failover.md b/quest/m0/failover.md deleted file mode 100644 index 97f27a84..00000000 --- a/quest/m0/failover.md +++ /dev/null @@ -1,18 +0,0 @@ -# [M] Relay failover - -## Goal - -`moq-studio` publishes the same broadcast epoch to two relays at once, so -when one relay or the path to it fails, viewers resume from the other -mid-group and the stream does not drop. - -## Plan - -Publisher epochs already support this: the same epoch on several relays -resumes mid-group, and a restart is a newer epoch (`doc/concept/moq-lite.md`, -Publisher epochs and Multiple CDNs). The studio dials one relay today. Encode -once and publish to both. The second relay is configured, never inferred. -Both links must negotiate the same protocol and draft, one with publisher -epochs, since a copy whose track info differs ends the track; a relay that -does not match is refused at setup with the reason, rather than failing over -as a restart. Show each relay's state in the Health panel. diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 5d32f466..989957ed 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -201,12 +201,18 @@ pub struct Output { /// Whether the program video is copied and encoded only while someone /// watches, rather than all the time. pub on_demand: bool, - /// The relay connection, which reconnects on its own and dials only while - /// held. The Health panel reads its state. - _connection: moq_tokio::Connection, + /// A connection to each relay, which reconnects on its own and dials only + /// while held. The Health panel reads their states. + _connections: Vec, publication: moq_net::origin::Publication, } +/// A relay the program is published to. +pub struct Relay { + pub client: moq_tokio::Client, + pub url: url::Url, +} + impl Drop for Output { fn drop(&mut self) { self.publication.close(); @@ -237,15 +243,15 @@ const AUDIO_PERIOD: Duration = Duration::from_millis(10); impl Engine { /// Open the `loaded` collection and history on `gpu`, start the program /// clock, and save both to `autosave`'s files as they change. - /// With `connect`, publish the program at `broadcast` on that relay, - /// encoding its video only while someone watches when `on_demand`. + /// Publish the program at `broadcast` on every one of `relays`, preferred + /// first, encoding its video only while someone watches when `on_demand`. /// - /// Call inside the tokio runtime that serves the relay connection. + /// Call inside the tokio runtime that serves the relay connections. pub async fn start( gpu: Gpu, autosave: Autosave, loaded: autosave::Loaded, - connect: Option<(moq_tokio::Client, url::Url)>, + relays: Vec, broadcast: &str, on_demand: bool, ) -> anyhow::Result> { @@ -270,12 +276,11 @@ impl Engine { compositor.cut(); let canvas = collection.canvas; - let (output, origin, publish) = match connect { - Some((client, url)) => { - let (output, publish) = publish(&gpu, client, url, subscribe, broadcast, canvas, on_demand).await?; - (Some(output), None, Some(publish)) - } - None => (None, Some(subscribe), None), + let (output, origin, publish) = if relays.is_empty() { + (None, Some(subscribe), None) + } else { + let (output, publish) = publish(&gpu, relays, subscribe, broadcast, canvas, on_demand).await?; + (Some(output), None, Some(publish)) }; let engine = Arc::new(Self { @@ -537,12 +542,11 @@ struct Publish { health: health::Publisher, } -/// Announce the program broadcast and connect to the relay, which also -/// carries `subscribe`'s broadcast sources. +/// Announce the program broadcast under one epoch and connect to every relay, +/// each also carrying `subscribe`'s broadcast sources. async fn publish( gpu: &Gpu, - client: moq_tokio::Client, - url: url::Url, + relays: Vec, subscribe: moq_net::origin::Producer, path: &str, canvas: moq_scene::Canvas, @@ -550,15 +554,15 @@ async fn publish( ) -> anyhow::Result<(Output, Publish)> { let origin = moq_tokio::origin::spawn(); let path = moq_net::Path::new(path).with_context(|| format!("cannot publish at `{path}`"))?; - tracing::info!(url = %moq_tokio::RedactedUrl::new(&url), broadcast = %path, "connecting"); let reports = subscribe.consume(); - let connection = client.with_publisher(&origin).with_subscriber(subscribe).connect(url); + tracing::info!(broadcast = %path, "publishing"); + let (connections, links) = connect(relays, &origin, &subscribe); let publication = origin.create_broadcast(&path).context("cannot create the broadcast")?; let mut broadcast = publication.broadcast().clone(); - // Each track claims its bitrate on the connection's estimate, which the + // Each track claims its bitrate on the first relay's estimate, which that // connection then probes to fund: the bandwidth the stream is granted. - let catalog = moq_mux::catalog::Config::default().with_bandwidth(connection.allocator()); + let catalog = moq_mux::catalog::Config::default().with_bandwidth(connections[0].allocator()); let catalog = moq_mux::catalog::Producer::new(&mut broadcast, catalog)?; let clock = catalog.clock(); let (echo, viewers) = health::echo(&path)?; @@ -588,6 +592,8 @@ async fn publish( let stats = catalog.clone(); let audio = moq_audio::encode::Producer::new(&mut broadcast, catalog, input, &options)?; + // One epoch on every relay, so a viewer moving between them resumes + // mid-group rather than restarting. let epoch = moq_net::Epoch::mint(); publication .announce(moq_net::origin::Route::default().with_epoch(epoch.clone())) @@ -604,7 +610,7 @@ async fn publish( observer: hang::health::Observer::new(path.clone(), Some(epoch)), catalog: stats, clock: clock.clone(), - monitor: connection.monitor(), + relays: links, feedback, watching, }; @@ -612,7 +618,7 @@ async fn publish( let output = Output { path: path.to_string(), on_demand, - _connection: connection, + _connections: connections, publication, }; Ok(( @@ -628,6 +634,39 @@ async fn publish( )) } +/// Connect to every relay, publishing `origin` to each and feeding `subscribe` +/// the broadcast sources of each, and watch each connection for the Health +/// panel. +fn connect( + relays: Vec, + origin: &moq_net::origin::Producer, + subscribe: &moq_net::origin::Producer, +) -> (Vec, Vec) { + relays + .into_iter() + .enumerate() + .map(|(rank, Relay { client, url })| { + tracing::info!(url = %moq_tokio::RedactedUrl::new(&url), "connecting"); + // Only the scheme, host, and port: a path can carry a credential, and + // the panel may be on a shared screen. + let mut endpoint = format!("{}://{}", url.scheme(), url.host_str().unwrap_or_default()); + if let Some(port) = url.port() { + endpoint.push_str(&format!(":{port}")); + } + // A broadcast source comes through the first relay while it has a + // route, and through the failover only while the first has none. + let preference = match rank { + 0 => moq_net::origin::Preference::default(), + _ => moq_net::origin::Preference::new(-1), + }; + let sources = subscribe.clone().with_preference(preference); + let connection = client.with_publisher(origin).with_subscriber(sources).connect(url); + let monitor = connection.monitor(); + (connection, health::Link { endpoint, monitor }) + }) + .unzip() +} + /// Run a background reader, logging why it stopped. async fn log(what: &'static str, task: impl Future>) { if let Err(err) = task.await { @@ -806,14 +845,17 @@ mod tests { let gpu = Gpu::new().await.expect("a GPU"); let (autosave, _) = Autosave::open(dir.path().join("show.toml")).unwrap(); - let connect = Some((client, fixture.url.clone())); + let relays = vec![Relay { + client, + url: fixture.url.clone(), + }]; let on_demand = false; let loaded = autosave::Loaded { collection, history: History::new(), dropped: 0, }; - let engine = Engine::start(gpu, autosave, loaded, connect, "studio", on_demand) + let engine = Engine::start(gpu, autosave, loaded, relays, "studio", on_demand) .await .expect("the studio starts"); @@ -930,7 +972,8 @@ mod tests { .tracks .iter() .all(|track| track.sent.is_some_and(|sent| sent.as_bps() > 0)); - if report.relay == Some(health::Relay::Connected) && sending { + let connected = matches!(&report.relays[..], [relay] if relay.state == health::State::Connected); + if connected && sending { return; } } @@ -940,4 +983,78 @@ mod tests { .await .expect("health grades the program's tracks"); } + + /// With a failover relay, both relays route the program under its one + /// epoch over the one protocol version, so a viewer moving between them + /// resumes; and a broadcast source only the failover has still reaches + /// the studio. + #[tokio::test(flavor = "multi_thread")] + async fn every_relay_routes_one_epoch() { + let _ = moq_tokio::crypto::install_default(); + let version: moq_net::Version = "moq-lite-07-wip".parse().unwrap(); + let mut relays = Vec::new(); + let mut routes = Vec::new(); + let mut guests = Vec::new(); + for _ in 0..2 { + let fixture = moq_relay::test_relay().await.expect("test relay"); + let ready = fixture.relay.ready(); + routes.push(fixture.relay.cluster().origin.consume()); + tokio::spawn(fixture.relay.run()); + ready.wait().await.expect("relay ready"); + let mut client = moq_tokio::connect::Config::default(); + client.tls.fingerprint = vec![fixture.fingerprint.clone()]; + client.version = vec![version]; + let client = client.init(Default::default()).expect("client"); + guests.push((client.clone(), fixture.url.clone())); + relays.push(Relay { + client, + url: fixture.url, + }); + } + + let origin = moq_tokio::origin::spawn(); + let subscribe = moq_tokio::origin::spawn(); + let (_connections, links) = connect(relays, &origin, &subscribe); + let publication = origin.create_broadcast("studio").unwrap(); + let epoch = moq_net::Epoch::mint(); + publication + .announce(moq_net::origin::Route::default().with_epoch(epoch.clone())) + .unwrap(); + + let guest = moq_tokio::origin::spawn(); + let source = guest.create_broadcast("guest").unwrap(); + source.announce(Default::default()).unwrap(); + let (client, url) = guests.pop().unwrap(); + let _guest = client.with_publisher(&guest).connect(url); + + let routed = async { + for relay in &routes { + let mut announced = relay.announced(); + loop { + match announced.next().await.expect("announcements") { + moq_net::announce::Event::Start(announce) if announce.prefix.as_str() == "studio" => { + assert_eq!(announce.route.epoch, Some(epoch.clone())); + break; + } + _ => continue, + } + } + } + for link in &links { + let mut state = link.monitor.state(); + while state.version != Some(version) { + state = link + .monitor + .state_changed(state.revision) + .await + .expect("the connection"); + } + } + subscribe.consume().follow("guest", None).unwrap().available().await + }; + kio::clock::timeout(TIMEOUT, routed) + .await + .expect("both relays route the program, and the failover the source") + .expect("the source"); + } } diff --git a/rs/moq-studio/src/health.rs b/rs/moq-studio/src/health.rs index 27444322..0a950a16 100644 --- a/rs/moq-studio/src/health.rs +++ b/rs/moq-studio/src/health.rs @@ -3,8 +3,8 @@ //! //! Every [`INTERVAL`] the studio grades its stream with `hang::health`, the //! grader the rest of the stack shares, from three points of view: -//! - its own: what each track of the broadcast wrote, and the relay -//! connection's round trip, estimated capacity, and loss; +//! - its own: what each track of the broadcast wrote, each relay connection's +//! state, and the serving one's round trip, estimated capacity, and loss; //! - its viewers': the feedback each publishes under the broadcast's `echo` //! path, which the catalog solicits (`moq export --echo` answers it); //! - the relay's: how many sessions watch the broadcast, read from the relay's @@ -42,8 +42,9 @@ const MAX_VIEWERS: usize = 64; /// What the Health panel draws, rebuilt every [`INTERVAL`]. #[derive(Debug, Default)] pub struct Report { - /// The relay connection, `None` when the studio publishes nothing. - pub relay: Option, + /// Each relay the broadcast goes to, the preferred first; empty when the + /// studio publishes nothing. + pub relays: Vec, pub uplink: Uplink, pub viewers: Viewers, /// Each track of the broadcast, in catalog order. @@ -52,9 +53,19 @@ pub struct Report { pub problems: Vec, } -/// The relay connection's state. +/// One relay connection. #[derive(Debug, Clone, PartialEq)] -pub enum Relay { +pub struct Relay { + /// Its scheme, host, and port. + pub endpoint: String, + pub state: State, + /// The protocol its session negotiated, while connected. + pub version: Option, +} + +/// A relay connection's state. +#[derive(Debug, Clone, PartialEq)] +pub enum State { /// No session has connected yet. Connecting, Connected, @@ -62,9 +73,12 @@ pub enum Relay { Disconnected(String), } -/// The connection to the relay, graded as the publisher's link. +/// The link to the first connected relay, graded as the publisher's: viewers +/// who rank the relays as the studio does reach the stream through it. #[derive(Debug, Default)] pub struct Uplink { + /// The relay it reaches, `None` while none is connected. + pub relay: Option, /// The worst of the grades below. pub verdict: Verdict, pub rtt: Option, @@ -178,11 +192,19 @@ pub struct Publisher { /// The catalog's clock, which maps a track's timestamps to the wall time /// the program drew them. pub clock: moq_mux::Clock, - pub monitor: moq_tokio::connection::Monitor, + /// Each relay the broadcast goes to, the preferred first. + pub relays: Vec, pub feedback: Feedback, pub watching: Watching, } +/// The connection to one relay. +pub struct Link { + /// Its scheme, host, and port. + pub endpoint: String, + pub monitor: moq_tokio::connection::Monitor, +} + /// Grades the stream every interval. pub struct Sampler { publisher: Option, @@ -195,6 +217,8 @@ struct Previous { at: Instant, local: Local, report: Option>, + /// Which relay's link the report's transport readings came from. + serving: Option, } impl Sampler { @@ -217,16 +241,25 @@ impl Sampler { let mut report = Report::default(); let mut graded = Graded::default(); let mut current = None; + let mut serving = None; if let Some(publisher) = &self.publisher { - report.relay = Some(relay(&publisher.monitor)); - let snapshot = snapshot(publisher); + report.relays = publisher.relays.iter().map(relay).collect(); + serving = report.relays.iter().position(|relay| relay.state == State::Connected); + let link = serving.map(|index| &publisher.relays[index]); + let snapshot = snapshot(publisher, link.map(|link| &link.monitor)); let ours = health::Report::new(publisher.observer.clone(), snapshot); let mut sample = health::Sample::new(&ours, Duration::ZERO); - if let (Some(previous), Some(interval)) = (previous.as_ref().and_then(|p| p.report.as_ref()), interval) { + // Another relay's link counted its transport from another start. + let previous_report = previous + .as_ref() + .filter(|previous| previous.serving == serving) + .and_then(|previous| previous.report.as_ref()); + if let (Some(previous), Some(interval)) = (previous_report, interval) { sample = sample.with_previous(previous, interval); } let graded_ours = health::publisher(&sample, &self.thresholds); report.uplink = uplink(&ours.snapshot, &graded_ours, local.watched); + report.uplink.relay = link.map(|link| link.endpoint.clone()); graded.publisher = Some(graded_ours); graded.viewers = publisher.feedback.grade(now, &self.thresholds); report.viewers.unread = publisher.feedback.unread(); @@ -242,7 +275,7 @@ impl Sampler { encoded: local.encoded.saturating_sub(previous.local.encoded), }); report.problems = problems(&Facts { - relay: report.relay.as_ref(), + relays: &report.relays, local: &local, drops, uplink: &report.uplink, @@ -255,28 +288,37 @@ impl Sampler { at: now, local, report: current, + serving, }); report } } -fn relay(monitor: &moq_tokio::connection::Monitor) -> Relay { - match monitor.state().phase { - moq_tokio::connection::Phase::Connecting => Relay::Connecting, - moq_tokio::connection::Phase::Connected | moq_tokio::connection::Phase::Migrating => Relay::Connected, - moq_tokio::connection::Phase::Disconnected | moq_tokio::connection::Phase::Closed => Relay::Disconnected( - monitor - .failure() +fn relay(link: &Link) -> Relay { + use moq_tokio::connection::Phase; + let observed = link.monitor.state(); + let state = match observed.phase { + Phase::Connecting => State::Connecting, + Phase::Connected | Phase::Migrating => State::Connected, + Phase::Disconnected | Phase::Closed => State::Disconnected( + observed + .failure .map_or_else(|| "the connection closed".to_string(), |err| err.to_string()), ), + }; + Relay { + endpoint: link.endpoint.clone(), + state, + version: observed.version, } } -/// What the studio's tracks have written, and the connection's transport. -fn snapshot(publisher: &Publisher) -> hang::stats::Snapshot { +/// What the studio's tracks have written, and the transport of the link +/// `monitor` watches. +fn snapshot(publisher: &Publisher, monitor: Option<&moq_tokio::connection::Monitor>) -> hang::stats::Snapshot { let mut snapshot = hang::stats::Snapshot::default(); snapshot.renditions = publisher.catalog.stats(); - if let Some(stats) = publisher.monitor.stats() { + if let Some(stats) = monitor.and_then(moq_tokio::connection::Monitor::stats) { snapshot.transport.rtt = stats.rtt; snapshot.transport.rate = stats.estimated_send_rate.map(Rate::as_bps); snapshot.transport.lost_bytes = stats.bytes_lost; @@ -303,6 +345,7 @@ fn uplink(snapshot: &hang::stats::Snapshot, graded: &Health, watched: bool) -> U .sum() }); Uplink { + relay: None, verdict: grades .iter() .map(|(_, verdict)| *verdict) @@ -436,7 +479,7 @@ struct Drops { /// Everything a problem can rest on. struct Facts<'a> { - relay: Option<&'a Relay>, + relays: &'a [Relay], local: &'a Local, drops: Option, uplink: &'a Uplink, @@ -456,13 +499,28 @@ fn problems(facts: &Facts<'_>) -> Vec { }; let local = facts.local; - match facts.relay { - Some(Relay::Disconnected(reason)) => add( - Verdict::Unhealthy, - format!("The relay is unreachable ({reason}), so nothing reaches viewers until it reconnects."), - ), - Some(Relay::Connecting) => add(Verdict::Degraded, "Connecting to the relay.".into()), - Some(Relay::Connected) | None => {} + // While another relay serves, one lost is a lost fallback rather than an + // outage. + let serving = facts.relays.iter().find(|relay| relay.state == State::Connected); + for relay in facts.relays { + let endpoint = &relay.endpoint; + match (&relay.state, serving) { + (State::Disconnected(reason), Some(serving)) => add( + Verdict::Degraded, + format!( + "The relay {endpoint} is unreachable ({reason}), so the stream reaches viewers only through {}.", + serving.endpoint + ), + ), + (State::Disconnected(reason), None) => add( + Verdict::Unhealthy, + format!( + "The relay {endpoint} is unreachable ({reason}), so nothing reaches viewers until a relay reconnects." + ), + ), + (State::Connecting, _) => add(Verdict::Degraded, format!("Connecting to the relay {endpoint}.")), + (State::Connected, _) => {} + } } if let Some(failed) = &local.failed { add(Verdict::Unhealthy, format!("The output stopped: {failed}.")); @@ -502,11 +560,15 @@ fn problems(facts: &Facts<'_>) -> Vec { } let uplink = facts.uplink; + let relay = uplink + .relay + .as_ref() + .map_or_else(String::new, |relay| format!(" {relay}")); if let (Some(capacity), Some(media)) = (uplink.capacity, uplink.media) { add( uplink.grade(Input::Headroom), format!( - "The uplink is congested: the connection carries about {} and the stream needs {}.", + "The uplink to the relay{relay} is congested: the connection carries about {} and the stream needs {}.", rate(capacity), rate(media) ), @@ -516,7 +578,7 @@ fn problems(facts: &Facts<'_>) -> Vec { add( uplink.grade(Input::Loss), format!( - "The uplink is losing packets: {:.1}% of the media sent had to be resent.", + "The uplink to the relay{relay} is losing packets: {:.1}% of the media sent had to be resent.", loss * 100.0 ), ); @@ -525,7 +587,7 @@ fn problems(facts: &Facts<'_>) -> Vec { add( uplink.grade(Input::Rtt), format!( - "The relay is far away: a round trip to it takes {} ms.", + "The relay{relay} is far away: a round trip to it takes {} ms.", rtt.as_millis() ), ); @@ -858,7 +920,7 @@ mod tests { fn facts<'a>(local: &'a Local, uplink: &'a Uplink, viewers: &'a [GradedViewer], tracks: &'a [Track]) -> Facts<'a> { static THRESHOLDS: std::sync::LazyLock = std::sync::LazyLock::new(Thresholds::default); Facts { - relay: Some(&Relay::Connected), + relays: &[], local, drops: None, uplink, @@ -930,11 +992,14 @@ mod tests { watched: true, ..Default::default() }; - let watched = uplink(¤t.snapshot, &graded, true); + let mut watched = uplink(¤t.snapshot, &graded, true); + watched.relay = Some("https://a".into()); assert_eq!(watched.verdict, Verdict::Unhealthy); assert_eq!( causes(problems(&facts(&local, &watched, &[], &[]))), - ["The uplink is congested: the connection carries about 2.5 Mb/s and the stream needs 3.0 Mb/s."] + [ + "The uplink to the relay https://a is congested: the connection carries about 2.5 Mb/s and the stream needs 3.0 Mb/s." + ] ); local.watched = false; let idle = uplink(¤t.snapshot, &graded, false); @@ -1039,19 +1104,66 @@ mod tests { grades: vec![(Input::Rtt, Verdict::Degraded)], ..Default::default() }; - let relay = Relay::Disconnected("timed out".into()); + let relays = [relay("https://a", State::Disconnected("timed out".into()))]; let mut facts = facts(&local, &uplink, &[], &[]); - facts.relay = Some(&relay); + facts.relays = &relays; assert_eq!( causes(problems(&facts)), [ - "The relay is unreachable (timed out), so nothing reaches viewers until it reconnects.", + "The relay https://a is unreachable (timed out), so nothing reaches viewers until a relay reconnects.", "Source `camera` stalled: no new picture for 3 s, so the program repeats its last one.", "The relay is far away: a round trip to it takes 400 ms.", ] ); } + fn relay(endpoint: &str, state: State) -> Relay { + Relay { + endpoint: endpoint.into(), + state, + version: None, + } + } + + /// While another relay serves, an unreachable one costs viewers their + /// fallback rather than the stream, whichever of the two it is; with none + /// serving, nothing reaches viewers. + #[test] + fn a_lost_relay_is_an_outage_only_without_another() { + let local = Local::default(); + let uplink = Uplink::default(); + let lost = |relays: &[Relay]| { + let mut facts = facts(&local, &uplink, &[], &[]); + facts.relays = relays; + problems(&facts) + .into_iter() + .map(|problem| (problem.verdict, problem.cause)) + .collect::>() + }; + let down = || State::Disconnected("timed out".into()); + assert_eq!( + lost(&[relay("https://a", down()), relay("https://b", State::Connected)]), + [( + Verdict::Degraded, + "The relay https://a is unreachable (timed out), so the stream reaches viewers only through https://b." + .to_string() + )] + ); + assert_eq!( + lost(&[relay("https://a", State::Connected), relay("https://b", down())]), + [( + Verdict::Degraded, + "The relay https://b is unreachable (timed out), so the stream reaches viewers only through https://a." + .to_string() + )] + ); + let outage = lost(&[relay("https://a", down()), relay("https://b", State::Connecting)]); + assert_eq!( + outage.iter().map(|(verdict, _)| *verdict).collect::>(), + [Verdict::Unhealthy, Verdict::Degraded] + ); + } + /// The relay's count is the sessions on the broadcast itself, over every /// tier, not those on broadcasts nested beneath it. #[test] @@ -1116,7 +1228,10 @@ mod tests { observer: Observer::new(path, None), catalog: catalog.clone(), clock: clock.clone(), - monitor: connection.monitor(), + relays: vec![Link { + endpoint: "https://127.0.0.1:1".into(), + monitor: connection.monitor(), + }], feedback: feedback.clone(), watching: Watching::default(), })); diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index 2f8e4e8b..001ae7c0 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -38,6 +38,14 @@ pub struct Config { #[usage(flatten)] pub client: moq_tokio::connect::Config, + /// A second relay to publish the same broadcast to, so a viewer linked to + /// both resumes from it when the `--connect` relay fails. Both must speak + /// the one `--connect-version`, which must carry publisher epochs. Its + /// credential rides its URL's `?jwt=`: `--connect-token` goes to the + /// `--connect` relay only. + #[usage(long)] + pub failover: Option, + /// QUIC transport tuning (`--quic-*`). #[usage(flatten)] pub quic: moq_tokio::quic::Config, @@ -55,6 +63,44 @@ impl Config { anyhow::ensure!(deprecated.is_empty(), "{deprecated}"); Ok(()) } + + /// The relays to publish to, the `--connect` one first; none without it. + fn relays(&self) -> anyhow::Result> { + let Some(url) = self.client.url.clone() else { + anyhow::ensure!(self.failover.is_none(), "--failover needs --connect"); + return Ok(Vec::new()); + }; + if self.failover.is_some() { + self.check_failover()?; + } + let client = self.client.clone().init(self.quic.clone())?; + let mut relays = vec![engine::Relay { client, url }]; + if let Some(url) = self.failover.clone() { + let mut client = self.client.clone(); + // Its own credentials ride its URL; another operator's never reach it. + client.tokens.clear(); + let client = client + .init(self.quic.clone()) + .context("cannot set up the failover relay")?; + relays.push(engine::Relay { client, url }); + } + Ok(relays) + } + + /// Refuse relays a viewer could not move between without a restart: a copy + /// resumes mid-group only under the same epoch, which needs both links on + /// one protocol version that carries it. + fn check_failover(&self) -> anyhow::Result<()> { + let [version] = self.client.version.as_slice() else { + anyhow::bail!("--failover needs exactly one --connect-version, so both relays negotiate the same protocol"); + }; + anyhow::ensure!( + matches!(version, moq_net::Version::Lite(lite) if lite.has_epoch()), + "--failover needs a --connect-version with publisher epochs (moq-lite-07 or newer), not {version}, \ + or a viewer moving between relays restarts the stream rather than resuming it" + ); + Ok(()) + } } fn main() -> anyhow::Result<()> { @@ -66,17 +112,14 @@ fn main() -> anyhow::Result<()> { let runtime = tokio::runtime::Runtime::new()?; // The QUIC client binds its socket on the current runtime. let _runtime = runtime.enter(); + let relays = config.relays()?; let gpu = pollster::block_on(engine::Gpu::new())?; let (autosave, loaded) = autosave::Autosave::open(config.scenes.clone())?; - let connect = match config.client.url.clone() { - Some(url) => Some((config.client.clone().init(config.quic.clone())?, url)), - None => None, - }; let engine = runtime.block_on(engine::Engine::start( gpu.clone(), autosave, loaded, - connect, + relays, &config.broadcast, config.on_demand, ))?; @@ -138,4 +181,28 @@ mod tests { assert!(!config.on_demand, "the program encodes whatever the demand by default"); assert_eq!(config.client.url.unwrap().port(), Some(4443)); } + + /// A failover relay is accepted only on one protocol version that carries + /// publisher epochs, so a viewer moving between the relays resumes rather + /// than restarts. + #[test] + fn failover_needs_one_version_with_epochs() { + let check = |versions: &[&str]| { + let mut args = vec!["show.toml", "--connect", "https://a/", "--failover", "https://b/"]; + for version in versions { + args.extend(["--connect-version", version]); + } + let args: Vec<_> = args.into_iter().map(std::ffi::OsStr::new).collect(); + Config::parse_from(&args).unwrap().check_failover() + }; + check(&["moq-lite-07-wip"]).expect("one version with epochs"); + for refused in [ + &[][..], + &["moq-lite-06"], + &["moq-transport-22"], + &["moq-lite-07-wip", "moq-lite-06"], + ] { + assert!(check(refused).is_err(), "{refused:?} was accepted"); + } + } } diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 05185cc9..ae6f12ee 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1070,11 +1070,11 @@ impl Panels { } } - /// How the stream reaches viewers: what is wrong and why, the uplink, each - /// track, and each viewer that reports. + /// How the stream reaches viewers: what is wrong and why, each relay, the + /// uplink, each track, and each viewer that reports. fn health(&mut self, ui: &mut Ui) { let report = self.engine.health.lock().unwrap(); - let Some(relay) = &report.relay else { + if report.relays.is_empty() { ui.label("Not publishing; start with --connect to see how the stream is delivered."); for problem in &report.problems { ui.colored_label(color(ui, problem.verdict), &problem.cause); @@ -1091,20 +1091,37 @@ impl Panels { } ui.separator(); + egui::Grid::new("relays").striped(true).show(ui, |ui| { + for heading in ["Relay", "State", "Protocol"] { + ui.strong(heading); + } + ui.end_row(); + for relay in &report.relays { + ui.label(&relay.endpoint); + match &relay.state { + health::State::Connecting => ui.weak("connecting"), + health::State::Connected => ui.colored_label(HEALTHY, "connected"), + health::State::Disconnected(reason) => ui + .colored_label(ui.visuals().error_fg_color, "unreachable") + .on_hover_text(reason), + }; + ui.label(or_unknown(relay.version.map(|version| version.to_string()))); + ui.end_row(); + } + }); + ui.separator(); + let uplink = &report.uplink; let viewers = &report.viewers; egui::Grid::new("uplink").show(ui, |ui| { - ui.label("Relay"); - match relay { - health::Relay::Connecting => ui.weak("connecting"), - health::Relay::Connected => ui.colored_label(HEALTHY, "connected"), - health::Relay::Disconnected(reason) => ui - .colored_label(ui.visuals().error_fg_color, "unreachable") - .on_hover_text(reason), - }; - ui.end_row(); ui.label("Uplink"); - verdict(ui, uplink.verdict); + ui.horizontal(|ui| { + verdict(ui, uplink.verdict); + match &uplink.relay { + Some(relay) => ui.weak(format!("to {relay}")), + None => ui.weak("no relay connected"), + }; + }); ui.end_row(); ui.label("Round trip"); ui.label(or_unknown(uplink.rtt.map(|rtt| format!("{} ms", rtt.as_millis())))); @@ -1120,8 +1137,8 @@ impl Panels { ui.end_row(); ui.label("Watching"); match viewers.watching { - Some(watching) => ui.label(format!("{watching} sessions, by the relay's stats")), - None => ui.weak("unknown; the relay shares no stats with this connection"), + Some(watching) => ui.label(format!("{watching} sessions, by the relays' stats")), + None => ui.weak("unknown; no relay shares its stats with this studio"), }; ui.end_row(); ui.label("Reporting"); From 5f0864f382d3261890fff1eb3307c847a7b987e7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 22:02:49 -0800 Subject: [PATCH 3/5] fix(studio): refuse a fixed bind port with a failover relay Each relay's client binds its own socket, so a fixed --connect-bind port would be bound twice. Document that relays sharing a stats node name count once. Co-Authored-By: Claude Opus 5.5 --- README.md | 11 +++++++---- rs/moq-studio/src/main.rs | 24 ++++++++++++++++-------- 2 files changed, 23 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 7d4965f4..a9c8587e 100644 --- a/README.md +++ b/README.md @@ -21,9 +21,10 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ same publisher epoch, so a viewer linked to both resumes mid-group from it when the first fails. Both relays must speak the one `--connect-version`, which must carry publisher epochs (`moq-lite-07-wip` today, which a relay - accepts only when configured to), or the studio refuses to start. Its - credential rides its URL's `?jwt=`; `--connect-token` goes to the first - relay only. + accepts only when configured to), or the studio refuses to start; it also + refuses a fixed `--connect-bind` port, since each relay's connection binds + its own. Its credential rides its URL's `?jwt=`; `--connect-token` goes to + the first relay only. - `capture` adds camera, display, window, microphone, and system-audio sources. On Linux it links ALSA; add `pipewire` for PipeWire cameras. `openh264` is the software H.264 fallback for a machine without a hardware encoder. @@ -252,7 +253,9 @@ it is unhealthy. Below that it shows: - **Watching**: the sessions subscribed to the broadcast, from the relays' `.stats` broadcasts. A relay shares those only with a connection allowed to read them, so most show "unknown". In a cluster, a relay forwarding the - stream to another counts as one session. + stream to another counts as one session. Two relays whose stats use the + same node name publish them at the same path, so only the first relay's + count. - **Tracks**: each track's bitrate and frame rate, and, from viewers that report, the lowest bitrate any received, its worst grade, and the latency from the program clock to the slowest viewer. Latency compares the two diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index 001ae7c0..508c820f 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -99,6 +99,12 @@ impl Config { "--failover needs a --connect-version with publisher epochs (moq-lite-07 or newer), not {version}, \ or a viewer moving between relays restarts the stream rather than resuming it" ); + // Each relay's client binds its own socket, so a fixed port would be + // bound twice. + anyhow::ensure!( + self.client.bind.is_none_or(|bind| bind.port() == 0), + "--failover cannot share a fixed --connect-bind port between the two relays; leave the port 0" + ); Ok(()) } } @@ -187,20 +193,22 @@ mod tests { /// than restarts. #[test] fn failover_needs_one_version_with_epochs() { - let check = |versions: &[&str]| { + let check = |extra: &[&str]| { let mut args = vec!["show.toml", "--connect", "https://a/", "--failover", "https://b/"]; - for version in versions { - args.extend(["--connect-version", version]); - } + args.extend(extra); let args: Vec<_> = args.into_iter().map(std::ffi::OsStr::new).collect(); Config::parse_from(&args).unwrap().check_failover() }; - check(&["moq-lite-07-wip"]).expect("one version with epochs"); + let lite07 = ["--connect-version", "moq-lite-07-wip"]; + check(&lite07).expect("one version with epochs"); + check(&[&lite07[..], &["--connect-bind", "127.0.0.1:0"]].concat()).expect("an ephemeral port"); for refused in [ &[][..], - &["moq-lite-06"], - &["moq-transport-22"], - &["moq-lite-07-wip", "moq-lite-06"], + &["--connect-version", "moq-lite-06"], + &["--connect-version", "moq-transport-22"], + &[&lite07[..], &["--connect-version", "moq-lite-06"]].concat(), + // Each relay binds its own socket. + &[&lite07[..], &["--connect-bind", "127.0.0.1:4444"]].concat(), ] { assert!(check(refused).is_err(), "{refused:?} was accepted"); } From 73d9825d14107511599ef957efe93bc277b40698 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 06:40:37 -0800 Subject: [PATCH 4/5] docs(studio): note that bandwidth follows the first relay Co-Authored-By: Claude Opus 5.5 --- README.md | 4 +++- rs/moq-studio/src/engine.rs | 3 ++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 3d16422e..d9952e41 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,9 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ accepts only when configured to), or the studio refuses to start; it also refuses a fixed `--connect-bind` port, since each relay's connection binds its own. Its credential rides its URL's `?jwt=`; `--connect-token` goes to - the first relay only. + the first relay only. Captured sources follow the first relay's bandwidth + estimate alone, so while only the failover serves, they hold their last + bitrate rather than fit its link. - `capture` adds camera, display, window, microphone, and system-audio sources. On Linux it links ALSA; add `pipewire` for PipeWire cameras. `openh264` is the software H.264 fallback for a machine without a hardware encoder. diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 991381cb..817ca4ff 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -574,7 +574,8 @@ async fn publish( let publication = origin.create_broadcast(&path).context("cannot create the broadcast")?; let mut broadcast = publication.broadcast().clone(); // Each track claims its bitrate on the first relay's estimate, which that - // connection then probes to fund: the bandwidth the stream is granted. + // connection then probes to fund: the bandwidth the stream is granted. An + // allocator takes one connection's estimate, so the failover's goes unused. let bandwidth = connections[0].allocator(); let catalog = moq_mux::catalog::Config::default() .with_bandwidth(bandwidth.clone()) From 3038387c4e485307838d62d63596e5a2c8be778b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 13:13:51 -0800 Subject: [PATCH 5/5] fix(studio): document how failover sources and bandwidth actually behave A newer instance of a broadcast source wins before the first relay's rank, so the README and the engine comment say so, and the failover test proves it. No encoder adapts to a relay's estimate: the tracks only claim their bitrate on the first relay's link, which is what makes it the one probed. Name the version the epoch check accepts, and show a host-less relay's path. Co-Authored-By: Claude Opus 5.5 --- README.md | 20 ++++----- rs/moq-studio/src/engine.rs | 81 ++++++++++++++++++++++++++++--------- rs/moq-studio/src/main.rs | 2 +- rs/moq-studio/src/setup.rs | 3 +- 4 files changed, 77 insertions(+), 29 deletions(-) diff --git a/README.md b/README.md index d9315523..633c0da1 100644 --- a/README.md +++ b/README.md @@ -25,9 +25,9 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ accepts only when configured to), or the studio refuses to start; it also refuses a fixed `--connect-bind` port, since each relay's connection binds its own. Its credential rides its URL's `?jwt=`; `--connect-token` goes to - the first relay only. Captured sources follow the first relay's bandwidth - estimate alone, so while only the failover serves, they hold their last - bitrate rather than fit its link. + the first relay only. The stream's tracks claim their bitrate on the first + relay's link alone, so only that link is probed toward the stream's rate; + no encoder adapts to either estimate. - `capture` adds camera, display, window, microphone, and system-audio sources. On Linux it links ALSA; add `pipewire` for PipeWire cameras. `openh264` is the software H.264 fallback for a machine without a hardware encoder. @@ -57,8 +57,8 @@ the studio. It runs three checks, and a check that fails shows its reason: path for about six seconds and subscribes to it through the relay, so the relay pulls it. The congestion controller's send-rate estimate is the measurement. With `--failover`, only the `--connect` relay is measured, - since the stream's bandwidth follows it. A connection that reports no estimate, such as WebSocket, fails - the check. + since only its link is probed toward the stream's rate. A connection that + reports no estimate, such as WebSocket, fails the check. Setup then proposes a scene named Main and a canvas, explaining each choice. The scene has the first display filling the canvas, the first camera in the @@ -126,8 +126,10 @@ comes along, renamed if its name is taken. Copied filters paste onto the selected source or item, after its own. The clipboard holds text in the scene file's format. -A broadcast source subscribes through the same relay connections, from the -first relay while it has the broadcast and the failover only while it does not, plays the +A broadcast source subscribes through the same relay connections. It plays the +newest instance of its broadcast through either relay, preferring the first +relay among routes to the same instance, so a publisher restarted behind only +the failover replaces the first relay's copy. It plays the best video and audio its catalog lists, switches when the catalog changes, and comes back on its own when the publisher restarts. Audio from every source is mixed into the program whether or not its scene is live; mute a source to leave it out. @@ -348,8 +350,8 @@ it is unhealthy. Below that it shows: `.stats` broadcasts. A relay shares those only with a connection allowed to read them, so most show "unknown". In a cluster, a relay forwarding the stream to another counts as one session. Two relays whose stats use the - same node name publish them at the same path, so only the first relay's - count. + same node name publish them at the same path, so only one of them counts: + the newer instance, or the first relay's when they are one instance. - **Tracks**: each track's bitrate and frame rate, and, from viewers that report, the lowest bitrate any received, its worst grade, and the latency from the program clock to the slowest viewer. Latency compares the two diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index e111ea67..32b0fa76 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -588,7 +588,7 @@ async fn publish( let mut broadcast = publication.broadcast().clone(); // Each track claims its bitrate on the first relay's estimate, which that // connection then probes to fund: the bandwidth the stream is granted. An - // allocator takes one connection's estimate, so the failover's goes unused. + // allocator spans one connection, so the failover's link is never probed. let bandwidth = connections[0].allocator(); let catalog = moq_mux::catalog::Config::default() .with_bandwidth(bandwidth.clone()) @@ -688,13 +688,18 @@ fn connect( .map(|(rank, Relay { client, url })| { tracing::info!(url = %moq_tokio::RedactedUrl::new(&url), "connecting"); // Only the scheme, host, and port: a path can carry a credential, and - // the panel may be on a shared screen. - let mut endpoint = format!("{}://{}", url.scheme(), url.host_str().unwrap_or_default()); - if let Some(port) = url.port() { - endpoint.push_str(&format!(":{port}")); - } - // A broadcast source comes through the first relay while it has a - // route, and through the failover only while the first has none. + // the panel may be on a shared screen. A URL without a host, such as + // a unix socket's, keeps its path, which is the address. + let endpoint = match url.host_str() { + Some(host) => match url.port() { + Some(port) => format!("{}://{host}:{port}", url.scheme()), + None => format!("{}://{host}", url.scheme()), + }, + None => format!("{}:{}", url.scheme(), url.path()), + }; + // Among routes of one instance of a broadcast source, the first + // relay's wins while it has one. A newer instance wins first, + // whichever relay it comes through. let preference = match rank { 0 => moq_net::origin::Preference::default(), _ => moq_net::origin::Preference::new(-1), @@ -1078,8 +1083,9 @@ mod tests { /// With a failover relay, both relays route the program under its one /// epoch over the one protocol version, so a viewer moving between them - /// resumes; and a broadcast source only the failover has still reaches - /// the studio. + /// resumes. A broadcast source only the failover has still reaches the + /// studio, and a newer instance of a source behind the failover replaces + /// the first relay's. #[tokio::test(flavor = "multi_thread")] async fn every_relay_routes_one_epoch() { let _ = moq_tokio::crypto::install_default(); @@ -1113,11 +1119,21 @@ mod tests { .announce(moq_net::origin::Route::default().with_epoch(epoch.clone())) .unwrap(); - let guest = moq_tokio::origin::spawn(); - let source = guest.create_broadcast("guest").unwrap(); - source.announce(Default::default()).unwrap(); - let (client, url) = guests.pop().unwrap(); - let _guest = client.with_publisher(&guest).connect(url); + // Publish `path` under `epoch` through one relay's guest. + let publish = |path: &str, epoch: Option<&moq_net::Epoch>, (client, url): &(moq_tokio::Client, url::Url)| { + let origin = moq_tokio::origin::spawn(); + let source = origin.create_broadcast(path).unwrap(); + let route = moq_net::origin::Route::default(); + let route = match epoch { + Some(epoch) => route.with_epoch(epoch.clone()), + None => route, + }; + source.announce(route).unwrap(); + let connection = client.clone().with_publisher(&origin).connect(url.clone()); + (origin, source, connection) + }; + let _guest = publish("guest", None, &guests[1]); + let mut sources = subscribe.consume().announced(); let routed = async { for relay in &routes { @@ -1142,11 +1158,40 @@ mod tests { .expect("the connection"); } } - subscribe.consume().follow("guest", None).unwrap().available().await + subscribe + .consume() + .follow("guest", None) + .unwrap() + .available() + .await + .expect("the source"); + + // The newest instance of a source wins before the first relay's rank, + // so one restarted behind the failover replaces the first relay's. + let older: moq_net::Epoch = "01900000-0000-7000-8000-000000000000".parse().unwrap(); + let newer: moq_net::Epoch = "01910000-0000-7000-8000-000000000000".parse().unwrap(); + let _first = publish("show", Some(&older), &guests[0]); + routed_at(&mut sources, &older).await; + let _restarted = publish("show", Some(&newer), &guests[1]); + routed_at(&mut sources, &newer).await; }; kio::clock::timeout(TIMEOUT, routed) .await - .expect("both relays route the program, and the failover the source") - .expect("the source"); + .expect("both relays route the program, the failover the sources"); + } + + /// Wait until `announced` routes `show` under `epoch`. + async fn routed_at(announced: &mut moq_net::announce::Consumer, epoch: &moq_net::Epoch) { + use moq_net::announce::Event; + loop { + let announce = match announced.next().await.expect("announcements") { + Event::Start(announce) | Event::Update(announce) | Event::Restart(announce) => announce, + Event::Takeover { announce, .. } => announce, + Event::End(_) => continue, + }; + if announce.prefix.as_str() == "show" && announce.route.epoch.as_ref() == Some(epoch) { + return; + } + } } } diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index 37d007e9..3e8d1d7e 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -101,7 +101,7 @@ impl Config { }; anyhow::ensure!( matches!(version, moq_net::Version::Lite(lite) if lite.has_epoch()), - "--failover needs a --connect-version with publisher epochs (moq-lite-07 or newer), not {version}, \ + "--failover needs a --connect-version with publisher epochs (moq-lite-07-wip today), not {version}, \ or a viewer moving between relays restarts the stream rather than resuming it" ); // Each relay's client binds its own socket, so a fixed port would be diff --git a/rs/moq-studio/src/setup.rs b/rs/moq-studio/src/setup.rs index a12c9790..acb8d557 100644 --- a/rs/moq-studio/src/setup.rs +++ b/rs/moq-studio/src/setup.rs @@ -589,7 +589,8 @@ impl Setup { .map(|(_, encoding)| encoding.total()) .max() .map(|top| top.scaled(1.0 / headroom())); - // The first relay's, which the stream's bandwidth follows. + // The first relay's: the stream's tracks claim their bitrate on its + // link alone, so it is the one probed toward the stream's rate. let connect = self .launch .relays