diff --git a/src/mux/codec/ac3.rs b/src/mux/codec/ac3.rs index 6ae1b35..e2fabc3 100644 --- a/src/mux/codec/ac3.rs +++ b/src/mux/codec/ac3.rs @@ -450,6 +450,7 @@ mod tests { pts: None, dts: None, data: vec![], + discontinuity: false, }; assert!(parser.parse(&pes).is_empty()); } @@ -464,6 +465,7 @@ mod tests { pts: Some(90000), dts: None, data: frame_data.clone(), + discontinuity: false, }; let frames = parser.parse(&pes); assert_eq!(frames.len(), 1); @@ -483,6 +485,7 @@ mod tests { pts: Some(90000), dts: None, data: frame_data[..mid].to_vec(), + discontinuity: false, }; let frames1 = parser.parse(&pes1); assert!(frames1.is_empty(), "partial frame should not emit"); @@ -494,6 +497,7 @@ mod tests { pts: Some(93000), dts: None, data: frame_data[mid..].to_vec(), + discontinuity: false, }; let frames2 = parser.parse(&pes2); assert_eq!(frames2.len(), 1); @@ -512,6 +516,7 @@ mod tests { pts: None, dts: None, data, + discontinuity: false, }; let frames = parser.parse(&pes); assert_eq!(frames.len(), 1); @@ -535,6 +540,7 @@ mod tests { pts: Some(90000), dts: None, data: pes1_data, + discontinuity: false, }; let frames1 = parser.parse(&pes1); assert_eq!(frames1.len(), 1, "first complete frame emitted"); @@ -548,6 +554,7 @@ mod tests { pts: Some(93000), dts: None, data: pes2_data, + discontinuity: false, }; let frames2 = parser.parse(&pes2); assert_eq!(frames2.len(), 1, "split-sync frame must be recovered"); @@ -574,6 +581,7 @@ mod tests { pts: None, dts: None, data, + discontinuity: false, }; let frames = parser.parse(&pes); assert!(frames.is_empty()); @@ -602,6 +610,7 @@ mod tests { pts: None, dts: None, data, + discontinuity: false, }; assert!(parser.parse(&pes).is_empty()); assert_eq!(parser.buf, vec![0x0B], "lone trailing 0x0B retained"); @@ -639,6 +648,7 @@ mod tests { pts: Some(90000), dts: None, data, + discontinuity: false, }; let f = parser.parse(&pes); assert_eq!(f.len(), 1, "frame 1 emitted in parse"); @@ -671,6 +681,7 @@ mod tests { pts: Some(90000), dts: None, data, + discontinuity: false, }; let f = parser.parse(&pes); assert_eq!(f.len(), 2); @@ -714,6 +725,7 @@ mod tests { pts: Some(90000), dts: None, data, + discontinuity: false, }; let frames = parser.parse(&pes); assert_eq!(frames.len(), 1, "only the real AC-3 frame is emitted"); @@ -1179,6 +1191,7 @@ mod tests { pts: Some(90000), dts: None, data, + discontinuity: false, } } } diff --git a/src/mux/codec/dts.rs b/src/mux/codec/dts.rs index 099c7f2..b5fc8e4 100644 --- a/src/mux/codec/dts.rs +++ b/src/mux/codec/dts.rs @@ -482,6 +482,7 @@ mod tests { pts, dts: None, data, + discontinuity: false, } } diff --git a/src/mux/codec/dvdsub.rs b/src/mux/codec/dvdsub.rs index 21f852e..47b0320 100644 --- a/src/mux/codec/dvdsub.rs +++ b/src/mux/codec/dvdsub.rs @@ -233,6 +233,7 @@ mod tests { pts, dts: None, data, + discontinuity: false, } } diff --git a/src/mux/codec/h264.rs b/src/mux/codec/h264.rs index 83d5763..4757128 100644 --- a/src/mux/codec/h264.rs +++ b/src/mux/codec/h264.rs @@ -537,6 +537,7 @@ mod tests { pts, dts: None, data, + discontinuity: false, } } @@ -923,6 +924,7 @@ mod tests { pts: Some(180000), // 2 seconds (presentation) dts: Some(90000), // 1 second (decode) data, + discontinuity: false, }; let frames = parser.parse(&pes); assert_eq!(frames.len(), 1); @@ -1232,6 +1234,7 @@ mod tests { pts: None, dts: Some(90000), data: vec![0x00, 0x00, 0x01, 0x41, 0x10], + discontinuity: false, }; let f = parser.parse(&pes); assert_eq!(f.len(), 1); @@ -1247,6 +1250,7 @@ mod tests { pts: None, dts: None, data: vec![0x00, 0x00, 0x01, 0x41, 0x10], + discontinuity: false, }; let f = parser.parse(&pes); assert_eq!(f.len(), 1); diff --git a/src/mux/codec/hevc.rs b/src/mux/codec/hevc.rs index 91ac64d..dd83527 100644 --- a/src/mux/codec/hevc.rs +++ b/src/mux/codec/hevc.rs @@ -973,6 +973,7 @@ mod tests { pts, dts: None, data, + discontinuity: false, } } @@ -2243,6 +2244,7 @@ mod tests { pts: Some(180000), // 2 s (presentation) dts: Some(90000), // 1 s (decode) data, + discontinuity: false, }; let frames = parser.parse(&pes); assert_eq!(frames.len(), 1); @@ -2850,6 +2852,7 @@ mod tests { d.extend_from_slice(&[0x10, 0x20]); d }, + discontinuity: false, }; let f = parser.parse(&pes); assert_eq!(f.len(), 1); diff --git a/src/mux/codec/lpcm.rs b/src/mux/codec/lpcm.rs index 4350372..3bdd896 100644 --- a/src/mux/codec/lpcm.rs +++ b/src/mux/codec/lpcm.rs @@ -98,6 +98,7 @@ mod tests { pts, dts: None, data, + discontinuity: false, } } diff --git a/src/mux/codec/mod.rs b/src/mux/codec/mod.rs index 8ef9f28..bd37662 100644 --- a/src/mux/codec/mod.rs +++ b/src/mux/codec/mod.rs @@ -198,6 +198,7 @@ mod tests { pts, dts: None, data, + discontinuity: false, } } diff --git a/src/mux/codec/mpeg2.rs b/src/mux/codec/mpeg2.rs index ed30874..23e8472 100644 --- a/src/mux/codec/mpeg2.rs +++ b/src/mux/codec/mpeg2.rs @@ -707,6 +707,7 @@ mod tests { pts: None, dts: None, data, + discontinuity: false, }; let mut p = Mpeg2Parser::new(); let mut frames = Vec::new(); @@ -822,6 +823,7 @@ mod tests { pts, dts: None, data, + discontinuity: false, } } @@ -1415,6 +1417,7 @@ mod tests { pts: None, dts: Some(90000), data, + discontinuity: false, }; let f = parse_then_flush(&mut parser, &pes); assert_eq!(f[0].pts_ns, 1_000_000_000, "DTS fallback"); @@ -1428,6 +1431,7 @@ mod tests { pts: None, dts: None, data: data2, + discontinuity: false, }; let f2 = parse_then_flush(&mut parser2, &pes2); assert_eq!(f2[0].pts_ns, 0, "no PTS/DTS → 0"); diff --git a/src/mux/codec/pgs.rs b/src/mux/codec/pgs.rs index b7a9fff..7779592 100644 --- a/src/mux/codec/pgs.rs +++ b/src/mux/codec/pgs.rs @@ -223,6 +223,7 @@ mod tests { pts, dts: None, data, + discontinuity: false, } } diff --git a/src/mux/codec/truehd.rs b/src/mux/codec/truehd.rs index e8480dc..d6b8302 100644 --- a/src/mux/codec/truehd.rs +++ b/src/mux/codec/truehd.rs @@ -425,6 +425,7 @@ mod tests { pts, dts: None, data, + discontinuity: false, } } diff --git a/src/mux/codec/vc1.rs b/src/mux/codec/vc1.rs index e5bee00..334c153 100644 --- a/src/mux/codec/vc1.rs +++ b/src/mux/codec/vc1.rs @@ -451,6 +451,7 @@ mod tests { pts, dts: None, data, + discontinuity: false, } } @@ -757,6 +758,7 @@ mod tests { pts: Some(180000), // presentation dts: Some(90000), // decode data, + discontinuity: false, }; let frames = parser.parse(&pes); assert_eq!(frames.len(), 1); @@ -1007,6 +1009,7 @@ mod tests { pts: None, dts: Some(90000), data: vec![0x00, 0x00, 0x01, SC_FRAME, 0x55], + discontinuity: false, }; let f = parser.parse(&pes); assert_eq!(f[0].pts_ns, 1_000_000_000, "DTS fallback"); @@ -1018,6 +1021,7 @@ mod tests { pts: None, dts: None, data: vec![0x00, 0x00, 0x01, SC_FRAME, 0x55], + discontinuity: false, }; let f2 = parser2.parse(&pes2); assert_eq!(f2[0].pts_ns, 0, "no PTS/DTS → 0"); diff --git a/src/mux/disc.rs b/src/mux/disc.rs index 0dea489..0dca5ad 100644 --- a/src/mux/disc.rs +++ b/src/mux/disc.rs @@ -723,6 +723,8 @@ impl crate::pes::Stream for DiscStream { pts: ps.pts.map(|p| p as i64), dts: ps.dts.map(|d| d as i64), data: ps.data.clone(), + // PS (DVD/CSS) path: no AACS conceal, no gap flag. + discontinuity: false, }; if let Some((_, parser)) = self.parsers.iter_mut().find(|(p, _)| *p == pid) { @@ -846,6 +848,8 @@ impl crate::pes::Stream for DiscStream { pts: ps.pts.map(|p| p as i64), dts: ps.dts.map(|d| d as i64), data: ps.data.clone(), + // PS (DVD/CSS) path: no AACS conceal, no gap flag. + discontinuity: false, }; if let Some((_, parser)) = self.parsers.iter_mut().find(|(p, _)| *p == pid) { diff --git a/src/mux/mkv.rs b/src/mux/mkv.rs index fd99cdb..4397a30 100644 --- a/src/mux/mkv.rs +++ b/src/mux/mkv.rs @@ -2665,6 +2665,7 @@ mod tests { pts: Some(gop_pts), dts: None, data: es, + discontinuity: false, })); } frames.extend(parser.flush()); diff --git a/src/mux/mod.rs b/src/mux/mod.rs index 59bd885..f582d5f 100644 --- a/src/mux/mod.rs +++ b/src/mux/mod.rs @@ -91,6 +91,7 @@ pub(crate) mod mkvstream; pub(crate) mod network; pub(crate) mod null; pub(crate) mod ps; +pub(crate) mod resync; pub(crate) mod stdio; /// Shared clip-boundary timeline-continuity corrector (used by the MKV muxer /// and the `demux://` sink). diff --git a/src/mux/pipelined_stream.rs b/src/mux/pipelined_stream.rs index c3dc900..558e3da 100644 --- a/src/mux/pipelined_stream.rs +++ b/src/mux/pipelined_stream.rs @@ -68,6 +68,13 @@ pub struct PipelinedPesStream { /// are expected on every disc; instead of a per-packet WARN they're tallied /// and summarised once at EOF. dropped_nav_packets: u64, + /// Per-track (by stream index) B1 drop-to-keyframe gate. After a TS gap on a + /// video track, drop inter-coded frames until the next IRAP so the muxed + /// stream stays decode-clean across an upstream concealed loss (P3/B1). + resync: Vec, + /// Per-track "is inter-coded video" flag (only video has cross-frame + /// references the gate must protect). Indexed by stream index. + is_video: Vec, } impl PipelinedPesStream { @@ -87,6 +94,14 @@ impl PipelinedPesStream { parsers: Vec<(u16, Box)>, pid_to_track: Vec<(u16, usize)>, ) -> Self { + let is_video: Vec = title + .streams + .iter() + .map(|s| matches!(s, crate::disc::Stream::Video(_))) + .collect(); + let resync = (0..title.streams.len()) + .map(|_| super::resync::ResyncGate::new()) + .collect(); Self { title, parsers, @@ -98,6 +113,8 @@ impl PipelinedPesStream { skip_parse: std::env::var_os("FREEMKV_SKIP_PARSE").is_some(), decrypt_loss: None, dropped_nav_packets: 0, + resync, + is_video, } } @@ -164,9 +181,35 @@ impl PipelinedPesStream { } else if let Some((_, parser)) = self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid) { + let is_video = self.is_video.get(track).copied().unwrap_or(false); + let discontinuity = pes.discontinuity; for frame in parser.parse(&pes) { - self.pending_frames - .push_back(PesFrame::from_codec_frame(track, frame)); + // B1: after a TS gap on a video track, drop forward to the + // next keyframe so no frame with a dangling reference is + // emitted. Audio/subtitle always admit (independent frames). + // A track with no gate (out-of-range index) emits as-is. + let emit = match self.resync.get_mut(track) { + Some(gate) => { + let was_armed = gate.is_armed(); + let dropped = gate.dropped_in_run(); + let admit = gate.admit(is_video, discontinuity, frame.keyframe); + if admit && was_armed { + tracing::warn!( + target: "mux", + track, + pid = pes.pid, + dropped, + "B1: resynced at keyframe after concealed gap" + ); + } + admit + } + None => true, + }; + if emit { + self.pending_frames + .push_back(PesFrame::from_codec_frame(track, frame)); + } } } } @@ -217,6 +260,8 @@ impl PipelinedPesStream { pts: ps.pts.map(|p| p as i64), dts: ps.dts.map(|d| d as i64), data: ps.data, + // PS (DVD/CSS) path: no AACS conceal → no continuity-gap flag. + discontinuity: false, }; if let Some((_, parser)) = self.parsers.iter_mut().find(|(p, _)| *p == pid) { for frame in parser.parse(&pes) { @@ -411,6 +456,7 @@ mod tests { pts: Some(90_000), dts: None, data, + discontinuity: false, } } @@ -498,6 +544,125 @@ mod tests { ); } + /// A parser that emits exactly one frame per PES, marking it a keyframe iff + /// the PES payload's first byte is `b'K'`. Lets a test script a precise + /// keyframe/inter-frame sequence to exercise the B1 resync gate. + struct KeyframeParser; + impl CodecParser for KeyframeParser { + fn parse(&mut self, pes: &PesPacket) -> Vec { + vec![super::super::codec::Frame { + coding: None, + source: None, + pts_ns: pes.pts.unwrap_or(0), + keyframe: pes.data.first() == Some(&b'K'), + data: pes.data.clone(), + duration_ns: None, + }] + } + fn flush(&mut self) -> Vec { + vec![] + } + fn codec_private(&self) -> Option> { + None + } + } + + fn ts_pes_disc(pid: u16, data: Vec, discontinuity: bool) -> PesPacket { + PesPacket { + source: None, + pid, + pts: Some(90_000), + dts: None, + data, + discontinuity, + } + } + + /// B1 end-to-end: after a TS discontinuity on a VIDEO track the consumer + /// must DROP every inter-coded frame until the next keyframe, so no frame + /// with a dangling reference reaches the muxer (an ffmpeg deep-scan would + /// otherwise report a missing reference). The frame carrying the + /// discontinuity and the inter frames behind it are dropped; the stream + /// resumes cleanly at the next keyframe. + #[test] + fn b1_video_drops_to_keyframe_after_discontinuity() { + let title = video_title(false); // one HEVC video stream, PID 0x1011 + let parsers: Vec<(u16, Box)> = vec![(0x1011, Box::new(KeyframeParser))]; + let pid_to_track = vec![(0x1011u16, 0usize)]; + let (mut stream, tx) = make_stream(title, parsers, pid_to_track); + + // K0,P1 clean → emit. P2 carries the gap (inter frame referencing the + // lost data) → arms the gate; P2,P3 drop. K4 is the next keyframe → + // resync + emit. P5 then admits cleanly. + tx.send(DemuxBatch::Ts(vec![ + ts_pes_disc(0x1011, b"K0".to_vec(), false), + ts_pes_disc(0x1011, b"P1".to_vec(), false), + ts_pes_disc(0x1011, b"P2".to_vec(), true), + ts_pes_disc(0x1011, b"P3".to_vec(), false), + ts_pes_disc(0x1011, b"K4".to_vec(), false), + ts_pes_disc(0x1011, b"P5".to_vec(), false), + ])) + .unwrap(); + tx.send(DemuxBatch::Eof).unwrap(); + + let mut emitted = Vec::new(); + while let Some(f) = stream.read().unwrap() { + emitted.push(f.data); + } + // P2 (gap) and P3 (still no keyframe) are dropped; the rest survive in + // order. Crucially the FIRST frame after the gap that we emit is the + // keyframe K4 — never a dangling-reference inter frame. + assert_eq!( + emitted, + vec![ + b"K0".to_vec(), + b"P1".to_vec(), + b"K4".to_vec(), + b"P5".to_vec() + ], + "post-gap inter frames dropped, stream resumes at the keyframe" + ); + } + + /// Counterpart to B1: a discontinuity on a NON-video track must NOT drop + /// frames — audio/subtitle access units are independent, so the gate admits + /// every frame the parser still produces. + #[test] + fn b1_audio_does_not_drop_on_discontinuity() { + let mut title = DiscTitle::empty(); + title.streams.push(crate::disc::Stream::Audio(AudioStream { + pid: 0x1100, + codec: Codec::Ac3, + channels: AudioChannels::Surround51, + language: "eng".into(), + sample_rate: SampleRate::S48, + secondary: false, + purpose: LabelPurpose::Normal, + label: String::new(), + })); + let parsers: Vec<(u16, Box)> = vec![(0x1100, Box::new(KeyframeParser))]; + let pid_to_track = vec![(0x1100u16, 0usize)]; + let (mut stream, tx) = make_stream(title, parsers, pid_to_track); + + tx.send(DemuxBatch::Ts(vec![ + ts_pes_disc(0x1100, b"a0".to_vec(), false), + ts_pes_disc(0x1100, b"a1".to_vec(), true), // gap — but audio is independent + ts_pes_disc(0x1100, b"a2".to_vec(), false), + ])) + .unwrap(); + tx.send(DemuxBatch::Eof).unwrap(); + + let mut emitted = Vec::new(); + while let Some(f) = stream.read().unwrap() { + emitted.push(f.data); + } + assert_eq!( + emitted, + vec![b"a0".to_vec(), b"a1".to_vec(), b"a2".to_vec()], + "audio frames are never dropped on a discontinuity" + ); + } + /// At EOF the consumer must call `flush()` on every parser and emit the /// buffered tail frames — a parser that holds the final access unit (e.g. /// DTS-HD) must NOT have it dropped. Without the flush the last frame is diff --git a/src/mux/resync.rs b/src/mux/resync.rs new file mode 100644 index 0000000..4980217 --- /dev/null +++ b/src/mux/resync.rs @@ -0,0 +1,123 @@ +//! B1 drop-to-IRAP gate — keep a muxed elementary stream decode-clean across a +//! mid-stream gap (e.g. an undecryptable unit the mux concealed as NULL TS, +//! P3/A2). +//! +//! When packets are lost, the affected access unit is already dropped at the TS +//! layer (the assembler drops the partial PES on the continuity gap). But for +//! INTER-CODED video the frames that follow reference the lost frame (and each +//! other) until the next IRAP/IDR keyframe — emitting them yields an ffmpeg +//! "missing reference / non-existing PPS" deep-scan error and visibly broken +//! decode. So after a gap on a video track we DROP FORWARD to the next keyframe +//! and resume cleanly there. The gap rounds up to (at most) one GOP — the price +//! of never emitting a dangling reference; it is logged. +//! +//! Audio and subtitle frames are independent (no inter-frame references), so a +//! gap there costs only the single already-dropped frame; the gate is a no-op +//! for non-video tracks (it always admits). + +/// Per-track keyframe-resync state. One gate per elementary stream; a video +/// track's gate stays "armed" from a discontinuity until the next keyframe. +#[derive(Debug, Default)] +pub(crate) struct ResyncGate { + /// True while dropping post-gap inter-coded frames until the next keyframe. + armed: bool, + /// Count of frames dropped while armed (for a single summary log on resync). + dropped: u64, +} + +impl ResyncGate { + pub(crate) fn new() -> Self { + Self { + armed: false, + dropped: 0, + } + } + + /// Decide whether a parsed frame should be EMITTED (`true`) or DROPPED + /// (`false`). + /// + /// * `is_video` — inter-coded video track (the only kind with cross-frame + /// references); `false` for audio/subtitle, which always admit. + /// * `discontinuity` — this frame's source PES followed a TS continuity gap. + /// * `keyframe` — this frame is a self-contained IRAP/IDR. + /// + /// A non-video track always admits. A video track arms on a discontinuity + /// and then drops every non-keyframe until (and excluding the drop of) the + /// next keyframe, which disarms and is emitted. + pub(crate) fn admit(&mut self, is_video: bool, discontinuity: bool, keyframe: bool) -> bool { + if !is_video { + return true; + } + if discontinuity { + self.armed = true; + } + if self.armed { + if keyframe { + self.armed = false; + self.dropped = 0; + true + } else { + self.dropped += 1; + false + } + } else { + true + } + } + + /// Frames dropped so far in the CURRENT armed run (0 when not armed / just + /// resynced). Lets the consumer log the resync cost once at the keyframe. + pub(crate) fn dropped_in_run(&self) -> u64 { + self.dropped + } + + /// Whether the gate is currently dropping frames (armed, awaiting keyframe). + pub(crate) fn is_armed(&self) -> bool { + self.armed + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn non_video_always_admits_even_on_discontinuity() { + let mut g = ResyncGate::new(); + // Audio/subtitle: a gap drops only the (already TS-dropped) frame; every + // frame the parser still emits is independent and must pass. + assert!(g.admit(false, true, false)); + assert!(g.admit(false, true, false)); + assert!(!g.is_armed(), "non-video never arms"); + } + + #[test] + fn video_drops_inter_frames_until_next_keyframe() { + let mut g = ResyncGate::new(); + // Clean run: everything admits. + assert!(g.admit(true, false, true)); // IDR + assert!(g.admit(true, false, false)); // P + // Gap arrives on the next frame (a P referencing lost data) → drop it + // and every inter frame until the next keyframe. + assert!(!g.admit(true, true, false), "post-gap P dropped"); + assert!(!g.admit(true, false, false), "still dropping (no key yet)"); + assert!(!g.admit(true, false, false)); + assert_eq!(g.dropped_in_run(), 3); + // Next keyframe resyncs and is emitted. + assert!(g.admit(true, false, true), "keyframe resumes the stream"); + assert!(!g.is_armed()); + assert_eq!(g.dropped_in_run(), 0); + // Back to a clean run. + assert!(g.admit(true, false, false), "post-resync P admits"); + } + + #[test] + fn discontinuity_landing_on_a_keyframe_emits_immediately() { + let mut g = ResyncGate::new(); + // If the first surviving frame after the gap is itself an IRAP, there is + // nothing to drop — it is self-contained. + assert!(g.admit(true, true, true), "gap+keyframe emits, no drop"); + assert!(!g.is_armed()); + assert_eq!(g.dropped_in_run(), 0); + } +} diff --git a/src/mux/ts.rs b/src/mux/ts.rs index e0bbd1b..bd66375 100644 --- a/src/mux/ts.rs +++ b/src/mux/ts.rs @@ -28,6 +28,15 @@ pub struct PesPacket { /// from the producer's known stream offset. `None` when the demuxer was fed /// without a base offset (callers that don't need provenance). pub source: Option, + /// True when a TS continuity gap (a CC discontinuity, or an adaptation-field + /// discontinuity_indicator) was seen on this PID since the previous PES + /// completed — i.e. one or more packets for this stream were lost (e.g. the + /// mux replaced an undecryptable unit with NULL TS packets, P3/A2). This is + /// the FIRST surviving PES after the gap, so for inter-coded video it (and + /// every later frame up to the next IRAP/IDR) may reference data that is now + /// gone. The codec-parse consumer uses it to drop forward to the next + /// keyframe (B1) instead of emitting frames with dangling references. + pub discontinuity: bool, } /// Per-PID PES reassembly state. @@ -58,6 +67,11 @@ struct PesAssembler { /// Stamped at PES start, emitted on the completed packet — provenance is /// carried, never reconstructed downstream. pes_source: Option, + /// Sticky "a continuity gap occurred on this PID" flag. Set whenever a CC + /// gap or an explicit discontinuity_indicator is seen; carried onto the NEXT + /// completed PES (which is the first surviving frame after the loss) and then + /// cleared. Drives B1 drop-to-keyframe in the codec consumer. + pending_discontinuity: bool, } /// Initial capacity for a fresh PES buffer. Sized to cover the @@ -92,6 +106,7 @@ impl PesAssembler { header_remaining: 0, last_cc: None, pes_source: None, + pending_discontinuity: false, } } @@ -105,12 +120,15 @@ impl PesAssembler { source: Option, ) -> Option { let completed = if self.active && !self.buffer.is_empty() { + let discontinuity = self.pending_discontinuity; + self.pending_discontinuity = false; Some(PesPacket { pid: self.pid, pts: self.pts, dts: self.dts, data: std::mem::replace(&mut self.buffer, Vec::with_capacity(PES_BUFFER_INIT_CAP)), source: self.pes_source, + discontinuity, }) } else { self.buffer.clear(); @@ -151,12 +169,15 @@ impl PesAssembler { fn flush(&mut self) -> Option { if self.active && !self.buffer.is_empty() { self.active = false; + let discontinuity = self.pending_discontinuity; + self.pending_discontinuity = false; Some(PesPacket { pid: self.pid, pts: self.pts, dts: self.dts, data: std::mem::take(&mut self.buffer), source: self.pes_source, + discontinuity, }) } else { None @@ -391,6 +412,16 @@ impl TsDemuxer { None => false, }; asm.last_cc = Some(cc); + // Any continuity gap — at a PUSI boundary or mid-PES — means packets for + // this stream were lost (an upstream NULL-TS conceal, a damaged source). + // Mark it sticky so the NEXT completed PES carries `discontinuity` and the + // codec consumer can drop forward to the next keyframe (B1). The partial + // PES is still dropped below only for a NON-PUSI continuation (a hole in + // the middle of the current frame); a gap landing exactly on a PUSI starts + // a clean new frame, but it is still the first frame after the loss. + if discontinuity_flag || cc_gap { + asm.pending_discontinuity = true; + } if !pusi && (discontinuity_flag || cc_gap) && asm.active { tracing::trace!( target: "mux", @@ -916,6 +947,50 @@ mod tests { ); } + /// B1 plumbing: a continuity gap must STAMP `discontinuity = true` on the + /// next completed PES so the codec consumer can drop forward to the next + /// keyframe. A clean in-sequence PES carries `discontinuity = false`. We + /// open A (cc=0), flush it cleanly via B's PUSI (cc=1), then jump the CC + /// (cc 1 -> 5) on C's PUSI: the gap is sticky and lands on the PES flushed + /// at that boundary (B — the frame whose tail packets were the lost ones). + #[test] + fn continuity_gap_stamps_discontinuity_on_next_pes() { + let pid = 0x1011; + let mut demux = TsDemuxer::new(&[pid]); + + // Open A (cc=0): nothing completed yet. + let out = demux.feed(&ts_payload_packet(pid, true, 0, &pes_start(b"AAAA"))); + assert!(out.is_empty()); + + // B's PUSI (cc=1, in sequence) flushes A. A saw no gap → clean. + let out = demux.feed(&ts_payload_packet(pid, true, 1, &pes_start(b"BBBB"))); + assert_eq!(out.len(), 1, "A completes"); + assert_eq!(&out[0].data[..4], b"AAAA"); + assert!( + !out[0].discontinuity, + "in-sequence PES is not a discontinuity" + ); + + // C's PUSI jumps cc 1 -> 5: packets were lost. The gap is sticky and is + // attributed to the PES flushed here (B), which lost its tail packets. + let out = demux.feed(&ts_payload_packet(pid, true, 5, &pes_start(b"CCCC"))); + assert_eq!(out.len(), 1, "B completes"); + assert_eq!(&out[0].data[..4], b"BBBB"); + assert!( + out[0].discontinuity, + "the PES at the continuity gap must be flagged so B1 can resync" + ); + + // C itself was opened clean (after the gap) and carries no new gap. + let out = demux.flush(); + assert_eq!(out.len(), 1); + assert_eq!(&out[0].data[..4], b"CCCC"); + assert!( + !out[0].discontinuity, + "post-gap PES with no further gap is clean" + ); + } + /// In-sequence continuation (cc 0 -> 1) must still splice normally — the /// continuity check must not break the happy path. #[test]