test(e2e): require quota admission on over-limit 4x4 PUT

Do not treat SlowDown or AccessDenied as a hard-quota rejection.
Match InvalidRequest/QuotaExceeded only with a Bucket quota exceeded
message, matching the single-node quota e2e contract.

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

.config/e2e-distributed-selection.txt

@@ -0,0 +1,2 @@
+sha256-linux=4988bad7f5929152e0744f07393bc5d24aba2726e5daafa0eeca9a8c6a1f5683
+sha256-darwin=4988bad7f5929152e0744f07393bc5d24aba2726e5daafa0eeca9a8c6a1f5683

.config/nextest.toml

@@ -183,6 +183,13 @@ test-group = 'e2e-reliability'
 filter = 'package(e2e_test) & test(/^inline_fast_path_cluster_test::/)'
 test-group = 'e2e-inline-boundaries'
 
+# 4-node 4-drive distributed Actions suite: each case starts four rustfs
+# processes and up to sixteen data directories. Serialize across nextest's
+# process boundary so several 4x4 clusters never overlap.
+[[profile.default.overrides]]
+filter = 'package(e2e_test) & test(/^distributed::/)'
+test-group = 'e2e-cluster-nightly'
+
 # Vault KMS tests share the fixed dev-server port 8200. serial_test's #[serial]
 # does not cross nextest process boundaries, so keep every Vault-backed test in
 # one group.
@@ -526,6 +533,26 @@ path = "junit.xml"
 filter = 'package(e2e_test)'
 test-group = 'e2e-cluster-nightly'
 
+# ---------------------------------------------------------------------------
+# e2e-distributed profile — 4-node 4-disk Actions suite
+# ---------------------------------------------------------------------------
+# Nightly / dispatch lane owned by .github/workflows/e2e-distributed.yml.
+# Each case starts four rustfs processes (and for site replication, two
+# clusters). Upgrade cases also require RUSTFS_UPGRADE_SOURCE_BINARY.
+# Serialized via e2e-cluster-nightly. Not a PR merge gate.
+[profile.e2e-distributed]
+default-filter = 'package(e2e_test) & test(/^distributed::/)'
+fail-fast = false
+# Decommission / rebalance cases poll for up to 180s with little stdout.
+slow-timeout = { period = "120s", terminate-after = 6 }
+
+[profile.e2e-distributed.junit]
+path = "junit.xml"
+
+[[profile.e2e-distributed.overrides]]
+filter = 'package(e2e_test)'
+test-group = 'e2e-cluster-nightly'
+
 # ---------------------------------------------------------------------------
 # e2e-odm-interop profile — on-demand migration provider interop lane (ODM-20)
 # ---------------------------------------------------------------------------
@@ -586,6 +613,10 @@ path = "junit.xml"
 #     cluster-fault lane. heal_erasure_disk_rebuild is intentionally not
 #     excluded here because backlog#2213 promotes core heal rebuild coverage to
 #     this merge/main lane while retaining nightly coverage.
+#   * distributed:: — 4-node 4-disk Actions suite (S3, lock, versioning,
+#     replication, quota, observability, expand/decommission/rebalance, site
+#     replication, chaos, upgrade history/IAM). Owns [profile.e2e-distributed] and
+#     .github/workflows/e2e-distributed.yml.
 #   * on_demand_migration::interop_test — the ODM-20 provider interoperability
 #     cases, which are meaningless without a source: they run in the dedicated
 #     [profile.e2e-odm-interop] lane below, where the workflow points them at a
@@ -607,6 +638,7 @@ default-filter = """
     package(e2e_test)
     & !test(/^protocols::/)
     & !test(/^(admin_timeout_regression_test|cluster_concurrency_test|cluster_multidrive_pool_test|degraded_listing_availability_test|namespace_lock_quorum_test|object_lambda_test|stale_multipart_cleanup_cluster_test)::/)
+    & !test(/^distributed::/)
     & !test(/^replication_extension_test::/)
     & !test(/^replication_target_matrix_test::/)
     & !test(/^on_demand_migration::(concurrency_test|fault_test|interop_test|real_source_test)::/)

.github/scheduled-validations.json

@@ -4,6 +4,11 @@
   { "workflow": ".github/workflows/ci.yml", "max_age_hours": 192 },
   { "workflow": ".github/workflows/coverage.yml", "max_age_hours": 192 },
   { "workflow": ".github/workflows/e2e-replication-nightly.yml", "max_age_hours": 36 },
+  {
+    "workflow": ".github/workflows/e2e-distributed.yml",
+    "max_age_hours": 36,
+    "never_ran_grace_until": "2026-09-18T00:00:00Z"
+  },
   { "workflow": ".github/workflows/e2e-s3tests.yml", "max_age_hours": 192 },
   { "workflow": ".github/workflows/fuzz.yml", "max_age_hours": 36 },
   { "workflow": ".github/workflows/mint.yml", "max_age_hours": 192 },

.github/workflows/e2e-distributed.yml

@@ -0,0 +1,140 @@
+# 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/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-disk distributed e2e lane.
+#
+# Each selected test starts a real localhost cluster via
+# `RustFSTestClusterEnvironment` (4 processes; 4 drives per node unless the
+# case is a two-site 4-node 1-drive pair or a 4-node upgrade). Membership is
+# `[profile.e2e-distributed]` in `.config/nextest.toml`. This is not a required
+# merge check: it is the scheduled/dispatch counterpart to the hardware
+# functional chain that currently clones rustfs/auto-testing onto three VMs.
+# Upgrade cases download the same pinned previous release as e2e-upgrade.yml.
+
+name: e2e-distributed
+
+on:
+  workflow_dispatch:
+    inputs:
+      filter:
+        description: "Optional nextest -E filter (default: the whole e2e-distributed profile)"
+        required: false
+        default: ""
+  schedule:
+    # 05:53 UTC nightly — clear of e2e-nightly (04:29) and ODM interop (05:23).
+    - cron: "53 5 * * *"
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: ${{ github.event_name != 'schedule' }}
+
+jobs:
+  distributed:
+    name: Distributed 4-node 4-disk e2e
+    runs-on: sm-standard-4
+    timeout-minutes: 180
+    env:
+      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+      NO_PROXY: 127.0.0.1,localhost
+      HTTP_PROXY: ""
+      HTTPS_PROXY: ""
+      # Pinned previous release used by distributed::upgrade_test (same pin as e2e-upgrade.yml).
+      UPGRADE_SOURCE_VERSION: 1.0.0-rc.2
+      UPGRADE_SOURCE_ASSET: rustfs-linux-x86_64-gnu-v1.0.0-rc.2.zip
+      UPGRADE_SOURCE_SHA256: 7c789386bf85278f865b8e0d359bf4edb84d5aa408cc3fa54a18c25ca74cd6e7
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+
+      - name: Setup Rust environment
+        uses: ./.github/actions/setup
+        with:
+          rust-version: stable
+          cache-shared-key: ci-e2e-distributed
+          cache-save-if: ${{ github.ref == 'refs/heads/main' }}
+          install-build-packaging-tools: 'false'
+
+      - name: Download pinned previous release
+        env:
+          SOURCE_DIR: ${{ runner.temp }}/rustfs-upgrade-source
+        run: |
+          set -euo pipefail
+          mkdir -p "$SOURCE_DIR"
+          archive="$SOURCE_DIR/$UPGRADE_SOURCE_ASSET"
+          curl --fail --location --retry 3 --output "$archive" \
+            "https://github.com/${GITHUB_REPOSITORY}/releases/download/${UPGRADE_SOURCE_VERSION}/${UPGRADE_SOURCE_ASSET}"
+          echo "$UPGRADE_SOURCE_SHA256  $archive" | sha256sum --check --strict
+          unzip -q "$archive" -d "$SOURCE_DIR"
+          chmod +x "$SOURCE_DIR/rustfs"
+          test -x "$SOURCE_DIR/rustfs"
+          echo "RUSTFS_UPGRADE_SOURCE_BINARY=$SOURCE_DIR/rustfs" >> "$GITHUB_ENV"
+
+      - name: Build rustfs binary
+        run: |
+          cargo build -p rustfs --bins
+          : > target/debug/rustfs.features
+
+      - name: Verify distributed e2e membership
+        env:
+          NEXTEST_LISTING: ${{ runner.temp }}/rustfs-e2e-distributed-list.json
+        run: |
+          cargo nextest list --profile e2e-distributed -p e2e_test --message-format json > "${NEXTEST_LISTING}"
+          python3 ./scripts/check_test_wiring.py --check-profile e2e-distributed "${NEXTEST_LISTING}"
+
+      - name: Run distributed 4-node e2e suite
+        env:
+          RUSTFS_E2E_LOG_DIR: ${{ runner.temp }}/rustfs-e2e-distributed-logs
+          NEXTEST_FILTER: ${{ github.event.inputs.filter }}
+        run: |
+          set -euo pipefail
+          if [ -n "${NEXTEST_FILTER}" ]; then
+            cargo nextest run --profile e2e-distributed -p e2e_test -E "${NEXTEST_FILTER}" --no-tests=fail
+          else
+            cargo nextest run --profile e2e-distributed -p e2e_test --no-tests=fail
+          fi
+
+      - name: Upload distributed e2e diagnostics
+        if: always()
+        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
+        with:
+          name: e2e-distributed-${{ github.run_number }}
+          path: |
+            target/nextest/e2e-distributed/junit.xml
+            ${{ runner.temp }}/rustfs-e2e-distributed-list.json
+            ${{ runner.temp }}/rustfs-e2e-distributed-logs/
+          retention-days: 7
+          if-no-files-found: warn
+
+  alert-on-failure:
+    name: Alert on scheduled failure
+    needs: [distributed]
+    if: always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure')
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: read
+      issues: write
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+      - name: Open or update failure-tracking issue
+        uses: ./.github/actions/schedule-failure-issue
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/scheduled-validation-watchdog.yml

@@ -22,6 +22,7 @@ on:
       - "Continuous Integration"
       - "coverage"
       - "e2e-nightly"
+      - "e2e-distributed"
       - "e2e-s3tests"
       - "Fuzz"
       - "mint"

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, 4-node upgrade of historical data and IAM AK/SK. 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,9 @@ 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 / upgrade)
+# Upgrade cases need RUSTFS_UPGRADE_SOURCE_BINARY; without it they fail closed.
+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

@@ -1483,8 +1483,9 @@ impl RustFSTestClusterEnvironment {
             self.spawn_node(node_idx, binary_path, &volumes_arg)?;
         }
 
-        for (i, node) in self.nodes.iter().enumerate() {
-            self.wait_for_node_ready(&node.address, i).await?;
+        for i in 0..self.nodes.len() {
+            let address = self.nodes[i].address.clone();
+            self.wait_for_node_ready(&address, i).await?;
         }
 
         for node_idx in 0..self.nodes.len() {
@@ -1510,7 +1511,8 @@ impl RustFSTestClusterEnvironment {
         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?;
+        let address = self.nodes[node_idx].address.clone();
+        self.wait_for_node_ready(&address, node_idx).await?;
         self.wait_for_node_service_ready(node_idx).await?;
         Ok(())
     }
@@ -1559,8 +1561,18 @@ impl RustFSTestClusterEnvironment {
     ///
     /// 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<dyn std::error::Error + Send + Sync>> {
+    fn node_process_exited(&mut self, idx: usize) -> Result<bool, Box<dyn std::error::Error + Send + Sync>> {
+        let Some(process) = self.nodes.get_mut(idx).and_then(|node| node.process.as_mut()) else {
+            return Ok(true);
+        };
+        Ok(process.try_wait()?.is_some())
+    }
+
+    async fn wait_for_node_ready(&mut self, address: &str, idx: usize) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         for attempt in 0..60 {
+            if self.node_process_exited(idx)? {
+                return Err(format!("cluster node {idx} process exited before TCP ready").into());
+            }
             if TcpStream::connect(address).await.is_ok() {
                 info!("Node {} ({}) TCP ready after {} attempts", idx, address, attempt + 1);
                 return Ok(());
@@ -1574,10 +1586,13 @@ impl RustFSTestClusterEnvironment {
     ///
     /// Verifies service availability by calling the S3 `list_buckets` API against the requested node,
     /// retries up to 120 times with a 1-second interval between attempts.
-    async fn wait_for_node_service_ready(&self, node_idx: usize) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    async fn wait_for_node_service_ready(&mut self, node_idx: usize) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         let client = self.create_s3_client(node_idx)?;
 
         for attempt in 0..120 {
+            if self.node_process_exited(node_idx)? {
+                return Err(format!("cluster node {node_idx} process exited before S3 ready").into());
+            }
             match client.list_buckets().send().await {
                 Ok(_) => {
                     info!("Cluster node {} service ready after {} attempts", node_idx, attempt + 1);
@@ -1700,6 +1715,64 @@ 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);
+
+        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,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, bring_drive_online, put_object, retrying_get_equals,
+    take_drive_offline, unique_bucket, wait_for_ready,
+};
+use crate::common::init_logging;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::Barrier;
+
+#[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 concurrent_gets_survive_peer_node_kill() -> TestResult {
+    init_logging();
+    let mut dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let bucket = unique_bucket("getkill");
+    dist.create_bucket(&bucket).await?;
+    let body = vec![0x7Au8; 96 * 1024];
+    put_object(&dist.client(0)?, &bucket, "steady.bin", body.clone()).await?;
+
+    let live: Vec<_> = (0..3).map(|idx| dist.client(idx)).collect::<Result<Vec<_>, _>>()?;
+    let start = Arc::new(Barrier::new(13));
+    let mut handles = Vec::new();
+    for idx in 0..12 {
+        let client = live[idx % live.len()].clone();
+        let bucket = bucket.clone();
+        let body = body.clone();
+        let start = start.clone();
+        handles.push(tokio::spawn(async move {
+            start.wait().await;
+            retrying_get_equals(&client, &bucket, "steady.bin", &body, Duration::from_secs(20)).await
+        }));
+    }
+    start.wait().await;
+    dist.cluster.stop_node(3)?;
+    for handle in handles {
+        handle.await??;
+    }
+
+    dist.cluster.start_node(3).await?;
+    wait_for_ready(&dist.cluster).await?;
+    assert_object_bytes(&dist.client(3)?, &bucket, "steady.bin", &body).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,67 @@
+// 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, decommission_started_or_refused, payload_for, put_inventory_retrying,
+    retrying_get_equals, retrying_put, 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::FourByFour).await?;
+    let bucket = unique_bucket("concdecom");
+    dist.create_bucket(&bucket).await?;
+    let baseline_client = dist.client(0)?;
+    let inventory = put_inventory_retrying(&baseline_client, &bucket, 10, 24 * 1024, Duration::from_secs(30)).await?;
+
+    let decommission_started = decommission_started_or_refused(&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??);
+    }
+
+    if decommission_started {
+        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,44 @@
+// 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, decommission_started_or_refused, put_inventory_retrying, sha256_hex,
+    unique_bucket, wait_for_decommission_complete,
+};
+use crate::common::init_logging;
+use std::time::Duration;
+
+#[tokio::test]
+async fn decommission_attempt_does_not_alter_object_sha256() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let bucket = unique_bucket("integrity");
+    dist.create_bucket(&bucket).await?;
+    let client = dist.client(0)?;
+    let inventory = put_inventory_retrying(&client, &bucket, 20, 64 * 1024, Duration::from_secs(30)).await?;
+    let before: Vec<(String, String)> = inventory.iter().map(|(key, body)| (key.clone(), sha256_hex(body))).collect();
+
+    if decommission_started_or_refused(&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,71 @@
+// 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, decommission_started_or_refused, list_pools_json, put_inventory,
+    rebalance_started_or_refused, 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_four_drive_restart_preserves_objects_then_rebalance_attempt() -> TestResult {
+    init_logging();
+    let mut dist = DistCluster::start(DistLayout::FourByFour).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.start().await?;
+
+    let after_restart = dist.client(0)?;
+    assert_inventory(&after_restart, &bucket, &inventory).await?;
+    let peer = dist.client(3)?;
+    assert_inventory(&peer, &bucket, &inventory).await?;
+
+    if rebalance_started_or_refused(&dist.cluster).await? {
+        wait_for_rebalance_idle(&dist.cluster, Duration::from_secs(90)).await?;
+    }
+    assert_inventory(&peer, &bucket, &inventory).await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn four_node_four_drive_decommission_attempt_does_not_lose_objects() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourByFour).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(1);
+    assert!(pool_count >= 1, "expected at least one pool before decommission: {pools_before}");
+
+    if decommission_started_or_refused(&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,149 @@
+// 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, wait_until,
+};
+use crate::common::init_logging;
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
+use std::time::Duration;
+
+#[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(())
+}
+
+#[tokio::test]
+async fn four_node_list_buckets_agree_across_all_nodes() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let bucket = unique_bucket("listed");
+    dist.create_bucket(&bucket).await?;
+    put_object(&dist.client(0)?, &bucket, "seed.bin", b"seed".to_vec()).await?;
+
+    for node_idx in 0..dist.cluster.nodes.len() {
+        let client = dist.client(node_idx)?;
+        let name = bucket.clone();
+        wait_until(
+            Duration::from_secs(20),
+            || {
+                let client = client.clone();
+                let name = name.clone();
+                async move {
+                    let listed = client.list_buckets().send().await?;
+                    Ok(listed.buckets().iter().any(|entry| entry.name() == Some(name.as_str())))
+                }
+            },
+            &format!("node {node_idx} lists {bucket}"),
+        )
+        .await?;
+        wait_until(
+            Duration::from_secs(20),
+            || {
+                let client = dist.client(node_idx).expect("client");
+                let name = bucket.clone();
+                async move { Ok(get_object_bytes(&client, &name, "seed.bin").await.ok() == Some(b"seed".to_vec())) }
+            },
+            &format!("node {node_idx} reads seed.bin"),
+        )
+        .await?;
+    }
+    Ok(())
+}

crates/e2e_test/src/distributed/harness.rs

@@ -0,0 +1,933 @@
+// 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** — `append_single_node_pool` exists for harness unit
+//!   tests. Live expand-then-restart currently hits `pool metadata recovery
+//!   required` on localhost DistErasure. That is a production bootstrap-proof
+//!   limitation this test lane does not change. Movement tests use 4×4 single
+//!   pool and classify decommission/rebalance product refusals (and opaque
+//!   500 InternalError) as a refused move while still asserting object bytes.
+//!
+//! 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, build_test_s3_config,
+    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,
+}
+
+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 mut dist = Self::new_stopped_with_env(layout, extra_env).await?;
+        dist.cluster.start().await?;
+        Ok(dist)
+    }
+
+    /// Allocate ports and data dirs without spawning processes.
+    ///
+    /// Upgrade tests configure capture logs, then start a pinned previous
+    /// binary against the same directories.
+    pub async fn new_stopped(layout: DistLayout) -> TestResult<Self> {
+        Self::new_stopped_with_env(layout, &[]).await
+    }
+
+    pub async fn new_stopped_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),
+        };
+        let mut cluster = RustFSTestClusterEnvironment::with_topology(topology).await?;
+        cluster.set_env("NO_PROXY", "127.0.0.1,localhost");
+        cluster.set_env("HTTP_PROXY", "");
+        cluster.set_env("HTTPS_PROXY", "");
+        for &(key, value) in extra_env {
+            cluster.set_env(key, value);
+        }
+        Ok(Self { cluster })
+    }
+
+    /// Start every node with a specific `rustfs` binary, keeping the allocated
+    /// data directories. Used to seed an old on-disk format before upgrading.
+    pub async fn start_from_binary(&mut self, binary: &Path) -> TestResult {
+        self.cluster.start_with_binary(binary).await?;
+        wait_for_ready(&self.cluster).await?;
+        Ok(())
+    }
+
+    /// Stop every node and bring the same data directories up on the workspace
+    /// binary (direct upgrade).
+    pub async fn restart_with_current_binary(&mut self) -> TestResult {
+        self.cluster.stop();
+        self.cluster.start().await?;
+        wait_for_ready(&self.cluster).await?;
+        Ok(())
+    }
+
+    /// Replace one running node with the workspace binary (rolling upgrade).
+    pub async fn replace_node_with_current_binary(&mut self, node_idx: usize) -> TestResult {
+        self.cluster.stop_node(node_idx)?;
+        self.cluster.start_node(node_idx).await?;
+        wait_for_ready(&self.cluster).await?;
+        Ok(())
+    }
+
+    pub fn client_with_credentials(&self, node_idx: usize, access_key: &str, secret_key: &str) -> TestResult<Client> {
+        if node_idx >= self.cluster.nodes.len() {
+            return Err("node_idx is invalid".into());
+        }
+        Ok(Client::from_conf(build_test_s3_config(
+            &self.cluster.nodes[node_idx].url,
+            access_key,
+            secret_key,
+            None,
+            "cluster-iam-test",
+        )))
+    }
+
+    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)
+}
+
+/// Localhost DistErasure can 500 a PUT while heal_bucket hits a pool-meta
+/// write fence. Retry only those transient codes.
+pub(crate) async fn put_inventory_retrying(
+    client: &Client,
+    bucket: &str,
+    count: usize,
+    size: usize,
+    timeout: Duration,
+) -> 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);
+        retrying_put(client, bucket, &key, body.clone(), timeout).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
+}
+
+/// Localhost DistErasure can boot and serve S3 while refusing pool.bin
+/// mutations (`pool metadata writes remain blocked` / missing fleet
+/// capability proof). Single-pool 4×4 also rejects decommission/rebalance
+/// with a product error. Tests must not pretend a move ran.
+pub(crate) fn is_pool_meta_write_fence(body: &str) -> bool {
+    body.contains("pool metadata writes remain blocked")
+        || body.contains("pool metadata recovery required")
+        || body.contains("pool activation requires a live fleet capability proof")
+        || body.contains("pool activation fleet capability proof expired")
+        || body.contains("live fleet capability proof")
+}
+
+/// Product refusals that movement tests observe. Opaque 500 InternalError stays
+/// in [`classify_data_movement_http`] because admin often wraps the fence as
+/// InternalError XML without the inner string. 502/503 and auth failures are
+/// not refusals.
+pub(crate) fn is_known_data_movement_refusal(body: &str) -> bool {
+    is_pool_meta_write_fence(body)
+        || body.contains("NotImplemented")
+        || body.contains("single pool deployments do not support")
+        || body.contains("at least one active pool must remain")
+}
+
+#[derive(Debug)]
+pub(crate) enum DataMovementStart {
+    Started,
+    Refused(String),
+}
+
+pub(crate) fn classify_data_movement_http(status: StatusCode, body: &str) -> Result<DataMovementStart, String> {
+    if status.is_success() {
+        return Ok(DataMovementStart::Started);
+    }
+    if is_known_data_movement_refusal(body) || status.as_u16() == 501 || status == StatusCode::INTERNAL_SERVER_ERROR {
+        return Ok(DataMovementStart::Refused(format!("{status} {body}")));
+    }
+    Err(format!("{status} {body}"))
+}
+
+pub(crate) async fn try_start_decommission(
+    cluster: &RustFSTestClusterEnvironment,
+    pool_id: usize,
+) -> TestResult<DataMovementStart> {
+    let path = format!("/rustfs/admin/v3/pools/decommission?pool={pool_id}&by-id=true");
+    let (status, response) = cluster_admin(cluster, Method::POST, &path, None).await?;
+    classify_data_movement_http(status, &response).map_err(|detail| format!("POST {path} failed: {detail}").into())
+}
+
+/// Returns whether decommission actually started. A product refusal or opaque
+/// 500 InternalError is not a test failure: callers still assert object bytes.
+pub(crate) async fn decommission_started_or_refused(cluster: &RustFSTestClusterEnvironment, pool_id: usize) -> TestResult<bool> {
+    match try_start_decommission(cluster, pool_id).await? {
+        DataMovementStart::Started => Ok(true),
+        DataMovementStart::Refused(detail) => {
+            eprintln!("decommission POST refused; objects still asserted: {detail}");
+            Ok(false)
+        }
+    }
+}
+
+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)
+}
+
+/// `Ok(true)` complete, `Ok(false)` still running, `Err` terminal failure.
+pub(crate) fn decommission_progress(status: &serde_json::Value, pool_id: usize) -> Result<bool, String> {
+    if decommission_failed(status, pool_id) {
+        return Err(format!("decommission failed for pool {pool_id}: {status}"));
+    }
+    Ok(decommission_complete(status, pool_id))
+}
+
+pub(crate) async fn wait_for_decommission_complete(
+    cluster: &RustFSTestClusterEnvironment,
+    pool_id: usize,
+    timeout: Duration,
+) -> TestResult {
+    let deadline = Instant::now() + timeout;
+    let mut delay = Duration::from_millis(50);
+    let mut last_error;
+    loop {
+        last_error = match decommission_status_json(cluster).await {
+            Ok(status) => match decommission_progress(&status, pool_id) {
+                Ok(true) => return Ok(()),
+                Ok(false) => format!("decommission complete still false: {status}"),
+                Err(failed) => return Err(failed.into()),
+            },
+            Err(error) => error.to_string(),
+        };
+        if Instant::now() >= deadline {
+            return Err(format!("decommission complete did not become true within {timeout:?}: {last_error}").into());
+        }
+        sleep(delay).await;
+        delay = (delay * 2).min(Duration::from_secs(1));
+    }
+}
+
+pub(crate) async fn try_start_rebalance(cluster: &RustFSTestClusterEnvironment) -> TestResult<DataMovementStart> {
+    let path = "/rustfs/admin/v3/rebalance/start";
+    let (status, response) = cluster_admin(cluster, Method::POST, path, None).await?;
+    classify_data_movement_http(status, &response).map_err(|detail| format!("POST {path} failed: {detail}").into())
+}
+
+pub(crate) async fn rebalance_started_or_refused(cluster: &RustFSTestClusterEnvironment) -> TestResult<bool> {
+    match try_start_rebalance(cluster).await? {
+        DataMovementStart::Started => Ok(true),
+        DataMovementStart::Refused(detail) => {
+            eprintln!("rebalance POST refused; objects still asserted: {detail}");
+            Ok(false)
+        }
+    }
+}
+
+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(DRIVES_PER_NODE, 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(), DRIVES_PER_NODE);
+    let volumes = env.rustfs_volumes_arg();
+    assert_eq!(volumes.split(' ').count(), 3, "expected three pool arguments, got: {volumes}");
+    assert!(volumes.contains("/drive{0...3}"), "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}"
+    );
+}
+
+#[cfg(unix)]
+#[tokio::test]
+async fn cluster_start_fails_fast_when_node_process_exits() {
+    let mut dist = DistCluster::new_stopped(DistLayout::FourNodeFourDisk)
+        .await
+        .expect("stopped 4-node cluster");
+    let script = format!("{}/immediate-exit.sh", dist.cluster.temp_dir);
+    std::fs::write(&script, "#!/bin/sh\nexit 1\n").expect("write exit stub");
+    let mut perms = std::fs::metadata(&script).expect("stat exit stub").permissions();
+    std::os::unix::fs::PermissionsExt::set_mode(&mut perms, 0o755);
+    std::fs::set_permissions(&script, perms).expect("chmod exit stub");
+
+    let started = Instant::now();
+    let err = dist
+        .start_from_binary(Path::new(&script))
+        .await
+        .expect_err("a node that exits immediately must fail start");
+    let elapsed = started.elapsed();
+    let message = err.to_string();
+    assert!(
+        message.contains("exited before TCP ready") || message.contains("exited before S3 ready"),
+        "unexpected start error: {message}"
+    );
+    assert!(
+        elapsed < Duration::from_secs(30),
+        "cluster start must fail fast when a node exits, took {elapsed:?}"
+    );
+}
+
+#[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));
+    assert!(decommission_progress(&status, 0).expect("complete pool"));
+    assert!(!decommission_progress(&status, 1).expect("other pool is not complete"));
+}
+
+#[test]
+fn decommission_progress_fails_closed_on_failed_flag() {
+    let failed = serde_json::json!({
+        "pools": [{
+            "id": 0,
+            "status": "failed",
+            "decommissionInfo": { "complete": false, "failed": true, "canceled": false }
+        }]
+    });
+    let err = decommission_progress(&failed, 0).expect_err("failed decommission must not look complete");
+    assert!(err.contains("decommission failed for pool 0"), "{err}");
+    assert!(!decommission_progress(&failed, 1).expect("missing pool is still running"));
+}
+
+#[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));
+}
+
+#[test]
+fn pool_meta_write_fence_matches_known_product_gates() {
+    assert!(is_pool_meta_write_fence(
+        "heal_bucket: pool metadata writes remain blocked after a recovery-required replica state"
+    ));
+    assert!(is_pool_meta_write_fence(
+        "rebalance meta save failed: pool activation requires a live fleet capability proof"
+    ));
+    assert!(is_pool_meta_write_fence("pool metadata recovery required: no durable bootstrap identity"));
+    assert!(!is_pool_meta_write_fence("NotImplemented: single pool cannot decommission"));
+    assert!(!is_pool_meta_write_fence("AccessDenied"));
+}
+
+#[test]
+fn classify_data_movement_http_observes_product_refusals_not_auth_failures() {
+    assert!(matches!(classify_data_movement_http(StatusCode::OK, ""), Ok(DataMovementStart::Started)));
+    assert!(matches!(
+        classify_data_movement_http(
+            StatusCode::BAD_REQUEST,
+            "failed to start decommission: single pool deployments do not support decommission"
+        ),
+        Ok(DataMovementStart::Refused(_))
+    ));
+    assert!(matches!(
+        classify_data_movement_http(
+            StatusCode::BAD_REQUEST,
+            "failed to start decommission: at least one active pool must remain after decommission start"
+        ),
+        Ok(DataMovementStart::Refused(_))
+    ));
+    assert!(matches!(
+        classify_data_movement_http(StatusCode::NOT_IMPLEMENTED, "NotImplemented"),
+        Ok(DataMovementStart::Refused(_))
+    ));
+    assert!(matches!(
+        classify_data_movement_http(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "pool metadata writes remain blocked after a recovery-required replica state"
+        ),
+        Ok(DataMovementStart::Refused(_))
+    ));
+    assert!(matches!(
+        classify_data_movement_http(StatusCode::INTERNAL_SERVER_ERROR, "InternalError"),
+        Ok(DataMovementStart::Refused(_))
+    ));
+    let denied = classify_data_movement_http(StatusCode::FORBIDDEN, "AccessDenied").expect_err("auth failure is not a refusal");
+    assert!(denied.contains("AccessDenied"), "{denied}");
+    let unavailable = classify_data_movement_http(StatusCode::SERVICE_UNAVAILABLE, "ServiceUnavailable")
+        .expect_err("503 is not a product refusal");
+    assert!(unavailable.contains("ServiceUnavailable"), "{unavailable}");
+    let bad_gateway =
+        classify_data_movement_http(StatusCode::BAD_GATEWAY, "Bad Gateway").expect_err("502 is not a product refusal");
+    assert!(bad_gateway.contains("502") || bad_gateway.contains("Bad Gateway"), "{bad_gateway}");
+}

crates/e2e_test/src/distributed/mod.rs

@@ -0,0 +1,35 @@
+// 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 upgrade_test;
+mod versioning_test;

crates/e2e_test/src/distributed/object_lock_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, unique_bucket};
+use crate::common::init_logging;
+use crate::object_lock::common::{delete_object_with_bypass, put_object_with_legal_hold, put_object_with_retention};
+use aws_sdk_s3::Client;
+use aws_sdk_s3::error::ProvideErrorMetadata;
+use aws_sdk_s3::error::SdkError;
+use aws_sdk_s3::operation::delete_object::DeleteObjectError;
+use aws_sdk_s3::types::{ObjectLockLegalHoldStatus, ObjectLockRetentionMode};
+use chrono::{Duration as ChronoDuration, Utc};
+
+fn delete_denied(error: &SdkError<DeleteObjectError>, context: &str) -> TestResult {
+    let code = error.as_service_error().and_then(ProvideErrorMetadata::code);
+    if code == Some("AccessDenied") {
+        Ok(())
+    } else {
+        Err(format!("{context}: expected AccessDenied, got {error:?}").into())
+    }
+}
+
+async fn expect_versioned_delete_denied(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    version_id: &str,
+    bypass: bool,
+    context: &str,
+) -> TestResult {
+    match delete_object_with_bypass(client, bucket, key, Some(version_id), bypass).await {
+        Ok(_) => Err(format!("{context}: DeleteObject of retained version must be denied").into()),
+        Err(error) => delete_denied(error.as_ref(), context),
+    }
+}
+
+#[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 compliance_key = "compliance.bin";
+    let compliance_version = put_object_with_retention(
+        &client,
+        &bucket,
+        compliance_key,
+        b"locked-compliance",
+        ObjectLockRetentionMode::Compliance,
+        retain_until,
+    )
+    .await?;
+
+    // Unversioned DELETE is allowed: it only creates a delete marker. WORM
+    // applies to a specific version id.
+    let marker = peer.delete_object().bucket(&bucket).key(compliance_key).send().await?;
+    assert_eq!(
+        marker.delete_marker(),
+        Some(true),
+        "unversioned DELETE on a locked object must create a delete marker"
+    );
+
+    expect_versioned_delete_denied(&peer, &bucket, compliance_key, &compliance_version, false, "COMPLIANCE without bypass")
+        .await?;
+    expect_versioned_delete_denied(&peer, &bucket, compliance_key, &compliance_version, true, "COMPLIANCE with bypass").await?;
+
+    let governance_key = "governance.bin";
+    let governance_version = put_object_with_retention(
+        &client,
+        &bucket,
+        governance_key,
+        b"locked-governance",
+        ObjectLockRetentionMode::Governance,
+        retain_until,
+    )
+    .await?;
+
+    expect_versioned_delete_denied(&peer, &bucket, governance_key, &governance_version, false, "GOVERNANCE without bypass")
+        .await?;
+    delete_object_with_bypass(&peer, &bucket, governance_key, Some(&governance_version), true).await?;
+
+    let hold_key = "legal-hold.bin";
+    let hold_version =
+        put_object_with_legal_hold(&client, &bucket, hold_key, b"legal-hold", ObjectLockLegalHoldStatus::On).await?;
+    expect_versioned_delete_denied(&peer, &bucket, hold_key, &hold_version, false, "legal hold without bypass").await?;
+    expect_versioned_delete_denied(&peer, &bucket, hold_key, &hold_version, true, "legal hold with bypass").await?;
+
+    Ok(())
+}

crates/e2e_test/src/distributed/observability_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, cluster_admin, 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());
+    }
+
+    // Optional surfaces: 404/400/501 are acceptable (route missing or stubbed);
+    // unexpected 5xx is not. A 2xx body must be non-empty.
+    for path in [
+        "/rustfs/admin/v3/log/search",
+        "/rustfs/admin/v4/runtime/capabilities",
+        "/minio/v2/metrics/cluster",
+    ] {
+        let (status, body) = cluster_admin(&dist.cluster, Method::GET, path, None).await?;
+        assert!(
+            status.is_success() || status.is_client_error() || status.as_u16() == 501,
+            "observability path {path} returned {status}: {body}"
+        );
+        if status.is_success() {
+            assert!(!body.trim().is_empty(), "empty body from {path}");
+        }
+    }
+
+    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,144 @@
+// 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, retrying_put, set_bucket_quota,
+    set_remote_target, unique_bucket, wait_for_replicated_bytes, wait_until,
+};
+use crate::common::{FAST_DATA_USAGE_SCANNER_ENV, init_logging};
+use aws_sdk_s3::error::ProvideErrorMetadata;
+use http::Method;
+use std::time::Duration;
+
+/// `Ok(true)` quota admission rejected the PUT, `Ok(false)` retry, `Err` not quota.
+fn quota_over_limit_put_outcome(code: Option<&str>, message: Option<&str>) -> Result<bool, String> {
+    let quota_message = message.is_some_and(|text| text.starts_with("Bucket quota exceeded"));
+    match code {
+        Some("InvalidRequest" | "QuotaExceeded") if quota_message => Ok(true),
+        Some("SlowDown" | "ServiceUnavailable") => Ok(false),
+        Some("AccessDenied") => Err("AccessDenied is not a quota admission rejection".to_string()),
+        Some("InvalidRequest" | "QuotaExceeded") => {
+            Err(format!("InvalidRequest/QuotaExceeded without quota admission message: {message:?}"))
+        }
+        other => Err(format!("unexpected over-quota error code {other:?} message {message:?}")),
+    }
+}
+
+#[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)?;
+    retrying_put(&client, &bucket, "small.bin", vec![0u8; 1024], Duration::from_secs(30)).await?;
+    wait_until(
+        Duration::from_secs(30),
+        || async {
+            let (status, body) = super::harness::cluster_admin(
+                &dist.cluster,
+                Method::GET,
+                &format!("/rustfs/admin/v3/quota-stats/{bucket}"),
+                None,
+            )
+            .await?;
+            if !status.is_success() {
+                return Ok(false);
+            }
+            let stats: serde_json::Value = serde_json::from_str(&body).unwrap_or_default();
+            Ok(stats.get("current_usage").and_then(serde_json::Value::as_u64).unwrap_or(0) >= 1024)
+        },
+        "quota stats observe small object",
+    )
+    .await?;
+
+    let mut oversized_attempt = 0u32;
+    wait_until(
+        Duration::from_secs(30),
+        || {
+            oversized_attempt += 1;
+            let key = format!("too-big-{oversized_attempt}.bin");
+            let client = client.clone();
+            let bucket = bucket.clone();
+            async move {
+                match client
+                    .put_object()
+                    .bucket(&bucket)
+                    .key(key)
+                    .body(vec![0u8; 16 * 1024].into())
+                    .send()
+                    .await
+                {
+                    Ok(_) => Ok(false),
+                    Err(error) => {
+                        let code = error.as_service_error().and_then(ProvideErrorMetadata::code);
+                        let message = error.as_service_error().and_then(ProvideErrorMetadata::message);
+                        match quota_over_limit_put_outcome(code, message) {
+                            Ok(done) => Ok(done),
+                            Err(detail) => Err(format!("{detail}: {error:?}").into()),
+                        }
+                    }
+                }
+            }
+        },
+        "hard quota rejects oversized PUT",
+    )
+    .await?;
+    Ok(())
+}
+
+#[test]
+fn quota_over_limit_put_outcome_requires_quota_admission() {
+    assert_eq!(
+        quota_over_limit_put_outcome(Some("InvalidRequest"), Some("Bucket quota exceeded for bucket x")),
+        Ok(true)
+    );
+    assert_eq!(
+        quota_over_limit_put_outcome(Some("QuotaExceeded"), Some("Bucket quota exceeded")),
+        Ok(true)
+    );
+    assert_eq!(quota_over_limit_put_outcome(Some("SlowDown"), Some("slow down")), Ok(false));
+    assert_eq!(quota_over_limit_put_outcome(Some("ServiceUnavailable"), Some("unavailable")), Ok(false));
+    assert!(quota_over_limit_put_outcome(Some("AccessDenied"), Some("Access Denied")).is_err());
+    assert!(quota_over_limit_put_outcome(Some("InvalidRequest"), Some("invalid argument")).is_err());
+}

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,82 @@
+// 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, decommission_started_or_refused, put_inventory_retrying,
+    rebalance_started_or_refused, retrying_get_equals, retrying_put, 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::FourByFour).await?;
+    let bucket = unique_bucket("s3move");
+    dist.create_bucket(&bucket).await?;
+    let client = dist.client(0)?;
+    let inventory = put_inventory_retrying(&client, &bucket, 8, 16 * 1024, Duration::from_secs(30)).await?;
+
+    let decommission_started = decommission_started_or_refused(&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"
+    );
+
+    if decommission_started {
+        wait_for_decommission_complete(&dist.cluster, 0, Duration::from_secs(180)).await?;
+    }
+    assert_inventory(&live, &bucket, &inventory).await?;
+
+    let _ = rebalance_started_or_refused(&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/upgrade_test.rs

@@ -0,0 +1,363 @@
+// 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.
+
+//! 4-node upgrade coverage for historical objects and IAM AK/SK.
+//!
+//! Complements `upgrade_compatibility_test` (single-node SSE/multipart and
+//! mixed-version listing). This module pins the distributed contract the
+//! hardware upgrade chain is meant to catch: after a 4-node upgrade, objects
+//! written on the previous release still read back, and IAM user credentials
+//! created before the upgrade still authenticate.
+//!
+//! Requires `RUSTFS_UPGRADE_SOURCE_BINARY` pointing at the pinned previous
+//! release. The `e2e-distributed` workflow downloads that binary; a local run
+//! without it fails closed rather than skipping.
+
+use super::harness::{
+    DistCluster, DistLayout, TestResult, assert_object_bytes, cluster_admin_ok, enable_versioning, get_object_bytes, put_object,
+    unique_bucket, wait_until,
+};
+use crate::common::{
+    AdminTransport, admin_add_canned_policy_via, admin_attach_user_policy_via, admin_create_user_via, init_logging,
+};
+use aws_sdk_s3::Client;
+use aws_sdk_s3::error::ProvideErrorMetadata;
+use std::ffi::OsString;
+use std::path::{Path, PathBuf};
+use std::time::Duration;
+use uuid::Uuid;
+
+const SOURCE_BINARY_ENV: &str = "RUSTFS_UPGRADE_SOURCE_BINARY";
+const IAM_SECRET: &str = "UpgradeTestSecretKey1";
+const WRONG_SECRET: &str = "WrongSecretKey000000";
+const CREDENTIAL_TIMEOUT: Duration = Duration::from_secs(30);
+
+struct UpgradeSeed {
+    history_bucket: String,
+    history_key: &'static str,
+    history_body: Vec<u8>,
+    versioned_bucket: String,
+    versioned_key: &'static str,
+    version1: String,
+    version1_body: Vec<u8>,
+    version2: String,
+    version2_body: Vec<u8>,
+    iam_bucket: String,
+    iam_key: &'static str,
+    iam_body: Vec<u8>,
+    iam_user: String,
+    iam_secret: &'static str,
+}
+
+fn resolve_source_binary(value: Option<OsString>) -> TestResult<PathBuf> {
+    let path = value.map(PathBuf::from).ok_or_else(|| {
+        format!(
+            "{SOURCE_BINARY_ENV} must point to the pinned previous release binary (the e2e-distributed workflow downloads it)"
+        )
+    })?;
+    if !path.is_file() {
+        return Err(format!("upgrade source binary does not exist: {}", path.display()).into());
+    }
+    Ok(path)
+}
+
+fn source_binary() -> TestResult<PathBuf> {
+    resolve_source_binary(std::env::var_os(SOURCE_BINARY_ENV))
+}
+
+fn capture_upgrade_logs(cluster: &mut DistCluster, label: &str) -> 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.cluster.nodes.len() {
+        let path = Path::new(&log_dir).join(format!("{label}-node-{node_idx}.log"));
+        cluster
+            .cluster
+            .set_node_capture_log_path(node_idx, path.to_string_lossy().into_owned())?;
+    }
+    Ok(())
+}
+
+fn iam_rw_policy(bucket: &str) -> String {
+    serde_json::json!({
+        "Version": "2012-10-17",
+        "Statement": [{
+            "Effect": "Allow",
+            "Action": ["s3:*"],
+            "Resource": [
+                format!("arn:aws:s3:::{bucket}"),
+                format!("arn:aws:s3:::{bucket}/*")
+            ]
+        }]
+    })
+    .to_string()
+}
+
+async fn create_iam_user(dist: &DistCluster, user: &str, secret: &str, policy_name: &str, bucket: &str) -> TestResult {
+    let url = &dist.cluster.nodes[0].url;
+    let access = &dist.cluster.access_key;
+    let admin_secret = &dist.cluster.secret_key;
+    admin_create_user_via(AdminTransport::Signed, url, access, admin_secret, user, secret).await?;
+    admin_add_canned_policy_via(AdminTransport::Signed, url, access, admin_secret, policy_name, &iam_rw_policy(bucket)).await?;
+    admin_attach_user_policy_via(AdminTransport::Signed, url, access, admin_secret, policy_name, user).await?;
+    Ok(())
+}
+
+async fn wait_for_put(client: &Client, bucket: &str, key: &str, body: Vec<u8>, label: &str) -> TestResult {
+    wait_until(
+        CREDENTIAL_TIMEOUT,
+        || {
+            let client = client.clone();
+            let bucket = bucket.to_string();
+            let key = key.to_string();
+            let body = body.clone();
+            async move {
+                put_object(&client, &bucket, &key, body).await?;
+                Ok(true)
+            }
+        },
+        label,
+    )
+    .await
+}
+
+async fn wait_for_bytes(client: &Client, bucket: &str, key: &str, expected: &[u8], label: &str) -> TestResult {
+    wait_until(
+        CREDENTIAL_TIMEOUT,
+        || {
+            let client = client.clone();
+            let bucket = bucket.to_string();
+            let key = key.to_string();
+            let expected = expected.to_vec();
+            async move {
+                let got = get_object_bytes(&client, &bucket, &key).await?;
+                Ok(got == expected)
+            }
+        },
+        label,
+    )
+    .await
+}
+
+async fn seed_history_and_iam(dist: &DistCluster) -> TestResult<UpgradeSeed> {
+    let history_bucket = unique_bucket("upg-hist");
+    let versioned_bucket = unique_bucket("upg-ver");
+    let iam_bucket = unique_bucket("upg-iam");
+    dist.create_bucket(&history_bucket).await?;
+    dist.create_bucket(&versioned_bucket).await?;
+    dist.create_bucket(&iam_bucket).await?;
+
+    let root = dist.client(0)?;
+    enable_versioning(&root, &versioned_bucket).await?;
+
+    let history_key = "plain-history.bin";
+    let history_body = b"written by the previous 4-node release".to_vec();
+    put_object(&root, &history_bucket, history_key, history_body.clone()).await?;
+
+    let versioned_key = "versioned-history.txt";
+    let version1_body = b"version-one-before-upgrade".to_vec();
+    let version1 = root
+        .put_object()
+        .bucket(&versioned_bucket)
+        .key(versioned_key)
+        .body(aws_sdk_s3::primitives::ByteStream::from(version1_body.clone()))
+        .send()
+        .await?
+        .version_id()
+        .ok_or("first versioned PUT omitted version ID")?
+        .to_string();
+    let version2_body = b"version-two-before-upgrade".to_vec();
+    let version2 = root
+        .put_object()
+        .bucket(&versioned_bucket)
+        .key(versioned_key)
+        .body(aws_sdk_s3::primitives::ByteStream::from(version2_body.clone()))
+        .send()
+        .await?
+        .version_id()
+        .ok_or("second versioned PUT omitted version ID")?
+        .to_string();
+
+    let iam_user = format!("upg{}", &Uuid::new_v4().simple().to_string()[..8]);
+    let policy_name = format!("upgpol{}", &Uuid::new_v4().simple().to_string()[..8]);
+    create_iam_user(dist, &iam_user, IAM_SECRET, &policy_name, &iam_bucket).await?;
+
+    let iam_key = "iam-history.bin";
+    let iam_body = b"written with pre-upgrade IAM AK/SK".to_vec();
+    let iam_client = dist.client_with_credentials(1, &iam_user, IAM_SECRET)?;
+    wait_for_put(&iam_client, &iam_bucket, iam_key, iam_body.clone(), "IAM user PUT before upgrade").await?;
+
+    Ok(UpgradeSeed {
+        history_bucket,
+        history_key,
+        history_body,
+        versioned_bucket,
+        versioned_key,
+        version1,
+        version1_body,
+        version2,
+        version2_body,
+        iam_bucket,
+        iam_key,
+        iam_body,
+        iam_user,
+        iam_secret: IAM_SECRET,
+    })
+}
+
+async fn assert_history_and_iam(dist: &DistCluster, seed: &UpgradeSeed, context: &str) -> TestResult {
+    let root_a = dist.client(0)?;
+    let root_b = dist.client(3)?;
+    wait_for_bytes(
+        &root_b,
+        &seed.history_bucket,
+        seed.history_key,
+        &seed.history_body,
+        &format!("{context}: root GET historical object"),
+    )
+    .await?;
+    assert_object_bytes(&root_a, &seed.history_bucket, seed.history_key, &seed.history_body).await?;
+
+    let v1 = root_b
+        .get_object()
+        .bucket(&seed.versioned_bucket)
+        .key(seed.versioned_key)
+        .version_id(&seed.version1)
+        .send()
+        .await?;
+    let v1_body = v1.body.collect().await?.into_bytes();
+    if v1_body.as_ref() != seed.version1_body.as_slice() {
+        return Err(format!("{context}: version 1 bytes changed after upgrade").into());
+    }
+    let v2 = root_a
+        .get_object()
+        .bucket(&seed.versioned_bucket)
+        .key(seed.versioned_key)
+        .version_id(&seed.version2)
+        .send()
+        .await?;
+    let v2_body = v2.body.collect().await?.into_bytes();
+    if v2_body.as_ref() != seed.version2_body.as_slice() {
+        return Err(format!("{context}: version 2 bytes changed after upgrade").into());
+    }
+
+    let users = cluster_admin_ok(&dist.cluster, http::Method::GET, "/rustfs/admin/v3/list-users", None).await?;
+    if !users.contains(&seed.iam_user) {
+        return Err(format!("{context}: list-users lost IAM user {}: {users}", seed.iam_user).into());
+    }
+
+    let iam_on_upgraded = dist.client_with_credentials(0, &seed.iam_user, seed.iam_secret)?;
+    let iam_on_peer = dist.client_with_credentials(3, &seed.iam_user, seed.iam_secret)?;
+    wait_for_bytes(
+        &iam_on_upgraded,
+        &seed.iam_bucket,
+        seed.iam_key,
+        &seed.iam_body,
+        &format!("{context}: IAM GET historical object on node 0"),
+    )
+    .await?;
+    wait_for_bytes(
+        &iam_on_peer,
+        &seed.iam_bucket,
+        seed.iam_key,
+        &seed.iam_body,
+        &format!("{context}: IAM GET historical object on node 3"),
+    )
+    .await?;
+
+    let post_key = format!("after-upgrade-{context}.txt");
+    let post_body = format!("{context}: written with the same IAM AK/SK after upgrade").into_bytes();
+    wait_for_put(
+        &iam_on_peer,
+        &seed.iam_bucket,
+        &post_key,
+        post_body.clone(),
+        &format!("{context}: IAM PUT after upgrade"),
+    )
+    .await?;
+    assert_object_bytes(&iam_on_upgraded, &seed.iam_bucket, &post_key, &post_body).await?;
+
+    let bad = dist.client_with_credentials(1, &seed.iam_user, WRONG_SECRET)?;
+    match bad.get_object().bucket(&seed.iam_bucket).key(seed.iam_key).send().await {
+        Ok(_) => return Err(format!("{context}: wrong secret must not read the IAM object").into()),
+        Err(error) => {
+            let code = error.as_service_error().and_then(ProvideErrorMetadata::code);
+            if code == Some("SignatureDoesNotMatch")
+                || code == Some("InvalidAccessKeyId")
+                || code == Some("AccessDenied")
+                || code == Some("InvalidArgument")
+            {
+            } else if error.raw_response().is_some_and(|response| response.status().as_u16() == 403) {
+            } else {
+                return Err(format!("{context}: wrong secret failed with unexpected error {error:?}").into());
+            }
+        }
+    }
+
+    let post_root_key = format!("root-after-{context}.bin");
+    let post_root_body = format!("{context}: root write after upgrade").into_bytes();
+    put_object(&root_a, &seed.history_bucket, &post_root_key, post_root_body.clone()).await?;
+    assert_object_bytes(&root_b, &seed.history_bucket, &post_root_key, &post_root_body).await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn four_node_direct_upgrade_preserves_history_and_iam_credentials() -> TestResult {
+    init_logging();
+    let previous = source_binary()?;
+    let mut dist = DistCluster::new_stopped(DistLayout::FourNodeFourDisk).await?;
+    capture_upgrade_logs(&mut dist, "direct-upgrade")?;
+    dist.start_from_binary(&previous).await?;
+
+    let seed = seed_history_and_iam(&dist).await?;
+    dist.restart_with_current_binary().await?;
+    assert_history_and_iam(&dist, &seed, "direct").await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn four_node_rolling_upgrade_preserves_history_and_iam_credentials() -> TestResult {
+    init_logging();
+    let previous = source_binary()?;
+    let mut dist = DistCluster::new_stopped(DistLayout::FourNodeFourDisk).await?;
+    capture_upgrade_logs(&mut dist, "rolling-upgrade")?;
+    dist.start_from_binary(&previous).await?;
+
+    let seed = seed_history_and_iam(&dist).await?;
+
+    dist.replace_node_with_current_binary(0).await?;
+    assert_history_and_iam(&dist, &seed, "one-current-node").await?;
+
+    for node_idx in [1, 2] {
+        dist.replace_node_with_current_binary(node_idx).await?;
+    }
+    assert_history_and_iam(&dist, &seed, "one-previous-node").await?;
+
+    dist.replace_node_with_current_binary(3).await?;
+    assert_history_and_iam(&dist, &seed, "homogeneous-current").await?;
+    Ok(())
+}
+
+#[test]
+fn missing_upgrade_source_binary_fails_closed() {
+    let err = resolve_source_binary(None).expect_err("absent env must fail closed");
+    assert!(err.to_string().contains(SOURCE_BINARY_ENV), "{err}");
+}
+
+#[test]
+fn missing_upgrade_source_binary_file_fails_closed() {
+    let err = resolve_source_binary(Some("/no/such/rustfs-upgrade-source".into())).expect_err("missing file must fail closed");
+    assert!(err.to_string().contains("does not exist"), "{err}");
+}

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;

docs/testing/README.md

@@ -12,10 +12,10 @@ Pick the lowest layer that can prove the change; add a higher-layer test only wh
 |---|---|---|---|
 | Unit & crate integration | Per-crate logic and in-process integration tests | `cargo nextest run --all --exclude e2e_test` (or `-p <crate>`); `make test` wraps it | Every PR, required (`Test and Lint`, `ci` profile) |
 | ecstore black-box | Erasure-coded read/write/recovery validation; profiles `quick` / `full` / `destructive` / `fuzz` | `scripts/run_ecstore_validation_suite.sh --profile quick` | Local and release validation only; not wired into any workflow. Contract: [ecstore-validation-suite-design.md](ecstore-validation-suite-design.md) |
-| e2e (`e2e_test` crate) | A real `rustfs` binary per test, driven over the S3, admin, and protocol APIs | `cargo nextest run --profile e2e-smoke -p e2e_test` | PR: `e2e-smoke` (report-only); merge queue / main push: `e2e-full`; nightly: `e2e-repl-nightly`, `e2e-nightly`, `e2e-protocols`. Guide: [`crates/e2e_test/README.md`](../../crates/e2e_test/README.md) |
+| e2e (`e2e_test` crate) | A real `rustfs` binary per test, driven over the S3, admin, and protocol APIs | `cargo nextest run --profile e2e-smoke -p e2e_test` | PR: `e2e-smoke` (report-only); merge queue / main push: `e2e-full`; nightly: `e2e-repl-nightly`, `e2e-nightly`, `e2e-protocols`, `e2e-distributed`. Guide: [`crates/e2e_test/README.md`](../../crates/e2e_test/README.md); 4-node 4-disk map: [distributed-e2e.md](distributed-e2e.md) |
 | s3s-e2e conformance | External S3 conformance tool against a live server | `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>` | PR, report-only (second half of the `End-to-End Tests` job) |
 | S3 compatibility | `ceph/s3-tests` (boto3; allow-list `scripts/s3-tests/implemented_tests.txt`) and MinIO `mint` | `scripts/s3-tests/run.sh`; mint via `.github/workflows/mint.yml` | s3-tests: PR report-only plus a weekly full sweep; mint: weekly, report-only |
-| Chaos / fault-injection | Single-node disk fault injection (`crates/e2e_test/src/chaos.rs`, `crates/e2e_test/src/fault_proxy.rs`) used by the reliability and heal e2e modules | Part of the e2e crate (`e2e-reliability` test-group) | With the `e2e-full` and nightly e2e lanes. A multi-node power-loss harness is not in tree |
+| Chaos / fault-injection | Single-node disk fault injection (`crates/e2e_test/src/chaos.rs`, `crates/e2e_test/src/fault_proxy.rs`) plus the 4-node kill/offline-drive/blackhole cases in `crates/e2e_test/src/distributed/chaos_test.rs` | Part of the e2e crate (`e2e-reliability` and `e2e-distributed`) | Reliability cases with `e2e-full`; 4-node chaos with the `e2e-distributed` nightly lane |
 | Fuzz | `cargo-fuzz` targets over untrusted parsing surfaces; isolated sub-workspace under `fuzz/` | `./scripts/fuzz/run.sh` (see [`fuzz/README.md`](../../fuzz/README.md)) | PR smoke on the paths listed in `.github/workflows/fuzz.yml`, plus nightly corpus |
 | Benchmarks | Criterion benches under each crate's `benches/` | `cargo bench -p <crate>` | On demand; never a gate |
 
@@ -61,6 +61,7 @@ All profiles are defined in `.config/nextest.toml`; its block comments hold the
 | `e2e-full` | Merge-queue / main-push single-node e2e lane |
 | `e2e-repl-nightly` | Nightly slow / cross-process replication lane |
 | `e2e-nightly` | Nightly serial multi-process cluster fault lane |
+| `e2e-distributed` | Nightly 4-node 4-disk S3 / lock / versioning / replication / quota / expand / decommission / rebalance / site-replication / chaos / upgrade (history + IAM AK/SK) lane |
 | `e2e-protocols` | Nightly fixed-port FTPS/SFTP/WebDAV lane, run with `-j 1` |
 
 Membership of each e2e profile is pinned by a digest in `.config/e2e-<profile>-selection.txt` and checked by `scripts/check_test_wiring.py --check-profile <profile>` before the lane runs. To list what a profile selects on your platform (the result is platform-dependent because some modules are linux-only):

docs/testing/ci-gates.md

@@ -74,6 +74,7 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run, sched
 | `ci.yml` (weekly) | full matrix, including the schedule/dispatch-only rio-v2 jobs `build-rustfs-debug-binary-rio-v2` and `e2e-tests-rio-v2` | per-job | yes | dispatch `ci.yml` |
 | `build.yml` (weekly) | `build-rustfs` over the six-target platform matrix in `prepare-platform-matrix` (four Linux, macOS aarch64, Windows x86_64) | build/package integrity | yes | dispatch `build.yml` with an exact platform set |
 | `e2e-replication-nightly.yml` (nightly) | `repl-nightly`, `cluster-nightly`, `protocols-nightly` | three independent gates; JUnit, membership listing, server logs | yes | `cargo nextest run --profile e2e-repl-nightly -p e2e_test`; `--profile e2e-nightly`; `-j 1 --profile e2e-protocols` |
+| `e2e-distributed.yml` (nightly) | `distributed` | 4-node 4-disk e2e gate including direct/rolling upgrade of historical objects and IAM AK/SK; JUnit, membership listing, server logs | yes, with `never_ran_grace_until` | download the pinned previous release as in the workflow, export `RUSTFS_UPGRADE_SOURCE_BINARY`, then `cargo nextest run --profile e2e-distributed -p e2e_test` |
 | `e2e-s3tests.yml` (weekly) | `s3tests` (single and distributed, four shards each), `upstream-head-canary` | compatibility gate; report, JUnit, node IDs, server logs | yes | `scripts/s3-tests/run.sh` against an existing single or distributed target |
 | `fuzz.yml` (nightly) | `nightly-fuzz-corpus` per target | gate; corpus and crash artifacts | yes | `MAX_TOTAL_TIME=<seconds> ./scripts/fuzz/run.sh` |
 | `minio-interop.yml` (nightly) | `minio-interop` | EC + SSE read-parity gate | yes, with `never_ran_grace_until` | pinned Docker fixture steps in the workflow |

docs/testing/distributed-e2e.md

@@ -0,0 +1,70 @@
+# Distributed 4-node 4-disk e2e
+
+**Use this when:** adding or diagnosing GitHub Actions coverage for a 4-node cluster, or deciding whether a behaviour belongs in `e2e-distributed` versus the single-node `e2e-full` lane, the nightly cluster-fault lane, or the hardware functional chain.
+**Source of truth:** `crates/e2e_test/src/distributed/`, `[profile.e2e-distributed]` in `.config/nextest.toml`, `.github/workflows/e2e-distributed.yml`.
+
+## Topology
+
+The in-tree harness runs every node on `127.0.0.1` with a distinct port. That matches `RustFSTestClusterEnvironment` in `crates/e2e_test/src/common.rs`:
+
+| Layout | Constructor | Use |
+|---|---|---|
+| 4 nodes × 4 drives, one pool | `ClusterTopology::single_pool_multidrive(4, 4)` | S3, object lock, versioning, quota, observability, concurrency, chaos |
+| 4 nodes × 1 drive, one pool | `ClusterTopology::single_pool(4)` | Two-site replication (8 processes total); direct/rolling upgrade from the pinned previous release |
+| 2 single-node pools × 4 drives | `ClusterTopology::per_node_pools(4, [[0],[1]])` | Harness-only: `append_single_node_pool` unit tests. Live multi-pool expand/restart currently dies with `pool metadata recovery required`; this lane does not change that production gate |
+
+A pool striped across several localhost ports is not expressible (`RUSTFS_VOLUMES` host ellipses would collide on disk paths). Multi-host striped pools remain the hardware functional-chain / backlog #1313 / #1314 lane.
+
+Decommission and rebalance POST on the 4×4 single-pool layout is refused by the current product (`single pool deployments do not support decommission`, NotImplemented, or opaque 500 InternalError when the inner pool-meta fence is wrapped). 502/503 are not treated as a product refusal. Those cases still assert object bytes and SHA-256; when the API starts they wait for completion and assert post-move integrity. They do not treat a refusal as a successful move. This lane does not change production pool-meta bootstrap, write-fence, or decommission policy; it only observes the current server behavior.
+
+## What this lane covers
+
+`cargo nextest run --profile e2e-distributed -p e2e_test` selects `distributed::*`:
+
+- S3 put / get / head / list / copy / rename / delete / presign / empty object
+- Object Lock COMPLIANCE, GOVERNANCE (with bypass), legal hold
+- Versioning, version GET, delete marker
+- Bucket replication between two 4-node clusters; hard quota
+- Health / admin info / storageinfo / audit target list
+- Pool restart, decommission/rebalance *attempts*, checksum integrity, S3 during those attempts on 4×4. Live multi-pool expand/restart is a production pool-meta bootstrap limitation and is not patched here; `append_single_node_pool` is covered by harness unit tests
+- Site replication object convergence
+- High-concurrency PUT/GET; concurrent PUT during decommission
+- Node kill/restart, full process restart, drive offline (4×4). Volume-proxy blackhole stays in `cluster_volume_fault_proxy_pass_smoke` (2×2); a 4-node volume proxy cannot format because RPC audience is the listen port
+- Multipart, cross-node listing, list-buckets agreement
+- Concurrent GET while a peer node is killed
+- Direct and rolling upgrade from the pinned previous release: historical objects, versioned history, and IAM user AK/SK still work afterwards
+
+## Existing Actions gaps this lane does not replace
+
+Those suites stay in place; this lane fills the in-tree 4×4 hole they leave.
+
+| Existing lane | Gap |
+|---|---|
+| `rustfs-*-test.yml` functional chain | Clones private `rustfs/auto-testing`, runs on three shared VMs (`vm000`–`vm002`), `continue-on-error: true`, not a merge signal, not 4 nodes. Hardware `rustfs-upgrade-test.yml` stays there |
+| `e2e-upgrade.yml` | Single-node SSE/multipart/delete-marker contracts plus mixed-version listing; does not pin IAM user AK/SK on a 4-node cluster |
+| `e2e-smoke` / `e2e-full` | Almost all cases are single-node |
+| `e2e-nightly` | 4-node cluster faults and heal, not S3/lock/versioning/quota/decommission matrix |
+| `e2e-repl-nightly` | Site and bucket replication on 1–3 *single-node* processes |
+| `e2e-s3tests.yml` `multi` | Weekly ceph/s3-tests against Docker 4-node; not lock/WORM, decommission, chaos, or checksum integrity |
+| `crates/e2e_test/src/chaos.rs` | Single-node disk faults only |
+
+Hardware power-loss, NIC pull, and real disk replacement still belong on the smoke-testing VMs. This lane simulates those with SIGKILL, directory rename, and `FaultProxy` blackhole.
+
+## Run
+
+```bash
+cargo build -p rustfs --bins
+# Upgrade cases require the pinned previous binary (CI downloads it).
+export RUSTFS_UPGRADE_SOURCE_BINARY=/path/to/rustfs-1.0.0-rc.2
+cargo nextest run --profile e2e-distributed -p e2e_test
+```
+
+Without `RUSTFS_UPGRADE_SOURCE_BINARY` the two `distributed::upgrade_test::*` cases fail closed. Filter them out for a local run that is not checking upgrade:
+
+```bash
+cargo nextest run --profile e2e-distributed -p e2e_test -E 'not test(/^distributed::upgrade_test::/)'
+```
+
+The upgrade topology is `ClusterTopology::single_pool(4)` (4 nodes × 1 drive). That matches the proven mixed-version fixture in `upgrade_compatibility_test`; 4×4 localhost drives are rejected by the previous release's same-device disk check.
+
+Membership is pinned by `.config/e2e-distributed-selection.txt`. Update it with `python3 ./scripts/check_test_wiring.py --update-profile e2e-distributed <listing.json> linux` after adding or renaming a case.