test: use .test not .invalid for the unresolvable key-service hosts

The round-2 guard tests used `keyserver.invalid` — RFC 6761 reserved and
guaranteed non-resolvable, which is the correct property. But the shared
leak-guard net lists `.invalid` among the private-infra TLD suffixes
(.internal/.local/.lan/.corp), so every such host tripped it and turned the
repo red on push — the exact cost the precommit script's own comment records
from 1.6.3. `.test` is equally RFC-6761-reserved and equally non-resolvable,
carries the same test guarantee, and is not in the infra net. No behaviour
change; the tests assert the identical never-resolves path.

.github/dependabot.yml

@@ -0,0 +1,63 @@
+version: 2
+
+# Dependency updates land on `dev`, never on `main`.
+#
+# `main` here is a RELEASE POINTER that release.sh moves to each tag. A bot
+# commit on it would put work there that no tag contains, which is exactly the
+# state that aborted the 1.6.2 cascade at the last step -- so pointing
+# Dependabot at main would recreate that failure on a schedule.
+updates:
+  - package-ecosystem: cargo
+    directory: /
+    target-branch: dev
+    schedule:
+      interval: weekly
+    open-pull-requests-limit: 5
+    # One PR per week for the routine bumps instead of one per crate. Eight
+    # repos times a handful of crates is a volume nobody reads, and an
+    # unread PR queue is indistinguishable from no updates at all.
+    groups:
+      minor-and-patch:
+        update-types:
+          - minor
+          - patch
+    ignore:
+      # The freemkv crates depend on each other by GIT TAG, re-pinned by
+      # release.sh as part of the release commit. Dependabot cannot see that
+      # cascade, so a PR bumping one of these would fight the release process
+      # and could pin a version whose tag does not exist yet.
+      - dependency-name: freemkv-unlock
+      - dependency-name: libfreemkv
+      - dependency-name: freemkv-keysources
+      - dependency-name: freemkv-i18n
+      - dependency-name: freemkv-engine
+
+  # The workflows are now real infrastructure -- the release cascade, the
+  # cross-platform hash matrix, the disc gate -- so their actions need the same
+  # attention as the crates.
+  - package-ecosystem: github-actions
+    directory: /
+    target-branch: dev
+    schedule:
+      interval: weekly
+    open-pull-requests-limit: 5
+    groups:
+      actions:
+        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

.github/workflows/ci.yml

@@ -41,15 +41,15 @@ jobs:
   lint:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           path: freemkv-keysources
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           repository: freemkv/libfreemkv
           ref: "${{ github.ref_name == 'qa' && 'qa' || 'dev' }}"
           path: libfreemkv
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           repository: freemkv/freemkv-unlock
           ref: "${{ github.ref_name == 'qa' && 'qa' || 'dev' }}"
@@ -78,15 +78,15 @@ jobs:
   test:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           path: freemkv-keysources
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           repository: freemkv/libfreemkv
           ref: "${{ github.ref_name == 'qa' && 'qa' || 'dev' }}"
           path: libfreemkv
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           repository: freemkv/freemkv-unlock
           ref: "${{ github.ref_name == 'qa' && 'qa' || 'dev' }}"
@@ -111,17 +111,30 @@ jobs:
   # libfreemkv's platform code (SCSI transports included). Build the tests, not
   # `cargo check` — check skips codegen and never compiles test-only code.
   check-macos:
+    # dev is the FAST lane: this job still runs, but on the release-candidate
+    # branches rather than on every push to dev. Nothing is deleted and no
+    # platform stops being checked before a release -- qa.yml independently
+    # covers macOS and Windows, and the jobs unique to this file (the Intel
+    # macOS build, the Windows release build) run here on qa and main. A push
+    # to dev is meant to be cheap and frequent; waiting on three runner pools
+    # to agree is what a release candidate is for.
+    #
+    # `if` SKIPS the job (it does not queue). A queued job would be far worse
+    # than a slow one: release.sh's CI gate refuses while any run for the
+    # commit is still in progress, so a never-scheduled job blocks releases
+    # silently -- see the note on real-media in qa.yml.
+    if: github.ref_name == 'qa' || github.ref_name == 'main'
     runs-on: macos-latest
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           path: freemkv-keysources
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           repository: freemkv/libfreemkv
           ref: "${{ github.ref_name == 'qa' && 'qa' || 'dev' }}"
           path: libfreemkv
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           repository: freemkv/freemkv-unlock
           ref: "${{ github.ref_name == 'qa' && 'qa' || 'dev' }}"
@@ -143,17 +156,30 @@ jobs:
         working-directory: freemkv-keysources
 
   check-windows:
+    # dev is the FAST lane: this job still runs, but on the release-candidate
+    # branches rather than on every push to dev. Nothing is deleted and no
+    # platform stops being checked before a release -- qa.yml independently
+    # covers macOS and Windows, and the jobs unique to this file (the Intel
+    # macOS build, the Windows release build) run here on qa and main. A push
+    # to dev is meant to be cheap and frequent; waiting on three runner pools
+    # to agree is what a release candidate is for.
+    #
+    # `if` SKIPS the job (it does not queue). A queued job would be far worse
+    # than a slow one: release.sh's CI gate refuses while any run for the
+    # commit is still in progress, so a never-scheduled job blocks releases
+    # silently -- see the note on real-media in qa.yml.
+    if: github.ref_name == 'qa' || github.ref_name == 'main'
     runs-on: windows-latest
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           path: freemkv-keysources
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           repository: freemkv/libfreemkv
           ref: "${{ github.ref_name == 'qa' && 'qa' || 'dev' }}"
           path: libfreemkv
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           repository: freemkv/freemkv-unlock
           ref: "${{ github.ref_name == 'qa' && 'qa' || 'dev' }}"

.github/workflows/leak-guard.yml

@@ -11,7 +11,7 @@ jobs:
   leak-guard:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           fetch-depth: 0
       - name: Compute commit range

.github/workflows/qa.yml

@@ -43,7 +43,7 @@ jobs:
     permissions:
       contents: write
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           fetch-depth: 0
       - name: Stamp the next rc
@@ -71,15 +71,15 @@ jobs:
         os: ['ubuntu-latest', 'macos-latest']
     runs-on: ${{ matrix.os }}
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           path: freemkv-keysources
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           repository: freemkv/libfreemkv
           ref: qa
           path: libfreemkv
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           repository: freemkv/freemkv-unlock
           ref: qa
@@ -106,15 +106,15 @@ jobs:
   cross-lint:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           path: freemkv-keysources
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           repository: freemkv/libfreemkv
           ref: qa
           path: libfreemkv
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           repository: freemkv/freemkv-unlock
           ref: qa
@@ -145,15 +145,15 @@ jobs:
   windows-build:
     runs-on: windows-latest
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           path: freemkv-keysources
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           repository: freemkv/libfreemkv
           ref: qa
           path: libfreemkv
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
         with:
           repository: freemkv/freemkv-unlock
           ref: qa

.github/workflows/release.yml

@@ -17,7 +17,7 @@ jobs:
   verify:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
       - name: Verify Cargo.toml version matches tag
         run: |
           CARGO_VER="v$(grep '^version' Cargo.toml | head -1 | sed 's/.*"\(.*\)"/\1/')"
@@ -40,7 +40,7 @@ jobs:
     needs: verify
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
       - uses: dtolnay/rust-toolchain@1.97.0
       - uses: Swatinem/rust-cache@v2
       - run: cargo test
@@ -58,8 +58,8 @@ jobs:
     needs: verify
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v7
       - name: Create GitHub Release
-        uses: softprops/action-gh-release@v2
+        uses: softprops/action-gh-release@v3
         with:
           generate_release_notes: true

CHANGELOG.md

@@ -1,5 +1,86 @@
 # 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
+
+- **The key service is reached over a newer HTTP client, with the same
+  protections.** Before contacting a key service this crate resolves the host,
+  rejects any private or link-local address, and then pins the connection to the
+  addresses it just checked, so a hostname cannot be re-pointed at an internal
+  machine in the moment between the check and the request. That behaviour is
+  unchanged; only the client underneath it moved.
+
+### Security
+
+- **The address pinning is now proven by a test rather than assumed.** The checks
+  that reject a bad address were well covered, but nothing verified that the
+  approved addresses were the ones actually connected to — and every existing
+  test still passed with that wiring removed entirely, because none of them
+  opened a connection. A connection made to an unreachable name can only
+  succeed if the pinned address is honoured, and that is now asserted. No
+  defect is being fixed here: the pinning was correct. It simply had no way of
+  telling anyone if it stopped being correct.
+
+## [1.6.2] — 2026-08-08
+
+Version sync with the workspace. No functional change in this crate.
+
 ## [1.6.1] — 2026-08-07
 
 Version sync with the workspace. No functional change in this crate.

CONTRIBUTING.md

@@ -0,0 +1,18 @@
+# Contributing
+
+Thanks for your interest!
+
+- **Report a bug** — open an issue
+- **Fix a bug** — fork, branch, PR
+- **Suggest a feature** — open an issue first
+
+## Development
+
+```bash
+cargo build
+cargo test
+```
+
+## License
+
+MIT

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "freemkv-keysources"
-version = "1.6.1"
+version = "1.6.4"
 edition = "2024"
 rust-version = "1.97"
 license = "MIT"
@@ -19,13 +19,13 @@ publish = false
 # bare version req to the libfreemkv git tag (local dev overrides it to a path).
 libfreemkv = "1.6"
 # OnlineSource: POST disc inputs + samples to a key service over HTTP.
-ureq = { version = "2", features = ["json"] }
+ureq = { version = "3", features = ["json"] }
 serde_json = "1"
-base64 = "0.22"
+base64 = "0.23"
 # KeydbSource::save: decompress a downloaded keydb (.zip / .gz). Same versions
 # already in the resolved graph via libfreemkv — no new tree, no new HTTP stack.
 flate2 = "1"
-zip = { version = "2", default-features = false, features = ["deflate"] }
+zip = { version = "8", default-features = false, features = ["deflate"] }
 # Structural begin/end logging around the keyserver round-trip (never logs
 # key material).
 tracing = "0.1"
@@ -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.1" }
+libfreemkv = { git = "https://github.com/freemkv/libfreemkv", tag = "v1.6.4" }

SECURITY.md

@@ -0,0 +1,22 @@
+# Security Policy
+
+## Supported versions
+
+| Version | Supported |
+| ------- | --------- |
+| 1.6.x   | Yes       |
+| < 1.6   | No        |
+
+Only the current 1.6.x line receives security fixes.
+
+## Reporting a vulnerability
+
+Report vulnerabilities privately through GitHub Security Advisories:
+https://github.com/freemkv/freemkv-keysources/security/advisories/new
+
+Do not open a public issue for a security report. Include the affected
+version, steps to reproduce, and the impact you believe the issue has.
+
+## Response time
+
+You will get an initial response within 7 days.

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/keydb_format.rs

@@ -9,13 +9,19 @@
 // The parser is copied verbatim, so it carries the full KeyDb/DiscEntry API
 // even though this crate's consumer (`keydb.rs`) only exercises a subset
 // (`load`, `find_disc`, `iter_disc_entries`, and the public fields read by
-// `candidates_from`/`host_certs`). The unused items — `empty`, `find_vuk`,
-// `DiscEntry::{title, vid}` — are part of the faithful copy and are
-// retained rather than pruned; allow dead_code so the byte-for-byte copy
+// `KeydbSource::unit_keys_from` / `KeydbSource::host_certs`). NOTE: an earlier
+// revision of this comment cited a `candidates_from` function — no such
+// function has ever existed in this crate; the real consumer is
+// `KeydbSource::unit_keys_from` (keydb.rs), which reads `unit_keys`, `vuk`,
+// `media_key` AND `vid` (`vid` is load-bearing for the MK+VID derivation the
+// kat_c / kat_d KATs pin — it is NOT unused). The genuinely unused items —
+// `empty`, `find_vuk`, `DiscEntry::title` — are part of the faithful copy and
+// are retained rather than pruned; allow dead_code so the byte-for-byte copy
 // compiles clean without diverging from the libfreemkv original.
 #![allow(dead_code)]
 
 use std::collections::HashMap;
+use std::sync::Arc;
 
 use libfreemkv::aacs::types::{DeviceKey, HostCert};
 
@@ -34,7 +40,10 @@ const MAX_KEYDB_BYTES: u64 = 128 * 1024 * 1024;
 const MAX_DISC_ENTRIES: usize = 500_000;
 
 /// Parsed AACS key database.
-#[derive(Debug)]
+///
+/// NO `#[derive(Debug)]`: see the hand-written redacting impl below. The
+/// derive dumped every processing key and all ~181k discs' key material into
+/// any `{:?}`.
 pub struct KeyDb {
     /// Device keys for MKB processing
     pub device_keys: Vec<DeviceKey>,
@@ -44,8 +53,55 @@ pub struct KeyDb {
     /// keydb's revocation metadata (libfreemkv's `HostCert` stays pure; the
     /// `Revoked in MKBv<N>` annotation is tracked in this crate).
     pub host_certs: Vec<KeydbHostCert>,
-    /// Per-disc VUK entries indexed by disc hash (hex lowercase)
-    pub disc_entries: HashMap<String, DiscEntry>,
+    /// Per-disc VUK entries indexed by disc hash (hex lowercase).
+    ///
+    /// The key is `Arc<str>` and is the SAME allocation as the entry's own
+    /// [`DiscEntry::disc_hash`] — the map used to own a second `String` clone
+    /// of every hash, ~13 MB of pure duplication across the real keydb's 181k
+    /// entries (and now that [`crate::KeydbSource`] caches the parsed db for
+    /// the process lifetime, that duplication is resident, not transient).
+    /// `Arc<str>: Borrow<str>`, so lookups still take a plain `&str`.
+    pub disc_entries: HashMap<Arc<str>, DiscEntry>,
+}
+
+// Key material must never reach a log, a panic message, or a bug report. Both
+// types below carry raw AACS key bytes, and both are public API, so a
+// `#[derive(Debug)]` on either is a leak of the crown jewels: `{:?}` on a
+// loaded `KeyDb` printed every processing key plus every disc's Media Key,
+// VID, VUK and Unit Keys. libfreemkv's equivalent types (`aacs::types`:
+// `DeviceKey`, `HostCert`, `Vid`, `MediaKey`, `Vuk`, `ProcessingKey`,
+// `UnitKey`, its own `DiscEntry`) all hand-write a redacting `Debug` for
+// exactly this reason; this crate's copy of the parser dropped that
+// convention when it was relocated. These restore it, field-for-field in the
+// same style (`<redacted>` for key bytes, a `_len` for key-bearing
+// collections). Pinned by `keydb_debug_is_redacted` /
+// `disc_entry_debug_is_redacted`.
+
+impl std::fmt::Debug for KeyDb {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("KeyDb")
+            .field("device_keys_len", &self.device_keys.len())
+            .field("processing_keys_len", &self.processing_keys.len())
+            .field("host_certs_len", &self.host_certs.len())
+            .field("disc_entries_len", &self.disc_entries.len())
+            .finish()
+    }
+}
+
+impl std::fmt::Debug for DiscEntry {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("DiscEntry")
+            .field("disc_hash", &self.disc_hash)
+            .field("title", &self.title)
+            .field("media_key", &self.media_key.map(|_| "<redacted>"))
+            .field("vid", &self.vid.map(|_| "<redacted>"))
+            .field("vuk", &self.vuk.map(|_| "<redacted>"))
+            .field("unit_keys_len", &self.unit_keys.len())
+            .field("mkb_version", &self.mkb_version)
+            .field("volume_size", &self.volume_size)
+            .field("is_uhd", &self.is_uhd)
+            .finish()
+    }
 }
 
 /// A keydb host certificate together with its revocation generation.
@@ -65,10 +121,14 @@ pub struct KeydbHostCert {
 }
 
 /// A per-disc entry from the key database.
-#[derive(Debug, Clone)]
+///
+/// NO `#[derive(Debug)]`: see the hand-written redacting impl above.
+#[derive(Clone)]
 pub struct DiscEntry {
-    /// Disc hash (20 bytes, hex)
-    pub disc_hash: String,
+    /// Disc hash (20 bytes, hex). `Arc<str>` so the entry and the
+    /// [`KeyDb::disc_entries`] key that points at it share ONE allocation
+    /// instead of two.
+    pub disc_hash: Arc<str>,
     /// Disc title
     pub title: String,
     /// Media Key (16 bytes) — from MKB processing
@@ -145,6 +205,92 @@ pub(crate) fn parse_hex20(s: &str) -> Option<[u8; 20]> {
     libfreemkv::hex::parse_hex_fixed::<20>(s)
 }
 
+/// What [`KeyDb::parse`] threw away, by reason.
+///
+/// The parser is fed a THIRD-PARTY file: a mirror can serve a truncated or
+/// byte-shifted download that is still valid UTF-8, still contains enough
+/// recognisable rows to pass `KeydbSource::save`'s "at least one entry" check,
+/// and is then atomically committed as the new keydb — after which every disc
+/// past the corruption point silently resolves nothing. Before this counter,
+/// each malformed `| DK` / `| PK` / `| HC` / `| HC2` / `0x…` row was dropped by
+/// an `if let Some(..) = .. { }` with no `else`, no counter and no log, so that
+/// file was indistinguishable from a healthy one at every layer above.
+///
+/// The counts are logged ONCE per parse (a per-line log would be 181k lines on
+/// a real keydb, which is why the rate limit is "one summary", never "hide the
+/// count").
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct ParseStats {
+    /// `| DK` rows that parsed as neither a positioned nor an orphan device key.
+    pub dk_rejected: usize,
+    /// `| PK` rows whose key field did not yield 16 bytes.
+    pub pk_rejected: usize,
+    /// `| HC` rows rejected (bad hex, short cert, missing field).
+    pub hc_rejected: usize,
+    /// `| HC2` rows rejected (bad hex, wrong priv length, short cert).
+    pub hc2_rejected: usize,
+    /// `0x… = …` disc rows the field parser refused outright.
+    pub disc_rejected: usize,
+    /// Disc rows dropped because [`MAX_DISC_ENTRIES`] was already reached.
+    pub disc_over_cap: usize,
+    /// Disc rows that REPLACED an earlier row with the same hash (last wins).
+    /// Not necessarily corruption, but never something to discover silently:
+    /// a duplicated hash means one of the two rows' keys is now unreachable.
+    pub disc_duplicate: usize,
+}
+
+impl ParseStats {
+    /// Total rows the parser refused. Excludes duplicates, which were parsed
+    /// successfully and merely overwrote a sibling.
+    fn rejected(&self) -> usize {
+        self.dk_rejected
+            + self.pk_rejected
+            + self.hc_rejected
+            + self.hc2_rejected
+            + self.disc_rejected
+            + self.disc_over_cap
+    }
+
+    /// Emit the one summary line, if there is anything to say. Returns whether
+    /// a line was emitted — the only way a test can observe the warning without
+    /// a `tracing` subscriber dependency, and the seam
+    /// [`crate::KeydbSource::cached_db`] counts to prove the warning survives
+    /// the cache.
+    ///
+    /// `pub(crate)` because [`crate::KeydbSource`] re-emits it on a CACHE HIT
+    /// too: the cache short-circuits `parse`, so without that the corruption
+    /// summary for a given file fired once per process and a long-running daemon
+    /// went on serving the damaged keydb in silence forever after.
+    pub(crate) fn log(&self) -> bool {
+        if self.rejected() == 0 && self.disc_duplicate == 0 {
+            return false;
+        }
+        tracing::warn!(
+            target: "freemkv::keysource",
+            dk_rejected = self.dk_rejected,
+            pk_rejected = self.pk_rejected,
+            hc_rejected = self.hc_rejected,
+            hc2_rejected = self.hc2_rejected,
+            disc_rejected = self.disc_rejected,
+            disc_over_cap = self.disc_over_cap,
+            disc_duplicate = self.disc_duplicate,
+            "keydb.cfg lines were rejected while parsing; the file may be truncated or corrupt (keys past the damage will not resolve)"
+        );
+        true
+    }
+}
+
+/// True when `line` opens a per-disc row. The `0x` prefix is matched
+/// CASE-INSENSITIVELY: `parse_hex` in this same file deliberately tolerates
+/// `0x` and `0X` (and is tested for it), so a case-sensitive `starts_with("0x")`
+/// here would drop an uppercase `0X…` disc row on the floor with no diagnostic
+/// while every other hex field in the format accepted it. Shared with
+/// `KeydbSource::save`, whose entry count must stay in EXACT lockstep with what
+/// this parser accepts.
+pub(crate) fn is_disc_entry_line(line: &str) -> bool {
+    (line.starts_with("0x") || line.starts_with("0X")) && line.contains(" = ")
+}
+
 impl KeyDb {
     /// Construct an empty KeyDb. Used by unit tests; production code
     /// reaches a populated KeyDb via [`KeyDb::load`] or [`KeyDb::parse`].
@@ -158,7 +304,20 @@ impl KeyDb {
     }
 
     /// Parse a KEYDB.cfg file from a string.
+    ///
+    /// Lines the parser cannot use are counted and summarised in ONE
+    /// `tracing::warn!` (see [`ParseStats`]) — they are never dropped silently.
     pub fn parse(data: &str) -> Self {
+        let (db, stats) = Self::parse_counted(data);
+        stats.log();
+        db
+    }
+
+    /// [`Self::parse`] without the log — the seam the rejection-counting tests
+    /// assert on (a `tracing` subscriber would be the only other way to observe
+    /// a count, and the count is the thing under test, not its formatting).
+    pub(crate) fn parse_counted(data: &str) -> (Self, ParseStats) {
+        let mut stats = ParseStats::default();
         let mut db = KeyDb {
             device_keys: Vec::new(),
             processing_keys: Vec::new(),
@@ -187,6 +346,8 @@ impl KeyDb {
                     db.device_keys.push(dk);
                 } else if let Some(key) = Self::parse_orphan_dk(line) {
                     db.processing_keys.push(key);
+                } else {
+                    stats.dk_rejected += 1;
                 }
                 continue;
             }
@@ -195,6 +356,8 @@ impl KeyDb {
             if line.starts_with("| PK") {
                 if let Some(pk) = Self::parse_processing_key(line) {
                     db.processing_keys.push(pk);
+                } else {
+                    stats.pk_rejected += 1;
                 }
                 continue;
             }
@@ -230,6 +393,8 @@ impl KeyDb {
                             revoked_at_mkb,
                         });
                     }
+                } else {
+                    stats.hc2_rejected += 1;
                 }
                 continue;
             }
@@ -238,22 +403,39 @@ impl KeyDb {
             if line.starts_with("| HC") {
                 if let Some(hc) = Self::parse_host_cert(line) {
                     db.host_certs.push(hc);
+                } else {
+                    stats.hc_rejected += 1;
                 }
                 continue;
             }
 
-            // Disc entry: starts with 0x
-            if line.starts_with("0x") && line.contains(" = ") {
+            // Disc entry: starts with 0x / 0X (see `is_disc_entry_line`).
+            if is_disc_entry_line(line) {
+                // The cap is a memory bound against a pathological file, but
+                // hitting it means real discs are being dropped — count it, the
+                // way `load`'s size/UTF-8 caps produce a diagnostic.
                 if db.disc_entries.len() >= MAX_DISC_ENTRIES {
+                    stats.disc_over_cap += 1;
                     continue;
                 }
-                if let Some(entry) = Self::parse_disc_entry(line) {
-                    db.disc_entries.insert(entry.disc_hash.clone(), entry);
+                match Self::parse_disc_entry(line) {
+                    // The map key IS the entry's own `disc_hash` allocation
+                    // (`Arc` clone, no second copy of the string).
+                    Some(entry) => {
+                        if db
+                            .disc_entries
+                            .insert(entry.disc_hash.clone(), entry)
+                            .is_some()
+                        {
+                            stats.disc_duplicate += 1;
+                        }
+                    }
+                    None => stats.disc_rejected += 1,
                 }
             }
         }
 
-        db
+        (db, stats)
     }
 
     /// Load a KEYDB.cfg from disk.
@@ -265,6 +447,27 @@ impl KeyDb {
     /// [`KeyDb`] rather than an error — callers needing a non-empty db must
     /// check the parsed contents.
     pub fn load(path: &std::path::Path) -> std::io::Result<Self> {
+        let (db, stats) = Self::load_counted(std::fs::File::open(path)?, path)?;
+        stats.log();
+        Ok(db)
+    }
+
+    /// [`Self::load`] from an ALREADY-OPEN file, without the log.
+    ///
+    /// Two callers need this seam, both in [`crate::KeydbSource`]:
+    ///
+    /// * The identity stamp and the bytes must come from the SAME descriptor.
+    ///   Stat-the-path-then-load-the-path is two independent resolutions of one
+    ///   name, so an atomic rename landing between them stored a stamp
+    ///   describing one file next to the parsed contents of another — and the
+    ///   cache then served that pairing until the stamp changed again. `fstat`
+    ///   on the open handle cannot be pointed at a different inode.
+    /// * The rejection counts have to be RETAINED, not just logged, so a cache
+    ///   hit can re-emit them.
+    pub(crate) fn load_counted(
+        f: std::fs::File,
+        path: &std::path::Path,
+    ) -> std::io::Result<(Self, ParseStats)> {
         // Read through a hard `take` cap rather than trusting a stat. A
         // `metadata` pre-check is only advisory: it is skipped entirely when the
         // stat fails, and it reports len 0 for a FIFO or character device, so
@@ -274,7 +477,6 @@ impl KeyDb {
         // exactly-at-cap file and rejects anything larger, matching
         // `read_capped_to_string` on the download path.
         use std::io::Read;
-        let f = std::fs::File::open(path)?;
         let mut buf = Vec::new();
         f.take(MAX_KEYDB_BYTES + 1).read_to_end(&mut buf)?;
         if buf.len() as u64 > MAX_KEYDB_BYTES {
@@ -292,7 +494,7 @@ impl KeyDb {
                 format!("keydb.cfg is not valid UTF-8: {}", path.display()),
             )
         })?;
-        Ok(Self::parse(&data))
+        Ok(Self::parse_counted(&data))
     }
 
     /// Look up a disc by its hash. Returns the VUK if found.
@@ -307,8 +509,8 @@ impl KeyDb {
         // prefix and the no-prefix fallback is currently unreachable; it is
         // retained as a defensive match for the prefix-agnostic lookup contract.
         self.disc_entries
-            .get(&format!("0x{hash}"))
-            .or_else(|| self.disc_entries.get(&hash))
+            .get(format!("0x{hash}").as_str())
+            .or_else(|| self.disc_entries.get(hash.as_str()))
             .and_then(|e| e.vuk)
     }
 
@@ -323,8 +525,8 @@ impl KeyDb {
         // key carries the "0x" prefix, see find_vuk); kept as a defensive
         // match for the prefix-agnostic lookup contract.
         self.disc_entries
-            .get(&format!("0x{hash}"))
-            .or_else(|| self.disc_entries.get(&hash))
+            .get(format!("0x{hash}").as_str())
+            .or_else(|| self.disc_entries.get(hash.as_str()))
     }
 
     /// Iterate every disc entry. Used by Path 3 (scan for matching VID).
@@ -373,7 +575,7 @@ impl KeyDb {
         self.disc_entries
             .values()
             .filter(|e| !e.unit_keys.is_empty())
-            .map(|e| (e.disc_hash.clone(), e.unit_keys.clone()))
+            .map(|e| (e.disc_hash.to_string(), e.unit_keys.clone()))
             .collect()
     }
 
@@ -445,10 +647,10 @@ impl KeyDb {
         }
 
         // Per-disc entries, sorted by hash for a deterministic, diff-friendly file.
-        let mut hashes: Vec<&String> = self.disc_entries.keys().collect();
+        let mut hashes: Vec<&Arc<str>> = self.disc_entries.keys().collect();
         hashes.sort();
         for h in hashes {
-            let d = &self.disc_entries[h];
+            let d = &self.disc_entries[&**h];
             // `parse` keeps the `hash_part` verbatim, so the stored `disc_hash`
             // already carries its `0x` prefix — emit it as-is (prefixing another
             // `0x` would double it on re-parse).
@@ -634,7 +836,7 @@ impl KeyDb {
     fn parse_disc_entry(line: &str) -> Option<DiscEntry> {
         // 0x<hash> = <title> | D | <date> | M | 0x<mk> | I | 0x<id> | V | 0x<vuk> | U | <unit_keys> ; <comment>
         let (hash_part, rest) = line.split_once(" = ")?;
-        let disc_hash = hash_part.trim().to_lowercase();
+        let disc_hash: Arc<str> = Arc::from(hash_part.trim().to_lowercase());
 
         // The trailing `;` comment (e.g.
         // "; MKBv76/BEE/FindVUK 1.74 - VolumeSize: 81309007872 (UHD)") carries
@@ -853,13 +1055,18 @@ mod tests {
     /// is NOT expected — but once normalized, a re-load+re-serialize must be
     /// stable. Idempotence here proves `parse` is lossless on its own output and
     /// `to_keydb_cfg` is deterministic. Also asserts no rows are dropped.
-    /// Skipped unless `KEYDB_PATH` points at a real keydb.cfg.
+    /// REQUIRES a real keydb; `#[ignore]` by default. Run with:
+    /// `KEYDB_PATH=/path/to/keydb.cfg cargo test --release -- --ignored`.
+    ///
+    /// It used to carry no `#[ignore]` and `return` early when `KEYDB_PATH` was
+    /// unset — so CI ran it, asserted NOTHING about 181k-entry behaviour, and
+    /// reported green. A skip has to be visible in the test output to mean
+    /// anything, and an ignored test that is RUN without its fixture must fail
+    /// loudly rather than silently pass.
     #[test]
+    #[ignore = "needs a real keydb.cfg: KEYDB_PATH=<path> cargo test -- --ignored"]
     fn to_keydb_cfg_is_idempotent_on_real_keydb() {
-        let path = match keydb_path() {
-            Some(p) => p,
-            None => return,
-        };
+        let path = real_keydb_path();
         let db1 = KeyDb::load(&path).unwrap();
         let s1 = db1.to_keydb_cfg();
         let db2 = KeyDb::parse(&s1);
@@ -881,12 +1088,54 @@ mod tests {
         assert_eq!(db1.host_certs.len(), db2.host_certs.len(), "HC drift");
     }
 
+    /// REAL-SCALE parser health: a genuine published keydb must parse with ZERO
+    /// rejected rows. This is the assertion the rejection counter exists to
+    /// make possible — a corrupt or byte-shifted mirror download differs from a
+    /// healthy one precisely here, and nothing above the parser can tell them
+    /// apart. `#[ignore]` by default; run with:
+    /// `KEYDB_PATH=/path/to/keydb.cfg cargo test --release -- --ignored`.
+    #[test]
+    #[ignore = "needs a real keydb.cfg: KEYDB_PATH=<path> cargo test -- --ignored"]
+    fn a_real_keydb_parses_with_no_rejected_rows() {
+        let path = real_keydb_path();
+        let text = std::fs::read_to_string(&path).expect("keydb must be readable");
+        let (db, stats) = KeyDb::parse_counted(&text);
+        assert!(
+            db.disc_entries.len() > 170_000,
+            "expected a full-size keydb, got {} entries",
+            db.disc_entries.len()
+        );
+        assert_eq!(
+            stats.rejected(),
+            0,
+            "a healthy keydb must not lose rows to the parser: {stats:?}"
+        );
+        assert_eq!(
+            stats.disc_duplicate, 0,
+            "duplicate disc hashes make one row's keys unreachable: {stats:?}"
+        );
+    }
+
     /// Get KEYDB path from KEYDB_PATH environment variable. Returns None if not set or not found.
     fn keydb_path() -> Option<std::path::PathBuf> {
         let path = std::path::PathBuf::from(std::env::var("KEYDB_PATH").ok()?);
         if path.exists() { Some(path) } else { None }
     }
 
+    /// The real keydb for the `#[ignore]`d real-scale tests, or a LOUD failure.
+    ///
+    /// The old pattern (`match keydb_path() { Some(p) => p, None => return }`)
+    /// on a non-ignored test is the worst of both worlds: CI ran the test,
+    /// asserted nothing, and reported success. These tests are `#[ignore]`d, so
+    /// running one at all is an explicit request — and an explicit request with
+    /// no fixture must fail, not quietly pass.
+    fn real_keydb_path() -> std::path::PathBuf {
+        keydb_path().expect(
+            "KEYDB_PATH must point at a real keydb.cfg; run: \
+             KEYDB_PATH=/path/to/keydb.cfg cargo test --release -- --ignored",
+        )
+    }
+
     #[test]
     fn test_parse_disc_entry() {
         // All-zero placeholders — synthetic; no real key material in code.
@@ -1033,12 +1282,12 @@ mod tests {
         assert!(KeyDb::parse_host_cert(&line).is_none());
     }
 
+    /// REQUIRES a real keydb; `#[ignore]` by default. Run with:
+    /// `KEYDB_PATH=/path/to/keydb.cfg cargo test --release -- --ignored`.
     #[test]
+    #[ignore = "needs a real keydb.cfg: KEYDB_PATH=<path> cargo test -- --ignored"]
     fn test_parse_full_keydb() {
-        let path = match keydb_path() {
-            Some(p) => p,
-            None => return,
-        }; // skip if not available
+        let path = real_keydb_path();
 
         let db = KeyDb::load(&path).unwrap();
 
@@ -1065,6 +1314,349 @@ mod tests {
         );
     }
 
+    /// `load_counted` must read the HANDLE it was given, never re-resolve the
+    /// path. `KeydbSource::cached_db` stamps the open file with `fstat` and then
+    /// parses from that same descriptor precisely so the recorded identity and
+    /// the parsed bytes cannot describe two different files; the moment this
+    /// function re-opens the path, that guarantee is gone and an atomic rename
+    /// landing in between stores one file's stamp beside another file's keys —
+    /// a mismatch the cache then serves as if it were valid.
+    ///
+    /// The race itself cannot be scheduled deterministically in-process, so the
+    /// SEAM is what is pinned: with the path already replaced, the handle's
+    /// contents must still win. Catches the mutation that restores an internal
+    /// `File::open(path)`.
+    #[cfg(unix)]
+    #[test]
+    fn load_counted_reads_the_handle_not_the_path() {
+        let dir = std::env::temp_dir().join(format!(
+            "fmk-keysources-load-seam-{}-{:?}",
+            std::process::id(),
+            std::thread::current().id()
+        ));
+        std::fs::create_dir_all(&dir).unwrap();
+        let path = dir.join("keydb.cfg");
+        let other = dir.join("replacement.cfg");
+        let row = |k: &str| format!("0xaabb = T | U | 1-0x{k}\n");
+        std::fs::write(&path, row(&"01".repeat(16))).unwrap();
+        std::fs::write(&other, row(&"02".repeat(16))).unwrap();
+
+        let handle = std::fs::File::open(&path).unwrap();
+        // The path now names a DIFFERENT inode; the handle still names the old
+        // one, exactly as it would mid-race.
+        std::fs::rename(&other, &path).unwrap();
+
+        let (db, _) = KeyDb::load_counted(handle, &path).expect("the open handle is still valid");
+        assert_eq!(
+            db.disc_entries
+                .values()
+                .next()
+                .expect("one entry")
+                .unit_keys[0]
+                .1,
+            [0x01u8; 16],
+            "the parse must come from the handle that was stamped, not from a re-opened path"
+        );
+
+        std::fs::remove_dir_all(&dir).ok();
+    }
+
+    // ── Debug redaction: key material must never reach a log or a panic ────
+
+    /// The only key bytes in these fixtures are the byte 0x5A repeated; the
+    /// assertion is that NO run of them survives into the formatted string, in
+    /// either the hex or the Rust-array rendering a derived `Debug` produces.
+    fn assert_no_key_bytes(what: &str, rendered: &str) {
+        assert!(
+            !rendered.contains("90"),
+            "{what} Debug leaked a decimal key byte: {rendered}"
+        );
+        assert!(
+            !rendered.contains("5a") && !rendered.contains("5A"),
+            "{what} Debug leaked a hex key byte: {rendered}"
+        );
+        // A derived `Debug` over `[u8; 16]` / `Vec<[u8; 16]>` renders a byte
+        // array — `[90, 90, ...]`. A redacting impl never emits one. Matched as
+        // "a `[` immediately followed by a digit" rather than "any `[`", because
+        // callers legitimately render a `Vec` OF redacting structs
+        // (`{:?}` on `KeyDb::host_certs`), whose own `[` opens a list of
+        // `Name { .. }` — never a byte. Every byte-array rendering, decimal or
+        // nested, still starts `[<digit>` (or `[[<digit>`), and the two byte
+        // checks above catch the payload independently.
+        let bytes = rendered.as_bytes();
+        assert!(
+            !rendered
+                .match_indices('[')
+                .any(|(i, _)| bytes.get(i + 1).is_some_and(u8::is_ascii_digit)),
+            "{what} Debug rendered a raw byte array: {rendered}"
+        );
+    }
+
+    /// `{:?}` on a `DiscEntry` must not print the Media Key, VID, VUK or any
+    /// Unit Key. Catches the mutation that restores `#[derive(Debug)]`: the
+    /// derive renders `media_key: Some([90, 90, ...])`, so a single `{:?}` in a
+    /// caller's log or panic message published a disc's whole key set.
+    #[test]
+    fn disc_entry_debug_is_redacted() {
+        let e = DiscEntry {
+            disc_hash: Arc::from("0xaabb"),
+            title: "T".into(),
+            media_key: Some([0x5A; 16]),
+            vid: Some([0x5A; 16]),
+            vuk: Some([0x5A; 16]),
+            unit_keys: vec![(1, [0x5A; 16])],
+            mkb_version: Some(76),
+            volume_size: Some(1),
+            is_uhd: true,
+        };
+        let rendered = format!("{e:?}");
+        assert_no_key_bytes("DiscEntry", &rendered);
+        // The non-secret identifying fields are still useful and still printed.
+        assert!(rendered.contains("0xaabb"), "{rendered}");
+        assert!(rendered.contains("unit_keys_len"), "{rendered}");
+    }
+
+    /// `{:?}` on a whole `KeyDb` must print COUNTS only. The derive dumped every
+    /// processing key plus all ~181k disc entries' key material — the single
+    /// worst leak in the crate, since a `KeyDb` is what a key source holds.
+    #[test]
+    fn keydb_debug_is_redacted() {
+        let mut db = KeyDb::empty();
+        db.processing_keys.push([0x5A; 16]);
+        db.device_keys.push(DeviceKey {
+            key: [0x5A; 16],
+            node: 1,
+            uv: 2,
+            u_mask_shift: 3,
+        });
+        db.host_certs.push(KeydbHostCert {
+            cert: HostCert {
+                private_key: [0x5A; 20],
+                certificate: vec![0x5A; 92],
+                private_key_v2: None,
+                certificate_v2: None,
+            },
+            revoked_at_mkb: None,
+        });
+        let e = DiscEntry {
+            disc_hash: Arc::from("0xaabb"),
+            title: "T".into(),
+            media_key: Some([0x5A; 16]),
+            vid: None,
+            vuk: Some([0x5A; 16]),
+            unit_keys: vec![(1, [0x5A; 16])],
+            mkb_version: None,
+            volume_size: None,
+            is_uhd: false,
+        };
+        db.disc_entries.insert(e.disc_hash.clone(), e);
+
+        let rendered = format!("{db:?}");
+        assert_no_key_bytes("KeyDb", &rendered);
+        assert!(
+            rendered.contains("disc_entries_len"),
+            "counts, not contents: {rendered}"
+        );
+        assert!(
+            rendered.contains("processing_keys_len"),
+            "counts, not contents: {rendered}"
+        );
+    }
+
+    /// The SIBLING type. `KeyDb`'s hand-written `Debug` prints
+    /// `host_certs_len`, so it protects the whole-db rendering and nothing else
+    /// — [`KeyDb::host_certs`] is a public field, and `{:?}` on ONE element of
+    /// it walks a completely different code path: `KeydbHostCert`'s derive,
+    /// then `HostCert`'s impl. That path carries the AACS host PRIVATE key, the
+    /// only key material in the crate that authenticates US rather than opening
+    /// a disc.
+    ///
+    /// It is safe today only because `libfreemkv::HostCert` hand-writes a
+    /// redacting `Debug` (`aacs::types`, `private_key: "<redacted>"` +
+    /// `certificate_len`) — an EXTERNAL guarantee this crate silently inherits
+    /// through the derive, with nothing on this side pinning it. This test pins
+    /// it: it catches a libfreemkv upgrade that swaps that impl for
+    /// `#[derive(Debug)]`, which would leak the host private key out of this
+    /// crate's public API with no diff in this repository at all.
+    ///
+    /// Rendered three ways because callers reach the cert three ways: the whole
+    /// `Vec` (`?db.host_certs`), one element, and the inner `HostCert` a caller
+    /// gets back from `KeyDb::host_certs(mkb)`.
+    #[test]
+    fn host_cert_debug_is_redacted_including_the_sibling_wrapper() {
+        let hc = KeydbHostCert {
+            cert: HostCert {
+                private_key: [0x5A; 20],
+                certificate: vec![0x5A; 92],
+                private_key_v2: Some([0x5A; 32]),
+                certificate_v2: Some(vec![0x5A; 132]),
+            },
+            revoked_at_mkb: Some(76),
+        };
+        assert_no_key_bytes("KeydbHostCert", &format!("{hc:?}"));
+        assert_no_key_bytes("Vec<KeydbHostCert>", &format!("{:?}", vec![hc.clone()]));
+        assert_no_key_bytes("HostCert", &format!("{:?}", hc.cert));
+        // Positive half: the private keys must be present-but-masked, not
+        // omitted. A `Debug` that simply dropped the fields would satisfy the
+        // negative assertions above while hiding that a key is held at all.
+        assert_eq!(
+            format!("{hc:?}").matches("<redacted>").count(),
+            2,
+            "both host private keys (1.0 and 2.0) must render as <redacted>: {hc:?}"
+        );
+        // The non-secret revocation field is what the wrapper exists for and is
+        // still printed — redaction must not become "print nothing useful".
+        assert!(
+            format!("{hc:?}").contains("76"),
+            "the revocation generation is not secret and must survive: {hc:?}"
+        );
+    }
+
+    /// The `Vec<DeviceKey>` on the other public `KeyDb` field, for the same
+    /// reason: `device_keys` is reachable element-wise and each entry holds a
+    /// 16-byte device key. Also inherited from libfreemkv's redacting impl.
+    #[test]
+    fn device_key_debug_is_redacted() {
+        let dk = DeviceKey {
+            key: [0x5A; 16],
+            node: 1,
+            uv: 2,
+            u_mask_shift: 3,
+        };
+        assert_no_key_bytes("DeviceKey", &format!("{dk:?}"));
+        assert_no_key_bytes("Vec<DeviceKey>", &format!("{:?}", vec![dk]));
+    }
+
+    // ── Parser rejection accounting ────────────────────────────────────────
+
+    /// Every malformed row the parser drops must be COUNTED, per reason. The
+    /// mutation this catches: restoring the bare `if let Some(x) = .. { }` with
+    /// no `else`, which is how a byte-shifted mirror download passed `save()`'s
+    /// "at least one entry" validation, was atomically committed as the new
+    /// keydb, and then resolved nothing for every disc past the damage — with
+    /// not one line of log to notice it by.
+    #[test]
+    fn parse_counts_every_rejected_row_by_reason() {
+        let good_pk = format!("| PK | 0x{}", "ab".repeat(16));
+        let cfg = format!(
+            "{good_pk}\n\
+             | DK | DEVICE_KEY 0xZZ | DEVICE_NODE 0x0800 | KEY_UV 0x00000400 | KEY_U_MASK_SHIFT 0x17\n\
+             | PK | 0xnothex\n\
+             | HC | HOST_PRIV_KEY 0x00 | HOST_CERT 0x00\n\
+             | HC2 | HOST_PRIV_KEY 0x00 | HOST_CERT 0x00\n\
+             0xdeadbeef and no equals sign here\n"
+        );
+        let (db, stats) = KeyDb::parse_counted(&cfg);
+        assert_eq!(stats.dk_rejected, 1, "the bad DK row must be counted");
+        assert_eq!(stats.pk_rejected, 1, "the bad PK row must be counted");
+        assert_eq!(stats.hc_rejected, 1, "the short HC row must be counted");
+        assert_eq!(stats.hc2_rejected, 1, "the short HC2 row must be counted");
+        assert_eq!(stats.rejected(), 4, "four rows rejected in total");
+        // The one GOOD row still loaded — counting must not change parsing.
+        assert_eq!(db.processing_keys, vec![[0xABu8; 16]]);
+        // A `0x` line with no " = " is not a disc row at all, so it is not a
+        // rejection (it is a comment-shaped line the format allows).
+        assert_eq!(stats.disc_rejected, 0);
+    }
+
+    /// A duplicated disc hash silently last-wins through `HashMap::insert`:
+    /// one of the two rows' keys becomes unreachable. Not necessarily
+    /// corruption, but never something to discover silently.
+    #[test]
+    fn parse_counts_duplicate_disc_hashes() {
+        let k1 = "01".repeat(16);
+        let k2 = "02".repeat(16);
+        let cfg = format!("0xAAAA = T | U | 1-0x{k1}\n0xaaaa = T | U | 1-0x{k2}\n");
+        let (db, stats) = KeyDb::parse_counted(&cfg);
+        assert_eq!(db.disc_entries.len(), 1, "same hash ⇒ one entry survives");
+        assert_eq!(
+            stats.disc_duplicate, 1,
+            "the overwritten row must be counted, not vanish"
+        );
+        // Last wins — pinned so the counter documents the ACTUAL behaviour.
+        assert_eq!(db.get_uk("0xaaaa"), vec![(1, [0x02u8; 16])]);
+    }
+
+    /// Hitting `MAX_DISC_ENTRIES` drops real discs, so it must produce a
+    /// diagnostic like `load`'s size / UTF-8 caps do — not a bare `continue`.
+    /// Driven through the cap constant itself so it cannot rot if the cap moves.
+    #[test]
+    fn parse_counts_disc_rows_dropped_over_the_entry_cap() {
+        let k = "01".repeat(16);
+        let mut cfg = String::with_capacity((MAX_DISC_ENTRIES + 2) * 16);
+        for i in 0..MAX_DISC_ENTRIES + 2 {
+            cfg.push_str(&format!("0x{i:x} = T | U | 1-0x{k}\n"));
+        }
+        let (db, stats) = KeyDb::parse_counted(&cfg);
+        assert_eq!(
+            db.disc_entries.len(),
+            MAX_DISC_ENTRIES,
+            "cap still enforced"
+        );
+        assert_eq!(
+            stats.disc_over_cap, 2,
+            "the two rows dropped by the cap must be counted"
+        );
+    }
+
+    /// A disc row rejected by the FIELD parser is counted separately from one
+    /// dropped by the cap.
+    #[test]
+    fn parse_counts_unparseable_disc_row() {
+        // ` = ` present (so it IS a disc row) but `split_once(" = ")` succeeds
+        // and the entry parses — to get a genuine rejection the row must fail
+        // `parse_disc_entry`, which only returns None without a " = ". Nothing
+        // in today's grammar can be a disc row AND fail, so the counter is
+        // proven by construction elsewhere; assert the healthy case is zero so
+        // a future parser change that starts rejecting rows cannot do it
+        // silently.
+        let k = "01".repeat(16);
+        let (_, stats) = KeyDb::parse_counted(&format!("0xAAAA = T | U | 1-0x{k}\n"));
+        assert_eq!(stats.rejected(), 0, "a healthy keydb rejects nothing");
+        assert_eq!(stats.disc_duplicate, 0);
+    }
+
+    // ── `0X` disc rows / the save() lockstep ────────────────────────────────
+
+    /// The disc-row test is `starts_with("0x")` — CASE-SENSITIVE, while
+    /// `parse_hex` in this same file deliberately accepts `0x` AND `0X` (and is
+    /// tested for it). An uppercase `0X…` row was therefore dropped with no
+    /// diagnostic at all: not parsed, not counted, not logged. Catches the
+    /// mutation that removes the `0X` arm of `is_disc_entry_line`.
+    #[test]
+    fn disc_row_with_uppercase_0x_prefix_is_parsed() {
+        let v = "33".repeat(16);
+        let db = KeyDb::parse(&format!("0XABCDEF = T | V | 0x{v}\n"));
+        assert_eq!(
+            db.disc_entries.len(),
+            1,
+            "an uppercase 0X disc row must parse, like every other hex field"
+        );
+        // The hash is lowercased on the way in, so lookup is unaffected.
+        assert_eq!(db.find_vuk("0xabcdef"), Some([0x33u8; 16]));
+        assert!(is_disc_entry_line("0XAA = T"));
+        assert!(is_disc_entry_line("0xaa = T"));
+        // Still not a disc row without the " = " separator.
+        assert!(!is_disc_entry_line("0XAABBCC"));
+    }
+
+    // ── One allocation per disc hash, not two ──────────────────────────────
+
+    /// The map key and the entry's own `disc_hash` must be the SAME allocation.
+    /// They used to be two `String`s (`insert(entry.disc_hash.clone(), entry)`),
+    /// ~13 MB of duplicate hash text across the real keydb's 181k entries — now
+    /// resident for the process lifetime, since `KeydbSource` caches the parse.
+    #[test]
+    fn disc_hash_is_stored_once_not_twice() {
+        let v = "33".repeat(16);
+        let db = KeyDb::parse(&format!("0xABCDEF = T | V | 0x{v}\n"));
+        let (key, entry) = db.disc_entries.iter().next().expect("one entry");
+        assert!(
+            Arc::ptr_eq(key, &entry.disc_hash),
+            "the map key must BE the entry's disc_hash allocation, not a clone"
+        );
+    }
+
     // ════════════════════════════════════════════════════════════════════
     // Hardening additions
     // ════════════════════════════════════════════════════════════════════
@@ -1131,7 +1723,7 @@ mod tests {
         let z32 = "00".repeat(16);
         let line = format!("0xABCDEF = T | M | 0x{z32}");
         let e = KeyDb::parse_disc_entry(&line).unwrap();
-        assert_eq!(e.disc_hash, "0xabcdef");
+        assert_eq!(&*e.disc_hash, "0xabcdef");
     }
 
     #[test]
@@ -1476,14 +2068,14 @@ mod tests {
     // the env is unset; they must still COMPILE.
     // ════════════════════════════════════════════════════════════════════
 
+    /// REQUIRES a real keydb; `#[ignore]` by default. Run with:
+    /// `KEYDB_PATH=/path/to/keydb.cfg cargo test --release -- --ignored`.
     #[test]
+    #[ignore = "needs a real keydb.cfg: KEYDB_PATH=<path> cargo test -- --ignored"]
     fn test_vuk_derivation() {
         // Pick any UHD entry with a known MK, VID, and VUK from KEYDB.
         // VUK = AES-DEC(MK, VID) XOR VID
-        let path = match keydb_path() {
-            Some(p) => p,
-            None => return,
-        };
+        let path = real_keydb_path();
 
         let db = KeyDb::load(&path).unwrap();
 
@@ -1507,20 +2099,26 @@ mod tests {
         eprintln!("VUK derivation verified for: {}", entry.title);
     }
 
+    /// REQUIRES a real keydb AND an encrypted unit dump; `#[ignore]` by default.
+    /// Run with: `KEYDB_PATH=<keydb.cfg> ENCRYPTED_UNIT_PATH=<unit.bin> \
+    /// cargo test --release -- --ignored`.
     #[test]
+    #[ignore = "needs KEYDB_PATH + ENCRYPTED_UNIT_PATH: cargo test -- --ignored"]
     fn test_decrypt_real_unit() {
         // Try decrypting a real encrypted aligned unit from a UHD sample.
         // This disc is AACS 2.0 (BEE) so unit key alone won't work —
         // we need bus decryption first. But this verifies the pipeline.
         // Path comes from ENCRYPTED_UNIT_PATH (same env-driven pattern as the
         // KEYDB_PATH fixture); no-ops in CI when unset.
-        let unit_path = match std::env::var("ENCRYPTED_UNIT_PATH").ok() {
-            Some(p) => std::path::PathBuf::from(p),
-            None => return,
-        };
-        if !unit_path.exists() {
-            return;
-        }
+        let unit_path = std::path::PathBuf::from(
+            std::env::var("ENCRYPTED_UNIT_PATH")
+                .expect("ENCRYPTED_UNIT_PATH must point at an encrypted aligned unit"),
+        );
+        assert!(
+            unit_path.exists(),
+            "ENCRYPTED_UNIT_PATH does not exist: {}",
+            unit_path.display()
+        );
 
         let original = std::fs::read(&unit_path).unwrap();
         assert_eq!(original.len(), libfreemkv::aacs::content::ALIGNED_UNIT_LEN);
@@ -1529,10 +2127,7 @@ mod tests {
             "Unit should be encrypted"
         );
 
-        let kp = match keydb_path() {
-            Some(p) => p,
-            None => return,
-        };
+        let kp = real_keydb_path();
         let db = KeyDb::load(&kp).unwrap();
 
         // Candidate entries: any UHD entry that carries unit keys.
@@ -1543,13 +2138,24 @@ mod tests {
             .collect();
 
         eprintln!("Found {} entries with unit keys", candidate_entries.len());
+        // Without candidates nothing below runs and the test would "pass" having
+        // exercised nothing — a parser regression that dropped every unit key
+        // would look exactly like the expected AACS 2.0 outcome.
+        assert!(
+            !candidate_entries.is_empty(),
+            "the fixture keydb must carry at least one entry with unit keys"
+        );
 
         // Try each entry's unit keys: apply the key, then ask whether it opened
         // the unit (the segregated primitives — decrypt, then structural check).
+        let mut attempts = 0usize;
+        let mut any_key_changed_the_unit = false;
         for entry in &candidate_entries {
             for (_, key) in &entry.unit_keys {
                 let mut unit = original.clone();
                 libfreemkv::aacs::content::decrypt_unit(&mut unit, key);
+                attempts += 1;
+                any_key_changed_the_unit |= unit != original;
                 if libfreemkv::aacs::content::is_clean(&unit, libfreemkv::disc::ContentFormat::BdTs)
                 {
                     eprintln!("SUCCESS: Decrypted with entry {}", entry.disc_hash);
@@ -1561,29 +2167,49 @@ mod tests {
             }
         }
 
-        // Expected: none work because this is AACS 2.0 and needs bus decryption first
-        eprintln!("No unit key worked (expected for AACS 2.0 BEE disc — needs read_data_key)");
+        // Reaching here is the EXPECTED outcome for the AACS 2.0 (BEE) sample
+        // this test was written against: the unit is still bus-encrypted, so no
+        // unit key alone can open it. That makes "no key worked" an unusable
+        // assertion — which is exactly why this test used to end in an
+        // `eprintln!` and fall off the end, i.e. pass unconditionally. A no-op
+        // `decrypt_unit`, an `is_clean` stuck at `false`, or a parser that
+        // yielded zero keys would all have produced this same clean exit.
+        //
+        // What CAN be asserted regardless of the sample: the pipeline actually
+        // ran, and the decrypt primitive actually transformed the bytes.
+        assert!(attempts > 0, "no unit key was tried at all");
+        assert!(
+            any_key_changed_the_unit,
+            "decrypt_unit left the aligned unit byte-identical for all {attempts} keys — \
+             the decrypt primitive is a no-op, not a failed key"
+        );
+        eprintln!(
+            "No unit key worked over {attempts} attempts \
+             (expected for AACS 2.0 BEE disc — needs read_data_key)"
+        );
     }
 
+    /// REQUIRES a real keydb; `#[ignore]` by default. Run with:
+    /// `KEYDB_PATH=/path/to/keydb.cfg cargo test --release -- --ignored`.
     #[test]
+    #[ignore = "needs a real keydb.cfg: KEYDB_PATH=<path> cargo test -- --ignored"]
     fn test_resolve_keys_vuk_path() {
         // Test the full resolve chain using VUK path
-        let path = match keydb_path() {
-            Some(p) => p,
-            None => return,
-        };
+        let path = real_keydb_path();
         let db = KeyDb::load(&path).unwrap();
 
         // Find any BD entry that carries a VUK and unit keys, then exercise
         // the lookup-by-hash + VUK-derivation chain against it.
+        // `expect`, not an early `return`: returning here asserted NOTHING while
+        // reporting success, so a parser regression that stopped populating
+        // `vid`/`vuk` — the very fields this test exists to exercise — would make
+        // the `find` miss and ship green. If the fixture genuinely has no such
+        // entry, that is a broken fixture and must be loud.
         let entry = db
             .disc_entries
             .values()
-            .find(|e| e.vuk.is_some() && !e.unit_keys.is_empty() && e.vid.is_some());
-        if entry.is_none() {
-            return;
-        }
-        let entry = entry.unwrap();
+            .find(|e| e.vuk.is_some() && !e.unit_keys.is_empty() && e.vid.is_some())
+            .expect("the fixture keydb must carry an entry with VUK + unit keys + VID");
         let vuk = entry.vuk.unwrap();
         let vid = entry.vid.unwrap();
         let hash_hex = format!("0x{}", libfreemkv::hex::strip_hex_prefix(&entry.disc_hash));

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;
@@ -9,6 +10,10 @@ use base64::Engine;
 use libfreemkv::aacs::types::UnitKey;
 use libfreemkv::keysource::{DecodeSampleSet, ResolveCtx};
 use libfreemkv::{Error, KeySource};
+use ureq::config::Config;
+use ureq::http::Uri;
+use ureq::unversioned::resolver::{ResolvedSocketAddrs, Resolver};
+use ureq::unversioned::transport::{DefaultConnector, NextTimeout};
 
 // Upper bound on the MKB forwarded to the key service — kept in lockstep with
 // libfreemkv's `read_mkb_content` MAX_BYTES (64 MiB) so an MKB the library is
@@ -92,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
@@ -144,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)
@@ -217,21 +246,85 @@ pub fn validate_keyserver_url(url: &str) -> Result<(), String> {
     resolve_and_guard(url).map(|_| ()).map_err(|(_, msg)| msg)
 }
 
+/// ureq's `ResolvedSocketAddrs` is a fixed 16-slot array and its `push` writes
+/// straight into that array, so handing it a 17th address is an out-of-bounds
+/// panic — in the rip thread, on a host that merely publishes a lot of A
+/// records. Keep the first 16; every one of them was validated by
+/// [`resolve_and_guard`], so a subset is still safe, just less redundant.
+const MAX_PINNED_ADDRS: usize = 16;
+
+/// The pinned-address resolver behind [`hardened_agent`].
+///
+/// ureq 3 replaced v2's resolver closure with this trait. The agent MUST be
+/// built through `Agent::with_parts` to take one: `Agent::new_with_config`
+/// silently keeps the default resolver, which would send the request to live
+/// DNS and quietly reopen the rebinding window this module exists to close.
+/// That failure is invisible — it has no symptom short of an actual attack —
+/// so it is pinned by `hardened_agent_connects_to_the_pinned_address_not_dns`.
+#[derive(Debug)]
+struct PinnedResolver(Vec<SocketAddr>);
+
+impl Resolver for PinnedResolver {
+    fn resolve(
+        &self,
+        _uri: &Uri,
+        _config: &Config,
+        _timeout: NextTimeout,
+    ) -> Result<ResolvedSocketAddrs, ureq::Error> {
+        let mut out = self.empty();
+        for addr in self.0.iter().take(MAX_PINNED_ADDRS) {
+            out.push(*addr);
+        }
+        if out.is_empty() {
+            // The trait's contract: at least one address, or this error.
+            return Err(ureq::Error::HostNotFound);
+        }
+        Ok(out)
+    }
+}
+
 /// Build a ureq agent that follows zero redirects (so a public URL can't
 /// 30x-redirect to an internal host) and pins DNS resolution to `pinned`
 /// (the addresses already validated by [`resolve_and_guard`]).
 fn hardened_agent(pinned: Vec<SocketAddr>) -> ureq::Agent {
-    ureq::AgentBuilder::new()
-        .redirects(0)
-        .timeout_connect(Duration::from_secs(10))
-        .timeout_read(Duration::from_secs(TIMEOUT_SECS))
-        .resolver(move |_netloc: &str| Ok(pinned.clone()))
-        .build()
+    let config = Config::builder()
+        .max_redirects(0)
+        .timeout_connect(Some(Duration::from_secs(10)))
+        .timeout_recv_response(Some(Duration::from_secs(TIMEOUT_SECS)))
+        .build();
+    // `with_parts`, never `new_with_config` — see [`PinnedResolver`].
+    ureq::Agent::with_parts(config, DefaultConnector::new(), PinnedResolver(pinned))
 }
 
 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 {
@@ -239,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
@@ -280,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",
@@ -373,18 +513,38 @@ 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.set("Authorization", &value);
+            req = req.header("Authorization", &value);
         }
         // Begin/end around the keyserver round-trip — a slow or unresponsive
         // service is the suspected DVD-scan hang. The agent is built with a
@@ -433,11 +593,11 @@ fn classify_http_status(code: u16) -> Error {
 /// body carries base64 AACS key material and the reply carries keys — neither is
 /// ever logged, and nor is a serde parse error (it quotes the offending input).
 fn interpret_reply(
-    sent: Result<ureq::Response, ureq::Error>,
+    sent: Result<ureq::http::Response<ureq::Body>, ureq::Error>,
     ctx: &dyn ResolveCtx,
     elapsed_ms: u64,
 ) -> Result<Vec<UnitKey>, Error> {
-    let resp = match sent {
+    let mut resp = match sent {
         Ok(r) => r,
         Err(e) => {
             // Record the HTTP status when the service actually answered:
@@ -452,7 +612,7 @@ fn interpret_reply(
             // Each arm now RETURNS the classified error instead of an empty
             // vec, so the distinction survives past this function.
             return Err(match e {
-                ureq::Error::Status(code, _) => {
+                ureq::Error::StatusCode(code) => {
                     let err = classify_http_status(code);
                     tracing::warn!(
                         target: "freemkv::keysource",
@@ -464,7 +624,14 @@ fn interpret_reply(
                     );
                     err
                 }
-                ureq::Error::Transport(_) => {
+                // ureq 3 fans v2's single `Transport` variant out into `Io`,
+                // `Timeout`, `Tls`, `Protocol`, `HostNotFound` and more, and the
+                // enum is non_exhaustive. Every one of them means the same thing
+                // here — nothing answered, so nothing is known about this disc —
+                // and a catch-all is the only shape that stays correct as the
+                // enum grows. The status arm above is the sole case where the
+                // service actually replied.
+                _ => {
                     tracing::warn!(
                         target: "freemkv::keysource",
                         phase = "keyserver_post",
@@ -486,7 +653,8 @@ fn interpret_reply(
     // Reading MAX_RESPONSE_BYTES+1 lets us detect (and reject) an over-cap body.
     let mut buf = Vec::new();
     if resp
-        .into_reader()
+        .body_mut()
+        .as_reader()
         .take(MAX_RESPONSE_BYTES as u64 + 1)
         .read_to_end(&mut buf)
         .is_err()
@@ -752,6 +920,80 @@ mod tests {
         assert_eq!(addrs[0].port(), 8080);
     }
 
+    // ── the pin is actually consulted ──────────────────────────────────────
+    //
+    // `resolve_and_guard` VALIDATES addresses; `hardened_agent` PINS them, so a
+    // hostname cannot re-resolve to an internal address between validation and
+    // connection. The validation half is well covered above — but every one of
+    // those tests still passes if the pinned resolver is never consulted,
+    // because none of them makes a connection. Nothing else in this file proves
+    // the agent honours the pin, and a mis-wired resolver fails OPEN: the
+    // connection quietly falls back to live DNS and the rebinding window this
+    // module exists to close is reopened, with no visible symptom.
+    //
+    // The discriminator: pin the agent to a loopback listener this test owns,
+    // then ask it for a host that CANNOT resolve — `.test` is reserved by
+    // RFC 6761 and never resolves in the public DNS. Only a consulted resolver
+    // can turn that
+    // name into a connection; a fallback to live DNS fails instead. No network
+    // is touched, and `query` (whose guard blocks loopback by design) is not
+    // involved — this drives `hardened_agent` directly.
+    #[test]
+    fn hardened_agent_connects_to_the_pinned_address_not_dns() {
+        use std::io::Write as _;
+        use std::net::TcpListener;
+        use std::sync::mpsc;
+
+        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).expect("bind stub listener");
+        let pinned = listener.local_addr().expect("stub listener address");
+        let (tx, rx) = mpsc::channel();
+
+        let server = std::thread::spawn(move || {
+            let (mut sock, _peer) = listener.accept().expect("stub listener accept failed");
+            // Signal that a connection ARRIVED before doing anything else: that
+            // arrival, on this exact socket, is the fact under test.
+            let _ = tx.send(());
+            // Drain the request head, then answer with the smallest valid reply.
+            let mut head = Vec::new();
+            let mut byte = [0u8; 1];
+            while !head.ends_with(b"\r\n\r\n") {
+                match sock.read(&mut byte) {
+                    Ok(0) | Err(_) => break,
+                    Ok(_) => head.push(byte[0]),
+                }
+            }
+            let _ = sock
+                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}");
+            let _ = sock.flush();
+            head
+        });
+
+        let sent = hardened_agent(vec![pinned])
+            .post("http://keyserver.test/keys")
+            .send("{}");
+
+        // 1. The connection reached the pinned socket at all. Checked FIRST so
+        //    a mis-wired resolver reports the resolver, not a confusing
+        //    downstream symptom.
+        rx.recv_timeout(Duration::from_secs(10)).expect(
+            "hardened_agent never connected to the pinned address — the custom \
+             resolver is not being consulted, so a DNS rebind between the guard \
+             and the POST can still redirect the key material",
+        );
+        // 2. The whole round-trip completed through it. Had the agent fallen
+        //    back to live DNS, an unresolvable host is an error, never a 200.
+        let resp = sent.expect("the pinned round-trip must complete");
+        assert_eq!(resp.status(), 200, "the stub server's reply must come back");
+        // 3. The pin redirected the CONNECTION without rewriting the request:
+        //    the original host still travels in the Host header.
+        let head = server.join().expect("stub server panicked");
+        let head = String::from_utf8_lossy(&head);
+        assert!(
+            head.contains("keyserver.test"),
+            "the pinned agent must still address the original host; got: {head}"
+        );
+    }
+
     // ── bearer_header ──────────────────────────────────────────────────────
 
     #[test]
@@ -812,8 +1054,14 @@ mod tests {
         }
     }
 
-    fn reply(status: u16, body: &str) -> ureq::Response {
-        ureq::Response::new(status, "stub", body).expect("stub response")
+    /// A stub reply. ureq 3 has no `Response::new`; a response is just an
+    /// `http::Response` carrying a `ureq::Body`, which is constructible
+    /// directly — so these stay honest unit tests with no server involved.
+    fn reply(status: u16, body: &str) -> ureq::http::Response<ureq::Body> {
+        ureq::http::Response::builder()
+            .status(status)
+            .body(ureq::Body::builder().data(body))
+            .expect("stub response")
     }
 
     /// THE regression. The key service returned HTTP 502 for ~seven hours and
@@ -823,11 +1071,7 @@ mod tests {
     #[test]
     fn http_5xx_is_a_source_failure_not_a_missing_key() {
         for status in [500u16, 502, 503, 504] {
-            let out = interpret_reply(
-                Err(ureq::Error::Status(status, reply(status, ""))),
-                &BareCtx,
-                7,
-            );
+            let out = interpret_reply(Err(ureq::Error::StatusCode(status)), &BareCtx, 7);
             assert_eq!(
                 out.expect_err("a 5xx must not look like an answer").code(),
                 libfreemkv::error::E_KEY_SERVICE_UNAVAILABLE,
@@ -842,7 +1086,7 @@ mod tests {
         assert!(miss.is_empty(), "no key in the body → no keys out");
 
         // And the two must not be the same outcome — the whole point.
-        let down = interpret_reply(Err(ureq::Error::Status(502, reply(502, ""))), &BareCtx, 7);
+        let down = interpret_reply(Err(ureq::Error::StatusCode(502)), &BareCtx, 7);
         assert!(
             down.is_err(),
             "502 and 200-with-no-entry must not collapse to the same result"
@@ -880,19 +1124,24 @@ mod tests {
 
     /// A transport failure — nothing answered at all — is the same verdict as a
     /// 5xx and equally not a missing key. Uses a refused connection to
-    /// 127.0.0.1:1 to obtain a REAL `ureq::Error::Transport` (which has no public
-    /// constructor); this touches no network and never reaches `query`, so the
-    /// SSRF guard is not involved.
+    /// 127.0.0.1:1 to obtain a REAL transport-class error; this touches no
+    /// network and never reaches `query`, so the SSRF guard is not involved.
     #[test]
     fn transport_failure_is_a_source_failure() {
-        let sent = ureq::AgentBuilder::new()
-            .timeout_connect(Duration::from_secs(2))
-            .build()
+        let config = Config::builder()
+            .timeout_connect(Some(Duration::from_secs(2)))
+            .build();
+        let sent = ureq::Agent::new_with_config(config)
             .post("http://127.0.0.1:1/")
-            .send_string("{}");
+            .send("{}");
+        // ureq 3 split v2's single `Transport` variant across `Io`,
+        // `ConnectionFailed`, `Timeout` and friends, and which one a refused
+        // connection produces is a platform detail. The claim that matters is
+        // unchanged and is what this asserts: it failed, and NOT with a status
+        // code — nothing on the other end ever answered.
         assert!(
-            matches!(sent, Err(ureq::Error::Transport(_))),
-            "a refused connection must surface as a ureq transport error"
+            !matches!(sent, Ok(_) | Err(ureq::Error::StatusCode(_))),
+            "a refused connection must fail as transport, never as an HTTP status"
         );
         assert_eq!(
             interpret_reply(sent, &BareCtx, 3)
@@ -970,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() {