v0.18.22: bump version

Cargo.toml

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

src/disc/mod.rs

@@ -3412,9 +3412,9 @@ pub fn detect_max_batch_sectors(device_path: &str) -> u16 {
     }
 
     // Check if optical drive (0x05 = CD/DVD)
-   let is_optical = (|| -> bool {
+    let is_optical = (|| -> bool {
         use std::path::Path;
-        let scsi_device_dir = format!("/sys/class/scsi_device/");
+        let scsi_device_dir = "/sys/class/scsi_device/".to_string();
         if let Ok(entries) = std::fs::read_dir(&scsi_device_dir) {
             for entry in entries.flatten() {
                 let device_type_path = entry.path().join("device/type");
@@ -3458,7 +3458,7 @@ pub fn detect_max_batch_sectors(device_path: &str) -> u16 {
                 }
             }
         }
-     DEFAULT_BATCH_SECTORS_OPTICAL
+        DEFAULT_BATCH_SECTORS_OPTICAL
     } else {
         DEFAULT_BATCH_SECTORS_BLOCK
     }

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/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};

src/mux/resolve.rs

@@ -207,7 +207,8 @@ pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::S
             // 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);
+            let mut stream =
+                DiscStream::new(Box::new(reader), title, keys, ISO_MUX_BATCH_SECTORS, format);
             if opts.raw {
                 stream.set_raw();
             }