fix(kms): distinguish key directory outages from missing keys (#7470)

* test(kms): cover directory outages and missing keys on main

(cherry picked from commit 8fc1f0c41d)

* fix(kms): preserve directory availability errors on main

(cherry picked from commit c2a8e476f7)

crates/e2e_test/src/kms/kms_fault_recovery_test.rs

@@ -59,6 +59,36 @@ async fn test_kms_key_directory_unavailable() -> Result<(), Box<dyn std::error::
 
     assert_eq!(put_response.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
 
+    // A missing key in the healthy store is a client error, unlike a store outage.
+    let missing_key_object = "test-missing-kms-key";
+    let missing_key_error = s3_client
+        .put_object()
+        .bucket(TEST_BUCKET)
+        .key(missing_key_object)
+        .body(aws_sdk_s3::primitives::ByteStream::from_static(b"must not be published"))
+        .server_side_encryption(ServerSideEncryption::AwsKms)
+        .ssekms_key_id("rustfs-e2e-test-missing-key")
+        .send()
+        .await
+        .expect_err("an unknown key in a healthy Local KMS store must reject the write");
+    assert_eq!(missing_key_error.raw_response().map(|response| response.status().as_u16()), Some(400));
+    assert_eq!(
+        missing_key_error.as_service_error().and_then(ProvideErrorMetadata::code),
+        Some("KMS.NotFoundException")
+    );
+    let missing_key_absence = s3_client
+        .get_object()
+        .bucket(TEST_BUCKET)
+        .key(missing_key_object)
+        .send()
+        .await
+        .expect_err("a write rejected by a missing KMS key must not publish an object");
+    assert_eq!(missing_key_absence.raw_response().map(|response| response.status().as_u16()), Some(404));
+    assert_eq!(
+        missing_key_absence.as_service_error().and_then(ProvideErrorMetadata::code),
+        Some("NoSuchKey")
+    );
+
     // Temporarily rename the key directory to simulate unavailability
     info!("🔧 Simulating key directory unavailability");
     let backup_dir = format!("{}.backup", kms_env.kms_keys_dir);

crates/kms/src/backends/local.rs

@@ -1233,6 +1233,9 @@ impl LocalKmsClient {
     async fn decode_stored_key(&self, key_id: &str) -> Result<(StoredMasterKey, Vec<u8>)> {
         let key_path = self.master_key_path(key_id)?;
         if !fs::try_exists(&key_path).await? {
+            // Only an accessible key store can establish that a single key is
+            // missing; a directory outage must retain its filesystem error.
+            let _ = fs::read_dir(&self.config.key_dir).await?;
             return Err(KmsError::key_not_found(key_id));
         }
 
@@ -2387,6 +2390,76 @@ mod tests {
         (client, temp_dir)
     }
 
+    #[tokio::test]
+    async fn local_key_directory_outage_is_io_error_and_recovers_original_key() {
+        let root = TempDir::new().expect("create isolated key store");
+        let key_dir = root.path().join("keys");
+        let unavailable_dir = root.path().join("keys-unavailable");
+        let config = KmsConfig::local(key_dir.clone()).with_insecure_development_defaults();
+        let backend = LocalKmsBackend::new(config).await.expect("start Local KMS");
+        let key_id = "directory-outage-key";
+        backend
+            .create_key(CreateKeyRequest {
+                key_name: Some(key_id.to_string()),
+                ..Default::default()
+            })
+            .await
+            .expect("create the original key");
+        let request = |key_id: &str| GenerateDataKeyRequest {
+            key_id: key_id.to_string(),
+            key_spec: KeySpec::Aes256,
+            encryption_context: HashMap::new(),
+        };
+        let before = backend
+            .generate_data_key(request(key_id))
+            .await
+            .expect("generate a data key before the outage");
+        let missing_key = backend.generate_data_key(request("no-such-key")).await;
+        let key_path = key_dir.join(format!("{key_id}.key"));
+        let original_record = fs::read(&key_path).await.expect("read the original key record");
+
+        fs::rename(&key_dir, &unavailable_dir)
+            .await
+            .expect("make the key directory unavailable");
+        let unavailable = backend.generate_data_key(request(key_id)).await;
+        // Restore before checking the error so the failing regression leaves no
+        // orphaned key store; both paths also belong to the same temporary root.
+        fs::rename(&unavailable_dir, &key_dir)
+            .await
+            .expect("restore the original key directory");
+
+        let after = backend
+            .generate_data_key(request(key_id))
+            .await
+            .expect("generate a data key after directory restoration");
+        for data_key in [&before, &after] {
+            let decrypted = backend
+                .decrypt(DecryptRequest {
+                    ciphertext: data_key.ciphertext_blob.clone(),
+                    encryption_context: HashMap::new(),
+                    grant_tokens: Vec::new(),
+                })
+                .await
+                .expect("the original master key must decrypt both data keys");
+            assert!(
+                decrypted.plaintext == data_key.plaintext_key,
+                "directory restoration must preserve the original key material"
+            );
+        }
+        assert!(
+            fs::read(&key_path).await.expect("read the restored key record") == original_record,
+            "reads and recovery must not rewrite the key record"
+        );
+        assert!(
+            matches!(missing_key, Err(KmsError::KeyNotFound { key_id }) if key_id == "no-such-key"),
+            "a missing key in a readable directory must remain KeyNotFound"
+        );
+        assert!(
+            matches!(unavailable, Err(KmsError::IoError { .. })),
+            "an unavailable key directory must remain an I/O error, not KeyNotFound"
+        );
+    }
+
     /// With the AAD write switch on, the Local backend seals the stored
     /// encryption context into the wrap exactly like KV2: the bound envelope
     /// round-trips, a rewritten stored context fails authentication even with