From d327ae20290a37a53446f2ab30a199d8737929fa Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 00:21:35 -0700 Subject: [PATCH 1/2] chore(quest): claim cli-import-clock From 6ddfab7eeddc3853ef3631043848b2a993faf121 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 00:46:27 -0700 Subject: [PATCH 2/2] feat(cli): publish stdin imports on the broadcast clock `moq import ts|fmp4|flv` published source timestamps verbatim against a wall sampled at startup, so a late first frame or a large starting PTS advertised the wrong wall time, and a restarted source rewound or was refused. The fMP4, TS, and FLV importers gain `live()`, which translates onto the catalog's broadcast clock through one shared mapping per source: the first frame is live on arrival, every track keeps its source offset, and a restart continues forward after the real idle gap. fMP4 passthrough rewrites `tfdt`. `SourceMap` now runs on the same per-source anchor, and `Clock::new` puts PTS zero ten seconds back so earlier-muxed frames still map. Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 7 + doc/lib/rs/moq-mux.md | 8 + ...l-clock-latency-target-for-synchronized.md | 4 - quest/m1/README.md | 1 - quest/m1/cli-import-clock.md | 33 -- quest/m2/teleop/correlation.md | 1 - rs/moq-cli/src/publish.rs | 57 +++- rs/moq-mux/src/clock.rs | 321 ++++++++++++++---- rs/moq-mux/src/container/flv/import.rs | 44 ++- rs/moq-mux/src/container/flv/import_test.rs | 142 ++++++++ rs/moq-mux/src/container/fmp4/import.rs | 39 ++- rs/moq-mux/src/container/fmp4/import_test.rs | 154 +++++++++ rs/moq-mux/src/container/test_util.rs | 62 ++++ rs/moq-mux/src/container/ts/import.rs | 86 +++-- rs/moq-mux/src/container/ts/import_test.rs | 107 ++++++ 15 files changed, 930 insertions(+), 136 deletions(-) delete mode 100644 quest/m1/cli-import-clock.md diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 1376686f9d..43f26fdbcd 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -53,6 +53,13 @@ moq --connect https://relay.example.com/anon --broadcast my-stream.hang export t moq --connect "https://relay.example.com/rooms/1?jwt=$TOKEN" --broadcast alice.hang import ts ``` +The `ts`, `fmp4`, and `flv` imports publish on the broadcast clock the catalog +advertises, not the input's own timestamps. The first frame is stamped when it +arrives, every track keeps its offset from the others, and an input that +restarts its timestamps, such as a restarted encoder, continues forward +after the real gap rather than rewinding. So a feed whose PTS starts hours in, +or whose first frame arrives late, still names the right wall time. + MPEG-TS import carries H.264/H.265 and AAC/MP2/AC-3/E-AC-3, passes SCTE-35 and subtitle PIDs through as tracks, and round-trips the service tables. A `discontinuity_indicator` on the program's PCR PID is a system time-base reset, diff --git a/doc/lib/rs/moq-mux.md b/doc/lib/rs/moq-mux.md index a3fd8d42e8..5fe596a4fd 100644 --- a/doc/lib/rs/moq-mux.md +++ b/doc/lib/rs/moq-mux.md @@ -80,6 +80,14 @@ The producer sets the entry's `mode` and encodes the track with its `compression`. Read it back from `Catalog` and subscribe with `catalog::Entry::new(name, &entry.binary)`. +The fMP4, MPEG-TS, and FLV importers publish the source's own timestamps unless +built with `live()`, which translates them onto the catalog's broadcast clock: +the first frame is live on arrival, every track of the input shares that one +mapping, and a source that restarts its timestamps continues forward after the +real idle gap. fMP4 passthrough rewrites each fragment's `tfdt` to match. Use +it for a live feed with its own zero; publish verbatim only when the catalog's +clock (`Config::with_clock`) already names the source's zero. + ```bash cargo add moq-mux ``` diff --git a/quest/m1/2278-watch-absolute-wall-clock-latency-target-for-synchronized.md b/quest/m1/2278-watch-absolute-wall-clock-latency-target-for-synchronized.md index 80800d1b8a..53be1396c0 100644 --- a/quest/m1/2278-watch-absolute-wall-clock-latency-target-for-synchronized.md +++ b/quest/m1/2278-watch-absolute-wall-clock-latency-target-for-synchronized.md @@ -28,10 +28,6 @@ application knows whether remote clocks are synchronized. Verify application access using the built-in publisher integration, including a live-only broadcast with no archive timeline. -## Required - -- [CLI import clock](/quest/m1/cli-import-clock.md) - built-in publishers populate the mapping applications read - ## Closes - [#2278](https://github.com/moq-dev/moq/issues/2278) - close this issue when the quest finishes diff --git a/quest/m1/README.md b/quest/m1/README.md index 45e07d2450..c6ce5d66f4 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -33,7 +33,6 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Track demand](/quest/m1/track-demand.md) - Rust and JS watch a track's subscribers through `demand()` alone - [Broadcast close](/quest/m1/broadcast-close/README.md) - `close()` is the one way to end a broadcast in every language, a permanent retraction that leaves in-flight tracks alone - [Relay peer set](/quest/m1/relay-peer-set.md) - a wire consumer tells a client hop from a peer hop, and every mesh credential can mark a peer -- [CLI import clock](/quest/m1/cli-import-clock.md) - fMP4, TS, and FLV imports publish on the shared broadcast clock across restarts - [CLI inspection](/quest/m1/cli-inspect/README.md) - `moq ls` lists what is live and `moq fetch` reads a group over MoQ, and a guide shows how to inspect a relay - [JS caught up](/quest/m1/js-announce-caught-up.md) - @moq/net's announce consumer says when the initial set has landed, like Rust - [Bindings caught up](/quest/m1/announce-live-bindings.md) - moq-ffi, libmoq, and every wrapper yield the same flat announce event, `Live` included diff --git a/quest/m1/cli-import-clock.md b/quest/m1/cli-import-clock.md deleted file mode 100644 index 0aca2b75f7..0000000000 --- a/quest/m1/cli-import-clock.md +++ /dev/null @@ -1,33 +0,0 @@ -# [M] CLI imports publish on the broadcast clock - -## Goal - -`moq import` of fMP4, TS, and FLV publishes timestamps on the shared broadcast -clock, like native capture and `js/publish` already do, including source -restarts, late first frames, and real idle gaps. Today the imports publish -source PTS verbatim against a wall clock sampled at startup, so a TS feed with -a large starting PTS or a late first frame advertises the wrong wall time. - -## Plan - -Use `moq_mux::Clock` and `SourceMap` with the root catalog `clock`; this adds -no clock API or catalog field. Select each source's initial mapping once, -account for a delayed first frame, and translate source resets onto the same -monotonic clock while preserving real idle gaps. System-wall adjustments do not -retime a running broadcast or old archive records. Preserve allowed B-frame -ordering within a group. - -- fMP4 is passthrough, so translation must rewrite `tfdt`. -- A muxed source needs one mapping for all of its tracks, since interleaved - audio and video can step back further than `SourceMap::MAX_REORDER`. -- Keep conversion at the adapter boundary and refuse an unmappable source - explicitly. Discontinuity markers signal the existing playhead contract; - they do not replace the wall epoch. - -CI fixtures drive the import path, not only the clock helper: simultaneous -A/V, a late first frame, a restart to zero, a restart after idle, and retained -archive playback. Update the import docs. - -## Related - -- [GStreamer clock](/quest/m1/3021-moq-gst-anchor-generated-media-timelines-to-wall-clock.md) - separate source adapter diff --git a/quest/m2/teleop/correlation.md b/quest/m2/teleop/correlation.md index 779d6f6e07..bb8e60ac91 100644 --- a/quest/m2/teleop/correlation.md +++ b/quest/m2/teleop/correlation.md @@ -30,4 +30,3 @@ the same property that makes an MCAP recording valuable. ## Required - [Robot teleoperation primitive](/quest/m2/teleop/robot.md) -- [CLI import clock](/quest/m1/cli-import-clock.md) - publishers populate the fixed broadcast mapping used to join tracks diff --git a/rs/moq-cli/src/publish.rs b/rs/moq-cli/src/publish.rs index 5fa8036fc1..306480c9ae 100644 --- a/rs/moq-cli/src/publish.rs +++ b/rs/moq-cli/src/publish.rs @@ -255,6 +255,9 @@ impl Publish { /// `broadcast`. Announce the broadcast afterwards: this constructor creates /// the catalog tracks, so announcing after it lands the advertisement with /// the tracks already in place. + /// + /// Stdin is a live feed with its own zero, so the container importers translate its + /// timestamps onto the broadcast clock the catalog advertises (`live`). pub fn new( mut broadcast: moq_net::broadcast::Producer, format: &PublishFormat, @@ -267,7 +270,7 @@ impl Publish { if let PublishFormat::Ts = format { let config = config.with_catalog(moq_mux::catalog::hang::Catalog::::default()); let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config)?; - let ts = ts::Import::new(broadcast.clone(), catalog.reserve()); + let ts = ts::Import::new(broadcast.clone(), catalog.reserve()).live(); return Ok(Self { source: Source::Stream(PublishDecoder::Ts(Box::new(ts))), broadcast, @@ -286,12 +289,12 @@ impl Publish { }) } PublishFormat::Fmp4 => { - let fmp4 = fmp4::Import::new(broadcast.clone(), catalog.reserve()); + let fmp4 = fmp4::Import::new(broadcast.clone(), catalog.reserve()).live(); Source::Stream(PublishDecoder::Fmp4(Box::new(fmp4))) } PublishFormat::Ts => unreachable!("TS is handled above with the mpegts catalog extension"), PublishFormat::Flv => { - let flv = flv::Import::new(broadcast.clone(), catalog.reserve()); + let flv = flv::Import::new(broadcast.clone(), catalog.reserve()).live(); Source::Stream(PublishDecoder::Flv(Box::new(flv))) } }; @@ -757,6 +760,54 @@ mod tests { ); } + /// `moq import ts` publishes on the broadcast clock it advertises: a feed arriving a minute + /// after the broadcast began is live on arrival, not stamped with its own PTS (1.4s into bbb). + #[tokio::test] + async fn ts_import_publishes_on_the_broadcast_clock() { + let ago = Duration::from_secs(60); + let clock = moq_mux::Clock::at(std::time::Instant::now() - ago, std::time::SystemTime::now() - ago).unwrap(); + let broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let config = moq_mux::catalog::Config::default().with_clock(clock); + let mut publish = Publish::new(broadcast, &PublishFormat::Ts, config).unwrap(); + #[allow(irrefutable_let_patterns)] + let Source::Stream(decoder) = &mut publish.source else { + panic!("expected a stream source"); + }; + let before = clock.now(); + decoder.decode_chunk(BBB).unwrap(); + let after = clock.now(); + decoder.finish().unwrap(); + + let catalog = hang::catalog::Catalog::<()>::subscribe(&consumer) + .await + .unwrap() + .next() + .await + .unwrap() + .expect("a catalog"); + assert_eq!( + catalog.clock, + Some(clock.wall()), + "the advertised clock is the one stamped on" + ); + let (name, config) = catalog.video.renditions.iter().next().expect("a video rendition"); + let track = consumer.track(name).unwrap().subscribe(None).await.unwrap(); + let container = moq_mux::catalog::hang::Container::try_from(config).unwrap(); + let first = Consumer::new(track, container) + .read() + .await + .unwrap() + .expect("a video frame") + .timestamp; + // The PES that anchors the mapping need not be this frame: the mux spaces them apart. + let skew = Duration::from_secs(2).as_micros(); + assert!( + before.as_micros() - skew <= first.as_micros() && first.as_micros() <= after.as_micros() + skew, + "the first frame is live on arrival: {first:?} not in {before:?}..={after:?}" + ); + } + /// Read the first frame of a verbatim track back as raw bytes. async fn read_frame(consumer: &moq_net::broadcast::Consumer, name: &str) -> Vec { let track = consumer.track(name).unwrap().subscribe(None).await.unwrap(); diff --git a/rs/moq-mux/src/clock.rs b/rs/moq-mux/src/clock.rs index 2929b0591e..39961a2773 100644 --- a/rs/moq-mux/src/clock.rs +++ b/rs/moq-mux/src/clock.rs @@ -59,10 +59,24 @@ impl Clock { /// enough that the wall value stays within the JSON-safe integer range until the year 2255. pub const TIMESCALE: moq_net::Timescale = moq_net::Timescale::MICRO; - /// Start a clock anchored at the current instant, with PTS zero at the current wall time. + /// How far before construction a fresh clock puts PTS zero. + /// + /// A translated source anchors its first frame at the current instant, but the frames + /// muxed beside it can carry earlier timestamps: a B-frame presenting before the keyframe + /// decoded ahead of it, or audio leading video in the mux. Starting the clock this far back + /// leaves them room instead of landing before the broadcast began. + const LEAD: Duration = Duration::from_secs(10); + + /// Start a clock at the current instant, with PTS zero ten seconds earlier on both the + /// monotonic and the wall clock, so earlier-stamped frames of a source anchored now still map. pub fn new() -> Self { - Self::at(Instant::now(), SystemTime::now()) - .expect("the current wall time is representable as a broadcast clock") + let (now, wall) = (Instant::now(), SystemTime::now()); + // Shortly after boot the monotonic clock may not reach back that far; start at now then. + let (epoch, wall) = match (now.checked_sub(Self::LEAD), wall.checked_sub(Self::LEAD)) { + (Some(epoch), Some(wall)) => (epoch, wall), + _ => (now, wall), + }; + Self::at(epoch, wall).expect("the current wall time is representable as a broadcast clock") } /// Start a clock at an explicit monotonic epoch and wall time. @@ -122,7 +136,7 @@ impl Default for Clock { /// live edge, preserving the source's spacing from there on; a reset re-anchors forward, /// preserving the real idle gap measured on the broadcast's monotonic clock. Backwards steps /// within [`MAX_REORDER`](Self::MAX_REORDER) keep their offset, so permitted B-frame reordering -/// inside a group survives verbatim. +/// inside a group survives verbatim. Translated timestamps keep the source's timescale. /// /// The broadcast wall mapping is never touched: translating a reset is not a new epoch, and a /// discontinuity marker the adapter emits alongside is a delivery event the playhead reacts to, @@ -131,13 +145,8 @@ impl Default for Clock { /// Each publisher adapter owns one per source and wires its own restart detection to /// [`reset`](Self::reset); the automatic path only separates reordering from resets by size. pub struct SourceMap { - clock: Clock, - /// Broadcast micros minus source micros; `None` until the first frame anchors it. - offset: Option, - last_source: Option, - last_broadcast: Option, - /// `clock.now()` when the last frame was translated: the idle gap's start. - last_arrival: Option, + anchor: Anchor, + lane: Lane, } impl SourceMap { @@ -152,22 +161,19 @@ impl SourceMap { /// A translator onto `clock`, unanchored until the first frame. pub fn new(clock: Clock) -> Self { Self { - clock, - offset: None, - last_source: None, - last_broadcast: None, - last_arrival: None, + anchor: Anchor::new(clock), + lane: Lane::default(), } } /// The broadcast clock this source translates onto. pub fn clock(&self) -> Clock { - self.clock + self.anchor.clock } /// Translate `pts` onto the broadcast clock, sampling the arrival time. pub fn translate(&mut self, pts: moq_net::Timestamp) -> crate::Result { - self.translate_at(pts, self.clock.now().value()) + self.anchor.translate(&mut self.lane, pts) } /// Translate `pts` onto the broadcast clock, arriving at monotonic `now` micros. @@ -175,41 +181,7 @@ impl SourceMap { /// The deterministic core behind [`translate`](Self::translate): synthetic sources pin the /// arrival instants instead of sampling them. pub fn translate_at(&mut self, pts: moq_net::Timestamp, now: u64) -> crate::Result { - let src = pts.as_micros(); - - let broadcast = match self.offset { - Some(offset) => { - let mapped = src as i128 + offset; - if mapped < 0 { - return Err(crate::Error::UnmappableTimestamp(format!( - "{pts:?} lands before the broadcast began" - ))); - } - let mapped = u64::try_from(mapped).map_err(|_| { - crate::Error::UnmappableTimestamp(format!("{pts:?} lands outside the representable range")) - })?; - match self.last_broadcast { - Some(last) if mapped < last && last - mapped > Self::MAX_REORDER.as_micros() as u64 => { - // A source reset: re-anchor forward, counting the downtime as content. - self.reanchor(src, now)? - } - // Forward, steady, or reordered within a group: the offset stands. - _ => mapped, - } - } - // The first frame is live now; the source keeps its spacing from there. Rebasing by - // the frame's own PTS (rather than pretending it is timestamp zero) is what keeps a - // delayed first frame honest. - None => { - self.offset = Some(now as i128 - src as i128); - now - } - }; - - self.last_source = Some(src); - self.last_broadcast = Some(broadcast); - self.last_arrival = Some(now); - moq_net::Timestamp::from_micros(broadcast).map_err(crate::Error::from) + self.anchor.translate_at(&mut self.lane, pts, now) } /// Re-anchor after an explicitly detected source restart, preserving the idle gap. @@ -218,33 +190,171 @@ impl SourceMap { /// the next frame continues after everything published so far plus the downtime since the /// previous frame, instead of rewinding the broadcast. pub fn reset(&mut self, pts: moq_net::Timestamp) -> crate::Result { - self.reset_at(pts, self.clock.now().value()) + self.lane.restart(); + self.translate(pts) } /// [`reset`](Self::reset) with an explicit arrival instant, for synthetic sources. pub fn reset_at(&mut self, pts: moq_net::Timestamp, now: u64) -> crate::Result { + self.lane.restart(); + self.translate_at(pts, now) + } +} + +/// One source's mapping onto the broadcast clock, shared by every track the source muxes. +/// +/// Tracks of one source must share an offset or they drift apart by however far their first +/// frames' PTS differ. They can't share a single [`SourceMap`] either: interleaved audio and +/// video step back further than [`SourceMap::MAX_REORDER`], which would read as a reset. So +/// each track keeps its own [`Lane`] that detects its own backwards steps, and a restart any +/// lane detects moves the anchor once; the other lanes adopt that mapping when they restart too. +pub(crate) struct Anchor { + clock: Clock, + /// Broadcast micros minus source micros for the current generation; `None` until the first + /// frame anchors it. + offset: Option, + /// Bumped at each re-anchor, so a lane knows whether its restart was already applied. + generation: u64, + /// The latest broadcast micros published so far, by any lane: the idle gap's origin. + last_broadcast: Option, + /// Where the latest frames end, by any lane, so a restart never lands on one of them. + last_end: Option, + /// `clock.now()` when the last frame was translated: the idle gap's start. + last_arrival: Option, +} + +/// One track's position on its source's [`Anchor`]. +#[derive(Default)] +pub(crate) struct Lane { + /// The offset this lane translates with, in micros; `None` until it adopts one. + offset: Option, + generation: u64, + last_source: Option, + /// The shortest forward step this lane's source took: its frame duration, near enough. + step: Option, + /// The adapter observed a restart on this lane out of band. + restart: bool, +} + +impl Lane { + /// The next frame starts a new source timeline, however its PTS compares to the last. + pub(crate) fn restart(&mut self) { + self.restart = true; + } +} + +impl Anchor { + pub(crate) fn new(clock: Clock) -> Self { + Self { + clock, + offset: None, + generation: 0, + last_broadcast: None, + last_end: None, + last_arrival: None, + } + } + + /// Translate one of `lane`'s timestamps, sampling the arrival time. + pub(crate) fn translate(&mut self, lane: &mut Lane, pts: moq_net::Timestamp) -> crate::Result { + self.translate_at(lane, pts, self.clock.now().value()) + } + + /// Translate one of `lane`'s timestamps, arriving at monotonic `now` micros. + pub(crate) fn translate_at( + &mut self, + lane: &mut Lane, + pts: moq_net::Timestamp, + now: u64, + ) -> crate::Result { let src = pts.as_micros(); - let broadcast = self.reanchor(src, now)?; - self.last_source = Some(src); - self.last_broadcast = Some(broadcast); + + match self.offset { + // The first frame is live now; the source keeps its spacing from there. Rebasing by + // the frame's own PTS (rather than pretending it is timestamp zero) is what keeps a + // delayed first frame honest. + None => self.offset = Some(now as i128 - src as i128), + Some(_) => { + let stepped_back = lane + .last_source + .is_some_and(|last| last > src + SourceMap::MAX_REORDER.as_micros()); + // A restart another lane already applied is adopted, not applied twice. + if (lane.restart || stepped_back) && lane.offset.is_some() && lane.generation == self.generation { + self.reanchor(src, now); + } + if lane.restart || stepped_back { + lane.offset = None; + } + } + } + lane.restart = false; + + // A lane joining late, or following a restart, takes the source's current mapping. + let offset = *lane.offset.get_or_insert_with(|| { + lane.generation = self.generation; + self.offset.expect("anchored above") + }); + + let mapped = src as i128 + offset; + if mapped < 0 { + return Err(crate::Error::UnmappableTimestamp(format!( + "{pts:?} lands before the broadcast began" + ))); + } + // The broadcast clock counts in micros, so the mapping must be nameable there too. + u64::try_from(mapped) + .ok() + .and_then(|mapped| moq_net::Timestamp::from_micros(mapped).ok()) + .ok_or_else(|| { + crate::Error::UnmappableTimestamp(format!("{pts:?} lands outside the representable range")) + })?; + + // Keep the source's timescale: the offset is constant, so the spacing stays exact. + let scale = pts.scale(); + let shift = offset * scale.as_u64() as i128 / 1_000_000; + let value = u64::try_from(pts.value() as i128 + shift) + .map_err(|_| crate::Error::UnmappableTimestamp(format!("{pts:?} lands outside the representable range")))?; + let translated = moq_net::Timestamp::new(value, scale) + .map_err(|_| crate::Error::UnmappableTimestamp(format!("{pts:?} lands outside the representable range")))?; + + if let Some(step) = lane + .last_source + .and_then(|last| src.checked_sub(last)) + .filter(|step| *step > 0) + { + lane.step = Some(lane.step.map_or(step, |min| min.min(step))); + } + lane.last_source = Some(src); + + let start = mapped as u128; + self.last_broadcast = Some(self.last_broadcast.map_or(start, |last| last.max(start))); + self.extend_micros(start + lane.step.unwrap_or(0)); self.last_arrival = Some(now); - moq_net::Timestamp::from_micros(broadcast).map_err(crate::Error::from) + Ok(translated) + } + + /// Record that the broadcast has published up to `end`, e.g. a fragment's last sample end. + pub(crate) fn extend(&mut self, end: moq_net::Timestamp) { + self.extend_micros(end.as_micros()); } - /// Move the offset so `src` continues after the last broadcast plus the idle gap since the - /// previous arrival. Returns the rebased broadcast micros. - fn reanchor(&mut self, src: u128, now: u64) -> crate::Result { + fn extend_micros(&mut self, end: u128) { + self.last_end = Some(self.last_end.map_or(end, |last| last.max(end))); + } + + /// Move the anchor so `src` continues after everything published plus the idle gap since the + /// previous arrival: the real downtime for a paced source, and at least the last frames' end + /// for one arriving in a burst. + fn reanchor(&mut self, src: u128, now: u64) { let base = match (self.last_broadcast, self.last_arrival) { - (Some(last), Some(arrival)) => last as u128 + now.saturating_sub(arrival) as u128, - // Unanchored: the reset frame itself is live now. + (Some(last), Some(arrival)) => { + let idle = last + now.saturating_sub(arrival) as u128; + idle.max(self.last_end.unwrap_or(0)) + } _ => now as u128, }; self.offset = Some(base as i128 - src as i128); - let broadcast = - u64::try_from(base).map_err(|_| crate::Error::UnmappableTimestamp(format!("{base} is out of range")))?; - // Refuse a mapping that contradicts the range instead of publishing it. - moq_net::Timestamp::from_micros(broadcast)?; - Ok(broadcast) + self.generation += 1; } } @@ -427,6 +537,81 @@ mod tests { )); } + #[test] + fn fresh_clock_leaves_room_before_now() { + let clock = Clock::new(); + // PTS zero sits before construction, so a frame stamped a little before now still maps, + // and the mapping still names the current wall time. + assert!(clock.now().as_micros() >= Clock::LEAD.as_micros()); + let now = clock.wall_clock(clock.now()).unwrap(); + let drift = now + .duration_since(SystemTime::now()) + .unwrap_or_else(|err| err.duration()); + assert!(drift < Duration::from_secs(1), "wall + now is the current wall time"); + } + + #[test] + fn muxed_lanes_share_one_mapping() { + let clock = Clock::at(epoch(), moq_epoch()).unwrap(); + let mut anchor = Anchor::new(clock); + let (mut video, mut audio) = (Lane::default(), Lane::default()); + + // Video anchors live; audio, muxed 800ms earlier, joins on the same offset. + let v = anchor.translate_at(&mut video, us(10_800_000), 2_000_000).unwrap(); + assert_eq!(v.as_micros(), 2_000_000); + let a = anchor.translate_at(&mut audio, us(10_000_000), 2_000_000).unwrap(); + assert_eq!(a.as_micros(), 1_200_000); + + // Interleaving steps back further than a reorder across lanes, which is not a reset. + let v = anchor.translate_at(&mut video, us(11_800_000), 3_000_000).unwrap(); + assert_eq!(v.as_micros(), 3_000_000); + let a = anchor.translate_at(&mut audio, us(11_000_000), 3_000_000).unwrap(); + assert_eq!(a.as_micros(), 2_200_000); + } + + #[test] + fn muxed_restart_reanchors_once() { + let clock = Clock::at(epoch(), moq_epoch()).unwrap(); + let mut anchor = Anchor::new(clock); + let (mut video, mut audio) = (Lane::default(), Lane::default()); + + anchor.translate_at(&mut video, us(5_000_000), 1_000_000).unwrap(); + anchor.translate_at(&mut audio, us(5_000_000), 1_000_000).unwrap(); + + // The source restarts at zero after 4s idle: video notices first and moves the anchor to + // the last published instant plus the gap. + let v = anchor.translate_at(&mut video, us(0), 5_000_000).unwrap(); + assert_eq!(v.as_micros(), 5_000_000); + // Audio's own step back adopts that mapping rather than adding the gap again. + let a = anchor.translate_at(&mut audio, us(20_000), 5_020_000).unwrap(); + assert_eq!(a.as_micros(), 5_020_000); + + // An out-of-band restart flagged on every lane is applied once as well. + video.restart(); + audio.restart(); + let v = anchor.translate_at(&mut video, us(0), 6_000_000).unwrap(); + assert_eq!(v.as_micros(), 6_000_000); + let a = anchor.translate_at(&mut audio, us(0), 6_000_000).unwrap(); + assert_eq!(a.as_micros(), 6_000_000); + } + + #[test] + fn translation_keeps_the_source_timescale() { + let clock = Clock::at(epoch(), moq_epoch()).unwrap(); + let mut source = clock.source(); + let scale = moq_net::Timescale::new(90_000).unwrap(); + + let first = source + .translate_at(moq_net::Timestamp::new(3003, scale).unwrap(), 1_000_000) + .unwrap(); + let second = source + .translate_at(moq_net::Timestamp::new(6006, scale).unwrap(), 1_033_000) + .unwrap(); + // 90 kHz in, 90 kHz out, with the frame spacing exact in ticks. + assert_eq!(first.scale(), scale); + assert_eq!(second.value() - first.value(), 3003); + } + #[test] fn mapping_before_the_broadcast_began_is_refused() { let clock = Clock::at(epoch(), moq_epoch()).unwrap(); diff --git a/rs/moq-mux/src/container/flv/import.rs b/rs/moq-mux/src/container/flv/import.rs index 5ba6fc30e3..d4d7fe267b 100644 --- a/rs/moq-mux/src/container/flv/import.rs +++ b/rs/moq-mux/src/container/flv/import.rs @@ -80,6 +80,10 @@ pub struct Import { video: BTreeMap, /// Demuxed audio tracks keyed by RTMP track id. audio: BTreeMap, + + /// The source's mapping onto the broadcast clock, set by [`live`](Self::live). `None` + /// publishes the tag timestamps verbatim. + anchor: Option, } /// The demuxed video track plus its current catalog config, so a repeated @@ -89,12 +93,14 @@ struct VideoStream { config: VideoConfig, stalled: hang::catalog::stalled::Detector, last_source: Option, + lane: crate::clock::Lane, } /// The demuxed audio track plus its current catalog config. struct AudioStream { track: crate::container::Producer, config: AudioConfig, + lane: crate::clock::Lane, } impl Import { @@ -110,9 +116,22 @@ impl Import { header_seen: false, video: BTreeMap::new(), audio: BTreeMap::new(), + anchor: None, } } + /// Publish on the broadcast clock rather than the source's own tag timestamps. + /// + /// For a live feed with its own zero: the first frame is live on arrival, every track shares + /// that one mapping, and an encoder that restarts its timestamps continues forward after the + /// real idle gap. Without this, tag timestamps are published verbatim, which suits a source + /// already on the clock the catalog advertises + /// ([`Config::with_clock`](crate::catalog::Config::with_clock)). + pub fn live(mut self) -> Self { + self.anchor = Some(crate::clock::Anchor::new(self.catalog.clock())); + self + } + /// Select the container this importer wraps decoded media renditions in. /// /// [`Legacy`](hang::catalog::Container::Legacy) unless selected. It applies to every rendition @@ -470,9 +489,13 @@ impl Import { // A media frame means every sequence header has arrived (FLV sends config before data), so // the track set is declared; release the reservation to publish. self.initial_reservation = None; - let timestamp = Timestamp::from_millis(pts_ms as u64)?; let written = { let stream = self.video.get_mut(&track_id).expect("checked above"); + let timestamp = Timestamp::from_millis(pts_ms as u64)?; + let timestamp = match self.anchor.as_mut() { + Some(anchor) => anchor.translate(&mut stream.lane, timestamp)?, + None => timestamp, + }; match stream.track.write(Frame { timestamp, duration: None, @@ -502,8 +525,13 @@ impl Import { // A media frame means every sequence header has arrived (FLV sends config before data), so // the track set is declared; release the reservation to publish. self.initial_reservation = None; + let timestamp = Timestamp::from_millis(timestamp)?; + let timestamp = match self.anchor.as_mut() { + Some(anchor) => anchor.translate(&mut stream.lane, timestamp)?, + None => timestamp, + }; stream.track.write(Frame { - timestamp: Timestamp::from_millis(timestamp)?, + timestamp, duration: None, payload: Bytes::copy_from_slice(data), keyframe: true, @@ -537,6 +565,7 @@ impl Import { config, stalled: hang::catalog::stalled::Detector::new(), last_source: None, + lane: Default::default(), }, ); Ok(()) @@ -558,7 +587,14 @@ impl Import { Some(reserved) => reserved.audio(net_track, wire, config.clone())?, None => self.catalog.audio(net_track, wire, config.clone())?, }; - self.audio.insert(track_id, AudioStream { track: media, config }); + self.audio.insert( + track_id, + AudioStream { + track: media, + config, + lane: Default::default(), + }, + ); Ok(()) } @@ -628,9 +664,11 @@ impl Import { pub fn seek(&mut self, sequence: u64) -> crate::Result<()> { for stream in self.video.values_mut() { stream.track.seek(sequence)?; + stream.lane.restart(); } for stream in self.audio.values_mut() { stream.track.seek(sequence)?; + stream.lane.restart(); } Ok(()) } diff --git a/rs/moq-mux/src/container/flv/import_test.rs b/rs/moq-mux/src/container/flv/import_test.rs index a7cddfde56..ceb6a78897 100644 --- a/rs/moq-mux/src/container/flv/import_test.rs +++ b/rs/moq-mux/src/container/flv/import_test.rs @@ -664,3 +664,145 @@ async fn import_rejects_non_flv() { let buf = bytes::BytesMut::from(&b"NOTFLV\x00\x00\x00"[..]); assert!(importer.decode(&buf).is_err()); } + +/// One encoder session: sequence headers (after the file header when `header`), then `frames` +/// video frames 40ms apart whose composition offsets reorder like B-frames, interleaved with AAC +/// frames up to 300ms earlier in PTS, the way a muxer leads audio. +fn session(header: bool, start_ms: u32, frames: u32) -> Vec { + let mut out = if header { flv_header(0x05) } else { Vec::new() }; + + let mut vseq = vec![ + (super::FRAME_TYPE_KEY << 4) | super::VIDEO_CODEC_AVC, + super::AVC_SEQUENCE_HEADER, + 0, + 0, + 0, + ]; + vseq.extend_from_slice(&avcc()); + write_tag(&mut out, super::TAG_VIDEO, start_ms, &vseq); + let mut aseq = vec![super::AAC_AUDIO_TAG_HEADER, super::AAC_SEQUENCE_HEADER]; + aseq.extend_from_slice(&ASC); + write_tag(&mut out, super::TAG_AUDIO, start_ms, &aseq); + + for i in 0..frames { + let dts = start_ms + i * 40; + let frame_type = if i % 25 == 0 { + super::FRAME_TYPE_KEY + } else { + super::FRAME_TYPE_INTER + }; + // IPBPB...: P-frames present two slots late and the B-frames between them step back. + let cts: u8 = match i % 25 { + 0 => 40, + j if j % 2 == 1 => 80, + _ => 0, + }; + let mut video = vec![(frame_type << 4) | super::VIDEO_CODEC_AVC, super::AVC_NALU, 0, 0, cts]; + video.extend_from_slice(&[0, 0, 0, 5, 0x65, 0x88, 0x84, 0x21, 0x00]); + write_tag(&mut out, super::TAG_VIDEO, dts, &video); + + let mut audio = vec![super::AAC_AUDIO_TAG_HEADER, super::AAC_RAW]; + audio.extend_from_slice(&[0xde, 0xad, 0xbe, 0xef]); + write_tag(&mut out, super::TAG_AUDIO, dts + 20 - start_ms.min(300), &audio); + } + out +} + +/// What an FLV import published, plus the broadcast clock's reading around the first chunk. +struct Imported { + published: std::collections::BTreeMap>, + video: String, + before: u128, + after: u128, +} + +/// Import `chunks` in order, idling `idle` between them, on a clock that began `ago` earlier. +/// `live` translates onto that clock; otherwise tag timestamps publish verbatim. +async fn import(chunks: &[Vec], live: bool, ago: Duration, idle: Duration) -> Imported { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let config = crate::catalog::Config::default().with_clock(crate::container::test_util::late_clock(ago)); + let catalog = crate::catalog::Producer::new(&mut broadcast, config).unwrap(); + let clock = catalog.clock(); + let mut importer = Import::new(broadcast, catalog.reserve()); + if live { + importer = importer.live(); + } + + let before = clock.now().as_micros(); + let mut after = before; + for (i, chunk) in chunks.iter().enumerate() { + if i > 0 { + std::thread::sleep(idle); + } + importer.decode(chunk).unwrap(); + if i == 0 { + after = clock.now().as_micros(); + } + } + importer.finish().unwrap(); + + let snapshot = catalog.snapshot(); + Imported { + published: crate::container::test_util::published(&consumer, &snapshot).await, + video: snapshot.video.renditions.keys().next().unwrap().clone(), + before, + after, + } +} + +/// A feed an hour into its own timeline, arriving 30s after the broadcast began, publishes on the +/// broadcast clock: the first frame is live on arrival, and audio and video keep the one offset +/// their tags gave them, B-frame reordering included. +#[tokio::test] +async fn live_import_anchors_a_late_first_frame() { + let input = [session(true, 3_600_000, 50)]; + let ago = Duration::from_secs(30); + let verbatim = import(&input, false, ago, Duration::ZERO).await; + let live = import(&input, true, ago, Duration::ZERO).await; + + crate::container::test_util::common_offset(&verbatim.published, &live.published); + let first = live.published[&live.video][0]; + assert!( + (live.before..=live.after).contains(&first), + "the first frame is live on arrival: {first} not in {}..={}", + live.before, + live.after + ); +} + +/// An encoder restarting its timestamps at zero continues the broadcast forward: after the real +/// idle gap, and with every track moving onto the one new mapping. +#[tokio::test] +async fn live_import_restarts_forward_after_idle() { + for idle in [Duration::ZERO, Duration::from_millis(300)] { + let input = [session(true, 5_000, 50), session(false, 0, 50)]; + let live = import(&input, true, Duration::from_secs(30), idle).await; + + let last_before = live + .published + .values() + .map(|t| t[..50].iter().max().unwrap()) + .max() + .unwrap(); + let video = &live.published[&live.video]; + let gap = video[50] as i128 - *last_before as i128; + assert!( + gap >= idle.as_micros() as i128, + "the restart lands after the idle gap: {gap}us after {idle:?}" + ); + assert!( + gap < (idle + Duration::from_secs(5)).as_micros() as i128, + "the restart is not pushed further: {gap}us" + ); + + // The second session keeps its own A/V relationship on the new mapping. + let verbatim = import(&[session(true, 0, 50)], false, Duration::ZERO, Duration::ZERO).await; + let second = live + .published + .iter() + .map(|(name, t)| (name.clone(), t[50..].to_vec())) + .collect(); + crate::container::test_util::common_offset(&verbatim.published, &second); + } +} diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index 714845fef8..15d2cf26ea 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -89,6 +89,10 @@ pub struct Import { // Only the timeline report is anchored. Each fragment still carries its own timestamp on the // wire, and `Recorder::end` still reports real content time. segment_start: Option, + + // The source's mapping onto the broadcast clock, set by `live`. `None` publishes the source's + // decode times verbatim. + anchor: Option, } /// The catalog entry for one imported track, whichever section it lives in. @@ -132,6 +136,9 @@ struct Fmp4Track { // Sequence to use for the next group, set by `Import::seek`. pending_sequence: Option, + // This track's position on the source's broadcast-clock mapping. + lane: crate::clock::Lane, + // The segment this track's open group belongs to. A mismatch with `Import::segment` rolls the // group, which is what keeps audio on the same boundaries as video. segment: Option, @@ -174,9 +181,22 @@ impl Import { segment: 0, pending_timeline_cut: false, segment_start: None, + anchor: None, } } + /// Publish on the broadcast clock rather than the source's own decode times. + /// + /// For a live feed with its own zero: the first fragment is live on arrival, every track + /// shares that one mapping, and a source that restarts its decode times continues forward + /// after the real idle gap instead of being refused. Each fragment's `tfdt` is rewritten to + /// match. Without this, decode times are published verbatim, which suits a source already on + /// the clock the catalog advertises ([`Config::with_clock`](crate::catalog::Config::with_clock)). + pub fn live(mut self) -> Self { + self.anchor = Some(crate::clock::Anchor::new(self.catalog.clock())); + self + } + /// Declare that the next fragment starts a new segment, for callers that know the source's /// segmentation out of band (e.g. an HLS import following its playlist). /// @@ -347,6 +367,7 @@ impl Import { last_decode_time: None, sample_duration: None, pending_sequence: None, + lane: Default::default(), estimator: Estimator::new(), claim: crate::catalog::Claim::new(self.catalog.bandwidth()), }, @@ -698,8 +719,16 @@ impl Import { let default_sample_flags = trex.map(|trex| trex.default_sample_flags).unwrap_or_default(); let tfdt = traf.tfdt.as_ref().ok_or(Error::MissingTfdt)?; - let mut dts = tfdt.base_media_decode_time; let timescale = moq_net::Timescale::new(trak.mdia.mdhd.timescale as u64)?; + // The decode time this fragment is published at, and so rewritten into its `tfdt`. + let base_decode_time = match self.anchor.as_mut() { + Some(anchor) => { + let source = Timestamp::new(tfdt.base_media_decode_time, timescale)?; + anchor.translate(&mut track.lane, source)?.value() + } + None => tfdt.base_media_decode_time, + }; + let mut dts = base_decode_time; // Every fragment restates its decode time, so a stale one puts two different samples // on the same timestamp, which reads downstream as an undeclared hole. @@ -845,6 +874,9 @@ impl Import { // and ensuring trun.data_offset is Some(...) reserves 4 bytes per trun. for traf_mut in &mut adjusted_moof.traf { traf_mut.tfhd.base_data_offset = None; + traf_mut.tfdt = Some(mp4_atom::Tfdt { + base_media_decode_time: base_decode_time, + }); // A zero default/sample duration is "unknown", not "instantaneous": drop it so // the re-emitted fragment carries no bogus zero that a decoder would honor. if traf_mut.tfhd.default_sample_duration == Some(0) { @@ -962,6 +994,10 @@ impl Import { track.estimator.write(timestamp, fragment_len); let end = max_end.ok_or(Error::MissingTrun)?; + if let Some(anchor) = self.anchor.as_mut() { + // A restart continues after this fragment's last sample, not merely its start. + anchor.extend(end); + } if let Some(recorder) = track.recorder.as_mut() { recorder.end(end); } @@ -1013,6 +1049,7 @@ impl Import { } track.pending_sequence = Some(sequence); track.last_decode_time = None; + track.lane.restart(); } Ok(()) } diff --git a/rs/moq-mux/src/container/fmp4/import_test.rs b/rs/moq-mux/src/container/fmp4/import_test.rs index 9f0108e436..9d54992e80 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/rs/moq-mux/src/container/fmp4/import_test.rs @@ -1192,3 +1192,157 @@ fn fragment_jitter_uses_sample_endpoints() { assert_eq!(jitter.as_nanos().div_ceil(1_000_000), expected); } } + +/// One encoder session of bbb-shaped fragments: `frames` 100ms video fragments from `start_us`, +/// a keyframe each second, interleaved with audio fragments up to 300ms earlier in PTS. +fn live_session(start_us: u64, frames: u64) -> Vec { + let (_, (video_id, video_scale), (audio_id, audio_scale)) = bbb_init(); + let lead = start_us.min(300_000); + let mut out = Vec::new(); + for j in 0..frames { + let pts = start_us + j * 100_000; + let video = sample(pts, j % 10 == 0, Some(100_000)); + out.extend_from_slice(&super::encode_fragment(info(video_id, video_scale, j as u32), &[video]).unwrap()); + let audio = sample(pts - lead, true, Some(100_000)); + out.extend_from_slice(&super::encode_fragment(info(audio_id, audio_scale, j as u32), &[audio]).unwrap()); + } + out +} + +/// What an fMP4 import published, plus the broadcast clock's reading around the first chunk. +struct LiveImport { + published: std::collections::BTreeMap>, + video: String, + clock: crate::Clock, + before: u128, + after: u128, + /// The first segment the broadcast timeline recorded, the index an archive replays from. + record: moq_net::Timestamp, +} + +/// Import bbb's init then `chunks`, idling `idle` between them, on a clock that began `ago` +/// earlier. `live` translates onto that clock; otherwise decode times publish verbatim. +async fn live_import( + chunks: &[Vec], + live: bool, + ago: std::time::Duration, + idle: std::time::Duration, +) -> LiveImport { + let (init, _, _) = bbb_init(); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let config = crate::catalog::Config::default().with_clock(crate::container::test_util::late_clock(ago)); + let mut catalog = crate::catalog::Producer::new(&mut broadcast, config).unwrap(); + let clock = catalog.clock(); + let mut fmp4 = crate::container::fmp4::Import::new(broadcast, catalog.reserve()); + if live { + fmp4 = fmp4.live(); + } + fmp4.decode(&init).unwrap(); + + let snapshot = catalog.snapshot(); + let section = snapshot.archive.clone().expect("the import advertises a timeline"); + let mut timeline = crate::timeline::Consumer::<()>::subscribe(&consumer, §ion) + .await + .unwrap(); + + let before = clock.now().as_micros(); + let mut after = before; + for (i, chunk) in chunks.iter().enumerate() { + if i > 0 { + std::thread::sleep(idle); + } + fmp4.decode(chunk).unwrap(); + if i == 0 { + after = clock.now().as_micros(); + } + } + fmp4.finish().unwrap(); + catalog.finish().unwrap(); + + let event = timeline.next().await.unwrap().expect("a recorded segment"); + let crate::timeline::Event::Push { entry, .. } = event else { + panic!("the first timeline event was not a segment"); + }; + + LiveImport { + published: crate::container::test_util::published(&consumer, &snapshot).await, + video: snapshot.video.renditions.keys().next().unwrap().clone(), + clock, + before, + after, + record: entry.pts, + } +} + +/// A feed an hour into its own decode timeline, arriving 30s after the broadcast began, publishes +/// on the broadcast clock: the `tfdt` a decoder reads is rewritten so the first fragment is live +/// on arrival, and every track moves by the one offset. The archive index records the same times, +/// so a replay names each segment's real wall time. +#[tokio::test] +async fn live_import_rewrites_tfdt_onto_the_broadcast_clock() { + let input = [live_session(3_600_000_000, 20)]; + let ago = std::time::Duration::from_secs(30); + let verbatim = live_import(&input, false, ago, std::time::Duration::ZERO).await; + let wall_before = std::time::SystemTime::now(); + let live = live_import(&input, true, ago, std::time::Duration::ZERO).await; + let wall_after = std::time::SystemTime::now(); + + crate::container::test_util::common_offset(&verbatim.published, &live.published); + let first = live.published[&live.video][0]; + assert!( + (live.before..=live.after).contains(&first), + "the first fragment is live on arrival: {first} not in {}..={}", + live.before, + live.after + ); + + // The timeline records at a coarser scale, so allow its rounding. + let tick = std::time::Duration::from_millis(1); + assert!( + first.abs_diff(live.record.as_micros()) < tick.as_micros(), + "the archive indexes the translated time" + ); + let wall = live.clock.wall_clock(live.record).unwrap(); + assert!( + wall_before - tick <= wall && wall <= wall_after, + "a replayed segment names the wall time it went live" + ); +} + +/// A source whose decode times restart at zero continues forward after the real idle gap rather +/// than being refused as non-monotonic, with every track moving onto one new mapping. +#[tokio::test] +async fn live_import_restarts_forward_after_idle() { + for idle in [std::time::Duration::ZERO, std::time::Duration::from_millis(300)] { + let input = [live_session(5_000_000, 20), live_session(0, 20)]; + let live = live_import(&input, true, std::time::Duration::from_secs(30), idle).await; + + // The restart lands the arrival gap after the first session's last fragment, and never on + // top of it: that fragment lasts 100ms. + let last = live + .published + .values() + .map(|t| t[..20].iter().max().unwrap()) + .max() + .unwrap(); + let gap = live.published[&live.video][20] as i128 - *last as i128; + let floor = idle.max(std::time::Duration::from_millis(100)); + assert!( + gap >= floor.as_micros() as i128, + "the restart lands after the idle gap: {gap}us after {idle:?}" + ); + assert!( + gap < (idle + std::time::Duration::from_secs(5)).as_micros() as i128, + "the restart is not pushed further: {gap}us" + ); + + let verbatim = live_import(&input[1..], false, std::time::Duration::ZERO, std::time::Duration::ZERO).await; + let second = live + .published + .iter() + .map(|(name, t)| (name.clone(), t[20..].to_vec())) + .collect(); + crate::container::test_util::common_offset(&verbatim.published, &second); + } +} diff --git a/rs/moq-mux/src/container/test_util.rs b/rs/moq-mux/src/container/test_util.rs index c95462d376..50d5f4de97 100644 --- a/rs/moq-mux/src/container/test_util.rs +++ b/rs/moq-mux/src/container/test_util.rs @@ -128,3 +128,65 @@ pub(crate) fn raw_frame(timestamp_us: u64, payload: &'static [u8], keyframe: boo duration: None, } } + +/// A broadcast clock that began `ago` before now, so a first frame arriving now reads as late. +pub(crate) fn late_clock(ago: std::time::Duration) -> crate::Clock { + crate::Clock::at(std::time::Instant::now() - ago, std::time::SystemTime::now() - ago).unwrap() +} + +/// Every frame timestamp, in micros, that each media rendition in `catalog` published, by track. +/// +/// Reads through each rendition's own container, so an fMP4 timestamp comes from the fragment's +/// `tfdt` rather than the wire header. The importer must have finished its tracks. +pub(crate) async fn published( + consumer: &moq_net::broadcast::Consumer, + catalog: &hang::catalog::Catalog, +) -> std::collections::BTreeMap> { + let mut containers = Vec::new(); + for (name, config) in &catalog.video.renditions { + containers.push((name.clone(), crate::catalog::hang::Container::try_from(config).unwrap())); + } + for (name, config) in &catalog.audio.renditions { + containers.push((name.clone(), crate::catalog::hang::Container::try_from(config).unwrap())); + } + + let mut out = std::collections::BTreeMap::new(); + for (name, container) in containers { + let replay = moq_net::track::Subscription::default().with_max_age(std::time::Duration::from_secs(3600)); + let track = consumer.track(&name).unwrap().subscribe(replay).await.unwrap(); + let mut reader = crate::container::Consumer::new(track, container); + let mut timestamps = Vec::new(); + while let Some(frame) = tokio::time::timeout(std::time::Duration::from_secs(5), reader.read()) + .await + .expect("the importer finished its tracks") + .unwrap() + { + timestamps.push(frame.timestamp.as_micros()); + } + out.insert(name, timestamps); + } + out +} + +/// The one offset every published timestamp moved by between a verbatim and a live import of the +/// same input, within a tick of rounding: one mapping for every track, so A/V sync and B-frame +/// order survive exactly. +pub(crate) fn common_offset( + verbatim: &std::collections::BTreeMap>, + live: &std::collections::BTreeMap>, +) -> i128 { + assert_eq!(verbatim.keys().count(), live.keys().count(), "the same tracks publish"); + let mut offset = None; + for (v, l) in verbatim.values().zip(live.values()) { + assert_eq!(v.len(), l.len(), "the same frames publish"); + for (v, l) in v.iter().zip(l) { + let delta = *l as i128 - *v as i128; + let first = *offset.get_or_insert(delta); + assert!( + (delta - first).abs() <= 1_000, + "every frame moves by one offset: {delta} vs {first}" + ); + } + } + offset.expect("frames were published") +} diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index d811e24cc1..0cfa35a533 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -134,6 +134,9 @@ pub struct Import { /// advances it, so a cue could be stamped with another program's PTS. last_pts: Option, media_unwrap: PtsUnwrap, + /// The source's mapping onto the broadcast clock, set by [`live`](Self::live). `None` + /// publishes the source's unwrapped PTS verbatim. + anchor: Option, } impl Import { @@ -176,9 +179,22 @@ impl Import { identity_recorded: false, last_pts: None, media_unwrap: PtsUnwrap::default(), + anchor: None, } } + /// Publish on the broadcast clock rather than the source's own PTS. + /// + /// For a live feed with its own zero: the first frame is live on arrival, every elementary + /// stream shares that one mapping, and a restart (a PTS rewind or a signalled time-base + /// discontinuity) continues forward after the real idle gap. Without this, the unwrapped PTS + /// is published verbatim, which suits a source already on the clock the catalog advertises + /// ([`Config::with_clock`](crate::catalog::Config::with_clock)). + pub fn live(mut self) -> Self { + self.anchor = Some(crate::clock::Anchor::new(self.catalog.clock())); + self + } + /// Select the container this importer wraps decoded media renditions in. /// /// [`Legacy`](hang::catalog::Container::Legacy) unless selected. It applies to every rendition @@ -646,7 +662,11 @@ impl Import { // frame must be timestamped with this frame's PTS ("now"), not the // previous one's. if pes.header.pts.is_some() { - let pts = unwrap_pts(&mut self.media_unwrap, pes.header.pts.map(|t| t.as_u64()))?; + let pts = unwrap_pts( + &mut self.media_unwrap, + pes.header.pts.map(|t| t.as_u64()), + self.anchor.as_mut(), + )?; let video = match self.streams.get(&pid) { Some(Stream::H264 { reanchor, import, .. }) => { Some((reanchor, import.floor(false), import.floor(true))) @@ -717,7 +737,7 @@ impl Import { let Some(stream) = self.streams.get_mut(&pid) else { return Ok(()); }; - self.published |= stream.write(pending, batched)?; + self.published |= stream.write(pending, batched, self.anchor.as_mut())?; // Record the decoded media track's PID + PMT descriptors (language, ...) once // its lazily created track exists, so export can preserve them. @@ -1219,7 +1239,7 @@ impl VerbatimStream { /// Publish one reassembled PES payload verbatim, in its own group, stamped with /// its PTS (or the live edge when the PES carried none). - fn write(&mut self, pending: Pending) -> anyhow::Result { + fn write(&mut self, pending: Pending, anchor: Option<&mut crate::clock::Anchor>) -> anyhow::Result { // Record the original PES stream_id once, from the first PES, so export // re-emits the stream under its real id (e.g. 0xBD for teletext/DVB AC-3). if !self.stream_id_recorded { @@ -1233,7 +1253,7 @@ impl VerbatimStream { } let edge = self.track.live_edge(); - let pts = match unwrap_pts(&mut self.unwrap, pending.pts)? { + let pts = match unwrap_pts(&mut self.unwrap, pending.pts, anchor)? { Some(pts) => self.reanchor.apply(pts, edge)?, // No clock to shift, so land on the edge and leave the shift alone. None => edge.unwrap_or(Timestamp::ZERO), @@ -1547,7 +1567,12 @@ enum Stream { } impl Stream { - fn write(&mut self, pending: Pending, batched: bool) -> anyhow::Result { + fn write( + &mut self, + pending: Pending, + batched: bool, + anchor: Option<&mut crate::clock::Anchor>, + ) -> anyhow::Result { match self { Stream::H264 { split, @@ -1556,7 +1581,7 @@ impl Stream { reanchor, } => { let reorder = reorder_delay(pending.pts, pending.dts); - let pts = unwrap_pts(unwrap, pending.pts)?; + let pts = unwrap_pts(unwrap, pending.pts, anchor)?; // Each PES is one access unit, so flush to emit it immediately. let mut frames = split.decode(&pending.data, pts)?; frames.extend(split.flush(pts)?); @@ -1578,7 +1603,7 @@ impl Stream { reanchor, } => { let reorder = reorder_delay(pending.pts, pending.dts); - let pts = unwrap_pts(unwrap, pending.pts)?; + let pts = unwrap_pts(unwrap, pending.pts, anchor)?; // Each PES is one access unit, so flush to emit it immediately. let mut frames = split.decode(&pending.data, pts)?; frames.extend(split.flush(pts)?); @@ -1592,10 +1617,10 @@ impl Stream { } Ok(published) } - Stream::Aac(stream) => stream.write(pending, batched), - Stream::Opus(stream) => stream.write(pending), - Stream::Legacy(stream) => stream.write(pending), - Stream::Verbatim(stream) => stream.write(pending), + Stream::Aac(stream) => stream.write(pending, batched, anchor), + Stream::Opus(stream) => stream.write(pending, anchor), + Stream::Legacy(stream) => stream.write(pending, anchor), + Stream::Verbatim(stream) => stream.write(pending, anchor), Stream::Clock | Stream::Ignored => Ok(false), } } @@ -2005,8 +2030,13 @@ struct AacStream { } impl AacStream { - fn write(&mut self, pending: Pending, batched: bool) -> anyhow::Result { - let pes_base = unwrap_pts(&mut self.unwrap, pending.pts)?; + fn write( + &mut self, + pending: Pending, + batched: bool, + anchor: Option<&mut crate::clock::Anchor>, + ) -> anyhow::Result { + let pes_base = unwrap_pts(&mut self.unwrap, pending.pts, anchor)?; // Prepend the partial frame left by the previous PES, if any. let carried = self.tail.len(); @@ -2232,7 +2262,8 @@ impl AacStream { // of it. if !self.tail.is_empty() && self.import.is_some() { self.resync.drain(); - self.write(Pending::empty(), true)?; + // No PTS to translate, so no mapping needed. + self.write(Pending::empty(), true, None)?; } // A partial frame at end of stream isn't emissible; drop it, but leave a trace for // diagnosing truncated captures. @@ -2268,8 +2299,8 @@ struct OpusStream { } impl OpusStream { - fn write(&mut self, pending: Pending) -> anyhow::Result { - let base = unwrap_pts(&mut self.unwrap, pending.pts)?; + fn write(&mut self, pending: Pending, anchor: Option<&mut crate::clock::Anchor>) -> anyhow::Result { + let base = unwrap_pts(&mut self.unwrap, pending.pts, anchor)?; let edge = self.import.live_edge(); let base = base.map(|base| self.reanchor.apply(base, edge)).transpose()?; @@ -2424,9 +2455,9 @@ struct LegacyStream { } impl LegacyStream { - fn write(&mut self, pending: Pending) -> anyhow::Result { + fn write(&mut self, pending: Pending, anchor: Option<&mut crate::clock::Anchor>) -> anyhow::Result { let mut published = false; - let pes_base = unwrap_pts(&mut self.unwrap, pending.pts)?; + let pes_base = unwrap_pts(&mut self.unwrap, pending.pts, anchor)?; // Prepend the partial frame left by the previous PES, if any. let carried = self.tail.len(); @@ -2654,7 +2685,8 @@ impl LegacyStream { // of it. if !self.tail.is_empty() && self.import.is_some() { self.resync.drain(); - self.write(Pending::empty())?; + // No PTS to translate, so no mapping needed. + self.write(Pending::empty(), None)?; } // A partial frame at end of stream isn't emissible verbatim; drop it, but // leave a trace for diagnosing truncated captures. @@ -2722,12 +2754,19 @@ fn advance_pts(pts: Option, samples: u64, sample_rate: u32) -> anyhow /// Convert a raw 90 kHz PTS to a microsecond [`Timestamp`], unwrapping the /// 33-bit field. Returns `None` when the PES carried no PTS. -fn unwrap_pts(unwrap: &mut PtsUnwrap, pts: Option) -> anyhow::Result> { +fn unwrap_pts( + unwrap: &mut PtsUnwrap, + pts: Option, + anchor: Option<&mut crate::clock::Anchor>, +) -> anyhow::Result> { let Some(raw) = pts else { return Ok(None); }; - let extended = unwrap.unwrap(raw); - Ok(Some(Timestamp::from_scale(extended, 90_000)?)) + let extended = Timestamp::from_scale(unwrap.unwrap(raw), 90_000)?; + Ok(Some(match anchor { + Some(anchor) => anchor.translate(&mut unwrap.lane, extended)?, + None => extended, + })) } /// The reorder delay `PTS - DTS` for one PES, as a microsecond [`Timestamp`]. `None` unless @@ -2752,6 +2791,8 @@ fn reorder_delay(pts: Option, dts: Option) -> Option { struct PtsUnwrap { last: Option, offset: u64, + /// This stream's position on the source's broadcast-clock mapping, when publishing live. + lane: crate::clock::Lane, } impl PtsUnwrap { @@ -2775,6 +2816,7 @@ impl PtsUnwrap { /// whatever the source does with its clock next. fn discontinuity(&mut self) { self.last = None; + self.lane.restart(); } } diff --git a/rs/moq-mux/src/container/ts/import_test.rs b/rs/moq-mux/src/container/ts/import_test.rs index 348f2cf926..05ce77c687 100644 --- a/rs/moq-mux/src/container/ts/import_test.rs +++ b/rs/moq-mux/src/container/ts/import_test.rs @@ -549,3 +549,110 @@ fn import_handles_unaligned_chunks() { assert_eq!(snapshot.video.renditions.len(), 1); assert_eq!(snapshot.audio.renditions.len(), 1); } + +/// What a TS import published, plus the broadcast clock's reading around the first chunk. +struct LiveImport { + published: std::collections::BTreeMap>, + before: u128, + after: u128, +} + +/// Import `chunks` in order, idling `idle` between them, on a clock that began `ago` earlier. +/// `live` translates onto that clock; otherwise the unwrapped PTS publishes verbatim. +async fn live_import(chunks: &[&[u8]], live: bool, ago: std::time::Duration, idle: std::time::Duration) -> LiveImport { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let config = crate::catalog::Config::default().with_clock(crate::container::test_util::late_clock(ago)); + let catalog = crate::catalog::Producer::new(&mut broadcast, config).unwrap(); + let clock = catalog.clock(); + let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve()); + if live { + import = import.live(); + } + + let before = clock.now().as_micros(); + let mut after = before; + for (i, chunk) in chunks.iter().enumerate() { + if i > 0 { + std::thread::sleep(idle); + } + import.decode(chunk).unwrap(); + if i == 0 { + after = clock.now().as_micros(); + } + } + import.finish().unwrap(); + + LiveImport { + published: crate::container::test_util::published(&consumer, &catalog.snapshot()).await, + before, + after, + } +} + +/// A TS feed arriving 30s after the broadcast began publishes on the broadcast clock rather than +/// its own PTS: the stream is live on arrival, and H.264 and AAC keep the one offset their PES +/// headers gave them. +#[tokio::test] +async fn live_import_anchors_a_late_first_frame() { + let data: &[u8] = include_bytes!("test_data/bbb_cbr.ts"); + let ago = std::time::Duration::from_secs(30); + let verbatim = live_import(&[data], false, ago, std::time::Duration::ZERO).await; + let live = live_import(&[data], true, ago, std::time::Duration::ZERO).await; + + let offset = crate::container::test_util::common_offset(&verbatim.published, &live.published); + // The earliest PES anchors at its arrival, which frames muxed ahead of it may precede. + let earliest = verbatim.published.values().map(|t| t[0]).min().unwrap() as i128 + offset; + let skew = std::time::Duration::from_secs(2).as_micros() as i128; + assert!( + live.before as i128 - skew <= earliest && earliest <= live.after as i128, + "the stream is live on arrival: {earliest} not near {}..={}", + live.before, + live.after + ); +} + +/// The same feed played twice, as when an encoder restarts its PTS, continues forward after the +/// real idle gap instead of rewinding. +#[tokio::test] +async fn live_import_restarts_forward_after_idle() { + let data: &[u8] = include_bytes!("test_data/bbb_cbr.ts"); + let once = live_import(&[data], false, std::time::Duration::ZERO, std::time::Duration::ZERO).await; + + // How much source time one pass covers, across every stream. + let starts = once.published.values().map(|t| *t.iter().min().unwrap()); + let ends = once.published.values().map(|t| *t.iter().max().unwrap()); + let span = (ends.max().unwrap() - starts.min().unwrap()) as i128; + + for idle in [std::time::Duration::ZERO, std::time::Duration::from_millis(300)] { + let live = live_import(&[data, data], true, std::time::Duration::from_secs(30), idle).await; + + // Each pass lands on one mapping for every stream: the first frames on the first, the last + // frames on the second. + let offset = |pick: fn(&Vec) -> u128| { + let deltas: Vec = live + .published + .iter() + .map(|(name, t)| pick(t) as i128 - pick(&once.published[name]) as i128) + .collect(); + assert!( + deltas.iter().all(|d| (d - deltas[0]).abs() <= 1_000), + "every stream shares one mapping: {deltas:?}" + ); + deltas[0] + }; + let first = offset(|t| t[0]); + let second = offset(|t| *t.last().unwrap()); + + // The second pass continues after the first plus the real idle gap, not on top of it. + let shift = second - first; + assert!( + shift >= span + idle.as_micros() as i128, + "the restart resumes after the first pass and the idle gap: {shift} < {span} + {idle:?}" + ); + assert!( + shift < span + (idle + std::time::Duration::from_secs(5)).as_micros() as i128, + "the restart is not pushed further: {shift}" + ); + } +}