demux_sink: authored PGS clear beats a capped duration, hostile chapter names neutralised; driver: stop paths pinned, dead live_keyed param removed

src/mux/demux_sink.rs

@@ -396,8 +396,10 @@ impl EsWriter for PgsSupWriter {
         let pts90 = ns_to_90k(pts_ns);
         let is_pcs = f.data.first() == Some(&SEG_PCS) && f.data.len() > 13;
         let mut written = 0;
+        // An authored clear states the real end, even past a duration the parser capped.
+        let is_clear = is_pcs && f.data[13] == 0;
         if let Some((end, width, height)) = self.pending_clear {
-            if end < pts_ns || (end == pts_ns && !is_pcs) {
+            if (end < pts_ns && !is_clear) || (end == pts_ns && !is_pcs) {
                 let clear = Self::synthetic_clear_display_set(width, height);
                 let pts = ns_to_90k(end);
                 written += Self::emit_segments(&clear, pts, pts, w)?.0;
@@ -611,7 +613,7 @@ pub(crate) fn chapters_ogm(chapters: &[Chapter]) -> String {
         let name = if c.name.is_empty() {
             n.to_string()
         } else {
-            c.name.clone()
+            escape_controls(&c.name, &[])
         };
         s.push_str(&format!("CHAPTER{n:02}={h:02}:{m:02}:{sec:02}.{ms:03}\n"));
         s.push_str(&format!("CHAPTER{n:02}NAME={name}\n"));
@@ -620,11 +622,28 @@ pub(crate) fn chapters_ogm(chapters: &[Chapter]) -> String {
 }
 
 fn xml_escape(s: &str) -> String {
-    s.replace('&', "&amp;")
+    // XML 1.0 forbids C0 controls other than tab / LF / CR.
+    escape_controls(s, &['\t', '\n', '\r'])
+        .replace('&', "&amp;")
         .replace('<', "&lt;")
         .replace('>', "&gt;")
 }
 
+// Disc-derived text in a line-oriented or XML file: control chars (and U+FFFE / U+FFFF)
+// outside `allow` are written in their `escape_debug()` form, as `Error::DirNameCollision`
+// does, instead of raw.
+fn escape_controls(s: &str, allow: &[char]) -> String {
+    let mut out = String::with_capacity(s.len());
+    for c in s.chars() {
+        if (c.is_control() && !allow.contains(&c)) || matches!(c, '\u{FFFE}' | '\u{FFFF}') {
+            out.extend(c.escape_debug());
+        } else {
+            out.push(c);
+        }
+    }
+    out
+}
+
 // Replace path-hostile characters (control chars incl. NUL) in a filename component;
 // disc-derived text (e.g. STN language codes) is unvalidated.
 fn sanitize(s: &str) -> String {
@@ -1521,6 +1540,84 @@ mod tests {
         }
     }
 
+    // The parser caps a display at 30 s; an authored clear later than that is the real end.
+    #[test]
+    fn pgs_sup_authored_clear_beyond_a_capped_duration_is_not_preempted() {
+        let mut writer = PgsSupWriter::default();
+        let mut bytes = Vec::new();
+        for frame in [
+            sup_frame(0, Some(30_000_000_000), true),
+            sup_frame(40_000_000_000, Some(0), false),
+        ] {
+            writer.write_frame(&mut bytes, &frame, frame.pts).unwrap();
+        }
+        writer.finish(&mut bytes).unwrap();
+        assert_eq!(sup_compositions(&bytes), [(0, 1), (3_600_000, 0)]);
+    }
+
+    // A continuation segment stamped exactly at the wipe time follows the clear, not precedes it.
+    #[test]
+    fn pgs_sup_clear_precedes_a_non_pcs_frame_at_the_wipe_time() {
+        let mut writer = PgsSupWriter::default();
+        let mut bytes = Vec::new();
+        let display = sup_frame(0, Some(2_000_000_000), true);
+        writer.write_frame(&mut bytes, &display, 0).unwrap();
+        let mut end = sup_frame(2_000_000_000, None, false);
+        end.data = vec![SEG_END, 0, 0];
+        writer.write_frame(&mut bytes, &end, end.pts).unwrap();
+        writer.finish(&mut bytes).unwrap();
+        let mut types = Vec::new();
+        let mut pos = 0;
+        while pos < bytes.len() {
+            let size = usize::from(u16::from_be_bytes([
+                bytes[pos + SUP_HEADER_LEN + 1],
+                bytes[pos + SUP_HEADER_LEN + 2],
+            ]));
+            types.push(bytes[pos + SUP_HEADER_LEN]);
+            pos += SUP_HEADER_LEN + PGS_SEG_HEADER_LEN + size;
+        }
+        // display PCS + END, clear PCS + END, then the continuation END.
+        assert_eq!(types, [SEG_PCS, SEG_END, SEG_PCS, SEG_END, SEG_END]);
+    }
+
+    // sub:// / demux:// finish() must flush the writer so the last subtitle is cleared.
+    #[test]
+    fn sink_finish_writes_the_last_pgs_subtitles_synthetic_clear() {
+        let dir = tempdir();
+        let title = title_with(vec![subtitle_stream(Codec::Pgs, "eng")], vec![None]);
+        let opts = DemuxOptions {
+            base: "Sub".to_string(),
+            export_chapters: false,
+            ..Default::default()
+        };
+        let mut sink = DemuxSink::create(&dir, &title, &opts).unwrap();
+        sink.write(&sup_frame(0, Some(2_000_000_000), true))
+            .unwrap();
+        sink.finish().unwrap();
+        let sup = std::fs::read_dir(&dir)
+            .unwrap()
+            .map(|e| e.unwrap().path())
+            .find(|p| p.extension().is_some_and(|e| e == "sup"))
+            .expect("a .sup file");
+        let bytes = std::fs::read(&sup).unwrap();
+        assert_eq!(sup_compositions(&bytes), [(0, 1), (180_000, 0)]);
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
+    // Hostile chapter names must not forge OGM lines or make the XML ill-formed.
+    #[test]
+    fn chapter_names_with_control_chars_are_neutralised() {
+        let chaps = vec![Chapter {
+            time_secs: 0.0,
+            name: "x\nCHAPTER99=00:00:00.000\u{1}y".to_string(),
+        }];
+        let ogm = chapters_ogm(&chaps);
+        assert_eq!(ogm.lines().count(), 2, "{ogm:?}");
+        assert!(!ogm.lines().any(|l| l.starts_with("CHAPTER99")));
+        let xml = chapters_xml(&chaps);
+        assert!(!xml.contains('\u{1}'), "{xml:?}");
+    }
+
     #[test]
     fn pgs_sup_legacy_duration_only_input_clears_before_a_later_display() {
         let mut writer = PgsSupWriter::default();

src/mux/driver.rs

@@ -164,14 +164,19 @@ pub fn mux_url(
 ) -> std::io::Result<MuxOutcome> {
     let source = match super::source::open_source(url, crate::disc::ScanOptions::default(), ctx) {
         Ok(s) => s,
-        Err(Error::Halted) if ctx.halt.is_cancelled() => {
-            return Ok(MuxOutcome::stopped_before_output(0, 0));
-        }
-        Err(e) => return Err(e.into()),
+        Err(e) => return open_failed(e, ctx),
     };
     mux_with_keys(source, keys, dest_url, opts, ctx)
 }
 
+// A failed open: the stopped outcome when it is a Stop, the error otherwise.
+fn open_failed(e: Error, ctx: &Ctx) -> std::io::Result<MuxOutcome> {
+    match e {
+        Error::Halted if ctx.halt.is_cancelled() => Ok(MuxOutcome::stopped_before_output(0, 0)),
+        e => Err(e.into()),
+    }
+}
+
 // `mux_with_keys` before its one halt rule.
 fn mux_routed(
     source: Source,
@@ -251,7 +256,7 @@ pub(crate) fn open_pes(
             match set.or(keyless.as_ref()) {
                 Some(set) => {
                     let title = live_keyed_title(&*reader, title, set, opts)?;
-                    live_keyed(reader, title, format, set, opts, ctx)
+                    live_keyed(reader, title, set, opts, ctx)
                 }
                 None => {
                     let mut title = title;
@@ -344,7 +349,7 @@ fn session_pes(
             let staged = session.staged_reader().ok_or_else(not_ready)?;
             let title = live_keyed_title(staged, title, set, opts)?;
             let reader = session.take_reader().ok_or_else(not_ready)?;
-            live_keyed(reader, title, format, set, opts, ctx)
+            live_keyed(reader, title, set, opts, ctx)
         }
         None => {
             let mut title = title;
@@ -411,7 +416,6 @@ fn live_keyed_title(
 fn live_keyed(
     reader: Box<dyn SectorSource>,
     title: DiscTitle,
-    _format: crate::disc::ContentFormat,
     set: &crate::keys::KeyRing,
     opts: &MuxOptions,
     ctx: &Ctx,
@@ -1401,6 +1405,72 @@ mod tests {
         }
     }
 
+    // A never-syncing TrueHD title with `n` non-sync frames of `size` bytes each.
+    fn unsynced_truehd_stream(n: usize, size: usize) -> FakeStream {
+        use crate::disc::{AudioChannels, AudioStream, Codec, LabelPurpose, SampleRate, Stream};
+        let mut s = FakeStream::new(0);
+        s.info.streams = vec![Stream::Audio(AudioStream {
+            pid: 0x1100,
+            codec: Codec::TrueHd,
+            channels: AudioChannels::Surround51,
+            language: "eng".into(),
+            sample_rate: SampleRate::S48,
+            secondary: false,
+            purpose: LabelPurpose::Normal,
+            label: String::new(),
+        })];
+        for pts in 0..n as i64 {
+            s.frames.push_back(PesFrame {
+                discard_padding_ns: 0,
+                track: 0,
+                pts,
+                keyframe: true,
+                data: vec![0u8; size],
+                duration_ns: None,
+                source: None,
+                coding: None,
+            });
+        }
+        s
+    }
+
+    // A Stop during the TrueHD probe ends the mux before any output opens, for both a
+    // frame sink and a metadata sink, whether it shows as a cancelled halt or Err(Halted).
+    #[test]
+    fn a_stop_during_the_truehd_probe_stops_before_the_output_opens() {
+        let dir = tempfile::tempdir().unwrap();
+        let json = format!("json://{}", dir.path().join("t.json").display());
+        for dest in ["null://", json.as_str()] {
+            let halt = Halt::new();
+            let s = unsynced_truehd_stream(5, 64).cancels(halt.clone(), 1);
+            let out = drive_mux(Box::new(s), dest, &Ctx::new(halt), None, None).unwrap();
+            assert!(out.halted && !out.completed && !out.output_opened, "{dest}");
+
+            let s = unsynced_truehd_stream(5, 64).halt_errs_at(1);
+            let out = drive_mux(Box::new(s), dest, &Ctx::default(), None, None).unwrap();
+            assert!(!out.completed && !out.output_opened, "{dest} Err(Halted)");
+        }
+    }
+
+    // The probe reads at most TRUEHD_PROBE_BYTES before the output opens; a title that never
+    // syncs is not buffered whole.
+    #[test]
+    fn the_truehd_probe_is_byte_capped() {
+        let mut s = unsynced_truehd_stream(20, 1 << 20);
+        let reads = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        s.read_observer = Some(reads.clone());
+        let at_open = Arc::new(std::sync::atomic::AtomicUsize::new(usize::MAX));
+        let (seen, counter) = (at_open.clone(), reads.clone());
+        let ctx = Ctx::default().with_events(Arc::new(move |e: &Event<'_>| {
+            if let Event::OutputOpened { .. } = e {
+                seen.store(counter.load(Ordering::SeqCst), Ordering::SeqCst);
+            }
+        }));
+        drive_mux(Box::new(s), "null://", &ctx, None, None).unwrap();
+        let n = at_open.load(Ordering::SeqCst);
+        assert!(n <= 10, "{n} MiB frames read before the output opened");
+    }
+
     fn mp2_ext_title() -> DiscTitle {
         use crate::disc::{AudioChannels, AudioStream, Codec, LabelPurpose, SampleRate, Stream};
         let audio = |pid, label: &str| {
@@ -1680,6 +1750,22 @@ mod tests {
         assert!(out.halted && !out.completed && !out.output_opened);
     }
 
+    // A Stop landing in `open_source` (the scan / bring-up) is the documented stopped
+    // outcome from `mux_url`, not an `E_HALTED` error; any other open failure stays an error.
+    #[test]
+    fn a_stop_during_mux_urls_open_is_a_halted_outcome() {
+        let ctx = Ctx::default();
+        ctx.halt.cancel();
+        let out = open_failed(Error::Halted, &ctx).expect("a Stop is not an error");
+        assert!(out.halted && !out.completed && !out.output_opened);
+        assert!(open_failed(Error::NoStreams, &ctx).is_err());
+        let live = Ctx::default();
+        assert!(
+            open_failed(Error::Halted, &live).is_err(),
+            "Err(Halted) with no Stop requested is not a stopped outcome"
+        );
+    }
+
     // ── halt mid-pump stops cleanly with completed=false, no panic. ──
     #[test]
     fn halt_mid_pump_stops_cleanly() {
@@ -1700,6 +1786,10 @@ mod tests {
         .expect("halt is a clean stop, not an error");
         assert!(!out.completed, "an interrupted mux is not complete");
         assert!(out.output_opened, "the sink was opened before the halt");
+        assert!(
+            out.halted,
+            "a Stop is reported as halted so callers map it to Cancelled"
+        );
     }
 
     // A halt that ends the stream as Ok(None) can also release the header gate
@@ -1740,6 +1830,7 @@ mod tests {
             None,
         )
         .expect("a halt mid frame-read is a clean stop, not an Err");
+        assert!(!out.halted, "halt was never cancelled: not a Stop");
         assert!(!out.completed, "interrupted mux is not complete");
         assert!(out.output_opened, "sink opened before the mid-read halt");
     }