feat(studio): quitting delivers the broadcast's tail before the relay session closes

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.

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

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

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)