io: phase 1 buffering — read-side flatness

Three changes targeting 0.20.9's "muxer never read-stalls on NFS read
latency" invariant:

A. FileSectorSource gets a 32 MiB internal read-ahead buffer
   (READAHEAD_BUF_BYTES). Splits out from src/sector/file.rs into
   src/io/file_sector_source/ with per-OS open hints (Linux
   posix_fadvise(SEQUENTIAL), macOS fcntl(F_RDADVISE) with 64 MiB
   cap, Windows TODO stub, BSD/illumos no-op). Backward seeks
   rebuffer; partial reads at EOF return only the bytes that exist;
   oversize-request bypass for count > BUF_SECTORS.

B. WritebackFile inline #[cfg(target_os = "linux")] blocks split
   into per-OS files under src/io/writeback_file/. Linux unchanged
   (fallocate KEEP_SIZE, fsync via bounded_syscall). macOS gets a
   real F_PREALLOCATE + F_FULLFSYNC impl (was a "skipped (non-linux)"
   debug log before). Windows is a stub (FlushFileBuffers via
   std sync_all; TODO for SetFileValidData). BSDs/illumos fall back
   to std sync_all.

C. New byte_channel module — byte-bounded producer/consumer wrapping
   std sync_channel with Mutex/Condvar byte accounting. Sender blocks
   when used_bytes + item.byte_size() > capacity. HasByteSize impl
   for PesFrame. Default cap BYTE_CHANNEL_DEFAULT_CAPACITY = 64 MiB,
   sized to absorb worst-case NFS read p99 (~2 s × UHD peak compressed
   ~15 MB/s). The mux call site lives in autorip (out of scope here);
   this lands the primitive in libfreemkv for autorip to adopt.

Test counts: byte_channel +6, file_sector_source +5, sector::file
round-trip suite (3) preserved. passn_handler_ab.rs A/B fixture
(8 profiles) still green.

precommit.sh libfreemkv: fmt + clippy + test all green on Rust 1.86.

No version bump; no Cargo.lock changes; no forbidden-file edits
(disc/patch.rs, disc/read_error.rs, io/pipeline.rs,
tests/passn_handler_ab.rs).

src/io/byte_channel.rs

@@ -0,0 +1,330 @@
+//! Byte-sized bounded producer/consumer channel.
+//!
+//! Wraps `std::sync::mpsc::sync_channel` with a byte-accounting
+//! `Mutex<usize> + Condvar` cap. Sender blocks (cooperatively) when
+//! `used_bytes + item.byte_size() > capacity_bytes`. Receiver
+//! decrements `used_bytes` when it takes the item.
+//!
+//! Why: the existing producer→consumer channel between `DiscStream`
+//! (PES producer) and `MuxSink` (PES consumer) is bounded by frame
+//! count. Frame sizes vary 100× between metadata and keyframes, so a
+//! count-based cap either starves on small frames or buffers far too
+//! much memory on big ones. Byte-sized accounting sizes the buffer for
+//! the worst-case input stall (NFS read p99 ≈ 1–2 s × ~15 MB/s peak
+//! compressed bitrate ≈ ~30 MB) directly.
+//!
+//! The underlying mpsc channel is created with a very large slot count
+//! so the byte cap (not the slot count) is the real backpressure. Slot
+//! count is only there to give the kernel a small chunk to wake on.
+//!
+//! See `freemkv-private/memory/project_buffering_architecture.md` §
+//! Pipeline channel — sizing.
+
+use std::sync::mpsc::{Receiver as MpscReceiver, RecvError, SendError, SyncSender, sync_channel};
+use std::sync::{Arc, Condvar, Mutex};
+
+/// Default byte cap for the muxer's input channel. Sized to hide a
+/// worst-case ~2 s NFS read refill at UHD peak compressed bitrate
+/// (~15 MB/s); 64 MiB gives headroom. Tweakable; not magic.
+pub const BYTE_CHANNEL_DEFAULT_CAPACITY: usize = 64 * 1024 * 1024;
+
+/// Slot capacity of the inner `sync_channel`. Large so the byte cap is
+/// the real backpressure mechanism — the mpsc slot count only exists
+/// to give the kernel a chunk to wake on. PES frames are typically
+/// ~700 B each, so 64 MiB ≈ 90 k frames; 200 k is comfortable headroom.
+const INNER_SLOT_CAPACITY: usize = 200_000;
+
+/// Anything whose in-memory cost can be accounted by a single
+/// `usize`. Implement on the item type sent through [`Sender`].
+pub trait HasByteSize {
+    /// Bytes this item contributes to the channel's used budget.
+    /// Must be > 0 to make progress (a 0-byte item would never
+    /// block the sender no matter the cap; see send_blocks_at_capacity
+    /// test).
+    fn byte_size(&self) -> usize;
+}
+
+impl HasByteSize for crate::pes::PesFrame {
+    fn byte_size(&self) -> usize {
+        // Frame data + the fixed header overhead the serializer
+        // writes (track + pts + keyframe + len). The `Vec<u8>` heap
+        // allocation also has alloc-header overhead but that's
+        // <0.1 % at typical frame sizes — folding it in would just
+        // add noise to the budget.
+        self.data.len() + 14
+    }
+}
+
+/// Shared book-keeping between [`Sender`] and [`Receiver`]. Wrapped in
+/// an `Arc` because both halves hold it independently.
+struct Accounting {
+    used: Mutex<usize>,
+    cv: Condvar,
+    capacity: usize,
+}
+
+/// Send half of the byte-bounded channel.
+///
+/// `send` blocks (on a `Condvar`) when adding the item would push
+/// `used_bytes` past `capacity_bytes`. Unblocks when the receiver
+/// `recv`s items out and notifies. Returns `Err(item)` if the
+/// receiver has been dropped — mirrors `mpsc::SyncSender::send`.
+pub struct Sender<T: HasByteSize> {
+    tx: SyncSender<T>,
+    acct: Arc<Accounting>,
+}
+
+impl<T: HasByteSize> Clone for Sender<T> {
+    fn clone(&self) -> Self {
+        Sender {
+            tx: self.tx.clone(),
+            acct: self.acct.clone(),
+        }
+    }
+}
+
+impl<T: HasByteSize> Sender<T> {
+    /// Push one item. Blocks until adding it would not exceed the
+    /// capacity, then sends through the inner mpsc channel.
+    pub fn send(&self, item: T) -> Result<(), SendError<T>> {
+        let sz = item.byte_size();
+        // Reserve capacity first. The reservation is observable to
+        // other senders via `used`; only after we win the slot do we
+        // hand the item to the inner mpsc channel. That ordering means
+        // `used` is always a conservative upper bound on what's in the
+        // mpsc queue + about-to-be-sent.
+        {
+            let mut used = self.acct.used.lock().expect("byte_channel poisoned");
+            // An item bigger than the whole capacity will never fit; let
+            // it through anyway as a one-shot reservation, otherwise the
+            // sender deadlocks forever waiting for `used == 0` AND
+            // nothing in flight. The receiver will drain it on the
+            // other side. Same behaviour as `std::sync::mpsc` for
+            // arbitrarily large messages.
+            while *used + sz > self.acct.capacity && *used > 0 {
+                used = self.acct.cv.wait(used).expect("byte_channel cv poisoned");
+            }
+            *used += sz;
+        }
+        match self.tx.send(item) {
+            Ok(()) => Ok(()),
+            Err(SendError(returned)) => {
+                // Receiver dropped — refund the reservation so a later
+                // sender on a clone doesn't observe phantom used bytes
+                // (the receiver is gone so nobody will decrement).
+                let mut used = self.acct.used.lock().expect("byte_channel poisoned");
+                *used = used.saturating_sub(sz);
+                self.acct.cv.notify_all();
+                Err(SendError(returned))
+            }
+        }
+    }
+}
+
+/// Receive half of the byte-bounded channel.
+///
+/// `recv` blocks on the inner mpsc until an item is available, then
+/// decrements the byte-accounting and wakes any sender waiting on
+/// capacity.
+pub struct Receiver<T: HasByteSize> {
+    rx: MpscReceiver<T>,
+    acct: Arc<Accounting>,
+}
+
+impl<T: HasByteSize> Receiver<T> {
+    /// Take the next item. Returns `Err(RecvError)` when all senders
+    /// have been dropped and the channel is empty.
+    pub fn recv(&self) -> Result<T, RecvError> {
+        let item = self.rx.recv()?;
+        let sz = item.byte_size();
+        let mut used = self.acct.used.lock().expect("byte_channel poisoned");
+        *used = used.saturating_sub(sz);
+        // Notify all so multi-sender setups wake every blocked sender,
+        // not just one. Wasted wakeups are cheap; missed wakeups would
+        // be a deadlock.
+        self.acct.cv.notify_all();
+        Ok(item)
+    }
+}
+
+/// Create a byte-bounded channel with the given capacity in bytes.
+/// Returns a `(Sender, Receiver)` pair; clone the `Sender` for
+/// multi-producer setups.
+pub fn channel<T: HasByteSize>(capacity_bytes: usize) -> (Sender<T>, Receiver<T>) {
+    let (tx, rx) = sync_channel::<T>(INNER_SLOT_CAPACITY);
+    let acct = Arc::new(Accounting {
+        used: Mutex::new(0),
+        cv: Condvar::new(),
+        capacity: capacity_bytes,
+    });
+    (
+        Sender {
+            tx,
+            acct: acct.clone(),
+        },
+        Receiver { rx, acct },
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::thread;
+    use std::time::{Duration, Instant};
+
+    /// Test payload — its `byte_size` returns whatever we passed at
+    /// construction so capacity math is exact and predictable.
+    #[derive(Clone, Debug, PartialEq, Eq)]
+    struct Item {
+        sz: usize,
+        tag: u32,
+    }
+
+    impl HasByteSize for Item {
+        fn byte_size(&self) -> usize {
+            self.sz
+        }
+    }
+
+    #[test]
+    fn send_recv_round_trip() {
+        let (tx, rx) = channel::<Item>(1024);
+        for i in 0..5 {
+            tx.send(Item { sz: 100, tag: i }).unwrap();
+        }
+        for i in 0..5 {
+            let got = rx.recv().unwrap();
+            assert_eq!(got, Item { sz: 100, tag: i });
+        }
+    }
+
+    #[test]
+    fn byte_accounting_decrements_on_recv() {
+        // Internal book-keeping check via observable side-effect: after
+        // sending K items totalling N bytes and receiving them all, a
+        // subsequent send of an N-byte item must NOT block (no items
+        // in flight, all capacity refunded).
+        let (tx, rx) = channel::<Item>(1024);
+        for _ in 0..4 {
+            tx.send(Item { sz: 256, tag: 0 }).unwrap();
+        }
+        for _ in 0..4 {
+            rx.recv().unwrap();
+        }
+        // Cap is now fully available again. Send a 1024-byte item; the
+        // `used > 0` guard means it goes through alone (no wait).
+        let start = Instant::now();
+        tx.send(Item { sz: 1024, tag: 99 }).unwrap();
+        assert!(start.elapsed() < Duration::from_millis(100));
+        let got = rx.recv().unwrap();
+        assert_eq!(got.tag, 99);
+    }
+
+    #[test]
+    fn send_blocks_at_capacity_unblocks_on_recv() {
+        // Cap = 200 bytes, item = 100 bytes. First two sends fit
+        // exactly; the third must block until a recv frees capacity.
+        let (tx, rx) = channel::<Item>(200);
+        tx.send(Item { sz: 100, tag: 0 }).unwrap();
+        tx.send(Item { sz: 100, tag: 1 }).unwrap();
+
+        let tx2 = tx.clone();
+        let sent_at = Arc::new(Mutex::new(None::<Instant>));
+        let sent_at2 = sent_at.clone();
+        let h = thread::spawn(move || {
+            tx2.send(Item { sz: 100, tag: 2 }).unwrap();
+            *sent_at2.lock().unwrap() = Some(Instant::now());
+        });
+
+        // Give the sender thread a head start; it should be parked in
+        // `cv.wait` because used (200) + 100 > capacity (200).
+        thread::sleep(Duration::from_millis(100));
+        assert!(
+            sent_at.lock().unwrap().is_none(),
+            "third send should be blocked at capacity"
+        );
+
+        // Drain one. Sender wakes and completes.
+        let recv_at = Instant::now();
+        let got = rx.recv().unwrap();
+        assert_eq!(got.tag, 0);
+        h.join().unwrap();
+
+        let sent_when = sent_at.lock().unwrap().unwrap();
+        assert!(
+            sent_when >= recv_at,
+            "sender must complete AFTER receiver freed capacity"
+        );
+
+        // Drain the remaining two.
+        assert_eq!(rx.recv().unwrap().tag, 1);
+        assert_eq!(rx.recv().unwrap().tag, 2);
+    }
+
+    #[test]
+    fn item_larger_than_capacity_still_goes_through() {
+        // Pathological case: a single item bigger than the capacity.
+        // The guard `*used > 0` lets it through when the channel is
+        // empty (otherwise the sender deadlocks forever). Matches
+        // `mpsc::SyncSender` semantics for oversize messages.
+        let (tx, rx) = channel::<Item>(100);
+        tx.send(Item { sz: 1000, tag: 7 }).unwrap();
+        let got = rx.recv().unwrap();
+        assert_eq!(got, Item { sz: 1000, tag: 7 });
+    }
+
+    #[test]
+    fn concurrent_send_recv_stress() {
+        // 4 sender threads × 1k items each, 1 receiver. Verify byte
+        // accounting stays sane (channel never deadlocks, every item
+        // arrives exactly once) under contention.
+        const SENDERS: u32 = 4;
+        const PER_SENDER: u32 = 1000;
+        const TOTAL: u32 = SENDERS * PER_SENDER;
+
+        let (tx, rx) = channel::<Item>(8 * 1024);
+        let sent = Arc::new(AtomicUsize::new(0));
+        let mut handles = Vec::new();
+        for s in 0..SENDERS {
+            let tx = tx.clone();
+            let sent = sent.clone();
+            handles.push(thread::spawn(move || {
+                for i in 0..PER_SENDER {
+                    // Vary item size so accounting actually has to
+                    // multiplex differently-sized blockers. 1B → 256B.
+                    let sz = 1 + ((i as usize) % 256);
+                    tx.send(Item {
+                        sz,
+                        tag: s * PER_SENDER + i,
+                    })
+                    .unwrap();
+                    sent.fetch_add(1, Ordering::SeqCst);
+                }
+            }));
+        }
+        // Drop our local sender so the receiver can eventually see
+        // RecvError once all sender clones are done. Cloning the
+        // sender into each producer means each clone Drop'd separately.
+        drop(tx);
+
+        let mut received = 0u32;
+        while let Ok(_item) = rx.recv() {
+            received += 1;
+        }
+        for h in handles {
+            h.join().unwrap();
+        }
+        assert_eq!(received, TOTAL);
+        assert_eq!(sent.load(Ordering::SeqCst) as u32, TOTAL);
+    }
+
+    #[test]
+    fn send_after_recv_dropped_returns_err() {
+        let (tx, rx) = channel::<Item>(1024);
+        drop(rx);
+        let r = tx.send(Item { sz: 10, tag: 0 });
+        assert!(r.is_err());
+    }
+}

src/io/file_sector_source/linux.rs

@@ -0,0 +1,15 @@
+//! Linux: hint the kernel that this fd will be read sequentially so
+//! readahead widens. `posix_fadvise(POSIX_FADV_SEQUENTIAL)` is a hint,
+//! not a guarantee — the kernel still owns the policy decision.
+
+use std::fs::File;
+use std::os::unix::io::AsRawFd;
+
+pub(super) fn hint_sequential(file: &File, _len_bytes: u64) {
+    // Best-effort: return value ignored. A fadvise failure has no
+    // user-observable consequence (reads still work, just without the
+    // widened readahead window).
+    unsafe {
+        libc::posix_fadvise(file.as_raw_fd(), 0, 0, libc::POSIX_FADV_SEQUENTIAL);
+    }
+}

src/io/file_sector_source/macos.rs

@@ -0,0 +1,38 @@
+//! macOS: hint the kernel to prefetch a generous chunk. macOS has no
+//! direct `POSIX_FADV_SEQUENTIAL` equivalent; the idiomatic hint is
+//! `fcntl(F_RDADVISE, &radvisory)` describing the byte range you
+//! intend to read soon. We point it at the whole file (clamped to a
+//! ceiling so a multi-TB ISO doesn't ask the kernel to prefetch
+//! everything at once).
+
+use std::fs::File;
+use std::os::unix::io::AsRawFd;
+
+/// `F_RDADVISE` opcode — not in libc's named constants on all SDKs.
+const F_RDADVISE: libc::c_int = 44;
+
+/// Cap on the byte length we pass to `F_RDADVISE`. Asking for a
+/// multi-GB readahead window is counterproductive — the OS doesn't
+/// have that much cache to throw at one fd. 64 MiB is generous for
+/// our use case (sweep, mux) and matches the byte-channel cap so the
+/// kernel's prefetch ≥ our app-level pipeline depth.
+const RDADVISE_MAX_BYTES: i64 = 64 * 1024 * 1024;
+
+/// `radvisory` per `<sys/fcntl.h>`. repr(C) layout is stable.
+#[repr(C)]
+struct RadAdvisory {
+    ra_offset: libc::off_t,
+    ra_count: libc::c_int,
+}
+
+pub(super) fn hint_sequential(file: &File, len_bytes: u64) {
+    let bytes = (len_bytes as i64).min(RDADVISE_MAX_BYTES);
+    let mut ra = RadAdvisory {
+        ra_offset: 0,
+        ra_count: bytes as libc::c_int,
+    };
+    // Best-effort.
+    unsafe {
+        libc::fcntl(file.as_raw_fd(), F_RDADVISE, &mut ra);
+    }
+}

src/io/file_sector_source/mod.rs

@@ -0,0 +1,393 @@
+//! [`FileSectorSource`] — read 2048-byte sectors from an ISO file on
+//! disk, with an internal 32 MiB read-ahead buffer.
+//!
+//! ## Why the buffer
+//!
+//! On NFS-mounted ISOs, an unbuffered `pread(2048)` per sector pays an
+//! NFS round-trip for every sector. With `rsize=1 MiB` and a 100-150 ms
+//! NFS RTT, that's three orders of magnitude more round trips than
+//! necessary — the muxer goes read-bound on every read, even though
+//! the local NFS client could deliver MB/s on bigger requests.
+//!
+//! Internally this source keeps a [`READAHEAD_BUF_BYTES`] (32 MiB)
+//! window pre-read from the file. `read_sectors(lba, count)` slices
+//! into the window if `[lba, lba+count)` is contained in it; otherwise
+//! the window is refilled (full-size aligned to the requested LBA's
+//! buffer position).
+//!
+//! ## Access pattern assumption
+//!
+//! The buffer is sized for **forward-sequential** reads (sweep, mux).
+//! Reverse-mode patch is range-local, so a refill per range works out
+//! fine (the buffer covers the whole range for typical bad-range
+//! sizes). Random-access reads thrash the buffer — at which point the
+//! 32 MiB pre-read is wasted work. We accept that: the use case is
+//! mux + sweep, both forward-sequential.
+//!
+//! Backward seeks rebuffer from the new LBA; partial reads at EOF
+//! return only the bytes that exist (the underlying file is shorter
+//! than a full buffer slot).
+//!
+//! ## Platform open hints
+//!
+//! On `open()` each platform issues its "sequential access expected"
+//! hint to the kernel so OS-level readahead widens. The hint lives in
+//! a per-OS sibling module ([`linux::hint_sequential`] et al.) — no
+//! inline `#[cfg]` in this file.
+
+#[cfg(target_os = "linux")]
+mod linux;
+#[cfg(target_os = "macos")]
+mod macos;
+#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
+mod other;
+#[cfg(target_os = "windows")]
+mod windows;
+
+#[cfg(target_os = "linux")]
+use linux as platform;
+#[cfg(target_os = "macos")]
+use macos as platform;
+#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
+use other as platform;
+#[cfg(target_os = "windows")]
+use windows as platform;
+
+use std::fs::File;
+use std::io::{Read, Seek, SeekFrom};
+use std::path::Path;
+
+use crate::error::{Error, Result};
+use crate::sector::SectorSource;
+
+/// Internal read-ahead buffer size. 32 MiB amortises one NFS round
+/// trip across ~16 k sectors — three orders of magnitude fewer trips
+/// than per-sector pread, and large enough to coast through a typical
+/// NFS server commit blip.
+///
+/// Tweakable. Named const, not a magic number.
+pub const READAHEAD_BUF_BYTES: usize = 32 * 1024 * 1024;
+
+const SECTOR_SIZE: usize = 2048;
+/// Sectors per refill: [`READAHEAD_BUF_BYTES`] / [`SECTOR_SIZE`]. The
+/// buffer always tries to hold this many, except at the tail of the
+/// file where less data exists.
+const BUF_SECTORS: u32 = (READAHEAD_BUF_BYTES / SECTOR_SIZE) as u32;
+
+/// SectorSource backed by a file (ISO image) with an internal
+/// `READAHEAD_BUF_BYTES`-sized read-ahead window.
+///
+/// `read_sectors` is satisfied from the buffer when possible; otherwise
+/// a full-buffer refill is issued at the requested LBA's position and
+/// the call is re-tried against the freshly populated window.
+pub struct FileSectorSource {
+    file: File,
+    /// Total file size in sectors. Constant after construction;
+    /// surfaced via [`SectorSource::capacity_sectors`].
+    capacity: u32,
+    /// Heap-allocated read-ahead buffer. Boxed slice (not `Vec`) so
+    /// the allocation is sized exactly once and the layout is
+    /// stable. Length is at most [`READAHEAD_BUF_BYTES`].
+    buf: Box<[u8]>,
+    /// First LBA covered by `buf`. `buf_len_sectors == 0` means the
+    /// buffer is empty (e.g. immediately after construction or a
+    /// backward seek); the value of `buf_start_lba` is then
+    /// meaningless.
+    buf_start_lba: u32,
+    /// Number of valid sectors in `buf`. `[buf_start_lba,
+    /// buf_start_lba + buf_len_sectors)` is the half-open LBA range
+    /// the buffer currently caches.
+    buf_len_sectors: u32,
+}
+
+impl FileSectorSource {
+    /// Open an existing ISO file for reading. Capacity is derived
+    /// from `metadata().len() / 2048`. Returns
+    /// [`Error::IsoTooLarge`] if the file would exceed the 32-bit
+    /// LBA address space (~8 TB).
+    ///
+    /// Issues the platform's "sequential access expected" hint on the
+    /// fd (Linux `posix_fadvise(SEQUENTIAL)`, macOS `fcntl(F_RDADVISE)`,
+    /// Windows TODO stub) so the kernel's readahead widens.
+    pub fn open(path: &Path) -> std::io::Result<Self> {
+        let file = File::open(path)?;
+        let len = file.metadata()?.len();
+        let sectors = len / SECTOR_SIZE as u64;
+        if sectors > u32::MAX as u64 {
+            return Err(Error::IsoTooLarge {
+                path: path.to_string_lossy().into_owned(),
+            }
+            .into());
+        }
+        let capacity = sectors as u32;
+
+        // Best-effort sequential hint. Ignored on platforms without
+        // an equivalent primitive (or where the API exists but the
+        // FS doesn't honour it).
+        platform::hint_sequential(&file, len);
+
+        // Pre-allocate the buffer once. `vec![0u8; N].into_boxed_slice()`
+        // is the canonical way to fix the allocation size up-front;
+        // `Vec::with_capacity` would leave `len == 0` and force callers
+        // to do unsafe length manipulation to write into it.
+        let buf = vec![0u8; READAHEAD_BUF_BYTES].into_boxed_slice();
+
+        Ok(Self {
+            file,
+            capacity,
+            buf,
+            buf_start_lba: 0,
+            buf_len_sectors: 0,
+        })
+    }
+
+    /// True if `[lba, lba + count)` is wholly inside the current
+    /// buffer window. `count == 0` is vacuously true.
+    fn buffer_covers(&self, lba: u32, count: u32) -> bool {
+        if self.buf_len_sectors == 0 {
+            return false;
+        }
+        let end = match lba.checked_add(count) {
+            Some(e) => e,
+            None => return false,
+        };
+        let buf_end = self.buf_start_lba.saturating_add(self.buf_len_sectors);
+        lba >= self.buf_start_lba && end <= buf_end
+    }
+
+    /// Refill the buffer so it starts at `lba`. Read as many sectors
+    /// as we have buffer space AND file capacity for. Caller has
+    /// already checked `lba < capacity`.
+    fn refill(&mut self, lba: u32) -> Result<()> {
+        debug_assert!(lba < self.capacity, "refill past capacity");
+        // Don't read past EOF — clamp the request to remaining
+        // sectors. partial-buffer-at-EOF behaviour is intentional.
+        let want = BUF_SECTORS.min(self.capacity - lba);
+        let want_bytes = want as usize * SECTOR_SIZE;
+        let offset = lba as u64 * SECTOR_SIZE as u64;
+        self.file
+            .seek(SeekFrom::Start(offset))
+            .map_err(|e| Error::IoError { source: e })?;
+        self.file
+            .read_exact(&mut self.buf[..want_bytes])
+            .map_err(|e| Error::IoError { source: e })?;
+        self.buf_start_lba = lba;
+        self.buf_len_sectors = want;
+        Ok(())
+    }
+}
+
+impl SectorSource for FileSectorSource {
+    fn capacity_sectors(&self) -> u32 {
+        self.capacity
+    }
+
+    fn read_sectors(
+        &mut self,
+        lba: u32,
+        count: u16,
+        out: &mut [u8],
+        _recovery: bool,
+    ) -> Result<usize> {
+        let count = count as u32;
+        let bytes = count as usize * SECTOR_SIZE;
+        debug_assert!(
+            out.len() >= bytes,
+            "FileSectorSource::read_sectors: out len {} < requested {}",
+            out.len(),
+            bytes
+        );
+        if count == 0 {
+            return Ok(0);
+        }
+        // Refill if the requested range isn't entirely buffered.
+        // `buffer_covers` also handles the empty-buffer case
+        // (buf_len_sectors == 0).
+        if !self.buffer_covers(lba, count) {
+            // A request larger than the buffer itself can never fit;
+            // fall back to a one-shot direct pread for that pathological
+            // case so callers can't deadlock the source.
+            if count > BUF_SECTORS {
+                let offset = lba as u64 * SECTOR_SIZE as u64;
+                self.file
+                    .seek(SeekFrom::Start(offset))
+                    .map_err(|e| Error::IoError { source: e })?;
+                self.file
+                    .read_exact(&mut out[..bytes])
+                    .map_err(|e| Error::IoError { source: e })?;
+                // Invalidate buffer state — we bypassed it, the
+                // window is no longer authoritative for this LBA.
+                self.buf_len_sectors = 0;
+                return Ok(bytes);
+            }
+            self.refill(lba)?;
+        }
+
+        // Slice the buffer at the requested LBA's offset within it.
+        let off_sectors = (lba - self.buf_start_lba) as usize;
+        let off_bytes = off_sectors * SECTOR_SIZE;
+        out[..bytes].copy_from_slice(&self.buf[off_bytes..off_bytes + bytes]);
+        Ok(bytes)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Write;
+    use tempfile::tempdir;
+
+    /// Build a deterministic ISO of `sectors` sectors where sector `n`
+    /// is filled with the byte pattern `((n & 0xff) as u8)`. Lets us
+    /// verify any sector by content alone.
+    fn make_iso(path: &std::path::Path, sectors: u32) {
+        let mut f = std::fs::File::create(path).unwrap();
+        let mut chunk = vec![0u8; SECTOR_SIZE];
+        for n in 0..sectors {
+            let b = (n & 0xff) as u8;
+            chunk.iter_mut().for_each(|c| *c = b);
+            f.write_all(&chunk).unwrap();
+        }
+        f.flush().unwrap();
+    }
+
+    #[test]
+    fn sequential_reads_match_file() {
+        // Two full buffer windows + a tail = exercise refill across
+        // boundaries.
+        let total = BUF_SECTORS * 2 + 17;
+        let dir = tempdir().unwrap();
+        let path = dir.path().join("seq.iso");
+        make_iso(&path, total);
+
+        let mut src = FileSectorSource::open(&path).unwrap();
+        assert_eq!(src.capacity_sectors(), total);
+
+        let mut got = vec![0u8; SECTOR_SIZE];
+        for lba in 0..total {
+            src.read_sectors(lba, 1, &mut got, false).unwrap();
+            let expected = (lba & 0xff) as u8;
+            assert!(
+                got.iter().all(|b| *b == expected),
+                "sector {lba} content mismatch: expected 0x{expected:02x}"
+            );
+        }
+    }
+
+    #[test]
+    fn multi_sector_read_spanning_buffer_boundary() {
+        // A read that lands exactly on the last sector of the buffer
+        // plus the first sector of the next refill must rebuffer
+        // mid-read. Bypass path triggers when count > BUF_SECTORS; we
+        // want the in-window path, so count stays small but
+        // straddles the boundary.
+        let total = BUF_SECTORS * 2;
+        let dir = tempdir().unwrap();
+        let path = dir.path().join("span.iso");
+        make_iso(&path, total);
+
+        let mut src = FileSectorSource::open(&path).unwrap();
+
+        // Prime: read sector 0 to populate buffer #0.
+        let mut got = vec![0u8; SECTOR_SIZE];
+        src.read_sectors(0, 1, &mut got, false).unwrap();
+        assert_eq!(src.buf_start_lba, 0);
+        assert!(src.buf_len_sectors > 0);
+
+        // Now read 4 sectors crossing the buffer boundary at
+        // BUF_SECTORS - 2 → BUF_SECTORS + 1. Spans the refill.
+        let span_lba = BUF_SECTORS - 2;
+        let mut buf4 = vec![0u8; SECTOR_SIZE * 4];
+        src.read_sectors(span_lba, 4, &mut buf4, false).unwrap();
+        for i in 0..4 {
+            let lba = span_lba + i as u32;
+            let expected = (lba & 0xff) as u8;
+            for b in &buf4[i * SECTOR_SIZE..(i + 1) * SECTOR_SIZE] {
+                assert_eq!(*b, expected, "byte mismatch at sub-sector {i}");
+            }
+        }
+    }
+
+    #[test]
+    fn backward_seek_rebuffers() {
+        // Read forward across two windows, then jump back to sector
+        // 0. Buffer must refill from the start.
+        let total = BUF_SECTORS * 2 + 5;
+        let dir = tempdir().unwrap();
+        let path = dir.path().join("back.iso");
+        make_iso(&path, total);
+
+        let mut src = FileSectorSource::open(&path).unwrap();
+        let mut got = vec![0u8; SECTOR_SIZE];
+
+        // Forward to the second window.
+        src.read_sectors(BUF_SECTORS + 1, 1, &mut got, false)
+            .unwrap();
+        let start_after_forward = src.buf_start_lba;
+        assert!(start_after_forward >= BUF_SECTORS);
+
+        // Backward to sector 0. The current buffer doesn't cover it
+        // → refill must happen.
+        src.read_sectors(0, 1, &mut got, false).unwrap();
+        assert_eq!(src.buf_start_lba, 0);
+        assert!(got.iter().all(|b| *b == 0));
+    }
+
+    #[test]
+    fn partial_buffer_at_eof() {
+        // File is smaller than one buffer window. The buffer must
+        // populate with only the available sectors and reads must
+        // still succeed.
+        let total: u32 = 100;
+        assert!(total < BUF_SECTORS);
+        let dir = tempdir().unwrap();
+        let path = dir.path().join("small.iso");
+        make_iso(&path, total);
+
+        let mut src = FileSectorSource::open(&path).unwrap();
+        assert_eq!(src.capacity_sectors(), total);
+
+        let mut got = vec![0u8; SECTOR_SIZE];
+        // First read triggers refill clamped to `total`.
+        src.read_sectors(0, 1, &mut got, false).unwrap();
+        assert_eq!(src.buf_len_sectors, total);
+
+        // Read the very last sector — still inside the buffer.
+        src.read_sectors(total - 1, 1, &mut got, false).unwrap();
+        let expected = ((total - 1) & 0xff) as u8;
+        assert!(got.iter().all(|b| *b == expected));
+    }
+
+    #[test]
+    fn oversized_read_bypasses_buffer() {
+        // A request larger than the buffer must not deadlock the
+        // refill (which only loads BUF_SECTORS at a time). Bypass
+        // path handles it via direct pread.
+        let total = BUF_SECTORS + 100;
+        let dir = tempdir().unwrap();
+        let path = dir.path().join("over.iso");
+        make_iso(&path, total);
+
+        let mut src = FileSectorSource::open(&path).unwrap();
+        // Read more than BUF_SECTORS in one call. count is u16, so we
+        // can't actually exceed BUF_SECTORS (16k) — but the path also
+        // triggers via `out.len() / SECTOR_SIZE > BUF_SECTORS` check
+        // implicitly because count > BUF_SECTORS. BUF_SECTORS for
+        // 32 MiB is 16384, which does fit in u16 (max 65535). Cap
+        // at BUF_SECTORS + 1 to exercise the bypass.
+        let req = (BUF_SECTORS + 1) as u16;
+        let req_bytes = req as usize * SECTOR_SIZE;
+        let mut big = vec![0u8; req_bytes];
+        src.read_sectors(0, req, &mut big, false).unwrap();
+        // Spot-check sector 0 and the last requested sector.
+        assert!(big[..SECTOR_SIZE].iter().all(|b| *b == 0));
+        let last_lba = req as u32 - 1;
+        let exp = (last_lba & 0xff) as u8;
+        let last_off = (req as usize - 1) * SECTOR_SIZE;
+        assert!(
+            big[last_off..last_off + SECTOR_SIZE]
+                .iter()
+                .all(|b| *b == exp)
+        );
+    }
+}

src/io/file_sector_source/other.rs

@@ -0,0 +1,7 @@
+//! Fallback for targets without a known sequential-readahead hint
+//! (BSDs, illumos, etc.). No-op — reads still work, they just don't
+//! get the OS-level prefetch widening.
+
+use std::fs::File;
+
+pub(super) fn hint_sequential(_file: &File, _len_bytes: u64) {}

src/io/file_sector_source/windows.rs

@@ -0,0 +1,19 @@
+//! Windows: the canonical sequential-access hint is
+//! `FILE_FLAG_SEQUENTIAL_SCAN` passed to `CreateFile` at open time —
+//! it cannot be set after the fact via `SetFileInformationByHandle`.
+//! Routing the open call through this module would mean a custom
+//! `File::from_raw_handle` plumb for every `FileSectorSource::open`
+//! caller, which is more invasive than the Phase 1 scope.
+//!
+//! TODO: replumb `FileSectorSource::open` to take an
+//! `OpenOptions`-style builder so the Windows path can flip the flag
+//! at open time. For now this is a no-op stub.
+
+use std::fs::File;
+
+pub(super) fn hint_sequential(_file: &File, _len_bytes: u64) {
+    tracing::debug!(
+        target: "mux",
+        "FileSectorSource hint_sequential: windows stub (TODO: FILE_FLAG_SEQUENTIAL_SCAN at open)"
+    );
+}

src/io/mod.rs

@@ -10,11 +10,22 @@
 //! no-op pipeline — their default cache policies have not been shown
 //! to exhibit the same pathology for this access pattern.
 //!
+//! `FileSectorSource` is the read-side dual — it implements
+//! [`crate::sector::SectorSource`] for an ISO file with an internal
+//! 32 MiB read-ahead buffer that amortises NFS round-trip latency
+//! across thousands of sector reads.
+//!
 //! `Pipeline` + `Sink` (0.18) is the generic producer/consumer primitive
 //! used by sweep, patch, and mux to overlap reads with writes via a
 //! bounded channel + dedicated consumer thread.
+//!
+//! `byte_channel` is a byte-sized producer/consumer channel for the
+//! mux pipeline, sized to absorb worst-case input read stalls (see
+//! `freemkv-private/memory/project_buffering_architecture.md`).
 
 pub(crate) mod bounded;
+pub mod byte_channel;
+pub mod file_sector_source;
 mod writeback;
 mod writeback_file;
 

src/io/writeback_file/linux.rs

@@ -0,0 +1,66 @@
+//! Linux platform impl for [`super::WritebackFile`].
+//!
+//! - `preallocate`: `fallocate(FALLOC_FL_KEEP_SIZE)` — reserve extents
+//!   without growing the reported file size. Reduces extent
+//!   fragmentation on large sequential writes (mux output on NFS in
+//!   particular).
+//! - `durable_sync`: `fsync` wrapped in
+//!   [`crate::io::bounded::bounded_syscall`] with a 60 s deadline so a
+//!   wedged NFS server can't trap the calling thread indefinitely.
+
+use std::fs::File;
+use std::io;
+use std::os::unix::io::AsRawFd;
+use std::time::Duration;
+
+/// Pre-reserve extents for `size_bytes` of upcoming sequential writes.
+/// Best-effort: a non-zero rc is logged but not propagated, since the
+/// caller would just continue with the unreserved file anyway.
+pub(super) fn preallocate(file: &File, size_bytes: u64) {
+    // FALLOC_FL_KEEP_SIZE = 0x01 — keep the reported file size at 0
+    // (writes grow it normally) while still pre-reserving the extents.
+    let rc = unsafe {
+        libc::fallocate(
+            file.as_raw_fd(),
+            libc::FALLOC_FL_KEEP_SIZE,
+            0,
+            size_bytes as i64,
+        )
+    };
+    tracing::debug!(
+        target: "mux",
+        "WritebackFile fallocate size_hint={size_bytes} rc={rc} ok={}",
+        rc == 0
+    );
+}
+
+/// Run `fsync` on `file` with a 60 s deadline. On timeout we log loudly
+/// and return `Ok(())` — the kernel will still flush on close, so the
+/// data is best-effort durable; the alternative (trap the thread for
+/// the rest of the rip) defeats `/api/stop`.
+pub(super) fn durable_sync(file: &File) -> io::Result<()> {
+    let fd = file.as_raw_fd();
+    match crate::io::bounded::bounded_syscall(
+        None,
+        Duration::from_secs(60),
+        move || -> io::Result<()> {
+            let rc = unsafe { libc::fsync(fd) };
+            if rc == 0 {
+                Ok(())
+            } else {
+                Err(io::Error::last_os_error())
+            }
+        },
+    ) {
+        Ok(inner) => inner,
+        Err(crate::io::bounded::BoundedError::Timeout) => {
+            tracing::error!(
+                target: "mux",
+                "WritebackFile::sync_all fsync timed out after 60s; kernel will flush on close (best-effort)"
+            );
+            Ok(())
+        }
+        Err(crate::io::bounded::BoundedError::Halted) => Ok(()),
+        Err(crate::io::bounded::BoundedError::WorkerLost) => Ok(()),
+    }
+}

src/io/writeback_file/macos.rs

@@ -0,0 +1,104 @@
+//! macOS platform impl for [`super::WritebackFile`].
+//!
+//! - `preallocate`: `fcntl(F_PREALLOCATE)` — macOS's fallocate-equiv.
+//!   Reserves a contiguous extent when possible, falling back to a
+//!   non-contiguous reservation if the FS can't satisfy it. Reported
+//!   file size is unchanged (`F_ALLOCATEALL` is not set, so allocation
+//!   is "best effort up to length"; growth happens via writes).
+//! - `durable_sync`: `fcntl(F_FULLFSYNC)` wrapped in
+//!   [`crate::io::bounded::bounded_syscall`] with a 60 s deadline.
+//!   F_FULLFSYNC is HFS+/APFS's true-fsync (flushes the disk's own
+//!   write cache) — what `fsync` should have been on macOS. Falls back
+//!   to plain `fsync` if F_FULLFSYNC returns ENOTSUP.
+
+use std::fs::File;
+use std::io;
+use std::os::unix::io::AsRawFd;
+use std::time::Duration;
+
+/// libc `F_PREALLOCATE` — not exposed by the `libc` crate on all macOS
+/// SDK versions, so define it here.
+const F_PREALLOCATE: libc::c_int = 42;
+/// Allocate from current EOF.
+const F_PEOFPOSMODE: libc::c_int = 3;
+/// Hint: contiguous extent preferred.
+const F_ALLOCATECONTIG: libc::c_uint = 0x00000002;
+/// Allocate all the requested bytes (fall back to non-contig if needed).
+const F_ALLOCATEALL: libc::c_uint = 0x00000004;
+
+/// `fcntl(F_FULLFSYNC)` opcode. Documented in `man 2 fcntl` on macOS;
+/// not in the `libc` crate as a named constant.
+const F_FULLFSYNC: libc::c_int = 51;
+
+/// `fstore_t` layout matches `sys/fcntl.h`. Repr is C-stable so we can
+/// build it manually.
+#[repr(C)]
+struct Fstore {
+    fst_flags: libc::c_uint,
+    fst_posmode: libc::c_int,
+    fst_offset: libc::off_t,
+    fst_length: libc::off_t,
+    fst_bytesalloc: libc::off_t,
+}
+
+pub(super) fn preallocate(file: &File, size_bytes: u64) {
+    let mut fst = Fstore {
+        fst_flags: F_ALLOCATECONTIG | F_ALLOCATEALL,
+        fst_posmode: F_PEOFPOSMODE,
+        fst_offset: 0,
+        fst_length: size_bytes as libc::off_t,
+        fst_bytesalloc: 0,
+    };
+    // First attempt: contiguous.
+    let mut rc = unsafe { libc::fcntl(file.as_raw_fd(), F_PREALLOCATE, &mut fst) };
+    if rc == -1 {
+        // Fall back: drop the contiguous hint, allow scattered extents.
+        fst.fst_flags = F_ALLOCATEALL;
+        rc = unsafe { libc::fcntl(file.as_raw_fd(), F_PREALLOCATE, &mut fst) };
+    }
+    tracing::debug!(
+        target: "mux",
+        "WritebackFile F_PREALLOCATE size_hint={size_bytes} rc={rc} bytes_allocated={} ok={}",
+        fst.fst_bytesalloc,
+        rc != -1
+    );
+}
+
+pub(super) fn durable_sync(file: &File) -> io::Result<()> {
+    let fd = file.as_raw_fd();
+    match crate::io::bounded::bounded_syscall(
+        None,
+        Duration::from_secs(60),
+        move || -> io::Result<()> {
+            // Try F_FULLFSYNC first. If it isn't supported on this
+            // filesystem (older HFS, some network mounts) fall back to
+            // plain fsync — better than nothing.
+            let rc = unsafe { libc::fcntl(fd, F_FULLFSYNC, 0) };
+            if rc == 0 {
+                return Ok(());
+            }
+            let err = io::Error::last_os_error();
+            if err.raw_os_error() == Some(libc::ENOTSUP) {
+                let rc = unsafe { libc::fsync(fd) };
+                if rc == 0 {
+                    Ok(())
+                } else {
+                    Err(io::Error::last_os_error())
+                }
+            } else {
+                Err(err)
+            }
+        },
+    ) {
+        Ok(inner) => inner,
+        Err(crate::io::bounded::BoundedError::Timeout) => {
+            tracing::error!(
+                target: "mux",
+                "WritebackFile::sync_all F_FULLFSYNC timed out after 60s; kernel will flush on close (best-effort)"
+            );
+            Ok(())
+        }
+        Err(crate::io::bounded::BoundedError::Halted) => Ok(()),
+        Err(crate::io::bounded::BoundedError::WorkerLost) => Ok(()),
+    }
+}

src/io/writeback_file/mod.rs

@@ -20,6 +20,34 @@
 //!
 //! See `super::writeback::linux` for the underlying pathology and the
 //! strategy.
+//!
+//! ## Platform split
+//!
+//! The platform-specific pieces of this wrapper — extent preallocation
+//! (Linux `fallocate(KEEP_SIZE)`, macOS `F_PREALLOCATE`, Windows
+//! `SetFileValidData`) and the durable-flush primitive (Linux/macOS
+//! `fsync`/`F_FULLFSYNC` wrapped in a bounded syscall, Windows
+//! `FlushFileBuffers`) — live in per-OS sibling modules. The dispatch
+//! happens once at the bottom of this file via cfg-gated `mod` decls.
+//! No inline `#[cfg(target_os = "...")]` in the business-logic above.
+
+#[cfg(target_os = "linux")]
+mod linux;
+#[cfg(target_os = "macos")]
+mod macos;
+#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
+mod other;
+#[cfg(target_os = "windows")]
+mod windows;
+
+#[cfg(target_os = "linux")]
+use linux as platform;
+#[cfg(target_os = "macos")]
+use macos as platform;
+#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
+use other as platform;
+#[cfg(target_os = "windows")]
+use windows as platform;
 
 use std::fs::{File, OpenOptions};
 use std::io::{self, Seek, SeekFrom, Write};
@@ -65,43 +93,20 @@ impl WritebackFile {
     }
 
     /// Like [`Self::create`] but pre-reserves `size_bytes` of disk
-    /// space via `fallocate(FALLOC_FL_KEEP_SIZE)` on Linux. The
+    /// space via the platform's extent-preallocation primitive (Linux
-    /// reported file size is unchanged (writes still grow the file
+    /// `fallocate(KEEP_SIZE)`, macOS `F_PREALLOCATE`, Windows
-    /// naturally) — only the on-disk extent allocation is preallocated,
+    /// `SetFileValidData` stub). The reported file size is unchanged
-    /// which reduces extent fragmentation on large sequential writes
+    /// (writes still grow the file naturally) — only the on-disk extent
-    /// (mux output, especially on slow storage / NFS).
+    /// allocation is preallocated, which reduces extent fragmentation
+    /// on large sequential writes (mux output, especially on slow
+    /// storage / NFS).
     ///
-    /// On macOS / Windows the size hint is ignored and this is
+    /// On platforms without an extent-preallocation primitive this is
-    /// equivalent to `create`.
+    /// equivalent to `create` — the size hint is dropped after a debug
+    /// log.
     pub(crate) fn create_with_size_hint(path: &Path, size_bytes: u64) -> io::Result<Self> {
         let file = File::create(path)?;
-        #[cfg(target_os = "linux")]
+        platform::preallocate(&file, size_bytes);
-        {
-            use std::os::unix::io::AsRawFd;
-            // FALLOC_FL_KEEP_SIZE = 0x01 — keep the reported file
-            // size at 0 (writes grow it normally) while still
-            // pre-reserving the extents.
-            let rc = unsafe {
-                libc::fallocate(
-                    file.as_raw_fd(),
-                    libc::FALLOC_FL_KEEP_SIZE,
-                    0,
-                    size_bytes as i64,
-                )
-            };
-            tracing::debug!(
-                target: "mux",
-                "WritebackFile fallocate size_hint={size_bytes} rc={rc} ok={}",
-                rc == 0
-            );
-        }
-        #[cfg(not(target_os = "linux"))]
-        {
-            tracing::debug!(
-                target: "mux",
-                "WritebackFile fallocate size_hint={size_bytes} skipped (non-linux)"
-            );
-        }
         Self::new(file)
     }
 
@@ -118,7 +123,8 @@ impl WritebackFile {
     /// place of `File::sync_all`.
     ///
     /// The final fsync is wrapped in
-    /// [`crate::io::bounded::bounded_syscall`] with a 60 s deadline.
+    /// [`crate::io::bounded::bounded_syscall`] with a 60 s deadline on
+    /// platforms that have a usable bounded primitive (Linux + macOS).
     /// fsync on a wedged NFS server (or a degraded local disk) can
     /// hang the calling thread; the wrapper ensures the worst case is
     /// 60 s + log-and-continue rather than indefinite. On timeout the
@@ -126,39 +132,7 @@ impl WritebackFile {
     /// best effort, but bounded.
     pub(crate) fn sync_all(&mut self) -> io::Result<()> {
         self.pipeline.finalize();
-        #[cfg(unix)]
+        platform::durable_sync(&self.file)
-        {
-            use std::os::unix::io::AsRawFd;
-            use std::time::Duration;
-            let fd = self.file.as_raw_fd();
-            match crate::io::bounded::bounded_syscall(
-                None,
-                Duration::from_secs(60),
-                move || -> io::Result<()> {
-                    let rc = unsafe { libc::fsync(fd) };
-                    if rc == 0 {
-                        Ok(())
-                    } else {
-                        Err(io::Error::last_os_error())
-                    }
-                },
-            ) {
-                Ok(inner) => inner,
-                Err(crate::io::bounded::BoundedError::Timeout) => {
-                    tracing::error!(
-                        target: "mux",
-                        "WritebackFile::sync_all fsync timed out after 60s; kernel will flush on close (best-effort)"
-                    );
-                    Ok(())
-                }
-                Err(crate::io::bounded::BoundedError::Halted) => Ok(()),
-                Err(crate::io::bounded::BoundedError::WorkerLost) => Ok(()),
-            }
-        }
-        #[cfg(not(unix))]
-        {
-            self.file.sync_all()
-        }
     }
 }
 

src/io/writeback_file/other.rs

@@ -0,0 +1,23 @@
+//! Fallback platform impl for [`super::WritebackFile`] on targets
+//! without a dedicated implementation (BSDs, illumos, etc.).
+//!
+//! - `preallocate` is a logged no-op.
+//! - `durable_sync` calls `File::sync_all` directly (no bounded-syscall
+//!   wrapper — the wrapper depends on Linux/macOS unix idioms that
+//!   aren't universally portable). If a future BSD impl needs the
+//!   60-s deadline, it should land in its own per-OS file rather than
+//!   bloat this fallback.
+
+use std::fs::File;
+use std::io;
+
+pub(super) fn preallocate(_file: &File, size_bytes: u64) {
+    tracing::debug!(
+        target: "mux",
+        "WritebackFile preallocate size_hint={size_bytes} skipped (no impl on this target)"
+    );
+}
+
+pub(super) fn durable_sync(file: &File) -> io::Result<()> {
+    file.sync_all()
+}

src/io/writeback_file/windows.rs

@@ -0,0 +1,32 @@
+//! Windows platform impl for [`super::WritebackFile`].
+//!
+//! TODO: this stub matches the design's "validate without a Windows
+//! build env, leave a stub" carve-out. The real impl should use:
+//!
+//! - `SetEndOfFile` + `SetFileValidData` for extent preallocation
+//!   (caller needs `SE_MANAGE_VOLUME_NAME` privilege; if unavailable
+//!   fall back to a write-zero path or just skip).
+//! - `FlushFileBuffers` for fsync-equivalent durable flush.
+//!
+//! Until then: preallocate is a debug-logged no-op; durable_sync calls
+//! the std `File::sync_all` (which on Windows maps to
+//! `FlushFileBuffers` internally).
+
+use std::fs::File;
+use std::io;
+
+pub(super) fn preallocate(_file: &File, size_bytes: u64) {
+    tracing::debug!(
+        target: "mux",
+        "WritebackFile preallocate size_hint={size_bytes} skipped (windows stub; TODO: SetFileValidData)"
+    );
+}
+
+pub(super) fn durable_sync(file: &File) -> io::Result<()> {
+    // `File::sync_all` on Windows is `FlushFileBuffers`. Acceptable
+    // for now; the bounded-syscall wrapper is not used here because
+    // the stub also skips the worker-thread + leak machinery (the
+    // wrapper would need an `unsafe impl Send` for `RawHandle`, and
+    // designing that without a Windows test env is asking for it).
+    file.sync_all()
+}

src/sector/file.rs

@@ -1,148 +1,19 @@
-//! File-backed sector I/O — read and write 2048-byte sectors against
+//! File-backed sector sink — write 2048-byte sectors to an ISO image
-//! an ISO image on disk.
+//! on disk.
 //!
-//! [`FileSectorSource`] is the read side (open-only). [`FileSectorSink`]
+//! The read-side counterpart ([`crate::io::file_sector_source::FileSectorSource`])
-//! is the write side (create or open-rw); writes go through
+//! lives under `io/` because its internals (read-ahead buffer, per-OS
-//! [`crate::io::WritebackFile`] so big sequential ISO writes share
+//! `fadvise`/`F_RDADVISE` hints) are I/O infrastructure rather than
-//! the same bounded-cache writeback pipeline used by sweep / patch /
+//! sector-trait business logic. Both types remain re-exported at
-//! mux.
+//! [`crate::sector`] for ergonomic imports.
 
-use std::fs::{File, OpenOptions};
+use std::fs::OpenOptions;
-use std::io::{Read, Seek, SeekFrom, Write};
+use std::io::{Seek, SeekFrom, Write};
 use std::path::Path;
 
 use crate::error::{Error, Result};
 
-use super::{SectorSink, SectorSource};
+use super::SectorSink;
-
-/// Bytes-read threshold per `posix_fadvise(DONTNEED)` drop on the
-/// read side. Mirrors the writeback chunk size so the read-side
-/// page cache stays bounded the same way the write side does.
-#[cfg(target_os = "linux")]
-const READ_DROP_CHUNK_BYTES: u64 = 32 * 1024 * 1024;
-
-/// SectorSource backed by a file (ISO image).
-///
-/// Seeks to `lba * 2048`, reads `count * 2048` bytes per call. The
-/// file is held directly: every `read_sectors` call performs an
-/// absolute seek, so a wrapping `BufReader` would have its buffer
-/// invalidated on every call (its internal cursor moves with the
-/// `Seek` impl) — pure overhead. Callers that benefit from buffered
-/// reads should compose their own `BufReader` at the `read_sectors`
-/// granularity they care about.
-pub struct FileSectorSource {
-    file: File,
-    capacity: u32,
-    /// Bytes read since the last `posix_fadvise(DONTNEED)` drop.
-    /// Only updated on Linux; on other targets it stays at 0.
-    #[cfg(target_os = "linux")]
-    bytes_read_since_drop: u64,
-    /// Byte offset at which the current drop window starts (the
-    /// next `posix_fadvise(DONTNEED)` call drops from here).
-    #[cfg(target_os = "linux")]
-    drop_window_start: u64,
-}
-
-impl FileSectorSource {
-    /// Open an existing ISO file for reading. Capacity is derived
-    /// from `metadata().len() / 2048`. Returns
-    /// [`Error::IsoTooLarge`] if the file would exceed the 32-bit
-    /// LBA address space (~8 TB).
-    pub fn open(path: &Path) -> std::io::Result<Self> {
-        let file = File::open(path)?;
-        let len = file.metadata()?.len();
-        let sectors = len / 2048;
-        if sectors > u32::MAX as u64 {
-            return Err(Error::IsoTooLarge {
-                path: path.to_string_lossy().into_owned(),
-            }
-            .into());
-        }
-        let capacity = sectors as u32;
-
-        // Hint sequential access on Linux so the kernel's readahead
-        // window widens for the ISO sweep. Best-effort: return value
-        // is ignored. On macOS / Windows this is a no-op.
-        #[cfg(target_os = "linux")]
-        {
-            use std::os::unix::io::AsRawFd;
-            unsafe {
-                libc::posix_fadvise(file.as_raw_fd(), 0, 0, libc::POSIX_FADV_SEQUENTIAL);
-            }
-        }
-
-        Ok(Self {
-            file,
-            capacity,
-            #[cfg(target_os = "linux")]
-            bytes_read_since_drop: 0,
-            #[cfg(target_os = "linux")]
-            drop_window_start: 0,
-        })
-    }
-}
-
-// Implement the legacy `SectorSource` trait. The blanket impl in
-// `super` produces the `SectorSource` impl automatically — no need
-// to write both, and writing both would conflict. This keeps the
-// 0.17 method-resolution path intact (callers with `SectorSource`
-// in scope can still write `fsr.read_sectors(..)` against a
-// `FileSectorSource`).
-impl SectorSource for FileSectorSource {
-    fn capacity_sectors(&self) -> u32 {
-        self.capacity
-    }
-
-    fn read_sectors(
-        &mut self,
-        lba: u32,
-        count: u16,
-        buf: &mut [u8],
-        _recovery: bool,
-    ) -> Result<usize> {
-        let offset = lba as u64 * 2048;
-        let bytes = count as usize * 2048;
-        self.file
-            .seek(SeekFrom::Start(offset))
-            .map_err(|e| Error::IoError { source: e })?;
-        self.file
-            .read_exact(&mut buf[..bytes])
-            .map_err(|e| Error::IoError { source: e })?;
-
-        // On Linux, periodically drop the just-read region from the
-        // page cache to keep cache pressure bounded during multi-GB
-        // sequential ISO reads. Mirrors the write-side pipeline.
-        #[cfg(target_os = "linux")]
-        {
-            use std::os::unix::io::AsRawFd;
-            self.bytes_read_since_drop += bytes as u64;
-            if self.bytes_read_since_drop >= READ_DROP_CHUNK_BYTES {
-                let drop_start = self.drop_window_start;
-                let drop_len = self.bytes_read_since_drop;
-                let t0 = std::time::Instant::now();
-                unsafe {
-                    libc::posix_fadvise(
-                        self.file.as_raw_fd(),
-                        drop_start as i64,
-                        drop_len as i64,
-                        libc::POSIX_FADV_DONTNEED,
-                    );
-                }
-                let elapsed_ms = t0.elapsed().as_millis();
-                let start_lba = drop_start / 2048;
-                let end_lba = (drop_start + drop_len) / 2048;
-                tracing::trace!(
-                    target: "mux",
-                    "FileSectorSource fadvise DONTNEED lba=[{start_lba}..{end_lba}) bytes={drop_len} elapsed_ms={elapsed_ms}"
-                );
-                self.drop_window_start = drop_start + drop_len;
-                self.bytes_read_since_drop = 0;
-            }
-        }
-
-        Ok(bytes)
-    }
-}
 
 /// SectorSink backed by a file (ISO image).
 ///
@@ -211,13 +82,8 @@ impl SectorSink for FileSectorSink {
 
 #[cfg(test)]
 mod tests {
-    // Bring the 0.18 trait into scope (not super::*: the super
+    use super::FileSectorSink;
-    // module also re-exports the legacy `SectorSource`, and
+    use crate::io::file_sector_source::FileSectorSource;
-    // having both `SectorSource::read_sectors` and
-    // `SectorSource::read_sectors` visible would force every
-    // call site to disambiguate). External consumers see the
-    // same surface this test exercises.
-    use super::{FileSectorSink, FileSectorSource};
     use crate::sector::{SectorSink, SectorSource};
     use tempfile::tempdir;
 

src/sector/mod.rs

@@ -109,5 +109,6 @@ pub trait SectorSink: Send {
     fn finish(self: Box<Self>) -> Result<()>;
 }
 
+pub use crate::io::file_sector_source::FileSectorSource;
 pub use decrypting::DecryptingSectorSource;
-pub use file::{FileSectorSink, FileSectorSource};
+pub use file::FileSectorSink;