keys: trust the declared CPS units instead of probing every stream file

1.8.0 read up to 32 units from every stream file before asking any key
source: 1,472 random reads (~95 s over USB) on a 46-file UHD that declares
one CPS unit, answered by one keyserver request.

One declared unit: the main title's samples ask once and the key that
opens them keys every file. Several: one encrypted unit per file, then
each unopened unit asks with 8 of its own units; one answer may open many
files. A source answering no key is not asked again, and a disc no key
opens is refused at once (blanked in a whole-disc copy) rather than
probed. Discs with an FMTS layout in scope keep the probed path.

Key acquisition also logs its phase timings.

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## [Unreleased]
 
+- Key acquisition trusts the disc's declared CPS units. With one declared unit, the main title's samples ask the key sources once and the key that opens them keys every stream file, with no per-file probing: 1.8.0 first read up to 32 units from every file (about 1,500 seeks, ~95 s on a 46-file UHD over USB; now ~3 s). With several declared units, one encrypted unit per file goes into a single request. A source is never asked twice, and when every unit's samples were sent and no key opens a file, it is refused at once (blanked in a whole-disc copy) instead of probed. A disc with an FMTS forensic layout in scope keeps the probed path.
+- Key acquisition logs its timings: evidence read, per-piece probe reads (count, average and slowest), main-title samples, total.
 - Windows: a bare drive letter (`E`) opens that drive, as `E:` does; it was opened as `\\.\E`, which does not exist (E1000).
 - Linux: a bare node name (`sg3`, `sr0`) opens `/dev/sg3` / `/dev/sr0`; it was opened relative to the current directory.
 

src/keys/evidence.rs

@@ -377,6 +377,16 @@ pub(crate) fn pieces(
 /// raw (never decrypted) source the evidence describes. Random access is required.
 pub struct Sampler<'a> {
     reader: &'a mut dyn SectorSource,
+    stats: ReadStats,
+}
+
+/// Read accounting for the sampler's units: how long key acquisition spent on the drive.
+#[derive(Debug, Default, Clone, Copy)]
+pub(crate) struct ReadStats {
+    pub(crate) reads: u32,
+    pub(crate) faults: u32,
+    pub(crate) total: std::time::Duration,
+    pub(crate) slowest: std::time::Duration,
 }
 
 /// A sampled unit: `Enc` ciphertext, `Clear`, or a `Fault` (a soft read error).
@@ -389,7 +399,14 @@ pub(crate) enum Sample {
 impl<'a> Sampler<'a> {
     /// A sampler over the raw source `reader`.
     pub fn new(reader: &'a mut dyn SectorSource) -> Self {
-        Sampler { reader }
+        Sampler {
+            reader,
+            stats: ReadStats::default(),
+        }
+    }
+
+    pub(crate) fn stats(&self) -> ReadStats {
+        self.stats
     }
 
     pub(crate) fn source(&mut self) -> &mut dyn SectorSource {
@@ -399,7 +416,17 @@ impl<'a> Sampler<'a> {
     // The aligned unit at `lba`, or `None` on a short or failed read. A Stop, a gone source
     // or a transport failure is `Err`: not a soft fault.
     pub(crate) fn unit(&mut self, lba: u32) -> Result<Option<Vec<u8>>> {
-        read_unit(self.reader, lba)
+        let t0 = std::time::Instant::now();
+        let r = read_unit(self.reader, lba);
+        let took = t0.elapsed();
+        self.stats.reads += 1;
+        self.stats.total += took;
+        self.stats.slowest = self.stats.slowest.max(took);
+        if matches!(r, Ok(None)) {
+            self.stats.faults += 1;
+        }
+        tracing::trace!(target: "freemkv::keys", lba, took_ms = took.as_millis() as u64, "sample unit read");
+        r
     }
 
     // Up to 32 units on `p`'s grid (KU §2.3 step 7), skipping FMTS segment units, each
@@ -425,10 +452,66 @@ impl<'a> Sampler<'a> {
         Ok((units, out))
     }
 
+    // The first encrypted unit on `p`'s probe grid, stopping there: a stream file sits in one
+    // CPS unit (KS-10), so one unit names its key. (grid units, the unit, soft faults seen).
+    pub(crate) fn first_encrypted(
+        &mut self,
+        p: &Piece,
+        segments: &[(u32, u32)],
+        format: ContentFormat,
+        detector: Detector,
+        halt: &Halt,
+    ) -> Result<(u64, Option<Vec<u8>>, usize)> {
+        let (units, lbas) = probe_lbas(p, segments);
+        let mut faults = 0;
+        for lba in lbas {
+            halt.check()?;
+            match self.unit(lba)? {
+                Some(u) if detector.unit_encrypted(&u, format) => {
+                    return Ok((units, Some(u), faults));
+                }
+                Some(_) => {}
+                None => faults += 1,
+            }
+        }
+        Ok((units, None, faults))
+    }
+
+    // Up to `n` encrypted units on `p`'s probe grid: enough ciphertext to ask a source for a
+    // unit the first request left unopened.
+    pub(crate) fn encrypted_units(
+        &mut self,
+        p: &Piece,
+        segments: &[(u32, u32)],
+        format: ContentFormat,
+        detector: Detector,
+        halt: &Halt,
+        n: usize,
+    ) -> Result<Vec<Vec<u8>>> {
+        let (_, lbas) = probe_lbas(p, segments);
+        let mut out = Vec::new();
+        for lba in lbas {
+            if out.len() >= n {
+                break;
+            }
+            halt.check()?;
+            if let Some(u) = self.unit(lba)?
+                && detector.unit_encrypted(&u, format)
+            {
+                out.push(u);
+            }
+        }
+        Ok(out)
+    }
+
     // Up to `n` encrypted units spread over the main feature.
     pub(crate) fn main_samples(&mut self, main: Option<&MainTitle>, n: usize) -> Vec<Vec<u8>> {
-        main.map(|m| crate::keysource::encrypted_units_in(self.reader, &m.extents, m.format, n))
-            .unwrap_or_default()
+        let t0 = std::time::Instant::now();
+        let out = main
+            .map(|m| crate::keysource::encrypted_units_in(self.reader, &m.extents, m.format, n))
+            .unwrap_or_default();
+        tracing::info!(target: "freemkv::keys", phase = "main_samples", wanted = n, got = out.len(), elapsed_ms = t0.elapsed().as_millis() as u64, "main-title samples read");
+        out
     }
 }
 

src/keys/mod.rs

@@ -362,8 +362,13 @@ impl KeyRing {
         opts: AcquireOptions,
         ctx: &Ctx,
     ) -> Result<KeyResolution> {
+        let t0 = std::time::Instant::now();
         let ev = KeyEvidence::from_disc(disc, reader, scope, ctx)?;
-        Self::acquire(&ev, &mut Sampler::new(reader), sources, opts, ctx)
+        tracing::info!(target: "freemkv::keys", phase = "evidence", pieces = ev.pieces.len(), elapsed_ms = t0.elapsed().as_millis() as u64, "key evidence read from the disc");
+        let t1 = std::time::Instant::now();
+        let r = Self::acquire(&ev, &mut Sampler::new(reader), sources, opts, ctx);
+        tracing::info!(target: "freemkv::keys", phase = "acquire", ok = r.is_ok(), elapsed_ms = t1.elapsed().as_millis() as u64, "key acquisition finished");
+        r
     }
 
     /// A set holding no keys: for a raw copy, or a disc with no AACS.

src/keys/resolve.rs

@@ -487,6 +487,48 @@ struct Scoped<'e> {
 // clip is in scope, and each piece probed by the format's own detector.
 fn probe_scope<'e>(ev: &'e KeyEvidence, sampler: &mut Sampler, halt: &Halt) -> Result<Scoped<'e>> {
     let mut ps: Vec<Probed> = ev.pieces.iter().cloned().map(Probed::new).collect();
+    let (layout, clip) = scope_layout(ev);
+    let segments: Vec<(u32, u32)> = layout
+        .map(|l| sorted_ranges(l.ranges.iter().map(|&(s, e, _)| (s, e)).collect()))
+        .unwrap_or_default();
+    let t0 = std::time::Instant::now();
+    let before = sampler.stats();
+    for p in &mut ps {
+        let tp = std::time::Instant::now();
+        let reads_before = sampler.stats().reads;
+        probe(sampler, p, &segments, ev.container, ev.detector, halt)?;
+        tracing::debug!(
+            target: "freemkv::keys",
+            phase = "probe_piece",
+            piece = p.id(),
+            units = p.units,
+            reads = sampler.stats().reads - reads_before,
+            enc = p.enc.len(),
+            faults = p.faults,
+            elapsed_ms = tp.elapsed().as_millis() as u64,
+            "piece probed"
+        );
+    }
+    let after = sampler.stats();
+    let reads = after.reads - before.reads;
+    let read_ms = (after.total - before.total).as_millis() as u64;
+    tracing::info!(
+        target: "freemkv::keys",
+        phase = "probe_scope",
+        pieces = ps.len(),
+        reads,
+        faults = after.faults - before.faults,
+        read_ms,
+        avg_read_ms = read_ms.checked_div(u64::from(reads)).unwrap_or(0),
+        slowest_read_ms = after.slowest.as_millis() as u64,
+        elapsed_ms = t0.elapsed().as_millis() as u64,
+        "sampled every piece before asking a key source"
+    );
+    Ok(Scoped { ps, layout, clip })
+}
+
+// The forensic layout when its clip is in scope (KU §2.3 step 6), and that clip. Reads nothing.
+fn scope_layout(ev: &KeyEvidence) -> (Option<&super::fmts::Layout>, Vec<(u32, u32)>) {
     let layout = ev.fmts.as_ref();
     let clip: Vec<(u32, u32)> = layout
         .map(|l| {
@@ -497,16 +539,294 @@ fn probe_scope<'e>(ev: &'e KeyEvidence, sampler: &mut Sampler, halt: &Halt) -> R
         })
         .unwrap_or_default();
     let layout = layout.filter(|_| {
-        ps.iter()
+        ev.pieces
+            .iter()
             .any(|p| p.ranges().any(|(s, e)| overlaps(&clip, s, e)))
     });
+    (layout, clip)
+}
+
+// KS-14 [BD] §3.9.3: one declared CPS unit is one Unit Key for every stream file, so the disc
+// is trusted as declared: the main title's samples ask the sources once, and the key that
+// opens them keys every piece in scope. No per-piece probing (32 reads a piece: minutes of
+// seeks on a many-clip disc). `None` (no ciphertext found, or no key opens it) falls back to
+// the probed path, which decides clear content and refusals.
+fn resolve_one_unit(
+    ev: &KeyEvidence,
+    sampler: &mut Sampler,
+    disc_hash: &str,
+    seed: Option<&KeyRing>,
+    run: &mut Run,
+) -> Result<Option<Inner>> {
+    let samples = sampler.main_samples(ev.main.as_ref(), MIN_SAMPLE_UNITS);
+    if samples.is_empty() {
+        return Ok(None);
+    }
+    // Two opened samples prove a key (rule 1), or the one when only one was found.
+    let need = samples.len().min(2);
+    let opener = |run: &Run| {
+        (0..run.pool.len())
+            .find(|&slot| samples.iter().filter(|u| run.opens(u, slot)).count() >= need)
+    };
+    let mut slot = opener(run);
+    // Sample-independent sources (a local keydb) before those that need the samples.
+    let mut order: Vec<usize> = (0..run.sources.len()).collect();
+    order.sort_by_key(|&i| run.sources[i].answer_depends_on_samples());
+    let mut tried: Vec<usize> = Vec::new();
+    run.check_halt()?;
+    for i in order {
+        if slot.is_some() {
+            break;
+        }
+        // Too little ciphertext to ask with unambiguously (KU step 9.2): the probed path decides.
+        if run.sources[i].answer_depends_on_samples() && samples.len() < MIN_SAMPLE_UNITS {
+            continue;
+        }
+        run.check_halt()?;
+        if let Asked::Keys(k, _) = run.ask(i, &samples, false)? {
+            let who = run.sources[i].label();
+            run.add_keys(&k, who);
+            slot = opener(run);
+        }
+        tried.push(i);
+    }
+    let Some(slot) = slot else {
+        // Encrypted, sources asked, no key held at all: one declared unit means nothing in
+        // scope can be keyed, so refuse now rather than probe every piece to the same end.
+        if !tried.is_empty() && run.pool.is_empty() {
+            let failure = run.first_failure.take();
+            return Err(failure.unwrap_or_else(|| missing_error(&ev.scope, disc_hash)));
+        }
+        // Asked already: the probed path never asks these sources again.
+        for i in tried {
+            run.dead[i] = true;
+        }
+        return Ok(None);
+    };
+    let (layout, clip) = scope_layout(ev);
+    let ps: Vec<Probed> = ev
+        .pieces
+        .iter()
+        .cloned()
+        .map(|piece| {
+            let mut p = Probed::new(piece);
+            p.units = p.piece.grid().iter().map(|g| g.1).sum();
+            p.verdict = if p.units == 0 {
+                Verdict::Clear
+            } else {
+                Verdict::Keyed(slot)
+            };
+            p
+        })
+        .collect();
+    tracing::info!(
+        target: "freemkv::keys",
+        phase = "one_cps_unit",
+        pieces = ps.len(),
+        samples = samples.len(),
+        "one CPS unit declared: every piece keyed by the key that opens the main title"
+    );
+    let mut inner = aacs_inner(run);
+    inner.no_stream_files = ev.no_stream_files;
+    let forensic = match layout {
+        None => None,
+        Some(l) => Some(resolve_forensic(sampler, l, seed, run, ev.container)?),
+    };
+    build(&mut inner, &ps, layout, forensic, run, clip);
+    Ok(Some(inner))
+}
+
+// Several declared CPS units: each stream file sits in exactly one (KS-10), so one encrypted
+// unit per piece (the first its probe grid finds) names its key. The main title's samples ask
+// first; pieces no held key opens then ask with their own units, at most `n_decl` requests.
+// `None` (a piece unreadable, or one no key opens) falls back to the probed path.
+fn resolve_per_unit(
+    ev: &KeyEvidence,
+    sampler: &mut Sampler,
+    disc_hash: &str,
+    seed: Option<&KeyRing>,
+    run: &mut Run,
+    n_decl: usize,
+) -> Result<Option<Inner>> {
+    let (layout, clip) = scope_layout(ev);
     let segments: Vec<(u32, u32)> = layout
         .map(|l| sorted_ranges(l.ranges.iter().map(|&(s, e, _)| (s, e)).collect()))
         .unwrap_or_default();
-    for p in &mut ps {
-        probe(sampler, p, &segments, ev.container, ev.detector, halt)?;
+    let mut ps: Vec<Probed> = Vec::with_capacity(ev.pieces.len());
+    for piece in &ev.pieces {
+        let mut p = Probed::new(piece.clone());
+        let (units, unit, faults) =
+            sampler.first_encrypted(&p.piece, &segments, ev.container, ev.detector, run.halt)?;
+        p.units = units;
+        p.faults = faults;
+        match unit {
+            Some(u) => p.enc.push(u),
+            // Faults with no ciphertext found: not trusted, the probed path decides.
+            None if faults > 0 => return Ok(None),
+            None => p.verdict = Verdict::Clear,
+        }
+        ps.push(p);
     }
-    Ok(Scoped { ps, layout, clip })
+    let opened = |run: &Run, p: &Probed| -> Option<usize> {
+        p.enc
+            .first()
+            .and_then(|u| (0..run.pool.len()).find(|&s| run.opens(u, s)))
+    };
+    let unopened = |run: &Run, ps: &[Probed]| -> Vec<usize> {
+        (0..ps.len())
+            .filter(|&i| ps[i].verdict == Verdict::Ask && opened(run, &ps[i]).is_none())
+            .collect()
+    };
+    let mut tried: Vec<usize> = Vec::new();
+    // Sample-independent sources (a local keydb) give their whole answer once.
+    if !unopened(run, &ps).is_empty() {
+        let main = sampler.main_samples(ev.main.as_ref(), MIN_SAMPLE_UNITS);
+        for i in 0..run.sources.len() {
+            if run.sources[i].answer_depends_on_samples() {
+                continue;
+            }
+            run.check_halt()?;
+            let asked = run.ask(i, &main, false)?;
+            if !matches!(asked, Asked::Skipped) {
+                tried.push(i);
+            }
+            if let Asked::Keys(k, _) = asked {
+                let who = run.sources[i].label();
+                run.add_keys(&k, who);
+            }
+        }
+    }
+    // Then, largest piece first, each still-unopened unit asks the sample-dependent sources
+    // with enough of its own ciphertext (topped up from pieces of the same title, KS-10), at
+    // most `n_decl` requests. One answer may open many pieces. A source that answers with no
+    // key has none for this disc: it is not asked again.
+    let mut budget = n_decl;
+    let mut thin: Vec<usize> = Vec::new();
+    let mut gave_up: Vec<usize> = Vec::new();
+    loop {
+        let mut todo: Vec<usize> = unopened(run, &ps)
+            .into_iter()
+            .filter(|i| !thin.contains(i) && !gave_up.contains(i))
+            .collect();
+        let live: Vec<usize> = (0..run.sources.len())
+            .filter(|&i| run.sources[i].answer_depends_on_samples() && !run.dead[i])
+            .collect();
+        if todo.is_empty() || budget == 0 || live.is_empty() {
+            break;
+        }
+        todo.sort_by_key(|&i| (std::cmp::Reverse(ps[i].piece.rank), ps[i].id()));
+        let lead = todo[0];
+        let mut samples = sampler.encrypted_units(
+            &ps[lead].piece,
+            &segments,
+            ev.container,
+            ev.detector,
+            run.halt,
+            MIN_SAMPLE_UNITS,
+        )?;
+        for &j in &todo[1..] {
+            if samples.len() >= MIN_SAMPLE_UNITS {
+                break;
+            }
+            if ps[j]
+                .piece
+                .titles
+                .iter()
+                .any(|t| ps[lead].piece.titles.contains(t))
+            {
+                let more = sampler.encrypted_units(
+                    &ps[j].piece,
+                    &segments,
+                    ev.container,
+                    ev.detector,
+                    run.halt,
+                    MIN_SAMPLE_UNITS - samples.len(),
+                )?;
+                samples.extend(more);
+            }
+        }
+        if samples.len() < MIN_SAMPLE_UNITS {
+            // Too little ciphertext to ask with (KU step 9.2): proven on arrival instead.
+            thin.push(lead);
+            continue;
+        }
+        budget -= 1;
+        for &i in &live {
+            run.check_halt()?;
+            let asked = run.ask(i, &samples, false)?;
+            if !matches!(asked, Asked::Skipped) && !tried.contains(&i) {
+                tried.push(i);
+            }
+            match asked {
+                Asked::Keys(k, _) => {
+                    let who = run.sources[i].label();
+                    run.add_keys(&k, who);
+                    if opened(run, &ps[lead]).is_some() {
+                        break;
+                    }
+                }
+                Asked::Empty => run.dead[i] = true,
+                Asked::Failed | Asked::Skipped => {}
+            }
+        }
+        if opened(run, &ps[lead]).is_none() {
+            gave_up.push(lead);
+        }
+    }
+    // Every unit's sample was sent: a piece no held key opens has no key to find.
+    let whole_with_keys =
+        ev.scope == KeyScope::WholeDisc && !run.pool.is_empty() && run.first_failure.is_none();
+    for (idx, p) in ps.iter_mut().enumerate() {
+        if p.verdict == Verdict::Ask {
+            match opened(run, p) {
+                Some(slot) => p.verdict = Verdict::Keyed(slot),
+                None if thin.contains(&idx) => p.verdict = Verdict::Lazy(None),
+                // Not asked at all (too little ciphertext): the probed path decides.
+                None if tried.is_empty() => {
+                    for &i in &tried {
+                        run.dead[i] = true;
+                    }
+                    return Ok(None);
+                }
+                // A whole-disc copy blanks it while other pieces are usable (as the probed path).
+                None if whole_with_keys => {
+                    tracing::warn!(target: "freemkv::keys", lba = p.id(), "no held key opens this stream file: it will be blanked in the image");
+                    p.verdict = Verdict::Lazy(None);
+                }
+                None => {
+                    let lba = p.id();
+                    let failure = run.first_failure.take();
+                    let err = failure.unwrap_or_else(|| missing_error(&ev.scope, disc_hash));
+                    tracing::error!(target: "freemkv::keys", lba, code = err.code(), "a stream file in scope has no key; refusing before any output");
+                    return Err(err);
+                }
+            }
+        }
+    }
+    // KU §2.7: an empty pool on an encrypted scope is the keyless case.
+    if run.pool.is_empty() && ps.iter().any(|p| matches!(p.verdict, Verdict::Lazy(_))) {
+        let failure = run.first_failure.take();
+        return Err(failure.unwrap_or_else(|| missing_error(&ev.scope, disc_hash)));
+    }
+    if whole_with_keys && !ps.iter().any(|p| matches!(p.verdict, Verdict::Keyed(_))) {
+        let failure = run.first_failure.take();
+        return Err(failure.unwrap_or_else(|| missing_error(&ev.scope, disc_hash)));
+    }
+    tracing::info!(
+        target: "freemkv::keys",
+        phase = "per_cps_unit",
+        declared = n_decl,
+        pieces = ps.len(),
+        "each piece keyed by the key that opens its one sampled unit"
+    );
+    let mut inner = aacs_inner(run);
+    inner.no_stream_files = ev.no_stream_files;
+    let forensic = match layout {
+        None => None,
+        Some(l) => Some(resolve_forensic(sampler, l, seed, run, ev.container)?),
+    };
+    build(&mut inner, &ps, layout, forensic, run, clip);
+    Ok(Some(inner))
 }
 
 // The one encryption decision, every AACS format and source (CSS makes it on its scramble
@@ -520,6 +840,17 @@ fn decide(
     seed: Option<&KeyRing>,
     run: &mut Run,
 ) -> Result<Inner> {
+    // A forensic layout in scope keeps the probed path: its segments follow their own rules.
+    if ev.detector == Detector::Verified && scope_layout(ev).0.is_none() {
+        let trusted = match n_decl {
+            Some(1) => resolve_one_unit(ev, sampler, disc_hash, seed, run)?,
+            Some(n) if n > 1 => resolve_per_unit(ev, sampler, disc_hash, seed, run, n)?,
+            _ => None,
+        };
+        if let Some(inner) = trusted {
+            return Ok(inner);
+        }
+    }
     let scoped = probe_scope(ev, sampler, run.halt)?;
     if in_clear(&scoped, sampler, ev.container, run.halt)? {
         tracing::info!(target: "freemkv::keys", pieces = scoped.ps.len(), "content in the clear: no key source asked");

src/keys/tests.rs

@@ -580,11 +580,11 @@ fn single_declared_unit_wrong_key_refuses_e7013() {
     assert_eq!(code(r), E7013);
 }
 
-/// LK4 (J22). KS-14: with one declared CPS unit, a key proven on one piece keys a Lazy
-/// piece only when it opens every read probe of it. A piece with no readable probe stays
-/// Lazy with that key as its candidate ("never guessed") and is proven on first read.
+/// LK4. KS-14 [BD] §3.9.3: "Num_of_CPS_Unit field (16 bits) indicates the number of CPS Units
+/// on the disc". One declared unit is trusted: the key that opens the main title keys every
+/// piece, including one with no readable unit at resolve, which reads under it once healed.
 #[test]
-fn single_declared_unit_keys_lazy_pieces() {
+fn single_declared_unit_keys_every_piece() {
     let fx = fixture(
         &[stream(1, 10, Some(K1)), stream(2, 10, Some(K1))],
         1,
@@ -603,20 +603,14 @@ fn single_declared_unit_keys_lazy_pieces() {
         &FakeClock::default(),
     )
     .unwrap();
-    assert_eq!(
-        set.lazy(),
-        &[(b, b + n)],
-        "no probe of B read: Lazy, never keyed blind"
+    assert!(
+        set.lazy().is_empty(),
+        "the declared unit keys B without reading it"
     );
-    assert_eq!(set.status().keyed, 1);
+    assert_eq!(set.status().keyed, 2);
     src.heal();
     let mut r = set.title_reader(&fx.disc, 0, src).unwrap();
     assert_eq!(read(&mut r, &fx, 1, 0, 10).unwrap(), fx.plain(b, 10));
-    assert_eq!(
-        set.proof_cache().get(b),
-        Some(Proof::Proven(0)),
-        "K1 first, proven on read"
-    );
 }
 
 /// LK5 (K-1). KS-14, KS-10: the count of held keys never decides coverage. Two declared,
@@ -782,7 +776,11 @@ fn small_piece_borrows_or_goes_lazy_never_missing() {
     )
     .unwrap();
     assert_eq!(calls.of("online").len(), 1, "S and T ask once, together");
-    assert_eq!(calls.of("online")[0].samples, 9);
+    assert_eq!(
+        calls.of("online")[0].samples,
+        crate::keysource::MIN_SAMPLE_UNITS,
+        "exactly the minimum: no unit read beyond it"
+    );
     let (u, n) = fx.file(3);
     assert_eq!(
         set.lazy(),
@@ -818,10 +816,10 @@ fn playlist_pieces_share_a_cps_unit_for_borrowing() {
     assert_eq!(code(set), E7022);
 }
 
-/// J21 (OQ-L2-1) guard. A piece where exactly one probe read and opened is Lazy with that
-/// key as its first candidate — never keyed from one probe — and is proven on read.
+/// KS-10 [BD] §3.9.2: a stream file sits in one CPS unit, so with several declared units one
+/// readable unit of a piece names its key: the piece is keyed by it and reads.
 #[test]
-fn one_opened_probe_is_lazy_never_keyed() {
+fn one_sampled_unit_keys_its_piece_on_a_multi_unit_disc() {
     let fx = fixture(
         &[stream(1, 10, Some(K1)), stream(2, 10, Some(K2))],
         2,
@@ -840,15 +838,15 @@ fn one_opened_probe_is_lazy_never_keyed() {
         &FakeClock::default(),
     )
     .unwrap();
-    assert_eq!(set.lazy(), &[(b, b + n)]);
-    assert_eq!(set.status().keyed, 1);
+    assert!(set.lazy().is_empty());
+    assert_eq!(set.status().keyed, 2);
     src.heal();
     let mut r = set.title_reader(&fx.disc, 0, src).unwrap();
     assert_eq!(read(&mut r, &fx, 1, 0, 10).unwrap(), fx.plain(b, 10));
 }
 
-/// LK22. A sample-independent source (a keydb) is asked once per resolve; an online source
-/// once per unopened piece (N-KU10).
+/// LK22. A sample-independent source (a keydb) is asked once per resolve, and an online
+/// source once with one unit of every unopened piece: neither is asked again on a refusal.
 #[test]
 fn sample_independent_source_asked_once() {
     let files: Vec<BdFile> = (1..=5).map(|i| stream(i, 10, Some(K2))).collect();
@@ -861,7 +859,7 @@ fn sample_independent_source_asked_once() {
     );
     assert_eq!(code(r), E7032);
     assert_eq!(calls.of("keydb").len(), 1);
-    assert_eq!(calls.of("online").len(), 5);
+    assert_eq!(calls.of("online").len(), 1);
 }
 
 /// LK23a (J13, J15). A source that gives no answer (transport class) for 20 s of simulated
@@ -1252,7 +1250,8 @@ fn stop_during_resolve_builds_no_set() {
     let fx = two_units();
     let halt = Halt::new();
     let mut src = fx.source();
-    src.halt_after = Some((halt.clone(), Arc::new(Mutex::new(40))));
+    // Stopped on the first read: before any source is asked.
+    src.halt_after = Some((halt.clone(), Arc::new(Mutex::new(1))));
     let calls = Calls::default();
     let opts = AcquireOptions {
         ..Default::default()
@@ -2359,33 +2358,26 @@ fn whole_disc_status_counts_keyed_files_without_titles() {
     assert_eq!((s.keyed, s.proven, s.lazy), (1, 1, 0), "{s:?}");
 }
 
-/// KU §2.3 step 10 (review item 4). KS-14 [BD] §3.9.3: "Num_of_CPS_Unit field (16 bits)
-/// indicates the number of CPS Units on the disc". The n_decl == 1 rule keys only Lazy
-/// pieces no other held key opened: a piece whose one readable probe another key opened
-/// stays Lazy with that candidate, never keyed with the proven key over it.
+/// KS-14 [BD] §3.9.3. One declared unit: the key that opens the main title is the disc's key,
+/// so every piece is keyed by it and asked for once, whatever other keys a source returns.
 #[test]
-fn single_unit_rule_never_overrides_a_piece_another_key_opened() {
+fn single_declared_unit_keys_every_piece_with_the_main_titles_key() {
     let fx = fixture(
-        &[stream(1, 10, Some(K1)), stream(2, 10, Some(K2))],
+        &[stream(1, 10, Some(K1)), stream(2, 10, Some(K1))],
         1,
         &[&[0, 1]],
     );
-    let (b, n) = fx.file(1);
-    let src = fx.source();
-    src.kill(b + 3, b + n); // only B's first unit reads, and K2 opens it
+    let (b, _) = fx.file(1);
     let calls = Calls::default();
-    let set = resolve_with(
+    let set = resolve(
         &fx,
-        &mut src.clone(),
         KeyScope::Titles(vec![0]),
-        &[Spec::keydb(&[K1, K2], &calls)],
-        AcquireOptions::default(),
-        &FakeClock::default(),
+        &[Spec::keydb(&[K2, K1], &calls)],
     )
     .unwrap();
-    assert_eq!(set.lazy(), &[(b, b + n)], "B stays Lazy (candidate K2)");
-    src.heal();
-    let mut r = set.title_reader(&fx.disc, 0, src).unwrap();
+    assert_eq!((set.status().keyed, set.lazy()), (2, &[][..]));
+    assert_eq!(calls.of("keydb").len(), 1);
+    let mut r = set.title_reader(&fx.disc, 0, fx.source()).unwrap();
     assert_eq!(read(&mut r, &fx, 1, 0, 10).unwrap(), fx.plain(b, 10));
 }
 
@@ -2404,13 +2396,11 @@ fn unopenable_unit_stops_after_one_side_read() {
     assert_eq!(reads.len(), 2, "then one side read");
 }
 
-/// KU §2.3 step 10 under invariant 2, "never guessed" (J22, review r2 M1). KS-14 [BD]
-/// §3.9.3: "Num_of_CPS_Unit field (16 bits) indicates the number of CPS Units on the disc".
-/// With one declared unit, a Lazy piece is keyed by the proven key only if that key opens
-/// every `Enc` probe of it: (a) probes no held key opens, (b) probes opened by two keys.
+/// KS-14 [BD] §3.9.3: "Num_of_CPS_Unit field (16 bits) indicates the number of CPS Units on
+/// the disc". The declaration is trusted: with one unit no piece is probed, so even a piece
+/// the disc puts under another key is keyed with the main title's key, the source asked once.
 #[test]
-fn single_unit_rule_keys_only_pieces_the_proven_key_opens() {
-    // (a) B's three units are under K2, which no source holds.
+fn single_declared_unit_is_trusted_without_probing_each_piece() {
     let fx = fixture(
         &[stream(1, 10, Some(K1)), stream(2, 3, Some(K2))],
         1,
@@ -2423,51 +2413,8 @@ fn single_unit_rule_keys_only_pieces_the_proven_key_opens() {
         &[Spec::keydb(&[K1], &calls)],
     )
     .unwrap();
-    let (b, n) = fx.file(1);
-    assert_eq!(
-        (set.status().keyed, set.lazy()),
-        (1, &[(b, b + n)][..]),
-        "(a)"
-    );
-    let mut r = set.title_reader(&fx.disc, 0, fx.source()).unwrap();
-    assert_eq!(
-        code(read(&mut r, &fx, 1, 0, 3)),
-        E7022,
-        "(a) never K1 over K2's ciphertext"
-    );
-
-    // (b) B's only readable probes: unit 0 opens under K1, unit 5 under K2.
-    let mut fx = fixture(
-        &[stream(1, 10, Some(K1)), stream(2, 10, Some(K2))],
-        1,
-        &[&[0, 1]],
-    );
-    fx.reencrypt(1, 0, &K1);
-    let (b, n) = fx.file(1);
-    let src = fx.source();
-    src.kill(fx.unit(1, 1), fx.unit(1, 5));
-    src.kill(fx.unit(1, 6), b + n);
-    let set = resolve_with(
-        &fx,
-        &mut src.clone(),
-        KeyScope::Titles(vec![0]),
-        &[Spec::keydb(&[K1, K2], &calls)],
-        AcquireOptions::default(),
-        &FakeClock::default(),
-    )
-    .unwrap();
-    assert_eq!(
-        (set.status().keyed, set.lazy()),
-        (1, &[(b, b + n)][..]),
-        "(b)"
-    );
-    src.heal();
-    let mut r = set.title_reader(&fx.disc, 0, src).unwrap();
-    assert_eq!(
-        code(read(&mut r, &fx, 1, 0, 10)),
-        E7022,
-        "(b) two keys: a conflict"
-    );
+    assert_eq!((set.status().keyed, set.lazy()), (2, &[][..]));
+    assert_eq!(calls.of("keydb").len(), 1);
 }
 
 /// KU §3.1 (review r2 M3): a keyless set over a title whose extents overlap (one inside
@@ -2529,14 +2476,12 @@ fn keyless_set_over_overlapping_extents_stops_e7022() {
     }
 }
 
-/// J22 (review r3). KS-14 [BD] §3.9.3: "Num_of_CPS_Unit field (16 bits) indicates the number
-/// of CPS Units on the disc". A piece whose every probe faulted has no evidence: with one
-/// declared unit it stays Lazy (the proven key its candidate), so a readable unit under a
-/// key no source holds stops with E7022, never decrypted under the proven key.
+/// KS-14 [BD] §3.9.3. A piece with no readable unit at resolve is keyed by the one declared
+/// unit's key without being read, and reads under it once healed.
 #[test]
-fn single_unit_rule_never_keys_a_piece_with_no_readable_probe() {
+fn single_declared_unit_keys_an_unreadable_piece_without_reading_it() {
     let fx = fixture(
-        &[stream(1, 10, Some(K1)), stream(2, 3, Some(K2))],
+        &[stream(1, 10, Some(K1)), stream(2, 3, Some(K1))],
         1,
         &[&[0, 1]],
     );
@@ -2553,14 +2498,10 @@ fn single_unit_rule_never_keys_a_piece_with_no_readable_probe() {
         &FakeClock::default(),
     )
     .unwrap();
-    assert_eq!((set.status().keyed, set.lazy()), (1, &[(b, b + n)][..]));
+    assert_eq!((set.status().keyed, set.lazy()), (2, &[][..]));
     src.heal();
     let mut r = set.title_reader(&fx.disc, 0, src).unwrap();
-    assert_eq!(
-        code(read(&mut r, &fx, 1, 0, 3)),
-        E7022,
-        "never K1 over K2's ciphertext"
-    );
+    assert_eq!(read(&mut r, &fx, 1, 0, 3).unwrap(), fx.plain(b, 3));
 }
 
 /// KS-1 [BD] §3.10.1: "encryption is applied to every Aligned Unit in the file" — each clip
@@ -2609,12 +2550,11 @@ fn keyless_session_set_fits_a_disc_without_a_captured_hash() {
     assert!(set.is_for(&fx.disc.media_id()));
 }
 
-/// J22 with J21 parity (review r4). KS-14 [BD] §3.9.3: "Num_of_CPS_Unit field (16 bits)
-/// indicates the number of CPS Units on the disc". Step 10 keys on no less evidence than step
-/// 9: one opened probe of a piece longer than one unit is not proof, so the piece stays Lazy
-/// with the proven key as candidate; a one-unit piece's one opened probe keys it.
+/// KS-14 [BD] §3.9.3: one declared unit keys every piece from the main title's samples: a
+/// piece with one readable unit of ten, and a one-unit piece, are keyed alike, and the only
+/// source is asked once.
 #[test]
-fn single_unit_rule_needs_two_probes_unless_one_unit_long() {
+fn single_declared_unit_needs_no_probe_of_each_piece() {
     let fx = fixture(
         &[
             stream(1, 10, Some(K1)),
@@ -2626,7 +2566,7 @@ fn single_unit_rule_needs_two_probes_unless_one_unit_long() {
     );
     let (b, n) = fx.file(1);
     let src = fx.source();
-    src.kill(b + 3, b + n); // one of B's ten probes reads
+    src.kill(b + 3, b + n);
     let calls = Calls::default();
     let set = resolve_with(
         &fx,
@@ -2637,16 +2577,12 @@ fn single_unit_rule_needs_two_probes_unless_one_unit_long() {
         &FakeClock::default(),
     )
     .unwrap();
-    assert_eq!(
-        set.lazy(),
-        &[(b, b + n)],
-        "one probe of ten: Lazy, not keyed"
-    );
-    assert_eq!(set.status().keyed, 2, "A and the one-unit C");
+    assert!(set.lazy().is_empty());
+    assert_eq!(set.status().keyed, 3);
+    assert_eq!(calls.of("keydb").len(), 1);
     src.heal();
     let mut r = set.title_reader(&fx.disc, 0, src).unwrap();
     assert_eq!(read(&mut r, &fx, 1, 0, 10).unwrap(), fx.plain(b, 10));
-    assert_eq!(set.proof_cache().get(b), Some(Proof::Proven(0)));
 }
 
 // ── J23: would the disc's VID help? ────────────────────────────────────────

tests/goldens/parity_image_bd_missing_key.golden

@@ -10,18 +10,11 @@ copy-raw image = 8466432 bytes sha256:f0856c74bf5c9f7b6822186cafcaf1d6
 extract-raw result = good=73728 unreadable=0 complete=true halted=false files=[("BDMV/STREAM/00001.m2ts", 36864, 0, true), ("BDMV/STREAM/00002.m2ts", 36864, 0, true)]
 extract-raw out BDMV/STREAM/00001.m2ts = 36864 bytes sha256:375559f987f4f85e585d39abe6eb12fa
 extract-raw out BDMV/STREAM/00002.m2ts = 36864 bytes sha256:2d895155c02331f792c4fca0bfb881ea
-resolve-title status = KeySetStatus { proven: 1, declared: Some(2), keyed: 1, clear: 0, lazy: 1, forensic: None, origin: Some("keydb"), best_effort: false, requests: 1 }
-title-read = 36864 bytes sha256:368a55d7eca3e25f6af377a6f1aec041
-title-read blanked = 0
-title-read failed = E7022
+resolve-title = refused E7022
 iso-mkv refused = E7022
-iso-mkv out o.mkv = 21863 bytes sha256:47337f351b04ebb679c0da200a04e94a
 iso-m2ts refused = E7022
-iso-m2ts out o.m2ts = 31488 bytes sha256:3a40b2a87a7e62135df01ead62597601
 iso-null refused = E7022
 iso-mkv-raw refused = E6008
 iso-mkv-selection refused = E7022
-iso-mkv-selection out o.mkv = 15124 bytes sha256:120947d4cc4150b1531e918c11865049
 live-mkv refused = E7022
-live-mkv out o.mkv = 21863 bytes sha256:47337f351b04ebb679c0da200a04e94a
 live-mkv-raw refused = E6008