feat(nightly): publish packages as assets of the rolling 'nightly' release (#7592)

Replace the assets-branch scheme with a proper GitHub Release on
rustfs/auto-testing: a single 'nightly' release whose deb/rpm assets
are replaced in place on every build. This is the standard channel —
visible on the repo's Releases page, stable download URLs, no git
history growth (release assets live outside the repository).

- New scripts/release/publish_nightly_assets.sh: resolves-or-creates
  the 'nightly' release via the REST API, deletes same-name assets,
  uploads rustfs-nightly-latest.{deb,rpm}, then PATCHes the release
  body with the build provenance (ref@sha, run link, sizes, SHA256).
  Plain curl + python3, no gh CLI (the build fleet has none — #7586).
- The workflow step shrinks to invoking the script; full flow
  exercised end-to-end against the real release with probe files
  (create / upload / overwrite / download round-trip / body update).

.config/make/lint-fmt.mak

@@ -89,6 +89,7 @@ offline-enrollment-e2e-check: core-deps ## Build and exercise the dedicated offl
 test-wiring-check: ## Check tests stay registered and selected by their intended runners
 	@echo "🧪 Checking test wiring..."
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_test_wiring.py
+	$(RUSTFS_PYTHON_BIN) ./scripts/ci_gate.py --check-workflow
 
 .PHONY: log-analyzer-rules-check
 log-analyzer-rules-check: core-deps ## Check log-analyzer rule anchors still exist verbatim in source

.config/make/tests.mak

@@ -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/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
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_security_workflow.py

.config/nextest.toml

@@ -197,6 +197,12 @@ 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`)
 # ---------------------------------------------------------------------------

.github/actions/quick-checks/action.yml

@@ -111,10 +111,6 @@ runs:
       shell: bash
       run: ./scripts/check_no_planning_docs.sh
 
-    - name: Check CI paths stay in sync
-      shell: bash
-      run: ./scripts/check_ci_paths_sync.sh
-
     - name: Check io_uring lane --lib precondition
       shell: bash
       run: ./scripts/check_uring_lane_lib_only.sh

.github/workflows/ci-docs-only.yml

@@ -1,79 +0,0 @@
-# Copyright 2026 RustFS Team
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-# Reports the existing required checks for paths excluded by ci.yml.
-# Mixed PRs can trigger both workflows; their Quick Checks jobs use one shared
-# action to keep validation coverage aligned. Keep this paths list in sync with
-# ci.yml's pull_request.paths-ignore via scripts/check_ci_paths_sync.sh.
-
-name: Continuous Integration (docs only)
-
-on:
-  pull_request:
-    types: [ opened, synchronize, reopened ]
-    branches: [ main ]
-    paths:
-      - "**.md"
-      - "docs/**"
-      - "deploy/**"
-      - "scripts/dev_*.sh"
-      - "scripts/probe.sh"
-      - "LICENSE*"
-      - ".gitignore"
-      - ".dockerignore"
-      - "README*"
-      - "**/*.png"
-      - "**/*.jpg"
-      - "**/*.svg"
-      - ".github/workflows/build.yml"
-      - ".github/workflows/docker.yml"
-      - ".github/workflows/audit.yml"
-      - "flake.lock"
-
-permissions:
-  contents: read
-
-jobs:
-  quick-checks:
-    name: Quick Checks
-    runs-on: ubuntu-latest
-    timeout-minutes: 10
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
-        with:
-          persist-credentials: false
-
-      - name: Run shared quick checks
-        uses: ./.github/actions/quick-checks
-
-  test-and-lint:
-    name: Test and Lint
-    runs-on: ubuntu-latest
-    timeout-minutes: 10
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
-        with:
-          persist-credentials: false
-
-      # Docs-only PRs skip the full code CI, but they are exactly where a
-      # planning-type document could be slipped in (git add -f bypasses
-      # .gitignore). Run the guard here so the required "Test and Lint" check
-      # stays meaningful for docs-only changes.
-      - name: Check no planning docs committed
-        run: ./scripts/check_no_planning_docs.sh
-
-      - name: Satisfy required check for docs-only changes
-        run: echo "Docs-only change — code CI is skipped by paths-ignore; planning-docs guard passed, reporting success for the required 'Test and Lint' check."

.github/workflows/ci.yml

@@ -37,25 +37,6 @@ on:
   pull_request:
     types: [ opened, synchronize, reopened, closed ]
     branches: [ main ]
-    # Keep this list in sync with the `paths` list in ci-docs-only.yml, which
-    # reports the required "Test and Lint" check for PRs skipped here.
-    paths-ignore:
-      - "**.md"
-      - "docs/**"
-      - "deploy/**"
-      - "scripts/dev_*.sh"
-      - "scripts/probe.sh"
-      - "LICENSE*"
-      - ".gitignore"
-      - ".dockerignore"
-      - "README*"
-      - "**/*.png"
-      - "**/*.jpg"
-      - "**/*.svg"
-      - ".github/workflows/build.yml"
-      - ".github/workflows/docker.yml"
-      - ".github/workflows/audit.yml"
-      - "flake.lock"
   merge_group:
     types: [ checks_requested ]
   schedule:
@@ -88,6 +69,32 @@ jobs:
       - name: Explain cancellation run
         run: echo "PR closed; this run only cancels older runs in the same concurrency group."
 
+  classify-changes:
+    name: Select CI scope
+    if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    outputs:
+      mode: ${{ steps.scope.outputs.mode }}
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          fetch-depth: 2
+          persist-credentials: false
+      - name: Select scope using the base revision's policy
+        id: scope
+        env:
+          CI_BASE_SHA: ${{ github.event.pull_request.base.sha }}
+        run: |
+          if [[ "$GITHUB_EVENT_NAME" != "pull_request" ]]; then
+            printf '%s\n' 'mode=full' >> "$GITHUB_OUTPUT"
+          elif [[ "$CI_BASE_SHA" =~ ^[0-9a-f]{40}$ ]] && git show "$CI_BASE_SHA:scripts/ci_gate.py" > "$RUNNER_TEMP/ci-gate-base.py"; then
+            python3 -I "$RUNNER_TEMP/ci-gate-base.py" select
+          else
+            printf '%s\n' 'mode=full' >> "$GITHUB_OUTPUT"
+            echo "Base CI policy unavailable; running the full matrix."
+          fi
+
   typos:
     name: Typos
     if: github.event_name != 'pull_request' || github.event.action != 'closed'
@@ -100,7 +107,7 @@ jobs:
       - name: Typos check with custom config file
         uses: crate-ci/typos@37bb98842b0d8c4ffebdb75301a13db0267cef89 # master
 
-  # Fail early with compile-free checks shared with docs-only CI.
+  # Fail early with compile-free checks for every pull request.
   quick-checks:
     name: Quick Checks
     if: github.event_name != 'pull_request' || github.event.action != 'closed'
@@ -116,9 +123,9 @@ jobs:
         uses: ./.github/actions/quick-checks
 
   test-and-lint:
-    name: Test and Lint
+    name: Workspace Test and Lint
-    if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    if: needs.classify-changes.outputs.mode == 'full' && (github.event_name != 'pull_request' || github.event.action != 'closed')
-    needs: [ quick-checks ]
+    needs: [ quick-checks, classify-changes ]
     runs-on: sm-standard-4
     timeout-minutes: 90
     env:
@@ -289,45 +296,6 @@ jobs:
       - name: Run rebalance/decommission migration proofs
         run: ./scripts/check_migration_gate_count.sh
 
-      # Record the reason before this job completes as FAILURE. A separate
-      # dependent job cancels sibling lanes only after GitHub has preserved this
-      # required check's failure verdict.
-      - name: Annotate early-stop reason
-        if: >-
-          failure() && github.event_name == 'pull_request'
-          && github.event.pull_request.head.repo.full_name == github.repository
-        run: |
-          {
-            echo "## CI early-stop"
-            echo "Job \`${GITHUB_JOB}\` (Test and Lint) failed; a follow-up job will cancel sibling lanes to free runners."
-            echo "Sibling jobs showing **cancelled** were stopped by the early-stop follow-up, not by their own failure."
-          } >> "$GITHUB_STEP_SUMMARY"
-
-  # Preserve the required Test and Lint FAILURE verdict before stopping sibling
-  # lanes. Cancelling from inside test-and-lint changed its own conclusion to
-  # CANCELLED and hid the actionable failure in the PR checks UI.
-  cancel-after-test-and-lint-failure:
-    name: Cancel siblings after Test and Lint failure
-    if: >-
-      failure() && needs.test-and-lint.result == 'failure'
-      && github.event_name == 'pull_request'
-      && github.event.pull_request.head.repo.full_name == github.repository
-    needs: [ test-and-lint ]
-    runs-on: ubuntu-latest
-    timeout-minutes: 5
-    permissions:
-      actions: write
-    steps:
-      - name: Cancel remaining jobs
-        env:
-          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-        run: |
-          curl -fsS -X POST \
-            -H "Authorization: Bearer ${GH_TOKEN}" \
-            -H "Accept: application/vnd.github+json" \
-            -H "X-GitHub-Api-Version: 2022-11-28" \
-            "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}/cancel"
-
   # Dedicated serial lane for the ILM / lifecycle integration tests. These tests
   # drive the object layer through process-global singletons (the GLOBAL_ENV
   # ECStore, the global tier-config manager, background-expiry workers) and bind
@@ -340,8 +308,8 @@ jobs:
   # See rustfs/backlog#1148 (ilm-1) and #1155.
   test-ilm-integration-serial:
     name: ILM Integration (serial)
-    if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    if: needs.classify-changes.outputs.mode == 'full' && (github.event_name != 'pull_request' || github.event.action != 'closed')
-    needs: [ quick-checks ]
+    needs: [ quick-checks, classify-changes ]
     runs-on: sm-standard-4
     timeout-minutes: 90
     env:
@@ -408,8 +376,8 @@ jobs:
 
   test-and-lint-rio-v2:
     name: Test and Lint (rio-v2)
-    if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    if: needs.classify-changes.outputs.mode == 'full' && (github.event_name != 'pull_request' || github.event.action != 'closed')
-    needs: [ quick-checks ]
+    needs: [ quick-checks, classify-changes ]
     runs-on: sm-standard-4
     timeout-minutes: 90
     env:
@@ -449,8 +417,8 @@ jobs:
 
   connect-short-credential-boundary:
     name: Connect Short Credential Boundary
-    if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    if: needs.classify-changes.outputs.mode == 'full' && (github.event_name != 'pull_request' || github.event.action != 'closed')
-    needs: [ quick-checks ]
+    needs: [ quick-checks, classify-changes ]
     runs-on: sm-standard-4
     timeout-minutes: 60
     env:
@@ -507,8 +475,8 @@ jobs:
 
   test-and-lint-protocols:
     name: "Test and Lint (${{ matrix.features.name }})"
-    if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    if: needs.classify-changes.outputs.mode == 'full' && (github.event_name != 'pull_request' || github.event.action != 'closed')
-    needs: [ quick-checks ]
+    needs: [ quick-checks, classify-changes ]
     runs-on: sm-standard-4
     timeout-minutes: 90
     strategy:
@@ -561,8 +529,8 @@ jobs:
 
   build-rustfs-debug-binary:
     name: Build RustFS Debug Binary
-    if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    if: needs.classify-changes.outputs.mode == 'full' && (github.event_name != 'pull_request' || github.event.action != 'closed')
-    needs: [ quick-checks ]
+    needs: [ quick-checks, classify-changes ]
     runs-on: sm-standard-4
     timeout-minutes: 30
     env:
@@ -684,8 +652,8 @@ jobs:
     # job had neither, so each closed/merged PR really ran the whole io_uring
     # suite (measured 4m17s / 7m19s / 7m31s on runs 30678272341 / 30678117601 /
     # 30662728539) and kept the cancellation run in progress for minutes.
-    if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    if: needs.classify-changes.outputs.mode == 'full' && (github.event_name != 'pull_request' || github.event.action != 'closed')
-    needs: [ quick-checks ]
+    needs: [ quick-checks, classify-changes ]
     # GitHub-hosted ubuntu-latest runs a recent kernel with io_uring and, unlike
     # a container, applies no seccomp filter that would block io_uring_setup — so
     # the probe succeeds and the tests exercise the real UringBackend/FdCache/
@@ -1212,9 +1180,44 @@ jobs:
           if-no-files-found: ignore
           retention-days: 3
 
+  required-checks:
+    name: Test and Lint
+    if: always() && (github.event_name != 'pull_request' || github.event.action != 'closed')
+    needs:
+      - classify-changes
+      - typos
+      - quick-checks
+      - test-and-lint
+      - test-ilm-integration-serial
+      - test-and-lint-rio-v2
+      - connect-short-credential-boundary
+      - test-and-lint-protocols
+      - build-rustfs-debug-binary
+      - uring-integration
+      - e2e-tests
+      - s3-implemented-tests
+      - s3-lifecycle-behavior-tests
+      - build-rustfs-debug-binary-rio-v2
+      - e2e-tests-rio-v2
+      - e2e-full
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+      - name: Require the expected result of every CI lane
+        env:
+          CI_NEEDS: ${{ toJSON(needs) }}
+        shell: bash
+        run: python3 scripts/ci_gate.py verify
+
   alert-on-failure:
     name: Alert on scheduled failure
     needs:
+      - classify-changes
+      - connect-short-credential-boundary
+      - required-checks
       - typos
       - quick-checks
       - test-and-lint

.github/workflows/nightly-gnu.yml

@@ -19,17 +19,27 @@ 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-main-${{ github.event_name }}
+  group: nightly-gnu-build-${{ github.event_name }}-${{ github.event_name == 'schedule' && (vars.NIGHTLY_BRANCH || 'main') || (inputs.branch || github.ref_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:
@@ -43,6 +53,7 @@ jobs:
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
+          ref: ${{ env.NIGHTLY_BUILD_REF }}
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup
@@ -152,13 +163,104 @@ 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)
@@ -187,11 +289,10 @@ 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}"
@@ -247,6 +348,20 @@ 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
@@ -284,6 +399,7 @@ jobs:
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
+          ref: ${{ env.NIGHTLY_BUILD_REF }}
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup
@@ -372,6 +488,7 @@ jobs:
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
+          ref: ${{ env.NIGHTLY_BUILD_REF }}
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup

.github/workflows/rustfs-performance-test.yml

@@ -82,6 +82,12 @@ jobs:
   performance-test:
     runs-on: pf-testing
     timeout-minutes: 900
+    env:
+      RUSTFS_BENCH_SCRIPT: ${{ github.workspace }}/auto-testing/rustfs_performance_testing.sh
+      RUSTFS_WARP_METHODS: ${{ inputs.test_method }}
+      RUSTFS_WARP_SIZES: ${{ inputs.object_size }}
+      RUSTFS_WARP_DURATION: ${{ inputs.warp_duration || '5m' }}
+      RUSTFS_WARP_CONCURRENCY: ${{ inputs.warp_concurrency || '64' }}
     # Run on manual dispatch, or when the nightly build completed successfully.
     # Skipped when nightly failed.
     if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
@@ -158,19 +164,15 @@ jobs:
       - name: Run benchmark (GET/PUT/MIXED)
         id: benchmark
         run: |
-          # Empty on automatic (workflow_run) runs -> full 30 rounds.
-          # Manual dispatch can restrict method(s)/size(s).
-          export WARP_METHODS="${{ inputs.test_method }}"
-          export WARP_SIZES="${{ inputs.object_size }}"
           ./auto-testing/rustfs_performance_test.sh \
             --step 5 -y \
-            --warp-duration "${{ inputs.warp_duration || '5m' }}" \
-            --warp-concurrency "${{ inputs.warp_concurrency || '64' }}" \
             --log-file "${LOG_FILE}"
 
       - name: Analyze results
         if: ${{ steps.benchmark.conclusion == 'success' }}
         run: |
+          export WARP_METHODS="${RUSTFS_WARP_METHODS}" WARP_SIZES="${RUSTFS_WARP_SIZES}"
+          export WARP_DURATION="${RUSTFS_WARP_DURATION}" WARP_CONCURRENCY="${RUSTFS_WARP_CONCURRENCY}"
           ./auto-testing/rustfs_performance_test.sh --step 6 -y --log-file "${LOG_FILE:-/dev/null}"
 
       - name: Collect RustFS version info

.gitignore

@@ -52,6 +52,7 @@ docs
 __pycache__/
 !docs/
 docs/*
+!docs/README.md
 !docs/architecture/
 !docs/architecture/**
 !docs/operations/

CHANGELOG.md

@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Replication
+
+- Object Lock replication PUTs now carry a required integrity header, fixing target rejection introduced by the plain-payload default ([#7097](https://github.com/rustfs/rustfs/pull/7097)). This changes the default outbound request for locked objects but adds no persisted format.
+- Multipart source objects stay on the multipart transport even when their checksum record is a whole-object checksum, so objects above the single-PUT limit remain replicable ([#7047](https://github.com/rustfs/rustfs/pull/7047)).
+- Targets that mint their own version IDs now use a per-target version ledger for tag, retention, legal-hold, and permanent-delete mutations; ambiguous pre-ledger matches fail with backoff instead of guessing ([#7368](https://github.com/rustfs/rustfs/pull/7368)). This adds dual-prefixed internal metadata keys that older readers ignore.
+- Single-part source checksums are forwarded as `x-amz-checksum-*` headers instead of user metadata, so the replica preserves checksum responses ([#7313](https://github.com/rustfs/rustfs/pull/7313)). This changes the default outbound headers for checksummed objects.
+- Site-replication outage recovery now uses a bounded 30-second retry drain plus the 600-second full reconciliation pass, persists destructive liabilities before local deletion, and fences replay settlement and peer edits ([#7148](https://github.com/rustfs/rustfs/pull/7148)). Persisted additions are optional and ignored by older readers.
+- IAM snapshot/deletion replay, target-assigned delete-marker purges, timestamp ordering, and best-effort peer broadcast now close the control-plane gaps found by the R6 review ([#7195](https://github.com/rustfs/rustfs/pull/7195)).
+- Upgrade and rollback: upgrade every node in one site consecutively and verify reconciliation before moving to the next site; do not intentionally run a site mixed-version. Target-version ledger keys are harmless on rollback, although old code cannot use their routing. Before rolling back past [#7307](https://github.com/rustfs/rustfs/pull/7307), drain or repair every pending version purge: older code can free a retained version's data directory before its remote purge is acknowledged. See `docs/operations/site-replication-operations.md`.
+
 ### Security
 - **Presigned URLs honour only signed headers** (GHSA-g8w9-qw9q-fghr): a SigV4 presigned request that carries an `x-amz-*` request header not listed in `X-Amz-SignedHeaders` is now rejected with `403 AccessDenied` ("There were headers present in the request which were not signed"), matching AWS S3. Previously the holder of a presigned `PutObject` URL could add unsigned `x-amz-tagging`, `x-amz-storage-class`, `x-amz-website-redirect-location`, ACL, metadata, Object Lock or SSE headers and have them applied. Presigners that intend a property must set it before signing so the SDK lists the header in `SignedHeaders`; `x-amz-cf-id` (CloudFront) remains tolerated unsigned. Header-signed SigV4 and SigV2 requests are unchanged.
 

README.md

@@ -211,7 +211,10 @@ For developers who want to build RustFS Docker images from source with multi-arc
 
 ```bash
 # Build multi-architecture images locally
-./docker-buildx.sh --build-arg RELEASE=latest
+./docker-buildx.sh
+
+# Build a single-platform image locally
+./docker-buildx.sh -p linux/amd64
 
 # Build and push to registry
 ./docker-buildx.sh --push

README_ZH.md

@@ -150,7 +150,10 @@ docker compose -f docker-compose-simple.yml up -d
 
 ```bash
 # 在本地构建多架构镜像
-./docker-buildx.sh --build-arg RELEASE=latest
+./docker-buildx.sh
+
+# 在本地构建单平台镜像
+./docker-buildx.sh -p linux/amd64
 
 # 构建并推送到仓库
 ./docker-buildx.sh --push

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -1047,6 +1047,8 @@ 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));

crates/e2e_test/src/namespace_lock_quorum_test.rs

@@ -25,6 +25,190 @@ 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()

crates/e2e_test/src/on_demand_migration/interaction_test.rs

@@ -1137,7 +1137,12 @@ async fn test_odm_admin_config_is_redacted_and_status_counts_match_the_source()
         let miss = env.raw_get(bucket, miss_key).await?;
         assert_eq!(miss.status, 404, "{}", String::from_utf8_lossy(&miss.body));
     }
-    assert!(env.wait_local_listed(bucket, hit_key, SETTLE).await?);
+    let (listed, _, _) = tokio::try_join!(
+        env.wait_local_listed(bucket, hit_key, SETTLE),
+        env.wait_for_status_counter(bucket, "/counters/pulled_objects_total/inline", 1, SETTLE),
+        env.wait_for_status_counter(bucket, "/counters/pulled_bytes_total", body.len() as u64, SETTLE),
+    )?;
+    assert!(listed);
 
     let status = env.status_json(bucket).await?;
     assert_eq!(status.pointer("/configured").and_then(Value::as_bool), Some(true), "{status}");

crates/e2e_test/src/replication_extension_test.rs

@@ -4235,6 +4235,16 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR
     <ExistingObjectReplication><Status>Enabled</Status></ExistingObjectReplication>
     <Destination><Bucket>{target_b_arn}</Bucket></Destination>
   </Rule>
+  <Rule>
+    <ID>matrix-and-tags</ID>
+    <Priority>135</Priority>
+    <Status>Enabled</Status>
+    <Filter><And><Prefix>and-tags/</Prefix><Tag><Key>env</Key><Value>prod</Value></Tag><Tag><Key>tier</Key><Value>gold</Value></Tag></And></Filter>
+    <DeleteMarkerReplication><Status>Disabled</Status></DeleteMarkerReplication>
+    <DeleteReplication><Status>Enabled</Status></DeleteReplication>
+    <ExistingObjectReplication><Status>Enabled</Status></ExistingObjectReplication>
+    <Destination><Bucket>{target_b_arn}</Bucket></Destination>
+  </Rule>
   <Rule>
     <ID>matrix-disabled</ID>
     <Priority>140</Priority>
@@ -4289,6 +4299,7 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR
         "matrix-prefix",
         "matrix-tag",
         "matrix-disabled",
+        "matrix-and-tags",
         "matrix-priority-high",
         "Priority>200",
         "<Status>Disabled</Status>",
@@ -4409,6 +4420,30 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR
     put_single_tag_current(&source_client, source_bucket, "tagged/no-match.txt", "route", "tagged").await?;
     assert_replication_key_absent(&target_client_b, target_bucket_b, "tagged/no-match.txt", Duration::from_secs(3)).await?;
 
+    // S3 and MinIO both read `And.Tags` as AND: an object carrying only one of
+    // the required tags is not admitted. Matching any single tag would push
+    // data to a destination the rule never selected (backlog#2366 P1-1), and
+    // the two-tag rule is the shape `mc replicate add --tags "k1=v1&k2=v2"`
+    // writes, so a single-tag rule passing is not evidence for this.
+    source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key("and-tags/partial.txt")
+        .tagging("env=prod")
+        .body(ByteStream::from_static(b"one of two tags"))
+        .send()
+        .await?;
+    assert_replication_key_absent(&target_client_b, target_bucket_b, "and-tags/partial.txt", Duration::from_secs(3)).await?;
+    source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key("and-tags/full.txt")
+        .tagging("env=prod&tier=gold")
+        .body(ByteStream::from_static(b"both tags"))
+        .send()
+        .await?;
+    wait_for_user_get_object(&target_client_b, target_bucket_b, "and-tags/full.txt").await?;
+
     source_client
         .put_object()
         .bucket(source_bucket)

crates/ecstore/src/core/pools.rs

@@ -3463,6 +3463,11 @@ 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>,
@@ -3473,17 +3478,21 @@ where
             NamespaceLock = rustfs_lock::NamespaceLockWrapper,
         >,
 {
-    // Activation lock order is always pool.bin -> rebalance.bin.
+    // 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)?;
     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,
@@ -22094,7 +22103,7 @@ mod pools_tests {
                 .resources
                 .lock()
                 .expect("activation lock recorder should not be poisoned"),
-            vec![POOL_META_NAME.to_string(), REBAL_META_NAME.to_string()]
+            vec![REBAL_META_NAME.to_string(), POOL_META_NAME.to_string()]
         );
 
         let mut second_acquire = Box::pin(acquire_pool_rebalance_activation_locks(second.clone(), None));
@@ -22110,10 +22119,50 @@ mod pools_tests {
                 .resources
                 .lock()
                 .expect("activation lock recorder should not be poisoned"),
-            vec![POOL_META_NAME.to_string(), REBAL_META_NAME.to_string()]
+            vec![REBAL_META_NAME.to_string(), POOL_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");

crates/ecstore/src/disk/error.rs

@@ -627,7 +627,13 @@ impl From<tokio::task::JoinError> for DiskError {
 impl Clone for DiskError {
     fn clone(&self) -> Self {
         match self {
-            DiskError::Io(io_error) => DiskError::Io(std::io::Error::new(io_error.kind(), io_error.to_string())),
+            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::MaxVersionsExceeded => DiskError::MaxVersionsExceeded,
             DiskError::Unexpected => DiskError::Unexpected,
             DiskError::CorruptedFormat => DiskError::CorruptedFormat,
@@ -1265,6 +1271,49 @@ 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};
@@ -1309,11 +1358,57 @@ 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());
@@ -1325,6 +1420,17 @@ 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]

crates/ecstore/src/disk/error_reduce.rs

@@ -226,6 +226,78 @@ 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");

crates/ecstore/src/services/rebalance/control.rs

@@ -572,7 +572,7 @@ impl ECStore {
     where
         S: EcstoreObjectIO + StorageNamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
     {
-        // Lock order: pool_meta_save_gate -> pool.bin -> rebalance.bin.
+        // Lock order: pool_meta_save_gate -> rebalance.bin -> pool.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

crates/ecstore/src/services/rebalance/entry.rs

@@ -50,6 +50,11 @@ 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,
@@ -256,9 +261,15 @@ 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
@@ -1237,6 +1248,130 @@ 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() {

crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs

@@ -1907,6 +1907,124 @@ 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"))));

crates/ecstore/src/services/rebalance/worker.rs

@@ -244,6 +244,7 @@ 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
@@ -256,6 +257,15 @@ 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,
@@ -309,6 +319,7 @@ 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()),
@@ -585,3 +596,48 @@ 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,
+                );
+            }
+        }
+    }
+}

crates/ecstore/src/set_disk/metadata.rs

@@ -326,14 +326,15 @@ impl SetDisks {
         let parity_blocks = Self::common_parity(&parities, default_parity_count as i32);
 
         if parity_blocks < 0 {
-            // No parity value reached read quorum. Distinguish two cases:
+            // A consistent layout can require more replies than the initial
-            // enough disks answered with valid-looking metadata that simply
+            // half-set probe. Reaching that probe alone is not corruption;
-            // cannot be reconciled (corrupt/foreign entries — retrying cannot
+            // only invalid or conflicting healthy replies establish that.
-            // 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();
-            if healthy_replies >= expected_rquorum {
+            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() {
                 error!(
                     "object_quorum_from_meta: irreconcilable parity across {healthy_replies} healthy replies (corrupt metadata), errs={errs:?}"
                 );
@@ -1652,6 +1653,40 @@ 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]

crates/ecstore/src/set_disk/mod.rs

@@ -865,6 +865,7 @@ 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;

crates/ecstore/src/set_disk/ops/bucket.rs

@@ -21,12 +21,72 @@
 
 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_write_quorum_errs,
+    MakeBucketOptions, Result, SetDisks, is_reserved_or_invalid_bucket, join_all, reduce_read_quorum_errs,
+    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;
@@ -131,59 +191,12 @@ impl BucketOperations for SetDisks {
 
     #[tracing::instrument(skip(self))]
     async fn get_bucket_info(&self, bucket: &str, _opts: &BucketOptions) -> Result<BucketInfo> {
-        let disks = self.disk_inventory().await;
+        let mut info = self.stat_bucket_with_quorum(bucket, BucketInfoQuorum::Write).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),
-                }
-            });
-        }
-
-        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();
+            info.versioning = sys.versioning();
-            object_locking = sys.object_locking();
+            info.object_locking = sys.object_locking();
         }
-
+        Ok(info)
-        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))]

crates/ecstore/src/store/bucket.rs

@@ -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::get_lock_acquire_timeout;
+use crate::set_disk::{BucketInfoQuorum, 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,17 +772,30 @@ 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 =
+        let mut scoped_results = futures::future::join_all(self.bucket_sets().map(|(pool_index, set_index, set)| async move {
-            futures::future::join_all(self.bucket_sets().map(|(pool_index, set_index, set)| async move {
+            let result = match quorum {
-                (pool_index, set_index, set.get_bucket_info(bucket, opts).await)
+                BucketInfoQuorum::Read => set.stat_bucket_with_quorum(bucket, quorum).await,
-            }))
+                BucketInfoQuorum::Write => set.get_bucket_info(bucket, opts).await,
-            .await;
+            };
+            (pool_index, set_index, result)
+        }))
+        .await;
         scoped_results.sort_unstable_by_key(|(pool_index, set_index, _)| (*pool_index, *set_index));
 
         let mut first_info = None;
@@ -806,7 +819,11 @@ impl ECStore {
 
     #[instrument(skip(self))]
     pub(super) async fn handle_get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
-        let mut info = self.get_bucket_info_from_sets(bucket, opts).await?;
+        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),
+        };
 
         if let Ok(sys) = metadata_sys::get_in(&self.ctx, bucket).await {
             if should_override_created_from_metadata(sys.created) {
@@ -819,6 +836,35 @@ 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
@@ -1049,7 +1095,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::table_bucket_catalog_metadata_prefix;
+    use crate::bucket::metadata::{BucketMetadata, 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};
@@ -1076,6 +1122,7 @@ 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;
@@ -1359,11 +1406,18 @@ 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 in 0..2 {
+        for (pool_index, &drive_count) in drives_per_pool.iter().enumerate() {
             let mut endpoints = Vec::new();
-            for disk_index in 0..4 {
+            for disk_index in 0..drive_count {
                 let disk_path = temp_dir.path().join(format!("pool{pool_index}-disk{disk_index}"));
                 tokio::fs::create_dir_all(&disk_path)
                     .await
@@ -1378,7 +1432,7 @@ mod tests {
             pools.push(PoolEndpoints {
                 legacy: false,
                 set_count: 1,
-                drives_per_set: 4,
+                drives_per_set: drive_count,
                 endpoints: Endpoints::from(endpoints),
                 cmd_line: format!("bucket-test-pool-{pool_index}"),
                 platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
@@ -1399,9 +1453,12 @@ mod tests {
         )
         .await
         .expect("multi-pool ECStore should initialize");
-        let storage_class =
+        let mut storage_class_kvs = rustfs_config::server_config::KVS::new();
-            crate::config::storageclass::lookup_config_for_pools_without_env(&rustfs_config::server_config::KVS::new(), &[4, 4])
+        if let Some(parity) = standard_parity {
-                .expect("multi-pool storage class should match both four-disk pools");
+            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");
         for pool in &ecstore.pools {
             for set in &pool.disk_set {
                 set.set_test_storage_class_config(storage_class.clone());
@@ -2067,6 +2124,218 @@ 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() {
@@ -2090,6 +2359,7 @@ 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())
@@ -2097,6 +2367,7 @@ 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())
@@ -2108,7 +2379,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::ErasureWriteQuorum);
+        assert_eq!(info_err, StorageError::ErasureReadQuorum);
     }
 
     #[tokio::test]

crates/kms/src/service_manager.rs

@@ -577,13 +577,16 @@ impl KmsServiceManager {
         Some(service_version.probe_worker.as_ref()?.status())
     }
 
-    /// Health check for the KMS service
+    /// 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.
     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) => {
@@ -594,8 +597,6 @@ 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)
                     }
                 }
@@ -739,17 +740,6 @@ 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 {
@@ -1004,19 +994,6 @@ 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};

crates/kms/tests/behavior_resilience.rs

@@ -75,6 +75,44 @@ 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();

crates/replication/src/config.rs

@@ -166,6 +166,24 @@ fn rule_replicates(rule: &ReplicationRule, obj: &ObjectOpts) -> bool {
     }
 }
 
+fn replication_filter_tags_match(filter: &s3s::dto::ReplicationRuleFilter, object_tags: &HashMap<String, String>) -> bool {
+    let tag_matches = |tag: &s3s::dto::Tag| match (&tag.key, &tag.value) {
+        (None, None) => true,
+        (Some(key), _) if key.is_empty() => true,
+        (Some(key), Some(value)) => object_tags.get(key) == Some(value),
+        _ => false,
+    };
+
+    filter
+        .and
+        .as_ref()
+        .and_then(|and| and.tags.as_deref())
+        .into_iter()
+        .flatten()
+        .chain(filter.tag.iter())
+        .all(tag_matches)
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum ReplicationTargetValidationError {
     RoleWithMultipleDestinations,
@@ -704,7 +722,7 @@ impl ReplicationConfigurationExt for ReplicationConfiguration {
 
             if let Some(filter) = &rule.filter {
                 let object_tags = ReplicationTagFilter::decode_tags_to_map(&obj.user_tags);
-                if filter.test_tags(&object_tags) {
+                if replication_filter_tags_match(filter, &object_tags) {
                     rules.push(rule.clone());
                 }
             } else {
@@ -1139,6 +1157,47 @@ mod tests {
         assert_eq!(validate_replication_config_structure(&structure_config(vec![rule])), Ok(()));
     }
 
+    #[test]
+    fn actionable_rules_require_every_and_tag_to_match() {
+        let mut rule = replication_rule("rule-1", "arn:target:a");
+        rule.filter = Some(s3s::dto::ReplicationRuleFilter {
+            and: Some(s3s::dto::ReplicationRuleAndOperator {
+                prefix: None,
+                tags: Some(vec![
+                    s3s::dto::Tag {
+                        key: Some("env".to_string()),
+                        value: Some("prod".to_string()),
+                    },
+                    s3s::dto::Tag {
+                        key: Some("tier".to_string()),
+                        value: Some("gold".to_string()),
+                    },
+                ]),
+            }),
+            ..Default::default()
+        });
+        let config = structure_config(vec![rule]);
+        let object = |user_tags: &str| ObjectOpts {
+            name: "object".to_string(),
+            user_tags: user_tags.to_string(),
+            ..Default::default()
+        };
+
+        assert!(config.filter_target_arns(&object("env=prod")).is_empty());
+        assert_eq!(config.filter_target_arns(&object("env=prod&tier=gold")), vec!["arn:target:a"]);
+        assert!(config.filter_target_arns(&object("")).is_empty());
+
+        let mut malformed = config;
+        malformed.rules[0].filter.as_mut().unwrap().and.as_mut().unwrap().tags = Some(vec![s3s::dto::Tag {
+            key: Some("env".to_string()),
+            value: None,
+        }]);
+        assert!(
+            malformed.filter_target_arns(&object("env=prod")).is_empty(),
+            "a malformed tag filter must fail closed"
+        );
+    }
+
     #[test]
     fn structure_validation_allows_tag_filter_when_delete_marker_replication_disabled() {
         let mut rule = replication_rule("rule-1", "arn:target:a");

crates/replication/src/stats.rs

@@ -580,6 +580,30 @@ impl FailStats {
         FailedMetric { count, size }
     }
 
+    /// Both rolling windows from one walk of the samples. `short` must be the
+    /// narrower window; the walk stops at `long`. Callers that need both (the
+    /// per-node site snapshot) would otherwise scan the deque twice while
+    /// holding the bucket-stats read lock, and the deque is only bounded by
+    /// the one-hour window - an unreachable target under load fills it.
+    pub fn recent_windows(&self, short: Duration, long: Duration) -> (FailedMetric, FailedMetric) {
+        let now = Instant::now();
+        let mut short_metric = FailedMetric::default();
+        let mut long_metric = FailedMetric::default();
+        for sample in self.recent.iter().rev() {
+            let age = now.duration_since(sample.observed_at);
+            if age > long {
+                break;
+            }
+            if age <= short {
+                short_metric.count += 1;
+                short_metric.size += sample.size;
+            }
+            long_metric.count += 1;
+            long_metric.size += sample.size;
+        }
+        (short_metric, long_metric)
+    }
+
     pub fn merge(&self, other: &FailStats) -> Self {
         Self {
             count: self.count.saturating_add(other.count),
@@ -912,6 +936,26 @@ mod tests {
         assert_eq!(last_hour.size, 96);
     }
 
+    #[test]
+    fn fail_stats_recent_windows_matches_two_separate_scans() {
+        let mut stats = FailStats::default();
+        stats.add_size(64, None::<&()>);
+        stats.add_size(32, None::<&()>);
+
+        let (minute, hour) = stats.recent_windows(Duration::from_secs(60), Duration::from_secs(60 * 60));
+        let expected_minute = stats.recent_since(Duration::from_secs(60));
+        let expected_hour = stats.recent_since(Duration::from_secs(60 * 60));
+
+        assert_eq!((minute.count, minute.size), (expected_minute.count, expected_minute.size));
+        assert_eq!((hour.count, hour.size), (expected_hour.count, expected_hour.size));
+        assert_eq!(minute.count, 2);
+        assert_eq!(hour.size, 96);
+
+        let empty = FailStats::default();
+        let (minute, hour) = empty.recent_windows(Duration::from_secs(60), Duration::from_secs(60 * 60));
+        assert_eq!((minute.count, minute.size, hour.count, hour.size), (0, 0, 0, 0));
+    }
+
     #[test]
     fn fail_stats_saturate_instead_of_wrapping() {
         let mut stats = FailStats {

crates/targets/src/store.rs

@@ -919,6 +919,22 @@ where
             }
         }
 
+        // The filename's item count is untrusted. Reject a payload that contains
+        // more items than advertised instead of returning success and allowing the
+        // caller to delete the entry with trailing events still in the file.
+        match deserializer.next() {
+            None => {}
+            Some(Ok(_)) => {
+                return Err(StoreError::Deserialization(format!(
+                    "Batch for key {key} contains more than {} items",
+                    key.item_count
+                )));
+            }
+            Some(Err(e)) => {
+                return Err(StoreError::Deserialization(format!("Failed to deserialize trailing batch item: {e}")));
+            }
+        }
+
         if items.is_empty() && key.item_count > 0 {
             return Err(StoreError::Deserialization("No items found".to_string()));
         }
@@ -1381,6 +1397,39 @@ mod tests {
         let _ = store.delete();
     }
 
+    #[test]
+    fn get_multiple_errors_on_batch_with_trailing_items_instead_of_partial_success() {
+        let dir = temp_store_dir("trailing-batch-items");
+        let store = QueueStore::<String>::new_with_compression(&dir, 8, ".test", false);
+        store.open().unwrap();
+
+        let items = vec!["aa".to_string(), "bb".to_string(), "cc".to_string()];
+        let original_key = store.put_multiple(items).unwrap();
+        assert_eq!(original_key.item_count, 3);
+
+        // Keep the three-item payload but make its filename claim that it contains
+        // only two items, simulating a corrupt or otherwise untrusted queue key.
+        let original_path = store.file_path(&original_key);
+        let advertised_key = Key {
+            item_count: 2,
+            ..original_key
+        };
+        let advertised_path = store.file_path(&advertised_key);
+        std::fs::rename(&original_path, &advertised_path).unwrap();
+
+        let err = store.get_multiple(&advertised_key).unwrap_err();
+        assert!(
+            matches!(err, StoreError::Deserialization(_)),
+            "expected Deserialization error, got {err:?}"
+        );
+
+        // Because get_multiple failed, the batch entry remains available for
+        // inspection or recovery instead of being silently discarded.
+        assert!(advertised_path.exists());
+
+        let _ = store.delete();
+    }
+
     #[test]
     fn concurrent_put_raw_respects_entry_limit() {
         let dir = temp_store_dir("concurrent-limit");

docs/README.md

@@ -0,0 +1,23 @@
+# Documentation
+
+Use the focused indexes rather than treating this directory as an unordered
+collection:
+
+- [Architecture knowledge base](architecture/README.md)
+- [Testing references](testing/README.md)
+
+## Operations
+
+Operational runbooks live under [`operations/`](operations/). Replication
+operators should start with:
+
+| Runbook | Use it for |
+|---|---|
+| [Site replication operations](operations/site-replication-operations.md) | Health fields, pending operations, outage recovery, re-pair admission, IAM/SSE boundaries, and upgrades. |
+| [Replication target check](operations/replication-check.md) | Validating an S3 destination and version fidelity before enabling replication. |
+| [Replication object size limits](operations/replication-object-size-limits.md) | Multipart routing, large-object limits, and retry characteristics. |
+| [Replication outbound transport](operations/replication-outbound-transport.md) | Integrity headers, generic target behavior, and transport knobs. |
+
+Other runbooks remain grouped by filename in [`operations/`](operations/);
+architecture pages link to the relevant runbook where a cross-boundary
+procedure is required.

docs/architecture/README.md

@@ -60,6 +60,8 @@ Required headings and strings in these files are asserted by `scripts/check_arch
 | [minio-rustfs-router-compatibility.md](minio-rustfs-router-compatibility.md) | a client or `mc` call that works against MinIO fails against RustFS and you need to know whether the endpoint is missing, stubbed, or deliberately different |
 | [minio-file-format-compat.md](minio-file-format-compat.md) | deciding whether a MinIO drive set, bucket-metadata blob, or SSE object can be read or imported by a given RustFS build, or before touching a listed version anchor |
 
-Operations runbooks live in [../operations/](../operations/) and testing references in [../testing/README.md](../testing/README.md).
+Operations runbooks are registered in the [documentation operations index](../README.md#operations), and testing references live in [../testing/README.md](../testing/README.md).
+
+For replication operations, start with [site replication operations](../operations/site-replication-operations.md), [replication target check](../operations/replication-check.md), [replication object size limits](../operations/replication-object-size-limits.md), and [replication outbound transport](../operations/replication-outbound-transport.md).
 
 For per-node HTTP failure ratios and cached storage probe provenance, see [S3 write failure diagnostics](../operations/s3-write-failure-diagnostics.md).

docs/architecture/s3-compatibility-matrix.md

@@ -38,6 +38,42 @@ Counts ignore blank lines and comments; compute them from the files. The lifecyc
 
 "Supported" for the SSE row means RustFS encrypts and decrypts its own objects. MinIO SSE objects (SSE-S3, SSE-KMS, SSE-C) are not readable in default builds; see [minio-file-format-compat.md Part C](minio-file-format-compat.md#part-c--server-side-encryption-sse) for the `rio-v2` migration build.
 
+## Replication Support Boundary
+
+Site replication and bucket replication are not the same compatibility claim.
+Site replication requires RustFS-compatible peer admin APIs and coordinates
+IAM, topology, buckets, and metadata. A generic S3-compatible service can only
+be a bucket-replication data target.
+
+For a generic S3 target, RustFS supports object PUT/HEAD/DELETE, multipart
+uploads, tags, version deletes, and Object Lock mutations when the target
+implements the corresponding S3 APIs and has versioning enabled. Targets that
+mint their own version IDs are supported through a per-target version ledger;
+pre-ledger replicas are adopted only when exact key and ETag identify one
+unambiguous target version. `NoSuchVersion` for an already absent addressed
+replica is treated as converged.
+
+The following are capability boundaries, not universal S3 claims:
+
+- `GET /BUCKET?replication-check` must pass the phases required by the intended
+  workload. `VersionFidelity` may report a minting target as mismatched even
+  though ledger-addressed delete and Object Lock phases succeed.
+- A target that rejects standard multipart constraints, required Object Lock
+  integrity headers, or the configured checksum framing is unsupported until
+  its transport settings are made compatible.
+- SSE-S3 and SSE-KMS are decrypted at the source and re-encrypted by the
+  destination's KMS. SSE-C uses ciphertext passthrough and requires target
+  evidence. Unsupported or ambiguous encryption metadata fails closed.
+- ACL authorization is intentionally unsupported, and generic targets never
+  receive RustFS IAM/site-control-plane state.
+- RustFS does not guess between multiple target versions with the same key and
+  ETag. The mutation remains failed and retryable until repair establishes an
+  unambiguous mapping.
+
+See [site replication operations](../operations/site-replication-operations.md)
+for health, recovery, and upgrade rules and [replication outbound transport](../operations/replication-outbound-transport.md)
+for the tested target classes and knobs.
+
 ## Not Yet Passing
 
 Standard S3 areas that must not be described as complete:

docs/operations/site-replication-operations.md

@@ -0,0 +1,258 @@
+# Site Replication Operations
+
+**Use this when:** operating a site-replication deployment, diagnosing a peer
+outage or incomplete topology change, pairing sites that already contain data,
+or planning an upgrade.
+
+**Source of truth:** `rustfs/src/admin/handlers/site_replication.rs`,
+`rustfs/src/site_replication/`, and the bucket-replication worker under
+`crates/ecstore/src/bucket/replication/`.
+
+Site replication combines two different convergence paths:
+
+- the control plane replicates buckets, bucket metadata, IAM, and topology;
+- ordinary bucket replication moves object versions and delete operations.
+
+An `enabled: true` response only says that a site has more than one configured
+peer. It does not prove that every peer is reachable or caught up. Always read
+`pendingOperation`, `retryStats`, `PeerErrors`, and `Metrics` as well.
+
+## Routine checks
+
+Run these commands from an admin workstation with one alias per site:
+
+```console
+mc admin replicate info site-a
+mc admin replicate status site-a
+```
+
+Check more than one site. A partition can leave each side with a different but
+locally valid view.
+
+`replicate info` is the compact control-plane view:
+
+| Field | Interpretation |
+|---|---|
+| `enabled` | More than one site is configured; this is not a health verdict. |
+| `sites` | The locally persisted topology. Compare deployment IDs and endpoints on every site. |
+| `retryStats.pending` | Collapsed peer deliveries waiting to be retried. |
+| `retryStats.failed` | Deliveries that crossed the escalation threshold and require attention. |
+| `retryStats.lastError` | A redacted summary of the most recent delivery failure. |
+| `pendingOperation` | A durable multi-step topology operation described below. Absence is the healthy steady state. |
+
+`replicate status` adds detailed convergence state:
+
+| Field | Interpretation |
+|---|---|
+| `Sites` / `PeerStates` | Configured peers and derived reachability/configuration state. |
+| `PeerErrors` | A peer could not be queried. Its detailed counters may be absent; do not read zeros as success. |
+| `BucketStats` | Per-bucket presence and versioning, replication, lifecycle, Object Lock, and metadata mismatches. |
+| `PolicyStats`, `UserStats`, `GroupStats` | IAM inventory mismatches. |
+| `RetryStats` | Durable control-plane retry backlog and escalation count. |
+| `Metrics.replMetrics` | Per-destination online state, downtime, replicated counts/bytes, and `failed` totals/windows. |
+| `Metrics.queued` / `Metrics.inProgress` | Object work waiting or active on the responding node. |
+| `Metrics.errors` | Node-level object-replication failures. When only queue statistics are available, RustFS synthesizes a node entry and preserves this counter rather than reporting zero. |
+| `Metrics.retries` | Redeliveries. Always zero today: a failed object is not retried by an event, it waits for the scanner pass described below. Read `errors` instead. |
+
+Healthy means: the same topology is visible on all sites, no pending operation,
+no peer error, no failed retry escalation, required bucket/IAM state is in sync,
+and queue/error counters are stable or falling. Counters are cumulative; alert on
+their rate and on a backlog that does not drain, not merely on a non-zero total.
+
+## Pending operations and recovery
+
+`pendingOperation` contains `operation`, an opaque `id`, `pendingPeers`, and
+`ackedPeers`. Do not edit the site-replication state object by hand. The marker
+is the crash-recovery journal and removing it can make a partially applied
+operation look complete.
+
+The heavyweight reconciler runs once at startup and every 600 seconds. The
+lightweight retry drain runs every 30 seconds. A restart is therefore a valid
+way to cause an immediate heavyweight pass after the underlying fault has been
+fixed, but it is not a substitute for fixing connectivity, credentials, TLS,
+or the remote endpoint.
+
+### `remove`
+
+The original topology and each peer acknowledgement are persisted before the
+operation finalizes. While peers remain in `pendingPeers`, restore access to
+them and wait for reconciliation. If a peer is permanently gone, a new remove
+request may remove all currently active unacknowledged peers; RustFS permits
+that request and then finalizes against the remaining topology. Removing the
+local site or all sites is also an explicit completion path.
+
+Do not re-add a site merely to hide this marker. First compare the topology on
+all reachable peers. If the same operation ID makes no progress for more than
+one heavyweight interval, collect `PeerErrors`, `RetryStats`, and the
+site-replication logs before retrying the remove.
+
+### `rotate-svc-acct`
+
+Service-account rotation keeps the candidate secrets and peer acknowledgements
+until every current remote peer accepts the rotation. Restore the failing peer
+and allow the reconciler to resume it. Do not manually delete either candidate
+credential during this window: doing so can remove the only credential that a
+not-yet-acknowledged peer accepts.
+
+After the marker clears, verify `replicate status` from every site, then retire
+any separately retained old credential material according to local policy.
+
+### `endpoint-refresh`
+
+An endpoint, CA, or TLS-verification edit first refreshes the replication
+target on every active peer and records acknowledgements. On startup and every
+heavyweight pass, RustFS probes peer capability, uses the endpoint-refresh API
+when supported (or the legacy peer-edit fallback), refreshes local bucket
+targets, and commits the edit only after every still-active peer acknowledges.
+
+If this marker is stuck:
+
+1. Confirm that the proposed endpoint and CA are correct and reachable from
+   every site, not only from the admin workstation.
+2. Restore the site-replication service account and TLS trust path.
+3. Wait for one 600-second pass or restart one healthy node to trigger the
+   startup pass.
+4. Re-run the identical edit only if the operation remains visible; a different
+   endpoint edit is rejected while the existing refresh is pending. The journal
+   pins the edit's payload, so a re-run without `--replicate-ilm-expiry` keeps
+   the value the first attempt recorded, and a re-run asking for a different
+   value is rejected. Finish or remove the pending refresh before changing it.
+
+A peer removed from the topology no longer blocks completion. A remove request
+is accepted when it removes every active unacknowledged peer.
+
+While this marker is present, control-plane retry replay to the other peers
+keeps running, but bucket wiring reconciliation waits: it rewrites the same
+targets the refresh is changing. Expect bucket-level drift on this site to
+persist until the refresh settles.
+
+## Outage recovery and convergence time
+
+Control-plane retry begins on the 30-second drain, while heavyweight snapshots,
+pending topology operations, and bucket wiring are revisited on the 600-second
+pass. Object MRF entries are persisted every 10 seconds by default and target
+health is probed every 5 seconds. These are scheduling bounds, not delivery
+SLAs: network timeouts and the amount of queued work add to them.
+
+Objects that must be rediscovered by the scanner have this conservative upper
+bound before discovery:
+
+```text
+RUSTFS_DATA_USAGE_UPDATE_DIR_CYCLES
+    × max(RUSTFS_SCANNER_CYCLE, actual duration of one scanner cycle)
+```
+
+The defaults re-descend a compacted directory every 16 cycles. A practical
+production starting point for a tighter recovery objective is
+`RUSTFS_DATA_USAGE_UPDATE_DIR_CYCLES=4`; `1` forces re-descent every cycle.
+Measure the additional disk and metadata load before lowering it further or
+tuning the scanner cadence. For an immediate operator-driven recovery, start a
+site resync with `mc admin replicate resync start` and monitor its status.
+Transfer time after discovery remains proportional to backlog size, bandwidth,
+worker capacity, and target latency. Use queue depth and the rate of
+`Metrics.errors` rather than the formula alone to decide whether convergence is
+progressing.
+
+## Pairing sites that already contain data
+
+When more than one requested site is non-empty, preflight considers each bucket
+name held by more than one site:
+
+- versioning must be `Enabled` on every site holding the shared bucket;
+- Object Lock enablement must be identical on every holder.
+
+A bucket present on only one site is safe: post-add backfill creates it on the
+other peers. A shared unversioned bucket is rejected because merging can
+overwrite the only copy of an object. An Object Lock mismatch is rejected
+because lock enablement cannot be changed after bucket creation and convergence
+could otherwise strip a WORM guarantee.
+
+If preflight rejects the pair, keep the authoritative copy, delete the
+conflicting bucket (or its contents) from all other sites, run `replicate add`
+again, and then start `replicate resync` from the surviving site. Back up and
+validate the authoritative data before deleting anything.
+
+## IAM convergence and repair boundary
+
+Ordinary IAM changes are delivered to each peer. A successful bulk IAM import
+also schedules one collapsed full-IAM snapshot per remote peer. A failed IAM
+deletion is replayed before that snapshot so the snapshot cannot re-create a
+principal or grant that was already revoked.
+
+The safety state has two bounds:
+
+- deletion high-water marks are retained for 30 days;
+- deletion replay bodies are capped at 256 distinct entities per peer.
+
+Repeated deletion of the same entity replaces its saved body. When the per-peer
+cap is exceeded or the body cannot be serialized, the retry entry remains
+escalated rather than pretending the deletion is replayable. An item from an
+older sender without a source timestamp cannot install the 30-day high-water
+mark, so verify it explicitly after a prolonged split. A successful drain
+clears replay bodies; removing the peer prunes its bodies. For an escalated IAM
+retry, use the site-replication repair workflow for the affected peer and IAM
+family, then verify users, service accounts, groups, policies, and mappings on
+both sides. Repair is the operator's explicit accountability transfer and
+clears the saved deletion bodies only after the IAM repair succeeds.
+
+A group's status converges in one direction. An explicit disable is applied
+everywhere, including through a snapshot, but a membership change never
+carries an enable - it would otherwise re-enable a group frozen on the
+receiving site. If a group ended up disabled on one site only, re-enable it
+there explicitly with `mc admin group enable`; a snapshot or repair will not
+do it.
+
+Treat IAM divergence as a security incident: a user deleted on one site can
+remain usable on an unreachable peer until replay or repair completes. A peer
+whose IAM entry is escalated does not receive scheduled snapshots either -
+including the one a bulk import schedules - until the repair settles it.
+
+## Encrypted objects
+
+| Source form | Replication behavior | Fail-closed condition |
+|---|---|---|
+| SSE-S3 | The source decrypts the object; the request sends only `AES256` intent; the destination encrypts with its own KMS. Source envelope material never leaves the site. | The destination cannot satisfy the encryption request, or the source metadata is incomplete/unsupported. The replica is `FAILED`; plaintext is not silently stored. |
+| SSE-KMS | The source decrypts the object; the request sends `aws:kms` intent without the source-local key ID; the destination selects its own configured KMS key. | Either side cannot decrypt/encrypt, or the metadata mixes incompatible encryption evidence. |
+| SSE-C | Stored ciphertext and the required SSE-C replication transport metadata pass through. RustFS verifies target evidence before accepting the replica. | The target does not echo the customer-algorithm evidence, required material/layout is absent, or the metadata is ambiguous. |
+
+Unknown MinIO/RustFS encryption markers are never forwarded as ordinary user
+metadata. They fail replication so an operator must migrate or repair the
+object with a supported format.
+
+## Rolling upgrades and rollback
+
+Keep every node in one site on the same version whenever possible. Upgrade all
+nodes of one site consecutively, verify its startup reconciliation and status,
+then move to the next site. Do not intentionally leave a site mixed-version:
+admin requests can land on different nodes, and an older node may not resume a
+new pending-operation shape or expose its health fields.
+
+Current state additions are optional and defaulted, so older readers ignore
+them. The target-version ledger is stored as dual-prefixed internal object
+metadata and is also ignored by older readers; rollback does not corrupt the
+object format, but older code loses the assigned-version routing improvement.
+
+Before rolling back across the fix that retains the data directory of a version
+awaiting purge replication (rustfs/rustfs#7307), ensure no version purge is
+pending. Older code can free that retained version's data directory before the
+remote purge is acknowledged, leaving unreadable metadata and blocking bucket
+deletion. Drain or repair replication and take a metadata/data backup first.
+
+## Runtime knobs
+
+These values are read when the owning background task starts. Restart the
+server after changing them. The millisecond intervals have a 10 ms floor;
+invalid values fall back to the default with a warning.
+
+| Variable | Default | Effect |
+|---|---:|---|
+| `RUSTFS_REPL_HEALTH_CHECK_INTERVAL_MS` | `5000` | Remote-target health probe interval. Lowering it increases outbound probes. |
+| `RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS` | `10000` | Maximum periodic interval between MRF persistence flushes; 1,000 new entries also trigger a flush. |
+| `RUSTFS_REPL_RESYNC_POLL_MAX_MS` | `60000` | Upper bound for randomized resync retry-poll sleep. |
+| `RUSTFS_REPL_RESYNC_MAX_JOBS` | `2` | Concurrent resync jobs; values are bounded to `1..=32`. |
+
+Transport-specific controls and target behavior are documented in
+[Replication outbound transport](replication-outbound-transport.md). Validate a
+new destination with [Replication target check](replication-check.md), and read
+[Replication object size limits](replication-object-size-limits.md) before
+moving large objects.

docs/testing/ci-gates.md

@@ -3,7 +3,7 @@
 **Use this when:** a check is red and you need to know whether it blocks the merge, which workflow and job produced it, and how to reproduce it locally.
 **Source of truth:** the live `main` ruleset (command below) for required status; `.github/workflows/<file>.yml` for triggers, `paths`, `timeout-minutes`, and cron; `.config/nextest.toml` for e2e profile filters; `.github/scheduled-validations.json` for the freshness-watchdog list.
 
-A job blocks a merge only when its exact check name is in the live `main` ruleset. A workflow name, a `merge_group` trigger, or a red PR check does not make a job required by itself.
+A job blocks a merge when its exact check name is required by the live `main` ruleset, or when its result is required by the `Test and Lint` aggregate. A workflow name, a `merge_group` trigger, or an unrelated red PR check does not make a job required by itself.
 
 ## Required merge checks
 
@@ -13,9 +13,11 @@ The `main` ruleset (`6436880`) requires exactly these contexts, with `strict_req
 |---|---|---|
 | `CLA Check` | `cla.yml` | Contributor agreement |
 | `Quick Checks` | `ci.yml` job `quick-checks` | Formatting and repository guard scripts |
-| `Test and Lint` | `ci.yml` job `test-and-lint` | Clippy, workspace nextest (`ci` profile, excluding `e2e_test`), doctests, migration-gate count (`scripts/check_migration_gate_count.sh`) |
+| `Test and Lint` | `ci.yml` job `required-checks` | Exact expected results for every CI validation job, including workspace checks, critical E2E, feature lanes, and event-specific full suites |
 
-For PRs limited to the `paths-ignore` list in `ci.yml`, `ci-docs-only.yml` reports `Quick Checks` and `Test and Lint` under the same names; it runs the quick checks and `scripts/check_no_planning_docs.sh`, not a Rust build or tests. `scripts/check_ci_paths_sync.sh` keeps the two path lists aligned.
+Every PR enters `ci.yml`. The `classify-changes` job uses the base revision of `scripts/ci_gate.py` to select a conservative documentation-only path: root Markdown/licenses, `AGENTS.md`, Markdown under `docs/` or `.agents/skills/`, and documentation images. Unknown paths, unavailable Git history, an empty diff, or a missing base policy select the full matrix. Renames include their deleted source path. Documentation-only PRs still run Quick Checks and Typos; the aggregate requires the expensive jobs to be skipped exactly as selected.
+
+`required-checks` runs even after failed or skipped dependencies. `scripts/ci_gate.py verify` rejects missing jobs, unexpected jobs, failure, cancellation, and unexpected skips; optional lanes are required only on their declared events. `Workspace Test and Lint` is the ordinary Rust job, while `Test and Lint` uniquely names the aggregate. New validation jobs must update both its direct dependencies and the script contract. Test this wiring and its failure cases with `python3 scripts/ci_gate.py --self-test`.
 
 Verify the live rule before changing merge policy:
 
@@ -24,25 +26,25 @@ gh api repos/rustfs/rustfs/rulesets/6436880 \
   --jq '.rules[] | select(.type == "required_status_checks") | .parameters'
 ```
 
-Promotion rule: never promote a report-only lane to required from one green run. Require at least 14 days and 30 representative PRs with at least 99% complete execution, then update the ruleset and this file together.
+The aggregate requires the validation lanes already selected by `ci.yml`; this closes the gap where a failing critical lane left the required workspace check green. Independent workflows remain report-only unless separately required. Before adding a new expensive lane or moving existing PR coverage to a schedule, collect representative execution and regression evidence, establish ownership and a working scheduled replacement, and update this reference with the resulting policy.
 
 ## Pull request and merge matrix
 
-"Report-only" means visible and actionable but not in the required list. Budgets are each job's `timeout-minutes` in the named workflow and are not copied here.
+"Via aggregate" means a wrong result fails the required `Test and Lint` check. "Report-only" means visible and actionable but outside both the required list and aggregate. Budgets are each job's `timeout-minutes` in the named workflow and are not copied here.
 
 | Event | Check name | Workflow / job | Merge status | Reproduce |
 |---|---|---|---|---|
 | PR, non-doc change | `Quick Checks` | `ci.yml` `quick-checks` | Required | `make pre-commit` |
-| PR, non-doc change | `Test and Lint` | `ci.yml` `test-and-lint` | Required | `cargo clippy --all-targets -- -D warnings`; `cargo nextest run --profile ci --all --exclude e2e_test`; `cargo test --all --doc`; `scripts/check_migration_gate_count.sh` |
+| PR, non-doc change | `Workspace Test and Lint` | `ci.yml` `test-and-lint` | Via aggregate | `cargo clippy --all-targets -- -D warnings`; `cargo nextest run --profile ci --all --exclude e2e_test`; `cargo test --all --doc`; `scripts/check_migration_gate_count.sh` |
-| PR, non-doc change | `Typos` | `ci.yml` `typos` | Report-only | `typos` |
+| PR, non-doc change | `Typos` | `ci.yml` `typos` | Via aggregate | `typos` |
-| PR, non-doc change | `ILM Integration (serial)` | `ci.yml` `test-ilm-integration-serial` | Report-only | exact command in the job |
+| PR, non-doc change | `ILM Integration (serial)` | `ci.yml` `test-ilm-integration-serial` | Via aggregate | exact command in the job |
-| PR, non-doc change | `Test and Lint (rio-v2)`, `Test and Lint (swift)`, `Test and Lint (sftp)` | `ci.yml` `test-and-lint-rio-v2`, `test-and-lint-protocols` | Report-only | `cargo nextest run` with the job's `--features` |
+| PR, non-doc change | `Test and Lint (rio-v2)`, `Test and Lint (swift)`, `Test and Lint (sftp)` | `ci.yml` `test-and-lint-rio-v2`, `test-and-lint-protocols` | Via aggregate | `cargo nextest run` with the job's `--features` |
-| PR, non-doc change | `Connect Short Credential Boundary` | `ci.yml` `connect-short-credential-boundary` | Report-only | `cargo test -p rustfs --test connect_registration --features connect-e2e-short-credentials`; `cargo check -p rustfs --release --features connect-e2e-short-credentials` must fail |
+| PR, non-doc change | `Connect Short Credential Boundary` | `ci.yml` `connect-short-credential-boundary` | Via aggregate | `cargo test -p rustfs --test connect_registration --features connect-e2e-short-credentials`; `cargo check -p rustfs --release --features connect-e2e-short-credentials` must fail |
-| PR, non-doc change | `Build RustFS Debug Binary` | `ci.yml` `build-rustfs-debug-binary` | Report-only; prerequisite for the black-box jobs | `cargo build -p rustfs --bins` |
+| PR, non-doc change | `Build RustFS Debug Binary` | `ci.yml` `build-rustfs-debug-binary` | Via aggregate; prerequisite for black-box jobs | `cargo build -p rustfs --bins --features e2e-test-hooks` |
-| PR, non-doc change | `io_uring Integration (real)` | `ci.yml` `uring-integration` | Report-only | `cargo test -p rustfs-ecstore --lib uring_ -- --test-threads=1 --nocapture` |
+| PR, non-doc change | `io_uring Integration (real)` | `ci.yml` `uring-integration` | Via aggregate | `cargo test -p rustfs-ecstore --lib uring_ -- --test-threads=1 --nocapture` |
-| PR, non-doc change | `End-to-End Tests` | `ci.yml` `e2e-tests` | Report-only | `cargo nextest run --profile e2e-smoke -p e2e_test`, then `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>`; membership guards `scripts/check_test_wiring.py --check-profile e2e-smoke <listing.json>` and `scripts/check_security_smoke_count.sh check <listing.json>` |
+| PR, non-doc change | `End-to-End Tests` | `ci.yml` `e2e-tests` | Via aggregate | `cargo nextest run --profile e2e-smoke -p e2e_test`, then `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>`; membership guards `scripts/check_test_wiring.py --check-profile e2e-smoke <listing.json>` and `scripts/check_security_smoke_count.sh check <listing.json>` |
-| PR, non-doc change | `S3 Implemented Tests` | `ci.yml` `s3-implemented-tests` | Report-only | build `rustfs`, then `scripts/s3-tests/run.sh` with the job's `DEPLOY_MODE` / `TEST_MODE` / `MAXFAIL` env |
+| PR, non-doc change | `S3 Implemented Tests` | `ci.yml` `s3-implemented-tests` | Via aggregate | build `rustfs`, then `scripts/s3-tests/run.sh` with the job's `DEPLOY_MODE` / `TEST_MODE` / `MAXFAIL` env |
-| PR, non-doc change | `S3 Lifecycle Behavior Tests` | `ci.yml` `s3-lifecycle-behavior-tests` | Report-only | `scripts/s3-tests/run.sh` with the job's accelerated-scanner env |
+| PR, non-doc change | `S3 Lifecycle Behavior Tests` | `ci.yml` `s3-lifecycle-behavior-tests` | Via aggregate | `scripts/s3-tests/run.sh` with the job's accelerated-scanner env |
 | PR touching `paths` in `audit.yml` | `Cargo Deny`, `Workflow Pin Report`, `Dependency Review` | `audit.yml` `cargo-deny`, `workflow-pin-report`, `dependency-review` | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
 | PR touching `paths` in `architecture-migration-rules.yml` | `Architecture Migration Rules` | `architecture-migration-rules.yml` `architecture-migration-rules` | Report-only | `scripts/check_architecture_migration_rules.sh` |
 | PR touching `paths` in `nix.yml` | `Nix Build & Check` | `nix.yml` `nix-validation` | Report-only | `nix flake check` |
@@ -52,8 +54,8 @@ Promotion rule: never promote a report-only lane to required from one green run.
 | PR touching `paths` in `e2e-upgrade.yml` | `Direct upgrade from the previous release`, `Mixed-version rolling upgrade from the previous release`, `Bucket configuration survives the upgrade`, `Rollback reads current bucket metadata` | `e2e-upgrade.yml` `upgrade` matrix | Report-only | the `cargo test --locked -p e2e_test` command in the job with `RUSTFS_UPGRADE_SOURCE_BINARY` pointing at the pinned previous release (`UPGRADE_SOURCE_VERSION`) |
 | PR touching `paths` in `oidc-keycloak.yml` | `OIDC Keycloak live gate` | `oidc-keycloak.yml` `oidc-keycloak-live` | Report-only | `cargo build --locked -p rustfs --bin rustfs`, then `bash scripts/test/oidc_keycloak_live.sh ./target/debug/rustfs` |
 | PR touching `paths` in `targets-integration.yml` | `PostgreSQL, MySQL, AMQP, and NATS` | `targets-integration.yml` `targets-live` | Report-only | start the containers as in the job, export the `RUSTFS_TEST_*` DSNs, then the job's `cargo test --locked -p rustfs-targets --test <name> -- --ignored --test-threads=1` commands |
-| PR limited to main-CI-excluded paths | `Quick Checks`, `Test and Lint` | `ci-docs-only.yml` `quick-checks`, `test-and-lint` | Required | `git diff --check`; `make doc-paths-check`; `scripts/check_no_planning_docs.sh` |
+| PR, documentation-only selection | `Quick Checks`, `Typos`, `Test and Lint` | `ci.yml` `quick-checks`, `typos`, `required-checks` | Required directly or via aggregate | Quick Checks commands; `python3 scripts/ci_gate.py --self-test` |
-| `merge_group`; push to `main` | `End-to-End Tests (full merge gate)` | `ci.yml` `e2e-full` | Report-only | `cargo nextest run --profile e2e-full -p e2e_test` |
+| `merge_group`; push to `main` | `End-to-End Tests (full merge gate)` | `ci.yml` `e2e-full` | Via aggregate on these events | `cargo nextest run --profile e2e-full -p e2e_test` |
 
 e2e filters live in `.config/nextest.toml`; extend a profile instead of adding a second selector. Before a profile runs, `scripts/check_test_wiring.py` compares its listing to the committed digest in `.config/e2e-<profile>-selection.txt`, so a silent test drop fails closed.
 
@@ -67,11 +69,11 @@ the serialized cluster fault-domain suites for scheduled soak signal.
 
 ## Scheduled validation
 
-Scheduled lanes never block a PR. Their workflow-local gate fails the run, scheduled failures route to the shared failure-issue action, and `scheduled-validation-freshness.yml` fails when a workflow listed in `.github/scheduled-validations.json` has not run within its `max_age_hours` (a `never_ran_grace_until` entry covers the window before a newly enabled cron's first slot). Cadence is qualitative here; the cron lives in each workflow's `on.schedule`.
+Scheduled lanes never block a PR. Their workflow-local gate fails the run, scheduled failures route to the shared failure-issue action, and `scheduled-validation-freshness.yml` fails when a workflow listed in `.github/scheduled-validations.json` has no recent attempt or completed successful scheduled run within its `max_age_hours` (a `never_ran_grace_until` entry covers the window before a newly enabled cron's first slot). Cadence is qualitative here; the cron lives in each workflow's `on.schedule`.
 
 | Workflow (cadence) | Jobs | Verdict and artifacts | In freshness list | Reproduce |
 |---|---|---|---|---|
-| `ci.yml` (weekly) | full matrix, including the schedule/dispatch-only rio-v2 jobs `build-rustfs-debug-binary-rio-v2` and `e2e-tests-rio-v2` | per-job | yes | dispatch `ci.yml` |
+| `ci.yml` (weekly) | full matrix, including the schedule/dispatch-only rio-v2 jobs `build-rustfs-debug-binary-rio-v2` and `e2e-tests-rio-v2` | strict aggregate; the full E2E lane runs on dispatch, merge groups, and main pushes | yes | dispatch `ci.yml` |
 | `build.yml` (weekly) | `build-rustfs` over the six-target platform matrix in `prepare-platform-matrix` (four Linux, macOS aarch64, Windows x86_64) | build/package integrity | yes | dispatch `build.yml` with an exact platform set |
 | `e2e-replication-nightly.yml` (nightly) | `repl-nightly`, `cluster-nightly`, `protocols-nightly` | three independent gates; JUnit, membership listing, server logs | yes | `cargo nextest run --profile e2e-repl-nightly -p e2e_test`; `--profile e2e-nightly`; `-j 1 --profile e2e-protocols` |
 | `e2e-distributed.yml` (storage-sensitive PRs + nightly) | `distributed` | fail-closed 4-node 4-disk S3, durability, replication, movement, fault, and direct/rolling upgrade gate; JUnit, membership listing, per-node server logs | yes, with `never_ran_grace_until` | download the pinned previous release as in the workflow, export `RUSTFS_UPGRADE_SOURCE_BINARY`, then `cargo nextest run --profile e2e-distributed -p e2e_test` |
@@ -88,7 +90,7 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run, sched
 | `e2e-upgrade.yml` (weekly) | `upgrade` (4-case matrix) | upgrade and rollback gate; server logs | no | see the PR row |
 | `oidc-keycloak.yml` (weekly) | `oidc-keycloak-live` | live OIDC gate | no | see the PR row |
 | `targets-integration.yml` (nightly) | `targets-live` | live target gate; container logs | no | see the PR row |
-| `scheduled-validation-freshness.yml` (nightly) | `check-freshness` | fails on a never-created or stale schedule | n/a | dispatch |
+| `scheduled-validation-freshness.yml` (nightly) | `check-freshness` | fails on missing or stale attempts or completed successes | n/a | dispatch |
 
 Manual `workflow_dispatch` runs are debugging evidence and do not open scheduled-failure issues. A manual performance run may explicitly allow a known regression; that override is not a passing baseline.
 

rustfs/src/admin/handlers/replication.rs

@@ -627,6 +627,32 @@ pub(crate) async fn cluster_replication_stats(bucket: &str, context: Option<Arc<
         .await
 }
 
+/// Reload the bucket's metadata on every peer so a follow-up
+/// `put-bucket-replication` on another node does not read a stale target.
+///
+/// Best effort, like every S3 bucket-config write path
+/// (`app::bucket_usecase::notify_bucket_metadata_reload`): the target is
+/// already persisted and live on this node, and the 15-minute refresh closes
+/// the gap, so a peer that cannot be reached must not turn a completed write
+/// into a failed request.
+async fn notify_remote_target_metadata_reload(bucket: &str, context: Option<Arc<AppContext>>, action: &'static str) {
+    let Some(notification_system) = current_notification_system_for_context(context.as_deref()) else {
+        return;
+    };
+    if let Err(err) = notification_system.load_bucket_metadata(bucket).await {
+        warn!(
+            event = EVENT_ADMIN_REMOTE_TARGET_STATE,
+            component = LOG_COMPONENT_ADMIN,
+            subsystem = LOG_SUBSYSTEM_REPLICATION,
+            action = action,
+            result = "peer_metadata_reload_failed",
+            bucket = %bucket,
+            error = ?err,
+            "admin remote target state"
+        );
+    }
+}
+
 fn unique_replication_peers(peer_clients: &[Option<PeerRestClient>]) -> (Vec<&PeerRestClient>, u32) {
     let mut seen_grid_hosts = HashSet::new();
     let peers: Vec<_> = peer_clients
@@ -699,6 +725,7 @@ pub struct SetRemoteTargetHandler {}
 impl Operation for SetRemoteTargetHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let cred = validate_replication_admin_request(&req, AdminAction::SetBucketTargetAction).await?;
+        let app_context = app_context_from_req(&req);
 
         let queries = extract_query_params(&req.uri);
 
@@ -926,6 +953,8 @@ impl Operation for SetRemoteTargetHandler {
             .map_err(map_bucket_target_error)?;
         let _targets_guard = lock_bucket_targets_metadata(bucket).await;
         let arn = persist_remote_target_write(bucket, remote_target, incarnation, mode).await?;
+        drop(_targets_guard);
+        notify_remote_target_metadata_reload(bucket, app_context, "set_remote_target").await;
         let arn_str = serde_json::to_string(&arn)
             .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "Failed to serialize target ARN"))?;
 
@@ -1006,6 +1035,7 @@ pub struct RemoveRemoteTargetHandler {}
 impl Operation for RemoveRemoteTargetHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         validate_replication_admin_request(&req, AdminAction::SetBucketTargetAction).await?;
+        let app_context = app_context_from_req(&req);
 
         debug!("remove remote target called");
         let queries = extract_query_params(&req.uri);
@@ -1081,6 +1111,7 @@ impl Operation for RemoveRemoteTargetHandler {
         }
         let json_targets = serde_json::to_vec(&targets)
             .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "Failed to serialize targets"))?;
+        let notification_bucket = bucket.clone();
         let bucket = bucket.clone();
         let arn = arn_str.clone();
         // The pool cancellation owns a detached task. Both outer guards must
@@ -1101,6 +1132,8 @@ impl Operation for RemoveRemoteTargetHandler {
             S3Error::with_message(S3ErrorCode::InternalError, format!("remote target removal task failed: {error}"))
         })??;
 
+        notify_remote_target_metadata_reload(&notification_bucket, app_context, "remove_remote_target").await;
+
         Ok(S3Response::new((StatusCode::NO_CONTENT, Body::from("".to_string()))))
     }
 }
@@ -1787,6 +1820,25 @@ mod tests {
         pairs.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect()
     }
 
+    #[test]
+    fn remote_target_writes_notify_peer_metadata_caches() {
+        let source = include_str!("replication.rs");
+        for (start, end) in [
+            ("impl Operation for SetRemoteTargetHandler", "pub struct ListRemoteTargetHandler"),
+            ("impl Operation for RemoveRemoteTargetHandler", "async fn cancel_active_resync_intent"),
+        ] {
+            let body = source
+                .split(start)
+                .nth(1)
+                .and_then(|rest| rest.split(end).next())
+                .expect(start);
+            assert!(
+                body.contains("notify_remote_target_metadata_reload"),
+                "{start} must notify every node before returning success"
+            );
+        }
+    }
+
     #[test]
     fn update_ops_parse_minio_query_contract() {
         let ops = parse_remote_target_update_ops(&query_map(&[

rustfs/src/admin/handlers/user.rs

@@ -1318,6 +1318,24 @@ impl Operation for ImportIam {
             failed,
         };
 
+        // The entities are already imported locally. A snapshot that cannot be
+        // scheduled is a convergence delay the reconcile pass still closes, so
+        // it must not turn a completed import into a failed request - the same
+        // best-effort contract every other site-replication hook here follows.
+        if let Err(err) =
+            crate::site_replication::enqueue_site_replication_iam_snapshot("iam import scheduled a full snapshot").await
+        {
+            warn!(
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_USER,
+                event = EVENT_ADMIN_USER_STATE,
+                action = "import_iam",
+                result = "site_replication_snapshot_not_scheduled",
+                error = ?err,
+                "admin user state"
+            );
+        }
+
         let body = serde_json::to_vec(&ret).map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
 
         let mut header = HeaderMap::new();
@@ -1424,6 +1442,16 @@ mod tests {
         assert!(include_str!("user.rs").contains(mapper_call));
     }
 
+    #[test]
+    fn import_iam_enqueues_a_site_replication_snapshot() {
+        let body = source_block(include_str!("user.rs"), "impl Operation for ImportIam");
+
+        assert!(
+            body.contains("enqueue_site_replication_iam_snapshot"),
+            "a successful IAM import must schedule a full IAM snapshot for every remote site"
+        );
+    }
+
     #[test]
     fn test_should_check_deny_only_for_regular_self_request() {
         let cred = Credentials {

rustfs/src/admin/replication_metrics_wire.rs

@@ -409,6 +409,26 @@ fn transfer_summaries(stats: &InternalReplicationStats) -> (XferSummaryWire, Tar
     (summary, per_target)
 }
 
+/// Node-level failure counters for `errors`. The sibling `retries` field
+/// stays zero on purpose: it means redeliveries in the minio-go shape, and a
+/// failed object is not retried by an event today (it waits for the scanner's
+/// heal pass), so reporting failures there would claim a redelivery that
+/// never happened.
+fn failure_counters(stats: &InternalReplicationStats) -> CounterSummaryWire {
+    let (total, last1m, last1hr) = stats.stats.values().fold((0i64, 0i64, 0i64), |acc, stat| {
+        (
+            acc.0.saturating_add(stat.fail_stats.count),
+            acc.1.saturating_add(stat.fail_stats.last_minute.count),
+            acc.2.saturating_add(stat.fail_stats.last_hour.count),
+        )
+    });
+    CounterSummaryWire {
+        total: u64::try_from(total.max(0)).unwrap_or_default(),
+        last1m: u64::try_from(last1m.max(0)).unwrap_or_default(),
+        last1hr: u64::try_from(last1hr.max(0)).unwrap_or_default(),
+    }
+}
+
 impl MetricsV2Wire {
     /// Project the aggregated internal stats onto the `MetricsV2` shape.
     ///
@@ -418,6 +438,7 @@ impl MetricsV2Wire {
     /// `queueStats.nodes` and treats an empty list as "no data".
     pub(crate) fn from_stats(bucket_stats: &BucketStats, node_name: &str) -> Self {
         let (xfer_stats, tgt_xfer_stats) = transfer_summaries(&bucket_stats.replication_stats);
+        let failed = failure_counters(&bucket_stats.replication_stats);
         let mut nodes: Vec<ReplQNodeStatsWire> = bucket_stats
             .queue_stats
             .nodes
@@ -436,6 +457,7 @@ impl MetricsV2Wire {
                 q_stats: InQueueMetricWire::from(&bucket_stats.replication_stats.q_stat),
                 xfer_stats: xfer_stats.clone(),
                 tgt_xfer_stats: tgt_xfer_stats.clone(),
+                errors: failed,
                 ..Default::default()
             });
         } else {
@@ -444,6 +466,7 @@ impl MetricsV2Wire {
             if let Some(first) = nodes.first_mut() {
                 first.xfer_stats = xfer_stats.clone();
                 first.tgt_xfer_stats = tgt_xfer_stats.clone();
+                first.errors = failed;
             }
         }
 
@@ -478,6 +501,12 @@ mod tests {
         target.replicated_size = 4096;
         target.failed.count = 3;
         target.failed.size = 900;
+        target.fail_stats.count = 3;
+        target.fail_stats.size = 900;
+        target.fail_stats.last_minute.count = 2;
+        target.fail_stats.last_minute.size = 600;
+        target.fail_stats.last_hour.count = 3;
+        target.fail_stats.last_hour.size = 900;
         target.bandwidth_limit_bytes_per_sec = 1024;
         target.current_bandwidth_bytes_per_sec = 512.5;
         stats
@@ -537,6 +566,10 @@ mod tests {
         assert_eq!(node["queueStats"]["peak"], node["queueStats"]["max"]);
         assert!(node["activeWorkers"].get("curr").is_some());
         assert!(node["transferSummary"].get("Total").is_some());
+        assert_eq!(node["errors"]["total"], 3);
+        assert_eq!(node["errors"]["last1m"], 2);
+        assert_eq!(node["errors"]["last1hr"], 3);
+        assert_eq!(node["retries"]["total"], 0, "failures are not redeliveries; retries must not claim one");
         assert_eq!(json["downtimeInfo"], serde_json::json!({}));
     }
 

rustfs/src/site_replication/retry.rs

@@ -217,6 +217,26 @@ pub(crate) fn settle_observed_site_replication_retry_event(
     before.saturating_sub(queue.len())
 }
 
+/// Make sure `peer` has a collapsed entry for `path` without counting the
+/// call as a delivery failure. A bulk local mutation (`import-iam`) needs the
+/// entry to exist so the next drain sends the snapshot; routing it through
+/// [`upsert_site_replication_retry_event`] would raise `retry_count` on every
+/// import and escalate a healthy peer to `failed` after
+/// [`SITE_REPLICATION_RETRY_FAILED_AFTER`] of them, with the scheduling note
+/// shown to operators as `lastError`.
+pub(crate) fn ensure_site_replication_retry_event(
+    queue: &mut Vec<SiteReplicationRetryEvent>,
+    peer: &PeerInfo,
+    path: &str,
+    reason: &str,
+) -> S3Result<Vec<SiteReplicationRetryEvent>> {
+    let path = collapsed_retry_queue_path(path).unwrap_or(path);
+    if queue.iter().any(|event| retry_event_matches(event, peer, path)) {
+        return Ok(Vec::new());
+    }
+    push_site_replication_retry_event(queue, peer, path, summarize_peer_error_detail(reason), false, None)
+}
+
 pub(crate) fn upsert_site_replication_retry_event(
     queue: &mut Vec<SiteReplicationRetryEvent>,
     peer: &PeerInfo,
@@ -244,6 +264,17 @@ pub(crate) fn upsert_site_replication_retry_event(
         return Ok(Vec::new());
     }
 
+    push_site_replication_retry_event(queue, peer, path, detail, peer_unreachable, generation)
+}
+
+fn push_site_replication_retry_event(
+    queue: &mut Vec<SiteReplicationRetryEvent>,
+    peer: &PeerInfo,
+    path: &str,
+    detail: String,
+    peer_unreachable: bool,
+    generation: Option<u64>,
+) -> S3Result<Vec<SiteReplicationRetryEvent>> {
     let slots_needed = queue
         .len()
         .saturating_add(1)
@@ -274,7 +305,7 @@ pub(crate) fn upsert_site_replication_retry_event(
         retry_count: 1,
         failed: false,
         last_error: detail,
-        updated_at: Some(now),
+        updated_at: Some(OffsetDateTime::now_utc()),
         edit_generation: generation,
         peer_unreachable,
         deletions_recorded: false,
@@ -365,6 +396,60 @@ pub(crate) async fn enqueue_site_replication_retry_event_for_generation(
     }
 }
 
+/// Returns the number of peers whose snapshot entry is escalated and therefore
+/// will not carry this scheduling: the marker records a deletion that a
+/// snapshot cannot replay, and only a repair settles it, so clearing it to make
+/// the entry drainable again would drop that liability.
+pub(crate) fn record_iam_snapshot_retries(
+    state: &mut SiteReplicationState,
+    local_peer: &PeerInfo,
+    reason: &str,
+) -> S3Result<usize> {
+    let peers = state
+        .peers
+        .values()
+        .filter(|peer| {
+            peer.deployment_id != local_peer.deployment_id && !same_identity_endpoint(&peer.endpoint, &local_peer.endpoint)
+        })
+        .cloned()
+        .collect::<Vec<_>>();
+    let mut escalated = 0usize;
+    for peer in peers {
+        if state.retry_queue.iter().any(|event| {
+            retry_event_matches(event, &peer, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH)
+                && event.last_error == SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER
+        }) {
+            escalated += 1;
+            continue;
+        }
+        ensure_site_replication_retry_event(&mut state.retry_queue, &peer, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH, reason)?;
+    }
+    Ok(escalated)
+}
+
+/// Schedule one collapsed full-IAM snapshot per remote peer after a bulk
+/// local mutation such as `import-iam`.
+pub(crate) async fn enqueue_site_replication_iam_snapshot(reason: &str) -> S3Result<()> {
+    let state = load_site_replication_state().await?;
+    if !state.enabled() {
+        return Ok(());
+    }
+    let local_peer = current_local_runtime_peer(&state);
+    let reason = reason.to_string();
+    let escalated = update_site_replication_state(move |state| record_iam_snapshot_retries(state, &local_peer, &reason)).await?;
+    if escalated > 0 {
+        warn!(
+            component = LOG_COMPONENT_ADMIN,
+            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+            escalated,
+            result = "iam_snapshot_not_scheduled_for_escalated_peer",
+            "site replication peers hold an escalated IAM entry; the snapshot waits for a repair"
+        );
+    }
+    Ok(())
+}
+
 pub(crate) const SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH: &str = "/rustfs/admin/v3/site-replication/peer/iam-item";
 
 /// Per-peer cap on recorded deletion bodies. Beyond it the peer's collapsed

rustfs/src/site_replication/state.rs

@@ -168,6 +168,8 @@ mod rfc3339_map {
 pub(crate) struct PendingEndpointRefresh {
     pub(crate) id: String,
     pub(crate) peer: PeerInfo,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) ilm_expiry_override: Option<bool>,
     #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
     pub(crate) remote_peers: BTreeMap<String, PeerInfo>,
     #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]

rustfs/src/site_replication/tests.rs

@@ -693,6 +693,122 @@ fn test_record_iam_deletion_marks_newest_wins_and_expires_by_age_only() {
     );
 }
 
+/// Scheduling a snapshot is not a delivery failure. Repeated imports - the
+/// normal way a bulk IAM migration is done, one archive at a time - must not
+/// walk the peer's entry up to the escalation threshold and report a healthy
+/// site as `retryStats.failed` with the scheduling note as its `lastError`.
+#[test]
+fn repeated_iam_import_snapshots_do_not_escalate_a_healthy_peer() {
+    let local = PeerInfo {
+        deployment_id: "local-dep".to_string(),
+        ..peer("local", "https://local.example.com")
+    };
+    let remote = PeerInfo {
+        deployment_id: "remote-a".to_string(),
+        ..peer("remote-a", "https://a.example.com")
+    };
+    let mut state = SiteReplicationState {
+        peers: BTreeMap::from([
+            (local.deployment_id.clone(), local.clone()),
+            (remote.deployment_id.clone(), remote),
+        ]),
+        ..Default::default()
+    };
+
+    for _ in 0..(SITE_REPLICATION_RETRY_FAILED_AFTER + 2) {
+        record_iam_snapshot_retries(&mut state, &local, "iam import scheduled a full snapshot").expect("record snapshot");
+    }
+
+    assert_eq!(state.retry_queue.len(), 1);
+    let event = &state.retry_queue[0];
+    assert_eq!(event.retry_count, 1, "a schedule must not count as a delivery attempt");
+    assert!(!event.failed, "a scheduled snapshot must not report as an escalated failure");
+}
+
+/// An escalated entry records a deletion a snapshot cannot replay: only a
+/// repair settles it. Scheduling an import snapshot must not clear that
+/// marker to make the entry drainable again, and the peer it skips has to be
+/// reported rather than silently left behind.
+#[test]
+fn an_escalated_peer_keeps_its_marker_and_is_reported() {
+    let local = PeerInfo {
+        deployment_id: "local-dep".to_string(),
+        ..peer("local", "https://local.example.com")
+    };
+    let remote = PeerInfo {
+        deployment_id: "remote-a".to_string(),
+        ..peer("remote-a", "https://a.example.com")
+    };
+    let mut state = SiteReplicationState {
+        peers: BTreeMap::from([
+            (local.deployment_id.clone(), local.clone()),
+            (remote.deployment_id.clone(), remote.clone()),
+        ]),
+        retry_queue: vec![SiteReplicationRetryEvent {
+            id: "escalated".to_string(),
+            peer_deployment_id: remote.deployment_id.clone(),
+            peer_endpoint: remote.endpoint,
+            path: SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH.to_string(),
+            retry_count: SITE_REPLICATION_RETRY_FAILED_AFTER,
+            failed: true,
+            last_error: SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER.to_string(),
+            deletions_recorded: true,
+            ..Default::default()
+        }],
+        ..Default::default()
+    };
+
+    let escalated =
+        record_iam_snapshot_retries(&mut state, &local, "iam import scheduled a full snapshot").expect("record snapshot retries");
+
+    assert_eq!(escalated, 1);
+    assert_eq!(state.retry_queue.len(), 1);
+    assert_eq!(
+        state.retry_queue[0].last_error, SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER,
+        "the unreplayable-deletion marker must survive a snapshot schedule"
+    );
+}
+
+#[test]
+fn iam_import_snapshot_retry_is_recorded_once_per_remote_peer() {
+    let local = PeerInfo {
+        deployment_id: "local-dep".to_string(),
+        ..peer("local", "https://local.example.com")
+    };
+    let remote_a = PeerInfo {
+        deployment_id: "remote-a".to_string(),
+        ..peer("remote-a", "https://a.example.com")
+    };
+    let remote_b = PeerInfo {
+        deployment_id: "remote-b".to_string(),
+        ..peer("remote-b", "https://b.example.com")
+    };
+    let mut state = SiteReplicationState {
+        peers: BTreeMap::from([
+            (local.deployment_id.clone(), local.clone()),
+            (remote_a.deployment_id.clone(), remote_a),
+            (remote_b.deployment_id.clone(), remote_b),
+        ]),
+        ..Default::default()
+    };
+
+    record_iam_snapshot_retries(&mut state, &local, "IAM import snapshot pending").expect("record snapshot retries");
+
+    assert_eq!(state.retry_queue.len(), 2);
+    assert!(
+        state
+            .retry_queue
+            .iter()
+            .all(|event| event.path == SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH)
+    );
+    assert!(
+        state
+            .retry_queue
+            .iter()
+            .all(|event| event.peer_deployment_id != local.deployment_id)
+    );
+}
+
 /// A failed deletion delivery persists a replay record next to the collapsed
 /// retry entry; a fresh entry is stamped `deletions_recorded` so a later
 /// replay can settle it, and a repeated deletion of the same entity keeps the

rustfs/src/storage/storage_api.rs

@@ -17,6 +17,7 @@
 use std::collections::hash_map::DefaultHasher;
 use std::hash::{Hash, Hasher};
 use std::sync::{Arc, LazyLock};
+use std::time::Duration;
 
 use rand::RngExt as _;
 use rustfs_storage_api as storage_contracts;
@@ -836,6 +837,39 @@ impl StorageReplicationStatsHandle {
 
     pub(crate) async fn site_metrics_snapshot(&self) -> ReplicationSiteMetricsSnapshot {
         let metrics = self.inner.get_sr_metrics_for_node().await;
+        // Aggregate under the read lock rather than through `get_all`: that
+        // clones every bucket's stats, and `FailStats.recent` is bounded only
+        // by the one-hour window, so an unreachable target under load - the
+        // very case an operator polls this for - makes the copy large. The
+        // windows come from the live samples; the serialized `last_minute` /
+        // `last_hour` snapshots are stamped onto per-bucket clones elsewhere
+        // and stay zero in this node-local cache.
+        let (
+            failed_count,
+            failed_bytes,
+            failed_last_minute_count,
+            failed_last_minute_bytes,
+            failed_last_hour_count,
+            failed_last_hour_bytes,
+        ) = {
+            let cache = self.inner.cache.read().await;
+            cache
+                .values()
+                .flat_map(|bucket| bucket.stats.values())
+                .fold((0i64, 0i64, 0i64, 0i64, 0i64, 0i64), |totals, stat| {
+                    let (minute, hour) = stat
+                        .fail_stats
+                        .recent_windows(Duration::from_secs(60), Duration::from_secs(3600));
+                    (
+                        totals.0.saturating_add(stat.fail_stats.count),
+                        totals.1.saturating_add(stat.fail_stats.size),
+                        totals.2.saturating_add(minute.count),
+                        totals.3.saturating_add(minute.size),
+                        totals.4.saturating_add(hour.count),
+                        totals.5.saturating_add(hour.size),
+                    )
+                })
+        };
         ReplicationSiteMetricsSnapshot {
             uptime: metrics.uptime,
             queued_curr_count: metrics.queued.curr.count,
@@ -859,6 +893,12 @@ impl StorageReplicationStatsHandle {
             proxy_delete_tag_failed: metrics.proxied.delete_tag_failed,
             replica_size: metrics.replica_size,
             replica_count: metrics.replica_count,
+            failed_count,
+            failed_bytes,
+            failed_last_minute_count,
+            failed_last_minute_bytes,
+            failed_last_hour_count,
+            failed_last_hour_bytes,
         }
     }
 
@@ -899,6 +939,12 @@ pub(crate) struct ReplicationSiteMetricsSnapshot {
     pub(crate) proxy_delete_tag_failed: i64,
     pub(crate) replica_size: i64,
     pub(crate) replica_count: i64,
+    pub(crate) failed_count: i64,
+    pub(crate) failed_bytes: i64,
+    pub(crate) failed_last_minute_count: i64,
+    pub(crate) failed_last_minute_bytes: i64,
+    pub(crate) failed_last_hour_count: i64,
+    pub(crate) failed_last_hour_bytes: i64,
 }
 
 pub(crate) async fn get_local_server_property() -> rustfs_madmin::ServerProperties {
@@ -2043,13 +2089,32 @@ pub(crate) async fn init_compression_total_memory_from_backend(store: Arc<ECStor
 #[cfg(test)]
 mod tests {
     use super::{
-        BUCKET_RESYNC_LOCK_RETRY_MAX_MS, apply_active_resync_intents, bucket_resync_transaction_lock_retry_ceiling_ms,
+        BUCKET_RESYNC_LOCK_RETRY_MAX_MS, StorageReplicationStatsHandle, apply_active_resync_intents,
-        bucket_resync_transaction_lock_retry_delay, bucket_resync_transaction_lock_retry_reason,
+        bucket_resync_transaction_lock_retry_ceiling_ms, bucket_resync_transaction_lock_retry_delay,
-        bucket_targets_metadata_lock_shard, ecstore_bucket, lock_bucket_targets_metadata, new_instance_ctx,
+        bucket_resync_transaction_lock_retry_reason, bucket_targets_metadata_lock_shard, ecstore_bucket,
-        retry_bucket_resync_transaction_lock, scanner_maintenance_config_file,
+        lock_bucket_targets_metadata, new_instance_ctx, retry_bucket_resync_transaction_lock, scanner_maintenance_config_file,
     };
     use std::time::Duration;
 
+    #[tokio::test]
+    async fn site_metrics_snapshot_includes_live_failure_windows() {
+        let stats = StorageReplicationStatsHandle::new();
+        let mut target = ecstore_bucket::replication::BucketReplicationStat::default();
+        target.fail_stats.add_size(2048, None::<&std::io::Error>);
+        let mut bucket = ecstore_bucket::replication::BucketReplicationStats::new();
+        bucket.stats.insert("arn:replication::remote:photos".to_string(), target);
+        stats.inner.cache.write().await.insert("photos".to_string(), bucket);
+
+        let snapshot = stats.site_metrics_snapshot().await;
+
+        assert_eq!(snapshot.failed_count, 1);
+        assert_eq!(snapshot.failed_bytes, 2048);
+        assert_eq!(snapshot.failed_last_minute_count, 1);
+        assert_eq!(snapshot.failed_last_minute_bytes, 2048);
+        assert_eq!(snapshot.failed_last_hour_count, 1);
+        assert_eq!(snapshot.failed_last_hour_bytes, 2048);
+    }
+
     #[tokio::test]
     async fn bucket_target_metadata_locks_serialize_only_matching_shards() {
         let bucket = "bucket-target-lock";

scripts/check_ci_paths_sync.sh

@@ -1,83 +0,0 @@
-#!/usr/bin/env bash
-# ci.yml's pull_request paths-ignore and ci-docs-only.yml's paths must be equal.
-#
-# ci-docs-only.yml exists to report the required checks for pull requests that
-# ci.yml skips. The two lists are the complement of each other, so any drift
-# breaks one of two ways, both silent:
-#
-#   - an entry only in ci.yml's paths-ignore: a PR touching only those files
-#     triggers neither workflow, nobody reports "Test and Lint" or "Quick
-#     Checks", and the PR waits on a required check forever;
-#   - an entry only in ci-docs-only.yml's paths: both workflows run, which is
-#     merely wasteful — but it also means the lists no longer describe the same
-#     intent, and the next edit is made against a wrong assumption.
-#
-# The push paths-ignore in ci.yml is deliberately NOT compared: no required
-# check is reported for push events, so it does not have to pair with anything.
-#
-# Also asserts ci-docs-only.yml still declares both companion job names, since a
-# rename there produces exactly the permanent-pending failure above.
-#
-# Usage: scripts/check_ci_paths_sync.sh
-set -euo pipefail
-
-cd "$(dirname "$0")/.."
-
-CI=".github/workflows/ci.yml"
-DOCS=".github/workflows/ci-docs-only.yml"
-
-# Print the quoted list items that follow $2 within the block introduced by $1.
-# Both files keep these as a flat list of quoted scalars, so no YAML parser is
-# needed and the script stays dependency-free like its check_* siblings.
-extract() {
-    local file="$1" event="$2" key="$3"
-    awk -v event="$event" -v key="$key" '
-        $0 ~ "^  " event ":[[:space:]]*$" { in_event = 1; next }
-        in_event && /^  [a-z_]+:[[:space:]]*$/ { in_event = 0 }
-        in_event && $0 ~ "^    " key ":[[:space:]]*$" { in_list = 1; next }
-        in_list {
-            if ($0 ~ /^      - /) {
-                item = $0
-                sub(/^      - /, "", item)
-                gsub(/^"|"$/, "", item)
-                print item
-                next
-            }
-            if ($0 !~ /^[[:space:]]*#/ && $0 !~ /^[[:space:]]*$/) in_list = 0
-        }
-    ' "$file" | sort
-}
-
-ci_list="$(extract "$CI" "pull_request" "paths-ignore")"
-docs_list="$(extract "$DOCS" "pull_request" "paths")"
-
-if [ -z "$ci_list" ] || [ -z "$docs_list" ]; then
-    echo "ERROR: could not read one of the path lists — did the file structure change?" >&2
-    echo "  $CI pull_request.paths-ignore: $(printf '%s' "$ci_list" | grep -c . || true) entries" >&2
-    echo "  $DOCS pull_request.paths: $(printf '%s' "$docs_list" | grep -c . || true) entries" >&2
-    exit 1
-fi
-
-status=0
-
-if ! diff_out="$(diff <(printf '%s\n' "$ci_list") <(printf '%s\n' "$docs_list"))"; then
-    echo "ERROR: $CI pull_request paths-ignore and $DOCS paths have drifted." >&2
-    echo "  '<' is only in $CI, '>' is only in $DOCS:" >&2
-    printf '%s\n' "$diff_out" | sed 's/^/    /' >&2
-    status=1
-fi
-
-for job_name in "Test and Lint" "Quick Checks"; do
-    if ! grep -q "name: ${job_name}\$" "$DOCS"; then
-        echo "ERROR: $DOCS no longer declares a job named '${job_name}'." >&2
-        echo "  It is a required status check; without a companion job here, a" >&2
-        echo "  docs-only PR waits on it forever." >&2
-        status=1
-    fi
-done
-
-if [ "$status" -ne 0 ]; then
-    exit 1
-fi
-
-echo "OK: ci.yml and ci-docs-only.yml path lists agree ($(printf '%s\n' "$ci_list" | wc -l | tr -d ' ') entries)"

scripts/check_s3s_footprint.sh

@@ -55,8 +55,11 @@ cd "$(dirname "$0")/.."
 # now reports an unreadable configuration as a plain string instead of raising
 # an S3 error per arm (24 invocation lines removed from
 # rustfs/src/admin/handlers/bucket_meta.rs; measured after merging the two).
+# 1589 -> 1588 on 2026-09-08: the GA blocker set (rustfs/backlog#2366) added
+# three invocation lines to the endpoint-refresh paths and folded the five
+# copies of the concurrent-change error into one constructor, netting -1.
 S3S_IMPORT_FILES_BASELINE=213
-S3_ERROR_LINES_BASELINE=1589
+S3_ERROR_LINES_BASELINE=1588
 # ecstore-scoped ratchet (rustfs/backlog#1842): the storage engine must not
 # know S3 wire/DTO types (ARCHITECTURE.md invariant 4). The S3-*consuming*
 # client was extracted to crates/s3-client, where s3s usage is legitimate;

scripts/check_test_wiring.py

@@ -542,7 +542,7 @@ def yaml_scalar_continues(lines: list[str], index: int, indent: int) -> bool:
 def check_quick_checks(root: Path) -> list[str]:
     errors: list[str] = []
     bypass_key = r'''(?:if|continue-on-error|needs|"if"|"continue-on-error"|"needs"|'if'|'continue-on-error'|'needs')\s*:'''
-    for name in ("ci.yml", "ci-docs-only.yml"):
+    for name in ("ci.yml",):
         relative = f".github/workflows/{name}"
         path = root / relative
         job = yaml_block(path.read_text().splitlines(), "quick-checks", 2) if path.is_file() else None
@@ -1165,7 +1165,6 @@ class SelfTests(unittest.TestCase):
                 ".github/workflows/ci.yml": caller.replace(
                     "    steps:", "    if: github.event_name != 'pull_request' || github.event.action != 'closed'\n    steps:"
                 ),
-                ".github/workflows/ci-docs-only.yml": caller,
                 ".github/actions/quick-checks/action.yml": action,
             }
             for relative, source in sources.items():
@@ -1173,7 +1172,7 @@ class SelfTests(unittest.TestCase):
                 path.parent.mkdir(parents=True, exist_ok=True)
                 path.write_text(source)
             self.assertEqual(check_quick_checks(root), [])
-            for relative in (".github/workflows/ci.yml", ".github/workflows/ci-docs-only.yml"):
+            for relative in (".github/workflows/ci.yml",):
                 source = sources[relative]
                 mutations = {
                     "different action": source.replace("./.github/actions/quick-checks", "./.github/actions/other"),

scripts/ci_gate.py

@@ -0,0 +1,369 @@
+#!/usr/bin/env python3
+"""Select safe documentation-only CI and verify the complete required job set."""
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path, PurePosixPath
+import re
+import subprocess
+import sys
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parent.parent
+ALWAYS_JOBS = ("classify-changes", "typos", "quick-checks")
+CODE_JOBS = (
+    "test-and-lint", "test-ilm-integration-serial", "test-and-lint-rio-v2",
+    "connect-short-credential-boundary", "test-and-lint-protocols",
+    "build-rustfs-debug-binary", "uring-integration", "e2e-tests",
+    "s3-implemented-tests", "s3-lifecycle-behavior-tests",
+)
+OPTIONAL_JOBS = ("build-rustfs-debug-binary-rio-v2", "e2e-tests-rio-v2", "e2e-full")
+NON_VALIDATION_JOBS = {"required-checks", "cancel-closed-pr-runs", "alert-on-failure"}
+
+
+def documentation_path(path: str) -> bool:
+    parts = PurePosixPath(path).parts
+    if not parts or path.startswith("/") or any(part in (".", "..") for part in parts) or any(ord(c) < 32 for c in path):
+        return False
+    if parts[-1] == "AGENTS.md":
+        return True
+    if len(parts) == 1 and (path.endswith(".md") or path == "LICENSE" or path.startswith("LICENSE-")):
+        return True
+    if path.startswith(("docs/", ".agents/skills/")) and path.endswith(".md"):
+        return True
+    return path.startswith("docs/") and path.endswith((".png", ".jpg", ".svg"))
+
+
+def select_mode(event: str, base: str, head: str, root: Path) -> str:
+    if event != "pull_request" or not all(re.fullmatch(r"[0-9a-f]{40}", sha) for sha in (base, head)):
+        return "full"
+    try:
+        changed = subprocess.check_output(
+            ["git", "diff", "--no-ext-diff", "--no-textconv", "--no-renames", "--name-only", "-z", base, head, "--"],
+            cwd=root, stderr=subprocess.PIPE,
+        ).decode("utf-8")
+    except (subprocess.CalledProcessError, UnicodeError):
+        return "full"
+    paths = changed.rstrip("\0").split("\0") if changed else []
+    return "docs" if paths and all(documentation_path(path) for path in paths) else "full"
+
+
+def expected_results(mode: str, event: str, ref: str) -> dict[str, str]:
+    if event not in ("pull_request", "push", "merge_group", "schedule", "workflow_dispatch"):
+        raise ValueError(f"unsupported CI event: {event!r}")
+    if mode not in ("docs", "full") or (mode == "docs" and event != "pull_request"):
+        raise ValueError(f"invalid CI selection: {mode!r} for {event!r}")
+    expected = {job: "success" for job in ALWAYS_JOBS}
+    expected.update({job: "success" if mode == "full" else "skipped" for job in CODE_JOBS})
+    rio = mode == "full" and event in ("schedule", "workflow_dispatch")
+    expected.update({job: "success" if rio else "skipped" for job in OPTIONAL_JOBS[:2]})
+    full = mode == "full" and (event in ("merge_group", "workflow_dispatch") or (event == "push" and ref == "refs/heads/main"))
+    expected["e2e-full"] = "success" if full else "skipped"
+    return expected
+
+
+def verify_results(needs: object, event: str, ref: str) -> list[str]:
+    if not isinstance(needs, dict):
+        return ["needs must be a job-result object"]
+    selection = needs.get("classify-changes", {})
+    outputs = selection.get("outputs", {}) if isinstance(selection, dict) else {}
+    mode = outputs.get("mode") if isinstance(outputs, dict) else None
+    try:
+        expected = expected_results(mode, event, ref)
+    except ValueError as error:
+        return [str(error)]
+    errors = []
+    if set(needs) != set(expected):
+        errors.append(f"job set differs: missing={sorted(set(expected) - set(needs))}, unexpected={sorted(set(needs) - set(expected))}")
+    for job, required in expected.items():
+        result = needs.get(job, {})
+        actual = result.get("result") if isinstance(result, dict) else None
+        if actual != required:
+            errors.append(f"{job}: expected {required}, got {actual!r}")
+    return errors
+
+
+def check_workflow(root: Path) -> list[str]:
+    # Reuse the repository's canonical-indentation checker; actionlint validates YAML syntax.
+    from check_test_wiring import yaml_block, yaml_scalar_continues
+
+    errors = []
+    lines = (root / ".github/workflows/ci.yml").read_text().splitlines()
+    jobs = yaml_block(lines, "jobs", 0) or []
+    names = set()
+    for index, line in enumerate(jobs):
+        if not re.match(r"^  \S", line) or line.lstrip().startswith("#"):
+            continue
+        header = re.fullmatch(r'''  (["']?)([A-Za-z_][A-Za-z0-9_-]*)\1\s*:\s*(?:#.*)?''', line)
+        if header is None:
+            errors.append("CI job declarations must use single-line job IDs")
+            continue
+        name = header[2]
+        if name in names:
+            errors.append(f"duplicate CI job ID: {name}")
+        names.add(name)
+        jobs[index] = f"  {name}:"
+    required = set(ALWAYS_JOBS + CODE_JOBS + OPTIONAL_JOBS)
+    if names - NON_VALIDATION_JOBS != required:
+        errors.append("CI verification jobs and the required gate contract differ")
+    for job in required:
+        block = yaml_block(jobs, job, 2) or []
+        if any(re.match(r"\s+(?:- )?[\"']?continue-on-error[\"']?\s*:", line) for line in block):
+            errors.append(f"{job} cannot convert a validation failure into success")
+    gate = yaml_block(jobs, "required-checks", 2) or []
+    def scalar(block, key, indent):
+        prefix = " " * indent + key + ": "
+        matches = [index for index, line in enumerate(block) if line.startswith(prefix)]
+        if len(matches) != 1:
+            return None
+        index = matches[0]
+        if yaml_scalar_continues(block, index, indent):
+            return None
+        return block[index][len(prefix):]
+
+    display_names = {}
+    for job in names:
+        block = [re.sub(r'''^    (?:'name'|"name")\s*:\s*''', "    name: ", line)
+                 for line in yaml_block(jobs, job, 2) or []]
+        value = scalar(block, "name", 4)
+        display = re.fullmatch(r'''(?:"([^"\\]*)"|'([^']*)'|([^'"#][^#]*?))(?:\s+#.*)?\s*''', (value or "").strip())
+        if display is None or (display[3] is not None and display[3].startswith(tuple("|>*&!{[?"))):
+            errors.append(f"{job} must use a verifiable single-line display name")
+            continue
+        name = next(value for value in display.groups() if value is not None)
+        if "${{" in name and (job != "test-and-lint-protocols" or name != "Test and Lint (${{ matrix.features.name }})"):
+            errors.append(f"{job} has an unverifiable dynamic display name")
+        display_names[job] = name
+
+    dependencies = yaml_block(gate, "needs", 4) or []
+    declared = [line.strip().removeprefix("- ") for line in dependencies if line.strip()]
+    if set(declared) != required or len(declared) != len(required):
+        errors.append("required-checks must directly depend on every verification job exactly once")
+    if display_names.get("required-checks") != "Test and Lint" or list(display_names.values()).count("Test and Lint") != 1:
+        errors.append("Test and Lint must uniquely name the aggregate gate")
+    if scalar(gate, "if", 4) != "always() && (github.event_name != 'pull_request' || github.event.action != 'closed')":
+        errors.append("required-checks must run after failed or skipped dependencies")
+    if scalar(gate, "shell", 8) != "bash" or scalar(gate, "run", 8) != "python3 scripts/ci_gate.py verify" or scalar(gate, "CI_NEEDS", 10) != "${{ toJSON(needs) }}":
+        errors.append("required-checks must verify the actual needs results")
+    if any(re.match(r'''\s+(?:- )?(?:["']?continue-on-error["']?\s*:|["']?if["']?\s*:)''', line) and not line.startswith("    if:") for line in gate):
+        errors.append("required-checks cannot ignore failures")
+    pr = yaml_block(lines, "pull_request", 2) or []
+    if any(line.strip().startswith(("paths:", "paths-ignore:")) for line in pr):
+        errors.append("all pull requests must enter the single CI workflow")
+    if (root / ".github/workflows/ci-docs-only.yml").exists():
+        errors.append("the duplicate required-status companion must be removed")
+    return errors
+
+
+class SelfTests(unittest.TestCase):
+    def test_documentation_paths_do_not_hide_build_or_fixture_changes(self):
+        for path in ("README.md", "AGENTS.md", "crates/utils/AGENTS.md", "docs/testing/README.md", "docs/diagram.svg", ".agents/skills/example/SKILL.md"):
+            self.assertTrue(documentation_path(path), path)
+        for path in ("", "src/lib.rs", "crates/foo/tests/fixtures/data.md", "Cargo.lock", "build.rs", "deploy/chart.yaml", ".github/workflows/ci.yml", "scripts/dev_build.sh", "assets/logo.png", "docs/test.rs", "README.md\n", "../README.md"):
+            self.assertFalse(documentation_path(path), path)
+
+    def test_git_range_includes_deleted_source_and_rename_origins(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            def git(*args):
+                return subprocess.check_output(["git", "-c", "core.hooksPath=/dev/null", "-c", "user.name=CI Test", "-c", "user.email=ci@example.invalid", *args], cwd=root, stderr=subprocess.PIPE).decode().strip()
+            git("init", "-q")
+            (root / "server.rs").write_text("fn main() {}\n")
+            (root / "README.md").write_text("old\n")
+            git("add", "."); git("commit", "-qm", "base")
+            base = git("rev-parse", "HEAD")
+            (root / "README.md").write_text("new\n")
+            git("add", "."); git("commit", "-qm", "docs")
+            docs = git("rev-parse", "HEAD")
+            self.assertEqual(select_mode("pull_request", base, docs, root), "docs")
+            (root / "server.rs").rename(root / "server.md")
+            git("add", "-A"); git("commit", "-qm", "rename source")
+            head = git("rev-parse", "HEAD")
+            self.assertEqual(select_mode("pull_request", base, head, root), "full")
+            self.assertEqual(select_mode("pull_request", docs, docs, root), "full")
+            self.assertEqual(select_mode("pull_request", "0" * 40, head, root), "full")
+            self.assertEqual(select_mode("pull_request", "--output=bad", head, root), "full")
+            self.assertEqual(select_mode("merge_group", base, docs, root), "full")
+
+    def test_event_contract_requires_complete_candidate_and_optional_lanes(self):
+        ordinary = expected_results("full", "pull_request", "refs/pull/1/merge")
+        self.assertEqual({job for job, state in ordinary.items() if state == "skipped"}, set(OPTIONAL_JOBS))
+        docs = expected_results("docs", "pull_request", "refs/pull/1/merge")
+        self.assertEqual({job for job, state in docs.items() if state == "success"}, set(ALWAYS_JOBS))
+        for event in ("schedule", "workflow_dispatch", "merge_group", "push"):
+            result = expected_results("full", event, "refs/heads/main")
+            self.assertEqual(result["e2e-full"], "skipped" if event == "schedule" else "success")
+            self.assertEqual(result["e2e-tests-rio-v2"], "success" if event in ("schedule", "workflow_dispatch") else "skipped")
+            with self.assertRaises(ValueError):
+                expected_results("docs", event, "refs/heads/main")
+
+    def test_every_wrong_result_missing_job_or_selection_fails_closed(self):
+        for mode, event in (("full", "pull_request"), ("docs", "pull_request"), ("full", "schedule"), ("full", "workflow_dispatch"), ("full", "merge_group")):
+            good = {job: {"result": value} for job, value in expected_results(mode, event, "refs/heads/main").items()}
+            good["classify-changes"]["outputs"] = {"mode": mode}
+            self.assertEqual(verify_results(good, event, "refs/heads/main"), [])
+            for job in good:
+                for value in ("success", "skipped", "failure", "cancelled", "neutral", "", None):
+                    if value == good[job]["result"]:
+                        continue
+                    with self.subTest(mode=mode, event=event, job=job, result=value):
+                        bad = {**good, job: {**good[job], "result": value}}
+                        self.assertTrue(verify_results(bad, event, "refs/heads/main"))
+                self.assertTrue(verify_results({key: value for key, value in good.items() if key != job}, event, "refs/heads/main"))
+                missing_result = {key: value for key, value in good[job].items() if key != "result"}
+                self.assertTrue(verify_results({**good, job: missing_result}, event, "refs/heads/main"))
+            self.assertTrue(verify_results({**good, "unknown-job": {"result": "success"}}, event, "refs/heads/main"))
+            for selection in ({}, {"mode": ""}, {"mode": True}, []):
+                bad = {**good, "classify-changes": {"result": "success", "outputs": selection}}
+                self.assertTrue(verify_results(bad, event, "refs/heads/main"))
+
+    def test_repository_wiring_and_missing_dependency_regression(self):
+        self.assertEqual(check_workflow(ROOT), [])
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            (root / ".github/workflows").mkdir(parents=True)
+            source = (ROOT / ".github/workflows/ci.yml").read_text()
+            path = root / ".github/workflows/ci.yml"
+            for job in ALWAYS_JOBS + CODE_JOBS + OPTIONAL_JOBS:
+                before, gate = source.split("  required-checks:\n", 1)
+                path.write_text(before + "  required-checks:\n" + gate.replace(f"      - {job}\n", "", 1))
+                self.assertTrue(check_workflow(root), job)
+            for old, new in (
+                ("run: python3 scripts/ci_gate.py verify", "run: python3 scripts/ci_gate.py verify || true"),
+                ("run: python3 scripts/ci_gate.py verify", "run: python3 scripts/ci_gate.py verify\n          || true"),
+                ("CI_NEEDS: ${{ toJSON(needs) }}", "CI_NEEDS: '{}'"),
+                ("name: Test and Lint\n", "name: Unrequired result\n"),
+                ("        shell: bash\n        run: python3 scripts/ci_gate.py verify", "        shell: echo {0}\n        run: python3 scripts/ci_gate.py verify"),
+                ("        shell: bash\n        run: python3 scripts/ci_gate.py verify", "        run: python3 scripts/ci_gate.py verify"),
+                ("        run: python3 scripts/ci_gate.py verify", '        "if": false\n        run: python3 scripts/ci_gate.py verify'),
+            ):
+                path.write_text(source.replace(old, new))
+                self.assertTrue(check_workflow(root), new)
+            for job in ALWAYS_JOBS + CODE_JOBS + OPTIONAL_JOBS:
+                for field in ("continue-on-error", '"continue-on-error"', "'continue-on-error'"):
+                    path.write_text(source.replace(f"  {job}:\n", f"  {job}:\n    {field}: true\n", 1))
+                    self.assertTrue(check_workflow(root), (job, field))
+                    before, block = source.split(f"  {job}:\n", 1)
+                    block = block.replace("      - name:", f"      - {field}: true\n        name:", 1)
+                    path.write_text(before + f"  {job}:\n" + block)
+                    self.assertTrue(check_workflow(root), (job, field, "step"))
+            path.write_text(source + "\n  cancel-after-test-and-lint-failure:\n    runs-on: ubuntu-latest\n")
+            self.assertTrue(check_workflow(root))
+
+    def test_job_ids_and_display_names_cannot_hide_validation(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            (root / ".github/workflows").mkdir(parents=True)
+            source = (ROOT / ".github/workflows/ci.yml").read_text()
+            path = root / ".github/workflows/ci.yml"
+            for header in ("typos", "'typos'", '"typos"'):
+                path.write_text(source.replace("  typos:\n", f"  {header}: # spelling\n"))
+                self.assertEqual(check_workflow(root), [], header)
+            for name in ("Test and Lint # required", "'Test and Lint'", '"Test and Lint" # required'):
+                path.write_text(source.replace("    name: Test and Lint\n", f"    name: {name}\n"))
+                self.assertEqual(check_workflow(root), [], name)
+            for key in ("'name'", '"name"'):
+                path.write_text(source.replace("    name: Typos\n", f"    {key}: Typos\n"))
+                self.assertEqual(check_workflow(root), [], key)
+            for header in ("new_test", "NewTest", "_new_test", "'new_test'", '"new_test"', '"new\\u005ftest"'):
+                path.write_text(source + f"\n  {header}:\n    name: New test\n    runs-on: ubuntu-latest\n    steps:\n      - run: exit 1\n")
+                self.assertTrue(check_workflow(root), header)
+            path.write_text(source + "\n  'typos':\n    name: Duplicate\n    runs-on: ubuntu-latest\n    steps:\n      - run: exit 1\n")
+            self.assertIn("duplicate CI job ID: typos", check_workflow(root))
+            for name in (
+                "Test and Lint", "Test and Lint # duplicate", "'Test and Lint'",
+                '"Test and Lint" # duplicate', '"Test\\u0020and Lint"',
+                ">-\n      Test and Lint", "|-\n      Test and Lint", "Test and\n      Lint",
+                "*required_name", "&required_name Test and Lint", "!!str Test and Lint",
+                "${{ 'Test and Lint' }}", '"${{ github.event.inputs.check_name }}"',
+            ):
+                path.write_text(source.replace("    name: Typos\n", f"    name: {name}\n"))
+                self.assertTrue(check_workflow(root), name)
+            path.write_text(source.replace("    name: Typos\n", ""))
+            self.assertIn("typos must use a verifiable single-line display name", check_workflow(root))
+
+    def test_verify_command_preserves_failures(self):
+        good = {job: {"result": value} for job, value in expected_results("full", "pull_request", "refs/pull/1/merge").items()}
+        good["classify-changes"]["outputs"] = {"mode": "full"}
+        failed = {**good, "e2e-tests": {"result": "failure"}}
+        for needs, code in ((json.dumps(good), 0), (json.dumps(failed), 1), ("{}", 1), ("{", 1)):
+            with self.subTest(needs=needs):
+                env = dict(os.environ, CI_NEEDS=needs, GITHUB_EVENT_NAME="pull_request", GITHUB_REF="refs/pull/1/merge")
+                result = subprocess.run([sys.executable, str(Path(__file__).resolve()), "verify"], env=env, capture_output=True, text=True)
+                self.assertEqual(result.returncode, code, result.stderr)
+                self.assertIn("ERROR:" if code else "CI contract passed", result.stderr if code else result.stdout)
+
+    def test_actual_selector_bootstrap_uses_base_policy_and_fails_closed(self):
+        from check_test_wiring import yaml_block
+        jobs = yaml_block((ROOT / ".github/workflows/ci.yml").read_text().splitlines(), "jobs", 0)
+        selector = yaml_block(jobs, "classify-changes", 2)
+        body = "\n".join(line[10:] for line in selector[selector.index("        run: |") + 1:])
+        for event, changed, base_sha, available, broken, expected in (
+            ("pull_request", "README.md", "b" * 40, True, False, "docs"),
+            ("pull_request", "src/server.rs", "b" * 40, True, False, "full"),
+            ("pull_request", "README.md", "b" * 40, False, False, "full"),
+            ("merge_group", "README.md", "b" * 40, False, False, "full"),
+            ("pull_request", "README.md", "b" * 40, True, True, None),
+            ("pull_request", "README.md", "", True, True, "full"),
+        ):
+            with self.subTest(event=event, changed=changed, available=available, broken=broken), tempfile.TemporaryDirectory() as directory:
+                root = Path(directory)
+                (root / "scripts").mkdir()
+                (root / "scripts/ci_gate.py").write_text("raise SystemExit(71)\n")
+                (root / "python3").symlink_to(sys.executable)
+                base = root / "base-policy.py"
+                base.write_text("raise SystemExit(29)\n" if broken else Path(__file__).read_text())
+                git = root / "git"
+                git.write_text('''#!/bin/sh
+if [ "$1" = show ]; then
+  [ "$2" = "$CI_BASE_SHA:scripts/ci_gate.py" ] || exit 19
+  [ "$BASE_AVAILABLE" = yes ] || exit 128
+  cat "$BASE_POLICY"
+elif [ "$1" = diff ]; then
+  printf '%s\\0' "$CHANGED_PATH"
+else
+  exit 20
+fi
+''')
+                git.chmod(0o755)
+                output = root / "output"
+                output.touch()
+                env = dict(os.environ, GITHUB_EVENT_NAME=event, CI_BASE_SHA=base_sha, GITHUB_SHA="c" * 40,
+                           RUNNER_TEMP=str(root), GITHUB_OUTPUT=str(output), BASE_POLICY=str(base),
+                           BASE_AVAILABLE="yes" if available else "no", CHANGED_PATH=changed,
+                           PATH=f"{root}{os.pathsep}{os.environ['PATH']}")
+                result = subprocess.run(["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", body], cwd=root, env=env, capture_output=True, text=True)
+                self.assertEqual(result.returncode, 29 if expected is None else 0, result.stderr)
+                self.assertEqual(output.read_text(), "" if expected is None else f"mode={expected}\n")
+
+
+def main() -> int:
+    if sys.argv[1:] == ["--self-test"]:
+        return not unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(SelfTests)).wasSuccessful()
+    if sys.argv[1:] == ["select"]:
+        mode = select_mode(os.environ.get("GITHUB_EVENT_NAME", ""), os.environ.get("CI_BASE_SHA", ""), os.environ.get("GITHUB_SHA", ""), Path.cwd())
+        with open(os.environ["GITHUB_OUTPUT"], "a") as output:
+            output.write(f"mode={mode}\n")
+        print(f"CI selection: {mode}")
+        return 0
+    if sys.argv[1:] == ["verify"]:
+        try:
+            errors = verify_results(json.loads(os.environ["CI_NEEDS"]), os.environ.get("GITHUB_EVENT_NAME", ""), os.environ.get("GITHUB_REF", ""))
+        except (KeyError, ValueError) as error:
+            errors = [str(error)]
+    elif sys.argv[1:] == ["--check-workflow"]:
+        errors = check_workflow(ROOT)
+    else:
+        print("usage: ci_gate.py {select|verify|--check-workflow|--self-test}", file=sys.stderr)
+        return 2
+    for error in errors:
+        print(f"ERROR: {error}", file=sys.stderr)
+    if not errors:
+        print("CI contract passed")
+    return bool(errors)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/release/publish_nightly_assets.sh

@@ -0,0 +1,87 @@
+#!/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"

scripts/test_nightly_candidate.py

@@ -163,12 +163,13 @@ SH
                 self.assertFalse(self.store.exists())
                 self.assertEqual(list(self.root.glob("nightly-awscli.*")), [])
 
-    def test_checkout_sha_mismatch_fails_before_upload(self):
+    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.
         result = self.run_publish(GITHUB_SHA="f" * 40)
-        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(result.returncode, 0, result.stderr)
-        self.assertIn("Checkout SHA", result.stderr)
+        self.assertEqual(self.manifest()["source_sha"], self.sha)
-        self.assertFalse(self.output.exists())
-        self.assertFalse((self.root / "aws.log").exists())
 
     def test_same_date_builds_and_reruns_keep_distinct_candidates(self):
         urls = []

scripts/test_security_workflow.py

@@ -837,6 +837,55 @@ emit_step_result() {
             self.assertIn(value, contents)
         self.assertNotIn("OLD RUN EVIDENCE", contents)
 
+    def test_performance_commands_bind_runner_selection_and_preserve_failures(self) -> None:
+        self.prepare("performance")
+        source = self.source.splitlines()
+        job = yaml_block(source, "performance-test", 2)
+        runner = WorkflowSteps()
+        runner.directory = self.directory / "workspace with spaces"
+        scripts = runner.directory / "auto-testing"
+        scripts.mkdir(parents=True)
+        wrapper = scripts / "rustfs_performance_test.sh"
+        wrapper.write_text(f"#!{sys.executable}\nimport json, os, sys\n" +
+                           "print(json.dumps({'args': sys.argv[1:], 'env': {key: os.environ.get(key) for key in " +
+                           "('RUSTFS_BENCH_SCRIPT', 'RUSTFS_WARP_METHODS', 'RUSTFS_WARP_SIZES', " +
+                           "'RUSTFS_WARP_DURATION', 'RUSTFS_WARP_CONCURRENCY', 'WARP_METHODS', " +
+                           "'WARP_SIZES', 'WARP_DURATION', 'WARP_CONCURRENCY')}}))\n" +
+                           "sys.exit(int(os.environ['FAKE_BENCH_EXIT']))\n")
+        wrapper.chmod(0o755)
+        runner.steps = named_steps(job)
+        for methods, sizes, duration, concurrency in (
+            ("get", "1KiB", "1s", "7"), ("all", "all", "5m", "64"), ("", "", "5m", "64")
+        ):
+            runner.context = {"github.workspace": str(runner.directory), "inputs.test_method": methods,
+                              "inputs.object_size": sizes, "inputs.warp_duration || '5m'": duration,
+                              "inputs.warp_concurrency || '64'": concurrency}
+            runner.env = {**self.env, "RUSTFS_BENCH_SCRIPT": "/unverified/home-script.sh",
+                          "RUSTFS_WARP_METHODS": "put", "RUSTFS_WARP_SIZES": "64MiB",
+                          "RUSTFS_WARP_DURATION": "99h", "RUSTFS_WARP_CONCURRENCY": "2",
+                          "WARP_DURATION": "88h", "WARP_CONCURRENCY": "3", "WARP_METHODS": "mixed", "WARP_SIZES": "32MiB",
+                          "LOG_FILE": str(self.directory / "suite.log")}
+            runner.env.update(runner.step_env(job, indent=4))
+            for step, number in (("Run benchmark (GET/PUT/MIXED)", "5"), ("Analyze results", "6")):
+                for code in (0, 42):
+                    with self.subTest(methods=methods, sizes=sizes, step=step, exit=code):
+                        runner.env["FAKE_BENCH_EXIT"] = str(code)
+                        result = runner.run_step(step)
+                        self.assertEqual(result.returncode, code, result.stderr)
+                        invocation = json.loads(result.stdout)
+                        expected = ["--step", number, "-y", "--log-file", runner.env["LOG_FILE"]]
+                        self.assertEqual(invocation["args"], expected)
+                        self.assertEqual(invocation["env"]["RUSTFS_BENCH_SCRIPT"], str(scripts / "rustfs_performance_testing.sh"))
+                        self.assertEqual(invocation["env"]["RUSTFS_WARP_METHODS"], methods)
+                        self.assertEqual(invocation["env"]["RUSTFS_WARP_SIZES"], sizes)
+                        self.assertEqual(invocation["env"]["RUSTFS_WARP_DURATION"], duration)
+                        self.assertEqual(invocation["env"]["RUSTFS_WARP_CONCURRENCY"], concurrency)
+                        if number == "6":
+                            self.assertEqual(invocation["env"]["WARP_METHODS"], methods)
+                            self.assertEqual(invocation["env"]["WARP_SIZES"], sizes)
+                            self.assertEqual(invocation["env"]["WARP_DURATION"], duration)
+                            self.assertEqual(invocation["env"]["WARP_CONCURRENCY"], concurrency)
+
 
 if __name__ == "__main__":
     unittest.main()