fix(moq-mux): refuse a moov or Tracks after the importer finished

doc/setup/upgrade.md

@@ -69,7 +69,11 @@ These land with the next breaking release, not the 2026-09-23 train.
   `Tracks`, or the first PMT) on a provisional clock, then re-anchor it on the
   first frame. They now hold it until that frame, as FLV already did, so the
   first snapshot carries the final root `clock`. A reader waiting for the
-  catalog now waits for media, not just the init segment.
+  catalog now waits for media of a selected track, not just the init segment:
+  a track `with_select` deselects doesn't release it, even if its media
+  arrives first. A `moov` or `Tracks` decoded after `finish()` is refused with
+  `fmp4::Error::MoovAfterFinish` or `mkv::Error::TracksAfterFinish`, since the
+  tracks it declares could never finish.
 - **fMP4 export fixes its track set at the init segment.** moq-mux's
   `fmp4::Error` drops `MissingVideoTrack`, `MissingAudioTrack`, and
   `NoCatalogSnapshot`, and adds `TrackAdded`, `TrackChanged`, `TrackRewound`,

rs/moq-hls/src/export/mod.rs

@@ -1442,16 +1442,21 @@ mod tests {
 		let mut import = moq_mux::container::fmp4::Import::new(broadcast, catalog.reserve());
 		import.decode(&init).unwrap();
 
-		// Watch before the first fragment, as a live viewer would, and let it read whatever the
-		// moov published, so a provisional clock would stick.
+		// Watch before the first fragment, as a live viewer would, so a catalog the moov
+		// published on a provisional clock would stick.
 		let source = moq_mux::Source::new(origin.consume(), "live");
 		let broadcaster = Broadcaster::new(source, Config::default()).await.unwrap();
-		let _ = next_event(&mut broadcaster.renditions()).await;
+		assert!(
+			next_event(&mut broadcaster.renditions()).await.is_none(),
+			"the moov alone lists no rendition"
+		);
 
 		import.decode(&fragments).unwrap();
 		let anchored = catalog.snapshot().clock.expect("the catalog advertises a clock");
 
-		let _ = tokio::time::timeout(Duration::from_secs(5), broadcaster.ready()).await;
+		tokio::time::timeout(Duration::from_secs(5), broadcaster.ready())
+			.await
+			.expect("a rendition becomes playable");
 		let video = catalog
 			.snapshot()
 			.video

rs/moq-mux/src/container/fmp4/import.rs

@@ -279,12 +279,8 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 		if self.moov.is_some() {
 			return Err(Error::DuplicateMoov.into());
 		}
-		// `finish()` may have released the initial reservation already; a fresh one publishes at
-		// the end of this call.
-		let reserved = self
-			.initial_reservation
-			.clone()
-			.unwrap_or_else(|| self.catalog.reserve());
+		// Only `finish()` releases the reservation before a moov, since a fragment needs one.
+		let reserved = self.initial_reservation.clone().ok_or(Error::MoovAfterFinish)?;
 		let timeline = self.catalog.timeline();
 
 		// The tracks below enroll in the timeline, so advertise it in the same catalog update

rs/moq-mux/src/container/fmp4/import_test.rs

@@ -1463,18 +1463,44 @@ async fn a_moov_with_nothing_selected_publishes() {
 	assert_eq!(clocks.drain().len(), 1, "the catalog publishes without a fragment");
 }
 
-/// A moov decoded after `finish()` released the reservation still declares its tracks.
-#[tokio::test]
-async fn a_moov_after_finish_publishes() {
+/// A moov after `finish()` is refused: its tracks would be declared but never finished.
+#[test]
+fn a_moov_after_finish_is_refused() {
 	let (init, _, _) = bbb_init();
 	let mut broadcast = moq_net::broadcast::Info::new().produce();
-	let consumer = broadcast.consume();
 	let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap();
-	let mut clocks = crate::container::test_util::Clocks::subscribe(&consumer).await;
 	let mut fmp4 = crate::container::fmp4::Import::new(broadcast, catalog.reserve());
 
 	fmp4.finish().unwrap();
+	let err = fmp4.decode(&init).unwrap_err();
+	assert!(
+		matches!(err, crate::Error::Cmaf(crate::container::fmp4::Error::MoovAfterFinish)),
+		"{err:?}"
+	);
+	let snapshot = catalog.snapshot();
+	assert!(snapshot.video.renditions.is_empty() && snapshot.audio.renditions.is_empty());
+}
+
+/// The catalog waits for a selected track's fragment: a deselected track's fragment is skipped,
+/// so it doesn't release the reservation even when it arrives first.
+#[tokio::test]
+async fn a_deselected_fragment_does_not_publish() {
+	use crate::select::{Broadcast, Video};
+
+	let (init, (video_id, video_scale), (audio_id, audio_scale)) = bbb_init();
+	let mut broadcast = moq_net::broadcast::Info::new().produce();
+	let consumer = broadcast.consume();
+	let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap();
+	let mut clocks = crate::container::test_util::Clocks::subscribe(&consumer).await;
+	let mut fmp4 = crate::container::fmp4::Import::new(broadcast, catalog.reserve())
+		.with_select(Broadcast::default().video(Video::default()));
 	fmp4.decode(&init).unwrap();
-	assert!(!catalog.snapshot().video.renditions.is_empty(), "the moov was read");
-	assert!(!clocks.drain().is_empty(), "the catalog publishes");
+
+	let audio = super::encode_fragment(info(audio_id, audio_scale, 0), &[sample(0, true, Some(100_000))]).unwrap();
+	fmp4.decode(&audio).unwrap();
+	assert_eq!(clocks.drain(), vec![], "a deselected fragment publishes nothing");
+
+	let video = super::encode_fragment(info(video_id, video_scale, 0), &[sample(0, true, Some(100_000))]).unwrap();
+	fmp4.decode(&video).unwrap();
+	assert_eq!(clocks.drain().len(), 1, "the selected track's first fragment publishes");
 }

rs/moq-mux/src/container/fmp4/mod.rs

@@ -124,6 +124,10 @@ pub enum Error {
 	#[error("duplicate moov")]
 	DuplicateMoov,
 
+	/// A moov arrived after `finish()`, so the tracks it declares could never finish.
+	#[error("moov after finish")]
+	MoovAfterFinish,
+
 	#[error("missing trun")]
 	MissingTrun,
 

rs/moq-mux/src/container/mkv/import.rs

@@ -261,7 +261,9 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 			}
 			// A second Tracks element would redeclare a published track set.
 			MatroskaSpec::Tracks(Master::Full(children)) if !self.tracks_seen => {
-				self.handle_tracks(children)?;
+				// Only `finish()` releases the reservation before Tracks, since a block needs a track.
+				let reserved = self.initial_reservation.clone().ok_or(Error::TracksAfterFinish)?;
+				self.handle_tracks(&reserved, children)?;
 				self.tracks_seen = true;
 				// The reservation stays held until the first block anchors the clock, unless no track
 				// was declared: then no block ever will.
@@ -302,18 +304,18 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 		Err(Error::MissingDocType.into())
 	}
 
-	fn handle_tracks(&mut self, entries: Vec<MatroskaSpec>) -> Result<()> {
+	fn handle_tracks(&mut self, reserved: &crate::catalog::Reserved<E>, entries: Vec<MatroskaSpec>) -> Result<()> {
 		for entry in entries {
 			if let MatroskaSpec::TrackEntry(Master::Full(children)) = entry
-				&& let Err(e) = self.add_track(children)
+				&& let Err(e) = self.add_track(reserved, children)
 			{
 				tracing::warn!(error = ?e, "skipping MKV track");
 			}
 		}
 		Ok(())
 	}
 
-	fn add_track(&mut self, children: Vec<MatroskaSpec>) -> Result<()> {
+	fn add_track(&mut self, reserved: &crate::catalog::Reserved<E>, children: Vec<MatroskaSpec>) -> Result<()> {
 		let mut track_number: Option<u64> = None;
 		let mut track_type: Option<u64> = None;
 		let mut codec_id: Option<String> = None;
@@ -365,18 +367,12 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 			TrackKind::Video => {
 				let mut config = build_video_config(&codec_id, codec_private.as_ref(), video_children.as_deref())?;
 				config.container = self.container.clone();
-				Media::Video(match &self.initial_reservation {
-					Some(reserved) => reserved.video(track, wire, config)?,
-					None => self.catalog.video(track, wire, config)?,
-				})
+				Media::Video(reserved.video(track, wire, config)?)
 			}
 			TrackKind::Audio => {
 				let mut config = build_audio_config(&codec_id, codec_private.as_ref(), audio_children.as_deref())?;
 				config.container = self.container.clone();
-				Media::Audio(match &self.initial_reservation {
-					Some(reserved) => reserved.audio(track, wire, config)?,
-					None => self.catalog.audio(track, wire, config)?,
-				})
+				Media::Audio(reserved.audio(track, wire, config)?)
 			}
 		};
 

rs/moq-mux/src/container/mkv/import_test.rs

@@ -578,10 +578,12 @@ async fn first_catalog_carries_the_anchored_clock() {
 	let mut mkv = crate::container::mkv::Import::new(broadcast, catalog.reserve());
 
 	mkv.decode(header).unwrap();
-	assert_eq!(
-		catalog.snapshot().video.renditions.len(),
-		1,
-		"the Tracks element was read"
+	let snapshot = catalog.snapshot();
+	assert_eq!(snapshot.video.renditions.len(), 1, "the VP9 rendition was read");
+	let audio: Vec<_> = snapshot.audio.renditions.values().collect();
+	assert!(
+		matches!(audio.as_slice(), [rendition] if matches!(rendition.codec, AudioCodec::Opus)),
+		"the Opus rendition was read: {audio:?}"
 	);
 	assert_eq!(clocks.drain(), vec![], "the Tracks element alone publishes nothing");
 
@@ -595,3 +597,27 @@ async fn first_catalog_carries_the_anchored_clock() {
 	mkv.finish().unwrap();
 	assert!(clocks.drain().iter().all(|clock| *clock == Some(anchored)));
 }
+
+/// A Tracks element after `finish()` is refused: its tracks would be declared but never finished.
+#[test]
+fn tracks_after_finish_are_refused() {
+	let data = MkvBuilder::new()
+		.header("webm")
+		.segment_start()
+		.info(1_000_000)
+		.tracks(vec![track_entry_audio_opus(1, 48_000.0, 2)])
+		.segment_end()
+		.build();
+
+	let mut broadcast = moq_net::broadcast::Info::new().produce();
+	let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap();
+	let mut mkv = crate::container::mkv::Import::new(broadcast, catalog.reserve());
+
+	mkv.finish().unwrap();
+	let err = mkv.decode(&data).unwrap_err();
+	assert!(
+		matches!(err, crate::Error::Mkv(crate::container::mkv::Error::TracksAfterFinish)),
+		"{err:?}"
+	);
+	assert!(catalog.snapshot().audio.renditions.is_empty(), "no track was declared");
+}

rs/moq-mux/src/container/mkv/mod.rs

@@ -37,6 +37,10 @@ pub enum Error {
 	#[error("timestamp overflow")]
 	TimestampOverflow,
 
+	/// A Tracks element arrived after `finish()`, so the tracks it declares could never finish.
+	#[error("Tracks after finish")]
+	TracksAfterFinish,
+
 	#[error("TrackEntry missing TrackNumber")]
 	MissingTrackNumber,