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

test(e2e): prove operations overlap data movement
2026-09-05 19:55:37 +00:00 · 2026-09-06 03:32:13 +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 · 2026-09-05 11:57:33 +08:00
43 changed files with 4237 additions and 2161 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;
@@ -4498,12 +4498,6 @@ impl SetDisks {
        &self.ctx
    }

-    /// Read the persisted bucket identity through this set's metadata owner.
-    /// Missing or non-authoritative legacy identities remain errors.
-    pub async fn bucket_incarnation_id_from_disk(&self, bucket: &str) -> Result<Uuid> {
-        metadata_sys::get_bucket_incarnation_id_in(&self.ctx, bucket).await
-    }
-
    /// Admit one short scanner cache publication under this set's instance
    /// movement fence. The caller must hold the returned guard through its
    /// final conditional cache write; no scan-round work belongs under it.
@@ -14,7 +14,6 @@

 use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration};
 use serde::{Deserialize, Serialize, ser::SerializeMap};
-use sha2::{Digest, Sha256};
 use std::{
    collections::{HashMap, HashSet},
    future::Future,
@@ -401,85 +400,6 @@ impl DataUsageScanCheckpoint {
    }
 }

-/// Durable scope of a bucket checkpoint, independent of namespace mutation counters.
-#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq)]
-#[serde(deny_unknown_fields)]
-pub struct DataUsageScanIdentity {
-    pub version: u16,
-    pub bucket_incarnation: uuid::Uuid,
-    pub set_layout: DataUsageScanPlanDigest,
-    pub publication_epoch: u64,
-    pub tier_registry_generation: u64,
-    pub scan_mode: HealScanMode,
-}
-
-impl Serialize for DataUsageScanIdentity {
-    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
-        let mut map = serializer.serialize_map(Some(6))?;
-        map.serialize_entry("version", &self.version)?;
-        map.serialize_entry("bucket_incarnation", &self.bucket_incarnation)?;
-        map.serialize_entry("set_layout", &self.set_layout)?;
-        map.serialize_entry("publication_epoch", &self.publication_epoch)?;
-        map.serialize_entry("tier_registry_generation", &self.tier_registry_generation)?;
-        map.serialize_entry("scan_mode", &self.scan_mode)?;
-        map.end()
-    }
-}
-
-impl DataUsageScanIdentity {
-    pub(crate) fn is_valid(&self) -> bool {
-        self.version == 1
-            && !self.bucket_incarnation.is_nil()
-            && matches!(self.scan_mode, HealScanMode::Normal | HealScanMode::Deep)
-    }
-}
-
-/// A forward coverage sweep may span budgets, but not authorize mixed mutation generations.
-#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq)]
-#[serde(deny_unknown_fields)]
-pub struct DataUsageScanProgress {
-    pub started_plan: DataUsageScanPlanDigest,
-    pub requested_plan: DataUsageScanPlanDigest,
-}
-
-impl Serialize for DataUsageScanProgress {
-    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
-        let mut map = serializer.serialize_map(Some(2))?;
-        map.serialize_entry("started_plan", &self.started_plan)?;
-        map.serialize_entry("requested_plan", &self.requested_plan)?;
-        map.end()
-    }
-}
-
-#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
-#[serde(deny_unknown_fields)]
-pub struct DataUsageScanCoverageReceipt {
-    pub through: String,
-    pub digest: [u8; 32],
-}
-
-impl Serialize for DataUsageScanCoverageReceipt {
-    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
-        let mut map = serializer.serialize_map(Some(2))?;
-        map.serialize_entry("through", &self.through)?;
-        map.serialize_entry("digest", &self.digest)?;
-        map.end()
-    }
-}
-
-struct CheckpointDigestWriter(Sha256);
-
-impl std::io::Write for CheckpointDigestWriter {
-    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
-        self.0.update(bytes);
-        Ok(bytes.len())
-    }
-
-    fn flush(&mut self) -> std::io::Result<()> {
-        Ok(())
-    }
-}
-
 #[derive(Clone, Debug, Default, Serialize, Deserialize)]
 pub struct DataUsageEntryInfo {
    pub name: String,
@@ -550,12 +470,6 @@ pub struct DataUsageCacheInfo {
    #[serde(default)]
    pub scan_checkpoint: Option<DataUsageScanCheckpoint>,
    #[serde(default)]
-    pub scan_identity: Option<DataUsageScanIdentity>,
-    #[serde(default)]
-    pub scan_progress: Option<DataUsageScanProgress>,
-    #[serde(default)]
-    pub scan_coverage_receipt: Option<DataUsageScanCoverageReceipt>,
-    #[serde(default)]
    pub pending_heals: Vec<PendingScannerHeal>,
    #[serde(default)]
    pub object_lock: Option<Arc<ObjectLockConfiguration>>,
@@ -567,11 +481,6 @@ pub struct DataUsageCacheInfo {
    pub snapshot_complete: bool,
    #[serde(default)]
    pub scan_plan_digest: Option<DataUsageScanPlanDigest>,
-    /// Full activity and inventory scope of a set scan; only a complete
-    /// snapshot proves coverage. Bucket caches bind this scope into their
-    /// opaque scan plan digest instead.
-    #[serde(default)]
-    pub scan_coverage_digest: Option<DataUsageScanPlanDigest>,
    #[serde(default)]
    pub cache_key_format: u16,
    /// Registry generation used for the completed/partial scan. This is
@@ -608,10 +517,6 @@ impl Serialize for DataUsageCacheInfo {
        // Keep this metadata map-encoded so older readers can ignore fields
        // appended by newer scanner versions during rolling upgrades.
        let field_count = 16
-            + usize::from(self.scan_identity.is_some())
-            + usize::from(self.scan_progress.is_some())
-            + usize::from(self.scan_coverage_receipt.is_some())
-            + usize::from(self.scan_coverage_digest.is_some())
            + usize::from(self.tier_registry_generation.is_some())
            + usize::from(!self.size_reconciliation.is_empty())
            + usize::from(self.lkg_snapshot_complete)
@@ -631,23 +536,11 @@ impl Serialize for DataUsageCacheInfo {
        state.serialize_entry("failed_objects", &self.failed_objects)?;
        state.serialize_entry("scan_resume_after", &self.scan_resume_after)?;
        state.serialize_entry("scan_checkpoint", &self.scan_checkpoint)?;
-        if let Some(identity) = self.scan_identity {
-            state.serialize_entry("scan_identity", &identity)?;
-        }
-        if let Some(progress) = self.scan_progress {
-            state.serialize_entry("scan_progress", &progress)?;
-        }
-        if let Some(receipt) = &self.scan_coverage_receipt {
-            state.serialize_entry("scan_coverage_receipt", receipt)?;
-        }
        state.serialize_entry("pending_heals", &self.pending_heals)?;
        state.serialize_entry("object_lock", &self.object_lock)?;
        state.serialize_entry("source", &self.source)?;
        state.serialize_entry("snapshot_complete", &self.snapshot_complete)?;
        state.serialize_entry("scan_plan_digest", &self.scan_plan_digest)?;
-        if let Some(coverage) = self.scan_coverage_digest {
-            state.serialize_entry("scan_coverage_digest", &coverage)?;
-        }
        state.serialize_entry("cache_key_format", &self.cache_key_format)?;
        if let Some(generation) = self.tier_registry_generation {
            state.serialize_entry("tier_registry_generation", &generation)?;
@@ -810,174 +703,6 @@ impl DataUsageCache {
        }
    }

-    pub(crate) fn prepare_bucket_checkpoint(
-        &mut self,
-        name: &str,
-        next_cycle: u64,
-        leader_epoch: u64,
-        source: DataUsageCacheSource,
-        scan_plan_digest: DataUsageScanPlanDigest,
-        identity: DataUsageScanIdentity,
-    ) -> DataUsageCachePrepareOutcome {
-        if self.info.next_cycle > next_cycle {
-            return DataUsageCachePrepareOutcome::RejectedNewerCycle;
-        }
-        if self.info.leader_epoch > leader_epoch {
-            return DataUsageCachePrepareOutcome::RejectedNewerLeader;
-        }
-        let reusable = identity.is_valid()
-            && name != DATA_USAGE_ROOT
-            && self.info.name == name
-            && self.info.source == Some(source)
-            && self.info.leader_epoch == leader_epoch
-            && self.info.cache_key_format == DATA_USAGE_CACHE_KEY_FORMAT
-            && self.info.scan_identity == Some(identity)
-            && self.info.tier_registry_generation == Some(identity.tier_registry_generation)
-            && (self.cache.is_empty() || self.checked_flatten_complete_scope(name).is_some());
-        if reusable
-            && self.info.snapshot_complete
-            && self.info.scan_progress.is_none()
-            && self.info.scan_checkpoint.is_none()
-            && self.info.scan_resume_after.is_none()
-            && self.info.scan_coverage_receipt.is_none()
-            && self.info.scan_plan_digest == Some(scan_plan_digest)
-        {
-            return self.prepare_for_scan(name, next_cycle, leader_epoch, source, scan_plan_digest, true);
-        }
-        if !reusable {
-            let keep_debts = self.info.name == name
-                && self
-                    .info
-                    .scan_identity
-                    .is_none_or(|previous| previous.bucket_incarnation == identity.bucket_incarnation);
-            let (pending_heals, size_reconciliation) = if keep_debts {
-                (
-                    std::mem::take(&mut self.info.pending_heals),
-                    std::mem::take(&mut self.info.size_reconciliation),
-                )
-            } else {
-                (Vec::new(), HashMap::new())
-            };
-            *self = Self::default();
-            self.info.pending_heals = pending_heals;
-            self.info.size_reconciliation = size_reconciliation;
-        }
-        let cursor_is_valid = (self.info.scan_checkpoint.is_none()
-            && self.info.scan_resume_after.is_none()
-            && self.info.scan_coverage_receipt.is_none())
-            || self.validated_scan_frontier().is_some();
-        if !cursor_is_valid {
-            self.info.scan_progress = None;
-        }
-        self.info.name = name.to_owned();
-        self.info.next_cycle = next_cycle;
-        self.info.leader_epoch = leader_epoch;
-        self.info.source = Some(source);
-        self.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT;
-        self.info.tier_registry_generation = Some(identity.tier_registry_generation);
-        self.info.scan_identity = Some(identity);
-        self.info.snapshot_complete = false;
-        if let Some(progress) = &mut self.info.scan_progress {
-            progress.requested_plan = scan_plan_digest;
-        } else {
-            self.info.scan_progress = Some(DataUsageScanProgress {
-                started_plan: scan_plan_digest,
-                requested_plan: scan_plan_digest,
-            });
-            self.info.scan_resume_after = None;
-            self.info.scan_checkpoint = None;
-            self.info.scan_coverage_receipt = None;
-        }
-        // Old readers do not understand coverage sweeps. An absent plan makes
-        // their existing prepare path rebuild instead of promoting mixed data.
-        self.info.scan_plan_digest = None;
-        if reusable {
-            DataUsageCachePrepareOutcome::Reused
-        } else {
-            DataUsageCachePrepareOutcome::Reset
-        }
-    }
-
-    fn coverage_prefix_digest(&self, through: &str) -> Result<[u8; 32], serde_json::Error> {
-        let mut writer = CheckpointDigestWriter(Sha256::new());
-        serde_json::to_writer(
-            &mut writer,
-            &(
-                &self.info.name,
-                self.info.scan_identity,
-                self.info.source,
-                self.info.leader_epoch,
-                self.info.cache_key_format,
-                self.info.scan_progress.map(|progress| progress.started_plan),
-                through,
-            ),
-        )?;
-        let mut prefix = self
-            .cache
-            .iter()
-            .filter(|(key, _)| {
-                let ancestor = through
-                    .strip_prefix(key.as_str())
-                    .is_some_and(|suffix| suffix.starts_with('/'));
-                let descendant = key.strip_prefix(through).is_some_and(|suffix| suffix.starts_with('/'));
-                (key.as_str() <= through && !ancestor) || descendant
-            })
-            .collect::<Vec<_>>();
-        prefix.sort_unstable_by_key(|(key, _)| *key);
-        for (key, entry) in prefix {
-            let mut value = serde_json::to_value(entry)?;
-            value.sort_all_objects();
-            if let Some(children) = value.get_mut("children").and_then(serde_json::Value::as_array_mut) {
-                children.sort_unstable_by(|left, right| left.as_str().cmp(&right.as_str()));
-            }
-            serde_json::to_writer(&mut writer, &(key, value))?;
-        }
-        Ok(writer.0.finalize().into())
-    }
-
-    pub(crate) fn validated_scan_frontier(&self) -> Option<&str> {
-        let receipt = self.info.scan_coverage_receipt.as_ref()?;
-        let checkpoint = self.info.scan_checkpoint.as_ref()?;
-        (self.info.scan_progress.is_some()
-            && self.info.scan_identity.is_some_and(|identity| identity.is_valid())
-            && self.info.source.is_some()
-            && receipt.through.len() <= 16 * 1024
-            && checkpoint.version == DATA_USAGE_SCAN_CHECKPOINT_VERSION
-            && checkpoint.resume_after == receipt.through
-            && self.info.scan_resume_after.as_deref() == Some(receipt.through.as_str())
-            && receipt
-                .through
-                .strip_prefix(&self.info.name)
-                .is_some_and(|suffix| suffix.starts_with('/'))
-            && self.find(&receipt.through).is_some()
-            && self.coverage_prefix_digest(&receipt.through).ok() == Some(receipt.digest))
-        .then_some(receipt.through.as_str())
-    }
-
-    /// Seal only the frontier supplied by completed traversal, never a restored cursor.
-    pub(crate) fn seal_scan_frontier(&mut self, frontier: Option<&str>) -> Result<(), serde_json::Error> {
-        if self.info.scan_progress.is_none() {
-            self.info.scan_coverage_receipt = None;
-            return Ok(());
-        }
-        let frontier = frontier.filter(|path| path.len() <= 16 * 1024 && self.find(path).is_some());
-        self.info.scan_coverage_receipt = match frontier {
-            Some(through) => Some(DataUsageScanCoverageReceipt {
-                through: through.to_owned(),
-                digest: self.coverage_prefix_digest(through)?,
-            }),
-            None => None,
-        };
-        self.info.scan_resume_after = frontier.map(str::to_owned);
-        let reason = self
-            .info
-            .scan_checkpoint
-            .as_ref()
-            .map_or(DataUsageScanCheckpointReason::Unknown, |checkpoint| checkpoint.reason);
-        self.info.scan_checkpoint = frontier.map(|through| DataUsageScanCheckpoint::new(through.to_owned(), reason));
-        Ok(())
-    }
-
    fn ensure_cache_save_metrics_registered() {
        CACHE_SAVE_METRICS_ONCE.call_once(|| {
            describe_counter!(
@@ -1057,29 +782,6 @@ impl DataUsageCache {
        (visited == expected_entries).then_some(entry)
    }

-    pub(crate) fn has_complete_root_inventory(&self, bucket_keys: &HashSet<String>) -> bool {
-        let Some(root) = self.find(DATA_USAGE_ROOT) else {
-            return false;
-        };
-        // Set roots only connect bucket entries. Scalar data at the root, an
-        // extra bucket, or an orphan must not disappear during bucket folding.
-        root.children.len() == bucket_keys.len()
-            && bucket_keys.iter().all(|key| root.children.contains(key))
-            && root.size == 0
-            && root.objects == 0
-            && root.versions == 0
-            && root.delete_markers == 0
-            && root.failed_objects == 0
-            && !root.compacted
-            && root.obj_sizes.is_empty()
-            && root.obj_versions.is_empty()
-            && root.replication_stats.is_none()
-            && root.all_tier_stats.is_none()
-            && root.unknown_tier_stats.is_none()
-            && root.tier_accounting_proof.is_none()
-            && self.checked_flatten_complete(DATA_USAGE_ROOT).is_some()
-    }
-
    fn checked_flatten_inner(&self, path: &str) -> Option<(DataUsageEntry, usize)> {
        let root_key = hash_path(path).key();
        let (root_key, root) = self.cache.get_key_value(&root_key)?;
@@ -29,31 +29,6 @@ use tokio::sync::Mutex;

 const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([3; 32]);

-#[test]
-fn scoped_scan_coverage_metadata_preserves_map_compatibility() {
-    #[derive(serde::Deserialize)]
-    struct LegacyInfo {
-        name: String,
-        next_cycle: u64,
-    }
-    let mut info = DataUsageCacheInfo {
-        name: DATA_USAGE_ROOT.to_string(),
-        next_cycle: 7,
-        ..Default::default()
-    };
-    let old = serde_json::to_value(&info).expect("legacy metadata should encode");
-    assert!(old.get("scan_coverage_digest").is_none());
-    let old: DataUsageCacheInfo = serde_json::from_value(old).expect("missing coverage must remain readable");
-    assert!(old.scan_coverage_digest.is_none());
-    info.scan_coverage_digest = Some(TEST_PLAN_DIGEST);
-    let encoded = rmp_serde::to_vec(&info).expect("coverage metadata should remain map encoded");
-    let legacy: LegacyInfo = rmp_serde::from_slice(&encoded).expect("old map readers should ignore additive proof fields");
-    assert_eq!(legacy.name, DATA_USAGE_ROOT);
-    assert_eq!(legacy.next_cycle, 7);
-    let decoded: DataUsageCacheInfo = rmp_serde::from_slice(&encoded).expect("new reader should restore the coverage proof");
-    assert_eq!(decoded.scan_coverage_digest, Some(TEST_PLAN_DIGEST));
-}
-
 #[derive(Debug, PartialEq, Eq)]
 struct CachePutRecord {
    object: String,
@@ -728,15 +728,6 @@ async fn scan_and_persist_local_bucket(
        DataUsageCacheReuseOptions {
            require_source: true,
            tier_registry_generation: Some(tier_registry_generation),
-            checkpoint_identity: crate::scanner_io::scanner_bucket_checkpoint_identity(
-                &set,
-                &bucket,
-                expected_publication_epoch,
-                tier_registry_generation,
-                scan_mode,
-            )
-            .await
-            .ok(),
        },
    );
    match scan_state {
@@ -830,22 +821,6 @@ async fn scan_and_persist_local_bucket(
        ScannerDiskScanOutcome::Partial(cache) => (cache, Some(RemoteScannerFrameResult::Partial)),
        ScannerDiskScanOutcome::NamespaceNotFound(cache) => (cache, Some(RemoteScannerFrameResult::NamespaceNotFound)),
    };
-    if let Some(expected) = cache.info.scan_identity
-        && crate::scanner_io::scanner_bucket_checkpoint_identity(
-            &set,
-            &bucket,
-            expected_publication_epoch,
-            tier_registry_generation,
-            scan_mode,
-        )
-        .await
-        .ok()
-            != Some(expected)
-    {
-        return Err(RemoteScannerServerError::retry_bucket(
-            "remote scanner checkpoint identity changed during scanning",
-        ));
-    }

    if guard.is_lock_lost() {
        return Err(RemoteScannerServerError::worker(
@@ -1086,12 +1086,6 @@ impl ScannerMaintenanceFeatures {
    fn needs_regular_cycle(self) -> bool {
        self.lifecycle || self.replication || self.inspection_failed
    }
-
-    fn requires_full_scan(self, observed_generation: Option<u64>, current_generation: u64, wake: ScannerCycleWakeReason) -> bool {
-        self.needs_regular_cycle()
-            || observed_generation != Some(current_generation)
-            || !matches!(wake, ScannerCycleWakeReason::DirtyUsage | ScannerCycleWakeReason::ClusterActivity)
-    }
 }

 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -1289,18 +1283,18 @@ async fn configure_scanner_defaults(
    ctx: &CancellationToken,
    storeapi: &Arc<impl ScannerStorage>,
 ) -> (ScannerMaintenanceFeatures, Option<u64>) {
-    let (features, maintenance_generation) = detect_stable_scanner_maintenance_features(ctx, storeapi)
-        .await
-        .unwrap_or_else(|| {
-            (
-                ScannerMaintenanceFeatures {
-                    inspection_failed: true,
-                    ..Default::default()
-                },
-                scanner_maintenance_generation(),
-            )
-        });
    if storeapi.setup_is_erasure_sd().await {
+        let (features, maintenance_generation) = detect_stable_scanner_maintenance_features(ctx, storeapi)
+            .await
+            .unwrap_or_else(|| {
+                (
+                    ScannerMaintenanceFeatures {
+                        inspection_failed: true,
+                        ..Default::default()
+                    },
+                    scanner_maintenance_generation(),
+                )
+            });
        // Single-disk keeps the speed-preset-derived default cycle (60s at the
        // `default` preset) instead of a special shorter cycle: no measured
        // cold-start ILM latency basis for an override, and clean-idle backoff
@@ -1325,7 +1319,7 @@ async fn configure_scanner_defaults(
    } else {
        set_scanner_default_speed(ScannerSpeed::Default);
        set_scanner_default_cycle_secs(None);
-        (features, Some(maintenance_generation))
+        (ScannerMaintenanceFeatures::default(), None)
    }
 }

@@ -1570,7 +1564,7 @@ where
    S: ScannerStorage,
 {
    let cycle_budget = ScannerCycleBudget::new(ctx, scanner_cycle_budget_config());
-    run_data_scanner_cycle_with_budget(ctx, storeapi, cycle_info, cycle_revision, leader_epoch, cycle_budget, true).await
+    run_data_scanner_cycle_with_budget(ctx, storeapi, cycle_info, cycle_revision, leader_epoch, cycle_budget).await
 }

 #[instrument(skip_all)]
@@ -1582,7 +1576,6 @@ async fn run_data_scanner_cycle_with_budget<S>(
    cycle_revision: &mut DataUsageCacheRevision,
    leader_epoch: u64,
    cycle_budget: Arc<ScannerCycleBudget>,
-    requires_full_scan: bool,
 ) -> ScannerCycleOutcome
 where
    S: ScannerStorage,
@@ -1721,9 +1714,6 @@ where
            scan_mode,
            scan_scope: crate::scanner_io::ScannerBucketScanScope::default(),
            persisted_usage_baseline: usage_persist_baseline.data.clone(),
-            requires_full_scan,
-            #[cfg(test)]
-            resolved_scope_observer: None,
        },
    )
    .await;
@@ -2584,7 +2574,10 @@ where
    let mut superseded_backoff = ScannerRetryBackoff::default();
    let mut deferred_backoff = ScannerRetryBackoff::default();
    let initial_runtime_config = resolve_scanner_runtime_config();
-    if clean_idle_topology_supported && maintenance_generation_seen.is_none() {
+    if clean_idle_topology_supported
+        && scanner_clean_idle_backoff_configured(&initial_runtime_config)
+        && maintenance_generation_seen.is_none()
+    {
        let Some((features, generation)) = detect_stable_scanner_maintenance_features(&ctx, &storeapi).await else {
            global_metrics().set_cycle(None).await;
            finish_scanner_leader_iteration(false, "stopped", String::new()).await;
@@ -2792,7 +2785,6 @@ where
                &mut cycle_revision,
                leader_epoch,
                cycle_budget.clone(),
-                true,
            ),
            guard.lock_lost_notified(),
        )
@@ -2887,7 +2879,7 @@ where
        #[cfg(test)]
        notify_scanner_runtime_observed_for_test(&storeapi, pause_backlog_observation);
        let runtime_config = resolve_scanner_runtime_config();
-        if clean_idle_topology_supported {
+        if clean_idle_topology_supported && scanner_clean_idle_backoff_configured(&runtime_config) {
            let current_generation = scanner_maintenance_generation();
            if maintenance_generation_seen != Some(current_generation) {
                scanner_activity_seen = None;
@@ -3083,11 +3075,6 @@ where
                &mut cycle_revision,
                leader_epoch,
                cycle_budget.clone(),
-                maintenance_features.requires_full_scan(
-                    maintenance_generation_seen,
-                    scanner_maintenance_generation(),
-                    wake_reason,
-                ),
            ),
            guard.lock_lost_notified(),
        )
@@ -3145,7 +3132,10 @@ where
        let maintenance_config_changed =
            maintenance_generation_seen.is_some_and(|generation| generation != current_maintenance_generation);
        let retry_failed_inspection = maintenance_inspection_retry.retry_due(maintenance_features, wake_reason, Instant::now());
-        if clean_idle_topology_supported && (maintenance_config_changed || retry_failed_inspection) {
+        if clean_idle_topology_supported
+            && scanner_clean_idle_backoff_configured(&runtime_config)
+            && (maintenance_config_changed || retry_failed_inspection)
+        {
            let Some((features, generation)) = detect_stable_scanner_maintenance_features(&ctx, &storeapi).await else {
                break;
            };
@@ -1204,7 +1204,7 @@ async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledgin
    let mut revision = DataUsageCacheRevision::Missing;
    let outcome = tokio::time::timeout(
        Duration::from_secs(30),
-        run_data_scanner_cycle_with_budget(&ctx, &store, &mut cycle_info, &mut revision, 1, Arc::clone(&budget), true),
+        run_data_scanner_cycle_with_budget(&ctx, &store, &mut cycle_info, &mut revision, 1, Arc::clone(&budget)),
    )
    .await
    .expect("the coordinator must finish its namespace walk while a PUT is pending");
@@ -1240,7 +1240,7 @@ async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledgin
    let retry_budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, ScannerCycleBudgetConfig::default());
    let outcome = tokio::time::timeout(
        Duration::from_secs(30),
-        run_data_scanner_cycle_with_budget(&ctx, &store, &mut cycle_info, &mut revision, 1, Arc::clone(&retry_budget), true),
+        run_data_scanner_cycle_with_budget(&ctx, &store, &mut cycle_info, &mut revision, 1, Arc::clone(&retry_budget)),
    )
    .await
    .expect("the same cycle must converge after the pending PUT drains");
@@ -4869,28 +4869,6 @@ async fn usage_bootstrap_does_not_overwrite_concurrent_replacement() {
 #[serial]
 async fn scanner_usage_state_reset_publishes_fenced_bootstrap_marker() {
    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    let quota_ledger_path = "config/quota-ledger/reserved-bucket.json";
-    let quota_ledger = serde_json::to_vec(&serde_json::json!({
-        "version": 1,
-        "bucket_incarnation": "00000000-0000-0000-0000-000000000001",
-        "quota_revision_unix_nanos": 1,
-        "accounted_usage": 100,
-        "reservations": {
-            "00000000-0000-0000-0000-000000000002": {
-                "object": "pending-object",
-                "old_size": 0,
-                "new_size": 64,
-                "created_at": 1,
-                "pool_index": 0,
-                "set_index": 0,
-                "commit_started": true
-            }
-        }
-    }))
-    .expect("quota ledger fixture should encode");
-    save_config(store.clone(), quota_ledger_path, quota_ledger.clone())
-        .await
-        .expect("independent quota reservations should persist");
    let cycle = CurrentCycle {
        current: 41,
        next: 42,
@@ -4949,14 +4927,6 @@ async fn scanner_usage_state_reset_publishes_fenced_bootstrap_marker() {
    assert!(!data_usage_info_has_persisted_baseline_identity(&usage));
    assert_eq!(usage.scanner_epoch, Some(9));

-    assert_eq!(
-        read_config(store.clone(), quota_ledger_path)
-            .await
-            .expect("quota ledger must remain readable after scanner reset"),
-        quota_ledger,
-        "scanner reset must preserve incarnation and outstanding reserved bytes exactly"
-    );
-
    for path in [
        usage_backup_path.as_str(),
        LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str(),
@@ -8238,76 +8208,6 @@ fn clean_idle_backoff_policy_preserves_explicit_and_maintenance_cycles() {
    }
 }

-#[tokio::test]
-#[serial]
-async fn scoped_scan_explicit_bitrot_keeps_dirty_planning_without_idle_backoff() {
-    temp_env::async_with_vars([(ENV_SCANNER_CYCLE, None), (ENV_SCANNER_BITROT_CYCLE_SECS, Some("3600"))], async {
-        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
-        crate::scanner_io::clear_dirty_usage_buckets_for_tests();
-        let (_temp_dir, store) = setup_scanner_cycle_store_with_pool_count(true, 2).await;
-        let ctx = CancellationToken::new();
-        let (features, generation) = configure_scanner_defaults(&ctx, &store).await;
-        let config = resolve_scanner_runtime_config();
-        assert_eq!(config.cycle_interval_source, ScannerRuntimeConfigSource::Default);
-        assert_eq!(config.bitrot_cycle_source, ScannerRuntimeConfigSource::Env);
-        assert!(!scanner_clean_idle_backoff_configured(&config));
-        assert!(!features.needs_regular_cycle());
-        assert_eq!(
-            generation,
-            Some(scanner_maintenance_generation()),
-            "multi-disk startup must inspect maintenance independently"
-        );
-        let observed = ScannerCycleObservedGenerations::for_wait(&config, None, 7, 0, scanner_maintenance_generation());
-        assert_eq!(observed.dirty_usage, Some(7), "explicit bitrot still permits ordinary dirty wakeups");
-        for (wake, full) in [
-            (ScannerCycleWakeReason::DirtyUsage, false),
-            (ScannerCycleWakeReason::ClusterActivity, false),
-            (ScannerCycleWakeReason::Timer, true),
-            (ScannerCycleWakeReason::ClusterMaintenance, true),
-        ] {
-            assert_eq!(
-                features.requires_full_scan(generation, scanner_maintenance_generation(), wake),
-                full,
-                "{wake:?}"
-            );
-        }
-        assert!(features.requires_full_scan(None, scanner_maintenance_generation(), ScannerCycleWakeReason::DirtyUsage));
-        for unsafe_features in [
-            ScannerMaintenanceFeatures {
-                lifecycle: true,
-                ..Default::default()
-            },
-            ScannerMaintenanceFeatures {
-                replication: true,
-                ..Default::default()
-            },
-            ScannerMaintenanceFeatures {
-                inspection_failed: true,
-                ..Default::default()
-            },
-        ] {
-            assert!(unsafe_features.requires_full_scan(
-                generation,
-                scanner_maintenance_generation(),
-                ScannerCycleWakeReason::DirtyUsage
-            ));
-        }
-        crate::scanner_io::record_scanner_maintenance_change("maintenance-proof-change");
-        assert!(features.requires_full_scan(generation, scanner_maintenance_generation(), ScannerCycleWakeReason::DirtyUsage));
-        let (refreshed, refreshed_generation) = detect_stable_scanner_maintenance_features(&ctx, &store)
-            .await
-            .expect("changed maintenance generation should be inspected");
-        assert!(!refreshed.requires_full_scan(
-            Some(refreshed_generation),
-            scanner_maintenance_generation(),
-            ScannerCycleWakeReason::DirtyUsage
-        ));
-        crate::scanner_io::clear_dirty_usage_buckets_for_tests();
-    })
-    .await;
-    crate::runtime_config::refresh_scanner_runtime_config_for_tests();
-}
-
 #[test]
 fn clean_idle_backoff_requires_activity_probes() {
    let default_config = ScannerRuntimeConfig::default();
@@ -8696,63 +8596,6 @@ fn scanner_node_activity(epoch: &str, namespace_generation: u64, maintenance_gen
    }
 }

-#[test]
-fn scoped_scan_remote_dirty_coverage_invalidates_local_bucket_current() {
-    let before = BTreeMap::from([("remote".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
-    let mut after = before.clone();
-    let remote = after.get_mut("remote").expect("remote activity should exist");
-    remote.dirty_usage_generation += 1;
-    remote.dirty_usage_pending = true;
-    assert_eq!(scanner_activity_structural_digest(&before), scanner_activity_structural_digest(&after));
-    let old_plan = crate::scanner_io::checkpoint_fixture_bucket_digest(
-        DataUsageScanPlanDigest(scanner_activity_snapshot_digest(&before)),
-        None,
-    );
-    let new_plan = crate::scanner_io::checkpoint_fixture_bucket_digest(
-        DataUsageScanPlanDigest(scanner_activity_snapshot_digest(&after)),
-        None,
-    );
-    assert_ne!(
-        old_plan, new_plan,
-        "remote dirty changes must fence Current even without a local bucket hint"
-    );
-    let source = DataUsageCacheSource::new(0, 0);
-    let mut cache = DataUsageCache {
-        info: crate::DataUsageCacheInfo {
-            name: "bucket".to_string(),
-            next_cycle: 7,
-            leader_epoch: 11,
-            source: Some(source),
-            last_update: Some(std::time::SystemTime::UNIX_EPOCH),
-            snapshot_complete: true,
-            scan_plan_digest: Some(old_plan),
-            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
-            ..Default::default()
-        },
-        ..Default::default()
-    };
-    cache.replace("bucket", "", DataUsageEntry::default());
-    assert!(matches!(
-        crate::scanner_io::current_cache_root_or_prepare_with_generation(
-            &mut cache,
-            "bucket",
-            source,
-            7,
-            11,
-            new_plan,
-            crate::scanner_io::DataUsageCacheReuseOptions {
-                require_source: true,
-                tier_registry_generation: None,
-                checkpoint_identity: None,
-            },
-        ),
-        crate::scanner_io::DataUsageCacheScanState::Prepared {
-            outcome: DataUsageCachePrepareOutcome::Reset,
-            ..
-        }
-    ));
-}
-
 #[test]
 fn post_lease_activity_proof_rejects_a_put_tail_that_finished_before_lease_acquisition() {
    let before = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
@@ -707,9 +707,6 @@ pub struct FolderScanner {
    /// next scan and cannot mix generations in one aggregate.
    tier_registry: TierRegistrySnapshot,
    pending_heals_changed: bool,
-    coverage_frontier: Option<String>,
-    resume_frontier: Option<String>,
-    coverage_gap: bool,
    pending_size_reconciliation_keys: HashSet<String>,
    pending_size_reconciliation_scopes: HashSet<String>,
    pending_size_reconciliation_truncated: bool,
@@ -909,16 +906,6 @@ impl FolderScanner {
        self.update_cache.info.scan_checkpoint = Some(checkpoint);
    }

-    fn record_completed_child(&mut self, folder: &str, healthy: bool) {
-        if self.old_cache.info.scan_progress.is_some() {
-            self.coverage_gap |= !healthy;
-            if !self.coverage_gap {
-                self.coverage_frontier = Some(folder.to_owned());
-            }
-        }
-        self.record_scan_resume_hint(folder);
-    }
-
    fn record_scan_resume_hint_if_not_ancestor(&mut self, folder: &str) {
        let keep_existing = self
            .new_cache
@@ -1484,7 +1471,6 @@ impl FolderScanner {
                // (e.g. in the get_size error branch below). This branch only accounts
                // for subsequent skips of already-failed paths.
                if self.should_skip_failed(&item.path) {
-                    self.coverage_gap |= self.old_cache.info.scan_progress.is_some();
                    continue;
                }

@@ -1498,7 +1484,6 @@ impl FolderScanner {
                        let failure_action = classify_get_size_failure(&item, &e);

                        if failure_action != GetSizeFailureAction::Skip {
-                            self.coverage_gap |= self.old_cache.info.scan_progress.is_some();
                            // Track failed objects to prevent infinite retry loops
                            into.failed_objects += 1;
                            self.record_failed(&item.path);
@@ -1591,9 +1576,6 @@ impl FolderScanner {
                abandoned_children.remove(&path_join_buf(&[&item.bucket, &item.object_path()]));

                apply_scanner_size_summary(into, &sz);
-                if !sz.size_reconciliation.is_empty() {
-                    self.coverage_gap |= self.old_cache.info.scan_progress.is_some();
-                }
                self.apply_size_reconciliation(&sz);
                into.objects += 1;
                object_count += 1;
@@ -1638,7 +1620,6 @@ impl FolderScanner {
            }

            if self.is_erasure_mode && found_erasure_data_directory && !found_object_metadata {
-                self.coverage_gap |= self.old_cache.info.scan_progress.is_some();
                found_object_metadata = true;
                let metadata_path = path_join_buf(&[&dir_path, STORAGE_FORMAT_FILE]);

@@ -1760,13 +1741,7 @@ impl FolderScanner {
                source: FolderScanSource::Existing,
            }));
            let has_queued_folders = !queued_folders.is_empty();
-            let forward_sweep = self.old_cache.info.scan_progress.is_some();
-            let forward_resume_after = self.resume_frontier.clone();
-            let resume_order = if forward_sweep {
-                order_queued_folders_for_resume(&mut queued_folders, None)
-            } else {
-                order_queued_folders_for_resume(&mut queued_folders, scan_resume_after)
-            };
+            let resume_order = order_queued_folders_for_resume(&mut queued_folders, scan_resume_after);
            if checkpoint_tracks_child_order && has_queued_folders {
                match resume_order {
                    FolderResumeOrder::Used => global_metrics().record_scanner_checkpoint_used(),
@@ -1783,18 +1758,6 @@ impl FolderScanner {

                let mut folder_item = queued_folder.folder;
                let h = hash_path(&folder_item.name);
-                if forward_sweep
-                    && !into.compacted
-                    && forward_resume_after.as_deref().is_some_and(|resume| {
-                        folder_item.name.as_str() <= resume
-                            && !matches!(folder_resume_match(&folder_item.name, resume), Some(FolderResumeMatch::Descendant))
-                    })
-                    && self.old_cache.find(&folder_item.name).is_some()
-                {
-                    self.new_cache.copy_with_children(&self.old_cache, &h, &folder_item.parent);
-                    into.add_child(&h);
-                    continue;
-                }

                match queued_folder.source {
                    FolderScanSource::New => {
@@ -1826,7 +1789,7 @@ impl FolderScanner {
                        }
                    }
                    FolderScanSource::Existing => {
-                        if !forward_sweep && !into.compacted && self.old_cache.is_compacted(&h) {
+                        if !into.compacted && self.old_cache.is_compacted(&h) {
                            let next_cycle = self.old_cache.info.next_cycle as u32;
                            if !h.mod_(next_cycle, data_usage_update_dir_cycles()) {
                                // Transfer and add as child...
@@ -1847,7 +1810,7 @@ impl FolderScanner {
                    // In compacted mode child totals are accumulated directly into the parent entry.
                    let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), into));
                    fut.await.map_err(|e| ScannerError::Other(e.to_string()))?;
-                    self.record_completed_child(&folder_item.name, into.failed_objects == 0);
+                    self.record_scan_resume_hint(&folder_item.name);
                    self.send_update_for_entry(&this_hash, &folder.parent, into).await;
                    tokio::task::yield_now().await;
                } else {
@@ -1872,13 +1835,12 @@ impl FolderScanner {
                            error = %e,
                            "Scanner child folder scan failed"
                        );
-                        self.coverage_gap |= forward_sweep;
                        continue;
                    }
                    tokio::task::yield_now().await;

                    into.add_child(&h);
-                    self.record_completed_child(&folder_item.name, dst.failed_objects == 0);
+                    self.record_scan_resume_hint(&folder_item.name);
                    // We scanned a folder, optionally send update.
                    self.update_cache.delete_recursive(&h);
                    self.update_cache.copy_with_children(&self.new_cache, &h, &folder_item.parent);
@@ -2288,10 +2250,7 @@ impl FolderScanner {
            self.new_cache.replace_hashed(&this_hash, &folder.parent, into);
        }

-        // Keep independently accounted children while the sweep cursor may
-        // reference them; the hard cardinality compaction below still applies.
        if !into.compacted
-            && self.old_cache.info.scan_progress.is_none()
            && self.new_cache.info.name != folder.name
            && let Some(mut flat) = self.new_cache.size_recursive(&this_hash.key())
        {
@@ -2400,7 +2359,6 @@ pub async fn scan_data_folder(
    cache.fold_retired_tiers(&tier_registry.names);
    cache.info.tier_registry_generation = Some(tier_registry.generation);

-    let resume_frontier = cache.validated_scan_frontier().map(str::to_owned);
    // Create folder scanner
    let mut scanner = FolderScanner {
        root: base_path,
@@ -2430,9 +2388,6 @@ pub async fn scan_data_folder(
        local_disk,
        tier_registry,
        pending_heals_changed: false,
-        coverage_frontier: resume_frontier.clone(),
-        resume_frontier,
-        coverage_gap: false,
        pending_size_reconciliation_keys: HashSet::new(),
        pending_size_reconciliation_scopes: HashSet::new(),
        pending_size_reconciliation_truncated: false,
@@ -2463,41 +2418,23 @@ pub async fn scan_data_folder(
    match scanner.scan_folder(ctx.clone(), folder, &mut root).await {
        Ok(()) => {
            // Get the new cache and finalize it
-            let coverage_gap = scanner.coverage_gap;
            let new_cache = scanner.as_mut_new_cache();
            new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
            new_cache.info.last_update = Some(SystemTime::now());
            new_cache.info.next_cycle = cache.info.next_cycle;
-            let unresolved_objects = coverage_gap
-                || root.failed_objects > 0
+            let unresolved_objects = root.failed_objects > 0
                || !new_cache.info.failed_objects.is_empty()
                || !new_cache.info.size_reconciliation.is_empty();
-            let mixed_coverage = new_cache
-                .info
-                .scan_progress
-                .is_some_and(|progress| progress.started_plan != progress.requested_plan);
-            new_cache.info.snapshot_complete = !unresolved_objects && !mixed_coverage;
-            if let Some(progress) = &mut new_cache.info.scan_progress {
-                if new_cache.info.snapshot_complete {
-                    new_cache.info.scan_plan_digest = Some(progress.requested_plan);
-                    new_cache.info.scan_progress = None;
-                } else {
-                    // Retain observations, then verify from the beginning under
-                    // the latest plan. A clean tail cannot certify an old prefix.
-                    progress.started_plan = progress.requested_plan;
-                    new_cache.info.scan_plan_digest = None;
-                }
-            }
+            new_cache.info.snapshot_complete = !unresolved_objects;
            let had_scan_checkpoint = cache.info.scan_checkpoint.is_some() || new_cache.info.scan_checkpoint.is_some();
            new_cache.info.scan_resume_after = None;
            new_cache.info.scan_checkpoint = None;
-            new_cache.info.scan_coverage_receipt = None;
            if had_scan_checkpoint {
                global_metrics().record_scanner_checkpoint_cleared();
            }

            close_disk_guard.close().await;
-            if unresolved_objects || mixed_coverage {
+            if unresolved_objects {
                Err(ScannerError::PartialCache(Box::new(new_cache.clone())))
            } else {
                Ok(new_cache.clone())
@@ -2511,7 +2448,6 @@ pub async fn scan_data_folder(
                if root_has_progress {
                    scanner.carry_forward_old_children(&root_hash, &mut root);
                }
-                let coverage_frontier = scanner.coverage_frontier.clone();
                let new_cache = scanner.as_mut_new_cache();
                if root_has_progress {
                    new_cache.replace_hashed(&root_hash, &None, &root);
@@ -2526,18 +2462,6 @@ pub async fn scan_data_folder(
                    if root_has_progress {
                        set_scan_checkpoint(new_cache, checkpoint_reason_from_budget(budget.reason()));
                    }
-                    new_cache.seal_scan_frontier(coverage_frontier.as_deref())?;
-                    if new_cache.info.scan_progress.is_some() {
-                        if let Some(checkpoint) = &new_cache.info.scan_checkpoint {
-                            global_metrics().record_scanner_checkpoint_set(
-                                checkpoint.version,
-                                checkpoint.resume_after.clone(),
-                                checkpoint.reason.as_str(),
-                            );
-                        } else {
-                            global_metrics().record_scanner_checkpoint_cleared();
-                        }
-                    }
                    close_disk_guard.close().await;
                    return Err(ScannerError::PartialCache(Box::new(new_cache.clone())));
                }
@@ -346,9 +346,6 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
            refresh_failed: false,
        },
        pending_heals_changed: false,
-        coverage_frontier: None,
-        resume_frontier: None,
-        coverage_gap: false,
        pending_size_reconciliation_keys: HashSet::new(),
        pending_size_reconciliation_scopes: HashSet::new(),
        pending_size_reconciliation_truncated: false,
@@ -234,531 +234,6 @@ fn checkpoint_fixture_compaction_preserves_aggregate_not_child_enumeration() {
    );
 }

-fn bound_checkpoint() -> (DataUsageCache, crate::DataUsageScanIdentity) {
-    let identity = crate::DataUsageScanIdentity {
-        version: 1,
-        bucket_incarnation: Uuid::from_u128(1),
-        set_layout: DataUsageScanPlanDigest([41; 32]),
-        publication_epoch: 0,
-        tier_registry_generation: 7,
-        scan_mode: HealScanMode::Normal,
-    };
-    let mut cache = DataUsageCache::default();
-    cache.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, identity);
-    cache.replace("bucket", "", DataUsageEntry::default());
-    cache.replace(
-        "bucket/static",
-        "bucket",
-        DataUsageEntry {
-            objects: 3,
-            ..Default::default()
-        },
-    );
-    cache.info.scan_resume_after = Some("bucket/static".into());
-    cache.info.scan_checkpoint = Some(DataUsageScanCheckpoint::new(
-        "bucket/static".into(),
-        DataUsageScanCheckpointReason::Objects,
-    ));
-    cache
-        .seal_scan_frontier(Some("bucket/static"))
-        .expect("completed fixture prefix receipt");
-    (cache, identity)
-}
-
-#[test]
-fn checkpoint_fixture_roundtrip_retains_verified_scope_but_old_reader_rebuilds() {
-    let (cache, identity) = bound_checkpoint();
-    let mut cache = decode_fixture(&cache.marshal_msg().expect("encode bound progress")).expect("read bound progress");
-    let next_plan = DataUsageScanPlanDigest([42; 32]);
-    assert_eq!(
-        cache.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, next_plan, identity),
-        crate::DataUsageCachePrepareOutcome::Reused
-    );
-    assert_eq!(retained(&cache), 3);
-    assert_eq!(cache.info.scan_identity, Some(identity));
-    assert_eq!(
-        cache.info.scan_progress,
-        Some(crate::DataUsageScanProgress {
-            started_plan: PLAN,
-            requested_plan: next_plan
-        })
-    );
-    assert!(cache.info.scan_plan_digest.is_none());
-    let mut old_wire = serde_json::to_value(&cache).expect("map-encoded compatibility fixture");
-    let old_info = old_wire["info"].as_object_mut().expect("cache info is a map");
-    old_info.remove("scan_identity");
-    old_info.remove("scan_progress");
-    old_info.remove("scan_coverage_receipt");
-    let mut old_view: DataUsageCache = serde_json::from_value(old_wire).expect("old writer drops unknown metadata");
-    assert_eq!(
-        old_view.prepare_for_scan("bucket", 11, 7, SOURCE, next_plan, true),
-        crate::DataUsageCachePrepareOutcome::Reset
-    );
-    assert!(old_view.cache.is_empty());
-    assert!(!old_view.info.snapshot_complete);
-}
-
-#[test]
-fn checkpoint_fixture_optional_coverage_metadata_roundtrips_together() {
-    let (mut cache, _) = bound_checkpoint();
-    cache.info.scan_coverage_digest = Some(DataUsageScanPlanDigest([77; 32]));
-    let encoded = cache.marshal_msg().expect("encode every optional coverage field");
-    let decoded = DataUsageCache::unmarshal(&encoded).expect("decode combined W03/W04 metadata map");
-    assert_eq!(decoded.info.scan_identity, cache.info.scan_identity);
-    assert_eq!(decoded.info.scan_progress, cache.info.scan_progress);
-    assert_eq!(decoded.info.scan_coverage_receipt, cache.info.scan_coverage_receipt);
-    assert_eq!(decoded.info.scan_coverage_digest, cache.info.scan_coverage_digest);
-    assert_eq!(decoded.validated_scan_frontier(), Some("bucket/static"));
-    assert!(!decoded.info.snapshot_complete);
-}
-
-#[test]
-fn checkpoint_fixture_unchanged_complete_plan_keeps_existing_rescan_policy() {
-    let (mut cache, identity) = bound_checkpoint();
-    cache.info.scan_progress = None;
-    cache.info.scan_plan_digest = Some(PLAN);
-    cache.info.scan_resume_after = None;
-    cache.info.scan_checkpoint = None;
-    cache.info.scan_coverage_receipt = None;
-    cache.info.snapshot_complete = true;
-    assert_eq!(
-        cache.prepare_bucket_checkpoint("bucket", 12, 7, SOURCE, PLAN, identity),
-        crate::DataUsageCachePrepareOutcome::Reused
-    );
-    assert!(
-        cache.info.scan_progress.is_none(),
-        "unchanged complete coverage needs no forced verification sweep"
-    );
-    assert_eq!(cache.info.scan_plan_digest, Some(PLAN));
-    assert_eq!(retained(&cache), 3);
-    let next = DataUsageScanPlanDigest([44; 32]);
-    cache.prepare_bucket_checkpoint("bucket", 12, 7, SOURCE, next, identity);
-    assert_eq!(cache.info.scan_progress.expect("changed plan must be verified").started_plan, next);
-    assert!(cache.info.scan_plan_digest.is_none());
-    assert!(!cache.info.snapshot_complete);
-}
-
-#[test]
-fn checkpoint_fixture_identity_changes_and_future_state_fail_closed() {
-    let (cache, identity) = bound_checkpoint();
-    for next_identity in [
-        crate::DataUsageScanIdentity {
-            bucket_incarnation: Uuid::from_u128(2),
-            ..identity
-        },
-        crate::DataUsageScanIdentity {
-            set_layout: DataUsageScanPlanDigest([9; 32]),
-            ..identity
-        },
-        crate::DataUsageScanIdentity {
-            publication_epoch: 1,
-            ..identity
-        },
-        crate::DataUsageScanIdentity {
-            tier_registry_generation: 8,
-            ..identity
-        },
-        crate::DataUsageScanIdentity {
-            scan_mode: HealScanMode::Deep,
-            ..identity
-        },
-    ] {
-        let mut next = cache.clone();
-        assert_eq!(
-            next.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, next_identity),
-            crate::DataUsageCachePrepareOutcome::Reset
-        );
-        assert!(next.cache.is_empty());
-        assert!(next.info.scan_checkpoint.is_none());
-        assert!(!next.info.snapshot_complete);
-    }
-    for (source, epoch) in [(crate::DataUsageCacheSource::new(1, 0), 7), (SOURCE, 8)] {
-        let mut next = cache.clone();
-        assert_eq!(
-            next.prepare_bucket_checkpoint("bucket", 11, epoch, source, PLAN, identity),
-            crate::DataUsageCachePrepareOutcome::Reset
-        );
-        assert!(next.cache.is_empty());
-    }
-    for (cycle, epoch, expected) in [
-        (10, 7, crate::DataUsageCachePrepareOutcome::RejectedNewerCycle),
-        (11, 6, crate::DataUsageCachePrepareOutcome::RejectedNewerLeader),
-    ] {
-        let mut next = cache.clone();
-        assert_eq!(next.prepare_bucket_checkpoint("bucket", cycle, epoch, SOURCE, PLAN, identity), expected);
-        assert_eq!(
-            serde_json::to_value(&next).expect("current cache"),
-            serde_json::to_value(&cache).expect("saved cache")
-        );
-    }
-    for invalid in [
-        crate::DataUsageScanIdentity { version: 2, ..identity },
-        crate::DataUsageScanIdentity {
-            bucket_incarnation: Uuid::nil(),
-            ..identity
-        },
-    ] {
-        let mut next = cache.clone();
-        crate::scanner_io::current_cache_root_or_prepare_with_generation(
-            &mut next,
-            "bucket",
-            SOURCE,
-            11,
-            7,
-            PLAN,
-            crate::scanner_io::DataUsageCacheReuseOptions {
-                checkpoint_identity: Some(invalid),
-                ..Default::default()
-            },
-        );
-        assert!(next.cache.is_empty(), "unsupported identity must not retain coverage");
-    }
-}
-
-#[test]
-fn checkpoint_fixture_corrupt_cursor_restarts_validation_without_claiming_completion() {
-    let (cache, identity) = bound_checkpoint();
-    for resume in ["other/static", "bucket/missing"] {
-        let mut next = cache.clone();
-        next.info.scan_resume_after = Some(resume.into());
-        next.info.scan_checkpoint = Some(DataUsageScanCheckpoint::new(resume.into(), DataUsageScanCheckpointReason::Objects));
-        let plan = DataUsageScanPlanDigest([43; 32]);
-        next.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, plan, identity);
-        assert_eq!(retained(&next), 3, "observations may survive an invalid cursor");
-        assert!(next.info.scan_resume_after.is_none());
-        assert!(next.info.scan_checkpoint.is_none());
-        assert_eq!(next.info.scan_progress.expect("new verification sweep").started_plan, plan);
-        assert!(!next.info.snapshot_complete);
-        assert!(next.info.scan_plan_digest.is_none());
-    }
-}
-
-#[test]
-fn checkpoint_fixture_receipt_binds_covered_prefix_not_unvisited_suffix() {
-    let (mut cache, _) = bound_checkpoint();
-    cache.replace(
-        "bucket/z-unvisited",
-        "bucket",
-        DataUsageEntry {
-            objects: 99,
-            ..Default::default()
-        },
-    );
-    assert_eq!(cache.validated_scan_frontier(), Some("bucket/static"));
-    let saved = decode_fixture(&cache.marshal_msg().expect("persist receipt")).expect("load receipt");
-    assert_eq!(saved.validated_scan_frontier(), Some("bucket/static"));
-    cache.cache.get_mut("bucket/static").expect("covered prefix").objects = 100;
-    assert!(
-        cache.validated_scan_frontier().is_none(),
-        "altered covered content must invalidate the receipt"
-    );
-}
-
-#[tokio::test]
-#[serial]
-async fn checkpoint_fixture_existing_uncovered_cursor_cannot_skip_to_complete() {
-    let (scanner, root) = build_test_scanner().await;
-    let _guard = TestGuard {
-        temp_dir: Some(root.clone()),
-    };
-    for (object, size) in [
-        ("a-done/object", 1),
-        ("b-pending/first", 7),
-        ("b-pending/second", 1),
-        ("z-stale/object", 2),
-    ] {
-        write_checkpoint_object(&root, object, &[(None, size)]).await;
-    }
-    let identity = crate::DataUsageScanIdentity {
-        tier_registry_generation: crate::runtime_tier_registry_for_cycle(11, 7).await.generation,
-        ..bound_checkpoint().1
-    };
-    for tamper_receipt_path in [false, true] {
-        let store = FixtureStore::new();
-        let mut cache = DataUsageCache::default();
-        let revisions = cache
-            .load_with_revisions(store.clone(), CACHE_NAME)
-            .await
-            .expect("empty fixture revisions");
-        cache.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, identity);
-        cache.replace("bucket", "", DataUsageEntry::default());
-        for (prefix, objects) in [("a-done", 1), ("b-pending", 99), ("z-stale", 99)] {
-            cache.replace(
-                &format!("bucket/{prefix}"),
-                "bucket",
-                DataUsageEntry {
-                    objects,
-                    size: objects,
-                    compacted: true,
-                    ..Default::default()
-                },
-            );
-        }
-        cache
-            .seal_scan_frontier(Some("bucket/a-done"))
-            .expect("actual completed prefix receipt");
-        cache.info.scan_resume_after = Some("bucket/z-stale".into());
-        cache.info.scan_checkpoint = Some(DataUsageScanCheckpoint::new(
-            "bucket/z-stale".into(),
-            DataUsageScanCheckpointReason::Objects,
-        ));
-        if tamper_receipt_path {
-            cache.info.scan_coverage_receipt.as_mut().expect("receipt").through = "bucket/z-stale".into();
-        }
-        cache
-            .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0)
-            .await
-            .expect("persist corrupted existing cursor");
-        let mut loaded = store.strict_load().await;
-        let revisions = loaded
-            .load_with_revisions(store.clone(), CACHE_NAME)
-            .await
-            .expect("corrupt cursor CAS revision");
-        assert!(loaded.validated_scan_frontier().is_none());
-        loaded.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, identity);
-        assert!(loaded.info.scan_resume_after.is_none());
-        loaded.info.skip_healing = true;
-        let parent = CancellationToken::new();
-        let budget = ScannerCycleBudget::new_with_progress_tracking(
-            &parent,
-            ScannerCycleBudgetConfig {
-                max_objects: Some(2),
-                ..Default::default()
-            },
-        );
-        let outcome = scanner
-            .local_disk
-            .clone()
-            .nsscanner_disk(
-                budget.token(),
-                budget.clone(),
-                vec![scanner.local_disk.clone()],
-                loaded,
-                None,
-                HealScanMode::Normal,
-            )
-            .await
-            .expect("scan must revisit the prefix");
-        let ScannerDiskScanOutcome::Partial(cache) = outcome else {
-            panic!("uncovered suffix must not become complete")
-        };
-        assert_eq!(budget.progress().0, 2);
-        assert_eq!(cache.checked_flatten("bucket/b-pending").expect("revisited prefix").size, 7);
-        assert!(!cache.info.snapshot_complete);
-        cache
-            .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0)
-            .await
-            .expect("persist verified partial");
-        assert!(!store.strict_load().await.info.snapshot_complete);
-    }
-}
-
-#[tokio::test]
-#[serial]
-async fn checkpoint_fixture_failed_child_prevents_receipt_advancing_past_gap() {
-    let (scanner, root) = build_test_scanner().await;
-    let _guard = TestGuard {
-        temp_dir: Some(root.clone()),
-    };
-    for object in ["a-good", "b-skipped", "c-later"] {
-        write_checkpoint_object(&root, object, &[(None, 1)]).await;
-    }
-    let identity = crate::DataUsageScanIdentity {
-        tier_registry_generation: crate::runtime_tier_registry_for_cycle(11, 7).await.generation,
-        ..bound_checkpoint().1
-    };
-    let mut cache = DataUsageCache::default();
-    cache.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, identity);
-    cache.info.skip_healing = true;
-    cache.info.failed_objects.insert(
-        root.join("bucket/b-skipped/xl.meta").to_string_lossy().into_owned(),
-        FolderScanner::now_secs(),
-    );
-    let parent = CancellationToken::new();
-    let budget = ScannerCycleBudget::new_with_progress_tracking(
-        &parent,
-        ScannerCycleBudgetConfig {
-            max_objects: Some(2),
-            ..Default::default()
-        },
-    );
-    let result = scanner
-        .local_disk
-        .clone()
-        .nsscanner_disk(
-            budget.token(),
-            budget,
-            vec![scanner.local_disk.clone()],
-            cache,
-            None,
-            HealScanMode::Normal,
-        )
-        .await
-        .expect("scan with a known failed child");
-    let ScannerDiskScanOutcome::Partial(cache) = result else {
-        panic!("skipped failure is not complete")
-    };
-    assert_eq!(cache.validated_scan_frontier(), Some("bucket/a-good"));
-    assert!(!cache.info.failed_objects.is_empty());
-    assert!(!cache.info.snapshot_complete);
-}
-
-#[tokio::test]
-#[serial]
-async fn checkpoint_fixture_complete_sampling_partial_resumes_with_fixed_budget() {
-    check_complete_sampling_resumption(HealScanMode::Normal).await;
-}
-
-#[tokio::test]
-#[serial]
-async fn checkpoint_fixture_normal_partial_reenters_prefix_for_deep_scan() {
-    check_complete_sampling_resumption(HealScanMode::Deep).await;
-}
-
-async fn check_complete_sampling_resumption(resume_mode: HealScanMode) {
-    let (scanner, root) = build_test_scanner().await;
-    let _guard = TestGuard {
-        temp_dir: Some(root.clone()),
-    };
-    for index in 0..9 {
-        write_checkpoint_object(&root, &format!("prefix/{index:04}"), &[(None, 1)]).await;
-    }
-    let identity = crate::DataUsageScanIdentity {
-        tier_registry_generation: crate::runtime_tier_registry_for_cycle(11, 7).await.generation,
-        ..bound_checkpoint().1
-    };
-    let mut cache = DataUsageCache::default();
-    cache.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, identity);
-    cache.info.skip_healing = true;
-    let parent = CancellationToken::new();
-    let budget = ScannerCycleBudget::new(&parent, Default::default());
-    // Seed an existing complete baseline; every recovery attempt below is bounded.
-    let baseline = scanner
-        .local_disk
-        .clone()
-        .nsscanner_disk(
-            budget.token(),
-            budget,
-            vec![scanner.local_disk.clone()],
-            cache,
-            None,
-            HealScanMode::Normal,
-        )
-        .await
-        .expect("initial complete baseline");
-    let ScannerDiskScanOutcome::Complete(mut cache) = baseline else { panic!("baseline is complete") };
-    let mut deep_current = cache.clone();
-    let deep_identity = crate::DataUsageScanIdentity {
-        scan_mode: HealScanMode::Deep,
-        ..identity
-    };
-    let state = crate::scanner_io::current_cache_root_or_prepare_with_generation(
-        &mut deep_current,
-        "bucket",
-        SOURCE,
-        11,
-        7,
-        PLAN,
-        crate::scanner_io::DataUsageCacheReuseOptions {
-            checkpoint_identity: Some(deep_identity),
-            ..Default::default()
-        },
-    );
-    assert!(
-        matches!(state, crate::scanner_io::DataUsageCacheScanState::Prepared { .. }),
-        "Normal complete is not Deep Current"
-    );
-    cache.prepare_bucket_checkpoint("bucket", 12, 7, SOURCE, PLAN, identity);
-    assert!(cache.info.scan_progress.is_none(), "complete unchanged baseline uses existing sampling");
-    let parent = CancellationToken::new();
-    let budget = ScannerCycleBudget::new_with_progress_tracking(
-        &parent,
-        ScannerCycleBudgetConfig {
-            max_objects: Some(3),
-            ..Default::default()
-        },
-    );
-    let outcome = scanner
-        .local_disk
-        .clone()
-        .nsscanner_disk(
-            budget.token(),
-            budget,
-            vec![scanner.local_disk.clone()],
-            cache,
-            None,
-            HealScanMode::Normal,
-        )
-        .await
-        .expect("sampling interruption");
-    let ScannerDiskScanOutcome::Partial(cache) = outcome else {
-        panic!("sampling must exhaust the three-object budget")
-    };
-    assert!(cache.info.scan_progress.is_none());
-    assert_eq!(cache.info.scan_plan_digest, Some(PLAN));
-    assert!(cache.info.scan_checkpoint.is_some());
-    let store = FixtureStore::new();
-    let mut loaded = DataUsageCache::default();
-    let revisions = loaded
-        .load_with_revisions(store.clone(), CACHE_NAME)
-        .await
-        .expect("fixture revision");
-    cache
-        .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0)
-        .await
-        .expect("persist sampling partial");
-    if resume_mode == HealScanMode::Deep {
-        write_checkpoint_object(&root, "prefix/0000", &[(None, 7)]).await;
-    }
-    let resumed_identity = crate::DataUsageScanIdentity {
-        scan_mode: resume_mode,
-        ..identity
-    };
-    for round in 0..16 {
-        let mut loaded = DataUsageCache::default();
-        let revisions = loaded
-            .load_with_revisions(store.clone(), CACHE_NAME)
-            .await
-            .expect("reload partial each recovery round");
-        loaded.prepare_bucket_checkpoint("bucket", 12, 7, SOURCE, PLAN, resumed_identity);
-        assert!(loaded.info.scan_progress.is_some(), "sampling partial must enter forward validation");
-        loaded.info.skip_healing = true;
-        let parent = CancellationToken::new();
-        let budget = ScannerCycleBudget::new_with_progress_tracking(
-            &parent,
-            ScannerCycleBudgetConfig {
-                max_objects: Some(3),
-                ..Default::default()
-            },
-        );
-        let result = scanner
-            .local_disk
-            .clone()
-            .nsscanner_disk(budget.token(), budget, vec![scanner.local_disk.clone()], loaded, None, resume_mode)
-            .await
-            .expect("bounded recovery scan");
-        let (cache, complete) = match result {
-            ScannerDiskScanOutcome::Partial(cache) => (cache, false),
-            ScannerDiskScanOutcome::Complete(cache) => (cache, true),
-            _ => panic!("fixture namespace remains present"),
-        };
-        if round == 0 && resume_mode == HealScanMode::Deep {
-            assert_eq!(cache.find("bucket/prefix/0000").expect("Deep revisits earlier prefix").size, 7);
-        }
-        cache
-            .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0)
-            .await
-            .expect("save bounded recovery");
-        if complete {
-            let root = store.strict_load().await.checked_flatten("bucket").expect("certified root");
-            assert_eq!(root.objects, 9);
-            assert_eq!(root.size, if resume_mode == HealScanMode::Deep { 15 } else { 9 });
-            return;
-        }
-    }
-    panic!("sampling interruption must recover with the unchanged three-object budget");
-}
-
 #[tokio::test]
 #[serial]
 async fn checkpoint_fixture_save_reload_resume() {
@@ -767,48 +242,23 @@ async fn checkpoint_fixture_save_reload_resume() {

 #[tokio::test]
 #[serial]
-async fn checkpoint_fixture_hot_digest_retains_partial_progress() {
+async fn checkpoint_fixture_hot_digest_diagnostic() {
    run_checkpoint_fixture(true).await;
 }

-async fn write_checkpoint_object(root: &std::path::Path, object: &str, versions: &[(Option<Uuid>, i64)]) {
-    let mut metadata = FileMeta::new();
-    for (index, (version_id, size)) in versions.iter().enumerate() {
-        let mut info = FileInfo::new(object, 4, 2);
-        info.volume = "bucket".into();
-        info.version_id = *version_id;
-        info.versioned = version_id.is_some();
-        info.size = *size;
-        info.mod_time = Some(
-            OffsetDateTime::from_unix_timestamp(1_700_000_000 + i64::try_from(index).expect("fixture index"))
-                .expect("non-sentinel fixture modification time"),
-        );
-        metadata.add_version(info).expect("fixture version");
-    }
-    write_test_object_metadata_bytes(root, "bucket", object, &metadata.marshal_msg().expect("fixture metadata")).await;
-}
-
 async fn run_checkpoint_fixture(change_digest: bool) {
    let (scanner, root) = build_test_scanner().await;
    let _guard = TestGuard {
        temp_dir: Some(root.clone()),
    };
    for index in 0..STATIC_OBJECTS {
-        write_checkpoint_object(&root, &format!("static/{index:04}"), &[(None, 1)]).await;
+        write_test_object_metadata(&root, "bucket", &format!("static/{index:04}")).await;
    }
-    let identity = crate::DataUsageScanIdentity {
-        version: 1,
-        bucket_incarnation: Uuid::from_u128(1),
-        set_layout: DataUsageScanPlanDigest([41; 32]),
-        publication_epoch: 0,
-        tier_registry_generation: crate::runtime_tier_registry_for_cycle(11, 7).await.generation,
-        scan_mode: HealScanMode::Normal,
-    };
    let store = FixtureStore::new();
    let mut previous = 0;
    let mut visited = 0;
    for round in 0..3_u8 {
-        write_checkpoint_object(&root, "hot/current", &[(None, 1)]).await;
+        write_test_object_metadata(&root, "bucket", "hot/current").await;
        let mut cache = DataUsageCache::default();
        let revisions = cache
            .load_with_revisions(store.clone(), CACHE_NAME)
@@ -828,11 +278,9 @@ async fn run_checkpoint_fixture(change_digest: bool) {
            crate::scanner_io::DataUsageCacheReuseOptions {
                require_source: true,
                tier_registry_generation: None,
-                checkpoint_identity: Some(identity),
            },
        );
        let prepared = retained(&cache);
-        cache.info.skip_healing = true;
        let parent = CancellationToken::new();
        let budget = ScannerCycleBudget::new_with_progress_tracking(
            &parent,
@@ -878,12 +326,13 @@ async fn run_checkpoint_fixture(change_digest: bool) {
        eprintln!(
            "checkpoint_fixture round={round} hot_digest={change_digest} visited_total={visited} before={previous} prepared={prepared} scanned={scanned} reloaded={reloaded} diagnosis={diagnosis:?}"
        );
-        assert_eq!(
-            diagnosis,
-            CoverageDiagnosis::Progress,
-            "visited growth must produce durable static coverage"
-        );
-        assert!(loaded.info.scan_plan_digest.is_none(), "old readers must rebuild an uncertified sweep");
+        if !change_digest || std::env::var_os("RUSTFS_CHECKPOINT_REQUIRE_PROGRESS").is_some() {
+            assert_eq!(
+                diagnosis,
+                CoverageDiagnosis::Progress,
+                "visited growth must produce durable static coverage"
+            );
+        }
        crate::remote_scanner::checkpoint_fixture_partial_return(budget.progress(), budget.entries_visited()).await;
        previous = reloaded;
    }
@@ -934,85 +383,28 @@ async fn run_checkpoint_fixture(change_digest: bool) {
    assert!(result.is_err(), "pre-scan cancellation must not produce a complete root");
    assert_eq!(budget.reason(), None, "parent cancellation is not object budget exhaustion");

-    store.reject_save.store(false, Ordering::SeqCst);
-    write_checkpoint_object(&root, "static/0000", &[(Some(Uuid::from_u128(2)), 7), (Some(Uuid::from_u128(3)), 3)]).await;
-    tokio::fs::remove_dir_all(root.join("bucket/static/0001"))
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(&parent, Default::default());
+    let result = scanner
+        .local_disk
+        .clone()
+        .nsscanner_disk(
+            budget.token(),
+            budget,
+            vec![scanner.local_disk.clone()],
+            loaded,
+            None,
+            HealScanMode::Normal,
+        )
        .await
-        .expect("remove previously scanned fixture object");
-    write_checkpoint_object(&root, "hot/later", &[(None, 1)]).await;
-    let final_plan = crate::scanner_io::checkpoint_fixture_bucket_digest(PLAN, Some(3));
-    let mut saw_mixed_sweep_end = false;
-    for _ in 0..32 {
-        let mut cache = DataUsageCache::default();
-        let revisions = cache
-            .load_with_revisions(store.clone(), CACHE_NAME)
-            .await
-            .expect("reload every bounded round");
-        crate::scanner_io::current_cache_root_or_prepare_with_generation(
-            &mut cache,
-            "bucket",
-            SOURCE,
-            11,
-            7,
-            final_plan,
-            crate::scanner_io::DataUsageCacheReuseOptions {
-                require_source: true,
-                tier_registry_generation: Some(identity.tier_registry_generation),
-                checkpoint_identity: Some(identity),
-            },
-        );
-        cache.info.skip_healing = true;
-        let parent = CancellationToken::new();
-        let budget = ScannerCycleBudget::new_with_progress_tracking(
-            &parent,
-            ScannerCycleBudgetConfig {
-                max_objects: Some(4),
-                ..Default::default()
-            },
-        );
-        let outcome = scanner
-            .local_disk
-            .clone()
-            .nsscanner_disk(
-                budget.token(),
-                budget.clone(),
-                vec![scanner.local_disk.clone()],
-                cache,
-                None,
-                HealScanMode::Normal,
-            )
-            .await
-            .expect("bounded sweep outcome");
-        let (cache, complete) = match outcome {
-            ScannerDiskScanOutcome::Complete(cache) => (cache, true),
-            ScannerDiskScanOutcome::Partial(cache) => (cache, false),
-            ScannerDiskScanOutcome::NamespaceNotFound(_) => panic!("fixture namespace exists"),
-        };
-        cache
-            .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0)
-            .await
-            .expect("save each bounded sweep");
-        let saved = store.strict_load().await;
-        if complete {
-            assert!(
-                saw_mixed_sweep_end,
-                "a clean tail must first finish as partial before a new validation sweep"
-            );
-            assert!(saved.info.snapshot_complete);
-            assert!(saved.info.scan_progress.is_none());
-            assert!(saved.info.scan_checkpoint.is_none());
-            assert_eq!(saved.info.scan_plan_digest, Some(final_plan));
-            let total = saved.checked_flatten("bucket").expect("complete bucket root");
-            assert_eq!((total.objects, total.versions, total.size), (25, 2, 34));
-            assert_eq!(saved.checked_flatten("bucket/static").expect("static subtree").objects, 23);
-            assert_eq!(saved.checked_flatten("bucket/hot").expect("hot subtree").objects, 2);
-            return;
-        }
-        assert!(!saved.info.snapshot_complete);
-        assert!(saved.info.scan_plan_digest.is_none());
-        if !budget.budget_elapsed() {
-            saw_mixed_sweep_end = true;
-        }
-    }
-    panic!("finite stable fixture must converge using the same four-object budget without an unbounded final sweep");
+        .expect("unbounded scan must complete after durable partial progress");
+    let ScannerDiskScanOutcome::Complete(cache) = result else {
+        panic!("unbounded fixture must produce a complete disk cache");
+    };
+    assert!(cache.info.snapshot_complete);
+    assert!(cache.info.scan_checkpoint.is_none());
+    assert_eq!(
+        cache.checked_flatten("bucket").expect("complete bucket root").objects,
+        usize::try_from(STATIC_OBJECTS + 1).expect("fixture object count fits usize")
+    );
 }
@@ -183,18 +183,6 @@ fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<
        return None;
    }

-    // Completed maintenance also covers ordinary usage. Keep its exact stored
-    // proof for cache reuse, and reject mixtures of different set work proofs.
-    let baseline_plan_digest = DataUsageScanPlanDigest(baseline.usage_snapshot_set_states.first()?.scan_plan_digest?);
-    if ![
-        proof.scan_plan_digest,
-        scanner_bucket_work_digest(proof.scan_plan_digest, HealScanMode::Normal, true),
-        scanner_bucket_work_digest(proof.scan_plan_digest, HealScanMode::Deep, true),
-    ]
-    .contains(&baseline_plan_digest)
-    {
-        return None;
-    }
    let mut states = HashSet::with_capacity(baseline.usage_snapshot_set_states.len());
    for state in &baseline.usage_snapshot_set_states {
        let source = DataUsageCacheSource::new(usize::try_from(state.pool_index).ok()?, usize::try_from(state.set_index).ok()?);
@@ -204,13 +192,13 @@ fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<
            || state.tombstone
            || state.scanner_epoch != Some(proof.leader_epoch)
            || state.scanner_cycle.is_none_or(|cycle| cycle > proof.want_cycle)
-            || state.scan_plan_digest != Some(baseline_plan_digest.0)
+            || state.scan_plan_digest != Some(proof.scan_plan_digest.0)
        {
            return None;
        }
    }

-    (states == *proof.expected_sources).then_some(baseline_plan_digest)
+    (states == *proof.expected_sources).then_some(proof.scan_plan_digest)
 }

 fn scoped_scan_scope_from_dirty_buckets(
@@ -283,9 +271,6 @@ pub struct ScannerBucketScanPlan {
    all_buckets: Arc<Vec<BucketInfo>>,
    scope: ScannerBucketScanScope,
    digest: DataUsageScanPlanDigest,
-    /// Includes mutation generations even when the set planner uses a structural digest.
-    bucket_coverage_digest: DataUsageScanPlanDigest,
-    requires_full_scan: bool,
    // Cache work must invalidate on namespace completion even when its scoped baseline remains reusable.
    execution_digest: DataUsageScanPlanDigest,
    leader_epoch: u64,
@@ -329,53 +314,6 @@ fn scanner_bucket_plan_digest(buckets: &[BucketInfo], activity_digest: [u8; 32])
    DataUsageScanPlanDigest(hasher.finalize().into())
 }

-fn scanner_bucket_inventory_is_complete(
-    all_buckets: &[BucketInfo],
-    buckets_by_source: &HashMap<DataUsageCacheSource, Vec<BucketInfo>>,
-) -> bool {
-    let inventory = all_buckets
-        .iter()
-        .map(|bucket| (bucket.name.as_str(), bucket.created))
-        .collect::<HashMap<_, _>>();
-    if inventory.len() != all_buckets.len() || inventory.keys().any(|name| name.is_empty() || *name == DATA_USAGE_ROOT) {
-        return false;
-    }
-    let mut covered = HashSet::with_capacity(inventory.len());
-    for buckets in buckets_by_source.values() {
-        let mut set_names = HashSet::with_capacity(buckets.len());
-        for bucket in buckets {
-            if !set_names.insert(bucket.name.as_str()) || inventory.get(bucket.name.as_str()) != Some(&bucket.created) {
-                return false;
-            }
-            covered.insert(bucket.name.as_str());
-        }
-    }
-    covered.len() == inventory.len()
-}
-
-// Bind known work requirements before both local and remote cache admission.
-// Matching requirements remain reusable for the same intent; this is not a
-// new deadline or a durable generation for newly due maintenance.
-fn scanner_bucket_work_digest(
-    scan_plan_digest: DataUsageScanPlanDigest,
-    scan_mode: HealScanMode,
-    requires_full_scan: bool,
-) -> DataUsageScanPlanDigest {
-    if scan_mode == HealScanMode::Normal && !requires_full_scan {
-        return scan_plan_digest;
-    }
-    let mut hasher = Sha256::new();
-    hasher.update(b"scanner-bucket-work-v1");
-    hasher.update(scan_plan_digest.0);
-    hasher.update([match scan_mode {
-        HealScanMode::Unknown => 0,
-        HealScanMode::Normal => 1,
-        HealScanMode::Deep => 2,
-    }]);
-    hasher.update([u8::from(requires_full_scan || scan_mode == HealScanMode::Deep)]);
-    DataUsageScanPlanDigest(hasher.finalize().into())
-}
-
 fn scanner_bucket_cache_digest(
    scan_plan_digest: DataUsageScanPlanDigest,
    dirty_generation: Option<u64>,
@@ -773,39 +711,6 @@ pub(crate) async fn scanner_set_disk_inventory(set: &SetDisks) -> Vec<Arc<Disk>>
    disks
 }

-pub(crate) async fn scanner_bucket_checkpoint_identity(
-    set: &SetDisks,
-    bucket: &str,
-    publication_epoch: u64,
-    tier_registry_generation: u64,
-    scan_mode: HealScanMode,
-) -> Result<crate::DataUsageScanIdentity> {
-    let bucket_incarnation = set.bucket_incarnation_id_from_disk(bucket).await?;
-    let disks = set
-        .format
-        .erasure
-        .sets
-        .get(set.set_index)
-        .filter(|disks| !disks.is_empty())
-        .ok_or_else(|| Error::other("scanner checkpoint set layout is absent"))?;
-    if set.format.id.is_nil() || disks.iter().any(uuid::Uuid::is_nil) {
-        return Err(Error::other("scanner checkpoint set layout has a nil identity"));
-    }
-    let mut digest = Sha256::new();
-    digest.update(set.format.id.as_bytes());
-    for disk in disks {
-        digest.update(disk.as_bytes());
-    }
-    Ok(crate::DataUsageScanIdentity {
-        version: 1,
-        bucket_incarnation,
-        set_layout: crate::DataUsageScanPlanDigest(digest.finalize().into()),
-        publication_epoch,
-        tier_registry_generation,
-        scan_mode,
-    })
-}
-
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(crate) enum ScannerCycleDeferReason {
    ActivityBaselineUnavailable,
@@ -112,10 +112,6 @@ pub(crate) fn current_cache_root_entry_with_generation(
    let metadata_is_current = cache.info.name == name
        && cache.info.source == Some(source)
        && cache.info.snapshot_complete
-        && cache.info.scan_progress.is_none()
-        && cache.info.scan_checkpoint.is_none()
-        && cache.info.scan_resume_after.is_none()
-        && cache.info.scan_coverage_receipt.is_none()
        && cache.info.scan_plan_digest == Some(scan_plan_digest)
        && cache.info.last_update.is_some()
        && cache.info.next_cycle == next_cycle
@@ -141,7 +137,6 @@ pub(crate) enum DataUsageCacheScanState {
 pub(crate) struct DataUsageCacheReuseOptions {
    pub(crate) require_source: bool,
    pub(crate) tier_registry_generation: Option<u64>,
-    pub(crate) checkpoint_identity: Option<crate::DataUsageScanIdentity>,
 }

 #[cfg(test)]
@@ -164,7 +159,6 @@ pub(crate) fn current_cache_root_or_prepare(
        DataUsageCacheReuseOptions {
            require_source,
            tier_registry_generation: None,
-            checkpoint_identity: None,
        },
    )
 }
@@ -178,12 +172,6 @@ pub(crate) fn current_cache_root_or_prepare_with_generation(
    scan_plan_digest: DataUsageScanPlanDigest,
    options: DataUsageCacheReuseOptions,
 ) -> DataUsageCacheScanState {
-    if cache.info.next_cycle <= next_cycle
-        && cache.info.leader_epoch <= leader_epoch
-        && cache.info.scan_identity != options.checkpoint_identity
-    {
-        cache.info.scan_plan_digest = None;
-    }
    if options.tier_registry_generation.is_some_and(|generation| {
        cache.info.next_cycle <= next_cycle
            && cache.info.leader_epoch <= leader_epoch
@@ -205,12 +193,7 @@ pub(crate) fn current_cache_root_or_prepare_with_generation(
        Ok(Some(root)) => DataUsageCacheScanState::Current(Box::new(root)),
        current => DataUsageCacheScanState::Prepared {
            invalid_current: current.err(),
-            outcome: match options.checkpoint_identity.filter(crate::DataUsageScanIdentity::is_valid) {
-                Some(identity) => {
-                    cache.prepare_bucket_checkpoint(name, next_cycle, leader_epoch, source, scan_plan_digest, identity)
-                }
-                None => cache.prepare_for_scan(name, next_cycle, leader_epoch, source, scan_plan_digest, options.require_source),
-            },
+            outcome: cache.prepare_for_scan(name, next_cycle, leader_epoch, source, scan_plan_digest, options.require_source),
        },
    }
 }
@@ -230,87 +213,10 @@ pub(super) fn cache_snapshot_is_current(
    )
 }

-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub(super) struct ScannerSnapshotIdentity {
-    pub(super) cycle: u64,
-    pub(super) leader_epoch: u64,
-    pub(super) plan_digest: DataUsageScanPlanDigest,
-    pub(super) coverage_digest: DataUsageScanPlanDigest,
-    pub(super) tier_registry_generation: Option<u64>,
-}
-
-pub(super) struct ScannerSnapshotScope<'a> {
-    pub(super) sources: &'a HashSet<DataUsageCacheSource>,
-    pub(super) buckets: &'a [String],
-    pub(super) identity: ScannerSnapshotIdentity,
-}
-
-#[derive(Debug, PartialEq, Eq, thiserror::Error)]
-pub(super) enum ScannerSnapshotValidationError {
-    #[error("scanner snapshot does not cover the expected complete sets")]
-    IncompleteSets,
-    #[error("scanner snapshot does not match the requested generation")]
-    GenerationMismatch,
-    #[error("scanner snapshot bucket inventory is invalid")]
-    InvalidInventory,
-    #[error("scanner snapshot root is incomplete or corrupt")]
-    InvalidRoot,
-}
-
-struct ValidatedScannerSnapshot<'a> {
-    results: &'a [DataUsageCache],
-    last_update: SystemTime,
-}
-
-impl<'a> ValidatedScannerSnapshot<'a> {
-    fn validate(
-        results: &'a [DataUsageCache],
-        scope: &ScannerSnapshotScope<'_>,
-    ) -> std::result::Result<Self, ScannerSnapshotValidationError> {
-        if !scanner_results_form_complete_snapshot(results, scope.sources) {
-            return Err(ScannerSnapshotValidationError::IncompleteSets);
-        }
-        let bucket_keys = scope
-            .buckets
-            .iter()
-            .map(|bucket| crate::hash_path(bucket).key())
-            .collect::<HashSet<_>>();
-        if bucket_keys.len() != scope.buckets.len()
-            || scope
-                .buckets
-                .iter()
-                .any(|bucket| bucket.is_empty() || bucket == DATA_USAGE_ROOT)
-        {
-            return Err(ScannerSnapshotValidationError::InvalidInventory);
-        }
-        for result in results {
-            if result.info.next_cycle != scope.identity.cycle
-                || result.info.leader_epoch != scope.identity.leader_epoch
-                || result.info.scan_plan_digest != Some(scope.identity.plan_digest)
-                || result.info.scan_coverage_digest != Some(scope.identity.coverage_digest)
-                || result.info.tier_registry_generation != scope.identity.tier_registry_generation
-            {
-                return Err(ScannerSnapshotValidationError::GenerationMismatch);
-            }
-            if result.info.name != DATA_USAGE_ROOT
-                || result.info.cache_key_format != DATA_USAGE_CACHE_KEY_FORMAT
-                || !result.has_complete_root_inventory(&bucket_keys)
-            {
-                return Err(ScannerSnapshotValidationError::InvalidRoot);
-            }
-        }
-        let last_update = results
-            .iter()
-            .filter_map(|result| result.info.last_update)
-            .max()
-            .ok_or(ScannerSnapshotValidationError::IncompleteSets)?;
-        Ok(Self { results, last_update })
-    }
-}
-
 pub(super) fn completed_data_usage_info(
    results: &[DataUsageCache],
-    scope: &ScannerSnapshotScope<'_>,
+    expected_sources: &HashSet<DataUsageCacheSource>,
+    all_buckets: &[String],
    tier_registry_names: &[String],
    bucket_plan_complete: bool,
    budget_elapsed: bool,
@@ -323,10 +229,26 @@ pub(super) fn completed_data_usage_info(
    if !should_publish_completed_snapshot(completed_set_count, results.len(), budget_elapsed, cancelled) {
        return None;
    }
-    let validated = ValidatedScannerSnapshot::validate(results, scope).ok()?;
-    let results = validated.results;
-    let all_buckets = scope.buckets;
-    let registry_generation = scope.identity.tier_registry_generation;
+    if !scanner_results_form_complete_snapshot(results, expected_sources) {
+        return None;
+    }
+
+    // A generation is comparable across nodes because it is derived from the
+    // frozen registry names. Cycle and leader fencing remain separate cache
+    // metadata. Legacy peers omit the generation; an all-legacy result remains
+    // readable, but mixing legacy and new (or two new generations) would make
+    // the per-tier accounting ambiguous.
+    let registry_generation = results.first()?.info.tier_registry_generation;
+    if results.iter().any(|result| match registry_generation {
+        Some(generation) => result.info.tier_registry_generation != Some(generation),
+        None => result.info.tier_registry_generation.is_some(),
+    }) {
+        return None;
+    }
+
+    if results.iter().any(|result| result.root().is_none()) {
+        return None;
+    }

    let mut total = DataUsageEntry::default();
    let mut bucket_entries = HashMap::with_capacity(all_buckets.len());
@@ -351,7 +273,7 @@ pub(super) fn completed_data_usage_info(
        return None;
    }

-    let merged_last_update = validated.last_update;
+    let merged_last_update = results.iter().filter_map(|result| result.info.last_update).max()?;
    let buckets_usage = bucket_entries
        .iter()
        .map(|(bucket, entry)| Some((bucket.clone(), checked_bucket_usage_info(entry)?)))
@@ -378,8 +300,8 @@ pub(super) fn completed_data_usage_info(
    usage_snapshot_set_states.sort_by_key(|state| (state.pool_index, state.set_index));
    let data_usage_info = DataUsageInfo {
        last_update: Some(merged_last_update),
-        scanner_cycle: Some(scope.identity.cycle),
-        scanner_epoch: Some(scope.identity.leader_epoch),
+        scanner_cycle: Some(results.first()?.info.next_cycle),
+        scanner_epoch: Some(results.first()?.info.leader_epoch),
        objects_total_count: u64::try_from(total.objects).ok()?,
        versions_total_count: u64::try_from(total.versions).ok()?,
        delete_markers_total_count: u64::try_from(total.delete_markers).ok()?,
@@ -687,7 +609,6 @@ pub(super) async fn persist_and_publish_cache_snapshot(
    expected_publication_epoch: u64,
 ) -> Option<SystemTime> {
    let source = cache_snapshot.info.source?;
-    let coverage_digest = cache_snapshot.info.scan_coverage_digest?;
    let execution_digest = cache_snapshot.info.scan_execution_digest?;
    let guard = match acquire_scanner_cache_locks(store.as_ref(), DATA_USAGE_CACHE_NAME, source).await {
        Ok(guard) => guard,
@@ -753,8 +674,7 @@ pub(super) async fn persist_and_publish_cache_snapshot(
        );
        return None;
    }
-    if persisted.info.scan_coverage_digest == Some(coverage_digest)
-        && persisted.info.scan_execution_digest == Some(execution_digest)
+    if persisted.info.scan_execution_digest == Some(execution_digest)
        && matches!(
            current_cache_root_entry_with_generation(
                &persisted,
@@ -39,12 +39,6 @@ pub(super) fn prepare_scoped_set_scan(
    else {
        return None;
    };
-    // The existing cache does not bind each bucket to a durable incarnation.
-    // Listing creation times can come from volume metadata, so even Some(time)
-    // cannot prove that an unselected same-name bucket is the cached bucket.
-    if all_buckets.iter().any(|bucket| !selected_buckets.contains(&bucket.name)) {
-        return None;
-    }
    if selected_buckets.is_empty()
        || !old_cache.info.snapshot_complete
        || old_cache.info.last_update.is_none()
@@ -55,7 +49,7 @@ pub(super) fn prepare_scoped_set_scan(
        || old_cache.info.source != Some(generation.source)
        || old_cache.info.scan_plan_digest != Some(baseline_scan_plan_digest)
        || old_cache.info.cache_key_format != DATA_USAGE_CACHE_KEY_FORMAT
-        || !old_cache.has_complete_root_inventory(&old_cache.find(DATA_USAGE_ROOT)?.children)
+        || old_cache.checked_flatten_complete_scope(DATA_USAGE_ROOT).is_none()
    {
        return None;
    }
@@ -80,12 +74,21 @@ pub(super) fn prepare_scoped_set_scan(
        cache: HashMap::new(),
    };
    cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
+    let root_hash = crate::hash_path(DATA_USAGE_ROOT);
    let mut current_bucket_names = HashSet::with_capacity(all_buckets.len());
    for bucket in all_buckets {
        if !current_bucket_names.insert(bucket.name.as_str()) {
            return None;
        }
-        cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
+        if selected_buckets.contains(&bucket.name) {
+            cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
+            continue;
+        }
+
+        let bucket_hash = crate::hash_path(&bucket.name);
+        old_cache.find(&bucket.name)?;
+        cache.copy_with_children(old_cache, &bucket_hash, &Some(root_hash.clone()));
+        cache.find(&bucket.name)?;
    }

    Some(PreparedScopedSetScan {
@@ -115,8 +118,6 @@ impl ScannerIOCache for SetDisks {
            all_buckets,
            scope,
            digest: scan_plan_digest,
-            bucket_coverage_digest,
-            requires_full_scan,
            execution_digest,
            leader_epoch,
            tier_registry_generation,
@@ -126,8 +127,6 @@ impl ScannerIOCache for SetDisks {
            pending_maintenance_work,
            cache_cycle_floor,
        } = scan_plan;
-        let scan_plan_digest = scanner_bucket_work_digest(scan_plan_digest, scan_mode, requires_full_scan);
-        let bucket_work_digest = scanner_bucket_work_digest(bucket_coverage_digest, scan_mode, requires_full_scan);
        let pool_label = self.pool_index.to_string();
        let set_label = self.set_index.to_string();

@@ -170,9 +169,8 @@ impl ScannerIOCache for SetDisks {
                scan_plan_digest,
            },
        );
-        let mut scoped_cache = scoped_scan.map(|mut prepared| {
+        let mut scoped_cache = scoped_scan.map(|prepared| {
            buckets = prepared.buckets;
-            prepared.cache.info.scan_coverage_digest = Some(bucket_coverage_digest);
            prepared.cache
        });
        if buckets.is_empty() {
@@ -188,7 +186,6 @@ impl ScannerIOCache for SetDisks {
                            tier_registry_generation: Some(tier_registry_generation),
                            source: Some(source),
                            scan_plan_digest: Some(scan_plan_digest),
-                            scan_coverage_digest: Some(bucket_coverage_digest),
                            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
                            ..Default::default()
                        },
@@ -478,7 +475,6 @@ impl ScannerIOCache for SetDisks {
                    source: Some(source),
                    snapshot_complete: false,
                    scan_plan_digest: Some(scan_plan_digest),
-                    scan_coverage_digest: Some(bucket_coverage_digest),
                    cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
                    lkg_snapshot_complete: old_cache.info.lkg_snapshot_complete,
                    lkg_next_cycle: old_cache.info.lkg_next_cycle,
@@ -648,7 +644,7 @@ impl ScannerIOCache for SetDisks {

                    let cache_name = path_join_buf(&[&bucket.name, DATA_USAGE_CACHE_NAME]);
                    let bucket_scan_plan_digest =
-                        scanner_bucket_cache_digest(bucket_work_digest, dirty_usage_buckets_clone.get(&bucket.name).copied());
+                        scanner_bucket_cache_digest(execution_digest, dirty_usage_buckets_clone.get(&bucket.name).copied());

                    if let Some(server_epoch) = remote_server_epoch {
                        let request_sequence = remote_session_sequence;
@@ -894,17 +890,6 @@ impl ScannerIOCache for SetDisks {
                            continue;
                        }
                    };
-                    // Lack of an authoritative legacy identity disables the new
-                    // checkpoint protocol; the existing full rebuild remains available.
-                    let checkpoint_identity = scanner_bucket_checkpoint_identity(
-                        &store_clone_clone,
-                        &bucket.name,
-                        expected_publication_epoch_clone,
-                        tier_registry_generation,
-                        scan_mode,
-                    )
-                    .await
-                    .ok();
                    let scan_state = current_cache_root_or_prepare_with_generation(
                        &mut cache,
                        &bucket.name,
@@ -915,7 +900,6 @@ impl ScannerIOCache for SetDisks {
                        DataUsageCacheReuseOptions {
                            require_source: require_cache_source,
                            tier_registry_generation: Some(tier_registry_generation),
-                            checkpoint_identity,
                        },
                    );
                    let outcome = match scan_state {
@@ -1065,21 +1049,6 @@ impl ScannerIOCache for SetDisks {
                            }
                        }
                    };
-                    if let Some(expected) = checkpoint_identity
-                        && scanner_bucket_checkpoint_identity(
-                            &store_clone_clone,
-                            &bucket.name,
-                            expected_publication_epoch_clone,
-                            tier_registry_generation,
-                            scan_mode,
-                        )
-                        .await
-                        .ok()
-                            != Some(expected)
-                    {
-                        record_failed_dirty_bucket(&failed_dirty_buckets_clone, &bucket.name).await;
-                        continue;
-                    }
                    let scan_outcome = match scan_result {
                        Ok(scan_outcome) => scan_outcome,
                        Err(e) => {
@@ -72,9 +72,6 @@ where
        scan_mode,
        scan_scope: ScannerBucketScanScope::default(),
        persisted_usage_baseline: None,
-        requires_full_scan: true,
-        #[cfg(test)]
-        resolved_scope_observer: None,
    };
    nsscanner_with_storage_status_scoped(store, request).await
 }
@@ -88,10 +85,6 @@ pub(crate) struct ScannerCycleRequest {
    pub(crate) scan_mode: HealScanMode,
    pub(crate) scan_scope: ScannerBucketScanScope,
    pub(crate) persisted_usage_baseline: Option<Bytes>,
-    /// Scheduled maintenance must visit clean buckets even with a valid dirty scope.
-    pub(crate) requires_full_scan: bool,
-    #[cfg(test)]
-    pub(crate) resolved_scope_observer: Option<tokio::sync::oneshot::Sender<ScannerBucketScanScope>>,
 }

 struct ScannerBucketScopeResolution<'a> {
@@ -100,7 +93,6 @@ struct ScannerBucketScopeResolution<'a> {
    activity_before: &'a crate::scanner::ScannerActivitySnapshot,
    dirty_usage_snapshot: &'a DirtyUsageSnapshot,
    all_buckets: &'a [BucketInfo],
-    requires_full_scan: bool,
 }

 async fn resolve_scanner_bucket_scan_scope<S>(
@@ -111,9 +103,6 @@ async fn resolve_scanner_bucket_scan_scope<S>(
 where
    S: ScannerStorage,
 {
-    if resolution.requires_full_scan {
-        return ScannerBucketScanScope::default();
-    }
    if !resolution.requested_scope.is_default()
        || !resolution.dirty_usage_snapshot.covers_all_pending
        || resolution.dirty_usage_snapshot.generation == u64::MAX
@@ -183,9 +172,6 @@ where
        scan_mode,
        scan_scope,
        persisted_usage_baseline,
-        requires_full_scan,
-        #[cfg(test)]
-        resolved_scope_observer,
    } = request;
    let child_token = ctx.child_token();
    let _tier_cycle_guard = begin_tier_registry_cycle(want_cycle, leader_epoch);
@@ -274,13 +260,13 @@ where
        }
    }
    bucket_plan_complete &= buckets_by_source.keys().copied().collect::<HashSet<_>>() == *expected_sources;
-    bucket_plan_complete &= scanner_bucket_inventory_is_complete(&all_buckets, &buckets_by_source);
-    let structural_scan_plan_digest =
-        scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_structural_digest(&activity_before));
-    let scan_plan_digest = scanner_bucket_work_digest(structural_scan_plan_digest, scan_mode, requires_full_scan);
    let activity_digest = crate::scanner::scanner_activity_snapshot_digest(&activity_before);
-    let bucket_coverage_digest = scanner_bucket_plan_digest(&all_buckets, activity_digest);
-    let execution_digest = scanner_bucket_work_digest(bucket_coverage_digest, scan_mode, requires_full_scan);
+    let scan_plan_digest =
+        scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_structural_digest(&activity_before));
+    let mut execution_hasher = Sha256::new();
+    execution_hasher.update(scan_plan_digest.0);
+    execution_hasher.update(activity_digest);
+    let execution_digest = DataUsageScanPlanDigest(execution_hasher.finalize().into());
    let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets(&all_buckets, dirty_generation_before_bucket_list));
    let scan_scope = resolve_scanner_bucket_scan_scope(
        store,
@@ -292,19 +278,14 @@ where
                expected_sources: &expected_sources,
                leader_epoch,
                want_cycle,
-                scan_plan_digest: structural_scan_plan_digest,
+                scan_plan_digest,
            },
            activity_before: &activity_before,
            dirty_usage_snapshot: &dirty_usage_snapshot,
            all_buckets: &all_buckets,
-            requires_full_scan: requires_full_scan || scan_mode == HealScanMode::Deep,
        },
    )
    .await;
-    #[cfg(test)]
-    if let Some(observer) = resolved_scope_observer {
-        let _ = observer.send(scan_scope.clone());
-    }
    let cache_cycle_floor = Arc::new(AtomicU64::new(want_cycle));
    let tier_registry = runtime_tier_registry_for_cycle(want_cycle, leader_epoch).await;
    let tier_registry_generation = tier_registry.generation;
@@ -434,9 +415,7 @@ where
            buckets: set_buckets,
            all_buckets: Arc::clone(&all_buckets),
            scope: scan_scope.clone(),
-            digest: structural_scan_plan_digest,
-            bucket_coverage_digest,
-            requires_full_scan,
+            digest: scan_plan_digest,
            execution_digest,
            leader_epoch,
            tier_registry_generation,
@@ -564,17 +543,8 @@ where
    let all_bucket_names = all_buckets.iter().map(|bucket| bucket.name.clone()).collect::<Vec<_>>();
    let completed_usage = completed_data_usage_info(
        &results,
-        &ScannerSnapshotScope {
-            sources: &expected_sources,
-            buckets: &all_bucket_names,
-            identity: ScannerSnapshotIdentity {
-                cycle: want_cycle,
-                leader_epoch,
-                plan_digest: scan_plan_digest,
-                coverage_digest: bucket_coverage_digest,
-                tier_registry_generation: Some(tier_registry_generation),
-            },
-        },
+        &expected_sources,
+        &all_bucket_names,
        &tier_registry.names,
        bucket_plan_complete,
        budget_elapsed,
@@ -17,33 +17,6 @@ use crate::data_usage_define::{UNKNOWN_TIER, UnknownTierStats, hash_path};
 use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage, TierAccountingProof};

 const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([7; 32]);
-const TEST_COVERAGE_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([6; 32]);
-
-#[test]
-fn scanner_bucket_inventory_requires_exact_unique_set_union() {
-    let first = BucketInfo {
-        name: "first".to_string(),
-        ..Default::default()
-    };
-    let second = BucketInfo {
-        name: "second".to_string(),
-        ..Default::default()
-    };
-    let source = DataUsageCacheSource::new(0, 0);
-    let mut sets = HashMap::from([(source, vec![first.clone()])]);
-    assert!(scanner_bucket_inventory_is_complete(std::slice::from_ref(&first), &sets));
-    assert!(!scanner_bucket_inventory_is_complete(&[first.clone(), second.clone()], &sets));
-    assert!(!scanner_bucket_inventory_is_complete(&[], &sets));
-    assert!(!scanner_bucket_inventory_is_complete(&[first.clone(), first.clone()], &sets));
-    sets.insert(source, vec![first.clone(), first.clone()]);
-    assert!(!scanner_bucket_inventory_is_complete(std::slice::from_ref(&first), &sets));
-    sets.insert(source, vec![second]);
-    assert!(!scanner_bucket_inventory_is_complete(std::slice::from_ref(&first), &sets));
-    let mut recreated = first.clone();
-    recreated.created = Some(OffsetDateTime::UNIX_EPOCH);
-    sets.insert(source, vec![recreated]);
-    assert!(!scanner_bucket_inventory_is_complete(&[first], &sets));
-}

 #[test]
 fn should_publish_completed_snapshot_requires_full_clean_cycle() {
@@ -106,7 +79,6 @@ fn completed_root_cache(bucket: &str, objects: usize, update_secs: u64, source:
            source: Some(source),
            snapshot_complete: true,
            scan_plan_digest: Some(TEST_PLAN_DIGEST),
-            scan_coverage_digest: Some(TEST_COVERAGE_DIGEST),
            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
            ..Default::default()
        },
@@ -136,139 +108,7 @@ fn completed_data_usage_info_for_test(
    cancelled: bool,
 ) -> Option<(DataUsageInfo, SystemTime)> {
    let expected_sources = results.iter().filter_map(|result| result.info.source).collect::<HashSet<_>>();
-    completed_usage_for_scope(results, &expected_sources, all_buckets, &[], true, budget_elapsed, cancelled)
-}
-
-fn completed_usage_for_scope(
-    results: &[DataUsageCache],
-    expected_sources: &HashSet<DataUsageCacheSource>,
-    all_buckets: &[String],
-    tier_registry_names: &[String],
-    bucket_plan_complete: bool,
-    budget_elapsed: bool,
-    cancelled: bool,
-) -> Option<(DataUsageInfo, SystemTime)> {
-    let first = results.first()?;
-    completed_data_usage_info(
-        results,
-        &ScannerSnapshotScope {
-            sources: expected_sources,
-            buckets: all_buckets,
-            identity: ScannerSnapshotIdentity {
-                cycle: first.info.next_cycle,
-                leader_epoch: first.info.leader_epoch,
-                plan_digest: TEST_PLAN_DIGEST,
-                coverage_digest: TEST_COVERAGE_DIGEST,
-                tier_registry_generation: first.info.tier_registry_generation,
-            },
-        },
-        tier_registry_names,
-        bucket_plan_complete,
-        budget_elapsed,
-        cancelled,
-    )
-}
-
-#[test]
-fn completed_data_usage_info_rejects_duplicate_bucket_inventory() {
-    let set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
-    let buckets = vec!["bucket".to_string(), "bucket".to_string()];
-    assert!(completed_data_usage_info_for_test(&[set], &buckets, false, false).is_none());
-}
-
-#[test]
-fn completed_data_usage_info_rejects_extra_or_detached_bucket_data() {
-    let buckets = vec!["bucket".to_string()];
-    let mut set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
-    set.replace(
-        "unlisted",
-        DATA_USAGE_ROOT,
-        DataUsageEntry {
-            objects: 1,
-            ..Default::default()
-        },
-    );
-    assert!(completed_data_usage_info_for_test(&[set.clone()], &buckets, false, false).is_none());
-    set.cache
-        .get_mut(DATA_USAGE_ROOT)
-        .expect("set root")
-        .children
-        .remove(&hash_path("unlisted").key());
-    assert!(
-        completed_data_usage_info_for_test(&[set], &buckets, false, false).is_none(),
-        "orphaned data must not disappear from authoritative accounting"
-    );
-}
-
-#[test]
-fn completed_data_usage_info_rejects_disconnected_expected_bucket() {
-    let buckets = vec!["bucket".to_string()];
-    let mut set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
-    set.cache.get_mut(DATA_USAGE_ROOT).expect("set root").children.clear();
-    assert!(completed_data_usage_info_for_test(&[set], &buckets, false, false).is_none());
-}
-
-#[test]
-fn completed_data_usage_info_rejects_root_scalar_data_and_unknown_key_format() {
-    let buckets = vec!["bucket".to_string()];
-    let set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
-    let mut scalar_root = set.clone();
-    scalar_root.cache.get_mut(DATA_USAGE_ROOT).expect("set root").size = 10;
-    assert!(completed_data_usage_info_for_test(&[scalar_root], &buckets, false, false).is_none());
-    let mut future_format = set;
-    future_format.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT + 1;
-    assert!(completed_data_usage_info_for_test(&[future_format], &buckets, false, false).is_none());
-}
-
-#[test]
-fn completed_data_usage_info_binds_all_results_to_requested_identity() {
-    let buckets = vec!["bucket".to_string()];
-    let source = DataUsageCacheSource::new(0, 0);
-    let sources = HashSet::from([source]);
-    let set = completed_root_cache("bucket", 2, 10, source);
-    let identity = ScannerSnapshotIdentity {
-        cycle: 0,
-        leader_epoch: 0,
-        plan_digest: TEST_PLAN_DIGEST,
-        coverage_digest: TEST_COVERAGE_DIGEST,
-        tier_registry_generation: None,
-    };
-    let results = [set];
-    for expected in [
-        ScannerSnapshotIdentity { cycle: 1, ..identity },
-        ScannerSnapshotIdentity {
-            leader_epoch: 1,
-            ..identity
-        },
-        ScannerSnapshotIdentity {
-            plan_digest: DataUsageScanPlanDigest([9; 32]),
-            ..identity
-        },
-        ScannerSnapshotIdentity {
-            tier_registry_generation: Some(1),
-            ..identity
-        },
-        ScannerSnapshotIdentity {
-            coverage_digest: DataUsageScanPlanDigest([4; 32]),
-            ..identity
-        },
-    ] {
-        let scope = ScannerSnapshotScope {
-            sources: &sources,
-            buckets: &buckets,
-            identity: expected,
-        };
-        assert!(completed_data_usage_info(&results, &scope, &[], true, false, false).is_none());
-    }
-    let scope = ScannerSnapshotScope {
-        sources: &sources,
-        buckets: &buckets,
-        identity,
-    };
-    let (usage, _) = completed_data_usage_info(&results, &scope, &[], true, false, false)
-        .expect("the requested complete scope remains publishable");
-    assert_eq!(usage.objects_total_count, 2);
-    assert!(usage.is_complete_bucket_usage_snapshot());
+    completed_data_usage_info(results, &expected_sources, all_buckets, &[], true, budget_elapsed, cancelled)
 }

 fn lkg_root_cache(bucket: &str, objects: usize, source: DataUsageCacheSource) -> DataUsageCache {
@@ -296,7 +136,7 @@ fn partial_usage_is_observational_not_authoritative_for_quota() {
    let expected = HashSet::from([current_source, stalled_source]);

    assert!(
-        completed_usage_for_scope(&[current.clone(), stalled.clone()], &expected, &all_buckets, &[], true, false, false)
+        completed_data_usage_info(&[current.clone(), stalled.clone()], &expected, &all_buckets, &[], true, false, false)
            .is_none()
    );
    let (observed, _) = observational_data_usage_info(&[current, stalled], &expected, &all_buckets, &[], TEST_PLAN_DIGEST, 8, 3)
@@ -669,7 +509,7 @@ fn completed_data_usage_info_accepts_unknown_only_with_current_registry_generati

    let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]);
    assert!(
-        completed_usage_for_scope(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_some()
+        completed_data_usage_info(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_some()
    );
 }

@@ -707,7 +547,7 @@ fn completed_data_usage_info_rejects_non_registry_tier_in_current_generation() {

    let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]);
    assert!(
-        completed_usage_for_scope(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_none()
+        completed_data_usage_info(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_none()
    );
 }

@@ -779,7 +619,6 @@ fn current_cache_root_with_new_tier_generation_resets_old_cache() {
        DataUsageCacheReuseOptions {
            require_source: false,
            tier_registry_generation: Some(2),
-            checkpoint_identity: None,
        },
    );

@@ -859,8 +698,6 @@ fn completed_data_usage_info_publishes_confirmed_empty_namespace() {
            source: Some(DataUsageCacheSource::new(0, 0)),
            snapshot_complete: true,
            scan_plan_digest: Some(TEST_PLAN_DIGEST),
-            scan_coverage_digest: Some(TEST_COVERAGE_DIGEST),
-            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
            ..Default::default()
        },
        ..Default::default()
@@ -1018,7 +855,7 @@ fn completed_data_usage_info_requires_exact_topology_sources() {
    let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0), DataUsageCacheSource::new(1, 0)]);

    assert!(
-        completed_usage_for_scope(&[first_set, unexpected_set], &expected_sources, &all_buckets, &[], true, false, false)
+        completed_data_usage_info(&[first_set, unexpected_set], &expected_sources, &all_buckets, &[], true, false, false)
            .is_none()
    );
 }
@@ -1029,7 +866,7 @@ fn completed_data_usage_info_rejects_incomplete_bucket_plan() {
    let set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
    let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]);

-    assert!(completed_usage_for_scope(&[set], &expected_sources, &all_buckets, &[], false, false, false).is_none());
+    assert!(completed_data_usage_info(&[set], &expected_sources, &all_buckets, &[], false, false, false).is_none());
 }

 #[test]
@@ -1339,90 +1176,6 @@ fn dirty_bucket_cache_digest_changes_with_generation() {
    assert!(!cache_snapshot_is_current(&cache, "photos", source, 11, 0, second));
 }

-#[test]
-fn scoped_scan_bucket_work_proof_fences_same_cycle_cache() {
-    let source = DataUsageCacheSource::new(0, 0);
-    let structural_plan = DataUsageScanPlanDigest([9; 32]);
-    let normal_plan = scanner_bucket_work_digest(structural_plan, HealScanMode::Normal, false);
-    assert_eq!(normal_plan, structural_plan, "ordinary work keeps the existing digest contract");
-    for (scan_mode, full) in [(HealScanMode::Deep, false), (HealScanMode::Normal, true)] {
-        let requested_plan = scanner_bucket_work_digest(structural_plan, scan_mode, full);
-        assert_ne!(requested_plan, normal_plan);
-        let mut cache = DataUsageCache {
-            info: DataUsageCacheInfo {
-                name: "cold".to_string(),
-                next_cycle: 7,
-                leader_epoch: 11,
-                last_update: Some(SystemTime::UNIX_EPOCH),
-                source: Some(source),
-                snapshot_complete: true,
-                scan_plan_digest: Some(normal_plan),
-                cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
-                ..Default::default()
-            },
-            ..Default::default()
-        };
-        cache.replace("cold", "", DataUsageEntry::default());
-        assert!(cache_snapshot_is_current(&cache, "cold", source, 7, 11, normal_plan));
-        assert!(matches!(
-            current_cache_root_or_prepare(&mut cache, "cold", source, 7, 11, requested_plan, true),
-            DataUsageCacheScanState::Prepared {
-                outcome: DataUsageCachePrepareOutcome::Reset,
-                ..
-            }
-        ));
-        assert!(cache.cache.is_empty(), "different work requirements must enter a fresh walk");
-        cache.replace("cold", "", DataUsageEntry::default());
-        cache.info.snapshot_complete = true;
-        cache.info.last_update = Some(SystemTime::UNIX_EPOCH);
-        assert!(
-            matches!(
-                current_cache_root_or_prepare(&mut cache, "cold", source, 7, 11, requested_plan, true),
-                DataUsageCacheScanState::Current(_)
-            ),
-            "completed matching work may satisfy the same intent retry"
-        );
-    }
-    assert_eq!(
-        scanner_bucket_work_digest(structural_plan, HealScanMode::Deep, false),
-        scanner_bucket_work_digest(structural_plan, HealScanMode::Deep, true)
-    );
-}
-
-#[test]
-fn scoped_scan_complete_root_requires_current_coverage_from_every_set() {
-    let sources = HashSet::from([DataUsageCacheSource::new(0, 0), DataUsageCacheSource::new(1, 0)]);
-    let buckets = vec!["bucket".to_string()];
-    let coverage = DataUsageScanPlanDigest([4; 32]);
-    let scope = ScannerSnapshotScope {
-        sources: &sources,
-        buckets: &buckets,
-        identity: ScannerSnapshotIdentity {
-            cycle: 0,
-            leader_epoch: 0,
-            plan_digest: TEST_PLAN_DIGEST,
-            coverage_digest: coverage,
-            tier_registry_generation: None,
-        },
-    };
-    for (first_coverage, second_coverage, valid) in [
-        (Some(coverage), Some(coverage), true),
-        (None, Some(coverage), false),
-        (Some(coverage), None, false),
-        (None, None, false),
-        (Some(coverage), Some(DataUsageScanPlanDigest([5; 32])), false),
-    ] {
-        let mut first = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
-        let mut second = completed_root_cache("bucket", 3, 10, DataUsageCacheSource::new(1, 0));
-        first.info.scan_coverage_digest = first_coverage;
-        second.info.scan_coverage_digest = second_coverage;
-        assert_eq!(
-            completed_data_usage_info(&[first, second], &scope, &[], true, false, false).is_some(),
-            valid
-        );
-    }
-}
-
 #[test]
 fn scanner_cache_lock_resource_is_scoped_to_cache_source() {
    let cache_name = "photos/.usage-cache.bin";
@@ -124,41 +124,6 @@ async fn setup_two_pool_scanner_store() -> (tempfile::TempDir, Arc<ECStore>) {
    (temp_dir, store)
 }

-#[tokio::test]
-#[serial]
-async fn checkpoint_fixture_bucket_identity_uses_its_set_instance_owner() {
-    let (_first_dir, first) = setup_two_pool_scanner_store().await;
-    first
-        .make_bucket("checkpoint-identity", &MakeBucketOptions::default())
-        .await
-        .expect("first instance bucket");
-    let first_identity =
-        scanner_bucket_checkpoint_identity(&first.pools[0].disk_set[0], "checkpoint-identity", 0, 7, HealScanMode::Normal)
-            .await
-            .expect("first durable identity");
-    let (_second_dir, second) = setup_two_pool_scanner_store().await;
-    second
-        .make_bucket("checkpoint-identity", &MakeBucketOptions::default())
-        .await
-        .expect("second instance bucket");
-    let second_identity =
-        scanner_bucket_checkpoint_identity(&second.pools[0].disk_set[0], "checkpoint-identity", 0, 7, HealScanMode::Normal)
-            .await
-            .expect("second durable identity");
-    assert_ne!(first_identity.bucket_incarnation, second_identity.bucket_incarnation);
-    assert_eq!(
-        scanner_bucket_checkpoint_identity(&first.pools[0].disk_set[0], "checkpoint-identity", 0, 7, HealScanMode::Normal)
-            .await
-            .expect("first owner remains bound"),
-        first_identity
-    );
-    assert!(
-        scanner_bucket_checkpoint_identity(&first.pools[0].disk_set[0], "missing-checkpoint-bucket", 0, 7, HealScanMode::Normal)
-            .await
-            .is_err()
-    );
-}
-
 #[tokio::test]
 #[serial]
 async fn scanner_cache_locks_block_same_source_workers() {
@@ -318,211 +283,6 @@ async fn scanner_cycle_is_deferred_while_terminal_decommission_is_blocked() {
    }
 }

-#[tokio::test]
-#[serial]
-async fn scoped_scan_production_entry_preserves_deep_and_full_maintenance_work() {
-    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
-    clear_dirty_usage_buckets_for_tests();
-    for bucket in ["hot-bucket", "cold-bucket"] {
-        store
-            .make_bucket(bucket, &MakeBucketOptions::default())
-            .await
-            .expect("bucket should be created");
-        let mut reader = ScannerPutObjReader::from_vec(b"initial".to_vec());
-        store.pools[0].disk_set[0]
-            .put_object(bucket, "initial", &mut reader, &ScannerObjectOptions::default())
-            .await
-            .expect("initial object should persist");
-    }
-    let mut baseline = None;
-    for (index, (scan_mode, requires_full_scan, explicit_scope)) in [
-        (HealScanMode::Normal, true, false),
-        (HealScanMode::Normal, false, false),
-        (HealScanMode::Deep, false, false),
-        (HealScanMode::Normal, true, false),
-        (HealScanMode::Deep, false, true),
-        (HealScanMode::Normal, true, true),
-    ]
-    .into_iter()
-    .enumerate()
-    {
-        if index > 0 {
-            let mut reader = ScannerPutObjReader::from_vec(b"maintenance".to_vec());
-            store.pools[0].disk_set[0]
-                .put_object("cold-bucket", &format!("added-{index}"), &mut reader, &ScannerObjectOptions::default())
-                .await
-                .expect("cold bucket mutation should persist");
-            // Only the hot bucket is in the usage hint. The cold result must
-            // come from this cycle's storage walk, not its previous baseline.
-            record_dirty_usage_bucket("hot-bucket");
-        }
-        let requested_scope = if explicit_scope {
-            ScannerBucketScanScope::from_dirty_buckets(
-                HashSet::from(["hot-bucket".to_string()]),
-                DataUsageScanPlanDigest([7; 32]),
-            )
-        } else {
-            ScannerBucketScanScope::default()
-        };
-        let ctx = CancellationToken::new();
-        let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
-        let (updates, mut receiver) = mpsc::channel(1);
-        let (observer, observed_scope) = tokio::sync::oneshot::channel();
-        let cycle = u64::try_from(index + 1).expect("test cycle should fit");
-        let result = tokio::time::timeout(
-            Duration::from_secs(30),
-            nsscanner_with_storage_status_scoped(
-                store.as_ref(),
-                ScannerCycleRequest {
-                    ctx,
-                    budget,
-                    updates,
-                    want_cycle: cycle,
-                    leader_epoch: 11,
-                    scan_mode,
-                    scan_scope: requested_scope,
-                    persisted_usage_baseline: baseline,
-                    requires_full_scan,
-                    resolved_scope_observer: Some(observer),
-                },
-            ),
-        )
-        .await
-        .expect("cycle should finish within the test deadline")
-        .expect("cycle should succeed");
-        assert_eq!(result.status, ScannerCycleStatus::Complete, "cycle {cycle}");
-        let resolved = observed_scope.await.expect("production resolver should report its scope");
-        if index == 1 {
-            assert_eq!(
-                resolved.selected_buckets.as_deref(),
-                Some(&HashSet::from(["hot-bucket".to_string()])),
-                "ordinary dirty work must retain the existing planner"
-            );
-        } else {
-            assert!(resolved.is_default(), "cycle {cycle} must visit the full maintenance scope");
-        }
-        let mut snapshot = receiver.recv().await.expect("cycle should publish a snapshot");
-        assert!(snapshot.usage_snapshot_complete, "cycle {cycle}");
-        assert_eq!(
-            snapshot.buckets_usage["cold-bucket"].objects_count,
-            u64::try_from(index + 1).expect("count should fit")
-        );
-        assert_eq!(snapshot.buckets_usage["hot-bucket"].objects_count, 1);
-        assert_eq!(snapshot.scanner_cycle, Some(cycle));
-        assert_eq!(snapshot.scanner_epoch, Some(11));
-        snapshot.usage_snapshot_converged = Some(true);
-        baseline = Some(Bytes::from(serde_json::to_vec(&snapshot).expect("complete baseline should encode")));
-    }
-    clear_dirty_usage_buckets_for_tests();
-}
-
-#[tokio::test]
-#[serial]
-async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure() {
-    for (scan_mode, requires_full_scan) in [
-        (HealScanMode::Normal, false),
-        (HealScanMode::Deep, false),
-        (HealScanMode::Normal, true),
-    ] {
-        let (_temp_dir, store) = setup_two_pool_scanner_store().await;
-        clear_dirty_usage_buckets_for_tests();
-        for bucket in ["hot-bucket", "cold-bucket"] {
-            store
-                .make_bucket(bucket, &MakeBucketOptions::default())
-                .await
-                .expect("bucket should be created");
-            let mut reader = ScannerPutObjReader::from_vec(b"initial".to_vec());
-            store.pools[0].disk_set[0]
-                .put_object(bucket, "initial", &mut reader, &ScannerObjectOptions::default())
-                .await
-                .expect("initial object should persist");
-        }
-        let ctx = CancellationToken::new();
-        let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
-        let (updates, receiver) = mpsc::channel(1);
-        drop(receiver);
-        let failed = tokio::time::timeout(
-            Duration::from_secs(30),
-            nsscanner_with_storage_status_scoped(
-                store.as_ref(),
-                ScannerCycleRequest {
-                    ctx,
-                    budget,
-                    updates,
-                    want_cycle: 7,
-                    leader_epoch: 11,
-                    scan_mode: HealScanMode::Normal,
-                    scan_scope: ScannerBucketScanScope::default(),
-                    persisted_usage_baseline: None,
-                    requires_full_scan: false,
-                    resolved_scope_observer: None,
-                },
-            ),
-        )
-        .await
-        .expect("normal scan should finish")
-        .expect_err("root delivery must fail after bucket cache persistence");
-        assert!(failed.to_string().contains("receiver closed"), "{failed}");
-        let cache_name = path_join_buf(&["cold-bucket", DATA_USAGE_CACHE_NAME]);
-        let mut cached = DataUsageCache::default();
-        cached
-            .load(store.pools[0].disk_set[0].clone(), &cache_name)
-            .await
-            .expect("normal bucket cache should have committed");
-        assert!(cached.info.snapshot_complete);
-        assert_eq!(cached.info.next_cycle, 7);
-        assert_eq!(
-            cached
-                .checked_flatten("cold-bucket")
-                .expect("cached root should be valid")
-                .objects,
-            1
-        );
-
-        let mut reader = ScannerPutObjReader::from_vec(b"maintenance".to_vec());
-        store.pools[0].disk_set[0]
-            .put_object("cold-bucket", "new", &mut reader, &ScannerObjectOptions::default())
-            .await
-            .expect("new cold object should persist");
-        record_dirty_usage_bucket("hot-bucket");
-        if scan_mode == HealScanMode::Normal && !requires_full_scan {
-            record_dirty_usage_bucket("cold-bucket");
-        }
-        let ctx = CancellationToken::new();
-        let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
-        let (updates, mut receiver) = mpsc::channel(1);
-        let result = tokio::time::timeout(
-            Duration::from_secs(30),
-            nsscanner_with_storage_status_scoped(
-                store.as_ref(),
-                ScannerCycleRequest {
-                    ctx,
-                    budget,
-                    updates,
-                    want_cycle: 7,
-                    leader_epoch: 11,
-                    scan_mode,
-                    scan_scope: ScannerBucketScanScope::default(),
-                    persisted_usage_baseline: None,
-                    requires_full_scan,
-                    resolved_scope_observer: None,
-                },
-            ),
-        )
-        .await
-        .expect("maintenance scan should finish")
-        .expect("maintenance scan should succeed");
-        assert_eq!(result.status, ScannerCycleStatus::Complete);
-        let snapshot = receiver.recv().await.expect("maintenance snapshot should be published");
-        assert_eq!(snapshot.scanner_cycle, Some(7));
-        assert_eq!(
-            snapshot.buckets_usage["cold-bucket"].objects_count, 2,
-            "{scan_mode:?}/full={requires_full_scan} must not replay the same-cycle Normal root"
-        );
-        clear_dirty_usage_buckets_for_tests();
-    }
-}
-
 #[tokio::test]
 async fn data_usage_publish_fails_when_receiver_is_closed() {
    let (updates, receiver) = mpsc::channel(1);
@@ -1133,7 +893,6 @@ fn complete_set_usage_cache(buckets: &[(&str, usize)], scan_plan_digest: DataUsa
            source: Some(DataUsageCacheSource::new(1, 2)),
            snapshot_complete: true,
            scan_plan_digest: Some(scan_plan_digest),
-            scan_coverage_digest: Some(scan_plan_digest),
            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
            tier_registry_generation: Some(13),
            ..Default::default()
@@ -1251,8 +1010,6 @@ async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers
            all_buckets: Arc::new(Vec::new()),
            scope: ScannerBucketScanScope::default(),
            digest: DataUsageScanPlanDigest([6; 32]),
-            bucket_coverage_digest: DataUsageScanPlanDigest([6; 32]),
-            requires_full_scan: false,
            execution_digest: empty_execution,
            leader_epoch: 11,
            tier_registry_generation: 13,
@@ -1382,36 +1139,6 @@ fn scoped_scan_selects_only_current_dirty_buckets_after_baseline_validation() {
    assert_ne!(scope.baseline_scan_plan_digest, Some(baseline_scan_plan_digest));
 }

-#[test]
-fn scoped_scan_baseline_work_proof_requires_uniform_known_set_identity() {
-    let source = DataUsageCacheSource::new(1, 2);
-    let second_source = DataUsageCacheSource::new(1, 3);
-    let sources = HashSet::from([source, second_source]);
-    let structural = DataUsageScanPlanDigest([9; 32]);
-    let full = scanner_bucket_work_digest(structural, HealScanMode::Normal, true);
-    let deep = scanner_bucket_work_digest(structural, HealScanMode::Deep, true);
-    let encoded = complete_usage_baseline(source, full, 7, 11);
-    let baseline: DataUsageInfo = serde_json::from_slice(&encoded).expect("baseline should decode");
-    for (second_plan, expected) in [(full, Some(full)), (deep, None), (DataUsageScanPlanDigest([8; 32]), None)] {
-        let mut candidate = baseline.clone();
-        let mut second = candidate.usage_snapshot_set_states[0].clone();
-        second.set_index = 3;
-        second.scan_plan_digest = Some(second_plan.0);
-        candidate.usage_snapshot_set_states.push(second);
-        let data = Bytes::from(serde_json::to_vec(&candidate).expect("candidate should encode"));
-        assert_eq!(
-            complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
-                data: Some(&data),
-                expected_sources: &sources,
-                leader_epoch: 11,
-                want_cycle: 8,
-                scan_plan_digest: structural,
-            }),
-            expected
-        );
-    }
-}
-
 fn peer_dirty_usage_snapshot(
    instance_id: &str,
    generation: u64,
@@ -1494,15 +1221,8 @@ fn verified_remote_dirty_usage_buckets_rejects_incomplete_or_stale_peer_state()
    }
 }

-fn bucket_info_with_created_time(name: &str) -> BucketInfo {
-    BucketInfo {
-        created: Some(time::OffsetDateTime::UNIX_EPOCH),
-        ..bucket_info(name)
-    }
-}
-
 #[test]
-fn scoped_set_scan_rebuilds_selected_buckets_and_drops_deleted_buckets() {
+fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
    let baseline_digest = DataUsageScanPlanDigest([1; 32]);
    let current_digest = DataUsageScanPlanDigest([2; 32]);
    let mut old_cache = complete_set_usage_cache(&[("stable", 10), ("dirty", 20), ("deleted", 30)], baseline_digest);
@@ -1515,11 +1235,8 @@ fn scoped_set_scan_rebuilds_selected_buckets_and_drops_deleted_buckets() {
            ..Default::default()
        },
    );
-    let all_buckets = vec![
-        bucket_info_with_created_time("stable"),
-        bucket_info_with_created_time("dirty"),
-    ];
-    let selected_buckets = Arc::new(HashSet::from(["stable".to_string(), "dirty".to_string(), "deleted".to_string()]));
+    let all_buckets = vec![bucket_info("stable"), bucket_info("dirty")];
+    let selected_buckets = Arc::new(HashSet::from(["dirty".to_string(), "deleted".to_string()]));

    let prepared = prepare_scoped_set_scan(
        &old_cache,
@@ -1539,16 +1256,12 @@ fn scoped_set_scan_rebuilds_selected_buckets_and_drops_deleted_buckets() {
    )
    .expect("complete matching set cache should support a scoped scan");

-    assert_eq!(
-        prepared.buckets.iter().map(|bucket| bucket.name.as_str()).collect::<Vec<_>>(),
-        ["stable", "dirty"]
-    );
+    assert_eq!(prepared.buckets.iter().map(|bucket| bucket.name.as_str()).collect::<Vec<_>>(), ["dirty"]);
    let stable = prepared
        .cache
        .checked_flatten("stable")
-        .expect("selected bucket placeholder should exist");
-    assert_eq!((stable.size, stable.objects), (0, 0));
-    assert!(prepared.cache.find("stable/prefix").is_none());
+        .expect("unselected bucket subtree should be retained");
+    assert_eq!((stable.size, stable.objects), (15, 2));
    assert_eq!(prepared.cache.find("dirty").map(|entry| (entry.size, entry.objects)), Some((0, 0)));
    assert!(prepared.cache.find("deleted").is_none());
    assert_eq!(prepared.cache.info.scan_plan_digest, Some(current_digest));
@@ -1559,41 +1272,11 @@ fn scoped_set_scan_rebuilds_selected_buckets_and_drops_deleted_buckets() {
    assert_eq!(prepared.cache.info.lkg_scan_plan_digest, Some(baseline_digest));
 }

-#[test]
-fn scoped_set_scan_rejects_unbound_bucket_incarnations() {
-    let baseline_digest = DataUsageScanPlanDigest([1; 32]);
-    let old_cache = complete_set_usage_cache(&[("stable", 10), ("dirty", 20)], baseline_digest);
-    let scope = ScannerBucketScanScope {
-        selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))),
-        baseline_scan_plan_digest: Some(baseline_digest),
-    };
-    let generation = ScannerSetCacheGeneration {
-        want_cycle: 8,
-        leader_epoch: 11,
-        tier_registry_generation: 13,
-        source: DataUsageCacheSource::new(1, 2),
-        scan_plan_digest: DataUsageScanPlanDigest([2; 32]),
-    };
-    for created in [
-        None,
-        Some(OffsetDateTime::UNIX_EPOCH),
-        Some(OffsetDateTime::UNIX_EPOCH + time::Duration::days(1)),
-    ] {
-        let mut stable = bucket_info("stable");
-        stable.created = created;
-        let buckets = vec![stable, bucket_info_with_created_time("dirty")];
-        assert!(
-            prepare_scoped_set_scan(&old_cache, &buckets, &buckets, &scope, generation).is_none(),
-            "missing identity, volume timestamps and same-name recreation must all rebuild"
-        );
-    }
-}
-
 #[test]
 fn scoped_set_scan_falls_back_when_an_unselected_bucket_has_no_baseline() {
    let baseline_digest = DataUsageScanPlanDigest([3; 32]);
    let old_cache = complete_set_usage_cache(&[("stable", 10)], baseline_digest);
-    let all_buckets = vec![bucket_info_with_created_time("stable"), bucket_info_with_created_time("new")];
+    let all_buckets = vec![bucket_info("stable"), bucket_info("new")];

    assert!(
        prepare_scoped_set_scan(
@@ -1619,7 +1302,7 @@ fn scoped_set_scan_falls_back_when_an_unselected_bucket_has_no_baseline() {
 #[test]
 fn scoped_set_scan_requires_an_exact_complete_baseline() {
    let baseline_digest = DataUsageScanPlanDigest([5; 32]);
-    let all_buckets = vec![bucket_info_with_created_time("dirty")];
+    let all_buckets = vec![bucket_info("dirty")];
    let scope = ScannerBucketScanScope {
        selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))),
        baseline_scan_plan_digest: Some(baseline_digest),
@@ -1640,10 +1323,6 @@ fn scoped_set_scan_requires_an_exact_complete_baseline() {
    not_durable.info.last_update = None;
    assert!(prepare_scoped_set_scan(&not_durable, &all_buckets, &all_buckets, &scope, generation).is_none());

-    let mut unscoped_usage = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
-    unscoped_usage.cache.get_mut(DATA_USAGE_ROOT).expect("set root").objects = 1;
-    assert!(prepare_scoped_set_scan(&unscoped_usage, &all_buckets, &all_buckets, &scope, generation).is_none());
-
    let mut wrong_digest = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
    wrong_digest.info.scan_plan_digest = Some(DataUsageScanPlanDigest([7; 32]));
    assert!(prepare_scoped_set_scan(&wrong_digest, &all_buckets, &all_buckets, &scope, generation).is_none());
@@ -1654,13 +1333,6 @@ fn scoped_set_scan_requires_an_exact_complete_baseline() {
    };
    let complete = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
    assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &empty_scope, generation).is_none());
-    assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &scope, generation).is_some());
-
-    let unidentified_buckets = vec![bucket_info("dirty")];
-    assert!(
-        prepare_scoped_set_scan(&complete, &unidentified_buckets, &unidentified_buckets, &scope, generation).is_some(),
-        "fully selected buckets are rebuilt without reusing an unproven incarnation"
-    );

    let mut future_cache = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
    future_cache.info.next_cycle = generation.want_cycle.saturating_add(1);
@@ -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.
@@ -9,18 +9,14 @@ cargo test -p rustfs-scanner --lib checkpoint_fixture -- --list
 RUST_MIN_STACK=4194304 cargo test -p rustfs-scanner --lib checkpoint_fixture -- --nocapture
 ```

-Both the unchanged-plan and hot-plan cases require durable static coverage to increase each round. `LostAtPrepare` identifies invalidation before traversal; `LostAtReload` identifies loss between the returned cache and persisted data; `WalkWithoutRetention` identifies visited growth without durable coverage growth. Missing, corrupt, empty-root, and oversized checkpoint inputs are rejected by the strict fixture reader. Save failure and publication-epoch rejection must preserve the preceding file bytes. Parent cancellation is checked separately from object-budget exhaustion. Superseded classification is tested separately from either incomplete outcome.
+The unchanged-plan case requires durable static coverage to increase each round. The hot-plan diagnostic changes the bucket plan digest between rounds and reports where coverage is lost without asserting that a particular defect must remain present. To require progress in this diagnostic as well:

-After three interrupted rounds, the fixture overwrites a previously visited object with two versions, deletes another visited object, and creates one more hot object. It then keeps the same four-object budget until the stable namespace is certified. Finishing a sweep that spans different mutation plans must first return partial; a subsequent verification sweep must produce exactly 25 objects, 2 versioned entries, and 34 logical bytes. There is no final unbudgeted sweep.
+```sh
+RUST_MIN_STACK=4194304 RUSTFS_CHECKPOINT_REQUIRE_PROGRESS=1 cargo test -p rustfs-scanner --lib checkpoint_fixture_hot_digest_diagnostic -- --nocapture
+```

-The new bucket checkpoint binds the persisted bucket incarnation, set layout, publication epoch, tier generation and scan mode, with the existing source/leader/key-format checks. Its forward sweep records the starting and requested mutation plans separately. Partial sweeps omit the legacy `scan_plan_digest`, so older readers rebuild instead of treating mixed observations as a current complete snapshot. Completed sweeps restore that digest only after covering one mutation plan. Unsupported or missing identities retain the legacy rebuild path. The stable-plan fast path requires a complete snapshot without unfinished checkpoint state; its interrupted result must enter forward validation on reload.
-
-A coverage receipt binds the completed traversal frontier to its scope, starting plan and canonical covered-prefix digest. It excludes ancestor aggregates and the unvisited suffix, so unrelated suffix changes cannot invalidate completed work. Cancellation seals only the completed frontier; failed child traversal and known failed-metadata skips block further frontier advancement. A saved cursor pointing at an existing but unvisited old subtree is rejected unless it agrees with that receipt. The receipt is a consistency check for storage owned by the scanner, not authentication against a party able to forge the entire cache and recompute its digest. New metadata remains map-encoded with optional top-level fields.
-
-The Normal-to-Deep regression holds the mutation plan fixed, changes metadata in a previously visited prefix, and verifies that the real Deep disk-scan entry point reads that prefix again. It also checks that a complete Normal cache cannot satisfy the Deep `Current` path. The fixture disables heal side effects; it proves traversal re-entry, not actual bitrot detection or repair. Additional tests cover map round-trips, stale identities, existing-but-uncovered cursors, coverage gaps, and per-instance metadata ownership.
-
-This fixture bounds object processing after directory enumeration. It does not prove fixed-budget enumeration of arbitrarily wide directories or real process-restart convergence. Those gates require a storage-owned resumable enumeration capability, including its initial construction cost; a readdir offset, an in-memory iterator or a last-name filter is not that capability.
+A nonzero exit from the strict command means that walked work did not become additional retained static coverage. `LostAtPrepare` identifies invalidation before traversal; `LostAtReload` identifies loss between the returned cache and persisted data; `WalkWithoutRetention` identifies visited growth without durable coverage growth. Missing, corrupt, empty-root, and oversized checkpoint inputs are rejected by the strict fixture reader. Save failure and publication-epoch rejection must preserve the preceding file bytes. Parent cancellation is checked separately from object-budget exhaustion. Superseded classification is tested separately from either incomplete outcome.

 For every saved partial cache, the fixture also passes its progress through the production authenticated remote terminal-frame writer and stream consumer. A remote partial result must remain partial even when its progress reports visited objects. This covers the return-frame contract; it does not execute the remote RPC server, distributed locks, EC quorum persistence, mixed-version peers, process crashes, or fsync durability. The file backend models revision preconditions and persistence errors, not a concurrent object store.

-The synthetic namespace contains no customer data. Temporary files are removed with their owning fixture. Rolling back to a reader without the optional checkpoint metadata rebuilds partial coverage; it must not clear quota floors or complete authoritative snapshots. A passing fixture alone does not establish that the field report in [issue #7108](https://github.com/rustfs/rustfs/issues/7108) has been independently reproduced or fixed. A field diagnosis must separately identify the source capture, cycle and leader identity, and decoded bucket/set caches.
+The synthetic namespace contains no customer data. Temporary files are removed with their owning fixture. Production scan semantics and persistent formats are unchanged, so rollback consists of removing these tests and this guide. A passing fixture alone does not establish that the field report in [issue #7108](https://github.com/rustfs/rustfs/issues/7108) has been independently reproduced or fixed. A field diagnosis must separately identify the source capture, cycle and leader identity, and decoded bucket/set caches.
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	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