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..18f2047d5d 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) => { @@ -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::(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]