fix(mux): data tracks stamp on the catalog's current clock

doc/lib/rs/moq-mux.md

@@ -100,7 +100,9 @@ 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. A clock set with
+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
 `Config::with_clock` is never re-anchored, for a recording whose zero names its
 real start.
 

quest/m1/README.md

@@ -66,7 +66,6 @@ QUIC studies there on that rule.
 - [Audio capture without ALSA link](/quest/m1/capture-alsa-link.md) - moq-audio capture and playback build on Linux without linking libasound
 - [Capture by default](/quest/m1/capture-default.md) - moq-video and moq-audio build `capture` by default, so pre-merge checks test it and the capture gate goes away
 - [Ship capture and playback](/quest/m1/cli-packaging.md) - a released moq binary can capture and play, which no distribution currently enables
-- [Data track clock](/quest/m1/data-track-clock.md) - JSON and binary data tracks stamp on the catalog's clock at write time, matching the media's anchored clock
 - [Go and Dart doc samples](/quest/m1/doc-samples-go-dart.md) - Go and Dart doc samples compile against their wrappers
 - [Publishing requires a timestamp](/quest/m1/publish-timestamp.md) - every Rust and binding publish API takes a timestamp; nothing fills in now for the caller
 - [JS publishing requires a timestamp](/quest/m1/js-publish-timestamp.md) - the same in @moq/json, @moq/flate, and @moq/net

quest/m1/publish-timestamp.md

@@ -58,5 +58,4 @@ Public API: breaking, on `dev`. Wire: none.
 
 ## Related
 
-- [Data track clock](/quest/m1/data-track-clock.md) - the moq-mux mapping this keeps reads the catalog's clock at write time
 - [JS publishing requires a timestamp](/quest/m1/js-publish-timestamp.md) - the same change in the JS packages

quest/m3/watch-data-sync.md

@@ -19,7 +19,3 @@ releases it.
 - In m3 until an application needs synchronized data playback, such as
   teleop telemetry beside video; until then, a raw consumer reads payloads as
   they arrive.
-
-## Required
-
-- [Data track clock](/quest/m1/data-track-clock.md) - data timestamps share the media clock mapping, so the playhead can release them

rs/moq-mux/src/binary.rs

@@ -138,8 +138,6 @@ fn prepare(config: &mut impl AsMut<BinaryConfig>, mode: Mode) -> crate::Result<b
 pub struct Snapshot<E: CatalogExt = ()> {
 	inner: moq_flate::snapshot::Producer,
 	listing: Listing,
-	/// Maps a payload's capture instant onto the broadcast timeline.
-	clock: crate::Clock,
 	/// Which catalog the entry lives in. The entry's own type is erased by `Listing`.
 	_catalog: PhantomData<fn() -> E>,
 }
@@ -155,12 +153,10 @@ impl<E: CatalogExt> Snapshot<E> {
 			binary.compression = moq_flate::Compression::Deflate;
 		}
 		let inner = moq_flate::snapshot::Producer::new(track, binary);
-		let clock = rendition.clock();
 		let listing = Listing::new(rendition, config)?;
 		Ok(Self {
 			inner,
 			listing,
-			clock,
 			_catalog: PhantomData,
 		})
 	}
@@ -180,7 +176,7 @@ impl<E: CatalogExt> Snapshot<E> {
 	/// A payload timed with its capture instant is written at that time and measures the entry's
 	/// `jitter` and `delay`; one ahead of now is refused before anything is written.
 	pub fn update(&mut self, payload: impl Into<Timed<Bytes, Instant>>) -> crate::Result<()> {
-		let (payload, captured) = self.clock.stamp(payload.into())?;
+		let (payload, captured) = self.listing.stamp(payload.into())?;
 		let size = self.inner.update(payload)?;
 		self.listing.record(size, captured)
 	}
@@ -207,8 +203,6 @@ pub struct Stream<E: CatalogExt = ()> {
 	/// entry advertising a track that can no longer accept records only misleads a consumer that
 	/// discovers it afterwards.
 	listing: Option<Listing>,
-	/// Maps a payload's capture instant onto the broadcast timeline.
-	clock: crate::Clock,
 	/// Which catalog the entry lives in. The entry's own type is erased by `Listing`.
 	_catalog: PhantomData<fn() -> E>,
 }
@@ -224,13 +218,11 @@ impl<E: CatalogExt> Stream<E> {
 			binary.compression = moq_flate::Compression::Deflate;
 		}
 		let inner = moq_flate::stream::Producer::new(track, binary);
-		let clock = rendition.clock();
 		let listing = Listing::new(rendition, config)?;
 		Ok(Self {
 			inner,
 			name: listing.name().to_string(),
 			listing: Some(listing),
-			clock,
 			_catalog: PhantomData,
 		})
 	}
@@ -255,7 +247,11 @@ impl<E: CatalogExt> Stream<E> {
 	/// error publishing the measured bitrate is returned after the payload was written, so the track
 	/// stays open and a retry would duplicate it.
 	pub fn append(&mut self, payload: impl Into<Timed<Bytes, Instant>>) -> crate::Result<()> {
-		let (payload, captured) = self.clock.stamp(payload.into())?;
+		let (payload, captured) = match &mut self.listing {
+			Some(listing) => listing.stamp(payload.into())?,
+			// A failed write already ended the log, which refuses this payload whatever its time.
+			None => (Timed::from(payload.into().value), None),
+		};
 		let size = match self.inner.append(payload) {
 			Ok(size) => size,
 			Err(err) => {
@@ -407,6 +403,18 @@ mod test {
 			.clone()
 	}
 
+	/// Drain the timestamp of every frame a subscriber currently has, without blocking.
+	fn stamps(subscriber: &mut moq_net::track::Subscriber) -> Vec<u128> {
+		let waiter = kio::Waiter::noop();
+		let mut stamps = Vec::new();
+		while let Poll::Ready(Ok(Some(mut group))) = subscriber.poll_recv_group(&waiter) {
+			while let Poll::Ready(Ok(Some(frame))) = group.poll_read_frame(&waiter) {
+				stamps.push(frame.timestamp.as_millis());
+			}
+		}
+		stamps
+	}
+
 	/// Drain every payload a consumer currently has, without blocking.
 	///
 	/// One loop whichever mode the publisher chose, which is the point of the single `Consumer`.
@@ -501,13 +509,7 @@ mod test {
 			.unwrap();
 		let after = catalog.clock().now();
 
-		let waiter = kio::Waiter::noop();
-		let mut stamps = Vec::new();
-		while let Poll::Ready(Ok(Some(mut group))) = subscriber.poll_recv_group(&waiter) {
-			while let Poll::Ready(Ok(Some(frame))) = group.poll_read_frame(&waiter) {
-				stamps.push(frame.timestamp.as_millis());
-			}
-		}
+		let stamps = stamps(&mut subscriber);
 		assert_eq!(stamps.len(), 2);
 		assert!(
 			// The track stores milliseconds.
@@ -516,6 +518,61 @@ mod test {
 		);
 	}
 
+	/// A track created before an importer's first frame stamps on the clock that frame anchors, not
+	/// the one the catalog started with.
+	#[test]
+	fn a_write_follows_the_anchored_clock() {
+		let (mut broadcast, mut catalog) = catalog();
+		let mut snapshot = catalog
+			.binary_snapshot(track(&mut broadcast, "thumbnail"), Config::default())
+			.unwrap();
+		let mut stream = catalog
+			.binary_stream(track(&mut broadcast, "telemetry"), Config::default())
+			.unwrap();
+		let (mut snapshots, mut streams) = (snapshot.consume(), stream.consume());
+
+		// The stream starts an hour in, far from the ten seconds a fresh clock reads.
+		let first = moq_net::Timestamp::from_secs(3600).unwrap();
+		catalog.anchor(first).unwrap();
+		snapshot.update(&b"untimed"[..]).unwrap();
+		stream
+			.append(Timed::from(&b"timed"[..]).at(std::time::Instant::now()))
+			.unwrap();
+		let after = catalog.clock().now();
+
+		for stamps in [stamps(&mut snapshots), stamps(&mut streams)] {
+			assert_eq!(stamps.len(), 1);
+			assert!(
+				first.as_millis() <= stamps[0] && stamps[0] <= after.as_millis(),
+				"{first:?} {stamps:?} {after:?}"
+			);
+		}
+	}
+
+	/// An anchor stepping the clock back is a break in the timeline, not jitter. A fresh clock
+	/// reads ten seconds, and a stream starting at zero anchors it there.
+	#[test]
+	fn an_anchor_is_not_jitter() {
+		let (mut broadcast, mut catalog) = catalog();
+		let mut telemetry = catalog
+			.binary_stream(track(&mut broadcast, "telemetry"), Config::default())
+			.unwrap();
+
+		telemetry
+			.append(Timed::from(&b"before"[..]).at(std::time::Instant::now()))
+			.unwrap();
+		catalog.anchor(moq_net::Timestamp::ZERO).unwrap();
+		telemetry
+			.append(Timed::from(&b"after"[..]).at(std::time::Instant::now()))
+			.unwrap();
+
+		let jitter = entry(&catalog, "telemetry").jitter;
+		assert!(
+			jitter.is_none_or(|jitter| jitter < std::time::Duration::from_secs(1)),
+			"{jitter:?}"
+		);
+	}
+
 	#[test]
 	fn a_stream_track_roundtrips() {
 		let (mut broadcast, catalog) = catalog();

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

@@ -37,21 +37,24 @@ pub(crate) struct Listing {
 	rendition: Box<dyn Owned>,
 	/// Measures `delay` against the catalog's other renditions, media included.
 	estimator: Estimator,
+	/// The catalog clock as of the last write. Read again on every write, since an importer's first
+	/// frame re-anchors the catalog's clock however long after this track was created.
+	clock: crate::Clock,
 }
 
 /// The parts of a [`Rendition`] a [`Listing`] uses, without its config type.
 trait Owned: Send + Sync {
 	fn name(&self) -> &str;
-	fn timestamp(&self) -> crate::Result<moq_net::Timestamp>;
+	fn clock(&self) -> crate::Clock;
 	fn estimate(&mut self, estimate: super::Estimate) -> crate::Result<()>;
 }
 
 impl<E: CatalogExt, C: RenditionConfig<E>> Owned for Rendition<E, C> {
 	fn name(&self) -> &str {
 		Rendition::name(self)
 	}
-	fn timestamp(&self) -> crate::Result<moq_net::Timestamp> {
-		Rendition::timestamp(self, None)
+	fn clock(&self) -> crate::Clock {
+		Rendition::clock(self)
 	}
 	fn estimate(&mut self, estimate: super::Estimate) -> crate::Result<()> {
 		Rendition::estimate(self, estimate)
@@ -67,6 +70,7 @@ impl Listing {
 		rendition.set(config)?;
 		Ok(Self {
 			estimator: rendition.estimator(),
+			clock: rendition.clock(),
 			rendition: Box::new(rendition),
 		})
 	}
@@ -76,10 +80,25 @@ impl Listing {
 		self.rendition.name()
 	}
 
+	/// Stamp a payload on the catalog's current clock, at its capture instant or else now. Also
+	/// returns the capture time, if there was one.
+	pub(crate) fn stamp<P>(
+		&mut self,
+		timed: moq_net::Timed<P, std::time::Instant>,
+	) -> crate::Result<(moq_net::Timed<P>, Option<moq_net::Timestamp>)> {
+		let clock = self.rendition.clock();
+		// A re-anchor moves this track's timeline, which measured across would read as jitter.
+		if clock.wall() != self.clock.wall() {
+			self.estimator.discontinuity();
+		}
+		self.clock = clock;
+		clock.stamp(timed)
+	}
+
 	/// Measure a frame of `bytes` encoded bytes, just written, captured at `captured` on the
 	/// broadcast clock if known.
 	pub(crate) fn record(&mut self, bytes: usize, captured: Option<moq_net::Timestamp>) -> crate::Result<()> {
-		let now = self.rendition.timestamp()?;
+		let now = self.clock.now();
 		let flush = captured.map(|captured| (captured, std::time::Instant::now()));
 		self.record_at(now, bytes, flush)
 	}

rs/moq-mux/src/json.rs

@@ -165,8 +165,6 @@ fn delta_ratio_of<C: std::any::Any>(config: &C) -> Option<u32> {
 pub struct Snapshot<T, E: CatalogExt = ()> {
 	inner: moq_json::snapshot::Producer<T>,
 	listing: Listing,
-	/// Maps a value's capture instant onto the broadcast timeline.
-	clock: crate::Clock,
 	/// Which catalog the entry lives in. The entry's own type is erased by `Listing`.
 	_catalog: PhantomData<fn() -> E>,
 }
@@ -192,12 +190,10 @@ impl<T: Serialize, E: CatalogExt> Snapshot<T, E> {
 			json.delta_ratio = delta_ratio;
 		}
 		let inner = moq_json::snapshot::Producer::new(track, json);
-		let clock = rendition.clock();
 		let listing = Listing::new(rendition, config)?;
 		Ok(Self {
 			inner,
 			listing,
-			clock,
 			_catalog: PhantomData,
 		})
 	}
@@ -226,7 +222,7 @@ impl<T: Serialize, E: CatalogExt> Snapshot<T, E> {
 	where
 		T: 'a,
 	{
-		let (value, captured) = self.clock.stamp(value.into())?;
+		let (value, captured) = self.listing.stamp(value.into())?;
 		match self.inner.update(value)? {
 			Some(size) => self.listing.record(size, captured),
 			None => Ok(()),
@@ -255,8 +251,6 @@ pub struct Stream<T, E: CatalogExt = ()> {
 	/// entry advertising a track that can no longer accept records only misleads a consumer that
 	/// discovers it afterwards.
 	listing: Option<Listing>,
-	/// Maps a record's capture instant onto the broadcast timeline.
-	clock: crate::Clock,
 	/// Which catalog the entry lives in. The entry's own type is erased by `Listing`.
 	_catalog: PhantomData<fn() -> E>,
 }
@@ -272,13 +266,11 @@ impl<T: Serialize, E: CatalogExt> Stream<T, E> {
 			json.compression = moq_json::Compression::Deflate;
 		}
 		let inner = moq_json::stream::Producer::new(track, json);
-		let clock = rendition.clock();
 		let listing = Listing::new(rendition, config)?;
 		Ok(Self {
 			inner,
 			name: listing.name().to_string(),
 			listing: Some(listing),
-			clock,
 			_catalog: PhantomData,
 		})
 	}
@@ -310,7 +302,11 @@ impl<T: Serialize, E: CatalogExt> Stream<T, E> {
 	where
 		T: 'a,
 	{
-		let (value, captured) = self.clock.stamp(value.into())?;
+		let (value, captured) = match &mut self.listing {
+			Some(listing) => listing.stamp(value.into())?,
+			// A failed write already ended the log, which refuses this record whatever its time.
+			None => (Timed::from(value.into().value), None),
+		};
 		let size = match self.inner.append(value) {
 			Ok(size) => size,
 			Err(err) => {
@@ -517,6 +513,44 @@ mod test {
 		assert_eq!(drain(consumer), vec![json!({ "live": true })]);
 	}
 
+	/// A track created before an importer's first frame stamps on the clock that frame anchors, not
+	/// the one the catalog started with.
+	#[test]
+	fn a_write_follows_the_anchored_clock() {
+		let (mut broadcast, mut catalog) = catalog();
+		let mut status = catalog
+			.json_snapshot::<Value>(track(&mut broadcast, "status"), Config::default())
+			.unwrap();
+		let mut chat = catalog
+			.json_stream::<Value>(track(&mut broadcast, "chat"), Config::default())
+			.unwrap();
+		let mut subscribers = [status.consume(), chat.consume()];
+
+		// The stream starts an hour in, far from the ten seconds a fresh clock reads.
+		let first = moq_net::Timestamp::from_secs(3600).unwrap();
+		catalog.anchor(first).unwrap();
+		status.update(&json!({ "live": true })).unwrap();
+		chat.append(Timed::from(&json!({ "n": 1 })).at(std::time::Instant::now()))
+			.unwrap();
+		let after = catalog.clock().now();
+
+		let waiter = kio::Waiter::noop();
+		for subscriber in &mut subscribers {
+			let mut stamps = Vec::new();
+			while let Poll::Ready(Ok(Some(mut group))) = subscriber.poll_recv_group(&waiter) {
+				while let Poll::Ready(Ok(Some(frame))) = group.poll_read_frame(&waiter) {
+					stamps.push(frame.timestamp.as_millis());
+				}
+			}
+			assert_eq!(stamps.len(), 1);
+			assert!(
+				// The track stores milliseconds.
+				first.as_millis() <= stamps[0] && stamps[0] <= after.as_millis(),
+				"{first:?} {stamps:?} {after:?}"
+			);
+		}
+	}
+
 	/// `delta_ratio` is an encoder setting on [`Config`], not a catalog field. A ratio of 0
 	/// publishes each value as its own group; a positive ratio keeps the next value in that group.
 	#[test]