From e0281bef856c4a64e113594abb0034c92a0d08b3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 7 Nov 2026 07:34:05 -0800 Subject: [PATCH 1/2] chore(quest): claim quest/m0/catalog-finish.md Quest-Claim: 05a842c4b0abf435 From 66d36a463032a1157e21bf05af6a66e3668152b2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 7 Nov 2026 07:49:52 -0800 Subject: [PATCH 2/2] fix(studio): catalogs finish when the studio quits Quitting finishes the program's catalog and the recording's once their tracks end, then the sources' local catalog once the sources stop. A source's own catalog and a guest's return feed catalog finish with them. Viewers on the relay and recordings read each catalog to a clean end, the relay session drains instead of timing out, and quitting logs no `track::Producer dropped without finish() or abort()` warning. Co-Authored-By: Claude Opus 5.5 --- quest/m0/README.md | 1 - quest/m0/catalog-finish.md | 22 ---------- rs/moq-scene/src/publish.rs | 8 ++++ rs/moq-studio/src/engine.rs | 88 ++++++++++++++++++++++++++----------- rs/moq-studio/src/feed.rs | 15 ++++--- 5 files changed, 81 insertions(+), 53 deletions(-) delete mode 100644 quest/m0/catalog-finish.md diff --git a/quest/m0/README.md b/quest/m0/README.md index 5a7302f..df1bef9 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -102,7 +102,6 @@ repository. - [Browser sources on the GPU under Mesa](/quest/m0/cef-headless.md) - shared-texture pages paint on Intel and AMD under Wayland, and a page that cannot paint fails loudly - [Browser sources make no outside requests](/quest/m0/browser-offline.md) - no component updates, spellcheck downloads, or network time from the helper - [A media file's sound plays with its pictures](/quest/m0/media-av-sync.md) - media sources write a bus latency ahead, as broadcast sources do -- [Catalogs finish when the studio quits](/quest/m0/catalog-finish.md) - the program's and the recording's catalogs end with `finish()`, not a drop - [The UI keeps its frame rate while the program renders](/quest/m0/ui-frame-rate.md) - the editor draws without waiting on the program's render - [The Apple signing identity is available](/quest/m0/signing.md) - the Apple credentials packaging and the macOS camera need; Windows ships unsigned - [Packaging](/quest/m0/packaging.md) - a signed macOS app, a Windows installer, and a Flatpak for Linux diff --git a/quest/m0/catalog-finish.md b/quest/m0/catalog-finish.md deleted file mode 100644 index ffe5fc0..0000000 --- a/quest/m0/catalog-finish.md +++ /dev/null @@ -1,22 +0,0 @@ -# [S] Catalogs finish when the studio quits - -## Goal - -When the studio quits or stops publishing, the program's catalog and the -recording's catalog end with `finish()`, so a viewer reading either sees a -clean end, not a cut-off track, and the studio logs no -`track::Producer dropped without finish() or abort()` warning on quit. - -## Plan - -From [moq-dev/studio#91](https://github.com/moq-dev/studio/pull/91), picked -2026-11-07 (maintainer away under /afk, recommended option), m0 by the rule -that fixes join m0. - -On quit the studio logs that warning for `catalog.json` and `catalog`, twice. -`Program::drop` (`rs/moq-studio/src/engine.rs`) closes the program's broadcast -without finishing its catalog, and the recording's catalog is dropped the same -way. First confirm from the viewer's side what a viewer reading the catalog -sees today, then finish each catalog before its broadcast closes. Tests read -the program's catalog through a relay and the recording's catalog from the -local broadcast, quit, and see each one finish. diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 9ff46f1..c45615d 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -145,6 +145,12 @@ impl Publish { self.catalog.snapshot() } + /// End the local catalog, once every source has stopped, so whatever reads + /// it reaches its end rather than a cut. + pub fn finish(&self) -> Result<(), Error> { + Ok(self.catalog.clone().finish()?) + } + /// Encode the picture of each source `on_program` shows and the sound of /// each it hears, whether or not anything reads their tracks; any other /// track encodes only while read. @@ -282,6 +288,8 @@ impl Drop for Listing<'_> { root.ext.studio.sources.remove(&self.claim.name); } }); + // Fails only once the source's catalog failed, which ended its tracks. + let _ = self.catalog.finish(); } } diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index ea0813b..886ffb7 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -657,8 +657,8 @@ pub struct Program { /// What a recording reads the program from, on the same clock. recording: moq_net::origin::Route, /// The recording broadcast's catalog, held for as long as the engine runs, - /// so a recording still reading as the studio stops finds it open. - _recorded: moq_scene::publish::Catalog, + /// so a recording still reading as the studio stops reads it to its end. + recorded: moq_scene::publish::Catalog, /// The origin the studio's own broadcasts are read through, which no relay /// connection carries. local: moq_net::origin::Producer, @@ -674,6 +674,16 @@ impl Drop for Program { } impl Program { + /// Finish the program's catalog and the recording's, once every track + /// they list has ended, so a viewer or a recording reading either reaches + /// its end rather than a cut. + fn finish(&self) { + for catalog in [&self.catalog, &self.recorded] { + // Fails only once the catalog failed, which ended its tracks already. + let _ = catalog.clone().finish(); + } + } + /// What a replay of the program could copy: its video at each size, /// largest first, with its audio, up to now. pub fn program(&self) -> Vec { @@ -1049,8 +1059,9 @@ impl Engine { /// every picture, 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 published scene'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 + /// The program audio ends its track too, and then the program's catalog + /// and the recording's finish, so a viewer or 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. @@ -1063,6 +1074,7 @@ impl Engine { tracing::error!(name, "the thread panicked"); } } + self.program.finish(); // Taken out first, so the wait holds no lock the window draws with. let browser = { let mut state = self.state.lock().unwrap(); @@ -1077,8 +1089,8 @@ impl Engine { /// Once [`stop`](Self::stop) has ended the tracks of every picture, 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`]. + /// for each source's thread to end, having finished its own track, and then + /// finish the sources' local catalog. Bounded at [`CLOSE`]. pub async fn close(&self) { let running = std::mem::take(&mut *self.running.lock().unwrap()); let sources = async { @@ -1086,6 +1098,9 @@ impl Engine { // Fails once the thread ends, which is all it says. let _ = ended.await; } + // Every source has left the local catalog, so it lists nothing more. + // Fails only once it failed, which ended its tracks already. + let _ = self.program.sources.finish(); }; let relays = async { if let Some(output) = &self.output { @@ -1834,7 +1849,7 @@ async fn publish( clock: clock.clone(), sources: sources.clone(), recording, - _recorded: recorded, + recorded, local, _audio: audio_reservation, }, @@ -3184,15 +3199,33 @@ mod tests { } // Quitting while recording reads the program until stopping ends its - // tracks, into the one file. + // tracks, into the one file. Then the recording's catalog ends, and + // the sources' local one once they have stopped. engine.recorder.start(false).expect("recording starts"); let frames = recorded(&engine); encoded(&engine, frames + 30).await; + let recording = engine.program.recording.broadcast().consume(); + let mut listed = moq_mux::catalog::Consumer::<()>::new(&recording, Default::default()) + .await + .expect("the recording's catalog"); + let local = engine.program.sources.consume(); + let mut sources = moq_mux::catalog::Consumer::::new(&local, Default::default()) + .await + .expect("the sources' catalog"); engine.recorder.close(); let stopping = engine.clone(); tokio::task::spawn_blocking(move || stopping.stop()).await.unwrap(); let files = done(&engine.recorder).await; assert_eq!(files, [["soun", "vide"]]); + kio::clock::timeout(TIMEOUT, engine.close()) + .await + .expect("the studio closes"); + kio::clock::timeout(TIMEOUT, async { + while listed.next().await.expect("the recording's catalog finishes").is_some() {} + while sources.next().await.expect("the sources' catalog finishes").is_some() {} + }) + .await + .expect("both catalogs end"); assert!(engine.stats.error().is_none()); } @@ -3480,8 +3513,8 @@ mod tests { (track, first) } - /// Read a published scene's `video`, from the first group it was served, - /// and its `catalogs` to a clean end. + /// Read a broadcast's `video`, from the first group it was served, and its + /// `catalogs` to a clean end. async fn ended( mut catalogs: moq_mux::catalog::Consumer<()>, (mut video, first): (moq_net::track::Subscriber, moq_net::group::Consumer), @@ -4192,7 +4225,7 @@ mod tests { draft.set_canvas(canvas).unwrap(); state.commit(draft, None).unwrap(); } - let ended = kio::clock::timeout(TIMEOUT, async { + let old = kio::clock::timeout(TIMEOUT, async { let mut last = shown; while let Some(read) = pictures.read().await.expect("decode") { if let moq_video::decode::Read::Frame(frame) = read { @@ -4224,7 +4257,7 @@ mod tests { .expect("the resized picture arrives"); // Too small to halve, so one rendition, and one session. assert_eq!(engine.stats.sessions.open(), 1); - assert!(ended < first, "the new rendition continues the old one's timeline"); + assert!(old < first, "the new rendition continues the old one's timeline"); let next = kio::clock::timeout(TIMEOUT, async { loop { match pictures.read().await.expect("decode") { @@ -4318,24 +4351,29 @@ mod tests { assert_eq!(clip, dir.path().join("show-clip-1.mp4")); // 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(); + // published what they were handed, then its catalog in each format, and + // closes the relay session only once the relay has them, rather than + // leaving viewers dropped tracks. + let video = served(&broadcast, &resized).await; + let mut sound = broadcast.track(audio).unwrap().subscribe(None).await.unwrap(); + let msf = moq_mux::catalog::CatalogFormat::Msf; + let mut listed = moq_mux::catalog::Consumer::<()>::new(&broadcast, msf) + .await + .expect("the MSF catalog"); let stopping = engine.clone(); tokio::task::spawn_blocking(move || stopping.stop()).await.unwrap(); 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"); - } + kio::clock::timeout(TIMEOUT, async { + while let Some(mut group) = sound.recv_group().await.expect("the audio finishes") { + while group.read_frame().await.expect("the audio finishes").is_some() {} + } + while listed.next().await.expect("the MSF catalog finishes").is_some() {} + }) + .await + .expect("the audio and the MSF catalog end"); + ended(catalogs, video).await; let failed = engine.stats.error(); assert!(failed.is_none(), "{failed:?}"); } diff --git a/rs/moq-studio/src/feed.rs b/rs/moq-studio/src/feed.rs index 7a5d4d0..763eeb6 100644 --- a/rs/moq-studio/src/feed.rs +++ b/rs/moq-studio/src/feed.rs @@ -212,8 +212,8 @@ struct Running { } impl Running { - /// List the program's video and encode the picked audio until `stopped`, - /// then finish the audio track. + /// List the program's video and encode the picked audio until `stopped` + /// or the program ends, then finish the audio track and the catalog. async fn run(mut self, stopped: tokio::sync::oneshot::Receiver<()>) -> anyhow::Result<()> { let Self { catalog, @@ -233,9 +233,11 @@ impl Running { }; if let Err(err) = ended { self.audio.abort(moq_net::StreamError::Internal.into()); + self.catalog.abort(moq_net::StreamError::Internal.into()); return Err(err); } self.audio.finish()?; + self.catalog.finish()?; Ok(()) } } @@ -359,8 +361,8 @@ mod tests { } /// 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. + /// its own, which hears everyone on program but that guest, and ends + /// cleanly 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()); @@ -441,10 +443,13 @@ mod tests { .await .unwrap() .expect("an instance"); + let format = moq_mux::catalog::CatalogFormat::Hang; + let mut listings = moq_mux::catalog::Consumer::<()>::new(&bob, format).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. + // A guest who is not renewed has no feed, whose catalog ends cleanly. + while listings.next().await.expect("the catalog finishes").is_some() {} bob.closed().await; }