pq/libfreemkv
1
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
a0f76a4f1864af414211b6144a409549b144c884
libfreemkv/src/mux/disc.rs
T
Matthew Jackson 528a6b7345 Walk past empty extents iteratively instead of recursing 
fill_extents skipped an exhausted or zero-sector extent by calling itself,
which costs a stack frame per skipped extent. Nothing filters
sector_count == 0 out of a UDF or MPLS extent list, so a malformed disc
declaring a long run of empty extents recursed once per extent before
reading a single sector. Rust does not guarantee tail-call elimination, so
that overflows the stack — which aborts the process rather than returning
an io::Error, taking a long-running service down with it.

The skip is now a loop. The regression test runs on a 256 KiB stack, where
the recursive version dies and the loop finishes immediately.
2026-08-01 11:05:12 -07:00

2610 lines
109 KiB
Rust
Raw Blame History

//! DiscStream — read any disc (physical drive or ISO file) → PES frames.
//!
//! One stream type for all disc sources. The source is a SectorSource —
//! Drive (hardware) or FileSectorSource (file). DiscStream doesn't care.
//!
//! Read-only. For disc→ISO (raw sector copy), use `freemkv_engine::recovery::copy`.
use crate::disc::{DiscTitle, Extent};
use crate::drive::extract_scsi_context;
use crate::event::{BatchSizeReason, Event, EventKind};
use crate::halt::Halt;
use crate::sector::{DecryptingSectorSource, SectorSource};
use std::io;
#[cfg(test)]
use std::sync::Arc;
#[cfg(test)]
use std::sync::atomic::AtomicBool;
/// Ramp back up to the preferred batch size after this many sectors
/// of clean reading at the current (reduced) size. 100 MiB = 51,200 sectors.
///
/// Chosen so that an isolated transient failure doesn't lock the rip at
/// size 1: once past the bad zone, we probe up after ~100 ms of good reads.
/// And so that noisy zones with occasional successes can't trigger a
/// premature probe — we need a sustained clean run.
const PROBE_THRESHOLD_SECTORS: u32 = 100 * 1024 * 1024 / 2048;
/// Halve a batch size, keeping 3-sector alignment when >= 6
/// (3-sector alignment = one AACS unit). At sizes < 6 we descend
/// through 3 → 1 without intermediate unaligned sizes.
fn halve_batch_size(size: u16) -> u16 {
let h = (size / 2).max(1);
if h >= 6 { h - (h % 3) } else { h }
}
/// Double a batch size toward a preferred max, keeping 3-sector alignment
/// when the result is >= 6.
fn double_batch_size(size: u16, preferred: u16) -> u16 {
let d = size.saturating_mul(2).min(preferred);
if d >= 6 { d - (d % 3) } else { d }
}
/// Adaptive batch sizer. Shrinks on read failure, grows after a sustained
/// clean streak. Amortizes the cost of entering a bad zone — descent happens
/// once, not once per bad sector.
#[derive(Debug)]
struct AdaptiveBatch {
preferred: u16,
current: u16,
streak_sectors: u32,
}
impl AdaptiveBatch {
fn new(preferred: u16) -> Self {
Self {
preferred,
current: preferred,
streak_sectors: 0,
}
}
fn current(&self) -> u16 {
self.current
}
/// Record a successful read of `sectors`. Returns an event if the
/// sizer probed up to a larger batch size.
fn on_success(&mut self, sectors: u16) -> Option<EventKind> {
self.streak_sectors = self.streak_sectors.saturating_add(sectors as u32);
if self.current < self.preferred && self.streak_sectors >= PROBE_THRESHOLD_SECTORS {
let new_size = double_batch_size(self.current, self.preferred);
if new_size != self.current {
self.current = new_size;
self.streak_sectors = 0;
return Some(EventKind::BatchSizeChanged {
new_size,
reason: BatchSizeReason::Probed,
});
}
}
None
}
/// Record a read failure. Returns an event if the sizer shrank.
/// Does nothing at size 1 (caller handles skip/error).
fn on_failure(&mut self) -> Option<EventKind> {
self.streak_sectors = 0;
if self.current <= 1 {
return None;
}
let new_size = halve_batch_size(self.current);
self.current = new_size;
Some(EventKind::BatchSizeChanged {
new_size,
reason: BatchSizeReason::Shrunk,
})
}
}
/// Disc stream. Reads sectors from any source → PES frames.
///
/// Sources: physical drive, ISO file, or any SectorSource.
/// Decrypt, demux, and codec parsing happen internally.
pub struct DiscStream {
/// Underlying sector source wrapped in the 0.18
/// [`DecryptingSectorSource`] decorator. Every `read_sectors`
/// call yields plaintext, so `fill_extents` no longer needs an
/// inline `decrypt::decrypt_sectors` step. `DecryptKeys::None`
/// (raw / unencrypted disc) makes the decorator a pass-through.
reader: DecryptingSectorSource<Box<dyn SectorSource>>,
title: DiscTitle,
/// Mirror of the keys handed in at construction. The decorator
/// owns the cryptographic state; this field is kept for
/// metadata-side callers (`info()` and friends) that want to
/// know whether the disc was encrypted, without reaching through
/// the wrapper.
decrypt_keys: crate::decrypt::DecryptKeys,
/// Sector granularity the decrypt step requires each read buffer to start
/// on and span a multiple of. AACS decrypts whole 6144-byte (3-sector)
/// units keyed off the buffer's first 16 bytes, so every `read_sectors`
/// buffer must begin on a real on-disc unit boundary — hence reads and
/// error-skips must stay aligned to this. `3` for AACS, `1` for CSS /
/// unencrypted (per-sector, self-synchronizing). Mirrors the file-backed
/// highway's `PrefetchedSectorSource` guard; this is the inline live path.
unit_align: u16,
// Extents to read
extents: Vec<Extent>,
// Position
current_extent: usize,
current_offset: u32,
// Buffer
read_buf: Vec<u8>,
buf_valid: usize,
// Adaptive batch sizer — preferred comes from the caller
// (detect_max_batch_sectors), shrinks/grows based on read outcomes.
adaptive: AdaptiveBatch,
errors: u64,
/// Cumulative bytes actually skipped (zero-filled) on read error.
/// Distinct from `errors`, which counts skip *events*: one event can
/// cover a whole AACS unit (`unit_align` sectors = 6144 bytes), so
/// `errors * 2048` understates real loss by the alignment factor.
/// Consumers estimating lost video time must scale by this, not by
/// the event count.
lost_bytes: u64,
pub skip_errors: bool,
/// When set and the token is cancelled, fill_extents returns Err(Halted)
/// at the next retry boundary. Unlike skip_errors, this propagates the
/// error up so the rip terminates cleanly. Construct with
/// [`DiscStream::with_halt`], passing the same `Halt` clone handed to
/// sweep / patch / mux so every phase observes one Stop signal.
halt: Option<Halt>,
event_fn: Option<Box<dyn Fn(Event) + Send>>,
eof: bool,
/// Count of dropped DVD navigation packets (private_stream_2, 0xBF) — these
/// are expected on every disc; tallied and summarised once at EOF instead of
/// a per-packet WARN.
dropped_nav_packets: u64,
// 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 — single-threaded inline demux + codec parse. The
// pipeline-mode mux (3-stage threaded) lives in
// [`super::pipelined_stream::PipelinedPesStream`]; this type is
// the legacy in-thread path for live-disc reads where adaptive
// batch retry on bad sectors lives in `fill_extents`.
ts_demuxer: Option<super::ts::TsDemuxer>,
ps_demuxer: Option<super::ps::PsDemuxer>,
parsers: Vec<(u16, Box<dyn super::codec::CodecParser>)>,
pending_frames: std::collections::VecDeque<crate::pes::PesFrame>,
pid_to_track: Vec<(u16, usize)>,
/// Cached `FREEMKV_SKIP_PARSE` profiling flag. The env var cannot
/// change at runtime, and `std::env::var_os` takes a process-wide
/// lock; reading it once at construction keeps it out of the
/// per-batch read() hot loop.
skip_parse: bool,
/// Cached `FREEMKV_PROFILE` presence, read once at construction. When
/// false, the read() loop skips the four `Instant::now()` captures and the
/// `prof_tick` calls entirely, so profiling-off runs pay no per-iteration
/// timestamp cost or `prof_active()` env-var lookup (which takes a
/// process-wide lock).
profiling: bool,
/// B1 drop-to-keyframe resync gate, one per stream — mirrors
/// `PipelinedPesStream`. After an AACS-concealed (undecryptable) gap stamps
/// a frame `discontinuity`, the gate drops forward inter-coded video frames
/// until the next keyframe so no dangling-reference frame reaches the muxer.
/// Without this the live-drive single-pass path emitted decode-broken output
/// on an undecryptable unit (the file-backed highway path is already gated).
resync: Vec<super::resync::ResyncGate>,
/// Whether each stream (by track index) is video — audio/subtitle frames are
/// independent and always admit.
is_video: Vec<bool>,
}
impl DiscStream {
/// Create a disc stream from any sector reader.
///
/// Works with physical drives and ISO files — both implement SectorSource.
/// The caller opens the source, scans for titles/keys, and passes them in.
/// The stream handles demuxing, decryption, and codec parsing internally.
pub fn new(
mut reader: Box<dyn SectorSource>,
title: DiscTitle,
mut decrypt_keys: crate::decrypt::DecryptKeys,
batch_sectors: u16,
content_format: crate::disc::ContentFormat,
raw: bool,
halt: Option<Halt>,
) -> std::io::Result<Self> {
let mut title = title;
let extents = title.extents.clone();
// Resolve this title's CSS key from the reader if the caller supplied
// none — the SAME shared step the file-backed mux highway
// (`build_iso_pipeline`) uses, so single-pass and multi-pass descramble a
// DVD identically. No-op for AACS / already-keyed / genuinely-clear input
// or `raw`; a scrambled-but-uncrackable DVD is a hard `CssKeyMissing`.
// `halt` is passed here (not deferred to `with_halt`) so a Stop during the
// crack scan is honored — the scan runs at construction, before the caller
// can attach a token.
crate::css::resolve_dvd_title_key(
&mut *reader,
&extents,
&mut decrypt_keys,
batch_sectors,
content_format,
raw,
halt.as_ref(),
)?;
let bytes_total_extents: u64 = extents.iter().map(|e| e.sector_count as u64 * 2048).sum();
// Debug log reader type at construction — critical for diagnosing mux
// reading from drive instead of ISO. `type_name_of_val(&*reader)`
// resolves the CONCRETE type behind the box (Drive / FileSectorSource),
// unlike `type_name::<dyn SectorSource>()` which always prints the
// trait-object name regardless of the underlying source.
tracing::debug!(
target: "mux",
"DiscStream constructed with reader type: {}",
std::any::type_name_of_val(&*reader)
);
// CSS/unencrypted content needs a decrypting wrapper to yield plaintext
// VOB bytes before the AC-3 sub-stream probe can read real `acmod`s.
// MUX path (read > decrypt > mux): decrypt every unit in place and pass the
// bytes to the muxer; a unit that decrypts to broken TS is the muxer's
// concern, never conceal / re-fetch / count as loss (fail loud only on a
// genuine can't-decrypt). DiscStream is a decode/mux stream (live-drive
// single-pass / direct), never the ciphertext-preserving sweep.
let mut reader = DecryptingSectorSource::new(reader, decrypt_keys.clone());
// Wrong-substream fix (Silence-of-the-Lambs): re-route the title's
// declared AC-3 audio onto the physically-correct `0x8x` sub-streams by
// probing their real channel counts off the head of the feature. No-op
// for non-DVD or when the probe yields nothing.
crate::disc::dvd_audio_probe::probe_and_remap(&mut reader, &mut title);
// Parser table + PID map + demuxer come from the CANONICAL builder shared
// with the file-backed highway (`resolve::build_demux_state`). This used
// to be an open-coded copy of the same loop, which drifted: it built every
// parser with plain `parser_for_codec` and so never routed a Blu-ray 3D
// MVC dependent (right-eye) view to the param-set-passthrough parser
// (ISO/IEC 14496-10 Annex H) — the same 3D disc muxed correctly from an
// ISO and incorrectly from a live drive. Call the canonical builder rather
// than repeating its dispatch, so a future rule added there cannot go
// missing on the `disc://` path again.
let (parsers, pid_to_track, ts_demuxer, ps_demuxer) =
super::resolve::build_demux_state(&title, content_format);
// AACS decrypts whole 6144-byte (3-sector) units keyed off each read
// buffer's first 16 bytes, so reads/skips must stay 3-sector aligned.
// CSS and unencrypted content are per-2048-byte and self-synchronizing
// (align 1). Same rule the file-backed highway applies in resolve.rs.
let unit_align: u16 = match &decrypt_keys {
crate::decrypt::DecryptKeys::Aacs { .. } => 3,
_ => 1,
};
// `reader` is already wrapped in DecryptingSectorSource above (so the
// internal fill_extents path sees plaintext bytes; for DecryptKeys::None
// the decorator is a pass-through). Reset the unit base the probe read
// advanced so the first fill_extents read starts cleanly.
reader.set_unit_base(0);
// B1 resync gates: one per stream, video flagged so the gate only
// drop-to-keyframes video (audio/subtitle always admit). Computed before
// `title` is moved into the struct below.
let is_video: Vec<bool> = title
.streams
.iter()
.map(|s| matches!(s, crate::disc::Stream::Video(_)))
.collect();
let resync = (0..title.streams.len())
.map(|_| super::resync::ResyncGate::new())
.collect();
Ok(Self {
reader,
title,
decrypt_keys,
unit_align,
extents,
current_extent: 0,
current_offset: 0,
read_buf: Vec::with_capacity(batch_sectors as usize * 2048),
buf_valid: 0,
adaptive: AdaptiveBatch::new(batch_sectors),
errors: 0,
lost_bytes: 0,
skip_errors: false,
halt,
event_fn: None,
eof: false,
dropped_nav_packets: 0,
bytes_read_total: 0,
bytes_total_extents,
ts_demuxer,
ps_demuxer,
parsers,
pending_frames: std::collections::VecDeque::new(),
pid_to_track,
skip_parse: std::env::var_os("FREEMKV_SKIP_PARSE").is_some(),
profiling: std::env::var_os("FREEMKV_PROFILE").is_some(),
resync,
is_video,
})
}
/// Set event handler for sector-level events (binary search, skip, recover).
pub fn on_event(&mut self, f: impl Fn(Event) + Send + 'static) {
self.event_fn = Some(Box::new(f));
}
/// Constructor-time builder: attach a [`Halt`] token so that when
/// any clone is cancelled, the next read-retry boundary inside
/// `fill_extents` returns `Err(Halted)`. Required for Stop to work
/// during dense bad-sector regions (where the outer PES read() loop
/// can spend minutes inside fill_extents before emitting a frame).
///
/// Pass the same `Halt` clone you hand to sweep / patch / mux so every
/// phase observes a single Stop signal.
pub fn with_halt(mut self, halt: Halt) -> Self {
self.halt = Some(halt);
self
}
/// Install a proactive [`AacsKeyMap`](crate::decrypt::AacsKeyMap) on the inline
/// live-drive path — the counterpart to what
/// [`build_iso_pipeline`](crate::mux::resolve::build_iso_pipeline) does for the
/// file-backed highway. The map is the title's read plan: it decides which unit
/// each LBA is and, for an FMTS forensic segment, which phase is ours. The
/// extent walk is rewritten to the read plan so **only our-phase units are read
/// off the drive** (the alternate device-group units are never fetched,
/// decrypted, or muxed), and the map is installed so each unit decrypts with its
/// mapped key. A non-forensic map returns the extents unchanged, so a plain
/// single/multi-CPS disc reads exactly as before.
pub fn with_key_map(mut self, map: std::sync::Arc<crate::decrypt::AacsKeyMap>) -> Self {
self.extents = map.read_plan(&self.extents, self.unit_align.max(1) as u32);
self.bytes_total_extents = self
.extents
.iter()
.map(|e| e.sector_count as u64 * 2048)
.sum();
self.reader.set_key_map(map);
self
}
fn is_halted(&self) -> bool {
self.halt
.as_ref()
.map(|h| h.is_cancelled())
.unwrap_or(false)
}
fn emit(&self, kind: EventKind) {
if let Some(ref f) = self.event_fn {
f(Event { kind });
}
}
/// Skip decryption — return raw encrypted bytes. Updates both
/// the metadata-side key field and the wrapped reader's keys so
/// subsequent `read_sectors` calls become a pass-through.
pub fn set_raw(&mut self) {
self.decrypt_keys = crate::decrypt::DecryptKeys::None;
self.reader.set_keys(crate::decrypt::DecryptKeys::None);
}
/// Commit a SUCCESSFUL `read_sectors` into the read buffer and advance the
/// extent cursor. `got` is the byte count the source reported (already
/// clamped to `bytes`, the full span requested for `sectors`).
///
/// The full case (`got == bytes`) is the only one every in-tree
/// `SectorSource` produces — they are all full-or-error — and it behaves
/// exactly as before.
///
/// A SHORT read (`got < bytes`) used to be handled inconsistently: the
/// returned count was trusted for `buf_valid`, but `current_offset` still
/// advanced by the full requested `sectors`. The undelivered tail therefore
/// vanished from the muxed title with no error, no `SectorSkipped` event and
/// no `lost_bytes` — a silent hole, indistinguishable from clean output, in
/// a single-pass path that has no later pass to recover it. It is handled
/// here rather than trusted away because invisible data loss is the one
/// outcome this stream must never produce.
///
/// The tail is NOT re-read from a smaller offset: `current_offset` must stay
/// on an AACS unit boundary (`unit_align`), and resuming mid-unit desyncs
/// the decrypt of everything after it — the same reason the failed-unit skip
/// branch below advances by the whole unit. So the gap is either
///
/// - a hard [`crate::error::Error::DiscRead`] (E6000) when the caller has
/// NOT opted into holes, or
/// - zero-filled and ACCOUNTED under `skip_errors`, identically to a failed
/// unit: the stale buffer tail is cleared, `errors` / `lost_bytes` are
/// charged, and a `SectorSkipped` event is emitted.
fn commit_read(&mut self, lba: u32, got: usize, bytes: usize) -> io::Result<()> {
if got < bytes {
if !self.skip_errors {
return Err(crate::error::Error::DiscRead {
sector: lba as u64,
status: None,
sense: None,
}
.into());
}
// Clear the tail the source did not write — it holds stale bytes
// from a previous batch, which would mux as plausible garbage.
self.read_buf[got..bytes].fill(0);
self.errors += 1;
self.lost_bytes = self.lost_bytes.saturating_add((bytes - got) as u64);
self.emit(EventKind::SectorSkipped { sector: lba as u64 });
}
self.buf_valid = bytes;
// `bytes` is always a whole number of 2048-byte logical sectors (it is
// `sectors * 2048`), so this is the requested sector count — the cursor
// advances a whole unit span either way, keeping AACS alignment.
self.current_offset += (bytes / 2048) as u32;
// Only the bytes the source actually delivered count as read; the
// zero-filled tail is loss, already charged to `lost_bytes`.
self.bytes_read_total = self.bytes_read_total.saturating_add(got as u64);
self.emit(EventKind::BytesRead {
bytes: self.bytes_read_total,
total: self.bytes_total_extents,
});
Ok(())
}
fn fill_extents(&mut self) -> io::Result<bool> {
// Skip exhausted and zero-sector extents iteratively. This was a
// self-recursive call, which costs a stack frame per skipped extent:
// nothing filters `sector_count == 0` out of a UDF/MPLS extent list, so
// a malformed disc declaring thousands of empty extents in a row could
// overflow the stack. That aborts the process — uncatchable, and fatal
// to a long-running service — rather than surfacing as an io::Error.
let (ext_start, ext_sectors, remaining) = loop {
if self.current_extent >= self.extents.len() {
return Ok(false);
}
let ext = &self.extents[self.current_extent];
let remaining = ext.sector_count.saturating_sub(self.current_offset);
if remaining > 0 {
break (ext.start_lba, ext.sector_count, remaining);
}
self.current_extent += 1;
self.current_offset = 0;
};
// start_lba comes from UDF/MPLS extents; a malformed extent near
// u32::MAX would overflow (debug panic / release wrap to a wrong LBA).
// Saturate for consistency with the rest of the file's arithmetic.
let lba = ext_start.saturating_add(self.current_offset);
// AACS aligned units are anchored at this extent's start LBA — gate the
// decrypt-on-read source relative to it (clip-anchored), not absolute
// disc LBA 0. No-op for CSS / None.
self.reader.set_unit_base(ext_start);
// Adaptive sizer: start at current (preferred until a failure), shrink
// on failure, advance on success. One 5s read attempt per try — no
// retry loops, no sleeps. On size-1 failure, skip or error.
//
// Halt is checked at the top of every iteration — in a dense bad zone
// this loop can spend minutes shrinking and skipping sectors; without
// the check, Stop wouldn't take effect until the outer PES read() loop
// finally emits a frame, which may never happen.
let start_time = std::time::Instant::now();
loop {
if self.is_halted() {
return Err(crate::error::Error::Halted.into());
}
// Debug: log slow reads during mux — helps diagnose stalls
if cfg!(debug_assertions) && start_time.elapsed().as_secs() > 5 {
tracing::debug!(target: "mux", "fill_extents waiting at LBA {} ({}s elapsed, sectors={})", lba, start_time.elapsed().as_secs(), remaining);
}
// Keep every read buffer starting on a real on-disc unit boundary.
// AACS (unit_align=3) decrypts whole 6144-byte units keyed off the
// buffer's first bytes, so a sub-unit read mid-extent desyncs the
// rest of the title; always read at least one full unit. Only the
// final partial unit at the extent tail (remaining < align) is read
// short — nothing follows it to desync. CSS/raw (align=1) is
// per-sector and self-synchronizing, so this is a no-op there.
let align = self.unit_align.max(1) as u32;
let want = remaining.min(self.adaptive.current() as u32);
let sectors: u16 = if align <= 1 {
want as u16
} else if remaining < align {
remaining as u16
} else if want < align {
align as u16
} else {
(want - want % align) as u16
};
let bytes = sectors as usize * 2048;
self.read_buf.resize(bytes, 0);
let res = self
.reader
.read_sectors(lba, sectors, &mut self.read_buf[..bytes], false);
if let Ok(&got) = res.as_ref() {
// SectorSource::read_sectors returns the number of bytes
// written into buf. All in-tree sources return full-or-error,
// but a short count would leave the stale/zeroed tail of
// read_buf in place.
debug_assert!(got <= bytes, "read_sectors over-reported byte count");
if let Some(ev) = self.adaptive.on_success(sectors) {
self.emit(ev);
}
let got = got.min(bytes);
self.commit_read(lba, got, bytes)?;
break;
}
// Transport failure (status=0xFF: USB-bridge crash / disconnect) is
// NOT a skippable bad sector. The bridge is wedged and every
// subsequent read fails identically, so shrinking + skipping past it
// — even under `skip_errors` — just marches the whole disc at one
// ~15s bridge-recovery per probe, producing no usable output (the
// "runs forever, no MKV" report). Abort immediately, highest
// priority, mirroring the multipass sweep's transport-failure rule
// in `read_error::handle_read_error`. The CLI/UX surfaces this so the
// user power-cycles the drive (or switches to multipass recovery).
if let Some(e) = res.as_ref().err()
&& e.is_scsi_transport_failure()
{
let (status, sense) = extract_scsi_context(e);
return Err(crate::error::Error::DiscRead {
sector: lba as u64,
status: Some(status),
sense,
}
.into());
}
if (sectors as u32) <= align {
// Bottomed out at one unit (AACS) / one sector (CSS) / the
// extent tail. This is single-pass disc→MKV, which has NO Pass N
// to come back and recover later — so before we skip or bail,
// give the drive its full ECC recovery budget ONCE
// (`recovery=true` → READ_RECOVERY_TIMEOUT_MS, ~60s), exactly as
// the multipass patch does on its bad ranges. A single bounded
// read, never a loop (hard rule #2: tight retry loops on one LBA
// push the BU40N into fast-fail). On success we USE the recovered
// data, so the old "transient retry returns a bogus status for
// readable data" hole cannot reopen; the earlier 10s-timeout read
// gave the drive no chance to recover a marginal sector that a
// 60s ECC read can.
tracing::debug!(
target: "mux",
"fill_extents last-chance recovery read at LBA {} ({} sectors, 60s ECC)",
lba,
sectors
);
let rec = self
.reader
.read_sectors(lba, sectors, &mut self.read_buf[..bytes], true);
if let Ok(&got) = rec.as_ref() {
debug_assert!(got <= bytes, "recovery read over-reported byte count");
if let Some(ev) = self.adaptive.on_success(sectors) {
self.emit(ev);
}
let got = got.min(bytes);
// Same short-read rule as the first-attempt read above —
// one shared commit, so the two cannot drift apart.
self.commit_read(lba, got, bytes)?;
break;
}
// Recovery read also failed. A transport failure here (status
// 0xFF: USB-bridge crash / disconnect) is NOT a skippable bad
// unit — same as the original 10s read above. The line-442
// short-circuit only inspected `res`; the 60s recovery read
// (`rec`) can wedge the bridge on its own, and falling into the
// `skip_errors` branch below would zero-fill + advance, treating
// a dead bridge as a skippable unit and marching the whole disc
// at one bridge-recovery per probe (hard rule #2, "runs forever,
// no MKV"). Re-check `rec` and abort, mirroring line 442.
if let Some(e) = rec.as_ref().err()
&& e.is_scsi_transport_failure()
{
let (status, sense) = extract_scsi_context(e);
return Err(crate::error::Error::DiscRead {
sector: lba as u64,
status: Some(status),
sense,
}
.into());
}
// Recovery read also failed. Skip the WHOLE failed unit or bail.
// Zero-filling and advancing by the full unit keeps
// current_offset unit-aligned, so the next read still begins on a
// real AACS unit boundary (a 1-sector skip here would desync the
// rest of the title — the bug this guards).
if self.skip_errors {
let zb = sectors as usize * 2048;
self.read_buf.resize(zb, 0);
self.read_buf[..zb].fill(0);
self.buf_valid = zb;
self.errors += 1;
// `errors` counts skip events; `lost_bytes` counts the
// bytes actually zero-filled. For AACS (unit_align=3) a
// single event skips a whole 6144-byte unit, so loss
// estimates must use this, not `errors * 2048`.
self.lost_bytes = self.lost_bytes.saturating_add(zb as u64);
self.emit(EventKind::SectorSkipped { sector: lba as u64 });
self.current_offset += sectors as u32;
break;
} else {
// Build the error from the recovery failure we now hold
// (falling back to the original 10s-read failure).
let err = rec.err().or(res.err());
let (status, sense) =
err.as_ref().map(extract_scsi_context).unwrap_or((0, None));
return Err(crate::error::Error::DiscRead {
sector: lba as u64,
status: Some(status),
sense,
}
.into());
}
}
// Shrink and retry at the same LBA with a smaller batch.
if let Some(ev) = self.adaptive.on_failure() {
self.emit(ev);
}
}
if self.current_offset >= ext_sectors {
self.current_extent += 1;
self.current_offset = 0;
}
Ok(true)
}
}
/// Per-stage profiling state — populated only when `FREEMKV_PROFILE`
/// is set. Logs a percentage breakdown via `tracing` (target "mux")
/// every [`PROFILE_INTERVAL`]. Zero overhead in normal runs (the
/// `DiscStream::profiling` check is the only added cost).
struct StageProf {
started: std::time::Instant,
last_dump: std::time::Instant,
fill_ns: u128,
feed_ns: u128,
consume_ns: u128,
bytes_in: u64,
}
const PROFILE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5);
thread_local! {
static STAGE_PROF: std::cell::RefCell<Option<StageProf>> = const { std::cell::RefCell::new(None) };
}
fn prof_active() -> bool {
std::env::var_os("FREEMKV_PROFILE").is_some()
}
fn prof_tick(stage: &str, ns: u128, bytes: u64) {
STAGE_PROF.with(|cell| {
let mut slot = cell.borrow_mut();
if slot.is_none() {
if !prof_active() {
return;
}
let now = std::time::Instant::now();
*slot = Some(StageProf {
started: now,
last_dump: now,
fill_ns: 0,
feed_ns: 0,
consume_ns: 0,
bytes_in: 0,
});
}
let p = slot.as_mut().unwrap();
match stage {
"fill" => p.fill_ns += ns,
"feed" => p.feed_ns += ns,
"consume" => p.consume_ns += ns,
_ => {}
}
p.bytes_in += bytes;
let now = std::time::Instant::now();
if now.duration_since(p.last_dump) < PROFILE_INTERVAL {
return;
}
let elapsed_ms = now.duration_since(p.started).as_millis().max(1);
let fill_pct = p.fill_ns / 10_000 / elapsed_ms;
let feed_pct = p.feed_ns / 10_000 / elapsed_ms;
let consume_pct = p.consume_ns / 10_000 / elapsed_ms;
let mbps = p.bytes_in as u128 * 1000 / 1_000_000 / elapsed_ms;
tracing::debug!(
target: "mux",
"[profile] elapsed={}ms in={}MB/s fill={}% feed={}% consume={}%",
elapsed_ms, mbps, fill_pct, feed_pct, consume_pct,
);
p.last_dump = now;
});
}
impl crate::pes::Stream for DiscStream {
fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
if let Some(frame) = self.pending_frames.pop_front() {
return Ok(Some(frame));
}
if self.eof {
return Ok(None);
}
loop {
// Profiling timestamps only when FREEMKV_PROFILE is set; otherwise
// these stay None and no Instant::now() is taken in the hot loop.
let t0 = self.profiling.then(std::time::Instant::now);
if !self.fill_extents()? {
self.eof = true;
if self.dropped_nav_packets > 0 {
tracing::debug!(
target: "mux",
"dropped {} DVD navigation packets (private_stream_2/0xBF) — expected, carry no elementary stream",
self.dropped_nav_packets
);
}
// Flush demuxer — last PES packet may still be in the assembler
if let Some(ref mut demuxer) = self.ts_demuxer {
for pes in &demuxer.flush() {
if let Some((_, track)) =
self.pid_to_track.iter().find(|(pid, _)| *pid == pes.pid)
&& let Some((_, parser)) =
self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid)
{
let resync = &mut self.resync;
let is_video = &self.is_video;
let pending = &mut self.pending_frames;
for frame in parser.parse(pes) {
// Same B1 gate — a concealed gap can leave a
// post-gap frame in the demuxer's final flush.
let emit = match resync.get_mut(*track) {
Some(gate) => gate.admit(
is_video.get(*track).copied().unwrap_or(false),
frame.discontinuity,
frame.keyframe,
),
None => true,
};
if emit {
pending.push_back(crate::pes::PesFrame::from_codec_frame(
*track, frame,
));
}
}
}
}
}
// PS demuxer flush (DVD)
if let Some(ref mut demuxer) = self.ps_demuxer {
for ps in &demuxer.flush() {
// Route by the REAL DVD PID (see consume_ps in
// pipelined_stream.rs); the old (sub_id & 0x1F)+1
// heuristic mis-routed VobSub into the AC-3 parser.
let Some(pid) = ps.dvd_pid() else {
if ps.is_nav() {
// Expected DVD navigation packet (PCI/DSI) —
// tally, no WARN.
self.dropped_nav_packets += 1;
} else {
// Unexpected unmappable stream_id (a
// possibly-dropped real stream). Keep the WARN.
tracing::warn!(
target: "mux",
"dropping unmappable PS packet (stream_id={:#04x}, sub_stream_id={:?})",
ps.stream_id,
ps.sub_stream_id,
);
}
continue;
};
let Some((_, track)) =
self.pid_to_track.iter().find(|(p, _)| *p == pid).copied()
else {
tracing::warn!(
target: "mux",
"dropping PS packet for unmapped PID {:#06x} (stream_id={:#04x}, sub_stream_id={:?})",
pid,
ps.stream_id,
ps.sub_stream_id,
);
continue;
};
let pes = super::ts::PesPacket {
source: None,
pid,
pts: ps.pts.map(|p| p as i64),
dts: ps.dts.map(|d| d as i64),
data: ps.data.clone(),
// PS (DVD/CSS) path: no AACS conceal, no gap flag.
discontinuity: false,
};
if let Some((_, parser)) = self.parsers.iter_mut().find(|(p, _)| *p == pid)
{
for frame in parser.parse(&pes) {
self.pending_frames.push_back(
crate::pes::PesFrame::from_codec_frame(track, frame),
);
}
}
}
}
// Drain any access unit a codec parser buffered past the last
// PES (DTS-HD's final core+extension unit, assembled across
// PES boundaries).
let pid_to_track = &self.pid_to_track;
let pending = &mut self.pending_frames;
let resync = &mut self.resync;
let is_video = &self.is_video;
for (pid, parser) in self.parsers.iter_mut() {
let Some(&(_, track)) = pid_to_track.iter().find(|(p, _)| p == pid) else {
continue;
};
// Flush frames carry their own per-frame `discontinuity` (a
// post-gap picture buffered at EOF was stamped by the parser),
// so route them through the SAME B1 gate the in-stream path
// uses — otherwise a trailing dangling-reference frame would
// bypass the resync.
for frame in parser.flush() {
let emit = match resync.get_mut(track) {
Some(gate) => gate.admit(
is_video.get(track).copied().unwrap_or(false),
frame.discontinuity,
frame.keyframe,
),
None => true,
};
if emit {
pending.push_back(crate::pes::PesFrame::from_codec_frame(track, frame));
}
}
}
// A gate still armed at EOF dropped post-gap frames that never
// reached a keyframe (a concealed gap in the final GOP). Surface
// it once so the loss is visible, not silent.
for (track, gate) in self.resync.iter().enumerate() {
if gate.is_armed() {
tracing::warn!(
target: "mux",
track,
dropped = gate.dropped_in_run(),
"B1: stream ended while dropping to a keyframe after a concealed gap (no trailing keyframe)"
);
}
}
return Ok(self.pending_frames.pop_front());
}
let bytes = self.buf_valid;
let t1 = self.profiling.then(std::time::Instant::now);
if let (Some(t0), Some(t1)) = (t0, t1) {
prof_tick("fill", t1.duration_since(t0).as_nanos(), bytes as u64);
}
// Plaintext: the wrapped reader (DecryptingSectorSource)
// applied AACS / CSS in-place during fill_extents'
// read_sectors call. The pre-0.18 inline decrypt step
// lived here.
if let Some(ref mut demuxer) = self.ts_demuxer {
let packets = demuxer.feed(&self.read_buf[..bytes]);
let t2 = self.profiling.then(std::time::Instant::now);
if let (Some(t1), Some(t2)) = (t1, t2) {
prof_tick("feed", t2.duration_since(t1).as_nanos(), 0);
}
let skip_parse = self.skip_parse;
for pes in packets {
if let Some((_, track)) = self
.pid_to_track
.iter()
.find(|(pid, _)| *pid == pes.pid)
.copied()
{
if skip_parse {
// Profiling escape hatch — bypass the codec
// parser and pass the raw PES bytes straight
// through as a single PesFrame. Lets us
// attribute consumer-thread time to
// "demux + framing" vs "codec parse".
self.pending_frames.push_back(crate::pes::PesFrame {
coding: None,
source: None,
track,
pts: pes.pts.map(super::codec::pts_to_ns).unwrap_or(0),
keyframe: false,
data: pes.data,
duration_ns: None,
});
} else if let Some((_, parser)) =
self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid)
{
// Disjoint field borrows: the resync gate + is_video
// reads coexist with the mutable `self.parsers` drain.
let resync = &mut self.resync;
let is_video = &self.is_video;
let pending = &mut self.pending_frames;
for frame in parser.parse(&pes) {
// B1: after a concealed/lost gap, drop forward to
// the next keyframe on a video track so no frame
// with a dangling reference is emitted.
let emit = match resync.get_mut(track) {
Some(gate) => gate.admit(
is_video.get(track).copied().unwrap_or(false),
frame.discontinuity,
frame.keyframe,
),
None => true,
};
if emit {
pending.push_back(crate::pes::PesFrame::from_codec_frame(
track, frame,
));
}
}
}
}
}
let t3 = self.profiling.then(std::time::Instant::now);
if let (Some(t2), Some(t3)) = (t2, t3) {
prof_tick("consume", t3.duration_since(t2).as_nanos(), 0);
}
} else if let Some(ref mut demuxer) = self.ps_demuxer {
let packets = demuxer.feed(&self.read_buf[..bytes]);
for ps in &packets {
// Route by the REAL DVD PID (see consume_ps in
// pipelined_stream.rs); the old (sub_id & 0x1F)+1
// heuristic mis-routed VobSub into the AC-3 parser.
let Some(pid) = ps.dvd_pid() else {
if ps.is_nav() {
// Expected DVD navigation packet (PCI/DSI) — tally,
// no WARN.
self.dropped_nav_packets += 1;
} else {
// Unexpected unmappable stream_id (a possibly-dropped
// real stream). Keep the individual WARN.
tracing::warn!(
target: "mux",
"dropping unmappable PS packet (stream_id={:#04x}, sub_stream_id={:?})",
ps.stream_id,
ps.sub_stream_id,
);
}
continue;
};
let Some((_, track)) =
self.pid_to_track.iter().find(|(p, _)| *p == pid).copied()
else {
tracing::warn!(
target: "mux",
"dropping PS packet for unmapped PID {:#06x} (stream_id={:#04x}, sub_stream_id={:?})",
pid,
ps.stream_id,
ps.sub_stream_id,
);
continue;
};
let pes = super::ts::PesPacket {
source: None,
pid,
pts: ps.pts.map(|p| p as i64),
dts: ps.dts.map(|d| d as i64),
data: ps.data.clone(),
// PS (DVD/CSS) path: no AACS conceal, no gap flag.
discontinuity: false,
};
if let Some((_, parser)) = self.parsers.iter_mut().find(|(p, _)| *p == pid) {
for frame in parser.parse(&pes) {
self.pending_frames
.push_back(crate::pes::PesFrame::from_codec_frame(track, frame));
}
}
}
}
self.buf_valid = 0;
if let Some(frame) = self.pending_frames.pop_front() {
return Ok(Some(frame));
}
}
}
fn write(&mut self, _frame: &crate::pes::PesFrame) -> io::Result<()> {
Err(crate::error::Error::StreamReadOnly.into())
}
fn finish(&mut self) -> io::Result<()> {
Ok(())
}
fn info(&self) -> &DiscTitle {
&self.title
}
fn codec_private(&self, track: usize) -> Option<Vec<u8>> {
let pid = self
.pid_to_track
.iter()
.find(|(_, idx)| *idx == track)
.map(|(pid, _)| *pid)?;
self.parsers
.iter()
.find(|(p, _)| *p == pid)
.and_then(|(_, parser)| parser.codec_private())
}
fn headers_ready(&self) -> bool {
// FREEMKV_SKIP_PARSE bypasses codec parsers entirely for
// bottleneck profiling, so codec_private is never populated.
// Pretend headers are ready immediately in that mode so the
// CLI loop doesn't hang waiting for them.
if self.skip_parse {
return true;
}
for (idx, s) in self.title.streams.iter().enumerate() {
if let crate::disc::Stream::Video(v) = s
&& !v.secondary
&& self.codec_private(idx).is_none()
{
return false;
}
}
true
}
fn errors(&self) -> u64 {
// Read errors PLUS every frame the B1 resync gates discarded.
//
// Summed from the gates here rather than counted at each of the three
// `admit` call sites: one place cannot drift, three can, and a frame
// dropped by the demuxer flush is the same loss as one dropped on the
// main path. `ResyncGate::dropped` is zeroed at every resync, so only
// `dropped_total` sees a gap that RESOLVES — which is most of them, and
// was the whole reason concealed video loss reached no caller.
self.errors
+ self
.resync
.iter()
.map(super::resync::ResyncGate::dropped_total)
.sum::<u64>()
}
fn lost_bytes(&self) -> u64 {
// Read-error zero-fill loss (counted in fill_extents) — real missing
// content the abort gate must see. There is no decrypt-loss term: the
// decrypt path passes bad-encoded/undecryptable units through (a broken-TS
// unit is the muxer's concern, and a missing key is indistinguishable from
// bad authoring here), so only physical read loss is reported.
self.lost_bytes
}
}
#[cfg(test)]
mod tests {
//! `DiscStream` is the only read-only `Stream` impl in tree (every
//! other concrete impl in `mux/*` is bidirectional or write-only).
//! These tests lock down a static `Send` assertion plus a
//! `Box<dyn Stream>` round trip exercising every method through the
//! trait object, so future Send-breaking edits to `DiscStream`'s
//! interior types fail at compile time.
use super::*;
use crate::disc::{ContentFormat, DiscTitle};
use crate::pes::Stream;
/// Static-assert `DiscStream: Send`. The `Stream` trait has `Send` as a
/// supertrait — if a future field on `DiscStream` is non-`Send` (e.g.
/// a `Box<dyn Read>` instead of `Box<dyn SectorSource>`), this fails
/// at compile time, before the runtime trait-object test below.
fn _assert_disc_stream_is_send() {
fn requires_send<T: Send>() {}
requires_send::<DiscStream>();
}
/// Trivial `SectorSource` that yields zeroed sectors. Empty title means
/// the demuxer produces no PES frames, so `read()` walks the extents to
/// EOF and returns `Ok(None)`. That's enough to exercise the trait-object
/// dispatch — the goal here is the bridge, not the demuxer.
struct ZeroReader {
capacity: u32,
}
impl crate::sector::SectorSource for ZeroReader {
fn read_sectors(
&mut self,
_lba: u32,
count: u16,
buf: &mut [u8],
_recovery: bool,
) -> crate::error::Result<usize> {
let bytes = count as usize * 2048;
buf[..bytes].fill(0);
Ok(bytes)
}
fn capacity_sectors(&self) -> u32 {
self.capacity
}
}
/// A `SectorSource` that UNDER-reports: it writes the whole requested span
/// (leaving the tail as stale bytes a real short read would also leave) but
/// reports only the first sector as valid. Every in-tree source is
/// full-or-error; this models one that is not, which is exactly the case
/// `fill_extents` handled inconsistently — trusting the short count for
/// `buf_valid` while still advancing the extent cursor by the full request.
struct ShortReader {
capacity: u32,
}
impl crate::sector::SectorSource for ShortReader {
fn read_sectors(
&mut self,
_lba: u32,
count: u16,
buf: &mut [u8],
_recovery: bool,
) -> crate::error::Result<usize> {
let bytes = count as usize * 2048;
// Stale marker across the whole span; only the first sector is
// reported as actually delivered.
buf[..bytes].fill(0xAB);
Ok(2048usize.min(bytes))
}
fn capacity_sectors(&self) -> u32 {
self.capacity
}
}
fn short_read_stream(skip_errors: bool) -> DiscStream {
let mut s = DiscStream::new(
Box::new(ShortReader { capacity: 64 }),
synthetic_title(64),
crate::decrypt::DecryptKeys::None,
8, // request 8 sectors (16384 B); the source delivers 1 (2048 B)
ContentFormat::BdTs,
false,
None,
)
.unwrap();
s.skip_errors = skip_errors;
s
}
/// A short read must never become a SILENT gap. `buf_valid` trusts the
/// returned byte count, so advancing `current_offset` by the full requested
/// sector count drops the undelivered tail out of the muxed title with no
/// error, no skip event, and no `lost_bytes` — data loss invisible to the
/// caller and to the progress accounting.
///
/// Without `skip_errors` the caller has NOT opted into holes, so it is a
/// hard read error: numeric code E6000 (`Error::DiscRead`).
#[test]
fn short_read_without_skip_errors_is_reported_not_silently_skipped() {
let mut s = short_read_stream(false);
let err = s
.fill_extents()
.expect_err("a short read must not be reported as a clean fill");
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
assert!(
err.to_string().contains("E6000"),
"expected the disc-read code E6000, got {err}"
);
}
/// With `skip_errors` the caller HAS opted into holes, so the undelivered
/// tail is zero-filled and ACCOUNTED — never left as stale buffer bytes and
/// never dropped silently. `current_offset` still advances by the full
/// request so it stays on an AACS unit boundary (resuming mid-unit desyncs
/// the rest of the title, per the failed-unit skip branch).
#[test]
fn short_read_with_skip_errors_is_zero_filled_and_accounted() {
let mut s = short_read_stream(true);
assert!(
s.fill_extents()
.expect("skip_errors absorbs the short read")
);
// 8 sectors requested at 2048 B/sector (the ECMA-167 / UDF logical
// sector size) = 16384 B of buffer; the source delivered 2048 B.
assert_eq!(s.buf_valid, 16_384, "the whole requested span stays valid");
assert_eq!(s.current_offset, 8, "the cursor advances a whole unit span");
assert_eq!(
s.lost_bytes, 14_336,
"16384 requested - 2048 delivered = 14336 lost bytes must be counted"
);
assert_eq!(s.errors, 1, "the gap is counted as one skip event");
// The delivered sector survives; the undelivered tail is zeroed, not
// the stale 0xAB the source left behind.
assert!(s.read_buf[..2048].iter().all(|&b| b == 0xAB));
assert!(
s.read_buf[2048..16_384].iter().all(|&b| b == 0),
"the undelivered tail must be zero-filled, not stale bytes"
);
}
/// Nothing filters `sector_count == 0` out of a UDF/MPLS extent list, so a
/// malformed disc can declare a long run of empty extents. `fill_extents`
/// used to skip each one with a self-recursive call, costing a stack frame
/// apiece; Rust does not guarantee tail-call elimination, so a few thousand
/// of them overflowed the stack. A stack overflow aborts the process — it
/// is not catchable and takes the whole `autorip` service down — where the
/// iterative form just walks off the end and reports EOF.
///
/// Run on a deliberately small stack so the frame cost is unmissable: the
/// recursive version dies here, the loop finishes in microseconds.
#[test]
fn a_long_run_of_empty_extents_does_not_recurse_per_extent() {
let title = DiscTitle {
extents: (0..50_000)
.map(|_| crate::disc::Extent {
start_lba: 0,
sector_count: 0,
})
.collect(),
..DiscTitle::empty()
};
let handle = std::thread::Builder::new()
.stack_size(256 * 1024)
.spawn(move || {
let mut s = DiscStream::new(
Box::new(ZeroReader { capacity: 0 }),
title,
crate::decrypt::DecryptKeys::None,
8,
ContentFormat::BdTs,
false,
None,
)
.unwrap();
s.fill_extents()
})
.unwrap();
assert!(
!handle.join().unwrap().unwrap(),
"an all-empty extent list is EOF, not an error"
);
}
fn synthetic_title(sector_count: u32) -> DiscTitle {
DiscTitle {
extents: vec![crate::disc::Extent {
start_lba: 0,
sector_count,
}],
..DiscTitle::empty()
}
}
/// Smallest credible witness that `DiscStream` flows through `dyn Stream`:
/// build a `Box<dyn Stream>`, drive `read()` to EOF, exercise `info()` /
/// `headers_ready()` / `codec_private()` through the trait object.
#[test]
fn stream_via_dyn_object() {
let reader = ZeroReader { capacity: 8 };
let title = synthetic_title(8);
let stream = DiscStream::new(
Box::new(reader),
title,
crate::decrypt::DecryptKeys::None,
8,
ContentFormat::BdTs,
false,
None,
)
.unwrap();
let mut src: Box<dyn Stream> = Box::new(stream);
// Empty-title fixture has no streams configured, so headers are
// trivially ready and codec_private() yields nothing on track 0.
assert!(src.headers_ready());
assert!(src.codec_private(0).is_none());
let _ = src.info();
// Drive read() to EOF through the trait object — empty-title fixture
// produces no frames, but the call still routes through the blanket
// dispatch into Stream::read.
let mut frames = 0usize;
while src.read().expect("read").is_some() {
frames += 1;
if frames > 1024 {
panic!("unexpected unbounded frame stream from empty title");
}
}
assert_eq!(frames, 0);
}
/// A Blu-ray 3D title: an AVC base view plus the MVC dependent (right-eye)
/// view, marked by `MVC_DEPENDENT_LABEL`.
fn mvc_title() -> DiscTitle {
use crate::disc::{Codec, ColorSpace, FrameRate, HdrFormat, Resolution, VideoStream};
let view = |pid: u16, label: &str| {
crate::disc::Stream::Video(VideoStream {
pid,
codec: Codec::H264,
resolution: Resolution::R1080p,
frame_rate: FrameRate::F23_976,
hdr: HdrFormat::Sdr,
color_space: ColorSpace::Bt709,
display_aspect: None,
secondary: false,
label: label.to_string(),
measured_cicp: None,
})
};
let mut t = synthetic_title(8);
t.streams = vec![
view(0x1011, ""), // base view
view(0x1012, crate::disc::MVC_DEPENDENT_LABEL), // dependent view
];
t.content_format = ContentFormat::BdTs;
t
}
/// A dependent-view access unit: PPS (NAL 8) + coded-slice-extension (NAL 20),
/// no IDR. The base-view parser strips the PPS from a non-keyframe AU; the
/// MVC passthrough parser keeps every parameter set in-band so each dependent
/// frame is a self-contained access unit (ISO/IEC 14496-10 Annex H).
fn mvc_dependent_au_pes(pid: u16) -> crate::mux::ts::PesPacket {
let nal = |t: u8, body: &[u8]| {
let mut v = vec![0x00, 0x00, 0x01, t];
v.extend_from_slice(body);
v
};
let mut data = Vec::new();
data.extend_from_slice(&nal(0x68, &[0xCE, 0x01])); // PPS (8)
data.extend_from_slice(&nal(0x74, &[0x11, 0x22])); // slice extension (20)
crate::mux::ts::PesPacket {
source: None,
pid,
pts: Some(90_000),
dts: None,
data,
discontinuity: false,
}
}
/// Whether a parser's output for a dependent-view AU still carries the PPS —
/// the observable signature of the MVC param-set-passthrough parser.
fn keeps_pps_inband(parser: &mut dyn super::super::codec::CodecParser, pid: u16) -> bool {
let frames = parser.parse(&mvc_dependent_au_pes(pid));
assert_eq!(frames.len(), 1, "one access unit in, one frame out");
// Frame payload is length-prefixed NALs (4-byte BE length + NAL).
let d = &frames[0].data;
let mut i = 0;
while i + 4 <= d.len() {
let len = u32::from_be_bytes([d[i], d[i + 1], d[i + 2], d[i + 3]]) as usize;
i += 4;
if i + len > d.len() {
break;
}
if len > 0 && d[i] & 0x1F == 8 {
return true;
}
i += len;
}
false
}
/// The LIVE `disc://` path (`DiscStream::new`) must dispatch the Blu-ray 3D
/// MVC dependent view to the param-set-passthrough parser, exactly as the
/// file-backed ISO path (`resolve::build_demux_state`) does. Before this was
/// shared, the live path built every parser with plain `parser_for_codec`, so
/// the same 3D disc muxed correctly from an ISO and incorrectly from a drive.
#[test]
fn live_path_dispatches_mvc_dependent_view_to_passthrough_parser() {
let title = mvc_title();
let mut stream = DiscStream::new(
Box::new(ZeroReader { capacity: 8 }),
title.clone(),
crate::decrypt::DecryptKeys::None,
8,
ContentFormat::BdTs,
false,
None,
)
.unwrap();
let idx_of = |parsers: &Vec<(u16, Box<dyn super::super::codec::CodecParser>)>, pid: u16| {
parsers.iter().position(|(p, _)| *p == pid).unwrap()
};
// Live path: dependent view keeps its PPS in-band, base view does not.
let dep = idx_of(&stream.parsers, 0x1012);
assert!(
keeps_pps_inband(stream.parsers[dep].1.as_mut(), 0x1012),
"live disc:// path must give the MVC dependent view the passthrough parser"
);
let base = idx_of(&stream.parsers, 0x1011);
assert!(
!keeps_pps_inband(stream.parsers[base].1.as_mut(), 0x1011),
"base view keeps the ordinary parser (discriminator is real, not vacuous)"
);
// And it must agree with the ISO path, which is the canonical builder.
let (mut iso_parsers, _, _, _) =
crate::mux::resolve::build_demux_state(&title, ContentFormat::BdTs);
let iso_dep = idx_of(&iso_parsers, 0x1012);
assert!(
keeps_pps_inband(iso_parsers[iso_dep].1.as_mut(), 0x1012),
"ISO path reference behaviour"
);
}
/// `is_halted()` must observe a cancellation signal installed via
/// `with_halt(Halt)` — flipping the token must cause the next
/// `fill_extents` retry boundary to bail.
#[test]
fn halt_via_with_halt_observed_by_is_halted() {
let halt = Halt::new();
let stream = DiscStream::new(
Box::new(ZeroReader { capacity: 8 }),
synthetic_title(8),
crate::decrypt::DecryptKeys::None,
8,
crate::disc::ContentFormat::BdTs,
false,
None,
)
.unwrap()
.with_halt(halt.clone());
assert!(!stream.is_halted());
halt.cancel();
assert!(
stream.is_halted(),
"with_halt token cancellation must be observed by is_halted()"
);
}
/// `with_key_map` on the inline live-drive path applies the same FMTS read plan
/// the file-backed highway uses: within a forensic segment only our-phase units
/// survive the extent walk, so the alternate device-group units are never read.
#[test]
fn with_key_map_reads_only_our_phase_units() {
use crate::decrypt::{AacsKeyMap, DecryptKeys, Phase};
// AACS keys → unit_align = 3, so a unit is 3 sectors and the phase filter
// engages. Key contents are irrelevant to the read plan.
let aacs = DecryptKeys::Aacs {
unit_keys: vec![(0, [0u8; 16]), (1, [1u8; 16])],
read_data_key: None,
format: ContentFormat::BdTs,
};
// 100 units (300 sectors). A 10-unit Even forensic segment at LBA [30,60):
// even units (30,36,42,48,54) are ours; odd (33,39,45,51,57) are dropped.
let map = AacsKeyMap::from_ranges_phased(vec![(30, 60, 1, Phase::Even)]);
let stream = DiscStream::new(
Box::new(ZeroReader { capacity: 300 }),
synthetic_title(300),
aacs,
8,
ContentFormat::BdTs,
false,
None,
)
.unwrap()
.with_key_map(std::sync::Arc::new(map));
let total: u32 = stream.extents.iter().map(|e| e.sector_count).sum();
assert_eq!(
total,
300 - 5 * 3,
"exactly the 5 alternate-phase units (15 sectors) are dropped from the read walk"
);
assert!(
stream.extents.len() > 1,
"the forensic segment split the single extent into our-phase-only runs"
);
// The progress denominator tracks the reduced read set.
assert_eq!(stream.bytes_total_extents, total as u64 * 2048);
}
/// Recording `SectorSource`: logs every `(lba, count)` request and
/// returns `Err` whenever the requested range covers `bad_sector`.
/// Successful reads return zeroed sectors (which the content-clarity check
/// does not flag as scrambled, so `DecryptingSectorSource` passes them through
/// even with synthetic AACS keys — no real decrypt is attempted).
struct RecordingReader {
capacity: u32,
bad_sector: u32,
log: std::sync::Arc<std::sync::Mutex<Vec<(u32, u16)>>>,
}
impl crate::sector::SectorSource for RecordingReader {
fn read_sectors(
&mut self,
lba: u32,
count: u16,
buf: &mut [u8],
_recovery: bool,
) -> crate::error::Result<usize> {
self.log.lock().unwrap().push((lba, count));
let end = lba + count as u32;
if self.bad_sector >= lba && self.bad_sector < end {
return Err(crate::error::Error::DiscRead {
sector: self.bad_sector as u64,
status: Some(0x02),
sense: None,
});
}
let bytes = count as usize * 2048;
buf[..bytes].fill(0);
Ok(bytes)
}
fn capacity_sectors(&self) -> u32 {
self.capacity
}
}
/// `SectorSource` that fails every read covering `bad_sector` with a
/// SCSI **transport failure** (status=0xFF) — the USB-bridge-crash sentinel
/// that `Drive::read` surfaces as `DiscRead { status: Some(0xFF), .. }`.
/// Logs each `(lba, count)` so a test can prove the failure was not
/// retried/skipped.
struct TransportFailReader {
capacity: u32,
bad_sector: u32,
log: std::sync::Arc<std::sync::Mutex<Vec<(u32, u16)>>>,
}
impl crate::sector::SectorSource for TransportFailReader {
fn read_sectors(
&mut self,
lba: u32,
count: u16,
buf: &mut [u8],
_recovery: bool,
) -> crate::error::Result<usize> {
self.log.lock().unwrap().push((lba, count));
let end = lba + count as u32;
if self.bad_sector >= lba && self.bad_sector < end {
return Err(crate::error::Error::DiscRead {
sector: self.bad_sector as u64,
status: Some(crate::scsi::SCSI_STATUS_TRANSPORT_FAILURE),
sense: None,
});
}
let bytes = count as usize * 2048;
buf[..bytes].fill(0);
Ok(bytes)
}
fn capacity_sectors(&self) -> u32 {
self.capacity
}
}
/// `SectorSource` that mirrors a marginal sector recoverable only with the
/// drive's full ECC budget: every read covering `bad_sector` FAILS while
/// `recovery=false` (the fast 10s pass) and SUCCEEDS (zeroed bytes) once
/// `recovery=true` (the 60s ECC pass). Drives the single-pass bottom-out
/// "last-chance recovery read" success branch in `fill_extents`, which the
/// other test sources (ignoring the flag) never exercise.
struct RecoverableReader {
capacity: u32,
bad_sector: u32,
/// `(lba, count, recovery)` for every issued read.
log: std::sync::Arc<std::sync::Mutex<Vec<(u32, u16, bool)>>>,
}
impl crate::sector::SectorSource for RecoverableReader {
fn read_sectors(
&mut self,
lba: u32,
count: u16,
buf: &mut [u8],
recovery: bool,
) -> crate::error::Result<usize> {
self.log.lock().unwrap().push((lba, count, recovery));
let end = lba + count as u32;
let covers_bad = self.bad_sector >= lba && self.bad_sector < end;
// Fail on the fast (non-recovery) pass; the 60s ECC recovery read
// succeeds. Distinct non-0x02 sense byte so a transport-failure
// re-check (status 0xFF) is provably NOT triggered here.
if covers_bad && !recovery {
return Err(crate::error::Error::DiscRead {
sector: self.bad_sector as u64,
status: Some(0x02),
sense: None,
});
}
let bytes = count as usize * 2048;
buf[..bytes].fill(0);
Ok(bytes)
}
fn capacity_sectors(&self) -> u32 {
self.capacity
}
}
/// Coverage for the single-pass bottom-out RECOVERY-READ SUCCESS branch
/// (rc.5.2 audit #3): a sector that fails the fast 10s read but reads clean
/// on the 60s ECC recovery read must have its RECOVERED data muxed — the
/// cursor advances over the whole unit, byte counters move, and NO skip is
/// counted. Pre-fix the test sources ignored `recovery`, so this branch was
/// untested. Uses `unit_align=1` (None) so the bottom-out unit is a single
/// sector, exercising the `(sectors as u32) <= align` path precisely.
#[test]
fn recovery_read_success_muxes_recovered_data_no_skip() {
const COUNT: u32 = 10;
let bad = 4u32;
let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let reader = RecoverableReader {
capacity: COUNT,
bad_sector: bad,
log: log.clone(),
};
let mut stream = DiscStream::new(
Box::new(reader),
synthetic_title(COUNT),
crate::decrypt::DecryptKeys::None,
8,
ContentFormat::BdTs,
false,
None,
)
.unwrap();
// skip_errors=false: if the recovery read did NOT succeed, fill_extents
// would return Err — so reaching EOF cleanly proves recovery worked.
stream.skip_errors = false;
let mut guard = 0;
loop {
match stream.fill_extents() {
Ok(true) => {}
Ok(false) => break,
Err(e) => panic!("recovery read should have succeeded, got: {e}"),
}
guard += 1;
assert!(guard < 1000, "fill_extents did not reach EOF");
}
// No skip counted: the recovered unit was muxed, not zero-filled.
assert_eq!(
stream.errors, 0,
"a successful recovery read must not count as a skipped sector"
);
assert_eq!(
stream.lost_bytes, 0,
"a successful recovery read loses no bytes"
);
// All COUNT sectors' worth of bytes were read through to the cursor end.
assert_eq!(
stream.bytes_read_total,
COUNT as u64 * 2048,
"every sector (including the recovered one) must be counted as read"
);
// The bad sector was retried with recovery=true and that read SUCCEEDED.
let reads = log.lock().unwrap();
assert!(
reads
.iter()
.any(|&(lba, count, rec)| rec && lba == bad && count == 1),
"expected a recovery=true single-sector read at the bad sector; got {reads:?}"
);
// And the fast pass at the bad sector did happen with recovery=false.
assert!(
reads.iter().any(|&(lba, _c, rec)| !rec && lba == bad),
"expected a non-recovery read to have first failed at the bad sector"
);
}
/// `SectorSource` that fails the fast (non-recovery) read covering
/// `bad_sector` with an ordinary bad-sector error (status 0x02), then fails
/// the 60s ECC recovery read with a TRANSPORT failure (status 0xFF). Models
/// a bridge that wedges precisely during the last-chance recovery read.
struct RecoveryTransportFailReader {
capacity: u32,
bad_sector: u32,
log: std::sync::Arc<std::sync::Mutex<Vec<(u32, u16, bool)>>>,
}
impl crate::sector::SectorSource for RecoveryTransportFailReader {
fn read_sectors(
&mut self,
lba: u32,
count: u16,
buf: &mut [u8],
recovery: bool,
) -> crate::error::Result<usize> {
self.log.lock().unwrap().push((lba, count, recovery));
let end = lba + count as u32;
if self.bad_sector >= lba && self.bad_sector < end {
let status = if recovery {
crate::scsi::SCSI_STATUS_TRANSPORT_FAILURE
} else {
0x02
};
return Err(crate::error::Error::DiscRead {
sector: self.bad_sector as u64,
status: Some(status),
sense: None,
});
}
let bytes = count as usize * 2048;
buf[..bytes].fill(0);
Ok(bytes)
}
fn capacity_sectors(&self) -> u32 {
self.capacity
}
}
/// Regression (rc.5.2 audit #2): a transport failure on the 60s ECC
/// RECOVERY read (not just the initial 10s read) must ABORT, even under
/// `skip_errors=true`. The line-442 short-circuit only inspected the
/// original `res`; without a re-check the wedged-bridge recovery failure
/// fell into the skip branch — zero-fill + advance — marching the disc at
/// one bridge-recovery per unit ("runs forever, no MKV", hard rule #2). The
/// fix re-checks the recovery error for `is_scsi_transport_failure()` before
/// the skip block and returns `Error::DiscRead`.
#[test]
fn transport_failure_on_recovery_read_aborts_even_with_skip_errors() {
const COUNT: u32 = 10;
let bad = 4u32;
let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let reader = RecoveryTransportFailReader {
capacity: COUNT,
bad_sector: bad,
log: log.clone(),
};
let mut stream = DiscStream::new(
Box::new(reader),
synthetic_title(COUNT),
crate::decrypt::DecryptKeys::None,
8,
ContentFormat::BdTs,
false,
None,
)
.unwrap();
stream.skip_errors = true;
// Drive fill_extents across batches: the good leading sectors mux fine,
// and the batch covering the bad sector shrinks to size 1, fails the
// fast read (0x02), then the bottom-out recovery read returns the
// transport failure (0xFF) — which must abort.
let mut res = Ok(true);
for _ in 0..1000 {
res = stream.fill_extents();
if !matches!(res, Ok(true)) {
break;
}
}
assert!(
res.is_err(),
"a transport failure on the recovery read must abort fill_extents, got {res:?}"
);
assert_eq!(
stream.errors, 0,
"a recovery-read transport-failure abort must NOT count as a skip"
);
assert_eq!(
stream.lost_bytes, 0,
"a transport-failure abort zero-fills nothing"
);
// Prove the bottom-out recovery read was actually reached and aborted on.
let reads = log.lock().unwrap();
assert!(
reads
.iter()
.any(|&(lba, count, rec)| rec && lba == bad && count == 1),
"expected a recovery=true read at the bad sector to have been attempted; got {reads:?}"
);
}
/// Regression: a USB-bridge transport crash (status=0xFF) during a direct
/// single-pass `disc://→mkv://` rip must ABORT immediately, even under
/// `skip_errors=true`. The pre-fix behavior treated it as a skippable bad
/// sector: zero-fill, advance, repeat — marching the whole disc at one
/// ~15s bridge-recovery per probe, producing no MKV ("runs forever"). The
/// fix mirrors the multipass sweep: transport failure short-circuits to an
/// error before any shrink/skip, so exactly ONE read is issued and no skip
/// is counted.
#[test]
fn transport_failure_aborts_single_pass_even_with_skip_errors() {
const COUNT: u32 = 10;
let bad = 4u32;
let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let reader = TransportFailReader {
capacity: COUNT,
bad_sector: bad,
log: log.clone(),
};
let mut stream = DiscStream::new(
Box::new(reader),
synthetic_title(COUNT),
crate::decrypt::DecryptKeys::None,
8,
ContentFormat::BdTs,
false,
None,
)
.unwrap();
stream.skip_errors = true;
let res = stream.fill_extents();
assert!(
res.is_err(),
"transport failure must abort fill_extents, not skip past it"
);
assert_eq!(
stream.errors, 0,
"a transport-failure abort must NOT count as a skipped sector"
);
let reads = log.lock().unwrap();
assert_eq!(
reads.len(),
1,
"transport failure must abort after the first failed read with no \
shrink/retry/skip-ahead; got reads {reads:?}"
);
}
/// AACS unit-alignment skip (the #1 coverage gap). With `unit_align=3`
/// (DecryptKeys::Aacs) and `skip_errors=true`, a single bad mid-extent
/// sector must NOT desync the rest of the title: every `read_sectors`
/// request must start on a 3-sector unit boundary relative to the extent
/// start, and the skip over the failed unit must advance the cursor by a
/// whole 3-sector unit (never a single sector).
#[test]
fn aacs_reads_stay_unit_aligned_and_skip_whole_units() {
const COUNT: u32 = 30;
const ALIGN: u32 = 3;
// Bad sector at offset 13 — inside unit 4 (offsets 12,13,14). The
// whole unit must be skipped, keeping the cursor unit-aligned.
let bad = 13u32;
let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let reader = RecordingReader {
capacity: COUNT,
bad_sector: bad,
log: log.clone(),
};
let title = synthetic_title(COUNT);
let keys = crate::decrypt::DecryptKeys::Aacs {
unit_keys: vec![(0, [0u8; 16])],
read_data_key: None,
format: crate::disc::ContentFormat::BdTs,
};
let mut stream = DiscStream::new(
Box::new(reader),
title,
keys,
8,
ContentFormat::BdTs,
false,
None,
)
.unwrap();
stream.skip_errors = true;
assert_eq!(
stream.unit_align, ALIGN as u16,
"AACS keys must set unit_align=3"
);
// Drive fill_extents to EOF (no PES demux needed — we observe the
// raw read pattern directly).
let ext_start = 0u32;
let mut guard = 0;
loop {
match stream.fill_extents() {
Ok(true) => {}
Ok(false) => break,
Err(e) => panic!("fill_extents errored unexpectedly: {e}"),
}
guard += 1;
assert!(guard < 1000, "fill_extents did not reach EOF");
}
let reads = log.lock().unwrap();
assert!(!reads.is_empty(), "expected at least one read");
for &(lba, count) in reads.iter() {
assert_eq!(
(lba - ext_start) % ALIGN,
0,
"read at lba {lba} is not unit-aligned (offset {} % {ALIGN} != 0)",
lba - ext_start
);
// Non-tail reads must be a whole number of units; the only permitted
// short read is a final partial unit (below one unit). Assert it
// rather than documenting it — a mid-stream non-unit-multiple read
// would straddle AACS unit boundaries and decrypt under the wrong
// alignment.
assert!(
(count as u32).is_multiple_of(ALIGN) || (count as u32) < ALIGN,
"read count {count} is neither a whole number of units nor a sub-unit tail"
);
}
// At least one error was skipped (the bad unit) and a SectorSkipped
// event was emitted; errors counter advanced by exactly the bad units.
assert!(stream.errors >= 1, "expected the bad unit to be skipped");
// Regression: `lost_bytes` must account for the WHOLE skipped unit
// (3 sectors = 6144 bytes), not a single sector. A loss estimate
// built from `errors * 2048` would undercount AACS loss ~3x — the
// single-pass abort-gate bug this guards against. Exactly one unit
// is bad in this fixture, so lost_bytes == errors * ALIGN * 2048.
assert_eq!(
stream.lost_bytes,
stream.errors * ALIGN as u64 * 2048,
"AACS skip must record a whole unit (6144 B) per skip event, not 2048"
);
assert!(
stream.lost_bytes > stream.errors * 2048,
"lost_bytes must exceed the errors*2048 undercount for AACS units"
);
// Crucial anti-desync assertion: the read that bottomed out and was
// skipped must have been a single 3-sector unit starting at offset 12
// (the unit boundary at or below the bad sector 13), NOT a 1-sector
// read at 13. Find a recorded read of (12, 3).
assert!(
reads
.iter()
.any(|&(lba, count)| lba == 12 && count == ALIGN as u16),
"expected a unit-aligned (lba=12,count=3) read over the bad unit; got {reads:?}"
);
// And NO single-sector read at the bad sector itself (would be a desync).
assert!(
!reads.iter().any(|&(lba, count)| lba == bad && count == 1),
"a 1-sector read at the bad sector {bad} would desync the AACS unit stream"
);
}
/// `unit_align == 1` (DecryptKeys::None) variant: single-sector skips
/// still work (CSS/raw is self-synchronizing, so a 1-sector skip is
/// correct there — contrast with the AACS whole-unit skip above).
#[test]
fn unencrypted_single_sector_skip_works() {
const COUNT: u32 = 10;
let bad = 4u32;
let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let reader = RecordingReader {
capacity: COUNT,
bad_sector: bad,
log: log.clone(),
};
let mut stream = DiscStream::new(
Box::new(reader),
synthetic_title(COUNT),
crate::decrypt::DecryptKeys::None,
8,
ContentFormat::BdTs,
false,
None,
)
.unwrap();
stream.skip_errors = true;
assert_eq!(stream.unit_align, 1, "None keys must leave unit_align=1");
let mut guard = 0;
loop {
match stream.fill_extents() {
Ok(true) => {}
Ok(false) => break,
Err(e) => panic!("fill_extents errored unexpectedly: {e}"),
}
guard += 1;
assert!(guard < 1000, "fill_extents did not reach EOF");
}
let reads = log.lock().unwrap();
// The bad sector must have been retried down to a single sector and
// skipped at count==1 — the self-synchronizing per-sector path.
assert!(
reads.iter().any(|&(lba, count)| lba == bad && count == 1),
"align=1 must bottom out at a 1-sector read over the bad sector; got {reads:?}"
);
assert!(stream.errors >= 1);
// align=1: a skip event covers exactly one sector, so lost_bytes
// and errors*2048 agree (the AACS undercount does not apply here).
assert_eq!(
stream.lost_bytes,
stream.errors * 2048,
"single-sector (align=1) skip must record exactly 2048 B per event"
);
}
#[test]
fn halt_via_with_halt_from_arc_observed_by_is_halted() {
let arc = Arc::new(AtomicBool::new(false));
let stream = DiscStream::new(
Box::new(ZeroReader { capacity: 8 }),
synthetic_title(8),
crate::decrypt::DecryptKeys::None,
8,
crate::disc::ContentFormat::BdTs,
false,
None,
)
.unwrap()
.with_halt(Halt::from_arc(arc.clone()));
assert!(!stream.is_halted());
arc.store(true, std::sync::atomic::Ordering::Relaxed);
assert!(
stream.is_halted(),
"with_halt(Halt::from_arc) must observe Arc-side flips"
);
}
/// Every read fails CSS-locked (`05/6F/03`) — a scrambled DVD whose title key
/// can't be cracked. Drives `resolve_dvd_title_key` to `ScrambledUncracked`.
struct LockedReader;
impl crate::sector::SectorSource for LockedReader {
fn read_sectors(
&mut self,
lba: u32,
_count: u16,
_buf: &mut [u8],
_recovery: bool,
) -> crate::error::Result<usize> {
Err(crate::error::Error::DiscRead {
sector: lba as u64,
status: Some(2),
sense: Some(crate::scsi::ScsiSense {
sense_key: 0x05,
asc: 0x6F,
ascq: 0x03,
}),
})
}
fn capacity_sectors(&self) -> u32 {
64
}
}
fn mpegps_title(sector_count: u32) -> DiscTitle {
let mut t = synthetic_title(sector_count);
t.content_format = ContentFormat::MpegPs;
t
}
/// PARITY with `build_iso_pipeline_dvd_none_keys_scrambled_hard_fails`: the
/// live-drive single-pass constructor must ALSO hard-fail (not build a
/// scrambled-passthrough stream) for a `None`-keyed scrambled MPEG-PS DVD —
/// the exact 328k-decode-error corruption path, on the single-pass side.
#[test]
fn disc_stream_new_dvd_none_scrambled_hard_fails() {
let res = DiscStream::new(
Box::new(LockedReader),
mpegps_title(8),
crate::decrypt::DecryptKeys::None,
8,
ContentFormat::MpegPs,
false,
None,
);
assert!(
res.is_err(),
"single-pass DiscStream must hard-fail on a scrambled, keyless CSS DVD"
);
}
/// `raw` must bypass the CSS crack at the DiscStream boundary too: the same
/// scrambled-uncrackable input that hard-fails above must CONSTRUCT in raw
/// mode (ciphertext passthrough), never hard-fail.
#[test]
fn disc_stream_new_raw_bypasses_css_crack() {
let res = DiscStream::new(
Box::new(LockedReader),
mpegps_title(8),
crate::decrypt::DecryptKeys::None,
8,
ContentFormat::MpegPs,
true, // raw
None,
);
assert!(
res.is_ok(),
"raw single-pass must construct without cracking, even on scrambled-uncrackable input"
);
}
mod stream_surface_tests {
//! The `impl crate::pes::Stream for DiscStream` surface plus the adaptive
//! batch sizer. These are the accessors the CLI's abort gate and the
//! header-wait loop read; a constant in any of them hides read loss or
//! stalls the mux, so each is constrained against a value that is neither
//! the mutation constant nor the initial state.
use super::*;
use crate::pes::Stream as PesStream;
// ── errors() / lost_bytes(): honest loss reporting ─────────────────
/// `Stream::errors()` and `Stream::lost_bytes()` are the ONLY channel by
/// which a caller learns that bytes went missing (the abort gate and the
/// "N sectors skipped" report both read them). A constant there reports a
/// lossy rip as clean.
///
/// Two short-read fills are driven so both counters land on values that are
/// neither `0` nor `1` and are distinct from each other — a single fill
/// would leave `errors == 1`, indistinguishable from a stuck constant, and
/// equal counters would not prove the two accessors read different fields.
#[test]
fn errors_and_lost_bytes_report_real_short_read_loss_through_the_trait() {
let mut s = short_read_stream(true);
// Before any read the stream is clean — establishes the accessors are
// not simply echoing a preloaded value.
assert_eq!(PesStream::errors(&s), 0);
assert_eq!(PesStream::lost_bytes(&s), 0);
for _ in 0..2 {
assert!(
s.fill_extents()
.expect("skip_errors absorbs the short read")
);
}
// Two 8-sector (16384 B) requests, 2048 B delivered each: two skip
// events, 2 * (16384 - 2048) = 28672 bytes lost.
assert_eq!(
PesStream::errors(&s),
2,
"errors() must report BOTH short reads, not a constant"
);
assert_eq!(
PesStream::lost_bytes(&s),
28_672,
"lost_bytes() must report the byte total, not an event count or a constant"
);
assert_ne!(
PesStream::errors(&s),
PesStream::lost_bytes(&s),
"the two accessors must read different fields"
);
}
/// Frames the B1 resync gate discards must reach `errors()`, including
/// after the gap RESOLVES.
///
/// `ResyncGate::dropped` is zeroed the moment a keyframe disarms the
/// gate, and the only EOF warning fires for gates STILL armed — so a
/// mid-title gap that resolves left no trace anywhere. That is the
/// common case: most gaps do resolve. A rip with several concealed gaps
/// reported 0 errors and 0 lost bytes while whole GOPs were discarded.
///
/// This asserts the accessor, not the gate's own counter, because
/// `errors()` is the only channel through which a caller learns
/// anything went wrong.
#[test]
fn errors_reports_frames_the_resync_gate_dropped_after_the_gap_resolves() {
let mut s = short_read_stream(true);
assert_eq!(PesStream::errors(&s), 0, "clean before anything happens");
// Drive a gate directly: a discontinuity, two dropped inter-coded
// frames, then a keyframe that resyncs and zeroes the per-run count.
// The short-read fixture's title carries no streams, so give it a
// gate to drive.
s.resync.push(crate::mux::resync::ResyncGate::new());
let gate = s.resync.last_mut().expect("just pushed");
assert!(!gate.admit(true, true, false));
assert!(!gate.admit(true, false, false));
assert!(gate.admit(true, false, true));
assert_eq!(
gate.dropped_in_run(),
0,
"the resync zeroed the per-run counter, which is the whole trap"
);
assert_eq!(
PesStream::errors(&s),
2,
"the two discarded frames must still reach the caller after the \
gap resolved; reading the per-run counter here would report 0"
);
}
// ── write(): DiscStream is read-only ──────────────────────────────
/// `DiscStream` is the tree's only read-only `Stream`. `write()` returning
/// `Ok(())` would make a caller that muxed INTO a disc stream believe every
/// frame landed, producing a silent no-op rip. It must refuse with the
/// numeric code `E_STREAM_READ_ONLY`.
#[test]
fn write_refuses_with_the_read_only_code() {
let mut s = short_read_stream(false);
let frame = crate::pes::PesFrame {
coding: None,
source: None,
track: 0,
pts: 0,
keyframe: true,
data: vec![0u8; 4],
duration_ns: None,
};
let err = PesStream::write(&mut s, &frame)
.expect_err("a read-only stream must never accept a frame");
assert_eq!(err.kind(), std::io::ErrorKind::Unsupported);
let code = format!("E{}", crate::error::Error::StreamReadOnly.code());
assert!(
err.to_string().contains(&code),
"expected the read-only code {code}, got {err}"
);
}
// ── codec_private() / headers_ready() ─────────────────────────────
/// MPEG-2 sequence header (ISO/IEC 13818-2 §6.2.2.1) — start code `0x000001B3`
/// followed by 12-bit horizontal_size, 12-bit vertical_size, then
/// aspect_ratio_information / frame_rate_code and the bit-rate/VBV tail.
fn seq_header(width: u16, height: u16) -> Vec<u8> {
let mut h = vec![0x00, 0x00, 0x01, 0xB3];
h.push((width >> 4) as u8);
h.push((((width & 0x0F) as u8) << 4) | ((height >> 8) & 0x0F) as u8);
h.push((height & 0xFF) as u8);
h.push((3 << 4) | 4); // aspect_ratio_information=3, frame_rate_code=4
h.extend_from_slice(&[0xFF, 0xFF, 0xFF, 0x00]);
h
}
/// MPEG-2 picture header (ISO/IEC 13818-2 §6.2.3), coding type in bits 3..5
/// of the sixth byte. Type 1 = I-picture.
fn picture_header(coding_type: u8) -> Vec<u8> {
vec![
0x00,
0x00,
0x01,
0x00,
0x00,
(coding_type & 0x07) << 3,
0,
0,
]
}
/// One video (MPEG-2, track 1) behind one audio (AC-3, track 0). The video
/// is deliberately NOT track 0 so a `codec_private` that ignored its `track`
/// argument, or read the pid map in the wrong direction, would answer with
/// the audio parser (which has no codec private) and fail.
fn audio_then_video_title() -> DiscTitle {
use crate::disc::{
AudioChannels, AudioStream, Codec, ColorSpace, FrameRate, HdrFormat, LabelPurpose,
Resolution, SampleRate, VideoStream,
};
let mut t = DiscTitle {
extents: vec![crate::disc::Extent {
start_lba: 0,
sector_count: 8,
}],
..DiscTitle::empty()
};
t.content_format = ContentFormat::MpegPs;
t.streams = vec![
crate::disc::Stream::Audio(AudioStream {
pid: 0x00BD,
codec: Codec::Ac3,
channels: AudioChannels::Stereo,
language: "eng".to_string(),
sample_rate: SampleRate::S48,
secondary: false,
purpose: LabelPurpose::Normal,
label: String::new(),
}),
crate::disc::Stream::Video(VideoStream {
pid: 0x00E0,
codec: Codec::Mpeg2,
resolution: Resolution::R480i,
frame_rate: FrameRate::F29_97,
hdr: HdrFormat::Sdr,
color_space: ColorSpace::Smpte170m,
display_aspect: None,
secondary: false,
label: String::new(),
measured_cicp: None,
}),
];
t
}
fn mixed_stream() -> DiscStream {
DiscStream::new(
Box::new(ZeroReader { capacity: 8 }),
audio_then_video_title(),
crate::decrypt::DecryptKeys::None,
8,
ContentFormat::MpegPs,
false,
None,
)
.unwrap()
}
/// `headers_ready()` gates the CLI's "wait for codec private" loop: a
/// constant `true` starts the mux before the video's extradata exists
/// (an MKV with an empty CodecPrivate — unplayable video, RFC 9559 §5.1.4.1.5
/// requires it for V_MPEG2), and a constant `false` hangs forever.
///
/// Both directions are pinned in one case: not ready before the video
/// parser has seen a sequence header, ready after.
#[test]
fn headers_ready_follows_the_video_codec_private_and_flips_both_ways() {
let mut s = mixed_stream();
assert!(
PesStream::codec_private(&s, 1).is_none(),
"no sequence header parsed yet"
);
assert!(
!PesStream::headers_ready(&s),
"a non-secondary video track without codec private must NOT be reported ready"
);
// Feed the video parser a complete access unit: sequence header +
// I-picture, closed by a following picture so the AU boundary is hit.
let vpid = 0x00E0u16;
let (_, parser) = s
.parsers
.iter_mut()
.find(|(p, _)| *p == vpid)
.expect("video parser present");
let mut au = seq_header(720, 480);
au.extend_from_slice(&picture_header(1));
au.extend_from_slice(&[0xAA; 16]);
let pes = |data: Vec<u8>, pts: Option<i64>| crate::mux::ts::PesPacket {
source: None,
pid: vpid,
pts,
dts: None,
data,
discontinuity: false,
};
let _ = parser.parse(&pes(au, Some(0)));
let mut next = picture_header(3);
next.extend_from_slice(&[0xBB; 16]);
let _ = parser.parse(&pes(next, None));
let cp = PesStream::codec_private(&s, 1)
.expect("codec_private must reach the VIDEO track's parser, not track 0's");
assert_eq!(
&cp[..4],
&[0x00, 0x00, 0x01, 0xB3],
"codec private is the MPEG-2 sequence header"
);
assert_eq!(&cp[4..7], &[0x2D, 0x01, 0xE0], "720x480 as authored");
assert!(
PesStream::headers_ready(&s),
"with the video's codec private present the mux may start"
);
// The AC-3 track genuinely has none — so the Some() above is a real
// per-track lookup, not a fixed answer.
assert!(PesStream::codec_private(&s, 0).is_none());
// And a track index past the end of the pid map has none either.
assert!(PesStream::codec_private(&s, 7).is_none());
}
// ── on_event() / emit() ───────────────────────────────────────────
/// `on_event()` installs the sink and `emit()` feeds it. If either is a
/// no-op the CLI's progress bar never moves and skipped sectors are never
/// reported — the rip looks clean and stalled at 0 %.
#[test]
fn installed_event_sink_receives_skip_and_progress_events() {
let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
let sink = log.clone();
let mut s = short_read_stream(true);
s.on_event(move |e| {
let tag = match e.kind {
EventKind::SectorSkipped { sector } => format!("skip:{sector}"),
EventKind::BytesRead { bytes, total } => format!("bytes:{bytes}/{total}"),
other => format!("other:{other:?}"),
};
sink.lock().unwrap().push(tag);
});
assert!(s.fill_extents().expect("short read absorbed"));
let got = log.lock().unwrap().clone();
assert!(
got.contains(&"skip:0".to_string()),
"the skipped unit at LBA 0 must reach the installed sink; got {got:?}"
);
// 2048 B delivered out of a 64-sector (131072 B) title.
assert!(
got.contains(&"bytes:2048/131072".to_string()),
"progress must report DELIVERED bytes against the extent total; got {got:?}"
);
}
// ── set_raw() ─────────────────────────────────────────────────────
/// `set_raw()` must flip BOTH key holders — the metadata mirror read by
/// `info()`-side callers and the wrapped reader that actually decrypts. A
/// no-op leaves an AACS stream decrypting when the caller asked for
/// ciphertext (the `--raw` forensic path), silently returning plaintext.
#[test]
fn set_raw_clears_both_the_mirror_and_the_readers_keys() {
let mut s = DiscStream::new(
Box::new(ZeroReader { capacity: 8 }),
synthetic_title(8),
crate::decrypt::DecryptKeys::Aacs {
unit_keys: vec![(0, [0x11u8; 16])],
read_data_key: None,
format: ContentFormat::BdTs,
},
3,
ContentFormat::BdTs,
false,
None,
)
.unwrap();
assert!(
s.decrypt_keys.is_encrypted(),
"fixture must start encrypted or the test proves nothing"
);
// Behavioural witness for the WRAPPED reader's keys: while AACS keys
// are installed the decorator refuses a read that does not begin on a
// 6144-byte aligned unit (it would mis-decrypt every following unit),
// so a mid-unit LBA hard-fails.
let mut buf = vec![0u8; 2048];
let before = s.reader.read_sectors(1, 1, &mut buf, false);
assert!(
matches!(before, Err(crate::error::Error::DecryptFailed)),
"the encrypted fixture must reject a mid-unit read; got {before:?}"
);
s.set_raw();
assert!(
!s.decrypt_keys.is_encrypted(),
"the metadata mirror must report the stream as raw"
);
let after = s.reader.read_sectors(1, 1, &mut buf, false);
assert_eq!(
after.expect("raw mode must pass ciphertext straight through"),
2048,
"the wrapped reader must stop decrypting, not just the mirror"
);
}
// ── AdaptiveBatch ─────────────────────────────────────────────────
/// AACS decrypts whole 3-sector (6144 B) units, so every batch size at or
/// above one doubled unit must stay a multiple of 3 — an unaligned size
/// makes the next read straddle a unit boundary and mis-decrypt the rest of
/// the title. Below 6 the ladder descends 3 → 1 with no unaligned rungs.
#[test]
fn halve_batch_size_keeps_unit_alignment_and_bottoms_out_at_one() {
assert_eq!(
halve_batch_size(64),
30,
"32 rounded down to a unit multiple"
);
assert_eq!(halve_batch_size(30), 15);
assert_eq!(halve_batch_size(12), 6, "6 is already unit-aligned");
assert_eq!(halve_batch_size(11), 5, "below 6: no alignment rounding");
assert_eq!(halve_batch_size(6), 3);
assert_eq!(halve_batch_size(3), 1);
assert_eq!(halve_batch_size(2), 1);
assert_eq!(
halve_batch_size(1),
1,
"must never reach 0 — a 0-sector read"
);
for size in 1u16..=4096 {
let h = halve_batch_size(size);
assert!(h >= 1, "halve({size}) must never be 0");
assert!(h <= size, "halve({size}) = {h} must not grow");
assert!(
h < 6 || h.is_multiple_of(3),
"halve({size}) = {h} is unit-unaligned"
);
}
}
#[test]
fn double_batch_size_grows_toward_preferred_without_breaking_alignment() {
assert_eq!(double_batch_size(30, 64), 60);
assert_eq!(
double_batch_size(4, 64),
6,
"8 rounded down to a unit multiple"
);
assert_eq!(
double_batch_size(1, 64),
2,
"below 6: no alignment rounding"
);
assert_eq!(
double_batch_size(60, 64),
63,
"clamped to preferred, then aligned"
);
for size in 1u16..=2048 {
let d = double_batch_size(size, 4096);
assert!(d >= size, "double({size}) = {d} must not shrink");
assert!(
d < 6 || d.is_multiple_of(3),
"double({size}) = {d} is unit-unaligned"
);
}
}
/// The sizer must actually probe back up: after a failure drops the batch,
/// a sustained clean run has to return a `BatchSizeChanged{Probed}` event
/// AND raise `current`. Never probing locks a rip at the reduced size for
/// the rest of the disc (the whole point of the amortised descent).
#[test]
fn on_success_probes_up_after_a_sustained_clean_run_and_resets_the_streak() {
let mut b = AdaptiveBatch::new(64);
assert!(
matches!(
b.on_failure(),
Some(EventKind::BatchSizeChanged {
new_size: 30,
reason: BatchSizeReason::Shrunk
})
),
"a failure must shrink 64 -> 30"
);
assert_eq!(b.current(), 30);
// Just under the 51200-sector probe threshold: still silent.
let mut fed = 0u32;
while fed + 30 < PROBE_THRESHOLD_SECTORS {
assert!(
b.on_success(30).is_none(),
"no probe before {PROBE_THRESHOLD_SECTORS} clean sectors (at {fed})"
);
fed += 30;
}
assert_eq!(b.current(), 30, "still at the reduced size");
// The read that crosses the threshold probes up.
let ev = b
.on_success(30)
.expect("a sustained clean run must probe the batch size back up");
assert!(
matches!(
ev,
EventKind::BatchSizeChanged {
new_size: 60,
reason: BatchSizeReason::Probed
}
),
"expected a Probed grow to 60, got {ev:?}"
);
assert_eq!(
b.current(),
60,
"the sizer must actually adopt the new size"
);
assert_eq!(
b.streak_sectors, 0,
"the streak resets so the next probe needs a fresh clean run"
);
// At the preferred size a clean run must NOT keep firing events.
let mut b = AdaptiveBatch::new(64);
for _ in 0..(PROBE_THRESHOLD_SECTORS / 64 + 2) {
assert!(
b.on_success(64).is_none(),
"no probe is possible when already at the preferred size"
);
}
assert_eq!(b.current(), 64);
}
}
}
Reference in New Issue View Git Blame Copy Permalink