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.5"
+version = "0.20.6"
 edition = "2024"
 rust-version = "1.86"
 license = "AGPL-3.0-only"

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

@@ -56,8 +56,6 @@ use std::fs::File;
 use std::os::unix::io::{AsRawFd, RawFd};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::mpsc::{RecvTimeoutError, sync_channel};
-use std::thread;
 use std::time::{Duration, Instant};
 
 const ADAPTIVE_WINDOW: usize = 16;
@@ -332,34 +330,20 @@ fn detect_nfs(fd: RawFd) -> bool {
 /// 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
-/// intentionally leaked — it'll unwind whenever the syscall
+/// 0.20.6 generalizes the worker-thread + recv_timeout pattern into
-/// eventually returns or the process exits.
+/// [`crate::io::bounded::bounded_syscall`]; this helper now just adapts
-///
+/// the generic primitive to the WAIT_AFTER call shape (returns elapsed_ms
-/// Channel capacity is 0 (rendezvous). The worker sends `()` once
+/// instead of the syscall's `()` return, treats `WorkerLost` as a benign
-/// the syscall returns; if we time out first, the send blocks
+/// no-op to match the original semantics).
-/// forever inside the leaked worker — fine, the receiver is gone
-/// and the OS reaps the thread at process exit.
 fn wait_after_with_timeout(fd: RawFd, off: u64, len: u64) -> Option<u64> {
-    let (tx, rx) = sync_channel::<()>(0);
     let started = Instant::now();
-    // `RawFd` is `Copy` + `Send`; `off`/`len` are `u64`. Nothing
+    match crate::io::bounded::bounded_syscall(None, WAIT_AFTER_TIMEOUT, move || unsafe {
-    // borrows from the caller — the worker can safely outlive this
-    // function on timeout.
-    let _ = thread::Builder::new()
-        .name("freemkv-writeback-wait".into())
-        .spawn(move || {
-            unsafe {
         libc::sync_file_range(fd, off as i64, len as i64, libc::SYNC_FILE_RANGE_WAIT_AFTER);
-            }
+    }) {
-            // If the receiver is gone (we timed out) this send
-            // returns Err — fine, we just drop and the worker
-            // exits.
-            let _ = tx.send(());
-        });
-    match rx.recv_timeout(WAIT_AFTER_TIMEOUT) {
         Ok(()) => Some(started.elapsed().as_millis() as u64),
-        Err(RecvTimeoutError::Timeout) => None,
+        Err(crate::io::bounded::BoundedError::Timeout)
-        Err(RecvTimeoutError::Disconnected) => {
+        | 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

src/io/writeback_file.rs

@@ -116,10 +116,50 @@ 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();
+        #[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()
         }
+    }
 }
 
 impl Write for WritebackFile {