diff --git a/README.md b/README.md index aeb6222..633c0da 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,16 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ 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; 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 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. @@ -46,8 +56,9 @@ the studio. It runs three checks, and a check that fails shows its reason: - **Uplink:** with `--connect`, setup publishes a probe at the `--broadcast` 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. A connection that reports no estimate, such as WebSocket, fails - the check. + measurement. With `--failover`, only the `--connect` relay is measured, + 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 @@ -84,7 +95,7 @@ Every panel is a tab you can drag, split, or stack. | Filters | Crop the selected source or item, add, order, and set its filters, set an item's blend mode, and copy or paste filters. | | 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, the canvas, 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 @@ -115,7 +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 connection, 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. @@ -320,19 +334,24 @@ 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. + stream to another counts as one session. Two relays whose stats use the + 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/quest/m0/README.md b/quest/m0/README.md index 1336573..37e1902 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -90,7 +90,6 @@ repository. - [Live encode control](/quest/m0/encode-control.md) - simulcast rungs that follow the bandwidth grant, live bitrate and resolution, keyframes on request - [Health grades source tracks](/quest/m0/health-sources.md) - the Health panel covers each published source track, not only the program - [Audio sources report stalls](/quest/m0/audio-stall.md) - a dead microphone or system audio reads stalled like a camera -- [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 97f27a8..0000000 --- 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 c26910e..32b0fa7 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -202,12 +202,19 @@ 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. +#[derive(Clone)] +pub struct Relay { + pub client: moq_tokio::Client, + pub url: url::Url, +} + impl Drop for Output { fn drop(&mut self) { self.publication.close(); @@ -240,8 +247,8 @@ const AUDIO_PERIOD: Duration = Duration::from_millis(10); #[derive(Clone)] pub struct Launch { pub gpu: Gpu, - /// The relay to publish to, if any. - pub connect: Option<(moq_tokio::Client, url::Url)>, + /// The relays to publish to, the preferred first; none publishes nothing. + pub relays: Vec, /// The broadcast path the program is published at. pub broadcast: String, /// Whether the program video encodes only while someone watches. @@ -252,15 +259,15 @@ impl Engine { /// Open the `loaded` collection and history on `launch.gpu`, start the /// program clock, and save both to `autosave`'s files as they change, /// starting at once. - /// With `launch.connect`, publish the program at `launch.broadcast` on that - /// relay, encoding its video only while someone watches when - /// `launch.on_demand`. + /// Publish the program at `launch.broadcast` on every one of + /// `launch.relays`, preferred first, encoding its video only while someone + /// watches when `launch.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(launch: Launch, autosave: Autosave, loaded: autosave::Loaded) -> anyhow::Result> { let Launch { gpu, - connect, + relays, broadcast, on_demand, } = launch; @@ -274,12 +281,11 @@ impl Engine { } = loaded; 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)) }; // Absolute, so a copied image names the same file in another collection. @@ -561,13 +567,12 @@ struct Publish { sources: moq_scene::publish::Publish, } -/// Announce the program broadcast, which also carries each local source's own -/// track, and connect to the relay, which also carries `subscribe`'s broadcast -/// sources. +/// Announce the program broadcast under one epoch, which also carries each +/// local source's own track, 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, @@ -575,15 +580,16 @@ 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 - // connection then probes to fund: the bandwidth the stream is granted. - let bandwidth = connection.allocator(); + // 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 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()) .with_catalog(Default::default()); @@ -625,6 +631,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())) @@ -641,7 +649,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, }; @@ -649,7 +657,7 @@ async fn publish( let output = Output { path: path.to_string(), on_demand, - _connection: connection, + _connections: connections, publication, }; Ok(( @@ -666,6 +674,44 @@ 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. 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), + }; + 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 { @@ -872,7 +918,10 @@ mod tests { let launch = Launch { gpu: Gpu::new().await.expect("a GPU"), - connect: Some((client, fixture.url.clone())), + relays: vec![Relay { + client, + url: fixture.url.clone(), + }], broadcast: "studio".into(), on_demand: false, }; @@ -1020,7 +1069,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; } } @@ -1030,4 +1080,118 @@ 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. 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(); + 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(); + + // 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 { + 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 + .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, 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/health.rs b/rs/moq-studio/src/health.rs index a7ec374..215aabf 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] @@ -1117,7 +1229,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 ffa3a7f..3e8d1d7 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -43,6 +43,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, @@ -60,6 +68,50 @@ 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-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 + // 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(()) + } } fn main() -> anyhow::Result<()> { @@ -71,14 +123,11 @@ 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 connect = match config.client.url.clone() { - Some(url) => Some((config.client.clone().init(config.quic.clone())?, url)), - None => None, - }; let launch = engine::Launch { gpu: gpu.clone(), - connect, + relays, broadcast: config.broadcast.clone(), on_demand: config.on_demand, }; @@ -210,4 +259,30 @@ 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 = |extra: &[&str]| { + let mut args = vec!["show.toml", "--connect", "https://a/", "--failover", "https://b/"]; + args.extend(extra); + let args: Vec<_> = args.into_iter().map(std::ffi::OsStr::new).collect(); + Config::parse_from(&args).unwrap().check_failover() + }; + 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 [ + &[][..], + &["--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"); + } + } } diff --git a/rs/moq-studio/src/setup.rs b/rs/moq-studio/src/setup.rs index 103f0be..acb8d55 100644 --- a/rs/moq-studio/src/setup.rs +++ b/rs/moq-studio/src/setup.rs @@ -589,11 +589,13 @@ impl Setup { .map(|(_, encoding)| encoding.total()) .max() .map(|top| top.scaled(1.0 / headroom())); + // 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 - .connect - .clone() - .map(|(client, url)| (client, url, self.launch.broadcast.clone())); + .relays + .first() + .map(|relay| (relay.client.clone(), relay.url.clone(), self.launch.broadcast.clone())); self.uplink = Some(match (connect, ceiling) { (None, _) => Task::Done(Err("no relay is configured; start with --connect to measure it".into())), (Some(_), None) => Task::Done(Err("nothing encodes on this machine, so nothing would be sent".into())), diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 21a9105..cd77ae6 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1332,11 +1332,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); @@ -1353,20 +1353,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())))); @@ -1382,8 +1399,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");