io/writeback: medium-agnostic by construction, no DONTNEED, single detection point

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.

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
+//! `sync_file_range(WAIT_AFTER)` runs on every medium (local FS, NFS,
-//! indefinitely waiting for the server's commit ack. If the server
+//! etc.). It bounds the dirty-page set at ~2 × `chunk_bytes` by
-//! never acks (network partition, server-side hang, slow commit), the
+//! waiting for each chunk's kernel writeback to complete before
-//! syscall never returns and the consumer thread is stuck inside the
+//! advancing. There is no `if is_nfs` branch in the hot path. The
-//! kernel — `/api/stop` can't reach it because halt is cooperative.
+//! 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
+//! **No `posix_fadvise(DONTNEED)` after the wait.** Earlier revisions
-//! (`f_type == NFS_SUPER_MAGIC`), the pipeline skips the WAIT_AFTER +
+//! evicted the just-WAIT_AFTER'd range from the page cache on the
-//! `posix_fadvise(DONTNEED)` dance entirely. NFS clients have their
+//! theory that bounding *dirty* pages required also bounding *clean*
-//! own buffering and commit semantics that handle dirty-page bounds
+//! pages. That was unnecessary and counterproductive: clean cached
-//! without us forcing the issue. The async `SYNC_FILE_RANGE_WRITE`
+//! pages cost nothing (the kernel reclaims them under LRU when memory
-//! kickoff still runs (non-blocking by spec) so writeback still gets
+//! is needed) and DONTNEED forces a read-modify-write on any
-//! a nudge.
+//! 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
+//! A degraded disk, wedged NFS server, or odd FS driver could in
-//! could in principle wedge inside WAIT_AFTER. Each WAIT_AFTER call
+//! principle hang inside WAIT_AFTER. Each call runs on a worker
-//! runs on a worker thread with a 30s recv_timeout on its result
+//! thread with a 30s recv_timeout on its result channel
-//! channel. On timeout we log a loud error, set a `degraded` flag,
+//! ([`crate::io::bounded::bounded_syscall`]). On timeout the pipeline
-//! and from then on skip WAIT_AFTER + DONTNEED for the rest of the
+//! flips to `degraded`, logs a loud error, and skips WAIT_AFTER for
-//! pipeline's life (same shape as the NFS path). The worker thread
+//! the rest of its life. The worker thread is intentionally leaked —
-//! is intentionally leaked — it unwinds whenever the syscall
+//! it unwinds whenever the syscall eventually returns or the process
-//! eventually returns or the process exits. The mux continues; the
+//! exits. The mux continues; the original dirty-burst pathology
-//! original dirty-burst pathology re-emerges but the rip can still
+//! re-emerges only on the specific (medium, server, FS) combination
-//! finish instead of freezing.
+//! 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
+    /// What `detect_storage_class` saw at construction. Carried for
-    /// WAIT_AFTER unsafe (can block forever on missing server ack), so
+    /// diagnostic logging only — *not* keyed off in the hot path.
-    /// we skip it entirely and let the NFS client handle commit on
+    storage_class: StorageClass,
-    /// close.
-    is_nfs: bool,
     /// 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={}",
+            "WritebackPipeline fd={fd} storage_class={class:?} chunk_bytes={chunk_bytes}"
-            if is_nfs { "nfs-skip-wait" } else { "wait+dontneed" }
         );
         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
+    /// True if we should bypass the WAIT_AFTER finalisation step. The
-    /// step. NFS always bypasses; local storage bypasses once the
+    /// pipeline only skips after `WAIT_AFTER` has *empirically* hung
-    /// pipeline has flipped to degraded after a WAIT_AFTER timeout.
+    /// 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
+                // Degraded path (set only after an empirical
-                // timeout): the WAIT_AFTER + DONTNEED dance is what
+                // WAIT_AFTER timeout on this fd). We still advance
-                // hangs on NFS — skip it. We still advance `pending`
+                // `pending` so the next call has a stable cycle.
-                // 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();
+                        // Deliberately NO `posix_fadvise(DONTNEED)`
-                        unsafe {
+                        // here. See module-level docs: evicting clean
-                            libc::posix_fadvise(
+                        // pages costs us read-modify-write on
-                                self.fd,
+                        // matroska cluster backpatches and gains
-                                prev_off as i64,
+                        // nothing — the kernel will reclaim clean
-                                prev_len as i64,
+                        // pages under LRU when memory is needed.
-                                libc::POSIX_FADV_DONTNEED,
-                            );
-                        }
-                        fadvise_ms = t_fadv.elapsed().as_millis() as u64;
                         self.record_wait(wait_ms);
                     }
                     None => {
-                        // Timeout branch: switch to NFS-style skip
+                        // Timeout branch: this fd's FS or server is
-                        // for the rest of the pipeline's life. Do
+                        // unable to ack WAIT_AFTER. Flip to skip mode
-                        // NOT call DONTNEED — if WAIT_AFTER hasn't
+                        // for the rest of the pipeline's life.
-                        // returned, the pages aren't safely flushed.
                         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,27 +282,15 @@ 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) {
+            if wait_after_with_timeout(self.fd, prev_off, prev_len).is_none() {
-                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",
@@ -298,15 +301,32 @@ impl WritebackPipeline {
         }
     }
 }
+
+/// 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 whether `fd` lives on an NFS mount via `fstatfs`. Returns
+/// Probe the FS class via `fstatfs` and return both the class label
-/// `false` on any error — we fail open, not closed: better to run the
+/// and a recommended `chunk_bytes` seed. `seed == 0` means "no
-/// normal local-storage path on a misdetected NFS mount (and surface
+/// medium-specific recommendation; use the caller's hint." Failure
-/// the freeze loudly via the timeout) than to needlessly disable
+/// modes default to `Unknown` + seed 0 so the caller's hint applies.
-/// writeback bounding on every local file because of a transient
+///
-/// stat error.
+/// This is the single medium-detection point in the whole writeback
-fn detect_nfs(fd: RawFd) -> bool {
+/// 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