diff --git a/Cargo.lock b/Cargo.lock index 3e3d8032a1..aa4323ad61 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/rs/moq-mux/Cargo.toml b/rs/moq-mux/Cargo.toml index 846a02b873..d6d3a07afb 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" h264-parser = { version = "0.4.2" } hang = { workspace = true } kio = { workspace = true } diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 6bbb02801a..fe39363d79 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -6,8 +6,8 @@ //! catalog entries. Elementary streams we don't decode are carried verbatim, one //! MoQ track per PID, described in the `mpegts` catalog section: PES-framed streams //! ride the normal PES reassembly, while section-framed streams (SCTE-35 and -//! other private sections, which are not PES) are intercepted before the mpeg2ts -//! reader and reassembled. TS adds PAT/PMT discovery, PES reassembly, the +//! other private sections, which are not PES) are reassembled as sections, as the +//! PAT and PMT are. TS adds PAT/PMT discovery, PES reassembly, the //! private-section path, and the 90 kHz -> microsecond PTS conversion. The service //! layer is captured too: the identity parsed from the PAT as a //! [`Program`](catalog::Program) record in the `mpegts` catalog section, and the @@ -15,18 +15,14 @@ //! (see [`si`](super::si)), so both survive the round-trip. use std::collections::{BTreeMap, HashMap, HashSet}; -use std::io::Read; -use std::sync::{Arc, Mutex}; use anyhow::Context; -use bytes::BytesMut; use mpeg2ts::es::StreamType; -use mpeg2ts::pes::PesHeader; -use mpeg2ts::ts::payload::Pes; -use mpeg2ts::ts::{Pid, ReadTsPacket, TsPacket, TsPacketReader, TsPayload}; +use mpeg2ts::ts::{Pid, TsPacket}; use super::adts; use super::catalog; +use super::psi::{self, PesStart, Pmt}; use crate::catalog::hang::CatalogExt; use crate::codec::{aac, ac3, eac3, h264, h265, legacy, mp2, opus}; use moq_net::Timestamp; @@ -42,8 +38,12 @@ use moq_net::Timestamp; /// PID, when the catalog `E` carries the [`mpegts`](super::Mpegts) section: PES-framed /// streams ride the normal PES reassembly, section-framed streams (SCTE-35, marked /// by a program-level 'CUEI' registration descriptor, and other private sections) -/// are intercepted before the reader and reassembled. With a base `Catalog<()>` -/// they're logged and dropped instead. +/// are reassembled as sections. With a base `Catalog<()>` they're logged and dropped +/// instead. +/// +/// The PAT and every PMT are reassembled the same way, so a table that spans packets or +/// sits behind a nonzero pointer_field is read whole. A PAT or PMT section whose CRC-32 +/// fails is dropped and counted in [`Stats::crc_error`]; the last good table stays in force. /// /// The selected container applies only to decoded media renditions. Verbatim tracks in the /// `mpegts` catalog section continue to use the legacy Hang container. @@ -58,15 +58,13 @@ pub struct Import { /// snapshot rather than a half-converged one. See [`Reserved`](crate::catalog::Reserved). initial_reservation: Option>, - /// Shared, refillable byte source the persistent reader pulls whole packets - /// from. Kept beside the reader so [`decode`](Self::decode) can append bytes - /// between reads without recreating the reader (which holds PAT/PMT state). - feed: Feed, - reader: TsPacketReader, - - /// PMT PIDs announced by the PAT. With `streams` (the ES PIDs a PMT registers) - /// these are the only PIDs the reader can route; see [`Self::decode`]. - pmt_pids: HashSet, + /// The PAT, reassembled off PID 0. + pat: PatReader, + /// The PMT PIDs PATs have named for the imported program, each with its reassembler. + /// Several programs' PMTs may share one. + pmt_sections: HashMap, + /// PAT and PMT sections dropped for a CRC mismatch, keyed by PID. + crc_errors: BTreeMap, /// Per elementary-stream-PID codec routing. streams: HashMap>, /// Counters from routes a later PMT replaced, keyed by PID. @@ -92,20 +90,13 @@ pub struct Import { /// True once a PMT with at least one supported stream has been parsed. initialized: bool, - /// Whole-packet accumulator. Bytes are routed one TS packet at a time - /// (section-framed verbatim PIDs diverted, the rest fed to the reader); a - /// trailing partial packet is kept here for the next call. + /// Whole-packet accumulator. Bytes are routed one TS packet at a time; a trailing + /// partial packet is kept here for the next call. scratch: Vec, - /// Sync lock. 0x47 is the packet sync byte but also occurs freely in payload (TS - /// has no byte stuffing), so a lone 0x47 isn't a boundary. False until a candidate - /// is confirmed by the next packet's sync byte; once true we stride 188 at a time - /// and trust the per-packet check. Persists across `decode` calls so a candidate - /// pending confirmation at a buffer tail is re-confirmed, not trusted blindly. - synced: bool, - /// Section-framed verbatim PIDs, intercepted before the reader. Private sections - /// (SCTE-35 table_id 0xFC and others) are not PES, so the reader would - /// `Pes::read_from` and abort. Keyed by PID. SCTE-35 is detected via the PMT - /// 'CUEI' registration descriptor. + framer: Framer, + /// Section-framed verbatim PIDs. Private sections (SCTE-35 table_id 0xFC and others) + /// are not PES, so they are reassembled as sections rather than PES-parsed. Keyed by + /// PID. SCTE-35 is detected via the PMT 'CUEI' registration descriptor. sections: HashMap>, /// Whether the catalog can carry the `mpegts` section, sampled once at construction. /// A base `Catalog<()>` can't, so its undecoded PIDs route to `Stream::Ignored`. @@ -119,9 +110,9 @@ pub struct Import { /// Whether the PMT program-level descriptors have been recorded yet (set once; /// PMT `program_info` is stable for the program's life). program_recorded: bool, - /// Reassemblers for the standalone SI PIDs ([`catalog::SI_PIDS`]), keyed by PID. - /// Intercepted before the reader (like SCTE-35 sections) so the service layer - /// survives the round-trip. Only populated with `mpegts` catalog support. + /// Reassemblers for the standalone SI PIDs ([`catalog::SI_PIDS`]), keyed by PID, so + /// the service layer survives the round-trip. Only populated with `mpegts` catalog + /// support. si_sections: HashMap, /// The SI store: buffers each sub-table to a complete generation, publishes the /// per-`(PID, table_id)` snapshot tracks, and advertises them in the catalog. @@ -146,7 +137,6 @@ pub struct Import { impl Import { pub fn new(broadcast: moq_net::broadcast::Producer, reserved: crate::catalog::Reserved) -> Self { - let feed = Feed::default(); let container = hang::catalog::Container::default(); // A long-lived producer handle for catalog edits (mpegts sections, later PMTs); the passed // reservation gates the initial publish and is dropped once the first PMT is parsed. @@ -161,9 +151,9 @@ impl Import { catalog, container, initial_reservation: Some(reserved), - reader: TsPacketReader::new(feed.clone()), - feed, - pmt_pids: HashSet::new(), + pat: PatReader::default(), + pmt_sections: HashMap::new(), + crc_errors: BTreeMap::new(), streams: HashMap::new(), retired_stats: BTreeMap::new(), liveness: Liveness::default(), @@ -174,7 +164,7 @@ impl Import { published: false, initialized: false, scratch: Vec::new(), - synced: false, + framer: Framer::default(), sections: HashMap::new(), supports_mpegts, es_descriptors: HashMap::new(), @@ -239,51 +229,11 @@ impl Import { pub fn decode(&mut self, data: &[u8]) -> anyhow::Result<()> { self.scratch.extend_from_slice(data); - // Route one whole packet at a time. Section-framed verbatim PIDs are - // intercepted here (the reader would PES-parse their sections and abort); - // every other packet is fed to the reader. Per-packet so a PMT is parsed - // (and any section PID registered) before the packets that follow it in the - // same chunk route. + // Route one whole packet at a time, so a PMT is parsed (and any PID it declares + // registered) before the packets that follow it in the same chunk route. let mut off = 0; - while off + TsPacket::SIZE <= self.scratch.len() { - // A TS packet starts with the 0x47 sync byte. Once synced we trust it and stride - // 188 at a time; until then (or after a miss) we must (re)acquire the lock. - if !self.synced || self.scratch[off] != 0x47 { - self.synced = false; - // 0x47 also occurs freely in payload (TS has no byte stuffing), so a lone one - // isn't a boundary. Scan (SIMD via memchr) for a candidate whose next packet - // also begins with 0x47, confirming the 188 stride before locking onto it. - // Striding past a false candidate would route one bogus packet; jumping a flat - // 188 instead would only re-align on exact multiples and could desync forever. - loop { - let Some(rel) = memchr::memchr(0x47, &self.scratch[off..]) else { - // No sync byte left: the buffer is junk, drop it. - off = self.scratch.len(); - break; - }; - off += rel; - match self.scratch.get(off + TsPacket::SIZE) { - // Next packet also starts with a sync byte: lock onto this candidate. - Some(&0x47) => { - self.synced = true; - break; - } - // The byte 188 ahead isn't a sync byte: this 0x47 was payload, keep scanning. - Some(_) => off += 1, - // Can't confirm yet (candidate is near the buffer tail). Stay unsynced so - // it's re-confirmed next call (with the trailing bytes) instead of trusted. - None => break, - } - } - // Unsynced means the buffer had no confirmable sync byte, or the candidate is - // pending confirmation; either way there's nothing to route until more arrives. - if !self.synced { - break; - } - continue; - } - let pkt: [u8; TsPacket::SIZE] = self.scratch[off..off + TsPacket::SIZE].try_into().unwrap(); - off += TsPacket::SIZE; + while let Some(at) = self.framer.next(&self.scratch, &mut off) { + let pkt: [u8; TsPacket::SIZE] = self.scratch[at..at + TsPacket::SIZE].try_into().unwrap(); let pid = (((pkt[1] & 0x1f) as u16) << 8) | pkt[2] as u16; // Every packet paces the multiplex, null stuffing and retransmissions included. if self.supports_mpegts { @@ -298,8 +248,8 @@ impl Import { continue; } // Read the clock's own flag before routing, so the break lands between the media - // either side of it. The PCR PID can be a dedicated one the routing gate below - // drops, and the flag rides an adaptation-only packet as readily as a payload one. + // either side of it. The PCR PID can be a dedicated one nothing below routes, and + // the flag rides an adaptation-only packet as readily as a payload one. // A packet the demodulator flagged corrupt (`transport_error_indicator`) is not // read: its adaptation field is as untrustworthy as its payload, and taking a bit // out of it would break every track in the program on line noise. @@ -319,9 +269,8 @@ impl Import { self.liveness.delivered(pid, units); continue; } - // Intercept the standalone SI PIDs before the routing gate below drops them: - // they carry the service layer, not media, and are not fed to the reader - // (which only routes the PAT/PMT/ES PIDs it learns). + // The standalone SI PIDs carry the service layer, not media, and no PAT or PMT + // names them. if self.supports_mpegts && catalog::SI_PIDS.contains(&pid) { self.si_section(pid, &pkt)?; continue; @@ -353,33 +302,22 @@ impl Import { Continuation::Contiguous => {} } } - // PIDs we don't decode and don't carry (`Stream::Ignored`: a base catalog's - // undecoded streams, or an ambiguous 0x86 PID without CUEI) are dropped here, - // not fed to the PES reader, which aborts on private sections (spec section 7: - // never fatal). - if let Ok(p) = Pid::new(pid) - && matches!(self.streams.get(&p), Some(Stream::Ignored)) - { + let Ok(pid) = Pid::new(pid) else { continue; - } - // Feed the reader only PIDs it can route: the PAT, the PMT PIDs it - // announces, and the ES PIDs a PMT registers. A live capture joins - // mid-stream, so PES arrive before their PSI; feeding those would make - // the reader abort on an unknown PID. Drop them until the layout is - // learned (PSI repeats), then normal demux resumes. - if pid != Pid::PAT - && !Pid::new(pid).is_ok_and(|p| self.pmt_pids.contains(&p) || self.streams.contains_key(&p)) - { - continue; - } - { - let mut state = self.feed.lock(); - state.data.clear(); - state.data.extend_from_slice(&pkt); - state.pos = 0; - } - while let Some(packet) = self.reader.read_ts_packet()? { - self.handle_packet(packet)?; + }; + if pid.as_u16() == Pid::PAT { + self.pat_packet(&pkt)?; + } else if self.pmt_sections.contains_key(&pid) { + self.pmt_packet(pid, &pkt)?; + } else { + // PIDs we don't decode and don't carry (`Stream::Ignored`: a base catalog's + // undecoded streams, or an ambiguous 0x86 PID without CUEI) are dropped, as is + // every PID before the PSI that declares it: a live capture joins mid-stream, + // so PES arrive before their PMT. + match self.streams.get(&pid) { + None | Some(Stream::Ignored) => {} + Some(_) => self.pes_packet(pid, &pkt)?, + } } } @@ -395,81 +333,121 @@ impl Import { Ok(()) } - fn handle_packet(&mut self, packet: TsPacket) -> anyhow::Result<()> { - let pid = packet.header.pid; - match packet.payload { - Some(TsPayload::Pmt(pmt)) => { - // PMTs of several programs may share one PID; only the chosen one maps streams. - if self.program.is_some_and(|program| program != pmt.program_num) { - return Ok(()); - } - // Which PID speaks for the program clock, so a `discontinuity_indicator` there - // can be read as a timebase reset rather than a counter jump. - if self.pcr_pid != pmt.pcr_pid { - self.pcr_pid = pmt.pcr_pid; - // A new clock: the intervals straddling the switch measure nothing. - self.liveness.discontinuity(); - if self.mux_rate.discontinuity() { - self.record_mux_rate()?; - } - } + /// Feed one packet on PID 0 to the PAT's reassembler, applying every whole table. + fn pat_packet(&mut self, pkt: &[u8; TsPacket::SIZE]) -> anyhow::Result<()> { + let crc_error = self.crc_errors.entry(Pid::PAT).or_default(); + if let Some(pat) = self.pat.push(pkt, crc_error) { + self.handle_pat(&pat)?; + } + Ok(()) + } - // SCTE-35 is announced by a program-level registration descriptor with - // format_identifier 'CUEI' (ITU-T J.181). The stream itself uses - // stream_type 0x86, which mpeg2ts maps to a DTS audio variant, so - // detection keys off the CUEI descriptor, not the stream type alone. - let cuei = pmt - .program_info - .iter() - .any(|d| d.tag == 0x05 && d.data.len() >= 4 && &d.data[0..4] == b"CUEI"); - - // Record the program-level descriptors once (PMT program_info is stable); - // export re-emits them verbatim, including the original CUEI. - if self.supports_mpegts && !self.program_recorded && !pmt.program_info.is_empty() { - let program = to_descriptors(&pmt.program_info); - if let Some(mpegts) = self.catalog.modify()?.ext.mpegts_mut() { - mpegts.program_descriptors = program; - } - self.program_recorded = true; - } + /// Feed one packet on a PMT PID to its reassembler, applying every good PMT section. + fn pmt_packet(&mut self, pid: Pid, pkt: &[u8; TsPacket::SIZE]) -> anyhow::Result<()> { + let mut sections = Vec::new(); + self.pmt_sections.entry(pid).or_default().push(pkt, &mut sections); + for section in sections { + // Other tables may share the PID; only a PMT is this path's to check. + if section.first() != Some(&psi::PMT_TABLE_ID) { + continue; + } + if !psi::crc_ok(§ion) { + self.crc_error(pid); + continue; + } + if let Some(pmt) = Pmt::parse(§ion) { + self.handle_pmt(pmt)?; + } + } + Ok(()) + } - for es in &pmt.es_info { - let stream_type = es.stream_type as u8; - // Stash ES descriptors so a decoded media track can record them once its - // (lazily created) track exists; verbatim streams record their own. - if self.supports_mpegts { - self.es_descriptors - .insert(es.elementary_pid.as_u16(), to_descriptors(&es.descriptors)); - } - // Section-framed private data is intercepted before the reader (which - // aborts on private sections): private sections (0x05) and CUEI-marked - // SCTE-35 (0x86). Everything else routes through ensure_stream (a decoded - // codec, PES-framed verbatim, or dropped). - if stream_type == 0x05 || (cuei && matches!(es.stream_type, StreamType::Dts8ChannelLosslessAudio)) { - self.ensure_section(es.elementary_pid, stream_type, &es.descriptors)?; - } else { - self.ensure_stream(es.elementary_pid, es.stream_type, &es.descriptors)?; - } - } + /// A PAT or PMT section failed its CRC: it is dropped whole, and the table already in + /// force stays. + fn crc_error(&mut self, pid: Pid) { + *self.crc_errors.entry(pid.as_u16()).or_default() += 1; + tracing::warn!(pid = pid.as_u16(), "dropped a PSI section with a bad CRC"); + } + + 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(()); + } + // Which PID speaks for the program clock, so a `discontinuity_indicator` there + // can be read as a timebase reset rather than a counter jump. + if self.pcr_pid != pmt.pcr_pid { + self.pcr_pid = pmt.pcr_pid; + // A new clock: the intervals straddling the switch measure nothing. + self.liveness.discontinuity(); + if self.mux_rate.discontinuity() { + self.record_mux_rate()?; + } + } + + // SCTE-35 is announced by a program-level registration descriptor with + // format_identifier 'CUEI' (ITU-T J.181). The stream itself uses + // stream_type 0x86, which is also a DTS audio variant, so detection keys + // off the CUEI descriptor, not the stream type alone. + let cuei = pmt + .program_info + .iter() + .any(|d| d.tag == 0x05 && d.data.len() >= 4 && &d.data[0..4] == b"CUEI"); - // Every stream in the initial program is registered now; release the reservation - // so the catalog publishes once each rendition's config resolves, not before. - self.initial_reservation = None; + // Record the program-level descriptors once (PMT program_info is stable); + // export re-emits them verbatim, including the original CUEI. + if self.supports_mpegts && !self.program_recorded && !pmt.program_info.is_empty() { + if let Some(mpegts) = self.catalog.modify()?.ext.mpegts_mut() { + mpegts.program_descriptors = pmt.program_info.clone(); } - Some(TsPayload::PesStart(pes)) => self.handle_pes_start(pid, pes)?, - Some(TsPayload::PesContinuation(bytes)) => self.handle_pes_continuation(pid, &bytes)?, - Some(TsPayload::Pat(pat)) => self.handle_pat(&pat)?, - _ => {} + self.program_recorded = true; } + + for es in &pmt.streams { + // Stash ES descriptors so a decoded media track can record them once its + // (lazily created) track exists; verbatim streams record their own. + if self.supports_mpegts { + self.es_descriptors.insert(es.pid.as_u16(), es.descriptors.clone()); + } + // Section-framed private data is reassembled as sections, never PES-parsed: + // private sections (0x05) and CUEI-marked SCTE-35 (0x86). Everything else + // routes through ensure_stream (a decoded codec, PES-framed verbatim, or + // dropped). + if es.stream_type == 0x05 || (cuei && es.stream_type == StreamType::Dts8ChannelLosslessAudio as u8) { + self.ensure_section(es.pid, es.stream_type, &es.descriptors)?; + } else { + self.ensure_stream(es.pid, es.stream_type, &es.descriptors)?; + } + } + + // Every stream in the initial program is registered now; release the reservation + // so the catalog publishes once each rendition's config resolves, not before. + self.initial_reservation = None; Ok(()) } - fn ensure_stream( - &mut self, - pid: Pid, - stream_type: StreamType, - descriptors: &[mpeg2ts::ts::Descriptor], - ) -> anyhow::Result<()> { + /// Route one packet on an elementary stream PID into its PES reassembly. + fn pes_packet(&mut self, pid: Pid, pkt: &[u8; TsPacket::SIZE]) -> anyhow::Result<()> { + let payload = match payload(pkt) { + Payload::None => return Ok(()), + Payload::Bytes(payload) => payload, + // No payload can be found in it, so like a corrupt packet it breaks the PES. + Payload::Malformed => { + self.pending.remove(&pid); + if let Some(stream) = self.streams.get_mut(&pid) { + stream.desync(); + } + return Ok(()); + } + }; + if pkt[1] & 0x40 != 0 { + self.handle_pes_start(pid, PesStart::parse(payload)?) + } else { + self.handle_pes_continuation(pid, payload) + } + } + + fn ensure_stream(&mut self, pid: Pid, stream_type: u8, descriptors: &[catalog::Descriptor]) -> anyhow::Result<()> { // A later PMT can remap a PID that was section-framed (intercepted in // `decode`) to a PES codec/verbatim stream. Drop the stale section route first, // or it would keep intercepting the PID and the new stream would never get data. @@ -483,8 +461,8 @@ impl Import { return Ok(()); } - let stream = match stream_type { - StreamType::H264 => { + let stream = match StreamType::from_u8(stream_type).ok() { + Some(StreamType::H264) => { let track = self .broadcast .unique_track(".avc3", self.catalog.track_info(hang::catalog::PRIORITY.video))?; @@ -495,7 +473,7 @@ impl Import { reanchor: Reanchor::default(), } } - StreamType::H265 => { + Some(StreamType::H265) => { let track = self .broadcast .unique_track(".hev1", self.catalog.track_info(hang::catalog::PRIORITY.video))?; @@ -508,7 +486,7 @@ impl Import { } // Only ADTS-framed AAC (0x0F). 0x11 is LATM/LOAS, which uses a different // framing and syncword, so it falls through to the ignored arm below. - StreamType::AdtsAac => Stream::Aac(Box::new(AacStream { + Some(StreamType::AdtsAac) => Stream::Aac(Box::new(AacStream { import: None, broadcast: self.broadcast.clone(), reserved: Some(self.reserve()), @@ -523,15 +501,15 @@ impl Import { // Legacy broadcast audio, carried verbatim. Both MP2 stream types // (0x03 MPEG-1, 0x04 MPEG-2 half rate) share one parser; sample rate and // channels always come from the frame header, not the PMT. - StreamType::Mpeg1Audio | StreamType::Mpeg2HalvedSampleRateAudio => { + Some(StreamType::Mpeg1Audio | StreamType::Mpeg2HalvedSampleRateAudio) => { self.legacy_stream(pid, &mp2::DESCRIPTOR) } - StreamType::DolbyDigitalUpToSixChannelAudio => self.legacy_stream(pid, &ac3::DESCRIPTOR), - StreamType::DolbyDigitalPlusUpTo16ChannelAudioForAtsc => self.legacy_stream(pid, &eac3::DESCRIPTOR), + Some(StreamType::DolbyDigitalUpToSixChannelAudio) => self.legacy_stream(pid, &ac3::DESCRIPTOR), + Some(StreamType::DolbyDigitalPlusUpTo16ChannelAudioForAtsc) => self.legacy_stream(pid, &eac3::DESCRIPTOR), // Opus rides private-data PES (0x06), distinguished from other private streams // by an 'Opus' registration descriptor. Channels and the (always 48 kHz) rate // come from the descriptors, so the importer is built up front. - StreamType::Mpeg2PacketizedData if registration_format(descriptors) == Some(*b"Opus") => { + Some(StreamType::Mpeg2PacketizedData) if registration_format(descriptors) == Some(*b"Opus") => { let config = opus_config(descriptors)?; let track = self .broadcast @@ -544,19 +522,19 @@ impl Import { reanchor: Reanchor::default(), })) } - StreamType::Mpeg1Video | StreamType::Mpeg2Video => Stream::Clock, - // A codec we don't decode. Carry it verbatim as PES when the catalog supports - // the `mpegts` section. 0x86 is excluded: it's ambiguous (DTS audio, or a - // non-conformant SCTE-35 mux without CUEI, which is sections the PES reader - // would abort on), so drop it rather than risk feeding sections to the reader. - other => { - if self.supports_mpegts && !matches!(other, StreamType::Dts8ChannelLosslessAudio) { - let descriptors = to_descriptors(descriptors); + Some(StreamType::Mpeg1Video | StreamType::Mpeg2Video) => Stream::Clock, + // A codec we don't decode, named or not. Carry it verbatim as PES when the + // catalog supports the `mpegts` section. 0x86 is excluded: it's ambiguous (DTS + // audio, or a non-conformant SCTE-35 mux without CUEI, which is sections a PES + // parse would abort on), so drop it rather than risk PES-parsing sections. + _ => { + if self.supports_mpegts && stream_type != StreamType::Dts8ChannelLosslessAudio as u8 { + let descriptors = descriptors.to_vec(); match VerbatimStream::new( self.broadcast.clone(), self.catalog.clone(), pid.as_u16(), - stream_type as u8, + stream_type, descriptors, ) { Ok(stream) => Stream::Verbatim(Box::new(stream)), @@ -566,7 +544,7 @@ impl Import { } } } else { - tracing::warn!(?other, pid = pid.as_u16(), "unsupported TS stream type, dropping"); + tracing::warn!(stream_type, pid = pid.as_u16(), "unsupported TS stream type, dropping"); Stream::Ignored } } @@ -602,12 +580,7 @@ impl Import { /// Register a section-framed verbatim PID (SCTE-35 or other private sections): /// intercepted (see [`Self::decode`]) with a verbatim track when the catalog /// carries the `mpegts` section, dropped as `Ignored` when it can't. - fn ensure_section( - &mut self, - pid: Pid, - stream_type: u8, - descriptors: &[mpeg2ts::ts::Descriptor], - ) -> anyhow::Result<()> { + fn ensure_section(&mut self, pid: Pid, stream_type: u8, descriptors: &[catalog::Descriptor]) -> anyhow::Result<()> { if self.sections.contains_key(&pid.as_u16()) { return Ok(()); } @@ -616,7 +589,7 @@ impl Import { self.continuity.remove(&pid); if !self.supports_mpegts { // Always route to Ignored, replacing any prior codec on this PID (a later PMT - // can reassign it), so a private section never reaches the PES reader. Warn once. + // can reassign it), so a private section is never PES-parsed. Warn once. let previous = self.retire_stream(pid); self.streams.insert(pid, Stream::Ignored); if !matches!(previous, Some(Stream::Ignored)) { @@ -629,7 +602,7 @@ impl Import { } // A prior PMT may have routed this PID to Ignored; drop it so the PID has one route. self.retire_stream(pid); - let descriptors = to_descriptors(descriptors); + let descriptors = descriptors.to_vec(); let stream = SectionStream::new( self.broadcast.clone(), self.catalog.clone(), @@ -643,7 +616,7 @@ impl Import { tracing::debug!( pid = pid.as_u16(), stream_type, - "private section stream detected; intercepting before the reader" + "private section stream detected; reassembling its sections" ); Ok(()) } @@ -660,7 +633,7 @@ impl Import { Some(stream) } - fn handle_pes_start(&mut self, pid: Pid, pes: Pes) -> anyhow::Result<()> { + fn handle_pes_start(&mut self, pid: Pid, pes: PesStart) -> anyhow::Result<()> { // A new PES start means the previous one for this PID is complete. if self.pending.contains_key(&pid) { self.flush(pid)?; @@ -683,12 +656,8 @@ impl Import { // flushes on the next PES, so a SCTE-35 section arriving during this // frame must be timestamped with this frame's PTS ("now"), not the // previous one's. - if pes.header.pts.is_some() { - let pts = unwrap_pts( - &mut self.media_unwrap, - pes.header.pts.map(|t| t.as_u64()), - self.anchor.as_mut(), - )?; + if pes.pts.is_some() { + let pts = unwrap_pts(&mut self.media_unwrap, pes.pts, self.anchor.as_mut())?; let video = match self.streams.get(&pid) { Some(Stream::H264 { reanchor, import, .. }) => { Some((reanchor, import.floor(false), import.floor(true))) @@ -720,16 +689,15 @@ impl Import { return Ok(()); } - let data_len = pes_data_len(&pes.header, pes.pes_packet_len); let mut pending = Pending { - pts: pes.header.pts.map(|t| t.as_u64()), - dts: pes.header.dts.map(|t| t.as_u64()), - stream_id: pes.header.stream_id.as_u8(), + pts: pes.pts, + dts: pes.dts, + stream_id: pes.stream_id, data: Vec::with_capacity(pes.data.len()), - data_len, + data_len: pes.data_len, }; - pending.data.extend_from_slice(&pes.data); - let complete = matches!(data_len, Some(len) if pending.data.len() >= len); + pending.data.extend_from_slice(pes.data); + let complete = matches!(pes.data_len, Some(len) if pending.data.len() >= len); self.pending.insert(pid, pending); if complete { @@ -861,17 +829,17 @@ impl Import { Ok(()) } - /// Pick the imported program from a PAT and learn its PMT PID, so the routing gate in - /// `decode` lets that PMT through and no other. + /// Pick the imported program from a whole PAT and learn its PMT PID, so `decode` + /// reassembles that PMT and no other. /// /// Every PAT is checked, so a program added mid-stream ends an unselected import, and a /// selected program that disappears ends a selected one; media already published stays. - fn handle_pat(&mut self, pat: &mpeg2ts::ts::payload::Pat) -> anyhow::Result<()> { + pub(super) fn handle_pat(&mut self, pat: &psi::Pat) -> anyhow::Result<()> { // program_number 0 is the network PID association, not a program. - let programs: Vec<_> = pat.table.iter().filter(|entry| entry.program_num != 0).collect(); - let numbers = || programs.iter().map(|entry| entry.program_num).collect::>(); + let programs: Vec<_> = pat.programs.iter().filter(|entry| entry.program_number != 0).collect(); + let numbers = || programs.iter().map(|entry| entry.program_number).collect::>(); let entry = match self.program { - Some(program) => match programs.iter().find(|entry| entry.program_num == program) { + Some(program) => match programs.iter().find(|entry| entry.program_number == program) { Some(entry) => entry, None => anyhow::bail!( "transport stream has no program {program}; its PAT lists {}", @@ -884,25 +852,21 @@ impl Import { _ => return Err(MultipleProgramsError { programs: numbers() }.into()), }, }; - self.pmt_pids.insert(entry.program_map_pid); + self.pmt_sections.entry(entry.pmt_pid).or_default(); self.record_program_identity(pat.transport_stream_id, entry) } /// Capture the transport/service identity (TSID, service number, PMT PID) from the /// PAT into the catalog service record, once. No-op without `mpegts` support. - fn record_program_identity( - &mut self, - transport_stream_id: u16, - entry: &mpeg2ts::ts::ProgramAssociation, - ) -> anyhow::Result<()> { + fn record_program_identity(&mut self, transport_stream_id: u16, entry: &psi::Association) -> anyhow::Result<()> { if !self.supports_mpegts || self.identity_recorded { return Ok(()); } if let Some(mpegts) = self.catalog.modify()?.ext.mpegts_mut() { let program = mpegts.program.get_or_insert_with(Default::default); program.transport_stream_id = transport_stream_id; - program.program_number = entry.program_num; - program.pmt_pid = entry.program_map_pid.as_u16(); + program.program_number = entry.program_number; + program.pmt_pid = entry.pmt_pid.as_u16(); } self.identity_recorded = true; Ok(()) @@ -954,8 +918,8 @@ impl Import { Ok(()) } - /// Snapshot what every elementary stream has delivered, and the audio frame sync lost on - /// the way. + /// Snapshot what every elementary stream has delivered, the audio frame sync lost on the + /// way, and the PAT and PMT sections dropped for a bad CRC. /// /// Cheap: it reads counters the demuxer already keeps, so a caller can poll it per /// chunk and report the delta. @@ -975,7 +939,15 @@ impl Import { for (pid, stats) in &mut streams { (stats.units, stats.quiet) = self.liveness.stream(*pid); } - Stats { streams } + Stats { + streams, + crc_error: self.crc_errors.values().sum(), + } + } + + /// [`Stats::crc_error`] per PID, for merging importers that read the same PSI. + pub(super) fn crc_errors(&self) -> &BTreeMap { + &self.crc_errors } /// Abort every track with `err` instead of finishing, so subscribers see the @@ -1013,7 +985,8 @@ fn list_programs(programs: &[u16]) -> String { } /// What each demuxed elementary stream delivered, and the audio frame sync it lost or could -/// not verify, keyed by elementary stream PID. +/// not verify, keyed by elementary stream PID; and the PSI sections the stream as a whole +/// lost to corruption. /// /// Snapshot it with [`Import::stats`]. Every count is cumulative for the life of the /// importer, so what an operator alarms on is the rate: a feed that resyncs once an hour is @@ -1025,12 +998,17 @@ pub struct Stats { /// One row per elementary stream PID the importer carries, ordered by PID. A PID the /// importer drops (an undecoded stream without the `mpegts` catalog section) has none. pub streams: BTreeMap, + /// PAT and PMT sections dropped because their CRC-32 did not match. Each one left the + /// table already in force standing, or, before any, delayed the program's start to the + /// next good repetition. Counted for the stream rather than per elementary stream, since + /// the PSI PIDs have no row of their own. + pub crc_error: u64, } impl Stats { - /// Whether no elementary stream has been registered yet. + /// Whether no elementary stream has been registered yet and nothing has been lost. pub fn is_empty(&self) -> bool { - self.streams.is_empty() + self.streams.is_empty() && self.crc_error == 0 } } @@ -1108,17 +1086,6 @@ impl Pending { } } -/// Convert mpeg2ts PMT descriptors to the catalog's verbatim form. -fn to_descriptors(descriptors: &[mpeg2ts::ts::Descriptor]) -> Vec { - descriptors - .iter() - .map(|d| catalog::Descriptor { - tag: d.tag, - data: bytes::Bytes::copy_from_slice(&d.data), - }) - .collect() -} - /// Create a verbatim track and record it in the `mpegts` catalog section as a /// [`Track`](catalog::Track) with a `verbatim` carriage record. Shared by the /// section- and PES-framed paths. @@ -1471,7 +1438,7 @@ impl Continuity { /// Classify one 188-byte packet, recording what the next call needs. fn observe(&mut self, pkt: &[u8; 188]) -> Continuation { // transport_error_indicator: the demodulator flagged this packet as corrupt, so its - // payload can't be trusted (and we don't validate CRC-32). Forgetting the counter + // payload can't be trusted (and only a PAT or PMT has its CRC-32 checked). Forgetting the counter // keeps the next clean packet from also looking like a gap. if pkt[1] & 0x80 != 0 { self.last_cc = None; @@ -1582,6 +1549,116 @@ impl Liveness { } } +/// 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 +/// stuffing), so a lone 0x47 isn't a boundary. +#[derive(Default)] +pub(super) struct Framer { + /// False until a candidate is confirmed by the next packet's sync byte; once true we + /// stride 188 at a time and trust the per-packet check. Persists across calls so a + /// candidate pending confirmation at a buffer tail is re-confirmed, not trusted blindly. + synced: bool, +} + +impl Framer { + /// Where the next whole packet in `buf` at or after `*off` starts, advancing `*off` past + /// it. `None` once no confirmed packet remains, with `*off` at the first byte the next + /// call still needs. + pub(super) fn next(&mut self, buf: &[u8], off: &mut usize) -> Option { + while *off + TsPacket::SIZE <= buf.len() { + if self.synced && buf[*off] == 0x47 { + let at = *off; + *off += TsPacket::SIZE; + return Some(at); + } + self.synced = false; + // Scan (SIMD via memchr) for a candidate whose next packet also begins with 0x47, + // confirming the 188 stride before locking onto it. Striding past a false + // candidate would route one bogus packet; jumping a flat 188 instead would only + // re-align on exact multiples and could desync forever. + loop { + let Some(rel) = memchr::memchr(0x47, &buf[*off..]) else { + // No sync byte left: the buffer is junk, drop it. + *off = buf.len(); + break; + }; + *off += rel; + match buf.get(*off + TsPacket::SIZE) { + // Next packet also starts with a sync byte: lock onto this candidate. + Some(&0x47) => { + self.synced = true; + break; + } + // The byte 188 ahead isn't a sync byte: this 0x47 was payload, keep scanning. + Some(_) => *off += 1, + // Can't confirm yet (candidate is near the buffer tail). Stay unsynced so + // it's re-confirmed next call (with the trailing bytes) instead of trusted. + None => break, + } + } + // Unsynced means the buffer had no confirmable sync byte, or the candidate is + // pending confirmation; either way there's nothing to route until more arrives. + if !self.synced { + return None; + } + } + None + } +} + +/// 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)] +pub(super) struct PatReader { + sections: SectionReassembler, + table: psi::PatAssembler, +} + +impl PatReader { + /// Consume one packet on PID 0, counting each section dropped for a bad CRC into + /// `crc_error`. Returns the whole PAT if this packet completed one. + pub(super) fn push(&mut self, pkt: &[u8; TsPacket::SIZE], crc_error: &mut u64) -> Option { + let mut sections = Vec::new(); + self.sections.push(pkt, &mut sections); + let mut pat = None; + for section in sections { + if section.first() != Some(&psi::PAT_TABLE_ID) { + continue; + } + if !psi::crc_ok(§ion) { + *crc_error += 1; + tracing::warn!(pid = Pid::PAT, "dropped a PSI section with a bad CRC"); + continue; + } + pat = self.table.section(§ion).or(pat); + } + pat + } +} + +/// Where a packet's payload is. +enum Payload<'a> { + /// Adaptation field only, or reserved `adaptation_field_control`. + None, + Bytes(&'a [u8]), + /// The adaptation field claims to run past the packet. + Malformed, +} + +fn payload(pkt: &[u8; TsPacket::SIZE]) -> Payload<'_> { + let afc = (pkt[3] >> 4) & 0x3; + if afc & 0x1 == 0 { + return Payload::None; + } + let off = if afc & 0x2 != 0 { 5 + pkt[4] as usize } else { 4 }; + match pkt.get(off..) { + Some([]) => Payload::None, + Some(payload) => Payload::Bytes(payload), + None => Payload::Malformed, + } +} + /// Byte-level reassembler for MPEG-TS private sections on one PID. /// /// Private sections (SCTE-35 table_id 0xFC and others) are not PES. This handles @@ -1616,23 +1693,15 @@ impl SectionReassembler { } let pusi = pkt[1] & 0x40 != 0; - let afc = (pkt[3] >> 4) & 0x3; - let has_payload = afc & 0x1 != 0; - - let mut off = 4; - if afc & 0x2 != 0 { - let af_len = pkt[4] as usize; - off = 5 + af_len; - } - - if !has_payload { - return; - } - - if off >= pkt.len() { - return; - } - let payload = &pkt[off..]; + let payload = match payload(pkt) { + Payload::None => return, + Payload::Bytes(payload) => payload, + // A section can't be found in it, so like a counter gap it costs the partial. + Payload::Malformed => { + self.acc.clear(); + return; + } + }; if pusi { // pointer_field: payload[1..1+ptr] is the tail of the section already in @@ -2531,7 +2600,7 @@ impl OpusStream { /// The 4-byte registration `format_identifier` from a PMT registration descriptor /// (tag 0x05), if present. Identifies the codec of a private-data (0x06) stream. -fn registration_format(descriptors: &[mpeg2ts::ts::Descriptor]) -> Option<[u8; 4]> { +fn registration_format(descriptors: &[catalog::Descriptor]) -> Option<[u8; 4]> { descriptors .iter() .find(|d| d.tag == 0x05) @@ -2546,7 +2615,7 @@ fn registration_format(descriptors: &[mpeg2ts::ts::Descriptor]) -> Option<[u8; 4 /// stereo and the Vorbis family 1 mapping above that. Higher codes (0x81 explicitly-coded /// layouts, reserved values) aren't supported, so they fall back to stereo rather than being /// read as a raw 129..=255 count. -fn opus_config(descriptors: &[mpeg2ts::ts::Descriptor]) -> crate::Result { +fn opus_config(descriptors: &[catalog::Descriptor]) -> crate::Result { let channels = descriptors .iter() .find(|d| d.tag == 0x7f && d.data.first() == Some(&0x80)) @@ -2897,17 +2966,6 @@ impl LegacyStream { } } -/// Replicates [`PesHeader::optional_header_len`] (which is crate-private) to -/// compute the declared PES payload length for bounded packets. -fn pes_data_len(header: &PesHeader, pes_packet_len: u16) -> Option { - if pes_packet_len == 0 { - // Unbounded (the usual case for video); flush on the next PES start. - return None; - } - let optional = 3 + header.pts.map_or(0, |_| 5) + header.dts.map_or(0, |_| 5) + header.escr.map_or(0, |_| 6); - pes_packet_len.checked_sub(optional).map(|n| n as usize) -} - /// Swallow a [`MissingKeyframe`](crate::container::MissingKeyframe) from a video /// decode: a TS capture can join mid-GOP, so the deltas before the first keyframe /// have no group to anchor and are simply dropped rather than aborting the demux. @@ -3048,38 +3106,6 @@ impl Reanchor { } } -/// A cloneable, refillable [`Read`] source backed by a shared buffer. -/// -/// [`Import`] loads one whole TS packet at a time; the reader consumes it and -/// reaches end-of-stream before the next refill. -#[derive(Clone, Default)] -struct Feed(Arc>); - -#[derive(Default)] -struct FeedState { - data: BytesMut, - pos: usize, -} - -impl Feed { - fn lock(&self) -> std::sync::MutexGuard<'_, FeedState> { - self.0.lock().unwrap() - } -} - -impl Read for Feed { - fn read(&mut self, out: &mut [u8]) -> std::io::Result { - let mut state = self.lock(); - let n = out.len().min(state.data.len() - state.pos); - if n == 0 { - return Ok(0); - } - out[..n].copy_from_slice(&state.data[state.pos..state.pos + n]); - state.pos += n; - Ok(n) - } -} - #[cfg(test)] pub(super) mod test { use std::collections::BTreeMap; @@ -5488,6 +5514,232 @@ pub(super) mod test { assert!(err.contains("no program 3") && err.contains("1, 2"), "{err}"); } + /// `section` on `pid` as the packets a mux would send, counting from `*cc`: the first + /// opens it behind a pointer_field skipping `tail` (the end of an earlier section), and + /// the rest carry it on. + pub(in crate::container::ts) fn psi_packets(pid: u16, cc: &mut u8, tail: &[u8], section: &[u8]) -> Vec { + let mut payload = vec![tail.len() as u8]; + payload.extend_from_slice(tail); + payload.extend_from_slice(section); + let mut out = Vec::new(); + for (i, chunk) in payload.chunks(184).enumerate() { + let pusi = if i == 0 { 0x40 } else { 0 }; + out.extend_from_slice(&[0x47, pusi | (pid >> 8) as u8, pid as u8, 0x10 | *cc]); + out.extend_from_slice(chunk); + out.resize(out.len().next_multiple_of(188), 0xff); + *cc = (*cc + 1) & 0x0f; + } + out + } + + /// A PAT section listing `(program_number, pmt_pid)`. + pub(in crate::container::ts) fn pat_section(programs: &[(u16, u16)]) -> Vec { + let body: Vec = programs + .iter() + .flat_map(|&(number, pid)| [(number >> 8) as u8, number as u8, 0xe0 | (pid >> 8) as u8, pid as u8]) + .collect(); + super::psi::section(0x00, 1, 0, 0, 0, &body) + } + + /// 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 { + 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 { + body.extend_from_slice(&[ + stream_type, + 0xe0 | (pid >> 8) as u8, + pid as u8, + 0xf0, + descriptors.len() as u8, + ]); + body.extend_from_slice(descriptors); + } + super::psi::section(0x02, program, version, 0, 0, &body) + } + + /// `section` with one bit of its CRC flipped. + pub(in crate::container::ts) fn corrupt(mut section: Vec) -> Vec { + *section.last_mut().unwrap() ^= 0x01; + section + } + + 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. + pub(in crate::container::ts) fn fifty_programs() -> Vec { + let programs: Vec<_> = (1..=50).map(|n| (n, 0x0100 + n)).collect(); + let pat = pat_section(&programs); + let mut data = psi_packets(0, &mut 0, &[], &pat); + assert_eq!(data.len(), 2 * 188, "the PAT spans two packets"); + data.extend(psi_packets( + 0x0132, + &mut 0, + &[], + &pmt_section(50, 0, &[(MP2, 0x0061, &[])]), + )); + data.extend(mp2_pes(0x0061, 0, 90_000, [0xAA, 0xBB])); + data + } + + /// A PAT split into two sections, each in its own packet, listing program 1 then program + /// 2, then each program's PMT (one MP2 stream) and one PES on it. + pub(in crate::container::ts) fn two_section_pat() -> Vec { + let mut cc = 0; + let mut data = Vec::new(); + for (number, program) in [(0, 1u16), (1, 2)] { + let entry = [0x00, program as u8, 0xe0 | program as u8, 0x00]; + let section = super::psi::section(0x00, 1, 0, number, 1, &entry); + data.extend(psi_packets(0, &mut cc, &[], §ion)); + } + 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)); + } + data + } + + fn psi_import() -> (crate::catalog::Producer, super::Import) { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let import = super::Import::new(broadcast, catalog.reserve()); + (catalog, import) + } + + #[test] + fn a_pmt_spanning_two_packets_is_read_whole() { + let (catalog, mut import) = psi_import(); + // A private descriptor long enough to push the second stream into the next packet. + let mut descriptor = vec![0x80, 200]; + descriptor.resize(202, 0x00); + let pmt = pmt_section(1, 0, &[(MP2, 0x0061, &descriptor), (MP2, 0x0062, &[])]); + let mut data = psi_packets(0, &mut 0, &[], &pat_section(&[(1, 0x0100)])); + data.extend(psi_packets(0x0100, &mut 0, &[], &pmt)); + assert_eq!(data.len(), 3 * 188, "the PMT spans two packets"); + data.extend(mp2_pes(0x0061, 0, 90_000, [0xAA, 0xBB])); + data.extend(mp2_pes(0x0062, 0, 90_000, [0xCC, 0xDD])); + + import.decode(&data).unwrap(); + import.finish().unwrap(); + assert_eq!(catalog.snapshot().audio.renditions.len(), 2, "both streams publish"); + assert_eq!(import.stats().crc_error, 0); + } + + #[test] + fn a_pat_behind_a_nonzero_pointer_field_is_read() { + let (catalog, mut import) = psi_import(); + let mut data = psi_packets(0, &mut 0, &[0xab; 7], &pat_section(&[(1, 0x0100)])); + data.extend(psi_packets( + 0x0100, + &mut 0, + &[], + &pmt_section(1, 0, &[(MP2, 0x0061, &[])]), + )); + data.extend(mp2_pes(0x0061, 0, 90_000, [0xAA, 0xBB])); + + import.decode(&data).unwrap(); + import.finish().unwrap(); + assert_eq!(catalog.snapshot().audio.renditions.len(), 1); + } + + #[test] + fn a_program_listed_in_a_pats_second_packet_is_selected() { + let (catalog, import) = psi_import(); + let mut import = import.with_program(50); + import.decode(&fifty_programs()).unwrap(); + import.finish().unwrap(); + assert_eq!(catalog.snapshot().audio.renditions.len(), 1, "program 50 publishes"); + + // Unselected, the whole PAT is what refuses the input. + let (_, mut import) = psi_import(); + let err = import.decode(&fifty_programs()).unwrap_err(); + let err = err.downcast_ref::().unwrap(); + assert_eq!(err.programs, (1..=50).collect::>()); + } + + /// A corrupt PAT and a corrupt PMT between good repetitions are each dropped and counted + /// once, while the layout they would have changed holds; a later good PMT revision still + /// applies. + #[test] + fn corrupt_psi_between_good_repetitions_is_dropped_and_counted() { + let (catalog, mut import) = psi_import(); + let (mut pat_cc, mut pmt_cc) = (0, 0); + let pat = pat_section(&[(1, 0x0100)]); + let pmt = pmt_section(1, 0, &[(MP2, 0x0061, &[])]); + let two = [(MP2, 0x0061, &[][..]), (MP2, 0x0062, &[][..])]; + + let mut data = psi_packets(0, &mut pat_cc, &[], &pat); + data.extend(psi_packets(0x0100, &mut pmt_cc, &[], &pmt)); + data.extend(mp2_pes(0x0061, 0, 90_000, [0xAA, 0xBB])); + import.decode(&data).unwrap(); + let units = import.stats().streams[&0x0061].units; + + // Read, this PAT would end the unselected import with a second program, and this PMT + // would add a stream. + let mut data = psi_packets(0, &mut pat_cc, &[], &corrupt(pat_section(&[(1, 0x0100), (2, 0x0200)]))); + data.extend(psi_packets(0x0100, &mut pmt_cc, &[], &corrupt(pmt_section(1, 1, &two)))); + data.extend(psi_packets(0, &mut pat_cc, &[], &pat)); + data.extend(psi_packets(0x0100, &mut pmt_cc, &[], &pmt)); + data.extend(mp2_pes(0x0061, 1, 180_000, [0xAA, 0xBB])); + import.decode(&data).unwrap(); + + assert_eq!( + catalog.snapshot().audio.renditions.len(), + 1, + "the corrupt PMT added nothing" + ); + assert!( + import.stats().streams[&0x0061].units > units, + "the stream kept delivering" + ); + assert_eq!(import.crc_errors(), &BTreeMap::from([(0x0000, 1), (0x0100, 1)])); + assert_eq!(import.stats().crc_error, 2); + + let mut data = psi_packets(0x0100, &mut pmt_cc, &[], &pmt_section(1, 2, &two)); + data.extend(mp2_pes(0x0062, 0, 180_000, [0xCC, 0xDD])); + import.decode(&data).unwrap(); + import.finish().unwrap(); + assert_eq!( + catalog.snapshot().audio.renditions.len(), + 2, + "the good revision applied" + ); + assert_eq!(import.stats().crc_error, 2, "good sections count nothing"); + } + + /// The positive control: with no good PAT, nothing is learned, so nothing publishes. + #[test] + fn a_feed_whose_only_pat_is_corrupt_publishes_nothing() { + let (catalog, mut import) = psi_import(); + let mut data = psi_packets(0, &mut 0, &[], &corrupt(pat_section(&[(1, 0x0100)]))); + data.extend(psi_packets( + 0x0100, + &mut 0, + &[], + &pmt_section(1, 0, &[(MP2, 0x0061, &[])]), + )); + data.extend(mp2_pes(0x0061, 0, 90_000, [0xAA, 0xBB])); + + import.decode(&data).unwrap(); + import.finish().unwrap(); + assert!(catalog.snapshot().audio.renditions.is_empty()); + assert_eq!(import.stats().crc_error, 1); + assert!(!import.stats().is_empty(), "a dropped section is something to report"); + } + + #[test] + fn adaptation_field_past_the_packet_drops_partial() { + let section = fake_section(0xfc, 300); + let mut malformed = packet(false, 1, 0, §ion[183..]); + // afc 0b11 with an adaptation_field_length reaching past the packet. + malformed[3] = 0x31; + malformed[4] = 200; + let rest = packet(false, 2, 0, §ion[183..]); + assert!(run(&[packet(true, 0, 0, §ion[..183]), malformed, rest]).is_empty()); + } + /// Two MP2 renditions, the first of which the PMT designates as the PCR PID. const PCR_PID: u16 = 0x0061; const PEER_PID: u16 = 0x0062; diff --git a/rs/moq-mux/src/container/ts/mod.rs b/rs/moq-mux/src/container/ts/mod.rs index 05858df90f..3d5fdb454d 100644 --- a/rs/moq-mux/src/container/ts/mod.rs +++ b/rs/moq-mux/src/container/ts/mod.rs @@ -23,6 +23,7 @@ mod export; mod import; mod mux_rate; mod programs; +mod psi; mod si; // The `mpegts` catalog section (per-track PID + descriptors plus verbatim carriage diff --git a/rs/moq-mux/src/container/ts/programs.rs b/rs/moq-mux/src/container/ts/programs.rs index 8f2a139b74..8b7b64c283 100644 --- a/rs/moq-mux/src/container/ts/programs.rs +++ b/rs/moq-mux/src/container/ts/programs.rs @@ -1,9 +1,12 @@ //! Every program of a multiplex, each imported as its own broadcast. +use std::collections::BTreeMap; + use anyhow::Context; -use mpeg2ts::ts::{ReadTsPacket, TsPacket, TsPacketReader, TsPayload}; +use mpeg2ts::ts::{Pid, TsPacket}; -use super::{Ext, Import, Stats}; +use super::import::{Framer, PatReader}; +use super::{Ext, Import, Stats, psi}; use crate::catalog; /// Imports every program the first PAT lists, each as its own broadcast on `origin`, with its @@ -18,8 +21,10 @@ pub struct Programs { name: moq_net::PathOwned, config: catalog::Config, live: bool, - /// Input held until it contains a whole PAT. + /// Input not yet read for the PAT: a trailing partial packet, or what follows the PAT. pending: Vec, + /// Reads the PAT out of the input. + scan: PatScan, /// Empty until the PAT arrives. programs: Vec, } @@ -41,6 +46,7 @@ impl Programs { config, live: false, pending: Vec::new(), + scan: PatScan::default(), programs: Vec::new(), } } @@ -57,13 +63,10 @@ impl Programs { return self.feed(data); } self.pending.extend_from_slice(data); - let Some(programs) = pat_programs(&self.pending).filter(|programs| !programs.is_empty()) else { - // Only a packet's worth of tail can still hold the start of the PAT. - let keep = self.pending.len().saturating_sub(TsPacket::SIZE - 1); - self.pending.drain(..keep); + let Some(pat) = self.scan.scan(&mut self.pending) else { return Ok(()); }; - for program in programs { + for program in pat.program_numbers() { let name = program_broadcast(self.name.as_str(), program); let mut broadcast = self .origin @@ -80,6 +83,9 @@ impl Programs { if self.live { import = import.live(); } + // 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)?; broadcast .announce(Default::default()) .with_context(|| format!("failed to announce broadcast {name}"))?; @@ -101,11 +107,22 @@ impl Programs { } /// Every program's counters in one map: PIDs are unique across a multiplex. + /// + /// Every importer reads the same PAT, and programs sharing a PMT PID read the same PMT + /// sections, so a dropped section counts once however many read it. pub fn stats(&self) -> Stats { let mut stats = Stats::default(); + let mut crc_errors = BTreeMap::::new(); for program in &self.programs { stats.streams.extend(program.import.stats().streams); + for (&pid, &count) in program.import.crc_errors() { + let merged = crc_errors.entry(pid).or_default(); + *merged = (*merged).max(count); + } } + // 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 } @@ -137,32 +154,37 @@ fn program_broadcast(name: &str, program: u16) -> String { } } -/// The program numbers listed by the first whole PAT in `data`, in PAT order, or `None` if -/// there is none yet. -fn pat_programs(data: &[u8]) -> Option> { - let mut off = 0; - while let Some(rel) = memchr::memchr(0x47, &data[off..]) { - off += rel; - let packet = data.get(off..off + TsPacket::SIZE)?; - // PID 0 opening a section. The section CRC rejects a sync byte found in payload. - if packet[1] & 0x5f == 0x40 - && packet[2] == 0 - && let Ok(Some(TsPacket { - payload: Some(TsPayload::Pat(pat)), - .. - })) = TsPacketReader::new(packet).read_ts_packet() - { - return Some( - pat.table - .iter() - .map(|entry| entry.program_num) - .filter(|&program| program != 0) - .collect(), - ); +/// 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; + } } - off += 1; + pending.drain(..off); + found } - None } #[cfg(test)] @@ -172,7 +194,9 @@ mod test { use super::*; use crate::catalog::hang::Container; use crate::container::Consumer; - use crate::container::ts::import::test::two_programs; + use crate::container::ts::import::test::{ + corrupt, fifty_programs, pat_section, psi_packets, two_programs, two_section_pat, + }; #[test] fn program_broadcasts_keep_the_catalog_suffix_last() { @@ -181,6 +205,11 @@ mod test { assert_eq!(program_broadcast("event", 1), "event/1"); } + /// The PAT in a whole buffer, as one [`PatScan`] reads it. + fn pat_programs(data: &[u8]) -> Option> { + Some(PatScan::default().scan(&mut data.to_vec())?.program_numbers()) + } + #[test] fn pat_programs_reads_the_first_whole_pat() { let data = two_programs(); @@ -192,6 +221,118 @@ 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 + /// the program listed in its second packet publishes. + #[tokio::test] + async fn a_pat_spanning_two_packets_starts_every_program() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin.clone(), "event.hang", catalog::Config::default()); + let input = fifty_programs(); + for chunk in input.chunks(100) { + programs.decode(chunk).unwrap(); + } + programs.finish().unwrap(); + assert_eq!(programs.programs.len(), 50); + + let consumer = crate::Source::new(origin.consume(), "event/50.hang") + .broadcast() + .await + .unwrap(); + let catalog = hang::catalog::Catalog::<()>::subscribe(&consumer) + .await + .unwrap() + .next() + .await + .unwrap() + .expect("a catalog"); + 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. + #[test] + fn a_corrupt_pat_starts_no_program() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + // A null packet after each PAT confirms its sync byte. + let mut null = vec![0x47, 0x1f, 0xff, 0x10]; + null.resize(188, 0xff); + let mut cc = 0; + 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_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_eq!(programs.stats().crc_error, 1); + } + + /// How many audio renditions the catalog `path` publishes lists. + async fn renditions(origin: &moq_net::origin::Producer, path: &str) -> usize { + let consumer = crate::Source::new(origin.consume(), path).broadcast().await.unwrap(); + let catalog = hang::catalog::Catalog::<()>::subscribe(&consumer) + .await + .unwrap() + .next() + .await + .unwrap() + .expect("a catalog"); + catalog.audio.renditions.len() + } + + /// A PAT in two sections starts both programs, each importer seeded with the whole table + /// rather than only the section that completed it. + #[tokio::test] + async fn a_pat_in_two_sections_starts_every_program() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin.clone(), "event.hang", catalog::Config::default()); + programs.decode(&two_section_pat()).unwrap(); + programs.finish().unwrap(); + assert_eq!(programs.programs.len(), 2); + assert_eq!(renditions(&origin, "event/1.hang").await, 1); + assert_eq!(renditions(&origin, "event/2.hang").await, 1); + } + + /// A partial PAT holds only itself, not the multiplex that follows it. + #[test] + fn a_partial_pat_does_not_hold_the_multiplex() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + programs.decode(&fifty_programs()[..188]).unwrap(); + let mut null = vec![0x47, 0x1f, 0xff, 0x10]; + null.resize(188, 0xff); + for _ in 0..1000 { + programs.decode(&null).unwrap(); + } + assert!(programs.programs.is_empty()); + assert!( + programs.pending.len() < 2 * 188, + "held {} bytes", + programs.pending.len() + ); + } + + /// A corrupt PAT section sharing a packet with the good one that starts the programs is + /// counted once. + #[test] + fn a_corrupt_pat_beside_a_good_one_counts_once() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut sections = corrupt(pat_section(&[(1, 0x0100)])); + sections.extend(pat_section(&[(1, 0x0100)])); + let mut data = psi_packets(0, &mut 0, &[], §ions); + let mut null = vec![0x47, 0x1f, 0xff, 0x10]; + null.resize(188, 0xff); + data.extend_from_slice(&null); + programs.decode(&data).unwrap(); + assert_eq!(programs.programs.len(), 1); + assert_eq!(programs.stats().crc_error, 1); + } + /// The input is held until the PAT is whole, then each program publishes as its own /// broadcast carrying only that program, live on its own first frame rather than an hour /// apart on one shared clock. diff --git a/rs/moq-mux/src/container/ts/psi.rs b/rs/moq-mux/src/container/ts/psi.rs new file mode 100644 index 0000000000..49585017cc --- /dev/null +++ b/rs/moq-mux/src/container/ts/psi.rs @@ -0,0 +1,422 @@ +//! PAT, PMT, and PES header parsing for the importer's demux. +//! +//! Sections arrive whole from the importer's reassembler, so a table spanning packets or +//! sitting behind a nonzero pointer_field parses like any other. A PAT or PMT section is +//! used only when its CRC-32/MPEG-2 matches: a corrupt one is dropped whole, and the last +//! good table stays in force until a repetition replaces it. + +use std::collections::BTreeMap; + +use anyhow::Context; +use mpeg2ts::ts::Pid; + +use super::catalog::Descriptor; + +const CRC: crc::Crc = crc::Crc::::new(&crc::CRC_32_MPEG_2); + +/// Whether a section's trailing CRC-32/MPEG-2 matches the bytes before it. +pub(super) fn crc_ok(section: &[u8]) -> bool { + let Some(split) = section.len().checked_sub(4) else { + return false; + }; + let (body, crc) = section.split_at(split); + CRC.checksum(body) == u32::from_be_bytes(crc.try_into().expect("four bytes")) +} + +/// The table_id of a PAT section. +pub(super) const PAT_TABLE_ID: u8 = 0x00; +/// The table_id of a PMT section. +pub(super) const PMT_TABLE_ID: u8 = 0x02; + +/// The long-form section header a PAT or PMT carries (ISO/IEC 13818-1 2.4.4.4). +struct Header<'a> { + /// transport_stream_id on a PAT, program_number on a PMT. + extension: u16, + version: u8, + /// Clear on a table announced ahead of use, which is not in force yet. + current: bool, + number: u8, + last: u8, + /// Everything between the header and the CRC. + body: &'a [u8], +} + +impl<'a> Header<'a> { + /// A whole section as the reassembler delivers it: exactly `3 + section_length` bytes. + fn parse(section: &'a [u8]) -> Option { + // section_syntax_indicator: PAT and PMT always use the long form. + if section.len() < 12 || section[1] & 0x80 == 0 { + return None; + } + Some(Self { + extension: u16::from_be_bytes([section[3], section[4]]), + version: (section[5] >> 1) & 0x1f, + current: section[5] & 0x01 != 0, + number: section[6], + last: section[7], + body: §ion[8..section.len() - 4], + }) + } +} + +/// A 13-bit PID behind three reserved bits. +fn pid(bytes: [u8; 2]) -> Pid { + Pid::new(u16::from_be_bytes(bytes) & 0x1fff).expect("13-bit PID") +} + +/// A 12-bit length behind four reserved or unused bits. +fn length(bytes: [u8; 2]) -> usize { + usize::from(u16::from_be_bytes(bytes) & 0x0fff) +} + +/// One program the PAT lists. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) struct Association { + /// 0 names the network PID rather than a program. + pub program_number: u16, + pub pmt_pid: Pid, +} + +/// A whole program association table, every section of its version. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) struct Pat { + pub transport_stream_id: u16, + 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 +/// program is never missed for having been listed in a section still on its way. +#[derive(Default)] +pub(super) struct PatAssembler { + /// The version, transport_stream_id, and last_section_number being collected. + table: Option<(u8, u16, u8)>, + sections: BTreeMap>, +} + +impl PatAssembler { + /// Fold in one CRC-checked PAT section, returning the whole table once it is complete. + /// + /// Every repetition of a complete table returns it again, as the PAT repeats on the wire. + pub fn section(&mut self, section: &[u8]) -> Option { + let header = Header::parse(section)?; + if section[0] != PAT_TABLE_ID || !header.current || header.number > header.last { + return None; + } + let (entries, []) = header.body.as_chunks::<4>() else { + return None; + }; + let programs = entries + .iter() + .map(|entry| Association { + program_number: u16::from_be_bytes([entry[0], entry[1]]), + pmt_pid: pid([entry[2], entry[3]]), + }) + .collect(); + + let table = (header.version, header.extension, header.last); + if self.table != Some(table) { + self.table = Some(table); + self.sections.clear(); + } + self.sections.insert(header.number, programs); + if self.sections.len() <= usize::from(header.last) { + return None; + } + Some(Pat { + transport_stream_id: header.extension, + programs: self.sections.values().flatten().copied().collect(), + }) + } +} + +/// One elementary stream a PMT lists. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) struct Es { + /// The raw stream_type, kept whole so a type this crate has no name for still routes. + pub stream_type: u8, + pub pid: Pid, + pub descriptors: Vec, +} + +/// A program map table. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) struct Pmt { + pub program_number: u16, + /// `None` when the PMT names PID 0x1FFF: the program carries no clock of its own. + pub pcr_pid: Option, + pub program_info: Vec, + pub streams: Vec, +} + +impl Pmt { + /// Parse one CRC-checked PMT section. `None` for a table not yet in force, one that + /// claims more than one section (a PMT never does), or one whose lengths overrun it. + pub fn parse(section: &[u8]) -> Option { + let header = Header::parse(section)?; + if section[0] != PMT_TABLE_ID || !header.current || header.number != 0 || header.last != 0 { + return None; + } + let body = header.body; + if body.len() < 4 { + return None; + } + let pcr_pid = Some(pid([body[0], body[1]])).filter(|pid| pid.as_u16() != 0x1fff); + let info_len = length([body[2], body[3]]); + let program_info = descriptors(body.get(4..4 + info_len)?)?; + + let mut rest = &body[4 + info_len..]; + let mut streams = Vec::new(); + while !rest.is_empty() { + if rest.len() < 5 { + return None; + } + let len = length([rest[3], rest[4]]); + streams.push(Es { + stream_type: rest[0], + pid: pid([rest[1], rest[2]]), + descriptors: descriptors(rest.get(5..5 + len)?)?, + }); + rest = &rest[5 + len..]; + } + Some(Self { + program_number: header.extension, + pcr_pid, + program_info, + streams, + }) + } +} + +/// A descriptor loop, or `None` if a descriptor runs past it. +fn descriptors(mut data: &[u8]) -> Option> { + let mut out = Vec::new(); + while !data.is_empty() { + let [tag, len, ..] = *data else { + return None; + }; + let body = data.get(2..2 + usize::from(len))?; + out.push(Descriptor { + tag, + data: bytes::Bytes::copy_from_slice(body), + }); + data = &data[2 + usize::from(len)..]; + } + Some(out) +} + +/// The start of a PES packet: its header, and the payload bytes that follow it in this +/// TS packet. +pub(super) struct PesStart<'a> { + pub stream_id: u8, + /// Raw 90 kHz. + pub pts: Option, + /// Raw 90 kHz. + pub dts: Option, + /// The payload length a bounded PES declares, or `None` for an unbounded one (the + /// usual case for video). + pub data_len: Option, + pub data: &'a [u8], +} + +impl<'a> PesStart<'a> { + /// Parse the PES header at the start of a payload-unit-start packet's payload. + /// + /// Optional fields this demux has no use for (ESCR, ES rate, trick mode, copy info, the + /// PES CRC, extensions, stuffing) are skipped by `PES_header_data_length`. A payload that + /// is not a PES header at all is an error, as it was when an external reader parsed it. + pub fn parse(payload: &'a [u8]) -> anyhow::Result { + anyhow::ensure!(payload.len() >= 6, "PES header truncated"); + anyhow::ensure!(payload[..3] == [0, 0, 1], "missing PES start code prefix"); + let stream_id = payload[3]; + let packet_len = usize::from(u16::from_be_bytes([payload[4], payload[5]])); + + // Streams that carry no optional header (ISO/IEC 13818-1 table 2-21). + if matches!(stream_id, 0xbc | 0xbe | 0xbf | 0xf0 | 0xf1 | 0xf2 | 0xf8 | 0xff) { + return Ok(Self { + stream_id, + pts: None, + dts: None, + data_len: Some(packet_len).filter(|&len| len != 0), + data: &payload[6..], + }); + } + + anyhow::ensure!(payload.len() >= 9, "PES header truncated"); + anyhow::ensure!(payload[6] & 0xc0 == 0x80, "unexpected PES header marker bits"); + anyhow::ensure!(payload[6] & 0x30 == 0, "scrambled PES is not supported"); + let (pts_flag, dts_flag) = (payload[7] & 0x80 != 0, payload[7] & 0x40 != 0); + anyhow::ensure!(pts_flag || !dts_flag, "PES DTS without a PTS"); + let header_len = usize::from(payload[8]); + let optional = payload.get(9..9 + header_len).context("PES header truncated")?; + + let (pts, dts) = match (pts_flag, dts_flag) { + (false, _) => (None, None), + (true, false) => (Some(timestamp(optional.get(..5), 0b0010)?), None), + (true, true) => ( + Some(timestamp(optional.get(..5), 0b0011)?), + Some(timestamp(optional.get(5..10), 0b0001)?), + ), + }; + Ok(Self { + stream_id, + pts, + dts, + data_len: Some(packet_len) + .filter(|&len| len != 0) + .and_then(|len| len.checked_sub(3 + header_len)), + data: &payload[9 + header_len..], + }) + } +} + +/// A 33-bit PTS or DTS behind its 4-bit prefix and three marker bits. +fn timestamp(bytes: Option<&[u8]>, prefix: u8) -> anyhow::Result { + let bytes = bytes.context("PES header truncated")?; + anyhow::ensure!(bytes[0] >> 4 == prefix, "unexpected PES timestamp prefix"); + anyhow::ensure!( + bytes[0] & 1 == 1 && bytes[2] & 1 == 1 && bytes[4] & 1 == 1, + "unexpected PES timestamp marker bit" + ); + Ok((u64::from(bytes[0] >> 1 & 0x07) << 30) + | (u64::from(bytes[1]) << 22) + | (u64::from(bytes[2] >> 1) << 15) + | (u64::from(bytes[3]) << 7) + | u64::from(bytes[4] >> 1)) +} + +/// A long-form section with a valid CRC around `body`. +#[cfg(test)] +pub(super) fn section(table_id: u8, extension: u16, version: u8, number: u8, last: u8, body: &[u8]) -> Vec { + let len = 5 + body.len() + 4; + let mut out = vec![table_id, 0xb0 | (len >> 8) as u8, len as u8]; + out.extend_from_slice(&extension.to_be_bytes()); + out.extend_from_slice(&[0xc1 | (version << 1), number, last]); + out.extend_from_slice(body); + let crc = CRC.checksum(&out); + out.extend_from_slice(&crc.to_be_bytes()); + out +} + +#[cfg(test)] +mod test { + use super::*; + + #[test] + fn crc_matches_mpeg2() { + // The single-program PAT ffmpeg writes: TSID 1, program 1 on PID 0x1000. + let pat = [ + 0x00, 0xb0, 0x0d, 0x00, 0x01, 0xc1, 0x00, 0x00, 0x00, 0x01, 0xf0, 0x00, 0x2a, 0xb1, 0x04, 0xb2, + ]; + assert!(crc_ok(&pat)); + let mut flipped = pat; + flipped[9] ^= 0x01; + assert!(!crc_ok(&flipped)); + assert!(!crc_ok(&pat[..3])); + } + + #[test] + fn pat_waits_for_every_section() { + let mut pat = PatAssembler::default(); + let first = section(0, 7, 3, 0, 1, &[0x00, 0x01, 0xe1, 0x00]); + let second = section(0, 7, 3, 1, 1, &[0x00, 0x02, 0xe2, 0x00]); + 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]); + // A repetition of either section repeats the whole table. + assert_eq!(pat.section(&first), Some(whole)); + + // A new version starts over rather than mixing with the old one's sections. + let next = section(0, 7, 4, 0, 1, &[0x00, 0x03, 0xe3, 0x00]); + assert_eq!(pat.section(&next), None); + } + + #[test] + fn pat_skips_a_table_not_yet_in_force() { + let mut next = section(0, 1, 0, 0, 0, &[0x00, 0x01, 0xe1, 0x00]); + next[5] &= !0x01; + assert_eq!(PatAssembler::default().section(&next), None); + } + + #[test] + fn pmt_keeps_unnamed_stream_types_and_descriptors() { + let body = [ + 0xe1, 0x00, // PCR PID 0x100 + 0xf0, 0x06, 0x05, 0x04, b'C', b'U', b'E', b'I', // program_info: CUEI registration + 0x2d, 0xe1, 0x01, 0xf0, 0x00, // MPEG-H 3D audio, no descriptors + 0x1b, 0xe1, 0x00, 0xf0, 0x03, 0x0a, 0x01, 0x00, // H.264 with a language descriptor + ]; + let pmt = Pmt::parse(§ion(0x02, 9, 0, 0, 0, &body)).unwrap(); + assert_eq!(pmt.program_number, 9); + assert_eq!(pmt.pcr_pid, Some(Pid::new(0x100).unwrap())); + assert_eq!(pmt.program_info[0].tag, 0x05); + assert_eq!(&pmt.program_info[0].data[..], b"CUEI"); + assert_eq!(pmt.streams.len(), 2); + assert_eq!(pmt.streams[0].stream_type, 0x2d); + assert_eq!(pmt.streams[1].pid, Pid::new(0x100).unwrap()); + assert_eq!(pmt.streams[1].descriptors[0].tag, 0x0a); + } + + #[test] + fn pmt_refuses_an_overrun() { + let body = [0xff, 0xff, 0xf0, 0x00, 0x1b, 0xe1, 0x00, 0xf0, 0x09, 0x0a]; + assert_eq!(Pmt::parse(§ion(0x02, 1, 0, 0, 0, &body)), None); + // PCR PID 0x1FFF: no clock. + let body = [0xff, 0xff, 0xf0, 0x00]; + assert_eq!(Pmt::parse(§ion(0x02, 1, 0, 0, 0, &body)).unwrap().pcr_pid, None); + } + + #[test] + fn pes_header_reads_timestamps_and_skips_optional_fields() { + // PTS 0x1_2345_6789 and DTS 90000, then an ES rate field and three stuffing bytes. + let mut payload = vec![0, 0, 1, 0xe0, 0x00, 0x00, 0x80, 0xd0, 16]; + payload.extend_from_slice(&[0x39, 0x8d, 0x15, 0xcf, 0x13]); + payload.extend_from_slice(&[0x11, 0x00, 0x05, 0xbf, 0x21]); + payload.extend_from_slice(&[0x80, 0x00, 0x01, 0xff, 0xff, 0xff]); + payload.extend_from_slice(b"data"); + let pes = PesStart::parse(&payload).unwrap(); + assert_eq!(pes.stream_id, 0xe0); + assert_eq!(pes.pts, Some(0x1_2345_6789)); + assert_eq!(pes.dts, Some(90_000)); + assert_eq!(pes.data_len, None); + assert_eq!(pes.data, b"data"); + } + + #[test] + fn bounded_pes_length_counts_header_stuffing() { + // PTS only, one stuffing byte, four payload bytes: 3 + 6 + 4 = 13. + let mut payload = vec![0, 0, 1, 0xc0, 0x00, 13, 0x80, 0x80, 6]; + payload.extend_from_slice(&[0x21, 0x00, 0x01, 0x00, 0x01, 0xff]); + payload.extend_from_slice(b"mp2!"); + let pes = PesStart::parse(&payload).unwrap(); + assert_eq!(pes.pts, Some(0)); + assert_eq!(pes.data_len, Some(4)); + assert_eq!(pes.data, b"mp2!"); + } + + #[test] + fn pes_without_optional_header() { + let payload = [0, 0, 1, 0xbf, 0x00, 0x02, 0xaa, 0xbb]; + let pes = PesStart::parse(&payload).unwrap(); + assert_eq!((pes.pts, pes.data_len, pes.data), (None, Some(2), &[0xaa, 0xbb][..])); + } + + #[test] + fn not_a_pes_header() { + assert!(PesStart::parse(&[0xfc, 0x30, 0x11, 0x00, 0x00, 0x00, 0x00]).is_err()); + assert!(PesStart::parse(&[0, 0, 1, 0xe0, 0, 0, 0x80, 0x40, 0]).is_err()); + } +} diff --git a/rs/moq-mux/src/container/ts/stats.rs b/rs/moq-mux/src/container/ts/stats.rs index 9853721168..0172a22f03 100644 --- a/rs/moq-mux/src/container/ts/stats.rs +++ b/rs/moq-mux/src/container/ts/stats.rs @@ -1,4 +1,4 @@ -//! Reporting on the per-stream counters [`Import::stats`](super::Import::stats) returns. +//! Reporting on the counters [`Import::stats`](super::Import::stats) returns. use std::collections::BTreeSet; @@ -27,7 +27,7 @@ impl Log { /// Compare a snapshot taken [`INTERVAL`](Self::INTERVAL) after the last. /// /// Logs a stream whose access units did not advance since the last sample, once per - /// silence, and any stream whose frame-sync counters moved. + /// silence, any stream whose frame-sync counters moved, and PSI sections dropped since. pub fn sample(&mut self, latest: Stats) { self.sync(&latest); for (pid, stream) in &latest.streams { @@ -49,17 +49,24 @@ impl Log { /// Compare the snapshot taken after [`Import::finish`](super::Import::finish). /// - /// It covers less than an interval, so only the frame sync is compared: the drain at end - /// of input can publish a frame nothing vouched for. + /// It covers less than an interval, so only the losses are compared: the drain at end of + /// input can publish a frame nothing vouched for. pub fn finish(&self, latest: &Stats) { self.sync(latest); } - /// Report the streams whose frame-sync counters moved, one line each. + /// Report the streams whose frame-sync counters moved, one line each, and the PSI + /// sections dropped for a bad CRC if that count moved. /// - /// The importer already warns on each individual resync; this is the running total, which - /// is what an operator turns into a rate. + /// The importer already warns on each individual resync and dropped section; this is the + /// running total, which is what an operator turns into a rate. fn sync(&self, latest: &Stats) { + if self.previous.as_ref().map_or(0, |previous| previous.crc_error) != latest.crc_error { + tracing::info!( + crc_error = latest.crc_error, + "PAT or PMT sections dropped for a bad CRC" + ); + } let lost = |stream: &StreamStats| (stream.resyncs, stream.discarded, stream.unconfirmed); for (pid, stream) in &latest.streams { let previous = self.previous.as_ref().and_then(|previous| previous.streams.get(pid)); @@ -156,4 +163,29 @@ mod test { }); assert!(!logs_contain("stopped delivering"), "the stream kept counting"); } + + /// Dropped PSI sections are reported when the count moves, with the running total, and + /// not again while it holds. + #[test] + #[tracing_test::traced_test] + fn reports_crc_errors_when_they_move() { + let sample = |crc_error: u64| Stats { + crc_error, + ..Default::default() + }; + + let mut log = Log::default(); + log.sample(sample(0)); + log.sample(sample(2)); + log.sample(sample(2)); + log.finish(&sample(3)); + + logs_assert(|lines: &[&str]| { + let dropped: Vec<_> = lines.iter().filter(|line| line.contains("bad CRC")).collect(); + match dropped.as_slice() { + [first, second] if first.contains("crc_error=2") && second.contains("crc_error=3") => Ok(()), + _ => Err(format!("expected one line per move, got {dropped:?}")), + } + }); + } }