test(e2e): add 4-node 4-disk distributed Actions suite

Add a nightly e2e-distributed lane that boots localhost 4-node clusters
and covers S3, object lock, versioning, replication, quota, expand,
decommission, rebalance, site replication, concurrency, and chaos.

Co-authored-by: RustFS <hello@rustfs.com>

crates/e2e_test/README.md

@@ -26,6 +26,7 @@ Registered in [`src/lib.rs`](src/lib.rs). Grouped by concern:
 | **protocols** | [`src/protocols/`](src/protocols) | FTPS, WebDAV, SFTP compliance. Fixed ports, own guide: [`src/protocols/README.md`](src/protocols/README.md) |
 | **reliant** | [`src/reliant/`](src/reliant) | Tests that reuse an **externally started** server (SQL/select, conditional writes, lifecycle, deleted-object reads, node-interact). Run via [`scripts/run_e2e_tests.sh`](../../scripts/run_e2e_tests.sh); see [`src/reliant/README.md`](src/reliant/README.md) |
 | **cluster** | `cluster_concurrency_test`, `stale_multipart_cleanup_cluster_test`, `namespace_lock_quorum_test`, `admin_timeout_regression_test`, `object_lambda_test`, `replication_extension_test`, `tier_stats_cluster_test` | Multi-node scenarios via `RustFSTestClusterEnvironment` |
+| **distributed 4×4** | [`src/distributed/`](src/distributed) | Nightly `e2e-distributed` lane: S3, object lock/WORM, versioning, bucket/site replication, quota, expand/decommission/rebalance, concurrency, chaos. Map: [`docs/testing/distributed-e2e.md`](../../docs/testing/distributed-e2e.md) |
 | **chaos / reliability** | [`src/chaos.rs`](src/chaos.rs), `reliability_disk_fault_test`, `heal_erasure_disk_rebuild_test`, `server_startup_failfast_test` | Disk offline/replace/corrupt, EC rebuild, heal, fail-fast startup |
 | **upgrade compatibility** | `upgrade_compatibility_test` | Pinned previous-release writes followed by current-build reads on the same data directory |
 
@@ -171,6 +172,7 @@ the same profile for membership and execution with one nightly worker.
 | KMS suite | `e2e-full` job, merge queue + main | **Active** |
 | Direct and mixed-version rolling upgrades from pinned previous release | `e2e-upgrade.yml`, storage-sensitive PRs + release tags + weekly | **Active** |
 | Cluster faults (`e2e-nightly` profile) | consolidated nightly workflow | **Active** (backlog#1149 ci-7) |
+| Distributed 4-node 4-disk (`e2e-distributed` profile) | `.github/workflows/e2e-distributed.yml` | **Active** (nightly / dispatch; not a merge gate) |
 | Protocols (FTPS/WebDAV/SFTP) | consolidated nightly workflow, serial | **Active** (backlog#1149 ci-7) |
 | Replication (fast subset) | `e2e-smoke` profile, `e2e-tests` job, every PR | **Active** (backlog#1147 repl-1) |
 | Replication (slow + multi-node) | `e2e-repl-nightly` profile, consolidated nightly workflow | **Active** (backlog#1147 repl-1) |
@@ -191,6 +193,8 @@ cargo nextest run --profile e2e-smoke -p e2e_test
 cargo nextest run --profile e2e-full -p e2e_test
 # Cluster fault nightly lane
 cargo nextest run --profile e2e-nightly -p e2e_test
+# 4-node 4-disk distributed lane (S3 / lock / versioning / replication / decommission / chaos)
+cargo nextest run --profile e2e-distributed -p e2e_test
 # Replication nightly lane; awscurl is required for STS paths
 cargo nextest run --profile e2e-repl-nightly -p e2e_test
 # Fixed-port protocol nightly lane

crates/e2e_test/src/common.rs

@@ -1700,6 +1700,69 @@ impl RustFSTestClusterEnvironment {
         Ok(())
     }
 
+    /// Append a new single-node erasure pool to a stopped multi-pool cluster.
+    ///
+    /// Used to simulate pool expansion on localhost: every pool already owns
+    /// exactly one node with `drives_per_node >= 2` (the only multi-pool layout
+    /// the single-host `RUSTFS_VOLUMES` syntax can express). The new node is
+    /// allocated a fresh port and empty drive directories; callers must
+    /// [`Self::start`] afterwards so every process picks up the extended
+    /// volumes argument. Existing data directories are left untouched.
+    pub async fn append_single_node_pool(&mut self) -> Result<usize, Box<dyn std::error::Error + Send + Sync>> {
+        if self.nodes.iter().any(|node| node.process.is_some()) {
+            return Err("stop the cluster before appending a pool".into());
+        }
+        if self.topology.drives_per_node < 2 {
+            return Err(
+                "append_single_node_pool requires drives_per_node >= 2 (the server parser rejects a single-drive ellipses pool)"
+                    .into(),
+            );
+        }
+
+        let mut pools = self.topology.normalized_pools();
+        for (pool_idx, nodes) in pools.iter().enumerate() {
+            if nodes.len() != 1 {
+                return Err(format!(
+                    "pool {pool_idx} spans {} nodes; append_single_node_pool requires one node per pool",
+                    nodes.len()
+                )
+                .into());
+            }
+        }
+
+        let new_idx = self.nodes.len();
+        let port = RustFSTestEnvironment::find_available_port().await?;
+        let address = format!("127.0.0.1:{port}");
+        let data_dirs: Vec<String> = (0..self.topology.drives_per_node)
+            .map(|drive| format!("{}/node{}/drive{}", self.temp_dir, new_idx, drive))
+            .collect();
+        for dir in &data_dirs {
+            fs::create_dir_all(dir).await?;
+        }
+
+        self.nodes.push(ClusterNode {
+            url: format!("http://{address}"),
+            address,
+            data_dir: data_dirs[0].clone(),
+            data_dirs,
+            pool_idx: pools.len(),
+            process: None,
+        });
+        pools.push(vec![new_idx]);
+        self.topology.node_count = self.nodes.len();
+        self.topology.pools = pools;
+        self.node_extra_env.push(Vec::new());
+        self.node_capture_log_paths.push(None);
+        self.volume_proxy_addresses.push(None);
+
+        if !self.extra_env.iter().any(|(key, _)| key == "RUSTFS_UNSAFE_BYPASS_DISK_CHECK") {
+            self.extra_env
+                .push(("RUSTFS_UNSAFE_BYPASS_DISK_CHECK".to_string(), "true".to_string()));
+        }
+
+        Ok(new_idx)
+    }
+
     /// Gracefully stop one cluster node and wait for its process to exit.
     ///
     /// This is intentionally separate from [`Self::stop_node`]: the latter is

crates/e2e_test/src/distributed/chaos_test.rs

@@ -0,0 +1,102 @@
+// Copyright 2026 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/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 super::harness::{
+    DistCluster, DistLayout, TestResult, assert_object_bytes, bring_drive_online, put_object, retrying_get_equals,
+    take_drive_offline, unique_bucket, wait_for_ready,
+};
+use crate::common::init_logging;
+use crate::fault_proxy::FaultMode;
+use std::time::Duration;
+
+#[tokio::test]
+async fn kill_and_restart_node_preserves_objects() -> TestResult {
+    init_logging();
+    let mut dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let bucket = unique_bucket("killnode");
+    dist.create_bucket(&bucket).await?;
+    let body = vec![0x11u8; 128 * 1024];
+    put_object(&dist.client(0)?, &bucket, "keep.bin", body.clone()).await?;
+
+    dist.cluster.stop_node(3)?;
+    retrying_get_equals(&dist.client(0)?, &bucket, "keep.bin", &body, Duration::from_secs(20)).await?;
+
+    dist.cluster.start_node(3).await?;
+    wait_for_ready(&dist.cluster).await?;
+    assert_object_bytes(&dist.client(3)?, &bucket, "keep.bin", &body).await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn full_cluster_restart_preserves_objects() -> TestResult {
+    init_logging();
+    let mut dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let bucket = unique_bucket("pwr");
+    dist.create_bucket(&bucket).await?;
+    let body = vec![0x44u8; 64 * 1024];
+    put_object(&dist.client(1)?, &bucket, "survive.bin", body.clone()).await?;
+
+    dist.cluster.stop();
+    dist.cluster.start().await?;
+    wait_for_ready(&dist.cluster).await?;
+    for node_idx in 0..dist.cluster.nodes.len() {
+        assert_object_bytes(&dist.client(node_idx)?, &bucket, "survive.bin", &body).await?;
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn offline_drive_then_replace_keeps_object_readable() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let bucket = unique_bucket("baddrive");
+    dist.create_bucket(&bucket).await?;
+    let body = vec![0x22u8; 96 * 1024];
+    put_object(&dist.client(1)?, &bucket, "durable.bin", body.clone()).await?;
+
+    take_drive_offline(&dist.cluster, 0, 0)?;
+    retrying_get_equals(&dist.client(2)?, &bucket, "durable.bin", &body, Duration::from_secs(20)).await?;
+    bring_drive_online(&dist.cluster, 0, 0)?;
+    retrying_get_equals(&dist.client(3)?, &bucket, "durable.bin", &body, Duration::from_secs(20)).await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn volume_proxy_blackhole_then_restore_keeps_s3_available() -> TestResult {
+    init_logging();
+    let mut cluster =
+        crate::common::RustFSTestClusterEnvironment::with_topology(crate::common::ClusterTopology::single_pool_multidrive(4, 4))
+            .await?;
+    let proxy = cluster.start_volume_proxy_for_node(1).await?;
+    cluster.start().await?;
+    cluster.create_test_bucket("chaos-net").await?;
+    let client = cluster.create_s3_client(0)?;
+    let body = vec![0x33u8; 32 * 1024];
+    put_object(&client, "chaos-net", "via-proxy.bin", body.clone()).await?;
+
+    proxy.set_mode(FaultMode::Blackhole);
+    retrying_get_equals(
+        &cluster.create_s3_client(2)?,
+        "chaos-net",
+        "via-proxy.bin",
+        &body,
+        Duration::from_secs(20),
+    )
+    .await?;
+
+    proxy.set_mode(FaultMode::Pass);
+    assert_object_bytes(&cluster.create_s3_client(3)?, "chaos-net", "via-proxy.bin", &body).await?;
+    proxy.shutdown().await;
+    Ok(())
+}

crates/e2e_test/src/distributed/concurrency_stability_test.rs

@@ -0,0 +1,57 @@
+// Copyright 2026 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/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 super::harness::{DistCluster, DistLayout, TestResult, assert_object_bytes, payload_for, put_object, unique_bucket};
+use crate::common::init_logging;
+use std::sync::Arc;
+use tokio::sync::Barrier;
+
+#[tokio::test]
+async fn four_node_high_concurrency_puts_are_readable_from_every_node() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let bucket = unique_bucket("conc");
+    dist.create_bucket(&bucket).await?;
+    let clients = Arc::new(dist.clients()?);
+    let barrier = Arc::new(Barrier::new(32));
+
+    let mut handles = Vec::new();
+    for idx in 0..32 {
+        let clients = clients.clone();
+        let barrier = barrier.clone();
+        let bucket = bucket.clone();
+        handles.push(tokio::spawn(async move {
+            barrier.wait().await;
+            let client = &clients[idx % clients.len()];
+            let key = format!("c/{idx:02}.bin");
+            let body = payload_for(&key, 16 * 1024);
+            put_object(client, &bucket, &key, body.clone()).await?;
+            Ok::<_, Box<dyn std::error::Error + Send + Sync>>((key, body))
+        }));
+    }
+
+    let mut inventory = Vec::new();
+    for handle in handles {
+        inventory.push(handle.await??);
+    }
+
+    for (node_idx, client) in clients.iter().enumerate() {
+        for (key, body) in &inventory {
+            assert_object_bytes(client, &bucket, key, body)
+                .await
+                .map_err(|error| format!("node {node_idx} failed to read {key}: {error}"))?;
+        }
+    }
+    Ok(())
+}

crates/e2e_test/src/distributed/concurrent_data_movement_test.rs

@@ -0,0 +1,65 @@
+// Copyright 2026 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/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 super::harness::{
+    DistCluster, DistLayout, TestResult, assert_inventory, payload_for, put_inventory, retrying_get_equals, retrying_put,
+    start_decommission, unique_bucket, wait_for_decommission_complete,
+};
+use crate::common::init_logging;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::Barrier;
+
+#[tokio::test]
+async fn concurrent_puts_during_decommission_do_not_lose_baseline_or_new_objects() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourPoolFourDrive).await?;
+    let bucket = unique_bucket("concdecom");
+    dist.create_bucket(&bucket).await?;
+    let baseline_client = dist.client(0)?;
+    let inventory = put_inventory(&baseline_client, &bucket, 10, 24 * 1024).await?;
+
+    start_decommission(&dist.cluster, 0).await?;
+
+    let clients = Arc::new(dist.clients()?);
+    let barrier = Arc::new(Barrier::new(16));
+    let mut handles = Vec::new();
+    for idx in 0..16 {
+        let clients = clients.clone();
+        let barrier = barrier.clone();
+        let bucket = bucket.clone();
+        handles.push(tokio::spawn(async move {
+            barrier.wait().await;
+            let client = &clients[idx % clients.len()];
+            let key = format!("live/{idx:02}.bin");
+            let body = payload_for(&key, 8 * 1024);
+            retrying_put(client, &bucket, &key, body.clone(), Duration::from_secs(45)).await?;
+            Ok::<_, Box<dyn std::error::Error + Send + Sync>>((key, body))
+        }));
+    }
+
+    let mut live_objects = Vec::new();
+    for handle in handles {
+        live_objects.push(handle.await??);
+    }
+
+    wait_for_decommission_complete(&dist.cluster, 0, Duration::from_secs(180)).await?;
+
+    let checker = dist.client(3)?;
+    assert_inventory(&checker, &bucket, &inventory).await?;
+    for (key, body) in live_objects {
+        retrying_get_equals(&checker, &bucket, &key, &body, Duration::from_secs(30)).await?;
+    }
+    Ok(())
+}

crates/e2e_test/src/distributed/data_integrity_movement_test.rs

@@ -0,0 +1,43 @@
+// Copyright 2026 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/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 super::harness::{
+    DistCluster, DistLayout, TestResult, assert_inventory, put_inventory, sha256_hex, start_decommission, unique_bucket,
+    wait_for_decommission_complete,
+};
+use crate::common::init_logging;
+use std::time::Duration;
+
+#[tokio::test]
+async fn decommission_does_not_alter_object_sha256_across_pools() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourPoolFourDrive).await?;
+    let bucket = unique_bucket("integrity");
+    dist.create_bucket(&bucket).await?;
+    let client = dist.client(0)?;
+    let inventory = put_inventory(&client, &bucket, 20, 64 * 1024).await?;
+    let before: Vec<(String, String)> = inventory.iter().map(|(key, body)| (key.clone(), sha256_hex(body))).collect();
+
+    start_decommission(&dist.cluster, 0).await?;
+    wait_for_decommission_complete(&dist.cluster, 0, Duration::from_secs(180)).await?;
+
+    let after_client = dist.client(2)?;
+    assert_inventory(&after_client, &bucket, &inventory).await?;
+    for (key, expected_hash) in before {
+        let got = after_client.get_object().bucket(&bucket).key(&key).send().await?;
+        let body = got.body.collect().await?.into_bytes();
+        assert_eq!(sha256_hex(body.as_ref()), expected_hash, "checksum changed for {key} after decommission");
+    }
+    Ok(())
+}

crates/e2e_test/src/distributed/expand_decommission_rebalance_test.rs

@@ -0,0 +1,74 @@
+// Copyright 2026 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/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 super::harness::{
+    DistCluster, DistLayout, TestResult, assert_inventory, list_pools_json, put_inventory, start_decommission, start_rebalance,
+    unique_bucket, wait_for_decommission_complete, wait_for_rebalance_idle,
+};
+use crate::common::init_logging;
+use std::time::Duration;
+
+#[tokio::test]
+async fn four_node_pool_expand_preserves_objects_then_rebalance() -> TestResult {
+    init_logging();
+    let mut dist = DistCluster::start(DistLayout::TwoPoolFourDrive).await?;
+    let bucket = unique_bucket("expand");
+    dist.create_bucket(&bucket).await?;
+    let client = dist.client(0)?;
+    let inventory = put_inventory(&client, &bucket, 12, 32 * 1024).await?;
+    assert_inventory(&client, &bucket, &inventory).await?;
+
+    dist.cluster.stop();
+    dist.cluster.append_single_node_pool().await?;
+    dist.cluster.append_single_node_pool().await?;
+    assert_eq!(dist.cluster.nodes.len(), 4);
+    dist.cluster.start().await?;
+
+    let after_expand = dist.client(0)?;
+    assert_inventory(&after_expand, &bucket, &inventory).await?;
+    let peer = dist.client(3)?;
+    assert_inventory(&peer, &bucket, &inventory).await?;
+
+    let _ = start_rebalance(&dist.cluster).await?;
+    // Rebalance may finish immediately on a tiny dataset; either idle or a
+    // started-then-completed status is success. A hard failure is not.
+    let _ = wait_for_rebalance_idle(&dist.cluster, Duration::from_secs(90)).await;
+    assert_inventory(&peer, &bucket, &inventory).await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn four_pool_decommission_moves_objects_without_loss() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourPoolFourDrive).await?;
+    let bucket = unique_bucket("decom");
+    dist.create_bucket(&bucket).await?;
+    let client = dist.client(1)?;
+    let inventory = put_inventory(&client, &bucket, 16, 48 * 1024).await?;
+
+    let pools_before = list_pools_json(&dist.cluster).await?;
+    let pool_count = pools_before
+        .as_array()
+        .map(Vec::len)
+        .or_else(|| pools_before.get("pools").and_then(serde_json::Value::as_array).map(Vec::len))
+        .unwrap_or(4);
+    assert!(pool_count >= 4, "expected four pools before decommission: {pools_before}");
+
+    start_decommission(&dist.cluster, 0).await?;
+    wait_for_decommission_complete(&dist.cluster, 0, Duration::from_secs(180)).await?;
+
+    let after = dist.client(3)?;
+    assert_inventory(&after, &bucket, &inventory).await?;
+    Ok(())
+}

crates/e2e_test/src/distributed/extra_test.rs

@@ -0,0 +1,108 @@
+// Copyright 2026 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/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 super::harness::{DistCluster, DistLayout, TestResult, assert_object_bytes, get_object_bytes, put_object, unique_bucket};
+use crate::common::init_logging;
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
+
+#[tokio::test]
+async fn four_node_four_drive_multipart_and_cross_node_listing_agree() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let bucket = unique_bucket("extra");
+    dist.create_bucket(&bucket).await?;
+    let client = dist.client(0)?;
+
+    let key = "multipart.bin";
+    let part1 = vec![0x41u8; 5 * 1024 * 1024];
+    let part2 = vec![0x42u8; 5 * 1024 * 1024];
+    let upload = client.create_multipart_upload().bucket(&bucket).key(key).send().await?;
+    let upload_id = upload.upload_id().ok_or("missing upload id")?.to_string();
+
+    let uploaded1 = client
+        .upload_part()
+        .bucket(&bucket)
+        .key(key)
+        .upload_id(&upload_id)
+        .part_number(1)
+        .body(ByteStream::from(part1.clone()))
+        .send()
+        .await?;
+    let uploaded2 = client
+        .upload_part()
+        .bucket(&bucket)
+        .key(key)
+        .upload_id(&upload_id)
+        .part_number(2)
+        .body(ByteStream::from(part2.clone()))
+        .send()
+        .await?;
+
+    client
+        .complete_multipart_upload()
+        .bucket(&bucket)
+        .key(key)
+        .upload_id(&upload_id)
+        .multipart_upload(
+            CompletedMultipartUpload::builder()
+                .parts(
+                    CompletedPart::builder()
+                        .part_number(1)
+                        .e_tag(uploaded1.e_tag().unwrap_or_default())
+                        .build(),
+                )
+                .parts(
+                    CompletedPart::builder()
+                        .part_number(2)
+                        .e_tag(uploaded2.e_tag().unwrap_or_default())
+                        .build(),
+                )
+                .build(),
+        )
+        .send()
+        .await?;
+
+    let mut expected = part1;
+    expected.extend_from_slice(&part2);
+    for node_idx in 0..dist.cluster.nodes.len() {
+        assert_object_bytes(&dist.client(node_idx)?, &bucket, key, &expected).await?;
+    }
+
+    put_object(&client, &bucket, "list/a", b"a".to_vec()).await?;
+    put_object(&dist.client(2)?, &bucket, "list/b", b"b".to_vec()).await?;
+    let mut seen = Vec::new();
+    for node_idx in 0..dist.cluster.nodes.len() {
+        let listed = dist
+            .client(node_idx)?
+            .list_objects_v2()
+            .bucket(&bucket)
+            .prefix("list/")
+            .send()
+            .await?;
+        let keys: Vec<String> = listed
+            .contents()
+            .iter()
+            .filter_map(|object| object.key().map(str::to_string))
+            .collect();
+        seen.push(keys);
+    }
+    for keys in &seen[1..] {
+        assert_eq!(&seen[0], keys, "list results diverged across nodes: {seen:?}");
+    }
+
+    let got = get_object_bytes(&dist.client(3)?, &bucket, "list/a").await?;
+    assert_eq!(got, b"a");
+    Ok(())
+}

crates/e2e_test/src/distributed/harness.rs

@@ -0,0 +1,688 @@
+// Copyright 2026 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.
+
+//! Shared 4-node distributed e2e helpers.
+//!
+//! Two localhost-expressible layouts cover the suite:
+//!
+//! * **4×4 single pool** (`four_by_four`) — four processes, four drives each,
+//!   one `DistErasure` pool (16 explicit volume endpoints). This is the
+//!   default S3 / lock / versioning / chaos topology.
+//! * **4×4 four pool** (`four_pool_four_drive`) — four single-node pools of
+//!   four drives. Required for decommission/rebalance/expand, which the
+//!   server rejects on a single pool.
+//!
+//! Genuine multi-node *striped* pools still need multi-host CI (backlog
+//! #1313 / #1314). Site replication uses two 4-node 1-drive clusters so the
+//! process count stays at eight rather than sixteen.
+
+use crate::common::{
+    ClusterTopology, FAST_DATA_USAGE_SCANNER_ENV, RustFSTestClusterEnvironment, admin_request, local_http_client,
+    replication_fast_env, signed_request,
+};
+use crate::replication_extension_test::LOOPBACK_REPLICATION_TARGET_ENV;
+use aws_sdk_s3::Client;
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
+use http::{Method, StatusCode};
+use sha2::{Digest, Sha256};
+use std::collections::BTreeMap;
+use std::path::Path;
+use std::time::Duration;
+use tokio::time::{Instant, sleep};
+use uuid::Uuid;
+
+pub(crate) type TestResult<T = ()> = Result<T, Box<dyn std::error::Error + Send + Sync>>;
+
+pub(crate) const NODE_COUNT: usize = 4;
+pub(crate) const DRIVES_PER_NODE: usize = 4;
+
+#[derive(Clone, Copy, Debug)]
+pub(crate) enum DistLayout {
+    /// 4 nodes × 4 drives, one erasure pool spanning every endpoint.
+    FourByFour,
+    /// 4 nodes × 1 drive, one erasure pool (minimum 4-node 4-disk layout).
+    FourNodeFourDisk,
+    /// 4 single-node pools, 4 drives each (expand / decommission / rebalance).
+    FourPoolFourDrive,
+    /// 2 single-node pools, 4 drives each (expansion seed).
+    TwoPoolFourDrive,
+}
+
+pub(crate) struct DistCluster {
+    pub cluster: RustFSTestClusterEnvironment,
+}
+
+impl DistCluster {
+    pub async fn start(layout: DistLayout) -> TestResult<Self> {
+        Self::start_with_env(layout, &[]).await
+    }
+
+    pub async fn start_with_env(layout: DistLayout, extra_env: &[(&str, &str)]) -> TestResult<Self> {
+        let topology = match layout {
+            DistLayout::FourByFour => ClusterTopology::single_pool_multidrive(NODE_COUNT, DRIVES_PER_NODE),
+            DistLayout::FourNodeFourDisk => ClusterTopology::single_pool(NODE_COUNT),
+            DistLayout::FourPoolFourDrive => {
+                ClusterTopology::per_node_pools(DRIVES_PER_NODE, (0..NODE_COUNT).map(|idx| vec![idx]).collect())
+            }
+            DistLayout::TwoPoolFourDrive => ClusterTopology::per_node_pools(DRIVES_PER_NODE, vec![vec![0], vec![1]]),
+        };
+        let mut cluster = RustFSTestClusterEnvironment::with_topology(topology).await?;
+        for &(key, value) in extra_env {
+            cluster.set_env(key, value);
+        }
+        cluster.start().await?;
+        Ok(Self { cluster })
+    }
+
+    pub async fn start_replication_pair() -> TestResult<(Self, Self)> {
+        let mut extra: Vec<(&str, &str)> = replication_fast_env();
+        extra.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
+        extra.extend_from_slice(FAST_DATA_USAGE_SCANNER_ENV);
+        let source = Self::start_with_env(DistLayout::FourNodeFourDisk, &extra).await?;
+        let target = Self::start_with_env(DistLayout::FourNodeFourDisk, &extra).await?;
+        Ok((source, target))
+    }
+
+    pub fn client(&self, node_idx: usize) -> TestResult<Client> {
+        self.cluster.create_s3_client(node_idx)
+    }
+
+    pub fn clients(&self) -> TestResult<Vec<Client>> {
+        self.cluster.create_all_clients()
+    }
+
+    pub async fn create_bucket(&self, bucket: &str) -> TestResult {
+        self.cluster.create_test_bucket(bucket).await
+    }
+}
+
+pub(crate) fn unique_bucket(prefix: &str) -> String {
+    let id = Uuid::new_v4().simple().to_string();
+    format!("{prefix}-{}", &id[..12])
+}
+
+pub(crate) fn sha256_hex(bytes: &[u8]) -> String {
+    let digest = Sha256::digest(bytes);
+    digest.iter().map(|byte| format!("{byte:02x}")).collect()
+}
+
+pub(crate) fn payload_for(key: &str, size: usize) -> Vec<u8> {
+    let seed = key.as_bytes();
+    (0..size)
+        .map(|idx| seed.get(idx % seed.len()).copied().unwrap_or(0) ^ (idx as u8))
+        .collect()
+}
+
+pub(crate) async fn put_object(client: &Client, bucket: &str, key: &str, body: Vec<u8>) -> TestResult {
+    client
+        .put_object()
+        .bucket(bucket)
+        .key(key)
+        .body(ByteStream::from(body))
+        .send()
+        .await?;
+    Ok(())
+}
+
+pub(crate) async fn get_object_bytes(client: &Client, bucket: &str, key: &str) -> TestResult<Vec<u8>> {
+    let output = client.get_object().bucket(bucket).key(key).send().await?;
+    Ok(output.body.collect().await?.into_bytes().to_vec())
+}
+
+pub(crate) async fn assert_object_bytes(client: &Client, bucket: &str, key: &str, expected: &[u8]) -> TestResult {
+    let got = get_object_bytes(client, bucket, key).await?;
+    if got.as_slice() != expected {
+        return Err(format!(
+            "object {bucket}/{key} bytes mismatch: expected {} bytes sha256={} got {} bytes sha256={}",
+            expected.len(),
+            sha256_hex(expected),
+            got.len(),
+            sha256_hex(&got)
+        )
+        .into());
+    }
+    Ok(())
+}
+
+pub(crate) async fn put_inventory(
+    client: &Client,
+    bucket: &str,
+    count: usize,
+    size: usize,
+) -> TestResult<BTreeMap<String, Vec<u8>>> {
+    let mut inventory = BTreeMap::new();
+    for idx in 0..count {
+        let key = format!("obj-{idx:04}");
+        let body = payload_for(&key, size);
+        put_object(client, bucket, &key, body.clone()).await?;
+        inventory.insert(key, body);
+    }
+    Ok(inventory)
+}
+
+pub(crate) async fn assert_inventory(client: &Client, bucket: &str, inventory: &BTreeMap<String, Vec<u8>>) -> TestResult {
+    for (key, expected) in inventory {
+        assert_object_bytes(client, bucket, key, expected).await?;
+    }
+    Ok(())
+}
+
+pub(crate) async fn enable_versioning(client: &Client, bucket: &str) -> TestResult {
+    client
+        .put_bucket_versioning()
+        .bucket(bucket)
+        .versioning_configuration(
+            VersioningConfiguration::builder()
+                .status(BucketVersioningStatus::Enabled)
+                .build(),
+        )
+        .send()
+        .await?;
+    Ok(())
+}
+
+pub(crate) async fn wait_until<F, Fut>(timeout: Duration, mut probe: F, label: &str) -> TestResult
+where
+    F: FnMut() -> Fut,
+    Fut: std::future::Future<Output = TestResult<bool>>,
+{
+    let deadline = Instant::now() + timeout;
+    let mut delay = Duration::from_millis(50);
+    loop {
+        let last_error = match probe().await {
+            Ok(true) => return Ok(()),
+            Ok(false) => format!("{label} still false"),
+            Err(error) => error.to_string(),
+        };
+        if Instant::now() >= deadline {
+            return Err(format!("{label} did not become true within {timeout:?}: {last_error}").into());
+        }
+        sleep(delay).await;
+        delay = (delay * 2).min(Duration::from_secs(1));
+    }
+}
+
+pub(crate) async fn cluster_admin(
+    cluster: &RustFSTestClusterEnvironment,
+    method: Method,
+    path_and_query: &str,
+    body: Option<String>,
+) -> TestResult<(StatusCode, String)> {
+    admin_request(
+        &cluster.nodes[0].url,
+        method,
+        path_and_query,
+        body,
+        &cluster.access_key,
+        &cluster.secret_key,
+    )
+    .await
+}
+
+pub(crate) async fn cluster_admin_ok(
+    cluster: &RustFSTestClusterEnvironment,
+    method: Method,
+    path_and_query: &str,
+    body: Option<String>,
+) -> TestResult<String> {
+    let (status, response) = cluster_admin(cluster, method.clone(), path_and_query, body).await?;
+    if !status.is_success() {
+        return Err(format!("{method} {path_and_query} failed: {status} {response}").into());
+    }
+    Ok(response)
+}
+
+pub(crate) async fn wait_for_ready(cluster: &RustFSTestClusterEnvironment) -> TestResult {
+    let client = local_http_client();
+    for node in &cluster.nodes {
+        let url = format!("{}/health/ready", node.url);
+        wait_until(
+            Duration::from_secs(30),
+            || {
+                let client = client.clone();
+                let url = url.clone();
+                async move {
+                    match client.get(&url).send().await {
+                        Ok(response) if response.status().is_success() => Ok(true),
+                        _ => Ok(false),
+                    }
+                }
+            },
+            &format!("node {} ready", node.address),
+        )
+        .await?;
+    }
+    Ok(())
+}
+
+pub(crate) fn take_drive_offline(
+    cluster: &RustFSTestClusterEnvironment,
+    node_idx: usize,
+    drive_idx: usize,
+) -> TestResult<String> {
+    let dir = cluster
+        .nodes
+        .get(node_idx)
+        .and_then(|node| node.data_dirs.get(drive_idx))
+        .ok_or("invalid node/drive index")?;
+    let offline = format!("{dir}.offline");
+    if Path::new(&offline).exists() {
+        return Err(format!("drive already offline: {offline}").into());
+    }
+    std::fs::rename(dir, &offline)?;
+    Ok(offline)
+}
+
+pub(crate) fn bring_drive_online(cluster: &RustFSTestClusterEnvironment, node_idx: usize, drive_idx: usize) -> TestResult {
+    let dir = cluster
+        .nodes
+        .get(node_idx)
+        .and_then(|node| node.data_dirs.get(drive_idx))
+        .ok_or("invalid node/drive index")?;
+    let offline = format!("{dir}.offline");
+    if Path::new(dir).exists() {
+        std::fs::remove_dir_all(dir)?;
+    }
+    std::fs::rename(&offline, dir)?;
+    Ok(())
+}
+
+pub(crate) async fn set_remote_target(
+    source: &RustFSTestClusterEnvironment,
+    source_bucket: &str,
+    target: &RustFSTestClusterEnvironment,
+    target_bucket: &str,
+) -> TestResult<String> {
+    let body = serde_json::json!({
+        "endpoint": target.nodes[0].address,
+        "credentials": {
+            "accessKey": target.access_key,
+            "secretKey": target.secret_key
+        },
+        "targetbucket": target_bucket,
+        "secure": false,
+        "type": "replication"
+    });
+    let url = format!(
+        "{}/rustfs/admin/v3/set-remote-target?bucket={}",
+        source.nodes[0].url,
+        urlencoding::encode(source_bucket)
+    );
+    let response = signed_request(
+        Method::PUT,
+        &url,
+        &source.access_key,
+        &source.secret_key,
+        Some(body.to_string().into_bytes()),
+        Some("application/json"),
+    )
+    .await?;
+    if response.status() != StatusCode::OK {
+        let status = response.status();
+        let body = response.text().await.unwrap_or_default();
+        return Err(format!("set remote target failed: {status} {body}").into());
+    }
+    Ok(serde_json::from_slice(&response.bytes().await?)?)
+}
+
+pub(crate) async fn put_bucket_replication(source: &RustFSTestClusterEnvironment, bucket: &str, target_arn: &str) -> TestResult {
+    let body = format!(
+        r#"<ReplicationConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <Role></Role>
+  <Rule>
+    <ID>rule-1</ID>
+    <Priority>1</Priority>
+    <Status>Enabled</Status>
+    <DeleteMarkerReplication>
+      <Status>Enabled</Status>
+    </DeleteMarkerReplication>
+    <ExistingObjectReplication>
+      <Status>Enabled</Status>
+    </ExistingObjectReplication>
+    <Destination>
+      <Bucket>{target_arn}</Bucket>
+    </Destination>
+  </Rule>
+</ReplicationConfiguration>"#
+    );
+    let url = format!("{}/{bucket}?replication", source.nodes[0].url);
+    let response = signed_request(
+        Method::PUT,
+        &url,
+        &source.access_key,
+        &source.secret_key,
+        Some(body.into_bytes()),
+        Some("application/xml"),
+    )
+    .await?;
+    if !response.status().is_success() {
+        let status = response.status();
+        let body = response.text().await.unwrap_or_default();
+        return Err(format!("put bucket replication failed: {status} {body}").into());
+    }
+    Ok(())
+}
+
+pub(crate) async fn wait_for_replicated_bytes(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    expected: &[u8],
+    timeout: Duration,
+) -> TestResult {
+    wait_until(
+        timeout,
+        || async {
+            match get_object_bytes(client, bucket, key).await {
+                Ok(got) if got.as_slice() == expected => Ok(true),
+                Ok(_) => Ok(false),
+                Err(error) => {
+                    let message = error.to_string();
+                    if message.contains("NoSuchKey") || message.contains("NotFound") {
+                        Ok(false)
+                    } else {
+                        Err(error)
+                    }
+                }
+            }
+        },
+        &format!("replicated object {bucket}/{key}"),
+    )
+    .await
+}
+
+pub(crate) async fn set_bucket_quota(cluster: &RustFSTestClusterEnvironment, bucket: &str, quota_bytes: u64) -> TestResult {
+    wait_until(
+        Duration::from_secs(30),
+        || async {
+            let (status, _) =
+                cluster_admin(cluster, Method::GET, &format!("/rustfs/admin/v3/quota-stats/{bucket}"), None).await?;
+            Ok(status.is_success() || status == StatusCode::NOT_FOUND)
+        },
+        "quota stats ready",
+    )
+    .await?;
+    let body = serde_json::json!({ "quota": quota_bytes, "quota_type": "HARD" }).to_string();
+    wait_until(
+        Duration::from_secs(30),
+        || async {
+            let (status, response) =
+                cluster_admin(cluster, Method::PUT, &format!("/rustfs/admin/v3/quota/{bucket}"), Some(body.clone())).await?;
+            if status.is_success() {
+                return Ok(true);
+            }
+            if status == StatusCode::SERVICE_UNAVAILABLE {
+                return Ok(false);
+            }
+            Err(format!("failed to set quota for {bucket}: {status} {response}").into())
+        },
+        "set hard quota",
+    )
+    .await
+}
+
+pub(crate) async fn start_decommission(cluster: &RustFSTestClusterEnvironment, pool_id: usize) -> TestResult<String> {
+    cluster_admin_ok(
+        cluster,
+        Method::POST,
+        &format!("/rustfs/admin/v3/pools/decommission?pool={pool_id}&by-id=true"),
+        None,
+    )
+    .await
+}
+
+pub(crate) async fn decommission_status_json(cluster: &RustFSTestClusterEnvironment) -> TestResult<serde_json::Value> {
+    let body = cluster_admin_ok(cluster, Method::GET, "/rustfs/admin/v3/decommission/status", None).await?;
+    Ok(serde_json::from_str(&body)?)
+}
+
+fn pool_entry(status: &serde_json::Value, pool_id: usize) -> Option<&serde_json::Value> {
+    if let Some(pools) = status.get("pools").and_then(serde_json::Value::as_array) {
+        return pools
+            .iter()
+            .find(|pool| pool.get("id").and_then(serde_json::Value::as_u64) == Some(pool_id as u64));
+    }
+    if status.get("id").and_then(serde_json::Value::as_u64) == Some(pool_id as u64) {
+        Some(status)
+    } else {
+        None
+    }
+}
+
+fn decommission_pool_failed(pool: &serde_json::Value) -> bool {
+    let info = pool.get("decommissionInfo");
+    let flagged = |key: &str| info.and_then(|value| value.get(key)).and_then(serde_json::Value::as_bool) == Some(true);
+    flagged("failed")
+        || flagged("canceled")
+        || pool
+            .get("status")
+            .and_then(serde_json::Value::as_str)
+            .is_some_and(|status| status.eq_ignore_ascii_case("failed") || status.eq_ignore_ascii_case("canceled"))
+}
+
+pub(crate) fn decommission_complete(status: &serde_json::Value, pool_id: usize) -> bool {
+    let Some(pool) = pool_entry(status, pool_id) else {
+        return false;
+    };
+    if decommission_pool_failed(pool) {
+        return false;
+    }
+    let info_complete = pool
+        .get("decommissionInfo")
+        .and_then(|value| value.get("complete"))
+        .and_then(serde_json::Value::as_bool)
+        == Some(true);
+    let status_text = pool.get("status").and_then(serde_json::Value::as_str).unwrap_or("");
+    let pool_status = pool.get("poolStatus").and_then(serde_json::Value::as_str).unwrap_or("");
+    info_complete || status_text.eq_ignore_ascii_case("complete") || pool_status.eq_ignore_ascii_case("decommissioned")
+}
+
+pub(crate) fn decommission_failed(status: &serde_json::Value, pool_id: usize) -> bool {
+    pool_entry(status, pool_id).is_some_and(decommission_pool_failed)
+}
+
+pub(crate) async fn wait_for_decommission_complete(
+    cluster: &RustFSTestClusterEnvironment,
+    pool_id: usize,
+    timeout: Duration,
+) -> TestResult {
+    wait_until(
+        timeout,
+        || async {
+            let status = decommission_status_json(cluster).await?;
+            if decommission_failed(&status, pool_id) {
+                return Err(format!("decommission failed for pool {pool_id}: {status}").into());
+            }
+            Ok(decommission_complete(&status, pool_id))
+        },
+        "decommission complete",
+    )
+    .await
+}
+
+pub(crate) async fn start_rebalance(cluster: &RustFSTestClusterEnvironment) -> TestResult<String> {
+    cluster_admin_ok(cluster, Method::POST, "/rustfs/admin/v3/rebalance/start", None).await
+}
+
+pub(crate) async fn rebalance_status_json(cluster: &RustFSTestClusterEnvironment) -> TestResult<serde_json::Value> {
+    let body = cluster_admin_ok(cluster, Method::GET, "/rustfs/admin/v3/rebalance/status", None).await?;
+    Ok(serde_json::from_str(&body)?)
+}
+
+pub(crate) fn rebalance_active(status: &serde_json::Value) -> bool {
+    status
+        .get("pools")
+        .and_then(serde_json::Value::as_array)
+        .is_some_and(|pools| {
+            pools.iter().any(|pool| {
+                let stopping = pool.get("stopping").and_then(serde_json::Value::as_bool) == Some(true);
+                let value = pool.get("status").and_then(serde_json::Value::as_str).unwrap_or("");
+                stopping
+                    || value.eq_ignore_ascii_case("started")
+                    || value.eq_ignore_ascii_case("active")
+                    || value.eq_ignore_ascii_case("running")
+                    || value.eq_ignore_ascii_case("stopping")
+            })
+        })
+}
+
+pub(crate) async fn wait_for_rebalance_idle(cluster: &RustFSTestClusterEnvironment, timeout: Duration) -> TestResult {
+    wait_until(
+        timeout,
+        || async {
+            match rebalance_status_json(cluster).await {
+                Ok(status) => Ok(!rebalance_active(&status)),
+                Err(error) => {
+                    let message = error.to_string();
+                    if message.contains("NoSuchResource") || message.contains("404") || message.contains("not started") {
+                        Ok(true)
+                    } else {
+                        Err(error)
+                    }
+                }
+            }
+        },
+        "rebalance idle",
+    )
+    .await
+}
+
+pub(crate) async fn list_pools_json(cluster: &RustFSTestClusterEnvironment) -> TestResult<serde_json::Value> {
+    let body = cluster_admin_ok(cluster, Method::GET, "/rustfs/admin/v3/pools/list", None).await?;
+    Ok(serde_json::from_str(&body)?)
+}
+
+pub(crate) async fn retrying_put(client: &Client, bucket: &str, key: &str, body: Vec<u8>, timeout: Duration) -> TestResult {
+    wait_until(
+        timeout,
+        || {
+            let client = client.clone();
+            let bucket = bucket.to_string();
+            let key = key.to_string();
+            let body = body.clone();
+            async move {
+                match put_object(&client, &bucket, &key, body).await {
+                    Ok(()) => Ok(true),
+                    Err(error) => {
+                        let message = error.to_string();
+                        if message.contains("SlowDown")
+                            || message.contains("ServiceUnavailable")
+                            || message.contains("InternalError")
+                            || message.contains("503")
+                            || message.contains("500")
+                        {
+                            Ok(false)
+                        } else {
+                            Err(error)
+                        }
+                    }
+                }
+            }
+        },
+        &format!("put {bucket}/{key} during data movement"),
+    )
+    .await
+}
+
+pub(crate) async fn retrying_get_equals(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    expected: &[u8],
+    timeout: Duration,
+) -> TestResult {
+    wait_until(
+        timeout,
+        || async {
+            match get_object_bytes(client, bucket, key).await {
+                Ok(got) if got.as_slice() == expected => Ok(true),
+                Ok(_) => Ok(false),
+                Err(error) => {
+                    let message = error.to_string();
+                    if message.contains("NoSuchKey")
+                        || message.contains("SlowDown")
+                        || message.contains("ServiceUnavailable")
+                        || message.contains("InternalError")
+                        || message.contains("503")
+                        || message.contains("500")
+                    {
+                        Ok(false)
+                    } else {
+                        Err(error)
+                    }
+                }
+            }
+        },
+        &format!("get {bucket}/{key} during data movement"),
+    )
+    .await
+}
+
+#[tokio::test]
+async fn append_single_node_pool_extends_ellipses_volumes() {
+    let mut env = RustFSTestClusterEnvironment::with_topology(ClusterTopology::per_node_pools(2, vec![vec![0], vec![1]]))
+        .await
+        .expect("two-pool seed topology");
+    assert_eq!(env.rustfs_volumes_arg().split(' ').count(), 2);
+
+    let added = env.append_single_node_pool().await.expect("append third pool");
+    assert_eq!(added, 2);
+    assert_eq!(env.nodes.len(), 3);
+    assert_eq!(env.nodes[2].pool_idx, 2);
+    assert_eq!(env.nodes[2].data_dirs.len(), 2);
+    let volumes = env.rustfs_volumes_arg();
+    assert_eq!(volumes.split(' ').count(), 3, "expected three pool arguments, got: {volumes}");
+    assert!(volumes.contains("/drive{0...1}"), "expanded layout must keep drive ellipses: {volumes}");
+}
+
+#[tokio::test]
+async fn append_single_node_pool_rejects_striped_single_pool() {
+    let mut env = RustFSTestClusterEnvironment::new(4).await.expect("four-node single pool");
+    let err = env
+        .append_single_node_pool()
+        .await
+        .expect_err("a striped single pool cannot gain a localhost pool");
+    let message = err.to_string();
+    assert!(
+        message.contains("drives_per_node") || message.contains("one node per pool"),
+        "unexpected error: {message}"
+    );
+}
+
+#[test]
+fn decommission_complete_reads_pool_status_and_info_flag() {
+    let status = serde_json::json!({
+        "pools": [
+            {
+                "id": 0,
+                "status": "complete",
+                "poolStatus": "decommissioned",
+                "decommissionInfo": { "complete": true, "failed": false, "canceled": false }
+            },
+            { "id": 1, "status": "none", "poolStatus": "active" }
+        ]
+    });
+    assert!(decommission_complete(&status, 0));
+    assert!(!decommission_complete(&status, 1));
+    assert!(!decommission_failed(&status, 0));
+}
+
+#[test]
+fn rebalance_active_treats_started_as_in_progress() {
+    let started = serde_json::json!({ "pools": [{ "id": 0, "status": "Started", "stopping": false }] });
+    let done = serde_json::json!({ "pools": [{ "id": 0, "status": "Completed", "stopping": false }] });
+    assert!(rebalance_active(&started));
+    assert!(!rebalance_active(&done));
+}

crates/e2e_test/src/distributed/mod.rs

@@ -0,0 +1,34 @@
+// Copyright 2026 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.
+
+//! 4-node 4-drive distributed e2e coverage.
+//!
+//! Selected by `[profile.e2e-distributed]` and run from
+//! `.github/workflows/e2e-distributed.yml`. Excluded from `e2e-full` because
+//! each case starts four real `rustfs` processes.
+
+mod chaos_test;
+mod concurrency_stability_test;
+mod concurrent_data_movement_test;
+mod data_integrity_movement_test;
+mod expand_decommission_rebalance_test;
+mod extra_test;
+mod harness;
+mod object_lock_test;
+mod observability_test;
+mod replication_quota_test;
+mod s3_basic_test;
+mod s3_during_data_movement_test;
+mod site_replication_test;
+mod versioning_test;

crates/e2e_test/src/distributed/object_lock_test.rs

@@ -0,0 +1,137 @@
+// Copyright 2026 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/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 super::harness::{DistCluster, DistLayout, TestResult, put_object, unique_bucket};
+use crate::common::init_logging;
+use aws_sdk_s3::error::ProvideErrorMetadata;
+use aws_sdk_s3::types::{
+    DefaultRetention, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockLegalHold, ObjectLockLegalHoldStatus,
+    ObjectLockRetention, ObjectLockRetentionMode, ObjectLockRule,
+};
+use chrono::{Duration as ChronoDuration, Utc};
+
+#[tokio::test]
+async fn four_node_four_drive_object_lock_worm_blocks_delete() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let client = dist.client(0)?;
+    let peer = dist.client(2)?;
+    let bucket = unique_bucket("objlock");
+
+    client
+        .create_bucket()
+        .bucket(&bucket)
+        .object_lock_enabled_for_bucket(true)
+        .send()
+        .await?;
+
+    let retain_until = Utc::now() + ChronoDuration::days(1);
+    let retain_until_s3 = aws_sdk_s3::primitives::DateTime::from_secs(retain_until.timestamp());
+
+    client
+        .put_object_lock_configuration()
+        .bucket(&bucket)
+        .object_lock_configuration(
+            ObjectLockConfiguration::builder()
+                .object_lock_enabled(ObjectLockEnabled::Enabled)
+                .rule(
+                    ObjectLockRule::builder()
+                        .default_retention(
+                            DefaultRetention::builder()
+                                .mode(ObjectLockRetentionMode::Governance)
+                                .days(1)
+                                .build(),
+                        )
+                        .build(),
+                )
+                .build(),
+        )
+        .send()
+        .await?;
+
+    let compliance_key = "compliance.bin";
+    put_object(&client, &bucket, compliance_key, b"locked-compliance".to_vec()).await?;
+    client
+        .put_object_retention()
+        .bucket(&bucket)
+        .key(compliance_key)
+        .retention(
+            ObjectLockRetention::builder()
+                .mode(ObjectLockRetentionMode::Compliance)
+                .retain_until_date(retain_until_s3)
+                .build(),
+        )
+        .send()
+        .await?;
+
+    let compliance_delete = peer.delete_object().bucket(&bucket).key(compliance_key).send().await;
+    match compliance_delete {
+        Ok(_) => return Err("COMPLIANCE retention must block DeleteObject".into()),
+        Err(error) => {
+            let code = error.as_service_error().and_then(ProvideErrorMetadata::code);
+            assert_eq!(code, Some("AccessDenied"), "unexpected COMPLIANCE delete error: {error:?}");
+        }
+    }
+
+    let governance_key = "governance.bin";
+    put_object(&client, &bucket, governance_key, b"locked-governance".to_vec()).await?;
+    client
+        .put_object_retention()
+        .bucket(&bucket)
+        .key(governance_key)
+        .retention(
+            ObjectLockRetention::builder()
+                .mode(ObjectLockRetentionMode::Governance)
+                .retain_until_date(retain_until_s3)
+                .build(),
+        )
+        .send()
+        .await?;
+
+    let governance_blocked = peer.delete_object().bucket(&bucket).key(governance_key).send().await;
+    match governance_blocked {
+        Ok(_) => return Err("GOVERNANCE retention must block DeleteObject without bypass".into()),
+        Err(error) => {
+            let code = error.as_service_error().and_then(ProvideErrorMetadata::code);
+            assert_eq!(code, Some("AccessDenied"), "unexpected GOVERNANCE delete error: {error:?}");
+        }
+    }
+
+    peer.delete_object()
+        .bucket(&bucket)
+        .key(governance_key)
+        .bypass_governance_retention(true)
+        .send()
+        .await?;
+
+    let hold_key = "legal-hold.bin";
+    put_object(&client, &bucket, hold_key, b"legal-hold".to_vec()).await?;
+    client
+        .put_object_legal_hold()
+        .bucket(&bucket)
+        .key(hold_key)
+        .legal_hold(ObjectLockLegalHold::builder().status(ObjectLockLegalHoldStatus::On).build())
+        .send()
+        .await?;
+    let hold_delete = peer.delete_object().bucket(&bucket).key(hold_key).send().await;
+    match hold_delete {
+        Ok(_) => return Err("legal hold must block DeleteObject".into()),
+        Err(error) => {
+            let code = error.as_service_error().and_then(ProvideErrorMetadata::code);
+            assert_eq!(code, Some("AccessDenied"), "unexpected legal-hold delete error: {error:?}");
+        }
+    }
+
+    Ok(())
+}

crates/e2e_test/src/distributed/observability_test.rs

@@ -0,0 +1,61 @@
+// Copyright 2026 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/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 super::harness::{DistCluster, DistLayout, TestResult, cluster_admin_ok, put_object, unique_bucket, wait_for_ready};
+use crate::common::{init_logging, local_http_client};
+use http::Method;
+
+#[tokio::test]
+async fn four_node_four_drive_health_admin_info_and_audit_list() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourByFour).await?;
+    wait_for_ready(&dist.cluster).await?;
+
+    let http = local_http_client();
+    for node in &dist.cluster.nodes {
+        let ready = http.get(format!("{}/health/ready", node.url)).send().await?;
+        assert!(ready.status().is_success(), "node {} not ready: {}", node.address, ready.status());
+        let live = http.get(format!("{}/health/live", node.url)).send().await;
+        if let Ok(response) = live {
+            assert!(
+                response.status().is_success() || response.status().as_u16() == 404,
+                "unexpected live probe on {}: {}",
+                node.address,
+                response.status()
+            );
+        }
+    }
+
+    let info = cluster_admin_ok(&dist.cluster, Method::GET, "/rustfs/admin/v3/info", None).await?;
+    assert!(!info.is_empty(), "admin info was empty");
+    let storage = cluster_admin_ok(&dist.cluster, Method::GET, "/rustfs/admin/v3/storageinfo", None).await?;
+    assert!(
+        storage.contains("disks") || storage.contains("backend") || storage.contains("info"),
+        "storageinfo missing expected fields: {storage}"
+    );
+
+    let audit = cluster_admin_ok(&dist.cluster, Method::GET, "/rustfs/admin/v3/audit/target/list", None).await?;
+    let trimmed = audit.trim();
+    if !trimmed.is_empty() && trimmed != "null" && !trimmed.starts_with('[') && !trimmed.starts_with('{') {
+        return Err(format!("audit target list was not machine-readable: {audit}").into());
+    }
+
+    let bucket = unique_bucket("obs");
+    dist.create_bucket(&bucket).await?;
+    put_object(&dist.client(0)?, &bucket, "probe.log", b"observability".to_vec()).await?;
+
+    let trace = cluster_admin_ok(&dist.cluster, Method::GET, "/rustfs/admin/v3/info", None).await?;
+    assert!(!trace.is_empty());
+    Ok(())
+}

crates/e2e_test/src/distributed/replication_quota_test.rs

@@ -0,0 +1,98 @@
+// Copyright 2026 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/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 super::harness::{
+    DistCluster, DistLayout, TestResult, enable_versioning, put_bucket_replication, put_object, set_bucket_quota,
+    set_remote_target, unique_bucket, wait_for_replicated_bytes,
+};
+use crate::common::{FAST_DATA_USAGE_SCANNER_ENV, init_logging};
+use aws_sdk_s3::error::ProvideErrorMetadata;
+use std::time::Duration;
+
+#[tokio::test]
+async fn four_node_bucket_replication_converges_to_peer_cluster() -> TestResult {
+    init_logging();
+    let (source, target) = DistCluster::start_replication_pair().await?;
+    let source_bucket = unique_bucket("replsrc");
+    let target_bucket = unique_bucket("repldst");
+    source.create_bucket(&source_bucket).await?;
+    target.create_bucket(&target_bucket).await?;
+
+    let source_client = source.client(0)?;
+    let target_client = target.client(0)?;
+    enable_versioning(&source_client, &source_bucket).await?;
+    enable_versioning(&target_client, &target_bucket).await?;
+
+    let arn = set_remote_target(&source.cluster, &source_bucket, &target.cluster, &target_bucket).await?;
+    put_bucket_replication(&source.cluster, &source_bucket, &arn).await?;
+
+    let key = "replicated.bin";
+    let body = b"distributed-bucket-replication".to_vec();
+    put_object(&source_client, &source_bucket, key, body.clone()).await?;
+    wait_for_replicated_bytes(&target_client, &target_bucket, key, &body, Duration::from_secs(45)).await?;
+
+    let peer_read = target.client(3)?;
+    wait_for_replicated_bytes(&peer_read, &target_bucket, key, &body, Duration::from_secs(15)).await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn four_node_four_drive_hard_quota_rejects_over_limit_put() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start_with_env(DistLayout::FourByFour, FAST_DATA_USAGE_SCANNER_ENV).await?;
+    let bucket = unique_bucket("quota");
+    dist.create_bucket(&bucket).await?;
+    set_bucket_quota(&dist.cluster, &bucket, 8 * 1024).await?;
+
+    let client = dist.client(1)?;
+    put_object(&client, &bucket, "small.bin", vec![0u8; 1024]).await?;
+
+    let over_limit = client
+        .put_object()
+        .bucket(&bucket)
+        .key("too-big.bin")
+        .body(vec![0u8; 16 * 1024].into())
+        .send()
+        .await;
+    match over_limit {
+        Ok(_) => {
+            // Scanner-backed quota can lag a cycle; a second over-quota PUT must fail.
+            let second = client
+                .put_object()
+                .bucket(&bucket)
+                .key("too-big-2.bin")
+                .body(vec![0u8; 16 * 1024].into())
+                .send()
+                .await;
+            match second {
+                Ok(_) => return Err("hard quota admitted two oversized PUTs on a 4x4 cluster".into()),
+                Err(error) => {
+                    let code = error.as_service_error().and_then(ProvideErrorMetadata::code);
+                    assert!(
+                        matches!(code, Some("QuotaExceeded" | "SlowDown" | "AccessDenied" | "InvalidRequest")),
+                        "unexpected over-quota error: {error:?}"
+                    );
+                }
+            }
+        }
+        Err(error) => {
+            let code = error.as_service_error().and_then(ProvideErrorMetadata::code);
+            assert!(
+                matches!(code, Some("QuotaExceeded" | "SlowDown" | "AccessDenied" | "InvalidRequest")),
+                "unexpected over-quota error: {error:?}"
+            );
+        }
+    }
+    Ok(())
+}

crates/e2e_test/src/distributed/s3_basic_test.rs

@@ -0,0 +1,111 @@
+// Copyright 2026 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 super::harness::{DistCluster, DistLayout, TestResult, assert_object_bytes, get_object_bytes, put_object, unique_bucket};
+use crate::common::{init_logging, local_http_client};
+use aws_sdk_s3::presigning::PresigningConfig;
+use aws_sdk_s3::types::{Delete, MetadataDirective, ObjectIdentifier};
+use std::time::Duration;
+
+#[tokio::test]
+async fn four_node_four_drive_s3_put_get_head_list_copy_rename_delete_and_presign() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let bucket = unique_bucket("s3basic");
+    dist.create_bucket(&bucket).await?;
+
+    let writer = dist.client(0)?;
+    let reader = dist.client(3)?;
+    let key = "dir/object.bin";
+    let body = vec![0xA5u8; 256 * 1024];
+    put_object(&writer, &bucket, key, body.clone()).await?;
+
+    let head = reader.head_object().bucket(&bucket).key(key).send().await?;
+    assert_eq!(head.content_length(), Some(body.len() as i64));
+    assert_object_bytes(&reader, &bucket, key, &body).await?;
+
+    let ranged = reader
+        .get_object()
+        .bucket(&bucket)
+        .key(key)
+        .range("bytes=0-15")
+        .send()
+        .await?;
+    let ranged_body = ranged.body.collect().await?.into_bytes();
+    assert_eq!(ranged_body.as_ref(), &body[..16]);
+
+    let listed = reader.list_objects_v2().bucket(&bucket).prefix("dir/").send().await?;
+    let keys: Vec<_> = listed.contents().iter().filter_map(|object| object.key()).collect();
+    assert_eq!(keys, vec![key]);
+
+    let copy_key = "dir/object-copy.bin";
+    reader
+        .copy_object()
+        .bucket(&bucket)
+        .key(copy_key)
+        .copy_source(format!("{bucket}/{key}"))
+        .metadata_directive(MetadataDirective::Copy)
+        .send()
+        .await?;
+    assert_object_bytes(&writer, &bucket, copy_key, &body).await?;
+
+    let moved_key = "dir/object-moved.bin";
+    writer
+        .copy_object()
+        .bucket(&bucket)
+        .key(moved_key)
+        .copy_source(format!("{bucket}/{copy_key}"))
+        .send()
+        .await?;
+    writer.delete_object().bucket(&bucket).key(copy_key).send().await?;
+    match writer.head_object().bucket(&bucket).key(copy_key).send().await {
+        Ok(_) => return Err("copied source still present after rename delete".into()),
+        Err(error) if error.as_service_error().is_some_and(|err| err.is_not_found()) => {}
+        Err(error) => return Err(error.into()),
+    }
+    assert_object_bytes(&reader, &bucket, moved_key, &body).await?;
+
+    let presigned = writer
+        .get_object()
+        .bucket(&bucket)
+        .key(key)
+        .presigned(PresigningConfig::expires_in(Duration::from_secs(120))?)
+        .await?;
+    let response = local_http_client().get(presigned.uri().to_string()).send().await?;
+    assert!(response.status().is_success(), "presigned GET failed: {}", response.status());
+    let presigned_body = response.bytes().await?;
+    assert_eq!(presigned_body.as_ref(), body.as_slice());
+
+    let empty_key = "empty";
+    put_object(&writer, &bucket, empty_key, Vec::new()).await?;
+    let empty = get_object_bytes(&reader, &bucket, empty_key).await?;
+    assert!(empty.is_empty());
+
+    writer
+        .delete_objects()
+        .bucket(&bucket)
+        .delete(
+            Delete::builder()
+                .objects(ObjectIdentifier::builder().key(key).build()?)
+                .objects(ObjectIdentifier::builder().key(moved_key).build()?)
+                .objects(ObjectIdentifier::builder().key(empty_key).build()?)
+                .build()?,
+        )
+        .send()
+        .await?;
+
+    let remaining = reader.list_objects_v2().bucket(&bucket).send().await?;
+    assert!(remaining.contents().is_empty(), "bucket still has objects after delete");
+    Ok(())
+}

crates/e2e_test/src/distributed/s3_during_data_movement_test.rs

@@ -0,0 +1,80 @@
+// Copyright 2026 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/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 super::harness::{
+    DistCluster, DistLayout, TestResult, assert_inventory, put_inventory, retrying_get_equals, retrying_put, start_decommission,
+    start_rebalance, unique_bucket, wait_for_decommission_complete,
+};
+use crate::common::init_logging;
+use std::time::Duration;
+
+#[tokio::test]
+async fn s3_put_get_list_succeed_during_decommission_and_rebalance() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourPoolFourDrive).await?;
+    let bucket = unique_bucket("s3move");
+    dist.create_bucket(&bucket).await?;
+    let client = dist.client(0)?;
+    let inventory = put_inventory(&client, &bucket, 8, 16 * 1024).await?;
+
+    start_decommission(&dist.cluster, 0).await?;
+    let live = dist.client(2)?;
+    retrying_put(
+        &live,
+        &bucket,
+        "during-decommission.bin",
+        b"written-while-decommissioning".to_vec(),
+        Duration::from_secs(30),
+    )
+    .await?;
+    retrying_get_equals(
+        &live,
+        &bucket,
+        "during-decommission.bin",
+        b"written-while-decommissioning",
+        Duration::from_secs(30),
+    )
+    .await?;
+    let listed = live.list_objects_v2().bucket(&bucket).send().await?;
+    assert!(
+        listed
+            .contents()
+            .iter()
+            .any(|object| object.key() == Some("during-decommission.bin")),
+        "list during decommission missed the newly written key"
+    );
+
+    wait_for_decommission_complete(&dist.cluster, 0, Duration::from_secs(180)).await?;
+    assert_inventory(&live, &bucket, &inventory).await?;
+
+    let _ = start_rebalance(&dist.cluster).await;
+    retrying_put(
+        &live,
+        &bucket,
+        "during-rebalance.bin",
+        b"written-while-rebalancing".to_vec(),
+        Duration::from_secs(30),
+    )
+    .await?;
+    retrying_get_equals(
+        &live,
+        &bucket,
+        "during-rebalance.bin",
+        b"written-while-rebalancing",
+        Duration::from_secs(30),
+    )
+    .await?;
+    assert_inventory(&dist.client(1)?, &bucket, &inventory).await?;
+    Ok(())
+}

crates/e2e_test/src/distributed/site_replication_test.rs

@@ -0,0 +1,87 @@
+// Copyright 2026 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/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 super::harness::{
+    DistCluster, TestResult, cluster_admin_ok, enable_versioning, put_object, unique_bucket, wait_for_replicated_bytes,
+};
+use crate::common::{init_logging, signed_request};
+use http::{Method, StatusCode};
+use rustfs_madmin::PeerSite;
+use std::time::Duration;
+
+async fn site_replication_add(cluster: &crate::common::RustFSTestClusterEnvironment, sites: &[PeerSite]) -> TestResult<String> {
+    let url = format!("{}/rustfs/admin/v3/site-replication/add?replicateILMExpiry=false", cluster.nodes[0].url);
+    let response = signed_request(
+        Method::PUT,
+        &url,
+        &cluster.access_key,
+        &cluster.secret_key,
+        Some(serde_json::to_vec(sites)?),
+        Some("application/json"),
+    )
+    .await?;
+    if response.status() != StatusCode::OK {
+        let status = response.status();
+        let body = response.text().await.unwrap_or_default();
+        return Err(format!("site replication add failed: {status} {body}").into());
+    }
+    Ok(response.text().await?)
+}
+
+#[tokio::test]
+async fn four_node_site_replication_replicates_object_to_peer_site() -> TestResult {
+    init_logging();
+    let (site_a, site_b) = DistCluster::start_replication_pair().await?;
+    let bucket = unique_bucket("siterepl");
+    site_a.create_bucket(&bucket).await?;
+    site_b.create_bucket(&bucket).await?;
+
+    let client_a = site_a.client(0)?;
+    let client_b = site_b.client(0)?;
+    enable_versioning(&client_a, &bucket).await?;
+    enable_versioning(&client_b, &bucket).await?;
+
+    let sites = vec![
+        PeerSite {
+            name: "site-a".to_string(),
+            endpoint: site_a.cluster.nodes[0].url.clone(),
+            access_key: site_a.cluster.access_key.clone(),
+            secret_key: site_a.cluster.secret_key.clone(),
+            ..Default::default()
+        },
+        PeerSite {
+            name: "site-b".to_string(),
+            endpoint: site_b.cluster.nodes[0].url.clone(),
+            access_key: site_b.cluster.access_key.clone(),
+            secret_key: site_b.cluster.secret_key.clone(),
+            ..Default::default()
+        },
+    ];
+    site_replication_add(&site_a.cluster, &sites).await?;
+
+    let info = cluster_admin_ok(&site_a.cluster, Method::GET, "/rustfs/admin/v3/site-replication/info", None).await?;
+    assert!(
+        info.contains("site-a") || info.contains("enabled") || info.contains("true"),
+        "site replication info did not show a configured peer: {info}"
+    );
+
+    let key = "site-object.bin";
+    let body = b"four-node-site-replication".to_vec();
+    put_object(&client_a, &bucket, key, body.clone()).await?;
+    wait_for_replicated_bytes(&client_b, &bucket, key, &body, Duration::from_secs(60)).await?;
+
+    let peer_b = site_b.client(3)?;
+    wait_for_replicated_bytes(&peer_b, &bucket, key, &body, Duration::from_secs(20)).await?;
+    Ok(())
+}

crates/e2e_test/src/distributed/versioning_test.rs

@@ -0,0 +1,88 @@
+// Copyright 2026 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/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 super::harness::{DistCluster, DistLayout, TestResult, enable_versioning, get_object_bytes, put_object, unique_bucket};
+use crate::common::init_logging;
+use aws_sdk_s3::error::ProvideErrorMetadata;
+
+#[tokio::test]
+async fn four_node_four_drive_versioning_put_list_get_delete_marker() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let bucket = unique_bucket("version");
+    dist.create_bucket(&bucket).await?;
+    let writer = dist.client(0)?;
+    let reader = dist.client(3)?;
+    enable_versioning(&writer, &bucket).await?;
+
+    let key = "versioned.txt";
+    put_object(&writer, &bucket, key, b"v1".to_vec()).await?;
+    put_object(&writer, &bucket, key, b"v2".to_vec()).await?;
+
+    let versions = reader.list_object_versions().bucket(&bucket).prefix(key).send().await?;
+    let version_ids: Vec<String> = versions
+        .versions()
+        .iter()
+        .filter_map(|version| version.version_id().map(str::to_string))
+        .collect();
+    assert!(version_ids.len() >= 2, "expected at least two versions, got {version_ids:?}");
+
+    let latest = get_object_bytes(&reader, &bucket, key).await?;
+    assert_eq!(latest, b"v2");
+
+    let older_id = versions
+        .versions()
+        .iter()
+        .find(|version| version.is_latest() != Some(true))
+        .and_then(|version| version.version_id())
+        .ok_or("missing non-latest version id")?;
+    let older = reader
+        .get_object()
+        .bucket(&bucket)
+        .key(key)
+        .version_id(older_id)
+        .send()
+        .await?;
+    let older_body = older.body.collect().await?.into_bytes();
+    assert_eq!(older_body.as_ref(), b"v1");
+
+    writer.delete_object().bucket(&bucket).key(key).send().await?;
+    let after_delete = reader.list_object_versions().bucket(&bucket).prefix(key).send().await?;
+    assert!(
+        !after_delete.delete_markers().is_empty(),
+        "delete marker missing after unversioned-style delete: {after_delete:?}"
+    );
+
+    let latest_after_delete = reader.get_object().bucket(&bucket).key(key).send().await;
+    match latest_after_delete {
+        Ok(_) => return Err("current version should be a delete marker".into()),
+        Err(error)
+            if error
+                .as_service_error()
+                .and_then(ProvideErrorMetadata::code)
+                .is_some_and(|code| code == "NoSuchKey" || code == "NotFound") => {}
+        Err(error) => return Err(error.into()),
+    }
+
+    let restored = reader
+        .get_object()
+        .bucket(&bucket)
+        .key(key)
+        .version_id(older_id)
+        .send()
+        .await?;
+    let restored_body = restored.body.collect().await?.into_bytes();
+    assert_eq!(restored_body.as_ref(), b"v1");
+    Ok(())
+}

crates/e2e_test/src/lib.rs

@@ -378,6 +378,11 @@ mod bucket_stats_regression_test;
 #[cfg(test)]
 mod distributed_startup_regression_test;
 
+// 4-node / 4-disk distributed Actions suite (S3, lock, versioning, replication,
+// quota, observability, expand/decommission/rebalance, site replication, chaos).
+#[cfg(test)]
+mod distributed;
+
 // P1 regression: tier/ILM transition (rustfs#5218, #5130, #5011, #4826, #5024)
 #[cfg(test)]
 mod tier_transition_regression_test;