refactor(moq-mux): TS stats count a shared PID once

quest/m1/ts-reliability/README.md

@@ -100,7 +100,6 @@ catalog linger (#5305).
 ## Required
 
 - [A small step back keeps the TS export's video](/quest/m1/ts-reliability/step-back.md) - the export re-acquires on a sub-group step back from any source instead of dropping its video
-- [Shared PID stats](/quest/m1/ts-reliability/shared-pid-stats.md) - `ts::stats` counts a shared PID once in the shared parser, so its timing counts survive program changes
 
 ## Related
 

rs/moq-mux/src/container/ts/health.rs

@@ -47,6 +47,18 @@ pub(super) struct Errors {
 	pids: BTreeMap<u16, PidErrors>,
 }
 
+/// The checks a program times on a PID by the way it reads it, which programs reading the PID
+/// the same way count alike.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub(super) enum Checks {
+	/// The PTS cadence of an elementary stream it carries, and the PAT's interval on PID 0.
+	Stream,
+	/// The PCR checks on the PID it takes its clock from.
+	Clock,
+	/// The PMT interval on the PID it reads its PMT from.
+	Table,
+}
+
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 struct PidErrors {
 	continuity_count_error: u64,
@@ -83,6 +95,31 @@ impl Errors {
 		}
 	}
 
+	/// The `checks` counted on `pid` alone.
+	pub(super) fn only(&self, pid: u16, checks: Checks) -> Self {
+		let all = self.pids.get(&pid).copied().unwrap_or_default();
+		let mut errors = PidErrors::default();
+		let mut pat_error = 0;
+		match checks {
+			Checks::Stream => {
+				errors.pts_error = all.pts_error;
+				if pid == Pid::PAT {
+					pat_error = self.pat_error;
+				}
+			}
+			Checks::Clock => {
+				errors.pcr_repetition_error = all.pcr_repetition_error;
+				errors.pcr_discontinuity_indicator_error = all.pcr_discontinuity_indicator_error;
+			}
+			Checks::Table => errors.pmt_error = all.pmt_error,
+		}
+		Self {
+			pat_error,
+			pids: BTreeMap::from([(pid, errors)]),
+			..Default::default()
+		}
+	}
+
 	/// Write the stream-wide totals into `stats`, and each elementary stream's own counts into
 	/// its row.
 	pub(super) fn report(&self, stats: &mut Snapshot) {
@@ -173,7 +210,7 @@ impl Multiplex {
 }
 
 /// The TR 101 290 checks timed on one program's clock: table, PCR and PTS intervals.
-#[derive(Default)]
+#[derive(Clone, Default)]
 pub(super) struct Health {
 	errors: Errors,
 	pcr_pid: Option<u16>,
@@ -274,7 +311,7 @@ impl Health {
 	pub(super) fn pmt_pids(&mut self, pids: &[u16], now: Option<u64>) {
 		self.pmt.retain(|pid, _| pids.contains(pid));
 		for &pid in pids {
-			self.pmt.entry(pid).or_insert(Timer { last: now });
+			self.pmt.entry(pid).or_insert(Timer::at(now));
 		}
 	}
 
@@ -301,6 +338,32 @@ impl Health {
 	pub(super) fn restart(&mut self, pid: u16) {
 		self.pts.remove(&pid);
 	}
+
+	/// Carry what this checker has timed on `pid` over to `to`, which reads the same packets and
+	/// counts `pid` from here on, where `to` has not timed it yet: the PAT's interval on PID 0,
+	/// the PMT's on a PMT PID `to` reads, and the PTS cadence when `to` grades one there. So an
+	/// interval running across the handover counts once, from where it began, even from before
+	/// `to`'s clock started. `now` and `then` are this checker's clock and `to`'s; nothing
+	/// carries over before both run.
+	pub(super) fn hand_over(&self, pid: u16, to: &mut Self, pts: bool, now: Option<u64>, then: Option<u64>) {
+		let (Some(now), Some(then)) = (now, then) else {
+			return;
+		};
+		let carry = |timer: &Timer, theirs: &mut Timer| {
+			if theirs.last.is_none() {
+				theirs.last = timer.last.map(|last| last + ticks(then) - ticks(now));
+			}
+		};
+		if pid == Pid::PAT {
+			carry(&self.pat, &mut to.pat);
+		}
+		if let (Some(timer), Some(theirs)) = (self.pmt.get(&pid), to.pmt.get_mut(&pid)) {
+			carry(timer, theirs);
+		}
+		if pts && let Some(timer) = self.pts.get(&pid) {
+			carry(timer, to.pts.entry(pid).or_default());
+		}
+	}
 }
 
 /// The step between two readings of a clock that wraps: forward by up to half the range,
@@ -338,33 +401,43 @@ fn section_start(pkt: &[u8; TsPacket::SIZE]) -> Option<u8> {
 }
 
 /// How long since a table or a PTS last occurred, on the program clock.
-#[derive(Default)]
+#[derive(Clone, Copy, Default)]
 struct Timer {
-	/// `None` until the clock starts.
-	last: Option<u64>,
+	/// `None` until the clock starts. Signed, since one [handed over](Health::hand_over) from
+	/// an earlier clock may predate this one's start.
+	last: Option<i64>,
 }
 
 impl Timer {
+	fn at(now: Option<u64>) -> Self {
+		Self { last: now.map(ticks) }
+	}
+
 	fn reset(&mut self, now: Option<u64>) {
 		if now.is_some() {
-			self.last = now;
+			*self = Self::at(now);
 		}
 	}
 
 	/// How many times `limit` ran out since the last occurrence, restarting the wait at each,
 	/// so an outage counts once per limit it lasts rather than once in all.
 	fn expired(&mut self, now: u64, limit: u64) -> u64 {
-		let last = self.last.get_or_insert(now);
-		let late = now.saturating_sub(*last);
+		let last = self.last.get_or_insert(ticks(now));
+		let late = u64::try_from(ticks(now) - *last).unwrap_or(0);
 		if late <= limit {
 			return 0;
 		}
 		let count = (late - 1) / limit;
-		*last += count * limit;
+		*last += ticks(count * limit);
 		count
 	}
 }
 
+/// A program clock reading as a signed tick count, which 27 MHz keeps in range for millennia.
+fn ticks(now: u64) -> i64 {
+	i64::try_from(now).unwrap_or(i64::MAX)
+}
+
 /// `TS_sync_loss` and `Sync_byte_error`, on a grid of its own.
 ///
 /// The importer's framer re-scans at the first byte that is not a sync byte, which is right

rs/moq-mux/src/container/ts/import.rs

@@ -348,8 +348,18 @@ impl<E: catalog::Catalog> Import<E> {
 		self
 	}
 
-	/// Whether this importer reassembles the PMT sections on `pid`, and counts their CRC errors.
-	pub(super) fn reads_pmt(&self, pid: u16) -> bool {
+	/// The PID this importer reads its PMT from.
+	pub(super) fn pmt_pid(&self) -> Option<u16> {
+		self.pmt.as_ref().map(|(pid, _)| pid.as_u16())
+	}
+
+	/// The PID this importer takes its program clock from.
+	pub(super) fn pcr_pid(&self) -> Option<u16> {
+		self.pcr_pid.map(|pid| pid.as_u16())
+	}
+
+	/// Whether this importer reassembles the PMT sections on `pid`.
+	fn reads_pmt(&self, pid: u16) -> bool {
 		self.pmt.as_ref().is_some_and(|(pmt, _)| pmt.as_u16() == pid)
 	}
 
@@ -358,11 +368,20 @@ impl<E: catalog::Catalog> Import<E> {
 		self.sections.contains_key(&pid) || Pid::new(pid).is_ok_and(|pid| self.streams.contains_key(&pid))
 	}
 
-	/// Whether this importer holds a PMT section begun on `pid`, which it counts if corrupt.
-	pub(super) fn assembling_pmt(&self, pid: u16) -> bool {
-		self.pmt
-			.as_ref()
-			.is_some_and(|(pmt, sections)| pmt.as_u16() == pid && sections.assembling())
+	/// The TR 101 290 intervals this importer times, and its clock, for another to take over.
+	pub(super) fn timing(&self) -> (&Health, Option<u64>) {
+		(&self.health, self.liveness.now())
+	}
+
+	/// Take over the TR 101 290 intervals `health` timed on `pid` at clock reading `now`, from a
+	/// program that read the same packets, to count the PID from here on. See
+	/// [`Health::hand_over`].
+	pub(super) fn take_over(&mut self, pid: u16, (health, now): (&Health, Option<u64>)) {
+		let cadenced = Pid::new(pid)
+			.ok()
+			.and_then(|pid| self.streams.get(&pid))
+			.is_some_and(Stream::cadenced);
+		health.hand_over(pid, &mut self.health, cadenced, now, self.liveness.now());
 	}
 
 	/// A reservation on this program's timebase, so every stream shifts by its offset.
@@ -1243,9 +1262,7 @@ impl<E: catalog::Catalog> Import<E> {
 
 		let is_video = stream.video();
 		let is_clock = matches!(stream, Stream::Clock);
-		// Verbatim streams with no cadence (SCTE-35, DVB subtitles) are left ungraded
-		// rather than watched against an invented one.
-		if pes.pts.is_some() && (is_video || !matches!(stream, Stream::PesPacket(_) | Stream::Ignored)) {
+		if pes.pts.is_some() && stream.cadenced() {
 			self.health.pts(pid.as_u16(), self.liveness.now());
 		}
 		// Resolve the wrap at PES start so every elementary stream and private section shares
@@ -1945,13 +1962,14 @@ impl<E: catalog::Catalog> Import<E> {
 		stats
 	}
 
-	/// The TR 101 290 counts this program timed, for merging programs that read the same
-	/// multiplex.
+	/// The TR 101 290 counts this program timed, for [`Programs`](super::Programs) to count each
+	/// PID once.
 	pub(super) fn errors(&self) -> &super::health::Errors {
 		self.health.errors()
 	}
 
-	/// PMT sections dropped for a bad CRC per PID, for merging programs that read the same PMT.
+	/// PMT sections dropped for a bad CRC per PID.
+	#[cfg(test)]
 	pub(super) fn crc_errors(&self) -> &BTreeMap<u16, u64> {
 		&self.crc_errors
 	}
@@ -3029,21 +3047,9 @@ pub(super) struct SectionReassembler {
 	span: Option<Span>,
 	/// Shared with the PES path: a broken chain drops the partial either way.
 	continuity: Continuity,
-	/// Whether the first section the latest packet completed began in an earlier packet.
-	resumed: bool,
 }
 
 impl SectionReassembler {
-	/// Whether the first section the latest packet completed began in an earlier packet.
-	pub(super) fn resumed(&self) -> bool {
-		self.resumed
-	}
-
-	/// Whether a partial section is left assembling.
-	pub(super) fn assembling(&self) -> bool {
-		!self.acc.is_empty()
-	}
-
 	/// Consume one 188-byte TS packet, appending every completed section to `out`.
 	pub(super) fn push(&mut self, pkt: &[u8], out: &mut Vec<Vec<u8>>) {
 		let mut completed = Vec::new();
@@ -3053,7 +3059,6 @@ impl SectionReassembler {
 
 	fn push_at(&mut self, pkt: &[u8], source: Option<(u64, &Placement)>, out: &mut Vec<(Option<Span>, Vec<u8>)>) {
 		let pkt: &[u8; 188] = pkt.try_into().expect("section packet must be 188 bytes");
-		self.resumed = false;
 		match self.continuity.observe(pkt) {
 			Continuation::Duplicate => return,
 			// A packet flagged corrupt takes its own payload down with the partial: this is
@@ -3094,9 +3099,7 @@ impl SectionReassembler {
 				// progress (stream just joined, or a reset dropped the partial) these
 				// bytes are an orphaned fragment, so skip straight to the pointer_field.
 				self.acc.extend_from_slice(&payload[1..1 + ptr]);
-				let completed = out.len();
 				self.drain(out, source, end - (payload.len() - 1 - ptr) as u64);
-				self.resumed = out.len() > completed;
 			}
 			// The pointer_field is a hard section boundary: drop any leftover partial
 			// and start the section it points to.
@@ -3111,9 +3114,7 @@ impl SectionReassembler {
 			// discontinuity, or corrupt pointer dropped the partial, rather than
 			// resuming on stray bytes that merely look like a section.
 			self.acc.extend_from_slice(payload);
-			let completed = out.len();
 			self.drain(out, source, end);
-			self.resumed = out.len() > completed;
 		}
 	}
 
@@ -3453,6 +3454,12 @@ impl<E: catalog::Catalog> Stream<E> {
 		}
 	}
 
+	/// Whether the stream's PTS keeps a cadence `PTS_error` grades. Verbatim streams with none
+	/// (SCTE-35, DVB subtitles) are left ungraded rather than watched against an invented one.
+	fn cadenced(&self) -> bool {
+		self.video() || !matches!(self, Stream::PesPacket(_) | Stream::Ignored)
+	}
+
 	/// This route's track and the frame sync it has lost (only the self-describing audio
 	/// codecs scan for it), or `None` for a PID that is dropped rather than carried.
 	fn stats(&self) -> Option<stats::Stream> {