mux: B1 drop-to-keyframe resync after a concealed gap

Pairs with A2 (read-path NULL-TS concealment). When the demux assembler
sees a TS continuity gap it now stamps `discontinuity` on the next
completed PES; the codec-parse stage carries that onto a per-track
ResyncGate. After a gap on an inter-coded video track the gate drops
forward to the next IRAP/IDR keyframe so no frame with a dangling
reference reaches the muxer (an ffmpeg deep scan would otherwise report
a missing-reference / non-existing-PPS error). Audio and subtitle tracks
have no cross-frame references, so the gate is a no-op there.

- ts.rs: PesPacket gains `discontinuity`; PesAssembler tracks a sticky
  pending_discontinuity flag set on CC gap / discontinuity_indicator and
  carried to the next completed/flushed PES.
- resync.rs (new): ResyncGate — per-track arm-on-gap, drop non-keyframes
  until the next keyframe disarms and resumes. Logs the resync + drop
  count once at the keyframe.
- pipelined_stream.rs: precompute per-track is_video, apply the gate in
  consume_ts. Out-of-range track index emits as-is (defensive).

Tests: ResyncGate unit tests; ts.rs gap-stamps-discontinuity; end-to-end
B1 video-drops-to-keyframe and audio-never-drops through PipelinedPesStream.

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,
         }
     }
 }

src/mux/codec/dts.rs

@@ -482,6 +482,7 @@ mod tests {
             pts,
             dts: None,
             data,
+            discontinuity: false,
         }
     }
 

src/mux/codec/dvdsub.rs

@@ -233,6 +233,7 @@ mod tests {
             pts,
             dts: None,
             data,
+            discontinuity: false,
         }
     }
 

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);

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);

src/mux/codec/lpcm.rs

@@ -98,6 +98,7 @@ mod tests {
             pts,
             dts: None,
             data,
+            discontinuity: false,
         }
     }
 

src/mux/codec/mod.rs

@@ -198,6 +198,7 @@ mod tests {
             pts,
             dts: None,
             data,
+            discontinuity: false,
         }
     }
 

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");

src/mux/codec/pgs.rs

@@ -223,6 +223,7 @@ mod tests {
             pts,
             dts: None,
             data,
+            discontinuity: false,
         }
     }
 

src/mux/codec/truehd.rs

@@ -425,6 +425,7 @@ mod tests {
             pts,
             dts: None,
             data,
+            discontinuity: false,
         }
     }
 

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");

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) {

src/mux/mkv.rs

@@ -2665,6 +2665,7 @@ mod tests {
                 pts: Some(gop_pts),
                 dts: None,
                 data: es,
+                discontinuity: false,
             }));
         }
         frames.extend(parser.flush());

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).

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<super::resync::ResyncGate>,
+    /// Per-track "is inter-coded video" flag (only video has cross-frame
+    /// references the gate must protect). Indexed by stream index.
+    is_video: Vec<bool>,
 }
 
 impl PipelinedPesStream {
@@ -87,6 +94,14 @@ impl PipelinedPesStream {
         parsers: Vec<(u16, Box<dyn CodecParser>)>,
         pid_to_track: Vec<(u16, usize)>,
     ) -> Self {
+        let is_video: Vec<bool> = 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,7 +181,32 @@ 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) {
+                        // 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));
                         }
@@ -172,6 +214,7 @@ impl PipelinedPesStream {
                 }
             }
         }
+    }
 
     fn consume_ps(&mut self, packets: Vec<super::ps::PsPacket>) {
         for ps in packets {
@@ -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<super::super::codec::Frame> {
+            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<super::super::codec::Frame> {
+            vec![]
+        }
+        fn codec_private(&self) -> Option<Vec<u8>> {
+            None
+        }
+    }
+
+    fn ts_pes_disc(pid: u16, data: Vec<u8>, 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<dyn CodecParser>)> = 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<dyn CodecParser>)> = 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

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);
+    }
+}

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<crate::pes::SourcePos>,
+    /// 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<crate::pes::SourcePos>,
+    /// 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<crate::pes::SourcePos>,
     ) -> Option<PesPacket> {
         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<PesPacket> {
         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]