diff --git a/doc/lib/rs/moq-mux.md b/doc/lib/rs/moq-mux.md index 249e949b64..25d907b2c4 100644 --- a/doc/lib/rs/moq-mux.md +++ b/doc/lib/rs/moq-mux.md @@ -131,9 +131,11 @@ The fMP4, MPEG-TS, FLV, and MKV importers publish the source's own timestamps (MPEG-TS after unwrapping its 33-bit PTS; fMP4 passthrough keeps each `tfdt`) and anchor the catalog's broadcast clock instead: the first frame's timestamp maps to the time it arrived, and every track of the input, like every importer -sharing the catalog, keeps that one mapping. Data tracks stamp on it too, even -one created before that first frame, though anything it wrote earlier stays on -the clock the catalog started with. A clock set with +sharing the catalog, keeps that one mapping. Each importer withholds the +catalog until that first frame, so its first snapshot already carries the +anchored root `clock` for readers that copy it once. Data tracks stamp on the +clock too, even one created before that first frame, though anything it wrote +earlier stays on the clock the catalog started with. A clock set with `Config::with_clock` is never re-anchored, for a recording whose zero names its real start. diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md index 20f2c7c233..010f616fbf 100644 --- a/doc/setup/upgrade.md +++ b/doc/setup/upgrade.md @@ -64,6 +64,12 @@ These land with the next breaking release, not the 2026-09-23 train. `Instant` with `.at(catalog.clock().capture(instant)?)`, reading `catalog.clock()` at write time, since an importer's first frame re-anchors it. A timestamp ahead of now is published rather than refused. +- **moq-mux importers publish the catalog at their first frame.** The fMP4, + MKV, and MPEG-TS importers used to publish it at their init segment (`moov`, + `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. - **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/quest/m1/README.md b/quest/m1/README.md index af04ff2136..4888ee8e4e 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -28,7 +28,6 @@ blocks. The quests that gated m0 lines moved under them. ## Required -- [Import at the first frame](/quest/m1/import-first-frame.md) - fMP4, MKV, and MPEG-TS imports publish the catalog at their first frame, so a lone importer's root clock never moves - [HLS first catalog](/quest/m1/hls-first-catalog.md) - moq-hls import's first catalog lists every rendition from the master playlist, not only the first - [Typed timedness](/quest/m1/typed-timedness.md) - a track is all timed or all untimed, `Info.timescale` is optional, and `Timed` loses its clock parameter, in Rust, JS, and the bindings - [Untimed model](/quest/m1/untimed-model.md) - an untimed moq-net track reaches every subscriber untimed; no receiver fills in arrival time diff --git a/quest/m1/hls-first-catalog.md b/quest/m1/hls-first-catalog.md index af5b35af1f..b2034dfac6 100644 --- a/quest/m1/hls-first-catalog.md +++ b/quest/m1/hls-first-catalog.md @@ -16,10 +16,9 @@ and grows on later updates. Reported by Dryvnt in ## Plan -Decided (2026-10-05, maintainer): m1, [XS], ranked right after [Import at the -first frame](/quest/m1/import-first-frame.md). That quest moves each -importer's release to its first frame, which still lands before the next -rendition is reserved. +Decided (2026-10-05, maintainer): m1, [XS]. Each importer now releases its +reservation at its first frame, which still lands before the next rendition +is reserved. Guidance: @@ -44,5 +43,4 @@ lists every rendition instead of only the first. ## Related -- [Import at the first frame](/quest/m1/import-first-frame.md) - moves each importer's catalog release to its first frame - [Shared import clock](/quest/m1/shared-clock.md) - moq-hls renditions share one `catalog::Input`, whose `reserve()` this hold may use diff --git a/quest/m1/hop-aligned-import.md b/quest/m1/hop-aligned-import.md index 3f3fe0b336..b63352b87d 100644 --- a/quest/m1/hop-aligned-import.md +++ b/quest/m1/hop-aligned-import.md @@ -27,9 +27,8 @@ fed one encoded stream. Capture is out: two encoders never align. stream's own timestamps and refuse a rewind, so nothing shifts them. - The catalog's root `clock` must agree too. Today each importer anchors it to its own first-frame arrival (`Clock::arrival` in - `rs/moq-mux/src/catalog/producer.rs`), which - [Import at the first frame](/quest/m1/import-first-frame.md) makes the - first published clock, and [Shared import clock](/quest/m1/shared-clock.md) + `rs/moq-mux/src/catalog/producer.rs`), which is also the first published + clock, since each importer holds its catalog until that frame, and [Shared import clock](/quest/m1/shared-clock.md) offsets a joining importer by its arrival time. Decided in the 2026-10-05 audit: redundant importers derive the wall anchor from the input (its PTS or PCR) or from the shared epoch, never from arrival, so two catalogs of diff --git a/quest/m1/import-first-frame.md b/quest/m1/import-first-frame.md deleted file mode 100644 index 4356b04a6a..0000000000 --- a/quest/m1/import-first-frame.md +++ /dev/null @@ -1,69 +0,0 @@ -# [S] Importers publish the catalog at their first frame - -## Goal - -A catalog fed by one container importer is first published with its final -root `clock`. Today fMP4 and MKV imports publish the catalog when their init -segment (`moov`, `Tracks`) resolves the reservation, and an Opus-only MPEG-TS -import when its PMT does, on a provisional clock, then re-anchor it on the -first frame (`catalog::Producer::anchor`). Readers that copy the clock once -(moq-hls export's `EXT-X-PROGRAM-DATE-TIME` and `availabilityStartTime`, -derived broadcasts) keep the provisional one, and the hang draft says the -mapping is "fixed for the broadcast". - -Non-goal: a clock that never moves after any publish. A container set up -after a data track or catalog section has published still re-anchors once; -[shared-clock](/quest/m1/shared-clock.md) makes the first publish fix the -clock and offsets that container instead. - -## Plan - -The anchor is unreleased, and this breaks an in-tree path (an fMP4 or MKV -import read by moq-hls export), so it lands before the next release cut. -Requested by an external consumer (OneTooMany). Ranked first in m1, ahead of -shared-clock and cmaf-frame-timestamp (maintainer, 2026-10-05). - -Decided (2026-10-05): - -- fMP4, MKV, and MPEG-TS hold their initial reservation until their first - frame anchors, as FLV already does (`flv/import.rs`, "anchored before the - reservation below publishes"). The first snapshot then carries the anchored - clock. PTS stays verbatim for a single importer, so a passthrough fragment's - `tfdt` still agrees with its frame timestamp. -- Split from the strict "first publish fixes the clock" rule, which needs - shared-clock's offset for a container joining a fixed clock. The hold alone - needs no offset and regresses nothing: an earlier publish still lets the - anchor move the clock, as on `main`. -- Rejected: omitting `clock` until anchored. Adding it later is still a change - copy-once readers miss. - -Guidance: - -- An fMP4 `moov` declares every track at once, and MKV's `Tracks` likewise, so - the first frame of any track releases the hold. -- MPEG-TS drops its hold after the PMT (`ts/import.rs`, "Every stream in the - initial program is registered now"). Video configs resolve only on the first - frame, after the anchor, but Opus builds its config from PMT descriptors, so - an Opus-only stream publishes before its first PES anchors. Keep the hold - until the first anchor. -- fMP4 detects a duplicate `moov` by its reservation being gone - (`fmp4/import.rs`, `DuplicateMoov`). Holding it to the first fragment lets a - second `moov` slip through, so key the check on the parsed `moov` instead. - Likewise check that a TS PMT version change before the first PES does not - re-reserve against the held handle. -- Tests (fail on `main` for fMP4, MKV, and Opus-only TS): for each of fMP4, - MKV, FLV, and TS, a catalog consumer's first snapshot carries the anchored - clock and later snapshots never change it. The TS case is Opus-only with a - nonzero first PTS, since a video stream's config resolves only on its first - frame, after the anchor. Also a TS program of only verbatim PIDs, which - publish through `modify` with no reservation. Plus an end-to-end fMP4 import - into moq-hls export whose `EXT-X-PROGRAM-DATE-TIME` matches the anchored - clock. -- Docs: fix the `Config::with_clock` and `Producer::anchor` comments and - `doc/lib/rs/moq-mux.md`, and add an `Unreleased` line in - `doc/setup/upgrade.md`: the catalog now appears at an importer's first - frame, not its init segment. - -Public API: no new items. A catalog fed by an fMP4, MKV, or Opus-only TS -importer is published at the first frame instead of the init segment. Wire: -none; the hang draft already says the mapping is fixed. diff --git a/quest/m1/shared-clock.md b/quest/m1/shared-clock.md index 209609724a..cad69a29ed 100644 --- a/quest/m1/shared-clock.md +++ b/quest/m1/shared-clock.md @@ -12,8 +12,8 @@ backwards, and `container::Producer::write` refuses a group below the last one (`TimestampRewind`), even across a discontinuity. The catalog's root `clock` is also final from the first snapshot a consumer -sees. [import-first-frame](/quest/m1/import-first-frame.md) covers a lone -importer; a container set up after a data track or catalog section has +sees. A lone importer already holds its catalog until its first frame +anchors; a container set up after a data track or catalog section has published (any order moq-c and moq-ffi allow) still re-anchors it after copy-once readers (moq-hls export, derived broadcasts) took the old one. @@ -136,7 +136,6 @@ Wire: none. ## Required -- [Import at the first frame](/quest/m1/import-first-frame.md) - a lone importer anchors before the first publish, so the publish rule doesn't offset it - [CMAF frame timestamp](/quest/m1/cmaf-frame-timestamp.md) - decoders honour an offset frame timestamp on passthrough tracks ## Related diff --git a/quest/m2/rtsp-import.md b/quest/m2/rtsp-import.md index 98742721b5..903dcea7f0 100644 --- a/quest/m2/rtsp-import.md +++ b/quest/m2/rtsp-import.md @@ -93,7 +93,6 @@ The moq.pro guide link below resolves once moq.pro#2210 merges. ## Related - [Broadcast epochs](/quest/m0/broadcast-epoch/README.md) - the fresh epoch a caller and the CLI mint for each restarted broadcast -- [Import at the first frame](/quest/m1/import-first-frame.md) - a lone session's catalog publishes at its first frame - [moq.pro's Pronto truck](https://github.com/moq-dev/moq.pro/blob/main/pronto/truck/src/camera.rs) - the prior art this generalizes - [moq.pro: Pronto truck on moq-rtsp](https://github.com/moq-dev/moq.pro/blob/main/quest/m3/truck-rtsp.md) - runs the library entry point under its own supervisor - [moq.pro: Camera (RTSP) guide](https://github.com/moq-dev/moq.pro/blob/main/quest/m2/rtsp-guide.md) - documents `moq import rtsp` for customers once a release ships it diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index a5773ce0d0..6b3aacf041 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -1420,6 +1420,61 @@ mod tests { drop((media, registration, broadcast)); } + // The playlist copies the root clock when it first sees a rendition, so an fMP4 import must + // not publish its catalog on a provisional clock that the first fragment then re-anchors. + #[tokio::test(start_paused = true)] + async fn fmp4_import_times_playlists_by_the_anchored_clock() { + // An hour into the encoder's own timeline, a keyframe every 2s. + let start = 3_600_000_000; + let muxer = moq_mux::container::fmp4::Muxer::video(&video_config()).unwrap(); + let init = muxer.init().unwrap().expect("a VP8 init from the coded size"); + let mut fragments = Vec::new(); + for i in 0..6u64 { + let frame = vp8_frame(start + i * 1_000_000, i % 2 == 0); + fragments.extend_from_slice(&muxer.fragment(i as u32, &[frame]).unwrap()); + } + + let origin = produce_origin(); + let mut broadcast = origin.create_broadcast("live").expect("publish allowed"); + broadcast.announce(Default::default()).expect("publish allowed"); + settle().await; + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); + 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. + 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; + + 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; + let video = catalog + .snapshot() + .video + .renditions + .into_keys() + .next() + .expect("a video rendition"); + let rendition = broadcaster.rendition(Kind::Video, &video).expect("video discovered"); + let playlist = tokio::time::timeout(Duration::from_secs(5), rendition.playlist(None)) + .await + .expect("a listed segment") + .unwrap() + .expect("playlist renders"); + + let first = anchored + .wall_clock(moq_net::Timestamp::from_micros(start).unwrap()) + .unwrap(); + let expected = format!("#EXT-X-PROGRAM-DATE-TIME:{}\n", humantime::format_rfc3339_millis(first)); + assert!(playlist.contains(&expected), "{expected} in {playlist}"); + + drop((import, catalog)); + } + // A finished broadcast renders a static presentation, offset to its first listed segment. #[tokio::test] async fn dash_manifest_turns_static_when_finished() { diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index 698d1fd4d2..ae3c14e2f4 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -277,8 +277,9 @@ impl Config { /// /// The clock's wall mapping is advertised at the catalog root and fixed for the broadcast. /// Without this, the catalog starts a fresh clock, and a container importer (fMP4, MPEG-TS, - /// FLV, MKV) re-anchors it on its first timestamp, before writing any frame, so the stream's own - /// timestamps map to the arrival time. Pass one whose PTS zero names the content's real start + /// FLV, MKV) anchors it on its first frame's timestamp, so the stream's own timestamps map to + /// the arrival time. The importer withholds the catalog until then, so its first snapshot + /// already carries the anchored clock. Pass one whose PTS zero names the content's real start /// when importing a recording. pub fn with_clock(mut self, clock: crate::Clock) -> Self { self.clock = Some(clock); @@ -470,7 +471,10 @@ impl Producer { /// /// Container importers publish their stream's timestamps verbatim and call this before /// writing each frame: the first call places the wall mapping so `pts` is live on arrival, - /// and every later call, from any track or importer sharing this catalog, is a no-op. + /// and every later call, from any track or importer sharing this catalog, is a no-op. An + /// importer holds its initial reservation until this first call, so a catalog it feeds is + /// first published on the anchored clock. One already published by another producer still + /// moves here. pub(crate) fn anchor(&mut self, pts: moq_net::Timestamp) -> crate::Result<()> { if take(&self.current).anchored { return Ok(()); diff --git a/rs/moq-mux/src/container/flv/import.rs b/rs/moq-mux/src/container/flv/import.rs index 59480c0d25..363495b1a2 100644 --- a/rs/moq-mux/src/container/flv/import.rs +++ b/rs/moq-mux/src/container/flv/import.rs @@ -643,6 +643,8 @@ impl Import { /// Finish every track, flushing the current group. pub fn finish(&mut self) -> crate::Result<()> { + // No frame follows to anchor the clock, so publish the declared track set now. + self.initial_reservation = None; for stream in self.video.values_mut() { stream.track.finish()?; } diff --git a/rs/moq-mux/src/container/flv/import_test.rs b/rs/moq-mux/src/container/flv/import_test.rs index 7d50266ffa..be5a17a22a 100644 --- a/rs/moq-mux/src/container/flv/import_test.rs +++ b/rs/moq-mux/src/container/flv/import_test.rs @@ -788,3 +788,32 @@ async fn import_refuses_a_restart() { "the first session stays published" ); } + +/// The catalog is first published at the first frame, carrying the clock that frame anchors, +/// rather than at the sequence headers on a provisional clock a copy-once reader would keep. +#[tokio::test] +async fn first_catalog_carries_the_anchored_clock() { + let start_ms = 3_600_000; + let data = session(true, start_ms, 10); + let (headers, frames) = data.split_at(session(true, start_ms, 0).len()); + + 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 provisional = catalog.clock().wall(); + let mut clocks = crate::container::test_util::Clocks::subscribe(&consumer).await; + let mut importer = Import::new(broadcast, catalog.reserve()); + + importer.decode(headers).unwrap(); + assert_eq!(clocks.drain(), vec![], "the sequence headers alone publish nothing"); + + importer.decode(frames).unwrap(); + let anchored = catalog.clock().wall(); + assert_ne!(anchored, provisional, "the first frame anchors the clock"); + let published = clocks.drain(); + assert!(!published.is_empty(), "the first frame publishes the catalog"); + assert!(published.iter().all(|clock| *clock == Some(anchored)), "{published:?}"); + + importer.finish().unwrap(); + assert!(clocks.drain().iter().all(|clock| *clock == Some(anchored))); +} diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index fdd70ff050..ca85131901 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -33,9 +33,9 @@ pub struct Import { /// The catalog being produced catalog: crate::catalog::Producer, - /// Held until the moov's track set is declared, so the catalog is withheld from the broadcast - /// until every rendition is in (and, when composed with other importers, until they finish too). - /// Dropped in [`init`](Self::init). + /// Held until the first fragment anchors the clock, so the catalog is withheld from the broadcast + /// until every rendition is in and its root `clock` is final (and, when composed with other + /// importers, until they release theirs too). initial_reservation: Option>, // Which track roles to publish. `None` imports every supported track. @@ -275,9 +275,16 @@ impl Import { } fn init(&mut self, moov: Moov) -> Result<()> { - // Held from construction until the track set is declared, so a second moov would be - // re-declaring a track set the catalog already published. - let reserved = self.initial_reservation.clone().ok_or(Error::DuplicateMoov)?; + // A second moov would re-declare a track set the catalog already advertises. + 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()); let timeline = self.catalog.timeline(); // The tracks below enroll in the timeline, so advertise it in the same catalog update @@ -353,9 +360,11 @@ impl Import { ); } - // The moov's full track set is declared now; release the reservation so the catalog publishes. - drop(reserved); - self.initial_reservation = None; + // The reservation stays held until the first fragment anchors the clock, unless no track + // was declared: then no fragment ever will. + if self.tracks.is_empty() { + self.initial_reservation = None; + } self.moov = Some(moov); @@ -909,8 +918,11 @@ impl Import { // in the track's native timescale. The relay reads it off the wire; the // consumer still drives playback from the fragment's internal timing. let timestamp = min_timestamp.ok_or(Error::MissingTrun)?; - // The first fragment of the import is live on arrival. + // The first fragment of the import is live on arrival. Anchor before releasing the + // reservation, so the first snapshot carries the final clock; the moov declared every + // track, so any track's fragment releases it. self.catalog.anchor(timestamp)?; + self.initial_reservation = None; // Write the per-track fragment as a single MoQ frame (passthrough). The group rolls // once per segment, so a group is a segment and the fragments inside it are frames. @@ -987,6 +999,8 @@ impl Import { impl Import { /// Finish all tracks, flushing current groups. pub fn finish(&mut self) -> Result<()> { + // No frame follows to anchor the clock, so publish the declared track set now. + self.initial_reservation = None; for track in self.tracks.values_mut() { track.estimator.cut(None); track.publish_estimate()?; diff --git a/rs/moq-mux/src/container/fmp4/import_test.rs b/rs/moq-mux/src/container/fmp4/import_test.rs index e4737c85a0..461b4242ac 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/rs/moq-mux/src/container/fmp4/import_test.rs @@ -1405,3 +1405,76 @@ async fn seek_keeps_the_decode_time_until_a_discontinuity() { fmp4.decode(&live_session(0, 5)) .expect("a signalled discontinuity starts a new timeline"); } + +/// The catalog is first published at the first fragment, carrying the clock that fragment +/// anchors, rather than at the moov on a provisional clock a copy-once reader would keep. +#[tokio::test] +async fn first_catalog_carries_the_anchored_clock() { + 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 provisional = catalog.clock().wall(); + let mut clocks = crate::container::test_util::Clocks::subscribe(&consumer).await; + let mut fmp4 = crate::container::fmp4::Import::new(broadcast, catalog.reserve()); + + fmp4.decode(&init).unwrap(); + assert_eq!(clocks.drain(), vec![], "the moov alone publishes nothing"); + + fmp4.decode(&live_session(3_600_000_000, 20)).unwrap(); + let anchored = catalog.clock().wall(); + assert_ne!(anchored, provisional, "the first fragment anchors the clock"); + let published = clocks.drain(); + assert!(!published.is_empty(), "the first fragment publishes the catalog"); + assert!(published.iter().all(|clock| *clock == Some(anchored)), "{published:?}"); + + fmp4.finish().unwrap(); + assert!(clocks.drain().iter().all(|clock| *clock == Some(anchored))); +} + +/// A second moov is refused before the first fragment too, while the reservation is still held. +#[test] +fn a_second_moov_is_refused() { + let (init, _, _) = bbb_init(); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut fmp4 = crate::container::fmp4::Import::new(broadcast, catalog.reserve()); + + fmp4.decode(&init).unwrap(); + let err = fmp4.decode(&init).unwrap_err(); + assert!( + matches!(err, crate::Error::Cmaf(crate::container::fmp4::Error::DuplicateMoov)), + "{err:?}" + ); +} + +/// A moov with every track deselected publishes at once, since no fragment will ever anchor. +#[tokio::test] +async fn a_moov_with_nothing_selected_publishes() { + 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()) + .with_select(crate::select::Broadcast::default()); + + fmp4.decode(&init).unwrap(); + 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() { + 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(); + fmp4.decode(&init).unwrap(); + assert!(!catalog.snapshot().video.renditions.is_empty(), "the moov was read"); + assert!(!clocks.drain().is_empty(), "the catalog publishes"); +} diff --git a/rs/moq-mux/src/container/fmp4/mod.rs b/rs/moq-mux/src/container/fmp4/mod.rs index 5fee8e32f8..5bc68c1b85 100644 --- a/rs/moq-mux/src/container/fmp4/mod.rs +++ b/rs/moq-mux/src/container/fmp4/mod.rs @@ -120,7 +120,7 @@ pub enum Error { #[error("duplicate moof")] DuplicateMoof, - /// A second moov would re-declare the track set the first one already published. + /// A second moov would re-declare the track set the first one already declared. #[error("duplicate moov")] DuplicateMoov, diff --git a/rs/moq-mux/src/container/mkv/import.rs b/rs/moq-mux/src/container/mkv/import.rs index 141e6bacc1..72869ebe34 100644 --- a/rs/moq-mux/src/container/mkv/import.rs +++ b/rs/moq-mux/src/container/mkv/import.rs @@ -43,8 +43,9 @@ pub struct Import { catalog: crate::catalog::Producer, container: hang::catalog::Container, - /// Held until the Tracks element is processed, so the catalog is withheld from the broadcast - /// until every rendition is in (and, when composed with other importers, until they finish too). + /// Held until the first block anchors the clock, so the catalog is withheld from the broadcast + /// until every rendition is in and its root `clock` is final (and, when composed with other + /// importers, until they release theirs too). initial_reservation: Option>, /// Accumulated unparsed input. @@ -262,8 +263,11 @@ impl Import { MatroskaSpec::Tracks(Master::Full(children)) if !self.tracks_seen => { self.handle_tracks(children)?; self.tracks_seen = true; - // The full track set is declared now; release the reservation so the catalog publishes. - self.initial_reservation = None; + // The reservation stays held until the first block anchors the clock, unless no track + // was declared: then no block ever will. + if self.tracks.is_empty() { + self.initial_reservation = None; + } } MatroskaSpec::Cluster(Master::Start) => { self.cluster_timestamp = 0; @@ -430,8 +434,11 @@ impl Import { .checked_mul(self.timestamp_scale_ns) .ok_or(Error::TimestampOverflow)?; let timestamp = Timestamp::from_nanos(pts_ns)?; - // The first block is live on arrival. + // The first block is live on arrival. Anchor before releasing the reservation, so the + // first snapshot carries the final clock; Tracks declared every track, so any track's + // block releases it. self.catalog.anchor(timestamp)?; + self.initial_reservation = None; // Audio tracks: always treat as keyframes (matches fmp4 behavior). let keyframe = matches!(track.kind, TrackKind::Audio) || keyframe; @@ -473,6 +480,8 @@ impl Import { /// Finish all tracks, flushing current groups. pub fn finish(&mut self) -> Result<()> { + // No frame follows to anchor the clock, so publish the declared track set now. + self.initial_reservation = None; for track in self.tracks.values_mut() { if let Some(g) = track.group.take() { g.finish()?; diff --git a/rs/moq-mux/src/container/mkv/import_test.rs b/rs/moq-mux/src/container/mkv/import_test.rs index fe9f827b16..da446f2638 100644 --- a/rs/moq-mux/src/container/mkv/import_test.rs +++ b/rs/moq-mux/src/container/mkv/import_test.rs @@ -547,3 +547,51 @@ async fn import_refuses_a_restart() { assert!(matches!(err, crate::Error::TimestampRewind(_)), "{err:?}"); assert_eq!(import.published.values().next().unwrap().len(), 2); } + +/// The catalog is first published at the first block, carrying the clock that block anchors, +/// rather than at the Tracks element on a provisional clock a copy-once reader would keep. +#[tokio::test] +async fn first_catalog_carries_the_anchored_clock() { + let data = MkvBuilder::new() + .header("webm") + .segment_start() + .info(1_000_000) + .tracks(vec![ + track_entry_video_vp9(1, 16, 16), + track_entry_audio_opus(2, 48_000.0, 2), + ]) + .cluster(3_600_000, || { + vec![simple_block(2, 0, true, b"a0"), simple_block(1, 0, true, b"k0")] + }) + .segment_end() + .build(); + // The writer sizes the Segment at its end, so split at the Cluster's element ID. + const CLUSTER_ID: [u8; 4] = [0x1F, 0x43, 0xB6, 0x75]; + let split = data.windows(4).position(|w| w == CLUSTER_ID).expect("a cluster"); + let (header, cluster) = data.split_at(split); + + 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 provisional = catalog.clock().wall(); + let mut clocks = crate::container::test_util::Clocks::subscribe(&consumer).await; + 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" + ); + assert_eq!(clocks.drain(), vec![], "the Tracks element alone publishes nothing"); + + mkv.decode(cluster).unwrap(); + let anchored = catalog.clock().wall(); + assert_ne!(anchored, provisional, "the first block anchors the clock"); + let published = clocks.drain(); + assert!(!published.is_empty(), "the first block publishes the catalog"); + assert!(published.iter().all(|clock| *clock == Some(anchored)), "{published:?}"); + + mkv.finish().unwrap(); + assert!(clocks.drain().iter().all(|clock| *clock == Some(anchored))); +} diff --git a/rs/moq-mux/src/container/test_util.rs b/rs/moq-mux/src/container/test_util.rs index 3940408fe0..d73a3c7e3c 100644 --- a/rs/moq-mux/src/container/test_util.rs +++ b/rs/moq-mux/src/container/test_util.rs @@ -163,6 +163,33 @@ pub(crate) async fn published( out } +/// The root clocks a broadcast's catalog snapshots carry, read off its `catalog.json` track. +pub(crate) struct Clocks(crate::catalog::hang::Consumer); + +impl Clocks { + /// Subscribe before importing, so the first snapshot is not missed. + pub(crate) async fn subscribe(consumer: &moq_net::broadcast::Consumer) -> Self { + let track = consumer + .track(hang::Catalog::DEFAULT_NAME) + .unwrap() + .subscribe(None) + .await + .unwrap(); + Self(crate::catalog::hang::Consumer::new(track)) + } + + /// The clock of every snapshot published since the last call, without waiting. + pub(crate) fn drain(&mut self) -> Vec> { + let waiter = kio::Waiter::noop(); + let mut clocks = Vec::new(); + while let std::task::Poll::Ready(catalog) = self.0.poll_next(&waiter) { + let Some(catalog) = catalog.unwrap() else { break }; + clocks.push(catalog.clock); + } + clocks + } +} + /// The one offset every published timestamp moved by between two imports of the same media, /// within a tick of rounding: one mapping for every track, so A/V sync and B-frame order survive /// exactly. diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index fa2cbebd59..2a6d26dbe0 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -53,9 +53,9 @@ pub struct Import { catalog: crate::catalog::Producer, container: hang::catalog::Container, - /// Held while the first PMT is parsed so the catalog is withheld from the broadcast until every - /// stream in the initial program has reserved its rendition. Dropped once the PMT is fully - /// processed, so a one-shot muxer (fMP4, TS re-export) sees the complete track list in the first + /// Held until the first PES anchors the clock, so the catalog is withheld from the broadcast + /// until every stream in the initial program has reserved its rendition and the root `clock` is + /// final. A one-shot muxer (fMP4, TS re-export) sees the complete track list in the first /// snapshot rather than a half-converged one. See [`Reserved`](crate::catalog::Reserved). initial_reservation: Option>, @@ -138,7 +138,7 @@ impl Import { pub fn new(broadcast: moq_net::broadcast::Producer, reserved: crate::catalog::Reserved) -> Self { let container = hang::catalog::Container::default(); // A long-lived producer handle for catalog edits (mpegts sections, later PMTs); the passed - // reservation gates the initial publish and is dropped once the first PMT is parsed. + // reservation gates the initial publish and is dropped once the first PES anchors. let catalog = reserved.producer(); // Sample the real catalog once at construction, not E::default(): an extension // may carry the section by value, and a snapshot clones under the mutex (no publish). @@ -430,10 +430,6 @@ impl Import { } let pids: Vec = pmt.streams.iter().map(|es| es.pid.as_u16()).collect(); self.health.pmt_streams(&pids); - - // Every stream in the initial program is registered now; release the reservation - // so the catalog publishes once each rendition's config resolves, not before. - self.initial_reservation = None; Ok(()) } @@ -680,6 +676,7 @@ impl Import { // anchors the catalog here like a flushed PES would. if let Some(pts) = pes.pts { self.catalog.anchor(Timestamp::from_scale(pts, 90_000)?)?; + self.initial_reservation = None; } // Nothing is published, but each PES is a picture the source delivered. self.liveness.delivered(pid.as_u16(), 1); @@ -740,9 +737,13 @@ impl Import { let Some(stream) = self.streams.get_mut(&pid) else { return Ok(()); }; - // The first PES is live on arrival: its PTS needs no unwrap yet. + // The first PES is live on arrival: its PTS needs no unwrap yet. Anchor before releasing + // the reservation, so the first snapshot carries the final clock: an Opus config comes + // from the PMT and would otherwise publish first. Every stream in the initial program + // reserved at its PMT, so any stream's PES releases it. if let Some(pts) = pending.pts { self.catalog.anchor(Timestamp::from_scale(pts, 90_000)?)?; + self.initial_reservation = None; } let units = match stream.write(pending, batched) { Ok(units) => units, @@ -936,6 +937,8 @@ impl Import { for pid in pids { self.flush(pid)?; } + // No frame follows to anchor the clock, so publish the declared track set now. + self.initial_reservation = None; for (pid, stream) in &mut self.streams { let units = stream.finish()?; self.liveness.delivered(pid.as_u16(), units); @@ -5310,6 +5313,45 @@ pub(super) mod test { assert_eq!(&frame.payload[..], &payload[..], "verbatim PES payload round-trips"); } + /// A program of only verbatim PIDs reserves no rendition, so the held reservation alone keeps + /// its PMT from publishing the catalog before the first PES anchors the clock. + #[tokio::test] + async fn verbatim_only_program_publishes_at_the_first_pes() { + use crate::catalog::hang::Catalog; + use crate::container::ts::catalog::Ext; + + const DATA_PID: u16 = 0x0052; + + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(Catalog::::default()), + ) + .unwrap(); + let provisional = catalog.clock().wall(); + let mut clocks = crate::container::test_util::Clocks::subscribe(&consumer).await; + let mut import = super::Import::new(broadcast, catalog.reserve()); + + import + .decode(&synth_pmt(&[(StreamType::Mpeg2PacketizedData, DATA_PID)], false)) + .unwrap(); + assert_eq!(catalog.snapshot().ext.mpegts.tracks.len(), 1, "the PMT was read"); + assert_eq!(clocks.drain(), vec![], "the PMT alone publishes nothing"); + + import + .decode(&audio_pes_packet(DATA_PID, 0, 3_600 * 90_000, &[0xDE, 0xAD])) + .unwrap(); + let anchored = catalog.clock().wall(); + assert_ne!(anchored, provisional, "the first PES anchors the clock"); + let published = clocks.drain(); + assert!(!published.is_empty(), "the first PES publishes the catalog"); + assert!(published.iter().all(|clock| *clock == Some(anchored)), "{published:?}"); + + import.finish().unwrap(); + assert!(clocks.drain().iter().all(|clock| *clock == Some(anchored))); + } + /// A TS packet on `pid` carrying only an adaptation field (no payload) that sets /// `discontinuity_indicator`: the clock packet a mux flags a timebase reset on. fn clock_break_packet(pid: u16) -> Vec { diff --git a/rs/moq-mux/src/container/ts/import_test.rs b/rs/moq-mux/src/container/ts/import_test.rs index 5d47ba5e13..b3588a8d08 100644 --- a/rs/moq-mux/src/container/ts/import_test.rs +++ b/rs/moq-mux/src/container/ts/import_test.rs @@ -269,6 +269,36 @@ async fn import_opus_frames() { } } +/// An Opus config comes from the PMT, before any PES, yet the catalog is first published at the +/// first PES, carrying the clock it anchors rather than a provisional one a copy-once reader +/// would keep. +#[tokio::test] +async fn opus_catalog_carries_the_anchored_clock() { + // An hour in, so the anchored clock lands far from the provisional one. + let data = shift_clock(include_bytes!("test_data/opus.ts"), 3_600 * 90_000); + + 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 provisional = catalog.clock().wall(); + let mut clocks = crate::container::test_util::Clocks::subscribe(&consumer).await; + let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve()); + + // Packet by packet, so a snapshot published at the PMT is seen before the PES replaces it. + let mut published = Vec::new(); + for pkt in data.chunks(188) { + import.decode(pkt).unwrap(); + published.extend(clocks.drain()); + } + import.finish().unwrap(); + published.extend(clocks.drain()); + + let anchored = catalog.clock().wall(); + assert_ne!(anchored, provisional, "the first PES anchors the clock"); + assert!(!published.is_empty(), "the catalog publishes"); + assert!(published.iter().all(|clock| *clock == Some(anchored)), "{published:?}"); +} + /// `eac3.ts` is an ffmpeg-authored audio-only ATSC E-AC-3 program (stream_type /// 0x87 plus the 'EAC3' registration descriptor), regenerated with: /// `ffmpeg -f lavfi -i sine=frequency=440:sample_rate=48000:duration=0.5 diff --git a/rs/moq-srt/src/ts.rs b/rs/moq-srt/src/ts.rs index 0c9172aebd..4857465e2e 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -375,13 +375,41 @@ mod tests { seal(s) } - /// The PSI of a two-program multiplex, each program carrying one private PES stream - /// (carried verbatim, so the catalog lists it without waiting on media): program 1's on - /// PID 0x101, program 2's on 0x201. + /// One TS packet carrying a whole private PES with a PTS, stuffed in its adaptation field + /// so the PES ends exactly at its declared length. + fn pes_packet(pid: u16, pts: u64) -> Vec { + let mut pes = vec![0x00, 0x00, 0x01, 0xbd, 0x00, 10, 0x80, 0x80, 0x05]; + pes.extend_from_slice(&[ + 0x21 | (((pts >> 30) & 0x07) << 1) as u8, + (pts >> 22) as u8, + 0x01 | (((pts >> 15) & 0x7f) << 1) as u8, + (pts >> 7) as u8, + 0x01 | ((pts & 0x7f) << 1) as u8, + ]); + pes.extend_from_slice(&[0xde, 0xad]); + let stuffing = 188 - 4 - pes.len(); + let mut p = vec![ + 0x47, + 0x40 | (pid >> 8) as u8, + pid as u8, + 0x30, + (stuffing - 1) as u8, + 0x00, + ]; + p.resize(4 + stuffing, 0xff); + p.extend_from_slice(&pes); + p + } + + /// A two-program multiplex, each program carrying one private PES stream (carried + /// verbatim) and its first PES, which anchors the clock and so publishes the catalog: + /// program 1's on PID 0x101, program 2's on 0x201. fn two_programs() -> Bytes { let mut ts = psi_packet(0x0000, &pat(&[(1, 0x0100), (2, 0x0200)])); ts.extend_from_slice(&psi_packet(0x0100, &pmt(1, 0x06, 0x0101))); ts.extend_from_slice(&psi_packet(0x0200, &pmt(2, 0x06, 0x0201))); + ts.extend_from_slice(&pes_packet(0x0101, 90_000)); + ts.extend_from_slice(&pes_packet(0x0201, 90_000)); ts.into() }