fix(kms): prevent transient health failures from latching status (#7578)

Keep backend health checks from overwriting the running service lifecycle state so subsequent admin checks and probe-based readiness can recover without a restart.

Add a regression test that fails on the original implementation after backend recovery and verifies the service instance and version remain unchanged.

Validation: 39 focused resilience, lifecycle, concurrency, and service manager tests passed; one existing live AWS test remained ignored. cargo fmt --all --check and git diff --check passed.

Thanks to @stevapple for reporting the issue and providing a detailed diagnosis and reproduction.

Fixes #7554

crates/kms/src/service_manager.rs

@@ -577,13 +577,16 @@ impl KmsServiceManager {
         Some(service_version.probe_worker.as_ref()?.status())
     }
 
-    /// Health check for the KMS service
+    /// Check backend health without changing the service lifecycle state.
+    ///
+    /// A transient backend failure leaves the published service available for
+    /// subsequent checks and operations. Readiness uses the background probe
+    /// to evaluate backend availability independently of lifecycle state.
     pub async fn health_check(&self) -> Result<bool> {
         let checked_state = self.state.load_full();
         match checked_state.current_service.as_ref() {
             Some(service_version) => {
                 let manager = service_version.manager.clone();
-                let checked_version = service_version.version;
                 // Perform health check on the backend
                 match manager.health_check().await {
                     Ok(healthy) => {
@@ -594,8 +597,6 @@ impl KmsServiceManager {
                     }
                     Err(e) => {
                         error!("KMS health check error: {}", e);
-                        let _guard = self.lifecycle_mutex.lock().await;
-                        self.mark_health_error_if_current(checked_version, &e);
                         Err(e)
                     }
                 }
@@ -739,17 +740,6 @@ impl KmsServiceManager {
             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) {
-            self.state.store(Arc::new(RuntimeState {
-                config: current.config.clone(),
-                status: KmsServiceStatus::Error(format!("Health check failed: {error}")),
-                current_service: current.current_service.clone(),
-            }));
-        }
-    }
 }
 
 impl Default for KmsServiceManager {
@@ -1004,19 +994,6 @@ mod tests {
         assert!(manager.get_service_version().await.expect("restarted version") > first_version);
     }
 
-    #[tokio::test]
-    async fn stale_health_failure_cannot_poison_new_service_status() {
-        let manager = KmsServiceManager::new();
-        manager.configure(static_config("key-a", 0x11)).await.expect("configure");
-        manager.start().await.expect("start");
-        let old_version = manager.get_service_version().await.expect("old version");
-        manager.restart().await.expect("restart");
-
-        manager.mark_health_error_if_current(old_version, &KmsError::backend_error("stale failure"));
-
-        assert_eq!(manager.get_status().await, KmsServiceStatus::Running);
-    }
-
     #[tokio::test]
     async fn forbidden_local_master_key_change_preserves_running_config_and_service() {
         use crate::types::{CreateKeyRequest, KeyUsage};

crates/kms/tests/behavior_resilience.rs

@@ -75,6 +75,44 @@ fn unreachable_vault_config() -> KmsConfig {
     }
 }
 
+#[tokio::test]
+async fn transient_health_failure_does_not_latch_the_service_status() {
+    let kms = TestKms::local().await;
+    let manager = kms.manager();
+    let service = manager.get_encryption_service().await.expect("running service");
+    let version = manager.get_service_version().await.expect("running version");
+    assert!(manager.health_check().await.expect("initial backend health"));
+
+    // Move only this test's keys out of reach, then restore the same backend.
+    let key_dir = kms.key_dir().expect("local key directory");
+    let outage = tempfile::TempDir::new().expect("temporary outage directory");
+    let hidden_keys = outage.path().join("keys");
+    tokio::fs::rename(&key_dir, &hidden_keys)
+        .await
+        .expect("make backend unavailable");
+    let failure = manager.health_check().await;
+    let outage_status = manager.get_status().await;
+    tokio::fs::rename(&hidden_keys, &key_dir).await.expect("restore backend");
+
+    assert!(failure.is_err(), "the outage must surface as a health-check error");
+    assert!(manager.health_check().await.expect("backend recovers without restart"));
+    assert!(Arc::ptr_eq(
+        &service,
+        &manager.get_encryption_service().await.expect("service survives the outage")
+    ));
+    assert_eq!(manager.get_service_version().await, Some(version));
+    assert_eq!(
+        manager.get_status().await,
+        KmsServiceStatus::Running,
+        "a recovered backend must not leave service-status and readiness latched in Error"
+    );
+    assert_eq!(
+        outage_status,
+        KmsServiceStatus::Running,
+        "backend health does not change the running service's lifecycle state"
+    );
+}
+
 #[tokio::test]
 async fn starting_against_an_unreachable_backend_fails_without_publishing_a_service() {
     let manager = KmsServiceManager::new();