Merge branch 'audit-lib-G2b' into audit-fix-lib

src/dirimage/layout.rs

@@ -216,7 +216,14 @@ fn walk(dir: &Path, disc_path: &str, depth: u32, entries: &mut usize) -> Result<
                 // A broken symlink or a file gone between readdir and stat: skip it
                 // rather than plan an unreadable extent. Anything else (e.g. perm
                 // denied) is NOT a missing file, so skipping it would drop it silently.
-                Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
+                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
+                    tracing::warn!(
+                        target: "freemkv::dirimage",
+                        path = %child_path.escape_debug(),
+                        "entry is a dangling link or vanished mid-walk; omitted from the image"
+                    );
+                    continue;
+                }
                 Err(e) => return Err(Error::from(e)),
             };
             if !meta.is_file() {
@@ -676,7 +683,7 @@ pub(super) fn flatten<'a>(dir: &'a DirNode, out: &mut Vec<&'a FileNode>) {
     }
 }
 
-/// Metadata footprint in blocks, for diagnostics.
+/// Metadata footprint in blocks; `plan` bounds it against `MAX_META_BYTES`.
 pub(super) fn metadata_block_count(root: &DirNode) -> u64 {
     2 + metadata_blocks(root)
 }
@@ -907,6 +914,26 @@ mod tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    // `walk` counts every entry against `MAX_ENTRIES`: the entry that would pass it is refused.
+    #[test]
+    fn the_entry_cap_refuses_the_entry_past_max_entries() {
+        let dir = std::env::temp_dir().join(format!("fmkv-entries-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).unwrap();
+        std::fs::write(dir.join("a.bin"), b"x").unwrap();
+        let mut at_cap = MAX_ENTRIES - 1;
+        assert!(
+            walk(&dir, "/", 0, &mut at_cap).is_ok(),
+            "the last allowed entry"
+        );
+        let mut past = MAX_ENTRIES;
+        assert!(matches!(
+            walk(&dir, "/", 0, &mut past),
+            Err(Error::DirImageTooLarge)
+        ));
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
     // The nesting cap (`MAX_DEPTH`) refuses a tree deeper than the planner
     // represents, before it recurses without bound. Executes `walk`'s depth
     // guard on a real directory chain, not the constant.

src/dirimage/tests.rs

@@ -201,6 +201,59 @@ fn a_file_past_the_ad_ceiling_reads_back_as_multiple_extents() {
     );
 }
 
+// Past the 30-bit AD ceiling, each extent reads its own bytes of the file: a read at an
+// extent's start, one sector in, and a batch crossing the extent boundary.
+#[test]
+fn a_multi_extent_file_reads_each_extents_own_bytes() {
+    use std::io::Seek;
+    let s = Scratch::new("bigdata");
+    s.file("BDMV/index.bdmv", &pattern(1, 16));
+    s.dir("BDMV/STREAM");
+    let big = s.path().join("BDMV/STREAM/00000.m2ts");
+    let ad = super::layout::MAX_AD_BYTES;
+    let size = ad + 3 * SECTOR as u64;
+    let mut f = std::fs::File::create(&big).unwrap();
+    f.set_len(size).unwrap();
+    for (at, b) in [
+        (0, 0xA1u8),
+        (ad - SECTOR as u64, 0xB2),
+        (ad, 0xC3),
+        (ad + SECTOR as u64, 0xD4),
+    ] {
+        f.seek(SeekFrom::Start(at)).unwrap();
+        f.write_all(&[b]).unwrap();
+    }
+    drop(f);
+
+    let mut img = DirImage::open(s.path()).unwrap();
+    let fs = udf::read_filesystem(&mut img).unwrap();
+    let exts = fs
+        .file_extents(&mut img, "/BDMV/STREAM/00000.m2ts")
+        .unwrap();
+    assert_eq!(exts.len(), 2);
+    let (first, second) = (exts[0].0, exts[1].0);
+    let head = |img: &mut DirImage, lba: u32, n: u16| {
+        let mut buf = vec![0u8; n as usize * SECTOR];
+        img.read_sectors(lba, n, &mut buf, false).unwrap();
+        (0..n as usize)
+            .map(|i| buf[i * SECTOR])
+            .collect::<Vec<u8>>()
+    };
+    assert_eq!(head(&mut img, first, 1), [0xA1]);
+    assert_eq!(head(&mut img, second, 1), [0xC3]);
+    assert_eq!(
+        head(&mut img, second + 1, 1),
+        [0xD4],
+        "one sector into the extent"
+    );
+    let last = first + exts[0].1 - 1;
+    assert_eq!(
+        head(&mut img, last, 3),
+        [0xB2, 0xC3, 0xD4],
+        "a batch crossing the extent boundary"
+    );
+}
+
 /// An IN-RANGE gap (a sector below capacity that no extent covers) reads as
 /// zeros, not an error — a real image has unrecorded sectors too, and
 /// `read_filesystem` probes fixed LBAs (256, the VDS window) before it knows
@@ -1197,6 +1250,47 @@ fn a_skipped_symlink_does_not_collide_with_a_same_name_file() {
     DirImage::open(s.path()).expect("dangling symlink is omitted, not a collision");
 }
 
+/// A tree whose metadata exceeds the in-memory ceiling is refused at plan time.
+#[test]
+fn an_oversized_metadata_footprint_is_refused_at_plan_time() {
+    let (s, _, _) = bdmv_scratch();
+    for dir in ["BDMV/STREAM", "BDMV/CLIPINF"] {
+        for i in 0..17_000 {
+            s.file(&format!("{dir}/{i:05}"), b"");
+        }
+    }
+    let err = DirImage::open(s.path()).unwrap_err();
+    assert_eq!(err.code(), crate::error::E_DIR_IMAGE_TOO_LARGE, "got {err}");
+}
+
+/// A symlink to a directory is omitted, and does not take part in the collision check.
+#[cfg(unix)]
+#[test]
+fn a_directory_symlink_is_omitted_from_the_image() {
+    let (s, _, _) = bdmv_scratch();
+    s.file("BDMV/STREAM/A.M2TS", b"x");
+    s.dir("elsewhere");
+    std::os::unix::fs::symlink(
+        s.path().join("elsewhere"),
+        s.path().join("BDMV/STREAM/ A.M2TS"),
+    )
+    .unwrap();
+    let mut img = DirImage::open(s.path()).expect("a directory link is omitted, not a collision");
+    let fs = udf::read_filesystem(&mut img).unwrap();
+    let names: Vec<_> = fs
+        .find_dir("/BDMV/STREAM")
+        .unwrap()
+        .entries
+        .iter()
+        .map(|e| e.name.clone())
+        .collect();
+    assert_eq!(
+        names.iter().filter(|n| n.trim() == "A.M2TS").count(),
+        1,
+        "{names:?}"
+    );
+}
+
 /// A directory whose FID list exceeds the UDF reader's cap is refused at plan time, by name.
 #[test]
 fn an_over_wide_directory_is_refused_at_plan_time() {

src/io/block_sink.rs

@@ -128,9 +128,8 @@ impl BlockSink for NullBlockSink {
 }
 
 /// Open the block output `url`: `iso://<path>` (an image, created now) or `null://`.
-/// `dir://` is the folder output of a whole-disc extraction ([`crate::Disc::extract_tree`]);
-/// a PES output scheme is refused with [`Error::StreamWriteOnly`]'s sibling
-/// [`Error::StreamUrlInvalid`].
+/// `dir://` is the folder output of a whole-disc extraction ([`crate::Disc::extract_tree`])
+/// and is refused here, as is a PES output scheme, with [`Error::StreamUrlInvalid`].
 pub fn open_block_sink(url: &str) -> Result<Box<dyn BlockSink>> {
     match crate::mux::parse_url(url) {
         crate::mux::StreamUrl::Iso { path } => {

src/io/file_sector_source/macos.rs

@@ -34,6 +34,7 @@ pub(crate) fn prefetch(file: &File, offset: u64, len: u64) {
         ra_count: bytes as libc::c_int,
     };
     // Best-effort — kernel hint only.
+    // SAFETY: `file` is a live borrowed fd and `ra` a live stack radvisory for the call.
     unsafe {
         libc::fcntl(file.as_raw_fd(), libc::F_RDADVISE, &mut ra);
     }

src/io/file_sector_source/mod.rs

@@ -36,8 +36,8 @@ use crate::sector::SectorSource;
 
 use crate::consts::{SECTOR_BYTES, SECTOR_BYTES_U64};
 
-// Bytes-read threshold per `posix_fadvise(DONTNEED)` drop; mirrors WRITEBACK_CHUNK_BYTES. 32
-// MiB is empirically tuned (7200rpm HDD/SATA).
+// Bytes-read threshold per `posix_fadvise(DONTNEED)` drop; mirrors WRITEBACK_CHUNK_BYTES_DEFAULT.
+// 32 MiB is empirically tuned (7200rpm HDD/SATA).
 const READ_DROP_CHUNK_BYTES_DEFAULT: u64 = 32 * 1024 * 1024;
 
 // Max MiB from FREEMKV_READ_DROP_CHUNK_MIB: 64 GiB, and small enough n*1024*1024 can't overflow
@@ -661,4 +661,52 @@ mod tests {
         );
         let _ = std::fs::remove_file(&path);
     }
+
+    // A padded read of one sector from a longer file takes only that sector, and an lba past
+    // the file's end reads as zeros.
+    #[test]
+    fn open_padded_clamps_each_read_to_its_own_sectors() {
+        let td = tempfile::tempdir().unwrap();
+        let path = td.path().join("p.bin");
+        std::fs::write(&path, vec![7u8; 3 * 2048 + 100]).unwrap();
+        let mut s = FileSectorSource::open_padded(&path).unwrap();
+        let mut buf = vec![0xAAu8; 2048];
+        assert_eq!(s.read_sectors(1, 1, &mut buf, false).unwrap(), 2048);
+        assert!(buf.iter().all(|&b| b == 7));
+        assert_eq!(s.read_sectors(9, 1, &mut buf, false).unwrap(), 2048);
+        assert!(buf.iter().all(|&b| b == 0), "past EOF is the synthetic pad");
+    }
+
+    // The 32-bit LBA ceiling: u32::MAX sectors opens, one more is IsoTooLarge, for both
+    // open paths. Sparse files; skipped where the filesystem refuses the length.
+    #[test]
+    fn open_refuses_a_file_past_the_lba_space() {
+        let td = tempfile::tempdir().unwrap();
+        let path = td.path().join("huge.bin");
+        let f = std::fs::File::create(&path).unwrap();
+        let max = u32::MAX as u64 * SECTOR_BYTES_U64;
+        if f.set_len(max).is_err() {
+            return;
+        }
+        assert_eq!(
+            FileSectorSource::open(&path).unwrap().capacity_sectors(),
+            u32::MAX
+        );
+        assert_eq!(
+            FileSectorSource::open_padded(&path)
+                .unwrap()
+                .capacity_sectors(),
+            u32::MAX
+        );
+        f.set_len(max + SECTOR_BYTES_U64).unwrap();
+        assert!(matches!(
+            FileSectorSource::open(&path),
+            Err(Error::IsoTooLarge { .. })
+        ));
+        f.set_len(max + 1).unwrap();
+        assert!(matches!(
+            FileSectorSource::open_padded(&path),
+            Err(Error::IsoTooLarge { .. })
+        ));
+    }
 }

src/io/flush/tests.rs

@@ -705,6 +705,107 @@ fn durable_sync_file_stall_and_stop() {
     assert!(t.elapsed() < SLACK, "{:?}", t.elapsed());
 }
 
+// A flush whose primitives fail or whose counter jumps by `step` per sample.
+struct ScriptedOps {
+    range_err: bool,
+    finish_err: bool,
+    finish_sleep: Duration,
+    step: AtomicU64,
+}
+
+impl ScriptedOps {
+    fn new() -> Self {
+        Self {
+            range_err: false,
+            finish_err: false,
+            finish_sleep: Duration::ZERO,
+            step: AtomicU64::new(0),
+        }
+    }
+}
+
+impl FlushOps for ScriptedOps {
+    fn chunk(&self, _: &File) -> io::Result<()> {
+        Ok(())
+    }
+    fn range(&self, _: &File, _: u64, _: u64) -> Option<io::Result<()>> {
+        self.range_err.then(|| Err(io::Error::from_raw_os_error(5)))
+    }
+    fn finish(&self, _: &File) -> io::Result<()> {
+        std::thread::sleep(self.finish_sleep);
+        if self.finish_err {
+            return Err(io::Error::from_raw_os_error(5));
+        }
+        Ok(())
+    }
+    fn sample(&self) -> Option<u64> {
+        (self.step.load(Ordering::SeqCst) > 0)
+            .then(|| self.step.fetch_add(1 << 40, Ordering::SeqCst))
+    }
+}
+
+/// A failed range sync or final flush is the call's error, never a quiet success.
+#[test]
+fn durable_sync_file_surfaces_range_and_finish_errors() {
+    let (_d, f) = sized_file(8 * K);
+    let t = durable_timing(Duration::from_secs(30));
+    let range = ScriptedOps {
+        range_err: true,
+        ..ScriptedOps::new()
+    };
+    let e = durable_sync_file_with(&f, None, |_, _| {}, Arc::new(range), t)
+        .expect_err("a failed range sync fails");
+    assert_eq!(e.raw_os_error(), Some(5), "{e}");
+    let finish = ScriptedOps {
+        finish_err: true,
+        ..ScriptedOps::new()
+    };
+    let e = durable_sync_file_with(&f, None, |_, _| {}, Arc::new(finish), t)
+        .expect_err("a failed final flush fails");
+    assert_eq!(e.raw_os_error(), Some(5), "{e}");
+}
+
+/// A sampled counter past the file length (other writers on the mount) never reports over
+/// 100% and still ends with the final `(len, len)` report.
+#[test]
+fn durable_sync_file_clamps_sampled_progress_below_the_length() {
+    let ops = ScriptedOps {
+        finish_sleep: W * 3,
+        step: AtomicU64::new(1),
+        ..ScriptedOps::new()
+    };
+    let (_d, f) = sized_file(8 * K);
+    let mut seen = Vec::new();
+    durable_sync_file_with(
+        &f,
+        None,
+        |d, n| seen.push((d, n)),
+        Arc::new(ops),
+        durable_timing(W),
+    )
+    .unwrap();
+    assert!(seen.len() >= 2, "sampled while blocked: {seen:?}");
+    assert!(seen.iter().all(|(d, n)| d <= n), "over 100%: {seen:?}");
+    assert_eq!(seen.last(), Some(&(8 * K, 8 * K)));
+}
+
+/// An empty file still gets its one completion report.
+#[test]
+fn durable_sync_file_reports_an_empty_file_once() {
+    let (_d, f) = sized_file(0);
+    let mut seen = Vec::new();
+    let t = durable_timing(W);
+    durable_sync_file_with(
+        &f,
+        None,
+        |d, n| seen.push((d, n)),
+        Arc::new(ScriptedOps::new()),
+        t,
+    )
+    .unwrap();
+    assert_eq!(seen, [(0, 0)]);
+}
+
 const MOUNTSTATS: &str = "\
 device rootfs mounted on / with fstype rootfs
 device nas:/export mounted on /mnt/nas with fstype nfs4 statvers=1.1

src/io/image_writer.rs

@@ -339,6 +339,7 @@ mod tests {
         let err =
             write_image(&mut src, &dest, 100_000, &Ctx::new(halt.clone())).expect_err("must halt");
         assert!(matches!(err, Error::Halted), "got {err:?}");
+        assert!(dest.exists(), "a halt must keep the partial image");
         let _ = std::fs::remove_file(&dest);
     }
 

src/io/pipeline.rs

@@ -40,8 +40,8 @@ impl Default for JoinTiming {
     }
 }
 
-// Cached FREEMKV_DEBUG=1 lookup — called per item on the mux hot loop,
-// so the env lock is paid once, not per call.
+// Cached FREEMKV_DEBUG lookup ("1", "true" or "yes" enables per-item debug tracing) — called
+// per item on the mux hot loop, so the env lock is paid once, not per call.
 pub fn debug_enabled() -> bool {
     static ENABLED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
     *ENABLED.get_or_init(|| {

src/io/tree_sink.rs

@@ -7,7 +7,7 @@
 
 use crate::error::{Error, Result};
 use std::collections::HashMap;
-use std::io::Write;
+use std::io::{self, Write};
 use std::path::{Path, PathBuf};
 
 /// Attempts to delete the case-probe marker before giving up.
@@ -30,8 +30,14 @@ impl TreeSink {
         std::fs::create_dir_all(dest).map_err(|e| Error::DirWriteFailed {
             errno: e.raw_os_error(),
         })?;
-        if !force && dir_is_non_empty(dest) {
-            return Err(Error::DirNotEmpty);
+        if !force {
+            // An unlistable folder is not known to be empty: refuse rather than mix trees.
+            let non_empty = dir_is_non_empty(dest).map_err(|e| Error::DirWriteFailed {
+                errno: e.raw_os_error(),
+            })?;
+            if non_empty {
+                return Err(Error::DirNotEmpty);
+            }
         }
         Ok(TreeSink {
             dest: dest.to_path_buf(),
@@ -389,11 +395,9 @@ pub(crate) fn available_space(_dir: &Path) -> Option<u64> {
     None
 }
 
-/// Whether a directory exists and contains any entry.
-pub(crate) fn dir_is_non_empty(dir: &Path) -> bool {
-    std::fs::read_dir(dir)
-        .map(|mut it| it.next().is_some())
-        .unwrap_or(false)
+/// Whether a directory contains any entry; an unlistable one is an error.
+pub(crate) fn dir_is_non_empty(dir: &Path) -> io::Result<bool> {
+    Ok(std::fs::read_dir(dir)?.next().is_some())
 }
 
 #[cfg(test)]
@@ -421,6 +425,23 @@ mod tests {
         assert!(!dir.path().join("a.bin.partial").exists());
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn an_unlistable_target_is_refused_not_treated_as_empty() {
+        use std::os::unix::fs::PermissionsExt;
+        let dir = tempfile::tempdir().unwrap();
+        let out = dir.path().join("tree");
+        std::fs::create_dir(&out).unwrap();
+        std::fs::write(out.join("old.bin"), b"x").unwrap();
+        std::fs::set_permissions(&out, std::fs::Permissions::from_mode(0o300)).unwrap();
+        let listable = std::fs::read_dir(&out).is_ok();
+        let r = TreeSink::create(&out, false);
+        std::fs::set_permissions(&out, std::fs::Permissions::from_mode(0o700)).unwrap();
+        if !listable {
+            assert!(matches!(r, Err(Error::DirWriteFailed { .. })));
+        }
+    }
+
     #[test]
     fn the_aacs_directories_are_not_part_of_the_tree() {
         assert!(!TreeSink::keeps_top_level("AACS", false));

src/io/writeback/linux.rs

@@ -351,11 +351,10 @@ impl WritebackPipeline {
         }
     }
 
-    /// Caller is about to seek away from the current write region.
-    /// Drain any in-flight chunk and reset tracking.
-    // MKV seeks every cluster, so never wait here: kick the tail and let the
-    // pending chunk wait at the next boundary. An un-waited range keeps its error
-    // for the final fsync.
+    /// Caller is about to seek away from the current write region. Kicks off
+    /// the tail and resets tracking; never waits (MKV seeks every cluster), so
+    /// the pending chunk is waited at the next boundary or `finalize`, and an
+    /// un-waited range keeps its error for the final fsync.
     pub(crate) fn handle_seek(&mut self, new_pos: u64) {
         let tail_len = self.pos.saturating_sub(self.last_flush_pos);
         if tail_len > 0 && self.waitable {
@@ -745,6 +744,50 @@ mod tests {
         assert_eq!(p.error().and_then(|e| e.raw_os_error()), Some(libc::EIO));
     }
 
+    static ERRNO_WAITS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
+
+    fn eio_then_erofs_wait(_fd: RawFd, _off: u64, _len: u64) -> i32 {
+        match ERRNO_WAITS.fetch_add(1, Ordering::SeqCst) {
+            0 => libc::EIO,
+            _ => libc::EROFS,
+        }
+    }
+
+    // The first data-loss errno is the cause; a later one never replaces it.
+    #[test]
+    fn the_first_latched_errno_is_kept() {
+        let (_f, mut p) = local_pipeline(CHUNK_BYTES_MIN);
+        p.wait_op = eio_then_erofs_wait;
+        p.note_progress(CHUNK_BYTES_MIN);
+        p.note_progress(2 * CHUNK_BYTES_MIN);
+        p.note_progress(3 * CHUNK_BYTES_MIN);
+        p.finalize();
+        assert!(ERRNO_WAITS.load(Ordering::SeqCst) >= 2, "a second wait ran");
+        assert_eq!(p.error().and_then(|e| e.raw_os_error()), Some(libc::EIO));
+    }
+
+    fn ok_wait(_fd: RawFd, _off: u64, _len: u64) -> i32 {
+        0
+    }
+
+    // Each completed WAIT_AFTER credits exactly the waited chunk's length as durable, at a
+    // boundary and at finalize.
+    #[test]
+    fn completed_waits_credit_the_waited_length() {
+        let (_f, mut p) = local_pipeline(CHUNK_BYTES_MIN);
+        let flush = crate::io::flush::FlushProgress::default();
+        flush.note_total(10 * CHUNK_BYTES_MIN);
+        p.set_flush_progress(flush.clone());
+        p.wait_op = ok_wait;
+        p.note_progress(CHUNK_BYTES_MIN);
+        assert_eq!(flush.bytes_durable(), 0, "nothing waited yet");
+        // A second chunk twice as long: the wait on the first credits only its length.
+        p.note_progress(3 * CHUNK_BYTES_MIN);
+        assert_eq!(flush.bytes_durable(), CHUNK_BYTES_MIN);
+        p.finalize();
+        assert_eq!(flush.bytes_durable(), 3 * CHUNK_BYTES_MIN);
+    }
+
     // Only data-loss errnos latch: ESPIPE/EINVAL mean the call was unsupported.
     #[test]
     fn non_writeback_errno_is_not_latched() {

src/io/writeback_file/flusher.rs

@@ -1,6 +1,6 @@
 //! The in-write flusher (stop design §2.10 items 1–4): one worker per file makes written
-//! bytes durable in chunks of `C` while the writer runs. `C` halves (to a floor) when one
-//! chunk flush is slow; the writer waits, halt-aware, while more than `2 × C` is unflushed;
+//! bytes durable in chunks of `C` while the writer runs. `C` starts at a floor, halves when one
+//! chunk flush is slow and doubles (to a ceiling) when one is fast; the writer waits, halt-aware, while more than `2 × C` is unflushed;
 //! a flusher with no progress for the stall window latches `SyncTimeout` (sticky).
 
 use std::fs::File;
@@ -329,4 +329,50 @@ mod tests {
         writer.join().unwrap().unwrap();
         assert!(f.error().is_none());
     }
+
+    struct FailingChunk;
+    impl FlushOps for FailingChunk {
+        fn chunk(&self, _: &File) -> io::Result<()> {
+            Err(io::Error::from_raw_os_error(28))
+        }
+        fn range(&self, _: &File, _: u64, _: u64) -> Option<io::Result<()>> {
+            None
+        }
+        fn finish(&self, _: &File) -> io::Result<()> {
+            Ok(())
+        }
+        fn sample(&self) -> Option<u64> {
+            None
+        }
+    }
+
+    // A failed chunk flush latches its own errno: the waiter gets the real error at once,
+    // not `SyncTimeout` after the stall window, and it stays latched.
+    #[test]
+    fn a_chunk_error_is_latched_and_reported_as_itself() {
+        let file = tempfile::tempfile().unwrap();
+        let timing = FlushTiming {
+            stall: Duration::from_millis(400),
+            slow_chunk: Duration::from_secs(10),
+            chunk_min: 100,
+            chunk_max: 100,
+            sample_every: Duration::from_millis(50),
+        };
+        let f = Flusher::spawn(
+            &file,
+            Arc::new(FailingChunk),
+            timing,
+            FlushProgress::default(),
+            0,
+        )
+        .unwrap();
+        let start = Instant::now();
+        let e = f.drain(100, None, None).expect_err("the chunk failed");
+        assert_eq!(e.raw_os_error(), Some(28), "{e}");
+        assert!(
+            start.elapsed() < Duration::from_millis(300),
+            "waited a stall"
+        );
+        assert_eq!(f.error().and_then(|e| e.raw_os_error()), Some(28));
+    }
 }

src/io/writeback_file/mod.rs

@@ -695,6 +695,54 @@ mod tests {
         assert!(w.hinted, "reservation released before a successful sync");
     }
 
+    // Rebinding the flush counters carries the length so far, makes the new counters the
+    // file's own, and moves a running flusher's durable credit onto them.
+    #[test]
+    fn set_flush_progress_rebinds_every_holder() {
+        struct OkOps;
+        impl FlushOps for OkOps {
+            fn chunk(&self, _: &File) -> io::Result<()> {
+                Ok(())
+            }
+            fn range(&self, _: &File, _: u64, _: u64) -> Option<io::Result<()>> {
+                None
+            }
+            fn finish(&self, _: &File) -> io::Result<()> {
+                Ok(())
+            }
+            fn sample(&self) -> Option<u64> {
+                None
+            }
+        }
+        let mut file = tempfile::tempfile().unwrap();
+        file.write_all(b"12345").unwrap();
+        let timing = FlushTiming {
+            chunk_min: 100,
+            chunk_max: 100,
+            ..FlushTiming::default()
+        };
+        let mut w = WritebackFile::with_flush_ops(file, Arc::new(OkOps), timing).unwrap();
+        w.write_all(&[0u8; 100]).unwrap();
+        let fresh = FlushProgress::default();
+        w.set_flush_progress(fresh.clone());
+        assert_eq!(
+            fresh.bytes_total(),
+            105,
+            "the length so far is carried over"
+        );
+        w.write_all(&[0u8; 300]).unwrap();
+        assert_eq!(w.flush_progress().bytes_total(), 405);
+        assert_eq!(fresh.bytes_total(), 405, "writes count on the new counters");
+        let start = std::time::Instant::now();
+        while fresh.bytes_durable() == 0 && start.elapsed() < std::time::Duration::from_secs(2) {
+            std::thread::sleep(std::time::Duration::from_millis(5));
+        }
+        assert!(
+            fresh.bytes_durable() > 0,
+            "the flusher credits the new counters"
+        );
+    }
+
     // `sync_all` is idempotent: calling it twice, then Drop (also
     // finalizes), must not corrupt data or panic.
     #[test]

src/whole_disc_tests.rs

@@ -173,6 +173,16 @@ fn unit_aligned_widens_reads_onto_each_files_grid() {
     assert_eq!(r.inner.reads, vec![(305, 3)]);
 }
 
+// A partial last unit is read to the span's end, never onto the sectors beyond it.
+#[test]
+fn unit_aligned_does_not_widen_a_read_past_the_span_end() {
+    let mut r = reader(vec![(100, 4, 100)]);
+    let mut buf = vec![0u8; 2048];
+    r.read_sectors(103, 1, &mut buf, false).unwrap();
+    assert_eq!(r.inner.reads, vec![(103, 1)]);
+    assert_eq!(buf[0], 103);
+}
+
 #[test]
 fn unit_aligned_propagates_inner_read_errors_on_both_paths() {
     let mut r = reader(vec![(100, 30, 100)]);
@@ -208,6 +218,9 @@ fn unit_block_end_pulls_back_to_the_straddling_units_head() {
     assert_eq!(r.unit_block_end(90, 103), 103);
     assert_eq!(r.unit_block_end(103, 104), 104);
     assert_eq!(r.unit_block_end(0, 50), 50);
+    // The straddled unit's head lies before the span: no pull-back into the previous region.
+    let r = reader(vec![(300, 9, 299)]);
+    assert_eq!(r.unit_block_end(250, 300), 300);
 }
 
 // Spec guards (keys-upfront-design §7.8) on the per-file unit grid.
@@ -338,6 +351,33 @@ mod spec_guards {
     }
 }
 
+// The inner source's access pattern and speed control reach the mux/driver through the wrapper.
+#[test]
+fn unit_aligned_forwards_random_access_and_speed() {
+    #[derive(Default)]
+    struct Sequential {
+        speed: Option<u16>,
+    }
+    impl SectorSource for Sequential {
+        fn capacity_sectors(&self) -> u32 {
+            0
+        }
+        fn read_sectors(&mut self, _: u32, _: u16, _: &mut [u8], _: bool) -> Result<usize> {
+            Ok(0)
+        }
+        fn random_access(&self) -> bool {
+            false
+        }
+        fn set_speed(&mut self, kbs: u16) {
+            self.speed = Some(kbs);
+        }
+    }
+    let mut r = UnitAligned::new(Sequential::default(), Vec::new());
+    assert!(!r.random_access());
+    r.set_speed(1234);
+    assert_eq!(r.inner.speed, Some(1234));
+}
+
 // UnitAligned wraps the decrypting reader for whole-disc/image copies; it must
 // relay the inner source's unmapped list (an image built on it would else drop it).
 #[test]