Merge branch 'main' into test/distributed-e2e-hardening

test(odm): exercise overflow and invalid cursors reliably (#7236 )
2026-09-05 19:55:37 +00:00 · 2026-09-06 03:32:13 +08:00 · 2026-09-06 03:09:39 +08:00 · 2026-09-05 23:56:44 +08:00 · 2026-09-05 22:50:00 +08:00 · 2026-09-05 22:21:31 +08:00
53 changed files with 4387 additions and 3244 deletions
@@ -0,0 +1,2 @@
+sha256-linux=9785867929047dfd8c6f768e0d2b1e0a8fdba85216f4a4139093b1619d03ff07
+sha256-darwin=9785867929047dfd8c6f768e0d2b1e0a8fdba85216f4a4139093b1619d03ff07
@@ -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,27 @@ path = "junit.xml"
 filter = 'package(e2e_test)'
 test-group = 'e2e-cluster-nightly'

+# ---------------------------------------------------------------------------
+# e2e-distributed profile — 4-node 4-disk Actions suite
+# ---------------------------------------------------------------------------
+# Storage-sensitive PR / 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 with no retries.
+[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 +614,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 +639,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)::/)
@@ -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 },
@@ -0,0 +1,199 @@
+# 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`. Storage-sensitive PRs,
+# nightly runs, and manual dispatches all execute the same fail-closed suite.
+# Upgrade cases download the same pinned previous release as e2e-upgrade.yml.
+
+name: e2e-distributed
+
+on:
+  pull_request:
+    paths:
+      - "Cargo.lock"
+      - "Cargo.toml"
+      - ".config/nextest.toml"
+      - ".github/workflows/e2e-distributed.yml"
+      - "crates/audit/**"
+      - "crates/common/**"
+      - "crates/config/**"
+      - "crates/e2e_test/**"
+      - "crates/ecstore/**"
+      - "crates/filemeta/**"
+      - "crates/heal/**"
+      - "crates/iam/**"
+      - "crates/lock/**"
+      - "crates/madmin/**"
+      - "crates/notify/**"
+      - "crates/replication/**"
+      - "crates/s3-client/**"
+      - "crates/s3-ops/**"
+      - "crates/s3-types/**"
+      - "crates/scanner/**"
+      - "crates/storage-api/**"
+      - "crates/utils/**"
+      - "rustfs/**"
+  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: Prepare isolated filesystems for pool movement
+        run: |
+          set -euo pipefail
+          mount_base="${RUNNER_TEMP}/rustfs-e2e-pools"
+          mkdir -p "${mount_base}"
+          roots=()
+          for pool in 0 1 2 3; do
+            image="${mount_base}/pool-${pool}.img"
+            mountpoint="${mount_base}/pool-${pool}"
+            truncate -s 1G "${image}"
+            mkfs.ext4 -q -F "${image}"
+            mkdir -p "${mountpoint}"
+            sudo mount -o loop,nosuid,nodev "${image}" "${mountpoint}"
+            sudo chmod 1777 "${mountpoint}"
+            roots+=("${mountpoint}")
+          done
+          printf -v joined_roots '%s:' "${roots[@]}"
+          echo "RUSTFS_E2E_POOL_ROOTS=${joined_roots%:}" >> "${GITHUB_ENV}"
+          findmnt --noheadings --output TARGET,SOURCE,FSTYPE --target "${roots[0]}"
+          findmnt --noheadings --output TARGET,SOURCE,FSTYPE --target "${roots[1]}"
+          findmnt --noheadings --output TARGET,SOURCE,FSTYPE --target "${roots[2]}"
+          findmnt --noheadings --output TARGET,SOURCE,FSTYPE --target "${roots[3]}"
+
+      - 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
+          FILTER: ${{ inputs.filter }}
+        run: |
+          set -euo pipefail
+          if [ -n "${FILTER}" ]; then
+            cargo nextest run --profile e2e-distributed -p e2e_test -E "${FILTER}"
+          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
+
+      - name: Unmount isolated pool filesystems
+        if: always()
+        run: |
+          set -euo pipefail
+          mount_base="${RUNNER_TEMP}/rustfs-e2e-pools"
+          for pool in 0 1 2 3; do
+            mountpoint="${mount_base}/pool-${pool}"
+            if mountpoint --quiet "${mountpoint}"; then
+              sudo umount "${mountpoint}"
+            fi
+          done
+
+  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 }}
@@ -22,6 +22,7 @@ on:
      - "Continuous Integration"
      - "coverage"
      - "e2e-nightly"
+      - "e2e-distributed"
      - "e2e-s3tests"
      - "Fuzz"
      - "mint"
@@ -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) | Storage-sensitive PR and 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** (storage-sensitive PR / nightly / dispatch) |
 | 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
@@ -1700,6 +1700,69 @@ impl RustFSTestClusterEnvironment {
        Ok(())
    }

+    /// Append a new single-node erasure pool to a stopped multi-pool cluster.
+    ///
+    /// Used to simulate pool expansion on localhost: every pool already owns
+    /// exactly one node with `drives_per_node >= 2` (the only multi-pool layout
+    /// the single-host `RUSTFS_VOLUMES` syntax can express). The new node is
+    /// allocated a fresh port and empty drive directories; callers must
+    /// [`Self::start`] afterwards so every process picks up the extended
+    /// volumes argument. Existing data directories are left untouched.
+    pub async fn append_single_node_pool(&mut self) -> Result<usize, Box<dyn std::error::Error + Send + Sync>> {
+        if self.nodes.iter().any(|node| node.process.is_some()) {
+            return Err("stop the cluster before appending a pool".into());
+        }
+        if self.topology.drives_per_node < 2 {
+            return Err(
+                "append_single_node_pool requires drives_per_node >= 2 (the server parser rejects a single-drive ellipses pool)"
+                    .into(),
+            );
+        }
+
+        let mut pools = self.topology.normalized_pools();
+        for (pool_idx, nodes) in pools.iter().enumerate() {
+            if nodes.len() != 1 {
+                return Err(format!(
+                    "pool {pool_idx} spans {} nodes; append_single_node_pool requires one node per pool",
+                    nodes.len()
+                )
+                .into());
+            }
+        }
+
+        let new_idx = self.nodes.len();
+        let port = RustFSTestEnvironment::find_available_port().await?;
+        let address = format!("127.0.0.1:{port}");
+        let data_dirs: Vec<String> = (0..self.topology.drives_per_node)
+            .map(|drive| format!("{}/node{}/drive{}", self.temp_dir, new_idx, drive))
+            .collect();
+        for dir in &data_dirs {
+            fs::create_dir_all(dir).await?;
+        }
+
+        self.nodes.push(ClusterNode {
+            url: format!("http://{address}"),
+            address,
+            data_dir: data_dirs[0].clone(),
+            data_dirs,
+            pool_idx: pools.len(),
+            process: None,
+        });
+        pools.push(vec![new_idx]);
+        self.topology.node_count = self.nodes.len();
+        self.topology.pools = pools;
+        self.node_extra_env.push(Vec::new());
+        self.node_capture_log_paths.push(None);
+        self.volume_proxy_addresses.push(None);
+
+        if !self.extra_env.iter().any(|(key, _)| key == "RUSTFS_UNSAFE_BYPASS_DISK_CHECK") {
+            self.extra_env
+                .push(("RUSTFS_UNSAFE_BYPASS_DISK_CHECK".to_string(), "true".to_string()));
+        }
+
+        Ok(new_idx)
+    }
+
    /// Gracefully stop one cluster node and wait for its process to exit.
    ///
    /// This is intentionally separate from [`Self::stop_node`]: the latter is
@@ -0,0 +1,222 @@
+// 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, retrying_get_equals, unique_bucket,
+    wait_for_ready, wait_until,
+};
+use crate::chaos::{census_object_version_on_disk, signed_admin_post};
+use crate::common::{build_test_s3_config, init_logging};
+use crate::fault_proxy::FaultMode;
+use aws_sdk_s3::Client;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::{Barrier, mpsc};
+use tokio::time::timeout;
+
+#[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 fresh_drive_replacement_is_physically_healed_without_data_change() -> TestResult {
+    init_logging();
+    let mut dist = DistCluster::start_with_env(DistLayout::FourByFour, &[("RUSTFS_HEAL_ENABLED", "true")]).await?;
+    let bucket = unique_bucket("baddrive");
+    dist.create_bucket(&bucket).await?;
+    let body = payload_for("fresh-drive/durable.bin", 8 * 1024 * 1024);
+    put_object(&dist.client(1)?, &bucket, "durable.bin", body.clone()).await?;
+
+    let replaced_drive = PathBuf::from(&dist.cluster.nodes[0].data_dirs[0]);
+    let baseline = census_object_version_on_disk(&replaced_drive, &bucket, "durable.bin", None)?;
+    assert!(
+        baseline.is_complete(),
+        "replacement target did not hold a complete baseline shard: {baseline:?}"
+    );
+    assert!(
+        !baseline.expected_part_numbers.is_empty(),
+        "replacement witness must use physical part shards: {baseline:?}"
+    );
+
+    dist.cluster.stop_node(0)?;
+    let format_path = replaced_drive.join(".rustfs.sys/format.json");
+    let format = std::fs::read(&format_path)?;
+    let retired_drive = PathBuf::from(format!("{}.retired", replaced_drive.display()));
+    std::fs::rename(&replaced_drive, &retired_drive)?;
+    std::fs::create_dir_all(format_path.parent().ok_or("replacement format path omitted parent")?)?;
+    std::fs::write(&format_path, format)?;
+    let empty = census_object_version_on_disk(&replaced_drive, &bucket, "durable.bin", None)?;
+    assert!(!empty.has_xl_meta, "fresh replacement unexpectedly retained object metadata: {empty:?}");
+
+    dist.cluster.start_node(0).await?;
+    wait_for_ready(&dist.cluster).await?;
+    let heal_body =
+        r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#;
+    let heal_url = format!("{}/rustfs/admin/v3/heal/{bucket}?forceStart=true", dist.cluster.nodes[1].url);
+    signed_admin_post(&heal_url, Some(heal_body), &dist.cluster.access_key, &dist.cluster.secret_key).await?;
+    wait_until(
+        Duration::from_secs(90),
+        || async {
+            let healed = census_object_version_on_disk(&replaced_drive, &bucket, "durable.bin", None)?;
+            Ok(healed.matches_manifest(&baseline))
+        },
+        "fresh replacement contains the original complete shard manifest",
+    )
+    .await?;
+
+    for node_idx in 0..dist.cluster.nodes.len() {
+        assert_object_bytes(&dist.client(node_idx)?, &bucket, "durable.bin", &body).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 = payload_for("inflight/steady.bin", 8 * 1024 * 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 worker_count = 12;
+    let release = Arc::new(Barrier::new(worker_count + 1));
+    let (started_tx, mut started_rx) = mpsc::unbounded_channel();
+    let mut handles = Vec::new();
+    for idx in 0..worker_count {
+        let client = live[idx % live.len()].clone();
+        let bucket = bucket.clone();
+        let body = body.clone();
+        let release = release.clone();
+        let started_tx = started_tx.clone();
+        handles.push(tokio::spawn(async move {
+            let response = client.get_object().bucket(&bucket).key("steady.bin").send().await?;
+            if response.content_length() != Some(body.len() as i64) {
+                return Err::<(), Box<dyn std::error::Error + Send + Sync>>(
+                    format!("worker {idx} received a wrong content length").into(),
+                );
+            }
+            started_tx.send(idx)?;
+            release.wait().await;
+            let actual = response.body.collect().await?.into_bytes();
+            if actual.as_ref() != body.as_slice() {
+                return Err(format!("worker {idx} received corrupted bytes after peer kill").into());
+            }
+            Ok(())
+        }));
+    }
+    drop(started_tx);
+    for _ in 0..worker_count {
+        timeout(Duration::from_secs(30), started_rx.recv())
+            .await?
+            .ok_or("a streaming GET exited before reaching the kill barrier")?;
+    }
+
+    dist.cluster.stop_node(3)?;
+    release.wait().await;
+    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(())
+}
+
+#[tokio::test]
+async fn blackholed_node_client_network_preserves_cluster_availability_and_recovers() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let proxy = crate::fault_proxy::FaultProxy::start(dist.cluster.nodes[3].address.parse()?).await?;
+    let proxied_url = format!("http://{}", proxy.local_addr());
+    let proxied_client = Client::from_conf(build_test_s3_config(
+        &proxied_url,
+        &dist.cluster.access_key,
+        &dist.cluster.secret_key,
+        None,
+        "distributed-network-chaos",
+    ));
+
+    let result: TestResult = async {
+        let bucket = unique_bucket("netfault");
+        dist.create_bucket(&bucket).await?;
+        let baseline = payload_for("network/baseline.bin", 1024 * 1024);
+        put_object(&dist.client(0)?, &bucket, "baseline.bin", baseline.clone()).await?;
+        assert_object_bytes(&proxied_client, &bucket, "baseline.bin", &baseline).await?;
+
+        proxy.set_mode(FaultMode::Blackhole);
+        assert_eq!(proxy.mode(), FaultMode::Blackhole);
+        if let Ok(Ok(_)) = timeout(
+            Duration::from_secs(5),
+            proxied_client.get_object().bucket(&bucket).key("baseline.bin").send(),
+        )
+        .await
+        {
+            return Err("blackholed node endpoint unexpectedly completed a GET".into());
+        }
+
+        let during = payload_for("network/during.bin", 1024 * 1024);
+        timeout(Duration::from_secs(30), async {
+            put_object(&dist.client(1)?, &bucket, "during-blackhole.bin", during.clone()).await?;
+            assert_object_bytes(&dist.client(2)?, &bucket, "baseline.bin", &baseline).await?;
+            assert_object_bytes(&dist.client(0)?, &bucket, "during-blackhole.bin", &during).await?;
+            Ok::<_, Box<dyn std::error::Error + Send + Sync>>(())
+        })
+        .await??;
+
+        proxy.set_mode(FaultMode::Pass);
+        retrying_get_equals(&proxied_client, &bucket, "during-blackhole.bin", &during, Duration::from_secs(30)).await?;
+        Ok(())
+    }
+    .await;
+
+    proxy.set_mode(FaultMode::Pass);
+    proxy.shutdown().await;
+    result
+}
@@ -0,0 +1,98 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::harness::{DistCluster, DistLayout, TestResult, assert_object_bytes, payload_for, put_object, unique_bucket};
+use crate::common::init_logging;
+use std::collections::BTreeSet;
+use std::sync::Arc;
+use tokio::sync::Barrier;
+
+#[tokio::test]
+async fn four_node_high_concurrency_mixed_workload_is_consistent_on_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 worker_count = 24;
+    let rounds = 4;
+    let barrier = Arc::new(Barrier::new(worker_count));
+
+    let mut handles = Vec::new();
+    for idx in 0..worker_count {
+        let clients = clients.clone();
+        let barrier = barrier.clone();
+        let bucket = bucket.clone();
+        handles.push(tokio::spawn(async move {
+            barrier.wait().await;
+            let writer = &clients[idx % clients.len()];
+            let reader = &clients[(idx + 1) % clients.len()];
+            let copier = &clients[(idx + 2) % clients.len()];
+            let mut retained = Vec::with_capacity(rounds);
+            for round in 0..rounds {
+                let key = format!("source/worker-{idx:02}-round-{round}.bin");
+                let copy_key = format!("retained/worker-{idx:02}-round-{round}.bin");
+                let body = payload_for(&key, 64 * 1024);
+                put_object(writer, &bucket, &key, body.clone()).await?;
+
+                let head = reader.head_object().bucket(&bucket).key(&key).send().await?;
+                if head.content_length() != Some(body.len() as i64) {
+                    return Err(format!("HEAD returned the wrong size for {key}: {head:?}").into());
+                }
+                assert_object_bytes(reader, &bucket, &key, &body).await?;
+
+                copier
+                    .copy_object()
+                    .bucket(&bucket)
+                    .key(&copy_key)
+                    .copy_source(format!("{bucket}/{key}"))
+                    .send()
+                    .await?;
+                assert_object_bytes(writer, &bucket, &copy_key, &body).await?;
+
+                writer.delete_object().bucket(&bucket).key(&key).send().await?;
+                let missing = reader
+                    .head_object()
+                    .bucket(&bucket)
+                    .key(&key)
+                    .send()
+                    .await
+                    .expect_err("deleted source key must not remain visible");
+                if missing.raw_response().map(|response| response.status().as_u16()) != Some(404) {
+                    return Err(format!("deleted source {key} returned an unexpected result: {missing:?}").into());
+                }
+                retained.push((copy_key, body));
+            }
+            Ok::<_, Box<dyn std::error::Error + Send + Sync>>(retained)
+        }));
+    }
+
+    let mut inventory = Vec::new();
+    for handle in handles {
+        inventory.extend(handle.await??);
+    }
+
+    let expected_keys: BTreeSet<_> = inventory.iter().map(|(key, _)| key.as_str()).collect();
+    for (node_idx, client) in clients.iter().enumerate() {
+        let listed = client.list_objects_v2().bucket(&bucket).prefix("retained/").send().await?;
+        let listed_keys: BTreeSet<_> = listed.contents().iter().filter_map(|object| object.key()).collect();
+        assert_eq!(listed_keys, expected_keys, "node {node_idx} returned a divergent retained-key listing");
+        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(())
+}
@@ -0,0 +1,74 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::harness::{
+    DECOMMISSION_POOL_ID, DistCluster, DistLayout, TestResult, assert_inventory, decommission_running_with_progress,
+    decommission_status_json, payload_for, put_inventory_retrying, retrying_get_equals, retrying_put, start_decommission,
+    unique_bucket, wait_for_decommission_complete, wait_for_decommission_running_with_progress,
+};
+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 mut dist = DistCluster::start(DistLayout::SingleNodeFourDrive).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, 96, 256 * 1024, Duration::from_secs(30)).await?;
+    dist.expand_to_four_pools().await?;
+
+    start_decommission(&dist.cluster, DECOMMISSION_POOL_ID).await?;
+
+    let clients = Arc::new(dist.clients()?);
+    let barrier = Arc::new(Barrier::new(17));
+    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))
+        }));
+    }
+
+    wait_for_decommission_running_with_progress(&dist.cluster, DECOMMISSION_POOL_ID, Duration::from_secs(30)).await?;
+    barrier.wait().await;
+
+    let mut live_objects = Vec::new();
+    for handle in handles {
+        live_objects.push(handle.await??);
+    }
+    let status = decommission_status_json(&dist.cluster).await?;
+    if !decommission_running_with_progress(&status, DECOMMISSION_POOL_ID)? {
+        return Err(format!("decommission did not remain active across concurrent PUTs: {status}").into());
+    }
+
+    wait_for_decommission_complete(&dist.cluster, DECOMMISSION_POOL_ID, Duration::from_secs(180)).await?;
+
+    let checker = dist.client(2)?;
+    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(())
+}
@@ -0,0 +1,156 @@
+// 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::{
+    DECOMMISSION_POOL_ID, DistCluster, DistLayout, TestResult, assert_inventory, enable_versioning, put_inventory_retrying,
+    sha256_hex, start_decommission, unique_bucket, wait_for_decommission_active, wait_for_decommission_complete,
+};
+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 decommission_does_not_alter_object_sha256_across_pools() -> TestResult {
+    init_logging();
+    let mut dist = DistCluster::start(DistLayout::SingleNodeFourDrive).await?;
+    let bucket = unique_bucket("integrity");
+    dist.create_bucket(&bucket).await?;
+    let client = dist.client(0)?;
+    enable_versioning(&client, &bucket).await?;
+    let inventory = put_inventory_retrying(&client, &bucket, 96, 256 * 1024, Duration::from_secs(30)).await?;
+    let before: Vec<(String, String)> = inventory.iter().map(|(key, body)| (key.clone(), sha256_hex(body))).collect();
+
+    let versioned_key = "history/versioned.bin";
+    let version_one = b"historical bytes before data movement".to_vec();
+    let version_two = b"current bytes before data movement".to_vec();
+    let version_one_id = client
+        .put_object()
+        .bucket(&bucket)
+        .key(versioned_key)
+        .body(ByteStream::from(version_one.clone()))
+        .send()
+        .await?
+        .version_id()
+        .ok_or("historical PUT omitted version ID")?
+        .to_string();
+    let version_two_id = client
+        .put_object()
+        .bucket(&bucket)
+        .key(versioned_key)
+        .body(ByteStream::from(version_two.clone()))
+        .send()
+        .await?
+        .version_id()
+        .ok_or("current PUT omitted version ID")?
+        .to_string();
+
+    let multipart_key = "multipart/moved.bin";
+    let first_part = vec![0x31; 5 * 1024 * 1024];
+    let second_part = vec![0x72; 1024 * 1024];
+    let upload = client
+        .create_multipart_upload()
+        .bucket(&bucket)
+        .key(multipart_key)
+        .send()
+        .await?;
+    let upload_id = upload.upload_id().ok_or("movement multipart upload omitted upload ID")?;
+    let uploaded_one = client
+        .upload_part()
+        .bucket(&bucket)
+        .key(multipart_key)
+        .upload_id(upload_id)
+        .part_number(1)
+        .body(ByteStream::from(first_part.clone()))
+        .send()
+        .await?;
+    let uploaded_two = client
+        .upload_part()
+        .bucket(&bucket)
+        .key(multipart_key)
+        .upload_id(upload_id)
+        .part_number(2)
+        .body(ByteStream::from(second_part.clone()))
+        .send()
+        .await?;
+    client
+        .complete_multipart_upload()
+        .bucket(&bucket)
+        .key(multipart_key)
+        .upload_id(upload_id)
+        .multipart_upload(
+            CompletedMultipartUpload::builder()
+                .parts(
+                    CompletedPart::builder()
+                        .part_number(1)
+                        .e_tag(uploaded_one.e_tag().ok_or("movement part 1 omitted ETag")?)
+                        .build(),
+                )
+                .parts(
+                    CompletedPart::builder()
+                        .part_number(2)
+                        .e_tag(uploaded_two.e_tag().ok_or("movement part 2 omitted ETag")?)
+                        .build(),
+                )
+                .build(),
+        )
+        .send()
+        .await?;
+
+    dist.expand_to_four_pools().await?;
+
+    start_decommission(&dist.cluster, DECOMMISSION_POOL_ID).await?;
+    wait_for_decommission_active(&dist.cluster, DECOMMISSION_POOL_ID, Duration::from_secs(30)).await?;
+    wait_for_decommission_complete(&dist.cluster, DECOMMISSION_POOL_ID, 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");
+    }
+    for (version_id, expected) in [(&version_one_id, &version_one), (&version_two_id, &version_two)] {
+        let got = after_client
+            .get_object()
+            .bucket(&bucket)
+            .key(versioned_key)
+            .version_id(version_id)
+            .send()
+            .await?
+            .body
+            .collect()
+            .await?
+            .into_bytes();
+        assert_eq!(got.as_ref(), expected.as_slice(), "version {version_id} changed after decommission");
+    }
+    let mut expected_multipart = first_part;
+    expected_multipart.extend_from_slice(&second_part);
+    let got_multipart = after_client
+        .get_object()
+        .bucket(&bucket)
+        .key(multipart_key)
+        .send()
+        .await?
+        .body
+        .collect()
+        .await?
+        .into_bytes();
+    assert_eq!(
+        sha256_hex(got_multipart.as_ref()),
+        sha256_hex(&expected_multipart),
+        "multipart checksum changed after decommission"
+    );
+    Ok(())
+}
@@ -0,0 +1,81 @@
+// 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::{
+    DECOMMISSION_POOL_ID, DistCluster, DistLayout, TestResult, assert_inventory, list_pools_json, put_inventory,
+    put_inventory_retrying, start_decommission, start_rebalance, unique_bucket, wait_for_decommission_active,
+    wait_for_decommission_complete, wait_for_rebalance_active, wait_for_rebalance_complete,
+};
+use crate::common::init_logging;
+use std::time::Duration;
+
+#[tokio::test]
+async fn four_node_pool_expand_preserves_objects_then_rebalance() -> TestResult {
+    init_logging();
+    let mut dist = DistCluster::start(DistLayout::SingleNodeFourDrive).await?;
+    let bucket = unique_bucket("expand");
+    dist.create_bucket(&bucket).await?;
+    let client = dist.client(0)?;
+    let inventory = put_inventory(&client, &bucket, 64, 256 * 1024).await?;
+    assert_inventory(&client, &bucket, &inventory).await?;
+
+    for expected_nodes in 2..=4 {
+        let new_node = dist.append_pool_and_restart().await?;
+        assert_eq!(new_node + 1, expected_nodes);
+        assert_inventory(&dist.client(new_node)?, &bucket, &inventory).await?;
+    }
+    assert_eq!(dist.cluster.nodes.len(), 4);
+
+    // Prove that the expanded pool map is durable, and clear any recovery
+    // latch raised while the newly-added pool replicas converged.
+    dist.restart_current_binary_gracefully().await?;
+
+    let after_expand = dist.client(0)?;
+    assert_inventory(&after_expand, &bucket, &inventory).await?;
+    let peer = dist.client(3)?;
+    assert_inventory(&peer, &bucket, &inventory).await?;
+
+    let rebalance_id = start_rebalance(&dist.cluster).await?;
+    wait_for_rebalance_active(&dist.cluster, &rebalance_id, Duration::from_secs(30)).await?;
+    wait_for_rebalance_complete(&dist.cluster, &rebalance_id, Duration::from_secs(180)).await?;
+    assert_inventory(&peer, &bucket, &inventory).await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn four_pool_decommission_moves_objects_without_loss() -> TestResult {
+    init_logging();
+    let mut dist = DistCluster::start(DistLayout::SingleNodeFourDrive).await?;
+    let bucket = unique_bucket("decom");
+    dist.create_bucket(&bucket).await?;
+    let client = dist.client(0)?;
+    let inventory = put_inventory_retrying(&client, &bucket, 96, 128 * 1024, Duration::from_secs(30)).await?;
+    dist.expand_to_four_pools().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))
+        .ok_or_else(|| format!("pool list omitted an array: {pools_before}"))?;
+    assert_eq!(pool_count, 4, "expected exactly four pools before decommission: {pools_before}");
+
+    start_decommission(&dist.cluster, DECOMMISSION_POOL_ID).await?;
+    wait_for_decommission_active(&dist.cluster, DECOMMISSION_POOL_ID, Duration::from_secs(30)).await?;
+    wait_for_decommission_complete(&dist.cluster, DECOMMISSION_POOL_ID, Duration::from_secs(180)).await?;
+
+    let after = dist.client(2)?;
+    assert_inventory(&after, &bucket, &inventory).await?;
+    Ok(())
+}
@@ -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(())
+}
@@ -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;
@@ -0,0 +1,219 @@
+// 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_lock_configuration, 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::primitives::ByteStream;
+use aws_sdk_s3::types::{
+    DefaultRetention, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockLegalHoldStatus, ObjectLockRetentionMode,
+    ObjectLockRule,
+};
+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 deleted_governance = peer
+        .head_object()
+        .bucket(&bucket)
+        .key(governance_key)
+        .version_id(&governance_version)
+        .send()
+        .await
+        .expect_err("GOVERNANCE bypass must remove the retained version");
+    assert_eq!(
+        deleted_governance.raw_response().map(|response| response.status().as_u16()),
+        Some(404),
+        "deleted GOVERNANCE version returned an unexpected HEAD result: {deleted_governance:?}"
+    );
+
+    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(())
+}
+
+#[tokio::test]
+async fn four_node_default_retention_is_visible_and_non_lock_bucket_rejects_configuration() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let writer = dist.client(0)?;
+    let reader = dist.client(3)?;
+    let bucket = unique_bucket("default-lock");
+
+    writer
+        .create_bucket()
+        .bucket(&bucket)
+        .object_lock_enabled_for_bucket(true)
+        .send()
+        .await?;
+    put_object_lock_configuration(&writer, &bucket, ObjectLockRetentionMode::Governance, Some(1), None).await?;
+
+    let key = "default-governance.bin";
+    let put = writer
+        .put_object()
+        .bucket(&bucket)
+        .key(key)
+        .body(ByteStream::from_static(b"default retention payload"))
+        .send()
+        .await?;
+    let version_id = put.version_id().ok_or("default-retained PUT omitted version ID")?;
+
+    let config = reader.get_object_lock_configuration().bucket(&bucket).send().await?;
+    let default_retention = config
+        .object_lock_configuration()
+        .and_then(|configuration| configuration.rule())
+        .and_then(|rule| rule.default_retention())
+        .ok_or("GetObjectLockConfiguration omitted default retention")?;
+    assert_eq!(default_retention.mode().map(|mode| mode.as_str()), Some("GOVERNANCE"));
+    assert_eq!(default_retention.days(), Some(1));
+
+    let retention = reader
+        .get_object_retention()
+        .bucket(&bucket)
+        .key(key)
+        .version_id(version_id)
+        .send()
+        .await?;
+    let retention = retention.retention().ok_or("GetObjectRetention omitted applied retention")?;
+    assert_eq!(retention.mode().map(|mode| mode.as_str()), Some("GOVERNANCE"));
+    let retain_until = retention
+        .retain_until_date()
+        .ok_or("default retention omitted retain-until date")?;
+    assert!(retain_until.secs() > Utc::now().timestamp(), "default retention is not in the future");
+
+    let versioning = reader.get_bucket_versioning().bucket(&bucket).send().await?;
+    assert_eq!(versioning.status().map(|status| status.as_str()), Some("Enabled"));
+    expect_versioned_delete_denied(&reader, &bucket, key, version_id, false, "default GOVERNANCE retention without bypass")
+        .await?;
+
+    let plain_bucket = unique_bucket("no-lock");
+    dist.create_bucket(&plain_bucket).await?;
+    let configuration = ObjectLockConfiguration::builder()
+        .object_lock_enabled(ObjectLockEnabled::Enabled)
+        .rule(
+            ObjectLockRule::builder()
+                .default_retention(
+                    DefaultRetention::builder()
+                        .mode(ObjectLockRetentionMode::Governance)
+                        .days(1)
+                        .build(),
+                )
+                .build(),
+        )
+        .build();
+    let error = writer
+        .put_object_lock_configuration()
+        .bucket(&plain_bucket)
+        .object_lock_configuration(configuration)
+        .send()
+        .await
+        .expect_err("an unversioned bucket must reject Object Lock enablement");
+    let service_error = error
+        .as_service_error()
+        .ok_or("non-lock bucket rejection was not an S3 service error")?;
+    assert_eq!(service_error.code(), Some("InvalidBucketState"), "unexpected error: {error:?}");
+    assert_eq!(
+        service_error.message(),
+        Some("Object Lock configuration cannot be enabled on existing buckets"),
+        "unexpected error: {error:?}"
+    );
+
+    Ok(())
+}
@@ -0,0 +1,236 @@
+// 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, cluster_admin_ok, unique_bucket, wait_for_ready};
+use crate::common::{admin_request, init_logging, local_http_client};
+use aws_sdk_s3::operation::RequestId;
+use aws_sdk_s3::primitives::ByteStream;
+use bytes::Bytes;
+use http::Method;
+use http_body_util::{BodyExt, Empty};
+use hyper::body::Incoming;
+use hyper::service::service_fn;
+use hyper::{Request, Response};
+use hyper_util::rt::TokioIo;
+use local_ip_address::local_ip;
+use rustfs_madmin::metrics::RealtimeMetrics;
+use rustfs_utils::egress::ENV_OUTBOUND_ALLOW_ORIGINS;
+use serde_json::Value;
+use std::convert::Infallible;
+use std::time::Duration;
+use tokio::net::TcpListener;
+use tokio::sync::mpsc;
+use tokio::task::JoinHandle;
+use tokio::time::{Instant, timeout};
+
+async fn spawn_audit_collector() -> TestResult<(String, mpsc::UnboundedReceiver<Value>, JoinHandle<()>)> {
+    let listener = TcpListener::bind("0.0.0.0:0").await?;
+    let endpoint = format!("http://{}/audit", std::net::SocketAddr::new(local_ip()?, listener.local_addr()?.port()));
+    let (tx, rx) = mpsc::unbounded_channel();
+    let handle = tokio::spawn(async move {
+        loop {
+            let Ok((stream, _)) = listener.accept().await else {
+                return;
+            };
+            let tx = tx.clone();
+            tokio::spawn(async move {
+                let service = service_fn(move |request: Request<Incoming>| {
+                    let tx = tx.clone();
+                    async move {
+                        let method = request.method().clone();
+                        if let Ok(body) = request.into_body().collect().await
+                            && method == Method::POST
+                            && let Ok(payload) = serde_json::from_slice::<Value>(&body.to_bytes())
+                        {
+                            if let Some(records) = payload["Records"].as_array() {
+                                for entry in records {
+                                    let _ = tx.send(entry.clone());
+                                }
+                            } else {
+                                let _ = tx.send(payload);
+                            }
+                        }
+                        Ok::<_, Infallible>(Response::new(Empty::<Bytes>::new()))
+                    }
+                });
+                let _ = hyper::server::conn::http1::Builder::new()
+                    .serve_connection(TokioIo::new(stream), service)
+                    .await;
+            });
+        }
+    });
+    Ok((endpoint, rx, handle))
+}
+
+async fn wait_for_audit_entry(
+    rx: &mut mpsc::UnboundedReceiver<Value>,
+    bucket: &str,
+    key: &str,
+    request_id: &str,
+) -> TestResult<Value> {
+    let deadline = Instant::now() + Duration::from_secs(30);
+    let mut seen = Vec::new();
+    loop {
+        let remaining = deadline.saturating_duration_since(Instant::now());
+        if remaining.is_zero() {
+            return Err(format!(
+                "audit webhook did not receive PutObject for {bucket}/{key}; received {} other records: {seen:?}",
+                seen.len()
+            )
+            .into());
+        }
+        let entry = match timeout(remaining, rx.recv()).await {
+            Ok(Some(entry)) => entry,
+            Ok(None) => return Err("audit collector stopped before the expected entry arrived".into()),
+            Err(_) => {
+                return Err(format!(
+                    "audit webhook did not receive PutObject for {bucket}/{key}; received {} other records: {seen:?}",
+                    seen.len()
+                )
+                .into());
+            }
+        };
+        if entry["api"]["name"].as_str() == Some("s3:PutObject")
+            && entry["api"]["bucket"].as_str() == Some(bucket)
+            && entry["api"]["object"].as_str() == Some(key)
+            && entry["requestID"].as_str() == Some(request_id)
+        {
+            return Ok(entry);
+        }
+        if seen.len() < 8 {
+            seen.push(format!(
+                "api={:?} bucket={:?} object={:?} requestID={:?}",
+                entry["api"]["name"].as_str(),
+                entry["api"]["bucket"].as_str(),
+                entry["api"]["object"].as_str(),
+                entry["requestID"].as_str()
+            ));
+        }
+    }
+}
+
+#[tokio::test]
+async fn four_node_health_inventory_metrics_and_audit_delivery_are_consistent() -> TestResult {
+    init_logging();
+    let (audit_endpoint, mut audit_entries, collector) = spawn_audit_collector().await?;
+    let audit_origin = reqwest::Url::parse(&audit_endpoint)?.origin().ascii_serialization();
+    let audit_env = [
+        ("RUSTFS_AUDIT_ENABLE", "true"),
+        ("RUSTFS_AUDIT_WEBHOOK_ENABLE_DISTRIBUTED", "on"),
+        ("RUSTFS_AUDIT_WEBHOOK_ENDPOINT_DISTRIBUTED", audit_endpoint.as_str()),
+        (ENV_OUTBOUND_ALLOW_ORIGINS, audit_origin.as_str()),
+    ];
+    let mut dist = DistCluster::new_stopped_with_env(DistLayout::FourByFour, &audit_env).await?;
+    for node_idx in 0..dist.cluster.nodes.len() {
+        let queue_dir = format!("{}/audit-queue-node-{node_idx}", dist.cluster.temp_dir);
+        tokio::fs::create_dir_all(&queue_dir).await?;
+        dist.cluster
+            .set_node_env(node_idx, "RUSTFS_AUDIT_WEBHOOK_QUEUE_DIR_DISTRIBUTED", queue_dir)?;
+    }
+    dist.cluster.start().await?;
+    wait_for_ready(&dist.cluster).await?;
+
+    let http = local_http_client();
+    for node in &dist.cluster.nodes {
+        for probe in ["ready", "live"] {
+            let response = http.get(format!("{}/health/{probe}", node.url)).send().await?;
+            assert!(
+                response.status().is_success(),
+                "node {} {probe} probe failed: {}",
+                node.address,
+                response.status()
+            );
+        }
+    }
+
+    let info_body = cluster_admin_ok(&dist.cluster, Method::GET, "/rustfs/admin/v3/info", None).await?;
+    let info: Value = serde_json::from_str(&info_body)?;
+    let servers = info["info"]["servers"]
+        .as_array()
+        .ok_or_else(|| format!("admin info omitted servers: {info}"))?;
+    assert_eq!(servers.len(), 4, "admin info did not report all four nodes: {info}");
+
+    let storage_body = cluster_admin_ok(&dist.cluster, Method::GET, "/rustfs/admin/v3/storageinfo", None).await?;
+    let storage: Value = serde_json::from_str(&storage_body)?;
+    let disks = storage["info"]["disks"]
+        .as_array()
+        .ok_or_else(|| format!("storageinfo omitted disks: {storage}"))?;
+    assert_eq!(disks.len(), 16, "storageinfo did not report all sixteen drives: {storage}");
+    assert!(
+        disks.iter().all(|disk| {
+            disk["state"].as_str().is_some_and(|state| state.eq_ignore_ascii_case("ok"))
+                && disk["runtimeState"]
+                    .as_str()
+                    .is_some_and(|state| state.eq_ignore_ascii_case("online"))
+        }),
+        "storageinfo reported a drive that was not healthy and online: {storage}"
+    );
+
+    for (node_idx, node) in dist.cluster.nodes.iter().enumerate() {
+        let (status, metrics_body) = admin_request(
+            &node.url,
+            Method::GET,
+            "/rustfs/admin/v3/metrics?n=1&by-host=true&by-disk=true",
+            None,
+            &dist.cluster.access_key,
+            &dist.cluster.secret_key,
+        )
+        .await?;
+        assert!(status.is_success(), "node {node_idx} metrics failed: {status} {metrics_body}");
+        let sample: RealtimeMetrics = serde_json::from_str(
+            metrics_body
+                .lines()
+                .next()
+                .ok_or_else(|| format!("node {node_idx} returned empty metrics"))?,
+        )?;
+        assert!(sample.finally, "node {node_idx} metrics sample was not terminal");
+        assert!(sample.errors.is_empty(), "node {node_idx} metrics reported errors: {:?}", sample.errors);
+        assert!(!sample.hosts.is_empty(), "node {node_idx} metrics omitted hosts");
+    }
+
+    let targets_body = cluster_admin_ok(&dist.cluster, Method::GET, "/rustfs/admin/v3/audit/target/list", None).await?;
+    let targets: Value = serde_json::from_str(&targets_body)?;
+    let configured = targets["audit_endpoints"]
+        .as_array()
+        .ok_or_else(|| format!("audit target list omitted audit_endpoints: {targets}"))?
+        .iter()
+        .any(|target| target["account_id"].as_str() == Some("distributed") && target["service"].as_str() == Some("webhook"));
+    assert!(configured, "configured audit webhook was missing: {targets}");
+
+    let bucket = unique_bucket("audit");
+    dist.create_bucket(&bucket).await?;
+    let key = "correlated/audit-object.bin";
+    let put = dist
+        .client(2)?
+        .put_object()
+        .bucket(&bucket)
+        .key(key)
+        .body(ByteStream::from_static(b"distributed audit payload"))
+        .send()
+        .await?;
+    let request_id = put.request_id().ok_or("PutObject response omitted request ID")?;
+    let audit = wait_for_audit_entry(&mut audit_entries, &bucket, key, request_id).await?;
+    assert_eq!(
+        audit["api"]["status_code"].as_i64(),
+        Some(200),
+        "audit entry did not report success: {audit}"
+    );
+    assert!(
+        !audit.to_string().contains(&dist.cluster.secret_key),
+        "audit entry leaked the root secret key"
+    );
+
+    collector.abort();
+    Ok(())
+}
@@ -0,0 +1,191 @@
+// 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_ready, wait_for_replicated_bytes, wait_until,
+};
+use crate::common::{FAST_DATA_USAGE_SCANNER_ENV, init_logging};
+use aws_sdk_s3::error::ProvideErrorMetadata;
+use aws_sdk_s3::primitives::ByteStream;
+use http::Method;
+use std::time::Duration;
+
+async fn wait_for_replication_status(
+    client: &aws_sdk_s3::Client,
+    bucket: &str,
+    key: &str,
+    expected: &[&str],
+    timeout: Duration,
+) -> TestResult {
+    wait_until(
+        timeout,
+        || async {
+            let head = client.head_object().bucket(bucket).key(key).send().await?;
+            Ok(head
+                .replication_status()
+                .is_some_and(|status| expected.contains(&status.as_str())))
+        },
+        &format!("replication status for {bucket}/{key} in {expected:?}"),
+    )
+    .await
+}
+
+#[tokio::test]
+async fn four_node_bucket_replication_converges_to_peer_cluster() -> TestResult {
+    init_logging();
+    let (source, mut 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/metadata-and-tags.bin";
+    let body = b"distributed-bucket-replication".to_vec();
+    source_client
+        .put_object()
+        .bucket(&source_bucket)
+        .key(key)
+        .metadata("origin", "four-node-source")
+        .tagging("suite=distributed&shape=metadata")
+        .body(ByteStream::from(body.clone()))
+        .send()
+        .await?;
+    wait_for_replicated_bytes(&target_client, &target_bucket, key, &body, Duration::from_secs(45)).await?;
+    wait_for_replication_status(&source_client, &source_bucket, key, &["COMPLETED"], Duration::from_secs(30)).await?;
+
+    let peer_read = target.client(3)?;
+    wait_for_replicated_bytes(&peer_read, &target_bucket, key, &body, Duration::from_secs(15)).await?;
+    let replica_head = peer_read.head_object().bucket(&target_bucket).key(key).send().await?;
+    assert_eq!(
+        replica_head
+            .metadata()
+            .and_then(|metadata| metadata.get("origin"))
+            .map(String::as_str),
+        Some("four-node-source")
+    );
+    assert_eq!(replica_head.replication_status().map(|status| status.as_str()), Some("REPLICA"));
+    let replica_tags = peer_read.get_object_tagging().bucket(&target_bucket).key(key).send().await?;
+    let tags: std::collections::BTreeMap<_, _> = replica_tags.tag_set().iter().map(|tag| (tag.key(), tag.value())).collect();
+    assert_eq!(tags.get("suite"), Some(&"distributed"));
+    assert_eq!(tags.get("shape"), Some(&"metadata"));
+
+    target.cluster.stop();
+    let outage_key = "replicated/queued-during-target-outage.bin";
+    let outage_body = b"retry-after-target-restart".to_vec();
+    put_object(&source_client, &source_bucket, outage_key, outage_body.clone()).await?;
+    wait_for_replication_status(
+        &source_client,
+        &source_bucket,
+        outage_key,
+        &["PENDING", "FAILED"],
+        Duration::from_secs(30),
+    )
+    .await?;
+
+    target.cluster.start().await?;
+    wait_for_ready(&target.cluster).await?;
+    wait_for_replicated_bytes(&target.client(2)?, &target_bucket, outage_key, &outage_body, Duration::from_secs(90)).await?;
+    wait_for_replication_status(&source_client, &source_bucket, outage_key, &["COMPLETED"], Duration::from_secs(45)).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).map_err(|error| format!("quota stats returned invalid JSON: {error}: {body}"))?;
+            let usage = stats
+                .get("current_usage")
+                .and_then(serde_json::Value::as_u64)
+                .ok_or_else(|| format!("quota stats omitted current_usage: {stats}"))?;
+            Ok(usage >= 1024)
+        },
+        "quota stats observe small object",
+    )
+    .await?;
+
+    let oversized_key = "too-big.bin";
+    let error = client
+        .put_object()
+        .bucket(&bucket)
+        .key(oversized_key)
+        .body(vec![0u8; 16 * 1024].into())
+        .send()
+        .await
+        .expect_err("hard quota must reject the oversized PUT");
+    let service_error = error
+        .as_service_error()
+        .ok_or("quota rejection was not an S3 service error")?;
+    assert_eq!(
+        error.raw_response().map(|response| response.status().as_u16()),
+        Some(400),
+        "quota rejection must be HTTP 400: {error:?}"
+    );
+    assert_eq!(service_error.code(), Some("InvalidRequest"), "unexpected quota error: {error:?}");
+    assert!(
+        service_error
+            .message()
+            .is_some_and(|message| message.starts_with("Bucket quota exceeded")),
+        "PUT must fail specifically at quota admission: {error:?}"
+    );
+
+    let missing = client
+        .head_object()
+        .bucket(&bucket)
+        .key(oversized_key)
+        .send()
+        .await
+        .expect_err("an object rejected by quota must not become visible");
+    assert_eq!(
+        missing.raw_response().map(|response| response.status().as_u16()),
+        Some(404),
+        "quota-rejected object returned an unexpected HEAD result: {missing:?}"
+    );
+
+    let listed = client.list_objects_v2().bucket(&bucket).send().await?;
+    assert!(
+        listed.contents().iter().all(|object| object.key() != Some(oversized_key)),
+        "quota-rejected key leaked into ListObjectsV2"
+    );
+    Ok(())
+}
@@ -0,0 +1,258 @@
+// 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::error::ProvideErrorMetadata;
+use aws_sdk_s3::presigning::PresigningConfig;
+use aws_sdk_s3::primitives::ByteStream;
+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());
+
+    let deleted = 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?;
+    assert!(deleted.errors().is_empty(), "DeleteObjects reported failures: {deleted:?}");
+    assert_eq!(deleted.deleted().len(), 3, "DeleteObjects did not acknowledge every key");
+
+    let remaining = reader.list_objects_v2().bucket(&bucket).send().await?;
+    assert!(remaining.contents().is_empty(), "bucket still has objects after delete");
+    Ok(())
+}
+
+#[tokio::test]
+async fn four_node_s3_metadata_tags_special_keys_pagination_and_multipart_abort() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let bucket = unique_bucket("s3matrix");
+    dist.create_bucket(&bucket).await?;
+    let writer = dist.client(0)?;
+    let reader = dist.client(3)?;
+
+    let special_key = "unicode/测试 space+percent%25.txt";
+    let special_body = b"metadata and tagging survive distributed routing".to_vec();
+    let put = writer
+        .put_object()
+        .bucket(&bucket)
+        .key(special_key)
+        .metadata("test-meta", "distributed")
+        .tagging("purpose=compatibility&scope=four-by-four")
+        .body(ByteStream::from(special_body.clone()))
+        .send()
+        .await?;
+    let etag = put.e_tag().ok_or("PutObject omitted ETag")?.to_string();
+
+    let head = reader.head_object().bucket(&bucket).key(special_key).send().await?;
+    assert_eq!(
+        head.metadata()
+            .and_then(|metadata| metadata.get("test-meta"))
+            .map(String::as_str),
+        Some("distributed")
+    );
+    assert_eq!(head.e_tag(), Some(etag.as_str()));
+    let tags = reader.get_object_tagging().bucket(&bucket).key(special_key).send().await?;
+    let actual_tags: std::collections::BTreeMap<_, _> = tags
+        .tag_set()
+        .iter()
+        .map(|tag| (tag.key().to_string(), tag.value().to_string()))
+        .collect();
+    assert_eq!(actual_tags.get("purpose").map(String::as_str), Some("compatibility"));
+    assert_eq!(actual_tags.get("scope").map(String::as_str), Some("four-by-four"));
+
+    let conditional = reader
+        .get_object()
+        .bucket(&bucket)
+        .key(special_key)
+        .if_match(&etag)
+        .send()
+        .await?;
+    assert_eq!(conditional.body.collect().await?.into_bytes().as_ref(), special_body.as_slice());
+    let invalid_range = reader
+        .get_object()
+        .bucket(&bucket)
+        .key(special_key)
+        .range("bytes=999999-1000000")
+        .send()
+        .await
+        .expect_err("an unsatisfiable range must fail");
+    assert_eq!(
+        invalid_range.as_service_error().and_then(ProvideErrorMetadata::code),
+        Some("InvalidRange"),
+        "unexpected invalid-range error: {invalid_range:?}"
+    );
+
+    let upload_key = "multipart/aborted.bin";
+    let upload = writer
+        .create_multipart_upload()
+        .bucket(&bucket)
+        .key(upload_key)
+        .send()
+        .await?;
+    let upload_id = upload.upload_id().ok_or("CreateMultipartUpload omitted upload ID")?;
+    writer
+        .upload_part()
+        .bucket(&bucket)
+        .key(upload_key)
+        .upload_id(upload_id)
+        .part_number(1)
+        .body(ByteStream::from(vec![0x5Au8; 5 * 1024 * 1024]))
+        .send()
+        .await?;
+    let pending = reader
+        .list_multipart_uploads()
+        .bucket(&bucket)
+        .prefix("multipart/")
+        .send()
+        .await?;
+    assert!(pending.uploads().iter().any(|entry| entry.upload_id() == Some(upload_id)));
+    writer
+        .abort_multipart_upload()
+        .bucket(&bucket)
+        .key(upload_key)
+        .upload_id(upload_id)
+        .send()
+        .await?;
+    let after_abort = reader
+        .list_multipart_uploads()
+        .bucket(&bucket)
+        .prefix("multipart/")
+        .send()
+        .await?;
+    assert!(after_abort.uploads().iter().all(|entry| entry.upload_id() != Some(upload_id)));
+    let aborted_head = reader
+        .head_object()
+        .bucket(&bucket)
+        .key(upload_key)
+        .send()
+        .await
+        .expect_err("aborted multipart upload must not create an object");
+    assert_eq!(
+        aborted_head.raw_response().map(|response| response.status().as_u16()),
+        Some(404),
+        "aborted multipart object returned an unexpected HEAD result: {aborted_head:?}"
+    );
+
+    for index in 0..113 {
+        let key = format!("page/{index:04}.txt");
+        put_object(&writer, &bucket, &key, format!("page-{index}").into_bytes()).await?;
+    }
+    let mut token = None;
+    let mut paged_keys = Vec::new();
+    loop {
+        let page = reader
+            .list_objects_v2()
+            .bucket(&bucket)
+            .prefix("page/")
+            .max_keys(37)
+            .set_continuation_token(token.take())
+            .send()
+            .await?;
+        paged_keys.extend(page.contents().iter().filter_map(|object| object.key().map(str::to_string)));
+        if page.is_truncated() != Some(true) {
+            break;
+        }
+        token = Some(
+            page.next_continuation_token()
+                .ok_or("truncated ListObjectsV2 page omitted next continuation token")?
+                .to_string(),
+        );
+    }
+    assert_eq!(paged_keys.len(), 113);
+    let expected: Vec<_> = (0..113).map(|index| format!("page/{index:04}.txt")).collect();
+    assert_eq!(paged_keys, expected, "pagination lost, duplicated, or reordered keys");
+    Ok(())
+}
@@ -0,0 +1,94 @@
+// 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::{
+    DECOMMISSION_POOL_ID, DistCluster, DistLayout, TestResult, assert_inventory, decommission_running_with_progress,
+    decommission_status_json, put_inventory_retrying, rebalance_running_with_progress, rebalance_status_json,
+    retrying_get_equals, retrying_put, start_decommission, start_rebalance, unique_bucket, wait_for_decommission_complete,
+    wait_for_decommission_running_with_progress, wait_for_rebalance_complete, wait_for_rebalance_running_with_progress,
+};
+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 mut dist = DistCluster::start(DistLayout::SingleNodeFourDrive).await?;
+    let bucket = unique_bucket("s3move");
+    dist.create_bucket(&bucket).await?;
+    let client = dist.client(0)?;
+    let inventory = put_inventory_retrying(&client, &bucket, 96, 256 * 1024, Duration::from_secs(30)).await?;
+    dist.expand_to_four_pools().await?;
+
+    start_decommission(&dist.cluster, DECOMMISSION_POOL_ID).await?;
+    wait_for_decommission_running_with_progress(&dist.cluster, DECOMMISSION_POOL_ID, Duration::from_secs(30)).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"
+    );
+    let status = decommission_status_json(&dist.cluster).await?;
+    if !decommission_running_with_progress(&status, DECOMMISSION_POOL_ID)? {
+        return Err(format!("decommission did not remain active across the S3 operations: {status}").into());
+    }
+
+    wait_for_decommission_complete(&dist.cluster, DECOMMISSION_POOL_ID, Duration::from_secs(180)).await?;
+    assert_inventory(&live, &bucket, &inventory).await?;
+
+    let rebalance_id = start_rebalance(&dist.cluster).await?;
+    wait_for_rebalance_running_with_progress(&dist.cluster, &rebalance_id, Duration::from_secs(30)).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?;
+    let status = rebalance_status_json(&dist.cluster).await?;
+    if !rebalance_running_with_progress(&status, &rebalance_id)? {
+        return Err(format!("rebalance did not remain active across the S3 operations: {status}").into());
+    }
+    wait_for_rebalance_complete(&dist.cluster, &rebalance_id, Duration::from_secs(180)).await?;
+    assert_inventory(&dist.client(1)?, &bucket, &inventory).await?;
+    Ok(())
+}
@@ -0,0 +1,128 @@
+// 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,
+    wait_until,
+};
+use crate::common::{init_logging, signed_request};
+use http::{Method, StatusCode};
+use rustfs_madmin::{PeerSite, ReplicateAddStatus, SiteReplicationInfo, SyncStatus};
+use std::time::Duration;
+
+async fn site_replication_add(
+    cluster: &crate::common::RustFSTestClusterEnvironment,
+    sites: &[PeerSite],
+) -> TestResult<ReplicateAddStatus> {
+    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(serde_json::from_slice(&response.bytes().await?)?)
+}
+
+async fn site_replication_info(cluster: &crate::common::RustFSTestClusterEnvironment) -> TestResult<SiteReplicationInfo> {
+    let body = cluster_admin_ok(cluster, Method::GET, "/rustfs/admin/v3/site-replication/info", None).await?;
+    Ok(serde_json::from_str(&body)?)
+}
+
+async fn wait_for_site_replication_enabled(cluster: &crate::common::RustFSTestClusterEnvironment) -> TestResult {
+    wait_until(
+        Duration::from_secs(30),
+        || async {
+            let info = site_replication_info(cluster).await?;
+            Ok(info.enabled && info.sites.len() == 2 && info.sites.iter().all(|site| site.sync_state == SyncStatus::Enable))
+        },
+        "site replication enabled with two synchronized sites",
+    )
+    .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()
+        },
+    ];
+    let add_status = site_replication_add(&site_a.cluster, &sites).await?;
+    assert!(
+        add_status.success && add_status.err_detail.is_empty() && add_status.initial_sync_error_message.is_empty(),
+        "site replication add reported failure: {add_status:?}"
+    );
+    wait_for_site_replication_enabled(&site_a.cluster).await?;
+    wait_for_site_replication_enabled(&site_b.cluster).await?;
+
+    let info_a = site_replication_info(&site_a.cluster).await?;
+    let remote = info_a
+        .sites
+        .iter()
+        .find(|site| site.name == "site-b")
+        .ok_or_else(|| format!("site A info omitted the configured site-b peer: {info_a:?}"))?;
+    assert_eq!(remote.endpoint, site_b.cluster.nodes[0].url);
+    let deployment_ids: std::collections::BTreeSet<_> = info_a.sites.iter().map(|site| site.deployment_id.as_str()).collect();
+    assert!(
+        deployment_ids.iter().all(|deployment_id| !deployment_id.is_empty()) && deployment_ids.len() == 2,
+        "site peers must have two distinct non-empty deployment IDs: {info_a:?}"
+    );
+    assert!(info_a.retry_stats.is_none(), "site A has pending replication retries: {info_a:?}");
+    assert!(info_a.pending_operation.is_none(), "site A has a pending operation: {info_a:?}");
+
+    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?;
+
+    let reverse_key = "reverse/site-object.bin";
+    let reverse_body = b"site-b-to-site-a".to_vec();
+    put_object(&site_b.client(2)?, &bucket, reverse_key, reverse_body.clone()).await?;
+    wait_for_replicated_bytes(&site_a.client(3)?, &bucket, reverse_key, &reverse_body, Duration::from_secs(60)).await?;
+    Ok(())
+}
@@ -0,0 +1,345 @@
+// 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::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 source_binary() -> TestResult<PathBuf> {
+    let path = std::env::var_os(SOURCE_BINARY_ENV).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 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);
+            let rejected = code == Some("SignatureDoesNotMatch")
+                || code == Some("InvalidAccessKeyId")
+                || code == Some("AccessDenied")
+                || code == Some("InvalidArgument")
+                || error.raw_response().is_some_and(|response| response.status().as_u16() == 403);
+            if !rejected {
+                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(())
+}
@@ -0,0 +1,188 @@
+// 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;
+use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
+
+#[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";
+    let v1_id = writer
+        .put_object()
+        .bucket(&bucket)
+        .key(key)
+        .body(b"v1".to_vec().into())
+        .send()
+        .await?
+        .version_id()
+        .ok_or("v1 PUT omitted version ID")?
+        .to_string();
+    let v2_id = writer
+        .put_object()
+        .bucket(&bucket)
+        .key(key)
+        .body(b"v2".to_vec().into())
+        .send()
+        .await?
+        .version_id()
+        .ok_or("v2 PUT omitted version ID")?
+        .to_string();
+
+    let versions = reader.list_object_versions().bucket(&bucket).prefix(key).send().await?;
+    let matching_versions: Vec<_> = versions
+        .versions()
+        .iter()
+        .filter(|version| version.key() == Some(key))
+        .collect();
+    assert_eq!(matching_versions.len(), 2, "fresh key must have exactly two versions: {versions:?}");
+    assert!(versions.delete_markers().is_empty(), "fresh key unexpectedly has a delete marker");
+    assert!(
+        matching_versions
+            .iter()
+            .any(|version| version.version_id() == Some(v1_id.as_str()) && version.is_latest() != Some(true)),
+        "v1 was not the historical version: {versions:?}"
+    );
+    assert!(
+        matching_versions
+            .iter()
+            .any(|version| version.version_id() == Some(v2_id.as_str()) && version.is_latest() == Some(true)),
+        "v2 was not the latest version: {versions:?}"
+    );
+
+    let latest = get_object_bytes(&reader, &bucket, key).await?;
+    assert_eq!(latest, b"v2");
+
+    let older = reader.get_object().bucket(&bucket).key(key).version_id(&v1_id).send().await?;
+    let older_body = older.body.collect().await?.into_bytes();
+    assert_eq!(older_body.as_ref(), b"v1");
+
+    let deleted = writer.delete_object().bucket(&bucket).key(key).send().await?;
+    assert_eq!(deleted.delete_marker(), Some(true));
+    let marker_id = deleted.version_id().ok_or("DeleteObject omitted delete-marker version ID")?;
+    let after_delete = reader.list_object_versions().bucket(&bucket).prefix(key).send().await?;
+    let matching_markers: Vec<_> = after_delete
+        .delete_markers()
+        .iter()
+        .filter(|marker| marker.key() == Some(key))
+        .collect();
+    assert_eq!(
+        matching_markers.len(),
+        1,
+        "delete marker missing or duplicated after current-version delete: {after_delete:?}"
+    );
+    assert!(
+        matching_markers[0].version_id() == Some(marker_id) && matching_markers[0].is_latest() == Some(true),
+        "DeleteObject response and ListObjectVersions disagree about the marker: {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(&v1_id).send().await?;
+    let restored_body = restored.body.collect().await?.into_bytes();
+    assert_eq!(restored_body.as_ref(), b"v1");
+
+    writer
+        .delete_object()
+        .bucket(&bucket)
+        .key(key)
+        .version_id(marker_id)
+        .send()
+        .await?;
+    assert_eq!(get_object_bytes(&reader, &bucket, key).await?, b"v2");
+    Ok(())
+}
+
+#[tokio::test]
+async fn four_node_versioning_suspension_keeps_one_null_version_and_history() -> TestResult {
+    init_logging();
+    let dist = DistCluster::start(DistLayout::FourByFour).await?;
+    let bucket = unique_bucket("suspend");
+    dist.create_bucket(&bucket).await?;
+    let writer = dist.client(0)?;
+    let reader = dist.client(3)?;
+    enable_versioning(&writer, &bucket).await?;
+
+    let key = "suspended.txt";
+    let original = writer
+        .put_object()
+        .bucket(&bucket)
+        .key(key)
+        .body(b"enabled-history".to_vec().into())
+        .send()
+        .await?
+        .version_id()
+        .ok_or("enabled PUT omitted version ID")?
+        .to_string();
+    writer
+        .put_bucket_versioning()
+        .bucket(&bucket)
+        .versioning_configuration(
+            VersioningConfiguration::builder()
+                .status(BucketVersioningStatus::Suspended)
+                .build(),
+        )
+        .send()
+        .await?;
+
+    put_object(&writer, &bucket, key, b"null-one".to_vec()).await?;
+    put_object(&writer, &bucket, key, b"null-two".to_vec()).await?;
+    assert_eq!(get_object_bytes(&reader, &bucket, key).await?, b"null-two");
+
+    let versions = reader.list_object_versions().bucket(&bucket).prefix(key).send().await?;
+    let matching: Vec<_> = versions
+        .versions()
+        .iter()
+        .filter(|version| version.key() == Some(key))
+        .collect();
+    assert!(matching.iter().any(|version| version.version_id() == Some(original.as_str())));
+    let null_version_count = matching
+        .iter()
+        .filter(|version| {
+            matches!(
+                version.version_id(),
+                None | Some("") | Some("null") | Some("00000000-0000-0000-0000-000000000000")
+            )
+        })
+        .count();
+    assert_eq!(null_version_count, 1, "suspended overwrites must keep one null version: {versions:?}");
+
+    let historical = reader
+        .get_object()
+        .bucket(&bucket)
+        .key(key)
+        .version_id(&original)
+        .send()
+        .await?;
+    assert_eq!(historical.body.collect().await?.into_bytes().as_ref(), b"enabled-history");
+    Ok(())
+}
@@ -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;
@@ -20,9 +20,10 @@
 //! journal (`count_requests`) carries the assertion in every one of them.

 use super::common::{BoxError, OdmTestEnv, RawResponse, SeedObject, start_configured_env};
-use crate::fake_s3_target::Operation;
+use crate::fake_s3_target::{FaultAction, Operation};
 use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
 use bytes::Bytes;
+use futures::{StreamExt, TryStreamExt};
 use std::time::Duration;

 type TestResult = Result<(), BoxError>;
@@ -145,14 +146,38 @@ async fn test_odm_range_burst_overflows_the_pull_queue_without_failing_clients()
    .await?;

    let body = payload(128 * 1024);
+    let blocker = "queue/blocker.bin";
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(blocker, body.clone())]);
+    // The one-chunk range completes immediately; its full background pull
+    // occupies the only slot while the remaining requests fill the queue.
+    env.source.inject_for_key(
+        Operation::GetObject,
+        blocker,
+        FaultAction::SlowSendBody {
+            chunk_bytes: 1024,
+            delay: Duration::from_millis(100),
+        },
+        2,
+    );
+    let response = env
+        .raw_object_request(http::Method::GET, bucket, blocker, &[("range", "bytes=0-1023")])
+        .await?;
+    assert_eq!(response.status, 206);
+    assert_eq!(response.body, body.slice(0..1024));
+    env.wait_for_status_counter(bucket, "/inflight_pulls", 1, SETTLE).await?;
+
    let keys: Vec<String> = (0..REQUESTS).map(|index| format!("queue/object-{index:03}.bin")).collect();
    let seeds: Vec<SeedObject> = keys.iter().map(|key| SeedObject::new(key.clone(), body.clone())).collect();
    env.seed_source(SOURCE_BUCKET, &seeds);

-    let responses: Vec<RawResponse> = futures::future::try_join_all(
+    // Bound source connections below the fixture's limit while still
+    // submitting all 100 requests to the eight-slot background queue.
+    let responses: Vec<RawResponse> = futures::stream::iter(
        keys.iter()
            .map(|key| env.raw_object_request(http::Method::GET, bucket, key, &[("range", "bytes=0-1023")])),
    )
+    .buffered(16)
+    .try_collect()
    .await?;
    for (key, response) in keys.iter().zip(&responses) {
        assert_eq!(response.status, 206, "{key}: {}", String::from_utf8_lossy(&response.body));
@@ -168,6 +193,15 @@ async fn test_odm_range_burst_overflows_the_pull_queue_without_failing_clients()
        .wait_for_status_counter(bucket, "/counters/pull_failures_total/queue_full", 1, SETTLE)
        .await?;
    assert!(queue_full > 0, "a 100-deep burst must overflow an 8-slot queue");
+    let queue_full = usize::try_from(queue_full)?;
+    assert!(queue_full <= REQUESTS);
+    env.wait_for_status_counter(
+        bucket,
+        "/counters/pulled_objects_total/background",
+        u64::try_from(REQUESTS + 1 - queue_full)?,
+        SETTLE,
+    )
+    .await?;

    let ranged_reads: usize = keys.iter().map(|key| source_get_count(&env, key)).sum();
    assert!(
@@ -175,9 +209,6 @@ async fn test_odm_range_burst_overflows_the_pull_queue_without_failing_clients()
        "every reader is served from the source: {ranged_reads} GETs for {REQUESTS} readers"
    );
    let dropped = keys.iter().filter(|key| source_get_count(&env, key) == 1).count();
-    assert!(
-        dropped > 0,
-        "the overflowed keys are the ones with no backfill GET, but every key got one"
-    );
+    assert_eq!(dropped, queue_full, "only overflowed keys remain without a background GET");
    Ok(())
 }
@@ -265,16 +265,13 @@ async fn list_through_rejects_a_tampered_continuation_token() -> TestResult {
    let decoded = String::from_utf8(base64_simd::STANDARD.decode_to_vec(token.as_bytes())?)?;
    assert!(decoded.contains("\"t\":\"odm-list\""), "the merged token is an envelope: {decoded}");

-    let tampered = base64_simd::STANDARD.encode_to_string(decoded.replace("\"v\":1", "\"v\":2").as_bytes());
-    let rejected = env
-        .raw_list_objects_v2(bucket, &format!("continuation-token={tampered}"))
-        .await?;
-    assert_eq!(
-        rejected.status,
-        400,
-        "a bumped token version is a client error: {}",
-        String::from_utf8_lossy(&rejected.body)
-    );
+    let tampered = base64_simd::STANDARD.encode_to_string(decoded.replace("\"v\":1", "\"v\":3").as_bytes());
+    assert_ne!(tampered, token, "the test must change the token version");
+    let query = serde_urlencoded::to_string([("continuation-token", tampered.as_str())])?;
+    let rejected = env.raw_list_objects_v2(bucket, &query).await?;
+    let error_body = String::from_utf8_lossy(&rejected.body);
+    assert_eq!(rejected.status, 400, "a bumped token version is a client error: {}", error_body);
+    assert!(error_body.contains("<Code>InvalidArgument</Code>"), "{error_body}");
    Ok(())
 }

@@ -151,17 +151,15 @@ pub mod bucket {
            BUCKET_CONFIG_PUBLISH_HOOK, BucketConfigPublishHook, BucketMetadataMutationGuard, BucketMetadataSys,
            ObjectLockConfigState, acquire_bucket_metadata_transaction_lock,
            acquire_bucket_metadata_transaction_lock_for_incarnation, acquire_scanner_bucket_incarnation_fence,
-            capture_bucket_metadata_incarnation, delete, delete_if_incarnation, delete_if_incarnation_at,
-            delete_under_transaction_lock, get, get_accelerate_config, get_bucket_policy, get_bucket_policy_raw,
-            get_bucket_targets_config, get_config_from_disk, get_cors_config, get_durability_config,
-            get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config, get_notification_config,
-            get_object_lock_config, get_object_lock_config_state, get_on_demand_migration_config,
+            capture_bucket_metadata_incarnation, delete, delete_if_incarnation, delete_under_transaction_lock, get,
+            get_accelerate_config, get_bucket_policy, get_bucket_policy_raw, get_bucket_targets_config, get_config_from_disk,
+            get_cors_config, get_durability_config, get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config,
+            get_notification_config, get_object_lock_config, get_object_lock_config_state, get_on_demand_migration_config,
            get_on_demand_migration_config_in, get_public_access_block_config, get_quota_config, get_replication_config,
            get_request_payment_config, get_sse_config, get_tagging_config, get_versioning_config, get_website_config,
            init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata, remove_bucket_metadata, set_bucket_metadata,
            update, update_bucket_targets_under_transaction_lock, update_config_with, update_if_incarnation,
-            update_if_incarnation_at, update_quota_if_incarnation, update_quota_if_incarnation_at, update_under_transaction_lock,
-            update_under_transaction_lock_at,
+            update_quota_if_incarnation, update_under_transaction_lock,
        };
        #[cfg(feature = "test-util")]
        pub use crate::bucket::metadata_sys::{ConfigWriteLockProbe, test_support};
@@ -791,22 +791,9 @@ impl BucketMetadata {
        }
    }

-    /// Replace one config payload and stamp its `*_config_updated_at` with the
-    /// local clock. This is the entry for edits that originate here: the
-    /// local write time is the edit's source time.
    pub fn update_config(&mut self, config_file: &str, data: Vec<u8>) -> Result<OffsetDateTime> {
-        self.update_config_at(config_file, data, OffsetDateTime::now_utc())
-    }
+        let updated = OffsetDateTime::now_utc();

-    /// [`Self::update_config`] with an explicit `updated_at` stamp.
-    ///
-    /// For a config replicated from another site the edit's source time is
-    /// the peer's `updated_at`, not the moment it lands here: staleness of
-    /// the next incoming item is judged against the stored stamp, so stamping
-    /// the local apply time would reject a newer source edit that was merely
-    /// delivered late (backlog#2292). Only replication receivers should pass
-    /// a foreign time; local edits keep [`Self::update_config`].
-    pub fn update_config_at(&mut self, config_file: &str, data: Vec<u8>, updated: OffsetDateTime) -> Result<OffsetDateTime> {
        match config_file {
            BUCKET_POLICY_CONFIG => {
                self.policy_config_json = data;
@@ -1538,39 +1525,6 @@ mod test {
        assert_eq!(metadata.bucket_incarnation_id, incarnation);
    }

-    /// backlog#2292: a replicated config is stamped with the source
-    /// `updated_at` it was given, not the local clock, while the plain
-    /// `update_config` entry keeps stamping the local clock.
-    #[test]
-    fn update_config_at_stamps_the_given_time_and_update_config_stamps_now() {
-        let source_time = OffsetDateTime::now_utc() - time::Duration::hours(3);
-        let mut metadata = BucketMetadata::new("source-stamped");
-
-        let stamped = metadata
-            .update_config_at(BUCKET_POLICY_CONFIG, br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec(), source_time)
-            .unwrap();
-        assert_eq!(stamped, source_time);
-        assert_eq!(metadata.policy_config_updated_at, source_time);
-
-        let tagging = b"<Tagging><TagSet><Tag><Key>k</Key><Value>v</Value></Tag></TagSet></Tagging>".to_vec();
-        let stamped = metadata
-            .update_config_at(BUCKET_TAGGING_CONFIG, tagging, source_time)
-            .unwrap();
-        assert_eq!(stamped, source_time);
-        assert_eq!(metadata.tagging_config_updated_at, source_time);
-
-        let before = OffsetDateTime::now_utc();
-        let stamped = metadata
-            .update_config(BUCKET_POLICY_CONFIG, br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec())
-            .unwrap();
-        assert!(stamped >= before, "a local edit is stamped with the local clock");
-        assert_eq!(metadata.policy_config_updated_at, stamped);
-        assert_eq!(
-            metadata.tagging_config_updated_at, source_time,
-            "restamping one config must not move another config's stamp"
-        );
-    }
-
    #[test]
    fn object_locking_requires_lock_metadata_not_plain_versioning() {
        use s3s::dto::ObjectLockEnabled;
@@ -567,32 +567,6 @@ pub async fn update_if_incarnation(
        config_file,
        data,
        Some(expected_incarnation_id),
-        None,
-    ))
-    .await
-}
-
-/// [`update_if_incarnation`] stamping the config with `updated_at` instead of
-/// the local clock.
-///
-/// For a site-replication receiver the edit's source time is the peer's
-/// `updated_at`; persisting it keeps the stored `*_config_updated_at` on the
-/// source clock so the next item's staleness is judged source-time against
-/// source-time (backlog#2292). See [`BucketMetadata::update_config_at`].
-pub async fn update_if_incarnation_at(
-    bucket: &str,
-    config_file: &str,
-    data: Vec<u8>,
-    expected_incarnation_id: Uuid,
-    updated_at: OffsetDateTime,
-) -> Result<OffsetDateTime> {
-    Box::pin(update_with_sys_expected(
-        get_bucket_metadata_sys()?,
-        bucket,
-        config_file,
-        data,
-        Some(expected_incarnation_id),
-        Some(updated_at),
    ))
    .await
 }
@@ -603,30 +577,6 @@ pub async fn delete_if_incarnation(bucket: &str, config_file: &str, expected_inc
        bucket,
        config_file,
        Some(expected_incarnation_id),
-        None,
-    ))
-    .await
-}
-
-/// [`delete_if_incarnation`] stamping the cleared config with `updated_at`
-/// (a replicated deletion's source time) instead of the local clock.
-///
-/// The stamp survives the deletion as the config's `*_config_updated_at`, and
-/// that is what the next incoming item is judged against: a local stamp on
-/// the delete would reject a newer source re-create that was merely delivered
-/// later (backlog#2292). See [`update_if_incarnation_at`].
-pub async fn delete_if_incarnation_at(
-    bucket: &str,
-    config_file: &str,
-    expected_incarnation_id: Uuid,
-    updated_at: OffsetDateTime,
-) -> Result<OffsetDateTime> {
-    Box::pin(delete_with_sys_expected(
-        get_bucket_metadata_sys()?,
-        bucket,
-        config_file,
-        Some(expected_incarnation_id),
-        Some(updated_at),
    ))
    .await
 }
@@ -648,41 +598,34 @@ async fn update_with_sys(
    config_file: &str,
    data: Vec<u8>,
 ) -> Result<OffsetDateTime> {
-    update_with_sys_expected(sys, bucket, config_file, data, None, None).await
+    update_with_sys_expected(sys, bucket, config_file, data, None).await
 }

-/// `updated_at` is the stamp persisted on the config; `None` uses the local
-/// clock (the edit originates here), `Some` carries a replicated edit's
-/// source time (backlog#2292).
 async fn update_with_sys_expected(
    sys: Arc<RwLock<BucketMetadataSys>>,
    bucket: &str,
    config_file: &str,
    data: Vec<u8>,
    expected_incarnation_id: Option<Uuid>,
-    updated_at: Option<OffsetDateTime>,
 ) -> Result<OffsetDateTime> {
    let guard = acquire_config_write_guard_for_incarnation(sys.clone(), bucket, expected_incarnation_id).await?;
-    update_under_config_write_guard(sys, &guard, config_file, data, updated_at).await
+    update_under_config_write_guard(sys, &guard, config_file, data).await
 }

 /// [`delete`] against an explicitly supplied metadata system. See
 /// [`update_with_sys`].
 async fn delete_with_sys(sys: Arc<RwLock<BucketMetadataSys>>, bucket: &str, config_file: &str) -> Result<OffsetDateTime> {
-    delete_with_sys_expected(sys, bucket, config_file, None, None).await
+    delete_with_sys_expected(sys, bucket, config_file, None).await
 }

-/// `updated_at`: `None` stamps the local clock; `Some` persists a replicated
-/// deletion's source time (backlog#2292).
 async fn delete_with_sys_expected(
    sys: Arc<RwLock<BucketMetadataSys>>,
    bucket: &str,
    config_file: &str,
    expected_incarnation_id: Option<Uuid>,
-    updated_at: Option<OffsetDateTime>,
 ) -> Result<OffsetDateTime> {
    let guard = acquire_config_write_guard_for_incarnation(sys.clone(), bucket, expected_incarnation_id).await?;
-    delete_under_config_write_guard(sys, &guard, config_file, updated_at).await
+    delete_under_config_write_guard(sys, &guard, config_file).await
 }

 /// Owns the complete bucket-config mutation fence.
@@ -829,21 +772,7 @@ pub async fn update_under_transaction_lock(
    data: Vec<u8>,
 ) -> Result<OffsetDateTime> {
    guard.ensure_valid(bucket)?;
-    update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data, None).await
-}
-
-/// [`update_under_transaction_lock`] stamping the config with `updated_at`
-/// (a replicated edit's source time) instead of the local clock; see
-/// [`update_if_incarnation_at`] (backlog#2292).
-pub async fn update_under_transaction_lock_at(
-    guard: &BucketMetadataMutationGuard,
-    bucket: &str,
-    config_file: &str,
-    data: Vec<u8>,
-    updated_at: OffsetDateTime,
-) -> Result<OffsetDateTime> {
-    guard.ensure_valid(bucket)?;
-    update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data, Some(updated_at)).await
+    update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data).await
 }

 /// Clear one config file while the caller holds this bucket's transaction lock.
@@ -853,7 +782,7 @@ pub async fn delete_under_transaction_lock(
    config_file: &str,
 ) -> Result<OffsetDateTime> {
    guard.ensure_valid(bucket)?;
-    delete_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, None).await
+    delete_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file).await
 }

 pub async fn update_quota_if_incarnation(
@@ -861,29 +790,6 @@ pub async fn update_quota_if_incarnation(
    data: Vec<u8>,
    expected_incarnation_id: Uuid,
    proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken,
-) -> Result<OffsetDateTime> {
-    update_quota_if_incarnation_stamped(bucket, data, expected_incarnation_id, proof, None).await
-}
-
-/// [`update_quota_if_incarnation`] stamping the quota config with
-/// `updated_at` (a replicated edit's source time) instead of the local
-/// clock; see [`update_if_incarnation_at`] (backlog#2292).
-pub async fn update_quota_if_incarnation_at(
-    bucket: &str,
-    data: Vec<u8>,
-    expected_incarnation_id: Uuid,
-    proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken,
-    updated_at: OffsetDateTime,
-) -> Result<OffsetDateTime> {
-    update_quota_if_incarnation_stamped(bucket, data, expected_incarnation_id, proof, Some(updated_at)).await
-}
-
-async fn update_quota_if_incarnation_stamped(
-    bucket: &str,
-    data: Vec<u8>,
-    expected_incarnation_id: Uuid,
-    proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken,
-    updated_at: Option<OffsetDateTime>,
 ) -> Result<OffsetDateTime> {
    let sys = get_bucket_metadata_sys()?;
    let guard = Box::pin(acquire_config_write_guard_for_incarnation(
@@ -901,7 +807,7 @@ async fn update_quota_if_incarnation_stamped(
            achieved: 0,
        });
    }
-    update_under_config_write_guard(sys, &guard, rustfs_config::QUOTA_CONFIG_FILE, data, updated_at).await
+    update_under_config_write_guard(sys, &guard, rustfs_config::QUOTA_CONFIG_FILE, data).await
 }

 pub async fn update_bucket_targets_under_transaction_lock(
@@ -917,7 +823,6 @@ async fn update_under_config_write_guard(
    guard: &BucketMetadataMutationGuard,
    config_file: &str,
    data: Vec<u8>,
-    updated_at: Option<OffsetDateTime>,
 ) -> Result<OffsetDateTime> {
    guard.ensure_valid(&guard.bucket)?;
    let metadata_sys = sys.read().await.clone();
@@ -929,7 +834,7 @@ async fn update_under_config_write_guard(
            Some(&guard.transaction_guard),
            &guard.bucket,
            "bucket config transaction",
-            metadata_sys.update_checked(&guard.bucket, config_file, data, true, guard.incarnation_id, updated_at),
+            metadata_sys.update_checked(&guard.bucket, config_file, data, true, guard.incarnation_id),
        ),
    )
    .await?;
@@ -941,7 +846,6 @@ async fn delete_under_config_write_guard(
    sys: Arc<RwLock<BucketMetadataSys>>,
    guard: &BucketMetadataMutationGuard,
    config_file: &str,
-    updated_at: Option<OffsetDateTime>,
 ) -> Result<OffsetDateTime> {
    guard.ensure_valid(&guard.bucket)?;
    let metadata_sys = sys.read().await.clone();
@@ -953,7 +857,7 @@ async fn delete_under_config_write_guard(
            Some(&guard.transaction_guard),
            &guard.bucket,
            "bucket config deletion transaction",
-            metadata_sys.update_checked(&guard.bucket, config_file, Vec::new(), false, guard.incarnation_id, updated_at),
+            metadata_sys.update_checked(&guard.bucket, config_file, Vec::new(), false, guard.incarnation_id),
        ),
    )
    .await?;
@@ -1858,17 +1762,15 @@ impl BucketMetadataSys {
    /// `update` and the config read alone). Keep these boxed.
    pub async fn update(&self, bucket: &str, config_file: &str, data: Vec<u8>) -> Result<OffsetDateTime> {
        let incarnation_id = Box::pin(self.get_bucket_incarnation_id(bucket)).await?;
-        Box::pin(self.update_checked(bucket, config_file, data, true, incarnation_id, None)).await
+        Box::pin(self.update_checked(bucket, config_file, data, true, incarnation_id)).await
    }

    pub async fn delete(&self, bucket: &str, config_file: &str) -> Result<OffsetDateTime> {
        let incarnation_id = self.get_bucket_incarnation_id(bucket).await?;
-        self.update_checked(bucket, config_file, Vec::new(), false, incarnation_id, None)
+        self.update_checked(bucket, config_file, Vec::new(), false, incarnation_id)
            .await
    }

-    /// `updated_at`: `None` stamps the local clock; `Some` persists a
-    /// replicated edit's source time (backlog#2292).
    async fn update_checked(
        &self,
        bucket: &str,
@@ -1876,7 +1778,6 @@ impl BucketMetadataSys {
        data: Vec<u8>,
        parse: bool,
        expected_incarnation_id: Uuid,
-        updated_at: Option<OffsetDateTime>,
    ) -> Result<OffsetDateTime> {
        // Load through this system's own store, the one `save` persists to
        // (backlog#1052 S7). Reading from the ambient handle instead made the
@@ -1887,10 +1788,7 @@ impl BucketMetadataSys {
            return Err(Error::BucketNotFound(bucket.to_string()));
        }

-        let updated = match updated_at {
-            Some(updated_at) => bm.update_config_at(config_file, data, updated_at)?,
-            None => bm.update_config(config_file, data)?,
-        };
+        let updated = bm.update_config(config_file, data)?;

        Box::pin(self.save(bm)).await?;

@@ -3857,106 +3755,6 @@ mod tests {
        );
    }

-    /// backlog#2292: the explicit-stamp write path persists the given source
-    /// time as the config's `*_config_updated_at` — through the incarnation
-    /// path and through an already-held transaction guard — and survives a
-    /// reload from disk, while the plain path keeps stamping the local clock.
-    #[tokio::test]
-    async fn explicit_updated_at_is_persisted_as_the_config_stamp() {
-        let (dirs, ecstore) = isolated_store_over_temp_disks().await;
-        let bucket = "source-stamped-config";
-        for dir in &dirs {
-            std::fs::create_dir_all(dir.path().join(bucket)).expect("bucket volume should be created");
-        }
-        let sys = Arc::new(RwLock::new(BucketMetadataSys::new(ecstore)));
-        let source_time = OffsetDateTime::now_utc() - Duration::from_secs(3 * 3600);
-        let policy = br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec();
-        let tagging = b"<Tagging><TagSet><Tag><Key>k</Key><Value>v</Value></Tag></TagSet></Tagging>".to_vec();
-
-        // Incarnation path (`update_if_incarnation_at` minus the ambient lookup).
-        let stamped =
-            update_with_sys_expected(sys.clone(), bucket, BUCKET_POLICY_CONFIG, policy.clone(), None, Some(source_time))
-                .await
-                .expect("source-stamped policy write should persist");
-        assert_eq!(stamped, source_time);
-
-        // Held-guard path (`update_under_transaction_lock_at` minus the ambient lookup).
-        let guard = acquire_config_write_guard(sys.clone(), bucket).await.expect("write guard");
-        let stamped = update_under_config_write_guard(sys.clone(), &guard, BUCKET_TAGGING_CONFIG, tagging, Some(source_time))
-            .await
-            .expect("source-stamped tagging write should persist");
-        drop(guard);
-        assert_eq!(stamped, source_time);
-
-        let metadata_sys = sys.read().await.clone();
-        metadata_sys.metadata_map.write().await.clear();
-        let reloaded = metadata_sys.get_config_from_disk(bucket).await.expect("reload from disk");
-        assert_eq!(reloaded.policy_config_updated_at, source_time);
-        assert_eq!(reloaded.tagging_config_updated_at, source_time);
-
-        // The plain path is unchanged: a local edit is stamped with the local clock.
-        let before = OffsetDateTime::now_utc();
-        let stamped = update_with_sys(sys.clone(), bucket, BUCKET_POLICY_CONFIG, policy)
-            .await
-            .expect("locally stamped policy write should persist");
-        assert!(stamped >= before, "the plain write path must keep stamping the local clock");
-        let reloaded = metadata_sys.get_config_from_disk(bucket).await.expect("reload from disk");
-        assert_eq!(reloaded.policy_config_updated_at, stamped);
-        assert_eq!(
-            reloaded.tagging_config_updated_at, source_time,
-            "an unrelated config keeps its source stamp"
-        );
-    }
-
-    /// backlog#2292: a replicated delete persists the source time as the
-    /// cleared config's `*_config_updated_at`, so the receive-side gate
-    /// (source time against stored stamp) lets a newer source re-create land
-    /// even when the delete was applied later than the re-create's source
-    /// time; the plain delete keeps stamping the local clock.
-    #[tokio::test]
-    async fn explicit_updated_at_is_persisted_by_a_delete() {
-        let (dirs, ecstore) = isolated_store_over_temp_disks().await;
-        let bucket = "source-stamped-delete";
-        for dir in &dirs {
-            std::fs::create_dir_all(dir.path().join(bucket)).expect("bucket volume should be created");
-        }
-        let sys = Arc::new(RwLock::new(BucketMetadataSys::new(ecstore)));
-        let policy = br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec();
-        let created_at = OffsetDateTime::now_utc() - Duration::from_secs(3 * 3600);
-        let deleted_at = created_at + Duration::from_secs(60);
-        let recreated_at = deleted_at + Duration::from_secs(60);
-
-        update_with_sys_expected(sys.clone(), bucket, BUCKET_POLICY_CONFIG, policy.clone(), None, Some(created_at))
-            .await
-            .expect("source-stamped policy write should persist");
-        let stamped = delete_with_sys_expected(sys.clone(), bucket, BUCKET_POLICY_CONFIG, None, Some(deleted_at))
-            .await
-            .expect("source-stamped policy delete should persist");
-        assert_eq!(stamped, deleted_at);
-
-        let metadata_sys = sys.read().await.clone();
-        metadata_sys.metadata_map.write().await.clear();
-        let reloaded = metadata_sys.get_config_from_disk(bucket).await.expect("reload from disk");
-        assert!(reloaded.policy_config_json.is_empty(), "the delete cleared the payload");
-        assert_eq!(reloaded.policy_config_updated_at, deleted_at, "the delete kept the source stamp");
-        assert!(
-            recreated_at >= reloaded.policy_config_updated_at,
-            "a re-create newer than the delete's source time is not stale against the stored stamp"
-        );
-
-        // The plain delete path is unchanged: stamped with the local clock.
-        update_with_sys_expected(sys.clone(), bucket, BUCKET_POLICY_CONFIG, policy, None, Some(recreated_at))
-            .await
-            .expect("re-create should persist");
-        let before = OffsetDateTime::now_utc();
-        let stamped = delete_with_sys_expected(sys.clone(), bucket, BUCKET_POLICY_CONFIG, None, None)
-            .await
-            .expect("locally stamped delete should persist");
-        assert!(stamped >= before, "the plain delete path must keep stamping the local clock");
-        let reloaded = metadata_sys.get_config_from_disk(bucket).await.expect("reload from disk");
-        assert_eq!(reloaded.policy_config_updated_at, stamped);
-    }
-
    /// The load and the persisted write share one write guard, so concurrent
    /// rewrites of the same config compose instead of clobbering each other.
    /// Moving the load outside that guard loses all but the last tag.
@@ -4173,16 +3971,10 @@ mod tests {
        let new_incarnation = store.bucket_incarnation_id_from_disk(bucket).await.unwrap();
        assert_ne!(old_incarnation, new_incarnation);

-        let err = update_with_sys_expected(
-            sys.clone(),
-            bucket,
-            BUCKET_TAGGING_CONFIG,
-            b"<Tagging/>".to_vec(),
-            Some(old_incarnation),
-            None,
-        )
-        .await
-        .expect_err("a request authorized for the deleted incarnation must fail closed");
+        let err =
+            update_with_sys_expected(sys.clone(), bucket, BUCKET_TAGGING_CONFIG, b"<Tagging/>".to_vec(), Some(old_incarnation))
+                .await
+                .expect_err("a request authorized for the deleted incarnation must fail closed");
        assert!(matches!(err, Error::BucketNotFound(name) if name == bucket));

        let persisted = sys.read().await.get_config_from_disk(bucket).await.unwrap();
@@ -4217,7 +4009,7 @@ mod tests {
            }],
        })
        .unwrap();
-        update_under_config_write_guard(sys, &guard, BUCKET_TAGGING_CONFIG, tagging, None)
+        update_under_config_write_guard(sys, &guard, BUCKET_TAGGING_CONFIG, tagging)
            .await
            .unwrap();
        assert!(!delete.is_finished());
@@ -20,9 +20,9 @@ pub use rustfs_replication::{
 pub(crate) use rustfs_replication::{
    ReplicationDeleteSource, ReplicationMultipartPartInput, ReplicationResyncTargetObject, delete_marker_purge_mrf_entry,
    delete_marker_purge_version_id, delete_replication_creates_marker, delete_replication_missing_source_decision,
-    delete_replication_object_opts, delete_replication_target_version_id, heal_uses_delete_replication_path,
-    is_object_lock_denied_delete, is_retryable_delete_replication_head_error, is_version_delete_replication,
-    replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size,
-    replication_multipart_part_plan, replication_single_put_size_error, resync_existing_delete_replication_info,
-    resync_target_for_object, should_retry_delete_marker_purge, single_part_replica_etag_mismatch,
+    delete_replication_object_opts, heal_uses_delete_replication_path, is_object_lock_denied_delete,
+    is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome, replication_etags_match,
+    replication_multipart_complete_actual_size, replication_multipart_part_plan, replication_single_put_size_error,
+    resync_existing_delete_replication_info, resync_target_for_object, should_retry_delete_marker_purge,
+    single_part_replica_etag_mismatch, target_delete_version_id,
 };
@@ -882,20 +882,6 @@ fn reconstructed_heal_delete_info(
 ) -> DeletedObjectReplicationInfo {
    let mut rstate = oi.replication_state();
    rstate.replicate_decision_str = dsc.to_string();
-    // The caller hands us a blank ObjectInfo (the source marker may already be
-    // gone), so the state above carries no target-assigned marker version ids.
-    // Restore them from the journal: `delete_marker_purge_version_id` must hit
-    // the id the target reported, not fall back to the source marker id, which
-    // a target that mints its own ids answers with an idempotent 204 that would
-    // acknowledge the intent while the real marker stays behind (backlog#2290).
-    // The corrupt flag rides along so a refusal stays a refusal after restart.
-    for (arn, version_id) in &entry.target_delete_marker_version_ids {
-        rstate
-            .target_delete_marker_version_ids
-            .entry(arn.clone())
-            .or_insert_with(|| version_id.clone());
-    }
-    rstate.target_delete_marker_version_ids_corrupt |= entry.target_delete_marker_version_ids_corrupt;

    let delete_marker_mtime = entry
        .delete_marker_mtime
@@ -6615,87 +6601,4 @@ mod tests {
            replacement_data
        );
    }
-
-    /// backlog#2290: a delete-marker purge intent that survives a restart
-    /// through the MRF journal addresses the marker version the TARGET
-    /// assigned, exactly as the live watcher does (see the
-    /// `requires_delayed_purge` spawn). The journal carries the per-ARN ids
-    /// (`targetDeleteMarkerVersionIDs`) and replay restores them into the
-    /// reconstructed replication state; without that the replay would fall
-    /// back to the source marker id, which a target that mints its own ids
-    /// answers with an idempotent 204 — the entry would be acknowledged while
-    /// the real marker stayed behind.
-    #[test]
-    fn mrf_delete_marker_purge_replay_preserves_target_assigned_marker_version() {
-        use super::super::replication_object_decision_boundary::{delete_marker_purge_mrf_entry, delete_marker_purge_version_id};
-
-        let arn = "arn:minio:replication::generic-target:photos".to_string();
-        let source_marker = uuid::Uuid::new_v4();
-        let remote_marker = "remote-assigned-marker-version".to_string();
-
-        let live_oi = ObjectInfo {
-            bucket: "photos".to_string(),
-            name: "obj".to_string(),
-            version_id: Some(source_marker),
-            delete_marker: true,
-            ..Default::default()
-        };
-        let mut live_state = live_oi.replication_state();
-        live_state.replicate_decision_str = replicate_decision_for_admitted_targets(std::slice::from_ref(&arn)).to_string();
-        live_state
-            .target_delete_marker_version_ids
-            .insert(arn.clone(), remote_marker.clone());
-        let live = DeletedObjectReplicationInfo {
-            delete_object: ReplicationDeletedObject {
-                object_name: "obj".to_string(),
-                delete_marker: true,
-                delete_marker_version_id: Some(source_marker),
-                replication_state: Some(live_state),
-                ..Default::default()
-            },
-            bucket: "photos".to_string(),
-            ..Default::default()
-        };
-        assert_eq!(
-            delete_marker_purge_version_id(live.delete_object.replication_state.as_ref(), &arn, source_marker),
-            Some(Some(remote_marker.clone())),
-            "the live purge addresses the recorded target version"
-        );
-
-        // Watch window exhausted: persist the intent, restart, replay it.
-        let entry = delete_marker_purge_mrf_entry(&live, vec![arn.clone()]);
-        let replay_oi = ObjectInfo {
-            bucket: entry.bucket.clone(),
-            name: entry.object.clone(),
-            version_id: entry.version_id,
-            delete_marker: entry.delete_marker,
-            ..Default::default()
-        };
-        let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
-        let replayed = reconstructed_heal_delete_info(&entry, &replay_oi, &dsc);
-
-        assert_eq!(
-            delete_marker_purge_version_id(replayed.delete_object.replication_state.as_ref(), &arn, source_marker),
-            Some(Some(remote_marker)),
-            "the MRF replay must address the target-assigned marker version, not source marker {source_marker}"
-        );
-
-        // A refusal (inconsistent recorded ids) must stay a refusal across the
-        // journal round trip instead of degrading into the source-id fallback.
-        let mut refused = live;
-        refused
-            .delete_object
-            .replication_state
-            .as_mut()
-            .expect("state was set above")
-            .target_delete_marker_version_ids_corrupt = true;
-        let entry = delete_marker_purge_mrf_entry(&refused, vec![arn.clone()]);
-        assert!(entry.target_delete_marker_version_ids_corrupt);
-        let replayed = reconstructed_heal_delete_info(&entry, &replay_oi, &dsc);
-        assert_eq!(
-            delete_marker_purge_version_id(replayed.delete_object.replication_state.as_ref(), &arn, source_marker),
-            None,
-            "the MRF replay must keep refusing to guess when the recorded ids were inconsistent"
-        );
-    }
 }
@@ -32,11 +32,11 @@ use super::replication_msgp_boundary::ReplicationMsgpCodec;
 use super::replication_object_config::{ReplicationConfig, get_replication_config, must_replicate};
 use super::replication_object_decision_boundary::{
    MustReplicateOptions, ReplicationMultipartPartInput, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
-    delete_replication_creates_marker, delete_replication_target_version_id, heal_uses_delete_replication_path,
-    is_object_lock_denied_delete, is_retryable_delete_replication_head_error, is_version_delete_replication,
-    replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size,
-    replication_multipart_part_plan, replication_single_put_size_error, resync_existing_delete_replication_info,
-    should_retry_delete_marker_purge, single_part_replica_etag_mismatch,
+    delete_replication_creates_marker, heal_uses_delete_replication_path, is_object_lock_denied_delete,
+    is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome, replication_etags_match,
+    replication_multipart_complete_actual_size, replication_multipart_part_plan, replication_single_put_size_error,
+    resync_existing_delete_replication_info, should_retry_delete_marker_purge, single_part_replica_etag_mismatch,
+    target_delete_version_id,
 };
 use super::replication_queue_boundary::{DeletedObjectReplicationInfo, ReplicationQueueAdmission};
 use super::replication_resync_boundary::ResyncStatusType;
@@ -2051,11 +2051,7 @@ pub(crate) async fn replicate_delete_with_outcome<S: ReplicationStorage>(

    let is_version_purge = is_version_delete_replication(&dobj.delete_object);

-    // The watcher exists to purge a replicated marker once the SOURCE marker
-    // vanishes. A version purge is that purge already (its failures reach the
-    // journal as a purge entry), so it must not spawn a second watcher that
-    // journals a duplicate intent (backlog#2290).
-    let requires_delayed_purge = should_retry_delete_marker_purge(&dobj.delete_object) && !is_version_purge;
+    let requires_delayed_purge = should_retry_delete_marker_purge(&dobj.delete_object);

    let (replication_status, prev_status) = if !is_version_purge {
        (
@@ -2765,6 +2761,12 @@ fn unavailable_delete_target_info(dobj: &DeletedObjectReplicationInfo, arn: &str
 }

 async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo {
+    let version_id = if let Some(version_id) = &dobj.delete_object.delete_marker_version_id {
+        version_id.to_owned()
+    } else {
+        dobj.delete_object.version_id.unwrap_or_default()
+    };
+
    let mut rinfo = dobj
        .delete_object
        .replication_state
@@ -2797,25 +2799,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
        return rinfo;
    }

-    // Purging a replicated delete marker addresses the version the target
-    // assigned (recorded when the marker was created there); see
-    // `delete_replication_target_version_id`. A corrupt record is a failure,
-    // not a guess: the entry stays visible until the metadata is repaired.
-    let Some(version_id) = delete_replication_target_version_id(&dobj.delete_object, &tgt_client.arn) else {
-        warn!(
-            event = EVENT_DELETE_MARKER_PURGE_FAILED,
-            component = LOG_COMPONENT_ECSTORE,
-            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-            bucket = tgt_client.bucket,
-            object = dobj.delete_object.object_name,
-            arn = %tgt_client.arn,
-            reason = "recorded_target_version_inconsistent",
-            "Replicated version purge refused: recorded target delete-marker version metadata is inconsistent"
-        );
-        rinfo.version_purge_status = VersionPurgeStatusType::Failed;
-        rinfo.error = Some("recorded target delete-marker version metadata is inconsistent".to_string());
-        return rinfo;
-    };
+    let version_id = target_delete_version_id(version_id, is_version_purge);

    if dobj.delete_object.delete_marker && dobj.delete_object.delete_marker_version_id.is_some() {
        match head_object_for_worker(
@@ -429,27 +429,6 @@ where
        }
    }

-    /// The cached mapping record for one user or group, looked up in the same
-    /// cache partition `policy_db_set` writes it to (group / STS / regular+service
-    /// user). `None` when no mapping is stored.
-    pub async fn get_mapped_policy_record(&self, name: &str, user_type: UserType, is_group: bool) -> Option<MappedPolicy> {
-        let cache = self.cache.snapshot();
-        if is_group {
-            cache.group_policies.get(name).cloned()
-        } else if user_type == UserType::Sts {
-            cache.sts_policies.get(name).cloned()
-        } else {
-            cache.user_policies.get(name).cloned()
-        }
-    }
-
-    /// The cached group record (members, status, own timestamp) without the
-    /// mapped-policy overlay `get_group_description` applies. `None` when the
-    /// group does not exist.
-    pub async fn get_group_info(&self, name: &str) -> Option<GroupInfo> {
-        self.cache.snapshot().groups.get(name).cloned()
-    }
-
    pub async fn get_policy(&self, name: &str) -> Result<Policy> {
        if name.is_empty() {
            return Err(Error::InvalidArgument);
@@ -555,17 +534,6 @@ where
    }

    pub async fn set_policy(&self, name: &str, policy: Policy) -> Result<OffsetDateTime> {
-        self.set_policy_at(name, policy, OffsetDateTime::now_utc()).await
-    }
-
-    /// [`Self::set_policy`] stamping the document with `updated_at` instead
-    /// of the local clock.
-    ///
-    /// A site-replication receiver passes the edit's source time: the next
-    /// incoming revision is judged against the stored `UpdateDate`, so a
-    /// local stamp would reject a newer source edit that was merely delivered
-    /// later (backlog#2291). The returned stamp is the one persisted.
-    pub async fn set_policy_at(&self, name: &str, policy: Policy, updated_at: OffsetDateTime) -> Result<OffsetDateTime> {
        if name.is_empty() || policy.is_empty() {
            return Err(Error::InvalidArgument);
        }
@@ -576,17 +544,18 @@ where
            .get(name)
            .map(|v| {
                let mut p = v.clone();
-                p.update_at(policy.clone(), updated_at);
+                p.update(policy.clone());
                p
            })
-            .unwrap_or_else(|| PolicyDoc::new_at(policy, updated_at));
+            .unwrap_or_else(|| PolicyDoc::new(policy));

        self.api.save_policy_doc(name, policy_doc.clone()).await?;

-        self.cache
-            .add_or_update_policy_doc(name, &policy_doc, OffsetDateTime::now_utc());
+        let now = OffsetDateTime::now_utc();

-        Ok(updated_at)
+        self.cache.add_or_update_policy_doc(name, &policy_doc, now);
+
+        Ok(now)
    }

    pub async fn list_policies(&self, bucket_name: &str) -> Result<HashMap<String, Policy>> {
@@ -820,12 +789,6 @@ where

    /// create a service account and update cache
    pub async fn add_service_account(&self, cred: Credentials) -> Result<OffsetDateTime> {
-        self.add_service_account_at(cred, OffsetDateTime::now_utc()).await
-    }
-
-    /// [`Self::add_service_account`] stamping the identity with `updated_at`
-    /// instead of the local clock; see [`Self::set_policy_at`] (backlog#2291).
-    pub async fn add_service_account_at(&self, cred: Credentials, updated_at: OffsetDateTime) -> Result<OffsetDateTime> {
        if cred.access_key.is_empty() || cred.parent_user.is_empty() {
            return Err(Error::InvalidArgument);
        }
@@ -837,8 +800,7 @@ where
        }
        drop(cache);

-        let mut u = UserIdentity::new(cred);
-        u.update_at = Some(updated_at);
+        let u = UserIdentity::new(cred);

        self.api
            .save_user_identity(&u.credentials.access_key, UserType::Svc, u.clone(), None)
@@ -846,22 +808,10 @@ where

        self.update_user_with_claims(&u.credentials.access_key, u.clone())?;

-        Ok(updated_at)
+        Ok(OffsetDateTime::now_utc())
    }

    pub async fn update_service_account(&self, name: &str, opts: UpdateServiceAccountOpts) -> Result<OffsetDateTime> {
-        self.update_service_account_at(name, opts, OffsetDateTime::now_utc()).await
-    }
-
-    /// [`Self::update_service_account`] stamping the identity with
-    /// `updated_at` instead of the local clock; see [`Self::set_policy_at`]
-    /// (backlog#2291).
-    pub async fn update_service_account_at(
-        &self,
-        name: &str,
-        opts: UpdateServiceAccountOpts,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
        let _mutation_guard = self.cache.service_account_mutation_lock().lock().await;
        let cache = self.cache.snapshot();
        let Some(ui) = cache.users.get(name).cloned() else {
@@ -908,7 +858,13 @@ where
        }

        if let Some(status) = opts.status {
-            cr.status = account_status_flag(&status).to_owned();
+            match status.as_str() {
+                val if val == AccountStatus::Enabled.as_ref() => cr.status = auth::ACCOUNT_ON.to_owned(),
+                val if val == AccountStatus::Disabled.as_ref() => cr.status = auth::ACCOUNT_OFF.to_owned(),
+                auth::ACCOUNT_ON => cr.status = auth::ACCOUNT_ON.to_owned(),
+                auth::ACCOUNT_OFF => cr.status = auth::ACCOUNT_OFF.to_owned(),
+                _ => cr.status = auth::ACCOUNT_OFF.to_owned(),
+            }
        }

        let mut m: HashMap<String, Value> = if token_without_expiration {
@@ -960,8 +916,8 @@ where

        cr.session_token = jwt_sign(&m, &cr.secret_key)?;

-        let mut u = UserIdentity::new(cr);
-        u.update_at = Some(updated_at);
+        let u = UserIdentity::new(cr);
+        let updated_at = u.update_at.unwrap_or_else(OffsetDateTime::now_utc);
        self.api
            .save_user_identity(&u.credentials.access_key, UserType::Svc, u.clone(), None)
            .await?;
@@ -1193,20 +1149,6 @@ where
        Ok((policies.into_iter().collect(), update_at))
    }
    pub async fn policy_db_set(&self, name: &str, user_type: UserType, is_group: bool, policy: &str) -> Result<OffsetDateTime> {
-        self.policy_db_set_at(name, user_type, is_group, policy, OffsetDateTime::now_utc())
-            .await
-    }
-
-    /// [`Self::policy_db_set`] stamping the mapping with `updated_at` instead
-    /// of the local clock; see [`Self::set_policy_at`] (backlog#2291).
-    pub async fn policy_db_set_at(
-        &self,
-        name: &str,
-        user_type: UserType,
-        is_group: bool,
-        policy: &str,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
        if name.is_empty() {
            return Err(Error::InvalidArgument);
        }
@@ -1226,11 +1168,10 @@ where
                self.cache.delete_user_policy(name, OffsetDateTime::now_utc());
            }

-            return Ok(updated_at);
+            return Ok(OffsetDateTime::now_utc());
        }

-        let mut mp = MappedPolicy::new(policy);
-        mp.update_at = updated_at;
+        let mp = MappedPolicy::new(policy);

        let cache = self.cache.snapshot();
        let policy_docs_cache = Arc::clone(&cache.policy_docs);
@@ -1253,7 +1194,7 @@ where
            self.cache.add_or_update_user_policy(name, &mp, OffsetDateTime::now_utc());
        }

-        Ok(updated_at)
+        Ok(OffsetDateTime::now_utc())
    }

    pub async fn set_temp_user(&self, access_key: &str, cred: &Credentials, policy_name: Option<&str>) -> Result<OffsetDateTime> {
@@ -1450,17 +1391,6 @@ where
    }

    pub async fn add_user(&self, access_key: &str, args: &AddOrUpdateUserReq) -> Result<OffsetDateTime> {
-        self.add_user_at(access_key, args, OffsetDateTime::now_utc()).await
-    }
-
-    /// [`Self::add_user`] stamping the identity with `updated_at` instead of
-    /// the local clock; see [`Self::set_policy_at`] (backlog#2291).
-    pub async fn add_user_at(
-        &self,
-        access_key: &str,
-        args: &AddOrUpdateUserReq,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
        let cache = self.cache.snapshot();
        let users = Arc::clone(&cache.users);
        if let Some(x) = users.get(access_key) {
@@ -1478,13 +1408,12 @@ where
                _ => auth::ACCOUNT_OFF,
            }
        };
-        let mut user_entry = UserIdentity::from(Credentials {
+        let user_entry = UserIdentity::from(Credentials {
            access_key: access_key.to_string(),
            secret_key: args.secret_key.to_string(),
            status: status.to_owned(),
            ..Default::default()
        });
-        user_entry.update_at = Some(updated_at);

        self.api
            .save_user_identity(access_key, UserType::Reg, user_entry.clone(), None)
@@ -1492,7 +1421,7 @@ where

        self.update_user_with_claims(access_key, user_entry)?;

-        Ok(updated_at)
+        Ok(OffsetDateTime::now_utc())
    }

    pub async fn delete_user(&self, access_key: &str, utype: UserType) -> Result<()> {
@@ -1670,17 +1599,6 @@ where
    }

    pub async fn set_user_status(&self, access_key: &str, status: AccountStatus) -> Result<OffsetDateTime> {
-        self.set_user_status_at(access_key, status, OffsetDateTime::now_utc()).await
-    }
-
-    /// [`Self::set_user_status`] stamping the identity with `updated_at`
-    /// instead of the local clock; see [`Self::set_policy_at`] (backlog#2291).
-    pub async fn set_user_status_at(
-        &self,
-        access_key: &str,
-        status: AccountStatus,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
        if access_key.is_empty() {
            return Err(Error::InvalidArgument);
        }
@@ -1707,13 +1625,12 @@ where
            }
        };

-        let mut user_entry = UserIdentity::from(Credentials {
+        let user_entry = UserIdentity::from(Credentials {
            access_key: access_key.to_string(),
            secret_key: u.credentials.secret_key.clone(),
            status: status.to_owned(),
            ..Default::default()
        });
-        user_entry.update_at = Some(updated_at);
        drop(cache);
        drop(users);

@@ -1723,7 +1640,7 @@ where

        self.update_user_with_claims(access_key, user_entry)?;

-        Ok(updated_at)
+        Ok(OffsetDateTime::now_utc())
    }

    fn update_user_with_claims(&self, k: &str, u: UserIdentity) -> Result<()> {
@@ -1759,17 +1676,6 @@ where
    }

    pub async fn add_users_to_group(&self, group: &str, members: Vec<String>) -> Result<OffsetDateTime> {
-        self.add_users_to_group_at(group, members, OffsetDateTime::now_utc()).await
-    }
-
-    /// [`Self::add_users_to_group`] stamping the group with `updated_at`
-    /// instead of the local clock; see [`Self::set_policy_at`] (backlog#2291).
-    pub async fn add_users_to_group_at(
-        &self,
-        group: &str,
-        members: Vec<String>,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
        if group.is_empty() {
            return Err(Error::InvalidArgument);
        }
@@ -1787,10 +1693,6 @@ where
            }
        }

-        // The group's own timestamp moves with every membership or status
-        // change: site replication judges an incoming group item against it
-        // (backlog#2291), so it must reflect the last change, not creation.
-        let now = updated_at;
        let gi = match cache.groups.get(group) {
            Some(res) => {
                let mut gi = res.clone();
@@ -1799,20 +1701,16 @@ where
                uniq_set.extend(members.iter().cloned());

                gi.members = uniq_set.into_iter().collect();
-                gi.update_at = Some(now);
-                gi
-            }
-            None => {
-                let mut gi = GroupInfo::new(members.clone());
-                gi.update_at = Some(now);
                gi
            }
+            None => GroupInfo::new(members.clone()),
        };
        drop(cache);

        self.api.save_group_info(group, gi.clone()).await?;

-        self.cache.with_write_lock(|cache| {
+        let now = self.cache.with_write_lock(|cache| {
+            let now = OffsetDateTime::now_utc();
            cache.add_or_update_group(group, &gi, now);

            let user_group_memberships = Arc::clone(&cache.state().user_group_memberships);
@@ -1821,18 +1719,13 @@ where
                m.insert(group.to_string());
                cache.add_or_update_user_group_membership(member, &m, now);
            });
+            now
        });

        Ok(now)
    }

    pub async fn set_group_status(&self, name: &str, enable: bool) -> Result<OffsetDateTime> {
-        self.set_group_status_at(name, enable, OffsetDateTime::now_utc()).await
-    }
-
-    /// [`Self::set_group_status`] stamping the group with `updated_at` instead
-    /// of the local clock; see [`Self::set_policy_at`] (backlog#2291).
-    pub async fn set_group_status_at(&self, name: &str, enable: bool, updated_at: OffsetDateTime) -> Result<OffsetDateTime> {
        if name.is_empty() {
            return Err(Error::InvalidArgument);
        }
@@ -1850,14 +1743,12 @@ where
        } else {
            gi.status = STATUS_DISABLED.to_owned();
        }
-        let now = updated_at;
-        gi.update_at = Some(now);

        self.api.save_group_info(name, gi.clone()).await?;

-        self.cache.add_or_update_group(name, &gi, now);
+        self.cache.add_or_update_group(name, &gi, OffsetDateTime::now_utc());

-        Ok(now)
+        Ok(OffsetDateTime::now_utc())
    }

    pub async fn get_group_description(&self, name: &str) -> Result<GroupDesc> {
@@ -1927,20 +1818,6 @@ where
        name: &str,
        members: Vec<String>,
        update_cache_only: bool,
-    ) -> Result<OffsetDateTime> {
-        self.remove_members_from_group_at(name, members, update_cache_only, OffsetDateTime::now_utc())
-            .await
-    }
-
-    /// [`Self::remove_members_from_group`] stamping the group with
-    /// `updated_at` instead of the local clock; see [`Self::set_policy_at`]
-    /// (backlog#2291).
-    pub async fn remove_members_from_group_at(
-        &self,
-        name: &str,
-        members: Vec<String>,
-        update_cache_only: bool,
-        updated_at: OffsetDateTime,
    ) -> Result<OffsetDateTime> {
        let cache = self.cache.snapshot();
        let mut gi = cache
@@ -1953,14 +1830,13 @@ where
        let s: HashSet<&String> = HashSet::from_iter(gi.members.iter());
        let d: HashSet<&String> = HashSet::from_iter(members.iter());
        gi.members = s.difference(&d).map(|v| v.to_string()).collect::<Vec<String>>();
-        let now = updated_at;
-        gi.update_at = Some(now);

        if !update_cache_only {
            self.api.save_group_info(name, gi.clone()).await?;
        }

-        self.cache.with_write_lock(|cache| {
+        let now = self.cache.with_write_lock(|cache| {
+            let now = OffsetDateTime::now_utc();
            cache.add_or_update_group(name, &gi, now);

            let user_group_memberships = Arc::clone(&cache.state().user_group_memberships);
@@ -1971,25 +1847,13 @@ where
                    cache.add_or_update_user_group_membership(member, &m, now);
                }
            });
+            now
        });

        Ok(now)
    }

    pub async fn remove_users_from_group(&self, group: &str, members: Vec<String>) -> Result<OffsetDateTime> {
-        self.remove_users_from_group_at(group, members, OffsetDateTime::now_utc())
-            .await
-    }
-
-    /// [`Self::remove_users_from_group`] stamping the group with `updated_at`
-    /// instead of the local clock; a group delete (no members) leaves no
-    /// record and returns the stamp unchanged (backlog#2291).
-    pub async fn remove_users_from_group_at(
-        &self,
-        group: &str,
-        members: Vec<String>,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
        if group.is_empty() {
            return Err(Error::InvalidArgument);
        }
@@ -2038,17 +1902,18 @@ where
                return Err(err);
            }

-            self.cache.with_write_lock(|cache| {
+            let now = self.cache.with_write_lock(|cache| {
                let now = OffsetDateTime::now_utc();
                self.remove_group_from_memberships_map_unlocked(cache, group, now);
                cache.delete_group(group, now);
                cache.delete_group_policy(group, now);
+                now
            });

-            return Ok(updated_at);
+            return Ok(now);
        }

-        self.remove_members_from_group_at(group, members, false, updated_at).await
+        self.remove_members_from_group(group, members, false).await
    }

    fn remove_group_from_memberships_map_unlocked(&self, cache: &mut LockedCache, group: &str, now: OffsetDateTime) {
@@ -2370,19 +2235,6 @@ where
    }
 }

-/// The stored `status` flag for a service-account status given on the admin
-/// or replication wire: the madmin `enabled` / `disabled` words and the stored
-/// `on` / `off` flags are both accepted; anything else disables the account.
-pub(crate) fn account_status_flag(status: &str) -> &'static str {
-    match status {
-        val if val == AccountStatus::Enabled.as_ref() => auth::ACCOUNT_ON,
-        val if val == AccountStatus::Disabled.as_ref() => auth::ACCOUNT_OFF,
-        auth::ACCOUNT_ON => auth::ACCOUNT_ON,
-        auth::ACCOUNT_OFF => auth::ACCOUNT_OFF,
-        _ => auth::ACCOUNT_OFF,
-    }
-}
-
 pub fn get_default_policies() -> HashMap<String, PolicyDoc> {
    let default_policies = &DEFAULT_POLICIES;
    default_policies
@@ -385,14 +385,7 @@ impl<T: Store> IamSys<T> {
    }

    pub async fn set_policy(&self, name: &str, policy: Policy) -> Result<OffsetDateTime> {
-        self.set_policy_at(name, policy, OffsetDateTime::now_utc()).await
-    }
-
-    /// [`Self::set_policy`] stamping the document with `updated_at` (a
-    /// replicated edit's source time) instead of the local clock; see
-    /// `IamCache::set_policy_at` (backlog#2291).
-    pub async fn set_policy_at(&self, name: &str, policy: Policy, updated_at: OffsetDateTime) -> Result<OffsetDateTime> {
-        let updated_at = self.store.set_policy_at(name, policy, updated_at).await?;
+        let updated_at = self.store.set_policy(name, policy).await?;

        if !self.has_watcher() {
            for r in notify_iam_load_policy(name).await {
@@ -650,18 +643,7 @@ impl<T: Store> IamSys<T> {
    }

    pub async fn set_user_status(&self, name: &str, status: rustfs_madmin::AccountStatus) -> Result<OffsetDateTime> {
-        self.set_user_status_at(name, status, OffsetDateTime::now_utc()).await
-    }
-
-    /// [`Self::set_user_status`] stamping the identity with `updated_at` (a
-    /// replicated edit's source time) instead of the local clock (backlog#2291).
-    pub async fn set_user_status_at(
-        &self,
-        name: &str,
-        status: rustfs_madmin::AccountStatus,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
-        let updated_at = self.store.set_user_status_at(name, status, updated_at).await?;
+        let updated_at = self.store.set_user_status(name, status).await?;

        self.notify_for_user(name, false).await;

@@ -673,20 +655,6 @@ impl<T: Store> IamSys<T> {
        parent_user: &str,
        groups: Option<Vec<String>>,
        opts: NewServiceAccountOpts,
-    ) -> Result<(Credentials, OffsetDateTime)> {
-        self.new_service_account_at(parent_user, groups, opts, OffsetDateTime::now_utc())
-            .await
-    }
-
-    /// [`Self::new_service_account`] stamping the identity with `updated_at`
-    /// (a replicated edit's source time) instead of the local clock
-    /// (backlog#2291).
-    pub async fn new_service_account_at(
-        &self,
-        parent_user: &str,
-        groups: Option<Vec<String>>,
-        opts: NewServiceAccountOpts,
-        updated_at: OffsetDateTime,
    ) -> Result<(Credentials, OffsetDateTime)> {
        if parent_user.is_empty() {
            return Err(IamError::InvalidArgument);
@@ -756,18 +724,11 @@ impl<T: Store> IamSys<T> {
        let mut cred = create_new_credentials_with_metadata(&access_key, &secret_key, &m, &secret_key)?;
        cred.parent_user = parent_user.to_owned();
        cred.groups = groups;
-        // The status is part of the created identity: a replicated disabled
-        // account must never exist enabled, not even between a create and a
-        // follow-up status write (backlog#2289).
-        cred.status = opts
-            .status
-            .as_deref()
-            .map_or(ACCOUNT_ON, crate::manager::account_status_flag)
-            .to_owned();
+        cred.status = ACCOUNT_ON.to_owned();
        cred.name = opts.name;
        cred.description = opts.description;

-        let create_at = self.store.add_service_account_at(cred.clone(), updated_at).await?;
+        let create_at = self.store.add_service_account(cred.clone()).await?;

        self.notify_for_service_account(&cred.access_key).await;

@@ -775,23 +736,11 @@ impl<T: Store> IamSys<T> {
    }

    pub async fn update_service_account(&self, name: &str, opts: UpdateServiceAccountOpts) -> Result<OffsetDateTime> {
-        self.update_service_account_at(name, opts, OffsetDateTime::now_utc()).await
-    }
-
-    /// [`Self::update_service_account`] stamping the identity with
-    /// `updated_at` (a replicated edit's source time) instead of the local
-    /// clock (backlog#2291).
-    pub async fn update_service_account_at(
-        &self,
-        name: &str,
-        opts: UpdateServiceAccountOpts,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
        if name == SITE_REPLICATOR_SERVICE_ACCOUNT && !opts.allow_site_replicator_account {
            return Err(IamError::IAMActionNotAllowed);
        }

-        let updated_at = self.store.update_service_account_at(name, opts, updated_at).await?;
+        let updated_at = self.store.update_service_account(name, opts).await?;

        self.notify_for_service_account(name).await;

@@ -991,17 +940,6 @@ impl<T: Store> IamSys<T> {
    }

    pub async fn create_user(&self, access_key: &str, args: &AddOrUpdateUserReq) -> Result<OffsetDateTime> {
-        self.create_user_at(access_key, args, OffsetDateTime::now_utc()).await
-    }
-
-    /// [`Self::create_user`] stamping the identity with `updated_at` (a
-    /// replicated edit's source time) instead of the local clock (backlog#2291).
-    pub async fn create_user_at(
-        &self,
-        access_key: &str,
-        args: &AddOrUpdateUserReq,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
        if !is_access_key_valid(access_key) {
            return Err(IamError::InvalidAccessKeyLength);
        }
@@ -1014,7 +952,7 @@ impl<T: Store> IamSys<T> {
            return Err(IamError::InvalidSecretKeyLength);
        }

-        let updated_at = self.store.add_user_at(access_key, args, updated_at).await?;
+        let updated_at = self.store.add_user(access_key, args).await?;
        self.load_user(access_key, UserType::Reg).await?;

        self.notify_for_user(access_key, false).await;
@@ -1088,21 +1026,10 @@ impl<T: Store> IamSys<T> {
    }

    pub async fn add_users_to_group(&self, group: &str, users: Vec<String>) -> Result<OffsetDateTime> {
-        self.add_users_to_group_at(group, users, OffsetDateTime::now_utc()).await
-    }
-
-    /// [`Self::add_users_to_group`] stamping the group with `updated_at` (a
-    /// replicated edit's source time) instead of the local clock (backlog#2291).
-    pub async fn add_users_to_group_at(
-        &self,
-        group: &str,
-        users: Vec<String>,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
        if contains_reserved_chars(group) {
            return Err(IamError::GroupNameContainsReservedChars);
        }
-        let updated_at = self.store.add_users_to_group_at(group, users, updated_at).await?;
+        let updated_at = self.store.add_users_to_group(group, users).await?;

        self.notify_for_group(group).await;

@@ -1110,19 +1037,7 @@ impl<T: Store> IamSys<T> {
    }

    pub async fn remove_users_from_group(&self, group: &str, users: Vec<String>) -> Result<OffsetDateTime> {
-        self.remove_users_from_group_at(group, users, OffsetDateTime::now_utc()).await
-    }
-
-    /// [`Self::remove_users_from_group`] stamping the group with `updated_at`
-    /// (a replicated edit's source time) instead of the local clock
-    /// (backlog#2291).
-    pub async fn remove_users_from_group_at(
-        &self,
-        group: &str,
-        users: Vec<String>,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
-        let updated_at = self.store.remove_users_from_group_at(group, users, updated_at).await?;
+        let updated_at = self.store.remove_users_from_group(group, users).await?;

        self.notify_for_group(group).await;

@@ -1130,13 +1045,7 @@ impl<T: Store> IamSys<T> {
    }

    pub async fn set_group_status(&self, group: &str, enable: bool) -> Result<OffsetDateTime> {
-        self.set_group_status_at(group, enable, OffsetDateTime::now_utc()).await
-    }
-
-    /// [`Self::set_group_status`] stamping the group with `updated_at` (a
-    /// replicated edit's source time) instead of the local clock (backlog#2291).
-    pub async fn set_group_status_at(&self, group: &str, enable: bool, updated_at: OffsetDateTime) -> Result<OffsetDateTime> {
-        let updated_at = self.store.set_group_status_at(group, enable, updated_at).await?;
+        let updated_at = self.store.set_group_status(group, enable).await?;

        self.notify_for_group(group).await;

@@ -1146,22 +1055,6 @@ impl<T: Store> IamSys<T> {
        self.store.get_group_description(group).await
    }

-    /// The stored group record itself (see `IamCache::get_group_info`).
-    pub async fn get_group_info(&self, group: &str) -> Option<GroupInfo> {
-        self.store.get_group_info(group).await
-    }
-
-    /// The stored policy document, `Error::NoSuchPolicy` when absent.
-    pub async fn get_policy_doc(&self, name: &str) -> Result<PolicyDoc> {
-        self.store.get_policy_doc(name).await
-    }
-
-    /// The stored mapping record for one user or group (see
-    /// `IamCache::get_mapped_policy_record`).
-    pub async fn get_mapped_policy_record(&self, name: &str, user_type: UserType, is_group: bool) -> Option<MappedPolicy> {
-        self.store.get_mapped_policy_record(name, user_type, is_group).await
-    }
-
    pub async fn list_groups_load(&self) -> Result<Vec<String>> {
        self.store.update_groups().await
    }
@@ -1171,24 +1064,7 @@ impl<T: Store> IamSys<T> {
    }

    pub async fn policy_db_set(&self, name: &str, user_type: UserType, is_group: bool, policy: &str) -> Result<OffsetDateTime> {
-        self.policy_db_set_at(name, user_type, is_group, policy, OffsetDateTime::now_utc())
-            .await
-    }
-
-    /// [`Self::policy_db_set`] stamping the mapping with `updated_at` (a
-    /// replicated edit's source time) instead of the local clock (backlog#2291).
-    pub async fn policy_db_set_at(
-        &self,
-        name: &str,
-        user_type: UserType,
-        is_group: bool,
-        policy: &str,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
-        let updated_at = self
-            .store
-            .policy_db_set_at(name, user_type, is_group, policy, updated_at)
-            .await?;
+        let updated_at = self.store.policy_db_set(name, user_type, is_group, policy).await?;

        if !self.has_watcher() {
            for r in notify_iam_load_policy_mapping(name, user_type.to_u64(), is_group).await {
@@ -1970,11 +1846,6 @@ pub struct NewServiceAccountOpts {
    pub expiration: Option<OffsetDateTime>,
    pub allow_site_replicator_account: bool,
    pub claims: Option<HashMap<String, Value>>,
-    /// Status the account is created with (`enabled` / `disabled` or the
-    /// stored `on` / `off` flags); `None` creates it enabled. Site
-    /// replication passes the source account's status so a disabled account
-    /// is never enabled on the peer, not even transiently (backlog#2289).
-    pub status: Option<String>,
 }

 pub struct UpdateServiceAccountOpts {
@@ -45,33 +45,18 @@ pub struct PolicyDoc {

 impl PolicyDoc {
    pub fn new(policy: Policy) -> Self {
-        Self::new_at(policy, OffsetDateTime::now_utc())
-    }
-
-    /// [`Self::new`] with an explicit `UpdateDate` (and `CreateDate`).
-    ///
-    /// A replicated document keeps the edit's source time: the receiver
-    /// judges the next incoming revision against the stored stamp, so a
-    /// local stamp would reject a newer source edit that was merely
-    /// delivered later.
-    pub fn new_at(policy: Policy, at: OffsetDateTime) -> Self {
        Self {
            version: 1,
            policy,
-            create_date: Some(at),
-            update_date: Some(at),
+            create_date: Some(OffsetDateTime::now_utc()),
+            update_date: Some(OffsetDateTime::now_utc()),
        }
    }

    pub fn update(&mut self, policy: Policy) {
-        self.update_at(policy, OffsetDateTime::now_utc());
-    }
-
-    /// [`Self::update`] with an explicit `UpdateDate`; see [`Self::new_at`].
-    pub fn update_at(&mut self, policy: Policy, at: OffsetDateTime) {
        self.version += 1;
        self.policy = policy;
-        self.update_date = Some(at);
+        self.update_date = Some(OffsetDateTime::now_utc());

        if self.create_date.is_none() {
            self.create_date = self.update_date;
@@ -76,21 +76,6 @@ impl ReplicationWorkerOperation for DeletedObjectReplicationInfo {
                .delete_object
                .delete_marker_mtime
                .and_then(|t| i64::try_from(t.unix_timestamp_nanos()).ok()),
-            // Carry the target-assigned marker version ids (and the fail-closed corrupt
-            // flag) into the journal so a purge intent replayed after a restart addresses
-            // the same version the live path did (backlog#2290). Only delete-marker state
-            // ever records these; other deletes serialize an empty map.
-            target_delete_marker_version_ids: self
-                .delete_object
-                .replication_state
-                .as_ref()
-                .map(|state| state.target_delete_marker_version_ids.clone())
-                .unwrap_or_default(),
-            target_delete_marker_version_ids_corrupt: self
-                .delete_object
-                .replication_state
-                .as_ref()
-                .is_some_and(|state| state.target_delete_marker_version_ids_corrupt),
            target_arns: self.admitted_target_arns(),
            force_delete_id: self.delete_object.force_delete_id,
            force_delete_generation: self.delete_object.force_delete_generation,
@@ -253,28 +238,6 @@ pub fn delete_marker_purge_version_id(
    })
 }

-/// The version a delete replication addresses on `arn`, or `None` to refuse.
-///
-/// A version purge whose purged version is a delete marker must address the
-/// marker version the TARGET assigned — the recorded mapping, exactly as the
-/// delayed-purge watcher does. The source-side `DELETE ?versionId=<marker>`
-/// replicates as such a purge, and a generic S3 target answers a DELETE of an
-/// unknown versionId with 204 while keeping its marker, so addressing it by
-/// the source id reported success and left the marker behind (backlog#2290,
-/// R6.1 on the VMs). Nothing recorded falls back to the source-derived id
-/// (id-mirroring peers); a corrupt record refuses, as the watcher does.
-pub fn delete_replication_target_version_id(dobj: &DeletedObject, arn: &str) -> Option<Option<String>> {
-    let is_version_purge = is_version_delete_replication(dobj);
-    if is_version_purge
-        && !dobj.delete_marker
-        && let Some(marker) = dobj.delete_marker_version_id
-    {
-        return delete_marker_purge_version_id(dobj.replication_state.as_ref(), arn, marker);
-    }
-    let source_version = dobj.delete_marker_version_id.or(dobj.version_id).unwrap_or_default();
-    Some(target_delete_version_id(source_version, is_version_purge))
-}
-
 /// Shape an exhausted purge intent as a marker-creation delete entry. Replay
 /// reconstructs it with `delete_marker: true`, finds the source marker gone,
 /// and funnels into the stale-marker branch of `replicate_delete_with_outcome`
@@ -295,9 +258,9 @@ mod tests {

    use super::{
        DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
-        delete_replication_creates_marker, delete_replication_target_version_id, is_object_lock_denied_delete,
-        is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome,
-        resync_existing_delete_replication_info, should_retry_delete_marker_purge, target_delete_version_id,
+        delete_replication_creates_marker, is_object_lock_denied_delete, is_retryable_delete_replication_head_error,
+        is_version_delete_replication, replicate_delete_outcome, resync_existing_delete_replication_info,
+        should_retry_delete_marker_purge, target_delete_version_id,
    };
    use crate::storage_api::DeletedObject;
    use crate::{
@@ -632,76 +595,6 @@ mod tests {
        assert_eq!(entry.retry_count, 0);
        assert_eq!(entry.bucket, "bucket-a");
        assert_eq!(entry.object, "doc.txt");
-        assert!(
-            entry.target_delete_marker_version_ids.is_empty(),
-            "no recorded target marker ids means the journal carries none"
-        );
-        assert!(!entry.target_delete_marker_version_ids_corrupt);
-    }
-
-    /// backlog#2290: a purge intent journaled to MRF must carry the marker
-    /// version ids the targets assigned, plus the fail-closed corrupt flag,
-    /// so a replay after restart addresses the same version the live path did.
-    #[test]
-    fn delete_marker_purge_mrf_entry_carries_target_assigned_marker_versions() {
-        let delete_marker_version_id = Uuid::new_v4();
-        let mut state = ReplicationState::default();
-        state
-            .target_delete_marker_version_ids
-            .insert("arn:a".to_string(), "remote-marker-a".to_string());
-        state
-            .target_delete_marker_version_ids
-            .insert("arn:b".to_string(), "remote-marker-b".to_string());
-        let mut dobj = DeletedObjectReplicationInfo {
-            delete_object: DeletedObject {
-                object_name: "doc.txt".to_string(),
-                delete_marker: false,
-                version_id: Some(Uuid::new_v4()),
-                delete_marker_version_id: Some(delete_marker_version_id),
-                replication_state: Some(state),
-                ..Default::default()
-            },
-            bucket: "bucket-a".to_string(),
-            ..Default::default()
-        };
-
-        let entry = delete_marker_purge_mrf_entry(&dobj, vec!["arn:a".to_string()]);
-        assert_eq!(
-            entry.target_delete_marker_version_ids,
-            HashMap::from([
-                ("arn:a".to_string(), "remote-marker-a".to_string()),
-                ("arn:b".to_string(), "remote-marker-b".to_string()),
-            ]),
-            "every recorded target marker id survives the journal, regardless of the retried ARN subset"
-        );
-        assert!(!entry.target_delete_marker_version_ids_corrupt);
-        assert_eq!(
-            delete_marker_purge_version_id(
-                Some(&ReplicationState {
-                    target_delete_marker_version_ids: entry.target_delete_marker_version_ids,
-                    ..Default::default()
-                }),
-                "arn:a",
-                delete_marker_version_id
-            ),
-            Some(Some("remote-marker-a".to_string()))
-        );
-
-        // The live path refuses to purge on inconsistent metadata and reports the target
-        // as failed; the journaled intent must keep refusing after a restart.
-        dobj.delete_object
-            .replication_state
-            .as_mut()
-            .expect("state was set above")
-            .target_delete_marker_version_ids_corrupt = true;
-        let entry = delete_marker_purge_mrf_entry(&dobj, vec!["arn:a".to_string()]);
-        assert!(entry.target_delete_marker_version_ids_corrupt);
-
-        // A delete without replication state journals an empty map.
-        dobj.delete_object.replication_state = None;
-        let entry = dobj.to_mrf_entry();
-        assert!(entry.target_delete_marker_version_ids.is_empty());
-        assert!(!entry.target_delete_marker_version_ids_corrupt);
    }

    #[test]
@@ -763,57 +656,4 @@ mod tests {
        assert!(!is_object_lock_denied_delete(Some("InternalError"), Some("retention lookup failed")));
        assert!(!is_object_lock_denied_delete(None, Some("legal hold")));
    }
-
-    fn purge_of_marker(marker: Uuid, state: Option<ReplicationState>) -> DeletedObject {
-        DeletedObject {
-            object_name: "obj".to_string(),
-            delete_marker: false,
-            delete_marker_version_id: Some(marker),
-            version_id: None,
-            replication_state: state,
-            ..Default::default()
-        }
-    }
-
-    #[test]
-    fn delete_replication_target_version_id_addresses_recorded_marker_for_purges() {
-        let arn = "arn:minio:replication::generic:photos";
-        let marker = Uuid::new_v4();
-        let mut state = ReplicationState::default();
-        state
-            .target_delete_marker_version_ids
-            .insert(arn.to_string(), "remote-marker".to_string());
-
-        // purge of a replicated marker: the target's own version
-        assert_eq!(
-            delete_replication_target_version_id(&purge_of_marker(marker, Some(state.clone())), arn),
-            Some(Some("remote-marker".to_string()))
-        );
-        // nothing recorded for this arn: the source-derived id (id-mirroring peers)
-        assert_eq!(
-            delete_replication_target_version_id(&purge_of_marker(marker, None), arn),
-            Some(Some(marker.to_string()))
-        );
-        // corrupt record: refuse instead of guessing
-        state.target_delete_marker_version_ids_corrupt = true;
-        assert_eq!(delete_replication_target_version_id(&purge_of_marker(marker, Some(state)), arn), None);
-
-        // marker creation keeps the source id (the target mints its own on a
-        // versionless DELETE; the id only travels in the source header)
-        let creation = DeletedObject {
-            object_name: "obj".to_string(),
-            delete_marker: true,
-            delete_marker_version_id: Some(marker),
-            ..Default::default()
-        };
-        assert_eq!(delete_replication_target_version_id(&creation, arn), Some(Some(marker.to_string())));
-        // plain version purge: the source version id
-        let version = Uuid::new_v4();
-        let purge = DeletedObject {
-            object_name: "obj".to_string(),
-            version_id: Some(version),
-            ..Default::default()
-        };
-        assert_eq!(delete_replication_target_version_id(&purge, arn), Some(Some(version.to_string())));
-    }
 }
@@ -641,26 +641,6 @@ pub struct MrfReplicateEntry {
    #[serde(rename = "deleteMarkerMtime", skip_serializing_if = "Option::is_none", default)]
    pub delete_marker_mtime: Option<i64>,

-    // For delete-marker purge intents: the exact version id each target assigned to the
-    // replicated marker, keyed by target ARN. A generic S3 target mints its own version ids
-    // and answers a DELETE of an unknown id with 204, so a replay that fell back to the source
-    // marker id would be acknowledged while the real marker stayed behind (backlog#2290).
-    // Old files lack this key; default=empty means "unknown" and replay keeps the source-id
-    // fallback it always had.
-    #[serde(rename = "targetDeleteMarkerVersionIDs", skip_serializing_if = "HashMap::is_empty", default)]
-    pub target_delete_marker_version_ids: HashMap<String, String>,
-
-    // Companion to the map above: the source metadata disagreed about the recorded ids when
-    // the intent was journaled, so the live path refused to guess and reported the target as
-    // failed. Replay must keep refusing instead of falling back to the source id. Old files
-    // lack this key; default=false.
-    #[serde(
-        rename = "targetDeleteMarkerVersionIDsCorrupt",
-        skip_serializing_if = "std::ops::Not::not",
-        default
-    )]
-    pub target_delete_marker_version_ids_corrupt: bool,
-
    #[serde(rename = "targetARNs", skip_serializing_if = "Vec::is_empty", default)]
    pub target_arns: Vec<String>,

@@ -41,9 +41,9 @@ pub use config::{
 };
 pub use delete::{
    DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
-    delete_replication_creates_marker, delete_replication_target_version_id, is_object_lock_denied_delete,
-    is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome,
-    resync_existing_delete_replication_info, should_retry_delete_marker_purge, target_delete_version_id,
+    delete_replication_creates_marker, is_object_lock_denied_delete, is_retryable_delete_replication_head_error,
+    is_version_delete_replication, replicate_delete_outcome, resync_existing_delete_replication_info,
+    should_retry_delete_marker_purge, target_delete_version_id,
 };
 pub use filemeta::{
    NULL_VERSION_ID, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING,
@@ -31,13 +31,8 @@ const CAPABILITY_OPERATION_KIND: u64 = 1 << 0;
 const CAPABILITY_TARGET_ARNS: u64 = 1 << 1;
 const CAPABILITY_FORCE_DELETE: u64 = 1 << 2;
 const CAPABILITY_DELETE_MARKER_MTIME: u64 = 1 << 3;
-// Per-ARN target-assigned delete-marker version ids on purge intents (backlog#2290).
-const CAPABILITY_TARGET_DELETE_MARKER_VERSION_IDS: u64 = 1 << 4;
-const MRF_KNOWN_CAPABILITIES: u64 = CAPABILITY_OPERATION_KIND
-    | CAPABILITY_TARGET_ARNS
-    | CAPABILITY_FORCE_DELETE
-    | CAPABILITY_DELETE_MARKER_MTIME
-    | CAPABILITY_TARGET_DELETE_MARKER_VERSION_IDS;
+const MRF_KNOWN_CAPABILITIES: u64 =
+    CAPABILITY_OPERATION_KIND | CAPABILITY_TARGET_ARNS | CAPABILITY_FORCE_DELETE | CAPABILITY_DELETE_MARKER_MTIME;

 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum MrfCapability {
@@ -45,7 +40,6 @@ pub enum MrfCapability {
    TargetArns,
    ForceDelete,
    DeleteMarkerMtime,
-    TargetDeleteMarkerVersionIds,
 }

 impl MrfCapability {
@@ -55,7 +49,6 @@ impl MrfCapability {
            Self::TargetArns => CAPABILITY_TARGET_ARNS,
            Self::ForceDelete => CAPABILITY_FORCE_DELETE,
            Self::DeleteMarkerMtime => CAPABILITY_DELETE_MARKER_MTIME,
-            Self::TargetDeleteMarkerVersionIds => CAPABILITY_TARGET_DELETE_MARKER_VERSION_IDS,
        }
    }
 }
@@ -608,17 +601,9 @@ pub fn decode_mrf_file(data: &[u8]) -> Result<Vec<MrfReplicateEntry>> {
 #[cfg(test)]
 mod tests {
    use super::*;
-    use std::collections::HashMap;
    use uuid::Uuid;

-    // Capability word 31 = OperationKind | TargetArns | ForceDelete | DeleteMarkerMtime |
-    // TargetDeleteMarkerVersionIds (backlog#2290).
    const ENVELOPE_FIXTURE: &[u8] = &[
-        b'M', b'R', b'F', b'E', 1, 0, 1, 0, 1, 0, 0, 0, 31, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 1, 2, 3,
-    ];
-
-    // The envelope a binary from before backlog#2290 writes: same header, capability word 15.
-    const PRE_TARGET_MARKER_IDS_ENVELOPE_FIXTURE: &[u8] = &[
        b'M', b'R', b'F', b'E', 1, 0, 1, 0, 1, 0, 0, 0, 15, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 1, 2, 3,
    ];

@@ -641,8 +626,6 @@ mod tests {
                delete_marker_version_id: None,
                delete_marker: false,
                delete_marker_mtime: None,
-                target_delete_marker_version_ids: HashMap::new(),
-                target_delete_marker_version_ids_corrupt: false,
                target_arns: vec!["arn:target-a".to_string()],
            },
            MrfReplicateEntry {
@@ -659,8 +642,6 @@ mod tests {
                delete_marker_version_id: None,
                delete_marker: false,
                delete_marker_mtime: None,
-                target_delete_marker_version_ids: HashMap::new(),
-                target_delete_marker_version_ids_corrupt: false,
                target_arns: vec!["arn:target-a".to_string(), "arn:target-b".to_string()],
            },
            MrfReplicateEntry {
@@ -677,11 +658,6 @@ mod tests {
                delete_marker_version_id: Some(del_vid),
                delete_marker: true,
                delete_marker_mtime: Some(1_705_312_200_123_456_789),
-                target_delete_marker_version_ids: HashMap::from([
-                    ("arn:target-a".to_string(), "remote-marker-a".to_string()),
-                    ("arn:target-b".to_string(), "remote-marker-b".to_string()),
-                ]),
-                target_delete_marker_version_ids_corrupt: false,
                target_arns: vec!["arn:target-a".to_string()],
            },
        ];
@@ -709,54 +685,6 @@ mod tests {
            Some(1_705_312_200_123_456_789),
            "delete-marker mtime must survive the MRF disk round-trip"
        );
-        assert!(decoded[0].target_delete_marker_version_ids.is_empty());
-        assert!(decoded[1].target_delete_marker_version_ids.is_empty());
-        assert_eq!(
-            decoded[2].target_delete_marker_version_ids,
-            HashMap::from([
-                ("arn:target-a".to_string(), "remote-marker-a".to_string()),
-                ("arn:target-b".to_string(), "remote-marker-b".to_string()),
-            ]),
-            "target-assigned marker version ids must survive the MRF disk round-trip (backlog#2290)"
-        );
-        assert!(!decoded[2].target_delete_marker_version_ids_corrupt);
-    }
-
-    /// backlog#2290: the corrupt flag rides the same journal round trip, and an
-    /// entry that carries neither field encodes exactly as it did before the
-    /// field existed (both keys are skipped when empty/false).
-    #[test]
-    fn mrf_file_round_trips_target_marker_ids_corrupt_flag_and_skips_empty_keys() {
-        let corrupt = MrfReplicateEntry {
-            bucket: "bucket-a".to_string(),
-            object: "delete-a".to_string(),
-            op: MrfOpKind::Delete,
-            delete_marker: true,
-            delete_marker_version_id: Some(Uuid::new_v4()),
-            target_delete_marker_version_ids_corrupt: true,
-            target_arns: vec!["arn:target-a".to_string()],
-            ..Default::default()
-        };
-        let decoded = decode_mrf_file(&encode_mrf_file(std::slice::from_ref(&corrupt)).expect("mrf file should encode"))
-            .expect("mrf file should decode");
-        assert_eq!(decoded, vec![corrupt]);
-        assert!(decoded[0].target_delete_marker_version_ids_corrupt);
-
-        let plain = MrfReplicateEntry {
-            bucket: "bucket-a".to_string(),
-            object: "delete-a".to_string(),
-            op: MrfOpKind::Delete,
-            delete_marker: true,
-            target_arns: vec!["arn:target-a".to_string()],
-            ..Default::default()
-        };
-        let encoded = encode_mrf_file(std::slice::from_ref(&plain)).expect("mrf file should encode");
-        let payload = String::from_utf8_lossy(&encoded);
-        assert!(
-            !payload.contains("targetDeleteMarkerVersionIDs"),
-            "an entry without recorded ids must not grow the new keys: {payload}"
-        );
-        assert_eq!(decode_mrf_file(&encoded).expect("mrf file should decode"), vec![plain]);
    }

    #[test]
@@ -791,99 +719,6 @@ mod tests {
        // Old files lack the deleteMarkerMtime key; it must default to None so replay keeps the
        // pre-#867 fallback to the current time.
        assert_eq!(decoded[0].delete_marker_mtime, None);
-        // Old files also lack the target marker id keys; they must default to an empty map
-        // and a clear corrupt flag so replay keeps the pre-#2290 source-id fallback.
-        assert!(decoded[0].target_delete_marker_version_ids.is_empty());
-        assert!(!decoded[0].target_delete_marker_version_ids_corrupt);
-    }
-
-    /// backlog#2290: a delete-marker entry written by a binary that predates the
-    /// `targetDeleteMarkerVersionIDs` key decodes with an empty map and a clear
-    /// corrupt flag — the exact shape replay handled before the field existed.
-    #[test]
-    fn mrf_pre_target_marker_ids_delete_entry_decodes_with_empty_map() {
-        let marker_version_id = Uuid::new_v4();
-        let mut payload = Vec::new();
-        rmp::encode::write_array_len(&mut payload, 1).expect("array len should encode");
-        rmp::encode::write_map_len(&mut payload, 9).expect("map len should encode");
-        rmp::encode::write_str(&mut payload, "bucket").expect("bucket key should encode");
-        rmp::encode::write_str(&mut payload, "old-bucket").expect("bucket value should encode");
-        rmp::encode::write_str(&mut payload, "object").expect("object key should encode");
-        rmp::encode::write_str(&mut payload, "old-key").expect("object value should encode");
-        rmp::encode::write_str(&mut payload, "retryCount").expect("retry key should encode");
-        rmp::encode::write_i32(&mut payload, 0).expect("retry value should encode");
-        rmp::encode::write_str(&mut payload, "size").expect("size key should encode");
-        rmp::encode::write_i64(&mut payload, 0).expect("size value should encode");
-        rmp::encode::write_str(&mut payload, "op").expect("op key should encode");
-        rmp::encode::write_str(&mut payload, "delete").expect("op value should encode");
-        rmp::encode::write_str(&mut payload, "forceDelete").expect("forceDelete key should encode");
-        rmp::encode::write_bool(&mut payload, false).expect("forceDelete value should encode");
-        rmp::encode::write_str(&mut payload, "deleteMarkerVersionID").expect("marker id key should encode");
-        // Uuid serializes as a 16-byte bin in the MessagePack journal.
-        rmp::encode::write_bin(&mut payload, marker_version_id.as_bytes()).expect("marker id value should encode");
-        rmp::encode::write_str(&mut payload, "deleteMarker").expect("deleteMarker key should encode");
-        rmp::encode::write_bool(&mut payload, true).expect("deleteMarker value should encode");
-        rmp::encode::write_str(&mut payload, "targetARNs").expect("targetARNs key should encode");
-        rmp::encode::write_array_len(&mut payload, 1).expect("targetARNs len should encode");
-        rmp::encode::write_str(&mut payload, "arn:target-a").expect("targetARNs value should encode");
-
-        let mut data = Vec::with_capacity(4 + payload.len());
-        data.extend_from_slice(&MRF_META_FORMAT.to_le_bytes());
-        data.extend_from_slice(&MRF_META_VERSION.to_le_bytes());
-        data.extend_from_slice(&payload);
-
-        let decoded = decode_mrf_file(&data).expect("pre-#2290 delete-marker entry should decode");
-
-        assert_eq!(decoded.len(), 1);
-        assert_eq!(decoded[0].op, MrfOpKind::Delete);
-        assert!(decoded[0].delete_marker);
-        assert_eq!(decoded[0].delete_marker_version_id, Some(marker_version_id));
-        assert_eq!(decoded[0].target_arns, vec!["arn:target-a".to_string()]);
-        assert!(decoded[0].target_delete_marker_version_ids.is_empty());
-        assert!(!decoded[0].target_delete_marker_version_ids_corrupt);
-    }
-
-    /// backlog#2290: the new field is fenced by its own capability bit exactly
-    /// like the earlier optional fields — a reader without the bit refuses an
-    /// envelope that advertises it, while the current reader still accepts the
-    /// pre-#2290 envelope.
-    #[test]
-    fn envelope_target_marker_ids_capability_is_fenced_and_backward_compatible() {
-        assert!(MrfCapabilities::current().contains(MrfCapability::TargetDeleteMarkerVersionIds));
-        assert_eq!(MrfCapabilities::with(MrfCapability::TargetDeleteMarkerVersionIds).bits(), 1 << 4);
-
-        // Old envelope, current reader: accepted, and the negotiated set lacks the new bit.
-        let legacy = MrfEnvelope::decode(PRE_TARGET_MARKER_IDS_ENVELOPE_FIXTURE, MrfProtocolCapabilities::current())
-            .expect("pre-#2290 envelope should decode");
-        assert_eq!(legacy.protocol().capabilities().bits(), 15);
-        assert!(
-            !legacy
-                .protocol()
-                .capabilities()
-                .contains(MrfCapability::TargetDeleteMarkerVersionIds)
-        );
-        assert_eq!(legacy.payload(), &[1, 2, 3]);
-
-        // Current envelope, reader that only knows the pre-#2290 bits: refused.
-        let pre_2290_reader = MrfProtocolCapabilities::new(1, 1, MrfCapabilities::from_bits(15).expect("known bits"));
-        assert_eq!(
-            MrfEnvelope::decode(ENVELOPE_FIXTURE, pre_2290_reader),
-            Err(MrfEnvelopeError::MissingCapabilities {
-                required: 31,
-                available: 15,
-            })
-        );
-
-        // Negotiation with such a peer drops the bit instead of failing.
-        let negotiated = MrfProtocolCapabilities::current()
-            .negotiate(pre_2290_reader)
-            .expect("negotiation with a pre-#2290 peer should succeed");
-        assert!(
-            !negotiated
-                .capabilities()
-                .contains(MrfCapability::TargetDeleteMarkerVersionIds)
-        );
-        assert!(negotiated.capabilities().contains(MrfCapability::DeleteMarkerMtime));
    }

    #[test]
@@ -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/fresh-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 on storage-sensitive PRs and nightly via `e2e-distributed` |
 | 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 |

@@ -63,6 +63,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` | Storage-sensitive PR and 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):
@@ -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` (storage-sensitive PRs + nightly) | `distributed` | fail-closed 4-node 4-disk S3, durability, replication, movement, fault, and direct/rolling upgrade gate; JUnit, membership listing, per-node 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 |
@@ -0,0 +1,75 @@
+# 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 |
+| 1 single-node pool × 4 drives, then `append_single_node_pool` three times | expansion seed | Pool expand, then decommission / rebalance / integrity |
+
+A multi-pool layout in which any pool spans several localhost ports is not expressible (`RUSTFS_VOLUMES` host ellipses would collide on disk paths). Multi-host striped expansion pools remain the hardware functional-chain / backlog #1313 / #1314 lane.
+
+Data-movement cases fail closed. A decommission or rebalance test must observe a successful start response, an active state, a clean terminal state, non-zero movement counters, and post-operation object integrity. An unsupported response, HTTP 5xx, missing status fields, cleanup warning, or zero-progress terminal response fails the case; pre/post S3 availability alone is not evidence that movement ran.
+
+The four expansion pools must report independent capacity. Four directories on one runner filesystem all return the same `statfs` totals, so RustFS correctly concludes that no pool is less free than the cluster average and performs no rebalance. The Actions job mounts four isolated ext4 loopback filesystems and exports their absolute paths through `RUSTFS_E2E_POOL_ROOTS`. The harness rejects missing, duplicate, relative, nonexistent, or same-device roots instead of allowing a vacuous movement pass. Planned pool additions stop every process with SIGTERM; hard process termination remains a chaos-only fault. After the fourth pool joins, the harness performs one full graceful persistent restart: this proves the expanded pool map survives restart and ensures movement begins only after every replica can load the converged metadata.
+
+The expansion fixture is an all-current-binary fleet, so it initializes pool metadata with the documented V3 write and fleet-confirmation gates. Decommission cases write their baseline objects, version history, and multipart data into pool 0 before adding pools 1–3, then retire pool 0. This makes a passing result evidence of user-data movement rather than merely an internal-metadata counter changing.
+
+## What this lane covers
+
+`cargo nextest run --profile e2e-distributed -p e2e_test` selects `distributed::*`:
+
+- S3 put / get / head / list / copy / rename / delete / presign, range and conditional reads, special keys, metadata, tags, pagination, empty objects, multipart complete and abort
+- Object Lock COMPLIANCE, GOVERNANCE and bypass, legal hold, bucket default retention, and non-lock bucket rejection
+- Versioning, exact historical reads, delete-marker removal, and suspended null-version overwrite semantics
+- Bucket replication between two 4-node clusters, including metadata/tags and target-outage retry; hard quota admission and absence of rejected keys
+- Ready/live probes on every node, exact 4-server/16-disk inventory, realtime metrics on every node, and correlated audit-webhook delivery
+- Pool expand, decommission, rebalance, checksum integrity, versioned and multipart data, and S3 during active movement
+- Bidirectional site-replication convergence plus enabled/synchronized peer state on both sites
+- A 24-worker mixed PUT/HEAD/GET/COPY/DELETE workload; concurrent PUT during active decommission
+- Node kill/restart, full process restart, node-facing TCP blackhole/recovery, in-flight streaming GETs across a peer kill, and fresh-drive replacement verified by physical `xl.meta`/part-shard census
+- Multipart, cross-node listing, list-buckets agreement
+- 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` | Most selected cases are single-node; distributed modules are intentionally owned by this serialized lane |
+| `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, physical NIC pull, authenticated inter-node partition, firmware/media errors, and replacement-server provisioning still belong on the hardware validation VMs. This lane provides deterministic process kill, fresh local-volume replacement, and node-facing TCP blackhole analogues; it does not claim physical fault certification.
+
+## Run
+
+```bash
+cargo build -p rustfs --bins
+# Expansion/decommission/rebalance cases require four paths on distinct filesystems.
+export RUSTFS_E2E_POOL_ROOTS=/mnt/rustfs-pool-0:/mnt/rustfs-pool-1:/mnt/rustfs-pool-2:/mnt/rustfs-pool-3
+# 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. Without four distinct `RUSTFS_E2E_POOL_ROOTS`, the expansion and data-movement cases fail closed. Filter upgrades 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 the Linux and Darwin entries with `python3 ./scripts/check_test_wiring.py --update-profile e2e-distributed <listing.json> <platform>` after adding or renaming a case.
@@ -1131,7 +1131,6 @@ impl Operation for ImportIam {
                        expiration: req.expiration,
                        allow_site_replicator_account: false,
                        claims: Some(req.claims),
-                        status: None,
                    };

                    let groups = if req.groups.is_empty() { None } else { Some(req.groups) };
@@ -325,25 +325,6 @@ pub(crate) mod metadata_sys {
        super::ecstore_bucket::metadata_sys::update_if_incarnation(bucket, config_file, data, expected_incarnation_id).await
    }

-    /// [`update_if_incarnation`] stamping the config with a replicated edit's
-    /// source `updated_at` instead of the local clock (backlog#2292).
-    pub(crate) async fn update_if_incarnation_at(
-        bucket: &str,
-        config_file: &str,
-        data: Vec<u8>,
-        expected_incarnation_id: uuid::Uuid,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
-        super::ecstore_bucket::metadata_sys::update_if_incarnation_at(
-            bucket,
-            config_file,
-            data,
-            expected_incarnation_id,
-            updated_at,
-        )
-        .await
-    }
-
    pub(crate) async fn update_quota_if_incarnation(
        bucket: &str,
        data: Vec<u8>,
@@ -353,25 +334,6 @@ pub(crate) mod metadata_sys {
        super::ecstore_bucket::metadata_sys::update_quota_if_incarnation(bucket, data, expected_incarnation_id, proof).await
    }

-    /// [`update_quota_if_incarnation`] stamping the quota with a replicated
-    /// edit's source `updated_at` instead of the local clock (backlog#2292).
-    pub(crate) async fn update_quota_if_incarnation_at(
-        bucket: &str,
-        data: Vec<u8>,
-        expected_incarnation_id: uuid::Uuid,
-        proof: &super::ecstore_notification::CrossPoolFenceFleetProofToken,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
-        super::ecstore_bucket::metadata_sys::update_quota_if_incarnation_at(
-            bucket,
-            data,
-            expected_incarnation_id,
-            proof,
-            updated_at,
-        )
-        .await
-    }
-
    pub(crate) async fn capture_bucket_metadata_incarnation(bucket: &str) -> Result<uuid::Uuid> {
        super::ecstore_bucket::metadata_sys::capture_bucket_metadata_incarnation(bucket).await
    }
@@ -426,18 +388,6 @@ pub(crate) mod metadata_sys {
        super::ecstore_bucket::metadata_sys::delete_if_incarnation(bucket, config_file, expected_incarnation_id).await
    }

-    /// [`delete_if_incarnation`] stamping the cleared config with a replicated
-    /// deletion's source `updated_at` instead of the local clock (backlog#2292).
-    pub(crate) async fn delete_if_incarnation_at(
-        bucket: &str,
-        config_file: &str,
-        expected_incarnation_id: uuid::Uuid,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
-        super::ecstore_bucket::metadata_sys::delete_if_incarnation_at(bucket, config_file, expected_incarnation_id, updated_at)
-            .await
-    }
-
    pub(crate) async fn get_bucket_policy(bucket: &str) -> Result<(BucketPolicy, OffsetDateTime)> {
        super::ecstore_bucket::metadata_sys::get_bucket_policy(bucket).await
    }
@@ -712,8 +712,6 @@ pub(crate) mod bucket {
                    delete_marker_version_id: None,
                    delete_marker: false,
                    delete_marker_mtime: None,
-                    target_delete_marker_version_ids: Default::default(),
-                    target_delete_marker_version_ids_corrupt: false,
                    target_arns,
                    force_delete_id: Some(operation_id),
                    force_delete_generation: Some(i64::try_from(generation.unix_timestamp_nanos()).unwrap_or(i64::MAX)),
@@ -302,164 +302,7 @@ pub(crate) fn site_replication_state_replicates_ilm_expiry(state: &SiteReplicati
    state.peers.values().any(|peer| peer.replicate_ilm_expiry)
 }

-/// Secret-bearing half of the IAM snapshot. `SRInfo` is served to admin
-/// callers (`site-replication/info`, status, add preflight) and must stay
-/// secret-free, so the bootstrap plan receives credentials through this
-/// separate value, built only on the paths that deliver to peers (site add
-/// bootstrap, repair, retry snapshot resend). Never persisted, never served.
-#[derive(Debug, Clone, Default)]
-pub(crate) struct SiteReplicationIamCredentials {
-    /// Built-in users (access key -> credential); temp and service accounts
-    /// are excluded, external/IdP users never appear here.
-    pub(crate) users: BTreeMap<String, SiteReplicationUserCredential>,
-    /// Every service account except the site replicator's own, already
-    /// shaped as the `service-account` create item the live hook emits.
-    pub(crate) service_accounts: Vec<SiteReplicationServiceAccountSnapshot>,
-}
-
-#[derive(Debug, Clone)]
-pub(crate) struct SiteReplicationUserCredential {
-    pub(crate) secret_key: String,
-    pub(crate) status: AccountStatus,
-    /// The user record's own update time (the axis the receiver's staleness
-    /// check compares against), unlike `UserInfo::updated_at` which
-    /// `list_users` overwrites with the policy mapping's time.
-    pub(crate) updated_at: Option<OffsetDateTime>,
-}
-
-#[derive(Debug, Clone)]
-pub(crate) struct SiteReplicationServiceAccountSnapshot {
-    pub(crate) create: SRSvcAccCreate,
-    pub(crate) envelope: Option<SRSvcAccReplicationEnvelope>,
-    pub(crate) updated_at: Option<OffsetDateTime>,
-}
-
-pub(crate) const SERVICE_ACCOUNT_ENVELOPE_VERSION: u64 = 2;
-
-pub(crate) fn encode_service_account_replication_policy(
-    claims: &HashMap<String, Value>,
-    session_policy: Option<&str>,
-) -> S3Result<(SRSessionPolicy, Option<SRSvcAccReplicationEnvelope>)> {
-    if !claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM) {
-        return session_policy
-            .map(SRSessionPolicy::from_json)
-            .transpose()
-            .map(|policy| policy.unwrap_or_default())
-            .map(|policy| (policy, None))
-            .map_err(|err| s3_error!(InvalidArgument, "marshal policy failed: {:?}", err));
-    }
-
-    let policy = match session_policy {
-        Some(policy) => serde_json::from_str::<Policy>(policy)
-            .map_err(|err| s3_error!(InvalidArgument, "invalid service account replication policy: {:?}", err))?,
-        None => Policy::default(),
-    };
-    if policy.statements.is_empty() && (!policy.id.is_empty() || !policy.version.is_empty())
-        || policy.version.is_empty() && !policy.statements.is_empty()
-    {
-        return Err(s3_error!(InvalidArgument, "service account replication policy is not normalized"));
-    }
-    let policy = serde_json::to_string(&policy)
-        .map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?;
-    let policy = SRSessionPolicy::from_json(&policy)
-        .map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?;
-    Ok((
-        policy,
-        Some(SRSvcAccReplicationEnvelope {
-            version: SERVICE_ACCOUNT_ENVELOPE_VERSION,
-        }),
-    ))
-}
-
-/// Read the credentials the IAM snapshot needs straight from the IAM store:
-/// `list_users` deliberately strips secret keys and skips service accounts,
-/// which is right for an admin listing and wrong for a peer snapshot (the
-/// plan builder used to drop every user for lack of a secret, so a status
-/// change or secret rotation committed while a peer was unreachable never
-/// reached it — backlog#2289).
-pub(crate) async fn build_sr_iam_credentials() -> S3Result<SiteReplicationIamCredentials> {
-    let mut credentials = SiteReplicationIamCredentials::default();
-    let Some(iam_sys) = current_iam_handle() else {
-        return Ok(credentials);
-    };
-
-    let mut users = HashMap::new();
-    iam_sys.load_users(UserType::Reg, &mut users).await.map_err(ApiError::from)?;
-    for (access_key, identity) in users {
-        if identity.credentials.is_temp() || identity.credentials.is_service_account() {
-            continue;
-        }
-        credentials.users.insert(
-            access_key,
-            SiteReplicationUserCredential {
-                secret_key: identity.credentials.secret_key,
-                status: if identity.credentials.status == "off" {
-                    AccountStatus::Disabled
-                } else {
-                    AccountStatus::Enabled
-                },
-                updated_at: identity.update_at,
-            },
-        );
-    }
-
-    let mut service_accounts = HashMap::new();
-    iam_sys
-        .load_users(UserType::Svc, &mut service_accounts)
-        .await
-        .map_err(ApiError::from)?;
-    let mut service_accounts: Vec<_> = service_accounts.into_iter().collect();
-    service_accounts.sort_by(|(a, _), (b, _)| a.cmp(b));
-    for (access_key, identity) in service_accounts {
-        // The replicator account is installed by join / rotate, never by a snapshot.
-        if access_key == SITE_REPLICATOR_SERVICE_ACCOUNT || !identity.credentials.is_service_account() {
-            continue;
-        }
-        let claims = iam_sys.get_claims_for_svc_acc(&access_key).await.map_err(ApiError::from)?;
-        let (account, session_policy) = iam_sys.get_service_account(&access_key).await.map_err(ApiError::from)?;
-        let session_policy = session_policy
-            .map(|policy| serde_json::to_string(&policy))
-            .transpose()
-            .map_err(|err| {
-                S3Error::with_message(
-                    S3ErrorCode::InternalError,
-                    format!("marshal service account session policy failed: {err:?}"),
-                )
-            })?;
-        let (session_policy, envelope) = encode_service_account_replication_policy(&claims, session_policy.as_deref())?;
-        credentials.service_accounts.push(SiteReplicationServiceAccountSnapshot {
-            create: SRSvcAccCreate {
-                parent: identity.credentials.parent_user,
-                access_key,
-                secret_key: identity.credentials.secret_key,
-                groups: identity.credentials.groups.unwrap_or_default(),
-                claims,
-                session_policy,
-                status: identity.credentials.status,
-                name: account.name.unwrap_or_default(),
-                description: account.description.unwrap_or_default(),
-                expiration: account.expiration,
-                api_version: Some(SITE_REPL_API_VERSION.to_string()),
-            },
-            envelope,
-            updated_at: identity.update_at,
-        });
-    }
-
-    Ok(credentials)
-}
-
-/// The bootstrap plan for peer delivery: `info` (secret-free) plus the IAM
-/// credentials read at this moment.
-pub(crate) async fn build_site_replication_bootstrap_plan(info: &SRInfo) -> S3Result<SiteReplicationBootstrapPlan> {
-    let credentials = build_sr_iam_credentials().await?;
-    site_replication_bootstrap_plan(info, &credentials)
-}
-
-pub(crate) fn site_replication_bootstrap_plan(
-    info: &SRInfo,
-    credentials: &SiteReplicationIamCredentials,
-) -> S3Result<SiteReplicationBootstrapPlan> {
+pub(crate) fn site_replication_bootstrap_plan(info: &SRInfo) -> S3Result<SiteReplicationBootstrapPlan> {
    let mut plan = SiteReplicationBootstrapPlan::default();
    let replicate_ilm_expiry = site_replication_info_replicates_ilm_expiry(info);

@@ -475,57 +318,24 @@ pub(crate) fn site_replication_bootstrap_plan(
    }

    for (access_key, user) in &info.user_info_map {
-        // Credentials come from the store snapshot; an inline `secret_key` on
-        // the SRInfo entry (older callers, tests) is accepted as a fallback.
-        // Users with neither (external / IdP identities) have nothing a peer
-        // could install and are skipped.
-        let credential = credentials.users.get(access_key);
-        let Some(secret_key) = credential
-            .map(|credential| credential.secret_key.clone())
-            .or_else(|| user.secret_key.clone())
-            .filter(|secret_key| !secret_key.is_empty())
-        else {
-            continue;
-        };
-        let status = credential
-            .map(|credential| credential.status.clone())
-            .unwrap_or_else(|| user.status.clone());
-        let updated_at = credential.and_then(|credential| credential.updated_at).or(user.updated_at);
-        plan.iam_items.push(SRIAMItem {
-            r#type: "iam-user".to_string(),
-            iam_user: Some(rustfs_madmin::SRIAMUser {
-                access_key: access_key.clone(),
-                is_delete_req: false,
-                user_req: Some(AddOrUpdateUserReq {
-                    secret_key,
-                    policy: user.policy_name.clone(),
-                    status,
+        if let Some(secret_key) = &user.secret_key {
+            plan.iam_items.push(SRIAMItem {
+                r#type: "iam-user".to_string(),
+                iam_user: Some(rustfs_madmin::SRIAMUser {
+                    access_key: access_key.clone(),
+                    is_delete_req: false,
+                    user_req: Some(AddOrUpdateUserReq {
+                        secret_key: secret_key.clone(),
+                        policy: user.policy_name.clone(),
+                        status: user.status.clone(),
+                    }),
+                    api_version: Some(SITE_REPL_API_VERSION.to_string()),
                }),
-                api_version: Some(SITE_REPL_API_VERSION.to_string()),
-            }),
-            updated_at,
-            api_version: Some(SITE_REPL_API_VERSION.to_string()),
-            ..Default::default()
-        });
-    }
-
-    // Service accounts follow their parents: the receiver creates a missing
-    // account under `parent` and updates an existing one (secret, status,
-    // session policy), so a rotation or disable committed during an outage
-    // converges through the same snapshot as users do.
-    for account in &credentials.service_accounts {
-        plan.iam_items.push(SRIAMItem {
-            r#type: "service-account".to_string(),
-            svc_acc_change: Some(SRSvcAccChange {
-                create: Some(account.create.clone()),
-                oidc_service_account_envelope: account.envelope.clone(),
+                updated_at: user.updated_at,
                api_version: Some(SITE_REPL_API_VERSION.to_string()),
                ..Default::default()
-            }),
-            updated_at: account.updated_at,
-            api_version: Some(SITE_REPL_API_VERSION.to_string()),
-            ..Default::default()
-        });
+            });
+        }
    }

    for (name, desc) in &info.group_desc_map {
@@ -708,12 +518,7 @@ pub(crate) async fn broadcast_site_replication_make_bucket(
    } else {
        path
    };
-    // Both steps run to completion on their own: the broadcast attempts every
-    // peer and reports the first failure (backlog#2293), so stopping here on
-    // that error would skip `configure-replication` for the peers whose
-    // `make` just succeeded — and nothing records a retry for that gap. The
-    // failed peer's retry events cover both steps independently.
-    let make_result = broadcast_site_replication_json_using_runtime(runtime, &path, &serde_json::json!({})).await;
+    broadcast_site_replication_json_using_runtime(runtime, &path, &serde_json::json!({})).await?;

    let configure_path = bootstrap_bucket_op_path(bucket, "configure-replication");
    let configure_path = if let Some(token) = bootstrap_token {
@@ -721,8 +526,7 @@ pub(crate) async fn broadcast_site_replication_make_bucket(
    } else {
        configure_path
    };
-    let configure_result = broadcast_site_replication_json_using_runtime(runtime, &configure_path, &serde_json::json!({})).await;
-    make_result.and(configure_result)
+    broadcast_site_replication_json_using_runtime(runtime, &configure_path, &serde_json::json!({})).await
 }

 const SITE_REPLICATION_DELETE_INTENT_PENDING: &str =
@@ -1028,21 +832,6 @@ pub async fn site_replication_iam_change_hook(item: SRIAMItem) -> S3Result<()> {
    let Some(runtime) = runtime_site_replication_targets().await? else {
        return Ok(());
    };
-    // A local revoke must out-rank a stale grant a peer delivers later, so its
-    // mark is committed before the broadcast (backlog#2291). The broadcast
-    // still goes out when the mark cannot be persisted: the peers' own records
-    // remain the primary gate, the mark only covers the deleted case.
-    if let Err(err) = record_iam_deletion_marks_for_item(&item).await {
-        warn!(
-            component = LOG_COMPONENT_ADMIN,
-            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
-            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
-            item_type = %item.r#type,
-            result = "iam_deletion_mark_not_recorded",
-            error = ?err,
-            "failed to record local IAM deletion mark before broadcast"
-        );
-    }
    let mut first_error: Option<S3Error> = None;
    for peer in runtime.state.peers.values() {
        if peer.deployment_id == runtime.local_peer.deployment_id
@@ -79,16 +79,13 @@ use http::header::{CONTENT_TYPE, HOST};
 use http::{HeaderMap, HeaderValue, Uri};
 use hyper::{Method, StatusCode};
 use rustfs_config::{DEFAULT_CONSOLE_ADDRESS, DEFAULT_RUSTFS_TLS_PATH, ENV_RUSTFS_CONSOLE_ADDRESS, ENV_RUSTFS_TLS_PATH};
-use rustfs_iam::federation::OIDC_VIRTUAL_PARENT_CLAIM;
 use rustfs_iam::store::{MappedPolicy, UserType, sr_wire_user_type};
 use rustfs_iam::sys::SITE_REPLICATOR_SERVICE_ACCOUNT;
 use rustfs_madmin::{
-    AccountStatus, AddOrUpdateUserReq, GroupAddRemove, GroupStatus, PeerInfo, PeerSite, ReplicateEditStatus,
-    SITE_REPL_API_VERSION, SRBucketInfo, SRBucketMeta, SRGroupInfo, SRIAMItem, SRIAMPolicy, SRInfo, SRPolicyMapping, SRRemoveReq,
-    SRResyncOpStatus, SRRetryStats, SRSessionPolicy, SRStateInfo, SRSvcAccChange, SRSvcAccCreate, SRSvcAccDelete,
-    SRSvcAccReplicationEnvelope, SyncStatus,
+    AddOrUpdateUserReq, GroupAddRemove, GroupStatus, PeerInfo, PeerSite, ReplicateEditStatus, SITE_REPL_API_VERSION,
+    SRBucketInfo, SRBucketMeta, SRGroupInfo, SRIAMItem, SRIAMPolicy, SRInfo, SRPolicyMapping, SRRemoveReq, SRResyncOpStatus,
+    SRRetryStats, SRStateInfo, SyncStatus,
 };
-use rustfs_policy::policy::Policy;
 use rustfs_signer::constants::UNSIGNED_PAYLOAD;
 use rustfs_signer::sign_v4;
 use rustfs_tls_runtime::{GlobalPublishedOutboundTlsState, TlsGeneration};
@@ -110,26 +107,6 @@ use tracing::{info, warn};
 use url::{Url, form_urlencoded};
 use uuid::Uuid;

-/// Serialize `value` with every JSON object's keys sorted, for hashing and
-/// equality checks. `HashMap` fields (service-account claims) iterate in a
-/// per-instance random order and `serde_json` is built with `preserve_order`,
-/// so two identical plans would otherwise hash differently: the repair
-/// preflight token went stale between dry-run and execute, and a retry
-/// snapshot resend never looked "stable" (backlog#2289 follow-up).
-pub(crate) fn canonical_json_vec<T: Serialize>(value: &T) -> serde_json::Result<Vec<u8>> {
-    fn sort_keys(value: Value) -> Value {
-        match value {
-            Value::Object(map) => {
-                let sorted: BTreeMap<String, Value> = map.into_iter().map(|(key, value)| (key, sort_keys(value))).collect();
-                Value::Object(sorted.into_iter().collect())
-            }
-            Value::Array(items) => Value::Array(items.into_iter().map(sort_keys).collect()),
-            other => other,
-        }
-    }
-    serde_json::to_vec(&sort_keys(serde_json::to_value(value)?))
-}
-
 pub(crate) const LOG_COMPONENT_ADMIN: &str = "admin";

 pub(crate) const LOG_SUBSYSTEM_SITE_REPLICATION: &str = "site_replication";
@@ -234,9 +234,9 @@ impl SiteReplicationRepairTask<'_> {

    pub(crate) fn id(&self) -> S3Result<String> {
        let payload = match self {
-            Self::Iam(item) => canonical_json_vec(item),
+            Self::Iam(item) => serde_json::to_vec(item),
            Self::BucketMake(_) | Self::Replication(_) => serde_json::to_vec(&serde_json::json!({})),
-            Self::BucketMetadata(item) => canonical_json_vec(item),
+            Self::BucketMetadata(item) => serde_json::to_vec(item),
        }
        .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize repair task failed: {err}")))?;
        let mut digest = Sha256::new();
@@ -726,7 +726,7 @@ pub(crate) async fn execute_site_replication_repair_locked(
        return Err(s3_error!(InvalidRequest, "site replication is not configured"));
    }
    let info = build_sr_info(&state, &request.local_peer).await?;
-    let plan = build_site_replication_bootstrap_plan(&info).await?;
+    let plan = site_replication_bootstrap_plan(&info)?;
    let plan_token = site_replication_repair_plan_token(&state, &plan)?;
    let preflight_token = site_replication_repair_preflight_token(&state, &plan, request.signing_key.as_bytes())?;
    let sites = site_replication_repair_sites(&state, &request.local_peer, &plan, request.signing_key.as_bytes())?;
@@ -397,12 +397,12 @@ pub(crate) fn iam_deletion_replay_matches(record: &SiteReplicationIamDeletionRep
 /// newer revision of one another.
 pub(crate) fn iam_item_deletion_entity(item: &SRIAMItem) -> Option<String> {
    match item.r#type.as_str() {
-        "policy" if item.policy.is_none() => Some(iam_policy_deletion_mark_entity(&item.name)),
+        "policy" if item.policy.is_none() => Some(format!("policy:{}", item.name)),
        "iam-user" => item
            .iam_user
            .as_ref()
            .filter(|user| user.is_delete_req)
-            .map(|user| iam_user_deletion_mark_entity(&user.access_key)),
+            .map(|user| format!("iam-user:{}", user.access_key)),
        "group-info" => item
            .group_info
            .as_ref()
@@ -416,7 +416,7 @@ pub(crate) fn iam_item_deletion_entity(item: &SRIAMItem) -> Option<String> {
            .policy_mapping
            .as_ref()
            .filter(|mapping| mapping.policy.is_empty())
-            .map(|mapping| iam_policy_mapping_deletion_mark_entity(&mapping.user_or_group, mapping.user_type, mapping.is_group)),
+            .map(|mapping| format!("policy-mapping:{}:{}:{}", mapping.user_or_group, mapping.user_type, mapping.is_group)),
        "service-account" => item
            .svc_acc_change
            .as_ref()
@@ -426,82 +426,6 @@ pub(crate) fn iam_item_deletion_entity(item: &SRIAMItem) -> Option<String> {
    }
 }

-/// The entities whose deletion a deletion-shaped IAM item commits, keyed the
-/// way the receive-side staleness gate looks them up once the local record is
-/// gone (backlog#2291); empty for creates and updates. Group member removal
-/// yields one entity per removed member so a stale re-add of that member can
-/// be judged, and a group delete (no members) yields the group itself.
-pub(crate) fn iam_item_deletion_mark_entities(item: &SRIAMItem) -> Vec<String> {
-    if item.r#type == "group-info" {
-        let Some(update) = item
-            .group_info
-            .as_ref()
-            .map(|group| &group.update_req)
-            .filter(|update| update.is_remove)
-        else {
-            return Vec::new();
-        };
-        if update.members.is_empty() {
-            return vec![iam_group_deletion_mark_entity(&update.group)];
-        }
-        return update
-            .members
-            .iter()
-            .map(|member| iam_group_member_deletion_mark_entity(&update.group, member))
-            .collect();
-    }
-    iam_item_deletion_entity(item).into_iter().collect()
-}
-
-pub(crate) fn iam_policy_deletion_mark_entity(name: &str) -> String {
-    format!("policy:{name}")
-}
-
-pub(crate) fn iam_user_deletion_mark_entity(access_key: &str) -> String {
-    format!("iam-user:{access_key}")
-}
-
-/// `user_type` is the SR wire integer, as carried by the item on both sides.
-pub(crate) fn iam_policy_mapping_deletion_mark_entity(user_or_group: &str, user_type: i64, is_group: bool) -> String {
-    format!("policy-mapping:{user_or_group}:{user_type}:{is_group}")
-}
-
-pub(crate) fn iam_group_deletion_mark_entity(group: &str) -> String {
-    format!("group:{group}")
-}
-
-pub(crate) fn iam_group_member_deletion_mark_entity(group: &str, member: &str) -> String {
-    format!("group-member:{group}:{member}")
-}
-
-/// Persist the deletion marks of `item` (its source `updated_at` per entity
-/// of [`iam_item_deletion_mark_entities`]) through the state transaction.
-/// No-op for creates/updates and for items without a source timestamp
-/// (older peers): a mark without a source clock could not be ordered against
-/// later items. Called before a local deletion is broadcast and after a
-/// replicated deletion is applied, so both sides out-rank a stale grant that
-/// arrives later.
-pub(crate) async fn record_iam_deletion_marks_for_item(item: &SRIAMItem) -> S3Result<()> {
-    let entities = iam_item_deletion_mark_entities(item);
-    let Some(deleted_at) = item.updated_at.filter(|_| !entities.is_empty()) else {
-        return Ok(());
-    };
-    commit_iam_deletion_marks(entities, deleted_at).await
-}
-
-/// [`record_iam_deletion_marks`] under the state transaction; the write is
-/// skipped when no mark moves.
-pub(crate) async fn commit_iam_deletion_marks(entities: Vec<String>, deleted_at: OffsetDateTime) -> S3Result<()> {
-    update_site_replication_state_when_changed(move |state| {
-        Ok(if record_iam_deletion_marks(state, &entities, deleted_at) {
-            StateCommit::Changed(())
-        } else {
-            StateCommit::Unchanged(())
-        })
-    })
-    .await
-}
-
 /// Failure bookkeeping for one IAM item delivery: upsert the collapsed retry
 /// event and, when the item is a deletion, record its body for replay. Both
 /// live in the same state so the caller commits them in one transaction — a
@@ -867,8 +791,8 @@ impl RetrySnapshot {

    pub(crate) fn fingerprint(&self) -> S3Result<Vec<Vec<u8>>> {
        let mut payloads = match self {
-            Self::Iam(items) => items.iter().map(canonical_json_vec).collect::<Result<Vec<_>, _>>(),
-            Self::BucketMetadata(items) => items.iter().map(canonical_json_vec).collect::<Result<Vec<_>, _>>(),
+            Self::Iam(items) => items.iter().map(serde_json::to_vec).collect::<Result<Vec<_>, _>>(),
+            Self::BucketMetadata(items) => items.iter().map(serde_json::to_vec).collect::<Result<Vec<_>, _>>(),
        }
        .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize retry snapshot failed: {err}")))?;
        payloads.sort_unstable();
@@ -1030,7 +954,6 @@ pub(crate) enum IamSnapshotKey {
    User(String),
    Group(String),
    PolicyMapping { target: String, user_type: i64, is_group: bool },
-    ServiceAccount(String),
 }

 pub(crate) fn iam_snapshot_key(item: &SRIAMItem) -> Option<IamSnapshotKey> {
@@ -1049,11 +972,6 @@ pub(crate) fn iam_snapshot_key(item: &SRIAMItem) -> Option<IamSnapshotKey> {
            user_type: mapping.user_type,
            is_group: mapping.is_group,
        }),
-        "service-account" => item
-            .svc_acc_change
-            .as_ref()
-            .and_then(|change| change.create.as_ref())
-            .map(|create| IamSnapshotKey::ServiceAccount(create.access_key.clone())),
        _ => None,
    }
 }
@@ -1088,24 +1006,6 @@ pub(crate) fn iam_snapshot_tombstones(item: &SRIAMItem, observed_at: OffsetDateT
                mapping.policy.clear();
            }
        }
-        "service-account" => {
-            let Some(access_key) = item
-                .svc_acc_change
-                .as_ref()
-                .and_then(|change| change.create.as_ref())
-                .map(|create| create.access_key.clone())
-            else {
-                return Vec::new();
-            };
-            tombstone.svc_acc_change = Some(SRSvcAccChange {
-                delete: Some(SRSvcAccDelete {
-                    access_key,
-                    api_version: Some(SITE_REPL_API_VERSION.to_string()),
-                }),
-                api_version: Some(SITE_REPL_API_VERSION.to_string()),
-                ..Default::default()
-            });
-        }
        _ => return Vec::new(),
    }
    vec![tombstone]
@@ -1801,7 +1701,7 @@ pub(crate) async fn drain_site_replication_retry_queue_locked(
    // tick and only when a snapshot resend is actually due.
    let plan = if needs_plan {
        let info = build_sr_info(&runtime.state, &runtime.local_peer).await?;
-        Some(build_site_replication_bootstrap_plan(&info).await?)
+        Some(site_replication_bootstrap_plan(&info)?)
    } else {
        None
    };
@@ -1941,7 +1841,7 @@ pub(crate) async fn drain_one_site_replication_retry_event(
                    }
                }
                let fresh_info = build_sr_info(&runtime.state, &runtime.local_peer).await?;
-                let fresh_plan = build_site_replication_bootstrap_plan(&fresh_info).await?;
+                let fresh_plan = site_replication_bootstrap_plan(&fresh_info)?;
                let fresh_snapshot = RetrySnapshot::from_plan(&action, &fresh_plan).expect("snapshot action has a snapshot");
                if fresh_snapshot.fingerprint()? == current_fingerprint {
                    if is_iam {
@@ -64,104 +64,6 @@ pub(crate) struct SiteReplicationState {
    /// newer edit that already landed.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub(crate) applied_edit_generations: BTreeMap<String, u64>,
-    /// Source timestamp of the newest IAM deletion committed on this site,
-    /// keyed by the deleted entity (`iam_item_deletion_mark_entities`). A
-    /// deletion leaves no local record to judge a later item against, so this
-    /// is what lets the receive-side staleness gate reject a grant that is
-    /// older than the revoke it would otherwise undo (backlog#2291). Marks
-    /// are kept for [`SITE_REPLICATION_IAM_DELETION_MARK_RETENTION`] and never
-    /// evicted by count: see that constant for why a count bound would open
-    /// exactly the window the marks exist to close.
-    #[serde(default, with = "rfc3339_map", skip_serializing_if = "BTreeMap::is_empty")]
-    pub(crate) iam_deletion_marks: BTreeMap<String, OffsetDateTime>,
-}
-
-/// How long an IAM deletion mark outlives the deletion it records.
-///
-/// A mark fences the delivery paths that can still carry an older grant for
-/// the deleted entity: a live delivery delayed in transit, the same grant
-/// arriving on a sibling node while the revoke is being applied, and a
-/// snapshot (bootstrap / repair / resend) built by a peer that has not yet
-/// received the deletion — which is bounded by this site's own retry queue
-/// towards that peer, whose backoff tops out at one day
-/// (`SITE_REPLICATION_RETRY_DRAIN_MAX_BACKOFF_SECS`). The retry drain itself
-/// never replays a stale grant: it resends snapshots of the current records
-/// and the recorded deletion bodies. Thirty days is an order of magnitude
-/// beyond every one of those windows. Marks are pruned by age only — a count
-/// bound would drop a mark that is still inside the delivery window as soon
-/// as enough newer deletions happen, letting the delayed grant re-create the
-/// entity, which is the very hole the marks close.
-pub(crate) const SITE_REPLICATION_IAM_DELETION_MARK_RETENTION: time::Duration = time::Duration::days(30);
-
-/// Record that deletions of `entities` with source timestamp `deleted_at`
-/// were committed here. Newest wins per entity: an older deletion never
-/// lowers a mark. Marks older than the retention are pruned in the same
-/// pass. Returns whether the state changed.
-pub(crate) fn record_iam_deletion_marks(
-    state: &mut SiteReplicationState,
-    entities: &[String],
-    deleted_at: OffsetDateTime,
-) -> bool {
-    record_iam_deletion_marks_at(state, entities, deleted_at, OffsetDateTime::now_utc())
-}
-
-/// [`record_iam_deletion_marks`] pruning against an explicit `now`.
-pub(crate) fn record_iam_deletion_marks_at(
-    state: &mut SiteReplicationState,
-    entities: &[String],
-    deleted_at: OffsetDateTime,
-    now: OffsetDateTime,
-) -> bool {
-    let mut changed = false;
-    for entity in entities {
-        if state
-            .iam_deletion_marks
-            .get(entity)
-            .is_some_and(|existing| *existing >= deleted_at)
-        {
-            continue;
-        }
-        state.iam_deletion_marks.insert(entity.clone(), deleted_at);
-        changed = true;
-    }
-    let expired_before = now - SITE_REPLICATION_IAM_DELETION_MARK_RETENTION;
-    let before = state.iam_deletion_marks.len();
-    state.iam_deletion_marks.retain(|_, deleted_at| *deleted_at >= expired_before);
-    changed || state.iam_deletion_marks.len() != before
-}
-
-/// Newest deletion mark among `entities`, or `None` when no deletion of any
-/// of them was recorded here. The receive-side staleness gate feeds this in
-/// as the local timestamp when the targeted record is absent.
-pub(crate) fn iam_deletion_mark(state: &SiteReplicationState, entities: &[String]) -> Option<OffsetDateTime> {
-    entities
-        .iter()
-        .filter_map(|entity| state.iam_deletion_marks.get(entity).copied())
-        .max()
-}
-
-/// RFC 3339 map values, matching the other timestamps in the state object
-/// (`time::serde::rfc3339` only applies to a single field).
-mod rfc3339_map {
-    use serde::{Deserialize, Deserializer, Serialize, Serializer};
-    use std::collections::BTreeMap;
-    use time::OffsetDateTime;
-
-    #[derive(Serialize, Deserialize)]
-    #[serde(transparent)]
-    struct Stamp(#[serde(with = "time::serde::rfc3339")] OffsetDateTime);
-
-    pub(super) fn serialize<S: Serializer>(map: &BTreeMap<String, OffsetDateTime>, serializer: S) -> Result<S::Ok, S::Error> {
-        serializer.collect_map(map.iter().map(|(entity, deleted_at)| (entity, Stamp(*deleted_at))))
-    }
-
-    pub(super) fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<BTreeMap<String, OffsetDateTime>, D::Error> {
-        let map = BTreeMap::<String, Stamp>::deserialize(deserializer)?;
-        Ok(map
-            .into_iter()
-            .map(|(entity, Stamp(deleted_at))| (entity, deleted_at))
-            .collect())
-    }
 }

 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
@@ -421,33 +323,6 @@ where
    update_site_replication_state_when_changed(move |state| update(state).map(StateCommit::Changed)).await
 }

-/// The state transaction for work that has to await inside it: an IAM write
-/// that must be ordered with the staleness verdict taken before it and the
-/// deletion mark committed after it (backlog#2291). Same boundary as
-/// [`update_site_replication_state`] — load and persist under the
-/// distributed state-object write lock, so two nodes of this site cannot
-/// interleave their verdicts and writes — and the same rules inside: no peer
-/// network calls and no other config locks. The closure hands the state back
-/// as `Some` when it changed it; `None` skips the write.
-pub(crate) async fn with_site_replication_state_transaction<T, F, Fut>(transaction: F) -> S3Result<T>
-where
-    T: Send + 'static,
-    F: FnOnce(SiteReplicationState) -> Fut + Send + 'static,
-    Fut: std::future::Future<Output = S3Result<(T, Option<SiteReplicationState>)>> + Send + 'static,
-{
-    with_site_replication_state_lock(move || async move {
-        let store = current_object_store_handle()
-            .ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()))?;
-        let state = load_site_replication_state_no_lock(store.clone()).await?;
-        let (result, changed) = transaction(state).await?;
-        if let Some(state) = changed {
-            persist_site_replication_state_no_lock(store, state).await?;
-        }
-        Ok(result)
-    })
-    .await
-}
-
 /// [`update_site_replication_state`] for closures that may find nothing to
 /// do — see [`StateCommit`].
 pub(crate) async fn update_site_replication_state_when_changed<T, F>(update: F) -> S3Result<T>
@@ -554,145 +554,6 @@ fn test_iam_item_deletion_entity_shapes() {
    assert!(iam_item_deletion_entity(&policy_set).is_none());
 }

-/// Deletion marks (backlog#2291) key on the same entities as the replay
-/// records, except that a group member removal is marked per member (so a
-/// stale re-add of one member can be judged) and a group delete marks the
-/// group itself. Creates and updates leave no mark.
-#[test]
-fn test_iam_item_deletion_mark_entities_shapes() {
-    assert_eq!(
-        iam_item_deletion_mark_entities(&user_delete_item("alice")),
-        vec!["iam-user:alice".to_string()]
-    );
-    assert_eq!(
-        iam_item_deletion_mark_entities(&policy_delete_item("readonly")),
-        vec!["policy:readonly".to_string()]
-    );
-
-    let mut group_remove = SRIAMItem {
-        r#type: "group-info".to_string(),
-        group_info: Some(SRGroupInfo {
-            update_req: GroupAddRemove {
-                group: "devs".to_string(),
-                members: vec!["bob".to_string(), "alice".to_string()],
-                status: GroupStatus::Enabled,
-                is_remove: true,
-            },
-            api_version: None,
-        }),
-        ..Default::default()
-    };
-    assert_eq!(
-        iam_item_deletion_mark_entities(&group_remove),
-        vec!["group-member:devs:bob".to_string(), "group-member:devs:alice".to_string()]
-    );
-    group_remove
-        .group_info
-        .as_mut()
-        .expect("group info")
-        .update_req
-        .members
-        .clear();
-    assert_eq!(
-        iam_item_deletion_mark_entities(&group_remove),
-        vec!["group:devs".to_string()],
-        "a removal without members deletes the group"
-    );
-    group_remove.group_info.as_mut().expect("group info").update_req.is_remove = false;
-    assert!(iam_item_deletion_mark_entities(&group_remove).is_empty());
-
-    let mapping_clear = SRIAMItem {
-        r#type: "policy-mapping".to_string(),
-        policy_mapping: Some(SRPolicyMapping {
-            user_or_group: "alice".to_string(),
-            user_type: 0,
-            is_group: false,
-            policy: String::new(),
-            ..Default::default()
-        }),
-        ..Default::default()
-    };
-    assert_eq!(
-        iam_item_deletion_mark_entities(&mapping_clear),
-        vec!["policy-mapping:alice:0:false".to_string()]
-    );
-
-    let mut user_create = user_delete_item("alice");
-    user_create.iam_user.as_mut().expect("iam user").is_delete_req = false;
-    assert!(iam_item_deletion_mark_entities(&user_create).is_empty());
-}
-
-/// Newest wins per entity, marks are pruned by age only (never by count: a
-/// count bound would drop a mark still inside the delivery window as soon as
-/// enough newer deletions happen), and the timestamps survive the state
-/// object as RFC 3339.
-#[test]
-fn test_record_iam_deletion_marks_newest_wins_and_expires_by_age_only() {
-    let at = |seconds: i64| OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(seconds);
-    let now = at(1_000_000);
-    let mut state = SiteReplicationState::default();
-    let alice = vec!["iam-user:alice".to_string()];
-
-    assert!(record_iam_deletion_marks_at(&mut state, &alice, at(20), now));
-    assert!(
-        !record_iam_deletion_marks_at(&mut state, &alice, at(10), now),
-        "an older deletion does not move the mark"
-    );
-    assert!(
-        !record_iam_deletion_marks_at(&mut state, &alice, at(20), now),
-        "a replayed deletion is not a change"
-    );
-    assert_eq!(iam_deletion_mark(&state, &alice), Some(at(20)));
-    assert!(record_iam_deletion_marks_at(&mut state, &alice, at(30), now));
-    assert_eq!(iam_deletion_mark(&state, &alice), Some(at(30)));
-    assert_eq!(iam_deletion_mark(&state, &["iam-user:bob".to_string()]), None);
-    assert!(!record_iam_deletion_marks_at(&mut state, &[], at(40), now));
-
-    // Many newer deletions never evict an older mark that is still within the retention.
-    let members: Vec<String> = (0..4096).map(|index| format!("group-member:devs:user-{index:04}")).collect();
-    for (index, member) in members.iter().enumerate() {
-        record_iam_deletion_marks_at(&mut state, std::slice::from_ref(member), at(100 + index as i64), now);
-    }
-    assert_eq!(state.iam_deletion_marks.len(), members.len() + 1);
-    assert_eq!(iam_deletion_mark(&state, &alice), Some(at(30)), "no count-based eviction");
-
-    // Marks older than the retention are pruned, on the pass that records a
-    // newer one and on a pass that changes nothing else; younger ones stay.
-    let later = at(100) + SITE_REPLICATION_IAM_DELETION_MARK_RETENTION;
-    assert!(
-        record_iam_deletion_marks_at(&mut state, &["iam-user:carol".to_string()], at(200_000), later),
-        "pruning alone is a change"
-    );
-    assert_eq!(iam_deletion_mark(&state, &alice), None, "alice's mark aged out");
-    assert_eq!(
-        iam_deletion_mark(&state, &members[..1]),
-        Some(at(100)),
-        "a mark exactly at the retention edge stays, and so do the younger ones"
-    );
-    assert_eq!(state.iam_deletion_marks.len(), members.len() + 1);
-    assert_eq!(iam_deletion_mark(&state, &["iam-user:carol".to_string()]), Some(at(200_000)));
-    let mut state = SiteReplicationState::default();
-    record_iam_deletion_marks_at(&mut state, &alice, at(30), now);
-    let past_edge = at(30) + SITE_REPLICATION_IAM_DELETION_MARK_RETENTION + time::Duration::seconds(1);
-    assert!(
-        record_iam_deletion_marks_at(&mut state, &[], at(0), past_edge),
-        "a pass that only prunes reports the change"
-    );
-    assert_eq!(iam_deletion_mark(&state, &alice), None);
-    record_iam_deletion_marks_at(&mut state, &alice, at(30), now);
-
-    let json = serde_json::to_value(&state).expect("serialize state");
-    assert_eq!(json["iam_deletion_marks"]["iam-user:alice"], serde_json::json!("1970-01-01T00:00:30Z"));
-    let reloaded = parse_site_replication_state(&serde_json::to_vec(&state).expect("serialize state")).expect("parse state");
-    assert_eq!(reloaded.iam_deletion_marks, state.iam_deletion_marks);
-    assert!(
-        parse_site_replication_state(br#"{"name":"a","service_account_access_key":"","service_account_parent":"","peers":{},"updated_at":null,"resync_status":{}}"#)
-            .expect("state without marks")
-            .iam_deletion_marks
-            .is_empty()
-    );
-}
-
 /// A failed deletion delivery persists a replay record next to the collapsed
 /// retry entry; a fresh entry is stamped `deletions_recorded` so a later
 /// replay can settle it, and a repeated deletion of the same entity keeps the
@@ -1818,8 +1679,7 @@ fn test_site_replication_bootstrap_plan_includes_replayable_snapshot_items() {
        },
    );

-    let plan =
-        site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build");
+    let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build");

    assert_eq!(plan.iam_items.iter().map(|item| item.r#type.as_str()).collect::<Vec<_>>(), {
        vec!["policy", "iam-user", "group-info", "policy-mapping"]
@@ -1857,8 +1717,7 @@ fn test_site_replication_bootstrap_plan_skips_lifecycle_by_default() {
        },
    );

-    let plan =
-        site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build");
+    let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build");

    assert!(!plan.bucket_items.iter().any(|item| item.r#type == "lc-config"));
 }
@@ -1889,8 +1748,7 @@ fn test_site_replication_bootstrap_plan_emits_timestamped_lifecycle_delete() {
        },
    );

-    let plan =
-        site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build");
+    let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build");

    let item = plan
        .bucket_items
@@ -2077,8 +1935,8 @@ fn test_site_replication_repair_preflight_token_is_deterministic_for_equal_state
        },
    );

-    let plan_a = site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("first plan");
-    let plan_b = site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("second plan");
+    let plan_a = site_replication_bootstrap_plan(&info).expect("first plan");
+    let plan_b = site_replication_bootstrap_plan(&info).expect("second plan");
    let token_a = site_replication_repair_preflight_token(&state, &plan_a, b"test-signing-key").expect("first token");
    let token_b = site_replication_repair_preflight_token(&state, &plan_b, b"test-signing-key").expect("second token");

@@ -3361,345 +3219,3 @@ fn test_reconcile_adds_missing_peer_rules_to_existing_config() {
    assert!(rule_ids.contains(&"site-repl-dep-b"));
    assert!(rule_ids.contains(&"site-repl-dep-c"));
 }
-
-/// backlog#2289: the IAM snapshot (retry resend, repair, site-add bootstrap)
-/// used to be built from `list_users`, whose `UserInfo` never carries a
-/// secret key, so the plan dropped every user and a status change or secret
-/// rotation committed while a peer was unreachable never reached it. The
-/// credentials now come from a separate store read; SRInfo stays secret-free.
-#[test]
-fn test_bootstrap_plan_carries_users_from_the_credential_snapshot() {
-    let mut info = SRInfo::default();
-    // Exactly what `list_users` builds: status, policy, updated_at — never secret_key.
-    info.user_info_map.insert(
-        "alice".to_string(),
-        rustfs_madmin::UserInfo {
-            status: rustfs_madmin::AccountStatus::Disabled,
-            policy_name: Some("readwrite".to_string()),
-            updated_at: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")),
-            ..Default::default()
-        },
-    );
-    info.user_info_map.insert(
-        "external-idp-user".to_string(),
-        rustfs_madmin::UserInfo {
-            status: rustfs_madmin::AccountStatus::Enabled,
-            ..Default::default()
-        },
-    );
-    let user_updated_at = OffsetDateTime::from_unix_timestamp(1_700_000_500).expect("timestamp");
-    let mut credentials = SiteReplicationIamCredentials::default();
-    credentials.users.insert(
-        "alice".to_string(),
-        SiteReplicationUserCredential {
-            secret_key: "alice-secret".to_string(),
-            status: rustfs_madmin::AccountStatus::Disabled,
-            updated_at: Some(user_updated_at),
-        },
-    );
-
-    let plan = site_replication_bootstrap_plan(&info, &credentials).expect("bootstrap plan should build");
-
-    let users: Vec<_> = plan.iam_items.iter().filter(|item| item.r#type == "iam-user").collect();
-    assert_eq!(users.len(), 1, "only the user with a credential travels: {:?}", plan.iam_items);
-    let alice = users[0].iam_user.as_ref().expect("iam user body");
-    assert_eq!(alice.access_key, "alice");
-    let req = alice.user_req.as_ref().expect("user request");
-    assert_eq!(req.secret_key, "alice-secret");
-    assert_eq!(req.status, rustfs_madmin::AccountStatus::Disabled);
-    assert_eq!(req.policy.as_deref(), Some("readwrite"));
-    // the user record's own axis, not the policy-mapping time list_users reports
-    assert_eq!(users[0].updated_at, Some(user_updated_at));
-}
-
-fn service_account_snapshot(access_key: &str, parent: &str, status: &str) -> SiteReplicationServiceAccountSnapshot {
-    SiteReplicationServiceAccountSnapshot {
-        create: rustfs_madmin::SRSvcAccCreate {
-            parent: parent.to_string(),
-            access_key: access_key.to_string(),
-            secret_key: format!("{access_key}-secret"),
-            groups: Vec::new(),
-            claims: HashMap::new(),
-            session_policy: SRSessionPolicy::default(),
-            status: status.to_string(),
-            name: String::new(),
-            description: String::new(),
-            expiration: None,
-            api_version: Some(SITE_REPL_API_VERSION.to_string()),
-        },
-        envelope: None,
-        updated_at: Some(OffsetDateTime::from_unix_timestamp(1_700_000_600).expect("timestamp")),
-    }
-}
-
-/// backlog#2289: service accounts were absent from every snapshot (the
-/// listing filters them). They now travel as the create item the live hook
-/// emits — after their parents — carrying secret and status.
-#[test]
-fn test_bootstrap_plan_emits_service_accounts_after_their_parents() {
-    let mut info = SRInfo::default();
-    info.user_info_map
-        .insert("alice".to_string(), rustfs_madmin::UserInfo::default());
-    let mut credentials = SiteReplicationIamCredentials::default();
-    credentials.users.insert(
-        "alice".to_string(),
-        SiteReplicationUserCredential {
-            secret_key: "alice-secret".to_string(),
-            status: rustfs_madmin::AccountStatus::Enabled,
-            updated_at: None,
-        },
-    );
-    credentials
-        .service_accounts
-        .push(service_account_snapshot("alice-svc", "alice", "off"));
-
-    let plan = site_replication_bootstrap_plan(&info, &credentials).expect("bootstrap plan should build");
-
-    let types: Vec<_> = plan.iam_items.iter().map(|item| item.r#type.as_str()).collect();
-    assert_eq!(types, vec!["iam-user", "service-account"]);
-    let change = plan.iam_items[1].svc_acc_change.as_ref().expect("service account change");
-    let create = change.create.as_ref().expect("create body");
-    assert_eq!((create.access_key.as_str(), create.parent.as_str()), ("alice-svc", "alice"));
-    assert_eq!(create.secret_key, "alice-svc-secret");
-    assert_eq!(create.status, "off", "a disabled account must arrive disabled");
-    assert!(change.delete.is_none() && change.update.is_none());
-}
-
-/// A service account present in the previous snapshot but gone from the
-/// fresh one is replayed as an explicit delete, like the other IAM kinds.
-#[test]
-fn test_retry_snapshot_tombstones_removed_service_accounts() {
-    let observed_at = OffsetDateTime::from_unix_timestamp(1_700_001_000).expect("timestamp");
-    let mut info = SRInfo::default();
-    info.user_info_map
-        .insert("alice".to_string(), rustfs_madmin::UserInfo::default());
-    let mut credentials = SiteReplicationIamCredentials::default();
-    credentials.users.insert(
-        "alice".to_string(),
-        SiteReplicationUserCredential {
-            secret_key: "alice-secret".to_string(),
-            status: rustfs_madmin::AccountStatus::Enabled,
-            updated_at: None,
-        },
-    );
-    let mut with_account = credentials.clone();
-    with_account
-        .service_accounts
-        .push(service_account_snapshot("alice-svc", "alice", "on"));
-    let previous = site_replication_bootstrap_plan(&info, &with_account).expect("previous plan");
-    let fresh = site_replication_bootstrap_plan(&info, &credentials).expect("fresh plan");
-
-    let replay = RetrySnapshot::replay_after_change(
-        &RetrySnapshot::Iam(previous.iam_items),
-        &RetrySnapshot::Iam(fresh.iam_items),
-        observed_at,
-    );
-    let RetrySnapshot::Iam(items) = replay else {
-        panic!("IAM snapshot expected");
-    };
-    let tombstone = items
-        .iter()
-        .find(|item| item.r#type == "service-account")
-        .expect("service account tombstone");
-    let change = tombstone.svc_acc_change.as_ref().expect("change");
-    assert_eq!(change.delete.as_ref().map(|delete| delete.access_key.as_str()), Some("alice-svc"));
-    assert!(change.create.is_none());
-    assert_eq!(tombstone.updated_at, Some(observed_at));
-}
-
-/// Spawns a one-shot HTTP peer that answers 200 and flips the returned flag
-/// once a request head has arrived.
-async fn spawn_reached_probe_peer() -> (String, Arc<AtomicBool>, tokio::task::JoinHandle<()>) {
-    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind healthy peer");
-    let endpoint = format!("http://{}", listener.local_addr().expect("healthy peer address"));
-    let reached = Arc::new(AtomicBool::new(false));
-    let reached_by_server = reached.clone();
-    let server = tokio::spawn(async move {
-        let Ok((mut stream, _)) = listener.accept().await else {
-            return;
-        };
-        let mut request = Vec::new();
-        let mut buffer = [0_u8; 1024];
-        loop {
-            let Ok(read) = stream.read(&mut buffer).await else {
-                return;
-            };
-            if read == 0 {
-                return;
-            }
-            request.extend_from_slice(&buffer[..read]);
-            if request.windows(4).any(|window| window == b"\r\n\r\n") {
-                break;
-            }
-        }
-        reached_by_server.store(true, Ordering::SeqCst);
-        let _ = stream
-            .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\nconnection: close\r\n\r\nok")
-            .await;
-    });
-    (endpoint, reached, server)
-}
-
-/// Three-peer runtime whose local peer is `local`; BTreeMap order visits the
-/// failing peer `b` before the healthy peer `c`.
-fn broadcast_runtime_with_failing_peer_before_healthy(failing_endpoint: &str, healthy_endpoint: &str) -> SiteReplicationRuntime {
-    let local_peer = PeerInfo {
-        deployment_id: "local".to_string(),
-        ..peer("local", "http://127.0.0.1:9")
-    };
-    let mut state = SiteReplicationState {
-        name: "local".to_string(),
-        service_account_access_key: "site-replicator-0".to_string(),
-        ..Default::default()
-    };
-    state.peers.insert("local".to_string(), local_peer.clone());
-    state.peers.insert(
-        "b".to_string(),
-        PeerInfo {
-            deployment_id: "b".to_string(),
-            ..peer("b", failing_endpoint)
-        },
-    );
-    state.peers.insert(
-        "c".to_string(),
-        PeerInfo {
-            deployment_id: "c".to_string(),
-            ..peer("c", healthy_endpoint)
-        },
-    );
-    SiteReplicationRuntime {
-        state,
-        local_peer,
-        service_account_secret_key: "site-replicator-secret".to_string(),
-    }
-}
-
-const BROADCAST_PROBE_DELETE_BUCKET_PATH: &str =
-    "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=delete-bucket";
-
-/// The generic JSON broadcast (bucket make/delete, bucket-meta hook, bucket
-/// ops) attempts every remote peer: a peer whose request fails must not stop
-/// delivery to the peers that follow it in deployment-id order, and the
-/// failure is still reported to the caller (backlog#2293).
-#[tokio::test]
-#[serial]
-async fn test_broadcast_json_reaches_healthy_peers_after_a_failed_peer() {
-    // Peer "b": nothing listens on the port, so the connect is refused.
-    let refused = TcpListener::bind("127.0.0.1:0").await.expect("bind refused-peer probe");
-    let refused_endpoint = format!("http://{}", refused.local_addr().expect("refused-peer address"));
-    drop(refused);
-
-    let (healthy_endpoint, reached, server) = spawn_reached_probe_peer().await;
-    let runtime = broadcast_runtime_with_failing_peer_before_healthy(&refused_endpoint, &healthy_endpoint);
-
-    let result = temp_env::async_with_vars([(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV, Some("true"))], async {
-        broadcast_site_replication_json_with_runtime(&runtime, BROADCAST_PROBE_DELETE_BUCKET_PATH, &serde_json::json!({})).await
-    })
-    .await;
-
-    let err = result.expect_err("peer b refuses connections, the broadcast must report it");
-    assert!(
-        reached.load(Ordering::SeqCst),
-        "peer c never received the broadcast once peer b failed: {err}"
-    );
-    server.abort();
-}
-
-/// Same guarantee when the failing peer never gets a transport: an endpoint
-/// that `PeerTransport::for_runtime_peer` rejects must be skipped past (and
-/// reported), not abort the broadcast before the healthy peers (backlog#2293).
-#[tokio::test]
-#[serial]
-async fn test_broadcast_json_reaches_healthy_peers_after_a_peer_without_transport() {
-    // Peer "b": a scheme the peer connection validator refuses outright.
-    let forbidden_endpoint = "ftp://peer-b.example.com";
-
-    let (healthy_endpoint, reached, server) = spawn_reached_probe_peer().await;
-    let runtime = broadcast_runtime_with_failing_peer_before_healthy(forbidden_endpoint, &healthy_endpoint);
-
-    let result = temp_env::async_with_vars([(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV, Some("true"))], async {
-        broadcast_site_replication_json_with_runtime(&runtime, BROADCAST_PROBE_DELETE_BUCKET_PATH, &serde_json::json!({})).await
-    })
-    .await;
-
-    let err = result.expect_err("peer b has no usable transport, the broadcast must report it");
-    assert!(
-        err.to_string().contains("invalid persisted site replication peer"),
-        "the reported error must be peer b's transport failure: {err}"
-    );
-    assert!(
-        reached.load(Ordering::SeqCst),
-        "peer c never received the broadcast once peer b failed to get a transport: {err}"
-    );
-    server.abort();
-}
-
-fn service_account_item_with_claims(order: &[&str]) -> SRIAMItem {
-    let mut claims = HashMap::new();
-    for key in order {
-        claims.insert((*key).to_string(), serde_json::json!(format!("value-of-{key}")));
-    }
-    SRIAMItem {
-        r#type: "service-account".to_string(),
-        svc_acc_change: Some(SRSvcAccChange {
-            create: Some(rustfs_madmin::SRSvcAccCreate {
-                parent: "alice".to_string(),
-                access_key: "alice-svc".to_string(),
-                secret_key: "alice-svc-secret".to_string(),
-                groups: Vec::new(),
-                claims,
-                session_policy: SRSessionPolicy::default(),
-                status: "on".to_string(),
-                name: String::new(),
-                description: String::new(),
-                expiration: None,
-                api_version: Some(SITE_REPL_API_VERSION.to_string()),
-            }),
-            api_version: Some(SITE_REPL_API_VERSION.to_string()),
-            ..Default::default()
-        }),
-        updated_at: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")),
-        api_version: Some(SITE_REPL_API_VERSION.to_string()),
-        ..Default::default()
-    }
-}
-
-/// The repair preflight token and the retry-snapshot fingerprint hash the
-/// serialized items. Service-account claims live in a `HashMap`, whose
-/// iteration order differs between instances, so the hash must not depend on
-/// it (the real-VM repair returned 412 "preflight is stale" between dry-run
-/// and execute once snapshots carried service accounts).
-#[test]
-fn test_repair_task_id_and_retry_fingerprint_ignore_claim_map_order() {
-    let forward = service_account_item_with_claims(&["accessKey", "exp", "parent", "sa-policy", "sub", "tenant"]);
-    let backward = service_account_item_with_claims(&["tenant", "sub", "sa-policy", "parent", "exp", "accessKey"]);
-
-    let canonical = canonical_json_vec(&forward).expect("canonical json");
-    let text = String::from_utf8(canonical).expect("utf8");
-    let positions: Vec<usize> = [
-        "\"accessKey\"",
-        "\"exp\"",
-        "\"parent\"",
-        "\"sa-policy\"",
-        "\"sub\"",
-        "\"tenant\"",
-    ]
-    .iter()
-    .map(|key| text.find(key).expect("claim key present"))
-    .collect();
-    assert!(
-        positions.windows(2).all(|pair| pair[0] < pair[1]),
-        "claim keys must serialize sorted: {text}"
-    );
-
-    assert_eq!(
-        SiteReplicationRepairTask::Iam(&forward).id().expect("id"),
-        SiteReplicationRepairTask::Iam(&backward).id().expect("id"),
-        "identical items must yield the same repair task id regardless of claim map order"
-    );
-    assert_eq!(
-        RetrySnapshot::Iam(vec![forward]).fingerprint().expect("fingerprint"),
-        RetrySnapshot::Iam(vec![backward]).fingerprint().expect("fingerprint"),
-        "identical snapshots must fingerprint equal regardless of claim map order"
-    );
-}
@@ -876,14 +876,6 @@ pub(crate) async fn broadcast_site_replication_json<T: Serialize>(path: &str, bo
    broadcast_site_replication_json_with_runtime(&runtime, path, body).await
 }

-/// PUT `body` to `path` on every remote peer of the runtime.
-///
-/// Every peer is attempted: one peer's failure — transport construction
-/// included — must not skip the peers that follow it in deployment-id order,
-/// or they silently miss the change with no retry record (backlog#2293). A
-/// success settles the peer/path's queued retry event, a failure enqueues one
-/// under the request `path` (so the drain classifies it as today), and the
-/// first error is returned once all peers were attempted.
 pub(crate) async fn broadcast_site_replication_json_with_runtime<T: Serialize>(
    runtime: &SiteReplicationRuntime,
    path: &str,
@@ -891,30 +883,20 @@ pub(crate) async fn broadcast_site_replication_json_with_runtime<T: Serialize>(
 ) -> S3Result<()> {
    let state = &runtime.state;
    let local_peer = &runtime.local_peer;
-    let mut first_error: Option<S3Error> = None;

    for peer in state.peers.values() {
        if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) {
            continue;
        }

-        let sent = match PeerTransport::for_runtime_peer(peer).await {
-            Ok(transport) => PeerAdminRequest::put(&transport.connection, path, &state.service_account_access_key)
-                .with_client(&transport.client)
-                .send_with_retry_event(peer, &runtime.service_account_secret_key, body)
-                .await
-                .map(|_| ()),
-            Err(err) => {
-                enqueue_site_replication_retry_event(peer, path, &err).await;
-                Err(err)
-            }
-        };
-        if let Err(err) = sent {
-            first_error.get_or_insert(err);
-        }
+        let transport = PeerTransport::for_runtime_peer(peer).await?;
+        PeerAdminRequest::put(&transport.connection, path, &state.service_account_access_key)
+            .with_client(&transport.client)
+            .send_with_retry_event(peer, &runtime.service_account_secret_key, body)
+            .await?;
    }

-    first_error.map_or(Ok(()), Err)
+    Ok(())
 }

 pub(crate) fn parse_endpoint_refresh_status(peer: &PeerInfo, body: &[u8]) -> S3Result<()> {
Author	SHA1	Message	Date
Zhengchao An	043842d571	Merge branch 'main' into test/distributed-e2e-hardening	2026-09-06 03:32:13 +08:00
Zhengchao An	e1608fbd9c	test(odm): exercise overflow and invalid cursors reliably (#7236 )	2026-09-06 03:09:39 +08:00
Zhengchao An	4da48f888e	Merge branch 'main' into test/distributed-e2e-hardening	2026-09-05 23:56:44 +08:00
overtrue	d8580ec970	test(e2e): prove operations overlap data movement	2026-09-05 22:50:00 +08:00
Zhengchao An	034efa8dac	Merge branch 'main' into test/distributed-e2e-hardening	2026-09-05 22:21:31 +08:00
Zhengchao An	1277c11f04	Merge branch 'main' into test/distributed-e2e-hardening	2026-09-05 11:57:33 +08:00
RustFS	bdb069c0fd	Merge branch 'main' into test/distributed-e2e-hardening	2026-09-05 09:29:28 +08:00
RustFS	b65ab747d3	Merge branch 'main' into test/distributed-e2e-hardening	2026-09-05 09:17:48 +08:00
loverustfs	8c48f184b5	test(e2e): add distributed 4x4 validation	2026-09-05 02:51:03 +08:00