fix(kms): cover concurrent Vault KV2 rotation races (#5632)

* fix(kms): handle concurrent Vault KV2 baseline races

* test(kms): keep concurrent rotation regression fail-closed

crates/kms/src/backends/vault.rs

@@ -1895,6 +1895,19 @@ mod tests {
         (vault, client)
     }
 
+    async fn scripted_kv2_client(key_data: &VaultKeyData) -> (ScriptedVault, VaultKmsClient) {
+        let vault = ScriptedVault::serve_kv2(
+            "rustfs/kms/keys/wired-key",
+            serde_json::to_value(key_data).expect("serialize scripted KV2 key"),
+        )
+        .await;
+        let (vault_config, kms_config) = scripted_configs(&vault.address);
+        let client = VaultKmsClient::new(vault_config, &kms_config)
+            .await
+            .expect("scripted Vault client");
+        (vault, client)
+    }
+
     fn healthy_key_data() -> VaultKeyData {
         VaultKeyData {
             algorithm: "AES_256".to_string(),
@@ -3223,6 +3236,102 @@ mod tests {
         );
     }
 
+    /// Concurrent rotations use KV2 create-only records and a CAS pointer
+    /// switch. The stateful scripted Vault applies those preconditions to real
+    /// HTTP requests, so this test proves committed versions are unique and
+    /// contiguous instead of only checking that several calls returned.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn wired_concurrent_kv2_rotations_commit_unique_monotonic_versions() {
+        use std::collections::HashSet;
+        use std::sync::Arc;
+
+        const ATTEMPTS: usize = 8;
+        let mut key_data = healthy_key_data();
+        key_data.baseline_version = Some(1);
+        let (vault, client) = scripted_kv2_client(&key_data).await;
+        let client = Arc::new(client);
+        let barrier = Arc::new(tokio::sync::Barrier::new(ATTEMPTS));
+        let tasks: Vec<_> = (0..ATTEMPTS)
+            .map(|_| {
+                let client = Arc::clone(&client);
+                let barrier = Arc::clone(&barrier);
+                tokio::spawn(async move {
+                    barrier.wait().await;
+                    client.rotate_key("wired-key", None).await
+                })
+            })
+            .collect();
+
+        let mut committed_versions = Vec::new();
+        let mut errors = Vec::new();
+        for task in tasks {
+            match task.await.expect("join concurrent rotation task") {
+                Ok(result) => committed_versions.push(result.version),
+                Err(error) => errors.push(error),
+            }
+        }
+
+        assert!(
+            !committed_versions.is_empty(),
+            "at least one concurrent rotation must commit; errors: {errors:?}"
+        );
+        assert!(
+            errors.iter().all(
+                |error| matches!(error, KmsError::InvalidOperation { message } if message.contains("Concurrent modification"))
+            ),
+            "a lost CAS race must be the only expected failure: {errors:?}"
+        );
+
+        let mut sorted_versions = committed_versions.clone();
+        sorted_versions.sort_unstable();
+        let unique_versions: HashSet<_> = sorted_versions.iter().copied().collect();
+        assert_eq!(
+            unique_versions.len(),
+            sorted_versions.len(),
+            "concurrent rotations must never return a version twice: {committed_versions:?}"
+        );
+
+        let successful_rotations = u32::try_from(sorted_versions.len()).expect("test attempts fit u32");
+        let current_version = 1u32 + successful_rotations;
+        assert_eq!(
+            sorted_versions,
+            (2..=current_version).collect::<Vec<_>>(),
+            "committed versions must form one monotonic sequence: {sorted_versions:?}"
+        );
+
+        let snapshot = vault.kv2_snapshot().expect("stateful KV2 snapshot");
+        let persisted_version = snapshot.current_data["version"]
+            .as_u64()
+            .and_then(|version| u32::try_from(version).ok())
+            .expect("current KV2 record must carry a u32 key version");
+        assert_eq!(persisted_version, current_version);
+        assert!(
+            snapshot.current_secret_version >= u64::from(current_version),
+            "the KV2 secret version must advance with each committed pointer switch"
+        );
+        assert_eq!(
+            snapshot.version_records.keys().copied().collect::<Vec<_>>(),
+            (1..=current_version).collect::<Vec<_>>(),
+            "every committed KMS version must have exactly one immutable record"
+        );
+
+        let mut materials = HashSet::new();
+        for (version, record) in &snapshot.version_records {
+            let material = record["encrypted_key_material"]
+                .as_str()
+                .expect("version record must carry encrypted key material");
+            assert!(materials.insert(material), "version {version} reuses another version's material");
+        }
+        assert_eq!(
+            snapshot.current_data["encrypted_key_material"],
+            snapshot
+                .version_records
+                .get(&current_version)
+                .expect("current version record")["encrypted_key_material"],
+            "the top-level fast path must match the current immutable version record"
+        );
+    }
+
     /// The tags write-back after a create is a check-and-set read-modify-write
     /// that carries the key material over from the freshly read record.
     #[tokio::test]