From 1d9a55859478d4d27bf9e490bce6a14ceee5a015 Mon Sep 17 00:00:00 2001 From: t0ms Date: Wed, 30 Sep 2026 12:18:23 +0100 Subject: [PATCH 1/3] feat(moq-mux): per-program SI for a selected TS program An import with a program selected (`moq import ts --program `, and each broadcast of `--program all`) now carries SI for that service alone. Other services' EIT actual is dropped, and the SDT actual is rebuilt as one section listing only the selected service, with a fresh CRC. Network-wide tables pass through, and an import without a selection is unchanged. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + doc/bin/cli.md | 3 + quest/m1/ts-psi-reassembly.md | 5 +- quest/m2/README.md | 1 - quest/m2/ts-program-si.md | 52 ----- rs/moq-mux/Cargo.toml | 1 + rs/moq-mux/src/container/ts/export_test.rs | 232 ++++++++++++++++++++ rs/moq-mux/src/container/ts/import.rs | 13 +- rs/moq-mux/src/container/ts/si.rs | 239 +++++++++++++++------ 9 files changed, 417 insertions(+), 130 deletions(-) delete mode 100644 quest/m2/ts-program-si.md diff --git a/Cargo.lock b/Cargo.lock index b397cdaf8f..eb190e1a5d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4459,6 +4459,7 @@ dependencies = [ "anyhow", "base64 0.23.1", "bytes", + "crc", "futures", "h264-parser", "hang", diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 65c66fa6a8..a9bb1fab4a 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -87,6 +87,9 @@ the first PAT lists as its own broadcast, with its own clock and catalog, keepin the catalog suffix last: `--broadcast event.hang` publishes `event/1.hang`, `event/2.hang`, and so on. `export ts` writes one program per broadcast. `import srt` takes the same `--program`. +A selected program's SI describes that service alone: its SDT lists only the +selected service, and other services' EIT is dropped. Network-wide tables (NIT, +BAT, TDT/TOT, and the SDT and EIT of other transport streams) pass through. ```bash moq --connect https://relay.example.com/anon --broadcast event.hang import ts --program all < mux.ts diff --git a/quest/m1/ts-psi-reassembly.md b/quest/m1/ts-psi-reassembly.md index d5fa4556ed..ee0b00e375 100644 --- a/quest/m1/ts-psi-reassembly.md +++ b/quest/m1/ts-psi-reassembly.md @@ -32,9 +32,8 @@ Settled decisions: elementary-stream loop with descriptors) are parsed in moq-mux. PES is routed by the PIDs of the parsed PMT, and its header (PTS/DTS) is also parsed in moq-mux. -- Each reassembled section's CRC-32/MPEG-2 is checked with the `crc` crate. - [Per-program SI](/quest/m2/ts-program-si.md) needs the same dependency, and - whichever change lands first adds it. A section with a bad CRC is dropped and +- Each reassembled section's CRC-32/MPEG-2 is checked with the `crc` crate, + which moq-mux already depends on. A section with a bad CRC is dropped and the last good table stays in force, so the import continues; it no longer aborts as the `mpeg2ts` reader does. The section is refused whole, never half-applied, so fail-loud holds at section granularity. A malformed diff --git a/quest/m2/README.md b/quest/m2/README.md index 95f0845640..1f8ac1ad06 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -75,7 +75,6 @@ upstream release waits in [m4](/quest/m4/README.md). - [Send buffer pools](/quest/m2/quic-buffer-pool.md) - whether pooled send buffers beat Bytes in the stream send path - [AF_XDP UDP path](/quest/m2/af-xdp.md) - the kernel-bypass verdict on today's virtio hosts that gates DPDK - [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 diff --git a/quest/m2/ts-program-si.md b/quest/m2/ts-program-si.md deleted file mode 100644 index 168022f0a1..0000000000 --- a/quest/m2/ts-program-si.md +++ /dev/null @@ -1,52 +0,0 @@ -# [M] Per-program SI - -## Goal - -When an MPEG-TS import selects a program (`ts::Import::with_program`, so -`moq import ts --program ` and each broadcast of `--program all`), that -program's broadcast carries SI describing only its own service. Other services' -EIT sub-tables are dropped, and the SDT for this transport stream is rewritten -to list only the selected service. Network-wide tables (NIT, BAT, TDT/TOT, SDT -other, EIT other) pass through unchanged. An import without a selection does -no filtering by program and keeps today's capture behavior. - -## Plan - -Settled decisions: - -- Filter at import, in `si::Capture`, before a generation commits, so a - program's SI tracks never carry another service's bytes and exporters stay - unchanged. Filtering at export was rejected: every subscriber would still - receive the whole multiplex's schedule. -- Filter only under an explicit selection. A single-program input whose SDT - lists extra services is left alone, because default capture keeps the - sections it accepts verbatim. -- EIT actual (`table_id` 0x4E and 0x50..=0x5F): keep a sub-table only when its - `table_id_extension` (the service_id) is the selected program_number. Whole - sub-tables are dropped; nothing is rewritten. EIT other (0x4F and - 0x60..=0x6F) describes other transport streams and passes through. -- SDT actual (`table_id` 0x42 on PID 0x11): its one sub-table lists every - service, so once a generation is complete, rebuild it as a single section - (`section_number` and `last_section_number` 0) holding the selected service's - loop entry, wherever it sat in the source's sections. Keep the header fields - (transport_stream_id, version_number, original_network_id), recompute - `section_length`, and write a fresh CRC-32/MPEG-2 using the `crc` crate, which - is already in `Cargo.lock`, as a direct moq-mux dependency rather than a - hand-written table. [TS PSI reassembly](/quest/m1/ts-psi-reassembly.md) - needs it too, and whichever change lands first adds it. -- When the SDT has no entry for the selected service, carry no SDT actual rather - than fabricating a table the source never gave for this service. If a later - version drops the service, the next snapshot retires the SDT captured before - it, and its catalog entry goes too if nothing else is left on it. A later - version that lists the service again is captured normally. -- A `(PID, table_id)` whose every sub-table is filtered away gets no catalog - `mpegts.si` entry or track. -- The same change updates the program-selection paragraph in `doc/bin/cli.md` - and the `si.rs` module docs; no new page is needed. - -Test with a synthetic two-service multiplex whose SDT lists both services -across two sections, with EIT present/following for each. Each selected import -carries one SDT entry with a valid CRC and only its own EIT; an unselected -import keeps the sections verbatim; a selection missing from the SDT carries -none; an SDT version that drops the selected service retires the earlier one; -and the re-exported TS still parses. diff --git a/rs/moq-mux/Cargo.toml b/rs/moq-mux/Cargo.toml index 5828a2b46a..128a3b498a 100644 --- a/rs/moq-mux/Cargo.toml +++ b/rs/moq-mux/Cargo.toml @@ -16,6 +16,7 @@ categories = ["multimedia", "network-programming", "web-programming"] anyhow = { workspace = true } base64 = { workspace = true } bytes = { workspace = true } +crc = "3.4" h264-parser = { version = "0.4.2" } hang = { workspace = true } kio = { workspace = true } diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index b6b4eef985..d7eceb3838 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -2892,6 +2892,238 @@ async fn eit_now_next_and_schedule_are_captured() { ); } +/// [`si_rig`] importing only `program`. +fn si_rig_selecting(program: u16) -> SiRig { + let SiRig { + import, + catalog, + consumer, + } = si_rig(); + SiRig { + import: import.with_program(program), + catalog, + consumer, + } +} + +/// An SDT service loop entry for `service`, running, with a service_descriptor naming it. +fn sdt_service_entry(service: u16, name: &[u8]) -> Vec { + let mut descriptor = vec![0x48, 0, 0x01, 1, b'P', name.len() as u8]; + descriptor.extend_from_slice(name); + descriptor[1] = (descriptor.len() - 2) as u8; + let mut entry = service.to_be_bytes().to_vec(); + entry.push(0xfd); + entry.extend_from_slice(&[0x80, descriptor.len() as u8]); + entry.extend_from_slice(&descriptor); + entry +} + +/// SDT actual section `number` of `last` for TSID 1 on ONID 2, listing `services`. +fn sdt_actual(version: u8, number: u8, last: u8, services: &[(u16, &[u8])]) -> Vec { + let mut body = vec![0x00, 0x02, 0xff]; + for &(service, name) in services { + body.extend(sdt_service_entry(service, name)); + } + make_long_section(0x42, 1, version, number, last, &body) +} + +/// EIT present/following actual section `number` of 1 for `service` on TSID 1, ONID 2. +fn eit_pf(service: u16, number: u8) -> Bytes { + let body = [0x00, 0x01, 0x00, 0x02, 0x01, 0x4E, service as u8, number]; + Bytes::from(make_long_section(0x4E, service, 0, number, 1, &body)) +} + +fn nit() -> Bytes { + Bytes::from(make_long_section(0x40, 1, 0, 0, 0, &[0xbb; 4])) +} + +/// SI for services 1 and 2: an SDT listing service 1 in section 0 and service 2 in +/// section 1, EIT present/following for each, and a NIT. +fn two_service_si() -> Vec { + let mut out = si_packet(0x0011, &sdt_actual(0, 0, 1, &[(1, b"One")])); + out.extend(si_packet_cc(0x0011, &sdt_actual(0, 1, 1, &[(2, b"Two")]), 1)); + for (cc, section) in [eit_pf(1, 0), eit_pf(1, 1), eit_pf(2, 0), eit_pf(2, 1)] + .iter() + .enumerate() + { + out.extend(si_packet_cc(0x0012, section, cc as u8)); + } + out.extend(si_packet(0x0010, &nit())); + out +} + +/// `section` is one whole SDT actual for TSID 1 on ONID 2 at `version`, listing only +/// `service` named `name`, under a valid CRC. +fn assert_sdt_lists_only(section: &[u8], version: u8, service: u16, name: &[u8]) { + let crc = crc::Crc::::new(&crc::CRC_32_MPEG_2); + assert_eq!(crc.checksum(section), 0, "a valid CRC-32/MPEG-2"); + let section_length = (usize::from(section[1] & 0x0f) << 8) | usize::from(section[2]); + assert_eq!(section.len(), 3 + section_length, "section_length covers the section"); + assert_eq!(section[0], 0x42, "SDT actual"); + assert_eq!( + u16::from_be_bytes([section[3], section[4]]), + 1, + "transport_stream_id kept" + ); + assert_eq!((section[5] >> 1) & 0x1f, version, "version_number kept"); + assert_eq!((section[6], section[7]), (0, 0), "section 0 of 0"); + assert_eq!( + u16::from_be_bytes([section[8], section[9]]), + 2, + "original_network_id kept" + ); + assert_eq!( + §ion[11..section.len() - 4], + &sdt_service_entry(service, name)[..], + "the service loop holds the selected service alone" + ); +} + +/// A selected program's SI describes its own service alone: one SDT actual section +/// listing it, wherever it sat in the source, and only its own EIT. Network-wide +/// tables pass through. +#[tokio::test(start_paused = true)] +async fn a_selected_program_carries_only_its_own_si() { + for (program, name) in [(1u16, &b"One"[..]), (2, b"Two")] { + let mut input = crate::container::ts::import::test::two_programs(); + input.extend(two_service_si()); + let mut rig = si_rig_selecting(program); + rig.import.decode(&BytesMut::from(&input[..])).unwrap(); + rig.import.finish().unwrap(); + + let si = rig.catalog.snapshot().ext.mpegts.si.clone(); + let sdt = read_si_sections(&rig.consumer, &si[&0x0011][&0x42].track).await; + assert_eq!(sdt.len(), 1, "program {program}: one SDT actual section"); + assert_sdt_lists_only(&sdt[0], 0, program, name); + assert_eq!( + read_si_sections(&rig.consumer, &si[&0x0012][&0x4E].track).await, + vec![eit_pf(program, 0), eit_pf(program, 1)], + "program {program}: only its own EIT" + ); + assert_eq!( + read_si_sections(&rig.consumer, &si[&0x0010][&0x40].track).await, + vec![nit()], + "the NIT passes through" + ); + } +} + +/// Without a selection nothing is filtered by program: every section is kept verbatim. +#[tokio::test(start_paused = true)] +async fn an_unselected_import_keeps_every_service() { + let mut rig = si_rig(); + rig.import.decode(&BytesMut::from(&two_service_si()[..])).unwrap(); + rig.import.finish().unwrap(); + + let si = rig.catalog.snapshot().ext.mpegts.si.clone(); + assert_eq!( + read_si_sections(&rig.consumer, &si[&0x0011][&0x42].track).await, + vec![ + Bytes::from(sdt_actual(0, 0, 1, &[(1, b"One")])), + Bytes::from(sdt_actual(0, 1, 1, &[(2, b"Two")])), + ], + "the SDT verbatim" + ); + assert_eq!( + read_si_sections(&rig.consumer, &si[&0x0012][&0x4E].track).await, + vec![eit_pf(1, 0), eit_pf(1, 1), eit_pf(2, 0), eit_pf(2, 1)], + "every service's EIT" + ); +} + +/// A selected service the SDT does not list gets no SDT actual rather than a +/// fabricated one, and a table filtered to nothing gets no catalog entry. +#[tokio::test(start_paused = true)] +async fn a_selection_missing_from_the_sdt_carries_none() { + let mut rig = si_rig_selecting(3); + rig.import.decode(&BytesMut::from(&two_service_si()[..])).unwrap(); + rig.import.finish().unwrap(); + + let si = rig.catalog.snapshot().ext.mpegts.si.clone(); + assert!(si.contains_key(&0x0010), "the NIT was captured (control)"); + assert!(!si.contains_key(&0x0011), "no SDT entry: {si:?}"); + assert!(!si.contains_key(&0x0012), "no EIT entry: {si:?}"); +} + +/// An SDT revision that drops the selected service retires the SDT captured before +/// it, catalog entry and all; a later revision listing it again is captured normally. +/// Driven on the capture itself so each revision cuts without waiting out the debounce. +#[tokio::test(start_paused = true)] +async fn an_sdt_revision_dropping_the_service_retires_it() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); + let mut capture = crate::container::ts::si::Capture::new(broadcast, catalog.clone()); + capture.select(1); + let mut revise = |version: u8, services: &[(u16, &[u8])]| { + capture.section(0x0011, sdt_actual(version, 0, 0, services)).unwrap(); + capture.flush(Timestamp::ZERO, true).unwrap(); + catalog + .snapshot() + .ext + .mpegts + .si + .get(&0x0011) + .map(|tables| tables[&0x42].track.clone()) + }; + + let track = revise(0, &[(1, b"One"), (2, b"Two")]).expect("the SDT is advertised"); + assert_eq!(revise(1, &[(2, b"Two")]), None, "the revision retired the SDT entry"); + assert_eq!( + revise(2, &[(1, b"One"), (2, b"Two")]), + Some(track.clone()), + "listed again, on the same track" + ); + capture.finish(Timestamp::ZERO).unwrap(); + + let groups = read_si_groups(&consumer, &track).await; + assert_eq!(groups.len(), 3, "one snapshot per revision"); + assert_sdt_lists_only(&groups[0][0], 0, 1, b"One"); + assert!(groups[1].is_empty(), "the retiring snapshot carries no SDT"); + assert_sdt_lists_only(&groups[2][0], 2, 1, b"One"); +} + +/// A selected program's reduced SI survives export: the TS parses, and importing it +/// again finds the same single-service SDT and the same EIT. +#[tokio::test(start_paused = true)] +async fn a_selected_programs_si_survives_export() { + let mut input = crate::container::ts::import::test::two_programs(); + input.extend(two_service_si()); + let mut rig = si_rig_selecting(2); + rig.import.decode(&BytesMut::from(&input[..])).unwrap(); + rig.import.finish().unwrap(); + + let ts = drain_with( + Export::with_ts( + crate::source::announced(&rig.consumer), + crate::catalog::CatalogFormat::Hang, + ) + .await + .unwrap(), + ) + .await; + assert_packet_aligned(&ts); + let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref())); + while reader.read_ts_packet().unwrap().is_some() {} + + let mut again = si_rig(); + again.import.decode(&BytesMut::from(ts.as_ref())).unwrap(); + again.import.finish().unwrap(); + let si = again.catalog.snapshot().ext.mpegts.si.clone(); + let sdt = read_si_sections(&again.consumer, &si[&0x0011][&0x42].track).await; + assert_eq!(sdt.len(), 1, "one SDT actual section"); + assert_sdt_lists_only(&sdt[0], 0, 2, b"Two"); + assert_eq!( + read_si_sections(&again.consumer, &si[&0x0012][&0x4E].track).await, + vec![eit_pf(2, 0), eit_pf(2, 1)], + "only program 2's EIT" + ); +} + /// #2842: SDT other sections from two networks that reuse a transport_stream_id /// must not collide. The identity reads original_network_id (bytes 8..10) for /// table_id 0x46, so both survive as separate sub-tables; a revision within one diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 5f4c10ef23..3643d77aee 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -194,9 +194,11 @@ impl Import { /// /// Without this, a PAT that lists more than one program fails the import with /// [`MultipleProgramsError`] rather than merging them onto one clock. A PAT that does not - /// list `program` fails it too. + /// list `program` fails it too. The SI describes only this program's service: other + /// services' EIT actual is dropped and the SDT actual lists this service alone. pub fn with_program(mut self, program: u16) -> Self { self.program = Some(program); + self.si.select(program); self } @@ -914,10 +916,11 @@ impl Import { /// /// Every section is captured, whatever its `table_id`: the SDT PID also carries /// the BAT, the EIT PID carries now/next and the schedule, and a table we don't - /// recognize is exactly as worth preserving as one we do. The store buffers each - /// sub-table to a complete generation and commits it atomically, so a plain - /// repetition (SI repeats every couple of seconds) publishes nothing and a torn - /// multi-section transition is never visible. + /// recognize is exactly as worth preserving as one we do. A selected program is the + /// exception: its SI describes that service alone ([`with_program`](Self::with_program)). + /// The store buffers each sub-table to a complete generation and commits it atomically, + /// so a plain repetition (SI repeats every couple of seconds) publishes nothing and a + /// torn multi-section transition is never visible. fn si_section(&mut self, pid: u16, pkt: &[u8]) -> anyhow::Result<()> { let mut sections = Vec::new(); self.si_sections.entry(pid).or_default().push(pkt, &mut sections); diff --git a/rs/moq-mux/src/container/ts/si.rs b/rs/moq-mux/src/container/ts/si.rs index eb934cc49b..c9ab176e3d 100644 --- a/rs/moq-mux/src/container/ts/si.rs +++ b/rs/moq-mux/src/container/ts/si.rs @@ -15,6 +15,14 @@ //! new snapshot group, so a torn multi-section transition (half old version, half //! new) is unrepresentable on the track. [`Snapshot`] is the export half: it //! reduces a track's frames back into the current section set. +//! +//! An import that selects a program ([`Capture::select`]) carries SI for that +//! service alone. Other services' EIT actual sub-tables are dropped on arrival, and +//! each SDT actual snapshot is rebuilt as one section holding only the selected +//! service's entry, with a fresh CRC. An SDT that does not list the service carries +//! no SDT actual, so a revision that drops it retires the earlier one. Network-wide +//! tables (NIT, BAT, SDT other, EIT other, TDT/TOT) pass through verbatim, as does +//! everything when no program is selected. use std::collections::{BTreeMap, HashMap}; use std::time::Duration; @@ -24,6 +32,13 @@ use moq_net::Timestamp; use super::catalog; +const SDT_PID: u16 = 0x0011; +const SDT_ACTUAL: u8 = 0x42; +const EIT_PID: u16 = 0x0012; + +/// The checksum closing every long-form DVB section. +const CRC: crc::Crc = crc::Crc::::new(&crc::CRC_32_MPEG_2); + /// Coalesce snapshot cuts: a junction revises many sub-tables inside a second /// (every service's now/next rolls on the hour), and one group carrying all of them /// beats a burst of near-identical groups. Measured on the host clock, not the @@ -106,6 +121,37 @@ pub(super) fn split_sections(payload: &Bytes) -> Vec { out } +/// Rebuild an SDT sub-table as one section listing only `service`, wherever its +/// entry sat among `sections`. The header (transport_stream_id, version_number, +/// original_network_id) is kept, the numbering reset to section 0 of 0, and the +/// length and CRC recomputed. `None` when no section lists the service. +fn sdt_service<'a>(sections: impl IntoIterator, service: u16) -> Option> { + for section in sections { + // The service loop runs from the 11-byte header to the CRC. + let end = section.len().saturating_sub(4); + let mut off = 11; + while off + 5 <= end { + let len = 5 + ((usize::from(section[off + 3] & 0x0f) << 8) | usize::from(section[off + 4])); + if off + len > end { + break; + } + if u16::from_be_bytes([section[off], section[off + 1]]) == service { + let mut out = section[..11].to_vec(); + out.extend_from_slice(§ion[off..off + len]); + let section_length = out.len() - 3 + 4; + out[1] = (out[1] & 0xf0) | (section_length >> 8) as u8; + out[2] = section_length as u8; + out[6] = 0; + out[7] = 0; + out.extend_from_slice(&CRC.checksum(&out).to_be_bytes()); + return Some(out); + } + off += len; + } + } + None +} + /// One committed generation of a sub-table: the sections currently in force. /// /// Keyed by `section_number`, sparse on purpose: DVB EIT schedule segments skip @@ -151,16 +197,21 @@ impl Pending { /// One `(PID, table_id)` entry: its snapshot track and the state of every /// sub-table riding it. struct Entry { - track: moq_net::track::Producer, - name: String, + /// Created with the first snapshot that carries anything, so a table filtered + /// to nothing never gets a track. + track: Option, interval: Option, + /// The selected service an SDT actual snapshot is reduced to. + service: Option, active: BTreeMap, /// In-flight generations, keyed by sub-table and version. pending: HashMap<(Identity, u8), Pending>, /// A commit changed `active` since the last cut. dirty: bool, - /// The entry appears in the catalog's `mpegts.si` map (deferred until the first - /// snapshot group exists, so a subscriber never finds an empty track). + /// The newest snapshot group carries at least one sub-table. + populated: bool, + /// The entry appears in the catalog's `mpegts.si` map: exactly while it is + /// `populated`, so a subscriber never finds an empty track. advertised: bool, /// Host-clock micros of the last cut ([`DEBOUNCE`]); `None` until the first. last_cut: Option, @@ -279,19 +330,44 @@ impl Entry { } } - /// Cut a snapshot group: the complete active set, one frame per sub-table. + /// Cut a snapshot group: the complete active set, one frame per sub-table, an + /// SDT reduced to the selected service. Nothing is cut until a snapshot carries + /// something, and that snapshot creates the entry's track on `broadcast`. /// `pts` stamps the frames; `now` is the host clock for the debounce. - fn cut(&mut self, pts: Timestamp, now: u128) -> anyhow::Result<()> { - let mut group = self.track.append_group()?; - for generation in self.active.values() { - let mut payload = Vec::new(); - for section in generation.sections.values() { - payload.extend_from_slice(section); - } + fn cut( + &mut self, + broadcast: &moq_net::broadcast::Producer, + (pid, table_id): (u16, u8), + pts: Timestamp, + now: u128, + ) -> anyhow::Result<()> { + let frames: Vec> = self + .active + .values() + .filter_map(|generation| match self.service { + Some(service) => sdt_service(generation.sections.values(), service), + None => Some(generation.sections.values().flatten().copied().collect()), + }) + .collect(); + self.dirty = false; + self.populated = !frames.is_empty(); + let track = match &mut self.track { + Some(track) => track, + None if frames.is_empty() => return Ok(()), + // Deterministic, greppable name; fall back to a unique suffix on the + // (pathological) collision with an existing track. + None => self.track.insert( + match broadcast.create_track(format!("{pid:#06x}-{table_id:#04x}.si"), None) { + Ok(track) => track, + Err(_) => broadcast.unique_track(".si", None)?, + }, + ), + }; + let mut group = track.append_group()?; + for payload in frames { group.write_frame(pts, payload)?; } group.finish()?; - self.dirty = false; self.last_cut = Some(now); Ok(()) } @@ -309,6 +385,9 @@ pub(super) struct Capture { entries: BTreeMap<(u16, u8), Entry>, /// Host clock driving the cut debounce (see [`DEBOUNCE`]). clock: crate::Clock, + /// The selected program_number, which is the service_id of the only service + /// carried; `None` captures every service. + program: Option, } impl Capture { @@ -318,11 +397,18 @@ impl Capture { catalog, entries: BTreeMap::new(), clock: crate::Clock::new(), + program: None, } } + /// Carry only the service numbered `program`: its EIT actual, and an SDT actual + /// listing it alone. + pub fn select(&mut self, program: u16) { + self.program = Some(program); + } + /// Fold one completed section from `pid` into its `(pid, table_id)` entry, - /// creating the entry (and its track) on first sight of the pair. + /// creating the entry on first sight of the pair. pub fn section(&mut self, pid: u16, section: Vec) -> anyhow::Result<()> { let Some(&table_id) = section.first() else { return Ok(()); @@ -335,43 +421,42 @@ impl Capture { if long_form(§ion) && section[5] & 0x01 == 0 { return Ok(()); } - - if !self.entries.contains_key(&(pid, table_id)) { - // Deterministic, greppable name; fall back to a unique suffix on the - // (pathological) collision with an existing track. - let name = format!("{pid:#06x}-{table_id:#04x}.si"); - let track = match self.broadcast.create_track(name, None) { - Ok(track) => track, - Err(_) => self.broadcast.unique_track(".si", None)?, - }; - self.entries.insert( - (pid, table_id), - Entry { - name: track.name().to_string(), - track, - interval: catalog::si_interval(table_id), - active: BTreeMap::new(), - pending: HashMap::new(), - dirty: false, - advertised: false, - last_cut: None, - }, - ); + // EIT actual (now/next and schedule) is keyed by service_id: another + // service's sub-table describes a program this import does not carry. + if let Some(program) = self.program + && pid == EIT_PID + && matches!(table_id, 0x4E | 0x50..=0x5F) + && !matches!(identity(§ion), Identity::Sub { ext, .. } if ext == program) + { + return Ok(()); } - let entry = self.entries.get_mut(&(pid, table_id)).unwrap(); + + let service = self.program.filter(|_| (pid, table_id) == (SDT_PID, SDT_ACTUAL)); + let entry = self.entries.entry((pid, table_id)).or_insert_with(|| Entry { + track: None, + interval: catalog::si_interval(table_id), + service, + active: BTreeMap::new(), + pending: HashMap::new(), + dirty: false, + populated: false, + advertised: false, + last_cut: None, + }); entry.section(section); Ok(()) } - /// Cut snapshot groups for the dirty entries and advertise any entry with its - /// first snapshot in the catalog (one lock for the whole batch). + /// Cut snapshot groups for the dirty entries, then advertise each entry whose + /// snapshot now carries something and retract each whose snapshot went empty + /// (one catalog lock for the whole batch). /// /// `pts` is the media clock, used to stamp the frames; the cut debounce runs on /// the host clock ([`DEBOUNCE`]), and a first snapshot is never delayed. `force` /// overrides the debounce, for end of stream. pub fn flush(&mut self, pts: Timestamp, force: bool) -> anyhow::Result<()> { let now = self.clock.now().as_micros(); - for entry in self.entries.values_mut() { + for (key, entry) in self.entries.iter_mut() { if !entry.dirty { continue; } @@ -381,27 +466,33 @@ impl Capture { { continue; } - entry.cut(pts, now)?; + entry.cut(&self.broadcast, *key, pts, now)?; } - if self.entries.values().any(|e| !e.advertised && e.last_cut.is_some()) { + if self.entries.values().any(|e| e.advertised != e.populated) { let mut guard = self.catalog.modify()?; let Some(mpegts) = guard.ext.mpegts_mut() else { anyhow::bail!("catalog extension no longer carries an mpegts section"); }; for ((pid, table_id), entry) in self.entries.iter_mut() { - if entry.advertised || entry.last_cut.is_none() { + // Either flag implies the track exists. + let Some(track) = &entry.track else { continue }; + if entry.advertised == entry.populated { continue; } - mpegts.si.entry(*pid).or_default().insert( - *table_id, - catalog::SiEntry { - track: entry.name.clone(), - interval: entry.interval, - ..Default::default() - }, - ); - entry.advertised = true; + if entry.populated { + mpegts.si.entry(*pid).or_default().insert( + *table_id, + catalog::SiEntry { + track: track.name().to_string(), + interval: entry.interval, + ..Default::default() + }, + ); + } else { + retract(&mut mpegts.si, *pid, *table_id, track.name()); + } + entry.advertised = entry.populated; } } Ok(()) @@ -410,8 +501,8 @@ impl Capture { /// Flush everything and finish every track. pub fn finish(&mut self, pts: Timestamp) -> anyhow::Result<()> { self.flush(pts, true)?; - for entry in self.entries.values_mut() { - entry.track.finish()?; + for track in self.entries.values_mut().filter_map(|entry| entry.track.as_mut()) { + track.finish()?; } Ok(()) } @@ -421,7 +512,9 @@ impl Capture { pub fn abort(mut self, err: moq_net::Error) { self.unadvertise(); for (_, entry) in std::mem::take(&mut self.entries) { - let _ = entry.track.abort(err.clone()); + if let Some(track) = entry.track { + let _ = track.abort(err.clone()); + } } } @@ -440,26 +533,34 @@ impl Capture { return; }; for ((pid, table_id), entry) in self.entries.iter_mut() { - if !entry.advertised { + let Some(track) = entry.track.as_ref().filter(|_| entry.advertised) else { continue; - } + }; entry.advertised = false; - if let Some(tables) = mpegts.si.get_mut(pid) { - // Remove only a mapping this capture still owns: a second capture on - // the same broadcast (same key, unique fallback track name) may have - // overwritten it, and it will not re-advertise, so blindly removing - // here would strip the live capture's table for good. - if tables.get(table_id).is_some_and(|e| e.track == entry.name) { - tables.remove(table_id); - } - if tables.is_empty() { - mpegts.si.remove(pid); - } - } + retract(&mut mpegts.si, *pid, *table_id, track.name()); } } } +/// Remove the `(pid, table_id)` mapping from the catalog's `si` map, and the PID +/// once nothing is left on it. +/// +/// Only a mapping still naming `track` goes: a second capture on the same broadcast +/// (same key, unique fallback track name) may have overwritten it, and it will not +/// re-advertise, so blindly removing here would strip the live capture's table for +/// good. +fn retract(si: &mut BTreeMap>, pid: u16, table_id: u8, track: &str) { + let Some(tables) = si.get_mut(&pid) else { + return; + }; + if tables.get(&table_id).is_some_and(|e| e.track == track) { + tables.remove(&table_id); + } + if tables.is_empty() { + si.remove(&pid); + } +} + impl Drop for Capture { fn drop(&mut self) { self.unadvertise(); From 36f99f32de6f12a491889ae0f95fac118cd36342 Mon Sep 17 00:00:00 2001 From: t0ms Date: Wed, 30 Sep 2026 15:54:36 +0100 Subject: [PATCH 2/3] fix(moq-mux): keep a corrupt SDT out of a selected program's SI The rebuilt SDT actual gets a fresh CRC, so a section whose source CRC fails would go out certified. Drop it on arrival instead, keeping the last good snapshot. Test fixtures now carry real CRCs. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/export_test.rs | 51 ++++++++++++++++++++-- rs/moq-mux/src/container/ts/si.rs | 14 ++++-- 2 files changed, 57 insertions(+), 8 deletions(-) diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index d7eceb3838..39af3d831f 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -1741,9 +1741,9 @@ async fn scte35_fixtures_survive_roundtrip() { } /// Build a well-formed long-form section: the full generic header (extension, -/// current version, `number` of `last`), then `body` and a CRC placeholder -/// (capture is verbatim, nothing checks it). The SI store buffers a sub-table -/// until its generation completes, so the header fields must be coherent. +/// current version, `number` of `last`), then `body` and a valid CRC-32/MPEG-2. +/// The SI store buffers a sub-table until its generation completes, so the header +/// fields must be coherent. fn make_long_section(table_id: u8, ext: u16, version: u8, number: u8, last: u8, body: &[u8]) -> Vec { let section_length = 5 + body.len() + 4; let mut s = vec![ @@ -1757,7 +1757,8 @@ fn make_long_section(table_id: u8, ext: u16, version: u8, number: u8, last: u8, last, ]; s.extend_from_slice(body); - s.extend_from_slice(&[0xde, 0xad, 0xbe, 0xef]); + let crc = crc::Crc::::new(&crc::CRC_32_MPEG_2).checksum(&s); + s.extend_from_slice(&crc.to_be_bytes()); s } @@ -3087,6 +3088,48 @@ async fn an_sdt_revision_dropping_the_service_retires_it() { assert_sdt_lists_only(&groups[2][0], 2, 1, b"One"); } +/// A corrupted SDT actual is not rebuilt under a fresh, valid CRC: the section is +/// dropped and the last good snapshot stays in force, rather than the corruption +/// reading as a revision (or as the service leaving). An intact revision after it +/// is captured normally. +#[tokio::test(start_paused = true)] +async fn a_corrupt_sdt_keeps_the_last_good_snapshot() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); + let mut capture = crate::container::ts::si::Capture::new(broadcast, catalog.clone()); + capture.select(1); + let mut feed = |section: Vec| { + capture.section(0x0011, section).unwrap(); + capture.flush(Timestamp::ZERO, true).unwrap(); + catalog + .snapshot() + .ext + .mpegts + .si + .get(&0x0011) + .map(|tables| tables[&0x42].track.clone()) + }; + + let track = feed(sdt_actual(0, 0, 0, &[(1, b"One")])).expect("the SDT is advertised"); + // One unflagged bit flip turns the service name "One" into "Nne" under the source CRC. + let mut corrupt = sdt_actual(1, 0, 0, &[(1, b"One")]); + let name = corrupt.len() - 4 - 3; + corrupt[name] ^= 0x01; + assert_eq!(feed(corrupt), Some(track.clone()), "the corrupt revision kept the SDT"); + assert_eq!(feed(sdt_actual(2, 0, 0, &[(1, b"Uno")])), Some(track.clone())); + capture.finish(Timestamp::ZERO).unwrap(); + + let groups = read_si_groups(&consumer, &track).await; + assert_eq!(groups.len(), 2, "no snapshot for the corrupt revision"); + assert_sdt_lists_only(&groups[0][0], 0, 1, b"One"); + assert_sdt_lists_only(&groups[1][0], 2, 1, b"Uno"); +} + /// A selected program's reduced SI survives export: the TS parses, and importing it /// again finds the same single-service SDT and the same EIT. #[tokio::test(start_paused = true)] diff --git a/rs/moq-mux/src/container/ts/si.rs b/rs/moq-mux/src/container/ts/si.rs index c9ab176e3d..d0a58c8c8e 100644 --- a/rs/moq-mux/src/container/ts/si.rs +++ b/rs/moq-mux/src/container/ts/si.rs @@ -19,10 +19,11 @@ //! An import that selects a program ([`Capture::select`]) carries SI for that //! service alone. Other services' EIT actual sub-tables are dropped on arrival, and //! each SDT actual snapshot is rebuilt as one section holding only the selected -//! service's entry, with a fresh CRC. An SDT that does not list the service carries -//! no SDT actual, so a revision that drops it retires the earlier one. Network-wide -//! tables (NIT, BAT, SDT other, EIT other, TDT/TOT) pass through verbatim, as does -//! everything when no program is selected. +//! service's entry, with a fresh CRC. An SDT actual section whose own CRC fails is +//! dropped, so the last good snapshot stays in force. An SDT that does not list the +//! service carries no SDT actual, so a revision that drops it retires the earlier +//! one. Network-wide tables (NIT, BAT, SDT other, EIT other, TDT/TOT) pass through +//! verbatim, as does everything when no program is selected. use std::collections::{BTreeMap, HashMap}; use std::time::Duration; @@ -432,6 +433,11 @@ impl Capture { } let service = self.program.filter(|_| (pid, table_id) == (SDT_PID, SDT_ACTUAL)); + // The rebuilt SDT gets a fresh CRC, which would certify corruption the + // source's CRC flags; dropping the section keeps the last good generation. + if service.is_some() && CRC.checksum(§ion) != 0 { + return Ok(()); + } let entry = self.entries.entry((pid, table_id)).or_insert_with(|| Entry { track: None, interval: catalog::si_interval(table_id), From 932cce2a1e4c317cd4051fdc8a9db698571316e1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 30 Sep 2026 21:38:31 -0700 Subject: [PATCH 3/3] fix(moq-mux): commit a selected SDT revision only once it is complete A wrap commit treats a dense SDT missing a section as complete, so a revision whose selected-service section failed its CRC retired the last good snapshot. Under a selection, wait for every section instead. Also note the program_number == service_id assumption and cover EIT schedule filtering. Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 2 + rs/moq-mux/src/container/ts/export_test.rs | 77 ++++++++++++++++++++++ rs/moq-mux/src/container/ts/si.rs | 12 +++- 3 files changed, 89 insertions(+), 2 deletions(-) diff --git a/doc/bin/cli.md b/doc/bin/cli.md index a9bb1fab4a..07c732f12e 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -90,6 +90,8 @@ the catalog suffix last: `--broadcast event.hang` publishes `event/1.hang`, A selected program's SI describes that service alone: its SDT lists only the selected service, and other services' EIT is dropped. Network-wide tables (NIT, BAT, TDT/TOT, and the SDT and EIT of other transport streams) pass through. +SI matches the selection by DVB `service_id`, which is assumed to equal the PAT +`program_number`. ```bash moq --connect https://relay.example.com/anon --broadcast event.hang import ts --program all < mux.ts diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 39af3d831f..a858514b9b 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -3130,6 +3130,83 @@ async fn a_corrupt_sdt_keeps_the_last_good_snapshot() { assert_sdt_lists_only(&groups[1][0], 2, 1, b"Uno"); } +/// A multi-section SDT revision with the selected service's section corrupt does not +/// commit as complete-as-observed when the other section repeats: that would read as +/// the service leaving. The revision commits once the section arrives intact. +#[tokio::test(start_paused = true)] +async fn a_partial_sdt_revision_keeps_the_last_good_snapshot() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); + let mut capture = crate::container::ts::si::Capture::new(broadcast, catalog.clone()); + capture.select(1); + let mut feed = |section: Vec| { + capture.section(0x0011, section).unwrap(); + capture.flush(Timestamp::ZERO, true).unwrap(); + catalog + .snapshot() + .ext + .mpegts + .si + .get(&0x0011) + .map(|tables| tables[&0x42].track.clone()) + }; + + feed(sdt_actual(0, 0, 1, &[(1, b"One")])); + let track = feed(sdt_actual(0, 1, 1, &[(2, b"Two")])).expect("the SDT is advertised"); + let mut corrupt = sdt_actual(1, 0, 1, &[(1, b"One")]); + let name = corrupt.len() - 4 - 3; + corrupt[name] ^= 0x01; + assert_eq!(feed(corrupt), Some(track.clone())); + assert_eq!(feed(sdt_actual(1, 1, 1, &[(2, b"Two")])), Some(track.clone())); + assert_eq!( + feed(sdt_actual(1, 1, 1, &[(2, b"Two")])), + Some(track.clone()), + "the repeated section did not commit the partial revision" + ); + assert_eq!(feed(sdt_actual(1, 0, 1, &[(1, b"Uno")])), Some(track.clone())); + capture.finish(Timestamp::ZERO).unwrap(); + + let groups = read_si_groups(&consumer, &track).await; + assert_eq!(groups.len(), 2, "no snapshot for the partial revision"); + assert_sdt_lists_only(&groups[0][0], 0, 1, b"One"); + assert_sdt_lists_only(&groups[1][0], 1, 1, b"Uno"); +} + +/// A selection filters EIT schedule actual (0x50..=0x5F) by service, as it does +/// present/following. +#[tokio::test(start_paused = true)] +async fn a_selection_filters_eit_schedule() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); + let mut capture = crate::container::ts::si::Capture::new(broadcast, catalog.clone()); + capture.select(1); + let schedule = |service: u16| { + let body = [0x00, 0x01, 0x00, 0x02, 0x00, 0x50, service as u8]; + make_long_section(0x50, service, 0, 0, 0, &body) + }; + for service in [1, 2] { + capture.section(0x0012, schedule(service)).unwrap(); + } + capture.finish(Timestamp::ZERO).unwrap(); + + let si = catalog.snapshot().ext.mpegts.si.clone(); + assert_eq!( + read_si_sections(&consumer, &si[&0x0012][&0x50].track).await, + vec![Bytes::from(schedule(1))], + "only the selected service's schedule" + ); +} + /// A selected program's reduced SI survives export: the TS parses, and importing it /// again finds the same single-service SDT and the same EIT. #[tokio::test(start_paused = true)] diff --git a/rs/moq-mux/src/container/ts/si.rs b/rs/moq-mux/src/container/ts/si.rs index d0a58c8c8e..4dbdd6fa91 100644 --- a/rs/moq-mux/src/container/ts/si.rs +++ b/rs/moq-mux/src/container/ts/si.rs @@ -20,7 +20,8 @@ //! service alone. Other services' EIT actual sub-tables are dropped on arrival, and //! each SDT actual snapshot is rebuilt as one section holding only the selected //! service's entry, with a fresh CRC. An SDT actual section whose own CRC fails is -//! dropped, so the last good snapshot stays in force. An SDT that does not list the +//! dropped, and a revision commits only once every section has arrived, so the last +//! good snapshot stays in force until then. An SDT that does not list the //! service carries no SDT actual, so a revision that drops it retires the earlier //! one. Network-wide tables (NIT, BAT, SDT other, EIT other, TDT/TOT) pass through //! verbatim, as does everything when no program is selected. @@ -265,6 +266,12 @@ impl Entry { if let Some(prior) = pending.sections.get(&number) { if *prior == section { + // A reduced SDT reads a missing service as removal, so a dense SDT + // missing a section (dropped for a bad CRC, or lost) must not commit as + // complete-as-observed; it waits for the cycle that completes it. + if self.service.is_some() { + return; + } // The same section came round again before the set completed: the // transmission cycle wrapped, so what we hold is the whole sub-table as // this mux transmits it. For EIT schedule this is the *only* commit @@ -403,7 +410,8 @@ impl Capture { } /// Carry only the service numbered `program`: its EIT actual, and an SDT actual - /// listing it alone. + /// listing it alone. SI keys services by `service_id`, assumed equal to the PAT + /// `program_number`. pub fn select(&mut self, program: u16) { self.program = Some(program); }