From c621acc7165afdf0f8537015dbf97a4e2369a811 Mon Sep 17 00:00:00 2001 From: t0ms Date: Sat, 3 Oct 2026 12:03:48 +0100 Subject: [PATCH 1/2] feat(mux): count the TR 101 290 errors of a TS feed at import `moq import ts` and `moq-srt` count the ETSI TR 101 290 V1.4.1 checks that grade a contribution feed at ingest, at its fixed limits: TS_sync_loss, Sync_byte_error, PAT_error_2, Continuity_count_error, PMT_error_2, Transport_error, PCR_repetition_error, PCR_discontinuity_indicator_error and PTS_error, beside the existing CRC_error. Counters only: nothing published changes. The checks live in a private `ts::health` module. The importer's continuity classifier moves there, so the counter and routing read one verdict per packet. Sync is graded on a grid of its own with ISO 13818-1 G.1 hysteresis, since the framer re-scans at the first bad byte. Table and PTS intervals run on the program clock the importer already keeps; PCR checks grade values. `Programs` merges its importers' counts by maximum, as it does CRC errors. Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 14 + doc/bin/srt.md | 5 +- quest/m1/ts-damaged-units.md | 6 +- quest/m1/ts-stats-module.md | 2 +- quest/m2/README.md | 1 - quest/m2/ts-health-stats.md | 1 - quest/m2/ts-import-health.md | 111 --- rs/moq-mux/src/container/ts/health.rs | 967 ++++++++++++++++++++++++ rs/moq-mux/src/container/ts/import.rs | 220 +++--- rs/moq-mux/src/container/ts/mod.rs | 1 + rs/moq-mux/src/container/ts/programs.rs | 6 +- rs/moq-mux/src/container/ts/stats.rs | 53 +- 12 files changed, 1142 insertions(+), 245 deletions(-) delete mode 100644 quest/m2/ts-import-health.md create mode 100644 rs/moq-mux/src/container/ts/health.rs diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 4ed16dcdd9..35ceeba4b3 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -83,6 +83,20 @@ downstream sees it. A sparse stream such as SCTE-35 goes quiet between cues, so the line reports rather than alarms; `Import::stats` carries the same counters for a caller that sets its own limit. +`import ts` also counts the ETSI TR 101 290 errors of the feed it receives, at +that standard's fixed limits: `TS_sync_loss`, `Sync_byte_error`, `PAT_error`, +`Continuity_count_error`, `PMT_error`, `Transport_error`, `CRC_error` (PAT and +PMT), `PCR_repetition_error`, `PCR_discontinuity_indicator_error` and +`PTS_error`. They are cumulative, logged with their totals in the sample after +any of them moves, and carried on `Import::stats` stream-wide and per PID. They +change nothing that is published. They grade the stream as it reached the +importer, not the wire a receiver sees downstream, and the PCR checks grade +consecutive PCR values rather than arrival times, so they speak for the encoder +and not for the network in front of the importer. Table, PCR and PTS intervals +run on the program clock, which starts at the first PCR after the PMT. +`PID_error` is covered, more strictly, by the access-unit counts above. +`PCR_accuracy_error` is not measured. + MPEG-TS import takes one program. A multi-program stream is refused before anything is published, naming its programs, rather than merged onto one clock; a PAT that adds a program mid-stream ends the import the same way. diff --git a/doc/bin/srt.md b/doc/bin/srt.md index ee4ff66711..eef692c56c 100644 --- a/doc/bin/srt.md +++ b/doc/bin/srt.md @@ -10,8 +10,9 @@ source (`--connect`); `moq export srt` serves SRT to players or pushes to a remote. The payload is MPEG-TS, so the same codecs as [`import ts`](/bin/cli) apply: H.264/H.265 video and AAC, MP2, AC-3, or E-AC-3 audio. Ingest logs the same per-stream lines as `import ts`, under an `srt{path=...}` span: an -elementary stream that stopped delivering access units, and audio frame sync -lost. +elementary stream that stopped delivering access units, audio frame sync lost, +and the TR 101 290 counters when one moves. They grade the TS as SRT delivered +it, after retransmission. ```bash # Accept a contribution feed and publish it diff --git a/quest/m1/ts-damaged-units.md b/quest/m1/ts-damaged-units.md index e7124059e6..8db92f2eb4 100644 --- a/quest/m1/ts-damaged-units.md +++ b/quest/m1/ts-damaged-units.md @@ -30,7 +30,7 @@ Decided (maintainer, 2026-09-30): - Count each drop in a new cumulative per-PID `damaged` counter on the stream's stats row, beside `resyncs` and `discarded`, reported by `ts::stats::Log`. It names what happened; no TR 101 290 check matches a codec error, and - [TS import health](/quest/m2/ts-import-health.md) keeps the ETSI names. + the TR 101 290 counters keep the ETSI names. The rows are `#[non_exhaustive]`, so the field is additive; follow [TS stats module](/quest/m1/ts-stats-module.md)'s names if it has landed. - Errors that are not confined to one unit (the producer refusing a rewind, @@ -50,7 +50,3 @@ Public API: additive, one stats field. Wire: none. ## Closes - [#4581](https://github.com/moq-dev/moq/issues/4581) - one malformed packet ends the TS import - -## Related - -- [TS import health](/quest/m2/ts-import-health.md) - the TR 101 290 counters for the same feed diff --git a/quest/m1/ts-stats-module.md b/quest/m1/ts-stats-module.md index 96cbdb6563..8cf07e1226 100644 --- a/quest/m1/ts-stats-module.md +++ b/quest/m1/ts-stats-module.md @@ -31,7 +31,7 @@ Decided while planning the follow-ups of - Update moq-cli's `publish.rs`, the only consumer that names these types. moq-srt only uses `stats::Log`. - Main still adds fields under the old names: #4584 added `crc_error`, and - [TS import health](/quest/m2/ts-import-health.md) adds more. `Export::stats` + the TR 101 290 counters added more. `Export::stats` (#4577) also returns the old `ts::Stats`. The rename carries those at merge time. Update the type names in the quests still open when this lands, including [media stats schema](/quest/m1/stats/schema.md) and [Rust reporters](/quest/m1/stats/rust.md). diff --git a/quest/m2/README.md b/quest/m2/README.md index c27f801da8..68b2cb3851 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -100,7 +100,6 @@ waiting on an upstream release waits in [m4](/quest/m4/README.md). - [Keep-alive by deadline](/quest/m2/quic-keep-alive.md) - a PING only when the idle deadline nears, no fixed timer - [Socket close](/quest/m2/noq-socket-close.md) - `moq-quic` releases an endpoint's socket on close, so moq-tokio drops its wrapper - [GOP overhead](/quest/m2/gop-overhead.md) - price the I-frames a short GOP pays for, deciding whether a long GOP plus a keyframe request is worth designing -- [TS import health](/quest/m2/ts-import-health.md) - `moq import ts` counts the TR 101 290 errors of the feed it receives, PCR and PTS graded on its own values - [TS health stats](/quest/m2/ts-health-stats.md) - the TS counters ride the stats plumbing beside the media counters - [TS stopped log: audio and video only](/quest/m2/ts-stopped-av-only.md) - sparse data PIDs like SCTE-35 stop logging "stopped delivering" every quiet second - [Teleoperation](/quest/m2/teleop/README.md) - MoQ carries robot video down and control up on one session as a library capability diff --git a/quest/m2/ts-health-stats.md b/quest/m2/ts-health-stats.md index 17a521065c..1f96be9c7a 100644 --- a/quest/m2/ts-health-stats.md +++ b/quest/m2/ts-health-stats.md @@ -42,4 +42,3 @@ Decided while planning [#1838](https://github.com/moq-dev/moq/issues/1838): - [TS stats module](/quest/m1/ts-stats-module.md) - the owned `track` that lets the rows deserialize - [Media stats schema](/quest/m1/stats/schema.md) - the snapshot the counters flatten into - [Rust reporters](/quest/m1/stats/rust.md) - `moq import --stats` and the stats interval -- [TS import health](/quest/m2/ts-import-health.md) - the ingest counters diff --git a/quest/m2/ts-import-health.md b/quest/m2/ts-import-health.md deleted file mode 100644 index 4ef2a6f64b..0000000000 --- a/quest/m2/ts-import-health.md +++ /dev/null @@ -1,111 +0,0 @@ -# [M] moq import ts counts the TR 101 290 errors of the feed it receives - -## Goal - -An operator polling `moq import ts`, or `moq-srt` through the same importer, -reads cumulative counters for the TR 101 290 checks that grade a contribution -feed at ingest: `TS_sync_loss`, `Sync_byte_error`, `PAT_error`, -`Continuity_count_error`, `PMT_error`, `Transport_error`, `CRC_error` (on -PAT and PMT), `PCR_repetition_error`, `PCR_discontinuity_indicator_error` and `PTS_error`. -Counters only: ETSI's fixed limits, no configuration, no verdict, and nothing -changes in what is published. - -## Plan - -Decided while planning [#1838](https://github.com/moq-dev/moq/issues/1838): - -- **The checks live beside the container**, in a private module under - `rs/moq-mux/src/container/ts` that takes 188-byte packets and keeps the - counters. No new crate: nothing outside the tree consumes one. The importer - already classifies sync, continuity and TEI for routing; routing reads the - module's classification rather than keeping a second. A packet with - `transport_error_indicator` set counts `Transport_error` and nothing else: - TR 101 290 derives no further error from it, so it feeds no continuity, - PCR, PTS or CRC check. -- **Counters are named for the ETSI check** in snake case - (`continuity_count_error`, ...), cumulative for the importer's life like - `resyncs`: stream-wide on `ts::Stats`, per PID on the PID's row. No - last-event timestamp and no green/amber/red roll-up; the rate is what an - operator alarms on, and the stats sample dates the event. -- **Limits are ETSI TR 101 290 V1.4.1's and fixed**: 0.5 s for PAT and PMT, - loss of sync after two or more consecutive corrupted sync bytes and - acquisition after five consecutive correct ones (ISO 13818-1 G.1), 100 ms - for PCR repetition (the 40 ms limit left TS 101 154 in 2005), 0 to 100 ms - for PCR discontinuity, 700 ms for PTS. The PMT names the PIDs, and every - packet is checked, so there is no sampling to configure. -- **No `PID_error` field.** ETSI defines it as a PMT-listed PID with no - packets for a user-specified period. The per-PID rows #3489 landed in - #4502 (`units`, access units delivered, and `quiet`, the time since the - last) cover it and are stricter: a dead video path behind a - live mux still sends adaptation-field-only PCR packets on its PID, which a - packet count reads as live - ([T27](https://github.com/tdrapier-wbd/moq-mpegts-paper/blob/d4c7573f9519a1c9c4882fab2021fbf82616fdcc/lab/test-27-liveness-detector.md)). - The rows keep #3489's names so no field claims the ETSI name for a different - measurement, and the consumer picks the window, as #3489 decided. The docs - map `PID_error` onto them. -- **`CRC_error` covers PAT and PMT only.** TR 101 290 names CAT, PAT, PMT, - NIT, EIT, BAT, SDT and TOT, and its table 5.1b cuts that to PAT and PMT for - systems with reduced SI, which a contribution feed is. The importer parses - only PAT and PMT, and since #4584 it - already drops a bad-CRC section and counts it as `crc_error`; this quest - adopts that field. SI captured verbatim (NIT, SDT, EIT, BAT, TOT) is not - CRC-checked, and CAT is out with the scrambled services below. -- **Intervals run on the program clock** #4502 reads, accumulated step by - step with a jump bound: one corrupt PCR taken as an absolute distance - fabricated a 23,861 s outage on every PID in the campaign's detector (T27). - A clock that stops freezes every interval; the consumer sees that case as - every count standing still across two samples. -- **PCR and PTS are graded on values, and documented as such.** Repetition and - discontinuity compare consecutive PCR values on the PCR PID, not arrival - times: that grades the encoder's insertion, which is what ingest can speak - for, while the network in front of the importer (SRT, UDP, a pipe) re-times - arrival. The docs name the domain beside the counters, because a - value-domain pass is not a wire pass - ([evidence §3.2](https://github.com/tdrapier-wbd/moq-mpegts-paper/blob/d4c7573f9519a1c9c4882fab2021fbf82616fdcc/docs/evidence.md)). - - A signalled jump is legal: the interval across a packet carrying - `discontinuity_indicator` on the PCR PID is dropped from both checks (ISO - 13818-1 2.4.3.4); the importer already reads it in `timebase_break`. The - 33-bit wrap is modular, not a jump. - - `PCR_discontinuity_indicator_error` is a consecutive difference outside 0 - to 100 ms without the flag, backwards included. On values alone a run of - missing PCRs and an unsignalled jump are the same difference, so one - forward difference over 100 ms counts under both checks; the docs say so - rather than guess which it was. - - `PTS_error` is per elementary stream on the transport clock. Verbatim - streams with no cadence (SCTE-35, DVB subtitles) are left ungraded rather - than watched against an invented number. -- **`PCR_accuracy_error` is out.** ±500 ns is graded either against a - constant-rate model of byte position, which holds only on a CBR feed, or - against arrival on the wire, which is an analyser measurement - ([T33](https://github.com/tdrapier-wbd/moq-mpegts-paper/blob/d4c7573f9519a1c9c4882fab2021fbf82616fdcc/lab/test-33-gate2-preparation.md)). -- **No opaque whole-mux lane**: - [#1861](https://github.com/moq-dev/moq/issues/1861) is closed not planned - and verbatim TS is a non-goal in - [MSFTS convergence](/quest/m2/msfts-convergence.md). P3 is out of this set; - `CAT_error` too, since the lane carries no scrambled service. A scrambled - PAT or PMT still counts under its own check. - -Implementation: - -- `rs/moq-mux/src/container/ts/import.rs`: `Stats` and `StreamStats` are - `#[non_exhaustive]`, so the fields are additive. A continuity break - declared by `discontinuity_indicator`, the one duplicate ISO 13818-1 - 2.4.3.3 permits, and a payload-less packet repeating its counter are not - errors. -- `ts::stats::Log` logs a line when a counter moves, as it does for - resyncs. -- Measure `decode` throughput on `test_data/kyrion_mpeg2av_ac3.ts` before and - after; the checks read header bytes the loop already reads, so a measurable - cost is a finding. -- Tests, one stimulus per check with a control each way: a clean capture - reads zero, the faulted one reads exactly the count injected. Feed at least - a pair of packets, since a single packet never routes. Timing arms: a - 150 ms PCR gap counts one repetition and one discontinuity error, an 80 ms - gap counts nothing, a signalled 500 ms jump counts nothing, a placed wrap - counts nothing, and a 1 s PTS gap on one PID counts one there and nowhere - else. A TEI packet that also jumps its continuity counter counts one - `Transport_error` and no `Continuity_count_error`. - -## Related - -- [TS health stats](/quest/m2/ts-health-stats.md) - where these counters are published diff --git a/rs/moq-mux/src/container/ts/health.rs b/rs/moq-mux/src/container/ts/health.rs new file mode 100644 index 0000000000..287db90203 --- /dev/null +++ b/rs/moq-mux/src/container/ts/health.rs @@ -0,0 +1,967 @@ +//! The ETSI TR 101 290 checks an [`Import`](super::Import) grades its input against. +//! +//! Counters only, at the fixed limits of TR 101 290 V1.4.1: nothing here decides what is +//! published, and nothing is configured. The importer feeds every packet through +//! [`Health::packet`], and the media PIDs and the clock route on its continuity verdict, so +//! the counter and the PES reassembly cannot disagree about which packets broke a chain. +//! +//! Intervals run on the program clock the importer already keeps +//! ([`Liveness`](super::import::Liveness)): the PCR summed step by step, which the 33-bit wrap, +//! a signalled reset or one corrupt value cannot jump. The clock starts at the first PCR after +//! the PMT, and a clock that stops freezes every interval with it. One that jumps forward +//! without a flag, by less than the 1 s bound, stretches every interval by the jump: the PCR +//! check counts the jump, and a table or PTS the stretch makes late counts too. + +use std::collections::{BTreeMap, HashMap}; + +use mpeg2ts::ts::{Pid, TsPacket}; + +use super::Stats; +use super::import::{Payload, discontinuity_indicator, payload, pcr}; +use super::mux_rate::PCR_WRAP; +use super::psi::PMT_TABLE_ID; + +const PCR_HZ: u64 = 27_000_000; +/// `PAT_error_2` and `PMT_error_2`: each table recurs at least every 0.5 s. +const TABLE_LIMIT: u64 = PCR_HZ / 2; +/// `PCR_repetition_error` and `PCR_discontinuity_indicator_error`: 100 ms. The 40 ms limit +/// left TS 101 154 in 2005. +const PCR_LIMIT: u64 = PCR_HZ / 10; +/// `PTS_error`: 700 ms. +const PTS_LIMIT: u64 = PCR_HZ * 7 / 10; +/// ISO/IEC 13818-1 G.1: sync is lost after two consecutive corrupt sync bytes, and acquired +/// after five consecutive correct ones. +const SYNC_LOST: u8 = 2; +const SYNC_ACQUIRED: u8 = 5; +const NULL_PID: u16 = 0x1fff; +const SYNC_BYTE: u8 = 0x47; + +/// What one importer counted, kept per PID so importers reading the same multiplex can be +/// merged without counting a packet twice. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub(super) struct Errors { + ts_sync_loss: u64, + sync_byte_error: u64, + pat_error: u64, + pids: BTreeMap, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +struct PidErrors { + continuity_count_error: u64, + transport_error: u64, + pmt_error: u64, + pcr_repetition_error: u64, + pcr_discontinuity_indicator_error: u64, + pts_error: u64, +} + +impl Errors { + /// Fold in another importer's count of the same input. Every importer of a multiplex sees + /// every packet, so each check takes the larger count rather than the sum. + pub(super) fn max(&mut self, other: &Self) { + self.ts_sync_loss = self.ts_sync_loss.max(other.ts_sync_loss); + self.sync_byte_error = self.sync_byte_error.max(other.sync_byte_error); + self.pat_error = self.pat_error.max(other.pat_error); + for (pid, theirs) in &other.pids { + let ours = self.pids.entry(*pid).or_default(); + ours.continuity_count_error = ours.continuity_count_error.max(theirs.continuity_count_error); + ours.transport_error = ours.transport_error.max(theirs.transport_error); + ours.pmt_error = ours.pmt_error.max(theirs.pmt_error); + ours.pcr_repetition_error = ours.pcr_repetition_error.max(theirs.pcr_repetition_error); + ours.pcr_discontinuity_indicator_error = ours + .pcr_discontinuity_indicator_error + .max(theirs.pcr_discontinuity_indicator_error); + ours.pts_error = ours.pts_error.max(theirs.pts_error); + } + } + + /// Write the stream-wide totals into `stats`, and each elementary stream's own counts into + /// its row. + pub(super) fn report(&self, stats: &mut Stats) { + stats.ts_sync_loss = self.ts_sync_loss; + stats.sync_byte_error = self.sync_byte_error; + stats.pat_error = self.pat_error; + let sum = |field: fn(&PidErrors) -> u64| self.pids.values().map(field).sum(); + stats.continuity_count_error = sum(|pid| pid.continuity_count_error); + stats.transport_error = sum(|pid| pid.transport_error); + stats.pmt_error = sum(|pid| pid.pmt_error); + stats.pcr_repetition_error = sum(|pid| pid.pcr_repetition_error); + stats.pcr_discontinuity_indicator_error = sum(|pid| pid.pcr_discontinuity_indicator_error); + stats.pts_error = sum(|pid| pid.pts_error); + for (pid, row) in &mut stats.streams { + let errors = self.pids.get(pid).copied().unwrap_or_default(); + row.continuity_count_error = errors.continuity_count_error; + row.transport_error = errors.transport_error; + row.pts_error = errors.pts_error; + } + } + + fn pid(&mut self, pid: u16) -> &mut PidErrors { + self.pids.entry(pid).or_default() + } +} + +/// The TR 101 290 checks on one importer's input. See the [module](self). +#[derive(Default)] +pub(super) struct Health { + errors: Errors, + sync: Sync, + /// Per PID, every PID but the null PID. + continuity: HashMap, + pcr_pid: Option, + /// The last PCR value on the PCR PID, forgotten across a signalled discontinuity. + pcr: Option, + pat: Timer, + /// The PMT PIDs the PAT names for the imported program. + pmt: HashMap, + /// The elementary streams with a PTS cadence, from their first PTS. + pts: HashMap, +} + +impl Health { + pub(super) fn errors(&self) -> &Errors { + &self.errors + } + + /// The framer found a packet at `at` in `buf`. Checks every sync byte position up to it. + pub(super) fn routed(&mut self, buf: &[u8], at: usize) { + self.sync.routed(buf, at, &mut self.errors); + } + + /// The importer is about to drain `buf` up to `off`: check the sync byte positions in the + /// rest of it, which no packet will be routed from, and carry the grid over. + pub(super) fn drained(&mut self, buf: &[u8], off: usize) { + self.sync.drained(buf, off, &mut self.errors); + } + + /// Grade one packet, and say what it means for the bytes already accumulated for its PID. + /// + /// A packet with `transport_error_indicator` set counts `Transport_error` and nothing else: + /// TR 101 290 derives no further error from it, so it feeds no continuity, table or PCR + /// check. A duplicate feeds none of them either, after its own continuity check. + pub(super) fn packet(&mut self, pkt: &[u8; TsPacket::SIZE], now: Option) -> Continuation { + let pid = (u16::from(pkt[1] & 0x1f) << 8) | u16::from(pkt[2]); + if pkt[1] & 0x80 != 0 { + self.errors.pid(pid).transport_error += 1; + } + // The null PID's counter is undefined (ISO/IEC 13818-1 2.4.3.3). + if pid == NULL_PID { + return Continuation::Contiguous; + } + + let continuity = self.continuity.entry(pid).or_default(); + let continuation = continuity.observe(pkt); + match continuation { + Continuation::Corrupt => return continuation, + // ISO/IEC 13818-1 2.4.3.3 permits one duplicate; a third copy is an error. + Continuation::Duplicate => { + if continuity.repeats > 1 { + self.errors.pid(pid).continuity_count_error += 1; + } + return continuation; + } + Continuation::Broken if !discontinuity_indicator(pkt) => { + self.errors.pid(pid).continuity_count_error += 1; + } + Continuation::Broken | Continuation::Contiguous => {} + } + + let scrambled = pkt[3] & 0xc0 != 0; + if pid == Pid::PAT { + if scrambled { + self.errors.pat_error += 1; + } else { + match section_start(pkt) { + Some(0x00) => self.pat.reset(now), + Some(_) => self.errors.pat_error += 1, + None => {} + } + } + } else if let Some(timer) = self.pmt.get_mut(&pid) { + if scrambled { + self.errors.pid(pid).pmt_error += 1; + } else if section_start(pkt) == Some(PMT_TABLE_ID) { + timer.reset(now); + } + } + if self.pcr_pid == Some(pid) { + self.pcr(pid, pkt); + } + continuation + } + + /// `PCR_repetition_error` and `PCR_discontinuity_indicator_error`, graded on consecutive + /// PCR values rather than on arrival, so they speak for the encoder's insertion and not for + /// the network in front of the importer. + /// + /// On values alone a run of missing PCRs and an unsignalled jump are the same difference, + /// so one forward difference over 100 ms counts under both. A backward one counts only as + /// a discontinuity. The interval across a packet that sets `discontinuity_indicator` is + /// dropped from both (ISO/IEC 13818-1 2.4.3.4), and the wrap is modular. + fn pcr(&mut self, pid: u16, pkt: &[u8; TsPacket::SIZE]) { + let value = pcr(pkt); + if discontinuity_indicator(pkt) { + self.pcr = value; + return; + } + let Some(value) = value else { + return; + }; + let Some(last) = self.pcr.replace(value) else { + return; + }; + let step = (value + PCR_WRAP - last) % PCR_WRAP; + let forward = step <= PCR_WRAP / 2; + let errors = self.errors.pid(pid); + if forward && step > PCR_LIMIT { + errors.pcr_repetition_error += 1; + } + if !forward || step > PCR_LIMIT { + errors.pcr_discontinuity_indicator_error += 1; + } + } + + /// The program clock advanced to `now`: count every table and PTS overdue at it. + pub(super) fn tick(&mut self, now: u64) { + self.errors.pat_error += self.pat.expired(now, TABLE_LIMIT); + for (pid, timer) in &mut self.pmt { + let late = timer.expired(now, TABLE_LIMIT); + if late > 0 { + self.errors.pids.entry(*pid).or_default().pmt_error += late; + } + } + for (pid, timer) in &mut self.pts { + let late = timer.expired(now, PTS_LIMIT); + if late > 0 { + self.errors.pids.entry(*pid).or_default().pts_error += late; + } + } + } + + /// The PMT PIDs the latest PAT names for the imported program. A PID it no longer names is + /// no longer graded; a new one is overdue 0.5 s from now. + pub(super) fn pmt_pids(&mut self, pids: &[u16], now: Option) { + self.pmt.retain(|pid, _| pids.contains(pid)); + for &pid in pids { + self.pmt.entry(pid).or_insert(Timer { last: now }); + } + } + + /// The PMT named the PID that carries the program clock. + pub(super) fn pcr_pid(&mut self, pid: Option) { + if self.pcr_pid != pid { + self.pcr_pid = pid; + self.pcr = None; + } + } + + /// An elementary stream with a PTS cadence started a PES that carries one. + pub(super) fn pts(&mut self, pid: u16, now: Option) { + self.pts.entry(pid).or_default().reset(now); + } + + /// `pid` changed route: forget the chain behind it, so the new route's first packet is not + /// judged against bytes it never accumulated, and stop grading its PTS until it has one. + pub(super) fn restart(&mut self, pid: u16) { + self.continuity.remove(&pid); + self.pts.remove(&pid); + } +} + +/// The table_id of the first section a PSI packet starts, if it starts one. +fn section_start(pkt: &[u8; TsPacket::SIZE]) -> Option { + if pkt[1] & 0x40 == 0 { + return None; + } + let Payload::Bytes(payload) = payload(pkt) else { + return None; + }; + let pointer = usize::from(*payload.first()?); + payload.get(1 + pointer).copied().filter(|&table_id| table_id != 0xff) +} + +/// How long since a table or a PTS last occurred, on the program clock. +#[derive(Default)] +struct Timer { + /// `None` until the clock starts. + last: Option, +} + +impl Timer { + fn reset(&mut self, now: Option) { + if now.is_some() { + self.last = 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); + if late <= limit { + return 0; + } + let count = (late - 1) / limit; + *last += count * limit; + count + } +} + +/// `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 +/// for routing and wrong for grading: ISO/IEC 13818-1 G.1 keeps striding on the old grid, and +/// only declares sync lost after two corrupt sync bytes in a row. So this follows the grid +/// the framer last locked, checks every position on it whether or not a packet was routed +/// from there, and adopts the framer's next lock once it has lost its own. +#[derive(Default)] +struct Sync { + /// The next sync byte position on the grid, relative to the importer's buffer. `None` + /// until the framer locks, and again after sync is lost. + next: Option, + acquired: bool, + /// Consecutive correct sync bytes while acquiring, or corrupt ones while acquired. + run: u8, +} + +impl Sync { + fn routed(&mut self, buf: &[u8], at: usize, errors: &mut Errors) { + self.walk(buf, at, errors); + match self.next { + // A packet on the grid. + Some(next) if next == at => { + self.byte(true, errors); + self.next = Some(at + TsPacket::SIZE); + } + // The framer locked another phase. The grid runs on into this packet's payload, + // where it is checked when the bytes are walked. + Some(_) => {} + None => { + self.next = Some(at + TsPacket::SIZE); + self.byte(true, errors); + } + } + } + + fn drained(&mut self, buf: &[u8], off: usize, errors: &mut Errors) { + self.walk(buf, buf.len(), errors); + // Every position before the end of `buf` has been checked, so the next is at or past + // `off`. + self.next = self.next.map(|next| next - off); + } + + /// Check every grid position before `end`. + fn walk(&mut self, buf: &[u8], end: usize, errors: &mut Errors) { + while let Some(next) = self.next + && next < end + { + self.next = Some(next + TsPacket::SIZE); + self.byte(buf[next] == SYNC_BYTE, errors); + } + } + + fn byte(&mut self, good: bool, errors: &mut Errors) { + if self.acquired { + if good { + self.run = 0; + return; + } + errors.sync_byte_error += 1; + self.run += 1; + if self.run >= SYNC_LOST { + errors.ts_sync_loss += 1; + self.acquired = false; + self.run = 0; + self.next = None; + } + } else if good { + self.run += 1; + if self.run >= SYNC_ACQUIRED { + self.acquired = true; + self.run = 0; + } + } else { + self.run = 0; + self.next = None; + } + } +} + +/// Whether one PID's TS packets are still an unbroken chain, for whoever is accumulating +/// bytes out of them. +/// +/// Both reassemblers on this PID need the same answer: a section and a PES are equally +/// meaningless when spliced onto bytes that didn't follow them. Keeping one implementation +/// keeps a subtle rule (which packets advance the counter, which retransmissions to ignore) +/// from drifting into two, and `Continuity_count_error` reads the same verdict. +#[derive(Default)] +pub(super) struct Continuity { + /// Last continuity_counter seen on a packet with payload, to spot gaps. + last_cc: Option, + /// Last payload packet, to identify ISO 13818-1 retransmissions. + last_pkt: Option<[u8; 188]>, + /// Copies of `last_pkt` since it arrived. + repeats: u8, +} + +/// What one packet says about the bytes already accumulated for its PID. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum Continuation { + /// A retransmission, which may refresh PCR/OPCR. Ignore it entirely to avoid duplicate bytes. + Duplicate, + /// Contiguous with the previous payload packet. + Contiguous, + /// A counter gap or a declared discontinuity. Whatever was accumulating for this PID is + /// lost, but this packet's own payload is intact and still worth routing. + Broken, + /// The demodulator flagged this packet corrupt. The partial is lost like [`Broken`], and + /// so is the packet: nothing in it can be trusted. + Corrupt, +} + +/// Whether two packets differ only in the clock fields a retransmission may refresh. +fn is_duplicate(last: &[u8; 188], pkt: &[u8; 188]) -> bool { + if last == pkt { + return true; + } + + // A refreshed clock requires the same adaptation-field shape in both packets. Comparing + // through the flags byte also pins the PID, payload-unit start, counter, length, and every + // other header bit before any bytes are ignored. + let afc = (last[3] >> 4) & 0x3; + if afc & 0x2 == 0 || last[..6] != pkt[..6] { + return false; + } + + let af_len = last[4] as usize; + let flags = last[5]; + let clock_len = usize::from(flags & 0x10 != 0) * 6 + usize::from(flags & 0x08 != 0) * 6; + let clock_end = 6 + clock_len; + let adaptation_end = 5 + af_len; + if clock_len == 0 || clock_end > adaptation_end || adaptation_end > last.len() { + return false; + } + + last[clock_end..] == pkt[clock_end..] +} + +impl Continuity { + /// Classify one 188-byte packet, recording what the next call needs. + pub(super) fn observe(&mut self, pkt: &[u8; 188]) -> Continuation { + // transport_error_indicator: the demodulator flagged this packet as corrupt, so its + // payload can't be trusted (and only a PAT or PMT has its CRC-32 checked). Forgetting the counter + // keeps the next clean packet from also looking like a gap. + if pkt[1] & 0x80 != 0 { + self.last_cc = None; + self.last_pkt = None; + return Continuation::Corrupt; + } + + let afc = (pkt[3] >> 4) & 0x3; + let has_payload = afc & 0x1 != 0; + // Read the adaptation field before the no-payload case: a discontinuity can ride on + // an adaptation-only packet, and it counts just the same. + let discontinuity = discontinuity_indicator(pkt); + + if !has_payload { + if discontinuity { + self.last_cc = None; + self.last_pkt = None; + return Continuation::Broken; + } + return Continuation::Contiguous; + } + + // A retransmission repeats the counter, and so does a loss of exactly 15 packets. + // Only the bytes tell them apart, which is why the counter alone can't decide: taking + // every repeat for a duplicate would carry a partial straight across that loss and + // join it to unrelated bytes. + if self.last_pkt.as_ref().is_some_and(|last| is_duplicate(last, pkt)) { + self.repeats = self.repeats.saturating_add(1); + return Continuation::Duplicate; + } + self.last_pkt = Some(*pkt); + self.repeats = 0; + + // Only payload packets advance the counter, so this is the one place it moves. + let cc = pkt[3] & 0x0f; + let cc_gap = matches!(self.last_cc, Some(last) if cc != (last + 1) & 0x0f); + self.last_cc = Some(cc); + if discontinuity || cc_gap { + Continuation::Broken + } else { + Continuation::Contiguous + } + } +} + +impl Stats { + /// The stream-wide TR 101 290 counters in the order of the checks, less `crc_error`, which + /// predates them and is reported on its own. + pub(super) fn health(&self) -> [u64; 9] { + [ + self.ts_sync_loss, + self.sync_byte_error, + self.pat_error, + self.continuity_count_error, + self.pmt_error, + self.transport_error, + self.pcr_repetition_error, + self.pcr_discontinuity_indicator_error, + self.pts_error, + ] + } +} + +#[cfg(test)] +mod test { + use std::collections::HashMap; + + use super::super::{Import, Programs, Stats, psi}; + use super::PCR_WRAP; + + const PMT_PID: u16 = 0x0100; + /// MPEG-2 video, read for its clock only. It carries the PCR, on adaptation-only packets + /// that repeat its counter. + const VIDEO: u16 = 0x0101; + /// A second MPEG-2 video PID, so a fault on one can be shown to stay there. + const PEER: u16 = 0x0102; + /// PCR ticks per millisecond. + const MS: u64 = 27_000; + + /// One part of the feed, for a test to leave out. + #[derive(Clone, Copy, Debug, PartialEq, Eq)] + enum Part { + Pat, + Pmt, + Pcr, + Pes(u16), + } + + /// A single-program feed built packet by packet, so a test can leave a part out without + /// breaking a continuity counter, or break one on purpose. + #[derive(Default)] + struct Feed { + cc: HashMap, + out: Vec, + } + + impl Feed { + fn header(&mut self, pid: u16, pusi: bool, adaptation: bool, payload: bool) -> Vec { + let counter = self.cc.entry(pid).or_default(); + // Only a payload packet advances the counter; one without repeats the last. + let cc = if payload { + let cc = *counter; + *counter = (cc + 1) & 0x0f; + cc + } else { + counter.wrapping_sub(1) & 0x0f + }; + let afc = (u8::from(adaptation) << 1) | u8::from(payload); + vec![ + 0x47, + (u8::from(pusi) << 6) | (pid >> 8) as u8, + pid as u8, + (afc << 4) | cc, + ] + } + + fn push(&mut self, mut pkt: Vec) { + pkt.resize(188, 0xff); + self.out.extend(pkt); + } + + fn psi(&mut self, pid: u16, section: &[u8]) { + let mut pkt = self.header(pid, true, false, true); + pkt.push(0); + pkt.extend_from_slice(section); + self.push(pkt); + } + + fn pat(&mut self) { + let entry = [0x00, 0x01, 0xe0 | (PMT_PID >> 8) as u8, PMT_PID as u8]; + self.psi(0, &psi::section(0x00, 1, 0, 0, 0, &entry)); + } + + fn pmt(&mut self) { + let mut body = vec![0xe0 | (VIDEO >> 8) as u8, VIDEO as u8, 0xf0, 0x00]; + for pid in [VIDEO, PEER] { + body.extend_from_slice(&[0x02, 0xe0 | (pid >> 8) as u8, pid as u8, 0xf0, 0x00]); + } + self.psi(PMT_PID, &psi::section(0x02, 1, 0, 0, 0, &body)); + } + + /// An adaptation-only clock packet; `flag` sets `discontinuity_indicator`. + fn pcr(&mut self, ticks: u64, flag: bool) { + let (base, ext) = (ticks / 300, ticks % 300); + let mut pkt = self.header(VIDEO, false, true, false); + pkt.extend_from_slice(&[ + 183, + 0x10 | if flag { 0x80 } else { 0 }, + (base >> 25) as u8, + (base >> 17) as u8, + (base >> 9) as u8, + (base >> 1) as u8, + ((base as u8 & 1) << 7) | 0x7e | (ext >> 8) as u8, + ext as u8, + ]); + self.push(pkt); + } + + /// A PES start with `pts` (90 kHz); `flag` gives it an adaptation field that sets + /// `discontinuity_indicator`. + fn pes(&mut self, pid: u16, pts: u64, flag: bool) { + let pts = pts & ((1 << 33) - 1); + let mut pkt = self.header(pid, true, flag, true); + if flag { + pkt.extend_from_slice(&[1, 0x80]); + } + pkt.extend_from_slice(&[ + 0x00, + 0x00, + 0x01, + 0xe0, + 0x00, + 0x00, + 0x80, + 0x80, + 0x05, + 0x21 | ((pts >> 29) as u8 & 0x0e), + (pts >> 22) as u8, + ((pts >> 14) as u8 & 0xfe) | 1, + (pts >> 7) as u8, + ((pts << 1) as u8 & 0xfe) | 1, + ]); + self.push(pkt); + } + + fn null(&mut self) { + self.push(vec![0x47, 0x1f, 0xff, 0x10]); + } + + /// The feed from `from` to `to` ms on a clock reading `clock(t)` PCR ticks at `t` ms: the + /// PAT and PMT every 100 ms, a PCR every 10 ms, a PES on each video PID every 40 ms + /// stamped on the same clock, and a null packet closing each 10 ms. `keep` drops parts. + fn run(&mut self, from: u64, to: u64, clock: impl Fn(u64) -> u64, keep: impl Fn(u64, Part) -> bool) { + for t in (from..to).step_by(10) { + if t % 100 == 0 { + if keep(t, Part::Pat) { + self.pat(); + } + if keep(t, Part::Pmt) { + self.pmt(); + } + } + if keep(t, Part::Pcr) { + self.pcr(clock(t) % PCR_WRAP, false); + } + if t % 40 == 0 { + for pid in [VIDEO, PEER] { + if keep(t, Part::Pes(pid)) { + self.pes(pid, clock(t) / 300, false); + } + } + } + self.null(); + } + } + + fn clean(&mut self, from: u64, to: u64) { + self.run(from, to, |t| t * MS, |_, _| true); + } + + /// Where each null packet starts. + fn nulls(&self) -> Vec { + (0..self.out.len()) + .step_by(188) + .filter(|&at| self.out[at + 1] & 0x1f == 0x1f && self.out[at + 2] == 0xff) + .collect() + } + + fn last_packet(&mut self) -> &mut [u8] { + let len = self.out.len(); + &mut self.out[len - 188..] + } + + /// Import the feed in `chunk`-byte pieces. + fn import(&self, chunk: usize) -> Stats { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let mut import = Import::new(broadcast, catalog.reserve()); + for piece in self.out.chunks(chunk) { + import.decode(piece).unwrap(); + } + import.stats() + } + + /// The counts, the same whether the feed arrives in whole packets or in pieces that + /// split them. + fn stats(&self) -> Stats { + let stats = self.import(188 * 7); + assert_eq!(stats, self.import(100), "the counts depend on how the input was split"); + stats + } + } + + fn all(_: u64, _: Part) -> bool { + true + } + + /// `part` is missing for `len` ms after 1 s. + fn gap(part: Part, len: u64) -> impl Fn(u64, Part) -> bool { + move |t, p| p != part || !(t > 1_000 && t < 1_000 + len) + } + + #[test] + fn a_clean_feed_counts_nothing() { + let mut feed = Feed::default(); + feed.clean(0, 3_000); + let stats = feed.stats(); + assert_eq!(stats.health(), [0; 9], "{stats:?}"); + assert_eq!(stats.crc_error, 0); + for pid in [VIDEO, PEER] { + assert_eq!(stats.streams[&pid].units, 75, "every PES on {pid:#x} was read"); + } + } + + /// The PCR is graded on its values: a gap over 100 ms counts under both checks, a shorter + /// one under neither, the wrap is no gap, and a step back is a discontinuity only. + #[test] + fn pcr_gaps_are_graded_on_their_values() { + let stats = |clock: &dyn Fn(u64) -> u64, keep: &dyn Fn(u64, Part) -> bool| { + let mut feed = Feed::default(); + feed.run(0, 2_000, clock, keep); + feed.stats() + }; + let pcr = |clock: &dyn Fn(u64) -> u64, keep: &dyn Fn(u64, Part) -> bool| { + let stats = stats(clock, keep); + let health = stats.health(); + assert_eq!( + (&health[..6], health[8]), + (&[0; 6][..], 0), + "only the PCR checks move: {stats:?}" + ); + (stats.pcr_repetition_error, stats.pcr_discontinuity_indicator_error) + }; + let steady = |t| t * MS; + assert_eq!(pcr(&steady, &gap(Part::Pcr, 150)), (1, 1), "a 150 ms gap"); + assert_eq!(pcr(&steady, &gap(Part::Pcr, 80)), (0, 0), "an 80 ms gap"); + assert_eq!(pcr(&steady, &gap(Part::Pcr, 100)), (0, 0), "a gap of exactly 100 ms"); + assert_eq!(pcr(&|t| PCR_WRAP - 1_000 * MS + t * MS, &all), (0, 0), "the wrap"); + assert_eq!( + pcr(&|t| if t < 1_000 { t * MS } else { (t - 500) * MS }, &all), + (0, 1), + "an unsignalled step back" + ); + + // An unsignalled 500 ms jump also stretches the program clock by 510 ms, which makes + // the PAT and the PMT each just late. The 40 ms PES cadence absorbs it. + let jump = stats(&|t| if t < 1_000 { t * MS } else { (t + 500) * MS }, &all); + assert_eq!( + (jump.pcr_repetition_error, jump.pcr_discontinuity_indicator_error), + (1, 1), + "an unsignalled 500 ms jump" + ); + assert_eq!((jump.pat_error, jump.pmt_error), (1, 1), "{jump:?}"); + assert_eq!(jump.health().iter().sum::(), 4, "{jump:?}"); + } + + /// The same 500 ms jump, signalled on the PCR PID, is a new time base and grades nothing. + #[test] + fn a_signalled_jump_is_not_graded() { + let mut feed = Feed::default(); + feed.clean(0, 1_000); + feed.pcr(1_500 * MS, true); + feed.run(1_000, 2_000, |t| (t + 500) * MS, all); + let stats = feed.stats(); + assert_eq!(stats.health(), [0; 9], "{stats:?}"); + } + + /// A missing PAT counts once for each half second of program clock it is missing, and + /// a PAT exactly 0.5 s late is on time. + #[test] + fn a_late_pat_counts_once_per_half_second() { + for (len, count) in [(400, 0), (500, 0), (800, 1), (1_200, 2)] { + let mut feed = Feed::default(); + feed.run(0, 3_000, |t| t * MS, gap(Part::Pat, len)); + let stats = feed.stats(); + assert_eq!(stats.pat_error, count, "PAT missing for {len} ms"); + assert_eq!(stats.pmt_error, 0, "PAT missing for {len} ms"); + } + } + + /// PID 0 carries only PATs, in the clear. + #[test] + fn pid_zero_carries_only_clear_pats() { + let mut feed = Feed::default(); + feed.clean(0, 500); + feed.psi(0, &psi::section(0x02, 1, 0, 0, 0, &[0xe1, 0x01, 0xf0, 0x00])); + feed.pat(); + feed.last_packet()[3] |= 0x80; + feed.clean(500, 1_000); + let stats = feed.stats(); + assert_eq!(stats.pat_error, 2, "{stats:?}"); + } + + #[test] + fn a_late_pmt_counts_on_its_own_check() { + let mut feed = Feed::default(); + feed.run(0, 3_000, |t| t * MS, gap(Part::Pmt, 800)); + let stats = feed.stats(); + assert_eq!((stats.pmt_error, stats.pat_error), (1, 0), "{stats:?}"); + } + + /// A 1 s gap in one stream's PTS counts once, on that stream. + #[test] + fn a_pts_gap_counts_on_its_own_pid() { + let mut feed = Feed::default(); + feed.run(0, 3_000, |t| t * MS, gap(Part::Pes(PEER), 1_000)); + let stats = feed.stats(); + assert_eq!(stats.pts_error, 1, "{stats:?}"); + assert_eq!(stats.streams[&PEER].pts_error, 1); + assert_eq!(stats.streams[&VIDEO].pts_error, 0); + } + + /// The continuity counts after a feed with `fault` applied at 500 ms on [`PEER`]: stream + /// wide, on `PEER`'s row, and the transport errors. + fn continuity(fault: impl Fn(&mut Feed)) -> (u64, u64, u64) { + let mut feed = Feed::default(); + feed.clean(0, 500); + fault(&mut feed); + feed.clean(500, 1_000); + let stats = feed.stats(); + ( + stats.continuity_count_error, + stats.streams[&PEER].continuity_count_error, + stats.transport_error, + ) + } + + #[test] + fn continuity_errors_follow_iso_13818_1() { + assert_eq!(continuity(|_| {}), (0, 0, 0), "the control"); + let lost = |feed: &mut Feed| { + feed.pes(PEER, 500 * 90, false); + feed.out.truncate(feed.out.len() - 188); + }; + assert_eq!(continuity(lost), (1, 1, 0), "a lost packet"); + let copies = |n: usize| { + move |feed: &mut Feed| { + feed.pes(PEER, 500 * 90, false); + let packet = feed.last_packet().to_vec(); + for _ in 0..n { + feed.out.extend_from_slice(&packet); + } + } + }; + assert_eq!( + continuity(copies(1)), + (0, 0, 0), + "the one duplicate ISO 13818-1 permits" + ); + assert_eq!(continuity(copies(2)), (1, 1, 0), "a third copy"); + let jump = |flag: bool| { + move |feed: &mut Feed| { + feed.cc.insert(PEER, 9); + feed.pes(PEER, 500 * 90, flag); + } + }; + assert_eq!(continuity(jump(false)), (1, 1, 0), "a counter jump"); + assert_eq!( + continuity(jump(true)), + (0, 0, 0), + "a jump declared by discontinuity_indicator" + ); + } + + /// A packet flagged corrupt counts `Transport_error` and nothing else, even when its + /// counter jumps too. + #[test] + fn a_corrupt_packet_counts_only_a_transport_error() { + let tei = |feed: &mut Feed| { + feed.cc.insert(PEER, 9); + feed.pes(PEER, 500 * 90, false); + feed.last_packet()[1] |= 0x80; + }; + assert_eq!(continuity(tei), (0, 0, 1)); + + let mut feed = Feed::default(); + feed.clean(0, 500); + tei(&mut feed); + feed.clean(500, 1_000); + let stats = feed.stats(); + assert_eq!(stats.streams[&PEER].transport_error, 1); + assert_eq!(stats.health().iter().sum::(), 1, "{stats:?}"); + + // Nor does a corrupt clock packet 300 ms out, or a corrupt packet on PID 0 that starts + // some other table. + let mut feed = Feed::default(); + feed.clean(0, 500); + feed.pcr(800 * MS, false); + feed.last_packet()[1] |= 0x80; + feed.psi(0, &psi::section(0x02, 1, 0, 0, 0, &[0xe1, 0x01, 0xf0, 0x00])); + feed.last_packet()[1] |= 0x80; + feed.clean(500, 1_000); + let stats = feed.stats(); + assert_eq!(stats.transport_error, 2, "{stats:?}"); + assert_eq!(stats.health().iter().sum::(), 2, "{stats:?}"); + } + + /// One corrupt sync byte is an error without losing sync; two in a row lose it, and five + /// good ones acquire it again, so the next corrupt byte counts once more. + #[test] + fn sync_is_lost_after_two_corrupt_sync_bytes() { + let mut feed = Feed::default(); + feed.clean(0, 2_000); + let nulls = feed.nulls(); + // A null packet: no counter to break. + feed.out[nulls[50]] = 0x00; + let stats = feed.stats(); + assert_eq!( + (stats.sync_byte_error, stats.ts_sync_loss), + (1, 0), + "one corrupt sync byte" + ); + assert_eq!(stats.health().iter().sum::(), 1, "{stats:?}"); + + // At 1.01 s the null is followed by an adaptation-only PCR packet, whose loss breaks + // no counter either. + feed.out[nulls[101]] = 0x00; + feed.out[nulls[101] + 188] = 0x00; + feed.out[nulls[150]] = 0x00; + let stats = feed.stats(); + assert_eq!((stats.sync_byte_error, stats.ts_sync_loss), (4, 1), "{stats:?}"); + assert_eq!(stats.health().iter().sum::(), 5, "{stats:?}"); + } + + /// Bytes inserted into the stream move the packet grid: the old grid finds no sync byte + /// twice running, so sync is lost once, and acquired again on the new one. + #[test] + fn a_shifted_grid_loses_sync_once() { + let mut feed = Feed::default(); + feed.clean(0, 1_000); + feed.out.extend_from_slice(&[0x00; 10]); + feed.clean(1_000, 2_000); + let stats = feed.stats(); + assert_eq!((stats.sync_byte_error, stats.ts_sync_loss), (2, 1), "{stats:?}"); + assert_eq!(stats.health().iter().sum::(), 3, "{stats:?}"); + } + + /// Every program's importer grades every packet of the multiplex, so a fault counts once. + #[test] + fn a_multiplex_counts_each_fault_once() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", crate::catalog::Config::default()); + let mut data = super::super::import::test::two_programs(); + let mut corrupt = vec![0x47, 0x9f, 0xff, 0x10]; + corrupt.resize(188, 0xff); + data.extend_from_slice(&corrupt); + let mut null = vec![0x47, 0x1f, 0xff, 0x10]; + null.resize(188, 0xff); + data.extend_from_slice(&null); + programs.decode(&data).unwrap(); + assert_eq!(programs.stats().transport_error, 1); + } +} diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index c44b833ceb..16fa63747c 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -22,6 +22,7 @@ use mpeg2ts::ts::{Pid, TsPacket}; use super::adts; use super::catalog; +use super::health::{Continuation, Continuity, Health}; use super::psi::{self, PesStart, Pmt}; use crate::catalog::hang::CatalogExt; use crate::codec::{aac, ac3, eac3, h264, h265, legacy, mp2, opus}; @@ -73,12 +74,12 @@ pub struct Import { liveness: Liveness, /// In-progress PES reassembly, keyed by elementary PID. pending: HashMap, - /// Per elementary-stream-PID TS continuity state. - continuity: HashMap, + /// The TR 101 290 checks, and the per-PID continuity state routing reads. + health: Health, /// PID the PMT designates as carrying the program clock reference. An /// adaptation-field `discontinuity_indicator` means a *system time-base* /// discontinuity only here; on any other PID it says nothing but that the - /// continuity counter jumped ([`Continuity`]). + /// continuity counter jumped ([`Continuation::Broken`]). pcr_pid: Option, /// The multiplex rate measured off the PCR PID, recorded in the `mpegts` section /// while the source holds one. Only fed with `mpegts` catalog support. @@ -154,7 +155,7 @@ impl Import { retired_stats: BTreeMap::new(), liveness: Liveness::default(), pending: HashMap::new(), - continuity: HashMap::new(), + health: Health::default(), pcr_pid: None, mux_rate: Default::default(), published: false, @@ -218,16 +219,20 @@ impl Import { // registered) before the packets that follow it in the same chunk route. let mut off = 0; while let Some(at) = self.framer.next(&self.scratch, &mut off) { + self.health.routed(&self.scratch, at); let pkt: [u8; TsPacket::SIZE] = self.scratch[at..at + TsPacket::SIZE].try_into().unwrap(); let pid = (((pkt[1] & 0x1f) as u16) << 8) | pkt[2] as u16; // Every packet paces the multiplex, null stuffing and retransmissions included. if self.supports_mpegts { self.mux_rate.packet(); } + // Every packet is graded, but only the media PIDs and the clock route on the + // verdict: PSI and section PIDs keep their reassemblers' own. + let continuation = self.health.packet(&pkt, self.liveness.now()); let continuation = Pid::new(pid) .ok() .filter(|pid| self.streams.contains_key(pid) || self.pcr_pid == Some(*pid)) - .map(|pid| self.continuity.entry(pid).or_default().observe(&pkt)); + .map(|_| continuation); // A retransmitted payload must not repeat the clock reset it carried. if matches!(continuation, Some(Continuation::Duplicate)) { continue; @@ -243,6 +248,9 @@ impl Import { self.timebase_break()?; } else if let Some(pcr) = pcr(&pkt) { self.liveness.pcr(pcr); + if let Some(now) = self.liveness.now() { + self.health.tick(now); + } if self.supports_mpegts && self.mux_rate.pcr(pcr) { self.record_mux_rate()?; } @@ -306,6 +314,7 @@ impl Import { } } + self.health.drained(&self.scratch, off); self.scratch.drain(..off); // Cut the snapshot groups for whatever SI committed in this batch. Batching per // decode call (plus the store's own host-clock debounce) coalesces a junction's @@ -363,6 +372,7 @@ impl Import { // can be read as a timebase reset rather than a counter jump. if self.pcr_pid != pmt.pcr_pid { self.pcr_pid = pmt.pcr_pid; + self.health.pcr_pid(pmt.pcr_pid.map(|pid| pid.as_u16())); // A new clock: the intervals straddling the switch measure nothing. self.liveness.discontinuity(); if self.mux_rate.discontinuity() { @@ -539,7 +549,7 @@ impl Import { self.liveness.register(pid.as_u16()); } self.streams.insert(pid, stream); - self.continuity.entry(pid).or_default(); + self.health.restart(pid.as_u16()); Ok(()) } @@ -566,7 +576,7 @@ impl Import { } // This PID is becoming section-framed; drop any partial PES a prior codec left pending. self.pending.remove(&pid); - self.continuity.remove(&pid); + self.health.restart(pid.as_u16()); if !self.supports_mpegts { // Always route to Ignored, replacing any prior codec on this PID (a later PMT // can reassign it), so a private section is never PES-parsed. Warn once. @@ -626,6 +636,11 @@ impl Import { let is_video = matches!(stream, Stream::H264 { .. } | Stream::H265 { .. } | Stream::Clock); 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() && !matches!(stream, Stream::Verbatim(_) | Stream::Ignored) { + self.health.pts(pid.as_u16(), self.liveness.now()); + } if is_video { for stream in self.streams.values_mut() { if let Stream::Aac(audio) = stream { @@ -820,6 +835,7 @@ impl Import { }, }; self.pmt_sections.entry(entry.pmt_pid).or_default(); + self.health.pmt_pids(&[entry.pmt_pid.as_u16()], self.liveness.now()); self.record_program_identity(pat.transport_stream_id, entry) } @@ -907,10 +923,19 @@ impl Import { for (pid, stats) in &mut streams { (stats.units, stats.quiet) = self.liveness.stream(*pid); } - Stats { + let mut stats = Stats { streams, crc_error: self.crc_errors.values().sum(), - } + ..Default::default() + }; + self.health.errors().report(&mut stats); + stats + } + + /// The TR 101 290 counts behind [`Self::stats`], for merging importers that read the same + /// multiplex. + pub(super) fn errors(&self) -> &super::health::Errors { + self.health.errors() } /// [`Stats::crc_error`] per PID, for merging importers that read the same PSI. @@ -953,17 +978,23 @@ fn list_programs(programs: &[u16]) -> String { } /// What each demuxed elementary stream delivered, and the audio frame sync it lost or could -/// not verify, keyed by elementary stream PID; and the PSI sections the stream as a whole -/// lost to corruption. +/// not verify, keyed by elementary stream PID; the PSI sections the stream as a whole lost to +/// corruption; and its TR 101 290 errors. /// /// Snapshot it with [`Import::stats`]. Every count is cumulative for the life of the /// importer, so what an operator alarms on is the rate: a feed that resyncs once an hour is /// healthy, one that resyncs every second is losing audio, and one whose access units stop /// advancing while the mux keeps flowing has lost that stream. /// +/// The fields named for ETSI TR 101 290 (V1.4.1) checks count them at that standard's fixed +/// limits. They grade the stream as the importer received it, not the wire a receiver sees +/// downstream. TR 101 290's `PID_error` is covered, more strictly, by each row's `units` and +/// `quiet`: a dead video path behind a live mux still sends PCR-only packets on its PID, which +/// a packet count reads as live. `PCR_accuracy_error` is not measured. +/// /// [`Export::stats`](super::Export::stats) returns the same rows for the streams it writes, /// so one schema reads both edges. Only `units` and `quiet` move there: the exporter builds -/// every frame header itself, so it has no frame sync to lose. +/// every frame header itself, so it has no frame sync to lose, and it grades nothing. #[derive(Clone, Debug, Default, PartialEq, Eq)] #[non_exhaustive] pub struct Stats { @@ -974,14 +1005,50 @@ pub struct Stats { /// PAT and PMT sections dropped because their CRC-32 did not match. Each one left the /// table already in force standing, or, before any, delayed the program's start to the /// next good repetition. Counted for the stream rather than per elementary stream, since - /// the PSI PIDs have no row of their own. + /// the PSI PIDs have no row of their own. TR 101 290 2.2 `CRC_error`, on the PAT and PMT + /// only: the reduced-SI set of its table 5.1b. Captured SI is not CRC-checked. pub crc_error: u64, + /// 1.1 `TS_sync_loss`: sync lost after two consecutive corrupt sync bytes on the packet + /// grid, and acquired again after five correct ones (ISO/IEC 13818-1 G.1). + pub ts_sync_loss: u64, + /// 1.2 `Sync_byte_error`: a byte other than 0x47 where the grid expects a sync byte, while + /// in sync. + pub sync_byte_error: u64, + /// 1.3 `PAT_error_2`: each 0.5 s of program clock without a PAT section on PID 0, and each + /// packet on PID 0 that is scrambled or starts a section other than a PAT. + pub pat_error: u64, + /// 1.4 `Continuity_count_error`, on every PID but the null PID: a counter gap, or a third + /// copy of a packet. A gap declared by `discontinuity_indicator`, the one duplicate ISO/IEC + /// 13818-1 2.4.3.3 permits, and a payload-less packet repeating its counter are not errors. + /// A loss of exactly 16 packets leaves the counter in step, and only a byte comparison + /// tells a loss of 15 from a duplicate, so the count is a floor. + pub continuity_count_error: u64, + /// 1.5 `PMT_error_2`: each 0.5 s of program clock without a PMT section on a PMT PID the + /// PAT names for the imported program, and each scrambled packet on one. + pub pmt_error: u64, + /// 2.1 `Transport_error`: packets with `transport_error_indicator` set, on any PID. Such a + /// packet counts nothing else. + pub transport_error: u64, + /// 2.3a `PCR_repetition_error`: consecutive PCR values on the PCR PID more than 100 ms + /// apart. Graded on the values, not on arrival: that speaks for the encoder's insertion, + /// which is what ingest can, while the network in front of the importer re-times arrival. + /// The interval across a `discontinuity_indicator` is not graded. + pub pcr_repetition_error: u64, + /// 2.3b `PCR_discontinuity_indicator_error`: consecutive PCR values outside 0 to 100 ms + /// apart, backwards included, without `discontinuity_indicator`. On values a run of missing + /// PCRs and an unsignalled jump are the same difference, so a forward step over 100 ms + /// counts here and as a repetition error both. + pub pcr_discontinuity_indicator_error: u64, + /// 2.5 `PTS_error`: each 700 ms of program clock without a PTS on an elementary stream + /// that has carried one. Verbatim streams are not graded: SCTE-35 and subtitles have no + /// cadence to hold. + pub pts_error: u64, } impl Stats { /// Whether no elementary stream has been registered yet and nothing has been lost. pub fn is_empty(&self) -> bool { - self.streams.is_empty() && self.crc_error == 0 + self.streams.is_empty() && self.crc_error == 0 && self.health() == [0; 9] } } @@ -1015,6 +1082,12 @@ pub struct StreamStats { /// confirm it. That substitutes audio rather than leaving a gap, which is why it is /// counted separately from a resync. pub unconfirmed: u64, + /// This PID's share of [`Stats::continuity_count_error`]. + pub continuity_count_error: u64, + /// This PID's share of [`Stats::transport_error`]. + pub transport_error: u64, + /// This PID's share of [`Stats::pts_error`]. + pub pts_error: u64, } impl StreamStats { @@ -1327,46 +1400,17 @@ impl VerbatimStream { } } -/// Whether one PID's TS packets are still an unbroken chain, for whoever is accumulating -/// bytes out of them. -/// -/// Both reassemblers on this PID need the same answer: a section and a PES are equally -/// meaningless when spliced onto bytes that didn't follow them. Keeping one implementation -/// keeps a subtle rule (which packets advance the counter, which retransmissions to ignore) -/// from drifting into two. -#[derive(Default)] -struct Continuity { - /// Last continuity_counter seen on a packet with payload, to spot gaps. - last_cc: Option, - /// Last payload packet, to identify ISO 13818-1 retransmissions. - last_pkt: Option<[u8; 188]>, -} - -/// What one packet says about the bytes already accumulated for its PID. -enum Continuation { - /// A retransmission, which may refresh PCR/OPCR. Ignore it entirely to avoid duplicate bytes. - Duplicate, - /// Contiguous with the previous payload packet. - Contiguous, - /// A counter gap or a declared discontinuity. Whatever was accumulating for this PID is - /// lost, but this packet's own payload is intact and still worth routing. - Broken, - /// The demodulator flagged this packet corrupt. The partial is lost like [`Broken`], and - /// so is the packet: nothing in it can be trusted. - Corrupt, -} - /// Whether a packet's adaptation field sets `discontinuity_indicator`. /// /// What that declares depends on the PID: a continuity-counter break on an elementary /// stream, and additionally a system time-base break on the program's PCR PID. It rides an /// adaptation-only packet (a clock packet with no payload) as readily as a payload one. -fn discontinuity_indicator(pkt: &[u8; 188]) -> bool { +pub(super) fn discontinuity_indicator(pkt: &[u8; 188]) -> bool { pkt[3] & 0x20 != 0 && pkt[4] > 0 && pkt[5] & 0x80 != 0 } /// The PCR a packet's adaptation field carries, in 27 MHz ticks. -fn pcr(pkt: &[u8; 188]) -> Option { +pub(super) fn pcr(pkt: &[u8; 188]) -> Option { if pkt[3] & 0x20 == 0 || pkt[4] < 7 || pkt[5] & 0x10 == 0 { return None; } @@ -1379,80 +1423,6 @@ fn pcr(pkt: &[u8; 188]) -> Option { Some(base * 300 + ext) } -/// Whether two packets differ only in the clock fields a retransmission may refresh. -fn is_duplicate(last: &[u8; 188], pkt: &[u8; 188]) -> bool { - if last == pkt { - return true; - } - - // A refreshed clock requires the same adaptation-field shape in both packets. Comparing - // through the flags byte also pins the PID, payload-unit start, counter, length, and every - // other header bit before any bytes are ignored. - let afc = (last[3] >> 4) & 0x3; - if afc & 0x2 == 0 || last[..6] != pkt[..6] { - return false; - } - - let af_len = last[4] as usize; - let flags = last[5]; - let clock_len = usize::from(flags & 0x10 != 0) * 6 + usize::from(flags & 0x08 != 0) * 6; - let clock_end = 6 + clock_len; - let adaptation_end = 5 + af_len; - if clock_len == 0 || clock_end > adaptation_end || adaptation_end > last.len() { - return false; - } - - last[clock_end..] == pkt[clock_end..] -} - -impl Continuity { - /// Classify one 188-byte packet, recording what the next call needs. - fn observe(&mut self, pkt: &[u8; 188]) -> Continuation { - // transport_error_indicator: the demodulator flagged this packet as corrupt, so its - // payload can't be trusted (and only a PAT or PMT has its CRC-32 checked). Forgetting the counter - // keeps the next clean packet from also looking like a gap. - if pkt[1] & 0x80 != 0 { - self.last_cc = None; - self.last_pkt = None; - return Continuation::Corrupt; - } - - let afc = (pkt[3] >> 4) & 0x3; - let has_payload = afc & 0x1 != 0; - // Read the adaptation field before the no-payload case: a discontinuity can ride on - // an adaptation-only packet, and it counts just the same. - let discontinuity = discontinuity_indicator(pkt); - - if !has_payload { - if discontinuity { - self.last_cc = None; - self.last_pkt = None; - return Continuation::Broken; - } - return Continuation::Contiguous; - } - - // A retransmission repeats the counter, and so does a loss of exactly 15 packets. - // Only the bytes tell them apart, which is why the counter alone can't decide: taking - // every repeat for a duplicate would carry a partial straight across that loss and - // join it to unrelated bytes. - if self.last_pkt.as_ref().is_some_and(|last| is_duplicate(last, pkt)) { - return Continuation::Duplicate; - } - self.last_pkt = Some(*pkt); - - // Only payload packets advance the counter, so this is the one place it moves. - let cc = pkt[3] & 0x0f; - let cc_gap = matches!(self.last_cc, Some(last) if cc != (last + 1) & 0x0f); - self.last_cc = Some(cc); - if discontinuity || cc_gap { - Continuation::Broken - } else { - Continuation::Contiguous - } - } -} - /// How long each elementary stream has gone without delivering an access unit. /// /// Measured on the program clock, the PCR summed interval by interval, so it runs straight @@ -1497,6 +1467,11 @@ impl Liveness { } } + /// PCR ticks elapsed on the program clock since its first PCR, or `None` before one. + pub(super) fn now(&self) -> Option { + self.elapsed + } + /// The clock restarted or changed PID: the next interval measures nothing. pub(super) fn discontinuity(&mut self) { self.pcr = None; @@ -1621,7 +1596,7 @@ impl PatReader { } /// Where a packet's payload is. -enum Payload<'a> { +pub(super) enum Payload<'a> { /// Adaptation field only, or reserved `adaptation_field_control`. None, Bytes(&'a [u8]), @@ -1629,7 +1604,7 @@ enum Payload<'a> { Malformed, } -fn payload(pkt: &[u8; TsPacket::SIZE]) -> Payload<'_> { +pub(super) fn payload(pkt: &[u8; TsPacket::SIZE]) -> Payload<'_> { let afc = (pkt[3] >> 4) & 0x3; if afc & 0x1 == 0 { return Payload::None; @@ -4128,6 +4103,7 @@ pub(super) mod test { resyncs: 1, discarded: 47, unconfirmed: 0, + ..Default::default() } )]), "the resync left no trace an operator could alarm on" @@ -4321,6 +4297,7 @@ pub(super) mod test { resyncs: 1, discarded: 72, unconfirmed: 0, + ..Default::default() } )]), "the resync left no trace an operator could alarm on" @@ -4508,6 +4485,7 @@ pub(super) mod test { resyncs: 0, discarded: 0, unconfirmed: 1, + ..Default::default() } )]), "the drained frame was published without a trace" diff --git a/rs/moq-mux/src/container/ts/mod.rs b/rs/moq-mux/src/container/ts/mod.rs index 3d5fdb454d..7f53ec95a4 100644 --- a/rs/moq-mux/src/container/ts/mod.rs +++ b/rs/moq-mux/src/container/ts/mod.rs @@ -20,6 +20,7 @@ mod adts; mod export; +mod health; mod import; mod mux_rate; mod programs; diff --git a/rs/moq-mux/src/container/ts/programs.rs b/rs/moq-mux/src/container/ts/programs.rs index 335ae6a161..7489c4b308 100644 --- a/rs/moq-mux/src/container/ts/programs.rs +++ b/rs/moq-mux/src/container/ts/programs.rs @@ -98,20 +98,24 @@ impl Programs { /// Every program's counters in one map: PIDs are unique across a multiplex. /// /// Every importer reads the same PAT, and programs sharing a PMT PID read the same PMT - /// sections, so a dropped section counts once however many read it. + /// sections, so a dropped section counts once however many read it. Every importer also + /// grades every packet of the multiplex, so each TR 101 290 error counts once too. pub fn stats(&self) -> Stats { let mut stats = Stats::default(); let mut crc_errors = BTreeMap::::new(); + let mut errors = super::health::Errors::default(); for program in &self.programs { stats.streams.extend(program.import.stats().streams); for (&pid, &count) in program.import.crc_errors() { let merged = crc_errors.entry(pid).or_default(); *merged = (*merged).max(count); } + errors.max(program.import.errors()); } // The importers only see the PATs after the one that started them. *crc_errors.entry(Pid::PAT).or_default() += self.scan.crc_error; stats.crc_error = crc_errors.values().sum(); + errors.report(&mut stats); stats } diff --git a/rs/moq-mux/src/container/ts/stats.rs b/rs/moq-mux/src/container/ts/stats.rs index 519c62f180..00b0dee0fb 100644 --- a/rs/moq-mux/src/container/ts/stats.rs +++ b/rs/moq-mux/src/container/ts/stats.rs @@ -56,8 +56,8 @@ impl Log { self.sync(latest); } - /// Report the streams whose frame-sync counters moved, one line each, and the PSI - /// sections dropped for a bad CRC if that count moved. + /// Report the streams whose frame-sync counters moved, one line each, the PSI sections + /// dropped for a bad CRC if that count moved, and the TR 101 290 counters if any moved. /// /// The importer already warns on each individual resync and dropped section; this is the /// running total, which is what an operator turns into a rate. @@ -68,6 +68,21 @@ impl Log { "PAT or PMT sections dropped for a bad CRC" ); } + if self.previous.as_ref().map_or([0; 9], Stats::health) != latest.health() { + tracing::info!( + ts_sync_loss = latest.ts_sync_loss, + sync_byte_error = latest.sync_byte_error, + pat_error = latest.pat_error, + continuity_count_error = latest.continuity_count_error, + pmt_error = latest.pmt_error, + transport_error = latest.transport_error, + crc_error = latest.crc_error, + pcr_repetition_error = latest.pcr_repetition_error, + pcr_discontinuity_indicator_error = latest.pcr_discontinuity_indicator_error, + pts_error = latest.pts_error, + "TR 101 290 errors in the received transport stream" + ); + } let lost = |stream: &StreamStats| (stream.resyncs, stream.discarded, stream.unconfirmed); for (pid, stream) in &latest.streams { let previous = self.previous.as_ref().and_then(|previous| previous.streams.get(pid)); @@ -188,5 +203,39 @@ mod test { _ => Err(format!("expected one line per move, got {dropped:?}")), } }); + assert!(!logs_contain("TR 101 290"), "a CRC error is reported on its own line"); + } + + /// The TR 101 290 counters are reported together, with their totals, when any of them + /// moves, and not while they hold. + #[test] + #[tracing_test::traced_test] + fn reports_tr_101_290_errors_when_they_move() { + let sample = |transport_error: u64, pcr_repetition_error: u64| Stats { + transport_error, + pcr_repetition_error, + ..Default::default() + }; + + let mut log = Log::default(); + log.sample(sample(0, 0)); + log.sample(sample(1, 0)); + log.sample(sample(1, 0)); + log.sample(sample(1, 2)); + log.finish(&sample(1, 2)); + + logs_assert(|lines: &[&str]| { + let moved: Vec<_> = lines.iter().filter(|line| line.contains("TR 101 290")).collect(); + match moved.as_slice() { + [first, second] + if [first, second].iter().all(|line| line.contains(" transport_error=1")) + && first.contains(" pcr_repetition_error=0") + && second.contains(" pcr_repetition_error=2") => + { + Ok(()) + } + _ => Err(format!("expected one line per move, got {moved:?}")), + } + }); } } From 19f7ae3a8c680eedfc497f176acd103ed8104fe3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 3 Oct 2026 09:17:47 -0700 Subject: [PATCH 2/2] fix(mux): stop grading PTS on a stream a PMT revision drops Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/health.rs | 33 +++++++++++++++++++++++---- rs/moq-mux/src/container/ts/import.rs | 2 ++ 2 files changed, 31 insertions(+), 4 deletions(-) diff --git a/rs/moq-mux/src/container/ts/health.rs b/rs/moq-mux/src/container/ts/health.rs index 287db90203..d5fb6327a3 100644 --- a/rs/moq-mux/src/container/ts/health.rs +++ b/rs/moq-mux/src/container/ts/health.rs @@ -261,6 +261,12 @@ impl Health { self.pts.entry(pid).or_default().reset(now); } + /// The elementary PIDs the latest PMT lists. A PID it no longer lists has no PTS cadence + /// left to grade. + pub(super) fn pmt_streams(&mut self, pids: &[u16]) { + self.pts.retain(|pid, _| pids.contains(pid)); + } + /// `pid` changed route: forget the chain behind it, so the new route's first packet is not /// judged against bytes it never accumulated, and stop grading its PTS until it has one. pub(super) fn restart(&mut self, pid: u16) { @@ -585,12 +591,13 @@ mod test { self.psi(0, &psi::section(0x00, 1, 0, 0, 0, &entry)); } - fn pmt(&mut self) { + /// A PMT listing `pids` as MPEG-2 video, with the clock on [`VIDEO`]. + fn pmt(&mut self, version: u8, pids: &[u16]) { let mut body = vec![0xe0 | (VIDEO >> 8) as u8, VIDEO as u8, 0xf0, 0x00]; - for pid in [VIDEO, PEER] { + for &pid in pids { body.extend_from_slice(&[0x02, 0xe0 | (pid >> 8) as u8, pid as u8, 0xf0, 0x00]); } - self.psi(PMT_PID, &psi::section(0x02, 1, 0, 0, 0, &body)); + self.psi(PMT_PID, &psi::section(0x02, 1, version, 0, 0, &body)); } /// An adaptation-only clock packet; `flag` sets `discontinuity_indicator`. @@ -651,7 +658,7 @@ mod test { self.pat(); } if keep(t, Part::Pmt) { - self.pmt(); + self.pmt(0, &[VIDEO, PEER]); } } if keep(t, Part::Pcr) { @@ -824,6 +831,24 @@ mod test { assert_eq!(stats.streams[&VIDEO].pts_error, 0); } + /// A stream a PMT revision drops has no PTS cadence left to grade. + #[test] + fn a_stream_the_pmt_drops_has_no_pts_cadence() { + let mut feed = Feed::default(); + feed.clean(0, 1_000); + for t in (1_000..3_000).step_by(100) { + feed.pmt(1, &[VIDEO]); + feed.run( + t, + t + 100, + |t| t * MS, + |_, part| part != Part::Pmt && part != Part::Pes(PEER), + ); + } + let stats = feed.stats(); + assert_eq!(stats.health(), [0; 9], "{stats:?}"); + } + /// The continuity counts after a feed with `fault` applied at 500 ms on [`PEER`]: stream /// wide, on `PEER`'s row, and the transport errors. fn continuity(fault: impl Fn(&mut Feed)) -> (u64, u64, u64) { diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 16fa63747c..fcd24f38d6 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -414,6 +414,8 @@ impl Import { self.ensure_stream(es.pid, es.stream_type, &es.descriptors)?; } } + let pids: Vec = pmt.streams.iter().map(|es| es.pid.as_u16()).collect(); + self.health.pmt_streams(&pids); // Every stream in the initial program is registered now; release the reservation // so the catalog publishes once each rendition's config resolves, not before.