Strip a trailing ; comment on M/I/V keydb fields, not just U

parse_disc_entry's U arm stripped a `; MKBv…` comment before hex-parsing the
value, but the M/I/V arms did not. When a real keydb.cfg disc row's LAST field
is M/I/V (no U after it), the comment glues onto that value and parse_hex16 of
"0x… ; MKBv64" fails — silently dropping a media/id/VUK key the file actually
holds, so the disc reports no key. Strip the comment in M/I/V too (a hex16 never
contains ';', so it is safe), with a red-before-green regression test.

.github/dependabot.yml

@@ -46,3 +46,18 @@ updates:
         update-types:
           - minor
           - patch
+    ignore:
+      # NOT a dependency: `dtolnay/rust-toolchain` is versioned by the RUST
+      # release it installs, and the tag we pin is the toolchain CI is pinned
+      # to on purpose -- precommit.sh runs the same one locally so a lint that
+      # passes on a developer's newer default cannot pass CI by accident.
+      #
+      # Dependabot reads those tags as semver and proposed 1.97.0 -> 1.100.0,
+      # a Rust version that does not exist. Every such PR 404s on toolchain
+      # download across all eight repos, and they regenerate weekly -- eight
+      # permanently-red PRs that promote.yml then has to special-case when it
+      # decides whether dev is green.
+      #
+      # Bumping the toolchain is a deliberate, all-eight-repos change, made by
+      # hand together with precommit.sh. There is nothing here for a bot.
+      - dependency-name: dtolnay/rust-toolchain

CHANGELOG.md

@@ -1,5 +1,60 @@
 # Changelog
 
+## [Unreleased]
+
+### Fixed
+
+- **A misconfigured key service is no longer quieter than an outage.** A
+  mistyped port, an `http://` URL, or a key-service address the SSRF guard
+  refuses all mean the same thing: the service was never asked, so nothing is
+  known about the disc. They were reported as an empty result — the same value a
+  service returns when it genuinely holds no key — so a standing
+  misconfiguration surfaced to the operator as "no key for this disc" and never
+  self-corrected, while a transient DNS failure was correctly reported as a
+  failure. All three are now failures, logged at error level and separated in
+  the log text from an outage: one says fix the URL, the other says wait. A
+  later key source's real key still wins, so nothing that used to rip stops
+  ripping.
+- **A replaced `keydb.cfg` can no longer be missed.** The parsed keydb is cached
+  behind the file's identity, and that identity was size plus modification time
+  — which are byte-identical when a file is replaced within the same second by
+  one of the same length, the common shape of a daily refresh on a filesystem
+  that records whole seconds. The old, superseded keys were then served for the
+  rest of the process's life while every lookup reported success. The identity
+  now includes the file's inode, and a cache entry is not trusted until the file
+  has been unmodified long enough for its timestamp to be meaningful.
+- **A corrupt keydb keeps saying so.** The summary warning that counts rejected
+  and duplicated rows is produced while parsing, and the cache skips parsing —
+  so a damaged keydb warned once, at whatever moment it was first read, and was
+  then served in silence to every later disc. The summary is now re-emitted on
+  every lookup that serves that file. An intact keydb stays silent.
+- **A keydb's identity and its contents are read through one file handle.** The
+  cache used to stat the path and then separately open it, so a replacement
+  landing between the two stored one file's identity beside another file's keys.
+
+### Security
+
+- **The host certificate's redaction is now pinned by a test.** `{:?}` on a
+  keydb host certificate does not print the AACS host private key, and never
+  did — but that depended entirely on libfreemkv's own redacting formatter,
+  with nothing in this crate to notice if it changed. The whole-database
+  formatter's test did not cover it. It does now, along with device keys.
+
+### Changed
+
+- **Breaking (API):** `KeyDb::disc_entries`' key and `DiscEntry::disc_hash` are
+  `Arc<str>` rather than `String` (a ~13 MB saving on the published keydb, which
+  is now held in memory). Code that requires an owned `String` from either needs
+  `.to_string()`. Unreleased in 1.6.4; recorded here because it shipped to git
+  consumers without a note.
+
+## [1.6.4] — 2026-08-15
+
+### Changed
+
+- **No functional change.** This crate ships alongside the rest of freemkv at a
+  matching version; its behaviour is untouched.
+
 ## [1.6.3] — 2026-08-10
 
 ### Changed

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "freemkv-keysources"
-version = "1.6.3"
+version = "1.6.4"
 edition = "2024"
 rust-version = "1.97"
 license = "MIT"
@@ -34,10 +34,10 @@ tracing = "0.1"
 lto = "thin"
 codegen-units = 1
 
-# libfreemkv is git-only (off crates.io). Redirect the bare `libfreemkv = "1.5"`
+# libfreemkv is git-only (off crates.io). Redirect the bare `libfreemkv = "1.6"`
 # req above to the git tag — committed and CI-visible. The release script
 # rewrites this tag to the new version before regenerating Cargo.lock. Local dev
 # overrides it with a path patch via the gitignored .cargo/config.toml (a
 # config-level [patch.crates-io] wins over this manifest one for the same crate).
 [patch.crates-io]
-libfreemkv = { git = "https://github.com/freemkv/libfreemkv", tag = "v1.6.2" }
+libfreemkv = { git = "https://github.com/freemkv/libfreemkv", tag = "v1.6.4" }

src/keydb.rs

@@ -31,6 +31,8 @@
 
 use std::io::{Read, Write};
 use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
 
 use crate::uks_from_vuk;
 use libfreemkv::aacs::derive::{derive_media_key_from_dk, derive_media_key_from_pk, derive_vuk};
@@ -55,15 +57,324 @@ pub struct UpdateResult {
     pub bytes: usize,
 }
 
+/// Widest observed granularity of a filesystem's stored mtime. HFS+, many
+/// container overlay filesystems and most network filesystems record whole
+/// seconds; 2 s leaves room for that plus rounding. Used by
+/// [`CacheEntry::is_settled`] — see the argument there.
+const MTIME_GRANULARITY: std::time::Duration = std::time::Duration::from_secs(2);
+
+/// The identity of the keydb file a cache entry was parsed from.
+///
+/// `(len, modified)` ALONE IS NOT AN IDENTITY. mtime is stored at whole-second
+/// resolution on HFS+, on many container overlays and on most network
+/// filesystems, so a writer that replaces `keydb.cfg` within the same second
+/// with a file of the same byte length leaves `(len, modified)` bit-identical —
+/// and the cache then serves the superseded parse for the rest of the process's
+/// life. For a key database that is not a stale page: it is serving retired or
+/// since-corrected keys while reporting success. The writers that hit this are
+/// real and named in this crate's own docs: the daily-refresh job, a second
+/// freemkv process, an operator's editor, a sync tool.
+///
+/// Two independent discriminators close it:
+///
+/// 1. `dev` + `ino`. Every writer that publishes ATOMICALLY (temp file, fsync,
+///    rename — what [`KeydbSource::save`] does, and what any correct publisher
+///    does) creates a NEW inode, so the stamp differs immediately and
+///    unconditionally, whatever the clock or the mtime resolution did. Zero on
+///    platforms without them (see [`Self::of`]), which costs nothing: rule 2
+///    stands alone.
+/// 2. A freshness rule on the cache entry — [`CacheEntry::is_settled`] — which
+///    covers the residual in-place rewrite that reuses the inode.
+///
+/// What this still does NOT catch: a writer that FORGES the timestamp, i.e.
+/// restores an mtime equal to the cached one (`touch -t`, `rsync --times`, a
+/// tar/backup restore) into the SAME inode with the SAME length. Nothing short
+/// of hashing the content can see that, and hashing 62 MiB per lookup is
+/// precisely the cost the cache exists to remove (~141 ms × ≥3 loads per rip).
+/// [`KeydbSource::save`] therefore keeps invalidating the cache directly, so the
+/// one write path this crate owns never depends on any of the above.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct FileStamp {
+    /// Filesystem device id. `0` where unavailable.
+    dev: u64,
+    /// Inode number — the discriminator an atomic rename always changes. `0`
+    /// where unavailable.
+    ino: u64,
+    len: u64,
+    modified: Option<std::time::SystemTime>,
+}
+
+impl FileStamp {
+    /// Stamp from a metadata block. Take it from `File::metadata` on an OPEN
+    /// handle (an `fstat`), never from a second `std::fs::metadata` on the path:
+    /// only the handle guarantees the stamp and the bytes describe one file.
+    fn of(meta: &std::fs::Metadata) -> Self {
+        // dev/ino are a Unix concept. Windows has a rough equivalent
+        // (volume serial + file index) but exposes it only through
+        // `MetadataExt` methods that are documented as unreliable on some
+        // filesystems; rule 2 alone is correct there, so the cross-platform
+        // fallback is simply "no inode discriminator" rather than a shaky one.
+        #[cfg(unix)]
+        let (dev, ino) = {
+            use std::os::unix::fs::MetadataExt;
+            (meta.dev(), meta.ino())
+        };
+        #[cfg(not(unix))]
+        let (dev, ino) = (0u64, 0u64);
+        Self {
+            dev,
+            ino,
+            len: meta.len(),
+            modified: meta.modified().ok(),
+        }
+    }
+}
+
+/// One cached parse: the file identity it came from, WHEN that identity was
+/// observed, the parsed database, and the parser's rejection counts.
+struct CacheEntry {
+    stamp: FileStamp,
+    /// Wall-clock time at which `stamp` was read off the open handle. The whole
+    /// point of storing it is [`Self::is_settled`].
+    stamped_at: std::time::SystemTime,
+    db: Arc<KeyDb>,
+    /// Retained, not just logged: [`KeydbSource::cached_db`] re-emits the
+    /// summary on every hit, because a cache hit skips `KeyDb::parse` and with
+    /// it the one warning that a corrupt keydb is being served.
+    stats: crate::keydb_format::ParseStats,
+}
+
+impl CacheEntry {
+    /// Whether this entry's stamp can be TRUSTED to change if the file changes.
+    ///
+    /// The proof, for a filesystem storing mtime at granularity `G`: suppose
+    /// this entry was stamped at `stamped_at` with `stamped_at - modified >= G`,
+    /// and some writer later rewrites the file in place at wall time
+    /// `T > stamped_at`. The recorded mtime `m'` of that write is within one
+    /// granule of `T`, so `m' >= T - G > stamped_at - G >= modified`. Hence
+    /// `m' != modified` and the stamp DOES change. The ambiguous window is
+    /// therefore exactly the one where the cached entry was stamped less than
+    /// `G` after the file's own mtime — i.e. we looked at the file while it was
+    /// still "hot" — and in that window the entry is refused and the file is
+    /// re-read. Once the file has been quiet for `G`, its entry settles and the
+    /// cache is fully effective again.
+    ///
+    /// Cost: after every keydb write, lookups re-parse for up to `G` (2 s).
+    /// The published keydb is rewritten daily, so that is ~2 s of re-parsing per
+    /// day against a cache that saves ~420 ms per rip.
+    ///
+    /// `modified: None` (a filesystem that reports no mtime) never settles: the
+    /// only two discriminators left would be size and inode, so the entry is
+    /// re-read every time rather than being trusted on a weaker basis.
+    ///
+    /// CLOCK CAVEAT: `stamped_at` is the local clock and `modified` is the
+    /// filesystem's. On a network share whose server clock runs AHEAD of the
+    /// client's by more than `G`, entries never settle (safe: more re-parsing).
+    /// A server clock BEHIND the client's makes an entry settle early, which
+    /// re-opens the same-second window it closes — the inode rule (1) is what
+    /// covers the realistic writers there.
+    fn is_settled(&self, granularity: std::time::Duration) -> bool {
+        self.stamp
+            .modified
+            .and_then(|m| self.stamped_at.duration_since(m).ok())
+            .is_some_and(|age| age >= granularity)
+    }
+}
+
 /// A [`KeySource`] backed by a local `keydb.cfg` file.
+///
+/// The parsed database is CACHED behind the file's identity stamp. The
+/// published keydb is ~62 MiB / ~181k entries and a single AACS-cert rip parses
+/// it repeatedly — `host_certs()` for the drive handshake (freemkv's
+/// `pipe.rs`/`engine.rs`, autorip's `keysource.rs` each build their own source),
+/// then the trait `host_certs(mkb)` during resolution, then `get_unit_keys` —
+/// and every one of those calls used to re-read all 62 MiB from disk and re-parse
+/// it from scratch, because the struct held nothing but a `PathBuf`.
+///
+/// A keydb replaced underneath this source is picked up on the next call, with
+/// ONE stated exception: a writer that rewrites the file in place, into the same
+/// inode, at the same byte length, having restored the previous mtime exactly
+/// (`touch -t` / `rsync --times` / a backup restore). Everything else — any
+/// atomic rename, any length change, any real clock advance — changes the stamp.
+/// [`FileStamp`] and [`CacheEntry::is_settled`] carry the full argument and the
+/// clock caveats; do not weaken either without re-reading them. (The earlier
+/// wording here — "an externally replaced keydb is still picked up" —
+/// was simply false for a same-second, same-length replacement, which is the
+/// common shape of a daily-refresh job on a whole-second-mtime filesystem.)
+///
+/// MEMORY RESIDENCY (accepted, and deliberately recorded rather than fixed
+/// here): a hit keeps the entire parsed keydb — every disc's Media Key / VID /
+/// VUK / Unit Keys, the processing- and device-key pools, and the host-cert
+/// PRIVATE keys — resident for the process's life, where the pre-cache code held
+/// one parse transiently. Nothing in this crate zeroizes on drop, so that
+/// material was always exposed to a core dump or a swapped page; the cache
+/// widens the WINDOW, it does not create the class. Closing it properly means
+/// zeroize-on-drop across `KeyDb`, `DiscEntry` and libfreemkv's `aacs::types`
+/// (which owns most of the buffers) — a cross-crate change, tracked separately,
+/// NOT smuggled in behind a cache fix.
 pub struct KeydbSource {
     path: PathBuf,
+    /// `Mutex` (not `RwLock`): the guarded section is a stamp comparison plus an
+    /// `Arc` clone, so there is nothing for concurrent readers to win. Poisoning
+    /// is recovered from rather than propagated — a panic elsewhere must not
+    /// turn every later key lookup into a panic.
+    cache: Mutex<Option<CacheEntry>>,
+    /// How many times the file was actually read + parsed (cache MISSES). The
+    /// only honest way to assert the cache from a test: timing is flaky and the
+    /// parsed value is identical either way. Cheap enough to keep in release.
+    parses: AtomicUsize,
+    /// The `G` of [`CacheEntry::is_settled`]. A field, not the constant inlined,
+    /// so the two discriminators can be tested ONE AT A TIME: with `G` at zero
+    /// every entry settles, which isolates the inode/len rule; with `G` enormous
+    /// nothing settles, which isolates the freshness rule. Testing them together
+    /// is how a stale-cache test ends up passing whether or not the fix is
+    /// present — on APFS (nanosecond mtimes) the naive same-second scenario is
+    /// simply not reproducible, so a test that "reproduces" it there is
+    /// asserting nothing.
+    mtime_granularity: std::time::Duration,
+    /// How many corruption summaries were emitted (see
+    /// [`KeydbSource::emit_parse_stats`]). Same rationale as `parses`: the
+    /// alternative instrument is a `tracing` subscriber in dev-dependencies.
+    warnings: AtomicUsize,
 }
 
 impl KeydbSource {
     /// A keydb source reading the given `keydb.cfg` path.
     pub fn new(path: impl Into<PathBuf>) -> Self {
-        Self { path: path.into() }
+        Self {
+            path: path.into(),
+            cache: Mutex::new(None),
+            parses: AtomicUsize::new(0),
+            mtime_granularity: MTIME_GRANULARITY,
+            warnings: AtomicUsize::new(0),
+        }
+    }
+
+    /// Log the parser's rejection summary and count the emission.
+    ///
+    /// The count is the only way to assert, without pulling a `tracing`
+    /// subscriber into dev-dependencies, that the corruption warning is NOT
+    /// swallowed by the cache. See [`Self::cached_db`].
+    fn emit_parse_stats(&self, stats: &crate::keydb_format::ParseStats) {
+        if stats.log() {
+            self.warnings.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+
+    /// Corruption summaries emitted so far. Test-only observability, paired with
+    /// [`Self::parse_count`]: a corrupt keydb must warn once per LOOKUP, while
+    /// still parsing once.
+    #[cfg(test)]
+    fn warning_count(&self) -> usize {
+        self.warnings.load(Ordering::Relaxed)
+    }
+
+    /// Test-only: override the settle window (see the field's doc).
+    #[cfg(test)]
+    fn with_mtime_granularity(mut self, g: std::time::Duration) -> Self {
+        self.mtime_granularity = g;
+        self
+    }
+
+    /// The parsed keydb, from cache when the file is provably unchanged.
+    ///
+    /// Errors are the same ones [`KeyDb::load`] produces, unchanged: a missing
+    /// file is `NotFound` (the documented benign case), an over-cap or non-UTF-8
+    /// file is `InvalidData`. The stamp costs an `fstat`, not a 62 MiB read.
+    ///
+    /// ONE OPEN, ONE IDENTITY. The file is opened once and both the stamp
+    /// (`fstat` on that handle) and, on a miss, the bytes come from it. The
+    /// previous shape — `std::fs::metadata(path)` and then a separate
+    /// `KeyDb::load(path)` — resolved the path TWICE with no atomicity between
+    /// them, so a rename landing in the gap cached the new file's stamp beside
+    /// the old file's contents (or the reverse); the pairing was then served,
+    /// looking perfectly valid, until the stamp changed again. The stamp is read
+    /// BEFORE the bytes on purpose: a write racing the read then leaves the OLD
+    /// stamp cached, which the next call detects, whereas stamping afterwards
+    /// would file torn content under a fresh-looking identity and pin it.
+    fn cached_db(&self) -> std::io::Result<Arc<KeyDb>> {
+        let file = std::fs::File::open(&self.path)?;
+        let stamp = FileStamp::of(&file.metadata()?);
+        let stamped_at = std::time::SystemTime::now();
+        let mut guard = self.cache.lock().unwrap_or_else(|e| e.into_inner());
+        if let Some(entry) = guard.as_ref()
+            && entry.stamp == stamp
+            && entry.is_settled(self.mtime_granularity)
+        {
+            // Re-emit the parser's rejection summary on EVERY hit, not just on
+            // the parse that produced it. `KeyDb::parse` is what logs it, and a
+            // hit skips `parse` entirely — so a daemon that cached a corrupt
+            // keydb logged its rejected-row counts exactly once, at whatever
+            // moment it first read the file, and then served that same damaged
+            // database to every later rip in silence. Repetition is the correct
+            // failure mode here: the line is emitted only when the file really
+            // does have rejected or duplicate rows, and a warning an operator
+            // can still see on the tenth rip is worth more than one they had to
+            // catch on the first.
+            self.emit_parse_stats(&entry.stats);
+            return Ok(entry.db.clone());
+        }
+        let (db, stats) = KeyDb::load_counted(file, &self.path)?;
+        self.emit_parse_stats(&stats);
+        let db = Arc::new(db);
+        self.parses.fetch_add(1, Ordering::Relaxed);
+        *guard = Some(CacheEntry {
+            stamp,
+            stamped_at,
+            db: db.clone(),
+            stats,
+        });
+        Ok(db)
+    }
+
+    /// Drop the cached parse. Called after this source WRITES the file, so the
+    /// next lookup re-reads it without depending on mtime resolution.
+    fn invalidate_cache(&self) {
+        *self.cache.lock().unwrap_or_else(|e| e.into_inner()) = None;
+    }
+
+    /// Cache misses so far — the number of real file reads + parses. Test-only
+    /// observability for the cache (see [`Self::cached_db`]).
+    #[cfg(test)]
+    fn parse_count(&self) -> usize {
+        self.parses.load(Ordering::Relaxed)
+    }
+
+    /// Turn a [`KeyDb::load`] failure into this source's verdict.
+    ///
+    /// A MISSING keydb is the documented benign case: the app may simply not
+    /// have one, another source may hold the key, and "no keydb" is genuinely
+    /// "no key here". Everything else is NOT that: `InvalidData` is a keydb over
+    /// the 128 MiB cap or one that is not valid UTF-8 — i.e. corrupt, truncated,
+    /// or a half-finished download — and a permission/IO failure means the file
+    /// could not be read at all. Reporting either as an empty result made a
+    /// broken keydb indistinguishable from "this disc has no key" (the same
+    /// conflation as the seven-hour 502), with no log to notice it by. Both are
+    /// now logged AND surfaced as errors so the resolver reports a source
+    /// failure instead of `E7022`.
+    fn load_failure(&self, e: &std::io::Error) -> Option<Error> {
+        match e.kind() {
+            std::io::ErrorKind::NotFound => None,
+            std::io::ErrorKind::InvalidData => {
+                tracing::warn!(
+                    target: "freemkv::keysource",
+                    path = %self.path.display(),
+                    "keydb.cfg is unusable (over the size cap or not valid UTF-8) — this is a corrupt or truncated keydb, NOT a disc without a key"
+                );
+                Some(Error::KeydbInvalid)
+            }
+            kind => {
+                tracing::warn!(
+                    target: "freemkv::keysource",
+                    path = %self.path.display(),
+                    io_kind = ?kind,
+                    "keydb.cfg could not be read — NOT a disc without a key"
+                );
+                Some(Error::KeydbLoad {
+                    path: self.path.display().to_string(),
+                })
+            }
+        }
     }
 
     /// Validate, decompress, and crash-safely persist raw keydb bytes (plain
@@ -92,11 +403,12 @@ impl KeydbSource {
             .lines()
             .filter(|l| {
                 let t = l.trim();
-                // Mirror KeyDb::parse's disc-entry rule EXACTLY (keydb_format.rs:
-                // a "0x" line is only an entry if it also contains " = "), so
-                // save() never validates + persists content that parses to zero
-                // usable entries (e.g. a stray "0xDEADBEEF" comment line).
-                (t.starts_with("0x") && t.contains(" = "))
+                // Mirror KeyDb::parse's disc-entry rule EXACTLY by CALLING it
+                // (`is_disc_entry_line`), so save() never validates + persists
+                // content that parses to zero usable entries (e.g. a stray
+                // "0xDEADBEEF" comment line) and the two can no longer drift —
+                // they did drift once already, on the `0X` case rule.
+                crate::keydb_format::is_disc_entry_line(t)
                     || t.starts_with("| DK")
                     || t.starts_with("| PK")
                     || t.starts_with("| HC")
@@ -108,6 +420,8 @@ impl KeydbSource {
         }
 
         write_atomic(&self.path, &text)?;
+        // The file this source reads was just replaced; drop the parsed copy.
+        self.invalidate_cache();
 
         Ok(UpdateResult {
             path: self.path.clone(),
@@ -143,9 +457,14 @@ impl KeydbSource {
     /// filtering is applied (passes `None`). The [`KeySource::host_certs`] TRAIT
     /// method wires the real MKB generation through for revocation filtering.
     pub fn host_certs(&self) -> Vec<HostCert> {
-        match KeyDb::load(&self.path) {
+        match self.cached_db() {
             Ok(db) => db.host_certs(None),
-            Err(_) => Vec::new(),
+            // No error channel here (the scan-options builder wants a list), so
+            // the failure can only be LOGGED — but it must not be invisible.
+            Err(e) => {
+                let _ = self.load_failure(&e);
+                Vec::new()
+            }
         }
     }
 
@@ -330,17 +649,27 @@ fn write_atomic(path: &Path, text: &str) -> Result<(), Error> {
 }
 
 impl KeySource for KeydbSource {
-    /// Resolve this disc's base per-CPS-unit Unit Keys from the keydb. A missing /
-    /// unreadable keydb is not an error — it simply yields no keys (another
-    /// source may have them), the same as the library's own loader.
+    /// Resolve this disc's base per-CPS-unit Unit Keys from the keydb.
+    ///
+    /// A MISSING keydb is not an error — it simply yields no keys (another
+    /// source may have them), the same as the library's own loader. A keydb that
+    /// exists but cannot be USED (over the size cap, not UTF-8, unreadable) is a
+    /// source failure and returns `Err`: it says nothing about whether this disc
+    /// has a key, and reporting it as an empty result is the same
+    /// looks-like-success failure as a 502 reported as `E7022`.
     ///
     /// The keydb carries no AACS 2.1 forensic index keys today, so it does not
     /// override `get_fmts_indexes` — the default (empty) opts it out, and an FMTS
     /// disc's forensic set comes from the online source.
     fn get_unit_keys(&self, ctx: &dyn ResolveCtx) -> Result<Vec<UnitKey>, Error> {
-        match KeyDb::load(&self.path) {
+        match self.cached_db() {
             Ok(db) => Ok(Self::unit_keys_from(&db, ctx)),
-            Err(_) => Ok(Vec::new()),
+            Err(e) => match self.load_failure(&e) {
+                // Missing keydb: the documented benign miss.
+                None => Ok(Vec::new()),
+                // Corrupt / unreadable keydb: a SOURCE FAILURE, not a miss.
+                Some(err) => Err(err),
+            },
         }
     }
 
@@ -350,9 +679,13 @@ impl KeySource for KeydbSource {
     /// `; Revoked in MKBv<N>` annotation): a cert revoked at generation `R` is
     /// withheld once the disc's generation reaches `R`.
     fn host_certs(&self, mkb: Option<u32>) -> Vec<HostCert> {
-        match KeyDb::load(&self.path) {
+        match self.cached_db() {
             Ok(db) => db.host_certs(mkb),
-            Err(_) => Vec::new(),
+            // Vec-returning trait method: log the failure, return nothing.
+            Err(e) => {
+                let _ = self.load_failure(&e);
+                Vec::new()
+            }
         }
     }
 
@@ -365,6 +698,29 @@ impl KeySource for KeydbSource {
 mod tests {
     use super::*;
     use crate::keydb_format::DiscEntry;
+
+    /// A fixed, in-the-past mtime — `touch -t CCYYMMDDhhmm.ss`.
+    #[cfg(unix)]
+    const PINNED_MTIME: &str = "202601011200.00";
+
+    /// Force a file's mtime to [`PINNED_MTIME`], so two different files can be
+    /// made bit-identical to a `(len, mtime)` stamp on purpose. The cache tests
+    /// that matter all depend on FORCING the indistinguishable case rather than
+    /// hoping the filesystem produces it: on APFS (nanosecond mtimes) a
+    /// same-second replacement is otherwise detected by mtime alone, and a test
+    /// that relies on that passes with the fix reverted — which is worth nothing.
+    #[cfg(unix)]
+    fn pin_mtime(p: &std::path::Path) {
+        let ok = std::process::Command::new("touch")
+            .arg("-t")
+            .arg(PINNED_MTIME)
+            .arg(p)
+            .status()
+            .expect("touch must run")
+            .success();
+        assert!(ok, "touch -t failed");
+    }
+
     use libfreemkv::aacs::derive::derive_vuk;
     use libfreemkv::aacs::types::DeviceKey;
     use std::collections::HashMap;
@@ -796,6 +1152,402 @@ mod tests {
         std::fs::remove_dir_all(&dir).ok();
     }
 
+    // ── Caching: the 62 MiB / 181k-entry keydb is parsed ONCE ────────────────
+
+    /// A single AACS-cert rip drives this source at least three times — the
+    /// scan-options builder's inherent `host_certs()`, the trait
+    /// `host_certs(mkb)` during the handshake, then `get_unit_keys` — and each
+    /// call used to re-read and re-parse the WHOLE file, because `KeydbSource`
+    /// held nothing but a `PathBuf`. Measured on the real 62 MiB / 184,860-entry
+    /// keydb: ~140 ms per parse in `--release`, i.e. ~420 ms and ~186 MiB of
+    /// file reads per rip, against ~1 µs for a cache hit.
+    ///
+    /// Catches the mutation that drops the cache (every call re-parses: count 3)
+    /// and the mutation that makes the cache unconditional (see the
+    /// invalidation tests below).
+    ///
+    /// The mtime is pinned into the PAST first, because a cache entry is only
+    /// trusted once the file has been quiet for the mtime granularity
+    /// ([`CacheEntry::is_settled`]) — a keydb written microseconds ago is
+    /// deliberately re-read. Unix-only: pinning an mtime needs `touch -t` (no
+    /// `filetime` dependency for a test).
+    #[cfg(unix)]
+    #[test]
+    fn repeated_lookups_parse_the_keydb_once() {
+        let dir = scratch("cache-hit");
+        let path = dir.join("keydb.cfg");
+        let line = format!(
+            "| HC | HOST_PRIV_KEY 0x{} | HOST_CERT 0x{}\n",
+            "00".repeat(20),
+            "00".repeat(92)
+        );
+        std::fs::write(&path, &line).unwrap();
+        pin_mtime(&path);
+
+        let src = KeydbSource::new(&path);
+        assert_eq!(src.host_certs().len(), 1);
+        assert_eq!(KeySource::host_certs(&src, Some(70)).len(), 1);
+        assert!(
+            src.get_unit_keys(&ctx(HASH, Vec::new(), None))
+                .unwrap()
+                .is_empty()
+        );
+        assert_eq!(
+            src.parse_count(),
+            1,
+            "three lookups over an unchanged file must parse it once"
+        );
+    }
+
+    /// The cache must not go stale: a keydb replaced UNDERNEATH the source (the
+    /// daily-refresh thread writing through a different `KeydbSource`, or an
+    /// operator dropping in a new file) has a different size/mtime stamp and
+    /// must be re-read. Catches the mutation that caches on first load and never
+    /// re-stats — which would pin a rip to a keydb deleted hours earlier.
+    #[test]
+    fn a_changed_keydb_file_is_reparsed() {
+        let dir = scratch("cache-stamp");
+        let path = dir.join("keydb.cfg");
+        let hc = |b: &str, n: usize| {
+            format!(
+                "| HC | HOST_PRIV_KEY 0x{} | HOST_CERT 0x{}\n",
+                "00".repeat(20),
+                b.repeat(n)
+            )
+        };
+        std::fs::write(&path, hc("00", 92)).unwrap();
+        let src = KeydbSource::new(&path);
+        assert_eq!(src.host_certs()[0].certificate.len(), 92);
+
+        // A different LENGTH guarantees a different stamp regardless of the
+        // filesystem's mtime resolution.
+        std::fs::write(&path, hc("11", 100)).unwrap();
+        assert_eq!(
+            src.host_certs()[0].certificate.len(),
+            100,
+            "a replaced keydb must be re-read, not served from a stale cache"
+        );
+        assert_eq!(src.parse_count(), 2, "one parse per distinct file");
+    }
+
+    // ── The two stale-cache discriminators, tested ONE AT A TIME ────────────
+    //
+    // The defect both of these pin: `(len, mtime)` is not a file identity. An
+    // external writer — the daily-refresh job, a second freemkv process, an
+    // editor, a sync tool — that replaces `keydb.cfg` inside one mtime granule
+    // with a file of the SAME length leaves that pair bit-identical, and the
+    // source then serves the superseded parse for the rest of the process's
+    // life: retired or since-corrected keys, reported as success.
+    //
+    // Each test disables the OTHER discriminator through `mtime_granularity`, so
+    // neither can pass on the strength of the one it is not testing, and neither
+    // depends on the host filesystem's real mtime resolution (on APFS the naive
+    // scenario is not reproducible at all).
+
+    /// Discriminator 1, the inode: a keydb replaced by ATOMIC RENAME — how
+    /// every correct publisher, including this crate's own `save()`, installs a
+    /// new file — must be re-read even when length and mtime are identical.
+    /// `mtime_granularity: 0` makes every entry settle, so ONLY dev+ino can
+    /// distinguish the two files here.
+    ///
+    /// Catches the mutation that drops `dev`/`ino` from `FileStamp` (the
+    /// original `(len, modified)` stamp): the source then keeps answering with
+    /// the previous keydb's keys.
+    #[cfg(unix)]
+    #[test]
+    fn a_same_length_same_mtime_rename_is_still_re_read() {
+        let dir = scratch("cache-inode");
+        let path = dir.join("keydb.cfg");
+        let replacement = dir.join("keydb.cfg.new");
+        let body = |k: &str| format!("0xaabb = T | U | 1-0x{k}\n");
+
+        std::fs::write(&path, body(&"01".repeat(16))).unwrap();
+        pin_mtime(&path);
+        let src = KeydbSource::new(&path).with_mtime_granularity(std::time::Duration::ZERO);
+        assert_eq!(
+            src.get_unit_keys(&ctx(HASH, Vec::new(), None)).unwrap()[0].key,
+            [0x01u8; 16]
+        );
+
+        // Same length, same mtime, different inode.
+        std::fs::write(&replacement, body(&"02".repeat(16))).unwrap();
+        pin_mtime(&replacement);
+        std::fs::rename(&replacement, &path).unwrap();
+        assert_eq!(
+            std::fs::metadata(&path).unwrap().len(),
+            body(&"01".repeat(16)).len() as u64,
+            "the two keydbs must be the same size for this test to mean anything"
+        );
+
+        assert_eq!(
+            src.get_unit_keys(&ctx(HASH, Vec::new(), None)).unwrap()[0].key,
+            [0x02u8; 16],
+            "a renamed-in keydb of the same size and mtime must not be served from the stale cache"
+        );
+        assert_eq!(src.parse_count(), 2, "the replacement must be re-parsed");
+    }
+
+    /// Discriminator 2, the settle window: a keydb rewritten IN PLACE — same
+    /// inode, same length, mtime forced back to the same value — must still be
+    /// re-read while the cached entry is too young to trust. `mtime_granularity`
+    /// is set to a decade so nothing ever settles, which is the same condition a
+    /// real same-second, whole-second-mtime replacement creates and the only way
+    /// to reproduce it deterministically on a nanosecond-mtime filesystem.
+    ///
+    /// Catches the mutation that deletes the `is_settled` check from
+    /// `cached_db` (every stamp match trusted unconditionally, which is what the
+    /// original code did): dev, ino, len and mtime are ALL identical here, so
+    /// without the freshness rule the source serves the old keys forever.
+    #[cfg(unix)]
+    #[test]
+    fn an_in_place_rewrite_under_an_unsettled_stamp_is_still_re_read() {
+        use std::io::Write as _;
+        let dir = scratch("cache-settle");
+        let path = dir.join("keydb.cfg");
+        let body = |k: &str| format!("0xaabb = T | U | 1-0x{k}\n");
+
+        std::fs::write(&path, body(&"01".repeat(16))).unwrap();
+        pin_mtime(&path);
+        let ino_before = std::fs::metadata(&path)
+            .map(|m| std::os::unix::fs::MetadataExt::ino(&m))
+            .unwrap();
+        let src = KeydbSource::new(&path)
+            .with_mtime_granularity(std::time::Duration::from_secs(10 * 365 * 24 * 3600));
+        assert_eq!(
+            src.get_unit_keys(&ctx(HASH, Vec::new(), None)).unwrap()[0].key,
+            [0x01u8; 16]
+        );
+
+        // Overwrite the SAME inode in place (no truncate, no rename), same
+        // length, then force the mtime back: every field of the old stamp
+        // matches.
+        let mut f = std::fs::OpenOptions::new().write(true).open(&path).unwrap();
+        f.write_all(body(&"02".repeat(16)).as_bytes()).unwrap();
+        f.sync_all().unwrap();
+        drop(f);
+        pin_mtime(&path);
+        assert_eq!(
+            std::fs::metadata(&path)
+                .map(|m| std::os::unix::fs::MetadataExt::ino(&m))
+                .unwrap(),
+            ino_before,
+            "the rewrite must reuse the inode for this test to mean anything"
+        );
+
+        assert_eq!(
+            src.get_unit_keys(&ctx(HASH, Vec::new(), None)).unwrap()[0].key,
+            [0x02u8; 16],
+            "an in-place rewrite under an indistinguishable stamp must not be served from cache"
+        );
+    }
+
+    /// A corrupt-but-parseable keydb must keep saying so. `ParseStats::log` runs
+    /// inside `KeyDb::parse`, and a cache hit skips `parse` — so once the cache
+    /// landed, the rejected-row summary for a given file was emitted exactly
+    /// ONCE for the entire life of the process, while that same damaged file
+    /// went on being served to every subsequent rip. A long-running daemon
+    /// (autorip) logs it at startup, the operator misses it in the startup noise
+    /// or loses it to log rotation, and there is no second chance for as long as
+    /// the process runs.
+    ///
+    /// Catches the mutation that deletes the `emit_parse_stats` call from the
+    /// cache-HIT arm of `cached_db`: warnings would drop to 1 while the file is
+    /// still served three times.
+    #[cfg(unix)]
+    #[test]
+    fn a_corrupt_keydb_warns_on_every_lookup_not_only_on_the_parse() {
+        let dir = scratch("cache-warn");
+        let path = dir.join("keydb.cfg");
+        // Parseable (one good disc row) but damaged: two rows the parser must
+        // reject and count.
+        std::fs::write(
+            &path,
+            "0xaabb = T | U | 1-0x00112233445566778899aabbccddeeff\n\
+             | PK | 0xnothex\n\
+             | DK | DEVICE_KEY 0xZZ | DEVICE_NODE 0x0800 | KEY_UV 0x00000400 | KEY_U_MASK_SHIFT 0x17\n",
+        )
+        .unwrap();
+        pin_mtime(&path);
+
+        let src = KeydbSource::new(&path);
+        for _ in 0..3 {
+            assert!(
+                !src.get_unit_keys(&ctx(HASH, Vec::new(), None))
+                    .unwrap()
+                    .is_empty(),
+                "the healthy row must still resolve — corruption is not a source failure here"
+            );
+        }
+        assert_eq!(
+            src.parse_count(),
+            1,
+            "the cache must still do its job: one parse for three lookups"
+        );
+        assert_eq!(
+            src.warning_count(),
+            3,
+            "the corruption summary must be re-emitted on every cache hit, not once per process"
+        );
+    }
+
+    /// A HEALTHY keydb must stay silent — the re-emission above must not turn
+    /// into a per-lookup log line for every operator with an intact file.
+    ///
+    /// Catches the mutation that makes `emit_parse_stats` unconditional (drops
+    /// `ParseStats::log`'s empty-counts early return).
+    #[cfg(unix)]
+    #[test]
+    fn a_healthy_keydb_warns_on_no_lookup() {
+        let dir = scratch("cache-quiet");
+        let path = dir.join("keydb.cfg");
+        std::fs::write(
+            &path,
+            "0xaabb = T | U | 1-0x00112233445566778899aabbccddeeff\n",
+        )
+        .unwrap();
+        pin_mtime(&path);
+
+        let src = KeydbSource::new(&path);
+        for _ in 0..3 {
+            let _ = src.get_unit_keys(&ctx(HASH, Vec::new(), None)).unwrap();
+        }
+        assert_eq!(
+            src.warning_count(),
+            0,
+            "an intact keydb must produce no corruption summary at all"
+        );
+    }
+
+    /// The cost side of the settle rule, asserted so it is a decision and not an
+    /// accident: a keydb written just now is re-read on every lookup until it
+    /// has been quiet for the granularity. That is the price of never serving a
+    /// same-second replacement, and it is bounded — 2 s after each write of a
+    /// file that is rewritten daily.
+    ///
+    /// Catches the mutation that widens the window to something unbounded (a
+    /// granularity of hours would re-parse 62 MiB on every lookup all day) by
+    /// pairing with `repeated_lookups_parse_the_keydb_once`, which proves the
+    /// entry DOES settle once the mtime is in the past.
+    #[test]
+    fn a_freshly_written_keydb_is_not_trusted_from_cache() {
+        let dir = scratch("cache-fresh");
+        let path = dir.join("keydb.cfg");
+        std::fs::write(
+            &path,
+            "0xaabb = T | U | 1-0x00112233445566778899aabbccddeeff\n",
+        )
+        .unwrap();
+
+        let src = KeydbSource::new(&path);
+        assert!(
+            !src.get_unit_keys(&ctx(HASH, Vec::new(), None))
+                .unwrap()
+                .is_empty()
+        );
+        assert!(
+            !src.get_unit_keys(&ctx(HASH, Vec::new(), None))
+                .unwrap()
+                .is_empty()
+        );
+        assert_eq!(
+            src.parse_count(),
+            2,
+            "a keydb whose mtime is younger than the granularity must be re-read, not trusted"
+        );
+    }
+
+    /// `save()` replaces the very file this source reads, so it drops the cached
+    /// parse ITSELF rather than trusting the (size, mtime) stamp: a same-size
+    /// rewrite that lands inside the filesystem's timestamp granularity is the
+    /// one change the stamp cannot see — and it is exactly the change this crate
+    /// performs (the daily-refresh thread re-saving a keydb through the source
+    /// it then reads from).
+    ///
+    /// The test forces that indistinguishable case rather than hoping for it:
+    /// the file is written, stamped to a FIXED mtime, cached, re-saved with
+    /// different keys of the SAME length, and stamped back to the same mtime. To
+    /// the stamp the two files are identical, so only the explicit invalidation
+    /// can produce the new key. Unix-only because pinning an mtime needs
+    /// `touch -t` (no `filetime` dependency for one test).
+    #[cfg(unix)]
+    #[test]
+    fn save_invalidates_the_cache_even_when_the_file_stamp_is_unchanged() {
+        let dir = scratch("cache-save");
+        let path = dir.join("keydb.cfg");
+        let k1 = "01".repeat(16);
+        let k2 = "02".repeat(16);
+        let body = |k: &str| format!("0xaabb = T | U | 1-0x{k}\n");
+
+        std::fs::write(&path, body(&k1)).unwrap();
+        pin_mtime(&path);
+
+        let src = KeydbSource::new(&path);
+        let first = src.get_unit_keys(&ctx(HASH, Vec::new(), None)).unwrap();
+        assert_eq!(first[0].key, [0x01u8; 16]);
+
+        // Same byte length, written THROUGH this source, then stamped back so
+        // (size, mtime) is bit-identical to what the cache recorded.
+        src.save(body(&k2).as_bytes()).expect("save must succeed");
+        pin_mtime(&path);
+        assert_eq!(
+            std::fs::metadata(&path).unwrap().len(),
+            body(&k1).len() as u64,
+            "the two keydbs must be the same size for this test to mean anything"
+        );
+
+        let second = src.get_unit_keys(&ctx(HASH, Vec::new(), None)).unwrap();
+        assert_eq!(
+            second[0].key, [0x02u8; 16],
+            "a save through this source must invalidate its own cache, not wait for the stamp to change"
+        );
+    }
+
+    // ── A corrupt keydb is a source FAILURE, not "this disc has no key" ──────
+
+    /// `KeyDb::load` fails distinguishably for a keydb that is over the 128 MiB
+    /// cap or not valid UTF-8 — a truncated download, a half-written file, a
+    /// binary blob dropped in by mistake. Both used to collapse into
+    /// `Ok(Vec::new())` with ZERO tracing, i.e. into the benign "no key for this
+    /// disc" the resolver reports as `E7022` — the same looks-like-success
+    /// failure as the seven-hour 502. Only a MISSING file is genuinely benign.
+    ///
+    /// Catches the mutation that restores `Err(_) => Ok(Vec::new())`.
+    #[test]
+    fn a_corrupt_keydb_is_an_error_not_an_empty_answer() {
+        let dir = scratch("corrupt");
+        let path = dir.join("keydb.cfg");
+        // Invalid UTF-8 (a lone continuation byte) → InvalidData from load.
+        std::fs::write(&path, [0x30u8, 0x78, 0xFF, 0xFE, 0x0A]).unwrap();
+
+        let src = KeydbSource::new(&path);
+        let err = src
+            .get_unit_keys(&ctx(HASH, Vec::new(), None))
+            .expect_err("a corrupt keydb must not look like a disc with no key");
+        assert_eq!(
+            err.code(),
+            libfreemkv::error::E_KEYDB_INVALID,
+            "an unusable keydb must report itself, not E7022"
+        );
+        assert_ne!(
+            err.code(),
+            libfreemkv::error::E_NO_DISC_KEY,
+            "a corrupt keydb says NOTHING about whether this disc has a key"
+        );
+    }
+
+    /// The contrast case, through the same code: a keydb that simply is not
+    /// there stays the documented benign miss (`Ok(empty)`), so the fix above
+    /// cannot have turned "no keydb configured" into a hard failure.
+    #[test]
+    fn a_missing_keydb_is_still_a_benign_empty() {
+        let src = KeydbSource::new("/nonexistent/path/keydb.cfg");
+        assert!(
+            src.get_unit_keys(&ctx(HASH, Vec::new(), None))
+                .expect("a missing keydb is not an error")
+                .is_empty()
+        );
+    }
+
     // ── save / update (moved from libfreemkv::keydb) ──────────────────────────
 
     // Per project convention, tests never touch /tmp (wiped on reboot). Anchor

src/lib.rs

@@ -60,6 +60,16 @@ pub(crate) fn uks_from_vuk(vuk: &[u8; 16], enc_title_keys: &[[u8; 16]]) -> Vec<U
 /// so the "which sources, in what order" policy lives entirely with the
 /// application, not the library. `MultiSource` is itself a [`KeySource`], so it
 /// nests and composes.
+///
+/// **A source that could not ANSWER is not a source that answered "no key".**
+/// When no inner source produces keys, the composition returns `Err` if any
+/// inner source failed, and `Ok(empty)` only when every source genuinely
+/// answered and none held a key. This mirrors libfreemkv's own resolver
+/// (`keysource::drive_unit_keys`, which tracks an `errored` flag, and
+/// `source_failure`, which stamps the first failure onto the disc) — a
+/// composition that swallowed the `Err` would hand the resolver a clean
+/// `Ok(empty)`, and an outage would once again be reported to the operator as
+/// `E7022 No key source has a decryption key for this disc`.
 pub struct MultiSource {
     sources: Vec<Box<dyn KeySource>>,
 }
@@ -71,35 +81,61 @@ impl MultiSource {
     }
 }
 
-impl KeySource for MultiSource {
-    /// Try each inner source in order; the FIRST to return a non-empty base Unit
-    /// Key set wins. An inner source that returns empty OR errors is treated as
-    /// "no key here" and the next is tried (a single source failure never blocks
-    /// the chain). All sources exhausted → empty.
-    fn get_unit_keys(&self, ctx: &dyn ResolveCtx) -> Result<Vec<UnitKey>, libfreemkv::Error> {
-        for s in &self.sources {
-            if let Ok(uks) = s.get_unit_keys(ctx)
-                && !uks.is_empty()
-            {
-                return Ok(uks);
+/// Drive `sources` in order and return the first non-empty result — preserving
+/// the Ok/Err distinction when nothing resolves.
+///
+/// A source failure never BLOCKS the chain (a later source is still tried, and a
+/// later success still wins outright, error or no error), but it is not
+/// forgotten either: with no keys anywhere, the FIRST failure is returned. First
+/// (not last) matches libfreemkv's `source_failure` rule, so the most-preferred
+/// source's reason is the one the operator is told about, in the same order the
+/// caller expressed a preference for successes.
+///
+/// `get` selects which trait method to drive, so the base and forensic paths
+/// share ONE implementation of this rule and cannot drift apart — the forensic
+/// half carried the identical bug.
+fn first_non_empty(
+    sources: &[Box<dyn KeySource>],
+    get: impl Fn(&dyn KeySource, &dyn ResolveCtx) -> Result<Vec<UnitKey>, libfreemkv::Error>,
+    ctx: &dyn ResolveCtx,
+) -> Result<Vec<UnitKey>, libfreemkv::Error> {
+    let mut first_failure: Option<libfreemkv::Error> = None;
+    for s in sources {
+        match get(s.as_ref(), ctx) {
+            Ok(uks) if !uks.is_empty() => return Ok(uks),
+            Ok(_) => {}
+            Err(e) => {
+                if first_failure.is_none() {
+                    first_failure = Some(e);
+                }
             }
         }
-        Ok(Vec::new())
+    }
+    match first_failure {
+        // At least one source could not answer, and nothing else had a key: the
+        // composition does NOT know that this disc has no key.
+        Some(e) => Err(e),
+        // Every source answered; none holds a key. The genuine miss.
+        None => Ok(Vec::new()),
+    }
+}
+
+impl KeySource for MultiSource {
+    /// Try each inner source in order; the FIRST to return a non-empty base Unit
+    /// Key set wins, even if an earlier source failed. All sources exhausted →
+    /// `Ok(empty)` if they all ANSWERED, or the first source failure if any could
+    /// not (see [`MultiSource`]).
+    fn get_unit_keys(&self, ctx: &dyn ResolveCtx) -> Result<Vec<UnitKey>, libfreemkv::Error> {
+        first_non_empty(&self.sources, |s, c| s.get_unit_keys(c), ctx)
     }
 
     /// Forensic-index counterpart: try each inner source's `get_fmts_indexes` in
-    /// the same order and return the first non-empty set. A source with no
-    /// forensic material (the keydb, via the trait default) contributes empty and
-    /// is skipped; on today's discs the online source answers.
+    /// the same order and return the first non-empty set, under the same
+    /// failure-preserving rule. A source with no forensic material (the keydb,
+    /// via the trait default) contributes empty and is skipped; on today's discs
+    /// the online source answers.
     fn get_fmts_indexes(&self, ctx: &dyn ResolveCtx) -> Result<Vec<UnitKey>, libfreemkv::Error> {
-        for s in &self.sources {
-            if let Ok(uks) = s.get_fmts_indexes(ctx)
-                && !uks.is_empty()
-            {
-                return Ok(uks);
-            }
-        }
-        Ok(Vec::new())
+        first_non_empty(&self.sources, |s, c| s.get_fmts_indexes(c), ctx)
     }
 
     /// UNION every inner source's host certs (filtered at the given MKB

src/online.rs

@@ -2,6 +2,7 @@
 
 use std::io::Read;
 use std::net::{IpAddr, SocketAddr, ToSocketAddrs};
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use crate::uks_from_vuk;
@@ -96,6 +97,17 @@ fn is_blocked_ip(ip: &IpAddr) -> bool {
 /// standing misconfiguration that will never fix itself, while `Unreachable` is
 /// the key service being down right now — exactly the condition this module's
 /// callers must not report as "this disc has no key".
+///
+/// SEPARABLE, NOT RANKED. Distinguishing the two is about the *message*, never
+/// about whether the caller hears anything at all. BOTH variants mean the
+/// service was never asked, so BOTH are `Err` out of [`OnlineSource::query`]
+/// (see the match there). The first cut of this enum routed `Config` to
+/// `Ok(Vec::new())` and only `Unreachable` to `Err`, which made a permanent
+/// fault — a mistyped port, an `http://` URL, a host that resolves into RFC1918
+/// — strictly QUIETER than a transient one: one `warn!` and then a benign-looking
+/// empty that `MultiSource::first_non_empty` cannot tell from "this keydb simply
+/// has no row for this disc". That is the seven-hour-502 collapse with the
+/// polarity reversed: the outage at least self-heals, the typo never does.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum GuardFail {
     /// Malformed URL, bad scheme, or a host that is (or resolves to) a
@@ -148,7 +160,20 @@ fn resolve_and_guard(url: &str) -> Result<Vec<SocketAddr>, (GuardFail, String)>
     } else if let Some((h, p)) = authority.rsplit_once(':') {
         match p.parse::<u16>() {
             Ok(p) => (h.to_string(), p),
-            Err(_) => (authority.to_string(), default_port),
+            // A `host:port` authority whose port is not a u16 is a TYPO in the
+            // operator's config — `https://example.com:notaport/keys`. This used
+            // to fall back to the WHOLE authority as the hostname (port text
+            // included), which of course never resolves, so the URL surfaced as
+            // `GuardFail::Unreachable` → `Err(KeyServiceUnavailable)`: a standing
+            // misconfiguration reported as a transient outage, the exact
+            // collapse `GuardFail`'s own doc says must not happen, because the
+            // two halves demand OPPOSITE operator actions (fix the URL vs. wait).
+            // Reject it as `Config`, identical to the bracketed-IPv6 branch a few
+            // lines above, which already returns `Config` for an unparseable
+            // port. Silently substituting `default_port` would be worse still —
+            // it would ship the bearer token and key material to a port the
+            // operator never configured.
+            Err(_) => return Err((GuardFail::Config, "invalid port".into())),
         }
     } else {
         (authority.to_string(), default_port)
@@ -274,6 +299,32 @@ fn hardened_agent(pinned: Vec<SocketAddr>) -> ureq::Agent {
 pub struct OnlineSource {
     base_url: String,
     secret: String,
+    /// The last agent built, with the address set it was pinned to.
+    ///
+    /// A `query` used to build a FRESH `ureq::Agent` every call, and an FMTS disc
+    /// makes two calls per rip (`get_unit_keys` + `get_fmts_indexes`), so it paid
+    /// two TLS handshakes to the same host for no reason — an agent owns the
+    /// connection pool, and a discarded agent discards the pooled, already
+    /// negotiated TLS connection with it.
+    ///
+    /// SECURITY: the agent is reused ONLY when the freshly resolved + guarded
+    /// address SET is identical to the one the cached agent is pinned to. Every
+    /// query still re-resolves and re-runs the SSRF guard — that is the whole
+    /// anti-rebinding defence and is deliberately not cached; what is reused is
+    /// the connection, and only to addresses just re-validated this call.
+    ///
+    /// The stored `Vec<SocketAddr>` is the SET KEY: sorted and deduped, so the
+    /// comparison is order-insensitive. `to_socket_addrs` hands back whatever
+    /// order the resolver felt like, and a round-robin keyserver reorders its
+    /// A/AAAA records on essentially every lookup — an ordered `==` therefore
+    /// missed the cache every single time against exactly the deployment that
+    /// most needs it, silently degrading to the per-query TLS handshake this
+    /// field exists to avoid. Set equality is also the honest predicate: the
+    /// security property being asserted is "these are the same validated
+    /// addresses", which says nothing about their order. The agent itself stays
+    /// pinned to the order it was BUILT with; that is safe because an equal set
+    /// means every address it holds was re-resolved and re-guarded this call.
+    agent: Mutex<Option<(Vec<SocketAddr>, Arc<ureq::Agent>)>>,
 }
 
 impl OnlineSource {
@@ -281,9 +332,33 @@ impl OnlineSource {
         Self {
             base_url: base_url.into(),
             secret: secret.into(),
+            agent: Mutex::new(None),
         }
     }
 
+    /// The agent pinned to `pinned` — the cached one when the address set is
+    /// unchanged, a fresh one otherwise (which then becomes the cached one).
+    /// Poisoning is recovered from: a panic elsewhere must not make every later
+    /// key request panic.
+    fn agent_for(&self, pinned: Vec<SocketAddr>) -> Arc<ureq::Agent> {
+        // Compare SETS, not sequences — see the `agent` field's doc.
+        let key = {
+            let mut k = pinned.clone();
+            k.sort();
+            k.dedup();
+            k
+        };
+        let mut guard = self.agent.lock().unwrap_or_else(|e| e.into_inner());
+        if let Some((cached_key, agent)) = guard.as_ref()
+            && *cached_key == key
+        {
+            return agent.clone();
+        }
+        let agent = Arc::new(hardened_agent(pinned));
+        *guard = Some((key, agent.clone()));
+        agent
+    }
+
     /// The server-resolved Unit Keys for this disc. Runs exactly one network
     /// round-trip. The service returns either a terminal `UK` (used directly) or
     /// a `VUK` (derived to Unit Keys locally via the disc's encrypted title keys
@@ -322,21 +397,44 @@ impl OnlineSource {
         // worst case for cleartext. A self-hosted keyserver on a LAN is already
         // impossible by that guard, whatever its scheme.
         //
-        // Falls through to the next key source (a local keydb) rather than
-        // failing the rip, and says why — misconfiguration should be visible,
-        // not silently indistinguishable from "this disc has no key".
+        // Still falls THROUGH to the next key source (a local keydb) — an `Err`
+        // here does not block the chain, because `MultiSource::first_non_empty`
+        // returns any later source's non-empty key set outright and only
+        // surfaces the failure when nothing anywhere resolved. What it does stop
+        // is the silent case: a cleartext URL is the same permanent operator
+        // fault as the address-guard rejection below (see [`GuardFail`]), so it
+        // gets the same verdict. Returning `Ok(empty)` here made "we refused to
+        // ask" indistinguishable from "the service answered and has no key".
         if !self.base_url.starts_with("https://") {
-            tracing::warn!(
+            tracing::error!(
                 target: "freemkv::keysource",
                 "key-service URL is not https:// — refusing to send key material \
-                 and credentials in cleartext; skipping the online source"
+                 and credentials in cleartext; the service was NOT asked about this disc \
+                 (fix the URL scheme), which is not the same as this disc having no key"
             );
-            return Ok(Vec::new());
+            return Err(Error::KeyServiceUnavailable);
         }
         let mkb = ctx.mkb().unwrap_or(&[]);
-        // An over-cap MKB cannot be forwarded — bound the body. Log it: a silent
-        // empty return here is indistinguishable from "no key", so surface the
-        // real cause (the cap is 64 MiB, far above any real trimmed MKB).
+        // THE LINE this source draws between its skips (written out after the
+        // `Config`-was-quieter-than-`Unreachable` bug), in three parts:
+        //
+        // * NO URL AT ALL is not a fault: the operator did not configure an
+        //   online source, so there is nothing to report. `Ok(empty)`, silently
+        //   (the first guard in this function).
+        // * A bad URL — wrong scheme, mistyped port, guard-blocked address — is
+        //   an OPERATOR FAULT. Every disc is affected, nothing was asked, and
+        //   the fix is to edit the config. That is `Err` (above).
+        // * An over-cap MKB or too few content samples is a property of THIS
+        //   DISC's inputs, not a fault: the request cannot be *formed* for this
+        //   disc, so this source has nothing for it and never will, whatever the
+        //   operator does and however long they wait. Reporting that as
+        //   `KeyServiceUnavailable` would send the operator waiting out an
+        //   outage that does not exist — the same mislabelling in the opposite
+        //   direction. These stay `Ok(empty)`, and stay logged.
+        //
+        // Log it: a silent empty return here is indistinguishable from "no key",
+        // so surface the real cause (the cap is 64 MiB, far above any real
+        // trimmed MKB).
         if mkb.len() > MAX_MKB_BYTES {
             tracing::warn!(
                 target: "freemkv::keysource",
@@ -415,15 +513,35 @@ impl OnlineSource {
                 // static label is enough to separate "misconfigured/blocked
                 // key-service URL" from "this disc has no key", which is the
                 // only distinction the operator needs here.
-                tracing::warn!(
+                //
+                // `error!`, not `warn!`, and `Err`, not `Ok(empty)`. This arm
+                // used to return `Ok(Vec::new())`, which made the PERMANENT
+                // fault quieter than the transient one directly above it: the
+                // service was never asked, yet the composition was handed a
+                // clean empty and — with no other source holding the key — told
+                // the operator `E7022 no key for this disc`. Same lie as the
+                // seven-hour 502, from a config typo that never self-heals. A
+                // failure to ask is never evidence about the disc, so it leaves
+                // as `Err`; `MultiSource::first_non_empty` still lets a later
+                // source's real key win outright, so this cannot fail a rip that
+                // the local keydb could have served.
+                tracing::error!(
                     target: "freemkv::keysource",
                     phase = "keyserver_post",
-                    "key-service URL failed the address guard; skipping the online source"
+                    "key-service URL failed the address guard — the service was NOT asked about this disc; \
+                     this is a standing misconfiguration (fix the URL), not a disc without a key"
                 );
-                return Ok(Vec::new());
+                // NOTE (reported upstream): libfreemkv has no
+                // `KeyServiceMisconfigured` code, so the permanent fault borrows
+                // the transient E7028. E7028's contract — "the source never got
+                // as far as answering the question" — is exactly true here; only
+                // its "transient, retry later" hint is wrong, and the `error!`
+                // above carries the correcting detail until a 70xx config code
+                // exists.
+                return Err(Error::KeyServiceUnavailable);
             }
         };
-        let agent = hardened_agent(pinned);
+        let agent = self.agent_for(pinned);
         let mut req = agent.post(&self.base_url);
         if let Some(value) = bearer_header(&self.secret) {
             req = req.header("Authorization", &value);
@@ -1101,6 +1219,301 @@ mod tests {
         }
     }
 
+    // ── The pre-flight guards in `query` (nothing leaves the process) ───────
+    //
+    // Each of the three guards below fires BEFORE any address is resolved and
+    // before anything is sent. None was exercised: a reordering that let the
+    // cleartext POST through would have shipped green, and that POST carries the
+    // bearer token plus base64 key material.
+    //
+    // Their VERDICTS differ on purpose (see the comment above the MKB cap in
+    // `query`): the cleartext-scheme guard is an operator fault and yields
+    // `Err`, while the two input-shaped guards yield `Ok(empty)` — this source
+    // has nothing for THIS disc, permanently, and calling that an outage would
+    // send the operator waiting for nothing.
+    //
+    // The discriminator in all three: the configured host is `.test`, which
+    // RFC 6761 guarantees never resolves. On the guarded path nothing resolves
+    // it, so the test touches no network and returns `Ok(empty)`. Remove the
+    // guard and control reaches `resolve_and_guard`, which fails as
+    // `Unreachable` → `Err(KeyServiceUnavailable)` — a different Result, so the
+    // mutation cannot pass.
+
+    /// A `ResolveCtx` whose MKB size and sample COUNT are dialled per guard.
+    struct GuardCtx {
+        mkb: Vec<u8>,
+        samples: usize,
+    }
+    impl ResolveCtx for GuardCtx {
+        fn disc_hash(&self) -> &str {
+            "0x422EB"
+        }
+        fn title(&self) -> Option<&str> {
+            None
+        }
+        fn vid(&self) -> Option<libfreemkv::aacs::types::Vid> {
+            None
+        }
+        fn mkb(&self) -> Result<&[u8], Error> {
+            Ok(&self.mkb)
+        }
+        fn enc_title_keys(&self) -> Result<&[[u8; 16]], Error> {
+            Ok(&[])
+        }
+        fn samples(&self, _n: usize) -> Result<Vec<Vec<u8>>, Error> {
+            Ok(vec![vec![0u8; 16]; self.samples])
+        }
+    }
+
+    /// An `http://` key-service URL must never be POSTed to: the body carries
+    /// base64 AACS key material and the header carries a replayable bearer
+    /// token. The source refuses — and, because refusing means the service was
+    /// never asked, says so with `Err` rather than an empty that reads as "this
+    /// disc has no key". A later source's real key still wins
+    /// (`MultiSource::first_non_empty`), so refusing does not fail a rip the
+    /// local keydb can serve.
+    ///
+    /// Catches two mutations: sending the request anyway (the host is `.test`
+    /// and would surface as `Unreachable`, but the assertion below would still
+    /// hold, which is why `too_few_samples_skips_the_request` guards the
+    /// send-path ordering) and — the one this test was rewritten for — returning
+    /// `Ok(Vec::new())`, which made a permanent cleartext misconfiguration
+    /// indistinguishable from a genuine miss.
+    #[test]
+    fn cleartext_http_url_is_refused_before_anything_is_sent() {
+        let src = OnlineSource::new("http://keyserver.test/keys", "s3cr3t");
+        let ctx = GuardCtx {
+            mkb: Vec::new(),
+            // Enough samples that ONLY the scheme guard can stop the request.
+            samples: MIN_SAMPLE_UNITS,
+        };
+        assert_eq!(
+            src.get_unit_keys(&ctx)
+                .expect_err("refusing to ask is a source failure, never a miss")
+                .code(),
+            libfreemkv::error::E_KEY_SERVICE_UNAVAILABLE,
+            "an http:// key-service URL must be refused, not sent — and reported"
+        );
+    }
+
+    /// An MKB larger than the forward cap cannot be sent; the source skips
+    /// rather than truncating the MKB (which would ask about a different disc).
+    #[test]
+    fn over_cap_mkb_skips_the_request() {
+        let src = OnlineSource::new("https://keyserver.test/keys", "s3cr3t");
+        let ctx = GuardCtx {
+            mkb: vec![0u8; MAX_MKB_BYTES + 1],
+            samples: MIN_SAMPLE_UNITS,
+        };
+        assert!(
+            src.get_unit_keys(&ctx)
+                .expect("an un-forwardable MKB is a skip, not a service failure")
+                .is_empty(),
+            "an over-cap MKB must skip the online source"
+        );
+    }
+
+    /// Too few content samples make the service's answer ambiguous (it
+    /// identifies the key by which submitted unit decrypts), so the request is
+    /// never built. Proven at the boundary: `MIN_SAMPLE_UNITS - 1` skips.
+    #[test]
+    fn too_few_samples_skips_the_request() {
+        let src = OnlineSource::new("https://keyserver.test/keys", "s3cr3t");
+        let ctx = GuardCtx {
+            mkb: Vec::new(),
+            samples: MIN_SAMPLE_UNITS - 1,
+        };
+        assert!(
+            src.get_unit_keys(&ctx)
+                .expect("under-sampling is a skip, not a service failure")
+                .is_empty(),
+            "fewer than MIN_SAMPLE_UNITS samples must skip the online source"
+        );
+    }
+
+    // ── MAX_RESPONSE_BYTES: the stated anti-OOM defence ─────────────────────
+
+    /// The reply cap is the crate's only defence against a hostile or broken
+    /// key service driving the client to OOM with an unbounded body, and it had
+    /// never been exercised. Both edges are asserted so the `+1` read cannot
+    /// quietly become a truncating read (which would hand a half-body to the
+    /// JSON parser) or an off-by-one that rejects a legal reply.
+    #[test]
+    fn over_cap_reply_is_rejected_and_an_at_cap_reply_still_parses() {
+        // The over-cap body is deliberately VALID, key-bearing JSON. A body of
+        // junk would be rejected by the JSON parser whether or not the cap
+        // exists, so the test would pass with the cap deleted — worthless. This
+        // one is only rejectable BY the cap: remove the bounded read and it
+        // parses into a key.
+        let head = r#"{"UK":["000102030405060708090a0b0c0d0e0f"],"pad":""#;
+        let tail = r#""}"#;
+        let over = format!(
+            "{head}{}{tail}",
+            "p".repeat(MAX_RESPONSE_BYTES + 1 - head.len() - tail.len())
+        );
+        assert_eq!(over.len(), MAX_RESPONSE_BYTES + 1);
+        assert_eq!(
+            interpret_reply(Ok(reply(200, &over)), &BareCtx, 1)
+                .expect_err("an over-cap body must not be treated as an answer")
+                .code(),
+            libfreemkv::error::E_KEY_SERVICE_UNAVAILABLE,
+        );
+
+        // EXACTLY at the cap, and valid: still read and still answered. Padded
+        // with a JSON string so the length is exact and the key is real.
+        let pad = MAX_RESPONSE_BYTES - head.len() - tail.len();
+        let at_cap = format!("{head}{}{tail}", "p".repeat(pad));
+        assert_eq!(at_cap.len(), MAX_RESPONSE_BYTES);
+        let keys = interpret_reply(Ok(reply(200, &at_cap)), &BareCtx, 1)
+            .expect("a reply exactly at the cap is legal and must be read");
+        assert_eq!(keys.len(), 1, "the key in an at-cap reply must survive");
+    }
+
+    // ── A bad port is CONFIG, not an outage ────────────────────────────────
+
+    /// `https://example.com:notaport/keys` is a typo in the operator's config.
+    /// The unparseable port used to make the WHOLE authority (port text
+    /// included) the hostname, which of course never resolved — so a standing
+    /// misconfiguration was reported as `Unreachable`, i.e. as the key service
+    /// being down. The two demand OPPOSITE actions (fix the URL vs. wait), which
+    /// is precisely what `GuardFail`'s doc says must not be collapsed.
+    #[test]
+    fn unparseable_port_is_a_config_fault_not_an_outage() {
+        for url in [
+            "https://example.com:notaport/keys",
+            "http://example.com:99999/keys", // out of u16 range
+            "https://example.com:/keys",     // empty port
+        ] {
+            assert_eq!(
+                resolve_and_guard(url).expect_err("must be rejected").0,
+                GuardFail::Config,
+                "{url} is a configuration typo, not a service outage"
+            );
+        }
+        // A WELL-FORMED port is still split off the host and honoured.
+        let addrs = resolve_and_guard("https://8.8.8.8:8443/keys").expect("valid port accepted");
+        assert_eq!(addrs[0].port(), 8443);
+    }
+
+    /// The same distinction as seen by a caller — and the CRITICAL half of it:
+    /// separating a config fault from an outage must never make the config fault
+    /// quieter. A mistyped port means the service was never asked, so the caller
+    /// gets `Err`, exactly like an outage; only the log text differs.
+    ///
+    /// Catches the mutation that returns `Ok(Vec::new())` from the
+    /// `GuardFail::Config` arm of `query` (which is what this code did, and is
+    /// how a permanent misconfiguration became an `E7022 no key for this disc`
+    /// with no `Err` for `MultiSource::first_non_empty` to surface).
+    #[test]
+    fn query_with_a_mistyped_port_reports_a_failure_not_a_miss() {
+        let src = OnlineSource::new("https://example.com:notaport/keys", "s3cr3t");
+        let ctx = GuardCtx {
+            mkb: Vec::new(),
+            samples: MIN_SAMPLE_UNITS,
+        };
+        assert_eq!(
+            src.get_unit_keys(&ctx)
+                .expect_err("a config fault means the service was never asked — never Ok(empty)")
+                .code(),
+            libfreemkv::error::E_KEY_SERVICE_UNAVAILABLE,
+        );
+    }
+
+    /// A guard-BLOCKED address (the SSRF/metadata case) travels the same
+    /// `GuardFail::Config` arm and must be just as loud: the POST never left the
+    /// process, so nothing is known about this disc. Distinct from the mistyped
+    /// port above because it fails AFTER resolution, on the address check.
+    ///
+    /// Catches the mutation that makes only the parse-level config faults `Err`
+    /// while letting the resolved-to-private-address case fall back to
+    /// `Ok(empty)`.
+    #[test]
+    fn query_against_a_guard_blocked_address_reports_a_failure_not_a_miss() {
+        // 127.0.0.1 needs no DNS and is unconditionally rejected by is_blocked_ip.
+        let src = OnlineSource::new("https://127.0.0.1/keys", "s3cr3t");
+        let ctx = GuardCtx {
+            mkb: Vec::new(),
+            samples: MIN_SAMPLE_UNITS,
+        };
+        assert_eq!(
+            src.get_unit_keys(&ctx)
+                .expect_err("a blocked address means the service was never asked")
+                .code(),
+            libfreemkv::error::E_KEY_SERVICE_UNAVAILABLE,
+        );
+    }
+
+    // ── One agent per address set, not one per query ────────────────────────
+
+    /// A round-robin keyserver returns the SAME addresses in a different order
+    /// on each lookup. That is the same validated set, so it must reuse the
+    /// agent — an ordered `==` on the cache key missed on every single query
+    /// against exactly the deployment the cache was written for, silently
+    /// reverting to a fresh TLS handshake per request.
+    ///
+    /// Catches the mutation that drops the sort/dedup from `agent_for`'s key
+    /// (the reversed order would then build a second agent) and, via the second
+    /// half, the mutation that "fixes" order-sensitivity by comparing only
+    /// lengths or the first element — which would wrongly reuse an agent pinned
+    /// to a DIFFERENT address.
+    #[test]
+    fn a_reordered_but_identical_address_set_reuses_the_agent() {
+        let src = OnlineSource::new("https://keyserver.test/keys", "");
+        let forward: Vec<SocketAddr> = vec![
+            "8.8.8.8:443".parse().unwrap(),
+            "8.8.4.4:443".parse().unwrap(),
+        ];
+        let reversed: Vec<SocketAddr> = forward.iter().rev().copied().collect();
+        assert_ne!(forward, reversed, "the two orders must really differ");
+
+        let first = src.agent_for(forward);
+        let again = src.agent_for(reversed);
+        assert!(
+            Arc::ptr_eq(&first, &again),
+            "a reordered but identical address SET must reuse the pinned agent"
+        );
+
+        // Same cardinality, one address swapped: a genuinely different set, so
+        // never the same agent.
+        let changed: Vec<SocketAddr> = vec![
+            "8.8.8.8:443".parse().unwrap(),
+            "1.1.1.1:443".parse().unwrap(),
+        ];
+        assert!(
+            !Arc::ptr_eq(&first, &src.agent_for(changed)),
+            "a different address set must never reuse an agent pinned elsewhere"
+        );
+    }
+
+    /// An FMTS disc calls `query` twice per rip (`get_unit_keys` +
+    /// `get_fmts_indexes`), and a fresh `ureq::Agent` per call throws away the
+    /// connection pool with it — a second full TLS handshake to the same host
+    /// for nothing. The agent is reused only while the freshly guarded address
+    /// set is IDENTICAL, so the anti-rebinding guarantee is untouched: a
+    /// different address set builds (and re-pins) a new agent.
+    #[test]
+    fn the_agent_is_reused_per_address_set_only() {
+        let src = OnlineSource::new("https://keyserver.test/keys", "");
+        let a: Vec<SocketAddr> = vec!["8.8.8.8:443".parse().unwrap()];
+        let b: Vec<SocketAddr> = vec!["1.1.1.1:443".parse().unwrap()];
+
+        let first = src.agent_for(a.clone());
+        let second = src.agent_for(a.clone());
+        assert!(
+            Arc::ptr_eq(&first, &second),
+            "the same pinned address set must reuse the agent (and its pooled TLS connection)"
+        );
+
+        let other = src.agent_for(b);
+        assert!(
+            !Arc::ptr_eq(&first, &other),
+            "a DIFFERENT address set must never reuse an agent pinned elsewhere"
+        );
+        // And the address set is re-pinned, so going back re-builds.
+        let back = src.agent_for(a);
+        assert!(!Arc::ptr_eq(&first, &back));
+    }
+
     /// Finding #9 regression: parse_uk must reject any non-hex byte up front so
     /// sign prefixes / whitespace can't slip through the windowed 2-char parse
     /// (`u8::from_str_radix` accepts "+5", "-A", etc.).

tests/key_sources.rs

@@ -362,6 +362,147 @@ fn multi_source_real_keydb_resolves_through_chain() {
     );
 }
 
+// ── MultiSource: a source that could not ANSWER is not a source that said no ─
+//
+// `MultiSource` re-collapsed the Ok/Err distinction `OnlineSource` was
+// engineered to draw: `if let Ok(uks) = s.get_unit_keys(ctx)` discarded every
+// `Err`, so a composition whose sources ALL failed returned `Ok(Vec::new())` —
+// which the resolver reports as `E7022 No key source has a decryption key for
+// this disc`. That is the seven-hour-502 incident reproduced one layer up.
+
+/// A source that always fails, with a chosen error — the key-service outage.
+struct FailingSource {
+    err: fn() -> libfreemkv::Error,
+    label: &'static str,
+}
+
+impl KeySource for FailingSource {
+    fn get_unit_keys(&self, _ctx: &dyn ResolveCtx) -> Result<Vec<UnitKey>, libfreemkv::Error> {
+        Err((self.err)())
+    }
+    fn get_fmts_indexes(&self, _ctx: &dyn ResolveCtx) -> Result<Vec<UnitKey>, libfreemkv::Error> {
+        Err((self.err)())
+    }
+    fn label(&self) -> &'static str {
+        self.label
+    }
+}
+
+fn unavailable() -> libfreemkv::Error {
+    libfreemkv::Error::KeyServiceUnavailable
+}
+
+fn unauthorized() -> libfreemkv::Error {
+    libfreemkv::Error::KeyServiceUnauthorized
+}
+
+/// THE regression, one layer up from `interpret_reply`: when no source holds a
+/// key AND a source could not answer, the composition must report the FAILURE,
+/// never a clean "no key". Catches the mutation that restores
+/// `if let Ok(uks) = ..` (which drops the `Err` and returns `Ok(empty)`).
+#[test]
+fn multi_source_reports_a_source_failure_instead_of_a_clean_no_key() {
+    let multi = MultiSource::new(vec![
+        Box::new(FailingSource {
+            err: unavailable,
+            label: "down",
+        }),
+        Box::new(ScriptedSource::new("empty", vec![])),
+    ]);
+    let err = multi
+        .get_unit_keys(&DiscInputsCtx::new(&inputs("x")))
+        .expect_err("a failed source must not collapse into a clean no-key");
+    assert_eq!(
+        err.code(),
+        libfreemkv::error::E_KEY_SERVICE_UNAVAILABLE,
+        "the composition must surface the service outage"
+    );
+    assert_ne!(
+        err.code(),
+        libfreemkv::error::E_NO_DISC_KEY,
+        "an outage must never be reported as 'this disc has no key'"
+    );
+}
+
+/// The forensic half carried the identical bug and is now driven through the
+/// same helper — pinned separately so a future edit cannot fix one and leave
+/// the other.
+#[test]
+fn multi_source_fmts_indexes_also_preserve_a_source_failure() {
+    let multi = MultiSource::new(vec![Box::new(FailingSource {
+        err: unavailable,
+        label: "down",
+    })]);
+    assert_eq!(
+        multi
+            .get_fmts_indexes(&DiscInputsCtx::new(&inputs("x")))
+            .expect_err("the forensic path must preserve the failure too")
+            .code(),
+        libfreemkv::error::E_KEY_SERVICE_UNAVAILABLE,
+    );
+}
+
+/// A source failure must NOT block the chain: a later source that actually has
+/// the key still wins outright. Pins the other half of the contract, so the fix
+/// cannot be over-applied into "any error aborts the composition".
+#[test]
+fn multi_source_failure_never_blocks_a_later_success() {
+    let multi = MultiSource::new(vec![
+        Box::new(FailingSource {
+            err: unavailable,
+            label: "down",
+        }),
+        Box::new(ScriptedSource::new("real", vec![uk(0xc1)])),
+    ]);
+    assert_eq!(
+        resolve(&multi, &inputs("x")),
+        vec![uk(0xc1)],
+        "a failing first source must not stop the second from resolving"
+    );
+}
+
+/// FIRST failure wins, matching libfreemkv's `source_failure` rule: the
+/// most-preferred source's reason is the one the operator is told about, in the
+/// same order the caller expressed a preference for successes.
+#[test]
+fn multi_source_reports_the_first_failure_not_the_last() {
+    let multi = MultiSource::new(vec![
+        Box::new(FailingSource {
+            err: unauthorized,
+            label: "bad-token",
+        }),
+        Box::new(FailingSource {
+            err: unavailable,
+            label: "down",
+        }),
+    ]);
+    assert_eq!(
+        multi
+            .get_unit_keys(&DiscInputsCtx::new(&inputs("x")))
+            .expect_err("both sources failed")
+            .code(),
+        libfreemkv::error::E_KEY_SERVICE_UNAUTHORIZED,
+        "the first source's failure is the one reported"
+    );
+}
+
+/// The contrast case: every source ANSWERED and none holds a key. That is the
+/// genuine miss, and `E7022` is then the truth — so the fix must not turn an
+/// honest empty into an error.
+#[test]
+fn multi_source_all_sources_answering_empty_is_still_a_clean_no_key() {
+    let multi = MultiSource::new(vec![
+        Box::new(ScriptedSource::new("a", vec![])),
+        Box::new(ScriptedSource::new("b", vec![])),
+    ]);
+    assert!(
+        multi
+            .get_unit_keys(&DiscInputsCtx::new(&inputs("x")))
+            .expect("every source answered; this is a genuine miss")
+            .is_empty()
+    );
+}
+
 /// The `Path`-typed constructor accepts a borrowed path.
 #[test]
 fn keydb_source_accepts_borrowed_path() {