fix(ci): integrate strict gate with E2E provenance

test(e2e): record compiled Linux receipt test membership
test(e2e): record compiled Darwin receipt test membership
2026-09-09 13:46:05 +00:00 · 2026-09-09 05:56:33 +08:00 · 2026-09-09 01:48:41 +08:00 · 2026-09-09 00:31:36 +08:00 · 2026-09-08 23:55:05 +08:00 · 2026-09-08 22:45:34 +08:00
35 changed files with 919 additions and 1648 deletions
@@ -1,2 +1,2 @@
-sha256-darwin=f0c78fdb93471575d9a64c5c46eae6c806bdd0bc10a6e33d7fb574aabd8db5a3
-sha256-linux=03ed7016cab672de9320e31375a0358eceacb4408b0e79cf063614fa7c878b87
+sha256-darwin=845feb5859c4063c38307ada8f263f4039ebaf54c510bbc9177c4ab0dba2d8a9
+sha256-linux=22320a04e541ef27cf1d0df3670ab3fafb62c57beef9a57eaff895e84a1e8380
@@ -39,6 +39,7 @@ script-tests: ## Run shell script tests
 	./scripts/test_python_bin.sh
 	./scripts/check_embedded_secrets.sh --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_test_wiring.py --self-test
+	$(RUSTFS_PYTHON_BIN) ./scripts/test_e2e_binary.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/ci_gate.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_security_coverage.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_scheduled_validation_freshness.py --self-test
@@ -197,12 +197,6 @@ test-group = 'e2e-cluster-nightly'
 filter = 'package(e2e_test) & (test(/^kms::kms_vault_test::/) | test(/^kms::kms_rekey_sweep_test::/) | test(/^kms::configured_roundtrip_test::test_configured_vault_kms_admin_and_versioned_cleanup$/))'
 test-group = 'e2e-vault'

-# This four-disk, 65-member rollback probe already drives up to 32 concurrent
-# durable deletions. Reserve this nextest run's capacity for its progress oracle.
-[[profile.default.overrides]]
-filter = 'package(rustfs-ecstore) & test(=store::init::tests::dispatch_manifest_rollback_bounded_concurrency_reaches_tail_behind_slow_member)'
-threads-required = "num-test-threads"
-
 # ---------------------------------------------------------------------------
 # ci profile — the strict CI gate (ci.yml `cargo nextest run --profile ci`)
 # ---------------------------------------------------------------------------
@@ -568,7 +568,7 @@ jobs:
                      digest.update(chunk)
              return digest.hexdigest()

-          argv = ["cargo", "build", "-p", "rustfs", "--bins", "--features", "e2e-test-hooks"]
+          argv = ["python3", "scripts/e2e_binary.py", "build", "--bins", "--features", "e2e-test-hooks"]
          commit, tree = git("rev-parse", "HEAD"), git("rev-parse", "HEAD^{tree}")
          clean_before = not git("status", "--porcelain", "--untracked-files=normal")
          if not clean_before:
@@ -602,6 +602,7 @@ jobs:
          name: rustfs-debug-binary
          path: |
            target/debug/rustfs
+            target/debug/rustfs.e2e.json
            target/debug/rustfs.e2e-startup-cas-build.json
          if-no-files-found: error
          retention-days: 1
@@ -634,13 +635,15 @@ jobs:
          install-build-packaging-tools: 'false'

      - name: Build debug binary with rio-v2
-        run: cargo build -p rustfs --bins --features rio-v2,e2e-test-hooks
+        run: python3 scripts/e2e_binary.py build --bins --features rio-v2,e2e-test-hooks

      - name: Upload debug binary
        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
        with:
          name: rustfs-debug-binary-rio-v2
-          path: target/debug/rustfs
+          path: |
+            target/debug/rustfs
+            target/debug/rustfs.e2e.json
          if-no-files-found: error
          retention-days: 1

@@ -789,7 +792,7 @@ jobs:
          NEXTEST_ARCHIVE: ${{ runner.temp }}/rustfs-e2e-smoke.tar.zst
          RUSTFS_E2E_LOG_DIR: ${{ runner.temp }}/rustfs-e2e-smoke-logs
        run: |
-          cargo nextest run --profile e2e-smoke --archive-file "${NEXTEST_ARCHIVE}" \
+          python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-smoke --archive-file "${NEXTEST_ARCHIVE}" \
            --status-level all --final-status-level all --failure-output final

      - name: Upload e2e smoke diagnostics
@@ -825,7 +828,7 @@ jobs:
          RUSTFS_TEST_PORT="$(python3 -c 'import socket; s=socket.socket(); s.bind(("127.0.0.1", 0)); print(s.getsockname()[1]); s.close()')"
          RUSTFS_TEST_PORT="${RUSTFS_TEST_PORT}" \
          RUSTFS_TEST_LOG="${RUN_ROOT}/rustfs.log" \
-          ./scripts/e2e-run.sh ./target/debug/rustfs "${RUN_ROOT}/data"
+          python3 scripts/e2e_binary.py run --features e2e-test-hooks -- ./scripts/e2e-run.sh ./target/debug/rustfs "${RUN_ROOT}/data"

      - name: Upload test logs
        if: failure()
@@ -953,6 +956,7 @@ jobs:
          if manifest["binary_sha256"] != digest.hexdigest() or manifest["commit"] != commit:
              raise SystemExit("downloaded hooks binary identity mismatch")
          shutil.copy2(manifest_path, target / manifest_path.name)
+          shutil.copy2(source.with_name("rustfs.e2e.json"), target / "rustfs.e2e.json")
          binary.chmod(0o755)
          PYINPUT

@@ -972,7 +976,7 @@ jobs:
          RUSTFS_E2E_STARTUP_CAS_BINARY: ${{ runner.temp }}/rustfs-startup-cas-input/rustfs
          RUSTFS_E2E_STARTUP_CAS_BUILD_MANIFEST: ${{ runner.temp }}/rustfs-startup-cas-input/rustfs.e2e-startup-cas-build.json
          RUSTFS_E2E_STARTUP_CAS_ARTIFACT_DIR: ${{ runner.temp }}/rustfs-startup-cas-evidence
-        run: cargo nextest run --profile e2e-full -p e2e_test
+        run: python3 scripts/e2e_binary.py run --binary "$RUSTFS_E2E_STARTUP_CAS_BINARY" --features e2e-test-hooks -- cargo nextest run --profile e2e-full -p e2e_test

      - name: Upload junit
        if: always()
@@ -1044,7 +1048,7 @@ jobs:
      - name: Run end-to-end tests
        run: |
          s3s-e2e --version
-          ./scripts/e2e-run.sh ./target/debug/rustfs /tmp/rustfs
+          python3 scripts/e2e_binary.py run --features rio-v2,e2e-test-hooks -- ./scripts/e2e-run.sh ./target/debug/rustfs /tmp/rustfs

      - name: Upload test logs
        if: failure()
@@ -1087,7 +1091,7 @@ jobs:
          S3_PORT="${S3_PORT}" \
          DATA_ROOT="${RUN_ROOT}" \
          S3TESTS_CONF=artifacts/s3tests-single/s3tests.conf \
-          ./scripts/s3-tests/run.sh
+          python3 scripts/e2e_binary.py run --features e2e-test-hooks -- ./scripts/s3-tests/run.sh

      - name: Upload s3 test artifacts
        if: always()
@@ -1169,7 +1173,7 @@ jobs:
          S3_PORT="${S3_PORT}" \
          DATA_ROOT="${RUN_ROOT}" \
          S3TESTS_CONF=artifacts/s3tests-single/s3tests.conf \
-          ./scripts/s3-tests/run.sh
+          python3 scripts/e2e_binary.py run --features e2e-test-hooks -- ./scripts/s3-tests/run.sh

      - name: Upload s3 test artifacts
        if: always()
@@ -151,8 +151,7 @@ jobs:

      - name: Build rustfs binary
        run: |
-          cargo build -p rustfs --bins
-          : > target/debug/rustfs.features
+          python3 scripts/e2e_binary.py build --bins

      - name: Verify distributed e2e membership
        env:
@@ -168,9 +167,9 @@ jobs:
        run: |
          set -euo pipefail
          if [ -n "${FILTER}" ]; then
-            cargo nextest run --profile e2e-distributed -p e2e_test -E "${FILTER}"
+            python3 scripts/e2e_binary.py run -- cargo nextest run --profile e2e-distributed -p e2e_test -E "${FILTER}"
          else
-            cargo nextest run --profile e2e-distributed -p e2e_test --no-tests=fail
+            python3 scripts/e2e_binary.py run -- cargo nextest run --profile e2e-distributed -p e2e_test --no-tests=fail
          fi

      - name: Upload distributed e2e diagnostics
@@ -89,14 +89,10 @@ jobs:
      - name: Verify awscurl
        run: test -x "$AWSCURL_PATH"

-      # Build the rustfs binary once up front. The e2e tests spawn it as a
-      # child process (crates/e2e_test/src/common.rs) and will build it on
-      # demand otherwise, but a single explicit build avoids several parallel
-      # nextest test processes racing to build it at once.
+      # Build once and carry its source/binary identity into the test invocation.
      - name: Build rustfs binary
        run: |
-          cargo build -p rustfs --bins
-          : > target/debug/rustfs.features
+          python3 scripts/e2e_binary.py build --bins

      - name: Verify replication e2e membership
        env:
@@ -108,7 +104,7 @@ jobs:
      - name: Run replication e2e nightly suite
        env:
          RUSTFS_E2E_LOG_DIR: ${{ runner.temp }}/rustfs-e2e-repl-nightly-logs
-        run: cargo nextest run --profile e2e-repl-nightly -p e2e_test
+        run: python3 scripts/e2e_binary.py run -- cargo nextest run --profile e2e-repl-nightly -p e2e_test

      - name: Upload nextest junit report
        if: always()
@@ -144,8 +140,7 @@ jobs:

      - name: Build rustfs binary
        run: |
-          cargo build -p rustfs --bins --features e2e-test-hooks
-          : > target/debug/rustfs.features
+          python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks

      - name: Verify cluster fault e2e membership
        env:
@@ -157,7 +152,7 @@ jobs:
      - name: Run cluster fault e2e nightly suite
        env:
          RUSTFS_E2E_LOG_DIR: ${{ runner.temp }}/rustfs-e2e-nightly-logs
-        run: cargo nextest run --profile e2e-nightly -p e2e_test
+        run: python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-nightly -p e2e_test

      - name: Upload cluster fault diagnostics
        if: always()
@@ -198,6 +193,9 @@ jobs:
          sudo apt-get install -y -qq iproute2
          ss -tn state CLOSE-WAIT >/dev/null

+      - name: Build protocol server
+        run: python3 scripts/e2e_binary.py build --features "$RUSTFS_BUILD_FEATURES"
+
      # The suite owns fixed protocol ports and serializes its internal cases.
      - name: Verify protocol e2e membership
        env:
@@ -210,7 +208,7 @@ jobs:
        env:
          RUSTFS_E2E_LOG_DIR: ${{ runner.temp }}/rustfs-protocol-e2e-logs
        run: >-
-          cargo nextest run -j 1 --profile e2e-protocols -p e2e_test --no-capture
+          python3 scripts/e2e_binary.py run --features "$RUSTFS_BUILD_FEATURES" -- cargo nextest run -j 1 --profile e2e-protocols -p e2e_test --no-capture

      - name: Upload protocol diagnostics
        if: always()
@@ -125,12 +125,11 @@ jobs:

      - name: Build current RustFS binary
        run: |
-          cargo build --locked -p rustfs --bin rustfs
-          : > target/debug/rustfs.features
+          python3 scripts/e2e_binary.py build

      - name: Run upgrade compatibility test
        run: |
-          cargo test --locked -p e2e_test \
+          python3 scripts/e2e_binary.py run -- cargo test --locked -p e2e_test \
            "upgrade_compatibility_test::${{ matrix.test }}" \
            -- --ignored --exact --nocapture

@@ -19,27 +19,17 @@ on:
    - cron: "7 0 * * *"
      timezone: "Asia/Shanghai"
  workflow_dispatch:
-    inputs:
-      branch:
-        description: 'Branch/ref to build and publish as the nightly (empty = scheduled source, see NIGHTLY_BUILD_REF)'
-        required: false
-        default: ''

 permissions:
  contents: read

-# Scheduled builds follow the NIGHTLY_BRANCH repo variable so the channel can
-# be pointed at e.g. `release` for the GA cycle and back to `main` afterwards
-# without touching this file. Manual runs take the `branch` input, falling
-# back to the branch the run was dispatched from.
 concurrency:
-  group: nightly-gnu-build-${{ github.event_name }}-${{ github.event_name == 'schedule' && (vars.NIGHTLY_BRANCH || 'main') || (inputs.branch || github.ref_name) }}
+  group: nightly-gnu-build-main-${{ github.event_name }}
  cancel-in-progress: ${{ github.event_name == 'workflow_dispatch' }}

 env:
  CARGO_TERM_COLOR: always
  RUST_BACKTRACE: 1
-  NIGHTLY_BUILD_REF: ${{ github.event_name == 'schedule' && (vars.NIGHTLY_BRANCH || 'main') || (inputs.branch || github.ref_name) }}

 jobs:
  build:
@@ -53,7 +43,6 @@ jobs:
        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
        with:
          persist-credentials: false
-          ref: ${{ env.NIGHTLY_BUILD_REF }}

      - name: Setup Rust environment
        uses: ./.github/actions/setup
@@ -163,104 +152,13 @@ jobs:

          fakeroot dpkg-deb --build "${PKG_DIR}"
          ls -lh "${DEB_FILE}"
-          echo "deb_date=${DEB_DATE}" >> "${GITHUB_OUTPUT}"
          echo "deb_file=${DEB_FILE}" >> "${GITHUB_OUTPUT}"

-      # Same packaging scheme as .github/workflows/package.yml (fpm), but from
-      # the locally built nightly binary instead of a release artifact, with a
-      # date-based version that mirrors the DEB.
-      - name: Build RPM package
-        id: rpm
-        shell: bash
-        env:
-          DEB_DATE: ${{ steps.deb.outputs.deb_date }}
-        run: |
-          set -euo pipefail
-
-          if ! command -v fpm >/dev/null 2>&1; then
-            SUDO=""; [ "$(id -u)" -ne 0 ] && SUDO="sudo -n"
-            ${SUDO} apt-get update -qq && ${SUDO} apt-get install -y -qq ruby ruby-dev build-essential rpm >/dev/null
-            ${SUDO} gem install fpm --no-document >/dev/null
-          fi
-
-          RPM_FILE="rustfs-nightly-${DEB_DATE}.rpm"
-          RPM_VERSION="0"
-          RPM_RELEASE="0.nightly.${DEB_DATE//-/.}"
-
-          echo "Building RPM: ${RPM_FILE} (version ${RPM_VERSION}-${RPM_RELEASE})"
-
-          # fpm wants the config file to exist before packaging.
-          mkdir -p ./tmp-pkg/etc/default
-          cat > ./tmp-pkg/etc/default/rustfs << 'ENVEOF'
-          # RustFS Environment Configuration
-          # See https://rustfs.com/docs/ for more information
-          # RUSTFS_VOLUMES=""
-          # RUSTFS_ROOT_USER=""
-          # RUSTFS_ROOT_PASSWORD=""
-          ENVEOF
-
-          fpm -s dir -t rpm \
-            --name rustfs \
-            --version "$RPM_VERSION" \
-            --iteration "$RPM_RELEASE" \
-            --architecture x86_64 \
-            --package "$RPM_FILE" \
-            --depends "glibc >= 2.31" \
-            --maintainer "RustFS Team <support@rustfs.com>" \
-            --description "High-performance distributed object storage" \
-            --url "https://rustfs.com" \
-            --license "Apache-2.0" \
-            --after-install <(cat << 'POSTINST'
-          #!/bin/bash
-          set -e
-          if ! getent passwd rustfs > /dev/null 2>&1; then
-              useradd -r -s /bin/false -d /opt/rustfs rustfs
-          fi
-          mkdir -p /opt/rustfs /data/rustfs /var/log/rustfs
-          chown rustfs:rustfs /opt/rustfs /data/rustfs /var/log/rustfs
-          if [ -d /run/systemd/system ]; then
-              systemctl daemon-reload
-          fi
-          POSTINST
-            ) \
-            --before-remove <(cat << 'PRERM'
-          #!/bin/bash
-          set -e
-          if [ -d /run/systemd/system ] && systemctl is-active --quiet rustfs; then
-              systemctl stop rustfs
-          fi
-          PRERM
-            ) \
-            --after-remove <(cat << 'POSTRM'
-          #!/bin/bash
-          set -e
-          if [ -d /run/systemd/system ]; then
-              systemctl daemon-reload
-          fi
-          POSTRM
-            ) \
-            --config-files /etc/default/rustfs \
-            "rustfs-nightly-${DEB_DATE}/usr/bin/rustfs=/usr/bin/rustfs" \
-            ./tmp-pkg/etc/default/rustfs=/etc/default/rustfs \
-            deploy/build/rustfs.service=/lib/systemd/system/rustfs.service \
-            LICENSE=/usr/share/doc/rustfs/LICENSE \
-            README.md=/usr/share/doc/rustfs/README.md
-
-          [[ -f "$RPM_FILE" ]] || { echo "RPM build failed"; exit 1; }
-          rpm -qpl "$RPM_FILE" | grep -Fx '/usr/bin/rustfs' >/dev/null
-          stat --printf='%n %s bytes\n' "$RPM_FILE"
-          echo "rpm_file=$RPM_FILE" >> "$GITHUB_OUTPUT"
-
      - name: Upload DEB artifact
        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
        with:
          name: ${{ steps.deb.outputs.deb_file }}
          path: ${{ steps.deb.outputs.deb_file }}
-      - name: Upload RPM artifact
-        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
-        with:
-          name: ${{ steps.rpm.outputs.rpm_file }}
-          path: ${{ steps.rpm.outputs.rpm_file }}
          if-no-files-found: error

      # Persist the nightly deb on Cloudflare R2 (same channel as package.yml)
@@ -289,10 +187,11 @@ jobs:
          export AWS_SECRET_ACCESS_KEY="$R2_SECRET_ACCESS_KEY"
          export AWS_DEFAULT_REGION="auto"

-          # The candidate manifest must describe the tree that was actually
-          # built. With a ref override (NIGHTLY_BRANCH / dispatch input) that
-          # is not necessarily GITHUB_SHA, so always advertise HEAD.
          SOURCE_SHA="$(git rev-parse HEAD)"
+          if [[ "${SOURCE_SHA}" != "${GITHUB_SHA}" ]]; then
+            echo "Checkout SHA does not match the nightly build run" >&2
+            exit 1
+          fi
          DEB_SHA256="$(sha256sum "${DEB_FILE}" | cut -d ' ' -f 1)"
          CANDIDATE_KEY="artifacts/rustfs/packages/nightly/runs/${GITHUB_RUN_ID}/${GITHUB_RUN_ATTEMPT}/${DEB_SHA256}/rustfs.deb"
          CANDIDATE_URL="https://dl.rustfs.com/${CANDIDATE_KEY}"
@@ -348,20 +247,6 @@ jobs:
          path: ${{ steps.publish.outputs.candidate_file }}
          if-no-files-found: error

-      # Publish the deb/rpm pair to the auto-testing repo's `assets` branch so
-      # engineers can download and install the nightly directly. The branch is
-      # a single-commit orphan rewritten on every build, which keeps the repo
-      # small while the latest files stay reachable at stable raw URLs.
-      # Publish the deb/rpm pair as assets of the rolling `nightly` release on
-      # rustfs/auto-testing (see scripts/release/publish_nightly_assets.sh).
-      - name: Publish packages to auto-testing release assets
-        env:
-          ASSETS_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
-          DEB_FILE: ${{ steps.deb.outputs.deb_file }}
-          RPM_FILE: ${{ steps.rpm.outputs.rpm_file }}
-          DEB_DATE: ${{ steps.deb.outputs.deb_date }}
-          BUILD_REF: ${{ env.NIGHTLY_BUILD_REF }}
-        run: bash scripts/release/publish_nightly_assets.sh
  # Live-Vault lane for the rustfs-kms suite (rustfs/backlog#1774).
  #
  # RUSTFS_KMS_VAULT_TOKEN is the single switch that adds the Vault KV2 and
@@ -399,7 +284,6 @@ jobs:
        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
        with:
          persist-credentials: false
-          ref: ${{ env.NIGHTLY_BUILD_REF }}

      - name: Setup Rust environment
        uses: ./.github/actions/setup
@@ -488,7 +372,6 @@ jobs:
        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
        with:
          persist-credentials: false
-          ref: ${{ env.NIGHTLY_BUILD_REF }}

      - name: Setup Rust environment
        uses: ./.github/actions/setup
@@ -132,7 +132,7 @@ jobs:
            s3api create-bucket --bucket "${RUSTFS_ODM_INTEROP_BUCKET}"

      - name: Build the RustFS binary under test
-        run: cargo build --locked -p rustfs --bins
+        run: python3 scripts/e2e_binary.py build --bins

      # The lane selects tests by module, so a rename would quietly shrink it.
      # The committed digest in .config/e2e-odm-interop-selection.txt fails
@@ -143,7 +143,7 @@ jobs:
          python3 ./scripts/check_test_wiring.py --check-profile e2e-odm-interop "${NEXTEST_LISTING}"

      - name: Run the interop cases against MinIO
-        run: cargo nextest run --profile e2e-odm-interop -p e2e_test --no-tests=fail
+        run: python3 scripts/e2e_binary.py run -- cargo nextest run --profile e2e-odm-interop -p e2e_test --no-tests=fail

      - name: Build the MinIO interop report
        if: always()
@@ -251,7 +251,7 @@ jobs:

      - name: Build the RustFS binary under test
        if: steps.credentials.outputs.present == 'true'
-        run: cargo build --locked -p rustfs --bins
+        run: python3 scripts/e2e_binary.py build --bins

      # A filterset that matches nothing is valid, so the count is asserted
      # rather than inferred from a green run.
@@ -272,7 +272,7 @@ jobs:
      - name: Run the three-case minimum
        if: steps.credentials.outputs.present == 'true'
        run: |
-          cargo nextest run --profile e2e-odm-interop -p e2e_test \
+          python3 scripts/e2e_binary.py run -- cargo nextest run --profile e2e-odm-interop -p e2e_test \
            -E "${CLOUD_CASE_FILTER}" --no-tests=fail

      - name: Build the ${{ matrix.provider }} interop report
@@ -211,10 +211,7 @@ For developers who want to build RustFS Docker images from source with multi-arc

 ```bash
 # Build multi-architecture images locally
-./docker-buildx.sh
-
-# Build a single-platform image locally
-./docker-buildx.sh -p linux/amd64
+./docker-buildx.sh --build-arg RELEASE=latest

 # Build and push to registry
 ./docker-buildx.sh --push
@@ -150,10 +150,7 @@ docker compose -f docker-compose-simple.yml up -d

 ```bash
 # 在本地构建多架构镜像
-./docker-buildx.sh
-
-# 在本地构建单平台镜像
-./docker-buildx.sh -p linux/amd64
+./docker-buildx.sh --build-arg RELEASE=latest

 # 构建并推送到仓库
 ./docker-buildx.sh --push
@@ -1,7 +1,7 @@
 # e2e_test

 End-to-end test suite for RustFS. Each test spawns a **real `rustfs` binary**
-(built on demand from the workspace) and drives it over the network with the
+(built and identified before the test invocation) and drives it over the network with the
 AWS SDK (`aws-sdk-s3`), raw HTTP (`reqwest` / `awscurl`), or a protocol client
 (FTPS / WebDAV / SFTP). This is the black-box integration layer: exhaustive
 end-to-end behavior lives here, unit behavior stays in the source crates
@@ -32,32 +32,28 @@ Registered in [`src/lib.rs`](src/lib.rs). Grouped by concern:

 ## How to run

-All commands assume repo root. `cargo test` triggers an on-demand build of the
-`rustfs` binary from [`src/common.rs`](src/common.rs) (`rustfs_binary_path`) on
-first use — the first invocation is slow, later ones reuse the binary.
+All commands assume repo root and Python 3.9 or newer on Linux or macOS. Build the server once through the provenance entry point, then run the test command through the same script:

 ```bash
-# Whole crate (default = ignored tests skipped)
-cargo nextest run -p e2e_test
+python3 scripts/e2e_binary.py build --features e2e-test-hooks
+
+# Whole crate (ignored tests remain skipped)
+python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run -p e2e_test

 # One module
-cargo nextest run -p e2e_test -E 'test(list_objects_v2_pagination_test)'
-
-# PR smoke subset (see "CI smoke subset" below)
-cargo nextest run --profile e2e-smoke -p e2e_test
-
-# ILM serial lane — ignored lifecycle tests, single-threaded (mirrors CI)
-cargo nextest run -j1 --run-ignored ignored-only -p rustfs-scanner -p rustfs \
-  -E 'binary(lifecycle_integration_test) or (package(rustfs) and test(lifecycle_transition_api_test))'
+python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run -p e2e_test -E 'test(list_objects_v2_pagination_test)'

+# PR smoke subset
+python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-smoke -p e2e_test
 ```

-The protocols suite has its own contract (fixed bind ports 9022–9301,
-single-worker execution, feature-gated scheduling) documented in
-[`src/protocols/README.md`](src/protocols/README.md). `RUSTFS_BUILD_FEATURES`
-selects which features the spawned binary is built with; leave it unset to run
-every protocol entry. Use the exact profile command under
-[Troubleshooting](#troubleshooting) for CI-equivalent execution.
+`build` records the source contents, HEAD, resolved Cargo features, profile, toolchain, and binary SHA-256 beside the executable in `rustfs.e2e.json`. `run` validates that identity before and after the command, preserves command failures, and removes its temporary run receipt on completion. The Rust harness checks that receipt before starting each server; it never compiles a server inside a test process. Source or binary changes during a run invalidate the result, even when the test command succeeds. Use an isolated worktree and keep it unchanged until the command finishes.
+
+The additional `--features` arguments must match between `build` and `run`; Cargo defaults remain enabled. The wrapper supplies `RUSTFS_BUILD_FEATURES` from Cargo's resolved feature list, including features enabled by `full`. Protocol helpers require a subset of that list. `CARGO_TARGET_DIR` and `--profile release` are supported. An in-workspace target directory must be Git-ignored; tracked files are always included in the source identity. `build --bins` preserves CI lanes that compile all RustFS binary targets. For a downloaded artifact, copy both the executable and its sidecar, then use `run`; do not generate a new identity for an arbitrary prebuilt binary. `CARGO_BIN_EXE_rustfs` cannot override the verified executable.
+
+Each build/run holds an exclusive `rustfs.e2e.lock` marker beside the binary; concurrent wrappers fail immediately. Use a private target directory and do not run ordinary Cargo builds against it while tests are active: Cargo does not honor this marker. Interrupted runs fail and terminate their command group. After an uncatchable kill, inspect the PID recorded in a leftover marker and remove it only after confirming its owner has stopped. Embedded file symlinks are hashed through their target; embedded directory symlinks are rejected because their contents cannot be enumerated safely by this entry point.
+
+The protocols suite has its own fixed-port and single-worker contract in [`src/protocols/README.md`](src/protocols/README.md). Use its command under [Troubleshooting](#troubleshooting).

 ### `#[ignore]` semantics

@@ -123,7 +119,7 @@ via `create_s3_client(idx)` / `create_all_clients()`. See
 | `wait_for_server_ready` | Poll readiness before issuing requests |
 | `create_s3_client` / `create_test_bucket` / `delete_test_bucket` | aws-sdk-s3 client + bucket lifecycle |
 | `find_available_port` | Random free port (isolation primitive) |
-| `rustfs_binary_path` / `_with_features` | Locate/build the binary; honors `RUSTFS_BUILD_FEATURES` |
+| `rustfs_binary_path` / `_with_features` | Verify this run's binary receipt and required feature subset |
 | `requested_rustfs_build_features` / `rustfs_build_feature_enabled` | Feature-gate a test to what the binary was built with |
 | `execute_awscurl` / `awscurl_post` / `_get` / `_put` / `_delete` / `awscurl_post_sts_form_urlencoded` | Admin/STS API calls via `awscurl`; missing binaries are test failures |
 | `replication_fast_env` | Env vars that shrink replication timers (from repl-4); pass to `start_rustfs_server_with_env` |
@@ -189,35 +185,33 @@ the wiring source of truth. Committed test-ID digests under
 **Reproduce a CI failure locally** — run the exact profile/lane:

 ```bash
-# Smoke (e2e-tests job) — includes the 20 fast replication tests
-cargo nextest run --profile e2e-smoke -p e2e_test
-# Full single-node merge/main lane
-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
-RUSTFS_BUILD_FEATURES=ftps,webdav,sftp \
-  cargo nextest run -j 1 --profile e2e-protocols -p e2e_test --no-capture
-# ILM serial lane
+# Smoke, full, and cluster lanes share a server with fault-test hooks.
+python3 scripts/e2e_binary.py build --features e2e-test-hooks
+python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-smoke -p e2e_test
+python3 scripts/e2e_binary.py run --binary "$RUSTFS_E2E_STARTUP_CAS_BINARY" --features e2e-test-hooks -- cargo nextest run --profile e2e-full -p e2e_test
+python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-nightly -p e2e_test
+
+# Distributed 4-node 4-disk lane uses the default server.
+# Upgrade cases require RUSTFS_UPGRADE_SOURCE_BINARY and fail closed without it.
+python3 scripts/e2e_binary.py build
+python3 scripts/e2e_binary.py run -- cargo nextest run --profile e2e-distributed -p e2e_test
+
+# Replication nightly uses the default server; awscurl is required for STS.
+python3 scripts/e2e_binary.py build
+python3 scripts/e2e_binary.py run -- cargo nextest run --profile e2e-repl-nightly -p e2e_test
+
+# Protocol nightly owns fixed ports.
+python3 scripts/e2e_binary.py build --features ftps,webdav,sftp
+python3 scripts/e2e_binary.py run --features ftps,webdav,sftp -- cargo nextest run -j 1 --profile e2e-protocols -p e2e_test --no-capture
+
+# The ILM serial lane does not use this server harness.
 cargo nextest run -j1 --run-ignored ignored-only -p rustfs-scanner -p rustfs \
  -E 'binary(lifecycle_integration_test) or (package(rustfs) and test(lifecycle_transition_api_test))'
-# s3s-e2e black box
-./scripts/e2e-run.sh ./target/debug/rustfs /tmp/rustfs-e2e-data
 ```

-**Stale binary.** Tests build the `rustfs` binary once and reuse it. To avoid
-rebuilding while iterating on tests, `common.rs` reuses an existing binary when
-running *inside* the e2e test process even if sources changed
-(`can_reuse_inside_e2e`, [`src/common.rs`](src/common.rs) line 98). Downside: if
-you changed **server** code, force a rebuild with
-`cargo build -p rustfs` (or `touch` a source file outside the reuse window)
-before re-running, or CI's freshly built artifact will diverge from your local
-one.
+The full lane also requires the startup-CAS build manifest generated by the `Build debug binary` step in `.github/workflows/ci.yml`. Preserve that binary and both sidecars as its `Preserve startup CAS binary input` step does, and use the same `RUSTFS_E2E_STARTUP_CAS_*` environment as `Run e2e full suite`. A generic local build alone does not supply that fixture evidence.
+
+**Stale or unverified binary.** Re-run the matching `build` command after changing source or features, then invoke tests through `run`. A missing receipt, copied old executable, or mismatched build identity is a prerequisite failure. Bare Cargo invocations that start a server deliberately fail; unit tests that do not start a server can still run directly.

 **Port already in use / orphan processes.** A hard-killed run can leak a
 `rustfs` child holding its port. Find and kill it:
@@ -249,7 +243,8 @@ spawn error. Install the pinned CI version before running their profiles.
 A subset of this crate runs on every PR via the `e2e-tests` job:

 ```bash
-cargo nextest run --profile e2e-smoke -p e2e_test
+python3 scripts/e2e_binary.py build --features e2e-test-hooks
+python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-smoke -p e2e_test
 ```

 The selection lives in `.config/nextest.toml` under `[profile.e2e-smoke]`
@@ -31,7 +31,6 @@ use rustfs_signer::constants::UNSIGNED_PAYLOAD;
 use rustfs_signer::sign_v4;
 use s3s::Body;
 use serde_json;
-use std::ffi::OsStr;
 use std::fs as stdfs;
 use std::io::ErrorKind;
 use std::net::SocketAddr;
@@ -44,7 +43,6 @@ use tokio::net::TcpStream;
 use tokio::time::sleep;
 use tracing::{error, info, warn};
 use uuid::Uuid;
-use walkdir::WalkDir;

 // Common constants for all E2E tests
 pub const DEFAULT_ACCESS_KEY: &str = "rustfsadmin";
@@ -428,59 +426,75 @@ fn resolve_rustfs_binary_path(workspace: &Path, configured_target_dir: Option<&P
    path
 }

-/// Resolve the RustFS binary relative to the workspace, optionally requesting build features.
+/// Resolve the server verified by `scripts/e2e_binary.py run` for this test invocation.
+/// Requested features are a required subset of the server's resolved Cargo features.
 pub fn rustfs_binary_path_with_features(requested_features: Option<&str>) -> PathBuf {
-    if let Some(path) = std::env::var_os("CARGO_BIN_EXE_rustfs") {
-        return PathBuf::from(path);
-    }
-    let requested_features = requested_features.and_then(normalize_rustfs_build_features);
-
    let workspace = workspace_root();
    let configured_target_dir = std::env::var_os("CARGO_TARGET_DIR").map(PathBuf::from);
-    let binary_path = resolve_rustfs_binary_path(&workspace, configured_target_dir.as_deref());
-
-    let features_match = binary_features_match(&binary_path, requested_features.as_deref());
-    let source_is_newer = workspace_sources_newer_than_binary(&binary_path);
-    let can_reuse_inside_e2e = running_inside_e2e_test_binary() && requested_features.is_none() && features_match;
-    if binary_path.is_file() && features_match && (!source_is_newer || can_reuse_inside_e2e) {
-        if source_is_newer {
-            warn!(
-                "RustFS binary at {:?} appears older than workspace sources; reusing it inside cargo test to avoid nested builds",
-                binary_path
-            );
-        }
-        info!("Using existing RustFS binary at {:?}", binary_path);
-        return binary_path;
+    let binary_path = std::env::var_os("CARGO_BIN_EXE_rustfs")
+        .map(PathBuf::from)
+        .unwrap_or_else(|| resolve_rustfs_binary_path(&workspace, configured_target_dir.as_deref()));
+    let receipt_path = std::env::var_os("RUSTFS_E2E_BINARY_RECEIPT").map(PathBuf::from);
+    receipt_path
+        .ok_or_else(|| std::io::Error::new(ErrorKind::NotFound, "missing E2E run receipt"))
+        .and_then(|receipt| verify_e2e_binary_receipt(&receipt, &workspace, &binary_path, requested_features))
+        .unwrap_or_else(|error| {
+            panic!(
+                "E2E server prerequisite failed: {error}. Build with `python3 scripts/e2e_binary.py build --features <features>` and run tests with `python3 scripts/e2e_binary.py run --features <features> -- cargo nextest run ...`"
+            )
+        })
 }

-    info!("Building RustFS binary to ensure it's up to date...");
-    build_rustfs_binary(requested_features.as_deref(), &binary_path);
-
-    info!("Using RustFS binary at {:?}", binary_path);
-    binary_path
+#[derive(serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+struct E2eBinaryReceipt {
+    schema: u32,
+    workspace: PathBuf,
+    binary: PathBuf,
+    size: u64,
+    modified_ns: u128,
+    features: Vec<String>,
 }

-fn workspace_sources_newer_than_binary(binary_path: &PathBuf) -> bool {
-    let Ok(binary_meta) = std::fs::metadata(binary_path) else {
-        return true;
-    };
-    let Ok(binary_modified) = binary_meta.modified() else {
-        return true;
-    };
-
-    let workspace = workspace_root();
-    let watch_roots = [
-        workspace.join("Cargo.toml"),
-        workspace.join("Cargo.lock"),
-        workspace.join("rustfs"),
-        workspace.join("crates"),
-    ];
-
-    watch_roots.iter().any(|path| path_is_newer_than(binary_modified, path))
+fn verify_e2e_binary_receipt(
+    receipt_path: &Path,
+    workspace: &Path,
+    binary_path: &Path,
+    requested_features: Option<&str>,
+) -> std::io::Result<PathBuf> {
+    let receipt: E2eBinaryReceipt = serde_json::from_slice(&stdfs::read(receipt_path)?)?;
+    let binary = binary_path.canonicalize()?;
+    let metadata = binary.metadata()?;
+    let modified_ns = metadata
+        .modified()?
+        .duration_since(std::time::UNIX_EPOCH)
+        .map_err(std::io::Error::other)?
+        .as_nanos();
+    // The runner hashes source and binary before/after the entire suite. Each
+    // nextest process checks only this invocation's path, features, and file stat.
+    if receipt.schema != 1
+        || receipt.workspace != workspace.canonicalize()?
+        || receipt.binary != binary
+        || !metadata.is_file()
+        || receipt.size != metadata.len()
+        || receipt.modified_ns != modified_ns
+    {
+        return Err(std::io::Error::new(
+            ErrorKind::InvalidData,
+            "E2E server differs from this run's verified binary",
+        ));
    }
-
-fn running_inside_e2e_test_binary() -> bool {
-    std::env::var("CARGO_PKG_NAME").is_ok_and(|value| value == "e2e_test")
+    if let Some(requested) = requested_features.and_then(normalize_rustfs_build_features)
+        && requested
+            .split(',')
+            .any(|feature| !receipt.features.iter().any(|actual| actual == feature))
+    {
+        return Err(std::io::Error::new(
+            ErrorKind::InvalidInput,
+            "E2E server is missing a requested build feature",
+        ));
+    }
+    Ok(binary)
 }

 pub fn requested_rustfs_build_features() -> Option<String> {
@@ -510,96 +524,6 @@ pub fn rustfs_build_feature_enabled(requested_features: Option<&str>, required_f
        .any(|feature| feature.eq_ignore_ascii_case(RUSTFS_FULL_FEATURE) || feature.eq_ignore_ascii_case(required_feature))
 }

-fn rustfs_binary_features_stamp_path(binary_path: &Path) -> PathBuf {
-    binary_path.with_extension("features")
-}
-
-fn binary_features_match(binary_path: &Path, requested_features: Option<&str>) -> bool {
-    let stamp_path = rustfs_binary_features_stamp_path(binary_path);
-    let recorded = stdfs::read_to_string(stamp_path)
-        .ok()
-        .and_then(|value| normalize_rustfs_build_features(&value));
-    let requested = requested_features.and_then(normalize_rustfs_build_features);
-
-    match requested.as_deref() {
-        Some(features) => recorded.as_deref() == Some(features),
-        None => recorded.is_none(),
-    }
-}
-
-fn path_is_newer_than(binary_modified: std::time::SystemTime, path: &Path) -> bool {
-    if path.is_file() {
-        return std::fs::metadata(path)
-            .and_then(|meta| meta.modified())
-            .map(|modified| modified > binary_modified)
-            .unwrap_or(false);
-    }
-
-    if !path.is_dir() {
-        return false;
-    }
-
-    WalkDir::new(path)
-        .into_iter()
-        .filter_entry(|entry| {
-            let name = entry.file_name();
-            name != OsStr::new("target") && name != OsStr::new(".git")
-        })
-        .filter_map(Result::ok)
-        .filter(|entry| entry.file_type().is_file())
-        .any(|entry| {
-            std::fs::metadata(entry.path())
-                .and_then(|meta| meta.modified())
-                .map(|modified| modified > binary_modified)
-                .unwrap_or(false)
-        })
-}
-
-/// Build the RustFS binary using cargo
-fn build_rustfs_binary(requested_features: Option<&str>, binary_path: &Path) {
-    let workspace = workspace_root();
-    info!("Building RustFS binary from workspace: {:?}", workspace);
-
-    let _profile = if cfg!(debug_assertions) {
-        info!("Building in debug mode");
-        "dev"
-    } else {
-        info!("Building in release mode");
-        "release"
-    };
-
-    let mut cmd = Command::new("cargo");
-    cmd.current_dir(&workspace).args(["build", "--bin", "rustfs"]);
-
-    if let Some(features) = requested_features {
-        cmd.arg("--features").arg(features);
-        info!("Building with features: {}", features);
-    }
-
-    if !cfg!(debug_assertions) {
-        cmd.arg("--release");
-    }
-
-    info!(
-        "Executing: cargo build --bin rustfs {}",
-        if cfg!(debug_assertions) { "" } else { "--release" }
-    );
-
-    let output = cmd.output().expect("Failed to execute cargo build command");
-
-    if !output.status.success() {
-        let stderr = String::from_utf8_lossy(&output.stderr);
-        panic!("Failed to build RustFS binary. Error: {stderr}");
-    }
-
-    let stamp_path = rustfs_binary_features_stamp_path(binary_path);
-    if let Err(err) = stdfs::write(&stamp_path, requested_features.unwrap_or_default()) {
-        warn!("Failed to write RustFS feature stamp {:?}: {}", stamp_path, err);
-    }
-
-    info!("✅ RustFS binary built successfully");
-}
-
 fn awscurl_binary_path() -> PathBuf {
    std::env::var_os("AWSCURL_PATH")
        .map(PathBuf::from)
@@ -2229,16 +2153,66 @@ mod tests {
    }

    #[test]
-    fn binary_feature_stamp_matching_uses_normalized_features() {
-        let binary_path = std::env::temp_dir().join(format!("rustfs-feature-stamp-test-{}", Uuid::new_v4()));
-        let stamp_path = rustfs_binary_features_stamp_path(&binary_path);
+    fn explicit_binary_without_run_receipt_is_rejected() {
+        const CHILD_ENV: &str = "RUSTFS_E2E_RECEIPT_TEST_CHILD";
+        if std::env::var_os(CHILD_ENV).is_some() {
+            rustfs_binary_path_with_features(None);
+            return;
+        }
+        let executable = std::env::current_exe().expect("locate isolated test process");
+        let output = Command::new(&executable)
+            .args([
+                "--exact",
+                "common::tests::explicit_binary_without_run_receipt_is_rejected",
+                "--nocapture",
+            ])
+            .env(CHILD_ENV, "1")
+            .env("CARGO_BIN_EXE_rustfs", &executable)
+            .env_remove("RUSTFS_E2E_BINARY_RECEIPT")
+            .output()
+            .expect("run the missing-receipt scenario with isolated environment variables");
+        assert!(!output.status.success(), "an explicit binary must not bypass run verification");
+        assert!(String::from_utf8_lossy(&output.stderr).contains("missing E2E run receipt"));
+    }

-        stdfs::write(&stamp_path, " SFTP, ftps ").expect("write feature stamp");
-        assert!(binary_features_match(&binary_path, Some("sftp,ftps")));
-        assert!(binary_features_match(&binary_path, Some(" SFTP, FTPS ")));
-        assert!(!binary_features_match(&binary_path, Some("sftp")));
-
-        stdfs::remove_file(stamp_path).ok();
+    #[test]
+    fn e2e_run_receipt_rejects_replaced_binary_and_missing_features() {
+        let directory = std::env::temp_dir().join(format!("rustfs-e2e-receipt-test-{}", Uuid::new_v4()));
+        stdfs::create_dir(&directory).expect("create receipt fixture");
+        let binary = directory.join("rustfs");
+        let receipt = directory.join("receipt.json");
+        stdfs::write(&binary, "server").expect("write fixture binary");
+        let metadata = binary.metadata().expect("stat fixture binary");
+        let record = serde_json::json!({
+            "schema": 1,
+            "workspace": directory.canonicalize().expect("canonical workspace"),
+            "binary": binary.canonicalize().expect("canonical binary"),
+            "size": metadata.len(),
+            "modified_ns": metadata.modified().expect("modified time").duration_since(std::time::UNIX_EPOCH).expect("positive timestamp").as_nanos(),
+            "features": ["default", "full", "ftps", "webdav", "sftp"]
+        });
+        stdfs::write(&receipt, serde_json::to_vec(&record).expect("serialize receipt")).expect("write receipt");
+        verify_e2e_binary_receipt(&receipt, &directory, &binary, Some("sftp,webdav")).expect("resolved feature subset");
+        verify_e2e_binary_receipt(&receipt, &directory, &binary, Some("full")).expect("full was actually requested");
+        assert_eq!(
+            verify_e2e_binary_receipt(&receipt, &directory, &binary, Some("rio-v2"))
+                .expect_err("full does not enable rio-v2")
+                .kind(),
+            ErrorKind::InvalidInput
+        );
+        let other = directory.join("old-server");
+        stdfs::write(&other, "server").expect("write alternate binary");
+        assert!(verify_e2e_binary_receipt(&receipt, &directory, &other, None).is_err());
+        stdfs::write(&binary, "different server").expect("replace fixture binary");
+        assert!(verify_e2e_binary_receipt(&receipt, &directory, &binary, None).is_err());
+        stdfs::remove_file(&receipt).expect("remove expired receipt");
+        assert_eq!(
+            verify_e2e_binary_receipt(&receipt, &directory, &binary, None)
+                .expect_err("expired receipt")
+                .kind(),
+            ErrorKind::NotFound
+        );
+        stdfs::remove_dir_all(directory).expect("remove receipt fixture");
    }

    /// Build a cluster environment struct in-memory (no ports, no processes) so
@@ -1047,8 +1047,6 @@ mod tests {
        let mut cluster = RustFSTestClusterEnvironment::new(4).await?;
        cluster.set_env("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", "true");
        cluster.set_env("RUSTFS_HEAL_ENABLED", "true");
-        // Capture physical baselines after the PUT rename fanout has drained.
-        cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false");
        // Heal control uses the first lexicographically sorted grid host.
        // Keep that coordinator distinct from the remote target at index 1.
        cluster.nodes.sort_by(|left, right| left.url.cmp(&right.url));
@@ -25,190 +25,6 @@ const KEY: &str = "thumb/79/concurrent-overwrite.jpg";

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

-async fn assert_quorum_object_body(client: &Client, bucket: &str, key: &str, expected: &[u8]) -> TestResult {
-    let body = client
-        .get_object()
-        .bucket(bucket)
-        .key(key)
-        .send()
-        .await?
-        .body
-        .collect()
-        .await?
-        .into_bytes();
-    assert_eq!(body.as_ref(), expected, "quorum read returned incorrect contents for {key}");
-    Ok(())
-}
-
-async fn wait_for_quorum_read_admission(clients: &[Client], bucket: &str) -> TestResult {
-    // SIGKILL can orphan a granted lease. Wait for shared metadata-lock
-    // admission before asserting the stable quorum boundary; cold bodies
-    // remain unread throughout this readiness probe.
-    let deadline =
-        tokio::time::Instant::now() + rustfs_lock::fast_lock::DEFAULT_LOCK_TIMEOUT + std::time::Duration::from_secs(15);
-    loop {
-        let mut ready = true;
-        for client in clients {
-            for key in ["warm-small", "warm-large"] {
-                match client.head_object().bucket(bucket).key(key).send().await {
-                    Ok(_) => {}
-                    Err(error) if error.raw_response().is_some_and(|response| response.status().as_u16() == 503) => {
-                        ready = false;
-                        break;
-                    }
-                    Err(error) => return Err(error.into()),
-                }
-            }
-            if !ready {
-                break;
-            }
-        }
-        if ready {
-            return Ok(());
-        }
-        if tokio::time::Instant::now() >= deadline {
-            return Err(format!("read quorum did not become available after lease convergence for {bucket}").into());
-        }
-        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
-    }
-}
-
-#[tokio::test]
-async fn test_degraded_cluster_read_quorum_follows_erasure_layout() -> TestResult {
-    crate::common::init_logging();
-
-    for (node_count, parity) in [(4, 2), (6, 3), (6, 2)] {
-        let read_quorum = node_count - parity;
-        let write_quorum = read_quorum + usize::from(read_quorum == parity);
-        let mut cluster = RustFSTestClusterEnvironment::new(node_count).await?;
-        cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", format!("EC:{parity}"));
-        // Wait for every seed fanout before removing any physical shard.
-        cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false");
-        cluster.set_env("RUSTFS_OBS_METRICS_EXPORT_ENABLED", "false");
-        cluster.set_env("RUST_LOG", "warn,rustfs_lock=debug");
-        cluster.start().await?;
-
-        let clients = cluster
-            .create_all_clients()?
-            .into_iter()
-            .map(|client| {
-                Client::from_conf(
-                    client
-                        .config()
-                        .to_builder()
-                        .retry_config(aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(1))
-                        .build(),
-                )
-            })
-            .collect::<Vec<_>>();
-        let bucket = format!("read-quorum-{node_count}-{parity}");
-        clients[0].create_bucket().bucket(&bucket).send().await?;
-        let small = b"read quorum is derived from the erasure layout".to_vec();
-        let large = (0..1_048_576)
-            .map(|index| u8::try_from(index % 251).expect("bounded payload byte"))
-            .collect::<Vec<_>>();
-        for (key, body) in [
-            ("warm-small", &small),
-            ("warm-large", &large),
-            ("cold-small", &small),
-            ("cold-large", &large),
-            ("below-quorum", &large),
-        ] {
-            clients[node_count - 1]
-                .put_object()
-                .bucket(&bucket)
-                .key(key)
-                .body(Bytes::copy_from_slice(body).into())
-                .send()
-                .await?;
-        }
-        for node in &cluster.nodes {
-            for key in ["warm-small", "warm-large", "cold-small", "cold-large", "below-quorum"] {
-                let census =
-                    crate::chaos::census_object_version_on_disk(std::path::Path::new(&node.data_dir), &bucket, key, None)?;
-                assert!(census.is_complete(), "seed shard must be complete before fault injection: {census:?}");
-                assert_eq!(census.data_blocks, Some(read_quorum));
-                assert_eq!(census.parity_blocks, Some(parity));
-            }
-        }
-        for client in &clients {
-            assert_quorum_object_body(client, &bucket, "warm-small", &small).await?;
-            assert_quorum_object_body(client, &bucket, "warm-large", &large).await?;
-        }
-
-        for offline_node in (read_quorum..node_count).rev() {
-            cluster.stop_node(offline_node)?;
-            wait_for_quorum_read_admission(&clients[..offline_node], &bucket).await?;
-            for client in clients.iter().take(offline_node) {
-                client.head_bucket().bucket(&bucket).send().await?;
-                assert_quorum_object_body(client, &bucket, "warm-large", &large).await?;
-            }
-        }
-
-        // Exercise more than the five-second positive bucket-validation TTL.
-        // Every sample must succeed; polling must not hide a transient failure.
-        let validation_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(6);
-        loop {
-            for client in clients.iter().take(read_quorum) {
-                assert_quorum_object_body(client, &bucket, "warm-small", &small).await?;
-                assert_quorum_object_body(client, &bucket, "warm-large", &large).await?;
-                let listing = client.list_objects_v2().bucket(&bucket).send().await?;
-                for key in ["warm-small", "warm-large", "cold-small", "cold-large", "below-quorum"] {
-                    assert!(listing.contents().iter().any(|entry| entry.key() == Some(key)), "listing omitted {key}");
-                }
-            }
-            if tokio::time::Instant::now() >= validation_deadline {
-                break;
-            }
-            tokio::time::sleep(std::time::Duration::from_millis(250)).await;
-        }
-        for client in clients.iter().take(read_quorum) {
-            assert_quorum_object_body(client, &bucket, "cold-small", &small).await?;
-            assert_quorum_object_body(client, &bucket, "cold-large", &large).await?;
-        }
-
-        let write = clients[0]
-            .put_object()
-            .bucket(&bucket)
-            .key("quorum-write")
-            .body(Bytes::copy_from_slice(&small).into())
-            .send()
-            .await;
-        if read_quorum >= write_quorum {
-            write?;
-        } else {
-            let error = write.expect_err("a read quorum must not authorize a write that needs more votes");
-            assert_eq!(error.as_service_error().and_then(|error| error.meta().code()), Some("ServiceUnavailable"));
-        }
-
-        cluster.stop_node(read_quorum - 1)?;
-        for client in clients.iter().take(read_quorum - 1) {
-            match client.get_object().bucket(&bucket).key("below-quorum").send().await {
-                Ok(response) => assert!(
-                    response.body.collect().await.is_err(),
-                    "fewer than {read_quorum} valid fragments must not reconstruct an uncached object"
-                ),
-                Err(error) => assert_eq!(
-                    error.as_service_error().and_then(|error| error.meta().code()),
-                    Some("ServiceUnavailable"),
-                    "a quorum loss must not be mistaken for a missing object"
-                ),
-            }
-        }
-
-        for node in 0..read_quorum - 1 {
-            cluster.stop_node(node)?;
-        }
-        cluster.start().await?;
-        for client in &clients {
-            assert_quorum_object_body(client, &bucket, "warm-large", &large).await?;
-            assert_quorum_object_body(client, &bucket, "below-quorum", &large).await?;
-        }
-    }
-
-    Ok(())
-}
-
 async fn put_object(client: Client, payload: Vec<u8>, writer_id: usize) -> Result<(), String> {
    client
        .put_object()
@@ -17,15 +17,13 @@ Use the canonical CI-equivalent protocol command in the parent
 For targeted debugging of the core suite only:

 ```bash
-RUSTFS_BUILD_FEATURES=ftps,webdav,sftp cargo test --package e2e_test test_protocol_core_suite -- --test-threads=1 --nocapture
+python3 scripts/e2e_binary.py build --features ftps,webdav,sftp
+python3 scripts/e2e_binary.py run --features ftps,webdav,sftp -- cargo test --package e2e_test test_protocol_core_suite -- --test-threads=1 --nocapture
 ```

 This targeted command does not cover the full `e2e-protocols` profile.

-`RUSTFS_BUILD_FEATURES` controls which features the test rustfs binary is
-built with. When this variable is set, the protocol test runner schedules
-only entries whose protocol is present in the requested feature list. Leave
-it unset to run every protocol entry.
+`e2e_binary.py` supplies `RUSTFS_BUILD_FEATURES` from the verified server's resolved Cargo features. The protocol runner schedules only entries present in that feature list; helpers check that their required features are available without rebuilding the server.
 `--test-threads=1` is required because every entry spawns a rustfs server
 on fixed bind ports.

@@ -3463,11 +3463,6 @@ impl PoolRebalanceActivationFence {
    }
 }

-#[cfg(test)]
-tokio::task_local! {
-    pub(crate) static REBALANCE_ACTIVATION_LOCK_ATTEMPT: Arc<tokio::sync::Notify>;
-}
-
 pub(crate) async fn acquire_pool_rebalance_activation_locks<S>(
    pool: Arc<S>,
    fleet_proof: Option<crate::services::notification_sys::CrossPoolFenceFleetProofToken>,
@@ -3478,21 +3473,17 @@ where
            NamespaceLock = rustfs_lock::NamespaceLockWrapper,
        >,
 {
-    // Match entry admission: rebalance.bin -> pool.bin. An entry retains its
-    // run read fence while target mutations acquire the pool metadata fence;
-    // activation must not hold pool.bin while waiting for that entry to drain.
-    let rebalance_meta_lock = pool.new_ns_lock(RUSTFS_META_BUCKET, REBAL_META_NAME).await?;
-    #[cfg(test)]
-    let _ = REBALANCE_ACTIVATION_LOCK_ATTEMPT.try_with(|attempted| attempted.notify_one());
-    let rebalance_meta_guard = rebalance_meta_lock
-        .get_write_lock(get_lock_acquire_timeout())
-        .await
-        .map_err(activation_rebalance_meta_lock_error)?;
+    // Activation lock order is always pool.bin -> rebalance.bin.
    let pool_meta_lock = pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?;
    let pool_meta_guard = pool_meta_lock
        .get_write_lock(get_lock_acquire_timeout())
        .await
        .map_err(activation_pool_meta_lock_error)?;
+    let rebalance_meta_lock = pool.new_ns_lock(RUSTFS_META_BUCKET, REBAL_META_NAME).await?;
+    let rebalance_meta_guard = rebalance_meta_lock
+        .get_write_lock(get_lock_acquire_timeout())
+        .await
+        .map_err(activation_rebalance_meta_lock_error)?;

    Ok(PoolRebalanceActivationFence {
        pool_meta_guard,
@@ -22103,7 +22094,7 @@ mod pools_tests {
                .resources
                .lock()
                .expect("activation lock recorder should not be poisoned"),
-            vec![REBAL_META_NAME.to_string(), POOL_META_NAME.to_string()]
+            vec![POOL_META_NAME.to_string(), REBAL_META_NAME.to_string()]
        );

        let mut second_acquire = Box::pin(acquire_pool_rebalance_activation_locks(second.clone(), None));
@@ -22119,50 +22110,10 @@ mod pools_tests {
                .resources
                .lock()
                .expect("activation lock recorder should not be poisoned"),
-            vec![REBAL_META_NAME.to_string(), POOL_META_NAME.to_string()]
+            vec![POOL_META_NAME.to_string(), REBAL_META_NAME.to_string()]
        );
    }

-    #[tokio::test]
-    async fn test_activation_cancellation_releases_rebalance_fence_while_pool_fence_is_contended() {
-        use crate::storage_api_contracts::namespace::NamespaceLocking as _;
-        let pool = Arc::new(ActivationLockRecorder {
-            lock_manager: Arc::new(rustfs_lock::GlobalLockManager::new()),
-            owner: "activation-cancellation",
-            resources: StdMutex::new(Vec::new()),
-        });
-        let pool_lock = pool
-            .new_ns_lock(crate::disk::RUSTFS_META_BUCKET, POOL_META_NAME)
-            .await
-            .expect("pool lock should be created");
-        let pool_reader = pool_lock
-            .get_read_lock(std::time::Duration::from_secs(5))
-            .await
-            .expect("ordinary mutation should hold the pool read fence");
-        pool.resources.lock().expect("recorder should not be poisoned").clear();
-        let mut activation = Box::pin(acquire_pool_rebalance_activation_locks(Arc::clone(&pool), None));
-        assert!(matches!(futures::poll!(&mut activation), Poll::Pending));
-        assert_eq!(
-            *pool.resources.lock().expect("recorder should not be poisoned"),
-            vec![REBAL_META_NAME.to_string(), POOL_META_NAME.to_string()],
-            "activation must hold the run fence before waiting for the pool fence",
-        );
-        drop(activation);
-        let rebalance_lock = pool
-            .new_ns_lock(crate::disk::RUSTFS_META_BUCKET, REBAL_META_NAME)
-            .await
-            .expect("run lock should be created");
-        let run_writer = rebalance_lock
-            .get_write_lock(std::time::Duration::from_secs(5))
-            .await
-            .expect("cancelling activation must release its already-acquired run fence");
-        assert!(
-            !pool_reader.is_released(),
-            "cancelling activation must not release another caller's pool fence"
-        );
-        assert!(!run_writer.is_lock_lost());
-    }
-
    #[test]
    fn decommission_receipt_run_token_changes_with_persisted_start_time() {
        let first = OffsetDateTime::from_unix_timestamp(1_000).expect("first run timestamp should be valid");
@@ -627,13 +627,7 @@ impl From<tokio::task::JoinError> for DiskError {
 impl Clone for DiskError {
    fn clone(&self) -> Self {
        match self {
-            DiskError::Io(io_error) => DiskError::Io(
-                rustfs_rio::clone_internode_http_io_error(io_error)
-                    .and_then(std::io::Error::into_inner)
-                    // The helper derives a kind from the source; Clone must retain the original outer kind.
-                    .map(|source| std::io::Error::new(io_error.kind(), source))
-                    .unwrap_or_else(|| std::io::Error::new(io_error.kind(), io_error.to_string())),
-            ),
+            DiskError::Io(io_error) => DiskError::Io(std::io::Error::new(io_error.kind(), io_error.to_string())),
            DiskError::MaxVersionsExceeded => DiskError::MaxVersionsExceeded,
            DiskError::Unexpected => DiskError::Unexpected,
            DiskError::CorruptedFormat => DiskError::CorruptedFormat,
@@ -1271,49 +1265,6 @@ mod tests {
        assert!(!bad_request.is_retryable_internode_write_failure());
    }

-    #[test]
-    fn test_internode_http_clone_preserves_retryability_status_and_context() {
-        use http::StatusCode;
-        use rustfs_rio::InternodeHttpErrorKind::{ConnectionRefused, ConnectionReset, HttpStatus, Unknown};
-
-        for (kind, retryable) in [
-            (ConnectionRefused, true),
-            (ConnectionReset, true),
-            (HttpStatus(StatusCode::TOO_MANY_REQUESTS), true),
-            (HttpStatus(StatusCode::SERVICE_UNAVAILABLE), true),
-            (HttpStatus(StatusCode::CONFLICT), true),
-            (Unknown, false),
-            (HttpStatus(StatusCode::BAD_REQUEST), false),
-            (HttpStatus(StatusCode::INTERNAL_SERVER_ERROR), false),
-        ] {
-            let original = DiskError::from(rustfs_rio::new_test_internode_http_io_error(kind));
-            assert_eq!(original.internode_http_error_kind(), Some(kind));
-            assert_eq!(original.is_retryable_internode_write_failure(), retryable);
-
-            let cloned = original.clone();
-            assert_eq!(cloned, original, "clone must preserve the error bucket for {kind:?}");
-            assert_eq!(
-                cloned.is_retryable_internode_write_failure(),
-                retryable,
-                "clone changed retryability for {kind:?}"
-            );
-            assert_eq!(cloned.internode_http_error_kind(), Some(kind));
-            if let HttpStatus(status) = kind {
-                assert!(cloned.is_internode_http_status(status.as_u16()));
-            }
-            let DiskError::Io(io_error) = &cloned else {
-                panic!("unmarked internode error must remain Io: {cloned:?}");
-            };
-            let source = io_error
-                .get_ref()
-                .and_then(|source| source.downcast_ref::<InternodeHttpError>())
-                .expect("clone must retain the structured internode error");
-            assert_eq!(source.context().method(), "PUT");
-            assert_eq!(source.context().target(), "/rustfs/rpc/put_file_stream");
-            assert_eq!(source.context().operation(), Some(INTERNODE_OPERATION_PUT_FILE_STREAM));
-        }
-    }
-
    #[tokio::test]
    async fn read_stream_conflict_is_not_a_retryable_put_file_failure() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
@@ -1358,57 +1309,11 @@ mod tests {
                !error.is_retryable_internode_write_failure(),
                "read-operation 409 must not trigger put-file retry"
            );
-            let cloned = error.clone();
-            let reduced = crate::disk::error_reduce::reduce_write_quorum_errs(&[Some(error)], &[], 1)
-                .expect("the read conflict must remain the dominant error");
-            for preserved in [&cloned, &reduced] {
-                assert!(
-                    !preserved.is_retryable_internode_write_failure(),
-                    "cloning or reducing a read conflict must not turn it into a PUT retry"
-                );
-                assert!(preserved.is_internode_http_status(409));
-                let DiskError::Io(io_error) = preserved else {
-                    panic!("read conflict must remain Io: {preserved:?}");
-                };
-                let source = io_error
-                    .get_ref()
-                    .and_then(|source| source.downcast_ref::<InternodeHttpError>())
-                    .expect("read conflict must retain its request context");
-                assert_eq!(source.context().method(), "GET");
-                assert_eq!(source.context().target(), "/rustfs/rpc/read_file_stream");
-                assert_eq!(
-                    source.context().operation(),
-                    Some(rustfs_io_metrics::internode_metrics::INTERNODE_OPERATION_READ_FILE_STREAM)
-                );
-            }
        })
        .await
        .expect("isolated read-conflict test must finish within its budget");
    }

-    #[test]
-    fn test_internode_http_clone_preserves_outer_io_kind_and_message() {
-        let source = rustfs_rio::new_test_internode_http_io_error(InternodeHttpErrorKind::ConnectionReset)
-            .into_inner()
-            .expect("the internode helper must provide a typed source");
-        let original_io = io::Error::new(io::ErrorKind::InvalidData, source);
-        let message = original_io.to_string();
-        let original = DiskError::from(original_io);
-        assert_eq!(original.internode_http_error_kind(), Some(InternodeHttpErrorKind::ConnectionReset));
-        assert!(original.is_retryable_internode_write_failure());
-
-        let cloned = original.clone();
-        let reduced = crate::disk::error_reduce::reduce_write_quorum_errs(&[Some(original)], &[], 1)
-            .expect("the wrapped internode error must remain the dominant error");
-        for preserved in [&cloned, &reduced] {
-            let DiskError::Io(io_error) = preserved else {
-                panic!("the wrapped error must remain Io: {preserved:?}");
-            };
-            assert_eq!(io_error.kind(), io::ErrorKind::InvalidData);
-            assert_eq!(io_error.to_string(), message);
-        }
-    }
-
    #[test]
    fn test_internode_missing_errors_preserve_disk_error_types() {
        let file_missing = DiskError::from(rustfs_rio::new_test_remote_file_not_found_http_io_error());
@@ -1420,17 +1325,6 @@ mod tests {
        assert_eq!(file_missing, DiskError::FileNotFound);
        assert_eq!(volume_missing, DiskError::VolumeNotFound);
        assert!(matches!(unmarked_server_error, DiskError::Io(_)));
-        for missing in [file_missing, volume_missing] {
-            assert_eq!(missing.clone(), missing);
-            assert_eq!(
-                crate::disk::error_reduce::reduce_write_quorum_errs(
-                    &[Some(missing.clone()), Some(missing.clone()), None],
-                    &[],
-                    2
-                ),
-                Some(missing)
-            );
-        }
    }

    #[test]
@@ -226,78 +226,6 @@ mod tests {
        assert_eq!(res, Some(quorum_err));
    }

-    #[test]
-    fn test_write_quorum_reduction_preserves_internode_http_identity() {
-        use http::StatusCode;
-        use rustfs_rio::InternodeHttpErrorKind::{ConnectionRefused, HttpStatus, Unknown};
-
-        for (kind, retryable) in [
-            (ConnectionRefused, true),
-            (HttpStatus(StatusCode::SERVICE_UNAVAILABLE), true),
-            (HttpStatus(StatusCode::CONFLICT), true),
-            (Unknown, false),
-            (HttpStatus(StatusCode::BAD_REQUEST), false),
-        ] {
-            // Construct both producer errors independently: the reducer owns the first clone.
-            let first = Error::from(rustfs_rio::new_test_internode_http_io_error(kind));
-            let second = Error::from(rustfs_rio::new_test_internode_http_io_error(kind));
-            assert_eq!(first.internode_http_error_kind(), Some(kind));
-            assert_eq!(second.internode_http_error_kind(), Some(kind));
-            assert_eq!(first.is_retryable_internode_write_failure(), retryable);
-            let errors = [Some(first), Some(second), None];
-            let reduced = reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, 2)
-                .expect("two equal producer errors must dominate one successful write");
-
-            assert_eq!(Some(&reduced), errors[0].as_ref());
-            assert_eq!(
-                reduced.is_retryable_internode_write_failure(),
-                retryable,
-                "quorum reduction changed retryability for {kind:?}"
-            );
-            assert_eq!(reduced.internode_http_error_kind(), Some(kind));
-            if let HttpStatus(status) = kind {
-                assert!(reduced.is_internode_http_status(status.as_u16()));
-            }
-            let Error::Io(io_error) = &reduced else {
-                panic!("the dominant error must remain Io: {reduced:?}");
-            };
-            let source = io_error
-                .get_ref()
-                .and_then(|source| source.downcast_ref::<rustfs_rio::InternodeHttpError>())
-                .expect("quorum reduction must retain the structured internode error");
-            assert_eq!(source.context().method(), "PUT");
-            assert_eq!(source.context().target(), "/rustfs/rpc/put_file_stream");
-            assert_eq!(
-                source.context().operation(),
-                Some(rustfs_io_metrics::internode_metrics::INTERNODE_OPERATION_PUT_FILE_STREAM)
-            );
-        }
-    }
-
-    #[test]
-    fn test_clone_and_write_quorum_do_not_promote_non_retryable_errors() {
-        use http::StatusCode;
-        use rustfs_rio::InternodeHttpErrorKind::{HttpStatus, Unknown};
-
-        for original in [
-            Error::from(rustfs_rio::new_test_internode_http_io_error(Unknown)),
-            Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(StatusCode::BAD_REQUEST))),
-            Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(StatusCode::FORBIDDEN))),
-            Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(StatusCode::NOT_FOUND))),
-            Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(
-                StatusCode::INTERNAL_SERVER_ERROR,
-            ))),
-            err_io("internode connection reset: PUT /rustfs/rpc/put_file_stream"),
-        ] {
-            assert!(!original.is_retryable_internode_write_failure());
-            let cloned = original.clone();
-            let reduced =
-                reduce_write_quorum_errs(&[Some(original)], &[], 1).expect("a non-retryable error must remain an error");
-            assert!(!cloned.is_retryable_internode_write_failure());
-            assert!(!reduced.is_retryable_internode_write_failure());
-        }
-    }
-
    #[test]
    fn test_count_errs() {
        let e1 = err_io("a");
@@ -572,7 +572,7 @@ impl ECStore {
    where
        S: EcstoreObjectIO + StorageNamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
    {
-        // Lock order: pool_meta_save_gate -> rebalance.bin -> pool.bin.
+        // Lock order: pool_meta_save_gate -> pool.bin -> rebalance.bin.
        let mut pool_meta_guard = self.pool_meta_save_gate.lock().await;
        pool_meta_guard.ensure_write_safe("rebalance worker activation")?;
        // Classify the durable rebalance record while holding both namespace
@@ -50,11 +50,6 @@ fn ensure_rebalance_entry_active(cancel: &CancellationToken) -> Result<()> {
    Ok(())
 }

-#[cfg(test)]
-tokio::task_local! {
-    static REBALANCE_ENTRY_RUN_FENCE_BARRIER: (Arc<tokio::sync::Notify>, Arc<tokio::sync::Notify>);
-}
-
 #[derive(Debug)]
 struct RebalanceEntryTarget {
    bucket: String,
@@ -261,15 +256,9 @@ impl ECStore {
            .sort_by_key(|v| (v.mod_time.is_none(), std::cmp::Reverse(v.mod_time)));

        // Entry lock order is bucket incarnation -> activation_gate -> rebalance.bin -> movement gate.
-        // Target capacity admission can then acquire pool.bin under the run fence.
        // Stop waits for in-flight entries through cleanup, but not for entries admitted later.
        ensure_rebalance_entry_active(&cancel)?;
        let run_guard = self.rebalance_run_guard(rebalance_id.as_ref(), "rebalance entry").await?;
-        #[cfg(test)]
-        if let Ok((arrived, release)) = REBALANCE_ENTRY_RUN_FENCE_BARRIER.try_with(Clone::clone) {
-            arrived.notify_one();
-            release.notified().await;
-        }
        let lock_lost_signal = run_guard.lock_lost_signal();
        #[cfg(test)]
        let _run_signal_test_fence = lock_lost_signal
@@ -1248,130 +1237,6 @@ mod tests {
        assert_eq!(pool_stats.cleanup_warnings.count, 1, "deferred cleanup must not add a permanent warning");
    }

-    #[tokio::test]
-    #[serial_test::serial]
-    async fn real_rebalance_entry_progresses_while_peer_activation_waits_for_run_fence() {
-        const REBALANCE_ID: &str = "rebalance-peer-activation-lock-order";
-        let (_temp_dirs, store, peer) = crate::services::rebalance::test_two_pool_stores_with_isolated_node_contexts(Some(
-            active_rebalance_meta(REBALANCE_ID),
-        ))
-        .await;
-        assert!(!Arc::ptr_eq(&store.ctx, &peer.ctx), "node-local movement gates must be independent");
-        {
-            let mut meta = peer.rebalance_meta.write().await;
-            let meta = meta.as_mut().expect("peer should know the durable run");
-            meta.activation_gate = Arc::default();
-            meta.cancel = None;
-        }
-        let bucket = crate::disk::RUSTFS_META_BUCKET;
-        let object = "rebalance-peer-activation-object";
-        let version_id = uuid::Uuid::new_v4();
-        let payload = b"entry must drain before peer activation takes the pool fence".repeat(1024);
-        let source_set = store.pools[0].get_disks_by_key(object);
-        let target_set = store.pools[1].get_disks_by_key(object);
-        let opts = ObjectOptions {
-            versioned: true,
-            version_id: Some(version_id.to_string()),
-            ..Default::default()
-        };
-        let mut writer = PutObjReader::from_vec(payload.clone());
-        let source_before = source_set
-            .put_object(bucket, object, &mut writer, &opts)
-            .await
-            .expect("source version should be written");
-        let entry = metacache_entry_from_source(&source_set, bucket, object).await;
-        let arrived = Arc::new(tokio::sync::Notify::new());
-        let release = Arc::new(tokio::sync::Notify::new());
-        // JoinSet aborts both scoped tasks if an assertion or timeout fails.
-        let mut tasks = tokio::task::JoinSet::new();
-        let entry_store = Arc::clone(&store);
-        tasks.spawn(
-            REBALANCE_ENTRY_RUN_FENCE_BARRIER.scope((Arc::clone(&arrived), Arc::clone(&release)), async move {
-                entry_store
-                    .rebalance_entry(
-                        RebalanceEntryTarget {
-                            bucket: bucket.to_string(),
-                            pool_index: 0,
-                        },
-                        entry,
-                        source_set,
-                        Arc::new(RebalanceBucketConfigs::default()),
-                        Arc::from(REBALANCE_ID),
-                        CancellationToken::new(),
-                    )
-                    .await
-            }),
-        );
-        tokio::time::timeout(StdDuration::from_secs(30), arrived.notified())
-            .await
-            .expect("real entry must acquire its persisted run read fence");
-
-        let attempted = Arc::new(tokio::sync::Notify::new());
-        let peer_pool = Arc::clone(&peer.pools[0]);
-        let (activation_done, activation_result) = tokio::sync::oneshot::channel();
-        tasks.spawn(
-            crate::core::pools::REBALANCE_ACTIVATION_LOCK_ATTEMPT.scope(Arc::clone(&attempted), async move {
-                let result = peer.fence_rebalance_worker_activation(peer_pool, REBALANCE_ID).await;
-                let result = result.map(|fence| match fence {
-                    super::super::control::RebalanceWorkerActivationFence::Ready(fence) => {
-                        fence.ensure_held().expect("peer activation must retain both fences");
-                    }
-                    super::super::control::RebalanceWorkerActivationFence::NotStartedTerminal => {
-                        panic!("the paused entry's run must still require activation");
-                    }
-                });
-                activation_done.send(result).expect("activation receiver should remain alive");
-                Ok(RebalanceEntryOutcome::Completed)
-            }),
-        );
-        tokio::time::timeout(StdDuration::from_secs(30), attempted.notified())
-            .await
-            .expect("peer activation must attempt the persisted rebalance write fence");
-        release.notify_one();
-
-        tokio::time::timeout(StdDuration::from_secs(30), async {
-            while let Some(result) = tasks.join_next().await {
-                assert!(matches!(
-                    result
-                        .expect("scoped task must not panic")
-                        .expect("entry must not fail or defer"),
-                    RebalanceEntryOutcome::Completed
-                ));
-            }
-        })
-        .await
-        .expect("entry and peer activation must both make progress");
-        activation_result
-            .await
-            .expect("peer activation result should be sent")
-            .expect("peer activation must not time out behind the entry it blocks");
-
-        let mut reader = target_set
-            .get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
-            .await
-            .expect("the exact target version must be readable");
-        let mut actual = Vec::new();
-        reader
-            .stream
-            .read_to_end(&mut actual)
-            .await
-            .expect("target body should drain completely");
-        assert_eq!(actual, payload);
-        assert_eq!(reader.object_info.version_id, source_before.version_id);
-        assert_eq!(reader.object_info.etag, source_before.etag);
-        assert_eq!(reader.object_info.mod_time, source_before.mod_time);
-        let source_error = store.pools[0]
-            .get_object_info(bucket, object, &opts)
-            .await
-            .expect_err("completed entry must clean up the source version");
-        assert!(crate::error::is_err_object_not_found(&source_error) || crate::error::is_err_version_not_found(&source_error));
-        let meta = store.rebalance_meta.read().await;
-        let stats = &meta.as_ref().expect("local run must remain installed").pool_stats[0];
-        assert_eq!(stats.num_objects, 1);
-        assert_eq!(stats.num_versions, 1);
-        assert_eq!(stats.cleanup_warnings.count, 0);
-    }
-
    #[tokio::test]
    #[serial_test::serial]
    async fn real_rebalance_run_fence_loss_before_target_commit_preserves_target_and_source() {
@@ -1907,124 +1907,6 @@ fn test_is_transient_rebalance_error_accepts_wrapped_disk_timeout() {
    assert!(is_transient_rebalance_error(&Error::Io(std::io::Error::other(DiskError::Timeout))));
 }

-#[test]
-fn test_rebalance_stage_wrapped_transient_errors_remain_retryable() {
-    let cases = [
-        Error::Lock(rustfs_lock::LockError::timeout(".rustfs.sys/pool.bin@latest", Duration::from_secs(5))),
-        Error::Lock(rustfs_lock::LockError::network(
-            "peer unavailable",
-            std::io::Error::from(std::io::ErrorKind::ConnectionReset),
-        )),
-        Error::SlowDown,
-        Error::ErasureReadQuorum,
-        Error::ErasureWriteQuorum,
-        Error::Io(std::io::Error::other(DiskError::Timeout)),
-        Error::Io(std::io::Error::from(std::io::ErrorKind::TimedOut)),
-    ];
-    for mut error in cases {
-        for depth in 0..=3 {
-            assert!(is_transient_rebalance_error(&error), "transient source lost at depth {depth}: {error:?}");
-            assert!(
-                should_defer_rebalance_entry_failure(&error),
-                "exhausted transient entries must be deferred"
-            );
-            assert!(should_retry_rebalance_listing(&error, 0, 3));
-            assert!(
-                !should_retry_rebalance_listing(&error, 2, 3),
-                "wrapping must not bypass the attempt limit"
-            );
-            error = data_movement::data_movement_stage_error_for_test(
-                "rebalance_object",
-                "put_object",
-                "bucket",
-                "baseline/00042.bin",
-                error,
-            );
-        }
-    }
-}
-
-#[test]
-fn test_rebalance_stage_wrapped_terminal_errors_remain_terminal() {
-    let cases = [
-        Error::FileAccessDenied,
-        Error::FileCorrupt,
-        Error::OperationCanceled,
-        Error::DataMovementOverwriteErr("bucket".to_string(), "object".to_string(), "version".to_string()),
-        Error::Lock(rustfs_lock::LockError::already_locked("bucket/object", "owner")),
-        Error::other("permission denied"),
-    ];
-    for mut error in cases {
-        for depth in 0..=3 {
-            assert!(
-                !is_transient_rebalance_error(&error),
-                "terminal source must survive depth {depth}: {error:?}"
-            );
-            assert!(!should_defer_rebalance_entry_failure(&error));
-            // Object names are untrusted context, not evidence of a transient failure.
-            error = data_movement::data_movement_stage_error_for_test(
-                "rebalance_object",
-                "put_object",
-                "bucket",
-                "remote lock rpc timed out",
-                error,
-            );
-        }
-    }
-}
-
-#[tokio::test]
-async fn test_rebalance_stage_wrapped_lock_timeout_retries_real_migration_loop() {
-    for succeeds_on_retry in [true, false] {
-        let backend = MigrationBackendSpy::new(None, None);
-        let attempts = AtomicUsize::new(0);
-        let waits = AtomicUsize::new(0);
-        let mut transfer = |_, _, _| {
-            let attempt = attempts.fetch_add(1, Ordering::SeqCst);
-            async move {
-                if succeeds_on_retry && attempt > 0 {
-                    return Ok(());
-                }
-                Err(data_movement::data_movement_stage_error_for_test(
-                    "rebalance_object",
-                    "put_object",
-                    "bucket",
-                    "baseline/00042.bin",
-                    Error::Lock(rustfs_lock::LockError::timeout(".rustfs.sys/pool.bin@latest", Duration::from_secs(5))),
-                ))
-            }
-        };
-        let version = version_normal();
-        let result = migrate_entry_version_with_retry_wait(
-            &backend,
-            "bucket".to_string(),
-            0,
-            &version,
-            None,
-            3,
-            false,
-            &mut transfer,
-            |_: String, _: String, _: ObjectOptions| async { Ok::<_, Error>(ObjectInfo::default()) },
-            |_| {
-                waits.fetch_add(1, Ordering::SeqCst);
-                std::future::ready(())
-            },
-        )
-        .await;
-        assert_eq!(result.moved, succeeds_on_retry);
-        assert_eq!(result.failed, !succeeds_on_retry);
-        assert_eq!(attempts.load(Ordering::SeqCst), if succeeds_on_retry { 2 } else { 3 });
-        assert_eq!(backend.get_calls(), attempts.load(Ordering::SeqCst));
-        assert_eq!(waits.load(Ordering::SeqCst), attempts.load(Ordering::SeqCst) - 1);
-        if !succeeds_on_retry {
-            assert_eq!(result.stage, Some("write_target"));
-            assert!(should_defer_rebalance_entry_failure(
-                result.error.as_ref().expect("exhaustion must retain its source error")
-            ));
-        }
-    }
-}
-
 #[test]
 fn test_is_transient_rebalance_error_accepts_io_timeout_message() {
    assert!(is_transient_rebalance_error(&Error::Io(std::io::Error::other("timeout"))));
@@ -244,7 +244,6 @@ pub(super) fn resolve_rebalance_bucket_result(
 }

 pub(super) fn is_transient_rebalance_error(err: &Error) -> bool {
-    let err = rebalance_error_source(err);
    match err {
        Error::SlowDown
        | Error::ErasureReadQuorum
@@ -257,15 +256,6 @@ pub(super) fn is_transient_rebalance_error(err: &Error) -> bool {
    }
 }

-fn rebalance_error_source(mut err: &Error) -> &Error {
-    // Stage context contains object names, so classify the preserved source,
-    // not timeout-like text supplied by an object name. Iterate nested stages.
-    while let Some(source) = crate::data_movement::data_movement_stage_source(err) {
-        err = source;
-    }
-    err
-}
-
 fn is_rebalance_transient_lock_error(err: &rustfs_lock::LockError) -> bool {
    match err {
        rustfs_lock::LockError::Timeout { .. } | rustfs_lock::LockError::Network { .. } => true,
@@ -319,7 +309,6 @@ pub(super) fn rebalance_listing_retry_delay(attempt: usize) -> Duration {
 }

 fn is_rebalance_lock_or_rpc_timeout(err: &Error) -> bool {
-    let err = rebalance_error_source(err);
    match err {
        Error::Lock(rustfs_lock::LockError::Timeout { .. }) | Error::Lock(rustfs_lock::LockError::Network { .. }) => true,
        Error::Io(io_err) => is_rebalance_lock_or_rpc_timeout_message(&io_err.to_string()),
@@ -596,48 +585,3 @@ impl SetDisks {
        Ok(())
    }
 }
-
-#[cfg(test)]
-mod error_source_tests {
-    use super::*;
-
-    #[test]
-    fn stage_wrapped_errors_select_the_source_backoff_policy() {
-        let cases = [
-            (
-                Error::Lock(rustfs_lock::LockError::timeout(".rustfs.sys/pool.bin@latest", Duration::from_secs(5))),
-                true,
-            ),
-            (
-                Error::Lock(rustfs_lock::LockError::network(
-                    "peer unavailable",
-                    std::io::Error::from(std::io::ErrorKind::ConnectionReset),
-                )),
-                true,
-            ),
-            (Error::other("remote lock rpc timed out"), true),
-            (Error::SlowDown, false),
-            (Error::Io(std::io::Error::other(DiskError::Timeout)), false),
-            (Error::FileAccessDenied, false),
-        ];
-        for (mut error, lock_backoff) in cases {
-            for depth in 0..=3 {
-                assert_eq!(
-                    is_rebalance_lock_or_rpc_timeout(&error),
-                    lock_backoff,
-                    "wrong backoff at depth {depth}: {error:?}"
-                );
-                if !lock_backoff {
-                    assert_eq!(rebalance_migration_retry_delay(1, &error), REBALANCE_MIGRATION_RETRY_BASE_DELAY * 2);
-                }
-                error = crate::data_movement::data_movement_stage_error_for_test(
-                    "rebalance_object",
-                    "put_object",
-                    "bucket",
-                    "remote lock rpc timed out",
-                    error,
-                );
-            }
-        }
-    }
-}
@@ -326,15 +326,14 @@ impl SetDisks {
        let parity_blocks = Self::common_parity(&parities, default_parity_count as i32);

        if parity_blocks < 0 {
-            // A consistent layout can require more replies than the initial
-            // half-set probe. Reaching that probe alone is not corruption;
-            // only invalid or conflicting healthy replies establish that.
+            // No parity value reached read quorum. Distinguish two cases:
+            // enough disks answered with valid-looking metadata that simply
+            // cannot be reconciled (corrupt/foreign entries — retrying cannot
+            // help, and heal should see Corrupt, rustfs#5801) versus too few
+            // healthy answers (a genuine quorum condition where retry may
+            // succeed once disks recover).
            let healthy_replies = errs.iter().filter(|err| err.is_none()).count();
-            let consistent_parity = parities
-                .iter()
-                .find(|&&parity| parity >= 0)
-                .filter(|&&parity| parities.iter().filter(|&&candidate| candidate == parity).count() == healthy_replies);
-            if healthy_replies >= expected_rquorum && consistent_parity.is_none() {
+            if healthy_replies >= expected_rquorum {
                error!(
                    "object_quorum_from_meta: irreconcilable parity across {healthy_replies} healthy replies (corrupt metadata), errs={errs:?}"
                );
@@ -1653,40 +1652,6 @@ mod tests {
        assert_eq!(err, DiskError::FileCorrupt);
    }

-    #[test]
-    fn consistent_parity_below_its_data_shard_quorum_is_not_corruption() {
-        for (drive_count, parity) in [(6, 2), (8, 2), (12, 4)] {
-            let data = drive_count - parity;
-            let mut metas = (1..=drive_count)
-                .map(|index| {
-                    let mut info = FileInfo::new("bucket/object", data, parity);
-                    info.size = 1024;
-                    info.erasure.index = index;
-                    info
-                })
-                .collect::<Vec<_>>();
-            let mut errs = vec![Some(DiskError::DiskNotFound); drive_count];
-            errs[..data].fill(None);
-            assert_eq!(
-                SetDisks::object_quorum_from_meta(&metas, &errs, parity).expect("exact data quorum should resolve"),
-                (data as i32, data as i32)
-            );
-
-            errs[data - 1] = Some(DiskError::DiskNotFound);
-            assert_eq!(
-                SetDisks::object_quorum_from_meta(&metas, &errs, parity).expect_err("one fewer shard cannot resolve"),
-                DiskError::ErasureReadQuorum,
-                "layout {drive_count}/{parity} has consistent metadata but insufficient shards"
-            );
-
-            metas[0].erasure.parity_blocks = usize::MAX;
-            assert_eq!(
-                SetDisks::object_quorum_from_meta(&metas, &errs, parity).expect_err("corrupt healthy replies must be rejected"),
-                DiskError::FileCorrupt
-            );
-        }
-    }
-
    /// Too few healthy replies remains a genuine quorum condition where a
    /// retry may succeed once disks recover.
    #[test]
@@ -865,7 +865,6 @@ pub(crate) use core::io_primitives::{ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, ren
 mod ctx;
 mod metadata;
 mod ops;
-pub(crate) use ops::bucket::BucketInfoQuorum;

 #[cfg(test)]
 pub(crate) use ops::hermetic_set_disks_isolated;
@@ -21,72 +21,12 @@

 use super::super::{
    BUCKET_OP_IGNORED_ERRS, BucketInfo, BucketOperations, BucketOptions, DeleteBucketOptions, DiskError, Error, HashMap,
-    MakeBucketOptions, Result, SetDisks, is_reserved_or_invalid_bucket, join_all, reduce_read_quorum_errs,
-    reduce_write_quorum_errs,
+    MakeBucketOptions, Result, SetDisks, is_reserved_or_invalid_bucket, join_all, reduce_write_quorum_errs,
 };
 use crate::api::bucket::metadata_sys;
 use crate::disk::DiskAPI;

-#[derive(Clone, Copy)]
-pub(crate) enum BucketInfoQuorum {
-    Read,
-    Write,
-}
-
 impl SetDisks {
-    pub(crate) async fn stat_bucket_with_quorum(&self, bucket: &str, quorum: BucketInfoQuorum) -> Result<BucketInfo> {
-        let disks = self.disk_inventory().await;
-        let disk_count = disks.len();
-        let mut futures = Vec::with_capacity(disk_count);
-        for disk in disks {
-            let bucket = bucket.to_string();
-            futures.push(async move {
-                match disk {
-                    Some(disk) => disk.stat_volume(&bucket).await,
-                    None => Err(DiskError::DiskNotFound),
-                }
-            });
-        }
-
-        let results = join_all(futures).await;
-        let mut infos = Vec::with_capacity(results.len());
-        let mut errs = Vec::with_capacity(results.len());
-        for result in results {
-            match result {
-                Ok(info) => {
-                    infos.push(Some(info));
-                    errs.push(None);
-                }
-                Err(err) => {
-                    infos.push(None);
-                    errs.push(Some(err));
-                }
-            }
-        }
-
-        let error = match quorum {
-            // Bucket mutations use a majority regardless of object storage
-            // class. A namespace read must intersect that majority; object
-            // readers still enforce the persisted layout's data-shard quorum.
-            BucketInfoQuorum::Read => reduce_read_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, disk_count.div_ceil(2).max(1)),
-            BucketInfoQuorum::Write => reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, disk_count / 2 + 1),
-        };
-        if let Some(err) = error {
-            return Err(err.into());
-        }
-
-        infos
-            .into_iter()
-            .flatten()
-            .next()
-            .map(|info| BucketInfo {
-                name: info.name,
-                created: info.created,
-                ..Default::default()
-            })
-            .ok_or(Error::VolumeNotFound)
-    }
-
    pub(crate) async fn list_bucket_for_scanner(&self, _opts: &BucketOptions) -> Result<(Vec<BucketInfo>, bool)> {
        let disks = self.disk_inventory().await;
        let write_quorum = (disks.len() / 2) + 1;
@@ -191,12 +131,59 @@ impl BucketOperations for SetDisks {

    #[tracing::instrument(skip(self))]
    async fn get_bucket_info(&self, bucket: &str, _opts: &BucketOptions) -> Result<BucketInfo> {
-        let mut info = self.stat_bucket_with_quorum(bucket, BucketInfoQuorum::Write).await?;
-        if let Ok(sys) = metadata_sys::get(bucket).await {
-            info.versioning = sys.versioning();
-            info.object_locking = sys.object_locking();
+        let disks = self.disk_inventory().await;
+        let write_quorum = (disks.len() / 2) + 1;
+
+        let mut futures = Vec::with_capacity(disks.len());
+        for disk in disks {
+            let bucket = bucket.to_string();
+            futures.push(async move {
+                match disk {
+                    Some(disk) => disk.stat_volume(&bucket).await,
+                    None => Err(DiskError::DiskNotFound),
                }
-        Ok(info)
+            });
+        }
+
+        let results = join_all(futures).await;
+        let mut infos = Vec::with_capacity(results.len());
+        let mut errs = Vec::with_capacity(results.len());
+        for result in results {
+            match result {
+                Ok(info) => {
+                    infos.push(Some(info));
+                    errs.push(None);
+                }
+                Err(err) => {
+                    infos.push(None);
+                    errs.push(Some(err));
+                }
+            }
+        }
+
+        if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) {
+            return Err(err.into());
+        }
+
+        let mut versioning = false;
+        let mut object_locking = false;
+        if let Ok(sys) = metadata_sys::get(bucket).await {
+            versioning = sys.versioning();
+            object_locking = sys.object_locking();
+        }
+
+        infos
+            .into_iter()
+            .flatten()
+            .next()
+            .map(|info| BucketInfo {
+                name: info.name,
+                created: info.created,
+                versioning,
+                object_locking,
+                ..Default::default()
+            })
+            .ok_or(Error::VolumeNotFound)
    }

    #[tracing::instrument(skip(self))]
@@ -19,7 +19,7 @@ use crate::bucket::{
 };
 use crate::error::is_err_bucket_not_found;
 use crate::runtime::sources as runtime_sources;
-use crate::set_disk::{BucketInfoQuorum, get_lock_acquire_timeout};
+use crate::set_disk::get_lock_acquire_timeout;
 use crate::storage_api_contracts::bucket::{BUCKET_LIFECYCLE_LOCK_OBJECT, SRBucketDeleteOp};
 use crate::storage_api_contracts::namespace::NamespaceLocking as _;
 use futures::stream::{self, StreamExt};
@@ -772,28 +772,15 @@ impl ECStore {

    #[instrument(skip(self))]
    pub(crate) async fn get_bucket_info_from_sets(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
-        self.get_bucket_info_from_sets_with_quorum(bucket, opts, BucketInfoQuorum::Write)
-            .await
-    }
-
-    async fn get_bucket_info_from_sets_with_quorum(
-        &self,
-        bucket: &str,
-        opts: &BucketOptions,
-        quorum: BucketInfoQuorum,
-    ) -> Result<BucketInfo> {
        // One host may participate in several pools after expansion. Resolve the
        // namespace against each erasure set so disks from different pools can
        // never be combined into one bucket quorum.
        // Bucket validation is request-path IO. Keep the previous peer fanout's
        // latency shape by probing every set concurrently; scanner listings use
        // a separate bounded path below because they run continuously.
-        let mut scoped_results = futures::future::join_all(self.bucket_sets().map(|(pool_index, set_index, set)| async move {
-            let result = match quorum {
-                BucketInfoQuorum::Read => set.stat_bucket_with_quorum(bucket, quorum).await,
-                BucketInfoQuorum::Write => set.get_bucket_info(bucket, opts).await,
-            };
-            (pool_index, set_index, result)
+        let mut scoped_results =
+            futures::future::join_all(self.bucket_sets().map(|(pool_index, set_index, set)| async move {
+                (pool_index, set_index, set.get_bucket_info(bucket, opts).await)
            }))
            .await;
        scoped_results.sort_unstable_by_key(|(pool_index, set_index, _)| (*pool_index, *set_index));
@@ -819,11 +806,7 @@ impl ECStore {

    #[instrument(skip(self))]
    pub(super) async fn handle_get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
-        let mut info = match self.get_bucket_info_from_sets(bucket, opts).await {
-            Ok(info) => info,
-            Err(Error::ErasureWriteQuorum) => return self.get_bucket_info_at_read_quorum(bucket, opts).await,
-            Err(err) => return Err(err),
-        };
+        let mut info = self.get_bucket_info_from_sets(bucket, opts).await?;

        if let Ok(sys) = metadata_sys::get_in(&self.ctx, bucket).await {
            if should_override_created_from_metadata(sys.created) {
@@ -836,35 +819,6 @@ impl ECStore {
        Ok(info)
    }

-    async fn get_bucket_info_at_read_quorum(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
-        // Lock order: bucket lifecycle -> internal metadata object read locks.
-        // Keep create/delete from changing the namespace while a read quorum
-        // confirms both physical presence and persisted bucket metadata.
-        let guard = self.acquire_bucket_lifecycle_read_lock(bucket).await?;
-        await_bucket_namespace_operation(Some(&guard), bucket, "bucket read quorum validation", async {
-            let mut info = self
-                .get_bucket_info_from_sets_with_quorum(bucket, opts, BucketInfoQuorum::Read)
-                .await?;
-            let (metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?;
-            if !persisted {
-                // A minority of directories left by failed creation is not an
-                // authoritative bucket. Never turn fabricated defaults into
-                // permission to serve degraded reads.
-                return Err(Error::ErasureReadQuorum);
-            }
-            if metadata.name != bucket {
-                return Err(Error::FileCorrupt);
-            }
-            if should_override_created_from_metadata(metadata.created) {
-                info.created = Some(metadata.created);
-            }
-            info.versioning = metadata.versioning();
-            info.object_locking = metadata.object_locking();
-            Ok(info)
-        })
-        .await
-    }
-
    #[instrument(skip(self))]
    pub(super) async fn handle_list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
        // TODO(backlog): support cached bucket listing via opts.cached
@@ -1095,7 +1049,7 @@ mod tests {
        run_physical_bucket_deletion, scan_metadata_less_residue, scan_metadata_less_residue_with_budget,
        should_override_created_from_metadata, validate_table_bucket_delete_allowed,
    };
-    use crate::bucket::metadata::{BucketMetadata, table_bucket_catalog_metadata_prefix};
+    use crate::bucket::metadata::table_bucket_catalog_metadata_prefix;
    use crate::bucket::metadata_sys;
    use crate::cluster::rpc::peer_s3_client::install_delete_bucket_empty_scan_barrier;
    use crate::disk::{BUCKET_META_PREFIX, DiskAPI, RUSTFS_META_BUCKET, STORAGE_FORMAT_FILE};
@@ -1122,7 +1076,6 @@ mod tests {
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::time::{Duration, SystemTime};
    use time::OffsetDateTime;
-    use tokio::io::AsyncReadExt;
    use tokio::sync::{Notify, OnceCell};
    use tokio_util::sync::CancellationToken;
    use uuid::Uuid;
@@ -1406,18 +1359,11 @@ mod tests {
    }

    async fn setup_multi_pool_bucket_test_env() -> (tempfile::TempDir, Arc<ECStore>) {
-        setup_bucket_quorum_test_env(&[4, 4], None).await
-    }
-
-    async fn setup_bucket_quorum_test_env(
-        drives_per_pool: &[usize],
-        standard_parity: Option<usize>,
-    ) -> (tempfile::TempDir, Arc<ECStore>) {
        let temp_dir = tempfile::tempdir().expect("multi-pool bucket test directory should be created");
        let mut pools = Vec::new();
-        for (pool_index, &drive_count) in drives_per_pool.iter().enumerate() {
+        for pool_index in 0..2 {
            let mut endpoints = Vec::new();
-            for disk_index in 0..drive_count {
+            for disk_index in 0..4 {
                let disk_path = temp_dir.path().join(format!("pool{pool_index}-disk{disk_index}"));
                tokio::fs::create_dir_all(&disk_path)
                    .await
@@ -1432,7 +1378,7 @@ mod tests {
            pools.push(PoolEndpoints {
                legacy: false,
                set_count: 1,
-                drives_per_set: drive_count,
+                drives_per_set: 4,
                endpoints: Endpoints::from(endpoints),
                cmd_line: format!("bucket-test-pool-{pool_index}"),
                platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
@@ -1453,12 +1399,9 @@ mod tests {
        )
        .await
        .expect("multi-pool ECStore should initialize");
-        let mut storage_class_kvs = rustfs_config::server_config::KVS::new();
-        if let Some(parity) = standard_parity {
-            storage_class_kvs.insert(crate::config::storageclass::CLASS_STANDARD.to_string(), format!("EC:{parity}"));
-        }
-        let storage_class = crate::config::storageclass::lookup_config_for_pools_without_env(&storage_class_kvs, drives_per_pool)
-            .expect("storage class should match every test erasure set");
+        let storage_class =
+            crate::config::storageclass::lookup_config_for_pools_without_env(&rustfs_config::server_config::KVS::new(), &[4, 4])
+                .expect("multi-pool storage class should match both four-disk pools");
        for pool in &ecstore.pools {
            for set in &pool.disk_set {
                set.set_test_storage_class_config(storage_class.clone());
@@ -2124,218 +2067,6 @@ mod tests {
        }
    }

-    #[tokio::test]
-    #[serial]
-    async fn bucket_info_read_quorum_tracks_erasure_layout() {
-        for (drive_count, parity) in [(2, 1), (3, 1), (4, 2), (5, 2), (6, 3), (8, 4), (6, 2), (12, 6)] {
-            let (_temp_dir, store) = setup_bucket_quorum_test_env(&[drive_count], Some(parity)).await;
-            metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
-            let bucket = format!("read-quorum-{drive_count}-{parity}");
-            let object = "uncached-object";
-            let body = b"erasure read quorum must follow the persisted layout".repeat(32_768);
-            store
-                .make_bucket(&bucket, &MakeBucketOptions::default())
-                .await
-                .expect("healthy namespace should accept bucket creation");
-            store
-                .put_object(&bucket, object, &mut PutObjReader::from_vec(body.clone()), &ObjectOptions::default())
-                .await
-                .expect("healthy erasure set should accept the seed object");
-            let set = &store.pools[0].disk_set[0];
-            let lock = set
-                .new_ns_lock(&bucket, object)
-                .await
-                .expect("seed namespace lock should resolve");
-            drop(
-                lock.get_write_lock(Duration::from_secs(30))
-                    .await
-                    .expect("seed physical fanout must finish before taking disks offline"),
-            );
-            if (drive_count, parity) == (6, 3) {
-                let mut kvs = rustfs_config::server_config::KVS::new();
-                kvs.insert(crate::config::storageclass::CLASS_STANDARD.to_string(), "EC:2".to_string());
-                set.set_test_storage_class_config(
-                    crate::config::storageclass::lookup_config_for_pools_without_env(&kvs, &[drive_count])
-                        .expect("a later storage-class change must not raise old objects' read quorum"),
-                );
-            }
-
-            let offline_indexes = (0..parity).collect::<Vec<_>>();
-            let offline = take_set_disks_offline(&store, set, &offline_indexes).await;
-            let info = store
-                .get_bucket_info(&bucket, &BucketOptions::default())
-                .await
-                .expect("bucket validation must admit the object's exact read quorum");
-            assert_eq!(info.name, bucket);
-
-            let mut reader = store
-                .get_object_reader(&bucket, object, None, Default::default(), &ObjectOptions::default())
-                .await
-                .expect("the persisted layout should remain readable at its exact data-shard quorum");
-            let mut restored = Vec::new();
-            reader
-                .stream
-                .read_to_end(&mut restored)
-                .await
-                .expect("quorum read should reconstruct the body");
-            assert_eq!(restored, body, "layout {drive_count}/{parity} must retain exact object contents");
-            drop(reader);
-
-            if drive_count - parity == drive_count / 2 {
-                let error = store
-                    .get_bucket_info_from_sets(&bucket, &BucketOptions::default())
-                    .await
-                    .expect_err("bucket mutations must retain their majority namespace check");
-                assert_eq!(error, StorageError::ErasureWriteQuorum);
-            }
-
-            let below_quorum = take_set_disks_offline(&store, set, &[parity]).await;
-            let read = store
-                .get_object_reader(&bucket, object, None, Default::default(), &ObjectOptions::default())
-                .await;
-            match read {
-                Ok(mut reader) => assert!(
-                    reader.stream.read_to_end(&mut Vec::new()).await.is_err(),
-                    "layout {drive_count}/{parity} must reject fewer than its data-shard quorum"
-                ),
-                Err(error) => assert!(
-                    matches!(error, StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _)),
-                    "a missing shard must report read quorum loss, got {error}"
-                ),
-            }
-            restore_set_disks(&store, set, below_quorum).await;
-            restore_set_disks(&store, set, offline).await;
-        }
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn bucket_info_read_quorum_is_scoped_to_each_erasure_set() {
-        let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4, 6], None).await;
-        metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
-        let bucket = "read-quorum-mixed-pools";
-        store
-            .make_bucket(bucket, &MakeBucketOptions::default())
-            .await
-            .expect("healthy pools should accept bucket creation");
-
-        let first_set = &store.pools[0].disk_set[0];
-        let second_set = &store.pools[1].disk_set[0];
-        let first_offline = take_set_disks_offline(&store, first_set, &[0, 1]).await;
-        let second_offline = take_set_disks_offline(&store, second_set, &[0, 1, 2]).await;
-        store
-            .get_bucket_info(bucket, &BucketOptions::default())
-            .await
-            .expect("each set independently satisfies its namespace read quorum");
-
-        for (set, extra_disk) in [(first_set, 2), (second_set, 3)] {
-            let extra_offline = take_set_disks_offline(&store, set, &[extra_disk]).await;
-            assert_eq!(
-                store
-                    .get_bucket_info(bucket, &BucketOptions::default())
-                    .await
-                    .expect_err("another pool must not subsidize a set below its read quorum"),
-                StorageError::ErasureReadQuorum
-            );
-            restore_set_disks(&store, set, extra_offline).await;
-        }
-        restore_set_disks(&store, first_set, first_offline).await;
-        restore_set_disks(&store, second_set, second_offline).await;
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn bucket_info_read_quorum_requires_authoritative_metadata() {
-        for state in ["missing", "corrupt", "foreign", "incarnation"] {
-            let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4], None).await;
-            metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
-            let bucket = format!("read-quorum-{state}-metadata");
-            let mut metadata = if state == "missing" {
-                store
-                    .make_bucket_on_sets(&bucket, &MakeBucketOptions::default())
-                    .await
-                    .expect("simulate directories left before bucket metadata is published");
-                BucketMetadata::new(&bucket)
-            } else {
-                store
-                    .make_bucket(&bucket, &MakeBucketOptions::default())
-                    .await
-                    .expect("healthy bucket should publish metadata");
-                metadata_sys::get_in(&store.ctx, &bucket)
-                    .await
-                    .expect("seed metadata should be cached")
-                    .as_ref()
-                    .clone()
-            };
-            let path = metadata.save_file_path();
-            match state {
-                "corrupt" => crate::config::com::save_config(store.clone(), &path, b"corrupt".to_vec())
-                    .await
-                    .expect("persist corrupt metadata while the cached copy remains valid"),
-                "foreign" => {
-                    metadata.name = "different-bucket".to_string();
-                    let mut encoded = vec![1, 0, 1, 0];
-                    encoded.extend(metadata.marshal_msg().expect("foreign metadata should encode"));
-                    crate::config::com::save_config(store.clone(), &path, encoded)
-                        .await
-                        .expect("persist metadata for a different bucket at the requested path");
-                }
-                "incarnation" => crate::bucket::metadata::save_bucket_incarnation(store.clone(), &bucket, Uuid::new_v4())
-                    .await
-                    .expect("persist a different bucket generation"),
-                _ => {}
-            }
-
-            let set = &store.pools[0].disk_set[0];
-            let offline = take_set_disks_offline(&store, set, &[0, 1]).await;
-            let error = store
-                .get_bucket_info(&bucket, &BucketOptions::default())
-                .await
-                .expect_err("read admission must not trust residual directories or cached metadata");
-            match state {
-                "missing" => assert_eq!(error, StorageError::ErasureReadQuorum),
-                "foreign" => assert_eq!(error, StorageError::FileCorrupt),
-                "incarnation" => assert!(error.to_string().contains("sidecar does not match bucket metadata")),
-                "corrupt" => assert!(error.to_string().contains("format invalid"), "unexpected corruption error: {error}"),
-                _ => unreachable!(),
-            }
-            restore_set_disks(&store, set, offline).await;
-        }
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn bucket_info_read_quorum_accepts_persisted_legacy_metadata() {
-        let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4], None).await;
-        metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
-        let bucket = "interop";
-        store
-            .make_bucket_on_sets(bucket, &MakeBucketOptions::default())
-            .await
-            .expect("legacy bucket directories should exist");
-        let hex = include_str!("../../tests/fixtures/minio/bucket_metadata.blob.hex")
-            .split_whitespace()
-            .collect::<String>();
-        let body = (0..hex.len())
-            .step_by(2)
-            .map(|index| u8::from_str_radix(&hex[index..index + 2], 16).expect("pinned MinIO metadata fixture"))
-            .collect();
-        crate::config::com::save_config(store.clone(), &BucketMetadata::new(bucket).save_file_path(), body)
-            .await
-            .expect("legacy metadata should be persisted without an incarnation sidecar");
-
-        let set = &store.pools[0].disk_set[0];
-        let offline = take_set_disks_offline(&store, set, &[0, 1]).await;
-        let info = store
-            .get_bucket_info(bucket, &BucketOptions::default())
-            .await
-            .expect("persisted MinIO metadata should authorize reads at the namespace read quorum");
-        assert_eq!(info.name, bucket);
-        assert!(info.versioning);
-        assert!(info.object_locking);
-        restore_set_disks(&store, set, offline).await;
-    }
-
    #[tokio::test]
    #[serial]
    async fn bucket_namespace_reads_report_missing_when_every_set_is_absent() {
@@ -2359,7 +2090,6 @@ mod tests {
    #[serial]
    async fn bucket_namespace_reads_fail_closed_when_any_set_loses_quorum() {
        let (_temp_dir, ecstore) = setup_multi_pool_bucket_test_env().await;
-        metadata_sys::init_bucket_metadata_sys(ecstore.clone(), Vec::new()).await;
        let bucket = format!("degraded-expansion-{}", Uuid::new_v4().simple());
        ecstore.pools[0].disk_set[0]
            .make_bucket(&bucket, &MakeBucketOptions::default())
@@ -2367,7 +2097,6 @@ mod tests {
            .expect("bucket should be created in the original pool only");
        ecstore.pools[1].disk_set[0].disks.write().await[0] = None;
        ecstore.pools[1].disk_set[0].disks.write().await[1] = None;
-        ecstore.pools[1].disk_set[0].disks.write().await[2] = None;

        let list_err = ecstore
            .list_bucket(&BucketOptions::default())
@@ -2379,7 +2108,7 @@ mod tests {
            .get_bucket_info(&bucket, &BucketOptions::default())
            .await
            .expect_err("bucket validation must fail when an expansion pool is unavailable");
-        assert_eq!(info_err, StorageError::ErasureReadQuorum);
+        assert_eq!(info_err, StorageError::ErasureWriteQuorum);
    }

    #[tokio::test]
@@ -577,16 +577,13 @@ impl KmsServiceManager {
        Some(service_version.probe_worker.as_ref()?.status())
    }

-    /// Check backend health without changing the service lifecycle state.
-    ///
-    /// A transient backend failure leaves the published service available for
-    /// subsequent checks and operations. Readiness uses the background probe
-    /// to evaluate backend availability independently of lifecycle state.
+    /// Health check for the KMS service
    pub async fn health_check(&self) -> Result<bool> {
        let checked_state = self.state.load_full();
        match checked_state.current_service.as_ref() {
            Some(service_version) => {
                let manager = service_version.manager.clone();
+                let checked_version = service_version.version;
                // Perform health check on the backend
                match manager.health_check().await {
                    Ok(healthy) => {
@@ -597,6 +594,8 @@ impl KmsServiceManager {
                    }
                    Err(e) => {
                        error!("KMS health check error: {}", e);
+                        let _guard = self.lifecycle_mutex.lock().await;
+                        self.mark_health_error_if_current(checked_version, &e);
                        Err(e)
                    }
                }
@@ -740,6 +739,17 @@ impl KmsServiceManager {
            task: std::sync::Mutex::new(Some(task)),
        }))
    }
+
+    fn mark_health_error_if_current(&self, checked_version: u64, error: &KmsError) {
+        let current = self.state.load_full();
+        if current.current_service.as_ref().map(|version| version.version) == Some(checked_version) {
+            self.state.store(Arc::new(RuntimeState {
+                config: current.config.clone(),
+                status: KmsServiceStatus::Error(format!("Health check failed: {error}")),
+                current_service: current.current_service.clone(),
+            }));
+        }
+    }
 }

 impl Default for KmsServiceManager {
@@ -994,6 +1004,19 @@ mod tests {
        assert!(manager.get_service_version().await.expect("restarted version") > first_version);
    }

+    #[tokio::test]
+    async fn stale_health_failure_cannot_poison_new_service_status() {
+        let manager = KmsServiceManager::new();
+        manager.configure(static_config("key-a", 0x11)).await.expect("configure");
+        manager.start().await.expect("start");
+        let old_version = manager.get_service_version().await.expect("old version");
+        manager.restart().await.expect("restart");
+
+        manager.mark_health_error_if_current(old_version, &KmsError::backend_error("stale failure"));
+
+        assert_eq!(manager.get_status().await, KmsServiceStatus::Running);
+    }
+
    #[tokio::test]
    async fn forbidden_local_master_key_change_preserves_running_config_and_service() {
        use crate::types::{CreateKeyRequest, KeyUsage};
@@ -75,44 +75,6 @@ fn unreachable_vault_config() -> KmsConfig {
    }
 }

-#[tokio::test]
-async fn transient_health_failure_does_not_latch_the_service_status() {
-    let kms = TestKms::local().await;
-    let manager = kms.manager();
-    let service = manager.get_encryption_service().await.expect("running service");
-    let version = manager.get_service_version().await.expect("running version");
-    assert!(manager.health_check().await.expect("initial backend health"));
-
-    // Move only this test's keys out of reach, then restore the same backend.
-    let key_dir = kms.key_dir().expect("local key directory");
-    let outage = tempfile::TempDir::new().expect("temporary outage directory");
-    let hidden_keys = outage.path().join("keys");
-    tokio::fs::rename(&key_dir, &hidden_keys)
-        .await
-        .expect("make backend unavailable");
-    let failure = manager.health_check().await;
-    let outage_status = manager.get_status().await;
-    tokio::fs::rename(&hidden_keys, &key_dir).await.expect("restore backend");
-
-    assert!(failure.is_err(), "the outage must surface as a health-check error");
-    assert!(manager.health_check().await.expect("backend recovers without restart"));
-    assert!(Arc::ptr_eq(
-        &service,
-        &manager.get_encryption_service().await.expect("service survives the outage")
-    ));
-    assert_eq!(manager.get_service_version().await, Some(version));
-    assert_eq!(
-        manager.get_status().await,
-        KmsServiceStatus::Running,
-        "a recovered backend must not leave service-status and readiness latched in Error"
-    );
-    assert_eq!(
-        outage_status,
-        KmsServiceStatus::Running,
-        "backend health does not change the running service's lifecycle state"
-    );
-}
-
 #[tokio::test]
 async fn starting_against_an_unreachable_backend_fails_without_publishing_a_service() {
    let manager = KmsServiceManager::new();
@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+"""Build an identified E2E server and verify it around one test invocation."""
+
+import argparse
+from contextlib import contextmanager
+import hashlib
+import json
+import os
+from pathlib import Path
+import stat
+import signal
+import subprocess
+import sys
+import tempfile
+
+ROOT = Path(__file__).resolve().parent.parent
+RECEIPT_ENV = "RUSTFS_E2E_BINARY_RECEIPT"
+
+
+def feature_set(value):
+    return sorted(set(part.strip() for part in value.split(",") if part.strip()))
+
+
+def file_hash(path):
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def source_identity():
+    head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
+    tracked = subprocess.check_output(["git", "ls-files", "--cached", "--others", "--exclude-standard", "-z"], cwd=ROOT)
+    paths = set(tracked.decode("utf-8").rstrip("\0").split("\0")) - {""}
+    # RustEmbed consumes ignored console assets as well as tracked Rust sources.
+    static_dir = ROOT / "rustfs/static"
+    if static_dir.is_symlink():
+        raise ValueError("The embedded static directory must not be a symlink")
+    if static_dir.is_dir():
+        for path in static_dir.rglob("*"):
+            if path.is_symlink() and path.is_dir():
+                raise ValueError(f"Unsupported embedded directory symlink: {path}")
+            if not path.is_dir():
+                paths.add(str(path.relative_to(ROOT)))
+    elif static_dir.exists():
+        paths.add("rustfs/static")
+    digest = hashlib.sha256()
+    digest.update(b"static-present\0" if static_dir.is_dir() else b"static-absent\0")
+    for name in sorted(paths):
+        path = ROOT / name
+        digest.update(name.encode("utf-8") + b"\0")
+        try:
+            metadata = path.lstat()
+        except FileNotFoundError:
+            digest.update(b"deleted\0")
+            continue
+        if stat.S_ISLNK(metadata.st_mode):
+            digest.update(b"symlink\0" + os.fsencode(os.readlink(path)) + b"\0")
+            if path.is_dir():
+                target = path.resolve()
+                if ROOT not in target.parents:
+                    raise ValueError(f"Directory link escapes the source inventory: {name}")
+                # Directory aliases such as .claude/skills share already-hashed inputs.
+                for child in target.rglob("*"):
+                    if child.is_dir() and not child.is_symlink():
+                        continue
+                    if child.is_dir() or str(child.relative_to(ROOT)) not in paths:
+                        raise ValueError(f"Directory link contains an unrecorded input: {child}")
+                digest.update(b"directory\0" + str(target.relative_to(ROOT)).encode("utf-8") + b"\0")
+                continue
+        elif not stat.S_ISREG(metadata.st_mode):
+            raise ValueError(f"Unsupported build input: {name}")
+        digest.update(str(metadata.st_mode & 0o111).encode() + b"\0")
+        digest.update(file_hash(path).encode() + b"\0")
+    return {"head": head, "sha256": digest.hexdigest()}
+
+
+def sidecar_path(binary):
+    return binary.with_name(binary.name + ".e2e.json")
+
+
+def validate_target_directory(target_dir):
+    if target_dir == ROOT or target_dir in ROOT.parents:
+        raise ValueError("CARGO_TARGET_DIR must not contain the source workspace")
+    if ROOT in target_dir.parents:
+        ignored = subprocess.run(["git", "check-ignore", "--quiet", "--no-index", str(target_dir.relative_to(ROOT))], cwd=ROOT)
+        if ignored.returncode != 0:
+            raise ValueError("An in-workspace CARGO_TARGET_DIR must be Git-ignored; use target/ or an external directory")
+
+
+@contextmanager
+def exclusive_binary(binary):
+    marker = binary.with_name(binary.name + ".e2e.lock")
+    try:
+        descriptor = os.open(marker, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+    except FileExistsError as error:
+        raise ValueError(f"Another E2E build/run owns {marker}; do not share a target directory between concurrent runs") from error
+    try:
+        identity = os.fstat(descriptor)
+        with os.fdopen(descriptor, "w") as lock:
+            lock.write(f"pid={os.getpid()}\n")
+        yield
+    finally:
+        current = marker.stat()
+        if (current.st_dev, current.st_ino) != (identity.st_dev, identity.st_ino):
+            raise ValueError("The E2E ownership marker changed during the command")
+        marker.unlink()
+
+
+def terminate_command(process):
+    if process.poll() is not None:
+        return
+    try:
+        os.killpg(process.pid, signal.SIGTERM)
+    except ProcessLookupError:
+        return
+    try:
+        process.wait(timeout=5)
+    except subprocess.TimeoutExpired:
+        os.killpg(process.pid, signal.SIGKILL)
+        process.wait()
+
+
+def build(binary, target_dir, profile, requested, all_bins):
+    sidecar = sidecar_path(binary)
+    sidecar.unlink(missing_ok=True)
+    before = source_identity()
+    command = ["cargo", "build", "--locked", "-p", "rustfs", "--target-dir", str(target_dir), "--message-format=json-render-diagnostics"]
+    command.extend(["--bins"] if all_bins else ["--bin", "rustfs"])
+    if requested:
+        command.extend(["--features", ",".join(requested)])
+    if profile == "release":
+        command.append("--release")
+    artifact = None
+    with subprocess.Popen(command, cwd=ROOT, stdout=subprocess.PIPE, text=True, start_new_session=True) as process:
+        try:
+            for line in process.stdout:
+                message = json.loads(line)
+                if message.get("reason") == "compiler-message":
+                    print(message["message"].get("rendered", ""), end="", file=sys.stderr)
+                if message.get("reason") == "compiler-artifact" and message.get("target", {}).get("name") == "rustfs" and "bin" in message.get("target", {}).get("kind", []):
+                    artifact = message
+            if process.wait() != 0:
+                raise ValueError("RustFS build failed; no E2E identity was recorded")
+        except BaseException:
+            terminate_command(process)
+            raise
+    if not artifact or Path(artifact.get("executable", "")).resolve() != binary:
+        raise ValueError("Cargo did not produce the requested RustFS executable")
+    if source_identity() != before:
+        raise ValueError("Build inputs changed during compilation; finish preparing embedded assets and rebuild in an isolated worktree")
+    record = {
+        "schema": 1,
+        "source": before,
+        "requested_features": requested,
+        "features": sorted(artifact["features"]),
+        "profile": profile,
+        "rustc": subprocess.check_output(["rustc", "-Vv"], text=True),
+        "binary_sha256": file_hash(binary),
+    }
+    sidecar.write_text(json.dumps(record, sort_keys=True) + "\n")
+    print(f"Built E2E server: {binary}\nIdentity: {sidecar}", file=sys.stderr)
+
+
+def verify(binary, profile, requested):
+    record = json.loads(sidecar_path(binary).read_text())
+    if not isinstance(record, dict) or set(record) != {"schema", "source", "requested_features", "features", "profile", "rustc", "binary_sha256"} or type(record["schema"]) is not int or record["schema"] != 1:
+        raise ValueError("Missing or unsupported E2E binary identity; run the build command")
+    if not isinstance(record["rustc"], str) or not record["rustc"].strip():
+        raise ValueError("Missing E2E build toolchain identity")
+    if record["requested_features"] != requested or record["profile"] != profile:
+        raise ValueError("E2E binary build features/profile differ from this test invocation")
+    if not isinstance(record["features"], list) or not all(isinstance(item, str) for item in record["features"]) or not set(requested) <= set(record["features"]):
+        raise ValueError("Invalid resolved E2E binary features")
+    if record["source"] != source_identity():
+        raise ValueError("E2E binary was built from different inputs; rebuild before testing")
+    if record["binary_sha256"] != file_hash(binary):
+        raise ValueError("E2E binary content differs from its build identity")
+    return record
+
+
+def run(binary, profile, requested, command):
+    if not command:
+        raise ValueError("run requires a test command after --")
+    override = os.environ.get("CARGO_BIN_EXE_rustfs")
+    if override and Path(override).resolve() != binary:
+        raise ValueError("CARGO_BIN_EXE_rustfs selects a different server; use --binary explicitly")
+    record = verify(binary, profile, requested)
+    metadata = binary.stat()
+    with tempfile.TemporaryDirectory(prefix="rustfs-e2e-receipt-") as directory:
+        receipt = Path(directory) / "receipt.json"
+        receipt.write_text(json.dumps({
+            "schema": 1,
+            "workspace": str(ROOT),
+            "binary": str(binary),
+            "size": metadata.st_size,
+            "modified_ns": metadata.st_mtime_ns,
+            "features": record["features"],
+        }))
+        env = dict(os.environ, CARGO_BIN_EXE_rustfs=str(binary), RUSTFS_BUILD_FEATURES=",".join(record["features"]))
+        env[RECEIPT_ENV] = str(receipt)
+        with subprocess.Popen(command, cwd=ROOT, env=env, start_new_session=True) as process:
+            try:
+                status = process.wait()
+            except (KeyboardInterrupt, SystemExit):
+                terminate_command(process)
+                raise
+        try:
+            if verify(binary, profile, requested) != record:
+                raise ValueError("E2E build identity changed during testing")
+        except (OSError, ValueError, subprocess.SubprocessError) as error:
+            print(f"E2E validation invalidated: {error}", file=sys.stderr)
+            return status if status else 1
+        return status
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("mode", choices=("build", "run"))
+    parser.add_argument("--features", default="", help="additional Cargo features; defaults remain enabled")
+    parser.add_argument("--profile", choices=("debug", "release"), default="debug")
+    parser.add_argument("--binary", type=Path, help="prebuilt server path for run")
+    parser.add_argument("--bins", action="store_true", help="build all RustFS binary targets, preserving the CI build matrix")
+    # Parse the child command separately so its options are never interpreted here.
+    args = sys.argv[1:]
+    separator = args.index("--") if "--" in args else len(args)
+    command = args[separator + 1:] if separator < len(args) else []
+    options = parser.parse_args(args[:separator])
+    target_dir = Path(os.environ.get("CARGO_TARGET_DIR", ROOT / "target")).resolve()
+    binary = (options.binary or target_dir / options.profile / ("rustfs.exe" if os.name == "nt" else "rustfs")).resolve()
+    try:
+        validate_target_directory(target_dir)
+        requested = feature_set(options.features)
+        if options.mode == "build":
+            binary.parent.mkdir(parents=True, exist_ok=True)
+        with exclusive_binary(binary):
+            if options.mode == "build":
+                if options.binary or command:
+                    raise ValueError("build does not accept --binary or a child command")
+                build(binary, target_dir, options.profile, requested, options.bins)
+                return 0
+            if options.bins:
+                raise ValueError("--bins is a build option")
+            return run(binary, options.profile, requested, command)
+    except (OSError, ValueError, subprocess.SubprocessError) as error:
+        print(f"E2E prerequisite failed: {error}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    signal.signal(signal.SIGTERM, lambda signum, frame: sys.exit(128 + signum))
+    raise SystemExit(main())
@@ -1,87 +0,0 @@
-#!/usr/bin/env bash
-# Publish the nightly DEB/RPM as assets of the rolling `nightly` release on
-# rustfs/auto-testing, replacing the previous build's files in place.
-#
-# Required environment:
-#   ASSETS_TOKEN  token with contents:write on rustfs/auto-testing
-#   DEB_FILE      path to the built .deb
-#   RPM_FILE      path to the built .rpm
-#   DEB_DATE      build date (YYYY-MM-DD)
-#   BUILD_REF     branch/ref the nightly was built from
-# Optional environment:
-#   GITHUB_SHA / GITHUB_RUN_ID / GITHUB_REPOSITORY / GITHUB_SERVER_URL
-#
-# Plain curl + python3 by design: the nightly build fleet has no gh CLI.
-
-set -euo pipefail
-
-: "${ASSETS_TOKEN:?ASSETS_TOKEN is required}"
-: "${DEB_FILE:?DEB_FILE is required}"
-: "${RPM_FILE:?RPM_FILE is required}"
-: "${DEB_DATE:?DEB_DATE is required}"
-: "${BUILD_REF:?BUILD_REF is required}"
-
-for f in "${DEB_FILE}" "${RPM_FILE}"; do
-  [ -f "$f" ] || { echo "missing package: $f" >&2; exit 1; }
-done
-
-API="https://api.github.com/repos/rustfs/auto-testing"
-UPLOADS="https://uploads.github.com/repos/rustfs/auto-testing/releases"
-AUTH="Authorization: token ${ASSETS_TOKEN}"
-SOURCE_SHA="$(git rev-parse HEAD 2>/dev/null || echo "${GITHUB_SHA:-unknown}")"
-
-release_id="$(curl -fsS --retry 3 -H "${AUTH}" "${API}/releases/tags/nightly" \
-  | python3 -c 'import json,sys; print(json.load(sys.stdin).get("id", ""))' 2>/dev/null || true)"
-if [ -z "${release_id}" ]; then
-  echo "creating the rolling nightly release"
-  release_id="$(curl -fsS --retry 3 -X POST -H "${AUTH}" -H "Content-Type: application/json" \
-    -d '{"tag_name":"nightly","name":"Nightly builds","body":"Rolling nightly builds. Assets are replaced on every build; the release body documents the provenance of the current files."}' \
-    "${API}/releases" | python3 -c 'import json,sys; print(json.load(sys.stdin)["id"])')"
-fi
-[ -n "${release_id}" ] || { echo "could not resolve the nightly release id" >&2; exit 1; }
-
-upload_asset() {
-  local name="$1" file="$2" asset_id
-  asset_id="$(curl -fsS --retry 3 -H "${AUTH}" "${API}/releases/tags/nightly" \
-    | ASSET_NAME="${name}" python3 -c '
-import json, sys, os
-d = json.load(sys.stdin)
-name = os.environ["ASSET_NAME"]
-print(next((a["id"] for a in d.get("assets", []) if a["name"] == name), ""))')"
-  if [ -n "${asset_id}" ]; then
-    curl -fsS --retry 3 -X DELETE -H "${AUTH}" "${API}/releases/assets/${asset_id}" >/dev/null
-  fi
-  curl -fsS --retry 3 --max-time 900 -X POST \
-    -H "${AUTH}" -H "Content-Type: application/octet-stream" \
-    --data-binary "@${file}" \
-    "${UPLOADS}/${release_id}/assets?name=${name}" >/dev/null
-  echo "uploaded ${name}"
-}
-
-upload_asset "rustfs-nightly-latest.deb" "${DEB_FILE}"
-upload_asset "rustfs-nightly-latest.rpm" "${RPM_FILE}"
-
-DEB_SHA="$(sha256sum "${DEB_FILE}" | cut -d ' ' -f 1)"
-RPM_SHA="$(sha256sum "${RPM_FILE}" | cut -d ' ' -f 1)"
-RUN_URL="${GITHUB_SERVER_URL:-https://github.com}/${GITHUB_REPOSITORY:-/rustfs/rustfs}/actions/runs/${GITHUB_RUN_ID:-0}"
-export BUILD_REF SOURCE_SHA DEB_DATE DEB_FILE RPM_FILE DEB_SHA RPM_SHA RUN_URL
-
-python3 - << 'PY' > /tmp/release-body.json
-import json, os
-e = os.environ
-deb_mb = os.path.getsize(e["DEB_FILE"]) // 1048576
-rpm_mb = os.path.getsize(e["RPM_FILE"]) // 1048576
-body = (
-    f"Nightly build from `{e['BUILD_REF']}@{e['SOURCE_SHA'][:12]}`, built {e['DEB_DATE']}.\n\n"
-    f"[Build run]({e['RUN_URL']}). The `latest` assets are replaced in place on every nightly.\n\n"
-    f"| Asset | Size | SHA256 |\n|---|---|---|\n"
-    f"| rustfs-nightly-latest.deb (={e['DEB_FILE']}) | {deb_mb} MB | `{e['DEB_SHA']}` |\n"
-    f"| rustfs-nightly-latest.rpm (={e['RPM_FILE']}) | {rpm_mb} MB | `{e['RPM_SHA']}` |\n"
-)
-print(json.dumps({"body": body}))
-PY
-
-curl -fsS --retry 3 -X PATCH -H "${AUTH}" -H "Content-Type: application/json" \
-  --data-binary @/tmp/release-body.json "${API}/releases/${release_id}" >/dev/null
-
-echo "published ${DEB_FILE} and ${RPM_FILE} to the rustfs/auto-testing 'nightly' release"
@@ -14,7 +14,12 @@ NC='\033[0m' # No Color

 # Default values
 PROJECT_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
-TARGET_DIR="$PROJECT_ROOT/target/debug"
+CARGO_TARGET_DIR="${CARGO_TARGET_DIR:-$PROJECT_ROOT/target}"
+if [[ "$CARGO_TARGET_DIR" != /* ]]; then
+    CARGO_TARGET_DIR="$PROJECT_ROOT/$CARGO_TARGET_DIR"
+fi
+export CARGO_TARGET_DIR
+TARGET_DIR="$CARGO_TARGET_DIR/debug"
 RUSTFS_BINARY="$TARGET_DIR/rustfs"
 DATA_DIR="$TARGET_DIR/rustfs_test_data"
 RUSTFS_PID=""
@@ -94,7 +99,7 @@ build_rustfs() {
    print_info "Building RustFS..."
    cd "$PROJECT_ROOT"
    
-    if ! cargo build --bin rustfs --features "$RUSTFS_BUILD_FEATURES"; then
+    if ! python3 scripts/e2e_binary.py build --features "$RUSTFS_BUILD_FEATURES"; then
        print_error "Failed to build RustFS"
        exit 1
    fi
@@ -115,6 +120,10 @@ check_dependencies() {
        missing_tools+=("curl")
    fi
    
+    if ! command -v python3 >/dev/null 2>&1; then
+        missing_tools+=("python3")
+    fi
+
    if ! command -v cargo >/dev/null 2>&1; then
        missing_tools+=("cargo")
    fi
@@ -203,7 +212,7 @@ run_tests() {

    print_info "Test command: ${test_cmd[*]}"

-    if "${test_cmd[@]}"; then
+    if python3 scripts/e2e_binary.py run --features "$RUSTFS_BUILD_FEATURES" -- "${test_cmd[@]}"; then
        print_success "All tests passed!"
        return 0
    else
@@ -243,9 +243,10 @@ run_quick_e2e_steps() {
    return
  fi

-  run_step "e2e-reliability-disk-fault" cargo test --package e2e_test reliability_disk_fault_test -- --nocapture
-  run_step "e2e-heal-erasure-disk-rebuild" cargo test --package e2e_test heal_erasure_disk_rebuild_test -- --nocapture
-  run_step "e2e-namespace-lock-quorum" cargo test --package e2e_test namespace_lock_quorum_test -- --nocapture
+  run_step "build-e2e-server" python3 scripts/e2e_binary.py build
+  run_step "e2e-reliability-disk-fault" python3 scripts/e2e_binary.py run -- cargo test --package e2e_test reliability_disk_fault_test -- --nocapture
+  run_step "e2e-heal-erasure-disk-rebuild" python3 scripts/e2e_binary.py run -- cargo test --package e2e_test heal_erasure_disk_rebuild_test -- --nocapture
+  run_step "e2e-namespace-lock-quorum" python3 scripts/e2e_binary.py run -- cargo test --package e2e_test namespace_lock_quorum_test -- --nocapture
 }

 run_quick_profile() {
@@ -313,15 +314,15 @@ write_blackbox_matrix() {

 	  {
 	    printf 'profile\tscenario\tgate\tcommand\tfixture_env\tstatus\n'
-	    printf 'quick\tsingle-node disk fault read/write\tblack-box\tcargo test --package e2e_test reliability_disk_fault_test -- --nocapture\tnone\t%s\n' "$e2e_status"
-	    printf 'quick\theal degraded erasure disk rebuild\tblack-box\tcargo test --package e2e_test heal_erasure_disk_rebuild_test -- --nocapture\tnone\t%s\n' "$e2e_status"
-	    printf 'quick\tnamespace lock quorum under EC ops\tblack-box\tcargo test --package e2e_test namespace_lock_quorum_test -- --nocapture\tnone\t%s\n' "$e2e_status"
+	    printf 'quick\tsingle-node disk fault read/write\tblack-box\tpython3 scripts/e2e_binary.py run -- cargo test --package e2e_test reliability_disk_fault_test -- --nocapture\tnone\t%s\n' "$e2e_status"
+	    printf 'quick\theal degraded erasure disk rebuild\tblack-box\tpython3 scripts/e2e_binary.py run -- cargo test --package e2e_test heal_erasure_disk_rebuild_test -- --nocapture\tnone\t%s\n' "$e2e_status"
+	    printf 'quick\tnamespace lock quorum under EC ops\tblack-box\tpython3 scripts/e2e_binary.py run -- cargo test --package e2e_test namespace_lock_quorum_test -- --nocapture\tnone\t%s\n' "$e2e_status"
 	    printf 'full\tlegacy bitrot read fixture restore\tfixture\tcargo test -p rustfs-ecstore --test legacy_bitrot_read_test -- --nocapture\tRUSTFS_LEGACY_TEST_ROOT,RUSTFS_LEGACY_TEST_DISK\t%s\n' "$legacy_status"
 	    printf 'full\tMinIO generated encrypted read and negative restore fixture\tfixture\tcargo test -p rustfs --features rio-v2 storage::minio_generated_read_test --lib -- --ignored --nocapture\tRUSTFS_MINIO_FIXTURE_ROOT,RUSTFS_MINIO_STATIC_KMS_KEY_B64\t%s\n' "$minio_status"
 	    printf 'full\tS3 multipart range versioning delete subset\tblack-box\tenv TESTEXPR=\"multipart or range or versioning or delete\" DEPLOY_MODE=build MAXFAIL=0 ./scripts/s3-tests/run.sh\tnone\t%s\n' "$s3_status"
-	    printf 'destructive\tdistributed cluster concurrency\tblack-box\tcargo test --package e2e_test cluster_concurrency_test -- --nocapture\tnone\t%s\n' "$destructive_status"
-	    printf 'destructive\tstale multipart cleanup cluster\tblack-box\tcargo test --package e2e_test stale_multipart_cleanup_cluster_test -- --nocapture\tnone\t%s\n' "$destructive_status"
-	    printf 'destructive\tdelete marker migration semantics\tblack-box\tcargo test --package e2e_test delete_marker_migration_semantics_test -- --nocapture\tnone\t%s\n' "$destructive_status"
+	    printf 'destructive\tdistributed cluster concurrency\tblack-box\tpython3 scripts/e2e_binary.py run -- cargo test --package e2e_test cluster_concurrency_test -- --nocapture\tnone\t%s\n' "$destructive_status"
+	    printf 'destructive\tstale multipart cleanup cluster\tblack-box\tpython3 scripts/e2e_binary.py run -- cargo test --package e2e_test stale_multipart_cleanup_cluster_test -- --nocapture\tnone\t%s\n' "$destructive_status"
+	    printf 'destructive\tdelete marker migration semantics\tblack-box\tpython3 scripts/e2e_binary.py run -- cargo test --package e2e_test delete_marker_migration_semantics_test -- --nocapture\tnone\t%s\n' "$destructive_status"
 	  } >"$BLACKBOX_MATRIX"
 }

@@ -566,9 +567,9 @@ run_destructive_profile() {
    return
  fi

-  run_step "e2e-cluster-concurrency" cargo test --package e2e_test cluster_concurrency_test -- --nocapture
-  run_step "e2e-stale-multipart-cleanup-cluster" cargo test --package e2e_test stale_multipart_cleanup_cluster_test -- --nocapture
-  run_step "e2e-delete-marker-migration-semantics" cargo test --package e2e_test delete_marker_migration_semantics_test -- --nocapture
+  run_step "e2e-cluster-concurrency" python3 scripts/e2e_binary.py run -- cargo test --package e2e_test cluster_concurrency_test -- --nocapture
+  run_step "e2e-stale-multipart-cleanup-cluster" python3 scripts/e2e_binary.py run -- cargo test --package e2e_test stale_multipart_cleanup_cluster_test -- --nocapture
+  run_step "e2e-delete-marker-migration-semantics" python3 scripts/e2e_binary.py run -- cargo test --package e2e_test delete_marker_migration_semantics_test -- --nocapture
 }

 run_fuzz_profile() {
@@ -0,0 +1,314 @@
+#!/usr/bin/env python3
+"""Exercise the E2E build/run boundary without compiling RustFS."""
+
+import json
+import os
+from pathlib import Path
+import shutil
+import signal
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+class BinaryProvenanceTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        (self.root / "scripts").mkdir()
+        shutil.copy(Path(__file__).with_name("e2e_binary.py"), self.root / "scripts/e2e_binary.py")
+        (self.root / "Cargo.toml").write_text("[workspace]\n")
+        (self.root / "source.rs").write_text("original source\n")
+        (self.root / ".gitignore").write_text("/target/\n/rustfs/static/\n")
+        (self.root / ".agents/skills").mkdir(parents=True)
+        (self.root / ".agents/skills/SKILL.md").write_text("tracked instructions\n")
+        (self.root / ".claude").mkdir()
+        (self.root / ".claude/skills").symlink_to("../.agents/skills", target_is_directory=True)
+        subprocess.run(["git", "init", "-q", str(self.root)], check=True)
+        for args in (["add", "."], ["-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-qm", "fixture"]):
+            subprocess.run(["git", "-C", str(self.root), *args], check=True)
+        self.commands = self.root / "target/commands"
+        self.commands.mkdir(parents=True)
+        cargo = self.commands / "cargo"
+        cargo.write_text(f"#!{sys.executable}\n" + '''import json, os, pathlib, sys
+if os.environ.get("FAKE_BUILD_FAIL"):
+    raise SystemExit(23)
+args = sys.argv[1:]
+if args[:2] == ["nextest", "run"]:
+    receipt = json.loads(pathlib.Path(os.environ["RUSTFS_E2E_BINARY_RECEIPT"]).read_text())
+    assert pathlib.Path(receipt["binary"]) == pathlib.Path(os.environ["CARGO_BIN_EXE_rustfs"]).resolve()
+    if os.environ.get("RUSTFS_E2E_STARTUP_CAS_BINARY"):
+        assert pathlib.Path(receipt["binary"]) == pathlib.Path(os.environ["RUSTFS_E2E_STARTUP_CAS_BINARY"]).resolve()
+    pathlib.Path("target/nextest-command.json").write_text(json.dumps(args))
+    raise SystemExit(int(os.environ.get("FAKE_TEST_EXIT", "0")))
+target = pathlib.Path(args[args.index("--target-dir") + 1])
+binary = target / ("release" if "--release" in args else "debug") / "rustfs"
+binary.parent.mkdir(parents=True, exist_ok=True)
+binary.write_text("#!/bin/sh\\nexit 0\\n")
+binary.chmod(0o755)
+features = ["default", "ftps", "webdav"]
+if "--features" in args:
+    features.extend(args[args.index("--features") + 1].split(","))
+if "full" in features:
+    features.extend(["sftp", "swift", "metrics-gpu", "pyroscope"])
+print(json.dumps({"reason": "compiler-artifact", "target": {"name": "rustfs", "kind": ["bin"]}, "executable": str(binary), "features": sorted(set(features))}))
+if os.environ.get("FAKE_BUILD_MUTATE"):
+    pathlib.Path("source.rs").write_text("changed during build")
+''')
+        cargo.chmod(0o755)
+        rustc = self.commands / "rustc"
+        rustc.write_text("#!/bin/sh\nprintf 'rustc fixture\\nhost: fixture\\n'\n")
+        rustc.chmod(0o755)
+        self.env = dict(os.environ, PATH=f"{self.commands}{os.pathsep}{os.environ['PATH']}")
+        for name in ("CARGO_TARGET_DIR", "CARGO_BIN_EXE_rustfs", "RUSTFS_BUILD_FEATURES", "RUSTFS_E2E_BINARY_RECEIPT"):
+            self.env.pop(name, None)
+        self.binary = self.root / "target/debug/rustfs"
+        self.sidecar = self.binary.with_name("rustfs.e2e.json")
+
+    def invoke(self, *args, env=None):
+        return subprocess.run([sys.executable, str(self.root / "scripts/e2e_binary.py"), *args], cwd=self.root, env=env or self.env, text=True, capture_output=True)
+
+    def build(self, features=""):
+        result = self.invoke("build", "--features", features)
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def run_code(self, code="pass", features="", env=None):
+        return self.invoke("run", "--features", features, "--", sys.executable, "-c", code, env=env)
+
+    def test_build_run_and_receipt_cleanup(self):
+        self.build("full,e2e-test-hooks")
+        result = self.run_code("import os,pathlib; print(os.environ['RUSTFS_E2E_BINARY_RECEIPT']); assert pathlib.Path(os.environ['CARGO_BIN_EXE_rustfs']).is_file(); assert 'sftp' in os.environ['RUSTFS_BUILD_FEATURES']", "e2e-test-hooks,full")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertFalse(Path(result.stdout.strip()).exists(), "run receipts must not survive their command")
+        self.assertIn("sftp", json.loads(self.sidecar.read_text())["features"])
+
+    def test_source_changes_are_not_hidden_by_timestamps_or_head(self):
+        self.build()
+        path = self.root / "source.rs"
+        old = path.stat()
+        path.write_text("different bytes\n")
+        os.utime(path, ns=(old.st_atime_ns, old.st_mtime_ns))
+        self.assertNotEqual(self.run_code().returncode, 0)
+
+    def test_deleted_untracked_and_ignored_embedded_inputs(self):
+        for mutation in ("delete", "untracked", "static"):
+            with self.subTest(mutation=mutation):
+                self.build()
+                path = self.root / "source.rs"
+                if mutation == "delete":
+                    path.unlink()
+                elif mutation == "untracked":
+                    (self.root / "new.rs").write_text("new source")
+                else:
+                    static = self.root / "rustfs/static"
+                    static.mkdir(parents=True)
+                    (static / "index.html").write_text("embedded content")
+                self.assertNotEqual(self.run_code().returncode, 0)
+                path.write_text("original source\n")
+
+    def test_wrong_binary_features_and_manifest_fail_closed(self):
+        self.build("sftp")
+        self.assertNotEqual(self.run_code(features="webdav").returncode, 0)
+        self.binary.write_text("old server")
+        self.assertNotEqual(self.run_code(features="sftp").returncode, 0)
+        self.sidecar.write_text("{}")
+        self.assertNotEqual(self.run_code(features="sftp").returncode, 0)
+        self.sidecar.unlink()
+        self.assertNotEqual(self.run_code(features="sftp").returncode, 0)
+
+    def test_build_failure_or_source_race_does_not_leave_a_receipt(self):
+        for failure in ("FAKE_BUILD_FAIL", "FAKE_BUILD_MUTATE"):
+            self.build()
+            result = self.invoke("build", env=dict(self.env, **{failure: "1"}))
+            self.assertNotEqual(result.returncode, 0)
+            self.assertFalse(self.sidecar.exists())
+
+    def test_child_failure_and_changes_during_run_fail(self):
+        self.build()
+        failed = self.run_code("raise SystemExit(37)")
+        self.assertEqual(failed.returncode, 37, failed.stderr)
+        for code in ("import pathlib; pathlib.Path('source.rs').write_text('changed while testing')", "import pathlib; pathlib.Path('target/debug/rustfs').write_text('different server')"):
+            self.build()
+            self.assertNotEqual(self.run_code(code).returncode, 0)
+
+    def test_override_cannot_select_an_unverified_server(self):
+        self.build()
+        result = self.run_code(env=dict(self.env, CARGO_BIN_EXE_rustfs="/some/old/server"))
+        self.assertNotEqual(result.returncode, 0)
+
+    def test_artifact_moves_between_clean_checkouts(self):
+        self.build()
+        with tempfile.TemporaryDirectory() as destination:
+            clone = Path(destination) / "clone"
+            subprocess.run(["git", "clone", "-q", str(self.root), str(clone)], check=True)
+            (clone / "target/debug").mkdir(parents=True)
+            shutil.copy2(self.binary, clone / "target/debug/rustfs")
+            shutil.copy2(self.sidecar, clone / "target/debug/rustfs.e2e.json")
+            result = subprocess.run([sys.executable, str(clone / "scripts/e2e_binary.py"), "run", "--", sys.executable, "-c", "pass"], cwd=clone, env=self.env, text=True, capture_output=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_ci_build_preserves_both_manifests_and_runs_the_copied_server(self):
+        from check_test_wiring import yaml_block
+        from test_security_workflow import named_steps, shell_body
+
+        source = (Path(__file__).resolve().parents[1] / ".github/workflows/ci.yml").read_text().splitlines()
+        build_steps = named_steps(yaml_block(source, "build-rustfs-debug-binary", 2))
+        run_steps = named_steps(yaml_block(source, "e2e-full", 2))
+        (self.root / "Cargo.lock").write_text("fixture lock\n")
+        subprocess.run(["git", "add", "Cargo.lock"], cwd=self.root, check=True)
+        subprocess.run(["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-qm", "lock"], cwd=self.root, check=True)
+        copied = self.root / "target/startup-cas-input/rustfs"
+        env = dict(self.env, STARTUP_CAS_INPUT=str(copied.parent), RUSTFS_E2E_STARTUP_CAS_BINARY=str(copied))
+        for step in (build_steps["Build debug binary"], run_steps["Preserve startup CAS binary input"]):
+            result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", shell_body(step)], cwd=self.root, env=env, capture_output=True, text=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+        for name in ("rustfs.e2e.json", "rustfs.e2e-startup-cas-build.json"):
+            self.assertIn("            target/debug/" + name, build_steps["Upload debug binary"])
+            self.assertEqual((self.binary.parent / name).read_bytes(), (copied.parent / name).read_bytes())
+        manifest = json.loads(copied.with_name("rustfs.e2e-startup-cas-build.json").read_text())
+        self.assertEqual(manifest["argv"], ["python3", "scripts/e2e_binary.py", "build", "--bins", "--features", "e2e-test-hooks"])
+        self.assertTrue(manifest["clean_before"] and manifest["clean_after"])
+        body = next(line.removeprefix("        run: ") for line in run_steps["Run e2e full suite"] if line.startswith("        run: "))
+        for status in (0, 23):
+            result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", body], cwd=self.root, env=dict(env, FAKE_TEST_EXIT=str(status)), capture_output=True, text=True)
+            self.assertEqual(result.returncode, status, result.stderr)
+        copied.write_text("replaced preserved binary")
+        result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", body], cwd=self.root, env=env, capture_output=True, text=True)
+        self.assertNotEqual(result.returncode, 0)
+
+    def test_distributed_workflow_runs_both_filter_branches_with_receipts(self):
+        from check_test_wiring import yaml_block
+        from test_security_workflow import named_steps, shell_body
+
+        source = (Path(__file__).resolve().parents[1] / ".github/workflows/e2e-distributed.yml").read_text().splitlines()
+        steps = named_steps(yaml_block(source, "distributed", 2))
+        result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", shell_body(steps["Build rustfs binary"])], cwd=self.root, env=self.env, capture_output=True, text=True)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        for selected in ("", "test(distributed::s3_basic)"):
+            for status in (0, 23):
+                result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", shell_body(steps["Run distributed 4-node e2e suite"])], cwd=self.root, env=dict(self.env, FILTER=selected, FAKE_TEST_EXIT=str(status)), capture_output=True, text=True)
+                self.assertEqual(result.returncode, status, result.stderr)
+                argv = json.loads((self.root / "target/nextest-command.json").read_text())
+                self.assertEqual(argv, ["nextest", "run", "--profile", "e2e-distributed", "-p", "e2e_test", *(["-E", selected] if selected else ["--no-tests=fail"])])
+
+    def test_target_directory_and_profile_are_explicit(self):
+        env = dict(self.env, CARGO_TARGET_DIR="target/custom")
+        built = self.invoke("build", "--profile", "release", env=env)
+        self.assertEqual(built.returncode, 0, built.stderr)
+        run = self.invoke("run", "--profile", "release", "--", sys.executable, "-c", "pass", env=env)
+        self.assertEqual(run.returncode, 0, run.stderr)
+        self.assertNotEqual(self.invoke("run", "--", sys.executable, "-c", "pass", env=env).returncode, 0)
+
+    def test_target_directory_cannot_hide_source_inputs(self):
+        for target in (str(self.root), str(self.root / "crates"), str(self.root.parent)):
+            with self.subTest(target=target):
+                result = self.invoke("build", env=dict(self.env, CARGO_TARGET_DIR=target))
+                self.assertNotEqual(result.returncode, 0)
+                self.assertIn("CARGO_TARGET_DIR", result.stderr)
+        tracked = self.root / "target/tracked.rs"
+        tracked.write_text("tracked build input")
+        subprocess.run(["git", "add", "-f", "target/tracked.rs"], cwd=self.root, check=True)
+        self.build()
+        tracked.write_text("changed tracked build input")
+        self.assertNotEqual(self.run_code().returncode, 0)
+
+    def test_unsupported_embedded_directory_links_fail_closed(self):
+        self.build()
+        destination = self.root / "target/embedded-assets"
+        destination.mkdir()
+        (destination / "index.html").write_text("untracked embedded input")
+        static = self.root / "rustfs/static"
+        static.mkdir(parents=True)
+        (static / "linked-assets").symlink_to(destination, target_is_directory=True)
+        self.assertNotEqual(self.run_code().returncode, 0)
+
+    def test_directory_aliases_cannot_hide_unrecorded_inputs(self):
+        self.build()
+        target = self.root / ".agents/skills/SKILL.md"
+        target.write_text("changed instructions\n")
+        self.assertNotEqual(self.run_code().returncode, 0)
+        self.build()
+        (target.parent / ".gitignore").write_text("hidden.rs\n")
+        (target.parent / "hidden.rs").write_text("ignored build input\n")
+        result = self.invoke("build")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("unrecorded input", result.stderr)
+        alias = self.root / ".claude/skills"
+        alias.unlink()
+        with tempfile.TemporaryDirectory() as external:
+            alias.symlink_to(external, target_is_directory=True)
+            result = self.invoke("build")
+            self.assertNotEqual(result.returncode, 0)
+            self.assertIn("escapes the source inventory", result.stderr)
+
+    def test_directory_alias_indirection_is_part_of_the_identity(self):
+        for name in ("first", "second"):
+            directory = self.root / name
+            directory.mkdir()
+            (directory / "input.rs").write_text(name)
+        selection = self.root / "target/selection"
+        selection.symlink_to(self.root / "first", target_is_directory=True)
+        (self.root / "source-alias").symlink_to("target/selection", target_is_directory=True)
+        self.build()
+        selection.unlink()
+        selection.symlink_to(self.root / "second", target_is_directory=True)
+        self.assertNotEqual(self.run_code().returncode, 0)
+
+    def test_existing_embedded_files_and_symlink_targets_are_hashed(self):
+        static = self.root / "rustfs/static"
+        static.mkdir(parents=True)
+        index = static / "index.html"
+        index.write_text("embedded version one")
+        external = self.root / "target/embedded-file"
+        external.write_text("linked version one")
+        (static / "linked.html").symlink_to(external)
+        self.build()
+        index.write_text("embedded version two")
+        self.assertNotEqual(self.run_code().returncode, 0)
+        self.build()
+        external.write_text("linked version two")
+        self.assertNotEqual(self.run_code().returncode, 0)
+
+    def test_each_run_hashes_binary_twice_and_never_calls_cargo(self):
+        script = self.root / "scripts/e2e_binary.py"
+        script.write_text(script.read_text().replace("def file_hash(path):\n", "def file_hash(path):\n    if path.name == 'rustfs':\n        with (ROOT / 'target/hash-count').open('a') as count:\n            count.write('hash\\n')\n"))
+        self.build()
+        count = self.root / "target/hash-count"
+        count.write_text("")
+        result = self.run_code(env=dict(self.env, FAKE_BUILD_FAIL="1"))
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(count.read_text().splitlines(), ["hash", "hash"])
+
+    def test_concurrent_build_or_run_is_rejected(self):
+        self.build()
+        command = [sys.executable, str(self.root / "scripts/e2e_binary.py"), "run", "--", sys.executable, "-c", "print('ready', flush=True); input()"]
+        with subprocess.Popen(command, cwd=self.root, env=self.env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) as process:
+            self.assertEqual(process.stdout.readline().strip(), "ready")
+            try:
+                for args in (("build", "--features", "sftp"), ("run", "--", sys.executable, "-c", "pass")):
+                    rejected = self.invoke(*args)
+                    self.assertNotEqual(rejected.returncode, 0)
+                    self.assertIn("Another E2E build/run", rejected.stderr)
+            finally:
+                output, error = process.communicate("\n", timeout=10)
+            self.assertEqual(process.returncode, 0, error + output)
+        self.assertFalse(self.binary.with_name("rustfs.e2e.lock").exists())
+
+    def test_interruption_cleans_receipt_and_releases_ownership(self):
+        self.build()
+        for signum in (signal.SIGINT, signal.SIGTERM):
+            command = [sys.executable, str(self.root / "scripts/e2e_binary.py"), "run", "--", sys.executable, "-c", "import os; print(os.environ['RUSTFS_E2E_BINARY_RECEIPT'], flush=True); input()"]
+            with subprocess.Popen(command, cwd=self.root, env=self.env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) as process:
+                receipt = Path(process.stdout.readline().strip())
+                self.assertTrue(receipt.is_file())
+                process.send_signal(signum)
+                process.communicate(timeout=10)
+                self.assertNotEqual(process.returncode, 0)
+            self.assertFalse(receipt.exists())
+            self.assertFalse(self.binary.with_name("rustfs.e2e.lock").exists())
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -163,13 +163,12 @@ SH
                self.assertFalse(self.store.exists())
                self.assertEqual(list(self.root.glob("nightly-awscli.*")), [])

-    def test_manifest_advertises_checked_out_head_even_when_github_sha_differs(self):
-        # With a ref override (NIGHTLY_BRANCH variable / dispatch `branch`
-        # input) the checked-out HEAD intentionally differs from GITHUB_SHA;
-        # the candidate manifest must record the tree that was built.
+    def test_checkout_sha_mismatch_fails_before_upload(self):
        result = self.run_publish(GITHUB_SHA="f" * 40)
-        self.assertEqual(result.returncode, 0, result.stderr)
-        self.assertEqual(self.manifest()["source_sha"], self.sha)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("Checkout SHA", result.stderr)
+        self.assertFalse(self.output.exists())
+        self.assertFalse((self.root / "aws.log").exists())

    def test_same_date_builds_and_reruns_keep_distinct_candidates(self):
        urls = []
Author	SHA1	Message	Date
overtrue	dddb4aa817	fix(ci): integrate strict gate with E2E provenance	2026-09-09 05:56:33 +08:00
overtrue	d70bc931b3	test(e2e): record compiled Linux receipt test membership	2026-09-09 01:48:41 +08:00
overtrue	5832623e70	test(e2e): record compiled Darwin receipt test membership	2026-09-09 00:31:36 +08:00
overtrue	e0f2d56c96	chore(ci): integrate current main for E2E provenance	2026-09-08 23:55:05 +08:00
overtrue	832c7dab9f	chore: merge main into E2E binary provenance	2026-09-08 22:45:34 +08:00
overtrue	6aeacdd961	test(e2e): record verified Linux receipt test membership	2026-09-08 22:24:01 +08:00
overtrue	a8cecf6462	test(e2e): register verified Darwin test membership	2026-09-05 20:18:52 +08:00
overtrue	980f3abbd3	chore: merge main PR evidence guidance	2026-09-05 19:00:05 +08:00
overtrue	e36650827b	chore: integrate shared quick checks for E2E validation	2026-09-05 18:59:07 +08:00
overtrue	09c8e10d5e	feat(test): verify the E2E server build and source identity	2026-09-05 18:58:01 +08:00
overtrue	0ff03c596c	chore: merge main after ECStore compile repair	2026-09-05 18:46:20 +08:00
overtrue	a74919db8e	fix(ci): reject dependencies on required quick checks	2026-09-05 18:17:13 +08:00
overtrue	e77c6f0ca5	fix(ci): install actionlint from its verified release	2026-09-05 17:42:14 +08:00
overtrue	3149411943	fix(ci): share quick checks and lint workflows	2026-09-05 17:37:20 +08:00