fix(moq-mux): importers publish the catalog at their first frame

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.
 

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`,

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

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

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

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

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

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

rs/moq-mux/src/catalog/producer.rs

@@ -277,8 +277,9 @@ impl<E: CatalogExt> Config<E> {
 	///
 	/// 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<E: CatalogExt> Producer<E> {
 	///
 	/// 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(());

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

@@ -643,6 +643,8 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 
 	/// 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()?;
 		}

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

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

@@ -33,9 +33,9 @@ pub struct Import<E: crate::catalog::hang::CatalogExt = ()> {
 	/// The catalog being produced
 	catalog: crate::catalog::Producer<E>,
 
-	/// 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<crate::catalog::Reserved<E>>,
 
 	// Which track roles to publish. `None` imports every supported track.
@@ -275,9 +275,16 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 	}
 
 	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<E: crate::catalog::hang::CatalogExt> Import<E> {
 			);
 		}
 
-		// 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<E: crate::catalog::hang::CatalogExt> Import<E> {
 			// 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<E: crate::catalog::hang::CatalogExt> Import<E> {
 impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 	/// 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()?;