fix(kms): report unreadable keys and bound list-keys page size (#5764)

A key that cannot be described was handled two incompatible ways. Vault KV2 swallowed every describe failure and dropped the key from the page, so a damaged or newer-format record silently disappeared from the operator's inventory and from the deletion sweep's census. Local failed the whole listing instead, so one bad record stopped every scheduled deletion on the node for as long as the damage lasted. Both force a per-key problem into a whole-page answer.

ListKeysResponse now carries unreadable_key_ids, and the backends that read local key records classify per-key failures in one place: KeyNotFound is a concurrent deletion and is skipped, a material-level error names the key on the page, and anything else fails the listing, because it says nothing about a particular key and reporting it as key damage would turn a backend outage into a false data-loss alarm. A listing that covered the entire key set and found nothing readable still fails, since an empty page there is indistinguishable from a deployment with no keys; the guard is scoped to a page with no successor so a damaged key can never strand the keys behind it. The deletion sweep destroys the expired keys it can read, counts the unreadable ones, and withholds its lifecycle gauges rather than publishing a census over a key set it did not fully see.

Vault Transit needs the same treatment and is easy to miss: its per-key metadata records live in KV2 too, so folding every non-404 failure into a backend error left its per-key classification unreachable and one metadata record written by a newer build still failed every listing on the node.

Vault KV2 record reads gain the typed errors this needs: an unparseable body is MaterialCorrupt and an absent data envelope is MaterialMissing, where both were previously indistinguishable from Vault being unreachable. Only the parse failure's category and position are reported, because serde's own message embeds the offending scalar and that message reaches a log line and an admin HTTP body.

The admin list handlers refuse a malformed limit with 400 instead of silently substituting the default page size, and every page is capped at 1000 where it is cut, so a single request can no longer fan out one metadata lookup per key without bound. The four operation-level KMS metrics gain a backend label, since operation names are shared across backends and a Transit latency regression was previously indistinguishable from an AWS one. The Static backend captures its reported creation date once instead of reading the clock on every describe and list. POST /kms/clear-cache gains a named response type with an unchanged wire shape.

crates/kms/src/backends/aws.rs

@@ -585,13 +585,16 @@ impl KmsBackend for AwsKmsBackend {
         // AWS rejects a `Limit` of zero, and clamping it up to one would return
         // a key to a caller that asked for none; the empty page is answered
         // here instead.
-        if list_keys_page_size(request.limit).is_none() {
+        let Some(page_size) = list_keys_page_size(request.limit) else {
             return Ok(empty_key_page());
-        }
+        };
 
+        // Taking the remote page size from the shared resolver keeps the AWS
+        // request under the same ceiling every other backend obeys; the AWS API
+        // maximum is the same 1000, so this never widens the remote page.
         let limit = request
             .limit
-            .map(|limit| i32::try_from(limit).unwrap_or(i32::MAX).clamp(1, 1000));
+            .map(|_| i32::try_from(page_size).unwrap_or(i32::MAX).clamp(1, 1000));
         let marker = request.marker.clone();
 
         let output = self
@@ -617,7 +620,17 @@ impl KmsBackend for AwsKmsBackend {
             let Some(key_id) = entry.key_id() else {
                 continue;
             };
-            let metadata = self.describe(key_id).await?;
+            let metadata = match self.describe(key_id).await {
+                Ok(metadata) => metadata,
+                // AWS `ListKeys` is eventually consistent, so a key destroyed
+                // between the listing and the describe is routine: it is
+                // dropped and the remote cursor still advances past it. There
+                // is no local record to be damaged here — key state lives in
+                // AWS — so `unreadable_key_ids` stays empty on this backend and
+                // every other failure fails the listing.
+                Err(KmsError::KeyNotFound { .. }) => continue,
+                Err(error) => return Err(error),
+            };
             if request
                 .usage_filter
                 .as_ref()
@@ -649,6 +662,8 @@ impl KmsBackend for AwsKmsBackend {
             keys,
             next_marker: output.next_marker.clone(),
             truncated: output.truncated,
+            // AWS owns key state; nothing here can be present-but-unreadable.
+            unreadable_key_ids: Vec::new(),
         })
     }
 

crates/kms/src/backends/local.rs

@@ -15,8 +15,8 @@
 //! Local file-based KMS backend implementation
 
 use crate::backends::{
-    BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_status_permits,
-    ensure_tag_keys_are_mutable, paginate_keys,
+    BackendCapabilities, ExpiredKeyRemoval, KmsBackend, ListedKeyFailure, StateGatedOperation, UnreadableKeys,
+    classify_listed_key_failure, ensure_key_status_permits, ensure_tag_keys_are_mutable, paginate_keys, started_at_the_first_key,
 };
 use crate::config::KmsConfig;
 use crate::config::LocalConfig;
@@ -1599,22 +1599,25 @@ impl LocalKmsClient {
         // Only the page is read from disk, so the cost of a list stays bounded
         // by the requested limit rather than by the size of the key set.
         let mut keys = Vec::with_capacity(page.items.len());
+        let mut unreadable = UnreadableKeys::default();
         for key_id in page.items {
             let key_info = match self.describe_key(key_id, None).await {
-                Ok(key_info) => key_info,
-                // A key that vanished between the scan and the read is dropped
-                // from the page: concurrent removal is normal, and the cursor
-                // is derived from the identifier list, so the listing still
-                // advances past it.
-                Err(KmsError::KeyNotFound { .. }) => {
-                    debug!(key_id, "skipping key removed while listing");
-                    continue;
+                Ok(key_info) => {
+                    unreadable.saw_readable();
+                    key_info
                 }
-                // Anything else means the record is still there and this build
-                // cannot interpret it. Dropping it would answer "these are
-                // your keys" with a set that silently omits one, and the
-                // deletion sweep would count a census it never fully saw.
-                Err(error) => return Err(error),
+                Err(error) => match classify_listed_key_failure(&error) {
+                    Some(ListedKeyFailure::Vanished) => {
+                        debug!(key_id, "skipping key removed while listing");
+                        continue;
+                    }
+                    Some(ListedKeyFailure::Unreadable) => {
+                        warn!(key_id, %error, "listing a key record this build cannot describe");
+                        unreadable.record(key_id, error);
+                        continue;
+                    }
+                    None => return Err(error),
+                },
             };
 
             if let Some(ref status_filter) = request.status_filter
@@ -1635,6 +1638,7 @@ impl LocalKmsClient {
             keys,
             next_marker: page.next_marker,
             truncated: page.truncated,
+            unreadable_key_ids: unreadable.into_reported_ids(!page.truncated && started_at_the_first_key(request))?,
         })
     }
 
@@ -3625,8 +3629,13 @@ mod tests {
     /// key that is still on disk, and the deletion sweep — which counts the
     /// lifecycle gauges out of the pages it lists — would report a census it
     /// never fully saw as complete.
+    ///
+    /// It must not fail the whole listing either: one damaged record would then
+    /// stop every readable key from ever being listed, and with it every
+    /// scheduled deletion on this node. The identifier is reported alongside the
+    /// keys that did read, so the page is honest and the caller still advances.
     #[tokio::test]
-    async fn list_keys_fails_closed_on_a_record_it_cannot_interpret() {
+    async fn list_keys_reports_a_record_it_cannot_interpret_without_dropping_it() {
         let (client, _temp_dir) = create_test_client().await;
         client.create_key("alpha", "AES_256", None).await.expect("create alpha");
         client.create_key("beta", "AES_256", None).await.expect("create beta");
@@ -3641,10 +3650,27 @@ mod tests {
             .await
             .expect("write record");
 
-        let error = client
+        let response = client
             .list_keys(&ListKeysRequest::default(), None)
             .await
-            .expect_err("a listing must not quietly omit a key it cannot read");
+            .expect("one unreadable record must not fail the whole listing");
+        assert_eq!(
+            response.keys.iter().map(|key| key.key_id.as_str()).collect::<Vec<_>>(),
+            vec!["alpha"],
+            "the readable key must still be listed"
+        );
+        assert_eq!(
+            response.unreadable_key_ids,
+            vec!["beta".to_string()],
+            "a key this build cannot read must be named, not quietly omitted"
+        );
+
+        // Describing it directly still fails closed with the typed error, and
+        // the raw marker value stays out of the message.
+        let error = client
+            .describe_key("beta", None)
+            .await
+            .expect_err("describe must fail closed");
         assert!(
             matches!(&error, KmsError::UnsupportedFormatVersion { key_id, version }
                 if key_id == "beta" && version == UNKNOWN_STORED_KEY_PROTECTION),
@@ -3653,6 +3679,110 @@ mod tests {
         assert!(!error.to_string().contains("secret-marker-value-must-not-leak"));
     }
 
+    /// Per-key attribution is only honest while some key on the page reads.
+    ///
+    /// When none does, the cause is almost certainly shared — a node reading
+    /// records written in a format it has no reader for, or a policy that
+    /// denies the whole subtree — and answering `200 OK` with an empty `keys`
+    /// list is indistinguishable, to every client that predates
+    /// `unreadable_key_ids`, from a deployment that simply has no keys. The
+    /// operator response to that is to provision a new key, which is the
+    /// destructive move the fail-closed rules exist to prevent.
+    #[tokio::test]
+    async fn a_page_whose_keys_are_all_unreadable_fails_instead_of_looking_empty() {
+        let (client, _temp_dir) = create_test_client().await;
+        for key_id in ["alpha", "beta"] {
+            client.create_key(key_id, "AES_256", None).await.expect("create key");
+            let key_path = client.master_key_path(key_id).expect("valid key id");
+            let mut record: serde_json::Value =
+                serde_json::from_slice(&fs::read(&key_path).await.expect("read record")).expect("decode record");
+            record["at_rest_protection"] = serde_json::json!({ "future_mode": ["opaque"] });
+            fs::write(&key_path, serde_json::to_vec_pretty(&record).expect("encode record"))
+                .await
+                .expect("write record");
+        }
+
+        let error = client
+            .list_keys(&ListKeysRequest::default(), None)
+            .await
+            .expect_err("a page with nothing readable must fail, not report an empty key set");
+        assert!(
+            matches!(&error, KmsError::UnsupportedFormatVersion { .. }),
+            "the failure must name what went wrong: {error:?}"
+        );
+
+        // An empty marker is not the same as no marker to a caller, but it is
+        // to the pager: both start at the first key. A generated client that
+        // always emits its cursor parameter must not fall through the guard.
+        let error = client
+            .list_keys(
+                &ListKeysRequest {
+                    marker: Some(String::new()),
+                    ..Default::default()
+                },
+                None,
+            )
+            .await
+            .expect_err("an empty marker starts at the first key and must not bypass the guard");
+        assert!(matches!(&error, KmsError::UnsupportedFormatVersion { .. }), "got {error:?}");
+    }
+
+    /// The all-unreadable guard must never become a cursor trap.
+    ///
+    /// It only fires for a listing that both started at the beginning and
+    /// reached the end, because such a page has no successor to advance to.
+    /// Applying it per page instead would mean a caller with `limit=1` gets a
+    /// failure — and a failure carries no `next_marker` — the moment its page
+    /// lands on the damaged key, leaving every key behind it permanently
+    /// unreachable.
+    #[tokio::test]
+    async fn a_damaged_key_never_blocks_paging_past_it() {
+        let (client, _temp_dir) = create_test_client().await;
+        for key_id in ["a-first", "b-damaged", "c-last"] {
+            client.create_key(key_id, "AES_256", None).await.expect("create key");
+        }
+        let key_path = client.master_key_path("b-damaged").expect("valid key id");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&fs::read(&key_path).await.expect("read record")).expect("decode record");
+        record["at_rest_protection"] = serde_json::json!({ "future_mode": ["opaque"] });
+        fs::write(&key_path, serde_json::to_vec_pretty(&record).expect("encode record"))
+            .await
+            .expect("write record");
+
+        // Walk the whole key set one key at a time, exactly as a client that
+        // pages until `truncated` is false would.
+        let mut marker = None;
+        let mut seen = Vec::new();
+        let mut reported_unreadable = Vec::new();
+        loop {
+            let page = client
+                .list_keys(
+                    &ListKeysRequest {
+                        limit: Some(1),
+                        marker: marker.clone(),
+                        ..Default::default()
+                    },
+                    None,
+                )
+                .await
+                .expect("a one-key page containing the damaged key must still be answerable");
+            seen.extend(page.keys.iter().map(|key| key.key_id.clone()));
+            reported_unreadable.extend(page.unreadable_key_ids.clone());
+            if !page.truncated {
+                break;
+            }
+            marker = page.next_marker;
+            assert!(marker.is_some(), "a truncated page must carry a cursor");
+        }
+
+        assert_eq!(
+            seen,
+            vec!["a-first".to_string(), "c-last".to_string()],
+            "paging must reach past the damage"
+        );
+        assert_eq!(reported_unreadable, vec!["b-damaged".to_string()]);
+    }
+
     #[tokio::test]
     async fn missing_salt_with_pre_marker_key_records_still_initializes() {
         let (dev_client, temp_dir) = create_dev_mode_client().await;

crates/kms/src/backends/mod.rs

@@ -162,6 +162,15 @@ pub(crate) fn ensure_rewrap_context_matches(
 /// Page size used when a [`ListKeysRequest`] does not ask for one.
 pub(crate) const DEFAULT_LIST_KEYS_PAGE_SIZE: u32 = 100;
 
+/// Largest page a single [`ListKeysRequest`] can be served.
+///
+/// A page is not a cheap slice: every listed identifier costs the backend one
+/// metadata lookup — a disk read on Local, an HTTP round trip on Vault Transit —
+/// so an unbounded `limit` turns one request into an unbounded fan-out against
+/// the key store. The ceiling is applied where the page is cut rather than at
+/// each caller, so no backend can opt out of it.
+pub(crate) const MAX_LIST_KEYS_PAGE_SIZE: u32 = 1_000;
+
 /// One page of a key set the backend has to slice itself.
 pub(crate) struct KeyPage<'a, T> {
     /// The identifiers this page covers, in listing order.
@@ -228,10 +237,14 @@ pub(crate) fn paginate_keys<'a, T>(sorted: &'a [T], request: &ListKeysRequest, k
 /// already gives `max-keys=0` on the S3 listing path — not a malformed one and
 /// not an omitted value. Rounding it up to a default would hand back a full
 /// page of keys to a caller that explicitly asked for none.
+///
+/// A larger request is clamped to [`MAX_LIST_KEYS_PAGE_SIZE`] rather than
+/// rejected: the caller still reaches every key by following `next_marker`, so
+/// clamping costs it an extra round trip where rejecting would break it.
 pub(crate) fn list_keys_page_size(limit: Option<u32>) -> Option<usize> {
     match limit.unwrap_or(DEFAULT_LIST_KEYS_PAGE_SIZE) {
         0 => None,
-        size => Some(size as usize),
+        size => Some(size.min(MAX_LIST_KEYS_PAGE_SIZE) as usize),
     }
 }
 
@@ -246,6 +259,106 @@ pub(crate) fn empty_key_page() -> ListKeysResponse {
         keys: Vec::new(),
         next_marker: None,
         truncated: false,
+        unreadable_key_ids: Vec::new(),
+    }
+}
+
+/// What a failed per-key describe means for the page being assembled.
+pub(crate) enum ListedKeyFailure {
+    /// The key disappeared between the identifier scan and the read. Concurrent
+    /// removal is normal and the cursor comes from the identifier list, so the
+    /// listing drops it and advances.
+    Vanished,
+    /// The record is still in the store and this build cannot interpret it.
+    /// Reported through [`ListKeysResponse::unreadable_key_ids`] rather than
+    /// omitted or turned into a whole-page failure.
+    Unreadable,
+}
+
+/// Decide whether a per-key describe failure may be attributed to that one key.
+///
+/// `None` means it may not: the error describes something outside the record,
+/// so the caller must fail the whole listing. Downgrading a timeout to "this key
+/// is unreadable" would report a Vault outage as mass key corruption.
+///
+/// The material-level variants are all per-record by construction: each names
+/// one key and stays true on re-read. That includes
+/// [`KmsError::MaterialAuthenticationFailed`], which on the local backend means
+/// one record's AEAD tag did not verify — bit rot or a torn write. The
+/// systemic reading of the same variant, a process holding the wrong master
+/// key, cannot reach a listing: it is rejected when the backend is constructed.
+/// The whole-key-set guard in [`UnreadableKeys`] covers whatever slips past
+/// that.
+pub(crate) fn classify_listed_key_failure(error: &KmsError) -> Option<ListedKeyFailure> {
+    match error {
+        KmsError::KeyNotFound { .. } => Some(ListedKeyFailure::Vanished),
+        KmsError::MaterialMissing { .. }
+        | KmsError::MaterialCorrupt { .. }
+        | KmsError::MaterialAuthenticationFailed { .. }
+        | KmsError::UnsupportedFormatVersion { .. }
+        | KmsError::BaselineVersionLost { .. } => Some(ListedKeyFailure::Unreadable),
+        _ => None,
+    }
+}
+
+/// Whether this request starts at the very beginning of the key set.
+///
+/// An empty `marker` is not the same as no marker to a caller, but it is to
+/// [`paginate_keys`]: no identifier sorts at or below the empty string, so the
+/// page starts at the first key either way. Testing `Option::is_none` alone
+/// would let `?marker=` — which a generated pager that always emits its cursor
+/// parameter sends on its first request — slip past the whole-key-set guard
+/// below and get the empty, healthy-looking page that guard exists to prevent.
+pub(crate) fn started_at_the_first_key(request: &ListKeysRequest) -> bool {
+    request.marker.as_deref().is_none_or(str::is_empty)
+}
+
+/// The unreadable identifiers of one page, and the guarantee that a *complete*
+/// listing never reports every key as damaged while looking empty.
+///
+/// The failure mode being guarded against is a shared cause — a mixed-version
+/// node reading records in a format it has no reader for — that makes every key
+/// unreadable at once. Answering that with `200 OK` and an empty `keys` list
+/// looks, to any client written before `unreadable_key_ids` existed, exactly
+/// like a deployment that has no keys.
+///
+/// The guard is deliberately scoped to a listing that reached the end of the key
+/// set on its first page. Firing it per page instead would be a trap: with
+/// `limit=1` a single damaged key would fail its own page, and since a failed
+/// page carries no `next_marker` the caller could never advance past it — every
+/// key behind the damaged one becomes permanently unreachable. A page that is
+/// truncated, or that resumed from a marker, always reports per-key so paging
+/// can advance; and an empty `keys` array on such a page is already a documented
+/// state, because filters are applied after the page is cut.
+#[derive(Default)]
+pub(crate) struct UnreadableKeys {
+    ids: Vec<String>,
+    first_error: Option<KmsError>,
+    readable: usize,
+}
+
+impl UnreadableKeys {
+    pub(crate) fn saw_readable(&mut self) {
+        self.readable += 1;
+    }
+
+    pub(crate) fn record(&mut self, key_id: &str, error: KmsError) {
+        self.ids.push(key_id.to_string());
+        self.first_error.get_or_insert(error);
+    }
+
+    /// The identifiers to report.
+    ///
+    /// `whole_key_set` says this page both started at the beginning and reached
+    /// the end, so there is no further page a caller could advance to — which is
+    /// what makes failing here safe. Every identifier has already been logged
+    /// individually by the backend, so the single returned error does not hide
+    /// the rest.
+    pub(crate) fn into_reported_ids(self, whole_key_set: bool) -> Result<Vec<String>> {
+        match self.first_error {
+            Some(error) if whole_key_set && self.readable == 0 => Err(error),
+            _ => Ok(self.ids),
+        }
     }
 }
 
@@ -860,6 +973,43 @@ mod tests {
         assert_eq!(page_of(&keys, Some(u32::MAX), Some("key-01")), (vec![keys[2].clone()], None, false));
     }
 
+    /// A page is capped however large a limit the caller asks for, and the
+    /// capped page still carries a cursor, so the caller reaches every key
+    /// instead of being cut off at the ceiling.
+    #[test]
+    fn page_size_is_capped_and_the_capped_page_still_advances() {
+        // Pinned as a number, not only symbolically: the published contract in
+        // `docs/operations/kms-admin-contract.md` states this exact ceiling, so
+        // raising it is an API change and has to be a deliberate edit here.
+        assert_eq!(MAX_LIST_KEYS_PAGE_SIZE, 1_000);
+        assert_eq!(
+            list_keys_page_size(Some(u32::MAX)),
+            Some(MAX_LIST_KEYS_PAGE_SIZE as usize),
+            "an unbounded limit must not become an unbounded per-key fan-out"
+        );
+        assert_eq!(list_keys_page_size(Some(MAX_LIST_KEYS_PAGE_SIZE)), Some(MAX_LIST_KEYS_PAGE_SIZE as usize));
+        // Under the ceiling the caller's limit is still honoured exactly.
+        assert_eq!(
+            list_keys_page_size(Some(MAX_LIST_KEYS_PAGE_SIZE - 1)),
+            Some(MAX_LIST_KEYS_PAGE_SIZE as usize - 1)
+        );
+
+        // Padded to a fixed width so the vector is sorted by identifier, which
+        // is what `paginate_keys` requires of its input.
+        let keys: Vec<String> = (0..MAX_LIST_KEYS_PAGE_SIZE as usize + 5)
+            .map(|index| format!("key-{index:04}"))
+            .collect();
+        let (items, next_marker, truncated) = page_of(&keys, Some(u32::MAX), None);
+        assert_eq!(items.len(), MAX_LIST_KEYS_PAGE_SIZE as usize);
+        assert!(truncated);
+        let next_marker = next_marker.expect("a capped page must hand back a cursor");
+        assert_eq!(next_marker, items.last().cloned().expect("page is non-empty"));
+
+        let (rest, _, still_truncated) = page_of(&keys, Some(u32::MAX), Some(&next_marker));
+        assert_eq!(rest.len(), 5, "following the cursor must reach the keys the cap held back");
+        assert!(!still_truncated);
+    }
+
     /// The cursor is an identifier, so a marker naming a key that no longer
     /// exists resumes after where it would have been instead of restarting.
     #[test]

crates/kms/src/backends/static_kms.rs

@@ -54,6 +54,18 @@ pub struct StaticKmsBackend {
     key_id: String,
     /// The raw 32-byte AES-256 key material (zeroed on drop).
     key: Zeroizing<[u8; KEY_SIZE]>,
+    /// When this backend was constructed, reported as the key's creation date.
+    ///
+    /// A statically configured key has no creation event to report, but the
+    /// value still has to be *stable*: reading it from the clock on each call
+    /// made `describe_key` and `list_keys` answer differently every time, so no
+    /// caller could diff an inventory or cache a description.
+    ///
+    /// The stability this buys is per process. The reported date still moves
+    /// across a restart and differs between nodes of one cluster, because there
+    /// is no creation event to anchor it to; callers must treat it as "when this
+    /// node loaded the key", not as the key's birth date.
+    created_at: Zoned,
 }
 
 impl StaticKmsBackend {
@@ -77,6 +89,7 @@ impl StaticKmsBackend {
         Ok(Self {
             key_id: static_config.key_id.clone(),
             key: key.into(),
+            created_at: Zoned::now(),
         })
     }
 
@@ -87,7 +100,7 @@ impl StaticKmsBackend {
             key_state: KeyState::Enabled,
             key_usage: KeyUsage::EncryptDecrypt,
             description: Some("Static single-key KMS backend".to_string()),
-            creation_date: Zoned::now(),
+            creation_date: self.created_at.clone(),
             deletion_date: None,
             origin: "EXTERNAL".to_string(),
             key_manager: "STATIC".to_string(),
@@ -277,19 +290,9 @@ impl StaticKmsBackend {
             return Ok(empty_key_page());
         }
 
-        let key_info = KeyInfo {
-            key_id: self.key_id.clone(),
-            description: Some("Static single-key KMS backend".to_string()),
-            algorithm: "AES_256".to_string(),
-            usage: KeyUsage::EncryptDecrypt,
-            status: KeyStatus::Active,
-            version: 1,
-            metadata: HashMap::new(),
-            tags: HashMap::new(),
-            created_at: Zoned::now(),
-            rotated_at: None,
-            created_by: None,
-        };
+        // Built through the same constructor `describe_key` uses, so a listed
+        // key and a described key can never drift apart.
+        let key_info = self.configured_key_info(&self.key_id)?;
 
         // The marker is an exclusive lower bound on the identifier, as it is
         // for every other backend.
@@ -316,6 +319,9 @@ impl StaticKmsBackend {
             keys: vec![key_info],
             next_marker: None,
             truncated: false,
+            // The configured key is described from memory, so it can never be
+            // present-but-unreadable.
+            unreadable_key_ids: Vec::new(),
         })
     }
 }
@@ -834,4 +840,37 @@ mod tests {
             .expect_err("create_key for configured key should return KeyAlreadyExists");
         assert!(create_err.to_string().contains("already exists"));
     }
+
+    /// The configured key's reported creation date must not move within a
+    /// process.
+    ///
+    /// It used to be read from the clock inside every describe and every list,
+    /// so two reads of the same unchanged key disagreed and no caller could
+    /// diff an inventory or cache a description. It also made `describe_key`
+    /// and `list_keys` report different dates for the one key that exists.
+    /// Stability across restarts and across nodes is not claimed — see the
+    /// field's own doc for why there is nothing to anchor it to.
+    #[tokio::test]
+    async fn configured_key_reports_a_stable_creation_date_within_a_process() {
+        let (backend, key_id, _key) = create_test_backend().await;
+
+        let first = KmsBackendTrait::describe_key(&backend, DescribeKeyRequest { key_id: key_id.clone() })
+            .await
+            .expect("describe_key should succeed")
+            .key_metadata
+            .creation_date;
+        tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+        let second = KmsBackendTrait::describe_key(&backend, DescribeKeyRequest { key_id: key_id.clone() })
+            .await
+            .expect("describe_key should succeed")
+            .key_metadata
+            .creation_date;
+        assert_eq!(first, second, "describing the same unchanged key twice must report one date");
+
+        let listed = KmsBackendTrait::list_keys(&backend, ListKeysRequest::default())
+            .await
+            .expect("list_keys should succeed");
+        let listed = listed.keys.first().expect("the configured key must be listed");
+        assert_eq!(listed.created_at, first, "the listed key must report the described date");
+    }
 }

crates/kms/src/backends/vault.rs

@@ -19,8 +19,9 @@ use crate::backends::vault_credentials::{
     token_source_for,
 };
 use crate::backends::{
-    BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, empty_key_page, ensure_key_state_permits,
-    ensure_key_status_permits, ensure_rewrap_context_matches, ensure_tag_keys_are_mutable, list_keys_page_size, paginate_keys,
+    BackendCapabilities, ExpiredKeyRemoval, KmsBackend, ListedKeyFailure, StateGatedOperation, UnreadableKeys,
+    classify_listed_key_failure, empty_key_page, ensure_key_state_permits, ensure_key_status_permits,
+    ensure_rewrap_context_matches, ensure_tag_keys_are_mutable, list_keys_page_size, paginate_keys, started_at_the_first_key,
 };
 use crate::config::{KmsConfig, VaultConfig};
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
@@ -185,6 +186,42 @@ fn is_cas_conflict(error: &ClientError) -> bool {
     )
 }
 
+/// Map a KV2 record read failure onto the typed error surface.
+///
+/// The three record-level outcomes are told apart from a backend outcome here,
+/// at the only place that knows a per-key record was being read: a 404 or a wrap
+/// failure is a key that is not there, an unparseable body is a record that is
+/// there and cannot be interpreted, and an empty `data` field is a record whose
+/// contents are gone. Folding the last two into a generic backend error made a
+/// damaged record indistinguishable from a Vault outage, which is exactly the
+/// distinction the fail-closed material rules and the listing contract turn on.
+///
+/// `record` names what was being read (`"key record"`, `"transit key
+/// metadata"`) and only reaches error text, never a metric label. Shared with
+/// the Transit backend, whose per-key metadata records live in KV2 too and
+/// otherwise had no way to be reported as damaged rather than as an outage.
+pub(super) fn map_key_record_read_error(key_id: &str, record: &str, error: ClientError) -> KmsError {
+    match error {
+        ClientError::ResponseWrapError | ClientError::APIError { code: 404, .. } => KmsError::key_not_found(key_id),
+        // Only the position and category of the parse failure are reported.
+        // serde's own message embeds the offending scalar ("invalid type:
+        // string \"…\", expected u32"), which for a key record is a value out
+        // of the stored material, and this message reaches both a log line and
+        // an admin HTTP body.
+        ClientError::JsonParseError { source } => KmsError::material_corrupt(
+            key_id,
+            format!(
+                "stored {record} does not deserialize ({:?} at line {}, column {})",
+                source.classify(),
+                source.line(),
+                source.column()
+            ),
+        ),
+        ClientError::ResponseDataEmptyError => KmsError::material_missing(key_id),
+        error => KmsError::backend_error(format!("Failed to read {record} from Vault: {error}")),
+    }
+}
+
 /// Decode and validate the stored master key material of a [`VaultKeyData`] record.
 ///
 /// This is the single read-side gate for KV2 key material: missing or undecodable
@@ -391,14 +428,7 @@ impl VaultKmsClient {
                 let vault = self.vault().map_err(AttemptError::fatal)?;
                 kv2::read_version(&vault.client, &self.kv_mount, path, secret_version)
                     .await
-                    .map_err(|e| {
-                        AttemptError::from_vaultrs(e, |e| match e {
-                            ClientError::ResponseWrapError | ClientError::APIError { code: 404, .. } => {
-                                KmsError::key_not_found(key_id)
-                            }
-                            e => KmsError::backend_error(format!("Failed to read key from Vault: {e}")),
-                        })
-                    })
+                    .map_err(|e| AttemptError::from_vaultrs(e, |e| map_key_record_read_error(key_id, "key record", e)))
             })
             .await?;
 
@@ -596,14 +626,9 @@ impl VaultKmsClient {
         let secret: VaultKeyData = self
             .run("vault_kv2_read_key", OpClass::ReadIdempotent, move || async move {
                 let vault = self.vault().map_err(AttemptError::fatal)?;
-                kv2::read(&vault.client, &self.kv_mount, path).await.map_err(|e| {
-                    AttemptError::from_vaultrs(e, |e| match e {
-                        ClientError::ResponseWrapError | ClientError::APIError { code: 404, .. } => {
-                            KmsError::key_not_found(key_id)
-                        }
-                        e => KmsError::backend_error(format!("Failed to read key from Vault: {e}")),
-                    })
-                })
+                kv2::read(&vault.client, &self.kv_mount, path)
+                    .await
+                    .map_err(|e| AttemptError::from_vaultrs(e, |e| map_key_record_read_error(key_id, "key record", e)))
             })
             .await?;
 
@@ -1222,12 +1247,25 @@ impl VaultKmsClient {
         let page = paginate_keys(&all_keys, request, String::as_str);
 
         let mut key_infos = Vec::with_capacity(page.items.len());
+        let mut unreadable = UnreadableKeys::default();
         for key_id in page.items {
-            // A key that disappeared between the listing and the read is
-            // dropped from the page rather than failing it; the cursor comes
-            // from the identifier list, so the listing still advances past it.
-            let Ok(key_info) = self.describe_key(key_id, None).await else {
-                continue;
+            let key_info = match self.describe_key(key_id, None).await {
+                Ok(key_info) => {
+                    unreadable.saw_readable();
+                    key_info
+                }
+                Err(error) => match classify_listed_key_failure(&error) {
+                    Some(ListedKeyFailure::Vanished) => {
+                        debug!(key_id, "skipping key removed while listing");
+                        continue;
+                    }
+                    Some(ListedKeyFailure::Unreadable) => {
+                        warn!(key_id, %error, "listing a KV2 key record this build cannot describe");
+                        unreadable.record(key_id, error);
+                        continue;
+                    }
+                    None => return Err(error),
+                },
             };
             if request
                 .status_filter
@@ -1246,6 +1284,7 @@ impl VaultKmsClient {
             keys: key_infos,
             next_marker: page.next_marker,
             truncated: page.truncated,
+            unreadable_key_ids: unreadable.into_reported_ids(!page.truncated && started_at_the_first_key(request))?,
         })
     }
 
@@ -1986,6 +2025,125 @@ mod tests {
         );
     }
 
+    /// A KV2 record that does not deserialize is a property of that one key, so
+    /// the listing names it and keeps going. Dropping it silently — which is
+    /// what this backend used to do for every describe failure — answered "these
+    /// are your keys" with a set that omitted one, and the deletion sweep took
+    /// its census over that partial set.
+    #[tokio::test]
+    async fn wired_list_reports_an_undeserializable_record_and_keeps_the_rest() {
+        let (_vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(serde_json::json!({ "keys": ["key-a", "key-b"] })),
+            ScriptedResponse::ok(kv2_read_data(&healthy_key_data())),
+            // `algorithm` is typed as a string; a number makes the record
+            // undecodable exactly as a newer or damaged writer would.
+            ScriptedResponse::ok(serde_json::json!({
+                "data": { "algorithm": 42 },
+                "metadata": { "created_time": "2026-01-01T00:00:00Z", "deletion_time": "", "custom_metadata": null, "destroyed": false, "version": 1 },
+            })),
+        ])
+        .await;
+
+        let response = client
+            .list_keys(&ListKeysRequest::default(), None)
+            .await
+            .expect("one undecodable record must not fail the whole listing");
+
+        assert_eq!(response.keys.len(), 1, "the readable key must still be listed");
+        assert_eq!(response.unreadable_key_ids, vec!["key-b".to_string()]);
+    }
+
+    /// The counterpart: an error that says nothing about a specific key must not
+    /// be reported as a damaged key. A Vault outage that listed itself as mass
+    /// key corruption would send an operator hunting for a data-loss event that
+    /// never happened, and would let the sweep proceed on a key set it could not
+    /// actually read.
+    #[tokio::test]
+    async fn wired_list_fails_when_the_backend_itself_is_unavailable() {
+        let mut responses = vec![
+            ScriptedResponse::ok(serde_json::json!({ "keys": ["key-a", "key-b"] })),
+            ScriptedResponse::ok(kv2_read_data(&healthy_key_data())),
+        ];
+        for _ in 0..SCRIPTED_RETRY_ATTEMPTS {
+            responses.push(ScriptedResponse::error(503, "temporarily unavailable"));
+        }
+        let (_vault, client) = scripted_client(responses).await;
+
+        let error = client
+            .list_keys(&ListKeysRequest::default(), None)
+            .await
+            .expect_err("an unreachable backend must fail the listing, not shrink it");
+        assert!(
+            matches!(error, KmsError::BackendError { .. }),
+            "a transient backend failure must not be reported as a damaged record: {error:?}"
+        );
+    }
+
+    /// A record whose body is present but not a key record is corrupt material,
+    /// not a backend problem — and the reported message carries only where the
+    /// parse failed, never the values it tripped over.
+    #[tokio::test]
+    async fn wired_read_reports_an_uninterpretable_record_as_corrupt_material() {
+        let (_vault, client) = scripted_client(vec![ScriptedResponse::ok(serde_json::json!({
+            "data": null,
+            "metadata": { "created_time": "2026-01-01T00:00:00Z", "deletion_time": "", "custom_metadata": null, "destroyed": false, "version": 1 },
+        }))])
+        .await;
+
+        let error = client
+            .describe_key("wired-key", None)
+            .await
+            .expect_err("an uninterpretable key record must fail closed");
+        assert!(matches!(&error, KmsError::MaterialCorrupt { .. }), "got {error:?}");
+    }
+
+    /// A `200` whose envelope carries no `data` at all is material that is gone
+    /// — a distinct outcome from a record that cannot be parsed, and from Vault
+    /// being unreachable.
+    #[tokio::test]
+    async fn wired_read_reports_an_absent_record_body_as_missing_material() {
+        let (_vault, client) = scripted_client(vec![ScriptedResponse::Http {
+            status: 200,
+            body: serde_json::json!({
+                "request_id": "scripted",
+                "lease_id": "",
+                "lease_duration": 0,
+                "renewable": false,
+            })
+            .to_string(),
+        }])
+        .await;
+
+        let error = client
+            .describe_key("wired-key", None)
+            .await
+            .expect_err("a record with no body must fail closed");
+        assert!(matches!(&error, KmsError::MaterialMissing { .. }), "got {error:?}");
+    }
+
+    /// serde names the offending scalar in its own message ("invalid type:
+    /// string \"…\", expected u32"). For a key record that scalar comes out of
+    /// the stored material, and this message reaches both a log line and an
+    /// admin HTTP body, so only the position and category may survive.
+    #[tokio::test]
+    async fn wired_read_does_not_echo_record_values_into_the_error() {
+        let (_vault, client) = scripted_client(vec![ScriptedResponse::ok(serde_json::json!({
+            "data": { "version": "sensitive-value-must-not-leak" },
+            "metadata": { "created_time": "2026-01-01T00:00:00Z", "deletion_time": "", "custom_metadata": null, "destroyed": false, "version": 1 },
+        }))])
+        .await;
+
+        let error = client
+            .describe_key("wired-key", None)
+            .await
+            .expect_err("a type-mismatched key record must fail closed");
+        assert!(matches!(&error, KmsError::MaterialCorrupt { .. }), "got {error:?}");
+        assert!(
+            !error.to_string().contains("sensitive-value-must-not-leak"),
+            "the error echoed a stored record value: {error}"
+        );
+    }
+
     #[tokio::test]
     async fn wired_read_retries_transient_status_then_succeeds() {
         let (vault, client) = scripted_client(vec![

crates/kms/src/backends/vault_transit.rs

@@ -14,13 +14,15 @@
 
 //! Vault Transit-based KMS backend.
 
+use crate::backends::vault::map_key_record_read_error;
 use crate::backends::vault_credentials::{
     CredentialTaskHandle, VaultClientHandle, VaultConnectionSettings, VaultCredentialPolicy, VaultCredentialProvider,
     token_source_for,
 };
 use crate::backends::{
-    BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, empty_key_page, ensure_key_state_permits,
-    ensure_rewrap_context_matches, ensure_tag_keys_are_mutable, list_keys_page_size, paginate_keys,
+    BackendCapabilities, ExpiredKeyRemoval, KmsBackend, ListedKeyFailure, StateGatedOperation, UnreadableKeys,
+    classify_listed_key_failure, empty_key_page, ensure_key_state_permits, ensure_rewrap_context_matches,
+    ensure_tag_keys_are_mutable, list_keys_page_size, paginate_keys, started_at_the_first_key,
 };
 use crate::config::{KmsConfig, VaultTransitConfig};
 use crate::encryption::{DataKeyEnvelope, generate_key_material};
@@ -37,7 +39,7 @@ use std::future::Future;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio_util::sync::CancellationToken;
-use tracing::info;
+use tracing::{debug, info, warn};
 use vaultrs::{
     api::kv2::requests::SetSecretRequestOptions,
     api::transit::{
@@ -434,8 +436,13 @@ impl VaultTransitKmsClient {
                 Ok(persisted) => Ok(Some(persisted.into())),
                 Err(vaultrs::error::ClientError::ResponseWrapError)
                 | Err(vaultrs::error::ClientError::APIError { code: 404, .. }) => Ok(None),
+                // A metadata record that is present but undecodable is a
+                // property of this one key, so it is reported as such rather
+                // than as a backend outage: otherwise a single record written
+                // by a newer build fails every listing on the node, and with it
+                // every scheduled deletion.
                 Err(e) => Err(AttemptError::from_vaultrs(e, |e| {
-                    KmsError::backend_error(format!("Failed to read transit key metadata from Vault KV: {e}"))
+                    map_key_record_read_error(key_id, "transit key metadata", e)
                 })),
             }
         })
@@ -1073,8 +1080,26 @@ impl VaultTransitKmsClient {
         // Reading metadata only for the page keeps a list bounded by the
         // requested limit instead of by the size of the transit mount.
         let mut keys = Vec::with_capacity(page.items.len());
+        let mut unreadable = UnreadableKeys::default();
         for key_id in page.items {
-            let key_info = self.key_info(key_id).await?;
+            let key_info = match self.key_info(key_id).await {
+                Ok(key_info) => {
+                    unreadable.saw_readable();
+                    key_info
+                }
+                Err(error) => match classify_listed_key_failure(&error) {
+                    Some(ListedKeyFailure::Vanished) => {
+                        debug!(key_id, "skipping key removed while listing");
+                        continue;
+                    }
+                    Some(ListedKeyFailure::Unreadable) => {
+                        warn!(key_id, %error, "listing a transit key this build cannot describe");
+                        unreadable.record(key_id, error);
+                        continue;
+                    }
+                    None => return Err(error),
+                },
+            };
             let usage_matches = request.usage_filter.as_ref().is_none_or(|usage| usage == &key_info.usage);
             let status_matches = request.status_filter.as_ref().is_none_or(|status| status == &key_info.status);
             if usage_matches && status_matches {
@@ -1086,6 +1111,7 @@ impl VaultTransitKmsClient {
             keys,
             next_marker: page.next_marker,
             truncated: page.truncated,
+            unreadable_key_ids: unreadable.into_reported_ids(!page.truncated && started_at_the_first_key(request))?,
         })
     }
 
@@ -1664,6 +1690,86 @@ mod tests {
         serde_json::to_value(&response).expect("serialize transit key read response")
     }
 
+    /// A listing must not silently shrink when the backend is the problem.
+    ///
+    /// Before the per-key classification this path used `?`, so any describe
+    /// failure failed the page; the risk introduced by classifying is the
+    /// opposite one — quietly dropping a key on an error that says nothing
+    /// about it. A transit mount that stops answering must still fail loudly.
+    #[tokio::test]
+    async fn list_fails_when_a_transit_key_read_is_unavailable() {
+        let mut responses = vec![ScriptedResponse::ok(serde_json::json!({ "keys": ["key-a"] }))];
+        for _ in 0..3 {
+            responses.push(ScriptedResponse::error(503, "temporarily unavailable"));
+        }
+        let (_vault, client) = scripted_client(responses).await;
+
+        let error = client
+            .list_keys(&ListKeysRequest::default(), None)
+            .await
+            .expect_err("an unreachable transit mount must fail the listing, not empty it");
+        assert!(
+            matches!(error, KmsError::BackendError { .. }),
+            "a transient backend failure must not be reported as a damaged key: {error:?}"
+        );
+    }
+
+    /// A transit key whose persisted metadata record cannot be decoded is
+    /// reported per key, not as a backend outage.
+    ///
+    /// The metadata record lives in KV2 exactly like a KV2 key record, so it has
+    /// the same failure mode: without this classification one record written by
+    /// a newer build fails every listing on the node, and the deletion sweep —
+    /// which aborts on a listing error — stops destroying every other expired
+    /// key for as long as that record is there.
+    #[tokio::test]
+    async fn list_reports_an_undecodable_metadata_record_per_key() {
+        let (_vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(serde_json::json!({ "keys": ["key-a", "key-b"] })),
+            ScriptedResponse::ok(transit_key_read_data("key-a")),
+            ScriptedResponse::ok(metadata_read_data(&TransitKeyMetadata::from_create_request(
+                &CreateKeyRequest::default(),
+            ))),
+            ScriptedResponse::ok(transit_key_read_data("key-b")),
+            // `key_usage` is an enum; a number cannot be decoded into it.
+            ScriptedResponse::ok(serde_json::json!({
+                "data": { "key_usage": 42 },
+                "metadata": { "created_time": "2026-01-01T00:00:00Z", "deletion_time": "", "custom_metadata": null, "destroyed": false, "version": 1 },
+            })),
+        ])
+        .await;
+
+        let response = client
+            .list_keys(&ListKeysRequest::default(), None)
+            .await
+            .expect("one undecodable metadata record must not fail the whole listing");
+        assert_eq!(response.keys.len(), 1, "the readable key must still be listed");
+        assert_eq!(response.unreadable_key_ids, vec!["key-b".to_string()]);
+    }
+
+    /// A key destroyed between the listing and the read is dropped, and the
+    /// listing still succeeds — the cursor comes from the identifier list, so
+    /// it advances past the gap on its own.
+    #[tokio::test]
+    async fn list_drops_a_key_that_vanished_between_the_scan_and_the_read() {
+        let (_vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(serde_json::json!({ "keys": ["key-a"] })),
+            ScriptedResponse::error(404, "no handler for route"),
+        ])
+        .await;
+
+        let response = client
+            .list_keys(&ListKeysRequest::default(), None)
+            .await
+            .expect("a key removed mid-listing must not fail the page");
+        assert!(response.keys.is_empty());
+        assert!(
+            response.unreadable_key_ids.is_empty(),
+            "a concurrent deletion is not damage: {:?}",
+            response.unreadable_key_ids
+        );
+    }
+
     /// A caller asking for no keys gets an empty page, and the page arithmetic
     /// never reaches for the element before an empty page. The scripted key
     /// listing stays unused: a request for zero keys has nothing to ask Vault.

crates/kms/src/deletion_worker.rs

@@ -63,7 +63,7 @@ const METRIC_TOMBSTONE_KEYS: &str = "rustfs_kms_deletion_tombstone_keys";
 /// (its creation time when it was never rotated); `0` when there are none.
 const METRIC_OLDEST_ROTATION_AGE_SECONDS: &str = "rustfs_kms_oldest_key_rotation_age_seconds";
 /// Counter: keys the sweep acted on, by `outcome` (`removed`, `blocked`,
-/// `skipped`, `failed`).
+/// `skipped`, `failed`, `unreadable`).
 const METRIC_SWEEP_KEYS_TOTAL: &str = "rustfs_kms_deletion_sweep_keys_total";
 
 /// Register metric descriptions once per process.
@@ -140,6 +140,7 @@ fn record_sweep(report: &SweepReport, census: Option<KeyCensus>) {
         ("blocked", report.blocked.len()),
         ("skipped", report.skipped),
         ("failed", report.failed),
+        ("unreadable", report.unreadable),
     ] {
         // Emitted even at zero so every outcome series exists from the first
         // sweep on and a rate over it is defined.
@@ -193,6 +194,13 @@ pub struct SweepReport {
     pub skipped: usize,
     /// Keys whose removal attempt failed; retried on the next sweep.
     pub failed: usize,
+    /// Keys the backend listed but could not describe.
+    ///
+    /// The sweep keeps going past them — one damaged record must not stop every
+    /// other expired key from being destroyed — but their existence means the
+    /// key set was only partially observed, so the lifecycle gauges are
+    /// withheld for this round rather than published over an incomplete census.
+    pub unreadable: usize,
 }
 
 pub(crate) struct DeletionWorker {
@@ -243,12 +251,13 @@ impl DeletionWorker {
                 _ = ticker.tick() => {}
             }
             let report = self.sweep(&Zoned::now()).await;
-            if !report.removed.is_empty() || !report.blocked.is_empty() || report.failed > 0 {
+            if !report.removed.is_empty() || !report.blocked.is_empty() || report.failed > 0 || report.unreadable > 0 {
                 info!(
                     removed = ?report.removed,
                     blocked = ?report.blocked,
                     skipped = report.skipped,
                     failed = report.failed,
+                    unreadable = report.unreadable,
                     "KMS deletion sweep completed"
                 );
             }
@@ -280,6 +289,13 @@ impl DeletionWorker {
                     break false;
                 }
             };
+            if !response.unreadable_key_ids.is_empty() {
+                warn!(
+                    key_ids = ?response.unreadable_key_ids,
+                    "KMS deletion sweep listed key records it cannot describe; census withheld this round"
+                );
+                report.unreadable += response.unreadable_key_ids.len();
+            }
             for key in &response.keys {
                 // Keys this sweep destroys are left out of the census: the
                 // gauges describe the key set as it stands once the sweep is
@@ -310,7 +326,12 @@ impl DeletionWorker {
                 None => break false,
             }
         };
-        record_sweep(&report, listed_everything.then_some(census));
+        // A census taken over a key set with unreadable members would report an
+        // oldest-rotation age and a pending-deletion count computed from the
+        // keys that happened to be readable, which is exactly the kind of quiet
+        // undercount the gauges exist to catch.
+        let observed_every_key = listed_everything && report.unreadable == 0;
+        record_sweep(&report, observed_every_key.then_some(census));
         report
     }
 
@@ -849,4 +870,52 @@ mod tests {
 
         assert_eq!(gauge_value(&snapshot, METRIC_PENDING_DELETION_KEYS), Some(total as f64));
     }
+
+    /// One key record this build cannot read must not stop the sweep.
+    ///
+    /// Before the listing reported unreadable identifiers, the damaged record
+    /// either vanished from the page — so the census counted a key set it had
+    /// not fully seen — or failed the listing outright, which aborted the sweep
+    /// and left every expired key on the node undeleted for as long as the
+    /// damage lasted. Now the expired key is still destroyed, and the gauges are
+    /// withheld for the round instead of being published over a partial census.
+    #[test]
+    fn sweep_destroys_expired_keys_past_a_record_it_cannot_read() {
+        let (snapshot, ()) = record_metrics(|| {
+            Box::pin(async move {
+                let temp_dir = tempfile::tempdir().expect("temp dir");
+                let backend = local_backend(&temp_dir).await;
+                let expired = create_key(&backend, "zz-expired").await;
+                schedule(&backend, &expired).await;
+                let damaged = create_key(&backend, "aa-damaged").await;
+
+                // Stamp a protection marker no build understands, exactly as a
+                // record written by a newer node would look here.
+                let key_path = temp_dir.path().join(format!("{damaged}.key"));
+                assert!(key_path.exists(), "the key record must exist before it is damaged");
+                let mut record: serde_json::Value =
+                    serde_json::from_slice(&tokio::fs::read(&key_path).await.expect("read record")).expect("decode record");
+                record["at_rest_protection"] = serde_json::json!({ "future_mode": ["opaque"] });
+                tokio::fs::write(&key_path, serde_json::to_vec_pretty(&record).expect("encode record"))
+                    .await
+                    .expect("write record");
+
+                let report = worker(backend.clone()).sweep(&after_window()).await;
+                assert_eq!(report.unreadable, 1, "the damaged record must be reported, not hidden");
+                assert_eq!(report.removed, vec![expired.clone()], "the expired key must still be destroyed");
+                assert_eq!(report.failed, 0, "an unreadable record is not a removal failure");
+                assert_key_gone(&backend, &expired).await;
+            })
+        });
+
+        assert_eq!(
+            gauge_value(&snapshot, METRIC_PENDING_DELETION_KEYS),
+            None,
+            "a census taken over a partially readable key set must not be published"
+        );
+        // The runbook points operators at this series as the signal that the
+        // gauges above have gone quiet on purpose, so it has to be emitted.
+        assert_eq!(counter_value(&snapshot, METRIC_SWEEP_KEYS_TOTAL, "unreadable"), 1);
+        assert_eq!(counter_value(&snapshot, METRIC_SWEEP_KEYS_TOTAL, "removed"), 1);
+    }
 }

crates/kms/src/manager.rs

@@ -655,6 +655,7 @@ mod tests {
                 keys: Vec::new(),
                 next_marker: None,
                 truncated: false,
+                unreadable_key_ids: Vec::new(),
             })
         }
 

crates/kms/src/policy.rs

@@ -256,6 +256,9 @@ struct BackendCapacity {
 
 #[derive(Debug)]
 struct BackendRuntime {
+    /// Static backend name, carried so per-operation metrics can be attributed
+    /// to the backend that served them rather than aggregated across all of them.
+    backend: &'static str,
     capacity: Arc<BackendCapacity>,
     capacity_class: CapacityClass,
     queued: Arc<Semaphore>,
@@ -284,6 +287,7 @@ impl BackendRuntime {
         in_flight.increment(0.0);
         circuit_open.increment(0.0);
         Self {
+            backend,
             capacity: active,
             capacity_class: capacity,
             queued: Arc::new(Semaphore::new(DEFAULT_MAX_QUEUED_OPERATIONS)),
@@ -495,16 +499,16 @@ fn equal_jitter(rng: &mut impl RngExt, cap: Duration) -> Duration {
 // ciphertext, and tokens must never reach a metric label.
 // ---------------------------------------------------------------------------
 
-/// Counter: operations executed, by `operation`, `op_class`, and `outcome`.
+/// Counter: operations executed, by `backend`, `operation`, `op_class`, and `outcome`.
 const METRIC_OPERATIONS_TOTAL: &str = "rustfs_kms_backend_operations_total";
-/// Counter: failed attempts, by `operation` and `error_class` (including
-/// `attempt_timeout` for attempts cut off by the per-attempt timeout).
+/// Counter: failed attempts, by `backend`, `operation` and `error_class`
+/// (including `attempt_timeout` for attempts cut off by the per-attempt timeout).
 const METRIC_ATTEMPT_FAILURES_TOTAL: &str = "rustfs_kms_backend_attempt_failures_total";
 /// Histogram: wall-clock duration of a whole operation (attempts plus
-/// backoff), in seconds, by `operation` and `outcome`.
+/// backoff), in seconds, by `backend`, `operation` and `outcome`.
 const METRIC_OPERATION_DURATION_SECONDS: &str = "rustfs_kms_backend_operation_duration_seconds";
-/// Histogram: attempts one operation used before completing, by `operation`
-/// and `outcome`.
+/// Histogram: attempts one operation used before completing, by `backend`,
+/// `operation` and `outcome`.
 const METRIC_OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
 /// Gauge: backend attempts currently in flight.
 const METRIC_IN_FLIGHT: &str = "rustfs_kms_backend_in_flight";
@@ -572,19 +576,19 @@ fn describe_metrics() {
     DESCRIBE.call_once(|| {
         metrics::describe_counter!(
             METRIC_OPERATIONS_TOTAL,
-            "Total KMS backend operations executed under the operation policy, by operation, operation class, and outcome"
+            "Total KMS backend operations executed under the operation policy, by backend, operation, operation class, and outcome"
         );
         metrics::describe_counter!(
             METRIC_ATTEMPT_FAILURES_TOTAL,
-            "Total failed KMS backend attempts, by operation and retry classification"
+            "Total failed KMS backend attempts, by backend, operation and retry classification"
         );
         metrics::describe_histogram!(
             METRIC_OPERATION_DURATION_SECONDS,
-            "Wall-clock duration of KMS backend operations including retries and backoff, in seconds"
+            "Wall-clock duration of KMS backend operations including retries and backoff, in seconds, by backend and operation"
         );
         metrics::describe_histogram!(
             METRIC_OPERATION_ATTEMPTS,
-            "Number of attempts a KMS backend operation used before completing"
+            "Number of attempts a KMS backend operation used before completing, by backend and operation"
         );
         metrics::describe_gauge!(METRIC_IN_FLIGHT, "KMS backend attempts currently in flight, by backend and policy scope");
         metrics::describe_gauge!(
@@ -595,9 +599,10 @@ fn describe_metrics() {
 }
 
 /// Record one failed attempt with its retry classification.
-fn record_attempt_failure(operation: &'static str, error_class: &'static str) {
+fn record_attempt_failure(backend: &'static str, operation: &'static str, error_class: &'static str) {
     metrics::counter!(
         METRIC_ATTEMPT_FAILURES_TOTAL,
+        "backend" => backend,
         "operation" => operation,
         "error_class" => error_class
     )
@@ -605,9 +610,17 @@ fn record_attempt_failure(operation: &'static str, error_class: &'static str) {
 }
 
 /// Record the terminal outcome of one policy execution.
-fn record_operation(operation: &'static str, class: OpClass, outcome: Outcome, attempts: u32, elapsed: Duration) {
+fn record_operation(
+    backend: &'static str,
+    operation: &'static str,
+    class: OpClass,
+    outcome: Outcome,
+    attempts: u32,
+    elapsed: Duration,
+) {
     metrics::counter!(
         METRIC_OPERATIONS_TOTAL,
+        "backend" => backend,
         "operation" => operation,
         "op_class" => class.as_label(),
         "outcome" => outcome.as_label()
@@ -615,12 +628,14 @@ fn record_operation(operation: &'static str, class: OpClass, outcome: Outcome, a
     .increment(1);
     metrics::histogram!(
         METRIC_OPERATION_DURATION_SECONDS,
+        "backend" => backend,
         "operation" => operation,
         "outcome" => outcome.as_label()
     )
     .record(elapsed.as_secs_f64());
     metrics::histogram!(
         METRIC_OPERATION_ATTEMPTS,
+        "backend" => backend,
         "operation" => operation,
         "outcome" => outcome.as_label()
     )
@@ -750,7 +765,7 @@ where
     let started = Instant::now();
     let mut attempts_made = 0u32;
     let (outcome, result) = drive_attempts(operation, class, policy, cancel, jitter, attempt, &mut attempts_made).await;
-    record_operation(operation, class, outcome, attempts_made, started.elapsed());
+    record_operation(policy.runtime.backend, operation, class, outcome, attempts_made, started.elapsed());
     result
 }
 
@@ -821,11 +836,11 @@ where
                 return (Outcome::Success, Ok(value));
             }
             Ok(Err(failure)) => {
-                record_attempt_failure(operation, failure.class.as_label());
+                record_attempt_failure(policy.runtime.backend, operation, failure.class.as_label());
                 failure
             }
             Err(_) => {
-                record_attempt_failure(operation, "attempt_timeout");
+                record_attempt_failure(policy.runtime.backend, operation, "attempt_timeout");
                 AttemptError {
                     class: ErrorClass::RetryableConn,
                     error: KmsError::operation_timed_out(format!(
@@ -874,11 +889,16 @@ where
 
 #[cfg(test)]
 fn test_runtime(max_concurrent: usize, max_queued: usize) -> Arc<BackendRuntime> {
+    named_test_runtime("test-backend", max_concurrent, max_queued)
+}
+
+#[cfg(test)]
+fn named_test_runtime(backend: &'static str, max_concurrent: usize, max_queued: usize) -> Arc<BackendRuntime> {
     let capacity = Arc::new(BackendCapacity {
         total: Arc::new(Semaphore::new(max_concurrent)),
         operations: Arc::new(Semaphore::new(max_concurrent)),
     });
-    let mut runtime = BackendRuntime::new("test-backend", "test-scope", capacity, CapacityClass::Credentials);
+    let mut runtime = BackendRuntime::new(backend, "test-scope", capacity, CapacityClass::Credentials);
     runtime.queued = Arc::new(Semaphore::new(max_queued));
     Arc::new(runtime)
 }
@@ -1765,6 +1785,83 @@ mod tests {
         assert!((durations[0] - 0.3).abs() < 1e-9, "expected 0.3s of virtual backoff, got {durations:?}");
     }
 
+    /// Request, error and latency series must be attributable to one backend.
+    ///
+    /// Operation names are shared across backends (`decrypt` exists on every
+    /// one of them), so without a `backend` label a Vault Transit latency
+    /// regression is indistinguishable from an AWS one in the same series.
+    #[test]
+    fn operation_metrics_separate_series_per_backend() {
+        fn policy_for(backend: &'static str) -> RetryPolicy {
+            let mut policy = policy_of(1_000, 60_000, 2, 100, 2_000);
+            policy.runtime = named_test_runtime(backend, DEFAULT_MAX_CONCURRENT_OPERATIONS, DEFAULT_MAX_QUEUED_OPERATIONS);
+            policy
+        }
+
+        let (snapshot, ()) = record_metrics(|| {
+            Box::pin(async {
+                let cancel = CancellationToken::new();
+                execute_with_jitter(
+                    "decrypt",
+                    OpClass::ReadIdempotent,
+                    &policy_for("vault-transit"),
+                    &cancel,
+                    full_jitter,
+                    || async { Ok(()) },
+                )
+                .await
+                .expect("transit attempt succeeds");
+
+                let result: Result<()> =
+                    execute_with_jitter("decrypt", OpClass::ReadIdempotent, &policy_for("aws"), &cancel, full_jitter, || async {
+                        Err(AttemptError {
+                            class: ErrorClass::Fatal,
+                            error: KmsError::access_denied("permission denied (403)"),
+                        })
+                    })
+                    .await;
+                assert!(matches!(result, Err(KmsError::AccessDenied { .. })), "got {result:?}");
+            })
+        });
+
+        let succeeded = [("operation", "decrypt"), ("outcome", "success")];
+        let failed = [("operation", "decrypt"), ("outcome", "fatal")];
+
+        assert_eq!(counter_value(&snapshot, METRIC_OPERATIONS_TOTAL, &[("backend", "vault-transit")]), 1);
+        assert_eq!(counter_value(&snapshot, METRIC_OPERATIONS_TOTAL, &[("backend", "aws")]), 1);
+        assert_eq!(
+            counter_value(&snapshot, METRIC_OPERATIONS_TOTAL, &[("backend", "aws"), succeeded[0], succeeded[1]]),
+            0,
+            "the AWS failure must not be counted under the transit success series"
+        );
+        assert_eq!(
+            counter_value(
+                &snapshot,
+                METRIC_ATTEMPT_FAILURES_TOTAL,
+                &[("backend", "aws"), ("operation", "decrypt"), ("error_class", "fatal")]
+            ),
+            1
+        );
+        assert_eq!(
+            counter_value(
+                &snapshot,
+                METRIC_ATTEMPT_FAILURES_TOTAL,
+                &[("backend", "vault-transit"), ("operation", "decrypt")]
+            ),
+            0
+        );
+
+        for (backend, labels) in [("vault-transit", succeeded), ("aws", failed)] {
+            let with_backend = [("backend", backend), labels[0], labels[1]];
+            assert_eq!(
+                histogram_values(&snapshot, METRIC_OPERATION_DURATION_SECONDS, &with_backend).len(),
+                1,
+                "{backend} must own a latency series of its own"
+            );
+            assert_eq!(histogram_values(&snapshot, METRIC_OPERATION_ATTEMPTS, &with_backend), vec![1.0]);
+        }
+    }
+
     #[test]
     fn metrics_record_fatal_outcome_with_single_attempt() {
         let (snapshot, ()) = record_metrics(|| {

crates/kms/src/types.rs

@@ -448,6 +448,18 @@ pub struct ListKeysResponse {
     pub next_marker: Option<String>,
     /// Whether there are more keys available
     pub truncated: bool,
+    /// Identifiers that are present in the key store but that this build could
+    /// not describe, in listing order.
+    ///
+    /// A key whose record cannot be interpreted must not simply be missing from
+    /// `keys`: an inventory that silently omits it reads as "you do not have
+    /// this key", and the deletion sweep's census would be taken over a key set
+    /// it never fully saw. Reporting the identifier separately keeps the page
+    /// honest while still letting the caller page past the damage. Errors that
+    /// say nothing about a specific key — timeouts, 5xx, auth failures — still
+    /// fail the whole listing rather than landing here.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub unreadable_key_ids: Vec<String>,
 }
 
 /// Operation context for auditing and access control