diff --git a/rs/moq-mux/src/container/ts/passthrough.rs b/rs/moq-mux/src/container/ts/passthrough.rs index 0e2ff52ee8..fada71317b 100644 --- a/rs/moq-mux/src/container/ts/passthrough.rs +++ b/rs/moq-mux/src/container/ts/passthrough.rs @@ -37,14 +37,18 @@ //! its base, ten bytes into its packet, so even an object that starts at a PCR packet sits ten //! bytes of that rate before its own PCR. //! +//! Every timestamp is then shifted onto the catalog's [`Clock`](crate::Clock) by one offset, fixed +//! through the reservation's [`Timebase`](crate::catalog::Timebase) by the first object: zero when +//! that object places the clock, and otherwise whatever lands it at now on a clock already in use. +//! //! PCRs are unwrapped into a 64-bit count of 27 MHz ticks that never goes back. A step forward of //! at most half the 2^33 * 300 tick period adds to the count, across the wrap included. A PCR with //! `discontinuity_indicator` whose step is forward by that measure adds the step too, after the //! bytes before it are timed at the rate before it, and starts a group behind a break marker. Any //! step back ends the import: [`PcrRewindError`] without the flag, [`PcrRestartError`] with it, //! since a signalled restart is new content and needs a new broadcast. The timeline therefore -//! depends only on the bytes since the importer's first PCR, and two importers of one feed started -//! at the same packet publish identical objects. +//! depends only on the bytes since the importer's first PCR and its shift, and two importers of one +//! feed started at the same packet each placing its own catalog's clock publish identical objects. use std::collections::{BTreeMap, HashMap, VecDeque}; @@ -167,6 +171,8 @@ pub struct Passthrough { /// Held until the first group is written, so the first catalog already names the track with /// its final `randomAccess`. reservation: Option>, + /// Shifts every object's PCR time onto the catalog clock, anchored by the first object. + shift: crate::catalog::Timebase, media: crate::container::Producer, /// The section as last written to the catalog, `None` before the first group. section: Option>, @@ -293,6 +299,7 @@ impl Passthrough { )?; Ok(Self { catalog, + shift: reserved.timebase(), reservation: Some(reserved), media, section: None, @@ -602,6 +609,8 @@ impl Passthrough { let ticks = ticks.max(self.written); self.written = ticks; let timestamp = moq_net::Timestamp::from_micros(ticks / (PCR_HZ / 1_000_000))?; + // The first object, which starts a group, anchors the shift and is live on arrival. + let timestamp = self.shift.shift(timestamp)?; let size = object.payload.len() as u64; let start = object.start.or_else(|| { self.group @@ -617,10 +626,6 @@ impl Passthrough { if start != Start::RandomAccess || !self.single { self.all_random_access = false; } - if self.reservation.is_some() { - // The first object of the import is live on arrival. - self.catalog.anchor(timestamp)?; - } self.group = Some(Group { ticks, bytes: 0, diff --git a/rs/moq-mux/src/container/ts/passthrough_test.rs b/rs/moq-mux/src/container/ts/passthrough_test.rs index 7b2616d3a1..ad2260e65d 100644 --- a/rs/moq-mux/src/container/ts/passthrough_test.rs +++ b/rs/moq-mux/src/container/ts/passthrough_test.rs @@ -63,8 +63,19 @@ impl Imported { /// Import `data` in chunks of `chunk` bytes and read back every object. async fn import(data: &[u8], chunk: usize, pcr_pid: Option) -> anyhow::Result { let mut broadcast = moq_net::broadcast::Info::new().produce(); - let consumer = broadcast.consume(); let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default())?; + import_onto(broadcast, catalog, data, chunk, pcr_pid).await +} + +/// [`import`] onto an existing `catalog` of `broadcast`. +async fn import_onto( + broadcast: moq_net::broadcast::Producer, + catalog: crate::catalog::Producer, + data: &[u8], + chunk: usize, + pcr_pid: Option, +) -> anyhow::Result { + let consumer = broadcast.consume(); let mut passthrough = Passthrough::new(broadcast.clone(), catalog.reserve())?; if let Some(pid) = pcr_pid { passthrough = passthrough.with_pcr_pid(pid); @@ -723,6 +734,50 @@ async fn a_flagged_jump_forward_starts_a_group_behind_a_break() { } } +/// A capture that took the catalog's clock first keeps it: the import shifts every object by one +/// offset that lands the first at now, so the timeline still never steps back and a flagged jump +/// still adds its step. +#[tokio::test(start_paused = true)] +async fn an_import_shifts_onto_a_clock_already_in_use() { + let mut mux = Mux::default(); + program(&mut mux, 0x100); + paced(&mut mux, 0x100, 3_600_000 * MS, 50, 10, 4); + mux.packet(0x100, Flags::pcr(3_620_000 * MS).discontinuity()); + mux.fill(0x101, 4); + paced(&mut mux, 0x100, 3_620_040 * MS, 50, 10, 4); + let data = mux.bytes(); + let verbatim = import(&data, data.len(), None).await.unwrap(); + + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let clock = catalog.clock(); + let fixed = catalog.snapshot().clock; + tokio::time::advance(Duration::from_secs(5)).await; + let before = clock.now().as_micros() as u64; + let shifted = import_onto(broadcast, catalog.clone(), &data, data.len(), None) + .await + .unwrap(); + let after = clock.now().as_micros() as u64; + + assert_eq!(catalog.snapshot().clock, fixed, "the import never moves the clock"); + let first = shifted.objects[0].micros; + assert!( + (before..=after).contains(&first), + "lands at now: {before} {first} {after}" + ); + let offset = first as i64 - verbatim.objects[0].micros as i64; + assert!(offset < 0, "an hour in shifts back onto the capture's five seconds"); + assert_eq!(shifted.objects.len(), verbatim.objects.len()); + for (shifted, verbatim) in shifted.objects.iter().zip(&verbatim.objects) { + assert_eq!((shifted.group, &shifted.payload), (verbatim.group, &verbatim.payload)); + assert_eq!( + shifted.micros as i64 - verbatim.micros as i64, + offset, + "one offset for every object" + ); + } +} + /// A flagged step back is a restart: new content, which a timeline cannot step back into. #[tokio::test(start_paused = true)] async fn a_flagged_step_back_is_a_restart() {