0.18 round 1 polish: address libfreemkv code-review findings

src/halt.rs

@@ -19,7 +19,12 @@ use std::sync::atomic::{AtomicBool, Ordering};
 /// Clones share the same underlying flag. `cancel()` is one-way; there is
 /// no `reset()` by design — construct a fresh `Halt` for a fresh
 /// operation.
-#[derive(Clone, Debug, Default)]
+///
+/// Construct with [`Halt::new`]. We intentionally don't derive
+/// `Default` — `Halt::new()` is more discoverable, matches the
+/// stdlib `Mutex::new` / `Arc::new` convention, and keeps the
+/// uncancelled-by-construction invariant in one named place.
+#[derive(Clone, Debug)]
 pub struct Halt(Arc<AtomicBool>);
 
 impl Halt {
@@ -99,9 +104,4 @@ mod tests {
         assert!(h.is_cancelled());
     }
 
-    #[test]
-    fn default_impl_is_uncancelled() {
-        let h = Halt::default();
-        assert!(!h.is_cancelled());
-    }
 }

src/io/mod.rs

@@ -26,4 +26,4 @@ pub(crate) use writeback_file::WritebackFile;
 // `disc/sweep_pipeline.rs`; patch and mux have no pipeline). The next
 // 0.18 slice removes this allow as it wires up the first consumer.
 #[allow(unused_imports)]
-pub use pipeline::{Apply, DEFAULT_DEPTH, Pipeline, Sink};
+pub use pipeline::{DEFAULT_PIPELINE_DEPTH, Flow, Pipeline, Sink, WRITE_THROUGH_DEPTH};

src/io/pipeline.rs

@@ -15,7 +15,7 @@
 //! - Producer dropping the channel (via `Pipeline::finish` dropping
 //!   `tx`) signals end-of-stream; consumer flushes via `close()` and
 //!   returns its `Output`.
-//! - Consumer returning [`Apply::Stop`] also calls `close()` and
+//! - Consumer returning [`Flow::Stop`] also calls `close()` and
 //!   returns its `Output`. `send()` from the producer will then either
 //!   succeed (if the item already fit in the channel buffer) or fail
 //!   with `Err(item)` once the consumer has dropped its receiver.
@@ -28,13 +28,13 @@
 //!
 //! ## Dead-code suppression
 //!
-//! The `Pipeline` / `Sink` / `Apply` / `DEFAULT_DEPTH` items are
-//! crate-internal API today (the parent `io` module is
-//! `pub(crate)`) but have no in-tree callers in this slice — sweep
-//! is still on `disc/sweep_pipeline.rs`, patch and mux still have
-//! no pipeline at all. Wiring them up is the next slice of the
-//! 0.18 redesign. The `#[allow]` below is removed once any of
-//! those three call sites lands on this primitive.
+//! The `Pipeline` / `Sink` / `Flow` / `DEFAULT_PIPELINE_DEPTH` /
+//! `WRITE_THROUGH_DEPTH` items are crate-internal API today (the
+//! parent `io` module is `pub(crate)`) but have no in-tree callers
+//! in this slice — sweep is still on `disc/sweep_pipeline.rs`, patch
+//! and mux still have no pipeline at all. Wiring them up is the
+//! next slice of the 0.18 redesign. The `#[allow]` below is removed
+//! once any of those three call sites lands on this primitive.
 
 #![allow(dead_code)]
 
@@ -44,14 +44,29 @@ use std::thread::{self, JoinHandle};
 
 use crate::error::Error;
 
-/// Default channel depth for callers that don't have a specific
-/// reason to pick another value. Sweep and mux are both expected to
-/// use this; patch may want `1` (write-through).
-pub const DEFAULT_DEPTH: usize = 4;
+/// Default channel depth for callers without a specific reason to
+/// pick another value.
+///
+/// Empirically tuned for sweep and mux — both want enough slack that
+/// short consumer stalls don't immediately back up onto the producer,
+/// but not so much that a producer outpacing the consumer accumulates
+/// arbitrary buffered work. `4` matches the depth `disc/sweep_pipeline.rs`
+/// has used since 0.17.11. Patch should usually use
+/// [`WRITE_THROUGH_DEPTH`] (`1`) instead — write-through gives clean
+/// back-pressure between every read attempt and the matching write,
+/// which matters when the consumer is updating the mapfile in lockstep.
+pub const DEFAULT_PIPELINE_DEPTH: usize = 4;
 
-/// Outcome of [`Sink::apply`]: either keep feeding items, or stop the
-/// pipeline early and run `close()`.
-pub enum Apply {
+/// Channel depth for write-through pipelines. Each `send` fully
+/// drains before the next can enqueue. Use this when the producer
+/// must observe consumer side-effects (e.g. mapfile state) before
+/// emitting the next item.
+pub const WRITE_THROUGH_DEPTH: usize = 1;
+
+/// Outcome of [`Sink::apply`]: either keep feeding items
+/// ([`Flow::Continue`]), or stop the pipeline early and run `close()`
+/// ([`Flow::Stop`]).
+pub enum Flow {
     Continue,
     Stop,
 }
@@ -64,16 +79,16 @@ pub trait Sink<I>: Send + 'static {
     /// [`Pipeline::finish`].
     type Output: Send + 'static;
 
-    /// Apply one item. Returning [`Apply::Continue`] keeps the
-    /// pipeline running; [`Apply::Stop`] ends it cleanly (still calls
+    /// Apply one item. Returning [`Flow::Continue`] keeps the
+    /// pipeline running; [`Flow::Stop`] ends it cleanly (still calls
     /// `close()`). An error short-circuits: `close()` is *not* called
     /// and the error is what `finish()` will return, but the consumer
     /// keeps draining the channel so the producer never blocks on a
     /// dead receiver.
-    fn apply(&mut self, item: I) -> Result<Apply, Error>;
+    fn apply(&mut self, item: I) -> Result<Flow, Error>;
 
     /// Called once at end-of-stream — either because the producer
-    /// dropped `tx` or because `apply` returned [`Apply::Stop`]. Use
+    /// dropped `tx` or because `apply` returned [`Flow::Stop`]. Use
     /// this to flush, fsync, finalise. Skipped if any prior `apply`
     /// returned `Err`.
     fn close(self) -> Result<Self::Output, Error>;
@@ -91,8 +106,11 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
     /// [`Sink`].
     ///
     /// The thread is named `freemkv-pipeline-consumer` so it shows up
-    /// distinctly in stack traces and `top -H`.
-    pub fn spawn<S: Sink<I, Output = R>>(depth: usize, sink: S) -> Self {
+    /// distinctly in stack traces and `top -H`. Returns an
+    /// `Error::IoError` if the OS refuses the thread spawn (resource
+    /// exhaustion); callers already operate in fallible context, so
+    /// this is propagated rather than panicked.
+    pub fn spawn<S: Sink<I, Output = R>>(depth: usize, sink: S) -> Result<Self, Error> {
         let (tx, rx) = sync_channel::<I>(depth);
         let handle = thread::Builder::new()
             .name("freemkv-pipeline-consumer".into())
@@ -109,8 +127,8 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
                         continue;
                     }
                     match sink.apply(item) {
-                        Ok(Apply::Continue) => {}
-                        Ok(Apply::Stop) => {
+                        Ok(Flow::Continue) => {}
+                        Ok(Flow::Stop) => {
                             stopped = true;
                         }
                         Err(e) => {
@@ -124,15 +142,22 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
                     None => sink.close(),
                 }
             })
-            .expect("spawning a thread should not fail");
+            .map_err(|e| Error::IoError { source: e })?;
 
-        Pipeline { tx, handle }
+        Ok(Pipeline { tx, handle })
     }
 
     /// Push one item. Blocks if the channel is full — that's the
     /// back-pressure the whole primitive exists to provide. Returns
     /// the item back if the consumer thread is gone (panicked or
     /// already returned).
+    ///
+    /// After the consumer returns [`Flow::Stop`], `send` will silently
+    /// buffer items into the channel until the channel fills, then
+    /// return `Err(item)` once the consumer has dropped its receiver.
+    /// Producers that need to stop pushing on `Stop` should track an
+    /// independent signal (e.g. `Halt`) — `send` alone is not the
+    /// notification edge.
     pub fn send(&self, item: I) -> Result<(), I> {
         self.tx.send(item).map_err(|e| e.0)
     }
@@ -141,7 +166,8 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
     /// thread to finish. Returns whatever the consumer's `close()`
     /// produced, or the first `apply` error, or — on consumer panic —
     /// an `Error::IoError` whose source is `io::Error::other(...)`
-    /// with a "panicked" message.
+    /// with a "pipeline consumer panicked: <payload>" message
+    /// (callers can match on the constant prefix).
     pub fn finish(self) -> Result<R, Error> {
         let Pipeline { tx, handle } = self;
         // Explicit drop, although the destructure already drops `tx`
@@ -149,9 +175,20 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
         drop(tx);
         match handle.join() {
             Ok(result) => result,
-            Err(_) => Err(Error::IoError {
-                source: io::Error::other("pipeline consumer panicked"),
-            }),
+            Err(payload) => {
+                // Preserve the original panic message when the
+                // consumer's panic payload was a `&str` or `String`
+                // (the two stdlib formats that `panic!` produces).
+                // Anything else falls back to "(no message)".
+                let msg = payload
+                    .downcast_ref::<&'static str>()
+                    .copied()
+                    .or_else(|| payload.downcast_ref::<String>().map(|s| s.as_str()))
+                    .unwrap_or("(no message)");
+                Err(Error::IoError {
+                    source: io::Error::other(format!("pipeline consumer panicked: {msg}")),
+                })
+            }
         }
     }
 }
@@ -171,9 +208,9 @@ mod tests {
     impl Sink<u64> for SumSink {
         type Output = u64;
 
-        fn apply(&mut self, item: u64) -> Result<Apply, Error> {
+        fn apply(&mut self, item: u64) -> Result<Flow, Error> {
             self.total += item;
-            Ok(Apply::Continue)
+            Ok(Flow::Continue)
         }
 
         fn close(self) -> Result<u64, Error> {
@@ -183,7 +220,8 @@ mod tests {
 
     #[test]
     fn happy_path_sums_items() {
-        let pipe = Pipeline::spawn(DEFAULT_DEPTH, SumSink { total: 0 });
+        let pipe = Pipeline::spawn(DEFAULT_PIPELINE_DEPTH, SumSink { total: 0 })
+            .expect("spawn should succeed");
         let mut expected = 0u64;
         for i in 0..100u64 {
             expected += i;
@@ -203,10 +241,10 @@ mod tests {
     impl Sink<()> for SlowSink {
         type Output = usize;
 
-        fn apply(&mut self, _item: ()) -> Result<Apply, Error> {
+        fn apply(&mut self, _item: ()) -> Result<Flow, Error> {
             std::thread::sleep(self.delay);
             self.count.fetch_add(1, Ordering::SeqCst);
-            Ok(Apply::Continue)
+            Ok(Flow::Continue)
         }
 
         fn close(self) -> Result<usize, Error> {
@@ -229,7 +267,7 @@ mod tests {
             delay: Duration::from_millis(50),
             count: count.clone(),
         };
-        let pipe = Pipeline::spawn(2, sink);
+        let pipe = Pipeline::spawn(2, sink).expect("spawn should succeed");
 
         let start = Instant::now();
         for _ in 0..5 {
@@ -257,12 +295,12 @@ mod tests {
     impl Sink<u64> for FailOnNthSink {
         type Output = ();
 
-        fn apply(&mut self, _item: u64) -> Result<Apply, Error> {
+        fn apply(&mut self, _item: u64) -> Result<Flow, Error> {
             let i = self.seen.fetch_add(1, Ordering::SeqCst) + 1;
             if i == self.n {
                 Err(Error::DecryptFailed)
             } else {
-                Ok(Apply::Continue)
+                Ok(Flow::Continue)
             }
         }
 
@@ -277,13 +315,14 @@ mod tests {
         let seen = Arc::new(AtomicUsize::new(0));
         let close_called = Arc::new(AtomicUsize::new(0));
         let pipe = Pipeline::spawn(
-            DEFAULT_DEPTH,
+            DEFAULT_PIPELINE_DEPTH,
             FailOnNthSink {
                 n: 3,
                 seen: seen.clone(),
                 close_called: close_called.clone(),
             },
-        );
+        )
+        .expect("spawn should succeed");
 
         // Send 10 items. Subsequent sends after the 3rd error must
         // still succeed (the consumer is draining).
@@ -304,7 +343,7 @@ mod tests {
         assert_eq!(seen.load(Ordering::SeqCst), 3);
     }
 
-    /// Returns `Apply::Stop` on the Nth apply.
+    /// Returns `Flow::Stop` on the Nth apply.
     struct StopOnNthSink {
         n: usize,
         seen: Arc<AtomicUsize>,
@@ -314,12 +353,12 @@ mod tests {
     impl Sink<u64> for StopOnNthSink {
         type Output = usize;
 
-        fn apply(&mut self, _item: u64) -> Result<Apply, Error> {
+        fn apply(&mut self, _item: u64) -> Result<Flow, Error> {
             let i = self.seen.fetch_add(1, Ordering::SeqCst) + 1;
             if i >= self.n {
-                Ok(Apply::Stop)
+                Ok(Flow::Stop)
             } else {
-                Ok(Apply::Continue)
+                Ok(Flow::Continue)
             }
         }
 
@@ -334,13 +373,14 @@ mod tests {
         let seen = Arc::new(AtomicUsize::new(0));
         let close_called = Arc::new(AtomicUsize::new(0));
         let pipe = Pipeline::spawn(
-            DEFAULT_DEPTH,
+            DEFAULT_PIPELINE_DEPTH,
             StopOnNthSink {
                 n: 3,
                 seen: seen.clone(),
                 close_called: close_called.clone(),
             },
-        );
+        )
+        .expect("spawn should succeed");
 
         // Send 10 items. After Stop, subsequent sends may either
         // succeed (already buffered) or fail with Err(I) (channel
@@ -365,7 +405,7 @@ mod tests {
     impl Sink<u64> for PanickingSink {
         type Output = ();
 
-        fn apply(&mut self, _item: u64) -> Result<Apply, Error> {
+        fn apply(&mut self, _item: u64) -> Result<Flow, Error> {
             panic!("synthetic test panic");
         }
 
@@ -381,7 +421,8 @@ mod tests {
         let prev = std::panic::take_hook();
         std::panic::set_hook(Box::new(|_| {}));
 
-        let pipe = Pipeline::spawn(DEFAULT_DEPTH, PanickingSink);
+        let pipe =
+            Pipeline::spawn(DEFAULT_PIPELINE_DEPTH, PanickingSink).expect("spawn should succeed");
         // First send may succeed (item buffered before panic) or fail
         // (channel closed after panic) — either is fine.
         let _ = pipe.send(1);
@@ -397,9 +438,18 @@ mod tests {
         match res {
             Err(Error::IoError { source }) => {
                 let msg = source.to_string();
+                // Constant prefix lets callers match without parsing
+                // the variable payload tail.
                 assert!(
-                    msg.contains("panicked"),
-                    "expected panic message, got: {msg}"
+                    msg.contains("pipeline consumer panicked"),
+                    "expected constant panic prefix, got: {msg}"
+                );
+                // The original `panic!` payload (a `&'static str`) must
+                // be preserved — without the downcast the message
+                // would just be the prefix.
+                assert!(
+                    msg.contains("synthetic test panic"),
+                    "expected original panic payload, got: {msg}"
                 );
             }
             other => panic!("expected Err(IoError), got {other:?}"),

src/io/writeback/linux.rs

@@ -19,6 +19,11 @@ use std::fs::File;
 use std::os::unix::io::{AsRawFd, RawFd};
 
 pub(crate) struct WritebackPipeline {
+    /// Aliases the wrapping `WritebackFile::file`. Only valid for the
+    /// lifetime of that struct — moving the `File` independently
+    /// would silently UAF this fd. The pipeline is a private field of
+    /// `WritebackFile` and never exposed outside that wrapper, which
+    /// is what keeps the alias sound.
     fd: RawFd,
     chunk_bytes: u64,
     last_flush_pos: u64,
@@ -26,6 +31,10 @@ pub(crate) struct WritebackPipeline {
 }
 
 impl WritebackPipeline {
+    /// Construct a pipeline aliasing `file`'s file descriptor. The
+    /// returned `WritebackPipeline` MUST be dropped before `file`
+    /// itself, or kept inside the same struct that owns `file` — the
+    /// alias is unchecked.
     pub(crate) fn new(file: &File, start_pos: u64, chunk_bytes: u64) -> Self {
         Self {
             fd: file.as_raw_fd(),

src/io/writeback_file.rs

@@ -110,3 +110,20 @@ impl Seek for WritebackFile {
         Ok(p)
     }
 }
+
+impl Drop for WritebackFile {
+    fn drop(&mut self) {
+        // Run the pipeline's tail finalize so the last in-flight chunk
+        // gets its `WAIT_AFTER` + `posix_fadvise(DONTNEED)`. Without
+        // this, callers that drop a `WritebackFile` without calling
+        // `sync_all` (panic, early-return, idiomatic `let _ = w;`)
+        // leave the trailing chunk in cache; the kernel still flushes
+        // on close, but the bounded-cache invariant fails at the tail.
+        // We deliberately do *not* call `self.file.sync_all()` here —
+        // close already triggers a flush, and an `fsync` from `Drop`
+        // would silently swallow its `io::Error` anyway. `finalize` is
+        // idempotent so an explicit `sync_all` followed by drop is
+        // still safe.
+        self.pipeline.finalize();
+    }
+}

src/lib.rs

@@ -147,6 +147,10 @@ pub use decrypt::{DecryptKeys, decrypt_sectors};
 // for displaying disc name + format quickly while a full scan runs in the
 // background. The codec / channel / resolution enums are the canonical
 // structured representation; never compare against display strings.
+// Note: `disc::Stream` here is the codec enum (audio / video / sub kind)
+// — not the `pes::Stream` trait re-exported below as `PesStream`. Two
+// different concepts, the same short name; both stay because both are
+// load-bearing in their respective domains.
 pub use disc::{
     AacsState, AudioChannels, AudioStream, Clip, Codec, ColorSpace, ContentFormat, DamageSeverity,
     Disc, DiscFormat, DiscId, DiscTitle, Extent, FrameRate, HdrFormat, KeySource, LabelPurpose,
@@ -171,7 +175,11 @@ pub use disc::{
 // that need to wire custom readers (e.g. autorip's drive-session reuse).
 // 0.18 trait split: `FrameSource` (read-only) and `FrameSink` (write-only)
 // supersede the unified `pes::Stream`. The old `Stream` re-export below
-// stays available for the deprecation window.
+// stays available for the deprecation window — re-exported as
+// `PesStream` to disambiguate from `disc::Stream` (the codec-kind enum
+// re-exported above), which would otherwise collide at the crate root.
+#[allow(deprecated)]
+pub use pes::Stream as PesStream;
 pub use pes::{FrameSink, FrameSource, PesFrame};
 
 pub use mux::DiscStream;

src/pes.rs

@@ -78,47 +78,19 @@ impl PesFrame {
     }
 }
 
-/// A PES frame source or sink. Each implementor is **either** read-only or
-/// write-only — never both.
-///
-/// Implementors fall into two camps:
-///
-/// - **Read sources**: `DiscStream` (drive or ISO), `M2tsStream` (when
-///   constructed from an existing file), `MkvStream` (demux), `NetworkStream`
-///   (TCP listener), `StdioStream::input()`. These return frames from
-///   `read()` and surface `StreamWriteOnly` (E9001) from `write()`.
-/// - **Write sinks**: `MkvStream::create`, `M2tsStream::create`,
-///   `NetworkStream::connect`, `StdioStream::output()`, `NullStream`.
-///   These accept frames in `write()` and surface `StreamReadOnly` (E9000)
-///   from `read()`. Always call `finish()` when done — that's where MKV
-///   writes its `Cues` index and `M2tsStream` flushes the TS muxer.
-///
-/// Direction is established at construction; mixing produces an error code,
-/// not a panic. Most consumers don't construct streams directly — call
-/// `mux::input(url, opts)` / `mux::output(url, title)` and let URL parsing
-/// pick the right type.
-///
-/// `info()` returns the stream's `DiscTitle` metadata (track list, codec
-/// info, duration). For sources it's parsed from the input; for sinks it's
-/// the metadata supplied at creation. Stable across all reads.
-///
-/// `codec_private(track)` exposes per-track initialization data
-/// (H.264 SPS/PPS, HEVC VPS/SPS/PPS, AC-3 fscod, etc.) that some output
-/// formats need before any frame can be written. `headers_ready()` returns
-/// false until enough input frames have been seen to populate every video
-/// track's codec-private blob — callers buffer frames they read until
-/// `headers_ready()` returns true.
+/// Deprecated; use [`FrameSource`] for read-only sources or [`FrameSink`]
+/// for write-only sinks. The runtime direction-error semantics
+/// (`StreamReadOnly` / `StreamWriteOnly` from a wrong-direction call) are
+/// removed in 0.18 — direction is type-checked.
 #[deprecated(
-    since = "0.18.0-dev",
+    since = "0.18.0",
     note = "use FrameSource (read-only) or FrameSink (write-only) instead"
 )]
 pub trait Stream {
-    /// Read the next frame, or `Ok(None)` at end of stream. Returns
-    /// `StreamWriteOnly` (E9001) on a write-only sink.
+    /// Read the next frame, or `Ok(None)` at end of stream.
     fn read(&mut self) -> std::io::Result<Option<PesFrame>>;
 
-    /// Write a frame to the sink. Returns `StreamReadOnly` (E9000) on a
-    /// read-only source.
+    /// Write a frame to the sink.
     fn write(&mut self, frame: &PesFrame) -> std::io::Result<()>;
 
     /// Finalize the stream: flush buffered frames, write any container
@@ -207,10 +179,25 @@ pub trait FrameSink: Send {
     fn info(&self) -> &crate::disc::DiscTitle;
 }
 
-// Bridge: any type implementing the deprecated `Stream` trait is also a
-// `FrameSource`. This lets existing concrete `Stream` impls in `mux/*`
-// satisfy `FrameSource` bounds without per-type migration during the
-// 0.18 deprecation window.
+// Bridge: any **`Send`** type implementing the deprecated `Stream` trait
+// is also a `FrameSource`. This lets existing concrete `Stream` impls in
+// `mux/*` satisfy `FrameSource` bounds without per-type migration during
+// the 0.18 deprecation window.
+//
+// **Send caveat (read me before tightening `Stream` itself).** This
+// blanket carries a `T: Send` bound rather than promoting `Send` to a
+// supertrait of `Stream`, because not every concrete in-tree `Stream`
+// impl is `Send`: `MkvStream` and `M2tsStream` carry `Box<dyn Read>`
+// and `Box<dyn Write>` fields whose trait objects don't include `Send`.
+// Adding `Stream: Send` would force a wider audit (every `Box<dyn Read>`
+// becomes `Box<dyn Read + Send>`) than this commit is taking on, and
+// the type-level migration target is `FrameSource` / `FrameSink`
+// directly anyway. Consequence: coercing a non-Send `Box<dyn Stream>`
+// (the return shape of `crate::mux::input` / `output`) to
+// `Box<dyn FrameSource>` will fail with a `T: Send` trait-bound error.
+// The fix on the consumer side is to construct a Send-compliant
+// `FrameSource` / `FrameSink` directly rather than relying on this
+// bridge for non-Send streams.
 //
 // Note: `FrameSink` cannot be blanket-impl'd from `Stream` because
 // `Stream::finish` takes `&mut self` while `FrameSink::finish` takes

src/sector/file.rs

@@ -8,7 +8,7 @@
 //! mux.
 
 use std::fs::{File, OpenOptions};
-use std::io::{BufReader, Read, Seek, SeekFrom, Write};
+use std::io::{Read, Seek, SeekFrom, Write};
 use std::path::Path;
 
 use crate::error::{Error, Result};
@@ -18,10 +18,14 @@ use super::{SectorReader, SectorSink};
 /// SectorSource backed by a file (ISO image).
 ///
 /// Seeks to `lba * 2048`, reads `count * 2048` bytes per call. The
-/// underlying file is wrapped in a 4 MiB `BufReader` so adjacent
-/// small reads coalesce into single syscalls.
+/// file is held directly: every `read_sectors` call performs an
+/// absolute seek, so a wrapping `BufReader` would have its buffer
+/// invalidated on every call (its internal cursor moves with the
+/// `Seek` impl) — pure overhead. Callers that benefit from buffered
+/// reads should compose their own `BufReader` at the `read_sectors`
+/// granularity they care about.
 pub struct FileSectorSource {
-    file: BufReader<File>,
+    file: File,
     capacity: u32,
 }
 
@@ -30,21 +34,18 @@ impl FileSectorSource {
     /// from `metadata().len() / 2048`. Returns
     /// [`Error::IsoTooLarge`] if the file would exceed the 32-bit
     /// LBA address space (~8 TB).
-    pub fn open(path: &str) -> std::io::Result<Self> {
+    pub fn open(path: &Path) -> std::io::Result<Self> {
         let file = File::open(path)?;
         let len = file.metadata()?.len();
         let sectors = len / 2048;
         if sectors > u32::MAX as u64 {
             return Err(Error::IsoTooLarge {
-                path: path.to_string(),
+                path: path.to_string_lossy().into_owned(),
             }
             .into());
         }
         let capacity = sectors as u32;
-        Ok(Self {
-            file: BufReader::with_capacity(4 * 1024 * 1024, file),
-            capacity,
-        })
+        Ok(Self { file, capacity })
     }
 }
 
@@ -173,7 +174,7 @@ mod tests {
         sink.write_sectors(2, &payload).unwrap();
         Box::new(sink).finish().unwrap();
 
-        let mut src = FileSectorSource::open(path.to_str().unwrap()).unwrap();
+        let mut src = FileSectorSource::open(&path).unwrap();
         assert_eq!(src.capacity_sectors(), 4);
 
         let mut got = [0u8; 2048];
@@ -200,7 +201,7 @@ mod tests {
         sink.write_sectors(0, &payload).unwrap();
         Box::new(sink).finish().unwrap();
 
-        let mut src = FileSectorSource::open(path.to_str().unwrap()).unwrap();
+        let mut src = FileSectorSource::open(&path).unwrap();
         assert_eq!(src.capacity_sectors(), 8);
 
         let mut got = vec![0u8; 8 * 2048];
@@ -226,7 +227,7 @@ mod tests {
         sink.write_sectors(1, &pat_b).unwrap();
         Box::new(sink).finish().unwrap();
 
-        let mut src = FileSectorSource::open(path.to_str().unwrap()).unwrap();
+        let mut src = FileSectorSource::open(&path).unwrap();
         assert_eq!(src.capacity_sectors(), 4);
         let mut got = [0u8; 2048];
 

tests/integration_progress_and_halt.rs

@@ -324,7 +324,7 @@ fn test_file_sector_reader_round_trip() {
     tmp.write_all(&data).expect("write data");
     tmp.flush().expect("flush");
 
-    let path = tmp.path().to_str().expect("path utf-8").to_string();
+    let path = tmp.path().to_path_buf();
     let mut fsr = FileSectorReader::open(&path).expect("open FileSectorReader");
 
     assert_eq!(fsr.capacity(), N_SECTORS as u32, "capacity mismatch");