From a2b88a79ec63756570926e8a608383244bb3ba4f Mon Sep 17 00:00:00 2001 From: LeonWang0735 Date: Fri, 6 Feb 2026 15:01:59 +0800 Subject: [PATCH] test(e2e_test): add automated cluster environment for conditional PUT race test (#1673) Co-authored-by: houseme Co-authored-by: loverustfs --- .../e2e_test/src/cluster_concurrency_test.rs | 241 ++++++++++++++++ crates/e2e_test/src/common.rs | 273 ++++++++++++++++++ crates/e2e_test/src/lib.rs | 3 + 3 files changed, 517 insertions(+) create mode 100644 crates/e2e_test/src/cluster_concurrency_test.rs diff --git a/crates/e2e_test/src/cluster_concurrency_test.rs b/crates/e2e_test/src/cluster_concurrency_test.rs new file mode 100644 index 000000000..7702dfac3 --- /dev/null +++ b/crates/e2e_test/src/cluster_concurrency_test.rs @@ -0,0 +1,241 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::common::RustFSTestClusterEnvironment; +use aws_sdk_s3::Client; +use aws_sdk_s3::error::SdkError; +use bytes::Bytes; +use serial_test::serial; +use std::sync::Arc; +use tokio::sync::Barrier; +use tracing::{info, warn}; + +const BUCKET: &str = "conditional-put-race-bucket"; + +async fn cleanup_object(client: &Client, key: &str) { + if let Err(e) = client.delete_object().bucket(BUCKET).key(key).send().await { + warn!("Failed to delete object '{}' from bucket '{}' during cleanup: {:?}", key, BUCKET, e); + } +} + +async fn conditional_put( + client: &Client, + key: &str, + data: &[u8], + client_id: usize, +) -> Result> { + let result = client + .put_object() + .bucket(BUCKET) + .key(key) + .body(Bytes::copy_from_slice(data).into()) + .if_none_match("*") + .send() + .await; + + match result { + Ok(resp) => { + info!(" Client {} SUCCEEDED - ETag: {}", client_id, resp.e_tag().unwrap_or("none")); + Ok(true) + } + Err(SdkError::ServiceError(e)) => { + let code = e.err().meta().code().unwrap_or(""); + let message = e.err().meta().message().unwrap_or(""); + if code == "PreconditionFailed" { + warn!(" Client {} got 412 PreconditionFailed", client_id); + Ok(false) + } else { + warn!(" Client {} got ServiceError: code={}, message={}", client_id, code, message); + Err(e.into_err().into()) + } + } + Err(e) => { + warn!(" Client {} got non-ServiceError: {:?}", client_id, e); + Err(e.into()) + } + } +} + +async fn run_race_iteration( + clients: &[Client], + test_key: &str, + iteration: usize, +) -> Result> { + cleanup_object(&clients[0], test_key).await; + tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; + + let head_result = clients[0].head_object().bucket(BUCKET).key(test_key).send().await; + + if head_result.is_ok() { + warn!("Warning: Object still exists after cleanup, skipping iteration {}", iteration); + return Ok(0); + } + + info!("\n=== Iteration {} ===", iteration); + info!("Launching {} concurrent conditional PUTs to different nodes...", clients.len()); + + let barrier = Arc::new(Barrier::new(clients.len())); + let test_key = test_key.to_string(); + + let mut handles = vec![]; + for (i, client) in clients.iter().enumerate() { + let client = client.clone(); + let barrier = barrier.clone(); + let key = test_key.clone(); + let data = format!("data from client {}", i).into_bytes(); + + let handle = tokio::spawn(async move { + barrier.wait().await; + conditional_put(&client, &key, &data, i).await + }); + handles.push(handle); + } + + let mut success_count = 0; + let mut had_error = false; + for handle in handles { + match handle.await { + Ok(Ok(true)) => success_count += 1, + Ok(Ok(false)) => {} + Ok(Err(e)) => { + had_error = true; + info!(" Error: {}", e); + } + Err(e) => { + had_error = true; + info!(" Task error: {}", e); + } + } + } + + info!("Result: {} out of {} succeeded", success_count, clients.len()); + + if success_count > 1 { + info!(">>> RACE CONDITION DETECTED!"); + } else if success_count == 1 { + info!(">>> Correct behavior: exactly 1 writer succeeded."); + } else if had_error { + return Err("all conditional PUTs failed (e.g. cluster/bucket not ready)".into()); + } else { + info!(">>> Unexpected: no writers succeeded."); + } + + Ok(success_count) +} + +#[tokio::test] +#[serial] +async fn test_conditional_put_race_cluster() -> Result<(), Box> { + crate::common::init_logging(); + info!("Starting conditional PUT race test with auto cluster"); + + let mut cluster = RustFSTestClusterEnvironment::new(4).await?; + cluster.start().await?; + + cluster.create_test_bucket(BUCKET).await?; + + let clients = cluster.create_all_clients()?; + + let iterations = 5; + let mut races_detected = 0; + let mut correct_count = 0; + let mut error_count = 0; + + for i in 1..=iterations { + let test_key = format!("race-test-{}-{}", std::process::id(), i); + + match run_race_iteration(&clients, &test_key, i).await { + Ok(success_count) => { + if success_count > 1 { + races_detected += 1; + } else if success_count == 1 { + correct_count += 1; + } + } + Err(e) => { + error_count += 1; + warn!("Iteration {} failed (not a race; e.g. cluster/network): {}", i, e) + } + } + + cleanup_object(&clients[0], &test_key).await; + tokio::time::sleep(tokio::time::Duration::from_millis(50)).await; + } + + info!("\n===================================="); + info!("SUMMARY"); + info!("===================================="); + info!("Total iterations: {}", iterations); + info!("Correct (1 winner): {}", correct_count); + info!("Race conditions: {}", races_detected); + info!("Errors (skipped): {}", error_count); + + assert_eq!(races_detected, 0, "Race conditions detected: {}/{}", races_detected, iterations); + assert_eq!( + error_count, 0, + "{} iteration(s) failed due to errors (e.g. cluster not ready)", + error_count + ); + + Ok(()) +} + +#[tokio::test] +#[serial] +async fn test_conditional_put_basic_cluster() -> Result<(), Box> { + crate::common::init_logging(); + info!("Starting basic conditional PUT test with auto cluster"); + + let mut cluster = RustFSTestClusterEnvironment::new(4).await?; + cluster.start().await?; + + cluster.create_test_bucket(BUCKET).await?; + + let client = cluster.create_s3_client(0)?; + let test_key = "basic-conditional-put"; + cleanup_object(&client, test_key).await; + + let result = client + .put_object() + .bucket(BUCKET) + .key(test_key) + .body(Bytes::from("first write").into()) + .if_none_match("*") + .send() + .await; + assert!(result.is_ok(), "First PUT with If-None-Match:* should succeed"); + + let result = client + .put_object() + .bucket(BUCKET) + .key(test_key) + .body(Bytes::from("second write").into()) + .if_none_match("*") + .send() + .await; + assert!(result.is_err(), "Second PUT with If-None-Match:* should fail"); + + assert!( + matches!(result, Err(SdkError::ServiceError(_))), + "Expected ServiceError but got different error type" + ); + + if let Err(SdkError::ServiceError(e)) = result { + let code = e.err().meta().code().unwrap_or(""); + assert_eq!(code, "PreconditionFailed"); + } + + cleanup_object(&client, test_key).await; + Ok(()) +} diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index da70db839..666e2a83f 100644 --- a/crates/e2e_test/src/common.rs +++ b/crates/e2e_test/src/common.rs @@ -374,3 +374,276 @@ pub async fn awscurl_delete( ) -> Result> { execute_awscurl(url, "DELETE", None, access_key, secret_key).await } + +/// Represents a single RustFS server instance in a test cluster. +/// +/// Each `ClusterNode` tracks the node's network address, base URL for +/// S3-compatible requests, on-disk data directory, and the underlying +/// child process handle when the node is running. +pub struct ClusterNode { + pub address: String, + pub url: String, + pub data_dir: String, + pub process: Option, +} + +/// Test environment for managing a multi-node RustFS cluster. +/// +/// `RustFSTestClusterEnvironment` is responsible for starting and stopping +/// a group of `ClusterNode`s, managing their temporary storage directory, +/// and providing the shared access and secret keys used by tests to +/// interact with the cluster. +pub struct RustFSTestClusterEnvironment { + pub nodes: Vec, + pub temp_dir: String, + pub access_key: String, + pub secret_key: String, +} + +impl RustFSTestClusterEnvironment { + /// Create a new RustFS test cluster environment with the specified number of nodes. + /// + /// Generates a unique temporary root directory for the cluster, allocates an available TCP port + /// for each node, creates an independent data directory for every node, and initializes basic + /// cluster node configurations (node processes are not started at this stage). + /// + /// # Arguments + /// + /// * `node_count` - The number of nodes to create in the cluster, must be a positive integer + /// (an empty cluster will cause errors in subsequent startup operations). + /// + /// # Returns + /// + /// * `Ok(Self)` - A new instance of `RustFSTestClusterEnvironment` with initialized node + /// configurations and temporary directory info on success. + /// * `Err(Box)` - An error if any step fails, such as temporary + /// directory creation failure or available port lookup failure. + pub async fn new(node_count: usize) -> Result> { + if node_count == 0 { + return Err("Node count must be greater than zero".into()); + } + let temp_dir = format!("/tmp/rustfs_cluster_test_{}", Uuid::new_v4()); + fs::create_dir_all(&temp_dir).await?; + + let mut nodes = Vec::with_capacity(node_count); + for i in 0..node_count { + let port = RustFSTestEnvironment::find_available_port().await?; + let address = format!("127.0.0.1:{}", port); + let url = format!("http://{}", address); + let data_dir = format!("{}/node{}", temp_dir, i); + fs::create_dir_all(&data_dir).await?; + + nodes.push(ClusterNode { + address, + url, + data_dir, + process: None, + }); + } + + Ok(Self { + nodes, + temp_dir, + access_key: DEFAULT_ACCESS_KEY.to_string(), + secret_key: DEFAULT_SECRET_KEY.to_string(), + }) + } + + /// Build the volumes argument string for RustFS binary (internal helper method). + /// + /// Concatenates the address and data directory of all cluster nodes into a single string + /// used as the `RUSTFS_VOLUMES` environment variable for RustFS node processes. + fn build_volumes_arg(&self) -> String { + self.nodes + .iter() + .map(|n| format!("http://{}{}", n.address, n.data_dir)) + .collect::>() + .join(" ") + } + + /// Start all node processes in the RustFS cluster and wait for the cluster service to be ready. + /// + /// Spawns a RustFS binary process for each node with necessary environment variable configurations, + /// first waits for each node's TCP port to be reachable, then verifies the cluster's S3-compatible + /// service availability via the S3 API. + /// + /// # Returns + /// + /// * `Ok(())` - All nodes start successfully and the cluster S3 service is ready for requests. + /// * `Err(Box)` - An error if process spawning fails, TCP port readiness + /// times out, or cluster service readiness times out. + pub async fn start(&mut self) -> Result<(), Box> { + let binary_path = rustfs_binary_path(); + let volumes_arg = self.build_volumes_arg(); + + for (i, node) in self.nodes.iter_mut().enumerate() { + info!("Starting cluster node {} on {}", i, node.address); + + let process = Command::new(&binary_path) + .env("RUSTFS_VOLUMES", &volumes_arg) + .env("RUSTFS_ADDRESS", &node.address) + .env("RUSTFS_ACCESS_KEY", &self.access_key) + .env("RUSTFS_SECRET_KEY", &self.secret_key) + .env("RUSTFS_CONSOLE_ENABLE", "false") + .current_dir(&node.data_dir) + .spawn()?; + + node.process = Some(process); + } + + for (i, node) in self.nodes.iter().enumerate() { + self.wait_for_node_ready(&node.address, i).await?; + } + + self.wait_for_service_ready().await?; + + Ok(()) + } + + /// Wait for a single cluster node's TCP port to become reachable (internal helper method). + /// + /// Attempts to establish a TCP connection to the node's address, retries up to 60 times + /// with a 1-second interval between attempts. Fails if the port is unreachable after all retries. + async fn wait_for_node_ready(&self, address: &str, idx: usize) -> Result<(), Box> { + for attempt in 0..60 { + if TcpStream::connect(address).await.is_ok() { + info!("Node {} ({}) TCP ready after {} attempts", idx, address, attempt + 1); + return Ok(()); + } + sleep(Duration::from_secs(1)).await; + } + Err(format!("Node {} failed to become ready", idx).into()) + } + + /// Wait for the entire cluster's S3-compatible service to be ready (internal helper method). + /// + /// Verifies service availability by calling the S3 `list_buckets` API, retries up to 120 times + /// with a 1-second interval between attempts. Fails if the API call remains unsuccessful after all retries. + async fn wait_for_service_ready(&self) -> Result<(), Box> { + let client = self.create_s3_client(0)?; + + for attempt in 0..120 { + match client.list_buckets().send().await { + Ok(_) => { + info!("Cluster service ready after {} attempts", attempt + 1); + return Ok(()); + } + Err(_) => { + sleep(Duration::from_secs(1)).await; + } + } + } + Err("Cluster service failed to become ready".into()) + } + + /// Create an S3 client configured to communicate with a specific cluster node. + /// + /// Configures the S3 client with the cluster's authentication credentials, a fixed `us-east-1` region, + /// the target node's endpoint URL, and enforces path-style access (required for RustFS S3 compatibility). + /// Performs a validity check on the node index before creating the client to avoid out-of-bounds errors. + /// + /// # Arguments + /// + /// * `node_idx` - The zero-based index of the target cluster node. Must be in the range `[0, total_nodes - 1]`. + /// + /// # Returns + /// + /// * `Ok(Client)` - A fully configured AWS S3 `Client` instance for the specified node on success. + /// * `Err(Box)` - An error if the node index is invalid, or if the S3 client configuration fails. + pub fn create_s3_client(&self, node_idx: usize) -> Result> { + if node_idx >= self.nodes.len() { + return Err("node_idx is invalid".into()); + } + let credentials = Credentials::new(&self.access_key, &self.secret_key, None, None, "cluster-test"); + let config = Config::builder() + .credentials_provider(credentials) + .region(Region::new("us-east-1")) + .endpoint_url(&self.nodes[node_idx].url) + .force_path_style(true) + .behavior_version_latest() + .build(); + Ok(Client::from_conf(config)) + } + + /// Create S3 clients for all nodes in the RustFS cluster and collect them into a vector. + /// + /// Iterates over all cluster node indices, calls `create_s3_client` for each index, and aggregates + /// the resulting clients into a pre-allocated vector. Terminates immediately and returns an error + /// if any single node's S3 client creation fails (fails fast behavior). + /// + /// # Returns + /// + /// * `Ok(Vec)` - A vector of configured S3 `Client` instances (one per cluster node) on full success. + /// * `Err(Box)` - An error with a descriptive message if any client creation fails, + /// including the underlying error from `create_s3_client`. + pub fn create_all_clients(&self) -> Result, Box> { + (0..self.nodes.len()).map(|i| self.create_s3_client(i)).try_fold( + Vec::with_capacity(self.nodes.len()), + |mut clients, result| match result { + Ok(client) => { + clients.push(client); + Ok(clients) + } + Err(e) => Err(format!("Failed to create S3 client for node: {}", e).into()), + }, + ) + } + + /// Create a test S3 bucket in the RustFS cluster. + /// + /// Uses the S3 client of the first cluster node to call the S3 `create_bucket` API and + /// create a bucket with the specified name (follows S3 bucket naming conventions). + /// + /// # Arguments + /// + /// * `bucket_name` - The name of the bucket to create, must comply with S3 bucket naming + /// rules (lowercase, no spaces, valid characters only). + /// + /// # Returns + /// + /// * `Ok(())` - The test bucket is created successfully via the S3 API. + /// * `Err(Box)` - An error if the S3 `create_bucket` API call fails, + /// such as invalid bucket name, insufficient permissions, or an unready cluster. + pub async fn create_test_bucket(&self, bucket_name: &str) -> Result<(), Box> { + let client = self.create_s3_client(0)?; + client.create_bucket().bucket(bucket_name).send().await?; + info!("Created test bucket: {}", bucket_name); + Ok(()) + } + + /// Stop all running node processes in the RustFS cluster. + /// + /// Iterates over all cluster nodes, attempts to kill the spawned RustFS process (if running), + /// and waits for the process to exit. Logs an error if process termination or waiting fails, + /// but does not panic (fails gracefully). + /// + /// This method is automatically called by the `Drop` trait when the cluster environment + /// is destroyed, and can also be called manually to stop the cluster early. + pub fn stop(&mut self) { + for (i, node) in self.nodes.iter_mut().enumerate() { + if let Some(mut process) = node.process.take() { + info!("Stopping cluster node {}", i); + if let Err(e) = process.kill() { + error!("Failed to kill cluster node {}: {}", i, e); + } + if let Err(e) = process.wait() { + error!("Failed to wait for cluster node {} to exit: {}", i, e); + } + } + } + } +} + +impl Drop for RustFSTestClusterEnvironment { + /// Clean up the RustFS test cluster environment when the instance is dropped. + /// + /// Automatically calls the `stop` method to terminate all running node processes, then + /// attempts to delete the cluster's temporary root directory and all its contents. + /// Logs a warning if directory deletion fails (does not affect program exit). + fn drop(&mut self) { + self.stop(); + if let Err(e) = std::fs::remove_dir_all(&self.temp_dir) { + warn!("Failed to clean up cluster temp directory {}: {}", self.temp_dir, e); + } + } +} diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index a467e68bf..c1baee9fb 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -61,6 +61,9 @@ mod protocols; #[cfg(test)] mod object_lock; +#[cfg(test)] +mod cluster_concurrency_test; + // PutObject / MultipartUpload with checksum (Content-MD5, x-amz-checksum-*) #[cfg(test)] mod checksum_upload_test;