v0.17.11: sweep producer/consumer split — overlap drive read with file write

Pre-0.17.11 sweep ran strictly serialised: SCSI read → decrypt → seek
+ write → mapfile.record → next read. Drive idled for the post-read
work; throughput capped at the sum of both costs. On a healthy disc
that's ~7-12 ms read + ~5-15 ms write/record per 64 KB batch, limiting
sustained throughput to ~10-12 MB/s on the test bed (BU40N + UHD inner
zone), well below the ~14-16 MB/s drive ceiling.

Decouples them: producer thread (caller's) owns SectorReader +
read_error state + decrypt + set_speed + halt; consumer thread (one
spawn) owns Writer + Mapfile, receives WorkItem messages, applies
file write + mapfile record. Bounded mpsc::sync_channel(4) gives
natural back-pressure. While the consumer writes batch N, the
producer is already reading batch N+1 — steady-state throughput is
now bound by the slower of the two pipelines (drive on healthy
discs), not their sum.

Side effects:
- Bisect path now decrypts. Pre-0.17.11 the bisect inner loop wrote
  raw cyphertext for single-sector recoveries on encrypted discs —
  quiet correctness bug exercised only by batch-fail-then-
  bisect-succeed on encrypted media. New producer-side decrypt
  covers main + bisect success paths uniformly.
- All read_ctx state stays single-threaded on producer (damage
  window, jump multiplier, etc.). No locking added.
- Mapfile remains single-writer on consumer. No locking.
- Halt latency: producer breaks loop, sends Finish, consumer drains
  ≤4 in-flight items + sync_all. ~1 batch (~12 ms) typical.
- BU40N + Initio bridge wedge concern unchanged: still single SCSI
  command in flight, error-path timing identical, no new retries.

New module: src/disc/sweep_pipeline.rs (WorkItem, ProgressSnapshot,
ConsumerInputs, spawn_consumer, consumer_loop, helpers). Public API
unchanged — Disc::copy / CopyOptions / CopyResult identical.

Patch (Pass N) is NOT changed; it's bound by drive recovery time, not
the read/write serialisation.

CHANGELOG.md

@@ -1,5 +1,163 @@
 # Changelog
 
+## 0.17.11 (2026-05-09)
+
+### Sweep producer/consumer split — overlap drive read with file write
+
+Pre-0.17.11 the sweep loop ran strictly serialised: SCSI read → decrypt
+→ seek + write → mapfile.record → next read. The drive idled for the
+post-read work; throughput capped at the **sum** of both costs. On a
+healthy disc that's ~7-12 ms read + ~5-15 ms write/record per 64 KB
+batch — limiting sustained throughput to ~10-12 MB/s on the test bed
+(BU40N + UHD inner zone), well below the drive's ceiling of ~14-16 MB/s.
+
+This release decouples them with a producer / consumer split:
+
+- **Producer** (caller's thread): owns the `SectorReader`, the entire
+  `read_error` state machine (Retry, Bisect, SkipBlock, JumpAhead,
+  AbortPass), `set_speed` damage-zone transitions, halt check, and
+  decrypt. Hands plaintext bytes to the consumer.
+- **Consumer** (one spawned thread): owns the `crate::io::Writer` and
+  the `Mapfile`. Receives `WorkItem` messages and applies the file
+  write + mapfile record per item.
+- **Channel**: bounded `mpsc::sync_channel(4)` — natural back-pressure
+  via blocking `send` when consumer falls behind.
+
+While the consumer is writing batch N to disk and updating the mapfile,
+the producer is already reading batch N+1 from the drive. Steady-state
+throughput is now bound by the slower of the two pipelines (the drive,
+on a healthy disc) instead of their sum.
+
+Side effects of the refactor:
+
+- **Bisect path now decrypts.** Pre-0.17.11 the bisect inner loop wrote
+  raw cyphertext when `decrypt=true` and a single sector was recovered
+  via single-sector retry — a quiet correctness bug exercised only by
+  the (rare) batch-fail-then-bisect-succeed path on encrypted discs.
+  The new producer-side decrypt covers both the main success path and
+  the bisect inner success path.
+- All `read_error::ReadCtx` state stays single-threaded on the producer
+  (damage window, jump multiplier, consecutive-good count, etc.). No
+  locking added.
+- Mapfile remains single-writer (consumer-only). No locking.
+- Halt-flag responsiveness unchanged: producer breaks the loop on
+  signal, sends `Finish`, consumer drains its ≤4 in-flight items and
+  exits within ~1 batch (~12 ms typical).
+- BU40N + Initio bridge wedge concern unchanged: still one SCSI command
+  in flight, error-path timing identical, no new retry logic.
+
+New module: `src/disc/sweep_pipeline.rs` (`WorkItem`, `ProgressSnapshot`,
+`ConsumerInputs`, `spawn_consumer`, `consumer_loop`, send/recv helpers).
+Public API surface unchanged — `Disc::copy` / `CopyOptions` /
+`CopyResult` look identical to callers.
+
+Patch (Pass N) is **not** affected by this release. Patch is bound by
+drive recovery time (60 s timeouts on bad sectors), not the read↔write
+serialisation; a similar split there would yield negligible benefit.
+
+## 0.17.10 (2026-05-09)
+
+### Bounded-cache writeback for big sequential writes
+
+Pass 1 sweep speed on a healthy disc previously dipped from ~15 MB/s
+to ~1 MB/s every ~30 s on a host with default Linux dirty-page
+settings. Cause: the kernel's `vm.dirty_ratio` (~20 % of RAM) lets
+hundreds of MB of dirty pages accumulate, then bursts a flush at
+99 % disk utilisation that blocks app writes for ~1 s. Confirmed
+empirically on the BU40N test bed: dirty pages grew 112 → 563 MB
+between bursts; lowering `vm.dirty_bytes` to 64 MB at the host
+sysctl level eliminated the dips.
+
+This release ships the equivalent inside libfreemkv so users don't
+need to tune the host kernel:
+
+- New `crate::io::Writer` — drop-in `File` wrapper implementing
+  `Write` + `Seek`. Wraps a per-platform `WritebackPipeline` that on
+  Linux schedules `sync_file_range(WRITE)` + lagging
+  `sync_file_range(WAIT_AFTER)` + `posix_fadvise(DONTNEED)` calls in
+  32 MB chunks, keeping dirty cache bounded at ~64 MB. macOS and
+  Windows ship a no-op stub — their default cache policies don't
+  exhibit the same pathology for our access pattern.
+- Disc::sweep wraps its output `File` in `Writer`. No changes to the
+  loop body — `Writer` forwards `seek`/`write_all` to `File` and
+  drives the pipeline transparently.
+- Module is purpose-built for any large sequential output (sweep,
+  patch, future mux) — they can adopt `crate::io::Writer` with a
+  one-line wrapper and inherit the same behaviour.
+
+## 0.17.7 (2026-05-08)
+
+### Sync release — no functional libfreemkv changes
+
+Version bump to keep the four freemkv crates at unified versioning.
+autorip shipped UI smoothness + audit fixes today (v0.17.6, v0.17.7);
+libfreemkv carries no code changes, but is bumped + republished to
+crates.io so downstream `libfreemkv = "0.17"` consumers stay aligned
+on the latest patch version. See autorip CHANGELOG for the
+user-visible work in this release window.
+
+## 0.17.5 (2026-05-08)
+
+### Pass N recovery — kernel block-device fallback + per-range fixes
+
+Live testing on direct-SATA BU40N + AACS-encrypted UHD disc revealed that
+the v0.17.3 single-shot SCSI READ path matched 0/22 of the small bad-sector
+LBAs that `dd if=/dev/sr0` recovers on the same drive. This release closes
+that gap and fixes several adjacent bugs that were silently capping
+recovery.
+
+- **`/dev/sr0` pread fallback in `Drive::read` (Linux)**: when a SCSI READ
+  via `/dev/sg*` returns Err, fall back to `posix_fadvise(POSIX_FADV_DONTNEED)
+  + pread()` against the corresponding block device. The kernel `sr_mod`
+  driver runs ~5 internal retries per command without the per-attempt
+  error-escalation overhead that userspace SG_IO retries pay, which is the
+  source of dd's recovery advantage. End-to-end byte-verification confirms
+  the fallback path returns real disc data (md5-equivalent to fresh dd from
+  the same drive session). The block fd is opened in `Drive::open` by
+  resolving `/sys/class/scsi_generic/sgN/device/block`; best-effort, with no
+  fallback if open fails.
+
+- **`Disc::patch` per-range watchdog fix**: when a range hit `MAX_RANGE_SECS`,
+  the old code did `wedged_exit = true; break 'outer;` — a single slow range
+  killed the entire patch. Now `break;` (skip this range, advance the outer
+  for loop). Pre-fix, patch was dying after 4 sectors of the first slow
+  range and never reaching the other 46.
+
+- **Per-sector range budget**: replaced flat `MAX_RANGE_SECS = 180` with
+  `range_budget_secs = (range_sectors × SECONDS_PER_SECTOR).min(RANGE_BUDGET_CAP_SECS)`.
+  Tiny ranges exit fast (1-sector range = 25 s); medium ranges get
+  proportional time (51-sector range ≈ 1275 s); large ranges still bounded
+  by the 1800 s cap so they cannot monopolise pass 1.
+
+- **`consecutive_failures` resets per range**: the wedge-exit detector
+  (`>= 50 consecutive failures`) is for "stuck on the same range"; pre-fix
+  the counter persisted across ranges, so 50 small post-bisection ranges
+  with one failure each falsely tripped the wedge mid-pass. Reset at every
+  range boundary.
+
+- **Reverted inline 5× retry in patch**: a brief experiment that was
+  measurably harmful — each "2 s" SCSI timeout paid ~1.5 s kernel SCSI
+  mid-layer error-escalation overhead, so 5× retry took ~17 s per LBA and
+  triggered the per-range watchdog in 4 sectors. Restored
+  `READ_RECOVERY_TIMEOUT_MS = 60_000` (the v0.17.3 baseline; the kernel-
+  auto-retry pattern is now provided by the `/dev/sr0` fallback above).
+
+### Empirical results (Dune Part Two UHD, BU40N direct SATA)
+
+- Pass 1: **94.6 MB recovered** (28% of formerly-bad data, 33 sr0 fallback
+  saves), **11 s of main-title content** restored. Patch completed all 47
+  retryable ranges naturally (was wedging at range 1 of 47 in v0.17.3).
+- Pass 2 cumulative: 95.2 MB (+0.6 MB; diminishing returns curve).
+- Remaining ~233 MB on the test disc appears physically unrecoverable on
+  this hardware (kernel auto-retry can't decode it either).
+
+### Behavioural notes
+
+- The `/dev/sr0` fallback is Linux only; macOS and Windows fall back to
+  the existing single-shot SCSI behaviour. The fallback is gated to
+  `recovery=true` reads (only fires from the patch path, not the sweep
+  path) to avoid page-cache pressure during multi-GB sequential ripping.
+
 ## 0.17.0 (2026-05-04)
 
 ### Code quality: unwrap safety, clippy compliance, test coverage

Cargo.toml

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

src/disc/mod.rs

@@ -13,6 +13,7 @@ mod dvd;
 mod encrypt;
 pub mod mapfile;
 pub mod read_error;
+mod sweep_pipeline;
 
 use crate::drive::{Drive, extract_scsi_context};
 use crate::error::{Error, Result};
@@ -1369,7 +1370,10 @@ impl Disc {
         path: &std::path::Path,
         opts: &SweepOptions,
     ) -> Result<CopyResult> {
-        use std::io::{Seek, SeekFrom, Write};
+        use sweep_pipeline::{
+            ConsumerInputs, ProgressSnapshot, WorkItem, send_or_abort, spawn_consumer,
+            try_recv_progress, try_request_stats,
+        };
 
         let total_bytes = self.capacity_sectors as u64 * 2048;
         let keys = if opts.decrypt {
@@ -1383,7 +1387,7 @@ impl Disc {
         if !opts.resume {
             let _ = std::fs::remove_file(&mapfile_path);
         }
-        let mut map = mapfile::Mapfile::open_or_create(
+        let map = mapfile::Mapfile::open_or_create(
             &mapfile_path,
             total_bytes,
             concat!("libfreemkv v", env!("CARGO_PKG_VERSION")),
@@ -1414,13 +1418,31 @@ impl Disc {
             f
         };
 
-        let mut file = file;
+        // Wrap the raw `File` in our bounded-cache writer (drains
+        // dirty pages continuously instead of bursting; see
+        // `crate::io`). The Writer moves into the consumer thread.
+        let file = crate::io::Writer::new(file).map_err(|e| Error::IoError { source: e })?;
         let batch: u16 = match opts.batch_sectors {
             Some(b) => b,
             None if opts.skip_on_error => ecc_sectors(self.format),
             None => DEFAULT_BATCH_SECTORS,
         };
 
+        // Pre-compute the list of NonTried regions before handing the
+        // mapfile to the consumer thread. Each region is processed by
+        // the producer in order; the consumer mutates the mapfile per
+        // work-item. Any regions left as NonTrimmed/Unreadable after
+        // sweep finishes are the patch pass's job.
+        let regions: Vec<(u64, u64)> = map.ranges_with(&[mapfile::SectorStatus::NonTried]);
+
+        // Spawn the consumer. It owns Writer + Mapfile; the producer
+        // (this thread) keeps `reader`, `read_ctx`, halt + set_speed.
+        let (work_tx, prog_rx, consumer_handle) = spawn_consumer(ConsumerInputs {
+            file,
+            map,
+            is_regular,
+        });
+
         let mut buf = vec![0u8; batch as usize * 2048];
         let mut bytes_done = 0u64;
         let mut halt_requested = false;
@@ -1429,49 +1451,23 @@ impl Disc {
         let mut read_ok_count: u64 = 0;
         let mut read_err_count: u64 = 0;
         let mut last_log_iter: u64 = 0;
-        // ALL read state lives in one place. The single error-handling
-        // entry point (`read_error::handle_read_error`) owns the
-        // counters, retry budgets, and damage-window updates.
         let mut read_ctx = read_error::ReadCtx::for_sweep(batch);
-        // Speed control derives from damage-zone state. We track the
-        // transition locally so we only call set_speed on edges, not
-        // every iteration.
         let mut in_damage_zone = false;
         const DAMAGE_ZONE_EXIT_THRESHOLD: u64 = 16;
+        let mut cached_snapshot: Option<ProgressSnapshot> = None;
+        let mut producer_err: Option<Error> = None;
+
         tracing::trace!(
             target: "freemkv::disc",
             phase = "copy_start",
             total_bytes,
             batch,
             skip_on_error = opts.skip_on_error,
-            "Disc::copy entered"
+            regions = regions.len(),
+            "Disc::sweep entered (producer/consumer)"
         );
 
-        'outer: loop {
-            // Every pass retries every non-Finished range. Includes
-            // Unreadable so each pass gets its own shot at sectors prior
-            // passes gave up on — drive state may have changed (cooled
-            // down, bridge stabilized, etc.). Mapfile is binary in
-            // intent: Finished or not-yet-good.
-            let regions_to_do = map.ranges_with(&[
-                mapfile::SectorStatus::NonTried,
-                mapfile::SectorStatus::NonTrimmed,
-                mapfile::SectorStatus::NonScraped,
-                mapfile::SectorStatus::Unreadable,
-            ]);
-            tracing::trace!(
-                target: "freemkv::disc",
-                phase = "outer_loop",
-                regions_remaining = regions_to_do.len(),
-                "Disc::copy outer iter"
-            );
-            if regions_to_do.is_empty() {
-                break;
-            }
-            let Some((region_pos, region_size)) = map.next_with(0, mapfile::SectorStatus::NonTried)
-            else {
-                break;
-            };
+        'outer: for (region_pos, region_size) in regions {
             let region_end = region_pos + region_size;
             let mut pos = region_pos;
             tracing::trace!(
@@ -1505,12 +1501,9 @@ impl Disc {
 
                 match read_result {
                     Ok(_) => {
-                        // === SUCCESS PATH ===
                         read_ok_count += 1;
                         read_ctx.on_success();
 
-                        // Damage-zone exit: after enough consecutive good
-                        // reads, restore max speed and reset jump multiplier.
                         if read_ctx.consecutive_good >= DAMAGE_ZONE_EXIT_THRESHOLD {
                             read_ctx.jump_multiplier = 1;
                             if in_damage_zone {
@@ -1526,6 +1519,7 @@ impl Disc {
                         }
                         read_ctx.bridge_degradation_count = 0;
 
+                        // Decrypt on producer; consumer expects plaintext.
                         if opts.decrypt {
                             crate::decrypt::decrypt_sectors(
                                 &mut buf[..block_bytes as usize],
@@ -1533,31 +1527,30 @@ impl Disc {
                                 0,
                             )?;
                         }
-                        file.seek(SeekFrom::Start(pos))
-                            .map_err(|e| Error::IoError { source: e })?;
-                        file.write_all(&buf[..block_bytes as usize])
-                            .map_err(|e| Error::IoError { source: e })?;
-                        map.record(pos, block_bytes, mapfile::SectorStatus::Finished)
-                            .map_err(|e| Error::IoError { source: e })?;
+
+                        // Move the batch into the channel via fresh
+                        // owned Vec. The producer's `buf` is reused
+                        // for the next read.
+                        let send_buf = buf[..block_bytes as usize].to_vec();
+                        if let Err(e) =
+                            send_or_abort(&work_tx, WorkItem::Good { pos, buf: send_buf })
+                        {
+                            producer_err = Some(e);
+                            break 'outer;
+                        }
                         bytes_done = bytes_done.saturating_add(block_bytes);
                         pos += block_bytes;
                     }
                     Err(err) if !opts.skip_on_error => {
-                        // Caller asked us not to skip. Surface the error verbatim.
                         let (status, sense) = extract_scsi_context(&err);
-                        return Err(Error::DiscRead {
+                        producer_err = Some(Error::DiscRead {
                             sector: block_lba as u64,
                             status: Some(status),
                             sense,
                         });
+                        break 'outer;
                     }
                     Err(err) => {
-                        // === ERROR PATH — single source of truth ===
-                        // ALL errors flow through handle_read_error. New
-                        // error class = one new arm in that function.
-                        // Logging, counter updates, retry budgets all live
-                        // there. The dispatch below is purely the I/O side
-                        // of each action.
                         read_err_count += 1;
                         let action = read_error::handle_read_error(&err, &mut read_ctx);
 
@@ -1566,22 +1559,18 @@ impl Disc {
                                 if pause_secs > 0 {
                                     std::thread::sleep(std::time::Duration::from_secs(pause_secs));
                                 }
-                                // Don't advance pos — same LBA next iteration.
                             }
                             read_error::ReadAction::Bisect => {
-                                // Re-issue the failed batch as single-sector reads.
-                                // ctx.bisecting=true so the inner failures don't
-                                // recursively request another bisect.
                                 read_ctx.bisecting = true;
                                 let saved_batch = read_ctx.batch;
                                 read_ctx.batch = 1;
+                                let mut bisect_aborted = false;
                                 for sector_offset in 0..block_count {
                                     if let Some(ref h) = opts.halt {
                                         if h.load(std::sync::atomic::Ordering::Relaxed) {
                                             halt_requested = true;
-                                            read_ctx.bisecting = false;
-                                            read_ctx.batch = saved_batch;
-                                            break 'outer;
+                                            bisect_aborted = true;
+                                            break;
                                         }
                                     }
                                     let sector_lba = block_lba + (sector_offset as u32);
@@ -1595,54 +1584,63 @@ impl Disc {
                                     ) {
                                         Ok(_) => {
                                             read_ctx.on_success();
-                                            file.seek(SeekFrom::Start(write_pos))
-                                                .map_err(|e| Error::IoError { source: e })?;
-                                            file.write_all(&sector_buf)
-                                                .map_err(|e| Error::IoError { source: e })?;
-                                            map.record(
-                                                write_pos,
-                                                2048,
-                                                mapfile::SectorStatus::Finished,
-                                            )
-                                            .map_err(|e| Error::IoError { source: e })?;
+                                            // Decrypt single sector before send. Pre-split
+                                            // bisect path silently skipped this — encrypted
+                                            // bytes were written for bisect-recovered sectors.
+                                            if opts.decrypt {
+                                                crate::decrypt::decrypt_sectors(
+                                                    &mut sector_buf,
+                                                    &keys,
+                                                    0,
+                                                )?;
+                                            }
+                                            if let Err(e) = send_or_abort(
+                                                &work_tx,
+                                                WorkItem::BisectGood {
+                                                    pos: write_pos,
+                                                    buf: Box::new(sector_buf),
+                                                },
+                                            ) {
+                                                producer_err = Some(e);
+                                                bisect_aborted = true;
+                                                break;
+                                            }
                                         }
                                         Err(inner_err) => {
-                                            // Inner failure goes through the same handler — it'll
-                                            // see bisecting=true and won't recurse. We only honour
-                                            // the SkipBlock action here (the bisect by definition
-                                            // can't return another Bisect, and JumpAhead inside a
-                                            // single-sector retry doesn't make sense).
                                             let _ = read_error::handle_read_error(
                                                 &inner_err,
                                                 &mut read_ctx,
                                             );
-                                            let zero = [0u8; 2048];
-                                            file.seek(SeekFrom::Start(write_pos))
-                                                .map_err(|e| Error::IoError { source: e })?;
-                                            file.write_all(&zero)
-                                                .map_err(|e| Error::IoError { source: e })?;
-                                            map.record(
-                                                write_pos,
-                                                2048,
-                                                mapfile::SectorStatus::NonTrimmed,
-                                            )
-                                            .map_err(|e| Error::IoError { source: e })?;
+                                            if let Err(e) = send_or_abort(
+                                                &work_tx,
+                                                WorkItem::BisectBad { pos: write_pos },
+                                            ) {
+                                                producer_err = Some(e);
+                                                bisect_aborted = true;
+                                                break;
+                                            }
                                         }
                                     }
                                 }
                                 read_ctx.bisecting = false;
                                 read_ctx.batch = saved_batch;
+                                if bisect_aborted {
+                                    break 'outer;
+                                }
                                 bytes_done = bytes_done.saturating_add(block_bytes);
                                 pos += block_bytes;
                             }
                             read_error::ReadAction::SkipBlock { pause_secs } => {
-                                let zero = vec![0u8; block_bytes as usize];
-                                file.seek(SeekFrom::Start(pos))
-                                    .map_err(|e| Error::IoError { source: e })?;
-                                file.write_all(&zero)
-                                    .map_err(|e| Error::IoError { source: e })?;
-                                map.record(pos, block_bytes, mapfile::SectorStatus::NonTrimmed)
-                                    .map_err(|e| Error::IoError { source: e })?;
+                                if let Err(e) = send_or_abort(
+                                    &work_tx,
+                                    WorkItem::SkipFill {
+                                        pos,
+                                        len: block_bytes,
+                                    },
+                                ) {
+                                    producer_err = Some(e);
+                                    break 'outer;
+                                }
                                 bytes_done = bytes_done.saturating_add(block_bytes);
                                 if pause_secs > 0 {
                                     std::thread::sleep(std::time::Duration::from_secs(pause_secs));
@@ -1653,17 +1651,18 @@ impl Disc {
                                 sectors,
                                 pause_secs,
                             } => {
-                                // Mark the failed batch + the gap up to jump_pos NonTrimmed.
-                                let zero_batch = vec![0u8; block_bytes as usize];
-                                file.seek(SeekFrom::Start(pos))
-                                    .map_err(|e| Error::IoError { source: e })?;
-                                file.write_all(&zero_batch)
-                                    .map_err(|e| Error::IoError { source: e })?;
-                                map.record(pos, block_bytes, mapfile::SectorStatus::NonTrimmed)
-                                    .map_err(|e| Error::IoError { source: e })?;
+                                if let Err(e) = send_or_abort(
+                                    &work_tx,
+                                    WorkItem::SkipFill {
+                                        pos,
+                                        len: block_bytes,
+                                    },
+                                ) {
+                                    producer_err = Some(e);
+                                    break 'outer;
+                                }
                                 bytes_done = bytes_done.saturating_add(block_bytes);
 
-                                // Damage-zone enter: drop to minimum read speed.
                                 if !in_damage_zone {
                                     in_damage_zone = true;
                                     reader.set_speed(0x0000);
@@ -1679,22 +1678,16 @@ impl Disc {
                                 let gap_start = pos + block_bytes;
                                 let gap_bytes = jump_pos.saturating_sub(gap_start);
                                 if gap_bytes > 0 {
-                                    let zero_gap = vec![0u8; 65536];
-                                    let mut filled: u64 = 0;
-                                    while filled < gap_bytes {
-                                        let chunk = (gap_bytes - filled).min(zero_gap.len() as u64);
-                                        file.seek(SeekFrom::Start(gap_start + filled))
-                                            .map_err(|e| Error::IoError { source: e })?;
-                                        file.write_all(&zero_gap[..chunk as usize])
-                                            .map_err(|e| Error::IoError { source: e })?;
-                                        filled += chunk;
+                                    if let Err(e) = send_or_abort(
+                                        &work_tx,
+                                        WorkItem::GapFill {
+                                            pos: gap_start,
+                                            len: gap_bytes,
+                                        },
+                                    ) {
+                                        producer_err = Some(e);
+                                        break 'outer;
                                     }
-                                    map.record(
-                                        gap_start,
-                                        gap_bytes,
-                                        mapfile::SectorStatus::NonTrimmed,
-                                    )
-                                    .map_err(|e| Error::IoError { source: e })?;
                                     bytes_done = bytes_done.saturating_add(gap_bytes);
                                 }
                                 tracing::warn!(
@@ -1712,11 +1705,12 @@ impl Disc {
                             }
                             read_error::ReadAction::AbortPass => {
                                 let (status, sense) = extract_scsi_context(&err);
-                                return Err(Error::DiscRead {
+                                producer_err = Some(Error::DiscRead {
                                     sector: block_lba as u64,
                                     status: Some(status),
                                     sense,
                                 });
+                                break 'outer;
                             }
                         }
                     }
@@ -1724,45 +1718,65 @@ impl Disc {
 
                 iter_count += 1;
 
+                // Drain any consumer-side stats snapshot.
+                if let Some(snap) = try_recv_progress(&prog_rx) {
+                    cached_snapshot = Some(snap);
+                }
+
                 if iter_count - last_log_iter >= 100 {
                     last_log_iter = iter_count;
-                    let stats = map.stats();
-                    tracing::trace!(
-                        target: "freemkv::disc",
-                        phase = "iter_progress",
-                        iter_count,
-                        read_ok_count,
-                        read_err_count,
-                        pos,
-                        region_end,
-                        bytes_good = stats.bytes_good,
-                        bytes_pending = stats.bytes_pending,
-                        copy_elapsed_ms = copy_t0.elapsed().as_millis() as u64,
-                        "Disc::copy inner iter"
-                    );
+                    if let Some(ref snap) = cached_snapshot {
+                        tracing::trace!(
+                            target: "freemkv::disc",
+                            phase = "iter_progress",
+                            iter_count,
+                            read_ok_count,
+                            read_err_count,
+                            pos,
+                            region_end,
+                            bytes_good = snap.stats.bytes_good,
+                            bytes_pending = snap.stats.bytes_pending,
+                            copy_elapsed_ms = copy_t0.elapsed().as_millis() as u64,
+                            "Disc::sweep inner iter"
+                        );
+                    }
+                    // Throttled stats refresh request.
+                    try_request_stats(&work_tx);
                 }
 
                 if let Some(reporter) = opts.progress {
-                    let stats = map.stats();
-                    let bad_ranges = map.ranges_with(&[
-                        mapfile::SectorStatus::NonTrimmed,
-                        mapfile::SectorStatus::Unreadable,
-                        mapfile::SectorStatus::NonScraped,
-                        mapfile::SectorStatus::NonTried,
-                    ]);
-                    let main_title_bad = self
-                        .titles
-                        .first()
-                        .map(|t| bytes_bad_in_title(t, &bad_ranges))
-                        .unwrap_or(0);
+                    // Use the latest consumer snapshot if we have
+                    // one; otherwise synthesise a producer-side
+                    // placeholder. On a fresh sweep, before the
+                    // first stats round-trip lands, this means
+                    // bytes_good ≈ bytes_done (producer's notion of
+                    // good-so-far) and the bad-range list is empty —
+                    // close enough for an early UI tick; the next
+                    // real snapshot replaces it.
                     let main_title = self.titles.first();
+                    let main_title_bad = match &cached_snapshot {
+                        Some(snap) => self
+                            .titles
+                            .first()
+                            .map(|t| bytes_bad_in_title(t, &snap.bad_ranges))
+                            .unwrap_or(0),
+                        None => 0,
+                    };
+                    let (bytes_good, bytes_unreadable, bytes_pending) = match &cached_snapshot {
+                        Some(snap) => (
+                            snap.stats.bytes_good,
+                            snap.stats.bytes_unreadable,
+                            snap.stats.bytes_pending,
+                        ),
+                        None => (bytes_done, 0u64, total_bytes.saturating_sub(bytes_done)),
+                    };
                     let pp = crate::progress::PassProgress {
                         kind: crate::progress::PassKind::Sweep,
                         work_done: pos,
                         work_total: total_bytes,
-                        bytes_good_total: stats.bytes_good,
-                        bytes_unreadable_total: stats.bytes_unreadable,
-                        bytes_pending_total: stats.bytes_pending,
+                        bytes_good_total: bytes_good,
+                        bytes_unreadable_total: bytes_unreadable,
+                        bytes_pending_total: bytes_pending,
                         bytes_total_disc: total_bytes,
                         disc_duration_secs: main_title.map(|t| t.duration_secs),
                         bytes_bad_in_main_title: main_title_bad,
@@ -1777,32 +1791,26 @@ impl Disc {
             }
         }
 
-        tracing::debug!(
-            target: "freemkv::disc",
-            phase = "sweep_sync",
-            file_len = file.metadata().map(|m| m.len()).unwrap_or(0),
-            "sweep: calling sync_all"
-        );
-        if let Err(e) = file.sync_all() {
-            if is_regular {
-                tracing::warn!(
-                    target: "freemkv::disc",
-                    phase = "sweep_sync_failed",
-                    error = %e,
-                    os_error = e.raw_os_error(),
-                    error_kind = ?e.kind(),
-                    "sweep: sync_all failed"
-                );
-                return Err(Error::IoError { source: e });
-            }
-            tracing::debug!(
-                target: "freemkv::disc",
-                phase = "sweep_sync_skipped",
-                error = %e,
-                "sweep: sync_all failed for non-regular file; ignoring"
-            );
+        // Tell the consumer we're done. Even on producer error, send
+        // Finish so the consumer drains cleanly and we get a summary.
+        let _ = work_tx.send(WorkItem::Finish);
+        drop(work_tx);
+
+        let summary = consumer_handle.join().map_err(|_| Error::IoError {
+            source: std::io::Error::other("sweep consumer thread panicked"),
+        })?;
+
+        // Producer-side error wins over consumer-side (the read failure
+        // is what motivated quitting; the consumer's flush error, if
+        // any, is downstream).
+        if let Some(e) = producer_err {
+            return Err(e);
         }
-        let stats = map.stats();
+        if let Some(e) = summary.error {
+            return Err(e);
+        }
+
+        let stats = summary.stats;
         tracing::debug!(
             target: "freemkv::disc",
             phase = "sweep_done",
@@ -1995,14 +2003,33 @@ impl Disc {
         let mut blocks_attempted: u64 = 0;
         let mut blocks_read_ok: u64 = 0;
         let mut blocks_read_failed: u64 = 0;
-        let mut consecutive_failures: u64 = 0;
+        // Reset to 0 at the start of every range; declared without init
+        // because the per-range reset (below) always runs before any read.
+        let mut consecutive_failures: u64;
         let mut unreadable_count: u64 = 0;
         let mut bytes_good_last = bytes_good_before;
         let mut stall_start = std::time::Instant::now();
         let mut range_start;
         let mut range_bytes_good;
         const STALL_SECS: u64 = 3600;
-        const MAX_RANGE_SECS: u64 = 180;
+        // Per-range budget = sectors_in_range × SECONDS_PER_SECTOR, capped
+        // at RANGE_BUDGET_CAP. Replaces the old flat 180 s/range — that
+        // was unfair to medium ranges (a 51-sector range got the same
+        // 180 s as a 1-sector range, so multi-sector ranges couldn't
+        // even attempt every sector inside their budget) and pointlessly
+        // generous to single-sector ranges (180 s when ~5 s would do).
+        // The cap keeps catastrophic ranges (10s of MB) bounded so they
+        // can't consume the entire patch run; multi-pass orchestration
+        // raises the cap on later passes for the genuinely-stuck ones.
+        // Empirical per-failed-sector cost on direct-SATA BU40N (2026-05-08):
+        // ~3 s SCSI READ failure + ~15 s sr0 pread fallback (kernel sr_mod
+        // does ~5 internal retries) ≈ 18-25 s total. SECONDS_PER_SECTOR=25
+        // lets a small range fully sample within budget instead of bailing
+        // after one slow read. Previous value of 5 was too tight: a
+        // 3-sector range got 15 s budget but the first failed read alone
+        // took ~20 s, so the watchdog fired before sector 2 could be tried.
+        const SECONDS_PER_SECTOR: u64 = 25;
+        const RANGE_BUDGET_CAP_SECS: u64 = 1800;
         const MAX_SKIPS_PER_RANGE: u32 = 10;
         let mut skip_count: u32;
         let mut buf = vec![0u8; block_sectors as usize * 2048];
@@ -2132,6 +2159,24 @@ impl Disc {
             range_start = std::time::Instant::now();
             range_bytes_good = bytes_good_before;
             skip_count = 0;
+            // Reset consecutive_failures at each range boundary. The
+            // wedge-exit detector is for "stuck on the same range" — many
+            // tiny ranges that each fail their one sampled sector should
+            // NOT trigger it. Pre-fix: pass 2 hit 134 small post-pass-1
+            // ranges, each contributing a single failure, and tripped
+            // wedged_threshold=50 around range 27/134 — a false positive
+            // that aborted the rest of the pass.
+            consecutive_failures = 0;
+            let range_sectors = *range_size / 2048;
+            let range_budget_secs = (range_sectors * SECONDS_PER_SECTOR).min(RANGE_BUDGET_CAP_SECS);
+            tracing::debug!(
+                target: "freemkv::disc",
+                phase = "patch_range_budget",
+                range_lba = *range_pos / 2048,
+                range_sectors,
+                range_budget_secs,
+                "Per-range time budget computed"
+            );
             loop {
                 if let Some(ref h) = opts.halt {
                     if h.load(std::sync::atomic::Ordering::Relaxed) {
@@ -2140,39 +2185,47 @@ impl Disc {
                     }
                 }
 
-                // Test 1: Range timeout - max 3 minutes per range
-                if range_start.elapsed().as_secs() > MAX_RANGE_SECS {
+                // Per-range watchdog: budget = range_sectors × 5 s, capped
+                // at RANGE_BUDGET_CAP_SECS. Tiny ranges exit fast (1-sector
+                // range = 5 s budget); medium ranges get proportional time
+                // (51-sector range = 255 s); huge ranges still bounded by
+                // the cap so they can't monopolise pass 1.
+                //
+                // Both the absolute-elapsed and no-progress checks share
+                // the same per-range budget. The progress check resets
+                // range_start on every byte gained, so a steadily-recovering
+                // range can run as long as it makes progress.
+                if range_start.elapsed().as_secs() > range_budget_secs {
                     tracing::warn!(
                         target: "freemkv::disc",
                         phase = "patch_range_timeout",
                         range_lba = range_pos / 2048,
+                        range_sectors,
                         elapsed_secs = range_start.elapsed().as_secs(),
-                        bytes_recovered = bytes_good_before - range_bytes_good,
-                        "Range timeout - no progress in {}s, aborting",
-                        MAX_RANGE_SECS
+                        budget_secs = range_budget_secs,
+                        bytes_recovered = range_bytes_good.saturating_sub(bytes_good_before),
+                        "Range timeout - moving to next range"
                     );
-                    wedged_exit = true;
-                    break 'outer;
+                    break;
                 }
 
-                // Test 2: Range progress - must recover bytes in 60 seconds
                 let bytes_good_now = map.stats().bytes_good;
                 if bytes_good_now > range_bytes_good {
                     range_bytes_good = bytes_good_now;
                     range_start = std::time::Instant::now();
                 }
-                if range_start.elapsed().as_secs() > MAX_RANGE_SECS {
+                if range_start.elapsed().as_secs() > range_budget_secs {
                     tracing::warn!(
                         target: "freemkv::disc",
                         phase = "patch_range_stall",
                         range_lba = range_pos / 2048,
+                        range_sectors,
                         elapsed_secs = range_start.elapsed().as_secs(),
-                        bytes_recovered = bytes_good_before - range_bytes_good,
-                        "Range stalled - no recovery in {}s, aborting",
-                        MAX_RANGE_SECS
+                        budget_secs = range_budget_secs,
+                        bytes_recovered = range_bytes_good.saturating_sub(bytes_good_before),
+                        "Range stalled - moving to next range"
                     );
-                    wedged_exit = true;
-                    break 'outer;
+                    break;
                 }
 
                 // Test 3: Skip count - max 10 skips per range
@@ -2264,6 +2317,16 @@ impl Disc {
                     }
                 }
 
+                // Single-shot read. Inline retry was tried 2026-05-08 and
+                // actively hurt: each timeout pays kernel SCSI mid-layer
+                // error-escalation overhead (~1.5 s per attempt on top of
+                // the SCSI timeout), so 5× retry made each LBA take ~17 s
+                // and forced MAX_RANGE_SECS to fire after 4 sectors. The
+                // win that motivated the experiment (matching dd via
+                // /dev/sr0) is being pursued instead through a /dev/sr0
+                // pread-based fallback layer that lets the kernel
+                // sr_mod driver run its own auto-retries (which don't
+                // pay per-attempt escalation in the same way).
                 let read_start = std::time::Instant::now();
                 let read_result = reader.read_sectors(lba, count, &mut buf[..bytes], recovery);
                 let read_duration_ms = read_start.elapsed().as_millis();

src/disc/sweep_pipeline.rs

@@ -0,0 +1,309 @@
+//! Producer / consumer split for `Disc::sweep`.
+//!
+//! Background: the original sweep loop runs strictly serialised —
+//! SCSI read → decrypt → seek + write → mapfile.record → next iter.
+//! On a healthy disc the SCSI read costs ~5-12 ms per 64 KB batch and
+//! the post-read work (decrypt 1-3 ms + file write + mapfile fsync
+//! 5-15 ms) adds another batch's worth of latency. The drive idles
+//! during the post-read work; throughput tops out at the *sum* of
+//! both costs.
+//!
+//! This module decouples them. A consumer thread owns the
+//! [`crate::io::Writer`] (the ISO file) and the
+//! [`super::mapfile::Mapfile`]. The producer thread (the caller of
+//! `Disc::sweep`) keeps the [`crate::sector::SectorReader`], the
+//! [`super::read_error`] state machine, and decrypt — so what enters
+//! the channel is already-clean cleartext bytes, matching the "disc
+//! hands plaintext to its consumers" semantic that
+//! [`crate::mux::DiscStream`] already follows. Producer and consumer
+//! run concurrently; with a healthy disc the drive can read the next
+//! batch while the previous one is being written and recorded.
+//!
+//! Correctness invariants preserved:
+//! - 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.
+//! - Per-iteration ordering of file-write → mapfile-record is kept
+//!   intact in the consumer (write before record, same as today), so
+//!   the on-disk invariant "mapfile only marks Finished what the file
+//!   has received" survives a crash mid-pass.
+//! - The BU40N+Initio bridge wedge concern is unchanged: only one
+//!   SCSI command in flight at a time, error-path timing identical,
+//!   no new retry logic.
+
+use std::io::{Seek, SeekFrom, Write};
+use std::sync::mpsc::{Receiver, SyncSender, TrySendError, sync_channel};
+use std::thread::{self, JoinHandle};
+
+use crate::error::{Error, Result};
+
+use super::mapfile::{MapStats, Mapfile, SectorStatus};
+
+/// Channel depth for in-flight work items. 4 is enough to absorb a
+/// mapfile-flush burst on the consumer without growing memory
+/// unboundedly. Producer back-pressure is the natural rate limiter:
+/// `SyncSender::send` blocks when the channel is full.
+const CHANNEL_DEPTH: usize = 4;
+
+/// Reusable zero buffer for SkipFill / GapFill / BisectBad. 64 KB
+/// matches the existing zero_gap chunk size used by the pre-split
+/// sweep loop.
+const ZERO_CHUNK: usize = 65 * 1024;
+
+/// Producer → Consumer messages. The consumer applies these in FIFO
+/// order; ordering of file writes and mapfile records across items is
+/// preserved.
+pub(super) enum WorkItem {
+    /// Successful batch read. Producer has already decrypted `buf` if
+    /// `opts.decrypt` was set. Consumer writes `buf` at `pos` and
+    /// records the range as `Finished`.
+    Good { pos: u64, buf: Vec<u8> },
+
+    /// Bisect inner-loop good single sector (already decrypted by the
+    /// producer). 2048 bytes.
+    BisectGood { pos: u64, buf: Box<[u8; 2048]> },
+
+    /// Bisect inner-loop bad single sector. Consumer writes 2048
+    /// zeros at `pos` and records the sector as `NonTrimmed`.
+    BisectBad { pos: u64 },
+
+    /// Whole-batch zero-fill (failed batch on `SkipBlock`, or the
+    /// failed batch portion of `JumpAhead`). Consumer streams zeros
+    /// across `[pos, pos+len)` and records the range as `NonTrimmed`.
+    SkipFill { pos: u64, len: u64 },
+
+    /// Gap fill following a `JumpAhead`. Same effect as `SkipFill`;
+    /// distinguished only so future logging / instrumentation can
+    /// tell them apart without parsing a flag.
+    GapFill { pos: u64, len: u64 },
+
+    /// Producer wants the latest mapfile stats for the progress
+    /// callback. Consumer responds on `prog_tx` with a fresh
+    /// [`ProgressSnapshot`]. Best-effort: if the producer hasn't
+    /// drained the previous snapshot, the new one is silently
+    /// dropped — the producer's local cache stays current enough.
+    StatsRequest,
+
+    /// Producer is done. Consumer drains, runs `sync_all` on the
+    /// file, and exits. Final stats are returned via the
+    /// `JoinHandle<ConsumerSummary>` from `spawn_consumer`.
+    Finish,
+}
+
+/// Snapshot the consumer sends back to the producer for the progress
+/// callback.
+pub(super) struct ProgressSnapshot {
+    pub stats: MapStats,
+    pub bad_ranges: Vec<(u64, u64)>,
+}
+
+/// Final summary returned by the consumer thread on shutdown. The
+/// producer reads it via the `JoinHandle` returned from
+/// [`spawn_consumer`].
+pub(super) struct ConsumerSummary {
+    pub stats: MapStats,
+    /// First mapfile/write error the consumer hit, if any. The
+    /// producer treats this as fatal on the way back up.
+    pub error: Option<Error>,
+}
+
+/// Owned bundle the consumer thread takes ownership of. Decrypt
+/// happens on the producer side before send, so the consumer never
+/// sees keys.
+pub(super) struct ConsumerInputs {
+    pub file: crate::io::Writer,
+    pub map: Mapfile,
+    /// `sync_all`-on-failure-is-an-error iff the output is a regular
+    /// file. `/dev/null` and pipes always fail `sync_all`; that's not
+    /// a real error.
+    pub is_regular: bool,
+}
+
+/// Spawn the consumer thread. The producer keeps the work-tx and
+/// prog-rx; the join handle yields the final summary on `Finish` (or
+/// on channel close).
+pub(super) fn spawn_consumer(
+    inputs: ConsumerInputs,
+) -> (
+    SyncSender<WorkItem>,
+    Receiver<ProgressSnapshot>,
+    JoinHandle<ConsumerSummary>,
+) {
+    let (work_tx, work_rx) = sync_channel::<WorkItem>(CHANNEL_DEPTH);
+    let (prog_tx, prog_rx) = sync_channel::<ProgressSnapshot>(1);
+
+    let handle = thread::Builder::new()
+        .name("freemkv-sweep-consumer".into())
+        .spawn(move || consumer_loop(inputs, work_rx, prog_tx))
+        .expect("spawning sweep consumer thread should not fail");
+
+    (work_tx, prog_rx, handle)
+}
+
+/// Send a work item, translating a `SendError` (consumer thread died
+/// / panicked) into a useful library error so the caller can
+/// propagate cleanly.
+pub(super) fn send_or_abort(tx: &SyncSender<WorkItem>, item: WorkItem) -> Result<()> {
+    tx.send(item).map_err(|_| Error::IoError {
+        source: std::io::Error::other("sweep consumer terminated unexpectedly"),
+    })
+}
+
+/// Best-effort `StatsRequest` send. If the channel is full, skip —
+/// the producer's cached snapshot is fine for one more iteration.
+pub(super) fn try_request_stats(tx: &SyncSender<WorkItem>) {
+    if let Err(TrySendError::Full(_)) = tx.try_send(WorkItem::StatsRequest) {
+        // expected when the consumer is busy; cached snapshot is
+        // still recent enough.
+    }
+}
+
+/// Drain any pending progress snapshots from the consumer. Returns
+/// the most recent one, if any. The producer caches it and uses it
+/// for subsequent progress callbacks until a fresh one arrives.
+pub(super) fn try_recv_progress(rx: &Receiver<ProgressSnapshot>) -> Option<ProgressSnapshot> {
+    let mut latest = None;
+    while let Ok(snap) = rx.try_recv() {
+        latest = Some(snap);
+    }
+    latest
+}
+
+fn consumer_loop(
+    mut inputs: ConsumerInputs,
+    work_rx: Receiver<WorkItem>,
+    prog_tx: SyncSender<ProgressSnapshot>,
+) -> ConsumerSummary {
+    let zero = [0u8; ZERO_CHUNK];
+    let mut first_error: Option<Error> = None;
+
+    // Channel closed without a Finish == treat as Finish (producer
+    // dropped tx without explicit teardown — should not happen in
+    // normal operation but be defensive).
+    while let Ok(item) = work_rx.recv() {
+        // Once we have an error, drain remaining items without
+        // applying side-effects so the producer never blocks on a
+        // dead consumer. Loop until Finish or channel close.
+        if first_error.is_some() {
+            if matches!(item, WorkItem::Finish) {
+                break;
+            }
+            continue;
+        }
+
+        match apply_item(&mut inputs, item, &zero, &prog_tx) {
+            Ok(true) => {}
+            Ok(false) => break, // Finish received
+            Err(e) => first_error = Some(e),
+        }
+    }
+
+    // Final flush — drain the writeback pipeline + fsync the ISO.
+    if first_error.is_none() {
+        if let Err(e) = inputs.file.sync_all() {
+            if inputs.is_regular {
+                first_error = Some(Error::IoError { source: e });
+            }
+            // Non-regular outputs (/dev/null, pipes) always fail
+            // sync_all; that's not a real error.
+        }
+    }
+
+    ConsumerSummary {
+        stats: inputs.map.stats(),
+        error: first_error,
+    }
+}
+
+/// Apply a single `WorkItem`. Returns `Ok(true)` to continue, `Ok(false)`
+/// to break the consumer loop on `Finish`, `Err(_)` on first failure
+/// (caller captures and continues draining).
+fn apply_item(
+    inputs: &mut ConsumerInputs,
+    item: WorkItem,
+    zero: &[u8; ZERO_CHUNK],
+    prog_tx: &SyncSender<ProgressSnapshot>,
+) -> Result<bool> {
+    match item {
+        WorkItem::Good { pos, buf } => {
+            // Decrypt is on the producer; consumer assumes plaintext.
+            let len = buf.len() as u64;
+            inputs
+                .file
+                .seek(SeekFrom::Start(pos))
+                .map_err(|e| Error::IoError { source: e })?;
+            inputs
+                .file
+                .write_all(&buf)
+                .map_err(|e| Error::IoError { source: e })?;
+            inputs
+                .map
+                .record(pos, len, SectorStatus::Finished)
+                .map_err(|e| Error::IoError { source: e })?;
+        }
+        WorkItem::BisectGood { pos, buf } => {
+            inputs
+                .file
+                .seek(SeekFrom::Start(pos))
+                .map_err(|e| Error::IoError { source: e })?;
+            inputs
+                .file
+                .write_all(&buf[..])
+                .map_err(|e| Error::IoError { source: e })?;
+            inputs
+                .map
+                .record(pos, 2048, SectorStatus::Finished)
+                .map_err(|e| Error::IoError { source: e })?;
+        }
+        WorkItem::BisectBad { pos } => {
+            inputs
+                .file
+                .seek(SeekFrom::Start(pos))
+                .map_err(|e| Error::IoError { source: e })?;
+            inputs
+                .file
+                .write_all(&zero[..2048])
+                .map_err(|e| Error::IoError { source: e })?;
+            inputs
+                .map
+                .record(pos, 2048, SectorStatus::NonTrimmed)
+                .map_err(|e| Error::IoError { source: e })?;
+        }
+        WorkItem::SkipFill { pos, len } | WorkItem::GapFill { pos, len } => {
+            inputs
+                .file
+                .seek(SeekFrom::Start(pos))
+                .map_err(|e| Error::IoError { source: e })?;
+            // Subsequent writes are sequential; `crate::io::Writer`'s
+            // seek-elision keeps them on the writeback pipeline path.
+            let mut filled = 0u64;
+            while filled < len {
+                let chunk = (len - filled).min(zero.len() as u64) as usize;
+                inputs
+                    .file
+                    .write_all(&zero[..chunk])
+                    .map_err(|e| Error::IoError { source: e })?;
+                filled += chunk as u64;
+            }
+            inputs
+                .map
+                .record(pos, len, SectorStatus::NonTrimmed)
+                .map_err(|e| Error::IoError { source: e })?;
+        }
+        WorkItem::StatsRequest => {
+            let stats = inputs.map.stats();
+            let bad_ranges = inputs.map.ranges_with(&[
+                SectorStatus::NonTrimmed,
+                SectorStatus::Unreadable,
+                SectorStatus::NonScraped,
+                SectorStatus::NonTried,
+            ]);
+            // Best-effort: drop on backpressure; producer's cache
+            // stays current enough.
+            let _ = prog_tx.try_send(ProgressSnapshot { stats, bad_ranges });
+        }
+        WorkItem::Finish => return Ok(false),
+    }
+    Ok(true)
+}

src/drive/mod.rs

@@ -69,6 +69,16 @@ pub struct Drive {
     halt: Arc<AtomicBool>,
     /// Event handler — fires for read errors and library-level state changes.
     event_fn: Option<Box<dyn Fn(Event) + Send>>,
+    /// Linux only: raw fd for the corresponding block device (`/dev/sr*`)
+    /// used as a recovery fallback when SCSI READ via `/dev/sg*` returns
+    /// an error. The kernel `sr_mod` driver auto-retries failed reads
+    /// (~5× per command) — historically the reason `dd if=/dev/sr0`
+    /// recovers ~50% of bad sectors that single-shot `SG_IO` READ
+    /// misses on the same drive. `None` when the block device couldn't
+    /// be resolved or opened (no fallback in that case; SCSI read
+    /// errors propagate as before).
+    #[cfg(target_os = "linux")]
+    block_dev_fd: Option<std::os::unix::io::RawFd>,
 }
 
 impl Drive {
@@ -87,6 +97,9 @@ impl Drive {
             None => (None, None, None),
         };
 
+        #[cfg(target_os = "linux")]
+        let block_dev_fd = open_block_device_for_sg(device);
+
         Ok(Drive {
             scsi: transport,
             driver,
@@ -96,6 +109,8 @@ impl Drive {
             device_path: device.to_string_lossy().to_string(),
             halt: Arc::new(AtomicBool::new(false)),
             event_fn: None,
+            #[cfg(target_os = "linux")]
+            block_dev_fd,
         })
     }
 
@@ -483,6 +498,54 @@ impl Drive {
                     scsi_status = status,
                     "Drive::read checked_exec failed"
                 );
+
+                // /dev/sr0 pread fallback (Linux only). The kernel
+                // sr_mod driver auto-retries failed reads (~5× per
+                // command). Empirically (BU40N + Dune Part 2 UHD,
+                // 2026-05-08) dd via /dev/sr0 recovers ~50% of bad
+                // sectors that a single-shot SG_IO READ misses.
+                #[cfg(target_os = "linux")]
+                if recovery {
+                    if let Some(fd) = self.block_dev_fd {
+                        let len = count as usize * 2048;
+                        if buf.len() >= len {
+                            let offset = lba as i64 * 2048;
+                            // Drop kernel cache for this region so we get
+                            // a fresh device read, not stale page-cache
+                            // data from a prior successful neighbour read.
+                            let _ = unsafe {
+                                libc::posix_fadvise(
+                                    fd,
+                                    offset,
+                                    len as i64,
+                                    libc::POSIX_FADV_DONTNEED,
+                                )
+                            };
+                            let n = unsafe {
+                                libc::pread(fd, buf.as_mut_ptr() as *mut libc::c_void, len, offset)
+                            };
+                            if n == len as isize {
+                                tracing::info!(
+                                    target: "freemkv::drive",
+                                    lba,
+                                    count,
+                                    bytes = len,
+                                    "Drive::read recovered via /dev/sr0 pread fallback"
+                                );
+                                return Ok(len);
+                            }
+                            tracing::debug!(
+                                target: "freemkv::drive",
+                                lba,
+                                count,
+                                pread_ret = n as i64,
+                                errno = std::io::Error::last_os_error().raw_os_error().unwrap_or(0),
+                                "/dev/sr0 pread fallback also failed"
+                            );
+                        }
+                    }
+                }
+
                 Err(Error::DiscRead {
                     sector: lba as u64,
                     status: Some(status),
@@ -586,6 +649,61 @@ impl Drop for Drive {
     fn drop(&mut self) {
         self.cleanup();
         // SgIoTransport::drop() runs next, calling libc::close(fd)
+        #[cfg(target_os = "linux")]
+        if let Some(fd) = self.block_dev_fd.take() {
+            unsafe { libc::close(fd) };
+        }
+    }
+}
+
+/// Resolve a `/dev/sg*` path to the corresponding `/dev/sr*` block
+/// device by walking sysfs, then open it for read (no `O_DIRECT` —
+/// `posix_fadvise(POSIX_FADV_DONTNEED)` flushes the cache before each
+/// pread, which avoids buffer-alignment requirements while still
+/// forcing fresh device reads).
+///
+/// Returns `None` on any error (sysfs not present, no matching block
+/// device, open failed). Callers treat that as "no fallback available"
+/// and propagate the original SCSI READ error.
+#[cfg(target_os = "linux")]
+fn open_block_device_for_sg(sg_path: &Path) -> Option<std::os::unix::io::RawFd> {
+    let basename = sg_path.file_name()?.to_str()?;
+    if !basename.starts_with("sg") {
+        return None;
+    }
+    let sysfs_dir = format!("/sys/class/scsi_generic/{}/device/block", basename);
+    let entries = std::fs::read_dir(&sysfs_dir).ok()?;
+    let block_name = entries
+        .flatten()
+        .find_map(|e| e.file_name().into_string().ok())?;
+    let block_path = format!("/dev/{}", block_name);
+
+    let mut bytes = block_path.as_bytes().to_vec();
+    bytes.push(0);
+    let fd = unsafe {
+        libc::open(
+            bytes.as_ptr() as *const libc::c_char,
+            libc::O_RDONLY | libc::O_CLOEXEC,
+        )
+    };
+    if fd < 0 {
+        tracing::debug!(
+            target: "freemkv::drive",
+            sg = basename,
+            block_path,
+            errno = std::io::Error::last_os_error().raw_os_error().unwrap_or(0),
+            "Failed to open block device for fallback; sr0 fallback disabled"
+        );
+        None
+    } else {
+        tracing::info!(
+            target: "freemkv::drive",
+            sg = basename,
+            block_path,
+            fd,
+            "Opened /dev/sr* as recovery fallback for failed SCSI reads"
+        );
+        Some(fd)
     }
 }
 

src/io/mod.rs

@@ -0,0 +1,16 @@
+//! File I/O helpers that bound kernel cache pressure on big writes.
+//!
+//! `Writer` is a drop-in wrapper around `std::fs::File` for any call
+//! site that performs large sequential writes (sweep, mux, etc.). It
+//! implements `Write` and `Seek` so existing code paths can swap
+//! `File` for `Writer` with no body changes. Internally it drives a
+//! `WritebackPipeline` that, on Linux, drains dirty pages continuously
+//! at 32 MB granularity to avoid the kernel's accumulate-then-burst
+//! flush behaviour. macOS and Windows use a no-op pipeline — their
+//! default cache policies have not been shown to exhibit the same
+//! pathology for this access pattern.
+
+mod writeback;
+mod writer;
+
+pub(crate) use writer::Writer;

src/io/writeback.rs

@@ -0,0 +1,17 @@
+//! Per-platform writeback pipeline. On Linux, drains dirty pages
+//! continuously at chunk granularity to keep the kernel's writeback
+//! queue bounded. On macOS and Windows, a no-op stub.
+//!
+//! The platform decision lives entirely in this file (the cfg-gated
+//! `pub use` below). Callers — and `DiskWriter` itself — are
+//! platform-independent.
+
+#[cfg(target_os = "linux")]
+mod linux;
+#[cfg(not(target_os = "linux"))]
+mod noop;
+
+#[cfg(target_os = "linux")]
+pub(super) use linux::WritebackPipeline;
+#[cfg(not(target_os = "linux"))]
+pub(super) use noop::WritebackPipeline;

src/io/writeback/linux.rs

@@ -0,0 +1,95 @@
+//! Linux writeback pipeline using `sync_file_range` + `posix_fadvise`.
+//!
+//! Pathology this fixes: the kernel's default `vm.dirty_ratio` (~20 %
+//! of RAM) lets dirty pages accumulate to hundreds of MB during a
+//! big sequential write, then bursts a flush at 99 % disk utilisation.
+//! While the burst runs, app writes block on the writeback queue —
+//! observed empirically as instantaneous speed dropping from ~15 MB/s
+//! to ~1 MB/s every ~30 s during a Pass 1 sweep.
+//!
+//! Strategy: every `chunk_bytes` of new sequential output, kick async
+//! writeback (`SYNC_FILE_RANGE_WRITE`) on the just-completed chunk and
+//! finalise the *previous* chunk via `WAIT_AFTER` + `posix_fadvise
+//! (DONTNEED)`. By the time we finalise, that previous chunk has had
+//! a full chunk's worth of work to flush — the wait is near-instant.
+//! Dirty cache stays bounded at ~2 × `chunk_bytes` and writes drain
+//! continuously instead of in bursts.
+
+use std::fs::File;
+use std::os::unix::io::{AsRawFd, RawFd};
+
+pub(crate) struct WritebackPipeline {
+    fd: RawFd,
+    chunk_bytes: u64,
+    last_flush_pos: u64,
+    pending: Option<(u64, u64)>,
+}
+
+impl WritebackPipeline {
+    pub(crate) fn new(file: &File, start_pos: u64, chunk_bytes: u64) -> Self {
+        Self {
+            fd: file.as_raw_fd(),
+            chunk_bytes,
+            last_flush_pos: start_pos,
+            pending: None,
+        }
+    }
+
+    /// Caller advanced the file position to `pos`. If a chunk boundary
+    /// was crossed, kick async writeback for the just-completed chunk
+    /// and finalise the previous one.
+    pub(crate) fn note_progress(&mut self, pos: u64) {
+        if pos < self.last_flush_pos.saturating_add(self.chunk_bytes) {
+            return;
+        }
+        unsafe {
+            let chunk_off = self.last_flush_pos as i64;
+            let chunk_len = (pos - self.last_flush_pos) as i64;
+            libc::sync_file_range(self.fd, chunk_off, chunk_len, libc::SYNC_FILE_RANGE_WRITE);
+            if let Some((prev_off, prev_len)) = self.pending.take() {
+                libc::sync_file_range(
+                    self.fd,
+                    prev_off as i64,
+                    prev_len as i64,
+                    libc::SYNC_FILE_RANGE_WAIT_AFTER,
+                );
+                libc::posix_fadvise(
+                    self.fd,
+                    prev_off as i64,
+                    prev_len as i64,
+                    libc::POSIX_FADV_DONTNEED,
+                );
+            }
+            self.pending = Some((chunk_off as u64, chunk_len as u64));
+        }
+        self.last_flush_pos = pos;
+    }
+
+    /// Caller is about to seek away from the current write region.
+    /// Drain any in-flight chunk and reset tracking.
+    pub(crate) fn handle_seek(&mut self, new_pos: u64) {
+        self.finalize();
+        self.last_flush_pos = new_pos;
+    }
+
+    /// Drain any in-flight chunk. Idempotent. Call before `sync_all()`
+    /// or when discarding the pipeline.
+    pub(crate) fn finalize(&mut self) {
+        if let Some((prev_off, prev_len)) = self.pending.take() {
+            unsafe {
+                libc::sync_file_range(
+                    self.fd,
+                    prev_off as i64,
+                    prev_len as i64,
+                    libc::SYNC_FILE_RANGE_WAIT_AFTER,
+                );
+                libc::posix_fadvise(
+                    self.fd,
+                    prev_off as i64,
+                    prev_len as i64,
+                    libc::POSIX_FADV_DONTNEED,
+                );
+            }
+        }
+    }
+}

src/io/writeback/noop.rs

@@ -0,0 +1,18 @@
+//! No-op writeback pipeline for non-Linux targets. macOS and Windows
+//! page cache policies have not been shown to exhibit the Linux
+//! accumulate-then-burst flush pathology for our access pattern.
+//! If that changes, replace this stub with a real implementation
+//! (e.g. `F_NOCACHE` on macOS, `FILE_FLAG_WRITE_THROUGH` on Windows).
+
+use std::fs::File;
+
+pub(crate) struct WritebackPipeline;
+
+impl WritebackPipeline {
+    pub(crate) fn new(_file: &File, _start_pos: u64, _chunk_bytes: u64) -> Self {
+        Self
+    }
+    pub(crate) fn note_progress(&mut self, _pos: u64) {}
+    pub(crate) fn handle_seek(&mut self, _new_pos: u64) {}
+    pub(crate) fn finalize(&mut self) {}
+}

src/io/writer.rs

@@ -0,0 +1,84 @@
+//! `Writer` — a drop-in `File` wrapper that keeps the kernel page
+//! cache bounded during large sequential output.
+//!
+//! Implements `Write` and `Seek`, so any call site that uses `File`
+//! through those traits (sweep, mux, patch) can swap to `Writer`
+//! without touching the body of the loop. The wrapper tracks the
+//! current file position and forwards each write to a
+//! [`super::writeback::WritebackPipeline`], which on Linux schedules
+//! incremental `sync_file_range` + `posix_fadvise(DONTNEED)` calls
+//! to drain dirty pages continuously instead of letting the kernel
+//! burst-flush hundreds of MB at a time.
+//!
+//! See `super::writeback::linux` for the pathology and the strategy.
+
+use std::fs::File;
+use std::io::{self, Seek, SeekFrom, Write};
+
+use super::writeback::WritebackPipeline;
+
+const CHUNK_BYTES: u64 = 32 * 1024 * 1024;
+
+pub(crate) struct Writer {
+    file: File,
+    pipeline: WritebackPipeline,
+    pos: u64,
+}
+
+impl Writer {
+    /// Wrap an open `File`. The current OS file position is queried
+    /// once so the pipeline starts tracking from wherever the file
+    /// already is (typically 0 for fresh files; non-zero for resumed
+    /// 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);
+        Ok(Self {
+            file,
+            pipeline,
+            pos,
+        })
+    }
+
+    /// Drain in-flight writeback then issue a full fsync. Use this in
+    /// place of `File::sync_all`.
+    pub(crate) fn sync_all(&mut self) -> io::Result<()> {
+        self.pipeline.finalize();
+        self.file.sync_all()
+    }
+}
+
+impl Write for Writer {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        let n = self.file.write(buf)?;
+        self.pos += n as u64;
+        self.pipeline.note_progress(self.pos);
+        Ok(n)
+    }
+
+    fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
+        self.file.write_all(buf)?;
+        self.pos += buf.len() as u64;
+        self.pipeline.note_progress(self.pos);
+        Ok(())
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.file.flush()
+    }
+}
+
+impl Seek for Writer {
+    fn seek(&mut self, from: SeekFrom) -> io::Result<u64> {
+        let p = self.file.seek(from)?;
+        // Only treat seeks that actually move the position as
+        // boundaries — sweep does a redundant `seek(Current(pos))`
+        // before every write, and we don't want that to drain the
+        // pipeline on every iteration.
+        if p != self.pos {
+            self.pipeline.handle_seek(p);
+            self.pos = p;
+        }
+        Ok(p)
+    }
+}

src/lib.rs

@@ -81,6 +81,7 @@ pub mod error;
 pub mod event;
 pub(crate) mod identity;
 pub(crate) mod ifo;
+pub(crate) mod io;
 pub mod keydb;
 pub(crate) mod labels;
 pub(crate) mod mpls;

src/scsi/mod.rs

@@ -68,10 +68,25 @@ pub(crate) const TUR_TIMEOUT_MS: u32 = 5_000;
 pub(crate) const READ_TIMEOUT_MS: u32 = 10_000;
 
 /// Timeout for content READ commands on the recovery path —
-/// [`disc::Disc::patch`]'s targeted retries on bad ranges. Doubles
-/// the fast-path budget so a sector that fails at 30 s gets one more
-/// honest attempt. Matches sg_dd's default per-command timeout
-/// (`DEF_TIMEOUT = 60000`).
+/// [`disc::Disc::patch`]'s targeted retries on bad ranges. Matches
+/// `sg_dd`'s 60 s ceiling: long enough that any sector the drive can
+/// recover at all gets the time to do so, short enough that an
+/// unresponsive bus is detected before the per-range watchdog fires.
+///
+/// In practice failed reads return in 1–4 s (the drive itself gives up
+/// on uncorrectable ECC before the timeout); the 60 s value is a
+/// safety ceiling, not a steady-state cost.
+///
+/// Historical note (2026-05-08): briefly lowered to 2 s with a 5×
+/// inline retry loop in `Disc::patch` to mimic the kernel `sr_mod`
+/// driver's auto-retry pattern. The synthetic logic worked but on the
+/// live drive each "2 s" read paid ~1.5 s of kernel SCSI mid-layer
+/// error escalation on top, so 5× retries took ~17 s per LBA and
+/// triggered MAX_RANGE_SECS after 4 sectors — pushing recovery to
+/// 0/22 ranges (worse than the 0/22 baseline of v0.17.3 single-shot
+/// at 60 s, since that at least visited every range). Reverted; the
+/// kernel-auto-retry approach is being pursued via a `/dev/sr0` pread
+/// fallback instead.
 pub(crate) const READ_RECOVERY_TIMEOUT_MS: u32 = 60_000;
 
 // ── SCSI status bytes (SPC-4 §4.5.5) ────────────────────────────────────────

tests/scsi_recovery.rs

@@ -24,6 +24,7 @@ fn test_sgio_transport_timeout_does_not_kill_transport() {
         use libfreemkv::scsi::{DataDirection, SCSI_STATUS_TRANSPORT_FAILURE};
         use std::sync::Arc;
         use std::sync::atomic::{AtomicI32, Ordering};
+        use std::time::Duration;
 
         let mut transport = SgIoTransport::open(_path).expect("open device");
         let fd_before = transport.fd;
@@ -93,6 +94,8 @@ fn test_drive_read_per_cdb_timeout_bounds_call() {
 
     #[cfg(target_os = "linux")]
     {
+        use std::time::Duration;
+
         let mut drive = libfreemkv::Drive::open(_path).expect("open drive");
         let timeout_ms: u32 = 5_000;