From 3f486b6aa85c33f0bdb143ff95fa4e99cae4f36f Mon Sep 17 00:00:00 2001 From: Matthew Jackson Date: Sat, 9 May 2026 10:28:14 -0700 Subject: [PATCH] 0.18 round 2: refactor Disc::patch onto Pipeline + PatchSink MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Patch was strictly serial (per-sector recovery: read → seek+write → mapfile.record → next). Lifting the write+record onto a consumer thread lets the drive issue the next per-sector retry while the previous block's recovered bytes are being committed — small but real win on damaged discs with many bad sectors, and uniform with sweep's threading model. - New PatchSink: Sink impl in src/disc/patch.rs. Owns WritebackFile + Mapfile. apply() seeks+writes recovered bytes and records mapfile state per item; close() runs sync_all and mapfile.flush. - Channel depth: WRITE_THROUGH_DEPTH (1). Patch wants minimum buffering — back-pressure should kick in immediately so the drive's per-sector retry budget isn't ahead of the writer. - Disc::patch: keeps every existing recovery decision on the producer (reverse walk, damage-window skip, NOT_READY pauses, bridge-degradation handling, wedge exit, range watchdog). WritebackFile ownership moves to the sink. Behaviour-preserving: per-sector single-shot read budget unchanged (BU40N+Initio bridge wedge concern still respected); recovery algorithm bit-identical. See freemkv-private/memory/0_18_redesign.md. Single contributor: MattJackson. --- src/disc/mod.rs | 294 ++++++++++++++++++-------------- src/disc/patch.rs | 241 ++++++++++++++++++++++++++ src/io/mod.rs | 8 +- src/io/pipeline.rs | 12 +- tests/pass_n_size_aware_skip.rs | 111 ++++++++++++ 5 files changed, 527 insertions(+), 139 deletions(-) create mode 100644 src/disc/patch.rs diff --git a/src/disc/mod.rs b/src/disc/mod.rs index 3af4c11..33484e1 100644 --- a/src/disc/mod.rs +++ b/src/disc/mod.rs @@ -12,6 +12,7 @@ mod bluray; mod dvd; mod encrypt; pub mod mapfile; +mod patch; pub mod read_error; mod sweep_pipeline; @@ -1948,7 +1949,8 @@ impl Disc { path: &std::path::Path, opts: &PatchOpts, ) -> Result { - use std::io::{Seek, SeekFrom, Write}; + use crate::io::pipeline::{Pipeline, WRITE_THROUGH_DEPTH}; + use patch::{PatchItem, PatchSink}; const BRIDGE_DEGRADATION_PAUSE_SECS: u64 = 10; const POST_FAILURE_PAUSE_SECS: u64 = 1; @@ -1967,7 +1969,7 @@ impl Disc { } let mapfile_path = self.mapfile_for(path); - let mut map = + let map = mapfile::Mapfile::load(&mapfile_path).map_err(|e| Error::IoError { source: e })?; let total_bytes = map.total_size(); let keys = if opts.decrypt { @@ -1979,11 +1981,69 @@ impl Disc { let is_regular = std::fs::metadata(path) .map(|m| m.file_type().is_file()) .unwrap_or(false); - // Same bounded-cache `WritebackFile` sweep uses, so patch's - // recovery writes (sparse but can be many across a damaged region) - // get the burst-flush protection on slow / NFS-backed staging. - let mut file = - crate::io::WritebackFile::open(path).map_err(|e| Error::IoError { source: e })?; + + // Snapshot fields we need from the mapfile *before* it moves into + // the consumer thread: bytes_good baseline, total entries, the + // initial `bad_ranges` work list, and the start-of-patch + // diagnostic dump. The shared state (`shared`) republishes these + // throughout the pass; the consumer owns the live `Mapfile`. + let bytes_good_before = map.stats().bytes_good; + let bytes_good_start = bytes_good_before; + let initial_stats = map.stats(); + let initial_entries: Vec<_> = map.entries().to_vec(); + // Every retry pass acts on every non-Finished range. Including + // Unreadable means a sector that failed in pass N gets a fresh + // shot in pass N+1 — drive state evolves, the same read can + // succeed later. Each pass owns its own jumps/skips; if pass 5 + // jumps over the same zone as pass 2, fine. + let mut bad_ranges = map.ranges_with(&[ + mapfile::SectorStatus::NonTrimmed, + mapfile::SectorStatus::NonScraped, + mapfile::SectorStatus::Unreadable, + ]); + if opts.reverse { + bad_ranges.reverse(); + } + let work_total: u64 = bad_ranges.iter().map(|(_, sz)| *sz).sum(); + + // Spawn the consumer. The `WritebackFile` (same bounded-cache + // wrapper sweep uses, so patch's recovery writes — sparse but + // can be many across a damaged region — get the burst-flush + // protection on slow / NFS-backed staging) and the `Mapfile` + // both move into the sink. We hold an `Arc>` snapshot + // the sink republishes after every record so producer-side + // stall guards / progress callbacks can read consumer side- + // effects. + let (sink, shared) = PatchSink::new(path, map, is_regular)?; + // Why: WRITE_THROUGH_DEPTH (=1) — patch reads ONE sector per + // recovery decision and the producer's stall / damage-window + // logic checks consumer-published stats inline. Sweep's + // DEFAULT_PIPELINE_DEPTH (=4) would let several sectors of + // recovered bytes queue up between producer decisions and + // writes, which conflicts with the per-sector lockstep this + // loop was written against. + let pipe = Pipeline::::spawn(WRITE_THROUGH_DEPTH, sink)?; + + // Send a `PatchItem` and translate a `SendError` (consumer + // thread died / panicked) into a useful library error so the + // caller can propagate cleanly. Mirrors `sweep_pipeline.rs`'s + // `send_or_abort`. + let send_or_abort = |pipe: &Pipeline, item: PatchItem| -> Result<()> { + pipe.send(item).map_err(|_| Error::IoError { + source: std::io::Error::other("patch consumer terminated unexpectedly"), + }) + }; + + // Snapshot helper for producer-side stats reads. Holds the + // mutex briefly; we never read across operations so a fresh + // snapshot per call is fine. + let read_shared = |shared: &std::sync::Mutex| -> ( + mapfile::MapStats, + Vec<(u64, u64)>, + ) { + let g = shared.lock().expect("PatchSink shared state mutex poisoned"); + (g.stats, g.bad_ranges.clone()) + }; // Log ISO file size at patch start for write monitoring if let Ok(metadata) = std::fs::metadata(path) { @@ -1998,8 +2058,6 @@ impl Disc { let block_sectors = opts.block_sectors.unwrap_or(1); let recovery = opts.full_recovery; - let bytes_good_before = map.stats().bytes_good; - let bytes_good_start = bytes_good_before; let mut halted = false; let mut wedged_exit = false; let mut blocks_attempted: u64 = 0; @@ -2059,24 +2117,23 @@ impl Disc { tracing::info!( target: "freemkv::disc", phase = "patch_mapfile_snapshot", - total_entries = map.entries().len(), + total_entries = initial_entries.len(), bytes_good_before, - bytes_retryable = map.stats().bytes_retryable, - bytes_unreadable = map.stats().bytes_unreadable, - bytes_nontried = map.stats().bytes_nontried, + bytes_retryable = initial_stats.bytes_retryable, + bytes_unreadable = initial_stats.bytes_unreadable, + bytes_nontried = initial_stats.bytes_nontried, "Mapfile state snapshot at patch start" ); // Log first 10 and last 10 entries for inspection - let entries = map.entries(); - if !entries.is_empty() { + if !initial_entries.is_empty() { tracing::info!( target: "freemkv::disc", phase = "patch_mapfile_entries_start", - num_to_log = (entries.len().min(10)) as u32, + num_to_log = (initial_entries.len().min(10)) as u32, "First 10 entries" ); - for entry in entries.iter().take(10) { + for entry in initial_entries.iter().take(10) { tracing::debug!( target: "freemkv::disc", phase = "patch_mapfile_entry_start", @@ -2087,14 +2144,14 @@ impl Disc { ); } } - if entries.len() > 10 { + if initial_entries.len() > 10 { tracing::info!( target: "freemkv::disc", phase = "patch_mapfile_entries_end", - num_to_log = (entries.len().min(10)) as u32, + num_to_log = (initial_entries.len().min(10)) as u32, "Last 10 entries" ); - for entry in entries.iter().skip(entries.len() - 10) { + for entry in initial_entries.iter().skip(initial_entries.len() - 10) { tracing::debug!( target: "freemkv::disc", phase = "patch_mapfile_entry_end", @@ -2106,21 +2163,6 @@ impl Disc { } } - // Every retry pass acts on every non-Finished range. Including - // Unreadable means a sector that failed in pass N gets a fresh - // shot in pass N+1 — drive state evolves, the same read can - // succeed later. Each pass owns its own jumps/skips; if pass 5 - // jumps over the same zone as pass 2, fine. - let mut bad_ranges = map.ranges_with(&[ - mapfile::SectorStatus::NonTrimmed, - mapfile::SectorStatus::NonScraped, - mapfile::SectorStatus::Unreadable, - ]); - if opts.reverse { - bad_ranges.reverse(); - } - let work_total: u64 = bad_ranges.iter().map(|(_, sz)| *sz).sum(); - tracing::info!( target: "freemkv::disc", phase = "patch_bad_ranges", @@ -2211,7 +2253,7 @@ impl Disc { break; } - let bytes_good_now = map.stats().bytes_good; + let bytes_good_now = read_shared(&shared).0.bytes_good; if bytes_good_now > range_bytes_good { range_bytes_good = bytes_good_now; range_start = std::time::Instant::now(); @@ -2248,21 +2290,23 @@ impl Disc { // patch-marked Unreadable sectors are actually readable. let unmarked_bytes = block_end.saturating_sub(*range_pos); if opts.reverse { - map.record( - *range_pos, - unmarked_bytes, - mapfile::SectorStatus::NonTrimmed, - ) - .map_err(|e| Error::IoError { source: e })?; + send_or_abort( + &pipe, + PatchItem::NonTrimmed { + pos: *range_pos, + len: unmarked_bytes, + }, + )?; } else { let remaining_start = *range_pos + (end - block_end); if remaining_start < end { - map.record( - remaining_start, - end - remaining_start, - mapfile::SectorStatus::NonTrimmed, - ) - .map_err(|e| Error::IoError { source: e })?; + send_or_abort( + &pipe, + PatchItem::NonTrimmed { + pos: remaining_start, + len: end - remaining_start, + }, + )?; } } // Continue to next range (break inner loop only) @@ -2363,8 +2407,6 @@ impl Disc { crate::decrypt::decrypt_sectors(&mut buf[..bytes], &keys, 0)?; } let write_start = std::time::Instant::now(); - file.seek(SeekFrom::Start(pos)) - .map_err(|e| Error::IoError { source: e })?; tracing::debug!( target: "freemkv::disc", phase = "patch_write_start", @@ -2372,8 +2414,18 @@ impl Disc { bytes, "Starting ISO write" ); - file.write_all(&buf[..bytes]) - .map_err(|e| Error::IoError { source: e })?; + // Hand the recovered bytes off to the consumer: + // seek + write + mapfile.record(Finished) all + // happen on the consumer thread, so the producer + // can immediately move on to the next read while + // these bytes are being committed. + send_or_abort( + &pipe, + PatchItem::Recovered { + pos, + buf: buf[..bytes].to_vec(), + }, + )?; let write_duration_ms = write_start.elapsed().as_millis(); tracing::info!( target: "freemkv::disc", @@ -2383,21 +2435,21 @@ impl Disc { write_duration_ms, "ISO write succeeded" ); - let mapfile_record_start = std::time::Instant::now(); - map.record(pos, block_bytes, mapfile::SectorStatus::Finished) - .map_err(|e| Error::IoError { source: e })?; - let mapfile_record_duration_ms = mapfile_record_start.elapsed().as_millis(); tracing::info!( target: "freemkv::disc", phase = "patch_mapfile_record_ok", pos, block_bytes, - mapfile_record_duration_ms, - "Mapfile record written" + "Mapfile record dispatched" ); - // Stall guard: watch bytes_good (real progress), not pos (advances on skips) - let bytes_good_now = map.stats().bytes_good; + // Stall guard: watch bytes_good (real progress), + // not pos (advances on skips). With the consumer + // running in its own thread, this read can lag + // by up to one item; the watchdog operates at + // STALL_SECS=3600 granularity so single-item lag + // is irrelevant. + let bytes_good_now = read_shared(&shared).0.bytes_good; if bytes_good_now > bytes_good_last { stall_start = std::time::Instant::now(); bytes_good_last = bytes_good_now; @@ -2449,25 +2501,23 @@ impl Disc { 0, )?; } - file.seek(SeekFrom::Start(bt_pos)) - .map_err(|e| Error::IoError { source: e })?; - file.write_all(&buf[..bt_bytes]) - .map_err(|e| Error::IoError { source: e })?; - map.record( - bt_pos, - span, - mapfile::SectorStatus::Finished, - ) - .map_err(|e| Error::IoError { source: e })?; + send_or_abort( + &pipe, + PatchItem::Recovered { + pos: bt_pos, + buf: buf[..bt_bytes].to_vec(), + }, + )?; } Err(_err) => { blocks_read_failed += 1; - map.record( - bt_pos, - span, - mapfile::SectorStatus::Unreadable, - ) - .map_err(|e| Error::IoError { source: e })?; + send_or_abort( + &pipe, + PatchItem::Unreadable { + pos: bt_pos, + len: span, + }, + )?; tracing::info!( target: "freemkv::disc", phase = "patch_backtrack_stop", @@ -2601,8 +2651,6 @@ impl Disc { )?; } let write_start = std::time::Instant::now(); - file.seek(SeekFrom::Start(pos)) - .map_err(|e| Error::IoError { source: e })?; tracing::debug!( target: "freemkv::disc", phase = "patch_write_start", @@ -2610,8 +2658,13 @@ impl Disc { bytes, "Starting ISO write" ); - file.write_all(&buf[..bytes]) - .map_err(|e| Error::IoError { source: e })?; + send_or_abort( + &pipe, + PatchItem::Recovered { + pos, + buf: buf[..bytes].to_vec(), + }, + )?; let write_duration_ms = write_start.elapsed().as_millis(); tracing::info!( target: "freemkv::disc", @@ -2621,22 +2674,16 @@ impl Disc { write_duration_ms, "ISO write succeeded" ); - let mapfile_record_start = std::time::Instant::now(); - map.record(pos, block_bytes, mapfile::SectorStatus::Finished) - .map_err(|e| Error::IoError { source: e })?; - let mapfile_record_duration_ms = - mapfile_record_start.elapsed().as_millis(); tracing::info!( target: "freemkv::disc", phase = "patch_mapfile_record_ok", pos, block_bytes, - mapfile_record_duration_ms, - "Mapfile record written" + "Mapfile record dispatched" ); // Stall guard after successful retry - let bytes_good_now = map.stats().bytes_good; + let bytes_good_now = read_shared(&shared).0.bytes_good; if bytes_good_now > bytes_good_last { stall_start = std::time::Instant::now(); bytes_good_last = bytes_good_now; @@ -2656,8 +2703,13 @@ impl Disc { } // All retries exhausted - mark as Unreadable - map.record(pos, block_bytes, mapfile::SectorStatus::Unreadable) - .map_err(|e| Error::IoError { source: e })?; + send_or_abort( + &pipe, + PatchItem::Unreadable { + pos, + len: block_bytes, + }, + )?; damage_window.push(false); if damage_window.len() > PASSN_DAMAGE_WINDOW { @@ -2665,7 +2717,7 @@ impl Disc { } // Stall guard: check on failures too, not just successes - let bytes_good_now = map.stats().bytes_good; + let bytes_good_now = read_shared(&shared).0.bytes_good; if bytes_good_now > bytes_good_last { stall_start = std::time::Instant::now(); bytes_good_last = bytes_good_now; @@ -2770,19 +2822,25 @@ impl Disc { } } - // Log mapfile record for Unreadable status - let mapfile_record_start = std::time::Instant::now(); - map.record(pos, block_bytes, mapfile::SectorStatus::Unreadable) - .map_err(|e| Error::IoError { source: e })?; - let mapfile_record_duration_ms = mapfile_record_start.elapsed().as_millis(); + // Redundant second Unreadable mark — preserved + // bit-for-bit from the pre-split loop (`record` + // is idempotent for same-status replacement of + // the same range). Routes through the consumer + // like every other state change. + send_or_abort( + &pipe, + PatchItem::Unreadable { + pos, + len: block_bytes, + }, + )?; tracing::info!( target: "freemkv::disc", phase = "patch_mapfile_record_unreadable", pos, block_bytes, consecutive_failures, - mapfile_record_duration_ms, - "Mapfile record written as Unreadable" + "Mapfile record dispatched as Unreadable" ); let pause_secs = if err.is_bridge_degradation() { @@ -2896,7 +2954,7 @@ impl Disc { work_done = work_done.saturating_add(block_bytes); if let Some(reporter) = opts.progress { - let s = map.stats(); + let (s, bad_ranges_now) = read_shared(&shared); let kind = if block_sectors == 1 { crate::progress::PassKind::Scrape { reverse: opts.reverse, @@ -2906,16 +2964,10 @@ impl Disc { reverse: opts.reverse, } }; - 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)) + .map(|t| bytes_bad_in_title(t, &bad_ranges_now)) .unwrap_or(0); let main_title = self.titles.first(); let pp = crate::progress::PassProgress { @@ -2939,26 +2991,13 @@ impl Disc { } } - if let Err(e) = file.sync_all() { - if is_regular { - tracing::warn!( - target: "freemkv::disc", - phase = "patch_sync_failed", - error = %e, - os_error = e.raw_os_error(), - error_kind = ?e.kind(), - "patch: sync_all failed" - ); - return Err(Error::IoError { source: e }); - } - tracing::debug!( - target: "freemkv::disc", - phase = "patch_sync_skipped", - error = %e, - "patch: sync_all failed for non-regular file; ignoring" - ); - } - map.flush().map_err(|e| Error::IoError { source: e })?; + // Drain the consumer thread: drop tx, wait for `close` to run + // sync_all + mapfile.flush, then take the final stats from the + // sink's summary. `close` failing on a regular-file sync_all is + // surfaced here as `Error::IoError`, matching pre-split + // behaviour. + let summary = pipe.finish()?; + let stats = summary.stats; // Log final ISO file size for write verification if let Ok(metadata) = std::fs::metadata(path) { @@ -2966,12 +3005,11 @@ impl Disc { target: "freemkv::disc", phase = "patch_iso_size_end", iso_bytes = metadata.len(), - bytes_recovered = map.stats().bytes_good.saturating_sub(bytes_good_before), + bytes_recovered = stats.bytes_good.saturating_sub(bytes_good_before), "ISO file size at patch end" ); } - let stats = map.stats(); tracing::info!( target: "freemkv::disc", phase = "patch_done", diff --git a/src/disc/patch.rs b/src/disc/patch.rs new file mode 100644 index 0000000..a1171d6 --- /dev/null +++ b/src/disc/patch.rs @@ -0,0 +1,241 @@ +//! Producer / consumer split for `Disc::patch`. +//! +//! Background: pre-0.18 patch ran strictly serial — single-sector +//! recovery read → seek + write recovered bytes → mapfile.record → +//! next iteration. The drive sat idle while the previous block's +//! recovered bytes were committed. On a damaged disc with many bad +//! sectors that adds up: per-sector write + mapfile.record costs a +//! handful of milliseconds each, which the drive could be using to +//! issue the next per-sector retry. +//! +//! This module decouples them. A consumer thread owns the +//! [`crate::io::WritebackFile`] (the ISO file) and the +//! [`super::mapfile::Mapfile`]. The producer thread (`Disc::patch`) +//! keeps the [`crate::sector::SectorReader`], the wedge / damage-window +//! state, the per-range watchdog, decrypt — so what enters the channel +//! is already-clean cleartext bytes (or an "Unreadable" terminal mark). +//! +//! Producer and consumer run concurrently; the channel uses +//! [`crate::io::pipeline::WRITE_THROUGH_DEPTH`] (=1) so back-pressure +//! kicks in immediately. We want the drive's per-sector retry budget +//! to stay in lockstep with the writer — sweep's `DEFAULT_PIPELINE_DEPTH` +//! (4) would let several sectors of recovered bytes queue up between +//! the producer's retry decisions and the writer, and patch's recovery +//! loop reads stats (`bytes_good`, range progress) inline to drive its +//! skip / wedge decisions. WRITE_THROUGH_DEPTH gives "read N+1 while +//! writing N", no further pipelining — exactly the model the producer +//! logic was written against. +//! +//! Correctness invariants preserved: +//! - Mapfile is single-writer (consumer-only). No locking on it. +//! - All recovery state (damage window, consecutive_failures, skip +//! escalation, range watchdog) 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), 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. The threading primitive only overlaps the +//! *write* with the *next read*; the per-sector single-shot read +//! budget that the bridge wedge concern was originally about is +//! untouched. +//! +//! Per-range watchdog (`MAX_RANGE_SECS` / `RANGE_BUDGET_CAP_SECS`) +//! checks `bytes_good` for forward progress. With work in flight on +//! the consumer, the producer would otherwise see stale values; the +//! sink publishes a [`SharedPatchState`] snapshot after every record +//! so the producer's stall guards observe consumer side-effects with +//! at most one item of lag (which is fine — the watchdog uses minute- +//! scale budgets, not single-record latency). + +use std::io::{Seek, SeekFrom, Write}; +use std::sync::{Arc, Mutex}; + +use crate::error::{Error, Result}; +use crate::io::pipeline::{Flow, Sink}; + +use super::mapfile::{MapStats, Mapfile, SectorStatus}; + +/// Item the producer hands to the patch consumer. One per per-sector +/// recovery decision. +pub(super) enum PatchItem { + /// Sector / small batch successfully recovered (and decrypted on the + /// producer side if `opts.decrypt` was set). Consumer seeks to + /// `pos`, writes `buf`, records the range as `Finished`. + Recovered { pos: u64, buf: Vec }, + + /// Producer exhausted retries on `[pos, pos+len)`. Consumer records + /// the range as `Unreadable`. No file write — the existing zero-fill + /// from sweep is preserved in place. + Unreadable { pos: u64, len: u64 }, + + /// Producer hit the per-range skip limit and is leaving the + /// remaining bytes as `NonTrimmed` for a future pass. CRITICAL: + /// this is not the same as `Unreadable` — sectors we never tried + /// stay hopeful. (See the comment at the skip-limit branch in + /// `Disc::patch`: ~36% of patch-marked Unreadable sectors are + /// actually readable on a later pass.) No file write. + NonTrimmed { pos: u64, len: u64 }, +} + +/// Mapfile snapshot the sink republishes after every record so the +/// producer can drive its stall / progress logic without holding the +/// mapfile lock for long. `bad_ranges` mirrors what +/// `Mapfile::ranges_with(&[NonTrimmed, Unreadable, NonScraped, NonTried])` +/// would return — same set the pre-split patch loop computed inline +/// for the progress callback. +pub(super) struct SharedPatchState { + pub stats: MapStats, + pub bad_ranges: Vec<(u64, u64)>, +} + +impl SharedPatchState { + fn from_map(map: &Mapfile) -> Self { + Self { + stats: map.stats(), + bad_ranges: map.ranges_with(&[ + SectorStatus::NonTrimmed, + SectorStatus::Unreadable, + SectorStatus::NonScraped, + SectorStatus::NonTried, + ]), + } + } +} + +/// Final summary returned by [`Sink::close`] when the consumer drains +/// cleanly. Mirrors what the pre-split patch loop computed at the end +/// of the function — final mapfile stats plus whether `sync_all` +/// failed on a regular file (the only kind of fsync error patch ever +/// surfaced; `/dev/null` and pipes always fail `sync_all`, that's not +/// a real error). +pub(super) struct PatchSummary { + pub stats: MapStats, +} + +/// Consumer-side of the patch pipeline. Owns the ISO writeback file +/// and the mapfile; publishes a shared snapshot after every record so +/// the producer can read `bytes_good` for stall detection and +/// progress reporting. +pub(super) struct PatchSink { + file: crate::io::WritebackFile, + map: Mapfile, + /// Whether the output is a regular file (so a `sync_all` failure + /// is real). `/dev/null` etc. always fail `sync_all`; ignore those. + is_regular: bool, + /// Snapshot the producer reads. Updated after every successful + /// `record()` call. `Mutex` rather than separate atomics because + /// the producer wants stats + bad_ranges as a coherent pair. + shared: Arc>, +} + +impl PatchSink { + /// Open `path` as a [`crate::io::WritebackFile`] and pair it with + /// `map` for the consumer. The producer holds onto the returned + /// `Arc>` so it can poll mapfile state + /// while the consumer is mutating it. + pub(super) fn new( + path: &std::path::Path, + map: Mapfile, + is_regular: bool, + ) -> Result<(Self, Arc>)> { + let file = crate::io::WritebackFile::open(path).map_err(|e| Error::IoError { source: e })?; + let shared = Arc::new(Mutex::new(SharedPatchState::from_map(&map))); + let shared_clone = shared.clone(); + Ok(( + Self { + file, + map, + is_regular, + shared, + }, + shared_clone, + )) + } + + fn republish(&self) { + // Best-effort lock — only the producer reads, only the consumer + // writes; contention is single-acquire so the lock is never + // poisoned in practice. If it ever did get poisoned we'd want + // the underlying error surfaced rather than silently swallowed, + // so we propagate the poison panic. (Same posture as + // `sweep_pipeline.rs` — it never recovers from a poisoned + // mutex either.) + let mut guard = self.shared.lock().expect("PatchSink shared state mutex poisoned"); + *guard = SharedPatchState::from_map(&self.map); + } +} + +impl Sink for PatchSink { + type Output = PatchSummary; + + fn apply(&mut self, item: PatchItem) -> std::result::Result { + match item { + PatchItem::Recovered { pos, buf } => { + let len = buf.len() as u64; + self.file + .seek(SeekFrom::Start(pos)) + .map_err(|e| Error::IoError { source: e })?; + self.file + .write_all(&buf) + .map_err(|e| Error::IoError { source: e })?; + self.map + .record(pos, len, SectorStatus::Finished) + .map_err(|e| Error::IoError { source: e })?; + } + PatchItem::Unreadable { pos, len } => { + self.map + .record(pos, len, SectorStatus::Unreadable) + .map_err(|e| Error::IoError { source: e })?; + } + PatchItem::NonTrimmed { pos, len } => { + self.map + .record(pos, len, SectorStatus::NonTrimmed) + .map_err(|e| Error::IoError { source: e })?; + } + } + self.republish(); + Ok(Flow::Continue) + } + + fn close(mut self) -> std::result::Result { + // Drain in-flight writeback then issue a full fsync. A failure + // here matters only on regular files — pipes / `/dev/null` etc. + // always fail `sync_all`. + if let Err(e) = self.file.sync_all() { + if self.is_regular { + tracing::warn!( + target: "freemkv::disc", + phase = "patch_sync_failed", + error = %e, + os_error = e.raw_os_error(), + error_kind = ?e.kind(), + "patch: sync_all failed" + ); + return Err(Error::IoError { source: e }); + } + tracing::debug!( + target: "freemkv::disc", + phase = "patch_sync_skipped", + error = %e, + "patch: sync_all failed for non-regular file; ignoring" + ); + } + self.map + .flush() + .map_err(|e| Error::IoError { source: e })?; + // Final republish so anyone reading the shared snapshot after + // `Pipeline::finish` sees the post-flush state. (The producer + // already has its own copy of the final `MapStats` in the + // returned `PatchSummary`, but the snapshot is part of the + // public-ish contract of the consumer: it stays current + // through close.) + self.republish(); + Ok(PatchSummary { + stats: self.map.stats(), + }) + } +} diff --git a/src/io/mod.rs b/src/io/mod.rs index a119dbd..8ff9943 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -21,9 +21,9 @@ pub mod pipeline; pub(crate) use writeback_file::WritebackFile; -// Re-exports for the 0.18 redesign. Currently flagged unused because -// no in-tree call site has been migrated yet (sweep is still on -// `disc/sweep_pipeline.rs`; patch and mux have no pipeline). The next -// 0.18 slice removes this allow as it wires up the first consumer. +// Re-exports for the 0.18 redesign. `Disc::patch` (0.18 round 2) +// uses `Pipeline` + `Sink` + `Flow` + `WRITE_THROUGH_DEPTH`. Sweep +// and mux still need to be migrated; until they are, +// `DEFAULT_PIPELINE_DEPTH` is unused outside tests. #[allow(unused_imports)] pub use pipeline::{DEFAULT_PIPELINE_DEPTH, Flow, Pipeline, Sink, WRITE_THROUGH_DEPTH}; diff --git a/src/io/pipeline.rs b/src/io/pipeline.rs index 0c295b0..74002fd 100644 --- a/src/io/pipeline.rs +++ b/src/io/pipeline.rs @@ -28,13 +28,11 @@ //! //! ## Dead-code suppression //! -//! The `Pipeline` / `Sink` / `Flow` / `DEFAULT_PIPELINE_DEPTH` / -//! `WRITE_THROUGH_DEPTH` items are crate-internal API today (the -//! parent `io` module is `pub(crate)`) but have no in-tree callers -//! in this slice — sweep is still on `disc/sweep_pipeline.rs`, patch -//! and mux still have no pipeline at all. Wiring them up is the -//! next slice of the 0.18 redesign. The `#[allow]` below is removed -//! once any of those three call sites lands on this primitive. +//! `Disc::patch` is the first in-tree caller (0.18 round 2). Sweep +//! and mux still have their own bespoke loops; once they land on +//! `Pipeline` too, the `#[allow]` below can drop. Until then it +//! covers the constants and trait machinery the patch caller doesn't +//! exercise (e.g. `DEFAULT_PIPELINE_DEPTH`). #![allow(dead_code)] diff --git a/tests/pass_n_size_aware_skip.rs b/tests/pass_n_size_aware_skip.rs index 423f1de..a1f33dd 100644 --- a/tests/pass_n_size_aware_skip.rs +++ b/tests/pass_n_size_aware_skip.rs @@ -286,3 +286,114 @@ fn patch_recovers_multiple_good_middles() { pr.bytes_total, ); } + +/// 0.18 Pass N pipeline split: exercises the new producer/consumer +/// path end-to-end on a synthetic patterned reader. Bad range layout +/// is small (5 bad LBAs surrounded by good middle) so the producer +/// emits a mix of `Recovered` and `Unreadable` items and the consumer +/// thread must apply both kinds. Verifies: +/// +/// - `bytes_good` advances (good sectors flow producer→consumer→file +/// →mapfile with the data preserved). +/// - The recovered LBAs end up Finished; the bad LBAs end up Unreadable. +/// - Bytes written at the recovered offsets match what the producer +/// read from the patterned source (proves the channel hand-off +/// didn't drop or reorder buffers, and the consumer's seek+write +/// landed at the right offsets). +#[test] +fn patch_pipeline_split_recovers_and_records_correctly() { + let capacity_sectors: u32 = 512; + let total_bytes: u64 = capacity_sectors as u64 * SECTOR_SIZE as u64; + + // Layout: LBAs 200-204 inclusive are bad (5 sectors), 205-249 good. + // The pre-existing range is LBAs 200-249 NonTrimmed (100 KB). + let mut bad_lbas = HashSet::new(); + for lba in 200..205 { + bad_lbas.insert(lba); + } + + let (mut reader, _trace) = PatternedSectorReader::new(capacity_sectors, bad_lbas.clone()); + let disc = synthetic_disc(capacity_sectors); + + let tmp = tempfile::NamedTempFile::new().unwrap(); + let iso_path = tmp.path().to_path_buf(); + drop(tmp); + + let finished = [ + (0, 200 * 2048), + (250 * 2048, (capacity_sectors as u64 - 250) * 2048), + ]; + let nontrimmed = [(200 * 2048, 50 * 2048)]; + prep_iso_and_mapfile(&iso_path, total_bytes, &finished, &nontrimmed); + + let opts = CopyOptions { + decrypt: false, + multipass: true, + ..Default::default() + }; + let pr = disc + .copy(&mut reader, &iso_path, &opts) + .expect("copy returns Ok"); + + // Bytes_good_total should advance — the good LBAs in the bad range + // (205-249, 45 sectors) are all reachable via per-sector retry. + // Initial bytes_good = 200 * 2048 + (512-250) * 2048 = 462 sectors. + // After patch, bytes_good should be ≥ 462 + 45 = 507 sectors worth. + let initial_good_sectors: u64 = 200 + (capacity_sectors as u64 - 250); + let min_expected_good_bytes = (initial_good_sectors + 30) * 2048; + assert!( + pr.bytes_good >= min_expected_good_bytes, + "patch should have recovered most good LBAs in the bad range via the pipeline. \ + bytes_good={} (expected ≥ {}); bytes_total={}", + pr.bytes_good, + min_expected_good_bytes, + pr.bytes_total, + ); + + // Verify the mapfile records: every good LBA is Finished, every + // bad LBA is either Unreadable or NonTrimmed (not Finished). + let map_path = libfreemkv::disc::mapfile_path_for(&iso_path); + let map = Mapfile::load(&map_path).unwrap(); + let finished_ranges = map.ranges_with(&[SectorStatus::Finished]); + let in_finished = |lba: u32| -> bool { + let pos = lba as u64 * 2048; + finished_ranges + .iter() + .any(|&(p, sz)| pos >= p && pos < p + sz) + }; + + for lba in 205..250 { + assert!( + in_finished(lba), + "good LBA {lba} should be Finished after pipeline patch run" + ); + } + for lba in 200..205 { + assert!( + !in_finished(lba), + "bad LBA {lba} should NOT be Finished after pipeline patch run" + ); + } + + // Verify the consumer wrote the producer's bytes at the right + // offsets. PatternedSectorReader fills each sector with `(lba & 0xff) + // as u8` — picking LBA 220 (well inside the recovered region) gives + // a clean signature byte to check. + use std::io::{Read, Seek, SeekFrom}; + let mut iso = std::fs::File::open(&iso_path).unwrap(); + iso.seek(SeekFrom::Start(220 * 2048)).unwrap(); + let mut sector = [0u8; 2048]; + iso.read_exact(&mut sector).unwrap(); + let expected_byte = (220u32 & 0xff) as u8; + + let _ = std::fs::remove_file(&iso_path); + let _ = std::fs::remove_file(&map_path); + + assert!( + sector.iter().all(|&b| b == expected_byte), + "consumer should have written PatternedSectorReader's pattern \ + (byte {expected_byte:#x} for LBA 220) to the recovered offset; \ + got first 8 bytes = {:?}", + §or[..8] + ); +}