test(e2e): cover mixed-version rolling upgrades (#6975)

2026-09-07 12:35:54 +00:00 · 2026-09-01 07:09:00 +08:00
parent 14a77f9d79
commit b0256e3453
4 changed files with 204 additions and 40 deletions
@@ -1469,31 +1469,18 @@ impl RustFSTestClusterEnvironment {
    ///   times out, or cluster service readiness times out.
    pub async fn start(&mut self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        let binary_path = rustfs_binary_path();
+        self.start_with_binary(&binary_path).await
+    }
+
+    /// Start every cluster node with a specific RustFS binary.
+    ///
+    /// Upgrade compatibility tests use this to initialize a cluster with a
+    /// pinned previous release before replacing nodes with the workspace build.
+    pub async fn start_with_binary(&mut self, binary_path: &Path) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        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 mut command = Command::new(&binary_path);
-            command
-                .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")
-                .env("RUST_LOG", "rustfs=info,rustfs_notify=debug");
-
-            for (key, value) in &self.extra_env {
-                command.env(key, value);
-            }
-            for (key, value) in &self.node_extra_env[i] {
-                command.env(key, value);
-            }
-            capture_command_logs(&mut command, self.node_capture_log_paths[i].as_deref())?;
-
-            let process = command.current_dir(&node.data_dir).spawn()?;
-
-            node.process = Some(process);
+        for node_idx in 0..self.nodes.len() {
+            self.spawn_node(node_idx, binary_path, &volumes_arg)?;
        }

        for (i, node) in self.nodes.iter().enumerate() {
@@ -1509,20 +1496,46 @@ impl RustFSTestClusterEnvironment {

    /// Start one node process using the cluster's existing volume layout.
    pub async fn start_node(&mut self, node_idx: usize) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        let binary_path = rustfs_binary_path();
+        self.start_node_from_binary(node_idx, &binary_path).await
+    }
+
+    /// Start one stopped cluster node with a specific RustFS binary while
+    /// preserving the cluster's volume layout and that node's data directory.
+    pub async fn start_node_from_binary(
+        &mut self,
+        node_idx: usize,
+        binary_path: &Path,
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        let volumes_arg = self.build_volumes_arg();
+        self.spawn_node(node_idx, binary_path, &volumes_arg)?;
+
+        self.wait_for_node_ready(&self.nodes[node_idx].address, node_idx).await?;
+        self.wait_for_node_service_ready(node_idx).await?;
+        Ok(())
+    }
+
+    fn spawn_node(
+        &mut self,
+        node_idx: usize,
+        binary_path: &Path,
+        volumes_arg: &str,
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        self.ensure_node_index(node_idx)?;
        if self.nodes[node_idx].process.is_some() {
            return Err(format!("cluster node {node_idx} is already running").into());
        }
+        if !binary_path.is_file() {
+            return Err(format!("RustFS binary does not exist: {}", binary_path.display()).into());
+        }

-        let binary_path = rustfs_binary_path();
-        let volumes_arg = self.build_volumes_arg();
        let log_path = self.node_capture_log_paths[node_idx].clone();
        let node = &mut self.nodes[node_idx];
-        info!("Starting cluster node {} on {}", node_idx, node.address);
+        info!("Starting cluster node {} on {} with {}", node_idx, node.address, binary_path.display());

-        let mut command = Command::new(&binary_path);
+        let mut command = Command::new(binary_path);
        command
-            .env("RUSTFS_VOLUMES", &volumes_arg)
+            .env("RUSTFS_VOLUMES", volumes_arg)
            .env("RUSTFS_ADDRESS", &node.address)
            .env("RUSTFS_ACCESS_KEY", &self.access_key)
            .env("RUSTFS_SECRET_KEY", &self.secret_key)
@@ -1539,9 +1552,6 @@ impl RustFSTestClusterEnvironment {

        let process = command.current_dir(&node.data_dir).spawn()?;
        node.process = Some(process);
-
-        self.wait_for_node_ready(&self.nodes[node_idx].address, node_idx).await?;
-        self.wait_for_node_service_ready(node_idx).await?;
        Ok(())
    }

@@ -12,14 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use crate::common::{RustFSTestEnvironment, init_logging};
+use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging, rustfs_binary_path};
 use aws_sdk_s3::Client;
 use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::primitives::ByteStream;
 use aws_sdk_s3::types::{
    BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, ServerSideEncryption, VersioningConfiguration,
 };
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
+use tokio::task::JoinSet;

 type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;

@@ -28,6 +29,10 @@ const SSE_MASTER_KEY_ENV: &str = "RUSTFS_SSE_S3_MASTER_KEY";
 const SSE_MASTER_KEY: &str = "QkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkI=";
 const PLAIN_BUCKET: &str = "upgrade-plain-data";
 const VERSIONED_BUCKET: &str = "upgrade-versioned-data";
+const MIXED_BUCKET: &str = "upgrade-mixed-version-data";
+const MIXED_NODE_COUNT: usize = 4;
+const MULTIPART_WORKERS: usize = 16;
+const MULTIPART_UPLOADS_PER_WORKER: usize = 16;

 fn source_binary() -> Result<PathBuf, Box<dyn std::error::Error + Send + Sync>> {
    let path = std::env::var_os(SOURCE_BINARY_ENV)
@@ -103,6 +108,99 @@ async fn write_multipart(client: &Client, bucket: &str, key: &str, parts: &[Vec<
    Ok(())
 }

+fn configure_cluster_logs(cluster: &mut RustFSTestClusterEnvironment) -> TestResult {
+    let Some(log_dir) = std::env::var_os("RUSTFS_E2E_LOG_DIR") else {
+        return Ok(());
+    };
+    std::fs::create_dir_all(&log_dir)?;
+    for node_idx in 0..cluster.nodes.len() {
+        let path = Path::new(&log_dir).join(format!("mixed-upgrade-node-{node_idx}.log"));
+        cluster.set_node_capture_log_path(node_idx, path.to_string_lossy().into_owned())?;
+    }
+    Ok(())
+}
+
+async fn write_multipart_load(clients: &[Client], phase: &str) -> Result<Vec<String>, Box<dyn std::error::Error + Send + Sync>> {
+    let mut tasks = JoinSet::new();
+    for worker in 0..MULTIPART_WORKERS {
+        let client = clients[worker % clients.len()].clone();
+        let phase = phase.to_string();
+        tasks.spawn(async move {
+            let mut keys = Vec::with_capacity(MULTIPART_UPLOADS_PER_WORKER);
+            for upload in 0..MULTIPART_UPLOADS_PER_WORKER {
+                let key = format!("{phase}/multipart/{worker:02}/{upload:02}");
+                let part = vec![u8::try_from(worker)?; 64 * 1024];
+                write_multipart(&client, MIXED_BUCKET, &key, &[part]).await?;
+                keys.push(key);
+            }
+            Ok::<_, Box<dyn std::error::Error + Send + Sync>>(keys)
+        });
+    }
+
+    let mut keys = Vec::with_capacity(MULTIPART_WORKERS * MULTIPART_UPLOADS_PER_WORKER);
+    while let Some(result) = tasks.join_next().await {
+        keys.extend(result??);
+    }
+    Ok(keys)
+}
+
+async fn exercise_mixed_cluster(
+    cluster: &RustFSTestClusterEnvironment,
+    phase: &str,
+    current_node: usize,
+    previous_node: usize,
+) -> TestResult {
+    let clients = cluster.create_all_clients()?;
+    let current_client = &clients[current_node];
+    let previous_client = &clients[previous_node];
+
+    let current_key = format!("{phase}/written-by-current");
+    let current_body = format!("{phase}: current RustFS build").into_bytes();
+    current_client
+        .put_object()
+        .bucket(MIXED_BUCKET)
+        .key(&current_key)
+        .body(ByteStream::from(current_body.clone()))
+        .send()
+        .await?;
+    assert_eq!(read_object(previous_client, MIXED_BUCKET, &current_key, None).await?.1, current_body);
+
+    let previous_key = format!("{phase}/written-by-previous");
+    let previous_body = format!("{phase}: previous RustFS release").into_bytes();
+    previous_client
+        .put_object()
+        .bucket(MIXED_BUCKET)
+        .key(&previous_key)
+        .body(ByteStream::from(previous_body.clone()))
+        .send()
+        .await?;
+    assert_eq!(read_object(current_client, MIXED_BUCKET, &previous_key, None).await?.1, previous_body);
+
+    let multipart_keys = write_multipart_load(&clients, phase).await?;
+    let expected_count = multipart_keys.len() + 2;
+    for client in [current_client, previous_client] {
+        let listed = client
+            .list_objects_v2()
+            .bucket(MIXED_BUCKET)
+            .prefix(format!("{phase}/"))
+            .send()
+            .await?;
+        assert_eq!(
+            listed.contents().len(),
+            expected_count,
+            "both RustFS versions must stream the complete mixed-version listing"
+        );
+    }
+
+    let last_multipart_key = format!("{phase}/multipart/{:02}/{:02}", MULTIPART_WORKERS - 1, MULTIPART_UPLOADS_PER_WORKER - 1);
+    assert_eq!(
+        read_object(previous_client, MIXED_BUCKET, &last_multipart_key, None).await?.1,
+        vec![u8::try_from(MULTIPART_WORKERS - 1)?; 64 * 1024]
+    );
+
+    Ok(())
+}
+
 #[tokio::test]
 #[ignore = "requires a pinned previous RustFS release binary"]
 async fn direct_upgrade_from_rc2_preserves_object_contracts() -> TestResult {
@@ -252,3 +350,47 @@ async fn direct_upgrade_from_rc2_preserves_object_contracts() -> TestResult {

    Ok(())
 }
+
+#[tokio::test]
+#[ignore = "requires a pinned previous RustFS release binary"]
+async fn rolling_upgrade_from_rc2_preserves_mixed_version_contracts() -> TestResult {
+    init_logging();
+    let previous_binary = source_binary()?;
+    let current_binary = rustfs_binary_path();
+    let mut cluster = RustFSTestClusterEnvironment::new(MIXED_NODE_COUNT).await?;
+    cluster.set_env("RUST_LOG", "rustfs=warn,rustfs_notify=warn");
+    configure_cluster_logs(&mut cluster)?;
+    cluster.start_with_binary(&previous_binary).await?;
+    cluster.create_test_bucket(MIXED_BUCKET).await?;
+
+    cluster.stop_node(0)?;
+    cluster.start_node_from_binary(0, &current_binary).await?;
+    exercise_mixed_cluster(&cluster, "one-current-node", 0, 1).await?;
+
+    for node_idx in [1, 2] {
+        cluster.stop_node(node_idx)?;
+        cluster.start_node_from_binary(node_idx, &current_binary).await?;
+    }
+    exercise_mixed_cluster(&cluster, "one-previous-node", 0, 3).await?;
+
+    cluster.stop_node(3)?;
+    cluster.start_node_from_binary(3, &current_binary).await?;
+
+    for client in cluster.create_all_clients()? {
+        for phase in ["one-current-node", "one-previous-node"] {
+            let listed = client
+                .list_objects_v2()
+                .bucket(MIXED_BUCKET)
+                .prefix(format!("{phase}/"))
+                .send()
+                .await?;
+            assert_eq!(
+                listed.contents().len(),
+                MULTIPART_WORKERS * MULTIPART_UPLOADS_PER_WORKER + 2,
+                "the homogeneous current cluster must preserve every object from {phase}"
+            );
+        }
+    }
+
+    Ok(())
+}