gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-10 14:16:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ci): align release E2E gate and repair regression tests (#7607)

Browse Source
This commit is contained in:
Zhengchao An
2026-09-10 12:03:27 +08:00
committed by GitHub
parent 110f630a5c
commit 70fb8bf504
5 changed files with 273 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.config/e2e-full-selection.txt
+1 -1
View File
@@ -1,2 +1,2 @@
sha256-darwin=874c881d7b45f12378a5817c7f42c95c4981960a2ec9ce12dcf4af239ae1f9d5
sha256-linux=9515861be899ceb10e2e0ef93c34208bb7a7a8a7f8067a02db4cfba23270ebd6
sha256-linux=9351e25b45bf7dfce18b951a5e3740225f457cacc53b8bf9f500f6947763ec0e
crates/ecstore/src/store/init.rs
+18 -19
View File
@@ -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())
crates/kms/src/backends/vault_credentials.rs
+32
View File
@@ -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.
crates/kms/tests/vault_fault_injection.rs
+200 -27
View File
@@ -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<String>,
task: JoinHandle<()>,
}
impl FaultVault {
async fn serve(responses: Vec<Option<(u16, serde_json::Value)>>) -> 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
);
}
scripts/ci_gate.py
+22 -1
View File
@@ -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: