From 14e9b40c845dcdb186180dd2495aea5a608f90b4 Mon Sep 17 00:00:00 2001 From: t0ms Date: Wed, 30 Sep 2026 12:25:42 +0100 Subject: [PATCH 1/5] feat(mux): report each TS elementary stream's access units at export Export::stats returns ts::Stats with the import's per-PID row for every elementary stream written: units, and quiet on the output's own PCR. The other fields stay zero at egress. `moq export ts` logs a row that stops advancing, as `moq import ts` does. Quest: quest/m2/ts-export-liveness.md. Also corrects the PCR repetition limit to TR 101 290 V1.4.1's 100 ms in the exporter comment and the test/ts grader defaults, and describes the exporter's current padding in test/ts. Co-Authored-By: Claude Opus 5.5 --- rs/moq-cli/src/subscribe.rs | 9 ++ rs/moq-mux/src/container/ts/export.rs | 64 +++++++++- rs/moq-mux/src/container/ts/export_test.rs | 140 ++++++++++++++++++++- rs/moq-mux/src/container/ts/import.rs | 32 +++-- rs/moq-mux/src/container/ts/stats.rs | 7 +- test/ts/README.md | 8 +- test/ts/compliance.py | 2 +- test/ts/pcr-timing.py | 2 +- test/ts/run.sh | 3 +- 9 files changed, 244 insertions(+), 23 deletions(-) diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index 2d1d8f67c5..96d3ba18bc 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -322,6 +322,10 @@ impl Subscribe { // hand-rolled poll instead of `ts.next()`. let mut delivery = Delivery::new(self.args.max_age); let linger = self.args.linger; + // Reports a track that stops reaching the output while the rest keeps flowing, + // the way `publish` reports one that stops arriving. + let mut log = moq_mux::container::ts::stats::Log::default(); + let mut sampled = tokio::time::Instant::now(); loop { let end = loop { let mut waited = false; @@ -341,6 +345,11 @@ impl Subscribe { }; delivery.update(&frame, ts.discontinuity()); delivery.deliver(&frame, waited, &mut stdout).await?; + + if sampled.elapsed() >= moq_mux::container::ts::stats::Log::INTERVAL { + sampled = tokio::time::Instant::now(); + log.sample(ts.stats()); + } }; // Any end waits out the linger, and on expiry the last one is the result: a diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 5457c6eaa1..e4fd8d802f 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -58,7 +58,7 @@ const PSI_INTERVAL: Duration = Duration::from_millis(500); /// table asking for faster repetition than this still gets it. const SI_REVISION_INTERVAL: Duration = Duration::from_secs(1); /// Emit a PCR on every crossing of this media-time grid ([`Export::emit`]). -/// TR 101 290 flags a gap over 40 ms; broadcast muxes emit every 25-40 ms. +/// TR 101 290 V1.4.1 flags a gap over 100 ms; broadcast muxes emit every 25-40 ms. pub(super) const PCR_INTERVAL: Duration = Duration::from_millis(25); /// How many missed PCR slots to backfill at most: one second's worth. Frames /// coarser than the grid cross several slots at a time and every one is filled so @@ -197,6 +197,11 @@ pub struct Export { /// up before it ever configures video. `None` until the tables are built, and for /// programs with no video track (nothing to align to). video_start: Option, + /// Access units per PID muxed into the open span, counted once it goes out, since a + /// rewind discards it unwritten. + span_units: BTreeMap, + /// Each elementary stream's access units and silence on the PCR written ([`Self::stats`]). + liveness: super::import::Liveness, } struct Pending { @@ -409,6 +414,23 @@ enum Kind { }, } +impl Kind { + /// The track suffix [`Import`](super::Import) gives a PID carrying this kind, so a stream's + /// row is named alike at both edges. + fn suffix(&self) -> &'static str { + match self { + Kind::Video(StreamType::H265) => ".hev1", + Kind::Video(_) => ".avc3", + Kind::Aac { .. } => ".aac", + Kind::Opus { .. } => ".opus", + Kind::Mp2 { .. } => ".mp2", + Kind::Ac3 => ".ac3", + Kind::Eac3 => ".eac3", + Kind::Verbatim { .. } => ".ts", + } + } +} + /// The null stuffing owed to the multiplex rate ([`Export::stuff`]). #[derive(Default)] struct Stuffing { @@ -699,6 +721,8 @@ impl Export { mux_rate: None, mux_rate_override: None, stuffing: Stuffing::default(), + span_units: BTreeMap::new(), + liveness: Default::default(), }) } @@ -1258,6 +1282,30 @@ impl Export { self.emitted_epoch } + /// Snapshot the access units each elementary stream has written, and how long each has + /// been quiet on the PCR the output carries. + /// + /// A track stalled upstream stops advancing its row while the PSI and the other PIDs + /// keep flowing, which nothing graded on the output bytes alone can see. Empty until the + /// program tables are built. Cheap enough to poll per frame. + pub fn stats(&self) -> super::Stats { + let mut stats = super::Stats::default(); + if self.psi.is_none() { + return stats; + } + for track in self.tracks.values() { + let (units, quiet) = self.liveness.stream(track.pid); + let row = super::StreamStats { + track: track.kind.suffix(), + units, + quiet, + ..Default::default() + }; + stats.streams.insert(track.pid, row); + } + stats + } + /// Carry on with the broadcast that replaced the one this export was reading. /// /// Call it once [`Source::returned`](crate::Source::returned) has resolved, after @@ -1298,6 +1346,7 @@ impl Export { } self.pending.clear(); self.keyframes.clear(); + self.span_units.clear(); self.queue.clear(); self.watermark = None; // The new generation waits for every track again, with a fresh budget: a @@ -1498,6 +1547,9 @@ impl Export { es_info, }; + for track in &tracks { + self.liveness.register(track.pid); + } self.psi = Some(Psi { pat, pmt, @@ -1657,6 +1709,8 @@ impl Export { } } + // A malformed section writes nothing, so the unit counts only if bytes went in. + let written = out.len(); match es_payload { // Section-framed verbatim (SCTE-35, ...) rides in private sections, not PES; // carry the bytes verbatim. @@ -1682,6 +1736,9 @@ impl Export { self.write_pes(&mut out, &unit, &es_payload)?; } } + if out.len() > written { + *self.span_units.entry(pid).or_default() += 1; + } if keyframe { self.keyframes.push(self.pending.len()); } @@ -1821,6 +1878,9 @@ impl Export { if let Some(cc) = counter_before(&bytes, bytes.len(), pcr_pid, None) { self.pcr_cc = Some(cc); } + for (pid, units) in std::mem::take(&mut self.span_units) { + self.liveness.delivered(pid, units); + } Ok(()) } @@ -1933,7 +1993,9 @@ impl Export { let mut packet = pcr_packet(pcr_pid, ticks, cc)?; if std::mem::take(&mut self.pcr_discontinuity) { packet[5] |= 0x80; + self.liveness.discontinuity(); } + self.liveness.pcr((ticks & TS_TIMESTAMP_MASK) * 300); Ok(packet) } diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index b6b4eef985..74a2bf297f 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -17,7 +17,7 @@ use mpeg2ts::ts::{ReadTsPacket, TsPacketReader, TsPayload}; use crate::catalog::hang::Container as HangContainer; use crate::container::ts::export::PCR_INTERVAL; -use crate::container::ts::{Export, catalog as tscat}; +use crate::container::ts::{Export, Stats, catalog as tscat}; use crate::container::{Frame, Kind, Producer}; use moq_net::Timestamp; @@ -5474,3 +5474,141 @@ async fn resume_after_a_finish() { async fn resume_after_a_drop() { resume_after(false).await; } + +/// Export 25 fps video and two AAC tracks for [`TICKS`] video frames, the video and the first +/// audio track stopping after `stop` while the second carries on. Returns the stats sampled +/// after `sample` and at the end, and the frames rendered in between. +/// +/// Drained after every write, like [`export_twice`], so the export reads every frame and a row +/// stops only because its track did. +async fn export_liveness(sample: u64, stop: u64) -> (Stats, Stats, Vec) { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + + let track = broadcast + .create_track("video.avc1", hang::container::track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + let mut cfg = VideoConfig::new(H264 { + profile: 0x42, + constraints: 0xc0, + level: 0x1f, + inline: false, + }); + cfg.container = Container::Legacy; + cfg.description = + Some(crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap()); + catalog + .modify() + .unwrap() + .video + .renditions + .insert(track.name().to_string(), cfg); + let mut video = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data)); + let mut audio = [ + aac_rendition(&mut broadcast, &mut catalog, "primary.aac"), + aac_rendition(&mut broadcast, &mut catalog, "secondary.aac"), + ]; + + let mut export = Export::new(crate::source::announced(&consumer)).await.unwrap(); + assert!(export.stats().is_empty(), "no row before the program tables"); + let (mut sampled, mut frames) = (None, Vec::new()); + let mut audio_index = 0; + for tick in 0..TICKS { + if tick < stop { + let keyframe = tick % GOP == 0; + let slice = if keyframe { + vec![0x65u8; 3_000] + } else { + vec![0x41u8; 400] + }; + video + .write(Frame { + timestamp: Timestamp::from_micros(tick * VIDEO_US).unwrap(), + duration: None, + payload: length_prefixed(&[&slice]), + keyframe, + }) + .unwrap(); + } + while audio_index * AUDIO_US < (tick + 1) * VIDEO_US { + let running = if tick < stop { &mut audio[..] } else { &mut audio[1..] }; + for track in running { + track + .write(Frame { + timestamp: Timestamp::from_micros(audio_index * AUDIO_US).unwrap(), + duration: None, + payload: Bytes::from_iter((0..180u16).map(|i| (i ^ audio_index as u16) as u8)), + keyframe: audio_index % AUDIO_GROUP == 0, + }) + .unwrap(); + } + audio_index += 1; + } + + let out = drain_frames(&mut export).await; + if sampled.is_some() { + frames.extend(out); + } + if tick + 1 == sample { + sampled = Some(export.stats()); + } + } + (sampled.expect("sampled mid-run"), export.stats(), frames) +} + +/// A healthy export advances every elementary stream's row, each quiet for no longer than +/// its own frame spacing plus the mux buffer, and leaves the import-only counters at zero. +#[tokio::test(start_paused = true)] +async fn export_stats_advance_every_stream() { + let (mid, end, _) = export_liveness(TICKS / 2, TICKS).await; + + let tracks: Vec<&str> = end.streams.values().map(|row| row.track).collect(); + assert_eq!(tracks, [".aac", ".aac", ".avc3"], "one row per elementary stream"); + for (pid, row) in &end.streams { + let before = &mid.streams[pid]; + assert!(before.units > 0, "PID {pid:#x} wrote nothing by mid-run"); + assert!( + row.units > before.units, + "PID {pid:#x} stopped advancing: {before:?} -> {row:?}" + ); + let quiet = row.quiet.expect("the output carries a PCR"); + assert!(quiet < Duration::from_millis(200), "PID {pid:#x} quiet for {quiet:?}"); + assert_eq!((row.resyncs, row.discarded, row.unconfirmed), (0, 0, 0)); + } +} + +/// The #3533 shape: video and the primary audio stop reaching the exporter while the +/// secondary audio carries on. Their rows freeze and their quiet time grows on the output's +/// own PCR, which keeps running on the surviving track, as do the PSI and that track's row. +#[tokio::test(start_paused = true)] +async fn export_stats_catch_a_track_that_stops() { + let stop = TICKS / 2; + // Sampled once the mux buffer has written out what the stopped tracks sent last. + let (sampled, end, after) = export_liveness(stop + 5, stop).await; + let stalled = Duration::from_micros((TICKS - stop) * VIDEO_US); + + let advanced = |pid: &u16| end.streams[pid].units > sampled.streams[pid].units; + let (running, stopped): (Vec, Vec) = end.streams.keys().copied().partition(advanced); + let [running] = running[..] else { + panic!("expected one stream to keep advancing, got {running:?}: {end:?}"); + }; + assert_eq!(end.streams[&running].track, ".aac"); + assert_eq!(stopped.len(), 2, "video and the primary audio stopped"); + + let quiet = |pid: u16| end.streams[&pid].quiet.expect("the output carries a PCR"); + assert!(quiet(running) < Duration::from_millis(200), "{:?}", quiet(running)); + for pid in stopped { + // The mux buffer holds the last span back, so the output's clock trails the media. + assert!( + quiet(pid) > stalled - Duration::from_millis(200), + "PID {pid:#x} quiet for only {:?} of a {stalled:?} stall", + quiet(pid) + ); + } + assert!( + count_pid(&after, 0x0000) >= 3, + "the PAT kept repeating through the stall" + ); + assert!(count_pid(&after, running) > 0, "the surviving track kept flowing"); +} diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 5f4c10ef23..93ded5c7f4 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -1020,11 +1020,16 @@ fn list_programs(programs: &[u16]) -> String { /// 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. +/// +/// [`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. #[derive(Clone, Debug, Default, PartialEq, Eq)] #[non_exhaustive] pub struct Stats { - /// One row per elementary stream PID the importer carries, ordered by PID. A PID the - /// importer drops (an undecoded stream without the `mpegts` catalog section) has none. + /// One row per elementary stream PID the importer carries or the exporter writes, ordered + /// by PID. A PID the importer drops (an undecoded stream without the `mpegts` catalog + /// section) has none. pub streams: BTreeMap, } @@ -1040,13 +1045,16 @@ impl Stats { #[non_exhaustive] pub struct StreamStats { /// The current or most recent MoQ track suffix for this PID (`.avc3`, `.mp2`, `.ts`, ...), - /// or empty for MPEG-1/2 video, which is read for its clock and not published. + /// or empty for MPEG-1/2 video, which is read for its clock and not published. At export, + /// the suffix an import of the output would give the PID. pub track: &'static str, /// Access units the stream delivered: frames published for decoded media, PES payloads or - /// sections carried verbatim, and PES read on MPEG-1/2 video. + /// sections carried verbatim, and PES read on MPEG-1/2 video. At export, the PES or + /// sections written for each frame. pub units: u64, /// Transport time since the stream last delivered an access unit, or since the PMT - /// declared it, measured on the program clock (PCR). `None` until the first PCR arrives. + /// declared it, measured on the program clock (PCR): the one read at import, the one + /// written at export. `None` until the first PCR. /// /// No threshold applies: a sparse stream such as SCTE-35 is legitimately quiet for /// seconds, so how long is too long is for whoever alarms to decide. @@ -1521,9 +1529,9 @@ impl Continuity { /// Measured on the program clock, the PCR summed interval by interval, so it runs straight /// through the 33-bit wrap, a signalled time-base reset and a corrupt PCR instead of jumping /// with them. Not the media clock: that follows the video PTS, and stops with the very stream -/// this catches. +/// this catches. [`Export`](super::Export) runs the same meter on the PCR it writes. #[derive(Default)] -struct Liveness { +pub(super) struct Liveness { /// The last PCR in 27 MHz ticks, forgotten at a reset so no interval spans one. pcr: Option, /// PCR ticks elapsed since the first PCR, if one has arrived. @@ -1535,7 +1543,7 @@ struct Liveness { impl Liveness { /// A PCR arrived on the program's clock PID. - fn pcr(&mut self, pcr: u64) { + pub(super) fn pcr(&mut self, pcr: u64) { let elapsed = self.elapsed.get_or_insert(0); if let Some(last) = self.pcr.replace(pcr) { let step = (pcr + super::mux_rate::PCR_WRAP - last) % super::mux_rate::PCR_WRAP; @@ -1549,18 +1557,18 @@ impl Liveness { } /// The clock restarted or changed PID: the next interval measures nothing. - fn discontinuity(&mut self) { + pub(super) fn discontinuity(&mut self) { self.pcr = None; } /// The PMT declared `pid`. Its silence counts from here until it delivers. - fn register(&mut self, pid: u16) { + pub(super) fn register(&mut self, pid: u16) { let now = self.elapsed.unwrap_or(0); self.streams.entry(pid).or_insert((0, now)); } /// `pid` delivered `units` access units just now. - fn delivered(&mut self, pid: u16, units: u64) { + pub(super) fn delivered(&mut self, pid: u16, units: u64) { if units == 0 { return; } @@ -1572,7 +1580,7 @@ impl Liveness { } /// `pid`'s access units and how long it has been quiet. See [`StreamStats`]. - fn stream(&self, pid: u16) -> (u64, Option) { + pub(super) fn stream(&self, pid: u16) -> (u64, Option) { let Some(&(units, last)) = self.streams.get(&pid) else { return (0, None); }; diff --git a/rs/moq-mux/src/container/ts/stats.rs b/rs/moq-mux/src/container/ts/stats.rs index 9853721168..bc886b9714 100644 --- a/rs/moq-mux/src/container/ts/stats.rs +++ b/rs/moq-mux/src/container/ts/stats.rs @@ -1,11 +1,12 @@ -//! Reporting on the per-stream counters [`Import::stats`](super::Import::stats) returns. +//! Reporting on the per-stream counters [`Import::stats`](super::Import::stats) and +//! [`Export::stats`](super::Export::stats) return. use std::collections::BTreeSet; use super::{Stats, StreamStats}; -/// Logs what moved in an importer's [`Stats`] from one sample to the next, so every front -/// door that runs a TS importer reports the same lines. +/// Logs what moved in an importer's or exporter's [`Stats`] from one sample to the next, so +/// every front door that runs one reports the same lines. /// /// Feed it a snapshot every [`INTERVAL`](Self::INTERVAL) of wall time. The caller owns the /// timer, since what counts as "now" differs between a CLI, a server, and a test. diff --git a/test/ts/README.md b/test/ts/README.md index 4fb5950b9d..336ceab858 100644 --- a/test/ts/README.md +++ b/test/ts/README.md @@ -7,7 +7,9 @@ runs [TSDuck](https://tsduck.io) plus a custom analyzer over it. This is a diagnostic gate, not just a pass/fail: the exporter ([`rs/moq-mux/src/container/ts/export.rs`](../../rs/moq-mux/src/container/ts/export.rs)) -is VBR, inserts no null packets, and paces PCR once per media frame, so several +pads with null packets to the multiplex rate the source recorded (or `--mux-rate`), +leaves a source without one unpadded, never delays media to fit the rate, and +puts a PCR on its own packet every 25 ms of media time, so several broadcast-shape checks are expected to flag. The report quantifies exactly where and by how much. @@ -80,7 +82,7 @@ Severities: **hard** checks fail the run by default; **shape** checks report as | `pcr-presence` | hard | a PCR PID is declared and carries PCR | | `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-repetition` | shape | consecutive PCRs within the limit (default 100 ms, TR 101 290 V1.4.1) | | `pcr-jitter` | shape | per-interval PCR jitter vs the nominal bitrate (pcrverify model) | | `null-ratio` | shape | null/stuffing fraction (flags only a pathological excess) | | `service-descriptors` | shape | an SDT naming the service is present | @@ -146,7 +148,7 @@ grades only how evenly the bytes are laid over the PCRs. | `sync` | hard | no invalid sync bytes / transport-error packets | | `continuity` | hard | no discontinuities, and a payload-less packet must not advance the counter (ISO 13818-1 2.4.3.3) | | `pcr-single-pid` | hard | every PCR rides one PID | -| `pcr-value-interval` | hard | no interval above `--repetition-ms` (default 40, TR 101 290), within one time base | +| `pcr-value-interval` | hard | no interval above `--repetition-ms` (default 100, TR 101 290 V1.4.1), within one time base | | `pcr-release-timing` | hard | no more than `--release-pct-max` of intervals arrive further than `--release-ms` from the interval their own values assert, and accumulated drift stays within `--drift-ms`, being the standing lag the sender is allowed to hold; a sample below `--live-min-pcr` PCRs or `--live-cover-pct` of the window is a failure, not a pass (`--live` only) | | `pcr-position` | shape | share of PCR packets within `--adjacent-packets` of the previous one | | `pcr-schedule` | shape | share of PCR intervals, on the busiest PCR PID, whose bytes are within `--schedule-tolerance-pct` (default 1) or one packet of what `--mux-rate` implies (estimated from the capture if not given); hard, at that share, when `--schedule-pct-min` is given | diff --git a/test/ts/compliance.py b/test/ts/compliance.py index 66aae98fe7..2774948c6f 100755 --- a/test/ts/compliance.py +++ b/test/ts/compliance.py @@ -72,7 +72,7 @@ class Check: class Thresholds: """IRD limits and T-STD model parameters, all overridable from the CLI.""" - pcr_repetition_ms: float = 40.0 + pcr_repetition_ms: float = 100.0 pcr_jitter_us: float = 500.0 null_ratio_max: float = 0.90 bitrate_cov_max: float = 0.10 diff --git a/test/ts/pcr-timing.py b/test/ts/pcr-timing.py index b318fe3904..9c08abe889 100755 --- a/test/ts/pcr-timing.py +++ b/test/ts/pcr-timing.py @@ -605,7 +605,7 @@ def main(): ap.add_argument("--live", action="store_true", help="read stdin and stamp arrivals, grading release timing") ap.add_argument("--seconds", type=float, default=45.0, help="live capture window (default 45)") ap.add_argument( - "--repetition-ms", type=float, default=40.0, help="max PCR value interval, TR 101 290 (default 40)" + "--repetition-ms", type=float, default=100.0, help="max PCR value interval, TR 101 290 V1.4.1 (default 100)" ) ap.add_argument( "--release-ms", diff --git a/test/ts/run.sh b/test/ts/run.sh index 0858b89dad..d30320d1e4 100755 --- a/test/ts/run.sh +++ b/test/ts/run.sh @@ -7,7 +7,8 @@ # TSDuck + custom analyzer in compliance.py against the capture. The point is to # tell whether what the subscriber emits is something an Integrated # Receiver/Decoder would accept, and to quantify where it diverges (the exporter -# is VBR, emits no null packets, and paces PCR per frame). +# pads to the recorded multiplex rate but never delays media to fit it, and puts a +# PCR every 25 ms of media time). # # Modes: # ./run.sh # generate a clip, round-trip it, analyze From 3f136b171162f66c8a53cb94349ac564fcbe5bde Mon Sep 17 00:00:00 2001 From: t0ms Date: Wed, 30 Sep 2026 13:38:21 +0100 Subject: [PATCH 2/5] fix(mux): count the whole PCR jump across a long media gap at export The exporter backfills at most a second of PCRs across a media gap, then steps its clock forward unflagged. Liveness dropped that step as a corrupt PCR, so a stopped track's quiet undercounted by the rest of the gap. The exporter's own PCRs now count every forward step; the importer's filter is unchanged. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/export.rs | 2 +- rs/moq-mux/src/container/ts/export_test.rs | 39 ++++++++++++++++++++++ rs/moq-mux/src/container/ts/import.rs | 20 ++++++++--- 3 files changed, 56 insertions(+), 5 deletions(-) diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index e4fd8d802f..8e471c1311 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -1995,7 +1995,7 @@ impl Export { packet[5] |= 0x80; self.liveness.discontinuity(); } - self.liveness.pcr((ticks & TS_TIMESTAMP_MASK) * 300); + self.liveness.written_pcr((ticks & TS_TIMESTAMP_MASK) * 300); Ok(packet) } diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 74a2bf297f..f38c7470c7 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -5612,3 +5612,42 @@ async fn export_stats_catch_a_track_that_stops() { ); assert!(count_pid(&after, running) > 0, "the surviving track kept flowing"); } + +/// A stopped track's silence counts the whole of a media gap on the surviving track, not just +/// the one second of clock the exporter backfills: the output's PCR jumps across the rest +/// unflagged, and that jump is time the stopped track was silent. +#[tokio::test(start_paused = true)] +async fn export_stats_count_a_gap_longer_than_the_backfill() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let mut stopped = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); + let mut running = aac_rendition(&mut broadcast, &mut catalog, "b.aac"); + + let mut export = Export::new(crate::source::announced(&consumer)).await.unwrap(); + for ms in (0..2_000).step_by(20) { + write_aac(&mut stopped, ms); + write_aac(&mut running, ms); + drain_frames(&mut export).await; + } + // Five seconds with nothing on either track, then only the second resumes. + for ms in (7_000..8_000).step_by(20) { + write_aac(&mut running, ms); + drain_frames(&mut export).await; + } + + let stats = export.stats(); + let mut rows: Vec<_> = stats.streams.values().collect(); + rows.sort_by_key(|row| row.units); + let [stopped, running] = rows[..] else { + panic!("expected two rows: {stats:?}"); + }; + assert!(stopped.units < running.units, "{stats:?}"); + let quiet = stopped.quiet.expect("the output carries a PCR"); + assert!( + quiet > Duration::from_millis(5_800), + "stopped at 2 s, quiet for only {quiet:?} at 8 s" + ); + let quiet = running.quiet.expect("the output carries a PCR"); + assert!(quiet < Duration::from_millis(200), "{quiet:?}"); +} diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 93ded5c7f4..21484fe68b 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -1543,14 +1543,26 @@ pub(super) struct Liveness { impl Liveness { /// A PCR arrived on the program's clock PID. + /// + /// Stepped one interval at a time, so a corrupt PCR, or the clock stepping back without a + /// flag, costs one interval rather than inventing hours of silence on every PID. The bound + /// is the mux-rate meter's. pub(super) fn pcr(&mut self, pcr: u64) { + self.step(pcr, super::mux_rate::MAX_INTERVAL); + } + + /// A PCR the caller wrote itself. Its clock breaks only where the caller flags a + /// discontinuity, so every forward step counts, including the jump across a media gap + /// longer than the backfill covers. + pub(super) fn written_pcr(&mut self, pcr: u64) { + self.step(pcr, super::mux_rate::PCR_WRAP); + } + + fn step(&mut self, pcr: u64, max: u64) { let elapsed = self.elapsed.get_or_insert(0); if let Some(last) = self.pcr.replace(pcr) { let step = (pcr + super::mux_rate::PCR_WRAP - last) % super::mux_rate::PCR_WRAP; - // Stepped one interval at a time, so a corrupt PCR, or the clock stepping back - // without a flag, costs one interval rather than inventing hours of silence on - // every PID. The bound is the mux-rate meter's. - if step <= super::mux_rate::MAX_INTERVAL { + if step <= max { *elapsed += step; } } From 3c80593f844780dbb3b50fe67a5d669ed2c5ad28 Mon Sep 17 00:00:00 2001 From: t0ms Date: Wed, 30 Sep 2026 15:54:04 +0100 Subject: [PATCH 3/5] fix(mux): count TS export stats only for returned output emit() credited a span's units and PCRs as it queued them, so an error in poll_next before they were returned, followed by resume() dropping the queue, left the stats counting output that never went out. Each queued frame now carries its own tally, applied when it is returned. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/export.rs | 93 ++++++++++++++++------ rs/moq-mux/src/container/ts/export_test.rs | 60 ++++++++++++++ 2 files changed, 128 insertions(+), 25 deletions(-) diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 8e471c1311..2ec7f838d5 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -158,8 +158,9 @@ pub struct Export { /// Offsets into [`pending`](Self::pending) where a keyframe's packets begin, so /// the output frame carrying one keeps the flag. keyframes: Vec, - /// Output frames ready to hand out, one per grid slot the last span covered. - queue: VecDeque, + /// Output frames ready to hand out, one per grid slot the last span covered, each with + /// what it tells [`Self::stats`] once returned. + queue: VecDeque<(Frame, Tally)>, /// Continuity counter of the last packet emitted on the PCR PID. A clock packet /// carries no payload, so it repeats whatever preceded it on the wire rather /// than advancing the counter ([`Export::pcr_at`]). @@ -197,13 +198,26 @@ pub struct Export { /// up before it ever configures video. `None` until the tables are built, and for /// programs with no video track (nothing to align to). video_start: Option, - /// Access units per PID muxed into the open span, counted once it goes out, since a - /// rewind discards it unwritten. - span_units: BTreeMap, - /// Each elementary stream's access units and silence on the PCR written ([`Self::stats`]). + /// Offsets into [`pending`](Self::pending) where each access unit's packets begin, with + /// its PID, so the output frame carrying it counts it. + units: Vec<(usize, u16)>, + /// Clock packets written into the output frame being laid out ([`Self::push`]). + tally: Tally, + /// Each elementary stream's access units and silence on the PCR returned ([`Self::stats`]). liveness: super::import::Liveness, } +/// What a queued output frame tells [`Export::stats`], applied only once the frame is +/// returned: a rewind drops the queue unwritten. +#[derive(Default)] +struct Tally { + /// The frame's clock packets in order: the PCR in 27 MHz ticks, and whether it flags a + /// new time base. + pcrs: Vec<(u64, bool)>, + /// The PID of each access unit whose packets begin in the frame, after its clock packets. + units: Vec, +} + struct Pending { frame: Frame, discontinuity: u64, @@ -721,7 +735,8 @@ impl Export { mux_rate: None, mux_rate_override: None, stuffing: Stuffing::default(), - span_units: BTreeMap::new(), + units: Vec::new(), + tally: Tally::default(), liveness: Default::default(), }) } @@ -844,8 +859,7 @@ impl Export { // own values imply and lets the caller's pacer release each at the instant // it asserts. See [`Self::advance`]. loop { - if let Some(out) = self.queue.pop_front() { - self.emitted_epoch = self.epoch; + if let Some(out) = self.pop() { return Poll::Ready(Ok(Some(out))); } let Some(name) = self.pick_next_track(waiter) else { @@ -902,8 +916,7 @@ impl Export { let drained = !self.tracks.is_empty() && self.tracks.values().all(|t| t.finished); if drained { self.emit(None)?; - if let Some(out) = self.queue.pop_front() { - self.emitted_epoch = self.epoch; + if let Some(out) = self.pop() { return Poll::Ready(Ok(Some(out))); } // SI emission rides media frames, so a snapshot that arrived behind the @@ -924,6 +937,22 @@ impl Export { Poll::Pending } + /// Return the next queued frame, committing what it carries to the output boundary. + fn pop(&mut self) -> Option { + let (out, tally) = self.queue.pop_front()?; + self.emitted_epoch = self.epoch; + for (pcr, discontinuity) in tally.pcrs { + if discontinuity { + self.liveness.discontinuity(); + } + self.liveness.written_pcr(pcr); + } + for pid in tally.units { + self.liveness.delivered(pid, 1); + } + Some(out) + } + /// The trailing SI frame for end of stream: every entry's current sections, /// re-emitted once. Duplicates of already-emitted sections are harmless /// repetitions; `None` when there are no tables or no program was ever built. @@ -1346,7 +1375,8 @@ impl Export { } self.pending.clear(); self.keyframes.clear(); - self.span_units.clear(); + self.units.clear(); + self.tally = Tally::default(); self.queue.clear(); self.watermark = None; // The new generation waits for every track again, with a fresh budget: a @@ -1737,7 +1767,7 @@ impl Export { } } if out.len() > written { - *self.span_units.entry(pid).or_default() += 1; + self.units.push((self.pending.len() + written, pid)); } if keyframe { self.keyframes.push(self.pending.len()); @@ -1802,6 +1832,7 @@ impl Export { self.span_counters = None; let bytes = std::mem::take(&mut self.pending); let keyframes = std::mem::take(&mut self.keyframes); + let units = std::mem::take(&mut self.units); let Some(to) = self.low.take() else { return Ok(()) }; let packets = bytes.len() / TsPacket::SIZE; @@ -1864,7 +1895,7 @@ impl Export { }; self.send(&mut payload, &bytes[cut * TsPacket::SIZE..next * TsPacket::SIZE]); self.stuff(index, &mut payload); - self.push(at, payload, &keyframes, cut, next); + self.push(at, payload, &keyframes, &units, cut, next); cut = next; at = boundary; let before = counter_before(&bytes, cut * TsPacket::SIZE, pcr_pid, self.pcr_cc); @@ -1874,13 +1905,10 @@ impl Export { } self.send(&mut payload, &bytes[cut * TsPacket::SIZE..]); - self.push(at, payload, &keyframes, cut, packets); + self.push(at, payload, &keyframes, &units, cut, packets); if let Some(cc) = counter_before(&bytes, bytes.len(), pcr_pid, None) { self.pcr_cc = Some(cc); } - for (pid, units) in std::mem::take(&mut self.span_units) { - self.liveness.delivered(pid, units); - } Ok(()) } @@ -1944,19 +1972,34 @@ impl Export { } /// Queue one output frame, unless it would be empty. `from`..`to` are the packet - /// indices it carries, which decide whether a keyframe begins in it. - fn push(&mut self, timestamp: Timestamp, payload: Vec, keyframes: &[usize], from: usize, to: usize) { + /// indices it carries, which decide whether a keyframe, or which access units, begin in it. + fn push( + &mut self, + timestamp: Timestamp, + payload: Vec, + keyframes: &[usize], + units: &[(usize, u16)], + from: usize, + to: usize, + ) { if payload.is_empty() { return; } let (from, to) = (from * TsPacket::SIZE, to * TsPacket::SIZE); let keyframe = keyframes.iter().any(|&at| at >= from && at < to); - self.queue.push_back(Frame { + let mut tally = std::mem::take(&mut self.tally); + tally.units = units + .iter() + .filter(|&&(at, _)| at >= from && at < to) + .map(|&(_, pid)| pid) + .collect(); + let frame = Frame { timestamp, duration: None, payload: Bytes::from(payload), keyframe, - }); + }; + self.queue.push_back((frame, tally)); } /// The clock packet for grid slot `index`, and record that the slot is served. @@ -1991,11 +2034,11 @@ impl Export { }; self.last_pcr = Some(index); let mut packet = pcr_packet(pcr_pid, ticks, cc)?; - if std::mem::take(&mut self.pcr_discontinuity) { + let discontinuity = std::mem::take(&mut self.pcr_discontinuity); + if discontinuity { packet[5] |= 0x80; - self.liveness.discontinuity(); } - self.liveness.written_pcr((ticks & TS_TIMESTAMP_MASK) * 300); + self.tally.pcrs.push(((ticks & TS_TIMESTAMP_MASK) * 300, discontinuity)); Ok(packet) } diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index f38c7470c7..126c3b6c8f 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -5475,6 +5475,66 @@ async fn resume_after_a_drop() { resume_after(false).await; } +/// The stats count only output that was returned. A frame the muxer refuses fails the export +/// with the span before it queued but never returned, and the resume discards it. +#[tokio::test(start_paused = true)] +async fn export_stats_skip_output_a_failure_discards() { + let origin = crate::source::produce_origin(); + let source = crate::Source::new(origin.consume(), "live"); + let publish = || { + let mut broadcast = origin.publish("live", Default::default()).unwrap(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + (broadcast, catalog) + }; + let units = |stats: Stats| stats.streams.values().map(|row| row.units).sum::() as usize; + + let (mut broadcast, mut catalog) = publish(); + let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); + let ended = source.broadcast().await.unwrap(); + let mut export = Export::new(source.clone()) + .await + .unwrap() + .with_max_age(RECORDING_MAX_AGE); + for ms in (0..200).step_by(20) { + write_aac(&mut track, ms); + } + let mut frames = drain_frames(&mut export).await; + // One byte past what an ADTS header can frame. + track + .write(Frame { + timestamp: Timestamp::from_millis(200).unwrap(), + duration: None, + payload: Bytes::from(vec![0; 8185]), + keyframe: true, + }) + .unwrap(); + let (rest, end) = drain_to_end(&mut export).await; + frames.extend(rest); + assert!(end.is_err(), "an unframeable AAC frame fails the export"); + assert!(pes_count(&frames) < 10, "the span before the failure stayed queued"); + assert_eq!(units(export.stats()), pes_count(&frames)); + + drop((broadcast, catalog, track)); + let (mut broadcast, mut catalog) = publish(); + let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); + source.returned(&ended).await.unwrap(); + export.resume().await.unwrap(); + for ms in (0..200).step_by(20) { + write_aac(&mut track, ms); + } + frames.extend(drain_frames(&mut export).await); + track.finish().unwrap(); + catalog.finish().unwrap(); + let (rest, end) = drain_to_end(&mut export).await; + frames.extend(rest); + end.unwrap(); + assert_eq!( + units(export.stats()), + pes_count(&frames), + "the discarded span never counts" + ); +} + /// Export 25 fps video and two AAC tracks for [`TICKS`] video frames, the video and the first /// audio track stopping after `stop` while the second carries on. Returns the stats sampled /// after `sample` and at the end, and the frames rendered in between. From b245bbed2a17a77cd954a019e0cf80bf71f0efff Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 30 Sep 2026 21:39:04 -0700 Subject: [PATCH 4/5] quest: finish ts-export-liveness Co-Authored-By: Claude Opus 5.5 --- quest/m2/ts-export-liveness.md | 50 ---------------------------------- 1 file changed, 50 deletions(-) delete mode 100644 quest/m2/ts-export-liveness.md diff --git a/quest/m2/ts-export-liveness.md b/quest/m2/ts-export-liveness.md deleted file mode 100644 index f8c2a92b03..0000000000 --- a/quest/m2/ts-export-liveness.md +++ /dev/null @@ -1,50 +0,0 @@ -# [S] moq export ts reports each elementary stream's access units and how long it has been quiet - -## Goal - -`Export::stats` returns `ts::Stats` with the import's per-PID row (#4502) for -every elementary stream the exporter writes: `units`, the access-unit count, -and `quiet`, the time since the last, on the output's own PCR. An operator can tell that one track stalled -at the exporter while PSI and the other PIDs kept flowing. The other fields of -`ts::Stats` stay zero at egress. `moq export ts` logs a line when a row stops -advancing. - -## Plan - -Decided while planning [#1838](https://github.com/moq-dev/moq/issues/1838): - -- **Liveness only, no ETSI counters at egress.** The media-aware lane - regenerates PAT, PMT, PCR and CC, so an egress check grades our muxer, never - the contribution feed. Sync, TEI, CRC, CC, PAT and PMT on bytes the muxer - constructs fail only on a code defect, which the `test/ts` hard checks catch - in CI, and `PAT_error` timed on the output's own PCR never fires on a whole - stall because that clock freezes too. `PCR_accuracy_error` would be - permanently in alarm: the exporter's PCR fails ±500 ns on every file sample - and the downstream groomer's output passes on every one - ([T33](https://github.com/tdrapier-wbd/moq-mpegts-paper/blob/d4c7573f9519a1c9c4882fab2021fbf82616fdcc/lab/test-33-gate2-preparation.md)). - PCR accuracy and wire timing belong to the groomer's monitoring of its own - output, outside this tree. -- **The runtime failure worth catching is a per-track stall.** #3533 stalled - video and primary audio at the exporter while PSI and the other PIDs - continued and the relay kept transmitting; only a per-PID liveness detector - on a subscriber found it - ([T27](https://github.com/tdrapier-wbd/moq-mpegts-paper/blob/d4c7573f9519a1c9c4882fab2021fbf82616fdcc/lab/test-27-liveness-detector.md)). -- **Same type as ingest.** `Export::stats` returns `ts::Stats`, so a - dashboard reads one schema at both edges. Additive on main. -- **No window in-tree.** The raw exporter delivers by group, with 102 - simultaneous media-time gaps over 0.3 s in 39 s on every PID, and thresholds - learned there differ from the groomed wire's for the same content (T27). A - window means something only with its monitoring point named, so the - consumer picks it. - -Implementation: - -- `rs/moq-mux/src/container/ts/export.rs`: count an access unit per PID when - its PES is written, and the quiet time on the output PCR. Fix the stale "TR 101 - 290 flags a gap over 40 ms" comment near `PCR_INTERVAL` (V1.4.1's limit is - 100 ms), and the matching 40 ms defaults and the "inserts no null packets, - and paces PCR once per media frame" sentence in `test/ts/README.md`. -- `rs/moq-cli/src/subscribe.rs` logs a stopped row the way `publish.rs` does. -- Tests in `export_test.rs`: a healthy export advances every row; a track that - stops mid-run grows its quiet time while PSI and the other PIDs continue, the - #3533 shape. From 4ef76eeb0fd5bd4c7ab679849cfbd595b24fe270 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 30 Sep 2026 22:20:20 -0700 Subject: [PATCH 5/5] quest: drop links to the finished ts-export-liveness Co-Authored-By: Claude Opus 5.5 --- quest/m1/ts-stats-module.md | 7 +++---- quest/m2/README.md | 1 - quest/m2/ts-health-stats.md | 1 - 3 files changed, 3 insertions(+), 6 deletions(-) diff --git a/quest/m1/ts-stats-module.md b/quest/m1/ts-stats-module.md index a68a4203fb..7968893eda 100644 --- a/quest/m1/ts-stats-module.md +++ b/quest/m1/ts-stats-module.md @@ -31,10 +31,9 @@ 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 quests still add fields under the old names: - [TS PSI reassembly](/quest/m1/ts-psi-reassembly.md), - [TS import health](/quest/m2/ts-import-health.md) and - [TS export liveness](/quest/m2/ts-export-liveness.md). The rename carries - those fields at merge time. Update the type names in the quests still open + [TS PSI reassembly](/quest/m1/ts-psi-reassembly.md) and + [TS import health](/quest/m2/ts-import-health.md). `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 e9c6f6e9f1..c0a65ae6ec 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -98,7 +98,6 @@ waiting on an upstream release waits in [m4](/quest/m4/README.md). - [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 - [Per-program SI](/quest/m2/ts-program-si.md) - a selected TS program's broadcast carries only its own service's SDT and EIT - [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 export liveness](/quest/m2/ts-export-liveness.md) - `moq export ts` reports each elementary stream's access units and quiet time, catching a per-track stall - [TS health stats](/quest/m2/ts-health-stats.md) - the TS counters ride the stats plumbing beside the media counters - [Teleoperation](/quest/m2/teleop/README.md) - MoQ carries robot video down and control up on one session as a library capability - [Media QA on other engines](/quest/m2/browser-media-qa-engines.md) - the media harness measures a Firefox or WebKit player over the fallback and names what each engine lacks diff --git a/quest/m2/ts-health-stats.md b/quest/m2/ts-health-stats.md index 28e8fc6b9d..17a521065c 100644 --- a/quest/m2/ts-health-stats.md +++ b/quest/m2/ts-health-stats.md @@ -43,4 +43,3 @@ Decided while planning [#1838](https://github.com/moq-dev/moq/issues/1838): - [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 -- [TS export liveness](/quest/m2/ts-export-liveness.md) - the egress rows