From ec5b10f83a222bb3ee62dd0a1b0f0a64300a5c2b Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Fri, 24 Jul 2026 01:01:34 -0700 Subject: [PATCH] mux/driver: split on_progress; stage live drive as session reader MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Split the ambiguous MuxEvents::on_progress(bytes, total) into two callbacks — on_read_progress(bytes_read, total) and on_write_progress(bytes_written, total) — so a consumer no longer has to guess which side of the pipeline a progress figure came from. The CLI drives its bar from the write side; autorip from the read side. The reader EventFn's BytesRead now maps to on_read_progress; the per-frame emit in drive_mux to on_write_progress. Both keep empty defaults; NoopEvents and the driver tests are updated to match. Add DiscSession::stage_drive_as_reader so the live single-pass path can run through MuxInput::Session: the owned Drive (itself a SectorSource) moves into the reader slot for mux_stream to take. The drive is now held as Option with the device path cached up front, so the mux driver can still name the device after the drive has been staged; the driver's Session arm uses the new device_path() accessor. --- src/mux/driver.rs | 47 ++++++++++++++++++++------------- src/session.rs | 66 +++++++++++++++++++++++++++++++++++++---------- 2 files changed, 81 insertions(+), 32 deletions(-) diff --git a/src/mux/driver.rs b/src/mux/driver.rs index d7ab8cc..438dd03 100644 --- a/src/mux/driver.rs +++ b/src/mux/driver.rs @@ -108,17 +108,27 @@ pub struct MuxOptions { /// [`Arc`] and clones it into the constructors' `'static` /// [`EventFn`] (via [`reader_event_fn`]) so the reader-side events fire from the /// highway's producer thread / the live `DiscStream`'s read loop. The driver -/// fires [`Self::on_output_opened`] and the write-side [`Self::on_progress`] +/// fires [`Self::on_output_opened`] and the write-side [`Self::on_write_progress`] /// from the driving thread; the reader-side events /// ([`Self::on_sector_skipped`] / [`Self::on_batch_size_changed`] / -/// [`Self::on_read_error`], plus a read-side [`Self::on_progress`]) are fired +/// [`Self::on_read_error`], plus [`Self::on_read_progress`]) are fired /// by that cloned `EventFn`. +/// +/// Progress is split into two callbacks because consumers drive their progress +/// UI from different sides of the pipeline: the CLI renders from the WRITE side +/// (bytes finalised to the sink), autorip from the READ side (bytes pulled off +/// the disc). Keeping them distinct avoids the old ambiguous single +/// `on_progress(bytes, total)` where the caller couldn't tell which number it +/// was handed. pub trait MuxEvents: Send + Sync + 'static { /// Fired once, immediately after the output sink is created. fn on_output_opened(&self, _title: &DiscTitle) {} + /// Fired periodically from the reader side with the running read-byte count + /// and the source extents' total byte estimate. + fn on_read_progress(&self, _bytes_read: u64, _bytes_total: u64) {} /// Fired periodically during the frame pump with the running written-byte /// count and the title's total byte estimate. - fn on_progress(&self, _bytes_written: u64, _bytes_total: u64) {} + fn on_write_progress(&self, _bytes_written: u64, _bytes_total: u64) {} /// A bad sector was skipped (zero-filled) at `lba`. fn on_sector_skipped(&self, _lba: u32) {} /// The adaptive read batch size changed. @@ -216,7 +226,7 @@ pub fn mux_stream( // `take_reader` below. let (title, format, keys, playlist) = { let disc = session.disc().ok_or_else(|| Error::DeviceNotReady { - path: session.drive().device_path().to_string(), + path: session.device_path().to_string(), })?; let title = disc .titles @@ -235,7 +245,7 @@ pub fn mux_stream( // A missing staged reader ("already consumed" / never staged) is a // clean error, not a panic (contract Q2). let reader = session.take_reader().ok_or_else(|| Error::DeviceNotReady { - path: session.drive().device_path().to_string(), + path: session.device_path().to_string(), })?; let mut stream = crate::mux::DiscStream::new( reader, @@ -276,8 +286,8 @@ pub fn mux_stream( /// borrow cannot satisfy the `'static` bound. /// /// Mapping (real [`EventKind`] variants): -/// - `BytesRead { bytes, total }` → [`MuxEvents::on_progress`] (read-side; the -/// file highway's only reader event — `total` is the extents' byte total) +/// - `BytesRead { bytes, total }` → [`MuxEvents::on_read_progress`] (read-side; +/// the file highway's only reader event — `total` is the extents' byte total) /// - `SectorSkipped { sector }` → [`MuxEvents::on_sector_skipped`] (live only) /// - `BatchSizeChanged { new_size, reason }` → [`MuxEvents::on_batch_size_changed`] /// (live only) @@ -287,7 +297,7 @@ pub fn mux_stream( /// (the disc's LBA space), so they are narrowed with `as u32`. fn reader_event_fn(events: Arc) -> crate::sector::prefetched::EventFn { Box::new(move |e: Event| match e.kind { - EventKind::BytesRead { bytes, total } => events.on_progress(bytes, total), + EventKind::BytesRead { bytes, total } => events.on_read_progress(bytes, total), EventKind::SectorSkipped { sector } => events.on_sector_skipped(sector as u32), EventKind::BatchSizeChanged { new_size, reason } => { events.on_batch_size_changed(new_size, reason) @@ -387,7 +397,7 @@ fn drive_mux( // The write consumer runs on its own thread so the latency-bound sink write // overlaps the next `stream.read()`. `bytes` mirrors the consumer's running - // written-byte count out to the driving thread for `on_progress`. + // written-byte count out to the driving thread for `on_write_progress`. let bytes = Arc::new(AtomicU64::new(0)); let sink = WriteSink { output: output_stream, @@ -421,7 +431,7 @@ fn drive_mux( interrupted = true; break; } - events.on_progress(bytes.load(Ordering::Relaxed), total_bytes); + events.on_write_progress(bytes.load(Ordering::Relaxed), total_bytes); } Ok(None) => break, Err(e) => { @@ -717,10 +727,11 @@ mod tests { struct CountingEvents { opened: AtomicBool, progress_calls: AtomicU64, - /// Set once `on_progress` is called with a `total` equal to the ISO + /// Set once `on_read_progress` is called with a `total` equal to the ISO /// extents' byte total — the fingerprint of the *read-side* `BytesRead` - /// event (the write-side `on_progress` carries the title's `size_bytes`, - /// a different number), so it isolates the `EventFn` translation. + /// event (the write-side `on_write_progress` carries the title's + /// `size_bytes`, a different number), so it isolates the `EventFn` + /// translation. saw_read_total: AtomicBool, read_total: u64, skipped: AtomicU64, @@ -744,7 +755,7 @@ mod tests { fn on_output_opened(&self, _title: &DiscTitle) { self.opened.store(true, Ordering::SeqCst); } - fn on_progress(&self, _bytes_written: u64, bytes_total: u64) { + fn on_read_progress(&self, _bytes_read: u64, bytes_total: u64) { self.progress_calls.fetch_add(1, Ordering::SeqCst); if bytes_total == self.read_total { self.saw_read_total.store(true, Ordering::SeqCst); @@ -793,7 +804,7 @@ mod tests { assert_eq!( events.progress_calls.load(Ordering::SeqCst), 1, - "BytesRead → on_progress" + "BytesRead → on_read_progress" ); assert!( events.saw_read_total.load(Ordering::SeqCst), @@ -866,8 +877,8 @@ mod tests { /// /// Mutation: passing `None` (instead of `Some(reader_event_fn(...))`) to /// `build_iso_pipeline` in the ISO arm leaves `saw_read_total` false — the - /// write-side `on_progress` carries `size_bytes` (0 here), never the 6144 - /// extents total — and this test fails. + /// write-side `on_write_progress` carries `size_bytes` (0 here), never the + /// 6144 extents total — and this test fails. #[test] fn mux_stream_iso_forwards_reader_progress_through_arc() { let es = [0xDE, 0xAD, 0xBE, 0xEF, 0x11, 0x22]; @@ -919,7 +930,7 @@ mod tests { ); assert!( events.progress_calls.load(Ordering::SeqCst) > 0, - "at least one on_progress call observed" + "at least one on_read_progress call observed" ); } } diff --git a/src/session.rs b/src/session.rs index 16f0517..e5ea9c2 100644 --- a/src/session.rs +++ b/src/session.rs @@ -152,11 +152,19 @@ pub struct KeySpec { /// `DiscStream`) reach it via [`Self::drive_mut`] / [`Self::into_drive`]; the /// scanned [`Disc`] comes out via [`Self::disc`] / [`Self::take_disc`]. pub struct DiscSession { - drive: Drive, + /// The opened drive. `Some` from [`Self::open`] until + /// [`Self::stage_drive_as_reader`] (live-drive mux) or [`Self::into_drive`] + /// moves it out. The cached [`Self::device_path`] survives that move so the + /// mux driver can still name the device in an error without the drive. + drive: Option, + /// The drive's device path, cached at [`Self::open`] so it outlives a + /// [`Self::stage_drive_as_reader`] that moves the drive into `reader`. + device: String, spec: KeySpec, disc: Option, - /// Sector source for a later file/live mux to `.take()` (steps 3–4). - /// Unpopulated in the current step; shapes the struct for the mux hoist. + /// Sector source for a later file/live mux to `.take()` (steps 3–4). The + /// file path stages a `FileSectorSource`; the live-drive path stages the + /// drive itself via [`Self::stage_drive_as_reader`]. reader: Option>, /// The read-time AACS fetch closure, built by [`Self::resolve_keys`] and /// retained so a later mux (step 4) can install it into the decrypt @@ -214,8 +222,10 @@ impl DiscSession { tracing::warn!(target: "freemkv::session", error = %e, "probe_disc advisory failed (continuing)"); } + let device = drive.device_path().to_string(); Ok(DiscSession { - drive, + drive: Some(drive), + device, spec, disc: None, reader: None, @@ -226,7 +236,7 @@ impl DiscSession { /// Fast disc identification — name/format only, no playlist parse. Wraps /// [`Disc::identify`]. pub fn identify(&mut self) -> Result { - Disc::identify(&mut self.drive) + Disc::identify(self.drive_mut()) } /// Full structure scan. Forwards the session's [`KeySpec`] credentials / @@ -234,7 +244,7 @@ impl DiscSession { /// set), runs [`Disc::scan`], stores the result, and returns a borrow. pub fn scan(&mut self, opts: ScanOptions) -> Result<&Disc> { let opts = forward_key_material(&mut self.spec, opts); - let disc = Disc::scan(&mut self.drive, &opts)?; + let disc = Disc::scan(self.drive.as_mut().expect("drive present for scan"), &opts)?; self.disc = Some(disc); Ok(self.disc.as_ref().expect("disc just stored")) } @@ -252,7 +262,7 @@ impl DiscSession { // The disc must have been scanned so its AACS inputs are captured. if self.disc.is_none() { return Err(Error::DeviceNotReady { - path: self.drive.device_path().to_string(), + path: self.device.clone(), }); } // Sample through the staged reader when present (file-backed), else the @@ -263,7 +273,11 @@ impl DiscSession { resolve_keys_for(reader.as_mut(), disc, sources) } else { let disc = self.disc.as_mut().expect("disc present (checked above)"); - resolve_keys_for(&mut self.drive, disc, sources) + resolve_keys_for( + self.drive.as_mut().expect("drive present for key sampling"), + disc, + sources, + ) }; self.key_fetch = resolved.key_fetch; Ok(resolved.trace) @@ -293,27 +307,51 @@ impl DiscSession { self.disc.take() } - /// Shared access to the opened drive (identity, profile, path). + /// Shared access to the opened drive (identity, profile, path). Panics if the + /// drive has already been staged into the reader slot + /// ([`Self::stage_drive_as_reader`]) or moved out via [`Self::into_drive`] — + /// use [`Self::device_path`] for a name that survives those moves. pub fn drive(&self) -> &Drive { - &self.drive + self.drive.as_ref().expect("drive present") + } + + /// The opened drive's device path. Cached at [`Self::open`], so it remains + /// available after [`Self::stage_drive_as_reader`] moves the drive into the + /// reader slot (the mux driver names the device here without the drive). + pub fn device_path(&self) -> &str { + &self.device } /// Mutable access to the opened drive — for ciphertext sampling and other /// direct reads consumers still perform. pub fn drive_mut(&mut self) -> &mut Drive { - &mut self.drive + self.drive.as_mut().expect("drive present") } /// Lock the tray so the disc cannot eject mid-rip. Unlock is guaranteed by - /// `Drive::drop`. + /// `Drive::drop`. A no-op if the drive is no longer held by the session. pub fn lock_tray(&mut self) { - self.drive.lock_tray(); + if let Some(drive) = self.drive.as_mut() { + drive.lock_tray(); + } } /// Consume the session, returning the owned drive (e.g. to move into a /// `DiscStream` for a live-drive mux). pub fn into_drive(self) -> Drive { - self.drive + self.drive.expect("drive present") + } + + /// Stage the owned drive as the session's boxed sector source so a live + /// single-pass mux can drive it through + /// [`MuxInput::Session`](crate::mux::MuxInput::Session). Moves the `Drive` + /// (itself a [`SectorSource`]) into the `reader` slot; the cached + /// [`Self::device_path`] keeps the device name available afterward. A no-op + /// if the drive was already staged or moved out. + pub fn stage_drive_as_reader(&mut self) { + if let Some(drive) = self.drive.take() { + self.reader = Some(Box::new(drive)); + } } /// Consume the session, returning the sector source staged for a later mux