diff --git a/.github/workflows/nightly-gnu.yml b/.github/workflows/nightly-gnu.yml index 946761e54..a1f0260e1 100644 --- a/.github/workflows/nightly-gnu.yml +++ b/.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 diff --git a/crates/kms/tests/vault_ha_failover_live.rs b/crates/kms/tests/vault_ha_failover_live.rs index 197198d5d..13ece1e03 100644 --- a/crates/kms/tests/vault_ha_failover_live.rs +++ b/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>, + 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( request: DecryptRequest, expected: Vec, completed: Arc, - failed: Arc, + allow_failover_errors: Arc, + failure: Arc>>, stop: CancellationToken, ) { while !stop.is_cancelled() { @@ -197,8 +220,18 @@ async fn decrypt_loop( 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, diff --git a/scripts/test/vault_ha_kms_live.sh b/scripts/test/vault_ha_kms_live.sh index 59b2ff2fe..786874e16 100755 --- a/scripts/test/vault_ha_kms_live.sh +++ b/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=$!