From 9ed451cd18fc1a568c1e3a31df4855f66caaf94d Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Thu, 1 Oct 2026 11:38:57 +0200 Subject: [PATCH 1/3] quest: claim quest/m1/data-track-clock Co-Authored-By: Claude Opus 5.5 From 96388033b8f8386f475fb59705a1567513e32eeb Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Thu, 1 Oct 2026 12:24:57 +0200 Subject: [PATCH 2/3] fix(mux): data tracks stamp on the catalog's current clock A JSON or binary producer copied the catalog clock at creation, so one created before an importer's first frame never saw the anchor and kept stamping on the catalog's starting clock. Listing now reads the catalog clock on every write, and resets its estimator when the wall mapping changes so a re-anchor is not measured as jitter. Co-Authored-By: Claude Opus 5.5 --- doc/lib/rs/moq-mux.md | 4 +- quest/m1/README.md | 1 - quest/m1/data-track-clock.md | 18 ------- quest/m1/publish-timestamp.md | 1 - quest/m3/watch-data-sync.md | 4 -- rs/moq-mux/src/binary.rs | 91 +++++++++++++++++++++++++++------- rs/moq-mux/src/catalog/data.rs | 27 ++++++++-- rs/moq-mux/src/json.rs | 16 +++--- 8 files changed, 106 insertions(+), 56 deletions(-) delete mode 100644 quest/m1/data-track-clock.md diff --git a/doc/lib/rs/moq-mux.md b/doc/lib/rs/moq-mux.md index d240598c61..1037c315d0 100644 --- a/doc/lib/rs/moq-mux.md +++ b/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. diff --git a/quest/m1/README.md b/quest/m1/README.md index 681654ad77..a326dd7738 100644 --- a/quest/m1/README.md +++ b/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 diff --git a/quest/m1/data-track-clock.md b/quest/m1/data-track-clock.md deleted file mode 100644 index fcb277566b..0000000000 --- a/quest/m1/data-track-clock.md +++ /dev/null @@ -1,18 +0,0 @@ -# [XS] Data tracks stamp on the catalog's current clock - -## Goal - -A JSON or binary data track publishes timestamps on the same clock mapping as -the media next to it, however early it was created. Today a data track copies -the catalog clock when it's created. An importer sets that clock from its -first frame (remove-live, #4543), so a data track created before that frame -maps its timestamps to a different clock than the media. - -## Plan - -Remove live() (#4543) is done on `dev`, and the first-frame clock anchor -exists only there, so this targets `dev`. - -Have data tracks read the catalog's clock when they stamp, not a copy made -when they were created. Keep it crate-private if possible. Test: a data track -created before an importer's first frame stamps on the anchored clock. diff --git a/quest/m1/publish-timestamp.md b/quest/m1/publish-timestamp.md index 0c4aa6b8b0..78b11642cb 100644 --- a/quest/m1/publish-timestamp.md +++ b/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 diff --git a/quest/m3/watch-data-sync.md b/quest/m3/watch-data-sync.md index 16a287344c..c48272e879 100644 --- a/quest/m3/watch-data-sync.md +++ b/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 diff --git a/rs/moq-mux/src/binary.rs b/rs/moq-mux/src/binary.rs index 600c9b6977..a00ecae38a 100644 --- a/rs/moq-mux/src/binary.rs +++ b/rs/moq-mux/src/binary.rs @@ -138,8 +138,6 @@ fn prepare(config: &mut impl AsMut, mode: Mode) -> crate::Result { 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 E>, } @@ -155,12 +153,10 @@ impl Snapshot { 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 Snapshot { /// 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>) -> 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 { /// entry advertising a track that can no longer accept records only misleads a consumer that /// discovers it afterwards. listing: Option, - /// 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 E>, } @@ -224,13 +218,11 @@ impl Stream { 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 Stream { /// 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>) -> 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 { + 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(); diff --git a/rs/moq-mux/src/catalog/data.rs b/rs/moq-mux/src/catalog/data.rs index 4f84d42f80..b8513cd312 100644 --- a/rs/moq-mux/src/catalog/data.rs +++ b/rs/moq-mux/src/catalog/data.rs @@ -37,12 +37,15 @@ pub(crate) struct Listing { rendition: Box, /// 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; + fn clock(&self) -> crate::Clock; fn estimate(&mut self, estimate: super::Estimate) -> crate::Result<()>; } @@ -50,8 +53,8 @@ impl> Owned for Rendition { fn name(&self) -> &str { Rendition::name(self) } - fn timestamp(&self) -> crate::Result { - 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

( + &mut self, + timed: moq_net::Timed, + ) -> crate::Result<(moq_net::Timed

, Option)> { + 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) -> 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) } diff --git a/rs/moq-mux/src/json.rs b/rs/moq-mux/src/json.rs index 96deb633c2..7f8c44ca44 100644 --- a/rs/moq-mux/src/json.rs +++ b/rs/moq-mux/src/json.rs @@ -165,8 +165,6 @@ fn delta_ratio_of(config: &C) -> Option { pub struct Snapshot { inner: moq_json::snapshot::Producer, 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 E>, } @@ -192,12 +190,10 @@ impl Snapshot { 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 Snapshot { 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 { /// entry advertising a track that can no longer accept records only misleads a consumer that /// discovers it afterwards. listing: Option

, - /// 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 E>, } @@ -272,13 +266,11 @@ impl Stream { 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 Stream { 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) => { From c45a5ea0951b7c2000be87b827710f5cd3e2c033 Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Thu, 1 Oct 2026 13:45:58 +0200 Subject: [PATCH 3/3] test(mux): JSON data tracks follow the anchored clock too Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/json.rs | 38 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/rs/moq-mux/src/json.rs b/rs/moq-mux/src/json.rs index 7f8c44ca44..18f2047d5d 100644 --- a/rs/moq-mux/src/json.rs +++ b/rs/moq-mux/src/json.rs @@ -513,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::(track(&mut broadcast, "status"), Config::default()) + .unwrap(); + let mut chat = catalog + .json_stream::(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]