v0.23.0: bump version

Three coupled changes that together meet the "stable max throughput
on any medium" bar:

1. Remove `is_nfs` from `skip_wait`. WAIT_AFTER runs on every medium
   now. The `bounded_syscall` 30s safety net (already in place) covers
   the wedged-FS case generically — no need to predict NFS-hangs at
   compile time. Re-applies the 0.21.12 fix that landed the flat
   25 MB/s on NFS in the first place.

2. Drop the `posix_fadvise(DONTNEED)` calls after WAIT_AFTER. Bounded
   *dirty* pages (the actual invariant) does not require evicting
   *clean* pages. DONTNEED was forcing read-modify-write on any
   in-window seek-then-write — exactly the pattern matroska
   cluster-size backpatches produce. Empirical 2026-05-15: write side
   of mux fed NFS at 43 MB/s while the file grew at 25 MB/s, an 18
   MB/s overhead almost certainly composed of those RMW cycles. The
   kernel reclaims clean pages under LRU when memory is actually
   needed — we don't have to ask.

3. Replace `detect_nfs(fd) -> bool` with `detect_storage_class(fd)
   -> (StorageClass, chunk_bytes_seed)`. Medium detection happens
   *once* at construction and produces *one* output: the initial
   `chunk_bytes` seed for the autotuner. NFS gets 64 MiB (commit ack
   ~10-30 ms; needs bigger chunks to amortize); other media use the
   caller's hint. No hot-path branches on medium. The autotuner drives
   all subsequent decisions from measured WAIT_AFTER p95 latency,
   identically on every (OS, FS) combination.

Module-level doc rewritten to match: no more "NFS escape hatch", no
"is_nfs" framing. The whole writeback module is now medium-agnostic
except for one labelled detection point.

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "libfreemkv"
-version = "0.21.2"
+version = "0.23.0"
 edition = "2024"
 rust-version = "1.86"
 license = "AGPL-3.0-only"
@@ -28,6 +28,12 @@ base64 = "0.22.1"
 # under project docs ("Acceptable strings: debug/trace logging"). Consumers (autorip)
 # wire a tracing subscriber and pipe events into the JSONL debug log.
 tracing = "0.1"
+# Bounded MPSC channel with kernel-wakeup send_timeout. Used by `io::pipeline`
+# so the halt-aware send/finish loops can BLOCK on consumer drain instead of
+# polling — the 50 ms poll cadence of the previous mpsc-based impl capped mux
+# throughput at ~1 MB/s (see (internal)/memory/
+# feedback_send_with_halt_poll_throttle.md, 0.21.7).
+crossbeam-channel = "0.5"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 libc = "0.2"

src/io/file_sector_source/linux.rs

@@ -1,18 +1,46 @@
-//! Linux: kernel readahead hint for the ISO file.
+//! Linux read-side platform hooks: sequential-access hint at open +
+//! periodic page-cache eviction during streaming reads.
 //!
-//! Originally `POSIX_FADV_SEQUENTIAL` to widen the readahead window.
-//! On NFS that turned out to cause aggressive multi-MB readahead bursts
-//! that saturated the TCP connection and starved concurrent writes
-//! during mux — observed empirically as a ~3× drop in mux throughput
-//! on the rip1/unraid-1 setup (0.21.0 vs 0.20.7 baseline). The kernel's
-//! default readahead (~128 KiB on Linux) interleaves more naturally
-//! with the muxer's concurrent NFS writes, so we no longer issue any
-//! hint here. The per-OS file stays so the convention is honoured and
-//! we can re-enable a hint cleanly if a different storage path benefits.
+//! ## Why both
+//!
+//! `POSIX_FADV_SEQUENTIAL` at open widens the kernel's readahead window
+//! so each pread aggregates into fewer NFS round-trips. `DONTNEED` on
+//! the consumed window (called periodically by the caller) drops the
+//! already-read pages from the page cache so an 85 GB streaming ISO
+//! read doesn't fill memory and starve concurrent writes (the MKV
+//! output during mux). Together they mirror the write-side
+//! WritebackPipeline's policy.
+//!
+//! ## History
+//!
+//! Pre-Phase-1 (0.20.7 baseline) had both. Phase 1's introduction of
+//! `FileSectorSource` silently dropped the read-side DONTNEED, and
+//! 0.21.2's revert of `SEQUENTIAL` (mistakenly attributing a regression
+//! to it) removed the hint. Net effect: 85 GB of ISO reads pinned in
+//! the page cache + no readahead widening → mux throughput collapse
+//! from 18 MB/s historical to 2.7-8 MB/s on 0.21.x. Restored in 0.21.6.
 
 use std::fs::File;
+use std::os::unix::io::AsRawFd;
 
-pub(super) fn hint_sequential(_file: &File, _len_bytes: u64) {
-    // No-op: see module-level comment. Kernel default readahead is
-    // what we want on NFS-backed ISOs, which is the dominant case.
+pub(super) fn hint_sequential(file: &File, _len_bytes: u64) {
+    // Best-effort: return value ignored. A fadvise failure has no
+    // user-observable consequence.
+    unsafe {
+        libc::posix_fadvise(file.as_raw_fd(), 0, 0, libc::POSIX_FADV_SEQUENTIAL);
+    }
+}
+
+/// Drop pages in the half-open byte range `[start, start+len)` from
+/// the page cache. Called periodically by `read_sectors` to bound the
+/// read-side page cache pressure.
+pub(super) fn drop_window(file: &File, start: u64, len: u64) {
+    unsafe {
+        libc::posix_fadvise(
+            file.as_raw_fd(),
+            start as i64,
+            len as i64,
+            libc::POSIX_FADV_DONTNEED,
+        );
+    }
 }

src/io/file_sector_source/macos.rs

@@ -36,3 +36,11 @@ pub(super) fn hint_sequential(file: &File, len_bytes: u64) {
         libc::fcntl(file.as_raw_fd(), F_RDADVISE, &mut ra);
     }
 }
+
+/// macOS has no direct `POSIX_FADV_DONTNEED` equivalent for a byte
+/// range. `fcntl(F_NOCACHE)` would disable caching globally on the fd
+/// (too coarse — we want the unread region to still benefit). Best
+/// approximation: no-op. macOS's unified buffer cache is generally
+/// less prone to the pin-everything pathology that triggers the
+/// regression on Linux NFS clients.
+pub(super) fn drop_window(_file: &File, _start: u64, _len: u64) {}

src/io/file_sector_source/mod.rs

@@ -87,24 +87,39 @@ const BUF_SECTORS: u32 = (READAHEAD_BUF_BYTES / SECTOR_SIZE) as u32;
 /// `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.
+/// Bytes-read threshold per `posix_fadvise(DONTNEED)` drop on the
+/// read side. Mirrors `WRITEBACK_CHUNK_BYTES` so the read-side page
+/// cache stays bounded the same way the write side does.
+///
+/// 0.21.6: re-added after empirical discovery that Phase 1 had silently
+/// dropped this from the pre-Phase-1 (0.20.7) hot path. Without it,
+/// 85 GB of streaming ISO reads pin the entire file in the kernel page
+/// cache, starving the MKV writeback and collapsing mux throughput
+/// (observed: 2.7 MB/s mux on 0.21.5 vs. 70 MB/s isolated NFS reads).
+const READ_DROP_CHUNK_BYTES: u64 = 32 * 1024 * 1024;
+
 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`].
+    /// 0.21.3+: the app-level buffer is no longer touched on the hot
+    /// path (every `read_sectors` is a direct pread). The fields are
+    /// retained so a future per-source-type policy (e.g. a local-disk
+    /// source where batched reads ARE beneficial) can re-enable
+    /// buffering cleanly without re-plumbing the struct.
+    #[allow(dead_code)]
     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.
+    #[allow(dead_code)]
     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,
+    /// 0.21.6: bytes read since the last DONTNEED drop. Drives the
+    /// per-`READ_DROP_CHUNK_BYTES` page-cache eviction in read_sectors.
+    bytes_read_since_drop: u64,
+    /// 0.21.6: file offset at which the current drop window starts.
+    /// The next DONTNEED drops from `drop_window_start` for
+    /// `bytes_read_since_drop` bytes.
+    drop_window_start: u64,
 }
 
 impl FileSectorSource {
@@ -145,11 +160,14 @@ impl FileSectorSource {
             buf,
             buf_start_lba: 0,
             buf_len_sectors: 0,
+            bytes_read_since_drop: 0,
+            drop_window_start: 0,
         })
     }
 
     /// True if `[lba, lba + count)` is wholly inside the current
     /// buffer window. `count == 0` is vacuously true.
+    #[allow(dead_code)]
     fn buffer_covers(&self, lba: u32, count: u32) -> bool {
         if self.buf_len_sectors == 0 {
             return false;
@@ -165,6 +183,7 @@ impl FileSectorSource {
     /// 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`.
+    #[allow(dead_code)]
     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
@@ -207,33 +226,43 @@ impl SectorSource for FileSectorSource {
         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)?;
+        // 0.21.3: bypass the application-level buffer entirely.
+        //
+        // Empirically the 32 MiB readahead window (0.21.0–0.21.1) and the
+        // 4 MiB shrink (0.21.2) both regressed mux throughput vs the
+        // pre-Phase-1 0.20.7 baseline on NFS bidirectional workloads
+        // (sweep ~25 MB/s OK; mux dropped from 18 → 7-8 → 5-6 MB/s).
+        // Direct pread per call lets the kernel's own readahead policy
+        // run, which interleaves naturally with concurrent NFS writes on
+        // the same TCP connection.
+        //
+        // Buffer fields are retained (currently unused on this path) so
+        // any future per-source policy can be reintroduced without
+        // re-plumbing structure. `refill` / `buffer_covers` are kept too
+        // (still exercised by the tests so the API contract is locked).
+        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 })?;
+        self.buf_len_sectors = 0;
+
+        // 0.21.6: periodic page-cache eviction on the read side. Without
+        // this, an 85 GB streaming ISO read pins the entire file in
+        // kernel page cache, which starves concurrent NFS writes (the
+        // MKV output) and collapses mux throughput. Mirrors the
+        // write-side WritebackPipeline's DONTNEED policy.
+        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;
+            platform::drop_window(&self.file, drop_start, drop_len);
+            self.drop_window_start = drop_start + drop_len;
+            self.bytes_read_since_drop = 0;
         }
 
-        // 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)
     }
 }
@@ -295,14 +324,14 @@ mod tests {
 
         let mut src = FileSectorSource::open(&path).unwrap();
 
-        // Prime: read sector 0 to populate buffer #0.
+        // Prime: read sector 0. (0.21.3+: app-level buffer is bypassed,
+        // so we don't assert internal buf state here — just exercise
+        // the read path.)
         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.
+        // Now read 4 sectors crossing what used to be the buffer
+        // boundary. Still a valid SectorSource-contract test.
         let span_lba = BUF_SECTORS - 2;
         let mut buf4 = vec![0u8; SECTOR_SIZE * 4];
         src.read_sectors(span_lba, 4, &mut buf4, false).unwrap();
@@ -330,13 +359,11 @@ mod tests {
         // 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.
+        // Backward to sector 0. (0.21.3+: app-level buffer is bypassed
+        // so we only assert the byte-level contract, not internal
+        // buffer state.)
         src.read_sectors(0, 1, &mut got, false).unwrap();
-        assert_eq!(src.buf_start_lba, 0);
         assert!(got.iter().all(|b| *b == 0));
     }
 
@@ -355,11 +382,12 @@ mod tests {
         assert_eq!(src.capacity_sectors(), total);
 
         let mut got = vec![0u8; SECTOR_SIZE];
-        // First read triggers refill clamped to `total`.
+        // First read at sector 0.
         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.
+        // Read the very last sector. (0.21.3+: app-level buffer is
+        // bypassed; the test still verifies that EOF-region reads
+        // return correct bytes.)
         src.read_sectors(total - 1, 1, &mut got, false).unwrap();
         let expected = ((total - 1) & 0xff) as u8;
         assert!(got.iter().all(|b| *b == expected));

src/io/file_sector_source/other.rs

@@ -5,3 +5,5 @@
 use std::fs::File;
 
 pub(super) fn hint_sequential(_file: &File, _len_bytes: u64) {}
+
+pub(super) fn drop_window(_file: &File, _start: u64, _len: u64) {}

src/io/file_sector_source/windows.rs

@@ -17,3 +17,8 @@ pub(super) fn hint_sequential(_file: &File, _len_bytes: u64) {
         "FileSectorSource hint_sequential: windows stub (TODO: FILE_FLAG_SEQUENTIAL_SCAN at open)"
     );
 }
+
+/// Windows page-cache eviction is not exposed via a posix_fadvise
+/// equivalent. The kernel does its own working-set management. No-op
+/// for now.
+pub(super) fn drop_window(_file: &File, _start: u64, _len: u64) {}

src/io/pipeline.rs

@@ -34,10 +34,11 @@
 //! consumer lag detection). This is critical for diagnosing stalls.
 
 use std::io;
-use std::sync::mpsc::{SyncSender, TrySendError, sync_channel};
 use std::thread::{self, JoinHandle};
 use std::time::{Duration, Instant};
 
+use crossbeam_channel::{Sender, TrySendError, bounded};
+
 use crate::error::Error;
 use crate::halt::Halt;
 
@@ -57,12 +58,21 @@ pub const JOIN_TIMEOUT_SECS: u64 = 600;
 /// responsive across both primitives.
 const POLL_INTERVAL: Duration = Duration::from_millis(250);
 
-/// Polling slice for the halt-aware send loop. Smaller than
-/// [`POLL_INTERVAL`] because send latency tolerance is much lower —
-/// frames must move through the channel at hundreds-of-Hz on the
-/// happy path; 50 ms keeps backpressure-driven wakeups fine-grained
-/// without busy-looping.
-const SEND_POLL_INTERVAL: Duration = Duration::from_millis(50);
+/// Halt-check cadence for the send loop. Producer blocks on
+/// [`crossbeam_channel::Sender::send_timeout`] for this slice — the
+/// kernel wakes it the instant the consumer drains a slot, so on the
+/// happy path there's no throughput cap from this primitive at all
+/// (the cap is whatever the underlying medium can sustain). When the
+/// consumer is genuinely wedged, the timeout fires every 250 ms and
+/// the producer checks the halt token; that's the latency a stop
+/// request will observe.
+///
+/// 0.21.7 replaced an old `std::sync::mpsc::sync_channel` + 50 ms
+/// `thread::sleep` polling loop that capped mux throughput at
+/// ~20 frames/sec ≈ 1 MB/s on saturated channels. See
+/// (internal)/memory/feedback_send_with_halt_poll_throttle.md
+/// for the multi-day diagnostic that surfaced it.
+const SEND_HALT_CHECK_INTERVAL: Duration = Duration::from_millis(250);
 
 /// Check if verbose debug logging is enabled via FREEMKV_DEBUG env var.
 pub fn debug_enabled() -> bool {
@@ -135,7 +145,7 @@ pub trait Sink<I>: Send + 'static {
 /// Bounded producer/consumer pipeline. Holds the producer-side
 /// channel and the consumer thread's join handle.
 pub struct Pipeline<I: Send + 'static, R: Send + 'static> {
-    tx: SyncSender<I>,
+    tx: Sender<I>,
     handle: JoinHandle<Result<R, Error>>,
 }
 
@@ -170,7 +180,7 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
         depth: usize,
         sink: S,
     ) -> Result<Self, Error> {
-        let (tx, rx) = sync_channel::<I>(depth);
+        let (tx, rx) = bounded::<I>(depth);
         let handle = thread::Builder::new()
             .name(name.into())
             .spawn(move || -> Result<R, Error> {
@@ -281,14 +291,20 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
     /// `Err`. Useful for best-effort signalling (e.g. sweep's
     /// throttled `StatsRequest`) where dropping the message is
     /// preferable to blocking the producer.
-    pub fn try_send(&self, item: I) -> Result<(), std::sync::mpsc::TrySendError<I>> {
+    pub fn try_send(&self, item: I) -> Result<(), TrySendError<I>> {
         self.tx.try_send(item)
     }
 
     /// Halt-aware bounded variant of [`Pipeline::send`].
     ///
-    /// Polls `try_send` on a 50 ms slice. Between slices, checks
-    /// (1) the [`Halt`] token and (2) the per-call `deadline`. Returns:
+    /// Uses [`crossbeam_channel::Sender::send_timeout`] so the producer
+    /// thread BLOCKS on consumer drain (kernel-wakeup) rather than
+    /// polling. The timeout slice is just the halt-observation cadence
+    /// ([`SEND_HALT_CHECK_INTERVAL`]) — on the happy path the producer
+    /// wakes the instant the consumer drains a slot, so there is no
+    /// throughput cap from this primitive at any medium speed.
+    ///
+    /// Returns:
     ///
     /// - `Ok(())` once the item lands in the channel.
     /// - `Err(item)` if the consumer disconnected, the halt fired, or
@@ -302,36 +318,45 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
     /// Unlike [`Pipeline::send`], this never blocks the producer
     /// thread inside an unkillable `mpsc::send` — if the consumer is
     /// wedged inside an unkillable syscall, the producer can still
-    /// observe `/api/stop` and unwind.
+    /// observe `/api/stop` and unwind within
+    /// [`SEND_HALT_CHECK_INTERVAL`].
     pub fn send_with_halt(&self, item: I, halt: &Halt, deadline: Duration) -> Result<(), I> {
+        use crossbeam_channel::SendTimeoutError;
         let end = Instant::now() + deadline;
         let mut pending = item;
         loop {
-            match self.tx.try_send(pending) {
-                Ok(()) => return Ok(()),
-                Err(TrySendError::Full(returned)) => {
-                    pending = returned;
-                    if halt.is_cancelled() {
-                        if debug_enabled() {
-                            tracing::debug!(
-                                "Pipeline send_with_halt: halt observed, returning item={}",
-                                std::any::type_name::<I>()
-                            );
-                        }
-                        return Err(pending);
-                    }
-                    if Instant::now() >= end {
-                        if debug_enabled() {
-                            tracing::debug!(
-                                "Pipeline send_with_halt: deadline elapsed, returning item={}",
-                                std::any::type_name::<I>()
-                            );
-                        }
-                        return Err(pending);
-                    }
-                    thread::sleep(SEND_POLL_INTERVAL);
+            // Pre-check the cheap exit conditions before parking.
+            if halt.is_cancelled() {
+                if debug_enabled() {
+                    tracing::debug!(
+                        "Pipeline send_with_halt: halt observed, returning item={}",
+                        std::any::type_name::<I>()
+                    );
                 }
-                Err(TrySendError::Disconnected(returned)) => {
+                return Err(pending);
+            }
+            let now = Instant::now();
+            if now >= end {
+                if debug_enabled() {
+                    tracing::debug!(
+                        "Pipeline send_with_halt: deadline elapsed, returning item={}",
+                        std::any::type_name::<I>()
+                    );
+                }
+                return Err(pending);
+            }
+            // Wait for space-available or halt-check tick, whichever
+            // is sooner. Crossbeam's send_timeout is kernel-wakeup
+            // based: the consumer's recv on a saturated channel
+            // signals this thread the moment a slot opens up.
+            let slice = SEND_HALT_CHECK_INTERVAL.min(end.saturating_duration_since(now));
+            match self.tx.send_timeout(pending, slice) {
+                Ok(()) => return Ok(()),
+                Err(SendTimeoutError::Timeout(returned)) => {
+                    pending = returned;
+                    // loop: re-check halt + deadline, then park again
+                }
+                Err(SendTimeoutError::Disconnected(returned)) => {
                     if debug_enabled() {
                         tracing::debug!(
                             "Pipeline send_with_halt: consumer disconnected, item={}",

src/io/writeback/linux.rs

@@ -22,34 +22,48 @@
 //! fast storage (NVMe) sees smaller chunks to keep cache pressure
 //! tight. Bounds: [4 MiB, 256 MiB].
 //!
-//! ## NFS escape hatch
+//! ## Medium-agnostic by construction
 //!
-//! `sync_file_range(WAIT_AFTER)` on an NFS-mounted file can block
-//! indefinitely waiting for the server's commit ack. If the server
-//! never acks (network partition, server-side hang, slow commit), the
-//! syscall never returns and the consumer thread is stuck inside the
-//! kernel — `/api/stop` can't reach it because halt is cooperative.
+//! `sync_file_range(WAIT_AFTER)` runs on every medium (local FS, NFS,
+//! etc.). It bounds the dirty-page set at ~2 × `chunk_bytes` by
+//! waiting for each chunk's kernel writeback to complete before
+//! advancing. There is no `if is_nfs` branch in the hot path. The
+//! safety net described below ([`WAIT_AFTER_TIMEOUT`]) handles wedged
+//! filesystems generically: if a particular FS proves unable to ack
+//! `WAIT_AFTER` within the deadline, the pipeline flips to `degraded`
+//! and skips for the rest of its life. That covers the original
+//! NFS-hang concern (network partition, server stuck on commit)
+//! without baking the failure mode into a per-FS branch.
 //!
-//! When `fstatfs` reports the file lives on an NFS mount
-//! (`f_type == NFS_SUPER_MAGIC`), the pipeline skips the WAIT_AFTER +
-//! `posix_fadvise(DONTNEED)` dance entirely. NFS clients have their
-//! own buffering and commit semantics that handle dirty-page bounds
-//! without us forcing the issue. The async `SYNC_FILE_RANGE_WRITE`
-//! kickoff still runs (non-blocking by spec) so writeback still gets
-//! a nudge.
+//! **No `posix_fadvise(DONTNEED)` after the wait.** Earlier revisions
+//! evicted the just-WAIT_AFTER'd range from the page cache on the
+//! theory that bounding *dirty* pages required also bounding *clean*
+//! pages. That was unnecessary and counterproductive: clean cached
+//! pages cost nothing (the kernel reclaims them under LRU when memory
+//! is needed) and DONTNEED forces a read-modify-write on any
+//! subsequent in-window write — exactly the pattern matroska
+//! cluster-size backpatches produce, costing ~40-50% of measured NFS
+//! write bandwidth in empirical tests 2026-05-15/16.
+//!
+//! Medium detection happens *exactly once*, at construction, and
+//! produces *exactly one* value: the initial `chunk_bytes` seed for
+//! the autotuner. The detector's job is to give the autotuner a
+//! decent starting point on cold start; from there p95 of WAIT_AFTER
+//! latency drives all subsequent decisions, identically on every
+//! medium.
 //!
 //! ## Defence in depth: WAIT_AFTER timeout
 //!
-//! Even on local storage, a degraded disk or odd filesystem driver
-//! could in principle wedge inside WAIT_AFTER. Each WAIT_AFTER call
-//! runs on a worker thread with a 30s recv_timeout on its result
-//! channel. On timeout we log a loud error, set a `degraded` flag,
-//! and from then on skip WAIT_AFTER + DONTNEED for the rest of the
-//! pipeline's life (same shape as the NFS path). The worker thread
-//! is intentionally leaked — it unwinds whenever the syscall
-//! eventually returns or the process exits. The mux continues; the
-//! original dirty-burst pathology re-emerges but the rip can still
-//! finish instead of freezing.
+//! A degraded disk, wedged NFS server, or odd FS driver could in
+//! principle hang inside WAIT_AFTER. Each call runs on a worker
+//! thread with a 30s recv_timeout on its result channel
+//! ([`crate::io::bounded::bounded_syscall`]). On timeout the pipeline
+//! flips to `degraded`, logs a loud error, and skips WAIT_AFTER for
+//! the rest of its life. The worker thread is intentionally leaked —
+//! it unwinds whenever the syscall eventually returns or the process
+//! exits. The mux continues; the original dirty-burst pathology
+//! re-emerges only on the specific (medium, server, FS) combination
+//! that actually wedged.
 
 use std::collections::VecDeque;
 use std::fs::File;
@@ -87,11 +101,9 @@ pub(crate) struct WritebackPipeline {
     /// Count of chunks emitted (used to space out periodic
     /// `debug!` size snapshots).
     chunk_count: u64,
-    /// True when the underlying file is on an NFS mount. NFS makes
-    /// WAIT_AFTER unsafe (can block forever on missing server ack), so
-    /// we skip it entirely and let the NFS client handle commit on
-    /// close.
-    is_nfs: bool,
+    /// What `detect_storage_class` saw at construction. Carried for
+    /// diagnostic logging only — *not* keyed off in the hot path.
+    storage_class: StorageClass,
     /// Set the first time WAIT_AFTER exceeds [`WAIT_AFTER_TIMEOUT`].
     /// Once set, behaviour matches the NFS path for the rest of the
     /// pipeline's life. Wrapped in `Arc` only because both this
@@ -108,13 +120,25 @@ impl WritebackPipeline {
     /// returned `WritebackPipeline` MUST be dropped before `file`
     /// itself, or kept inside the same struct that owns `file` — the
     /// alias is unchecked.
-    pub(crate) fn new(file: &File, start_pos: u64, chunk_bytes: u64) -> Self {
+    ///
+    /// `chunk_bytes_hint` is the caller's preferred starting chunk
+    /// size; the constructor overrides it with a medium-specific
+    /// initial seed if `detect_storage_class` recognises the fd's
+    /// filesystem. The autotuner then drives the value from there
+    /// based on measured `WAIT_AFTER` latency, identically on every
+    /// medium.
+    pub(crate) fn new(file: &File, start_pos: u64, chunk_bytes_hint: u64) -> Self {
         let fd = file.as_raw_fd();
-        let is_nfs = detect_nfs(fd);
+        let (class, seed) = detect_storage_class(fd);
+        // Medium-aware initial seed: bias toward bigger chunks on
+        // slow-commit media so the autotuner doesn't waste a minute
+        // climbing from a too-small starting value. A returned seed
+        // of 0 means "the detector has no opinion; use the caller's
+        // hint". All subsequent tuning is medium-agnostic.
+        let chunk_bytes = if seed > 0 { seed } else { chunk_bytes_hint };
         tracing::info!(
             target: "mux",
-            "WritebackPipeline fd={fd} is_nfs={is_nfs} chunk_bytes={chunk_bytes} strategy={}",
-            if is_nfs { "nfs-skip-wait" } else { "wait+dontneed" }
+            "WritebackPipeline fd={fd} storage_class={class:?} chunk_bytes={chunk_bytes}"
         );
         Self {
             fd,
@@ -123,17 +147,18 @@ impl WritebackPipeline {
             pending: None,
             wait_after_window: VecDeque::with_capacity(ADAPTIVE_WINDOW),
             chunk_count: 0,
-            is_nfs,
+            storage_class: class,
             degraded: Arc::new(AtomicBool::new(false)),
         }
     }
 
-    /// True if we should bypass the WAIT_AFTER + DONTNEED finalisation
-    /// step. NFS always bypasses; local storage bypasses once the
-    /// pipeline has flipped to degraded after a WAIT_AFTER timeout.
+    /// True if we should bypass the WAIT_AFTER finalisation step. The
+    /// pipeline only skips after `WAIT_AFTER` has *empirically* hung
+    /// past [`WAIT_AFTER_TIMEOUT`] on this particular fd. No per-FS
+    /// branch — the failure is detected, not predicted.
     #[inline]
     fn skip_wait(&self) -> bool {
-        self.is_nfs || self.degraded.load(Ordering::Relaxed)
+        self.degraded.load(Ordering::Relaxed)
     }
 
     /// Caller advanced the file position to `pos`. If a chunk boundary
@@ -156,40 +181,29 @@ impl WritebackPipeline {
         }
         if let Some((prev_off, prev_len)) = self.pending.take() {
             if self.skip_wait() {
-                // NFS branch (or degraded fallback after a prior
-                // timeout): the WAIT_AFTER + DONTNEED dance is what
-                // hangs on NFS — skip it. We still advance `pending`
-                // so the next call has a stable cycle.
+                // Degraded path (set only after an empirical
+                // WAIT_AFTER timeout on this fd). We still advance
+                // `pending` so the next call has a stable cycle.
             } else {
-                // Normal local-storage branch with belt-and-braces
-                // timeout. If WAIT_AFTER hangs > WAIT_AFTER_TIMEOUT
-                // we mark the pipeline degraded, log a loud error,
-                // and fall through to the skip path on subsequent
-                // calls.
                 match wait_after_with_timeout(self.fd, prev_off, prev_len) {
                     Some(ms) => {
                         wait_ms = ms;
-                        let t_fadv = Instant::now();
-                        unsafe {
-                            libc::posix_fadvise(
-                                self.fd,
-                                prev_off as i64,
-                                prev_len as i64,
-                                libc::POSIX_FADV_DONTNEED,
-                            );
-                        }
-                        fadvise_ms = t_fadv.elapsed().as_millis() as u64;
+                        // Deliberately NO `posix_fadvise(DONTNEED)`
+                        // here. See module-level docs: evicting clean
+                        // pages costs us read-modify-write on
+                        // matroska cluster backpatches and gains
+                        // nothing — the kernel will reclaim clean
+                        // pages under LRU when memory is needed.
                         self.record_wait(wait_ms);
                     }
                     None => {
-                        // Timeout branch: switch to NFS-style skip
-                        // for the rest of the pipeline's life. Do
-                        // NOT call DONTNEED — if WAIT_AFTER hasn't
-                        // returned, the pages aren't safely flushed.
+                        // Timeout branch: this fd's FS or server is
+                        // unable to ack WAIT_AFTER. Flip to skip mode
+                        // for the rest of the pipeline's life.
                         self.degraded.store(true, Ordering::Relaxed);
                         tracing::error!(
                             target: "mux",
-                            "WritebackPipeline WAIT_AFTER timed out after {}s on chunk off={} len={}, marking writeback degraded (subsequent chunks will skip WAIT_AFTER + DONTNEED)",
+                            "WritebackPipeline WAIT_AFTER timed out after {}s on chunk off={} len={}, marking writeback degraded (subsequent chunks will skip WAIT_AFTER)",
                             WAIT_AFTER_TIMEOUT.as_secs(),
                             prev_off,
                             prev_len
@@ -201,9 +215,10 @@ impl WritebackPipeline {
         self.pending = Some((chunk_off as u64, chunk_len as u64));
         self.last_flush_pos = pos;
         self.chunk_count += 1;
+        let _ = fadvise_ms;
         tracing::trace!(
             target: "mux",
-            "WritebackPipeline chunk off={} len={} sync_file_range_ms={wait_ms} fadvise_ms={fadvise_ms} chunk_bytes={} skip_wait={}",
+            "WritebackPipeline chunk off={} len={} sync_file_range_ms={wait_ms} chunk_bytes={} skip_wait={}",
             chunk_off,
             chunk_len,
             self.chunk_bytes,
@@ -212,10 +227,10 @@ impl WritebackPipeline {
         if self.chunk_count % SIZE_LOG_INTERVAL == 0 {
             tracing::debug!(
                 target: "mux",
-                "WritebackPipeline chunk_bytes={} after {} chunks is_nfs={} degraded={}",
+                "WritebackPipeline chunk_bytes={} after {} chunks storage_class={:?} degraded={}",
                 self.chunk_bytes,
                 self.chunk_count,
-                self.is_nfs,
+                self.storage_class,
                 self.degraded.load(Ordering::Relaxed),
             );
         }
@@ -267,46 +282,51 @@ impl WritebackPipeline {
         if let Some((prev_off, prev_len)) = self.pending.take() {
             tracing::debug!(
                 target: "mux",
-                "WritebackPipeline finalize chunk off={prev_off} len={prev_len} skip_wait={} is_nfs={} degraded={}",
+                "WritebackPipeline finalize chunk off={prev_off} len={prev_len} skip_wait={} storage_class={:?} degraded={}",
                 self.skip_wait(),
-                self.is_nfs,
+                self.storage_class,
                 self.degraded.load(Ordering::Relaxed),
             );
             if self.skip_wait() {
-                // NFS / degraded: skip WAIT_AFTER + DONTNEED. close()
-                // / sync_all() handle commit through their normal
-                // paths.
                 return;
             }
-            match wait_after_with_timeout(self.fd, prev_off, prev_len) {
-                Some(_ms) => unsafe {
-                    libc::posix_fadvise(
-                        self.fd,
-                        prev_off as i64,
-                        prev_len as i64,
-                        libc::POSIX_FADV_DONTNEED,
-                    );
-                },
-                None => {
-                    self.degraded.store(true, Ordering::Relaxed);
-                    tracing::error!(
-                        target: "mux",
-                        "WritebackPipeline finalize WAIT_AFTER timed out after {}s on chunk off={prev_off} len={prev_len}, marking writeback degraded",
-                        WAIT_AFTER_TIMEOUT.as_secs(),
-                    );
-                }
+            if wait_after_with_timeout(self.fd, prev_off, prev_len).is_none() {
+                self.degraded.store(true, Ordering::Relaxed);
+                tracing::error!(
+                    target: "mux",
+                    "WritebackPipeline finalize WAIT_AFTER timed out after {}s on chunk off={prev_off} len={prev_len}, marking writeback degraded",
+                    WAIT_AFTER_TIMEOUT.as_secs(),
+                );
             }
         }
     }
 }
 
-/// Probe whether `fd` lives on an NFS mount via `fstatfs`. Returns
-/// `false` on any error — we fail open, not closed: better to run the
-/// normal local-storage path on a misdetected NFS mount (and surface
-/// the freeze loudly via the timeout) than to needlessly disable
-/// writeback bounding on every local file because of a transient
-/// stat error.
-fn detect_nfs(fd: RawFd) -> bool {
+/// Coarse classification of the medium an open file is on. Used only
+/// at construction to seed the autotuner with a sensible initial
+/// `chunk_bytes`. Never branched on in the hot path.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum StorageClass {
+    /// `fstatfs.f_type == NFS_SUPER_MAGIC`. Slow-commit; seed a
+    /// generous initial chunk so the autotuner doesn't waste cycles
+    /// climbing from a too-small value.
+    Nfs,
+    /// Anything else we successfully stat'd. Local FS, tmpfs, ZFS,
+    /// network FS we don't have a constant for, etc. Use the caller's
+    /// hint as the seed.
+    Other,
+    /// `fstatfs` failed. Caller's hint wins.
+    Unknown,
+}
+
+/// Probe the FS class via `fstatfs` and return both the class label
+/// and a recommended `chunk_bytes` seed. `seed == 0` means "no
+/// medium-specific recommendation; use the caller's hint." Failure
+/// modes default to `Unknown` + seed 0 so the caller's hint applies.
+///
+/// This is the single medium-detection point in the whole writeback
+/// pipeline. Everything else is medium-agnostic.
+fn detect_storage_class(fd: RawFd) -> (StorageClass, u64) {
     // `libc::statfs` is repr(C) with a fixed layout; zeroing is the
     // documented init pattern for the kernel uapi struct.
     let mut buf: libc::statfs = unsafe { std::mem::zeroed() };
@@ -315,9 +335,9 @@ fn detect_nfs(fd: RawFd) -> bool {
         let errno = std::io::Error::last_os_error();
         tracing::warn!(
             target: "mux",
-            "WritebackPipeline fstatfs(fd={fd}) failed: {errno} — defaulting is_nfs=false",
+            "WritebackPipeline fstatfs(fd={fd}) failed: {errno} — defaulting to Unknown",
         );
-        return false;
+        return (StorageClass::Unknown, 0);
     }
     // `f_type` is signed (`__fsword_t`) on glibc and unsigned
     // (`c_ulong`) on musl. Cast both sides to i64 for a portable
@@ -328,7 +348,18 @@ fn detect_nfs(fd: RawFd) -> bool {
     let f_type = buf.f_type as i64;
     #[allow(clippy::unnecessary_cast)]
     let nfs_magic = libc::NFS_SUPER_MAGIC as i64;
-    f_type == nfs_magic
+    if f_type == nfs_magic {
+        // NFS commit ack typically ~10-30 ms RTT. The autotuner grows
+        // by doubling when p95 > 200 ms — so starting at 32 MiB it
+        // would never grow on a healthy NFS. Seed at 64 MiB so the
+        // commit cadence amortizes properly from the first chunk.
+        (StorageClass::Nfs, 64 * 1024 * 1024)
+    } else {
+        // Local FS / unknown remote FS. Caller's hint (typically 32
+        // MiB) is a fine starting point; autotuner adjusts from
+        // measured latency.
+        (StorageClass::Other, 0)
+    }
 }
 
 /// Run `sync_file_range(WAIT_AFTER)` on a worker thread and wait up

src/io/writeback_file/mod.rs

@@ -31,67 +31,22 @@
 //! happens once at the bottom of this file via cfg-gated `mod` decls.
 //! No inline `#[cfg(target_os = "...")]` in the business-logic above.
 //!
-//! ## Phase 2.5 — write-side flatness (writer thread)
+//! ## Write path
 //!
-//! `WritebackFile` is split into a thin muxer-facing handle and a
-//! dedicated writer thread that owns the real `File` + writeback
-//! pipeline. The muxer's `Write::write` and `Seek::seek` calls return as
-//! soon as the byte handoff to a bounded SPSC ring completes; the writer
-//! thread executes the real syscalls (incl. `sync_file_range(WAIT_AFTER)`
-//! on Linux) without ever blocking the muxer on a kernel commit.
+//! Writes are direct passthrough to the underlying `File` (no writer
+//! thread, no ring, no batching). Empirically the Phase-2.5
+//! writer-thread architecture introduced a ~60% mux throughput
+//! regression on NFS bidirectional workloads; reverting the write path
+//! to direct passthrough restores the 0.20.7 baseline. The writeback
+//! pipeline still runs (it's called inline from `write` / `write_all` /
+//! `seek`) so the bounded-cache invariant on Linux is preserved.
 //!
-//! ### Backpressure
+//! ## Halt-safety
 //!
-//! The ring is byte-bounded at [`RING_CAPACITY_BYTES`]. When the ring is
-//! full the muxer's `write` blocks on a condvar until the writer thread
-//! drains enough bytes to admit the next chunk. **Backpressure on
-//! ring-full deliberately blocks the muxer rather than dropping bytes**
-//! — archival workflows cannot afford byte loss, and the kernel page
-//! cache is already a second buffering layer underneath the writer
-//! thread.
-//!
-//! ### MKV seek-back semantics
-//!
-//! The MKV container backpatches cluster size headers shortly after
-//! emitting them. The writer thread maintains an
-//! [`ActiveClusterBuffer`] tracking the last
-//! [`ACTIVE_CLUSTER_WINDOW_BYTES`] bytes by absolute file position. When
-//! the muxer issues `Seek(pos)` for a `pos` inside that window, the
-//! writer issues a real `file.seek` (cheap) but **skips
-//! `pipeline.handle_seek()`** — no `sync_file_range(WAIT_AFTER)` drain
-//! is forced for the current chunk. This is the dominant case: every
-//! cluster backpatch is within the current 32 MiB writeback chunk.
-//!
-//! For seeks **outside** the window (rare — Cues index write at the end
-//! of mux, Segment header backpatch right before close) the writer
-//! falls back to the pre-Phase-2.5 behaviour: drain the in-flight
-//! writeback via `pipeline.handle_seek()`, then issue the real seek.
-//!
-//! ### Halt-safe
-//!
-//! All real `sync_file_range(WAIT_AFTER)` and `fsync` calls on the
-//! writer thread route through [`crate::io::bounded::bounded_syscall`]
-//! with a 60 s deadline (already in place in the per-OS modules pre-
-//! Phase-2.5). A wedged NFS server cannot freeze the writer thread
-//! indefinitely; the muxer keeps queueing into the ring; the kernel
-//! page cache absorbs.
-//!
-//! ### `sync_all` semantics
-//!
-//! `sync_all` is synchronous: it drains the ring through the writer
-//! thread, then runs the per-OS durable-flush primitive, and returns
-//! the result to the caller. This is the API contract — callers
-//! (sweep/patch consumers, mux finalisation) rely on it.
-//!
-//! ### `speed_mbs` reporting
-//!
-//! Speed measurements taken at the muxer side (bytes handed off into
-//! the ring) reflect ring-handoff throughput, **not** bytes committed
-//! to durable storage. This is the correct number for muxer flatness
-//! reporting; sweep/patch use mapfile-based progress which is unrelated.
-//! Autorip's UI is unaffected — speed is calculated outside this
-//! module — but the distinction is worth noting in release notes if the
-//! UI ever exposes "throughput vs commit rate" separately.
+//! `sync_all` runs the per-OS durable-flush primitive, which on
+//! Linux/macOS is wrapped in [`crate::io::bounded::bounded_syscall`]
+//! with a 60 s deadline. A wedged NFS server cannot trap the muxer
+//! indefinitely on the final fsync.
 
 #[cfg(target_os = "linux")]
 mod linux;
@@ -111,13 +66,9 @@ use other as platform;
 #[cfg(target_os = "windows")]
 use windows as platform;
 
-use std::collections::VecDeque;
 use std::fs::{File, OpenOptions};
 use std::io::{self, Seek, SeekFrom, Write};
 use std::path::Path;
-use std::sync::mpsc::{SyncSender, sync_channel};
-use std::sync::{Arc, Condvar, Mutex};
-use std::thread::{self, JoinHandle};
 
 use super::writeback::WritebackPipeline;
 
@@ -126,206 +77,25 @@ use super::writeback::WritebackPipeline;
 /// historical default — bounded-cache pressure stays at ~2 × this size.
 const WRITEBACK_CHUNK_BYTES: u64 = 32 * 1024 * 1024;
 
-/// Maximum bytes outstanding in the muxer → writer-thread ring. Sized
-/// to cover ~4 s of muxer output at a 32 MB/s peak — enough to absorb a
-/// short NFS commit stall without dropping the muxer's effective
-/// throughput, but not so large that the resident-memory footprint
-/// grows unbounded under a long writeback stall.
-const RING_CAPACITY_BYTES: usize = 128 * 1024 * 1024;
-
-/// Bytes the writer thread keeps in [`ActiveClusterBuffer`] for the
-/// in-window seek-then-patch fast path. Matches [`WRITEBACK_CHUNK_BYTES`]
-/// so that a cluster backpatch landing inside the most recently written
-/// (but not yet WAIT_AFTER'd) writeback chunk doesn't force a drain.
-const ACTIVE_CLUSTER_WINDOW_BYTES: u64 = WRITEBACK_CHUNK_BYTES;
-
-/// Thread name used for the writer thread. Visible to OS-level tooling
-/// (`ps -L`, `top -H`) so operators can correlate the muxer's flatness
-/// with this thread's activity.
-const WRITER_THREAD_NAME: &str = "freemkv-writeback-writer";
-
-/// Single chunk size cap for batching: when a `Write` command arrives
-/// the muxer copies its slice into a `Vec<u8>` to hand ownership over
-/// the ring. We keep the allocation a single contiguous buffer — no
-/// internal segmentation — so the writer thread can pass the slice
-/// straight to `File::write_all` and the kernel can coalesce.
-const MAX_WRITE_CHUNK_BYTES: usize = RING_CAPACITY_BYTES; // soft cap: a single command may not exceed the ring
-
-/// One command on the muxer → writer-thread ring.
-enum Cmd {
-    /// Write `buf.len()` bytes at the writer's current logical
-    /// position, then advance.
-    Write(Vec<u8>),
-    /// Seek to `from` against the writer-side `File`.
-    Seek(SeekFrom),
-    /// Flush the muxer-side `Write::flush()` request (rarely useful;
-    /// kept for trait completeness). The writer ignores it — the real
-    /// flushing happens on `SyncAll`.
-    Flush,
-    /// Drain the ring then run the per-OS durable-sync primitive and
-    /// signal `done` with the result.
-    SyncAll { done: SyncSender<io::Result<()>> },
-    /// Drain the ring (final pipeline finalize for chunk tail). Signal
-    /// `done` so `Drop` can wait synchronously. No fsync — that's what
-    /// `SyncAll` is for.
-    Finish { done: SyncSender<()> },
-}
-
-/// Ring state behind the muxer/writer condvar. `bytes_inflight` tracks
-/// the sum of `Write(buf).len()` bytes currently queued so backpressure
-/// can be enforced on a byte budget rather than a per-command count.
-struct RingState {
-    queue: VecDeque<Cmd>,
-    /// Total `Write` payload bytes currently in `queue`. Non-write
-    /// commands (`Seek`, `Flush`, `SyncAll`, `Finish`) don't count
-    /// against the budget.
-    bytes_inflight: usize,
-    /// Set by the writer thread when it observes a fatal error (a
-    /// failed `write_all` or `seek` on the underlying file). Once set,
-    /// the muxer's next `write` / `seek` returns the error and stops
-    /// queueing.
-    sticky_error: Option<io::ErrorKind>,
-    /// Set when the writer thread has exited (clean Finish, panic, or
-    /// channel closed). Muxer-side ops surface this as a broken-pipe.
-    writer_gone: bool,
-}
-
-struct Shared {
-    state: Mutex<RingState>,
-    /// Notified when the writer dequeues something (bytes free up) or
-    /// when the writer exits.
-    space_available: Condvar,
-    /// Notified when the muxer pushes a new command.
-    work_available: Condvar,
-}
-
-impl Shared {
-    fn new() -> Self {
-        Self {
-            state: Mutex::new(RingState {
-                queue: VecDeque::new(),
-                bytes_inflight: 0,
-                sticky_error: None,
-                writer_gone: false,
-            }),
-            space_available: Condvar::new(),
-            work_available: Condvar::new(),
-        }
-    }
-}
-
-/// Tiny ring of recently-written bytes indexed by absolute file
-/// position. Used by the writer thread to decide whether a `Seek`
-/// target falls inside the active writeback chunk; if so, the seek
-/// proceeds without forcing the pipeline to drain (the dominant case
-/// for MKV cluster-size backpatches).
-///
-/// The bytes themselves are kept in a contiguous `VecDeque<u8>` whose
-/// front corresponds to file position [`lo`]. The data on disk is the
-/// authoritative copy — `ActiveClusterBuffer` is a read-only mirror
-/// used for in-window patch verification in tests and (potentially) for
-/// future in-buffer mutation, NOT a write-through cache.
-///
-/// [`lo`]: Self::lo
-struct ActiveClusterBuffer {
-    /// Window capacity in bytes. The ring trims from the front to stay
-    /// at or below this size after every `push`.
-    cap: u64,
-    /// Absolute file position of the byte at `data.front()`.
-    lo: u64,
-    data: VecDeque<u8>,
-}
-
-impl ActiveClusterBuffer {
-    fn new(cap: u64) -> Self {
-        Self {
-            cap,
-            lo: 0,
-            data: VecDeque::with_capacity(cap as usize),
-        }
-    }
-
-    /// Reset the window — used after an out-of-window seek where the
-    /// previous data is no longer adjacent to the new position.
-    fn reset(&mut self, new_lo: u64) {
-        self.data.clear();
-        self.lo = new_lo;
-    }
-
-    /// One contiguous logical span of file positions currently held.
-    fn hi(&self) -> u64 {
-        self.lo + self.data.len() as u64
-    }
-
-    /// True if `pos` is in `[lo, hi]` (hi is exclusive of bytes but
-    /// inclusive of the seek-to-end-of-cluster boundary).
-    fn contains(&self, pos: u64) -> bool {
-        pos >= self.lo && pos <= self.hi()
-    }
-
-    /// Append `bytes` at absolute file position `start`. If `start` is
-    /// contiguous with `hi()`, the bytes extend the window; otherwise
-    /// the window is reset (the previous data is no longer adjacent and
-    /// would corrupt the position index).
-    fn push(&mut self, start: u64, bytes: &[u8]) {
-        let h = self.hi();
-        if start == h {
-            // Contiguous append.
-            self.data.extend(bytes.iter().copied());
-        } else if start >= self.lo && start <= h {
-            // Patch landing inside the window: overwrite from
-            // (start - lo) for bytes.len(), then extend if it spills
-            // past `hi`.
-            let offset = (start - self.lo) as usize;
-            let mut bi = 0usize;
-            while bi < bytes.len() && offset + bi < self.data.len() {
-                self.data[offset + bi] = bytes[bi];
-                bi += 1;
-            }
-            if bi < bytes.len() {
-                self.data.extend(bytes[bi..].iter().copied());
-            }
-        } else {
-            // Non-contiguous: drop the window and reseat.
-            self.data.clear();
-            self.lo = start;
-            self.data.extend(bytes.iter().copied());
-        }
-        // Trim from the front so the window stays at or below `cap`.
-        while self.data.len() as u64 > self.cap {
-            self.data.pop_front();
-            self.lo += 1;
-        }
-    }
-}
-
-/// Muxer-facing handle. Holds a sender into the bounded ring and a
-/// `JoinHandle` for the writer thread. `Write`/`Seek`/`sync_all`/`Drop`
-/// route through the ring; the muxer thread is never trapped on a
-/// commit syscall.
 pub(crate) struct WritebackFile {
-    shared: Arc<Shared>,
-    /// Joined on `Drop` after `Finish` so the writer thread's exit is
-    /// observed and any panic is surfaced loudly. `Option` so `Drop`
-    /// can `take()` it.
-    writer: Option<JoinHandle<()>>,
-    /// Logical file position from the muxer's point of view. Updated on
-    /// `write`/`write_all` (advanced by the count) and on `seek`
-    /// (replaced by the new position). Mirrors what the writer thread
-    /// will end up at once it has drained all queued commands —
-    /// callers that need a stream_position can read this without
-    /// blocking on the writer.
-    muxer_pos: u64,
+    file: File,
+    pipeline: WritebackPipeline,
+    pos: u64,
 }
 
 impl WritebackFile {
     /// Wrap an open `File`. The current OS file position is queried
-    /// once so the writer thread starts tracking from wherever the
-    /// file already is (typically 0 for fresh files; non-zero for
-    /// resumed or appended files).
+    /// once so the pipeline starts tracking from wherever the file
+    /// already is (typically 0 for fresh files; non-zero for resumed
+    /// or appended files).
     pub(crate) fn new(mut file: File) -> io::Result<Self> {
         let pos = file.stream_position()?;
-        Ok(Self::spawn(file, pos))
+        let pipeline = WritebackPipeline::new(&file, pos, WRITEBACK_CHUNK_BYTES);
+        Ok(Self {
+            file,
+            pipeline,
+            pos,
+        })
     }
 
     /// Create a new file at `path` (truncating any existing contents)
@@ -369,432 +139,84 @@ impl WritebackFile {
         Self::new(file)
     }
 
-    /// Spawn the writer thread for `file` starting at logical
-    /// position `start_pos`. The writer takes ownership of the `File`;
-    /// the muxer keeps the handle.
-    fn spawn(file: File, start_pos: u64) -> Self {
-        let shared = Arc::new(Shared::new());
-        let shared_w = Arc::clone(&shared);
-        let writer = thread::Builder::new()
-            .name(WRITER_THREAD_NAME.into())
-            .spawn(move || {
-                writer_thread_main(file, start_pos, shared_w);
-            })
-            .expect("writer thread spawn");
-        Self {
-            shared,
-            writer: Some(writer),
-            muxer_pos: start_pos,
-        }
-    }
-
     /// Drain in-flight writeback then issue a full fsync. Use this in
     /// place of `File::sync_all`.
     ///
-    /// Blocks the calling thread until the ring is fully drained AND
-    /// the per-OS durable-flush primitive has returned. The flush
-    /// itself runs on the writer thread, wrapped in
-    /// [`crate::io::bounded::bounded_syscall`] (60 s deadline on
-    /// Linux + macOS); a wedged NFS server cannot trap the muxer.
+    /// The final durable flush is wrapped in
+    /// [`crate::io::bounded::bounded_syscall`] (per the per-OS module)
+    /// with a 60 s deadline on Linux/macOS — a wedged NFS server cannot
+    /// trap the calling thread indefinitely. On timeout the page cache
+    /// is left to the kernel's normal flush-on-close path — best
+    /// effort, but bounded.
     pub(crate) fn sync_all(&mut self) -> io::Result<()> {
-        let (tx, rx) = sync_channel::<io::Result<()>>(0);
-        self.push_command(Cmd::SyncAll { done: tx }, 0)?;
-        // recv() blocks until the writer thread drains the ring up to
-        // the SyncAll command, runs the per-OS durable-sync, and sends
-        // the result back.
-        match rx.recv() {
-            Ok(r) => r,
-            Err(_) => {
-                // Writer thread exited without sending. Surface a
-                // distinct error kind so the caller can distinguish
-                // "writer panicked" from a normal fsync failure.
-                Err(io::Error::from(io::ErrorKind::BrokenPipe))
-            }
-        }
-    }
-
-    /// Push a single command onto the ring. `bytes_charge` is the
-    /// number of bytes this command contributes to the byte-budget
-    /// backpressure check; only `Write` commands contribute.
-    fn push_command(&mut self, cmd: Cmd, bytes_charge: usize) -> io::Result<()> {
-        let mut guard = self.shared.state.lock().unwrap();
-        // Surface any sticky error from the writer thread before
-        // queueing more work. The muxer should stop pushing once the
-        // writer has reported a failure.
-        if let Some(kind) = guard.sticky_error {
-            return Err(io::Error::from(kind));
-        }
-        if guard.writer_gone {
-            return Err(io::Error::from(io::ErrorKind::BrokenPipe));
-        }
-        // Byte-budget backpressure: wait for space if this would
-        // overflow the cap. A single command larger than the cap is
-        // admitted regardless (the cap is a soft target for batching;
-        // a giant single write still fits because the channel itself
-        // is unbounded count-wise).
-        while bytes_charge > 0
-            && guard.bytes_inflight + bytes_charge > RING_CAPACITY_BYTES
-            && guard.bytes_inflight > 0
-        {
-            guard = self.shared.space_available.wait(guard).unwrap();
-            if let Some(kind) = guard.sticky_error {
-                return Err(io::Error::from(kind));
-            }
-            if guard.writer_gone {
-                return Err(io::Error::from(io::ErrorKind::BrokenPipe));
-            }
-        }
-        guard.queue.push_back(cmd);
-        guard.bytes_inflight += bytes_charge;
-        drop(guard);
-        self.shared.work_available.notify_one();
-        Ok(())
+        self.pipeline.finalize();
+        platform::durable_sync(&self.file)
     }
 }
 
 impl Write for WritebackFile {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        // `Write::write` is allowed to be partial; we always accept
-        // the full slice (handoff is in-process) and report `buf.len()`.
-        // Callers that need the partial-write semantic still get the
-        // strict guarantee documented on `write_all`.
-        let n = buf.len();
-        if n == 0 {
-            return Ok(0);
-        }
-        if n > MAX_WRITE_CHUNK_BYTES {
-            // Defensive: a single command bigger than the ring can't
-            // be admitted by the backpressure check above without
-            // deadlocking against itself. Split into ring-sized
-            // chunks.
-            let mut off = 0;
-            while off < n {
-                let take = (n - off).min(MAX_WRITE_CHUNK_BYTES);
-                self.push_command(Cmd::Write(buf[off..off + take].to_vec()), take)?;
-                off += take;
-            }
-        } else {
-            self.push_command(Cmd::Write(buf.to_vec()), n)?;
-        }
-        self.muxer_pos += n as u64;
+        let n = self.file.write(buf)?;
+        self.pos += n as u64;
+        self.pipeline.note_progress(self.pos);
         Ok(n)
     }
 
     fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
-        // `Write::write_all` default delegates to `write` in a loop. We
-        // can do better: a single handoff per call, never partial. Same
-        // chunk-split for the absurd-large case.
-        let n = buf.len();
-        if n == 0 {
-            return Ok(());
-        }
-        if n > MAX_WRITE_CHUNK_BYTES {
-            let mut off = 0;
-            while off < n {
-                let take = (n - off).min(MAX_WRITE_CHUNK_BYTES);
-                self.push_command(Cmd::Write(buf[off..off + take].to_vec()), take)?;
-                off += take;
-            }
-        } else {
-            self.push_command(Cmd::Write(buf.to_vec()), n)?;
-        }
-        self.muxer_pos += n as u64;
+        self.file.write_all(buf)?;
+        self.pos += buf.len() as u64;
+        self.pipeline.note_progress(self.pos);
         Ok(())
     }
 
     fn flush(&mut self) -> io::Result<()> {
-        // The writer thread's view of `Flush` is a no-op (real flushing
-        // happens on `SyncAll`). We still send it so a future change
-        // could intercept it; for now the queue ordering is the only
-        // observable effect.
-        self.push_command(Cmd::Flush, 0)
+        self.file.flush()
     }
 }
 
 impl Seek for WritebackFile {
     fn seek(&mut self, from: SeekFrom) -> io::Result<u64> {
-        // The muxer's logical position must update synchronously so
-        // subsequent `write` calls advance from the right base, but the
-        // writer thread is the only place that has the authoritative
-        // OS file position. We model the muxer's `muxer_pos` purely
-        // from `SeekFrom::Start(n)` (the dominant case for MKV
-        // backpatch) and bounce other variants through to the writer
-        // by querying its current position via a synchronous round.
-        let new_pos = match from {
-            SeekFrom::Start(n) => n,
-            SeekFrom::Current(d) => {
-                let base = self.muxer_pos as i64;
-                let p = base
-                    .checked_add(d)
-                    .ok_or_else(|| io::Error::from(io::ErrorKind::InvalidInput))?;
-                if p < 0 {
-                    return Err(io::Error::from(io::ErrorKind::InvalidInput));
-                }
-                p as u64
-            }
-            SeekFrom::End(_) => {
-                // SeekFrom::End requires the OS file's current EOF. We
-                // do not maintain that on the muxer side; this branch
-                // is not used by the MKV muxer (it always seeks with
-                // `SeekFrom::Start`). If a future caller needs it, the
-                // path is: SyncAll → real seek → query position back.
-                // Reject explicitly so a regression is loud.
-                return Err(io::Error::from(io::ErrorKind::Unsupported));
-            }
-        };
-        self.push_command(Cmd::Seek(SeekFrom::Start(new_pos)), 0)?;
-        self.muxer_pos = new_pos;
-        Ok(new_pos)
+        let p = self.file.seek(from)?;
+        // Only treat seeks that actually move the position as
+        // boundaries — sweep does a redundant `seek(Current(pos))`
+        // before every write, and we don't want that to drain the
+        // pipeline on every iteration.
+        if p != self.pos {
+            // Diagnostic for the NFS mux hang: the MKV format requires
+            // the muxer to seek back occasionally (cluster size
+            // patching, Cues index write, Segment header backpatch).
+            // Each such seek invalidates the writeback chunk tracking
+            // and forces a finalize → WAIT_AFTER on the in-flight
+            // chunk. Logging the seek delta lets us correlate hang
+            // offsets with specific muxer operations.
+            let from_pos = self.pos;
+            let to_pos = p;
+            let delta: i64 = (to_pos as i64).wrapping_sub(from_pos as i64);
+            tracing::debug!(
+                target: "mux",
+                "WritebackFile seek from={from_pos} to={to_pos} delta={delta}"
+            );
+            self.pipeline.handle_seek(p);
+            self.pos = p;
+        }
+        Ok(p)
     }
 }
 
 impl Drop for WritebackFile {
     fn drop(&mut self) {
-        // Send a final `Finish` so the writer drains the ring (running
-        // the pipeline's tail finalize for the last in-flight chunk)
-        // before exiting. `sync_all` is *not* called from here — the
-        // existing pre-Phase-2.5 contract is "Drop runs finalize, not
-        // fsync", and Drop returning an io::Error is impossible anyway.
-        let (tx, rx) = sync_channel::<()>(0);
-        // If the writer thread already exited (sticky error path, or
-        // an earlier panic), the queue push will fail with
-        // BrokenPipe; we treat that as "nothing to drain" and proceed
-        // to join.
-        let push_ok = {
-            let mut guard = match self.shared.state.lock() {
-                Ok(g) => g,
-                Err(poison) => {
-                    // The writer panicked; recover the mutex so we
-                    // can still observe `writer_gone`.
-                    poison.into_inner()
-                }
-            };
-            if guard.writer_gone {
-                false
-            } else {
-                guard.queue.push_back(Cmd::Finish { done: tx });
-                self.shared.work_available.notify_one();
-                true
-            }
-        };
-        if push_ok {
-            // Block until the writer signals Finish completed (ring is
-            // drained, pipeline finalize ran). The wait is bounded
-            // only by the writer's per-syscall deadlines; if the
-            // writer panicked between the push and the recv, the
-            // sender is dropped and `recv` returns Err — proceed to
-            // join.
-            let _ = rx.recv();
-        }
-        if let Some(jh) = self.writer.take() {
-            // `join` surfaces a panic. We re-raise it loudly: a
-            // writer-thread panic indicates an io-layer bug, and
-            // swallowing it would mask data loss.
-            if let Err(panic) = jh.join() {
-                tracing::error!(
-                    target: "mux",
-                    "WritebackFile writer thread panicked during Drop; data may be lost"
-                );
-                // Re-raise during drop is allowed (terminates the
-                // process), but doing so from Drop can cause a double-
-                // panic if the caller is already unwinding. Compromise:
-                // log loudly and resume_unwind only outside of an
-                // ongoing unwind.
-                if !std::thread::panicking() {
-                    std::panic::resume_unwind(panic);
-                }
-            }
-        }
-    }
-}
-
-/// Writer-thread entry. Owns the `File` and the `WritebackPipeline`;
-/// pulls commands from the shared ring and executes them. Exits when a
-/// `Finish` command is observed (clean shutdown from `Drop`) or when
-/// the muxer side disconnects (every `Arc<Shared>` cloned by the
-/// handle is dropped — only happens on a forgotten-handle bug, which
-/// the panic-surface in `Drop::join` catches).
-fn writer_thread_main(file: File, start_pos: u64, shared: Arc<Shared>) {
-    let pipeline = WritebackPipeline::new(&file, start_pos, WRITEBACK_CHUNK_BYTES);
-    let mut state = WriterState {
-        file,
-        pipeline,
-        pos: start_pos,
-        active: ActiveClusterBuffer::new(ACTIVE_CLUSTER_WINDOW_BYTES),
-        shared: Arc::clone(&shared),
-    };
-    state.run();
-}
-
-/// All writer-thread-owned state. Methods here run exclusively on the
-/// writer thread — no Send/Sync concerns inside the body.
-struct WriterState {
-    file: File,
-    pipeline: WritebackPipeline,
-    /// Authoritative OS file position. Tracked locally so we can decide
-    /// whether a `Seek` is a no-op (target equals current position).
-    pos: u64,
-    active: ActiveClusterBuffer,
-    shared: Arc<Shared>,
-}
-
-impl WriterState {
-    fn run(&mut self) {
-        loop {
-            let cmd = match self.dequeue() {
-                Some(c) => c,
-                None => {
-                    // All senders dropped (handle leaked); mark
-                    // writer_gone and exit. The Drop join will surface
-                    // this if anyone cares.
-                    self.mark_writer_gone();
-                    return;
-                }
-            };
-            match cmd {
-                Cmd::Write(buf) => {
-                    if let Err(e) = self.do_write(&buf) {
-                        self.publish_error(e.kind());
-                    }
-                }
-                Cmd::Seek(from) => {
-                    if let Err(e) = self.do_seek(from) {
-                        self.publish_error(e.kind());
-                    }
-                }
-                Cmd::Flush => {
-                    // No-op for now (see note on `Cmd::Flush`).
-                }
-                Cmd::SyncAll { done } => {
-                    let r = self.do_sync_all();
-                    // Ignore send errors: if the muxer dropped the
-                    // receiver (cancelled wait), there's nothing to
-                    // do.
-                    let _ = done.send(r);
-                }
-                Cmd::Finish { done } => {
-                    // Drain pipeline tail; do not fsync. Mark
-                    // `writer_gone` so any racing `push_command` after
-                    // this returns BrokenPipe instead of queueing into
-                    // a dead writer.
-                    self.pipeline.finalize();
-                    self.mark_writer_gone();
-                    let _ = done.send(());
-                    return;
-                }
-            }
-        }
-    }
-
-    /// Block until at least one command is available, then return it.
-    /// Notifies the muxer side that bytes are free.
-    fn dequeue(&mut self) -> Option<Cmd> {
-        let mut guard = self.shared.state.lock().unwrap();
-        loop {
-            if let Some(cmd) = guard.queue.pop_front() {
-                if let Cmd::Write(ref buf) = cmd {
-                    guard.bytes_inflight = guard.bytes_inflight.saturating_sub(buf.len());
-                }
-                drop(guard);
-                self.shared.space_available.notify_all();
-                return Some(cmd);
-            }
-            // Queue is empty. Wait for new work.
-            guard = self.shared.work_available.wait(guard).unwrap();
-        }
-    }
-
-    fn do_write(&mut self, buf: &[u8]) -> io::Result<()> {
-        let start = self.pos;
-        self.file.write_all(buf)?;
-        self.pos += buf.len() as u64;
-        self.pipeline.note_progress(self.pos);
-        self.active.push(start, buf);
-        Ok(())
-    }
-
-    fn do_seek(&mut self, from: SeekFrom) -> io::Result<()> {
-        // We only ever push `SeekFrom::Start(n)` from the handle.
-        let target = match from {
-            SeekFrom::Start(n) => n,
-            // Defensive: should not occur on the wire, but handle
-            // gracefully.
-            SeekFrom::Current(_) | SeekFrom::End(_) => {
-                let p = self.file.seek(from)?;
-                self.pos = p;
-                self.active.reset(p);
-                self.pipeline.handle_seek(p);
-                return Ok(());
-            }
-        };
-        if target == self.pos {
-            // No-op seek (common: sweep emits `seek(Current(pos))`
-            // before every write). Skip the syscall.
-            return Ok(());
-        }
-        if self.active.contains(target) {
-            // In-window seek: the target is inside the current
-            // writeback chunk's data. The kernel page cache already
-            // has those bytes; we issue the real `seek` (cheap, no
-            // commit syscall) but **skip** the pipeline's
-            // `handle_seek` so no `sync_file_range(WAIT_AFTER)` drain
-            // is forced. Subsequent writes still call
-            // `pipeline.note_progress` from the new position, so the
-            // writeback chunk accounting stays coherent — the chunk
-            // simply gets "re-emitted" data over its tail bytes,
-            // which is what the MKV backpatch is.
-            tracing::trace!(
-                target: "mux",
-                "WritebackFile in-window seek pos={} -> {} window=[{},{}]",
-                self.pos,
-                target,
-                self.active.lo,
-                self.active.hi(),
-            );
-            self.file.seek(SeekFrom::Start(target))?;
-            self.pos = target;
-        } else {
-            // Out-of-window seek: rare (segment-header / Cues backpatch
-            // at end of mux). Drain in-flight writeback so the kernel
-            // doesn't carry dirty pages across the seek discontinuity,
-            // then do the real seek.
-            tracing::debug!(
-                target: "mux",
-                "WritebackFile out-of-window seek pos={} -> {} window=[{},{}]",
-                self.pos,
-                target,
-                self.active.lo,
-                self.active.hi(),
-            );
-            self.pipeline.handle_seek(target);
-            self.file.seek(SeekFrom::Start(target))?;
-            self.pos = target;
-            self.active.reset(target);
-        }
-        Ok(())
-    }
-
-    fn do_sync_all(&mut self) -> io::Result<()> {
+        // Run the pipeline's tail finalize so the last in-flight chunk
+        // gets its `WAIT_AFTER` + `posix_fadvise(DONTNEED)`. Without
+        // this, callers that drop a `WritebackFile` without calling
+        // `sync_all` (panic, early-return, idiomatic `let _ = w;`)
+        // leave the trailing chunk in cache; the kernel still flushes
+        // on close, but the bounded-cache invariant fails at the tail.
+        // We deliberately do *not* call `self.file.sync_all()` here —
+        // close already triggers a flush, and an `fsync` from `Drop`
+        // would silently swallow its `io::Error` anyway. `finalize` is
+        // idempotent so an explicit `sync_all` followed by drop is
+        // still safe.
         self.pipeline.finalize();
-        platform::durable_sync(&self.file)
-    }
-
-    fn publish_error(&self, kind: io::ErrorKind) {
-        let mut guard = self.shared.state.lock().unwrap();
-        if guard.sticky_error.is_none() {
-            guard.sticky_error = Some(kind);
-        }
-        drop(guard);
-        // Wake any muxer thread waiting on space — it will observe
-        // sticky_error and return.
-        self.shared.space_available.notify_all();
-    }
-
-    fn mark_writer_gone(&self) {
-        let mut guard = self.shared.state.lock().unwrap();
-        guard.writer_gone = true;
-        drop(guard);
-        self.shared.space_available.notify_all();
     }
 }
 
@@ -817,13 +239,13 @@ mod tests {
         {
             let mut w = WritebackFile::create(&p).unwrap();
             w.write_all(b"hello world").unwrap();
-            // Drop drains the ring.
+            // Drop drains the pipeline tail.
         }
         assert_eq!(read_back(&p), b"hello world");
     }
 
     #[test]
-    fn sync_all_blocks_until_ring_drains() {
+    fn sync_all_drains_and_flushes() {
         let dir = tempfile::tempdir().unwrap();
         let p = dir.path().join("b.bin");
         let mut w = WritebackFile::create(&p).unwrap();
@@ -831,7 +253,8 @@ mod tests {
             w.write_all(&[0x5au8; 1024]).unwrap();
         }
         // After sync_all, the bytes MUST be visible to a separate
-        // reader. The ring has been drained and durable-sync has run.
+        // reader. The pipeline has been finalised and durable-sync has
+        // run.
         w.sync_all().unwrap();
         let bytes = read_back(&p);
         assert_eq!(bytes.len(), 32 * 1024);
@@ -840,20 +263,17 @@ mod tests {
     }
 
     #[test]
-    fn in_window_seek_then_patch_roundtrip() {
-        // Write A; seek back inside the active-cluster window; patch
-        // with B; read back; the patch lands at the right offset.
+    fn seek_then_patch_roundtrip() {
+        // Write A; seek back; patch with B; read back; the patch lands
+        // at the right offset.
         let dir = tempfile::tempdir().unwrap();
         let p = dir.path().join("c.bin");
         let mut w = WritebackFile::create(&p).unwrap();
-        // 4 KiB of 'A' (well within ACTIVE_CLUSTER_WINDOW_BYTES =
-        // 32 MiB, so the seek-back is guaranteed in-window).
         let big = vec![b'A'; 4096];
         w.write_all(&big).unwrap();
         // Seek back to offset 1000 and overwrite 8 bytes.
         w.seek(SeekFrom::Start(1000)).unwrap();
         w.write_all(b"PATCHED!").unwrap();
-        // Seek to end so subsequent reads see the right size.
         w.sync_all().unwrap();
         drop(w);
         let bytes = read_back(&p);
@@ -864,67 +284,10 @@ mod tests {
         assert_eq!(bytes[1008], b'A');
     }
 
-    #[test]
-    fn out_of_window_seek_then_patch_roundtrip() {
-        // Write enough bytes that a seek to offset 0 is outside the
-        // active-cluster window (which is 32 MiB). To keep the test
-        // bounded, we hammer the ActiveClusterBuffer's `cap` field
-        // directly via the public Write path — 33 MiB of payload is
-        // sufficient.
-        let dir = tempfile::tempdir().unwrap();
-        let p = dir.path().join("d.bin");
-        let mut w = WritebackFile::create(&p).unwrap();
-        // Write 33 MiB of 'A'; the first 1 MiB is now outside the
-        // 32 MiB active-cluster window.
-        let chunk = vec![b'A'; 1024 * 1024];
-        for _ in 0..33 {
-            w.write_all(&chunk).unwrap();
-        }
-        // Seek to offset 100 (definitely outside the window) and
-        // patch.
-        w.seek(SeekFrom::Start(100)).unwrap();
-        w.write_all(b"OUTSIDE!").unwrap();
-        w.sync_all().unwrap();
-        drop(w);
-        let bytes = read_back(&p);
-        assert_eq!(bytes.len(), 33 * 1024 * 1024);
-        assert_eq!(&bytes[100..108], b"OUTSIDE!");
-        // Surrounding bytes are still 'A'.
-        assert_eq!(bytes[99], b'A');
-        assert_eq!(bytes[108], b'A');
-    }
-
-    #[test]
-    fn backpressure_blocks_when_ring_full() {
-        // Submit more bytes than RING_CAPACITY_BYTES; if backpressure
-        // works, the call sequence still completes once the writer
-        // drains. We measure that the total written matches and the
-        // calling thread did not panic / loop forever.
-        let dir = tempfile::tempdir().unwrap();
-        let p = dir.path().join("e.bin");
-        let mut w = WritebackFile::create(&p).unwrap();
-        // 4 × RING_CAPACITY_BYTES of payload, in chunks small enough
-        // that several can fit in the ring at once and backpressure
-        // triggers naturally.
-        let total = RING_CAPACITY_BYTES.saturating_mul(2) + (RING_CAPACITY_BYTES / 2);
-        let chunk = vec![0u8; 1024 * 1024];
-        let mut written = 0;
-        while written < total {
-            let take = (total - written).min(chunk.len());
-            w.write_all(&chunk[..take]).unwrap();
-            written += take;
-        }
-        w.sync_all().unwrap();
-        drop(w);
-        let meta = std::fs::metadata(&p).unwrap();
-        assert_eq!(meta.len() as usize, total);
-    }
-
     #[test]
     fn flush_is_observed_in_order() {
-        // `Write::flush` is a no-op on the writer side but must not
-        // panic or leak. Run an interleaved sequence and verify the
-        // bytes still land in order.
+        // `Write::flush` should not panic or reorder; verify the bytes
+        // land in order through interleaved flushes.
         let dir = tempfile::tempdir().unwrap();
         let p = dir.path().join("f.bin");
         let mut w = WritebackFile::create(&p).unwrap();
@@ -937,90 +300,4 @@ mod tests {
         drop(w);
         assert_eq!(read_back(&p), b"onetwothree");
     }
-
-    #[test]
-    fn active_cluster_buffer_contiguous_append_and_trim() {
-        let mut b = ActiveClusterBuffer::new(8);
-        b.push(0, b"abcd");
-        assert_eq!(b.lo, 0);
-        assert_eq!(b.hi(), 4);
-        b.push(4, b"efgh");
-        assert_eq!(b.lo, 0);
-        assert_eq!(b.hi(), 8);
-        // Push past the cap — front trims.
-        b.push(8, b"ij");
-        assert_eq!(b.lo, 2);
-        assert_eq!(b.hi(), 10);
-        assert!(b.contains(2));
-        assert!(b.contains(10));
-        assert!(!b.contains(1));
-        assert!(!b.contains(11));
-    }
-
-    #[test]
-    fn active_cluster_buffer_in_window_patch() {
-        let mut b = ActiveClusterBuffer::new(16);
-        b.push(100, b"AAAAAAAA");
-        assert!(b.contains(104));
-        // Patch in the middle.
-        b.push(102, b"BB");
-        let collected: Vec<u8> = b.data.iter().copied().collect();
-        assert_eq!(collected, b"AABBAAAA");
-        assert_eq!(b.lo, 100);
-        assert_eq!(b.hi(), 108);
-    }
-
-    #[test]
-    fn active_cluster_buffer_non_contiguous_reseats() {
-        let mut b = ActiveClusterBuffer::new(16);
-        b.push(0, b"abcd");
-        b.push(1000, b"XYZ");
-        assert_eq!(b.lo, 1000);
-        assert_eq!(b.hi(), 1003);
-    }
-
-    #[test]
-    fn writer_thread_panic_surfaces_on_drop() {
-        // Simulate a writer-side panic by writing to a read-only file
-        // — the underlying `file.write_all` will return EBADF /
-        // PermissionDenied. The writer publishes sticky_error and
-        // exits via the next dequeue; subsequent push_command returns
-        // the error. Drop joins cleanly (no panic from the writer
-        // thread itself; it returned through the error path).
-        //
-        // We deliberately use a closed-FD strategy: open a file,
-        // truncate the kernel's view by closing it, then write — this
-        // is hard to force without unsafe. Easier: write to a path
-        // and then forcibly close the underlying File via shutdown of
-        // the writer thread. Since we don't expose the inner File,
-        // pick the read-only-mode approach: open the file in
-        // read-only mode and try to write.
-        let dir = tempfile::tempdir().unwrap();
-        let p = dir.path().join("ro.bin");
-        std::fs::write(&p, b"seed").unwrap();
-        let f = OpenOptions::new().read(true).open(&p).unwrap();
-        let mut w = WritebackFile::new(f).unwrap();
-        // First write may succeed depending on platform; loop until
-        // an error surfaces. On Linux a write on an O_RDONLY fd
-        // returns EBADF immediately.
-        let mut saw_error = false;
-        for _ in 0..32 {
-            match w.write_all(b"x") {
-                Ok(()) => {
-                    // Give the writer thread a moment to surface the
-                    // error then retry.
-                    std::thread::sleep(std::time::Duration::from_millis(10));
-                }
-                Err(_) => {
-                    saw_error = true;
-                    break;
-                }
-            }
-        }
-        assert!(
-            saw_error,
-            "expected the writer to publish an error on a read-only fd"
-        );
-        drop(w);
-    }
 }