diff --git a/.config/e2e-full-selection.txt b/.config/e2e-full-selection.txt index ceb69305a..1ab608350 100644 --- a/.config/e2e-full-selection.txt +++ b/.config/e2e-full-selection.txt @@ -1,2 +1,2 @@ sha256-darwin=874c881d7b45f12378a5817c7f42c95c4981960a2ec9ce12dcf4af239ae1f9d5 -sha256-linux=9515861be899ceb10e2e0ef93c34208bb7a7a8a7f8067a02db4cfba23270ebd6 +sha256-linux=9351e25b45bf7dfce18b951a5e3740225f457cacc53b8bf9f500f6947763ec0e diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 1c17a9b38..11b2ed59e 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -15661,25 +15661,24 @@ mod tests { assert!(deleted[0].found, "the aggregate error must retain the committed pool result"); drop(injection); - tokio::time::timeout(Duration::from_secs(30), async { - loop { - let mut metadata_absent = true; - for pool in &store.pools { - metadata_absent &= pool - .get_disks_by_key(object) - .load_file_info_versions_exact(bucket, object) - .await - .expect("aggregate-error cleanup metadata should remain readable") - .is_none(); - } - if metadata_absent && backend.remove_count().await == 1 { - return; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - }) - .await - .expect("aggregate failure must not suppress committed receipt dispatch"); + // Exact reads can see subquorum metadata while workers remove each + // disk's free version. Inspect the final state after cleanup drains. + wait_for_expiry_workers_idle(&store).await; + for pool in &store.pools { + assert!( + pool.get_disks_by_key(object) + .load_file_info_versions_exact(bucket, object) + .await + .expect("aggregate-error cleanup metadata should remain readable") + .is_none(), + "aggregate failure must not suppress committed receipt cleanup" + ); + } + assert_eq!( + backend.remove_count().await, + 1, + "committed receipts must remove the shared remote object once" + ); assert_eq!(backend.object_count().await, 0, "the shared remote object should be removed exactly once"); store .delete_bucket(bucket, &DeleteBucketOptions::default()) diff --git a/crates/kms/src/backends/vault_credentials.rs b/crates/kms/src/backends/vault_credentials.rs index fa9b107c7..a79269cd5 100644 --- a/crates/kms/src/backends/vault_credentials.rs +++ b/crates/kms/src/backends/vault_credentials.rs @@ -1321,6 +1321,38 @@ mod tests { .expect("token auth must map to a source"); } + /// A pending acquisition cannot take the returned-error fallback: the + /// outer login policy must cut it off before publishing a client. + #[tokio::test(start_paused = true)] + async fn test_stalled_initial_login_is_bounded_by_the_attempt_timeout() { + let state = Arc::new(ScriptedState::default()); + let source = ScriptedSource { + state: state.clone(), + ttl: Duration::ZERO, + renewable: false, + login_delay: Duration::from_secs(60), + }; + let policy = test_policy(Duration::from_secs(10), Duration::from_secs(5)); + let attempt_timeout = policy.retry.attempt_timeout; + let started = Instant::now(); + let result = VaultCredentialProvider::new(test_settings(), Box::new(source), policy).await; + assert!( + matches!(&result, Err(KmsError::OperationTimedOut { message }) if message.starts_with("vault_login attempt 1 timed out")), + "a stalled login must return its typed timeout without publishing a client" + ); + assert_eq!( + started.elapsed(), + attempt_timeout, + "login must consume exactly one virtual attempt budget" + ); + assert_eq!(state.login_calls.load(Ordering::SeqCst), 0, "the acquisition must not complete"); + assert_eq!( + state.renew_calls.load(Ordering::SeqCst), + 0, + "failed initialization must not start renewal" + ); + } + /// backlog#2369 P3: `vault token create` defaults to a 768-hour TTL, so /// hard-coding "no lease" for token auth left the renewal task unstarted /// and turned a healthy cluster into one that answers 403 a month later. diff --git a/crates/kms/tests/vault_fault_injection.rs b/crates/kms/tests/vault_fault_injection.rs index b84c9c5d7..bc772da2c 100644 --- a/crates/kms/tests/vault_fault_injection.rs +++ b/crates/kms/tests/vault_fault_injection.rs @@ -14,8 +14,7 @@ //! Fault-injection matrix for the Vault backend operation policy. //! -//! Offline cases run against locally injected transport faults (a listener -//! that never responds) — deterministic, no external +//! Offline cases run against locally injected HTTP and transport faults — no external //! dependencies. Real-Vault cases are `#[ignore]`d and need a dev Vault //! (default `http://127.0.0.1:8200`, override with `RUSTFS_KMS_VAULT_ADDR`). //! @@ -38,9 +37,100 @@ use rustfs_kms::backends::vault::VaultKmsBackend; use rustfs_kms::{ BackendConfig, DescribeKeyRequest, KmsBackend as KmsBackendKind, KmsConfig, KmsError, VaultAuthMethod, VaultConfig, }; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::TcpListener; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; const OPERATIONS_TOTAL: &str = "rustfs_kms_backend_operations_total"; const ATTEMPT_FAILURES_TOTAL: &str = "rustfs_kms_backend_attempt_failures_total"; +const LOGIN: &str = "vault_login"; +const READ_KEY: &str = "vault_kv2_read_key"; +const LOOKUP_REQUEST: &str = "GET /v1/auth/token/lookup-self HTTP/1.1"; + +/// Unlike the unit-test scripted Vault, this fixture records the credential +/// probe too and can fail it independently of the subsequent key request. +/// `None` parks a connection without responding; extra requests receive 599. +struct FaultVault { + address: String, + requests: mpsc::UnboundedReceiver, + task: JoinHandle<()>, +} + +impl FaultVault { + async fn serve(responses: Vec>) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind fault-injection Vault"); + let address = format!("http://{}", listener.local_addr().expect("fault-injection Vault address")); + let (recorded, requests) = mpsc::unbounded_channel(); + let task = tokio::spawn(async move { + let mut responses = responses.into_iter(); + let mut parked = Vec::new(); + loop { + let (stream, _) = listener.accept().await.expect("accept Vault request"); + let mut stream = BufReader::new(stream); + let mut line = String::new(); + assert_ne!(stream.read_line(&mut line).await.expect("read request line"), 0); + recorded.send(line.trim_end().to_string()).expect("record Vault request"); + loop { + line.clear(); + assert_ne!(stream.read_line(&mut line).await.expect("read request header"), 0); + if line == "\r\n" { + break; + } + } + let mut stream = stream.into_inner(); + let response = responses + .next() + .unwrap_or_else(|| Some((599, serde_json::json!({"errors": ["unexpected Vault request"]})))); + if let Some((status, body)) = response { + let body = body.to_string(); + let response = format!( + "HTTP/1.1 {status} Scripted\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ); + stream.write_all(response.as_bytes()).await.expect("write Vault response"); + stream.shutdown().await.expect("close Vault response"); + } else { + parked.push(stream); + } + } + }); + Self { address, requests, task } + } + + async fn finish(&mut self) { + assert!(self.requests.try_recv().is_err(), "no unexpected requests may remain"); + self.task.abort(); + let error = (&mut self.task).await.expect_err("fault server runs until aborted"); + assert!(error.is_cancelled(), "fault server must not panic: {error}"); + } +} + +impl Drop for FaultVault { + fn drop(&mut self) { + self.task.abort(); + } +} + +fn healthy_token_lookup() -> serde_json::Value { + serde_json::json!({ + "data": { + "accessor": "fault-injection-accessor", + "creation_time": 1_700_000_000u64, + "creation_ttl": 0, + "display_name": "token", + "entity_id": "", + "explicit_max_ttl": 0, + "id": "unused", + "num_uses": 0, + "orphan": true, + "path": "auth/token/create", + "policies": ["default"], + "renewable": false, + "ttl": 0 + } + }) +} fn vault_config(address: &str, token: &str) -> VaultConfig { VaultConfig { @@ -125,39 +215,112 @@ fn counter_value(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)]) fn stalled_connection_is_cut_off_by_the_attempt_timeout() { let snapshot = record_metrics(|| { Box::pin(async move { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + let mut vault = FaultVault::serve(vec![Some((200, healthy_token_lookup())), None]).await; + let attempt_timeout = Duration::from_millis(250); + let client = VaultKmsBackend::new(kms_config(vault_config(&vault.address, "unused"), attempt_timeout, 1)) .await - .expect("bind stall listener"); - let address = format!("http://{}", listener.local_addr().expect("stall listener addr")); - // Accept and park every connection without ever responding. - tokio::spawn(async move { - let mut parked = Vec::new(); - loop { - let Ok((socket, _)) = listener.accept().await else { return }; - parked.push(socket); - } - }); + .expect("the token lookup must succeed before injecting the stalled key read"); + assert_eq!(vault.requests.try_recv().as_deref(), Ok(LOOKUP_REQUEST)); - let client = VaultKmsBackend::new(kms_config(vault_config(&address, "unused"), Duration::from_millis(250), 1)) - .await - .expect("client construction performs no network calls"); - let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-stalled")) + let read = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-stalled")); + tokio::pin!(read); + tokio::select! { + request = vault.requests.recv() => assert_eq!( + request.as_deref(), + Some("GET /v1/secret/data/rustfs/kms/fault-injection/fault-injection-stalled? HTTP/1.1") + ), + result = &mut read => panic!("the key request must reach the stall listener: {result:?}"), + } + // Pause only after real loopback I/O reaches the intended request; + // otherwise auto-advancing time could expire the login instead. + tokio::time::pause(); + let stalled_at = tokio::time::Instant::now(); + // Tokio rounds timer deadlines up to the next millisecond. + let virtual_step = attempt_timeout + Duration::from_millis(1); + tokio::time::advance(virtual_step).await; + let error = tokio::time::timeout(Duration::from_secs(1), read) .await + .expect("the attempt timer must resolve without further network activity") .expect_err("a stalled request must be cut off by the attempt timeout"); + assert_eq!( + stalled_at.elapsed(), + virtual_step, + "the read must resolve within the attempt budget plus one timer tick" + ); assert!( matches!(error, KmsError::OperationTimedOut { .. } | KmsError::BackendError { .. }), "got {error:?}" ); + vault.finish().await; }) }); // The policy timer reports attempt_timeout; the client-level HTTP timeout // surfaces as a connection-class failure. Either way it is exactly one // attempt that was cut off. - let cut_off = counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "attempt_timeout")]) - + counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "retryable_conn")]); + let cut_off = counter_value( + &snapshot, + ATTEMPT_FAILURES_TOTAL, + &[("operation", READ_KEY), ("error_class", "attempt_timeout")], + ) + counter_value( + &snapshot, + ATTEMPT_FAILURES_TOTAL, + &[("operation", READ_KEY), ("error_class", "retryable_conn")], + ); assert_eq!(cut_off, 1, "the single budgeted attempt must be cut off by a timeout"); - assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]), 1); + assert_eq!(counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("operation", LOGIN)]), 0); + assert_eq!( + counter_value(&snapshot, OPERATIONS_TOTAL, &[("operation", LOGIN), ("outcome", "success")]), + 1 + ); + assert_eq!( + counter_value(&snapshot, OPERATIONS_TOTAL, &[("operation", READ_KEY), ("outcome", "budget_exhausted")]), + 1 + ); +} + +/// A returned lookup error degrades lease discovery, not Vault authorization: +/// the subsequent forbidden key read must still fail once, without retrying. +#[test] +fn token_lookup_errors_do_not_bypass_key_authorization() { + for lookup_status in [403, 503] { + let snapshot = record_metrics(|| { + Box::pin(async move { + let mut vault = FaultVault::serve(vec![ + Some((lookup_status, serde_json::json!({"errors": ["token lookup unavailable"]}))), + Some((403, serde_json::json!({"errors": ["permission denied"]}))), + ]) + .await; + let client = VaultKmsBackend::new(kms_config(vault_config(&vault.address, "unused"), Duration::from_secs(5), 3)) + .await + .expect("a returned token lookup error must preserve static-token fallback"); + assert_eq!(vault.requests.try_recv().as_deref(), Ok(LOOKUP_REQUEST)); + let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-forbidden")) + .await + .expect_err("lease discovery fallback must not authorize a forbidden key read"); + assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}"); + assert_eq!( + vault.requests.try_recv().as_deref(), + Ok("GET /v1/secret/data/rustfs/kms/fault-injection/fault-injection-forbidden? HTTP/1.1") + ); + vault.finish().await; + }) + }); + assert_eq!(counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("operation", LOGIN)]), 0); + assert_eq!( + counter_value(&snapshot, OPERATIONS_TOTAL, &[("operation", LOGIN), ("outcome", "success")]), + 1 + ); + assert_eq!(counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("operation", READ_KEY)]), 1); + assert_eq!( + counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("operation", READ_KEY), ("error_class", "fatal")]), + 1 + ); + assert_eq!( + counter_value(&snapshot, OPERATIONS_TOTAL, &[("operation", READ_KEY), ("outcome", "fatal")]), + 1 + ); + } } fn real_vault_address() -> String { @@ -174,7 +337,7 @@ fn real_vault_invalid_token_is_fatal_and_never_retried() { let config = vault_config(&real_vault_address(), "fault-injection-invalid-token"); let client = VaultKmsBackend::new(kms_config(config, Duration::from_secs(5), 3)) .await - .expect("client construction performs no network calls"); + .expect("a returned token lookup error must preserve static-token fallback"); let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-forbidden")) .await .expect_err("an invalid token must be rejected"); @@ -183,13 +346,20 @@ fn real_vault_invalid_token_is_fatal_and_never_retried() { }); assert_eq!( - counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "fatal")]), + counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("operation", READ_KEY), ("error_class", "fatal")]), 1, "a 403 must be observed by exactly one attempt" ); - assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "fatal")]), 1); assert_eq!( - counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "retryable_status")]), + counter_value(&snapshot, OPERATIONS_TOTAL, &[("operation", READ_KEY), ("outcome", "fatal")]), + 1 + ); + assert_eq!( + counter_value( + &snapshot, + ATTEMPT_FAILURES_TOTAL, + &[("operation", READ_KEY), ("error_class", "retryable_status")] + ), 0, "an auth failure must never be classified as retryable" ); @@ -207,7 +377,7 @@ fn real_vault_missing_key_is_resolved_in_one_attempt() { let config = vault_config(&real_vault_address(), &token); let client = VaultKmsBackend::new(kms_config(config, Duration::from_secs(5), 3)) .await - .expect("client construction performs no network calls"); + .expect("static-token initialization must complete against the running Vault"); let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-definitely-missing")) .await .expect_err("a missing key must resolve to key-not-found"); @@ -216,9 +386,12 @@ fn real_vault_missing_key_is_resolved_in_one_attempt() { }); assert_eq!( - counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "fatal")]), + counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("operation", READ_KEY), ("error_class", "fatal")]), 1, "a 404 must be observed by exactly one attempt" ); - assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "fatal")]), 1); + assert_eq!( + counter_value(&snapshot, OPERATIONS_TOTAL, &[("operation", READ_KEY), ("outcome", "fatal")]), + 1 + ); } diff --git a/scripts/ci_gate.py b/scripts/ci_gate.py index 4c0ae574c..74442691f 100644 --- a/scripts/ci_gate.py +++ b/scripts/ci_gate.py @@ -59,7 +59,7 @@ def expected_results(mode: str, event: str, ref: str) -> dict[str, str]: expected.update({job: "success" if mode == "full" else "skipped" for job in CODE_JOBS}) rio = mode == "full" and event in ("schedule", "workflow_dispatch") expected.update({job: "success" if rio else "skipped" for job in OPTIONAL_JOBS[:2]}) - full = mode == "full" and (event in ("merge_group", "workflow_dispatch") or (event == "push" and ref == "refs/heads/main")) + full = mode == "full" and (event in ("merge_group", "workflow_dispatch") or (event == "push" and ref in ("refs/heads/main", "refs/heads/release"))) expected["e2e-full"] = "success" if full else "skipped" return expected @@ -219,6 +219,27 @@ class SelfTests(unittest.TestCase): bad = {**good, "classify-changes": {"result": "success", "outputs": selection}} self.assertTrue(verify_results(bad, event, "refs/heads/main")) + def test_full_e2e_gate_preserves_workflow_branch_and_event_scope(self): + for event, ref, required in ( + ("push", "refs/heads/main", "success"), + ("push", "refs/heads/release", "success"), + ("push", "refs/heads/feature", "skipped"), + ("push", "refs/heads/release-candidate", "skipped"), + ("push", "refs/tags/release", "skipped"), + ("pull_request", "refs/pull/1/merge", "skipped"), + ("schedule", "refs/heads/release", "skipped"), + ("workflow_dispatch", "refs/heads/feature", "success"), + ("merge_group", "refs/heads/gh-readonly-queue/release/pr-1", "success"), + ): + with self.subTest(event=event, ref=ref): + expected = expected_results("full", event, ref) + self.assertEqual(expected["e2e-full"], required) + needs = {job: {"result": result} for job, result in expected.items()} + needs["classify-changes"]["outputs"] = {"mode": "full"} + for result in ("success", "skipped", "failure", "cancelled"): + needs["e2e-full"]["result"] = result + self.assertEqual(verify_results(needs, event, ref) == [], result == required) + def test_repository_wiring_and_missing_dependency_regression(self): self.assertEqual(check_workflow(ROOT), []) with tempfile.TemporaryDirectory() as directory: