From c33c9000bb9d6c293da92a2b96e4107af6c35fe3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 23:24:47 -0700 Subject: [PATCH] fix(moq-mux): refuse a moov or Tracks after the importer finished finish() releases the catalog reservation, so an init segment decoded afterwards declared tracks that could never finish. fMP4 now refuses a moov with fmp4::Error::MoovAfterFinish and MKV a Tracks element with mkv::Error::TracksAfterFinish. Also pins that a deselected track's fragment doesn't release the reservation, checks the Opus rendition in the MKV first-frame test, and asserts the moq-hls e2e test's timeouts instead of discarding them. Co-Authored-By: Claude Opus 5.5 --- doc/setup/upgrade.md | 6 ++- rs/moq-hls/src/export/mod.rs | 13 +++++-- rs/moq-mux/src/container/fmp4/import.rs | 8 +--- rs/moq-mux/src/container/fmp4/import_test.rs | 40 ++++++++++++++++---- rs/moq-mux/src/container/fmp4/mod.rs | 4 ++ rs/moq-mux/src/container/mkv/import.rs | 20 ++++------ rs/moq-mux/src/container/mkv/import_test.rs | 34 +++++++++++++++-- rs/moq-mux/src/container/mkv/mod.rs | 4 ++ 8 files changed, 95 insertions(+), 34 deletions(-) diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md index 010f616fbf..83a034cfb4 100644 --- a/doc/setup/upgrade.md +++ b/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`, diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 6b3aacf041..cccbf9bea3 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/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 diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index ca85131901..8d35cc29df 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -279,12 +279,8 @@ impl Import { 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 diff --git a/rs/moq-mux/src/container/fmp4/import_test.rs b/rs/moq-mux/src/container/fmp4/import_test.rs index 461b4242ac..f8630de4ee 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/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"); } diff --git a/rs/moq-mux/src/container/fmp4/mod.rs b/rs/moq-mux/src/container/fmp4/mod.rs index 5bc68c1b85..886cb43142 100644 --- a/rs/moq-mux/src/container/fmp4/mod.rs +++ b/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, diff --git a/rs/moq-mux/src/container/mkv/import.rs b/rs/moq-mux/src/container/mkv/import.rs index 72869ebe34..beef0c875f 100644 --- a/rs/moq-mux/src/container/mkv/import.rs +++ b/rs/moq-mux/src/container/mkv/import.rs @@ -261,7 +261,9 @@ impl Import { } // 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,10 +304,10 @@ impl Import { Err(Error::MissingDocType.into()) } - fn handle_tracks(&mut self, entries: Vec) -> Result<()> { + fn handle_tracks(&mut self, reserved: &crate::catalog::Reserved, entries: Vec) -> 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"); } @@ -313,7 +315,7 @@ impl Import { Ok(()) } - fn add_track(&mut self, children: Vec) -> Result<()> { + fn add_track(&mut self, reserved: &crate::catalog::Reserved, children: Vec) -> Result<()> { let mut track_number: Option = None; let mut track_type: Option = None; let mut codec_id: Option = None; @@ -365,18 +367,12 @@ impl Import { 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)?) } }; diff --git a/rs/moq-mux/src/container/mkv/import_test.rs b/rs/moq-mux/src/container/mkv/import_test.rs index da446f2638..cbcf698cff 100644 --- a/rs/moq-mux/src/container/mkv/import_test.rs +++ b/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"); +} diff --git a/rs/moq-mux/src/container/mkv/mod.rs b/rs/moq-mux/src/container/mkv/mod.rs index 5905d0b005..78d2f28ec4 100644 --- a/rs/moq-mux/src/container/mkv/mod.rs +++ b/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,