From 748f04c98918ae1f20cdbf2ffaa725b9c8c359ad Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 29 Sep 2026 19:52:39 -0700 Subject: [PATCH] fix(moq-mux): signal a new time base when the DTS reserve steps the PCR back Since #4500 the video DTS reserve can grow after PCRs are out (a deeper reorder than declared, or a catalog `jitter` that arrives late), and the PCR backs off by it, so each growth stepped the PCR backwards unflagged. Flag the first PCR after a growth with discontinuity_indicator, as ISO 13818-1 2.4.3.4 requires, and let compliance.py's pcr-monotonic skip a signalled new time base the way pcr-timing.py already does. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/export.rs | 20 ++++++++++++------ rs/moq-mux/src/container/ts/export_test.rs | 24 ++++++++++++++++++---- test/ts/README.md | 2 +- test/ts/compliance.py | 24 ++++++++++++++++++---- 4 files changed, 55 insertions(+), 15 deletions(-) diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 20308810ea..5457c6eaa1 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -150,6 +150,9 @@ pub struct Export { /// 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, /// TS packets muxed into the span that is still open. pending: Vec, /// Offsets into [`pending`](Self::pending) where a keyframe's packets begin, so @@ -682,6 +685,7 @@ impl Export { 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 Export { /// 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) -> anyhow::Result> { let pcr_pid = self.psi.as_ref().context("PSI not built")?.pcr_pid; let reserve = self @@ -1915,6 +1920,9 @@ impl Export { .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 { diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 0cdcb50524..b6b4eef985 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/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)]) { } } -/// 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 = 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 diff --git a/test/ts/README.md b/test/ts/README.md index 28bcf8e348..4fb5950b9d 100644 --- a/test/ts/README.md +++ b/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) | diff --git a/test/ts/compliance.py b/test/ts/compliance.py index b02f5990c0..66aae98fe7 100755 --- a/test/ts/compliance.py +++ b/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)