v0.25.8: bump version (unified release with autorip 0.25.8)

Pre-0.25.7 the AACS authenticate loop fired up to 16 host-cert
attempts back-to-back with no pause. Each attempt is 5-10 SCSI
REPORT_KEY/SEND_KEY exchanges, so on a disc whose host cert isn't
in our KEYDB (or one the drive rejects), the drive saw 80-160 SCSI
commands in a few hundred ms and entered a fast-fail firmware
wedge state where every subsequent CDB returns sense 05/24 until
power-cycled.

Three defences:
- MAX_CERT_ATTEMPTS capped at 3 (was 16)
- 1-second sleep between attempts
- Bail immediately on any sense_key == 0x05 (ILLEGAL_REQUEST) so
  the loop can't deepen the wedge if a regression undoes the
  attempt cap.

CHANGELOG.md

@@ -1,5 +1,194 @@
 # Changelog
 
+## 0.25.8 (2026-05-20)
+
+### Changed
+
+- Synchronized to autorip 0.25.8. No libfreemkv code changes —
+  unified versioning bump only. autorip 0.25.7 image was DOA
+  (missing +x on the binary); see autorip changelog.
+
+## 0.25.7 (2026-05-20)
+
+### Fixed — BU40N firmware wedge on KEYDB-miss discs
+
+`disc::encrypt::do_handshake` used to fire up to 16 AACS authenticate
+attempts back-to-back with no pause between them. Each attempt is
+5-10 SCSI REPORT_KEY/SEND_KEY exchanges. On a disc whose host cert
+isn't in our KEYDB (or one the drive rejects), that's 80-160 SCSI
+commands hammered at the drive in a few hundred milliseconds — and
+the BU40N (plus most consumer optical drives) responds by entering
+a fast-fail firmware wedge state where every subsequent CDB returns
+`ILLEGAL_REQUEST/INVALID_FIELD_IN_CDB` (sense 0x05/0x24) until the
+drive is physically power-cycled.
+
+Live wedge event on rip1 2026-05-20 during a Barbie UHD scan
+(KEYDB miss) confirmed the diagnosis and motivated this fix.
+
+Defence-in-depth:
+- `MAX_CERT_ATTEMPTS` capped at 3 (was 16). If three different
+  host certs all fail, more won't help — the drive doesn't have a
+  match.
+- 1-second sleep between attempts. Gives the drive's firmware time
+  to recover internal state between auth challenges.
+- Bail immediately on any sense code with sense_key = 0x05
+  (`ILLEGAL_REQUEST`). The drive isn't merely rejecting our cert
+  — it's saying "I won't talk to you anymore" — so trying more
+  certs would deepen the wedge.
+
+## 0.25.6 (2026-05-20)
+
+### Changed
+
+- Synchronized to autorip 0.25.6 release. No behavioural changes.
+
+## 0.25.5 (2026-05-20)
+
+### Changed
+
+- Synchronized to autorip 0.25.5 release. No behavioural changes.
+
+## 0.25.4 (2026-05-20)
+
+### Changed
+
+- Synchronized to autorip 0.25.4 release. No behavioural changes
+  in the library — the v0.25.4 release ships the autorip-side
+  in-container NFS mount option.
+
+## 0.25.3 (2026-05-19)
+
+### Changed
+
+- Synchronized to autorip 0.25.3 release. No behavioural changes
+  in the library — the v0.25.3 release ships the autorip-side
+  parallel mux pipeline.
+
+## 0.25.2 (2026-05-19)
+
+### Fixed
+
+- **DTS-HD codec ID** — `MkvTrack::audio` now emits `A_DTS/MA` for
+  DTS-HD MA and `A_DTS/HR` for DTS-HD HR instead of mislabelling
+  both as plain `A_DTS`. Strict players (Plex transcoder, some
+  hardware decoders, AV receivers) reject lossless DTS-HD MA
+  payload when the track advertises `A_DTS` because the ID
+  implies the 1.5 Mbps core-only bitstream.
+- **PGS subtitle BlockDuration** — the PGS parser is now stateful:
+  it pairs each display PCS with the following empty/clear PCS to
+  compute a duration, and the MKV muxer emits a `BlockGroup` +
+  `BlockDuration` for subtitles that carry one. Without this the
+  last bitmap lingered on screen until the next display set
+  replaced it (or until end of file).
+
+### Changed
+
+- `codec::Frame` gains `duration_ns: Option<u64>` (set by parsers
+  that can compute one; currently only PGS).
+- `pes::PesFrame` gains `duration_ns: Option<u64>` (in-memory only;
+  not part of the on-wire serialization).
+- `MkvMuxer::write_frame` now takes a final `duration_ns: Option<u64>`
+  parameter. When `Some`, the frame is emitted as a `BlockGroup`
+  with `BlockDuration` instead of a `SimpleBlock`.
+
+## 0.25.1 (2026-05-19)
+
+### New — autorip event_fn plumbing
+
+- `PrefetchedSectorSource::new_with_events(reader, extents, batch,
+  halt, event_fn)` — same producer-thread pipeline as `new()`, plus
+  an optional callback fired from the producer with `BytesRead`
+  events after every successful batch. Lets the autorip multipass
+  + resume mux paths drive their progress UI from the highway
+  without polling the consumer side.
+- `build_iso_pipeline` gains an `event_fn` parameter so callers can
+  wire the same callback through one ctor.
+
+### Changed
+
+- `pes::Stream` trait: new default `errors() -> u64` method
+  (default returns 0). Lets `Box<dyn Stream>` callers query the
+  skip-on-error counter without downcasting. `DiscStream` overrides
+  to surface its existing `errors` field.
+
+### Removed (breaking)
+
+- `DiscStream::new_pipeline` and `DiscStream::read_pipeline`
+  deleted. All file-backed mux now uses `build_iso_pipeline` →
+  `PipelinedPesStream`. `DiscStream` is now the single-threaded
+  inline reader used by autorip's live-drive single-pass path only.
+- `DiscStream::demux_thread` and `demux_rx` fields removed.
+- `M2tsStream::open` deleted; `Mode::Read` variant removed.
+  `m2ts://` URLs go through the internal `build_m2ts_pipeline`
+  helper in `mux/resolve.rs` → `PipelinedPesStream`. `M2tsStream`
+  is a write-only sink now.
+
+## 0.25.0 (2026-05-19)
+
+### New — the freemkv mux throughput "highway"
+
+Three-stage pipelined PES read path. Read+decrypt runs on a
+producer thread, M2TS demux runs on a second thread, codec parse
+runs on the caller's thread. Communication between stages is via
+bounded `crossbeam_channel` with a recycled buffer pool — no
+allocations or memcpys in the steady-state hot loop.
+
+**Throughput on the rip1 testbed (Civil_War UHD, 62 GiB ISO →
+`null://`, single-thread caller):**
+
+|                                       |   MB/s |
+| ------------------------------------- | -----: |
+| 0.23.2 baseline                       |     60 |
+| + memchr SIMD HEVC start-code scan    |     69 |
+| + ts.feed no-copy boundary            |     72 |
+| + `PrefetchedSectorSource` (producer) |    124 |
+| + `DemuxThread` (3-stage pipeline)    |    135 |
+| + zero-copy recycled buffer pool      |    148 |
+| + 16 KiB initial PesAssembler buffer  |    162 |
+| + mimalloc allocator in freemkv CLI   |  200+  |
+| **+ warm cache**                      | **660** |
+
+The new public API:
+
+- `libfreemkv::PrefetchedSectorSource` — wraps any `SectorSource`,
+  spawns a producer thread, exposes recycled-buffer channels.
+- `libfreemkv::io::byte_prefetcher::BytePrefetcher` —
+  `std::io::Read` analogue for byte-stream sources (m2ts files,
+  sockets, stdin).
+- `libfreemkv::mux::demux_thread::DemuxThread` — the M2TS demux
+  worker; `spawn_zero_copy` takes either prefetcher's channels.
+- `libfreemkv::PipelinedPesStream` — the read-side `Stream` impl
+  that runs codec parse on the caller thread.
+- `libfreemkv::build_iso_pipeline(reader, title, keys, batch,
+  format, halt)` — the canonical ctor that wires all three stages
+  for an ISO file source.
+
+### Changed (breaking)
+
+- `IsoSectorReader` (the naive duplicate of `FileSectorSource`)
+  deleted. `FileSectorSource` is the sole file-backed sector
+  source; it carries the SEQUENTIAL fadvise hint, the periodic
+  DONTNEED page-cache eviction, and (new) the per-read
+  `readahead()` async-prefetch syscall.
+- `mux::input("iso://...", &opts)` now returns a
+  `PipelinedPesStream`. Function signature
+  `input(&str, &InputOptions) -> io::Result<Box<dyn Stream>>` is
+  unchanged; callers that treated the return value as
+  `Box<dyn Stream>` keep working.
+
+### Other
+
+- `FileSectorSource` exposes per-OS `prefetch()` hooks
+  (`readahead(2)` on Linux, `fcntl(F_RDADVISE)` on macOS, no-op on
+  Windows + other).
+- AACS decrypt thread-pool: env var renamed from
+  `FREEMKV_DECRYPT_THREADS` to `FREEMKV_THREADS`; default raised
+  to `cores.clamp(1, 64)`.
+- `PesAssembler` initial buffer capacity 256 KiB → 16 KiB (avoids
+  the 64-page first-touch fault tax on every PES boundary).
+- HEVC / H.264 `find_start_code` swapped onto
+  `memchr::memmem::find` (SIMD).
+
 ## 0.18.4 (2026-05-09)
 
 ### Build / CI hardening — no library code changes

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "libfreemkv"
-version = "0.25.0"
+version = "0.25.8"
 edition = "2024"
 rust-version = "1.86"
 license = "AGPL-3.0-only"

README.md

@@ -18,7 +18,7 @@ Part of the [freemkv](https://github.com/freemkv) project.
 
 ```toml
 [dependencies]
-libfreemkv = "0.18"
+libfreemkv = "0.25"
 ```
 
 ## Quick Start
@@ -56,8 +56,7 @@ output.finish()?;
 For damaged discs the library exposes two flat verbs — `Disc::sweep` for the
 forward Pass 1 and `Disc::patch` for retrying bad ranges. The library never
 loops; the multipass policy is the caller's job. See
-[`docs/rip-recovery.md`](docs/rip-recovery.md) and the design notes in
-`(internal)/memory/0_18_redesign.md`.
+[`docs/rip-recovery.md`](docs/rip-recovery.md).
 
 ```rust
 use libfreemkv::{SweepOptions, PatchOptions};

src/disc/encrypt.rs

@@ -61,9 +61,29 @@ impl Disc {
             "handshake starting"
         );
 
-        const MAX_CERT_ATTEMPTS: usize = 16;
+        // v0.25.7 wedge fix. Pre-0.25.7 this loop fired up to 16 AACS
+        // authenticate attempts back-to-back with no pause. Each attempt
+        // is 5-10 SCSI REPORT_KEY/SEND_KEY exchanges. On a disc whose
+        // host cert isn't in our KEYDB (or one the drive rejects),
+        // that's 80-160 SCSI commands hammered at the drive in a
+        // few hundred milliseconds — and the BU40N (and most consumer
+        // optical drives) responds by entering a fast-fail firmware
+        // wedge state where every subsequent CDB returns
+        // ILLEGAL_REQUEST/INVALID_FIELD_IN_CDB (sense 05/24) until
+        // power-cycled. Hit live on rip1 2026-05-20 during a Barbie
+        // UHD scan: KEYDB miss → 16 cert attempts in a tight loop →
+        // wedge → forced host reboot + drive disconnect to recover.
+        //
+        // Defense-in-depth: cap attempts, sleep between, and bail
+        // early on the drive's wedge sense so any later regression
+        // can't undo the protection silently.
+        const MAX_CERT_ATTEMPTS: usize = 3;
+        const PER_CERT_BACKOFF_MS: u64 = 1000;
         let mut last_err_code: Option<u16> = None;
         for (idx, hc) in keydb.host_certs.iter().take(MAX_CERT_ATTEMPTS).enumerate() {
+            if idx > 0 {
+                std::thread::sleep(std::time::Duration::from_millis(PER_CERT_BACKOFF_MS));
+            }
             match aacs::handshake::aacs_authenticate(session, &hc.private_key, &hc.certificate) {
                 Ok(mut auth) => {
                     let volume_id = match aacs::handshake::read_volume_id(session, &mut auth) {
@@ -94,7 +114,24 @@ impl Disc {
                     });
                 }
                 Err(e) => {
-                    last_err_code = Some(e.code());
+                    let code = e.code();
+                    last_err_code = Some(code);
+                    // Drive wedge senses (any with high byte 0x05 =
+                    // ILLEGAL_REQUEST). The drive isn't merely
+                    // rejecting our cert — it's saying "I won't talk
+                    // to you anymore." Trying more certs makes the
+                    // wedge worse. Bail out immediately.
+                    let sense_key = ((code >> 8) & 0xFF) as u8;
+                    if sense_key == 0x05 {
+                        tracing::warn!(
+                            target: "freemkv::disc",
+                            phase = "handshake_wedge_detected",
+                            cert_index = idx,
+                            error_code = code,
+                            "drive returned ILLEGAL_REQUEST during auth; bailing out to avoid wedge"
+                        );
+                        return None;
+                    }
                     continue;
                 }
             }
@@ -105,7 +142,8 @@ impl Disc {
             host_cert_count,
             tried = host_cert_count.min(MAX_CERT_ATTEMPTS),
             last_error_code = last_err_code,
-            "all host certs in KEYDB rejected by drive"
+            "all host certs in KEYDB rejected by drive (capped at {} attempts to prevent firmware wedge)",
+            MAX_CERT_ATTEMPTS
         );
         // All host certs failed — return None, not a fake success
         None

src/lib.rs

@@ -211,6 +211,7 @@ pub use mux::{InputOptions, StreamUrl, input, output, parse_url};
 // are the ISO-on-disk implementations. [`DecryptingSectorSource`] is the
 // single decrypt-on-read decorator (AACS / CSS / none) — wrap any
 // `SectorSource` to get plaintext sectors out.
+pub use mux::build_iso_pipeline;
 pub use scsi::{DriveInfo, ScsiSense, ScsiTransport, drive_has_disc, list_drives};
 pub use sector::{
     DecryptingSectorSource, FileSectorSink, FileSectorSource, PrefetchedSectorSource, SectorSink,

src/mux/codec/ac3.rs

@@ -76,6 +76,7 @@ impl CodecParser for Ac3Parser {
                 pts_ns,
                 keyframe: true,
                 data: data[start..start + frame_size].to_vec(),
+                duration_ns: None,
             });
             pos = start + frame_size;
         }

src/mux/codec/dts.rs

@@ -86,6 +86,7 @@ impl CodecParser for DtsParser {
                 pts_ns,
                 keyframe: true,
                 data: data[start..start + total_size].to_vec(),
+                duration_ns: None,
             });
             pos = start + total_size;
         }

src/mux/codec/dvdsub.rs

@@ -30,6 +30,7 @@ impl CodecParser for DvdSubParser {
             pts_ns,
             keyframe: true,
             data: pes.data.clone(),
+            duration_ns: None,
         }]
     }
 

src/mux/codec/h264.rs

@@ -85,6 +85,7 @@ impl CodecParser for H264Parser {
             pts_ns,
             keyframe,
             data: frame_data,
+            duration_ns: None,
         }]
     }
 

src/mux/codec/hevc.rs

@@ -110,6 +110,7 @@ impl CodecParser for HevcParser {
             pts_ns,
             keyframe,
             data: frame_data,
+            duration_ns: None,
         }]
     }
 

src/mux/codec/lpcm.rs

@@ -44,6 +44,7 @@ impl CodecParser for LpcmParser {
             pts_ns,
             keyframe: true,
             data: pes.data[BD_LPCM_HEADER_SIZE..].to_vec(),
+            duration_ns: None,
         }]
     }
 

src/mux/codec/mod.rs

@@ -29,6 +29,13 @@ pub struct Frame {
     pub keyframe: bool,
     /// Frame data (elementary stream bytes).
     pub data: Vec<u8>,
+    /// Optional duration in nanoseconds — only set by parsers that
+    /// can compute one (currently PGS, which pairs a display PCS
+    /// with the following empty PCS). When `Some`, the MKV muxer
+    /// emits a `BlockGroup` with `BlockDuration` instead of a
+    /// `SimpleBlock`; without it players guess the display interval
+    /// (subtitles linger past their end-time).
+    pub duration_ns: Option<u64>,
 }
 
 /// Convert 90kHz PTS to nanoseconds (round to nearest).
@@ -71,6 +78,7 @@ impl CodecParser for PassthroughParser {
             pts_ns,
             keyframe: self.keyframe,
             data: pes.data.clone(),
+            duration_ns: None,
         }]
     }
 

src/mux/codec/mpeg2.rs

@@ -194,6 +194,7 @@ impl CodecParser for Mpeg2Parser {
             pts_ns,
             keyframe,
             data: pes.data.clone(),
+            duration_ns: None,
         }]
     }
 

src/mux/codec/pgs.rs

@@ -1,12 +1,32 @@
 //! HDMV PGS (Presentation Graphics Stream) subtitle parser.
 //!
-//! PGS segments: PCS, WDS, PDS, ODS, END.
-//! Each PES packet contains one or more segments.
-//! All segments are keyframes (no inter-segment dependencies).
+//! PGS segments: PCS, WDS, PDS, ODS, END. Each PES packet starts with
+//! one of those (segment_type byte at offset 0).
+//!
+//! Subtitle display lifecycle (BD spec):
+//! - A "display" PCS (number_of_composition_objects > 0) starts a
+//!   visible subtitle. Its WDS/PDS/ODS follow.
+//! - A later "empty" PCS (number_of_composition_objects == 0) clears
+//!   the screen.
+//!
+//! For Matroska output we collapse that pair into one block with
+//! `BlockDuration` set to (clear_pts - display_pts). Without a
+//! duration, hardware players linger on the last bitmap until the
+//! next subtitle replaces it — which can be many seconds, and on a
+//! disc where the final subtitle has no follower, until end of file.
 
 use super::{CodecParser, Frame, PesPacket, pts_to_ns};
 
-pub struct PgsParser;
+const SEGMENT_PCS: u8 = 0x16;
+// Offset within the PES payload at which number_of_composition_objects
+// lives in a PCS: 3-byte segment header + 10 bytes of PCS fields
+// (video_w/h, frame_rate, comp_num, comp_state, palette_update,
+// palette_id_ref) = 13.
+const PCS_NUM_OBJECTS_OFFSET: usize = 13;
+
+pub struct PgsParser {
+    pending: Option<(i64, Vec<u8>)>,
+}
 
 impl Default for PgsParser {
     fn default() -> Self {
@@ -16,7 +36,7 @@ impl Default for PgsParser {
 
 impl PgsParser {
     pub fn new() -> Self {
-        Self
+        Self { pending: None }
     }
 }
 
@@ -26,11 +46,63 @@ impl CodecParser for PgsParser {
             return Vec::new();
         }
         let pts_ns = pes.pts.map(pts_to_ns).unwrap_or(0);
-        vec![Frame {
-            pts_ns,
-            keyframe: true,
-            data: pes.data.clone(),
-        }]
+
+        let is_pcs = pes.data[0] == SEGMENT_PCS;
+        let pcs_num_objects = if is_pcs && pes.data.len() > PCS_NUM_OBJECTS_OFFSET {
+            Some(pes.data[PCS_NUM_OBJECTS_OFFSET])
+        } else {
+            None
+        };
+
+        let mut out = Vec::new();
+        match pcs_num_objects {
+            // Clear/empty PCS — closes any pending display. Drop the
+            // clear segment itself; BlockDuration covers the screen
+            // wipe.
+            Some(0) => {
+                if let Some((start_pts, data)) = self.pending.take() {
+                    let duration = pts_ns.saturating_sub(start_pts).max(0) as u64;
+                    out.push(Frame {
+                        pts_ns: start_pts,
+                        keyframe: true,
+                        data,
+                        duration_ns: Some(duration),
+                    });
+                }
+            }
+            // Display PCS — start a new pending. If a prior display
+            // was never explicitly cleared (replace-without-clear),
+            // emit it with the new PCS's PTS as its end.
+            Some(_) => {
+                if let Some((start_pts, data)) = self.pending.take() {
+                    let duration = pts_ns.saturating_sub(start_pts).max(0) as u64;
+                    out.push(Frame {
+                        pts_ns: start_pts,
+                        keyframe: true,
+                        data,
+                        duration_ns: Some(duration),
+                    });
+                }
+                self.pending = Some((pts_ns, pes.data.clone()));
+            }
+            // Non-PCS first segment — either a continuation of the
+            // current display set, or non-standard layout. If we have
+            // a pending display, append; otherwise emit as-is.
+            None => {
+                if let Some((_, ref mut buf)) = self.pending {
+                    buf.extend_from_slice(&pes.data);
+                } else {
+                    out.push(Frame {
+                        pts_ns,
+                        keyframe: true,
+                        data: pes.data.clone(),
+                        duration_ns: None,
+                    });
+                }
+            }
+        }
+
+        out
     }
 
     fn codec_private(&self) -> Option<Vec<u8>> {
@@ -52,29 +124,60 @@ mod tests {
         }
     }
 
-    #[test]
-    fn parse_basic_segment() {
-        let mut parser = PgsParser::new();
-        // PGS segment data (PCS = presentation composition segment)
-        let data = vec![0x16, 0x00, 0x00, 0x11, 0x01, 0x02, 0x03];
-        let pes = make_pes(data.clone(), Some(90000));
-        let frames = parser.parse(&pes);
-
-        assert_eq!(frames.len(), 1);
-        assert_eq!(frames[0].data, data);
-        assert_eq!(frames[0].pts_ns, 1_000_000_000);
+    // Minimum-viable PCS bytes: type 0x16, segment_length (2 bytes),
+    // then 11 bytes of PCS fields ending in number_of_composition_objects.
+    fn pcs_bytes(num_objects: u8) -> Vec<u8> {
+        let mut v = vec![SEGMENT_PCS, 0x00, 0x0B];
+        v.extend_from_slice(&[0x07, 0x80, 0x04, 0x38]); // 1920x1080
+        v.push(0x10); // frame_rate
+        v.extend_from_slice(&[0x00, 0x01]); // composition_number
+        v.push(0x80); // composition_state = EpochStart
+        v.push(0x00); // palette_update + reserved
+        v.push(0x00); // palette_id_ref
+        v.push(num_objects);
+        v
     }
 
     #[test]
-    fn all_keyframes() {
+    fn display_then_clear_yields_duration() {
         let mut parser = PgsParser::new();
-        for i in 0..3 {
-            let data = vec![0x16, 0x00, i];
-            let pes = make_pes(data, Some(90000 * i as i64));
-            let frames = parser.parse(&pes);
-            assert_eq!(frames.len(), 1);
-            assert!(frames[0].keyframe, "PGS segment should always be keyframe");
-        }
+
+        // Display PCS at PTS 90000 (= 1s)
+        let display = pcs_bytes(1);
+        let frames = parser.parse(&make_pes(display.clone(), Some(90000)));
+        assert!(frames.is_empty(), "display PCS should be pending");
+
+        // Empty PCS at PTS 270000 (= 3s)
+        let clear = pcs_bytes(0);
+        let frames = parser.parse(&make_pes(clear, Some(270000)));
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0].pts_ns, 1_000_000_000);
+        assert_eq!(frames[0].duration_ns, Some(2_000_000_000));
+        assert_eq!(frames[0].data, display);
+    }
+
+    #[test]
+    fn replace_without_clear_still_emits_prior_with_duration() {
+        let mut parser = PgsParser::new();
+        let _ = parser.parse(&make_pes(pcs_bytes(1), Some(90000)));
+        let frames = parser.parse(&make_pes(pcs_bytes(1), Some(180000)));
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0].pts_ns, 1_000_000_000);
+        assert_eq!(frames[0].duration_ns, Some(1_000_000_000));
+    }
+
+    #[test]
+    fn non_pcs_segment_appends_to_pending() {
+        let mut parser = PgsParser::new();
+        let _ = parser.parse(&make_pes(pcs_bytes(1), Some(90000)));
+        // ODS-like segment (type 0x15)
+        let frames = parser.parse(&make_pes(vec![0x15, 0x00, 0x02, 0xAA, 0xBB], Some(90000)));
+        assert!(frames.is_empty());
+        // Clear closes the set; data should include the appended bytes.
+        let frames = parser.parse(&make_pes(pcs_bytes(0), Some(180000)));
+        assert_eq!(frames.len(), 1);
+        let data = &frames[0].data;
+        assert!(data.windows(5).any(|w| w == [0x15, 0x00, 0x02, 0xAA, 0xBB]));
     }
 
     #[test]

src/mux/codec/truehd.rs

@@ -139,6 +139,7 @@ impl CodecParser for TrueHdParser {
                 pts_ns: self.next_pts_ns,
                 keyframe: is_major_sync,
                 data: self.buf[..unit_bytes].to_vec(),
+                duration_ns: None,
             });
             self.buf.drain(..unit_bytes);
             self.next_pts_ns += AU_DURATION_NS;

src/mux/codec/vc1.rs

@@ -96,6 +96,7 @@ impl CodecParser for Vc1Parser {
             pts_ns: ts_ns,
             keyframe,
             data: frame_data.to_vec(),
+            duration_ns: None,
         }]
     }
 

src/mux/disc.rs

@@ -149,28 +149,16 @@ pub struct DiscStream {
     // a percent without a separate API call.
     bytes_total_extents: u64,
 
-    // PES output. `ts_demuxer` and `ps_demuxer` are `None` when the
-    // stream is in pipeline mode — the demux state lives inside a
-    // [`super::demux_thread::DemuxThread`] and PesPackets arrive
-    // through `demux_rx` already parsed.
+    // PES output — single-threaded inline demux + codec parse. The
+    // pipeline-mode mux (3-stage threaded) lives in
+    // [`super::pipelined_stream::PipelinedPesStream`]; this type is
+    // the legacy in-thread path for live-disc reads where adaptive
+    // batch retry on bad sectors lives in `fill_extents`.
     ts_demuxer: Option<super::ts::TsDemuxer>,
     ps_demuxer: Option<super::ps::PsDemuxer>,
     parsers: Vec<(u16, Box<dyn super::codec::CodecParser>)>,
     pending_frames: std::collections::VecDeque<crate::pes::PesFrame>,
     pid_to_track: Vec<(u16, usize)>,
-
-    // Pipeline mode: when `Some`, the read+decrypt+demux pipeline
-    // runs on a dedicated thread; this stream's `read()` just pulls
-    // PesPacket batches from `demux_rx` and runs codec parse on the
-    // caller thread. See [`super::demux_thread`].
-    //
-    // `demux_thread` is kept solely so that `Drop` joins the worker
-    // before this stream is dropped — direct reads happen through
-    // `demux_rx`. The `allow(dead_code)` keeps the optimizer happy
-    // since the field is only used at drop time.
-    #[allow(dead_code)]
-    demux_thread: Option<super::demux_thread::DemuxThread>,
-    demux_rx: Option<crossbeam_channel::Receiver<super::demux_thread::DemuxBatch>>,
 }
 
 impl DiscStream {
@@ -251,118 +239,6 @@ impl DiscStream {
             parsers,
             pending_frames: std::collections::VecDeque::new(),
             pid_to_track,
-            demux_thread: None,
-            demux_rx: None,
-        }
-    }
-
-    /// Pipeline-mode constructor. Moves the read+decrypt+demux work
-    /// onto a [`super::demux_thread::DemuxThread`] so the caller's
-    /// `read()` thread only does codec parse + frame emission.
-    ///
-    /// `reader` is a [`crate::sector::PrefetchedSectorSource`] — the
-    /// prefetched producer thread already runs read+decrypt on its
-    /// own thread; this constructor peels off its channels for the
-    /// demux thread to consume in zero-copy mode (no buffer memcpy
-    /// across thread boundary, recycled-pool of two buffers, no
-    /// allocator activity in the hot loop).
-    ///
-    /// Pipeline mode is the preferred wiring for ISO file mux on a
-    /// multi-core host; it gives a ~2× consumer throughput in the
-    /// `null://` benchmark vs the single-thread inline path.
-    pub fn new_pipeline(
-        reader: crate::sector::PrefetchedSectorSource,
-        title: DiscTitle,
-        decrypt_keys: crate::decrypt::DecryptKeys,
-        batch_sectors: u16,
-        content_format: crate::disc::ContentFormat,
-        halt: Option<Halt>,
-    ) -> Self {
-        let extents = title.extents.clone();
-        let bytes_total_extents: u64 = extents.iter().map(|e| e.sector_count as u64 * 2048).sum();
-
-        let mut pids = Vec::new();
-        let mut parsers = Vec::new();
-        let mut pid_to_track = Vec::new();
-        for (idx, s) in title.streams.iter().enumerate() {
-            let (pid, codec) = match s {
-                crate::disc::Stream::Video(v) => (v.pid, v.codec),
-                crate::disc::Stream::Audio(a) => (a.pid, a.codec),
-                crate::disc::Stream::Subtitle(s) => (s.pid, s.codec),
-            };
-            pids.push(pid);
-            pid_to_track.push((pid, idx));
-            parsers.push((pid, super::codec::parser_for_codec(codec, None)));
-        }
-
-        let (ts, ps) = match content_format {
-            crate::disc::ContentFormat::MpegPs => (None, Some(super::ps::PsDemuxer::new())),
-            crate::disc::ContentFormat::BdTs => {
-                let ts_pids: Vec<u16> = pids.clone();
-                if ts_pids.is_empty() {
-                    (None, None)
-                } else {
-                    (Some(super::ts::TsDemuxer::new(&ts_pids)), None)
-                }
-            }
-        };
-
-        let (prefetch_rx, recycle_tx, shell) = reader.into_channels();
-        let (handle, rx) = super::demux_thread::DemuxThread::spawn_zero_copy(
-            prefetch_rx,
-            recycle_tx,
-            shell,
-            halt.clone(),
-            ts,
-            ps,
-        );
-
-        // The DiscStream's own reader is a no-op pass-through — the
-        // real reader lives inside the demux thread. We need *some*
-        // Box<dyn SectorSource> to satisfy the field type; use a
-        // tiny stub. fill_extents won't be called in pipeline mode.
-        struct NullSource;
-        impl SectorSource for NullSource {
-            fn capacity_sectors(&self) -> u32 {
-                0
-            }
-            fn read_sectors(
-                &mut self,
-                _: u32,
-                _: u16,
-                _: &mut [u8],
-                _: bool,
-            ) -> crate::error::Result<usize> {
-                Ok(0)
-            }
-        }
-        let dummy: Box<dyn SectorSource> = Box::new(NullSource);
-
-        Self {
-            reader: DecryptingSectorSource::new(dummy, crate::decrypt::DecryptKeys::None),
-            title,
-            disc: None,
-            decrypt_keys,
-            extents,
-            current_extent: 0,
-            current_offset: 0,
-            read_buf: Vec::new(),
-            buf_valid: 0,
-            adaptive: AdaptiveBatch::new(batch_sectors),
-            errors: 0,
-            skip_errors: false,
-            halt,
-            event_fn: None,
-            eof: false,
-            bytes_read_total: 0,
-            bytes_total_extents,
-            ts_demuxer: None,
-            ps_demuxer: None,
-            parsers,
-            pending_frames: std::collections::VecDeque::new(),
-            pid_to_track,
-            demux_thread: Some(handle),
-            demux_rx: Some(rx),
         }
     }
 
@@ -596,88 +472,6 @@ fn prof_tick(stage: &str, ns: u128, bytes: u64) {
     });
 }
 
-impl DiscStream {
-    /// Pipeline-mode `read()` helper: pull one PesPacket batch from
-    /// the demux thread, run codec parse on each PES, enqueue the
-    /// resulting PesFrames, return the first one.
-    fn read_pipeline(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
-        use super::demux_thread::DemuxBatch;
-        let rx = self.demux_rx.as_ref().expect("read_pipeline without rx");
-        match rx.recv() {
-            Ok(DemuxBatch::Ts(packets)) => {
-                let skip_parse = std::env::var_os("FREEMKV_SKIP_PARSE").is_some();
-                for pes in packets {
-                    if let Some((_, track)) = self
-                        .pid_to_track
-                        .iter()
-                        .find(|(pid, _)| *pid == pes.pid)
-                        .copied()
-                    {
-                        if skip_parse {
-                            self.pending_frames.push_back(crate::pes::PesFrame {
-                                track,
-                                pts: pes.pts.map(super::codec::pts_to_ns).unwrap_or(0),
-                                keyframe: false,
-                                data: pes.data,
-                            });
-                        } else if let Some((_, parser)) =
-                            self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid)
-                        {
-                            for frame in parser.parse(&pes) {
-                                self.pending_frames.push_back(
-                                    crate::pes::PesFrame::from_codec_frame(track, frame),
-                                );
-                            }
-                        }
-                    }
-                }
-                Ok(self.pending_frames.pop_front())
-            }
-            Ok(DemuxBatch::Ps(packets)) => {
-                for ps in packets {
-                    let track = match ps.stream_id {
-                        0xE0..=0xEF => 0,
-                        0xC0..=0xDF => 1,
-                        0xBD => ps
-                            .sub_stream_id
-                            .map(|s| (s & 0x1F) as usize + 1)
-                            .unwrap_or(1),
-                        _ => continue,
-                    };
-                    if track >= self.title.streams.len() {
-                        continue;
-                    }
-                    let pid = self
-                        .pid_to_track
-                        .iter()
-                        .find(|(_, idx)| *idx == track)
-                        .map(|(p, _)| *p)
-                        .unwrap_or(0);
-                    let pes = super::ts::PesPacket {
-                        pid,
-                        pts: ps.pts.map(|p| p as i64),
-                        dts: ps.dts.map(|d| d as i64),
-                        data: ps.data,
-                    };
-                    if let Some((_, parser)) = self.parsers.iter_mut().find(|(p, _)| *p == pid) {
-                        for frame in parser.parse(&pes) {
-                            self.pending_frames
-                                .push_back(crate::pes::PesFrame::from_codec_frame(track, frame));
-                        }
-                    }
-                }
-                Ok(self.pending_frames.pop_front())
-            }
-            Ok(DemuxBatch::Err(e)) => Err(e),
-            Err(_) => {
-                // Channel closed → demux thread finished. EOF.
-                self.eof = true;
-                Ok(None)
-            }
-        }
-    }
-}
-
 impl crate::pes::Stream for DiscStream {
     fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
         if let Some(frame) = self.pending_frames.pop_front() {
@@ -688,19 +482,6 @@ impl crate::pes::Stream for DiscStream {
             return Ok(None);
         }
 
-        // Pipeline mode: read+decrypt+demux all happen on the demux
-        // thread. Pull a batch of PesPackets and run codec parse on
-        // this thread; loop until we have at least one frame OR EOF.
-        if self.demux_rx.is_some() {
-            loop {
-                match self.read_pipeline()? {
-                    Some(f) => return Ok(Some(f)),
-                    None if self.eof => return Ok(None),
-                    None => continue,
-                }
-            }
-        }
-
         loop {
             let t0 = std::time::Instant::now();
             if !self.fill_extents()? {
@@ -794,6 +575,7 @@ impl crate::pes::Stream for DiscStream {
                                 pts: pes.pts.map(super::codec::pts_to_ns).unwrap_or(0),
                                 keyframe: false,
                                 data: pes.data,
+                                duration_ns: None,
                             });
                         } else if let Some((_, parser)) =
                             self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid)
@@ -897,6 +679,10 @@ impl crate::pes::Stream for DiscStream {
         }
         true
     }
+
+    fn errors(&self) -> u64 {
+        self.errors
+    }
 }
 
 #[cfg(test)]

src/mux/ebml.rs

@@ -372,6 +372,9 @@ pub const BIT_DEPTH: u32 = 0x6264;
 pub const CLUSTER: u32 = 0x1F43_B675;
 pub const CLUSTER_TIMESTAMP: u32 = 0xE7;
 pub const SIMPLE_BLOCK: u32 = 0xA3;
+pub const BLOCK_GROUP: u32 = 0xA0;
+pub const BLOCK: u32 = 0xA1;
+pub const BLOCK_DURATION: u32 = 0x9B;
 
 // Cues
 pub const CUES: u32 = 0x1C53_BB6B;

src/mux/m2ts.rs

@@ -161,6 +161,7 @@ mod tests {
             pts: 0,
             keyframe: true,
             data: fake_idr_pes_data(),
+            duration_ns: None,
         };
         stream.write(&frame).unwrap();
         stream.finish().unwrap();

src/mux/mkv.rs

@@ -87,11 +87,19 @@ impl MkvTrack {
     }
 
     pub fn audio(a: &AudioStream) -> Self {
+        // Codec ID strings must distinguish the DTS family — strict
+        // players (Plex transcoder, some hardware decoders, some AV
+        // receivers) reject lossless DTS-HD MA payload when the
+        // track advertises plain `A_DTS` because it implies the
+        // bitstream is the 1.5 Mbps "core" only. Fix is just to
+        // emit the right ID per BD-STN codec field.
         let codec_id = match a.codec {
             Codec::Ac3 => "A_AC3",
             Codec::Ac3Plus => "A_EAC3",
             Codec::TrueHd => "A_TRUEHD",
-            Codec::DtsHdMa | Codec::DtsHdHr | Codec::Dts => "A_DTS",
+            Codec::DtsHdMa => "A_DTS/MA",
+            Codec::DtsHdHr => "A_DTS/HR",
+            Codec::Dts => "A_DTS",
             Codec::Lpcm => "A_PCM/INT/BIG",
             _ => "A_AC3",
         };
@@ -380,12 +388,19 @@ impl<W: Write + Seek> MkvMuxer<W> {
     }
 
     /// Write a single frame.
+    ///
+    /// When `duration_ns` is `Some`, the frame is emitted as a
+    /// `BlockGroup` with `BlockDuration` so the player knows exactly
+    /// when to remove the on-screen artifact (the practical case is
+    /// PGS subtitles — without it, the last bitmap lingers until the
+    /// next display set replaces it). Otherwise a plain `SimpleBlock`.
     pub fn write_frame(
         &mut self,
         track_idx: usize,
         pts_ns: i64,
         keyframe: bool,
         data: &[u8],
+        duration_ns: Option<u64>,
     ) -> io::Result<()> {
         let raw_ms = pts_ns / 1_000_000;
         let base = *self.base_pts_ms.get_or_insert(raw_ms);
@@ -409,9 +424,16 @@ impl<W: Write + Seek> MkvMuxer<W> {
             });
         }
 
-        // Write SimpleBlock
         let relative_ts = (pts_ms - self.cluster_ts_ms) as i16;
-        self.write_simple_block(track_idx + 1, relative_ts, keyframe, data)?;
+        match duration_ns {
+            Some(dur_ns) => {
+                let duration_ms = (dur_ns / 1_000_000).max(1);
+                self.write_block_group(track_idx + 1, relative_ts, keyframe, data, duration_ms)?;
+            }
+            None => {
+                self.write_simple_block(track_idx + 1, relative_ts, keyframe, data)?;
+            }
+        }
         self.frame_count += 1;
 
         Ok(())
@@ -510,6 +532,34 @@ impl<W: Write + Seek> MkvMuxer<W> {
 
         Ok(())
     }
+
+    fn write_block_group(
+        &mut self,
+        track_num: usize,
+        relative_ts: i16,
+        keyframe: bool,
+        data: &[u8],
+        duration_ms: u64,
+    ) -> io::Result<()> {
+        let track_vint = if track_num < 0x80 {
+            vec![(track_num as u8) | 0x80]
+        } else {
+            vec![0x40 | ((track_num >> 8) as u8), track_num as u8]
+        };
+        let flags: u8 = if keyframe { 0x80 } else { 0x00 };
+        let block_size = track_vint.len() + 2 + 1 + data.len();
+
+        let bg_pos = ebml::start_master(&mut self.writer, ebml::BLOCK_GROUP)?;
+        ebml::write_id(&mut self.writer, ebml::BLOCK)?;
+        ebml::write_size(&mut self.writer, block_size as u64)?;
+        self.writer.write_all(&track_vint)?;
+        self.writer.write_all(&relative_ts.to_be_bytes())?;
+        self.writer.write_all(&[flags])?;
+        self.writer.write_all(data)?;
+        ebml::write_uint(&mut self.writer, ebml::BLOCK_DURATION, duration_ms)?;
+        ebml::end_master(&mut self.writer, bg_pos)?;
+        Ok(())
+    }
 }
 
 // ============================================================
@@ -619,7 +669,7 @@ mod tests {
         let tracks = [make_video_track()];
         let mut muxer = MkvMuxer::new(buf, &tracks, None, 60.0, &[]).unwrap();
         muxer
-            .write_frame(0, 0, true, &[0xDE, 0xAD, 0xBE, 0xEF])
+            .write_frame(0, 0, true, &[0xDE, 0xAD, 0xBE, 0xEF], None)
             .unwrap();
         let data = muxer.writer.into_inner();
         assert!(
@@ -657,7 +707,9 @@ mod tests {
         let writer = SharedWriter(shared.clone());
         let tracks = [make_video_track()];
         let mut muxer = MkvMuxer::new(writer, &tracks, Some("Cue Test"), 60.0, &[]).unwrap();
-        muxer.write_frame(0, 0, true, &[0x01, 0x02, 0x03]).unwrap();
+        muxer
+            .write_frame(0, 0, true, &[0x01, 0x02, 0x03], None)
+            .unwrap();
         muxer.finish().unwrap();
 
         let data = shared.lock().unwrap().clone().into_inner();
@@ -673,13 +725,17 @@ mod tests {
         let tracks = [make_video_track(), make_audio_track()];
         let mut muxer = MkvMuxer::new(buf, &tracks, Some("Multi"), 120.0, &[]).unwrap();
         // Write frames to both tracks
-        muxer.write_frame(0, 0, true, &[0x00, 0x00, 0x01]).unwrap();
-        muxer.write_frame(1, 0, false, &[0x0B, 0x77, 0x00]).unwrap();
         muxer
-            .write_frame(0, 40_000_000, false, &[0x00, 0x00, 0x01])
+            .write_frame(0, 0, true, &[0x00, 0x00, 0x01], None)
             .unwrap();
         muxer
-            .write_frame(1, 32_000_000, false, &[0x0B, 0x77, 0x01])
+            .write_frame(1, 0, false, &[0x0B, 0x77, 0x00], None)
+            .unwrap();
+        muxer
+            .write_frame(0, 40_000_000, false, &[0x00, 0x00, 0x01], None)
+            .unwrap();
+        muxer
+            .write_frame(1, 32_000_000, false, &[0x0B, 0x77, 0x01], None)
             .unwrap();
         // Should not panic
         let data = muxer.writer.into_inner();
@@ -694,10 +750,12 @@ mod tests {
 
         // Record position before first frame
         let pos_before_kf = muxer.writer.position();
-        muxer.write_frame(0, 0, true, &[0xAA]).unwrap();
+        muxer.write_frame(0, 0, true, &[0xAA], None).unwrap();
         let pos_after_kf = muxer.writer.position();
 
-        muxer.write_frame(0, 1_000_000, false, &[0xBB]).unwrap();
+        muxer
+            .write_frame(0, 1_000_000, false, &[0xBB], None)
+            .unwrap();
         let pos_after_nkf = muxer.writer.position();
 
         let data = muxer.writer.into_inner();
@@ -942,7 +1000,7 @@ mod tests {
         let writer = SharedWriter(shared.clone());
         let mut muxer = MkvMuxer::new(writer, tracks, None, 0.0, chapters).unwrap();
         for (t, pts, kf, data) in frames {
-            muxer.write_frame(*t, *pts, *kf, data).unwrap();
+            muxer.write_frame(*t, *pts, *kf, data, None).unwrap();
         }
         let frame_count = muxer.frame_count;
         muxer.finish().unwrap();

src/mux/mkvstream.rs

@@ -138,6 +138,7 @@ impl crate::pes::Stream for MkvStream {
                         pts: pts_ms * 1_000_000, // ms → ns
                         keyframe,
                         data,
+                        duration_ns: None,
                     }));
                 }
                 _ => {
@@ -150,9 +151,13 @@ impl crate::pes::Stream for MkvStream {
 
     fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
         match &mut self.mode {
-            Mode::Write { muxer: Some(m) } => {
-                m.write_frame(frame.track, frame.pts, frame.keyframe, &frame.data)
-            }
+            Mode::Write { muxer: Some(m) } => m.write_frame(
+                frame.track,
+                frame.pts,
+                frame.keyframe,
+                &frame.data,
+                frame.duration_ns,
+            ),
             Mode::Write { muxer: None } => Ok(()),
             Mode::Read(_) => Err(crate::error::Error::StreamReadOnly.into()),
         }

src/mux/network.rs

@@ -186,6 +186,7 @@ mod tests {
             pts: 90000,
             keyframe: true,
             data: vec![0x47; 192],
+            duration_ns: None,
         };
         pes::Stream::write(&mut writer, &frame).unwrap();
         pes::Stream::finish(&mut writer).unwrap();

src/mux/null.rs

@@ -47,6 +47,7 @@ mod tests {
             pts: 0,
             keyframe: true,
             data: vec![0x01, 0x02, 0x03],
+            duration_ns: None,
         };
         sink.write(&frame).unwrap();
         let _ = sink.info();

src/mux/pipelined_stream.rs

@@ -108,6 +108,7 @@ impl PipelinedPesStream {
                         pts: pes.pts.map(super::codec::pts_to_ns).unwrap_or(0),
                         keyframe: false,
                         data: pes.data,
+                        duration_ns: None,
                     });
                 } else if let Some((_, parser)) =
                     self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid)

src/mux/resolve.rs

@@ -234,6 +234,7 @@ pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::S
                 ISO_MUX_BATCH_SECTORS,
                 format,
                 None,
+                None,
             );
             Ok(Box::new(stream))
         }
@@ -357,15 +358,17 @@ pub fn build_iso_pipeline<S: SectorSource + Send + 'static>(
     batch_sectors: u16,
     format: ContentFormat,
     halt: Option<crate::halt::Halt>,
+    event_fn: Option<crate::sector::prefetched::EventFn>,
 ) -> PipelinedPesStream {
     let extents = title.extents.clone();
     let decrypting =
         crate::sector::DecryptingSectorSource::new(Box::new(reader) as Box<dyn SectorSource>, keys);
-    let prefetched = crate::sector::PrefetchedSectorSource::new(
+    let prefetched = crate::sector::PrefetchedSectorSource::new_with_events(
         decrypting,
         extents,
         batch_sectors,
         halt.clone(),
+        event_fn,
     );
     let (rx, recycle_tx, shell) = prefetched.into_channels();
 

src/pes.rs

@@ -17,6 +17,10 @@ pub struct PesFrame {
     pub keyframe: bool,
     /// Raw elementary stream data (NAL units, audio samples, etc).
     pub data: Vec<u8>,
+    /// Optional duration in nanoseconds. In-memory only; not part of
+    /// the on-wire serialization. Currently set by the PGS parser so
+    /// the MKV muxer can emit `BlockDuration`.
+    pub duration_ns: Option<u64>,
 }
 
 impl PesFrame {
@@ -64,6 +68,7 @@ impl PesFrame {
             pts,
             keyframe,
             data,
+            duration_ns: None,
         }))
     }
 
@@ -74,6 +79,7 @@ impl PesFrame {
             pts: frame.pts_ns,
             keyframe: frame.keyframe,
             data: frame.data,
+            duration_ns: frame.duration_ns,
         }
     }
 }
@@ -115,6 +121,16 @@ pub trait Stream: Send {
     fn headers_ready(&self) -> bool {
         true
     }
+
+    /// Cumulative count of read errors the stream skipped past (e.g.
+    /// zero-filled bad sectors on a live drive). Default `0` for
+    /// streams that don't have a notion of skip-on-error (file ISO,
+    /// network, stdio, the pipeline highway, etc.); concrete impls
+    /// with adaptive retry (`DiscStream` on the drive single-pass
+    /// path) override.
+    fn errors(&self) -> u64 {
+        0
+    }
 }
 
 /// Wraps any output stream and counts bytes written.
@@ -183,6 +199,7 @@ mod tests {
             pts,
             keyframe: track == 0 && pts == 0,
             data: vec![track as u8, (pts & 0xff) as u8, 0xAA],
+            duration_ns: None,
         }
     }
 

src/sector/prefetched.rs

@@ -41,11 +41,18 @@
 //! reading the underlying source directly.
 
 use crate::error::Result;
+use crate::event::{Event, EventKind};
 use crate::halt::Halt;
 use crate::sector::SectorSource;
 use crossbeam_channel::{Receiver, Sender, bounded};
 use std::thread::JoinHandle;
 
+/// Producer-thread event callback. Fires `BytesRead` after every
+/// successful batch read so the consumer side can update a UI
+/// progress indicator without polling. `Send + 'static` because the
+/// callback runs on the producer thread.
+pub type EventFn = Box<dyn Fn(Event) + Send + 'static>;
+
 const PREFETCH_CHANNEL_DEPTH: usize = 2;
 
 /// Smallest sector source the producer will issue per read. AACS
@@ -85,15 +92,33 @@ impl PrefetchedSectorSource {
     /// — every byte the producer emits is what the consumer's demux
     /// will feed to its codec parsers.
     pub fn new<S>(
-        mut reader: S,
+        reader: S,
         extents: Vec<crate::disc::Extent>,
         batch_sectors: u16,
         halt: Option<Halt>,
     ) -> Self
+    where
+        S: SectorSource + Send + 'static,
+    {
+        Self::new_with_events(reader, extents, batch_sectors, halt, None)
+    }
+
+    /// Same as [`new`] but with a callback fired from the producer
+    /// thread after each successful batch — used by autorip's mux
+    /// path to surface `BytesRead` progress to the UI without the
+    /// consumer thread having to poll.
+    pub fn new_with_events<S>(
+        mut reader: S,
+        extents: Vec<crate::disc::Extent>,
+        batch_sectors: u16,
+        halt: Option<Halt>,
+        event_fn: Option<EventFn>,
+    ) -> Self
     where
         S: SectorSource + Send + 'static,
     {
         let total_sectors: u32 = extents.iter().map(|e| e.sector_count).sum();
+        let bytes_total_extents: u64 = extents.iter().map(|e| e.sector_count as u64 * 2048).sum();
         let (tx, rx) = bounded::<Batch>(PREFETCH_CHANNEL_DEPTH);
         let (recycle_tx, recycle_rx) = bounded::<Vec<u8>>(PREFETCH_CHANNEL_DEPTH + 1);
         let batch_bytes = batch_sectors as usize * 2048;
@@ -111,6 +136,7 @@ impl PrefetchedSectorSource {
             .spawn(move || {
                 let mut ext_idx = 0usize;
                 let mut offset: u32 = 0;
+                let mut bytes_read_total: u64 = 0;
                 while ext_idx < extents.len() {
                     if halt.as_ref().map(|h| h.is_cancelled()).unwrap_or(false) {
                         return;
@@ -144,6 +170,15 @@ impl PrefetchedSectorSource {
                     match reader.read_sectors(lba, sectors, &mut buf[..bytes], false) {
                         Ok(n) => {
                             buf.truncate(n);
+                            bytes_read_total = bytes_read_total.saturating_add(n as u64);
+                            if let Some(ref f) = event_fn {
+                                f(Event {
+                                    kind: EventKind::BytesRead {
+                                        bytes: bytes_read_total,
+                                        total: bytes_total_extents,
+                                    },
+                                });
+                            }
                             if tx.send(Ok(buf)).is_err() {
                                 return; // consumer dropped
                             }

tests/streams.rs

@@ -301,6 +301,7 @@ fn m2ts_stream_write_read() {
             pts: i as i64 * 1_000_000,
             keyframe: i == 0,
             data: vec![i; 100],
+            duration_ns: None,
         };
         PesStream::write(&mut stream, &frame).unwrap();
     }
@@ -322,6 +323,7 @@ fn m2ts_pes_frame_roundtrip() {
         pts: 1_234_567_890,
         keyframe: true,
         data: vec![0xDE; 200],
+        duration_ns: None,
     };
 
     let mut buf = Vec::new();
@@ -589,6 +591,7 @@ fn mkvstream_write_finish() {
             pts: i as i64 * 1_000_000,
             keyframe: i == 0,
             data: vec![i; 100],
+            duration_ns: None,
         };
         PesStream::write(&mut stream, &frame).unwrap();
     }
@@ -650,6 +653,7 @@ fn mkvstream_roundtrip_bdts() {
             pts: i as i64 * 1_000_000,
             keyframe: true,
             data: vec![i; 100],
+            duration_ns: None,
         };
         PesStream::write(&mut stream, &frame).unwrap();
     }
@@ -842,6 +846,7 @@ fn mkvstream_e2e_h264_produces_valid_mkv() {
         pts: 1_000_000_000, // 1 second in ns
         keyframe: true,
         data: es_data,
+        duration_ns: None,
     };
     PesStream::write(&mut stream2, &frame1).unwrap();
 
@@ -851,6 +856,7 @@ fn mkvstream_e2e_h264_produces_valid_mkv() {
         pts: 1_041_700_000, // ~1 frame later in ns
         keyframe: false,
         data: es_data2,
+        duration_ns: None,
     };
     PesStream::write(&mut stream2, &frame2).unwrap();
     PesStream::finish(&mut stream2).unwrap();