v0.20.6: io::bounded — halt-safe wrapper for blocking syscalls

Generalizes 0.20.5's hand-written wait_after_with_timeout into a
reusable primitive. After this change, every blocking syscall in the
recovery + mux paths is wrapped, so cooperative Halt has bounded
~250 ms latency reach even into kernel-owned thread states.

New module src/io/bounded.rs:
- BoundedError { Halted, Timeout, WorkerLost }
- bounded_syscall<F, R>(halt: Option<&Halt>, timeout, op) -> Result<R, BoundedError>
- Worker thread runs op; main thread recv_timeouts on a rendezvous
  channel in 250 ms slices, polling halt between slices.
- Worker is intentionally leaked on timeout/halt — kernel reaps when
  the syscall finally returns or at process exit. Calling thread is
  NEVER trapped inside a kernel call.
- 6 unit tests cover the happy path + each error variant.

Refactored callsites:
- src/io/writeback/linux.rs::wait_after_with_timeout now delegates
  to bounded_syscall. ~30 LOC of duplicated channel/thread plumbing
  deleted. Same semantics, cleaner.
- src/io/writeback_file.rs::WritebackFile::sync_all now wraps the
  final libc::fsync(fd) with bounded_syscall (60 s deadline). On
  timeout: log error at target=mux and return Ok — kernel will flush
  on close, best-effort but bounded. Covers FileSectorSink::finish,
  PatchSink::close, SweepSink::close, and the mux MKV finalize path
  (they all sync through WritebackFile).

What still hangs (deliberately not wrapped — too hot a path):
- File::write itself. Per-frame write on a wedged NFS could still
  block; but back-pressure from a stuck consumer means the producer
  notices within seconds, not minutes — different failure mode than
  the WAIT_AFTER hang 0.20.5/0.20.6 fix.

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "libfreemkv"
-version = "0.20.2"
+version = "0.20.6"
 edition = "2024"
 rust-version = "1.86"
 license = "AGPL-3.0-only"

src/disc/patch.rs

@@ -895,6 +895,18 @@ impl Disc {
                                 );
                                 let mut bt_pos = backtrack_start;
                                 while bt_pos < backtrack_end {
+                                    // Honor cancellation inside the
+                                    // backtrack inner loop. A long
+                                    // backtrack span can run minutes
+                                    // of single-sector reads; without
+                                    // this check the outer halt only
+                                    // takes effect when control
+                                    // returns to the per-range loop.
+                                    if let Some(h) = &opts.halt {
+                                        if h.load(std::sync::atomic::Ordering::Relaxed) {
+                                            return Err(crate::error::Error::Halted);
+                                        }
+                                    }
                                     let span =
                                         // Backtrack always at count=1: this path
                                         // fills a gap that the main loop's damage-

src/io/bounded.rs

@@ -0,0 +1,246 @@
+//! Bounded-syscall primitive: run a (potentially-blocking) operation
+//! on a worker thread, with a hard wall-clock deadline and an optional
+//! cooperative [`Halt`] poll. The calling thread is never trapped
+//! inside a kernel call.
+//!
+//! ## Why this exists
+//!
+//! [`crate::halt::Halt`] is cooperative: callers poll
+//! `is_cancelled()`. It cannot reach inside a syscall the kernel
+//! currently owns the thread for — `libc::sync_file_range`,
+//! `libc::fsync`, `File::write` on NFS, and so on. `/api/stop` from
+//! autorip therefore can't unstick a thread sitting in such a syscall.
+//!
+//! [`bounded_syscall`] is the escape hatch: it runs `op` on a fresh
+//! worker thread, then `recv_timeout`s on a rendezvous channel for the
+//! result. The wait is broken into ~250 ms slices so the calling
+//! thread can poll the supplied [`Halt`] in between. If the deadline
+//! elapses or the halt fires, the worker is intentionally leaked — the
+//! syscall will unwind whenever the kernel decides, or at process
+//! exit, but the caller is free to fall back to a degraded code path
+//! (skip the sync, log loudly, etc.).
+//!
+//! ## Trade-offs
+//!
+//! - **Thread per call.** Cheap (`std::thread::spawn` is < 100 µs on
+//!   Linux/macOS), but not free. Use on coarse-grained finalisation
+//!   syscalls (`sync_all`, `sync_file_range(WAIT_AFTER)`), not on hot
+//!   inner-loop writes.
+//! - **Leak on timeout.** A wedged syscall keeps a kernel slot and a
+//!   user-space thread around for the rest of the process's life.
+//!   Bounded by the number of independent rip/mux sessions, which is
+//!   one per disc. The alternative — trapping the caller forever —
+//!   defeats the entire purpose of `/api/stop`.
+//! - **Halt granularity ~250 ms.** Halt observation is not instant;
+//!   it's the worst-case latency of the `recv_timeout` slice. Good
+//!   enough for human-driven stop requests; not suitable for hard
+//!   real-time deadlines.
+//!
+//! ## Single source of truth
+//!
+//! Do NOT inline this pattern. Every blocking-syscall wrapper in the
+//! rip + mux pipeline calls this helper, so changes (e.g. swapping the
+//! channel impl, adjusting the poll slice, adding metrics) land in one
+//! place.
+//!
+//! ## Platform
+//!
+//! Pure `std::thread` + `std::sync::mpsc`. No `cfg(target_os)` needed
+//! here — the helper itself is platform-agnostic. Callers that wrap
+//! Linux-only syscalls (`sync_file_range`) still need their own
+//! `#[cfg(target_os = "linux")]` gates; this helper does not.
+
+use std::sync::mpsc::{RecvTimeoutError, sync_channel};
+use std::thread;
+use std::time::{Duration, Instant};
+
+use crate::halt::Halt;
+
+/// Granularity of the halt poll. The receive loop wakes every
+/// [`POLL_INTERVAL`] to (a) check the [`Halt`] token, then (b) check
+/// the overall deadline, then go back to waiting. 250 ms is a
+/// pragmatic balance: short enough that human-driven `/api/stop` feels
+/// responsive (< 0.5 s p99), long enough that the polling overhead is
+/// negligible against multi-second syscalls.
+const POLL_INTERVAL: Duration = Duration::from_millis(250);
+
+/// Failure outcome from a bounded syscall wrapper.
+#[derive(Debug)]
+pub(crate) enum BoundedError {
+    /// The user-visible halt token fired during the wait. The worker
+    /// thread is intentionally leaked — the caller should fall back to
+    /// a degraded code path rather than waiting on the syscall to
+    /// return.
+    Halted,
+    /// The deadline elapsed before the syscall returned. Same leak
+    /// semantics as `Halted`.
+    Timeout,
+    /// The worker thread panicked, or its sender disconnected before
+    /// sending a result. Treat as a benign no-op (callers usually
+    /// log and continue) rather than a hard error — by definition no
+    /// syscall observably ran to completion in this case.
+    WorkerLost,
+}
+
+/// Run a (potentially-blocking) operation on a worker thread with a
+/// deadline and an optional cooperative halt-token poll. Returns the
+/// operation's result if it completes within `timeout`; otherwise one
+/// of [`BoundedError::Halted`] / [`BoundedError::Timeout`] /
+/// [`BoundedError::WorkerLost`].
+///
+/// On `Halted` / `Timeout` the worker thread is intentionally leaked:
+/// the syscall will unwind whenever the kernel decides, or when the
+/// process exits. The calling thread is never trapped inside a kernel
+/// call.
+///
+/// `halt` is polled at [`POLL_INTERVAL`] granularity. Pass `None` for
+/// callers that don't (yet) have a halt token plumbed through —
+/// behaviour degrades to deadline-only, matching the 0.20.5
+/// `wait_after_with_timeout` shape this helper generalises.
+///
+/// `op` returns `R: Send + 'static`. The closure must own everything
+/// it touches because it may outlive this call (timeout / halt cases).
+pub(crate) fn bounded_syscall<F, R>(
+    halt: Option<&Halt>,
+    timeout: Duration,
+    op: F,
+) -> Result<R, BoundedError>
+where
+    F: FnOnce() -> R + Send + 'static,
+    R: Send + 'static,
+{
+    // Rendezvous channel: the worker sends exactly one value (the
+    // op's return) and then exits. Capacity-0 means the send blocks
+    // until we receive — fine on the happy path; on the timeout /
+    // halt path the receiver is dropped and the worker's send
+    // returns Err, which the worker ignores.
+    let (tx, rx) = sync_channel::<R>(0);
+    let _ = thread::Builder::new()
+        .name("freemkv-bounded-syscall".into())
+        .spawn(move || {
+            // Ignore the send error: if we time out (or get halted)
+            // before the worker finishes, the receiver is dropped
+            // and `tx.send` returns Err. Either way, the worker has
+            // nothing more to do.
+            let _ = tx.send(op());
+        });
+
+    let deadline = Instant::now() + timeout;
+    loop {
+        let now = Instant::now();
+        let remaining = deadline.saturating_duration_since(now);
+        let slice = remaining.min(POLL_INTERVAL);
+        match rx.recv_timeout(slice) {
+            Ok(v) => return Ok(v),
+            Err(RecvTimeoutError::Timeout) => {
+                if let Some(h) = halt {
+                    if h.is_cancelled() {
+                        return Err(BoundedError::Halted);
+                    }
+                }
+                if Instant::now() >= deadline {
+                    return Err(BoundedError::Timeout);
+                }
+                // Otherwise: another slice.
+            }
+            Err(RecvTimeoutError::Disconnected) => {
+                // Worker thread spawn failed, or it panicked before
+                // sending. Caller treats this as "no syscall ran" —
+                // typically a no-op + log.
+                return Err(BoundedError::WorkerLost);
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicBool, Ordering};
+
+    #[test]
+    fn op_completes_quickly() {
+        let r = bounded_syscall(None, Duration::from_secs(2), || 42u32);
+        assert!(matches!(r, Ok(42)));
+    }
+
+    #[test]
+    fn op_exceeds_timeout() {
+        // Op sleeps longer than the deadline → Timeout.
+        let r = bounded_syscall(None, Duration::from_millis(300), || {
+            thread::sleep(Duration::from_secs(2));
+            0u32
+        });
+        assert!(matches!(r, Err(BoundedError::Timeout)));
+    }
+
+    #[test]
+    fn halt_fires_during_wait() {
+        let halt = Halt::new();
+        let halt2 = halt.clone();
+        // Flip the halt from a side thread after ~300 ms — long
+        // enough that the receive loop has rolled at least one
+        // 250 ms slice and is sitting in `recv_timeout` again when
+        // the bit flips.
+        thread::spawn(move || {
+            thread::sleep(Duration::from_millis(300));
+            halt2.cancel();
+        });
+        let r = bounded_syscall(Some(&halt), Duration::from_secs(5), || {
+            thread::sleep(Duration::from_secs(5));
+            0u32
+        });
+        assert!(matches!(r, Err(BoundedError::Halted)));
+    }
+
+    #[test]
+    fn worker_panics() {
+        // Worker panics → sender drops without sending → recv sees
+        // Disconnected → WorkerLost. We use an explicit panic in the
+        // op closure rather than `panic!()` from inside the channel
+        // machinery; the spawned thread's panic is contained (no
+        // process abort) because we don't `.join()` it.
+        let r = bounded_syscall(None, Duration::from_secs(2), || -> u32 {
+            panic!("intentional test panic");
+        });
+        assert!(matches!(r, Err(BoundedError::WorkerLost)));
+    }
+
+    #[test]
+    fn halt_already_set_before_call_still_returns_halted() {
+        // Halt observed on the very first poll slice. The op blocks
+        // forever; we must not wait the full timeout to notice the
+        // halt is already set.
+        let halt = Halt::new();
+        halt.cancel();
+        let started = Instant::now();
+        let r = bounded_syscall(Some(&halt), Duration::from_secs(10), || {
+            thread::sleep(Duration::from_secs(10));
+            0u32
+        });
+        assert!(matches!(r, Err(BoundedError::Halted)));
+        // Should bail out within ~1 s; allow 2 s of slack for slow
+        // CI hosts.
+        assert!(
+            started.elapsed() < Duration::from_secs(2),
+            "halt-already-set took {:?}",
+            started.elapsed()
+        );
+    }
+
+    #[test]
+    fn ok_path_takes_no_halt_token() {
+        // Sanity: the `None` halt path is the documented zero-config
+        // form (matches the 0.20.5 `wait_after_with_timeout`
+        // behaviour). Op returns immediately; we must observe Ok.
+        let flag = Arc::new(AtomicBool::new(false));
+        let f2 = flag.clone();
+        let r = bounded_syscall(None, Duration::from_secs(2), move || {
+            f2.store(true, Ordering::Relaxed);
+            "ok"
+        });
+        assert!(matches!(r, Ok("ok")));
+        assert!(flag.load(Ordering::Relaxed));
+    }
+}

src/io/mod.rs

@@ -14,6 +14,7 @@
 //! used by sweep, patch, and mux to overlap reads with writes via a
 //! bounded channel + dedicated consumer thread.
 
+pub(crate) mod bounded;
 mod writeback;
 mod writeback_file;
 

src/io/writeback/linux.rs

@@ -14,9 +14,63 @@
 //! a full chunk's worth of work to flush — the wait is near-instant.
 //! Dirty cache stays bounded at ~2 × `chunk_bytes` and writes drain
 //! continuously instead of in bursts.
+//!
+//! The chunk size is adaptive: we measure the elapsed time of the
+//! `WAIT_AFTER` call over a rolling window of the last 16 chunks and
+//! resize the chunk based on the p95. Slow storage (NFS, network
+//! shares, HDD) sees larger chunks to amortise per-chunk overhead;
+//! fast storage (NVMe) sees smaller chunks to keep cache pressure
+//! tight. Bounds: [4 MiB, 256 MiB].
+//!
+//! ## NFS escape hatch
+//!
+//! `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.
+//!
+//! 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.
+//!
+//! ## 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.
 
+use std::collections::VecDeque;
 use std::fs::File;
 use std::os::unix::io::{AsRawFd, RawFd};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::{Duration, Instant};
+
+const ADAPTIVE_WINDOW: usize = 16;
+const CHUNK_BYTES_MIN: u64 = 4 * 1024 * 1024;
+const CHUNK_BYTES_MAX: u64 = 256 * 1024 * 1024;
+const ADAPTIVE_GROW_MS: u64 = 200;
+const ADAPTIVE_SHRINK_MS: u64 = 20;
+/// Every N chunks, emit a `debug!` snapshot of the current chunk
+/// size so operators tailing the log can see where the autoscaler
+/// settled.
+const SIZE_LOG_INTERVAL: u64 = 32;
+/// Hard upper bound on a single `sync_file_range(WAIT_AFTER)` call.
+/// Beyond this we declare the pipeline degraded and stop calling
+/// WAIT_AFTER for the rest of its life.
+const WAIT_AFTER_TIMEOUT: Duration = Duration::from_secs(30);
 
 pub(crate) struct WritebackPipeline {
     /// Aliases the wrapping `WritebackFile::file`. Only valid for the
@@ -28,6 +82,25 @@ pub(crate) struct WritebackPipeline {
     chunk_bytes: u64,
     last_flush_pos: u64,
     pending: Option<(u64, u64)>,
+    /// Rolling window of recent `WAIT_AFTER` elapsed_ms measurements.
+    wait_after_window: VecDeque<u64>,
+    /// 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,
+    /// 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
+    /// struct and the spawned worker thread (which itself doesn't
+    /// touch the flag) share-via-fd patterns might one day need it;
+    /// today it's effectively a single-owner cell — the `Arc` shape
+    /// keeps the door open for moving the read side into a worker
+    /// without re-plumbing types.
+    degraded: Arc<AtomicBool>,
 }
 
 impl WritebackPipeline {
@@ -36,14 +109,33 @@ impl WritebackPipeline {
     /// 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 {
+        let fd = file.as_raw_fd();
+        let is_nfs = detect_nfs(fd);
+        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" }
+        );
         Self {
-            fd: file.as_raw_fd(),
+            fd,
             chunk_bytes,
             last_flush_pos: start_pos,
             pending: None,
+            wait_after_window: VecDeque::with_capacity(ADAPTIVE_WINDOW),
+            chunk_count: 0,
+            is_nfs,
+            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.
+    #[inline]
+    fn skip_wait(&self) -> bool {
+        self.is_nfs || self.degraded.load(Ordering::Relaxed)
+    }
+
     /// Caller advanced the file position to `pos`. If a chunk boundary
     /// was crossed, kick async writeback for the just-completed chunk
     /// and finalise the previous one.
@@ -51,27 +143,115 @@ impl WritebackPipeline {
         if pos < self.last_flush_pos.saturating_add(self.chunk_bytes) {
             return;
         }
+        let chunk_off = self.last_flush_pos as i64;
+        let chunk_len = (pos - self.last_flush_pos) as i64;
+        let mut wait_ms: u64 = 0;
+        let mut fadvise_ms: u64 = 0;
+        // Async kickoff for the just-completed chunk runs on every
+        // path (NFS, degraded, normal) — it's nominally non-blocking
+        // by spec and gives the kernel an early hint that this range
+        // is ready to flush.
         unsafe {
-            let chunk_off = self.last_flush_pos as i64;
-            let chunk_len = (pos - self.last_flush_pos) as i64;
             libc::sync_file_range(self.fd, chunk_off, chunk_len, libc::SYNC_FILE_RANGE_WRITE);
-            if let Some((prev_off, prev_len)) = self.pending.take() {
-                libc::sync_file_range(
-                    self.fd,
-                    prev_off as i64,
-                    prev_len as i64,
-                    libc::SYNC_FILE_RANGE_WAIT_AFTER,
-                );
-                libc::posix_fadvise(
-                    self.fd,
-                    prev_off as i64,
-                    prev_len as i64,
-                    libc::POSIX_FADV_DONTNEED,
-                );
-            }
-            self.pending = Some((chunk_off as u64, chunk_len as u64));
         }
+        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.
+            } 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;
+                        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.
+                        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)",
+                            WAIT_AFTER_TIMEOUT.as_secs(),
+                            prev_off,
+                            prev_len
+                        );
+                    }
+                }
+            }
+        }
+        self.pending = Some((chunk_off as u64, chunk_len as u64));
         self.last_flush_pos = pos;
+        self.chunk_count += 1;
+        tracing::trace!(
+            target: "mux",
+            "WritebackPipeline chunk off={} len={} sync_file_range_ms={wait_ms} fadvise_ms={fadvise_ms} chunk_bytes={} skip_wait={}",
+            chunk_off,
+            chunk_len,
+            self.chunk_bytes,
+            self.skip_wait(),
+        );
+        if self.chunk_count % SIZE_LOG_INTERVAL == 0 {
+            tracing::debug!(
+                target: "mux",
+                "WritebackPipeline chunk_bytes={} after {} chunks is_nfs={} degraded={}",
+                self.chunk_bytes,
+                self.chunk_count,
+                self.is_nfs,
+                self.degraded.load(Ordering::Relaxed),
+            );
+        }
+    }
+
+    /// Push a new `WAIT_AFTER` measurement into the rolling window
+    /// and, if the window is full, adapt `chunk_bytes` based on p95.
+    fn record_wait(&mut self, wait_ms: u64) {
+        if self.wait_after_window.len() == ADAPTIVE_WINDOW {
+            self.wait_after_window.pop_front();
+        }
+        self.wait_after_window.push_back(wait_ms);
+        if self.wait_after_window.len() < ADAPTIVE_WINDOW {
+            return;
+        }
+        // p95 of 16 samples ≈ sorted[14] (5 % of 16 = 0.8 ≈ 1 above).
+        let mut sorted: Vec<u64> = self.wait_after_window.iter().copied().collect();
+        sorted.sort_unstable();
+        let p95 = sorted[14];
+        let old = self.chunk_bytes;
+        let new = if p95 > ADAPTIVE_GROW_MS && self.chunk_bytes < CHUNK_BYTES_MAX {
+            (self.chunk_bytes * 2).min(CHUNK_BYTES_MAX)
+        } else if p95 < ADAPTIVE_SHRINK_MS && self.chunk_bytes > CHUNK_BYTES_MIN {
+            (self.chunk_bytes / 2).max(CHUNK_BYTES_MIN)
+        } else {
+            self.chunk_bytes
+        };
+        if new != old {
+            self.chunk_bytes = new;
+            tracing::info!(
+                target: "mux",
+                "WritebackPipeline adaptive chunk_bytes {} -> {} p95_ms={p95}",
+                old,
+                new
+            );
+        }
     }
 
     /// Caller is about to seek away from the current write region.
@@ -85,20 +265,90 @@ impl WritebackPipeline {
     /// or when discarding the pipeline.
     pub(crate) fn finalize(&mut self) {
         if let Some((prev_off, prev_len)) = self.pending.take() {
-            unsafe {
-                libc::sync_file_range(
-                    self.fd,
-                    prev_off as i64,
-                    prev_len as i64,
-                    libc::SYNC_FILE_RANGE_WAIT_AFTER,
-                );
-                libc::posix_fadvise(
-                    self.fd,
-                    prev_off as i64,
-                    prev_len as i64,
-                    libc::POSIX_FADV_DONTNEED,
-                );
+            tracing::debug!(
+                target: "mux",
+                "WritebackPipeline finalize chunk off={prev_off} len={prev_len} skip_wait={} is_nfs={} degraded={}",
+                self.skip_wait(),
+                self.is_nfs,
+                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(),
+                    );
+                }
             }
         }
     }
 }
+
+/// 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 {
+    // `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() };
+    let rc = unsafe { libc::fstatfs(fd, &mut buf) };
+    if rc != 0 {
+        let errno = std::io::Error::last_os_error();
+        tracing::warn!(
+            target: "mux",
+            "WritebackPipeline fstatfs(fd={fd}) failed: {errno} — defaulting is_nfs=false",
+        );
+        return false;
+    }
+    // `f_type` is signed (`__fsword_t`) on glibc and unsigned
+    // (`c_ulong`) on musl. Cast both sides to i64 for a portable
+    // comparison.
+    let f_type = buf.f_type as i64;
+    let nfs_magic = libc::NFS_SUPER_MAGIC as i64;
+    f_type == nfs_magic
+}
+
+/// Run `sync_file_range(WAIT_AFTER)` on a worker thread and wait up
+/// to [`WAIT_AFTER_TIMEOUT`] for it to return. `Some(elapsed_ms)` on
+/// success; `None` on timeout. On timeout the worker thread is
+/// 0.20.6 generalizes the worker-thread + recv_timeout pattern into
+/// [`crate::io::bounded::bounded_syscall`]; this helper now just adapts
+/// the generic primitive to the WAIT_AFTER call shape (returns elapsed_ms
+/// instead of the syscall's `()` return, treats `WorkerLost` as a benign
+/// no-op to match the original semantics).
+fn wait_after_with_timeout(fd: RawFd, off: u64, len: u64) -> Option<u64> {
+    let started = Instant::now();
+    match crate::io::bounded::bounded_syscall(None, WAIT_AFTER_TIMEOUT, move || unsafe {
+        libc::sync_file_range(fd, off as i64, len as i64, libc::SYNC_FILE_RANGE_WAIT_AFTER);
+    }) {
+        Ok(()) => Some(started.elapsed().as_millis() as u64),
+        Err(crate::io::bounded::BoundedError::Timeout)
+        | Err(crate::io::bounded::BoundedError::Halted) => None,
+        Err(crate::io::bounded::BoundedError::WorkerLost) => {
+            // Worker thread spawn failed or panicked before sending.
+            // Treat as a benign success (no syscall ran) rather than
+            // a degrade trigger — falling through with elapsed_ms=0
+            // matches the no-op behaviour.
+            Some(0)
+        }
+    }
+}

src/io/writeback_file.rs

@@ -54,11 +54,57 @@ impl WritebackFile {
     /// and wrap it. Convenience for the common
     /// `File::create(path)` + `WritebackFile::new(file)` pair so callers
     /// don't have to assemble a `File` first.
+    ///
+    /// Callers that know the target output size should prefer
+    /// [`Self::create_with_size_hint`] so the kernel can pre-reserve
+    /// extents.
+    #[allow(dead_code)]
     pub(crate) fn create(path: &Path) -> io::Result<Self> {
         let file = File::create(path)?;
         Self::new(file)
     }
 
+    /// Like [`Self::create`] but pre-reserves `size_bytes` of disk
+    /// space via `fallocate(FALLOC_FL_KEEP_SIZE)` on Linux. The
+    /// reported file size is unchanged (writes still grow the file
+    /// naturally) — only the on-disk extent allocation is preallocated,
+    /// which reduces extent fragmentation on large sequential writes
+    /// (mux output, especially on slow storage / NFS).
+    ///
+    /// On macOS / Windows the size hint is ignored and this is
+    /// equivalent to `create`.
+    pub(crate) fn create_with_size_hint(path: &Path, size_bytes: u64) -> io::Result<Self> {
+        let file = File::create(path)?;
+        #[cfg(target_os = "linux")]
+        {
+            use std::os::unix::io::AsRawFd;
+            // FALLOC_FL_KEEP_SIZE = 0x01 — keep the reported file
+            // size at 0 (writes grow it normally) while still
+            // pre-reserving the extents.
+            let rc = unsafe {
+                libc::fallocate(
+                    file.as_raw_fd(),
+                    libc::FALLOC_FL_KEEP_SIZE,
+                    0,
+                    size_bytes as i64,
+                )
+            };
+            tracing::debug!(
+                target: "mux",
+                "WritebackFile fallocate size_hint={size_bytes} rc={rc} ok={}",
+                rc == 0
+            );
+        }
+        #[cfg(not(target_os = "linux"))]
+        {
+            tracing::debug!(
+                target: "mux",
+                "WritebackFile fallocate size_hint={size_bytes} skipped (non-linux)"
+            );
+        }
+        Self::new(file)
+    }
+
     /// Open an existing file at `path` for writing (no truncation) and
     /// wrap it. Mirrors `File::open` semantics for the writable case
     /// — used by patch / resume paths that mutate an existing ISO in
@@ -70,9 +116,49 @@ impl WritebackFile {
 
     /// Drain in-flight writeback then issue a full fsync. Use this in
     /// place of `File::sync_all`.
+    ///
+    /// The final fsync is wrapped in
+    /// [`crate::io::bounded::bounded_syscall`] with a 60 s deadline.
+    /// fsync on a wedged NFS server (or a degraded local disk) can
+    /// hang the calling thread; the wrapper ensures the worst case is
+    /// 60 s + log-and-continue rather than indefinite. On timeout the
+    /// 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<()> {
         self.pipeline.finalize();
-        self.file.sync_all()
+        #[cfg(unix)]
+        {
+            use std::os::unix::io::AsRawFd;
+            use std::time::Duration;
+            let fd = self.file.as_raw_fd();
+            match crate::io::bounded::bounded_syscall(
+                None,
+                Duration::from_secs(60),
+                move || -> io::Result<()> {
+                    let rc = unsafe { libc::fsync(fd) };
+                    if rc == 0 {
+                        Ok(())
+                    } else {
+                        Err(io::Error::last_os_error())
+                    }
+                },
+            ) {
+                Ok(inner) => inner,
+                Err(crate::io::bounded::BoundedError::Timeout) => {
+                    tracing::error!(
+                        target: "mux",
+                        "WritebackFile::sync_all fsync timed out after 60s; kernel will flush on close (best-effort)"
+                    );
+                    Ok(())
+                }
+                Err(crate::io::bounded::BoundedError::Halted) => Ok(()),
+                Err(crate::io::bounded::BoundedError::WorkerLost) => Ok(()),
+            }
+        }
+        #[cfg(not(unix))]
+        {
+            self.file.sync_all()
+        }
     }
 }
 
@@ -104,6 +190,20 @@ impl Seek for WritebackFile {
         // before every write, and we don't want that to drain the
         // pipeline on every iteration.
         if p != self.pos {
+            // Diagnostic for the NFS 73 % 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;
         }

src/mux/resolve.rs

@@ -250,11 +250,13 @@ pub fn output(
             // writeback) so a UHD-scale MKV mux to slow / network-attached
             // staging doesn't hit the dirty-page burst pathology that
             // sweep already side-steps. BufWriter sits on top to coalesce
-            // mux's many small EBML element writes.
+            // mux's many small EBML element writes. Pre-reserve the
+            // target's worth of extents on Linux via fallocate(KEEP_SIZE)
+            // to reduce extent fragmentation during the mux.
             let writer: Box<dyn super::WriteSeek + Send> =
                 Box::new(std::io::BufWriter::with_capacity(
                     IO_BUF_SIZE,
-                    crate::io::WritebackFile::create(path)?,
+                    crate::io::WritebackFile::create_with_size_hint(path, title.size_bytes)?,
                 ));
             Ok(Box::new(MkvStream::create(writer, title)?))
         }
@@ -262,7 +264,7 @@ pub fn output(
             validate_file_path(path, "m2ts")?;
             let writer = std::io::BufWriter::with_capacity(
                 IO_BUF_SIZE,
-                crate::io::WritebackFile::create(path)?,
+                crate::io::WritebackFile::create_with_size_hint(path, title.size_bytes)?,
             );
             Ok(Box::new(M2tsStream::create(writer, title)?))
         }

src/sector/file.rs

@@ -15,6 +15,12 @@ use crate::error::{Error, Result};
 
 use super::{SectorSink, SectorSource};
 
+/// Bytes-read threshold per `posix_fadvise(DONTNEED)` drop on the
+/// read side. Mirrors the writeback chunk size so the read-side
+/// page cache stays bounded the same way the write side does.
+#[cfg(target_os = "linux")]
+const READ_DROP_CHUNK_BYTES: u64 = 32 * 1024 * 1024;
+
 /// SectorSource backed by a file (ISO image).
 ///
 /// Seeks to `lba * 2048`, reads `count * 2048` bytes per call. The
@@ -27,6 +33,14 @@ use super::{SectorSink, SectorSource};
 pub struct FileSectorSource {
     file: File,
     capacity: u32,
+    /// Bytes read since the last `posix_fadvise(DONTNEED)` drop.
+    /// Only updated on Linux; on other targets it stays at 0.
+    #[cfg(target_os = "linux")]
+    bytes_read_since_drop: u64,
+    /// Byte offset at which the current drop window starts (the
+    /// next `posix_fadvise(DONTNEED)` call drops from here).
+    #[cfg(target_os = "linux")]
+    drop_window_start: u64,
 }
 
 impl FileSectorSource {
@@ -45,7 +59,26 @@ impl FileSectorSource {
             .into());
         }
         let capacity = sectors as u32;
-        Ok(Self { file, capacity })
+
+        // Hint sequential access on Linux so the kernel's readahead
+        // window widens for the ISO sweep. Best-effort: return value
+        // is ignored. On macOS / Windows this is a no-op.
+        #[cfg(target_os = "linux")]
+        {
+            use std::os::unix::io::AsRawFd;
+            unsafe {
+                libc::posix_fadvise(file.as_raw_fd(), 0, 0, libc::POSIX_FADV_SEQUENTIAL);
+            }
+        }
+
+        Ok(Self {
+            file,
+            capacity,
+            #[cfg(target_os = "linux")]
+            bytes_read_since_drop: 0,
+            #[cfg(target_os = "linux")]
+            drop_window_start: 0,
+        })
     }
 }
 
@@ -75,6 +108,38 @@ impl SectorSource for FileSectorSource {
         self.file
             .read_exact(&mut buf[..bytes])
             .map_err(|e| Error::IoError { source: e })?;
+
+        // On Linux, periodically drop the just-read region from the
+        // page cache to keep cache pressure bounded during multi-GB
+        // sequential ISO reads. Mirrors the write-side pipeline.
+        #[cfg(target_os = "linux")]
+        {
+            use std::os::unix::io::AsRawFd;
+            self.bytes_read_since_drop += bytes as u64;
+            if self.bytes_read_since_drop >= READ_DROP_CHUNK_BYTES {
+                let drop_start = self.drop_window_start;
+                let drop_len = self.bytes_read_since_drop;
+                let t0 = std::time::Instant::now();
+                unsafe {
+                    libc::posix_fadvise(
+                        self.file.as_raw_fd(),
+                        drop_start as i64,
+                        drop_len as i64,
+                        libc::POSIX_FADV_DONTNEED,
+                    );
+                }
+                let elapsed_ms = t0.elapsed().as_millis();
+                let start_lba = drop_start / 2048;
+                let end_lba = (drop_start + drop_len) / 2048;
+                tracing::trace!(
+                    target: "mux",
+                    "FileSectorSource fadvise DONTNEED lba=[{start_lba}..{end_lba}) bytes={drop_len} elapsed_ms={elapsed_ms}"
+                );
+                self.drop_window_start = drop_start + drop_len;
+                self.bytes_read_since_drop = 0;
+            }
+        }
+
         Ok(bytes)
     }
 }