v0.20.3: add halt check to Disc::patch backtrack inner loop

WO-5 (partial): the patch backtrack inner loop ('while bt_pos <
backtrack_end' in disc/patch.rs) issues per-sector reads to fill the
gap created by a damage-window skip. A long backtrack span can run
minutes; without an inline halt poll, the outer halt only takes
effect when control returns to the per-range loop. Adds a halt poll
at the top of each iteration so cancellation propagates inside the
backtrack span.

Per-sector read failures inside the backtrack already drop through
to the main fail path; this only changes the cancellation latency
between an /api/stop call and the producer actually unwinding. Drops
worst-case unwind from 'whole backtrack span × per-sector recovery
timeout' (minutes) to 'one in-flight SCSI command' (seconds).

The broader Arc<AtomicBool> → Halt migration on CopyOptions /
SweepOptions / PatchOptions / Drive::halt and Pipeline::send halt-
awareness is deferred — separate cycle, larger API impact.

Cargo.toml

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

src/aacs/keys.rs

@@ -247,6 +247,7 @@ fn validate_processing_key(
 /// Find Verify Media Key Record (type 0x10) in MKB.
 fn mkb_find_mk_dv(mkb: &[u8]) -> Option<[u8; 16]> {
     let mut pos = 0;
+    let mut type10_seen: Vec<(usize, usize)> = Vec::new();
     while pos + 4 <= mkb.len() {
         let rec_type = mkb[pos];
         let rec_len = u32::from_be_bytes([0, mkb[pos + 1], mkb[pos + 2], mkb[pos + 3]]) as usize;
@@ -254,14 +255,32 @@ fn mkb_find_mk_dv(mkb: &[u8]) -> Option<[u8; 16]> {
             break;
         }
 
+        if rec_type == 0x10 {
+            type10_seen.push((pos, rec_len));
+        }
+
         if rec_type == 0x10 && rec_len >= 20 {
             // mk_dv is at offset 4 (after record header)
             let mut dv = [0u8; 16];
             dv.copy_from_slice(&mkb[pos + 4..pos + 20]);
+            tracing::warn!(
+                target: "freemkv::disc",
+                phase = "mkb_mk_dv_found",
+                pos,
+                rec_len,
+                "mk_dv extracted from MKB"
+            );
             return Some(dv);
         }
         pos += rec_len;
     }
+    tracing::warn!(
+        target: "freemkv::disc",
+        phase = "mkb_mk_dv_not_found",
+        type10_seen = ?type10_seen,
+        scanned_bytes = pos,
+        "no 0x10 record with rec_len>=20 found"
+    );
     None
 }
 
@@ -628,32 +647,70 @@ pub fn resolve_keys(
         }
     };
 
+    tracing::warn!(
+        target: "freemkv::disc",
+        phase = "resolve_keys_start",
+        aacs2,
+        bus_encryption,
+        disc_hash = %hash_hex,
+        mkb_present = mkb_data.is_some(),
+        "resolve_keys: starting"
+    );
+
     // Path 1: Look up VUK by disc hash in KEYDB
     if let Some(entry) = keydb.find_disc(&hash_hex) {
+        tracing::warn!(target: "freemkv::disc", phase = "resolve_keys_path1_hit_entry", "disc hash found in keydb");
         if let Some(vuk) = entry.vuk {
             return Some(build(vuk, 1));
         }
+        tracing::warn!(target: "freemkv::disc", phase = "resolve_keys_path1_no_vuk", "disc hash entry has no VUK");
+    } else {
+        tracing::warn!(target: "freemkv::disc", phase = "resolve_keys_path1_miss", "disc hash NOT in keydb");
     }
 
     // Path 2: Find entry with matching VID → derive VUK from MK + VID
+    let mut path2_mk_did_count = 0usize;
     for entry in keydb.disc_entries.values() {
         if let (Some(mk), Some(did)) = (entry.media_key, entry.disc_id) {
+            path2_mk_did_count += 1;
             if did == *volume_id {
+                tracing::warn!(target: "freemkv::disc", phase = "resolve_keys_path2_hit", "MK+VID entry matched volume_id");
                 return Some(build(derive_vuk(&mk, volume_id), 2));
             }
         }
     }
+    tracing::warn!(target: "freemkv::disc", phase = "resolve_keys_path2_miss", mk_did_entries = path2_mk_did_count, "no MK+VID entry matched volume_id");
 
     // Path 3: MKB + processing keys → media key → VUK
     if let Some(mkb) = mkb_data {
+        let mk_dv = mkb_find_mk_dv(mkb);
+        let subdiff = mkb_find_subdiff_records(mkb);
+        let cvalues = mkb_find_cvalues(mkb);
+        tracing::warn!(
+            target: "freemkv::disc",
+            phase = "resolve_keys_mkb_records",
+            mk_dv_found = mk_dv.is_some(),
+            subdiff_found = subdiff.is_some(),
+            subdiff_len = subdiff.as_ref().map(|s| s.len()).unwrap_or(0),
+            cvalues_found = cvalues.is_some(),
+            cvalues_len = cvalues.as_ref().map(|c| c.len()).unwrap_or(0),
+            "MKB record scan results"
+        );
+
         if let Some(mk) = derive_media_key_from_pk(mkb, &keydb.processing_keys) {
+            tracing::warn!(target: "freemkv::disc", phase = "resolve_keys_path3_hit", "media key derived from processing key");
             return Some(build(derive_vuk(&mk, volume_id), 3));
         }
+        tracing::warn!(target: "freemkv::disc", phase = "resolve_keys_path3_miss", pk_count = keydb.processing_keys.len(), "PK derivation failed");
 
         // Path 4: MKB + device keys → processing key → media key → VUK
         if let Some(mk) = derive_media_key_from_dk(mkb, &keydb.device_keys) {
+            tracing::warn!(target: "freemkv::disc", phase = "resolve_keys_path4_hit", "media key derived from device key");
             return Some(build(derive_vuk(&mk, volume_id), 4));
         }
+        tracing::warn!(target: "freemkv::disc", phase = "resolve_keys_path4_miss", dk_count = keydb.device_keys.len(), "DK derivation failed");
+    } else {
+        tracing::warn!(target: "freemkv::disc", phase = "resolve_keys_no_mkb", "no MKB data available; paths 3/4 skipped");
     }
 
     None

src/css/mod.rs

@@ -16,7 +16,7 @@ pub mod lfsr;
 pub(crate) mod tables;
 
 use crate::disc::Extent;
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 
 /// CSS decryption state for a DVD title.
 #[derive(Debug, Clone)]
@@ -34,7 +34,7 @@ pub struct CssState {
 /// 0x80 (start of the encrypted region). This only happens when a new PES
 /// packet begins at exactly sector offset 128. We scan up to 50000
 /// scrambled sectors sequentially across all extents.
-pub fn crack_key(reader: &mut dyn SectorReader, extents: &[Extent]) -> Option<CssState> {
+pub fn crack_key(reader: &mut dyn SectorSource, extents: &[Extent]) -> Option<CssState> {
     let mut tried = 0u32;
     let max_tries = 50_000;
 

src/disc/bluray.rs

@@ -3,13 +3,13 @@
 use super::*;
 use crate::clpi;
 use crate::mpls;
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf;
 
 impl Disc {
     /// Scan Blu-ray titles from MPLS playlists.
     pub(super) fn scan_bluray_titles(
-        reader: &mut dyn SectorReader,
+        reader: &mut dyn SectorSource,
         udf_fs: &udf::UdfFs,
     ) -> Vec<DiscTitle> {
         let mut titles = Vec::new();
@@ -31,7 +31,7 @@ impl Disc {
     }
 
     pub(super) fn parse_playlist(
-        reader: &mut dyn SectorReader,
+        reader: &mut dyn SectorSource,
         udf_fs: &udf::UdfFs,
         filename: &str,
         data: &[u8],
@@ -199,7 +199,7 @@ impl Disc {
     /// Prefers English, falls back to first available language.
     /// Returns None if META directory is empty or XML has no usable title.
     pub(super) fn read_meta_title(
-        reader: &mut dyn SectorReader,
+        reader: &mut dyn SectorSource,
         udf_fs: &udf::UdfFs,
     ) -> Option<String> {
         let meta_dir = udf_fs.find_dir("/BDMV/META")?;

src/disc/dvd.rs

@@ -2,13 +2,13 @@
 
 use super::*;
 use crate::ifo;
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf;
 
 impl Disc {
     /// Scan DVD titles from IFO files (VIDEO_TS.IFO + VTS_XX_0.IFO).
     pub(super) fn scan_dvd_titles(
-        reader: &mut dyn SectorReader,
+        reader: &mut dyn SectorSource,
         udf_fs: &udf::UdfFs,
     ) -> Vec<DiscTitle> {
         let dvd_info = match ifo::parse_vmg(reader, udf_fs) {

src/disc/encrypt.rs

@@ -2,7 +2,7 @@
 
 use super::*;
 use crate::error::{Error, Result};
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf;
 
 /// Result of SCSI AACS handshake (ECDH authentication).
@@ -22,42 +22,103 @@ impl Disc {
     ) -> Option<HandshakeResult> {
         use crate::aacs::{self, KeyDb};
 
-        let keydb_path = opts.resolve_keydb()?;
-        let keydb = KeyDb::load(&keydb_path).ok()?;
+        tracing::warn!(
+            target: "freemkv::disc",
+            phase = "handshake_entry",
+            "do_handshake entered"
+        );
+        let keydb_path = match opts.resolve_keydb() {
+            Some(p) => p,
+            None => {
+                tracing::warn!(
+                    target: "freemkv::disc",
+                    phase = "handshake_no_keydb",
+                    "no KEYDB found in search paths; handshake skipped"
+                );
+                return None;
+            }
+        };
+        let keydb = match KeyDb::load(&keydb_path) {
+            Ok(db) => db,
+            Err(e) => {
+                tracing::warn!(
+                    target: "freemkv::disc",
+                    phase = "handshake_keydb_load_failed",
+                    io_error_kind = ?e.kind(),
+                    keydb = %keydb_path.display(),
+                    "KEYDB load failed; handshake skipped"
+                );
+                return None;
+            }
+        };
+
+        let host_cert_count = keydb.host_certs.len();
+        tracing::warn!(
+            target: "freemkv::disc",
+            phase = "handshake_start",
+            host_cert_count,
+            keydb = %keydb_path.display(),
+            "handshake starting"
+        );
 
         const MAX_CERT_ATTEMPTS: usize = 16;
-        for hc in keydb.host_certs.iter().take(MAX_CERT_ATTEMPTS) {
+        let mut last_err_code: Option<u16> = None;
+        for (idx, hc) in keydb.host_certs.iter().take(MAX_CERT_ATTEMPTS).enumerate() {
             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) {
                         Ok(vid) => vid,
-                        Err(_) => return None, // handshake succeeded but can't read VID
+                        Err(e) => {
+                            tracing::warn!(
+                                target: "freemkv::disc",
+                                phase = "handshake_vid_read_failed",
+                                cert_index = idx,
+                                error_code = e.code(),
+                                "auth ok but volume ID read failed"
+                            );
+                            return None;
+                        }
                     };
                     let read_data_key = aacs::handshake::read_data_keys(session, &mut auth)
                         .ok()
                         .map(|(rdk, _)| rdk);
+                    tracing::debug!(
+                        target: "freemkv::disc",
+                        phase = "handshake_ok",
+                        cert_index = idx,
+                        has_read_data_key = read_data_key.is_some(),
+                    );
                     return Some(HandshakeResult {
                         volume_id,
                         read_data_key,
                     });
                 }
-                Err(_) => {
+                Err(e) => {
+                    last_err_code = Some(e.code());
                     continue;
                 }
             }
         }
+        tracing::warn!(
+            target: "freemkv::disc",
+            phase = "handshake_all_certs_failed",
+            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 failed — return None, not a fake success
         None
     }
 
     /// Resolve disc encryption — AACS 1.0, AACS 2.0, CSS, or none.
     ///
-    /// Reads AACS files from UDF (via SectorReader), resolves keys through
+    /// Reads AACS files from UDF (via SectorSource), resolves keys through
     /// whatever path works: KEYDB VUK lookup, media key derivation, processing
     /// keys, device keys. Uses handshake result (volume ID, bus key) if available.
     pub(super) fn resolve_encryption(
         udf_fs: &udf::UdfFs,
-        reader: &mut dyn SectorReader,
+        reader: &mut dyn SectorSource,
         keydb_path: &std::path::Path,
         handshake: Option<&HandshakeResult>,
     ) -> Result<AacsState> {
@@ -73,6 +134,19 @@ impl Disc {
             .or_else(|_| udf_fs.read_file(reader, "/AACS/DUPLICATE/Unit_Key_RO.inf"))
             .map_err(|_| Error::AacsNoKeys)?;
 
+        // Log the disc hash so we can confirm whether it's present in KEYDB
+        // when key resolution fails. The disc hash is SHA-1 of the full
+        // Unit_Key_RO.inf file bytes — same value KEYDB.cfg keys VUK entries by.
+        let dh = crate::aacs::disc_hash(&uk_ro_data);
+        let dh_hex = crate::aacs::disc_hash_hex(&dh);
+        tracing::warn!(
+            target: "freemkv::disc",
+            phase = "scan_aacs_disc_hash",
+            disc_hash = %dh_hex,
+            uk_ro_len = uk_ro_data.len(),
+            "disc hash computed (compare with keydb.cfg entries)"
+        );
+
         let cc_data = udf_fs
             .read_file(reader, "/AACS/Content000.cer")
             .or_else(|_| udf_fs.read_file(reader, "/AACS/Content001.cer"))
@@ -84,6 +158,28 @@ impl Disc {
             .ok();
         let mkb_ver = mkb_data.as_deref().and_then(aacs::mkb_version);
 
+        let mkb_first_64_hex = mkb_data
+            .as_deref()
+            .map(|m| {
+                m.iter()
+                    .take(64)
+                    .map(|b| format!("{b:02x}"))
+                    .collect::<String>()
+            })
+            .unwrap_or_default();
+        tracing::warn!(
+            target: "freemkv::disc",
+            phase = "scan_aacs_mkb_info",
+            mkb_present = mkb_data.is_some(),
+            mkb_len = mkb_data.as_deref().map(|m| m.len()).unwrap_or(0),
+            mkb_version = ?mkb_ver,
+            mkb_first_64 = %mkb_first_64_hex,
+            keydb_disc_count = keydb.disc_entries.len(),
+            keydb_dk_count = keydb.device_keys.len(),
+            keydb_pk_count = keydb.processing_keys.len(),
+            "AACS resolution inputs"
+        );
+
         // Use handshake volume ID if available, otherwise zeros
         // (KEYDB VUK lookup by disc hash works without volume ID)
         let volume_id = handshake.map(|h| h.volume_id).unwrap_or([0u8; 16]);

src/disc/sweep.rs

@@ -20,7 +20,7 @@
 //! - Mapfile is single-writer (consumer-only). No locking.
 //! - All `read_error::ReadCtx` state stays on the producer thread.
 //! - `set_speed` calls happen on the producer thread (same thread that
-//!   owns the `SectorReader`). No new SCSI concurrency.
+//!   owns the `SectorSource`). No new SCSI concurrency.
 //! - Per-iteration ordering of file-write → mapfile-record is kept
 //!   intact in the consumer (write before record), so the on-disk
 //!   invariant "mapfile only marks Finished what the file has

src/drive/mod.rs

@@ -29,7 +29,7 @@ use crate::platform::PlatformDriver;
 use crate::platform::mt1959::Mt1959;
 use crate::profile::{self, DriveProfile};
 use crate::scsi::ScsiTransport;
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use std::path::Path;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -707,7 +707,7 @@ fn open_block_device_for_sg(sg_path: &Path) -> Option<std::os::unix::io::RawFd>
     }
 }
 
-impl SectorReader for Drive {
+impl SectorSource for Drive {
     fn read_sectors(
         &mut self,
         lba: u32,

src/ifo.rs

@@ -9,7 +9,7 @@
 
 use crate::disc::{Codec, Resolution};
 use crate::error::{Error, Result};
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf::UdfFs;
 
 // ── Public types ────────────────────────────────────────────────────────────
@@ -184,7 +184,7 @@ fn bcd_byte(b: u8) -> u32 {
 ///
 /// Reads the VMG (Video Manager) to discover title sets, then reads each
 /// VTS IFO to extract PGC chains, cell addresses, and stream attributes.
-pub fn parse_vmg(reader: &mut dyn SectorReader, udf: &UdfFs) -> Result<DvdInfo> {
+pub fn parse_vmg(reader: &mut dyn SectorSource, udf: &UdfFs) -> Result<DvdInfo> {
     let vmg_data = udf.read_file(reader, "/VIDEO_TS/VIDEO_TS.IFO")?;
 
     // Validate VMG magic
@@ -263,7 +263,7 @@ pub fn parse_vmg(reader: &mut dyn SectorReader, udf: &UdfFs) -> Result<DvdInfo>
 ///
 /// `titles_info` is a list of (chapter_count, vts_title_number) from TT_SRPT.
 fn parse_vts(
-    reader: &mut dyn SectorReader,
+    reader: &mut dyn SectorSource,
     udf: &UdfFs,
     vts_number: u8,
     titles_info: &[(u16, u8)],

src/io/mod.rs

@@ -27,5 +27,7 @@ pub(crate) use writeback_file::WritebackFile;
 // caller — the targeted `#[allow]` keeps the re-export visible without
 // dragging the rest of the module under `dead_code`.
 #[allow(unused_imports)]
-pub use pipeline::WRITE_THROUGH_DEPTH;
-pub use pipeline::{DEFAULT_PIPELINE_DEPTH, Flow, Pipeline, Sink};
+pub use pipeline::{
+    DEFAULT_PIPELINE_DEPTH, Flow, Pipeline, READ_PIPELINE_DEPTH, Sink, WRITE_PIPELINE_DEPTH,
+    WRITE_THROUGH_DEPTH,
+};

src/io/pipeline.rs

@@ -27,15 +27,11 @@
 //! - Consumer panic is converted into
 //!   `Error::IoError { source: io::Error::other(...) }`.
 //!
-//! ## Dead-code suppression
+//! ## Debug logging
 //!
-//! Both `Disc::sweep` (`DEFAULT_PIPELINE_DEPTH`, `spawn_named`,
-//! `try_send`) and `Disc::patch` (`WRITE_THROUGH_DEPTH`, `Flow::Stop`)
-//! are now in-tree callers (0.18 round 2). The targeted `#[allow]`
-//! markers below cover the corners of the API not exercised by either
-//! call site (e.g. `Pipeline::spawn` without a name; `Flow::Stop`
-//! return path from sweep's sink — sweep always consumes the producer's
-//! full work-list). Drop the allows once mux migrates.
+//! Set `FREEMKV_DEBUG=1` environment variable to enable verbose debug
+//! logging throughout the pipeline (channel sends/receives, backpressure,
+//! consumer lag detection). This is critical for diagnosing stalls.
 
 use std::io;
 use std::sync::mpsc::{SyncSender, sync_channel};
@@ -43,20 +39,29 @@ use std::thread::{self, JoinHandle};
 
 use crate::error::Error;
 
+/// Check if verbose debug logging is enabled via FREEMKV_DEBUG env var.
+pub fn debug_enabled() -> bool {
+    std::env::var("FREEMKV_DEBUG")
+        .ok()
+        .map(|v| v == "1" || v == "true" || v == "yes")
+        .unwrap_or(false)
+}
+
 /// Default channel depth for callers without a specific reason to
-/// pick another value.
-///
-/// Empirically tuned for sweep and mux — both want enough slack that
-/// short consumer stalls don't immediately back up onto the producer,
-/// but not so much that a producer outpacing the consumer accumulates
-/// arbitrary buffered work. `4` matches the depth sweep has used
-/// since 0.17.11 (originally in `disc/sweep_pipeline.rs`, now in
-/// `disc/sweep.rs::SweepSink`). Patch should usually use
-/// [`WRITE_THROUGH_DEPTH`] (`1`) instead — write-through gives clean
-/// back-pressure between every read attempt and the matching write,
-/// which matters when the consumer is updating the mapfile in lockstep.
+/// pick another value. Kept conservative (4) — most callers should
+/// use READ_PIPELINE_DEPTH or WRITE_PIPELINE_DEPTH instead.
 pub const DEFAULT_PIPELINE_DEPTH: usize = 4;
 
+/// Read pipeline depth. Larger buffer compensates for drive variability
+/// and NFS sync_file_range stalls; keeps ISO reader thread fed even when
+/// consumer blocks on write.
+pub const READ_PIPELINE_DEPTH: usize = 32;
+
+/// Write pipeline depth. Smaller buffer reduces backpressure risk when
+/// sync_file_range blocks; prevents producer from accumulating too much
+/// work while consumer waits for NFS to drain.
+pub const WRITE_PIPELINE_DEPTH: usize = 16;
+
 /// Channel depth for write-through pipelines. Each `send` fully
 /// drains before the next can enqueue. Use this when the producer
 /// must observe consumer side-effects (e.g. mapfile state) before
@@ -149,6 +154,12 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
                 let mut stopped = false;
 
                 while let Ok(item) = rx.recv() {
+                    if debug_enabled() {
+                        tracing::debug!("Pipeline receive: item={}", std::any::type_name::<I>());
+                    }
+
+                    let apply_start = std::time::Instant::now();
+
                     if first_err.is_some() || stopped {
                         // Drain remaining items so the producer never
                         // blocks on a dead receiver. `apply` is not
@@ -159,11 +170,32 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
                         Ok(Flow::Continue) => {}
                         Ok(Flow::Stop) => {
                             stopped = true;
+                            if debug_enabled() {
+                                tracing::debug!("Pipeline: consumer returned Flow::Stop");
+                            }
                         }
                         Err(e) => {
+                            if debug_enabled() {
+                                tracing::debug!("Pipeline: apply error, stopping, err={:?}", e);
+                            }
                             first_err = Some(e);
                         }
                     }
+
+                    let apply_elapsed = apply_start.elapsed();
+                    if debug_enabled() && apply_elapsed > std::time::Duration::from_millis(100) {
+                        tracing::debug!(
+                            "Pipeline apply: took {:.2}s, item={}",
+                            apply_elapsed.as_secs_f64(),
+                            std::any::type_name::<I>()
+                        );
+                    } else if debug_enabled() {
+                        tracing::debug!(
+                            "Pipeline apply: OK in {:.3}ms, item={}",
+                            apply_elapsed.as_micros(),
+                            std::any::type_name::<I>()
+                        );
+                    }
                 }
 
                 match first_err {
@@ -188,7 +220,35 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
     /// independent signal (e.g. `Halt`) — `send` alone is not the
     /// notification edge.
     pub fn send(&self, item: I) -> Result<(), I> {
-        self.tx.send(item).map_err(|e| e.0)
+        let start = std::time::Instant::now();
+        match self.tx.send(item) {
+            Ok(()) => {
+                let elapsed = start.elapsed();
+                if debug_enabled() && elapsed > std::time::Duration::from_millis(10) {
+                    tracing::debug!(
+                        "Pipeline send: blocked {:.2}s, item={}",
+                        elapsed.as_secs_f64(),
+                        std::any::type_name::<I>()
+                    );
+                } else if debug_enabled() {
+                    tracing::debug!("Pipeline send: OK in {:.3}ms", elapsed.as_micros());
+                }
+                Ok(())
+            }
+            Err(e) => {
+                let elapsed = start.elapsed();
+                if debug_enabled() && elapsed > std::time::Duration::from_millis(10) {
+                    tracing::debug!(
+                        "Pipeline send: blocked {:.2}s before channel closed, item={}",
+                        elapsed.as_secs_f64(),
+                        std::any::type_name::<I>()
+                    );
+                } else if debug_enabled() {
+                    tracing::debug!("Pipeline send: failed after {:.3}ms", elapsed.as_micros());
+                }
+                Err(e.0)
+            }
+        }
     }
 
     /// Non-blocking variant of [`Pipeline::send`]. If the channel is

src/io/writeback_file.rs

@@ -27,7 +27,7 @@ use std::path::Path;
 
 use super::writeback::WritebackPipeline;
 
-const CHUNK_BYTES: u64 = 32 * 1024 * 1024;
+const WRITEBACK_CHUNK_BYTES: u64 = 32 * 1024 * 1024;
 
 pub(crate) struct WritebackFile {
     file: File,
@@ -42,7 +42,7 @@ impl WritebackFile {
     /// or appended files).
     pub(crate) fn new(mut file: File) -> io::Result<Self> {
         let pos = file.stream_position()?;
-        let pipeline = WritebackPipeline::new(&file, pos, CHUNK_BYTES);
+        let pipeline = WritebackPipeline::new(&file, pos, WRITEBACK_CHUNK_BYTES);
         Ok(Self {
             file,
             pipeline,

src/labels/bdmt.rs

@@ -26,7 +26,7 @@
 // The module wiring (registry hook + public re-export) is added
 // separately. Until then the parse/detect entry points have no
 use super::xml;
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf::UdfFs;
 use std::collections::BTreeMap;
 
@@ -60,7 +60,7 @@ pub fn detect(udf: &UdfFs) -> bool {
 /// Read every `bdmt_<lang>.xml` under `/BDMV/META/DL/` and return the
 /// aggregated [`DiscMetadata`]. Returns `None` if no titles could be
 /// extracted from any file.
-pub fn parse(reader: &mut dyn SectorReader, udf: &UdfFs) -> Option<DiscMetadata> {
+pub fn parse(reader: &mut dyn SectorSource, udf: &UdfFs) -> Option<DiscMetadata> {
     let dir = udf.find_dir("/BDMV/META/DL")?;
     let mut out = DiscMetadata::default();
 

src/labels/clpi_audit.rs

@@ -29,7 +29,7 @@
 //! report. Surfaced via the labels-analyze tool — not part of the
 //! `analyze()` pipeline (no impact on the label output).
 
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf::UdfFs;
 use std::collections::BTreeMap;
 
@@ -116,7 +116,7 @@ impl ClpiVsMplsAudit {
 /// dedup-by-PID table of (CLPI fields, MPLS fields), return the
 /// merged view. Missing files (read errors, parse failures) are
 /// silently skipped — this is diagnostic, not correctness-critical.
-pub fn audit(reader: &mut dyn SectorReader, udf: &UdfFs) -> ClpiVsMplsAudit {
+pub fn audit(reader: &mut dyn SectorSource, udf: &UdfFs) -> ClpiVsMplsAudit {
     // Aggregate by PID across all CLPI files. If a PID appears in
     // multiple clips (typical — main movie clip + trailers reference
     // the same audio stream PIDs), first encountered wins (they should

src/labels/criterion.rs

@@ -4,7 +4,7 @@
 //! stream number mapping via playbackconfig.
 
 use super::{LabelPurpose, LabelQualifier, ParseResult, StreamLabel, StreamLabelType, xml};
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf::UdfFs;
 use std::collections::HashMap;
 
@@ -12,7 +12,7 @@ pub fn detect(udf: &UdfFs) -> bool {
     super::jar_file_exists(udf, "streamproperties.xml")
 }
 
-pub fn parse(reader: &mut dyn SectorReader, udf: &UdfFs) -> Option<ParseResult> {
+pub fn parse(reader: &mut dyn SectorSource, udf: &UdfFs) -> Option<ParseResult> {
     let sp_data = super::read_jar_file(reader, udf, "streamproperties.xml")?;
     let sp_text = std::str::from_utf8(&sp_data).ok()?;
 

src/labels/ctrm.rs

@@ -5,7 +5,7 @@
 //! menu_base.prop provides stream number → button name mapping.
 
 use super::{LabelPurpose, LabelQualifier, ParseResult, StreamLabel, StreamLabelType, vocab};
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf::UdfFs;
 use std::collections::HashMap;
 
@@ -14,7 +14,7 @@ pub fn detect(udf: &UdfFs) -> bool {
         || super::jar_file_exists(udf, "language_streams.txt")
 }
 
-pub fn parse(reader: &mut dyn SectorReader, udf: &UdfFs) -> Option<ParseResult> {
+pub fn parse(reader: &mut dyn SectorSource, udf: &UdfFs) -> Option<ParseResult> {
     // Try language_streams.txt first (richer structured data)
     let ls_labels = parse_language_streams(reader, udf);
 
@@ -55,7 +55,7 @@ fn merge(ls: Vec<StreamLabel>, mb: Vec<StreamLabel>) -> Vec<StreamLabel> {
 
 // ── language_streams.txt parser ────────────────────────────────────────────
 
-fn parse_language_streams(reader: &mut dyn SectorReader, udf: &UdfFs) -> Option<Vec<StreamLabel>> {
+fn parse_language_streams(reader: &mut dyn SectorSource, udf: &UdfFs) -> Option<Vec<StreamLabel>> {
     let data = super::read_jar_file(reader, udf, "language_streams.txt")?;
     let text = std::str::from_utf8(&data).ok()?;
 
@@ -184,7 +184,7 @@ mod tests {
 
     /// Build a minimal menu_base.prop text and run `parse_menu_base`'s
     /// inner logic via a temporary closure. This isolates the prop
-    /// parsing without needing a SectorReader.
+    /// parsing without needing a SectorSource.
     fn parse_props(text: &str) -> Vec<StreamLabel> {
         // Mirror the inner loop of parse_menu_base exactly. Kept
         // separate so the test doesn't need disc fixtures.
@@ -338,7 +338,7 @@ mod tests {
 
 // ── menu_base.prop parser ──────────────────────────────────────────────────
 
-fn parse_menu_base(reader: &mut dyn SectorReader, udf: &UdfFs) -> Option<Vec<StreamLabel>> {
+fn parse_menu_base(reader: &mut dyn SectorSource, udf: &UdfFs) -> Option<Vec<StreamLabel>> {
     let data = super::read_jar_file(reader, udf, "menu_base.prop")?;
     let text = std::str::from_utf8(&data).ok()?;
 

src/labels/dbp.rs

@@ -35,11 +35,11 @@
 
 use super::class_reader::CpInfo;
 use super::{ParseResult, StreamLabel, StreamLabelType, jar, vocab};
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf::UdfFs;
 use std::collections::BTreeMap;
 
-/// dbp detect can't peek inside a jar without a SectorReader (the
+/// dbp detect can't peek inside a jar without a SectorSource (the
 /// trait function only takes `&UdfFs`), so we trigger on the cheap
 /// signal "any top-level .jar in /BDMV/JAR/." That fires on every
 /// BD-J disc, but parse() does the real `com/dbp/` check and
@@ -49,7 +49,7 @@ pub fn detect(udf: &UdfFs) -> bool {
     jar::has_any_top_level_jar(udf)
 }
 
-pub fn parse(reader: &mut dyn SectorReader, udf: &UdfFs) -> Option<ParseResult> {
+pub fn parse(reader: &mut dyn SectorSource, udf: &UdfFs) -> Option<ParseResult> {
     jar::for_each_jar(reader, udf, |_entry_name, archive| {
         if !jar::has_path_prefix(archive, "com/dbp/") {
             return None;

src/labels/deluxe.rs

@@ -84,18 +84,18 @@ use super::class_reader::{
     ICONST_2, ICONST_3, ICONST_4, ICONST_5, ICONST_M1, INVOKESPECIAL, LDC, LDC_W, NEW, SIPUSH,
 };
 use super::{LabelPurpose, LabelQualifier, ParseResult, StreamLabel, StreamLabelType, jar, vocab};
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf::UdfFs;
 use std::collections::{HashMap, HashSet};
 
 pub fn detect(udf: &UdfFs) -> bool {
     // Cheap pre-check at the dir level; the real signal is
     // `com/bydeluxe/` inside any top-level jar's central directory,
-    // which `parse()` confirms when given a `SectorReader`.
+    // which `parse()` confirms when given a `SectorSource`.
     jar::has_any_top_level_jar(udf)
 }
 
-pub fn parse(reader: &mut dyn SectorReader, udf: &UdfFs) -> Option<ParseResult> {
+pub fn parse(reader: &mut dyn SectorSource, udf: &UdfFs) -> Option<ParseResult> {
     jar::for_each_jar(reader, udf, |entry_name, archive| {
         if !jar::has_path_prefix(archive, "com/bydeluxe/") {
             return None;

src/labels/jar.rs

@@ -13,7 +13,7 @@
 #![allow(dead_code)]
 
 use super::class_reader::ClassFile;
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf::UdfFs;
 use std::io::Cursor;
 use zip::ZipArchive;
@@ -26,7 +26,7 @@ pub type Jar = ZipArchive<Cursor<Vec<u8>>>;
 /// True if `/BDMV/JAR/` contains at least one top-level `.jar` file
 /// (not under a subdir). Used by `detect()` in parsers whose real
 /// signal lives inside a jar — they can't open the jar without a
-/// `SectorReader`, so they use this cheap pre-check and do the real
+/// `SectorSource`, so they use this cheap pre-check and do the real
 /// `com/<vendor>/` discriminator in `parse()`.
 pub fn has_any_top_level_jar(udf: &UdfFs) -> bool {
     let Some(jar_dir) = udf.find_dir("/BDMV/JAR") else {
@@ -49,7 +49,7 @@ pub fn has_any_top_level_jar(udf: &UdfFs) -> bool {
 ///
 /// Entries that fail to read from UDF or that aren't valid zips are
 /// silently skipped — same defensive shape as the existing dbp parser.
-pub fn for_each_jar<R, F>(reader: &mut dyn SectorReader, udf: &UdfFs, mut f: F) -> Option<R>
+pub fn for_each_jar<R, F>(reader: &mut dyn SectorSource, udf: &UdfFs, mut f: F) -> Option<R>
 where
     F: FnMut(&str, &mut Jar) -> Option<R>,
 {

src/labels/mod.rs

@@ -4,7 +4,7 @@
 //! To add a new format:
 //!   1. Create `src/labels/myformat.rs`
 //!   2. Implement `pub fn detect(udf: &UdfFs) -> bool`
-//!   3. Implement `pub fn parse(reader: &mut dyn SectorReader, udf: &UdfFs) -> Option<Vec<StreamLabel>>`
+//!   3. Implement `pub fn parse(reader: &mut dyn SectorSource, udf: &UdfFs) -> Option<Vec<StreamLabel>>`
 //!   4. Add `mod myformat;` below and one line to `PARSERS` array
 
 mod bdmt;
@@ -18,13 +18,12 @@ pub(crate) mod jar;
 mod mpls_universal;
 mod paramount;
 mod pixelogic;
-mod png_filenames;
 pub(crate) mod text;
 pub mod vocab;
 pub(crate) mod xml;
 
 use crate::disc::{DiscTitle, Stream};
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf::UdfFs;
 
 // Re-export bdmt's public type so callers can construct/inspect
@@ -88,7 +87,7 @@ pub enum LabelQualifier {
 // to array order on confidence ties.
 
 type DetectFn = fn(&UdfFs) -> bool;
-type ParseFn = fn(&mut dyn SectorReader, &UdfFs) -> Option<ParseResult>;
+type ParseFn = fn(&mut dyn SectorSource, &UdfFs) -> Option<ParseResult>;
 
 /// Per-parser claim of how reliable its output is. Used by the
 /// registry to pick between parsers when more than one matches (e.g.
@@ -176,7 +175,7 @@ const PARSERS: &[(&str, DetectFn, ParseFn)] = &[
 
 /// Search disc for config files, extract labels, apply to streams.
 /// This is 100% optional — if anything fails, streams are untouched.
-pub fn apply(reader: &mut dyn SectorReader, udf: &UdfFs, titles: &mut [DiscTitle]) {
+pub fn apply(reader: &mut dyn SectorSource, udf: &UdfFs, titles: &mut [DiscTitle]) {
     let labels = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| extract(reader, udf)))
         .unwrap_or_default();
     if labels.is_empty() {
@@ -192,7 +191,7 @@ pub fn apply(reader: &mut dyn SectorReader, udf: &UdfFs, titles: &mut [DiscTitle
 /// `forced` flag.
 ///
 /// Extracted from `apply()` so the matching logic is unit-testable
-/// without needing a SectorReader / UdfFs.
+/// without needing a SectorSource / UdfFs.
 pub(crate) fn apply_labels(labels: &[StreamLabel], titles: &mut [DiscTitle]) {
     for title in titles.iter_mut() {
         let mut audio_idx: u16 = 0;
@@ -370,7 +369,7 @@ fn generate_audio_label(
     }
 }
 
-fn extract(reader: &mut dyn SectorReader, udf: &UdfFs) -> Vec<StreamLabel> {
+fn extract(reader: &mut dyn SectorSource, udf: &UdfFs) -> Vec<StreamLabel> {
     let mut best: Option<(&'static str, ParseResult)> = None;
     for (name, detect, parse) in PARSERS {
         if !detect(udf) {
@@ -483,7 +482,7 @@ fn type_tag(t: StreamLabelType) -> u8 {
 /// authoring tool left out of the published playlist.
 fn append_clpi_orphans(
     labels: &mut Vec<StreamLabel>,
-    reader: &mut dyn SectorReader,
+    reader: &mut dyn SectorSource,
     udf: &UdfFs,
 ) -> usize {
     // Index existing labels by PID — but StreamLabel doesn't carry
@@ -608,7 +607,7 @@ fn append_clpi_orphans(
 /// the return shape is richer (includes confidence, all detected
 /// parsers, and any parsers that produced empty results).
 #[doc(hidden)]
-pub fn analyze(reader: &mut dyn SectorReader, udf: &UdfFs) -> LabelAnalysis {
+pub fn analyze(reader: &mut dyn SectorSource, udf: &UdfFs) -> LabelAnalysis {
     let inventory = jar_inventory(udf);
     let mut parsers_detected: Vec<&'static str> = Vec::new();
     let mut all_results: Vec<(&'static str, ParseResult)> = Vec::new();
@@ -699,7 +698,7 @@ pub fn analyze(reader: &mut dyn SectorReader, udf: &UdfFs) -> LabelAnalysis {
 /// playlist filename. Skipped entries (read error, parse error, no
 /// marks) silently dropped — this is a diagnostic field, not a
 /// correctness-critical one.
-fn collect_chapter_summary(reader: &mut dyn SectorReader, udf: &UdfFs) -> Vec<ChapterSummary> {
+fn collect_chapter_summary(reader: &mut dyn SectorSource, udf: &UdfFs) -> Vec<ChapterSummary> {
     let Some(playlist_dir) = udf.find_dir("/BDMV/PLAYLIST") else {
         return Vec::new();
     };
@@ -842,7 +841,7 @@ pub(crate) fn find_jar_file(udf: &UdfFs, filename: &str) -> Option<String> {
 
 /// Read a file from any BDMV/JAR subdirectory by filename.
 pub(crate) fn read_jar_file(
-    reader: &mut dyn SectorReader,
+    reader: &mut dyn SectorSource,
     udf: &UdfFs,
     filename: &str,
 ) -> Option<Vec<u8>> {
@@ -1073,7 +1072,7 @@ mod gap_fill_tests {
 // ── apply() integration tests ──────────────────────────────────────────────
 //
 // End-to-end coverage for the apply_labels + fill_defaults pipeline
-// without needing a SectorReader / UdfFs. Synthetic DiscTitle +
+// without needing a SectorSource / UdfFs. Synthetic DiscTitle +
 // StreamLabel inputs, assert on the resulting Stream field values.
 
 #[cfg(test)]

src/labels/mpls_universal.rs

@@ -24,7 +24,7 @@ use super::{
     LabelPurpose, LabelQualifier, ParseResult, StreamLabel, StreamLabelType,
     vocab::{self, LangInfo},
 };
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf::UdfFs;
 
 /// True iff `/BDMV/PLAYLIST/` exists and contains at least one
@@ -45,7 +45,7 @@ pub fn detect(udf: &UdfFs) -> bool {
 /// Returns `None` if no labels could be produced (e.g. no .mpls files
 /// parsed successfully, or every parsed stream was a type we skip
 /// like IG / DV EL).
-pub fn parse(reader: &mut dyn SectorReader, udf: &UdfFs) -> Option<ParseResult> {
+pub fn parse(reader: &mut dyn SectorSource, udf: &UdfFs) -> Option<ParseResult> {
     let playlist_dir = udf.find_dir("/BDMV/PLAYLIST")?;
 
     // Collect mpls filenames first so we don't hold a borrow on udf

src/labels/paramount.rs

@@ -13,14 +13,14 @@
 //! ```
 
 use super::{LabelPurpose, LabelQualifier, ParseResult, StreamLabel, StreamLabelType, xml};
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf::UdfFs;
 
 pub fn detect(udf: &UdfFs) -> bool {
     super::jar_file_exists(udf, "playlists.xml")
 }
 
-pub fn parse(reader: &mut dyn SectorReader, udf: &UdfFs) -> Option<ParseResult> {
+pub fn parse(reader: &mut dyn SectorSource, udf: &UdfFs) -> Option<ParseResult> {
     let data = super::read_jar_file(reader, udf, "playlists.xml")?;
     let text = std::str::from_utf8(&data).ok()?;
 

src/labels/pixelogic.rs

@@ -9,7 +9,7 @@ use super::{
     Confidence, LabelPurpose, LabelQualifier, ParseResult, StreamLabel, StreamLabelType, text,
     vocab,
 };
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use crate::udf::UdfFs;
 use std::sync::atomic::{AtomicBool, Ordering};
 
@@ -24,7 +24,7 @@ pub fn detect(udf: &UdfFs) -> bool {
     super::jar_file_exists(udf, "bluray_project.bin")
 }
 
-pub fn parse(reader: &mut dyn SectorReader, udf: &UdfFs) -> Option<ParseResult> {
+pub fn parse(reader: &mut dyn SectorSource, udf: &UdfFs) -> Option<ParseResult> {
     let data = super::read_jar_file(reader, udf, "bluray_project.bin")?;
     // min_len=4 matches the prior local extract_strings impl. The token
     // grammar is `{lang3}_{codec?}_{purpose?}_{region?}_` so the

src/labels/png_filenames.rs

@@ -1,72 +0,0 @@
-//! PNG-filename language token parser — stubbed (noop) pending need.
-//!
-//! ## What this would do
-//!
-//! Some discs encode per-language menu localization as pre-rendered PNG
-//! menu buttons, one per language, with the language token embedded in
-//! the filename. Examples observed in the 2026-05-10 corpus:
-//!
-//! - **disc-01 (The Amateur)** — `<region>_<lang>_<context>_<format>.png`
-//!   Region prefix: `USA` / `UK` / `JPN` / etc.
-//!   Lang tokens (3-char, uppercase): `ENG`, `FRC`, `FRP`, `DEU`, `DUT`,
-//!   `ITA`, `JPN`, `LAS`, `CSP`, `POL`, `CZE` (11 languages)
-//!
-//! - **disc-09 (Dune orig)** — `<title>_<variant>_<lang>_Composite<N>.png`
-//!   Lang tokens (3-char, mixed-case): `Eng`, `Ger` (2 languages)
-//!
-//! ## Why stubbed
-//!
-//! MPLS already gives per-stream `language` + `coding_type` + stream-type
-//! (audio vs subtitle) on every disc. For the 2 unknown-framework discs
-//! that PNG filenames would close (disc-01, disc-09), MPLS will produce
-//! a strict superset of what filenames could give us, because MPLS knows
-//! per-stream attribution while filenames only know "the disc offers
-//! these N language buttons."
-//!
-//! The **only** thing PNG filenames give us that MPLS doesn't is **studio
-//! variant disambiguation**:
-//! - `FRC` (French Canadian) vs `FRP` (French Parisian) — MPLS just says `fra`
-//! - `LAS` (Latin American Spanish) vs `CSP` (Castilian Spanish) — MPLS just says `spa`
-//!
-//! That's niche enough that it doesn't justify implementing right now.
-//! Reactivate this parser only when:
-//! 1. We hit a disc where MPLS is malformed/empty AND PNG filenames are
-//!    the only language hint, OR
-//! 2. A downstream consumer needs the studio variant suffix for output
-//!    naming (e.g. `Title (French Canadian).mkv` vs `Title (French).mkv`).
-//!
-//! ## When reactivating
-//!
-//! Implement `parse` to:
-//! 1. Iterate top-level PNG paths in `/BDMV/JAR/` (and `<id>/` subdirs).
-//! 2. Tokenize each filename on `_` / `-` / `.`
-//! 3. Match each token against an alias table:
-//!    - ISO 639-1 / 639-2 standard codes
-//!    - Studio variants: `FRC`/`FRP` → `fra-CA`/`fra-FR`,
-//!      `LAS`/`CSP` → `spa-419`/`spa-ES`,
-//!      mixed-case shortforms `Eng`/`Ger`/`Fra`/`Spa`/`Jpn` → ISO 639-2
-//!    - Country prefix filter: drop `USA`/`UK`/`JPN`/`AUS`/`GER`/`FR` when
-//!      they appear in position 0 (those are region markers, not langs).
-//! 4. Deduplicate. Confidence stays `Low` because we still don't know
-//!    per-stream codec or audio/subtitle attribution.
-//!
-//! Wire as ENRICHMENT after MPLS in `mod.rs::analyze`, not as a primary
-//! parser: PNG filenames upgrade `lang=fra` to `lang=fra-CA` when both
-//! sources agree on the disc; they should never overwrite MPLS data.
-
-use super::ParseResult;
-use crate::sector::SectorReader;
-use crate::udf::UdfFs;
-
-/// Stub: returns false so the dispatcher never calls `parse`. Reactivate
-/// by checking for the patterns described in the module docs.
-#[allow(dead_code)] // module-level noop, not wired into PARSERS until needed
-pub fn detect(_udf: &UdfFs) -> bool {
-    false
-}
-
-/// Stub: returns None. See module docs for the implementation sketch.
-#[allow(dead_code)] // module-level noop, not wired into PARSERS until needed
-pub fn parse(_reader: &mut dyn SectorReader, _udf: &UdfFs) -> Option<ParseResult> {
-    None
-}

src/lib.rs

@@ -82,7 +82,7 @@ pub mod event;
 pub mod halt;
 pub(crate) mod identity;
 pub(crate) mod ifo;
-pub(crate) mod io;
+pub mod io;
 pub mod keydb;
 pub mod labels;
 pub(crate) mod mpls;
@@ -128,16 +128,19 @@ pub use halt::Halt;
 
 // Generic bounded producer/consumer primitive used by sweep, patch, and
 // mux to overlap reads with writes via a dedicated consumer thread.
-// `Pipeline::spawn(depth, sink)` spawns the consumer; `pipe.send(item)`
+// `Pipeline::spawn(name, depth, sink)` spawns a named consumer; `pipe.send(item)`
 // pushes one item with back-pressure; `pipe.finish()` joins the
 // consumer and surfaces its `close()` output. Callers implement `Sink`
 // to define per-item behaviour and end-of-stream finalisation.
 //
-// `DEFAULT_PIPELINE_DEPTH` (=4) is the depth sweep + mux use; patch
-// uses `WRITE_THROUGH_DEPTH` (=1) so each read fully drains before the
-// next can enqueue. Returning `Flow::Stop` from `apply` ends the
-// consumer cleanly (still calls `close()`).
-pub use io::pipeline::{DEFAULT_PIPELINE_DEPTH, Flow, Pipeline, Sink, WRITE_THROUGH_DEPTH};
+// `DEFAULT_PIPELINE_DEPTH` (=4) is for callers without specific needs;
+// most should use READ_PIPELINE_DEPTH or WRITE_PIPELINE_DEPTH instead.
+// Patch uses `WRITE_THROUGH_DEPTH` (=1). Returning `Flow::Stop` from
+// `apply` ends the consumer cleanly (still calls `close()`).
+pub use io::pipeline::{
+    DEFAULT_PIPELINE_DEPTH, Flow, Pipeline, READ_PIPELINE_DEPTH, Sink, WRITE_PIPELINE_DEPTH,
+    WRITE_THROUGH_DEPTH,
+};
 
 // ─── Drive events (low-level callbacks) ─────────────────────────────────────
 pub use event::{Event, EventKind};
@@ -162,8 +165,8 @@ pub use decrypt::{DecryptKeys, decrypt_sectors};
 // structured representation; never compare against display strings.
 // Note: `disc::Stream` here is the codec enum (audio / video / sub kind)
 // — not the `pes::Stream` trait re-exported below as `PesStream`. Two
-// different concepts, the same short name; both stay because both are
-// load-bearing in their respective domains.
+// different concepts, the same short name; the trait gets the `Pes`
+// prefix at the crate root to keep both addressable.
 pub use disc::{
     AacsState, AudioChannels, AudioStream, Clip, Codec, ColorSpace, ContentFormat, DamageSeverity,
     Disc, DiscFormat, DiscId, DiscTitle, Extent, FrameRate, HdrFormat, KeySource, LabelPurpose,
@@ -176,7 +179,7 @@ pub use disc::{
 // All stream types implement `pes::Stream` — read PES frames from a source,
 // write PES frames to a sink. Pick the right type at construction:
 //
-// - `DiscStream` — physical drive or ISO (any `SectorReader`). Always read.
+// - `DiscStream` — physical drive or ISO (any `SectorSource`). Read-only.
 // - `MkvStream`  — Matroska container. Read on `open()`, write on `create()`.
 // - `M2tsStream` — Blu-ray Transport Stream. Read on `open()`, write on `create()`.
 // - `NetworkStream` — TCP. Read on `listen()`, write on `connect()`.
@@ -186,14 +189,11 @@ pub use disc::{
 // Most consumers use the URL resolvers (`input()` / `output()`) which pick
 // the right type from a scheme:// URL. Direct construction is for callers
 // that need to wire custom readers (e.g. autorip's drive-session reuse).
-// 0.18 trait split: `FrameSource` (read-only) and `FrameSink` (write-only)
-// supersede the unified `pes::Stream`. The old `Stream` re-export below
-// stays available for the deprecation window — re-exported as
-// `PesStream` to disambiguate from `disc::Stream` (the codec-kind enum
-// re-exported above), which would otherwise collide at the crate root.
-#[allow(deprecated)]
+// The trait is re-exported as `PesStream` here to disambiguate from
+// `disc::Stream` (the codec-kind enum re-exported above), which would
+// otherwise collide at the crate root.
+pub use pes::PesFrame;
 pub use pes::Stream as PesStream;
-pub use pes::{FrameSink, FrameSource, PesFrame};
 
 pub use mux::DiscStream;
 pub use mux::M2tsStream;
@@ -206,18 +206,14 @@ pub use mux::{InputOptions, StreamUrl, input, output, parse_url};
 // ─── Lower-level surfaces ───────────────────────────────────────────────────
 //
 // `ScsiTransport` is the platform-abstraction trait Drive uses; expose for
-// out-of-tree platform backends. `SectorSource` / `SectorSink` are the 0.18
+// out-of-tree platform backends. `SectorSource` / `SectorSink` are the
 // direction-typed read/write traits; `FileSectorSource` and `FileSectorSink`
 // 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. The legacy `SectorReader` /
-// `FileSectorReader` names stay re-exported through the 0.18 migration
-// window so existing call sites compile unchanged; a blanket impl makes
-// every `SectorReader` automatically usable as a `SectorSource`.
+// `SectorSource` to get plaintext sectors out.
 pub use scsi::{DriveInfo, ScsiSense, ScsiTransport, drive_has_disc, list_drives};
 pub use sector::{
-    DecryptingSectorSource, FileSectorReader, FileSectorSink, FileSectorSource, SectorReader,
-    SectorSink, SectorSource,
+    DecryptingSectorSource, FileSectorSink, FileSectorSource, SectorSink, SectorSource,
 };
 pub use speed::DriveSpeed;
 pub use udf::{UdfFs, read_filesystem};

src/mux/disc.rs

@@ -1,6 +1,6 @@
 //! DiscStream — read any disc (physical drive or ISO file) → PES frames.
 //!
-//! One stream type for all disc sources. The source is a SectorReader —
+//! One stream type for all disc sources. The source is a SectorSource —
 //! Drive (hardware) or IsoSectorReader (file). DiscStream doesn't care.
 //!
 //! Read-only. For disc→ISO (raw sector copy), use `Disc::copy()`.
@@ -9,7 +9,7 @@ use crate::disc::{Disc, DiscTitle, Extent};
 use crate::drive::extract_scsi_context;
 use crate::event::{BatchSizeReason, Event, EventKind};
 use crate::halt::Halt;
-use crate::sector::{DecryptingSectorSource, SectorReader, SectorSource};
+use crate::sector::{DecryptingSectorSource, SectorSource};
 use std::io;
 use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
@@ -97,7 +97,7 @@ impl AdaptiveBatch {
 
 /// Disc stream. Reads sectors from any source → PES frames.
 ///
-/// Sources: physical drive, ISO file, or any SectorReader.
+/// Sources: physical drive, ISO file, or any SectorSource.
 /// Decrypt, demux, and codec parsing happen internally.
 pub struct DiscStream {
     /// Underlying sector source wrapped in the 0.18
@@ -105,7 +105,7 @@ pub struct DiscStream {
     /// call yields plaintext, so `fill_extents` no longer needs an
     /// inline `decrypt::decrypt_sectors` step. `DecryptKeys::None`
     /// (raw / unencrypted disc) makes the decorator a pass-through.
-    reader: DecryptingSectorSource<Box<dyn SectorReader>>,
+    reader: DecryptingSectorSource<Box<dyn SectorSource>>,
     title: DiscTitle,
     disc: Option<Disc>,
     /// Mirror of the keys handed in at construction. The decorator
@@ -160,11 +160,11 @@ pub struct DiscStream {
 impl DiscStream {
     /// Create a disc stream from any sector reader.
     ///
-    /// Works with physical drives and ISO files — both implement SectorReader.
+    /// Works with physical drives and ISO files — both implement SectorSource.
     /// The caller opens the source, scans for titles/keys, and passes them in.
     /// The stream handles demuxing, decryption, and codec parsing internally.
     pub fn new(
-        reader: Box<dyn SectorReader>,
+        reader: Box<dyn SectorSource>,
         title: DiscTitle,
         decrypt_keys: crate::decrypt::DecryptKeys,
         batch_sectors: u16,
@@ -173,6 +173,13 @@ impl DiscStream {
         let extents = title.extents.clone();
         let bytes_total_extents: u64 = extents.iter().map(|e| e.sector_count as u64 * 2048).sum();
 
+        // Debug log reader type at construction — critical for diagnosing mux reading from drive instead of ISO
+        tracing::debug!(
+            target: "mux",
+            "DiscStream constructed with reader type: {}",
+            std::any::type_name::<dyn SectorSource>()
+        );
+
         let mut pids = Vec::new();
         let mut parsers = Vec::new();
         let mut pid_to_track = Vec::new();
@@ -315,10 +322,19 @@ impl DiscStream {
         // this loop can spend minutes shrinking and skipping sectors; without
         // the check, Stop wouldn't take effect until the outer PES read() loop
         // finally emits a frame, which may never happen.
+
+        let start_time = std::time::Instant::now();
+
         loop {
             if self.is_halted() {
                 return Err(crate::error::Error::Halted.into());
             }
+
+            // Debug: log slow reads during mux — helps diagnose stalls
+            if cfg!(debug_assertions) && start_time.elapsed().as_secs() > 5 {
+                tracing::debug!(target: "mux", "fill_extents waiting at LBA {} ({}s elapsed, sectors={})", lba, start_time.elapsed().as_secs(), remaining);
+            }
+
             let mut sectors = remaining.min(self.adaptive.current() as u32) as u16;
             // Align to 3-sector AACS units when possible. Partial units at
             // extent boundaries are safely handled by decrypt_sectors().
@@ -387,7 +403,6 @@ impl DiscStream {
     }
 }
 
-#[allow(deprecated)] // 0.18 trait split: migrate to FrameSource/FrameSink in follow-up commit.
 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() {
@@ -568,31 +583,26 @@ impl crate::pes::Stream for DiscStream {
 
 #[cfg(test)]
 mod tests {
-    //! `DiscStream` is the only meaningful `FrameSource` impl in tree (every
-    //! other concrete `pes::Stream` impl in `mux/*` is a sink). The 0.18
-    //! round-1 blanket `impl<T: pes::Stream + Send> pes::FrameSource for T`
-    //! covers `DiscStream` for free as long as it is `Send`. These tests
-    //! lock that down: a static `Send` assertion plus a `Box<dyn FrameSource>`
-    //! round trip exercising every `FrameSource` method through the trait
-    //! object, so future Send-breaking edits to `DiscStream`'s interior
-    //! types fail at compile time and the trait-bridge dispatch is verified
-    //! at runtime.
-    #![allow(deprecated)] // exercising the 0.18 deprecation-window blanket bridge.
+    //! `DiscStream` is the only read-only `Stream` impl in tree (every
+    //! other concrete impl in `mux/*` is bidirectional or write-only).
+    //! These tests lock down a static `Send` assertion plus a
+    //! `Box<dyn Stream>` round trip exercising every method through the
+    //! trait object, so future Send-breaking edits to `DiscStream`'s
+    //! interior types fail at compile time.
     use super::*;
     use crate::disc::{ContentFormat, DiscTitle};
-    use crate::pes::FrameSource;
+    use crate::pes::Stream;
 
-    /// Static-assert `DiscStream: Send`. The blanket
-    /// `impl<T: pes::Stream + Send> pes::FrameSource for T` only fires for
-    /// `Send` types — if a future field on `DiscStream` is non-`Send` (e.g.
-    /// a `Box<dyn Read>` instead of `Box<dyn SectorReader>`), this fails
+    /// Static-assert `DiscStream: Send`. The `Stream` trait has `Send` as a
+    /// supertrait — if a future field on `DiscStream` is non-`Send` (e.g.
+    /// a `Box<dyn Read>` instead of `Box<dyn SectorSource>`), this fails
     /// at compile time, before the runtime trait-object test below.
     fn _assert_disc_stream_is_send() {
         fn requires_send<T: Send>() {}
         requires_send::<DiscStream>();
     }
 
-    /// Trivial `SectorReader` that yields zeroed sectors. Empty title means
+    /// Trivial `SectorSource` that yields zeroed sectors. Empty title means
     /// the demuxer produces no PES frames, so `read()` walks the extents to
     /// EOF and returns `Ok(None)`. That's enough to exercise the trait-object
     /// dispatch — the goal here is the bridge, not the demuxer.
@@ -600,7 +610,7 @@ mod tests {
         capacity: u32,
     }
 
-    impl crate::sector::SectorReader for ZeroReader {
+    impl crate::sector::SectorSource for ZeroReader {
         fn read_sectors(
             &mut self,
             _lba: u32,
@@ -613,7 +623,7 @@ mod tests {
             Ok(bytes)
         }
 
-        fn capacity(&self) -> u32 {
+        fn capacity_sectors(&self) -> u32 {
             self.capacity
         }
     }
@@ -628,13 +638,11 @@ mod tests {
         }
     }
 
-    /// Smallest credible witness that `DiscStream` flows through the
-    /// `FrameSource` blanket impl: build a `Box<dyn FrameSource>`, drive
-    /// `read()` to EOF, exercise `info()` / `headers_ready()` /
-    /// `codec_private()` through the trait object. The trait-bridge
-    /// correctness is what's being verified — not demuxer behaviour.
+    /// Smallest credible witness that `DiscStream` flows through `dyn Stream`:
+    /// build a `Box<dyn Stream>`, drive `read()` to EOF, exercise `info()` /
+    /// `headers_ready()` / `codec_private()` through the trait object.
     #[test]
-    fn frame_source_via_dyn_object() {
+    fn stream_via_dyn_object() {
         let reader = ZeroReader { capacity: 8 };
         let title = synthetic_title(8);
         let stream = DiscStream::new(
@@ -645,7 +653,7 @@ mod tests {
             ContentFormat::BdTs,
         );
 
-        let mut src: Box<dyn FrameSource> = Box::new(stream);
+        let mut src: Box<dyn Stream> = Box::new(stream);
 
         // Empty-title fixture has no streams configured, so headers are
         // trivially ready and codec_private() yields nothing on track 0.

src/mux/iso.rs

@@ -1,10 +1,10 @@
-//! ISO sector reader — file-backed SectorReader for Blu-ray ISO images.
+//! ISO sector reader — file-backed SectorSource for Blu-ray ISO images.
 //!
 //! An ISO is a flat image of 2048-byte sectors. Sector N starts at byte offset N * 2048.
 //! Used by DiscStream::open_iso() and Disc::scan_image().
 
 use crate::error::{Error, Result};
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use std::fs::File;
 use std::io::{Read, Seek, SeekFrom};
 use std::path::Path;
@@ -32,12 +32,12 @@ impl IsoSectorReader {
         Ok(Self { file, capacity })
     }
 
-    pub fn capacity(&self) -> u32 {
+    pub fn capacity_sectors(&self) -> u32 {
         self.capacity
     }
 }
 
-impl SectorReader for IsoSectorReader {
+impl SectorSource for IsoSectorReader {
     fn read_sectors(
         &mut self,
         lba: u32,
@@ -73,7 +73,7 @@ mod tests {
         std::fs::write(&dir, &data).unwrap();
 
         let mut reader = IsoSectorReader::open(dir.to_str().unwrap()).unwrap();
-        assert_eq!(reader.capacity(), 4);
+        assert_eq!(reader.capacity_sectors(), 4);
 
         let mut buf = [0u8; 2048];
         reader.read_sectors(0, 1, &mut buf, true).unwrap();
@@ -94,7 +94,7 @@ mod tests {
         std::fs::write(&dir, &data).unwrap();
 
         let reader = IsoSectorReader::open(dir.to_str().unwrap()).unwrap();
-        assert_eq!(reader.capacity(), 10);
+        assert_eq!(reader.capacity_sectors(), 10);
 
         std::fs::remove_file(&dir).ok();
     }

src/mux/m2ts.rs

@@ -173,7 +173,6 @@ impl M2tsStream {
     }
 }
 
-#[allow(deprecated)] // 0.18 trait split: migrate to FrameSource/FrameSink in follow-up commit.
 impl crate::pes::Stream for M2tsStream {
     fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
         if let Some(frame) = self.pending_frames.pop_front() {
@@ -284,27 +283,3 @@ impl crate::pes::Stream for M2tsStream {
         true
     }
 }
-
-/// FrameSink sibling to the deprecated Stream impl; both coexist during the
-/// 0.18 deprecation window. Caller may pick either at the trait-object
-/// boundary — `Box<dyn Stream>` (deprecated) or `Box<dyn FrameSink>` (new).
-/// Use `M2tsStream::create(writer, title)` to construct the write half;
-/// calling `FrameSink::write` on an `M2tsStream::open(reader)` instance
-/// returns `StreamReadOnly`.
-#[allow(deprecated)] // delegating to deprecated Stream during the 0.18 deprecation window so callers don't see the deprecation twice.
-impl crate::pes::FrameSink for M2tsStream {
-    fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
-        <Self as crate::pes::Stream>::write(self, frame)
-    }
-
-    fn finish(self: Box<Self>) -> io::Result<()> {
-        // Why: Stream::finish takes &mut self, FrameSink::finish takes Box<Self>.
-        // Re-borrow inside the box, call Stream::finish, drop the box.
-        let mut s: Self = *self;
-        <Self as crate::pes::Stream>::finish(&mut s)
-    }
-
-    fn info(&self) -> &crate::disc::DiscTitle {
-        <Self as crate::pes::Stream>::info(self)
-    }
-}

src/mux/mkvstream.rs

@@ -92,7 +92,6 @@ impl MkvStream {
     }
 }
 
-#[allow(deprecated)] // 0.18 trait split: migrate to FrameSource/FrameSink in follow-up commit.
 impl crate::pes::Stream for MkvStream {
     fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
         let rs = match self.mode {
@@ -189,35 +188,6 @@ impl crate::pes::Stream for MkvStream {
     }
 }
 
-/// FrameSink sibling to the deprecated Stream impl; both coexist during the
-/// 0.18 deprecation window. Caller may pick either at the trait-object
-/// boundary — `Box<dyn Stream>` (deprecated) or `Box<dyn FrameSink>` (new).
-/// Use `MkvStream::create(writer, title)` to construct the write half;
-/// calling `FrameSink::write` on a `MkvStream::open(reader)` instance returns
-/// `StreamReadOnly`. `finish` is where the Cues index is written, so it must
-/// be called for the resulting MKV to be seekable.
-#[allow(deprecated)] // delegating to deprecated Stream during the 0.18 deprecation window so callers don't see the deprecation twice.
-impl crate::pes::FrameSink for MkvStream {
-    fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
-        <Self as crate::pes::Stream>::write(self, frame)
-    }
-
-    fn finish(self: Box<Self>) -> io::Result<()> {
-        // Why: Stream::finish takes &mut self, FrameSink::finish takes Box<Self>.
-        // Re-borrow inside the box, call Stream::finish, drop the box.
-        // The inner `MkvMuxer` is owned via `Option`, and `Stream::finish`
-        // already takes it via `Option::take()` — moving `*self` out of the
-        // box hands it the same field by-value, so the muxer's own
-        // by-value `finish()` runs correctly.
-        let mut s: Self = *self;
-        <Self as crate::pes::Stream>::finish(&mut s)
-    }
-
-    fn info(&self) -> &crate::disc::DiscTitle {
-        <Self as crate::pes::Stream>::info(self)
-    }
-}
-
 // ── MKV header parsing (read side) ────────────────────────────
 
 /// Returns (DiscTitle, codec_privates: Vec<(track_number, codec_private_bytes)>)

src/mux/network.rs

@@ -70,7 +70,6 @@ impl NetworkStream {
     }
 }
 
-#[allow(deprecated)] // 0.18 trait split: migrate to FrameSource/FrameSink in follow-up commit.
 impl crate::pes::Stream for NetworkStream {
     fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
         match &mut self.mode {
@@ -109,30 +108,6 @@ impl crate::pes::Stream for NetworkStream {
     }
 }
 
-/// FrameSink sibling to the deprecated Stream impl; both coexist during the
-/// 0.18 deprecation window. Caller may pick either at the trait-object
-/// boundary — `Box<dyn Stream>` (deprecated) or `Box<dyn FrameSink>` (new).
-/// Use `NetworkStream::connect(addr).meta(title)` to construct the write
-/// half; calling `FrameSink::write` on `NetworkStream::listen(addr)` returns
-/// `StreamReadOnly`.
-#[allow(deprecated)] // delegating to deprecated Stream during the 0.18 deprecation window so callers don't see the deprecation twice.
-impl crate::pes::FrameSink for NetworkStream {
-    fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
-        <Self as crate::pes::Stream>::write(self, frame)
-    }
-
-    fn finish(self: Box<Self>) -> io::Result<()> {
-        // Why: Stream::finish takes &mut self, FrameSink::finish takes Box<Self>.
-        // Re-borrow inside the box, call Stream::finish, drop the box.
-        let mut s: Self = *self;
-        <Self as crate::pes::Stream>::finish(&mut s)
-    }
-
-    fn info(&self) -> &DiscTitle {
-        <Self as crate::pes::Stream>::info(self)
-    }
-}
-
 // NetworkStream is PES-only — no IOStream/Read/Write byte interface.
 
 #[cfg(test)]
@@ -182,7 +157,6 @@ mod tests {
 
     #[test]
     #[ignore] // Requires TCP; may be flaky in CI environments
-    #[allow(deprecated)] // 0.18 trait split: migrate to FrameSource/FrameSink in follow-up commit.
     fn network_pes_roundtrip() {
         use crate::pes;
 

src/mux/null.rs

@@ -16,7 +16,6 @@ impl NullStream {
     }
 }
 
-#[allow(deprecated)] // 0.18 trait split: migrate to FrameSink in follow-up commit.
 impl crate::pes::Stream for NullStream {
     fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
         Ok(None)
@@ -32,43 +31,16 @@ impl crate::pes::Stream for NullStream {
     }
 }
 
-/// FrameSink sibling to the deprecated Stream impl; both coexist during the
-/// 0.18 deprecation window. Caller may pick either at the trait-object
-/// boundary — `Box<dyn Stream>` (deprecated) or `Box<dyn FrameSink>` (new).
-/// The deprecation-window callers eventually migrate; this impl exists so
-/// new callers can target `FrameSink` without waiting for the rest of the
-/// migration to complete.
-#[allow(deprecated)] // delegating to deprecated Stream during the 0.18 deprecation window so callers don't see the deprecation twice.
-impl crate::pes::FrameSink for NullStream {
-    fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
-        <Self as crate::pes::Stream>::write(self, frame)
-    }
-
-    fn finish(self: Box<Self>) -> io::Result<()> {
-        // Why: Stream::finish takes &mut self, FrameSink::finish takes Box<Self>.
-        // Re-borrow inside the box, call Stream::finish, drop the box.
-        let mut s: Self = *self;
-        <Self as crate::pes::Stream>::finish(&mut s)
-    }
-
-    fn info(&self) -> &DiscTitle {
-        <Self as crate::pes::Stream>::info(self)
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::pes::FrameSink;
+    use crate::pes::Stream;
 
-    /// Smallest credible witness that the new FrameSink impl on a concrete
-    /// `mux/*` sink works through the trait object: build a boxed
-    /// `dyn FrameSink`, write a frame, finish it. The trait-bridge correctness
-    /// is what's being verified — not NullStream-specific behaviour.
+    /// Verify NullStream routes through the `Stream` trait object cleanly.
     #[test]
-    fn frame_sink_via_dyn_object_writes_and_finishes() {
+    fn stream_via_dyn_object_writes_and_finishes() {
         let title = DiscTitle::empty();
-        let mut sink: Box<dyn FrameSink> = Box::new(NullStream::new(&title));
+        let mut sink: Box<dyn Stream> = Box::new(NullStream::new(&title));
 
         let frame = crate::pes::PesFrame {
             track: 0,
@@ -77,9 +49,7 @@ mod tests {
             data: vec![0x01, 0x02, 0x03],
         };
         sink.write(&frame).unwrap();
-        // info() routes through the trait object.
         let _ = sink.info();
-        // finish() consumes the Box<Self> — must compile and run.
         sink.finish().unwrap();
     }
 }

src/mux/resolve.rs

@@ -167,7 +167,6 @@ pub struct InputOptions {
 }
 
 /// Open a PES input stream (produces PES frames).
-#[allow(deprecated)] // 0.18 trait split: migrate to FrameSource in follow-up commit.
 pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::Stream>> {
     let parsed = parse_url(url);
     match parsed {
@@ -187,7 +186,7 @@ pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::S
                 None => crate::disc::ScanOptions::default(),
             };
             let mut reader = super::iso::IsoSectorReader::open(&path.to_string_lossy())?;
-            let capacity = reader.capacity();
+            let capacity = reader.capacity_sectors();
             let disc = crate::disc::Disc::scan_image(&mut reader, capacity, &scan_opts)
                 .map_err(|e| -> io::Error { e.into() })?;
             if disc.titles.is_empty() {
@@ -204,7 +203,11 @@ pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::S
             let title = disc.titles[idx].clone();
             let keys = disc.decrypt_keys();
             let format = disc.content_format;
-            let mut stream = DiscStream::new(Box::new(reader), title, keys, 64, format);
+            // ISO file: use large batch size (16 MB) — sequential read from fast storage, no bad sectors.
+            // Physical drives need small batches for adaptive error handling and retry logic.
+            const ISO_MUX_BATCH_SECTORS: u16 = 8192;
+            let mut stream =
+                DiscStream::new(Box::new(reader), title, keys, ISO_MUX_BATCH_SECTORS, format);
             if opts.raw {
                 stream.set_raw();
             }
@@ -235,7 +238,6 @@ pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::S
 }
 
 /// Open a PES output stream (consumes PES frames).
-#[allow(deprecated)] // 0.18 trait split: migrate to FrameSink in follow-up commit.
 pub fn output(
     url: &str,
     title: &crate::disc::DiscTitle,

src/mux/stdio.rs

@@ -60,7 +60,6 @@ impl StdioStream {
     }
 }
 
-#[allow(deprecated)] // 0.18 trait split: migrate to FrameSource/FrameSink in follow-up commit.
 impl crate::pes::Stream for StdioStream {
     fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
         self.ensure_header_read()?;
@@ -105,26 +104,3 @@ impl crate::pes::Stream for StdioStream {
         self.header_read || self.writer.is_some()
     }
 }
-
-/// FrameSink sibling to the deprecated Stream impl; both coexist during the
-/// 0.18 deprecation window. Caller may pick either at the trait-object
-/// boundary — `Box<dyn Stream>` (deprecated) or `Box<dyn FrameSink>` (new).
-/// Use `StdioStream::output(title)` to construct the write half; calling
-/// `FrameSink::write` on a `StdioStream::input()` returns `StreamReadOnly`.
-#[allow(deprecated)] // delegating to deprecated Stream during the 0.18 deprecation window so callers don't see the deprecation twice.
-impl crate::pes::FrameSink for StdioStream {
-    fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
-        <Self as crate::pes::Stream>::write(self, frame)
-    }
-
-    fn finish(self: Box<Self>) -> io::Result<()> {
-        // Why: Stream::finish takes &mut self, FrameSink::finish takes Box<Self>.
-        // Re-borrow inside the box, call Stream::finish, drop the box.
-        let mut s: Self = *self;
-        <Self as crate::pes::Stream>::finish(&mut s)
-    }
-
-    fn info(&self) -> &DiscTitle {
-        <Self as crate::pes::Stream>::info(self)
-    }
-}

src/pes.rs

@@ -78,19 +78,20 @@ impl PesFrame {
     }
 }
 
-/// Deprecated; use [`FrameSource`] for read-only sources or [`FrameSink`]
-/// for write-only sinks. The runtime direction-error semantics
-/// (`StreamReadOnly` / `StreamWriteOnly` from a wrong-direction call) are
-/// removed in 0.18 — direction is type-checked.
-#[deprecated(
-    since = "0.18.0",
-    note = "use FrameSource (read-only) or FrameSink (write-only) instead"
-)]
-pub trait Stream {
-    /// Read the next frame, or `Ok(None)` at end of stream.
+/// A PES frame stream. One trait per format — same type opens for read
+/// (`open()` / `listen()` / `input()`) or write (`create()` / `connect()` /
+/// `output()`). Calling the wrong-direction method returns a typed
+/// `StreamReadOnly` / `StreamWriteOnly` error.
+///
+/// `Send` is required so streams can move across the producer/consumer
+/// threads in autorip's mux pipeline.
+pub trait Stream: Send {
+    /// Read the next frame, or `Ok(None)` at end of stream. Returns
+    /// `StreamWriteOnly` on a stream opened for writing.
     fn read(&mut self) -> std::io::Result<Option<PesFrame>>;
 
-    /// Write a frame to the sink.
+    /// Write a frame to the sink. Returns `StreamReadOnly` on a stream
+    /// opened for reading.
     fn write(&mut self, frame: &PesFrame) -> std::io::Result<()>;
 
     /// Finalize the stream: flush buffered frames, write any container
@@ -116,115 +117,10 @@ pub trait Stream {
     }
 }
 
-/// Read-only source of PES frames.
-///
-/// Replaces the read half of the deprecated [`Stream`] trait. Implementors
-/// produce frames via [`read`](FrameSource::read) and never accept writes —
-/// passing a `FrameSource` where a sink is expected is a compile error,
-/// not a runtime `E9001`.
-///
-/// `info()` returns the source's `DiscTitle` metadata (track list, codec
-/// info, duration). It must be stable across the lifetime of the source.
-///
-/// `codec_private(track)` exposes per-track codec initialization data
-/// (H.264 SPS/PPS, HEVC VPS/SPS/PPS, AC-3 fscod, etc.) that downstream
-/// muxers may need before any frame is written. `headers_ready()` returns
-/// false until enough input frames have been seen to populate every video
-/// track's codec-private blob — callers buffer frames they read until
-/// `headers_ready()` returns true.
-pub trait FrameSource: Send {
-    /// Read the next frame, or `Ok(None)` at end of stream.
-    fn read(&mut self) -> std::io::Result<Option<PesFrame>>;
-
-    /// Source metadata. Stable across reads — implementors must return a
-    /// consistent reference for the lifetime of the source.
-    fn info(&self) -> &crate::disc::DiscTitle;
-
-    /// Codec initialization data for a track (SPS/PPS, AC-3 fscod, etc.).
-    /// `None` for tracks that don't need codec_private (raw passthrough).
-    fn codec_private(&self, _track: usize) -> Option<Vec<u8>> {
-        None
-    }
-
-    /// True when `codec_private` is available for every video track —
-    /// callers buffer input frames until this flips, since some output
-    /// formats (MKV) can't write frames without codec init data.
-    fn headers_ready(&self) -> bool {
-        true
-    }
-}
-
-/// Write-only sink of PES frames.
-///
-/// Replaces the write half of the deprecated [`Stream`] trait. Implementors
-/// accept frames via [`write`](FrameSink::write) and finalize via
-/// [`finish`](FrameSink::finish) — passing a `FrameSink` where a source is
-/// expected is a compile error, not a runtime `E9000`.
-///
-/// `finish` takes `Box<Self>` (rather than `&mut self` like `Stream::finish`)
-/// so that finalization is a one-shot terminal operation: callers cannot use
-/// the sink after `finish` returns. This is the standard idiom for terminal
-/// methods on `dyn Trait` objects.
-pub trait FrameSink: Send {
-    /// Write a frame to the sink.
-    fn write(&mut self, frame: &PesFrame) -> std::io::Result<()>;
-
-    /// Finalize the sink: flush buffered frames, write any container index
-    /// (e.g. MKV `Cues`), close the underlying file/socket. Consumes the
-    /// sink — callers cannot use it afterwards.
-    fn finish(self: Box<Self>) -> std::io::Result<()>;
-
-    /// Sink metadata. Stable across writes — implementors must return a
-    /// consistent reference for the lifetime of the sink.
-    fn info(&self) -> &crate::disc::DiscTitle;
-}
-
-// Bridge: any **`Send`** type implementing the deprecated `Stream` trait
-// is also a `FrameSource`. This lets existing concrete `Stream` impls in
-// `mux/*` satisfy `FrameSource` bounds without per-type migration during
-// the 0.18 deprecation window.
-//
-// **Send caveat (read me before tightening `Stream` itself).** This
-// blanket carries a `T: Send` bound rather than promoting `Send` to a
-// supertrait of `Stream`, because not every concrete in-tree `Stream`
-// impl is `Send`: `MkvStream` and `M2tsStream` carry `Box<dyn Read>`
-// and `Box<dyn Write>` fields whose trait objects don't include `Send`.
-// Adding `Stream: Send` would force a wider audit (every `Box<dyn Read>`
-// becomes `Box<dyn Read + Send>`) than this commit is taking on, and
-// the type-level migration target is `FrameSource` / `FrameSink`
-// directly anyway. Consequence: coercing a non-Send `Box<dyn Stream>`
-// (the return shape of `crate::mux::input` / `output`) to
-// `Box<dyn FrameSource>` will fail with a `T: Send` trait-bound error.
-// The fix on the consumer side is to construct a Send-compliant
-// `FrameSource` / `FrameSink` directly rather than relying on this
-// bridge for non-Send streams.
-//
-// Note: `FrameSink` cannot be blanket-impl'd from `Stream` because
-// `Stream::finish` takes `&mut self` while `FrameSink::finish` takes
-// `Box<Self>`; concrete types will be migrated in a follow-up commit.
-#[allow(deprecated)]
-impl<T: Stream + Send + ?Sized> FrameSource for T {
-    fn read(&mut self) -> std::io::Result<Option<PesFrame>> {
-        <Self as Stream>::read(self)
-    }
-
-    fn info(&self) -> &crate::disc::DiscTitle {
-        <Self as Stream>::info(self)
-    }
-
-    fn codec_private(&self, track: usize) -> Option<Vec<u8>> {
-        <Self as Stream>::codec_private(self, track)
-    }
-
-    fn headers_ready(&self) -> bool {
-        <Self as Stream>::headers_ready(self)
-    }
-}
-
 /// Wraps any output stream and counts bytes written.
 ///
 /// Progress tracking is a CLI concern — streams don't know their size.
-/// Wrap the output with CountingStream, then query bytes_written().
+/// Wrap the output with `CountingStream`, then query `bytes_written()`.
 ///
 /// ```text
 /// let mut output = CountingStream::new(libfreemkv::output(dest, &title)?);
@@ -233,13 +129,11 @@ impl<T: Stream + Send + ?Sized> FrameSource for T {
 ///     let pct = output.bytes_written() as f64 / total as f64;
 /// }
 /// ```
-#[allow(deprecated)]
 pub struct CountingStream {
     inner: Box<dyn Stream>,
     written: u64,
 }
 
-#[allow(deprecated)]
 impl CountingStream {
     pub fn new(inner: Box<dyn Stream>) -> Self {
         Self { inner, written: 0 }
@@ -251,7 +145,6 @@ impl CountingStream {
     }
 }
 
-#[allow(deprecated)]
 impl Stream for CountingStream {
     fn read(&mut self) -> std::io::Result<Option<PesFrame>> {
         self.inner.read()
@@ -293,88 +186,35 @@ mod tests {
         }
     }
 
-    /// Direct `FrameSource` impl (not via the deprecated `Stream` blanket).
-    struct MockSource {
-        frames: std::vec::IntoIter<PesFrame>,
+    /// Minimal in-memory `Stream` for trait-shape tests. `read` replays
+    /// pre-seeded frames; `write` collects them.
+    struct MockStream {
+        read_queue: std::vec::IntoIter<PesFrame>,
+        written: Vec<PesFrame>,
         title: DiscTitle,
     }
 
-    impl MockSource {
-        fn new(frames: Vec<PesFrame>) -> Self {
+    impl MockStream {
+        fn new(read_frames: Vec<PesFrame>) -> Self {
             Self {
-                frames: frames.into_iter(),
+                read_queue: read_frames.into_iter(),
+                written: Vec::new(),
                 title: DiscTitle::empty(),
             }
         }
     }
 
-    impl FrameSource for MockSource {
+    impl Stream for MockStream {
         fn read(&mut self) -> std::io::Result<Option<PesFrame>> {
-            Ok(self.frames.next())
+            Ok(self.read_queue.next())
         }
 
-        fn info(&self) -> &DiscTitle {
-            &self.title
-        }
-    }
-
-    /// Direct `FrameSink` impl (not via the deprecated `Stream` blanket).
-    struct MockSink {
-        collected: Vec<PesFrame>,
-        title: DiscTitle,
-    }
-
-    impl MockSink {
-        fn new() -> Self {
-            Self {
-                collected: Vec::new(),
-                title: DiscTitle::empty(),
-            }
-        }
-    }
-
-    impl FrameSink for MockSink {
         fn write(&mut self, frame: &PesFrame) -> std::io::Result<()> {
-            self.collected.push(frame.clone());
+            self.written.push(frame.clone());
             Ok(())
         }
 
-        fn finish(self: Box<Self>) -> std::io::Result<()> {
-            // Drop self; in real sinks this is where flush/fsync/close happens.
-            Ok(())
-        }
-
-        fn info(&self) -> &DiscTitle {
-            &self.title
-        }
-    }
-
-    /// Variant of `MockSink` whose `finish` returns the collected frames so
-    /// the test can assert on them after the consuming `Box<Self>` call.
-    struct CollectingSink {
-        collected: Vec<PesFrame>,
-        title: DiscTitle,
-    }
-
-    impl CollectingSink {
-        fn new() -> Self {
-            Self {
-                collected: Vec::new(),
-                title: DiscTitle::empty(),
-            }
-        }
-    }
-
-    impl FrameSink for CollectingSink {
-        fn write(&mut self, frame: &PesFrame) -> std::io::Result<()> {
-            self.collected.push(frame.clone());
-            Ok(())
-        }
-
-        fn finish(self: Box<Self>) -> std::io::Result<()> {
-            // Real CollectingSink consumers would expose `take()` before
-            // finish; this trait method just confirms the boxed signature
-            // compiles and runs.
+        fn finish(&mut self) -> std::io::Result<()> {
             Ok(())
         }
 
@@ -384,72 +224,43 @@ mod tests {
     }
 
     #[test]
-    fn frame_source_yields_frames_then_eof() {
+    fn stream_read_yields_frames_then_eof() {
         let frames = vec![make_frame(0, 0), make_frame(1, 1_000), make_frame(0, 2_000)];
-        let mut src = MockSource::new(frames.clone());
+        let mut s = MockStream::new(frames.clone());
 
-        let f0 = src.read().unwrap().expect("first frame");
+        let f0 = s.read().unwrap().expect("first frame");
         assert_eq!(f0.track, frames[0].track);
         assert_eq!(f0.pts, frames[0].pts);
         assert!(f0.keyframe);
 
-        let f1 = src.read().unwrap().expect("second frame");
-        assert_eq!(f1.track, frames[1].track);
+        let f1 = s.read().unwrap().expect("second frame");
         assert_eq!(f1.pts, frames[1].pts);
 
-        let f2 = src.read().unwrap().expect("third frame");
-        assert_eq!(f2.track, frames[2].track);
+        let f2 = s.read().unwrap().expect("third frame");
         assert_eq!(f2.pts, frames[2].pts);
 
-        assert!(src.read().unwrap().is_none());
-        assert!(src.read().unwrap().is_none()); // idempotent at EOF
+        assert!(s.read().unwrap().is_none());
+        assert!(s.read().unwrap().is_none()); // idempotent at EOF
     }
 
     #[test]
-    fn frame_sink_collects_then_finishes() {
-        let mut sink = MockSink::new();
+    fn stream_write_collects_then_finishes() {
+        let mut s = MockStream::new(Vec::new());
         let frames = [make_frame(0, 0), make_frame(1, 100), make_frame(2, 200)];
 
         for f in &frames {
-            sink.write(f).unwrap();
+            s.write(f).unwrap();
         }
-        assert_eq!(sink.collected.len(), 3);
-        assert_eq!(sink.collected[0].pts, 0);
-        assert_eq!(sink.collected[1].pts, 100);
-        assert_eq!(sink.collected[2].pts, 200);
-
-        // Box-and-finish — the `self: Box<Self>` shape must compile and run.
-        Box::new(sink).finish().unwrap();
+        assert_eq!(s.written.len(), 3);
+        s.finish().unwrap();
     }
 
     #[test]
-    fn frame_sink_via_dyn_object() {
-        let frames = [make_frame(0, 0), make_frame(0, 33)];
-        let mut sink: Box<dyn FrameSink> = Box::new(CollectingSink::new());
-
-        for f in &frames {
-            sink.write(f).unwrap();
-        }
-        // info() routes through the trait object.
-        let _ = sink.info();
-        sink.finish().unwrap();
-    }
-
-    /// The deprecated blanket impl: any concrete `Stream` should also act as
-    /// a `FrameSource`. `NullStream` has the simplest constructor of every
-    /// concrete `Stream` impl in `mux/*`, so it's the smallest credible
-    /// witness that the bridge compiles and dispatches correctly.
-    #[test]
-    #[allow(deprecated)]
-    fn deprecated_stream_satisfies_frame_source() {
-        let title = DiscTitle::empty();
-        let mut null = crate::mux::NullStream::new(&title);
-
-        let src: &mut dyn FrameSource = &mut null;
-        // NullStream::read returns Ok(None) — it's a write-only sink.
-        assert!(src.read().unwrap().is_none());
-        // info() forwards through the blanket impl.
-        let _ = src.info();
-        assert!(src.headers_ready());
+    fn stream_via_dyn_object() {
+        let mut s: Box<dyn Stream> = Box::new(MockStream::new(vec![make_frame(0, 0)]));
+        let frame = s.read().unwrap().expect("first frame");
+        s.write(&frame).unwrap();
+        let _ = s.info();
+        s.finish().unwrap();
     }
 }

src/sector/file.rs

@@ -13,7 +13,7 @@ use std::path::Path;
 
 use crate::error::{Error, Result};
 
-use super::{SectorReader, SectorSink};
+use super::{SectorSink, SectorSource};
 
 /// SectorSource backed by a file (ISO image).
 ///
@@ -49,14 +49,14 @@ impl FileSectorSource {
     }
 }
 
-// Implement the legacy `SectorReader` trait. The blanket impl in
+// Implement the legacy `SectorSource` trait. The blanket impl in
 // `super` produces the `SectorSource` impl automatically — no need
 // to write both, and writing both would conflict. This keeps the
-// 0.17 method-resolution path intact (callers with `SectorReader`
+// 0.17 method-resolution path intact (callers with `SectorSource`
 // in scope can still write `fsr.read_sectors(..)` against a
 // `FileSectorSource`).
-impl SectorReader for FileSectorSource {
-    fn capacity(&self) -> u32 {
+impl SectorSource for FileSectorSource {
+    fn capacity_sectors(&self) -> u32 {
         self.capacity
     }
 
@@ -147,8 +147,8 @@ impl SectorSink for FileSectorSink {
 #[cfg(test)]
 mod tests {
     // Bring the 0.18 trait into scope (not super::*: the super
-    // module also re-exports the legacy `SectorReader`, and
-    // having both `SectorReader::read_sectors` and
+    // module also re-exports the legacy `SectorSource`, and
+    // having both `SectorSource::read_sectors` and
     // `SectorSource::read_sectors` visible would force every
     // call site to disambiguate). External consumers see the
     // same surface this test exercises.

src/sector/mod.rs

@@ -1,29 +1,17 @@
 //! Sector-level I/O traits.
 //!
-//! 0.18 splits the unidirectional read trait from a write trait at
-//! the sector layer, so the type system catches "wrong direction"
-//! mistakes at compile time instead of runtime. See
-//! `(internal)/memory/0_18_redesign.md`.
+//! The sector layer is direction-typed: [`SectorSource`] reads
+//! 2048-byte sectors, [`SectorSink`] writes them. Concrete impls
+//! never do both — physical drives are read-only, file-backed
+//! ISO images are opened for read OR write at construction time.
 //!
-//! - [`SectorSource`] reads 2048-byte sectors. Implemented by
-//!   `Drive` (via the legacy [`SectorReader`] alias) and
-//!   [`FileSectorSource`] (ISO-backed).
-//! - [`SectorSink`] writes 2048-byte sectors. Implemented by
-//!   [`FileSectorSink`] (ISO-backed) and, in later commits, by
-//!   sweep/patch consumer adapters.
+//! - [`SectorSource`] is implemented by `Drive` (hardware) and
+//!   [`FileSectorSource`] / `IsoSectorReader` (file-backed).
+//! - [`SectorSink`] is implemented by [`FileSectorSink`]
+//!   (ISO-backed) and sweep/patch consumer adapters.
 //! - [`DecryptingSectorSource`] is a decorator that wraps any
-//!   `SectorSource` and applies the existing AACS / CSS in-place
-//!   decrypt to plaintext-out.
-//!
-//! [`SectorReader`] is the 0.17 read trait. It stays on through
-//! the 0.18 migration window so existing call sites
-//! (`Drive`, `IsoSectorReader`, `BufferedSectorReader`,
-//! `DiscStream`, `verify`) compile unchanged. A blanket impl
-//! forwards every `SectorReader` impl to `SectorSource`, so new
-//! code should target `SectorSource` / `SectorSink` directly. The
-//! formal `#[deprecated]` attribute lands once the internal
-//! callers have migrated; see the comment on `SectorReader` for
-//! why this commit holds it back.
+//!   `SectorSource` and applies AACS / CSS in-place decrypt to
+//!   yield plaintext sectors.
 
 pub mod decrypting;
 pub mod file;
@@ -32,13 +20,14 @@ use crate::error::Result;
 
 /// Read 2048-byte sectors from a disc, image, or composed source.
 ///
-/// Direction-typed: a `SectorSource` cannot be written to. Wrap the
-/// inner source in [`DecryptingSectorSource`] to get plaintext
-/// sectors out of an encrypted disc.
+/// Wrap the inner source in [`DecryptingSectorSource`] to get
+/// plaintext sectors out of an encrypted disc.
 pub trait SectorSource: Send {
-    /// Total capacity in sectors, if known. Returns 0 when unknown
+    /// Total capacity in sectors, if known. Default `0` = unknown
     /// (e.g. live drives that haven't completed `READ CAPACITY` yet).
-    fn capacity_sectors(&self) -> u32;
+    fn capacity_sectors(&self) -> u32 {
+        0
+    }
 
     /// Read `count` sectors starting at `lba` into `buf`.
     /// `buf` must be at least `count * 2048` bytes.
@@ -60,10 +49,52 @@ pub trait SectorSource: Send {
     fn set_speed(&mut self, _kbs: u16) {}
 }
 
+// Forwarding impls so `Box<dyn SectorSource>` and `&mut dyn SectorSource`
+// satisfy the `SectorSource` trait bound when wrapped by generic
+// decorators like `DecryptingSectorSource<S: SectorSource>`.
+impl SectorSource for Box<dyn SectorSource> {
+    fn capacity_sectors(&self) -> u32 {
+        (**self).capacity_sectors()
+    }
+
+    fn read_sectors(
+        &mut self,
+        lba: u32,
+        count: u16,
+        buf: &mut [u8],
+        recovery: bool,
+    ) -> Result<usize> {
+        (**self).read_sectors(lba, count, buf, recovery)
+    }
+
+    fn set_speed(&mut self, kbs: u16) {
+        (**self).set_speed(kbs)
+    }
+}
+
+impl SectorSource for &mut (dyn SectorSource + '_) {
+    fn capacity_sectors(&self) -> u32 {
+        (**self).capacity_sectors()
+    }
+
+    fn read_sectors(
+        &mut self,
+        lba: u32,
+        count: u16,
+        buf: &mut [u8],
+        recovery: bool,
+    ) -> Result<usize> {
+        (**self).read_sectors(lba, count, buf, recovery)
+    }
+
+    fn set_speed(&mut self, kbs: u16) {
+        (**self).set_speed(kbs)
+    }
+}
+
 /// Write 2048-byte sectors to a disc image or composed sink.
 ///
-/// Direction-typed: a `SectorSink` cannot be read from. The
-/// terminal [`finish`] takes `Box<Self>` so it can run on `dyn
+/// The terminal [`finish`] takes `Box<Self>` so it can run on `dyn
 /// SectorSink` and consume the sink (`fsync` + close).
 ///
 /// [`finish`]: SectorSink::finish
@@ -78,125 +109,5 @@ pub trait SectorSink: Send {
     fn finish(self: Box<Self>) -> Result<()>;
 }
 
-/// 0.17 read trait. Slated for removal once internal call sites
-/// migrate to [`SectorSource`] in follow-up commits; until then
-/// it remains the trait that `Drive`, `IsoSectorReader`,
-/// `BufferedSectorReader`, and existing `&mut dyn SectorReader`
-/// signatures use unchanged.
-///
-/// New code should implement [`SectorSource`] directly. The
-/// blanket impl below makes any `SectorReader` automatically
-/// usable wherever a `SectorSource` is expected, so a one-way
-/// migration off `SectorReader` is possible per-callsite without
-/// touching the impls.
-//
-// NOTE: not marked `#[deprecated]` in this commit — `cargo clippy
-// -- -D warnings` (the CI gauntlet) treats deprecation as an
-// error, and the existing `Drive` / `udf::BufferedSectorReader` /
-// `mux::DiscStream` / `verify` call sites all go through this
-// trait. The deprecation attribute lands together with the
-// migration commits that move those call sites to
-// `SectorSource`. The behavioural contract — "this trait is
-// going away in 0.18" — is documented above and tracked in
-// `(internal)/memory/0_18_redesign.md`.
-pub trait SectorReader: Send {
-    /// Read `count` sectors starting at `lba` into `buf`.
-    /// See [`SectorSource::read_sectors`] for semantics.
-    fn read_sectors(
-        &mut self,
-        lba: u32,
-        count: u16,
-        buf: &mut [u8],
-        recovery: bool,
-    ) -> Result<usize>;
-
-    /// Total capacity in sectors, if known.
-    fn capacity(&self) -> u32 {
-        0
-    }
-
-    fn set_speed(&mut self, _kbs: u16) {}
-}
-
-// Blanket impl: anything implementing the legacy `SectorReader`
-// trait automatically satisfies `SectorSource`. This is what keeps
-// existing impls (`Drive`, `IsoSectorReader`, `BufferedSectorReader`,
-// etc.) compiling without source changes during the migration. The
-// reverse direction (impl SectorReader for SectorSource) is
-// intentionally NOT provided — new code targets the new trait.
-impl<T: SectorReader + ?Sized> SectorSource for T {
-    fn capacity_sectors(&self) -> u32 {
-        <T as SectorReader>::capacity(self)
-    }
-
-    fn read_sectors(
-        &mut self,
-        lba: u32,
-        count: u16,
-        buf: &mut [u8],
-        recovery: bool,
-    ) -> Result<usize> {
-        <T as SectorReader>::read_sectors(self, lba, count, buf, recovery)
-    }
-
-    fn set_speed(&mut self, kbs: u16) {
-        <T as SectorReader>::set_speed(self, kbs)
-    }
-}
-
-// Forwarding impls so callers can wrap `&mut dyn SectorReader` /
-// `Box<dyn SectorReader>` in [`DecryptingSectorSource`] without
-// having to unbox or re-borrow inside the lib's hot paths. The
-// generic `&mut T` / `Box<T>` blankets would conflict with the
-// `SectorReader → SectorSource` blanket above (a downstream crate
-// could `impl SectorReader for &mut U`); the specific
-// `dyn SectorReader` instantiations are unambiguous because
-// `SectorReader` is the very trait whose `dyn` we're targeting.
-impl SectorSource for &mut (dyn SectorReader + '_) {
-    fn capacity_sectors(&self) -> u32 {
-        <dyn SectorReader as SectorReader>::capacity(*self)
-    }
-
-    fn read_sectors(
-        &mut self,
-        lba: u32,
-        count: u16,
-        buf: &mut [u8],
-        recovery: bool,
-    ) -> Result<usize> {
-        <dyn SectorReader as SectorReader>::read_sectors(*self, lba, count, buf, recovery)
-    }
-
-    fn set_speed(&mut self, kbs: u16) {
-        <dyn SectorReader as SectorReader>::set_speed(*self, kbs)
-    }
-}
-
-impl SectorSource for Box<dyn SectorReader> {
-    fn capacity_sectors(&self) -> u32 {
-        <dyn SectorReader as SectorReader>::capacity(&**self)
-    }
-
-    fn read_sectors(
-        &mut self,
-        lba: u32,
-        count: u16,
-        buf: &mut [u8],
-        recovery: bool,
-    ) -> Result<usize> {
-        <dyn SectorReader as SectorReader>::read_sectors(&mut **self, lba, count, buf, recovery)
-    }
-
-    fn set_speed(&mut self, kbs: u16) {
-        <dyn SectorReader as SectorReader>::set_speed(&mut **self, kbs)
-    }
-}
-
 pub use decrypting::DecryptingSectorSource;
 pub use file::{FileSectorSink, FileSectorSource};
-
-// Backwards-compat alias for the public API. `FileSectorReader` is
-// the 0.17 name; new code uses `FileSectorSource`. Both point at
-// the same type. The `#[deprecated]` attribute lands together with
-// the migration commits that retire the alias from internal uses.
-pub type FileSectorReader = FileSectorSource;

src/udf.rs

@@ -19,7 +19,7 @@
 //!   BD-ROM Part 3 — Blu-ray filesystem profile
 
 use crate::error::{Error, Result};
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 
 /// A UDF filesystem parsed from disc.
 #[derive(Debug)]
@@ -84,7 +84,7 @@ impl UdfFs {
     /// Reads sector by sector from disc — no buffering.
     /// Get the absolute starting LBA of a file on disc.
     /// Used by the rip pipeline to locate m2ts content sectors.
-    pub fn file_start_lba(&self, reader: &mut dyn SectorReader, path: &str) -> Result<u32> {
+    pub fn file_start_lba(&self, reader: &mut dyn SectorSource, path: &str) -> Result<u32> {
         let parts: Vec<&str> = path.trim_matches('/').split('/').collect();
         let mut current = &self.root;
         for part in &parts[..parts.len() - 1] {
@@ -115,7 +115,7 @@ impl UdfFs {
         Ok(self.partition_start + data_lba)
     }
 
-    pub fn read_file(&self, reader: &mut dyn SectorReader, path: &str) -> Result<Vec<u8>> {
+    pub fn read_file(&self, reader: &mut dyn SectorSource, path: &str) -> Result<Vec<u8>> {
         let parts: Vec<&str> = path.trim_matches('/').split('/').collect();
         let mut current = &self.root;
 
@@ -175,7 +175,7 @@ impl UdfFs {
     ///
     /// Skips: STREAM/ (video), BACKUP/, DUPLICATE/,
     ///   MKB_RO.inf, ContentHash*, ContentRevocation*
-    pub fn metadata_sector_ranges(&self, reader: &mut dyn SectorReader) -> Result<Vec<(u32, u32)>> {
+    pub fn metadata_sector_ranges(&self, reader: &mut dyn SectorSource) -> Result<Vec<(u32, u32)>> {
         let mut ranges = Vec::new();
 
         // UDF structure: sector 0 through end of metadata partition
@@ -194,7 +194,7 @@ impl UdfFs {
 
     /// All sector ranges that contain data (metadata + all files including STREAM).
     /// For full disc-to-ISO dumps — reads only allocated sectors, skips gaps.
-    pub fn all_sector_ranges(&self, reader: &mut dyn SectorReader) -> Result<Vec<(u32, u32)>> {
+    pub fn all_sector_ranges(&self, reader: &mut dyn SectorSource) -> Result<Vec<(u32, u32)>> {
         let mut ranges = Vec::new();
 
         // UDF structure sectors
@@ -212,7 +212,7 @@ impl UdfFs {
 
     fn collect_all_file_ranges(
         &self,
-        reader: &mut dyn SectorReader,
+        reader: &mut dyn SectorSource,
         entry: &DirEntry,
         ranges: &mut Vec<(u32, u32)>,
     ) -> Result<()> {
@@ -238,7 +238,7 @@ impl UdfFs {
 
     fn collect_file_ranges(
         &self,
-        reader: &mut dyn SectorReader,
+        reader: &mut dyn SectorSource,
         entry: &DirEntry,
         ranges: &mut Vec<(u32, u32)>,
     ) -> Result<()> {
@@ -276,7 +276,7 @@ impl UdfFs {
     /// Read an Extended File Entry (tag 266) or File Entry (tag 261)
     /// and return its first allocation extent: (data_lba, data_length).
     /// The data_lba is partition-relative.
-    fn read_icb_extent(&self, reader: &mut dyn SectorReader, meta_lba: u32) -> Result<(u32, u32)> {
+    fn read_icb_extent(&self, reader: &mut dyn SectorSource, meta_lba: u32) -> Result<(u32, u32)> {
         let extents = self.read_icb_extents(reader, meta_lba)?;
         extents.first().copied().ok_or(Error::DiscRead {
             sector: 0,
@@ -290,7 +290,7 @@ impl UdfFs {
     /// Handles files with many extents (e.g. 88 GB m2ts files have ~90 extents).
     fn read_icb_extents(
         &self,
-        reader: &mut dyn SectorReader,
+        reader: &mut dyn SectorSource,
         meta_lba: u32,
     ) -> Result<Vec<(u32, u32)>> {
         let mut icb = [0u8; 2048];
@@ -367,7 +367,7 @@ impl UdfFs {
     /// Returns Vec of (absolute_lba, sector_count) covering the entire file.
     pub fn file_extents(
         &self,
-        reader: &mut dyn SectorReader,
+        reader: &mut dyn SectorSource,
         path: &str,
     ) -> Result<Vec<(u32, u32)>> {
         let parts: Vec<&str> = path.trim_matches('/').split('/').collect();
@@ -417,7 +417,7 @@ impl UdfFs {
 /// 3. Metadata partition file → metadata content location
 /// 4. FSD → root directory ICB
 /// 5. Root directory → file tree
-pub fn read_filesystem(reader: &mut dyn SectorReader) -> Result<UdfFs> {
+pub fn read_filesystem(reader: &mut dyn SectorSource) -> Result<UdfFs> {
     // Step 1: Anchor Volume Descriptor Pointer at sector 256
     // ECMA-167 §10.2 — always at sector 256
     let mut avdp = [0u8; 2048];
@@ -593,7 +593,7 @@ pub fn read_filesystem(reader: &mut dyn SectorReader) -> Result<UdfFs> {
 /// and points to its ICB.
 #[allow(clippy::only_used_in_recursion)]
 fn read_directory(
-    reader: &mut dyn SectorReader,
+    reader: &mut dyn SectorSource,
     part_start: u32,
     meta_start: u32,
     meta_lba: u32,
@@ -758,7 +758,7 @@ fn read_directory(
 }
 
 /// Read file size (info_length) from an Extended File Entry ICB.
-fn read_file_size(reader: &mut dyn SectorReader, meta_start: u32, meta_lba: u32) -> Result<u64> {
+fn read_file_size(reader: &mut dyn SectorSource, meta_start: u32, meta_lba: u32) -> Result<u64> {
     let mut icb = [0u8; 2048];
     read_sector(reader, meta_start + meta_lba, &mut icb)?;
 
@@ -874,7 +874,7 @@ fn parse_dstring(data: &[u8]) -> String {
 /// Each SCSI command has ~500ms overhead on USB drives, so reading 32 sectors
 /// at once (one command) is 32x faster than 32 individual reads.
 pub(crate) struct BufferedSectorReader<'a> {
-    inner: &'a mut dyn SectorReader,
+    inner: &'a mut dyn SectorSource,
     cache_start: u32,
     cache: Vec<u8>,
     cache_sectors: u32,
@@ -884,7 +884,7 @@ pub(crate) struct BufferedSectorReader<'a> {
 }
 
 impl<'a> BufferedSectorReader<'a> {
-    pub(crate) fn new(inner: &'a mut dyn SectorReader, batch: u16) -> Self {
+    pub(crate) fn new(inner: &'a mut dyn SectorSource, batch: u16) -> Self {
         Self {
             inner,
             cache_start: u32::MAX,
@@ -952,7 +952,7 @@ impl BufferedSectorReader<'_> {
     }
 }
 
-impl SectorReader for BufferedSectorReader<'_> {
+impl SectorSource for BufferedSectorReader<'_> {
     fn read_sectors(
         &mut self,
         lba: u32,
@@ -996,7 +996,7 @@ impl SectorReader for BufferedSectorReader<'_> {
     }
 }
 
-fn read_sector(reader: &mut dyn SectorReader, lba: u32, buf: &mut [u8]) -> Result<()> {
+fn read_sector(reader: &mut dyn SectorSource, lba: u32, buf: &mut [u8]) -> Result<()> {
     reader.read_sectors(lba, 1, buf, true)?;
     Ok(())
 }

src/verify.rs

@@ -2,7 +2,7 @@
 
 use crate::disc::{Chapter, DiscTitle};
 use crate::progress::Progress;
-use crate::sector::SectorReader;
+use crate::sector::SectorSource;
 use std::time::Instant;
 
 /// Health status of a single sector read.
@@ -82,7 +82,7 @@ impl VerifyResult {
 /// Reads in batches for speed, falls back to single-sector on failure.
 /// The progress callback returns false to request early stop.
 pub fn verify_title(
-    reader: &mut dyn SectorReader,
+    reader: &mut dyn SectorSource,
     title: &DiscTitle,
     batch_sectors: u16,
     on_progress: Option<&dyn Progress>,

tests/disc_tests.rs

@@ -1,6 +1,6 @@
 //! Disc scanning pipeline tests.
 
-use libfreemkv::SectorReader;
+use libfreemkv::SectorSource;
 use libfreemkv::error::Result;
 use libfreemkv::{Disc, DiscTitle, ScanOptions};
 use std::collections::HashMap;
@@ -20,7 +20,7 @@ impl MockSectorReader {
     }
 }
 
-impl SectorReader for MockSectorReader {
+impl SectorSource for MockSectorReader {
     fn read_sectors(
         &mut self,
         lba: u32,

tests/integration_progress_and_halt.rs

@@ -1,17 +1,12 @@
 //! Integration tests for progress reporting, halt behavior, drop safety,
 //! and the file-backed sector reader round trip.
 
-// 0.18 trait split: this suite still drives the deprecated `pes::Stream`
-// trait directly. It will be migrated to `FrameSource`/`FrameSink` in the
-// follow-up that ports concrete impls.
-#![allow(deprecated)]
-
 use libfreemkv::disc::{CopyOptions, DiscRegion};
 use libfreemkv::error::Result;
 use libfreemkv::pes::Stream as PesStream;
 use libfreemkv::{
-    ContentFormat, Disc, DiscFormat, DiscStream, DiscTitle, EventKind, Extent, FileSectorReader,
-    SectorReader,
+    ContentFormat, Disc, DiscFormat, DiscStream, DiscTitle, EventKind, Extent, FileSectorSource,
+    SectorSource,
 };
 use std::io::Write;
 use std::sync::Arc;
@@ -37,7 +32,7 @@ impl ZeroSectorReader {
     }
 }
 
-impl SectorReader for ZeroSectorReader {
+impl SectorSource for ZeroSectorReader {
     fn read_sectors(
         &mut self,
         _lba: u32,
@@ -51,7 +46,7 @@ impl SectorReader for ZeroSectorReader {
         Ok(bytes)
     }
 
-    fn capacity(&self) -> u32 {
+    fn capacity_sectors(&self) -> u32 {
         self.capacity
     }
 }
@@ -72,7 +67,7 @@ impl SlowZeroSectorReader {
     }
 }
 
-impl SectorReader for SlowZeroSectorReader {
+impl SectorSource for SlowZeroSectorReader {
     fn read_sectors(
         &mut self,
         _lba: u32,
@@ -86,7 +81,7 @@ impl SectorReader for SlowZeroSectorReader {
         Ok(bytes)
     }
 
-    fn capacity(&self) -> u32 {
+    fn capacity_sectors(&self) -> u32 {
         self.capacity
     }
 }
@@ -106,6 +101,7 @@ fn synthetic_disc(capacity_sectors: u32) -> Disc {
         aacs: None,
         css: None,
         encrypted: false,
+        aacs_error: None,
         content_format: ContentFormat::BdTs,
     }
 }
@@ -308,7 +304,7 @@ fn test_drop_impls_do_not_panic_or_block() {
     panic!("DiscStream drop did not complete within 100ms");
 }
 
-// ── 5. FileSectorReader round trip ────────────────────────────────────────
+// ── 5. FileSectorSource round trip ────────────────────────────────────────
 
 #[test]
 fn test_file_sector_reader_round_trip() {
@@ -325,9 +321,13 @@ fn test_file_sector_reader_round_trip() {
     tmp.flush().expect("flush");
 
     let path = tmp.path().to_path_buf();
-    let mut fsr = FileSectorReader::open(&path).expect("open FileSectorReader");
+    let mut fsr = FileSectorSource::open(&path).expect("open FileSectorSource");
 
-    assert_eq!(fsr.capacity(), N_SECTORS as u32, "capacity mismatch");
+    assert_eq!(
+        fsr.capacity_sectors(),
+        N_SECTORS as u32,
+        "capacity mismatch"
+    );
 
     // Read each sector individually and compare.
     let mut buf = vec![0u8; SECTOR_SIZE];
@@ -395,7 +395,7 @@ impl FailingSectorReader {
     }
 }
 
-impl SectorReader for FailingSectorReader {
+impl SectorSource for FailingSectorReader {
     fn read_sectors(
         &mut self,
         _lba: u32,
@@ -422,7 +422,7 @@ impl SectorReader for FailingSectorReader {
         })
     }
 
-    fn capacity(&self) -> u32 {
+    fn capacity_sectors(&self) -> u32 {
         self.capacity
     }
 }
@@ -571,7 +571,7 @@ struct BlockSizeFailingReader {
     capacity: u32,
 }
 
-impl SectorReader for BlockSizeFailingReader {
+impl SectorSource for BlockSizeFailingReader {
     fn read_sectors(
         &mut self,
         lba: u32,
@@ -597,7 +597,7 @@ impl SectorReader for BlockSizeFailingReader {
         }
     }
 
-    fn capacity(&self) -> u32 {
+    fn capacity_sectors(&self) -> u32 {
         self.capacity
     }
 }

tests/pass_n_size_aware_skip.rs

@@ -15,7 +15,7 @@ use libfreemkv::disc::CopyOptions;
 use libfreemkv::disc::DiscRegion;
 use libfreemkv::disc::mapfile::{Mapfile, SectorStatus};
 use libfreemkv::error::Result;
-use libfreemkv::{ContentFormat, Disc, DiscFormat, SectorReader};
+use libfreemkv::{ContentFormat, Disc, DiscFormat, SectorSource};
 use std::collections::HashSet;
 use std::sync::{Arc, Mutex};
 
@@ -46,7 +46,7 @@ impl PatternedSectorReader {
     }
 }
 
-impl SectorReader for PatternedSectorReader {
+impl SectorSource for PatternedSectorReader {
     fn read_sectors(
         &mut self,
         lba: u32,
@@ -82,7 +82,7 @@ impl SectorReader for PatternedSectorReader {
         Ok(buf.len())
     }
 
-    fn capacity(&self) -> u32 {
+    fn capacity_sectors(&self) -> u32 {
         self.capacity
     }
 }
@@ -100,6 +100,7 @@ fn synthetic_disc(capacity_sectors: u32) -> Disc {
         aacs: None,
         css: None,
         encrypted: false,
+        aacs_error: None,
         content_format: ContentFormat::BdTs,
     }
 }

tests/streams.rs

@@ -1,10 +1,5 @@
 //! Integration tests for the PES stream pipeline.
 
-// 0.18 trait split: this suite still drives the deprecated `pes::Stream`
-// trait directly. It will be migrated to `FrameSource`/`FrameSink` in the
-// follow-up that ports concrete impls.
-#![allow(deprecated)]
-
 use libfreemkv::mux::meta::M2tsMeta;
 use libfreemkv::pes::Stream as PesStream;
 use libfreemkv::*;

tests/udf_tests.rs

@@ -1,7 +1,7 @@
 //! UDF parser tests using a MockSectorReader.
 
 use libfreemkv::error::Result;
-use libfreemkv::{SectorReader, read_filesystem};
+use libfreemkv::{SectorSource, read_filesystem};
 use std::collections::HashMap;
 
 const SECTOR_SIZE: usize = 2048;
@@ -38,7 +38,7 @@ impl MockSectorReader {
     }
 }
 
-impl SectorReader for MockSectorReader {
+impl SectorSource for MockSectorReader {
     fn read_sectors(
         &mut self,
         lba: u32,
@@ -463,11 +463,11 @@ fn find_dir_case_insensitive() {
 
 #[test]
 fn sector_reader_is_object_safe() {
-    // Verify SectorReader can be used as a trait object
+    // Verify SectorSource can be used as a trait object
     let mut reader = MockSectorReader::new();
     reader.set_sector(0, vec![42u8; SECTOR_SIZE]);
 
-    let dyn_reader: &mut dyn SectorReader = &mut reader;
+    let dyn_reader: &mut dyn SectorSource = &mut reader;
     let mut buf = vec![0u8; SECTOR_SIZE];
     let n = dyn_reader.read_sectors(0, 1, &mut buf, true).unwrap();
     assert_eq!(n, SECTOR_SIZE);