sector: generic recovery seam; FMTS forensic segments as decrypt loss

Replace the AACS-specific inline key-fetch in the decrypt decorator with
a scheme-neutral recovery seam: the input stream (L3) installs a Recover
closure (none / AACS key-fetch) and the decorator (L2) runs it at the
single decrypt-miss point. FMTS (AACS 2.1) forensic-segment units that no
key opens are just undecryptable units, concealed and counted as ordinary
decrypt loss with no FMTS-specific branch ("a loss is a loss"), so the
separate bytes_undecryptable bucket collapses into one loss count.

- sector/recovery.rs: the seam (MissOutcome, none/key_fetch factories),
  naming no encryption scheme in its type.
- FMTS: segment routing primitives + BYPASS_FMTS_KEY, and an upfront
  ensure_forensic_segments_decryptable gate (Error::FmtsKeyMissing) in
  the mux input path, parallel to the unit-key gate.
- CSS descramble/rekey moves from decrypt_sectors into
  css::descramble_region: CSS self-recovers from the data itself, so it
  stays OFF the seam (which is only for external inputs).
- disc/mod.rs also: main-title selection aligned to largest physical
  size; is_regular read from the open file handle, not metadata(path),
  fixing a swallowed sync_all on a fresh-rip ISO. decrypt_threads()
  resolved once via OnceLock off the per-buffer hot path.

src/aacs/segment.rs

@@ -29,6 +29,21 @@ pub const SEGMENT_RECORD_LEN: usize = 16;
 /// Bytes per BDAV source packet (188-byte TS + 4-byte arrival-time header).
 pub const SOURCE_PACKET_LEN: u64 = 192;
 
+/// Whether a 2.1 (FMTS) disc may rip WITHOUT segment (variant) keys.
+///
+/// `true` (today): the forensic variant segments are skipped as expected loss
+/// and the bulk of the title decodes with the unit key, so a 2.1 disc rips
+/// mostly-complete. A unit key (VUK) is still required, exactly as for any AACS
+/// disc. `false`: the absence of a segment-key source is a hard, UPFRONT failure
+/// ([`Error::FmtsKeyMissing`]) — the same policy as a missing unit key, so a
+/// forensic-holed rip is refused rather than produced. No segment-key source
+/// exists yet, so `true` is the only value under which a 2.1 disc rips at all;
+/// flip to `false` once segment keys can be sourced and a partial rip should be
+/// refused. Hardcoded on purpose — not a user setting.
+///
+/// [`Error::FmtsKeyMissing`]: crate::error::Error::FmtsKeyMissing
+pub const BYPASS_FMTS_KEY: bool = true;
+
 /// One forensic variant segment: the inclusive source-packet range it occupies
 /// in the FMTS clip.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -63,6 +78,47 @@ impl Segment {
     pub fn contains_spn(&self, spn: u32) -> bool {
         spn >= self.start_spn && spn <= self.end_spn
     }
+
+    /// True when the inclusive source-packet span `[first, last]` overlaps this
+    /// segment. Used to decide whether an aligned unit (which spans several
+    /// packets) touches the segment at all, not just whether one packet does.
+    pub fn overlaps_spn(&self, first: u32, last: u32) -> bool {
+        first <= self.end_spn && last >= self.start_spn
+    }
+}
+
+/// Source packets spanned by one AACS aligned unit: `6144 / 192 = 32`.
+pub const PACKETS_PER_UNIT: u32 =
+    (crate::aacs::content::ALIGNED_UNIT_LEN as u64 / SOURCE_PACKET_LEN) as u32;
+
+/// Byte offset within the clip of a clip-relative 2048-byte sector `lba`. The
+/// FMTS decode reads the clip file directly, so `lba` 0 is the clip's first
+/// byte and this offset lines up with the source-packet grid the segment map
+/// uses.
+pub fn lba_byte_offset(lba: u32) -> u64 {
+    lba as u64 * 2048
+}
+
+/// The forensic segment an AACS aligned unit belongs to, if any, given the
+/// unit's clip-relative byte offset.
+///
+/// This is the routing decision behind a 2.1 decrypt-miss: a unit that
+/// overlaps a forensic segment must be opened with that segment's **variant
+/// key** (from `SegmentKeyNNNNN.tbl`), not the CPS Unit Key. Opening it with
+/// the Unit Key is exactly what yields the broken-reference-frame garbage a
+/// plain unit-key rip produces. A unit outside every segment is ordinary
+/// content and a miss on it is a Unit-Key miss, so this returns `None` and the
+/// caller falls back to the normal unit-key fetch.
+///
+/// The unit is tested as a packet *span* (`[off/192, (off+6144-1)/192]`) so a
+/// unit that only partly overlaps a segment edge is still classified as
+/// variant; on the observed disc segments are unit-aligned, but the span test
+/// does not rely on that.
+pub fn variant_segment_for_unit(segments: &[Segment], unit_offset: u64) -> Option<&Segment> {
+    let unit_len = crate::aacs::content::ALIGNED_UNIT_LEN as u64;
+    let first = (unit_offset / SOURCE_PACKET_LEN) as u32;
+    let last = ((unit_offset + unit_len - 1) / SOURCE_PACKET_LEN) as u32;
+    segments.iter().find(|s| s.overlaps_spn(first, last))
 }
 
 /// Parse `IndividualSegment.tbl` into its forensic variant segments, in table
@@ -160,4 +216,58 @@ mod tests {
         let tbl = build_tbl(&[]);
         assert_eq!(parse_individual_segments(&tbl), Some(Vec::new()));
     }
+
+    #[test]
+    fn packets_per_unit_is_thirty_two() {
+        // 6144-byte aligned unit / 192-byte source packet.
+        assert_eq!(PACKETS_PER_UNIT, 32);
+    }
+
+    #[test]
+    fn unit_inside_segment_routes_to_variant() {
+        // A real first-record segment: packets [343680, 346239].
+        let segs = parse_individual_segments(&build_tbl(&[(1, 343680, 346239)])).unwrap();
+        // A unit sitting squarely inside: start at packet 344000 → byte 344000*192.
+        let off = 344000u64 * SOURCE_PACKET_LEN;
+        let hit = variant_segment_for_unit(&segs, off).expect("inside the segment");
+        assert_eq!(hit.number, 1);
+    }
+
+    #[test]
+    fn unit_outside_every_segment_is_unit_key_miss() {
+        let segs = parse_individual_segments(&build_tbl(&[(1, 343680, 346239)])).unwrap();
+        // A unit well before the segment is ordinary content → None (unit-key path).
+        let off = 1000u64 * SOURCE_PACKET_LEN;
+        assert!(variant_segment_for_unit(&segs, off).is_none());
+    }
+
+    #[test]
+    fn unit_straddling_a_segment_edge_counts_as_variant() {
+        // Segment starts at packet 100. A unit that ENDS just inside it (its 32
+        // packets straddle the boundary) must still route to the variant key,
+        // because part of its ciphertext is variant-encrypted.
+        let segs = parse_individual_segments(&build_tbl(&[(7, 100, 200)])).unwrap();
+        // Unit covering packets [80, 111]: overlaps [100,200] at the tail.
+        let off = 80u64 * SOURCE_PACKET_LEN;
+        let hit = variant_segment_for_unit(&segs, off).expect("straddles the start edge");
+        assert_eq!(hit.number, 7);
+        // A unit ending exactly at packet 99 (offset s.t. last = 99) does NOT overlap.
+        let before = 68u64 * SOURCE_PACKET_LEN; // [68, 99]
+        assert!(variant_segment_for_unit(&segs, before).is_none());
+    }
+
+    #[test]
+    fn no_segments_never_routes_to_variant() {
+        // The 1.0 / 2.0 case: no forensic map, so every miss is a unit-key miss.
+        assert!(variant_segment_for_unit(&[], lba_byte_offset(0)).is_none());
+        assert!(variant_segment_for_unit(&[], lba_byte_offset(9_999_999)).is_none());
+    }
+
+    #[test]
+    fn lba_maps_to_the_packet_grid() {
+        // A unit is 3 sectors (6144 bytes) = 32 packets. Clip-relative LBA 3 is
+        // the second aligned unit, which starts at packet 32.
+        let off = lba_byte_offset(3);
+        assert_eq!(off / SOURCE_PACKET_LEN, 32);
+    }
 }

src/css/lfsr.rs

@@ -24,13 +24,12 @@ use super::tables::{TAB1, TAB2, TAB3, TAB4, TAB5};
 /// hierarchy, not the content cipher). Bytes 0x80..0x800 are recovered with
 /// `*p = TAB1[*p] ^ (i_t5 & 0xff)`.
 ///
-/// The scramble flag at byte 0x14 (bits 4-5) indicates encryption. Like
+/// The scramble flag at byte 0x14 (bits 4-5) indicates encryption. This
-/// libdvdcss, the flag byte is NOT modified here — the caller treats a
+/// descrambler CLEARS that flag after unscrambling, so a descrambled sector
-/// nonzero `sector[0x14] & 0x30` as "needs unscrambling" and the descramble
+/// reads as `sector[0x14] & 0x30 == 0`; callers and the tests use that to tell
-/// is its own inverse, so re-running it on plaintext would re-scramble.
+/// it from ciphertext, and re-running descramble on an already-cleared sector
-/// (freemkv historically cleared the flag; we keep clearing it so callers
+/// is a no-op (the flag guard below skips it). Clearing does not affect the
-/// and the existing tests can distinguish a descrambled sector. This does
+/// recovered body.
-/// not affect the recovered body.)
 ///
 /// No-op (returns without modifying `sector`) in two cases:
 /// - `sector.len() < 2048`: the encrypted region (0x80..0x800) is not

src/css/mod.rs

@@ -271,6 +271,52 @@ pub fn descramble_sector(state: &CssState, sector: &mut [u8]) {
     lfsr::descramble_sector(&state.title_key, sector);
 }
 
+/// Descramble a whole CSS buffer in place, re-cracking the title key on a VOB
+/// region boundary. `title_key` is a CACHE of the last crack, not a fixed disc
+/// key: it changes per VTS/VOB region, so it is validated on every scrambled
+/// sector and re-cracked on a miss (libdvdcss's on-demand per-region rekey).
+///
+/// This CSS key acquisition is intrinsic to the cipher — CSS has no external key
+/// source, the ONLY way to a title key is cracking the data — so it lives with
+/// the CSS primitives and runs inside `decrypt::decrypt_sectors` (a public,
+/// self-contained CSS decrypt), NOT at the post-decrypt recovery seam that AACS
+/// key-fetch and FMTS segment-skip use (those consume external inputs).
+///
+/// The clear header (`<0x80`) is never scrambled, so its periodic crib predicts
+/// the plaintext at `0x80`. Descramble with the cached key; if the crib fails to
+/// reappear the key region changed (or the primed key was wrong) — restore the
+/// ciphertext, re-crack from this very sector, and descramble again. A crib-less
+/// sector (no periodic run) can be neither validated nor cracked, so it rides the
+/// cached key — correct, because it lives in the same region as the nearby crib
+/// sector that set the cache.
+pub fn descramble_region(buf: &mut [u8], title_key: &mut [u8; 5]) {
+    for chunk in buf.chunks_mut(2048) {
+        if chunk.len() < 2048 || !is_scrambled(chunk) {
+            continue;
+        }
+        let crib = stevenson::attack_crib(chunk);
+        // Snapshot the ciphertext (chunk is exactly 2048 here) only when there is
+        // a crib to validate against, so the common cache-hit path costs no
+        // per-sector heap allocation.
+        let mut original = [0u8; 2048];
+        if crib.is_some() {
+            original.copy_from_slice(chunk);
+        }
+        lfsr::descramble_sector(title_key, chunk);
+        if let Some(crib) = crib {
+            if chunk[0x80..0x80 + 10] != crib[..] {
+                // Cached key is stale for this region — restore the ciphertext and
+                // crack this sector's own key.
+                chunk.copy_from_slice(&original);
+                if let Some(fresh) = stevenson::crack_title_key(chunk) {
+                    *title_key = fresh;
+                }
+                lfsr::descramble_sector(title_key, chunk);
+            }
+        }
+    }
+}
+
 /// Check if a sector has the CSS scramble flag set.
 ///
 /// This is the RAW flag test — bits 4-5 of the sub-header byte 0x14 — used by

src/css/stevenson.rs

@@ -260,9 +260,6 @@ pub fn crack_title_key(sector: &[u8]) -> Option<[u8; 5]> {
     result
 }
 
-/// Inner body of [`crack_title_key`] — the actual AttackPattern search. Split
-/// out so the public entry point can wall-clock the whole attempt for the
-/// runaway guard without threading a timer through every return path.
 /// AttackPattern crib: the predicted 10-byte plaintext at byte 0x80.
 ///
 /// Scans the clear header `sec[0x00..0x80]` (never scrambled) for the longest

src/decrypt.rs

@@ -123,6 +123,12 @@ pub fn decrypt_threads() -> usize {
     if explicit > 0 {
         return explicit;
     }
+    // Resolve the `FREEMKV_THREADS` env var + `available_parallelism()` ONCE and
+    // cache it — this runs on the per-buffer decrypt hot path, and a getenv +
+    // String alloc + parallelism syscall per call is pure overhead. The explicit
+    // `set_decrypt_threads` override above still takes effect dynamically.
+    static DEFAULT_THREADS: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
+    *DEFAULT_THREADS.get_or_init(|| {
         let env = std::env::var("FREEMKV_THREADS")
             .ok()
             .and_then(|v| v.parse::<usize>().ok())
@@ -134,6 +140,7 @@ pub fn decrypt_threads() -> usize {
             .map(|n| n.get())
             .unwrap_or(2);
         cores.clamp(1, MAX_THREADS)
+    })
 }
 
 /// Resolved decryption state from disc scanning.
@@ -418,49 +425,13 @@ fn decrypt_sectors_impl(
             dropped_bytes.into_inner()
         }
         DecryptKeys::Css { title_key } => {
-            // CSS has no supplied key list: the ONLY source of a title key is
+            // CSS SELF-recovers: the title key changes per VOB region and is
-            // cracking the data, and the key changes per VTS/VOB region. So
+            // re-cracked constantly, but always FROM THE DATA ITSELF — no external
-            // `title_key` is a CACHE of the last crack, not a fixed disc key —
+            // input. So the whole descramble-and-rekey is self-contained here (see
-            // applying it blindly across a region boundary descrambles with the
+            // `css::descramble_region`), and CSS does not need the post-decrypt
-            // wrong key (valid headers, garbage payload). Validate it on every
+            // recovery seam that AACS key-fetch / FMTS segment-skip use (those DO
-            // scrambled sector and re-crack on a miss (libdvdcss's on-demand
+            // consume external inputs a `decrypt_sectors` caller cannot supply).
-            // per-region rekey; the same validate-then-rekey shape the AACS arm
+            css::descramble_region(buf, title_key);
-            // above uses, but re-cracking instead of picking from a list).
-            //
-            // The clear header (<0x80) is never scrambled, so its periodic crib
-            // predicts the plaintext at 0x80. Descramble with the cached key; if
-            // the crib fails to reappear the key region changed (or the primed
-            // key was wrong) — restore the ciphertext, re-crack from this very
-            // sector, and descramble again. A crib-less sector (no periodic run)
-            // can be neither validated nor cracked, so it rides the cached key —
-            // correct, because it lives in the same region as the nearby crib
-            // sector that set the cache.
-            for chunk in buf.chunks_mut(2048) {
-                if chunk.len() < 2048 || !css::is_scrambled(chunk) {
-                    continue;
-                }
-                let crib = css::stevenson::attack_crib(chunk);
-                // Snapshot the ciphertext into a stack buffer (chunk is exactly
-                // 2048 here — guaranteed by the `< 2048` continue above) only
-                // when there's a crib to validate against, so the common
-                // cache-hit path costs no per-sector heap allocation.
-                let mut original = [0u8; 2048];
-                if crib.is_some() {
-                    original.copy_from_slice(chunk);
-                }
-                css::lfsr::descramble_sector(title_key, chunk);
-                if let Some(crib) = crib {
-                    if chunk[0x80..0x80 + 10] != crib[..] {
-                        // Cached key is stale for this region — restore the
-                        // ciphertext and crack this sector's own key.
-                        chunk.copy_from_slice(&original);
-                        if let Some(fresh) = css::stevenson::crack_title_key(chunk) {
-                            *title_key = fresh;
-                        }
-                        css::lfsr::descramble_sector(title_key, chunk);
-                    }
-                }
-            }
             0
         }
     };
@@ -713,6 +684,83 @@ mod tests {
         );
     }
 
+    /// Build a Stevenson-crackable scrambled CSS sector for `title_key` (mirrors
+    /// `crackable_sector` in the css::mod tests): a periodic run in the clear
+    /// header continues past 0x80 into the encrypted region, so
+    /// `stevenson::crack_title_key` recovers the key. Distinct `seed` values give
+    /// two sectors different cribs, standing in for two VOB regions.
+    fn crackable_css_sector(title_key: &[u8; 5], seed: &[u8; 5]) -> Vec<u8> {
+        const RUN_START: usize = 0x59;
+        const SEED_OFFSET: usize = 0x54;
+        const PERIOD: usize = 8;
+        let mut plaintext = vec![0u8; 2048];
+        plaintext[0x00..0x04].copy_from_slice(&css::PACK_START);
+        plaintext[0x14] = 0x10; // scramble flag
+        let pat: Vec<u8> = (0..PERIOD)
+            .map(|k| (0xA0u8.wrapping_add(k as u8)) ^ 0x5A)
+            .collect();
+        for (i, b) in plaintext.iter_mut().enumerate().skip(RUN_START) {
+            *b = pat[i % PERIOD];
+        }
+        plaintext[SEED_OFFSET..SEED_OFFSET + 5].copy_from_slice(seed);
+        css::lfsr::scramble_sector(title_key, &mut plaintext);
+        plaintext
+    }
+
+    /// CHARACTERIZATION (recovery refactor safety net): the CSS arm's per-region
+    /// re-crack (the `title_key` cache is stale for a new VOB region → restore
+    /// ciphertext, `crack_title_key` this sector, re-descramble). Two crackable
+    /// sectors scrambled under DIFFERENT keys sit back-to-back; the cache is
+    /// primed to the FIRST key. Sector 0 rides the cache (crib matches); sector 1
+    /// must trip the crib mismatch and re-crack to its own key. Both must land
+    /// correct plaintext, and the cache must end on region 1's key.
+    ///
+    /// This behaviour currently lives inline in `decrypt_sectors` (the `Css`
+    /// arm). It is the delicate logic the recovery refactor will move to the
+    /// input-stream seam, so it must stay green byte-for-byte across that move.
+    #[test]
+    fn css_region_change_recracks_the_title_key() {
+        let key_a = [0x11, 0x22, 0x33, 0x44, 0x55];
+        let key_b = [0xAA, 0xBB, 0xCC, 0xDD, 0xEE];
+        let sector_a = crackable_css_sector(&key_a, &[0x01, 0x02, 0x03, 0x04, 0x05]);
+        let sector_b = crackable_css_sector(&key_b, &[0x09, 0x08, 0x07, 0x06, 0x05]);
+
+        // Expected plaintext bodies: each sector descrambled under its true key.
+        let mut plain_a = sector_a.clone();
+        css::lfsr::descramble_sector(&key_a, &mut plain_a);
+        let mut plain_b = sector_b.clone();
+        css::lfsr::descramble_sector(&key_b, &mut plain_b);
+
+        let mut buf = Vec::with_capacity(4096);
+        buf.extend_from_slice(&sector_a);
+        buf.extend_from_slice(&sector_b);
+
+        // Cache primed to region A's key (as if A was the last crack). CSS
+        // descramble-and-rekey lives in `css::descramble_region` (the recovery
+        // seam calls it); the region change must re-crack region B's key.
+        let mut ended = key_a;
+        css::descramble_region(&mut buf, &mut ended);
+
+        assert_eq!(
+            &buf[0x80..2048],
+            &plain_a[0x80..2048],
+            "sector 0 rides the cached key (crib matches, no re-crack)"
+        );
+        assert_eq!(
+            &buf[2048 + 0x80..4096],
+            &plain_b[0x80..2048],
+            "sector 1 re-cracks its own region key and descrambles correctly"
+        );
+        // The cache must have advanced to a key that descrambles region B.
+        let mut check_b = sector_b.clone();
+        css::lfsr::descramble_sector(&ended, &mut check_b);
+        assert_eq!(
+            &check_b[0x80..2048],
+            &plain_b[0x80..2048],
+            "the ended cache key must round-trip region B's body"
+        );
+    }
+
     /// Mixed 3-unit buffer: only the in-content SCRAMBLED unit is counted; an
     /// in-content CLEAR unit and an out-of-content SCRAMBLED unit are both skipped.
     #[test]
@@ -912,11 +960,12 @@ mod tests {
     /// fixed wrong key -> the body no longer matches the plaintext.
     #[test]
     fn css_descrambles_with_title_key() {
-        let title_key = [0x42, 0x13, 0x37, 0xBE, 0xEF];
+        let mut title_key = [0x42, 0x13, 0x37, 0xBE, 0xEF];
         let seed = [0xDE, 0xAD, 0xBE, 0xEF, 0x42];
         let (mut sector, plaintext) = make_css_sector(&title_key, &seed, 0xA5);
-        let mut keys = DecryptKeys::Css { title_key };
+        // CSS descramble lives in `css::descramble_region` (the recovery seam
-        decrypt_sectors(&mut sector, &mut keys, 0).expect("CSS decrypt is Ok");
+        // calls it); `decrypt_sectors` only flags CSS sectors for recovery.
+        css::descramble_region(&mut sector, &mut title_key);
         assert_eq!(
             &sector[0x80..2048],
             &plaintext[0x80..2048],
@@ -945,8 +994,8 @@ mod tests {
         let (s1, p1) = make_css_sector(&title_key, &[0x66, 0x77, 0x88, 0x99, 0xAA], 0xC3);
         let mut buf = s0;
         buf.extend_from_slice(&s1);
-        let mut keys = DecryptKeys::Css { title_key };
+        let mut title_key = title_key;
-        decrypt_sectors(&mut buf, &mut keys, 0).expect("CSS multi-sector decrypt is Ok");
+        css::descramble_region(&mut buf, &mut title_key);
         assert_eq!(
             &buf[0x80..2048],
             &p0[0x80..2048],
@@ -1024,8 +1073,8 @@ mod tests {
         buf.extend_from_slice(&s1);
 
         // Cache primed to key_a only — exactly what the one-shot scan crack yields.
-        let mut keys = DecryptKeys::Css { title_key: key_a };
+        let mut title_key = key_a;
-        decrypt_sectors(&mut buf, &mut keys, 0).expect("CSS multi-region decrypt is Ok");
+        css::descramble_region(&mut buf, &mut title_key);
 
         assert_eq!(
             &buf[0x80..2048],
@@ -1038,13 +1087,10 @@ mod tests {
             "region B sector must descramble after the path re-cracks its own key"
         );
         // The cache must have advanced to region B's key.
-        match keys {
+        assert_eq!(
-            DecryptKeys::Css { title_key } => assert_eq!(
             title_key, key_b,
             "cache must hold region B's key after the rekey"
-            ),
+        );
-            _ => unreachable!(),
-        }
     }
 
     /// The CSS path leaves UNSCRAMBLED sectors (flag clear) byte-for-byte

src/disc/extract.rs

@@ -22,7 +22,6 @@ use crate::sector::{DecryptingSectorSource, SectorSource};
 use crate::udf::{self, DirEntry, UdfFs};
 use std::io::Write;
 use std::path::{Path, PathBuf};
-use std::sync::atomic::Ordering;
 
 use crate::consts::{SECTOR_BYTES, SECTOR_BYTES_U64};
 /// AACS aligned unit = 3 sectors / 6144 bytes. Content reads are issued in
@@ -66,10 +65,9 @@ pub struct FileResult {
     pub path: PathBuf,
     /// Bytes written that decrypted cleanly.
     pub bytes_good: u64,
-    /// Bytes lost to unreadable sectors (zero-filled holes).
+    /// Bytes lost — unreadable sectors AND undecryptable units both land here
+    /// (extract fails a bad decrypt loud, so it is zero-filled like a bad sector).
     pub bytes_unreadable: u64,
-    /// Bytes lost to undecryptable AACS/CSS units (still ciphertext / dropped).
-    pub bytes_undecryptable: u64,
     /// True when the file was fully written (renamed from `.partial`).
     pub complete: bool,
 }
@@ -81,10 +79,8 @@ pub struct ExtractResult {
     pub files: Vec<FileResult>,
     /// Aggregate good bytes across all files.
     pub bytes_good: u64,
-    /// Aggregate unreadable (bad-sector) bytes.
+    /// Aggregate lost bytes — bad sectors AND undecryptable units (one bucket).
     pub bytes_unreadable: u64,
-    /// Aggregate undecryptable (decrypt-loss) bytes.
-    pub bytes_undecryptable: u64,
     /// True when every file completed and no loss was recorded.
     pub complete: bool,
     /// True when the run stopped early on an interrupt / progress halt.
@@ -92,11 +88,10 @@ pub struct ExtractResult {
 }
 
 impl ExtractResult {
-    /// Total bytes lost (unreadable + undecryptable). A non-zero value means
+    /// Total bytes lost. A non-zero value means the extraction is holed; the CLI
-    /// the extraction is holed; the CLI exits non-zero so a script can re-run
+    /// exits non-zero so a script can re-run through the `iso://` multipass path.
-    /// through the `iso://` multipass path.
     pub fn bytes_lost(&self) -> u64 {
-        self.bytes_unreadable + self.bytes_undecryptable
+        self.bytes_unreadable
     }
 }
 
@@ -197,7 +192,6 @@ impl Disc {
         // borrowing wrapper (so the caller keeps `reader`), swap keys per CSS
         // VTS group via `set_keys`; AACS/None keep `base_keys` throughout.
         let mut dec = DecryptingSectorSource::new(Borrowed(reader), base_keys.clone());
-        let decrypt_loss = dec.decrypt_loss();
 
         let mut result = ExtractResult::default();
         let total_bytes = required;
@@ -232,26 +226,15 @@ impl Disc {
                 }
             }
 
-            // Acquire (rather than Relaxed) on these per-file delta loads:
+            // A unit that fails to decrypt fails the read loud (extract runs
-            // `extract_tree` drives `dec` single-threaded so there is no race
+            // non-tolerate), so extract_one_file already zero-filled it and
-            // today, and Acquire costs nothing on x86. Note this is only half
+            // counted it in bytes_unreadable — one 'lost' bucket covers both
-            // the synchronisation: the paired counter store
+            // media damage and decrypt failure.
-            // (sector/decrypting.rs `fetch_add`) is Relaxed, so an Acquire
+            let (fr, halted) =
-            // load alone does NOT yet establish a happens-before edge. Before
-            // file extraction is parallelised, upgrade that store to Release
-            // (or stronger) so the delta cannot read a stale counter.
-            let before_loss = decrypt_loss.load(Ordering::Acquire);
-            let (mut fr, halted) =
                 extract_one_file(&mut dec, dest, pf, total_bytes, &mut done_bytes, opts)?;
-            let after_loss = decrypt_loss.load(Ordering::Acquire);
-            fr.bytes_undecryptable = after_loss.saturating_sub(before_loss);
-            fr.bytes_good = fr.bytes_good.saturating_sub(fr.bytes_undecryptable);
 
             result.bytes_good = result.bytes_good.saturating_add(fr.bytes_good);
             result.bytes_unreadable = result.bytes_unreadable.saturating_add(fr.bytes_unreadable);
-            result.bytes_undecryptable = result
-                .bytes_undecryptable
-                .saturating_add(fr.bytes_undecryptable);
             result.files.push(fr);
             if halted {
                 result.halted = true;
@@ -261,7 +244,6 @@ impl Disc {
 
         result.complete = !result.halted
             && result.bytes_unreadable == 0
-            && result.bytes_undecryptable == 0
             && result.files.iter().all(|f| f.complete);
         Ok(result)
     }
@@ -492,7 +474,6 @@ fn extract_one_file<S: SectorSource>(
         path: pf.host_rel.clone(),
         bytes_good: 0,
         bytes_unreadable: 0,
-        bytes_undecryptable: 0,
         complete: false,
     };
 
@@ -1585,10 +1566,6 @@ mod tests {
             res.bytes_unreadable, 0,
             "per-extent unit base must keep the second extent off the hole path"
         );
-        assert_eq!(
-            res.bytes_undecryptable, 0,
-            "clear units decrypt-restore clean"
-        );
         assert!(
             res.complete,
             "a clean multi-extent AACS file extracts complete"

src/disc/mod.rs

@@ -2041,10 +2041,10 @@ impl Disc {
     /// 1. Real titles (`size_bytes ≤ capacity_bytes`) before virtual
     ///    composites. The capacity check is a hard "physically
     ///    possible data on this disc" gate.
-    /// 2. Among real titles, fewer clips first. A 1-clip playlist is
+    /// 2. Among real titles, LARGEST physical size first — the main
-    ///    the canonical main feature; multi-clip playlists are either
+    ///    feature is the biggest real title on the disc. (This replaced
-    ///    chapter-stitched (small count) or virtual composites
+    ///    the old clip-count ordering, which mis-ranked chapter-per-clip
-    ///    (large count). Fewer wins.
+    ///    discs like Fast & Furious.)
     /// 3. Tiebreak on longer duration first.
     ///
     /// **Effect on non-branching discs:** unchanged — the main movie
@@ -2616,6 +2616,28 @@ impl Disc {
         self.ensure_decryptable_keys(raw, keys)
     }
 
+    /// Upfront FMTS (AACS 2.1) key gate, parallel to
+    /// [`ensure_title_decryptable`](Self::ensure_title_decryptable). A 2.1 disc
+    /// carries forensic variant segments that need segment (variant) keys the
+    /// unit-key path cannot provide. When
+    /// [`BYPASS_FMTS_KEY`](crate::aacs::segment::BYPASS_FMTS_KEY) is `false`,
+    /// their absence is a hard upfront failure ([`Error::FmtsKeyMissing`]) — the
+    /// same policy as a missing unit key, so a forensic-holed rip is refused, not
+    /// produced. When `true` (the default today) the segments are skipped as
+    /// expected loss and this passes. `raw` mode and non-FMTS discs always pass.
+    pub fn ensure_forensic_segments_decryptable(&self, raw: bool) -> Result<()> {
+        if raw || crate::aacs::segment::BYPASS_FMTS_KEY {
+            return Ok(());
+        }
+        // A 2.1 (FMTS) disc carries forensic variant segments with no segment-key
+        // source (none exists yet), so its variant segments cannot be opened.
+        // Refuse upfront rather than emit a forensic-holed rip.
+        if self.format == DiscFormat::Fmts {
+            return Err(Error::FmtsKeyMissing);
+        }
+        Ok(())
+    }
+
     /// Inject pre-resolved AACS unit keys into a scanned disc — the deferred-mux
     /// / resume path. The keys come from the mapfile's `# freemkv-uk:` header
     /// (persisted at sweep time when the disc was keyed), so the mux decrypts
@@ -3231,25 +3253,37 @@ impl Disc {
         // ISO file: if resuming and mapfile has Finished ranges, open existing;
         // otherwise create fresh and pre-size to total_bytes (sparse holes for
         // non-tried regions).
-        let is_regular = std::fs::metadata(path)
+        //
-            .map(|m| m.file_type().is_file())
+        // `is_regular` MUST be read from the OPEN file handle, not from
-            .unwrap_or(false);
+        // `metadata(path)` — on a fresh rip the path does not exist yet, so a
-        let file = if resume
+        // pre-create `metadata(path)` always fails (is_regular=false), which both
+        // skips the pre-size AND makes `SweepSink::close` swallow a real
+        // `sync_all()` failure on the just-written ISO as if it were /dev/null.
+        let (file, is_regular) = if resume
             && std::fs::metadata(path)
                 .map(|m| m.len() > 0)
                 .unwrap_or(false)
         {
-            std::fs::OpenOptions::new()
+            let f = std::fs::OpenOptions::new()
                 .write(true)
                 .open(path)
-                .map_err(|e| Error::IoError { source: e })?
+                .map_err(|e| Error::IoError { source: e })?;
+            let reg = f
+                .metadata()
+                .map(|m| m.file_type().is_file())
+                .unwrap_or(false);
+            (f, reg)
         } else {
             let f = std::fs::File::create(path).map_err(|e| Error::IoError { source: e })?;
-            if is_regular {
+            let reg = f
+                .metadata()
+                .map(|m| m.file_type().is_file())
+                .unwrap_or(false);
+            if reg {
                 f.set_len(total_bytes)
                     .map_err(|e| Error::IoError { source: e })?;
             }
-            f
+            (f, reg)
         };
 
         // Wrap the raw `File` in our bounded-cache `WritebackFile`

src/error.rs

@@ -88,6 +88,7 @@ pub const E_NO_DISC_KEY: u16 = 7022;
 pub const E_CSS_KEY_MISSING: u16 = 7023;
 pub const E_AACS_NO_HOST_CERT: u16 = 7024;
 pub const E_AACS_BUS_KEY_UNAVAILABLE: u16 = 7025;
+pub const E_FMTS_KEY_MISSING: u16 = 7026;
 
 // Keydb (8xxx)
 pub const E_KEYDB_CONNECT: u16 = 8000;
@@ -360,6 +361,15 @@ pub enum Error {
     /// time and no handshake runs.
     AacsBusKeyUnavailable,
 
+    /// AACS 2.1 (FMTS) disc carries forensic variant segments, but no segment
+    /// (variant) key is available to open them, and `BYPASS_FMTS_KEY` is `false`
+    /// (strict mode). Raised UPFRONT — before the mux — exactly like a missing
+    /// unit key, so a 2.1 disc that would rip with holes is refused rather than
+    /// silently producing a forensic-holed output. When `BYPASS_FMTS_KEY` is
+    /// `true` (the default today) this is never raised: the bulk decodes with the
+    /// unit key and the forensic segments are skipped as expected loss.
+    FmtsKeyMissing,
+
     // Keydb (8xxx)
     KeydbConnect {
         host: String,
@@ -573,6 +583,7 @@ impl Error {
             Error::CssKeyMissing => E_CSS_KEY_MISSING,
             Error::AacsNoHostCert { .. } => E_AACS_NO_HOST_CERT,
             Error::AacsBusKeyUnavailable => E_AACS_BUS_KEY_UNAVAILABLE,
+            Error::FmtsKeyMissing => E_FMTS_KEY_MISSING,
             Error::KeydbConnect { .. } => E_KEYDB_CONNECT,
             Error::KeydbHttp { .. } => E_KEYDB_HTTP,
             Error::KeydbInvalid => E_KEYDB_INVALID,
@@ -1209,6 +1220,7 @@ mod tests {
             E_CSS_KEY_MISSING,
             E_AACS_NO_HOST_CERT,
             E_AACS_BUS_KEY_UNAVAILABLE,
+            E_FMTS_KEY_MISSING,
             E_KEYDB_CONNECT,
             E_KEYDB_HTTP,
             E_KEYDB_INVALID,

src/mux/resolve.rs

@@ -358,6 +358,12 @@ pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::S
             // case must NOT raise a false E7023.
             disc.ensure_title_decryptable(opts.raw, &keys, title_is_clear)
                 .map_err(|e| -> io::Error { e.into() })?;
+            // Upfront FMTS gate, parallel to the unit-key gate above. With
+            // BYPASS_FMTS_KEY this is a no-op and a 2.1 disc's forensic units are
+            // concealed as ordinary decrypt loss below; without it, a 2.1 disc
+            // lacking segment keys fails here rather than emitting a holed mux.
+            disc.ensure_forensic_segments_decryptable(opts.raw)
+                .map_err(|e| -> io::Error { e.into() })?;
             // Correct TrueHD channel counts (MPLS understates 7.1/Atmos as 5.1)
             // by probing the first DECRYPTED access units of the chosen title.
             // A fresh reader avoids disturbing the mux reader below. Skipped in
@@ -638,15 +644,17 @@ pub fn build_iso_pipeline<S: SectorSource + Send + 'static>(
             .tolerate_decrypt_loss();
     // Install the fresh-key-on-failure callback (if any) so a unit no held key
     // decrypts is re-tried via the application's key source before being counted
-    // as loss.
+    // as loss. An AACS 2.1 forensic-segment unit that no key opens is just an
+    // undecryptable unit like any other: concealed and counted as decrypt loss —
+    // a loss is a loss, no FMTS special casing.
     if let Some(cb) = fetch {
         decrypting = decrypting.with_key_fetch(cb);
     }
-    // Grab the decrypt-loss counter before the decorator is moved into the
+    // Grab the loss counters before the decorator is moved into the producer
-    // producer thread. It tracks bytes of scrambled AACS units no key could
+    // thread. It tracks bytes of scrambled AACS units no key could decrypt —
-    // decrypt — silent loss the demux drops; the consuming stream surfaces it
+    // silent loss the demux drops; the consuming stream surfaces it through
-    // through `lost_bytes()` so the mux abort gate sees a partial decrypt
+    // `lost_bytes()` so the mux abort gate sees a partial decrypt failure rather
-    // failure rather than a clean rip.
+    // than a clean rip. Forensic (2.1) undecryptable units land here too.
     let decrypt_loss = decrypting.decrypt_loss();
 
     // Wrong-substream fix (Silence-of-the-Lambs): before the prefetcher takes

src/sector/decrypting.rs

@@ -38,16 +38,6 @@ use super::SectorSource;
 /// so it can ride the mux highway's producer thread.
 pub type KeyFetch = std::sync::Arc<dyn Fn(&[Vec<u8>]) -> Vec<[u8; 16]> + Send + Sync>;
 
-/// Cap on how many times one decorator will call the fetch closure over its
-/// lifetime — bounds key-server traffic to roughly O(distinct CPS units) even
-/// if scrambled units keep arriving. A disc has only a handful of unit keys.
-const MAX_FETCH_CALLS: usize = 16;
-
-/// Cap on how many still-scrambled sample units are handed to the fetch
-/// closure per call — a few samples are plenty for a key service to identify
-/// and validate the key, and it bounds the request size.
-const MAX_FETCH_SAMPLES: usize = 8;
-
 /// Cap on how many per-unit decrypt-verify-failure diagnostics one read emits.
 /// The diagnostic runs only on the failure (cold) path and bounds log volume so
 /// a large undecryptable range can't flood the device log; the first few units
@@ -109,21 +99,12 @@ pub struct DecryptingSectorSource<S: SectorSource> {
     ///
     /// [`decrypt_loss`]: Self::decrypt_loss
     decrypt_dropped: Arc<AtomicU64>,
-    /// Optional "fetch a fresh key for THIS data" callback (see [`KeyFetch`]).
+    /// The miss policy (see [`crate::sector::recovery::Recover`]) — a generic,
-    /// `None` for the common case (keys fully resolved up front); set via
+    /// scheme-neutral recovery the input stream (L3) installs and this decorator
-    /// [`with_key_fetch`](Self::with_key_fetch) by an application that wants
+    /// (L2) executes at the one seam when a content unit will not decrypt. `None`
-    /// to ask its key source for a key when a unit fails to decrypt.
+    /// = no recovery (a miss is loss). Installed via
-    fetch: Option<KeyFetch>,
+    /// [`with_key_fetch`](Self::with_key_fetch).
-    /// Fingerprints (hash over the unit ciphertext) of failing units a fetch
+    recovery: Option<crate::sector::recovery::Recover>,
-    /// already returned NO new key for. A later failure re-asks the source only
-    /// for units NOT in this set — so on a multi-CPS disc the source is still
-    /// asked for the *second* CPS unit's key even after the first came back dry
-    /// (the old global latch blocked that), while the *same* failing unit is
-    /// never re-fetched (and the total is still bounded by `MAX_FETCH_CALLS`).
-    fetch_dry: std::collections::HashSet<u64>,
-    /// How many times the fetch closure has been invoked, capped at
-    /// [`MAX_FETCH_CALLS`].
-    fetch_calls: usize,
     /// Verify-only mode: a read decrypt-CHECKS a scratch copy of the bytes (to
     /// detect undecryptable units) but NEVER mutates `buf` — the inner
     /// ciphertext is returned unchanged. This is what makes a multipass sweep
@@ -173,9 +154,10 @@ impl<S: SectorSource> DecryptingSectorSource<S> {
             unit_key_idx: 0,
             unit_base: 0,
             decrypt_dropped: Arc::new(AtomicU64::new(0)),
-            fetch: None,
+            // No recovery by default. CSS self-decrypts in `decrypt_sectors`
-            fetch_dry: std::collections::HashSet::new(),
+            // (needs no external input); AACS installs a key-fetch via
-            fetch_calls: 0,
+            // `with_key_fetch`.
+            recovery: None,
             verify_only: false,
             content_ranges: None,
             scratch: Vec::new(),
@@ -238,7 +220,7 @@ impl<S: SectorSource> DecryptingSectorSource<S> {
     /// for [`DecryptKeys::Aacs`]; ignored otherwise. The library makes no network
     /// call — `cb` is the application's seam to its key source.
     pub fn with_key_fetch(mut self, cb: KeyFetch) -> Self {
-        self.fetch = Some(cb);
+        self.recovery = Some(crate::sector::recovery::key_fetch(cb));
         self
     }
 
@@ -286,86 +268,6 @@ impl<S: SectorSource> DecryptingSectorSource<S> {
         }
     }
 
-    /// Collect the still-scrambled aligned units in `buf`, hand them to the
-    /// fetch callback, add any returned keys not already held to the AACS
-    /// pool (the CACHE — every later unit this pass, and any later read, reuses
-    /// them), and re-decrypt `buf`. Returns the post-retry dropped-byte count
-    /// (equal to `prev_dropped` when the callback could not help). The re-decrypt
-    /// is content-gated identically to the first read so a non-content unit is
-    /// never re-attempted. Caller guarantees the keys are `DecryptKeys::Aacs`, a
-    /// callback is installed, and the call budget is not yet spent.
-    fn fetch_failed_units(
-        &mut self,
-        buf: &mut [u8],
-        lba: u32,
-        content: Option<&[(u32, u32)]>,
-        prev_dropped: usize,
-    ) -> usize {
-        let unit_len = crate::aacs::content::ALIGNED_UNIT_LEN;
-        // Gather up to MAX_FETCH_SAMPLES still-scrambled aligned units — the
-        // exact on-disc ciphertext no held key could open. A trailing partial
-        // unit (chunks_exact remainder) can't be a whole scrambled unit, so
-        // skipping it is correct.
-        let mut samples: Vec<Vec<u8>> = Vec::new();
-        for chunk in buf.chunks_exact(unit_len) {
-            if crate::aacs::content::aacs_unit_needs_decrypt(chunk) {
-                samples.push(chunk.to_vec());
-                if samples.len() >= MAX_FETCH_SAMPLES {
-                    break;
-                }
-            }
-        }
-        if samples.is_empty() {
-            return prev_dropped;
-        }
-        // Skip the call when EVERY failing unit here is one a prior fetch already
-        // came back empty for — re-asking the identical ciphertext only burns a
-        // key-server request. A unit we have NOT asked about yet (e.g. a second
-        // CPS unit on a multi-CPS disc) still gets its one chance, where the old
-        // global `fetch_spent` latch wrongly blocked it.
-        let fps: Vec<u64> = samples.iter().map(|s| Self::sample_fp(s)).collect();
-        if fps.iter().all(|fp| self.fetch_dry.contains(fp)) {
-            return prev_dropped;
-        }
-        // Ask the application's key source for keys that open this ciphertext.
-        self.fetch_calls += 1;
-        let fresh = match self.fetch.as_ref() {
-            Some(cb) => cb(&samples),
-            None => return prev_dropped,
-        };
-        // Add only keys we don't already hold (dedup by value).
-        let mut added = 0usize;
-        if let DecryptKeys::Aacs { unit_keys, .. } = &mut self.keys {
-            for k in fresh {
-                if !unit_keys.iter().any(|(_, have)| *have == k) {
-                    let idx = unit_keys.len() as u32;
-                    unit_keys.push((idx, k));
-                    added += 1;
-                }
-            }
-        }
-        if added == 0 {
-            // Nothing new for THESE units — remember them so we don't re-ask the
-            // same ciphertext, but leave the door open for other units.
-            self.fetch_dry.extend(fps);
-            return prev_dropped;
-        }
-        // Retry now that the pool has grown; a unit that still won't decrypt is
-        // genuine loss. A retry error must not mask the original count.
-        Self::decrypt_buf(buf, &mut self.keys, self.unit_key_idx, lba, content)
-            .unwrap_or(prev_dropped)
-    }
-
-    /// Stable per-run fingerprint of a failing unit's ciphertext, for the
-    /// `fetch_dry` set. `DefaultHasher` is fixed-seed, so equal samples map to
-    /// equal fingerprints within a process — all the dedup needs.
-    fn sample_fp(sample: &[u8]) -> u64 {
-        use std::hash::{Hash, Hasher};
-        let mut h = std::collections::hash_map::DefaultHasher::new();
-        sample.hash(&mut h);
-        h.finish()
-    }
-
     /// Emit a bounded, structured diagnostic for each undecryptable unit in a
     /// failed verify read. Called only on the failure (cold) path. On a fresh
     /// rip `buf` holds the post-decrypt bytes straight off the drive, so the
@@ -521,8 +423,11 @@ impl<S: SectorSource> SectorSource for DecryptingSectorSource<S> {
         // recover a unit no held key opened.
         let content = self.content_ranges.clone(); // cheap Arc bump; frees the &self borrow
         let content_ref = content.as_deref();
-        // Whether a fresh-key fetch is still worth attempting on this decorator.
+        // Copy out the small Copy fields the seam needs, so the `&mut self.recovery`
-        let fetch_viable = self.fetch.is_some() && self.fetch_calls < MAX_FETCH_CALLS;
+        // borrow below does not collide with reads of other `self` fields. The
+        // recovery closure self-limits (its budget lives in its captures), so the
+        // decorator simply calls it whenever there is a miss.
+        let unit_key_idx = self.unit_key_idx;
         // First decrypt, then the FRESH-KEY-ON-FAILURE retry (read → decrypt → on
         // fail fetch a new key → retry → CACHE or fail). This runs in BOTH modes:
         //   * VERIFY-ONLY (multipass sweep): decrypt a reused SCRATCH copy so `buf`
@@ -537,11 +442,11 @@ impl<S: SectorSource> SectorSource for DecryptingSectorSource<S> {
         //   * NORMAL (mux / --no-raw): decrypt `buf` in place, same retry.
         // The fetch re-decrypt targets the post-decrypt buffer (scratch / buf),
         // whose still-scrambled units ARE the failures.
-        let dropped = if self.verify_only {
+        let outcome = if self.verify_only {
             let mut scratch = std::mem::take(&mut self.scratch);
             scratch.clear();
             scratch.extend_from_slice(&buf[..n]);
-            let mut d = match Self::decrypt_buf(
+            let d = match Self::decrypt_buf(
                 &mut scratch,
                 &mut self.keys,
                 self.unit_key_idx,
@@ -554,24 +459,45 @@ impl<S: SectorSource> SectorSource for DecryptingSectorSource<S> {
                     return Err(e);
                 }
             };
-            if d > 0 && fetch_viable {
+            let o = match (d, self.recovery.as_mut()) {
-                d = self.fetch_failed_units(&mut scratch, lba, content_ref, d);
+                (0, _) | (_, None) => crate::sector::recovery::MissOutcome { dropped: d },
+                (d, Some(r)) => {
+                    let rctx = crate::sector::recovery::RecoverCtx {
+                        unit_key_idx,
+                        lba,
+                        content: content.clone(),
+                        prev_dropped: d,
+                    };
+                    r(&mut scratch, &mut self.keys, &rctx)
                 }
+            };
             self.scratch = scratch;
-            d
+            o
         } else {
-            let mut d = Self::decrypt_buf(
+            let d = Self::decrypt_buf(
                 &mut buf[..n],
                 &mut self.keys,
                 self.unit_key_idx,
                 lba,
                 content_ref,
             )?;
-            if d > 0 && fetch_viable {
+            match (d, self.recovery.as_mut()) {
-                d = self.fetch_failed_units(&mut buf[..n], lba, content_ref, d);
+                (0, _) | (_, None) => crate::sector::recovery::MissOutcome { dropped: d },
-            }
+                (d, Some(r)) => {
-            d
+                    let rctx = crate::sector::recovery::RecoverCtx {
+                        unit_key_idx,
+                        lba,
+                        content: content.clone(),
+                        prev_dropped: d,
                     };
+                    r(&mut buf[..n], &mut self.keys, &rctx)
+                }
+            }
+        };
+        // A loss is a loss: whatever recovery could not decrypt (a missing unit
+        // key, or an AACS 2.1 forensic-segment unit with no variant key — same
+        // thing to the read path) is concealed and counted the same way.
+        let dropped = outcome.dropped;
         if dropped > 0 {
             self.decrypt_dropped
                 .fetch_add(dropped as u64, Ordering::Relaxed);
@@ -656,6 +582,11 @@ impl<S: SectorSource> SectorSource for DecryptingSectorSource<S> {
             // fails (no clean data to mux). Scheme-agnostic (only AACS reaches a
             // non-zero count); clear filesystem (gated out) and zero-fill (not
             // scrambled) never get here.
+            //
+            // An undecryptable unit is an undecryptable unit whatever the scheme —
+            // a missing unit key or an AACS 2.1 forensic-segment unit with no
+            // variant key both land here and fail the verify read the same way.
+            // (`dropped > 0` already holds inside the enclosing block.)
             if DECRYPT_VERIFY_READ {
                 // FACT-FINDING: on a fresh rip these bytes came straight off the
                 // drive, so each failing unit's signature (all-zero? entropy?

src/sector/mod.rs

@@ -16,6 +16,7 @@
 pub mod decrypting;
 pub mod file;
 pub mod prefetched;
+pub mod recovery;
 
 use crate::error::Result;
 

src/sector/recovery.rs

@@ -0,0 +1,330 @@
+//! The recovery seam: what a read does when a content unit will not decrypt.
+//!
+//! Per-format miss policy does NOT belong in the generic decrypt decorator
+//! (L2). The input stream (L3, e.g. [`crate::mux::disc::DiscStream`]) knows what
+//! it is reading and installs a [`Recover`] at construction; the decorator
+//! executes it at the one seam and honours the returned outcome. This keeps
+//! "a DVD re-cracks, a BD/UHD fetches a fresh key" out of the decryptor, where
+//! it would otherwise smear across `if`-branches.
+//!
+//! The recovery type ([`Recover`]) names **no encryption scheme**. It is a
+//! generic `FnMut(&mut [u8], &mut DecryptKeys, &RecoverCtx) -> MissOutcome` that
+//! operates on the generic [`DecryptKeys`] the whole decrypt path already uses,
+//! so a scheme is never baked into the type — only into the factory that builds
+//! a recovery:
+//!   * [`none`] — no recovery; a miss is loss (raw sweep / clear).
+//!   * [`key_fetch`] — AACS key-fetch: hand the failing ciphertext to the
+//!     application's key source and add any returned keys to the pool. An AACS
+//!     2.1 forensic-segment unit that no key opens is just an undecryptable unit
+//!     like any other — a loss is a loss, with no FMTS-specific branch here.
+//!
+//! CSS is deliberately NOT on this seam — and the reason is precise: this seam is
+//! for recovery that needs something `decrypt_sectors` does not have (an EXTERNAL
+//! key source for AACS, a segment map for FMTS). CSS's title key changes per VOB
+//! region and is re-cracked constantly, but always FROM THE DATA ITSELF — no
+//! external input — so CSS SELF-recovers inside `decrypt_sectors` (see
+//! [`crate::css::descramble_region`]). The generic type here would accept a CSS
+//! recovery, but CSS has no reason to use it.
+
+use crate::decrypt::DecryptKeys;
+use crate::sector::KeyFetch;
+use std::collections::HashSet;
+use std::sync::Arc;
+
+/// The result of running a recovery on a read's still-scrambled units: how many
+/// bytes remain loss after recovery ran. A loss is a loss — an undecryptable
+/// unit is concealed and counted the same whatever the scheme (an AACS 2.1
+/// forensic-segment unit with no variant key is just another undecryptable
+/// unit).
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct MissOutcome {
+    /// Bytes that remain loss after recovery.
+    pub dropped: usize,
+}
+
+impl MissOutcome {
+    /// All `n` bytes are loss.
+    fn loss(n: usize) -> Self {
+        Self { dropped: n }
+    }
+}
+
+/// Cap on how many times one recovery will call its fetch closure over its
+/// lifetime — bounds key-server traffic to ~O(distinct CPS units) even if
+/// scrambled units keep arriving. A disc has only a handful of unit keys.
+const MAX_FETCH_CALLS: usize = 16;
+
+/// Cap on how many still-scrambled sample units are handed to the fetch closure
+/// per call — a few samples suffice for a key service to identify and validate
+/// the key, and it bounds the request size.
+const MAX_FETCH_SAMPLES: usize = 8;
+
+/// Stable per-run fingerprint of a failing unit's ciphertext, for the dedup set.
+/// `DefaultHasher` is fixed-seed, so equal samples map to equal fingerprints
+/// within a process — all the dedup needs.
+fn sample_fp(sample: &[u8]) -> u64 {
+    use std::hash::{Hash, Hasher};
+    let mut h = std::collections::hash_map::DefaultHasher::new();
+    sample.hash(&mut h);
+    h.finish()
+}
+
+/// Re-decrypt `buf` after the key pool grew, content-gated identically to the
+/// first read so a non-content unit is never re-attempted. Mirrors the
+/// decorator's `decrypt_buf` dispatch.
+fn redecrypt(
+    buf: &mut [u8],
+    keys: &mut DecryptKeys,
+    unit_key_idx: usize,
+    lba: u32,
+    content: Option<&[(u32, u32)]>,
+    prev_dropped: usize,
+) -> usize {
+    match content {
+        Some(ranges) => {
+            crate::decrypt::decrypt_sectors_in_content(buf, keys, unit_key_idx, lba, ranges)
+        }
+        None => crate::decrypt::decrypt_sectors(buf, keys, unit_key_idx),
+    }
+    .unwrap_or(prev_dropped)
+}
+
+/// What a read hands a recovery on a miss: the disc's decrypt parameters and how
+/// many bytes the held keys could not decrypt. Scheme-neutral — a recovery reads
+/// only the generic [`DecryptKeys`] and these fields.
+pub struct RecoverCtx {
+    /// Which AACS unit-key index the read decrypts with (ignored by non-AACS).
+    pub unit_key_idx: usize,
+    /// Base LBA of the read.
+    pub lba: u32,
+    /// The encrypted-content extent map, when the read is content-gated.
+    pub content: Option<Arc<[(u32, u32)]>>,
+    /// Bytes the held keys could not decrypt before recovery ran.
+    pub prev_dropped: usize,
+}
+
+/// A recovery: given a read's still-scrambled `buf` and the **generic**
+/// [`DecryptKeys`], make units decrypt (crack or fetch a key into `keys`) and/or
+/// classify the loss (see [`MissOutcome`]). The type names NO encryption scheme
+/// — the installed recovery decides what to do with the generic keys, so any
+/// scheme (an AACS key-fetch, a future CSS re-crack) is just a different
+/// [`Recover`] the input stream installs. `FnMut` so per-recovery
+/// state (the AACS dedup set / call budget) lives in the closure's captures with
+/// no lock; `Send` so it can ride the mux highway's producer thread.
+pub type Recover = Box<dyn FnMut(&mut [u8], &mut DecryptKeys, &RecoverCtx) -> MissOutcome + Send>;
+
+/// The AACS key-fetch step used by [`key_fetch`]: gather the
+/// still-scrambled units, ask `fetch` for keys, add any new ones to the pool and
+/// re-decrypt. `dry` / `calls` are the caller-owned dedup set and call budget.
+/// Returns the post-retry dropped-byte count.
+fn aacs_fetch_step(
+    dry: &mut HashSet<u64>,
+    calls: &mut usize,
+    fetch: &KeyFetch,
+    buf: &mut [u8],
+    keys: &mut DecryptKeys,
+    ctx: &RecoverCtx,
+) -> usize {
+    let prev_dropped = ctx.prev_dropped;
+    if *calls >= MAX_FETCH_CALLS {
+        return prev_dropped;
+    }
+    let unit_len = crate::aacs::content::ALIGNED_UNIT_LEN;
+    // Gather up to MAX_FETCH_SAMPLES still-scrambled aligned units — the exact
+    // on-disc ciphertext no held key could open. A trailing partial unit
+    // (chunks_exact remainder) can't be a whole scrambled unit, so skipping it is
+    // correct.
+    let mut samples: Vec<Vec<u8>> = Vec::new();
+    for chunk in buf.chunks_exact(unit_len) {
+        if crate::aacs::content::aacs_unit_needs_decrypt(chunk) {
+            samples.push(chunk.to_vec());
+            if samples.len() >= MAX_FETCH_SAMPLES {
+                break;
+            }
+        }
+    }
+    if samples.is_empty() {
+        return prev_dropped;
+    }
+    // Skip the call when EVERY failing unit here is one a prior fetch already
+    // came back empty for — re-asking identical ciphertext only burns a request.
+    // A unit not asked about yet (e.g. a second CPS unit) still gets its chance.
+    let fps: Vec<u64> = samples.iter().map(|s| sample_fp(s)).collect();
+    if fps.iter().all(|fp| dry.contains(fp)) {
+        return prev_dropped;
+    }
+    *calls += 1;
+    let fresh = (fetch)(&samples);
+    // Add only keys we don't already hold (dedup by value).
+    let mut added = 0usize;
+    if let DecryptKeys::Aacs { unit_keys, .. } = keys {
+        for k in fresh {
+            if !unit_keys.iter().any(|(_, have)| *have == k) {
+                let idx = unit_keys.len() as u32;
+                unit_keys.push((idx, k));
+                added += 1;
+            }
+        }
+    }
+    if added == 0 {
+        // Nothing new for THESE units — remember them so we don't re-ask the same
+        // ciphertext, but leave the door open for other units.
+        dry.extend(fps);
+        return prev_dropped;
+    }
+    // Retry now that the pool has grown; a unit that still won't decrypt is
+    // genuine loss. A retry error must not mask the original count.
+    redecrypt(
+        buf,
+        keys,
+        ctx.unit_key_idx,
+        ctx.lba,
+        ctx.content.as_deref(),
+        prev_dropped,
+    )
+}
+
+/// No recovery: a miss is loss. Equivalent to installing nothing — provided so a
+/// caller that wants an explicit "give up" recovery has one.
+pub fn none() -> Recover {
+    Box::new(|_buf, _keys, ctx| MissOutcome::loss(ctx.prev_dropped))
+}
+
+/// AACS key-fetch recovery (BD / UHD): on a miss, ask the application's key
+/// source for a key that opens the failing ciphertext and add it to the pool.
+pub fn key_fetch(fetch: KeyFetch) -> Recover {
+    let mut dry: HashSet<u64> = HashSet::new();
+    let mut calls: usize = 0;
+    Box::new(move |buf, keys, ctx| {
+        MissOutcome::loss(aacs_fetch_step(
+            &mut dry, &mut calls, &fetch, buf, keys, ctx,
+        ))
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::aacs::content::ALIGNED_UNIT_LEN;
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    /// A 6144-byte aligned unit that reads as still-scrambled: CPI bits set on
+    /// byte 0 (so `aacs_unit_encrypted` flags it) and every 192-byte TS-sync
+    /// probe position forced off 0x47. `tag` varies the whole body so distinct
+    /// tags produce distinct fingerprints (mirrors decrypt.rs `scrambled_region`).
+    fn scrambled_unit(tag: u8) -> Vec<u8> {
+        let len = ALIGNED_UNIT_LEN;
+        let mut v: Vec<u8> = (0..len).map(|i| (i as u8).wrapping_mul(31) ^ tag).collect();
+        let mut off = 4;
+        while off < len {
+            v[off] = 0xA5; // never a 0x47 sync
+            off += 192;
+        }
+        v[0] |= 0xC0; // CPI: reads as encrypted content
+        v
+    }
+
+    /// A recovery context reading at clip-relative `lba` with `prev` bytes the
+    /// held keys could not decrypt.
+    fn ctx(lba: u32, prev: usize) -> RecoverCtx {
+        RecoverCtx {
+            unit_key_idx: 0,
+            lba,
+            content: None,
+            prev_dropped: prev,
+        }
+    }
+
+    #[test]
+    fn none_recovers_nothing() {
+        let mut r = none();
+        let mut buf = scrambled_unit(0x33);
+        let mut keys = DecryptKeys::Aacs {
+            unit_keys: vec![],
+            read_data_key: None,
+        };
+        let out = r(&mut buf, &mut keys, &ctx(0, 6144));
+        assert_eq!(out.dropped, 6144);
+    }
+
+    #[test]
+    fn key_fetch_adds_returned_keys_to_the_pool() {
+        // The fetch returns one key; it must be appended to the (empty) pool. We
+        // assert the pool grew (the decrypt itself is exercised end-to-end by the
+        // decorator's integration tests); here we pin the seam's key-plumbing.
+        let calls = Arc::new(AtomicUsize::new(0));
+        let c2 = Arc::clone(&calls);
+        let fetch: KeyFetch = Arc::new(move |samples: &[Vec<u8>]| {
+            c2.fetch_add(1, Ordering::SeqCst);
+            assert!(!samples.is_empty(), "failing ciphertext is forwarded");
+            vec![[0xAB; 16]]
+        });
+        let mut r = key_fetch(fetch);
+        let mut buf = scrambled_unit(0x33);
+        let mut keys = DecryptKeys::Aacs {
+            unit_keys: vec![],
+            read_data_key: None,
+        };
+        r(&mut buf, &mut keys, &ctx(0, ALIGNED_UNIT_LEN));
+        assert_eq!(calls.load(Ordering::SeqCst), 1, "fetch called once");
+        let DecryptKeys::Aacs { unit_keys, .. } = &keys else {
+            unreachable!()
+        };
+        assert_eq!(unit_keys.len(), 1, "returned key added to the pool");
+        assert_eq!(unit_keys[0].1, [0xAB; 16]);
+    }
+
+    #[test]
+    fn key_fetch_does_not_re_ask_dry_ciphertext() {
+        // A fetch that returns nothing marks the ciphertext dry; a second miss on
+        // the SAME ciphertext must not call the fetch again.
+        let calls = Arc::new(AtomicUsize::new(0));
+        let c2 = Arc::clone(&calls);
+        let fetch: KeyFetch = Arc::new(move |_: &[Vec<u8>]| {
+            c2.fetch_add(1, Ordering::SeqCst);
+            Vec::new() // never helps
+        });
+        let mut r = key_fetch(fetch);
+        let mut keys = DecryptKeys::Aacs {
+            unit_keys: vec![],
+            read_data_key: None,
+        };
+        let mut buf = scrambled_unit(0x44);
+        r(&mut buf, &mut keys, &ctx(0, ALIGNED_UNIT_LEN));
+        let mut buf2 = scrambled_unit(0x44); // identical ciphertext
+        r(&mut buf2, &mut keys, &ctx(0, ALIGNED_UNIT_LEN));
+        assert_eq!(
+            calls.load(Ordering::SeqCst),
+            1,
+            "identical dry ciphertext is not re-asked"
+        );
+    }
+
+    #[test]
+    fn key_fetch_call_budget_bounds_fetches() {
+        let calls = Arc::new(AtomicUsize::new(0));
+        let c2 = Arc::clone(&calls);
+        let fetch: KeyFetch = Arc::new(move |_: &[Vec<u8>]| {
+            c2.fetch_add(1, Ordering::SeqCst);
+            Vec::new()
+        });
+        let mut r = key_fetch(fetch);
+        let mut keys = DecryptKeys::Aacs {
+            unit_keys: vec![],
+            read_data_key: None,
+        };
+        // Distinct ciphertext each time so the dry-set never short-circuits; only
+        // the internal call budget should stop the fetch. The closure self-limits,
+        // so the decorator can call it unconditionally.
+        for i in 0..(MAX_FETCH_CALLS as u8 + 5) {
+            let mut buf = scrambled_unit(i);
+            r(&mut buf, &mut keys, &ctx(0, ALIGNED_UNIT_LEN));
+        }
+        assert_eq!(
+            calls.load(Ordering::SeqCst),
+            MAX_FETCH_CALLS,
+            "fetch is capped at MAX_FETCH_CALLS"
+        );
+    }
+}