fix(studio): catalogs finish when the studio quits

quest/m0/README.md

@@ -101,7 +101,6 @@ repository.
 - [Virtual camera](/quest/m0/vcam/README.md) - the program or a published scene exposed as a system camera on Linux, Windows, and macOS
 - [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
 - [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

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();
 	}
 }
 

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<moq_scene::replay::Target> {
@@ -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,15 +1089,18 @@ 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 {
 			for ended in running {
 				// 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::<moq_scene::publish::Ext>::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:?}");
 	}

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::<Vec<_>>(), ["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;
 	}