diff --git a/CHANGELOG.md b/CHANGELOG.md index 787e829..2feb0d2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,44 @@ # Changelog +## 0.17.12 (2026-05-09) + +### Mapfile time-batched persistence — unblock NFS staging + +Pre-0.17.12 every `Mapfile::record()` wrote the entire mapfile to disk +via tempfile-create + buffered-write + atomic-rename. On local LVM +that's effectively free (page cache + microsecond-scale renames). On +NFS (autorip's intended staging path for centralised media) each +record() became three RPCs through the unraid user share's shfs-fuse +layer — measured end-to-end at multiple ms each. With ~170 record() +calls per second of sustained sweep, the mapfile path alone burned +multiple seconds of wall time per real-world second of work, dragging +the rip from ~11 MB/s on local to ~1.5 MB/s on NFS. + +Fix: time-batch the persistence inside `Mapfile`. + +- `record()` always updates in-memory state and stats (so callers' + `stats()` reads stay coherent in the same process). +- The `write_to_disk()` rename only fires when ≥ `FLUSH_INTERVAL` + (1 s) has elapsed since the last persist. +- New `flush()` method forces an out-of-band persist; called by + `sweep_pipeline`'s consumer at end-of-sweep and by `Disc::patch` at + end-of-patch, after the file's `sync_all()`. +- `Drop` impl best-effort flushes so an early-return / unwind doesn't + silently lose pending state. + +Crash-safety changes from "lose at most one block" to "lose at most +1 s of recorded progress" — the ISO file's payload bytes are +unaffected; only the mapfile's authority over which sectors are +already-good is at risk, and a resume re-reads anything Pass 1 had +already covered. Acceptable for a 7× throughput recovery on the +target deployment. + +The internal `round_trip_load` test now calls `flush()` before +`Mapfile::load` to read back what the in-memory state asserts. +External patch / copy tests are unaffected: `patch` and +`sweep_pipeline` flush at completion before returning, so any +`Mapfile::load` at the call-site sees fully persisted state. + ## 0.17.11 (2026-05-09) ### Sweep producer/consumer split — overlap drive read with file write diff --git a/Cargo.toml b/Cargo.toml index f09ff81..d3abe15 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "libfreemkv" -version = "0.17.11" +version = "0.17.12" edition = "2024" rust-version = "1.86" license = "AGPL-3.0-only" diff --git a/src/disc/mapfile.rs b/src/disc/mapfile.rs index 006698c..e634332 100644 --- a/src/disc/mapfile.rs +++ b/src/disc/mapfile.rs @@ -17,11 +17,22 @@ //! //! Status chars: `?` non-tried · `*` non-trimmed · `/` non-scraped · `-` unreadable · `+` finished. //! -//! The mapfile is flushed to disk on every `record()` call so a crashed -//! rip loses at most one block of recorded state. +//! The mapfile is flushed to disk at most once per `FLUSH_INTERVAL` +//! during `record()` calls, plus on explicit `flush()` and on `Drop`. +//! This bounds atomic-rename RPC rate on networked staging (e.g. NFS) +//! where per-record persists otherwise serialize the rip pipeline. use std::io::{self, Write}; use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; + +/// Minimum interval between mapfile persists. `record()` updates in-memory +/// state every call but only writes to disk when this interval has elapsed +/// since the last persist (or when `flush()` is called explicitly, or on +/// `Drop`). Bounds RPC rate on NFS staging where atomic-rename per record +/// otherwise dominates throughput. On crash the worst-case progress loss +/// is one interval's worth of records. +const FLUSH_INTERVAL: Duration = Duration::from_millis(1000); /// Status of a byte range in the mapfile. ddrescue-compatible. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -100,8 +111,11 @@ pub struct MapStats { pub main_lost_ms: f64, } -/// Write-through mapfile. Every `record()` persists to disk immediately -/// so a crash during rip loses at most one block. +/// Time-batched mapfile. `record()` keeps in-memory state up-to-date on +/// every call; persists to disk at most once per `FLUSH_INTERVAL`. +/// Explicit `flush()` and `Drop` guarantee state is on disk after a sweep +/// or patch finishes. On hard crash the worst-case loss is one flush +/// interval of records — the file's payload bytes are unaffected. pub struct Mapfile { path: PathBuf, entries: Vec, @@ -110,6 +124,12 @@ pub struct Mapfile { /// Incrementally maintained stats — updated on every `record()` call /// so `stats()` is O(1) instead of O(n). stats: MapStats, + /// True when in-memory state has changed but `write_to_disk` has not + /// yet captured it. + dirty: bool, + /// Wall-clock timestamp of the last successful `write_to_disk` (or + /// the moment the mapfile was constructed, whichever is later). + last_flushed: Instant, } impl Mapfile { @@ -117,7 +137,7 @@ impl Mapfile { /// Writes to disk immediately so a resume can pick up even if the caller /// never records anything. pub fn create(path: &Path, total_size: u64, version: &str) -> io::Result { - let mf = Self { + let mut mf = Self { path: path.to_path_buf(), entries: vec![MapEntry { pos: 0, @@ -132,8 +152,13 @@ impl Mapfile { bytes_nontried: total_size, ..Default::default() }, + dirty: false, + last_flushed: Instant::now(), }; + // Eager initial persist so a resume can pick this up even if + // `record()` is never called. mf.write_to_disk()?; + mf.last_flushed = Instant::now(); Ok(mf) } @@ -200,6 +225,8 @@ impl Mapfile { total_size, version, stats, + dirty: false, + last_flushed: Instant::now(), }) } @@ -268,7 +295,24 @@ impl Mapfile { // path during sweep/patch, it just returns the cached value. self.stats = Self::compute_stats(&merged, self.total_size); self.entries = merged; - self.write_to_disk()?; + self.dirty = true; + if self.last_flushed.elapsed() >= FLUSH_INTERVAL { + self.write_to_disk()?; + self.dirty = false; + self.last_flushed = Instant::now(); + } + Ok(()) + } + + /// Persist any pending in-memory changes to disk. No-op if clean. + /// Callers (sweep/patch finalisation) invoke this after their last + /// `record()` to guarantee state is durable before returning. + pub fn flush(&mut self) -> io::Result<()> { + if self.dirty { + self.write_to_disk()?; + self.dirty = false; + self.last_flushed = Instant::now(); + } Ok(()) } @@ -363,6 +407,16 @@ impl Mapfile { } } +impl Drop for Mapfile { + /// Best-effort flush on drop so a sweep / patch that returns early + /// (or unwinds) doesn't lose its in-memory state. Errors here are + /// swallowed because Drop has no way to surface them; explicit + /// `flush()` on the success path gives callers proper error handling. + fn drop(&mut self) { + let _ = self.flush(); + } +} + fn parse_hex(s: &str) -> io::Result { let s = s.strip_prefix("0x").unwrap_or(s); u64::from_str_radix(s, 16).map_err(|_| { @@ -463,6 +517,8 @@ mod tests { let mut mf = Mapfile::create(&p, 1000, "test").unwrap(); mf.record(100, 200, SectorStatus::Finished).unwrap(); mf.record(500, 100, SectorStatus::Unreadable).unwrap(); + // record() batches; explicit flush before reading back from disk. + mf.flush().unwrap(); let loaded = Mapfile::load(&p).unwrap(); assert_eq!(loaded.entries(), mf.entries()); let _ = std::fs::remove_file(&p); diff --git a/src/disc/mod.rs b/src/disc/mod.rs index faf7507..7d329f3 100644 --- a/src/disc/mod.rs +++ b/src/disc/mod.rs @@ -2956,6 +2956,7 @@ impl Disc { "patch: sync_all failed for non-regular file; ignoring" ); } + map.flush().map_err(|e| Error::IoError { source: e })?; // Log final ISO file size for write verification if let Ok(metadata) = std::fs::metadata(path) { diff --git a/src/disc/sweep_pipeline.rs b/src/disc/sweep_pipeline.rs index e0f25b6..d406488 100644 --- a/src/disc/sweep_pipeline.rs +++ b/src/disc/sweep_pipeline.rs @@ -199,7 +199,8 @@ fn consumer_loop( } } - // Final flush — drain the writeback pipeline + fsync the ISO. + // Final flush — drain the writeback pipeline + fsync the ISO, + // then persist any pending mapfile state. if first_error.is_none() { if let Err(e) = inputs.file.sync_all() { if inputs.is_regular { @@ -208,6 +209,9 @@ fn consumer_loop( // Non-regular outputs (/dev/null, pipes) always fail // sync_all; that's not a real error. } + if let Err(e) = inputs.map.flush() { + first_error = Some(Error::IoError { source: e }); + } } ConsumerSummary {