fix(kms): restore persisted configuration after restart (#6821)

* fix(kms): restore persisted configuration after restart

* docs(kms): cover the reload route and startup load states

The admin contract matrix pins every dynamic KMS route for the rc and
console handoff, so the new POST /kms/reload needs a row there, and the
reload response reuses the configure snapshot shape rather than adding a
wire type. The observability runbook gains the operator procedure the
reload exists for: telling a load_failed startup apart from a server
that was never configured, and recovering without resubmitting secrets.

crates/e2e_test/src/kms/configured_roundtrip_test.rs

@@ -17,6 +17,7 @@
 
 use super::common::{
     LocalKMSTestEnvironment, VAULT_KEY_NAME, VaultTestEnvironment, configure_kms, get_kms_status, kms_admin_request, start_kms,
+    test_sse_kms_encryption,
 };
 use aws_sdk_s3::primitives::ByteStream;
 use aws_sdk_s3::types::{BucketVersioningStatus, ServerSideEncryption, VersioningConfiguration};
@@ -431,6 +432,38 @@ async fn test_configured_local_kms_admin_and_versioned_cleanup() -> TestResult {
     Ok(())
 }
 
+#[tokio::test]
+async fn test_admin_configured_local_kms_is_restored_after_restart() -> TestResult {
+    let mut env = LocalKMSTestEnvironment::new().await?;
+    env.base_env.start_rustfs_server(Vec::new()).await?;
+
+    let default_key_id = env.configure_local_kms().await?;
+    start_kms(&env.base_env.url, &env.base_env.access_key, &env.base_env.secret_key).await?;
+
+    env.base_env.restart_server_preserving_data(Vec::new(), &[]).await?;
+    assert_configured_status(
+        &env.base_env.url,
+        &env.base_env.access_key,
+        &env.base_env.secret_key,
+        "local",
+        &default_key_id,
+    )
+    .await?;
+
+    let bucket = format!("kms-restart-{}", Uuid::new_v4());
+    env.base_env.create_test_bucket(&bucket).await?;
+    let client = env.base_env.create_s3_client();
+    test_sse_kms_encryption(&client, &bucket).await?;
+    client
+        .delete_object()
+        .bucket(&bucket)
+        .key("test-sse-kms-object")
+        .send()
+        .await?;
+    env.base_env.delete_test_bucket(&bucket).await?;
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_configured_vault_kms_admin_and_versioned_cleanup() -> TestResult {
     let mut env = VaultTestEnvironment::new().await?;

crates/kms/src/service_manager.rs

@@ -259,6 +259,25 @@ impl KmsServiceManager {
         (state.status.clone(), config)
     }
 
+    /// Publish an initialization failure when no usable KMS state exists yet.
+    ///
+    /// Startup configuration discovery happens outside this crate. Recording
+    /// its failure here keeps status truthful without allowing a late failure
+    /// to replace an already configured or running service.
+    pub async fn record_initialization_error(&self, message: impl Into<String>) {
+        let _guard = self.lifecycle_mutex.lock().await;
+        let current = self.state.load_full();
+        if current.config.is_some() || current.current_service.is_some() {
+            return;
+        }
+
+        self.state.store(Arc::new(RuntimeState {
+            config: None,
+            status: KmsServiceStatus::Error(message.into()),
+            current_service: None,
+        }));
+    }
+
     fn redact_config(config: &mut KmsConfig) {
         if let BackendConfig::Static(static_config) = &mut config.backend_config {
             use zeroize::Zeroize;
@@ -849,6 +868,39 @@ mod tests {
         assert!(manager.get_encryption_service().await.is_none());
     }
 
+    #[tokio::test]
+    async fn initialization_error_is_visible_until_configuration_succeeds() {
+        let manager = KmsServiceManager::new();
+
+        manager
+            .record_initialization_error("persisted configuration could not be loaded")
+            .await;
+
+        assert_eq!(
+            manager.get_status().await,
+            KmsServiceStatus::Error("persisted configuration could not be loaded".to_string())
+        );
+        assert!(manager.get_config().await.is_none());
+
+        manager
+            .configure(static_config("key-a", 0x11))
+            .await
+            .expect("configure after startup failure");
+        assert_eq!(manager.get_status().await, KmsServiceStatus::Configured);
+    }
+
+    #[tokio::test]
+    async fn initialization_error_never_replaces_a_running_service() {
+        let manager = KmsServiceManager::new();
+        manager.configure(static_config("key-a", 0x11)).await.expect("configure");
+        manager.start().await.expect("start");
+
+        manager.record_initialization_error("late startup failure").await;
+
+        assert_eq!(manager.get_status().await, KmsServiceStatus::Running);
+        assert!(manager.get_encryption_service().await.is_some());
+    }
+
     #[tokio::test]
     async fn configure_rejects_running_service_without_changing_snapshot() {
         let manager = KmsServiceManager::new();