fix(moq-mux): ts passthrough shifts onto the catalog clock

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<E: crate::catalog::hang::CatalogExt = ()> {
 	/// Held until the first group is written, so the first catalog already names the track with
 	/// its final `randomAccess`.
 	reservation: Option<crate::catalog::Reserved<E>>,
+	/// Shifts every object's PCR time onto the catalog clock, anchored by the first object.
+	shift: crate::catalog::Timebase<E>,
 	media: crate::container::Producer<crate::catalog::hang::Container>,
 	/// The section as last written to the catalog, `None` before the first group.
 	section: Option<Section<E>>,
@@ -293,6 +299,7 @@ impl<E: crate::catalog::hang::CatalogExt> Passthrough<E> {
 		)?;
 		Ok(Self {
 			catalog,
+			shift: reserved.timebase(),
 			reservation: Some(reserved),
 			media,
 			section: None,
@@ -602,6 +609,8 @@ impl<E: crate::catalog::hang::CatalogExt> Passthrough<E> {
 		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<E: crate::catalog::hang::CatalogExt> Passthrough<E> {
 			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,

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<u16>) -> anyhow::Result<Imported> {
 	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<u16>,
+) -> anyhow::Result<Imported> {
+	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() {