feat(kms): persist deletion deadlines and run a restartable deletion worker

Persist the scheduled deletion deadline in the Local (StoredMasterKey /
MasterKeyInfo) and Vault KV2 (VaultKeyData) records as an optional,
serde(default) field so old records keep deserializing, surface it from
describe_key, and clear it on cancel. Previously both backends lost the
deadline on restart and KV2 describe always reported none.

Add KmsBackend::remove_expired_key: a race-safe removal primitive that
re-checks state and deadline under the backend's own synchronization
(per-key write lock on Local), writes a tombstone (Deleted/Unavailable
record) before destroying material so a crashed removal can simply be
re-run, treats an already-removed key as success, and never auto-removes
pending records without a persisted deadline. The user-facing force
delete paths now write the same tombstone first.

Add a background deletion worker owned by each KmsServiceManager
service version: spawned at publish time when the backend supports
deletion scheduling, cancelled on stop/restart/reconfigure via a
CancellationToken, and sweeping every 60s for expired pending deletions
and leftover tombstones. Sweeps are stateless and idempotent, so they
are restart-safe and safe to run on every node concurrently without
leader election. Before removing anything the worker refuses keys that
are still the configured default key or that an injectable
DeletionReferenceChecker reports as referenced.

Refs rustfs/backlog#1571 (part of rustfs/backlog#1562)

crates/kms/src/backends/contract_tests.rs

@@ -0,0 +1,330 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Shared key state × operation contract tests for KMS backends.
+//!
+//! Every stateful backend must satisfy the same lifecycle matrix (see
+//! `ensure_key_state_permits`): Enabled permits everything, Disabled permits
+//! decryption and lifecycle recovery but rejects new cryptographic use, and
+//! PendingDeletion rejects everything except decryption and cancellation.
+//! Decryption staying available in Disabled/PendingDeletion is an explicit,
+//! tested deviation from AWS KMS: disabling a key must not break reads of
+//! objects already encrypted under it.
+//!
+//! The full matrix runs offline against the Local backend. The Vault KV2 and
+//! Vault Transit runs exercise the same helper but need a live Vault dev
+//! server, so they are `#[ignore]`d in CI. Static is covered by its own
+//! stateless contract below.
+
+use super::local::LocalKmsBackend;
+use super::static_kms::StaticKmsBackend;
+use super::vault::VaultKmsBackend;
+use super::vault_transit::VaultTransitKmsBackend;
+use super::{KmsBackend, KmsClient};
+use crate::config::KmsConfig;
+use crate::error::{KmsError, Result};
+use crate::manager::KmsManager;
+use crate::service::ObjectEncryptionService;
+use crate::types::{
+    CancelKeyDeletionRequest, CreateKeyRequest, DecryptRequest, DeleteKeyRequest, DescribeKeyRequest, EncryptRequest,
+    GenerateDataKeyRequest, KeySpec, KeyState, KeyUsage, ObjectEncryptionContext,
+};
+use base64::Engine as _;
+use base64::engine::general_purpose::STANDARD as BASE64;
+use rand::RngExt as _;
+use std::collections::HashMap;
+use std::sync::Arc;
+
+fn expect_invalid_key_state<T: std::fmt::Debug>(result: Result<T>, expected_fragment: &str) {
+    match result {
+        Err(KmsError::InvalidOperation { message }) => assert!(
+            message.contains(expected_fragment),
+            "expected invalid-key-state message containing {expected_fragment:?}, got {message:?}"
+        ),
+        other => panic!("expected InvalidOperation (invalid key state), got {other:?}"),
+    }
+}
+
+fn context() -> HashMap<String, String> {
+    HashMap::from([("bucket".to_string(), "contract".to_string())])
+}
+
+fn generate_request(key_id: &str) -> GenerateDataKeyRequest {
+    GenerateDataKeyRequest {
+        key_id: key_id.to_string(),
+        key_spec: KeySpec::Aes256,
+        encryption_context: context(),
+    }
+}
+
+fn encrypt_request(key_id: &str) -> EncryptRequest {
+    EncryptRequest {
+        key_id: key_id.to_string(),
+        plaintext: b"contract-plaintext".to_vec(),
+        encryption_context: context(),
+        grant_tokens: Vec::new(),
+    }
+}
+
+fn decrypt_request(ciphertext: Vec<u8>) -> DecryptRequest {
+    DecryptRequest {
+        ciphertext,
+        encryption_context: context(),
+        grant_tokens: Vec::new(),
+    }
+}
+
+fn schedule_request(key_id: &str) -> DeleteKeyRequest {
+    DeleteKeyRequest {
+        key_id: key_id.to_string(),
+        pending_window_in_days: Some(7),
+        force_immediate: None,
+    }
+}
+
+fn cancel_request(key_id: &str) -> CancelKeyDeletionRequest {
+    CancelKeyDeletionRequest {
+        key_id: key_id.to_string(),
+    }
+}
+
+fn create_request(key_name: String) -> CreateKeyRequest {
+    CreateKeyRequest {
+        key_name: Some(key_name),
+        key_usage: KeyUsage::EncryptDecrypt,
+        ..Default::default()
+    }
+}
+
+async fn assert_key_state(backend: &dyn KmsBackend, key_id: &str, expected: KeyState) {
+    let described = backend
+        .describe_key(DescribeKeyRequest {
+            key_id: key_id.to_string(),
+        })
+        .await
+        .expect("describe_key must succeed for an existing key");
+    assert_eq!(described.key_metadata.key_state, expected, "unexpected state for key {key_id}");
+}
+
+/// Drives one freshly created (Enabled) key through the full state matrix.
+///
+/// `backend` is the product surface; `client` drives the lifecycle
+/// transitions not yet exposed through `KmsBackend`.
+async fn assert_state_machine_contract(backend: &dyn KmsBackend, client: &dyn KmsClient, key_id: &str) {
+    // Enabled: cryptographic use is allowed. Keep an envelope around to prove
+    // decryption keeps working in later states.
+    let data_key = backend
+        .generate_data_key(generate_request(key_id))
+        .await
+        .expect("Enabled key must generate data keys");
+    backend
+        .encrypt(encrypt_request(key_id))
+        .await
+        .expect("Enabled key must encrypt");
+
+    // Enabled -> Disabled.
+    client
+        .disable_key(key_id, None)
+        .await
+        .expect("disable from Enabled must succeed");
+    assert_key_state(backend, key_id, KeyState::Disabled).await;
+
+    // Disabled: new cryptographic use and rotation are rejected...
+    expect_invalid_key_state(backend.encrypt(encrypt_request(key_id)).await, "disabled");
+    expect_invalid_key_state(backend.generate_data_key(generate_request(key_id)).await, "disabled");
+    expect_invalid_key_state(client.rotate_key(key_id, None).await, "");
+    // ...but decryption of existing data keeps working (explicit AWS deviation)...
+    let decrypted = backend
+        .decrypt(decrypt_request(data_key.ciphertext_blob.clone()))
+        .await
+        .expect("decrypt with a disabled key must keep working");
+    assert_eq!(decrypted.plaintext, data_key.plaintext_key, "decrypt must recover the original data key");
+    // ...disable stays idempotent, cancel has nothing to cancel, and enable recovers.
+    client.disable_key(key_id, None).await.expect("disable must be idempotent");
+    expect_invalid_key_state(backend.cancel_key_deletion(cancel_request(key_id)).await, "not pending deletion");
+    client
+        .enable_key(key_id, None)
+        .await
+        .expect("enable from Disabled must succeed");
+    assert_key_state(backend, key_id, KeyState::Enabled).await;
+
+    // Disabled keys may still be scheduled for deletion.
+    client
+        .disable_key(key_id, None)
+        .await
+        .expect("disable before scheduling must succeed");
+    backend
+        .delete_key(schedule_request(key_id))
+        .await
+        .expect("scheduling deletion of a disabled key must succeed");
+    assert_key_state(backend, key_id, KeyState::PendingDeletion).await;
+
+    // PendingDeletion: everything except decryption and cancellation is rejected.
+    expect_invalid_key_state(backend.encrypt(encrypt_request(key_id)).await, "pending deletion");
+    expect_invalid_key_state(backend.generate_data_key(generate_request(key_id)).await, "pending deletion");
+    expect_invalid_key_state(client.enable_key(key_id, None).await, "pending deletion");
+    expect_invalid_key_state(client.disable_key(key_id, None).await, "pending deletion");
+    expect_invalid_key_state(client.rotate_key(key_id, None).await, "");
+    expect_invalid_key_state(client.schedule_key_deletion(key_id, 7, None).await, "pending deletion");
+    expect_invalid_key_state(backend.delete_key(schedule_request(key_id)).await, "pending deletion");
+    let decrypted = backend
+        .decrypt(decrypt_request(data_key.ciphertext_blob.clone()))
+        .await
+        .expect("decrypt with a pending-deletion key must keep working");
+    assert_eq!(decrypted.plaintext, data_key.plaintext_key);
+
+    // PendingDeletion -> Enabled through cancellation.
+    backend
+        .cancel_key_deletion(cancel_request(key_id))
+        .await
+        .expect("cancel from PendingDeletion must succeed");
+    assert_key_state(backend, key_id, KeyState::Enabled).await;
+    backend
+        .generate_data_key(generate_request(key_id))
+        .await
+        .expect("cancelled key must be usable again");
+
+    // Cancel without a pending deletion is an invalid state transition.
+    expect_invalid_key_state(backend.cancel_key_deletion(cancel_request(key_id)).await, "not pending deletion");
+}
+
+async fn local_fixture() -> (tempfile::TempDir, KmsConfig, LocalKmsBackend, String) {
+    let temp_dir = tempfile::tempdir().expect("temp dir should be created");
+    let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
+    let backend = LocalKmsBackend::new(config.clone())
+        .await
+        .expect("local backend should build");
+    let created = backend
+        .create_key(create_request("contract-key".to_string()))
+        .await
+        .expect("key should be created");
+    (temp_dir, config, backend, created.key_id)
+}
+
+#[tokio::test]
+async fn local_backend_state_machine_contract() {
+    let (_temp_dir, _config, backend, key_id) = local_fixture().await;
+    assert_state_machine_contract(&backend, backend.lifecycle_client(), &key_id).await;
+}
+
+/// SSE-shaped regression: disabling a key must not break decryption of data
+/// keys created while it was enabled, while new data key creation must fail.
+#[tokio::test]
+async fn local_disabled_key_keeps_decrypting_existing_envelopes() {
+    let (_temp_dir, config, backend, key_id) = local_fixture().await;
+    let backend = Arc::new(backend);
+    let service = ObjectEncryptionService::new(KmsManager::new(backend.clone(), config));
+
+    let object_context = ObjectEncryptionContext::new("sse-bucket".to_string(), "dir/object.bin".to_string());
+    let kms_key = Some(key_id.clone());
+    let (_data_key, encrypted_blob) = service
+        .create_data_key(&kms_key, &object_context)
+        .await
+        .expect("data key creation must succeed while the key is enabled");
+
+    backend
+        .lifecycle_client()
+        .disable_key(&key_id, None)
+        .await
+        .expect("disable must succeed");
+
+    service
+        .decrypt_data_key(&encrypted_blob, &object_context)
+        .await
+        .expect("existing objects must stay readable after their KMS key is disabled");
+    expect_invalid_key_state(service.create_data_key(&kms_key, &object_context).await, "disabled");
+}
+
+/// Static is a stateless read-only backend: cryptographic operations always
+/// work against the single configured key and every lifecycle mutation is
+/// rejected as an invalid operation.
+#[tokio::test]
+async fn static_backend_stateless_contract() {
+    let key_id = "static-contract-key";
+    let mut raw_key = [0u8; 32];
+    rand::rng().fill(&mut raw_key[..]);
+    let config = KmsConfig::static_kms(key_id.to_string(), BASE64.encode(raw_key));
+    let static_backend = StaticKmsBackend::new(config).await.expect("static backend should build");
+    // StaticKmsBackend implements both traits with overlapping method names,
+    // so pin each surface once instead of qualifying every call.
+    let backend: &dyn KmsBackend = &static_backend;
+    let client: &dyn KmsClient = &static_backend;
+
+    let data_key = backend
+        .generate_data_key(generate_request(key_id))
+        .await
+        .expect("static backend must generate data keys");
+    let decrypted = backend
+        .decrypt(decrypt_request(data_key.ciphertext_blob.clone()))
+        .await
+        .expect("static backend must decrypt its own envelopes");
+    assert_eq!(decrypted.plaintext, data_key.plaintext_key);
+    assert_key_state(backend, key_id, KeyState::Enabled).await;
+
+    expect_invalid_key_state(backend.create_key(create_request("another-key".to_string())).await, "read-only");
+    expect_invalid_key_state(backend.delete_key(schedule_request(key_id)).await, "read-only");
+    expect_invalid_key_state(backend.cancel_key_deletion(cancel_request(key_id)).await, "read-only");
+    expect_invalid_key_state(client.disable_key(key_id, None).await, "read-only");
+    expect_invalid_key_state(client.schedule_key_deletion(key_id, 7, None).await, "read-only");
+    expect_invalid_key_state(client.rotate_key(key_id, None).await, "read-only");
+}
+
+fn vault_dev_config(constructor: fn(url::Url, String) -> KmsConfig) -> KmsConfig {
+    let address = std::env::var("RUSTFS_KMS_VAULT_ADDR").unwrap_or_else(|_| "http://127.0.0.1:8200".to_string());
+    let token = std::env::var("RUSTFS_KMS_VAULT_TOKEN").unwrap_or_else(|_| "dev-token".to_string());
+    let mut config = constructor(url::Url::parse(&address).expect("vault address should parse"), token);
+    config.allow_insecure_dev_defaults = true;
+    config
+}
+
+#[tokio::test]
+#[ignore] // Requires a running Vault instance (dev mode) with a KV2 mount
+async fn vault_kv2_backend_state_machine_contract() {
+    let config = vault_dev_config(KmsConfig::vault);
+    let backend = VaultKmsBackend::new(config).await.expect("vault kv2 backend should build");
+    let created = backend
+        .create_key(create_request(format!("contract-{}", uuid::Uuid::new_v4())))
+        .await
+        .expect("key should be created");
+
+    assert_state_machine_contract(&backend, backend.lifecycle_client(), &created.key_id).await;
+
+    // Cleanup: leave the key pending deletion so repeated runs stay tidy.
+    let _ = backend.delete_key(schedule_request(&created.key_id)).await;
+}
+
+#[tokio::test]
+#[ignore] // Requires a running Vault instance (dev mode) with the transit engine enabled
+async fn vault_transit_backend_state_machine_contract() {
+    let config = vault_dev_config(KmsConfig::vault_transit);
+    let backend = VaultTransitKmsBackend::new(config)
+        .await
+        .expect("vault transit backend should build");
+    let created = backend
+        .create_key(create_request(format!("contract-{}", uuid::Uuid::new_v4())))
+        .await
+        .expect("key should be created");
+
+    assert_state_machine_contract(&backend, backend.lifecycle_client(), &created.key_id).await;
+
+    // Transit additionally supports rotation, which must only work while the
+    // key is Enabled (the shared matrix already covered the rejections).
+    backend
+        .lifecycle_client()
+        .rotate_key(&created.key_id, None)
+        .await
+        .expect("rotation of an Enabled transit key must succeed");
+
+    let _ = backend.delete_key(schedule_request(&created.key_id)).await;
+}

crates/kms/src/backends/local.rs

@@ -14,7 +14,9 @@
 
 //! Local file-based KMS backend implementation
 
-use crate::backends::{BackendInfo, KmsBackend, KmsClient};
+use crate::backends::{
+    BackendCapabilities, BackendInfo, ExpiredKeyRemoval, KmsBackend, KmsClient, StateGatedOperation, ensure_key_status_permits,
+};
 use crate::config::KmsConfig;
 use crate::config::LocalConfig;
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
@@ -425,6 +427,10 @@ struct StoredMasterKey {
     #[serde(with = "crate::time_serde::option_zoned")]
     rotated_at: Option<Zoned>,
     created_by: Option<String>,
+    /// Scheduled deletion deadline; absent on records written before deadline
+    /// persistence landed, so it must stay optional for backward compatibility.
+    #[serde(default, with = "crate::time_serde::option_zoned")]
+    deletion_date: Option<Zoned>,
     /// Encrypted key material (32 bytes encoded in base64 for AES-256)
     encrypted_key_material: String,
     /// Nonce used for encryption
@@ -770,6 +776,7 @@ impl LocalKmsClient {
             created_at: stored_key.created_at,
             rotated_at: stored_key.rotated_at,
             created_by: stored_key.created_by,
+            deletion_date: stored_key.deletion_date,
         })
     }
 
@@ -843,6 +850,7 @@ impl LocalKmsClient {
             created_at: master_key.created_at.clone(),
             rotated_at: master_key.rotated_at.clone(),
             created_by: master_key.created_by.clone(),
+            deletion_date: master_key.deletion_date.clone(),
             encrypted_key_material,
             nonce,
             at_rest_protection,
@@ -931,9 +939,12 @@ impl LocalKmsClient {
 
 #[async_trait]
 impl KmsClient for LocalKmsClient {
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
+    async fn generate_data_key(&self, request: &GenerateKeyRequest, context: Option<&OperationContext>) -> Result<DataKeyInfo> {
         debug!("Generating data key for master key: {}", request.master_key_id);
 
+        let key_info = self.describe_key(&request.master_key_id, context).await?;
+        ensure_key_status_permits(&request.master_key_id, &key_info.status, StateGatedOperation::GenerateDataKey)?;
+
         // Generate random data key material
         let key_length = match request.key_spec.as_str() {
             "AES_256" => 32,
@@ -972,14 +983,9 @@ impl KmsClient for LocalKmsClient {
     async fn encrypt(&self, request: &EncryptRequest, context: Option<&OperationContext>) -> Result<EncryptResponse> {
         debug!("Encrypting data with key: {}", request.key_id);
 
-        // Verify key exists and is active
+        // Verify key exists and its state allows encryption
         let key_info = self.describe_key(&request.key_id, context).await?;
-        if key_info.status != KeyStatus::Active {
-            return Err(KmsError::invalid_operation(format!(
-                "Key {} is not active (status: {:?})",
-                request.key_id, key_info.status
-            )));
-        }
+        ensure_key_status_permits(&request.key_id, &key_info.status, StateGatedOperation::Encrypt)?;
 
         let (ciphertext, _nonce) = self.encrypt_with_master_key(&request.key_id, &request.plaintext).await?;
 
@@ -1110,6 +1116,7 @@ impl KmsClient for LocalKmsClient {
 
         let _write_guard = self.lock_key_for_write(key_id).await;
         let mut master_key = self.load_master_key(key_id).await?;
+        ensure_key_status_permits(key_id, &master_key.status, StateGatedOperation::Enable)?;
         master_key.status = KeyStatus::Active;
 
         // Preserve the existing key material. Regenerating it on a pure status change would
@@ -1127,6 +1134,7 @@ impl KmsClient for LocalKmsClient {
 
         let _write_guard = self.lock_key_for_write(key_id).await;
         let mut master_key = self.load_master_key(key_id).await?;
+        ensure_key_status_permits(key_id, &master_key.status, StateGatedOperation::Disable)?;
         master_key.status = KeyStatus::Disabled;
 
         // Preserve the existing key material (see enable_key): a status change must never
@@ -1141,14 +1149,16 @@ impl KmsClient for LocalKmsClient {
     async fn schedule_key_deletion(
         &self,
         key_id: &str,
-        _pending_window_days: u32,
+        pending_window_days: u32,
         _context: Option<&OperationContext>,
     ) -> Result<()> {
         debug!("Scheduling deletion for key: {}", key_id);
 
         let _write_guard = self.lock_key_for_write(key_id).await;
         let mut master_key = self.load_master_key(key_id).await?;
+        ensure_key_status_permits(key_id, &master_key.status, StateGatedOperation::ScheduleDeletion)?;
         master_key.status = KeyStatus::PendingDeletion;
+        master_key.deletion_date = Some(Zoned::now() + Duration::from_secs(pending_window_days as u64 * 86400));
 
         // Preserve the existing key material (see enable_key): scheduling deletion must not
         // regenerate the master key, or cancelling the deletion later would recover a key that
@@ -1165,7 +1175,11 @@ impl KmsClient for LocalKmsClient {
 
         let _write_guard = self.lock_key_for_write(key_id).await;
         let mut master_key = self.load_master_key(key_id).await?;
+        if master_key.status != KeyStatus::PendingDeletion {
+            return Err(KmsError::invalid_key_state(format!("Key {key_id} is not pending deletion")));
+        }
         master_key.status = KeyStatus::Active;
+        master_key.deletion_date = None;
 
         // Preserve the existing key material (see enable_key): cancelling deletion must recover
         // the ORIGINAL key, not mint a new one that cannot decrypt existing data.
@@ -1215,6 +1229,12 @@ pub struct LocalKmsBackend {
 }
 
 impl LocalKmsBackend {
+    /// Lifecycle driver for the shared state-machine contract tests.
+    #[cfg(test)]
+    pub(crate) fn lifecycle_client(&self) -> &LocalKmsClient {
+        &self.client
+    }
+
     /// Create a new LocalKmsBackend
     pub async fn new(config: KmsConfig) -> Result<Self> {
         config.validate()?;
@@ -1328,6 +1348,11 @@ impl KmsBackend for LocalKmsBackend {
 
     async fn describe_key(&self, request: DescribeKeyRequest) -> Result<DescribeKeyResponse> {
         let key_info = self.client.describe_key(&request.key_id, None).await?;
+        let deletion_date = if key_info.status == KeyStatus::PendingDeletion {
+            self.client.load_master_key(&request.key_id).await?.deletion_date
+        } else {
+            None
+        };
 
         let metadata = KeyMetadata {
             key_id: key_info.key_id,
@@ -1340,7 +1365,7 @@ impl KmsBackend for LocalKmsBackend {
             key_usage: key_info.usage,
             description: key_info.description,
             creation_date: key_info.created_at,
-            deletion_date: None,
+            deletion_date,
             origin: "KMS".to_string(),
             key_manager: "CUSTOMER".to_string(),
             tags: key_info.tags,
@@ -1371,7 +1396,22 @@ impl KmsBackend for LocalKmsBackend {
             .map_err(|_| KmsError::key_not_found(format!("Key {key_id} not found")))?;
 
         let (deletion_date_str, deletion_date_dt) = if request.force_immediate.unwrap_or(false) {
-            // For immediate deletion, actually delete the key from filesystem
+            // Tombstone first: mark the record Deleted before removing the
+            // file, so a crash between the two steps leaves a key that is
+            // already unusable and whose removal can simply be re-run.
+            match self.client.decode_stored_key(key_id).await {
+                Ok((_stored, key_material)) => {
+                    let mut tombstone = master_key.clone();
+                    tombstone.status = KeyStatus::Deleted;
+                    tombstone.deletion_date = Some(Zoned::now());
+                    self.client.save_master_key(&tombstone, &key_material).await?;
+                }
+                Err(error) => {
+                    // A record whose material can no longer be decoded cannot be
+                    // re-encrypted into a tombstone; proceed with the removal.
+                    warn!(key_id, %error, "skipping tombstone for undecodable key record");
+                }
+            }
             let key_path = self.client.master_key_path(key_id)?;
             durable_file::remove_durably(key_path)
                 .await
@@ -1399,6 +1439,8 @@ impl KmsBackend for LocalKmsBackend {
             });
         } else {
             // Schedule for deletion (default 30 days)
+            ensure_key_status_permits(key_id, &master_key.status, StateGatedOperation::ScheduleDeletion)?;
+
             let days = request.pending_window_in_days.unwrap_or(30);
             if !(7..=30).contains(&days) {
                 return Err(KmsError::invalid_parameter("pending_window_in_days must be between 7 and 30".to_string()));
@@ -1406,6 +1448,7 @@ impl KmsBackend for LocalKmsBackend {
 
             let deletion_date = Zoned::now() + Duration::from_secs(days as u64 * 86400);
             master_key.status = KeyStatus::PendingDeletion;
+            master_key.deletion_date = Some(deletion_date.clone());
 
             (Some(deletion_date.to_string()), Some(deletion_date))
         };
@@ -1459,6 +1502,7 @@ impl KmsBackend for LocalKmsBackend {
 
         // Cancel the deletion by resetting the state
         master_key.status = KeyStatus::Active;
+        master_key.deletion_date = None;
 
         // Save the updated key to disk - this is the missing critical step!
         // Preserve existing key material instead of generating new one
@@ -1492,6 +1536,59 @@ impl KmsBackend for LocalKmsBackend {
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
+
+    fn capabilities(&self) -> BackendCapabilities {
+        // Rotation stays unadvertised until historical key versions can be
+        // retained (see LocalKmsClient::rotate_key); without version history
+        // there is also no versioning capability.
+        BackendCapabilities::minimal()
+            .with_enable_disable(true)
+            .with_schedule_deletion(true)
+            .with_physical_delete(true)
+    }
+
+    async fn remove_expired_key(&self, key_id: &str, now: &Zoned) -> Result<ExpiredKeyRemoval> {
+        // The per-key write lock serializes this against a concurrent
+        // cancellation, closing the check-then-remove race.
+        let _write_guard = self.client.lock_key_for_write(key_id).await;
+
+        if !fs::try_exists(self.client.master_key_path(key_id)?).await? {
+            return Ok(ExpiredKeyRemoval::Removed);
+        }
+        let master_key = self.client.load_master_key(key_id).await?;
+        match master_key.status {
+            // Tombstone left by a crashed removal: complete it.
+            KeyStatus::Deleted => {}
+            KeyStatus::PendingDeletion => {
+                match &master_key.deletion_date {
+                    Some(deadline) if deadline <= now => {}
+                    // Not yet due, or a legacy record without a persisted
+                    // deadline — never auto-remove those.
+                    _ => return Ok(ExpiredKeyRemoval::NotExpired),
+                }
+                // Tombstone first (see delete_key): a crash between the state
+                // write and the file removal must leave an unusable record.
+                match self.client.decode_stored_key(key_id).await {
+                    Ok((_stored, key_material)) => {
+                        let mut tombstone = master_key.clone();
+                        tombstone.status = KeyStatus::Deleted;
+                        tombstone.deletion_date = Some(now.clone());
+                        self.client.save_master_key(&tombstone, &key_material).await?;
+                    }
+                    Err(error) => {
+                        warn!(key_id, %error, "skipping tombstone for undecodable key record");
+                    }
+                }
+            }
+            KeyStatus::Active | KeyStatus::Disabled => return Ok(ExpiredKeyRemoval::StateChanged),
+        }
+
+        durable_file::remove_durably(self.client.master_key_path(key_id)?)
+            .await
+            .map_err(|e| KmsError::internal_error(format!("Failed to delete key file: {e}")))?;
+        debug!(key_id, "Local KMS expired key removed");
+        Ok(ExpiredKeyRemoval::Removed)
+    }
 }
 
 #[cfg(test)]
@@ -2606,9 +2703,16 @@ mod tests {
             client.schedule_key_deletion(key_id, 7, None),
             client.enable_key(key_id, None),
         );
-        disable.expect("disable");
-        schedule.expect("schedule deletion");
-        enable.expect("enable");
+        // The per-key lock serializes the three transitions in an arbitrary
+        // order, and the state gate may legitimately reject a transition that
+        // lost the race (e.g. enable after deletion was scheduled). Any other
+        // error kind would still mean corrupted storage.
+        for result in [disable, schedule, enable] {
+            match result {
+                Ok(()) | Err(KmsError::InvalidOperation { .. }) => {}
+                Err(other) => panic!("concurrent transition must only fail with a state rejection, got {other:?}"),
+            }
+        }
 
         // Whatever the serialization order, the file must be one writer's
         // complete output with the original material intact.
@@ -2623,4 +2727,67 @@ mod tests {
             "concurrent status updates must never lose or regenerate key material"
         );
     }
+
+    /// Records written before deadline persistence landed have no
+    /// deletion_date field and must keep deserializing (as None).
+    #[tokio::test]
+    async fn stored_master_key_without_deletion_date_still_deserializes() {
+        let (client, _temp_dir) = create_test_client().await;
+        client.create_key("legacy-key", "AES_256", None).await.expect("create key");
+
+        let path = client.master_key_path("legacy-key").expect("key path");
+        let bytes = fs::read(&path).await.expect("read stored key");
+        let mut value: serde_json::Value = serde_json::from_slice(&bytes).expect("stored key must be JSON");
+        value
+            .as_object_mut()
+            .expect("stored key must be a JSON object")
+            .remove("deletion_date")
+            .expect("current records must carry the field");
+
+        let stored: StoredMasterKey = serde_json::from_value(value).expect("legacy record must deserialize");
+        assert!(stored.deletion_date.is_none());
+    }
+
+    #[tokio::test]
+    async fn remove_expired_key_completes_a_tombstone_and_stays_idempotent() {
+        let temp_dir = TempDir::new().expect("temp dir");
+        let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
+        let backend = LocalKmsBackend::new(config).await.expect("backend");
+        let created = backend
+            .create_key(CreateKeyRequest {
+                key_name: Some("tombstoned-key".to_string()),
+                key_usage: KeyUsage::EncryptDecrypt,
+                ..Default::default()
+            })
+            .await
+            .expect("create key");
+        let key_id = created.key_id;
+
+        // Craft the state a removal crashed in: tombstone written, file not
+        // yet removed.
+        let client = backend.lifecycle_client();
+        let (_stored, key_material) = client.decode_stored_key(&key_id).await.expect("decode stored key");
+        let mut tombstone = client.load_master_key(&key_id).await.expect("load key");
+        tombstone.status = KeyStatus::Deleted;
+        tombstone.deletion_date = Some(Zoned::now());
+        client
+            .save_master_key(&tombstone, &key_material)
+            .await
+            .expect("write tombstone");
+
+        // The sweep primitive completes the crashed removal...
+        let outcome = backend
+            .remove_expired_key(&key_id, &Zoned::now())
+            .await
+            .expect("tombstone completion");
+        assert_eq!(outcome, crate::backends::ExpiredKeyRemoval::Removed);
+        assert!(!client.master_key_path(&key_id).expect("key path").exists());
+
+        // ...and stays idempotent once the key is gone.
+        let outcome = backend
+            .remove_expired_key(&key_id, &Zoned::now())
+            .await
+            .expect("repeat removal");
+        assert_eq!(outcome, crate::backends::ExpiredKeyRemoval::Removed);
+    }
 }

crates/kms/src/backends/mod.rs

@@ -14,17 +14,89 @@
 
 //! KMS backend implementations
 
-use crate::error::Result;
+use crate::error::{KmsError, Result};
 use crate::types::*;
 use async_trait::async_trait;
+use jiff::Zoned;
+use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 
+#[cfg(test)]
+mod contract_tests;
 pub mod local;
 pub mod static_kms;
 pub mod vault;
 pub(crate) mod vault_credentials;
 pub mod vault_transit;
 
+/// Operations whose availability depends on the key's lifecycle state.
+///
+/// Decryption is deliberately absent: RustFS allows decryption with
+/// `Disabled` and `PendingDeletion` keys — an explicit deviation from AWS
+/// KMS — because rejecting it would break reads of every object encrypted
+/// under a key the moment it is disabled. Deletion cancellation is also
+/// absent: it is valid exactly when the key is `PendingDeletion`, which call
+/// sites enforce directly.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum StateGatedOperation {
+    Encrypt,
+    GenerateDataKey,
+    Rotate,
+    Enable,
+    Disable,
+    ScheduleDeletion,
+}
+
+impl StateGatedOperation {
+    fn describe(self) -> &'static str {
+        match self {
+            Self::Encrypt => "encryption",
+            Self::GenerateDataKey => "data key generation",
+            Self::Rotate => "rotation",
+            Self::Enable => "enabling",
+            Self::Disable => "disabling",
+            Self::ScheduleDeletion => "deletion scheduling",
+        }
+    }
+}
+
+/// Enforce the shared key state × operation matrix.
+///
+/// - `Enabled`: every operation is allowed.
+/// - `Disabled`: enabling, disabling (idempotent) and deletion scheduling are
+///   allowed; encryption, data key generation and rotation are rejected.
+/// - `PendingDeletion`: every state-gated operation is rejected, including a
+///   repeated deletion schedule; only cancellation and decryption proceed.
+/// - `PendingImport`/`Unavailable`: the key is not usable and is reported as
+///   not found.
+pub(crate) fn ensure_key_state_permits(key_id: &str, state: &KeyState, operation: StateGatedOperation) -> Result<()> {
+    match state {
+        KeyState::Enabled => Ok(()),
+        KeyState::Disabled => match operation {
+            StateGatedOperation::Enable | StateGatedOperation::Disable | StateGatedOperation::ScheduleDeletion => Ok(()),
+            StateGatedOperation::Encrypt | StateGatedOperation::GenerateDataKey | StateGatedOperation::Rotate => Err(
+                KmsError::invalid_key_state(format!("Key {key_id} is disabled: {} is not allowed", operation.describe())),
+            ),
+        },
+        KeyState::PendingDeletion => Err(KmsError::invalid_key_state(format!(
+            "Key {key_id} is pending deletion: {} is not allowed",
+            operation.describe()
+        ))),
+        KeyState::PendingImport | KeyState::Unavailable => Err(KmsError::key_not_found(key_id)),
+    }
+}
+
+/// [`ensure_key_state_permits`] for backends that persist [`KeyStatus`].
+pub(crate) fn ensure_key_status_permits(key_id: &str, status: &KeyStatus, operation: StateGatedOperation) -> Result<()> {
+    let state = match status {
+        KeyStatus::Active => KeyState::Enabled,
+        KeyStatus::Disabled => KeyState::Disabled,
+        KeyStatus::PendingDeletion => KeyState::PendingDeletion,
+        KeyStatus::Deleted => KeyState::Unavailable,
+    };
+    ensure_key_state_permits(key_id, &state, operation)
+}
+
 /// Abstract KMS client interface that all backends must implement
 #[async_trait]
 pub trait KmsClient: Send + Sync {
@@ -184,6 +256,45 @@ pub trait KmsBackend: Send + Sync {
 
     /// Health check
     async fn health_check(&self) -> Result<bool>;
+
+    /// Report which operations this backend actually supports.
+    ///
+    /// The default is conservative: only the operations every backend is
+    /// required to implement by this trait are advertised. Optional lifecycle
+    /// operations (rotation, enable/disable, deletion scheduling, ...) must be
+    /// opted in by overriding this method.
+    fn capabilities(&self) -> BackendCapabilities {
+        BackendCapabilities::minimal()
+    }
+
+    /// Remove a key whose scheduled deletion deadline has passed.
+    ///
+    /// Used by the background deletion worker. Implementations must re-check
+    /// state and deadline under their own write synchronization so that a
+    /// concurrent cancellation observed after the caller's inspection wins
+    /// ([`ExpiredKeyRemoval::StateChanged`]), must write a tombstone (a
+    /// `Deleted`/`Unavailable` record) before destroying material so a crashed
+    /// removal can simply be re-run, and must treat an already-removed key as
+    /// success so the operation stays idempotent across restarts and nodes.
+    ///
+    /// The default rejects the operation for backends without deletion
+    /// support.
+    async fn remove_expired_key(&self, _key_id: &str, _now: &Zoned) -> Result<ExpiredKeyRemoval> {
+        Err(KmsError::unsupported_capability("backend without deletion support", "remove_expired_key"))
+    }
+}
+
+/// Outcome of [`KmsBackend::remove_expired_key`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ExpiredKeyRemoval {
+    /// The key's record and material were removed, or were already gone.
+    Removed,
+    /// The key is no longer pending deletion (for example the deletion was
+    /// cancelled after the caller inspected it); nothing was removed.
+    StateChanged,
+    /// The key is pending deletion but its deadline has not passed, or it has
+    /// no persisted deadline (legacy record) and is never auto-removed.
+    NotExpired,
 }
 
 /// Information about a KMS backend
@@ -237,3 +348,237 @@ impl BackendInfo {
         self
     }
 }
+
+/// Set of operations a KMS backend supports.
+///
+/// Reported by [`KmsBackend::capabilities`] so callers (manager, admin API)
+/// can discover what the active backend can do without probing individual
+/// operations. Marked `#[non_exhaustive]` so new capability flags can be
+/// added without breaking downstream code; construct values through
+/// [`BackendCapabilities::minimal`] and the `with_*` builders.
+#[non_exhaustive]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub struct BackendCapabilities {
+    /// Direct encryption of caller-provided plaintext with a master key
+    pub encrypt: bool,
+    /// Decryption of previously produced ciphertext
+    pub decrypt: bool,
+    /// Data encryption key (DEK) generation
+    pub generate_data_key: bool,
+    /// Key rotation that retains prior versions for decryption
+    pub rotate: bool,
+    /// Enabling and disabling keys
+    pub enable_disable: bool,
+    /// Scheduling key deletion with a pending window
+    pub schedule_deletion: bool,
+    /// Multiple key versions addressable after rotation
+    pub versioning: bool,
+    /// Irreversible physical deletion of key material
+    pub physical_delete: bool,
+}
+
+impl BackendCapabilities {
+    /// Conservative baseline: only the operations that every [`KmsBackend`]
+    /// implementation is required to provide by the trait. All optional
+    /// lifecycle capabilities default to unsupported.
+    pub const fn minimal() -> Self {
+        Self {
+            encrypt: true,
+            decrypt: true,
+            generate_data_key: true,
+            rotate: false,
+            enable_disable: false,
+            schedule_deletion: false,
+            versioning: false,
+            physical_delete: false,
+        }
+    }
+
+    /// Set whether direct encryption is supported
+    pub const fn with_encrypt(mut self, encrypt: bool) -> Self {
+        self.encrypt = encrypt;
+        self
+    }
+
+    /// Set whether decryption is supported
+    pub const fn with_decrypt(mut self, decrypt: bool) -> Self {
+        self.decrypt = decrypt;
+        self
+    }
+
+    /// Set whether data key generation is supported
+    pub const fn with_generate_data_key(mut self, generate_data_key: bool) -> Self {
+        self.generate_data_key = generate_data_key;
+        self
+    }
+
+    /// Set whether version-retaining key rotation is supported
+    pub const fn with_rotate(mut self, rotate: bool) -> Self {
+        self.rotate = rotate;
+        self
+    }
+
+    /// Set whether enabling/disabling keys is supported
+    pub const fn with_enable_disable(mut self, enable_disable: bool) -> Self {
+        self.enable_disable = enable_disable;
+        self
+    }
+
+    /// Set whether scheduled deletion with a pending window is supported
+    pub const fn with_schedule_deletion(mut self, schedule_deletion: bool) -> Self {
+        self.schedule_deletion = schedule_deletion;
+        self
+    }
+
+    /// Set whether multiple key versions are supported
+    pub const fn with_versioning(mut self, versioning: bool) -> Self {
+        self.versioning = versioning;
+        self
+    }
+
+    /// Set whether physical deletion of key material is supported
+    pub const fn with_physical_delete(mut self, physical_delete: bool) -> Self {
+        self.physical_delete = physical_delete;
+        self
+    }
+}
+
+impl Default for BackendCapabilities {
+    fn default() -> Self {
+        Self::minimal()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::KmsConfig;
+    use base64::Engine as _;
+    use base64::engine::general_purpose::STANDARD as BASE64;
+
+    /// Backend that implements only the trait-mandated operations and relies
+    /// on the default `capabilities` implementation.
+    struct MinimalBackend;
+
+    #[async_trait]
+    impl KmsBackend for MinimalBackend {
+        async fn create_key(&self, _request: CreateKeyRequest) -> Result<CreateKeyResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn encrypt(&self, _request: EncryptRequest) -> Result<EncryptResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn decrypt(&self, _request: DecryptRequest) -> Result<DecryptResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn generate_data_key(&self, _request: GenerateDataKeyRequest) -> Result<GenerateDataKeyResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn describe_key(&self, _request: DescribeKeyRequest) -> Result<DescribeKeyResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn list_keys(&self, _request: ListKeysRequest) -> Result<ListKeysResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn delete_key(&self, _request: DeleteKeyRequest) -> Result<DeleteKeyResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn cancel_key_deletion(&self, _request: CancelKeyDeletionRequest) -> Result<CancelKeyDeletionResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn health_check(&self) -> Result<bool> {
+            Ok(true)
+        }
+    }
+
+    fn capabilities_snapshot(capabilities: BackendCapabilities) -> std::collections::BTreeMap<String, bool> {
+        serde_json::from_value(serde_json::to_value(capabilities).expect("capabilities should serialize"))
+            .expect("capabilities should deserialize into a flat bool map")
+    }
+
+    #[test]
+    fn default_capabilities_are_conservative() {
+        let capabilities = MinimalBackend.capabilities();
+        assert_eq!(capabilities, BackendCapabilities::minimal());
+        assert_eq!(capabilities, BackendCapabilities::default());
+
+        // The conservative baseline advertises only trait-mandated operations.
+        assert!(capabilities.encrypt);
+        assert!(capabilities.decrypt);
+        assert!(capabilities.generate_data_key);
+        assert!(!capabilities.rotate);
+        assert!(!capabilities.enable_disable);
+        assert!(!capabilities.schedule_deletion);
+        assert!(!capabilities.versioning);
+        assert!(!capabilities.physical_delete);
+    }
+
+    #[tokio::test]
+    async fn default_remove_expired_key_is_unsupported() {
+        let error = MinimalBackend
+            .remove_expired_key("any-key", &jiff::Zoned::now())
+            .await
+            .expect_err("backends without deletion support must reject expired-key removal");
+        assert!(
+            matches!(error, KmsError::UnsupportedCapability { .. }),
+            "expected UnsupportedCapability, got {error:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn local_backend_capabilities_golden() {
+        let temp_dir = tempfile::tempdir().expect("temp dir should be created");
+        let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
+        let backend = local::LocalKmsBackend::new(config).await.expect("local backend should build");
+
+        insta::assert_json_snapshot!("local_backend_capabilities", capabilities_snapshot(backend.capabilities()));
+    }
+
+    #[tokio::test]
+    async fn vault_kv2_backend_capabilities_golden() {
+        let config = KmsConfig::vault(
+            url::Url::parse("http://127.0.0.1:8200").expect("vault URL should parse"),
+            "dev-token".to_string(),
+        )
+        .with_insecure_development_defaults();
+        // Constructing the client performs no network I/O with token auth.
+        let backend = vault::VaultKmsBackend::new(config)
+            .await
+            .expect("vault kv2 backend should build");
+
+        insta::assert_json_snapshot!("vault_kv2_backend_capabilities", capabilities_snapshot(backend.capabilities()));
+    }
+
+    #[tokio::test]
+    async fn vault_transit_backend_capabilities_golden() {
+        let config = KmsConfig::vault_transit(
+            url::Url::parse("http://127.0.0.1:8200").expect("vault URL should parse"),
+            "dev-token".to_string(),
+        )
+        .with_insecure_development_defaults();
+        // Constructing the client performs no network I/O with token auth.
+        let backend = vault_transit::VaultTransitKmsBackend::new(config)
+            .await
+            .expect("vault transit backend should build");
+
+        insta::assert_json_snapshot!("vault_transit_backend_capabilities", capabilities_snapshot(backend.capabilities()));
+    }
+
+    #[tokio::test]
+    async fn static_backend_capabilities_golden() {
+        let config = KmsConfig::static_kms("static-key".to_string(), BASE64.encode([0u8; 32]));
+        let backend = static_kms::StaticKmsBackend::new(config)
+            .await
+            .expect("static backend should build");
+
+        insta::assert_json_snapshot!("static_backend_capabilities", capabilities_snapshot(backend.capabilities()));
+    }
+}

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__local_backend_capabilities.snap

@@ -0,0 +1,14 @@
+---
+source: crates/kms/src/backends/mod.rs
+expression: capabilities_snapshot(backend.capabilities())
+---
+{
+  "decrypt": true,
+  "enable_disable": true,
+  "encrypt": true,
+  "generate_data_key": true,
+  "physical_delete": true,
+  "rotate": false,
+  "schedule_deletion": true,
+  "versioning": false
+}

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__static_backend_capabilities.snap

@@ -0,0 +1,14 @@
+---
+source: crates/kms/src/backends/mod.rs
+expression: capabilities_snapshot(backend.capabilities())
+---
+{
+  "decrypt": true,
+  "enable_disable": false,
+  "encrypt": true,
+  "generate_data_key": true,
+  "physical_delete": false,
+  "rotate": false,
+  "schedule_deletion": false,
+  "versioning": false
+}

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__vault_kv2_backend_capabilities.snap

@@ -0,0 +1,14 @@
+---
+source: crates/kms/src/backends/mod.rs
+expression: capabilities_snapshot(backend.capabilities())
+---
+{
+  "decrypt": true,
+  "enable_disable": true,
+  "encrypt": true,
+  "generate_data_key": true,
+  "physical_delete": true,
+  "rotate": false,
+  "schedule_deletion": true,
+  "versioning": false
+}

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__vault_transit_backend_capabilities.snap

@@ -0,0 +1,14 @@
+---
+source: crates/kms/src/backends/mod.rs
+expression: capabilities_snapshot(backend.capabilities())
+---
+{
+  "decrypt": true,
+  "enable_disable": true,
+  "encrypt": true,
+  "generate_data_key": true,
+  "physical_delete": true,
+  "rotate": true,
+  "schedule_deletion": true,
+  "versioning": true
+}

crates/kms/src/backends/static_kms.rs

@@ -21,7 +21,7 @@
 //!
 //! encrypted_data(plaintext_len+16) || nonce (12 bytes)
 
-use crate::backends::{BackendInfo, KmsBackend, KmsClient};
+use crate::backends::{BackendCapabilities, BackendInfo, KmsBackend, KmsClient};
 use crate::config::{BackendConfig, KmsConfig};
 use crate::encryption::DataKeyEnvelope;
 use crate::error::{KmsError, Result};
@@ -435,6 +435,12 @@ impl KmsBackend for StaticKmsBackend {
     async fn health_check(&self) -> Result<bool> {
         Ok(true)
     }
+
+    fn capabilities(&self) -> BackendCapabilities {
+        // Static KMS is a read-only single-key backend: it only performs
+        // cryptographic operations and rejects every lifecycle mutation.
+        BackendCapabilities::minimal()
+    }
 }
 
 #[cfg(test)]

crates/kms/src/backends/vault.rs

@@ -15,7 +15,10 @@
 //! Vault-based KMS backend implementation using vaultrs
 
 use crate::backends::vault_credentials::{VaultClientHandle, VaultConnectionSettings, VaultCredentialProvider, token_source_for};
-use crate::backends::{BackendInfo, KmsBackend, KmsClient};
+use crate::backends::{
+    BackendCapabilities, BackendInfo, ExpiredKeyRemoval, KmsBackend, KmsClient, StateGatedOperation, ensure_key_state_permits,
+    ensure_key_status_permits,
+};
 use crate::config::{KmsConfig, VaultConfig};
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
 use crate::error::{KmsError, Result};
@@ -61,6 +64,10 @@ struct VaultKeyData {
     metadata: HashMap<String, String>,
     /// Key tags
     tags: HashMap<String, String>,
+    /// Scheduled deletion deadline; absent on records written before deadline
+    /// persistence landed, so it must stay optional for backward compatibility.
+    #[serde(default)]
+    deletion_date: Option<Zoned>,
     /// Encrypted key material (base64 encoded)
     encrypted_key_material: String,
     /// Version that pre-versioning envelopes (no `master_key_version`) resolve to.
@@ -373,6 +380,7 @@ impl VaultKmsClient {
             description: request.description.clone(),
             metadata: existing_key_data.metadata.clone(),
             tags: request.tags.clone(),
+            deletion_date: existing_key_data.deletion_date.clone(),
             encrypted_key_material: existing_key_data.encrypted_key_material.clone(), // Preserve the key material
             baseline_version: existing_key_data.baseline_version,
         };
@@ -464,6 +472,9 @@ impl KmsClient for VaultKmsClient {
     async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
         debug!("Generating data key for master key: {}", request.master_key_id);
 
+        let key_data = self.get_key_data(&request.master_key_id).await?;
+        ensure_key_status_permits(&request.master_key_id, &key_data.status, StateGatedOperation::GenerateDataKey)?;
+
         // Generate random data key material using the existing method
         let plaintext_key = generate_key_material(&request.key_spec)?;
 
@@ -502,8 +513,9 @@ impl KmsClient for VaultKmsClient {
     async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
         debug!("Encrypting data with key: {}", request.key_id);
 
-        // Get the master key
+        // Get the master key and verify its state allows encryption
         let key_data = self.get_key_data(&request.key_id).await?;
+        ensure_key_status_permits(&request.key_id, &key_data.status, StateGatedOperation::Encrypt)?;
         let key_material = self.decrypt_key_material(&key_data.encrypted_key_material).await?;
 
         // For simplicity, we'll use a basic encryption approach
@@ -585,6 +597,7 @@ impl KmsClient for VaultKmsClient {
             description: None,
             metadata: HashMap::new(),
             tags: HashMap::new(),
+            deletion_date: None,
             encrypted_key_material: encrypted_material,
             baseline_version: None,
         };
@@ -603,6 +616,7 @@ impl KmsClient for VaultKmsClient {
             created_at: key_data.created_at,
             rotated_at: None,
             created_by: None,
+            deletion_date: None,
         };
 
         debug!(key_id, "Vault KMS master key created");
@@ -670,6 +684,7 @@ impl KmsClient for VaultKmsClient {
         debug!("Enabling key: {}", key_id);
 
         let mut key_data = self.get_key_data(key_id).await?;
+        ensure_key_status_permits(key_id, &key_data.status, StateGatedOperation::Enable)?;
         key_data.status = KeyStatus::Active;
         self.store_key_data(key_id, &key_data).await?;
 
@@ -681,6 +696,7 @@ impl KmsClient for VaultKmsClient {
         debug!("Disabling key: {}", key_id);
 
         let mut key_data = self.get_key_data(key_id).await?;
+        ensure_key_status_permits(key_id, &key_data.status, StateGatedOperation::Disable)?;
         key_data.status = KeyStatus::Disabled;
         self.store_key_data(key_id, &key_data).await?;
 
@@ -691,13 +707,15 @@ impl KmsClient for VaultKmsClient {
     async fn schedule_key_deletion(
         &self,
         key_id: &str,
-        _pending_window_days: u32,
+        pending_window_days: u32,
         _context: Option<&OperationContext>,
     ) -> Result<()> {
         debug!("Scheduling key deletion: {}", key_id);
 
         let mut key_data = self.get_key_data(key_id).await?;
+        ensure_key_status_permits(key_id, &key_data.status, StateGatedOperation::ScheduleDeletion)?;
         key_data.status = KeyStatus::PendingDeletion;
+        key_data.deletion_date = Some(Zoned::now() + Duration::from_secs(pending_window_days as u64 * 86400));
         self.store_key_data(key_id, &key_data).await?;
 
         debug!(key_id, "Vault KMS key deletion scheduled");
@@ -708,7 +726,11 @@ impl KmsClient for VaultKmsClient {
         debug!("Canceling key deletion: {}", key_id);
 
         let mut key_data = self.get_key_data(key_id).await?;
+        if key_data.status != KeyStatus::PendingDeletion {
+            return Err(KmsError::invalid_key_state(format!("Key {key_id} is not pending deletion")));
+        }
         key_data.status = KeyStatus::Active;
+        key_data.deletion_date = None;
         self.store_key_data(key_id, &key_data).await?;
 
         debug!(key_id, "Vault KMS key deletion canceled");
@@ -810,6 +832,7 @@ impl KmsClient for VaultKmsClient {
             created_at: key_data.created_at.clone(),
             rotated_at: Some(Zoned::now()),
             created_by: None,
+            deletion_date: key_data.deletion_date.clone(),
         })
     }
 
@@ -852,6 +875,12 @@ pub struct VaultKmsBackend {
 }
 
 impl VaultKmsBackend {
+    /// Lifecycle driver for the shared state-machine contract tests.
+    #[cfg(test)]
+    pub(crate) fn lifecycle_client(&self) -> &VaultKmsClient {
+        &self.client
+    }
+
     /// Create a new VaultKmsBackend
     pub async fn new(config: KmsConfig) -> Result<Self> {
         config.validate()?;
@@ -890,6 +919,7 @@ impl VaultKmsBackend {
             KeyState::Unavailable => KeyStatus::Deleted,
             KeyState::PendingImport => KeyStatus::Disabled, // Treat as disabled until import completes
         };
+        key_data.deletion_date = metadata.deletion_date.clone();
 
         // Update the key data in Vault storage
         self.client.store_key_data(key_id, &key_data).await?;
@@ -989,7 +1019,7 @@ impl KmsBackend for VaultKmsBackend {
             key_usage: key_info.usage,
             description: key_info.description,
             creation_date: key_info.created_at,
-            deletion_date: None,
+            deletion_date: key_data.deletion_date.clone(),
             origin: "VAULT".to_string(),
             key_manager: "VAULT".to_string(),
             tags: key_data.tags,
@@ -1018,8 +1048,17 @@ impl KmsBackend for VaultKmsBackend {
         };
 
         let deletion_date = if request.force_immediate.unwrap_or(false) {
-            // Check if key is already in PendingDeletion state
-            if key_metadata.key_state == KeyState::PendingDeletion {
+            // Check if key is already in PendingDeletion state (or a tombstone
+            // left by a crashed removal, which may simply be completed)
+            if key_metadata.key_state == KeyState::PendingDeletion || key_metadata.key_state == KeyState::Unavailable {
+                // Tombstone first: mark the record Deleted before removing it,
+                // so a crash between the two steps leaves a key that is already
+                // unusable and whose removal can simply be re-run.
+                if key_metadata.key_state == KeyState::PendingDeletion {
+                    let mut key_data = self.client.get_key_data(key_id).await?;
+                    key_data.status = KeyStatus::Deleted;
+                    self.client.store_key_data(key_id, &key_data).await?;
+                }
                 // Force immediate deletion: physically delete the key from Vault storage
                 self.client.delete_key(key_id).await?;
 
@@ -1037,6 +1076,8 @@ impl KmsBackend for VaultKmsBackend {
             }
         } else {
             // Schedule for deletion (default 30 days)
+            ensure_key_state_permits(key_id, &key_metadata.key_state, StateGatedOperation::ScheduleDeletion)?;
+
             let days = request.pending_window_in_days.unwrap_or(30);
             if !(7..=30).contains(&days) {
                 return Err(crate::error::KmsError::invalid_parameter(
@@ -1095,6 +1136,53 @@ impl KmsBackend for VaultKmsBackend {
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
+
+    fn capabilities(&self) -> BackendCapabilities {
+        // Rotation is unadvertised: the KV2 backend cannot rotate without
+        // replacing key material in place, and no historical versions are
+        // retained, so versioning is unsupported as well.
+        BackendCapabilities::minimal()
+            .with_enable_disable(true)
+            .with_schedule_deletion(true)
+            .with_physical_delete(true)
+    }
+
+    async fn remove_expired_key(&self, key_id: &str, now: &Zoned) -> Result<ExpiredKeyRemoval> {
+        // Vault KV2 offers no compare-and-swap here, so a cancellation racing
+        // the read below can still lose; the window is a single read-write
+        // gap and the sweep re-reads on every pass.
+        let mut key_data = match self.client.get_key_data(key_id).await {
+            Ok(key_data) => key_data,
+            Err(KmsError::KeyNotFound { .. }) => return Ok(ExpiredKeyRemoval::Removed),
+            Err(error) => return Err(error),
+        };
+        match key_data.status {
+            // Tombstone left by a crashed removal: complete it.
+            KeyStatus::Deleted => {}
+            KeyStatus::PendingDeletion => {
+                match &key_data.deletion_date {
+                    Some(deadline) if deadline <= now => {}
+                    // Not yet due, or a legacy record without a persisted
+                    // deadline — never auto-remove those.
+                    _ => return Ok(ExpiredKeyRemoval::NotExpired),
+                }
+                // Tombstone first: mark the record Deleted before removing it,
+                // so a crash between the two steps leaves a key that is
+                // already unusable and whose removal can simply be re-run.
+                key_data.status = KeyStatus::Deleted;
+                self.client.store_key_data(key_id, &key_data).await?;
+            }
+            KeyStatus::Active | KeyStatus::Disabled => return Ok(ExpiredKeyRemoval::StateChanged),
+        }
+
+        match self.client.delete_key(key_id).await {
+            Ok(()) | Err(KmsError::KeyNotFound { .. }) => {
+                debug!(key_id, "Vault KV2 expired key removed");
+                Ok(ExpiredKeyRemoval::Removed)
+            }
+            Err(error) => Err(error),
+        }
+    }
 }
 
 #[cfg(test)]
@@ -1259,6 +1347,7 @@ mod tests {
             tags: HashMap::new(),
             encrypted_key_material: general_purpose::STANDARD.encode([0x42u8; 32]),
             baseline_version: Some(1),
+            deletion_date: None,
         };
 
         let mut value = serde_json::to_value(&key_data).expect("serialize key data");
@@ -1617,4 +1706,41 @@ mod tests {
             "cancel_key_deletion must persist Active status to Vault, not only mutate the response"
         );
     }
+
+    /// The persisted KV2 record round-trips its deletion deadline, and records
+    /// written before the field existed keep deserializing (as None). A revert
+    /// of deadline persistence turns this test red.
+    #[test]
+    fn vault_key_data_deletion_date_round_trips_and_stays_backward_compatible() {
+        let deadline = Zoned::now() + Duration::from_secs(7 * 86400);
+        let key_data = VaultKeyData {
+            algorithm: "AES_256".to_string(),
+            usage: KeyUsage::EncryptDecrypt,
+            created_at: Zoned::now(),
+            status: KeyStatus::PendingDeletion,
+            version: 1,
+            description: None,
+            metadata: HashMap::new(),
+            tags: HashMap::new(),
+            deletion_date: Some(deadline.clone()),
+            encrypted_key_material: "material".to_string(),
+            baseline_version: None,
+        };
+
+        let mut value = serde_json::to_value(&key_data).expect("serialize");
+        let restored: VaultKeyData = serde_json::from_value(value.clone()).expect("round trip");
+        assert_eq!(
+            restored.deletion_date.as_ref().map(Zoned::timestamp),
+            Some(deadline.timestamp()),
+            "deletion deadline must survive the KV2 round trip"
+        );
+
+        value
+            .as_object_mut()
+            .expect("record must be a JSON object")
+            .remove("deletion_date")
+            .expect("current records must carry the field");
+        let legacy: VaultKeyData = serde_json::from_value(value).expect("legacy record must deserialize");
+        assert!(legacy.deletion_date.is_none());
+    }
 }

crates/kms/src/backends/vault_transit.rs

@@ -15,7 +15,9 @@
 //! Vault Transit-based KMS backend.
 
 use crate::backends::vault_credentials::{VaultClientHandle, VaultConnectionSettings, VaultCredentialProvider, token_source_for};
-use crate::backends::{BackendInfo, KmsBackend, KmsClient};
+use crate::backends::{
+    BackendCapabilities, BackendInfo, ExpiredKeyRemoval, KmsBackend, KmsClient, StateGatedOperation, ensure_key_state_permits,
+};
 use crate::config::{KmsConfig, VaultTransitConfig};
 use crate::encryption::{DataKeyEnvelope, generate_key_material};
 use crate::error::{KmsError, Result};
@@ -81,6 +83,12 @@ impl TransitKeyMetadata {
         }
     }
 
+    // KNOWN RISK (rustfs/backlog#1571, residual of rustfs/backlog#808): this
+    // fallback defaults to Enabled, so a key whose KV metadata read fails is
+    // treated as usable — a disabled or pending-deletion key can transiently
+    // "revive" on that path. State gates therefore only hold as strongly as
+    // metadata reads do. Changing the fallback is out of scope here; the
+    // synthesized_metadata_defaults_to_enabled test pins the current behavior.
     fn synthesized() -> Self {
         Self {
             key_usage: KeyUsage::EncryptDecrypt,
@@ -363,14 +371,9 @@ impl VaultTransitKmsClient {
         })
     }
 
-    async fn ensure_key_active(&self, key_id: &str) -> Result<TransitKeyMetadata> {
+    async fn ensure_key_state_allows(&self, key_id: &str, operation: StateGatedOperation) -> Result<TransitKeyMetadata> {
         let metadata = self.get_key_metadata(key_id).await?;
-        if metadata.key_state != KeyState::Enabled {
-            return Err(KmsError::invalid_operation(format!(
-                "Key {key_id} is not active (state: {:?})",
-                metadata.key_state
-            )));
-        }
+        ensure_key_state_permits(key_id, &metadata.key_state, operation)?;
         Ok(metadata)
     }
 }
@@ -378,7 +381,8 @@ impl VaultTransitKmsClient {
 #[async_trait]
 impl KmsClient for VaultTransitKmsClient {
     async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
-        self.ensure_key_active(&request.master_key_id).await?;
+        self.ensure_key_state_allows(&request.master_key_id, StateGatedOperation::GenerateDataKey)
+            .await?;
 
         let plaintext_key = generate_key_material(&request.key_spec)?;
         let encrypted_key = self
@@ -409,7 +413,9 @@ impl KmsClient for VaultTransitKmsClient {
     }
 
     async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
-        let metadata = self.ensure_key_active(&request.key_id).await?;
+        let metadata = self
+            .ensure_key_state_allows(&request.key_id, StateGatedOperation::Encrypt)
+            .await?;
         let ciphertext = self
             .transit_encrypt(&request.key_id, &request.plaintext, &request.encryption_context)
             .await?;
@@ -475,6 +481,7 @@ impl KmsClient for VaultTransitKmsClient {
             created_at: metadata.created_at,
             rotated_at: None,
             created_by: metadata.created_by,
+            deletion_date: None,
         })
     }
 
@@ -521,14 +528,16 @@ impl KmsClient for VaultTransitKmsClient {
     }
 
     async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
-        let mut metadata = self.get_key_metadata(key_id).await?;
+        // A pending deletion must be reverted through cancel_key_deletion, not
+        // silently by enabling, so the gate rejects PendingDeletion here.
+        let mut metadata = self.ensure_key_state_allows(key_id, StateGatedOperation::Enable).await?;
         metadata.key_state = KeyState::Enabled;
         metadata.deletion_date = None;
         self.store_key_metadata(key_id, &metadata).await
     }
 
     async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
-        let mut metadata = self.get_key_metadata(key_id).await?;
+        let mut metadata = self.ensure_key_state_allows(key_id, StateGatedOperation::Disable).await?;
         metadata.key_state = KeyState::Disabled;
         self.store_key_metadata(key_id, &metadata).await
     }
@@ -539,7 +548,9 @@ impl KmsClient for VaultTransitKmsClient {
         pending_window_days: u32,
         _context: Option<&OperationContext>,
     ) -> Result<()> {
-        let mut metadata = self.get_key_metadata(key_id).await?;
+        let mut metadata = self
+            .ensure_key_state_allows(key_id, StateGatedOperation::ScheduleDeletion)
+            .await?;
         metadata.key_state = KeyState::PendingDeletion;
         metadata.deletion_date = Some(Zoned::now() + Duration::from_secs(pending_window_days as u64 * 86400));
         self.store_key_metadata(key_id, &metadata).await
@@ -547,12 +558,17 @@ impl KmsClient for VaultTransitKmsClient {
 
     async fn cancel_key_deletion(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         let mut metadata = self.get_key_metadata(key_id).await?;
+        if metadata.key_state != KeyState::PendingDeletion {
+            return Err(KmsError::invalid_key_state(format!("Key {key_id} is not pending deletion")));
+        }
         metadata.key_state = KeyState::Enabled;
         metadata.deletion_date = None;
         self.store_key_metadata(key_id, &metadata).await
     }
 
     async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+        self.ensure_key_state_allows(key_id, StateGatedOperation::Rotate).await?;
+
         key::rotate(&self.vault().client, &self.config.mount_path, key_id)
             .await
             .map_err(|e| KmsError::backend_error(format!("Failed to rotate Vault Transit key {key_id}: {e}")))?;
@@ -572,6 +588,7 @@ impl KmsClient for VaultTransitKmsClient {
             created_at: metadata.created_at,
             rotated_at: Some(Zoned::now()),
             created_by: metadata.created_by,
+            deletion_date: None,
         })
     }
 
@@ -593,6 +610,14 @@ pub struct VaultTransitKmsBackend {
 }
 
 impl VaultTransitKmsBackend {
+    /// Lifecycle driver for the shared state-machine contract tests. Using the
+    /// backend's own client keeps its in-process metadata cache coherent with
+    /// the transitions the tests perform.
+    #[cfg(test)]
+    pub(crate) fn lifecycle_client(&self) -> &VaultTransitKmsClient {
+        &self.client
+    }
+
     pub async fn new(config: KmsConfig) -> Result<Self> {
         config.validate()?;
 
@@ -722,6 +747,8 @@ impl KmsBackend for VaultTransitKmsBackend {
                 None
             }
         } else {
+            ensure_key_state_permits(&key_id, &key_metadata.key_state, StateGatedOperation::ScheduleDeletion)?;
+
             let days = request.pending_window_in_days.unwrap_or(30);
             if !(7..=30).contains(&days) {
                 return Err(KmsError::invalid_parameter("pending_window_in_days must be between 7 and 30"));
@@ -762,6 +789,72 @@ impl KmsBackend for VaultTransitKmsBackend {
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
+
+    fn capabilities(&self) -> BackendCapabilities {
+        // Vault Transit natively supports version-retaining rotation, keeps
+        // prior versions addressable for decryption, and allows physical
+        // deletion once a key is pending deletion.
+        BackendCapabilities::minimal()
+            .with_rotate(true)
+            .with_enable_disable(true)
+            .with_schedule_deletion(true)
+            .with_versioning(true)
+            .with_physical_delete(true)
+    }
+
+    async fn remove_expired_key(&self, key_id: &str, now: &Zoned) -> Result<ExpiredKeyRemoval> {
+        // The transit key's existence anchors "already removed": once it is
+        // gone only stale scheduling metadata can remain, so clean that up.
+        match self.client.read_transit_key(key_id).await {
+            Ok(_) => {}
+            Err(KmsError::KeyNotFound { .. }) => {
+                self.client.delete_key_metadata(key_id).await?;
+                return Ok(ExpiredKeyRemoval::Removed);
+            }
+            Err(error) => return Err(error),
+        }
+
+        // A metadata read failure synthesizes an Enabled record (see
+        // TransitKeyMetadata::synthesized), which lands in StateChanged below:
+        // the worker never destroys material based on synthesized state.
+        let mut metadata = self.client.get_key_metadata(key_id).await?;
+        match metadata.key_state {
+            // Tombstone left by a crashed removal: complete it.
+            KeyState::Unavailable => {}
+            KeyState::PendingDeletion => {
+                match &metadata.deletion_date {
+                    Some(deadline) if deadline <= now => {}
+                    // Not yet due, or no persisted deadline — never auto-remove.
+                    _ => return Ok(ExpiredKeyRemoval::NotExpired),
+                }
+                // Tombstone first: an Unavailable record is rejected by every
+                // state gate, and a crashed removal can simply be re-run.
+                metadata.key_state = KeyState::Unavailable;
+                self.client.store_key_metadata(key_id, &metadata).await?;
+            }
+            KeyState::Enabled | KeyState::Disabled | KeyState::PendingImport => {
+                return Ok(ExpiredKeyRemoval::StateChanged);
+            }
+        }
+
+        if !self.client.read_transit_key(key_id).await?.deletion_allowed {
+            let mut update_builder = UpdateKeyConfigurationRequestBuilder::default();
+            update_builder.deletion_allowed(true);
+            key::update(
+                &self.client.vault().client,
+                &self.client.config.mount_path,
+                key_id,
+                Some(&mut update_builder),
+            )
+            .await
+            .map_err(|e| KmsError::backend_error(format!("Failed to allow deletion of Vault Transit key {key_id}: {e}")))?;
+        }
+        key::delete(&self.client.vault().client, &self.client.config.mount_path, key_id)
+            .await
+            .map_err(|e| KmsError::backend_error(format!("Failed to delete Vault Transit key {key_id}: {e}")))?;
+        self.client.delete_key_metadata(key_id).await?;
+        Ok(ExpiredKeyRemoval::Removed)
+    }
 }
 
 #[cfg(test)]
@@ -973,4 +1066,15 @@ mod tests {
         // Cleanup so repeated runs against the same Vault do not accumulate keys.
         let _ = client.schedule_key_deletion(&key_id, 7, None).await;
     }
+
+    /// Pins the known-risk fallback documented on `TransitKeyMetadata::synthesized`:
+    /// when KV metadata cannot be read, the synthesized record defaults to Enabled,
+    /// which weakens every state gate on that path. If this test turns red the
+    /// fallback semantics changed on purpose — update the comment there as well.
+    #[test]
+    fn synthesized_metadata_defaults_to_enabled() {
+        let metadata = TransitKeyMetadata::synthesized();
+        assert_eq!(metadata.key_state, KeyState::Enabled);
+        assert!(metadata.deletion_date.is_none());
+    }
 }

crates/kms/src/deletion_worker.rs

@@ -0,0 +1,399 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Background worker that completes scheduled key deletions.
+//!
+//! Every sweep lists keys, picks the ones whose persisted deletion deadline
+//! has passed (plus tombstones left by a crashed removal) and hands each to
+//! [`KmsBackend::remove_expired_key`], which re-checks state under the
+//! backend's own synchronization. The sweep is idempotent and keeps no state
+//! of its own, so it is safe to re-run after a restart and safe to run on
+//! every node of a deployment concurrently — a key is only ever removed while
+//! its (re-read) record is an expired pending deletion or a tombstone.
+
+use crate::backends::{ExpiredKeyRemoval, KmsBackend};
+use crate::types::{KeyStatus, ListKeysRequest};
+use async_trait::async_trait;
+use jiff::Zoned;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, info, warn};
+
+/// How often the worker looks for expired pending deletions.
+pub const DEFAULT_SWEEP_INTERVAL: Duration = Duration::from_secs(60);
+
+/// Reports configuration that still references a KMS key.
+///
+/// Consulted before any material is destroyed; a non-empty result blocks the
+/// removal until the references disappear. Implementations live where the
+/// referencing configuration lives (for example bucket encryption settings in
+/// the server) and are injected via
+/// [`crate::service_manager::KmsServiceManager::set_deletion_reference_checker`].
+#[async_trait]
+pub trait DeletionReferenceChecker: Send + Sync {
+    /// Identifiers of configuration still referencing `key_id` (bucket names,
+    /// settings paths, ...). Errors must be reported as a reference so that
+    /// an unavailable checker never unblocks a deletion.
+    async fn references(&self, key_id: &str) -> Vec<String>;
+}
+
+/// Outcome of one sweep, for logging and tests.
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
+pub struct SweepReport {
+    /// Keys whose record and material were removed this sweep.
+    pub removed: Vec<String>,
+    /// Keys left in place because configuration still references them.
+    pub blocked: Vec<String>,
+    /// Keys that were pending but not yet due, without a persisted deadline,
+    /// or whose state changed between inspection and removal.
+    pub skipped: usize,
+    /// Keys whose removal attempt failed; retried on the next sweep.
+    pub failed: usize,
+}
+
+pub(crate) struct DeletionWorker {
+    backend: Arc<dyn KmsBackend>,
+    default_key_id: Option<String>,
+    reference_checker: Option<Arc<dyn DeletionReferenceChecker>>,
+    interval: Duration,
+}
+
+impl DeletionWorker {
+    pub(crate) fn new(
+        backend: Arc<dyn KmsBackend>,
+        default_key_id: Option<String>,
+        reference_checker: Option<Arc<dyn DeletionReferenceChecker>>,
+    ) -> Self {
+        Self {
+            backend,
+            default_key_id,
+            reference_checker,
+            interval: DEFAULT_SWEEP_INTERVAL,
+        }
+    }
+
+    pub(crate) fn spawn(self, cancel: CancellationToken) -> tokio::task::JoinHandle<()> {
+        tokio::spawn(async move { self.run(cancel).await })
+    }
+
+    async fn run(self, cancel: CancellationToken) {
+        let mut ticker = tokio::time::interval(self.interval);
+        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+        loop {
+            tokio::select! {
+                _ = cancel.cancelled() => {
+                    debug!("KMS deletion worker stopped");
+                    return;
+                }
+                _ = ticker.tick() => {}
+            }
+            let report = self.sweep(&Zoned::now()).await;
+            if !report.removed.is_empty() || !report.blocked.is_empty() || report.failed > 0 {
+                info!(
+                    removed = ?report.removed,
+                    blocked = ?report.blocked,
+                    skipped = report.skipped,
+                    failed = report.failed,
+                    "KMS deletion sweep completed"
+                );
+            }
+        }
+    }
+
+    /// Run one sweep at the given time. Exposed separately so tests can drive
+    /// the expiry logic deterministically.
+    pub(crate) async fn sweep(&self, now: &Zoned) -> SweepReport {
+        let mut report = SweepReport::default();
+        let mut marker: Option<String> = None;
+        loop {
+            let request = ListKeysRequest {
+                limit: Some(100),
+                marker: marker.clone(),
+                usage_filter: None,
+                status_filter: None,
+            };
+            let response = match self.backend.list_keys(request).await {
+                Ok(response) => response,
+                Err(error) => {
+                    warn!(%error, "KMS deletion sweep could not list keys");
+                    report.failed += 1;
+                    return report;
+                }
+            };
+            for key in &response.keys {
+                if matches!(key.status, KeyStatus::PendingDeletion | KeyStatus::Deleted) {
+                    self.process_key(&key.key_id, now, &mut report).await;
+                }
+            }
+            if !response.truncated {
+                break;
+            }
+            match response.next_marker {
+                Some(next_marker) => marker = Some(next_marker),
+                None => break,
+            }
+        }
+        report
+    }
+
+    async fn process_key(&self, key_id: &str, now: &Zoned, report: &mut SweepReport) {
+        // Never remove a key that live configuration still points at. The
+        // default key check is built in; broader references (bucket
+        // encryption settings, ...) come from the injected checker.
+        if self.default_key_id.as_deref() == Some(key_id) {
+            warn!(key_id, "expired KMS key is still the default key; refusing removal");
+            report.blocked.push(key_id.to_string());
+            return;
+        }
+        if let Some(checker) = &self.reference_checker {
+            let references = checker.references(key_id).await;
+            if !references.is_empty() {
+                warn!(key_id, ?references, "expired KMS key is still referenced; refusing removal");
+                report.blocked.push(key_id.to_string());
+                return;
+            }
+        }
+
+        // The backend re-checks state and deadline under its own write
+        // synchronization, so a cancellation racing this sweep wins there.
+        match self.backend.remove_expired_key(key_id, now).await {
+            Ok(ExpiredKeyRemoval::Removed) => report.removed.push(key_id.to_string()),
+            Ok(ExpiredKeyRemoval::StateChanged | ExpiredKeyRemoval::NotExpired) => report.skipped += 1,
+            Err(error) => {
+                warn!(key_id, %error, "failed to remove expired KMS key; will retry next sweep");
+                report.failed += 1;
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::backends::KmsClient as _;
+    use crate::backends::local::LocalKmsBackend;
+    use crate::config::KmsConfig;
+    use crate::error::KmsError;
+    use crate::types::{CreateKeyRequest, DeleteKeyRequest, DescribeKeyRequest, KeyState, KeyUsage};
+
+    async fn local_backend(temp_dir: &tempfile::TempDir) -> Arc<LocalKmsBackend> {
+        let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
+        Arc::new(LocalKmsBackend::new(config).await.expect("local backend should build"))
+    }
+
+    async fn create_key(backend: &LocalKmsBackend, key_name: &str) -> String {
+        backend
+            .create_key(CreateKeyRequest {
+                key_name: Some(key_name.to_string()),
+                key_usage: KeyUsage::EncryptDecrypt,
+                ..Default::default()
+            })
+            .await
+            .expect("key should be created")
+            .key_id
+    }
+
+    async fn schedule(backend: &LocalKmsBackend, key_id: &str) {
+        backend
+            .delete_key(DeleteKeyRequest {
+                key_id: key_id.to_string(),
+                pending_window_in_days: Some(7),
+                force_immediate: None,
+            })
+            .await
+            .expect("deletion should be scheduled");
+    }
+
+    fn worker(backend: Arc<LocalKmsBackend>) -> DeletionWorker {
+        DeletionWorker::new(backend, None, None)
+    }
+
+    fn after_window() -> Zoned {
+        Zoned::now() + Duration::from_secs(8 * 86400)
+    }
+
+    async fn assert_key_gone(backend: &LocalKmsBackend, key_id: &str) {
+        let error = backend
+            .describe_key(DescribeKeyRequest {
+                key_id: key_id.to_string(),
+            })
+            .await
+            .expect_err("removed key must not be describable");
+        assert!(matches!(error, KmsError::KeyNotFound { .. }), "expected KeyNotFound, got {error:?}");
+    }
+
+    #[tokio::test]
+    async fn sweep_removes_expired_pending_key_and_is_idempotent() {
+        let temp_dir = tempfile::tempdir().expect("temp dir");
+        let backend = local_backend(&temp_dir).await;
+        let key_id = create_key(&backend, "expired-key").await;
+        schedule(&backend, &key_id).await;
+
+        let worker = worker(backend.clone());
+
+        // Not yet due: nothing happens.
+        let report = worker.sweep(&Zoned::now()).await;
+        assert!(report.removed.is_empty());
+        assert_eq!(report.skipped, 1);
+        assert_eq!(report.failed, 0);
+
+        // Past the deadline: the key is removed.
+        let report = worker.sweep(&after_window()).await;
+        assert_eq!(report.removed, vec![key_id.clone()]);
+        assert_eq!(report.failed, 0);
+        assert_key_gone(&backend, &key_id).await;
+
+        // Re-running the sweep after the removal is a no-op.
+        let report = worker.sweep(&after_window()).await;
+        assert_eq!(report, SweepReport::default());
+    }
+
+    #[tokio::test]
+    async fn cancelled_deletion_always_beats_the_sweep() {
+        let temp_dir = tempfile::tempdir().expect("temp dir");
+        let backend = local_backend(&temp_dir).await;
+        let cancelled = create_key(&backend, "cancelled-key").await;
+        let doomed = create_key(&backend, "doomed-key").await;
+        schedule(&backend, &cancelled).await;
+        schedule(&backend, &doomed).await;
+
+        backend
+            .cancel_key_deletion(crate::types::CancelKeyDeletionRequest {
+                key_id: cancelled.clone(),
+            })
+            .await
+            .expect("cancel should succeed");
+
+        let report = worker(backend.clone()).sweep(&after_window()).await;
+        assert_eq!(report.removed, vec![doomed.clone()]);
+        assert_eq!(report.failed, 0);
+
+        // The cancelled key survives, enabled and usable.
+        let described = backend
+            .describe_key(DescribeKeyRequest {
+                key_id: cancelled.clone(),
+            })
+            .await
+            .expect("cancelled key must still exist");
+        assert_eq!(described.key_metadata.key_state, KeyState::Enabled);
+        assert_key_gone(&backend, &doomed).await;
+    }
+
+    #[tokio::test]
+    async fn default_key_and_external_references_block_removal() {
+        struct StaticReferences(Vec<String>);
+
+        #[async_trait]
+        impl DeletionReferenceChecker for StaticReferences {
+            async fn references(&self, _key_id: &str) -> Vec<String> {
+                self.0.clone()
+            }
+        }
+
+        let temp_dir = tempfile::tempdir().expect("temp dir");
+        let backend = local_backend(&temp_dir).await;
+        let key_id = create_key(&backend, "referenced-key").await;
+        schedule(&backend, &key_id).await;
+
+        // Blocked while it is the configured default key.
+        let as_default = DeletionWorker::new(backend.clone(), Some(key_id.clone()), None);
+        let report = as_default.sweep(&after_window()).await;
+        assert_eq!(report.blocked, vec![key_id.clone()]);
+        assert!(report.removed.is_empty());
+
+        // Blocked while external configuration still references it.
+        let with_references = DeletionWorker::new(
+            backend.clone(),
+            None,
+            Some(Arc::new(StaticReferences(vec!["bucket:sse-bucket".to_string()]))),
+        );
+        let report = with_references.sweep(&after_window()).await;
+        assert_eq!(report.blocked, vec![key_id.clone()]);
+        assert!(report.removed.is_empty());
+        backend
+            .describe_key(DescribeKeyRequest { key_id: key_id.clone() })
+            .await
+            .expect("blocked key must still exist");
+
+        // Removed once nothing references it anymore.
+        let unreferenced = DeletionWorker::new(backend.clone(), None, Some(Arc::new(StaticReferences(Vec::new()))));
+        let report = unreferenced.sweep(&after_window()).await;
+        assert_eq!(report.removed, vec![key_id.clone()]);
+    }
+
+    #[tokio::test]
+    async fn deadline_survives_backend_restart_and_sweep_completes_it() {
+        let temp_dir = tempfile::tempdir().expect("temp dir");
+        let key_id;
+        {
+            let backend = local_backend(&temp_dir).await;
+            key_id = create_key(&backend, "restart-key").await;
+            schedule(&backend, &key_id).await;
+        }
+
+        // "Restart": a fresh backend over the same directory must still see
+        // the persisted deadline...
+        let backend = local_backend(&temp_dir).await;
+        let described = backend
+            .describe_key(DescribeKeyRequest { key_id: key_id.clone() })
+            .await
+            .expect("key must survive the restart");
+        assert_eq!(described.key_metadata.key_state, KeyState::PendingDeletion);
+        assert!(
+            described.key_metadata.deletion_date.is_some(),
+            "deletion deadline must survive a backend restart"
+        );
+
+        // ...and the worker completes the deletion without any new schedule call.
+        let report = worker(backend.clone()).sweep(&after_window()).await;
+        assert_eq!(report.removed, vec![key_id.clone()]);
+        assert_key_gone(&backend, &key_id).await;
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn worker_loop_removes_due_keys_and_stops_on_cancel() {
+        let temp_dir = tempfile::tempdir().expect("temp dir");
+        let backend = local_backend(&temp_dir).await;
+        let key_id = create_key(&backend, "loop-key").await;
+        // A zero-day window through the lifecycle client produces a deadline
+        // that is already due for the worker's wall-clock sweep.
+        backend
+            .lifecycle_client()
+            .schedule_key_deletion(&key_id, 0, None)
+            .await
+            .expect("schedule with zero window");
+
+        let cancel = CancellationToken::new();
+        let task = worker(backend.clone()).spawn(cancel.clone());
+
+        // The paused clock auto-advances through the worker's interval ticks.
+        let mut removed = false;
+        for _ in 0..100 {
+            tokio::time::sleep(Duration::from_secs(1)).await;
+            if backend
+                .describe_key(DescribeKeyRequest { key_id: key_id.clone() })
+                .await
+                .is_err()
+            {
+                removed = true;
+                break;
+            }
+        }
+        assert!(removed, "worker loop must remove the due key");
+
+        cancel.cancel();
+        task.await.expect("worker task must stop after cancellation");
+    }
+}

crates/kms/src/error.rs

@@ -128,6 +128,10 @@ pub enum KmsError {
     /// Backup/restore bundle contract violation; see [`crate::backup::BackupError`]
     #[error(transparent)]
     Backup(#[from] crate::backup::BackupError),
+
+    /// Operation is not supported by the active KMS backend
+    #[error("Operation '{operation}' is not supported by KMS backend '{backend}'")]
+    UnsupportedCapability { backend: String, operation: String },
 }
 
 impl KmsError {
@@ -269,6 +273,14 @@ impl KmsError {
             version,
         }
     }
+
+    /// Create an unsupported capability error
+    pub fn unsupported_capability<S1: Into<String>, S2: Into<String>>(backend: S1, operation: S2) -> Self {
+        Self::UnsupportedCapability {
+            backend: backend.into(),
+            operation: operation.into(),
+        }
+    }
 }
 
 /// Convert from standard library errors

crates/kms/src/lib.rs

@@ -69,6 +69,7 @@ pub mod backends;
 pub mod backup;
 mod cache;
 pub mod config;
+pub mod deletion_worker;
 mod encryption;
 mod error;
 pub mod manager;
@@ -89,6 +90,7 @@ pub use api_types::{
     UpdateKeyDescriptionRequest, UpdateKeyDescriptionResponse,
 };
 pub use config::*;
+pub use deletion_worker::DeletionReferenceChecker;
 pub use encryption::is_data_key_envelope;
 pub use error::{KmsError, KmsUnavailableError, Result};
 pub use manager::KmsManager;

crates/kms/src/manager.rs

@@ -159,6 +159,17 @@ impl KmsManager {
     pub async fn health_check(&self) -> Result<bool> {
         self.backend.health_check().await
     }
+
+    /// Report the capabilities of the configured backend
+    pub fn backend_capabilities(&self) -> crate::backends::BackendCapabilities {
+        self.backend.capabilities()
+    }
+
+    /// Direct handle to the configured backend, bypassing the metadata cache.
+    /// Used by background maintenance that must observe fresh state.
+    pub(crate) fn backend(&self) -> Arc<dyn KmsBackend> {
+        self.backend.clone()
+    }
 }
 
 #[cfg(test)]

crates/kms/src/service.rs

@@ -184,6 +184,15 @@ impl ObjectEncryptionService {
         self.kms_manager.health_check().await
     }
 
+    /// Report the capabilities of the configured backend
+    ///
+    /// # Returns
+    /// The capability matrix advertised by the active KMS backend
+    ///
+    pub fn backend_capabilities(&self) -> crate::backends::BackendCapabilities {
+        self.kms_manager.backend_capabilities()
+    }
+
     /// Create a data encryption key for object encryption
     ///
     /// # Arguments

crates/kms/src/service_manager.rs

@@ -16,6 +16,7 @@
 
 use crate::backends::{KmsBackend, local::LocalKmsBackend};
 use crate::config::{BackendConfig, KmsConfig};
+use crate::deletion_worker::{DeletionReferenceChecker, DeletionWorker};
 use crate::error::{KmsError, Result};
 use crate::manager::KmsManager;
 use crate::service::ObjectEncryptionService;
@@ -28,6 +29,7 @@ use std::sync::{
 };
 use subtle::ConstantTimeEq;
 use tokio::sync::Mutex;
+use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, warn};
 
 const LOG_COMPONENT_KMS: &str = "kms";
@@ -110,6 +112,40 @@ struct ServiceVersion {
     service: Arc<ObjectEncryptionService>,
     /// The KMS manager instance
     manager: Arc<KmsManager>,
+    /// Background deletion worker owned by this service version, if the
+    /// backend supports deletion scheduling
+    deletion_worker: Option<Arc<DeletionWorkerHandle>>,
+}
+
+impl ServiceVersion {
+    fn shutdown_deletion_worker(&self) {
+        if let Some(worker) = &self.deletion_worker {
+            worker.shutdown();
+        }
+    }
+}
+
+/// Cancellation handle for one service version's deletion worker.
+struct DeletionWorkerHandle {
+    cancel: CancellationToken,
+    task: std::sync::Mutex<Option<tokio::task::JoinHandle<()>>>,
+}
+
+impl DeletionWorkerHandle {
+    fn shutdown(&self) {
+        self.cancel.cancel();
+        if let Ok(mut task) = self.task.lock() {
+            // Detach: the task observes the cancelled token on its next poll.
+            drop(task.take());
+        }
+    }
+}
+
+impl Drop for DeletionWorkerHandle {
+    fn drop(&mut self) {
+        // Safety net for versions that are replaced without an explicit stop.
+        self.cancel.cancel();
+    }
 }
 
 #[derive(Clone)]
@@ -128,6 +164,8 @@ pub struct KmsServiceManager {
     /// Mutex to protect lifecycle operations (start, stop, reconfigure)
     /// This ensures only one lifecycle operation happens at a time
     lifecycle_mutex: Arc<Mutex<()>>,
+    /// External reference checker consulted before expired keys are removed
+    deletion_reference_checker: std::sync::RwLock<Option<Arc<dyn DeletionReferenceChecker>>>,
 }
 
 impl KmsServiceManager {
@@ -141,9 +179,23 @@ impl KmsServiceManager {
             }),
             version_counter: Arc::new(AtomicU64::new(0)),
             lifecycle_mutex: Arc::new(Mutex::new(())),
+            deletion_reference_checker: std::sync::RwLock::new(None),
         }
     }
 
+    /// Install the reference checker consulted before the deletion worker
+    /// removes an expired key. Takes effect for workers spawned by the next
+    /// start or reconfigure.
+    pub fn set_deletion_reference_checker(&self, checker: Arc<dyn DeletionReferenceChecker>) {
+        if let Ok(mut slot) = self.deletion_reference_checker.write() {
+            *slot = Some(checker);
+        }
+    }
+
+    fn deletion_reference_checker(&self) -> Option<Arc<dyn DeletionReferenceChecker>> {
+        self.deletion_reference_checker.read().ok().and_then(|slot| slot.clone())
+    }
+
     /// Get current service status
     pub async fn get_status(&self) -> KmsServiceStatus {
         self.state.load().status.clone()
@@ -321,6 +373,9 @@ impl KmsServiceManager {
         // Atomically clear current service version (lock-free, instant)
         // Note: Existing Arc references will keep the service alive until operations complete
         let state = self.state.load_full();
+        if let Some(current) = state.current_service.as_ref() {
+            current.shutdown_deletion_worker();
+        }
         self.state.store(Arc::new(RuntimeState {
             config: state.config.clone(),
             status: if state.config.is_some() {
@@ -522,6 +577,7 @@ impl KmsServiceManager {
             version,
             service: encryption_service,
             manager: kms_manager,
+            deletion_worker: None,
         })
     }
 
@@ -533,7 +589,11 @@ impl KmsServiceManager {
         Ok(service_version)
     }
 
-    fn publish_running(&self, config: KmsConfig, service_version: ServiceVersion) {
+    fn publish_running(&self, config: KmsConfig, mut service_version: ServiceVersion) {
+        if let Some(previous) = self.state.load().current_service.as_ref() {
+            previous.shutdown_deletion_worker();
+        }
+        service_version.deletion_worker = self.spawn_deletion_worker(&config, &service_version);
         self.state.store(Arc::new(RuntimeState {
             config: Some(config),
             status: KmsServiceStatus::Running,
@@ -541,6 +601,24 @@ impl KmsServiceManager {
         }));
     }
 
+    /// Spawn the background deletion worker for a service version about to be
+    /// published, if its backend supports deletion scheduling. The worker is
+    /// only started at publish time so failed start/reconfigure candidates
+    /// never leak a running task.
+    fn spawn_deletion_worker(&self, config: &KmsConfig, service_version: &ServiceVersion) -> Option<Arc<DeletionWorkerHandle>> {
+        let backend = service_version.manager.backend();
+        if !backend.capabilities().schedule_deletion {
+            return None;
+        }
+        let cancel = CancellationToken::new();
+        let worker = DeletionWorker::new(backend, config.default_key_id.clone(), self.deletion_reference_checker());
+        let task = worker.spawn(cancel.clone());
+        Some(Arc::new(DeletionWorkerHandle {
+            cancel,
+            task: std::sync::Mutex::new(Some(task)),
+        }))
+    }
+
     fn mark_health_error_if_current(&self, checked_version: u64, error: &KmsError) {
         let current = self.state.load_full();
         if current.current_service.as_ref().map(|version| version.version) == Some(checked_version) {
@@ -824,6 +902,66 @@ mod tests {
         assert!(matches!(current.backend_config, BackendConfig::Local(_)));
     }
 
+    #[tokio::test]
+    async fn deletion_worker_follows_the_service_lifecycle() {
+        use tempfile::TempDir;
+
+        let key_dir = TempDir::new().expect("create local KMS directory");
+        let mut config = KmsConfig::local(key_dir.path().to_path_buf());
+        config.allow_insecure_dev_defaults = true;
+        let manager = KmsServiceManager::new();
+        manager.configure(config).await.expect("configure local KMS");
+        manager.start().await.expect("start local KMS");
+
+        let first_worker = manager
+            .state
+            .load()
+            .current_service
+            .as_ref()
+            .expect("running service")
+            .deletion_worker
+            .clone()
+            .expect("local backend must run a deletion worker");
+        assert!(!first_worker.cancel.is_cancelled());
+
+        // Replacing the service version replaces (and cancels) its worker.
+        manager.restart().await.expect("restart");
+        assert!(first_worker.cancel.is_cancelled(), "replaced version's worker must be cancelled");
+        let second_worker = manager
+            .state
+            .load()
+            .current_service
+            .as_ref()
+            .expect("running service")
+            .deletion_worker
+            .clone()
+            .expect("restarted service must run a fresh worker");
+        assert!(!second_worker.cancel.is_cancelled());
+
+        // Stopping the service stops its worker.
+        manager.stop().await.expect("stop");
+        assert!(second_worker.cancel.is_cancelled(), "stop must cancel the deletion worker");
+    }
+
+    #[tokio::test]
+    async fn static_backend_runs_no_deletion_worker() {
+        let manager = KmsServiceManager::new();
+        manager.configure(static_config("key-a", 0x11)).await.expect("configure");
+        manager.start().await.expect("start");
+
+        assert!(
+            manager
+                .state
+                .load()
+                .current_service
+                .as_ref()
+                .expect("running service")
+                .deletion_worker
+                .is_none(),
+            "a backend without deletion scheduling must not run a worker"
+        );
+    }
+
     #[tokio::test]
     async fn reconfigure_allows_safe_local_runtime_settings_only() {
         use tempfile::TempDir;

crates/kms/src/types.rs

@@ -117,6 +117,9 @@ pub struct MasterKeyInfo {
     pub rotated_at: Option<Zoned>,
     /// Key creator/owner
     pub created_by: Option<String>,
+    /// Scheduled deletion deadline while the key is pending deletion
+    #[serde(default)]
+    pub deletion_date: Option<Zoned>,
 }
 
 impl MasterKeyInfo {
@@ -142,6 +145,7 @@ impl MasterKeyInfo {
             created_at: Zoned::now(),
             rotated_at: None,
             created_by,
+            deletion_date: None,
         }
     }
 
@@ -173,6 +177,7 @@ impl MasterKeyInfo {
             created_at: Zoned::now(),
             rotated_at: None,
             created_by,
+            deletion_date: None,
         }
     }
 }

rustfs/src/admin/handlers/kms_management.rs

@@ -72,6 +72,10 @@ pub struct KmsStatusResponse {
     pub cache_enabled: bool,
     pub cache_stats: Option<CacheStatsResponse>,
     pub default_key_id: Option<String>,
+    /// Capability matrix of the active backend. Additive field: omitted by
+    /// older servers, so it must stay optional for consumers.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub capabilities: Option<rustfs_kms::backends::BackendCapabilities>,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -204,6 +208,7 @@ impl Operation for KmsStatusHandler {
             cache_enabled: config.as_ref().is_some_and(|cfg| cfg.enable_cache),
             cache_stats,
             default_key_id: service.get_default_key_id().cloned(),
+            capabilities: Some(service.backend_capabilities()),
         };
 
         let data = serde_json::to_vec(&response).map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?;
@@ -339,4 +344,34 @@ mod tests {
     fn kms_clear_cache_rejects_server_info_fallback() {
         assert_lacks_action(&kms_clear_cache_actions(), Action::AdminAction(AdminAction::ServerInfoAdminAction));
     }
+
+    /// The `capabilities` field is additive: payloads produced by older
+    /// servers (without the field) must keep deserializing, and the field
+    /// must be omitted from JSON when unset so existing consumers see an
+    /// unchanged response shape.
+    #[test]
+    fn kms_status_response_capabilities_field_is_additive() {
+        let legacy_json = serde_json::json!({
+            "backend_type": "local",
+            "backend_status": "healthy",
+            "cache_enabled": true,
+            "cache_stats": null,
+            "default_key_id": null,
+        });
+        let legacy: super::KmsStatusResponse =
+            serde_json::from_value(legacy_json).expect("legacy status payload should deserialize");
+        assert!(legacy.capabilities.is_none());
+
+        let serialized = serde_json::to_value(&legacy).expect("status response should serialize");
+        assert!(serialized.get("capabilities").is_none(), "unset capabilities must be omitted");
+
+        let with_capabilities = super::KmsStatusResponse {
+            capabilities: Some(rustfs_kms::backends::BackendCapabilities::minimal()),
+            ..legacy
+        };
+        let serialized = serde_json::to_value(&with_capabilities).expect("status response should serialize");
+        let capabilities = serialized.get("capabilities").expect("capabilities must be present when set");
+        assert_eq!(capabilities.get("encrypt"), Some(&serde_json::Value::Bool(true)));
+        assert_eq!(capabilities.get("rotate"), Some(&serde_json::Value::Bool(false)));
+    }
 }