sector: an unreadable AACS unit head is a guess, not a cached CPI; docs and batch/threshold dedupe; tests for blank reporting, byte-view clip, producer latch, BD-TS gates, forwarding

src/sector/bus_removal.rs

@@ -330,12 +330,18 @@ impl BusGate {
                     Some(h) => head_byte0(h),
                     None => None,
                 };
-                // An unreadable or unmapped head is cached as encrypted: never re-read.
+                // An unreadable head is a guess, not a measurement: it is not
+                // remembered, so a later read that has the head decides for real.
+                // An unmapped head can never be measured, so that guess is stable.
                 let e = byte0.is_none_or(|b| b & CPI_MASK != 0);
-                if self.cpi.len() >= CPI_CACHE_MAX {
-                    self.cpi.clear();
+                if byte0.is_some() || head.is_none() {
+                    if self.cpi.len() >= CPI_CACHE_MAX {
+                        self.cpi.clear();
+                    }
+                    self.cpi.insert(key, e);
+                } else {
+                    return e;
                 }
-                self.cpi.insert(key, e);
                 e
             }
         };
@@ -658,6 +664,39 @@ mod tests {
         assert_eq!(s.inner().fua, [true, false]);
     }
 
+    // The wrapper is transparent to the inner source's capabilities and speed control.
+    #[test]
+    fn bus_removal_forwards_random_access_and_set_speed() {
+        struct Seq {
+            speed: u16,
+        }
+        impl SectorSource for Seq {
+            fn capacity_sectors(&self) -> u32 {
+                0
+            }
+            fn read_sectors(&mut self, _: u32, _: u16, _: &mut [u8], _: bool) -> Result<usize> {
+                Ok(0)
+            }
+            fn random_access(&self) -> bool {
+                false
+            }
+            fn set_speed(&mut self, kbs: u16) {
+                self.speed = kbs;
+            }
+        }
+        let mut s = BusRemovalSectorSource::new(Seq { speed: 0 }, BusStage::Passthrough);
+        assert!(!s.random_access());
+        s.set_speed(4500);
+        assert_eq!(s.inner().speed, 4500);
+    }
+
+    // An extent running past the 32-bit LBA space is clipped at it, never wrapped.
+    #[test]
+    fn bus_map_extent_past_the_lba_space_is_clipped() {
+        let map = BusMap::from_files(vec![vec![(0xFFFF_FFF0, 100)]]);
+        assert_eq!(map.covered_ranges(), vec![(0xFFFF_FFF0, 16)]);
+    }
+
     // A partly overlapping second file keeps its file offset across the clip.
     #[test]
     fn bus_map_partial_overlap_clip_keeps_the_file_offset() {
@@ -987,10 +1026,10 @@ mod tests {
         assert_eq!(buf, clear, "unknown head CPI must de-bus");
     }
 
-    // A unit head that fails to read is fetched once, fast-path (recovery=false),
-    // and cached as encrypted: neighbours never re-read it.
+    // A unit head that fails to read is fetched fast-path (recovery=false) and the
+    // unit guessed encrypted; the guess is not remembered, so each read retries.
     #[test]
-    fn host_key_failed_head_fetch_is_cached_as_encrypted() {
+    fn host_key_failed_head_fetch_guesses_encrypted_without_caching() {
         struct BadHead {
             bytes: Vec<u8>,
             reads: Vec<(u32, bool)>,
@@ -1036,11 +1075,91 @@ mod tests {
         assert_eq!(got, clear);
         assert_eq!(
             s.inner().reads,
-            vec![(301, true), (300, false), (302, true)],
-            "one fast head fetch, then cached"
+            vec![(301, true), (300, false), (302, true), (300, false)],
+            "the unreadable head is retried, never cached"
         );
     }
 
+    // A head that was unreadable once and is clear when later read: the guess made
+    // meanwhile must not decide the whole unit.
+    #[test]
+    fn host_key_head_guess_is_replaced_by_the_measured_head() {
+        let rdk = [0x62u8; 16];
+        let mut clear = clear_content(3); // LBAs 300..303
+        for l in 0..3 {
+            clear[l * SECTOR_BYTES] &= !0xC0;
+        }
+        let mut wire = clear.clone();
+        encrypt_bus(&mut wire, &rdk);
+        let map = Arc::new(BusMap::from_files(vec![vec![(300, 3)]]));
+        let mut g = BusGate::new(map);
+        // First pass: sector 301 alone, head 300 unreadable -> guessed encrypted.
+        let mut b = sector(&wire, 300, 301).to_vec();
+        g.debus(&mut b, &rdk, 301, &mut |_| None);
+        // Second pass: the whole unit with its real (clear) head.
+        let mut all = clear.clone();
+        g.debus(&mut all, &rdk, 300, &mut |_| {
+            unreachable!("head is in the buffer")
+        });
+        assert_eq!(all, clear, "a clear unit must pass through untouched");
+    }
+
+    // Any copy-permission bit set means encrypted (CPI 01/10/11), not just the top bit.
+    #[test]
+    fn host_key_any_cpi_bit_marks_the_unit_encrypted() {
+        let rdk = [0x63u8; 16];
+        for (head, encrypted) in [(0x00u8, false), (0x40, true), (0x80, true), (0xC0, true)] {
+            let mut clear = clear_content(3);
+            clear[0] = (clear[0] & !0xC0) | head;
+            let mut wire = clear.clone();
+            if encrypted {
+                encrypt_bus(&mut wire, &rdk);
+            }
+            let mut g = BusGate::new(Arc::new(BusMap::from_files(vec![vec![(300, 3)]])));
+            g.debus(&mut wire, &rdk, 300, &mut |_| None);
+            assert_eq!(wire, clear, "head byte0 {head:#x}");
+        }
+    }
+
+    // The decision is kept per (file, unit): interleaved reads of two files' mid-unit
+    // sectors fetch each head once, and unit 0 of one file does not decide the other's.
+    #[test]
+    fn host_key_decisions_are_keyed_by_file_and_unit() {
+        let rdk = [0x64u8; 16];
+        let mut clear = clear_content(6); // file 0 = 300..303 (CPI set), file 1 = 303..306 (clear)
+        clear[3 * SECTOR_BYTES] &= !0xC0;
+        let mut wire = clear.clone();
+        encrypt_bus(&mut wire[..3 * SECTOR_BYTES], &rdk);
+        let map = Arc::new(BusMap::from_files(vec![vec![(300, 3)], vec![(303, 3)]]));
+        let mut g = BusGate::new(map);
+        let mut heads = Vec::new();
+        for lba in [301u32, 304, 302, 305] {
+            let mut b = sector(&wire, 300, lba).to_vec();
+            g.debus(&mut b, &rdk, lba, &mut |h| {
+                heads.push(h);
+                Some(sector(&clear, 300, h)[0])
+            });
+            assert_eq!(b, sector(&clear, 300, lba), "lba {lba}");
+        }
+        assert_eq!(heads, vec![300, 303], "each head fetched once");
+    }
+
+    // The per-unit cache is bounded: it is cleared rather than growing with the disc.
+    #[test]
+    fn host_key_cpi_cache_stays_bounded() {
+        let units = CPI_CACHE_MAX as u32 + 8;
+        let map = Arc::new(BusMap::from_files(vec![vec![(0, units * 3)]]));
+        let mut g = BusGate::new(map);
+        let rdk = [0x65u8; 16];
+        let mut buf = clear_content(1);
+        buf[0] &= !0xC0;
+        for u in 0..units {
+            g.debus(&mut buf, &rdk, u * 3 + 1, &mut |_| Some(0));
+            assert!(g.cpi.len() <= CPI_CACHE_MAX);
+        }
+        assert!(g.cpi.len() < units as usize);
+    }
+
     // Unrecorded extents keep file offsets: [500,1) + hole(1) + [700,4) puts
     // 700 in unit 0 (head 500) and 701 at the head of unit 1.
     #[test]

src/sector/decrypting.rs

@@ -291,8 +291,6 @@ impl<S: SectorSource> DecryptingSectorSource<S> {
     }
 
     // Decrypt `buf` with the active keys, gated by content ranges when set.
-    // Shared by the first read and post-fetch retry so both agree on which
-    // units are content and the unit-key try order.
     fn decrypt_buf(
         buf: &mut [u8],
         keys: &mut DecryptKeys,
@@ -1396,6 +1394,32 @@ mod tests {
         assert_eq!(dec.blanked_units(), 1);
     }
 
+    /// A blanked unit reaches the run: its loss counter and one `UnitBlanked` at the read's
+    /// LBA; a read with no damage reports nothing.
+    #[test]
+    fn blanked_units_are_reported_to_the_run() {
+        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
+        let sink = seen.clone();
+        let ctx =
+            crate::ctx::Ctx::default().with_events(Arc::new(move |e: &crate::event::Event<'_>| {
+                if let crate::event::Event::UnitBlanked { lba, units } = e {
+                    sink.lock().unwrap().push((*lba, *units));
+                }
+            }));
+        let mut dec = damaged_file_source(4, &[1, 2]);
+        dec.observe(&ctx);
+        read_units(&mut dec, 0, 1).expect("intact unit");
+        assert!(seen.lock().unwrap().is_empty(), "no damage, no event");
+        assert_eq!(ctx.stats.snapshot().units_blanked, 0);
+        read_units(&mut dec, 1, 2).expect("a damage cluster is blanked");
+        assert_eq!(
+            *seen.lock().unwrap(),
+            vec![(u64::from(FILE_LBA) + 3, 2)],
+            "one event: the read's LBA and the unit count"
+        );
+        assert_eq!(ctx.stats.snapshot().units_blanked, 2);
+    }
+
     /// A cluster of damaged units is blanked and counted however it is read: alone, first,
     /// with no intact unit around it. KS-2 [BD] §3.10.1: "Each MPEG source packet consists of
     /// the TP_extra_header (4 bytes) and an MPEG Transport packet".

src/sector/mod.rs

@@ -36,7 +36,8 @@ pub trait SectorSource: Send {
     ///
     /// # Panics
     ///
-    /// Implementations may panic if `buf` is undersized.
+    /// Implementations may panic if `buf` is undersized; `FileSectorSource`
+    /// returns `DiscRead` instead.
     fn read_sectors(
         &mut self,
         lba: u32,
@@ -44,7 +45,6 @@ pub trait SectorSource: Send {
         buf: &mut [u8],
         recovery: bool,
     ) -> Result<usize>;
-    // `FileSectorSource` checks this and returns `DiscRead` instead of panicking.
 
     /// Like [`read_sectors`], but with an explicit Force Unit Access request.
     ///
@@ -267,6 +267,50 @@ mod tests {
         s.set_unit_base(base);
     }
 
+    // FUA and random-access answers must cross the forwarding impls, or Pass-N recovery
+    // reads lose Force Unit Access and a prefetcher claims random access.
+    struct FuaSpy(Arc<Mutex<Vec<bool>>>);
+    impl SectorSource for FuaSpy {
+        fn capacity_sectors(&self) -> u32 {
+            0
+        }
+        fn read_sectors(&mut self, _: u32, _: u16, _: &mut [u8], _: bool) -> Result<usize> {
+            self.0.lock().unwrap().push(false);
+            Ok(0)
+        }
+        fn read_sectors_fua(
+            &mut self,
+            _: u32,
+            _: u16,
+            _: &mut [u8],
+            _: bool,
+            fua: bool,
+        ) -> Result<usize> {
+            self.0.lock().unwrap().push(fua);
+            Ok(0)
+        }
+        fn random_access(&self) -> bool {
+            false
+        }
+    }
+
+    fn fua_through<S: SectorSource>(mut s: S) -> bool {
+        s.read_sectors_fua(0, 1, &mut [0u8; 2048], false, true)
+            .unwrap();
+        s.random_access()
+    }
+
+    #[test]
+    fn forwarding_impls_keep_fua_and_random_access() {
+        let seen = Arc::new(Mutex::new(Vec::new()));
+        let boxed: Box<dyn SectorSource> = Box::new(FuaSpy(seen.clone()));
+        assert!(!fua_through(boxed));
+        let mut spy = FuaSpy(seen.clone());
+        let by_ref: &mut dyn SectorSource = &mut spy;
+        assert!(!fua_through(by_ref));
+        assert_eq!(*seen.lock().unwrap(), vec![true, true]);
+    }
+
     // Leaf sources with no bus stage: image bytes on file are already bus-clear.
     const UNMAPPED_LEAVES: &[&str] = &["FileSectorSource", "DirImage"];
 

src/sector/prefetched.rs

@@ -177,10 +177,7 @@ impl PrefetchedSectorSource {
             crate::sector::read_stage::ExtentWalk::new(extents, policy, unit_align, ctx)?;
         let loss = walk.loss();
         let total_sectors = walk.total_sectors();
-        let batch_sectors = match policy {
-            crate::sector::read_stage::ReadPolicy::Image { batch }
-            | crate::sector::read_stage::ReadPolicy::Live { batch, .. } => batch,
-        };
+        let batch_sectors = policy.batch();
         let unmapped = reader.unmapped_stream_files().to_vec();
         let (tx, rx) = bounded::<Batch>(PREFETCH_CHANNEL_DEPTH);
         let (recycle_tx, recycle_rx) = bounded::<Vec<u8>>(PREFETCH_CHANNEL_DEPTH + 1);
@@ -1349,6 +1346,86 @@ mod tests {
         });
     }
 
+    // After the producer reports an error, the closed channel that follows is a dead source,
+    // never a clean end of stream.
+    #[test]
+    fn a_producer_error_latches_the_source_as_terminated() {
+        let _serial = serial();
+        with_watchdog(Duration::from_secs(10), || {
+            let extents = vec![Extent {
+                start_lba: 0,
+                sector_count: 3,
+            }];
+            let mut pf = PrefetchedSectorSource::new(ErrorSource, extents, 3, &Ctx::default())
+                .expect("spawn");
+            let mut buf = vec![0u8; 3 * 2048];
+            pf.read_sectors(0, 3, &mut buf, false)
+                .expect_err("the reader's error");
+            let again = pf.read_sectors(0, 3, &mut buf, false);
+            assert!(
+                matches!(&again, Err(e) if e.is_source_terminated()),
+                "second read after a producer error: {again:?}"
+            );
+        });
+    }
+
+    // A reader that panics is an error to the consumer, not a dropped channel read as EOF.
+    #[test]
+    fn a_panicking_reader_is_an_error_not_eof() {
+        struct Panics;
+        impl SectorSource for Panics {
+            fn read_sectors(&mut self, _: u32, _: u16, _: &mut [u8], _: bool) -> Result<usize> {
+                panic!("reader panic (expected by the test)");
+            }
+            fn capacity_sectors(&self) -> u32 {
+                3
+            }
+        }
+        let _serial = serial();
+        with_watchdog(Duration::from_secs(10), || {
+            let extents = vec![Extent {
+                start_lba: 0,
+                sector_count: 3,
+            }];
+            let mut pf =
+                PrefetchedSectorSource::new(Panics, extents, 3, &Ctx::default()).expect("spawn");
+            let mut buf = vec![0u8; 3 * 2048];
+            let r = pf.read_sectors(0, 3, &mut buf, false);
+            assert!(
+                matches!(r, Err(crate::error::Error::DemuxThreadPanicked)),
+                "{r:?}"
+            );
+        });
+    }
+
+    // A byte view ends at the file's real length: the zero-padded tail of the last sector
+    // is cut, and nothing past it is read.
+    #[test]
+    fn a_byte_view_is_clipped_to_the_real_length() {
+        for len in [3 * 2048 + 700u64, 4 * 2048, 1000] {
+            let extents = vec![Extent {
+                start_lba: 0,
+                sector_count: 8,
+            }];
+            let policy = crate::sector::read_stage::ReadPolicy::Image { batch: 2 };
+            let view = ByteView {
+                prefix: Vec::new(),
+                len,
+            };
+            let (mut pf, _loss) = PrefetchedSectorSource::file_bytes(
+                PatternSource { capacity: 16 },
+                extents,
+                policy,
+                &Ctx::default(),
+                view,
+            )
+            .expect("spawn");
+            let (got, last) = drain_direct(&mut pf, 2, 16);
+            assert!(matches!(last, Ok(0)), "{last:?}");
+            assert_eq!(got.len() as u64, len, "len {len}");
+        }
+    }
+
     // An inner source that answers a mid-extent read with `Ok(0)` has quit early: the producer
     // must say so, not drop `tx` (which reads as clean EOF and lets `fill_extents` fabricate
     // zeros for the rest).

src/sector/read_stage.rs

@@ -20,7 +20,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
 // Ramp back to the preferred batch size after this many clean sectors (100
 // MiB = 51,200 sectors) — long enough that noisy zones can't trigger a
 // premature probe, short enough an isolated failure doesn't lock size 1.
-const PROBE_THRESHOLD_SECTORS: u32 = 100 * 1024 * 1024 / 2048;
+const PROBE_THRESHOLD_SECTORS: u32 = (100 * 1024 * 1024 / SECTOR) as u32;
 
 const SECTOR: usize = crate::consts::SECTOR_BYTES;
 

src/sector/stage.rs

@@ -448,6 +448,35 @@ mod tests {
         );
     }
 
+    // A stream shorter than one unit is BD-TS only if every source packet is synced at byte 4.
+    #[test]
+    fn a_short_stream_is_bd_ts_only_when_every_packet_syncs() {
+        let unit = ts_unit(false);
+        let short = unit[..3 * BD_SOURCE_PACKET_BYTES].to_vec();
+        assert_eq!(classify(&short), Kind::BdTs);
+        let mut one_bad = short.clone();
+        one_bad[BD_SOURCE_PACKET_BYTES + 4] = 0x00;
+        assert_eq!(classify(&one_bad), Kind::Opaque, "one packet off sync");
+        let mut sync_at_0 = short;
+        for p in sync_at_0.chunks_mut(BD_SOURCE_PACKET_BYTES) {
+            p[0] = 0x47;
+            p[4] = 0x00;
+        }
+        assert_eq!(classify(&sync_at_0), Kind::Opaque, "sync belongs at byte 4");
+    }
+
+    // KS-5: a decrypter that leaves CPI set on units it has already decrypted still streams:
+    // clear TS is never a missing-key refusal, wherever its CPI flag says otherwise.
+    #[test]
+    fn clear_ts_with_a_cpi_flag_is_not_refused() {
+        let bytes: Vec<u8> = (0..4).flat_map(|_| ts_unit(true)).collect();
+        let mut stage = Stage::lazy(Cursor::new(bytes.clone()), false);
+        let mut out = Vec::new();
+        stage.read_to_end(&mut out).expect("clear TS is not E7022");
+        assert_eq!(out, bytes);
+        assert_eq!(stage.kind(), Some(Kind::BdTs));
+    }
+
     // An Opaque stream is handed through byte-identical, even if a later sector looks like a
     // scrambled pack (a VOB muxed into an MKV attachment stays untouched).
     #[test]