From eb7005a432e646212b83c74e537abbf1ff52e04a Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Thu, 22 Oct 2026 20:55:00 -0700 Subject: [PATCH 01/10] chore(quest): claim quest/m1/ts-reliability/program-cap.md Quest-Claim: 8361e91e3dc918e8 From d2a2249ade3acc980f4ad8b020291147c67c3324 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Thu, 22 Oct 2026 21:19:06 -0700 Subject: [PATCH 02/10] fix(mux): start a TS program only once its PMT describes it `ts::Programs` created a broadcast, catalog and importer for every program number the first PAT listed, so a small hostile PAT naming tens of thousands of programs created that many broadcasts. Programs now waits on each PAT entry's PMT PID and starts a program only when its PMT lists an elementary stream, so the state scales with the PMT bytes the input carried. CRC errors on a PSI PID are counted here until an importer reads that PID, so each still counts once. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EeyiEy5FWz3z44DYZVo7vU --- quest/m1/ts-reliability/README.md | 1 - quest/m1/ts-reliability/pat-lifecycle.md | 6 +- quest/m1/ts-reliability/program-cap.md | 23 -- rs/moq-mux/src/container/ts/import.rs | 16 +- rs/moq-mux/src/container/ts/programs.rs | 346 ++++++++++++++++------- rs/moq-mux/src/container/ts/psi.rs | 14 +- 6 files changed, 264 insertions(+), 142 deletions(-) delete mode 100644 quest/m1/ts-reliability/program-cap.md diff --git a/quest/m1/ts-reliability/README.md b/quest/m1/ts-reliability/README.md index afe0a17fe83..ef166badb3e 100644 --- a/quest/m1/ts-reliability/README.md +++ b/quest/m1/ts-reliability/README.md @@ -106,7 +106,6 @@ remain with their current quests. - [Access units](/quest/m1/ts-reliability/pes-access-units.md) - a PES carrying several access units publishes each at its own time - [PAT changes](/quest/m1/ts-reliability/pat-lifecycle.md) - a revised PAT never leaves a broadcast meaning another program - [Stream routes](/quest/m1/ts-reliability/stream-routes.md) - every PID gets a declared route, and what is not carried is counted -- [Program cap](/quest/m1/ts-reliability/program-cap.md) - a PAT listing every program number cannot create a broadcast per program ## Related diff --git a/quest/m1/ts-reliability/pat-lifecycle.md b/quest/m1/ts-reliability/pat-lifecycle.md index 2f0de125d9d..4643709459c 100644 --- a/quest/m1/ts-reliability/pat-lifecycle.md +++ b/quest/m1/ts-reliability/pat-lifecycle.md @@ -32,8 +32,10 @@ Decided 2026-10-20 by the maintainer: If the program returns, it publishes a new epoch of its broadcast, never the retired instance's names for new content. A single-program import whose selected program leaves the PAT ends, as today. -- Open: a PAT that names a program whose PMT never arrives. Decide whether - the import refuses it or waits, and which clock bounds the wait (found in +- Open: a PAT that names a program whose PMT never arrives. `Programs` + creates a program's broadcast only once its PMT lists a stream, so such a + program waits holding only its PAT entry. Decide whether the import refuses + it instead, and which clock bounds the wait (found in [moq-dev/moq#6122](https://github.com/moq-dev/moq/pull/6122)). Rust: `rs/moq-mux` TS `Import::handle_pat` and `Programs`. diff --git a/quest/m1/ts-reliability/program-cap.md b/quest/m1/ts-reliability/program-cap.md deleted file mode 100644 index b8bd4b429e8..00000000000 --- a/quest/m1/ts-reliability/program-cap.md +++ /dev/null @@ -1,23 +0,0 @@ -# [XS] TS import bounds the programs a PAT can create - -## Goal - -`moq import ts --program all` creates a broadcast only for a program the input actually describes, instead of one per program number a hostile PAT names. - -Today `rs/moq-mux/src/container/ts/programs.rs` loops over every `pat.program_numbers()` and creates a broadcast, catalog producer, importer, and consumer for each. A sub-megabyte transport stream can list all 65,536 program numbers across PAT sections, so one input creates tens of thousands of live broadcasts. - -## Plan - -Decided 2026-10-22 from the security audit, with the maintainer: - -- Create a program's broadcast only once its PMT declares a supported stream, so the state scales with the PMT bytes the input carried, not with the PAT's list. `--program all` is the operator's opt-in to every described program, and an operator-selected program list narrows it; the default already refuses multi-program input. -- No new program-count constant or run budget: a cap below the PAT's domain would only refuse valid multiplexes. The session announce budget does not apply here, since it counts what a remote peer announces to us, not broadcasts `Programs` creates locally. -- Test with a synthesized PAT naming far more programs than it carries PMTs for, and assert only the described programs exist. - -Public API: none. Wire: none. - -## Related - -- [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - the TS import contract this belongs to -- [PMT lifecycle](/quest/m1/ts-reliability/pat-lifecycle.md) - the PAT handling beside it -- [Shared TS parser](/quest/m1/ts-reliability/shared-parser.md) - one parser for every program, so per-chunk cost does not grow with the programs this quest admits diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 1308354e8ac..702025cc1a3 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -302,6 +302,13 @@ impl Import { self } + /// Start on a packet boundary another framer already confirmed, so a packet ending the input + /// routes without a successor to confirm it. + pub(super) fn aligned(mut self) -> Self { + self.framer = Framer { synced: true }; + self + } + /// A reservation on this program's timebase, so every stream shifts by its offset. fn reserve(&self) -> crate::catalog::Reserved { self.timebase.reserve() @@ -521,7 +528,8 @@ impl Import { tracing::warn!(pid = pid.as_u16(), "dropped a PSI section with a bad CRC"); } - fn handle_pmt(&mut self, pmt: Pmt) -> anyhow::Result<()> { + /// Apply a whole PMT: route every stream it lists, and end those it no longer lists. + pub(super) fn handle_pmt(&mut self, pmt: Pmt) -> anyhow::Result<()> { // PMTs of several programs may share one PID; only the chosen one maps streams. if self.program.is_some_and(|program| program != pmt.program_number) { return Ok(()); @@ -7050,7 +7058,7 @@ pub(super) mod test { /// One PES on `pid` carrying two whole MP2 frames: enough for the second to confirm the /// first, which is what the legacy path publishes on. - fn mp2_pes(pid: u16, cc: u8, pts: u64, fills: [u8; 2]) -> Vec { + pub(in crate::container::ts) fn mp2_pes(pid: u16, cc: u8, pts: u64, fills: [u8; 2]) -> Vec { let mut payload = mp2_frame(fills[0]); payload.extend_from_slice(&mp2_frame(fills[1])); audio_pes_packet(pid, cc, pts, &payload) @@ -7204,7 +7212,7 @@ pub(super) mod test { } /// A PMT section listing `(stream_type, pid, es_descriptors)`, the clock on the first. - fn pmt_section(program: u16, version: u8, streams: &[(u8, u16, &[u8])]) -> Vec { + pub(in crate::container::ts) fn pmt_section(program: u16, version: u8, streams: &[(u8, u16, &[u8])]) -> Vec { let pcr = streams.first().map_or(0x1fff, |&(_, pid, _)| pid); let mut body = vec![0xe0 | (pcr >> 8) as u8, pcr as u8, 0xf0, 0x00]; for &(stream_type, pid, descriptors) in streams { @@ -7226,7 +7234,7 @@ pub(super) mod test { section } - const MP2: u8 = StreamType::Mpeg1Audio as u8; + pub(in crate::container::ts) const MP2: u8 = StreamType::Mpeg1Audio as u8; /// A PAT listing fifty programs, which takes two packets, then program 50's PMT (one /// MP2 stream on `0x0061`) and one PES on it. diff --git a/rs/moq-mux/src/container/ts/programs.rs b/rs/moq-mux/src/container/ts/programs.rs index e48dd5780ce..1b614d5c2b1 100644 --- a/rs/moq-mux/src/container/ts/programs.rs +++ b/rs/moq-mux/src/container/ts/programs.rs @@ -1,11 +1,11 @@ //! Every program of a multiplex, each imported as its own broadcast. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap, HashSet}; use anyhow::Context; use mpeg2ts::ts::{Pid, TsPacket}; -use super::import::{Framer, PatReader}; +use super::import::{Framer, PatReader, SectionReassembler}; use super::{Ext, Import, psi, stats}; use crate::catalog; @@ -13,21 +13,40 @@ use crate::catalog; /// own clock and catalog. /// /// Program `n` of `event.hang` publishes on `event/n.hang`, keeping the catalog suffix last so -/// format detection still sees it; a name without one gains a plain `/n`. Input ahead of that -/// PAT is dropped, as a single [`Import`] drops media ahead of its PSI. A program a later PAT -/// adds is not published; one it removes ends the import. +/// format detection still sees it; a name without one gains a plain `/n`. A program starts once +/// its PMT lists an elementary stream, so one the PAT names but the input never describes costs +/// nothing. Input ahead of that PMT is dropped, as a single [`Import`] drops media ahead of its +/// PSI. A program a later PAT adds is not published; one it removes ends the import. pub struct Programs { origin: moq_net::origin::Producer, name: moq_net::PathOwned, config: catalog::Config, - /// Input not yet read for the PAT: a trailing partial packet, or what follows the PAT. + /// Input not yet read for the PAT and the waiting PMTs, a trailing partial packet at most. + /// The started programs already hold its first `fed` bytes. pending: Vec, - /// Reads the PAT out of the input. - scan: PatScan, - /// Empty until the PAT arrives. + fed: usize, + framer: Framer, + /// Reads PATs until a program starts, and its importer reads them instead. + pat_reader: PatReader, + /// The first whole PAT listing a program, which seeds every importer. + pat: Option, + /// The programs that PAT lists and no PMT has started yet, by their PMT PID. + waiting: HashMap, + /// PSI sections dropped for a bad CRC while no importer read their PID, keyed by PID. + crc_errors: BTreeMap, + /// Empty until a PMT starts a program. programs: Vec, } +/// The programs waiting for their PMT on one PID. +#[derive(Default)] +struct Waiting { + sections: SectionReassembler, + programs: HashSet, + /// Whether a started program's importer reads this PID too, and counts its errors. + read: bool, +} + /// One program's broadcast. struct Program { /// The PAT's program number. @@ -69,66 +88,147 @@ impl Programs { name: name.as_path().to_owned(), config, pending: Vec::new(), - scan: PatScan::default(), + fed: 0, + framer: Framer::default(), + pat_reader: PatReader::default(), + pat: None, + waiting: HashMap::new(), + crc_errors: BTreeMap::new(), programs: Vec::new(), } } /// Demux a chunk of the multiplex. A trailing partial packet is retained for the next call. pub fn decode(&mut self, data: &[u8]) -> anyhow::Result<()> { - if !self.programs.is_empty() { - return self.feed(data); + if self.pat.is_some() && self.waiting.is_empty() { + // Every program has started, and holds the partial packet `pending` kept. + self.pending.clear(); + return feed(&mut self.programs, data); } self.pending.extend_from_slice(data); - let Some(pat) = self.scan.scan(&mut self.pending) else { - return Ok(()); + let mut off = 0; + while let Some(at) = self.framer.next(&self.pending, &mut off) { + let pkt: [u8; TsPacket::SIZE] = self.pending[at..at + TsPacket::SIZE].try_into().unwrap(); + let started = self.psi(&pkt)?; + if !started.is_empty() { + // The programs already running read through this packet; the new ones start after it. + feed(&mut self.programs, &self.pending[self.fed..off])?; + self.fed = off; + self.programs.extend(started); + } + } + feed(&mut self.programs, &self.pending[self.fed..])?; + self.pending.drain(..off); + self.fed = self.pending.len(); + Ok(()) + } + + /// Read one packet for the PAT or a waiting PMT, returning the programs it starts. + fn psi(&mut self, pkt: &[u8; TsPacket::SIZE]) -> anyhow::Result> { + let pid = (u16::from(pkt[1] & 0x1f) << 8) | u16::from(pkt[2]); + if pid == Pid::PAT { + if self.programs.is_empty() { + let crc_error = self.crc_errors.entry(Pid::PAT).or_default(); + if let Some(pat) = self.pat_reader.push(pkt, crc_error) + && self.pat.is_none() + { + self.wait(pat); + } + } + return Ok(Vec::new()); + } + let Some((pid, waiting)) = Pid::new(pid) + .ok() + .and_then(|pid| Some((pid, self.waiting.get_mut(&pid)?))) + else { + return Ok(Vec::new()); }; - for program in pat.program_numbers() { - let name = program_broadcast(self.name.as_str(), program); - let mut broadcast = self - .origin - .create_broadcast(&name) - .with_context(|| format!("failed to create broadcast {name}"))?; - // TS carries undecoded elementary streams verbatim, so each catalog carries the - // `mpegts` extension, and owns its broadcast's catalog tracks. - let config = self - .config - .clone() - .with_catalog(catalog::hang::Catalog::::default()); - let catalog = catalog::Producer::new(&mut broadcast, config)?; - let mut import = Import::new(broadcast.clone(), catalog.reserve()).with_program(program); - // Seeded with the scanner's PAT rather than replaying its packets, so every section - // reaches the importer and nothing the scanner counted is counted again. - import.handle_pat(&pat)?; - // Each program is its own publisher instance for this run, so a restart - // replaces it rather than resuming into it. - let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); - let listed = broadcast.consume().track(hang::Catalog::DEFAULT_NAME)?; - self.programs.push(Program { - number: program, - broadcast, - import, - catalog, - pending: Some((route, listed)), - }); + let mut sections = Vec::new(); + waiting.sections.push(pkt, &mut sections); + let mut described = Vec::new(); + for section in sections { + // Other tables may share the PID. + if section.first() != Some(&psi::PMT_TABLE_ID) { + continue; + } + if !psi::crc_ok(§ion) { + if !waiting.read { + *self.crc_errors.entry(pid.as_u16()).or_default() += 1; + tracing::warn!(pid = pid.as_u16(), "dropped a PSI section with a bad CRC"); + } + continue; + } + if let Some(pmt) = psi::Pmt::parse(§ion) + && !pmt.streams.is_empty() + && waiting.programs.remove(&pmt.program_number) + { + described.push(pmt); + } + } + // Set only now: the importers start after this packet, so they miss its later sections. + waiting.read |= !described.is_empty(); + if waiting.programs.is_empty() { + self.waiting.remove(&pid); } - let pending = std::mem::take(&mut self.pending); - self.feed(&pending) + described.into_iter().map(|pmt| self.start(pmt)).collect() } - fn feed(&mut self, data: &[u8]) -> anyhow::Result<()> { - for program in &mut self.programs { - program.import.decode(data)?; - program.announce_if_listed()?; + /// Wait for the PMT of each program `pat` lists. + fn wait(&mut self, pat: psi::Pat) { + let mut listed = HashSet::new(); + for entry in &pat.programs { + // An importer takes a program's first entry, so a repeated number waits there alone. + if entry.program_number != 0 && listed.insert(entry.program_number) { + let waiting = self.waiting.entry(entry.pmt_pid).or_default(); + waiting.programs.insert(entry.program_number); + } } - Ok(()) + if !listed.is_empty() { + self.pat = Some(pat); + } + } + + /// Create the broadcast of the program `pmt` describes, its importer seeded with the PAT and + /// that PMT rather than replaying their packets, so nothing counted here is counted again. + fn start(&self, pmt: psi::Pmt) -> anyhow::Result { + let pat = self.pat.as_ref().expect("a PMT waits on its PAT"); + let number = pmt.program_number; + let name = program_broadcast(self.name.as_str(), number); + let mut broadcast = self + .origin + .create_broadcast(&name) + .with_context(|| format!("failed to create broadcast {name}"))?; + // TS carries undecoded elementary streams verbatim, so each catalog carries the + // `mpegts` extension, and owns its broadcast's catalog tracks. + let config = self + .config + .clone() + .with_catalog(catalog::hang::Catalog::::default()); + let catalog = catalog::Producer::new(&mut broadcast, config)?; + let mut import = Import::new(broadcast.clone(), catalog.reserve()) + .with_program(number) + .aligned(); + import.handle_pat(pat)?; + import.handle_pmt(pmt)?; + // Each program is its own publisher instance for this run, so a restart + // replaces it rather than resuming into it. + let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + let listed = broadcast.consume().track(hang::Catalog::DEFAULT_NAME)?; + Ok(Program { + number, + broadcast, + import, + catalog, + pending: Some((route, listed)), + }) } /// 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. Every importer also - /// grades every packet of the multiplex, so each TR 101 290 error counts once too. + /// Every importer reads the PAT, and programs sharing a PMT PID read the same PMT sections, + /// each from when it started. The earliest importer read whatever a later one did, so a + /// dropped section counts once however many read it, and so does each TR 101 290 error. + /// Sections dropped before any importer read their PID are counted here. pub fn stats(&self) -> stats::Snapshot { let mut stats = stats::Snapshot::default(); let mut crc_errors = BTreeMap::::new(); @@ -141,9 +241,7 @@ impl Programs { } 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(); + stats.crc_error = crc_errors.values().chain(self.crc_errors.values()).sum(); errors.report(&mut stats); stats } @@ -154,7 +252,7 @@ impl Programs { /// until its media is stale; dropping one closes that program. /// Fails when the input ended before its PAT listed a program or any program declared a track. pub fn finish(&mut self) -> anyhow::Result> { - anyhow::ensure!(!self.programs.is_empty(), "input ended before its PAT listed a program"); + anyhow::ensure!(self.pat.is_some(), "input ended before its PAT listed a program"); let mut announced = false; let mut finished = Vec::with_capacity(self.programs.len()); for program in &mut self.programs { @@ -176,6 +274,15 @@ impl Programs { } } +/// Hand `data` to every program, announcing each whose catalog has gone out. +fn feed(programs: &mut [Program], data: &[u8]) -> anyhow::Result<()> { + for program in programs { + program.import.decode(data)?; + program.announce_if_listed()?; + } + Ok(()) +} + /// The broadcast one program of `name` publishes on: `event.hang` becomes `event/2.hang`. fn program_broadcast(name: &str, program: u16) -> String { match catalog::CatalogFormat::detect(name) { @@ -187,39 +294,6 @@ fn program_broadcast(name: &str, program: u16) -> String { } } -/// Finds the first whole PAT listing a program, before any importer exists to read one. -#[derive(Default)] -struct PatScan { - framer: Framer, - pat: PatReader, - /// PAT sections dropped for a bad CRC before the first whole PAT. - crc_error: u64, -} - -impl PatScan { - /// Read `pending` up to the first whole PAT listing a program, returning it with `pending` - /// trimmed to start just after the packet that completed it. Otherwise only a trailing - /// partial packet is kept: the reader holds a partial PAT itself. - fn scan(&mut self, pending: &mut Vec) -> Option { - let mut off = 0; - let mut found = None; - while let Some(at) = self.framer.next(pending, &mut off) { - let pkt: &[u8; TsPacket::SIZE] = pending[at..at + TsPacket::SIZE].try_into().unwrap(); - if (u16::from(pkt[1] & 0x1f) << 8 | u16::from(pkt[2])) != Pid::PAT { - continue; - } - if let Some(pat) = self.pat.push(pkt, &mut self.crc_error) - && !pat.program_numbers().is_empty() - { - found = Some(pat); - break; - } - } - pending.drain(..off); - found - } -} - #[cfg(test)] mod test { use std::time::Duration; @@ -228,7 +302,7 @@ mod test { use crate::catalog::hang::Container; use crate::container::Consumer; use crate::container::ts::import::test::{ - corrupt, fifty_programs, pat_section, psi_packets, two_programs, two_section_pat, + MP2, corrupt, fifty_programs, mp2_pes, pat_section, pmt_section, psi_packets, two_programs, two_section_pat, }; #[tokio::test(start_paused = true)] @@ -258,9 +332,22 @@ mod test { assert_eq!(program_broadcast("event", 1), "event/1"); } - /// The PAT in a whole buffer, as one [`PatScan`] reads it. + /// The programs of the PAT a whole buffer starts waiting on. fn pat_programs(data: &[u8]) -> Option> { - Some(PatScan::default().scan(&mut data.to_vec())?.program_numbers()) + let mut programs = Programs::new( + crate::source::produce_origin(), + "event.hang", + catalog::Config::default(), + ); + programs.decode(data).unwrap(); + Some( + programs + .pat? + .programs + .iter() + .map(|entry| entry.program_number) + .collect(), + ) } #[test] @@ -274,10 +361,10 @@ mod test { assert_eq!(pat_programs(&data[..100]), None); } - /// A PAT too big for one packet is read whole, and each importer is handed all of it, so + /// A PAT too big for one packet is read whole, and its importer is handed all of it, so /// the program listed in its second packet publishes. #[tokio::test] - async fn a_pat_spanning_two_packets_starts_every_program() { + async fn a_pat_spanning_two_packets_is_read_whole() { let origin = crate::source::produce_origin(); let mut programs = Programs::new(origin.clone(), "event.hang", catalog::Config::default()); let input = fifty_programs(); @@ -285,7 +372,7 @@ mod test { programs.decode(chunk).unwrap(); } let _finished = programs.finish().unwrap(); - assert_eq!(programs.programs.len(), 50); + assert_eq!(programs.programs.len(), 1, "only program 50 has a PMT"); let consumer = origin.consume().request_broadcast("event/50.hang", None).await.unwrap(); let catalog = hang::catalog::Catalog::<()>::subscribe(&consumer) @@ -298,8 +385,8 @@ mod test { assert_eq!(catalog.audio.renditions.len(), 1, "program 50 publishes its stream"); } - /// A corrupt PAT starts nothing and is counted; the good repetition after it starts the - /// program without counting again. + /// A corrupt PAT is counted and ignored; the good repetition after it is read without + /// counting again. #[test] fn a_corrupt_pat_starts_no_program() { let origin = crate::source::produce_origin(); @@ -311,16 +398,36 @@ mod test { let mut data = psi_packets(0, &mut cc, &[], &corrupt(pat_section(&[(1, 0x0100)]))); data.extend_from_slice(&null); programs.decode(&data).unwrap(); - assert!(programs.programs.is_empty()); + assert!(programs.pat.is_none()); assert_eq!(programs.stats().crc_error, 1); let mut data = psi_packets(0, &mut cc, &[], &pat_section(&[(1, 0x0100)])); data.extend_from_slice(&null); programs.decode(&data).unwrap(); - assert_eq!(programs.programs.len(), 1); + assert!(programs.pat.is_some()); assert_eq!(programs.stats().crc_error, 1); } + /// A corrupt PMT counts once whether it arrives before any importer reads its PID, while + /// one importer reads it beside a waiting program, or while both of them do. + #[test] + fn a_corrupt_pmt_counts_once_whoever_reads_it() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut null = vec![0x47, 0x1f, 0xff, 0x10]; + null.resize(188, 0xff); + let pmt = |program| pmt_section(program, 0, &[(MP2, 0x0060 + program, &[])]); + let mut data = psi_packets(0, &mut 0, &[], &pat_section(&[(1, 0x0100), (2, 0x0100)])); + let mut cc = 0; + for section in [corrupt(pmt(1)), pmt(1), corrupt(pmt(2)), pmt(2), corrupt(pmt(1))] { + data.extend(psi_packets(0x0100, &mut cc, &[], §ion)); + } + data.extend_from_slice(&null); + programs.decode(&data).unwrap(); + assert_eq!(programs.programs.len(), 2); + assert_eq!(programs.stats().crc_error, 3); + } + /// How many audio renditions the catalog `path` publishes lists. async fn renditions(origin: &moq_net::origin::Producer, path: &str) -> usize { let consumer = origin.consume().request_broadcast(path, None).await.unwrap(); @@ -347,6 +454,45 @@ mod test { assert_eq!(renditions(&origin, "event/2.hang").await, 1); } + /// A PAT naming a thousand programs starts only the two whose PMTs the input carries, even + /// where an undescribed program shares a described one's PMT PID. + #[tokio::test] + async fn only_programs_a_pmt_describes_publish() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin.clone(), "event.hang", catalog::Config::default()); + // Pairs of programs share a PMT PID, so 6 waits on the PID that starts 7. + let pmt_pid = |program: u16| 0x0100 + program / 2; + let mut cc = 0; + let mut data = Vec::new(); + for section in 0..4u8 { + let entries: Vec = (1..=250) + .map(|n| u16::from(section) * 250 + n) + .flat_map(|n| { + let pid = pmt_pid(n); + [(n >> 8) as u8, n as u8, 0xe0 | (pid >> 8) as u8, pid as u8] + }) + .collect(); + data.extend(psi_packets( + 0, + &mut cc, + &[], + &psi::section(0x00, 1, 0, section, 3, &entries), + )); + } + for (program, pid, fills) in [(7u16, 0x0061, [0xAA, 0xBB]), (900, 0x0071, [0xCC, 0xDD])] { + let pmt = pmt_section(program, 0, &[(MP2, pid, &[])]); + data.extend(psi_packets(pmt_pid(program), &mut 0, &[], &pmt)); + data.extend(mp2_pes(pid, 0, 90_000, fills)); + } + programs.decode(&data).unwrap(); + let _finished = programs.finish().unwrap(); + + let started: Vec<_> = programs.programs.iter().map(|program| program.number).collect(); + assert_eq!(started, [7, 900]); + assert_eq!(renditions(&origin, "event/7.hang").await, 1); + assert_eq!(renditions(&origin, "event/900.hang").await, 1); + } + /// Each program announces as its own publisher instance for the run, so a restart /// replaces it rather than resuming into it. #[test] @@ -464,7 +610,7 @@ mod test { null.resize(188, 0xff); data.extend_from_slice(&null); programs.decode(&data).unwrap(); - assert_eq!(programs.programs.len(), 1); + assert!(programs.pat.is_some()); assert_eq!(programs.stats().crc_error, 1); } diff --git a/rs/moq-mux/src/container/ts/psi.rs b/rs/moq-mux/src/container/ts/psi.rs index d68a912eafc..5feb2baaa7b 100644 --- a/rs/moq-mux/src/container/ts/psi.rs +++ b/rs/moq-mux/src/container/ts/psi.rs @@ -137,17 +137,6 @@ pub(super) struct Pat { pub programs: Vec, } -impl Pat { - /// The program numbers listed, without the network PID's. - pub fn program_numbers(&self) -> Vec { - self.programs - .iter() - .map(|entry| entry.program_number) - .filter(|&number| number != 0) - .collect() - } -} - /// Collects PAT sections until every section of one version has arrived. /// /// A PAT that needs more than one section is complete only once all of them are in, so a @@ -553,7 +542,8 @@ mod test { assert_eq!(pat.section(&first), None); let whole = pat.section(&second).expect("both sections are in"); assert_eq!(whole.transport_stream_id, 7); - assert_eq!(whole.program_numbers(), [1, 2]); + let numbers: Vec<_> = whole.programs.iter().map(|entry| entry.program_number).collect(); + assert_eq!(numbers, [1, 2]); // A repetition of either section repeats the whole table. assert_eq!(pat.section(&first), Some(whole)); From 148b4eae3369e8b502e714dde5fd94506c320dfc Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 23 Oct 2026 18:32:25 -0700 Subject: [PATCH 03/10] fix(mux): waiting TS programs follow PAT revisions and keep their grading A program waiting for its PMT now follows a later PAT (moved to its new PMT PID, or dropped when no longer listed), starts on the last PMT revision in the packet that completes it, and the first importer carries on the pre-PMT TR 101 290 grading instead of starting blind. Co-Authored-By: Claude Opus 5.5 --- quest/m1/ts-reliability/pat-lifecycle.md | 3 +- rs/moq-mux/src/container/ts/import.rs | 13 +- rs/moq-mux/src/container/ts/programs.rs | 221 ++++++++++++++++++----- 3 files changed, 191 insertions(+), 46 deletions(-) diff --git a/quest/m1/ts-reliability/pat-lifecycle.md b/quest/m1/ts-reliability/pat-lifecycle.md index 7dfe0e42c2a..5b0c202f30a 100644 --- a/quest/m1/ts-reliability/pat-lifecycle.md +++ b/quest/m1/ts-reliability/pat-lifecycle.md @@ -17,7 +17,8 @@ What the code does today: - Under `--program all`, a later PAT that removes one program reaches every per-program `Import`; the removed program's `handle_pat` errors, `Programs::feed` propagates it, and the whole multiplex ingest stops, - including unrelated programs. + including unrelated programs. A program still waiting for its PMT is + instead dropped, or moved to the PMT PID the later PAT names. - A PAT naming another `transport_stream_id`, or a single-program import's PAT naming another program, already refuses the import. diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 943e69ba501..59ff4751e36 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -302,10 +302,12 @@ impl Import { self } - /// Start on a packet boundary another framer already confirmed, so a packet ending the input - /// routes without a successor to confirm it. - pub(super) fn aligned(mut self) -> Self { + /// Carry on another reader's pass over the input: start on the packet boundary its framer + /// confirmed, so a packet ending the input routes without a successor to confirm it, and + /// keep grading where its `health` left off. + pub(super) fn resume(mut self, health: Health) -> Self { self.framer = Framer { synced: true }; + self.health = health; self } @@ -1676,6 +1678,11 @@ impl Import { &self.crc_errors } + /// Whether this importer reads the PMT sections on `pid`, and counts their CRC errors. + pub(super) fn reads_pmt(&self, pid: Pid) -> bool { + self.pmt_sections.contains_key(&pid) + } + /// Abort every track with `err` instead of finishing, so subscribers see the /// real cause rather than [`moq_net::Error::Dropped`]. Buffered PES is discarded. /// Consumes the importer. diff --git a/rs/moq-mux/src/container/ts/programs.rs b/rs/moq-mux/src/container/ts/programs.rs index 1b614d5c2b1..4ee9f872d33 100644 --- a/rs/moq-mux/src/container/ts/programs.rs +++ b/rs/moq-mux/src/container/ts/programs.rs @@ -5,6 +5,7 @@ use std::collections::{BTreeMap, HashMap, HashSet}; use anyhow::Context; use mpeg2ts::ts::{Pid, TsPacket}; +use super::health::Health; use super::import::{Framer, PatReader, SectionReassembler}; use super::{Ext, Import, psi, stats}; use crate::catalog; @@ -16,7 +17,8 @@ use crate::catalog; /// format detection still sees it; a name without one gains a plain `/n`. A program starts once /// its PMT lists an elementary stream, so one the PAT names but the input never describes costs /// nothing. Input ahead of that PMT is dropped, as a single [`Import`] drops media ahead of its -/// PSI. A program a later PAT adds is not published; one it removes ends the import. +/// PSI. A later PAT moves a waiting program to the PMT PID it names and drops one it no longer +/// lists. A program a later PAT adds is not published; removing a started one ends the import. pub struct Programs { origin: moq_net::origin::Producer, name: moq_net::PathOwned, @@ -26,14 +28,17 @@ pub struct Programs { pending: Vec, fed: usize, framer: Framer, - /// Reads PATs until a program starts, and its importer reads them instead. + /// Reads PATs while a program waits. pat_reader: PatReader, - /// The first whole PAT listing a program, which seeds every importer. + /// The latest whole PAT, from the first that lists a program, which seeds every importer. pat: Option, - /// The programs that PAT lists and no PMT has started yet, by their PMT PID. + /// The programs the first PAT lists and no PMT has started yet, by the PMT PID the latest + /// PAT names for them. waiting: HashMap, /// PSI sections dropped for a bad CRC while no importer read their PID, keyed by PID. crc_errors: BTreeMap, + /// Grades the input until the first program starts, whose importer carries it on. + health: Option, /// Empty until a PMT starts a program. programs: Vec, } @@ -43,8 +48,6 @@ pub struct Programs { struct Waiting { sections: SectionReassembler, programs: HashSet, - /// Whether a started program's importer reads this PID too, and counts its errors. - read: bool, } /// One program's broadcast. @@ -94,13 +97,14 @@ impl Programs { pat: None, waiting: HashMap::new(), crc_errors: BTreeMap::new(), + health: Some(Health::default()), programs: Vec::new(), } } /// Demux a chunk of the multiplex. A trailing partial packet is retained for the next call. pub fn decode(&mut self, data: &[u8]) -> anyhow::Result<()> { - if self.pat.is_some() && self.waiting.is_empty() { + if !self.programs.is_empty() && self.waiting.is_empty() { // Every program has started, and holds the partial packet `pending` kept. self.pending.clear(); return feed(&mut self.programs, data); @@ -109,88 +113,139 @@ impl Programs { let mut off = 0; while let Some(at) = self.framer.next(&self.pending, &mut off) { let pkt: [u8; TsPacket::SIZE] = self.pending[at..at + TsPacket::SIZE].try_into().unwrap(); - let started = self.psi(&pkt)?; - if !started.is_empty() { - // The programs already running read through this packet; the new ones start after it. - feed(&mut self.programs, &self.pending[self.fed..off])?; - self.fed = off; - self.programs.extend(started); + if let Some(health) = &mut self.health { + health.routed(&self.pending, at); + // No program clock runs before a PMT. + health.packet(&pkt, None); + } + let described = self.psi(&pkt); + if described.is_empty() { + continue; + } + // The programs already running read through this packet; the new ones start after it. + feed(&mut self.programs, &self.pending[self.fed..off])?; + self.fed = off; + for pmt in described { + // The first importer grades on from here, as one reading from the start would. + // Later ones count a subset of its errors, which `stats` merges. + let health = match self.health.take() { + Some(mut health) => { + health.drained(&self.pending[..off], off); + health + } + None => Health::default(), + }; + let program = self.start(pmt, health)?; + self.programs.push(program); } } + if let Some(health) = &mut self.health { + health.drained(&self.pending, off); + } feed(&mut self.programs, &self.pending[self.fed..])?; self.pending.drain(..off); self.fed = self.pending.len(); Ok(()) } - /// Read one packet for the PAT or a waiting PMT, returning the programs it starts. - fn psi(&mut self, pkt: &[u8; TsPacket::SIZE]) -> anyhow::Result> { + /// Read one packet for the PAT or a waiting PMT, returning the PMTs of the programs it starts. + fn psi(&mut self, pkt: &[u8; TsPacket::SIZE]) -> Vec { let pid = (u16::from(pkt[1] & 0x1f) << 8) | u16::from(pkt[2]); if pid == Pid::PAT { - if self.programs.is_empty() { - let crc_error = self.crc_errors.entry(Pid::PAT).or_default(); - if let Some(pat) = self.pat_reader.push(pkt, crc_error) - && self.pat.is_none() - { - self.wait(pat); - } + // Once a program starts, its importer counts the PAT's errors. + let mut counted = 0; + let crc_error = if self.programs.is_empty() { + self.crc_errors.entry(Pid::PAT).or_default() + } else { + &mut counted + }; + if let Some(pat) = self.pat_reader.push(pkt, crc_error) { + self.wait(pat); } - return Ok(Vec::new()); + return Vec::new(); } let Some((pid, waiting)) = Pid::new(pid) .ok() .and_then(|pid| Some((pid, self.waiting.get_mut(&pid)?))) else { - return Ok(Vec::new()); + return Vec::new(); }; let mut sections = Vec::new(); waiting.sections.push(pkt, &mut sections); - let mut described = Vec::new(); + // Decided before this packet starts anyone: the new importers miss its sections. + let read = self.programs.iter().any(|program| program.import.reads_pmt(pid)); + let mut latest: Vec = Vec::new(); for section in sections { // Other tables may share the PID. if section.first() != Some(&psi::PMT_TABLE_ID) { continue; } if !psi::crc_ok(§ion) { - if !waiting.read { + if !read { *self.crc_errors.entry(pid.as_u16()).or_default() += 1; tracing::warn!(pid = pid.as_u16(), "dropped a PSI section with a bad CRC"); } continue; } + // A revision later in the packet supersedes an earlier one, since the importer + // starts after it. if let Some(pmt) = psi::Pmt::parse(§ion) - && !pmt.streams.is_empty() - && waiting.programs.remove(&pmt.program_number) + && waiting.programs.contains(&pmt.program_number) { - described.push(pmt); + latest.retain(|seen| seen.program_number != pmt.program_number); + latest.push(pmt); } } - // Set only now: the importers start after this packet, so they miss its later sections. - waiting.read |= !described.is_empty(); + latest.retain(|pmt| !pmt.streams.is_empty()); + for pmt in &latest { + waiting.programs.remove(&pmt.program_number); + } if waiting.programs.is_empty() { self.waiting.remove(&pid); } - described.into_iter().map(|pmt| self.start(pmt)).collect() + latest } - /// Wait for the PMT of each program `pat` lists. + /// Wait for the PMT of each program the first whole PAT listing one names. A later PAT moves + /// a waiting program to the PMT PID it names, and drops one it no longer lists. fn wait(&mut self, pat: psi::Pat) { - let mut listed = HashSet::new(); + let mut listed = HashMap::new(); for entry in &pat.programs { // An importer takes a program's first entry, so a repeated number waits there alone. - if entry.program_number != 0 && listed.insert(entry.program_number) { - let waiting = self.waiting.entry(entry.pmt_pid).or_default(); - waiting.programs.insert(entry.program_number); + if entry.program_number != 0 { + listed.entry(entry.program_number).or_insert(entry.pmt_pid); } } - if !listed.is_empty() { - self.pat = Some(pat); + if self.pat.is_none() { + if listed.is_empty() { + return; + } + for (&number, &pid) in &listed { + self.waiting.entry(pid).or_default().programs.insert(number); + } + } else { + let mut moved = Vec::new(); + self.waiting.retain(|pid, waiting| { + waiting.programs.retain(|number| match listed.get(number) { + Some(listed) if listed == pid => true, + Some(&listed) => { + moved.push((*number, listed)); + false + } + None => false, + }); + !waiting.programs.is_empty() + }); + for (number, pid) in moved { + self.waiting.entry(pid).or_default().programs.insert(number); + } } + self.pat = Some(pat); } /// Create the broadcast of the program `pmt` describes, its importer seeded with the PAT and /// that PMT rather than replaying their packets, so nothing counted here is counted again. - fn start(&self, pmt: psi::Pmt) -> anyhow::Result { + fn start(&self, pmt: psi::Pmt, health: Health) -> anyhow::Result { let pat = self.pat.as_ref().expect("a PMT waits on its PAT"); let number = pmt.program_number; let name = program_broadcast(self.name.as_str(), number); @@ -207,7 +262,7 @@ impl Programs { let catalog = catalog::Producer::new(&mut broadcast, config)?; let mut import = Import::new(broadcast.clone(), catalog.reserve()) .with_program(number) - .aligned(); + .resume(health); import.handle_pat(pat)?; import.handle_pmt(pmt)?; // Each program is its own publisher instance for this run, so a restart @@ -228,11 +283,15 @@ impl Programs { /// Every importer reads the PAT, and programs sharing a PMT PID read the same PMT sections, /// each from when it started. The earliest importer read whatever a later one did, so a /// dropped section counts once however many read it, and so does each TR 101 290 error. - /// Sections dropped before any importer read their PID are counted here. + /// Sections dropped before any importer read their PID are counted here, and so is the + /// grading of the input before the first program started. pub fn stats(&self) -> stats::Snapshot { let mut stats = stats::Snapshot::default(); let mut crc_errors = BTreeMap::::new(); let mut errors = super::health::Errors::default(); + if let Some(health) = &self.health { + errors.max(health.errors()); + } for program in &self.programs { stats.streams.extend(program.import.stats().streams); for (&pid, &count) in program.import.crc_errors() { @@ -493,6 +552,84 @@ mod test { assert_eq!(renditions(&origin, "event/900.hang").await, 1); } + /// A later PAT moves a waiting program to its new PMT PID and drops one it no longer lists, + /// so neither the stale PMT nor the dropped program's PMT starts anything. + #[tokio::test] + async fn a_later_pat_revises_the_waiting_programs() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin.clone(), "event.hang", catalog::Config::default()); + let mut cc = 0; + let mut data = psi_packets(0, &mut cc, &[], &pat_section(&[(1, 0x0100), (2, 0x0101)])); + let revised = psi::section(0x00, 1, 1, 0, 0, &[0x00, 0x01, 0xe2, 0x00]); + data.extend(psi_packets(0, &mut cc, &[], &revised)); + data.extend(psi_packets( + 0x0100, + &mut 0, + &[], + &pmt_section(1, 0, &[(MP2, 0x0062, &[])]), + )); + data.extend(psi_packets( + 0x0101, + &mut 0, + &[], + &pmt_section(2, 0, &[(MP2, 0x0071, &[])]), + )); + data.extend(psi_packets( + 0x0200, + &mut 0, + &[], + &pmt_section(1, 1, &[(MP2, 0x0061, &[])]), + )); + data.extend(mp2_pes(0x0061, 0, 90_000, [0xAA, 0xBB])); + programs.decode(&data).unwrap(); + let _finished = programs.finish().unwrap(); + + let started: Vec<_> = programs.programs.iter().map(|program| program.number).collect(); + assert_eq!(started, [1]); + assert_eq!(renditions(&origin, "event/1.hang").await, 1); + } + + /// Of two revisions of a PMT in one packet, the program starts on the later, as its importer + /// only reads the packets after it. + #[tokio::test] + async fn a_program_starts_on_the_last_pmt_in_its_packet() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin.clone(), "event.hang", catalog::Config::default()); + let mut data = psi_packets(0, &mut 0, &[], &pat_section(&[(1, 0x0100)])); + let mut sections = pmt_section(1, 0, &[(MP2, 0x0062, &[])]); + sections.extend(pmt_section(1, 1, &[(MP2, 0x0061, &[])])); + data.extend(psi_packets(0x0100, &mut 0, &[], §ions)); + data.extend(mp2_pes(0x0061, 0, 90_000, [0xAA, 0xBB])); + programs.decode(&data).unwrap(); + let _finished = programs.finish().unwrap(); + assert_eq!(renditions(&origin, "event/1.hang").await, 1); + } + + /// A transport error ahead of every PMT counts once: the scan grades it, and the first + /// importer carries that grading on rather than starting blind. + #[test] + fn a_transport_error_before_the_pmts_counts_once() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut data = psi_packets(0, &mut 0, &[], &pat_section(&[(1, 0x0100), (2, 0x0200)])); + let mut flagged = vec![0x47, 0x9f, 0xff, 0x10]; + flagged.resize(188, 0xff); + data.extend_from_slice(&flagged); + programs.decode(&data).unwrap(); + assert!(programs.programs.is_empty()); + assert_eq!(programs.stats().transport_error, 1); + + let mut data = Vec::new(); + for (program, pid, fills) in [(1u16, 0x0061, [0xAA, 0xBB]), (2, 0x0071, [0xCC, 0xDD])] { + let pmt = pmt_section(program, 0, &[(MP2, pid, &[])]); + data.extend(psi_packets(program << 8, &mut 0, &[], &pmt)); + data.extend(mp2_pes(pid, 0, 90_000, fills)); + } + programs.decode(&data).unwrap(); + assert_eq!(programs.programs.len(), 2); + assert_eq!(programs.stats().transport_error, 1); + } + /// Each program announces as its own publisher instance for the run, so a restart /// replaces it rather than resuming into it. #[test] From deb19895e15caa74bad2a28a81ce0f65367e1684 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 23 Oct 2026 18:52:34 -0700 Subject: [PATCH 04/10] fix(mux): a started TS importer finishes its PID's partial PMT and sees each PAT first The importer a PMT starts now inherits that PID's section reassembler, so a revision the starting packet began is finished rather than lost. Running importers take each PAT before the scan reads on, so CRC ownership of a PMT PID a PAT just moved them to no longer depends on chunk boundaries. `Import::resume` takes the handoff as one `Resume`. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/health.rs | 2 +- rs/moq-mux/src/container/ts/import.rs | 20 +++-- rs/moq-mux/src/container/ts/programs.rs | 113 +++++++++++++++++++----- 3 files changed, 109 insertions(+), 26 deletions(-) diff --git a/rs/moq-mux/src/container/ts/health.rs b/rs/moq-mux/src/container/ts/health.rs index eb6e0fbfad2..30835f9e234 100644 --- a/rs/moq-mux/src/container/ts/health.rs +++ b/rs/moq-mux/src/container/ts/health.rs @@ -402,7 +402,7 @@ impl Sync { /// 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)] +#[derive(Clone, Default)] pub(super) struct Continuity { /// Last continuity_counter seen on a packet with payload, to spot gaps. last_cc: Option, diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 59ff4751e36..a8cf9846f57 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -303,11 +303,11 @@ impl Import { } /// Carry on another reader's pass over the input: start on the packet boundary its framer - /// confirmed, so a packet ending the input routes without a successor to confirm it, and - /// keep grading where its `health` left off. - pub(super) fn resume(mut self, health: Health) -> Self { + /// confirmed, so a packet ending the input routes without a successor to confirm it. + pub(super) fn resume(mut self, from: Resume) -> Self { self.framer = Framer { synced: true }; - self.health = health; + self.health = from.health; + self.pmt_sections.insert(from.pmt, from.sections); self } @@ -2484,6 +2484,16 @@ impl Liveness { } } +/// Where another reader's pass over the input left off, for [`Import::resume`]. +pub(super) struct Resume { + /// Its grading of the input so far. + pub health: Health, + /// The PMT PID that started the importer. + pub pmt: Pid, + /// That PID's reassembler, holding any section the starting packet began. + pub sections: SectionReassembler, +} + /// Locks onto the 188-byte packet grid of a TS byte stream. /// /// 0x47 is the packet sync byte but also occurs freely in payload (TS has no byte @@ -2630,7 +2640,7 @@ pub(super) fn payload(pkt: &[u8; TsPacket::SIZE]) -> Payload<'_> { /// header split, where section_length is not yet known), continuity-counter /// gaps, and adaptation-field discontinuities. Deliberately minimal: just enough to /// recover whole sections verbatim. -#[derive(Default)] +#[derive(Clone, Default)] pub(super) struct SectionReassembler { /// Bytes of the section currently being reassembled. Its 3-byte header (and /// thus section_length) may not all be present yet, so completeness is diff --git a/rs/moq-mux/src/container/ts/programs.rs b/rs/moq-mux/src/container/ts/programs.rs index 4ee9f872d33..d85ea0913bf 100644 --- a/rs/moq-mux/src/container/ts/programs.rs +++ b/rs/moq-mux/src/container/ts/programs.rs @@ -6,7 +6,7 @@ use anyhow::Context; use mpeg2ts::ts::{Pid, TsPacket}; use super::health::Health; -use super::import::{Framer, PatReader, SectionReassembler}; +use super::import::{Framer, PatReader, Resume, SectionReassembler}; use super::{Ext, Import, psi, stats}; use crate::catalog; @@ -50,6 +50,15 @@ struct Waiting { programs: HashSet, } +/// The programs one packet on a PMT PID starts. +struct Described { + pid: Pid, + /// That PID's reassembler after the packet, holding any section it began. + sections: SectionReassembler, + /// The last PMT of each program it starts. + pmts: Vec, +} + /// One program's broadcast. struct Program { /// The PAT's program number. @@ -118,14 +127,20 @@ impl Programs { // No program clock runs before a PMT. health.packet(&pkt, None); } - let described = self.psi(&pkt); - if described.is_empty() { + let pid = (u16::from(pkt[1] & 0x1f) << 8) | u16::from(pkt[2]); + let described = self.psi(pid, &pkt); + // The running importers take each PAT before the scan reads on, so `reads_pmt` + // already knows the PMT PIDs it names. + if described.is_none() && pid != Pid::PAT { continue; } // The programs already running read through this packet; the new ones start after it. feed(&mut self.programs, &self.pending[self.fed..off])?; self.fed = off; - for pmt in described { + let Some(described) = described else { + continue; + }; + for pmt in described.pmts { // The first importer grades on from here, as one reading from the start would. // Later ones count a subset of its errors, which `stats` merges. let health = match self.health.take() { @@ -135,7 +150,12 @@ impl Programs { } None => Health::default(), }; - let program = self.start(pmt, health)?; + let from = Resume { + health, + pmt: described.pid, + sections: described.sections.clone(), + }; + let program = self.start(pmt, from)?; self.programs.push(program); } } @@ -148,9 +168,8 @@ impl Programs { Ok(()) } - /// Read one packet for the PAT or a waiting PMT, returning the PMTs of the programs it starts. - fn psi(&mut self, pkt: &[u8; TsPacket::SIZE]) -> Vec { - let pid = (u16::from(pkt[1] & 0x1f) << 8) | u16::from(pkt[2]); + /// Read one packet on `pid` for the PAT or a waiting PMT, returning the programs it starts. + fn psi(&mut self, pid: u16, pkt: &[u8; TsPacket::SIZE]) -> Option { if pid == Pid::PAT { // Once a program starts, its importer counts the PAT's errors. let mut counted = 0; @@ -162,14 +181,11 @@ impl Programs { if let Some(pat) = self.pat_reader.push(pkt, crc_error) { self.wait(pat); } - return Vec::new(); + return None; } - let Some((pid, waiting)) = Pid::new(pid) + let (pid, waiting) = Pid::new(pid) .ok() - .and_then(|pid| Some((pid, self.waiting.get_mut(&pid)?))) - else { - return Vec::new(); - }; + .and_then(|pid| Some((pid, self.waiting.get_mut(&pid)?)))?; let mut sections = Vec::new(); waiting.sections.push(pkt, &mut sections); // Decided before this packet starts anyone: the new importers miss its sections. @@ -197,13 +213,22 @@ impl Programs { } } latest.retain(|pmt| !pmt.streams.is_empty()); + if latest.is_empty() { + return None; + } for pmt in &latest { waiting.programs.remove(&pmt.program_number); } - if waiting.programs.is_empty() { - self.waiting.remove(&pid); - } - latest + let sections = if waiting.programs.is_empty() { + self.waiting.remove(&pid)?.sections + } else { + waiting.sections.clone() + }; + Some(Described { + pid, + sections, + pmts: latest, + }) } /// Wait for the PMT of each program the first whole PAT listing one names. A later PAT moves @@ -245,7 +270,7 @@ impl Programs { /// Create the broadcast of the program `pmt` describes, its importer seeded with the PAT and /// that PMT rather than replaying their packets, so nothing counted here is counted again. - fn start(&self, pmt: psi::Pmt, health: Health) -> anyhow::Result { + fn start(&self, pmt: psi::Pmt, from: Resume) -> anyhow::Result { let pat = self.pat.as_ref().expect("a PMT waits on its PAT"); let number = pmt.program_number; let name = program_broadcast(self.name.as_str(), number); @@ -262,7 +287,7 @@ impl Programs { let catalog = catalog::Producer::new(&mut broadcast, config)?; let mut import = Import::new(broadcast.clone(), catalog.reserve()) .with_program(number) - .resume(health); + .resume(from); import.handle_pat(pat)?; import.handle_pmt(pmt)?; // Each program is its own publisher instance for this run, so a restart @@ -605,6 +630,54 @@ mod test { assert_eq!(renditions(&origin, "event/1.hang").await, 1); } + /// A PMT revision the starting packet begins is finished by the importer reading on after it. + #[tokio::test] + async fn a_pmt_revision_begun_in_the_starting_packet_is_finished() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin.clone(), "event.hang", catalog::Config::default()); + let mut data = psi_packets(0, &mut 0, &[], &pat_section(&[(1, 0x0100)])); + let mut sections = pmt_section(1, 0, &[(MP2, 0x0062, &[])]); + let mut private = vec![0x80, 200]; + private.resize(202, 0); + sections.extend(pmt_section(1, 1, &[(MP2, 0x0061, &private)])); + let packets = psi_packets(0x0100, &mut 0, &[], §ions); + assert_eq!(packets.len(), 2 * 188, "the revision spans two packets"); + data.extend(packets); + data.extend(mp2_pes(0x0061, 0, 90_000, [0xAA, 0xBB])); + programs.decode(&data).unwrap(); + let _finished = programs.finish().unwrap(); + assert_eq!(renditions(&origin, "event/1.hang").await, 1); + } + + /// A corrupt PMT counts once on a PID a later PAT moves a started program to, even when the + /// PAT and the PMT arrive in one chunk. + #[test] + fn a_corrupt_pmt_after_a_moving_pat_counts_once() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut cc = 0; + let mut data = psi_packets(0, &mut cc, &[], &pat_section(&[(1, 0x0100), (2, 0x0200)])); + data.extend(psi_packets( + 0x0100, + &mut 0, + &[], + &pmt_section(1, 0, &[(MP2, 0x0061, &[])]), + )); + programs.decode(&data).unwrap(); + assert_eq!(programs.programs.len(), 1); + + let moved = [0x00, 0x01, 0xe2, 0x00, 0x00, 0x02, 0xe2, 0x00]; + let mut data = psi_packets(0, &mut cc, &[], &psi::section(0x00, 1, 1, 0, 0, &moved)); + data.extend(psi_packets( + 0x0200, + &mut 0, + &[], + &corrupt(pmt_section(2, 0, &[(MP2, 0x0071, &[])])), + )); + programs.decode(&data).unwrap(); + assert_eq!(programs.stats().crc_error, 1); + } + /// A transport error ahead of every PMT counts once: the scan grades it, and the first /// importer carries that grading on rather than starting blind. #[test] From 53c178eeeb4ad72825d11f90eacb7a462035e047 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 23 Oct 2026 19:04:44 -0700 Subject: [PATCH 05/10] fix(mux): a started TS importer finishes a PAT revision the scan began `Resume` also carries the scan's PAT reader, so a multi-section PAT revision that began before a program's PMT is completed by its importer, which then follows the PMT PID it names. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/import.rs | 5 ++++- rs/moq-mux/src/container/ts/programs.rs | 30 +++++++++++++++++++++++++ rs/moq-mux/src/container/ts/psi.rs | 2 +- 3 files changed, 35 insertions(+), 2 deletions(-) diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index a8cf9846f57..4f11664dc1f 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -307,6 +307,7 @@ impl Import { pub(super) fn resume(mut self, from: Resume) -> Self { self.framer = Framer { synced: true }; self.health = from.health; + self.pat = from.pat; self.pmt_sections.insert(from.pmt, from.sections); self } @@ -2488,6 +2489,8 @@ impl Liveness { pub(super) struct Resume { /// Its grading of the input so far. pub health: Health, + /// Its PAT reader, holding any revision still assembling. + pub pat: PatReader, /// The PMT PID that started the importer. pub pmt: Pid, /// That PID's reassembler, holding any section the starting packet began. @@ -2554,7 +2557,7 @@ impl Framer { /// Reads whole PATs off PID 0: reassembles its sections, drops any whose CRC fails, and /// collects the rest until every section of a version is in. -#[derive(Default)] +#[derive(Clone, Default)] pub(super) struct PatReader { sections: SectionReassembler, table: psi::PatAssembler, diff --git a/rs/moq-mux/src/container/ts/programs.rs b/rs/moq-mux/src/container/ts/programs.rs index d85ea0913bf..a31a3695ad6 100644 --- a/rs/moq-mux/src/container/ts/programs.rs +++ b/rs/moq-mux/src/container/ts/programs.rs @@ -152,6 +152,7 @@ impl Programs { }; let from = Resume { health, + pat: self.pat_reader.clone(), pmt: described.pid, sections: described.sections.clone(), }; @@ -649,6 +650,35 @@ mod test { assert_eq!(renditions(&origin, "event/1.hang").await, 1); } + /// A PAT revision still assembling when a program starts is finished by its importer, which + /// follows the PMT PID it names. + #[tokio::test] + async fn a_pat_revision_begun_before_a_program_starts_is_finished() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin.clone(), "event.hang", catalog::Config::default()); + let mut cc = 0; + let mut data = psi_packets(0, &mut cc, &[], &pat_section(&[(1, 0x0100)])); + let revision = |number, entry: [u8; 4]| psi::section(0x00, 1, 1, number, 1, &entry); + data.extend(psi_packets(0, &mut cc, &[], &revision(0, [0x00, 0x01, 0xe2, 0x00]))); + data.extend(psi_packets( + 0x0100, + &mut 0, + &[], + &pmt_section(1, 0, &[(MP2, 0x0062, &[])]), + )); + data.extend(psi_packets(0, &mut cc, &[], &revision(1, [0x00, 0x02, 0xe3, 0x00]))); + data.extend(psi_packets( + 0x0200, + &mut 0, + &[], + &pmt_section(1, 1, &[(MP2, 0x0061, &[])]), + )); + data.extend(mp2_pes(0x0061, 0, 90_000, [0xAA, 0xBB])); + programs.decode(&data).unwrap(); + let _finished = programs.finish().unwrap(); + assert_eq!(renditions(&origin, "event/1.hang").await, 1); + } + /// A corrupt PMT counts once on a PID a later PAT moves a started program to, even when the /// PAT and the PMT arrive in one chunk. #[test] diff --git a/rs/moq-mux/src/container/ts/psi.rs b/rs/moq-mux/src/container/ts/psi.rs index 5feb2baaa7b..5fd011a8a85 100644 --- a/rs/moq-mux/src/container/ts/psi.rs +++ b/rs/moq-mux/src/container/ts/psi.rs @@ -141,7 +141,7 @@ pub(super) struct Pat { /// /// A PAT that needs more than one section is complete only once all of them are in, so a /// program is never missed for having been listed in a section still on its way. -#[derive(Default)] +#[derive(Clone, Default)] pub(super) struct PatAssembler { /// The version, transport_stream_id, and last_section_number being collected. table: Option<(u8, u16, u8)>, From 538734383a573437d071ba7a90284b1a948ace64 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 05:22:19 -0700 Subject: [PATCH 06/10] perf(mux): bench TS programs a PAT names but no PMT describes `ts_programs` gains an undescribed axis: the PAT also names 1000 programs whose PMT never arrives, in a multi-section PAT. A PAT repetition equal to the one already read no longer rebuilds the waiting map, which was most of that case's cost. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/benches/ts_programs.rs | 108 +++++++++++++++--------- rs/moq-mux/src/container/ts/programs.rs | 4 + 2 files changed, 72 insertions(+), 40 deletions(-) diff --git a/rs/moq-mux/benches/ts_programs.rs b/rs/moq-mux/benches/ts_programs.rs index 52be36723e5..138d738ce9a 100644 --- a/rs/moq-mux/benches/ts_programs.rs +++ b/rs/moq-mux/benches/ts_programs.rs @@ -1,6 +1,7 @@ //! `--program all` TS ingest across program count and per-program bitrate, over one second of //! a generated multiplex. Throughput is the whole input, so a parse that repeats per program -//! shows as a falling rate with program count. +//! shows as a falling rate with program count. A PAT that also names programs no PMT +//! describes shows what a program costs before it starts. use std::time::Duration; @@ -32,12 +33,13 @@ fn crc32(data: &[u8]) -> u32 { crc } -/// A long-form PSI section: `table_id`, the 16-bit extension, version 0, one section. -fn section(table_id: u8, extension: u16, body: &[u8]) -> Vec { +/// A long-form PSI section: `table_id`, the 16-bit extension, version 0, section `number` of +/// `last`. +fn section(table_id: u8, extension: u16, number: u8, last: u8, body: &[u8]) -> Vec { let length = 5 + body.len() + 4; let mut section = vec![table_id, 0xb0 | (length >> 8) as u8, length as u8]; section.extend_from_slice(&extension.to_be_bytes()); - section.extend_from_slice(&[0xc1, 0, 0]); + section.extend_from_slice(&[0xc1, number, last]); section.extend_from_slice(body); section.extend_from_slice(&crc32(§ion).to_be_bytes()); section @@ -51,6 +53,11 @@ fn video_pid(program: u16) -> u16 { 0x0200 + program } +/// The PMT PID a PAT names for an undescribed program, clear of every PID that carries data. +fn undescribed_pid(program: u16) -> u16 { + 0x1000 + program +} + #[derive(Default)] struct Mux { out: Vec, @@ -70,12 +77,14 @@ impl Mux { } fn psi(&mut self, pid: u16, section: &[u8]) { - assert!(section.len() < 184, "one packet per section"); - self.header(pid, true, false); - let end = self.out.len() + 184; - self.out.push(0); - self.out.extend_from_slice(section); - self.out.resize(end, 0xff); + let mut payload = vec![0]; + payload.extend_from_slice(section); + for (i, chunk) in payload.chunks(184).enumerate() { + self.header(pid, i == 0, false); + let end = self.out.len() + 184; + self.out.extend_from_slice(chunk); + self.out.resize(end, 0xff); + } } /// One access unit on `pid` in `packets` packets, its first carrying `pcr` (27 MHz) and a @@ -123,16 +132,27 @@ impl Mux { } /// One second of `programs` H.264 programs at `bitrate` bits per second each, interleaved -/// frame by frame, each program carrying its PCR on its video PID. -fn multiplex(programs: u16, bitrate: u64) -> Vec { +/// frame by frame, each program carrying its PCR on its video PID. The PAT also names +/// `undescribed` programs whose PMT never arrives. +fn multiplex(programs: u16, undescribed: u16, bitrate: u64) -> Vec { let packets = (bitrate / FPS / 8).div_ceil(188) as usize; let mut mux = Mux::default(); - let mut pat = Vec::new(); - for program in 1..=programs { - pat.extend_from_slice(&program.to_be_bytes()); - pat.extend_from_slice(&(0xe000 | pmt_pid(program)).to_be_bytes()); - } - let pat = section(0x00, 1, &pat); + let entries: Vec = (1..=programs) + .map(|program| (program, pmt_pid(program))) + .chain((programs + 1..=programs + undescribed).map(|program| (program, undescribed_pid(program)))) + .flat_map(|(program, pid)| { + let [a, b] = program.to_be_bytes(); + let [c, d] = (0xe000 | pid).to_be_bytes(); + [a, b, c, d] + }) + .collect(); + // A section holds at most 1021 bytes, so a long PAT spans several. + let chunks: Vec<_> = entries.chunks(250 * 4).collect(); + let last = (chunks.len() - 1) as u8; + let pat: Vec<_> = (0..) + .zip(&chunks) + .map(|(number, chunk)| section(0x00, 1, number, last, chunk)) + .collect(); let pmts: Vec<_> = (1..=programs) .map(|program| { let video = video_pid(program); @@ -140,12 +160,14 @@ fn multiplex(programs: u16, bitrate: u64) -> Vec { body.extend_from_slice(&[0xf0, 0x00, 0x1b]); body.extend_from_slice(&(0xe000 | video).to_be_bytes()); body.extend_from_slice(&[0xf0, 0x00]); - section(0x02, program, &body) + section(0x02, program, 0, 0, &body) }) .collect(); for frame in 0..FPS { if frame % PSI_EVERY == 0 { - mux.psi(0, &pat); + for pat in &pat { + mux.psi(0, pat); + } for (program, pmt) in (1..=programs).zip(&pmts) { mux.psi(pmt_pid(program), pmt); } @@ -180,26 +202,32 @@ fn programs(c: &mut Criterion) { group.warm_up_time(Duration::from_millis(300)); group.sample_size(10); for programs in [1u16, 4, 16] { - for mbps in [1u64, 4] { - let input = multiplex(programs, mbps * 1_000_000); - let (stats, _driver) = ingest(programs, &input); - assert!( - stats.streams.len() == usize::from(programs) && stats.streams.values().all(|s| s.units == FPS), - "every program delivers every frame" - ); - group.throughput(Throughput::Bytes(input.len() as u64)); - group.bench_function(BenchmarkId::new(format!("{programs}"), format!("{mbps}mbps")), |b| { - b.iter_batched( - setup, - |(mut programs, driver)| { - for chunk in input.chunks(188 * 7) { - programs.decode(chunk).unwrap(); - } - (programs, driver) - }, - BatchSize::PerIteration, - ) - }); + for undescribed in [0u16, 1000] { + for mbps in [1u64, 4] { + let input = multiplex(programs, undescribed, mbps * 1_000_000); + let (stats, _driver) = ingest(programs, &input); + assert!( + stats.streams.len() == usize::from(programs) && stats.streams.values().all(|s| s.units == FPS), + "every program delivers every frame" + ); + group.throughput(Throughput::Bytes(input.len() as u64)); + let name = match undescribed { + 0 => format!("{programs}"), + _ => format!("{programs}+{undescribed}"), + }; + group.bench_function(BenchmarkId::new(name, format!("{mbps}mbps")), |b| { + b.iter_batched( + setup, + |(mut programs, driver)| { + for chunk in input.chunks(188 * 7) { + programs.decode(chunk).unwrap(); + } + (programs, driver) + }, + BatchSize::PerIteration, + ) + }); + } } } group.finish(); diff --git a/rs/moq-mux/src/container/ts/programs.rs b/rs/moq-mux/src/container/ts/programs.rs index f6f0ed89bec..1e312863e13 100644 --- a/rs/moq-mux/src/container/ts/programs.rs +++ b/rs/moq-mux/src/container/ts/programs.rs @@ -170,6 +170,10 @@ impl Programs { /// Wait for the PMT of each program the first whole PAT listing one names. A later PAT moves /// a waiting program to the PMT PID it names, and drops one it no longer lists. fn wait(&mut self, pat: &psi::Pat) { + // The PAT repeats several times a second, mostly unchanged. + if self.pat.as_ref() == Some(pat) { + return; + } let mut listed = HashMap::new(); for entry in &pat.programs { // An importer takes a program's first entry, so a repeated number waits there alone. From 4e1b4ab6225de7b08f3bb4ff52fbd9f8286e2e86 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 05:27:47 -0700 Subject: [PATCH 07/10] fix(mux): count a waiting PMT's corrupt section by who saw it begin A program a PAT moves onto a waiting PID mid-section never sees that section begin, so its CRC failure is the scan's to count. The scan now judges each section by whether a program read the PID when it began. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/import.rs | 5 +++ rs/moq-mux/src/container/ts/programs.rs | 41 ++++++++++++++++++++++--- 2 files changed, 42 insertions(+), 4 deletions(-) diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 0f2f248d472..3f991083704 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -2660,6 +2660,11 @@ pub(super) struct SectionReassembler { } impl SectionReassembler { + /// Whether a section is partway assembled. + pub(super) fn assembling(&self) -> bool { + !self.acc.is_empty() + } + /// Consume one 188-byte TS packet, appending every completed section to `out`. pub(super) fn push(&mut self, pkt: &[u8], out: &mut Vec>) { let mut completed = Vec::new(); diff --git a/rs/moq-mux/src/container/ts/programs.rs b/rs/moq-mux/src/container/ts/programs.rs index 1e312863e13..4ad54712afd 100644 --- a/rs/moq-mux/src/container/ts/programs.rs +++ b/rs/moq-mux/src/container/ts/programs.rs @@ -48,6 +48,9 @@ pub struct Programs { struct Waiting { sections: SectionReassembler, programs: HashSet, + /// Whether a program read this PID when the section still assembling began, and so counts + /// it if it is corrupt. + began_read: bool, } /// One program's broadcast. @@ -219,17 +222,26 @@ impl Programs { return Ok(()); }; let before = waiting.sections.clone(); + let read = !self.readers[usize::from(packet.pid)].is_empty(); + let carried = waiting.sections.assembling(); + let began = waiting.began_read; let mut sections = Vec::new(); waiting.sections.push(packet.pkt, &mut sections); + // The section still assembling began in this packet, unless the one before carried on. + if waiting.sections.assembling() && (!carried || !sections.is_empty()) { + waiting.began_read = read; + } + // Corrupt sections no program read from their start. let mut corrupt = 0; let mut latest: Vec = Vec::new(); - for section in sections { + for (index, section) in sections.into_iter().enumerate() { // Other tables may share the PID. if section.first() != Some(&psi::PMT_TABLE_ID) { continue; } if !psi::crc_ok(§ion) { - corrupt += 1; + let read = if index == 0 && carried { began } else { read }; + corrupt += u64::from(!read); continue; } if let Some(pmt) = psi::Pmt::parse(§ion) @@ -241,8 +253,7 @@ impl Programs { } latest.retain(|pmt| !pmt.streams.is_empty()); if latest.is_empty() { - // A program reading this PID counts its errors. - if corrupt > 0 && self.readers[usize::from(packet.pid)].is_empty() { + if corrupt > 0 { *self.crc_errors.entry(packet.pid).or_default() += corrupt; tracing::warn!(pid = packet.pid, "dropped a PSI section with a bad CRC"); } @@ -642,6 +653,28 @@ mod test { programs.decode(&data).unwrap(); assert_eq!(programs.programs.len(), 1); assert_eq!(programs.stats().crc_error, 1); + + // The same move landing while the corrupt section is partway in: the program it moved + // never saw that section begin, so the scan counts it. + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut cc = 0; + let mut data = psi_packets(0, &mut cc, &[], &pat_section(&[(1, 0x0100), (2, 0x0200)])); + data.extend(psi_packets(0x0100, &mut 0, &[], &pmt(1))); + let mut private = vec![0x80, 200]; + private.resize(202, 0); + let split = psi_packets( + 0x0200, + &mut 0, + &[], + &corrupt(pmt_section(2, 0, &[(MP2, 0x0071, &private)])), + ); + assert_eq!(split.len(), 2 * 188, "the corrupt PMT spans two packets"); + data.extend_from_slice(&split[..188]); + data.extend(psi_packets(0, &mut cc, &[], &psi::section(0x00, 1, 1, 0, 0, &moved))); + data.extend_from_slice(&split[188..]); + programs.decode(&data).unwrap(); + assert_eq!(programs.stats().crc_error, 1); } /// Each program announces as its own publisher instance for the run, so a restart From a5be42ac172cf8b66d2f06dbfde903e96a06646c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 05:42:17 -0700 Subject: [PATCH 08/10] fix(mux): hand waiting programs the SI and CRC counts they missed A program that starts on a later PMT now gets the SI sections that completed while it waited. A corrupt PMT section is counted by Programs only when no started program reassembles PMTs on that PID, since a PCR or media reader does not. A section that begins in the packet starting a program is that program's to count. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/import.rs | 13 +++++ rs/moq-mux/src/container/ts/programs.rs | 76 ++++++++++++++++++++++++- rs/moq-mux/src/container/ts/si.rs | 19 +++++++ 3 files changed, 106 insertions(+), 2 deletions(-) diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 3f991083704..293840022ec 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -304,6 +304,19 @@ impl Import { self } + /// Whether this importer reassembles the PMT sections on `pid`, and counts their CRC errors. + pub(super) fn reads_pmt(&self, pid: u16) -> bool { + Pid::new(pid).is_ok_and(|pid| self.pmt_sections.contains_key(&pid)) + } + + /// Fold in an SI section on `pid` the parser completed before this importer started. + pub(super) fn si_section(&mut self, pid: u16, section: bytes::Bytes) -> anyhow::Result<()> { + if !self.supports_mpegts { + return Ok(()); + } + self.si.section(pid, section) + } + /// A reservation on this program's timebase, so every stream shifts by its offset. fn reserve(&self) -> crate::catalog::Reserved { self.timebase.reserve() diff --git a/rs/moq-mux/src/container/ts/programs.rs b/rs/moq-mux/src/container/ts/programs.rs index 4ad54712afd..33546aa25a4 100644 --- a/rs/moq-mux/src/container/ts/programs.rs +++ b/rs/moq-mux/src/container/ts/programs.rs @@ -3,11 +3,12 @@ use std::collections::{BTreeMap, HashMap, HashSet}; use anyhow::Context; +use bytes::Bytes; use mpeg2ts::ts::Pid; use super::demux::{Demux, Packet}; use super::import::SectionReassembler; -use super::{Ext, Import, catalog::SI_PIDS, psi, stats}; +use super::{Ext, Import, catalog::SI_PIDS, psi, si, stats}; use crate::catalog; /// Imports every program the first PAT lists, each as its own broadcast on `origin`, with its @@ -34,6 +35,10 @@ pub struct Programs { waiting: HashMap, /// PMT sections dropped for a bad CRC on a PID no program read, by PID. crc_errors: BTreeMap, + /// The latest SI section in each slot while a program waits, handed to each program that + /// starts. Keyed by PID, table_id, sub-table and section number, it holds no more than one + /// program's own capture would. + si: BTreeMap, /// Empty until a PMT starts a program. programs: Vec, /// The programs reading each PID they claim, indexed by PID, in program order. The PAT and @@ -101,6 +106,7 @@ impl Programs { pat: None, waiting: HashMap::new(), crc_errors: BTreeMap::new(), + si: BTreeMap::new(), programs: Vec::new(), readers: vec![Vec::new(); 1 << 13], touched: Vec::new(), @@ -121,6 +127,11 @@ impl Programs { if let Some(pat) = &packet.pat { self.wait(pat); } else if !self.waiting.is_empty() { + if SI_PIDS.contains(&packet.pid) { + for section in packet.sections { + self.si.insert(si::slot(packet.pid, section), section.clone()); + } + } self.describe(&packet)?; } // Every program reads the PAT, and the SI sections the parser reassembled. @@ -159,6 +170,9 @@ impl Programs { } } demux.drain(); + if self.waiting.is_empty() { + self.si = BTreeMap::new(); + } // Only a program that read something has anything to flush or announce, so a call // costs what it carried rather than one pass over every program. for index in self.touched.drain(..) { @@ -222,7 +236,9 @@ impl Programs { return Ok(()); }; let before = waiting.sections.clone(); - let read = !self.readers[usize::from(packet.pid)].is_empty(); + let read = self.readers[usize::from(packet.pid)] + .iter() + .any(|&index| self.programs[index].import.reads_pmt(packet.pid)); let carried = waiting.sections.assembling(); let began = waiting.began_read; let mut sections = Vec::new(); @@ -268,6 +284,10 @@ impl Programs { for pmt in latest { self.start(pmt.program_number, packet.pid, before.clone())?; } + // The programs just started read this packet, so they read any section it began. + if let Some(waiting) = self.waiting.get_mut(&packet.pid) { + waiting.began_read = true; + } Ok(()) } @@ -292,6 +312,9 @@ impl Programs { .with_program(number) .with_pmt(pmt_pid, sections); import.handle_pat(pat)?; + for (slot, section) in &self.si { + import.si_section(slot.pid, section.clone())?; + } // Each program is its own publisher instance for this run, so a restart // replaces it rather than resuming into it. let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); @@ -675,6 +698,55 @@ mod test { data.extend_from_slice(&split[188..]); programs.decode(&data).unwrap(); assert_eq!(programs.stats().crc_error, 1); + + // A corrupt section begun in the packet that starts a program sharing the PID: the new + // program reads that packet, so it alone counts the section. + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut data = psi_packets(0, &mut 0, &[], &pat_section(&[(1, 0x0100), (2, 0x0100)])); + let mut sections = pmt(1); + sections.extend(corrupt(pmt_section(2, 0, &[(MP2, 0x0071, &private)]))); + data.extend(psi_packets(0x0100, &mut 0, &[], §ions)); + programs.decode(&data).unwrap(); + assert_eq!(programs.programs.len(), 1); + assert_eq!(programs.stats().crc_error, 1); + + // A waiting program's PMT PID that a started program reads only as media: no program + // reassembles PMTs there, so the scan counts the corrupt section. + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut data = psi_packets(0, &mut 0, &[], &pat_section(&[(1, 0x0100), (2, 0x0061)])); + data.extend(psi_packets(0x0100, &mut 0, &[], &pmt(1))); + data.extend(psi_packets(0x0061, &mut 0, &[], &corrupt(pmt(2)))); + programs.decode(&data).unwrap(); + assert_eq!(programs.programs.len(), 1); + assert_eq!(programs.stats().crc_error, 1); + } + + /// Service information sent while a program waits for its PMT reaches it once it starts. + #[test] + fn si_before_the_pmt_reaches_the_program() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut data = psi_packets(0, &mut 0, &[], &pat_section(&[(1, 0x0100)])); + // A TDT: short-form, five bytes of UTC. + data.extend(psi_packets( + 0x0014, + &mut 0, + &[], + &[0x70, 0x70, 0x05, 0xc0, 0x79, 0x12, 0x45, 0x00], + )); + data.extend(psi_packets( + 0x0100, + &mut 0, + &[], + &pmt_section(1, 0, &[(MP2, 0x0061, &[])]), + )); + data.extend(mp2_pes(0x0061, 0, 90_000, [0xAA, 0xBB])); + programs.decode(&data).unwrap(); + let _finished = programs.finish().unwrap(); + let si = programs.programs[0].catalog.snapshot().ext.mpegts.si.clone(); + assert!(si.contains_key(&0x0014), "the TDT is captured: {si:?}"); } /// Each program announces as its own publisher instance for the run, so a restart diff --git a/rs/moq-mux/src/container/ts/si.rs b/rs/moq-mux/src/container/ts/si.rs index ac60dc02f96..ff6a9411bf7 100644 --- a/rs/moq-mux/src/container/ts/si.rs +++ b/rs/moq-mux/src/container/ts/si.rs @@ -83,6 +83,25 @@ pub(super) fn identity(section: &[u8]) -> Identity { } } +/// Where a section on `pid` lands in a capture: a later section in the same slot replaces it. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub(super) struct Slot { + pub pid: u16, + table_id: u8, + identity: Identity, + number: u8, +} + +/// The slot `section` on `pid` fills. +pub(super) fn slot(pid: u16, section: &[u8]) -> Slot { + Slot { + pid, + table_id: section.first().copied().unwrap_or_default(), + identity: identity(section), + number: if long_form(section) { section[6] } else { 0 }, + } +} + /// The table-specific bytes that extend the generic identity, for the two table /// families whose `table_id_extension` alone is ambiguous (#2842). /// From 8fafeab69e5f569090cfe860c93b0aad9c2a1437 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 05:57:13 -0700 Subject: [PATCH 09/10] fix(mux): drop the waiting programs' SI replay The cache had no bound while any program waited, and it kept only the latest bytes per section, so a next-version section could replace the current one. Doing it right needs Capture's generations and a program clock that Programs does not have. A program that starts late now gets its SI on the next repetition, as an import joining mid-stream does. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/import.rs | 8 ----- rs/moq-mux/src/container/ts/programs.rs | 47 ++----------------------- rs/moq-mux/src/container/ts/si.rs | 19 ---------- 3 files changed, 2 insertions(+), 72 deletions(-) diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 293840022ec..c168cad0638 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -309,14 +309,6 @@ impl Import { Pid::new(pid).is_ok_and(|pid| self.pmt_sections.contains_key(&pid)) } - /// Fold in an SI section on `pid` the parser completed before this importer started. - pub(super) fn si_section(&mut self, pid: u16, section: bytes::Bytes) -> anyhow::Result<()> { - if !self.supports_mpegts { - return Ok(()); - } - self.si.section(pid, section) - } - /// A reservation on this program's timebase, so every stream shifts by its offset. fn reserve(&self) -> crate::catalog::Reserved { self.timebase.reserve() diff --git a/rs/moq-mux/src/container/ts/programs.rs b/rs/moq-mux/src/container/ts/programs.rs index 33546aa25a4..707c08089df 100644 --- a/rs/moq-mux/src/container/ts/programs.rs +++ b/rs/moq-mux/src/container/ts/programs.rs @@ -3,12 +3,11 @@ use std::collections::{BTreeMap, HashMap, HashSet}; use anyhow::Context; -use bytes::Bytes; use mpeg2ts::ts::Pid; use super::demux::{Demux, Packet}; use super::import::SectionReassembler; -use super::{Ext, Import, catalog::SI_PIDS, psi, si, stats}; +use super::{Ext, Import, catalog::SI_PIDS, psi, stats}; use crate::catalog; /// Imports every program the first PAT lists, each as its own broadcast on `origin`, with its @@ -18,7 +17,7 @@ use crate::catalog; /// format detection still sees it; a name without one gains a plain `/n`. A program starts once /// its PMT lists an elementary stream, so one the PAT names but the input never describes costs /// nothing. Input ahead of that PMT is dropped, as a single [`Import`] drops media ahead of its -/// PSI. A later PAT moves a waiting program to the PMT PID it names and drops one it no longer +/// PSI; its SI arrives with the next repetition, as for an import joining mid-stream. A later PAT moves a waiting program to the PMT PID it names and drops one it no longer /// lists. A program a later PAT adds is not published; removing a started one ends the import. /// /// The multiplex is parsed once: framing, continuity, the PAT and the SI sections are shared, @@ -35,10 +34,6 @@ pub struct Programs { waiting: HashMap, /// PMT sections dropped for a bad CRC on a PID no program read, by PID. crc_errors: BTreeMap, - /// The latest SI section in each slot while a program waits, handed to each program that - /// starts. Keyed by PID, table_id, sub-table and section number, it holds no more than one - /// program's own capture would. - si: BTreeMap, /// Empty until a PMT starts a program. programs: Vec, /// The programs reading each PID they claim, indexed by PID, in program order. The PAT and @@ -106,7 +101,6 @@ impl Programs { pat: None, waiting: HashMap::new(), crc_errors: BTreeMap::new(), - si: BTreeMap::new(), programs: Vec::new(), readers: vec![Vec::new(); 1 << 13], touched: Vec::new(), @@ -127,11 +121,6 @@ impl Programs { if let Some(pat) = &packet.pat { self.wait(pat); } else if !self.waiting.is_empty() { - if SI_PIDS.contains(&packet.pid) { - for section in packet.sections { - self.si.insert(si::slot(packet.pid, section), section.clone()); - } - } self.describe(&packet)?; } // Every program reads the PAT, and the SI sections the parser reassembled. @@ -170,9 +159,6 @@ impl Programs { } } demux.drain(); - if self.waiting.is_empty() { - self.si = BTreeMap::new(); - } // Only a program that read something has anything to flush or announce, so a call // costs what it carried rather than one pass over every program. for index in self.touched.drain(..) { @@ -312,9 +298,6 @@ impl Programs { .with_program(number) .with_pmt(pmt_pid, sections); import.handle_pat(pat)?; - for (slot, section) in &self.si { - import.si_section(slot.pid, section.clone())?; - } // Each program is its own publisher instance for this run, so a restart // replaces it rather than resuming into it. let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); @@ -723,32 +706,6 @@ mod test { assert_eq!(programs.stats().crc_error, 1); } - /// Service information sent while a program waits for its PMT reaches it once it starts. - #[test] - fn si_before_the_pmt_reaches_the_program() { - let origin = crate::source::produce_origin(); - let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); - let mut data = psi_packets(0, &mut 0, &[], &pat_section(&[(1, 0x0100)])); - // A TDT: short-form, five bytes of UTC. - data.extend(psi_packets( - 0x0014, - &mut 0, - &[], - &[0x70, 0x70, 0x05, 0xc0, 0x79, 0x12, 0x45, 0x00], - )); - data.extend(psi_packets( - 0x0100, - &mut 0, - &[], - &pmt_section(1, 0, &[(MP2, 0x0061, &[])]), - )); - data.extend(mp2_pes(0x0061, 0, 90_000, [0xAA, 0xBB])); - programs.decode(&data).unwrap(); - let _finished = programs.finish().unwrap(); - let si = programs.programs[0].catalog.snapshot().ext.mpegts.si.clone(); - assert!(si.contains_key(&0x0014), "the TDT is captured: {si:?}"); - } - /// Each program announces as its own publisher instance for the run, so a restart /// replaces it rather than resuming into it. #[test] diff --git a/rs/moq-mux/src/container/ts/si.rs b/rs/moq-mux/src/container/ts/si.rs index ff6a9411bf7..ac60dc02f96 100644 --- a/rs/moq-mux/src/container/ts/si.rs +++ b/rs/moq-mux/src/container/ts/si.rs @@ -83,25 +83,6 @@ pub(super) fn identity(section: &[u8]) -> Identity { } } -/// Where a section on `pid` lands in a capture: a later section in the same slot replaces it. -#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] -pub(super) struct Slot { - pub pid: u16, - table_id: u8, - identity: Identity, - number: u8, -} - -/// The slot `section` on `pid` fills. -pub(super) fn slot(pid: u16, section: &[u8]) -> Slot { - Slot { - pid, - table_id: section.first().copied().unwrap_or_default(), - identity: identity(section), - number: if long_form(section) { section[6] } else { 0 }, - } -} - /// The table-specific bytes that extend the generic identity, for the two table /// families whose `table_id_extension` alone is ambiguous (#2842). /// From e0e689b3d535795badd66dd1dc69171b183a10e3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 06:10:38 -0700 Subject: [PATCH 10/10] fix(mux): take PMT CRC ownership from what the reassembler did Programs inferred whether a partial section survived a packet from whether the packet completed any section. A section start that abandons the partial without completing one kept the old owner, so a replacement the moved program read from its start counted twice. The reassembler now reports whether the latest packet resumed a section and whether it began the partial left assembling. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/import.rs | 23 ++++++++++++++++++++--- rs/moq-mux/src/container/ts/programs.rs | 25 +++++++++++++++++++++---- 2 files changed, 41 insertions(+), 7 deletions(-) diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index c168cad0638..1e8a8ec0ef8 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -2662,12 +2662,21 @@ pub(super) struct SectionReassembler { span: Option, /// Shared with the PES path: a broken chain drops the partial either way. continuity: Continuity, + /// Whether the first section the latest packet completed began in an earlier packet. + resumed: bool, + /// Whether the partial section left assembling began in the latest packet. + fresh: bool, } impl SectionReassembler { - /// Whether a section is partway assembled. - pub(super) fn assembling(&self) -> bool { - !self.acc.is_empty() + /// Whether the first section the latest packet completed began in an earlier packet. + pub(super) fn resumed(&self) -> bool { + self.resumed + } + + /// Whether the latest packet began the partial section left assembling. + pub(super) fn fresh(&self) -> bool { + self.fresh && !self.acc.is_empty() } /// Consume one 188-byte TS packet, appending every completed section to `out`. @@ -2679,6 +2688,8 @@ impl SectionReassembler { fn push_at(&mut self, pkt: &[u8], source: Option<(u64, &Placement)>, out: &mut Vec<(Option, Vec)>) { let pkt: &[u8; 188] = pkt.try_into().expect("section packet must be 188 bytes"); + self.resumed = false; + self.fresh = false; match self.continuity.observe(pkt) { Continuation::Duplicate => return, // A packet flagged corrupt takes its own payload down with the partial: this is @@ -2719,7 +2730,9 @@ impl SectionReassembler { // progress (stream just joined, or a reset dropped the partial) these // bytes are an orphaned fragment, so skip straight to the pointer_field. self.acc.extend_from_slice(&payload[1..1 + ptr]); + let completed = out.len(); self.drain(out, source, end - (payload.len() - 1 - ptr) as u64); + self.resumed = out.len() > completed; } // The pointer_field is a hard section boundary: drop any leftover partial // and start the section it points to. @@ -2727,6 +2740,7 @@ impl SectionReassembler { self.acc.extend_from_slice(&payload[1 + ptr..]); self.span = source.and_then(|(_, placement)| placement.span(end - self.acc.len() as u64)); self.drain(out, source, end); + self.fresh = true; } else if !self.acc.is_empty() { // Continuation of the section in progress. A non-PUSI packet with nothing // in progress is unaligned (no pointer_field to resync on), so it is @@ -2734,7 +2748,10 @@ impl SectionReassembler { // discontinuity, or corrupt pointer dropped the partial, rather than // resuming on stray bytes that merely look like a section. self.acc.extend_from_slice(payload); + let completed = out.len(); self.drain(out, source, end); + self.resumed = out.len() > completed; + self.fresh = self.resumed; } } diff --git a/rs/moq-mux/src/container/ts/programs.rs b/rs/moq-mux/src/container/ts/programs.rs index 707c08089df..d6afb0e286f 100644 --- a/rs/moq-mux/src/container/ts/programs.rs +++ b/rs/moq-mux/src/container/ts/programs.rs @@ -225,12 +225,11 @@ impl Programs { let read = self.readers[usize::from(packet.pid)] .iter() .any(|&index| self.programs[index].import.reads_pmt(packet.pid)); - let carried = waiting.sections.assembling(); let began = waiting.began_read; let mut sections = Vec::new(); waiting.sections.push(packet.pkt, &mut sections); - // The section still assembling began in this packet, unless the one before carried on. - if waiting.sections.assembling() && (!carried || !sections.is_empty()) { + let resumed = waiting.sections.resumed(); + if waiting.sections.fresh() { waiting.began_read = read; } // Corrupt sections no program read from their start. @@ -242,7 +241,7 @@ impl Programs { continue; } if !psi::crc_ok(§ion) { - let read = if index == 0 && carried { began } else { read }; + let read = if index == 0 && resumed { began } else { read }; corrupt += u64::from(!read); continue; } @@ -682,6 +681,24 @@ mod test { programs.decode(&data).unwrap(); assert_eq!(programs.stats().crc_error, 1); + // After that move, a section start abandons the partial for another split corrupt + // section, which the moved program reads from its start, so it alone counts it. + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut cc = 0; + let mut data = psi_packets(0, &mut cc, &[], &pat_section(&[(1, 0x0100), (2, 0x0200)])); + data.extend(psi_packets(0x0100, &mut 0, &[], &pmt(1))); + data.extend_from_slice(&split[..188]); + data.extend(psi_packets(0, &mut cc, &[], &psi::section(0x00, 1, 1, 0, 0, &moved))); + data.extend(psi_packets( + 0x0200, + &mut 1, + &[], + &corrupt(pmt_section(2, 0, &[(MP2, 0x0071, &private)])), + )); + programs.decode(&data).unwrap(); + assert_eq!(programs.stats().crc_error, 1); + // A corrupt section begun in the packet that starts a program sharing the PID: the new // program reads that packet, so it alone counts the section. let origin = crate::source::produce_origin();