Reapply "io/writeback_file: restore Phase 2.5 (writer thread + bounded ring)"

This reverts commit 579c94bd23.

src/io/writeback_file/mod.rs

@@ -31,22 +31,67 @@
 //! happens once at the bottom of this file via cfg-gated `mod` decls.
 //! No inline `#[cfg(target_os = "...")]` in the business-logic above.
 //!
-//! ## Write path
+//! ## Phase 2.5 — write-side flatness (writer thread)
 //!
-//! Writes are direct passthrough to the underlying `File` (no writer
+//! `WritebackFile` is split into a thin muxer-facing handle and a
-//! thread, no ring, no batching). Empirically the Phase-2.5
+//! dedicated writer thread that owns the real `File` + writeback
-//! writer-thread architecture introduced a ~60% mux throughput
+//! pipeline. The muxer's `Write::write` and `Seek::seek` calls return as
-//! regression on NFS bidirectional workloads; reverting the write path
+//! soon as the byte handoff to a bounded SPSC ring completes; the writer
-//! to direct passthrough restores the 0.20.7 baseline. The writeback
+//! thread executes the real syscalls (incl. `sync_file_range(WAIT_AFTER)`
-//! pipeline still runs (it's called inline from `write` / `write_all` /
+//! on Linux) without ever blocking the muxer on a kernel commit.
-//! `seek`) so the bounded-cache invariant on Linux is preserved.
 //!
-//! ## Halt-safety
+//! ### Backpressure
 //!
-//! `sync_all` runs the per-OS durable-flush primitive, which on
+//! The ring is byte-bounded at [`RING_CAPACITY_BYTES`]. When the ring is
-//! Linux/macOS is wrapped in [`crate::io::bounded::bounded_syscall`]
+//! full the muxer's `write` blocks on a condvar until the writer thread
-//! with a 60 s deadline. A wedged NFS server cannot trap the muxer
+//! drains enough bytes to admit the next chunk. **Backpressure on
-//! indefinitely on the final fsync.
+//! 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.
 
 #[cfg(target_os = "linux")]
 mod linux;
@@ -66,9 +111,13 @@ 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;
 
@@ -77,25 +126,206 @@ 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 {
-    file: File,
+    shared: Arc<Shared>,
-    pipeline: WritebackPipeline,
+    /// Joined on `Drop` after `Finish` so the writer thread's exit is
-    pos: u64,
+    /// 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,
 }
 
 impl WritebackFile {
     /// Wrap an open `File`. The current OS file position is queried
-    /// once so the pipeline starts tracking from wherever the file
+    /// once so the writer thread starts tracking from wherever the
-    /// already is (typically 0 for fresh files; non-zero for resumed
+    /// file already is (typically 0 for fresh files; non-zero for
-    /// or appended files).
+    /// resumed or appended files).
     pub(crate) fn new(mut file: File) -> io::Result<Self> {
         let pos = file.stream_position()?;
-        let pipeline = WritebackPipeline::new(&file, pos, WRITEBACK_CHUNK_BYTES);
+        Ok(Self::spawn(file, pos))
-        Ok(Self {
-            file,
-            pipeline,
-            pos,
-        })
     }
 
     /// Create a new file at `path` (truncating any existing contents)
@@ -139,84 +369,432 @@ 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`.
     ///
-    /// The final durable flush is wrapped in
+    /// Blocks the calling thread until the ring is fully drained AND
-    /// [`crate::io::bounded::bounded_syscall`] (per the per-OS module)
+    /// the per-OS durable-flush primitive has returned. The flush
-    /// with a 60 s deadline on Linux/macOS — a wedged NFS server cannot
+    /// itself runs on the writer thread, wrapped in
-    /// trap the calling thread indefinitely. On timeout the page cache
+    /// [`crate::io::bounded::bounded_syscall`] (60 s deadline on
-    /// is left to the kernel's normal flush-on-close path — best
+    /// Linux + macOS); a wedged NFS server cannot trap the muxer.
-    /// effort, but bounded.
     pub(crate) fn sync_all(&mut self) -> io::Result<()> {
-        self.pipeline.finalize();
+        let (tx, rx) = sync_channel::<io::Result<()>>(0);
-        platform::durable_sync(&self.file)
+        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(())
     }
 }
 
 impl Write for WritebackFile {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        let n = self.file.write(buf)?;
+        // `Write::write` is allowed to be partial; we always accept
-        self.pos += n as u64;
+        // the full slice (handoff is in-process) and report `buf.len()`.
-        self.pipeline.note_progress(self.pos);
+        // 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;
         Ok(n)
     }
 
     fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
-        self.file.write_all(buf)?;
+        // `Write::write_all` default delegates to `write` in a loop. We
-        self.pos += buf.len() as u64;
+        // can do better: a single handoff per call, never partial. Same
-        self.pipeline.note_progress(self.pos);
+        // 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;
         Ok(())
     }
 
     fn flush(&mut self) -> io::Result<()> {
-        self.file.flush()
+        // 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)
     }
 }
 
 impl Seek for WritebackFile {
     fn seek(&mut self, from: SeekFrom) -> io::Result<u64> {
-        let p = self.file.seek(from)?;
+        // The muxer's logical position must update synchronously so
-        // Only treat seeks that actually move the position as
+        // subsequent `write` calls advance from the right base, but the
-        // boundaries — sweep does a redundant `seek(Current(pos))`
+        // writer thread is the only place that has the authoritative
-        // before every write, and we don't want that to drain the
+        // OS file position. We model the muxer's `muxer_pos` purely
-        // pipeline on every iteration.
+        // from `SeekFrom::Start(n)` (the dominant case for MKV
-        if p != self.pos {
+        // backpatch) and bounce other variants through to the writer
-            // Diagnostic for the NFS mux hang: the MKV format requires
+        // by querying its current position via a synchronous round.
-            // the muxer to seek back occasionally (cluster size
+        let new_pos = match from {
-            // patching, Cues index write, Segment header backpatch).
+            SeekFrom::Start(n) => n,
-            // Each such seek invalidates the writeback chunk tracking
+            SeekFrom::Current(d) => {
-            // and forces a finalize → WAIT_AFTER on the in-flight
+                let base = self.muxer_pos as i64;
-            // chunk. Logging the seek delta lets us correlate hang
+                let p = base
-            // offsets with specific muxer operations.
+                    .checked_add(d)
-            let from_pos = self.pos;
+                    .ok_or_else(|| io::Error::from(io::ErrorKind::InvalidInput))?;
-            let to_pos = p;
+                if p < 0 {
-            let delta: i64 = (to_pos as i64).wrapping_sub(from_pos as i64);
+                    return Err(io::Error::from(io::ErrorKind::InvalidInput));
-            tracing::debug!(
-                target: "mux",
-                "WritebackFile seek from={from_pos} to={to_pos} delta={delta}"
-            );
-            self.pipeline.handle_seek(p);
-            self.pos = p;
                 }
-        Ok(p)
+                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)
     }
 }
 
 impl Drop for WritebackFile {
     fn drop(&mut self) {
-        // Run the pipeline's tail finalize so the last in-flight chunk
+        // Send a final `Finish` so the writer drains the ring (running
-        // gets its `WAIT_AFTER` + `posix_fadvise(DONTNEED)`. Without
+        // the pipeline's tail finalize for the last in-flight chunk)
-        // this, callers that drop a `WritebackFile` without calling
+        // before exiting. `sync_all` is *not* called from here — the
-        // `sync_all` (panic, early-return, idiomatic `let _ = w;`)
+        // existing pre-Phase-2.5 contract is "Drop runs finalize, not
-        // leave the trailing chunk in cache; the kernel still flushes
+        // fsync", and Drop returning an io::Error is impossible anyway.
-        // on close, but the bounded-cache invariant fails at the tail.
+        let (tx, rx) = sync_channel::<()>(0);
-        // We deliberately do *not* call `self.file.sync_all()` here —
+        // If the writer thread already exited (sticky error path, or
-        // close already triggers a flush, and an `fsync` from `Drop`
+        // an earlier panic), the queue push will fail with
-        // would silently swallow its `io::Error` anyway. `finalize` is
+        // BrokenPipe; we treat that as "nothing to drain" and proceed
-        // idempotent so an explicit `sync_all` followed by drop is
+        // to join.
-        // still safe.
+        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<()> {
+        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();
     }
 }
 
@@ -239,13 +817,13 @@ mod tests {
         {
             let mut w = WritebackFile::create(&p).unwrap();
             w.write_all(b"hello world").unwrap();
-            // Drop drains the pipeline tail.
+            // Drop drains the ring.
         }
         assert_eq!(read_back(&p), b"hello world");
     }
 
     #[test]
-    fn sync_all_drains_and_flushes() {
+    fn sync_all_blocks_until_ring_drains() {
         let dir = tempfile::tempdir().unwrap();
         let p = dir.path().join("b.bin");
         let mut w = WritebackFile::create(&p).unwrap();
@@ -253,8 +831,7 @@ mod tests {
             w.write_all(&[0x5au8; 1024]).unwrap();
         }
         // After sync_all, the bytes MUST be visible to a separate
-        // reader. The pipeline has been finalised and durable-sync has
+        // reader. The ring has been drained and durable-sync has run.
-        // run.
         w.sync_all().unwrap();
         let bytes = read_back(&p);
         assert_eq!(bytes.len(), 32 * 1024);
@@ -263,17 +840,20 @@ mod tests {
     }
 
     #[test]
-    fn seek_then_patch_roundtrip() {
+    fn in_window_seek_then_patch_roundtrip() {
-        // Write A; seek back; patch with B; read back; the patch lands
+        // Write A; seek back inside the active-cluster window; patch
-        // at the right offset.
+        // 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);
@@ -284,10 +864,67 @@ 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` should not panic or reorder; verify the bytes
+        // `Write::flush` is a no-op on the writer side but must not
-        // land in order through interleaved flushes.
+        // panic or leak. Run an interleaved sequence and verify the
+        // bytes still land in order.
         let dir = tempfile::tempdir().unwrap();
         let p = dir.path().join("f.bin");
         let mut w = WritebackFile::create(&p).unwrap();
@@ -300,4 +937,90 @@ 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);
+    }
 }