diff --git a/README.md b/README.md index b056c17..7353c6e 100644 --- a/README.md +++ b/README.md @@ -30,6 +30,10 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ the first relay only. The stream's tracks claim their bitrate on the first relay's link alone, so only that link is probed toward the stream's rate, and the video follows that link's estimate alone. +- Quitting while live ends every track after its last pictures and sound, + then closes each relay session once the relay has them, waiting up to about + a second, so viewers and recordings on the relay see the broadcast end + rather than cut off. - `capture` adds camera, display, window, microphone, and system-audio sources, and listening to a mix. On Linux it links ALSA; add `pipewire` for PipeWire cameras. `openh264` is the software H.264 fallback for a machine without a hardware encoder. diff --git a/quest/m0/README.md b/quest/m0/README.md index d56814e..94c8e65 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -101,7 +101,6 @@ repository. - [Browser sources on macOS and Windows](/quest/m0/browser-source.md) - the out-of-process CEF pages Linux has, with shared GPU textures on the other desktops - [Web studio](/quest/m0/web.md) - the full studio in wasm on WebGPU and WebCodecs, served from this repository - [Virtual camera](/quest/m0/vcam/README.md) - a canvas's program exposed as a system camera on Linux, Windows, and macOS -- [Clean quit](/quest/m0/clean-quit.md) - quitting finishes every track and closes the relay session, so the broadcast's tail arrives - [Recover crashed recordings](/quest/m0/recording-repair.md) - a recording a crash left unfinished is repaired beside the original at launch - [Premultiplied source sampling](/quest/m0/premultiply.md) - scaled transparent sources draw without edge fringes - [Deterministic browser source startup](/quest/m0/cef-startup.md) - CEF on the CPU path paints its first frame without the GPU fallback hang diff --git a/quest/m0/clean-quit.md b/quest/m0/clean-quit.md deleted file mode 100644 index 7b6e99e..0000000 --- a/quest/m0/clean-quit.md +++ /dev/null @@ -1,27 +0,0 @@ -# [S] Clean quit - -## Goal - -Quitting the studio delivers everything it published before the process -exits: the program's video and audio and each source's own track end -cleanly, and the relay session closes gracefully, so a viewer or a recording -on the relay receives the tail of the broadcast instead of a cut. -A worker that fails to start rolls back what it already started, rather -than leaving threads or tracks running. - -## Plan - -From [moq-dev/studio#23](https://github.com/moq-dev/studio/pull/23) and -[moq-dev/studio#22](https://github.com/moq-dev/studio/pull/22). - -- `Engine::stop` (`rs/moq-studio/src/engine.rs`) drains and finishes the - program video, but `main` then returns and drops the runtime, so the tail - may never reach the relay. -- `moq_tokio::Connection::close` drains, bounded at one second, once every - track is finished: program audio, and each source's track. Source threads - are not stopped at exit today, so stop them and finish their tracks first. -- A worker that fails partway through starting (the stop signal raised on - #10 and #22) stops the parts it already started before reporting the - error. -- Test on a paused `kio::clock`: quit while live to a local relay, and a - subscriber receives the last group of every track and then the end. diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 2eb6a92..7544385 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -901,6 +901,99 @@ mod tests { .await; } + /// Stopping the sources, as quitting does, ends each source's own track + /// cleanly after what it published, and takes it out of the listing. + #[tokio::test] + async fn stopping_the_sources_finishes_their_tracks() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (publish, catalog, consumer) = studio(); + let (mut context, _origin) = context(dir.path()); + context.publish = Some(publish); + let mut collection = Collection::new(); + collection.sources.insert( + "red".into(), + Source::Color { + color: Rgba([255, 0, 0, 255]), + }, + ); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let snapshot = listed(&catalog, |sources| sources.contains_key("red")).await; + let (name, _) = snapshot.ext.studio.sources["red"].binary.tracks.iter().next().unwrap(); + let mut track = consumer.track(name).unwrap().subscribe(None).await.unwrap(); + + sources.stop(); + assert!(sources.get("red").is_none()); + let mut group = track.recv_group().await.unwrap().expect("the still"); + assert!(group.read_frame().await.unwrap().is_some()); + assert!(group.read_frame().await.unwrap().is_none()); + assert!(track.recv_group().await.unwrap().is_none(), "the track finished"); + listed(&catalog, |sources| sources.is_empty()).await; + }) + .await; + } + + /// Stopping the sources ends a playing media file's picture and sound + /// tracks cleanly after what they published. + #[tokio::test(start_paused = true)] + async fn stopping_the_sources_finishes_a_media_files_tracks() { + local(async { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("bed.mkv"), crate::media::tests::CLIP).unwrap(); + let (publish, catalog, consumer) = studio(); + publish.set_on_program(OnProgram { + shown: ["bed".to_string()].into(), + heard: ["bed".to_string()].into(), + }); + let (mut context, _origin) = context(dir.path()); + context.publish = Some(publish.clone()); + let mut collection = Collection::new(); + let bed = Source::Media { + path: "bed.mkv".into(), + looping: true, + restart: false, + }; + collection.sources.insert("bed".into(), bed); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + + let bed = until(|| { + let listed = catalog.snapshot().ext.studio.sources.remove("bed"); + listed.filter(|listed| !listed.video.renditions.is_empty() && !listed.audio.renditions.is_empty()) + }) + .await; + let names = [ + bed.video.renditions.keys().next().unwrap().clone(), + bed.audio.renditions.keys().next().unwrap().clone(), + ]; + let mut tracks = Vec::new(); + for name in &names { + let subscription = + moq_net::track::Subscription::default().with_delay(moq_net::track::Delay::default().with_max(None)); + tracks.push(consumer.track(name).unwrap().subscribe(subscription).await.unwrap()); + } + let written = |id: &str| publish.stats().get(id).map_or(0, |rendition| rendition.frames); + pictures(sources.get("bed").unwrap(), kio::clock::now(), 4).await; + until(|| names.iter().all(|name| written(name) > 0).then_some(())).await; + + sources.stop(); + for (name, mut track) in names.iter().zip(tracks) { + let mut frames = 0; + while let Some(mut group) = track.recv_group().await.expect("a clean end") { + while group.read_frame().await.expect("a whole group").is_some() { + frames += 1; + } + } + assert!(frames > 0, "{name} published nothing"); + } + listed(&catalog, |sources| sources.is_empty()).await; + }) + .await; + } + /// A mid-gray picture of `size` at `millis` on a capture's own timeline. fn gray(size: moq_video::Size, millis: u64) -> moq_video::Frame { let i420 = moq_video::I420::new(size, vec![0x80; moq_video::I420::len(size).unwrap()]).unwrap(); diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index db4aafd..8c1a0ab 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -284,6 +284,15 @@ impl Sources { } } + /// Stop every source, as removing it would: each [`Driver`] returns and + /// its own track finishes cleanly with what it published. The mixes stay, + /// and the next [`sync`](Self::sync) starts the sources again. + pub fn stop(&mut self) { + for feed in std::mem::take(&mut self.feeds).into_values() { + leave(&self.mixes, &feed); + } + } + /// The running source called `name`. pub fn get(&self, name: &str) -> Option<&Feed> { self.feeds.get(name) diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 4cdc6b7..c26b25e 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -37,6 +37,7 @@ use moq_scene::publish::{Encoding, Sessions}; use moq_scene::source::{self, Driver, Sources, Status}; use moq_scene::{Canvas, Collection, Compositor, Draft, Edit, History, Scaler, wgpu}; use moq_video::render::export::{Exporter, Pending, Route}; +use tokio::sync::oneshot::error::TryRecvError; use crate::autosave::{self, Autosave}; use crate::ladder::{self, Gate, Joins, Step}; @@ -473,7 +474,29 @@ pub struct Output { pub on_demand: bool, /// A connection to each relay, which reconnects on its own and dials only /// while held. The Health panel reads their states. - _connections: Vec, + connections: Vec, + /// The client that dialed each, whose endpoint sends the last close. + clients: Vec, +} + +impl Output { + /// Close each relay session once the relay has received what it was sent + /// of every finished track, which the session bounds at about a second, + /// then each client's endpoint, so the relay hears the close rather than + /// waiting out an idle timeout. A track still live is cut. + async fn close(&self) { + let mut closing = tokio::task::JoinSet::new(); + for (connection, client) in self.connections.iter().zip(&self.clients) { + let (connection, client) = (connection.clone(), client.clone()); + closing.spawn(async move { + if let Err(err) = connection.close().await { + tracing::warn!(%err, "the relay session closed without draining"); + } + client.close().await; + }); + } + closing.join_all().await; + } } /// A relay the program is published to. @@ -611,6 +634,8 @@ pub struct Engine { /// The program clock's and the audio's threads, which [`stop`](Self::stop) /// waits on. threads: Mutex>>, + /// Each source thread's end, which [`close`](Self::close) waits on. + running: Mutex>>, } /// The audio the program publishes. @@ -620,6 +645,11 @@ pub const AUDIO_BITRATE: moq_net::bandwidth::Rate = moq_net::bandwidth::Rate::fr /// close to the bus's buffering, long enough to cost nothing. 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. +/// Each relay session bounds its own drain at about a second. +const CLOSE: Duration = Duration::from_secs(3); + /// Everything starting the engine takes besides the scene file, which guided /// setup holds back until the creator has settled on a collection. #[derive(Clone)] @@ -686,9 +716,11 @@ impl Engine { } else { tracing::info!(broadcast = %path, "publishing"); let reports = subscribe.consume(); + let clients = relays.iter().map(|relay| relay.client.clone()).collect(); let (connections, links) = connect(relays, &origin, &subscribe); let live = Live { connections, + clients, links, reports, on_demand, @@ -707,7 +739,8 @@ impl Engine { let output = live.map(|live| Output { path: path.to_string(), on_demand, - _connections: live.connections, + connections: live.connections, + clients: live.clients, }); // Absolute, so a copied image names the same file in another collection. @@ -751,6 +784,7 @@ impl Engine { _origin: subscribe, stopped: AtomicBool::new(false), threads: Mutex::default(), + running: Mutex::default(), }); let Publish { renditions, @@ -798,6 +832,10 @@ impl Engine { let opened = opened.inspect_err(|err| video.fail(format!("{err:#}"))).ok(); lines.insert(name, opened); } + // Each worker holds the engine, so one that cannot start stops those + // already running, which would otherwise keep this engine and its + // publication alive beside a retry's. + let starting = Starting(Some(engine.clone())); let program = engine.clone(); let thread = clocked("program".into(), async move || { program.run_program(start, epoch, main, extras, lines).await @@ -827,6 +865,7 @@ impl Engine { } let save = engine.clone(); clocked("autosave".into(), async move || save.autosave.run(&save.state).await)?; + starting.started(); tokio::spawn(engine.clone().grade(health::Sampler::new(health))); // Last, once nothing can fail: a start that fails is retried from setup, // and a follower already running would keep this engine and its @@ -850,13 +889,15 @@ impl Engine { Ok(engine) } - /// Stop the program clock, then wait for each rendition's encoder to - /// publish the copies on their way to it and what it still holds, and to - /// end its video track, so quitting loses none of the program's last - /// pictures. The program audio ends its track too, so a recording reading - /// the program reaches the end of each. Then stop the browser sources' - /// helper and wait for it to exit, so their profile is saved and free for - /// the next launch. + /// Stop the program clock, then wait for each rendition's encoder, on + /// every canvas, to publish the copies on their way to it and what it + /// still holds, and to end its video track, so quitting loses none of the + /// program's last pictures; a further canvas's catalog then finishes too. + /// The program audio ends its track too, so a recording reading + /// the program reaches the end of each. Then stop every source, which + /// finishes its own track on its thread, and the browser sources' helper, + /// waiting for it to exit, so their profile is saved and free for the next + /// launch. pub fn stop(&self) { self.stopped.store(true, Ordering::Relaxed); let threads = std::mem::take(&mut *self.threads.lock().unwrap()); @@ -867,12 +908,42 @@ impl Engine { } } // Taken out first, so the wait holds no lock the window draws with. - let browser = self.state.lock().unwrap().sources.context().browser.clone(); + let browser = { + let mut state = self.state.lock().unwrap(); + state.sources.stop(); + state.sources.context().browser.clone() + }; if let Some(browser) = browser { browser.stop(); } } + /// Once [`stop`](Self::stop) has ended the tracks of every canvas, close + /// each relay session once the relay has their tails, so a viewer or a + /// recording on the relay receives them rather than a cut. Meanwhile wait + /// for each source's thread to end, having finished its own track. Bounded + /// at [`CLOSE`]. + pub async fn close(&self) { + let running = std::mem::take(&mut *self.running.lock().unwrap()); + let sources = async { + for ended in running { + // Fails once the thread ends, which is all it says. + let _ = ended.await; + } + }; + let relays = async { + if let Some(output) = &self.output { + output.close().await; + } + }; + // A source's track is in the studio's own local broadcast, which no + // relay session carries, so neither waits on the other. + let closing = async { tokio::join!(sources, relays) }; + if kio::clock::timeout(CLOSE, closing).await.is_err() { + tracing::warn!(timeout = ?CLOSE, "quitting before every source stopped and every relay drained"); + } + } + /// Write the collection and its history to their files unless they /// already hold them, or the scene file changed outside the studio. pub fn save(&self) -> anyhow::Result<()> { @@ -1041,8 +1112,13 @@ impl Engine { if self.stopped.load(Ordering::Relaxed) { // Without a sender, each encoder publishes what it was sent and // still holds, then finishes its track. - retired.push(main.end()); - retired.extend(lines.into_values().flatten().map(|(line, _)| line.end())); + retired.push(main.end(None)); + retired.extend( + lines + .into_values() + .flatten() + .map(|(line, broadcast)| line.end(Some(broadcast))), + ); for ended in retired { ended.join(); } @@ -1052,8 +1128,12 @@ impl Engine { let mut guard = self.state.lock().unwrap(); let state = &mut *guard; if std::mem::take(&mut state.stale) { - for driver in state.sources.sync(&state.collection) { - spawn(driver); + let drivers = state.sources.sync(&state.collection); + if !drivers.is_empty() { + let mut running = self.running.lock().unwrap(); + // Forgets the threads that have ended. + running.retain_mut(|ended| ended.try_recv() == Err(TryRecvError::Empty)); + running.extend(drivers.into_iter().map(spawn)); } self.pick_mixes(state); } @@ -1089,8 +1169,8 @@ impl Engine { .cloned() .collect(); for name in gone { - if let Some(Some((line, _))) = lines.remove(&name) { - retired.push(line.end()); + if let Some(Some((line, broadcast))) = lines.remove(&name) { + retired.push(line.end(Some(broadcast))); } self.stats.canvases.lock().unwrap().remove(&name); } @@ -1203,6 +1283,25 @@ impl Engine { } } +/// An engine whose workers are starting. Unless every one +/// [started](Self::started), dropping it stops those that did, waiting for +/// them, which takes about a frame and the encoders' tails. +struct Starting(Option>); + +impl Starting { + fn started(mut self) { + self.0 = None; + } +} + +impl Drop for Starting { + fn drop(&mut self) { + if let Some(engine) = self.0.take() { + engine.stop(); + } + } +} + /// One encoder of a mix: the published program audio, or the recording /// broadcast's own. enum Audio { @@ -1214,6 +1313,8 @@ enum Audio { /// the origin their viewers' reports and the relays' stats arrive on. struct Live { connections: Vec, + /// The client that dialed each connection. + clients: Vec, links: Vec, reports: moq_net::origin::Consumer, /// The canvas size encodes only while someone watches it. @@ -1717,6 +1818,9 @@ struct Ended { threads: Vec>, /// Holds the encode threads it was waiting on. probe: Option>>, + /// A further canvas's broadcast, dropped only after the threads, so its + /// catalog finishes after the video it lists. + _broadcast: Option, } impl Ended { @@ -1834,8 +1938,9 @@ impl Line { } /// Stop handing the renditions frames, so each encode thread publishes - /// what it holds and finishes its track. - fn end(mut self) -> Ended { + /// what it holds and finishes its track, then ends a further canvas's + /// `broadcast`. + fn end(mut self, broadcast: Option) -> Ended { let mut threads = std::mem::take(&mut self.ending); threads.extend( self.ladder @@ -1848,6 +1953,7 @@ impl Line { Ended { threads, probe: self.probing.map(|(_, probe)| probe), + _broadcast: broadcast, } } } @@ -1856,15 +1962,18 @@ impl Line { /// to start, which is not retried until the canvas is removed and added again. type Lines = BTreeMap>; -/// A further canvas's broadcast, closed when dropped. +/// A further canvas's broadcast, whose catalog finishes and which closes +/// when dropped. struct Broadcast { publication: moq_net::origin::Publication, /// Lists the canvas's video and the main broadcast's audio. - _catalog: moq_scene::publish::Catalog, + catalog: moq_scene::publish::Catalog, } impl Drop for Broadcast { fn drop(&mut self) { + // Fails only once the catalog failed, which ended its track already. + let _ = self.catalog.finish(); self.publication.close(); } } @@ -1986,10 +2095,7 @@ impl Extras { video, grants: format!("{name}/"), }; - let broadcast = Broadcast { - publication, - _catalog: catalog, - }; + let broadcast = Broadcast { publication, catalog }; Ok((renditions, broadcast)) } } @@ -2428,13 +2534,16 @@ 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. -fn spawn(driver: Driver) { +/// 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<()> { // 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); + let (running, ended) = tokio::sync::oneshot::channel::<()>(); let name = format!("source-{}", driver.name()); let started = clocked(name, async move || { + let _running = running; if let Ok(driver) = received.recv() { driver.run().await; } @@ -2447,6 +2556,7 @@ fn spawn(driver: Driver) { } Err(err) => driver.fail(format!("{err:#}")), } + ended } /// Start a thread called `name` that runs `task` with its own clock and IO, @@ -2740,10 +2850,60 @@ mod tests { .await } + /// The broadcast of the further canvas `tall` as `relay` routes it, its + /// catalogs, and the first that lists its video. + async fn routed_tall( + relay: &moq_net::origin::Consumer, + ) -> ( + moq_net::broadcast::Consumer, + moq_mux::catalog::Consumer<()>, + hang::Catalog, + ) { + // Until the canvas's broadcast is announced, the main one's route covers + // its path and cannot serve it; the follow moves on once it is. + let mut following = relay.follow("studio/tall", None).unwrap(); + let (broadcast, mut catalogs) = kio::clock::timeout(TIMEOUT, async { + loop { + let broadcast = following.available().await.expect("broadcast"); + let format = moq_mux::catalog::CatalogFormat::default(); + if let Ok(catalogs) = moq_mux::catalog::Consumer::<()>::new(&broadcast, format).await { + return (broadcast, catalogs); + } + } + }) + .await + .expect("the relay routes the canvas"); + // Audio is listed from the start, and the video once its encoder opens. + let catalog = kio::clock::timeout(TIMEOUT, async { + loop { + let catalog = catalogs.next().await.expect("catalog").expect("a catalog"); + if !catalog.video.renditions.is_empty() { + return catalog; + } + } + }) + .await + .expect("the canvas lists its video"); + (broadcast, catalogs, catalog) + } + + /// Read a further canvas's `video` and `catalogs` to a clean end. + async fn ended(mut catalogs: moq_mux::catalog::Consumer<()>, mut video: moq_net::track::Subscriber) { + kio::clock::timeout(TIMEOUT, async { + while let Some(mut group) = video.recv_group().await.expect("the video finishes") { + while group.read_frame().await.expect("the video finishes").is_some() {} + } + while catalogs.next().await.expect("the catalog finishes").is_some() {} + }) + .await + .expect("the canvas's tracks end"); + } + /// A further canvas reaches the relay in a broadcast of its own beneath the /// main one, at its own size, with the main broadcast's audio listed by /// reference rather than encoded again, and holds one more encode session. - /// Removing it ends its broadcast and gives its session back. + /// Removing it, or quitting, ends its video and then its catalog cleanly, + /// then its broadcast; removing it gives its session back. #[tokio::test(flavor = "multi_thread")] #[ignore = "needs a Vulkan device and an H.264 encoder; run by just rs render-sw"] async fn the_program_of_a_further_canvas_has_its_own_broadcast() { @@ -2776,61 +2936,58 @@ mod tests { .await .expect("the studio starts"); - // Until the canvas's broadcast is announced, the main one's route covers - // its path and cannot serve it; the follow moves on once it is. - let mut following = relay.follow("studio/tall", None).unwrap(); - let (broadcast, mut catalogs) = kio::clock::timeout(TIMEOUT, async { - loop { - let broadcast = following.available().await.expect("broadcast"); - let format = moq_mux::catalog::CatalogFormat::default(); - if let Ok(catalogs) = moq_mux::catalog::Consumer::<()>::new(&broadcast, format).await { - return (broadcast, catalogs); - } - } - }) - .await - .expect("the relay routes the canvas"); - // Audio is listed from the start, and the video once its encoder opens. - let catalog = kio::clock::timeout(TIMEOUT, async { - loop { - let catalog = catalogs.next().await.expect("catalog").expect("a catalog"); - if !catalog.video.renditions.is_empty() { - return catalog; - } - } - }) - .await - .expect("the canvas lists its video"); + let (broadcast, catalogs, catalog) = routed_tall(&relay).await; let (_, audio) = catalog.audio.renditions.iter().next().expect("the program audio"); let main = audio.broadcast.as_ref().expect("a reference to the main broadcast"); let main = moq_net::Path::new("studio/tall").unwrap().try_resolve(main); assert_eq!(main.as_ref().map(moq_net::Path::as_str), Some("studio")); assert!(audio.instance.is_some(), "the main broadcast's instance"); - let (video, config) = catalog - .video - .renditions - .iter() - .find(|(_, config)| config.coded_width == Some(360)) - .expect("the canvas size"); - assert_eq!(config.coded_height, Some(640)); - let mut pictures = subscribe(&broadcast, video, config).await; + let canvas_size = |catalog: &hang::Catalog| { + let (video, config) = catalog + .video + .renditions + .iter() + .find(|(_, config)| config.coded_width == Some(360)) + .expect("the canvas size"); + assert_eq!(config.coded_height, Some(640)); + (video.clone(), config.clone()) + }; + let (video, config) = canvas_size(&catalog); + let mut pictures = subscribe(&broadcast, &video, &config).await; red(&mut pictures, moq_video::Size::new(360, 640)) .await .expect("the canvas's own layout arrives"); assert_eq!(engine.stats.sessions.open(), 2, "one canvas size each"); assert!(engine.stats.canvases()["tall"].error().is_none()); + let track = broadcast.track(&video).unwrap().subscribe(None).await.unwrap(); { let mut state = engine.state.lock().unwrap(); let mut draft = state.collection().draft("Remove the canvas"); draft.remove_extra("tall").unwrap(); state.commit(draft, None).unwrap(); } + ended(catalogs, track).await; kio::clock::timeout(TIMEOUT, broadcast.closed()) .await .expect("the canvas's broadcast ends"); assert!(engine.stats.canvases().is_empty()); - engine.stop(); + assert_eq!(engine.stats.sessions.open(), 1, "the canvas's session given back"); + + // Back again, quitting ends it as cleanly. + engine.state.lock().unwrap().undo().unwrap(); + let (broadcast, catalogs, catalog) = routed_tall(&relay).await; + let (video, _) = canvas_size(&catalog); + let track = broadcast.track(&video).unwrap().subscribe(None).await.unwrap(); + let stopping = engine.clone(); + tokio::task::spawn_blocking(move || stopping.stop()).await.unwrap(); + kio::clock::timeout(TIMEOUT, engine.close()) + .await + .expect("the studio closes"); + ended(catalogs, track).await; + kio::clock::timeout(TIMEOUT, broadcast.closed()) + .await + .expect("the canvas's broadcast ends"); assert_eq!(engine.stats.sessions.open(), 0); } @@ -3137,18 +3294,25 @@ mod tests { .expect("a clip"); assert_eq!(clip, dir.path().join("show-clip-1.mp4")); - // Quitting ends the program's video cleanly once the encoder published - // what it was handed, rather than leaving viewers a dropped track. - let mut track = broadcast.track(&resized).unwrap().subscribe(None).await.unwrap(); + // Quitting ends the program's video and audio cleanly once the encoders + // published what they were handed, and closes the relay session only + // once the relay has it, rather than leaving viewers dropped tracks. + let video = broadcast.track(&resized).unwrap().subscribe(None).await.unwrap(); + let sound = broadcast.track(audio).unwrap().subscribe(None).await.unwrap(); let stopping = engine.clone(); tokio::task::spawn_blocking(move || stopping.stop()).await.unwrap(); - kio::clock::timeout(TIMEOUT, async { - while let Some(mut group) = track.recv_group().await.expect("the video finishes") { - while group.read_frame().await.expect("the video finishes").is_some() {} - } - }) - .await - .expect("the video ends"); + kio::clock::timeout(TIMEOUT, engine.close()) + .await + .expect("the studio closes"); + for mut track in [video, sound] { + kio::clock::timeout(TIMEOUT, async { + while let Some(mut group) = track.recv_group().await.expect("the track finishes") { + while group.read_frame().await.expect("the track finishes").is_some() {} + } + }) + .await + .expect("the track ends"); + } let failed = engine.stats.error(); assert!(failed.is_none(), "{failed:?}"); } @@ -3273,6 +3437,129 @@ mod tests { .expect("both relays route the program, the failover the sources"); } + /// Closing the output as quitting does delivers the tail of every finished + /// track before the relay session goes: a subscriber at the relay reads + /// each track's last group whole and then its end, and the broadcast is + /// withdrawn. Each last group is still on its way as the close starts, so + /// dropping the session instead cuts it. + #[tokio::test(start_paused = true)] + async fn closing_delivers_every_tail() { + use moq_net::announce::Event; + + /// The last group's frames, each padded to this many bytes. + const FRAMES: usize = 10; + const PADDING: usize = 50_000; + + let _ = moq_tokio::crypto::install_default(); + let fixture = moq_relay::test_relay().await.expect("test relay"); + let ready = fixture.relay.ready(); + let relay = fixture.relay.cluster().origin.consume(); + tokio::spawn(fixture.relay.run()); + ready.wait().await.expect("relay ready"); + let mut client = moq_tokio::connect::Config::default(); + client.tls.fingerprint = vec![fixture.fingerprint.clone()]; + let client = client.init(Default::default()).expect("client"); + + let origin = moq_tokio::origin::spawn(); + let (connections, _links) = connect( + vec![Relay { + client: client.clone(), + url: fixture.url.clone(), + }], + &origin, + &moq_tokio::origin::spawn(), + ); + let output = Output { + path: "studio".into(), + on_demand: false, + connections, + clients: vec![client], + }; + let publication = origin.create_broadcast("studio").unwrap(); + publication.announce(Default::default()).unwrap(); + let info = || moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI); + let tracks = ["video", "audio"].map(|name| publication.create_track(name, info()).unwrap()); + + // Wait for an event `want` picks on `announced`. + let routed = async |announced: &mut moq_net::announce::Consumer, want: fn(&Event) -> bool| { + kio::clock::timeout(TIMEOUT, async { + while !want(&announced.next().await.expect("announcements")) {} + }) + .await + }; + let mut announced = relay.announced(); + routed( + &mut announced, + |event| matches!(event, Event::Start(announce) if announce.prefix.as_str() == "studio"), + ) + .await + .expect("the relay routes the broadcast"); + let broadcast = kio::clock::timeout(TIMEOUT, relay.follow("studio", None).unwrap().available()) + .await + .expect("the relay routes the broadcast") + .expect("broadcast"); + let mut subscriptions = Vec::new(); + for (name, track) in ["video", "audio"].into_iter().zip(&tracks) { + let mut subscription = broadcast.track(name).unwrap().subscribe(None).await.unwrap(); + // A head group read through the relay, so only the tail is at stake. + let mut group = track.append_group().unwrap(); + group.write_frame(moq_net::Timestamp::ZERO, vec![0; 1]).unwrap(); + group.finish().unwrap(); + let mut head = kio::clock::timeout(TIMEOUT, subscription.recv_group()) + .await + .expect("the head group arrives") + .unwrap() + .expect("a group"); + while head.read_frame().await.unwrap().is_some() {} + subscriptions.push(subscription); + } + + // Read as a viewer does, while the studio writes and closes. + let readers: Vec<_> = subscriptions + .into_iter() + .map(|mut subscription| { + tokio::spawn(async move { + let mut tail = subscription + .recv_group() + .await + .expect("the tail, not a cut") + .expect("a group"); + let mut frames = 0; + while let Some(frame) = tail.read_frame().await.expect("the whole tail") { + assert_eq!(frame.payload.len(), PADDING); + frames += 1; + } + assert_eq!(frames, FRAMES); + let end = subscription.recv_group().await.expect("a clean end"); + assert!(end.is_none(), "the track ends after its tail"); + }) + }) + .collect(); + for track in &tracks { + let mut group = track.append_group().unwrap(); + for _ in 0..FRAMES { + group.write_frame(moq_net::Timestamp::ZERO, vec![0; PADDING]).unwrap(); + } + group.finish().unwrap(); + track.finish().unwrap(); + } + kio::clock::timeout(TIMEOUT, output.close()) + .await + .expect("the output closes"); + for reader in readers { + kio::clock::timeout(TIMEOUT, reader) + .await + .expect("the tail arrives") + .unwrap(); + } + routed( + &mut announced, + |event| matches!(event, Event::End(announce) if announce.prefix.as_str() == "studio"), + ) + .await + .expect("the broadcast is withdrawn"); + } + /// Wait until `announced` routes `show` under `instance`. async fn routed_at(announced: &mut moq_net::announce::Consumer, instance: &moq_net::Instance) { use moq_net::announce::Event; diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index a494d5b..ca72f40 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -200,7 +200,9 @@ fn main() -> anyhow::Result { // The edits since the last autosave, so quitting loses none. let saved = engine.save(); engine.stop(); - runtime.block_on(engine.recorder.done()); + // Before the runtime goes, which would drop each relay session with the + // tail of the broadcast still on its way. + runtime.block_on(async { tokio::join!(engine.close(), engine.recorder.done()) }); if let (Err(err), Err(_)) = (&saved, &ran) { tracing::error!(error = format!("{err:#}"), "cannot save on quit"); }