From 63a4651e9f7013b5a258a530dce7cf4ee41d6dee Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 2 Oct 2026 10:00:44 -0700 Subject: [PATCH 1/7] chore: claim TS damaged units quest Co-Authored-By: GPT-6 From fd7f1c3ec076592a0d6a52bbdad840c465c9d62a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 2 Oct 2026 10:20:57 -0700 Subject: [PATCH 2/7] fix(mux): refuse damaged TS units without ending ingest Co-Authored-By: GPT-6 --- doc/bin/cli.md | 7 + doc/bin/srt.md | 4 +- doc/lib/rs/moq-mux.md | 8 + quest/m1/README.md | 1 - quest/m1/ts-damaged-units.md | 56 ---- rs/moq-mux/src/container/ts/import.rs | 402 ++++++++++++++++++++++++-- rs/moq-mux/src/container/ts/stats.rs | 59 +++- 7 files changed, 446 insertions(+), 91 deletions(-) delete mode 100644 quest/m1/ts-damaged-units.md diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 45fbbb909d..942f7de5d8 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -79,6 +79,13 @@ downstream sees it. A sparse stream such as SCTE-35 goes quiet between cues, so the line reports rather than alarms; `Import::stats` carries the same counters for a caller that sets its own limit. +A corrupt media packet, malformed PES header, or damaged codec access unit is +refused whole and counted in the PID's cumulative `damaged` counter, beside +`resyncs`, `discarded`, and `unconfirmed`. Ingest continues on every other PID; +the damaged video track resumes at its next keyframe. The shared TS stats log +reports each counter increase, including at the end of input. Publishing, +catalog, and clock errors still end the import. + MPEG-TS import takes one program. A multi-program stream is refused before anything is published, naming its programs, rather than merged onto one clock; a PAT that adds a program mid-stream ends the import the same way. diff --git a/doc/bin/srt.md b/doc/bin/srt.md index a2abd327ab..8997ac93dc 100644 --- a/doc/bin/srt.md +++ b/doc/bin/srt.md @@ -11,7 +11,9 @@ remote. The payload is MPEG-TS, so the same codecs as [`import ts`](/bin/cli) apply: H.264/H.265 video and AAC, MP2, AC-3, or E-AC-3 audio. Ingest logs the same per-stream lines as `import ts`, under an `srt{path=...}` span: an elementary stream that stopped delivering access units, and audio frame sync -lost. +lost, plus damaged units refused on each PID. Damage drops that unit and keeps +the session alive; video resumes at its next keyframe. The cumulative `damaged` +count is the same one reported by `import ts`. ```bash # Accept a contribution feed and publish it diff --git a/doc/lib/rs/moq-mux.md b/doc/lib/rs/moq-mux.md index 6ac125f6bf..2a24576112 100644 --- a/doc/lib/rs/moq-mux.md +++ b/doc/lib/rs/moq-mux.md @@ -25,6 +25,14 @@ the inverse and skip stalled groups past a max age. Per-codec producers (`import::Opus`, H.264, and so on) are available for feeding frames you already have. +MPEG-TS `Import::stats` returns cumulative per-PID `StreamStats`: delivered +`units`, transport-clock `quiet` time, audio `resyncs`, scanned bytes `discarded`, +frames `unconfirmed`, and damaged units refused in `damaged`. A malformed media +packet, PES header, or codec unit is dropped whole; only that PID loses sync, +and video waits for its next keyframe. Publishing and catalog failures remain +fatal. `ts::stats::Log` reports these counters for both the CLI and SRT gateway. +The exporter's `damaged` count remains zero. + fMP4 export emits one fragment per publisher group by default, including audio. A closed group flushes even if the live publisher pauses before its next frame. `fmp4::Export::with_fragment_duration` adds an explicit duration cap. A zero cap diff --git a/quest/m1/README.md b/quest/m1/README.md index 2a508b46f4..f9b528189f 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -61,7 +61,6 @@ QUIC studies there on that rule. - [Legacy end overshoot](/quest/m1/legacy-end-overshoot.md) - browser playback survives a group that starts inside the previous group's estimated end - [Plan: max-delay](/quest/m1/plan-max-delay.md) - decide whether `max_age` becomes `max_delay` across the CLI, APIs, and bindings - [Wire compatibility](/quest/m1/wire-compat.md) - a nightly run tests this checkout against the last published release for tokens, session wire, and catalog/container -- [TS damaged units](/quest/m1/ts-damaged-units.md) - one malformed PES or access unit is dropped, counted as `damaged`, and resynced at the next keyframe instead of ending the import - [RTMP interleaving](/quest/m1/rtmp-interleaving.md) - isolate partial messages before optimizing assembly copies - [Merge main into dev](/quest/m1/dev-sync.md) - dev picks up `ts::stats` from main - [TS stats module](/quest/m1/ts-stats-module.md) - on dev, the TS stats types move under `ts::stats` as `Snapshot` and `Stream`, with an owned `track` diff --git a/quest/m1/ts-damaged-units.md b/quest/m1/ts-damaged-units.md deleted file mode 100644 index e7124059e6..0000000000 --- a/quest/m1/ts-damaged-units.md +++ /dev/null @@ -1,56 +0,0 @@ -# [M] TS import drops a damaged unit instead of ending the ingest - -## Goal - -`moq import ts`, and `moq-srt` through the same importer, survive a packet, -PES, or access unit they cannot parse: a malformed PES header or media-PID -adaptation field with `transport_error_indicator` clear, or a codec error such -as an H.264 NAL with `forbidden_zero_bit` set. The damaged unit is dropped and -counted, its track waits for its next keyframe, and the ingest carries on. -Today the first such error ends the import, and the gateway hangs up. - -## Plan - -Today `decode` in `rs/moq-mux/src/container/ts/import.rs` runs -`while let Some(packet) = self.reader.read_ts_packet()? { self.handle_packet(packet)?; }`, -so any PES-header error from the reader and any codec error from `flush` ends -the import. The `transport_error_indicator` branch already drops a flagged -packet (clears `pending[pid]` and calls `stream.desync()`), and it is the model. - -Decided (maintainer, 2026-09-30): - -- Fail-loud holds at unit granularity, as the importer - holds it at section granularity for PSI (#4584): a damaged unit is refused whole, never half - published, and counted, and nothing else about the feed changes. PSI damage - is already handled. -- Drop the unit the way the TEI branch does: clear the PID's pending PES and - desync its stream, so a codec that needs one waits for the next keyframe. - Make sure the scratch buffer still advances past the packet on this path - (today an error skips `self.scratch.drain(..off)`). -- Count each drop in a new cumulative per-PID `damaged` counter on the stream's - stats row, beside `resyncs` and `discarded`, reported by `ts::stats::Log`. - It names what happened; no TR 101 290 check matches a codec error, and - [TS import health](/quest/m2/ts-import-health.md) keeps the ETSI names. - The rows are `#[non_exhaustive]`, so the field is additive; follow - [TS stats module](/quest/m1/ts-stats-module.md)'s names if it has landed. -- Errors that are not confined to one unit (the producer refusing a rewind, - origin or catalog failures) stay fatal. -- Built on the demux TS PSI reassembly owns, since that quest moves PES header - parsing into moq-mux. - -Tests, from [#4581](https://github.com/moq-dev/moq/issues/4581)'s stimuli on a -fixture: a video PES header with its flag and timestamp bytes zeroed (TEI -clear), and one H.264 NAL with `forbidden_zero_bit` set. Each import carries -on, counts exactly one `damaged` on that PID and none elsewhere, and publishes -again from the next keyframe; a clean fixture counts zero. Document the counter -wherever the TS stats fields are described. - -Public API: additive, one stats field. Wire: none. - -## Closes - -- [#4581](https://github.com/moq-dev/moq/issues/4581) - one malformed packet ends the TS import - -## Related - -- [TS import health](/quest/m2/ts-import-health.md) - the TR 101 290 counters for the same feed diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index fb00d66b82..1f9e651caf 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -69,6 +69,8 @@ pub struct Import { streams: HashMap>, /// Counters from routes a later PMT replaced, keyed by PID. retired_stats: BTreeMap, + /// Damaged packets, PES, or access units refused, cumulative across PID remaps. + damaged: BTreeMap, /// Access units per elementary stream, timed on the program clock. liveness: Liveness, /// In-progress PES reassembly, keyed by elementary PID. @@ -156,6 +158,7 @@ impl Import { crc_errors: BTreeMap::new(), streams: HashMap::new(), retired_stats: BTreeMap::new(), + damaged: BTreeMap::new(), liveness: Liveness::default(), pending: HashMap::new(), continuity: HashMap::new(), @@ -241,6 +244,15 @@ impl Import { if self.supports_mpegts { self.mux_rate.packet(); } + if let Ok(pid) = Pid::new(pid) + && self.streams.contains_key(&pid) + && pkt[1] & 0x80 == 0 + && !adaptation_valid(&pkt) + { + self.damage(pid, &anyhow::anyhow!("malformed TS adaptation field")); + self.continuity.remove(&pid); + continue; + } let continuation = Pid::new(pid) .ok() .filter(|pid| self.streams.contains_key(pid) || self.pcr_pid == Some(*pid)) @@ -289,10 +301,7 @@ impl Import { // Flagged corrupt, so the packet joins the partial rather than opening a // new PES out of bytes the demodulator already disowned. Continuation::Corrupt => { - self.pending.remove(&pid); - if let Some(stream) = self.streams.get_mut(&pid) { - stream.desync(); - } + self.damage(pid, &anyhow::anyhow!("transport error indicator")); continue; } Continuation::Broken => { @@ -435,15 +444,18 @@ impl Import { 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(); - } + self.damage(pid, &anyhow::anyhow!("malformed TS adaptation field")); return Ok(()); } }; if pkt[1] & 0x40 != 0 { - self.handle_pes_start(pid, PesStart::parse(payload)?) + match PesStart::parse(payload) { + Ok(pes) => self.handle_pes_start(pid, pes), + Err(err) => { + self.damage(pid, &err); + Ok(()) + } + } } else { self.handle_pes_continuation(pid, payload) } @@ -473,6 +485,7 @@ impl Import { import: Box::new(h264::Import::new(track, self.reserve(), self.video_hint())?), unwrap: PtsUnwrap::default(), reanchor: Reanchor::default(), + waiting: false, } } Some(StreamType::H265) => { @@ -484,6 +497,7 @@ impl Import { import: Box::new(h265::Import::new(track, self.reserve(), self.video_hint())?), unwrap: PtsUnwrap::default(), reanchor: Reanchor::default(), + waiting: false, } } // Only ADTS-framed AAC (0x0F). 0x11 is LATM/LOAS, which uses a different @@ -719,6 +733,16 @@ impl Import { Ok(()) } + /// Refuse one damaged unit without changing any other PID or the program clock. + fn damage(&mut self, pid: Pid, err: &anyhow::Error) { + self.pending.remove(&pid); + if let Some(stream) = self.streams.get_mut(&pid) { + stream.desync(); + } + *self.damaged.entry(pid.as_u16()).or_default() += 1; + tracing::warn!(pid = pid.as_u16(), error = %err, "dropped a damaged TS unit"); + } + fn flush(&mut self, pid: Pid) -> anyhow::Result<()> { let Some(pending) = self.pending.remove(&pid) else { return Ok(()); @@ -731,7 +755,14 @@ impl Import { let Some(stream) = self.streams.get_mut(&pid) else { return Ok(()); }; - let units = stream.write(pending, batched, self.anchor.as_mut())?; + let units = match stream.write(pending, batched, self.anchor.as_mut()) { + Ok(units) => units, + Err(err) if err.is::() => { + self.damage(pid, &err); + return Ok(()); + } + Err(err) => return Err(err), + }; self.published |= units > 0; self.liveness.delivered(pid.as_u16(), units); @@ -922,7 +953,7 @@ impl Import { } /// 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. + /// way, damaged units refused, 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. @@ -941,6 +972,7 @@ impl Import { } for (pid, stats) in &mut streams { (stats.units, stats.quiet) = self.liveness.stream(*pid); + stats.damaged = self.damaged.get(pid).copied().unwrap_or_default(); } Stats { streams, @@ -1050,6 +1082,8 @@ pub struct StreamStats { /// confirm it. That substitutes audio rather than leaving a gap, which is why it is /// counted separately from a resync. pub unconfirmed: u64, + /// Damaged packets, PES, or access units refused whole on this PID. Export stays zero. + pub damaged: u64, } impl StreamStats { @@ -1061,7 +1095,7 @@ impl StreamStats { } /// Add a newer route's frame-sync counters while naming the route that is active now. - /// Delivery is kept per PID rather than per route, so it needs no merging. + /// Delivery and damage are kept per PID rather than per route, so they need no merging. fn merge(&mut self, current: &Self) { self.track = current.track; self.resyncs += current.resyncs; @@ -1660,6 +1694,30 @@ impl PatReader { } } +/// Whether every adaptation field fits inside the length its packet declares. +fn adaptation_valid(pkt: &[u8; TsPacket::SIZE]) -> bool { + if pkt[3] & 0x20 == 0 { + return true; + } + let Some(field) = pkt.get(5..5 + usize::from(pkt[4])) else { + return false; + }; + let Some(&flags) = field.first() else { + return true; + }; + let mut off = + 1 + usize::from(flags & 0x10 != 0) * 6 + usize::from(flags & 0x08 != 0) * 6 + usize::from(flags & 0x04 != 0); + for flag in [0x02, 0x01] { + if flags & flag != 0 { + let Some(&len) = field.get(off) else { + return false; + }; + off += 1 + usize::from(len); + } + } + off <= field.len() +} + /// Where a packet's payload is. enum Payload<'a> { /// Adaptation field only, or reserved `adaptation_field_control`. @@ -1796,6 +1854,39 @@ impl SectionReassembler { } } +/// A parse failure confined to a unit, distinguished from publishing and clock failures. +#[derive(Debug, thiserror::Error)] +#[error("{0}")] +struct Damaged(anyhow::Error); + +fn unit_error(err: crate::Error) -> anyhow::Error { + let damaged = matches!( + &err, + crate::Error::Annexb(_) + | crate::Error::H264( + h264::Error::NalTooShort + | h264::Error::ForbiddenZeroBit + | h264::Error::SpsTooShort + | h264::Error::SpsParse + | h264::Error::NotInitialized + ) | crate::Error::Aac( + aac::Error::ProgramConfigMissing | aac::Error::ProgramConfigTruncated | aac::Error::ProgramConfigEmpty + ) | crate::Error::H265( + h265::Error::NalTooShort + | h265::Error::ForbiddenZeroBit + | h265::Error::SpsParse + | h265::Error::MissingLevelIdc + | h265::Error::NotInitialized + | h265::Error::MissingSps + ) + ); + if damaged { + Damaged(err.into()).into() + } else { + err.into() + } +} + /// One elementary stream's codec importer plus PTS-unwrap state. enum Stream { H264 { @@ -1803,12 +1894,14 @@ enum Stream { import: Box, unwrap: PtsUnwrap, reanchor: Reanchor, + waiting: bool, }, H265 { split: h265::Split, import: Box, unwrap: PtsUnwrap, reanchor: Reanchor, + waiting: bool, }, Aac(Box>), Opus(Box), @@ -1835,16 +1928,21 @@ impl Stream { import, unwrap, reanchor, + waiting, } => { let reorder = reorder_delay(pending.pts, pending.dts); let pts = unwrap_pts(unwrap, pending.pts, anchor)?; // Each PES is one access unit, so flush to emit it immediately. - let mut frames = split.decode(&pending.data, pts)?; - frames.extend(split.flush(pts)?); + let mut frames = split.decode(&pending.data, pts).map_err(unit_error)?; + frames.extend(split.flush(pts).map_err(unit_error)?); let mut published = 0; for mut frame in frames { + if *waiting && !frame.keyframe { + continue; + } frame.timestamp = reanchor.apply(frame.timestamp, import.floor(frame.keyframe))?; published += u64::from(skip_missing_keyframe(import.decode([frame]))?); + *waiting = false; } // After decode, so the track (and its catalog rendition) exists. if let Some(reorder) = reorder { @@ -1857,16 +1955,21 @@ impl Stream { import, unwrap, reanchor, + waiting, } => { let reorder = reorder_delay(pending.pts, pending.dts); let pts = unwrap_pts(unwrap, pending.pts, anchor)?; // Each PES is one access unit, so flush to emit it immediately. - let mut frames = split.decode(&pending.data, pts)?; - frames.extend(split.flush(pts)?); + let mut frames = split.decode(&pending.data, pts).map_err(unit_error)?; + frames.extend(split.flush(pts).map_err(unit_error)?); let mut published = 0; for mut frame in frames { + if *waiting && !frame.keyframe { + continue; + } frame.timestamp = reanchor.apply(frame.timestamp, import.floor(frame.keyframe))?; published += u64::from(skip_missing_keyframe(import.decode([frame]))?); + *waiting = false; } if let Some(reorder) = reorder { import.observe_reorder(reorder)?; @@ -1919,8 +2022,14 @@ impl Stream { /// break the stream recovers from in place. fn desync(&mut self) { match self { - Stream::H264 { split, .. } => split.reset(), - Stream::H265 { split, .. } => split.reset(), + Stream::H264 { split, waiting, .. } => { + split.reset(); + *waiting = true; + } + Stream::H265 { split, waiting, .. } => { + split.reset(); + *waiting = true; + } Stream::Aac(stream) => stream.desync(), Stream::Legacy(stream) => stream.desync(), Stream::Opus(_) | Stream::Verbatim(_) | Stream::Clock | Stream::Ignored => {} @@ -2429,7 +2538,8 @@ impl AacStream { header.sample_rate, header.channel_config, &mut block, - )?; + ) + .map_err(|err| unit_error(err.into()))?; let mut config = aac::config(&asc)?; config.container = self.container.clone(); // Consume the reservation held since the PMT: this resolves the gated rendition, @@ -2564,18 +2674,11 @@ impl OpusStream { let edge = self.import.live_edge(); let base = base.map(|base| self.reanchor.apply(base, edge)).transpose()?; - let data = &pending.data; - let mut offset = 0; + let packets = opus_packets(&pending.data).map_err(Damaged)?; let mut published = 0; // 48 kHz samples elapsed since this PES's PTS, advancing each packet after the first. let mut elapsed: u64 = 0; - while offset < data.len() { - let (header_len, size) = parse_opus_control_header(&data[offset..])?; - let start = offset + header_len; - let end = start + size; - anyhow::ensure!(end <= data.len(), "Opus access unit exceeds PES payload"); - let packet = &data[start..end]; - + for packet in packets { let pts = match base { Some(base) if elapsed > 0 => { let advance = Timestamp::from_scale(elapsed, 48_000)?; @@ -2593,7 +2696,6 @@ impl OpusStream { // Default to 20 ms (960 samples) if the TOC can't be read, so a malformed packet // doesn't stall the timeline for the rest of the PES. elapsed += opus::packet_samples(packet).unwrap_or(960) as u64; - offset = end; published += 1; } Ok(published) @@ -2659,6 +2761,21 @@ fn opus_config(descriptors: &[catalog::Descriptor]) -> crate::Result anyhow::Result> { + let mut packets = Vec::new(); + while !data.is_empty() { + let (header_len, size) = parse_opus_control_header(data)?; + let end = header_len + size; + let packet = data + .get(header_len..end) + .context("Opus access unit exceeds PES payload")?; + packets.push(packet); + data = &data[end..]; + } + Ok(packets) +} + /// Parse one Opus-in-TS access-unit control header, returning `(header_len, payload_size)`. fn parse_opus_control_header(data: &[u8]) -> anyhow::Result<(usize, usize)> { anyhow::ensure!(data.len() >= 2, "Opus control header truncated"); @@ -2996,7 +3113,7 @@ fn skip_missing_keyframe(result: crate::Result<()>) -> anyhow::Result { match result { Ok(()) => Ok(true), Err(crate::Error::MissingKeyframe(_)) => Ok(false), - Err(e) => Err(e.into()), + Err(e) => Err(unit_error(e)), } } @@ -3137,7 +3254,8 @@ pub(super) mod test { use moq_net::Timestamp; use mpeg2ts::es::StreamType; - use super::{Continuation, Continuity, SectionReassembler}; + use super::{Continuation, Continuity, SectionReassembler, Stream}; + use mpeg2ts::ts::{Pid, TsPacket}; /// A drift budget no test timeline comes close to, so the reader sees every group. /// @@ -3224,7 +3342,9 @@ pub(super) mod test { legacy.resync.recover(&[0; 10], 0, &codec); legacy.resync.published(false); import.streams.insert(pid, stream); + import.damage(pid, &anyhow::anyhow!("malformed PES")); let before = import.stats(); + assert_eq!(before.streams[&pid.as_u16()].damaged, 1); import .ensure_section(pid, StreamType::Dts8ChannelLosslessAudio as u8, &[]) @@ -4301,6 +4421,7 @@ pub(super) mod test { resyncs: 1, discarded: 47, unconfirmed: 0, + damaged: 0, } )]), "the resync left no trace an operator could alarm on" @@ -4494,6 +4615,7 @@ pub(super) mod test { resyncs: 1, discarded: 72, unconfirmed: 0, + damaged: 0, } )]), "the resync left no trace an operator could alarm on" @@ -4697,6 +4819,7 @@ pub(super) mod test { resyncs: 0, discarded: 0, unconfirmed: 1, + damaged: 0, } )]), "the drained frame was published without a trace" @@ -6107,6 +6230,223 @@ pub(super) mod test { const VIDEO: u16 = 0x0050; + /// Damage on one PID refuses that unit and its dependent pictures while another PID + /// carries on. A later decode call resumes from the next keyframe. + async fn damaged_video_recovers(damage: &str) { + let peer = VIDEO + 1; + let mut mux = Mux { + out: synth_pmt(&[(StreamType::H264, VIDEO), (StreamType::H264, peer)], false), + ..Default::default() + }; + mux.gop(VIDEO, 90_000); + mux.gop(peer, 90_000); + let damaged_at = mux.out.len(); + mux.gop(VIDEO, 90_000 + 4 * FRAME); + mux.gop(peer, 90_000 + 4 * FRAME); + let packet = &mut mux.out[damaged_at..damaged_at + TsPacket::SIZE]; + let payload_at = 5 + usize::from(packet[4]); + match damage { + "pes" => packet[payload_at + 6..payload_at + 19].fill(0), + "nal" => { + let nal = packet + .windows(5) + .position(|bytes| bytes[..4] == [0, 0, 0, 1] && bytes[4] & 0x1f == 5) + .unwrap(); + packet[nal + 4] |= 0x80; + } + "late-nal" => { + let mut au = annexb_au(true); + au.extend_from_slice(&[0, 0, 0, 1, 0xe1, 0x80]); + packet.copy_from_slice(&video_pes(VIDEO, 4, 90_000 + 5 * FRAME, Some(90_000 + 4 * FRAME), &au)); + } + "adaptation" => packet[4] = 255, + "adaptation-clock" => { + packet[4] = 1; + packet[5] = 0x90; + } + "tei" => packet[1] |= 0x80, + "clean" => {} + _ => unreachable!(), + } + assert_eq!( + packet[1] & 0x80 != 0, + damage == "tei", + "only flagged corruption sets TEI" + ); + let resume_at = mux.out.len(); + mux.gop(VIDEO, 90_000 + 8 * FRAME); + mux.gop(peer, 90_000 + 8 * FRAME); + + 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()).unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + import + .decode(&mux.out[..resume_at]) + .expect("damage must stay local to the unit"); + assert!(import.scratch.is_empty(), "every damaged packet must be consumed"); + assert!( + !import.pending.contains_key(&Pid::new(VIDEO).unwrap()), + "no damaged PES remains" + ); + import + .decode(&mux.out[resume_at..]) + .expect("next keyframe must recover"); + import.finish().unwrap(); + assert_eq!(import.stats().streams[&VIDEO].damaged, u64::from(damage != "clean")); + assert_eq!(import.stats().streams[&peer].damaged, 0); + for (pid, expected) in [(VIDEO, if damage == "clean" { 12 } else { 8 }), (peer, 12)] { + let Stream::H264 { import: video, .. } = &import.streams[&Pid::new(pid).unwrap()] else { + panic!("video route"); + }; + let frames = read_track(&consumer, video.name(), crate::container::Kind::Video).await; + assert_eq!(frames.len(), expected, "published pictures on PID {pid}"); + if pid == VIDEO && damage != "clean" { + assert!(frames[4].keyframe, "recovery starts at a keyframe"); + assert_eq!( + frames[4].timestamp.as_micros(), + u128::from(90_000 + 9 * FRAME) * 1_000_000 / 90_000 + ); + } + } + } + + #[tokio::test(start_paused = true)] + async fn damaged_pes_header_recovers() { + damaged_video_recovers("pes").await; + } + + #[tokio::test(start_paused = true)] + async fn damaged_h264_nal_recovers() { + damaged_video_recovers("nal").await; + } + + #[tokio::test(start_paused = true)] + async fn damaged_trailing_nal_refuses_the_whole_access_unit() { + damaged_video_recovers("late-nal").await; + } + + #[tokio::test(start_paused = true)] + async fn damaged_adaptation_recovers() { + damaged_video_recovers("adaptation").await; + } + + #[tokio::test(start_paused = true)] + async fn damaged_adaptation_cannot_reset_the_program_clock() { + damaged_video_recovers("adaptation-clock").await; + } + + #[tokio::test(start_paused = true)] + async fn damaged_tei_recovers() { + damaged_video_recovers("tei").await; + } + + #[tokio::test(start_paused = true)] + async fn clean_video_has_no_damage() { + damaged_video_recovers("clean").await; + } + + /// A malformed unbounded PES drained at EOF is still refused and counted. + #[test] + fn damaged_final_access_unit_is_refused() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + let mut au = annexb_au(true); + let nal = au + .windows(5) + .position(|bytes| bytes[..4] == [0, 0, 0, 1] && bytes[4] & 0x1f == 5) + .unwrap(); + au[nal + 4] |= 0x80; + let mut packet = video_pes(VIDEO, 0, 90_000, None, &au); + let start = 5 + usize::from(packet[4]); + packet[start + 4..start + 6].fill(0); + let mut data = synth_pmt(&[(StreamType::H264, VIDEO)], false); + data.extend(packet); + import.decode(&data).unwrap(); + assert_eq!(import.stats().streams[&VIDEO].damaged, 0, "PES has not ended yet"); + import.finish().unwrap(); + assert_eq!(import.stats().streams[&VIDEO].damaged, 1); + assert_eq!(import.stats().streams[&VIDEO].units, 0); + assert!(catalog.snapshot().video.renditions.is_empty()); + } + + /// An invalid trailing Opus control header refuses the complete PES before its first + /// packet is published. Independently decodable packets in the next PES recover. + #[tokio::test(start_paused = true)] + async fn damaged_opus_pes_is_not_half_published() { + 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()).unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + import + .ensure_stream( + Pid::new(VIDEO).unwrap(), + 0x06, + &[super::catalog::Descriptor { + tag: 0x05, + data: bytes::Bytes::from_static(b"Opus"), + }], + ) + .unwrap(); + let good = [0x7f, 0xe0, 3, 0xf8, 0xff, 0xfe]; + let mut bad = good.to_vec(); + bad.push(0x7f); + let mut data = Vec::new(); + for (cc, payload) in [(0, good.as_slice()), (1, bad.as_slice()), (2, good.as_slice())] { + data.extend(video_pes(VIDEO, cc, 90_000 + u64::from(cc) * FRAME, None, payload)); + } + import.decode(&data).unwrap(); + import.finish().unwrap(); + assert_eq!(import.stats().streams[&VIDEO].damaged, 1); + assert_eq!(import.stats().streams[&VIDEO].units, 2); + let name = catalog.snapshot().audio.renditions.keys().next().unwrap().clone(); + let frames = read_track(&consumer, &name, crate::container::Kind::Audio).await; + assert_eq!(frames.len(), 2, "no packet from the damaged PES is published"); + } + + /// An aborted track is a feed-wide publishing failure, never a damaged media unit. + #[test] + fn damaged_recovery_keeps_publishing_errors_fatal() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let track = broadcast + .create_track("video", hang::container::track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + let video = crate::codec::h264::Import::new(track.clone(), catalog.reserve(), Default::default()).unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + import.streams.insert( + Pid::new(VIDEO).unwrap(), + Stream::H264 { + split: crate::codec::h264::Split::new(), + import: Box::new(video), + unwrap: Default::default(), + reanchor: Default::default(), + waiting: false, + }, + ); + track.abort(moq_net::Error::Closed).unwrap(); + let mut data = Vec::new(); + for cc in 0..3 { + data.extend(video_pes( + VIDEO, + cc, + 90_000 + u64::from(cc) * FRAME, + None, + &annexb_au(true), + )); + } + let err = import.decode(&data).expect_err("closed producer must end ingest"); + assert!( + matches!(err.downcast_ref::(), Some(crate::Error::Moq(_))), + "{err:?}" + ); + assert!(import.damaged.is_empty(), "publishing failure is not local damage"); + assert!( + !super::unit_error(crate::Error::H264(crate::codec::h264::Error::MissingTimestamp)).is::() + ); + } + /// #3798's join: a new IDR presents below the last P-frame on a continuous clock, since /// the B-frames before it presented earlier than the P-frame decoded ahead of them. #[tokio::test(start_paused = true)] diff --git a/rs/moq-mux/src/container/ts/stats.rs b/rs/moq-mux/src/container/ts/stats.rs index 519c62f180..d5e8bc2abd 100644 --- a/rs/moq-mux/src/container/ts/stats.rs +++ b/rs/moq-mux/src/container/ts/stats.rs @@ -28,7 +28,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, any stream whose frame-sync counters moved, and PSI sections dropped since. + /// silence, any stream whose loss counters moved, and PSI sections dropped since. pub fn sample(&mut self, latest: Stats) { self.sync(&latest); for (pid, stream) in &latest.streams { @@ -56,7 +56,7 @@ impl Log { self.sync(latest); } - /// Report the streams whose frame-sync counters moved, one line each, and the PSI + /// Report the streams whose damage or frame-sync counters moved, and the PSI /// sections dropped for a bad CRC if that count moved. /// /// The importer already warns on each individual resync and dropped section; this is the @@ -71,6 +71,14 @@ impl Log { 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)); + if previous.map_or(0, |previous| previous.damaged) != stream.damaged { + tracing::info!( + pid = *pid, + track = stream.track, + damaged = stream.damaged, + "damaged TS units refused" + ); + } if previous.map_or((0, 0, 0), lost) == lost(stream) { continue; } @@ -189,4 +197,51 @@ mod test { } }); } + + /// Damage is reported per PID when its cumulative count moves, including the final sample. + #[test] + #[tracing_test::traced_test] + fn reports_damaged_units_when_they_move() { + let sample = |damaged| { + let mut stats = Stats::default(); + stats.streams.insert( + 256, + StreamStats { + track: ".avc3", + damaged, + ..Default::default() + }, + ); + stats.streams.insert( + 257, + StreamStats { + track: ".mp2", + ..Default::default() + }, + ); + stats + }; + let mut log = Log::default(); + log.sample(sample(0)); + log.sample(sample(1)); + log.sample(sample(1)); + log.finish(&sample(2)); + logs_assert(|lines: &[&str]| { + let refused: Vec<_> = lines + .iter() + .filter(|line| line.contains("damaged TS units refused")) + .collect(); + match refused.as_slice() { + [first, second] + if first.contains("pid=256") + && first.contains("damaged=1") + && second.contains("pid=256") + && second.contains("damaged=2") => + { + Ok(()) + } + _ => Err(format!("expected damage increases on the video PID, got {refused:?}")), + } + }); + } } From ad4923b0d8d73790f76ebad3cf567658aef78ac7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 3 Oct 2026 11:16:56 -0700 Subject: [PATCH 3/7] fix(mux): drop queued frames on split reset; refuse bad adaptation after duplicates A splitter reset now clears frames completed before a failed decode, so a refused PES cannot surface its picture on the next call. The malformed adaptation check runs after duplicate detection and salvages across a gap. Co-Authored-By: Claude Opus 5.5 --- doc/bin/srt.md | 3 ++- rs/moq-mux/src/codec/h264/split.rs | 8 ++++--- rs/moq-mux/src/codec/h265/split.rs | 8 ++++--- rs/moq-mux/src/container/ts/import.rs | 34 ++++++++++++++++++++------- 4 files changed, 37 insertions(+), 16 deletions(-) diff --git a/doc/bin/srt.md b/doc/bin/srt.md index c149e1cd12..12828b9c49 100644 --- a/doc/bin/srt.md +++ b/doc/bin/srt.md @@ -13,7 +13,8 @@ same per-stream lines as `import ts`, under an `srt{path=...}` span: an elementary stream that stopped delivering access units, audio frame sync lost, damaged units refused on each PID, and the TR 101 290 counters when one moves. They grade the TS as SRT delivered it, after retransmission. Damage drops that -unit and keeps the session alive; video resumes at its next keyframe. +unit and keeps the session alive; video closes its group at the break and +resumes at its next keyframe, freezing for up to one GOP. ```bash # Accept a contribution feed and publish it diff --git a/rs/moq-mux/src/codec/h264/split.rs b/rs/moq-mux/src/codec/h264/split.rs index cbbcac9a8e..8f6ba4d6ee 100644 --- a/rs/moq-mux/src/codec/h264/split.rs +++ b/rs/moq-mux/src/codec/h264/split.rs @@ -224,14 +224,16 @@ impl Split { Ok(()) } - /// Drop any in-flight access unit. + /// Drop any in-flight access unit, and any frame completed but not yet returned. /// /// Pre-reset NALs would otherwise leak into a later frame with the wrong - /// timestamp. The parameter-set cache is kept so subsequent keyframes stay - /// self-contained. + /// timestamp, and a frame queued before a failed [`decode`](Self::decode) would + /// surface on the next call. The parameter-set cache is kept so subsequent + /// keyframes stay self-contained. pub fn reset(&mut self) { self.current = Avc3Frame::default(); self.tail.clear(); + self.pending.clear(); } fn pts(&mut self, hint: Option) -> Result { diff --git a/rs/moq-mux/src/codec/h265/split.rs b/rs/moq-mux/src/codec/h265/split.rs index 42951f60fc..9d114416e6 100644 --- a/rs/moq-mux/src/codec/h265/split.rs +++ b/rs/moq-mux/src/codec/h265/split.rs @@ -228,14 +228,16 @@ impl Split { Ok(()) } - /// Drop any in-flight access unit. + /// Drop any in-flight access unit, and any frame completed but not yet returned. /// /// Pre-reset NALs would otherwise leak into a later frame with the wrong - /// timestamp. The parameter-set cache is kept so subsequent keyframes stay - /// self-contained. + /// timestamp, and a frame queued before a failed [`decode`](Self::decode) would + /// surface on the next call. The parameter-set cache is kept so subsequent + /// keyframes stay self-contained. pub fn reset(&mut self) { self.current = Au::default(); self.tail.clear(); + self.pending.clear(); } fn pts(&mut self, hint: Option) -> Result { diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index a04143a6e7..5549edcb4e 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -232,24 +232,28 @@ impl Import { // Every packet is graded, but only the media PIDs and the clock route on the // verdict: PSI and section PIDs keep their reassemblers' own. let continuation = self.health.packet(&pkt, self.liveness.now()); + let continuation = Pid::new(pid) + .ok() + .filter(|pid| self.streams.contains_key(pid) || self.pcr_pid == Some(*pid)) + .map(|_| continuation); + // A retransmitted payload must not repeat the clock reset it carried. + if matches!(continuation, Some(Continuation::Duplicate)) { + continue; + } // A media packet whose adaptation field overruns itself is refused before its - // clock bits can reach the program clock. Its counter already joined the chain. + // clock bits can reach the program clock. Its counter already joined the chain, + // so a gap in front of it still salvages what came before. if let Ok(pid) = Pid::new(pid) && self.streams.contains_key(&pid) && pkt[1] & 0x80 == 0 && !adaptation_valid(&pkt) { + if matches!(continuation, Some(Continuation::Broken)) { + self.salvage(pid)?; + } self.damage(pid, &anyhow::anyhow!("malformed TS adaptation field"))?; continue; } - let continuation = Pid::new(pid) - .ok() - .filter(|pid| self.streams.contains_key(pid) || self.pcr_pid == Some(*pid)) - .map(|_| continuation); - // A retransmitted payload must not repeat the clock reset it carried. - if matches!(continuation, Some(Continuation::Duplicate)) { - 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 nothing below routes, and // the flag rides an adaptation-only packet as readily as a payload one. @@ -6056,6 +6060,13 @@ pub(super) mod test { au.extend_from_slice(&[0, 0, 0, 1, 0xe1, 0x80]); packet.copy_from_slice(&video_pes(VIDEO, 4, 90_000 + 5 * FRAME, Some(90_000 + 4 * FRAME), &au)); } + "queued" => { + // The AUD completes the IDR before the forbidden NAL fails, so the splitter + // has already queued it when the unit is refused. + let mut au = annexb_au(true); + au.extend_from_slice(&[0, 0, 0, 1, 0x09, 0xf0, 0, 0, 0, 1, 0xe1, 0x80, 0, 0, 0, 1, 0x41, 0x9a]); + packet.copy_from_slice(&video_pes(VIDEO, 4, 90_000 + 5 * FRAME, Some(90_000 + 4 * FRAME), &au)); + } "adaptation" => packet[4] = 255, "adaptation-clock" => { packet[4] = 1; @@ -6123,6 +6134,11 @@ pub(super) mod test { damaged_video_recovers("late-nal").await; } + #[tokio::test(start_paused = true)] + async fn damaged_unit_drops_a_picture_the_splitter_queued() { + damaged_video_recovers("queued").await; + } + #[tokio::test(start_paused = true)] async fn damaged_adaptation_recovers() { damaged_video_recovers("adaptation").await; From 7287c476a9c33bd6014e3cbceae90027ff6183ac Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 3 Oct 2026 11:38:36 -0700 Subject: [PATCH 4/7] fix(mux): refuse a malformed adaptation field on a dedicated PCR PID Its discontinuity bit could otherwise declare a timebase break and reset every stream. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/import.rs | 32 +++++++++++++++++++++++---- 1 file changed, 28 insertions(+), 4 deletions(-) diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 5549edcb4e..f209090b57 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -240,11 +240,11 @@ impl Import { if matches!(continuation, Some(Continuation::Duplicate)) { continue; } - // A media packet whose adaptation field overruns itself is refused before its - // clock bits can reach the program clock. Its counter already joined the chain, - // so a gap in front of it still salvages what came before. + // A media or PCR packet whose adaptation field overruns itself is refused before + // its clock bits can reach the program clock. Its counter already joined the + // chain, so a gap in front of it still salvages what came before. if let Ok(pid) = Pid::new(pid) - && self.streams.contains_key(&pid) + && (self.streams.contains_key(&pid) || self.pcr_pid == Some(pid)) && pkt[1] & 0x80 == 0 && !adaptation_valid(&pkt) { @@ -6159,6 +6159,30 @@ pub(super) mod test { damaged_video_recovers("clean").await; } + /// A dedicated PCR PID's malformed adaptation field cannot declare a timebase break. + #[test] + fn damaged_adaptation_on_a_dedicated_pcr_pid_keeps_the_clock() { + const PCR: u16 = 0x0100; + let mut mux = Mux { + out: synth_pmt(&[(StreamType::H264, VIDEO)], false), + ..Default::default() + }; + mux.gop(VIDEO, 90_000); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + import.decode(&mux.out).unwrap(); + import.pcr_pid = Some(Pid::new(PCR).unwrap()); + assert!(import.last_pts.is_some()); + + // The flag byte is the field's only byte, so the PCR it announces overruns it. + let mut packet = clock_break_packet(PCR); + packet[4] = 1; + packet[5] = 0x90; + import.decode(&packet).unwrap(); + assert!(import.last_pts.is_some(), "a malformed field reset the program clock"); + } + /// A malformed unbounded PES drained at EOF is still refused and counted. #[test] fn damaged_final_access_unit_is_refused() { From 5822b1d0d78cbd5259342b27780a2709a56d9486 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 3 Oct 2026 11:56:38 -0700 Subject: [PATCH 5/7] fix(mux): report a dedicated PCR PID's damage in the stats snapshot A PCR-only PID gets a clock-only row carrying its damaged count. An ignored PID imports nothing, so damage on it is neither refused nor counted. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/import.rs | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index f209090b57..f8ec9c95e1 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -713,8 +713,11 @@ impl Import { /// Refuse one damaged unit without changing any other PID or the program clock. fn damage(&mut self, pid: Pid, err: &anyhow::Error) -> anyhow::Result<()> { self.pending.remove(&pid); - if let Some(stream) = self.streams.get_mut(&pid) { - stream.lost()?; + match self.streams.get_mut(&pid) { + // Nothing is imported from it, so there is nothing to refuse or report. + Some(Stream::Ignored) => return Ok(()), + Some(stream) => stream.lost()?, + None => {} } *self.damaged.entry(pid.as_u16()).or_default() += 1; tracing::warn!(pid = pid.as_u16(), error = %err, "dropped a damaged TS unit"); @@ -958,9 +961,12 @@ impl Import { .and_modify(|retired| retired.merge(¤t)) .or_insert(current); } + // A dedicated PCR PID routes no stream, so its damage gets a clock-only row. + for (&pid, &damaged) in &self.damaged { + streams.entry(pid).or_insert_with(|| StreamStats::new("")).damaged = damaged; + } for (pid, stats) in &mut streams { (stats.units, stats.quiet) = self.liveness.stream(*pid); - stats.damaged = self.damaged.get(pid).copied().unwrap_or_default(); } let mut stats = Stats { streams, @@ -6181,6 +6187,7 @@ pub(super) mod test { packet[5] = 0x90; import.decode(&packet).unwrap(); assert!(import.last_pts.is_some(), "a malformed field reset the program clock"); + assert_eq!(import.stats().streams[&PCR].damaged, 1); } /// A malformed unbounded PES drained at EOF is still refused and counted. From 15afe24683243cbad17f84b816d8cc6d5c3c0df4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 3 Oct 2026 12:27:09 -0700 Subject: [PATCH 6/7] test(mux): put the dedicated PCR PID clear of the PMT PID Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/import.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index f8ec9c95e1..dd01852e09 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -6168,7 +6168,8 @@ pub(super) mod test { /// A dedicated PCR PID's malformed adaptation field cannot declare a timebase break. #[test] fn damaged_adaptation_on_a_dedicated_pcr_pid_keeps_the_clock() { - const PCR: u16 = 0x0100; + // Clear of the PMT PID and every stream, so only the clock-only row can carry it. + const PCR: u16 = 0x0200; let mut mux = Mux { out: synth_pmt(&[(StreamType::H264, VIDEO)], false), ..Default::default() @@ -6187,7 +6188,8 @@ pub(super) mod test { packet[5] = 0x90; import.decode(&packet).unwrap(); assert!(import.last_pts.is_some(), "a malformed field reset the program clock"); - assert_eq!(import.stats().streams[&PCR].damaged, 1); + let stats = import.stats(); + assert_eq!((stats.streams[&PCR].track, stats.streams[&PCR].damaged), ("", 1)); } /// A malformed unbounded PES drained at EOF is still refused and counted. From 9f94d85dbef3f15287f922160e3e945a0b73986c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 3 Oct 2026 12:43:33 -0700 Subject: [PATCH 7/7] fix(mux): a damaged PES start still flushes the PES before it The start bit ends the previous unbounded PES whole, so only the new unit is refused. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/import.rs | 35 ++++++++++++++++++++++++++- 1 file changed, 34 insertions(+), 1 deletion(-) diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index dd01852e09..fa2cbebd59 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -450,7 +450,11 @@ impl Import { if pkt[1] & 0x40 != 0 { match PesStart::parse(payload) { Ok(pes) => self.handle_pes_start(pid, pes), - Err(err) => self.damage(pid, &err), + Err(err) => { + // The start still ends the PES before it, which is whole. + self.flush(pid)?; + self.damage(pid, &err) + } } } else { self.handle_pes_continuation(pid, payload) @@ -6192,6 +6196,35 @@ pub(super) mod test { assert_eq!((stats.streams[&PCR].track, stats.streams[&PCR].damaged), ("", 1)); } + /// A damaged PES start refuses only its own unit: the unbounded PES before it ends there, + /// whole, and still publishes. + #[test] + fn damaged_pes_start_flushes_the_unit_before_it() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + let unbounded = |cc, pts, keyframe| { + let mut packet = video_pes(VIDEO, cc, pts, None, &annexb_au(keyframe)); + let start = 5 + usize::from(packet[4]); + packet[start + 4..start + 6].fill(0); + packet + }; + let mut data = synth_pmt(&[(StreamType::H264, VIDEO)], false); + data.extend(unbounded(0, 90_000, true)); + let mut damaged = unbounded(1, 90_000 + FRAME, false); + let start = 5 + usize::from(damaged[4]); + damaged[start + 6..start + 19].fill(0); + data.extend(damaged); + import.decode(&data).unwrap(); + import.finish().unwrap(); + assert_eq!(import.stats().streams[&VIDEO].damaged, 1); + assert_eq!( + import.stats().streams[&VIDEO].units, + 1, + "the keyframe before the damage" + ); + } + /// A malformed unbounded PES drained at EOF is still refused and counted. #[test] fn damaged_final_access_unit_is_refused() {