From 256bb33211cb48ce944381bcea2d46f6fac49c16 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 15:02:01 -0800 Subject: [PATCH 1/9] chore(quest): claim quest/m2/compose/watch.md Quest-Claim: 2f5f5a4ef6ee078b From e2ec27b11e574bf789107817aee86a9d8e99491f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 15:26:58 -0800 Subject: [PATCH 2/9] feat(studio): moq-studio --watch plays a studio broadcast with its layer layout Co-Authored-By: Claude Opus 5.5 --- README.md | 25 ++ quest/m2/compose/controls.md | 8 +- quest/m2/compose/hang-layers.md | 5 +- quest/m2/compose/layout.md | 9 +- quest/m2/compose/watch.md | 30 -- rs/moq-scene/src/collection.rs | 21 ++ rs/moq-scene/src/compositor.rs | 6 + rs/moq-scene/src/playout.rs | 2 +- rs/moq-scene/src/source.rs | 399 ++++++++++++++++++-- rs/moq-studio/src/engine.rs | 9 +- rs/moq-studio/src/main.rs | 75 +++- rs/moq-studio/src/ui.rs | 9 +- rs/moq-studio/src/watch.rs | 626 ++++++++++++++++++++++++++++++++ 13 files changed, 1140 insertions(+), 84 deletions(-) delete mode 100644 quest/m2/compose/watch.md create mode 100644 rs/moq-studio/src/watch.rs diff --git a/README.md b/README.md index cc7da69a..c016cebe 100644 --- a/README.md +++ b/README.md @@ -69,6 +69,8 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ Security. - The `--connect-*`, `--quic-*`, and `--log-level` flags are [moq-cli's](https://doc.moq.dev/bin/cli); see `moq-studio --help`. +- `moq-studio --watch ` opens no scene file and plays a studio broadcast + instead, as a viewer sees it; see [Watching](#watching). ## First launch @@ -993,6 +995,29 @@ A take changes the layout on the frame the program starts showing the new scene, so the published sources of the scene leaving go off and those of the scene coming in go on at once, even through a transition. +#### Watching + +`moq-studio --watch https://relay.example/anon/my-show` plays a studio +broadcast in a window as a viewer sees it: the program, with each picture of +its layer layout over it and each sound mixed with it. The URL's path names +the broadcast, as `moq play`'s does, and `--connect-token` and the `--quic-*` +flags apply to it; `--connect`, `--failover`, and `--room` are refused +beside it. It is a prototype for settling the layout's format, which goes +once `moq play` plays layers. + +- Each layout takes effect when the program plays the frame it is stamped + with, and its pictures' keyframes play from its `start`. +- Each source plays its own track on the program's clock: a picture shows + when the program's picture stamped alike does, and a sound plays with the + program's sound stamped alike, at its `gain`, `muted`, `pan`, and `delay`. + A frame too late for its moment is dropped. +- A broadcast without a layout, or with a layout of another `version`, plays + its program alone; the window says why for the latter. +- Guests' broadcasts are not in the layout yet, so only sources local to the + studio play as layers. `priority` and `toggle` are not applied. +- Sound needs a build with `capture`; without it the window says so in the + log and plays the pictures alone. + ### Recording **Start recording** in the Output panel saves the program at the canvas size, diff --git a/quest/m2/compose/controls.md b/quest/m2/compose/controls.md index a79b3c2e..c3b2390c 100644 --- a/quest/m2/compose/controls.md +++ b/quest/m2/compose/controls.md @@ -11,8 +11,8 @@ the studio. ## Plan Decided 2026-11-08 (maintainer, /quest-plan): the controls land in the native -player, the [watch prototype](/quest/m2/compose/watch.md) (rejected: browser -only). moq's players take them from hang's layer spec. +player, the `moq-studio --watch` prototype (`rs/moq-studio/src/watch.rs`; +rejected: browser only). moq's players take them from hang's layer spec. Decided 2026-11-09 (maintainer, /quest-audit): ✅ viewer toggles also land in moq's native player, @@ -28,7 +28,3 @@ layer is unsubscribed, so it costs the viewer and the relay nothing. The studio keeps encoding it, so toggling it back on can reach its recent past. Subscriptions are per track, not per layer: a source shown by several layers stays subscribed while any of them is on, at the highest priority among them. - -## Required - -- [moq-studio watch prototype](/quest/m2/compose/watch.md) - the viewer that exposes the toggles diff --git a/quest/m2/compose/hang-layers.md b/quest/m2/compose/hang-layers.md index dc52f82b..d88df9bc 100644 --- a/quest/m2/compose/hang-layers.md +++ b/quest/m2/compose/hang-layers.md @@ -13,7 +13,7 @@ Decided 2026-11-08 (maintainer, /quest-plan): the layer layout becomes official moq, so `moq play` and `@moq/watch` play a studio broadcast. Once the wait clears, advance the gitlink with `just upgrade-moq` in this PR, move the studio to hang's types, and delete the -[watch prototype](/quest/m2/compose/watch.md) and its flag. +`moq-studio --watch` prototype (`rs/moq-studio/src/watch.rs`) and its flag. Decided 2026-11-09 (maintainer, /quest-audit): ✅ the wait names moq's three quests (rejected: "both quests", unnamed, which left out `@moq/watch`). @@ -21,7 +21,7 @@ quests (rejected: "both quests", unnamed, which left out `@moq/watch`). ## Waiting - moq-dev/moq: its three layer quests land, planned in moq's m3 from the - [watch prototype](/quest/m2/compose/watch.md)'s format: + `moq-studio --watch` prototype's format, the README's "Layer layout": [hang specifies a layer layout track](https://github.com/moq-dev/moq/blob/main/quest/m3/layers/spec.md), [moq play composites layers](https://github.com/moq-dev/moq/blob/main/quest/m3/layers/play.md), and @@ -34,6 +34,5 @@ quests (rejected: "both quests", unnamed, which left out `@moq/watch`). ## Required -- [moq-studio watch prototype](/quest/m2/compose/watch.md) - the format moq specifies - [Viewer controls](/quest/m2/compose/controls.md) - toggles and priorities the spec carries - [The studio is caught up to moq's main](/quest/m0/pin-main.md) - moq's breaking changes since moq-dev/moq#8272, so this pin takes only its own diff --git a/quest/m2/compose/layout.md b/quest/m2/compose/layout.md index 17dfe6a2..5e43258b 100644 --- a/quest/m2/compose/layout.md +++ b/quest/m2/compose/layout.md @@ -19,6 +19,11 @@ studio's broadcast. Bump the layout's `version` if the change is not additive. +The `moq-studio --watch` prototype (`rs/moq-studio/src/watch.rs`) plays only +local sources' layers today: its layers play tracks of the program's own +broadcast on the program's clock, so a guest's layer needs the prototype to +follow the named broadcast, on the offset the layout carries. + ## Waiting - Maintainer: decide how the layout names a guest's broadcast. The epic @@ -43,7 +48,3 @@ Bump the layout's `version` if the change is not additive. Check: the maintainer answers in /quest-unblock, and the choice is recorded in this Plan and the epic's. - -## Related - -- [moq-studio watch prototype](/quest/m2/compose/watch.md) - the first player of the layout diff --git a/quest/m2/compose/watch.md b/quest/m2/compose/watch.md deleted file mode 100644 index 135d35e2..00000000 --- a/quest/m2/compose/watch.md +++ /dev/null @@ -1,30 +0,0 @@ -# [M] moq-studio watch prototype - -## Goal - -`moq-studio --watch ` plays a studio broadcast: the program, with each -source the layer layout names composited over it on the media clock, and -their audio mixed with the program's. A broadcast without a layer layout plays -its program alone. - -## Plan - -Decided 2026-11-08 (maintainer, /quest-plan): a throwaway prototype, so the -layer format settles without moq round trips before hang specifies it -(rejected: going straight to `moq play`). It lives in `moq-studio`, not its -own binary, since [hang-layers](/quest/m2/compose/hang-layers.md) deletes it -once `moq play` plays layers. - -Build it on moq-scene, which already plays remote broadcasts as sources for -guests (`rs/moq-studio/src/guests.rs`): a scene that the layout drives rather -than a scene file. `README.md` documents the flag while it exists. - -It plays the layer layout the studio publishes on `dev.moq.studio.layers` -(`rs/moq-scene/src/layout.rs`, and the README's "Layer layout"), refusing a -layout of another `version` loudly and playing the program alone. Its working -format is the starting point for moq's hang layer spec, which waits on this -quest. - -## Related - -- [Guests in the layer layout](/quest/m2/compose/layout.md) - guests' layers, once the maintainer decides how the layout names their broadcasts diff --git a/rs/moq-scene/src/collection.rs b/rs/moq-scene/src/collection.rs index 46b6c023..56784c7d 100644 --- a/rs/moq-scene/src/collection.rs +++ b/rs/moq-scene/src/collection.rs @@ -469,6 +469,13 @@ impl Collection { "`{name}` draws scene `{scene}`, which is not a scene" ))); } + if let Source::Layer { broadcast, .. } = source + && !matches!(self.sources.get(broadcast), Some(Source::Broadcast { .. })) + { + return Err(Error::Invalid(format!( + "`{name}` plays on `{broadcast}`, which is not a broadcast source" + ))); + } let Source::Clone { source } = source else { continue }; match self.sources.get(source) { Some(Source::Clone { .. }) => { @@ -1848,6 +1855,19 @@ pub enum Source { /// same frame. A scene drawing it cannot go on program, and the top layer /// cannot draw it, since either would draw the program into itself. Program, + /// A source a studio broadcast publishes independently, played from the + /// broadcast the [`Source::Broadcast`] called `broadcast` plays, on that + /// source's playout clock, so it lands with the program as the + /// [layer layout](crate::layout) places it. Never in a scene file: only + /// a player of the layout makes one. + #[serde(skip)] + Layer { + /// The broadcast source whose broadcast and clock it plays on. + broadcast: String, + /// Its entry under the broadcast catalog's + /// [`SECTION`](crate::publish::SECTION) `sources`. + source: String, + }, } impl Source { @@ -1904,6 +1924,7 @@ impl Source { | Self::Broadcast { .. } | Self::Browser { .. } | Self::Replay + | Self::Layer { .. } ) } diff --git a/rs/moq-scene/src/compositor.rs b/rs/moq-scene/src/compositor.rs index 755c5ad3..8d2a478e 100644 --- a/rs/moq-scene/src/compositor.rs +++ b/rs/moq-scene/src/compositor.rs @@ -655,6 +655,12 @@ impl Compositor { Ok(()) } + /// Play the program scene's keyframes as if it went on program at + /// `start`, such as one a viewer joined after it went on. + pub fn set_program_start(&mut self, start: Instant) { + self.program_start = Some(start); + } + /// Refuse to take a preview scene to program that cannot go there. fn can_take(&self, collection: &Collection) -> Result<(), Error> { match self.preview_scene.as_deref() { diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index 86fcc0be..cf4c07d5 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -28,7 +28,7 @@ pub(crate) const VIDEO_DELAY: Duration = Duration::from_millis(100); /// represent, and so how long a stalled group is waited on before it is /// skipped, as under `moq play --delay auto`. A shorter wait would cap the /// estimate at it, since a skipped group is never measured. -pub(crate) const MAX_DELAY: Duration = Duration::from_secs(2); +pub const MAX_DELAY: Duration = Duration::from_secs(2); /// What a bus channel holds between a write and the mix that plays it, while /// it is kept fed. diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 574b725a..b89ff048 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -31,7 +31,8 @@ use crate::browser::Page; use crate::collection::{Collection, Source}; use crate::lut::Cube; use crate::media::Clip; -use crate::playout::{BUS_LATENCY, MAX_DELAY, Place, Speaker, VIDEO_DELAY, Writer}; +pub use crate::playout::MAX_DELAY; +use crate::playout::{BUS_LATENCY, Place, Speaker, VIDEO_DELAY, Writer}; use crate::publish::{Claim, Listing, Publish, Still}; use crate::replay::{MIN_SPEED, Take, Target}; @@ -321,6 +322,20 @@ impl Sources { } keep }); + // A layer plays on its broadcast source's clock, so it stops with it + // and starts again on the replacement's. + let orphaned: Vec = self + .feeds + .iter() + .filter( + |(_, feed)| matches!(&feed.source, Source::Layer { broadcast, .. } if !self.feeds.contains_key(broadcast)), + ) + .map(|(name, _)| name.clone()) + .collect(); + for name in orphaned { + let feed = self.feeds.remove(&name).expect("listed above"); + leave(&self.mixes, &feed); + } // A mix ends once its last handle drops, which may be a reader's. self.mixes.retain(|name, _| collection.audio_scene(name).is_some()); for scene in &collection.audio_scenes { @@ -337,10 +352,25 @@ impl Sources { .collect(); let mut drivers = Vec::new(); - for (name, source) in collection.sources.iter().filter(|(_, source)| source.opens()) { + // Layers last, once the broadcast sources whose clocks they play on run. + let (layers, opens): (Vec<_>, Vec<_>) = collection + .sources + .iter() + .filter(|(_, source)| source.opens()) + .partition(|(_, source)| matches!(source, Source::Layer { .. })); + for (name, source) in opens.into_iter().chain(layers) { if !self.feeds.contains_key(name) { let held = collection.transition.stinger() == Some(name.as_str()); - let (feed, driver) = open(name, source, &self.context, held); + let follows = match source { + Source::Layer { broadcast, .. } => match collection.sources.get(broadcast) { + Some(Source::Broadcast { path }) => { + self.feeds[broadcast].clock.clone().map(|clock| (path.clone(), clock)) + } + _ => None, + }, + _ => None, + }; + let (feed, driver) = open(name, source, &self.context, held, follows); self.feeds.insert(name.clone(), feed); drivers.push(driver); } @@ -449,7 +479,7 @@ pub struct Feed { runs: Runs, /// Set once a media source's file turns out to hold no sound. silent: Arc, - /// A broadcast source's playout clock. + /// A broadcast source's playout clock, or the one a layer plays on. clock: Option, } @@ -612,7 +642,7 @@ impl Feed { /// picture alone. pub fn stalled(&self, now: Instant) -> Option { let stall = match self.source { - Source::Camera { .. } | Source::Broadcast { .. } => Stall::Picture, + Source::Camera { .. } | Source::Broadcast { .. } | Source::Layer { .. } => Stall::Picture, Source::Microphone { .. } | Source::SystemAudio => Stall::Sound, _ => return None, }; @@ -632,6 +662,13 @@ impl Feed { (self.status() == Status::Live).then(|| clock.target().unwrap_or(VIDEO_DELAY)) } + /// When a broadcast source plays what its broadcast stamps, or a layer + /// what its broadcast source's does, so what else that broadcast carries + /// can play on the same clock. `None` for any other source. + pub fn clock(&self) -> Option<&timeline::Consumer> { + self.clock.as_ref() + } + /// The source's gain, mute, delay, and meter in every mix, if it has sound. /// A media file without any has none once it opens. pub fn strip(&self) -> Option<&Strip> { @@ -678,6 +715,9 @@ pub struct Driver { runs: Runs, silent: Arc, clock: Option, + /// A layer's broadcast path, and the clock of the broadcast source it + /// plays on. + follows: Option<(String, timeline::Consumer)>, } /// Where a source's own track goes, its claim on the local catalog's listing @@ -795,6 +835,7 @@ impl Driver { self.media(&self.context.dir.join(path), *looping, cues, listing).await } Source::Broadcast { path } => self.broadcast(path).await, + Source::Layer { source, .. } => self.layer(source).await, Source::Browser { url, width, height } => self.browser(url, *width, *height).await, Source::Replay => { let plays = self.plays.clone().expect("a replay source has plays"); @@ -845,6 +886,7 @@ impl Driver { | Source::Image { .. } | Source::Media { .. } | Source::Broadcast { .. } + | Source::Layer { .. } | Source::Browser { .. } | Source::Overlay { .. } | Source::Clone { .. } @@ -1023,23 +1065,52 @@ impl Driver { /// Follow the broadcast at `path` for as long as the feed lives, playing /// each instance of it that comes online. async fn broadcast(&self, path: &str) -> Result<(), Error> { + let mut clock = self.clock.clone().expect("a broadcast source has a clock"); + self.follow(path, |broadcast| { + *self.played.lock().unwrap() = Played { + broadcast: Some(broadcast.clone()), + ..Played::default() + }; + // Each instance's timestamps are its own. + clock = clock.clone().reset(VIDEO_DELAY); + Box::pin(self.play(broadcast, path, clock.clone())) + }) + .await + } + + /// Play the source called `source` that the broadcast this layer follows + /// publishes independently, from each instance of it, on the clock of the + /// broadcast source playing it. + async fn layer(&self, source: &str) -> Result<(), Error> { + let Some((path, schedule)) = &self.follows else { + return Err(Error::Invalid(format!( + "layer `{}` follows no broadcast source", + self.name + ))); + }; + self.follow(path, |broadcast| { + Box::pin(self.play_layer(broadcast, path, source, schedule)) + }) + .await + } + + /// Play each instance of the broadcast at `path` with `start` as it comes + /// online, until the next replaces it or it goes offline. + async fn follow<'a>( + &'a self, + path: &str, + mut start: impl FnMut(moq_net::broadcast::Consumer) -> Playing<'a>, + ) -> Result<(), Error> { let mut instances = self.context.origin.follow(path, None)?; set(&self.status, Status::Offline); - let mut clock = self.clock.clone().expect("a broadcast source has a clock"); let mut playing: Option> = None; loop { tokio::select! { event = instances.available() => match event { Ok(Some(broadcast)) => { set(&self.status, Status::Live); - *self.played.lock().unwrap() = Played { - broadcast: Some(broadcast.clone()), - ..Played::default() - }; - // Each instance's timestamps are its own. - clock = clock.reset(VIDEO_DELAY); - playing = Some(Box::pin(self.play(broadcast, path, clock.clone()))); + playing = Some(start(broadcast)); } Ok(None) => { playing = None; @@ -1059,6 +1130,175 @@ impl Driver { } } + /// Decode one instance of a studio broadcast's source called `source`: + /// the first video and audio renditions its entry under the catalog's + /// [`SECTION`](crate::publish::SECTION) `sources` offers, switching as + /// the catalog changes them, until the catalog ends. Both play on + /// `schedule`, the program's clock, without moving it. + async fn play_layer( + &self, + broadcast: moq_net::broadcast::Consumer, + path: &str, + source: &str, + schedule: &timeline::Consumer, + ) -> Result<(), Error> { + let format = moq_mux::catalog::CatalogFormat::detect(path).unwrap_or_default(); + let mut catalogs = moq_mux::catalog::Consumer::::new(&broadcast, format).await?; + let mut video: Option = None; + let mut audio: Option = None; + let mut video_playing: Playing<'_> = Box::pin(std::future::pending()); + let mut audio_playing: Playing<'_> = Box::pin(std::future::pending()); + loop { + tokio::select! { + catalog = catalogs.next() => { + let Some(catalog) = catalog? else { + return Ok(()); + }; + let listed = catalog.ext.studio.sources.get(source); + let next_video = listed.map(|listed| pick(listed.video.enabled())).transpose()?.flatten(); + let next_audio = listed.map(|listed| pick(listed.audio.enabled())).transpose()?.flatten(); + let track = next_video.as_ref().map(|(id, config)| Reference::track(config, id).to_string()); + if track != video { + video = track; + video_playing = match next_video { + Some((id, config)) => Box::pin(self.follow_video(&broadcast, id, config, schedule)), + None => { + self.picture.lock().unwrap().clear(); + Box::pin(std::future::pending()) + } + }; + } + let track = next_audio.as_ref().map(|(id, config)| Reference::track(config, id).to_string()); + if track != audio { + audio = track; + audio_playing = match next_audio { + Some((id, config)) => Box::pin(self.follow_audio(&broadcast, id, config, schedule)), + None => Box::pin(std::future::pending()), + }; + } + } + result = &mut video_playing => { + result?; + video_playing = Box::pin(std::future::pending()); + } + result = &mut audio_playing => { + result?; + audio_playing = Box::pin(std::future::pending()); + } + } + } + } + + /// Show each picture once `schedule`, another track's clock, says it is + /// due, the newest due one when several are, as + /// [`play_video`](Self::play_video) does, except that a picture waits only + /// for its moment, not for its sound. + async fn follow_video( + &self, + broadcast: &moq_net::broadcast::Consumer, + name: String, + config: hang::catalog::VideoConfig, + schedule: &timeline::Consumer, + ) -> Result<(), Error> { + let mut options = moq_video::decode::Options::new(); + options.start = moq_video::decode::Start::Latest; + options.decoder.output = self.context.output; + options.delay = moq_video::decode::Delay { + min: Some(VIDEO_DELAY), + max: Some(MAX_DELAY), + }; + let mut consumer = moq_video::decode::Consumer::new(broadcast, &config, name, options).await?; + let mut queue = VecDeque::::new(); + let mut ended = false; + loop { + let now = kio::clock::now(); + if let Some(index) = schedule.present(queue.iter().map(|frame| frame.timestamp), now) { + let frame = queue.drain(..=index).next_back().expect("an index into the queue"); + self.show(Picture::Frame(frame)); + } + if ended && queue.is_empty() { + return Ok(()); + } + let due = queue + .front() + .and_then(|frame| schedule.due(frame.timestamp)) + .filter(|due| *due > now); + tokio::select! { + read = consumer.read(), if !ended && queue.len() < QUEUED => match read? { + Some(moq_video::decode::Read::Frame(frame)) => queue.push_back(frame), + Some(moq_video::decode::Read::Gap { .. } | moq_video::decode::Read::End { .. }) => {} + None => ended = true, + }, + () = kio::clock::sleep_until(due.unwrap_or(now)), if due.is_some() => {} + } + } + } + + /// Play a track's sound into the bus when `schedule`, another track's + /// clock, says each frame sounds, dropping a frame that arrives too late + /// for its moment, or before the clock has one. + async fn follow_audio( + &self, + broadcast: &moq_net::broadcast::Consumer, + name: String, + config: hang::catalog::AudioConfig, + schedule: &timeline::Consumer, + ) -> Result<(), Error> { + let Some(strip) = self.strip.clone() else { + return Ok(()); + }; + let bus = self.context.bus.config(); + let mut options = moq_audio::decode::Options::new(); + options.start = moq_audio::decode::Start::Latest; + options.delay = moq_audio::decode::Delay { + min: None, + max: Some(MAX_DELAY), + }; + options.output.format = moq_audio::Format::F32; + options.output.sample_rate = Some(bus.sample_rate); + options.output.layout = Some(bus.layout); + let mut consumer = moq_audio::decode::Consumer::new(broadcast, &config, name, options).await?; + let channel = self.channel(strip)?; + let mut writer = Writer::new(channel, bus.sample_rate, bus.layout.channels()); + + // Each frame waits here until the bus's latency before it sounds. + let mut queue = VecDeque::<(Instant, moq_audio::Frame)>::new(); + let mut ended = false; + loop { + let now = kio::clock::now(); + while let Some((sounds, _)) = queue.front() + && *sounds - BUS_LATENCY <= now + { + let (sounds, frame) = queue.pop_front().expect("peeked"); + writer.write(sounds, &frame.data, now)?; + } + if ended && queue.is_empty() { + return Ok(()); + } + let write = queue.front().map(|(sounds, _)| *sounds - BUS_LATENCY); + tokio::select! { + read = consumer.read(), if !ended => match read? { + Some(moq_audio::decode::Read::Frame(frame)) => { + // A write half the bus's latency late finds nothing queued + // ahead of it, and one due past the longest target would + // hold whatever a peer stamps. + let now = kio::clock::now(); + let placed = schedule.due(frame.timestamp).filter(|sounds| { + *sounds >= now + BUS_LATENCY / 2 && *sounds <= now + MAX_DELAY + BUS_LATENCY + }); + if let Some(sounds) = placed { + queue.push_back((sounds, frame)); + } + } + // A hole or an encoder restart shows in the stamps that follow. + Some(_) => {} + None => ended = true, + }, + () = kio::clock::sleep_until(write.unwrap_or(now)), if write.is_some() => {} + } + } + } + /// Decode one instance of a broadcast: the first video and audio renditions /// each catalog offers, switching when a catalog update changes them, until /// the catalog ends. Both play on `clock`. @@ -1967,8 +2207,15 @@ pub(crate) fn pick<'a, C: Reference + Clone + 'a>( } /// Start the feed for `source`, and the driver that fills it, `held` from -/// playing a media file until the first [`Feed::play`]. -fn open(name: &str, source: &Source, context: &Context, held: bool) -> (Feed, Driver) { +/// playing a media file until the first [`Feed::play`]. A layer `follows` its +/// broadcast's path and the clock of the broadcast source playing it. +fn open( + name: &str, + source: &Source, + context: &Context, + held: bool, + follows: Option<(String, timeline::Consumer)>, +) -> (Feed, Driver) { let picture = Mailbox::default(); let delivered = Delivered::default(); let strip = source.has_audio().then(Strip::default); @@ -2021,7 +2268,10 @@ fn open(name: &str, source: &Source, context: &Context, held: bool) -> (Feed, Dr cues, runs: runs.clone(), silent: silent.clone(), - clock: clock.as_ref().map(timeline::Producer::consume), + clock: match &follows { + Some((_, schedule)) => Some(schedule.clone()), + None => clock.as_ref().map(timeline::Producer::consume), + }, }; let driver = Driver { name: name.to_string(), @@ -2039,6 +2289,7 @@ fn open(name: &str, source: &Source, context: &Context, held: bool) -> (Feed, Dr runs, silent, clock, + follows, }; (feed, driver) } @@ -2049,11 +2300,12 @@ impl Source { /// system audio, or media file as captured or decoded, before its gain and /// mute. /// - /// A broadcast is already published by its own publisher, and a clone shows - /// its source's track, its filters carried by the layout rather than encoded - /// again. Web pages and overlays are seen only in the program, since H.264 - /// would encode their transparency over black, a LUT is looked up, never - /// shown, and a scene is drawn from other sources. + /// A broadcast, or a layer of one, is already published by its own + /// publisher, and a clone shows its source's track, its filters carried by + /// the layout rather than encoded again. Web pages and overlays are seen + /// only in the program, since H.264 would encode their transparency over + /// black, a LUT is looked up, never shown, and a scene is drawn from other + /// sources. pub fn publishes(&self) -> bool { match self { Source::Camera { .. } @@ -2066,6 +2318,7 @@ impl Source { | Source::Media { .. } | Source::Replay => true, Source::Broadcast { .. } + | Source::Layer { .. } | Source::Browser { .. } | Source::Overlay { .. } | Source::Clone { .. } @@ -2335,7 +2588,7 @@ mod tests { let (context, _origin) = context(dir.path()); let start = kio::clock::now(); let opened = |source: Source| { - let (feed, driver) = open("source", &source, &context, false); + let (feed, driver) = open("source", &source, &context, false, None); set(&driver.status, Status::Live); (feed, driver) }; @@ -2628,9 +2881,9 @@ mod tests { } /// A video track on its own timestamp base. - async fn rendition( + async fn rendition( broadcast: &moq_net::broadcast::Producer, - timebase: moq_mux::catalog::Timebase, + timebase: moq_mux::catalog::Timebase, ) -> (moq_video::encode::Producer, moq_video::encode::Encoder) { let mut config = moq_video::encode::Config::new(64, 64, moq_video::FrameRate::new(30, 1).unwrap()); config.kind = moq_video::encode::Kind::Software; @@ -2711,6 +2964,104 @@ mod tests { .await; } + /// A layer shows each picture when its broadcast source's clock says, as + /// the program's picture stamped alike shows, not its own delay after it + /// arrives. + #[tokio::test(start_paused = true)] + async fn a_layer_plays_on_its_broadcasts_clock() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, origin) = context(dir.path()); + let publication = origin + .route("room/guest", moq_net::origin::RouteOptions::hidden()) + .unwrap(); + let mut broadcast = publication.broadcast().clone(); + let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); + let mut catalog = crate::publish::Catalog::new(&mut broadcast, config).unwrap(); + let (mut program, mut encoder) = rendition(&broadcast, catalog.timebase()).await; + // The camera's track is in the program's broadcast, listed under its + // name in the catalog's own section rather than as the program's. + let mut local = moq_net::broadcast::Info::new().produce(); + let listing = moq_mux::catalog::Producer::new(&mut local, Default::default()).unwrap(); + let (mut cam, mut cam_encoder) = rendition(&broadcast, listing.timebase()).await; + publication.announce().unwrap(); + + let mut collection = collection(); + collection.sources.insert( + "cam".into(), + Source::Layer { + broadcast: "guest".into(), + source: "cam".into(), + }, + ); + collection.validate().unwrap(); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let layer = sources.get("cam").unwrap(); + let clock = sources.get("guest").unwrap().clock.clone().unwrap(); + + until_time(kio::clock::now()).await; + let arrived = kio::clock::now(); + picture(&mut program, &mut encoder, 1_000); + let stamp = moq_net::Timestamp::from_millis(1_000).unwrap(); + let due = until(|| clock.due(stamp)).await; + assert_eq!(due, arrived + VIDEO_DELAY); + + // Half the delay later, so on a clock of its own it would show late. + until_time(arrived + VIDEO_DELAY / 2).await; + picture(&mut cam, &mut cam_encoder, 1_000); + let listed = until(|| Some(listing.snapshot()).filter(|listed| !listed.video.renditions.is_empty())).await; + catalog + .mutate(|root| { + root.ext.studio.sources.insert("cam".into(), listed); + }) + .unwrap(); + until_time(due - Duration::from_millis(1)).await; + assert!(layer.next_picture().is_none(), "shown before its moment"); + until_time(due).await; + assert_eq!(shown(layer).await, Duration::from(stamp)); + assert_eq!(layer.delay(), Some(VIDEO_DELAY), "the program's"); + }) + .await; + } + + /// A layer stops with its broadcast source, and starts again on its + /// replacement's clock. + #[tokio::test] + async fn a_layer_follows_its_broadcast_sources_replacement() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, _origin) = context(dir.path()); + let mut collection = collection(); + collection.sources.insert( + "cam".into(), + Source::Layer { + broadcast: "guest".into(), + source: "cam".into(), + }, + ); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection); + assert_eq!(drivers.len(), 3); + start(&sources, drivers).await; + collection.sources.insert( + "guest".into(), + Source::Broadcast { + path: "room/other".into(), + }, + ); + let mut names: Vec = sources + .sync(&collection) + .iter() + .map(|driver| driver.name().to_string()) + .collect(); + names.sort(); + assert_eq!(names, ["cam", "guest"]); + }) + .await; + } + /// A video track replaced by one stamped earlier starts its own delay over, /// rather than showing every picture on arrival against the old track's. #[tokio::test(start_paused = true)] diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 15dba650..50cc0334 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -835,7 +835,7 @@ pub const AUDIO_BITRATE: moq_net::bandwidth::Rate = moq_net::bandwidth::Rate::fr /// How often the audio thread advances the bus. Short enough that a block is never /// close to the bus's buffering, long enough to cost nothing. -const AUDIO_PERIOD: Duration = Duration::from_millis(10); +pub(crate) const AUDIO_PERIOD: Duration = Duration::from_millis(10); /// The longest [`Engine::close`] waits for the sources to stop and the relays /// to drain, so a source or relay that never answers cannot hold up quitting. @@ -3034,7 +3034,7 @@ fn browser_profile() -> anyhow::Result { /// Run a source on its own thread until its feed is dropped, or fail the /// source when its thread cannot start. Returns what closes as the source /// ends, having finished its own track. -fn spawn(driver: Driver) -> tokio::sync::oneshot::Receiver<()> { +pub(crate) fn spawn(driver: Driver) -> tokio::sync::oneshot::Receiver<()> { // Handed over once the thread exists, since a failed spawn drops what it // was given and the driver must stay here to report why. let (handoff, received) = mpsc::sync_channel::(1); @@ -3059,7 +3059,10 @@ fn spawn(driver: Driver) -> tokio::sync::oneshot::Receiver<()> { /// Start a thread called `name` that runs `task` with its own clock and IO, /// returning why if either cannot start. -fn clocked(name: String, task: impl AsyncFnOnce() + Send + 'static) -> anyhow::Result> { +pub(crate) fn clocked( + name: String, + task: impl AsyncFnOnce() + Send + 'static, +) -> anyhow::Result> { // The thread starts before its runtime is built and handed over, since a // runtime dropped in the async caller, as a refused spawn would, panics. let (handoff, received) = mpsc::sync_channel::(1); diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index a1b8f0ce..894f5817 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -32,6 +32,7 @@ mod snap; mod ui; mod uplink; mod vcam; +mod watch; #[derive(usage_rs::Cli, Clone)] #[usage(unknown_flags = "error", args_override_self = false)] @@ -41,7 +42,15 @@ pub struct Config { /// The scene collection to open, saved as it changes. If it does not exist, /// guided setup proposes one and creates it. #[usage(value_hint = usage_rs::ValueHint::FilePath, extensions("toml"))] - pub scenes: PathBuf, + pub scenes: Option, + + /// Play a studio broadcast instead, as a viewer sees it: the program with + /// each source published independently over it. The URL's path names the + /// broadcast, as `moq play`'s does, such as + /// `https://relay.example/anon/studio`. A prototype, until `moq play` plays + /// the layer layout. + #[usage(long)] + pub watch: Option, /// The broadcast path the program is published at. #[usage(long, default = "studio")] @@ -85,6 +94,30 @@ pub struct Config { } impl Config { + /// The scene file to open, refusing `--watch` beside the studio's own flags. + fn scenes(&self) -> anyhow::Result> { + if self.watch.is_none() { + let scenes = self + .scenes + .as_deref() + .context("name a scene file to open, or --watch a broadcast")?; + return Ok(Some(scenes)); + } + anyhow::ensure!( + self.scenes.is_none(), + "--watch plays a broadcast, so it opens no scene file" + ); + anyhow::ensure!( + self.client.url.is_none() && self.failover.is_none(), + "--watch names the relay in its URL, so it takes no --connect or --failover" + ); + anyhow::ensure!( + self.guests.room.is_none(), + "--watch publishes nothing, so it has no --room" + ); + Ok(None) + } + /// Refuse a config parsed from released spellings, naming what replaced each. fn check_deprecated(&self) -> anyhow::Result<()> { let mut deprecated = self.client.deprecated(); @@ -149,6 +182,12 @@ fn main() -> anyhow::Result { config.log.init()?; config.check_deprecated()?; moq_tokio::cli::check_env(&[Config::spec()], &[])?; + let Some(scenes) = config.scenes()? else { + let url = config.watch.clone().expect("checked by scenes"); + watch::run(config.client.clone(), config.quic.clone(), url)?; + return Ok(ExitCode::SUCCESS); + }; + let scenes = scenes.to_path_buf(); let runtime = tokio::runtime::Runtime::new()?; // The QUIC client binds its socket on the current runtime. @@ -157,8 +196,7 @@ fn main() -> anyhow::Result { let gpu = pollster::block_on(engine::Gpu::new())?; let recordings = match &config.recordings { Some(dir) => dir.clone(), - None => config - .scenes + None => scenes .parent() .filter(|dir| !dir.as_os_str().is_empty()) .unwrap_or(Path::new(".")) @@ -177,20 +215,19 @@ fn main() -> anyhow::Result { recordings, room, }; - let exists = config - .scenes + let exists = scenes .try_exists() - .with_context(|| format!("reading {}", config.scenes.display()))?; + .with_context(|| format!("reading {}", scenes.display()))?; let app = match exists { true => { - let (autosave, loaded) = autosave::Autosave::open(config.scenes.clone())?; + let (autosave, loaded) = autosave::Autosave::open(scenes.clone())?; App::Starting(runtime.block_on(engine::Engine::start(launch, autosave, loaded))?) } // Nothing is saved until setup starts the studio, so quitting before then // leaves no file and the next launch runs setup again. false => App::Setup(Box::new(setup::Setup::new( launch, - config.scenes.clone(), + scenes.clone(), runtime.handle().clone(), )?)), }; @@ -427,7 +464,7 @@ mod tests { ]) .expect("a scene file and a relay should parse"); config.check_deprecated().unwrap(); - assert_eq!(config.scenes, PathBuf::from("show.toml")); + assert_eq!(config.scenes().unwrap(), Some(Path::new("show.toml"))); assert_eq!(config.broadcast, "studio"); assert!(!config.on_demand, "the program encodes whatever the demand by default"); assert!(config.guests.room.is_none(), "guests are opt-in"); @@ -463,4 +500,24 @@ mod tests { assert!(check(refused).is_err(), "{refused:?} was accepted"); } } + + /// `--watch` opens no scene file and names its own relay, and the studio + /// needs a scene file. + #[test] + fn watch_takes_no_scene_file_or_relay() { + let parse = |args: &[&str]| { + let args: Vec<_> = args.iter().map(std::ffi::OsStr::new).collect(); + Config::parse_from(&args).unwrap() + }; + let watch = parse(&["--watch", "https://relay.example/anon/studio"]); + assert_eq!(watch.scenes().unwrap(), None); + for refused in [ + &["show.toml", "--watch", "https://a/b"][..], + &["--watch", "https://a/b", "--connect", "https://a/"], + &["--watch", "https://a/b", "--room", "guests"], + &[], + ] { + assert!(parse(refused).scenes().is_err(), "{refused:?} was accepted"); + } + } } diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 004cd8f2..c4228f87 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -314,13 +314,13 @@ impl Studio { } /// A compositor texture as egui shows it. -struct Registered { - id: egui::TextureId, +pub(crate) struct Registered { + pub(crate) id: egui::TextureId, texture: wgpu::Texture, } impl Registered { - fn new(render: &eframe::egui_wgpu::RenderState, texture: &wgpu::Texture) -> Self { + pub(crate) fn new(render: &eframe::egui_wgpu::RenderState, texture: &wgpu::Texture) -> Self { let view = texture.create_view(&Default::default()); let id = render .renderer @@ -333,7 +333,7 @@ impl Registered { } /// Show `texture` from now on, if a resize replaced the one shown. - fn follow(&mut self, render: &eframe::egui_wgpu::RenderState, texture: &wgpu::Texture) { + pub(crate) fn follow(&mut self, render: &eframe::egui_wgpu::RenderState, texture: &wgpu::Texture) { if self.texture == *texture { return; } @@ -4508,6 +4508,7 @@ fn describe(source: &Source) -> String { Source::Replay => "replay".into(), Source::Lut { path } => format!("LUT: {}", path.display()), Source::Scene { scene } => format!("scene: {scene}"), + Source::Layer { broadcast, source } => format!("{source}, published in {broadcast}"), Source::Program => "program".into(), } } diff --git a/rs/moq-studio/src/watch.rs b/rs/moq-studio/src/watch.rs new file mode 100644 index 00000000..08f74a8a --- /dev/null +++ b/rs/moq-studio/src/watch.rs @@ -0,0 +1,626 @@ +//! `--watch`: play a studio broadcast as a viewer would, the program with each +//! source the [layer layout](moq_scene::layout) names composited over it and +//! mixed with it, on the program's media clock. +//! +//! A prototype for settling the layout's format before hang specifies one, +//! built on moq-scene: a collection the layout drives, rather than a scene +//! file, whose one scene draws the program, a [`Source::Broadcast`], under +//! each picture's [`Source::Layer`]. Each layer plays its own tracks on the +//! program's playout clock, each layout takes effect when the program reaches +//! the media time it is stamped with, and its pictures' keyframes play from +//! its `start` on that clock. A layout of another version is refused loudly, +//! and the program plays alone. + +use std::collections::{BTreeSet, VecDeque}; +use std::time::Duration; + +use anyhow::Context; +use kio::clock::Instant; +use moq_scene::layout::{self, Layout}; +use moq_scene::publish::Ext; +use moq_scene::source::{self, MAX_DELAY, Sources, Status}; +use moq_scene::{AudioScene, Canvas, Collection, Compositor, Item, Rect, Scene, Source}; + +use tokio::sync::watch; + +use crate::engine::{self, Gpu}; +use crate::mix; + +/// The program's source in the collection. +const PROGRAM: &str = "program"; + +/// What each layer's source name starts with, so none can take the +/// program's. +const LAYER: &str = "layer/"; + +/// The scene on program. +const SCENE: &str = "watch"; + +/// The canvas until the program's catalog or a layout gives its size. +const CANVAS: Canvas = Canvas { + width: 1280, + height: 720, + framerate: 30, +}; + +/// Play the broadcast at `url` until the window closes. Its path names the +/// broadcast, as `moq play`'s does, and `client` dials it. +pub fn run(client: moq_tokio::connect::Config, quic: moq_tokio::quic::Config, url: url::Url) -> anyhow::Result<()> { + let runtime = tokio::runtime::Runtime::new()?; + let _runtime = runtime.enter(); + let client = client.init(quic)?; + let origin = moq_tokio::origin::spawn(); + tracing::info!(url = %moq_tokio::RedactedUrl::new(&url), "watching"); + let connection = client.with_subscriber(origin.clone()).connect(url.clone()); + + let gpu = pollster::block_on(Gpu::new())?; + let (bus, mixer) = moq_audio::bus::Bus::new(Default::default())?; + let config = bus.config().clone(); + let start = Instant::now(); + engine::clocked("audio".into(), async move || advance(start, config, mixer).await)?; + let monitor = match mix::Monitor::start(runtime.handle(), AudioScene::DEFAULT, None) { + Ok(monitor) => Some(monitor), + Err(err) => { + tracing::warn!(error = format!("{err:#}"), "playing without sound"); + None + } + }; + + let sources = Sources::new(source::Context { + bus, + origin: origin.consume(), + dir: std::env::current_dir()?, + output: moq_video::render::output(&gpu.device), + publish: None, + browser: None, + }); + let collection = collection(None, CANVAS)?; + let mut compositor = Compositor::new(&gpu.device, &gpu.queue, collection.canvas)?; + compositor.set_previewing(false); + compositor.set_preview(Some(SCENE.into())); + compositor.cut(&collection, &sources)?; + + let (read, _) = watch::channel(Read::default()); + runtime.spawn(follow(origin.consume(), read.clone())); + + let watch = Watch { + url: moq_tokio::RedactedUrl::new(&url).to_string(), + read, + collection, + synced: false, + layout: None, + start: None, + size: None, + sources, + compositor, + monitor, + shown: None, + }; + let options = eframe::NativeOptions { + viewport: egui::ViewportBuilder::default() + .with_title("MoQ Studio") + .with_app_id("moq-studio") + .with_inner_size([1280.0, 720.0]), + wgpu_options: eframe::egui_wgpu::WgpuConfiguration { + // The compositor's program reaches the screen without leaving the GPU. + wgpu_setup: eframe::egui_wgpu::WgpuSetup::Existing(eframe::egui_wgpu::WgpuSetupExisting { + instance: gpu.instance, + adapter: gpu.adapter, + device: gpu.device, + queue: gpu.queue, + }), + ..Default::default() + }, + ..Default::default() + }; + let ran = eframe::run_native( + "moq-studio-watch", + options, + Box::new(move |cc| { + let render = cc + .wgpu_render_state + .clone() + .context("the window is not drawn with wgpu")?; + Ok(Box::new(Window { render, watch })) + }), + ); + runtime.block_on(async { + let _ = connection.close().await; + }); + ran.map_err(|err| anyhow::anyhow!("the window failed: {err}")) +} + +/// Advance the bus on the wall clock from `start`, so each mix plays. +async fn advance(start: Instant, config: moq_audio::bus::Config, mut mixer: moq_audio::bus::Driver) { + let rate = u128::from(config.sample_rate); + let mut clock = kio::clock::interval_at(start, engine::AUDIO_PERIOD); + clock.set_missed_tick_behavior(kio::clock::MissedTickBehavior::Skip); + let mut written: u128 = 0; + loop { + let tick = clock.tick().await; + let due = tick.saturating_duration_since(start).as_nanos() * rate / 1_000_000_000; + let frames = (due - written) as usize; + if frames > 0 { + mixer.advance(frames); + written = due; + } + } +} + +/// What the broadcast's catalog and layout track say, for the window to apply. +#[derive(Default)] +struct Read { + /// The program's picture size, from its catalog. + size: Option<(u32, u32)>, + /// Each layout read, or `None` for the program alone, and the media time + /// it takes effect at, or `None` for at once, oldest first. + pending: VecDeque<(Option, Option)>, + /// Why the layout was refused, so the program plays alone. + refused: Option, +} + +impl Read { + /// Queue `layout` to take effect at `at`. The program plays at most + /// [`MAX_DELAY`] behind what arrives, so of the layouts stamped that far + /// before it, all due by now, only the newest is kept. + fn push(&mut self, layout: Option, at: Option) { + let due = |stamp: Option| match (stamp, at) { + (None, _) => true, + (Some(_), None) => true, + (Some(stamp), Some(at)) => Duration::from(stamp) + MAX_DELAY <= Duration::from(at), + }; + let passed = self.pending.iter().take_while(|(_, stamp)| due(*stamp)).count(); + self.pending.drain(..passed.saturating_sub(1)); + self.pending.push_back((layout, at)); + } +} + +/// Read each instance of the root broadcast as it comes online: the program's +/// size from its catalog, and each layout from the layout track while the +/// catalog lists it. +async fn follow(origin: moq_net::origin::Consumer, read: watch::Sender) { + let mut instances = match origin.follow("", None) { + Ok(instances) => instances, + Err(err) => { + tracing::error!(%err, "cannot follow the broadcast"); + return; + } + }; + let mut playing = None; + loop { + tokio::select! { + event = instances.available() => match event { + Ok(Some(broadcast)) => playing = Some(Box::pin(instance(broadcast, read.clone()))), + Ok(None) => playing = None, + Err(err) => { + tracing::error!(%err, "the broadcast's announcements failed"); + return; + } + }, + result = async { playing.as_mut().expect("guarded").await }, if playing.is_some() => { + playing = None; + if let Err(err) = result { + tracing::warn!(error = format!("{err:#}"), "cannot read the broadcast's layout"); + } + } + } + } +} + +/// Read one instance of the broadcast until its catalog ends. +async fn instance(broadcast: moq_net::broadcast::Consumer, read: watch::Sender) -> anyhow::Result<()> { + let mut catalogs = + moq_mux::catalog::Consumer::::new(&broadcast, moq_mux::catalog::CatalogFormat::default()).await?; + let mut layouts: Option> = None; + // A refused layout stays refused for this instance. + let mut refused = false; + let alone = || read.send_modify(|read| read.push(None, None)); + loop { + tokio::select! { + catalog = moq_mux::catalog::Stream::next(&mut catalogs) => { + let Some(catalog) = catalog? else { + alone(); + return Ok(()); + }; + let size = catalog.video.renditions.values().find_map(|config| config.coded_width.zip(config.coded_height)); + read.send_if_modified(|read| std::mem::replace(&mut read.size, size) != size); + let listed = catalog.json.tracks.contains_key(layout::TRACK); + if listed && layouts.is_none() && !refused { + let track = broadcast.track(layout::TRACK)?; + layouts = Some(moq_json::snapshot::Consumer::new(track, Default::default())); + } else if !listed && layouts.take().is_some() { + alone(); + } + } + next = async { layouts.as_mut().expect("guarded").next().await }, if layouts.is_some() => match next { + Ok(Some(timed)) => read.send_modify(|read| read.push(Some(timed.value), timed.at)), + Ok(None) => { + layouts = None; + alone(); + } + Err(err) => { + tracing::error!(%err, "refusing the layer layout; playing the program alone"); + layouts = None; + refused = true; + read.send_modify(|read| { + read.refused = Some(err.to_string()); + read.push(None, None); + }); + } + }, + } + } +} + +/// The collection `layout` drives on a canvas the program's `size` when it +/// has none: the program, and over it each picture's layer, with the program +/// and each sound's layer in the audio scene on program. +fn collection(layout: Option<&Layout>, size: Canvas) -> anyhow::Result { + let mut collection = Collection::new(); + collection.canvas = match layout { + Some(layout) => Canvas { + width: layout.canvas.width, + height: layout.canvas.height, + framerate: size.framerate, + }, + None => size, + }; + collection + .sources + .insert(PROGRAM.into(), Source::Broadcast { path: String::new() }); + let mut items = vec![Item::new(PROGRAM, Rect::fill(&collection.canvas))]; + let mut sounds = BTreeSet::from([PROGRAM.to_string()]); + let mut layer = |source: &str| { + let name = format!("{LAYER}{source}"); + collection.sources.insert( + name.clone(), + Source::Layer { + broadcast: PROGRAM.into(), + source: source.into(), + }, + ); + name + }; + let mut audio = Vec::new(); + for picture in layout.iter().flat_map(|layout| &layout.pictures) { + items.push(Item { + source: layer(&picture.item.source), + ..picture.item.clone() + }); + } + for sound in layout.iter().flat_map(|layout| &layout.sounds) { + let name = layer(&sound.source); + audio.push((name.clone(), sound.audio.clone())); + sounds.insert(name); + } + collection.audio.extend(audio); + collection.scenes = vec![Scene { + items, + ..Scene::new(SCENE) + }]; + collection.audio_scenes = vec![AudioScene { + program: true, + sounds, + ..AudioScene::new(AudioScene::DEFAULT) + }]; + collection.validate().context("cannot play the layer layout")?; + Ok(collection) +} + +/// The player: the collection the layout drives, and what plays it. +struct Watch { + /// What the window says it waits for. + url: String, + read: watch::Sender, + collection: Collection, + /// Whether the sources run what `collection` holds. + synced: bool, + /// The layout in effect, or `None` for the program alone. + layout: Option, + /// When the program scene's keyframes play from, as the layout in effect + /// set it. + start: Option, + /// The program's size, from its catalog. + size: Option<(u32, u32)>, + sources: Sources, + compositor: Compositor, + monitor: Option, + /// The program as the window shows it. + shown: Option, +} + +impl Watch { + /// Apply what was read and is now due at `now`, and draw the program. + fn render(&mut self, now: Instant) -> anyhow::Result<()> { + let schedule = self.sources.get(PROGRAM).and_then(|feed| feed.clock()).cloned(); + let mut changed = !self.synced; + self.read.send_if_modified(|read| { + if read.size != self.size { + self.size = read.size; + changed = true; + } + // Each layout once the program plays the frame it takes effect on. + while let Some((_, at)) = read.pending.front() { + let due = match at { + None => true, + Some(at) => schedule + .as_ref() + .and_then(|schedule| schedule.due(*at)) + .is_some_and(|due| due <= now), + }; + if !due { + break; + } + let (layout, _) = read.pending.pop_front().expect("peeked"); + self.layout = layout; + changed = true; + } + false + }); + + if changed { + let size = match self.size { + Some((width, height)) => Canvas { + width, + height, + ..CANVAS + }, + None => CANVAS, + }; + let collection = match collection(self.layout.as_ref(), size) { + Ok(collection) => collection, + Err(err) => { + tracing::error!(error = format!("{err:#}"), "playing the program alone"); + self.read.send_modify(|read| read.refused = Some(format!("{err:#}"))); + self.layout = None; + collection(None, size)? + } + }; + if collection.canvas != self.collection.canvas { + self.compositor.fit(&collection)?; + } + self.collection = collection; + for driver in self.sources.sync(&self.collection) { + engine::spawn(driver); + } + if let (Some(monitor), Some(mix)) = (&self.monitor, self.sources.mix(AudioScene::DEFAULT)) { + monitor.play(mix.clone()); + } + self.synced = true; + } + + // A new start is a scene taken, whose keyframes play from it. + let start = self.layout.as_ref().map(|layout| layout.start); + if start != self.start + && let Some(start) = start + && let Ok(stamp) = moq_net::Timestamp::try_from(start) + && let Some(due) = schedule.as_ref().and_then(|schedule| schedule.due(stamp)) + { + self.compositor.set_program_start(due); + self.start = Some(start); + } + + self.compositor.render(&self.collection, &self.sources, now)?; + Ok(()) + } + + /// What the window says over the program, if anything. + fn status(&self) -> Option { + if let Some(refused) = &self.read.borrow().refused { + return Some(format!("Playing the program alone: {refused}")); + } + match self.sources.get(PROGRAM).map(|feed| feed.status()) { + Some(Status::Live) => None, + Some(Status::Failed(err)) => Some(format!("{}: {err}", self.url)), + _ => Some(format!("Waiting for {}", self.url)), + } + } +} + +/// The window around the player. +struct Window { + render: eframe::egui_wgpu::RenderState, + watch: Watch, +} + +impl eframe::App for Window { + fn ui(&mut self, ui: &mut egui::Ui, _frame: &mut eframe::Frame) { + let watch = &mut self.watch; + if let Err(err) = watch.render(kio::clock::now()) { + tracing::error!(error = format!("{err:#}"), "cannot draw the program"); + } + let texture = watch.compositor.program(); + match &mut watch.shown { + Some(shown) => shown.follow(&self.render, texture), + None => watch.shown = Some(crate::ui::Registered::new(&self.render, texture)), + } + let rect = ui.max_rect(); + let painter = ui.painter(); + painter.rect_filled(rect, 0.0, egui::Color32::BLACK); + let canvas = watch.collection.canvas; + let scale = (rect.width() / canvas.width as f32).min(rect.height() / canvas.height as f32); + let fit = egui::Rect::from_center_size( + rect.center(), + egui::vec2(canvas.width as f32 * scale, canvas.height as f32 * scale), + ); + let uv = egui::Rect::from_min_max(egui::pos2(0.0, 0.0), egui::pos2(1.0, 1.0)); + let id = watch.shown.as_ref().expect("registered above").id; + painter.image(id, fit, uv, egui::Color32::WHITE); + if let Some(status) = watch.status() { + painter.text( + rect.left_top() + egui::vec2(12.0, 12.0), + egui::Align2::LEFT_TOP, + status, + egui::FontId::proportional(16.0), + egui::Color32::LIGHT_RED, + ); + } + ui.ctx().request_repaint(); + } +} + +#[cfg(test)] +mod tests { + use moq_scene::layout::{Picture, Sound, Version}; + use moq_scene::{Audio, Filter, Keyframe, Size}; + + use super::*; + + /// A camera named as the program is, placed twice, and a microphone. + fn layout(start: Duration) -> Layout { + let mut cam = Item::new( + "program", + Rect { + x: 10.0, + y: 10.0, + width: 64.0, + height: 36.0, + }, + ); + cam.filters = vec![Filter::Opacity { amount: 0.5 }]; + cam.keyframes = vec![Keyframe { + x: 100.0, + ..Keyframe::new(Duration::from_secs(1)) + }]; + let picture = Picture { + item: cam, + priority: 0, + toggle: false, + }; + Layout { + version: Version, + canvas: Size { + width: 640, + height: 360, + }, + start, + pictures: vec![picture.clone(), picture], + sounds: vec![Sound { + source: "mic".into(), + audio: Audio { + gain: 0.5, + delay: Duration::from_millis(80), + ..Default::default() + }, + priority: 0, + toggle: false, + }], + } + } + + fn stamp(secs: u64) -> moq_net::Timestamp { + moq_net::Timestamp::try_from(Duration::from_secs(secs)).unwrap() + } + + /// The layout's pictures draw over the program from layers named apart + /// from it, even one named as it is, a source placed twice shares one + /// layer, and the audio scene on program plays the program and each + /// sound at its gain and delay. + #[test] + fn a_layout_drives_the_collection() { + let layout = layout(Duration::ZERO); + let driven = collection(Some(&layout), CANVAS).unwrap(); + assert_eq!((driven.canvas.width, driven.canvas.height), (640, 360)); + let items = &driven.scene(SCENE).unwrap().items; + let sources: Vec<&str> = items.iter().map(|item| item.source.as_str()).collect(); + assert_eq!(sources, [PROGRAM, "layer/program", "layer/program"]); + assert_eq!(items[1].keyframes, layout.pictures[0].item.keyframes); + assert_eq!( + driven.sources["layer/program"], + Source::Layer { + broadcast: PROGRAM.into(), + source: "program".into() + } + ); + let heard = driven.heard(AudioScene::DEFAULT); + assert_eq!(heard, BTreeSet::from([PROGRAM, "layer/mic"])); + assert_eq!(driven.audio["layer/mic"], layout.sounds[0].audio); + + let alone = collection(None, CANVAS).unwrap(); + assert_eq!(alone.sources.len(), 1, "the program alone"); + assert_eq!(alone.canvas, CANVAS); + } + + /// Of the layouts the program must have reached by the time a new one + /// arrives, only the newest waits. + #[test] + fn only_the_newest_passed_layout_waits() { + let mut read = Read::default(); + for secs in [1, 2, 3, 4] { + read.push(None, Some(stamp(secs))); + } + read.push(None, Some(stamp(5))); + let stamps: Vec<_> = read.pending.iter().map(|(_, at)| *at).collect(); + assert_eq!(stamps, [Some(stamp(3)), Some(stamp(4)), Some(stamp(5))]); + } + + /// A layout of another version, as a newer studio would publish. + #[derive(serde::Serialize)] + struct Newer { + version: u32, + } + + /// A broadcast with a catalog, and the instance reader's view of it. + fn broadcast() -> ( + moq_net::broadcast::Producer, + moq_scene::publish::Catalog, + watch::Sender, + tokio::task::JoinHandle>, + ) { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); + let catalog = moq_scene::publish::Catalog::new(&mut broadcast, config).unwrap(); + let (read, _) = watch::channel(Read::default()); + let reader = tokio::spawn(instance(broadcast.consume(), read.clone())); + (broadcast, catalog, read, reader) + } + + /// What `read` holds once it has `count` layouts waiting, with their + /// stamps as durations, since one read back carries its track's timescale. + async fn pending(read: &watch::Sender, count: usize) -> Vec<(Option, Option)> { + let mut reads = read.subscribe(); + let read = reads.wait_for(|read| read.pending.len() >= count).await.unwrap(); + let pending = read.pending.iter().cloned(); + pending.map(|(layout, at)| (layout, at.map(Duration::from))).collect() + } + + /// The reader hands over each layout at the media time it took effect, + /// and the program alone once the catalog stops listing the track, until + /// it lists it again. + #[tokio::test] + async fn reads_each_layout() { + let (broadcast, catalog, read, reader) = broadcast(); + let mut publisher = layout::Publisher::new(broadcast.clone(), catalog.clone()); + publisher.update(layout(Duration::ZERO), stamp(1)).unwrap(); + let first = (Some(layout(Duration::ZERO)), Some(Duration::from_secs(1))); + assert_eq!(pending(&read, 1).await, [first]); + publisher.finish().unwrap(); + assert_eq!( + pending(&read, 2).await[1], + (None, None), + "the program alone once unlisted" + ); + publisher.update(layout(Duration::from_secs(2)), stamp(2)).unwrap(); + let again = (Some(layout(Duration::from_secs(2))), Some(Duration::from_secs(2))); + assert_eq!(pending(&read, 3).await[2], again, "listed again"); + reader.abort(); + } + + /// A layout of another version is refused, saying why, and the program + /// plays alone. + #[tokio::test] + async fn refuses_another_version() { + let (mut broadcast, catalog, read, reader) = broadcast(); + let info = moq_net::track::Info::default().with_timescale(moq_mux::Clock::TIMESCALE); + let track = broadcast.create_track(layout::TRACK, info).unwrap(); + let options = moq_mux::json::Options::default().with_deltas(false); + let mut newer = catalog + .json_snapshot::(track, moq_mux::json::Config::default(), options) + .unwrap(); + newer.update(&Newer { version: 2 }, stamp(1)).unwrap(); + assert_eq!(pending(&read, 1).await, [(None, None)]); + let refused = read.borrow().refused.clone().unwrap(); + assert!(refused.contains("layer layout version 2 is not supported"), "{refused}"); + reader.abort(); + } +} From 819931895a1d153b7e43bdc7c65263e86c981986 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 15:28:59 -0800 Subject: [PATCH 3/9] test(studio): drop an unused mut Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/watch.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rs/moq-studio/src/watch.rs b/rs/moq-studio/src/watch.rs index 08f74a8a..fb8e2e37 100644 --- a/rs/moq-studio/src/watch.rs +++ b/rs/moq-studio/src/watch.rs @@ -610,7 +610,7 @@ mod tests { /// plays alone. #[tokio::test] async fn refuses_another_version() { - let (mut broadcast, catalog, read, reader) = broadcast(); + let (broadcast, catalog, read, reader) = broadcast(); let info = moq_net::track::Info::default().with_timescale(moq_mux::Clock::TIMESCALE); let track = broadcast.create_track(layout::TRACK, info).unwrap(); let options = moq_mux::json::Options::default().with_deltas(false); From ca177498e0ef153bd0ca3618ac7dbd1e1c78fab0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 15:51:49 -0800 Subject: [PATCH 4/9] feat(scene): a layer shows a published still Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/source.rs | 97 ++++++++++++++++++++++++++++++++++---- 1 file changed, 87 insertions(+), 10 deletions(-) diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index b89ff048..472d558d 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -1157,12 +1157,22 @@ impl Driver { let listed = catalog.ext.studio.sources.get(source); let next_video = listed.map(|listed| pick(listed.video.enabled())).transpose()?.flatten(); let next_audio = listed.map(|listed| pick(listed.audio.enabled())).transpose()?.flatten(); - let track = next_video.as_ref().map(|(id, config)| Reference::track(config, id).to_string()); + // A source without video, such as an image or a color, is a still. + let still = listed.and_then(|listed| { + let mut stills = listed.binary.tracks.iter(); + let (id, config) = stills.find(|(_, config)| config.broadcast.is_none() && config.mime.as_deref().is_some_and(|mime| mime.starts_with("image/")))?; + Some(Reference::track(config, id).to_string()) + }); + let track = match &next_video { + Some((id, config)) => Some(Reference::track(config, id).to_string()), + None => still.clone(), + }; if track != video { video = track; - video_playing = match next_video { - Some((id, config)) => Box::pin(self.follow_video(&broadcast, id, config, schedule)), - None => { + video_playing = match (next_video, still) { + (Some((id, config)), _) => Box::pin(self.follow_video(&broadcast, id, config, schedule)), + (None, Some(still)) => Box::pin(self.follow_still(&broadcast, still)), + (None, None) => { self.picture.lock().unwrap().clear(); Box::pin(std::future::pending()) } @@ -1189,6 +1199,19 @@ impl Driver { } } + /// Show the newest picture of the track `name`, a PNG or JPEG file per + /// group, as [`Still`] publishes one, whenever it arrives: a still has no + /// moment. + async fn follow_still(&self, broadcast: &moq_net::broadcast::Consumer, name: String) -> Result<(), Error> { + let mut track = broadcast.track(&name)?.subscribe(None).await?; + while let Some(mut group) = track.recv_group().await? { + if let Some(frame) = group.read_frame().await? { + self.show(decode(&frame.payload).map_err(|err| Error::Image(format!("{name}: {err}")))?); + } + } + Ok(()) + } + /// Show each picture once `schedule`, another track's clock, says it is /// due, the newest due one when several are, as /// [`play_video`](Self::play_video) does, except that a picture waits only @@ -2367,14 +2390,17 @@ async fn capture(source: moq_audio::capture::Source) -> Result Result<(Picture, bytes::Bytes), Error> { let bytes = std::fs::read(path).map_err(|err| Error::Image(format!("{}: {err}", path.display())))?; - let image = image::load_from_memory(&bytes) - .map_err(|err| Error::Image(format!("{}: {err}", path.display())))? - .into_rgba8(); - let picture = Picture::Rgba { + let picture = decode(&bytes).map_err(|err| Error::Image(format!("{}: {err}", path.display())))?; + Ok((picture, bytes.into())) +} + +/// A PNG or JPEG file's picture, within the image crate's default limits. +fn decode(bytes: &[u8]) -> Result { + let image = image::load_from_memory(bytes)?.into_rgba8(); + Ok(Picture::Rgba { size: moq_video::Size::new(image.width(), image.height()), pixels: image.into_raw(), - }; - Ok((picture, bytes.into())) + }) } /// Helpers for tests across the crate. @@ -3026,6 +3052,57 @@ mod tests { .await; } + /// A layer whose source is a still, an image file on a binary track, + /// shows it as it arrives. + #[tokio::test] + async fn a_layer_shows_a_still() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, origin) = context(dir.path()); + let publication = origin + .route("room/guest", moq_net::origin::RouteOptions::hidden()) + .unwrap(); + let mut broadcast = publication.broadcast().clone(); + let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); + let mut catalog = crate::publish::Catalog::new(&mut broadcast, config).unwrap(); + let mut track = broadcast.create_track("source/logo.png", None).unwrap(); + let pixel = image::RgbaImage::from_raw(1, 1, vec![255, 0, 0, 255]).unwrap(); + let mut png = std::io::Cursor::new(Vec::new()); + pixel.write_to(&mut png, image::ImageFormat::Png).unwrap(); + track.write_frame(None, bytes::Bytes::from(png.into_inner())).unwrap(); + let mut still = hang::catalog::BinaryConfig::new(hang::catalog::Mode::Snapshot); + still.mime = Some("image/png".into()); + let mut logo = hang::Catalog::default(); + logo.binary.tracks.insert("source/logo.png".into(), still); + catalog + .mutate(|root| { + root.ext.studio.sources.insert("logo".into(), logo); + }) + .unwrap(); + publication.announce().unwrap(); + + let mut collection = collection(); + collection.sources.insert( + "logo".into(), + Source::Layer { + broadcast: "guest".into(), + source: "logo".into(), + }, + ); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let layer = sources.get("logo").unwrap(); + let pixels = until(|| match layer.next_picture() { + Some(Picture::Rgba { pixels, .. }) => Some(pixels), + _ => None, + }) + .await; + assert_eq!(pixels, [255, 0, 0, 255]); + }) + .await; + } + /// A layer stops with its broadcast source, and starts again on its /// replacement's clock. #[tokio::test] From a192e51ab3946e8aa8c3b3a3ea1fee7185cb53b0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 15:59:16 -0800 Subject: [PATCH 5/9] fix(studio): a watched layer waits for its program sound, and a new instance starts fresh A layer picture shows only once the program sound stamped alike has played, the watcher resets what an old broadcast instance said, and keyframes follow the program playhead. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/playout.rs | 16 +++++++++- rs/moq-scene/src/source.rs | 29 +++++++++++------ rs/moq-studio/src/watch.rs | 64 ++++++++++++++++++++++++++----------- 3 files changed, 79 insertions(+), 30 deletions(-) diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index cf4c07d5..3f9b06ef 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -172,16 +172,30 @@ impl Drop for Speaker<'_> { } } +/// How soon a picture waiting on a stalled playhead looks at it again: a +/// playhead fed by nothing moves only when audio arrives, which tells no one +/// reading the clock but the broadcast's own pictures. +const RECHECK: Duration = Duration::from_millis(10); + /// Whether a picture stamped `timestamp` may show at `now` on `schedule`: /// while audio plays, only once the sound stamped with it has played, so a /// stall in the audio alone holds the picture too. -pub(crate) fn heard(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp, now: Instant) -> bool { +pub fn heard(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp, now: Instant) -> bool { schedule.target().is_none() || schedule .playhead(now) .is_some_and(|head| Duration::from(timestamp) <= head + CONTIGUOUS) } +/// When to look again at whether a picture stamped `timestamp`, which +/// [`heard`] holds at `now`, may show: once the playhead could have reached it +/// at wall rate, and at least [`RECHECK`] later while it does not move. +pub(crate) fn hearing(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp, now: Instant) -> Instant { + let head = schedule.playhead(now).unwrap_or_default(); + let ahead = Duration::from(timestamp).saturating_sub(head + CONTIGUOUS); + now + ahead.max(RECHECK) +} + /// Writes placed frames into a bus channel, so each sounds when it was placed. /// /// A channel plays a write [`BUS_LATENCY`] later only while it holds that much: diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 472d558d..29ab0de8 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -31,8 +31,8 @@ use crate::browser::Page; use crate::collection::{Collection, Source}; use crate::lut::Cube; use crate::media::Clip; -pub use crate::playout::MAX_DELAY; use crate::playout::{BUS_LATENCY, Place, Speaker, VIDEO_DELAY, Writer}; +pub use crate::playout::{MAX_DELAY, heard}; use crate::publish::{Claim, Listing, Publish, Still}; use crate::replay::{MIN_SPEED, Take, Target}; @@ -1213,9 +1213,8 @@ impl Driver { } /// Show each picture once `schedule`, another track's clock, says it is - /// due, the newest due one when several are, as - /// [`play_video`](Self::play_video) does, except that a picture waits only - /// for its moment, not for its sound. + /// due and its sound stamped alike has played, the newest such one when + /// several are, as [`play_video`](Self::play_video) does. async fn follow_video( &self, broadcast: &moq_net::broadcast::Consumer, @@ -1235,24 +1234,34 @@ impl Driver { let mut ended = false; loop { let now = kio::clock::now(); - if let Some(index) = schedule.present(queue.iter().map(|frame| frame.timestamp), now) { + // Counted first: `present` holds the clock's lock while it iterates. + let heard_count = queue + .iter() + .take_while(|frame| heard(schedule, frame.timestamp, now)) + .count(); + let ready = queue.iter().take(heard_count).map(|frame| frame.timestamp); + if let Some(index) = schedule.present(ready, now) { let frame = queue.drain(..=index).next_back().expect("an index into the queue"); self.show(Picture::Frame(frame)); } if ended && queue.is_empty() { return Ok(()); } - let due = queue - .front() - .and_then(|frame| schedule.due(frame.timestamp)) - .filter(|due| *due > now); + // Its moment, or once due, when its sound may have played. + let wake = queue.front().and_then(|frame| match schedule.due(frame.timestamp) { + Some(due) if due > now => Some(due), + _ if !heard(schedule, frame.timestamp, now) => { + Some(crate::playout::hearing(schedule, frame.timestamp, now)) + } + _ => None, + }); tokio::select! { read = consumer.read(), if !ended && queue.len() < QUEUED => match read? { Some(moq_video::decode::Read::Frame(frame)) => queue.push_back(frame), Some(moq_video::decode::Read::Gap { .. } | moq_video::decode::Read::End { .. }) => {} None => ended = true, }, - () = kio::clock::sleep_until(due.unwrap_or(now)), if due.is_some() => {} + () = kio::clock::sleep_until(wake.unwrap_or(now)), if wake.is_some() => {} } } } diff --git a/rs/moq-studio/src/watch.rs b/rs/moq-studio/src/watch.rs index fb8e2e37..e98bcd43 100644 --- a/rs/moq-studio/src/watch.rs +++ b/rs/moq-studio/src/watch.rs @@ -89,7 +89,6 @@ pub fn run(client: moq_tokio::connect::Config, quic: moq_tokio::quic::Config, ur collection, synced: false, layout: None, - start: None, size: None, sources, compositor, @@ -160,6 +159,15 @@ struct Read { } impl Read { + /// Forget what an instance of the broadcast said, as another replaces it + /// or it goes offline: its stamps are not the next one's. + fn reset(&mut self) { + self.size = None; + self.refused = None; + self.pending.clear(); + self.pending.push_back((None, None)); + } + /// Queue `layout` to take effect at `at`. The program plays at most /// [`MAX_DELAY`] behind what arrives, so of the layouts stamped that far /// before it, all due by now, only the newest is kept. @@ -190,8 +198,14 @@ async fn follow(origin: moq_net::origin::Consumer, read: watch::Sender) { loop { tokio::select! { event = instances.available() => match event { - Ok(Some(broadcast)) => playing = Some(Box::pin(instance(broadcast, read.clone()))), - Ok(None) => playing = None, + Ok(Some(broadcast)) => { + read.send_modify(Read::reset); + playing = Some(Box::pin(instance(broadcast, read.clone()))); + } + Ok(None) => { + read.send_modify(Read::reset); + playing = None; + } Err(err) => { tracing::error!(%err, "the broadcast's announcements failed"); return; @@ -317,9 +331,6 @@ struct Watch { synced: bool, /// The layout in effect, or `None` for the program alone. layout: Option, - /// When the program scene's keyframes play from, as the layout in effect - /// set it. - start: Option, /// The program's size, from its catalog. size: Option<(u32, u32)>, sources: Sources, @@ -343,10 +354,9 @@ impl Watch { while let Some((_, at)) = read.pending.front() { let due = match at { None => true, - Some(at) => schedule - .as_ref() - .and_then(|schedule| schedule.due(*at)) - .is_some_and(|due| due <= now), + Some(at) => schedule.as_ref().is_some_and(|schedule| { + schedule.due(*at).is_some_and(|due| due <= now) && source::heard(schedule, *at, now) + }), }; if !due { break; @@ -389,15 +399,18 @@ impl Watch { self.synced = true; } - // A new start is a scene taken, whose keyframes play from it. - let start = self.layout.as_ref().map(|layout| layout.start); - if start != self.start - && let Some(start) = start - && let Ok(stamp) = moq_net::Timestamp::try_from(start) - && let Some(due) = schedule.as_ref().and_then(|schedule| schedule.due(stamp)) - { - self.compositor.set_program_start(due); - self.start = Some(start); + // The keyframes play from `start` as far as the program has played, + // following the clock as it re-anchors or stalls with the sound. + if let (Some(layout), Some(schedule)) = (&self.layout, &schedule) { + let went_on = match schedule.playhead(now) { + Some(head) => now.checked_sub(head.saturating_sub(layout.start)), + None => moq_net::Timestamp::try_from(layout.start) + .ok() + .and_then(|start| schedule.due(start)), + }; + if let Some(went_on) = went_on { + self.compositor.set_program_start(went_on); + } } self.compositor.render(&self.collection, &self.sources, now)?; @@ -554,6 +567,19 @@ mod tests { assert_eq!(stamps, [Some(stamp(3)), Some(stamp(4)), Some(stamp(5))]); } + /// Another instance of the broadcast starts from nothing: the last one's + /// stamps and refusal are not its own. + #[test] + fn a_new_instance_forgets_the_last() { + let mut read = Read::default(); + read.push(None, Some(stamp(1))); + read.refused = Some("another version".into()); + read.reset(); + assert_eq!(read.pending.len(), 1); + assert!(read.pending[0].0.is_none() && read.pending[0].1.is_none()); + assert!(read.size.is_none() && read.refused.is_none()); + } + /// A layout of another version, as a newer studio would publish. #[derive(serde::Serialize)] struct Newer { From ed14f887007addce9fb734ee487f7a3f41926c52 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 16:00:54 -0800 Subject: [PATCH 6/9] fix(studio): a watched layer waits for its program sound, and a new instance starts fresh A layer picture shows only once the program sound stamped alike has played, the watcher resets what an old broadcast instance said, and keyframes follow the program playhead. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/playout.rs | 17 ++++------------- rs/moq-scene/src/source.rs | 6 ++---- 2 files changed, 6 insertions(+), 17 deletions(-) diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index 3f9b06ef..a581c73f 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -172,10 +172,10 @@ impl Drop for Speaker<'_> { } } -/// How soon a picture waiting on a stalled playhead looks at it again: a -/// playhead fed by nothing moves only when audio arrives, which tells no one -/// reading the clock but the broadcast's own pictures. -const RECHECK: Duration = Duration::from_millis(10); +/// How soon a picture held by [`heard`] looks at the playhead again: another +/// broadcast's playhead moves, or re-anchors across a gap, without telling +/// anyone reading its clock, so a held picture polls it. +pub const RECHECK: Duration = Duration::from_millis(10); /// Whether a picture stamped `timestamp` may show at `now` on `schedule`: /// while audio plays, only once the sound stamped with it has played, so a @@ -187,15 +187,6 @@ pub fn heard(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp, now: .is_some_and(|head| Duration::from(timestamp) <= head + CONTIGUOUS) } -/// When to look again at whether a picture stamped `timestamp`, which -/// [`heard`] holds at `now`, may show: once the playhead could have reached it -/// at wall rate, and at least [`RECHECK`] later while it does not move. -pub(crate) fn hearing(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp, now: Instant) -> Instant { - let head = schedule.playhead(now).unwrap_or_default(); - let ahead = Duration::from(timestamp).saturating_sub(head + CONTIGUOUS); - now + ahead.max(RECHECK) -} - /// Writes placed frames into a bus channel, so each sounds when it was placed. /// /// A channel plays a write [`BUS_LATENCY`] later only while it holds that much: diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 29ab0de8..bc64c89d 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -1247,12 +1247,10 @@ impl Driver { if ended && queue.is_empty() { return Ok(()); } - // Its moment, or once due, when its sound may have played. + // Its moment, or once due, soon to see whether its sound has played. let wake = queue.front().and_then(|frame| match schedule.due(frame.timestamp) { Some(due) if due > now => Some(due), - _ if !heard(schedule, frame.timestamp, now) => { - Some(crate::playout::hearing(schedule, frame.timestamp, now)) - } + _ if !heard(schedule, frame.timestamp, now) => Some(now + crate::playout::RECHECK), _ => None, }); tokio::select! { From c87f738a6a8e700a2ccda0e82725d6e9dc5ff6f3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 16:40:35 -0800 Subject: [PATCH 7/9] fix(studio): the program alone supersedes a layout still waiting An untimed fallback (unlisted, ended, or refused layout track) kept the last timed layout ahead of it, so a stalled program playhead left the old layers up. It now clears what waits, and the render gate moves to Read::take_due so a test drives it on a stalled clock. Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/watch.rs | 108 ++++++++++++++++++++++++++++--------- 1 file changed, 83 insertions(+), 25 deletions(-) diff --git a/rs/moq-studio/src/watch.rs b/rs/moq-studio/src/watch.rs index e98bcd43..407e30e8 100644 --- a/rs/moq-studio/src/watch.rs +++ b/rs/moq-studio/src/watch.rs @@ -16,6 +16,7 @@ use std::time::Duration; use anyhow::Context; use kio::clock::Instant; +use moq_mux::timeline; use moq_scene::layout::{self, Layout}; use moq_scene::publish::Ext; use moq_scene::source::{self, MAX_DELAY, Sources, Status}; @@ -168,19 +169,43 @@ impl Read { self.pending.push_back((None, None)); } - /// Queue `layout` to take effect at `at`. The program plays at most - /// [`MAX_DELAY`] behind what arrives, so of the layouts stamped that far - /// before it, all due by now, only the newest is kept. + /// Queue `layout` to take effect at `at`, or at once for `None`, which + /// supersedes every layout still waiting so a stalled program cannot hold + /// them up. The program plays at most [`MAX_DELAY`] behind what arrives, + /// so of the layouts stamped that far before it, all due by now, only the + /// newest is kept. fn push(&mut self, layout: Option, at: Option) { - let due = |stamp: Option| match (stamp, at) { - (None, _) => true, - (Some(_), None) => true, - (Some(stamp), Some(at)) => Duration::from(stamp) + MAX_DELAY <= Duration::from(at), - }; - let passed = self.pending.iter().take_while(|(_, stamp)| due(*stamp)).count(); - self.pending.drain(..passed.saturating_sub(1)); + match at { + None => self.pending.clear(), + Some(at) => { + let passed = self.pending.iter().take_while(|(_, stamp)| { + stamp.is_none_or(|stamp| Duration::from(stamp) + MAX_DELAY <= Duration::from(at)) + }); + self.pending.drain(..passed.count().saturating_sub(1)); + } + } self.pending.push_back((layout, at)); } + + /// Take the newest layout due at `now`: each takes effect once the program + /// plays the frame it is stamped with on `schedule`, and its sound has + /// played. `None` while none is due; `Some(None)` for the program alone. + fn take_due(&mut self, schedule: Option<&timeline::Consumer>, now: Instant) -> Option> { + let mut taken = None; + while let Some((_, at)) = self.pending.front() { + let due = match at { + None => true, + Some(at) => schedule.is_some_and(|schedule| { + schedule.due(*at).is_some_and(|due| due <= now) && source::heard(schedule, *at, now) + }), + }; + if !due { + break; + } + taken = self.pending.pop_front().map(|(layout, _)| layout); + } + taken + } } /// Read each instance of the root broadcast as it comes online: the program's @@ -350,18 +375,7 @@ impl Watch { self.size = read.size; changed = true; } - // Each layout once the program plays the frame it takes effect on. - while let Some((_, at)) = read.pending.front() { - let due = match at { - None => true, - Some(at) => schedule.as_ref().is_some_and(|schedule| { - schedule.due(*at).is_some_and(|due| due <= now) && source::heard(schedule, *at, now) - }), - }; - if !due { - break; - } - let (layout, _) = read.pending.pop_front().expect("peeked"); + if let Some(layout) = read.take_due(schedule.as_ref(), now) { self.layout = layout; changed = true; } @@ -580,6 +594,42 @@ mod tests { assert!(read.size.is_none() && read.refused.is_none()); } + /// Once the layout track is unlisted, the program plays alone at once, + /// even while a stalled program holds back a layout still waiting. + #[tokio::test(start_paused = true)] + async fn unlisting_supersedes_a_waiting_layout() { + let clock = timeline::Producer::new(Duration::from_millis(100)); + let audio = timeline::Audio { + end: Duration::from_secs(1), + buffered: Duration::ZERO, + target: Duration::from_millis(100), + }; + clock.audio(audio, Instant::now()); + let schedule = clock.consume(); + + let (broadcast, catalog, read, reader) = broadcast(); + let mut publisher = layout::Publisher::new(broadcast.clone(), catalog.clone()); + publisher.update(layout(Duration::ZERO), stamp(5)).unwrap(); + pending(&read, 1).await; + tokio::time::advance(Duration::from_secs(60)).await; + let later = Instant::now(); + let take = || { + let mut taken = None; + read.send_if_modified(|read| { + taken = read.take_due(Some(&schedule), later); + false + }); + taken + }; + assert_eq!(take(), None, "the program stalled before it"); + + publisher.finish().unwrap(); + alone(&read).await; + assert_eq!(take(), Some(None), "the program alone"); + assert!(read.borrow().pending.is_empty()); + reader.abort(); + } + /// A layout of another version, as a newer studio would publish. #[derive(serde::Serialize)] struct Newer { @@ -610,6 +660,13 @@ mod tests { pending.map(|(layout, at)| (layout, at.map(Duration::from))).collect() } + /// Wait until the newest thing `read` holds is the program alone. + async fn alone(read: &watch::Sender) { + let mut reads = read.subscribe(); + let alone = |read: &Read| read.pending.back().is_some_and(|(_, at)| at.is_none()); + reads.wait_for(alone).await.unwrap(); + } + /// The reader hands over each layout at the media time it took effect, /// and the program alone once the catalog stops listing the track, until /// it lists it again. @@ -621,14 +678,15 @@ mod tests { let first = (Some(layout(Duration::ZERO)), Some(Duration::from_secs(1))); assert_eq!(pending(&read, 1).await, [first]); publisher.finish().unwrap(); + alone(&read).await; assert_eq!( - pending(&read, 2).await[1], - (None, None), + pending(&read, 1).await, + [(None, None)], "the program alone once unlisted" ); publisher.update(layout(Duration::from_secs(2)), stamp(2)).unwrap(); let again = (Some(layout(Duration::from_secs(2))), Some(Duration::from_secs(2))); - assert_eq!(pending(&read, 3).await[2], again, "listed again"); + assert_eq!(pending(&read, 2).await[1], again, "listed again"); reader.abort(); } From d8caf3f1bc225238ccf032b3a04d4bd139522f59 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 16:47:03 -0800 Subject: [PATCH 8/9] fix(scene): a layer's sound follows its broadcast's clock as it is, waiting while its sound stalls Each queued frame's moment is read from the clock when it is written, not when it arrived, so a re-anchor moves it, and it waits while the program's own sound stamped alike has not been written, as the layer's pictures do. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/playout.rs | 53 +++++++++++++++++++++++++++++++++++++ rs/moq-scene/src/source.rs | 38 ++++++++++++++++---------- 2 files changed, 77 insertions(+), 14 deletions(-) diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index a581c73f..f1f4442f 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -187,6 +187,40 @@ pub fn heard(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp, now: .is_some_and(|head| Duration::from(timestamp) <= head + CONTIGUOUS) } +/// What to do at `now` with a sound stamped `timestamp` that plays on +/// `schedule`, another track's clock. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum Follow { + /// Write it to sound at this moment. + Write(Instant), + /// Look again at this moment. + Wait(Instant), + /// Too late for its moment, or the clock has none. + Drop, +} + +/// Decide what to do with a followed sound from the clock as it is now, not +/// as it was when the sound arrived, so a re-anchor moves it. The clock's own +/// sound stamped alike must have been written first: when that stalls, this +/// one waits with the pictures, until its moment is too close to write. +pub(crate) fn follow(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp, now: Instant) -> Follow { + let Some(sounds) = schedule.due(timestamp) else { + return Follow::Drop; + }; + // A write half the bus's latency late finds nothing queued ahead of it. + if sounds < now + BUS_LATENCY / 2 { + return Follow::Drop; + } + let write = sounds - BUS_LATENCY; + if write > now { + return Follow::Wait(write); + } + if !heard(schedule, timestamp, sounds) { + return Follow::Wait(now + RECHECK); + } + Follow::Write(sounds) +} + /// Writes placed frames into a bus channel, so each sounds when it was placed. /// /// A channel plays a write [`BUS_LATENCY`] later only while it holds that much: @@ -311,6 +345,25 @@ mod tests { assert!(!heard(&schedule, stamp(1_100), at(500)), "shown past its sound"); } + /// A followed sound is written the bus's latency before its moment once + /// the clock's own sound stamped alike has been, waits while that has + /// not, and is dropped once its moment is too close to write. + #[test] + fn a_followed_sound_waits_for_the_clocks_own() { + let clock = timeline::Producer::new(VIDEO_DELAY); + let schedule = clock.consume(); + let changed = Notify::new(); + let mut speaker = Speaker::new(clock.clone(), &changed); + speaker.place(ms(1_000), FRAME, TARGET, at(0)); + assert_eq!(follow(&schedule, stamp(1_000), at(0)), Follow::Wait(at(100))); + assert_eq!(follow(&schedule, stamp(1_000), at(100)), Follow::Write(at(150))); + // Past what the clock's own sound has written: it stalled. + assert_eq!(follow(&schedule, stamp(1_100), at(200)), Follow::Wait(at(210))); + assert_eq!(follow(&schedule, stamp(1_100), at(226)), Follow::Drop); + let fresh = clock.clone().reset(VIDEO_DELAY).consume(); + assert_eq!(follow(&fresh, stamp(1_000), at(0)), Follow::Drop, "no moment yet"); + } + /// A stall longer than the target re-buffers: the next frame sounds the /// target after it arrived, and the pictures stamped with it wait as long. #[test] diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index bc64c89d..50f8c4c0 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -31,7 +31,7 @@ use crate::browser::Page; use crate::collection::{Collection, Source}; use crate::lut::Cube; use crate::media::Clip; -use crate::playout::{BUS_LATENCY, Place, Speaker, VIDEO_DELAY, Writer}; +use crate::playout::{BUS_LATENCY, Follow, Place, Speaker, VIDEO_DELAY, Writer, follow}; pub use crate::playout::{MAX_DELAY, heard}; use crate::publish::{Claim, Listing, Publish, Still}; use crate::replay::{MIN_SPEED, Take, Target}; @@ -1265,8 +1265,9 @@ impl Driver { } /// Play a track's sound into the bus when `schedule`, another track's - /// clock, says each frame sounds, dropping a frame that arrives too late - /// for its moment, or before the clock has one. + /// clock, says each frame sounds and its own sound stamped alike has been + /// written, dropping a frame too late for its moment, or before the clock + /// has one. async fn follow_audio( &self, broadcast: &moq_net::broadcast::Consumer, @@ -1292,20 +1293,29 @@ impl Driver { let mut writer = Writer::new(channel, bus.sample_rate, bus.layout.channels()); // Each frame waits here until the bus's latency before it sounds. - let mut queue = VecDeque::<(Instant, moq_audio::Frame)>::new(); + let mut queue = VecDeque::::new(); let mut ended = false; loop { let now = kio::clock::now(); - while let Some((sounds, _)) = queue.front() - && *sounds - BUS_LATENCY <= now - { - let (sounds, frame) = queue.pop_front().expect("peeked"); - writer.write(sounds, &frame.data, now)?; + let mut write = None; + while let Some(frame) = queue.front() { + match follow(schedule, frame.timestamp, now) { + Follow::Write(sounds) => { + let frame = queue.pop_front().expect("peeked"); + writer.write(sounds, &frame.data, now)?; + } + Follow::Wait(at) => { + write = Some(at); + break; + } + Follow::Drop => { + queue.pop_front(); + } + } } if ended && queue.is_empty() { return Ok(()); } - let write = queue.front().map(|(sounds, _)| *sounds - BUS_LATENCY); tokio::select! { read = consumer.read(), if !ended => match read? { Some(moq_audio::decode::Read::Frame(frame)) => { @@ -1313,11 +1323,11 @@ impl Driver { // ahead of it, and one due past the longest target would // hold whatever a peer stamps. let now = kio::clock::now(); - let placed = schedule.due(frame.timestamp).filter(|sounds| { - *sounds >= now + BUS_LATENCY / 2 && *sounds <= now + MAX_DELAY + BUS_LATENCY + let placed = schedule.due(frame.timestamp).is_some_and(|sounds| { + sounds >= now + BUS_LATENCY / 2 && sounds <= now + MAX_DELAY + BUS_LATENCY }); - if let Some(sounds) = placed { - queue.push_back((sounds, frame)); + if placed { + queue.push_back(frame); } } // A hole or an encoder restart shows in the stamps that follow. From 410ca230ed920c750c79fc34a0c5f9eea835f3cb Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 16:49:38 -0800 Subject: [PATCH 9/9] fix(scene): recheck a followed sound's moment while it waits Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/playout.rs | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index f1f4442f..741986c4 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -211,9 +211,11 @@ pub(crate) fn follow(schedule: &timeline::Consumer, timestamp: moq_net::Timestam if sounds < now + BUS_LATENCY / 2 { return Follow::Drop; } + // The clock re-anchors without telling its readers, so a wait for a + // moment it may yet move earlier is rechecked too. let write = sounds - BUS_LATENCY; if write > now { - return Follow::Wait(write); + return Follow::Wait(write.min(now + RECHECK)); } if !heard(schedule, timestamp, sounds) { return Follow::Wait(now + RECHECK); @@ -345,9 +347,10 @@ mod tests { assert!(!heard(&schedule, stamp(1_100), at(500)), "shown past its sound"); } - /// A followed sound is written the bus's latency before its moment once - /// the clock's own sound stamped alike has been, waits while that has - /// not, and is dropped once its moment is too close to write. + /// A followed sound, rechecked until then, is written the bus's latency + /// before its moment once the clock's own sound stamped alike has been, + /// waits while that has not, and is dropped once its moment is too close + /// to write. #[test] fn a_followed_sound_waits_for_the_clocks_own() { let clock = timeline::Producer::new(VIDEO_DELAY); @@ -355,7 +358,8 @@ mod tests { let changed = Notify::new(); let mut speaker = Speaker::new(clock.clone(), &changed); speaker.place(ms(1_000), FRAME, TARGET, at(0)); - assert_eq!(follow(&schedule, stamp(1_000), at(0)), Follow::Wait(at(100))); + assert_eq!(follow(&schedule, stamp(1_000), at(0)), Follow::Wait(at(10)), "rechecked"); + assert_eq!(follow(&schedule, stamp(1_000), at(95)), Follow::Wait(at(100))); assert_eq!(follow(&schedule, stamp(1_000), at(100)), Follow::Write(at(150))); // Past what the clock's own sound has written: it stalled. assert_eq!(follow(&schedule, stamp(1_100), at(200)), Follow::Wait(at(210)));