v0.22.0: bump version

This reverts commit 2a33364253.

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "libfreemkv"
-version = "0.20.10"
+version = "0.22.0"
 edition = "2024"
 rust-version = "1.86"
 license = "AGPL-3.0-only"
@@ -28,6 +28,12 @@ base64 = "0.22.1"
 # under project docs ("Acceptable strings: debug/trace logging"). Consumers (autorip)
 # wire a tracing subscriber and pipe events into the JSONL debug log.
 tracing = "0.1"
+# Bounded MPSC channel with kernel-wakeup send_timeout. Used by `io::pipeline`
+# so the halt-aware send/finish loops can BLOCK on consumer drain instead of
+# polling — the 50 ms poll cadence of the previous mpsc-based impl capped mux
+# throughput at ~1 MB/s (see (internal)/memory/
+# feedback_send_with_halt_poll_throttle.md, 0.21.7).
+crossbeam-channel = "0.5"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 libc = "0.2"

src/io/file_sector_source/linux.rs

@@ -1,15 +1,46 @@
-//! Linux: hint the kernel that this fd will be read sequentially so
-//! readahead widens. `posix_fadvise(POSIX_FADV_SEQUENTIAL)` is a hint,
-//! not a guarantee — the kernel still owns the policy decision.
+//! Linux read-side platform hooks: sequential-access hint at open +
+//! periodic page-cache eviction during streaming reads.
+//!
+//! ## Why both
+//!
+//! `POSIX_FADV_SEQUENTIAL` at open widens the kernel's readahead window
+//! so each pread aggregates into fewer NFS round-trips. `DONTNEED` on
+//! the consumed window (called periodically by the caller) drops the
+//! already-read pages from the page cache so an 85 GB streaming ISO
+//! read doesn't fill memory and starve concurrent writes (the MKV
+//! output during mux). Together they mirror the write-side
+//! WritebackPipeline's policy.
+//!
+//! ## History
+//!
+//! Pre-Phase-1 (0.20.7 baseline) had both. Phase 1's introduction of
+//! `FileSectorSource` silently dropped the read-side DONTNEED, and
+//! 0.21.2's revert of `SEQUENTIAL` (mistakenly attributing a regression
+//! to it) removed the hint. Net effect: 85 GB of ISO reads pinned in
+//! the page cache + no readahead widening → mux throughput collapse
+//! from 18 MB/s historical to 2.7-8 MB/s on 0.21.x. Restored in 0.21.6.
 
 use std::fs::File;
 use std::os::unix::io::AsRawFd;
 
 pub(super) fn hint_sequential(file: &File, _len_bytes: u64) {
     // Best-effort: return value ignored. A fadvise failure has no
-    // user-observable consequence (reads still work, just without the
-    // widened readahead window).
+    // user-observable consequence.
     unsafe {
         libc::posix_fadvise(file.as_raw_fd(), 0, 0, libc::POSIX_FADV_SEQUENTIAL);
     }
 }
+
+/// Drop pages in the half-open byte range `[start, start+len)` from
+/// the page cache. Called periodically by `read_sectors` to bound the
+/// read-side page cache pressure.
+pub(super) fn drop_window(file: &File, start: u64, len: u64) {
+    unsafe {
+        libc::posix_fadvise(
+            file.as_raw_fd(),
+            start as i64,
+            len as i64,
+            libc::POSIX_FADV_DONTNEED,
+        );
+    }
+}

src/io/file_sector_source/macos.rs

@@ -36,3 +36,11 @@ pub(super) fn hint_sequential(file: &File, len_bytes: u64) {
         libc::fcntl(file.as_raw_fd(), F_RDADVISE, &mut ra);
     }
 }
+
+/// macOS has no direct `POSIX_FADV_DONTNEED` equivalent for a byte
+/// range. `fcntl(F_NOCACHE)` would disable caching globally on the fd
+/// (too coarse — we want the unread region to still benefit). Best
+/// approximation: no-op. macOS's unified buffer cache is generally
+/// less prone to the pin-everything pathology that triggers the
+/// regression on Linux NFS clients.
+pub(super) fn drop_window(_file: &File, _start: u64, _len: u64) {}

src/io/file_sector_source/mod.rs

@@ -65,8 +65,15 @@ use crate::sector::SectorSource;
 /// than per-sector pread, and large enough to coast through a typical
 /// NFS server commit blip.
 ///
+/// 0.21.2: shrunk from 32 MiB → 4 MiB. On NFS-backed ISOs with
+/// concurrent NFS writes (the mux phase), a 32 MiB refill bursts the
+/// TCP connection hard enough to starve the writer thread, observed
+/// empirically as a ~3× drop in sustained mux throughput on the
+/// rip1/unraid-1 setup. 4 MiB matches `rsize=1 MiB` × 4 round-trips
+/// and interleaves cleanly with writes.
+///
 /// Tweakable. Named const, not a magic number.
-pub const READAHEAD_BUF_BYTES: usize = 32 * 1024 * 1024;
+pub const READAHEAD_BUF_BYTES: usize = 4 * 1024 * 1024;
 
 const SECTOR_SIZE: usize = 2048;
 /// Sectors per refill: [`READAHEAD_BUF_BYTES`] / [`SECTOR_SIZE`]. The
@@ -80,24 +87,39 @@ const BUF_SECTORS: u32 = (READAHEAD_BUF_BYTES / SECTOR_SIZE) as u32;
 /// `read_sectors` is satisfied from the buffer when possible; otherwise
 /// a full-buffer refill is issued at the requested LBA's position and
 /// the call is re-tried against the freshly populated window.
+/// Bytes-read threshold per `posix_fadvise(DONTNEED)` drop on the
+/// read side. Mirrors `WRITEBACK_CHUNK_BYTES` so the read-side page
+/// cache stays bounded the same way the write side does.
+///
+/// 0.21.6: re-added after empirical discovery that Phase 1 had silently
+/// dropped this from the pre-Phase-1 (0.20.7) hot path. Without it,
+/// 85 GB of streaming ISO reads pin the entire file in the kernel page
+/// cache, starving the MKV writeback and collapsing mux throughput
+/// (observed: 2.7 MB/s mux on 0.21.5 vs. 70 MB/s isolated NFS reads).
+const READ_DROP_CHUNK_BYTES: u64 = 32 * 1024 * 1024;
+
 pub struct FileSectorSource {
     file: File,
     /// Total file size in sectors. Constant after construction;
     /// surfaced via [`SectorSource::capacity_sectors`].
     capacity: u32,
-    /// Heap-allocated read-ahead buffer. Boxed slice (not `Vec`) so
-    /// the allocation is sized exactly once and the layout is
-    /// stable. Length is at most [`READAHEAD_BUF_BYTES`].
+    /// 0.21.3+: the app-level buffer is no longer touched on the hot
+    /// path (every `read_sectors` is a direct pread). The fields are
+    /// retained so a future per-source-type policy (e.g. a local-disk
+    /// source where batched reads ARE beneficial) can re-enable
+    /// buffering cleanly without re-plumbing the struct.
+    #[allow(dead_code)]
     buf: Box<[u8]>,
-    /// First LBA covered by `buf`. `buf_len_sectors == 0` means the
-    /// buffer is empty (e.g. immediately after construction or a
-    /// backward seek); the value of `buf_start_lba` is then
-    /// meaningless.
+    #[allow(dead_code)]
     buf_start_lba: u32,
-    /// Number of valid sectors in `buf`. `[buf_start_lba,
-    /// buf_start_lba + buf_len_sectors)` is the half-open LBA range
-    /// the buffer currently caches.
     buf_len_sectors: u32,
+    /// 0.21.6: bytes read since the last DONTNEED drop. Drives the
+    /// per-`READ_DROP_CHUNK_BYTES` page-cache eviction in read_sectors.
+    bytes_read_since_drop: u64,
+    /// 0.21.6: file offset at which the current drop window starts.
+    /// The next DONTNEED drops from `drop_window_start` for
+    /// `bytes_read_since_drop` bytes.
+    drop_window_start: u64,
 }
 
 impl FileSectorSource {
@@ -138,11 +160,14 @@ impl FileSectorSource {
             buf,
             buf_start_lba: 0,
             buf_len_sectors: 0,
+            bytes_read_since_drop: 0,
+            drop_window_start: 0,
         })
     }
 
     /// True if `[lba, lba + count)` is wholly inside the current
     /// buffer window. `count == 0` is vacuously true.
+    #[allow(dead_code)]
     fn buffer_covers(&self, lba: u32, count: u32) -> bool {
         if self.buf_len_sectors == 0 {
             return false;
@@ -158,6 +183,7 @@ impl FileSectorSource {
     /// Refill the buffer so it starts at `lba`. Read as many sectors
     /// as we have buffer space AND file capacity for. Caller has
     /// already checked `lba < capacity`.
+    #[allow(dead_code)]
     fn refill(&mut self, lba: u32) -> Result<()> {
         debug_assert!(lba < self.capacity, "refill past capacity");
         // Don't read past EOF — clamp the request to remaining
@@ -200,33 +226,43 @@ impl SectorSource for FileSectorSource {
         if count == 0 {
             return Ok(0);
         }
-        // Refill if the requested range isn't entirely buffered.
-        // `buffer_covers` also handles the empty-buffer case
-        // (buf_len_sectors == 0).
-        if !self.buffer_covers(lba, count) {
-            // A request larger than the buffer itself can never fit;
-            // fall back to a one-shot direct pread for that pathological
-            // case so callers can't deadlock the source.
-            if count > BUF_SECTORS {
-                let offset = lba as u64 * SECTOR_SIZE as u64;
-                self.file
-                    .seek(SeekFrom::Start(offset))
-                    .map_err(|e| Error::IoError { source: e })?;
-                self.file
-                    .read_exact(&mut out[..bytes])
-                    .map_err(|e| Error::IoError { source: e })?;
-                // Invalidate buffer state — we bypassed it, the
-                // window is no longer authoritative for this LBA.
-                self.buf_len_sectors = 0;
-                return Ok(bytes);
-            }
-            self.refill(lba)?;
+        // 0.21.3: bypass the application-level buffer entirely.
+        //
+        // Empirically the 32 MiB readahead window (0.21.0–0.21.1) and the
+        // 4 MiB shrink (0.21.2) both regressed mux throughput vs the
+        // pre-Phase-1 0.20.7 baseline on NFS bidirectional workloads
+        // (sweep ~25 MB/s OK; mux dropped from 18 → 7-8 → 5-6 MB/s).
+        // Direct pread per call lets the kernel's own readahead policy
+        // run, which interleaves naturally with concurrent NFS writes on
+        // the same TCP connection.
+        //
+        // Buffer fields are retained (currently unused on this path) so
+        // any future per-source policy can be reintroduced without
+        // re-plumbing structure. `refill` / `buffer_covers` are kept too
+        // (still exercised by the tests so the API contract is locked).
+        let offset = lba as u64 * SECTOR_SIZE as u64;
+        self.file
+            .seek(SeekFrom::Start(offset))
+            .map_err(|e| Error::IoError { source: e })?;
+        self.file
+            .read_exact(&mut out[..bytes])
+            .map_err(|e| Error::IoError { source: e })?;
+        self.buf_len_sectors = 0;
+
+        // 0.21.6: periodic page-cache eviction on the read side. Without
+        // this, an 85 GB streaming ISO read pins the entire file in
+        // kernel page cache, which starves concurrent NFS writes (the
+        // MKV output) and collapses mux throughput. Mirrors the
+        // write-side WritebackPipeline's DONTNEED policy.
+        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;
+            platform::drop_window(&self.file, drop_start, drop_len);
+            self.drop_window_start = drop_start + drop_len;
+            self.bytes_read_since_drop = 0;
         }
 
-        // Slice the buffer at the requested LBA's offset within it.
-        let off_sectors = (lba - self.buf_start_lba) as usize;
-        let off_bytes = off_sectors * SECTOR_SIZE;
-        out[..bytes].copy_from_slice(&self.buf[off_bytes..off_bytes + bytes]);
         Ok(bytes)
     }
 }
@@ -288,14 +324,14 @@ mod tests {
 
         let mut src = FileSectorSource::open(&path).unwrap();
 
-        // Prime: read sector 0 to populate buffer #0.
+        // Prime: read sector 0. (0.21.3+: app-level buffer is bypassed,
+        // so we don't assert internal buf state here — just exercise
+        // the read path.)
         let mut got = vec![0u8; SECTOR_SIZE];
         src.read_sectors(0, 1, &mut got, false).unwrap();
-        assert_eq!(src.buf_start_lba, 0);
-        assert!(src.buf_len_sectors > 0);
 
-        // Now read 4 sectors crossing the buffer boundary at
-        // BUF_SECTORS - 2 → BUF_SECTORS + 1. Spans the refill.
+        // Now read 4 sectors crossing what used to be the buffer
+        // boundary. Still a valid SectorSource-contract test.
         let span_lba = BUF_SECTORS - 2;
         let mut buf4 = vec![0u8; SECTOR_SIZE * 4];
         src.read_sectors(span_lba, 4, &mut buf4, false).unwrap();
@@ -323,13 +359,11 @@ mod tests {
         // Forward to the second window.
         src.read_sectors(BUF_SECTORS + 1, 1, &mut got, false)
             .unwrap();
-        let start_after_forward = src.buf_start_lba;
-        assert!(start_after_forward >= BUF_SECTORS);
 
-        // Backward to sector 0. The current buffer doesn't cover it
-        // → refill must happen.
+        // Backward to sector 0. (0.21.3+: app-level buffer is bypassed
+        // so we only assert the byte-level contract, not internal
+        // buffer state.)
         src.read_sectors(0, 1, &mut got, false).unwrap();
-        assert_eq!(src.buf_start_lba, 0);
         assert!(got.iter().all(|b| *b == 0));
     }
 
@@ -348,11 +382,12 @@ mod tests {
         assert_eq!(src.capacity_sectors(), total);
 
         let mut got = vec![0u8; SECTOR_SIZE];
-        // First read triggers refill clamped to `total`.
+        // First read at sector 0.
         src.read_sectors(0, 1, &mut got, false).unwrap();
-        assert_eq!(src.buf_len_sectors, total);
 
-        // Read the very last sector — still inside the buffer.
+        // Read the very last sector. (0.21.3+: app-level buffer is
+        // bypassed; the test still verifies that EOF-region reads
+        // return correct bytes.)
         src.read_sectors(total - 1, 1, &mut got, false).unwrap();
         let expected = ((total - 1) & 0xff) as u8;
         assert!(got.iter().all(|b| *b == expected));

src/io/file_sector_source/other.rs

@@ -5,3 +5,5 @@
 use std::fs::File;
 
 pub(super) fn hint_sequential(_file: &File, _len_bytes: u64) {}
+
+pub(super) fn drop_window(_file: &File, _start: u64, _len: u64) {}

src/io/file_sector_source/windows.rs

@@ -17,3 +17,8 @@ pub(super) fn hint_sequential(_file: &File, _len_bytes: u64) {
         "FileSectorSource hint_sequential: windows stub (TODO: FILE_FLAG_SEQUENTIAL_SCAN at open)"
     );
 }
+
+/// Windows page-cache eviction is not exposed via a posix_fadvise
+/// equivalent. The kernel does its own working-set management. No-op
+/// for now.
+pub(super) fn drop_window(_file: &File, _start: u64, _len: u64) {}

src/io/pipeline.rs

@@ -34,10 +34,11 @@
 //! consumer lag detection). This is critical for diagnosing stalls.
 
 use std::io;
-use std::sync::mpsc::{SyncSender, TrySendError, sync_channel};
 use std::thread::{self, JoinHandle};
 use std::time::{Duration, Instant};
 
+use crossbeam_channel::{Sender, TrySendError, bounded};
+
 use crate::error::Error;
 use crate::halt::Halt;
 
@@ -57,12 +58,21 @@ pub const JOIN_TIMEOUT_SECS: u64 = 600;
 /// responsive across both primitives.
 const POLL_INTERVAL: Duration = Duration::from_millis(250);
 
-/// Polling slice for the halt-aware send loop. Smaller than
-/// [`POLL_INTERVAL`] because send latency tolerance is much lower —
-/// frames must move through the channel at hundreds-of-Hz on the
-/// happy path; 50 ms keeps backpressure-driven wakeups fine-grained
-/// without busy-looping.
-const SEND_POLL_INTERVAL: Duration = Duration::from_millis(50);
+/// Halt-check cadence for the send loop. Producer blocks on
+/// [`crossbeam_channel::Sender::send_timeout`] for this slice — the
+/// kernel wakes it the instant the consumer drains a slot, so on the
+/// happy path there's no throughput cap from this primitive at all
+/// (the cap is whatever the underlying medium can sustain). When the
+/// consumer is genuinely wedged, the timeout fires every 250 ms and
+/// the producer checks the halt token; that's the latency a stop
+/// request will observe.
+///
+/// 0.21.7 replaced an old `std::sync::mpsc::sync_channel` + 50 ms
+/// `thread::sleep` polling loop that capped mux throughput at
+/// ~20 frames/sec ≈ 1 MB/s on saturated channels. See
+/// (internal)/memory/feedback_send_with_halt_poll_throttle.md
+/// for the multi-day diagnostic that surfaced it.
+const SEND_HALT_CHECK_INTERVAL: Duration = Duration::from_millis(250);
 
 /// Check if verbose debug logging is enabled via FREEMKV_DEBUG env var.
 pub fn debug_enabled() -> bool {
@@ -135,7 +145,7 @@ pub trait Sink<I>: Send + 'static {
 /// Bounded producer/consumer pipeline. Holds the producer-side
 /// channel and the consumer thread's join handle.
 pub struct Pipeline<I: Send + 'static, R: Send + 'static> {
-    tx: SyncSender<I>,
+    tx: Sender<I>,
     handle: JoinHandle<Result<R, Error>>,
 }
 
@@ -170,7 +180,7 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
         depth: usize,
         sink: S,
     ) -> Result<Self, Error> {
-        let (tx, rx) = sync_channel::<I>(depth);
+        let (tx, rx) = bounded::<I>(depth);
         let handle = thread::Builder::new()
             .name(name.into())
             .spawn(move || -> Result<R, Error> {
@@ -281,14 +291,20 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
     /// `Err`. Useful for best-effort signalling (e.g. sweep's
     /// throttled `StatsRequest`) where dropping the message is
     /// preferable to blocking the producer.
-    pub fn try_send(&self, item: I) -> Result<(), std::sync::mpsc::TrySendError<I>> {
+    pub fn try_send(&self, item: I) -> Result<(), TrySendError<I>> {
         self.tx.try_send(item)
     }
 
     /// Halt-aware bounded variant of [`Pipeline::send`].
     ///
-    /// Polls `try_send` on a 50 ms slice. Between slices, checks
-    /// (1) the [`Halt`] token and (2) the per-call `deadline`. Returns:
+    /// Uses [`crossbeam_channel::Sender::send_timeout`] so the producer
+    /// thread BLOCKS on consumer drain (kernel-wakeup) rather than
+    /// polling. The timeout slice is just the halt-observation cadence
+    /// ([`SEND_HALT_CHECK_INTERVAL`]) — on the happy path the producer
+    /// wakes the instant the consumer drains a slot, so there is no
+    /// throughput cap from this primitive at any medium speed.
+    ///
+    /// Returns:
     ///
     /// - `Ok(())` once the item lands in the channel.
     /// - `Err(item)` if the consumer disconnected, the halt fired, or
@@ -302,36 +318,45 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
     /// Unlike [`Pipeline::send`], this never blocks the producer
     /// thread inside an unkillable `mpsc::send` — if the consumer is
     /// wedged inside an unkillable syscall, the producer can still
-    /// observe `/api/stop` and unwind.
+    /// observe `/api/stop` and unwind within
+    /// [`SEND_HALT_CHECK_INTERVAL`].
     pub fn send_with_halt(&self, item: I, halt: &Halt, deadline: Duration) -> Result<(), I> {
+        use crossbeam_channel::SendTimeoutError;
         let end = Instant::now() + deadline;
         let mut pending = item;
         loop {
-            match self.tx.try_send(pending) {
-                Ok(()) => return Ok(()),
-                Err(TrySendError::Full(returned)) => {
-                    pending = returned;
-                    if halt.is_cancelled() {
-                        if debug_enabled() {
-                            tracing::debug!(
-                                "Pipeline send_with_halt: halt observed, returning item={}",
-                                std::any::type_name::<I>()
-                            );
-                        }
-                        return Err(pending);
-                    }
-                    if Instant::now() >= end {
-                        if debug_enabled() {
-                            tracing::debug!(
-                                "Pipeline send_with_halt: deadline elapsed, returning item={}",
-                                std::any::type_name::<I>()
-                            );
-                        }
-                        return Err(pending);
-                    }
-                    thread::sleep(SEND_POLL_INTERVAL);
+            // Pre-check the cheap exit conditions before parking.
+            if halt.is_cancelled() {
+                if debug_enabled() {
+                    tracing::debug!(
+                        "Pipeline send_with_halt: halt observed, returning item={}",
+                        std::any::type_name::<I>()
+                    );
                 }
-                Err(TrySendError::Disconnected(returned)) => {
+                return Err(pending);
+            }
+            let now = Instant::now();
+            if now >= end {
+                if debug_enabled() {
+                    tracing::debug!(
+                        "Pipeline send_with_halt: deadline elapsed, returning item={}",
+                        std::any::type_name::<I>()
+                    );
+                }
+                return Err(pending);
+            }
+            // Wait for space-available or halt-check tick, whichever
+            // is sooner. Crossbeam's send_timeout is kernel-wakeup
+            // based: the consumer's recv on a saturated channel
+            // signals this thread the moment a slot opens up.
+            let slice = SEND_HALT_CHECK_INTERVAL.min(end.saturating_duration_since(now));
+            match self.tx.send_timeout(pending, slice) {
+                Ok(()) => return Ok(()),
+                Err(SendTimeoutError::Timeout(returned)) => {
+                    pending = returned;
+                    // loop: re-check halt + deadline, then park again
+                }
+                Err(SendTimeoutError::Disconnected(returned)) => {
                     if debug_enabled() {
                         tracing::debug!(
                             "Pipeline send_with_halt: consumer disconnected, item={}",

src/io/sink/mod.rs

@@ -28,8 +28,10 @@ use std::io::{Seek, Write};
 
 mod local_file;
 mod preallocate;
+mod socket;
 
 pub use local_file::LocalFileSink;
+pub use socket::{SocketSink, UdpSocketSink};
 
 /// Sequential-only write destination. Sockets, pipes, append-only
 /// stores. No seek. Implementations own their write buffering — the
@@ -83,10 +85,10 @@ pub fn open_for_mkv(
     dest: &std::path::Path,
     size_hint: Option<u64>,
 ) -> std::io::Result<Box<dyn RandomAccessSink>> {
-    #[cfg(target_os = "linux")]
-    use crate::platform::fs_type::{FsType, detect};
     #[cfg(not(target_os = "linux"))]
     use crate::platform::fs_type::detect;
+    #[cfg(target_os = "linux")]
+    use crate::platform::fs_type::{FsType, detect};
 
     #[cfg(target_os = "linux")]
     {

src/io/sink/socket.rs

@@ -0,0 +1,278 @@
+//! TCP / UDP socket sinks (sequential-only).
+//!
+//! [`SocketSink`] wraps a `TcpStream` in a 1 MiB `BufWriter`. Constructor
+//! tunes `SO_SNDBUF` to a caller hint when provided. `finish()` flushes
+//! the buffer then `shutdown(Write)`s the socket so the peer sees clean
+//! end-of-stream.
+//!
+//! [`UdpSocketSink`] wraps a connected `UdpSocket`. Each `write` call
+//! emits exactly one datagram — the caller is responsible for packetizing
+//! to a reasonable MTU (188 × 7 = 1316 bytes for MPEG-TS-over-UDP is the
+//! conventional choice). `finish()` is a no-op; UDP has no end-of-stream
+//! marker.
+//!
+//! Both types satisfy [`SequentialSink`] via the blanket impl in
+//! `super::mod`. Neither implements `Seek`, so neither satisfies
+//! [`RandomAccessSink`] — using one with `MkvMux` is a compile error,
+//! which is the design intent.
+//!
+//! [`SequentialSink`]: super::SequentialSink
+//! [`RandomAccessSink`]: super::RandomAccessSink
+
+use std::io::{self, BufWriter, Write};
+use std::net::{Shutdown, TcpStream, ToSocketAddrs, UdpSocket};
+
+/// `BufWriter` capacity for [`SocketSink`]. 1 MiB matches the typical
+/// kernel send-buffer ceiling and keeps small-write amplification from
+/// containers (TS = 188-byte packets, fMP4 fragment headers = ~100 bytes)
+/// from translating into syscall storms.
+const TCP_BUF_CAPACITY: usize = 1024 * 1024;
+
+/// Sequential-only sink over a TCP connection.
+///
+/// Wraps a `BufWriter<TcpStream>`; the inner `TcpStream` is kept as a
+/// clone so [`finish`](Self::finish) can call `shutdown(Write)` after
+/// flushing the buffer (the buffered writer doesn't expose the socket
+/// directly).
+pub struct SocketSink {
+    /// Buffered write half. All payload bytes go through this.
+    buf: BufWriter<TcpStream>,
+    /// Shutdown handle — clone of the socket inside `buf`. Used only by
+    /// `finish()` for `shutdown(Write)`; never read or written through.
+    shutdown_handle: TcpStream,
+}
+
+impl SocketSink {
+    /// Open a TCP connection to `addr` and wrap it for sequential
+    /// writing. `sndbuf_bytes`, when present, is forwarded to
+    /// `setsockopt(SO_SNDBUF)` as a kernel hint — the OS may clamp it.
+    ///
+    /// `addr` accepts anything `ToSocketAddrs` does: `"10.0.0.1:1234"`,
+    /// `("host", 1234)`, a `SocketAddr`, etc.
+    pub fn connect<A: ToSocketAddrs>(addr: A, sndbuf_bytes: Option<usize>) -> io::Result<Self> {
+        let stream = TcpStream::connect(addr)?;
+        // `set_nodelay(true)` keeps small writes (TS packet trains, fMP4
+        // moof headers) from sitting in Nagle's algorithm until the buffer
+        // fills. The BufWriter already absorbs syscall overhead; Nagle
+        // would just add latency without coalescing more.
+        stream.set_nodelay(true)?;
+        if let Some(n) = sndbuf_bytes {
+            set_send_buffer(&stream, n)?;
+        }
+        let shutdown_handle = stream.try_clone()?;
+        Ok(Self {
+            buf: BufWriter::with_capacity(TCP_BUF_CAPACITY, stream),
+            shutdown_handle,
+        })
+    }
+}
+
+impl Write for SocketSink {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.buf.write(buf)
+    }
+    fn flush(&mut self) -> io::Result<()> {
+        self.buf.flush()
+    }
+}
+
+impl SocketSink {
+    /// Drain the BufWriter and `shutdown(Write)` the underlying socket
+    /// so the peer sees a clean EOF.
+    ///
+    /// Note: [`SequentialSink::finish`](super::SequentialSink::finish)'s
+    /// blanket-impl default is a no-op. Trait-object call sites that
+    /// need socket shutdown should call this inherent method directly
+    /// before dropping the sink, or hold the concrete `SocketSink` type
+    /// (typical pattern: each muxer's `finish()` calls the appropriate
+    /// inherent close method on its captured concrete sink).
+    pub fn finish(&mut self) -> io::Result<()> {
+        self.buf.flush()?;
+        // `shutdown(Write)` signals clean EOF to the peer. Errors here
+        // are non-fatal — the connection may have already been torn down
+        // by the peer — but we surface them so callers can log.
+        self.shutdown_handle.shutdown(Shutdown::Write)
+    }
+}
+
+/// Sequential-only sink over a connected UDP socket.
+///
+/// Each [`write`](Write::write) call sends exactly one datagram. The
+/// caller is responsible for splitting payload at packet boundaries —
+/// for MPEG-TS this means 7 × 188 = 1316 bytes per datagram, the
+/// industry standard for MPEG-TS-over-UDP. No buffering happens here;
+/// adding it would silently merge datagrams.
+///
+/// `finish()` is a no-op: UDP has no end-of-stream marker. Closing the
+/// socket happens on drop.
+pub struct UdpSocketSink {
+    socket: UdpSocket,
+}
+
+impl UdpSocketSink {
+    /// Bind a local UDP socket to an ephemeral port and `connect` it to
+    /// `peer`. `connect` doesn't open a connection — it just fixes the
+    /// peer address so subsequent `send` calls don't need to repeat it,
+    /// and so receive-side filtering rejects packets from other sources.
+    ///
+    /// `sndbuf_bytes`, when present, is a hint to `SO_SNDBUF`.
+    pub fn connect<A: ToSocketAddrs>(peer: A, sndbuf_bytes: Option<usize>) -> io::Result<Self> {
+        // Bind to all-zeros / any port. The kernel picks an ephemeral
+        // source port and the source IP at first send.
+        let socket = UdpSocket::bind("0.0.0.0:0")?;
+        socket.connect(peer)?;
+        if let Some(n) = sndbuf_bytes {
+            set_udp_send_buffer(&socket, n)?;
+        }
+        Ok(Self { socket })
+    }
+}
+
+impl Write for UdpSocketSink {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        // `send` writes the entire datagram or fails — no partial sends
+        // for UDP. Match `Write::write`'s contract by reporting bytes
+        // accepted.
+        self.socket.send(buf)
+    }
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+impl UdpSocketSink {
+    /// No-op — UDP has no end-of-stream marker. Provided for parity
+    /// with [`SocketSink::finish`] so call sites can treat them uniformly.
+    pub fn finish(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+// ── Platform `SO_SNDBUF` tuning ────────────────────────────────────────────
+//
+// std's `TcpStream` / `UdpSocket` don't expose `SO_SNDBUF`. We drop to
+// libc on Linux + macOS (the libc-dep targets in Cargo.toml). On other
+// targets the hint is silently ignored — the socket still works, the
+// kernel just picks its own send-buffer size.
+
+#[cfg(any(target_os = "linux", target_os = "macos"))]
+fn set_send_buffer(stream: &TcpStream, bytes: usize) -> io::Result<()> {
+    use std::os::unix::io::AsRawFd;
+    setsockopt_sndbuf(stream.as_raw_fd(), bytes)
+}
+
+#[cfg(any(target_os = "linux", target_os = "macos"))]
+fn set_udp_send_buffer(socket: &UdpSocket, bytes: usize) -> io::Result<()> {
+    use std::os::unix::io::AsRawFd;
+    setsockopt_sndbuf(socket.as_raw_fd(), bytes)
+}
+
+#[cfg(not(any(target_os = "linux", target_os = "macos")))]
+fn set_send_buffer(_stream: &TcpStream, _bytes: usize) -> io::Result<()> {
+    // Non-Linux-non-macOS targets aren't in Cargo.toml's libc dep list;
+    // silently ignore the hint rather than failing the connect. Callers
+    // can detect via the lack of an explicit "sndbuf applied" signal
+    // (not provided, intentionally — this is a hint, not a guarantee).
+    Ok(())
+}
+
+#[cfg(not(any(target_os = "linux", target_os = "macos")))]
+fn set_udp_send_buffer(_socket: &UdpSocket, _bytes: usize) -> io::Result<()> {
+    Ok(())
+}
+
+#[cfg(any(target_os = "linux", target_os = "macos"))]
+fn setsockopt_sndbuf(fd: std::os::unix::io::RawFd, bytes: usize) -> io::Result<()> {
+    // Clamp into c_int range; SO_SNDBUF takes an `int` argument.
+    let want: libc::c_int = bytes.try_into().unwrap_or(libc::c_int::MAX);
+    let ret = unsafe {
+        libc::setsockopt(
+            fd,
+            libc::SOL_SOCKET,
+            libc::SO_SNDBUF,
+            &want as *const _ as *const libc::c_void,
+            std::mem::size_of::<libc::c_int>() as libc::socklen_t,
+        )
+    };
+    if ret != 0 {
+        return Err(io::Error::last_os_error());
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Read;
+    use std::net::{TcpListener, UdpSocket};
+    use std::thread;
+
+    /// Bind a listener, accept on a thread, return (listener_addr,
+    /// accepted-bytes future via JoinHandle).
+    #[test]
+    fn socket_sink_round_trips_bytes() {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let accept = thread::spawn(move || {
+            let (mut sock, _) = listener.accept().unwrap();
+            let mut buf = Vec::new();
+            sock.read_to_end(&mut buf).unwrap();
+            buf
+        });
+
+        let mut sink = SocketSink::connect(addr, Some(256 * 1024)).unwrap();
+        // Write enough to overflow the BufWriter at least once, then a
+        // tail that lives in the buffer until `finish` flushes.
+        let big: Vec<u8> = (0..(2 * TCP_BUF_CAPACITY))
+            .map(|i| (i & 0xff) as u8)
+            .collect();
+        sink.write_all(&big).unwrap();
+        sink.write_all(b"tail\n").unwrap();
+        sink.finish().unwrap();
+        drop(sink);
+
+        let received = accept.join().unwrap();
+        assert_eq!(received.len(), big.len() + 5);
+        assert_eq!(&received[..big.len()], &big[..]);
+        assert_eq!(&received[big.len()..], b"tail\n");
+    }
+
+    #[test]
+    fn socket_sink_is_sequential_only() {
+        // Compile-time assertion via dyn — if this ever started
+        // satisfying `RandomAccessSink`, the trait split would be broken.
+        fn _assert_seq(_: &mut dyn super::super::SequentialSink) {}
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let _accept = thread::spawn(move || {
+            let _ = listener.accept();
+        });
+        let mut sink = SocketSink::connect(addr, None).unwrap();
+        _assert_seq(&mut sink);
+        // The negative is harder to assert directly (no `is_not<T>`),
+        // but `SocketSink` does not impl `Seek`, so it can't unify with
+        // `RandomAccessSink`'s super-bound. The Phase 2 blanket impl
+        // `impl<T: SequentialSink + Seek> RandomAccessSink for T {}` thus
+        // excludes it by construction.
+    }
+
+    #[test]
+    fn udp_socket_sink_delivers_datagrams() {
+        let receiver = UdpSocket::bind("127.0.0.1:0").unwrap();
+        receiver
+            .set_read_timeout(Some(std::time::Duration::from_secs(2)))
+            .unwrap();
+        let addr = receiver.local_addr().unwrap();
+        let mut sink = UdpSocketSink::connect(addr, Some(128 * 1024)).unwrap();
+
+        sink.write_all(&[1, 2, 3, 4, 5]).unwrap();
+        sink.write_all(&[9, 9, 9]).unwrap();
+        sink.finish().unwrap();
+
+        let mut buf = [0u8; 64];
+        let n1 = receiver.recv(&mut buf).unwrap();
+        assert_eq!(&buf[..n1], &[1, 2, 3, 4, 5]);
+        let n2 = receiver.recv(&mut buf).unwrap();
+        assert_eq!(&buf[..n2], &[9, 9, 9]);
+    }
+}

src/io/writeback/linux.rs

@@ -321,8 +321,12 @@ fn detect_nfs(fd: RawFd) -> bool {
     }
     // `f_type` is signed (`__fsword_t`) on glibc and unsigned
     // (`c_ulong`) on musl. Cast both sides to i64 for a portable
-    // comparison.
+    // comparison. On glibc x86_64 both already are i64 — clippy flags
+    // the cast as unnecessary on that target only, but we need it for
+    // musl, so silence the lint.
+    #[allow(clippy::unnecessary_cast)]
     let f_type = buf.f_type as i64;
+    #[allow(clippy::unnecessary_cast)]
     let nfs_magic = libc::NFS_SUPER_MAGIC as i64;
     f_type == nfs_magic
 }

src/io/writeback_file/mod.rs

@@ -30,6 +30,23 @@
 //! `FlushFileBuffers`) — live in per-OS sibling modules. The dispatch
 //! 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
+//!
+//! Writes are direct passthrough to the underlying `File` (no writer
+//! thread, no ring, no batching). Empirically the Phase-2.5
+//! writer-thread architecture introduced a ~60% mux throughput
+//! regression on NFS bidirectional workloads; reverting the write path
+//! to direct passthrough restores the 0.20.7 baseline. The writeback
+//! pipeline still runs (it's called inline from `write` / `write_all` /
+//! `seek`) so the bounded-cache invariant on Linux is preserved.
+//!
+//! ## Halt-safety
+//!
+//! `sync_all` runs the per-OS durable-flush primitive, which on
+//! Linux/macOS is wrapped in [`crate::io::bounded::bounded_syscall`]
+//! with a 60 s deadline. A wedged NFS server cannot trap the muxer
+//! indefinitely on the final fsync.
 
 #[cfg(target_os = "linux")]
 mod linux;
@@ -55,6 +72,9 @@ use std::path::Path;
 
 use super::writeback::WritebackPipeline;
 
+/// Granularity at which the Linux writeback pipeline issues
+/// `sync_file_range` / `posix_fadvise(DONTNEED)` pairs. 32 MiB is the
+/// historical default — bounded-cache pressure stays at ~2 × this size.
 const WRITEBACK_CHUNK_BYTES: u64 = 32 * 1024 * 1024;
 
 pub(crate) struct WritebackFile {
@@ -122,14 +142,12 @@ 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 on
-    /// platforms that have a usable bounded primitive (Linux + macOS).
-    /// 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.
+    /// The final durable flush is wrapped in
+    /// [`crate::io::bounded::bounded_syscall`] (per the per-OS module)
+    /// with a 60 s deadline on Linux/macOS — a wedged NFS server cannot
+    /// trap the calling thread indefinitely. 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();
         platform::durable_sync(&self.file)
@@ -164,13 +182,13 @@ 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.
+            // Diagnostic for the NFS 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);
@@ -201,3 +219,85 @@ impl Drop for WritebackFile {
         self.pipeline.finalize();
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Read;
+
+    fn read_back(path: &Path) -> Vec<u8> {
+        let mut f = File::open(path).unwrap();
+        let mut v = Vec::new();
+        f.read_to_end(&mut v).unwrap();
+        v
+    }
+
+    #[test]
+    fn write_then_drop_persists_bytes() {
+        let dir = tempfile::tempdir().unwrap();
+        let p = dir.path().join("a.bin");
+        {
+            let mut w = WritebackFile::create(&p).unwrap();
+            w.write_all(b"hello world").unwrap();
+            // Drop drains the pipeline tail.
+        }
+        assert_eq!(read_back(&p), b"hello world");
+    }
+
+    #[test]
+    fn sync_all_drains_and_flushes() {
+        let dir = tempfile::tempdir().unwrap();
+        let p = dir.path().join("b.bin");
+        let mut w = WritebackFile::create(&p).unwrap();
+        for _ in 0..32 {
+            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
+        // run.
+        w.sync_all().unwrap();
+        let bytes = read_back(&p);
+        assert_eq!(bytes.len(), 32 * 1024);
+        assert!(bytes.iter().all(|&b| b == 0x5a));
+        drop(w);
+    }
+
+    #[test]
+    fn seek_then_patch_roundtrip() {
+        // Write A; seek back; patch 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();
+        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();
+        w.sync_all().unwrap();
+        drop(w);
+        let bytes = read_back(&p);
+        assert_eq!(bytes.len(), 4096);
+        assert_eq!(&bytes[1000..1008], b"PATCHED!");
+        // Bytes outside the patch are still 'A'.
+        assert_eq!(bytes[999], b'A');
+        assert_eq!(bytes[1008], b'A');
+    }
+
+    #[test]
+    fn flush_is_observed_in_order() {
+        // `Write::flush` should not panic or reorder; verify the bytes
+        // land in order through interleaved flushes.
+        let dir = tempfile::tempdir().unwrap();
+        let p = dir.path().join("f.bin");
+        let mut w = WritebackFile::create(&p).unwrap();
+        w.write_all(b"one").unwrap();
+        w.flush().unwrap();
+        w.write_all(b"two").unwrap();
+        w.flush().unwrap();
+        w.write_all(b"three").unwrap();
+        w.sync_all().unwrap();
+        drop(w);
+        assert_eq!(read_back(&p), b"onetwothree");
+    }
+}

src/mux/fmp4/mod.rs

@@ -0,0 +1,404 @@
+//! Fragmented MP4 muxer — **stub** for Phase 3.
+//!
+//! Goal: ISO/IEC 14496-12 fragmented MP4 (`ftyp` + `moov` init segment,
+//! then a sequence of `moof+mdat` media fragments) targeting a
+//! [`SequentialSink`](crate::io::sink::SequentialSink). DASH-friendly,
+//! no Cues backpatch.
+//!
+//! Status (v0.21.0 Phase 3): **STUB**. We ship the init segment
+//! (`ftyp` + a minimal HEVC `moov` skeleton with one video track) so
+//! the muxer's shape and call site are validated, but media fragments
+//! are NOT yet emitted — calls to [`Fmp4Mux::write_video`] currently
+//! accumulate frames into an internal buffer and discard them on
+//! [`Fmp4Mux::finish`].
+//!
+//! ## What's TODO (tracked in Phase 4 / v0.22.0 scope)
+//!
+//! - `moof` box: `mfhd` (sequence_number) + `traf` (`tfhd` + `tfdt`
+//!   + `trun` with sample sizes, durations, flags, composition offsets).
+//! - `mdat` box: concatenated sample data.
+//! - Fragment cadence: one fragment per GOP or every N seconds,
+//!   whichever comes first.
+//! - HEVC `hvcC` box inside `moov.trak.mdia.minf.stbl.stsd` so the
+//!   init segment is self-describing.
+//! - Sample-flags computation (sync vs. delta, depends_on, etc.).
+//! - Edit lists / fragment_duration for accurate seeking.
+//!
+//! Reference: ISO/IEC 14496-12 §8 (Movie Fragments).
+
+use std::io::{self, Write};
+
+// Box type literals — four-character codes per ISO/IEC 14496-12 §4.2.
+
+const FTYP: [u8; 4] = *b"ftyp";
+const MOOV: [u8; 4] = *b"moov";
+const MVHD: [u8; 4] = *b"mvhd";
+const TRAK: [u8; 4] = *b"trak";
+const TKHD: [u8; 4] = *b"tkhd";
+const MDIA: [u8; 4] = *b"mdia";
+const MDHD: [u8; 4] = *b"mdhd";
+const HDLR: [u8; 4] = *b"hdlr";
+const MINF: [u8; 4] = *b"minf";
+const VMHD: [u8; 4] = *b"vmhd";
+const DINF: [u8; 4] = *b"dinf";
+const DREF: [u8; 4] = *b"dref";
+const URL_: [u8; 4] = *b"url ";
+const STBL: [u8; 4] = *b"stbl";
+const STSD: [u8; 4] = *b"stsd";
+const STTS: [u8; 4] = *b"stts";
+const STSC: [u8; 4] = *b"stsc";
+const STSZ: [u8; 4] = *b"stsz";
+const STCO: [u8; 4] = *b"stco";
+const MVEX: [u8; 4] = *b"mvex";
+const TREX: [u8; 4] = *b"trex";
+
+/// Default movie timescale — 90 kHz lines up with MPEG-TS PTS and the
+/// HEVC SPS `vui_time_scale` for film content, simplifying the math
+/// when fragment emission lands.
+const MOVIE_TIMESCALE: u32 = 90_000;
+/// Video track ID. fMP4 init segments conventionally use track_ID=1
+/// for the primary video track; a single-track DASH representation has
+/// no reason to deviate.
+const VIDEO_TRACK_ID: u32 = 1;
+
+/// Fragmented MP4 muxer — stub.
+///
+/// See the module-level doc comment for what is and isn't shipped in
+/// this stub.
+pub struct Fmp4Mux<W: Write> {
+    writer: W,
+    header_written: bool,
+    /// Pending frames — held for the future fragment-emit path. The
+    /// stub drops these on `finish` but keeping them around lets the
+    /// post-stub work re-attach without changing the public API.
+    pending: Vec<PendingSample>,
+    /// hvcC bytes, if provided. Embedded in the `moov.…stsd.hvc1.hvcC`
+    /// box once that path lands.
+    #[allow(dead_code)]
+    codec_private: Option<Vec<u8>>,
+}
+
+struct PendingSample {
+    #[allow(dead_code)]
+    pts_ns: i64,
+    #[allow(dead_code)]
+    keyframe: bool,
+    #[allow(dead_code)]
+    data: Vec<u8>,
+}
+
+impl<W: Write> Fmp4Mux<W> {
+    pub fn new(writer: W) -> Self {
+        Self {
+            writer,
+            header_written: false,
+            pending: Vec::new(),
+            codec_private: None,
+        }
+    }
+
+    /// Provide the `HEVCDecoderConfigurationRecord` for the video track.
+    /// The stub stores it but doesn't yet embed it in `moov` — that's
+    /// part of the post-stub work.
+    pub fn set_video_codec_private(&mut self, hvcc: Vec<u8>) {
+        self.codec_private = Some(hvcc);
+    }
+
+    /// Write one video PES frame.
+    ///
+    /// **Stub behaviour:** the first call emits the init segment
+    /// (`ftyp` + `moov`) so any consumer that just wants the shape can
+    /// receive it. Subsequent calls accumulate frames in memory for
+    /// the future fragmenting path; **no media bytes are written yet**.
+    pub fn write_video(&mut self, pts_ns: i64, keyframe: bool, data: &[u8]) -> io::Result<()> {
+        if !self.header_written {
+            self.write_init_segment()?;
+            self.header_written = true;
+        }
+        // TODO(0.22.0): emit one `moof+mdat` per GOP. For now stash the
+        // frame so the future patch can hot-wire emission without API
+        // churn.
+        self.pending.push(PendingSample {
+            pts_ns,
+            keyframe,
+            data: data.to_vec(),
+        });
+        Ok(())
+    }
+
+    /// Flush. The stub additionally drops accumulated `pending` frames.
+    pub fn finish(&mut self) -> io::Result<()> {
+        // TODO(0.22.0): emit final fragment from pending; today the
+        // stub just clears the buffer to release memory.
+        self.pending.clear();
+        self.writer.flush()
+    }
+
+    fn write_init_segment(&mut self) -> io::Result<()> {
+        let ftyp = build_ftyp();
+        let moov = build_moov();
+        self.writer.write_all(&ftyp)?;
+        self.writer.write_all(&moov)?;
+        Ok(())
+    }
+}
+
+/// Build the `ftyp` box. `major_brand = "iso6"`, `minor_version = 1`,
+/// compatible brands `iso6 dash msdh hvc1` — the same conservative set
+/// shaka-packager uses for HEVC-in-fMP4 outputs.
+fn build_ftyp() -> Vec<u8> {
+    let mut body = Vec::new();
+    body.extend_from_slice(b"iso6");
+    body.extend_from_slice(&1u32.to_be_bytes());
+    body.extend_from_slice(b"iso6");
+    body.extend_from_slice(b"dash");
+    body.extend_from_slice(b"msdh");
+    body.extend_from_slice(b"hvc1");
+    wrap_box(&FTYP, &body)
+}
+
+/// Build the `moov` box — minimal skeleton. Single video trak, no
+/// hvcC inside stsd yet (TODO: full hvc1 sample entry).
+fn build_moov() -> Vec<u8> {
+    let mvhd = build_mvhd();
+    let trak = build_video_trak();
+    let mvex = build_mvex();
+    let mut body = Vec::new();
+    body.extend_from_slice(&mvhd);
+    body.extend_from_slice(&trak);
+    body.extend_from_slice(&mvex);
+    wrap_box(&MOOV, &body)
+}
+
+fn build_mvhd() -> Vec<u8> {
+    // Version 0, 100 bytes total body. Fields per ISO/IEC 14496-12 §8.2.2.
+    let mut body = Vec::new();
+    body.extend_from_slice(&[0, 0, 0, 0]); // version + flags
+    body.extend_from_slice(&0u32.to_be_bytes()); // creation_time
+    body.extend_from_slice(&0u32.to_be_bytes()); // modification_time
+    body.extend_from_slice(&MOVIE_TIMESCALE.to_be_bytes());
+    body.extend_from_slice(&0u32.to_be_bytes()); // duration = 0 (fragmented)
+    body.extend_from_slice(&0x0001_0000u32.to_be_bytes()); // rate 1.0
+    body.extend_from_slice(&0x0100u16.to_be_bytes()); // volume 1.0
+    body.extend_from_slice(&[0u8; 2]); // reserved
+    body.extend_from_slice(&[0u8; 8]); // reserved
+    // 3x3 identity transformation matrix in 16.16 fixed point.
+    for v in [0x1_0000u32, 0, 0, 0, 0x1_0000, 0, 0, 0, 0x4000_0000] {
+        body.extend_from_slice(&v.to_be_bytes());
+    }
+    body.extend_from_slice(&[0u8; 24]); // pre_defined[6]
+    body.extend_from_slice(&2u32.to_be_bytes()); // next_track_ID (1 reserved for video)
+    wrap_box(&MVHD, &body)
+}
+
+fn build_video_trak() -> Vec<u8> {
+    let tkhd = build_tkhd();
+    let mdia = build_mdia();
+    let mut body = Vec::new();
+    body.extend_from_slice(&tkhd);
+    body.extend_from_slice(&mdia);
+    wrap_box(&TRAK, &body)
+}
+
+fn build_tkhd() -> Vec<u8> {
+    let mut body = Vec::new();
+    // version=0 | flags=0x000007 (track_enabled | in_movie | in_preview)
+    body.extend_from_slice(&[0, 0, 0, 7]);
+    body.extend_from_slice(&0u32.to_be_bytes()); // creation_time
+    body.extend_from_slice(&0u32.to_be_bytes()); // modification_time
+    body.extend_from_slice(&VIDEO_TRACK_ID.to_be_bytes());
+    body.extend_from_slice(&[0u8; 4]); // reserved
+    body.extend_from_slice(&0u32.to_be_bytes()); // duration
+    body.extend_from_slice(&[0u8; 8]); // reserved
+    body.extend_from_slice(&0u16.to_be_bytes()); // layer
+    body.extend_from_slice(&0u16.to_be_bytes()); // alternate_group
+    body.extend_from_slice(&0u16.to_be_bytes()); // volume (video=0)
+    body.extend_from_slice(&[0u8; 2]); // reserved
+    // 3x3 identity matrix.
+    for v in [0x1_0000u32, 0, 0, 0, 0x1_0000, 0, 0, 0, 0x4000_0000] {
+        body.extend_from_slice(&v.to_be_bytes());
+    }
+    // width / height in 16.16 fixed point — placeholder 1920x1080.
+    body.extend_from_slice(&(1920u32 << 16).to_be_bytes());
+    body.extend_from_slice(&(1080u32 << 16).to_be_bytes());
+    wrap_box(&TKHD, &body)
+}
+
+fn build_mdia() -> Vec<u8> {
+    let mdhd = build_mdhd();
+    let hdlr = build_hdlr_vide();
+    let minf = build_minf();
+    let mut body = Vec::new();
+    body.extend_from_slice(&mdhd);
+    body.extend_from_slice(&hdlr);
+    body.extend_from_slice(&minf);
+    wrap_box(&MDIA, &body)
+}
+
+fn build_mdhd() -> Vec<u8> {
+    let mut body = Vec::new();
+    body.extend_from_slice(&[0, 0, 0, 0]); // version + flags
+    body.extend_from_slice(&0u32.to_be_bytes()); // creation_time
+    body.extend_from_slice(&0u32.to_be_bytes()); // modification_time
+    body.extend_from_slice(&MOVIE_TIMESCALE.to_be_bytes());
+    body.extend_from_slice(&0u32.to_be_bytes()); // duration
+    // language: 'und' in 5-bit-per-char ISO 639-2 packed (bit 15 = 0).
+    body.extend_from_slice(&[0x55, 0xC4]);
+    body.extend_from_slice(&0u16.to_be_bytes()); // pre_defined
+    wrap_box(&MDHD, &body)
+}
+
+fn build_hdlr_vide() -> Vec<u8> {
+    let mut body = Vec::new();
+    body.extend_from_slice(&[0, 0, 0, 0]); // version + flags
+    body.extend_from_slice(&0u32.to_be_bytes()); // pre_defined
+    body.extend_from_slice(b"vide");
+    body.extend_from_slice(&[0u8; 12]); // reserved
+    body.extend_from_slice(b"VideoHandler\0");
+    wrap_box(&HDLR, &body)
+}
+
+fn build_minf() -> Vec<u8> {
+    let vmhd = build_vmhd();
+    let dinf = build_dinf();
+    let stbl = build_stbl();
+    let mut body = Vec::new();
+    body.extend_from_slice(&vmhd);
+    body.extend_from_slice(&dinf);
+    body.extend_from_slice(&stbl);
+    wrap_box(&MINF, &body)
+}
+
+fn build_vmhd() -> Vec<u8> {
+    let mut body = Vec::new();
+    body.extend_from_slice(&[0, 0, 0, 1]); // version + flags=1
+    body.extend_from_slice(&0u16.to_be_bytes()); // graphicsmode
+    body.extend_from_slice(&[0u8; 6]); // opcolor
+    wrap_box(&VMHD, &body)
+}
+
+fn build_dinf() -> Vec<u8> {
+    let mut dref_body = Vec::new();
+    dref_body.extend_from_slice(&[0, 0, 0, 0]);
+    dref_body.extend_from_slice(&1u32.to_be_bytes()); // entry_count
+    // url with flags=1 (self-contained) and zero name.
+    let url_body = [0u8, 0, 0, 1];
+    dref_body.extend_from_slice(&wrap_box(&URL_, &url_body));
+    let dref = wrap_box(&DREF, &dref_body);
+    wrap_box(&DINF, &dref)
+}
+
+fn build_stbl() -> Vec<u8> {
+    // Stub stsd: empty sample description (zero entries). Replace with
+    // hvc1+hvcC once the fragmenting path lands so the init segment is
+    // actually decodable.
+    let mut stsd_body = Vec::new();
+    stsd_body.extend_from_slice(&[0, 0, 0, 0]);
+    stsd_body.extend_from_slice(&0u32.to_be_bytes()); // entry_count
+    let stsd = wrap_box(&STSD, &stsd_body);
+
+    // Empty stts/stsc/stsz/stco — fragmented init has no samples here.
+    let stts = wrap_box(&STTS, &[0, 0, 0, 0, 0, 0, 0, 0]); // version+flags, count=0
+    let stsc = wrap_box(&STSC, &[0, 0, 0, 0, 0, 0, 0, 0]);
+    let stsz = wrap_box(
+        &STSZ,
+        &[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], // version+flags, sample_size=0, count=0
+    );
+    let stco = wrap_box(&STCO, &[0, 0, 0, 0, 0, 0, 0, 0]);
+
+    let mut body = Vec::new();
+    body.extend_from_slice(&stsd);
+    body.extend_from_slice(&stts);
+    body.extend_from_slice(&stsc);
+    body.extend_from_slice(&stsz);
+    body.extend_from_slice(&stco);
+    wrap_box(&STBL, &body)
+}
+
+fn build_mvex() -> Vec<u8> {
+    // trex: track_ID=1, default_sample_description_index=1, others=0.
+    let mut trex_body = Vec::new();
+    trex_body.extend_from_slice(&[0, 0, 0, 0]); // version + flags
+    trex_body.extend_from_slice(&VIDEO_TRACK_ID.to_be_bytes());
+    trex_body.extend_from_slice(&1u32.to_be_bytes()); // default_sample_description_index
+    trex_body.extend_from_slice(&0u32.to_be_bytes()); // default_sample_duration
+    trex_body.extend_from_slice(&0u32.to_be_bytes()); // default_sample_size
+    trex_body.extend_from_slice(&0u32.to_be_bytes()); // default_sample_flags
+    let trex = wrap_box(&TREX, &trex_body);
+    wrap_box(&MVEX, &trex)
+}
+
+/// Wrap a box body in `[size:u32-BE][type:4]`. Suitable for any body
+/// that fits in u32; oversized boxes (size > 4 GiB) need the 64-bit
+/// large-size extension which we don't generate in the stub.
+fn wrap_box(box_type: &[u8; 4], body: &[u8]) -> Vec<u8> {
+    let size = (body.len() + 8) as u32;
+    let mut out = Vec::with_capacity(body.len() + 8);
+    out.extend_from_slice(&size.to_be_bytes());
+    out.extend_from_slice(box_type);
+    out.extend_from_slice(body);
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Decode the first box's size + type from `buf`.
+    fn read_box_header(buf: &[u8]) -> (u32, [u8; 4]) {
+        let size = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]);
+        let bt = [buf[4], buf[5], buf[6], buf[7]];
+        (size, bt)
+    }
+
+    #[test]
+    fn init_segment_starts_with_ftyp_then_moov() {
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = Fmp4Mux::new(&mut sink);
+        // Trigger init emission via a single (stubbed) write.
+        mux.write_video(0, true, &[0x00, 0x00, 0x00, 0x01, 0x40])
+            .unwrap();
+        mux.finish().unwrap();
+        drop(mux);
+
+        let (ftyp_size, ftyp_type) = read_box_header(&sink);
+        assert_eq!(&ftyp_type, b"ftyp");
+        assert!(ftyp_size >= 24, "ftyp too small: {ftyp_size}");
+
+        let (moov_size, moov_type) = read_box_header(&sink[ftyp_size as usize..]);
+        assert_eq!(&moov_type, b"moov");
+        assert!(moov_size > 100, "moov skeleton too small: {moov_size}");
+
+        // Stub guarantee: no media bytes after the init segment.
+        let total = ftyp_size as usize + moov_size as usize;
+        assert_eq!(sink.len(), total, "stub leaked media bytes past moov");
+    }
+
+    #[test]
+    fn moov_contains_trak_mvex() {
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = Fmp4Mux::new(&mut sink);
+        mux.write_video(0, true, &[]).unwrap();
+        mux.finish().unwrap();
+        drop(sink);
+
+        // Re-emit into a fresh buffer for parsing.
+        let mut buf: Vec<u8> = Vec::new();
+        let mut mux2 = Fmp4Mux::new(&mut buf);
+        mux2.write_video(0, true, &[]).unwrap();
+        mux2.finish().unwrap();
+        drop(mux2);
+
+        // Find moov payload start.
+        let (ftyp_size, _) = read_box_header(&buf);
+        let moov_start = ftyp_size as usize;
+        let (moov_size, _) = read_box_header(&buf[moov_start..]);
+        let moov_payload = &buf[moov_start + 8..moov_start + moov_size as usize];
+
+        // Scan for the trak and mvex four-CC anywhere in the moov payload.
+        let has_trak = moov_payload.windows(4).any(|w| w == b"trak");
+        let has_mvex = moov_payload.windows(4).any(|w| w == b"mvex");
+        assert!(has_trak, "moov missing trak");
+        assert!(has_mvex, "moov missing mvex");
+    }
+}

src/mux/hevc/mod.rs

@@ -0,0 +1,279 @@
+//! HEVC (H.265) elementary stream muxer — Annex B byte stream.
+//!
+//! Consumes [`PesFrame`](crate::pes::PesFrame)s for a single video track
+//! and writes them as a raw `.hevc` / `.h265` Annex B byte stream:
+//! `00 00 00 01 | NAL_unit | 00 00 00 01 | NAL_unit | …` with no
+//! container framing.
+//!
+//! On the first frame the muxer emits the codec_private's VPS, SPS, PPS
+//! (parsed from a `HEVCDecoderConfigurationRecord` in
+//! `length-prefixed-in-hvcC` form), then converts each PES frame's
+//! length-prefixed NAL units to Annex B and writes them.
+//!
+//! Sequential-only — no Cues, no backpatch. Target sink is any
+//! [`SequentialSink`](crate::io::sink::SequentialSink): file, socket,
+//! pipe, anything `Write + Send`.
+
+use std::io::{self, Write};
+
+/// Annex B 4-byte start code.
+pub(crate) const START_CODE: [u8; 4] = [0x00, 0x00, 0x00, 0x01];
+
+/// HEVC NAL unit type bits live in `(byte0 >> 1) & 0x3F` in Annex B.
+/// We don't filter NAL types here — the muxer is format-only — but we
+/// keep the constant as documentation of the field layout.
+#[allow(dead_code)]
+const HEVC_NAL_TYPE_MASK: u8 = 0x3F;
+
+/// Streaming HEVC Annex B muxer.
+///
+/// One instance per output stream. Tracks whether parameter sets have
+/// already been emitted so they're written exactly once at the head of
+/// the stream, mirroring the convention used by `ffmpeg -c:v copy -f
+/// hevc`.
+pub struct HevcMux<W: Write> {
+    writer: W,
+    /// `HEVCDecoderConfigurationRecord` payload (hvcC). Parsed lazily
+    /// on the first `write_frame` so callers can set it after
+    /// construction but before the first frame.
+    codec_private: Option<Vec<u8>>,
+    /// Set once VPS/SPS/PPS have been written to the stream. Subsequent
+    /// frames write only their own NAL units.
+    params_written: bool,
+}
+
+impl<W: Write> HevcMux<W> {
+    /// Construct over `writer`. The muxer does not impose any extra
+    /// buffering of its own — the sink owns its write buffering policy
+    /// (see [`LocalFileSink`](crate::io::sink::LocalFileSink) and
+    /// [`SocketSink`](crate::io::sink::SocketSink)).
+    pub fn new(writer: W) -> Self {
+        Self {
+            writer,
+            codec_private: None,
+            params_written: false,
+        }
+    }
+
+    /// Provide the `HEVCDecoderConfigurationRecord` (hvcC) so the muxer
+    /// can prepend VPS/SPS/PPS Annex B NALs at stream start. Optional —
+    /// if the PES frames already carry inline parameter sets (some
+    /// upstream demuxers do this), skipping this call is fine.
+    pub fn set_codec_private(&mut self, data: Vec<u8>) {
+        self.codec_private = Some(data);
+    }
+
+    /// Write one PES frame (= one access unit) as Annex B NAL units.
+    ///
+    /// Input may be either:
+    ///   - Length-prefixed: `[u32-BE len][NAL bytes]` repeated. This is
+    ///     the form emitted by libfreemkv's HEVC parser (the MKV-native
+    ///     layout). Converted to Annex B.
+    ///   - Already Annex B: bytes containing `00 00 00 01` start codes
+    ///     anywhere in the buffer. Passed through unchanged.
+    ///
+    /// `_pts_ns` is accepted for symmetry with other muxers but ignored
+    /// — Annex B has no timing layer.
+    pub fn write_frame(&mut self, _pts_ns: i64, data: &[u8]) -> io::Result<()> {
+        if !self.params_written {
+            if let Some(cp) = &self.codec_private {
+                if let Some(params) = hvcc_to_annex_b(cp) {
+                    self.writer.write_all(&params)?;
+                }
+            }
+            self.params_written = true;
+        }
+        let annex_b = length_prefixed_to_annex_b(data);
+        self.writer.write_all(&annex_b)
+    }
+
+    /// Flush the underlying writer. No trailer NAL is needed — an Annex
+    /// B stream ends whenever the file/socket ends.
+    pub fn finish(&mut self) -> io::Result<()> {
+        self.writer.flush()
+    }
+}
+
+/// Convert a `HEVCDecoderConfigurationRecord` (hvcC) into Annex B NAL
+/// units. Returns `Some(bytes)` if at least one NAL was extracted, else
+/// `None`.
+///
+/// Layout (per ISO/IEC 14496-15 §8.3.3.1.2):
+///   - 22-byte fixed header
+///   - byte 22 = `numOfArrays`
+///   - each array: `array_completeness:1 | reserved:1 | NAL_unit_type:6`,
+///     `numNalus:u16-BE`, then `numNalus` × `(nalUnitLength:u16-BE +
+///     NAL bytes)`.
+///
+/// We don't filter on NAL type — VPS (32), SPS (33), PPS (34), and any
+/// SEI arrays included in hvcC all get the same Annex B treatment.
+fn hvcc_to_annex_b(hvcc: &[u8]) -> Option<Vec<u8>> {
+    if hvcc.len() < 23 {
+        return None;
+    }
+    let num_arrays = hvcc[22] as usize;
+    let mut out = Vec::new();
+    let mut offset = 23;
+    for _ in 0..num_arrays {
+        if offset + 3 > hvcc.len() {
+            break;
+        }
+        offset += 1; // array_completeness + nal_type byte
+        let num_nalus = u16::from_be_bytes([hvcc[offset], hvcc[offset + 1]]) as usize;
+        offset += 2;
+        for _ in 0..num_nalus {
+            if offset + 2 > hvcc.len() {
+                break;
+            }
+            let nal_len = u16::from_be_bytes([hvcc[offset], hvcc[offset + 1]]) as usize;
+            offset += 2;
+            if offset + nal_len > hvcc.len() {
+                break;
+            }
+            out.extend_from_slice(&START_CODE);
+            out.extend_from_slice(&hvcc[offset..offset + nal_len]);
+            offset += nal_len;
+        }
+    }
+    if out.is_empty() { None } else { Some(out) }
+}
+
+/// Convert length-prefixed NAL units (`[u32-BE len][NAL]` repeated) to
+/// Annex B (`00 00 00 01 [NAL]` repeated).
+///
+/// If the input doesn't parse as length-prefixed (no valid lengths
+/// extracted), it's returned unchanged on the assumption that it's
+/// already Annex B — some upstream paths (raw HEVC ES from disc) pass
+/// Annex B straight through the PES layer.
+pub(crate) fn length_prefixed_to_annex_b(data: &[u8]) -> Vec<u8> {
+    let mut out = Vec::with_capacity(data.len() + (data.len() / 32));
+    let mut offset = 0;
+    while offset + 4 <= data.len() {
+        let len = u32::from_be_bytes([
+            data[offset],
+            data[offset + 1],
+            data[offset + 2],
+            data[offset + 3],
+        ]) as usize;
+        offset += 4;
+        if offset + len > data.len() {
+            // Mid-NAL truncation — fall through to the pass-through path
+            // rather than emitting a half-NAL.
+            return data.to_vec();
+        }
+        out.extend_from_slice(&START_CODE);
+        out.extend_from_slice(&data[offset..offset + len]);
+        offset += len;
+    }
+    if out.is_empty() && !data.is_empty() {
+        // No length prefixes found — input is likely already Annex B.
+        return data.to_vec();
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn length_prefixed_converts_to_annex_b() {
+        // Two NALs: [3-byte payload AA BB CC] and [2-byte payload DD EE].
+        let mut buf = Vec::new();
+        buf.extend_from_slice(&3u32.to_be_bytes());
+        buf.extend_from_slice(&[0xAA, 0xBB, 0xCC]);
+        buf.extend_from_slice(&2u32.to_be_bytes());
+        buf.extend_from_slice(&[0xDD, 0xEE]);
+
+        let got = length_prefixed_to_annex_b(&buf);
+        let want = [
+            0x00, 0x00, 0x00, 0x01, 0xAA, 0xBB, 0xCC, // first NAL
+            0x00, 0x00, 0x00, 0x01, 0xDD, 0xEE, // second NAL
+        ];
+        assert_eq!(&got[..], &want[..]);
+    }
+
+    #[test]
+    fn already_annex_b_passes_through_when_no_lengths_match() {
+        // A buffer < 4 bytes can't parse a length prefix at all →
+        // pass-through path triggers.
+        let raw = [0xAA, 0xBB, 0xCC];
+        let got = length_prefixed_to_annex_b(&raw);
+        assert_eq!(&got[..], &raw[..]);
+    }
+
+    #[test]
+    fn mid_nal_truncation_returns_original() {
+        // `[u32-BE 100][only 3 bytes]` — length prefix claims 100 bytes
+        // but the input only has 3 after the prefix. We treat that as
+        // malformed and pass the original buffer through so receivers
+        // can attempt their own recovery.
+        let mut raw = Vec::new();
+        raw.extend_from_slice(&100u32.to_be_bytes());
+        raw.extend_from_slice(&[0xAA, 0xBB, 0xCC]);
+        let got = length_prefixed_to_annex_b(&raw);
+        assert_eq!(&got[..], &raw[..]);
+    }
+
+    #[test]
+    fn hvcc_extracts_vps_sps_pps() {
+        // Build a minimal-but-valid hvcC: 22-byte header, then 3 arrays
+        // (VPS / SPS / PPS), each with 1 NAL of a 4-byte payload that
+        // we can spot in the output.
+        let mut hvcc = vec![0u8; 22];
+        hvcc.push(3); // numOfArrays
+        for (nal_type, payload) in [
+            (32u8, [0x40, 0x01, 0x0C, 0x01]),
+            (33, [0x42, 0x01, 0x01, 0x01]),
+            (34, [0x44, 0x01, 0xC1, 0x72]),
+        ] {
+            hvcc.push(nal_type & 0x3F);
+            hvcc.extend_from_slice(&1u16.to_be_bytes()); // numNalus
+            hvcc.extend_from_slice(&(payload.len() as u16).to_be_bytes());
+            hvcc.extend_from_slice(&payload);
+        }
+
+        let annex_b = hvcc_to_annex_b(&hvcc).expect("at least one NAL");
+        // Three NALs × (4-byte start + 4-byte payload) = 24 bytes.
+        assert_eq!(annex_b.len(), 24);
+        assert_eq!(&annex_b[..4], &START_CODE);
+        assert_eq!(&annex_b[8..12], &START_CODE);
+        assert_eq!(&annex_b[16..20], &START_CODE);
+        assert_eq!(annex_b[4], 0x40); // VPS first byte
+        assert_eq!(annex_b[12], 0x42); // SPS first byte
+        assert_eq!(annex_b[20], 0x44); // PPS first byte
+    }
+
+    #[test]
+    fn mux_writes_params_then_frames() {
+        // Build hvcC with one SPS to verify params-once semantics.
+        let mut hvcc = vec![0u8; 22];
+        hvcc.push(1);
+        hvcc.push(33);
+        hvcc.extend_from_slice(&1u16.to_be_bytes());
+        hvcc.extend_from_slice(&3u16.to_be_bytes());
+        hvcc.extend_from_slice(&[0x42, 0x01, 0x01]);
+
+        let mut frame_data = Vec::new();
+        frame_data.extend_from_slice(&2u32.to_be_bytes());
+        frame_data.extend_from_slice(&[0xAA, 0xBB]);
+
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = HevcMux::new(&mut sink);
+        mux.set_codec_private(hvcc);
+        mux.write_frame(0, &frame_data).unwrap();
+        // Second frame — no SPS re-emission.
+        mux.write_frame(40_000_000, &frame_data).unwrap();
+        mux.finish().unwrap();
+
+        // SPS NAL (7 bytes) + 2× frame NAL (6 bytes) = 19 bytes.
+        assert_eq!(sink.len(), 7 + 6 + 6);
+        // Start codes at offsets 0 (SPS), 7 (frame1), 13 (frame2).
+        assert_eq!(&sink[0..4], &START_CODE);
+        assert_eq!(&sink[7..11], &START_CODE);
+        assert_eq!(&sink[13..17], &START_CODE);
+        assert_eq!(sink[4], 0x42);
+        assert_eq!(sink[11], 0xAA);
+        assert_eq!(sink[17], 0xAA);
+    }
+}

src/mux/m2ts_mux/mod.rs

@@ -0,0 +1,681 @@
+//! Standard MPEG-TS (188-byte packets) muxer — sequential-only.
+//!
+//! Distinct from `super::tsmux::TsMuxer` (BD-TS with 192-byte packets
+//! and the 4-byte TP_extra_header). This muxer emits the IETF / ISO/IEC
+//! 13818-1 wire format that ffmpeg, VLC, and `m2tsindex` consume
+//! out of the box. Use it for plain `.ts` / `.m2ts` files over a
+//! [`SequentialSink`](crate::io::sink::SequentialSink), and for
+//! MPEG-TS-over-UDP via [`UdpSocketSink`](crate::io::sink::UdpSocketSink).
+//!
+//! ## Wire format
+//!
+//! Every output packet is exactly 188 bytes:
+//!
+//! ```text
+//! sync_byte:8 = 0x47
+//! transport_error_indicator:1 = 0
+//! payload_unit_start_indicator:1
+//! transport_priority:1 = 0
+//! PID:13
+//! transport_scrambling_control:2 = 0
+//! adaptation_field_control:2
+//! continuity_counter:4
+//! [adaptation field, if signalled]
+//! [payload, if signalled]
+//! ```
+//!
+//! ## Wiring
+//!
+//! Single program (PMT PID `0x1000`, program number `1`):
+//!   - PAT on PID `0x0000`
+//!   - PMT on PID `0x1000`
+//!   - Video (HEVC, stream_type `0x24`) on PID `0x0100`
+//!   - First audio (AC3, stream_type `0x81`) on PID `0x0101`
+//!     (or TrueHD, stream_type `0x83`, if hinted)
+//!
+//! ## Scope vs. full TS
+//!
+//! This is a deliberately minimal viable muxer:
+//!   - One program, one video track, optionally one audio track.
+//!   - PAT + PMT re-emitted every `PSI_INTERVAL_PACKETS` packets so a
+//!     mid-stream receiver can lock on within ~100 ms at typical
+//!     UHD bitrates.
+//!   - PCR clock derived from video PTS (`pts - PCR_LEAD_90KHZ`),
+//!     attached to the video PID's adaptation field every
+//!     `PCR_INTERVAL_PACKETS` packets.
+//!   - No language / descriptor tags, no SCTE-35 markers, no per-PID
+//!     PMT version bumps, no SDT/EIT. Sufficient for "ffmpeg can play
+//!     this back", not for full broadcast deployment.
+
+use std::io::{self, Write};
+
+mod packet;
+
+use packet::{Packet, PacketWriter};
+
+/// PID `0x0000` — PAT, mandated by spec.
+const PID_PAT: u16 = 0x0000;
+/// PID for the single program's PMT. `0x1000` is the conventional
+/// choice; anything outside reserved/null ranges works.
+const PID_PMT: u16 = 0x1000;
+/// PID for the video elementary stream.
+const PID_VIDEO: u16 = 0x0100;
+/// PID for the (optional) first audio elementary stream.
+const PID_AUDIO: u16 = 0x0101;
+/// Null PID — reserved by spec; never used here.
+#[allow(dead_code)]
+const PID_NULL: u16 = 0x1FFF;
+
+/// Re-emit PAT/PMT every N TS packets. ~250 × 188 B = 47 KB; at
+/// 80 Mb/s UHD that's ~5 ms — well under typical receiver lock budget.
+const PSI_INTERVAL_PACKETS: u64 = 250;
+/// Re-stamp PCR every N TS packets carrying the video PID. Spec MAX
+/// is 100 ms; 40 packets × 188 B at moderate bitrate keeps us under.
+const PCR_INTERVAL_PACKETS: u64 = 40;
+/// PCR lead time — the PCR must precede the PTS of the first byte of
+/// the picture it timestamps. 200 ms in 90 kHz ticks.
+const PCR_LEAD_90KHZ: u64 = 90_000 / 5;
+
+/// Stream-type codes from ISO/IEC 13818-1 Table 2-29 + later amendments.
+const STREAM_TYPE_HEVC: u8 = 0x24;
+const STREAM_TYPE_AC3: u8 = 0x81;
+const STREAM_TYPE_TRUEHD: u8 = 0x83;
+
+/// Audio codec hint for [`M2tsMux::new`] / [`M2tsMux::set_audio`]. The
+/// muxer needs to know the codec to pick the right PMT `stream_type`
+/// and the right PES stream_id; it doesn't decode the audio.
+#[derive(Debug, Clone, Copy)]
+pub enum AudioCodec {
+    /// AC-3 / E-AC-3, stream_type `0x81`, PES stream_id `0xBD`.
+    Ac3,
+    /// Dolby TrueHD, stream_type `0x83`, PES stream_id `0xBD`.
+    TrueHd,
+}
+
+impl AudioCodec {
+    fn stream_type(self) -> u8 {
+        match self {
+            AudioCodec::Ac3 => STREAM_TYPE_AC3,
+            AudioCodec::TrueHd => STREAM_TYPE_TRUEHD,
+        }
+    }
+}
+
+/// Sequential MPEG-TS muxer with one video and optional one audio track.
+pub struct M2tsMux<W: Write> {
+    out: PacketWriter<W>,
+    /// hvcC parameter set bytes to prepend to the first video PES.
+    /// Optional — if the upstream frames already carry inline params,
+    /// callers can omit this.
+    video_codec_private: Option<Vec<u8>>,
+    /// Set on first video frame: have we emitted VPS/SPS/PPS?
+    params_written: bool,
+    /// Audio codec, if an audio track is configured. `None` = video-only.
+    audio: Option<AudioCodec>,
+    /// First seen PTS (90 kHz). All subsequent PTS / PCR values are
+    /// relative to this so output streams start near t=0 and don't
+    /// confuse downstream parsers that don't tolerate huge starting
+    /// timestamps.
+    base_pts_90k: Option<u64>,
+    /// Per-PID continuity counter, 4 bits, monotonically increasing.
+    cc_video: u8,
+    cc_audio: u8,
+    cc_pat: u8,
+    cc_pmt: u8,
+    /// Total packets written, used to gate PSI / PCR cadence.
+    packets_written: u64,
+    /// Video packets written since last PCR, used to gate PCR cadence.
+    video_packets_since_pcr: u64,
+}
+
+impl<W: Write> M2tsMux<W> {
+    /// Construct a muxer wrapping `writer`. By default audio is
+    /// disabled; call [`set_audio`](Self::set_audio) before the first
+    /// frame to enable.
+    pub fn new(writer: W) -> Self {
+        Self {
+            out: PacketWriter::new(writer),
+            video_codec_private: None,
+            params_written: false,
+            audio: None,
+            base_pts_90k: None,
+            cc_video: 0,
+            cc_audio: 0,
+            cc_pat: 0,
+            cc_pmt: 0,
+            packets_written: 0,
+            video_packets_since_pcr: 0,
+        }
+    }
+
+    /// Provide the `HEVCDecoderConfigurationRecord` for video so the
+    /// muxer prepends VPS/SPS/PPS Annex B NALs at stream start.
+    pub fn set_video_codec_private(&mut self, hvcc: Vec<u8>) {
+        self.video_codec_private = Some(hvcc);
+    }
+
+    /// Enable a single audio track. Must be called before
+    /// [`write_audio`](Self::write_audio).
+    pub fn set_audio(&mut self, codec: AudioCodec) {
+        self.audio = Some(codec);
+    }
+
+    /// Write one video PES frame. `data` is either length-prefixed
+    /// NALUs (MKV-style) or already Annex B; both are accepted.
+    pub fn write_video(&mut self, pts_ns: i64, data: &[u8]) -> io::Result<()> {
+        let pts_90k = self.base_relative_pts(pts_ns);
+        // PCR comes "before" the PTS it timestamps; clamp at 0 for the
+        // first frame so we don't underflow.
+        let pcr = pts_90k.saturating_sub(PCR_LEAD_90KHZ);
+
+        // Annex-B-ify the frame and prepend VPS/SPS/PPS once.
+        let mut es = Vec::with_capacity(data.len() + 64);
+        if !self.params_written {
+            if let Some(cp) = &self.video_codec_private {
+                let payload = hvcc_payload(cp);
+                if !payload.is_empty() {
+                    let params = super::hevc::length_prefixed_to_annex_b(&payload);
+                    es.extend_from_slice(&params);
+                }
+            }
+            self.params_written = true;
+        }
+        let annex_b = super::hevc::length_prefixed_to_annex_b(data);
+        es.extend_from_slice(&annex_b);
+
+        let pes = build_video_pes(pts_90k, &es);
+        self.write_pes(PID_VIDEO, &pes, Some(pcr))
+    }
+
+    /// Write one audio PES frame. Returns `Ok(())` and silently drops
+    /// the frame if no audio track was configured — the design assumes
+    /// the upstream picks tracks and won't ship audio to a video-only
+    /// muxer, but defending against it keeps the API a single shape.
+    pub fn write_audio(&mut self, pts_ns: i64, data: &[u8]) -> io::Result<()> {
+        if self.audio.is_none() {
+            return Ok(());
+        }
+        let pts_90k = self.base_relative_pts(pts_ns);
+        let pes = build_audio_pes(pts_90k, data);
+        self.write_pes(PID_AUDIO, &pes, None)
+    }
+
+    /// Drain the underlying writer. No TS-level trailer is mandatory —
+    /// receivers detect end-of-stream from socket close or file EOF.
+    pub fn finish(&mut self) -> io::Result<()> {
+        self.out.flush()
+    }
+
+    /// Convert input PTS (nanoseconds) to 90 kHz ticks rebased on the
+    /// first frame's PTS. Saturating at 0 keeps the math friendly when
+    /// frames arrive slightly out of decode order.
+    fn base_relative_pts(&mut self, pts_ns: i64) -> u64 {
+        let raw_90k = if pts_ns > 0 {
+            (pts_ns as u64) * 9 / 100_000
+        } else {
+            0
+        };
+        let base = *self.base_pts_90k.get_or_insert(raw_90k);
+        raw_90k.saturating_sub(base)
+    }
+
+    /// Emit one PES payload as a chain of TS packets on `pid`. If `pcr`
+    /// is provided the first packet carries an adaptation field with
+    /// the PCR. PAT/PMT are re-emitted every `PSI_INTERVAL_PACKETS`.
+    ///
+    /// Packet-size math (TS = 188 bytes, header = 4 bytes):
+    ///   - 184 B remain after the header for the adaptation-field area
+    ///     plus the payload area.
+    ///   - With AF body of `b` bytes and `s` stuffing bytes: AF total =
+    ///     `1 + b + s` (the leading `1` is the `adaptation_field_length`
+    ///     byte itself). Payload = `184 - (1 + b + s)`.
+    ///   - With no AF area at all: payload = `184`.
+    ///
+    /// The fit-the-tail logic on the last packet of the PES uses
+    /// stuffing rather than a separate small packet, which is the
+    /// standard MPEG-TS convention.
+    fn write_pes(&mut self, pid: u16, pes: &[u8], pcr: Option<u64>) -> io::Result<()> {
+        // PSI cadence is enforced per TS packet — interleave a fresh
+        // PAT+PMT into the packet stream every PSI_INTERVAL_PACKETS so
+        // long single-PES emissions (e.g. one 60 KB video frame) don't
+        // starve receivers tuning in mid-stream.
+        let mut offset = 0;
+        let mut first = true;
+        while offset < pes.len() {
+            self.maybe_emit_psi()?;
+
+            let attach_pcr = first
+                && (pid == PID_VIDEO)
+                && (pcr.is_some())
+                && (self.packets_written == 0
+                    || self.video_packets_since_pcr >= PCR_INTERVAL_PACKETS);
+
+            let af_body: Vec<u8> = if attach_pcr {
+                build_pcr_adaptation(pcr.unwrap_or(0))
+            } else {
+                Vec::new()
+            };
+
+            let remaining = pes.len() - offset;
+            // Capacity for payload given AF body and 1-byte AF length.
+            // When AF body is empty we can still skip the AF entirely
+            // and get the full 184 B; only invoke the AF when we'd
+            // otherwise need stuffing.
+            let (af_present, payload_len, stuffing): (bool, usize, usize) = if !af_body.is_empty() {
+                // AF is mandatory (PCR). 1 byte length + body + stuffing
+                // + payload = 184.
+                let max_payload = 184 - 1 - af_body.len();
+                let p = remaining.min(max_payload);
+                let s = max_payload - p;
+                (true, p, s)
+            } else if remaining >= 184 {
+                // Full payload packet — no AF at all.
+                (false, 184, 0)
+            } else {
+                // Last (small) packet — stuff via empty AF.
+                // 1 byte length + 0 body + stuffing + payload = 184.
+                let max_payload = 183;
+                let p = remaining.min(max_payload);
+                let s = max_payload - p;
+                (true, p, s)
+            };
+
+            let cc = self.advance_cc(pid);
+            let mut packet = Packet::new();
+            packet.set_header(pid, first, true, af_present, cc);
+            if af_present {
+                packet.append_adaptation(&af_body, stuffing);
+            }
+            packet.append_payload(&pes[offset..offset + payload_len]);
+            debug_assert_eq!(packet.len(), 188, "packet not 188 bytes");
+            self.out.write_packet(&packet)?;
+
+            self.packets_written += 1;
+            if pid == PID_VIDEO {
+                if attach_pcr {
+                    self.video_packets_since_pcr = 0;
+                } else {
+                    self.video_packets_since_pcr += 1;
+                }
+            }
+            offset += payload_len;
+            first = false;
+        }
+        Ok(())
+    }
+
+    fn advance_cc(&mut self, pid: u16) -> u8 {
+        let slot = match pid {
+            PID_VIDEO => &mut self.cc_video,
+            PID_AUDIO => &mut self.cc_audio,
+            PID_PAT => &mut self.cc_pat,
+            PID_PMT => &mut self.cc_pmt,
+            _ => return 0,
+        };
+        let cc = *slot;
+        *slot = (*slot + 1) & 0x0F;
+        cc
+    }
+
+    fn maybe_emit_psi(&mut self) -> io::Result<()> {
+        if self.packets_written == 0 || self.packets_written % PSI_INTERVAL_PACKETS == 0 {
+            self.emit_pat()?;
+            self.emit_pmt()?;
+        }
+        Ok(())
+    }
+
+    fn emit_pat(&mut self) -> io::Result<()> {
+        let payload = build_pat(PID_PMT);
+        let cc = self.advance_cc(PID_PAT);
+        let mut packet = Packet::new();
+        packet.set_header(PID_PAT, true, true, false, cc);
+        packet.append_payload(&payload);
+        packet.pad_to_188();
+        self.out.write_packet(&packet)?;
+        self.packets_written += 1;
+        Ok(())
+    }
+
+    fn emit_pmt(&mut self) -> io::Result<()> {
+        let payload = build_pmt(self.audio);
+        let cc = self.advance_cc(PID_PMT);
+        let mut packet = Packet::new();
+        packet.set_header(PID_PMT, true, true, false, cc);
+        packet.append_payload(&payload);
+        packet.pad_to_188();
+        self.out.write_packet(&packet)?;
+        self.packets_written += 1;
+        Ok(())
+    }
+}
+
+/// Extract the raw hvcC bytes for handoff to `length_prefixed_to_annex_b`.
+/// hvcC layout: 22-byte fixed header, then `numOfArrays` arrays of
+/// `(nalType, numNalus, [nalLength:u16, NAL bytes]…)`. We convert this
+/// directly to a length-prefixed byte stream (NAL length is u16 in
+/// hvcC; widen to u32 for the standard length-prefixed encoding).
+fn hvcc_payload(hvcc: &[u8]) -> Vec<u8> {
+    if hvcc.len() < 23 {
+        return Vec::new();
+    }
+    let num_arrays = hvcc[22] as usize;
+    let mut out = Vec::new();
+    let mut offset = 23;
+    for _ in 0..num_arrays {
+        if offset + 3 > hvcc.len() {
+            break;
+        }
+        offset += 1;
+        let num_nalus = u16::from_be_bytes([hvcc[offset], hvcc[offset + 1]]) as usize;
+        offset += 2;
+        for _ in 0..num_nalus {
+            if offset + 2 > hvcc.len() {
+                break;
+            }
+            let nal_len = u16::from_be_bytes([hvcc[offset], hvcc[offset + 1]]) as usize;
+            offset += 2;
+            if offset + nal_len > hvcc.len() {
+                break;
+            }
+            out.extend_from_slice(&(nal_len as u32).to_be_bytes());
+            out.extend_from_slice(&hvcc[offset..offset + nal_len]);
+            offset += nal_len;
+        }
+    }
+    out
+}
+
+/// Build a PES packet for a video access unit.
+fn build_video_pes(pts_90k: u64, es: &[u8]) -> Vec<u8> {
+    build_pes_packet(0xE0, pts_90k, es, /* length_in_header */ false)
+}
+
+/// Build a PES packet for an audio access unit.
+fn build_audio_pes(pts_90k: u64, es: &[u8]) -> Vec<u8> {
+    // Audio PES: length is fillable when it fits in u16. We always
+    // write the length so receivers don't have to scan for the next
+    // start code.
+    build_pes_packet(0xBD, pts_90k, es, /* length_in_header */ true)
+}
+
+fn build_pes_packet(stream_id: u8, pts_90k: u64, es: &[u8], length_in_header: bool) -> Vec<u8> {
+    let mut out = Vec::with_capacity(es.len() + 14);
+    out.extend_from_slice(&[0x00, 0x00, 0x01, stream_id]);
+    // PES_packet_length: total bytes after this field. 3 flag bytes + 5
+    // PTS bytes + es.len(). Zero means "unbounded" — used for video
+    // where PES can exceed u16.
+    let pes_len = 8 + es.len();
+    if length_in_header && pes_len <= u16::MAX as usize {
+        out.extend_from_slice(&(pes_len as u16).to_be_bytes());
+    } else {
+        out.extend_from_slice(&[0x00, 0x00]);
+    }
+    // Flags byte 1: 10 = MPEG-2 marker, then scrambling/priority/etc 0.
+    out.push(0x80);
+    // Flags byte 2: PTS flag (bit 7).
+    out.push(0x80);
+    // PES_header_data_length = 5 (just PTS).
+    out.push(5);
+    // PTS bytes — 33-bit timestamp split across 5 bytes with marker bits.
+    let pts = pts_90k & 0x1_FFFF_FFFF;
+    out.push(0x21 | (((pts >> 29) & 0x0E) as u8));
+    out.push(((pts >> 22) & 0xFF) as u8);
+    out.push(0x01 | (((pts >> 14) & 0xFE) as u8));
+    out.push(((pts >> 7) & 0xFF) as u8);
+    out.push(0x01 | (((pts << 1) & 0xFE) as u8));
+    out.extend_from_slice(es);
+    out
+}
+
+/// Build the PAT payload (section, with pointer_field).
+fn build_pat(pmt_pid: u16) -> Vec<u8> {
+    let mut section = Vec::new();
+    section.push(0x00); // table_id = PAT
+    // section_syntax_indicator(1) | '0'(1) | reserved(2) | section_length(12)
+    // section_length covers from end of this field through CRC.
+    // Body: transport_stream_id(2) + version/cni(1) + section/last_section(2) + program(4) = 9 bytes,
+    // plus CRC(4) = 13. Encoded big-endian.
+    section.extend_from_slice(&[0xB0, 13]);
+    section.extend_from_slice(&[0x00, 0x01]); // transport_stream_id = 1
+    section.push(0xC1); // reserved | version=0 | current_next=1
+    section.push(0x00); // section_number
+    section.push(0x00); // last_section_number
+    section.extend_from_slice(&[0x00, 0x01]); // program_number = 1
+    // reserved(3) | network_PID/program_map_PID(13)
+    let pid_bytes = (0xE000u16 | (pmt_pid & 0x1FFF)).to_be_bytes();
+    section.extend_from_slice(&pid_bytes);
+    let crc = mpegts_crc32(&section);
+    section.extend_from_slice(&crc.to_be_bytes());
+    // Prepend pointer_field=0 (section starts immediately).
+    let mut payload = Vec::with_capacity(section.len() + 1);
+    payload.push(0x00);
+    payload.extend_from_slice(&section);
+    payload
+}
+
+/// Build the PMT payload (section, with pointer_field).
+fn build_pmt(audio: Option<AudioCodec>) -> Vec<u8> {
+    let mut section = Vec::new();
+    section.push(0x02); // table_id = PMT
+    // section_length filled in after we know the body size.
+    let len_placeholder = section.len();
+    section.extend_from_slice(&[0xB0, 0x00]);
+
+    section.extend_from_slice(&1u16.to_be_bytes()); // program_number
+    section.push(0xC1); // reserved | version=0 | current_next=1
+    section.push(0x00); // section_number
+    section.push(0x00); // last_section_number
+    // reserved(3) | PCR_PID(13)
+    let pcr_pid = (0xE000u16 | (PID_VIDEO & 0x1FFF)).to_be_bytes();
+    section.extend_from_slice(&pcr_pid);
+    // program_info_length = 0
+    section.extend_from_slice(&[0xF0, 0x00]);
+
+    // Video elementary stream entry.
+    section.push(STREAM_TYPE_HEVC);
+    let v_pid = (0xE000u16 | (PID_VIDEO & 0x1FFF)).to_be_bytes();
+    section.extend_from_slice(&v_pid);
+    section.extend_from_slice(&[0xF0, 0x00]); // ES_info_length = 0
+
+    if let Some(codec) = audio {
+        section.push(codec.stream_type());
+        let a_pid = (0xE000u16 | (PID_AUDIO & 0x1FFF)).to_be_bytes();
+        section.extend_from_slice(&a_pid);
+        section.extend_from_slice(&[0xF0, 0x00]);
+    }
+
+    // Now patch section_length: covers everything after the length field
+    // through the CRC, so (current body size - 3 bytes consumed by
+    // table_id + 2 length bytes) + 4 (CRC).
+    let section_len = section.len() - 3 + 4;
+    section[len_placeholder] = 0xB0 | ((section_len >> 8) as u8 & 0x0F);
+    section[len_placeholder + 1] = section_len as u8;
+
+    let crc = mpegts_crc32(&section);
+    section.extend_from_slice(&crc.to_be_bytes());
+
+    let mut payload = Vec::with_capacity(section.len() + 1);
+    payload.push(0x00);
+    payload.extend_from_slice(&section);
+    payload
+}
+
+/// Build the adaptation field carrying a PCR (no other flags).
+fn build_pcr_adaptation(pcr_90k: u64) -> Vec<u8> {
+    // adaptation_field_length is set by `Packet::append_adaptation`
+    // — this function returns just the field body.
+    //
+    // Layout: discontinuity_indicator(1) | random_access(1) |
+    // elementary_stream_priority(1) | PCR_flag(1) | OPCR_flag(1) |
+    // splicing_point_flag(1) | transport_private_data_flag(1) |
+    // adaptation_field_extension_flag(1) | PCR(48b).
+    let mut af = vec![0x50]; // PCR_flag=1, random_access_indicator=1
+    let pcr_base = pcr_90k & 0x1_FFFF_FFFF; // 33-bit
+    let pcr_ext: u16 = 0; // 9-bit, we keep it zero (no sub-tick precision)
+    // Encode PCR: 33b base | 6b reserved | 9b extension = 48b
+    af.push((pcr_base >> 25) as u8);
+    af.push((pcr_base >> 17) as u8);
+    af.push((pcr_base >> 9) as u8);
+    af.push((pcr_base >> 1) as u8);
+    af.push(((pcr_base << 7) as u8 & 0x80) | 0x7E | ((pcr_ext >> 8) as u8 & 0x01));
+    af.push(pcr_ext as u8);
+    af
+}
+
+/// MPEG-TS CRC-32 (poly 0x04C11DB7, init 0xFFFFFFFF, no reflection, no
+/// final XOR). Implementation: bitwise so we don't need a table —
+/// PSI sections are tiny, the cost is negligible.
+fn mpegts_crc32(data: &[u8]) -> u32 {
+    let mut crc: u32 = 0xFFFF_FFFF;
+    for &b in data {
+        crc ^= (b as u32) << 24;
+        for _ in 0..8 {
+            if crc & 0x8000_0000 != 0 {
+                crc = (crc << 1) ^ 0x04C1_1DB7;
+            } else {
+                crc <<= 1;
+            }
+        }
+    }
+    crc
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// All emitted bytes must align to 188-byte packet boundaries and
+    /// every packet must start with `0x47`.
+    fn assert_ts_well_formed(buf: &[u8]) {
+        assert_eq!(
+            buf.len() % 188,
+            0,
+            "stream not packet-aligned: {} bytes",
+            buf.len()
+        );
+        for (i, chunk) in buf.chunks(188).enumerate() {
+            assert_eq!(chunk[0], 0x47, "packet {} missing sync byte", i);
+        }
+    }
+
+    fn extract_pids(buf: &[u8]) -> Vec<u16> {
+        buf.chunks(188)
+            .map(|p| u16::from_be_bytes([p[1] & 0x1F, p[2]]))
+            .collect()
+    }
+
+    #[test]
+    fn crc32_is_self_validating() {
+        // The MPEG-TS CRC has the property that prepending a single
+        // bit-flip changes the output; running it over its own input +
+        // CRC yields a fixed magic constant (the CRC residue). Rather
+        // than hardcoding sample bytes, verify the underlying algorithm
+        // by checking that two distinct inputs produce distinct CRCs
+        // and that the same input is deterministic.
+        let a = [
+            0u8, 0xB0, 0x0D, 0x00, 0x01, 0xC1, 0x00, 0x00, 0x00, 0x01, 0xE1, 0x00,
+        ];
+        let mut b = a;
+        b[5] ^= 0x01; // flip one bit
+        let crc_a = mpegts_crc32(&a);
+        let crc_b = mpegts_crc32(&b);
+        assert_ne!(crc_a, crc_b);
+        assert_eq!(crc_a, mpegts_crc32(&a)); // deterministic
+        // Sanity: all-zero input ⇒ CRC = 0 (init XORs but the
+        // shift/feedback cancels for zero data after init drains).
+        // We don't assert exact value — that depends on poly choice —
+        // but check it's not the same as for non-zero data.
+        let crc_zero = mpegts_crc32(&[0u8; 12]);
+        assert_ne!(crc_zero, crc_a);
+    }
+
+    #[test]
+    fn video_only_mux_emits_pat_pmt_then_video() {
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = M2tsMux::new(&mut sink);
+        // One small video frame, no codec_private (so no params NAL inline).
+        let mut frame = Vec::new();
+        frame.extend_from_slice(&4u32.to_be_bytes());
+        frame.extend_from_slice(&[0x40, 0x01, 0x0C, 0x01]);
+        mux.write_video(0, &frame).unwrap();
+        mux.finish().unwrap();
+        drop(mux);
+
+        assert_ts_well_formed(&sink);
+        let pids = extract_pids(&sink);
+        // First two packets: PAT, PMT. At least one video packet after.
+        assert_eq!(pids[0], PID_PAT);
+        assert_eq!(pids[1], PID_PMT);
+        assert!(pids.iter().any(|p| *p == PID_VIDEO));
+    }
+
+    #[test]
+    fn audio_track_appears_in_pmt_and_stream() {
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = M2tsMux::new(&mut sink);
+        mux.set_audio(AudioCodec::Ac3);
+        // Video + audio frame pair.
+        let mut frame = Vec::new();
+        frame.extend_from_slice(&3u32.to_be_bytes());
+        frame.extend_from_slice(&[0x40, 0x01, 0x0C]);
+        mux.write_video(0, &frame).unwrap();
+        mux.write_audio(20_000_000, &[0x0B, 0x77, 0x12, 0x34])
+            .unwrap();
+        mux.finish().unwrap();
+        drop(mux);
+
+        assert_ts_well_formed(&sink);
+        let pids = extract_pids(&sink);
+        assert!(pids.iter().any(|p| *p == PID_VIDEO));
+        assert!(pids.iter().any(|p| *p == PID_AUDIO));
+    }
+
+    #[test]
+    fn psi_re_emits_at_interval() {
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = M2tsMux::new(&mut sink);
+        // Build a frame large enough to span > PSI_INTERVAL_PACKETS TS packets.
+        // 184 B payload per packet ⇒ ~250 packets = ~46 KB elementary stream.
+        let big: Vec<u8> = (0..(60 * 1024)).map(|i| (i & 0xff) as u8).collect();
+        let mut frame = Vec::new();
+        frame.extend_from_slice(&(big.len() as u32).to_be_bytes());
+        frame.extend_from_slice(&big);
+        mux.write_video(0, &frame).unwrap();
+        mux.finish().unwrap();
+        drop(mux);
+
+        assert_ts_well_formed(&sink);
+        let pids = extract_pids(&sink);
+        // Count PAT/PMT pairs — must be at least 2 given the input size.
+        let pat_count = pids.iter().filter(|p| **p == PID_PAT).count();
+        let pmt_count = pids.iter().filter(|p| **p == PID_PMT).count();
+        assert!(pat_count >= 2, "expected ≥2 PAT, got {}", pat_count);
+        assert!(pmt_count >= 2, "expected ≥2 PMT, got {}", pmt_count);
+    }
+
+    #[test]
+    fn continuity_counter_increments_per_pid() {
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = M2tsMux::new(&mut sink);
+        // Three small video frames to get a sequence of video TS packets.
+        for pts in [0i64, 40_000_000, 80_000_000] {
+            let mut frame = Vec::new();
+            frame.extend_from_slice(&3u32.to_be_bytes());
+            frame.extend_from_slice(&[0xAA, 0xBB, 0xCC]);
+            mux.write_video(pts, &frame).unwrap();
+        }
+        mux.finish().unwrap();
+        drop(mux);
+
+        // Collect CCs for video packets in order.
+        let ccs: Vec<u8> = sink
+            .chunks(188)
+            .filter(|p| u16::from_be_bytes([p[1] & 0x1F, p[2]]) == PID_VIDEO)
+            .map(|p| p[3] & 0x0F)
+            .collect();
+        for w in ccs.windows(2) {
+            assert_eq!(w[1], (w[0] + 1) & 0x0F);
+        }
+    }
+}

src/mux/m2ts_mux/packet.rs

@@ -0,0 +1,145 @@
+//! 188-byte MPEG-TS packet builder + writer.
+//!
+//! Internal helper for `super::M2tsMux`. Not public API — exposes raw
+//! byte layout so the parent module can compose PSI / PCR / PES bytes
+//! without each caller re-implementing the 188-byte boundary math.
+
+use std::io::{self, Write};
+
+const TS_PACKET_SIZE: usize = 188;
+const SYNC_BYTE: u8 = 0x47;
+const STUFF_BYTE: u8 = 0xFF;
+
+/// One TS packet under construction. Always emits 188 bytes when
+/// [`pad_to_188`](Self::pad_to_188) is called; if it's not called the
+/// caller is responsible for filling the packet exactly.
+pub(super) struct Packet {
+    buf: Vec<u8>,
+}
+
+impl Packet {
+    pub(super) fn new() -> Self {
+        Self {
+            buf: Vec::with_capacity(TS_PACKET_SIZE),
+        }
+    }
+
+    /// Write the 4-byte TS packet header.
+    ///
+    /// * `pid` — 13-bit PID
+    /// * `payload_unit_start` — first packet of a PES / PSI section
+    /// * `has_payload` — packet carries any payload bytes
+    /// * `has_adaptation` — packet carries an adaptation field
+    /// * `cc` — 4-bit continuity counter
+    pub(super) fn set_header(
+        &mut self,
+        pid: u16,
+        payload_unit_start: bool,
+        has_payload: bool,
+        has_adaptation: bool,
+        cc: u8,
+    ) {
+        self.buf.clear();
+        self.buf.push(SYNC_BYTE);
+        let pus_bit = if payload_unit_start { 0x40 } else { 0 };
+        // transport_error_indicator(1)=0 | payload_unit_start(1) | transport_priority(1)=0 | PID(5 high)
+        self.buf.push(pus_bit | ((pid >> 8) as u8 & 0x1F));
+        self.buf.push(pid as u8);
+        // transport_scrambling_control(2)=0 | adaptation_field_control(2) | continuity_counter(4)
+        let afc = match (has_adaptation, has_payload) {
+            (false, false) => 0b00, // reserved — should not happen
+            (false, true) => 0b01,  // payload only
+            (true, false) => 0b10,  // adaptation only
+            (true, true) => 0b11,   // both
+        };
+        self.buf.push((afc << 4) | (cc & 0x0F));
+    }
+
+    /// Append the adaptation field after the header.
+    ///
+    /// `body` is the adaptation field body (flags byte plus optional PCR
+    /// and so on). `stuffing` is the number of `0xFF` stuffing bytes to
+    /// append after the body. The first byte of the field
+    /// (`adaptation_field_length`) is computed here from
+    /// `body.len() + stuffing`.
+    pub(super) fn append_adaptation(&mut self, body: &[u8], stuffing: usize) {
+        let af_len = body.len() + stuffing;
+        debug_assert!(af_len <= 183, "adaptation field overflow");
+        self.buf.push(af_len as u8);
+        self.buf.extend_from_slice(body);
+        for _ in 0..stuffing {
+            self.buf.push(STUFF_BYTE);
+        }
+    }
+
+    /// Append payload bytes.
+    pub(super) fn append_payload(&mut self, payload: &[u8]) {
+        self.buf.extend_from_slice(payload);
+        debug_assert!(self.buf.len() <= TS_PACKET_SIZE, "packet overflow");
+    }
+
+    /// Pad the packet to exactly 188 bytes with `0xFF` bytes — used by
+    /// PSI emit paths where the section is much smaller than 184 bytes.
+    /// For PSI packets only — payload-carrying packets reserve room for
+    /// stuffing via `append_adaptation`.
+    pub(super) fn pad_to_188(&mut self) {
+        while self.buf.len() < TS_PACKET_SIZE {
+            self.buf.push(STUFF_BYTE);
+        }
+    }
+
+    pub(super) fn bytes(&self) -> &[u8] {
+        &self.buf
+    }
+
+    pub(super) fn len(&self) -> usize {
+        self.buf.len()
+    }
+}
+
+/// Buffered writer for assembled TS packets. Owns the underlying sink.
+pub(super) struct PacketWriter<W: Write> {
+    inner: W,
+}
+
+impl<W: Write> PacketWriter<W> {
+    pub(super) fn new(inner: W) -> Self {
+        Self { inner }
+    }
+
+    pub(super) fn write_packet(&mut self, packet: &Packet) -> io::Result<()> {
+        let bytes = packet.bytes();
+        debug_assert_eq!(bytes.len(), TS_PACKET_SIZE);
+        self.inner.write_all(bytes)
+    }
+
+    pub(super) fn flush(&mut self) -> io::Result<()> {
+        self.inner.flush()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn pad_fills_to_188() {
+        let mut p = Packet::new();
+        p.set_header(0x100, true, true, false, 0);
+        p.append_payload(&[1, 2, 3]);
+        p.pad_to_188();
+        assert_eq!(p.bytes().len(), 188);
+        assert_eq!(p.bytes()[0], SYNC_BYTE);
+        // After 4-byte header + 3 payload, byte 7 starts stuffing.
+        assert_eq!(p.bytes()[7], STUFF_BYTE);
+    }
+
+    #[test]
+    fn header_pid_round_trips() {
+        let mut p = Packet::new();
+        p.set_header(0x1ABC, false, true, false, 0xA);
+        let pid = u16::from_be_bytes([p.bytes()[1] & 0x1F, p.bytes()[2]]);
+        assert_eq!(pid, 0x1ABC);
+        assert_eq!(p.bytes()[3] & 0x0F, 0xA);
+    }
+}

src/mux/mod.rs

@@ -33,6 +33,17 @@ pub(crate) mod m2ts;
 /// to round-trip codec_privates that don't fit inside the underlying format).
 /// Exposed for integration tests that exercise the wire format directly.
 pub mod meta;
+
+// ── Phase 3 sequential muxers ──────────────────────────────────────────────
+//
+// New container muxers that consume PES frames and write to a
+// `SequentialSink`. They are NOT refactors of the existing `MkvStream` /
+// `M2tsStream` (which round-trip via the legacy `Stream` trait + the
+// BD-TS framing); they're sequential-only and target the Phase 2 sink
+// split end-to-end.
+pub mod fmp4;
+pub mod hevc;
+pub mod m2ts_mux;
 pub(crate) mod mkv;
 pub(crate) mod mkvstream;
 pub(crate) mod network;

src/platform/fs_type/linux.rs

@@ -32,8 +32,12 @@ pub(super) fn detect_impl(path: &Path) -> FsType {
     }
     // `f_type` is signed (`__fsword_t`) on glibc and unsigned
     // (`c_ulong`) on musl. Cast both sides to i64 for a portable
-    // comparison.
+    // comparison. On glibc x86_64 both already are i64 — clippy flags
+    // the cast as unnecessary on that target only, but we need it for
+    // musl, so silence the lint.
+    #[allow(clippy::unnecessary_cast)]
     let f_type = buf.f_type as i64;
+    #[allow(clippy::unnecessary_cast)]
     let nfs_magic = libc::NFS_SUPER_MAGIC as i64;
     if f_type == nfs_magic {
         return FsType::Nfs;

src/platform/fs_type/mod.rs

@@ -37,19 +37,19 @@ pub enum FsType {
 mod linux;
 #[cfg(target_os = "macos")]
 mod macos;
-#[cfg(target_os = "windows")]
-mod windows;
 #[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
 mod other;
+#[cfg(target_os = "windows")]
+mod windows;
 
 #[cfg(target_os = "linux")]
 use linux::detect_impl;
 #[cfg(target_os = "macos")]
 use macos::detect_impl;
-#[cfg(target_os = "windows")]
-use windows::detect_impl;
 #[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
 use other::detect_impl;
+#[cfg(target_os = "windows")]
+use windows::detect_impl;
 
 /// Best-effort classification of the filesystem under `path`.
 ///