v0.19.0: bump version

Cargo.toml

@@ -1,8 +1,8 @@
 [package]
 name = "libfreemkv"
-version = "0.18.18"
+version = "0.19.0"
 edition = "2024"
-rust-version = "1.86"
+rust-version = "0.19.0"
 license = "AGPL-3.0-only"
 description = "Open source raw disc access library for optical drives"
 repository = "https://github.com/freemkv/libfreemkv"
@@ -10,7 +10,7 @@ keywords = ["bluray", "uhd", "optical", "scsi", "disc"]
 categories = ["hardware-support", "multimedia"]
 
 [dependencies]
-serde = { version = "1", features = ["derive"] }
+serde = { version = "0.19.0", features = ["derive"] }
 serde_json = "1"
 sha1 = "0.10"
 sha2 = "0.10"
@@ -22,7 +22,7 @@ num-traits = "0.2"
 num-integer = "0.1"
 rand = "0.8"
 cmac = "0.7"
-zip = { version = "2", default-features = false, features = ["deflate"] }
+zip = { version = "0.19.0", default-features = false, features = ["deflate"] }
 base64 = "0.22.1"
 # Trace-level instrumentation for Disc::copy + SgIoTransport::execute. Permitted
 # under project docs ("Acceptable strings: debug/trace logging"). Consumers (autorip)

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,28 @@ 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. `16` matches the depth needed for UHD-scale
-/// mux where WritebackFile sync_file_range on NFS can stall the consumer;
-/// sweep uses [`DEFAULT_PIPELINE_DEPTH`] directly, mux should use this
-/// or deeper if ISO read is moved to a separate producer thread. 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.
-pub const DEFAULT_PIPELINE_DEPTH: usize = 32;
+/// 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
@@ -150,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
@@ -160,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 {
@@ -189,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/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;
@@ -133,11 +133,14 @@ pub use halt::Halt;
 // 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};

src/mux/disc.rs

@@ -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 SectorReader>()
+        );
+
         let mut pids = Vec::new();
         let mut parsers = Vec::new();
         let mut pid_to_track = Vec::new();
@@ -311,14 +318,24 @@ impl DiscStream {
         // on failure, advance on success. One 5s read attempt per try — no
         // retry loops, no sleeps. On size-1 failure, skip or error.
         //
-        // Halt is checked at the top of every iteration — in a dense bad zone
+      // Halt is checked at the top of every iteration — in a dense bad zone
         // 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_lba = lba;
+        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().