fix(s3): align encrypted checksums and multipart completion (#7025)

* fix(s3): align encrypted checksum handling

* test(s3): align multipart SSE-C completion

* fix(ecstore): scope startup helper to tests

.github/workflows/audit.yml

@@ -28,6 +28,7 @@ on:
       - 'scripts/test_package_versions.sh'
       - 'scripts/security/check_performance_ab_workflow.sh'
       - 'scripts/security/check_preview_release_workflow.sh'
+      - 'scripts/security/check_tier_artifact_workflow.sh'
       - 'scripts/security/check_workflow_pins.sh'
   pull_request:
     types: [ opened, synchronize, reopened, closed ]
@@ -43,6 +44,7 @@ on:
       - 'scripts/test_package_versions.sh'
       - 'scripts/security/check_performance_ab_workflow.sh'
       - 'scripts/security/check_preview_release_workflow.sh'
+      - 'scripts/security/check_tier_artifact_workflow.sh'
       - 'scripts/security/check_workflow_pins.sh'
   schedule:
     # Daily, not weekly. This schedule exists to catch RustSec advisories
@@ -150,6 +152,9 @@ jobs:
       - name: Check performance A/B workflow trust boundary
         run: ./scripts/security/check_performance_ab_workflow.sh
 
+      - name: Check tier evidence workflow isolation
+        run: ./scripts/security/check_tier_artifact_workflow.sh
+
       - name: Check package version contract
         run: ./scripts/test_package_versions.sh
 

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

@@ -284,21 +284,52 @@ jobs:
 
       - name: "Continue functional chain (next: Pool expansion)"
         # Only chain-triggered runs forward to the next suite; standalone
-        # workflow_dispatch runs stop after their own cleanup.
+        # workflow_dispatch runs stop after their own cleanup. A failed
+        # handoff must never pass silently: it retries, then files an alert
+        # issue in rustfs/backlog so a stalled chain is visible.
         if: ${{ always() && github.event_name == 'repository_dispatch' }}
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
         run: |
-          set -euo pipefail
-          if [ -z "${GH_TOKEN:-}" ]; then
+          set -uo pipefail
+          if [ -z "${{GH_TOKEN:-}}" ]; then
             echo "PF_TESTING_GH_TOKEN is not configured; cannot dispatch the next suite" >&2
             exit 1
           fi
-          echo "Dispatching next functional suite: Pool expansion"
-          gh api --method POST repos/rustfs/rustfs/dispatches \
-            -f event_type='rustfs-chain-pool' \
-            -F 'client_payload[from_suite]=heal'
+          DISPATCHED=0
+          for attempt in 1 2 3; do
+            if gh api --method POST repos/rustfs/rustfs/dispatches \
+              -f event_type='rustfs-chain-pool' \
+              -F 'client_payload[from_suite]=heal'; then
+              echo "dispatched next suite Pool expansion (attempt ${{attempt}})"
+              DISPATCHED=1
+              break
+            fi
+            echo "dispatch attempt ${{attempt}} failed; retrying in ${{attempt}}0s" >&2
+            sleep "${{attempt}}0"
+          done
+          if [ "${{DISPATCHED:-0}}" -ne 1 ]; then
+            echo "ERROR: functional chain stalled: could not dispatch Pool expansion after 3 attempts" >&2
+            TITLE="[functional][chain] stalled after heal (run ${{GITHUB_RUN_ID}})"
+            BODY_FILE="$(mktemp)"
+            {
+              echo "The functional chain could not hand off from **heal** to **Pool expansion** after 3 attempts."
+              echo ""
+              echo "- Failed suite job: ${{GITHUB_SERVER_URL}}/${{GITHUB_REPOSITORY}}/actions/runs/${{GITHUB_RUN_ID}}"
+              echo "- Expected next event: `rustfs-chain-pool`"
+              echo "- Likely cause: PF_TESTING_GH_TOKEN lacks contents:write on rustfs/rustfs, or the GitHub API was unavailable."
+              echo "- Recovery: re-dispatch manually with"
+              echo "  ```"
+              echo "  gh api --method POST repos/rustfs/rustfs/dispatches -f event_type='rustfs-chain-pool'"
+              echo "  ```"
+            } > "${{BODY_FILE}}"
+            gh issue create -R rustfs/backlog --title "${{TITLE}}" \
+              --body-file "${{BODY_FILE}}" --label functional-test \
+              || gh issue create -R rustfs/backlog --title "${{TITLE}}" --body-file "${{BODY_FILE}}" \
+              || echo "could not file the stall alert issue either; check the token" >&2
+            exit 1
+          fi
 
       - name: Notify on failure
         if: failure()

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

@@ -340,21 +340,52 @@ jobs:
 
       - name: "Continue functional chain (next: Tier)"
         # Only chain-triggered runs forward to the next suite; standalone
-        # workflow_dispatch runs stop after their own cleanup.
+        # workflow_dispatch runs stop after their own cleanup. A failed
+        # handoff must never pass silently: it retries, then files an alert
+        # issue in rustfs/backlog so a stalled chain is visible.
         if: ${{ always() && github.event_name == 'repository_dispatch' }}
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
         run: |
-          set -euo pipefail
-          if [ -z "${GH_TOKEN:-}" ]; then
+          set -uo pipefail
+          if [ -z "${{GH_TOKEN:-}}" ]; then
             echo "PF_TESTING_GH_TOKEN is not configured; cannot dispatch the next suite" >&2
             exit 1
           fi
-          echo "Dispatching next functional suite: Tier"
-          gh api --method POST repos/rustfs/rustfs/dispatches \
-            -f event_type='rustfs-chain-tier' \
-            -F 'client_payload[from_suite]=kms'
+          DISPATCHED=0
+          for attempt in 1 2 3; do
+            if gh api --method POST repos/rustfs/rustfs/dispatches \
+              -f event_type='rustfs-chain-tier' \
+              -F 'client_payload[from_suite]=kms'; then
+              echo "dispatched next suite Tier (attempt ${{attempt}})"
+              DISPATCHED=1
+              break
+            fi
+            echo "dispatch attempt ${{attempt}} failed; retrying in ${{attempt}}0s" >&2
+            sleep "${{attempt}}0"
+          done
+          if [ "${{DISPATCHED:-0}}" -ne 1 ]; then
+            echo "ERROR: functional chain stalled: could not dispatch Tier after 3 attempts" >&2
+            TITLE="[functional][chain] stalled after kms (run ${{GITHUB_RUN_ID}})"
+            BODY_FILE="$(mktemp)"
+            {
+              echo "The functional chain could not hand off from **kms** to **Tier** after 3 attempts."
+              echo ""
+              echo "- Failed suite job: ${{GITHUB_SERVER_URL}}/${{GITHUB_REPOSITORY}}/actions/runs/${{GITHUB_RUN_ID}}"
+              echo "- Expected next event: `rustfs-chain-tier`"
+              echo "- Likely cause: PF_TESTING_GH_TOKEN lacks contents:write on rustfs/rustfs, or the GitHub API was unavailable."
+              echo "- Recovery: re-dispatch manually with"
+              echo "  ```"
+              echo "  gh api --method POST repos/rustfs/rustfs/dispatches -f event_type='rustfs-chain-tier'"
+              echo "  ```"
+            } > "${{BODY_FILE}}"
+            gh issue create -R rustfs/backlog --title "${{TITLE}}" \
+              --body-file "${{BODY_FILE}}" --label functional-test \
+              || gh issue create -R rustfs/backlog --title "${{TITLE}}" --body-file "${{BODY_FILE}}" \
+              || echo "could not file the stall alert issue either; check the token" >&2
+            exit 1
+          fi
 
       - name: Notify on failure
         if: failure()

.github/workflows/rustfs-pool-expand-test.yml

@@ -596,21 +596,52 @@ jobs:
 
       - name: "Continue functional chain (next: Security)"
         # Only chain-triggered runs forward to the next suite; standalone
-        # workflow_dispatch runs stop after their own cleanup.
+        # workflow_dispatch runs stop after their own cleanup. A failed
+        # handoff must never pass silently: it retries, then files an alert
+        # issue in rustfs/backlog so a stalled chain is visible.
         if: ${{ always() && github.event_name == 'repository_dispatch' }}
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
         run: |
-          set -euo pipefail
-          if [ -z "${GH_TOKEN:-}" ]; then
+          set -uo pipefail
+          if [ -z "${{GH_TOKEN:-}}" ]; then
             echo "PF_TESTING_GH_TOKEN is not configured; cannot dispatch the next suite" >&2
             exit 1
           fi
-          echo "Dispatching next functional suite: Security"
-          gh api --method POST repos/rustfs/rustfs/dispatches \
-            -f event_type='rustfs-chain-security' \
-            -F 'client_payload[from_suite]=pool'
+          DISPATCHED=0
+          for attempt in 1 2 3; do
+            if gh api --method POST repos/rustfs/rustfs/dispatches \
+              -f event_type='rustfs-chain-security' \
+              -F 'client_payload[from_suite]=pool'; then
+              echo "dispatched next suite Security (attempt ${{attempt}})"
+              DISPATCHED=1
+              break
+            fi
+            echo "dispatch attempt ${{attempt}} failed; retrying in ${{attempt}}0s" >&2
+            sleep "${{attempt}}0"
+          done
+          if [ "${{DISPATCHED:-0}}" -ne 1 ]; then
+            echo "ERROR: functional chain stalled: could not dispatch Security after 3 attempts" >&2
+            TITLE="[functional][chain] stalled after pool (run ${{GITHUB_RUN_ID}})"
+            BODY_FILE="$(mktemp)"
+            {
+              echo "The functional chain could not hand off from **pool** to **Security** after 3 attempts."
+              echo ""
+              echo "- Failed suite job: ${{GITHUB_SERVER_URL}}/${{GITHUB_REPOSITORY}}/actions/runs/${{GITHUB_RUN_ID}}"
+              echo "- Expected next event: `rustfs-chain-security`"
+              echo "- Likely cause: PF_TESTING_GH_TOKEN lacks contents:write on rustfs/rustfs, or the GitHub API was unavailable."
+              echo "- Recovery: re-dispatch manually with"
+              echo "  ```"
+              echo "  gh api --method POST repos/rustfs/rustfs/dispatches -f event_type='rustfs-chain-security'"
+              echo "  ```"
+            } > "${{BODY_FILE}}"
+            gh issue create -R rustfs/backlog --title "${{TITLE}}" \
+              --body-file "${{BODY_FILE}}" --label functional-test \
+              || gh issue create -R rustfs/backlog --title "${{TITLE}}" --body-file "${{BODY_FILE}}" \
+              || echo "could not file the stall alert issue either; check the token" >&2
+            exit 1
+          fi
 
       - name: Notify on failure
         if: failure()

.github/workflows/rustfs-s3-compat-test.yml

@@ -320,21 +320,52 @@ jobs:
 
       - name: "Continue functional chain (next: KMS)"
         # Only chain-triggered runs forward to the next suite; standalone
-        # workflow_dispatch runs stop after their own cleanup.
+        # workflow_dispatch runs stop after their own cleanup. A failed
+        # handoff must never pass silently: it retries, then files an alert
+        # issue in rustfs/backlog so a stalled chain is visible.
         if: ${{ always() && github.event_name == 'repository_dispatch' }}
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
         run: |
-          set -euo pipefail
-          if [ -z "${GH_TOKEN:-}" ]; then
+          set -uo pipefail
+          if [ -z "${{GH_TOKEN:-}}" ]; then
             echo "PF_TESTING_GH_TOKEN is not configured; cannot dispatch the next suite" >&2
             exit 1
           fi
-          echo "Dispatching next functional suite: KMS"
-          gh api --method POST repos/rustfs/rustfs/dispatches \
-            -f event_type='rustfs-chain-kms' \
-            -F 'client_payload[from_suite]=s3'
+          DISPATCHED=0
+          for attempt in 1 2 3; do
+            if gh api --method POST repos/rustfs/rustfs/dispatches \
+              -f event_type='rustfs-chain-kms' \
+              -F 'client_payload[from_suite]=s3'; then
+              echo "dispatched next suite KMS (attempt ${{attempt}})"
+              DISPATCHED=1
+              break
+            fi
+            echo "dispatch attempt ${{attempt}} failed; retrying in ${{attempt}}0s" >&2
+            sleep "${{attempt}}0"
+          done
+          if [ "${{DISPATCHED:-0}}" -ne 1 ]; then
+            echo "ERROR: functional chain stalled: could not dispatch KMS after 3 attempts" >&2
+            TITLE="[functional][chain] stalled after s3 (run ${{GITHUB_RUN_ID}})"
+            BODY_FILE="$(mktemp)"
+            {
+              echo "The functional chain could not hand off from **s3** to **KMS** after 3 attempts."
+              echo ""
+              echo "- Failed suite job: ${{GITHUB_SERVER_URL}}/${{GITHUB_REPOSITORY}}/actions/runs/${{GITHUB_RUN_ID}}"
+              echo "- Expected next event: `rustfs-chain-kms`"
+              echo "- Likely cause: PF_TESTING_GH_TOKEN lacks contents:write on rustfs/rustfs, or the GitHub API was unavailable."
+              echo "- Recovery: re-dispatch manually with"
+              echo "  ```"
+              echo "  gh api --method POST repos/rustfs/rustfs/dispatches -f event_type='rustfs-chain-kms'"
+              echo "  ```"
+            } > "${{BODY_FILE}}"
+            gh issue create -R rustfs/backlog --title "${{TITLE}}" \
+              --body-file "${{BODY_FILE}}" --label functional-test \
+              || gh issue create -R rustfs/backlog --title "${{TITLE}}" --body-file "${{BODY_FILE}}" \
+              || echo "could not file the stall alert issue either; check the token" >&2
+            exit 1
+          fi
 
       - name: Notify on failure
         if: failure()

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

@@ -335,21 +335,52 @@ jobs:
 
       - name: "Continue functional chain (next: Heal)"
         # Only chain-triggered runs forward to the next suite; standalone
-        # workflow_dispatch runs stop after their own cleanup.
+        # workflow_dispatch runs stop after their own cleanup. A failed
+        # handoff must never pass silently: it retries, then files an alert
+        # issue in rustfs/backlog so a stalled chain is visible.
         if: ${{ always() && github.event_name == 'repository_dispatch' }}
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
         run: |
-          set -euo pipefail
-          if [ -z "${GH_TOKEN:-}" ]; then
+          set -uo pipefail
+          if [ -z "${{GH_TOKEN:-}}" ]; then
             echo "PF_TESTING_GH_TOKEN is not configured; cannot dispatch the next suite" >&2
             exit 1
           fi
-          echo "Dispatching next functional suite: Heal"
-          gh api --method POST repos/rustfs/rustfs/dispatches \
-            -f event_type='rustfs-chain-heal' \
-            -F 'client_payload[from_suite]=storage'
+          DISPATCHED=0
+          for attempt in 1 2 3; do
+            if gh api --method POST repos/rustfs/rustfs/dispatches \
+              -f event_type='rustfs-chain-heal' \
+              -F 'client_payload[from_suite]=storage'; then
+              echo "dispatched next suite Heal (attempt ${{attempt}})"
+              DISPATCHED=1
+              break
+            fi
+            echo "dispatch attempt ${{attempt}} failed; retrying in ${{attempt}}0s" >&2
+            sleep "${{attempt}}0"
+          done
+          if [ "${{DISPATCHED:-0}}" -ne 1 ]; then
+            echo "ERROR: functional chain stalled: could not dispatch Heal after 3 attempts" >&2
+            TITLE="[functional][chain] stalled after storage (run ${{GITHUB_RUN_ID}})"
+            BODY_FILE="$(mktemp)"
+            {
+              echo "The functional chain could not hand off from **storage** to **Heal** after 3 attempts."
+              echo ""
+              echo "- Failed suite job: ${{GITHUB_SERVER_URL}}/${{GITHUB_REPOSITORY}}/actions/runs/${{GITHUB_RUN_ID}}"
+              echo "- Expected next event: `rustfs-chain-heal`"
+              echo "- Likely cause: PF_TESTING_GH_TOKEN lacks contents:write on rustfs/rustfs, or the GitHub API was unavailable."
+              echo "- Recovery: re-dispatch manually with"
+              echo "  ```"
+              echo "  gh api --method POST repos/rustfs/rustfs/dispatches -f event_type='rustfs-chain-heal'"
+              echo "  ```"
+            } > "${{BODY_FILE}}"
+            gh issue create -R rustfs/backlog --title "${{TITLE}}" \
+              --body-file "${{BODY_FILE}}" --label functional-test \
+              || gh issue create -R rustfs/backlog --title "${{TITLE}}" --body-file "${{BODY_FILE}}" \
+              || echo "could not file the stall alert issue either; check the token" >&2
+            exit 1
+          fi
 
       - name: Notify on failure
         if: failure()

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

@@ -54,6 +54,18 @@ jobs:
     timeout-minutes: 420
     if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
     steps:
+      - name: Initialize run evidence directory
+        id: evidence
+        run: |
+          set -euo pipefail
+          umask 077
+          if ! mkdir -- "${TIER_ARTIFACTS_DIR}"; then
+            echo "refusing to reuse tier evidence path: ${TIER_ARTIFACTS_DIR}" >&2
+            exit 1
+          fi
+          test -d "${TIER_ARTIFACTS_DIR}"
+          test ! -L "${TIER_ARTIFACTS_DIR}"
+
       # auto-testing is private: clone it with the dedicated PF token (not
       # GITHUB_TOKEN) and retry transient GitHub/network failures.
       - name: Checkout auto-testing scripts (with retry)
@@ -133,11 +145,11 @@ jobs:
         id: test
         continue-on-error: true
         env:
-          LOG_FILE: /tmp/rustfs-tier.log
           PACKAGE_URL_INPUT: ${{ inputs.package_url }}
           RUSTFS_VERSION_INPUT: ${{ inputs.rustfs_version }}
         run: |
           set -euo pipefail
+          LOG_FILE="${TIER_ARTIFACTS_DIR}/rustfs-tier.log"
           chmod +x auto-testing/rustfs-tier-test.sh
           RC_BIN="$(command -v rc)"
           PACKAGE_URL="${PACKAGE_URL_INPUT}"
@@ -162,7 +174,7 @@ jobs:
           ./auto-testing/rustfs-tier-test.sh "${ARGS[@]}"
 
       - name: Inject diagnostic case failure
-        if: ${{ always() && inputs.force_case_failure }}
+        if: ${{ always() && steps.evidence.outcome == 'success' && inputs.force_case_failure }}
         run: |
           set -euo pipefail
           RESULT_FILE="${TIER_ARTIFACTS_DIR}/cases/single-single--TIER-101.json"
@@ -172,12 +184,8 @@ jobs:
           mv "${TMP_FILE}" "${RESULT_FILE}"
 
       - name: Generate report
-        if: always()
+        if: ${{ always() && steps.evidence.outcome == 'success' }}
         env:
-          LOG_FILE: /tmp/rustfs-tier.log
-          REPORT_FILE: /tmp/rustfs-tier-report.md
-          CASE_TABLE: /tmp/rustfs-tier-cases.md
-          GATE_RC_FILE: /tmp/rustfs-tier-gate.rc
           PACKAGE_URL_INPUT: ${{ inputs.package_url }}
           RUSTFS_VERSION_INPUT: ${{ inputs.rustfs_version }}
           TEST_OUTCOME: ${{ steps.test.outcome }}
@@ -185,6 +193,12 @@ jobs:
           TRIGGER_NAME: ${{ github.event_name }}
         run: |
           set -euo pipefail
+          test -d "${TIER_ARTIFACTS_DIR}"
+          test ! -L "${TIER_ARTIFACTS_DIR}"
+          LOG_FILE="${TIER_ARTIFACTS_DIR}/rustfs-tier.log"
+          REPORT_FILE="${TIER_ARTIFACTS_DIR}/rustfs-tier-report.md"
+          CASE_TABLE="${TIER_ARTIFACTS_DIR}/rustfs-tier-cases.md"
+          GATE_RC_FILE="${TIER_ARTIFACTS_DIR}/rustfs-tier-gate.rc"
           PACKAGE_URL="${PACKAGE_URL_INPUT}"
           RUSTFS_VERSION="${RUSTFS_VERSION_INPUT}"
           if [ -n "${PACKAGE_URL}" ]; then
@@ -228,11 +242,11 @@ jobs:
           cat "${REPORT_FILE}" >> "${GITHUB_STEP_SUMMARY}"
 
       - name: Upload functional report to dashboard
-        if: always()
+        if: ${{ always() && steps.evidence.outcome == 'success' }}
         continue-on-error: true
         env:
           GH_TOKEN: ${{ env.PF_TESTING_GH_TOKEN }}
-          REPORT_FILE: /tmp/rustfs-tier-report.md
+          REPORT_FILE: ${{ env.TIER_ARTIFACTS_DIR }}/rustfs-tier-report.md
           SUITE: tier
         run: |
           set -euo pipefail
@@ -254,16 +268,110 @@ jobs:
               | gh api --method PUT "repos/rustfs/dashboard/contents/${REPORT_PATH}" --input - >/dev/null
           fi
 
+      - name: Verify required tier evidence
+        id: evidence_verify
+        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        run: |
+          set -euo pipefail
+          failed=0
+          for name in \
+            rustfs-tier.log \
+            rustfs-tier-report.md \
+            rustfs-tier-cases.md \
+            rustfs-tier-gate.rc \
+            provenance.json; do
+            if [ ! -s "${TIER_ARTIFACTS_DIR}/${name}" ]; then
+              echo "required tier evidence is missing or empty: ${name}" >&2
+              failed=1
+            fi
+          done
+          for name in cases logs; do
+            if [ ! -d "${TIER_ARTIFACTS_DIR}/${name}" ]; then
+              echo "required tier evidence directory is missing: ${name}" >&2
+              failed=1
+            fi
+          done
+          if ! find "${TIER_ARTIFACTS_DIR}/cases" -maxdepth 1 -type f -name '*.json' -print -quit 2>/dev/null | grep -q .; then
+            echo "no atomic tier case result was produced" >&2
+            failed=1
+          fi
+          [ "${failed}" -eq 0 ]
+
+      - name: Upload report and logs
+        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
+        with:
+          name: rustfs-tier-test-${{ github.run_id }}-${{ github.run_attempt }}
+          path: ${{ env.TIER_ARTIFACTS_DIR }}/
+          if-no-files-found: error
+
+      - name: Cleanup environment (after)
+        if: always()
+        run: |
+          set -euo pipefail
+          sudo docker rm -f rustfs-test-mqtt >/dev/null 2>&1 || true
+          sudo rm -f /tmp/rustfs-mosquitto.conf
+          read -r -a NODES <<< "${RUSTFS_NODES:-vm000 vm001 vm002}"
+          SSH_USER="${RUSTFS_SSH_USER:-azureuser}"
+          for node in "${NODES[@]}"; do
+            ssh -o BatchMode=yes -o ConnectTimeout=10 -o StrictHostKeyChecking=accept-new "${SSH_USER}@${node}" '
+              set -euo pipefail
+              SUDO=""; [ "$(id -u)" -ne 0 ] && SUDO="sudo -n"
+              ${SUDO} systemctl stop rustfs 2>/dev/null || true
+              if ${SUDO} dpkg -l rustfs 2>/dev/null | grep -q "^ii"; then
+                ${SUDO} dpkg -P rustfs
+              fi
+              for i in 1 2 3 4; do ${SUDO} rm -rf /data/rustfs${i}/mnmd; done
+              ${SUDO} rm -rf /var/log/rustfs /var/lib/rustfs/kms
+            '
+          done
+
+      - name: Enforce tier suite result
+        id: gate
+        if: always()
+        env:
+          EVIDENCE_OUTCOME: ${{ steps.evidence.outcome }}
+          TEST_OUTCOME: ${{ steps.test.outcome }}
+          GATE_RC_FILE: ${{ env.TIER_ARTIFACTS_DIR }}/rustfs-tier-gate.rc
+        run: |
+          set -euo pipefail
+          failed=0
+          if [ "${EVIDENCE_OUTCOME}" != "success" ]; then
+            echo "tier evidence directory initialization is ${EVIDENCE_OUTCOME}, expected success" >&2
+            failed=1
+          fi
+          if [ "${TEST_OUTCOME}" != "success" ]; then
+            echo "tier suite step outcome is ${TEST_OUTCOME}, expected success" >&2
+            failed=1
+          fi
+          if [ "${EVIDENCE_OUTCOME}" != "success" ]; then
+            echo "structured gate result is unavailable because evidence initialization failed" >&2
+          elif [ ! -s "${GATE_RC_FILE}" ]; then
+            echo "structured gate result is missing" >&2
+            failed=1
+          else
+            GATE_RC="$(tr -d '[:space:]' < "${GATE_RC_FILE}")"
+            if ! [[ "${GATE_RC}" =~ ^[0-9]+$ ]] || [ "${GATE_RC}" -ne 0 ]; then
+              echo "structured 56-case gate failed with exit ${GATE_RC:-invalid}" >&2
+              failed=1
+            fi
+          fi
+          [ "${failed}" -eq 0 ]
+
       - name: File failure issue in rustfs/backlog
-        if: ${{ always() && (failure() || steps.test.outcome == 'failure' || steps.test.outcome == 'cancelled') }}
+        if: ${{ always() && (failure() || steps.test.outcome == 'failure' || steps.test.outcome == 'cancelled' || steps.evidence_verify.outcome == 'failure' || steps.evidence_verify.outcome == 'cancelled' || steps.gate.outcome == 'failure' || steps.gate.outcome == 'cancelled') }}
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
           SUITE: 'tier'
           SUITE_LABEL: 'Tier'
           RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
-          REPORT_FILE: '/tmp/rustfs-tier-report.md'
-          LOG_FILE: '/tmp/rustfs-tier.log'
+          EVIDENCE_DIR: ${{ env.TIER_ARTIFACTS_DIR }}
+          EVIDENCE_OUTCOME: ${{ steps.evidence.outcome }}
+          VERIFY_OUTCOME: ${{ steps.evidence_verify.outcome }}
+          GATE_OUTCOME: ${{ steps.gate.outcome }}
+          REPORT_FILE: ${{ env.TIER_ARTIFACTS_DIR }}/rustfs-tier-report.md
+          LOG_FILE: ${{ env.TIER_ARTIFACTS_DIR }}/rustfs-tier.log
         run: |
           set -euo pipefail
           if [ -z "${GH_TOKEN:-}" ]; then
@@ -293,10 +401,17 @@ jobs:
             echo "- Run: ${RUN_URL}"
             echo "- Trigger: ${GITHUB_EVENT_NAME}"
             echo "- Date: $(date -u +%Y-%m-%d)"
+            echo "- Evidence initialization: ${EVIDENCE_OUTCOME}"
+            echo "- Evidence verification: ${VERIFY_OUTCOME}"
+            echo "- Final gate: ${GATE_OUTCOME}"
             echo ""
             echo "## Report (errors and symptoms)"
             echo ""
-            if [ -s "${REPORT_FILE}" ]; then
+            if [ "${EVIDENCE_OUTCOME}" != "success" ]; then
+              echo "(the run evidence directory was rejected; its contents were not read)"
+            elif [ ! -d "${EVIDENCE_DIR}" ] || [ -L "${EVIDENCE_DIR}" ]; then
+              echo "(the run evidence directory is missing or unsafe; its contents were not read)"
+            elif [ -s "${REPORT_FILE}" ]; then
               redact < "${REPORT_FILE}"
             elif [ -s "${LOG_FILE:-}" ]; then
               echo "(report file missing; log tail below)"
@@ -313,81 +428,54 @@ jobs:
           fi
           echo "filed backlog issue for suite ${SUITE}"
 
-      - name: Upload report and logs
-        if: always()
-        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
-        with:
-          name: rustfs-tier-test-${{ github.run_id }}-${{ github.run_attempt }}
-          path: |
-            /tmp/rustfs-tier.log
-            /tmp/rustfs-tier-report.md
-            /tmp/rustfs-tier-cases.md
-            /tmp/rustfs-tier-gate.rc
-            ${{ env.TIER_ARTIFACTS_DIR }}/
-          if-no-files-found: error
-
-      - name: Cleanup environment (after)
-        if: always()
-        run: |
-          set -euo pipefail
-          sudo docker rm -f rustfs-test-mqtt >/dev/null 2>&1 || true
-          sudo rm -f /tmp/rustfs-mosquitto.conf
-          read -r -a NODES <<< "${RUSTFS_NODES:-vm000 vm001 vm002}"
-          SSH_USER="${RUSTFS_SSH_USER:-azureuser}"
-          for node in "${NODES[@]}"; do
-            ssh -o BatchMode=yes -o ConnectTimeout=10 -o StrictHostKeyChecking=accept-new "${SSH_USER}@${node}" '
-              set -euo pipefail
-              SUDO=""; [ "$(id -u)" -ne 0 ] && SUDO="sudo -n"
-              ${SUDO} systemctl stop rustfs 2>/dev/null || true
-              if ${SUDO} dpkg -l rustfs 2>/dev/null | grep -q "^ii"; then
-                ${SUDO} dpkg -P rustfs
-              fi
-              for i in 1 2 3 4; do ${SUDO} rm -rf /data/rustfs${i}/mnmd; done
-              ${SUDO} rm -rf /var/log/rustfs /var/lib/rustfs/kms
-            '
-          done
-
-      - name: Enforce tier suite result
-        if: always()
-        env:
-          TEST_OUTCOME: ${{ steps.test.outcome }}
-          GATE_RC_FILE: /tmp/rustfs-tier-gate.rc
-        run: |
-          set -euo pipefail
-          failed=0
-          if [ "${TEST_OUTCOME}" != "success" ]; then
-            echo "tier suite step outcome is ${TEST_OUTCOME}, expected success" >&2
-            failed=1
-          fi
-          if [ ! -s "${GATE_RC_FILE}" ]; then
-            echo "structured gate result is missing" >&2
-            failed=1
-          else
-            GATE_RC="$(tr -d '[:space:]' < "${GATE_RC_FILE}")"
-            if ! [[ "${GATE_RC}" =~ ^[0-9]+$ ]] || [ "${GATE_RC}" -ne 0 ]; then
-              echo "structured 56-case gate failed with exit ${GATE_RC:-invalid}" >&2
-              failed=1
-            fi
-          fi
-          [ "${failed}" -eq 0 ]
-
       - name: "Continue functional chain (next: Storage engine)"
         # Only chain-triggered runs forward to the next suite; standalone
-        # workflow_dispatch runs stop after their own cleanup.
+        # workflow_dispatch runs stop after their own cleanup. A failed
+        # handoff must never pass silently: it retries, then files an alert
+        # issue in rustfs/backlog so a stalled chain is visible.
         if: ${{ always() && github.event_name == 'repository_dispatch' }}
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
         run: |
-          set -euo pipefail
-          if [ -z "${GH_TOKEN:-}" ]; then
+          set -uo pipefail
+          if [ -z "${{GH_TOKEN:-}}" ]; then
             echo "PF_TESTING_GH_TOKEN is not configured; cannot dispatch the next suite" >&2
             exit 1
           fi
-          echo "Dispatching next functional suite: Storage engine"
-          gh api --method POST repos/rustfs/rustfs/dispatches \
-            -f event_type='rustfs-chain-storage' \
-            -F 'client_payload[from_suite]=tier'
+          DISPATCHED=0
+          for attempt in 1 2 3; do
+            if gh api --method POST repos/rustfs/rustfs/dispatches \
+              -f event_type='rustfs-chain-storage' \
+              -F 'client_payload[from_suite]=tier'; then
+              echo "dispatched next suite Storage engine (attempt ${{attempt}})"
+              DISPATCHED=1
+              break
+            fi
+            echo "dispatch attempt ${{attempt}} failed; retrying in ${{attempt}}0s" >&2
+            sleep "${{attempt}}0"
+          done
+          if [ "${{DISPATCHED:-0}}" -ne 1 ]; then
+            echo "ERROR: functional chain stalled: could not dispatch Storage engine after 3 attempts" >&2
+            TITLE="[functional][chain] stalled after tier (run ${{GITHUB_RUN_ID}})"
+            BODY_FILE="$(mktemp)"
+            {
+              echo "The functional chain could not hand off from **tier** to **Storage engine** after 3 attempts."
+              echo ""
+              echo "- Failed suite job: ${{GITHUB_SERVER_URL}}/${{GITHUB_REPOSITORY}}/actions/runs/${{GITHUB_RUN_ID}}"
+              echo "- Expected next event: `rustfs-chain-storage`"
+              echo "- Likely cause: PF_TESTING_GH_TOKEN lacks contents:write on rustfs/rustfs, or the GitHub API was unavailable."
+              echo "- Recovery: re-dispatch manually with"
+              echo "  ```"
+              echo "  gh api --method POST repos/rustfs/rustfs/dispatches -f event_type='rustfs-chain-storage'"
+              echo "  ```"
+            } > "${{BODY_FILE}}"
+            gh issue create -R rustfs/backlog --title "${{TITLE}}" \
+              --body-file "${{BODY_FILE}}" --label functional-test \
+              || gh issue create -R rustfs/backlog --title "${{TITLE}}" --body-file "${{BODY_FILE}}" \
+              || echo "could not file the stall alert issue either; check the token" >&2
+            exit 1
+          fi
 
       - name: Notify on failure
         if: failure()

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

@@ -388,21 +388,52 @@ jobs:
 
       - name: "Continue functional chain (next: S3 compatibility)"
         # Only chain-triggered runs forward to the next suite; standalone
-        # workflow_dispatch runs stop after their own cleanup.
+        # workflow_dispatch runs stop after their own cleanup. A failed
+        # handoff must never pass silently: it retries, then files an alert
+        # issue in rustfs/backlog so a stalled chain is visible.
         if: ${{ always() && github.event_name == 'repository_dispatch' }}
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
         run: |
-          set -euo pipefail
-          if [ -z "${GH_TOKEN:-}" ]; then
+          set -uo pipefail
+          if [ -z "${{GH_TOKEN:-}}" ]; then
             echo "PF_TESTING_GH_TOKEN is not configured; cannot dispatch the next suite" >&2
             exit 1
           fi
-          echo "Dispatching next functional suite: S3 compatibility"
-          gh api --method POST repos/rustfs/rustfs/dispatches \
-            -f event_type='rustfs-chain-s3' \
-            -F 'client_payload[from_suite]=upgrade'
+          DISPATCHED=0
+          for attempt in 1 2 3; do
+            if gh api --method POST repos/rustfs/rustfs/dispatches \
+              -f event_type='rustfs-chain-s3' \
+              -F 'client_payload[from_suite]=upgrade'; then
+              echo "dispatched next suite S3 compatibility (attempt ${{attempt}})"
+              DISPATCHED=1
+              break
+            fi
+            echo "dispatch attempt ${{attempt}} failed; retrying in ${{attempt}}0s" >&2
+            sleep "${{attempt}}0"
+          done
+          if [ "${{DISPATCHED:-0}}" -ne 1 ]; then
+            echo "ERROR: functional chain stalled: could not dispatch S3 compatibility after 3 attempts" >&2
+            TITLE="[functional][chain] stalled after upgrade (run ${{GITHUB_RUN_ID}})"
+            BODY_FILE="$(mktemp)"
+            {
+              echo "The functional chain could not hand off from **upgrade** to **S3 compatibility** after 3 attempts."
+              echo ""
+              echo "- Failed suite job: ${{GITHUB_SERVER_URL}}/${{GITHUB_REPOSITORY}}/actions/runs/${{GITHUB_RUN_ID}}"
+              echo "- Expected next event: `rustfs-chain-s3`"
+              echo "- Likely cause: PF_TESTING_GH_TOKEN lacks contents:write on rustfs/rustfs, or the GitHub API was unavailable."
+              echo "- Recovery: re-dispatch manually with"
+              echo "  ```"
+              echo "  gh api --method POST repos/rustfs/rustfs/dispatches -f event_type='rustfs-chain-s3'"
+              echo "  ```"
+            } > "${{BODY_FILE}}"
+            gh issue create -R rustfs/backlog --title "${{TITLE}}" \
+              --body-file "${{BODY_FILE}}" --label functional-test \
+              || gh issue create -R rustfs/backlog --title "${{TITLE}}" --body-file "${{BODY_FILE}}" \
+              || echo "could not file the stall alert issue either; check the token" >&2
+            exit 1
+          fi
 
       - name: Notify on failure
         if: failure()

crates/e2e_test/src/checksum_upload_test.rs

@@ -22,7 +22,7 @@ mod tests {
     use aws_sdk_s3::config::{Credentials, Region, RequestChecksumCalculation};
     use aws_sdk_s3::error::ProvideErrorMetadata;
     use aws_sdk_s3::primitives::ByteStream;
-    use aws_sdk_s3::types::{ChecksumAlgorithm, ChecksumMode, CompletedMultipartUpload, CompletedPart};
+    use aws_sdk_s3::types::{ChecksumAlgorithm, ChecksumMode, CompletedMultipartUpload, CompletedPart, ServerSideEncryption};
     use aws_smithy_http_client::Builder as SmithyHttpClientBuilder;
     use md5::{Digest as Md5Digest, Md5};
     use rustfs_rio::{Checksum, ChecksumType as RioChecksumType};
@@ -260,6 +260,117 @@ mod tests {
         info!("PASSED: HeadObject returns stored SHA256 digest");
     }
 
+    #[tokio::test]
+    async fn test_head_object_returns_sse_s3_checksum() {
+        init_logging();
+
+        let mut env = RustFSTestEnvironment::new().await.expect("Failed to create test environment");
+        env.start_rustfs_server_with_env(
+            vec![],
+            &[
+                ("RUSTFS_SSE_S3_MASTER_KEY", "MTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0NTY3ODkwMTI="),
+                ("RUSTFS_CONSOLE_ENABLE", "false"),
+            ],
+        )
+        .await
+        .expect("Failed to start RustFS");
+
+        let client = create_s3_client(&env);
+        let bucket = "test-sse-s3-checksum-head";
+        create_bucket(&client, bucket).await.expect("Failed to create bucket");
+
+        let put = client
+            .put_object()
+            .bucket(bucket)
+            .key("encrypted.txt")
+            .body(ByteStream::from_static(b"encrypted checksum"))
+            .server_side_encryption(ServerSideEncryption::Aes256)
+            .checksum_algorithm(ChecksumAlgorithm::Crc32)
+            .send()
+            .await
+            .expect("SSE-S3 PutObject with CRC32 failed");
+        let expected = put.checksum_crc32().expect("PutObject must return CRC32");
+
+        let head = client
+            .head_object()
+            .bucket(bucket)
+            .key("encrypted.txt")
+            .checksum_mode(ChecksumMode::Enabled)
+            .send()
+            .await
+            .expect("SSE-S3 HeadObject failed");
+
+        assert_eq!(head.checksum_crc32(), Some(expected));
+
+        client
+            .copy_object()
+            .bucket(bucket)
+            .key("encrypted-copy.txt")
+            .copy_source(format!("{bucket}/encrypted.txt"))
+            .server_side_encryption(ServerSideEncryption::Aes256)
+            .send()
+            .await
+            .expect("SSE-S3 CopyObject failed");
+        let copy_head = client
+            .head_object()
+            .bucket(bucket)
+            .key("encrypted-copy.txt")
+            .checksum_mode(ChecksumMode::Enabled)
+            .send()
+            .await
+            .expect("SSE-S3 copied HeadObject failed");
+
+        assert_eq!(copy_head.checksum_crc32(), Some(expected));
+
+        let multipart_key = "encrypted-multipart.txt";
+        let create = client
+            .create_multipart_upload()
+            .bucket(bucket)
+            .key(multipart_key)
+            .server_side_encryption(ServerSideEncryption::Aes256)
+            .checksum_algorithm(ChecksumAlgorithm::Crc32)
+            .send()
+            .await
+            .expect("SSE-S3 CreateMultipartUpload with CRC32 failed");
+        let upload_id = create.upload_id().expect("CreateMultipartUpload must return an upload ID");
+        let part = client
+            .upload_part()
+            .bucket(bucket)
+            .key(multipart_key)
+            .upload_id(upload_id)
+            .part_number(1)
+            .body(ByteStream::from_static(b"encrypted multipart checksum"))
+            .checksum_algorithm(ChecksumAlgorithm::Crc32)
+            .send()
+            .await
+            .expect("SSE-S3 UploadPart with CRC32 failed");
+        let completed_part = CompletedPart::builder()
+            .part_number(1)
+            .e_tag(part.e_tag().expect("UploadPart must return an ETag"))
+            .checksum_crc32(part.checksum_crc32().expect("UploadPart must return CRC32"))
+            .build();
+        let complete = client
+            .complete_multipart_upload()
+            .bucket(bucket)
+            .key(multipart_key)
+            .upload_id(upload_id)
+            .multipart_upload(CompletedMultipartUpload::builder().parts(completed_part).build())
+            .send()
+            .await
+            .expect("SSE-S3 CompleteMultipartUpload with CRC32 failed");
+        let expected_multipart = complete.checksum_crc32().expect("CompleteMultipartUpload must return CRC32");
+        let multipart_head = client
+            .head_object()
+            .bucket(bucket)
+            .key(multipart_key)
+            .checksum_mode(ChecksumMode::Enabled)
+            .send()
+            .await
+            .expect("SSE-S3 multipart HeadObject failed");
+
+        assert_eq!(multipart_head.checksum_crc32(), Some(expected_multipart));
+    }
+
     /// Multipart upload with checksum: CreateMultipartUpload, UploadPart(s) with checksum_sha256, CompleteMultipartUpload; then GetObject verifies content.
     /// Uses part size >= 5MB (server minimum) for two parts.
     #[tokio::test]

crates/e2e_test/src/common.rs

@@ -1699,6 +1699,51 @@ impl RustFSTestClusterEnvironment {
         process.wait()?;
         Ok(())
     }
+
+    /// Gracefully stop one cluster node and wait for its process to exit.
+    ///
+    /// This is intentionally separate from [`Self::stop_node`]: the latter is
+    /// a hard kill used by crash-recovery tests, while this path lets RustFS
+    /// complete its normal shutdown hooks before a test restarts the node.
+    pub async fn stop_node_gracefully(&mut self, node_idx: usize) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        self.ensure_node_index(node_idx)?;
+
+        #[cfg(unix)]
+        {
+            let Some(process) = self.nodes[node_idx].process.as_ref() else {
+                return Ok(());
+            };
+            let pid = process.id().to_string();
+            let signal_status = Command::new("kill").args(["-TERM", &pid]).status()?;
+            if !signal_status.success() {
+                return Err(format!("failed to send SIGTERM to cluster node {node_idx} (pid {pid})").into());
+            }
+
+            let mut process = self.nodes[node_idx]
+                .process
+                .take()
+                .ok_or_else(|| format!("cluster node {node_idx} process disappeared while stopping"))?;
+            let deadline = std::time::Instant::now() + Duration::from_secs(45);
+            loop {
+                if let Some(status) = process.try_wait()? {
+                    info!("Cluster node {} stopped gracefully with {}", node_idx, status);
+                    return Ok(());
+                }
+                if std::time::Instant::now() >= deadline {
+                    let _ = process.kill();
+                    let _ = process.wait();
+                    return Err(format!("cluster node {node_idx} did not stop gracefully within 45 seconds").into());
+                }
+                sleep(Duration::from_millis(100)).await;
+            }
+        }
+
+        #[cfg(not(unix))]
+        {
+            let _ = node_idx;
+            Err("graceful cluster-node stop is only supported on Unix E2E hosts".into())
+        }
+    }
 }
 
 impl Drop for RustFSTestClusterEnvironment {

crates/e2e_test/src/kms/multipart_encryption_test.rs

@@ -560,18 +560,12 @@ async fn test_multipart_encryption_type(
         .set_parts(Some(completed_parts))
         .build();
 
-    let mut complete_request = s3_client
+    let complete_request = s3_client
         .complete_multipart_upload()
         .bucket(bucket)
         .key(object_key)
         .upload_id(upload_id)
         .multipart_upload(completed_multipart_upload);
-    if matches!(encryption_type, EncryptionType::SSEC) {
-        complete_request = complete_request
-            .sse_customer_algorithm("AES256")
-            .sse_customer_key(sse_c_key.as_ref().unwrap())
-            .sse_customer_key_md5(sse_c_md5.as_ref().unwrap());
-    }
     let _complete_output = complete_request.send().await?;
 
     // Download and verify

crates/e2e_test/src/replacement_privileged_e2e_test.rs

@@ -40,10 +40,10 @@ mod tests {
 
     const ENABLE_ENV: &str = "RUSTFS_PRIVILEGED_REPLACEMENT_E2E";
     const NAMESPACE_ENV: &str = "RUSTFS_PRIVILEGED_REPLACEMENT_E2E_IN_NAMESPACE";
+    const LOG_DIR_ENV: &str = "RUSTFS_PRIVILEGED_REPLACEMENT_LOG_DIR";
     const TARGET_NODE: usize = 1;
     const TARGET_DRIVE: usize = 0;
     const MOUNT_SIZE: &str = "size=128m,mode=0700";
-    const ABSENT_SCANNER_OBSERVATION_TIMEOUT_SECS: u64 = 180;
     const REPLACEMENT_RECOVERY_DIR: &str = ".rustfs.sys/buckets/ahm-replacement";
     const REPLACEMENT_INTENT_SUFFIX: &str = "_ahm_replacement_intent.json";
     const REPLACEMENT_COMPLETION_PROOF_SUFFIX: &str = "_ahm_replacement_completion_proof.json";
@@ -142,6 +142,23 @@ mod tests {
             run_command("dmsetup", &["resume", &self.dm_name])
         }
 
+        fn verify_raw_io_is_unavailable(&self) -> Result<(), Box<dyn Error + Send + Sync>> {
+            let mapper = format!("/dev/mapper/{}", self.dm_name);
+            let output = Command::new("dd")
+                .env("LC_ALL", "C")
+                .arg(format!("if={mapper}"))
+                .args(["of=/dev/null", "bs=4096", "count=1", "iflag=direct", "status=none"])
+                .output()?;
+            if output.status.success() {
+                return Err(format!("dm-error target unexpectedly allowed a raw read from {mapper}").into());
+            }
+            let stderr = String::from_utf8_lossy(&output.stderr);
+            if !stderr.contains("Input/output error") {
+                return Err(format!("raw read from dm-error target failed unexpectedly: {stderr}").into());
+            }
+            Ok(())
+        }
+
         fn restore_available(&self) -> Result<(), Box<dyn Error + Send + Sync>> {
             let sectors = run_command_stdout("blockdev", &["--getsz", &self.loop_device])?;
             let linear_table = format!("0 {sectors} linear {} 0", self.loop_device);
@@ -194,6 +211,72 @@ mod tests {
         }
     }
 
+    struct ZramBlockMount {
+        target: PathBuf,
+        device: String,
+        mounted: bool,
+    }
+
+    impl ZramBlockMount {
+        fn reserve(target: &Path) -> Result<Self, Box<dyn Error + Send + Sync>> {
+            if !Path::new("/dev/zram-control").exists() {
+                run_command("modprobe", &["zram"])?;
+            }
+            let device = run_command_stdout("zramctl", &["--find", "--size", "256M"])?;
+            if device.is_empty() {
+                return Err("zramctl --find --size returned an empty device".into());
+            }
+
+            Ok(Self {
+                target: target.to_path_buf(),
+                device,
+                mounted: false,
+            })
+        }
+
+        fn mount_target(&mut self) -> Result<(), Box<dyn Error + Send + Sync>> {
+            let result = (|| {
+                run_command("mkfs.ext4", &["-F", &self.device])?;
+                let target_arg = path_to_string(&self.target, "zram replacement mount target")?;
+                run_command("mount", &[&self.device, &target_arg])
+            })();
+            if let Err(error) = result {
+                let _ = self.cleanup();
+                return Err(error);
+            }
+            self.mounted = true;
+            Ok(())
+        }
+
+        fn cleanup(&mut self) -> Result<(), Box<dyn Error + Send + Sync>> {
+            let mut first_error: Option<Box<dyn Error + Send + Sync>> = None;
+            if self.mounted {
+                if let Err(error) = detach_mount(&self.target) {
+                    first_error.get_or_insert(error);
+                } else {
+                    self.mounted = false;
+                }
+            }
+            if !self.device.is_empty() {
+                if let Err(error) = run_command("zramctl", &["--reset", &self.device]) {
+                    first_error.get_or_insert(error);
+                } else {
+                    self.device.clear();
+                }
+            }
+            if let Some(error) = first_error {
+                return Err(error);
+            }
+            Ok(())
+        }
+    }
+
+    impl Drop for ZramBlockMount {
+        fn drop(&mut self) {
+            let _ = self.cleanup();
+        }
+    }
+
     fn checked_command_output(program: &str, args: &[&str]) -> Result<std::process::Output, Box<dyn Error + Send + Sync>> {
         let output = Command::new(program).args(args).output()?;
         if output.status.success() {
@@ -298,6 +381,18 @@ mod tests {
         Err(format!("{ENABLE_ENV}=1 requires root or CAP_SYS_ADMIN; unshare exited with status {status}").into())
     }
 
+    fn replacement_node_log_path(
+        cluster_temp_dir: &str,
+        parity: usize,
+        node_index: usize,
+    ) -> Result<PathBuf, Box<dyn Error + Send + Sync>> {
+        let log_dir = std::env::var_os(LOG_DIR_ENV)
+            .map(PathBuf::from)
+            .unwrap_or_else(|| PathBuf::from(cluster_temp_dir));
+        fs::create_dir_all(&log_dir)?;
+        Ok(log_dir.join(format!("replacement-ec{parity}-node{node_index}-{}.log", std::process::id())))
+    }
+
     fn payload(len: usize, seed: u8) -> Vec<u8> {
         let mut next = seed;
         (0..len)
@@ -472,8 +567,20 @@ mod tests {
             if let Some(version_id) = &version.version_id {
                 request = request.version_id(version_id);
             }
-            let response = request.send().await?;
-            let body = response.body.collect().await?.into_bytes();
+            let response = request.send().await.map_err(|error| {
+                format!("body GET failed for {}/{}@{:?}: {error}", version.bucket, version.key, version.version_id)
+            })?;
+            let body = response
+                .body
+                .collect()
+                .await
+                .map_err(|error| {
+                    format!(
+                        "body stream failed for {}/{}@{:?}: {error}",
+                        version.bucket, version.key, version.version_id
+                    )
+                })?
+                .into_bytes();
             assert_eq!(
                 sha256_hex(&body),
                 *expected_sha256,
@@ -582,81 +689,6 @@ mod tests {
         Ok(())
     }
 
-    fn log_tail(log: &str) -> String {
-        let mut lines = log.lines().rev().take(80).collect::<Vec<_>>();
-        lines.reverse();
-        lines.join("\n")
-    }
-
-    fn log_len(path: &Path) -> Result<u64, Box<dyn Error + Send + Sync>> {
-        match fs::metadata(path) {
-            Ok(metadata) => Ok(metadata.len()),
-            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(0),
-            Err(error) => Err(format!("failed to stat target node log {path:?}: {error}").into()),
-        }
-    }
-
-    fn log_from_offset(path: &Path, offset: u64) -> Result<String, Box<dyn Error + Send + Sync>> {
-        let log = match fs::read(path) {
-            Ok(log) => log,
-            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Vec::new(),
-            Err(error) => return Err(format!("failed to read target node log {path:?}: {error}").into()),
-        };
-        let start = usize::try_from(offset).unwrap_or(usize::MAX).min(log.len());
-        Ok(String::from_utf8_lossy(&log[start..]).into_owned())
-    }
-
-    fn live_disk_loss_scan_completed(log: &str, target_disk: &Path) -> bool {
-        let target = target_disk.to_string_lossy();
-        let mut saw_live_loss = false;
-        for line in log.lines() {
-            if line.contains("Heal auto-scan disk inspection failed")
-                && line.contains("check_failed")
-                && line.contains(target.as_ref())
-            {
-                saw_live_loss = true;
-                continue;
-            }
-            if saw_live_loss && (line.contains("Heal auto disk scanner idle") || line.contains("Heal auto-scan cycle completed"))
-            {
-                return true;
-            }
-        }
-        false
-    }
-
-    fn live_disk_loss_scan_completed_from_path(
-        log_path: &Path,
-        start_offset: u64,
-        target_disk: &Path,
-    ) -> Result<bool, Box<dyn Error + Send + Sync>> {
-        Ok(live_disk_loss_scan_completed(&log_from_offset(log_path, start_offset)?, target_disk))
-    }
-
-    async fn wait_for_live_disk_loss_observation(
-        log_path: &Path,
-        target_disk: &Path,
-        start_offset: u64,
-        timeout_secs: u64,
-    ) -> Result<(), Box<dyn Error + Send + Sync>> {
-        let deadline = Instant::now() + Duration::from_secs(timeout_secs);
-        let mut tick = interval(Duration::from_secs(1));
-        loop {
-            if live_disk_loss_scan_completed_from_path(log_path, start_offset, target_disk)? {
-                return Ok(());
-            }
-            if Instant::now() >= deadline {
-                let log = log_from_offset(log_path, start_offset)?;
-                return Err(format!(
-                    "scanner did not finish a live target-loss scan for {target_disk:?} within {timeout_secs}s; log tail:\n{}",
-                    log_tail(&log)
-                )
-                .into());
-            }
-            tick.tick().await;
-        }
-    }
-
     fn cluster_status_is_definitive(status: &serde_json::Value) -> Result<bool, Box<dyn Error + Send + Sync>> {
         status["cluster"]["definitive"]
             .as_bool()
@@ -707,6 +739,13 @@ mod tests {
             .collect()
     }
 
+    fn is_transient_recovery_version_absence(error: &(dyn Error + 'static)) -> bool {
+        matches!(
+            error.downcast_ref::<rustfs_filemeta::Error>(),
+            Some(rustfs_filemeta::Error::FileVersionNotFound)
+        )
+    }
+
     fn incomplete_versions(
         target_disk: &Path,
         versions: &[BaselineVersion],
@@ -714,7 +753,21 @@ mod tests {
         let mut missing = BTreeSet::new();
         for version in versions {
             let actual =
-                census_object_version_on_disk(target_disk, &version.bucket, &version.key, version.version_id.as_deref())?;
+                match census_object_version_on_disk(target_disk, &version.bucket, &version.key, version.version_id.as_deref()) {
+                    Ok(actual) => actual,
+                    // During replacement recovery, xl.meta may arrive before this
+                    // particular historical version. The generic census helper
+                    // correctly reports that as an error; this progress poll must
+                    // instead wait for the version to be restored.
+                    Err(error) if is_transient_recovery_version_absence(error.as_ref()) => {
+                        missing.insert(format!(
+                            "{}/{}@{:?}: version metadata not yet present on replacement",
+                            version.bucket, version.key, version.version_id
+                        ));
+                        continue;
+                    }
+                    Err(error) => return Err(error),
+                };
             if !actual.matches_manifest(&version.expected) {
                 missing.insert(format!("{}/{}@{:?}: {actual:?}", version.bucket, version.key, version.version_id));
             }
@@ -822,13 +875,15 @@ mod tests {
 
         let mut mount_ns = MountNamespaceGuard::new()?;
         let mut cluster = RustFSTestClusterEnvironment::with_topology(ClusterTopology::single_pool_multidrive(3, 4)).await?;
-        let target_log_path = PathBuf::from(&cluster.temp_dir).join(format!("replacement-node{TARGET_NODE}.log"));
-        cluster.set_node_capture_log_path(TARGET_NODE, target_log_path.to_string_lossy())?;
+        for node_index in 0..cluster.nodes.len() {
+            let node_log_path = replacement_node_log_path(&cluster.temp_dir, parity, node_index)?;
+            cluster.set_node_capture_log_path(node_index, node_log_path.to_string_lossy())?;
+        }
         let target_disk = PathBuf::from(&cluster.nodes[TARGET_NODE].data_dirs[TARGET_DRIVE]);
-        // Each drive below is an independent tmpfs mount, so this privileged
-        // path must exercise the production distinct-device/readiness fences.
+        // The blank target uses a temporary zram block device, so the
+        // replacement readiness fence sees no root or sibling alias.
         cluster.extra_env.retain(|(key, _)| key != "RUSTFS_UNSAFE_BYPASS_DISK_CHECK");
-        let image_root = PathBuf::from(&cluster.temp_dir).join("replacement-faultable-images");
+        let image_root = PathBuf::from(&cluster.temp_dir).join("replacement-block-images");
         let mut target_mount = None;
         for (node_index, node) in cluster.nodes.iter().enumerate() {
             for (drive_index, drive) in node.data_dirs.iter().enumerate() {
@@ -845,6 +900,7 @@ mod tests {
             }
         }
         let mut target_mount = target_mount.ok_or("target drive was not mounted with the faultable block fixture")?;
+        let mut replacement_mount = ZramBlockMount::reserve(&target_disk)?;
 
         cluster.set_env("RUSTFS_HEAL_ENABLED", "true");
         cluster.set_env("RUSTFS_SCANNER_ENABLED", "true");
@@ -852,28 +908,34 @@ mod tests {
         cluster.set_env("RUSTFS_SCANNER_CYCLE", "1");
         cluster.set_env("RUSTFS_SCANNER_START_DELAY_SECS", "0");
         cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", format!("EC:{parity}"));
-        cluster.set_node_env(TARGET_NODE, "RUST_LOG", "rustfs=info,rustfs::heal::manager=debug,rustfs_notify=debug")?;
+        for node_index in 0..cluster.nodes.len() {
+            cluster.set_node_env(node_index, "RUST_LOG", "rustfs=info,rustfs::heal::manager=debug,rustfs_notify=debug")?;
+        }
         cluster.start().await?;
 
         let clients = cluster.create_all_clients()?;
-        let versions = seed_baseline(&clients[0], &target_disk).await?;
-        verify_bodies(&clients[0], &versions).await?;
+        let versions = seed_baseline(&clients[0], &target_disk)
+            .await
+            .map_err(|error| format!("pre-fault baseline seeding failed: {error}"))?;
+        verify_bodies(&clients[0], &versions)
+            .await
+            .map_err(|error| format!("pre-fault body verification failed: {error}"))?;
 
-        let live_loss_log_offset = log_len(&target_log_path)?;
-        target_mount.make_unavailable()?;
-        wait_for_live_disk_loss_observation(
-            &target_log_path,
-            &target_disk,
-            live_loss_log_offset,
-            ABSENT_SCANNER_OBSERVATION_TIMEOUT_SECS,
-        )
-        .await?;
-        assert_no_replacement_status_records(&cluster, &target_disk).await?;
-        assert_no_replacement_admission_artifacts(&cluster, &target_disk)?;
+        target_mount
+            .make_unavailable()
+            .map_err(|error| format!("failed to install the dm-error target: {error}"))?;
+        target_mount
+            .verify_raw_io_is_unavailable()
+            .map_err(|error| format!("dm-error target was not proven by a direct raw read: {error}"))?;
+        assert_no_replacement_status_records(&cluster, &target_disk)
+            .await
+            .map_err(|error| format!("live-fault replacement status check failed: {error}"))?;
+        assert_no_replacement_admission_artifacts(&cluster, &target_disk)
+            .map_err(|error| format!("live-fault replacement artifact check failed: {error}"))?;
 
-        cluster.stop_node(TARGET_NODE)?;
+        cluster.stop_node_gracefully(TARGET_NODE).await?;
         target_mount.cleanup()?;
-        mount_ns.mount_tmpfs(&target_disk, &format!("rustfs-e2e-p{parity}-replacement"))?;
+        replacement_mount.mount_target()?;
         let missing_before_restart = incomplete_versions(&target_disk, &versions)?;
         assert_eq!(
             missing_before_restart.len(),
@@ -882,46 +944,26 @@ mod tests {
         );
         cluster.start_node(TARGET_NODE).await?;
 
-        wait_for_completed_replacement_with_census(&cluster, &target_disk, &versions, 420).await?;
-        verify_bodies(&clients[0], &versions).await?;
+        let recovery_result = async {
+            wait_for_completed_replacement_with_census(&cluster, &target_disk, &versions, 420).await?;
+            verify_bodies(&clients[0], &versions).await
+        }
+        .await;
+        let stop_result = cluster.stop_node_gracefully(TARGET_NODE).await;
+        let replacement_cleanup_result = replacement_mount.cleanup();
 
-        Ok(())
-    }
+        if let Err(error) = recovery_result {
+            if let Err(stop_error) = stop_result {
+                info!(%stop_error, "replacement target stop failed while preserving recovery failure");
+            }
+            if let Err(cleanup_error) = replacement_cleanup_result {
+                info!(%cleanup_error, "replacement zram cleanup failed while preserving recovery failure");
+            }
+            return Err(error);
+        }
+        stop_result?;
+        replacement_cleanup_result?;
 
-    #[test]
-    fn live_loss_barrier_requires_scanner_failure_after_log_offset() -> Result<(), Box<dyn Error + Send + Sync>> {
-        let target = Path::new("/mnt/target");
-        assert!(live_disk_loss_scan_completed(
-            "Heal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed\nHeal auto-scan cycle completed",
-            target
-        ));
-        assert!(live_disk_loss_scan_completed(
-            "Heal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed\nHeal auto disk scanner idle",
-            target
-        ));
-        assert!(!live_disk_loss_scan_completed(
-            "Heal auto disk scanner idle\nHeal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed",
-            target
-        ));
-        assert!(!live_disk_loss_scan_completed(
-            "event=disk_health_check_failed endpoint=/mnt/target disk_state=check_failed\nHeal auto disk scanner idle",
-            target
-        ));
-        assert!(!live_disk_loss_scan_completed(
-            "Heal auto-scan disk inspection failed endpoint=/mnt/other disk_state=check_failed\nHeal auto disk scanner idle",
-            target
-        ));
-        let path = std::env::temp_dir().join(format!("rustfs-replacement-scan-{}.log", std::process::id()));
-        let stale =
-            "Heal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed\nHeal auto disk scanner idle\n";
-        fs::write(&path, stale)?;
-        let offset = log_len(&path)?;
-        assert!(!live_disk_loss_scan_completed_from_path(&path, offset, target)?);
-        let fresh =
-            "Heal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed\nHeal auto disk scanner idle\n";
-        fs::write(&path, format!("{stale}{fresh}"))?;
-        assert!(live_disk_loss_scan_completed_from_path(&path, offset, target)?);
-        fs::remove_file(path)?;
         Ok(())
     }
 
@@ -954,6 +996,15 @@ mod tests {
         );
     }
 
+    #[test]
+    fn recovery_census_only_treats_missing_version_as_transient() {
+        let missing_version: Box<dyn Error + Send + Sync> = Box::new(rustfs_filemeta::Error::FileVersionNotFound);
+        let missing_file: Box<dyn Error + Send + Sync> = Box::new(rustfs_filemeta::Error::FileNotFound);
+
+        assert!(is_transient_recovery_version_absence(missing_version.as_ref()));
+        assert!(!is_transient_recovery_version_absence(missing_file.as_ref()));
+    }
+
     #[tokio::test]
     async fn completion_poll_samples_census_before_status() {
         let order = std::rc::Rc::new(std::cell::RefCell::new(Vec::new()));

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -485,6 +485,19 @@ impl BucketTargetSys {
         mutex
     }
 
+    /// Snapshot the heartbeat-tracked health of `url`'s endpoint.
+    ///
+    /// Returns `None` when the heartbeat has never seen the endpoint. Unlike
+    /// [`Self::is_offline`] this deliberately does not call `init_hc`: a caller
+    /// that only reports metrics must not create health entries as a side
+    /// effect, or merely rendering a status page would mark an unknown peer
+    /// online.
+    pub async fn endpoint_health(&self, url: &Url) -> Option<EpHealth> {
+        let key = endpoint_health_key(url);
+        let health_map = self.h_mutex.read().await;
+        health_map.get(&key).cloned()
+    }
+
     pub async fn is_offline(&self, url: &Url) -> bool {
         let key = endpoint_health_key(url);
         {

crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs

@@ -22,6 +22,7 @@ pub(crate) use rustfs_replication::{
     delete_marker_purge_version_id, delete_replication_creates_marker, delete_replication_missing_source_decision,
     delete_replication_object_opts, heal_uses_delete_replication_path, is_object_lock_denied_delete,
     is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome, replication_etags_match,
-    replication_multipart_complete_actual_size, replication_multipart_part_plan, resync_existing_delete_replication_info,
-    resync_target_for_object, should_retry_delete_marker_purge, single_part_replica_etag_mismatch, target_delete_version_id,
+    replication_multipart_complete_actual_size, replication_multipart_part_plan, replication_single_put_size_error,
+    resync_existing_delete_replication_info, resync_target_for_object, should_retry_delete_marker_purge,
+    single_part_replica_etag_mismatch, target_delete_version_id,
 };

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -32,8 +32,9 @@ use super::replication_object_decision_boundary::{
     MustReplicateOptions, ReplicationMultipartPartInput, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
     delete_replication_creates_marker, heal_uses_delete_replication_path, is_object_lock_denied_delete,
     is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome, replication_etags_match,
-    replication_multipart_complete_actual_size, replication_multipart_part_plan, resync_existing_delete_replication_info,
-    should_retry_delete_marker_purge, single_part_replica_etag_mismatch, target_delete_version_id,
+    replication_multipart_complete_actual_size, replication_multipart_part_plan, replication_single_put_size_error,
+    resync_existing_delete_replication_info, should_retry_delete_marker_purge, single_part_replica_etag_mismatch,
+    target_delete_version_id,
 };
 use super::replication_queue_boundary::{DeletedObjectReplicationInfo, ReplicationQueueAdmission};
 use super::replication_resync_boundary::ResyncStatusType;
@@ -88,6 +89,7 @@ use std::collections::{HashMap, HashSet};
 use std::fmt::Display;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, LazyLock, Mutex as StdMutex};
+use std::time::Instant;
 use time::OffsetDateTime;
 use time::format_description::well_known::Rfc3339;
 use tokio::io::AsyncRead;
@@ -118,6 +120,7 @@ const EVENT_DELETE_MARKER_PURGE_FAILED: &str = "replication_delete_marker_purge_
 const EVENT_DELETE_MARKER_PURGE_MRF: &str = "replication_delete_marker_purge_mrf";
 const METRIC_DELETE_MARKER_PURGE_TOTAL: &str = "rustfs_replication_delete_marker_purge_total";
 const EVENT_REPLICATION_VERSION_IDENTITY_DRIFT: &str = "replication_version_identity_drift";
+const EVENT_REPLICATION_OBJECT_FAILED: &str = "replication_object_failed";
 const EVENT_REPLICATION_PURGE_OBJECT_LOCK_DENIED: &str = "replication_purge_object_lock_denied";
 
 #[allow(
@@ -190,11 +193,19 @@ fn metadata_requires_existing_target(op_type: ReplicationType, object_info: &Obj
 
 const METRIC_VERSION_IDENTITY_DRIFT_TOTAL: &str = "rustfs_replication_version_identity_drift_total";
 
-/// Targets that already produced a version-identity-drift warning this
-/// process lifetime, by ARN. Deduping is advisory only (the metric still
-/// counts every drifting PUT), so a reconfigured target re-warning only
-/// after a restart is acceptable.
-static VERSION_IDENTITY_WARNED_ARNS: LazyLock<StdMutex<HashSet<String>>> = LazyLock::new(|| StdMutex::new(HashSet::new()));
+/// How long a target stays quiet after reporting version-identity drift.
+///
+/// This used to be a plain "once per ARN per process": one line ever, which on
+/// a long-lived server meant the single most important diagnostic for a
+/// non-converging generic S3 target scrolled away hours before anyone looked
+/// (rustfs#6822). Re-arming on an interval keeps the log bounded while leaving
+/// the condition discoverable in any recent window.
+const VERSION_IDENTITY_DRIFT_LOG_INTERVAL: TokioDuration = TokioDuration::from_secs(600);
+
+/// When each target last reported version-identity drift, by ARN. Throttling is
+/// advisory only — the metric still counts every drifting PUT.
+static VERSION_IDENTITY_WARNED_ARNS: LazyLock<StdMutex<HashMap<String, Instant>>> =
+    LazyLock::new(|| StdMutex::new(HashMap::new()));
 
 /// Version purges the peer denied under object lock (#6850). A RustFS peer
 /// with the replicated-purge GOVERNANCE exemption
@@ -322,20 +333,39 @@ fn audit_target_version_identity(tgt_client: &TargetClient, source_version_id: &
         return;
     }
     counter!(METRIC_VERSION_IDENTITY_DRIFT_TOTAL).increment(1);
+    if !version_identity_drift_log_due(&tgt_client.arn, Instant::now()) {
+        return;
+    }
+    // `error`, not `warn`: the target silently refuses the addressing scheme
+    // every version-addressed delete and heal on it depends on, so replication
+    // to it can never converge. At `warn` this sat below `DEFAULT_LOG_LEVEL`
+    // and no default deployment ever saw the one line that explains why a
+    // purged version is still on the target (rustfs#6822).
+    error!(
+        event = EVENT_REPLICATION_VERSION_IDENTITY_DRIFT,
+        component = LOG_COMPONENT_ECSTORE,
+        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+        arn = %tgt_client.arn,
+        endpoint = %tgt_client.endpoint,
+        sent_version_id = %source_version_id,
+        assigned_version_id = assigned_version_id.unwrap_or("<none>"),
+        "Replication target does not adopt source version ids; version-addressed replication cannot converge (run ?replication-check for details)"
+    );
+}
+
+/// Whether this ARN's version-identity drift is due to be logged again at
+/// `now`, re-arming the throttle when it is. Split out from the audit so the
+/// interval policy is testable without a target client.
+fn version_identity_drift_log_due(arn: &str, now: Instant) -> bool {
     let mut warned = VERSION_IDENTITY_WARNED_ARNS
         .lock()
         .unwrap_or_else(|poisoned| poisoned.into_inner());
-    if warned.insert(tgt_client.arn.clone()) {
-        warn!(
-            event = EVENT_REPLICATION_VERSION_IDENTITY_DRIFT,
-            component = LOG_COMPONENT_ECSTORE,
-            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-            arn = %tgt_client.arn,
-            endpoint = %tgt_client.endpoint,
-            sent_version_id = %source_version_id,
-            assigned_version_id = assigned_version_id.unwrap_or("<none>"),
-            "Replication target does not adopt source version ids; version-addressed replication cannot converge (run ?replication-check for details)"
-        );
+    match warned.get(arn) {
+        Some(last) if now.duration_since(*last) < VERSION_IDENTITY_DRIFT_LOG_INTERVAL => false,
+        _ => {
+            warned.insert(arn.to_string(), now);
+            true
+        }
     }
 }
 
@@ -2050,10 +2080,13 @@ pub(crate) async fn replicate_delete_with_outcome<S: ReplicationStorage>(
         }
     }
 
+    let delete_version_id = dobj.delete_object.version_id.map(|v| v.to_string());
+    note_replication_terminal_failure(&bucket, &dobj.delete_object.object_name, delete_version_id.as_deref(), &rinfos);
+
     let mut drs = get_replication_state(
         &rinfos,
         &dobj.delete_object.replication_state.clone().unwrap_or_default(),
-        dobj.delete_object.version_id.map(|v| v.to_string()),
+        delete_version_id,
     );
     if replication_status != prev_status {
         drs.replication_timestamp = Some(OffsetDateTime::now_utc());
@@ -3023,8 +3056,11 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
         }
     }
 
+    let version_id = roi.version_id.map(|v| v.to_string());
+    note_replication_terminal_failure(&bucket, &object, version_id.as_deref(), &rinfos);
+
     let previous_state = roi.replication_state.clone().unwrap_or_default();
-    let merged_state = get_replication_state(&rinfos, &previous_state, roi.version_id.map(|v| v.to_string()));
+    let merged_state = get_replication_state(&rinfos, &previous_state, version_id);
     let replication_status = merged_state.composite_replication_status();
     let new_replication_internal = merged_state.replication_status_internal.clone();
     let mut object_info = roi.to_object_info();
@@ -3101,6 +3137,61 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
     (merged_state, state_persisted)
 }
 
+/// Emit the operator-visible record of a replication attempt that ended FAILED.
+///
+/// Every per-branch failure log in this module is deliberately quieter than
+/// `error`: most of them sit on the replication hot path and fire once per
+/// object *per ARN*, so a target that stays unreachable would flood the log
+/// from inside the transfer loop. That left a hole customers fell into
+/// (rustfs#6825): `DEFAULT_LOG_LEVEL` is `error`, so on a stock deployment a
+/// failed object produced no line at all, and an operator staring at a replica
+/// that never arrived had nothing to correlate — the same trap already
+/// documented for the GET path in
+/// `crates/e2e_test/src/get_stream_failure_observability_test.rs`.
+///
+/// This is the one place that knows an object reached a *terminal* FAILED state
+/// for a target, so this is where the guaranteed-visible line belongs. It is
+/// bounded by the number of objects that actually fail rather than by attempts
+/// inside a transfer, and it carries the target's own error so a remote
+/// rejection is diagnosable without the operator first having to lower the
+/// global log level and reproduce.
+fn note_replication_terminal_failure(bucket: &str, object: &str, version_id: Option<&str>, rinfos: &ReplicatedInfos) {
+    for target in rinfos.targets.iter() {
+        if target.is_empty() {
+            continue;
+        }
+        let replication_failed = target.replication_status == ReplicationStatusType::Failed;
+        let purge_failed = target.version_purge_status == VersionPurgeStatusType::Failed;
+        if !replication_failed && !purge_failed {
+            continue;
+        }
+
+        error!(
+            event = EVENT_REPLICATION_OBJECT_FAILED,
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+            bucket = %bucket,
+            object = %object,
+            version_id = version_id.unwrap_or("-"),
+            arn = %target.arn,
+            endpoint = %target.endpoint,
+            op_type = %target.op_type,
+            size = target.size,
+            replication_status = %target.replication_status.as_str(),
+            version_purge_status = %target.version_purge_status.as_str(),
+            // The target's error can carry a signed URL or an echoed auth
+            // header, so it goes through the same redaction as the persisted
+            // resync detail rather than straight into the log.
+            error = %target
+                .error
+                .as_deref()
+                .and_then(sanitize_resync_error_detail)
+                .unwrap_or_else(|| "<none>".to_string()),
+            "Replication failed for object"
+        );
+    }
+}
+
 fn unavailable_object_target_info(roi: &ReplicateObjectInfo, arn: &str) -> ReplicatedTargetInfo {
     ReplicatedTargetInfo {
         arn: arn.to_string(),
@@ -3400,6 +3491,33 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
             }
         };
 
+        if let Some(reason) = replication_single_put_size_error(is_multipart, transfer_size) {
+            drop(gr);
+            rinfo.replication_status = ReplicationStatusType::Failed;
+            rinfo.error = Some(reason.clone());
+            warn!(
+                event = EVENT_RESYNC_TARGET_OPERATION_FAILED,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                bucket = %bucket,
+                target_bucket = %tgt_client.bucket,
+                arn = %tgt_client.arn,
+                object = %object,
+                operation = "put_object",
+                transfer_size = transfer_size,
+                error = %reason,
+                "Replication target operation failed"
+            );
+            send_local_event(EventArgs {
+                event_name: EventName::ObjectReplicationNotTracked.to_string(),
+                bucket_name: bucket.clone(),
+                object: object_info,
+                user_agent: "Internal: [Replication]".to_string(),
+                ..Default::default()
+            });
+            return rinfo;
+        }
+
         if let Some(err) = if is_multipart {
             drop(gr);
             let result = replicate_object_with_multipart(MultipartReplicationContext {
@@ -4068,6 +4186,14 @@ async fn replicate_all_payload_to_target<S: ReplicationObjectIO>(
     ctx: ReplicateAllPayloadContext<'_, S>,
     mut gr: GetObjectReader,
 ) -> Option<std::io::Error> {
+    // Fail before streaming a body the target is required to reject: an S3
+    // PutObject caps at 5 GiB, and this route is chosen by the source object's
+    // storage shape rather than its size (rustfs#6825).
+    if let Some(reason) = replication_single_put_size_error(ctx.is_multipart, ctx.transfer_size) {
+        drop(gr);
+        return Some(std::io::Error::other(reason));
+    }
+
     if ctx.is_multipart {
         drop(gr);
         let result = replicate_object_with_multipart(MultipartReplicationContext {
@@ -5716,4 +5842,207 @@ mod tests {
         assert!(!retry_scheduled.load(Ordering::SeqCst));
         assert_eq!(result.unwrap_err().to_string(), "transfer failed");
     }
+
+    /// A replication target's terminal outcome, as the operator sees it.
+    fn failed_target(arn: &str, error: &str) -> ReplicatedTargetInfo {
+        ReplicatedTargetInfo {
+            arn: arn.to_string(),
+            size: 6 * 1024 * 1024 * 1024,
+            op_type: ReplicationType::Object,
+            replication_status: ReplicationStatusType::Failed,
+            endpoint: "s3.wasabisys.com".to_string(),
+            error: Some(error.to_string()),
+            ..Default::default()
+        }
+    }
+
+    /// Capture the log this module writes, filtered exactly the way a stock
+    /// deployment filters it.
+    fn logs_at_default_level(emit: impl FnOnce()) -> String {
+        use std::sync::{Arc, Mutex};
+        use tracing_subscriber::EnvFilter;
+        use tracing_subscriber::fmt::MakeWriter;
+        use tracing_subscriber::layer::SubscriberExt;
+
+        #[derive(Clone, Default)]
+        struct CapturedLogs {
+            buffer: Arc<Mutex<Vec<u8>>>,
+        }
+        struct CapturedLogWriter {
+            buffer: Arc<Mutex<Vec<u8>>>,
+        }
+        impl std::io::Write for CapturedLogWriter {
+            fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+                self.buffer
+                    .lock()
+                    .expect("captured logs mutex should not be poisoned")
+                    .extend_from_slice(buf);
+                Ok(buf.len())
+            }
+            fn flush(&mut self) -> std::io::Result<()> {
+                Ok(())
+            }
+        }
+        impl<'a> MakeWriter<'a> for CapturedLogs {
+            type Writer = CapturedLogWriter;
+            fn make_writer(&'a self) -> Self::Writer {
+                CapturedLogWriter {
+                    buffer: Arc::clone(&self.buffer),
+                }
+            }
+        }
+
+        let logs = CapturedLogs::default();
+        let subscriber = tracing_subscriber::registry()
+            // Not a hand-picked level: this is the filter an operator who has
+            // changed nothing is actually running.
+            .with(EnvFilter::new(rustfs_config::DEFAULT_LOG_LEVEL))
+            .with(
+                tracing_subscriber::fmt::layer()
+                    .with_writer(logs.clone())
+                    .with_ansi(false)
+                    .without_time(),
+            );
+        let _guard = tracing::subscriber::set_default(subscriber);
+        let _callsite_pin = crate::test_tracing::pin_callsite_interest_for_test();
+
+        emit();
+
+        let buffer = logs
+            .buffer
+            .lock()
+            .expect("captured logs mutex should not be poisoned")
+            .clone();
+        String::from_utf8(buffer).expect("captured logs should be valid UTF-8")
+    }
+
+    /// rustfs#6825: a 6 GiB object never reached the target and the server said
+    /// nothing an operator could act on, because every failure line in this
+    /// module sat below `DEFAULT_LOG_LEVEL`. The object key, the target, and
+    /// the target's own error have to survive the default filter.
+    #[test]
+    fn failed_replication_names_the_object_at_the_default_log_level() {
+        let rinfos = ReplicatedInfos {
+            replication_timestamp: Some(OffsetDateTime::now_utc()),
+            targets: vec![failed_target("arn:replication::wasabi", "put_object failed: EntityTooLarge")],
+        };
+
+        let logs = logs_at_default_level(|| {
+            note_replication_terminal_failure("photos", "backups/vm-image.qcow2", Some("v-9"), &rinfos);
+        });
+
+        assert!(logs.contains("backups/vm-image.qcow2"), "the failed object must be named: {logs}");
+        assert!(logs.contains("arn:replication::wasabi"), "the target must be named: {logs}");
+        assert!(logs.contains("EntityTooLarge"), "the target's own error must survive: {logs}");
+        assert!(logs.contains("v-9"), "the version must be named: {logs}");
+        assert!(logs.contains(EVENT_REPLICATION_OBJECT_FAILED), "the event must be structured: {logs}");
+    }
+
+    #[test]
+    fn successful_replication_stays_quiet_at_the_default_log_level() {
+        let rinfos = ReplicatedInfos {
+            replication_timestamp: Some(OffsetDateTime::now_utc()),
+            targets: vec![ReplicatedTargetInfo {
+                arn: "arn:replication::wasabi".to_string(),
+                replication_status: ReplicationStatusType::Completed,
+                ..Default::default()
+            }],
+        };
+
+        let logs = logs_at_default_level(|| {
+            note_replication_terminal_failure("photos", "backups/ok.bin", None, &rinfos);
+        });
+
+        assert!(logs.is_empty(), "a completed replication must not log an error: {logs}");
+    }
+
+    /// A failed version purge is the 6822 symptom (the version stays on the
+    /// target); it must be as visible as a failed transfer even though the
+    /// replication status itself is not FAILED.
+    #[test]
+    fn failed_version_purge_is_reported_at_the_default_log_level() {
+        let rinfos = ReplicatedInfos {
+            replication_timestamp: Some(OffsetDateTime::now_utc()),
+            targets: vec![ReplicatedTargetInfo {
+                arn: "arn:replication::wasabi".to_string(),
+                op_type: ReplicationType::Delete,
+                replication_status: ReplicationStatusType::Empty,
+                version_purge_status: VersionPurgeStatusType::Failed,
+                error: Some("remove_object failed: NoSuchVersion".to_string()),
+                ..Default::default()
+            }],
+        };
+
+        let logs = logs_at_default_level(|| {
+            note_replication_terminal_failure("photos", "backups/purged.bin", Some("v-1"), &rinfos);
+        });
+
+        assert!(logs.contains("backups/purged.bin"), "the purged object must be named: {logs}");
+        assert!(logs.contains("NoSuchVersion"), "the target's own error must survive: {logs}");
+    }
+
+    /// The target's error is echoed remote text and can carry a signed URL or
+    /// an auth header, so it goes through the persisted-detail redaction rather
+    /// than straight into the log.
+    #[test]
+    fn failed_replication_redacts_a_sensitive_target_error() {
+        let rinfos = ReplicatedInfos {
+            replication_timestamp: Some(OffsetDateTime::now_utc()),
+            targets: vec![failed_target(
+                "arn:replication::wasabi",
+                "put_object failed: rejected Authorization: Bearer super-secret",
+            )],
+        };
+
+        let logs = logs_at_default_level(|| {
+            note_replication_terminal_failure("photos", "backups/vm-image.qcow2", None, &rinfos);
+        });
+
+        assert!(logs.contains("backups/vm-image.qcow2"), "the object must still be named: {logs}");
+        assert!(!logs.contains("super-secret"), "the credential must not reach the log: {logs}");
+    }
+
+    /// An empty target slot carries no outcome; reporting it would invent a
+    /// failure for a target that was never attempted.
+    #[test]
+    fn empty_target_slots_are_not_reported_as_failures() {
+        let rinfos = ReplicatedInfos {
+            replication_timestamp: Some(OffsetDateTime::now_utc()),
+            targets: vec![ReplicatedTargetInfo::default()],
+        };
+
+        let logs = logs_at_default_level(|| {
+            note_replication_terminal_failure("photos", "backups/unattempted.bin", None, &rinfos);
+        });
+
+        assert!(logs.is_empty(), "an empty target slot must not be reported: {logs}");
+    }
+
+    #[test]
+    fn version_identity_drift_re_arms_after_the_throttle_interval() {
+        let arn = "arn:replication::drift-throttle-test";
+        let start = Instant::now();
+
+        assert!(version_identity_drift_log_due(arn, start), "first drift must be reported");
+        assert!(
+            !version_identity_drift_log_due(arn, start + VERSION_IDENTITY_DRIFT_LOG_INTERVAL / 2),
+            "a second drift inside the interval must stay throttled"
+        );
+        assert!(
+            version_identity_drift_log_due(arn, start + VERSION_IDENTITY_DRIFT_LOG_INTERVAL),
+            "drift must become visible again once the interval elapses, instead of \
+             going silent for the rest of the process lifetime"
+        );
+    }
+
+    #[test]
+    fn version_identity_drift_throttles_each_target_independently() {
+        let now = Instant::now();
+
+        assert!(version_identity_drift_log_due("arn:replication::drift-a", now));
+        assert!(
+            version_identity_drift_log_due("arn:replication::drift-b", now),
+            "one target's report must not silence another's"
+        );
+    }
 }

crates/ecstore/src/core/pools.rs

@@ -49,7 +49,7 @@ use crate::error::{
     is_err_version_not_found,
 };
 use crate::layout::endpoints::EndpointServerPools;
-use crate::object_api::{DecommissionCapacityOptions, GetObjectReader, ObjectOptions};
+use crate::object_api::{DecommissionCapacityOptions, GetObjectReader, ObjectInfo, ObjectOptions};
 use crate::runtime::sources as runtime_sources;
 use crate::services::notification_sys::{
     acquire_tier_delete_journal_fleet_proof, tier_delete_journal_fleet_proof_matches, tier_delete_journal_topology_generation,
@@ -78,7 +78,9 @@ use rmp_serde::Serializer;
 use rustfs_filemeta::{FileInfoVersions, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
 use rustfs_heal_contracts::heal_channel::HealOpts;
 use rustfs_utils::crypto::{hex_sha256, is_sha256_checksum};
-use rustfs_utils::path::{encode_dir_object, path_join, path_to_bucket_object, path_to_bucket_object_with_base_path};
+use rustfs_utils::path::{
+    decode_dir_object, encode_dir_object, path_join, path_to_bucket_object, path_to_bucket_object_with_base_path,
+};
 use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, ReplicationConfiguration};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
@@ -4075,23 +4077,54 @@ pub(crate) struct PoolMetaWriteState {
     expected_cluster_id: Option<uuid::Uuid>,
     cluster_epoch: Option<u64>,
     pool_meta_absent: bool,
-    fresh_bootstrap_proven: bool,
+    bootstrap_authority: PoolMetaBootstrapAuthority,
     identity_initialized: Option<bool>,
     identity_fresh_bootstrap_nonce: Option<uuid::Uuid>,
     identity_needs_repair: bool,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub(crate) enum PoolMetaBootstrapAuthority {
+    #[default]
+    None,
+    Fresh,
+    LegacyAdoption,
+}
+
+impl PoolMetaBootstrapAuthority {
+    pub(crate) fn combine_across_pools(self, other: Self) -> Self {
+        if self == other { self } else { Self::None }
+    }
+
+    fn is_proven(self) -> bool {
+        !matches!(self, Self::None)
+    }
+}
+
 impl PoolMetaWriteState {
+    #[cfg(test)]
     pub(crate) fn for_startup(cluster_id: uuid::Uuid, fresh_bootstrap_proven: bool) -> Self {
+        let bootstrap_authority = if fresh_bootstrap_proven {
+            PoolMetaBootstrapAuthority::Fresh
+        } else {
+            PoolMetaBootstrapAuthority::None
+        };
+        Self::for_startup_with_bootstrap_authority(cluster_id, bootstrap_authority)
+    }
+
+    pub(crate) fn for_startup_with_bootstrap_authority(
+        cluster_id: uuid::Uuid,
+        bootstrap_authority: PoolMetaBootstrapAuthority,
+    ) -> Self {
         Self {
             expected_cluster_id: Some(cluster_id),
-            fresh_bootstrap_proven,
+            bootstrap_authority,
             ..Default::default()
         }
     }
 
-    pub(crate) fn fresh_bootstrap_proven(&self) -> bool {
-        self.fresh_bootstrap_proven
+    pub(crate) fn bootstrap_identity_proven(&self) -> bool {
+        self.bootstrap_authority.is_proven()
     }
 
     pub(crate) fn identity_is_pending(&self) -> bool {
@@ -4113,7 +4146,7 @@ impl PoolMetaWriteState {
     #[cfg(any(test, feature = "test-util"))]
     fn for_test_bootstrap() -> Self {
         Self {
-            fresh_bootstrap_proven: true,
+            bootstrap_authority: PoolMetaBootstrapAuthority::Fresh,
             identity_initialized: Some(false),
             identity_fresh_bootstrap_nonce: Some(uuid::Uuid::new_v4()),
             ..Default::default()
@@ -4174,7 +4207,7 @@ impl PoolMetaWriteState {
         self.identity_fresh_bootstrap_nonce = selection.identity.and_then(|identity| identity.fresh_bootstrap_nonce);
         if let Some(identity) = selection.identity {
             if identity.initialized {
-                self.fresh_bootstrap_proven = false;
+                self.bootstrap_authority = PoolMetaBootstrapAuthority::None;
             }
             if let Some(metadata_epoch) = self.cluster_epoch
                 && metadata_epoch != identity.epoch
@@ -4198,11 +4231,11 @@ impl PoolMetaWriteState {
             return Ok(());
         }
         match self.identity_initialized {
-            Some(false) if self.fresh_bootstrap_proven && self.identity_fresh_bootstrap_nonce.is_some() => Ok(()),
+            Some(false) if self.bootstrap_identity_proven() && self.identity_fresh_bootstrap_nonce.is_some() => Ok(()),
             Some(false) => {
                 self.block_writes();
                 Err(Error::other(
-                    "pool metadata recovery required: pending cluster identity exists but this startup has no verified fresh-bootstrap proof",
+                    "pool metadata recovery required: pending cluster identity exists but this startup has no verified fresh-bootstrap proof or legacy-adoption proof",
                 ))
             }
             Some(true) => {
@@ -5184,10 +5217,10 @@ where
                 ..identity
             },
             Some(identity) => identity,
-            None if !initialized && !write_state.fresh_bootstrap_proven() => {
+            None if !initialized && !write_state.bootstrap_identity_proven() => {
                 write_state.block_writes();
                 return Err(Error::other(
-                    "pool metadata recovery required: cannot create a pending cluster identity without verified fresh-bootstrap proof",
+                    "pool metadata recovery required: cannot create a pending cluster identity without verified fresh-bootstrap proof or legacy-adoption proof",
                 ));
             }
             None => PersistedPoolMetaIdentity {
@@ -7159,6 +7192,170 @@ pub(crate) fn decommission_capacity_mutation_id(
     uuid::Uuid::from_bytes(bytes)
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct ExactDeleteCapacityReconciliation {
+    source_pool_index: usize,
+    target_pool_index: usize,
+    mutation_id: uuid::Uuid,
+    expected_data_bytes: usize,
+    expected_target_physical_bytes: usize,
+}
+
+fn plan_exact_delete_capacity_reconciliations(
+    meta: &PoolMeta,
+    object: &str,
+    exact: &ObjectInfo,
+) -> Result<Vec<ExactDeleteCapacityReconciliation>> {
+    let version_id = exact.version_id.map(|version_id| version_id.to_string());
+    let mut matches = Vec::new();
+
+    for (source_pool_index, pool) in meta.pools.iter().enumerate() {
+        let Some(reservation) = pool
+            .decommission
+            .as_ref()
+            .and_then(|info| info.capacity_reservation.as_ref())
+            .filter(|reservation| reservation.active())
+        else {
+            continue;
+        };
+        let owner = DecommissionCapacityOwner {
+            source_pool_index,
+            operation_id: reservation.operation_id,
+            generation: reservation.generation,
+            owner_nonce: reservation.owner_nonce,
+            mutation_id: None,
+        };
+        let logical_mutation_id = decommission_capacity_mutation_id(
+            owner,
+            &exact.bucket,
+            &exact.name,
+            version_id.as_deref(),
+            exact.delete_marker,
+            exact.mod_time,
+        );
+        // Existing data-movement producers persist directory-key intents using
+        // either the logical name or its internal `__XLDIR__` representation.
+        // Accept both while retaining the exact persisted identity for CAS.
+        let internal_mutation_id = if object == exact.name {
+            logical_mutation_id
+        } else {
+            decommission_capacity_mutation_id(
+                owner,
+                &exact.bucket,
+                object,
+                version_id.as_deref(),
+                exact.delete_marker,
+                exact.mod_time,
+            )
+        };
+        let mut source_match = None;
+
+        for target in &reservation.targets {
+            if target.pending_physical_bytes == 0 {
+                continue;
+            }
+            let Some(pending_mutation_id) = target.pending_mutation_id else {
+                return Err(decommission_capacity_blocked_error(format!(
+                    "source pool {source_pool_index} target pool {} has pending capacity without an object identity",
+                    target.pool_index
+                )));
+            };
+            if pending_mutation_id != logical_mutation_id && pending_mutation_id != internal_mutation_id {
+                continue;
+            }
+            if source_match.is_some() {
+                return Err(decommission_capacity_blocked_error(format!(
+                    "source pool {source_pool_index} has the same exact-delete capacity intent on multiple targets"
+                )));
+            }
+            source_match = Some((target.pool_index, target.layout, target.pending_physical_bytes, pending_mutation_id));
+        }
+
+        let Some((target_pool_index, target_layout, pending_physical_bytes, mutation_id)) = source_match else {
+            continue;
+        };
+        if exact.version_id.is_none() && exact.mod_time.is_none() {
+            return Err(decommission_capacity_blocked_error(
+                "unversioned exact delete cannot identify pending capacity without a modification time",
+            ));
+        }
+        let expected_data_bytes = if exact.delete_marker {
+            0
+        } else {
+            usize::try_from(exact.size).map_err(|_| {
+                decommission_capacity_blocked_error("exact delete cannot reconcile a negative or overflowing object size")
+            })?
+        };
+        let expected_target_physical_bytes = capacity_target_physical_bytes(expected_data_bytes.max(1), target_layout)?;
+        if pending_physical_bytes != expected_target_physical_bytes {
+            return Err(decommission_capacity_blocked_error(format!(
+                "source pool {source_pool_index} target pool {target_pool_index} pending capacity does not match the exact object size"
+            )));
+        }
+        let remaining_target_physical_bytes = reservation
+            .targets
+            .iter()
+            .find(|target| target.pool_index == target_pool_index)
+            .map(|target| {
+                target.remaining_reserved_physical_bytes(reservation.temporary_copies)
+                    / 1usize.saturating_add(reservation.temporary_copies)
+            })
+            .unwrap_or_default();
+        let remaining_total_physical_bytes = reservation
+            .predicted_physical_bytes
+            .saturating_sub(reservation.consumed_target_physical_bytes);
+        let remaining_data_bytes = reservation
+            .source_data_equivalent_bytes
+            .saturating_sub(reservation.committed_data_bytes);
+        if expected_target_physical_bytes > remaining_target_physical_bytes
+            || expected_target_physical_bytes > remaining_total_physical_bytes
+            || expected_data_bytes > remaining_data_bytes
+        {
+            return Err(decommission_capacity_blocked_error(format!(
+                "source pool {source_pool_index} target pool {target_pool_index} lacks reservation capacity for the exact object"
+            )));
+        }
+        matches.push(ExactDeleteCapacityReconciliation {
+            source_pool_index,
+            target_pool_index,
+            mutation_id,
+            expected_data_bytes,
+            expected_target_physical_bytes,
+        });
+    }
+
+    Ok(matches)
+}
+
+fn ensure_exact_delete_capacity_namespace_fences(opts: &ObjectOptions, bucket: &str, object: &str) -> Result<()> {
+    let object_fence = opts.namespace_lock_fence.as_ref().ok_or_else(|| {
+        decommission_capacity_blocked_error("exact delete capacity reconciliation requires an object namespace fence")
+    })?;
+    if object_fence.is_lock_lost() {
+        return Err(StorageError::NamespaceLockQuorumUnavailable {
+            mode: "exact_delete_capacity_reconciliation",
+            bucket: bucket.to_string(),
+            object: decode_dir_object(object),
+            required: 1,
+            achieved: 0,
+        });
+    }
+    if opts
+        .bucket_lifecycle_lock_fence
+        .as_ref()
+        .is_some_and(crate::object_api::NamespaceLockFence::is_lock_lost)
+    {
+        return Err(StorageError::NamespaceLockQuorumUnavailable {
+            mode: "exact_delete_capacity_bucket_generation",
+            bucket: bucket.to_string(),
+            object: decode_dir_object(object),
+            required: 1,
+            achieved: 0,
+        });
+    }
+    Ok(())
+}
+
 pub(crate) fn ensure_decommission_capacity_mutation_id(bucket: &str, object: &str, opts: &mut ObjectOptions) {
     if opts
         .decommission_capacity
@@ -8222,6 +8419,117 @@ impl ECStore {
             .await
     }
 
+    pub(crate) async fn reconcile_decommission_capacity_before_exact_delete(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+        exact: &ObjectInfo,
+    ) -> Result<()> {
+        if exact.bucket != bucket || exact.name != decode_dir_object(object) {
+            return Err(decommission_capacity_blocked_error(
+                "exact delete object identity changed before capacity reconciliation",
+            ));
+        }
+
+        let reconciliations = {
+            let mut save_guard = self.pool_meta_save_gate.lock().await;
+            let (_read_guard, snapshot) = self
+                .acquire_pool_meta_read_guard(&mut save_guard, "exact delete capacity reconciliation failed")
+                .await?;
+            plan_exact_delete_capacity_reconciliations(&snapshot, object, exact)?
+        };
+        if reconciliations.is_empty() {
+            return Ok(());
+        }
+        ensure_exact_delete_capacity_namespace_fences(opts, bucket, object)?;
+
+        let target_lookup_options = ObjectOptions {
+            versioned: opts.versioned,
+            version_suspended: opts.version_suspended,
+            version_id: opts.version_id.clone(),
+            metadata_chg: opts.version_id.is_some(),
+            no_lock: true,
+            ..Default::default()
+        };
+        for reconciliation in &reconciliations {
+            let target_pool = self.pools.get(reconciliation.target_pool_index).ok_or_else(|| {
+                decommission_capacity_blocked_error(format!(
+                    "source pool {} exact-delete capacity target pool {} is out of range",
+                    reconciliation.source_pool_index, reconciliation.target_pool_index
+                ))
+            })?;
+            let target = target_pool
+                .get_object_info(bucket, object, &target_lookup_options)
+                .await
+                .map_err(|err| {
+                    decommission_capacity_blocked_error(format!(
+                        "source pool {} target pool {} exact object evidence could not be read: {err}",
+                        reconciliation.source_pool_index, reconciliation.target_pool_index
+                    ))
+                })?;
+            if !Self::is_equivalent_decommission_capacity_target(exact, &target) {
+                return Err(decommission_capacity_blocked_error(format!(
+                    "source pool {} target pool {} does not contain an equivalent exact object for its pending capacity intent",
+                    reconciliation.source_pool_index, reconciliation.target_pool_index
+                )));
+            }
+        }
+        ensure_exact_delete_capacity_namespace_fences(opts, bucket, object)?;
+
+        let mut save_guard = self.pool_meta_save_gate.lock().await;
+        let (write_guard, mut snapshot) = self
+            .acquire_pool_meta_write_guard(&mut save_guard, "exact delete capacity reconciliation failed")
+            .await?;
+        let current_reconciliations = plan_exact_delete_capacity_reconciliations(&snapshot, object, exact)?;
+        if current_reconciliations != reconciliations {
+            return Err(decommission_capacity_blocked_error(
+                "pending capacity changed while exact target evidence was being verified",
+            ));
+        }
+        ensure_exact_delete_capacity_namespace_fences(opts, bucket, object)?;
+
+        let now = OffsetDateTime::now_utc();
+        let mut source_pool_indices = Vec::with_capacity(current_reconciliations.len());
+        for reconciliation in current_reconciliations {
+            resolve_decommission_target_pending(
+                &mut snapshot,
+                reconciliation.source_pool_index,
+                reconciliation.target_pool_index,
+                reconciliation.expected_target_physical_bytes,
+                reconciliation.mutation_id,
+            )?;
+            record_decommission_target_consumption(
+                &mut snapshot,
+                reconciliation.source_pool_index,
+                reconciliation.target_pool_index,
+                DecommissionTargetConsumption {
+                    committed_data_bytes: reconciliation.expected_data_bytes,
+                    target_physical_bytes: reconciliation.expected_target_physical_bytes,
+                    observed_physical_bytes: 0,
+                },
+                reconciliation.mutation_id,
+                now,
+            )?;
+            source_pool_indices.push(reconciliation.source_pool_index);
+        }
+        source_pool_indices.sort_unstable();
+        source_pool_indices.dedup();
+        ensure_exact_delete_capacity_namespace_fences(opts, bucket, object)?;
+
+        let outcome = snapshot
+            .save_no_lock_armed(self.pools.clone(), &mut save_guard, write_guard.lock_lost_signal(), &source_pool_indices)
+            .await?;
+        ensure_pool_meta_write_fence(&write_guard, "exact delete capacity reconciliation save failed")?;
+        {
+            let mut pool_meta = self.pool_meta.write().await;
+            publish_pool_meta_updates(&mut pool_meta, &outcome.committed, &source_pool_indices);
+        }
+        ensure_pool_meta_write_fence(&write_guard, "exact delete capacity reconciliation save failed")?;
+        outcome.disarm();
+        Ok(())
+    }
+
     pub(crate) async fn reconcile_decommission_capacity_after_equivalent_target(
         &self,
         owner: DecommissionCapacityOwner,
@@ -17149,7 +17457,8 @@ mod pools_tests {
     use super::{
         DecommissionCapacityOwner, DecommissionCapacityReservation, DecommissionCapacityTemporaryMutation,
         decommission_capacity_mutation_id, ensure_decommission_target_owner_admission,
-        ensure_external_decommission_target_admission, is_decommission_capacity_blocked_error,
+        ensure_exact_delete_capacity_namespace_fences, ensure_external_decommission_target_admission,
+        is_decommission_capacity_blocked_error, plan_exact_delete_capacity_reconciliations,
         record_decommission_target_consumption, reserve_decommission_target_pending, resolve_decommission_target_pending,
         set_decommission_capacity_info_overrides_for_test,
     };
@@ -17164,7 +17473,7 @@ mod pools_tests {
     use crate::disk::{STORAGE_FORMAT_FILE, endpoint::Endpoint};
     use crate::error::{Error, StorageError};
     use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
-    use crate::object_api::ObjectOptions;
+    use crate::object_api::{ObjectInfo, ObjectOptions};
     use crate::runtime::instance::InstanceContext;
     use crate::services::rebalance::{RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats};
     use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
@@ -21285,6 +21594,135 @@ mod pools_tests {
         assert_eq!(first, recovered, "a lease nonce rotation must not change the mutation identity");
     }
 
+    #[test]
+    fn exact_delete_capacity_plan_requires_identity_and_exact_size() {
+        let now = OffsetDateTime::UNIX_EPOCH + Duration::minutes(2);
+        let layout = DecommissionErasureLayout { data: 1, parity: 0 };
+        let capacity_infos = vec![
+            DecommissionPoolCapacityInfo::for_test(0, layout, 0, 30, 30),
+            DecommissionPoolCapacityInfo::for_test(1, layout, 60, 60, 0),
+        ];
+        let mut meta = PoolMeta {
+            version: POOL_META_VERSION,
+            pools: vec![decommission_test_pool_status(0, None), decommission_test_pool_status(1, None)],
+            ..Default::default()
+        };
+        meta.decommission(0, capacity_infos[0].space).unwrap();
+        reserve_decommission_start_target_capacity(&mut meta, &[0], &capacity_infos, uuid::Uuid::new_v4(), 1, now)
+            .expect("the exact-delete test reservation should fit");
+        let exact = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "object".to_string(),
+            version_id: Some(uuid::Uuid::from_u128(7)),
+            mod_time: Some(now),
+            size: 10,
+            ..Default::default()
+        };
+        let owner = {
+            let reservation = meta.pools[0]
+                .decommission
+                .as_ref()
+                .and_then(|info| info.capacity_reservation.as_ref())
+                .expect("the exact-delete test reservation should exist");
+            DecommissionCapacityOwner {
+                source_pool_index: 0,
+                operation_id: reservation.operation_id,
+                generation: reservation.generation,
+                owner_nonce: reservation.owner_nonce,
+                mutation_id: None,
+            }
+        };
+        let version_id = exact.version_id.map(|version_id| version_id.to_string());
+        let mutation_id = decommission_capacity_mutation_id(
+            owner,
+            &exact.bucket,
+            &exact.name,
+            version_id.as_deref(),
+            exact.delete_marker,
+            exact.mod_time,
+        );
+        reserve_decommission_target_pending(&mut meta, 0, 1, 10, mutation_id, now + Duration::seconds(1))
+            .expect("the exact-delete test intent should be reserved");
+
+        let plan = plan_exact_delete_capacity_reconciliations(&meta, &exact.name, &exact)
+            .expect("the exact identity should match the pending intent");
+        assert_eq!(plan.len(), 1);
+        assert_eq!(plan[0].source_pool_index, 0);
+        assert_eq!(plan[0].target_pool_index, 1);
+        assert_eq!(plan[0].expected_data_bytes, 10);
+        assert_eq!(plan[0].expected_target_physical_bytes, 10);
+
+        let mismatched_size = ObjectInfo {
+            size: 9,
+            ..exact.clone()
+        };
+        let mismatched_size = plan_exact_delete_capacity_reconciliations(&meta, &mismatched_size.name, &mismatched_size)
+            .expect_err("a different exact size must not consume the pending intent");
+        assert!(mismatched_size.to_string().contains("does not match the exact object size"));
+
+        let directory_exact = ObjectInfo {
+            name: "directory/".to_string(),
+            ..exact.clone()
+        };
+        let internal_directory = rustfs_utils::path::encode_dir_object(&directory_exact.name);
+        let internal_directory_mutation_id = decommission_capacity_mutation_id(
+            owner,
+            &directory_exact.bucket,
+            &internal_directory,
+            version_id.as_deref(),
+            directory_exact.delete_marker,
+            directory_exact.mod_time,
+        );
+        meta.pools[0]
+            .decommission
+            .as_mut()
+            .and_then(|info| info.capacity_reservation.as_mut())
+            .expect("the exact-delete test reservation should exist")
+            .targets[0]
+            .pending_mutation_id = Some(internal_directory_mutation_id);
+        let directory_plan = plan_exact_delete_capacity_reconciliations(&meta, &internal_directory, &directory_exact)
+            .expect("an internally encoded directory intent should match its logical exact object");
+        assert_eq!(directory_plan[0].mutation_id, internal_directory_mutation_id);
+
+        let logical_directory_mutation_id = decommission_capacity_mutation_id(
+            owner,
+            &directory_exact.bucket,
+            &directory_exact.name,
+            version_id.as_deref(),
+            directory_exact.delete_marker,
+            directory_exact.mod_time,
+        );
+        meta.pools[0]
+            .decommission
+            .as_mut()
+            .and_then(|info| info.capacity_reservation.as_mut())
+            .expect("the exact-delete test reservation should exist")
+            .targets[0]
+            .pending_mutation_id = Some(logical_directory_mutation_id);
+        let directory_plan = plan_exact_delete_capacity_reconciliations(&meta, &internal_directory, &directory_exact)
+            .expect("a logical directory intent should match its internally encoded delete path");
+        assert_eq!(directory_plan[0].mutation_id, logical_directory_mutation_id);
+
+        meta.pools[0]
+            .decommission
+            .as_mut()
+            .and_then(|info| info.capacity_reservation.as_mut())
+            .expect("the exact-delete test reservation should exist")
+            .targets[0]
+            .pending_mutation_id = None;
+        let unidentified = plan_exact_delete_capacity_reconciliations(&meta, &exact.name, &exact)
+            .expect_err("an unidentified pending intent must fail closed");
+        assert!(unidentified.to_string().contains("without an object identity"));
+
+        let mut opts = ObjectOptions::default();
+        let unfenced = ensure_exact_delete_capacity_namespace_fences(&opts, &exact.bucket, &exact.name)
+            .expect_err("capacity reconciliation must reject a missing object namespace fence");
+        assert!(unfenced.to_string().contains("requires an object namespace fence"));
+        opts.ensure_namespace_lock_fence();
+        ensure_exact_delete_capacity_namespace_fences(&opts, &exact.bucket, &exact.name)
+            .expect("a live object namespace fence should admit capacity reconciliation");
+    }
+
     #[test]
     fn ordinary_write_admission_cannot_race_into_a_reserved_target() {
         let now = OffsetDateTime::UNIX_EPOCH + Duration::minutes(2);

crates/ecstore/src/core/pools_test.rs

@@ -196,7 +196,7 @@ mod decommission_lock_order_tests {
     use crate::bucket::lifecycle::lifecycle::TRANSITION_PENDING;
     use crate::core::pools::{
         DecommissionCapacityLockOrderBarrier, DecommissionCapacityOwner, DecommissionErasureLayout, DecommissionPoolCapacityInfo,
-        POOL_META_NAME, set_decommission_capacity_info_overrides_for_test,
+        POOL_META_NAME, decommission_capacity_mutation_id, set_decommission_capacity_info_overrides_for_test,
     };
     use crate::data_movement;
     use crate::disk::RUSTFS_META_BUCKET;
@@ -3047,6 +3047,198 @@ mod decommission_lock_order_tests {
         );
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn exact_delete_reconciles_pending_capacity_before_removing_replicas() {
+        let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
+        let bucket = test_bucket("exact-delete-capacity");
+        let object = "published-before-exact-delete.bin";
+        let body = vec![0x55; 64 * 1024];
+        let version_id = uuid::Uuid::new_v4().to_string();
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create the exact-delete reconciliation bucket");
+        let incarnation = store
+            .bucket_incarnation_id(&bucket)
+            .await
+            .expect("load the exact-delete bucket incarnation");
+
+        let mut source_data = PutObjReader::from_vec(body.clone());
+        let source = store.pools[0]
+            .put_object(
+                &bucket,
+                object,
+                &mut source_data,
+                &ObjectOptions {
+                    versioned: true,
+                    version_id: Some(version_id.clone()),
+                    expected_bucket_incarnation_id: Some(incarnation),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("seed the exact source version");
+        let layout = DecommissionErasureLayout { data: 1, parity: 0 };
+        let target_total = body.len().saturating_mul(4);
+        set_decommission_capacity_info_overrides_for_test(
+            store.id,
+            vec![vec![
+                DecommissionPoolCapacityInfo::for_test(0, layout, 0, body.len(), body.len()),
+                DecommissionPoolCapacityInfo::for_test(1, layout, 0, target_total, target_total),
+                DecommissionPoolCapacityInfo::for_test(2, layout, target_total, target_total, 0),
+            ]],
+        );
+        store
+            .save_current_pool_meta_for_decommission_start(&[0], Vec::new())
+            .await
+            .expect("activate the exact-delete capacity reservation");
+        let owner = decommission_capacity_owner(&*store.pool_meta.read().await);
+        let target_pool_index = store.pool_meta.read().await.pools[0]
+            .decommission
+            .as_ref()
+            .and_then(|info| info.capacity_reservation.as_ref())
+            .expect("the exact-delete capacity reservation should exist")
+            .targets[0]
+            .pool_index;
+        assert_eq!(target_pool_index, 2);
+
+        let target_options = ObjectOptions {
+            versioned: true,
+            version_id: Some(version_id.clone()),
+            mod_time: source.mod_time,
+            preserve_etag: source.etag.clone(),
+            user_defined: (*source.user_defined).clone(),
+            data_movement: true,
+            src_pool_idx: 0,
+            expected_bucket_incarnation_id: Some(incarnation),
+            ..Default::default()
+        };
+        let mut target_data = PutObjReader::from_vec(body.clone());
+        let target = store.pools[target_pool_index]
+            .put_object(&bucket, object, &mut target_data, &target_options)
+            .await
+            .expect("publish the target version before capacity progress");
+        assert_eq!(target.version_id, source.version_id);
+        assert_eq!(target.mod_time, source.mod_time);
+        assert_eq!(target.size, source.size);
+
+        let source_version_id = source.version_id.map(|version_id| version_id.to_string());
+        let mutation_id = decommission_capacity_mutation_id(
+            owner,
+            &source.bucket,
+            &source.name,
+            source_version_id.as_deref(),
+            source.delete_marker,
+            source.mod_time,
+        );
+        {
+            let mut pool_meta = store.pool_meta.write().await;
+            let source_pool = &mut pool_meta.pools[0];
+            let reservation = source_pool
+                .decommission
+                .as_mut()
+                .and_then(|info| info.capacity_reservation.as_mut())
+                .expect("the exact-delete capacity reservation should remain active");
+            let target = reservation
+                .targets
+                .iter_mut()
+                .find(|target| target.pool_index == target_pool_index)
+                .expect("the exact-delete target allocation should exist");
+            target.pending_physical_bytes = body.len();
+            target.pending_mutation_id = Some(mutation_id);
+            reservation.pending_target_physical_bytes = body.len();
+            source_pool.last_update = time::OffsetDateTime::now_utc();
+        }
+        store
+            .save_current_pool_meta_for_test(&[0])
+            .await
+            .expect("persist the simulated post-commit capacity intent");
+
+        let exact_delete_options = ObjectOptions {
+            versioned: true,
+            version_id: Some(version_id.clone()),
+            expected_bucket_incarnation_id: Some(incarnation),
+            ..Default::default()
+        };
+        store.pools[target_pool_index]
+            .delete_object(&bucket, object, exact_delete_options.clone())
+            .await
+            .expect("remove the target evidence before the fail-closed exact delete");
+        let delete_err = store
+            .delete_object(&bucket, object, exact_delete_options.clone())
+            .await
+            .expect_err("exact delete must fail while its pending target evidence is absent");
+        assert!(matches!(delete_err, crate::error::Error::DecommissionCapacityBlocked { .. }));
+        store.pools[0]
+            .get_object_info(
+                &bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    version_id: Some(version_id.clone()),
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("a failed reconciliation must preserve the source evidence");
+        let mut failed = crate::core::pools::PoolMeta::default();
+        failed
+            .load_no_lock_from_replicas(store.pools.clone())
+            .await
+            .expect("the failed exact delete must preserve readable capacity metadata");
+        let failed_reservation = failed.pools[0]
+            .decommission
+            .as_ref()
+            .and_then(|info| info.capacity_reservation.as_ref())
+            .expect("the failed exact delete reservation should remain present");
+        assert_eq!(failed_reservation.pending_target_physical_bytes, body.len());
+        assert_eq!(failed_reservation.consumed_target_physical_bytes, 0);
+
+        let mut replacement_target_data = PutObjReader::from_vec(body.clone());
+        store.pools[target_pool_index]
+            .put_object(&bucket, object, &mut replacement_target_data, &target_options)
+            .await
+            .expect("restore the equivalent target evidence for the exact-delete retry");
+
+        store
+            .delete_object(&bucket, object, exact_delete_options)
+            .await
+            .expect("the exact delete should reconcile capacity before removing replicas");
+
+        let mut persisted = crate::core::pools::PoolMeta::default();
+        persisted
+            .load_no_lock_from_replicas(store.pools.clone())
+            .await
+            .expect("the exact-delete reconciliation should remain durable");
+        let reservation = persisted.pools[0]
+            .decommission
+            .as_ref()
+            .and_then(|info| info.capacity_reservation.as_ref())
+            .expect("the reconciled reservation should remain present");
+        assert_eq!(reservation.pending_target_physical_bytes, 0);
+        assert_eq!(reservation.consumed_target_physical_bytes, body.len());
+        assert_eq!(reservation.committed_data_bytes, body.len());
+
+        for pool_index in [0, target_pool_index] {
+            let err = store.pools[pool_index]
+                .get_object_info(
+                    &bucket,
+                    object,
+                    &ObjectOptions {
+                        versioned: true,
+                        version_id: Some(version_id.clone()),
+                        no_lock: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect_err("the exact version should be absent after reconciliation and delete");
+            assert!(crate::error::is_err_object_not_found(&err) || crate::error::is_err_version_not_found(&err));
+        }
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn data_movement_equivalent_target_reconciles_published_capacity_after_restart() {

crates/ecstore/src/object_api/mod.rs

@@ -37,7 +37,9 @@ use rustfs_filemeta::{FileInfo, MetaCacheEntriesSorted, ObjectPartInfo, RestoreS
 use rustfs_rio::Checksum;
 use rustfs_utils::CompressionAlgorithm;
 use rustfs_utils::http::headers::AMZ_OBJECT_TAGGING;
-use rustfs_utils::http::{AMZ_BUCKET_REPLICATION_STATUS, AMZ_RESTORE, AMZ_STORAGE_CLASS};
+use rustfs_utils::http::{
+    AMZ_BUCKET_REPLICATION_STATUS, AMZ_RESTORE, AMZ_STORAGE_CLASS, SUFFIX_PLAINTEXT_CHECKSUM, get_consistent_str,
+};
 use rustfs_utils::path::decode_dir_object;
 use std::collections::HashMap;
 use std::fmt::Debug;

crates/ecstore/src/object_api/types.rs

@@ -1771,9 +1771,10 @@ impl ObjectInfo {
         }
 
         if let Some(data) = &self.checksum {
-            if self.is_encrypted() {
+            if self.is_encrypted() && get_consistent_str(&self.user_defined, SUFFIX_PLAINTEXT_CHECKSUM) != Some("true") {
                 // Object-level encrypted checksum bytes require SSE decrypt material,
-                // so do not expose them as plaintext checksum headers here. The
+                // unless RustFS marked the stored bytes as plaintext. Do not expose
+                // unmarked bytes as checksum headers here. The
                 // `false` multipart flag feeds the response-path COMPOSITE
                 // fallback; callers that need accurate multipart routing must
                 // consult `is_multipart()` instead of this value.
@@ -2479,6 +2480,31 @@ mod tests {
         assert!(checksums.is_empty());
     }
 
+    #[test]
+    fn decrypt_checksums_reads_marked_rustfs_encrypted_object_checksum() {
+        let checksum = rustfs_rio::Checksum::new_from_data(rustfs_rio::ChecksumType::CRC32, b"encrypted-object")
+            .expect("test checksum should be valid");
+        let checksum_key = checksum.checksum_type.to_string();
+        let expected_checksum = checksum.encoded.clone();
+        let mut user_defined =
+            HashMap::from([(rustfs_utils::http::headers::AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "AES256".to_string())]);
+        rustfs_utils::http::insert_str(&mut user_defined, SUFFIX_PLAINTEXT_CHECKSUM, "true".to_string());
+        assert_eq!(user_defined.get("x-rustfs-internal-plaintext-checksum").map(String::as_str), Some("true"));
+        assert_eq!(user_defined.get("x-minio-internal-plaintext-checksum").map(String::as_str), Some("true"));
+        let info = ObjectInfo {
+            checksum: Some(checksum.to_bytes(&[])),
+            user_defined: Arc::new(user_defined),
+            ..Default::default()
+        };
+
+        let (checksums, is_multipart) = info
+            .decrypt_checksums(0, &HeaderMap::new())
+            .expect("marked RustFS checksum should decode");
+
+        assert!(!is_multipart);
+        assert_eq!(checksums.get(&checksum_key), Some(&expected_checksum));
+    }
+
     #[test]
     fn decrypt_checksums_keeps_encrypted_multipart_flag_false_for_response_paths() {
         let checksum = rustfs_rio::Checksum::new_from_data(rustfs_rio::ChecksumType::CRC32, b"encrypted-object")

crates/ecstore/src/set_disk/mod.rs

@@ -127,7 +127,7 @@ use rustfs_filemeta::{
 };
 use rustfs_heal_contracts::heal_channel::{
     DriveState, HealAdmissionResult, HealChannelPriority, HealItemType, HealOpts, HealRequestSource, HealScanMode,
-    send_heal_disk, send_heal_request_with_admission,
+    send_heal_replacement_disk, send_heal_request_with_admission,
 };
 use rustfs_io_metrics::{
     record_object_lock_diag_acquire_duration, record_object_lock_diag_enabled, record_object_lock_diag_hold_duration,

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

@@ -22,12 +22,10 @@
 use super::super::{
     Arc, DiskError, DiskInfo, DiskInfoOptions, DiskOption, DiskStore, Endpoint, Error, FormatV3, HealChannelPriority, LockResult,
     NamespaceLock, NamespaceLockWrapper, ObjectKey, Result, SetDisks, StorageError, debug, disk, info, load_format_erasure,
-    send_heal_disk, warn,
+    send_heal_replacement_disk, warn,
 };
-use crate::disk::DiskAPI;
 use crate::disk::health_state::DriveMembershipSnapshot;
-#[cfg(test)]
-use crate::disk::new_disk;
+use crate::disk::{DiskAPI, new_disk};
 use crate::runtime::sources as runtime_sources;
 use rand::prelude::SliceRandom;
 #[cfg(test)]
@@ -356,11 +354,28 @@ impl SetDisks {
             Ok(res) => res,
             Err(e) => {
                 warn!("renew_disk: connect_endpoint err {:?}", &e);
-                if ep.is_local && e == DiskError::UnformattedDisk {
-                    info!("renew_disk unformatteddisk will trigger heal_disk, {:?}", ep);
-                    let set_disk_id = format!("pool_{}_set_{}", ep.pool_idx, ep.set_idx);
-                    let _ = send_heal_disk(set_disk_id, Some(HealChannelPriority::Normal)).await;
+                if !matches!(e, DiskError::UnformattedDisk | DiskError::Io(_)) {
+                    return;
                 }
+
+                let attached = match self.attach_unformatted_replacement_disk(ep).await {
+                    Ok(attached) => attached,
+                    Err(err) => {
+                        warn!(endpoint = %ep, error = ?err, "renew_disk: unformatted replacement probe failed");
+                        return;
+                    }
+                };
+                if !attached {
+                    return;
+                }
+
+                info!("renew_disk attached unformatted replacement and will trigger heal_disk, {:?}", ep);
+                let (Ok(pool_index), Ok(set_index)) = (usize::try_from(ep.pool_idx), usize::try_from(ep.set_idx)) else {
+                    warn!("renew_disk: replacement target has invalid pool or set index, {:?}", ep);
+                    return;
+                };
+                let _ =
+                    send_heal_replacement_disk(pool_index, set_index, ep.to_string(), Some(HealChannelPriority::Normal)).await;
                 return;
             }
         };
@@ -412,6 +427,60 @@ impl SetDisks {
         disk_lock[disk_idx] = Some(new_disk);
     }
 
+    /// Attach a replacement target only after proving that the exact local slot
+    /// is present and unformatted. A health-checked reconnect may reject a
+    /// blank target before it reaches the format-heal path; that target still
+    /// has to be visible in this set for the formatter to claim it safely.
+    async fn attach_unformatted_replacement_disk(&self, ep: &Endpoint) -> disk::error::Result<bool> {
+        if !ep.is_local
+            || usize::try_from(ep.pool_idx).ok() != Some(self.pool_index)
+            || usize::try_from(ep.set_idx).ok() != Some(self.set_index)
+        {
+            return Ok(false);
+        }
+
+        let Some(disk_idx) = self.set_endpoints.iter().position(|candidate| candidate == ep) else {
+            return Ok(false);
+        };
+
+        let replacement = new_disk(
+            ep,
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+        )
+        .await?;
+        match load_format_erasure(&replacement, false).await {
+            Err(DiskError::UnformattedDisk) => {}
+            Ok(_) => return Ok(false),
+            Err(err) => return Err(err),
+        }
+
+        {
+            let mut disks = self.disks.write().await;
+            if disks[disk_idx].as_ref().is_some_and(|existing| existing.endpoint() != *ep) {
+                warn!(endpoint = %ep, disk_idx, "renew_disk rejected unformatted replacement for an occupied foreign slot");
+                return Ok(false);
+            }
+            disks[disk_idx] = Some(replacement.clone());
+        }
+
+        let local_disk_map = runtime_sources::local_disk_map_handle();
+        local_disk_map
+            .write()
+            .await
+            .insert(replacement.endpoint().to_string(), Some(replacement.clone()));
+
+        if runtime_sources::setup_is_dist_erasure().await {
+            let local_disk_set_drives = runtime_sources::local_disk_set_drives_handle();
+            let mut local_set_drives = local_disk_set_drives.write().await;
+            local_set_drives[self.pool_index][self.set_index][disk_idx] = Some(replacement);
+        }
+
+        Ok(true)
+    }
+
     pub(in crate::set_disk) fn find_disk_index(&self, fm: &FormatV3) -> Result<(usize, usize)> {
         self.format.check_other(fm)?;
 
@@ -779,6 +848,75 @@ mod tests {
         drop(temp_dirs);
     }
 
+    #[tokio::test]
+    async fn renew_disk_attaches_only_a_verified_local_unformatted_replacement() {
+        let disk_count = 4;
+        let format = FormatV3::new(1, disk_count);
+        let mut temp_dirs = Vec::with_capacity(disk_count);
+        let mut endpoints = Vec::with_capacity(disk_count);
+        let mut disks = Vec::with_capacity(disk_count);
+
+        for disk_idx in 0..disk_count - 1 {
+            let (temp_dir, endpoint, disk) = make_formatted_local_disk(disk_idx, &format).await;
+            temp_dirs.push(temp_dir);
+            endpoints.push(endpoint);
+            disks.push(Some(disk));
+        }
+
+        let replacement_dir = tempfile::tempdir().expect("replacement tempdir should be created");
+        let mut replacement_endpoint =
+            Endpoint::try_from(replacement_dir.path().to_str().expect("replacement path should be utf8"))
+                .expect("replacement endpoint should parse");
+        replacement_endpoint.set_pool_index(0);
+        replacement_endpoint.set_set_index(0);
+        replacement_endpoint.set_disk_index(disk_count - 1);
+        temp_dirs.push(replacement_dir);
+        endpoints.push(replacement_endpoint.clone());
+        disks.push(None);
+
+        let set_disks = SetDisks::new(
+            "test-owner".to_string(),
+            Arc::new(RwLock::new(disks)),
+            disk_count,
+            disk_count / 2,
+            0,
+            0,
+            endpoints,
+            format,
+            Vec::new(),
+        )
+        .await;
+
+        assert!(
+            set_disks
+                .attach_unformatted_replacement_disk(&replacement_endpoint)
+                .await
+                .expect("a blank local replacement should be admitted")
+        );
+
+        let attached = set_disks.get_disks_internal().await;
+        let replacement = attached[disk_count - 1]
+            .as_ref()
+            .expect("the verified replacement must occupy its exact set slot");
+        assert_eq!(replacement.endpoint(), replacement_endpoint);
+        assert!(
+            !replacement.health_check_enabled_for_test(),
+            "the blank replacement must not start health checks before it receives a format"
+        );
+        assert_eq!(
+            load_format_erasure(replacement, false).await.unwrap_err(),
+            DiskError::UnformattedDisk,
+            "only a still-unformatted replacement may be attached by the fallback"
+        );
+
+        runtime_sources::local_disk_map_handle()
+            .write()
+            .await
+            .remove(&replacement_endpoint.to_string());
+        drop(set_disks);
+        drop(temp_dirs);
+    }
+
     #[tokio::test]
     async fn renew_disk_rejects_a_format_from_another_slot_or_cluster() {
         let disk_count = 3;

crates/ecstore/src/store/init.rs

@@ -14,8 +14,8 @@
 
 use super::*;
 use crate::core::pools::{
-    PoolMetaReplicaState, PoolMetaWriteState, load_pool_meta_identity_observing, local_decommission_queue_prefix,
-    persist_pool_meta_identity_for_startup, pool_meta_has_active_decommission,
+    PoolMetaBootstrapAuthority, PoolMetaReplicaState, PoolMetaWriteState, load_pool_meta_identity_observing,
+    local_decommission_queue_prefix, persist_pool_meta_identity_for_startup, pool_meta_has_active_decommission,
 };
 use crate::runtime::instance::InstanceContext;
 use crate::runtime::sources as runtime_sources;
@@ -173,7 +173,7 @@ async fn establish_pool_meta_bootstrap_identity_if_proven<S>(
 where
     S: EcstoreObjectIO,
 {
-    if elected_writer && write_state.fresh_bootstrap_proven() {
+    if elected_writer && write_state.bootstrap_identity_proven() {
         persist_pool_meta_identity_for_startup(pools, write_state, false).await?;
     }
     Ok(())
@@ -215,7 +215,7 @@ where
     }
     let mut committed = meta.clone();
     if should_write {
-        if write_state.fresh_bootstrap_proven() || write_state.identity_is_pending() {
+        if write_state.bootstrap_identity_proven() || write_state.identity_is_pending() {
             persist_pool_meta_identity_for_startup(pools.clone(), write_state, false)
                 .await
                 .map_err(|err| Error::other(format!("store init failed during prepare_pool_meta_identity: {err}")))?;
@@ -402,7 +402,7 @@ impl ECStore {
         preflight_startup_rpc_secret(&endpoint_pools)?;
 
         let mut deployment_id = None;
-        let mut fresh_bootstrap_proven = true;
+        let mut pool_meta_bootstrap_authority = None;
 
         // let (endpoint_pools, _) = EndpointServerPools::create_server_endpoints(address.as_str(), &layouts)?;
 
@@ -518,7 +518,12 @@ impl ECStore {
                     }
                 }
             }?;
-            fresh_bootstrap_proven &= loaded_format.fresh_bootstrap_proven;
+            pool_meta_bootstrap_authority = Some(pool_meta_bootstrap_authority.map_or(
+                loaded_format.pool_meta_bootstrap_authority,
+                |authority: PoolMetaBootstrapAuthority| {
+                    authority.combine_across_pools(loaded_format.pool_meta_bootstrap_authority)
+                },
+            ));
             let fm = loaded_format.format;
 
             // Format loading succeeded, enable health monitoring on all disks
@@ -559,6 +564,10 @@ impl ECStore {
         let peer_sys = S3PeerSys::new_with_instance_ctx(&endpoint_pools, instance_ctx.clone());
         let mut pool_meta = PoolMeta::new(&pools, &PoolMeta::default());
         pool_meta.dont_save = true;
+        let pool_meta_write_state = PoolMetaWriteState::for_startup_with_bootstrap_authority(
+            deployment_id,
+            pool_meta_bootstrap_authority.unwrap_or_default(),
+        );
 
         let decommission_cancelers = RwLock::new(vec![None; pools.len()]);
         let ec = Arc::new(ECStore {
@@ -570,7 +579,7 @@ impl ECStore {
             rebalance_meta: RwLock::new(None),
             decommission_cancelers,
             start_gate: Mutex::new(()),
-            pool_meta_save_gate: Mutex::new(PoolMetaWriteState::for_startup(deployment_id, fresh_bootstrap_proven)),
+            pool_meta_save_gate: Mutex::new(pool_meta_write_state),
             decommission_capacity_entry_gate: Mutex::default(),
             // Adopt the caller's context (the process bootstrap one on the
             // legacy path) so startup writes (erasure type recorded before
@@ -791,6 +800,7 @@ mod tests {
         run_local_decommission_watchdog, save_validated_pool_meta_for_startup, should_auto_start_rebalance_after_init,
         should_defer_rebalance_auto_start, should_retry_format_load, wait_for_local_decommission_resume_delay,
     };
+    use crate::core::pools::PoolMetaBootstrapAuthority;
     #[cfg(feature = "test-util")]
     use crate::disk::DiskAPI;
     #[cfg(feature = "test-util")]
@@ -1151,6 +1161,65 @@ mod tests {
             .expect("successful startup publication must disarm the transaction guard");
     }
 
+    #[tokio::test]
+    async fn test_legacy_adoption_pool_meta_bootstrap_reaches_v3_cas() {
+        let deployment_id = Uuid::new_v4();
+        let storage = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
+        let mut write_state =
+            PoolMetaWriteState::for_startup_with_bootstrap_authority(deployment_id, PoolMetaBootstrapAuthority::LegacyAdoption);
+
+        establish_pool_meta_bootstrap_identity_if_proven(vec![storage.clone()], &mut write_state, true)
+            .await
+            .expect("verified legacy adoption should persist a nonce-bound identity before loading pool metadata");
+        let (loaded, replica_state) = load_pool_meta_for_startup(vec![storage.clone()], &mut write_state)
+            .await
+            .expect("verified legacy adoption should authorize initially missing pool metadata");
+        assert!(loaded.pools.is_empty());
+
+        let committed = persist_pool_meta_for_startup_if_safe(
+            &init_test_pool_meta(None),
+            vec![storage.clone()],
+            replica_state,
+            &mut write_state,
+            true,
+            true,
+        )
+        .await
+        .expect("verified legacy adoption should publish initial pool metadata");
+        assert_eq!(committed.pools[0].cmd_line, "pool-0");
+
+        let objects = storage.objects.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        assert!(objects.contains_key(POOL_META_NAME));
+        let identity = objects
+            .get(POOL_META_IDENTITY_NAME)
+            .map(|(payload, _)| payload.clone())
+            .expect("legacy adoption should commit the bootstrap identity");
+        assert!(pool_meta_identity_initialized_for_test(&identity).expect("decode committed identity"));
+    }
+
+    #[tokio::test]
+    async fn test_non_elected_legacy_adoption_cannot_initialize_pool_meta() {
+        let storage = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
+        let mut write_state =
+            PoolMetaWriteState::for_startup_with_bootstrap_authority(Uuid::new_v4(), PoolMetaBootstrapAuthority::LegacyAdoption);
+
+        establish_pool_meta_bootstrap_identity_if_proven(vec![storage.clone()], &mut write_state, false)
+            .await
+            .expect("a non-elected distributed node must not create legacy adoption authority");
+        let err = load_pool_meta_for_startup(vec![storage.clone()], &mut write_state)
+            .await
+            .expect_err("legacy adoption still requires the elected writer to create a durable identity");
+        assert!(err.to_string().contains("no durable bootstrap identity"));
+        assert!(
+            !storage
+                .objects
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .contains_key(POOL_META_NAME),
+            "classification must not create pool metadata on non-elected nodes"
+        );
+    }
+
     #[tokio::test]
     async fn test_unproven_pending_identity_cannot_authorize_all_missing_pool_meta() {
         let deployment_id = Uuid::new_v4();

crates/ecstore/src/store/init_format.rs

@@ -14,6 +14,7 @@
 
 use crate::cluster::rpc::client::is_network_like_disk_error;
 use crate::config::storageclass;
+use crate::core::pools::PoolMetaBootstrapAuthority;
 use crate::disk::error_reduce::{count_errs, reduce_write_quorum_errs};
 use crate::disk::{self, DiskAPI};
 use crate::error::{Error, Result};
@@ -84,7 +85,7 @@ pub async fn connect_load_init_formats(
 
 pub(crate) struct LoadedFormat {
     pub(crate) format: FormatV3,
-    pub(crate) fresh_bootstrap_proven: bool,
+    pub(crate) pool_meta_bootstrap_authority: PoolMetaBootstrapAuthority,
 }
 
 pub(crate) async fn connect_load_init_formats_with_instance_ctx(
@@ -133,7 +134,7 @@ pub(crate) async fn connect_load_init_formats_with_instance_ctx(
                 retain_format_quorum_members(instance_ctx, disks, &format, &quorum_members, set_drive_count).await?;
                 return Ok(LoadedFormat {
                     format: *format,
-                    fresh_bootstrap_proven: false,
+                    pool_meta_bootstrap_authority: PoolMetaBootstrapAuthority::LegacyAdoption,
                 });
             }
             Ok(LegacyFormatOutcome::Incompatible) => {
@@ -153,7 +154,7 @@ pub(crate) async fn connect_load_init_formats_with_instance_ctx(
             let fm = init_format_erasure(instance_ctx, disks, set_count, set_drive_count, deployment_id).await?;
             return Ok(LoadedFormat {
                 format: fm,
-                fresh_bootstrap_proven: true,
+                pool_meta_bootstrap_authority: PoolMetaBootstrapAuthority::Fresh,
             });
         }
     }
@@ -182,10 +183,29 @@ pub(crate) async fn connect_load_init_formats_with_instance_ctx(
 
     Ok(LoadedFormat {
         format: fm,
-        fresh_bootstrap_proven: false,
+        pool_meta_bootstrap_authority: verified_legacy_adoption_source(disks, &formats, set_count, set_drive_count).await?,
     })
 }
 
+async fn verified_legacy_adoption_source(
+    disks: &[Option<DiskStore>],
+    rustfs_formats: &[Option<FormatV3>],
+    set_count: usize,
+    set_drive_count: usize,
+) -> Result<PoolMetaBootstrapAuthority> {
+    match try_migrate_format(disks, rustfs_formats, set_count, set_drive_count).await {
+        Ok(LegacyFormatOutcome::Migrated { .. }) => Ok(PoolMetaBootstrapAuthority::LegacyAdoption),
+        Ok(LegacyFormatOutcome::None | LegacyFormatOutcome::Incompatible) => Ok(PoolMetaBootstrapAuthority::None),
+        Err(err) => {
+            debug!(
+                error = %err,
+                "legacy adoption proof skipped because legacy format verification failed"
+            );
+            Ok(PoolMetaBootstrapAuthority::None)
+        }
+    }
+}
+
 async fn retain_format_quorum_members(
     instance_ctx: &Arc<InstanceContext>,
     disks: &mut [Option<DiskStore>],
@@ -1311,10 +1331,7 @@ mod tests {
         let loaded = connect_load_init_formats_with_instance_ctx(&current_ctx(), true, &mut disks, 1, 3, None)
             .await
             .expect("fresh disks should receive a storage format");
-        assert!(
-            loaded.fresh_bootstrap_proven,
-            "every configured disk explicitly reporting unformatted should establish fresh topology proof"
-        );
+        assert_eq!(loaded.pool_meta_bootstrap_authority, PoolMetaBootstrapAuthority::Fresh);
         let format = loaded.format;
 
         let (formats, errors) = load_format_erasure_all(&disks, false).await;
@@ -1358,12 +1375,11 @@ mod tests {
 
         let mut expected = legacy;
         expected.erasure.this = Uuid::nil();
-        assert_eq!(
-            connect_load_init_formats(true, &mut disks, 1, 3, None)
-                .await
-                .expect("compatible legacy format should migrate"),
-            expected
-        );
+        let loaded = connect_load_init_formats_with_instance_ctx(&current_ctx(), true, &mut disks, 1, 3, None)
+            .await
+            .expect("compatible legacy format should migrate");
+        assert_eq!(loaded.format, expected);
+        assert_eq!(loaded.pool_meta_bootstrap_authority, PoolMetaBootstrapAuthority::LegacyAdoption);
         let (formats, errors) = load_format_erasure_all(&disks, false).await;
         assert!(
             errors.iter().all(Option::is_none),
@@ -1378,6 +1394,51 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn compatible_single_drive_legacy_format_marks_adoption_proof() {
+        let (_temp_dir, mut disks) = local_disks(1).await;
+        let legacy = FormatV3::new(1, 1);
+        write_legacy_majority(&disks, &legacy).await;
+
+        let loaded = connect_load_init_formats_with_instance_ctx(&current_ctx(), true, &mut disks, 1, 1, None)
+            .await
+            .expect("single-drive MinIO format should migrate");
+
+        assert_eq!(loaded.pool_meta_bootstrap_authority, PoolMetaBootstrapAuthority::LegacyAdoption);
+    }
+
+    #[tokio::test]
+    async fn existing_migrated_format_keeps_legacy_adoption_proof() {
+        let (_temp_dir, mut disks) = local_disks(1).await;
+        let legacy = FormatV3::new(1, 1);
+        write_legacy_majority(&disks, &legacy).await;
+
+        connect_load_init_formats_with_instance_ctx(&current_ctx(), true, &mut disks, 1, 1, None)
+            .await
+            .expect("first run should migrate the MinIO format");
+        let loaded = connect_load_init_formats_with_instance_ctx(&current_ctx(), true, &mut disks, 1, 1, None)
+            .await
+            .expect("retry after a partial adoption should reload the migrated RustFS format");
+
+        assert_eq!(loaded.pool_meta_bootstrap_authority, PoolMetaBootstrapAuthority::LegacyAdoption);
+    }
+
+    #[tokio::test]
+    async fn existing_rustfs_format_without_legacy_source_is_not_legacy_adoption() {
+        let (_temp_dir, mut disks) = local_disks(1).await;
+        let mut format = FormatV3::new(1, 1);
+        format.erasure.this = format.erasure.sets[0][0];
+        save_format_file(&disks[0], &Some(format))
+            .await
+            .expect("existing RustFS format should be written");
+
+        let loaded = connect_load_init_formats_with_instance_ctx(&current_ctx(), true, &mut disks, 1, 1, None)
+            .await
+            .expect("existing RustFS format should load");
+
+        assert_eq!(loaded.pool_meta_bootstrap_authority, PoolMetaBootstrapAuthority::None);
+    }
+
     #[tokio::test]
     async fn compatible_legacy_format_migrates_when_the_file_is_missing() {
         let (_temp_dir, mut disks) = local_disks(3).await;

crates/ecstore/src/store/object.rs

@@ -2353,6 +2353,16 @@ impl ECStore {
         })
     }
 
+    pub(crate) fn is_equivalent_decommission_capacity_target(source: &ObjectInfo, target: &ObjectInfo) -> bool {
+        source.bucket == target.bucket
+            && source.name == decode_dir_object(&target.name)
+            && if source.delete_marker {
+                is_equivalent_data_movement_delete_marker(source, target)
+            } else {
+                crate::data_movement::is_equivalent_data_movement_object_identity(source, target, true, false)
+            }
+    }
+
     /// Captures Object Lock state once for a batch of PUTs to the same bucket.
     /// `handle_put_object` only reuses the token for the same store, bucket,
     /// bucket incarnation, and Object Lock configuration revision.
@@ -4440,7 +4450,9 @@ impl ECStore {
         }
 
         if should_delete_from_all_pools(&opts, errs.len()) {
-            let mut obj = self.delete_object_from_all_pools(bucket, object, &opts, errs).await?;
+            let mut obj = self
+                .delete_object_from_all_pools(bucket, object, &opts, &pinfo.object_info, errs)
+                .await?;
             obj.name = decode_dir_object(object);
             return Ok(obj);
         }
@@ -5823,6 +5835,36 @@ mod tests {
         );
     }
 
+    #[test]
+    fn exact_delete_capacity_target_requires_matching_namespace_and_object_identity() {
+        let source = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "directory/".to_string(),
+            version_id: Some(Uuid::from_u128(1)),
+            mod_time: Some(OffsetDateTime::UNIX_EPOCH),
+            size: 10,
+            etag: Some("source-etag".to_string()),
+            ..Default::default()
+        };
+        let target = ObjectInfo {
+            name: rustfs_utils::path::encode_dir_object(&source.name),
+            ..source.clone()
+        };
+        assert!(ECStore::is_equivalent_decommission_capacity_target(&source, &target));
+
+        let mismatched_identity = ObjectInfo {
+            etag: Some("different-etag".to_string()),
+            ..target.clone()
+        };
+        assert!(!ECStore::is_equivalent_decommission_capacity_target(&source, &mismatched_identity));
+
+        let wrong_bucket = ObjectInfo {
+            bucket: "other-bucket".to_string(),
+            ..target
+        };
+        assert!(!ECStore::is_equivalent_decommission_capacity_target(&source, &wrong_bucket));
+    }
+
     #[tokio::test]
     async fn generic_data_movement_put_rejects_transition_ownership_without_capability() {
         let (_dirs, set) = make_local_set_disks(4, 2).await;

crates/ecstore/src/store/rebalance.rs

@@ -939,8 +939,12 @@ impl ECStore {
         bucket: &str,
         object: &str,
         opts: &ObjectOptions,
+        exact: &ObjectInfo,
         errs: Vec<PoolErr>,
     ) -> Result<ObjectInfo> {
+        self.reconcile_decommission_capacity_before_exact_delete(bucket, object, opts, exact)
+            .await?;
+
         let mut results = Vec::with_capacity(errs.len());
 
         for pe in errs.iter() {

crates/heal-contracts/src/heal_channel.rs

@@ -347,6 +347,9 @@ pub struct HealChannelRequest {
     pub id: String,
     /// Disk ID for heal disk/erasure set task
     pub disk: Option<String>,
+    /// Exact endpoints of replacement disks for an automatic erasure-set
+    /// rebuild. An empty list retains the generic erasure-set heal behavior.
+    pub heal_endpoints: Vec<String>,
     /// Bucket name
     pub bucket: String,
     /// Object prefix (optional)
@@ -594,6 +597,7 @@ pub fn create_heal_request(
         timeout_seconds: None,
         source: HealRequestSource::Internal,
         disk: None,
+        heal_endpoints: Vec::new(),
     }
 }
 
@@ -634,12 +638,13 @@ pub fn create_heal_response(
     }
 }
 
-pub async fn send_heal_disk(set_disk_id: String, priority: Option<HealChannelPriority>) -> Result<(), String> {
-    let req = HealChannelRequest {
+fn create_auto_heal_disk_request(set_disk_id: String, priority: Option<HealChannelPriority>) -> HealChannelRequest {
+    HealChannelRequest {
         id: Uuid::new_v4().to_string(),
         bucket: "".to_string(),
         object_prefix: None,
         disk: Some(set_disk_id),
+        heal_endpoints: Vec::new(),
         object_version_id: None,
         force_start: false,
         priority: priority.unwrap_or(HealChannelPriority::Low),
@@ -654,8 +659,71 @@ pub async fn send_heal_disk(set_disk_id: String, priority: Option<HealChannelPri
         no_lock: None,
         timeout_seconds: None,
         source: HealRequestSource::AutoHeal,
-    };
-    send_heal_request(req).await
+    }
+}
+
+fn create_auto_replacement_disk_request(
+    pool_index: usize,
+    set_index: usize,
+    replacement_endpoint: String,
+    priority: Option<HealChannelPriority>,
+) -> HealChannelRequest {
+    let mut request = create_auto_heal_disk_request(format!("pool_{pool_index}_set_{set_index}"), priority);
+    request.heal_endpoints = vec![replacement_endpoint];
+    request.pool_index = Some(pool_index);
+    request.set_index = Some(set_index);
+    request
+}
+
+/// Submit the legacy generic erasure-set auto-heal request.
+pub async fn send_heal_disk(set_disk_id: String, priority: Option<HealChannelPriority>) -> Result<(), String> {
+    send_heal_request(create_auto_heal_disk_request(set_disk_id, priority)).await
+}
+
+/// Submit an automatic replacement heal for one known disk endpoint.
+///
+/// The endpoint makes the request eligible for the durable replacement intent
+/// and completion-proof path in the heal task.
+pub async fn send_heal_replacement_disk(
+    pool_index: usize,
+    set_index: usize,
+    replacement_endpoint: String,
+    priority: Option<HealChannelPriority>,
+) -> Result<(), String> {
+    send_heal_request(create_auto_replacement_disk_request(
+        pool_index,
+        set_index,
+        replacement_endpoint,
+        priority,
+    ))
+    .await
+}
+
+#[cfg(test)]
+mod auto_heal_disk_request_tests {
+    use super::*;
+
+    #[test]
+    fn replacement_disk_request_carries_its_exact_endpoint() {
+        let request =
+            create_auto_replacement_disk_request(2, 3, "http://node2:9000/drive3".to_string(), Some(HealChannelPriority::Normal));
+
+        assert_eq!(request.disk.as_deref(), Some("pool_2_set_3"));
+        assert_eq!(request.heal_endpoints, ["http://node2:9000/drive3"]);
+        assert_eq!(request.pool_index, Some(2));
+        assert_eq!(request.set_index, Some(3));
+        assert_eq!(request.source, HealRequestSource::AutoHeal);
+    }
+
+    #[test]
+    fn legacy_auto_heal_disk_request_has_no_replacement_endpoint() {
+        let request = create_auto_heal_disk_request("pool_2_set_3".to_string(), None);
+
+        assert!(request.heal_endpoints.is_empty());
+        assert_eq!(request.pool_index, None);
+        assert_eq!(request.set_index, None);
+        assert_eq!(request.source, HealRequestSource::AutoHeal);
+    }
 }
 
 #[cfg(test)]

crates/heal/src/heal/channel.rs

@@ -646,10 +646,12 @@ impl HealChannelProcessor {
     /// Convert channel request to heal request
     fn convert_to_heal_request(&self, request: HealChannelRequest) -> Result<HealRequest> {
         let recursive = request.recursive.unwrap_or(false);
+        let mut inferred_set_scope = None;
         let heal_type = if let Some(disk_id) = &request.disk {
             let set_disk_id = utils::normalize_set_disk_id(disk_id).ok_or_else(|| Error::InvalidHealType {
                 heal_type: format!("erasure-set({disk_id})"),
             })?;
+            inferred_set_scope = utils::parse_set_disk_id(&set_disk_id).ok();
             HealType::ErasureSet {
                 buckets: vec![],
                 set_disk_id,
@@ -712,13 +714,14 @@ impl HealChannelProcessor {
             dry_run: request.dry_run.unwrap_or(false),
             no_lock,
             timeout: request.timeout_seconds.map(std::time::Duration::from_secs),
-            pool_index: request.pool_index,
-            set_index: request.set_index,
+            pool_index: request.pool_index.or_else(|| inferred_set_scope.map(|(pool, _)| pool)),
+            set_index: request.set_index.or_else(|| inferred_set_scope.map(|(_, set)| set)),
         };
 
         let mut heal_request = HealRequest::new(heal_type, options, priority);
         heal_request.id = request.id;
         heal_request.source = request.source;
+        heal_request.heal_endpoints = request.heal_endpoints;
         // force_start controls admission/queue semantics only. Do not reinterpret it as
         // destructive heal options: admin clients commonly pass forceStart=true together
         // with remove=false, and turning that into remove_corrupted=true can delete the
@@ -906,6 +909,7 @@ mod tests {
             object_prefix: None,
             object_version_id: None,
             disk: None,
+            heal_endpoints: Vec::new(),
             priority: HealChannelPriority::Normal,
             scan_mode: None,
             remove_corrupted: None,
@@ -938,6 +942,7 @@ mod tests {
             object_prefix: None,
             object_version_id: None,
             disk: None,
+            heal_endpoints: Vec::new(),
             priority: HealChannelPriority::High,
             scan_mode: Some(HealScanMode::Normal),
             remove_corrupted: Some(false),
@@ -970,6 +975,7 @@ mod tests {
             object_prefix: Some("test-object".to_string()),
             object_version_id: None,
             disk: None,
+            heal_endpoints: Vec::new(),
             priority: HealChannelPriority::High,
             scan_mode: Some(HealScanMode::Deep),
             remove_corrupted: Some(true),
@@ -1023,6 +1029,7 @@ mod tests {
             object_prefix: Some("test-object".to_string()),
             object_version_id: None,
             disk: None,
+            heal_endpoints: Vec::new(),
             priority: HealChannelPriority::Low,
             scan_mode: None,
             remove_corrupted: None,
@@ -1060,6 +1067,7 @@ mod tests {
                 object_prefix: Some("test-object".to_string()),
                 object_version_id: None,
                 disk: None,
+                heal_endpoints: Vec::new(),
                 priority: HealChannelPriority::Normal,
                 scan_mode: None,
                 remove_corrupted: None,
@@ -1099,6 +1107,7 @@ mod tests {
                 object_prefix: Some("test-object".to_string()),
                 object_version_id: None,
                 disk: None,
+                heal_endpoints: Vec::new(),
                 priority: HealChannelPriority::Normal,
                 scan_mode: None,
                 remove_corrupted: None,
@@ -1131,6 +1140,7 @@ mod tests {
             object_prefix: Some("logs/".to_string()),
             object_version_id: None,
             disk: None,
+            heal_endpoints: Vec::new(),
             priority: HealChannelPriority::High,
             scan_mode: Some(HealScanMode::Normal),
             remove_corrupted: Some(false),
@@ -1166,7 +1176,10 @@ mod tests {
             object_prefix: None,
             object_version_id: None,
             disk: Some("pool_0_set_1".to_string()),
+            heal_endpoints: vec!["http://node0:9000/drive1".to_string()],
             priority: HealChannelPriority::Critical,
+            pool_index: Some(0),
+            set_index: Some(1),
             scan_mode: None,
             remove_corrupted: None,
             recreate_missing: None,
@@ -1175,15 +1188,16 @@ mod tests {
             dry_run: None,
             no_lock: None,
             timeout_seconds: None,
-            pool_index: None,
-            set_index: None,
             force_start: false,
-            source: HealRequestSource::Internal,
+            source: HealRequestSource::AutoHeal,
         };
 
         let heal_request = processor.convert_to_heal_request(channel_request).unwrap();
         assert!(matches!(heal_request.heal_type, HealType::ErasureSet { .. }));
         assert_eq!(heal_request.priority, HealPriority::Urgent);
+        assert_eq!(heal_request.heal_endpoints, ["http://node0:9000/drive1"]);
+        assert_eq!(heal_request.options.pool_index, Some(0));
+        assert_eq!(heal_request.options.set_index, Some(1));
     }
 
     #[tokio::test]
@@ -1197,6 +1211,7 @@ mod tests {
             object_prefix: None,
             object_version_id: None,
             disk: Some("invalid-disk-id".to_string()),
+            heal_endpoints: Vec::new(),
             priority: HealChannelPriority::Normal,
             scan_mode: None,
             remove_corrupted: None,
@@ -1235,6 +1250,7 @@ mod tests {
                 object_prefix: None,
                 object_version_id: None,
                 disk: None,
+                heal_endpoints: Vec::new(),
                 priority: channel_priority,
                 scan_mode: None,
                 remove_corrupted: None,
@@ -1266,6 +1282,7 @@ mod tests {
             object_prefix: None,
             object_version_id: None,
             disk: None,
+            heal_endpoints: Vec::new(),
             priority: HealChannelPriority::Normal,
             scan_mode: None,
             remove_corrupted: Some(false),
@@ -1299,6 +1316,7 @@ mod tests {
             object_prefix: Some("".to_string()), // Empty prefix should be treated as bucket heal
             object_version_id: None,
             disk: None,
+            heal_endpoints: Vec::new(),
             priority: HealChannelPriority::Normal,
             scan_mode: None,
             remove_corrupted: None,
@@ -1336,6 +1354,7 @@ mod tests {
             object_prefix: Some("object".to_string()),
             object_version_id: None,
             disk: None,
+            heal_endpoints: Vec::new(),
             priority: HealChannelPriority::Low,
             scan_mode: Some(HealScanMode::Normal),
             remove_corrupted: None,
@@ -1614,6 +1633,7 @@ mod tests {
             object_prefix: None,
             object_version_id: None,
             disk: Some("invalid".to_string()),
+            heal_endpoints: Vec::new(),
             priority: HealChannelPriority::Normal,
             scan_mode: None,
             remove_corrupted: None,

crates/heal/src/heal/mod.rs

@@ -26,12 +26,10 @@ pub mod utils;
 
 use storage_api::owner::{
     ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_RUSTFS_META_BUCKET,
-    EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskResult,
-    EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore, ObjectIO, ObjectOperations,
-    ecstore_local_disk_map_read,
+    EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskOption,
+    EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore, ObjectIO,
+    ObjectOperations, ecstore_local_disk_map_read, ecstore_new_disk,
 };
-#[cfg(test)]
-use storage_api::owner::{EcstoreDiskOption, ecstore_new_disk};
 
 pub use erasure_healer::ErasureSetHealer;
 pub use manager::{HealManager, HealOperationsSnapshot, HealPriorityCounts, HealSourceCounts};
@@ -247,10 +245,8 @@ pub(crate) async fn local_disk_map_read() -> tokio::sync::OwnedRwLockReadGuard<L
     ecstore_local_disk_map_read().await
 }
 
-#[cfg(test)]
 pub(crate) type DiskOption = EcstoreDiskOption;
 
-#[cfg(test)]
 pub(crate) async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> DiskResult<DiskStore> {
     ecstore_new_disk(ep, opt).await
 }

crates/heal/src/heal/replacement_readiness.rs

@@ -14,9 +14,9 @@
 
 use std::{fs, path::Path};
 
-#[cfg(test)]
-use super::Endpoint;
-use super::{DiskStore, HealDiskExt as _, local_disk_map_read, resume::ReplacementTargetIdentity};
+use super::{
+    DiskOption, DiskStore, Endpoint, HealDiskExt as _, local_disk_map_read, new_disk, resume::ReplacementTargetIdentity,
+};
 
 pub(crate) async fn auto_replacement_target_ready(disk: &DiskStore, local_disks: &[DiskStore]) -> bool {
     auto_replacement_target_identity(disk, local_disks).await.is_some()
@@ -72,8 +72,38 @@ pub(crate) async fn auto_replacement_target_identity(
     .flatten()
 }
 
-pub(crate) async fn auto_replacement_targets_ready(targets: &[String]) -> bool {
-    auto_replacement_target_identities(targets).await.is_some()
+fn local_replacement_endpoint(target: &str, local_grid_hosts: &[String]) -> Option<Endpoint> {
+    let mut endpoint = Endpoint::try_from(target).ok()?;
+    if endpoint.is_local {
+        return Some(endpoint);
+    }
+
+    let grid_host = endpoint.grid_host();
+    if grid_host.is_empty() || !local_grid_hosts.iter().any(|local_host| local_host == &grid_host) {
+        return None;
+    }
+
+    endpoint.is_local = true;
+    Some(endpoint)
+}
+
+async fn replacement_target_disk(target: &str, local_disks: &[DiskStore]) -> Option<DiskStore> {
+    if let Some(disk) = local_disks.iter().find(|disk| disk.endpoint().to_string() == target) {
+        return Some(disk.clone());
+    }
+
+    let local_grid_hosts = local_disks.iter().map(|disk| disk.endpoint().grid_host()).collect::<Vec<_>>();
+    let endpoint = local_replacement_endpoint(target, &local_grid_hosts)?;
+
+    new_disk(
+        &endpoint,
+        &DiskOption {
+            cleanup: false,
+            health_check: false,
+        },
+    )
+    .await
+    .ok()
 }
 
 pub(crate) async fn auto_replacement_target_identities(targets: &[String]) -> Option<Vec<ReplacementTargetIdentity>> {
@@ -88,8 +118,8 @@ pub(crate) async fn auto_replacement_target_identities(targets: &[String]) -> Op
 
     let mut identities = Vec::with_capacity(targets.len());
     for target in targets {
-        let disk = local_disks.iter().find(|disk| disk.endpoint().to_string() == *target)?;
-        identities.push(auto_replacement_target_identity(disk, &local_disks).await?);
+        let disk = replacement_target_disk(target, &local_disks).await?;
+        identities.push(auto_replacement_target_identity(&disk, &local_disks).await?);
     }
     identities.sort_by(|left, right| left.endpoint.cmp(&right.endpoint));
     identities.dedup_by(|left, right| left.endpoint == right.endpoint);
@@ -131,6 +161,29 @@ mod tests {
     use super::*;
     use tempfile::TempDir;
 
+    #[test]
+    fn local_replacement_endpoint_accepts_a_url_on_a_registered_local_grid_host() {
+        let local_grid_hosts = vec!["http://127.0.0.1:9000".to_owned()];
+        let endpoint = local_replacement_endpoint("http://127.0.0.1:9000/replacement", &local_grid_hosts)
+            .expect("matching local grid host should be accepted");
+
+        assert!(endpoint.is_local);
+    }
+
+    #[test]
+    fn local_replacement_endpoint_rejects_a_url_on_an_unregistered_grid_host() {
+        let local_grid_hosts = vec!["http://127.0.0.1:9000".to_owned()];
+
+        assert!(local_replacement_endpoint("http://127.0.0.1:9001/replacement", &local_grid_hosts).is_none());
+    }
+
+    #[test]
+    fn local_replacement_endpoint_keeps_a_local_path_local() {
+        let endpoint = local_replacement_endpoint("/replacement", &[]).expect("local path should be accepted");
+
+        assert!(endpoint.is_local);
+    }
+
     #[tokio::test]
     async fn runtime_environment_cannot_bypass_mount_admission() {
         temp_env::async_with_vars(

crates/heal/src/heal/storage.rs

@@ -394,11 +394,6 @@ pub trait HealStorageAPI: Send + Sync {
         Err(Error::other("target-scoped replacement format is unsupported"))
     }
 
-    /// Recheck admitted replacement targets immediately before destructive work.
-    async fn replacement_targets_ready(&self, _targets: &[String]) -> Result<bool> {
-        Ok(false)
-    }
-
     /// Read target-specific physical evidence for one replacement version.
     ///
     /// This is only used by automatic replacement healing after the normal
@@ -1171,10 +1166,6 @@ impl HealStorageAPI for ECStoreHealStorage {
             .map_err(Error::Storage)
     }
 
-    async fn replacement_targets_ready(&self, targets: &[String]) -> Result<bool> {
-        Ok(super::replacement_readiness::auto_replacement_targets_ready(targets).await)
-    }
-
     async fn replacement_targets_have_version(
         &self,
         bucket: &str,

crates/heal/src/heal/storage_api.rs

@@ -24,7 +24,6 @@ pub(crate) use rustfs_ecstore::api::disk::{
     DiskStore as EcstoreDiskStore, HEALING_MARKER_PATH as ECSTORE_HEALING_MARKER_PATH,
     RUSTFS_META_BUCKET as ECSTORE_RUSTFS_META_BUCKET,
 };
-#[cfg(test)]
 pub(crate) use rustfs_ecstore::api::disk::{DiskOption as EcstoreDiskOption, new_disk as ecstore_new_disk};
 pub(crate) use rustfs_ecstore::api::error::{Error as EcstoreErrorType, StorageError as EcstoreStorageError};
 pub(crate) use rustfs_ecstore::api::runtime::local_disk_map_read as ecstore_local_disk_map_read;
@@ -43,7 +42,6 @@ pub(crate) mod owner {
         EcstoreStorageError, EcstoreStore, ecstore_load_admin_data_usage_from_backend_cached, ecstore_local_disk_map_read,
     };
 
-    #[cfg(test)]
     pub(crate) use super::{EcstoreDiskOption, ecstore_new_disk};
 }
 

crates/heal/src/heal/task/heal_erasure_set.rs

@@ -93,16 +93,6 @@ impl HealTask {
             None
         };
 
-        if is_auto_replacement
-            && !self
-                .await_with_control(self.storage.replacement_targets_ready(&self.heal_endpoints))
-                .await?
-        {
-            return Err(Error::TaskExecutionFailed {
-                message: format!("Replacement target is no longer ready for automatic heal {set_disk_id}"),
-            });
-        }
-
         let replacement_resume_disk = if is_auto_replacement {
             Some(match replacement_resume_disk {
                 Some(disk) => disk,

crates/heal/src/heal/task/tests.rs

@@ -84,7 +84,7 @@ async fn automatic_replacement_uses_target_scoped_format() {
     let temp = TempDir::new().expect("temporary resume disk directory should be created");
     let disk = make_resume_disk(&temp).await;
     let storage = Arc::new(MockStorage {
-        replacement_targets_ready: Mutex::new(true),
+        replacement_target_identities_ready: Mutex::new(true),
         resume_disk: Mutex::new(Some(disk)),
         ..Default::default()
     });
@@ -123,7 +123,7 @@ async fn automatic_replacement_uses_target_scoped_format() {
 #[tokio::test]
 async fn automatic_replacement_persists_intent_before_format() {
     let storage = Arc::new(MockStorage {
-        replacement_targets_ready: Mutex::new(true),
+        replacement_target_identities_ready: Mutex::new(true),
         ..Default::default()
     });
     let mut request = HealRequest::new(
@@ -155,7 +155,7 @@ async fn automatic_replacement_persists_intent_before_format() {
 #[tokio::test]
 async fn recovered_replacement_never_uses_a_fresh_resume_disk() {
     let storage = Arc::new(MockStorage {
-        replacement_targets_ready: Mutex::new(true),
+        replacement_target_identities_ready: Mutex::new(true),
         ..Default::default()
     });
     let mut request = HealRequest::new(
@@ -193,7 +193,7 @@ async fn automatic_replacement_rejects_a_new_identity_after_format() {
     let first_identity = replacement_identity("replacement-a", "device-a", "filesystem-a");
     let second_identity = replacement_identity("replacement-a", "device-b", "filesystem-b");
     let storage = Arc::new(MockStorage {
-        replacement_targets_ready: Mutex::new(true),
+        replacement_target_identities_ready: Mutex::new(true),
         replacement_target_identity_sequences: Mutex::new(VecDeque::from([
             vec![first_identity.clone()],
             vec![first_identity.clone()],
@@ -259,7 +259,7 @@ async fn automatic_replacement_reuses_an_existing_non_target_resume_anchor() {
     .await
     .expect("existing intent should be stored on the non-target anchor");
     let storage = Arc::new(MockStorage {
-        replacement_targets_ready: Mutex::new(true),
+        replacement_target_identities_ready: Mutex::new(true),
         replacement_resume_disk: Mutex::new(Some(anchor.clone())),
         ..Default::default()
     });
@@ -493,7 +493,7 @@ async fn verified_recovery_keeps_state_when_marker_clear_fails() {
 
     let storage = Arc::new(MockStorage {
         replacement_resume_disk: Mutex::new(Some(anchor.clone())),
-        replacement_targets_ready: Mutex::new(true),
+        replacement_target_identities_ready: Mutex::new(true),
         ..Default::default()
     });
     let mut request = HealRequest::new(
@@ -551,7 +551,7 @@ struct MockStorage {
     format_error: Mutex<Option<Error>>,
     global_format_calls: Mutex<u32>,
     replacement_format_calls: Mutex<Vec<(usize, usize, Vec<String>)>>,
-    replacement_targets_ready: Mutex<bool>,
+    replacement_target_identities_ready: Mutex<bool>,
     replacement_target_identity_sequences: Mutex<VecDeque<Vec<crate::heal::resume::ReplacementTargetIdentity>>>,
     listed_prefixes: Mutex<Vec<String>>,
     truncate_without_token: Mutex<bool>,
@@ -943,10 +943,6 @@ impl HealStorageAPI for MockStorage {
         ))
     }
 
-    async fn replacement_targets_ready(&self, _targets: &[String]) -> Result<bool> {
-        Ok(*self.replacement_targets_ready.lock().unwrap())
-    }
-
     async fn list_objects_for_heal_page(
         &self,
         bucket: &str,
@@ -1028,7 +1024,7 @@ impl HealStorageAPI for MockStorage {
         &self,
         targets: &[String],
     ) -> Result<Vec<crate::heal::resume::ReplacementTargetIdentity>> {
-        if !*self.replacement_targets_ready.lock().unwrap() {
+        if !*self.replacement_target_identities_ready.lock().unwrap() {
             return Err(Error::other("replacement target is not ready"));
         }
         if let Some(identities) = self.replacement_target_identity_sequences.lock().unwrap().pop_front() {

crates/protos/src/heal_control.rs

@@ -88,6 +88,8 @@ impl From<Priority> for HealChannelPriority {
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct StartCommand {
     disk: Option<String>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    heal_endpoints: Vec<String>,
     bucket: String,
     object_prefix: Option<String>,
     object_version_id: Option<String>,
@@ -113,6 +115,7 @@ impl TryFrom<HealChannelRequest> for StartCommand {
     fn try_from(request: HealChannelRequest) -> Result<Self, Self::Error> {
         Ok(Self {
             disk: request.disk,
+            heal_endpoints: request.heal_endpoints,
             bucket: request.bucket,
             object_prefix: request.object_prefix,
             object_version_id: request.object_version_id,
@@ -146,6 +149,7 @@ impl StartCommand {
         Ok(HealChannelRequest {
             id: request_id,
             disk: self.disk,
+            heal_endpoints: self.heal_endpoints,
             bucket: self.bucket,
             object_prefix: self.object_prefix,
             object_version_id: self.object_version_id,
@@ -632,6 +636,12 @@ mod tests {
         }
     }
 
+    fn replacement_test_request(request_id: String) -> HealChannelRequest {
+        let mut request = test_request(request_id);
+        request.heal_endpoints = vec!["http://node1:9000/drive2".to_string()];
+        request
+    }
+
     fn metadata(byte: u8, epoch: u64) -> RequestMetadata {
         RequestMetadata::new([byte; 16], 1_000, 2_000, epoch)
     }
@@ -639,7 +649,7 @@ mod tests {
     #[test]
     fn round_trips_all_commands_and_results() {
         let request_id = uuid::Uuid::new_v4().to_string();
-        let start = Envelope::start(test_request(request_id), metadata(1, 7)).unwrap();
+        let start = Envelope::start(replacement_test_request(request_id), metadata(1, 7)).unwrap();
         let query = Envelope::query(
             uuid::Uuid::new_v4().to_string(),
             metadata(2, 7),

crates/replication/src/lib.rs

@@ -59,8 +59,9 @@ pub use mrf::{
     MrfV2Envelope, MrfV2Error, MrfV2Reader, MrfV2Readiness, decode_mrf_file, encode_mrf_file,
 };
 pub use multipart::{
-    ReplicationMultipartPartInput, ReplicationMultipartPartPlan, ReplicationMultipartPlanError, ReplicationMultipartRange,
-    replication_multipart_complete_actual_size, replication_multipart_part_plan,
+    REPLICATION_MAX_SINGLE_PUT_SIZE, ReplicationMultipartPartInput, ReplicationMultipartPartPlan, ReplicationMultipartPlanError,
+    ReplicationMultipartRange, replication_multipart_complete_actual_size, replication_multipart_part_plan,
+    replication_single_put_size_error,
 };
 pub use object::{
     ReplicationSourceObject, ReplicationTargetObject, SsecPassthroughCapability, SsecPassthroughGate, content_matches_by_etag,

crates/replication/src/multipart.rs

@@ -109,15 +109,49 @@ pub fn replication_multipart_complete_actual_size(user_defined: &HashMap<String,
     get_internal_metadata(user_defined, SUFFIX_ACTUAL_SIZE).unwrap_or_default()
 }
 
+/// Largest body S3 accepts on a single `PutObject`. Anything above this has to
+/// be uploaded as multipart; the limit is part of the S3 API, not a RustFS
+/// tunable, so every generic S3 target enforces it.
+pub const REPLICATION_MAX_SINGLE_PUT_SIZE: i64 = 5 * 1024 * 1024 * 1024;
+
+/// Reject a single-`PutObject` replication transfer the target can never accept.
+///
+/// Replication mirrors the object's *source-side storage shape*: an object
+/// written to the source with one `PutObject` replicates with one `PutObject`
+/// whatever its size, and a multipart object replays the source's own part
+/// layout. So a source object larger than [`REPLICATION_MAX_SINGLE_PUT_SIZE`]
+/// that was not written as multipart can never reach a generic S3 target — the
+/// remote rejects it with `EntityTooLarge`, but only after the whole body has
+/// been streamed to it (rustfs#6825).
+///
+/// Returning the failure up front turns an unbounded wasted transfer plus an
+/// opaque remote error into a stated, diagnosable limit. RustFS deliberately
+/// does not re-chunk such an object into multipart on the replication side:
+/// the target's part layout is the source's, and rewriting it would break the
+/// ETag/part identity that heal and delete convergence address.
+pub fn replication_single_put_size_error(is_multipart: bool, transfer_size: i64) -> Option<String> {
+    if is_multipart || transfer_size <= REPLICATION_MAX_SINGLE_PUT_SIZE {
+        return None;
+    }
+    Some(format!(
+        "object of {transfer_size} bytes was not written as multipart on the source and exceeds the \
+         {REPLICATION_MAX_SINGLE_PUT_SIZE} byte single-PutObject limit of an S3 target; \
+         re-upload it with multipart to make it replicable"
+    ))
+}
+
 #[cfg(test)]
 mod tests {
     use super::{
-        ReplicationMultipartPartInput, ReplicationMultipartPartPlan, ReplicationMultipartPlanError, ReplicationMultipartRange,
-        replication_multipart_complete_actual_size, replication_multipart_part_plan,
+        REPLICATION_MAX_SINGLE_PUT_SIZE, ReplicationMultipartPartInput, ReplicationMultipartPartPlan,
+        ReplicationMultipartPlanError, ReplicationMultipartRange, replication_multipart_complete_actual_size,
+        replication_multipart_part_plan, replication_single_put_size_error,
     };
     use crate::http::{SUFFIX_ACTUAL_SIZE, insert_internal_metadata};
     use std::collections::HashMap;
 
+    const MIB: i64 = 1024 * 1024;
+
     #[test]
     fn multipart_part_plan_builds_range_and_next_offset() {
         assert_eq!(
@@ -219,4 +253,44 @@ mod tests {
         assert_eq!(replication_multipart_complete_actual_size(&user_defined), "123");
         assert!(replication_multipart_complete_actual_size(&HashMap::new()).is_empty());
     }
+
+    #[test]
+    fn single_put_size_guard_admits_transfers_a_target_can_accept() {
+        for size in [
+            0,
+            1,
+            MIB,
+            REPLICATION_MAX_SINGLE_PUT_SIZE - 1,
+            REPLICATION_MAX_SINGLE_PUT_SIZE,
+        ] {
+            assert_eq!(
+                replication_single_put_size_error(false, size),
+                None,
+                "single PUT of {size} bytes is within the S3 limit and must not be rejected"
+            );
+        }
+    }
+
+    #[test]
+    fn single_put_size_guard_rejects_an_oversized_single_put() {
+        let size = REPLICATION_MAX_SINGLE_PUT_SIZE + 1;
+        let err = replication_single_put_size_error(false, size).expect("oversized single PUT must be rejected");
+
+        // The message is the operator's diagnosis: it has to name the actual
+        // size, the limit, and the reason the object is on this route at all.
+        assert!(err.contains(&size.to_string()), "message must name the object size: {err}");
+        assert!(
+            err.contains(&REPLICATION_MAX_SINGLE_PUT_SIZE.to_string()),
+            "message must name the limit: {err}"
+        );
+        assert!(err.contains("multipart"), "message must name the remedy: {err}");
+    }
+
+    #[test]
+    fn single_put_size_guard_never_rejects_the_multipart_route() {
+        // Multipart replays the source part layout, so object size alone says
+        // nothing about whether the target will accept it; the per-part limits
+        // are the target's to enforce.
+        assert_eq!(replication_single_put_size_error(true, REPLICATION_MAX_SINGLE_PUT_SIZE * 1024), None);
+    }
 }

crates/utils/src/http/metadata_compat.rs

@@ -50,6 +50,11 @@ pub const SUFFIX_ACTUAL_OBJECT_SIZE: &str = "actual-object-size";
 /// Used by replication; key stored with capital A
 pub const SUFFIX_ACTUAL_OBJECT_SIZE_CAP: &str = "Actual-Object-Size";
 pub const SUFFIX_CRC: &str = "crc";
+/// Marks checksum bytes produced by RustFS as plaintext checksum metadata.
+///
+/// MinIO encrypts the same on-disk field for SSE objects, so readers must not
+/// decode encrypted-object checksums unless this marker is present.
+pub const SUFFIX_PLAINTEXT_CHECKSUM: &str = "plaintext-checksum";
 /// JSON-encoded per-part S3 checksum maps retained across raw data movement.
 pub const SUFFIX_PART_CHECKSUMS: &str = "part-checksums";
 pub const SUFFIX_TRANSITION_STATUS: &str = "transition-status";

docs/operations/replication-object-size-limits.md

@@ -0,0 +1,101 @@
+# Replication object size and shape limits (generic S3 targets)
+
+What RustFS can and cannot replicate to a generic S3 target (AWS S3, Wasabi,
+MinIO, or any other S3-compatible endpoint configured as a bucket replication
+target), and how a rejected object shows up in the log.
+
+## The route is chosen by the source object's shape, not its size
+
+RustFS mirrors how the object was written on the source:
+
+| Source object was written as | Replication transport |
+| --- | --- |
+| a single `PutObject` | a single `PutObject` on the target |
+| a multipart upload | a multipart upload replaying **the source's own part layout** |
+
+RustFS does not re-chunk on the replication side. A single-`PutObject` object is
+never converted into a multipart upload for the target, and a multipart object's
+parts are never merged or re-split. The target's part layout is the source's,
+because heal and delete convergence address the replica by that identity.
+
+This is why object size alone does not tell you whether an object is
+replicable — how it was uploaded does.
+
+## Limits
+
+### Single-`PutObject` objects: 5 GiB
+
+S3 caps `PutObject` at **5 GiB**. This is an S3 API limit that every target
+enforces, not a RustFS tunable.
+
+An object larger than 5 GiB that was written to the source with a single
+`PutObject` therefore **cannot be replicated to a generic S3 target**. RustFS
+detects this before streaming the body and fails the object immediately, rather
+than uploading gigabytes only to collect an `EntityTooLarge` from the remote.
+
+**Remedy:** re-upload the object using multipart. Most S3 clients do this
+automatically above a threshold (the AWS CLI defaults to 8 MiB); a client
+configured with a very high multipart threshold, or one that streams a single
+`PutObject`, is the usual way an object ends up on the wrong side of this limit.
+
+### Multipart objects: the target's multipart limits, applied to the source's layout
+
+Because the source's part layout is replayed verbatim, the target's own
+multipart constraints apply to that layout:
+
+| Constraint | Target rejects with |
+| --- | --- |
+| every part except the last must be ≥ 5 MiB | `EntityTooSmall` |
+| no part may exceed 5 GiB | `EntityTooLarge` |
+| at most 10,000 parts | failure at `CompleteMultipartUpload` |
+
+A source object whose parts satisfy these is replicable up to the S3 multipart
+maximum of 5 TiB.
+
+## Reliability characteristics for large objects
+
+Worth knowing before replicating multi-gigabyte objects:
+
+- Parts are transferred **sequentially**.
+- There is **no part-level retry**. A failure on any single part fails the whole
+  object; the target-side multipart upload is then aborted so no incomplete
+  upload is left behind.
+- Retry happens at the object level (MRF replay / heal scanner), so a failure
+  late in a large transfer re-sends the object from the beginning.
+
+For a 6 GiB object this means one long all-or-nothing transfer window. Part-level
+retry and resumable transfer are tracked as a separate improvement.
+
+## What a failed object looks like in the log
+
+A replication attempt that ends in a terminal `FAILED` state emits one `error`
+line per failed target. It is at `error` deliberately: the default log level
+(`RUSTFS_OBS_LOGGER_LEVEL`, default `error`) must not hide an object that never
+reached its target.
+
+```
+ERROR ... event=replication_object_failed bucket=photos object=backups/vm-image.qcow2
+      version_id=... arn=arn:replication::wasabi endpoint=s3.wasabisys.com
+      op_type=OBJECT size=6442450944 replication_status=FAILED
+      error="object of 6442450944 bytes was not written as multipart on the source and
+             exceeds the 5368709120 byte single-PutObject limit of an S3 target;
+             re-upload it with multipart to make it replicable"
+      Replication failed for object
+```
+
+The `error` field carries the target's own error code and message where the
+target produced one, so a remote rejection is diagnosable without lowering the
+log level and reproducing. It is passed through the same redaction as the
+persisted resync detail, so an error echoing a credential or signed URL is
+replaced with `[redacted sensitive resync error detail]`.
+
+Raise `RUSTFS_OBS_LOGGER_LEVEL` to `warn` to additionally see the per-attempt
+failure branches (target offline, HEAD failures, per-part errors) that sit
+underneath this summary.
+
+## Related
+
+- [Replication target check](replication-check.md) — validate a target's
+  configuration, versioning, and version fidelity before relying on it.
+- [Presigned PUT size limit](presigned-put-size-limit.md)
+- [Presigned multipart size limit](presigned-multipart-size-limit.md)

rustfs/src/admin/handlers/site_replication.rs

@@ -35,6 +35,7 @@ use crate::admin::storage_api::bucket::utils::{deserialize, serialize};
 use crate::admin::storage_api::bucket::{
     AdminObjectLockConfigExt as _, AdminReplicationConfigExt as _, AdminVersioningConfigExt as _,
 };
+use crate::admin::storage_api::bucket_target_sys::BucketTargetSys;
 use crate::admin::storage_api::contract::bucket::{
     BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions, SRBucketDeleteOp,
 };
@@ -61,7 +62,7 @@ use rustfs_iam::sys::{
 };
 use rustfs_madmin::{
     BucketBandwidth, GroupStatus, IDPSettings, InProgressMetric, InQueueMetric, LDAPConfigSettings, LDAPSettings,
-    OpenIDProviderSettings, PeerInfo, PeerSite, QStat, ReplProxyMetric, ReplicateAddStatus, ReplicateEditStatus,
+    OpenIDProviderSettings, OpenIDSettings, PeerInfo, PeerSite, QStat, ReplProxyMetric, ReplicateAddStatus, ReplicateEditStatus,
     ReplicateRemoveStatus, ResyncBucketStatus, SITE_REPL_API_VERSION, SR_IAM_ITEM_STS_ACC, SR_IAM_ITEM_STS_ACC_LEGACY,
     SRBucketInfo, SRBucketMeta, SRBucketStatsSummary, SRGroupInfo, SRGroupStatsSummary, SRIAMItem, SRIAMUser,
     SRILMExpiryStatsSummary, SRInfo, SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq, SRPendingOperation, SRPolicyMapping,
@@ -87,6 +88,7 @@ use std::time::Duration;
 use time::OffsetDateTime;
 use tokio::sync::Mutex;
 use tracing::{info, warn};
+use url::Url;
 use url::form_urlencoded;
 use uuid::Uuid;
 
@@ -1079,6 +1081,62 @@ async fn add_preflight_infos(
     .collect()
 }
 
+/// The first field at which a peer's reported IDP settings diverge from the
+/// local ones, named so an operator can act on the rejection instead of
+/// re-deriving the comparison (rustfs#7003). Values are reported except for
+/// credential-derived leaves, which are named but never echoed.
+fn idp_settings_difference(local: &serde_json::Value, peer: &serde_json::Value) -> Option<String> {
+    // Only scalars are echoed: a nested object may carry credential-derived
+    // leaves whose own field name never reaches `path`.
+    fn scalar(value: &serde_json::Value) -> Option<String> {
+        match value {
+            serde_json::Value::String(text) => Some(text.clone()),
+            serde_json::Value::Object(_) | serde_json::Value::Array(_) => None,
+            other => Some(other.to_string()),
+        }
+    }
+
+    fn presence(value: &serde_json::Value) -> &'static str {
+        if value.is_null() { "absent" } else { "set" }
+    }
+
+    fn walk(path: &str, local: &serde_json::Value, peer: &serde_json::Value) -> Option<String> {
+        if local == peer {
+            return None;
+        }
+
+        if let (serde_json::Value::Object(local_fields), serde_json::Value::Object(peer_fields)) = (local, peer) {
+            let keys: BTreeSet<&String> = local_fields.keys().chain(peer_fields.keys()).collect();
+            for key in keys {
+                let child_path = if path.is_empty() {
+                    key.to_string()
+                } else {
+                    format!("{path}.{key}")
+                };
+                let difference = walk(
+                    &child_path,
+                    local_fields.get(key).unwrap_or(&serde_json::Value::Null),
+                    peer_fields.get(key).unwrap_or(&serde_json::Value::Null),
+                );
+                if difference.is_some() {
+                    return difference;
+                }
+            }
+        }
+
+        let field = if path.is_empty() { "<root>" } else { path };
+        if field.to_ascii_lowercase().contains("secret") {
+            return Some(format!("`{field}` differs (values redacted)"));
+        }
+        match (scalar(local), scalar(peer)) {
+            (Some(local), Some(peer)) => Some(format!("`{field}` differs (local `{local}`, peer `{peer}`)")),
+            _ => Some(format!("`{field}` differs (local {}, peer {})", presence(local), presence(peer))),
+        }
+    }
+
+    walk("", local, peer)
+}
+
 fn validate_add_preflight_topology(infos: &[SiteReplicationAddPreflightInfo], local_peer: &PeerInfo) -> S3Result<()> {
     let mut deployment_ids = HashSet::new();
     let mut local_seen = false;
@@ -1121,8 +1179,12 @@ fn validate_add_preflight_topology(infos: &[SiteReplicationAddPreflightInfo], lo
         ));
     };
     for info in infos {
-        if &info.idp_settings != local_idp {
-            return Err(s3_error!(InvalidRequest, "IDP settings mismatch for site `{}`", info.endpoint));
+        if let Some(difference) = idp_settings_difference(local_idp, &info.idp_settings) {
+            return Err(s3_error!(
+                InvalidRequest,
+                "IDP settings mismatch for site `{}`: {difference}",
+                info.endpoint
+            ));
         }
     }
 
@@ -1916,36 +1978,64 @@ async fn bootstrap_existing_metadata_after_add(
     errors
 }
 
+/// The OpenID half of the IDP settings a site reports to its site-replication
+/// peers, built from the providers this site has configured (in
+/// `list_providers` order) and the site's own region.
+fn open_id_settings(providers: Vec<(String, OpenIDProviderSettings)>, region: String) -> OpenIDSettings {
+    // `region` qualifies the provider identities, not the site: a site
+    // without OpenID must report the empty settings in every region, or the
+    // peer comparison in `validate_add_preflight_topology` would reject two
+    // sites whose IDP configuration is identically absent (rustfs#7003).
+    if providers.is_empty() {
+        return OpenIDSettings::default();
+    }
+
+    let mut settings = OpenIDSettings {
+        enabled: true,
+        region,
+        ..Default::default()
+    };
+
+    for (provider_id, provider_settings) in providers {
+        let claim_provider_unset = settings.claim_provider.client_id.is_empty()
+            && settings.claim_provider.claim_name.is_empty()
+            && settings.claim_provider.role_policy.is_empty()
+            && settings.claim_provider.hashed_client_secret.is_empty();
+
+        if provider_id == "default" || claim_provider_unset {
+            settings.claim_provider = provider_settings;
+        } else {
+            settings.roles.insert(provider_id, provider_settings);
+        }
+    }
+
+    settings
+}
+
 fn local_idp_settings() -> IDPSettings {
     let mut settings = IDPSettings::default();
     if let Some(federation) = current_federated_identity_service() {
-        let providers = federation.list_providers();
-        settings.open_id.enabled = !providers.is_empty();
-        settings.open_id.region = current_region().map(|region| region.to_string()).unwrap_or_default();
-
-        for provider in providers {
-            let Some(config) = federation.get_provider_config(&provider.provider_id) else {
-                continue;
-            };
-            let provider_settings = OpenIDProviderSettings {
-                claim_name: config.claim_name.clone(),
-                claim_userinfo_enabled: false,
-                role_policy: config.role_policy.clone(),
-                client_id: config.client_id.clone(),
-                hashed_client_secret: hash_client_secret(config.client_secret.as_deref()),
-            };
-
-            let claim_provider_unset = settings.open_id.claim_provider.client_id.is_empty()
-                && settings.open_id.claim_provider.claim_name.is_empty()
-                && settings.open_id.claim_provider.role_policy.is_empty()
-                && settings.open_id.claim_provider.hashed_client_secret.is_empty();
-
-            if provider.provider_id == "default" || claim_provider_unset {
-                settings.open_id.claim_provider = provider_settings.clone();
-            } else {
-                settings.open_id.roles.insert(provider.provider_id.clone(), provider_settings);
-            }
-        }
+        // A listed provider whose config cannot be resolved contributes
+        // nothing to the peer comparison, so it is dropped here rather than
+        // inside the settings builder.
+        let providers = federation
+            .list_providers()
+            .into_iter()
+            .filter_map(|provider| {
+                let config = federation.get_provider_config(&provider.provider_id)?;
+                Some((
+                    provider.provider_id.clone(),
+                    OpenIDProviderSettings {
+                        claim_name: config.claim_name.clone(),
+                        claim_userinfo_enabled: false,
+                        role_policy: config.role_policy.clone(),
+                        client_id: config.client_id.clone(),
+                        hashed_client_secret: hash_client_secret(config.client_secret.as_deref()),
+                    },
+                ))
+            })
+            .collect();
+        settings.open_id = open_id_settings(providers, current_region().map(|region| region.to_string()).unwrap_or_default());
     }
 
     let (ldap, ldap_configs) = load_ldap_idp_settings();
@@ -2004,25 +2094,111 @@ fn filter_sr_info(mut info: SRInfo, opts: &SRStatusOptions) -> SRInfo {
     info
 }
 
-async fn build_metrics_summary(local_peer: &PeerInfo) -> SRMetricsSummary {
+/// Resolve one peer's downtime and last-seen from the replication heartbeat.
+///
+/// The heartbeat already tracks every remote endpoint it replicates to
+/// (`EpHealth`, refreshed every `RUSTFS_REPL_HEALTH_CHECK_INTERVAL_MS`), so
+/// these come from real observations instead of being synthesized per request.
+/// `reachable` is the caller's live probe for this status call and wins over
+/// the heartbeat's cached flag, which can lag by up to one probe interval.
+///
+/// An endpoint the heartbeat has never seen (no replication target points at
+/// it yet) yields zero downtime rather than a fabricated outage.
+async fn peer_health_metric(endpoint: &str, reachable: bool) -> (i64, Option<OffsetDateTime>) {
+    let fallback = (0, reachable.then(OffsetDateTime::now_utc));
+
+    let Ok(url) = Url::parse(endpoint) else {
+        return fallback;
+    };
+    let Some(health) = BucketTargetSys::get().endpoint_health(&url).await else {
+        return fallback;
+    };
+
+    let total_downtime_ns = i64::try_from(health.offline_duration.as_nanos()).unwrap_or(i64::MAX);
+    let last_online = if reachable {
+        Some(OffsetDateTime::now_utc())
+    } else {
+        health.last_online
+    };
+
+    (total_downtime_ns, last_online)
+}
+
+/// Assemble one peer's metric entry from already-resolved inputs.
+///
+/// Split out of [`build_metrics_summary`] so the local/remote and
+/// reachable/unreachable matrix stays unit-testable without a live replication
+/// stats handle or a populated heartbeat map.
+fn peer_metric_entry(
+    deployment_id: &str,
+    endpoint: &str,
+    is_local: bool,
+    reachable: bool,
+    (total_downtime_ns, last_online): (i64, Option<OffsetDateTime>),
+    local_counters: (i64, i64),
+) -> SRMetric {
+    let (replica_size, replica_count) = local_counters;
+
+    SRMetric {
+        deployment_id: deployment_id.to_string(),
+        endpoint: endpoint.to_string(),
+        online: reachable,
+        total_downtime_ns,
+        last_online,
+        // Replication counters are node-local: they describe traffic this node
+        // handled, so they belong only on the local entry. Copying them onto
+        // remote entries would double-count them cluster-wide.
+        replicated_size: if is_local { replica_size } else { 0 },
+        replicated_count: if is_local { replica_count } else { 0 },
+        ..Default::default()
+    }
+}
+
+async fn build_metrics_summary(
+    local_peer: &PeerInfo,
+    peers: &BTreeMap<String, PeerInfo>,
+    reachable_peers: &HashSet<String>,
+) -> SRMetricsSummary {
     let Some(stats) = current_replication_stats_handle() else {
         return SRMetricsSummary::default();
     };
 
     let node = stats.site_metrics_snapshot().await;
     let mut metrics = BTreeMap::new();
-    metrics.insert(
-        local_peer.deployment_id.clone(),
-        SRMetric {
-            deployment_id: local_peer.deployment_id.clone(),
-            endpoint: local_peer.endpoint.clone(),
-            online: true,
-            replicated_size: node.replica_size,
-            replicated_count: node.replica_count,
-            last_online: Some(OffsetDateTime::now_utc()),
-            ..Default::default()
-        },
-    );
+
+    // Emit an entry for every peer, not just the local one. An operator reading
+    // this page needs to know whether the *remote* site is reachable; reporting
+    // only "I am online" made a peer outage invisible here, so a site could be
+    // down for minutes while the status page stayed green.
+    for (deployment_id, peer) in peers {
+        let is_local = deployment_id == &local_peer.deployment_id;
+        let reachable = is_local || reachable_peers.contains(deployment_id);
+        let health = peer_health_metric(&peer.endpoint, reachable).await;
+
+        metrics.insert(
+            deployment_id.clone(),
+            peer_metric_entry(
+                deployment_id,
+                &peer.endpoint,
+                is_local,
+                reachable,
+                health,
+                (node.replica_size, node.replica_count),
+            ),
+        );
+    }
+
+    // A peer map that somehow omits the local deployment must still report the
+    // local counters rather than silently dropping them.
+    metrics.entry(local_peer.deployment_id.clone()).or_insert_with(|| SRMetric {
+        deployment_id: local_peer.deployment_id.clone(),
+        endpoint: local_peer.endpoint.clone(),
+        online: true,
+        last_online: Some(OffsetDateTime::now_utc()),
+        replicated_size: node.replica_size,
+        replicated_count: node.replica_count,
+        ..Default::default()
+    });
 
     SRMetricsSummary {
         active_workers: WorkerStat {
@@ -2669,7 +2845,7 @@ async fn build_status_info(state: &SiteReplicationState, local_peer: &PeerInfo,
     }
 
     if metrics_requested {
-        status.metrics = build_metrics_summary(local_peer).await;
+        status.metrics = build_metrics_summary(local_peer, &state.peers, &reachable_peers).await;
     }
 
     if opts.peer_state {
@@ -7580,6 +7756,47 @@ mod tests {
     use super::*;
     use crate::site_replication::identity::deployment_id_for_endpoint;
 
+    /// A peer the status probe could not reach must render as offline.
+    ///
+    /// Regression: `build_metrics_summary` used to hardcode `online: true` and
+    /// only ever emitted the local deployment, so a peer outage was invisible
+    /// on the status page — `Link: ● online` stayed green through a multi-minute
+    /// outage while replication was failing.
+    #[test]
+    fn peer_metric_entry_reports_unreachable_peer_as_offline() {
+        let heartbeat_last_online = OffsetDateTime::UNIX_EPOCH;
+        let entry = peer_metric_entry(
+            "remote-deployment",
+            "http://remote.example:9000",
+            false,
+            false,
+            (5_000_000_000, Some(heartbeat_last_online)),
+            (4096, 8),
+        );
+
+        assert!(!entry.online, "an unreachable peer must not be reported online");
+        assert_eq!(entry.total_downtime_ns, 5_000_000_000);
+        assert_eq!(
+            entry.last_online,
+            Some(heartbeat_last_online),
+            "an offline peer keeps the heartbeat's last-seen instead of being stamped 'now'"
+        );
+    }
+
+    /// Node-local replication counters must not be copied onto remote entries,
+    /// or a two-site cluster would double-count its own traffic.
+    #[test]
+    fn peer_metric_entry_keeps_replication_counters_on_the_local_entry() {
+        let local = peer_metric_entry("local", "http://local.example:9000", true, true, (0, None), (4096, 8));
+        let remote = peer_metric_entry("remote", "http://remote.example:9000", false, true, (0, None), (4096, 8));
+
+        assert_eq!(local.replicated_size, 4096);
+        assert_eq!(local.replicated_count, 8);
+        assert_eq!(remote.replicated_size, 0);
+        assert_eq!(remote.replicated_count, 0);
+        assert!(remote.online, "a reachable remote peer is still online");
+    }
+
     #[test]
     fn test_rotation_secret_candidates_try_the_installed_secret_before_the_new_one() {
         let mut pending = PendingRotation {
@@ -9636,6 +9853,119 @@ mod tests {
         assert!(err.to_string().contains("must include the local deployment"));
     }
 
+    fn openid_provider(client_id: &str) -> OpenIDProviderSettings {
+        OpenIDProviderSettings {
+            claim_name: "groups".to_string(),
+            claim_userinfo_enabled: false,
+            role_policy: "readwrite".to_string(),
+            client_id: client_id.to_string(),
+            hashed_client_secret: "hashed".to_string(),
+        }
+    }
+
+    // rustfs#7003: the site region is per-site and must not reach the peer
+    // IDP comparison while OpenID is unconfigured, or two sites in different
+    // regions can never be paired even with an identical (empty) IDP config.
+    #[test]
+    fn test_open_id_settings_omits_region_without_providers() {
+        let settings = open_id_settings(Vec::new(), "eu-site-1".to_string());
+
+        assert!(!settings.enabled);
+        assert_eq!(settings.region, "");
+        assert!(settings.roles.is_empty());
+        assert_eq!(
+            serde_json::to_value(&settings).expect("serialize OpenID settings"),
+            serde_json::to_value(OpenIDSettings::default()).expect("serialize default OpenID settings"),
+            "a site without OpenID providers must report the same settings in every region"
+        );
+    }
+
+    #[test]
+    fn test_open_id_settings_keeps_region_with_providers() {
+        let settings = open_id_settings(vec![("default".to_string(), openid_provider("client-a"))], "eu-site-1".to_string());
+
+        assert!(settings.enabled);
+        assert_eq!(settings.region, "eu-site-1");
+        assert_eq!(settings.claim_provider.client_id, "client-a");
+    }
+
+    // A whole provider object present on one side only must not spill its
+    // credential-derived leaves into the rejection message.
+    #[test]
+    fn test_validate_add_preflight_topology_omits_nested_values_in_idp_diff() {
+        let local_peer = PeerInfo {
+            deployment_id: "local-dep".to_string(),
+            ..peer("local", "https://local.example.com")
+        };
+        let mut local = preflight_site("local", "https://local.example.com", "local-dep", 0);
+        local.idp_settings = serde_json::json!({
+            "OpenID": {"Enabled": true, "ClaimProvider": {"ClientID": "client-a", "HashedClientSecret": "local-hash"}},
+        });
+        let mut remote = preflight_site("remote", "https://remote.example.com", "remote-dep", 0);
+        remote.idp_settings = serde_json::json!({"OpenID": {"Enabled": true}});
+        let infos = vec![local, remote];
+
+        let err = validate_add_preflight_topology(&infos, &local_peer).expect_err("IDP mismatch should fail");
+        let message = err.to_string();
+
+        assert!(message.contains("OpenID.ClaimProvider"), "{message}");
+        assert!(!message.contains("local-hash"), "{message}");
+    }
+
+    // rustfs#7003: a bare "IDP settings mismatch" leaves the operator with no
+    // way to tell which element diverged.
+    #[test]
+    fn test_validate_add_preflight_topology_reports_differing_idp_field() {
+        let local_peer = PeerInfo {
+            deployment_id: "local-dep".to_string(),
+            ..peer("local", "https://local.example.com")
+        };
+        let mut local = preflight_site("local", "https://local.example.com", "local-dep", 0);
+        local.idp_settings = serde_json::json!({
+            "LDAP": {"IsLDAPEnabled": false},
+            "OpenID": {"Enabled": false, "Region": "eu-site-1"},
+        });
+        let mut remote = preflight_site("remote", "https://remote.example.com", "remote-dep", 0);
+        remote.idp_settings = serde_json::json!({
+            "LDAP": {"IsLDAPEnabled": false},
+            "OpenID": {"Enabled": false, "Region": "eu-site-2"},
+        });
+        let infos = vec![local, remote];
+
+        let err = validate_add_preflight_topology(&infos, &local_peer).expect_err("IDP mismatch should fail");
+        let message = err.to_string();
+
+        assert!(message.contains("OpenID.Region"), "{message}");
+        assert!(message.contains("eu-site-1"), "{message}");
+        assert!(message.contains("eu-site-2"), "{message}");
+    }
+
+    // Hashed client secrets are still credential-derived material: name the
+    // field that diverged, never its value.
+    #[test]
+    fn test_validate_add_preflight_topology_redacts_secret_values_in_idp_diff() {
+        let local_peer = PeerInfo {
+            deployment_id: "local-dep".to_string(),
+            ..peer("local", "https://local.example.com")
+        };
+        let mut local = preflight_site("local", "https://local.example.com", "local-dep", 0);
+        local.idp_settings = serde_json::json!({
+            "OpenID": {"ClaimProvider": {"HashedClientSecret": "local-hash"}},
+        });
+        let mut remote = preflight_site("remote", "https://remote.example.com", "remote-dep", 0);
+        remote.idp_settings = serde_json::json!({
+            "OpenID": {"ClaimProvider": {"HashedClientSecret": "remote-hash"}},
+        });
+        let infos = vec![local, remote];
+
+        let err = validate_add_preflight_topology(&infos, &local_peer).expect_err("IDP mismatch should fail");
+        let message = err.to_string();
+
+        assert!(message.contains("OpenID.ClaimProvider.HashedClientSecret"), "{message}");
+        assert!(!message.contains("local-hash"), "{message}");
+        assert!(!message.contains("remote-hash"), "{message}");
+    }
+
     #[test]
     fn test_validate_add_preflight_topology_rejects_idp_mismatch() {
         let local_peer = PeerInfo {

rustfs/src/app/multipart_usecase.rs

@@ -101,7 +101,7 @@ use rustfs_utils::CompressionAlgorithm;
 #[cfg(test)]
 use rustfs_utils::http::insert_header;
 use rustfs_utils::http::{
-    SUFFIX_MAX_TOTAL_OBJECT_SIZE, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, SUFFIX_REPLICATION_STATUS,
+    SUFFIX_MAX_TOTAL_OBJECT_SIZE, SUFFIX_PLAINTEXT_CHECKSUM, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, SUFFIX_REPLICATION_STATUS,
     SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST, contains_key_str, get_consistent_str, get_header,
     get_source_scheme,
     headers::{AMZ_DECODED_CONTENT_LENGTH, AMZ_OBJECT_TAGGING, AMZ_STORAGE_CLASS},
@@ -653,7 +653,7 @@ impl DefaultMultipartUsecase {
                 content_size: 0,
                 principal: None,
             }
-            .validate_multipart_ssec(&multipart_info.user_defined)?;
+            .validate_complete_multipart_ssec(&multipart_info.user_defined)?;
         }
         let cache_adapter = self.object_data_cache();
         let _ = invalidate_object_data_cache_before_mutation(&cache_adapter, &bucket, &key).await;
@@ -1023,6 +1023,9 @@ impl DefaultMultipartUsecase {
                 ..Default::default()
             });
         }
+        if effective_sse.is_some() && opts.want_checksum.is_some() {
+            insert_str(&mut opts.user_defined, SUFFIX_PLAINTEXT_CHECKSUM, "true".to_string());
+        }
 
         let MultipartUploadResult {
             upload_id,

rustfs/src/app/object/copy.rs

@@ -575,6 +575,7 @@ impl DefaultObjectUsecase {
         }
 
         strip_managed_encryption_metadata(&mut user_defined);
+        remove_str(&mut user_defined, SUFFIX_PLAINTEXT_CHECKSUM);
 
         let destination_storage_class = storage_class
             .as_ref()
@@ -665,6 +666,7 @@ impl DefaultObjectUsecase {
         // none is requested, carry the source object's stored checksum over unchanged — the copy
         // does not alter the plaintext, so re-hashing would be wasted work and would flatten a
         // multipart composite value.
+        let destination_has_checksum = requested_checksum_type.is_some() || src_checksum.is_some();
         match requested_checksum_type {
             Some(checksum_type) => {
                 reader.add_calculated_checksum(checksum_type).map_err(ApiError::from)?;
@@ -696,6 +698,9 @@ impl DefaultObjectUsecase {
             write_plan = write_plan.with_encryption(material.write_encryption(None));
 
             user_defined.extend(encryption_material_to_metadata(&material)?);
+            if destination_has_checksum {
+                insert_str(&mut user_defined, SUFFIX_PLAINTEXT_CHECKSUM, "true".to_string());
+            }
         }
 
         reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?;

rustfs/src/app/object/extract.rs

@@ -63,6 +63,14 @@ pin_project! {
     }
 }
 
+pin_project! {
+    struct ExtractArchiveDecoderReader<R> {
+        #[pin]
+        inner: R,
+        _permit: OwnedSemaphorePermit,
+    }
+}
+
 #[derive(Debug, Default)]
 struct ExtractArchiveUploadState {
     etag: Option<String>,
@@ -100,6 +108,21 @@ impl<R> ExtractArchiveEtagReader<R> {
     }
 }
 
+impl<R> ExtractArchiveDecoderReader<R> {
+    fn new(inner: R, permit: OwnedSemaphorePermit) -> Self {
+        Self { inner, _permit: permit }
+    }
+}
+
+impl<R> AsyncRead for ExtractArchiveDecoderReader<R>
+where
+    R: AsyncRead,
+{
+    fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+        self.project().inner.poll_read(cx, buf)
+    }
+}
+
 fn extract_archive_incomplete_body(remaining: u64) -> std::io::Error {
     let Ok(remaining) = i64::try_from(remaining) else {
         return std::io::Error::new(std::io::ErrorKind::InvalidData, "archive remaining body length exceeds i64");
@@ -510,6 +533,36 @@ fn try_acquire_extract_staging_permit(manager: &ConcurrencyManager, staging_weig
     })
 }
 
+async fn build_admitted_extract_archive_decoder<R>(
+    manager: &ConcurrencyManager,
+    key: &str,
+    tracked_archive: R,
+) -> S3Result<ExtractArchiveDecoderReader<Box<dyn AsyncRead + Send + Unpin>>>
+where
+    R: AsyncRead + Send + Unpin + 'static,
+{
+    // Admission precedes stream inspection so saturation cannot allocate or
+    // drive another codec. The returned reader owns the permit through archive
+    // finalization and transport-length validation.
+    let permit = manager.try_acquire_snowball_archive_decoder().ok_or_else(|| {
+        object_s3_error(
+            S3ErrorCode::SlowDown,
+            "Snowball archive decoder limit reached, please reduce your request rate",
+        )
+    })?;
+    let (detected_archive_format, sniffed_archive) =
+        CompressionFormat::sniff(tracked_archive).await.map_err(|err| match err {
+            ZipError::InspectStream(source) => map_extract_archive_error(source),
+            _ => s3_error!(InvalidArgument, "Failed to detect archive compression"),
+        })?;
+    let archive_format = resolve_extract_archive_format(key, detected_archive_format);
+    let decoder = archive_format.get_decoder(sniffed_archive).map_err(|e| {
+        error!(error = ?e, "Archive decoder creation failed");
+        s3_error!(InvalidArgument, "get_decoder err")
+    })?;
+    Ok(ExtractArchiveDecoderReader::new(decoder, permit))
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum ExtractBatchAction {
     Stage,
@@ -2067,16 +2120,7 @@ impl DefaultObjectUsecase {
         let extract_limits = put_object_extract_limits();
         let tracked_archive =
             ExtractArchiveEtagReader::new(archive_reader, expected_archive_length, archive_upload_state.clone());
-        let (detected_archive_format, sniffed_archive) =
-            CompressionFormat::sniff(tracked_archive).await.map_err(|err| match err {
-                ZipError::InspectStream(source) => map_extract_archive_error(source),
-                _ => s3_error!(InvalidArgument, "Failed to detect archive compression"),
-            })?;
-        let archive_format = resolve_extract_archive_format(&key, detected_archive_format);
-        let decoder = archive_format.get_decoder(sniffed_archive).map_err(|e| {
-            error!(error = ?e, "Archive decoder creation failed");
-            s3_error!(InvalidArgument, "get_decoder err")
-        })?;
+        let decoder = build_admitted_extract_archive_decoder(get_concurrency_manager(), &key, tracked_archive).await?;
         let decoder = ExtractDecodedLimitReader::new(decoder, extract_limits.max_decoded_size);
 
         let mut ar = build_put_object_extract_archive(decoder, extract_limits);
@@ -2621,6 +2665,9 @@ impl DefaultObjectUsecase {
             }
             state.etag.as_ref().map(|etag| to_s3s_etag(etag))
         };
+        // Keep decoder admission through body-complete validation, then release
+        // it before response checksum and completion bookkeeping.
+        drop(decoder);
         apply_trailing_checksums(
             input.checksum_algorithm.as_ref().map(|a| a.as_str()),
             &req.trailing_headers,
@@ -2693,6 +2740,120 @@ mod tests {
         })
     }
 
+    #[tokio::test]
+    async fn snowball_archive_decoder_admission_is_global_and_lifetime_bound() {
+        struct PanicOnRead;
+        struct ErrorOnRead;
+
+        impl AsyncRead for PanicOnRead {
+            fn poll_read(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+                panic!("a saturated decoder admission must not inspect the archive body")
+            }
+        }
+
+        impl AsyncRead for ErrorOnRead {
+            fn poll_read(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+                Poll::Ready(Err(std::io::Error::other("injected decoder source failure")))
+            }
+        }
+
+        let manager = ConcurrencyManager::new();
+        let clone = manager.clone();
+        let mut held = Vec::new();
+        while let Some(permit) = manager.try_acquire_snowball_archive_decoder() {
+            held.push(permit);
+        }
+        assert!(!held.is_empty(), "the global decoder gate must admit at least one archive");
+
+        let error = match build_admitted_extract_archive_decoder(&clone, "archive.tar", PanicOnRead).await {
+            Ok(_) => panic!("a saturated decoder gate must reject without constructing another decoder"),
+            Err(error) => error,
+        };
+        assert_eq!(error.code(), &S3ErrorCode::SlowDown);
+
+        drop(
+            held.pop()
+                .expect("one decoder permit must be available for the lifetime test"),
+        );
+        let error = match build_admitted_extract_archive_decoder(&clone, "archive.tar", ErrorOnRead).await {
+            Ok(_) => panic!("archive inspection failure must remain an error"),
+            Err(error) => error,
+        };
+        assert_eq!(error.code(), &S3ErrorCode::InvalidArgument);
+        let released_after_error = clone
+            .try_acquire_snowball_archive_decoder()
+            .expect("archive inspection failure must release decoder capacity");
+        drop(released_after_error);
+
+        let mut builder = Builder::new(Vec::new());
+        let mut header = Header::new_gnu();
+        header.set_size(0);
+        header.set_cksum();
+        builder
+            .append_data(&mut header, "member.txt", &b""[..])
+            .await
+            .expect("decoder lifetime fixture should append its member");
+        let archive_bytes = builder.into_inner().await.expect("decoder lifetime fixture should finalize");
+        let expected_length = u64::try_from(archive_bytes.len()).expect("fixture length must fit u64");
+        let upload_state = Arc::new(Mutex::new(ExtractArchiveUploadState::default()));
+        let tracked_archive =
+            ExtractArchiveEtagReader::new(std::io::Cursor::new(archive_bytes), expected_length, upload_state.clone());
+        let decoder = build_admitted_extract_archive_decoder(&clone, "archive.tar", tracked_archive)
+            .await
+            .expect("released decoder capacity must be reusable");
+        assert!(
+            manager.try_acquire_snowball_archive_decoder().is_none(),
+            "the decoder reader must retain admission while it is active"
+        );
+
+        let extract_limits = put_object_extract_limits();
+        let decoder = ExtractDecodedLimitReader::new(decoder, extract_limits.max_decoded_size);
+        let mut archive = build_put_object_extract_archive(decoder, extract_limits);
+        let mut entries = archive.entries().expect("admitted archive entries should be readable");
+        let entry = entries
+            .next()
+            .await
+            .expect("admitted archive should contain its member")
+            .expect("admitted archive member should parse");
+        assert_eq!(entry.path_bytes().expect("archive member path should parse").as_ref(), b"member.txt");
+        drop(entry);
+        assert!(entries.next().await.is_none(), "admitted archive should contain one member");
+        drop(entries);
+        let mut decoder = match archive.into_inner() {
+            Ok(decoder) => decoder,
+            Err(_) => panic!("admitted archive should finalize"),
+        };
+        tokio::io::copy(&mut decoder, &mut tokio::io::sink())
+            .await
+            .expect("admitted archive should consume its remaining transport body");
+        assert!(
+            upload_state
+                .lock()
+                .expect("archive upload state lock must remain healthy")
+                .body_complete,
+            "transport-length validation must complete while decoder admission is held"
+        );
+        assert!(
+            manager.try_acquire_snowball_archive_decoder().is_none(),
+            "archive finalization and transport validation must retain decoder admission"
+        );
+        drop(decoder);
+        assert!(
+            clone.try_acquire_snowball_archive_decoder().is_some(),
+            "dropping the finalized decoder must release its global slot"
+        );
+
+        let cancelled =
+            build_admitted_extract_archive_decoder(&manager, "archive.tar", std::io::Cursor::new(b"cancelled".to_vec()))
+                .await
+                .expect("the decoder gate must remain reusable");
+        drop(cancelled);
+        assert!(
+            clone.try_acquire_snowball_archive_decoder().is_some(),
+            "dropping an unfinished decoder must release admission for cancellation"
+        );
+    }
+
     #[test]
     fn snowball_max_inflight_has_a_serial_compatibility_floor_and_bounded_ceiling() {
         assert_eq!(EXTRACT_DEFAULT_MAX_INFLIGHT, 1);

rustfs/src/app/object/mod.rs

@@ -145,9 +145,9 @@ use rustfs_utils::http::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_K
 use rustfs_utils::http::insert_header;
 use rustfs_utils::http::{
     AMZ_BUCKET_REPLICATION_STATUS, AMZ_CHECKSUM_MODE, AMZ_CHECKSUM_TYPE, AMZ_WEBSITE_REDIRECT_LOCATION, CONTENT_TYPE,
-    SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICA_STATUS, SUFFIX_REPLICA_TIMESTAMP,
-    SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_RESTORE_OPERATION_ID, SUFFIX_SOURCE_REPLICATION_CHECK,
-    SUFFIX_SOURCE_REPLICATION_REQUEST, get_header,
+    SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_PLAINTEXT_CHECKSUM, SUFFIX_REPLICA_STATUS,
+    SUFFIX_REPLICA_TIMESTAMP, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_RESTORE_OPERATION_ID,
+    SUFFIX_SOURCE_REPLICATION_CHECK, SUFFIX_SOURCE_REPLICATION_REQUEST, get_header,
     headers::{
         AMZ_CONTENT_SHA256, AMZ_DECODED_CONTENT_LENGTH, AMZ_MINIO_SNOWBALL_IGNORE_DIRS, AMZ_MINIO_SNOWBALL_IGNORE_ERRORS,
         AMZ_MINIO_SNOWBALL_PREFIX, AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE,

rustfs/src/app/object/put.rs

@@ -1447,6 +1447,10 @@ impl DefaultObjectUsecase {
             let encryption_metadata = encryption_material_to_metadata(&material)?;
             metadata.extend(encryption_metadata.clone());
             opts.user_defined.extend(encryption_metadata);
+            if opts.want_checksum.is_some() {
+                insert_str(&mut metadata, SUFFIX_PLAINTEXT_CHECKSUM, "true".to_string());
+                insert_str(&mut opts.user_defined, SUFFIX_PLAINTEXT_CHECKSUM, "true".to_string());
+            }
         }
 
         reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?;

rustfs/src/storage/concurrency/manager.rs

@@ -35,6 +35,10 @@ use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 use tracing::debug;
 
 const DERIVED_LARGE_PUT_ADMISSION_LIMIT_MAX: usize = 32;
+// Framed S2 alone can retain one encoded and one decoded block of roughly
+// 4 MiB each, while other codecs have their own larger windows. Four keeps
+// useful request parallelism without scaling codec memory and CPU with clients.
+const SNOWBALL_ARCHIVE_DECODER_LIMIT: usize = 4;
 pub(crate) const SNOWBALL_MEMBER_COMMIT_LIMIT: usize = 32;
 pub(crate) const SNOWBALL_STAGING_BYTES_LIMIT: usize = 4 * MI_B;
 
@@ -71,6 +75,8 @@ pub struct ConcurrencyManager {
     metrics_collector: Arc<MetricsCollector>,
     /// Foreground write admission policy, resolved once at startup.
     foreground_write_admission_policy: ForegroundWriteAdmissionPolicy,
+    /// Bounds active Snowball archive inspection and decoding across requests.
+    snowball_archive_decoder_semaphore: Arc<Semaphore>,
     /// Snowball members are internal PUTs, so they use a separate global gate
     /// from preparation through the independently owned post-commit tail.
     snowball_member_commit_semaphore: Arc<Semaphore>,
@@ -425,6 +431,7 @@ impl ConcurrencyManager {
             bandwidth_monitor,
             metrics_collector,
             foreground_write_admission_policy,
+            snowball_archive_decoder_semaphore: Arc::new(Semaphore::new(SNOWBALL_ARCHIVE_DECODER_LIMIT)),
             snowball_member_commit_semaphore: Arc::new(Semaphore::new(SNOWBALL_MEMBER_COMMIT_LIMIT)),
             snowball_staging_bytes_semaphore: Arc::new(Semaphore::new(SNOWBALL_STAGING_BYTES_LIMIT)),
         }
@@ -578,6 +585,11 @@ impl ConcurrencyManager {
             .await
     }
 
+    /// Try to acquire one global Snowball archive decoder slot.
+    pub(crate) fn try_acquire_snowball_archive_decoder(&self) -> Option<OwnedSemaphorePermit> {
+        self.snowball_archive_decoder_semaphore.clone().try_acquire_owned().ok()
+    }
+
     /// Acquire one global Snowball member lifecycle slot.
     pub(crate) async fn acquire_snowball_member_commit(&self) -> Result<OwnedSemaphorePermit, tokio::sync::AcquireError> {
         self.snowball_member_commit_semaphore.clone().acquire_owned().await
@@ -1095,8 +1107,8 @@ mod integration_tests {
     use super::super::io_schedule::{IoLoadLevel, IoPriority};
     use super::super::request_guard::GetObjectGuard;
     use super::{
-        ConcurrencyManager, ForegroundWriteAdmission, SNOWBALL_MEMBER_COMMIT_LIMIT, SNOWBALL_STAGING_BYTES_LIMIT,
-        derive_large_put_admission_limit,
+        ConcurrencyManager, ForegroundWriteAdmission, SNOWBALL_ARCHIVE_DECODER_LIMIT, SNOWBALL_MEMBER_COMMIT_LIMIT,
+        SNOWBALL_STAGING_BYTES_LIMIT, derive_large_put_admission_limit,
     };
     use crate::storage::storage_api::concurrency_consumer::PutObjectGuard;
     use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass};
@@ -1109,6 +1121,20 @@ mod integration_tests {
         let manager = ConcurrencyManager::new();
         let clone = manager.clone();
 
+        let decoder_permits = manager
+            .snowball_archive_decoder_semaphore
+            .clone()
+            .try_acquire_many_owned(
+                u32::try_from(SNOWBALL_ARCHIVE_DECODER_LIMIT).expect("Snowball decoder limit must fit into u32"),
+            )
+            .expect("the exact Snowball decoder limit must be available");
+        assert!(
+            clone.try_acquire_snowball_archive_decoder().is_none(),
+            "a cloned manager must share the global decoder gate"
+        );
+        drop(decoder_permits);
+        assert!(clone.try_acquire_snowball_archive_decoder().is_some());
+
         let commit_permits = manager
             .snowball_member_commit_semaphore
             .clone()

rustfs/src/storage/rpc/node_service.rs

@@ -230,6 +230,11 @@ fn validate_admin_heal_control_start(request: &rustfs_heal_contracts::heal_chann
     if request.source != rustfs_heal_contracts::heal_channel::HealRequestSource::Admin {
         return Err(Status::permission_denied("heal control start source must be admin"));
     }
+    if !request.heal_endpoints.is_empty() {
+        return Err(Status::invalid_argument(
+            "admin heal control start cannot contain automatic replacement endpoints",
+        ));
+    }
     if request.pool_index.is_some() != request.set_index.is_some() {
         return Err(Status::invalid_argument("heal control start requires both pool and set"));
     }
@@ -2532,6 +2537,7 @@ mod tests {
         make_heal_control_server, make_heal_control_server_with_cache, make_server, make_server_for_context,
         make_tier_mutation_control_server_for_context, previous_scanner_activity_response, remove_heal_control_replay,
         scanner_activity_response_v7, start_decommission_failure_response, stop_rebalance_response,
+        validate_admin_heal_control_start,
     };
     use crate::storage::rpc::node_service::heal::heal_topology_fingerprint;
     use crate::storage::storage_api::rpc_consumer::node_service::{DiskError, HealBucketInfo};
@@ -2766,6 +2772,23 @@ mod tests {
         assert!(!cache.contains_key("request-1"), "expired idle entries must be purged before admission");
     }
 
+    #[test]
+    fn heal_control_admin_start_rejects_automatic_replacement_endpoints() {
+        let mut request = rustfs_heal_contracts::heal_channel::create_heal_request(
+            String::new(),
+            None,
+            false,
+            Some(rustfs_heal_contracts::heal_channel::HealChannelPriority::High),
+        );
+        request.source = rustfs_heal_contracts::heal_channel::HealRequestSource::Admin;
+        request.recursive = Some(true);
+        request.heal_endpoints = vec!["/mnt/replacement".to_string()];
+
+        let err = validate_admin_heal_control_start(&request)
+            .expect_err("admin heal-control must not accept automatic replacement targets");
+        assert_eq!(err.code(), tonic::Code::InvalidArgument);
+    }
+
     #[tokio::test]
     async fn heal_control_executor_preserves_canonical_token_and_drops_query_results() {
         let manager = Arc::new(HealManager::new(Arc::new(HealControlMockStorage), None));

rustfs/src/storage/sse.rs

@@ -460,6 +460,28 @@ pub struct EncryptionRequest<'a> {
 }
 
 impl EncryptionRequest<'_> {
+    pub fn validate_complete_multipart_ssec(&self, user_defined: &HashMap<String, String>) -> Result<(), ApiError> {
+        let request_uses_ssec =
+            self.sse_customer_algorithm.is_some() || self.sse_customer_key.is_some() || self.sse_customer_key_md5.is_some();
+        if !request_uses_ssec {
+            let stored_algorithm = user_defined.get("x-amz-server-side-encryption-customer-algorithm");
+            let stored_key_md5 = user_defined.get("x-amz-server-side-encryption-customer-key-md5");
+            return match (stored_algorithm, stored_key_md5) {
+                (None, None) => Ok(()),
+                (Some(algorithm), Some(key_md5))
+                    if algorithm == DEFAULT_SSE_ALGORITHM
+                        && BASE64_STANDARD
+                            .decode_to_vec(key_md5)
+                            .is_ok_and(|decoded| decoded.len() == 16) =>
+                {
+                    Ok(())
+                }
+                _ => Err(ssec_invalid_request("The multipart upload contains invalid SSE-C metadata.")),
+            };
+        }
+        self.validate_multipart_ssec(user_defined)
+    }
+
     pub fn validate_multipart_ssec(&self, user_defined: &HashMap<String, String>) -> Result<(), ApiError> {
         let stored_algorithm = user_defined.get("x-amz-server-side-encryption-customer-algorithm");
         let stored_key_md5 = user_defined.get("x-amz-server-side-encryption-customer-key-md5");
@@ -6374,6 +6396,61 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_validate_complete_multipart_ssec_allows_omitted_parameters() {
+        let no_ssec = EncryptionRequest {
+            sse_customer_algorithm: None,
+            sse_customer_key: None,
+            sse_customer_key_md5: None,
+            ..multipart_ssec_request(42)
+        };
+
+        assert!(no_ssec.validate_complete_multipart_ssec(&multipart_ssec_metadata(42)).is_ok());
+        assert!(no_ssec.validate_complete_multipart_ssec(&HashMap::new()).is_ok());
+
+        for invalid_metadata in [
+            HashMap::from([("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string())]),
+            HashMap::from([
+                ("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES128".to_string()),
+                ("x-amz-server-side-encryption-customer-key-md5".to_string(), md5_base64([42u8; 32])),
+            ]),
+            HashMap::from([
+                ("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string()),
+                ("x-amz-server-side-encryption-customer-key-md5".to_string(), "invalid".to_string()),
+            ]),
+        ] {
+            assert_eq!(
+                no_ssec
+                    .validate_complete_multipart_ssec(&invalid_metadata)
+                    .expect_err("corrupt SSE-C session metadata must fail closed")
+                    .code,
+                S3ErrorCode::InvalidRequest
+            );
+        }
+    }
+
+    #[test]
+    fn test_validate_complete_multipart_ssec_still_validates_present_parameters() {
+        let partial = EncryptionRequest {
+            sse_customer_key: None,
+            ..multipart_ssec_request(42)
+        };
+        assert_eq!(
+            partial
+                .validate_complete_multipart_ssec(&multipart_ssec_metadata(42))
+                .expect_err("partial SSE-C parameters must fail")
+                .code,
+            S3ErrorCode::InvalidRequest
+        );
+        assert_eq!(
+            multipart_ssec_request(43)
+                .validate_complete_multipart_ssec(&multipart_ssec_metadata(42))
+                .expect_err("wrong SSE-C parameters must fail")
+                .code,
+            S3ErrorCode::InvalidRequest
+        );
+    }
+
     #[test]
     fn test_validate_multipart_ssec_rejects_wrong_key_without_leaking_it() {
         let request = multipart_ssec_request(43);

scripts/s3-tests/patches/0003-complete-multipart-ssec-headers.patch

@@ -0,0 +1,28 @@
+diff --git a/s3tests/functional/test_s3.py b/s3tests/functional/test_s3.py
+--- a/s3tests/functional/test_s3.py
++++ b/s3tests/functional/test_s3.py
+@@ -20631,13 +20631,6 @@ def _test_copy_part_enc(file_size, source_mode_key, dest_mode_key, source_sc=Non
+     })
+ 
+     if dest_mode_key == 'sse-c':
+-        # make sure api is verifying the SSE-C headers
+-        e = assert_raises(ClientError, client.complete_multipart_upload,
+-                          Bucket=dest_bucket_name, Key='testobj2',
+-                          UploadId=upload_id, MultipartUpload={'Parts': parts})
+-        status, _ = _get_status_and_error_code(e.response)
+-        assert status == 400
+-
+         # and the key would be the same as the one used in upload part
+         # use the source key to complete the upload
+         # this is not allowed, so we expect an error
+@@ -20649,6 +20642,10 @@ def _test_copy_part_enc(file_size, source_mode_key, dest_mode_key, source_sc=Non
+         status, _ = _get_status_and_error_code(e.response)
+         assert status == 400
+ 
++        # CompleteMultipartUpload does not require SSE-C headers. The upload
++        # metadata already identifies the key used for the uploaded parts.
++        complete_args = {}
++
+     # complete the multipart upload
+     response = client.complete_multipart_upload(
+         Bucket=dest_bucket_name,

scripts/security/check_tier_artifact_workflow.sh

@@ -0,0 +1,168 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+WORKFLOW="${1:-.github/workflows/rustfs-tier-test.yml}"
+
+fail() {
+  printf 'tier artifact workflow check failed: %s\n' "$*" >&2
+  exit 1
+}
+
+[ -f "${WORKFLOW}" ] || fail "workflow not found: ${WORKFLOW}"
+
+for fixed_path in \
+  /tmp/rustfs-tier.log \
+  /tmp/rustfs-tier-report.md \
+  /tmp/rustfs-tier-cases.md \
+  /tmp/rustfs-tier-gate.rc; do
+  if grep -Fq -- "${fixed_path}" "${WORKFLOW}"; then
+    fail "fixed cross-run path remains: ${fixed_path}"
+  fi
+done
+
+for required in \
+  'TIER_ARTIFACTS_DIR: /tmp/rustfs-tier-artifacts-${{ github.run_id }}-${{ github.run_attempt }}' \
+  'id: evidence' \
+  'umask 077' \
+  'mkdir -- "${TIER_ARTIFACTS_DIR}"' \
+  'rustfs-tier.log' \
+  'rustfs-tier-report.md' \
+  'rustfs-tier-cases.md' \
+  'rustfs-tier-gate.rc' \
+  'provenance.json' \
+  'Verify required tier evidence' \
+  'id: evidence_verify' \
+  'id: gate' \
+  'path: ${{ env.TIER_ARTIFACTS_DIR }}/' \
+  'if-no-files-found: error'; do
+  grep -Fq -- "${required}" "${WORKFLOW}" || fail "required contract is missing: ${required}"
+done
+
+step_block() {
+  local name="$1"
+  awk -v name="${name}" '
+    $0 == "      - name: " name { capture=1; found=1 }
+    capture && $0 != "      - name: " name && $0 ~ /^      - name: / { exit }
+    capture { print }
+    END { if (!found) exit 1 }
+  ' "${WORKFLOW}"
+}
+
+require_in_block() {
+  local block="$1" expected="$2" label="$3"
+  grep -Fqx -- "${expected}" <<< "${block}" \
+    || fail "${label} is missing: ${expected}"
+}
+
+verify_block="$(step_block 'Verify required tier evidence')" \
+  || fail "required-evidence step was not found"
+require_in_block "${verify_block}" \
+  "        if: \${{ always() && steps.evidence.outcome == 'success' }}" \
+  "required-evidence condition"
+require_in_block "${verify_block}" \
+  '          failed=0' \
+  "required-evidence failure accumulator"
+require_in_block "${verify_block}" \
+  '          [ "${failed}" -eq 0 ]' \
+  "required-evidence fail-closed result"
+require_in_block "${verify_block}" \
+  '            if [ ! -s "${TIER_ARTIFACTS_DIR}/${name}" ]; then' \
+  "required-file missing/empty predicate"
+require_in_block "${verify_block}" \
+  '            if [ ! -d "${TIER_ARTIFACTS_DIR}/${name}" ]; then' \
+  "required-directory missing predicate"
+for required_name in \
+  rustfs-tier.log \
+  rustfs-tier-report.md \
+  rustfs-tier-cases.md \
+  rustfs-tier-gate.rc \
+  provenance.json; do
+  grep -Fq -- "${required_name}" <<< "${verify_block}" \
+    || fail "required-evidence step does not verify ${required_name}"
+done
+require_in_block "${verify_block}" \
+  '          if ! find "${TIER_ARTIFACTS_DIR}/cases" -maxdepth 1 -type f -name '\''*.json'\'' -print -quit 2>/dev/null | grep -q .; then' \
+  "required atomic-case predicate"
+[ "$(grep -Fc '            failed=1' <<< "${verify_block}")" -eq 3 ] \
+  || fail "required-evidence step must fail for files, directories, and atomic cases"
+
+upload_block="$(step_block 'Upload report and logs')" \
+  || fail "upload step was not found"
+[ -n "${upload_block}" ] || fail "upload step was not found"
+require_in_block "${upload_block}" \
+  "        if: \${{ always() && steps.evidence.outcome == 'success' }}" \
+  "artifact upload condition"
+[ "$(grep -c '^[[:space:]]*path:' <<< "${upload_block}")" -eq 1 ] \
+  || fail "upload step must contain exactly one path"
+grep -Fq 'path: ${{ env.TIER_ARTIFACTS_DIR }}/' <<< "${upload_block}" \
+  || fail "upload step must archive only the run-scoped evidence directory"
+require_in_block "${upload_block}" \
+  '          if-no-files-found: error' \
+  "artifact upload empty-evidence behavior"
+
+gate_block="$(step_block 'Enforce tier suite result')" \
+  || fail "final gate step was not found"
+require_in_block "${gate_block}" '        if: always()' "final gate condition"
+require_in_block "${gate_block}" \
+  '          EVIDENCE_OUTCOME: ${{ steps.evidence.outcome }}' \
+  "final gate evidence status"
+require_in_block "${gate_block}" \
+  '          TEST_OUTCOME: ${{ steps.test.outcome }}' \
+  "final gate suite status"
+require_in_block "${gate_block}" \
+  '          GATE_RC_FILE: ${{ env.TIER_ARTIFACTS_DIR }}/rustfs-tier-gate.rc' \
+  "final gate structured status path"
+require_in_block "${gate_block}" \
+  '          if [ "${EVIDENCE_OUTCOME}" != "success" ]; then' \
+  "final gate evidence enforcement"
+require_in_block "${gate_block}" \
+  '          if [ "${TEST_OUTCOME}" != "success" ]; then' \
+  "final gate suite enforcement"
+require_in_block "${gate_block}" \
+  '          elif [ ! -s "${GATE_RC_FILE}" ]; then' \
+  "final gate missing-result enforcement"
+require_in_block "${gate_block}" \
+  '            if ! [[ "${GATE_RC}" =~ ^[0-9]+$ ]] || [ "${GATE_RC}" -ne 0 ]; then' \
+  "final gate nonzero-result enforcement"
+require_in_block "${gate_block}" \
+  '          [ "${failed}" -eq 0 ]' \
+  "final gate fail-closed result"
+
+issue_block="$(step_block 'File failure issue in rustfs/backlog')" \
+  || fail "failure-issue step was not found"
+require_in_block "${issue_block}" \
+  '          EVIDENCE_DIR: ${{ env.TIER_ARTIFACTS_DIR }}' \
+  "failure-issue evidence directory"
+require_in_block "${issue_block}" \
+  '          EVIDENCE_OUTCOME: ${{ steps.evidence.outcome }}' \
+  "failure-issue initialization status"
+require_in_block "${issue_block}" \
+  '          VERIFY_OUTCOME: ${{ steps.evidence_verify.outcome }}' \
+  "failure-issue verification status"
+require_in_block "${issue_block}" \
+  '          GATE_OUTCOME: ${{ steps.gate.outcome }}' \
+  "failure-issue final-gate status"
+require_in_block "${issue_block}" \
+  '            if [ "${EVIDENCE_OUTCOME}" != "success" ]; then' \
+  "failure-issue rejected-evidence guard"
+require_in_block "${issue_block}" \
+  '            elif [ ! -d "${EVIDENCE_DIR}" ] || [ -L "${EVIDENCE_DIR}" ]; then' \
+  "failure-issue unsafe-evidence guard"
+grep -Fq "steps.evidence_verify.outcome == 'failure'" <<< "${issue_block}" \
+  || fail "failure-issue condition does not cover evidence verification"
+grep -Fq "steps.gate.outcome == 'failure'" <<< "${issue_block}" \
+  || fail "failure-issue condition does not cover the final gate"
+
+step_line() {
+  local name="$1"
+  awk -v name="${name}" '$0 == "      - name: " name { print NR; exit }' "${WORKFLOW}"
+}
+verify_line="$(step_line 'Verify required tier evidence')"
+gate_line="$(step_line 'Enforce tier suite result')"
+issue_line="$(step_line 'File failure issue in rustfs/backlog')"
+[ -n "${verify_line}" ] && [ -n "${gate_line}" ] && [ -n "${issue_line}" ] \
+  || fail "cannot determine evidence/final-gate/failure-issue order"
+[ "${verify_line}" -lt "${gate_line}" ] && [ "${gate_line}" -lt "${issue_line}" ] \
+  || fail "failure issue must run after evidence verification and the final gate"
+
+printf 'tier artifact workflow contract is isolated\n'