test(kms): stabilize Vault failover validation (#6385)

* test(kms): bound Vault failover progress wait

* ci(nightly): honor manual dispatch ref

* test(kms): preserve Vault worker failures

* test(kms): validate Vault circuit recovery

.github/workflows/nightly-gnu.yml

@@ -39,11 +39,10 @@ jobs:
     env:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
     steps:
-      - name: Checkout main branch
+      - name: Checkout repository
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
-          ref: main
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup
@@ -89,11 +88,10 @@ jobs:
       # either casing.
       NO_PROXY: 127.0.0.1,localhost
     steps:
-      - name: Checkout main branch
+      - name: Checkout repository
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
-          ref: main
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup
@@ -178,11 +176,10 @@ jobs:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
       NO_PROXY: 127.0.0.1,localhost
     steps:
-      - name: Checkout main branch
+      - name: Checkout repository
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
-          ref: main
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup

crates/ecstore/src/io_support/bitrot.rs

@@ -784,6 +784,24 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle(
 ///
 /// # Returns
 /// A Result containing the BitrotWriterWrapper or an error
+/// Size hint handed to `DiskAPI::create_file` for a bitrot-wrapped shard.
+///
+/// A known length is grown by one checksum per shard so the on-disk file size
+/// matches what the bitrot writer emits. A negative length is the
+/// unknown-size sentinel (`HashReader::SIZE_PRESERVE_LAYER`, used by SSE and
+/// compression) and must be preserved: `RemoteDisk::create_file` forwards it
+/// in the `put_file_stream` query, and the receiver only treats `size > 0` as
+/// a fixed body length when locating the authenticated trailer. Clamping it
+/// to `0` would claim an empty body and misframe the stream. `0` stays `0`
+/// because a genuinely empty object still means an empty body.
+fn bitrot_create_file_size(length: i64, shard_size: usize, checksum_algo: &HashAlgorithm) -> i64 {
+    if length <= 0 {
+        return length;
+    }
+    let length = length as usize;
+    (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
+}
+
 pub async fn create_bitrot_writer(
     is_inline_buffer: bool,
     disk: Option<&DiskStore>,
@@ -796,12 +814,7 @@ pub async fn create_bitrot_writer(
     let writer = if is_inline_buffer {
         CustomWriter::new_inline_buffer()
     } else if let Some(disk) = disk {
-        let length = if length > 0 {
-            let length = length as usize;
-            (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
-        } else {
-            0
-        };
+        let length = bitrot_create_file_size(length, shard_size, &checksum_algo);
 
         let file = disk.create_file("", volume, path, length).await?;
         #[cfg(feature = "hotpath")]
@@ -820,6 +833,25 @@ mod tests {
     use rustfs_rio::ChunkReader;
     use std::collections::VecDeque;
 
+    #[test]
+    fn bitrot_create_file_size_grows_known_length_by_checksums() {
+        // 10 bytes over 4-byte shards = 3 shards, each followed by a 32-byte hash.
+        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::HighwayHash256), 10 + 3 * 32);
+        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::None), 10);
+    }
+
+    #[test]
+    fn bitrot_create_file_size_keeps_empty_and_unknown_distinct() {
+        assert_eq!(bitrot_create_file_size(0, 4, &HashAlgorithm::HighwayHash256), 0);
+        // SSE/compression streams advertise SIZE_PRESERVE_LAYER (-1); the remote
+        // put_file_stream receiver relies on a non-positive size to parse the auth
+        // trailer from the stream tail, so the sentinel must survive untouched.
+        assert_eq!(
+            bitrot_create_file_size(rustfs_rio::HashReader::SIZE_PRESERVE_LAYER, 4, &HashAlgorithm::HighwayHash256),
+            rustfs_rio::HashReader::SIZE_PRESERVE_LAYER
+        );
+    }
+
     struct TestChunkReader {
         chunks: VecDeque<Bytes>,
     }

crates/kms/tests/vault_ha_failover_live.rs

@@ -16,14 +16,14 @@
 //!
 //! `scripts/test/vault_ha_kms_live.sh` owns the official Vault containers and
 //! kills the active node while this test continuously decrypts through a
-//! surviving standby. KV2 and Transit requests must remain successful, use a
-//! bounded number of attempts, and leave the circuit and in-flight gauges at
-//! zero after a new leader is elected.
+//! surviving standby. KV2 and Transit must recover after the bounded circuit
+//! interval, use a bounded number of attempts, and leave the circuit and
+//! in-flight gauges at zero after a new leader is elected.
 
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
-use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use metrics_util::MetricKind;
@@ -43,6 +43,11 @@ const OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
 const IN_FLIGHT: &str = "rustfs_kms_backend_in_flight";
 const CIRCUIT_OPEN: &str = "rustfs_kms_backend_circuit_open";
 const MAX_ATTEMPTS: u32 = 10;
+const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(2);
+const HEALTHY_PROGRESS_TIMEOUT: Duration = Duration::from_secs(20);
+// The circuit remains open for 30s after five failed attempts.
+const POST_FAILOVER_PROGRESS_TIMEOUT: Duration = Duration::from_secs(35);
+const FAILOVER_ERROR_POLL_INTERVAL: Duration = Duration::from_millis(100);
 
 type MetricEntry = (
     metrics_util::CompositeKey,
@@ -64,7 +69,7 @@ fn config(backend: KmsBackend, backend_config: BackendConfig) -> KmsConfig {
         backend,
         backend_config,
         allow_insecure_dev_defaults: true,
-        timeout: Duration::from_secs(2),
+        timeout: ATTEMPT_TIMEOUT,
         retry_attempts: MAX_ATTEMPTS,
         enable_cache: false,
         ..KmsConfig::default()
@@ -164,14 +169,31 @@ fn retryable_failures(snapshot: &[MetricEntry], operation: &str) -> u64 {
         .sum()
 }
 
-async fn wait_for_count(counter: &AtomicU64, minimum: u64, description: &str) {
-    tokio::time::timeout(Duration::from_secs(20), async {
+async fn wait_for_count(
+    counter: &AtomicU64,
+    failure: &Mutex<Option<String>>,
+    minimum: u64,
+    description: &str,
+    timeout: Duration,
+) {
+    tokio::time::timeout(timeout, async {
         while counter.load(Ordering::SeqCst) < minimum {
+            if let Some(error) = failure.lock().expect("decrypt failure lock poisoned").as_ref() {
+                panic!(
+                    "{description} worker failed after {} successful decrypts: {error}",
+                    counter.load(Ordering::SeqCst)
+                );
+            }
             tokio::time::sleep(Duration::from_millis(25)).await;
         }
     })
     .await
-    .unwrap_or_else(|_| panic!("timed out waiting for {description}"));
+    .unwrap_or_else(|_| {
+        panic!(
+            "timed out after {timeout:?} waiting for {description}: completed {}, expected {minimum}",
+            counter.load(Ordering::SeqCst)
+        )
+    });
 }
 
 async fn wait_for_file(path: &Path, description: &str) {
@@ -189,7 +211,8 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
     request: DecryptRequest,
     expected: Vec<u8>,
     completed: Arc<AtomicU64>,
-    failed: Arc<AtomicBool>,
+    allow_failover_errors: Arc<AtomicBool>,
+    failure: Arc<Mutex<Option<String>>>,
     stop: CancellationToken,
 ) {
     while !stop.is_cancelled() {
@@ -197,8 +220,18 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
             Ok(response) if response.plaintext == expected => {
                 completed.fetch_add(1, Ordering::SeqCst);
             }
-            Ok(_) | Err(_) => {
-                failed.store(true, Ordering::SeqCst);
+            Ok(_) => {
+                *failure.lock().expect("decrypt failure lock poisoned") =
+                    Some("decrypt returned unexpected plaintext".to_string());
+                return;
+            }
+            Err(rustfs_kms::KmsError::BackendError { .. } | rustfs_kms::KmsError::OperationTimedOut { .. })
+                if allow_failover_errors.load(Ordering::SeqCst) =>
+            {
+                tokio::time::sleep(FAILOVER_ERROR_POLL_INTERVAL).await;
+            }
+            Err(error) => {
+                *failure.lock().expect("decrypt failure lock poisoned") = Some(error.to_string());
                 return;
             }
         }
@@ -296,7 +329,9 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
     );
 
     let stop = CancellationToken::new();
-    let failed = Arc::new(AtomicBool::new(false));
+    let allow_failover_errors = Arc::new(AtomicBool::new(false));
+    let kv2_failure = Arc::new(Mutex::new(None));
+    let transit_failure = Arc::new(Mutex::new(None));
     let kv2_completed = Arc::new(AtomicU64::new(0));
     let transit_completed = Arc::new(AtomicU64::new(0));
     let kv2_worker = tokio::spawn(decrypt_loop(
@@ -304,7 +339,8 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
         kv2_request,
         kv2_data_key.plaintext_key,
         Arc::clone(&kv2_completed),
-        Arc::clone(&failed),
+        Arc::clone(&allow_failover_errors),
+        Arc::clone(&kv2_failure),
         stop.clone(),
     ));
     let transit_worker = tokio::spawn(decrypt_loop(
@@ -312,12 +348,21 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
         transit_request,
         transit_data_key.plaintext_key,
         Arc::clone(&transit_completed),
-        Arc::clone(&failed),
+        Arc::clone(&allow_failover_errors),
+        Arc::clone(&transit_failure),
         stop.clone(),
     ));
 
-    wait_for_count(&kv2_completed, 2, "two healthy KV2 decrypts").await;
-    wait_for_count(&transit_completed, 2, "two healthy Transit decrypts").await;
+    wait_for_count(&kv2_completed, &kv2_failure, 2, "two healthy KV2 decrypts", HEALTHY_PROGRESS_TIMEOUT).await;
+    wait_for_count(
+        &transit_completed,
+        &transit_failure,
+        2,
+        "two healthy Transit decrypts",
+        HEALTHY_PROGRESS_TIMEOUT,
+    )
+    .await;
+    allow_failover_errors.store(true, Ordering::SeqCst);
     std::fs::write(&marker, b"ready").expect("publish failover readiness marker");
 
     wait_for_file(&elected, "the replacement Vault leader").await;
@@ -326,18 +371,39 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
 
     let kv2_after_election = kv2_completed.load(Ordering::SeqCst) + 2;
     let transit_after_election = transit_completed.load(Ordering::SeqCst) + 2;
-    wait_for_count(&kv2_completed, kv2_after_election, "post-failover KV2 decrypts").await;
-    wait_for_count(&transit_completed, transit_after_election, "post-failover Transit decrypts").await;
+    wait_for_count(
+        &kv2_completed,
+        &kv2_failure,
+        kv2_after_election,
+        "post-failover KV2 decrypts",
+        POST_FAILOVER_PROGRESS_TIMEOUT,
+    )
+    .await;
+    wait_for_count(
+        &transit_completed,
+        &transit_failure,
+        transit_after_election,
+        "post-failover Transit decrypts",
+        POST_FAILOVER_PROGRESS_TIMEOUT,
+    )
+    .await;
 
     stop.cancel();
     kv2_worker.await.expect("KV2 decrypt worker must join");
     transit_worker.await.expect("Transit decrypt worker must join");
-    assert!(!failed.load(Ordering::SeqCst), "no decrypt may fail or return different plaintext");
+    assert!(
+        kv2_failure.lock().expect("KV2 failure lock poisoned").is_none(),
+        "no KV2 decrypt may fail or return different plaintext"
+    );
+    assert!(
+        transit_failure.lock().expect("Transit failure lock poisoned").is_none(),
+        "no Transit decrypt may fail or return different plaintext"
+    );
 }
 
 #[test]
 #[ignore = "requires a real three-node Vault Raft cluster; run scripts/test/vault_ha_kms_live.sh"]
-fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
+fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
     let recorder = DebuggingRecorder::new();
     let snapshotter = recorder.snapshotter();
     metrics::with_local_recorder(&recorder, || {
@@ -349,11 +415,6 @@ fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
     });
     let snapshot = snapshotter.snapshot().into_vec();
 
-    assert_eq!(
-        counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "circuit_open")]),
-        0,
-        "a bounded leader election must not open the circuit"
-    );
     assert_eq!(
         counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]),
         0,

scripts/test/vault_ha_kms_live.sh

@@ -241,7 +241,7 @@ env \
   RUSTFS_TEST_VAULT_FAILOVER_MARKER="$MARKER" \
   RUSTFS_TEST_VAULT_OLD_LEADER="$OLD_LEADER" \
   cargo test -p rustfs-kms --test vault_ha_failover_live \
-    vault_raft_leader_failure_preserves_kv2_and_transit_decrypts -- \
+    vault_raft_leader_failure_recovers_kv2_and_transit_decrypts -- \
     --ignored --nocapture --test-threads=1 &
 TEST_PID=$!