From a254ac120b1f1ce32126759877920dbc049d8909 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 15:15:00 -0700 Subject: [PATCH 1/3] fix(mux): refuse damaged TS units without ending ingest (backport #4733) One malformed PES header, adaptation field, or codec access unit no longer ends MPEG-TS/SRT ingest. The damaged unit is refused and logged, H.264/H.265 close their group at the break and resume at the next keyframe, and every other PID carries on. Publishing, catalog, and clock errors stay fatal. Ported from d8b9b0031 without the TR 101 290 Health counters (#4750) and the StreamStats::damaged field, which release lacks. Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 8 + doc/bin/srt.md | 6 +- 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 | 553 ++++++++++++++++++++++++-- 5 files changed, 537 insertions(+), 46 deletions(-) diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 45fbbb909d..e18de406be 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -79,6 +79,14 @@ 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 logged as a warning. Ingest continues on every other PID. +Video closes its group at the break and resumes at its next keyframe, as it does +after a continuity-counter gap: the pictures in between may reference the lost +one, so they are dropped rather than decoded with artefacts. Each break freezes +video for up to one GOP. 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..26e1b046cf 100644 --- a/doc/bin/srt.md +++ b/doc/bin/srt.md @@ -10,8 +10,10 @@ source (`--connect`); `moq export srt` serves SRT to players or pushes to a 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. +elementary stream that stopped delivering access units, audio frame sync lost, +and damaged units refused. Damage drops that 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 fb00d66b82..f79784ab61 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -249,6 +249,20 @@ impl Import { if matches!(continuation, Some(Continuation::Duplicate)) { continue; } + // 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.pcr_pid == Some(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; + } // 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. @@ -289,10 +303,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 +446,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(); - } - return Ok(()); + return self.damage(pid, &anyhow::anyhow!("malformed TS adaptation field")); } }; 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) => { + // 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) } @@ -719,6 +733,19 @@ 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) -> anyhow::Result<()> { + self.pending.remove(&pid); + 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 => {} + } + tracing::warn!(pid = pid.as_u16(), error = %err, "dropped a damaged TS unit"); + Ok(()) + } + fn flush(&mut self, pid: Pid) -> anyhow::Result<()> { let Some(pending) = self.pending.remove(&pid) else { return Ok(()); @@ -731,7 +758,11 @@ 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::() => return self.damage(pid, &err), + Err(err) => return Err(err), + }; self.published |= units > 0; self.liveness.delivered(pid.as_u16(), units); @@ -740,19 +771,25 @@ impl Import { self.record_media_track(pid) } - /// The packet chain on `pid` was cut: salvage the truncated PES only where its bytes - /// stand on their own, drop whatever is left mid-unit, and require the next frame to - /// prove its boundary. + /// The packet chain on `pid` was cut: salvage the truncated PES, and treat the bytes + /// after it as [lost](Stream::lost). fn broken(&mut self, pid: Pid) -> anyhow::Result<()> { + self.salvage(pid)?; + if let Some(stream) = self.streams.get_mut(&pid) { + stream.lost()?; + } + Ok(()) + } + + /// Publish the truncated PES on `pid` only where its bytes stand on their own, and drop + /// whatever is left mid-unit. + fn salvage(&mut self, pid: Pid) -> anyhow::Result<()> { if self.streams.get(&pid).is_some_and(Stream::salvages_partial_pes) { - self.flush(pid)?; + self.flush(pid) } else { self.pending.remove(&pid); + Ok(()) } - if let Some(stream) = self.streams.get_mut(&pid) { - stream.desync(); - } - Ok(()) } /// A system time-base discontinuity: the PCR PID declared that the clock every @@ -771,11 +808,13 @@ impl Import { fn timebase_break(&mut self) -> anyhow::Result<()> { // The bytes still accumulating belong to the old clock, and their continuation is // stamped on the new one, so cut every PES here rather than splicing across. This - // runs before the marker so a salvaged tail lands in the closing group. + // runs before the marker so a salvaged tail lands in the closing group, which the + // marker closes without guessing an end. for pid in self.streams.keys().copied().collect::>() { - self.broken(pid)?; + self.salvage(pid)?; } for stream in self.streams.values_mut() { + stream.desync(); stream.discontinuity(self.published)?; } for section in self.sections.values_mut() { @@ -1660,6 +1699,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 +1859,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 { @@ -1839,8 +1935,8 @@ impl Stream { 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 { frame.timestamp = reanchor.apply(frame.timestamp, import.floor(frame.keyframe))?; @@ -1861,8 +1957,8 @@ impl Stream { 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 { frame.timestamp = reanchor.apply(frame.timestamp, import.floor(frame.keyframe))?; @@ -1927,6 +2023,20 @@ impl Stream { } } + /// Bytes were lost mid-stream: [`desync`](Self::desync), and close the open video group + /// where its content stops. Every picture until the next keyframe may reference what was + /// lost, so the producer refuses them until a keyframe opens the next group. Left open, + /// the group would close a GOP later at that keyframe, with an end past its content. + fn lost(&mut self) -> anyhow::Result<()> { + self.desync(); + match self { + Stream::H264 { import, .. } => import.cut(None)?, + Stream::H265 { import, .. } => import.cut(None)?, + _ => {} + } + Ok(()) + } + /// The program clock restarted: mark the break on this track and stop unwrapping the /// next PTS against a sample from the timebase that just ended. fn discontinuity(&mut self, published: bool) -> anyhow::Result<()> { @@ -2429,7 +2539,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 +2675,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 +2697,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 +2762,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 +3114,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 +3255,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. /// @@ -6107,6 +6226,364 @@ 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)); + } + "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; + 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(); + 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_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; + } + + #[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 dedicated PCR PID's malformed adaptation field cannot declare a timebase break. + #[test] + fn damaged_adaptation_on_a_dedicated_pcr_pid_keeps_the_clock() { + // Clear of the PMT PID and every stream. + const PCR: u16 = 0x0200; + 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 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].units, + 1, + "the keyframe before the damage" + ); + } + + /// A malformed unbounded PES drained at EOF is still refused. + #[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(); + import.finish().unwrap(); + 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].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(), + }, + ); + 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!( + !super::unit_error(crate::Error::H264(crate::codec::h264::Error::MissingTimestamp)).is::() + ); + } + + /// A break closes the video group where its content stops, not a GOP later at the next + /// keyframe. An open GOP's leading pictures present below that keyframe, so an end marker + /// there would sit past them and the exporter refused them as a rewind. + #[tokio::test(start_paused = true)] + async fn damaged_video_closes_its_group_at_the_break() { + use mpeg2ts::ts::{ReadTsPacket, TsPacketReader, TsPayload}; + + const MS: u64 = 90; + let at = |ms: u64| 90_000 + ms * MS; + let mut mux = Mux { + out: synth_pmt(&[(StreamType::H264, VIDEO)], false), + ..Default::default() + }; + // A closed GOP whose third picture is damaged, then an open GOP whose leading B-frame + // presents below its IDR. + let pictures = [ + (true, at(0), None), + (false, at(40), None), + (false, at(80), None), + (true, at(1200), Some(at(1000))), + (false, at(1040), None), + (false, at(1280), Some(at(1080))), + ]; + let mut damaged_at = 0; + for (i, (keyframe, pts, dts)) in pictures.into_iter().enumerate() { + if i == 2 { + damaged_at = mux.out.len(); + } + let cc = mux.cc(VIDEO); + mux.out + .extend_from_slice(&video_pes(VIDEO, cc, pts, dts, &annexb_au(keyframe))); + } + mux.out[damaged_at + 1] |= 0x80; + + 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).unwrap(); + import.finish().unwrap(); + + // Each group as published, its empty end marker included. + let name = catalog.snapshot().video.renditions.keys().next().unwrap().clone(); + let subscription = moq_net::track::Subscription::default().with_max_age(Duration::from_secs(3600)); + let mut track = consumer.track(&name).unwrap().subscribe(subscription).await.unwrap(); + let mut groups = Vec::new(); + while let Some(mut group) = track.recv_group().await.unwrap() { + let mut frames = Vec::new(); + while let Some(mut frame) = group.next_frame().await.unwrap() { + let frame = hang::container::Frame::decode(frame.read_all().await.unwrap()).unwrap(); + frames.push((frame.timestamp.as_micros(), frame.payload.is_empty())); + } + groups.push(frames); + } + let us = |ms: u64| u128::from(at(ms)) * 1_000_000 / 90_000; + assert_eq!( + groups, + [ + vec![(us(0), false), (us(40), false), (us(80), true)], + vec![(us(1200), false), (us(1040), false), (us(1280), false)], + ], + "the first group ends at the break, one frame after its last picture" + ); + + let mut exporter = super::super::Export::new(crate::source::announced(&consumer)) + .await + .unwrap() + .with_max_age(Duration::from_secs(30)); + let mut ts = Vec::new(); + while let Ok(res) = tokio::time::timeout(Duration::from_secs(1), exporter.next()).await { + let Some(frame) = res.expect("the exporter must take the open GOP after the break") else { + break; + }; + ts.extend_from_slice(&frame.payload); + } + let mut reader = TsPacketReader::new(std::io::Cursor::new(ts)); + let mut pictures = 0; + while let Some(packet) = reader.read_ts_packet().unwrap() { + pictures += usize::from(matches!(packet.payload, Some(TsPayload::PesStart(_)))); + } + assert_eq!(pictures, 5, "every picture either side of the break is exported"); + } + /// #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)] From 918cb137b48559b4ca0d95beacdc6c838b06ceb3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 17:32:14 -0700 Subject: [PATCH 2/3] fix(mux): keep the last good parameter sets when refusing a TS access unit A keyframe refused for a malformed inline SPS left that SPS in the splitter's cache, so every later bare keyframe re-injected it and was refused too. The H.264/H.265 TS path now snapshots the retained parameter sets per PES and restores them when the unit is refused. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/codec/h264/split.rs | 21 +++++ rs/moq-mux/src/codec/h265/split.rs | 24 ++++++ rs/moq-mux/src/container/ts/import.rs | 112 +++++++++++++++++++++++--- 3 files changed, 145 insertions(+), 12 deletions(-) diff --git a/rs/moq-mux/src/codec/h264/split.rs b/rs/moq-mux/src/codec/h264/split.rs index 8f6ba4d6ee..b71634e3e4 100644 --- a/rs/moq-mux/src/codec/h264/split.rs +++ b/rs/moq-mux/src/codec/h264/split.rs @@ -42,6 +42,12 @@ pub struct Split { pending: Vec, } +/// A snapshot of [`Split`]'s retained parameter sets. +pub(crate) struct Params { + sps: Vec, + pps: Vec, +} + #[derive(Default)] struct Avc3Frame { chunks: BytesMut, @@ -236,6 +242,21 @@ impl Split { self.pending.clear(); } + /// The retained parameter sets, for [`restore`](Self::restore) to put back. + pub(crate) fn params(&self) -> Params { + Params { + sps: self.sps.clone(), + pps: self.pps.clone(), + } + } + + /// Put back parameter sets a refused access unit replaced, so a later bare keyframe + /// re-injects the last good ones rather than the refused unit's. + pub(crate) fn restore(&mut self, params: Params) { + self.sps = params.sps; + self.pps = params.pps; + } + fn pts(&mut self, hint: Option) -> Result { if let Some(pts) = hint { return Ok(pts); diff --git a/rs/moq-mux/src/codec/h265/split.rs b/rs/moq-mux/src/codec/h265/split.rs index 9d114416e6..b092e7f255 100644 --- a/rs/moq-mux/src/codec/h265/split.rs +++ b/rs/moq-mux/src/codec/h265/split.rs @@ -40,6 +40,13 @@ pub struct Split { pending: Vec, } +/// A snapshot of [`Split`]'s retained parameter sets. +pub(crate) struct Params { + vps: Vec, + sps: Vec, + pps: Vec, +} + #[derive(Default)] struct Au { chunks: BytesMut, @@ -240,6 +247,23 @@ impl Split { self.pending.clear(); } + /// The retained parameter sets, for [`restore`](Self::restore) to put back. + pub(crate) fn params(&self) -> Params { + Params { + vps: self.vps.clone(), + sps: self.sps.clone(), + pps: self.pps.clone(), + } + } + + /// Put back parameter sets a refused access unit replaced, so a later bare keyframe + /// re-injects the last good ones rather than the refused unit's. + pub(crate) fn restore(&mut self, params: Params) { + self.vps = params.vps; + self.sps = params.sps; + self.pps = params.pps; + } + fn pts(&mut self, hint: Option) -> Result { if let Some(pts) = hint { return Ok(pts); diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index f79784ab61..04b37bc995 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -1934,14 +1934,23 @@ impl Stream { } => { let reorder = reorder_delay(pending.pts, pending.dts); let pts = unwrap_pts(unwrap, pending.pts, anchor)?; + let params = split.params(); // Each PES is one access unit, so flush to emit it immediately. - 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 { - frame.timestamp = reanchor.apply(frame.timestamp, import.floor(frame.keyframe))?; - published += u64::from(skip_missing_keyframe(import.decode([frame]))?); + let published = (|| { + 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 { + frame.timestamp = reanchor.apply(frame.timestamp, import.floor(frame.keyframe))?; + published += u64::from(skip_missing_keyframe(import.decode([frame]))?); + } + anyhow::Ok(published) + })(); + // A refused unit must not leave its parameter sets for a bare keyframe to re-inject. + if published.as_ref().is_err_and(|err| err.is::()) { + split.restore(params); } + let published = published?; // After decode, so the track (and its catalog rendition) exists. if let Some(reorder) = reorder { import.observe_reorder(reorder)?; @@ -1956,14 +1965,23 @@ impl Stream { } => { let reorder = reorder_delay(pending.pts, pending.dts); let pts = unwrap_pts(unwrap, pending.pts, anchor)?; + let params = split.params(); // Each PES is one access unit, so flush to emit it immediately. - 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 { - frame.timestamp = reanchor.apply(frame.timestamp, import.floor(frame.keyframe))?; - published += u64::from(skip_missing_keyframe(import.decode([frame]))?); + let published = (|| { + 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 { + frame.timestamp = reanchor.apply(frame.timestamp, import.floor(frame.keyframe))?; + published += u64::from(skip_missing_keyframe(import.decode([frame]))?); + } + anyhow::Ok(published) + })(); + // A refused unit must not leave its parameter sets for a bare keyframe to re-inject. + if published.as_ref().is_err_and(|err| err.is::()) { + split.restore(params); } + let published = published?; if let Some(reorder) = reorder { import.observe_reorder(reorder)?; } @@ -6352,6 +6370,76 @@ pub(super) mod test { damaged_video_recovers("clean").await; } + /// A keyframe refused for a malformed inline SPS must not replace the splitter's last good + /// parameter sets: a later bare keyframe re-injects those and recovers. + async fn damaged_sps_keeps_the_last_good_parameter_sets(stream_type: StreamType) { + let (params, idr, delta): (&[&[u8]], &[u8], &[u8]) = match stream_type { + StreamType::H264 => { + use crate::container::test_util::{IDR, PPS, SPS}; + (&[SPS, PPS], IDR, &[0x41, 0x9a, 0x00, 0x01]) + } + StreamType::H265 => { + use crate::codec::h265::fixtures::{PPS, SPS, VPS}; + (&[VPS, SPS, PPS], &[0x26, 0x01, 0x80, 0xaa], &[0x02, 0x01, 0x80, 0x33]) + } + _ => unreachable!(), + }; + let good_sps = params[params.len() - 2]; + // The SPS keeps its NAL header and loses the rest. + let truncated = &good_sps[..2]; + let au = |nals: &[&[u8]]| { + let mut out = Vec::new(); + for nal in nals { + out.extend_from_slice(&[0, 0, 0, 1]); + out.extend_from_slice(nal); + } + out + }; + let mut keyframe = params.to_vec(); + keyframe.push(idr); + let mut damaged = params.to_vec(); + damaged[params.len() - 2] = truncated; + damaged.push(idr); + let units = [ + au(&keyframe), + au(&[delta]), + au(&damaged), + au(&[delta]), + au(&[idr]), + au(&[delta]), + ]; + let mut data = synth_pmt(&[(stream_type, VIDEO)], false); + for (cc, unit) in units.iter().enumerate() { + let pts = 90_000 + cc as u64 * FRAME; + data.extend(video_pes(VIDEO, cc as u8, pts, None, unit)); + } + + 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(&data).expect("a damaged SPS stays local to its unit"); + import.finish().unwrap(); + let name = catalog.snapshot().video.renditions.keys().next().unwrap().clone(); + let frames = read_track(&consumer, &name, crate::container::Kind::Video).await; + assert_eq!(frames.len(), 4, "the bare keyframe after the damage recovers"); + assert!(frames[2].keyframe); + assert!( + frames[2].payload.windows(good_sps.len()).any(|bytes| bytes == good_sps), + "the recovery keyframe carries the last good SPS" + ); + } + + #[tokio::test(start_paused = true)] + async fn damaged_h264_sps_keeps_the_last_good_parameter_sets() { + damaged_sps_keeps_the_last_good_parameter_sets(StreamType::H264).await; + } + + #[tokio::test(start_paused = true)] + async fn damaged_h265_sps_keeps_the_last_good_parameter_sets() { + damaged_sps_keeps_the_last_good_parameter_sets(StreamType::H265).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() { From a96df0bc46e8d96d79619c78aed4f16fc7fea47d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 06:00:07 -0700 Subject: [PATCH 3/3] docs(mux): note the per-PES parameter set rollback, and drop 'refused whole' Mirrors the comment #5153 added on main. The CLI docs said a damaged unit is refused whole, which a PES carrying several AUs does not guarantee. Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 2 +- rs/moq-mux/src/container/ts/import.rs | 4 ++++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/doc/bin/cli.md b/doc/bin/cli.md index e18de406be..5aa5564e7e 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -80,7 +80,7 @@ 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 logged as a warning. Ingest continues on every other PID. +dropped and logged as a warning. Ingest continues on every other PID. Video closes its group at the break and resumes at its next keyframe, as it does after a continuity-counter gap: the pictures in between may reference the lost one, so they are dropped rather than decoded with artefacts. Each break freezes diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 04b37bc995..f5f430cb3e 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -1947,6 +1947,8 @@ impl Stream { anyhow::Ok(published) })(); // A refused unit must not leave its parameter sets for a bare keyframe to re-inject. + // The snapshot is per PES, so if one carries several AUs, a later damaged AU also rolls + // back new parameter sets an earlier, already published AU brought. if published.as_ref().is_err_and(|err| err.is::()) { split.restore(params); } @@ -1978,6 +1980,8 @@ impl Stream { anyhow::Ok(published) })(); // A refused unit must not leave its parameter sets for a bare keyframe to re-inject. + // The snapshot is per PES, so if one carries several AUs, a later damaged AU also rolls + // back new parameter sets an earlier, already published AU brought. if published.as_ref().is_err_and(|err| err.is::()) { split.restore(params); }