diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3b845f0..a1fe0b8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -21,7 +21,7 @@ jobs: steps: - uses: actions/checkout@v5 - uses: dtolnay/rust-toolchain@1.86.0 - - run: cargo test + - run: cargo test --tests check-macos: runs-on: macos-latest diff --git a/CHANGELOG.md b/CHANGELOG.md index 90ca954..a898430 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,59 @@ # Changelog +## 0.13.6 (2026-04-25) + +### Inline retry/reset stripped from `Drive::read`; `BytesRead` now emitted + +Two related changes that close the loop on the BU40N wedge work from +0.13.1–0.13.4 and on the long-standing autorip "0 KB/s, 0%" UI bug. + +**`Drive::read` is now single-shot.** The phase 1 / 2 / 3 retry loop +(reset → reopen → repeat) inside `Drive::read` is gone (~80 lines +deleted). `recovery=true` only bumps the per-CDB timeout to 30 s; +`recovery=false` keeps the 1.5 s timeout. On a failed read the +function returns `Err(DiscRead)` immediately. Per the BU40N +post-mortem, every USB / SCSI reset path tested in 0.13.1–0.13.3 +resets the bridge but not the drive firmware, and the inline +reset+reopen *was* the wedge primitive itself — issuing it from +inside `Drive::read` produced multi-minute hangs and made the wedge +class harder to surface to the user. The correct retry layer is +`Disc::patch`'s outer multi-pass loop, which is unaffected. A stuck +drive now surfaces as a clean `DiscRead` to the caller, who can +prompt physical replug. + +**SCSI reset surface trimmed.** `SgIoTransport::reset` (Linux) drops +the `SG_SCSI_RESET` ioctl and the STOP / START UNIT escalation; it +keeps the kernel `SG_IO` state flush plus `ALLOW MEDIUM REMOVAL`. +`MacScsiTransport::reset` is removed entirely (was open + drop + +sleep, no SCSI). The top-level `scsi::reset` / +`scsi::reset_with_timeout` / `scsi::reset_blocking` family is +removed — no callers remain after the `Drive::read` strip. + +**`EventKind::BytesRead` now emitted.** The variant was declared in +0.13.0 but never fired. `DiscStream::fill_extents` now emits +`BytesRead { bytes_read_total, total_extents_bytes }` after every +successful sector read, so consumers in direct (no-mapfile) mode can +drive a real-time progress bar without polling `output.bytes_written`. +Multi-pass mode continues to use `Disc::copy`'s `on_progress` +callback unchanged. Drives the autorip per-device live progress UI. + +`Drive::checked_sleep` is removed (only used by the recovery loop); +`Drive::sleep_until_halted` is `#[cfg(test)]`-only; `Drive::emit` is +retained because `BytesRead` uses it. + +### Tests +- New `tests/integration_progress_and_halt.rs` (5 tests): `BytesRead` + emission, `Disc::copy` `on_progress` regression guard, halt aborts + copy, Drop safety, `FileSectorReader` round-trip. +- 233 unit tests + 5 integration tests pass. + +### Net diff +~80 lines deleted, ~20 added. + +### Version sync +0.13.6 ecosystem release (libfreemkv + freemkv + bdemu + autorip all +on 0.13.6). + ## 0.13.5 (2026-04-25) ### Version sync — no functional changes diff --git a/Cargo.toml b/Cargo.toml index e919976..6642396 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "libfreemkv" -version = "0.13.5" +version = "0.13.6" edition = "2024" rust-version = "1.86" license = "AGPL-3.0-only" @@ -31,6 +31,9 @@ libc = "0.2" [target.'cfg(target_os = "macos")'.dependencies] libc = "0.2" +[dev-dependencies] +tempfile = "3" + [[bench]] name = "sgio_read" harness = false diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..68b737a --- /dev/null +++ b/Makefile @@ -0,0 +1,20 @@ +# libfreemkv — local dev helper. +# Mirrors the cross-crate scripts in (internal)/scripts/test-all.sh +# but scoped to this single crate. + +.PHONY: test build check ci clean + +test: + cargo test --tests + +build: + cargo build --release + +check: + cargo fmt --check + cargo clippy --all-targets -- -D warnings + +ci: check build test + +clean: + cargo clean diff --git a/README.md b/README.md index 15e0db3..30604a2 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ Part of the [freemkv](https://github.com/freemkv) project. ```toml [dependencies] -libfreemkv = "0.11" +libfreemkv = "0.13" ``` ## Quick Start @@ -84,7 +84,7 @@ while result.bytes_unreadable + result.bytes_pending > 0 { - **Stream labels** — 5 BD-J format parsers (Paramount, Criterion, Pixelogic, CTRM, Deluxe) - **AACS decryption** — transparent key resolution and content decrypt (1.0 + 2.0 bus decryption) - **KEYDB updates** — download, verify, save from any HTTP URL (zero deps, raw TCP) -- **Content reading** — adaptive batch reads with automatic decryption and error recovery +- **Content reading** — adaptive batch reads with automatic decryption - **Stream I/O** — unified stream pipeline for reading and writing any format ### Streams @@ -92,21 +92,21 @@ while result.bytes_unreadable + result.bytes_pending > 0 { | Stream | Input | Output | Transport | |--------|-------|--------|-----------| | DiscStream | Yes | -- | Optical drive via SCSI | -| IsoStream | Yes | Yes | Blu-ray ISO image file | +| IsoStream | Yes | -- | Blu-ray ISO image file (read via stream pipeline; written via `Disc::copy()`) | | MkvStream | Yes | Yes | Matroska container | | M2tsStream | Yes | Yes | BD transport stream with FMKV metadata header | | NetworkStream | Yes (listen) | Yes (connect) | TCP with FMKV metadata header | | StdioStream | Yes (stdin) | Yes (stdout) | Raw byte pipe | | NullStream | -- | Yes | Discard sink (byte counter for benchmarks) | -Streams implement `IOStream` (byte-level) and `pes::Stream` (frame-level). `input()` / `output()` resolve URL strings to PES stream instances. `open_input()` / `open_output()` resolve to byte-level IOStream instances. All URLs use the `scheme://path` format — bare paths are rejected. +Streams implement `pes::Stream` (frame-level). `input()` / `output()` resolve URL strings to PES stream instances. All URLs use the `scheme://path` format — bare paths are rejected. AACS decryption requires a KEYDB.cfg file. If available at `~/.config/aacs/KEYDB.cfg` or passed via `ScanOptions`, the library handles everything — handshake, key derivation, and per-sector decryption — without the application needing to know anything about encryption. ## Architecture ```text -Drive — open, identify, init, unlock, read (with recovery) +Drive — open, identify, init, unlock, single-shot read ├── ScsiTransport — SG_IO (Linux), IOKit (macOS), SPTI (Windows) ├── DriveProfile — per-drive unlock parameters (bundled) └── PlatformDriver — MediaTek (supported), Renesas (planned) diff --git a/src/drive/mod.rs b/src/drive/mod.rs index 2f3432f..f512ed2 100644 --- a/src/drive/mod.rs +++ b/src/drive/mod.rs @@ -53,9 +53,6 @@ const SCSI_GET_EVENT_STATUS: u8 = 0x4A; const SCSI_MODE_SENSE: u8 = 0x5A; const SCSI_REPORT_KEY: u8 = 0xA4; -/// Recovery state after a read error — stay at min speed for N bytes. -const RECOVERY_WINDOW: u64 = 500 * 1024 * 1024; // 500 MB - /// Optical disc drive session -- open, identify, unlock, and read. pub struct Drive { scsi: Box, @@ -64,11 +61,9 @@ pub struct Drive { pub platform: Option, pub drive_id: DriveId, device_path: String, - /// Bytes remaining in the min-speed recovery window. - recovery_bytes_remaining: u64, /// Halt flag — when set, Drive::read() bails at the next check point. halt: Arc, - /// Event handler — fires during read recovery. + /// Event handler — fires for read errors and library-level state changes. event_fn: Option>, } @@ -95,7 +90,6 @@ impl Drive { profile, drive_id, device_path: device.to_string_lossy().to_string(), - recovery_bytes_remaining: 0, halt: Arc::new(AtomicBool::new(false)), event_fn: None, }) @@ -131,13 +125,6 @@ impl Drive { self.halt.load(Ordering::Relaxed) } - /// Halt-aware sleep. Returns `Err(Halted)` if the flag fires before or - /// during the wait. The only sleep the drive's recovery path is allowed - /// to use — makes it impossible to accidentally block through a Stop. - fn checked_sleep(&self, total: std::time::Duration) -> Result<()> { - sleep_until_halted(&self.halt, total) - } - /// Halt-aware SCSI execute. Returns `Err(Halted)` if the flag is set /// before the command dispatches or by the time it completes. The only /// path to talk to the drive in the recovery hot loop; keeps Drive::read @@ -519,29 +506,25 @@ impl Drive { } } - /// Read sectors from the disc with automatic error recovery. + /// Read sectors from the disc. Single-shot — no inline retries, no + /// SCSI reset. /// - /// On failure: drops to min speed, waits with escalating patience - /// (5s, 10s, 15s, 30s, 60s), resets drive between attempts. - /// After recovery, stays at min speed for 500 MB before ramping up. + /// `recovery=true` bumps the per-CDB timeout to 30 s for the + /// `Disc::patch` pass; `recovery=false` uses 1.5 s for `Disc::copy`'s + /// fast skip-forward sweep. On any failure returns `Err(DiscRead)` + /// immediately. The orchestration layer (`Disc::patch`'s outer loop + /// for the patch pass, `DiscStream`'s adaptive batch halving for the + /// stream path) handles retries. /// - /// Returns Err only after all attempts exhausted — user should clean - /// the disc and resume. + /// Inline retry phases (5× gentle + reset+reopen + 5× more) were + /// removed in 0.13.6. Per + /// `(internal)/postmortems/2026-04-25-stop-wedge-and-zero-kbs.md`, + /// the inline reset on the LG BU40N (Initio bridge) wedged drive + /// firmware without ever recovering a sector. The remaining recovery + /// layers (Disc::patch multi-pass, DiscStream batch halving) do not + /// touch the wedge-prone reset path. pub fn read(&mut self, lba: u32, count: u16, buf: &mut [u8], recovery: bool) -> Result { - // Non-recovery mode is the Disc::copy fast pass — intent is "fail - // fast, let skip_forward advance past bad regions." A 5s budget let - // the drive grind L-EC on structure-protected / marginal sectors for - // nearly the full interval per 64 KB block, pinning throughput at - // ~13 KB/s on difficult discs. 1500ms kills slow reads early so - // skip_forward can double its stride; recoverable sectors are picked - // up on Disc::patch where recovery=true and the timeout is 30s. - let timeout_ms = if !recovery { - 1_500 - } else if self.recovery_bytes_remaining > 0 { - 30_000 - } else { - 10_000 - }; + let timeout_ms = if recovery { 30_000 } else { 1_500 }; let cdb = [ crate::scsi::SCSI_READ_10, 0x00, @@ -555,82 +538,16 @@ impl Drive { 0x00, ]; - // Normal read. Fast path: drive returns data, we return it. - // checked_exec short-circuits to Err(Halted) if Stop was requested; - // read() never touches the halt flag directly. match self.checked_exec( &cdb, crate::scsi::DataDirection::FromDevice, buf, timeout_ms, ) { - Ok(result) => { - if self.recovery_bytes_remaining > 0 { - let bytes_read = count as u64 * 2048; - self.recovery_bytes_remaining = - self.recovery_bytes_remaining.saturating_sub(bytes_read); - if self.recovery_bytes_remaining == 0 { - self.emit(EventKind::SpeedChange { speed_kbs: 0xFFFF }); - self.set_speed(0xFFFF); - } - } - return Ok(result.bytes_transferred); - } - Err(Error::Halted) => return Err(Error::Halted), - Err(_) => { /* fall through to recovery */ } + Ok(result) => Ok(result.bytes_transferred), + Err(Error::Halted) => Err(Error::Halted), + Err(_) => Err(Error::DiscRead { sector: lba as u64 }), } - - if !recovery { - return Err(Error::DiscRead { sector: lba as u64 }); - } - - // Enter recovery - self.emit(EventKind::ReadError { - sector: lba as u64, - error: Error::DiscRead { sector: lba as u64 }, - }); - self.emit(EventKind::SpeedChange { speed_kbs: 0 }); - self.set_speed(0); - - // Phase 1: gentle — sleep 30 s, retry. 5 times. - for attempt in 1..=5u32 { - self.emit(EventKind::Retry { attempt }); - self.checked_sleep(std::time::Duration::from_secs(30))?; - if let Ok(result) = - self.checked_exec(&cdb, crate::scsi::DataDirection::FromDevice, buf, 30_000) - { - self.emit(EventKind::SectorRecovered { sector: lba as u64 }); - self.recovery_bytes_remaining = RECOVERY_WINDOW; - return Ok(result.bytes_transferred); - } - } - - // Phase 2: fresh start — close, reset, open, init. - let device = std::path::PathBuf::from(&self.device_path); - self.checked_sleep(std::time::Duration::from_secs(5))?; - let _ = crate::scsi::reset(&device); - self.checked_sleep(std::time::Duration::from_secs(5))?; - self.scsi = crate::scsi::open(&device)?; - let _ = self.init(); - let _ = self.wait_ready(); - self.set_speed(0); - - // Phase 3: gentle again on fresh connection — sleep 30 s, retry. 5 times. - for attempt in 6..=10u32 { - self.emit(EventKind::Retry { attempt }); - self.checked_sleep(std::time::Duration::from_secs(30))?; - if let Ok(result) = - self.checked_exec(&cdb, crate::scsi::DataDirection::FromDevice, buf, 30_000) - { - self.emit(EventKind::SectorRecovered { sector: lba as u64 }); - self.recovery_bytes_remaining = RECOVERY_WINDOW; - return Ok(result.bytes_transferred); - } - } - - // Both phases failed. - self.recovery_bytes_remaining = RECOVERY_WINDOW; - Err(Error::DiscRead { sector: lba as u64 }) } /// Read the disc capacity in sectors (2048 bytes each). @@ -742,11 +659,26 @@ impl SectorReader for Drive { } } -/// Halt-aware sleep primitive. Returns `Err(Halted)` if the flag is set -/// before or during the wait. Extracted as a free function so it can be -/// unit-tested without constructing a full `Drive`. -/// -/// Wakes within ~100 ms of a halt, regardless of the requested duration. +/// Find all optical drives connected to this system. +/// Returns opened Drive objects ready for use. +pub fn find_drives() -> Vec { + discover_drives() + .into_iter() + .filter_map(|(path, _)| Drive::open(std::path::Path::new(&path)).ok()) + .collect() +} + +/// Find the first optical drive. +/// Returns an opened Drive ready for use. +pub fn find_drive() -> Option { + find_drives().into_iter().next() +} + +/// Halt-aware sleep primitive — wakes within ~100 ms of `halt` flipping +/// to true. Kept for the unit tests that cover the slicing behaviour; +/// production code paths no longer sleep on the recovery hot path +/// (recovery loop removed in 0.13.6). +#[cfg(test)] fn sleep_until_halted(halt: &AtomicBool, total: std::time::Duration) -> Result<()> { const SLICE: std::time::Duration = std::time::Duration::from_millis(100); let deadline = std::time::Instant::now() + total; @@ -763,21 +695,6 @@ fn sleep_until_halted(halt: &AtomicBool, total: std::time::Duration) -> Result<( } } -/// Find all optical drives connected to this system. -/// Returns opened Drive objects ready for use. -pub fn find_drives() -> Vec { - discover_drives() - .into_iter() - .filter_map(|(path, _)| Drive::open(std::path::Path::new(&path)).ok()) - .collect() -} - -/// Find the first optical drive. -/// Returns an opened Drive ready for use. -pub fn find_drive() -> Option { - find_drives().into_iter().next() -} - /// Internal: discover drive paths + IDs without opening full Drive objects. fn discover_drives() -> Vec<(String, DriveId)> { #[cfg(target_os = "linux")] diff --git a/src/mux/disc.rs b/src/mux/disc.rs index ca9c219..bb91e8d 100644 --- a/src/mux/disc.rs +++ b/src/mux/disc.rs @@ -127,6 +127,14 @@ pub struct DiscStream { event_fn: Option>, eof: bool, + // Cumulative bytes successfully read from the source. Drives + // EventKind::BytesRead emission and autorip's per-device progress. + bytes_read_total: u64, + // Pre-computed total of all extents in bytes (or 0 if extents are + // empty). Carried in EventKind::BytesRead.total so consumers can show + // a percent without a separate API call. + bytes_total_extents: u64, + // PES output ts_demuxer: Option, ps_demuxer: Option, @@ -149,6 +157,8 @@ impl DiscStream { content_format: crate::disc::ContentFormat, ) -> Self { let extents = title.extents.clone(); + let bytes_total_extents: u64 = + extents.iter().map(|e| e.sector_count as u64 * 2048).sum(); let mut pids = Vec::new(); let mut parsers = Vec::new(); @@ -194,6 +204,8 @@ impl DiscStream { halt: None, event_fn: None, eof: false, + bytes_read_total: 0, + bytes_total_extents, ts_demuxer, ps_demuxer, parsers, @@ -287,6 +299,11 @@ impl DiscStream { } self.buf_valid = bytes; self.current_offset += sectors as u32; + self.bytes_read_total = self.bytes_read_total.saturating_add(bytes as u64); + self.emit(EventKind::BytesRead { + bytes: self.bytes_read_total, + total: self.bytes_total_extents, + }); break; } diff --git a/src/scsi/linux.rs b/src/scsi/linux.rs index 9911161..28903e8 100644 --- a/src/scsi/linux.rs +++ b/src/scsi/linux.rs @@ -12,8 +12,6 @@ use crate::error::{Error, Result}; use std::path::Path; const SG_IO: u32 = 0x2285; -const SG_SCSI_RESET: u32 = 0x2284; -const SG_SCSI_RESET_DEVICE: i32 = 1; const SG_DXFER_NONE: i32 = -1; const SG_DXFER_TO_DEV: i32 = -2; const SG_DXFER_FROM_DEV: i32 = -3; @@ -80,49 +78,32 @@ impl SgIoTransport { }) } - /// Reset the drive to a known good state — equivalent to unplug/replug. - /// After reset, the drive is clean and no fd is held open. + /// Clean up kernel SG_IO state and unlock the tray. NOT a hardware + /// reset — purely software cleanup before this process opens the + /// device for real work. /// - /// ## Why each step exists + /// When a previous process is killed (SIGKILL) mid-SG_IO, the kernel + /// may hold queued commands against the dead fd, and `Drop` never + /// ran so the tray may still be locked via PREVENT MEDIUM REMOVAL. + /// This routine handles both: open + close flushes the kernel SG + /// queue (sg_release cancels commands tied to the fd), the 2 s sleep + /// gives the kernel time to finish that cleanup, then a fresh fd + /// sends ALLOW MEDIUM REMOVAL to clear any stale tray lock. /// - /// When a process is killed (SIGKILL/kill -9) mid-SG_IO ioctl, two things - /// go wrong: (1) the kernel's SG driver may have stale pending commands - /// queued for the dead process's fd, and (2) the drive firmware may still - /// be mid-operation (seeking, reading, processing a vendor command). - /// - /// A new process opening the same /dev/sg* device gets a fresh fd, but the - /// kernel doesn't automatically abort the dead process's commands — the - /// drive can appear hung on the first SCSI command. - /// - /// Additionally, killed processes skip Drop, so the tray may be locked - /// via PREVENT MEDIUM REMOVAL with no process alive to unlock it. - /// - /// ## Sequence - /// - /// 1. **open** — allocates kernel SG state for this fd - /// 2. **close** — triggers kernel cleanup: aborts any pending SG_IO - /// commands associated with this fd. The key operation — - /// the kernel's sg_release() cancels queued commands. - /// 3. **sleep 2s** — the drive firmware needs time to finish/abort whatever - /// it was doing when the previous process died. Without - /// this, the next command may block on drive-internal state. - /// 4. **open** — fresh fd with no stale commands in the kernel queue - /// 5. **unlock** — ALLOW MEDIUM REMOVAL (CDB 0x1E, prevent=0). Clears - /// any tray lock left by a killed process that never - /// ran its Drop/cleanup. - /// 6. **TUR** — TEST UNIT READY (CDB 0x00) with 3s timeout. If the - /// drive responds, it's in a good state. - /// 7. **escalate** — if TUR fails: - /// - SG_SCSI_RESET (device level) — kernel sends a SCSI - /// bus reset to the device, clearing all firmware state. - /// - STOP + START UNIT (CDB 0x1B) — power-cycles the - /// drive's logical unit, like pressing the eject button - /// and reinserting. - /// 8. **close** — release the fd. Drive is clean, nobody holds it. + /// We do NOT verify the drive with TUR or escalate to SG_SCSI_RESET / + /// STOP+START UNIT. Both escalations were tried in 0.13.0–0.13.5 + /// against the LG BU40N (Initio USB-SATA bridge); both failed to + /// recover wedged drives and made the wedge worse — see + /// (internal)/postmortems/2026-04-25-bu40n-wedge-recovery.md. + /// If the drive is genuinely unresponsive, the next workload command + /// fails naturally and the caller surfaces a "physical reconnect + /// required" prompt. Software has no path back from a wedged Initio + /// bridge — only physical replug clears it. pub fn reset(device: &Path) -> Result<()> { let c_path = Self::to_c_path(device); - // Step 1-2: open + close — flush stale kernel SG_IO state + // open + close — make the kernel cancel any SG_IO commands queued + // against a previous fd that didn't close cleanly. let probe_fd = unsafe { libc::open( c_path.as_ptr() as *const libc::c_char, @@ -133,10 +114,10 @@ impl SgIoTransport { unsafe { libc::close(probe_fd) }; } - // Step 3: let drive settle + // Let the kernel finish that cancellation before we reopen. std::thread::sleep(std::time::Duration::from_secs(2)); - // Step 4: open clean fd + // Fresh fd just to send the unlock command, then close. let fd = unsafe { libc::open( c_path.as_ptr() as *const libc::c_char, @@ -147,27 +128,10 @@ impl SgIoTransport { return Self::open_error(device); } - // Step 5: unlock tray + // ALLOW MEDIUM REMOVAL — clear any tray lock left by a killed + // process whose Drop never ran. Best-effort; ignore result. let _ = Self::raw_command(fd, &[0x1E, 0, 0, 0, 0, 0], 3_000); - // Step 6: TUR — if drive responds, we're done - if Self::raw_command(fd, &[0, 0, 0, 0, 0, 0], 3_000).is_err() { - // Step 7: escalate — SG_SCSI_RESET - let mut reset_type: i32 = SG_SCSI_RESET_DEVICE; - unsafe { libc::ioctl(fd, SG_SCSI_RESET as _, &mut reset_type) }; - std::thread::sleep(std::time::Duration::from_secs(3)); - - if Self::raw_command(fd, &[0, 0, 0, 0, 0, 0], 3_000).is_err() { - // STOP + START - let _ = Self::raw_command(fd, &[0x1B, 0, 0, 0, 0x00, 0], 3_000); - std::thread::sleep(std::time::Duration::from_secs(1)); - let _ = Self::raw_command(fd, &[0x1B, 0, 0, 0, 0x01, 0], 3_000); - std::thread::sleep(std::time::Duration::from_secs(3)); - let _ = Self::raw_command(fd, &[0, 0, 0, 0, 0, 0], 3_000); - } - } - - // Step 8: close — drive is clean unsafe { libc::close(fd) }; Ok(()) } diff --git a/src/scsi/macos.rs b/src/scsi/macos.rs index dc28582..d8d8180 100644 --- a/src/scsi/macos.rs +++ b/src/scsi/macos.rs @@ -260,23 +260,7 @@ impl MacScsiTransport { }) } - /// Reset the drive to a known good state. - /// On macOS, we open the device, release exclusive access, wait for - /// the system to reclaim it, then the next open() re-acquires. - /// IOKit's USB layer handles device-level resets internally when the - /// exclusive access is released and re-acquired. - /// - /// NOTE: untested — macOS reset may need IOUSBDeviceInterface::ResetDevice() - /// for USB drives. This is a best-effort implementation. - pub fn reset(device: &Path) -> Result<()> { - // Opening and immediately dropping triggers release of exclusive access - // which forces IOKit to reset the device state. - if let Ok(transport) = Self::open(device) { - drop(transport); // Drop releases exclusive access + closes plugin - } - std::thread::sleep(std::time::Duration::from_secs(2)); - Ok(()) - } + // `reset()` removed in 0.13.6 — see scsi/mod.rs for rationale. } /// Enumerate optical drives on macOS. Mirrors `drive::macos::find_drives` diff --git a/src/scsi/mod.rs b/src/scsi/mod.rs index 2391688..415f089 100644 --- a/src/scsi/mod.rs +++ b/src/scsi/mod.rs @@ -98,103 +98,13 @@ pub fn open(device: &Path) -> Result> { } } -/// Default upper bound on `scsi::reset()`. The platform-specific reset -/// sequence does ~15 s of bounded sleeps + ioctls in the happy path, so -/// 30 s is roughly 2× the worst-case happy time — long enough that a -/// healthy-but-slow drive isn't false-positively timed out, short enough -/// that a kernel-wedged ioctl doesn't take down the caller's poll loop -/// for minutes. -pub(crate) const DEFAULT_RESET_TIMEOUT_SECS: u64 = 30; - -/// Reset a SCSI device to a known good state, with a hard wallclock -/// bound (`DEFAULT_RESET_TIMEOUT_SECS`). On Linux: open/close fd cycle + -/// TUR + SG_SCSI_RESET escalation. -/// -/// **Why the timeout matters.** SG_SCSI_RESET is an ioctl that can block -/// indefinitely on a kernel-wedged USB target (the kernel waits for the -/// SCSI subsystem to ack the reset, which never comes from a dead-bus -/// device). Without an outer bound, the call hangs the caller's thread -/// forever — observed in production on a wedged BU40N where the autorip -/// poll loop sat in the ioctl for 60+ s. The bounded version returns -/// `DeviceResetFailed` after `DEFAULT_RESET_TIMEOUT_SECS`; the inner -/// thread keeps running until the kernel eventually unblocks it (we -/// can't cancel a Linux ioctl from userspace), so this leaks one OS -/// thread per wedge — acceptable cost for a daemon that recovers -/// instead of hanging. -/// -/// `pub(crate)` — outside callers use the higher-level `drive_has_disc` -/// (which folds in recovery escalation) or `Drive::reset` (instance-level). -/// Direct primitive exposure removed in 0.13.2 to enforce the -/// architectural rule that no consumer crate issues SCSI commands. -pub(crate) fn reset(device: &Path) -> Result<()> { - reset_with_timeout( - device, - std::time::Duration::from_secs(DEFAULT_RESET_TIMEOUT_SECS), - ) -} - -/// Reset with a caller-specified timeout. See [`reset`] for the full -/// rationale on why an outer wallclock bound is required. -pub(crate) fn reset_with_timeout(device: &Path, timeout: std::time::Duration) -> Result<()> { - let device_owned = device.to_path_buf(); - let (tx, rx) = std::sync::mpsc::channel(); - - // Detach a worker thread for the actual reset. We never `join` it — - // if the kernel ioctl is wedged, the join would block forever, which - // is the very thing we're protecting the caller from. The thread will - // exit on its own when the kernel eventually returns from the ioctl - // (or never, if the device is permanently dead — process exit cleans - // it up). - std::thread::Builder::new() - .name("scsi-reset".into()) - .spawn(move || { - let r = reset_blocking(&device_owned); - let _ = tx.send(r); - }) - .map_err(|_| Error::DeviceResetFailed { - path: device.display().to_string(), - })?; - - match rx.recv_timeout(timeout) { - Ok(result) => result, - Err(std::sync::mpsc::RecvTimeoutError::Timeout) => Err(Error::DeviceResetFailed { - path: device.display().to_string(), - }), - Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { - // The worker thread panicked before sending. Surface as a - // reset failure rather than a generic error. - Err(Error::DeviceResetFailed { - path: device.display().to_string(), - }) - } - } -} - -/// Inner reset — runs on the worker thread. May block indefinitely if -/// the kernel's SCSI subsystem is wedged. Callers must use the bounded -/// `reset()` wrapper above; this raw function isn't exposed. -fn reset_blocking(device: &Path) -> Result<()> { - #[cfg(target_os = "linux")] - { - linux::SgIoTransport::reset(device) - } - - #[cfg(target_os = "macos")] - { - macos::MacScsiTransport::reset(device) - } - - #[cfg(target_os = "windows")] - { - windows::SptiTransport::reset(device) - } - - #[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))] - { - let _ = device; - Ok(()) - } -} +// Note: a top-level `scsi::reset()` used to live here, wrapping a +// platform reset in a thread+recv_timeout so a kernel-wedged ioctl +// couldn't hang the caller. Removed in 0.13.6 along with the +// SG_SCSI_RESET / STOP+START UNIT escalation that needed it. The +// remaining platform reset (Linux: SgIoTransport::reset, called only +// from SgIoTransport::open) does pure userspace state cleanup with +// bounded sleeps — no escape-hatch wrapper required. // ── USB-layer recovery: rolled back in 0.13.4 ─────────────────────────────── // diff --git a/tests/integration_progress_and_halt.rs b/tests/integration_progress_and_halt.rs new file mode 100644 index 0000000..56ffb97 --- /dev/null +++ b/tests/integration_progress_and_halt.rs @@ -0,0 +1,337 @@ +//! Integration tests for progress reporting, halt behavior, drop safety, +//! and the file-backed sector reader round trip. + +use libfreemkv::disc::{CopyOptions, DiscRegion}; +use libfreemkv::error::Result; +use libfreemkv::pes::Stream as PesStream; +use libfreemkv::{ + ContentFormat, Disc, DiscFormat, DiscStream, DiscTitle, EventKind, Extent, FileSectorReader, + SectorReader, +}; +use std::io::Write; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +const SECTOR_SIZE: usize = 2048; + +// ── helpers ──────────────────────────────────────────────────────────────── + +/// Returns zeroed sectors. Always succeeds. Counts each call. +struct ZeroSectorReader { + capacity: u32, + calls: Arc, +} + +impl ZeroSectorReader { + fn new(capacity: u32) -> Self { + Self { + capacity, + calls: Arc::new(AtomicU64::new(0)), + } + } +} + +impl SectorReader for ZeroSectorReader { + fn read_sectors( + &mut self, + _lba: u32, + count: u16, + buf: &mut [u8], + _recovery: bool, + ) -> Result { + self.calls.fetch_add(1, Ordering::Relaxed); + let bytes = count as usize * SECTOR_SIZE; + buf[..bytes].fill(0); + Ok(bytes) + } + + fn capacity(&self) -> u32 { + self.capacity + } +} + +/// Like ZeroSectorReader but sleeps a configurable duration per call. +/// Used by the halt test so the copy takes >1 s. +struct SlowZeroSectorReader { + capacity: u32, + sleep_per_call: Duration, +} + +impl SlowZeroSectorReader { + fn new(capacity: u32, sleep_per_call: Duration) -> Self { + Self { + capacity, + sleep_per_call, + } + } +} + +impl SectorReader for SlowZeroSectorReader { + fn read_sectors( + &mut self, + _lba: u32, + count: u16, + buf: &mut [u8], + _recovery: bool, + ) -> Result { + std::thread::sleep(self.sleep_per_call); + let bytes = count as usize * SECTOR_SIZE; + buf[..bytes].fill(0); + Ok(bytes) + } + + fn capacity(&self) -> u32 { + self.capacity + } +} + +/// Build a Disc instance with a known capacity, no titles, no encryption. +/// Sufficient for `Disc::copy` (which only uses capacity_sectors + decrypt keys). +fn synthetic_disc(capacity_sectors: u32) -> Disc { + Disc { + volume_id: String::new(), + meta_title: None, + format: DiscFormat::BluRay, + capacity_sectors, + capacity_bytes: capacity_sectors as u64 * SECTOR_SIZE as u64, + layers: 1, + titles: Vec::new(), + region: DiscRegion::Free, + aacs: None, + css: None, + encrypted: false, + content_format: ContentFormat::BdTs, + } +} + +/// Build a DiscTitle with a single extent of `sector_count` sectors and no +/// streams (DiscStream still iterates sectors and would emit BytesRead). +fn synthetic_title(sector_count: u32) -> DiscTitle { + DiscTitle { + playlist: String::new(), + playlist_id: 0, + duration_secs: 0.0, + size_bytes: sector_count as u64 * SECTOR_SIZE as u64, + clips: Vec::new(), + streams: Vec::new(), + chapters: Vec::new(), + extents: vec![Extent { + start_lba: 0, + sector_count, + }], + content_format: ContentFormat::BdTs, + codec_privates: Vec::new(), + } +} + +// ── 1. BytesRead events emitted during disc copy (TDD red) ──────────────── + +#[test] +fn test_bytes_read_emitted_during_disc_copy() { + // Build a tiny synthetic disc and stream it through DiscStream. + let reader = ZeroSectorReader::new(64); + let title = synthetic_title(64); + let keys = libfreemkv::DecryptKeys::None; + + let mut stream = DiscStream::new(Box::new(reader), title, keys, 60, ContentFormat::BdTs); + + let count = Arc::new(AtomicU64::new(0)); + let count_cb = count.clone(); + stream.on_event(move |ev| { + if let EventKind::BytesRead { .. } = ev.kind { + count_cb.fetch_add(1, Ordering::Relaxed); + } + }); + + // Drive the stream to EOF. With no streams configured, read() returns + // Ok(None) once all extents are exhausted. + loop { + match stream.read() { + Ok(Some(_frame)) => {} + Ok(None) => break, + Err(e) => panic!("stream read failed: {e:?}"), + } + } + + let n = count.load(Ordering::Relaxed); + // EXPECTED TO FAIL until BytesRead emission is wired up. TDD red. + assert!( + n > 0, + "expected at least one BytesRead event, got {n} (lib does not yet emit BytesRead)" + ); +} + +// ── 2. Disc::copy on_progress callback fires (regression guard) ─────────── + +#[test] +fn test_disc_copy_progress_callback_fires() { + let disc = synthetic_disc(64); + let mut reader = ZeroSectorReader::new(64); + + let tmp = tempfile::NamedTempFile::new().expect("tempfile create"); + let iso_path = tmp.path().to_path_buf(); + drop(tmp); // we want the path, not the file handle + + let calls = Arc::new(AtomicU64::new(0)); + let last_bytes = Arc::new(AtomicU64::new(0)); + let calls_cb = calls.clone(); + let last_bytes_cb = last_bytes.clone(); + + let progress = move |bytes: u64, _total: u64| { + calls_cb.fetch_add(1, Ordering::Relaxed); + last_bytes_cb.store(bytes, Ordering::Relaxed); + }; + + let opts = CopyOptions { + decrypt: false, + on_progress: Some(&progress), + ..Default::default() + }; + + let result = disc.copy(&mut reader, &iso_path, &opts).expect("copy ok"); + + // Cleanup any sidecar mapfile + ISO before assertions. + let _ = std::fs::remove_file(&iso_path); + let _ = std::fs::remove_file(libfreemkv::disc::mapfile_path_for(&iso_path)); + + assert!(result.complete, "copy should be complete"); + let n = calls.load(Ordering::Relaxed); + let last = last_bytes.load(Ordering::Relaxed); + assert!(n > 0, "on_progress should fire at least once, got {n}"); + assert!(last > 0, "final progress bytes should be non-zero, got {last}"); +} + +// ── 3. Halt aborts disc copy promptly ───────────────────────────────────── + +#[test] +fn test_halt_aborts_disc_copy_promptly() { + // 6000 sectors, 60-sector batches → 100 read_sectors() calls. + // 10 ms sleep per call → ~1 s total without halt. + let capacity_sectors: u32 = 6000; + let mut reader = SlowZeroSectorReader::new(capacity_sectors, Duration::from_millis(10)); + let disc = synthetic_disc(capacity_sectors); + + let tmp = tempfile::NamedTempFile::new().expect("tempfile create"); + let iso_path = tmp.path().to_path_buf(); + drop(tmp); + + let halt = Arc::new(AtomicBool::new(false)); + let halt_for_thread = halt.clone(); + let iso_path_for_thread = iso_path.clone(); + + let join = std::thread::spawn(move || { + let opts = CopyOptions { + decrypt: false, + halt: Some(halt_for_thread), + ..Default::default() + }; + let t0 = Instant::now(); + let res = disc.copy(&mut reader, &iso_path_for_thread, &opts); + (res, t0.elapsed()) + }); + + // Let copy run, then halt. + std::thread::sleep(Duration::from_millis(200)); + halt.store(true, Ordering::Relaxed); + + // Bound the join: should exit far before the full 1 s otherwise needed. + let started = Instant::now(); + let mut joined = None; + while started.elapsed() < Duration::from_millis(2000) { + if join.is_finished() { + joined = Some(join.join().expect("thread join")); + break; + } + std::thread::sleep(Duration::from_millis(20)); + } + let (result, elapsed) = joined.expect("copy thread did not exit within 2s of halt"); + + // Cleanup + let _ = std::fs::remove_file(&iso_path); + let _ = std::fs::remove_file(libfreemkv::disc::mapfile_path_for(&iso_path)); + + let copy_result = result.expect("copy returns Ok with halted=true on halt"); + assert!( + copy_result.halted, + "copy_result.halted should be true after halt" + ); + assert!( + !copy_result.complete, + "copy_result.complete should be false when halted" + ); + assert!( + elapsed < Duration::from_millis(2000), + "copy thread exit elapsed {elapsed:?} exceeded 2s" + ); +} + +// ── 4. DiscStream Drop does not panic or block ──────────────────────────── + +#[test] +fn test_drop_impls_do_not_panic_or_block() { + let reader = ZeroSectorReader::new(64); + let title = synthetic_title(64); + let keys = libfreemkv::DecryptKeys::None; + let stream = DiscStream::new(Box::new(reader), title, keys, 60, ContentFormat::BdTs); + + // Drop on a worker thread; main thread enforces the timeout. + let handle = std::thread::spawn(move || { + drop(stream); + }); + + let started = Instant::now(); + while started.elapsed() < Duration::from_millis(100) { + if handle.is_finished() { + handle.join().expect("drop thread join"); + return; + } + std::thread::sleep(Duration::from_millis(5)); + } + panic!("DiscStream drop did not complete within 100ms"); +} + +// ── 5. FileSectorReader round trip ──────────────────────────────────────── + +#[test] +fn test_file_sector_reader_round_trip() { + // Build 8 sectors of pseudo-random bytes (sector-aligned). + const N_SECTORS: usize = 8; + let mut data = vec![0u8; N_SECTORS * SECTOR_SIZE]; + for (i, b) in data.iter_mut().enumerate() { + // Cheap PRNG: just a multiplicative pattern, deterministic for asserts. + *b = ((i as u64).wrapping_mul(2654435761) >> 16) as u8; + } + + let mut tmp = tempfile::NamedTempFile::new().expect("tempfile create"); + tmp.write_all(&data).expect("write data"); + tmp.flush().expect("flush"); + + let path = tmp.path().to_str().expect("path utf-8").to_string(); + let mut fsr = FileSectorReader::open(&path).expect("open FileSectorReader"); + + assert_eq!(fsr.capacity(), N_SECTORS as u32, "capacity mismatch"); + + // Read each sector individually and compare. + let mut buf = vec![0u8; SECTOR_SIZE]; + for lba in 0..N_SECTORS as u32 { + let n = fsr + .read_sectors(lba, 1, &mut buf, false) + .expect("read_sectors"); + assert_eq!(n, SECTOR_SIZE); + let off = lba as usize * SECTOR_SIZE; + assert_eq!( + &buf[..], + &data[off..off + SECTOR_SIZE], + "sector {lba} mismatch" + ); + } + + // Read all sectors at once and compare. + let mut all = vec![0u8; N_SECTORS * SECTOR_SIZE]; + let n = fsr + .read_sectors(0, N_SECTORS as u16, &mut all, false) + .expect("read all sectors"); + assert_eq!(n, N_SECTORS * SECTOR_SIZE); + assert_eq!(all, data, "bulk read mismatch"); +}