fix(kms): report real cache counters through the admin status API

KmsStatusResponse.cache_stats mapped the old (entry_count, 0) tuple onto
hit_count and miss_count, so operators polling KMS status read the entry
count as a hit count and a miss count that was always zero.

Map the fields to the counters they claim to be, and add entry_count and
eviction_count as additive, defaulted fields so the entry number that
hit_count used to carry is still available.

Refs rustfs/backlog#1584

.github/workflows/ci-docs-only.yml

@@ -12,24 +12,18 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# Companion to ci.yml for required status checks.
+# Companion to ci.yml for the required "Test and Lint" status check.
 #
-# ci.yml skips docs-only pull requests via paths-ignore, but the branch ruleset
-# requires a check named "Test and Lint" — without this workflow a docs-only PR
-# would wait on it forever. This workflow triggers on exactly the paths ci.yml
-# ignores and reports success under the same job name. Mixed PRs trigger both
-# workflows and the real check still gates: a required check with any failing
-# run blocks the merge.
+# ci.yml skips docs-only pull requests via paths-ignore, but the branch
+# ruleset requires a check named "Test and Lint" — without this workflow a
+# docs-only PR would wait on that check forever. This workflow triggers on
+# exactly the paths ci.yml ignores and reports an instant success under the
+# same job name. Mixed PRs trigger both workflows and the real check still
+# gates: a required check with any failing run blocks the merge.
 # https://docs.github.com/en/repositories/configuring-branches-and-merges-in-your-repository/defining-the-mergeability-of-pull-requests/troubleshooting-required-status-checks#handling-skipped-but-required-checks
 #
-# "Quick Checks" is mirrored here ahead of the ruleset change that will make it
-# required too (rustfs/backlog#1599). It is not required yet, so today this job
-# is inert; adding it first is what lets that ruleset change land without
-# stranding docs-only PRs on a check nobody reports.
-#
 # Keep the paths list below in sync with the pull_request paths-ignore list
-# in ci.yml, and keep the quick-checks steps below byte-identical to the
-# quick-checks job in ci.yml.
+# in ci.yml.
 
 name: Continuous Integration (docs only)
 
@@ -58,63 +52,9 @@ permissions:
   contents: read
 
 jobs:
-  # Deliberately NOT a bare `echo`. Once "Quick Checks" becomes a required
-  # check, ci.yml gates every expensive job behind it, so a mixed PR reports
-  # two check runs with this name: the real one (45-51s) and this companion.
-  # GitHub has no written contract for how it picks between same-named
-  # required check runs ("latest wins" vs "any failure blocks"), so instead of
-  # relying on ordering we make both runs execute the same commands against
-  # the same merge ref — their conclusions are then necessarily identical and
-  # the choice does not matter. Keep these steps byte-identical to the
-  # quick-checks job in ci.yml (a guard script that asserts this, and the paths
-  # sync below, is tracked in rustfs/backlog#1603).
-  #
-  # For a genuinely docs-only PR this adds no strictness (no code changed, so
-  # fmt and the guards always pass) and costs ~50s of ubuntu-latest.
-  quick-checks:
-    name: Quick Checks
-    runs-on: ubuntu-latest
-    timeout-minutes: 10
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
-
-      - name: Install ripgrep
-        run: sudo apt-get update && sudo apt-get install -y ripgrep
-
-      - name: Install Rust toolchain
-        uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
-        with:
-          components: rustfmt
-
-      - name: Check code formatting
-        run: cargo fmt --all --check
-
-      - name: Check unsafe code allowances
-        run: ./scripts/check_unsafe_code_allowances.sh
-
-      - name: Check layered dependencies
-        run: ./scripts/check_layer_dependencies.sh
-
-      - name: Check architecture migration rules
-        run: ./scripts/check_architecture_migration_rules.sh
-
-      - name: Check tokio io-uring feature guard
-        run: ./scripts/check_no_tokio_io_uring.sh
-
-      - name: Check extension schema boundaries
-        run: ./scripts/check_extension_schema_boundaries.sh
-
-      - name: Check body-cache whitelist guard
-        run: ./scripts/check_body_cache_whitelist.sh
-
-      - name: Check no planning docs committed
-        run: ./scripts/check_no_planning_docs.sh
-
   test-and-lint:
     name: Test and Lint
     runs-on: ubuntu-latest
-    timeout-minutes: 10
     steps:
       - name: Checkout repository
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7

.github/workflows/ci.yml

@@ -96,10 +96,6 @@ jobs:
 
   # Fast, compile-free checks that fail early so contributors get feedback in
   # ~1 minute instead of waiting for the full test job.
-  #
-  # These steps are mirrored byte-for-byte in ci-docs-only.yml so that a mixed
-  # PR, which reports two check runs named "Quick Checks", cannot get one red
-  # and one green. Edit both jobs together.
   quick-checks:
     name: Quick Checks
     if: github.event_name != 'pull_request' || github.event.action != 'closed'
@@ -452,13 +448,6 @@ jobs:
 
   uring-integration:
     name: io_uring Integration (real)
-    # The pull_request trigger includes `closed` purely so the concurrency
-    # group cancels in-flight runs of a closed PR; every other job opts out of
-    # that run with this guard (or is skipped through its `needs` chain). This
-    # job had neither, so each closed/merged PR really ran the whole io_uring
-    # suite (measured 4m17s / 7m19s / 7m31s on runs 30678272341 / 30678117601 /
-    # 30662728539) and kept the cancellation run in progress for minutes.
-    if: github.event_name != 'pull_request' || github.event.action != 'closed'
     # GitHub-hosted ubuntu-latest runs a recent kernel with io_uring and, unlike
     # a container, applies no seccomp filter that would block io_uring_setup — so
     # the probe succeeds and the tests exercise the real UringBackend/FdCache/
@@ -757,13 +746,7 @@ jobs:
   # evaluates ILM within ~2s of the due time, well inside the poll window.
   s3-lifecycle-behavior-tests:
     name: S3 Lifecycle Behavior Tests
-    # Also gated on e2e-tests, matching s3-implemented-tests: when the e2e smoke
-    # suite is already red this lane cannot tell us anything new, and it holds a
-    # sm-standard-4 for up to 30 minutes doing so. Both lanes only download the
-    # prebuilt debug binary (no cargo build), and s3-implemented-tests — which
-    # already waits on e2e-tests — finishes later anyway, so a green PR's total
-    # wall clock is unchanged.
-    needs: [ build-rustfs-debug-binary, e2e-tests ]
+    needs: [ build-rustfs-debug-binary ]
     runs-on: sm-standard-4
     timeout-minutes: 30
     steps:

crates/e2e_test/src/object_lock/object_lock_test.rs

@@ -26,7 +26,6 @@
 
 use super::common::*;
 use aws_sdk_s3::Client;
-use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::primitives::{ByteStream, DateTimeFormat};
 use aws_sdk_s3::types::{
     CompletedMultipartUpload, CompletedPart, Delete, MetadataDirective, ObjectIdentifier, ObjectLockLegalHoldStatus,
@@ -2121,127 +2120,6 @@ async fn test_multipart_default_retention_fixed_at_create() {
 // Versioning Auto-Enable Tests
 // ============================================================================
 
-#[tokio::test]
-#[serial]
-async fn test_unretained_object_lock_object_delete_and_bucket_cleanup() {
-    init_logging();
-    info!("🧪 Test: Unretained Object Lock object delete and bucket cleanup (Issue #5339)");
-
-    let mut env = ObjectLockTestEnvironment::new()
-        .await
-        .expect("failed to create Object Lock test environment");
-    env.start_rustfs().await.expect("failed to start RustFS");
-
-    let bucket = "test-object-lock-delete-cleanup";
-    let key = "unretained-object";
-
-    env.create_object_lock_bucket(bucket)
-        .await
-        .expect("failed to create Object Lock bucket");
-    let client = env.s3_client();
-
-    let put_response = client
-        .put_object()
-        .bucket(bucket)
-        .key(key)
-        .body(ByteStream::from_static(b"unretained data"))
-        .send()
-        .await
-        .expect("failed to upload unretained object");
-    let object_version_id = put_response
-        .version_id()
-        .expect("Object Lock buckets must create versioned objects")
-        .to_string();
-
-    let delete_response = client
-        .delete_object()
-        .bucket(bucket)
-        .key(key)
-        .send()
-        .await
-        .expect("failed to create delete marker");
-    assert_eq!(delete_response.delete_marker(), Some(true));
-    let delete_marker_version_id = delete_response
-        .version_id()
-        .expect("Deleting without a version ID must create a delete marker")
-        .to_string();
-
-    let get_error = client
-        .get_object()
-        .bucket(bucket)
-        .key(key)
-        .send()
-        .await
-        .expect_err("GET must not return an object hidden by a delete marker");
-    assert_eq!(get_error.raw_response().map(|response| response.status().as_u16()), Some(404));
-    assert_eq!(get_error.as_service_error().and_then(|error| error.code()), Some("NoSuchKey"));
-
-    let listed_objects = client
-        .list_objects_v2()
-        .bucket(bucket)
-        .send()
-        .await
-        .expect("failed to list current objects");
-    assert!(
-        listed_objects.contents().iter().all(|object| object.key() != Some(key)),
-        "ListObjectsV2 must hide objects whose latest version is a delete marker"
-    );
-
-    let listed_versions = client
-        .list_object_versions()
-        .bucket(bucket)
-        .send()
-        .await
-        .expect("failed to list object versions");
-    assert!(
-        listed_versions
-            .versions()
-            .iter()
-            .any(|version| version.key() == Some(key) && version.version_id() == Some(object_version_id.as_str())),
-        "The data version must remain until it is explicitly deleted"
-    );
-    assert!(
-        listed_versions
-            .delete_markers()
-            .iter()
-            .any(|marker| marker.key() == Some(key) && marker.version_id() == Some(delete_marker_version_id.as_str())),
-        "ListObjectVersions must expose the delete marker"
-    );
-
-    client
-        .delete_object()
-        .bucket(bucket)
-        .key(key)
-        .version_id(object_version_id)
-        .send()
-        .await
-        .expect("failed to delete the data version");
-    client
-        .delete_object()
-        .bucket(bucket)
-        .key(key)
-        .version_id(delete_marker_version_id)
-        .send()
-        .await
-        .expect("failed to delete the delete marker");
-
-    let remaining_versions = client
-        .list_object_versions()
-        .bucket(bucket)
-        .send()
-        .await
-        .expect("failed to list versions after cleanup");
-    assert!(remaining_versions.versions().is_empty());
-    assert!(remaining_versions.delete_markers().is_empty());
-
-    client
-        .delete_bucket()
-        .bucket(bucket)
-        .send()
-        .await
-        .expect("Deleting every version must remove xl.meta so the bucket can be deleted normally");
-}
-
 #[tokio::test]
 #[serial]
 async fn test_versioning_auto_enabled_with_object_lock() {

crates/kms/examples/kms_local_demo.rs

@@ -227,10 +227,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // Step 12: Check cache statistics
     println!("12. Checking cache statistics...");
-    if let Some((hits, misses)) = encryption_service.cache_stats().await {
+    if let Some(stats) = encryption_service.cache_stats().await {
         println!("   ✓ Cache statistics:");
-        println!("     - Cache hits: {}", hits);
-        println!("     - Cache misses: {}\n", misses);
+        println!("     - Cached entries: {}", stats.entries);
+        println!("     - Cache hits: {}", stats.hits);
+        println!("     - Cache misses: {}", stats.misses);
+        println!("     - Entries evicted: {}\n", stats.evictions);
     } else {
         println!("   - Cache is disabled\n");
     }

crates/kms/examples/kms_vault_kv_demo.rs

@@ -263,10 +263,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // Step 12: Check cache statistics
     println!("12. Checking cache statistics...");
-    if let Some((hits, misses)) = encryption_service.cache_stats().await {
+    if let Some(stats) = encryption_service.cache_stats().await {
         println!("   ✓ Cache statistics:");
-        println!("     - Cache hits: {}", hits);
-        println!("     - Cache misses: {}\n", misses);
+        println!("     - Cached entries: {}", stats.entries);
+        println!("     - Cache hits: {}", stats.hits);
+        println!("     - Cache misses: {}", stats.misses);
+        println!("     - Entries evicted: {}\n", stats.evictions);
     } else {
         println!("   - Cache is disabled\n");
     }

crates/kms/src/cache.rs

@@ -16,11 +16,82 @@
 
 use crate::types::KeyMetadata;
 use moka::future::Cache;
+use moka::notification::RemovalCause;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
 
+/// Default lifetime of a cached key metadata entry.
+const DEFAULT_METADATA_TTL: Duration = Duration::from_secs(300);
+
+// ---------------------------------------------------------------------------
+// Metrics
+//
+// Lookup outcomes and removals are counted here because moka exposes neither
+// hit nor miss counts; the numbers below are the cache's own, not derived.
+// Label values are exclusively static strings (lookup result, removal cause) —
+// key identifiers and key metadata must never reach a metric label.
+// ---------------------------------------------------------------------------
+
+/// Counter: key metadata lookups, by `result` (`hit` or `miss`).
+const METRIC_CACHE_LOOKUPS_TOTAL: &str = "rustfs_kms_metadata_cache_lookups_total";
+/// Counter: entries dropped from the cache, by `cause` (`expired`, `size`,
+/// `explicit`, `replaced`); only `expired` and `size` are true evictions, the
+/// rest are invalidations driven by key lifecycle operations.
+const METRIC_CACHE_EVICTIONS_TOTAL: &str = "rustfs_kms_metadata_cache_evictions_total";
+/// Gauge: entries currently held by the cache.
+const METRIC_CACHE_ENTRIES: &str = "rustfs_kms_metadata_cache_entries";
+
+/// Register metric descriptions once per process.
+fn describe_metrics() {
+    static DESCRIBE: std::sync::Once = std::sync::Once::new();
+    DESCRIBE.call_once(|| {
+        metrics::describe_counter!(METRIC_CACHE_LOOKUPS_TOTAL, "Total KMS key metadata cache lookups, by hit or miss");
+        metrics::describe_counter!(
+            METRIC_CACHE_EVICTIONS_TOTAL,
+            "Total entries dropped from the KMS key metadata cache, by removal cause"
+        );
+        metrics::describe_gauge!(METRIC_CACHE_ENTRIES, "Entries currently held by the KMS key metadata cache");
+    });
+}
+
+fn removal_cause_label(cause: RemovalCause) -> &'static str {
+    match cause {
+        RemovalCause::Expired => "expired",
+        RemovalCause::Explicit => "explicit",
+        RemovalCause::Replaced => "replaced",
+        RemovalCause::Size => "size",
+    }
+}
+
+/// Cumulative cache counters. Shared with moka's eviction listener, which runs
+/// outside any `&self` borrow, hence the `Arc` and the atomics.
+#[derive(Debug, Default)]
+struct CacheCounters {
+    hits: AtomicU64,
+    misses: AtomicU64,
+    evictions: AtomicU64,
+}
+
+/// Snapshot of the KMS key metadata cache counters.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct KmsCacheStats {
+    /// Entries currently held. Eventually consistent: moka applies pending
+    /// maintenance lazily.
+    pub entries: u64,
+    /// Lookups served from the cache since process start.
+    pub hits: u64,
+    /// Lookups that fell through to the backend since process start.
+    pub misses: u64,
+    /// Entries dropped since process start, whatever the cause (expiry,
+    /// capacity, invalidation, replacement).
+    pub evictions: u64,
+}
+
 /// KMS cache for storing frequently accessed keys and metadata
 pub struct KmsCache {
     key_metadata_cache: Cache<String, KeyMetadata>,
+    counters: Arc<CacheCounters>,
 }
 
 impl KmsCache {
@@ -33,11 +104,26 @@ impl KmsCache {
     /// A new instance of `KmsCache`
     ///
     pub fn new(capacity: u64) -> Self {
+        Self::with_ttl(capacity, DEFAULT_METADATA_TTL)
+    }
+
+    /// Create a new KMS cache with an explicit metadata time-to-live
+    fn with_ttl(capacity: u64, metadata_ttl: Duration) -> Self {
+        describe_metrics();
+
+        let counters = Arc::new(CacheCounters::default());
+        let eviction_counters = Arc::clone(&counters);
+
         Self {
             key_metadata_cache: Cache::builder()
                 .max_capacity(capacity)
-                .time_to_live(Duration::from_secs(300)) // 5 minutes default TTL
+                .time_to_live(metadata_ttl)
+                .eviction_listener(move |_key: Arc<String>, _metadata: KeyMetadata, cause: RemovalCause| {
+                    eviction_counters.evictions.fetch_add(1, Ordering::Relaxed);
+                    metrics::counter!(METRIC_CACHE_EVICTIONS_TOTAL, "cause" => removal_cause_label(cause)).increment(1);
+                })
                 .build(),
+            counters,
         }
     }
 
@@ -50,7 +136,9 @@ impl KmsCache {
     /// An `Option` containing the `KeyMetadata` if found, or `None` if not found
     ///
     pub async fn get_key_metadata(&self, key_id: &str) -> Option<KeyMetadata> {
-        self.key_metadata_cache.get(key_id).await
+        let cached = self.key_metadata_cache.get(key_id).await;
+        self.record_lookup(cached.is_some());
+        cached
     }
 
     /// Put key metadata into cache
@@ -62,6 +150,7 @@ impl KmsCache {
     pub async fn put_key_metadata(&mut self, key_id: &str, metadata: &KeyMetadata) {
         self.key_metadata_cache.insert(key_id.to_string(), metadata.clone()).await;
         self.key_metadata_cache.run_pending_tasks().await;
+        self.record_entry_count();
     }
 
     /// Remove key metadata from cache
@@ -71,6 +160,11 @@ impl KmsCache {
     ///
     pub async fn remove_key_metadata(&mut self, key_id: &str) {
         self.key_metadata_cache.remove(key_id).await;
+
+        // Flush maintenance so the removal notification and the entry gauge
+        // describe the cache as it is once this call returns.
+        self.key_metadata_cache.run_pending_tasks().await;
+        self.record_entry_count();
     }
 
     /// Clear all cached entries
@@ -79,18 +173,35 @@ impl KmsCache {
 
         // Wait for invalidation to complete
         self.key_metadata_cache.run_pending_tasks().await;
+        self.record_entry_count();
     }
 
-    /// Get cache statistics (hit count, miss count)
+    /// Get cache statistics
     ///
     /// # Returns
-    /// A tuple containing total entries and total misses
+    /// A [`KmsCacheStats`] snapshot of entry count, hits, misses and evictions
     ///
-    pub fn stats(&self) -> (u64, u64) {
-        (
-            self.key_metadata_cache.entry_count(),
-            0u64, // moka doesn't provide miss count directly
-        )
+    pub fn stats(&self) -> KmsCacheStats {
+        KmsCacheStats {
+            entries: self.key_metadata_cache.entry_count(),
+            hits: self.counters.hits.load(Ordering::Relaxed),
+            misses: self.counters.misses.load(Ordering::Relaxed),
+            evictions: self.counters.evictions.load(Ordering::Relaxed),
+        }
+    }
+
+    fn record_lookup(&self, hit: bool) {
+        let (counter, result) = if hit {
+            (&self.counters.hits, "hit")
+        } else {
+            (&self.counters.misses, "miss")
+        };
+        counter.fetch_add(1, Ordering::Relaxed);
+        metrics::counter!(METRIC_CACHE_LOOKUPS_TOTAL, "result" => result).increment(1);
+    }
+
+    fn record_entry_count(&self) {
+        metrics::gauge!(METRIC_CACHE_ENTRIES).set(self.key_metadata_cache.entry_count() as f64);
     }
 }
 
@@ -99,37 +210,90 @@ mod tests {
     use super::*;
     use crate::types::{KeyState, KeyUsage};
     use jiff::Zoned;
+    use metrics_util::MetricKind;
+    use metrics_util::debugging::{DebugValue, DebuggingRecorder};
     use std::time::Duration;
 
-    #[derive(Debug, Clone)]
-    struct CacheInfo {
-        key_metadata_count: u64,
-    }
-
-    impl CacheInfo {
-        fn total_entries(&self) -> u64 {
-            self.key_metadata_count
-        }
-    }
-
     impl KmsCache {
-        fn with_ttl_for_tests(capacity: u64, metadata_ttl: Duration) -> Self {
-            Self {
-                key_metadata_cache: Cache::builder().max_capacity(capacity).time_to_live(metadata_ttl).build(),
-            }
-        }
-
-        fn info_for_tests(&self) -> CacheInfo {
-            CacheInfo {
-                key_metadata_count: self.key_metadata_cache.entry_count(),
-            }
-        }
-
         fn contains_key_metadata_for_tests(&self, key_id: &str) -> bool {
             self.key_metadata_cache.contains_key(key_id)
         }
     }
 
+    fn test_metadata(key_id: &str) -> KeyMetadata {
+        KeyMetadata {
+            key_id: key_id.to_string(),
+            key_state: KeyState::Enabled,
+            key_usage: KeyUsage::EncryptDecrypt,
+            description: None,
+            creation_date: Zoned::now(),
+            deletion_date: None,
+            origin: "KMS".to_string(),
+            key_manager: "CUSTOMER".to_string(),
+            tags: std::collections::HashMap::new(),
+        }
+    }
+
+    type MetricEntry = (
+        metrics_util::CompositeKey,
+        Option<metrics::Unit>,
+        Option<metrics::SharedString>,
+        DebugValue,
+    );
+
+    /// Drive `test` on a current-thread runtime under a thread-local debugging
+    /// recorder and return its output plus one snapshot of everything emitted.
+    ///
+    /// A single snapshot per test on purpose: `Snapshotter::snapshot` drains
+    /// the recorded state, so taking it per assertion would leave every
+    /// assertion after the first with nothing to look at.
+    fn record_metrics<F, Fut, T>(test: F) -> (T, Vec<MetricEntry>)
+    where
+        F: FnOnce() -> Fut,
+        Fut: std::future::Future<Output = T>,
+    {
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        let output = metrics::with_local_recorder(&recorder, || {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("current-thread runtime must build");
+            runtime.block_on(test())
+        });
+        (output, snapshotter.snapshot().into_vec())
+    }
+
+    /// Sum of counters with `name` whose labels include every `(key, value)` pair.
+    fn counter_value(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)]) -> u64 {
+        snapshot
+            .iter()
+            .filter_map(|(composite, _unit, _description, value)| {
+                let key = composite.key();
+                let matches = composite.kind() == MetricKind::Counter
+                    && key.name() == name
+                    && labels
+                        .iter()
+                        .all(|(label, expected)| key.labels().any(|l| l.key() == *label && l.value() == *expected));
+                match (matches, value) {
+                    (true, DebugValue::Counter(count)) => Some(*count),
+                    _ => None,
+                }
+            })
+            .sum()
+    }
+
+    /// Last value recorded for the unlabelled gauge `name`.
+    fn gauge_value(snapshot: &[MetricEntry], name: &str) -> Option<f64> {
+        snapshot.iter().find_map(|(composite, _unit, _description, value)| {
+            let matches = composite.kind() == MetricKind::Gauge && composite.key().name() == name;
+            match (matches, value) {
+                (true, DebugValue::Gauge(gauge)) => Some(**gauge),
+                _ => None,
+            }
+        })
+    }
+
     #[tokio::test]
     async fn test_cache_operations() {
         let mut cache = KmsCache::new(100);
@@ -154,19 +318,16 @@ mod tests {
         assert_eq!(retrieved.expect("metadata should be cached").key_id, "test-key-1");
 
         // Test cache info
-        let info = cache.info_for_tests();
-        assert_eq!(info.key_metadata_count, 1);
-        assert_eq!(info.total_entries(), 1);
+        assert_eq!(cache.stats().entries, 1);
 
         // Test cache clearing
         cache.clear().await;
-        let info_after_clear = cache.info_for_tests();
-        assert_eq!(info_after_clear.total_entries(), 0);
+        assert_eq!(cache.stats().entries, 0);
     }
 
     #[tokio::test]
     async fn test_cache_with_custom_ttl() {
-        let mut cache = KmsCache::with_ttl_for_tests(
+        let mut cache = KmsCache::with_ttl(
             100,
             Duration::from_millis(100), // Short TTL for testing
         );
@@ -217,4 +378,59 @@ mod tests {
 
         assert!(cache.contains_key_metadata_for_tests("contains-test"));
     }
+
+    #[test]
+    fn lookups_report_real_hit_and_miss_counts() {
+        let (stats, snapshot) = record_metrics(|| async {
+            let mut cache = KmsCache::new(100);
+
+            assert!(cache.get_key_metadata("absent").await.is_none());
+            cache.put_key_metadata("present", &test_metadata("present")).await;
+            assert!(cache.get_key_metadata("present").await.is_some());
+            assert!(cache.get_key_metadata("absent").await.is_none());
+
+            cache.stats()
+        });
+
+        assert_eq!(stats.hits, 1);
+        assert_eq!(stats.misses, 2);
+        assert_eq!(counter_value(&snapshot, METRIC_CACHE_LOOKUPS_TOTAL, &[("result", "hit")]), 1);
+        assert_eq!(counter_value(&snapshot, METRIC_CACHE_LOOKUPS_TOTAL, &[("result", "miss")]), 2);
+    }
+
+    #[test]
+    fn removals_report_their_cause_and_the_resulting_entry_count() {
+        let (stats, snapshot) = record_metrics(|| async {
+            let mut cache = KmsCache::new(100);
+
+            cache.put_key_metadata("key", &test_metadata("key")).await;
+            cache.put_key_metadata("key", &test_metadata("key")).await;
+            cache.remove_key_metadata("key").await;
+
+            cache.stats()
+        });
+
+        assert_eq!(counter_value(&snapshot, METRIC_CACHE_EVICTIONS_TOTAL, &[("cause", "replaced")]), 1);
+        assert_eq!(counter_value(&snapshot, METRIC_CACHE_EVICTIONS_TOTAL, &[("cause", "explicit")]), 1);
+        assert_eq!(stats.evictions, 2);
+        assert_eq!(stats.entries, 0);
+        assert_eq!(gauge_value(&snapshot, METRIC_CACHE_ENTRIES), Some(0.0));
+    }
+
+    #[test]
+    fn capacity_pressure_reports_size_evictions() {
+        let (stats, snapshot) = record_metrics(|| async {
+            let mut cache = KmsCache::with_ttl(1, DEFAULT_METADATA_TTL);
+
+            cache.put_key_metadata("first", &test_metadata("first")).await;
+            cache.put_key_metadata("second", &test_metadata("second")).await;
+
+            cache.stats()
+        });
+
+        assert_eq!(counter_value(&snapshot, METRIC_CACHE_EVICTIONS_TOTAL, &[("cause", "size")]), 1);
+        assert_eq!(stats.evictions, 1);
+        assert_eq!(stats.entries, 1);
+        assert_eq!(gauge_value(&snapshot, METRIC_CACHE_ENTRIES), Some(1.0));
+    }
 }

crates/kms/src/lib.rs

@@ -86,6 +86,7 @@ pub use api_types::{
     StartKmsResponse, StopKmsResponse, TagKeyRequest, TagKeyResponse, UntagKeyRequest, UntagKeyResponse,
     UpdateKeyDescriptionRequest, UpdateKeyDescriptionResponse,
 };
+pub use cache::KmsCacheStats;
 pub use config::*;
 pub use deletion_worker::DeletionReferenceChecker;
 pub use encryption::is_data_key_envelope;

crates/kms/src/manager.rs

@@ -15,7 +15,7 @@
 //! KMS manager for handling key operations and backend coordination
 
 use crate::backends::KmsBackend;
-use crate::cache::KmsCache;
+use crate::cache::{KmsCache, KmsCacheStats};
 use crate::config::KmsConfig;
 use crate::error::Result;
 use crate::types::{
@@ -113,8 +113,8 @@ impl KmsManager {
         self.backend.list_keys(request).await
     }
 
-    /// Get cache statistics
-    pub async fn cache_stats(&self) -> Option<(u64, u64)> {
+    /// Get cache statistics, or `None` when caching is disabled
+    pub async fn cache_stats(&self) -> Option<KmsCacheStats> {
         if self.enable_cache {
             let cache = self.cache.read().await;
             Some(cache.stats())
@@ -254,9 +254,12 @@ mod tests {
         let describe_response = manager.describe_key(describe_request).await.expect("Failed to describe key");
         assert_eq!(describe_response.key_metadata.key_id, create_response.key_id);
 
-        // Test cache stats
-        let stats = manager.cache_stats().await;
-        assert!(stats.is_some());
+        // Creating the key populated the cache, so the describe above was
+        // served from it rather than from the backend.
+        let stats = manager.cache_stats().await.expect("cache is enabled");
+        assert_eq!(stats.entries, 1);
+        assert_eq!(stats.hits, 1);
+        assert_eq!(stats.misses, 0);
 
         // Test health check
         let health = manager.health_check().await.expect("Health check failed");

crates/kms/src/service.rs

@@ -14,6 +14,7 @@
 
 //! Object encryption service for S3-compatible encryption
 
+use crate::cache::KmsCacheStats;
 use crate::encryption::ciphers::{create_cipher, generate_iv};
 use crate::error::{KmsError, Result};
 use crate::manager::KmsManager;
@@ -160,9 +161,9 @@ impl ObjectEncryptionService {
     /// Get cache statistics
     ///
     /// # Returns
-    /// Option with (hits, misses) if caching is enabled
+    /// A [`KmsCacheStats`] snapshot if caching is enabled, `None` otherwise
     ///
-    pub async fn cache_stats(&self) -> Option<(u64, u64)> {
+    pub async fn cache_stats(&self) -> Option<KmsCacheStats> {
         self.kms_manager.cache_stats().await
     }
 

rustfs/src/admin/handlers/kms_management.rs

@@ -82,6 +82,14 @@ pub struct KmsStatusResponse {
 pub struct CacheStatsResponse {
     pub hit_count: u64,
     pub miss_count: u64,
+    /// Entries currently cached. Additive field: omitted by older servers, so
+    /// it must stay defaulted for consumers.
+    #[serde(default)]
+    pub entry_count: u64,
+    /// Entries dropped since process start, whatever the cause. Additive
+    /// field, same compatibility rule as `entry_count`.
+    #[serde(default)]
+    pub eviction_count: u64,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -193,9 +201,11 @@ impl Operation for KmsStatusHandler {
             }
         };
 
-        let cache_stats = service.cache_stats().await.map(|(hits, misses)| CacheStatsResponse {
-            hit_count: hits,
-            miss_count: misses,
+        let cache_stats = service.cache_stats().await.map(|stats| CacheStatsResponse {
+            hit_count: stats.hits,
+            miss_count: stats.misses,
+            entry_count: stats.entries,
+            eviction_count: stats.evictions,
         });
         let config = kms_service_manager_from_context().get_redacted_config().await;