From fca1fd48730efe69b53748c7737f530b4e4e9a31 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 10:26:01 -0800 Subject: [PATCH 1/8] chore(quest): claim quest/m0/guest-return.md Quest-Claim: c8368344135001f0 From cc22ab1cead2f1e3d59df48ba71d184bb0ae8911 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 11:01:03 -0800 Subject: [PATCH 2/8] feat(studio): publish a return feed for each guest in the room Each renewed guest gets the program's video by reference and a mix-minus of the mix on program without their microphone, published by the studio's session in the room. Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/engine.rs | 27 ++- rs/moq-studio/src/feed.rs | 463 ++++++++++++++++++++++++++++++++++++ rs/moq-studio/src/guests.rs | 309 ++++++++++++++++-------- rs/moq-studio/src/main.rs | 1 + rs/moq-studio/src/mix.rs | 47 +++- 5 files changed, 735 insertions(+), 112 deletions(-) create mode 100644 rs/moq-studio/src/feed.rs diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 6c10d3fe..473dfd02 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -9,6 +9,7 @@ //! - **audio** advances the audio bus, and with it every mix, on the same //! clock. //! - **audio-encode** encodes the mix on program into the program audio. +//! - Each guest's return feed encodes its mix-minus on the tokio runtime. //! - **encode**: one per program rendition, which waits for each program copy //! and encodes it. A new canvas size or frame rate starts a new rendition. //! - Recordings write their files on the tokio runtime. @@ -434,6 +435,8 @@ pub struct Engine { pub recorder: record::Recorder, /// The room guests join, when configured. pub room: Option>, + /// Each guest's return feed, once the studio's session in the room dials. + feeds: Mutex>, /// The program's broadcast. pub program: Program, /// Keeps broadcast sources' origin open when no connection holds it. @@ -569,6 +572,7 @@ impl Engine { output, recorder, room, + feeds: Mutex::default(), program, _origin: subscribe, stopped: AtomicBool::new(false), @@ -587,6 +591,13 @@ impl Engine { // Where the program clock starts on the catalog's, so the wall time the // catalog advertises for a frame is when the program drew it. let epoch = Duration::from(clock.capture(start)?); + if let Some(room) = &engine.room + && let Some(origin) = room.origin() + { + let program = engine.program.publication.broadcast().consume(); + let feeds = crate::feed::Feeds::new(room.clone(), origin.clone(), program, clock.clone(), epoch)?; + *engine.feeds.lock().unwrap() = Some(feeds); + } let program = engine.clone(); let thread = clocked("program".into(), async move || { program.run_program(start, epoch, renditions, rendition).await @@ -614,6 +625,16 @@ impl Engine { if let Some(guests) = guests { tokio::spawn(crate::guests::run(engine.clone(), guests)); } + if let Some(room) = &engine.room { + // Who has a return feed follows who is renewed, synced with the mixes. + let mut changes = room.changes(); + let follow = engine.clone(); + tokio::spawn(async move { + while changes.changed().await.is_ok() { + follow.state.lock().unwrap().stale = true; + } + }); + } Ok(engine) } @@ -693,12 +714,16 @@ impl Engine { .map(|monitor| (monitor.name().to_string(), monitor.failed())) } - /// Point the program audio and the monitor at the mixes `state` now has. + /// Point the program audio, the monitor, and each guest's return feed at + /// the mixes `state` now has. fn pick_mixes(&self, state: &State) { let mixes = &state.collection.mixes; if let Some(mix) = state.sources.mix(&mixes.program) { self.program_audio.set(mix.clone()); } + if let Some(feeds) = self.feeds.lock().unwrap().as_mut() { + feeds.sync(&state.collection, &state.sources); + } if let Some(monitor) = self.monitor.lock().unwrap().as_mut() { if !mixes.names.iter().any(|name| name == monitor.name()) { monitor.set_name(&mixes.program); diff --git a/rs/moq-studio/src/feed.rs b/rs/moq-studio/src/feed.rs new file mode 100644 index 00000000..b2ad9803 --- /dev/null +++ b/rs/moq-studio/src/feed.rs @@ -0,0 +1,463 @@ +//! Return feeds: what each guest in the room sees and hears of the show. +//! +//! Every guest the studio renews gets a broadcast at +//! `.return//` in the room, which only their token may +//! read. Its video is the program's: the studio's session in the room serves the +//! program broadcast once at `.program`, and each feed's catalog lists its video +//! renditions by reference, so a guest costs no video encode. Its audio is its +//! own: a mix-minus of the mix on program without the guest's microphone, so +//! they follow the show without hearing themselves back. +//! +//! The studio's session in the room publishes them, holding only what the room +//! key signs. A guest costs one subtraction per audio block and an Opus encoder +//! for as long as they are renewed, so the cost is bounded by the guests the +//! studio signed for and kept in the room. + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::Context; +use bytes::Bytes; +use moq_mux::catalog::Stream; +use moq_net::Path; +use moq_scene::Collection; +use moq_scene::source::Sources; + +use crate::guests::{PROGRAM, Room}; +use crate::mix; + +/// A guest's return audio: speech and the show's sound, for following along, +/// not for broadcast. +const BITRATE: moq_net::bandwidth::Rate = moq_net::bandwidth::Rate::from_kbps(64); + +/// The return feeds of every renewed guest, and the program they show. +pub struct Feeds { + room: Arc, + /// What the studio's session in the room publishes. + origin: moq_net::origin::Producer, + /// The program broadcast, which each feed's video names. + program: moq_net::broadcast::Consumer, + /// The program catalog's clock, which each feed's catalog shares so its + /// audio lines up with the program's video. + clock: moq_mux::Clock, + /// Where the program audio's timestamps start on that clock. + epoch: Duration, + feeds: BTreeMap, + /// Serves the program at [`PROGRAM`] until dropped. + serving: tokio::task::JoinHandle<()>, +} + +impl Feeds { + /// Serve `program` at [`PROGRAM`] in `room`, through `origin`, the studio's + /// session there, and publish a feed for each guest [`sync`](Self::sync) + /// finds renewed. `clock` and `epoch` are the program audio's, which the + /// feeds' audio is stamped against. + pub fn new( + room: Arc, + origin: moq_net::origin::Producer, + program: moq_net::broadcast::Consumer, + clock: moq_mux::Clock, + epoch: Duration, + ) -> anyhow::Result { + let requests = origin + .requests(PROGRAM) + .context("cannot serve the program in the room")?; + let served = program.clone(); + let serving = tokio::spawn(async move { + while let Ok(request) = requests.requested_broadcast().await { + // The route claims everything beneath it too, which holds nothing. + match request.path().as_str() == PROGRAM { + true => request.serve(&served), + false => request.reject(moq_net::Error::NotFound), + } + } + }); + Ok(Self { + room, + origin, + program, + clock, + epoch, + feeds: BTreeMap::new(), + serving, + }) + } + + /// Publish a feed for each renewed guest and end the others', and point + /// each at the mix `collection` puts on program, without the guest's + /// microphone as `sources` runs it. + pub fn sync(&mut self, collection: &Collection, sources: &Sources) { + let renewed = self.room.renewed(); + self.feeds + .retain(|identity, feed| renewed.contains(&(identity.clone(), feed.generation))); + let program = sources.mix(&collection.mixes.program); + for (identity, generation) in renewed { + if !self.feeds.contains_key(&identity) { + match self.open(&identity, generation) { + Ok(feed) => self.feeds.insert(identity.clone(), feed), + Err(err) => { + tracing::error!(%identity, error = format!("{err:#}"), "cannot publish a guest's return feed"); + continue; + } + }; + } + let Some(program) = program else { continue }; + let pick = &self.feeds[&identity].pick; + // A guest the host never added, or removed, is in no mix to take out. + let voice = self.room.voice(collection, &identity); + match voice.and_then(|name| sources.get(name)?.strip()) { + Some(strip) => pick.set(program.minus(strip)), + None => pick.set(program.clone()), + } + } + } + + /// Announce a feed for `identity` in `generation`, its audio following the + /// mix-minus it is picked. + fn open(&self, identity: &str, generation: u64) -> anyhow::Result { + let path = Room::feed_path(identity, generation)?; + // Rendition references resolve against the feed's own path. + let program = Path::new(PROGRAM)? + .relative(&path) + .context("the program cannot be named from a feed")?; + let publication = self.origin.create_broadcast(path)?; + let mut broadcast = publication.broadcast().clone(); + let config = moq_mux::catalog::Config::default().with_clock(self.clock.clone()); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config)?; + + let bus = moq_audio::bus::Config::default(); + let mut input = moq_audio::encode::Input::new(bus.sample_rate, bus.layout); + input.format = moq_audio::Format::F32; + let mut options = moq_audio::encode::Options::default(); + options.track = Some("audio".into()); + options.config = moq_audio::encode::Config::from_input(moq_audio::encode::Codec::Opus, &input); + options.config.bitrate = Some(BITRATE); + let audio = moq_audio::encode::Producer::new(&mut broadcast, catalog.clone(), input, &options)?; + publication.announce(moq_net::origin::Route::default())?; + + let pick = mix::Pick::default(); + let (stop, stopped) = tokio::sync::oneshot::channel(); + let running = Running { + _publication: publication, + catalog, + audio, + follow: pick.follow(), + program: self.program.clone(), + reference: program, + epoch: self.epoch, + rate: bus.sample_rate, + }; + let identity = identity.to_string(); + tokio::spawn(async move { + if let Err(err) = running.run(stopped).await { + tracing::error!(%identity, error = format!("{err:#}"), "a guest's return feed stopped"); + } + }); + Ok(Feed { + generation, + pick, + _stop: stop, + }) + } +} + +impl Drop for Feeds { + fn drop(&mut self) { + self.serving.abort(); + } +} + +/// One guest's return feed, which ends when dropped. +struct Feed { + /// The generation it is published under. + generation: u64, + /// The mix-minus its audio reads. + pick: mix::Pick, + /// Dropped to end the feed. + _stop: tokio::sync::oneshot::Sender<()>, +} + +/// A feed's broadcast, as its task runs it. +struct Running { + /// Ends the broadcast once the task returns. + _publication: moq_net::origin::Publication, + catalog: moq_mux::catalog::Producer, + audio: moq_audio::encode::Producer, + follow: mix::Follow, + /// The program broadcast, whose video renditions the catalog lists. + program: moq_net::broadcast::Consumer, + /// The program, relative to the feed. + reference: moq_net::path::Relative, + epoch: Duration, + rate: u32, +} + +impl Running { + /// List the program's video and encode the picked audio until `stopped`, + /// then finish the audio track. + async fn run(mut self, stopped: tokio::sync::oneshot::Receiver<()>) -> anyhow::Result<()> { + let Self { + catalog, + audio, + follow, + program, + reference, + epoch, + rate, + .. + } = &mut self; + let ended = tokio::select! { + listed = list(program, catalog, reference) => listed, + encoded = encode(follow, audio, *epoch, *rate) => encoded, + // Dropped with the feed. + _ = stopped => Ok(()), + }; + if let Err(err) = ended { + self.audio.abort(moq_net::StreamError::Internal.into()); + return Err(err); + } + self.audio.finish()?; + Ok(()) + } +} + +/// List the video renditions of each catalog `program` publishes in +/// `catalog`, read from the broadcast at `reference`, until the program ends. +async fn list( + program: &moq_net::broadcast::Consumer, + catalog: &mut moq_mux::catalog::Producer, + reference: &moq_net::path::Relative, +) -> anyhow::Result<()> { + let format = moq_mux::catalog::CatalogFormat::Hang; + let mut listings = moq_mux::catalog::Consumer::<()>::new(program, format) + .await + .context("cannot read the program's catalog")?; + while let Some(listed) = listings.next().await? { + let mut video = listed.video; + for rendition in video.renditions.values_mut() { + rendition.broadcast = Some(reference.clone()); + } + catalog.mutate(|catalog| catalog.video = video)?; + } + Ok(()) +} + +/// Encode what `follow` reads into `audio`, stamped `epoch` past the bus's +/// start at `rate`, as the program audio is, until the bus stops. +async fn encode( + follow: &mut mix::Follow, + audio: &mut moq_audio::encode::Producer, + epoch: Duration, + rate: u32, +) -> anyhow::Result<()> { + let rate = u128::from(rate); + let mut next = None; + while let Some(block) = follow.read().await { + // Audio this encoder fell too far behind to read leaves a gap. + if next.is_some_and(|next| next != block.position) { + audio.reanchor(); + } + next = Some(block.position + block.frames() as u64); + let micros = epoch.as_micros() + u128::from(block.position) * 1_000_000 / rate; + let Ok(timestamp) = moq_net::Timestamp::from_micros(micros as u64) else { + continue; + }; + let data: Vec = block.data.iter().flat_map(|sample| sample.to_le_bytes()).collect(); + audio.write(&moq_audio::Frame::new(Bytes::from(data), timestamp))?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use moq_audio::bus::{Bus, Channel}; + use moq_scene::{History, Source}; + use url::Url; + + use super::*; + use crate::guests::tests::{EXPIRY, Relay, key}; + + /// The default bus's 10 ms block. + const BLOCK: usize = 480; + + const TIMEOUT: Duration = Duration::from_secs(10); + + /// A program broadcast listing one video rendition, `video`. + fn program() -> (moq_net::broadcast::Producer, moq_mux::catalog::Producer) { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let codec: hang::catalog::VideoCodec = "avc1.64001f".parse().unwrap(); + let config = hang::catalog::VideoConfig::new(codec); + catalog + .mutate(|catalog| { + catalog.video.renditions.insert("video".into(), config); + }) + .unwrap(); + (broadcast, catalog) + } + + /// Return feeds on `origin` for the guests in `room`, of a program that + /// lives as long as the returned broadcast and catalog. + fn feeds( + room: &Arc, + origin: &moq_net::origin::Producer, + ) -> (Feeds, (moq_net::broadcast::Producer, moq_mux::catalog::Producer)) { + let (broadcast, catalog) = program(); + let feeds = Feeds::new( + room.clone(), + origin.clone(), + broadcast.consume(), + catalog.clock(), + Duration::ZERO, + ); + (feeds.unwrap(), (broadcast, catalog)) + } + + /// Alice and bob in `room` with their cameras on, and a collection playing + /// each camera, whose sources run on `bus`. + fn guests(room: &Room, bus: &Bus) -> (Collection, Sources) { + let mut collection = Collection::new(); + for guest in ["alice", "bob"] { + let camera = format!("{guest}/camera.hang"); + room.observe(&Path::new(&camera).unwrap(), true, &collection, &History::new()); + let path = format!("guests/{camera}"); + collection.sources.insert(guest.into(), Source::Broadcast { path }); + } + let origin = moq_tokio::origin::spawn(); + let mut sources = Sources::new(moq_scene::source::Context { + bus: bus.clone(), + origin: origin.consume(), + dir: ".".into(), + output: moq_video::Output::Cpu, + browser: None, + publish: None, + }); + // Not run: the test feeds each guest's strip itself. + drop(sources.sync(&collection)); + (collection, sources) + } + + /// A stereo channel into `bus` under the strip of the source `name`. + fn channel(bus: &Bus, sources: &Sources, name: &str) -> Channel { + let strip = sources.get(name).unwrap().strip().unwrap().clone(); + let mut input = moq_audio::encode::Input::new(48_000, moq_audio::Layout::Stereo); + input.format = moq_audio::Format::F32; + bus.channel(input, strip).unwrap() + } + + fn feed(channel: &mut Channel, value: f32) { + let data: Vec = std::iter::repeat_n(value, BLOCK * 2) + .flat_map(f32::to_le_bytes) + .collect(); + channel.write(&data).unwrap(); + } + + /// The catalog `broadcast` lists once it lists video. + async fn listed(broadcast: &moq_net::broadcast::Consumer) -> hang::Catalog { + let format = moq_mux::catalog::CatalogFormat::Hang; + let mut listings = moq_mux::catalog::Consumer::<()>::new(broadcast, format).await.unwrap(); + loop { + let catalog = listings.next().await.unwrap().expect("a catalog"); + if !catalog.video.renditions.is_empty() { + return catalog; + } + } + } + + /// Each guest's feed lists the program's video by reference and audio of + /// its own, which hears everyone on program but that guest, and ends once + /// the guest is no longer renewed. + #[tokio::test] + async fn each_guest_hears_the_program_without_themselves() { + let _clock = kio::clock::Manual::new().install(&tokio::runtime::Handle::current()); + let connect = Url::parse("https://relay.example/show").unwrap(); + let page = Url::parse("https://studio.example/join/").unwrap(); + let key = key("show/guests", &["**"], &["**"]); + let room = Arc::new(Room::new(&connect, "guests", key, page, EXPIRY).unwrap()); + let origin = moq_tokio::origin::spawn(); + let (mut feeds, _program) = feeds(&room, &origin); + let (bus, mut driver) = Bus::new(Default::default()).unwrap(); + let (collection, sources) = guests(&room, &bus); + let mut alice = channel(&bus, &sources, "alice"); + let mut bob = channel(&bus, &sources, "bob"); + feeds.sync(&collection, &sources); + + let path = Room::feed_path("alice", 0).unwrap(); + let broadcast = origin.consume().follow(path, None).unwrap().available().await.unwrap(); + let catalog = listed(&broadcast).await; + let video = &catalog.video.renditions["video"]; + assert_eq!(video.broadcast.as_ref().unwrap().as_str(), "../../.program"); + assert!(catalog.audio.renditions.contains_key("audio"), "{catalog:?}"); + + let mut heard_by_alice = feeds.feeds["alice"].pick.follow(); + let mut heard_by_bob = feeds.feeds["bob"].pick.follow(); + // Past the bus's latency, so each mix carries both guests. + let mut last = (None, None); + for _ in 0..20 { + feed(&mut alice, 0.25); + feed(&mut bob, 0.5); + driver.advance(BLOCK); + last = (heard_by_alice.read().await, heard_by_bob.read().await); + } + let (alice_hears, bob_hears) = (last.0.unwrap(), last.1.unwrap()); + assert_eq!(alice_hears.data[0], 0.5, "alice hears only bob"); + assert_eq!(bob_hears.data[0], 0.25, "bob hears only alice"); + + let path = Room::feed_path("bob", 0).unwrap(); + let bob = origin.consume().follow(path, None).unwrap().available().await.unwrap(); + room.stop_renewing("bob").unwrap(); + feeds.sync(&collection, &sources); + assert_eq!(feeds.feeds.keys().collect::>(), ["alice"]); + // A guest who is not renewed has no feed. + bob.closed().await; + } + + /// Through a relay that verifies tokens, a guest reads their own feed and + /// the program it shows, but not another guest's feed. + #[tokio::test(flavor = "multi_thread")] + async fn a_guest_reads_only_their_own_feed_through_a_relay() { + let relay = Relay::start().await; + let room = relay.room(); + let origin = room.origin().unwrap().clone(); + let (mut feeds, _program) = feeds(&room, &origin); + let (bus, _driver) = Bus::new(Default::default()).unwrap(); + let (collection, sources) = guests(&room, &bus); + feeds.sync(&collection, &sources); + + let read = moq_tokio::origin::spawn(); + let link = room.invite("alice").unwrap().link; + let _alice = relay.join(&link, &moq_tokio::origin::spawn(), &read); + let follow = async |path: Path| read.consume().follow(path, None)?.available().await; + + let path = Room::feed_path("alice", 0).unwrap(); + let feed = kio::clock::timeout(TIMEOUT, follow(path.clone())) + .await + .expect("alice's feed reaches her") + .unwrap(); + let catalog = kio::clock::timeout(TIMEOUT, listed(&feed)).await.unwrap(); + let reference = catalog.video.renditions["video"].broadcast.clone().unwrap(); + let program = path.resolve(&reference).unwrap(); + assert_eq!(program.as_str(), PROGRAM); + let shown = kio::clock::timeout(TIMEOUT, follow(program)) + .await + .expect("the program reaches alice") + .unwrap(); + let shown = kio::clock::timeout(TIMEOUT, listed(&shown)).await.unwrap(); + assert!(shown.video.renditions.contains_key("video")); + + // Bob reads his own, so the relay routes it. + let bob = moq_tokio::origin::spawn(); + let link = room.invite("bob").unwrap().link; + let _bob = relay.join(&link, &moq_tokio::origin::spawn(), &bob); + let path = Room::feed_path("bob", 0).unwrap(); + kio::clock::timeout(TIMEOUT, bob.consume().follow(path.clone(), None).unwrap().available()) + .await + .expect("bob's feed reaches him") + .unwrap(); + let other = kio::clock::timeout(TIMEOUT, read.consume().request(path, None)) + .await + .expect("the relay answers"); + assert!(other.is_err(), "alice read bob's feed"); + } +} diff --git a/rs/moq-studio/src/guests.rs b/rs/moq-studio/src/guests.rs index a768435e..2025cd24 100644 --- a/rs/moq-studio/src/guests.rs +++ b/rs/moq-studio/src/guests.rs @@ -17,6 +17,10 @@ //! band. Removing a guest from scenes stops their renewals, across restarts, //! until a new link, which moves them to the next generation, so their last //! token bounds how long they stay. +//! +//! A renewed guest also gets a return feed, the program with a mix-minus, at +//! `.return//`, which only their token may read either; +//! see [`crate::feed`]. use std::collections::{BTreeMap, BTreeSet}; use std::path::PathBuf; @@ -42,6 +46,14 @@ const RENEW: &str = ".renew"; /// the same name. const TOKEN_TRACK: &str = "token"; +/// The prefix of each guest's return feed, which moq-room's announcements hide; +/// [`Room::feed_path`] names the rest. `js/join` reads the same path. +const RETURN: &str = ".return"; + +/// Where the studio's session in the room serves the program, which every +/// guest's return feed shows. moq-room's announcements hide it. +pub const PROGRAM: &str = ".program"; + /// The shortest `--guest-expiry`: half of it must leave room for a renewal to /// reach the guest and the relay before the token it replaces expires. const MIN_EXPIRY: Duration = Duration::from_secs(60); @@ -131,10 +143,13 @@ pub struct Room { participants: Mutex>, /// Where removals are kept, or `None` to keep them in memory alone. removals: Option, - /// Wakes the renewals when who is renewed changes. - changed: tokio::sync::Notify, + /// Bumped when who is renewed changes, which wakes the renewals and the + /// return feeds. + changed: tokio::sync::watch::Sender<()>, /// The session publishing the renewals, once it dials. session: OnceLock, + /// What that session publishes, once it dials. + origin: OnceLock, } /// One guest the studio knows of. @@ -230,6 +245,7 @@ impl Room { removals: None, changed: Default::default(), session: Default::default(), + origin: Default::default(), }; // Sign once now, so a key that cannot sign for guests or for the studio's // own session fails at startup. @@ -346,7 +362,7 @@ impl Room { let invite = Invite { link, expires }; participants.entry(name.to_string()).or_default().invite = Some(invite.clone()); drop(participants); - self.changed.notify_one(); + self.changed.send_replace(()); Ok(invite) } @@ -361,7 +377,7 @@ impl Room { self.change(&mut participants, identity, |participant| participant.removed = true)?; } drop(participants); - self.changed.notify_one(); + self.changed.send_replace(()); Ok(()) } @@ -370,6 +386,21 @@ impl Room { Ok(Path::new(&format!("{RENEW}/{identity}/{generation}"))?) } + /// Where `identity`'s return feed in `generation` is published, in the room. + pub fn feed_path(identity: &str, generation: u64) -> anyhow::Result { + Ok(Path::new(&format!("{RETURN}/{identity}/{generation}"))?) + } + + /// A receiver that sees each change to who is renewed. + pub fn changes(&self) -> tokio::sync::watch::Receiver<()> { + self.changed.subscribe() + } + + /// What the studio's session in the room publishes, once it dials. + pub fn origin(&self) -> Option<&moq_net::origin::Producer> { + self.origin.get() + } + /// Why the session publishing guests' renewed tokens is not connected, if it /// is not. pub fn renewal_failure(&self) -> Option { @@ -378,24 +409,31 @@ impl Room { } /// A token for `identity`, and when it expires. It publishes moq-room's - /// paths but subscribes only to its own renewals, so no guest reads - /// another's. + /// paths but subscribes only to its own renewals and return feed, and the + /// program that feed shows, so no guest reads another's. fn sign(&self, identity: &str, generation: u64) -> anyhow::Result<(String, kio::clock::Wall)> { let invalid = || format!("cannot invite `{identity}`"); let renewals = Self::renewal_path(identity, generation).with_context(invalid)?; + let feed = Self::feed_path(identity, generation).with_context(invalid)?; let claims = moq_room::claims(self.root.as_str(), identity) .with_context(invalid)? - .with_subscribe([moq_auth::Pattern::literal(renewals.as_str())?]); + .with_subscribe([ + moq_auth::Pattern::literal(renewals.as_str())?, + moq_auth::Pattern::literal(feed.as_str())?, + moq_auth::Pattern::literal(PROGRAM)?, + ]); self.sign_claims(claims) .context("the room key cannot sign a guest's token") } /// A token for the studio's own session in the room, which publishes the - /// renewals and nothing else. + /// renewals, the return feeds, and the program they show, and nothing else. fn sign_session(&self) -> anyhow::Result<(String, kio::clock::Wall)> { - let claims = moq_auth::Claims::default() - .with_root(self.root.as_str()) - .with_publish([moq_auth::Pattern::subtree(RENEW)?]); + let claims = moq_auth::Claims::default().with_root(self.root.as_str()).with_publish([ + moq_auth::Pattern::subtree(RENEW)?, + moq_auth::Pattern::subtree(RETURN)?, + moq_auth::Pattern::subtree(PROGRAM)?, + ]); self.sign_claims(claims) .context("the room key cannot sign the studio's own session in the room") } @@ -414,7 +452,7 @@ impl Room { } /// The guests whose tokens are renewed, and in which generation. - fn renewed(&self) -> Vec<(String, u64)> { + pub fn renewed(&self) -> Vec<(String, u64)> { let participants = self.participants.lock().unwrap(); let renewed = participants.iter().filter(|(_, participant)| participant.renewed()); renewed @@ -428,7 +466,7 @@ impl Room { /// /// Every guest is tracked, including one invited before this session whose /// link still works, so a restart finds the guests already connected. - fn observe( + pub(crate) fn observe( &self, path: &Path, online: bool, @@ -457,7 +495,7 @@ impl Room { Kind::Screen => participant.screen = online, } drop(participants); - self.changed.notify_one(); + self.changed.send_replace(()); if !online || existing { return None; } @@ -532,21 +570,37 @@ impl Room { /// The sources in `collection` that show the guest called `identity`. pub fn sources_of(&self, collection: &Collection, identity: &str) -> Vec { - collection - .sources - .iter() - .filter(|(_, source)| { - let Source::Broadcast { path } = source else { - return false; - }; - let Ok(path) = Path::new(path) else { return false }; - path.strip_prefix(&self.path) - .and_then(moq_room::parse) - .is_some_and(|parsed| parsed.identity.as_str() == identity) - }) + self.guest_sources(collection, identity) .map(|(name, _)| name.clone()) .collect() } + + /// The source in `collection` playing the camera of the guest called + /// `identity`, whose sound is their microphone. + pub fn voice<'a>(&self, collection: &'a Collection, identity: &str) -> Option<&'a str> { + self.guest_sources(collection, identity) + .find(|(_, kind)| *kind == Kind::Camera) + .map(|(name, _)| name.as_str()) + } + + /// The sources in `collection` that show the guest called `identity`, with + /// which of their broadcasts each plays. + fn guest_sources<'a>( + &self, + collection: &'a Collection, + identity: &str, + ) -> impl Iterator { + collection.sources.iter().filter_map(move |(name, source)| { + let Source::Broadcast { path } = source else { + return None; + }; + let parsed = Path::new(path) + .ok()? + .strip_prefix(&self.path) + .and_then(moq_room::parse)?; + (parsed.identity.as_str() == identity).then_some((name, parsed.kind)) + }) + } } /// Whether `url` names this machine, so a browser treats an `http://` page @@ -654,6 +708,7 @@ impl Renewal { let origin = moq_tokio::origin::spawn(); let connection = client.with_publisher(&origin).connect(room.url.clone()); let _ = room.session.set(connection.monitor()); + let _ = room.origin.set(origin.clone()); let renewals = Renewals::new(room, origin); Ok(Self { renewals, connection }) } @@ -665,6 +720,7 @@ impl Renewal { let room = self.renewals.room.clone(); let period = room.expiry / 2; let mut renew = kio::clock::interval_at(kio::clock::now() + period, period); + let mut changes = room.changes(); loop { self.renewals.sync(); tokio::select! { @@ -675,7 +731,8 @@ impl Renewal { } self.renewals.renew(); } - () = room.changed.notified() => {} + // The room outlives this loop, so its changes never end. + _ = changes.changed() => {} } } } @@ -780,12 +837,12 @@ fn admit(draft: &mut Draft, name: &str, source: Source) -> Result<(), moq_scene: } #[cfg(test)] -mod tests { +pub(crate) mod tests { use super::*; - const EXPIRY: Duration = Duration::from_secs(3600); + pub(crate) const EXPIRY: Duration = Duration::from_secs(3600); - fn key(root: &str, publish: &[&str], subscribe: &[&str]) -> moq_auth::Key { + pub(crate) fn key(root: &str, publish: &[&str], subscribe: &[&str]) -> moq_auth::Key { let patterns = |texts: &[&str]| texts.iter().map(|text| text.parse().unwrap()).collect(); moq_auth::Key::generate(moq_auth::Algorithm::ES256, Some(moq_auth::KeyId::random())) .unwrap() @@ -802,7 +859,7 @@ mod tests { Room::new(&Url::parse(connect).unwrap(), "guests", key, page, EXPIRY) } - fn fragment(link: &Url) -> BTreeMap { + pub(crate) fn fragment(link: &Url) -> BTreeMap { url::form_urlencoded::parse(link.fragment().unwrap().as_bytes()) .into_owned() .collect() @@ -974,8 +1031,8 @@ mod tests { assert_eq!(claims.publish, ["alice/**".parse().unwrap()].into_iter().collect()); assert_eq!( claims.subscribe, - [".renew/alice/0".parse().unwrap()].into_iter().collect(), - "a guest reads only its own renewals" + subscribes("alice", 0), + "a guest reads only its own renewals and feed, and the program" ); let (issued, expires) = (claims.issued.unwrap(), claims.expires.unwrap()); let lifetime = expires.saturating_duration_since(issued); @@ -1063,6 +1120,116 @@ mod tests { assert!(room.participants()["alice"].screen); } + /// What a guest's token in `generation` subscribes to. + fn subscribes(identity: &str, generation: u64) -> moq_auth::Patterns { + [ + format!(".renew/{identity}/{generation}"), + format!(".return/{identity}/{generation}"), + PROGRAM.to_string(), + ] + .iter() + .map(|pattern| pattern.parse().unwrap()) + .collect() + } + + /// A relay that verifies tokens signed by the operator's key or a room + /// key scoped to `show/guests`, as `moq auth serve` does. + pub(crate) struct Relay { + /// Dials the relay, trusting the certificate it generated. + pub settings: moq_tokio::connect::Config, + /// The studio's `--connect` URL, whose token may subscribe to the show. + pub connect: Url, + /// The room key. + pub key: moq_auth::Key, + _keys: tempfile::TempDir, + } + + impl Relay { + pub(crate) async fn start() -> Self { + let _ = moq_tokio::crypto::install_default(); + let keys = tempfile::tempdir().unwrap(); + let admin = moq_auth::Key::generate(moq_auth::Algorithm::ES256, Some(moq_auth::KeyId::random())).unwrap(); + let scoped = key("show/guests", &["**"], &["**"]); + for key in [&admin, &scoped] { + let kid = key.kid.as_ref().unwrap().encode(); + key.to_public() + .unwrap() + .to_file(keys.path().join(format!("{kid}.jwk"))) + .unwrap(); + } + let mut policy = moq_auth::serve::Policy::default(); + policy.keys = Some(moq_auth::serve::Keys::Dir(keys.path().into())); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let auth = format!("http://{}/", listener.local_addr().unwrap()).parse().unwrap(); + let server = moq_auth::serve::Server::new(policy).unwrap(); + tokio::spawn(async move { server.serve(listener).await }); + + let mut config = moq_relay::Config::default(); + config.drain_timeout = Some(Duration::ZERO.into()); + config.listen.bind = Some("127.0.0.1:0".parse().unwrap()); + config.listen.tls.generate = vec!["localhost".into()]; + config.auth.url = Some(auth); + // Every version, as a relay serving the studio and its join page opts into + // the one that renews tokens in band. + config.listen.version = moq_net::Versions::all() + .iter() + .copied() + .chain([LITE_07.parse().unwrap()]) + .collect(); + let relay = moq_relay::Relay::load(config).await.unwrap().with_signals(false); + let addr = relay.quic_addr().unwrap(); + let ready = relay.ready(); + tokio::spawn(relay.run()); + ready.wait().await.unwrap(); + + let mut settings = moq_tokio::connect::Config::default(); + // The relay's certificate is generated for this test alone. + settings.tls.insecure = Some(true); + let studio = moq_auth::Claims::default() + .with_root("show") + .with_subscribe([moq_auth::Pattern::all()]); + let mut connect = Url::parse(&format!("https://localhost:{}/show", addr.port())).unwrap(); + connect + .query_pairs_mut() + .append_pair("jwt", &admin.sign(&studio).unwrap()); + Self { + settings, + connect, + key: scoped, + _keys: keys, + } + } + + /// The room `guests`, its session in the room dialed and running. + pub(crate) fn room(&self) -> Arc { + let page = Url::parse("https://studio.example/join/").unwrap(); + let room = Arc::new(Room::new(&self.connect, "guests", self.key.clone(), page, EXPIRY).unwrap()); + let renewal = Renewal::connect(room.clone(), &self.settings, Default::default()).unwrap(); + tokio::spawn(renewal.run()); + room + } + + /// Dial as the join page does with `link`, publishing `publish` and + /// reading into `read`: its token kept out of the URL, on the version + /// that renews it in band. + pub(crate) fn join( + &self, + link: &Url, + publish: &moq_net::origin::Producer, + read: &moq_net::origin::Producer, + ) -> moq_tokio::Connection { + let fragment = fragment(link); + let mut page = self.settings.clone(); + page.tokens = vec![fragment["jwt"].clone().try_into().unwrap()]; + page.version = vec![LITE_07.parse().unwrap()]; + page.init(Default::default()) + .unwrap() + .with_publisher(publish) + .with_subscriber(read.clone()) + .connect(Url::parse(&fragment["url"]).unwrap()) + } + } + /// A guest who opens the link reaches the relay with its token, and the /// studio's own connection sees its camera in the room. The studio's session /// in the room then hands the guest a fresh token, which the relay accepts in @@ -1070,65 +1237,15 @@ mod tests { #[tokio::test(flavor = "multi_thread")] async fn a_guest_joins_and_renews_through_a_relay_that_verifies_tokens() { const TIMEOUT: Duration = Duration::from_secs(10); - let _ = moq_tokio::crypto::install_default(); + let relay = Relay::start().await; + let client = relay.settings.clone().init(Default::default()).unwrap(); - // The operator's key for the studio's own connection, and the room's. - let dir = tempfile::tempdir().unwrap(); - let admin = moq_auth::Key::generate(moq_auth::Algorithm::ES256, Some(moq_auth::KeyId::random())).unwrap(); - let scoped = key("show/guests", &["**"], &["**"]); - for key in [&admin, &scoped] { - let kid = key.kid.as_ref().unwrap().encode(); - key.to_public() - .unwrap() - .to_file(dir.path().join(format!("{kid}.jwk"))) - .unwrap(); - } - let mut policy = moq_auth::serve::Policy::default(); - policy.keys = Some(moq_auth::serve::Keys::Dir(dir.path().into())); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let auth = format!("http://{}/", listener.local_addr().unwrap()).parse().unwrap(); - let server = moq_auth::serve::Server::new(policy).unwrap(); - tokio::spawn(async move { server.serve(listener).await }); - - let mut config = moq_relay::Config::default(); - config.drain_timeout = Some(Duration::ZERO.into()); - config.listen.bind = Some("127.0.0.1:0".parse().unwrap()); - config.listen.tls.generate = vec!["localhost".into()]; - config.auth.url = Some(auth); - // Every version, as a relay serving the studio and its join page opts into - // the one that renews tokens in band. - config.listen.version = moq_net::Versions::all() - .iter() - .copied() - .chain([LITE_07.parse().unwrap()]) - .collect(); - let relay = moq_relay::Relay::load(config).await.unwrap().with_signals(false); - let addr = relay.quic_addr().unwrap(); - let ready = relay.ready(); - tokio::spawn(relay.run()); - ready.wait().await.unwrap(); - - let mut settings = moq_tokio::connect::Config::default(); - // The relay's certificate is generated for this test alone. - settings.tls.insecure = Some(true); - let client = settings.clone().init(Default::default()).unwrap(); - - let studio = moq_auth::Claims::default() - .with_root("show") - .with_subscribe([moq_auth::Pattern::all()]); - let mut connect = Url::parse(&format!("https://localhost:{}/show", addr.port())).unwrap(); - connect - .query_pairs_mut() - .append_pair("jwt", &admin.sign(&studio).unwrap()); let subscribe = moq_tokio::origin::spawn(); let all = moq_auth::Pattern::all().into(); let mut announced = moq_room::Room::new(&subscribe.consume().scope("guests", &all).unwrap(), None); - let _studio = client.with_subscriber(subscribe).connect(connect.clone()); + let _studio = client.with_subscriber(subscribe).connect(relay.connect.clone()); - let page = Url::parse("https://studio.example/join/").unwrap(); - let room = Arc::new(Room::new(&connect, "guests", scoped.clone(), page, EXPIRY).unwrap()); - let renewal = Renewal::connect(room.clone(), &settings, Default::default()).unwrap(); - tokio::spawn(renewal.run()); + let room = relay.room(); // Another guest in the room, whose renewals alice must not read. room.observe( &Path::new("bob/camera.hang").unwrap(), @@ -1137,22 +1254,11 @@ mod tests { &History::new(), ); - // As the join page dials: its token kept out of the URL, on the version - // that renews it in band. - let fragment = fragment(&room.invite("alice").unwrap().link); - let mut page = settings.clone(); - page.tokens = vec![fragment["jwt"].clone().try_into().unwrap()]; - page.version = vec![LITE_07.parse().unwrap()]; let guest = moq_tokio::origin::spawn(); let camera = guest.create_broadcast("alice/camera.hang").unwrap(); camera.announce(moq_net::origin::Route::default()).unwrap(); let renewals = moq_tokio::origin::spawn(); - let alice = page - .init(Default::default()) - .unwrap() - .with_publisher(&guest) - .with_subscriber(renewals.clone()) - .connect(Url::parse(&fragment["url"]).unwrap()); + let alice = relay.join(&room.invite("alice").unwrap().link, &guest, &renewals); let event = kio::clock::timeout(TIMEOUT, announced.next()) .await @@ -1253,10 +1359,7 @@ mod tests { room.invite("alice").unwrap(); renewals.sync(); let back = key.verify(&next_token(&consumer, "alice", 1).await.unwrap()).unwrap(); - assert_eq!( - back.subscribe, - [".renew/alice/1".parse().unwrap()].into_iter().collect() - ); + assert_eq!(back.subscribe, subscribes("alice", 1)); let before = consumer.request(Room::renewal_path("alice", 0).unwrap(), None).await; assert!(before.is_err(), "the earlier generation is still renewed"); diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index f08ee87e..5b8cbebb 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -13,6 +13,7 @@ use anyhow::Context; mod autosave; mod devices; mod engine; +mod feed; mod guests; mod health; mod interact; diff --git a/rs/moq-studio/src/mix.rs b/rs/moq-studio/src/mix.rs index f36968e8..62b623db 100644 --- a/rs/moq-studio/src/mix.rs +++ b/rs/moq-studio/src/mix.rs @@ -1,5 +1,5 @@ -//! Outputs that read one of the collection's mixes: the program audio and -//! the headphone monitor. +//! Outputs that read one of the collection's mixes: the program audio, the +//! headphone monitor, and each guest's return feed, which reads a mix-minus. //! //! Each output follows a [`Pick`], which the program clock sets from the //! collection, and reads the picked mix on its own thread or task. A mix's @@ -9,7 +9,37 @@ use std::sync::{Arc, Mutex}; use std::task::{Poll, ready}; -use moq_audio::bus::{Block, Mix, Reader}; +use moq_audio::bus::{Block, Minus, Mix, Reader}; + +/// What an output reads: one of the collection's mixes, or a mix-minus of one. +#[derive(Clone)] +pub enum Mixed { + /// A mix, as the collection names it. + Mix(Mix), + /// A mix without one of its sources. + Minus(Minus), +} + +impl Mixed { + fn reader(&self) -> Reader { + match self { + Self::Mix(mix) => mix.reader(), + Self::Minus(minus) => minus.reader(), + } + } +} + +impl From for Mixed { + fn from(mix: Mix) -> Self { + Self::Mix(mix) + } +} + +impl From for Mixed { + fn from(minus: Minus) -> Self { + Self::Minus(minus) + } +} /// Which mix an output reads, as last picked. #[derive(Clone, Default)] @@ -20,13 +50,14 @@ struct Picked { /// How many times a mix was picked, so a reader sees each pick once. generation: u64, /// The mix, and a reader starting at the first block after the pick. - mix: Option<(Mix, Reader)>, + mix: Option<(Mixed, Reader)>, } impl Pick { /// Read `mix` from its next block on. Picking the mix already read /// changes nothing a reader hears. - pub fn set(&self, mix: Mix) { + pub fn set(&self, mix: impl Into) { + let mix = mix.into(); let reader = mix.reader(); // Fails only once every handle is gone, when nothing reads it. if let Ok(mut picked) = self.0.write() { @@ -65,10 +96,10 @@ pub struct Follow { end: u64, /// The mix being read, held so it outlives a pick that drops it from the /// collection until the next one takes over. - current: Option<(Mix, Reader)>, + current: Option<(Mixed, Reader)>, /// The mix picked since, with its first block once read, until the /// current one has caught up to it. - next: Option<(Mix, Reader, Option)>, + next: Option<(Mixed, Reader, Option)>, } impl Follow { @@ -419,7 +450,7 @@ mod tests { { let mut picked = pick.0.write().ok().unwrap(); picked.generation += 1; - picked.mix = Some((b.clone(), late)); + picked.mix = Some((b.clone().into(), late)); } step(&mut driver); let next = ready(&mut follow).unwrap(); From 5fcef3edc4ec7ca3768be4f9620b622dea072f41 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 11:02:44 -0800 Subject: [PATCH 3/8] feat(join): play the return feed, and document it Co-Authored-By: Claude Opus 5.5 --- README.md | 30 +++++++++++++++++++++++------- bun.lock | 1 + js/join/index.html | 2 ++ js/join/package.json | 3 ++- js/join/src/feed.ts | 17 +++++++++++++++++ js/join/src/index.css | 5 +++++ js/join/src/index.ts | 13 ++++++++++++- quest/m0/README.md | 1 - quest/m0/guest-return.md | 25 ------------------------- 9 files changed, 62 insertions(+), 35 deletions(-) create mode 100644 js/join/src/feed.ts delete mode 100644 quest/m0/guest-return.md diff --git a/README.md b/README.md index a94fa40a..04ab3f33 100644 --- a/README.md +++ b/README.md @@ -226,7 +226,8 @@ program. ## Guests Guests and phones join a room the studio owns on its relay, from a web page, -and their camera and screen appear as broadcast sources. The room follows +and their camera and screen appear as broadcast sources. The page plays the +show back to them, without their own voice. The room follows [moq-room](https://github.com/moq-dev/moq/tree/main/rs/moq-room): a guest named `alice` publishes `alice/camera.hang` and `alice/screen.hang` under it. Guests need a relay that verifies tokens, so run your own with @@ -260,9 +261,10 @@ moq-studio show.toml --connect 'https://relay.example.com/show?jwt=...' \ and each token the studio renews it with. While a guest is in the room, the studio signs them a fresh token every half of it, over a session of its own in the room that the room key signs for, and the join page presents it - without reconnecting. A guest's token may read only their own renewals, at - `.renew//` in the room, but anything else the relay lets subscribe - there can read every guest's, so grant that only to the studio. A guest who + without reconnecting. A guest's token may read only their own renewals and + return feed, at `.renew//` and `.return//` + in the room, and the program at `.program`, but anything else the relay lets + subscribe there can read every guest's, so grant that only to the studio. A guest who leaves needs a link, or a renewed token, that has not expired to come back. The Guests panel says when renewal is failing. - **Upgrading from a studio without renewal:** generate a new room key, drop @@ -283,7 +285,7 @@ moq-studio show.toml --connect 'https://relay.example.com/show?jwt=...' \ a QR code; **Copy link** puts it on the clipboard instead. The guest's token is in the link's fragment, which the browser never sends to a server, and lets them publish only under their own name, so they cannot appear as someone else, -and read only their own renewals. +and read only their own renewals and return feed. Treat the link like a password: anyone holding it can join as that guest before it expires, and is then renewed for as long as they stay, until you remove them. A guest's broadcasts become sources named after them (`alice`, @@ -297,13 +299,26 @@ who reconnects after you removed their source comes back under a new name **Remove from scenes** is your edit, so it undoes: it takes a guest, any clone of their sources, and any mask reading them out of every scene and the top layer and mutes them. A signed token cannot be taken back, so it also stops -renewing theirs, and they are disconnected within `--guest-expiry`. That +renewing theirs and ends their return feed, and they are disconnected within +`--guest-expiry`. That holds across a restart, whichever scene file you open: removed guests are kept beside the room key, as `room.guests.toml` for `room.jwk`, so its directory must be writable. Undo puts them back in the scenes but not on renewal; **New link** lets them back in and renews them again, under a new generation, so a link from before the removal cannot read their renewals. +While a guest is in the room, the join page plays their return feed: the +program's video, and the mix on program without that guest's microphone (a +mix-minus), so they follow the show without hearing themselves back. Ask +guests to wear headphones, so the show does not reach their microphone. The +studio publishes it over its session in the room, never with its own +credentials: the program once at `.program`, and each guest's feed at +`.return//`. The video is the program's own encode, which +this session uploads a second time while any guest watches it; each guest +in the room costs one 64 kbps Opus encode of their mix-minus. A guest's +screen share stays in their feed, since only their camera's microphone is +taken out. + ## Scene file The collection is TOML, so it diffs and reviews like code. It names devices, @@ -666,7 +681,8 @@ just join # Serve the join page locally ``` `rs/moq-scene` is the UI-free compositor, `rs/moq-studio` the app, and -`js/join` the join page, built on `@moq/room` and `@moq/publish`. Like the +`js/join` the join page, built on `@moq/room`, `@moq/publish`, and +`@moq/watch`. Like the Rust crates, those come from the `moq` submodule rather than npm, so `bun install` needs it checked out, and Nix to build `@moq/libav`. See [AGENTS.md](AGENTS.md) and [CONTRIBUTING.md](CONTRIBUTING.md), and the diff --git a/bun.lock b/bun.lock index e402ed0f..0e3bbc08 100644 --- a/bun.lock +++ b/bun.lock @@ -16,6 +16,7 @@ "@moq/publish": "workspace:*", "@moq/room": "workspace:*", "@moq/signals": "workspace:*", + "@moq/watch": "workspace:*", }, "devDependencies": { "@types/bun": "^1.4.4", diff --git a/js/join/index.html b/js/join/index.html index c7cbadff..e63f610b 100644 --- a/js/join/index.html +++ b/js/join/index.html @@ -23,6 +23,8 @@

Join the studio