v0.20.4: mux performance + observability — universal across storage

Four targeted changes to maximize mux throughput regardless of storage
backend (local SSD, local HDD, NFS, network share) and surface enough
log data to diagnose 'mux slow' reports without a re-rip:

1. POSIX_FADV_SEQUENTIAL on FileSectorSource::open (Linux only).
   Widens the kernel readahead window for sequential ISO reads. One
   syscall at open, free on every storage type.

2. POSIX_FADV_DONTNEED on the ISO read side after every 32 MiB chunk.
   Mirrors the writeback DONTNEED that already runs on the write
   side. Keeps the read-side page cache bounded during multi-GB ISO
   reads — eliminates the OOM-pressure / eviction-storm risk on
   long mux runs. Linux only; per-drop trace at target="mux".

3. WritebackFile::create_with_size_hint(path, size_bytes) calls
   fallocate(FALLOC_FL_KEEP_SIZE) on Linux to pre-reserve extents
   for the output. Reported file size stays 0 (writes grow it
   naturally) but the on-disk extent allocation is contiguous —
   reduces extent fragmentation for big sequential muxes. Wired
   into mkv:// and m2ts:// output paths via DiscTitle::size_bytes.
   No-op on macOS/Windows; old create() kept with #[allow(dead_code)]
   for callers without a size hint.

4. Adaptive WRITEBACK_CHUNK_BYTES in the Linux writeback pipeline.
   Tracks sync_file_range(WAIT_AFTER) elapsed_ms in a rolling
   16-sample window. p95 > 200 ms → double chunk size (cap 256 MiB).
   p95 < 20 ms → halve (floor 4 MiB). One algorithm, both
   fast-storage (small chunks, responsive) and slow-storage (big
   chunks, fewer commit round-trips) optimized. Per-chunk trace +
   per-32-chunk debug snapshot + info-on-resize so an operator can
   see where the autoscaler settled.

All four are universal — no storage-type detection, no env vars to
flip, no per-deploy tuning required. Total +201/-6 across four files.

src/io/writeback/linux.rs

@@ -14,9 +14,28 @@
 //! 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].
 
+use std::collections::VecDeque;
 use std::fs::File;
 use std::os::unix::io::{AsRawFd, RawFd};
+use std::time::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;
 
 pub(crate) struct WritebackPipeline {
     /// Aliases the wrapping `WritebackFile::file`. Only valid for the
@@ -28,6 +47,11 @@ 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,
 }
 
 impl WritebackPipeline {
@@ -41,6 +65,8 @@ impl WritebackPipeline {
             chunk_bytes,
             last_flush_pos: start_pos,
             pending: None,
+            wait_after_window: VecDeque::with_capacity(ADAPTIVE_WINDOW),
+            chunk_count: 0,
         }
     }
 
@@ -51,27 +77,83 @@ 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;
         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() {
+                let t_wait = Instant::now();
                 libc::sync_file_range(
                     self.fd,
                     prev_off as i64,
                     prev_len as i64,
                     libc::SYNC_FILE_RANGE_WAIT_AFTER,
                 );
+                wait_ms = t_wait.elapsed().as_millis() as u64;
+                let t_fadv = Instant::now();
                 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);
             }
             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={}",
+            chunk_off,
+            chunk_len,
+            self.chunk_bytes
+        );
+        if self.chunk_count % SIZE_LOG_INTERVAL == 0 {
+            tracing::debug!(
+                target: "mux",
+                "WritebackPipeline chunk_bytes={} after {} chunks",
+                self.chunk_bytes,
+                self.chunk_count
+            );
+        }
+    }
+
+    /// 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.