From 7ab0955f8b37d79323a6a835d27be5c4c776877c Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 25 Jul 2026 22:21:52 +0800 Subject: [PATCH] test(e2e): stabilize tier and storage class checks (#5228) Co-authored-by: heihutu --- .../src/inline_fast_path_cluster_test.rs | 60 ++++++++++-------- crates/e2e_test/src/multipart_auth_test.rs | 6 +- crates/ecstore/src/services/tier/tier.rs | 63 ++++++++++++------- .../src/services/tier/tier_mutation_intent.rs | 27 +++++++- .../src/services/tier/tier_mutation_peer.rs | 11 +--- 5 files changed, 103 insertions(+), 64 deletions(-) diff --git a/crates/e2e_test/src/inline_fast_path_cluster_test.rs b/crates/e2e_test/src/inline_fast_path_cluster_test.rs index 72ed10517..93874d77e 100644 --- a/crates/e2e_test/src/inline_fast_path_cluster_test.rs +++ b/crates/e2e_test/src/inline_fast_path_cluster_test.rs @@ -56,6 +56,7 @@ use tokio::net::TcpListener; use tokio::sync::Mutex; use tokio::task::JoinHandle; use tokio::time::{Instant, sleep}; +use uuid::Uuid; use walkdir::WalkDir; type TestResult = Result>; @@ -81,7 +82,6 @@ const FALLBACK_REQUEST_DIRECTION: &str = "request"; const FALLBACK_RESPONSE_DIRECTION: &str = "response"; const MPU_PART_1_SIZE: usize = 5 * 1024 * 1024; const MPU_PART_2_SIZE: usize = 16 * KIB; -const TIER_NAME: &str = "COLDTIER"; const TIER_BUCKET: &str = "inline-fallback-cold-tier"; const TIER_PREFIX: &str = "tiered"; const MSGPACK_FALLBACK_CONTROL_SERIES: [(&str, &str); 4] = [ @@ -961,11 +961,15 @@ async fn signed_admin_request( Ok((status, text)) } -async fn add_rustfs_tier(hot: &RustFSTestClusterEnvironment, cold: &RustFSTestEnvironment) -> TestResult { +fn unique_tier_name() -> String { + format!("COLDTIER{}", Uuid::new_v4().simple()).to_ascii_uppercase() +} + +async fn add_rustfs_tier(hot: &RustFSTestClusterEnvironment, cold: &RustFSTestEnvironment, tier_name: &str) -> TestResult { let body = serde_json::json!({ "type": "rustfs", "rustfs": { - "name": TIER_NAME, + "name": tier_name, "endpoint": cold.url.as_str(), "accessKey": cold.access_key.as_str(), "secretKey": cold.secret_key.as_str(), @@ -990,7 +994,7 @@ async fn add_rustfs_tier(hot: &RustFSTestClusterEnvironment, cold: &RustFSTestEn .await?; let attempt = format!("status={status}, body={}", compact_body(&response)); if status.is_success() { - wait_for_tier_verifiable(hot, &format!("status={status}, body={response}")).await?; + wait_for_tier_verifiable(hot, tier_name, &format!("status={status}, body={response}")).await?; return Ok(()); } attempts.push(attempt); @@ -1005,10 +1009,10 @@ async fn add_rustfs_tier(hot: &RustFSTestClusterEnvironment, cold: &RustFSTestEn Err(format!("AddTier(RustFS) failed after readiness polling: {final_error}").into()) } -async fn wait_for_tier_verifiable(hot: &RustFSTestClusterEnvironment, add_tier_response: &str) -> TestResult { +async fn wait_for_tier_verifiable(hot: &RustFSTestClusterEnvironment, tier_name: &str, add_tier_response: &str) -> TestResult { let deadline = Instant::now() + Duration::from_secs(60); let final_error = loop { - let snapshot = tier_readiness_snapshot(hot).await?; + let snapshot = tier_readiness_snapshot(hot, tier_name).await?; if snapshot.iter().any(|node| node.verify_status.is_success()) { return Ok(()); } @@ -1021,7 +1025,7 @@ async fn wait_for_tier_verifiable(hot: &RustFSTestClusterEnvironment, add_tier_r sleep(Duration::from_millis(500)).await; }; Err(format!( - "tier {TIER_NAME} was not verifiable on any hot node within 60s after AddTier({add_tier_response}): {final_error}" + "tier {tier_name} was not verifiable on any hot node within 60s after AddTier({add_tier_response}): {final_error}" ) .into()) } @@ -1036,7 +1040,7 @@ struct TierNodeReadiness { verify_body: String, } -async fn tier_readiness_snapshot(hot: &RustFSTestClusterEnvironment) -> TestResult> { +async fn tier_readiness_snapshot(hot: &RustFSTestClusterEnvironment, tier_name: &str) -> TestResult> { let mut snapshot = Vec::with_capacity(hot.nodes.len()); for (node_index, node) in hot.nodes.iter().enumerate() { let (list_status, list_body) = @@ -1044,7 +1048,7 @@ async fn tier_readiness_snapshot(hot: &RustFSTestClusterEnvironment) -> TestResu let (verify_status, verify_body) = signed_admin_request( &node.url, Method::GET, - &format!("/rustfs/admin/v3/tier/{TIER_NAME}"), + &format!("/rustfs/admin/v3/tier/{tier_name}"), None, &hot.access_key, &hot.secret_key, @@ -1054,7 +1058,7 @@ async fn tier_readiness_snapshot(hot: &RustFSTestClusterEnvironment) -> TestResu node_index, node_url: node.url.clone(), list_status, - list_has_tier: tier_list_contains(&list_body), + list_has_tier: tier_list_contains(&list_body, tier_name), list_body, verify_status, verify_body, @@ -1063,18 +1067,18 @@ async fn tier_readiness_snapshot(hot: &RustFSTestClusterEnvironment) -> TestResu Ok(snapshot) } -fn tier_list_contains(response: &str) -> bool { +fn tier_list_contains(response: &str, tier_name: &str) -> bool { serde_json::from_str::(response) .ok() .and_then(|value| value.as_array().cloned()) .is_some_and(|tiers| { tiers.iter().any(|tier| { - tier.get("name").and_then(serde_json::Value::as_str) == Some(TIER_NAME) + tier.get("name").and_then(serde_json::Value::as_str) == Some(tier_name) || tier .get("rustfs") .and_then(|rustfs| rustfs.get("name")) .and_then(serde_json::Value::as_str) - == Some(TIER_NAME) + == Some(tier_name) }) }) } @@ -1121,30 +1125,30 @@ fn is_retryable_add_tier_error(response: &str) -> bool { response.contains("Remote tier configuration is already being replaced") } -fn transition_rule() -> TestResult { +fn transition_rule(tier_name: &str) -> TestResult { Ok(LifecycleRule::builder() .id("inline-fallback-transition") .filter(LifecycleRuleFilter::builder().prefix("transition/").build()) .transitions( Transition::builder() .days(0) - .storage_class(TransitionStorageClass::from(TIER_NAME)) + .storage_class(TransitionStorageClass::from(tier_name)) .build(), ) .status(ExpirationStatus::Enabled) .build()?) } -async fn wait_for_transition(client: &Client, bucket: &str, key: &str) -> TestResult { +async fn wait_for_transition(client: &Client, bucket: &str, key: &str, tier_name: &str) -> TestResult { let deadline = Instant::now() + Duration::from_secs(90); loop { let head = client.head_object().bucket(bucket).key(key).send().await?; - if head.storage_class().map(|storage_class| storage_class.as_str()) == Some(TIER_NAME) { + if head.storage_class().map(|storage_class| storage_class.as_str()) == Some(tier_name) { return Ok(()); } if Instant::now() >= deadline { return Err(format!( - "object {bucket}/{key} was not transitioned to {TIER_NAME} within 90s (storage_class={:?})", + "object {bucket}/{key} was not transitioned to {tier_name} within 90s (storage_class={:?})", head.storage_class() ) .into()); @@ -1163,10 +1167,12 @@ async fn cold_tier_object_count(cold_client: &Client) -> TestResult { .len()) } -async fn put_lifecycle_with_transition_retry(client: &Client, bucket: &str) -> TestResult { +async fn put_lifecycle_with_transition_retry(client: &Client, bucket: &str, tier_name: &str) -> TestResult { let deadline = Instant::now() + Duration::from_secs(30); loop { - let lifecycle = BucketLifecycleConfiguration::builder().rules(transition_rule()?).build()?; + let lifecycle = BucketLifecycleConfiguration::builder() + .rules(transition_rule(tier_name)?) + .build()?; match client .put_bucket_lifecycle_configuration() .bucket(bucket) @@ -1541,14 +1547,15 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_ let fallback_before = collector.msgpack_json_fallback_totals().await; - add_rustfs_tier(&hot, &cold).await?; + let tier_name = unique_tier_name(); + add_rustfs_tier(&hot, &cold, &tier_name).await?; let bucket = "inline-transitioned-mixed-msgpack-controls"; hot_client.create_bucket().bucket(bucket).send().await?; - put_lifecycle_with_transition_retry(&hot_client, bucket).await?; + put_lifecycle_with_transition_retry(&hot_client, bucket, &tier_name).await?; let key = "transition/mixed-multipart.bin"; let (body, second_part, etag) = put_two_part_multipart(&hot_client, bucket, key).await?; - wait_for_transition(&hot_client, bucket, key).await?; + wait_for_transition(&hot_client, bucket, key, &tier_name).await?; assert!( cold_tier_object_count(&cold_client).await? >= 1, "cold-tier bucket must hold transitioned objects" @@ -1638,14 +1645,15 @@ async fn four_node_transitioned_inline_fallback() -> TestResult { hot.start().await?; let hot_client = hot.create_s3_client(0)?; - add_rustfs_tier(&hot, &cold).await?; + let tier_name = unique_tier_name(); + add_rustfs_tier(&hot, &cold, &tier_name).await?; let bucket = "inline-transitioned-fallback"; hot_client.create_bucket().bucket(bucket).send().await?; - put_lifecycle_with_transition_retry(&hot_client, bucket).await?; + put_lifecycle_with_transition_retry(&hot_client, bucket, &tier_name).await?; let key = "transition/two-part.bin"; let (body, _, etag) = put_two_part_multipart(&hot_client, bucket, key).await?; - wait_for_transition(&hot_client, bucket, key).await?; + wait_for_transition(&hot_client, bucket, key, &tier_name).await?; assert!( cold_tier_object_count(&cold_client).await? >= 1, "cold-tier bucket must hold the transitioned object" diff --git a/crates/e2e_test/src/multipart_auth_test.rs b/crates/e2e_test/src/multipart_auth_test.rs index feabb0ead..e29d1c635 100644 --- a/crates/e2e_test/src/multipart_auth_test.rs +++ b/crates/e2e_test/src/multipart_auth_test.rs @@ -1267,7 +1267,7 @@ async fn test_anonymous_post_object_accepts_storage_class_exact_policy_match() let bucket = "anon-post-storage-class"; let object_key = "post-storage-class-object.txt"; let expected_body = b"post-storage-class-body".to_vec(); - let storage_class = "STANDARD_IA"; + let storage_class = "REDUCED_REDUNDANCY"; let admin_client = env.create_s3_client(); admin_client.create_bucket().bucket(bucket).send().await?; @@ -5138,7 +5138,7 @@ async fn test_signed_put_object_extract_preserves_storage_class() -> Result<(), .put_object() .bucket(bucket) .key(archive_key) - .storage_class(aws_sdk_s3::types::StorageClass::StandardIa) + .storage_class(aws_sdk_s3::types::StorageClass::ReducedRedundancy) .body(ByteStream::from(tar_bytes)) .customize() .mutate_request(move |req| { @@ -5155,7 +5155,7 @@ async fn test_signed_put_object_extract_preserves_storage_class() -> Result<(), .send() .await?; - assert_eq!(head.storage_class().map(|value| value.as_str()), Some("STANDARD_IA")); + assert_eq!(head.storage_class().map(|value| value.as_str()), Some("REDUCED_REDUNDANCY")); Ok(()) } diff --git a/crates/ecstore/src/services/tier/tier.rs b/crates/ecstore/src/services/tier/tier.rs index 314b1d84a..15651efac 100644 --- a/crates/ecstore/src/services/tier/tier.rs +++ b/crates/ecstore/src/services/tier/tier.rs @@ -20,7 +20,7 @@ use byteorder::{ByteOrder, LittleEndian}; use bytes::Bytes; -use futures::{FutureExt, future::join_all}; +use futures::FutureExt; use http::HeaderMap; use http::status::StatusCode; use lazy_static::lazy_static; @@ -956,14 +956,10 @@ struct TierMutationPrepareFailure { } async fn abort_tier_mutation_peers(mutation_id: uuid::Uuid, peers: Vec>) -> io::Result<()> { - let results = join_all(peers.into_iter().map(|peer| async move { + let mut failures = Vec::new(); + for peer in peers { let label = peer.peer_label(); let result = peer.abort_tier_mutation(mutation_id).await; - (label, result) - })) - .await; - let mut failures = Vec::new(); - for (label, result) in results { match result { Ok(PeerTierMutationState::Aborted) => {} Ok(state) => { @@ -985,16 +981,9 @@ async fn commit_tier_mutation_peers( committed_config_etag: &str, ) -> io::Result<()> { let payload = Bytes::copy_from_slice(committed_config_etag.as_bytes()); - let results = join_all(peers.into_iter().map(|peer| { - let payload = payload.clone(); - async move { - let label = peer.peer_label(); - let result = peer.commit_tier_mutation(mutation_id, payload).await; - (label, result) - } - })) - .await; - for (label, result) in results { + for peer in peers { + let label = peer.peer_label(); + let result = peer.commit_tier_mutation(mutation_id, payload.clone()).await; match result { Ok(PeerTierMutationState::Committed) => {} Ok(state) => { @@ -5954,6 +5943,7 @@ mod tests { } async fn abort_tier_mutation(&self, _mutation_id: uuid::Uuid) -> Result { + self.track("abort").await; Ok(PeerTierMutationState::Aborted) } } @@ -5991,7 +5981,7 @@ mod tests { } #[tokio::test] - async fn commit_tier_mutation_peers_keeps_peer_commits_concurrent() { + async fn commit_tier_mutation_peers_serializes_shared_record_writes() { let mutation_id = uuid::Uuid::from_u128(35); let calls = Arc::new(Mutex::new(Vec::new())); let active = Arc::new(AtomicUsize::new(0)); @@ -6009,13 +5999,38 @@ mod tests { .await .expect("successful commit fanout should commit every peer"); - assert!( - max_active.load(Ordering::SeqCst) > 1, - "peer commit fanout should remain concurrent after serializing prepare" + assert_eq!( + max_active.load(Ordering::SeqCst), + 1, + "peer commit fanout must not write the same intent concurrently" ); - let mut calls = lock_unpoisoned(&calls).clone(); - calls.sort(); - assert_eq!(calls.as_slice(), &["peer-a:commit", "peer-b:commit", "peer-c:commit"]); + assert_eq!(lock_unpoisoned(&calls).as_slice(), &["peer-a:commit", "peer-b:commit", "peer-c:commit"]); + } + + #[tokio::test] + async fn abort_tier_mutation_peers_serializes_shared_record_writes() { + let mutation_id = uuid::Uuid::from_u128(36); + let calls = Arc::new(Mutex::new(Vec::new())); + let active = Arc::new(AtomicUsize::new(0)); + let max_active = Arc::new(AtomicUsize::new(0)); + + abort_tier_mutation_peers( + mutation_id, + vec![ + ConcurrencyTrackingTierMutationPeer::boxed("peer-a", calls.clone(), active.clone(), max_active.clone()), + ConcurrencyTrackingTierMutationPeer::boxed("peer-b", calls.clone(), active.clone(), max_active.clone()), + ConcurrencyTrackingTierMutationPeer::boxed("peer-c", calls.clone(), active.clone(), max_active.clone()), + ], + ) + .await + .expect("successful abort fanout should abort every peer"); + + assert_eq!( + max_active.load(Ordering::SeqCst), + 1, + "peer abort fanout must not write the same intent concurrently" + ); + assert_eq!(lock_unpoisoned(&calls).as_slice(), &["peer-a:abort", "peer-b:abort", "peer-c:abort"]); } #[tokio::test] diff --git a/crates/ecstore/src/services/tier/tier_mutation_intent.rs b/crates/ecstore/src/services/tier/tier_mutation_intent.rs index fa473cc8b..28a303e32 100644 --- a/crates/ecstore/src/services/tier/tier_mutation_intent.rs +++ b/crates/ecstore/src/services/tier/tier_mutation_intent.rs @@ -256,6 +256,15 @@ impl TierMutationIntent { } } + pub(crate) fn same_identity_as(&self, other: &Self) -> bool { + self.mutation_id == other.mutation_id + && self.kind == other.kind + && self.old_config_etag == other.old_config_etag + && self.candidate_digest == other.candidate_digest + && self.affected_targets == other.affected_targets + && self.expires_at_unix_nanos == other.expires_at_unix_nanos + } + pub(crate) fn encode(&self) -> Result> { self.validate()?; let intent_bytes = serde_json::to_vec(self)?; @@ -567,7 +576,23 @@ where } match save_tier_mutation_intent_record_if_current_with_prefix(api.clone(), prefix, &intent, ¤t_etag).await { Ok(()) => return Ok((intent, true)), - Err(Error::PreconditionFailed) if attempt + 1 < TIER_MUTATION_INTENT_ADVANCE_CAS_ATTEMPTS => continue, + Err(Error::PreconditionFailed) => { + let (mut current, _) = + load_tier_mutation_intent_record_with_etag_at_prefix(api.clone(), prefix, mutation_id).await?; + if !current.same_identity_as(&intent) { + return Err(Error::PreconditionFailed); + } + let replayed = current + .advance_idempotent(next, intent.committed_config_etag.clone()) + .map_err(tier_mutation_intent_store_error)?; + if replayed { + if attempt + 1 < TIER_MUTATION_INTENT_ADVANCE_CAS_ATTEMPTS { + continue; + } + return Err(Error::PreconditionFailed); + } + return Ok((current, false)); + } Err(err) => return Err(err), } } diff --git a/crates/ecstore/src/services/tier/tier_mutation_peer.rs b/crates/ecstore/src/services/tier/tier_mutation_peer.rs index fd408d304..2a249ad7c 100644 --- a/crates/ecstore/src/services/tier/tier_mutation_peer.rs +++ b/crates/ecstore/src/services/tier/tier_mutation_peer.rs @@ -111,7 +111,7 @@ async fn handle_prepare( } Err(Error::PreconditionFailed) => { let existing = load_tier_mutation_intent_record(api, mutation_id).await?; - if !same_mutation_identity(&existing, &intent) { + if !existing.same_identity_as(&intent) { return Err(TierMutationPeerError::ConflictingIntent); } match existing.state { @@ -236,15 +236,6 @@ fn peer_state_from_intent(state: TierMutationIntentState) -> TierMutationPeerSta } } -fn same_mutation_identity(existing: &TierMutationIntent, expected: &TierMutationIntent) -> bool { - existing.mutation_id == expected.mutation_id - && existing.kind == expected.kind - && existing.old_config_etag == expected.old_config_etag - && existing.candidate_digest == expected.candidate_digest - && existing.affected_targets == expected.affected_targets - && existing.expires_at_unix_nanos == expected.expires_at_unix_nanos -} - #[cfg(test)] mod tests { use super::*;