From cf55621271637ba61cea529648cd5f2e1e77a548 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 22:13:47 -0800 Subject: [PATCH 1/3] chore(quest): claim quest/m0/health-sources.md Quest-Claim: 9d403c94ee1f6ecb From 738ab24a2fac5c1994552e0d68bd777e6e1d2f1e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 07:07:31 -0800 Subject: [PATCH 2/3] feat(studio): Health grades each source's own tracks The sampler grades the sources' tracks beside the program's, from what each source listing wrote and what viewers report on them, grouped by source in the Tracks table. Uplink media stays the program's, since source tracks encode whether or not anyone pulls them. Co-Authored-By: Claude Opus 5.5 --- README.md | 13 ++- quest/m0/README.md | 1 - quest/m0/health-sources.md | 22 ----- rs/moq-scene/src/publish.rs | 26 ++++- rs/moq-studio/src/engine.rs | 1 + rs/moq-studio/src/health.rs | 192 ++++++++++++++++++++++++++++++------ rs/moq-studio/src/ui.rs | 13 +++ 7 files changed, 208 insertions(+), 60 deletions(-) delete mode 100644 quest/m0/health-sources.md diff --git a/README.md b/README.md index 0cf243c..d2b24f6 100644 --- a/README.md +++ b/README.md @@ -241,17 +241,20 @@ 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. + estimates it can carry, the media the program's tracks write, and the share + lost and resent. Loss and headroom are graded only while someone watches, + since the relay pulls nothing otherwise. Source tracks are left out of the + media, since they encode whether or not anyone pulls them. - **Watching**: the sessions subscribed to the broadcast, from the relay's `.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. - **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 - hosts' wall clocks, so keep both synced. + from the program clock to the slowest viewer. The program's tracks come + first, then each source's own tracks, grouped by source; a still image or + color carries no media to grade. Latency compares the two hosts' wall + clocks, so keep both synced. - **Viewers**: each viewer that reports, with its round trip, latency, and playout delay. The studio reads 64 at most, a sample the grades rest on: it counts the rest, and a reader one frees goes to the next viewer to arrive. diff --git a/quest/m0/README.md b/quest/m0/README.md index c21f350..1f39978 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -89,7 +89,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 -- [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 diff --git a/quest/m0/health-sources.md b/quest/m0/health-sources.md deleted file mode 100644 index 95354ea..0000000 --- a/quest/m0/health-sources.md +++ /dev/null @@ -1,22 +0,0 @@ -# [S] Health grades source tracks - -## Goal - -The Health panel grades each published source track the way it grades the -program's: sent rate, viewers reporting, and the worst grade and latency they -report, so a source someone watches directly is covered too. - -## Plan - -Source tracks are listed in their own nested catalogs under the studio's -catalog (see `README.md`, Source tracks), and the health sampler reads only -the root renditions today. Viewers report per rendition through `.echo`, so -the per-track table and its causes extend to the source renditions rather than -adding a second path. Keep the table readable when there are many sources, for -example grouped by source. Raised on -[#3](https://github.com/moq-dev/studio/pull/3). - -## Related - -- [Source tracks](/quest/m0/compose/sources.md) - the audio source tracks this - grades once they land diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 3fbb6b8..03b375a 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -60,11 +60,13 @@ pub struct Publish { owners: Arc>, } -/// Which [`Claim`] owns each source name's entry in the studio catalog. +/// Which [`Claim`] owns each source name's entry in the studio catalog, and +/// each open [`Listing`]'s catalog by its claim. #[derive(Default)] struct Owners { next: u64, by_name: BTreeMap, + catalogs: BTreeMap, } impl Publish { @@ -90,6 +92,15 @@ impl Publish { } } + /// What each source track has written, keyed by rendition ID, as + /// [`Catalog::stats`](moq_mux::catalog::Producer::stats) reports the + /// program's. The broadcast names every track uniquely, so all sources share + /// one map, and a replaced source's tracks count until it stops. + pub fn stats(&self) -> BTreeMap { + let owners = self.owners.lock().unwrap(); + owners.catalogs.values().flat_map(|catalog| catalog.stats()).collect() + } + /// Claim `name`'s entry in the studio catalog for a source that is opening, /// taking it from any older source of that name. pub(crate) fn claim(&self, name: &str) -> Claim { @@ -155,6 +166,7 @@ impl<'a> Listing<'a> { .with_bandwidth(publish.bandwidth.clone()) .with_hls(None); let catalog = moq_mux::catalog::Producer::new(&mut scratch, config)?; + claim.owners.lock().unwrap().catalogs.insert(claim.id, catalog.clone()); Ok(Self { claim, kind: publish.kind.clone(), @@ -194,6 +206,7 @@ impl<'a> Listing<'a> { impl Drop for Listing<'_> { fn drop(&mut self) { + self.claim.owners.lock().unwrap().catalogs.remove(&self.claim.id); // Fails only once the studio catalog closed, which took the entry with it. let _ = self.studio.mutate(|root| { if self.claim.owned() { @@ -539,6 +552,17 @@ mod tests { } } => {} } + + // Every frame written counts for the Health panel, until the + // source stops. + let stats = publish.stats(); + assert_eq!( + stats.values().map(|rendition| rendition.frames).sum::(), + 5, + "{stats:?}" + ); + drop(listing); + assert!(publish.stats().is_empty()); }) .await; } diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index d4d466b..85017aa 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -632,6 +632,7 @@ async fn publish( monitor: connection.monitor(), feedback, watching, + sources: sources.clone(), }; let output = Output { diff --git a/rs/moq-studio/src/health.rs b/rs/moq-studio/src/health.rs index a7ec374..0cd4fff 100644 --- a/rs/moq-studio/src/health.rs +++ b/rs/moq-studio/src/health.rs @@ -46,7 +46,8 @@ pub struct Report { pub relay: Option, pub uplink: Uplink, pub viewers: Viewers, - /// Each track of the broadcast, in catalog order. + /// Each track of the broadcast in catalog order: the program's, then each + /// source's by source name. pub tracks: Vec, /// What is wrong, worst first, each naming its cause. pub problems: Vec, @@ -70,7 +71,9 @@ pub struct Uplink { pub rtt: Option, /// What the congestion controller estimates the connection can carry. pub capacity: Option, - /// The media every track wrote over the last interval. + /// The media the program's tracks wrote over the last interval. Source + /// tracks are left out: they encode whether or not anyone pulls them, and + /// the relay uploads only what someone does. pub media: Option, /// Transport bytes lost per media byte written over the last interval. pub loss: Option, @@ -121,6 +124,8 @@ pub struct Viewer { /// One track of the broadcast. #[derive(Debug)] pub struct Track { + /// The source whose own track it is, or `None` for the program's. + pub source: Option, /// Its catalog rendition ID. pub id: String, /// What it carries, such as `video H.264 1920x1080`. @@ -139,6 +144,16 @@ pub struct Track { pub latency: Option, } +impl Track { + /// How a problem names it. + fn name(&self) -> String { + match &self.source { + Some(source) => format!("source `{source}`'s {}", self.label), + None => self.label.clone(), + } + } +} + /// Something wrong with the stream, and why. #[derive(Debug, Clone, PartialEq)] pub struct Problem { @@ -181,6 +196,8 @@ pub struct Publisher { pub monitor: moq_tokio::connection::Monitor, pub feedback: Feedback, pub watching: Watching, + /// Where the sources publish their own tracks, for what those wrote. + pub sources: moq_scene::publish::Publish, } /// Grades the stream every interval. @@ -195,6 +212,7 @@ struct Previous { at: Instant, local: Local, report: Option>, + sources: Option>, } impl Sampler { @@ -216,24 +234,28 @@ impl Sampler { let mut report = Report::default(); let mut graded = Graded::default(); - let mut current = None; + let (mut current, mut sources) = (None, None); if let Some(publisher) = &self.publisher { report.relay = Some(relay(&publisher.monitor)); - let snapshot = snapshot(publisher); - 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) { - sample = sample.with_previous(previous, interval); - } - let graded_ours = health::publisher(&sample, &self.thresholds); + let ours = health::Report::new(publisher.observer.clone(), snapshot(publisher)); + let previous_ours = previous.as_ref().and_then(|previous| previous.report.as_ref()); + let graded_ours = self.grade(&ours, previous_ours.zip(interval)); report.uplink = uplink(&ours.snapshot, &graded_ours, local.watched); graded.publisher = Some(graded_ours); + // Graded apart from the program's, so their media stays out of the + // uplink's; see [`Uplink::media`]. + let mut snapshot = hang::stats::Snapshot::default(); + snapshot.renditions = publisher.sources.stats(); + let theirs = health::Report::new(publisher.observer.clone(), snapshot); + let previous_theirs = previous.as_ref().and_then(|previous| previous.sources.as_ref()); + graded.sources = Some(self.grade(&theirs, previous_theirs.zip(interval))); graded.viewers = publisher.feedback.grade(now, &self.thresholds); report.viewers.unread = publisher.feedback.unread(); report.viewers.watching = publisher.watching.get(); - report.tracks = tracks(publisher, &graded); + report.tracks = tracks(&publisher.catalog.snapshot(), &publisher.clock, &graded); report.viewers.reports = viewers(publisher, &graded.viewers); current = Some(ours); + sources = Some(theirs); } let drops = previous.as_ref().map(|previous| Drops { @@ -255,9 +277,23 @@ impl Sampler { at: now, local, report: current, + sources, }); report } + + /// Grade what the studio wrote, against `previous` and the interval since. + fn grade( + &self, + current: &health::Report, + previous: Option<(&health::Report, Duration)>, + ) -> Health { + let mut sample = health::Sample::new(current, Duration::ZERO); + if let Some((previous, interval)) = previous { + sample = sample.with_previous(previous, interval); + } + health::publisher(&sample, &self.thresholds) + } } fn relay(monitor: &moq_tokio::connection::Monitor) -> Relay { @@ -318,28 +354,21 @@ fn uplink(snapshot: &hang::stats::Snapshot, graded: &Health, watched: bool) -> U } } -/// Each catalog track, with what it sent and what its viewers received. -fn tracks(publisher: &Publisher, graded: &Graded) -> Vec { - let catalog = publisher.catalog.snapshot(); - let sent = graded.publisher.as_ref(); - let video = catalog.video.renditions.iter().map(|(id, config)| { - let size = match (config.coded_width, config.coded_height) { - (Some(width), Some(height)) => format!(" {width}x{height}"), - _ => String::new(), - }; - (id, format!("video {}{size}", config.codec)) +/// Each media track `catalog` lists, the program's and then each source's, +/// with what it sent and what its viewers received. A source's stills carry +/// no media to grade, so they are left out. +fn tracks(catalog: &hang::Catalog, clock: &moq_mux::Clock, graded: &Graded) -> Vec { + let program = renditions(catalog).map(|(id, label)| (None, id, label, graded.publisher.as_ref())); + let sources = catalog.ext.studio.sources.iter().flat_map(|(name, catalog)| { + renditions(catalog).map(move |(id, label)| (Some(name), id, label, graded.sources.as_ref())) }); - let audio = catalog - .audio - .renditions - .iter() - .map(|(id, config)| (id, format!("audio {} {} ch", config.codec, config.channel_count))); - video - .chain(audio) - .map(|(id, label)| { + program + .chain(sources) + .map(|(source, id, label, sent)| { let ours = sent.and_then(|health| health.renditions.get(id)); let interval = sent.and_then(|health| health.interval); let mut track = Track { + source: source.cloned(), id: id.clone(), label, sent: ours.and_then(|rendition| rendition.bitrate).map(Rate::from_bps), @@ -362,7 +391,7 @@ fn tracks(publisher: &Publisher, graded: &Graded) -> Vec { if let Some(bitrate) = rendition.bitrate.map(Rate::from_bps) { track.received = Some(track.received.map_or(bitrate, |received| received.min(bitrate))); } - let latency = viewer.latency(id, &publisher.clock); + let latency = viewer.latency(id, clock); track.latency = track.latency.max(latency); } track @@ -370,6 +399,23 @@ fn tracks(publisher: &Publisher, graded: &Graded) -> Vec { .collect() } +/// Each video and audio rendition `catalog` lists, by ID, with what it carries. +fn renditions(catalog: &hang::Catalog) -> impl Iterator { + let video = catalog.video.renditions.iter().map(|(id, config)| { + let size = match (config.coded_width, config.coded_height) { + (Some(width), Some(height)) => format!(" {width}x{height}"), + _ => String::new(), + }; + (id, format!("video {}{size}", config.codec)) + }); + let audio = catalog + .audio + .renditions + .iter() + .map(|(id, config)| (id, format!("audio {} {} ch", config.codec, config.channel_count))); + video.chain(audio) +} + fn viewers(publisher: &Publisher, graded: &[GradedViewer]) -> Vec { graded .iter() @@ -402,7 +448,10 @@ fn viewers(publisher: &Publisher, graded: &[GradedViewer]) -> Vec { /// The grades of one interval. #[derive(Default)] struct Graded { + /// The program's tracks and the uplink. publisher: Option, + /// The sources' own tracks. + sources: Option, viewers: Vec, } @@ -592,7 +641,7 @@ fn viewer_problems(facts: &Facts<'_>, add: &mut impl FnMut(Verdict, String)) { let track = |id: Option<&str>| { id.and_then(|id| facts.labels.iter().find(|track| track.id == id)) - .map_or_else(|| "the stream".to_string(), |track| track.label.clone()) + .map_or_else(|| "the stream".to_string(), Track::name) }; for (input, verdict, count, id) in worst { let of = format!("{count} of {reporting} viewers"); @@ -896,6 +945,7 @@ mod tests { fn track(id: &str, label: &str) -> Track { Track { + source: None, id: id.into(), label: label.into(), sent: None, @@ -1026,6 +1076,80 @@ mod tests { ); } + /// What the studio wrote over one interval, graded: `bytes` and 30 frames + /// for each rendition. + fn wrote(renditions: &[(&str, u64)]) -> Health { + let observer = Observer::new(moq_net::Path::new("show").unwrap(), None); + let report = |scale: u64| { + let mut snapshot = hang::stats::Snapshot::<()>::default(); + for (id, bytes) in renditions { + let mut rendition = hang::stats::Rendition::default(); + rendition.bytes = bytes * scale; + rendition.frames = 30 * scale; + snapshot.renditions.insert(id.to_string(), rendition); + } + health::Report::new(observer.clone(), snapshot) + }; + let (previous, current) = (report(0), report(1)); + let sample = health::Sample::new(¤t, Duration::ZERO).with_previous(&previous, INTERVAL); + health::publisher(&sample, &Thresholds::default()) + } + + /// Each source's media tracks are graded like the program's, after them and + /// by source name, from what the source wrote and what viewers report on + /// them; a still has no media to grade. A viewer stalling on one names + /// its source. + #[test] + fn source_tracks_are_graded_by_source() { + let video = |width: u32, height: u32| { + let codec: hang::catalog::VideoCodec = "avc3.42c01f".parse().unwrap(); + let mut config = hang::catalog::VideoConfig::new(codec); + config.coded_width = Some(width); + config.coded_height = Some(height); + config + }; + let mut catalog = hang::Catalog::::default(); + catalog.video.renditions.insert("0.avc3".into(), video(1920, 1080)); + let mut camera = hang::Catalog::default(); + camera.video.renditions.insert("2.avc3".into(), video(1280, 720)); + let mut logo = hang::Catalog::default(); + let png = hang::catalog::BinaryConfig::new(hang::catalog::Mode::Snapshot); + logo.binary.tracks.insert("1.png".into(), png); + let sources = &mut catalog.ext.studio.sources; + sources.insert("logo".into(), logo); + sources.insert("camera".into(), camera); + + // 375 kB in a second is 3 Mb/s, and 125 kB is 1 Mb/s. + let graded = Graded { + publisher: Some(wrote(&[("0.avc3", 375_000)])), + sources: Some(wrote(&[("2.avc3", 125_000)])), + viewers: vec![stalling("alice", "2.avc3", Duration::from_millis(100))], + }; + let tracks = tracks(&catalog, &moq_mux::Clock::new(), &graded); + let rows: Vec<_> = tracks + .iter() + .map(|track| (track.source.as_deref(), track.id.as_str(), track.sent.map(rate))) + .collect(); + assert_eq!( + rows, + [ + (None, "0.avc3", Some("3.0 Mb/s".into())), + (Some("camera"), "2.avc3", Some("1.0 Mb/s".into())), + ] + ); + let camera = &tracks[1]; + assert_eq!(camera.framerate, Some(30.0)); + assert_eq!((camera.viewers, camera.verdict), (1, Verdict::Unhealthy)); + assert_eq!((tracks[0].viewers, tracks[0].verdict), (0, Verdict::Unknown)); + + let local = Local::default(); + let uplink = Uplink::default(); + assert_eq!( + causes(problems(&facts(&local, &uplink, &graded.viewers, &tracks))), + ["1 of 1 viewers are stalling on source `camera`'s video avc3.42c01f 1280x720."] + ); + } + /// Problems list worst first: an unreachable relay and a frozen source /// above a slow round trip. #[test] @@ -1120,6 +1244,12 @@ mod tests { monitor: connection.monitor(), feedback: feedback.clone(), watching: Watching::default(), + sources: moq_scene::publish::Publish::new( + broadcast.clone(), + catalog.clone(), + moq_net::bandwidth::Allocator::unlimited(), + moq_video::encode::Kind::Auto, + ), })); let alice = origin.create_broadcast("show/viewers/alice.echo").unwrap(); diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 74e76e9..715c054 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1152,6 +1152,7 @@ impl Panels { egui::Grid::new("tracks").striped(true).show(ui, |ui| { for heading in [ + "Source", "Track", "Sent", "Frames", @@ -1163,7 +1164,19 @@ impl Panels { ui.strong(heading); } ui.end_row(); + // Each source named once, beside its first track. + let mut group = None; for track in &report.tracks { + let source = Some(track.source.as_deref()); + if group != source { + match &track.source { + Some(name) => ui.label(name), + None => ui.weak("program"), + }; + group = source; + } else { + ui.label(""); + } ui.label(&track.label).on_hover_text(&track.id); ui.label(or_unknown(track.sent.map(health::rate))); ui.label(or_unknown(track.framerate.map(|framerate| format!("{framerate:.0}/s")))); From f3f9022f295f18e79699f8e9bf22a69e92788e36 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 17:58:52 -0800 Subject: [PATCH 3/3] fix(studio): grade only listed source tracks, and leave the program's Source cell blank A replaced source still stopping no longer counts tracks the studio catalog has dropped, so stats and the Tracks table agree. The program's group names no source, so a source called `program` cannot pass for it. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/publish.rs | 21 +++++++++++++-------- rs/moq-studio/src/ui.rs | 19 +++++++++---------- 2 files changed, 22 insertions(+), 18 deletions(-) diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 03b375a..3d54ad3 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -92,13 +92,15 @@ impl Publish { } } - /// What each source track has written, keyed by rendition ID, as - /// [`Catalog::stats`](moq_mux::catalog::Producer::stats) reports the - /// program's. The broadcast names every track uniquely, so all sources share - /// one map, and a replaced source's tracks count until it stops. + /// What each source track the studio catalog lists has written, keyed by + /// rendition ID, as [`Catalog::stats`](moq_mux::catalog::Producer::stats) + /// reports the program's. The broadcast names every track uniquely, so all + /// sources share one map. A replaced source's tracks drop out with its + /// catalog entry, even while it is still stopping. pub fn stats(&self) -> BTreeMap { let owners = self.owners.lock().unwrap(); - owners.catalogs.values().flat_map(|catalog| catalog.stats()).collect() + let listed = owners.by_name.values().filter_map(|id| owners.catalogs.get(id)); + listed.flat_map(|catalog| catalog.stats()).collect() } /// Claim `name`'s entry in the studio catalog for a source that is opening, @@ -553,16 +555,19 @@ mod tests { } => {} } - // Every frame written counts for the Health panel, until the - // source stops. + // Every frame written counts for the Health panel, while the + // studio catalog lists the source. let stats = publish.stats(); assert_eq!( stats.values().map(|rendition| rendition.frames).sum::(), 5, "{stats:?}" ); - drop(listing); + // A replacement takes the listing, and the old tracks with it. + let _newer = publish.claim("camera"); assert!(publish.stats().is_empty()); + drop(listing); + assert!(publish.owners.lock().unwrap().catalogs.is_empty()); }) .await; } diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 0ed1e11..8a24348 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1625,19 +1625,18 @@ impl Panels { ui.strong(heading); } ui.end_row(); - // Each source named once, beside its first track. + // Each source named once, beside its first track. The program's + // tracks come first and name none, so no source name can pass + // for them. let mut group = None; for track in &report.tracks { - let source = Some(track.source.as_deref()); - if group != source { - match &track.source { - Some(name) => ui.label(name), - None => ui.weak("program"), - }; - group = source; + let source = track.source.as_deref(); + ui.label(if source == group { + "" } else { - ui.label(""); - } + source.unwrap_or_default() + }); + group = source; ui.label(&track.label).on_hover_text(&track.id); ui.label(or_unknown(track.sent.map(health::rate))); ui.label(or_unknown(track.framerate.map(|framerate| format!("{framerate:.0}/s"))));