fix(moq-mux): signal a new time base when the DTS reserve steps the PCR back

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

@@ -150,6 +150,9 @@ pub struct Export<E: catalog::Catalog = ()> {
 	/// Generation of the last returned frame, updated only at the output boundary.
 	emitted_epoch: u64,
 	pcr_discontinuity: bool,
+	/// Back-off of the last PCR emitted ([`Self::pcr_at`]), so a reserve that grows past it
+	/// flags the next one as a new time base.
+	pcr_reserve: Option<u64>,
 	/// TS packets muxed into the span that is still open.
 	pending: Vec<u8>,
 	/// Offsets into [`pending`](Self::pending) where a keyframe's packets begin, so
@@ -682,6 +685,7 @@ impl<E: catalog::Catalog> Export<E> {
 			epoch: 0,
 			emitted_epoch: 0,
 			pcr_discontinuity: false,
+			pcr_reserve: None,
 			pending: Vec::new(),
 			keyframes: Vec::new(),
 			queue: VecDeque::new(),
@@ -1900,12 +1904,13 @@ impl<E: catalog::Catalog> Export<E> {
 	/// The value backs off by the largest reserve of any track, not just the PCR
 	/// track's: every rendition's PES must decode at or after the clock, and each
 	/// video track backs its DTS off by its own [`Reserve`]. A reserve that grows steps
-	/// the clock back with it; the DTS already sent stay ahead of the clock values sent
-	/// with them, and [`author_dts`] keeps every later DTS above those. Back off through the
-	/// 33-bit wrap rather than saturating: a timeline that starts inside the reserve
-	/// would otherwise clamp its first slots to zero and break the uniform step. The
-	/// wire field is a circular clock, so the masked wrapped value is the correct
-	/// mod-2^33 back-off.
+	/// the clock back with it, which only a new time base allows (ISO 13818-1 2.4.3.4),
+	/// so that PCR sets `discontinuity_indicator`. The DTS already sent stay ahead of
+	/// the clock values sent with them, and [`author_dts`] keeps every later DTS above
+	/// those. Back off through the 33-bit wrap rather than saturating: a timeline that
+	/// starts inside the reserve would otherwise clamp its first slots to zero and break
+	/// the uniform step. The wire field is a circular clock, so the masked wrapped value
+	/// is the correct mod-2^33 back-off.
 	fn pcr_at(&mut self, index: u128, before: Option<u8>) -> anyhow::Result<Vec<u8>> {
 		let pcr_pid = self.psi.as_ref().context("PSI not built")?.pcr_pid;
 		let reserve = self
@@ -1915,6 +1920,9 @@ impl<E: catalog::Catalog> Export<E> {
 			.max()
 			.unwrap_or(DEFAULT_DTS_RESERVE);
 		let ticks = slot_ticks(index, PCR_INTERVAL).wrapping_sub(reserve);
+		if self.pcr_reserve.replace(reserve).is_some_and(|last| reserve > last) {
+			self.pcr_discontinuity = true;
+		}
 		// Nothing has gone out on this PID yet, so there is no counter to repeat and
 		// any value starts a valid run; take the one before the next to be used.
 		let cc = match before {

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

@@ -2560,7 +2560,13 @@ async fn reordered_video_keeps_the_table_cadence() {
 	assert_eq!(export.discontinuity(), 0, "a reorder is not a rewind");
 	assert_eq!(count_pid(&out, 0x0000), 10, "PAT once per 500ms slot, not per reorder");
 	assert_eq!(count_pid(&out, 0x0011), 3, "SDT once per 2s slot, not per reorder");
-	assert_eq!(count_discontinuity(&out), 0, "a reorder is not a discontinuity");
+	// The SPS declares no reorder depth, so the first B-frame grows the reserve once, after
+	// the first PCR: that one step back is a new time base, the reorders after it are not.
+	assert_eq!(
+		count_discontinuity(&out),
+		1,
+		"only the reserve's growth restarts the clock"
+	);
 }
 
 /// A program with more than one track marks the break once, not once per track. A
@@ -5170,14 +5176,24 @@ fn assert_decodes_before_presenting(timing: &[(u64, Option<u64>)]) {
 	}
 }
 
-/// In transport order, every PES unit decodes at or after the last PCR preceding it.
+/// In transport order, every PES unit decodes at or after the last PCR preceding it, and the
+/// clock only steps back where a PCR signals a new time base.
 fn assert_decodes_after_the_clock(frames: &[Frame]) {
 	let bytes: Vec<u8> = frames.iter().flat_map(|f| f.payload.iter().copied()).collect();
 	let mut reader = TsPacketReader::new(Cursor::new(bytes));
 	let mut last_pcr = None;
 	while let Some(packet) = reader.read_ts_packet().unwrap() {
-		if let Some(pcr) = packet.adaptation_field.as_ref().and_then(|af| af.pcr) {
-			last_pcr = Some(pcr.as_u64() / 300);
+		if let Some(af) = packet.adaptation_field.as_ref()
+			&& let Some(pcr) = af.pcr
+		{
+			let pcr = pcr.as_u64() / 300;
+			if let Some(last) = last_pcr {
+				assert!(
+					pcr > last || af.discontinuity_indicator,
+					"the clock steps back from {last} to {pcr} without a discontinuity"
+				);
+			}
+			last_pcr = Some(pcr);
 		}
 		if let Some(TsPayload::PesStart(pes)) = packet.payload
 			&& let Some(pcr) = last_pcr

test/ts/README.md

@@ -78,7 +78,7 @@ Severities: **hard** checks fail the run by default; **shape** checks report as
 | `psi-crc` | hard | no section dropped for a bad CRC |
 | `continuity` | hard | no continuity-counter discontinuities |
 | `pcr-presence` | hard | a PCR PID is declared and carries PCR |
-| `pcr-monotonic` | hard | PCR strictly increases (one 33-bit wrap tolerated) |
+| `pcr-monotonic` | hard | PCR strictly increases (one 33-bit wrap tolerated), except into a PCR that signals `discontinuity_indicator` |
 | `duration-fidelity` | hard | exported PCR span tracks the source's duration (round-trip only) |
 | `pcr-repetition` | shape | consecutive PCRs within the limit (default 40 ms) |
 | `pcr-jitter` | shape | per-interval PCR jitter vs the nominal bitrate (pcrverify model) |

test/ts/compliance.py

@@ -112,6 +112,8 @@ class Scan:
     pid_packets: dict[int, list[tuple[int, int]]]
     # (ts_index, pcr_27mhz) samples for every PID that carries PCR.
     pcr_by_pid: dict[int, list[tuple[int, int]]]
+    # ts_index of every PCR that starts a new time base (discontinuity_indicator).
+    pcr_new_base: set[int]
 
 
 def scan_packets(ts_path: str, packet_size: int) -> Scan:
@@ -125,6 +127,10 @@ def scan_packets(ts_path: str, packet_size: int) -> Scan:
 
     pid_packets: dict[int, list[tuple[int, int]]] = {}
     pcr_by_pid: dict[int, list[tuple[int, int]]] = {}
+    pcr_new_base: set[int] = set()
+    # PIDs whose discontinuity_indicator rode a packet without a PCR, so the new time
+    # base starts at their next PCR (ISO 13818-1 2.4.3.5).
+    pending_base: set[int] = set()
     index = 0
     offset = 0
     n = len(data)
@@ -147,6 +153,8 @@ def scan_packets(ts_path: str, packet_size: int) -> Scan:
             af_len = data[offset + 4]
             if af_len > 0:
                 flags = data[offset + 5]
+                if flags & 0x80:
+                    pending_base.add(pid)
                 if (flags & 0x10) and af_len >= 7:  # PCR present
                     base = (
                         (data[offset + 6] << 25)
@@ -157,6 +165,9 @@ def scan_packets(ts_path: str, packet_size: int) -> Scan:
                     )
                     ext = ((data[offset + 10] & 0x01) << 8) | data[offset + 11]
                     pcr_by_pid.setdefault(pid, []).append((index, base * 300 + ext))
+                    if pid in pending_base:
+                        pending_base.discard(pid)
+                        pcr_new_base.add(index)
         if afc in (1, 3):
             # Payload = 184 minus the adaptation field (its length byte + body).
             consumed = (1 + af_len) if afc == 3 else 0
@@ -173,6 +184,7 @@ def scan_packets(ts_path: str, packet_size: int) -> Scan:
         total_packets=index,
         pid_packets=pid_packets,
         pcr_by_pid=pcr_by_pid,
+        pcr_new_base=pcr_new_base,
     )
 
 
@@ -331,18 +343,22 @@ def check_pcr_presence(analysis: dict, clock_by_pid: dict[int, PcrClock]) -> Che
 
 
 def check_pcr_monotonic(scan: Scan) -> Check:
-    """PCR must strictly increase per PID (a single 33-bit wrap is tolerated)."""
+    """PCR must strictly increase per PID (a single 33-bit wrap is tolerated).
+
+    Except into a PCR that signals a new time base (ISO 13818-1 2.4.3.4), which may
+    take any value; those are counted so a stream that leans on them still shows it.
+    """
     breaks = 0
     for _pid, samples in scan.pcr_by_pid.items():
         prev = None
-        for _i, pcr in samples:
-            if prev is not None:
+        for i, pcr in samples:
+            if prev is not None and i not in scan.pcr_new_base:
                 delta = pcr - prev
                 # A legitimate wrap shows as a large negative jump; anything else is a fault.
                 if delta <= 0 and not delta < -PCR_WRAP / 2:
                     breaks += 1
             prev = pcr
-    metrics = {"pcr_backwards": breaks}
+    metrics = {"pcr_backwards": breaks, "signalled_discontinuities": len(scan.pcr_new_base)}
     if breaks == 0:
         return Check("pcr-monotonic", Severity.HARD, Status.PASS, "PCR strictly increasing", metrics)
     return Check("pcr-monotonic", Severity.HARD, Status.FAIL, f"{breaks} backwards PCR step(s)", metrics)