feat(ci): publish functional reports and keep workflows non-blocking

The transition_matrix_tests use #[serial_test::serial] which has no
effect under nextest (each test runs in a separate process). When running
alongside thousands of other ecstore tests, the shared metadata cache
generation counter can race, causing intermittent 'metadata read should
publish the generation under test' panics.

Add both tests to the ecstore-serial-flaky test group in both default
and ci nextest profiles so they run single-threaded.

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
 sha256-darwin=d6aa36cfaae2c4d8590482c7e47138c5965b335b34a75f50d11ffc3366e9021e
-sha256-linux=96db8060fce98addda4f69092d297ca236bec4892d820617a26a261eedac61b0
+sha256-linux=e3eb4ab7fc72224abf58c546ac0706d6605d3bd26bac7d8ce338829fd3daecc2

.config/nextest.toml

@@ -100,6 +100,16 @@ test-group = 'embedded-test-ports'
 filter = 'package(rustfs-ecstore) & test(manual_transition_page_checkpoint_persists_durable_job_progress)'
 test-group = 'ecstore-serial-flaky'
 
+# Serialize the transition matrix tests. They build a 4-disk hermetic erasure
+# set, populate the get_object_metadata_cache, and assert generation lifecycle
+# semantics. serial_test's #[serial] has no effect across nextest's process
+# boundary, so concurrent execution races the shared metadata-cache generation
+# counter and causes spurious "metadata read should publish the generation"
+# panics. Preventive serialization, no retries.
+[[profile.default.overrides]]
+filter = 'package(rustfs-ecstore) & test(set_disk::transition_matrix_tests::)'
+test-group = 'ecstore-serial-flaky'
+
 # The durable ILM decommission regressions build isolated multi-pool stores and
 # deliberately take source or target disks offline while checking fencing.
 [[profile.default.overrides]]
@@ -232,6 +242,12 @@ test-group = 'embedded-test-ports'
 filter = 'package(rustfs-ecstore) & test(manual_transition_page_checkpoint_persists_durable_job_progress)'
 test-group = 'ecstore-serial-flaky'
 
+# Serialize the transition matrix tests under the ci profile too (see the
+# matching default-profile override near the top). No retries.
+[[profile.ci.overrides]]
+filter = 'package(rustfs-ecstore) & test(set_disk::transition_matrix_tests::)'
+test-group = 'ecstore-serial-flaky'
+
 [[profile.ci.overrides]]
 filter = 'package(rustfs-ecstore) & (test(decommission_migrates_and_verifies_registered_durable_ilm_records) | test(decommission_durable_ilm_target_read_error_is_not_masked_by_peer_success) | test(decommission_durable_ilm_terminal_receipt_recovers_failed_source_cleanup) | test(decommission_durable_ilm_receipt_pagination_fails_closed_on_second_page) | test(decommission_durable_ilm_recovery_keeps_multiple_active_sources))'
 test-group = 'ecstore-serial-flaky'

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

@@ -38,6 +38,7 @@ env:
 jobs:
   kms-test:
     runs-on: smoke-testing
+    continue-on-error: true
     timeout-minutes: 420
     if: ${{ github.event_name == 'workflow_dispatch' || github.event.workflow_run.conclusion == 'success' }}
     steps:
@@ -86,6 +87,7 @@ jobs:
 
       - name: Run KMS suite
         id: test
+        continue-on-error: true
         env:
           LOG_FILE: /tmp/rustfs-kms.log
         run: |
@@ -119,12 +121,65 @@ jobs:
           else
             PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
           fi
+          CASE_TABLE="/tmp/rustfs-kms-cases.md"
+          python3 - "${LOG_FILE}" "${CASE_TABLE}" <<'PY'
+          import re
+          import sys
+
+          log_file, out_file = sys.argv[1], sys.argv[2]
+          ansi = re.compile(r'\x1b\[[0-9;]*m')
+          start_re = re.compile(r'^---\s+([A-Z]+-[0-9]+)\s+(.+?)\s+---$')
+          done_re = re.compile(r'^\[(PASS|FAIL|UNSUPPORTED)\]\s+([A-Z]+-[0-9]+)\b')
+
+          rows = []
+          index = {}
+          try:
+            with open(log_file, 'r', encoding='utf-8', errors='replace') as fh:
+              for raw in fh:
+                line = ansi.sub('', raw).strip()
+                m = start_re.match(line)
+                if m:
+                  case_id, name = m.group(1), m.group(2)
+                  if case_id not in index:
+                    index[case_id] = len(rows)
+                    rows.append([case_id, name, 'RUNNING'])
+                  continue
+                m = done_re.match(line)
+                if m:
+                  status, case_id = m.group(1), m.group(2)
+                  if case_id in index:
+                    rows[index[case_id]][2] = status
+                  else:
+                    rows.append([case_id, case_id, status])
+                    index[case_id] = len(rows) - 1
+          except FileNotFoundError:
+            rows = []
+
+          counts = {'PASS': 0, 'FAIL': 0, 'UNSUPPORTED': 0, 'RUNNING': 0}
+          for _, _, status in rows:
+            counts[status] = counts.get(status, 0) + 1
+
+          with open(out_file, 'w', encoding='utf-8') as out:
+            out.write('## Case Summary\n\n')
+            out.write(f"- Total: {len(rows)}\\n")
+            out.write(f"- PASS: {counts.get('PASS', 0)}\\n")
+            out.write(f"- FAIL: {counts.get('FAIL', 0)}\\n")
+            out.write(f"- UNSUPPORTED: {counts.get('UNSUPPORTED', 0)}\\n")
+            out.write('\\n')
+            out.write('| Case | Name | Status |\\n')
+            out.write('| --- | --- | --- |\\n')
+            for case_id, name, status in rows:
+              out.write(f'| {case_id} | {name} | {status} |\\n')
+          PY
           {
             echo "# RustFS KMS test report"
             echo ""
             echo "- Run: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
             echo "- Trigger: ${{ github.event_name }}"
             echo "- Package: ${PACKAGE_SOURCE}"
+            echo "- Test Step Outcome: ${{ steps.test.outcome }}"
+            echo ""
+            cat "${CASE_TABLE}" || true
             echo ""
             echo "## Log tail"
             echo '```text'
@@ -133,6 +188,129 @@ jobs:
           } | tee "${REPORT_FILE}"
           cat "${REPORT_FILE}" >> "${GITHUB_STEP_SUMMARY}"
 
+      - name: Upload functional report to dashboard
+        if: always()
+        continue-on-error: true
+        env:
+          GH_TOKEN: ${{ env.PF_TESTING_GH_TOKEN }}
+          REPORT_FILE: /tmp/rustfs-kms-report.md
+          SUITE: kms
+        run: |
+          set -euo pipefail
+          if [ -z "${GH_TOKEN:-}" ]; then
+            echo "PF_TESTING_GH_TOKEN is not configured; skipping dashboard upload"
+            exit 0
+          fi
+          DATE="$(date -u +%Y-%m-%d)"
+          REPORT_PATH="functional-reports/${SUITE}/${DATE}.md"
+          CONTENT="$(python3 -c 'import base64,sys;print(base64.b64encode(open(sys.argv[1],"rb").read()).decode())' "${REPORT_FILE}")"
+          SHA="$(gh api "repos/rustfs/dashboard/contents/${REPORT_PATH}" -q '.sha' 2>/dev/null || true)"
+          if [ -n "${SHA}" ]; then
+            jq -n --arg msg "report(${SUITE}): ${DATE}" --arg content "${CONTENT}" --arg sha "${SHA}" \
+              '{message:$msg, content:$content, sha:$sha}' \
+              | gh api --method PUT "repos/rustfs/dashboard/contents/${REPORT_PATH}" --input - >/dev/null
+          else
+            jq -n --arg msg "report(${SUITE}): ${DATE}" --arg content "${CONTENT}" \
+              '{message:$msg, content:$content}' \
+              | gh api --method PUT "repos/rustfs/dashboard/contents/${REPORT_PATH}" --input - >/dev/null
+          fi
+
+          cat > /tmp/rustfs-functional-index.html <<'EOF'
+          <!doctype html>
+          <html lang="en">
+          <head>
+            <meta charset="utf-8" />
+            <meta name="viewport" content="width=device-width, initial-scale=1" />
+            <title>RustFS Functional Test Reports</title>
+            <style>
+              :root { --bg:#f4f6fb; --card:#fff; --text:#1f2937; --muted:#6b7280; --line:#e5e7eb; --accent:#0f766e; }
+              * { box-sizing: border-box; }
+              body { margin: 0; font-family: ui-sans-serif, -apple-system, Segoe UI, Helvetica, Arial, sans-serif; background: var(--bg); color: var(--text); }
+              .wrap { max-width: 980px; margin: 32px auto; padding: 0 16px; }
+              .card { background: var(--card); border: 1px solid var(--line); border-radius: 14px; padding: 20px; }
+              h1 { margin: 0 0 8px; font-size: 26px; }
+              p { margin: 0 0 14px; color: var(--muted); }
+              .tabs { display: flex; gap: 10px; margin: 14px 0 18px; flex-wrap: wrap; }
+              button { border: 1px solid var(--line); background: #fff; color: var(--text); border-radius: 10px; padding: 8px 14px; cursor: pointer; }
+              button.active { background: var(--accent); color: #fff; border-color: var(--accent); }
+              ul { list-style: none; margin: 0; padding: 0; }
+              li { padding: 10px 0; border-bottom: 1px solid var(--line); }
+              a { color: var(--accent); text-decoration: none; }
+              a:hover { text-decoration: underline; }
+            </style>
+          </head>
+          <body>
+            <div class="wrap">
+              <div class="card">
+                <h1>RustFS Functional Test Reports</h1>
+                <p>S3, KMS, Tier report tabs. Each tab lists reports by date.</p>
+                <div class="tabs" id="tabs"></div>
+                <ul id="list"></ul>
+              </div>
+            </div>
+            <script>
+              const suites = [
+                { key: 's3', label: 'S3 Compatibility' },
+                { key: 'kms', label: 'KMS' },
+                { key: 'tier', label: 'Tier' },
+              ];
+              const tabs = document.getElementById('tabs');
+              const list = document.getElementById('list');
+
+              async function loadSuite(suite) {
+                list.innerHTML = '<li>Loading...</li>';
+                const api = `https://api.github.com/repos/rustfs/dashboard/contents/functional-reports/${suite}`;
+                try {
+                  const res = await fetch(api);
+                  if (!res.ok) {
+                    list.innerHTML = '<li>No reports yet.</li>';
+                    return;
+                  }
+                  const data = await res.json();
+                  const files = data.filter(f => f.type === 'file' && f.name.endsWith('.md')).sort((a,b) => b.name.localeCompare(a.name));
+                  if (!files.length) {
+                    list.innerHTML = '<li>No reports yet.</li>';
+                    return;
+                  }
+                  list.innerHTML = files.map(f => `<li><a href="${f.html_url}" target="_blank" rel="noreferrer">${f.name.replace('.md','')}</a></li>`).join('');
+                } catch (_e) {
+                  list.innerHTML = '<li>Failed to load reports.</li>';
+                }
+              }
+
+              function setActive(key) {
+                for (const btn of tabs.querySelectorAll('button')) {
+                  btn.classList.toggle('active', btn.dataset.key === key);
+                }
+                loadSuite(key);
+              }
+
+              for (const suite of suites) {
+                const btn = document.createElement('button');
+                btn.textContent = suite.label;
+                btn.dataset.key = suite.key;
+                btn.addEventListener('click', () => setActive(suite.key));
+                tabs.appendChild(btn);
+              }
+              setActive('s3');
+            </script>
+          </body>
+          </html>
+          EOF
+
+          INDEX_PATH="functional/index.html"
+          INDEX_CONTENT="$(python3 -c 'import base64;print(base64.b64encode(open("/tmp/rustfs-functional-index.html","rb").read()).decode())')"
+          INDEX_SHA="$(gh api "repos/rustfs/dashboard/contents/${INDEX_PATH}" -q '.sha' 2>/dev/null || true)"
+          if [ -n "${INDEX_SHA}" ]; then
+            jq -n --arg msg "functional ui update" --arg content "${INDEX_CONTENT}" --arg sha "${INDEX_SHA}" \
+              '{message:$msg, content:$content, sha:$sha}' \
+              | gh api --method PUT "repos/rustfs/dashboard/contents/${INDEX_PATH}" --input - >/dev/null
+          else
+            jq -n --arg msg "functional ui init" --arg content "${INDEX_CONTENT}" \
+              '{message:$msg, content:$content}' \
+              | gh api --method PUT "repos/rustfs/dashboard/contents/${INDEX_PATH}" --input - >/dev/null
+          fi
+
       - name: Upload report and logs
         if: always()
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6

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

@@ -33,10 +33,12 @@ env:
   RUSTFS_NODES: ${{ secrets.RUSTFS_NODES || vars.RUSTFS_NODES }}
   RUSTFS_SSH_USER: ${{ secrets.RUSTFS_SSH_USER || vars.RUSTFS_SSH_USER }}
   RUSTFS_NIGHTLY_PACKAGE_URL: ${{ vars.RUSTFS_NIGHTLY_PACKAGE_URL || 'https://dl.rustfs.com/artifacts/rustfs/packages/nightly/rustfs-nightly-latest.deb' }}
+  PF_TESTING_GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
 
 jobs:
   s3-compat-test:
     runs-on: smoke-testing
+    continue-on-error: true
     timeout-minutes: 360
     if: ${{ github.event_name == 'workflow_dispatch' || github.event.workflow_run.conclusion == 'success' }}
     steps:
@@ -76,6 +78,7 @@ jobs:
 
       - name: Run S3 compatibility suite
         id: test
+        continue-on-error: true
         env:
           LOG_FILE: /tmp/rustfs-s3-compat.log
         run: |
@@ -109,12 +112,68 @@ jobs:
           else
             PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
           fi
+          CASE_TABLE="/tmp/rustfs-s3-compat-cases.md"
+          python3 - "${LOG_FILE}" "${CASE_TABLE}" <<'PY'
+          import re
+          import sys
+
+          log_file, out_file = sys.argv[1], sys.argv[2]
+          ansi = re.compile(r'\x1b\[[0-9;]*m')
+          start_re = re.compile(r'^---\s+([A-Z]+-[0-9]+)\s+(.+?)\s+---$')
+          done_re = re.compile(r'^\[(PASS|FAIL|UNSUPPORTED)\]\s+([A-Z]+-[0-9]+)\b')
+
+          rows = []
+          index = {}
+          current = None
+          try:
+            with open(log_file, 'r', encoding='utf-8', errors='replace') as fh:
+              for raw in fh:
+                line = ansi.sub('', raw).strip()
+                m = start_re.match(line)
+                if m:
+                  case_id, name = m.group(1), m.group(2)
+                  current = case_id
+                  if case_id not in index:
+                    index[case_id] = len(rows)
+                    rows.append([case_id, name, 'RUNNING'])
+                  continue
+                m = done_re.match(line)
+                if m:
+                  status, case_id = m.group(1), m.group(2)
+                  if case_id in index:
+                    rows[index[case_id]][2] = status
+                  else:
+                    rows.append([case_id, case_id, status])
+                    index[case_id] = len(rows) - 1
+                  current = None
+          except FileNotFoundError:
+            rows = []
+
+          counts = {'PASS': 0, 'FAIL': 0, 'UNSUPPORTED': 0, 'RUNNING': 0}
+          for _, _, status in rows:
+            counts[status] = counts.get(status, 0) + 1
+
+          with open(out_file, 'w', encoding='utf-8') as out:
+            out.write('## Case Summary\n\n')
+            out.write(f"- Total: {len(rows)}\\n")
+            out.write(f"- PASS: {counts.get('PASS', 0)}\\n")
+            out.write(f"- FAIL: {counts.get('FAIL', 0)}\\n")
+            out.write(f"- UNSUPPORTED: {counts.get('UNSUPPORTED', 0)}\\n")
+            out.write('\\n')
+            out.write('| Case | Name | Status |\\n')
+            out.write('| --- | --- | --- |\\n')
+            for case_id, name, status in rows:
+              out.write(f'| {case_id} | {name} | {status} |\\n')
+          PY
           {
             echo "# RustFS S3 compatibility test report"
             echo ""
             echo "- Run: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
             echo "- Trigger: ${{ github.event_name }}"
             echo "- Package: ${PACKAGE_SOURCE}"
+            echo "- Test Step Outcome: ${{ steps.test.outcome }}"
+            echo ""
+            cat "${CASE_TABLE}" || true
             echo ""
             echo "## Log tail"
             echo '```text'
@@ -123,6 +182,129 @@ jobs:
           } | tee "${REPORT_FILE}"
           cat "${REPORT_FILE}" >> "${GITHUB_STEP_SUMMARY}"
 
+      - name: Upload functional report to dashboard
+        if: always()
+        continue-on-error: true
+        env:
+          GH_TOKEN: ${{ env.PF_TESTING_GH_TOKEN }}
+          REPORT_FILE: /tmp/rustfs-s3-compat-report.md
+          SUITE: s3
+        run: |
+          set -euo pipefail
+          if [ -z "${GH_TOKEN:-}" ]; then
+            echo "PF_TESTING_GH_TOKEN is not configured; skipping dashboard upload"
+            exit 0
+          fi
+          DATE="$(date -u +%Y-%m-%d)"
+          REPORT_PATH="functional-reports/${SUITE}/${DATE}.md"
+          CONTENT="$(python3 -c 'import base64,sys;print(base64.b64encode(open(sys.argv[1],"rb").read()).decode())' "${REPORT_FILE}")"
+          SHA="$(gh api "repos/rustfs/dashboard/contents/${REPORT_PATH}" -q '.sha' 2>/dev/null || true)"
+          if [ -n "${SHA}" ]; then
+            jq -n --arg msg "report(${SUITE}): ${DATE}" --arg content "${CONTENT}" --arg sha "${SHA}" \
+              '{message:$msg, content:$content, sha:$sha}' \
+              | gh api --method PUT "repos/rustfs/dashboard/contents/${REPORT_PATH}" --input - >/dev/null
+          else
+            jq -n --arg msg "report(${SUITE}): ${DATE}" --arg content "${CONTENT}" \
+              '{message:$msg, content:$content}' \
+              | gh api --method PUT "repos/rustfs/dashboard/contents/${REPORT_PATH}" --input - >/dev/null
+          fi
+
+          cat > /tmp/rustfs-functional-index.html <<'EOF'
+          <!doctype html>
+          <html lang="en">
+          <head>
+            <meta charset="utf-8" />
+            <meta name="viewport" content="width=device-width, initial-scale=1" />
+            <title>RustFS Functional Test Reports</title>
+            <style>
+              :root { --bg:#f4f6fb; --card:#fff; --text:#1f2937; --muted:#6b7280; --line:#e5e7eb; --accent:#0f766e; }
+              * { box-sizing: border-box; }
+              body { margin: 0; font-family: ui-sans-serif, -apple-system, Segoe UI, Helvetica, Arial, sans-serif; background: var(--bg); color: var(--text); }
+              .wrap { max-width: 980px; margin: 32px auto; padding: 0 16px; }
+              .card { background: var(--card); border: 1px solid var(--line); border-radius: 14px; padding: 20px; }
+              h1 { margin: 0 0 8px; font-size: 26px; }
+              p { margin: 0 0 14px; color: var(--muted); }
+              .tabs { display: flex; gap: 10px; margin: 14px 0 18px; flex-wrap: wrap; }
+              button { border: 1px solid var(--line); background: #fff; color: var(--text); border-radius: 10px; padding: 8px 14px; cursor: pointer; }
+              button.active { background: var(--accent); color: #fff; border-color: var(--accent); }
+              ul { list-style: none; margin: 0; padding: 0; }
+              li { padding: 10px 0; border-bottom: 1px solid var(--line); }
+              a { color: var(--accent); text-decoration: none; }
+              a:hover { text-decoration: underline; }
+            </style>
+          </head>
+          <body>
+            <div class="wrap">
+              <div class="card">
+                <h1>RustFS Functional Test Reports</h1>
+                <p>S3, KMS, Tier report tabs. Each tab lists reports by date.</p>
+                <div class="tabs" id="tabs"></div>
+                <ul id="list"></ul>
+              </div>
+            </div>
+            <script>
+              const suites = [
+                { key: 's3', label: 'S3 Compatibility' },
+                { key: 'kms', label: 'KMS' },
+                { key: 'tier', label: 'Tier' },
+              ];
+              const tabs = document.getElementById('tabs');
+              const list = document.getElementById('list');
+
+              async function loadSuite(suite) {
+                list.innerHTML = '<li>Loading...</li>';
+                const api = `https://api.github.com/repos/rustfs/dashboard/contents/functional-reports/${suite}`;
+                try {
+                  const res = await fetch(api);
+                  if (!res.ok) {
+                    list.innerHTML = '<li>No reports yet.</li>';
+                    return;
+                  }
+                  const data = await res.json();
+                  const files = data.filter(f => f.type === 'file' && f.name.endsWith('.md')).sort((a,b) => b.name.localeCompare(a.name));
+                  if (!files.length) {
+                    list.innerHTML = '<li>No reports yet.</li>';
+                    return;
+                  }
+                  list.innerHTML = files.map(f => `<li><a href="${f.html_url}" target="_blank" rel="noreferrer">${f.name.replace('.md','')}</a></li>`).join('');
+                } catch (_e) {
+                  list.innerHTML = '<li>Failed to load reports.</li>';
+                }
+              }
+
+              function setActive(key) {
+                for (const btn of tabs.querySelectorAll('button')) {
+                  btn.classList.toggle('active', btn.dataset.key === key);
+                }
+                loadSuite(key);
+              }
+
+              for (const suite of suites) {
+                const btn = document.createElement('button');
+                btn.textContent = suite.label;
+                btn.dataset.key = suite.key;
+                btn.addEventListener('click', () => setActive(suite.key));
+                tabs.appendChild(btn);
+              }
+              setActive('s3');
+            </script>
+          </body>
+          </html>
+          EOF
+
+          INDEX_PATH="functional/index.html"
+          INDEX_CONTENT="$(python3 -c 'import base64;print(base64.b64encode(open("/tmp/rustfs-functional-index.html","rb").read()).decode())')"
+          INDEX_SHA="$(gh api "repos/rustfs/dashboard/contents/${INDEX_PATH}" -q '.sha' 2>/dev/null || true)"
+          if [ -n "${INDEX_SHA}" ]; then
+            jq -n --arg msg "functional ui update" --arg content "${INDEX_CONTENT}" --arg sha "${INDEX_SHA}" \
+              '{message:$msg, content:$content, sha:$sha}' \
+              | gh api --method PUT "repos/rustfs/dashboard/contents/${INDEX_PATH}" --input - >/dev/null
+          else
+            jq -n --arg msg "functional ui init" --arg content "${INDEX_CONTENT}" \
+              '{message:$msg, content:$content}' \
+              | gh api --method PUT "repos/rustfs/dashboard/contents/${INDEX_PATH}" --input - >/dev/null
+          fi
+
       - name: Upload report and logs
         if: always()
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6

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

@@ -38,6 +38,7 @@ env:
 jobs:
   tier-test:
     runs-on: smoke-testing
+    continue-on-error: true
     timeout-minutes: 420
     if: ${{ github.event_name == 'workflow_dispatch' || github.event.workflow_run.conclusion == 'success' }}
     steps:
@@ -60,6 +61,8 @@ jobs:
       - name: Cleanup environment (before)
         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
@@ -77,18 +80,35 @@ jobs:
 
       - name: Ensure MQTT broker + clients
         run: |
+          set -euo pipefail
           if ! command -v mosquitto_sub >/dev/null 2>&1; then
             sudo apt-get update
-            sudo apt-get install -y mosquitto mosquitto-clients
+            sudo apt-get install -y mosquitto-clients
           fi
-          sudo mkdir -p /etc/mosquitto/conf.d
-          printf 'listener 1883 0.0.0.0\nallow_anonymous true\n' | sudo tee /etc/mosquitto/conf.d/rustfs-test.conf >/dev/null
-          sudo systemctl restart mosquitto
-          sleep 2
-          ss -tln 2>/dev/null | grep -q ':1883' || { echo 'mosquitto not listening on 1883'; exit 1; }
+          command -v docker >/dev/null 2>&1 || { echo 'docker not found on runner'; exit 1; }
+          sudo docker rm -f rustfs-test-mqtt >/dev/null 2>&1 || true
+          cat <<'EOF' | sudo tee /tmp/rustfs-mosquitto.conf >/dev/null
+          listener 1883 0.0.0.0
+          allow_anonymous true
+          EOF
+          sudo docker run -d --name rustfs-test-mqtt -p 1883:1883 \
+            -v /tmp/rustfs-mosquitto.conf:/mosquitto/config/mosquitto.conf:ro \
+            eclipse-mosquitto:2 >/dev/null
+          for _ in {1..10}; do
+            if ss -tln 2>/dev/null | grep -q ':1883'; then
+              break
+            fi
+            sleep 1
+          done
+          ss -tln 2>/dev/null | grep -q ':1883' || {
+            echo 'mosquitto container is not listening on 1883'
+            sudo docker logs rustfs-test-mqtt || true
+            exit 1
+          }
 
       - name: Run tier suite
         id: test
+        continue-on-error: true
         env:
           LOG_FILE: /tmp/rustfs-tier.log
         run: |
@@ -122,12 +142,65 @@ jobs:
           else
             PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
           fi
+          CASE_TABLE="/tmp/rustfs-tier-cases.md"
+          python3 - "${LOG_FILE}" "${CASE_TABLE}" <<'PY'
+          import re
+          import sys
+
+          log_file, out_file = sys.argv[1], sys.argv[2]
+          ansi = re.compile(r'\x1b\[[0-9;]*m')
+          start_re = re.compile(r'^---\s+([A-Z]+-[0-9]+)\s+(.+?)\s+---$')
+          done_re = re.compile(r'^\[(PASS|FAIL|UNSUPPORTED)\]\s+([A-Z]+-[0-9]+)\b')
+
+          rows = []
+          index = {}
+          try:
+            with open(log_file, 'r', encoding='utf-8', errors='replace') as fh:
+              for raw in fh:
+                line = ansi.sub('', raw).strip()
+                m = start_re.match(line)
+                if m:
+                  case_id, name = m.group(1), m.group(2)
+                  if case_id not in index:
+                    index[case_id] = len(rows)
+                    rows.append([case_id, name, 'RUNNING'])
+                  continue
+                m = done_re.match(line)
+                if m:
+                  status, case_id = m.group(1), m.group(2)
+                  if case_id in index:
+                    rows[index[case_id]][2] = status
+                  else:
+                    rows.append([case_id, case_id, status])
+                    index[case_id] = len(rows) - 1
+          except FileNotFoundError:
+            rows = []
+
+          counts = {'PASS': 0, 'FAIL': 0, 'UNSUPPORTED': 0, 'RUNNING': 0}
+          for _, _, status in rows:
+            counts[status] = counts.get(status, 0) + 1
+
+          with open(out_file, 'w', encoding='utf-8') as out:
+            out.write('## Case Summary\n\n')
+            out.write(f"- Total: {len(rows)}\\n")
+            out.write(f"- PASS: {counts.get('PASS', 0)}\\n")
+            out.write(f"- FAIL: {counts.get('FAIL', 0)}\\n")
+            out.write(f"- UNSUPPORTED: {counts.get('UNSUPPORTED', 0)}\\n")
+            out.write('\\n')
+            out.write('| Case | Name | Status |\\n')
+            out.write('| --- | --- | --- |\\n')
+            for case_id, name, status in rows:
+              out.write(f'| {case_id} | {name} | {status} |\\n')
+          PY
           {
             echo "# RustFS tier test report"
             echo ""
             echo "- Run: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
             echo "- Trigger: ${{ github.event_name }}"
             echo "- Package: ${PACKAGE_SOURCE}"
+            echo "- Test Step Outcome: ${{ steps.test.outcome }}"
+            echo ""
+            cat "${CASE_TABLE}" || true
             echo ""
             echo "## Log tail"
             echo '```text'
@@ -136,6 +209,129 @@ jobs:
           } | tee "${REPORT_FILE}"
           cat "${REPORT_FILE}" >> "${GITHUB_STEP_SUMMARY}"
 
+      - name: Upload functional report to dashboard
+        if: always()
+        continue-on-error: true
+        env:
+          GH_TOKEN: ${{ env.PF_TESTING_GH_TOKEN }}
+          REPORT_FILE: /tmp/rustfs-tier-report.md
+          SUITE: tier
+        run: |
+          set -euo pipefail
+          if [ -z "${GH_TOKEN:-}" ]; then
+            echo "PF_TESTING_GH_TOKEN is not configured; skipping dashboard upload"
+            exit 0
+          fi
+          DATE="$(date -u +%Y-%m-%d)"
+          REPORT_PATH="functional-reports/${SUITE}/${DATE}.md"
+          CONTENT="$(python3 -c 'import base64,sys;print(base64.b64encode(open(sys.argv[1],"rb").read()).decode())' "${REPORT_FILE}")"
+          SHA="$(gh api "repos/rustfs/dashboard/contents/${REPORT_PATH}" -q '.sha' 2>/dev/null || true)"
+          if [ -n "${SHA}" ]; then
+            jq -n --arg msg "report(${SUITE}): ${DATE}" --arg content "${CONTENT}" --arg sha "${SHA}" \
+              '{message:$msg, content:$content, sha:$sha}' \
+              | gh api --method PUT "repos/rustfs/dashboard/contents/${REPORT_PATH}" --input - >/dev/null
+          else
+            jq -n --arg msg "report(${SUITE}): ${DATE}" --arg content "${CONTENT}" \
+              '{message:$msg, content:$content}' \
+              | gh api --method PUT "repos/rustfs/dashboard/contents/${REPORT_PATH}" --input - >/dev/null
+          fi
+
+          cat > /tmp/rustfs-functional-index.html <<'EOF'
+          <!doctype html>
+          <html lang="en">
+          <head>
+            <meta charset="utf-8" />
+            <meta name="viewport" content="width=device-width, initial-scale=1" />
+            <title>RustFS Functional Test Reports</title>
+            <style>
+              :root { --bg:#f4f6fb; --card:#fff; --text:#1f2937; --muted:#6b7280; --line:#e5e7eb; --accent:#0f766e; }
+              * { box-sizing: border-box; }
+              body { margin: 0; font-family: ui-sans-serif, -apple-system, Segoe UI, Helvetica, Arial, sans-serif; background: var(--bg); color: var(--text); }
+              .wrap { max-width: 980px; margin: 32px auto; padding: 0 16px; }
+              .card { background: var(--card); border: 1px solid var(--line); border-radius: 14px; padding: 20px; }
+              h1 { margin: 0 0 8px; font-size: 26px; }
+              p { margin: 0 0 14px; color: var(--muted); }
+              .tabs { display: flex; gap: 10px; margin: 14px 0 18px; flex-wrap: wrap; }
+              button { border: 1px solid var(--line); background: #fff; color: var(--text); border-radius: 10px; padding: 8px 14px; cursor: pointer; }
+              button.active { background: var(--accent); color: #fff; border-color: var(--accent); }
+              ul { list-style: none; margin: 0; padding: 0; }
+              li { padding: 10px 0; border-bottom: 1px solid var(--line); }
+              a { color: var(--accent); text-decoration: none; }
+              a:hover { text-decoration: underline; }
+            </style>
+          </head>
+          <body>
+            <div class="wrap">
+              <div class="card">
+                <h1>RustFS Functional Test Reports</h1>
+                <p>S3, KMS, Tier report tabs. Each tab lists reports by date.</p>
+                <div class="tabs" id="tabs"></div>
+                <ul id="list"></ul>
+              </div>
+            </div>
+            <script>
+              const suites = [
+                { key: 's3', label: 'S3 Compatibility' },
+                { key: 'kms', label: 'KMS' },
+                { key: 'tier', label: 'Tier' },
+              ];
+              const tabs = document.getElementById('tabs');
+              const list = document.getElementById('list');
+
+              async function loadSuite(suite) {
+                list.innerHTML = '<li>Loading...</li>';
+                const api = `https://api.github.com/repos/rustfs/dashboard/contents/functional-reports/${suite}`;
+                try {
+                  const res = await fetch(api);
+                  if (!res.ok) {
+                    list.innerHTML = '<li>No reports yet.</li>';
+                    return;
+                  }
+                  const data = await res.json();
+                  const files = data.filter(f => f.type === 'file' && f.name.endsWith('.md')).sort((a,b) => b.name.localeCompare(a.name));
+                  if (!files.length) {
+                    list.innerHTML = '<li>No reports yet.</li>';
+                    return;
+                  }
+                  list.innerHTML = files.map(f => `<li><a href="${f.html_url}" target="_blank" rel="noreferrer">${f.name.replace('.md','')}</a></li>`).join('');
+                } catch (_e) {
+                  list.innerHTML = '<li>Failed to load reports.</li>';
+                }
+              }
+
+              function setActive(key) {
+                for (const btn of tabs.querySelectorAll('button')) {
+                  btn.classList.toggle('active', btn.dataset.key === key);
+                }
+                loadSuite(key);
+              }
+
+              for (const suite of suites) {
+                const btn = document.createElement('button');
+                btn.textContent = suite.label;
+                btn.dataset.key = suite.key;
+                btn.addEventListener('click', () => setActive(suite.key));
+                tabs.appendChild(btn);
+              }
+              setActive('s3');
+            </script>
+          </body>
+          </html>
+          EOF
+
+          INDEX_PATH="functional/index.html"
+          INDEX_CONTENT="$(python3 -c 'import base64;print(base64.b64encode(open("/tmp/rustfs-functional-index.html","rb").read()).decode())')"
+          INDEX_SHA="$(gh api "repos/rustfs/dashboard/contents/${INDEX_PATH}" -q '.sha' 2>/dev/null || true)"
+          if [ -n "${INDEX_SHA}" ]; then
+            jq -n --arg msg "functional ui update" --arg content "${INDEX_CONTENT}" --arg sha "${INDEX_SHA}" \
+              '{message:$msg, content:$content, sha:$sha}' \
+              | gh api --method PUT "repos/rustfs/dashboard/contents/${INDEX_PATH}" --input - >/dev/null
+          else
+            jq -n --arg msg "functional ui init" --arg content "${INDEX_CONTENT}" \
+              '{message:$msg, content:$content}' \
+              | gh api --method PUT "repos/rustfs/dashboard/contents/${INDEX_PATH}" --input - >/dev/null
+          fi
+
       - name: Upload report and logs
         if: always()
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
@@ -150,6 +346,8 @@ jobs:
         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

crates/ecstore/src/api/mod.rs

@@ -406,7 +406,7 @@ pub mod notification {
     pub use crate::services::notification_sys::{
         CrossPoolFenceFleetProofToken, NotificationPeerErr, NotificationSys, ScannerPublicationLeaseGrant,
         acquire_cross_pool_fence_fleet_proof, cross_pool_fence_fleet_proof_matches, get_global_notification_sys,
-        new_global_notification_sys, start_remote_version_state_fleet_probe,
+        new_global_notification_sys, scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe,
     };
 }
 
@@ -416,9 +416,10 @@ pub mod object {
         GetObjectBodyCacheHookLookup, GetObjectBodySource, GetObjectReader, NamespaceLockFence, ObjectEncryptionResolver,
         ObjectInfo, ObjectLockConfigSnapshot, ObjectMutationHook, ObjectOptions, PutObjReader, QuotaAdmission,
         RangedDecompressReader, ReadEncryptionMaterial, ReadEncryptionMode, ReadEncryptionRequest,
-        SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, StreamConsumer, get_object_body_cache_plaintext_len,
-        lookup_get_object_body_cache_hook, register_get_object_body_cache_hook, register_object_mutation_hook,
-        unregister_get_object_body_cache_hook, unregister_object_mutation_hook,
+        SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, ScannerPublicationCommitScope, ScannerPublicationCommitStartError,
+        ScannerPublicationCommitState, StreamConsumer, get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook,
+        register_get_object_body_cache_hook, register_object_mutation_hook, unregister_get_object_body_cache_hook,
+        unregister_object_mutation_hook,
     };
     pub use crate::store::{
         PrepareSelectObjectSnapshotError, PreparedGetObjectReader, SelectObjectSnapshot, SelectObjectSnapshotReadError,

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -22,6 +22,7 @@ use crate::bucket::target::{self, BucketTarget, BucketTargets, Credentials};
 use crate::bucket::versioning_sys::BucketVersioningSys;
 use crate::runtime::sources as runtime_sources;
 use aws_credential_types::Credentials as SdkCredentials;
+use aws_credential_types::provider::{ProvideCredentials, error::CredentialsError, future};
 use aws_sdk_s3::config::Region as SdkRegion;
 use aws_sdk_s3::config::SharedHttpClient;
 use aws_sdk_s3::error::ProvideErrorMetadata;
@@ -77,7 +78,7 @@ use std::str::FromStr as _;
 use std::sync::Arc;
 use std::sync::OnceLock;
 use std::sync::Weak;
-use std::time::{Duration, Instant};
+use std::time::{Duration, Instant, SystemTime};
 use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 use tokio::sync::Mutex;
 use tokio::sync::RwLock;
@@ -89,6 +90,71 @@ use uuid::Uuid;
 
 const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16;
 const REDACTED_CREDENTIAL: &str = "<redacted>";
+const EXPIRED_REMOTE_TARGET_CREDENTIALS: &str = "remote target credentials have expired";
+
+#[derive(Clone)]
+struct RemoteTargetCredentialsProvider {
+    credentials: SdkCredentials,
+}
+
+impl RemoteTargetCredentialsProvider {
+    fn resolve_at(&self, now: SystemTime) -> aws_credential_types::provider::Result {
+        if self.credentials.expiry().is_some_and(|expiration| expiration <= now) {
+            return Err(CredentialsError::provider_error(std::io::Error::other(EXPIRED_REMOTE_TARGET_CREDENTIALS)));
+        }
+        Ok(self.credentials.clone())
+    }
+}
+
+impl fmt::Debug for RemoteTargetCredentialsProvider {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("RemoteTargetCredentialsProvider")
+            .field("temporary", &self.credentials.session_token().is_some())
+            .field("expiration", &self.credentials.expiry())
+            .finish()
+    }
+}
+
+impl ProvideCredentials for RemoteTargetCredentialsProvider {
+    fn provide_credentials<'a>(&'a self) -> future::ProvideCredentials<'a>
+    where
+        Self: 'a,
+    {
+        future::ProvideCredentials::ready(self.resolve_at(SystemTime::now()))
+    }
+
+    fn fallback_on_interrupt(&self) -> Option<SdkCredentials> {
+        self.resolve_at(SystemTime::now()).ok()
+    }
+}
+
+fn remote_target_sdk_credentials(
+    credentials: &Credentials,
+    account_id: &str,
+    now: SystemTime,
+) -> Result<SdkCredentials, &'static str> {
+    let session_token = credentials.effective_session_token();
+    let expiration = credentials.effective_expiration().map(SystemTime::from);
+    if expiration.is_some() && session_token.is_none() {
+        return Err("remote target credential expiration requires a session token");
+    }
+    if expiration.is_some_and(|expiration| expiration <= now) {
+        return Err(EXPIRED_REMOTE_TARGET_CREDENTIALS);
+    }
+
+    let mut builder = SdkCredentials::builder()
+        .access_key_id(credentials.access_key.clone())
+        .secret_access_key(credentials.secret_key.clone())
+        .account_id(account_id.to_string())
+        .provider_name("bucket_target_sys");
+    if let Some(session_token) = session_token {
+        builder = builder.session_token(session_token.to_string());
+    }
+    if let Some(expiration) = expiration {
+        builder = builder.expiry(expiration);
+    }
+    Ok(builder.build())
+}
 
 pub type HeadObjectSdkError = Box<SdkError<HeadObjectError>>;
 pub type GetObjectSdkError = Box<SdkError<GetObjectError>>;
@@ -845,13 +911,26 @@ impl BucketTargetSys {
         Ok(BucketTargets { targets: new_targets })
     }
 
+    async fn mark_refresh_attempt(&self, arn: &str) {
+        // Rate-limit a failed config fetch as well as a failed client build.
+        // A successful rebuild replaces this timestamp during publication.
+        self.arn_remotes_map
+            .write()
+            .await
+            .entry(arn.to_string())
+            .or_default()
+            .last_refresh = OffsetDateTime::now_utc();
+    }
+
     pub async fn mark_refresh_in_progress(&self, bucket: &str, arn: &str) {
         let mut arn_errs = self.arn_errs_map.write().await;
-        arn_errs.entry(arn.to_string()).or_insert_with(|| ArnErrs {
-            bucket: bucket.to_string(),
-            update_in_progress: true,
+        let err = arn_errs.entry(arn.to_string()).or_insert_with(|| ArnErrs {
             count: 1,
+            bucket: bucket.to_string(),
+            ..Default::default()
         });
+        err.update_in_progress = true;
+        err.bucket = bucket.to_string();
     }
 
     pub async fn mark_refresh_done(&self, bucket: &str, arn: &str) {
@@ -863,15 +942,21 @@ impl BucketTargetSys {
     }
 
     pub async fn is_reloading_target(&self, _bucket: &str, arn: &str) -> bool {
-        let arn_errs = self.arn_errs_map.read().await;
-        arn_errs.get(arn).map(|err| err.update_in_progress).unwrap_or(false)
+        self.arn_errs_map
+            .read()
+            .await
+            .get(arn)
+            .is_some_and(|err| err.update_in_progress)
     }
 
-    pub async fn inc_arn_errs(&self, _bucket: &str, arn: &str) {
+    pub async fn inc_arn_errs(&self, bucket: &str, arn: &str) {
         let mut arn_errs = self.arn_errs_map.write().await;
-        if let Some(err) = arn_errs.get_mut(arn) {
-            err.count += 1;
-        }
+        let err = arn_errs.entry(arn.to_string()).or_insert_with(|| ArnErrs {
+            bucket: bucket.to_string(),
+            ..Default::default()
+        });
+        err.count += 1;
+        err.bucket = bucket.to_string();
     }
 
     pub async fn get_remote_target_client(&self, bucket: &str, arn: &str) -> Option<Arc<TargetClient>> {
@@ -884,15 +969,15 @@ impl BucketTargetSys {
                 .unwrap_or((None, None))
         };
 
-        if let Some(cli) = cli {
+        let credentials_expired = cli
+            .as_ref()
+            .is_some_and(|client| client.credentials_expired_at(jiff::Timestamp::now()));
+        if let Some(cli) = cli
+            && !credentials_expired
+        {
             return Some(cli);
         }
 
-        // TODO(backlog): spawn an async task to proactively reload the replication target
-        if self.is_reloading_target(bucket, arn).await {
-            return None;
-        }
-
         if let Some(last_refresh) = last_refresh {
             let now = OffsetDateTime::now_utc();
             if now - last_refresh < Duration::from_secs(60 * 5) {
@@ -900,16 +985,24 @@ impl BucketTargetSys {
             }
         }
 
+        // The existing per-bucket publication lock is also the reload claim:
+        // try-locking keeps the request path non-blocking, is cancellation-safe,
+        // and prevents a stale reload from publishing after a credential update.
+        let update_mutex = self.target_update_mutex(bucket).await;
+        let Ok(update_guard) = update_mutex.try_lock() else {
+            return None;
+        };
+        self.mark_refresh_attempt(arn).await;
+
         match get_bucket_targets_config(bucket).await {
             Ok(bucket_targets) => {
-                self.mark_refresh_in_progress(bucket, arn).await;
-                self.update_all_targets(bucket, Some(&bucket_targets)).await;
-                self.mark_refresh_done(bucket, arn).await;
+                self.update_all_targets_locked(bucket, Some(&bucket_targets)).await;
             }
             Err(e) => {
                 error!("get bucket targets config error:{}", e);
             }
         };
+        drop(update_guard);
 
         let cli = self
             .arn_remotes_map
@@ -917,8 +1010,10 @@ impl BucketTargetSys {
             .await
             .get(arn)
             .and_then(|target| target.client.clone());
-        if cli.is_some() {
-            return cli;
+        if let Some(cli) = cli
+            && !cli.credentials_expired_at(jiff::Timestamp::now())
+        {
+            return Some(cli);
         }
 
         self.inc_arn_errs(bucket, arn).await;
@@ -948,12 +1043,13 @@ impl BucketTargetSys {
             });
         };
 
-        let creds = SdkCredentials::builder()
-            .access_key_id(credentials.access_key.clone())
-            .secret_access_key(credentials.secret_key.clone())
-            .account_id(target.reset_id.clone())
-            .provider_name("bucket_target_sys")
-            .build();
+        let creds = remote_target_sdk_credentials(credentials, &target.reset_id, SystemTime::now()).map_err(|error| {
+            BucketTargetError::RemoteTargetConnectionErr {
+                bucket: target.target_bucket.clone(),
+                access_key: credentials.access_key.clone(),
+                error: error.to_string(),
+            }
+        })?;
 
         let endpoint = if target.secure {
             format!("https://{}", target.endpoint)
@@ -973,7 +1069,7 @@ impl BucketTargetSys {
 
         let mut config_builder = S3Config::builder()
             .endpoint_url(endpoint.clone())
-            .credentials_provider(SharedCredentialsProvider::new(creds))
+            .credentials_provider(SharedCredentialsProvider::new(RemoteTargetCredentialsProvider { credentials: creds }))
             .region(SdkRegion::new(target.region.clone()))
             .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest());
 
@@ -1047,6 +1143,13 @@ impl BucketTargetSys {
         let update_mutex = self.target_update_mutex(bucket).await;
         let _update_guard = update_mutex.lock().await;
 
+        self.update_all_targets_locked(bucket, targets).await;
+    }
+
+    /// Builds and publishes one bucket snapshot while its update mutex is held.
+    /// Keeping persisted-config reads under the same mutex prevents a stale
+    /// reload from overwriting a concurrent credential rotation.
+    async fn update_all_targets_locked(&self, bucket: &str, targets: Option<&BucketTargets>) {
         let mut clients = Vec::new();
         if let Some(new_targets) = targets {
             for target in &new_targets.targets {
@@ -1078,6 +1181,17 @@ impl BucketTargetSys {
             && !new_targets.is_empty()
         {
             for (target, client) in clients {
+                // Keep a timestamped placeholder for configured targets whose
+                // client cannot be built. Replication records these attempts as
+                // failed, while the placeholder prevents every object from
+                // triggering another metadata reload/client build for five minutes.
+                arn_remotes_map.insert(
+                    target.arn.clone(),
+                    ArnTarget {
+                        client: None,
+                        last_refresh: OffsetDateTime::now_utc(),
+                    },
+                );
                 match client {
                     Ok(client) => {
                         arn_remotes_map.insert(
@@ -1090,11 +1204,6 @@ impl BucketTargetSys {
                         health_map.insert(client.arn.clone(), target_health(&client));
                         self.update_bandwidth_limit(bucket, &target.arn, target.bandwidth_limit);
                     }
-                    // The target stays in `targets_map`, so it keeps showing up in
-                    // `bucket remote ls` while no client exists to replicate through it —
-                    // replication then drops every object for this ARN. Without this the
-                    // rejection (loopback endpoint, bad CA, unparseable URL) left no trace
-                    // anywhere.
                     Err(err) => warn!(
                         bucket = %bucket,
                         arn = %target.arn,
@@ -1962,6 +2071,13 @@ pub struct TargetClient {
 }
 
 impl TargetClient {
+    fn credentials_expired_at(&self, now: jiff::Timestamp) -> bool {
+        self.credentials
+            .as_ref()
+            .and_then(Credentials::effective_expiration)
+            .is_some_and(|expiration| expiration <= now)
+    }
+
     pub fn to_url(&self) -> Url {
         Url::parse(&self.endpoint).unwrap()
     }
@@ -2557,6 +2673,26 @@ mod tests {
         }
     }
 
+    #[derive(Clone, Debug)]
+    struct RecordingAuthConnector {
+        signed_requests: Arc<std::sync::Mutex<Vec<(bool, bool)>>>,
+    }
+
+    impl SmithyHttpConnector for RecordingAuthConnector {
+        fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
+            let has_expected_token = request.headers().get("x-amz-security-token") == Some("temporary-session-token");
+            let has_authorization = request.headers().contains_key("authorization");
+            self.signed_requests
+                .lock()
+                .expect("recorded auth request lock should not be poisoned")
+                .push((has_expected_token, has_authorization));
+            HttpConnectorFuture::ready(Ok(HttpResponse::new(
+                aws_smithy_runtime_api::http::StatusCode::try_from(200_u16).expect("200 should be a valid response status"),
+                SdkBody::empty(),
+            )))
+        }
+    }
+
     fn recording_target_client() -> (TargetClient, Arc<std::sync::Mutex<Vec<String>>>) {
         let request_uris = Arc::new(std::sync::Mutex::new(Vec::new()));
         let connector = SharedHttpConnector::new(RecordingHttpConnector {
@@ -2582,6 +2718,150 @@ mod tests {
         )
     }
 
+    #[test]
+    fn remote_target_sdk_credentials_preserve_temporary_credential_fields() {
+        let now = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000);
+        let expiration = SystemTime::UNIX_EPOCH + Duration::from_secs(2_000);
+        let credentials = Credentials {
+            access_key: "access".to_string(),
+            secret_key: "secret".to_string(),
+            session_token: Some("temporary-session-token".to_string()),
+            expiration: Some(jiff::Timestamp::try_from(expiration).expect("test expiration should convert")),
+        };
+
+        let sdk_credentials =
+            remote_target_sdk_credentials(&credentials, "account", now).expect("unexpired temporary credentials should build");
+
+        assert_eq!(sdk_credentials.session_token(), Some("temporary-session-token"));
+        assert_eq!(sdk_credentials.expiry(), Some(expiration));
+        assert_eq!(sdk_credentials.account_id().map(|id| id.as_str()), Some("account"));
+    }
+
+    #[test]
+    fn remote_target_sdk_credentials_normalize_go_zero_expiration() {
+        let credentials = Credentials {
+            access_key: "access".to_string(),
+            secret_key: "secret".to_string(),
+            session_token: None,
+            expiration: Some("0001-01-01T00:00:00Z".parse().expect("Go zero time should parse")),
+        };
+
+        let sdk_credentials = remote_target_sdk_credentials(&credentials, "", SystemTime::now())
+            .expect("Go zero expiration should remain compatible with static credentials");
+
+        assert!(sdk_credentials.session_token().is_none());
+        assert!(sdk_credentials.expiry().is_none());
+    }
+
+    #[test]
+    fn remote_target_sdk_credentials_reject_invalid_expiration_boundaries() {
+        let expiration = SystemTime::UNIX_EPOCH + Duration::from_secs(2_000);
+        let mut credentials = Credentials {
+            access_key: "access".to_string(),
+            secret_key: "secret".to_string(),
+            session_token: None,
+            expiration: Some(jiff::Timestamp::try_from(expiration).expect("test expiration should convert")),
+        };
+
+        assert_eq!(
+            remote_target_sdk_credentials(&credentials, "", SystemTime::UNIX_EPOCH + Duration::from_secs(1_000))
+                .expect_err("expiration without a session token must fail"),
+            "remote target credential expiration requires a session token"
+        );
+
+        credentials.session_token = Some("temporary-session-token".to_string());
+        assert_eq!(
+            remote_target_sdk_credentials(&credentials, "", expiration)
+                .expect_err("credentials expire at the exact expiration boundary"),
+            EXPIRED_REMOTE_TARGET_CREDENTIALS
+        );
+    }
+
+    #[test]
+    fn remote_target_credentials_provider_fails_closed_after_expiration() {
+        let expiration = SystemTime::UNIX_EPOCH + Duration::from_secs(2_000);
+        let provider = RemoteTargetCredentialsProvider {
+            credentials: SdkCredentials::new(
+                "access",
+                "secret",
+                Some("temporary-session-token".to_string()),
+                Some(expiration),
+                "test",
+            ),
+        };
+
+        assert!(provider.resolve_at(expiration - Duration::from_nanos(1)).is_ok());
+        let err = provider
+            .resolve_at(expiration)
+            .expect_err("expired credentials must not be returned");
+        assert_eq!(err.source().map(ToString::to_string).as_deref(), Some(EXPIRED_REMOTE_TARGET_CREDENTIALS));
+        assert!(!format!("{provider:?}").contains("temporary-session-token"));
+        assert!(!format!("{provider:?}").contains("secret"));
+    }
+
+    #[test]
+    fn target_client_detects_expiration_for_cache_refresh() {
+        let expiration: jiff::Timestamp = "2099-01-01T00:00:00Z".parse().expect("expiration should parse");
+        let (mut client, _) = recording_target_client();
+        client.credentials = Some(Credentials {
+            access_key: "access".to_string(),
+            secret_key: "secret".to_string(),
+            session_token: Some("temporary-session-token".to_string()),
+            expiration: Some(expiration),
+        });
+
+        assert!(!client.credentials_expired_at("2098-12-31T23:59:59Z".parse().expect("pre-expiration timestamp should parse")));
+        assert!(client.credentials_expired_at(expiration));
+
+        client.credentials.as_mut().expect("credentials should exist").expiration =
+            Some("0001-01-01T00:00:00Z".parse().expect("Go zero time should parse"));
+        assert!(!client.credentials_expired_at(jiff::Timestamp::now()));
+    }
+
+    #[tokio::test]
+    async fn temporary_credentials_add_security_token_to_sigv4_requests() {
+        let signed_requests = Arc::new(std::sync::Mutex::new(Vec::new()));
+        let connector = SharedHttpConnector::new(RecordingAuthConnector {
+            signed_requests: Arc::clone(&signed_requests),
+        });
+        let http_client = http_client_fn(move |_settings, _components| connector.clone());
+        let credentials = Credentials {
+            access_key: "access".to_string(),
+            secret_key: "secret".to_string(),
+            session_token: Some("temporary-session-token".to_string()),
+            expiration: Some("2099-01-01T00:00:00Z".parse().expect("future expiration should parse")),
+        };
+        let sdk_credentials = remote_target_sdk_credentials(&credentials, "", SystemTime::now())
+            .expect("unexpired temporary credentials should build");
+        let client = S3Client::from_conf(
+            S3Config::builder()
+                .endpoint_url("https://target.example")
+                .credentials_provider(SharedCredentialsProvider::new(RemoteTargetCredentialsProvider {
+                    credentials: sdk_credentials,
+                }))
+                .region(SdkRegion::new("us-east-1"))
+                .http_client(http_client)
+                .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
+                .build(),
+        );
+
+        client
+            .head_bucket()
+            .bucket("target-bucket")
+            .send()
+            .await
+            .expect("recording connector should accept the signed request");
+
+        assert_eq!(
+            signed_requests
+                .lock()
+                .expect("recorded auth request lock should not be poisoned")
+                .as_slice(),
+            &[(true, true)],
+            "SigV4 request must include both authorization and the session-token header"
+        );
+    }
+
     fn spawn_https_server(cert: &rcgen::CertifiedKey<rcgen::KeyPair>, requests: usize) -> (u16, std::thread::JoinHandle<()>) {
         use std::io::{Read, Write};
 
@@ -3513,6 +3793,29 @@ mod tests {
         assert!(mutexes.contains_key("second"));
     }
 
+    #[tokio::test]
+    async fn target_refresh_attempt_updates_retry_timestamp_and_error_count() {
+        let sys = BucketTargetSys::default();
+
+        sys.mark_refresh_attempt("arn:reload").await;
+        let last_refresh = sys.arn_remotes_map.read().await["arn:reload"].last_refresh;
+        assert!(OffsetDateTime::now_utc() - last_refresh < Duration::from_secs(5));
+
+        sys.inc_arn_errs("bucket", "arn:reload").await;
+        sys.inc_arn_errs("bucket", "arn:reload").await;
+        let errors = sys.arn_errs_map.read().await;
+        assert_eq!(errors["arn:reload"].count, 2);
+        assert_eq!(errors["arn:reload"].bucket, "bucket");
+        drop(errors);
+
+        sys.mark_refresh_in_progress("bucket", "arn:reload").await;
+        assert!(sys.is_reloading_target("bucket", "arn:reload").await);
+        sys.mark_refresh_done("bucket", "arn:reload").await;
+        assert!(!sys.is_reloading_target("bucket", "arn:reload").await);
+        sys.mark_refresh_in_progress("bucket", "arn:reload").await;
+        assert!(sys.is_reloading_target("bucket", "arn:reload").await);
+    }
+
     #[tokio::test]
     async fn update_all_targets_publishes_disable_proxy_on_target_client() {
         // The read-proxy selector (replication_proxy::get_proxy_targets) skips
@@ -3551,6 +3854,88 @@ mod tests {
         assert!(opted_out.disable_proxy, "disable_proxy must reach the published TargetClient");
     }
 
+    #[tokio::test]
+    async fn update_all_targets_keeps_failed_client_placeholder() {
+        let sys = BucketTargetSys::default();
+        let target = BucketTarget {
+            arn: "arn:expired".to_string(),
+            endpoint: "192.168.1.10:9000".to_string(),
+            target_bucket: "target-bucket".to_string(),
+            region: "us-east-1".to_string(),
+            credentials: Some(Credentials {
+                access_key: "access".to_string(),
+                secret_key: "secret".to_string(),
+                session_token: Some("temporary-session-token".to_string()),
+                expiration: Some("2000-01-01T00:00:00Z".parse().expect("expired timestamp should parse")),
+            }),
+            ..Default::default()
+        };
+        let targets = BucketTargets { targets: vec![target] };
+
+        sys.update_all_targets("bucket", Some(&targets)).await;
+
+        let remotes = sys.arn_remotes_map.read().await;
+        let placeholder = remotes
+            .get("arn:expired")
+            .expect("configured target should retain a cache entry");
+        assert!(placeholder.client.is_none());
+        assert!(OffsetDateTime::now_utc() - placeholder.last_refresh < Duration::from_secs(5));
+        drop(remotes);
+        assert!(sys.get_remote_target_client("bucket", "arn:expired").await.is_none());
+    }
+
+    #[tokio::test]
+    async fn credential_rotation_atomically_replaces_published_client() {
+        let sys = BucketTargetSys::default();
+        let target = |session_token: &str| BucketTarget {
+            arn: "arn:rotating".to_string(),
+            endpoint: "192.168.1.10:9000".to_string(),
+            target_bucket: "target-bucket".to_string(),
+            region: "us-east-1".to_string(),
+            credentials: Some(Credentials {
+                access_key: "access".to_string(),
+                secret_key: "secret".to_string(),
+                session_token: Some(session_token.to_string()),
+                expiration: None,
+            }),
+            ..Default::default()
+        };
+
+        sys.update_all_targets(
+            "bucket",
+            Some(&BucketTargets {
+                targets: vec![target("old-session-token")],
+            }),
+        )
+        .await;
+        let old_client = sys
+            .get_remote_target_client("bucket", "arn:rotating")
+            .await
+            .expect("initial client should be published");
+
+        sys.update_all_targets(
+            "bucket",
+            Some(&BucketTargets {
+                targets: vec![target("new-session-token")],
+            }),
+        )
+        .await;
+        let new_client = sys
+            .get_remote_target_client("bucket", "arn:rotating")
+            .await
+            .expect("rotated client should be published");
+
+        assert!(!Arc::ptr_eq(&old_client, &new_client));
+        assert_eq!(
+            old_client.credentials.as_ref().and_then(Credentials::effective_session_token),
+            Some("old-session-token")
+        );
+        assert_eq!(
+            new_client.credentials.as_ref().and_then(Credentials::effective_session_token),
+            Some("new-session-token")
+        );
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn target_updates_serialize_client_build_through_publication_per_bucket() {
         let sys = Arc::new(BucketTargetSys::default());

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

@@ -436,16 +436,21 @@ pub(crate) async fn check_replicate_delete_strict(
     }
 
     for target in decision.targets_map.values_mut() {
-        if let Some(client) = ReplicationTargetStore::remote_target_client(bucket, &target.arn).await {
-            target.synchronous = client.replicate_sync;
-        } else {
-            target.replicate = false;
-            target.synchronous = false;
-        }
+        let replicate_sync = ReplicationTargetStore::remote_target_client(bucket, &target.arn)
+            .await
+            .map(|client| client.replicate_sync);
+        apply_target_delivery_mode(target, replicate_sync);
     }
     Ok(decision)
 }
 
+fn apply_target_delivery_mode(target: &mut ReplicateTargetDecision, replicate_sync: Option<bool>) {
+    // A missing runtime client is a delivery failure, not a rule mismatch.
+    // Preserve admission and fall back to the asynchronous worker, which can
+    // persist FAILED state for the heal/retry path.
+    target.synchronous = replicate_sync.unwrap_or(false);
+}
+
 pub(crate) fn check_replicate_delete_with_snapshot(
     dobj: &ObjectToDelete,
     oi: &ObjectInfo,
@@ -629,6 +634,23 @@ mod tests {
         }));
     }
 
+    #[test]
+    fn missing_target_client_preserves_delete_admission_as_async() {
+        let mut target = ReplicateTargetDecision::new("arn:target".to_string(), true, true);
+
+        apply_target_delivery_mode(&mut target, None);
+
+        assert!(target.replicate, "a runtime client miss must not erase the replication rule decision");
+        assert!(
+            !target.synchronous,
+            "unavailable synchronous targets must fall back to the async retry path"
+        );
+
+        apply_target_delivery_mode(&mut target, Some(true));
+        assert!(target.replicate);
+        assert!(target.synchronous);
+    }
+
     #[test]
     fn must_replicate_options_preserve_request_flag() {
         let user_defined = HashMap::new();

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

@@ -102,6 +102,7 @@ const BACKGROUND_WALKDIR_TIMEOUT: TokioDuration = TokioDuration::from_secs(60);
 const ENV_REPL_RESYNC_MAX_JOBS: &str = "RUSTFS_REPL_RESYNC_MAX_JOBS";
 const DEFAULT_REPL_RESYNC_MAX_JOBS: usize = 2;
 const MAX_REPL_RESYNC_MAX_JOBS: usize = 32;
+const TARGET_CLIENT_UNAVAILABLE_ERROR: &str = "replication target client is unavailable";
 use uuid::Uuid;
 
 const EVENT_RESYNC_STATUS_UPDATE_SKIPPED: &str = "replication_resync_status_update_skipped";
@@ -1847,19 +1848,7 @@ pub(crate) async fn replicate_delete_with_outcome<S: ReplicationStorage>(
                 reason = "target_client_missing",
                 "Skipping replication delete because target client is unavailable"
             );
-            send_local_event(EventArgs {
-                event_name: EventName::ObjectReplicationNotTracked.to_string(),
-                bucket_name: bucket.clone(),
-                object: ObjectInfo {
-                    bucket: bucket.clone(),
-                    name: dobj.delete_object.object_name.clone(),
-                    version_id,
-                    delete_marker: dobj.delete_object.delete_marker,
-                    ..Default::default()
-                },
-                user_agent: "Internal: [Replication]".to_string(),
-                ..Default::default()
-            });
+            rinfos.targets.push(unavailable_delete_target_info(&dobj, &tgt_entry.arn));
             continue;
         };
 
@@ -2584,6 +2573,32 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
     all_succeeded
 }
 
+fn unavailable_delete_target_info(dobj: &DeletedObjectReplicationInfo, arn: &str) -> ReplicatedTargetInfo {
+    let mut rinfo = dobj
+        .delete_object
+        .replication_state
+        .as_ref()
+        .map(|state| state.target_state(arn))
+        .unwrap_or_else(|| ReplicatedTargetInfo {
+            arn: arn.to_string(),
+            ..Default::default()
+        });
+    rinfo.op_type = dobj.op_type;
+    if is_version_delete_replication(&dobj.delete_object) {
+        if rinfo.version_purge_status != VersionPurgeStatusType::Complete {
+            rinfo.version_purge_status = VersionPurgeStatusType::Failed;
+            rinfo.error = Some(TARGET_CLIENT_UNAVAILABLE_ERROR.to_string());
+        }
+    } else if rinfo.prev_replication_status == ReplicationStatusType::Completed && dobj.op_type != ReplicationType::ExistingObject
+    {
+        rinfo.replication_status = ReplicationStatusType::Completed;
+    } else {
+        rinfo.replication_status = ReplicationStatusType::Failed;
+        rinfo.error = Some(TARGET_CLIENT_UNAVAILABLE_ERROR.to_string());
+    }
+    rinfo
+}
+
 async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo {
     let version_id = if let Some(version_id) = &dobj.delete_object.delete_marker_version_id {
         version_id.to_owned()
@@ -2796,6 +2811,10 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
     };
 
     let mut join_set = JoinSet::new();
+    let mut rinfos = ReplicatedInfos {
+        replication_timestamp: Some(OffsetDateTime::now_utc()),
+        targets: Vec::with_capacity(tgt_arns.len()),
+    };
 
     for arn in tgt_arns {
         let Some(tgt_client) = ReplicationTargetStore::remote_target_client(&bucket, &arn).await else {
@@ -2803,7 +2822,8 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
             // stays unreachable would flood the log from the replication hot path. The
             // condition is reported once per pass by the site-replication reconciler and
             // once per rebuild by `update_all_targets`, which is where an operator can act
-            // on it; the per-object event below still records each dropped object.
+            // on it; the FAILED state below preserves retry visibility and the
+            // aggregate result emits the user-visible failure event once.
             debug!(
                 event = EVENT_RESYNC_RUNTIME_SKIPPED,
                 component = LOG_COMPONENT_ECSTORE,
@@ -2812,15 +2832,9 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
                 object = %object,
                 arn = %arn,
                 reason = "target_client_missing",
-                "Replication rule has no bucket target for its destination ARN; object not replicated"
+                "Replication target client unavailable"
             );
-            send_local_event(EventArgs {
-                event_name: EventName::ObjectReplicationNotTracked.to_string(),
-                bucket_name: bucket.clone(),
-                object: roi.to_object_info(),
-                user_agent: "Internal: [Replication]".to_string(),
-                ..Default::default()
-            });
+            rinfos.targets.push(unavailable_object_target_info(&roi, &arn));
             continue;
         };
 
@@ -2835,11 +2849,6 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
         });
     }
 
-    let mut rinfos = ReplicatedInfos {
-        replication_timestamp: Some(OffsetDateTime::now_utc()),
-        targets: Vec::with_capacity(join_set.len()),
-    };
-
     while let Some(result) = join_set.join_next().await {
         match result {
             Ok(tgt_info) => {
@@ -2945,6 +2954,23 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
     (merged_state, state_persisted)
 }
 
+fn unavailable_object_target_info(roi: &ReplicateObjectInfo, arn: &str) -> ReplicatedTargetInfo {
+    ReplicatedTargetInfo {
+        arn: arn.to_string(),
+        size: roi.actual_size,
+        replication_action: if roi.op_type == ReplicationType::Object {
+            ReplicationAction::All
+        } else {
+            ReplicationAction::Metadata
+        },
+        op_type: roi.op_type,
+        replication_status: ReplicationStatusType::Failed,
+        prev_replication_status: roi.target_replication_status(arn),
+        error: Some(TARGET_CLIENT_UNAVAILABLE_ERROR.to_string()),
+        ..Default::default()
+    }
+}
+
 trait ReplicateObjectInfoExt {
     async fn replicate_object<S: ReplicationObjectIO>(
         &self,
@@ -4157,6 +4183,88 @@ async fn replicate_multipart_parts_and_complete<S: ReplicationObjectIO>(
 #[cfg(test)]
 mod tests {
     use super::super::replication_filemeta_boundary::ReplicateTargetDecision;
+
+    #[test]
+    fn unavailable_object_target_is_persisted_as_failed() {
+        let arn = "arn:object-target";
+        let roi = ReplicateObjectInfo {
+            actual_size: 42,
+            op_type: ReplicationType::Object,
+            replication_status_internal: Some(format!("{arn}=PENDING;")),
+            ..Default::default()
+        };
+
+        let target_info = unavailable_object_target_info(&roi, arn);
+        let merged = get_replication_state(
+            &ReplicatedInfos {
+                replication_timestamp: Some(OffsetDateTime::now_utc()),
+                targets: vec![target_info.clone()],
+            },
+            &ReplicationState::default(),
+            None,
+        );
+
+        assert_eq!(target_info.replication_status, ReplicationStatusType::Failed);
+        assert_eq!(target_info.prev_replication_status, ReplicationStatusType::Pending);
+        assert_eq!(target_info.replication_action, ReplicationAction::All);
+        assert_eq!(target_info.error.as_deref(), Some(TARGET_CLIENT_UNAVAILABLE_ERROR));
+        assert_eq!(merged.targets.get(arn), Some(&ReplicationStatusType::Failed));
+    }
+
+    #[test]
+    fn unavailable_delete_target_is_failed_without_overwriting_completed_state() {
+        let arn = "arn:delete-target";
+        let mut previous_state = ReplicationState::default();
+        previous_state.targets.insert(arn.to_string(), ReplicationStatusType::Pending);
+        let mut dobj = DeletedObjectReplicationInfo {
+            delete_object: ReplicationDeletedObject {
+                delete_marker: true,
+                replication_state: Some(previous_state),
+                ..Default::default()
+            },
+            op_type: ReplicationType::Delete,
+            ..Default::default()
+        };
+
+        let failed = unavailable_delete_target_info(&dobj, arn);
+        assert_eq!(failed.replication_status, ReplicationStatusType::Failed);
+        assert_eq!(failed.prev_replication_status, ReplicationStatusType::Pending);
+        assert_eq!(failed.error.as_deref(), Some(TARGET_CLIENT_UNAVAILABLE_ERROR));
+
+        dobj.delete_object
+            .replication_state
+            .as_mut()
+            .expect("previous state should exist")
+            .targets
+            .insert(arn.to_string(), ReplicationStatusType::Completed);
+        let completed = unavailable_delete_target_info(&dobj, arn);
+        assert_eq!(completed.replication_status, ReplicationStatusType::Completed);
+        assert!(completed.error.is_none());
+    }
+
+    #[test]
+    fn unavailable_version_purge_target_is_persisted_as_failed() {
+        let arn = "arn:purge-target";
+        let mut previous_state = ReplicationState::default();
+        previous_state
+            .purge_targets
+            .insert(arn.to_string(), VersionPurgeStatusType::Pending);
+        let dobj = DeletedObjectReplicationInfo {
+            delete_object: ReplicationDeletedObject {
+                version_id: Some(Uuid::new_v4()),
+                replication_state: Some(previous_state),
+                ..Default::default()
+            },
+            op_type: ReplicationType::Delete,
+            ..Default::default()
+        };
+
+        let target_info = unavailable_delete_target_info(&dobj, arn);
+
+        assert_eq!(target_info.version_purge_status, VersionPurgeStatusType::Failed);
+        assert_eq!(target_info.error.as_deref(), Some(TARGET_CLIENT_UNAVAILABLE_ERROR));
+    }
+
     fn resync_target_state(resync_id: &str, status: ResyncStatusType, replicated_count: i64) -> TargetReplicationResyncStatus {
         TargetReplicationResyncStatus {
             resync_id: resync_id.to_string(),

crates/ecstore/src/bucket/target/bucket_target.rs

@@ -25,6 +25,8 @@ use time::OffsetDateTime;
 use url::Url;
 
 const REDACTED_CREDENTIAL: &str = "<redacted>";
+const GO_YEAR_ONE_START_UNIX_SECONDS: i64 = -62_135_596_800;
+const GO_YEAR_TWO_START_UNIX_SECONDS: i64 = -62_104_060_800;
 
 #[derive(Deserialize, Serialize, Default, Clone)]
 pub struct Credentials {
@@ -41,6 +43,26 @@ pub struct Credentials {
 }
 
 impl Credentials {
+    /// Returns the session token used for request signing.
+    ///
+    /// MinIO-compatible payloads may carry an empty token. Treat whitespace-only
+    /// values as absent without rewriting a real token, whose bytes are opaque.
+    pub fn effective_session_token(&self) -> Option<&str> {
+        self.session_token.as_deref().filter(|token| !token.trim().is_empty())
+    }
+
+    /// Returns the credential expiry after normalizing Go's zero `time.Time`.
+    ///
+    /// Go JSON encoders emit year 1 for an unset `time.Time`; persisted MinIO
+    /// target metadata can therefore contain that sentinel even for static
+    /// credentials.
+    pub fn effective_expiration(&self) -> Option<Timestamp> {
+        self.expiration.filter(|expiration| {
+            let unix_seconds = expiration.as_second();
+            !(GO_YEAR_ONE_START_UNIX_SECONDS..GO_YEAR_TWO_START_UNIX_SECONDS).contains(&unix_seconds)
+        })
+    }
+
     pub fn redacted(&self) -> Self {
         Self {
             access_key: self.access_key.clone(),
@@ -355,6 +377,24 @@ mod tests {
     use std::time::Duration;
     use time::OffsetDateTime;
 
+    #[test]
+    fn credential_effective_values_normalize_only_compatibility_sentinels() {
+        let mut credentials = Credentials {
+            access_key: "access".to_string(),
+            secret_key: "secret".to_string(),
+            session_token: Some("  ".to_string()),
+            expiration: Some("0001-01-01T08:00:00+08:00".parse().expect("Go zero time should parse")),
+        };
+
+        assert!(credentials.effective_session_token().is_none());
+        assert!(credentials.effective_expiration().is_none());
+
+        credentials.session_token = Some(" opaque token ".to_string());
+        credentials.expiration = Some("2099-01-01T00:00:00Z".parse().expect("future timestamp should parse"));
+        assert_eq!(credentials.effective_session_token(), Some(" opaque token "));
+        assert_eq!(credentials.effective_expiration(), credentials.expiration);
+    }
+
     #[test]
     fn test_bucket_target_json_deserialize() {
         let json = r#"

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -122,6 +122,14 @@ fn control_plane_failure(op: &str, bucket: Option<&str>, error_code: Option<i32>
     if error_code == Some(rustfs_protos::proto_gen::node_service::ControlPlaneErrorCode::ControlPlaneErrorNotInitialized as i32) {
         return Error::RemoteNotInitialized;
     }
+    if error_code == Some(rustfs_protos::proto_gen::node_service::ControlPlaneErrorCode::ControlPlaneErrorInvalidArgument as i32)
+    {
+        return Error::InvalidArgument(
+            "control-plane".to_string(),
+            op.to_string(),
+            error_info.unwrap_or_else(|| format!("{op}: peer rejected invalid argument without details")),
+        );
+    }
     match error_info {
         Some(msg) => Error::other(msg),
         None => peer_failure_without_details(op, bucket),
@@ -2335,6 +2343,29 @@ mod tests {
         use rustfs_protos::proto_gen::node_service::ControlPlaneErrorCode;
         assert_eq!(ControlPlaneErrorCode::ControlPlaneErrorUnspecified as i32, 0);
         assert_eq!(ControlPlaneErrorCode::ControlPlaneErrorNotInitialized as i32, 1);
+        assert_eq!(ControlPlaneErrorCode::ControlPlaneErrorInvalidArgument as i32, 2);
+    }
+
+    #[test]
+    fn control_plane_failure_preserves_typed_invalid_argument_reason() {
+        use rustfs_protos::proto_gen::node_service::ControlPlaneErrorCode;
+
+        let reason = "durable unresolved-entry recovery requires pool metadata V2 or V3";
+        let err = control_plane_failure(
+            "start_decommission",
+            None,
+            Some(ControlPlaneErrorCode::ControlPlaneErrorInvalidArgument as i32),
+            Some(reason.to_string()),
+        );
+
+        assert!(
+            matches!(
+                err,
+                Error::InvalidArgument(ref scope, ref operation, ref actual_reason)
+                    if scope == "control-plane" && operation == "start_decommission" && actual_reason == reason
+            ),
+            "forwarded validation failures must remain typed and actionable"
+        );
     }
 
     #[test]

crates/ecstore/src/core/pools.rs

@@ -176,6 +176,34 @@ fn pool_meta_v3_writer_enabled() -> bool {
     )
 }
 
+fn ensure_decommission_ledger_persistence_supported_for(
+    version: u16,
+    v2_writer_enabled: bool,
+    v3_writer_enabled: bool,
+) -> Result<()> {
+    if matches!(version, POOL_META_VERSION | POOL_META_GENERATION_VERSION) || v2_writer_enabled || v3_writer_enabled {
+        return Ok(());
+    }
+
+    Err(Error::InvalidArgument(
+        "decommission".to_string(),
+        "pool-metadata-version".to_string(),
+        format!(
+            "durable unresolved-entry recovery requires pool metadata V2 or V3; enable both {} and {} only after every reader and writer supports V2",
+            rustfs_config::ENV_POOL_META_V2_WRITE,
+            rustfs_config::ENV_POOL_META_V2_FLEET_CONFIRMED,
+        ),
+    ))
+}
+
+fn ensure_decommission_ledger_persistence_supported(pool_meta: &PoolMeta) -> Result<()> {
+    ensure_decommission_ledger_persistence_supported_for(
+        pool_meta.version,
+        pool_meta_v2_writer_enabled(),
+        pool_meta_v3_writer_enabled(),
+    )
+}
+
 #[derive(Clone, Debug)]
 pub struct DecommissionCanceler {
     operation: Arc<DecommissionOperation>,
@@ -2027,6 +2055,7 @@ pub(crate) async fn pause_pool_activation_after_durable_save<S>(pool: &Arc<S>, f
 #[cfg(test)]
 struct PoolActivationStartProbeState {
     kind: PoolActivationStartKind,
+    preflight_side_effect_attempted: std::sync::atomic::AtomicBool,
     attempted: std::sync::atomic::AtomicBool,
     notify: tokio::sync::Notify,
 }
@@ -2045,6 +2074,7 @@ impl PoolActivationStartProbe {
     pub(crate) fn install(kind: PoolActivationStartKind) -> Self {
         let state = Arc::new(PoolActivationStartProbeState {
             kind,
+            preflight_side_effect_attempted: std::sync::atomic::AtomicBool::new(false),
             attempted: std::sync::atomic::AtomicBool::new(false),
             notify: tokio::sync::Notify::new(),
         });
@@ -2061,6 +2091,14 @@ impl PoolActivationStartProbe {
             self.state.notify.notified().await;
         }
     }
+
+    pub(crate) fn preflight_side_effect_was_attempted(&self) -> bool {
+        self.state.preflight_side_effect_attempted.load(Ordering::Acquire)
+    }
+
+    pub(crate) fn activation_was_attempted(&self) -> bool {
+        self.state.attempted.load(Ordering::Acquire)
+    }
 }
 
 #[cfg(test)]
@@ -2090,6 +2128,21 @@ pub(crate) fn observe_pool_activation_start_attempt(kind: PoolActivationStartKin
     }
 }
 
+#[cfg(test)]
+fn observe_pool_activation_preflight_side_effect_attempt(kind: PoolActivationStartKind) {
+    let probes = POOL_ACTIVATION_START_PROBES
+        .get_or_init(|| std::sync::Mutex::new(Vec::new()))
+        .lock()
+        .expect("pool activation start probe should not be poisoned")
+        .iter()
+        .filter(|state| state.kind == kind)
+        .cloned()
+        .collect::<Vec<_>>();
+    for state in probes {
+        state.preflight_side_effect_attempted.store(true, Ordering::Release);
+    }
+}
+
 fn rollback_decommission_pool_meta(pool_meta: &mut PoolMeta, previous_pool_meta: &PoolMeta, indices: &[usize]) {
     publish_pool_meta_updates(pool_meta, previous_pool_meta, indices);
 }
@@ -6189,6 +6242,7 @@ impl ECStore {
     ) -> Result<()> {
         {
             let mut pool_meta = self.pool_meta.write().await;
+            ensure_decommission_ledger_persistence_supported(&pool_meta)?;
             record_decommission_unresolved_entry(&mut pool_meta, idx, generation, entry)?;
         }
         self.save_current_pool_meta(&[idx])
@@ -6339,6 +6393,7 @@ impl ECStore {
         }
 
         ensure_decommission_start_pool_states(&latest_pool_meta, indices)?;
+        ensure_decommission_ledger_persistence_supported(&latest_pool_meta)?;
 
         let previous_pool_meta = latest_pool_meta.clone();
         let first_idx = indices.first().copied();
@@ -7040,10 +7095,14 @@ impl ECStore {
         save_guard.ensure_write_safe("decommission cannot be scheduled while pool metadata requires recovery")?;
         let indices = {
             let pool_meta = self.pool_meta.read().await;
-            resumable_decommission_queue_indices(&pool_meta)
+            let indices = resumable_decommission_queue_indices(&pool_meta)
                 .into_iter()
                 .filter(|idx| indices.contains(idx))
-                .collect::<Vec<_>>()
+                .collect::<Vec<_>>();
+            if !indices.is_empty() {
+                ensure_decommission_ledger_persistence_supported(&pool_meta)?;
+            }
+            indices
         };
         if indices.is_empty() {
             return Ok(Vec::new());
@@ -9299,8 +9358,11 @@ impl ECStore {
         {
             let pool_meta = self.pool_meta.read().await;
             ensure_decommission_start_pool_states(&pool_meta, &indices)?;
+            ensure_decommission_ledger_persistence_supported(&pool_meta)?;
         }
 
+        #[cfg(test)]
+        observe_pool_activation_preflight_side_effect_attempt(PoolActivationStartKind::Decommission);
         let decom_buckets = self.get_buckets_to_decommission().await?;
 
         let mut healed_buckets = HashSet::with_capacity(decom_buckets.len());
@@ -10856,10 +10918,98 @@ mod tests {
         assert!(!is_pool_activation_fleet_proof_error(&Error::ConfigNotFound));
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn decommission_v1_start_preflights_reject_before_metadata_writes() {
+        let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await;
+        let baseline = load_pool_meta_replicas(store.pools.clone(), true)
+            .await
+            .expect("baseline pool metadata should be readable");
+        assert_eq!(baseline.meta.version, POOL_META_V1_VERSION);
+        *store.pool_meta.write().await = baseline.meta.clone();
+        let start_probe = PoolActivationStartProbe::install(PoolActivationStartKind::Decommission);
+        let err = store
+            .start_decommission(vec![0])
+            .await
+            .expect_err("the initial V1 start preflight must reject before side effects");
+        assert!(matches!(err, Error::InvalidArgument(..)));
+        assert!(
+            !start_probe.preflight_side_effect_was_attempted(),
+            "V1 rejection must precede bucket listing, healing, and metadata-bucket creation"
+        );
+        assert!(
+            !start_probe.activation_was_attempted(),
+            "V1 rejection must not enter the authoritative activation save"
+        );
+        let after_early_rejection = load_pool_meta_replicas(store.pools.clone(), true)
+            .await
+            .expect("early rejection must leave durable pool metadata readable");
+        assert_eq!(after_early_rejection.canonical, baseline.canonical);
+        assert!(
+            store
+                .pool_meta
+                .read()
+                .await
+                .pools
+                .iter()
+                .all(|pool| pool.decommission.is_none())
+        );
+        assert!(store.decommission_cancelers.read().await.iter().all(Option::is_none));
+        store
+            .ensure_pool_meta_side_effects_safe("V1 start preflight")
+            .await
+            .expect("a deterministic start rejection must not latch recovery");
+        drop(start_probe);
+
+        let err = store
+            .save_current_pool_meta_for_decommission_start(
+                &[0],
+                vec![(
+                    0,
+                    PoolSpaceInfo {
+                        free: 50,
+                        total: 100,
+                        used: 50,
+                    },
+                )],
+                Vec::new(),
+            )
+            .await
+            .expect_err("the authoritative V1 start preflight must reject before saving");
+        assert!(matches!(err, Error::InvalidArgument(..)));
+
+        let after_authoritative_rejection = load_pool_meta_replicas(store.pools.clone(), true)
+            .await
+            .expect("authoritative rejection must leave durable pool metadata readable");
+        assert_eq!(after_authoritative_rejection.canonical, baseline.canonical);
+        assert!(
+            after_authoritative_rejection
+                .meta
+                .pools
+                .iter()
+                .all(|pool| pool.decommission.is_none())
+        );
+        assert!(
+            store
+                .pool_meta
+                .read()
+                .await
+                .pools
+                .iter()
+                .all(|pool| pool.decommission.is_none())
+        );
+        assert!(store.decommission_cancelers.read().await.iter().all(Option::is_none));
+        store
+            .ensure_pool_meta_side_effects_safe("authoritative V1 start preflight")
+            .await
+            .expect("an authoritative capability rejection must not latch recovery");
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn decommission_activation_fence_loss_after_durable_save_blocks_publication() {
         let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await;
+        crate::services::rebalance::promote_test_pool_meta_to_v2(&store).await;
         let barrier = PoolActivationDurableSaveBarrier::install(&store.pools[0]);
         let start_store = Arc::clone(&store);
         let start_task = tokio::spawn(async move {
@@ -10914,6 +11064,7 @@ mod tests {
     #[serial_test::serial]
     async fn decommission_activation_adopts_canonical_commit_after_replica_failure() {
         let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await;
+        crate::services::rebalance::promote_test_pool_meta_to_v2(&store).await;
         let barrier = PoolActivationDurableSaveBarrier::install(&store.pools[0]);
         let start_store = Arc::clone(&store);
         let start_task = tokio::spawn(async move {
@@ -11226,6 +11377,35 @@ mod tests {
         assert!(pool_meta_v3_writer_enabled_for(true, true));
     }
 
+    #[test]
+    fn decommission_ledger_persistence_requires_an_observed_or_confirmed_format() {
+        for (version, v2_enabled, v3_enabled, expected) in [
+            (POOL_META_V1_VERSION, false, false, false),
+            (POOL_META_V1_VERSION, true, false, true),
+            (POOL_META_V1_VERSION, false, true, true),
+            (POOL_META_VERSION, false, false, true),
+            (super::POOL_META_GENERATION_VERSION, false, false, true),
+        ] {
+            let result = super::ensure_decommission_ledger_persistence_supported_for(version, v2_enabled, v3_enabled);
+            assert_eq!(
+                result.is_ok(),
+                expected,
+                "unexpected capability result for pool metadata version {version}"
+            );
+        }
+
+        let half_confirmed_v2 = pool_meta_v2_writer_enabled_for(true, false);
+        let half_confirmed_v3 = pool_meta_v3_writer_enabled_for(false, true);
+        let err = super::ensure_decommission_ledger_persistence_supported_for(
+            POOL_META_V1_VERSION,
+            half_confirmed_v2,
+            half_confirmed_v3,
+        )
+        .expect_err("half-enabled rollout gates must not admit decommission");
+        assert!(matches!(err, Error::InvalidArgument(..)));
+        assert!(err.to_string().contains("durable unresolved-entry recovery"));
+    }
+
     #[test]
     fn pool_meta_stale_write_rejection_metric_is_countable() {
         let recorder = metrics_util::debugging::DebuggingRecorder::new();
@@ -14494,6 +14674,113 @@ mod pools_tests {
         assert!(store.decommission_cancelers.read().await[0].is_none());
     }
 
+    #[tokio::test]
+    async fn test_v1_unresolved_ledger_rejection_keeps_live_state_and_write_gate_safe() {
+        let generation = OffsetDateTime::UNIX_EPOCH;
+        let status = decommission_test_pool_status(
+            0,
+            Some(PoolDecommissionInfo {
+                start_time: Some(generation),
+                ..Default::default()
+            }),
+        );
+        let last_update = status.last_update;
+        let store = decommission_worker_test_store(
+            PoolMeta {
+                version: POOL_META_V1_VERSION,
+                pools: vec![status],
+                ..Default::default()
+            },
+            vec![None],
+        );
+        let entry = DecommissionUnresolvedEntry {
+            bucket: "bucket-a".to_string(),
+            object: "directory/".to_string(),
+            pool_index: 0,
+            set_index: 0,
+            source_generation: generation,
+            candidate_count: 1,
+            disk_error_count: 0,
+            observed_at: generation,
+            reason: "metadata_resolution_failed".to_string(),
+        };
+
+        let err = store
+            .persist_decommission_unresolved_entry(0, generation, entry)
+            .await
+            .expect_err("V1 must reject the ledger before changing live state");
+
+        assert!(matches!(err, Error::InvalidArgument(..)));
+        let pool_meta = store.pool_meta.read().await;
+        let status = &pool_meta.pools[0];
+        assert_eq!(status.last_update, last_update);
+        assert!(
+            status
+                .decommission
+                .as_ref()
+                .expect("active decommission metadata should remain present")
+                .unresolved_entries
+                .is_empty()
+        );
+        drop(pool_meta);
+        store
+            .pool_meta_save_gate
+            .lock()
+            .await
+            .ensure_write_safe("V1 unresolved-entry preflight")
+            .expect("a deterministic capability rejection must not latch recovery");
+    }
+
+    #[tokio::test]
+    async fn test_v1_runtime_recovery_rejects_worker_but_keeps_cancel_persistable() {
+        let generation = OffsetDateTime::UNIX_EPOCH;
+        let store = decommission_worker_test_store(
+            PoolMeta {
+                version: POOL_META_V1_VERSION,
+                pools: vec![decommission_test_pool_status(
+                    0,
+                    Some(PoolDecommissionInfo {
+                        start_time: Some(generation),
+                        ..Default::default()
+                    }),
+                )],
+                ..Default::default()
+            },
+            vec![None],
+        );
+
+        let err = store
+            .reserve_decommission_routines(&CancellationToken::new(), &[0])
+            .await
+            .err()
+            .expect("V1 recovery must not install a worker that cannot persist an unresolved ledger");
+        assert!(matches!(err, Error::InvalidArgument(..)));
+        assert!(store.decommission_cancelers.read().await[0].is_none());
+
+        let save_called = Arc::new(AtomicBool::new(false));
+        store
+            .decommission_cancel_with_owner_and_save(0, None, {
+                let save_called = save_called.clone();
+                move |snapshot, _| async move {
+                    snapshot.encode_config_data_for_v2_gate(false)?;
+                    save_called.store(true, Ordering::SeqCst);
+                    Ok(())
+                }
+            })
+            .await
+            .expect("a rejected V1 recovery must remain cancelable without restart");
+
+        assert!(save_called.load(Ordering::SeqCst));
+        let pool_meta = store.pool_meta.read().await;
+        let info = pool_meta.pools[0]
+            .decommission
+            .as_ref()
+            .expect("cancel metadata should remain present");
+        assert!(info.canceled);
+        assert!(!info.failed);
+        assert!(!info.complete);
+    }
+
     #[tokio::test]
     async fn test_decommission_transition_waits_without_registered_canceler() {
         let store = decommission_worker_test_store(PoolMeta::default(), vec![None]);
@@ -17487,6 +17774,7 @@ mod pools_tests {
     #[tokio::test]
     async fn test_runtime_recovery_reserves_the_startup_resumable_queue() {
         let meta = PoolMeta {
+            version: super::POOL_META_VERSION,
             pools: vec![
                 decommission_test_pool_status(
                     0,
@@ -17544,6 +17832,7 @@ mod pools_tests {
     #[tokio::test]
     async fn test_runtime_recovery_does_not_reserve_behind_active_predecessor() {
         let meta = PoolMeta {
+            version: super::POOL_META_VERSION,
             pools: vec![
                 decommission_test_pool_status(
                     0,

crates/ecstore/src/disk/disk_store.rs

@@ -250,6 +250,40 @@ pub(crate) trait DiskStoreRenameDataExt {
         dst_volume: &str,
         dst_path: &str,
     ) -> Result<RenameDataResp>;
+
+    async fn rename_data_borrowed_with_guard(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: &FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+        external_guard: Option<Arc<dyn Send + Sync>>,
+    ) -> Result<RenameDataResp> {
+        let _ = external_guard;
+        self.rename_data_borrowed(src_volume, src_path, fi, dst_volume, dst_path)
+            .await
+    }
+}
+
+/// Run a mutation in an owned task when a caller supplied publication guard.
+/// RPC cancellation drops only the waiter; the mutation owner keeps the guard
+/// until its operation has returned, including any detached blocking syscall.
+async fn run_owned_mutation<T, F, Fut>(external_guard: Option<Arc<dyn Send + Sync>>, operation: F) -> Result<T>
+where
+    T: Send + 'static,
+    F: FnOnce() -> Fut + Send + 'static,
+    Fut: std::future::Future<Output = Result<T>> + Send + 'static,
+{
+    if external_guard.is_none() {
+        return operation().await;
+    }
+    tokio::spawn(async move {
+        let _external_guard = external_guard;
+        operation().await
+    })
+    .await
+    .map_err(|_| Error::other("owned mutation task failed"))?
 }
 
 impl DiskStoreRenameDataExt for LocalDiskWrapper {
@@ -273,6 +307,49 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper {
         )
         .await
     }
+
+    async fn rename_data_borrowed_with_guard(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: &FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+        external_guard: Option<Arc<dyn Send + Sync>>,
+    ) -> Result<RenameDataResp> {
+        let operation = self.clone();
+        let src_volume = src_volume.to_owned();
+        let src_path = src_path.to_owned();
+        let fi = fi.clone();
+        let dst_volume = dst_volume.to_owned();
+        let dst_path = dst_path.to_owned();
+        let timeout_duration = if external_guard.is_some() {
+            // A fenced mutation owns the publication guard until the storage
+            // operation returns. Timing out this waiter would cancel the
+            // LocalDisk future while a spawn_blocking namespace syscall could
+            // still be committing, reopening the movement window. The caller
+            // may drop its waiter; the owned task drains the mutation.
+            Duration::ZERO
+        } else {
+            get_max_timeout_duration()
+        };
+        run_owned_mutation(external_guard, move || async move {
+            operation
+                .track_disk_health_mutation(
+                    "rename_data",
+                    DiskMetricMutation::Write,
+                    || async {
+                        operation
+                            .disk
+                            .rename_data_borrowed(&src_volume, &src_path, &fi, &dst_volume, &dst_path)
+                            .await
+                    },
+                    timeout_duration,
+                )
+                .await
+        })
+        .await
+    }
 }
 
 pub fn get_drive_walkdir_timeout() -> Duration {
@@ -678,17 +755,20 @@ impl DiskOperationMetrics {
         let elapsed_nanos = u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX);
         let slot = &self.last_minute[(now_sec % 60) as usize];
         loop {
-            let version = slot.version.load(Ordering::Acquire);
+            // The successful CAS below is AcqRel, so it is the publication
+            // fence for the writer that owns this slot. The initial parity
+            // check does not need to acquire the slot payload.
+            let version = slot.version.load(Ordering::Relaxed);
             if !version.is_multiple_of(2) {
                 std::hint::spin_loop();
                 continue;
             }
             if slot
                 .version
-                .compare_exchange(version, version.wrapping_add(1), Ordering::AcqRel, Ordering::Acquire)
+                .compare_exchange(version, version.wrapping_add(1), Ordering::AcqRel, Ordering::Relaxed)
                 .is_ok()
             {
-                if slot.unix_sec.load(Ordering::Acquire) != now_sec {
+                if slot.unix_sec.load(Ordering::Relaxed) != now_sec {
                     slot.count.store(0, Ordering::Relaxed);
                     slot.acc_time.store(0, Ordering::Relaxed);
                     slot.unix_sec.store(now_sec, Ordering::Release);
@@ -704,14 +784,10 @@ impl DiskOperationMetrics {
     fn last_minute_snapshot(&self, now_sec: u64) -> TimedAction {
         let mut snapshot = TimedAction::default();
         for slot in &self.last_minute {
-            let version = slot.version.load(Ordering::Acquire);
-            if !version.is_multiple_of(2) {
+            let Some((slot_sec, count, acc_time)) = slot.snapshot() else {
                 continue;
-            }
-            let slot_sec = slot.unix_sec.load(Ordering::Acquire);
-            let count = slot.count.load(Ordering::Acquire);
-            let acc_time = slot.acc_time.load(Ordering::Acquire);
-            if slot.version.load(Ordering::Acquire) == version && slot_sec <= now_sec && now_sec.saturating_sub(slot_sec) < 60 {
+            };
+            if slot_sec <= now_sec && now_sec.saturating_sub(slot_sec) < 60 {
                 snapshot.count = snapshot.count.saturating_add(count);
                 snapshot.acc_time = snapshot.acc_time.saturating_add(acc_time);
             }
@@ -720,6 +796,23 @@ impl DiskOperationMetrics {
     }
 }
 
+impl TimedActionSlot {
+    fn snapshot(&self) -> Option<(u64, u64, u64)> {
+        let version = self.version.load(Ordering::Acquire);
+        if !version.is_multiple_of(2) {
+            return None;
+        }
+
+        // The first Acquire load publishes the payload written before the
+        // matching Release store. Relaxed payload loads are sufficient while
+        // the final Acquire version load validates that no writer intervened.
+        let slot_sec = self.unix_sec.load(Ordering::Relaxed);
+        let count = self.count.load(Ordering::Relaxed);
+        let acc_time = self.acc_time.load(Ordering::Relaxed);
+        (self.version.load(Ordering::Acquire) == version).then_some((slot_sec, count, acc_time))
+    }
+}
+
 pub(crate) struct DiskHealthWaitingGuard<'a> {
     health: &'a DiskHealthTracker,
 }
@@ -1097,6 +1190,37 @@ impl LocalDiskWrapper {
         )
     }
 
+    /// Run a delete under an owned coordinator task when a publication guard
+    /// is present. This keeps the guard alive if the RPC waiter is cancelled
+    /// while the local namespace mutation is still in progress.
+    pub(crate) async fn delete_with_publication_guard(
+        &self,
+        volume: &str,
+        path: &str,
+        options: DeleteOptions,
+        external_guard: Option<Arc<dyn Send + Sync>>,
+    ) -> Result<()> {
+        let operation = self.clone();
+        let volume = volume.to_owned();
+        let path = path.to_owned();
+        let timeout_duration = if external_guard.is_some() {
+            Duration::ZERO
+        } else {
+            get_max_timeout_duration()
+        };
+        run_owned_mutation(external_guard, move || async move {
+            operation
+                .track_disk_health_mutation(
+                    "delete",
+                    DiskMetricMutation::Delete,
+                    || async { operation.disk.delete(&volume, &path, options).await },
+                    timeout_duration,
+                )
+                .await
+        })
+        .await
+    }
+
     pub(crate) fn new_with_reconnect_state(
         disk: Arc<LocalDisk>,
         health_check: bool,
@@ -2247,6 +2371,44 @@ mod tests {
     };
     use tokio::io::AsyncWrite;
 
+    struct DropProbe(Arc<std::sync::atomic::AtomicUsize>);
+
+    impl Drop for DropProbe {
+        fn drop(&mut self) {
+            self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        }
+    }
+
+    #[tokio::test]
+    async fn owned_mutation_keeps_publication_guard_after_waiter_cancellation() {
+        let drops = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let guard: Arc<dyn Send + Sync> = Arc::new(DropProbe(Arc::clone(&drops)));
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let (release_tx, release_rx) = tokio::sync::oneshot::channel();
+        let (finished_tx, finished_rx) = tokio::sync::oneshot::channel();
+
+        let waiter = tokio::spawn(run_owned_mutation(Some(guard), move || async move {
+            started_tx.send(()).expect("mutation should signal start");
+            release_rx.await.expect("mutation should be released");
+            finished_tx.send(()).expect("mutation should signal completion");
+            Ok::<_, Error>(())
+        }));
+
+        started_rx.await.expect("mutation owner should start");
+        waiter.abort();
+        assert_eq!(drops.load(std::sync::atomic::Ordering::SeqCst), 0);
+
+        release_tx.send(()).expect("mutation owner should still be alive");
+        finished_rx.await.expect("mutation owner should finish");
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while drops.load(std::sync::atomic::Ordering::SeqCst) == 0 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("publication guard should be released after mutation completion");
+    }
+
     struct PendingWriter;
 
     #[test]

crates/ecstore/src/disk/local.rs

@@ -3148,9 +3148,10 @@ impl LocalIoBackend for StdBackend {
                 direct_read_copy_fault_delta: MmapPageFaultDelta,
                 blocking_task_duration: StdDuration,
                 used_direct_io: bool,
-                /// The descriptor opened by THIS call (None on a cache hit), handed
-                /// back so the async caller can index it in the fd cache.
-                opened_fd: Option<Arc<std::fs::File>>,
+                /// The descriptor and size snapshot opened by THIS call (None on a
+                /// cache hit), handed back so the async caller can index it in the
+                /// fd cache.
+                opened_fd: Option<Arc<FdCacheEntry>>,
             }
 
             enum MmapCopyReadError {
@@ -3197,12 +3198,12 @@ impl LocalIoBackend for StdBackend {
                 (cache, key, gen_at_open)
             });
             #[cfg(target_os = "linux")]
-            let cached_fd: Option<Arc<std::fs::File>> = match &fd_lookup {
+            let cached_fd: Option<Arc<FdCacheEntry>> = match &fd_lookup {
                 Some((cache, key, _)) => cache.get(key).await,
                 None => None,
             };
             #[cfg(not(target_os = "linux"))]
-            let cached_fd: Option<Arc<std::fs::File>> = None;
+            let cached_fd: Option<Arc<FdCacheEntry>> = None;
 
             let blocking_wait_start = metrics_enabled.then(std::time::Instant::now);
             let read_result = tokio::task::spawn_blocking(move || {
@@ -3224,8 +3225,15 @@ impl LocalIoBackend for StdBackend {
                 // the read below is positioned (mmap offset argument / `read_exact_at`)
                 // and never depends on the descriptor's current offset. `cached_fd` being
                 // None also marks this call as a miss for the cache-insert side-channel.
-                let (file, access_check_duration) = if let Some(cached) = cached_fd.as_ref() {
-                    (cached.as_ref().try_clone().map_err(DiskError::from)?, StdDuration::ZERO)
+                // The cached length is the metadata snapshot captured at open time;
+                // all in-place/replacement writers invalidate this entry before
+                // publishing a mutation, so cache hits avoid a redundant fstat.
+                let (file, cached_len, access_check_duration) = if let Some(cached) = cached_fd.as_ref() {
+                    (
+                        cached.file.as_ref().try_clone().map_err(DiskError::from)?,
+                        Some(cached.len),
+                        StdDuration::ZERO,
+                    )
                 } else {
                     // Measure the volume access probe only — the part-path resolution
                     // above is accounted in `path_resolve_duration` (rustfs/backlog#1801).
@@ -3236,20 +3244,27 @@ impl LocalIoBackend for StdBackend {
                             .map_err(|e| DiskError::from(to_access_error(e, DiskError::VolumeAccessDenied)))?;
                     }
                     let access_check_duration = access_check_start.map_or(StdDuration::ZERO, |started_at| started_at.elapsed());
-                    (std::fs::File::open(&file_path).map_err(DiskError::from)?, access_check_duration)
+                    (std::fs::File::open(&file_path).map_err(DiskError::from)?, None, access_check_duration)
                 };
                 let file_open_duration = file_open_start.map_or(StdDuration::ZERO, |started_at| started_at.elapsed());
 
-                let metadata_lookup_start = metrics_enabled.then(StdInstant::now);
-                // On a cache hit this fstats the cached descriptor — the inode it was
-                // opened against, which invalidation keeps current for live entries. EC
-                // shards are fixed-length, so a still-cached pre-heal length is benign.
-                let meta = file.metadata().map_err(DiskError::from)?;
-                let metadata_lookup_duration = metadata_lookup_start.map_or(StdDuration::ZERO, |started_at| started_at.elapsed());
+                let (metadata_len, metadata_lookup_duration) = if let Some(len) = cached_len {
+                    // Reuse the open-time metadata snapshot on a cache hit. The
+                    // generation fence and mutation invalidation keep this value
+                    // tied to the inode held by `file`.
+                    (len, StdDuration::ZERO)
+                } else {
+                    let metadata_lookup_start = metrics_enabled.then(StdInstant::now);
+                    let meta = file.metadata().map_err(DiskError::from)?;
+                    let duration = metadata_lookup_start.map_or(StdDuration::ZERO, |started_at| started_at.elapsed());
+                    (meta.len(), duration)
+                };
 
                 let metadata_validate_start = metrics_enabled.then(StdInstant::now);
-                if meta.len() < end_offset_u64 {
-                    return Err(MmapCopyReadError::OutOfBounds { actual_size: meta.len() });
+                if metadata_len < end_offset_u64 {
+                    return Err(MmapCopyReadError::OutOfBounds {
+                        actual_size: metadata_len,
+                    });
                 }
                 let metadata_validate_duration =
                     metadata_validate_start.map_or(StdDuration::ZERO, |started_at| started_at.elapsed());
@@ -3395,9 +3410,14 @@ impl LocalIoBackend for StdBackend {
                 // Arc; `cached_fd.is_none()` is true exactly when this call did the open.
                 // Non-Linux has no fd cache, so skip the Arc allocation there.
                 #[cfg(target_os = "linux")]
-                let opened_fd: Option<Arc<std::fs::File>> = cached_fd.is_none().then(|| Arc::new(file));
+                let opened_fd: Option<Arc<FdCacheEntry>> = cached_fd.is_none().then(|| {
+                    Arc::new(FdCacheEntry {
+                        file: Arc::new(file),
+                        len: metadata_len,
+                    })
+                });
                 #[cfg(not(target_os = "linux"))]
-                let opened_fd: Option<Arc<std::fs::File>> = None;
+                let opened_fd: Option<Arc<FdCacheEntry>> = None;
 
                 Ok::<MmapCopyReadResult, MmapCopyReadError>(MmapCopyReadResult {
                     bytes,
@@ -3520,7 +3540,7 @@ impl LocalIoBackend for StdBackend {
                     }
                 }
             }
-            // Index the freshly opened descriptor for future cache hits
+            // Index the freshly opened descriptor and metadata snapshot for future cache hits
             // (rustfs/backlog#1801). `insert_if_fresh` refuses to cache if an
             // invalidation (heal/delete/rename) bumped the generation between the
             // open snapshot and now, so a stale pre-mutation inode is never served
@@ -3872,6 +3892,18 @@ struct FdKey {
     direct: bool,
 }
 
+/// Descriptor and immutable size snapshot retained for one cached shard inode.
+///
+/// The generation fence and explicit mutation invalidation keep the snapshot
+/// tied to the inode held by `file`, allowing cache hits to avoid a repeated
+/// metadata syscall without weakening replacement/heal semantics.
+struct FdCacheEntry {
+    /// An independently cloneable descriptor for the immutable shard inode.
+    file: Arc<std::fs::File>,
+    /// File length captured together with the descriptor.
+    len: u64,
+}
+
 /// Per-disk cache of open descriptors for io_uring reads (backlog#1145).
 ///
 /// Why this exists: `pread_uring` opened the file on the blocking pool for every
@@ -3901,7 +3933,7 @@ struct FdKey {
 /// the descriptor once no in-flight read still holds it.
 #[cfg(target_os = "linux")]
 struct FdCache {
-    cache: moka::future::Cache<FdKey, Arc<std::fs::File>>,
+    cache: moka::future::Cache<FdKey, Arc<FdCacheEntry>>,
     /// Bumped by every invalidation. A miss-path open snapshots this before it
     /// opens and refuses to insert if it moved, so an fd opened before a
     /// heal/delete commit can never be resurrected into the cache after the
@@ -3931,7 +3963,7 @@ impl FdCache {
         }
     }
 
-    async fn get(&self, key: &FdKey) -> Option<Arc<std::fs::File>> {
+    async fn get(&self, key: &FdKey) -> Option<Arc<FdCacheEntry>> {
         self.cache.get(key).await
     }
 
@@ -3946,11 +3978,11 @@ impl FdCache {
     /// open bumped the generation, so a stale pre-heal/pre-delete inode is never
     /// cached. The post-insert re-check closes the tiny window where an
     /// invalidate races the insert itself, by removing the entry we just added.
-    async fn insert_if_fresh(&self, key: FdKey, file: Arc<std::fs::File>, gen_at_open: u64) {
+    async fn insert_if_fresh(&self, key: FdKey, entry: Arc<FdCacheEntry>, gen_at_open: u64) {
         if self.generation.load(Ordering::Acquire) != gen_at_open {
             return;
         }
-        self.cache.insert(key.clone(), file).await;
+        self.cache.insert(key.clone(), entry).await;
         if self.generation.load(Ordering::Acquire) != gen_at_open {
             self.cache.invalidate(&key).await;
         }
@@ -3986,7 +4018,7 @@ impl FdCache {
         self.generation.fetch_add(1, Ordering::AcqRel);
         let volume = volume.to_owned();
         let prefix = prefix.trim_end_matches('/').to_owned();
-        let matches = move |k: &FdKey, _: &Arc<std::fs::File>| {
+        let matches = move |k: &FdKey, _: &Arc<FdCacheEntry>| {
             k.volume == volume && (k.path == prefix || k.path.strip_prefix(&prefix).is_some_and(|r| r.starts_with('/')))
         };
         if self.cache.invalidate_entries_if(matches).is_err() {
@@ -4002,7 +4034,7 @@ impl FdCache {
     fn invalidate_volume(&self, volume: &str) {
         self.generation.fetch_add(1, Ordering::AcqRel);
         let volume = volume.to_owned();
-        let matches = move |k: &FdKey, _: &Arc<std::fs::File>| k.volume == volume;
+        let matches = move |k: &FdKey, _: &Arc<FdCacheEntry>| k.volume == volume;
         if self.cache.invalidate_entries_if(matches).is_err() {
             self.cache.invalidate_all();
         }
@@ -4020,7 +4052,8 @@ impl FdCache {
     /// tests that drive the cache directly.
     #[cfg(test)]
     async fn insert(&self, key: FdKey, file: Arc<std::fs::File>) {
-        self.cache.insert(key, file).await;
+        let len = file.metadata().map(|metadata| metadata.len()).unwrap_or_default();
+        self.cache.insert(key, Arc::new(FdCacheEntry { file, len })).await;
     }
 
     #[cfg(test)]
@@ -4399,7 +4432,12 @@ impl UringBackend {
         };
 
         let file = match cached {
-            Some(file) => file,
+            Some(entry) => {
+                if entry.len < u64::try_from(end_offset).map_err(|_| DiskError::FileCorrupt)? {
+                    return Err(DiskError::FileCorrupt);
+                }
+                Arc::clone(&entry.file)
+            }
             None => {
                 // Snapshot the cache generation BEFORE opening (rustfs/backlog#1176):
                 // if a heal/delete invalidation runs while this open is in flight,
@@ -4409,7 +4447,7 @@ impl UringBackend {
                 let root = self.root.clone();
                 let volume_owned = volume.to_owned();
                 let path_owned = path.to_owned();
-                let file = tokio::task::spawn_blocking(move || -> Result<std::fs::File> {
+                let (file, len) = tokio::task::spawn_blocking(move || -> Result<(std::fs::File, u64)> {
                     let file_path = resolve_uring_object_path(&root, &volume_owned, &path_owned)?;
                     let file = std::fs::File::open(&file_path).map_err(DiskError::from)?;
                     let meta = file.metadata().map_err(DiskError::from)?;
@@ -4417,30 +4455,22 @@ impl UringBackend {
                     if meta.len() < end_offset_u64 {
                         return Err(DiskError::FileCorrupt);
                     }
-                    Ok(file)
+                    Ok((file, meta.len()))
                 })
                 .await
                 .map_err(|e| DiskError::other(format!("uring pread join error: {e}")))??;
-                let file = Arc::new(file);
+                let file = Arc::new(FdCacheEntry {
+                    file: Arc::new(file),
+                    len,
+                });
                 if let (Some((cache, key)), Some(gen_at_open)) = (cache_entry, gen_at_open) {
                     cache.insert_if_fresh(key, Arc::clone(&file), gen_at_open).await;
                 }
-                file
+                file.file.clone()
             }
         };
 
         if length == 0 {
-            // Parity with StdBackend and the miss path (rustfs/backlog#1173): a
-            // zero-length read still rejects an offset past EOF. The miss path
-            // validated `meta.len() < end_offset` (end_offset == offset here), but
-            // a cache hit skipped it — so fstat the descriptor and match. This is
-            // a rare path (callers do not issue zero-length reads), so the one
-            // extra fstat is negligible.
-            match file.metadata() {
-                Ok(meta) if offset_u64 > meta.len() => return Err(DiskError::FileCorrupt),
-                Ok(_) => {}
-                Err(e) => return Err(DiskError::from(e)),
-            }
             return Ok(Bytes::new());
         }
 
@@ -21407,11 +21437,10 @@ mod test {
 
     /// Zero-length read bounds parity on the cache-HIT path (backlog#1173/#1180).
     /// A `length == 0` read past EOF must be rejected identically whether the
-    /// descriptor is freshly opened (miss path) or served from the cache: the
-    /// cache-hit branch fstats the descriptor to reproduce the miss path's
-    /// `offset > len` check instead of returning empty unconditionally. Seeds
-    /// the cache with a normal read so the zero-length reads are hits, then pins
-    /// that UringBackend and StdBackend agree on every case.
+    /// descriptor is freshly opened (miss path) or served from the cache. Seeds
+    /// the cache with a normal read so the zero-length reads reuse the same
+    /// open-time size snapshot, then pins that UringBackend and StdBackend agree
+    /// on every case.
     #[cfg(target_os = "linux")]
     #[tokio::test(flavor = "multi_thread")]
     async fn uring_zero_length_read_bounds_match_std_on_cache_hit() {

crates/ecstore/src/disk/mod.rs

@@ -677,15 +677,20 @@ impl DiskAPI for Disk {
 }
 
 impl Disk {
-    pub(crate) async fn delete_with_scanner_publication_lease(
+    pub async fn delete_with_scanner_publication_lease_and_guard(
         &self,
         volume: &str,
         path: &str,
         opts: DeleteOptions,
         scanner_publication_lease_token: Option<Uuid>,
+        external_guard: Option<Arc<dyn Send + Sync>>,
     ) -> Result<()> {
         match self {
-            Disk::Local(local_disk) => local_disk.delete(volume, path, opts).await,
+            Disk::Local(local_disk) => {
+                local_disk
+                    .delete_with_publication_guard(volume, path, opts, external_guard)
+                    .await
+            }
             Disk::Remote(remote_disk) => {
                 remote_disk
                     .delete_with_scanner_publication_lease(volume, path, opts, scanner_publication_lease_token)
@@ -714,11 +719,34 @@ impl Disk {
         dst_volume: &str,
         dst_path: &str,
         scanner_publication_lease_token: Option<Uuid>,
+    ) -> Result<RenameDataResp> {
+        self.rename_data_borrowed_with_fence_and_guard(
+            src_volume,
+            src_path,
+            fi,
+            dst_volume,
+            dst_path,
+            scanner_publication_lease_token,
+            None,
+        )
+        .await
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    pub async fn rename_data_borrowed_with_fence_and_guard(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: &FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+        scanner_publication_lease_token: Option<Uuid>,
+        external_guard: Option<Arc<dyn Send + Sync>>,
     ) -> Result<RenameDataResp> {
         match self {
             Disk::Local(local_disk) => {
                 local_disk
-                    .rename_data_borrowed(src_volume, src_path, fi, dst_volume, dst_path)
+                    .rename_data_borrowed_with_guard(src_volume, src_path, fi, dst_volume, dst_path, external_guard)
                     .await
             }
             Disk::Remote(remote_disk) => {

crates/ecstore/src/object_api/types.rs

@@ -19,6 +19,9 @@ use crate::storage_api_contracts::{
         HTTPPreconditions, ObjectLockRetentionOptions, ObjectPreconditionError, ObjectPreconditionPart, ObjectPreconditionState,
     },
 };
+use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
+use tokio::sync::{Mutex, Notify, OwnedRwLockReadGuard};
+use tokio_util::sync::CancellationToken;
 
 #[derive(Clone)]
 pub struct NamespaceLockFence {
@@ -347,6 +350,337 @@ impl QuotaAdmission {
     }
 }
 
+const SCANNER_PUBLICATION_SCOPE_ADMITTED: u8 = 0;
+const SCANNER_PUBLICATION_SCOPE_IN_FLIGHT: u8 = 1;
+const SCANNER_PUBLICATION_SCOPE_COMMITTED: u8 = 2;
+const SCANNER_PUBLICATION_SCOPE_ABORTED_BEFORE_COMMIT: u8 = 3;
+const SCANNER_PUBLICATION_SCOPE_INDETERMINATE: u8 = 4;
+
+/// The terminal result of a storage-owned scanner publication mutation.
+///
+/// This state is deliberately not serialized. It is the ownership hand-off
+/// between the scanner coordinator and the storage mutation task, so a
+/// detached rename/cleanup task can retain the movement permit until it has
+/// reported a definitive result.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ScannerPublicationCommitState {
+    Admitted,
+    InFlight,
+    Committed,
+    AbortedBeforeCommit,
+    Indeterminate,
+}
+
+impl ScannerPublicationCommitState {
+    fn as_u8(self) -> u8 {
+        match self {
+            Self::Admitted => SCANNER_PUBLICATION_SCOPE_ADMITTED,
+            Self::InFlight => SCANNER_PUBLICATION_SCOPE_IN_FLIGHT,
+            Self::Committed => SCANNER_PUBLICATION_SCOPE_COMMITTED,
+            Self::AbortedBeforeCommit => SCANNER_PUBLICATION_SCOPE_ABORTED_BEFORE_COMMIT,
+            Self::Indeterminate => SCANNER_PUBLICATION_SCOPE_INDETERMINATE,
+        }
+    }
+
+    fn from_u8(value: u8) -> Self {
+        match value {
+            SCANNER_PUBLICATION_SCOPE_IN_FLIGHT => Self::InFlight,
+            SCANNER_PUBLICATION_SCOPE_COMMITTED => Self::Committed,
+            SCANNER_PUBLICATION_SCOPE_ABORTED_BEFORE_COMMIT => Self::AbortedBeforeCommit,
+            SCANNER_PUBLICATION_SCOPE_INDETERMINATE => Self::Indeterminate,
+            _ => Self::Admitted,
+        }
+    }
+
+    /// A caller may release its remote lease only after one of these states.
+    /// `Indeterminate` is intentionally excluded: the mutation may have
+    /// committed after cancellation or a transport failure.
+    pub fn permits_lease_release(self) -> bool {
+        matches!(self, Self::Committed | Self::AbortedBeforeCommit)
+    }
+}
+
+/// Why a storage-owned publication scope could not start its mutation.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ScannerPublicationCommitStartError {
+    Cancelled,
+    DeadlineExceeded,
+    AlreadyStarted,
+    Terminal,
+}
+
+struct ScannerPublicationCommitScopeInner {
+    expected_movement_epoch: u64,
+    safe_deadline: tokio::time::Instant,
+    remote_lease_tokens: Arc<[Uuid]>,
+    cancellation: CancellationToken,
+    state: AtomicU8,
+    completed: Notify,
+    /// Set once a storage mutation task has taken ownership of the scope.
+    /// The caller-side RAII guard must not classify cancellation as
+    /// indeterminate while that owner can still report a definitive result.
+    owner_attached: AtomicBool,
+    /// The permit is storage-owned rather than borrowed from the scanner
+    /// future. A detached mutation task keeps the scope alive and therefore
+    /// keeps this guard alive until it reports a terminal state.
+    movement_permit: Mutex<Option<OwnedRwLockReadGuard<()>>>,
+    lease_release_safe: Arc<AtomicBool>,
+}
+
+/// Storage-owned ownership scope for one fenced scanner metadata mutation.
+///
+/// The scope is an in-memory capability. It is intentionally carried through
+/// [`ObjectOptions`] as a hidden field and never participates in serde, object
+/// metadata, RPC wire structures, or on-disk formats.
+#[derive(Clone)]
+pub struct ScannerPublicationCommitScope {
+    inner: Arc<ScannerPublicationCommitScopeInner>,
+}
+
+/// RAII fallback for storage paths that return before their commit closure
+/// takes ownership. An in-flight scope is never guessed to be aborted: it is
+/// marked indeterminate so remote lease release remains blocked.
+pub(crate) struct ScannerPublicationCommitScopeGuard {
+    scope: Option<ScannerPublicationCommitScope>,
+}
+
+impl ScannerPublicationCommitScopeGuard {
+    pub(crate) fn new(scope: ScannerPublicationCommitScope) -> Self {
+        Self { scope: Some(scope) }
+    }
+
+    pub(crate) fn disarm(&mut self) {
+        self.scope = None;
+    }
+}
+
+impl Drop for ScannerPublicationCommitScopeGuard {
+    fn drop(&mut self) {
+        let Some(scope) = self.scope.as_ref() else {
+            return;
+        };
+        if scope.owner_attached() {
+            return;
+        }
+        match scope.state() {
+            ScannerPublicationCommitState::Admitted => {
+                let _ = scope.mark_aborted_before_commit();
+            }
+            ScannerPublicationCommitState::InFlight => {
+                let _ = scope.mark_indeterminate();
+            }
+            ScannerPublicationCommitState::Committed
+            | ScannerPublicationCommitState::AbortedBeforeCommit
+            | ScannerPublicationCommitState::Indeterminate => {}
+        }
+    }
+}
+
+impl Debug for ScannerPublicationCommitScope {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ScannerPublicationCommitScope")
+            .field("expected_movement_epoch", &self.expected_movement_epoch())
+            .field("safe_deadline", &self.safe_deadline())
+            .field("remote_lease_token_count", &self.remote_lease_tokens().len())
+            .field("state", &self.state())
+            .finish()
+    }
+}
+
+impl ScannerPublicationCommitScope {
+    /// Construct a scope after the storage layer has acquired its movement
+    /// read permit. Callers must keep the scope attached to the actual
+    /// mutation owner until [`Self::wait_for_completion`] has resolved.
+    pub(crate) fn new_storage_owned(
+        expected_movement_epoch: u64,
+        safe_deadline: tokio::time::Instant,
+        remote_lease_tokens: Vec<Uuid>,
+        movement_permit: OwnedRwLockReadGuard<()>,
+    ) -> Self {
+        Self::new_storage_owned_with_release_flag(
+            expected_movement_epoch,
+            safe_deadline,
+            remote_lease_tokens,
+            movement_permit,
+            Arc::new(AtomicBool::new(true)),
+        )
+    }
+
+    pub(crate) fn new_storage_owned_with_release_flag(
+        expected_movement_epoch: u64,
+        safe_deadline: tokio::time::Instant,
+        remote_lease_tokens: Vec<Uuid>,
+        movement_permit: OwnedRwLockReadGuard<()>,
+        lease_release_safe: Arc<AtomicBool>,
+    ) -> Self {
+        lease_release_safe.store(false, Ordering::Release);
+        Self {
+            inner: Arc::new(ScannerPublicationCommitScopeInner {
+                expected_movement_epoch,
+                safe_deadline,
+                remote_lease_tokens: remote_lease_tokens.into(),
+                cancellation: CancellationToken::new(),
+                state: AtomicU8::new(SCANNER_PUBLICATION_SCOPE_ADMITTED),
+                completed: Notify::new(),
+                owner_attached: AtomicBool::new(false),
+                movement_permit: Mutex::new(Some(movement_permit)),
+                lease_release_safe,
+            }),
+        }
+    }
+
+    pub fn expected_movement_epoch(&self) -> u64 {
+        self.inner.expected_movement_epoch
+    }
+
+    pub fn safe_deadline(&self) -> tokio::time::Instant {
+        self.inner.safe_deadline
+    }
+
+    pub fn is_expired(&self) -> bool {
+        tokio::time::Instant::now() >= self.safe_deadline()
+    }
+
+    pub fn remote_lease_tokens(&self) -> &[Uuid] {
+        &self.inner.remote_lease_tokens
+    }
+
+    pub fn cancellation_token(&self) -> CancellationToken {
+        self.inner.cancellation.clone()
+    }
+
+    pub fn is_cancelled(&self) -> bool {
+        self.inner.cancellation.is_cancelled()
+    }
+
+    /// Whether a mutation that has already begun may still enter its durable
+    /// commit boundary. The storage owner must check this immediately before
+    /// starting each irreversible fan-out/rename operation.
+    pub fn can_commit(&self) -> bool {
+        self.state() == ScannerPublicationCommitState::InFlight && !self.is_cancelled() && !self.is_expired()
+    }
+
+    /// Transfer terminal-state responsibility from the caller to a detached
+    /// storage mutation owner. Once set, dropping a scanner waiter leaves the
+    /// scope in-flight until that owner reports committed or indeterminate.
+    pub fn attach_mutation_owner(&self) {
+        self.inner.owner_attached.store(true, Ordering::Release);
+    }
+
+    fn owner_attached(&self) -> bool {
+        self.inner.owner_attached.load(Ordering::Acquire)
+    }
+
+    pub fn state(&self) -> ScannerPublicationCommitState {
+        ScannerPublicationCommitState::from_u8(self.inner.state.load(Ordering::Acquire))
+    }
+
+    /// Request cancellation without claiming that a mutation has stopped.
+    /// The owner must still report `AbortedBeforeCommit` or `Indeterminate`.
+    pub fn cancel(&self) {
+        self.inner.cancellation.cancel();
+    }
+
+    pub fn try_begin(&self) -> std::result::Result<(), ScannerPublicationCommitStartError> {
+        if self.inner.cancellation.is_cancelled() {
+            return Err(ScannerPublicationCommitStartError::Cancelled);
+        }
+        if self.is_expired() {
+            return Err(ScannerPublicationCommitStartError::DeadlineExceeded);
+        }
+        self.inner
+            .state
+            .compare_exchange(
+                SCANNER_PUBLICATION_SCOPE_ADMITTED,
+                SCANNER_PUBLICATION_SCOPE_IN_FLIGHT,
+                Ordering::AcqRel,
+                Ordering::Acquire,
+            )
+            .map(|_| ())
+            .map_err(|state| {
+                if ScannerPublicationCommitState::from_u8(state).permits_lease_release() {
+                    ScannerPublicationCommitStartError::Terminal
+                } else {
+                    ScannerPublicationCommitStartError::AlreadyStarted
+                }
+            })
+    }
+
+    pub fn mark_committed(&self) -> bool {
+        self.mark_terminal(ScannerPublicationCommitState::Committed)
+    }
+
+    pub fn mark_aborted_before_commit(&self) -> bool {
+        if self
+            .inner
+            .state
+            .compare_exchange(
+                SCANNER_PUBLICATION_SCOPE_ADMITTED,
+                SCANNER_PUBLICATION_SCOPE_ABORTED_BEFORE_COMMIT,
+                Ordering::AcqRel,
+                Ordering::Acquire,
+            )
+            .is_ok()
+        {
+            self.inner.lease_release_safe.store(true, Ordering::Release);
+            self.inner.completed.notify_waiters();
+            return true;
+        }
+        false
+    }
+
+    pub fn mark_indeterminate(&self) -> bool {
+        self.mark_terminal(ScannerPublicationCommitState::Indeterminate)
+    }
+
+    fn mark_terminal(&self, terminal: ScannerPublicationCommitState) -> bool {
+        self.inner
+            .state
+            .compare_exchange(SCANNER_PUBLICATION_SCOPE_IN_FLIGHT, terminal.as_u8(), Ordering::AcqRel, Ordering::Acquire)
+            .is_ok()
+            .then(|| {
+                if terminal.permits_lease_release() {
+                    self.inner.lease_release_safe.store(true, Ordering::Release);
+                }
+                self.inner.completed.notify_waiters()
+            })
+            .is_some()
+    }
+
+    /// Wait until the mutation owner has reported a definitive terminal
+    /// state. The permit remains owned by this scope until all scope clones are
+    /// dropped or [`Self::release_movement_permit`] is called safely.
+    pub async fn wait_for_completion(&self) -> ScannerPublicationCommitState {
+        loop {
+            let notified = self.inner.completed.notified();
+            tokio::pin!(notified);
+            notified.as_mut().enable();
+            let state = self.state();
+            if state != ScannerPublicationCommitState::Admitted && state != ScannerPublicationCommitState::InFlight {
+                return state;
+            }
+            notified.await;
+        }
+    }
+
+    /// Release the storage-owned movement permit only after a known-safe
+    /// terminal result. Returns `false` for in-flight or indeterminate work.
+    pub async fn release_movement_permit(&self) -> bool {
+        if !self.state().permits_lease_release() {
+            return false;
+        }
+        self.inner.movement_permit.lock().await.take().is_some()
+    }
+}
+
+impl Drop for ScannerPublicationCommitScopeInner {
+    fn drop(&mut self) {
+        if !ScannerPublicationCommitState::from_u8(self.state.load(Ordering::Acquire)).permits_lease_release() {
+            self.lease_release_safe.store(false, Ordering::Release);
+        }
+    }
+}
+
 #[derive(Default, Clone)]
 pub struct ObjectOptions {
     // Use the maximum parity (N/2), used when saving server configuration files
@@ -384,6 +718,11 @@ pub struct ObjectOptions {
     #[doc(hidden)]
     pub put_object_cancellation: Option<tokio_util::sync::CancellationToken>,
 
+    /// Storage-owned scanner publication capability. This field is an
+    /// in-memory hand-off only; it is never copied into object metadata.
+    #[doc(hidden)]
+    pub scanner_publication_commit_scope: Option<ScannerPublicationCommitScope>,
+
     pub data_movement: bool,
     pub raw_data_movement_read: bool,
     /// Materialize the data-movement per-part checksum sidecar for APIs that
@@ -473,6 +812,7 @@ impl std::fmt::Debug for ObjectOptions {
             .field("skip_rebalancing", &self.skip_rebalancing)
             .field("skip_free_version", &self.skip_free_version)
             .field("put_object_cancellation", &self.put_object_cancellation.is_some())
+            .field("scanner_publication_commit_scope", &self.scanner_publication_commit_scope)
             .field("data_movement", &self.data_movement)
             .field("raw_data_movement_read", &self.raw_data_movement_read)
             .field("include_part_checksums", &self.include_part_checksums)

crates/ecstore/src/services/notification_sys.rs

@@ -1962,6 +1962,12 @@ where
         .map_err(|_| Error::other(format!("scanner activity peer {host} timed out after {timeout_duration:?}")))?
 }
 
+/// Classify transport-only activity failures without treating an answered
+/// peer's application error as an outage.
+pub fn scanner_peer_transport_error_message_is_retryable(error: &str) -> bool {
+    crate::cluster::rpc::client::message_has_network_needle(error)
+}
+
 fn scanner_activity_should_retry(first_error: Option<&Error>, timed_out: bool) -> bool {
     timed_out || first_error.is_some_and(PeerRestClient::is_network_like_error)
 }

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

@@ -1667,6 +1667,8 @@ mod tests {
     async fn assert_real_activation_start_race(paused_kind: PoolActivationStartKind) {
         let (_temp_dirs, rebalance_store, decommission_store) =
             crate::services::rebalance::test_two_pool_stores_with_isolated_node_contexts(None).await;
+        crate::services::rebalance::promote_test_pool_meta_to_v2(&rebalance_store).await;
+        crate::services::rebalance::promote_test_pool_meta_to_v2(&decommission_store).await;
         let disk_stats = vec![
             DiskStat {
                 total_space: 100,

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

@@ -111,6 +111,17 @@ pub(crate) async fn test_two_pool_stores_with_isolated_node_contexts(
     test_two_pool_stores_with_contexts(rebalance_meta, true).await
 }
 
+#[cfg(test)]
+pub(crate) async fn promote_test_pool_meta_to_v2(store: &std::sync::Arc<crate::store::ECStore>) {
+    let mut pool_meta = store.pool_meta.read().await.clone();
+    pool_meta.version = crate::core::pools::POOL_META_VERSION;
+    pool_meta
+        .save(store.pools.clone())
+        .await
+        .expect("test pool metadata should be promoted to V2");
+    *store.pool_meta.write().await = pool_meta;
+}
+
 #[cfg(test)]
 async fn test_two_pool_stores_with_contexts(
     rebalance_meta: Option<RebalanceMeta>,

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -3657,6 +3657,7 @@ pub(in crate::set_disk) struct RenameTailOutcome {
 pub(in crate::set_disk) struct RenameDataFenceOptions<'a> {
     write_quorum: usize,
     scanner_publication_lease_tokens: Option<&'a HashMap<String, Uuid>>,
+    scanner_publication_commit_scope: Option<crate::object_api::ScannerPublicationCommitScope>,
 }
 
 impl<'a> RenameDataFenceOptions<'a> {
@@ -3667,8 +3668,17 @@ impl<'a> RenameDataFenceOptions<'a> {
         Self {
             write_quorum,
             scanner_publication_lease_tokens,
+            scanner_publication_commit_scope: None,
         }
     }
+
+    pub(in crate::set_disk) fn with_publication_scope(
+        mut self,
+        scanner_publication_commit_scope: Option<crate::object_api::ScannerPublicationCommitScope>,
+    ) -> Self {
+        self.scanner_publication_commit_scope = scanner_publication_commit_scope;
+        self
+    }
 }
 
 #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
@@ -3778,6 +3788,37 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal<
     }
 }
 
+async fn run_scanner_publication_delete_owner<F, Fut>(
+    scope: Option<crate::object_api::ScannerPublicationCommitScope>,
+    operation: F,
+) -> disk::error::Result<()>
+where
+    F: FnOnce() -> Fut + Send + 'static,
+    Fut: Future<Output = disk::error::Result<()>> + Send + 'static,
+{
+    if scope.is_none() {
+        return operation().await;
+    }
+    if let Some(scope) = scope.as_ref() {
+        scope.attach_mutation_owner();
+    }
+    tokio::spawn(async move {
+        let result = operation().await;
+        if let Some(scope) = scope.as_ref() {
+            if result.is_ok() {
+                let _ = scope.mark_committed();
+            } else {
+                // A failed quorum does not prove that no replica committed;
+                // keep the permit indeterminate for supervisor reconciliation.
+                let _ = scope.mark_indeterminate();
+            }
+        }
+        result
+    })
+    .await
+    .map_err(|_| DiskError::other("scanner publication delete owner failed"))?
+}
+
 impl SetDisks {
     pub(in crate::set_disk) fn default_read_quorum(&self) -> usize {
         self.set_drive_count - self.default_parity_count
@@ -3995,6 +4036,7 @@ impl SetDisks {
         let RenameDataFenceOptions {
             write_quorum,
             scanner_publication_lease_tokens,
+            scanner_publication_commit_scope: _scanner_publication_commit_scope,
         } = fence_options;
         if let Some(file_info) = disks
             .iter()
@@ -4352,6 +4394,7 @@ impl SetDisks {
         let RenameDataFenceOptions {
             write_quorum,
             scanner_publication_lease_tokens,
+            scanner_publication_commit_scope,
         } = fence_options;
         if let Some(file_info) = disks
             .iter()
@@ -4383,11 +4426,15 @@ impl SetDisks {
         let fanout_src_object = src_object.clone();
         let fanout_dst_bucket = dst_bucket.clone();
         let fanout_dst_object = dst_object.clone();
+        let fanout_publication_scope = scanner_publication_commit_scope.clone();
         // Keep one coordinator task so a cancelled caller cannot drop partially
         // completed disk mutations. Per-disk futures stay ordered in `join_all`,
         // preserving slot-indexed quorum and convergence accounting without a
         // scheduler task for every disk.
         let fanout = tokio::spawn(async move {
+            // Keep the storage-owned movement permit attached to the actual
+            // fan-out owner, even if the caller future is cancelled.
+            let _fanout_publication_scope = fanout_publication_scope;
             let successful_rename_completion_rank =
                 rustfs_io_metrics::put_stage_metrics_enabled().then(|| Arc::new(AtomicUsize::new(0)));
             let futures = fanout_disks
@@ -4401,6 +4448,7 @@ impl SetDisks {
                     let dst_object = fanout_dst_object.clone();
                     let dst_bucket = fanout_dst_bucket.clone();
                     let successful_rename_completion_rank = successful_rename_completion_rank.clone();
+                    let publication_scope = scanner_publication_commit_scope.clone();
 
                     std::panic::AssertUnwindSafe(async move {
                         // Test-only introspection guard: counts this operation as
@@ -4433,6 +4481,13 @@ impl SetDisks {
                             return Err(err);
                         }
 
+                        if let Some(scope) = publication_scope.as_ref()
+                            && !scope.can_commit()
+                        {
+                            let _ = scope.mark_indeterminate();
+                            return Err(DiskError::other("scanner publication commit scope deadline or cancellation reached"));
+                        }
+
                         let disk_wait_started = rustfs_io_metrics::put_stage_timer();
                         let result = disk
                             .rename_data_borrowed_with_fence(
@@ -5841,7 +5896,8 @@ impl SetDisks {
 
     #[cfg(test)]
     pub(in crate::set_disk) async fn delete_prefix(&self, bucket: &str, prefix: &str) -> disk::error::Result<()> {
-        self.delete_prefix_with_scanner_publication_lease(bucket, prefix, None).await
+        self.delete_prefix_with_scanner_publication_lease(bucket, prefix, None, None)
+            .await
     }
 
     /// Delete a prefix with an optional per-remote-disk scanner publication
@@ -5852,6 +5908,7 @@ impl SetDisks {
         bucket: &str,
         prefix: &str,
         scanner_publication_lease_tokens: Option<&HashMap<String, Uuid>>,
+        scanner_publication_commit_scope: Option<crate::object_api::ScannerPublicationCommitScope>,
     ) -> disk::error::Result<()> {
         let disks = self.get_disks_internal().await;
         let write_quorum = disks.len() / 2 + 1;
@@ -5860,11 +5917,21 @@ impl SetDisks {
         let mut futures = Vec::with_capacity(disks.len());
 
         for (disk_op, scanner_publication_lease_token) in disks.iter().zip(fanout_fence_tokens) {
+            let disk_op = disk_op.clone();
             let bucket = bucket.to_string();
             let prefix = prefix.to_string();
+            let scanner_publication_commit_scope = scanner_publication_commit_scope.clone();
             futures.push(async move {
                 if let Some(disk) = disk_op {
-                    disk.delete_with_scanner_publication_lease(
+                    if let Some(scope) = scanner_publication_commit_scope.as_ref()
+                        && !scope.can_commit()
+                    {
+                        return Err(DiskError::other("scanner publication delete scope cannot commit"));
+                    }
+                    let external_guard = scanner_publication_commit_scope
+                        .as_ref()
+                        .map(|scope| Arc::new(scope.clone()) as Arc<dyn Send + Sync>);
+                    disk.delete_with_scanner_publication_lease_and_guard(
                         &bucket,
                         &prefix,
                         DeleteOptions {
@@ -5873,6 +5940,7 @@ impl SetDisks {
                             ..Default::default()
                         },
                         scanner_publication_lease_token,
+                        external_guard,
                     )
                     .await
                 } else {
@@ -5881,7 +5949,10 @@ impl SetDisks {
             });
         }
 
-        Self::reduce_delete_prefix_results(join_all(futures).await, write_quorum)
+        run_scanner_publication_delete_owner(scanner_publication_commit_scope, move || async move {
+            Self::reduce_delete_prefix_results(join_all(futures).await, write_quorum)
+        })
+        .await
     }
 
     /// Scan a single disk's copy of `prefix` and decide whether it is an orphan
@@ -6809,6 +6880,63 @@ mod tests {
     use tempfile::TempDir;
     use tokio::io::AsyncReadExt;
 
+    #[tokio::test]
+    async fn scanner_delete_owner_survives_waiter_cancellation() {
+        let movement_gate = Arc::new(tokio::sync::RwLock::new(()));
+        let movement_permit = movement_gate.clone().read_owned().await;
+        let scope = crate::object_api::ScannerPublicationCommitScope::new_storage_owned(
+            7,
+            tokio::time::Instant::now() + std::time::Duration::from_secs(30),
+            Vec::new(),
+            movement_permit,
+        );
+        scope.try_begin().expect("delete scope should enter flight");
+        let scope_guard = crate::object_api::ScannerPublicationCommitScopeGuard::new(scope.clone());
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let (release_tx, release_rx) = tokio::sync::oneshot::channel();
+        let (finished_tx, finished_rx) = tokio::sync::oneshot::channel();
+
+        let waiter = tokio::spawn(run_scanner_publication_delete_owner(Some(scope.clone()), move || async move {
+            started_tx.send(()).expect("delete owner should start");
+            release_rx.await.expect("delete owner should be released");
+            finished_tx.send(()).expect("delete owner should finish");
+            Ok(())
+        }));
+        started_rx.await.expect("delete owner should run");
+        drop(scope_guard);
+        waiter.abort();
+        assert_eq!(
+            scope.state(),
+            crate::object_api::ScannerPublicationCommitState::InFlight,
+            "caller cancellation must not classify an owned delete as indeterminate"
+        );
+
+        let mut movement_writer = Box::pin(movement_gate.write_owned());
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_millis(20), &mut movement_writer)
+                .await
+                .is_err(),
+            "movement transition must remain fenced while delete owner drains"
+        );
+        release_tx.send(()).expect("delete owner should remain alive");
+        finished_rx.await.expect("delete owner should drain");
+        tokio::time::timeout(std::time::Duration::from_secs(1), async {
+            loop {
+                if scope.state() == crate::object_api::ScannerPublicationCommitState::Committed {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("delete owner should report a terminal result");
+        assert!(
+            scope.release_movement_permit().await,
+            "terminal delete should release its movement permit"
+        );
+        movement_writer.await;
+    }
+
     #[test]
     fn write_precondition_lookup_errors_fail_closed_unless_absence_is_known() {
         let create_only = HTTPPreconditions {

crates/ecstore/src/set_disk/mod.rs

@@ -3938,6 +3938,44 @@ impl SetDisks {
         owner.scanner_data_usage_publication_admission_guard().await
     }
 
+    pub async fn scanner_data_usage_publication_commit_scope(
+        &self,
+        expected_movement_epoch: u64,
+        safe_deadline: tokio::time::Instant,
+        remote_lease_tokens: Vec<Uuid>,
+    ) -> Option<crate::object_api::ScannerPublicationCommitScope> {
+        let (movement_permit, epoch) = self.scanner_data_usage_publication_admission_guard().await?;
+        if epoch != expected_movement_epoch {
+            return None;
+        }
+        Some(crate::object_api::ScannerPublicationCommitScope::new_storage_owned(
+            epoch,
+            safe_deadline,
+            remote_lease_tokens,
+            movement_permit,
+        ))
+    }
+
+    pub async fn scanner_data_usage_publication_commit_scope_with_release_flag(
+        &self,
+        expected_movement_epoch: u64,
+        safe_deadline: tokio::time::Instant,
+        remote_lease_tokens: Vec<Uuid>,
+        lease_release_safe: Arc<std::sync::atomic::AtomicBool>,
+    ) -> Option<crate::object_api::ScannerPublicationCommitScope> {
+        let (movement_permit, epoch) = self.scanner_data_usage_publication_admission_guard().await?;
+        if epoch != expected_movement_epoch {
+            return None;
+        }
+        Some(crate::object_api::ScannerPublicationCommitScope::new_storage_owned_with_release_flag(
+            epoch,
+            safe_deadline,
+            remote_lease_tokens,
+            movement_permit,
+            lease_release_safe,
+        ))
+    }
+
     /// Whether both sets' namespace-lock implementations cover the same object key.
     pub(crate) async fn shares_namespace_lock_domain(&self, other: &Self) -> bool {
         match (self.ctx.is_dist_erasure().await, other.ctx.is_dist_erasure().await) {

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

@@ -66,6 +66,7 @@ use crate::bucket::lifecycle::bucket_lifecycle_ops::LifecycleOps;
 use crate::bucket::utils::is_meta_bucketname;
 use crate::bucket::versioning::VersioningApi;
 use crate::disk::DiskAPI;
+use crate::object_api::ScannerPublicationCommitScopeGuard;
 use crate::set_disk::coding;
 use crate::set_disk::core::io_primitives::GetCodecStreamingReaderBuildOutcome;
 use crate::set_disk::mem;
@@ -272,6 +273,22 @@ const OLD_DATA_CLEANUP_RECEIPT_FILE: &str = ".rustfs-old-data-cleanup-receipt.js
 const SCANNER_PUBLICATION_LEASE_FENCE_MAX_BYTES: usize = 64 * 1024;
 const SCANNER_PUBLICATION_LEASE_FENCE_MAX_ENTRIES: usize = 256;
 
+fn begin_scanner_publication_delete_mutation(scope: Option<&crate::object_api::ScannerPublicationCommitScope>) -> Result<()> {
+    let Some(scope) = scope else {
+        return Ok(());
+    };
+    if scope.state() == crate::object_api::ScannerPublicationCommitState::Admitted {
+        scope
+            .try_begin()
+            .map_err(|_| Error::other("scanner publication delete scope cannot start"))?;
+    }
+    if !scope.can_commit() {
+        let _ = scope.mark_indeterminate();
+        return Err(StorageError::OperationCanceled);
+    }
+    Ok(())
+}
+
 fn take_scanner_publication_lease_tokens(user_defined: &mut HashMap<String, String>) -> Result<Option<HashMap<String, Uuid>>> {
     let Some(encoded) = user_defined.remove(SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY) else {
         return Ok(None);
@@ -2627,6 +2644,10 @@ impl SetDisks {
         opts: &ObjectOptions,
     ) -> Result<(ObjectInfo, Option<OldCurrentSize>)> {
         crate::hp_guard!("SetDisks::put_object");
+        let mut scope_outcome_guard = opts
+            .scanner_publication_commit_scope
+            .clone()
+            .map(ScannerPublicationCommitScopeGuard::new);
         let storage_class_config = self.storage_class_config_snapshot();
         self.invalidate_get_object_metadata_cache(bucket, object).await;
 
@@ -3397,8 +3418,16 @@ impl SetDisks {
             let commit_tmp_dir = tmp_dir.clone();
             let commit_object_lock_guard = object_lock_guard.take();
             let commit_bucket_lifecycle_guard = bucket_lifecycle_guard.take();
-            let commit_allows_early_ack = commit_object_lock_guard.is_some();
-            let detach_commit_owner = commit_allows_early_ack || commit_bucket_lifecycle_guard.is_some() || quota_mutation_fence;
+            let commit_scanner_publication_scope = opts.scanner_publication_commit_scope.clone();
+            // A scanner publication scope owns the movement permit until the
+            // complete rename fan-out drains. Keep this path synchronous so
+            // its terminal state is known before the coordinator releases
+            // remote leases.
+            let commit_allows_early_ack = commit_object_lock_guard.is_some() && commit_scanner_publication_scope.is_none();
+            let detach_commit_owner = commit_scanner_publication_scope.is_some()
+                || commit_allows_early_ack
+                || commit_bucket_lifecycle_guard.is_some()
+                || quota_mutation_fence;
             let commit_write_path_label = write_path.metric_label();
             let commit_is_versioned = opts.versioned || opts.version_suspended;
             let commit_versioned = opts.versioned;
@@ -3494,7 +3523,7 @@ impl SetDisks {
                     }
                     Ok(())
                 };
-                let pre_rename_result = if cancellation.is_some() || request_cancellation.is_some() {
+                let mut pre_rename_result = if cancellation.is_some() || request_cancellation.is_some() {
                     tokio::select! {
                         biased;
                         _ = wait_for_put_object_commit_cancellation(cancellation.as_ref(), request_cancellation.as_ref()) => {
@@ -3505,6 +3534,20 @@ impl SetDisks {
                 } else {
                     pre_rename.await
                 };
+                if pre_rename_result.is_ok()
+                    && let Some(scope) = commit_scanner_publication_scope.as_ref()
+                    && let Err(err) = scope.try_begin()
+                {
+                    let _ = scope.mark_aborted_before_commit();
+                    pre_rename_result = Err(Error::other(format!("scanner publication commit scope cannot start: {err:?}")));
+                }
+                if pre_rename_result.is_ok()
+                    && let Some(scope) = commit_scanner_publication_scope.as_ref()
+                    && !scope.can_commit()
+                {
+                    let _ = scope.mark_indeterminate();
+                    pre_rename_result = Err(StorageError::OperationCanceled);
+                }
                 if let Err(err) = pre_rename_result {
                     SetDisks::abort_quota_reservation_after_fence(
                         quota_reservation,
@@ -3540,9 +3583,17 @@ impl SetDisks {
                     crate::set_disk::core::io_primitives::RenameDataFenceOptions::new(
                         write_quorum,
                         commit_scanner_publication_lease_tokens.as_ref(),
-                    ),
+                    )
+                    .with_publication_scope(commit_scanner_publication_scope.clone()),
                 )
                 .await;
+                if let Some(scope) = commit_scanner_publication_scope.as_ref() {
+                    if rename_result.is_ok() {
+                        let _ = scope.mark_committed();
+                    } else {
+                        let _ = scope.mark_indeterminate();
+                    }
+                }
                 #[cfg(any(test, feature = "test-util"))]
                 if rename_result.is_ok() {
                     pause_put_object_commit(&commit_bucket, &commit_object, PutObjectCommitPause::AfterRenameQuorum).await;
@@ -3857,6 +3908,11 @@ impl SetDisks {
                 let _ = handoff.send(());
             }
             if detach_commit_owner {
+                if let Some(scope_outcome_guard) = scope_outcome_guard.as_mut() {
+                    // The spawned commit closure owns the scope clone and is
+                    // now responsible for its terminal outcome.
+                    scope_outcome_guard.disarm();
+                }
                 let mut cancellation = PutObjectCommitCancellation::new();
                 let child_token = cancellation.child_token();
                 let result = tokio::spawn(async move { Box::pin(commit(Some(child_token))).await })
@@ -7054,6 +7110,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
 
     #[tracing::instrument(skip(self, opts))]
     async fn delete_object(&self, bucket: &str, object: &str, mut opts: ObjectOptions) -> Result<ObjectInfo> {
+        let _scope_outcome_guard = opts
+            .scanner_publication_commit_scope
+            .clone()
+            .map(ScannerPublicationCommitScopeGuard::new);
+        let scanner_publication_commit_scope = opts.scanner_publication_commit_scope.clone();
         // Scanner cleanup carries the per-peer lease fence as transient
         // request metadata. Consume it before any delete-prefix fanout so it
         // cannot be persisted or treated as user metadata.
@@ -7148,6 +7209,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                                 }
                                 delete_request.set_skip_tier_free_version();
                             }
+                            begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
                             self.delete_object_version(bucket, object, &delete_request, false).await?;
                             if let Some((_, deleted_object)) = replication_delete {
                                 ReplicationLifecycleBridge::schedule_delete(bucket.to_string(), deleted_object).await;
@@ -7162,6 +7224,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                                 ..Default::default()
                             };
                             delete_request.set_tier_free_version_id(&Uuid::new_v4().to_string());
+                            begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
                             self.delete_object_version(bucket, object, &delete_request, false).await?;
                         }
                         for version in &versions.free_versions {
@@ -7173,10 +7236,14 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                                 ..Default::default()
                             };
                             delete_request.set_tier_free_version();
+                            begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
                             self.delete_object_version(bucket, object, &delete_request, false).await?;
                         }
                     }
                 }
+                if let Some(scope) = scanner_publication_commit_scope.as_ref() {
+                    let _ = scope.mark_committed();
+                }
                 self.invalidate_get_object_metadata_cache(bucket, object).await;
                 return Ok(ObjectInfo::default());
             }
@@ -7184,10 +7251,19 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 self.validate_bucket_incarnation(bucket, expected_incarnation_id).await?;
             }
             ensure_delete_commit_locks_held(_lock_guard.as_ref(), bucket, object, &opts)?;
-            self.delete_prefix_with_scanner_publication_lease(bucket, object, scanner_publication_lease_tokens.as_ref())
-                .await
-                .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
+            begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
+            self.delete_prefix_with_scanner_publication_lease(
+                bucket,
+                object,
+                scanner_publication_lease_tokens.as_ref(),
+                scanner_publication_commit_scope.clone(),
+            )
+            .await
+            .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
 
+            if let Some(scope) = scanner_publication_commit_scope.as_ref() {
+                let _ = scope.mark_committed();
+            }
             self.invalidate_all_get_object_metadata_cache();
             return Ok(ObjectInfo::default());
         }
@@ -7260,10 +7336,14 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 ..Default::default()
             };
             ensure_delete_commit_locks_held(_lock_guard.as_ref(), bucket, object, &opts)?;
+            begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
             self.delete_object_version(bucket, object, &dfi, false)
                 .await
                 .map_err(|e| to_object_err(e, vec![bucket, object]))?;
             self.invalidate_get_object_metadata_cache(bucket, object).await;
+            if let Some(scope) = scanner_publication_commit_scope.as_ref() {
+                let _ = scope.mark_committed();
+            }
             return Ok(ObjectInfo::from_file_info(&dfi, bucket, object, opts.versioned || opts.version_suspended));
         }
 
@@ -7337,6 +7417,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             };
 
             ensure_delete_commit_locks_held(_lock_guard.as_ref(), bucket, object, &opts)?;
+            begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
             self.delete_object_version(bucket, object, &fi, should_force_delete_marker_for_missing_version(&opts))
                 .await
                 .map_err(|e| to_object_err(e, vec![bucket, object]))?;
@@ -7348,6 +7429,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             oi.user_tags = Arc::clone(&goi.user_tags);
             oi.replication_decision = goi.replication_decision;
             self.invalidate_get_object_metadata_cache(bucket, object).await;
+            if let Some(scope) = scanner_publication_commit_scope.as_ref() {
+                let _ = scope.mark_committed();
+            }
             return Ok(oi);
         }
 
@@ -7373,6 +7457,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         }
 
         ensure_delete_commit_locks_held(_lock_guard.as_ref(), bucket, object, &opts)?;
+        begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
         self.delete_object_version(bucket, object, &dfi, opts.delete_marker)
             .await
             .map_err(|e| to_object_err(e, vec![bucket, object]))?;
@@ -7398,6 +7483,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             obj_info.delete_marker = true;
         }
         self.invalidate_get_object_metadata_cache(bucket, object).await;
+        if let Some(scope) = scanner_publication_commit_scope.as_ref() {
+            let _ = scope.mark_committed();
+        }
         Ok(obj_info)
     }
 

crates/ecstore/src/store/mod.rs

@@ -58,7 +58,7 @@ use crate::{
     core::sets::Sets,
     disk::{BUCKET_META_PREFIX, DiskOption, DiskStore, RUSTFS_META_BUCKET},
     layout::endpoints::EndpointServerPools,
-    object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader},
+    object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader, ScannerPublicationCommitScope},
 };
 use futures::future::join_all;
 use http::HeaderMap;
@@ -522,6 +522,48 @@ impl ECStore {
         Some((operation_guard, self.ctx.data_movement_operation_epoch()))
     }
 
+    /// Acquire a storage-owned scanner publication scope. Unlike the legacy
+    /// admission helper, the movement permit is owned by the returned scope
+    /// and therefore survives cancellation of the scanner coordinator while
+    /// the actual metadata mutation drains.
+    pub async fn scanner_data_usage_publication_commit_scope(
+        &self,
+        expected_movement_epoch: u64,
+        safe_deadline: tokio::time::Instant,
+        remote_lease_tokens: Vec<Uuid>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        let (movement_permit, epoch) = self.scanner_data_usage_publication_admission_guard().await?;
+        if epoch != expected_movement_epoch {
+            return None;
+        }
+        Some(ScannerPublicationCommitScope::new_storage_owned(
+            epoch,
+            safe_deadline,
+            remote_lease_tokens,
+            movement_permit,
+        ))
+    }
+
+    pub async fn scanner_data_usage_publication_commit_scope_with_release_flag(
+        &self,
+        expected_movement_epoch: u64,
+        safe_deadline: tokio::time::Instant,
+        remote_lease_tokens: Vec<Uuid>,
+        lease_release_safe: Arc<std::sync::atomic::AtomicBool>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        let (movement_permit, epoch) = self.scanner_data_usage_publication_admission_guard().await?;
+        if epoch != expected_movement_epoch {
+            return None;
+        }
+        Some(ScannerPublicationCommitScope::new_storage_owned_with_release_flag(
+            epoch,
+            safe_deadline,
+            remote_lease_tokens,
+            movement_permit,
+            lease_release_safe,
+        ))
+    }
+
     /// Capture the current publication epoch without holding the movement
     /// gate across backend I/O. Callers must re-admit the same epoch before a
     /// mutation commits.
@@ -1409,9 +1451,29 @@ mod tests {
             .await
             .expect("movement writer should proceed after lease expiry")
             .expect("expiry writer task should not panic");
+        assert!(
+            store.validate_scanner_publication_lease(expiring_token, 0).await.is_err(),
+            "an expired lease must not validate after its read guard is released"
+        );
         assert!(!store.release_scanner_publication_lease(expiring_token).await);
     }
 
+    #[tokio::test]
+    async fn scanner_publication_lease_rejects_a_new_movement_generation() {
+        let store = build_store_with_ctx(Arc::new(InstanceContext::new()));
+        let (token, generation) = store
+            .acquire_scanner_publication_lease(0, crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL)
+            .await
+            .expect("an idle store should grant a publication lease");
+
+        assert_eq!(store.ctx.advance_data_movement_generation(), Some(1));
+        assert!(
+            store.validate_scanner_publication_lease(token, generation).await.is_err(),
+            "a lease from the prior movement generation must fail closed"
+        );
+        assert!(store.release_scanner_publication_lease(token).await);
+    }
+
     #[tokio::test]
     async fn scanner_publication_lease_rejects_stale_generation_before_install() {
         let store = build_store_with_ctx(Arc::new(InstanceContext::new()));
@@ -1422,6 +1484,102 @@ mod tests {
         assert!(error.to_string().contains("generation is stale"));
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn scanner_publication_commit_scope_owns_permit_until_terminal_drain() {
+        let store = build_store_with_ctx(Arc::new(InstanceContext::new()));
+        let scope = store
+            .scanner_data_usage_publication_commit_scope(
+                0,
+                tokio::time::Instant::now() + Duration::from_secs(30),
+                vec![Uuid::new_v4()],
+            )
+            .await
+            .expect("idle storage should grant a publication scope");
+        assert_eq!(scope.state(), crate::object_api::ScannerPublicationCommitState::Admitted);
+        assert_eq!(scope.remote_lease_tokens().len(), 1);
+
+        let gate = store.ctx.data_movement_operation_gate();
+        let writer = tokio::spawn(async move { gate.write_owned().await });
+        tokio::task::yield_now().await;
+        assert!(!writer.is_finished(), "the scope must own its movement permit after the caller returns");
+
+        scope.cancel();
+        assert!(scope.mark_aborted_before_commit());
+        assert_eq!(
+            scope.wait_for_completion().await,
+            crate::object_api::ScannerPublicationCommitState::AbortedBeforeCommit
+        );
+        assert!(scope.release_movement_permit().await);
+        tokio::time::timeout(Duration::from_secs(1), writer)
+            .await
+            .expect("movement writer should proceed after the scope drains")
+            .expect("movement writer task should not panic");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn scanner_publication_commit_scope_rejects_late_start_and_keeps_indeterminate_permit() {
+        let store = build_store_with_ctx(Arc::new(InstanceContext::new()));
+        let scope = store
+            .scanner_data_usage_publication_commit_scope(0, tokio::time::Instant::now() + Duration::from_secs(1), Vec::new())
+            .await
+            .expect("idle storage should grant a publication scope");
+        tokio::time::advance(Duration::from_secs(1)).await;
+        assert_eq!(
+            scope.try_begin(),
+            Err(crate::object_api::ScannerPublicationCommitStartError::DeadlineExceeded)
+        );
+        assert!(
+            !scope.release_movement_permit().await,
+            "an admitted scope is not safe to release before owner resolution"
+        );
+        assert!(scope.mark_aborted_before_commit());
+        assert!(scope.release_movement_permit().await);
+
+        let scope = store
+            .scanner_data_usage_publication_commit_scope(0, tokio::time::Instant::now() + Duration::from_secs(30), Vec::new())
+            .await
+            .expect("a second idle publication scope should be granted");
+        scope.try_begin().expect("scope should enter the mutation state");
+        scope.cancel();
+        assert!(scope.mark_indeterminate());
+        assert_eq!(
+            scope.wait_for_completion().await,
+            crate::object_api::ScannerPublicationCommitState::Indeterminate
+        );
+        assert!(!scope.release_movement_permit().await, "indeterminate mutation must retain the permit");
+    }
+
+    #[tokio::test]
+    async fn scanner_publication_scope_guard_classifies_early_returns_conservatively() {
+        let store = build_store_with_ctx(Arc::new(InstanceContext::new()));
+        let permit = store.ctx.data_movement_operation_gate().read_owned().await;
+        let scope = ScannerPublicationCommitScope::new_storage_owned(
+            0,
+            tokio::time::Instant::now() + Duration::from_secs(30),
+            Vec::new(),
+            permit,
+        );
+        {
+            let _guard = crate::object_api::ScannerPublicationCommitScopeGuard::new(scope.clone());
+        }
+        assert_eq!(scope.state(), crate::object_api::ScannerPublicationCommitState::AbortedBeforeCommit);
+        assert!(scope.release_movement_permit().await);
+
+        let permit = store.ctx.data_movement_operation_gate().read_owned().await;
+        let scope = ScannerPublicationCommitScope::new_storage_owned(
+            0,
+            tokio::time::Instant::now() + Duration::from_secs(30),
+            Vec::new(),
+            permit,
+        );
+        scope.try_begin().expect("scope should enter the mutation state");
+        {
+            let _guard = crate::object_api::ScannerPublicationCommitScopeGuard::new(scope.clone());
+        }
+        assert_eq!(scope.state(), crate::object_api::ScannerPublicationCommitState::Indeterminate);
+        assert!(!scope.release_movement_permit().await);
+    }
+
     #[tokio::test]
     async fn scanner_target_guard_keeps_movement_writer_fenced_after_lease_release() {
         let store = build_store_with_ctx(Arc::new(InstanceContext::new()));

crates/protos/src/generated/proto_gen/node_service.rs

@@ -1553,6 +1553,9 @@ pub enum ControlPlaneErrorCode {
     /// The peer answered but its storage/IAM layer is not initialized yet
     /// (legacy string form: "errServerNotInitialized").
     ControlPlaneErrorNotInitialized = 1,
+    /// The peer rejected a control-plane request before changing durable state.
+    /// error_info carries the actionable validation reason.
+    ControlPlaneErrorInvalidArgument = 2,
 }
 impl ControlPlaneErrorCode {
     /// String value of the enum field names used in the ProtoBuf definition.
@@ -1563,6 +1566,7 @@ impl ControlPlaneErrorCode {
         match self {
             Self::ControlPlaneErrorUnspecified => "CONTROL_PLANE_ERROR_UNSPECIFIED",
             Self::ControlPlaneErrorNotInitialized => "CONTROL_PLANE_ERROR_NOT_INITIALIZED",
+            Self::ControlPlaneErrorInvalidArgument => "CONTROL_PLANE_ERROR_INVALID_ARGUMENT",
         }
     }
     /// Creates an enum from field names used in the ProtoBuf definition.
@@ -1570,6 +1574,7 @@ impl ControlPlaneErrorCode {
         match value {
             "CONTROL_PLANE_ERROR_UNSPECIFIED" => Some(Self::ControlPlaneErrorUnspecified),
             "CONTROL_PLANE_ERROR_NOT_INITIALIZED" => Some(Self::ControlPlaneErrorNotInitialized),
+            "CONTROL_PLANE_ERROR_INVALID_ARGUMENT" => Some(Self::ControlPlaneErrorInvalidArgument),
             _ => None,
         }
     }

crates/protos/src/node.proto

@@ -30,6 +30,9 @@ enum ControlPlaneErrorCode {
   // The peer answered but its storage/IAM layer is not initialized yet
   // (legacy string form: "errServerNotInitialized").
   CONTROL_PLANE_ERROR_NOT_INITIALIZED = 1;
+  // The peer rejected a control-plane request before changing durable state.
+  // error_info carries the actionable validation reason.
+  CONTROL_PLANE_ERROR_INVALID_ARGUMENT = 2;
 }
 
 message PingRequest {

crates/replication/src/config.rs

@@ -60,16 +60,17 @@ pub const REPLICATION_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &[
     "Destination.ReplicationTime",
 ];
 
-// v3: temporary-credential fields are advertised as read-only historical
-// metadata. They remain decodable for MinIO and persisted-data compatibility,
-// but set-remote-target rejects them until refresh and rotation are supported.
-pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 3;
+// v4: temporary-credential fields moved from read-only historical metadata to
+// writable fields because remote targets now use them for request signing.
+pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 4;
 
 pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
     "sourcebucket",
     "endpoint",
     "credentials.accessKey",
     "credentials.secretKey",
+    "credentials.sessionToken",
+    "credentials.expiration",
     "targetbucket",
     "secure",
     "path",
@@ -91,7 +92,12 @@ pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
     "disableProxy",
 ];
 
-pub const REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &["credentials.sessionToken", "credentials.expiration"];
+/// Remote target fields that are readable for persisted-data compatibility but
+/// cannot be written through the admin API.
+///
+/// The empty slice remains public for source compatibility with consumers of
+/// the v3 capability API.
+pub const REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &[];
 
 pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["edge", "edgeSyncBeforeExpiry"];
 

crates/scanner/src/lib.rs

@@ -39,11 +39,11 @@ use storage_api::owner::{
     EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts, EcstoreReplicationConfigurationExt,
     EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore,
     EcstoreVersioningApi, HTTPPreconditions, HTTPRangeSpec, ObjectIO, ObjectOperations, ObjectToDelete,
-    ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule,
-    ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config,
-    ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache,
-    ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, ecstore_is_erasure_sd,
-    ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info,
+    ScannerPublicationCommitScope, ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission,
+    ecstore_apply_expiry_rule, ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr,
+    ecstore_get_lifecycle_config, ecstore_get_object_lock_config, ecstore_get_replication_config,
+    ecstore_invalidate_admin_data_usage_snapshot_cache, ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure,
+    ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info,
     ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, ecstore_read_config,
     ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_save_config,
     scanner_replication_config_for_lifecycle_eval,
@@ -55,6 +55,7 @@ use storage_api::owner::{
     ecstore_new_disk,
 };
 use tokio_util::sync::CancellationToken;
+use uuid::Uuid;
 
 pub mod data_usage_define;
 pub mod error;
@@ -752,6 +753,32 @@ pub(crate) fn scanner_publication_epoch_changed(error: &EcstoreError) -> bool {
     )
 }
 
+pub(crate) async fn delete_config_with_publication_scope_for_epoch<S>(
+    api: Arc<S>,
+    bucket: &str,
+    object: &str,
+    mut opts: ScannerObjectOptions,
+    expected_epoch: u64,
+    scanner_publication_commit_scope: Option<ScannerPublicationCommitScope>,
+) -> EcstoreResult<ScannerObjectInfo>
+where
+    S: ScannerObjectIO + ScannerConfigObjectDelete,
+{
+    let legacy_admission = if scanner_publication_commit_scope.is_none() {
+        Some(
+            scanner_publication_admission_for_epoch(api.clone(), expected_epoch)
+                .await
+                .ok_or_else(|| EcstoreError::other(SCANNER_PUBLICATION_EPOCH_CHANGED))?,
+        )
+    } else {
+        None
+    };
+    opts.scanner_publication_commit_scope = scanner_publication_commit_scope;
+    let result = api.delete_config_object(bucket, object, opts).await;
+    drop(legacy_admission);
+    result
+}
+
 pub(crate) async fn delete_config_with_publication_admission_for_epoch<S>(
     api: Arc<S>,
     bucket: &str,
@@ -762,10 +789,7 @@ pub(crate) async fn delete_config_with_publication_admission_for_epoch<S>(
 where
     S: ScannerObjectIO + ScannerConfigObjectDelete,
 {
-    let Some(_admission) = scanner_publication_admission_for_epoch(api.clone(), expected_epoch).await else {
-        return Err(EcstoreError::other(SCANNER_PUBLICATION_EPOCH_CHANGED));
-    };
-    api.delete_config_object(bucket, object, opts).await
+    delete_config_with_publication_scope_for_epoch(api, bucket, object, opts, expected_epoch, None).await
 }
 
 /// Capture the storage-owned publication epoch without retaining the read
@@ -796,13 +820,14 @@ where
     Some(admission)
 }
 
-pub(crate) async fn save_config_shared_with_preconditions_and_lease_fence<S>(
+pub(crate) async fn save_config_shared_with_preconditions_and_lease_fence_and_scope<S>(
     api: Arc<S>,
     file: &str,
     data: Bytes,
     sha256hex: Option<String>,
     preconditions: HTTPPreconditions,
     scanner_publication_lease_fence: Option<&str>,
+    scanner_publication_commit_scope: Option<ScannerPublicationCommitScope>,
 ) -> EcstoreResult<ScannerObjectInfo>
 where
     S: ScannerObjectIO,
@@ -822,6 +847,7 @@ where
         &ScannerObjectOptions {
             max_parity: true,
             http_preconditions: Some(preconditions),
+            scanner_publication_commit_scope,
             user_defined,
             ..Default::default()
         },
@@ -886,6 +912,27 @@ pub trait ScannerConfigObjectDelete: Send + Sync + std::fmt::Debug + 'static {
     async fn scanner_data_usage_publication_admission(&self) -> Option<ScannerDataUsagePublicationAdmission> {
         None
     }
+
+    /// Acquire a storage-owned scope for a fenced scanner metadata mutation.
+    /// Implementations without a storage movement owner fail closed.
+    async fn scanner_data_usage_publication_commit_scope(
+        &self,
+        _expected_movement_epoch: u64,
+        _safe_deadline: tokio::time::Instant,
+        _remote_lease_tokens: Vec<Uuid>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        None
+    }
+
+    async fn scanner_data_usage_publication_commit_scope_with_release_flag(
+        &self,
+        _expected_movement_epoch: u64,
+        _safe_deadline: tokio::time::Instant,
+        _remote_lease_tokens: Vec<Uuid>,
+        _lease_release_safe: Arc<std::sync::atomic::AtomicBool>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        None
+    }
 }
 
 pub struct ScannerDataUsagePublicationAdmission {
@@ -929,6 +976,32 @@ impl ScannerConfigObjectDelete for ECStore {
         let (read_guard, epoch) = self.scanner_data_usage_publication_admission_guard().await?;
         Some(ScannerDataUsagePublicationAdmission::fenced(read_guard, epoch))
     }
+
+    async fn scanner_data_usage_publication_commit_scope(
+        &self,
+        expected_movement_epoch: u64,
+        safe_deadline: tokio::time::Instant,
+        remote_lease_tokens: Vec<Uuid>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        self.scanner_data_usage_publication_commit_scope(expected_movement_epoch, safe_deadline, remote_lease_tokens)
+            .await
+    }
+
+    async fn scanner_data_usage_publication_commit_scope_with_release_flag(
+        &self,
+        expected_movement_epoch: u64,
+        safe_deadline: tokio::time::Instant,
+        remote_lease_tokens: Vec<Uuid>,
+        lease_release_safe: Arc<std::sync::atomic::AtomicBool>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        self.scanner_data_usage_publication_commit_scope_with_release_flag(
+            expected_movement_epoch,
+            safe_deadline,
+            remote_lease_tokens,
+            lease_release_safe,
+        )
+        .await
+    }
 }
 
 #[async_trait::async_trait]
@@ -946,6 +1019,32 @@ impl ScannerConfigObjectDelete for SetDisks {
         let (read_guard, epoch) = self.scanner_data_usage_publication_admission_guard().await?;
         Some(ScannerDataUsagePublicationAdmission::fenced(read_guard, epoch))
     }
+
+    async fn scanner_data_usage_publication_commit_scope(
+        &self,
+        expected_movement_epoch: u64,
+        safe_deadline: tokio::time::Instant,
+        remote_lease_tokens: Vec<Uuid>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        self.scanner_data_usage_publication_commit_scope(expected_movement_epoch, safe_deadline, remote_lease_tokens)
+            .await
+    }
+
+    async fn scanner_data_usage_publication_commit_scope_with_release_flag(
+        &self,
+        expected_movement_epoch: u64,
+        safe_deadline: tokio::time::Instant,
+        remote_lease_tokens: Vec<Uuid>,
+        lease_release_safe: Arc<std::sync::atomic::AtomicBool>,
+    ) -> Option<ScannerPublicationCommitScope> {
+        self.scanner_data_usage_publication_commit_scope_with_release_flag(
+            expected_movement_epoch,
+            safe_deadline,
+            remote_lease_tokens,
+            lease_release_safe,
+        )
+        .await
+    }
 }
 
 #[cfg(test)]

crates/scanner/src/scanner.rs

@@ -16,6 +16,7 @@ use std::collections::BTreeMap;
 use std::future::Future;
 #[cfg(test)]
 use std::sync::Mutex as StdMutex;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, LazyLock, RwLock};
 
 use self::heal_info::{BackgroundHealInfoReadStatus, read_background_heal_info_with_epoch, save_background_heal_info_for_epoch};
@@ -62,8 +63,8 @@ use tokio::time::{Duration, Instant};
 use tokio_util::sync::CancellationToken;
 use tokio_util::task::AbortOnDropHandle;
 use tracing::{debug, error, info, instrument, warn};
+use uuid::Uuid;
 
-use crate::storage_api::owner::SCANNER_PUBLICATION_LEASE_TTL_MS;
 use crate::storage_api::scan::{
     BucketOperations, BucketOptions, NamespaceLocking as _, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION,
     SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_ACTIVITY_PROTOCOL_VERSION,
@@ -72,7 +73,7 @@ use crate::{
     ECStore, EcstoreError, RUSTFS_META_BUCKET, SCANNER_PUBLICATION_EPOCH_CHANGED, ScannerLifecycleConfigExt as _,
     ScannerReplicationConfigExt as _, delete_config_with_publication_admission_for_epoch, get_lifecycle_config,
     get_replication_config, invalidate_admin_data_usage_snapshot_cache, invalidate_data_usage_snapshot_cache, read_config,
-    replace_bucket_usage_memory_from_info, save_config, save_config_shared_with_preconditions_and_lease_fence,
+    replace_bucket_usage_memory_from_info, save_config, save_config_shared_with_preconditions_and_lease_fence_and_scope,
     save_config_with_preconditions, save_config_with_publication_admission_for_epoch, scanner_is_erasure_sd,
     scanner_publication_admission_for_epoch, scanner_publication_epoch, scanner_publication_epoch_changed,
 };
@@ -454,6 +455,7 @@ fn data_usage_backup_due(data_usage_info: &DataUsageInfo) -> bool {
 }
 
 #[cfg(test)]
+#[allow(dead_code)]
 async fn sync_data_usage_backup_from_primary(
     ctx: &CancellationToken,
     storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
@@ -461,12 +463,34 @@ async fn sync_data_usage_backup_from_primary(
     sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence(ctx, storeapi, None, None, None).await
 }
 
+#[allow(dead_code)]
 async fn sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence(
     ctx: &CancellationToken,
     storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
     expected_publication_epoch: Option<u64>,
     remote_lease_deadline: Option<std::time::Instant>,
     scanner_publication_lease_fence: Option<&str>,
+) -> Result<(), EcstoreError> {
+    sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence_and_scope(
+        ctx,
+        storeapi,
+        expected_publication_epoch,
+        remote_lease_deadline,
+        scanner_publication_lease_fence,
+        Vec::new(),
+        Arc::new(AtomicBool::new(true)),
+    )
+    .await
+}
+
+async fn sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence_and_scope(
+    ctx: &CancellationToken,
+    storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
+    expected_publication_epoch: Option<u64>,
+    remote_lease_deadline: Option<std::time::Instant>,
+    scanner_publication_lease_fence: Option<&str>,
+    remote_lease_tokens: Vec<Uuid>,
+    lease_release_safe: Arc<AtomicBool>,
 ) -> Result<(), EcstoreError> {
     let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
     for retry in 0..=SCANNER_PERSIST_CAS_RETRIES {
@@ -531,15 +555,48 @@ async fn sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence(
                 }
                 return Err(EcstoreError::other(SCANNER_PUBLICATION_EPOCH_CHANGED));
             };
-            save_config_shared_with_preconditions_and_lease_fence(
+            let publication_scope = match expected_publication_epoch {
+                Some(expected_epoch) => {
+                    storeapi
+                        .scanner_data_usage_publication_commit_scope_with_release_flag(
+                            expected_epoch,
+                            usage_store::scanner_publication_scope_deadline(data_usage_persist_timeout(), remote_lease_deadline),
+                            remote_lease_tokens.clone(),
+                            Arc::clone(&lease_release_safe),
+                        )
+                        .await
+                }
+                None => None,
+            };
+            if expected_publication_epoch.is_some() && publication_scope.is_none() {
+                if retry < SCANNER_PERSIST_CAS_RETRIES {
+                    continue;
+                }
+                return Err(EcstoreError::other(SCANNER_PUBLICATION_EPOCH_CHANGED));
+            }
+            let save_result = save_config_shared_with_preconditions_and_lease_fence_and_scope(
                 storeapi.clone(),
                 &backup_path,
                 primary.clone(),
                 sha256hex,
                 revision.preconditions(),
                 scanner_publication_lease_fence,
+                publication_scope.clone(),
             )
-            .await
+            .await;
+            if let Some(scope) = publication_scope {
+                match scope.wait_for_completion().await {
+                    crate::storage_api::owner::ScannerPublicationCommitState::Committed
+                    | crate::storage_api::owner::ScannerPublicationCommitState::AbortedBeforeCommit => save_result,
+                    crate::storage_api::owner::ScannerPublicationCommitState::Indeterminate
+                    | crate::storage_api::owner::ScannerPublicationCommitState::Admitted
+                    | crate::storage_api::owner::ScannerPublicationCommitState::InFlight => Err(EcstoreError::other(
+                        "scanner backup publication commit scope did not reach a safe terminal state",
+                    )),
+                }
+            } else {
+                save_result
+            }
         };
 
         match save_result {
@@ -1205,10 +1262,6 @@ fn data_usage_persist_timeout() -> Duration {
     DataUsageCache::persistence_timeout()
 }
 
-fn scanner_publication_lease_budget_allows_persistence(timeout: Duration) -> bool {
-    timeout < Duration::from_millis(SCANNER_PUBLICATION_LEASE_TTL_MS)
-}
-
 #[cfg(not(test))]
 const SCANNER_CYCLE_EPOCH_FENCE_TIMEOUT: Duration = Duration::from_secs(30);
 #[cfg(test)]
@@ -1493,21 +1546,24 @@ async fn run_data_scanner_cycle_with_budget(
     let mut remote_publication_leases = None;
     let remote_lease_defer_reason = if remote_publication_lease_targets.is_empty() {
         None
-    } else if !scanner_publication_lease_budget_allows_persistence(usage_persist_timeout) {
-        // The lease is intentionally fixed-duration and has no renewal path.
-        // Refuse a persistence budget that could outlive it instead of
-        // allowing the peer to admit movement while a local PUT is in flight.
-        Some(ScannerCycleDeferReason::PublicationLeaseBudgetExceeded)
     } else if let Some(notification_system) = storeapi.notification_system() {
-        match notification_system
-            .acquire_scanner_publication_leases(remote_publication_lease_targets.clone())
-            .await
-        {
-            Ok(grants) => {
+        let publication_proof_ctx = cycle_budget.token();
+        let lease_result = await_scanner_publication_proof(
+            &publication_proof_ctx,
+            cycle_info.current,
+            "lease_acquire",
+            || notification_system.acquire_scanner_publication_leases(remote_publication_lease_targets.clone()),
+            |err| scanner_publication_lease_error_is_retryable(&err.to_string()),
+        )
+        .await;
+        match lease_result {
+            ScannerPublicationProofWait::Ready(grants) => {
                 remote_publication_leases = Some((notification_system, grants));
                 None
             }
-            Err(_) => Some(ScannerCycleDeferReason::ActivityBaselineUnavailable),
+            ScannerPublicationProofWait::Rejected(_) | ScannerPublicationProofWait::Cancelled => {
+                Some(ScannerCycleDeferReason::ActivityBaselineUnavailable)
+            }
         }
     } else {
         Some(ScannerCycleDeferReason::ActivityBaselineUnavailable)
@@ -1548,26 +1604,16 @@ async fn run_data_scanner_cycle_with_budget(
         remote_lease_fence.is_some(),
     ))
     .then_some(ScannerCycleDeferReason::ActivityBaselineUnavailable);
-    let remote_lease_covers_persistence = remote_lease_deadline.is_none_or(|deadline| {
-        std::time::Instant::now()
-            .checked_add(usage_persist_timeout)
-            .is_some_and(|latest_finish| latest_finish < deadline)
-    });
     let publication_defer_reason = publication_defer_reason
         .or(remote_lease_defer_reason)
         .or(remote_lease_fence_defer_reason);
-    let publication_defer_reason = (!remote_lease_covers_persistence)
-        .then_some(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded)
-        .or(publication_defer_reason);
     // Include reasons discovered while acquiring or validating remote leases.
-    // In particular, the static budget gate above is reached after the scan
-    // result is classified, so computing this flag earlier would suppress its
-    // deferred metric.
     let publication_deferred = publication_defer_reason.is_some();
     let budget_elapsed = cycle_budget.budget_elapsed() && !ctx.is_cancelled();
     let remote_lease_probe = remote_publication_leases
         .as_ref()
         .map(|(notification_system, grants)| (Arc::clone(notification_system), grants.clone()));
+    let remote_lease_release_safe = Arc::new(AtomicBool::new(true));
     let mut usage_persist_outcome = match publication_defer_reason {
         Some(reason) => {
             drop(receiver);
@@ -1581,6 +1627,11 @@ async fn run_data_scanner_cycle_with_budget(
             let ctx_clone = ctx.clone();
             let route_probe_store = storeapi.clone();
             let remote_lease_fence = remote_lease_fence.clone();
+            let remote_lease_release_safe_for_task = Arc::clone(&remote_lease_release_safe);
+            let remote_lease_tokens = remote_publication_leases
+                .as_ref()
+                .map(|(_, grants)| grants.iter().map(|grant| grant.lease.token).collect())
+                .unwrap_or_default();
             let mut usage_persist_task = AbortOnDropHandle::new(tokio::spawn(async move {
                 store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence(
                     ctx_clone,
@@ -1592,7 +1643,9 @@ async fn run_data_scanner_cycle_with_budget(
                         publication_epoch,
                         remote_lease_deadline,
                         remote_lease_fence,
-                    ),
+                    )
+                    .with_remote_lease_tokens(remote_lease_tokens)
+                    .with_lease_release_flag(remote_lease_release_safe_for_task),
                     move || {
                         let storeapi = route_probe_store.clone();
                         let remote_lease_probe = remote_lease_probe.clone();
@@ -1657,7 +1710,16 @@ async fn run_data_scanner_cycle_with_budget(
     let lease_expired = remote_publication_leases
         .as_ref()
         .is_some_and(|(_, grants)| grants.iter().any(|grant| !grant.lease.is_valid()));
-    if let Some((notification_system, grants)) = remote_publication_leases.take() {
+    if !remote_lease_release_safe.load(Ordering::Acquire) {
+        // A cancelled or detached storage mutation did not report a safe
+        // terminal state. Keep remote grants until their own expiry rather
+        // than releasing movement admission while a commit may be unknown.
+        usage_persist_outcome = if usage_persist_outcome == DataUsagePersistOutcome::Failed {
+            DataUsagePersistOutcome::Failed
+        } else {
+            DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded)
+        };
+    } else if let Some((notification_system, grants)) = remote_publication_leases.take() {
         let release_result = notification_system.release_scanner_publication_leases(grants).await;
         let lease_release_failed = release_result.is_err();
         if lease_expired || lease_release_failed {

crates/scanner/src/scanner/activity.rs

@@ -108,6 +108,149 @@ impl ScannerRetryBackoff {
     }
 }
 
+const SCANNER_PUBLICATION_PROOF_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(30);
+
+pub(crate) fn scanner_publication_proof_retry_delay(consecutive_failures: u32) -> Duration {
+    let exponent = consecutive_failures.saturating_sub(1).min(31);
+    let multiplier = 1u32.checked_shl(exponent).unwrap_or(u32::MAX);
+    SCANNER_RETRY_BASE_INTERVAL
+        .saturating_mul(multiplier)
+        .min(SCANNER_PUBLICATION_PROOF_RETRY_MAX_INTERVAL)
+}
+
+pub(crate) fn scanner_publication_activity_error_is_retryable(error: &str) -> bool {
+    crate::storage_api::scanner_peer_transport_error_message_is_retryable(error)
+}
+
+pub(crate) fn scanner_publication_lease_error_is_retryable(error: &str) -> bool {
+    scanner_publication_activity_error_is_retryable(error)
+        || error.ends_with("scanner publication lease capacity is exhausted")
+        || error.ends_with("scanner publication lease response arrived after its safety window")
+}
+
+pub(crate) enum ScannerPublicationProofWait<T, E> {
+    Ready(T),
+    Rejected(E),
+    Cancelled,
+}
+
+pub(crate) async fn await_scanner_publication_proof<T, E, F, Fut, Retryable>(
+    ctx: &CancellationToken,
+    cycle: u64,
+    stage: &'static str,
+    mut proof: F,
+    retryable: Retryable,
+) -> ScannerPublicationProofWait<T, E>
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<T, E>>,
+    E: std::fmt::Display,
+    Retryable: Fn(&E) -> bool,
+{
+    let started_at = Instant::now();
+    let mut consecutive_failures = 0u32;
+
+    loop {
+        if ctx.is_cancelled() {
+            return ScannerPublicationProofWait::Cancelled;
+        }
+
+        match proof().await {
+            Ok(value) => {
+                if consecutive_failures > 0 {
+                    info!(
+                        target: "rustfs::scanner",
+                        event = EVENT_SCANNER_CYCLE_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_RUNTIME,
+                        state = "publication_proof_recovered",
+                        cycle,
+                        stage,
+                        attempts = consecutive_failures.saturating_add(1),
+                        pending_duration = ?started_at.elapsed(),
+                        "Scanner publication proof recovered"
+                    );
+                }
+                return ScannerPublicationProofWait::Ready(value);
+            }
+            Err(err) if !retryable(&err) => {
+                warn!(
+                    target: "rustfs::scanner",
+                    event = EVENT_SCANNER_CYCLE_STATE,
+                    component = LOG_COMPONENT_SCANNER,
+                    subsystem = LOG_SUBSYSTEM_RUNTIME,
+                    state = "publication_proof_rejected",
+                    cycle,
+                    stage,
+                    error = %err,
+                    "Scanner publication proof failed with a non-retryable cluster state"
+                );
+                return ScannerPublicationProofWait::Rejected(err);
+            }
+            Err(err) => {
+                consecutive_failures = consecutive_failures.saturating_add(1);
+                let retry_delay = scanner_publication_proof_retry_delay(consecutive_failures);
+                if consecutive_failures == 1 || consecutive_failures.is_multiple_of(20) {
+                    warn!(
+                        target: "rustfs::scanner",
+                        event = EVENT_SCANNER_CYCLE_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_RUNTIME,
+                        state = "publication_proof_pending",
+                        cycle,
+                        stage,
+                        attempt = consecutive_failures,
+                        retry_delay = ?retry_delay,
+                        error = %err,
+                        "Scanner retained a completed scan while publication proof is unavailable"
+                    );
+                } else {
+                    debug!(
+                        target: "rustfs::scanner",
+                        event = EVENT_SCANNER_CYCLE_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_RUNTIME,
+                        state = "publication_proof_retry",
+                        cycle,
+                        stage,
+                        attempt = consecutive_failures,
+                        retry_delay = ?retry_delay,
+                        error = %err,
+                        "Scanner publication activity proof retry scheduled"
+                    );
+                }
+
+                tokio::select! {
+                    _ = ctx.cancelled() => return ScannerPublicationProofWait::Cancelled,
+                    _ = tokio::time::sleep(retry_delay) => {}
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+pub(crate) async fn await_scanner_publication_activity<F, Fut>(
+    ctx: &CancellationToken,
+    cycle: u64,
+    stage: &'static str,
+    probe: F,
+) -> Result<ScannerActivitySnapshot, String>
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<ScannerActivitySnapshot, String>>,
+{
+    match await_scanner_publication_proof(ctx, cycle, stage, probe, |err: &String| {
+        scanner_publication_activity_error_is_retryable(err)
+    })
+    .await
+    {
+        ScannerPublicationProofWait::Ready(snapshot) => Ok(snapshot),
+        ScannerPublicationProofWait::Rejected(err) => Err(err),
+        ScannerPublicationProofWait::Cancelled => Err(format!("scanner publication activity proof was cancelled during {stage}")),
+    }
+}
+
 impl Default for ScannerCleanIdleBackoff {
     fn default() -> Self {
         Self { interval_multiplier: 1 }

crates/scanner/src/scanner/tests.rs

@@ -352,6 +352,7 @@ struct MemoryConfigStore {
     objects: Mutex<HashMap<String, Vec<u8>>>,
     revisions: Mutex<HashMap<String, u64>>,
     insert_after_gets: Mutex<HashMap<String, Vec<u8>>>,
+    delayed_gets: Mutex<HashMap<String, Duration>>,
     non_regular_objects: Mutex<HashSet<String>>,
     fail_put_number: Mutex<HashMap<String, usize>>,
     object_not_found_put_number: Mutex<HashMap<String, usize>>,
@@ -399,6 +400,9 @@ impl crate::storage_api::scanner_io::ObjectIO for MemoryConfigStore {
         _opts: &ObjectOptions,
     ) -> EcstoreResult<GetObjectReader> {
         let key = memory_config_key(bucket, object);
+        if let Some(delay) = self.delayed_gets.lock().await.remove(&key) {
+            tokio::time::sleep(delay).await;
+        }
         let inserted_data = self.insert_after_gets.lock().await.remove(&key);
         let data = {
             let mut objects = self.objects.lock().await;
@@ -3532,6 +3536,47 @@ async fn coordinator_classifies_an_expired_publication_lease() {
     assert!(store.put_counts.lock().await.is_empty(), "expired lease must prevent a PUT");
 }
 
+#[tokio::test]
+async fn backup_sync_checks_the_lease_deadline_after_a_slow_backup_read() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let primary_path = DATA_USAGE_OBJ_NAME_PATH.as_str();
+    let backup_path = format!("{primary_path}.bkp");
+    let primary_key = memory_config_key(RUSTFS_META_BUCKET, primary_path);
+    let backup_key = memory_config_key(RUSTFS_META_BUCKET, &backup_path);
+    let primary = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    store
+        .objects
+        .lock()
+        .await
+        .insert(primary_key, serde_json::to_vec(&primary).expect("primary usage snapshot should encode"));
+    store
+        .delayed_gets
+        .lock()
+        .await
+        .insert(backup_key.clone(), Duration::from_millis(20));
+
+    // The primary read is allowed to start, but the backup read consumes the
+    // remaining lease window. The second deadline check must prevent a stale
+    // backup PUT after that window has elapsed.
+    let deadline = std::time::Instant::now()
+        .checked_add(std::time::Duration::from_millis(5))
+        .expect("test deadline should support a five-millisecond window");
+    let result = sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence(
+        &CancellationToken::new(),
+        store.clone(),
+        None,
+        Some(deadline),
+        None,
+    )
+    .await;
+
+    assert!(scanner_publication_epoch_changed(
+        &result.expect_err("an expired backup lease must defer publication")
+    ));
+    assert!(!store.objects.lock().await.contains_key(&backup_key));
+    assert_eq!(store.put_counts.lock().await.get(&backup_key), None);
+}
+
 #[tokio::test]
 #[serial]
 async fn test_deferred_usage_save_keeps_last_real_save_metric() {
@@ -4583,7 +4628,6 @@ fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() {
     for reason in [
         ScannerCycleDeferReason::DataMovement,
         ScannerCycleDeferReason::ActivityBaselineUnavailable,
-        ScannerCycleDeferReason::PublicationLeaseBudgetExceeded,
         ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded,
         ScannerCycleDeferReason::PublicationLeaseReleaseFailed,
     ] {
@@ -4755,29 +4799,6 @@ fn data_usage_persist_wait_covers_cache_retries_and_backup() {
     crate::runtime_config::refresh_scanner_runtime_config_for_tests();
 }
 
-#[test]
-fn scanner_publication_lease_budget_has_a_strict_ttl_boundary() {
-    let ttl = Duration::from_millis(SCANNER_PUBLICATION_LEASE_TTL_MS);
-
-    assert!(scanner_publication_lease_budget_allows_persistence(
-        ttl.saturating_sub(Duration::from_millis(1))
-    ));
-    assert!(!scanner_publication_lease_budget_allows_persistence(ttl));
-    assert!(!scanner_publication_lease_budget_allows_persistence(ttl + Duration::from_millis(1)));
-    assert_eq!(
-        ScannerCycleDeferReason::PublicationLeaseBudgetExceeded.as_str(),
-        "publication_lease_budget_exceeded"
-    );
-    assert_eq!(
-        ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded.as_str(),
-        "publication_lease_deadline_exceeded"
-    );
-    assert_eq!(
-        ScannerCycleDeferReason::PublicationLeaseReleaseFailed.as_str(),
-        "publication_lease_release_failed"
-    );
-}
-
 #[tokio::test]
 async fn data_usage_persist_wait_aborts_when_scanner_is_cancelled() {
     let ctx = CancellationToken::new();
@@ -4807,6 +4828,29 @@ async fn data_usage_persist_wait_aborts_after_timeout() {
     assert!(task.is_finished());
 }
 
+#[tokio::test(start_paused = true)]
+async fn data_usage_persist_timeout_drops_owned_task_without_a_late_commit() {
+    let ctx = CancellationToken::new();
+    let commit_started = Arc::new(AtomicBool::new(false));
+    let commit_started_by_task = commit_started.clone();
+    let task_ready = Arc::new(tokio::sync::Notify::new());
+    let task_ready_by_task = task_ready.clone();
+    let mut task = AbortOnDropHandle::new(tokio::spawn(async move {
+        task_ready_by_task.notify_one();
+        std::future::pending::<()>().await;
+        commit_started_by_task.store(true, Ordering::Release);
+        DataUsagePersistOutcome::Saved
+    }));
+    task_ready.notified().await;
+
+    let result = wait_for_data_usage_persist_task(&ctx, &mut task, Duration::from_secs(1)).await;
+
+    assert!(matches!(result, DataUsagePersistTaskResult::TimedOut));
+    assert!(task.is_finished(), "the timed-out persistence task must be drained before return");
+    tokio::task::yield_now().await;
+    assert!(!commit_started.load(Ordering::Acquire), "an owned task must not commit after its timeout");
+}
+
 #[tokio::test(start_paused = true)]
 async fn maintenance_feature_inspection_preserves_base_cycle_after_timeout() {
     let ctx = CancellationToken::new();
@@ -5032,6 +5076,140 @@ fn superseded_retry_backoff_grows_from_the_default_cycle() {
     }
 }
 
+#[test]
+fn publication_proof_retry_backoff_reaches_its_short_cap() {
+    for (failures, expected) in [(1, 5), (2, 10), (3, 20), (4, 30), (20, 30)] {
+        assert_eq!(scanner_publication_proof_retry_delay(failures), Duration::from_secs(expected));
+    }
+}
+
+#[test]
+fn publication_proof_retry_classifies_availability_without_masking_protocol_errors() {
+    for error in [
+        "peer node3 is temporarily offline",
+        "scanner activity peer node3 timed out after 5s",
+        "transport error: connection refused",
+    ] {
+        assert!(scanner_publication_activity_error_is_retryable(error), "{error}");
+    }
+
+    for error in [
+        "scanner activity peer node3 uses protocol 6, expected 7",
+        "scanner activity peer node3 has a different storage topology",
+        "scanner activity peer node3 omitted its movement generation",
+        "duplicate scanner activity peer: node3",
+        "scanner activity peer[2] is unreachable",
+        "scanner publication lease peer node3 is unavailable",
+    ] {
+        assert!(!scanner_publication_activity_error_is_retryable(error), "{error}");
+    }
+}
+
+#[test]
+fn publication_lease_retry_preserves_only_recoverable_candidates() {
+    for error in [
+        "scanner publication lease acquisition failed: scanner publication lease capacity is exhausted",
+        "scanner publication lease acquisition failed: scanner publication lease response arrived after its safety window",
+        "scanner publication lease acquisition failed: peer node3 is temporarily offline",
+    ] {
+        assert!(scanner_publication_lease_error_is_retryable(error), "{error}");
+    }
+
+    for error in [
+        "scanner publication lease acquisition failed: scanner publication lease generation is stale",
+        "scanner publication lease acquisition failed: peer returned a different scanner publication lease session",
+        "scanner publication lease acquisition failed: scanner publication lease is blocked by data movement",
+        "scanner publication lease acquisition failed: peer returned an invalid scanner publication lease proof",
+    ] {
+        assert!(!scanner_publication_lease_error_is_retryable(error), "{error}");
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_proof_retains_candidate_until_activity_recovers() {
+    let ctx = CancellationToken::new();
+    let attempts = Arc::new(AtomicUsize::new(0));
+    let probe_attempts = attempts.clone();
+    let started_at = Instant::now();
+
+    let snapshot = await_scanner_publication_activity(&ctx, 17, "postscan", move || {
+        let attempt = probe_attempts.fetch_add(1, Ordering::SeqCst);
+        async move {
+            if attempt == 0 {
+                Err("peer temporarily offline".to_string())
+            } else {
+                Ok(ScannerActivitySnapshot::new())
+            }
+        }
+    })
+    .await
+    .expect("a retained publication candidate should survive one transient probe failure");
+
+    assert!(snapshot.is_empty());
+    assert_eq!(attempts.load(Ordering::SeqCst), 2);
+    assert_eq!(started_at.elapsed(), Duration::from_secs(5));
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_proof_does_not_retry_a_protocol_mismatch() {
+    let ctx = CancellationToken::new();
+    let attempts = Arc::new(AtomicUsize::new(0));
+    let probe_attempts = attempts.clone();
+
+    let err = await_scanner_publication_activity(&ctx, 17, "postscan", move || {
+        probe_attempts.fetch_add(1, Ordering::SeqCst);
+        async { Err("scanner activity peer node3 uses protocol 6, expected 7".to_string()) }
+    })
+    .await
+    .expect_err("a protocol mismatch must not be hidden behind availability retries");
+
+    assert!(err.contains("uses protocol"));
+    assert_eq!(attempts.load(Ordering::SeqCst), 1);
+}
+
+#[tokio::test]
+async fn publication_proof_stops_waiting_when_the_cycle_is_cancelled() {
+    let ctx = CancellationToken::new();
+    ctx.cancel();
+    let attempts = Arc::new(AtomicUsize::new(0));
+    let probe_attempts = attempts.clone();
+
+    let err = await_scanner_publication_activity(&ctx, 17, "postscan", move || {
+        probe_attempts.fetch_add(1, Ordering::SeqCst);
+        async { Ok(ScannerActivitySnapshot::new()) }
+    })
+    .await
+    .expect_err("a cancelled cycle must release its retained publication candidate");
+
+    assert!(err.contains("cancelled"));
+    assert_eq!(attempts.load(Ordering::SeqCst), 0);
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_proof_releases_candidate_when_cancelled_during_backoff() {
+    let ctx = CancellationToken::new();
+    let cancel_ctx = ctx.clone();
+    let attempts = Arc::new(AtomicUsize::new(0));
+    let probe_attempts = attempts.clone();
+    let started_at = Instant::now();
+
+    let cancel = tokio::spawn(async move {
+        tokio::time::sleep(Duration::from_secs(1)).await;
+        cancel_ctx.cancel();
+    });
+    let err = await_scanner_publication_activity(&ctx, 17, "postscan", move || {
+        probe_attempts.fetch_add(1, Ordering::SeqCst);
+        async { Err("peer temporarily offline".to_string()) }
+    })
+    .await
+    .expect_err("cycle cancellation must release a candidate waiting to retry publication proof");
+    cancel.await.expect("cancellation task should complete");
+
+    assert!(err.contains("cancelled"));
+    assert_eq!(attempts.load(Ordering::SeqCst), 1);
+    assert_eq!(started_at.elapsed(), Duration::from_secs(1));
+}
+
 #[tokio::test(start_paused = true)]
 async fn corrupt_cycle_state_backoff_uses_virtual_clock() {
     let mut backoff = ScannerRetryBackoff::default();

crates/scanner/src/scanner/usage_store.rs

@@ -13,7 +13,10 @@
 // limitations under the License.
 /// Data-usage snapshot persistence: CAS store pipeline, epoch baselines, and observed-snapshot cleanup.
 use super::*;
+use crate::storage_api::owner::ScannerPublicationCommitState;
 use std::collections::HashMap;
+use std::sync::atomic::AtomicBool;
+use uuid::Uuid;
 
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub(super) enum DataUsagePersistOutcome {
@@ -34,6 +37,16 @@ fn remote_lease_expired(deadline: Option<std::time::Instant>) -> bool {
     deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline)
 }
 
+pub(super) fn scanner_publication_scope_deadline(
+    persist_timeout: Duration,
+    remote_lease_deadline: Option<std::time::Instant>,
+) -> tokio::time::Instant {
+    let configured_deadline = tokio::time::Instant::now() + persist_timeout;
+    remote_lease_deadline
+        .map(tokio::time::Instant::from_std)
+        .map_or(configured_deadline, |lease_deadline| configured_deadline.min(lease_deadline))
+}
+
 #[derive(Clone, Debug)]
 pub(super) struct DataUsagePersistBaseline {
     pub(super) data: Option<Bytes>,
@@ -126,6 +139,8 @@ pub(super) struct ScannerPublicationFence {
     pub(super) expected_publication_epoch: Option<u64>,
     pub(super) remote_lease_deadline: Option<std::time::Instant>,
     pub(super) scanner_publication_lease_fence: Option<String>,
+    pub(super) remote_lease_tokens: Vec<Uuid>,
+    pub(super) lease_release_safe: Arc<AtomicBool>,
 }
 
 impl ScannerPublicationFence {
@@ -138,8 +153,20 @@ impl ScannerPublicationFence {
             expected_publication_epoch,
             remote_lease_deadline,
             scanner_publication_lease_fence,
+            remote_lease_tokens: Vec::new(),
+            lease_release_safe: Arc::new(AtomicBool::new(true)),
         }
     }
+
+    pub(super) fn with_remote_lease_tokens(mut self, remote_lease_tokens: Vec<Uuid>) -> Self {
+        self.remote_lease_tokens = remote_lease_tokens;
+        self
+    }
+
+    pub(super) fn with_lease_release_flag(mut self, lease_release_safe: Arc<AtomicBool>) -> Self {
+        self.lease_release_safe = lease_release_safe;
+        self
+    }
 }
 
 #[derive(Debug)]
@@ -290,6 +317,8 @@ where
         expected_publication_epoch,
         remote_lease_deadline,
         scanner_publication_lease_fence,
+        remote_lease_tokens,
+        lease_release_safe,
     } = publication_fence;
     let mut outcome = DataUsagePersistOutcome::NoUpdate;
     let mut next_baseline = initial_baseline;
@@ -580,25 +609,54 @@ where
 
             let done_save = Metrics::time(Metric::SaveUsage);
             let save_result = {
-                let Some(_publication_admission) =
-                    scanner_publication_admission_for_epoch(storeapi.clone(), publication_epoch_for_save).await
-                else {
-                    done_save();
-                    break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement);
+                let publication_scope = storeapi
+                    .scanner_data_usage_publication_commit_scope_with_release_flag(
+                        publication_epoch_for_save,
+                        scanner_publication_scope_deadline(data_usage_persist_timeout(), remote_lease_deadline),
+                        remote_lease_tokens.clone(),
+                        Arc::clone(&lease_release_safe),
+                    )
+                    .await;
+                let legacy_publication_admission = if publication_scope.is_none() {
+                    let Some(admission) =
+                        scanner_publication_admission_for_epoch(storeapi.clone(), publication_epoch_for_save).await
+                    else {
+                        done_save();
+                        break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement);
+                    };
+                    Some(admission)
+                } else {
+                    None
                 };
                 if remote_lease_expired(remote_lease_deadline) {
                     done_save();
                     break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded);
                 }
-                save_config_shared_with_preconditions_and_lease_fence(
+                let save_result = crate::save_config_shared_with_preconditions_and_lease_fence_and_scope(
                     storeapi.clone(),
                     target_path,
                     data.clone(),
                     sha256hex.clone(),
                     revision.preconditions(),
                     scanner_publication_lease_fence.as_deref(),
+                    publication_scope.clone(),
                 )
-                .await
+                .await;
+                drop(legacy_publication_admission);
+                if let Some(scope) = publication_scope {
+                    match scope.wait_for_completion().await {
+                        ScannerPublicationCommitState::Committed | ScannerPublicationCommitState::AbortedBeforeCommit => {
+                            save_result
+                        }
+                        ScannerPublicationCommitState::Indeterminate
+                        | ScannerPublicationCommitState::Admitted
+                        | ScannerPublicationCommitState::InFlight => Err(EcstoreError::other(
+                            "scanner publication commit scope did not reach a safe terminal state",
+                        )),
+                    }
+                } else {
+                    save_result
+                }
             };
             done_save();
 
@@ -696,6 +754,8 @@ where
                         expected_publication_epoch,
                         remote_lease_deadline,
                         scanner_publication_lease_fence.as_deref(),
+                        &remote_lease_tokens,
+                        Arc::clone(&lease_release_safe),
                     )
                     .await;
                     if expected_publication_epoch.is_some() && !cleanup_ok {
@@ -719,6 +779,8 @@ where
                         expected_publication_epoch,
                         remote_lease_deadline,
                         scanner_publication_lease_fence.as_deref(),
+                        &remote_lease_tokens,
+                        Arc::clone(&lease_release_safe),
                     )
                     .await;
                     if expected_publication_epoch.is_some() && !cleanup_ok {
@@ -761,6 +823,8 @@ where
                         expected_publication_epoch,
                         remote_lease_deadline,
                         scanner_publication_lease_fence.as_deref(),
+                        &remote_lease_tokens,
+                        Arc::clone(&lease_release_safe),
                     )
                     .await;
                     if expected_publication_epoch.is_some() && !cleanup_ok {
@@ -778,12 +842,14 @@ where
 
         if backup_due {
             let done_save = Metrics::time(Metric::SaveUsage);
-            let backup_result = sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence(
+            let backup_result = sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence_and_scope(
                 &ctx,
                 storeapi.clone(),
                 expected_publication_epoch,
                 remote_lease_deadline,
                 scanner_publication_lease_fence.as_deref(),
+                remote_lease_tokens.clone(),
+                Arc::clone(&lease_release_safe),
             )
             .await;
             done_save();
@@ -817,6 +883,8 @@ async fn cleanup_observed_data_usage_snapshot_for_epoch_and_lease(
     expected_publication_epoch: Option<u64>,
     remote_lease_deadline: Option<std::time::Instant>,
     scanner_publication_lease_fence: Option<&str>,
+    remote_lease_tokens: &[Uuid],
+    lease_release_safe: Arc<AtomicBool>,
 ) -> bool {
     if remote_lease_expired(remote_lease_deadline) {
         return false;
@@ -885,7 +953,15 @@ async fn cleanup_observed_data_usage_snapshot_for_epoch_and_lease(
         return false;
     }
 
-    let result = delete_config_with_publication_admission_for_epoch(
+    let publication_scope = storeapi
+        .scanner_data_usage_publication_commit_scope_with_release_flag(
+            read_epoch,
+            scanner_publication_scope_deadline(data_usage_persist_timeout(), remote_lease_deadline),
+            remote_lease_tokens.to_vec(),
+            Arc::clone(&lease_release_safe),
+        )
+        .await;
+    let result = crate::delete_config_with_publication_scope_for_epoch(
         storeapi,
         RUSTFS_META_BUCKET,
         DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(),
@@ -904,9 +980,23 @@ async fn cleanup_observed_data_usage_snapshot_for_epoch_and_lease(
             ..Default::default()
         },
         read_epoch,
+        publication_scope.clone(),
     )
     .await;
 
+    let result = if let Some(scope) = publication_scope {
+        match scope.wait_for_completion().await {
+            ScannerPublicationCommitState::Committed | ScannerPublicationCommitState::AbortedBeforeCommit => result,
+            ScannerPublicationCommitState::Indeterminate
+            | ScannerPublicationCommitState::Admitted
+            | ScannerPublicationCommitState::InFlight => Err(EcstoreError::other(
+                "scanner publication cleanup scope did not reach a safe terminal state",
+            )),
+        }
+    } else {
+        result
+    };
+
     match result {
         Ok(_)
         | Err(

crates/scanner/src/scanner_io.rs

@@ -574,10 +574,6 @@ pub(crate) async fn scanner_set_disk_inventory(set: &SetDisks) -> Vec<Arc<Disk>>
 pub(crate) enum ScannerCycleDeferReason {
     ActivityBaselineUnavailable,
     DataMovement,
-    /// The configured persistence budget cannot fit within the fixed remote
-    /// publication-lease TTL. This is a deterministic configuration/contract
-    /// mismatch, not evidence that a peer activity probe failed.
-    PublicationLeaseBudgetExceeded,
     /// A granted lease's absolute deadline cannot cover the persistence
     /// operation. This can occur even when the configured budget fits the
     /// nominal TTL because lease acquisition consumed part of the window.
@@ -592,7 +588,6 @@ impl ScannerCycleDeferReason {
         match self {
             Self::ActivityBaselineUnavailable => "activity_baseline_unavailable",
             Self::DataMovement => "data_movement",
-            Self::PublicationLeaseBudgetExceeded => "publication_lease_budget_exceeded",
             Self::PublicationLeaseDeadlineExceeded => "publication_lease_deadline_exceeded",
             Self::PublicationLeaseReleaseFailed => "publication_lease_release_failed",
         }

crates/scanner/src/storage_api.rs

@@ -92,7 +92,10 @@ pub(crate) use rustfs_ecstore::api::event::{EventArgs as EcstoreEventArgs, send_
 pub(crate) use rustfs_ecstore::api::layout::{
     EndpointServerPools as EcstoreEndpointServerPools, Endpoints as EcstoreEndpoints, PoolEndpoints as EcstorePoolEndpoints,
 };
-pub(crate) use rustfs_ecstore::api::object::SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY;
+pub(crate) use rustfs_ecstore::api::notification::scanner_peer_transport_error_message_is_retryable;
+pub(crate) use rustfs_ecstore::api::object::{
+    SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, ScannerPublicationCommitScope, ScannerPublicationCommitState,
+};
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::rebalance::{
     RebalStatus as EcstoreRebalStatus, RebalanceInfo as EcstoreRebalanceInfo, RebalanceMeta as EcstoreRebalanceMeta,
@@ -106,9 +109,9 @@ pub(crate) use rustfs_ecstore::api::runtime::{
     setup_is_erasure_sd as ecstore_is_erasure_sd,
 };
 pub(crate) use rustfs_ecstore::api::set_disk::SetDisks as EcstoreSetDisks;
+pub(crate) use rustfs_ecstore::api::storage::ECStore as EcstoreStore;
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::storage::init_local_disks_with_instance_ctx as ecstore_init_local_disks_with_instance_ctx;
-pub(crate) use rustfs_ecstore::api::storage::{ECStore as EcstoreStore, SCANNER_PUBLICATION_LEASE_TTL_MS};
 use rustfs_storage_api as storage_contracts;
 
 pub(crate) mod owner {
@@ -125,7 +128,7 @@ pub(crate) mod owner {
         EcstoreNsScannerOpenRequest, EcstoreObjectLockConfiguration, EcstoreObjectOpts, EcstoreReplicationConfigurationExt,
         EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore,
         EcstoreVersioningApi, EcstoreVersioningConfiguration, SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY,
-        SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerReplicationHealObject, ScannerReplicationHealResult,
+        ScannerPublicationCommitScope, ScannerPublicationCommitState, ScannerReplicationHealObject, ScannerReplicationHealResult,
         ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, ecstore_apply_transition_rule, ecstore_expiry_state_handle,
         ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, ecstore_get_object_lock_config,
         ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache,

rustfs/src/admin/handlers/replication.rs

@@ -57,13 +57,6 @@ use url::Host;
 
 const SUPPORTED_REMOTE_TARGET_API: &str = "s3v4";
 
-/// Go encodes the zero `time.Time` as the year-1 instant
-/// (`0001-01-01T00:00:00Z`, possibly re-encoded with an offset); no real
-/// credential expiry lives in year 1, so any such timestamp means "unset".
-fn is_go_zero_time(timestamp: Timestamp) -> bool {
-    timestamp.to_zoned(jiff::tz::TimeZone::UTC).year() == 1
-}
-
 /// Field groups a `set-remote-target?update=true` request may modify, mirroring
 /// MinIO's `TargetUpdateType` / `GetTargetUpdateOps` query contract: the update
 /// overlays only the requested groups onto the stored target, so a client can
@@ -250,7 +243,7 @@ impl RemoteTargetRequest {
 
     fn into_bucket_target(self) -> S3Result<BucketTarget> {
         self.validate_connection_fields()?;
-        self.into_bucket_target_common()
+        self.into_bucket_target_common(true)
     }
 
     /// Partial-update parse: only the field groups named by `ops` are validated;
@@ -259,43 +252,18 @@ impl RemoteTargetRequest {
         if self.arn.trim().is_empty() {
             return Err(s3_error!(InvalidRequest, "arn is required for update"));
         }
-        if ops.contains(&TargetUpdateOp::Credentials) {
+        let replacing_credentials = ops.contains(&TargetUpdateOp::Credentials);
+        if replacing_credentials {
             self.validate_connection_fields()?;
         }
-        self.into_bucket_target_common()
+        self.into_bucket_target_common(replacing_credentials)
     }
 
-    fn into_bucket_target_common(self) -> S3Result<BucketTarget> {
+    fn into_bucket_target_common(self, validate_credentials: bool) -> S3Result<BucketTarget> {
         if !self.target_type.is_valid() {
             return Err(s3_error!(InvalidRequest, "type is invalid"));
         }
 
-        if self
-            .credentials
-            .session_token
-            .as_deref()
-            .is_some_and(|token| !token.trim().is_empty())
-        {
-            return Err(s3_error!(
-                InvalidRequest,
-                "remote target field credentials.session_token is not supported by this RustFS version"
-            ));
-        }
-
-        // Go's `omitempty` never elides a zero `time.Time`, so every madmin
-        // marshal carries `"expiration":"0001-01-01T00:00:00Z"`; only a real
-        // (non-year-1) expiry means the client wants expiring credentials.
-        if self
-            .credentials
-            .expiration
-            .is_some_and(|expiration| !is_go_zero_time(expiration))
-        {
-            return Err(s3_error!(
-                InvalidRequest,
-                "remote target field credentials.expiration is not supported by this RustFS version"
-            ));
-        }
-
         if !self.api.is_empty() && self.api != SUPPORTED_REMOTE_TARGET_API {
             return Err(s3_error!(
                 InvalidRequest,
@@ -317,9 +285,17 @@ impl RemoteTargetRequest {
         }
 
         let mut credentials = TargetCredentials::from(self.credentials);
-        // Past the check above the expiration can only be the zero-value
-        // sentinel, i.e. "no expiration" — never persist it.
-        credentials.expiration = None;
+        credentials.expiration = credentials.effective_expiration();
+        if validate_credentials && credentials.expiration.is_some() && credentials.effective_session_token().is_none() {
+            return Err(s3_error!(InvalidRequest, "credentials.expiration requires credentials.session_token"));
+        }
+        if validate_credentials
+            && credentials
+                .expiration
+                .is_some_and(|expiration| expiration <= Timestamp::now())
+        {
+            return Err(s3_error!(InvalidRequest, "credentials.expiration must be in the future"));
+        }
 
         Ok(BucketTarget {
             source_bucket: self.source_bucket,
@@ -2104,27 +2080,13 @@ mod tests {
 
     #[test]
     fn remote_target_request_rejects_unimplemented_fields() {
-        for (field, value, historical_field) in [
-            (
-                "credentials.session_token",
-                serde_json::json!("session-token"),
-                Some("credentials.sessionToken"),
-            ),
-            (
-                "credentials.expiration",
-                serde_json::json!("2026-01-01T00:00:00Z"),
-                Some("credentials.expiration"),
-            ),
-            ("api", serde_json::json!("s3v2"), None),
-            ("edge", serde_json::json!(true), None),
-            ("edgeSyncBeforeExpiry", serde_json::json!(true), None),
+        for (field, value) in [
+            ("api", serde_json::json!("s3v2")),
+            ("edge", serde_json::json!(true)),
+            ("edgeSyncBeforeExpiry", serde_json::json!(true)),
         ] {
             let mut request = valid_remote_target_request();
-            if let Some((credential_field, credential_name)) = field.split_once('.') {
-                request[credential_field][credential_name] = value;
-            } else {
-                request[field] = value;
-            }
+            request[field] = value;
             let request: RemoteTargetRequest =
                 serde_json::from_value(request).expect("unsupported field should still deserialize");
             let err = request
@@ -2133,15 +2095,86 @@ mod tests {
 
             assert!(err.to_string().contains(field));
             assert!(err.to_string().contains("not supported by this RustFS version"));
-            if let Some(historical_field) = historical_field {
-                assert!(
-                    REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS.contains(&historical_field),
-                    "rejected field {field} must be advertised as historical-only"
-                );
-            }
         }
     }
 
+    #[test]
+    fn remote_target_request_accepts_temporary_credentials() {
+        let mut request = valid_remote_target_request();
+        request["credentials"]["sessionToken"] = serde_json::json!("session-token");
+        request["credentials"]["expiration"] = serde_json::json!("2099-01-01T00:00:00Z");
+
+        let target = serde_json::from_value::<RemoteTargetRequest>(request)
+            .expect("temporary credentials should deserialize")
+            .into_bucket_target()
+            .expect("unexpired temporary credentials should be accepted");
+        let credentials = target.credentials.expect("credentials should be preserved");
+
+        assert_eq!(credentials.session_token.as_deref(), Some("session-token"));
+        assert_eq!(
+            serde_json::to_value(credentials.expiration.expect("expiration should be preserved"))
+                .expect("expiration should serialize"),
+            serde_json::json!("2099-01-01T00:00:00Z")
+        );
+        for field in ["credentials.sessionToken", "credentials.expiration"] {
+            assert!(REMOTE_TARGET_WRITABLE_FIELDS.contains(&field));
+        }
+    }
+
+    #[test]
+    fn remote_target_request_accepts_session_token_without_expiration() {
+        let mut request = valid_remote_target_request();
+        request["credentials"]["session_token"] = serde_json::json!("session-token");
+
+        let target = serde_json::from_value::<RemoteTargetRequest>(request)
+            .expect("temporary credentials should deserialize")
+            .into_bucket_target()
+            .expect("a session token without a reported expiration should remain compatible");
+
+        assert_eq!(
+            target
+                .credentials
+                .as_ref()
+                .and_then(|credentials| credentials.session_token.as_deref()),
+            Some("session-token")
+        );
+        assert!(target.credentials.and_then(|credentials| credentials.expiration).is_none());
+    }
+
+    #[test]
+    fn remote_target_request_rejects_expiration_without_session_token() {
+        let mut request = valid_remote_target_request();
+        request["credentials"]["expiration"] = serde_json::json!("2099-01-01T00:00:00Z");
+
+        let err = serde_json::from_value::<RemoteTargetRequest>(request)
+            .expect("request should deserialize")
+            .into_bucket_target()
+            .expect_err("an expiring credential bundle requires a session token");
+
+        assert!(
+            err.to_string()
+                .contains("credentials.expiration requires credentials.session_token")
+        );
+    }
+
+    #[test]
+    fn remote_target_request_rejects_expired_temporary_credentials_without_leaking_secrets() {
+        let mut request = valid_remote_target_request();
+        request["credentials"]["secretKey"] = serde_json::json!("secret-must-not-leak");
+        request["credentials"]["sessionToken"] = serde_json::json!("session-token-must-not-leak");
+        request["credentials"]["expiration"] = serde_json::json!("2000-01-01T00:00:00Z");
+
+        let err = serde_json::from_value::<RemoteTargetRequest>(request)
+            .expect("request should deserialize")
+            .into_bucket_target()
+            .expect_err("expired credentials must fail before persistence");
+        let message = err.to_string();
+
+        assert!(message.contains("credentials.expiration must be in the future"));
+        assert!(!message.contains("secret-must-not-leak"));
+        assert!(!message.contains("session-token-must-not-leak"));
+    }
+
     #[test]
     fn remote_target_request_accepts_real_madmin_add_marshal() {
         let request: RemoteTargetRequest =
@@ -2192,6 +2225,32 @@ mod tests {
         );
     }
 
+    #[test]
+    fn non_credential_update_accepts_redacted_temporary_credential_round_trip() {
+        // list-remote-targets redacts both the secret and session token but
+        // intentionally retains the non-secret expiration. mc echoes that
+        // shape on a sync-only update; the credentials group is not applied.
+        let body = serde_json::json!({
+            "endpoint": "192.168.1.10:9000",
+            "credentials": {
+                "accessKey": "access",
+                "expiration": "2099-01-01T00:00:00Z"
+            },
+            "targetbucket": "target",
+            "arn": "arn:rustfs:replication:us-east-1:dep:target",
+            "type": "replication",
+            "replicationSync": true
+        });
+
+        let target = serde_json::from_value::<RemoteTargetRequest>(body)
+            .expect("redacted mc round-trip should deserialize")
+            .into_update_bucket_target(&[TargetUpdateOp::Sync])
+            .expect("a sync-only update must ignore the redacted credential group");
+
+        assert!(target.replication_sync);
+        assert!(target.credentials.and_then(|credentials| credentials.expiration).is_some());
+    }
+
     #[test]
     fn remote_target_request_ignores_client_supplied_latency() {
         // Latency is a server-measured runtime stat; mc echoes the
@@ -2238,22 +2297,6 @@ mod tests {
         assert!(!target.ca_cert_pem.is_empty());
     }
 
-    #[test]
-    fn remote_target_request_validation_does_not_echo_credential_values() {
-        let mut request = valid_remote_target_request();
-        request["credentials"]["session_token"] = serde_json::json!("session-token-must-not-leak");
-
-        let request: RemoteTargetRequest = serde_json::from_value(request).expect("request should deserialize");
-        let err = request
-            .into_bucket_target()
-            .expect_err("session tokens must be rejected before persistence");
-        let message = err.to_string();
-
-        assert!(message.contains("credentials.session_token"));
-        assert!(!message.contains("session-token-must-not-leak"));
-        assert!(!message.contains("secret"));
-    }
-
     #[test]
     fn remote_target_request_accepts_go_duration_wire_values() {
         // `mc replicate add` defaults `--healthcheck-seconds` to 60; madmin
@@ -2514,6 +2557,10 @@ mod tests {
 
     #[test]
     fn remote_target_capability_fields_do_not_overlap() {
+        assert!(
+            REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS.is_empty(),
+            "v4 must not retain writable temporary-credential fields as historical-only"
+        );
         for field in REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS {
             assert!(
                 !REMOTE_TARGET_WRITABLE_FIELDS.contains(field),

rustfs/src/admin/handlers/system.rs

@@ -1306,8 +1306,8 @@ mod tests {
         assert_eq!(response.summary.manual_transition_jobs.state, CapabilityState::Supported);
         assert_eq!(response.replication.contract_version, 1);
         assert_eq!(response.replication.bucket_replication.contract_version, 1);
-        // v3: temporary-credential fields are explicitly historical-only.
-        assert_eq!(response.replication.remote_targets.contract_version, 3);
+        // v4: temporary-credential fields moved from historical-only to writable.
+        assert_eq!(response.replication.remote_targets.contract_version, 4);
         assert_eq!(response.replication.bucket_replication.status.state, CapabilityState::Supported);
         assert_eq!(response.replication.remote_targets.status.state, CapabilityState::Supported);
         assert_eq!(
@@ -1363,8 +1363,8 @@ mod tests {
                     .remote_targets
                     .fields
                     .iter()
-                    .any(|field| field.name == name && field.state == super::ReplicationFieldState::ReadOnlyHistorical),
-                "remote target field {name} must be advertised as historical-only"
+                    .any(|field| field.name == name && field.state == super::ReplicationFieldState::Supported),
+                "remote target field {name} must be advertised as writable"
             );
         }
         assert_eq!(response.manual_transition_jobs.contract_version, 1);
@@ -1428,7 +1428,7 @@ mod tests {
         assert_eq!(value["summary"]["manual_transition_jobs"]["state"], "supported");
         assert_eq!(value["replication"]["contract_version"], 1);
         assert_eq!(value["replication"]["bucket_replication"]["contract_version"], 1);
-        assert_eq!(value["replication"]["remote_targets"]["contract_version"], 3);
+        assert_eq!(value["replication"]["remote_targets"]["contract_version"], 4);
         assert_eq!(value["replication"]["bucket_replication"]["status"]["state"], "supported");
         assert_eq!(value["replication"]["remote_targets"]["status"]["state"], "supported");
         assert_eq!(
@@ -1469,8 +1469,8 @@ mod tests {
                     .as_array()
                     .expect("remote target fields should be an array")
                     .iter()
-                    .any(|field| field["name"] == name && field["state"] == "read_only_historical"),
-                "serialized remote target field {name} must be historical-only"
+                    .any(|field| field["name"] == name && field["state"] == "supported"),
+                "serialized remote target field {name} must be writable"
             );
         }
         assert_eq!(value["manual_transition_jobs"]["contract_version"], 1);

rustfs/src/storage/rpc/node_service.rs

@@ -123,6 +123,21 @@ fn verify_node_mutation_body<T: CanonicalMutationBody>(request: &Request<T>, ope
         .map_err(|err| Status::permission_denied(format!("{operation} authentication failed: {err}")))
 }
 
+fn start_decommission_failure_response(err: Error) -> StartDecommissionResponse {
+    match err {
+        Error::InvalidArgument(_, _, reason) => StartDecommissionResponse {
+            success: false,
+            error_info: Some(reason),
+            error_code: Some(ControlPlaneErrorCode::ControlPlaneErrorInvalidArgument as i32),
+        },
+        err => StartDecommissionResponse {
+            success: false,
+            error_info: Some(err.to_string()),
+            error_code: None,
+        },
+    }
+}
+
 fn supports_dynamic_config_rpc(sub_system: &str) -> bool {
     NOTIFY_SUB_SYSTEMS.contains(&sub_system)
         || matches!(
@@ -2334,11 +2349,7 @@ impl Node for NodeService {
                 success: true,
                 error_info: None,
             })),
-            Err(err) => Ok(Response::new(StartDecommissionResponse {
-                error_code: None,
-                success: false,
-                error_info: Some(err.to_string()),
-            })),
+            Err(err) => Ok(Response::new(start_decommission_failure_response(err))),
         }
     }
 
@@ -2456,7 +2467,7 @@ mod tests {
         initialize_heal_topology_fingerprint, initialize_heal_topology_fingerprint_with_probe, legacy_scanner_activity_response,
         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, stop_rebalance_response,
+        scanner_activity_response_v7, start_decommission_failure_response, stop_rebalance_response,
     };
     use crate::storage::rpc::node_service::heal::heal_topology_fingerprint;
     use crate::storage::storage_api::rpc_consumer::node_service::{DiskError, HealBucketInfo};
@@ -2479,14 +2490,14 @@ mod tests {
     use rustfs_protos::models::PingBodyBuilder;
     use rustfs_protos::proto_gen::node_service::{
         BackgroundHealStatusRequest, BatchGenerallyLockRequest, CancelDecommissionRequest, CheckPartsRequest,
-        ClearDecommissionRequest, DeleteBucketMetadataRequest, DeleteBucketRequest, DeletePathsRequest, DeletePolicyRequest,
-        DeleteRequest, DeleteServiceAccountRequest, DeleteUserRequest, DeleteVersionRequest, DeleteVersionsRequest,
-        DeleteVolumeRequest, DiskInfoRequest, DownloadProfileDataRequest, GenerallyLockRequest, GetAllBucketStatsRequest,
-        GetBucketInfoRequest, GetBucketStatsDataRequest, GetCpusRequest, GetMemInfoRequest, GetMetacacheListingRequest,
-        GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest,
-        GetSrMetricsDataRequest, GetSysConfigRequest, GetSysErrorsRequest, HealBucketRequest, HealControlRequest,
-        ListBucketRequest, ListDirRequest, ListVolumesRequest, LoadBucketMetadataRequest, LoadGroupRequest,
-        LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest,
+        ClearDecommissionRequest, ControlPlaneErrorCode, DeleteBucketMetadataRequest, DeleteBucketRequest, DeletePathsRequest,
+        DeletePolicyRequest, DeleteRequest, DeleteServiceAccountRequest, DeleteUserRequest, DeleteVersionRequest,
+        DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, DownloadProfileDataRequest, GenerallyLockRequest,
+        GetAllBucketStatsRequest, GetBucketInfoRequest, GetBucketStatsDataRequest, GetCpusRequest, GetMemInfoRequest,
+        GetMetacacheListingRequest, GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest,
+        GetProcInfoRequest, GetSeLinuxInfoRequest, GetSrMetricsDataRequest, GetSysConfigRequest, GetSysErrorsRequest,
+        HealBucketRequest, HealControlRequest, ListBucketRequest, ListDirRequest, ListVolumesRequest, LoadBucketMetadataRequest,
+        LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest,
         LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, MakeBucketRequest, MakeVolumeRequest,
         MakeVolumesRequest, Mss, PingRequest, PreparePartTransactionRequest, ReadAllRequest, ReadAtRequest, ReadMultipleRequest,
         ReadVersionRequest, ReadXlRequest, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, RenameDataRequest,
@@ -2576,6 +2587,20 @@ mod tests {
         }
     }
 
+    #[test]
+    fn start_decommission_failure_response_preserves_invalid_argument_reason() {
+        let reason = "durable unresolved-entry recovery requires pool metadata V2 or V3";
+        let response = start_decommission_failure_response(Error::InvalidArgument(
+            "decommission".to_string(),
+            "pool-metadata-version".to_string(),
+            reason.to_string(),
+        ));
+
+        assert!(!response.success);
+        assert_eq!(response.error_info.as_deref(), Some(reason));
+        assert_eq!(response.error_code, Some(ControlPlaneErrorCode::ControlPlaneErrorInvalidArgument as i32));
+    }
+
     struct HealControlMockStorage;
 
     #[async_trait::async_trait]

rustfs/src/storage/rpc/node_service/disk.rs

@@ -33,6 +33,7 @@ use rustfs_io_metrics::internode_metrics::{
 use rustfs_protos::proto_gen::node_service::*;
 use serde::de::DeserializeOwned;
 use std::io::Cursor;
+use std::sync::Arc;
 use std::time::Instant;
 use tonic::{Request, Response, Status};
 use tracing::debug;
@@ -1230,37 +1231,40 @@ impl NodeService {
             // The target owns this read guard.  It must span the complete
             // disk rename, not merely the preflight, so a movement transition
             // cannot restart after validation and before rename linearization.
-            let _scanner_publication_lease_guard = if let Some(token) = scanner_publication_lease_token {
-                let Some(store) = self.resolve_object_store() else {
-                    return Ok(Response::new(RenameDataResponse {
-                        success: false,
-                        rename_data_resp: String::new(),
-                        rename_data_resp_bin: Vec::new().into(),
-                        error: Some(DiskError::other("scanner publication lease owner is unavailable").into()),
-                    }));
-                };
-                match store.acquire_scanner_publication_lease_guard(token).await {
-                    Ok(guard) => Some(guard),
-                    Err(err) => {
+            let scanner_publication_lease_guard: Option<Arc<dyn Send + Sync>> =
+                if let Some(token) = scanner_publication_lease_token {
+                    let Some(store) = self.resolve_object_store() else {
                         return Ok(Response::new(RenameDataResponse {
                             success: false,
                             rename_data_resp: String::new(),
                             rename_data_resp_bin: Vec::new().into(),
-                            error: Some(DiskError::other(err.to_string()).into()),
+                            error: Some(DiskError::other("scanner publication lease owner is unavailable").into()),
                         }));
+                    };
+                    match store.acquire_scanner_publication_lease_guard(token).await {
+                        Ok(guard) => Some(Arc::new(guard)),
+                        Err(err) => {
+                            return Ok(Response::new(RenameDataResponse {
+                                success: false,
+                                rename_data_resp: String::new(),
+                                rename_data_resp_bin: Vec::new().into(),
+                                error: Some(DiskError::other(err.to_string()).into()),
+                            }));
+                        }
                     }
-                }
-            } else {
-                None
-            };
+                } else {
+                    None
+                };
             let request_decoded_from_msgpack = decoded_file_info.from_msgpack;
             match disk
-                .rename_data(
+                .rename_data_borrowed_with_fence_and_guard(
                     &request.src_volume,
                     &request.src_path,
                     &decoded_file_info.value,
                     &request.dst_volume,
                     &request.dst_path,
+                    scanner_publication_lease_token,
+                    scanner_publication_lease_guard,
                 )
                 .await
             {
@@ -1641,26 +1645,36 @@ impl NodeService {
             // The target-side guard spans the complete delete operation. A
             // lease expiry or movement transition cannot occur between this
             // validation and the disk delete linearization point.
-            let _scanner_publication_lease_guard = if let Some(token) = scanner_publication_lease_token {
-                let Some(store) = self.resolve_object_store() else {
-                    return Ok(Response::new(DeleteResponse {
-                        success: false,
-                        error: Some(DiskError::other("scanner publication lease owner is unavailable").into()),
-                    }));
-                };
-                match store.acquire_scanner_publication_lease_guard(token).await {
-                    Ok(guard) => Some(guard),
-                    Err(err) => {
+            let scanner_publication_lease_guard: Option<Arc<dyn Send + Sync>> =
+                if let Some(token) = scanner_publication_lease_token {
+                    let Some(store) = self.resolve_object_store() else {
                         return Ok(Response::new(DeleteResponse {
                             success: false,
-                            error: Some(DiskError::other(err.to_string()).into()),
+                            error: Some(DiskError::other("scanner publication lease owner is unavailable").into()),
                         }));
+                    };
+                    match store.acquire_scanner_publication_lease_guard(token).await {
+                        Ok(guard) => Some(Arc::new(guard)),
+                        Err(err) => {
+                            return Ok(Response::new(DeleteResponse {
+                                success: false,
+                                error: Some(DiskError::other(err.to_string()).into()),
+                            }));
+                        }
                     }
-                }
-            } else {
-                None
-            };
-            match disk.delete(&request.volume, &request.path, options).await {
+                } else {
+                    None
+                };
+            match disk
+                .delete_with_scanner_publication_lease_and_guard(
+                    &request.volume,
+                    &request.path,
+                    options,
+                    scanner_publication_lease_token,
+                    scanner_publication_lease_guard,
+                )
+                .await
+            {
                 Ok(_) => Ok(Response::new(DeleteResponse {
                     success: true,
                     error: None,

rustfs/src/storage/storage_api.rs

@@ -1301,14 +1301,6 @@ pub(crate) trait StorageDiskRpcExt {
     async fn list_volumes(&self) -> DiskResult<Vec<VolumeInfo>>;
     async fn make_volume(&self, volume: &str) -> DiskResult<()>;
     async fn make_volumes(&self, volume: Vec<&str>) -> DiskResult<()>;
-    async fn rename_data(
-        &self,
-        src_volume: &str,
-        src_path: &str,
-        file_info: &rustfs_filemeta::FileInfo,
-        dst_volume: &str,
-        dst_path: &str,
-    ) -> DiskResult<RenameDataResp>;
     async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult<Vec<String>>;
     async fn read_file(&self, volume: &str, path: &str) -> DiskResult<FileReader>;
     async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> DiskResult<FileReader>;
@@ -1452,17 +1444,6 @@ where
         ecstore_disk::DiskAPI::make_volumes(self, volume).await
     }
 
-    async fn rename_data(
-        &self,
-        src_volume: &str,
-        src_path: &str,
-        file_info: &rustfs_filemeta::FileInfo,
-        dst_volume: &str,
-        dst_path: &str,
-    ) -> DiskResult<RenameDataResp> {
-        ecstore_disk::DiskAPI::rename_data(self, src_volume, src_path, file_info.clone(), dst_volume, dst_path).await
-    }
-
     async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult<Vec<String>> {
         ecstore_disk::DiskAPI::list_dir(self, origvolume, volume, dir_path, count).await
     }