fix(kms): honour the configured metadata cache TTL and metrics switch (#5569)

* fix(kms): honour the configured metadata cache TTL and metrics switch

KmsManager::new built the KmsCache from cache_config.max_keys alone, so
cache_config.ttl was dead configuration: every deployment ran the
hardcoded 300s window whatever the admin configure API was given, while
CacheSummary and the KMS config endpoint reported the configured value
back. cache_config.enable_metrics was never read anywhere.

Build the cache from the whole CacheConfig. The documented default is
reconciled down to the 300s the cache has always used rather than up to
the advertised 3600s, and now lives in one place (DEFAULT_CACHE_TTL)
instead of being duplicated across the four configure-request
converters, so the default path behaves exactly as before.

Behaviour change: a deployment configured through the admin API already
has ttl 3600 persisted, because the old converters wrote that default
into the stored config, so its describe_key staleness window widens from
an effective 300s to the 3600s it asked for. No cryptographic or
authorization path widens - encrypt, decrypt and generate_data_key go
straight to the backend and never read this cache. The Vault Transit
backend's own metadata cache, which does gate crypto through
ensure_key_state_allows, stays fixed at 300s and is now documented as
deliberately not operator-tunable.

The configured duration now reaches moka's builder, which panics above
1000 years, so CacheConfig::effective_ttl clamps to a 24h maximum the
way effective_timeout already clamps its own, and validate rejects a
zero TTL beside the existing max_keys check. Both config summaries and
the KMS config endpoint report the effective value, so the admin API
cannot advertise a lifetime the cache does not honour.

enable_metrics gates publication of the rustfs_kms_metadata_cache_*
families only; the counters behind the admin status API keep running
either way. No configure-request field sets it yet.

Refs rustfs/backlog#1584

* docs(kms): state why the Transit metadata TTL is not bound to the default

The comment claimed the constant matches config::DEFAULT_CACHE_TTL, which
reads as an invariant the code does not enforce. Say plainly that the
equality is a coincidence rather than a contract, and why binding the two
would be wrong: this cache gates crypto through ensure_key_state_allows,
so a later change to the operator-facing describe-cache default must not
be able to widen its staleness window.

crates/kms/src/api_types.rs

@@ -15,9 +15,9 @@
 //! API types for KMS dynamic configuration
 
 use crate::config::{
-    BackendConfig, CacheConfig, DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX, DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT, KmsBackend,
-    KmsConfig, LocalConfig, StaticConfig, TlsConfig, VaultAuthMethod, VaultConfig, VaultTransitConfig,
-    allow_immediate_deletion_from_env, redacted_secret, redacted_secret_option,
+    BackendConfig, CacheConfig, DEFAULT_CACHE_TTL, DEFAULT_MAX_CACHED_KEYS, DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX,
+    DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT, KmsBackend, KmsConfig, LocalConfig, StaticConfig, TlsConfig, VaultAuthMethod,
+    VaultConfig, VaultTransitConfig, allow_immediate_deletion_from_env, redacted_secret, redacted_secret_option,
 };
 use crate::service_manager::KmsServiceStatus;
 use crate::types::{KeyMetadata, KeyUsage};
@@ -429,10 +429,15 @@ pub enum BackendSummary {
 
 impl From<&KmsConfig> for KmsConfigSummary {
     fn from(config: &KmsConfig) -> Self {
+        // Report the lifetime the cache was built with, not the raw configured
+        // value: an oversized `ttl` is clamped rather than rejected, and this
+        // response is what operators check the cache against.
+        let cache_ttl_seconds = config.cache_config.effective_ttl().as_secs();
+
         let cache_summary = if config.enable_cache {
             Some(CacheSummary {
                 max_keys: config.cache_config.max_keys,
-                ttl_seconds: config.cache_config.ttl.as_secs(),
+                ttl_seconds: cache_ttl_seconds,
                 enable_metrics: config.cache_config.enable_metrics,
             })
         } else {
@@ -487,7 +492,7 @@ impl From<&KmsConfig> for KmsConfigSummary {
             retry_attempts: config.retry_attempts,
             enable_cache: config.enable_cache,
             max_cached_keys: config.cache_config.max_keys,
-            cache_ttl_seconds: config.cache_config.ttl.as_secs(),
+            cache_ttl_seconds,
             cache_summary,
             backend_summary,
         }
@@ -514,9 +519,9 @@ impl ConfigureLocalKmsRequest {
             retry_attempts: self.retry_attempts.unwrap_or(3),
             enable_cache: self.enable_cache.unwrap_or(true),
             cache_config: CacheConfig {
-                max_keys: self.max_cached_keys.unwrap_or(1000),
-                ttl: Duration::from_secs(self.cache_ttl_seconds.unwrap_or(3600)),
-                enable_metrics: true,
+                max_keys: self.max_cached_keys.unwrap_or(DEFAULT_MAX_CACHED_KEYS),
+                ttl: self.cache_ttl_seconds.map_or(DEFAULT_CACHE_TTL, Duration::from_secs),
+                ..CacheConfig::default()
             },
         }
     }
@@ -555,9 +560,9 @@ impl ConfigureVaultKmsRequest {
             retry_attempts: self.retry_attempts.unwrap_or(3),
             enable_cache: self.enable_cache.unwrap_or(true),
             cache_config: CacheConfig {
-                max_keys: self.max_cached_keys.unwrap_or(1000),
-                ttl: Duration::from_secs(self.cache_ttl_seconds.unwrap_or(3600)),
-                enable_metrics: true,
+                max_keys: self.max_cached_keys.unwrap_or(DEFAULT_MAX_CACHED_KEYS),
+                ttl: self.cache_ttl_seconds.map_or(DEFAULT_CACHE_TTL, Duration::from_secs),
+                ..CacheConfig::default()
             },
         }
     }
@@ -596,9 +601,9 @@ impl ConfigureVaultTransitKmsRequest {
             retry_attempts: self.retry_attempts.unwrap_or(3),
             enable_cache: self.enable_cache.unwrap_or(true),
             cache_config: CacheConfig {
-                max_keys: self.max_cached_keys.unwrap_or(1000),
-                ttl: Duration::from_secs(self.cache_ttl_seconds.unwrap_or(3600)),
-                enable_metrics: true,
+                max_keys: self.max_cached_keys.unwrap_or(DEFAULT_MAX_CACHED_KEYS),
+                ttl: self.cache_ttl_seconds.map_or(DEFAULT_CACHE_TTL, Duration::from_secs),
+                ..CacheConfig::default()
             },
         }
     }
@@ -623,9 +628,9 @@ impl ConfigureStaticKmsRequest {
             retry_attempts: self.retry_attempts.unwrap_or(3),
             enable_cache: self.enable_cache.unwrap_or(true),
             cache_config: CacheConfig {
-                max_keys: self.max_cached_keys.unwrap_or(1000),
-                ttl: Duration::from_secs(self.cache_ttl_seconds.unwrap_or(3600)),
-                enable_metrics: true,
+                max_keys: self.max_cached_keys.unwrap_or(DEFAULT_MAX_CACHED_KEYS),
+                ttl: self.cache_ttl_seconds.map_or(DEFAULT_CACHE_TTL, Duration::from_secs),
+                ..CacheConfig::default()
             },
         }
     }
@@ -886,8 +891,8 @@ mod tests {
         assert_eq!(summary.backend_type, KmsBackend::VaultTransit);
         assert_eq!(summary.timeout_seconds, 30);
         assert_eq!(summary.retry_attempts, 3);
-        assert_eq!(summary.max_cached_keys, 1000);
-        assert_eq!(summary.cache_ttl_seconds, 3600);
+        assert_eq!(summary.max_cached_keys, DEFAULT_MAX_CACHED_KEYS);
+        assert_eq!(summary.cache_ttl_seconds, DEFAULT_CACHE_TTL.as_secs());
 
         match summary.backend_summary {
             BackendSummary::VaultTransit {

crates/kms/src/backends/vault_transit.rs

@@ -59,7 +59,14 @@ const METADATA_CAS_ATTEMPTS: usize = 3;
 /// TTL bound on cached metadata records. This caps how long one node can keep
 /// acting on lifecycle state another node has since changed (disable,
 /// schedule-deletion): the divergence window is one TTL instead of "until
-/// process restart". Matches the manager-level `KmsCache` TTL.
+/// process restart".
+///
+/// Deliberately fixed rather than derived from `CacheConfig`: this cache gates
+/// cryptographic operations through `ensure_key_state_allows`, so its staleness
+/// window must not follow a knob an operator turns to tune the manager-level
+/// describe cache. It happens to equal `config::DEFAULT_CACHE_TTL` today, but
+/// that is a coincidence rather than a contract, and binding the two would let
+/// a later change to the operator-facing default silently widen this window.
 const METADATA_CACHE_TTL: Duration = Duration::from_secs(300);
 
 /// Capacity bound on the metadata cache so an unbounded key namespace cannot

crates/kms/src/cache.rs

@@ -14,15 +14,12 @@
 
 //! Caching layer for KMS operations to improve performance
 
+use crate::config::CacheConfig;
 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
@@ -31,6 +28,10 @@ const DEFAULT_METADATA_TTL: Duration = Duration::from_secs(300);
 // 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.
+//
+// Publication is gated by `CacheConfig::enable_metrics`; the atomics backing
+// `KmsCache::stats` are maintained regardless, so the admin status API keeps
+// reporting real numbers with the metrics switch off.
 // ---------------------------------------------------------------------------
 
 /// Counter: key metadata lookups, by `result` (`hit` or `miss`).
@@ -94,38 +95,44 @@ pub struct KmsCacheStats {
 pub struct KmsCache {
     key_metadata_cache: Cache<String, KeyMetadata>,
     counters: Arc<CacheCounters>,
+    metrics_enabled: bool,
 }
 
 impl KmsCache {
-    /// Create a new KMS cache with the specified capacity
+    /// Create a new KMS cache from the operator-supplied cache configuration
+    ///
+    /// The entry lifetime is [`CacheConfig::effective_ttl`] rather than the raw
+    /// configured value: this value arrives straight from the admin configure
+    /// API, and moka panics when built with a time-to-live beyond 1000 years.
     ///
     /// # Arguments
-    /// * `capacity` - Maximum number of entries in the cache
+    /// * `config` - Capacity, metadata lifetime and metrics switch to build with
     ///
     /// # Returns
     /// 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();
+    pub fn new(config: &CacheConfig) -> Self {
+        let metrics_enabled = config.enable_metrics;
+        if metrics_enabled {
+            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(metadata_ttl)
+                .max_capacity(config.max_keys as u64)
+                .time_to_live(config.effective_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);
+                    if metrics_enabled {
+                        metrics::counter!(METRIC_CACHE_EVICTIONS_TOTAL, "cause" => removal_cause_label(cause)).increment(1);
+                    }
                 })
                 .build(),
             counters,
+            metrics_enabled,
         }
     }
 
@@ -216,11 +223,15 @@ impl KmsCache {
             (&self.counters.misses, "miss")
         };
         counter.fetch_add(1, Ordering::Relaxed);
-        metrics::counter!(METRIC_CACHE_LOOKUPS_TOTAL, "result" => result).increment(1);
+        if self.metrics_enabled {
+            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);
+        if self.metrics_enabled {
+            metrics::gauge!(METRIC_CACHE_ENTRIES).set(self.key_metadata_cache.entry_count() as f64);
+        }
     }
 }
 
@@ -239,6 +250,14 @@ mod tests {
         }
     }
 
+    /// A cache holding `max_keys` entries for the default lifetime.
+    fn sized(max_keys: usize) -> KmsCache {
+        KmsCache::new(&CacheConfig {
+            max_keys,
+            ..Default::default()
+        })
+    }
+
     fn test_metadata(key_id: &str) -> KeyMetadata {
         KeyMetadata {
             key_id: key_id.to_string(),
@@ -315,7 +334,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_cache_operations() {
-        let mut cache = KmsCache::new(100);
+        let mut cache = sized(100);
 
         // Test key metadata caching
         let metadata = KeyMetadata {
@@ -346,10 +365,11 @@ mod tests {
 
     #[tokio::test]
     async fn test_cache_with_custom_ttl() {
-        let mut cache = KmsCache::with_ttl(
-            100,
-            Duration::from_millis(100), // Short TTL for testing
-        );
+        let mut cache = KmsCache::new(&CacheConfig {
+            max_keys: 100,
+            ttl: Duration::from_millis(100), // Short TTL for testing
+            ..Default::default()
+        });
 
         let metadata = KeyMetadata {
             key_id: "ttl-test-key".to_string(),
@@ -377,7 +397,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_cache_contains_methods() {
-        let mut cache = KmsCache::new(100);
+        let mut cache = sized(100);
 
         assert!(!cache.contains_key_metadata_for_tests("nonexistent"));
 
@@ -401,7 +421,7 @@ mod tests {
     #[test]
     fn lookups_report_real_hit_and_miss_counts() {
         let (stats, snapshot) = record_metrics(|| async {
-            let mut cache = KmsCache::new(100);
+            let mut cache = sized(100);
 
             assert!(cache.get_key_metadata("absent").await.is_none());
             cache.put_key_metadata("present", &test_metadata("present")).await;
@@ -417,10 +437,39 @@ mod tests {
         assert_eq!(counter_value(&snapshot, METRIC_CACHE_LOOKUPS_TOTAL, &[("result", "miss")]), 2);
     }
 
+    #[test]
+    fn disabling_metrics_stops_publication_without_blinding_the_status_api() {
+        let (stats, snapshot) = record_metrics(|| async {
+            let mut cache = KmsCache::new(&CacheConfig {
+                max_keys: 1,
+                enable_metrics: false,
+                ..Default::default()
+            });
+
+            assert!(cache.get_key_metadata("absent").await.is_none());
+            cache.put_key_metadata("first", &test_metadata("first")).await;
+            assert!(cache.get_key_metadata("first").await.is_some());
+            // Capacity is one, so this insert evicts "first".
+            cache.put_key_metadata("second", &test_metadata("second")).await;
+
+            cache.stats()
+        });
+
+        // The counters behind the admin status API keep moving...
+        assert_eq!(stats.hits, 1);
+        assert_eq!(stats.misses, 1);
+        assert_eq!(stats.evictions, 1);
+
+        // ...while nothing reaches the metrics recorder.
+        assert_eq!(counter_value(&snapshot, METRIC_CACHE_LOOKUPS_TOTAL, &[]), 0);
+        assert_eq!(counter_value(&snapshot, METRIC_CACHE_EVICTIONS_TOTAL, &[]), 0);
+        assert_eq!(gauge_value(&snapshot, METRIC_CACHE_ENTRIES), None);
+    }
+
     #[test]
     fn removals_report_their_cause_and_the_resulting_entry_count() {
         let (stats, snapshot) = record_metrics(|| async {
-            let mut cache = KmsCache::new(100);
+            let mut cache = sized(100);
 
             cache.put_key_metadata("key", &test_metadata("key")).await;
             cache.put_key_metadata("key", &test_metadata("key")).await;
@@ -439,7 +488,7 @@ mod tests {
     #[test]
     fn capacity_pressure_reports_size_evictions() {
         let (stats, snapshot) = record_metrics(|| async {
-            let mut cache = KmsCache::with_ttl(1, DEFAULT_METADATA_TTL);
+            let mut cache = sized(1);
 
             cache.put_key_metadata("first", &test_metadata("first")).await;
             cache.put_key_metadata("second", &test_metadata("second")).await;
@@ -466,7 +515,11 @@ mod tests {
         let ttl = Duration::from_millis(100);
 
         let (stats, snapshot) = record_metrics(move || async move {
-            let mut cache = KmsCache::with_ttl(100, ttl);
+            let mut cache = KmsCache::new(&CacheConfig {
+                max_keys: 100,
+                ttl,
+                ..Default::default()
+            });
 
             cache.put_key_metadata("expiring", &test_metadata("expiring")).await;
             assert_eq!(cache.stats().entries, 1);

crates/kms/src/config.rs

@@ -50,6 +50,19 @@ pub(crate) const MAX_OPERATION_TIMEOUT: Duration = Duration::from_secs(300);
 /// Upper bound applied to `KmsConfig::retry_attempts` when deriving backend behavior.
 pub(crate) const MAX_RETRY_ATTEMPTS: u32 = 10;
 
+/// Default number of key metadata entries the cache holds.
+pub const DEFAULT_MAX_CACHED_KEYS: usize = 1000;
+
+/// Default lifetime of a cached key metadata entry.
+pub const DEFAULT_CACHE_TTL: Duration = Duration::from_secs(300);
+
+/// Upper bound applied to `CacheConfig::ttl` when building the metadata cache.
+///
+/// Out-of-range values are clamped at use rather than rejected, matching
+/// `MAX_OPERATION_TIMEOUT`, so existing deployments with an oversized setting
+/// keep starting after an upgrade.
+pub(crate) const MAX_CACHE_TTL: Duration = Duration::from_secs(24 * 60 * 60);
+
 fn default_vault_transit_metadata_kv_mount() -> String {
     DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT.to_string()
 }
@@ -555,22 +568,40 @@ pub struct TlsConfig {
 pub struct CacheConfig {
     /// Maximum number of keys to cache
     pub max_keys: usize,
-    /// TTL for cached keys
+    /// Lifetime of a cached key metadata entry.
+    ///
+    /// This bounds how long a describe can answer from metadata that another
+    /// node has since changed (disable, schedule-deletion); encrypt, decrypt
+    /// and data key generation never read the cache. Values above 24 hours are
+    /// clamped at use (see [`CacheConfig::effective_ttl`]).
     pub ttl: Duration,
-    /// Enable cache metrics
+    /// Publish the `rustfs_kms_metadata_cache_*` metrics.
+    ///
+    /// Only metrics-recorder output is gated: the counters behind the admin
+    /// status API are maintained either way.
     pub enable_metrics: bool,
 }
 
 impl Default for CacheConfig {
     fn default() -> Self {
         Self {
-            max_keys: 1000,
-            ttl: Duration::from_secs(3600), // 1 hour
+            max_keys: DEFAULT_MAX_CACHED_KEYS,
+            ttl: DEFAULT_CACHE_TTL,
             enable_metrics: true,
         }
     }
 }
 
+impl CacheConfig {
+    /// Metadata lifetime with the configured value clamped to the supported maximum.
+    ///
+    /// This is the value the cache is built with and the value reported back to
+    /// operators, so what the admin API advertises is what the cache does.
+    pub fn effective_ttl(&self) -> Duration {
+        self.ttl.min(MAX_CACHE_TTL)
+    }
+}
+
 /// AWS KMS backend configuration.
 ///
 /// Deliberately holds no credential material: the backend resolves credentials
@@ -906,8 +937,17 @@ impl KmsConfig {
         }
 
         // Validate cache configuration
-        if self.enable_cache && self.cache_config.max_keys == 0 {
-            return Err(KmsError::configuration_error("Cache max_keys must be greater than 0"));
+        if self.enable_cache {
+            if self.cache_config.max_keys == 0 {
+                return Err(KmsError::configuration_error("Cache max_keys must be greater than 0"));
+            }
+
+            // A zero TTL expires every entry on insert, which is a cache that
+            // cannot serve anything; reject it instead of silently disabling
+            // the cache the operator asked for.
+            if self.cache_config.ttl.is_zero() {
+                return Err(KmsError::configuration_error("Cache ttl must be greater than 0"));
+            }
         }
 
         Ok(())
@@ -1238,6 +1278,60 @@ mod tests {
         assert_eq!(local_config.key_dir, temp_dir.path());
     }
 
+    #[test]
+    fn oversized_cache_ttl_is_clamped_not_rejected() {
+        let temp_dir = TempDir::new().expect("Failed to create temp dir");
+        let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
+
+        assert_eq!(config.cache_config.ttl, DEFAULT_CACHE_TTL);
+        assert_eq!(config.cache_config.effective_ttl(), DEFAULT_CACHE_TTL);
+
+        // An oversized lifetime must not keep the service from starting, and
+        // must not reach moka, whose builder panics beyond 1000 years.
+        let config = KmsConfig {
+            cache_config: CacheConfig {
+                ttl: Duration::from_secs(u64::MAX),
+                ..Default::default()
+            },
+            ..config
+        };
+        assert!(config.validate().is_ok());
+        assert_eq!(config.cache_config.effective_ttl(), MAX_CACHE_TTL);
+
+        // In-range values pass through unchanged.
+        let config = KmsConfig {
+            cache_config: CacheConfig {
+                ttl: Duration::from_secs(600),
+                ..Default::default()
+            },
+            ..config
+        };
+        assert!(config.validate().is_ok());
+        assert_eq!(config.cache_config.effective_ttl(), Duration::from_secs(600));
+    }
+
+    #[test]
+    fn zero_cache_ttl_is_rejected_only_while_caching_is_enabled() {
+        let temp_dir = TempDir::new().expect("Failed to create temp dir");
+        let config = KmsConfig {
+            cache_config: CacheConfig {
+                ttl: Duration::ZERO,
+                ..Default::default()
+            },
+            ..KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults()
+        };
+
+        // A cache that expires every entry on insert is a misconfiguration, not
+        // a way to turn caching off.
+        assert!(config.validate().is_err());
+
+        let config = KmsConfig {
+            enable_cache: false,
+            ..config
+        };
+        assert!(config.validate().is_ok());
+    }
+
     #[test]
     fn test_oversized_timeout_and_retries_clamped_not_rejected() {
         let temp_dir = TempDir::new().expect("Failed to create temp dir");

crates/kms/src/manager.rs

@@ -46,7 +46,7 @@ pub struct KmsManager {
 impl KmsManager {
     /// Create a new KMS manager with the given backend and config
     pub fn new(backend: Arc<dyn KmsBackend>, config: KmsConfig) -> Self {
-        let cache = Arc::new(RwLock::new(KmsCache::new(config.cache_config.max_keys as u64)));
+        let cache = Arc::new(RwLock::new(KmsCache::new(&config.cache_config)));
         if config.allow_immediate_deletion {
             warn!(
                 "KMS immediate key deletion is enabled: a DeleteKey request may destroy key material without any waiting window, and every object encrypted under that key becomes permanently unreadable"
@@ -462,6 +462,7 @@ mod tests {
     use jiff::Zoned;
     use std::collections::HashMap;
     use std::sync::Mutex;
+    use std::time::Duration;
     use tempfile::tempdir;
 
     /// Sink that keeps every record so tests can assert on the audit trail.
@@ -971,6 +972,45 @@ mod tests {
         assert!(health);
     }
 
+    #[tokio::test]
+    async fn configured_cache_ttl_bounds_how_long_metadata_is_reused() {
+        let temp_dir = tempdir().expect("Failed to create temp dir");
+        let mut config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
+        config.cache_config.ttl = Duration::from_millis(100);
+
+        let backend = Arc::new(LocalKmsBackend::new(config.clone()).await.expect("Failed to create backend"));
+        let manager = KmsManager::new(backend, config);
+
+        let key_id = manager
+            .create_key(CreateKeyRequest {
+                key_name: Some("cache-ttl-wiring".to_string()),
+                ..Default::default()
+            })
+            .await
+            .expect("Failed to create key")
+            .key_id;
+
+        // Creating the key populated the cache, so this describe is served from it.
+        manager
+            .describe_key(DescribeKeyRequest { key_id: key_id.clone() })
+            .await
+            .expect("describe should succeed");
+        assert_eq!(manager.cache_stats().await.expect("cache is enabled").hits, 1);
+
+        // Past the configured lifetime the entry is gone and the describe falls
+        // through to the backend. A cache built with a hardcoded lifetime would
+        // still be serving the entry here.
+        tokio::time::sleep(Duration::from_millis(150)).await;
+        manager
+            .describe_key(DescribeKeyRequest { key_id })
+            .await
+            .expect("describe should succeed");
+
+        let stats = manager.cache_stats().await.expect("cache is enabled");
+        assert_eq!(stats.hits, 1);
+        assert_eq!(stats.misses, 1);
+    }
+
     #[tokio::test]
     async fn lifecycle_round_trip_invalidates_cached_metadata() {
         let temp_dir = tempdir().expect("Failed to create temp dir");

crates/kms/src/snapshots/rustfs_kms__api_types__tests__kms_vault_transit_config_summary.snap

@@ -16,9 +16,9 @@ expression: "serde_json::to_value(&summary).expect(\"KMS config summary should s
   "cache_summary": {
     "enable_metrics": true,
     "max_keys": 1000,
-    "ttl_seconds": 3600
+    "ttl_seconds": 300
   },
-  "cache_ttl_seconds": 3600,
+  "cache_ttl_seconds": 300,
   "default_key_id": "rustfs-master-key",
   "enable_cache": true,
   "max_cached_keys": 1000,

rustfs/src/admin/handlers/kms_management.rs

@@ -344,7 +344,7 @@ impl Operation for KmsConfigHandler {
             backend: backend_name(&config.backend).to_string(),
             cache_enabled: config.enable_cache,
             cache_max_keys: config.cache_config.max_keys,
-            cache_ttl_seconds: config.cache_config.ttl.as_secs(),
+            cache_ttl_seconds: config.cache_config.effective_ttl().as_secs(),
             default_key_id: service.get_default_key_id().cloned(),
         };