test(put): align eager boundary assertions

Co-Authored-By: heihutu <heihutu@gmail.com>

.config/e2e-full-selection.txt

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

.config/nextest.toml

@@ -100,16 +100,6 @@ 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]]
@@ -242,12 +232,6 @@ 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,7 +38,6 @@ 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:
@@ -87,7 +86,6 @@ jobs:
 
       - name: Run KMS suite
         id: test
-        continue-on-error: true
         env:
           LOG_FILE: /tmp/rustfs-kms.log
         run: |
@@ -121,65 +119,12 @@ 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'
@@ -188,129 +133,6 @@ 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,12 +33,10 @@ 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:
@@ -78,7 +76,6 @@ jobs:
 
       - name: Run S3 compatibility suite
         id: test
-        continue-on-error: true
         env:
           LOG_FILE: /tmp/rustfs-s3-compat.log
         run: |
@@ -112,68 +109,12 @@ 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'
@@ -182,129 +123,6 @@ 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,7 +38,6 @@ 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:
@@ -61,8 +60,6 @@ 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
@@ -80,35 +77,18 @@ 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-clients
+            sudo apt-get install -y mosquitto mosquitto-clients
           fi
-          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
-          }
+          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; }
 
       - name: Run tier suite
         id: test
-        continue-on-error: true
         env:
           LOG_FILE: /tmp/rustfs-tier.log
         run: |
@@ -142,65 +122,12 @@ 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'
@@ -209,129 +136,6 @@ 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
@@ -346,8 +150,6 @@ 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, scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe,
+        new_global_notification_sys, start_remote_version_state_fleet_probe,
     };
 }
 
@@ -416,10 +416,9 @@ pub mod object {
         GetObjectBodyCacheHookLookup, GetObjectBodySource, GetObjectReader, NamespaceLockFence, ObjectEncryptionResolver,
         ObjectInfo, ObjectLockConfigSnapshot, ObjectMutationHook, ObjectOptions, PutObjReader, QuotaAdmission,
         RangedDecompressReader, ReadEncryptionMaterial, ReadEncryptionMode, ReadEncryptionRequest,
-        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,
+        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,
     };
     pub use crate::store::{
         PrepareSelectObjectSnapshotError, PreparedGetObjectReader, SelectObjectSnapshot, SelectObjectSnapshotReadError,

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -22,7 +22,6 @@ 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;
@@ -78,7 +77,7 @@ use std::str::FromStr as _;
 use std::sync::Arc;
 use std::sync::OnceLock;
 use std::sync::Weak;
-use std::time::{Duration, Instant, SystemTime};
+use std::time::{Duration, Instant};
 use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 use tokio::sync::Mutex;
 use tokio::sync::RwLock;
@@ -90,71 +89,6 @@ 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>>;
@@ -911,26 +845,13 @@ 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;
-        let err = arn_errs.entry(arn.to_string()).or_insert_with(|| ArnErrs {
-            count: 1,
+        arn_errs.entry(arn.to_string()).or_insert_with(|| ArnErrs {
             bucket: bucket.to_string(),
-            ..Default::default()
+            update_in_progress: true,
+            count: 1,
         });
-        err.update_in_progress = true;
-        err.bucket = bucket.to_string();
     }
 
     pub async fn mark_refresh_done(&self, bucket: &str, arn: &str) {
@@ -942,21 +863,15 @@ impl BucketTargetSys {
     }
 
     pub async fn is_reloading_target(&self, _bucket: &str, arn: &str) -> bool {
-        self.arn_errs_map
-            .read()
-            .await
-            .get(arn)
-            .is_some_and(|err| err.update_in_progress)
+        let arn_errs = self.arn_errs_map.read().await;
+        arn_errs.get(arn).map(|err| err.update_in_progress).unwrap_or(false)
     }
 
-    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;
-        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();
+        if let Some(err) = arn_errs.get_mut(arn) {
+            err.count += 1;
+        }
     }
 
     pub async fn get_remote_target_client(&self, bucket: &str, arn: &str) -> Option<Arc<TargetClient>> {
@@ -969,15 +884,15 @@ impl BucketTargetSys {
                 .unwrap_or((None, None))
         };
 
-        let credentials_expired = cli
-            .as_ref()
-            .is_some_and(|client| client.credentials_expired_at(jiff::Timestamp::now()));
-        if let Some(cli) = cli
-            && !credentials_expired
-        {
+        if let Some(cli) = cli {
             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) {
@@ -985,24 +900,16 @@ 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.update_all_targets_locked(bucket, Some(&bucket_targets)).await;
+                self.mark_refresh_in_progress(bucket, arn).await;
+                self.update_all_targets(bucket, Some(&bucket_targets)).await;
+                self.mark_refresh_done(bucket, arn).await;
             }
             Err(e) => {
                 error!("get bucket targets config error:{}", e);
             }
         };
-        drop(update_guard);
 
         let cli = self
             .arn_remotes_map
@@ -1010,10 +917,8 @@ impl BucketTargetSys {
             .await
             .get(arn)
             .and_then(|target| target.client.clone());
-        if let Some(cli) = cli
-            && !cli.credentials_expired_at(jiff::Timestamp::now())
-        {
-            return Some(cli);
+        if cli.is_some() {
+            return cli;
         }
 
         self.inc_arn_errs(bucket, arn).await;
@@ -1043,13 +948,12 @@ impl BucketTargetSys {
             });
         };
 
-        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 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 endpoint = if target.secure {
             format!("https://{}", target.endpoint)
@@ -1069,7 +973,7 @@ impl BucketTargetSys {
 
         let mut config_builder = S3Config::builder()
             .endpoint_url(endpoint.clone())
-            .credentials_provider(SharedCredentialsProvider::new(RemoteTargetCredentialsProvider { credentials: creds }))
+            .credentials_provider(SharedCredentialsProvider::new(creds))
             .region(SdkRegion::new(target.region.clone()))
             .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest());
 
@@ -1143,13 +1047,6 @@ 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 {
@@ -1181,17 +1078,6 @@ 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(
@@ -1204,6 +1090,11 @@ 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,
@@ -2071,13 +1962,6 @@ 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()
     }
@@ -2673,26 +2557,6 @@ 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 {
@@ -2718,150 +2582,6 @@ 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};
 
@@ -3793,29 +3513,6 @@ 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
@@ -3854,88 +3551,6 @@ 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,21 +436,16 @@ pub(crate) async fn check_replicate_delete_strict(
     }
 
     for target in decision.targets_map.values_mut() {
-        let replicate_sync = ReplicationTargetStore::remote_target_client(bucket, &target.arn)
-            .await
-            .map(|client| client.replicate_sync);
-        apply_target_delivery_mode(target, replicate_sync);
+        if let Some(client) = ReplicationTargetStore::remote_target_client(bucket, &target.arn).await {
+            target.synchronous = client.replicate_sync;
+        } else {
+            target.replicate = false;
+            target.synchronous = false;
+        }
     }
     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,
@@ -634,23 +629,6 @@ 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,7 +102,6 @@ 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";
@@ -1848,7 +1847,19 @@ pub(crate) async fn replicate_delete_with_outcome<S: ReplicationStorage>(
                 reason = "target_client_missing",
                 "Skipping replication delete because target client is unavailable"
             );
-            rinfos.targets.push(unavailable_delete_target_info(&dobj, &tgt_entry.arn));
+            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()
+            });
             continue;
         };
 
@@ -2573,32 +2584,6 @@ 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()
@@ -2811,10 +2796,6 @@ 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 {
@@ -2822,8 +2803,7 @@ 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 FAILED state below preserves retry visibility and the
-            // aggregate result emits the user-visible failure event once.
+            // on it; the per-object event below still records each dropped object.
             debug!(
                 event = EVENT_RESYNC_RUNTIME_SKIPPED,
                 component = LOG_COMPONENT_ECSTORE,
@@ -2832,9 +2812,15 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
                 object = %object,
                 arn = %arn,
                 reason = "target_client_missing",
-                "Replication target client unavailable"
+                "Replication rule has no bucket target for its destination ARN; object not replicated"
             );
-            rinfos.targets.push(unavailable_object_target_info(&roi, &arn));
+            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()
+            });
             continue;
         };
 
@@ -2849,6 +2835,11 @@ 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) => {
@@ -2954,23 +2945,6 @@ 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,
@@ -4183,88 +4157,6 @@ 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,8 +25,6 @@ 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 {
@@ -43,26 +41,6 @@ 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(),
@@ -377,24 +355,6 @@ 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,14 +122,6 @@ 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),
@@ -2343,29 +2335,6 @@ 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,34 +176,6 @@ 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>,
@@ -2055,7 +2027,6 @@ 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,
 }
@@ -2074,7 +2045,6 @@ 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(),
         });
@@ -2091,14 +2061,6 @@ 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)]
@@ -2128,21 +2090,6 @@ 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);
 }
@@ -6242,7 +6189,6 @@ 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])
@@ -6393,7 +6339,6 @@ 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();
@@ -7095,14 +7040,10 @@ 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;
-            let indices = resumable_decommission_queue_indices(&pool_meta)
+            resumable_decommission_queue_indices(&pool_meta)
                 .into_iter()
                 .filter(|idx| indices.contains(idx))
-                .collect::<Vec<_>>();
-            if !indices.is_empty() {
-                ensure_decommission_ledger_persistence_supported(&pool_meta)?;
-            }
-            indices
+                .collect::<Vec<_>>()
         };
         if indices.is_empty() {
             return Ok(Vec::new());
@@ -9358,11 +9299,8 @@ 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());
@@ -10918,98 +10856,10 @@ 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 {
@@ -11064,7 +10914,6 @@ 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 {
@@ -11377,35 +11226,6 @@ 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();
@@ -14674,113 +14494,6 @@ 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]);
@@ -17774,7 +17487,6 @@ 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,
@@ -17832,7 +17544,6 @@ 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,40 +250,6 @@ 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 {
@@ -307,49 +273,6 @@ 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 {
@@ -755,20 +678,17 @@ 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 {
-            // 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);
+            let version = slot.version.load(Ordering::Acquire);
             if !version.is_multiple_of(2) {
                 std::hint::spin_loop();
                 continue;
             }
             if slot
                 .version
-                .compare_exchange(version, version.wrapping_add(1), Ordering::AcqRel, Ordering::Relaxed)
+                .compare_exchange(version, version.wrapping_add(1), Ordering::AcqRel, Ordering::Acquire)
                 .is_ok()
             {
-                if slot.unix_sec.load(Ordering::Relaxed) != now_sec {
+                if slot.unix_sec.load(Ordering::Acquire) != now_sec {
                     slot.count.store(0, Ordering::Relaxed);
                     slot.acc_time.store(0, Ordering::Relaxed);
                     slot.unix_sec.store(now_sec, Ordering::Release);
@@ -784,10 +704,14 @@ impl DiskOperationMetrics {
     fn last_minute_snapshot(&self, now_sec: u64) -> TimedAction {
         let mut snapshot = TimedAction::default();
         for slot in &self.last_minute {
-            let Some((slot_sec, count, acc_time)) = slot.snapshot() else {
+            let version = slot.version.load(Ordering::Acquire);
+            if !version.is_multiple_of(2) {
                 continue;
-            };
-            if slot_sec <= now_sec && now_sec.saturating_sub(slot_sec) < 60 {
+            }
+            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 {
                 snapshot.count = snapshot.count.saturating_add(count);
                 snapshot.acc_time = snapshot.acc_time.saturating_add(acc_time);
             }
@@ -796,23 +720,6 @@ 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,
 }
@@ -1190,37 +1097,6 @@ 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,
@@ -2371,44 +2247,6 @@ 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,10 +3148,9 @@ impl LocalIoBackend for StdBackend {
                 direct_read_copy_fault_delta: MmapPageFaultDelta,
                 blocking_task_duration: StdDuration,
                 used_direct_io: bool,
-                /// 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>>,
+                /// 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>>,
             }
 
             enum MmapCopyReadError {
@@ -3198,12 +3197,12 @@ impl LocalIoBackend for StdBackend {
                 (cache, key, gen_at_open)
             });
             #[cfg(target_os = "linux")]
-            let cached_fd: Option<Arc<FdCacheEntry>> = match &fd_lookup {
+            let cached_fd: Option<Arc<std::fs::File>> = match &fd_lookup {
                 Some((cache, key, _)) => cache.get(key).await,
                 None => None,
             };
             #[cfg(not(target_os = "linux"))]
-            let cached_fd: Option<Arc<FdCacheEntry>> = None;
+            let cached_fd: Option<Arc<std::fs::File>> = None;
 
             let blocking_wait_start = metrics_enabled.then(std::time::Instant::now);
             let read_result = tokio::task::spawn_blocking(move || {
@@ -3225,15 +3224,8 @@ 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.
-                // 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,
-                    )
+                let (file, access_check_duration) = if let Some(cached) = cached_fd.as_ref() {
+                    (cached.as_ref().try_clone().map_err(DiskError::from)?, StdDuration::ZERO)
                 } else {
                     // Measure the volume access probe only — the part-path resolution
                     // above is accounted in `path_resolve_duration` (rustfs/backlog#1801).
@@ -3244,27 +3236,20 @@ 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)?, None, access_check_duration)
+                    (std::fs::File::open(&file_path).map_err(DiskError::from)?, access_check_duration)
                 };
                 let file_open_duration = file_open_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_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_validate_start = metrics_enabled.then(StdInstant::now);
-                if metadata_len < end_offset_u64 {
-                    return Err(MmapCopyReadError::OutOfBounds {
-                        actual_size: metadata_len,
-                    });
+                if meta.len() < end_offset_u64 {
+                    return Err(MmapCopyReadError::OutOfBounds { actual_size: meta.len() });
                 }
                 let metadata_validate_duration =
                     metadata_validate_start.map_or(StdDuration::ZERO, |started_at| started_at.elapsed());
@@ -3410,14 +3395,9 @@ 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<FdCacheEntry>> = cached_fd.is_none().then(|| {
-                    Arc::new(FdCacheEntry {
-                        file: Arc::new(file),
-                        len: metadata_len,
-                    })
-                });
+                let opened_fd: Option<Arc<std::fs::File>> = cached_fd.is_none().then(|| Arc::new(file));
                 #[cfg(not(target_os = "linux"))]
-                let opened_fd: Option<Arc<FdCacheEntry>> = None;
+                let opened_fd: Option<Arc<std::fs::File>> = None;
 
                 Ok::<MmapCopyReadResult, MmapCopyReadError>(MmapCopyReadResult {
                     bytes,
@@ -3540,7 +3520,7 @@ impl LocalIoBackend for StdBackend {
                     }
                 }
             }
-            // Index the freshly opened descriptor and metadata snapshot for future cache hits
+            // Index the freshly opened descriptor 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
@@ -3892,18 +3872,6 @@ 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
@@ -3933,7 +3901,7 @@ struct FdCacheEntry {
 /// the descriptor once no in-flight read still holds it.
 #[cfg(target_os = "linux")]
 struct FdCache {
-    cache: moka::future::Cache<FdKey, Arc<FdCacheEntry>>,
+    cache: moka::future::Cache<FdKey, Arc<std::fs::File>>,
     /// 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
@@ -3963,7 +3931,7 @@ impl FdCache {
         }
     }
 
-    async fn get(&self, key: &FdKey) -> Option<Arc<FdCacheEntry>> {
+    async fn get(&self, key: &FdKey) -> Option<Arc<std::fs::File>> {
         self.cache.get(key).await
     }
 
@@ -3978,11 +3946,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, entry: Arc<FdCacheEntry>, gen_at_open: u64) {
+    async fn insert_if_fresh(&self, key: FdKey, file: Arc<std::fs::File>, gen_at_open: u64) {
         if self.generation.load(Ordering::Acquire) != gen_at_open {
             return;
         }
-        self.cache.insert(key.clone(), entry).await;
+        self.cache.insert(key.clone(), file).await;
         if self.generation.load(Ordering::Acquire) != gen_at_open {
             self.cache.invalidate(&key).await;
         }
@@ -4018,7 +3986,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<FdCacheEntry>| {
+        let matches = move |k: &FdKey, _: &Arc<std::fs::File>| {
             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() {
@@ -4034,7 +4002,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<FdCacheEntry>| k.volume == volume;
+        let matches = move |k: &FdKey, _: &Arc<std::fs::File>| k.volume == volume;
         if self.cache.invalidate_entries_if(matches).is_err() {
             self.cache.invalidate_all();
         }
@@ -4052,8 +4020,7 @@ impl FdCache {
     /// tests that drive the cache directly.
     #[cfg(test)]
     async fn insert(&self, key: FdKey, file: Arc<std::fs::File>) {
-        let len = file.metadata().map(|metadata| metadata.len()).unwrap_or_default();
-        self.cache.insert(key, Arc::new(FdCacheEntry { file, len })).await;
+        self.cache.insert(key, file).await;
     }
 
     #[cfg(test)]
@@ -4432,12 +4399,7 @@ impl UringBackend {
         };
 
         let file = match cached {
-            Some(entry) => {
-                if entry.len < u64::try_from(end_offset).map_err(|_| DiskError::FileCorrupt)? {
-                    return Err(DiskError::FileCorrupt);
-                }
-                Arc::clone(&entry.file)
-            }
+            Some(file) => file,
             None => {
                 // Snapshot the cache generation BEFORE opening (rustfs/backlog#1176):
                 // if a heal/delete invalidation runs while this open is in flight,
@@ -4447,7 +4409,7 @@ impl UringBackend {
                 let root = self.root.clone();
                 let volume_owned = volume.to_owned();
                 let path_owned = path.to_owned();
-                let (file, len) = tokio::task::spawn_blocking(move || -> Result<(std::fs::File, u64)> {
+                let file = tokio::task::spawn_blocking(move || -> Result<std::fs::File> {
                     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)?;
@@ -4455,22 +4417,30 @@ impl UringBackend {
                     if meta.len() < end_offset_u64 {
                         return Err(DiskError::FileCorrupt);
                     }
-                    Ok((file, meta.len()))
+                    Ok(file)
                 })
                 .await
                 .map_err(|e| DiskError::other(format!("uring pread join error: {e}")))??;
-                let file = Arc::new(FdCacheEntry {
-                    file: Arc::new(file),
-                    len,
-                });
+                let file = Arc::new(file);
                 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.clone()
+                file
             }
         };
 
         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());
         }
 
@@ -21437,10 +21407,11 @@ 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. 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.
+    /// 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.
     #[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,20 +677,15 @@ impl DiskAPI for Disk {
 }
 
 impl Disk {
-    pub async fn delete_with_scanner_publication_lease_and_guard(
+    pub(crate) async fn delete_with_scanner_publication_lease(
         &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_with_publication_guard(volume, path, opts, external_guard)
-                    .await
-            }
+            Disk::Local(local_disk) => local_disk.delete(volume, path, opts).await,
             Disk::Remote(remote_disk) => {
                 remote_disk
                     .delete_with_scanner_publication_lease(volume, path, opts, scanner_publication_lease_token)
@@ -719,34 +714,11 @@ 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_with_guard(src_volume, src_path, fi, dst_volume, dst_path, external_guard)
+                    .rename_data_borrowed(src_volume, src_path, fi, dst_volume, dst_path)
                     .await
             }
             Disk::Remote(remote_disk) => {

crates/ecstore/src/object_api/types.rs

@@ -19,9 +19,6 @@ 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 {
@@ -350,337 +347,6 @@ 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
@@ -718,11 +384,6 @@ 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
@@ -812,7 +473,6 @@ 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,12 +1962,6 @@ 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,8 +1667,6 @@ 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,17 +111,6 @@ 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,7 +3657,6 @@ 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> {
@@ -3668,17 +3667,8 @@ 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)")]
@@ -3788,37 +3778,6 @@ 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
@@ -4036,7 +3995,6 @@ 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()
@@ -4394,7 +4352,6 @@ impl SetDisks {
         let RenameDataFenceOptions {
             write_quorum,
             scanner_publication_lease_tokens,
-            scanner_publication_commit_scope,
         } = fence_options;
         if let Some(file_info) = disks
             .iter()
@@ -4426,15 +4383,11 @@ 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
@@ -4448,7 +4401,6 @@ 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
@@ -4481,13 +4433,6 @@ 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(
@@ -5896,8 +5841,7 @@ 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, None)
-            .await
+        self.delete_prefix_with_scanner_publication_lease(bucket, prefix, None).await
     }
 
     /// Delete a prefix with an optional per-remote-disk scanner publication
@@ -5908,7 +5852,6 @@ 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;
@@ -5917,21 +5860,11 @@ 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 {
-                    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(
+                    disk.delete_with_scanner_publication_lease(
                         &bucket,
                         &prefix,
                         DeleteOptions {
@@ -5940,7 +5873,6 @@ impl SetDisks {
                             ..Default::default()
                         },
                         scanner_publication_lease_token,
-                        external_guard,
                     )
                     .await
                 } else {
@@ -5949,10 +5881,7 @@ impl SetDisks {
             });
         }
 
-        run_scanner_publication_delete_owner(scanner_publication_commit_scope, move || async move {
-            Self::reduce_delete_prefix_results(join_all(futures).await, write_quorum)
-        })
-        .await
+        Self::reduce_delete_prefix_results(join_all(futures).await, write_quorum)
     }
 
     /// Scan a single disk's copy of `prefix` and decide whether it is an orphan
@@ -6880,63 +6809,6 @@ 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,44 +3938,6 @@ 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,7 +66,6 @@ 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;
@@ -273,22 +272,6 @@ 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);
@@ -2644,10 +2627,6 @@ 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;
 
@@ -3418,16 +3397,8 @@ 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_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_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_write_path_label = write_path.metric_label();
             let commit_is_versioned = opts.versioned || opts.version_suspended;
             let commit_versioned = opts.versioned;
@@ -3523,7 +3494,7 @@ impl SetDisks {
                     }
                     Ok(())
                 };
-                let mut pre_rename_result = if cancellation.is_some() || request_cancellation.is_some() {
+                let 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()) => {
@@ -3534,20 +3505,6 @@ 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,
@@ -3583,17 +3540,9 @@ 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;
@@ -3908,11 +3857,6 @@ 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 })
@@ -7110,11 +7054,6 @@ 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.
@@ -7209,7 +7148,6 @@ 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;
@@ -7224,7 +7162,6 @@ 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 {
@@ -7236,14 +7173,10 @@ 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());
             }
@@ -7251,19 +7184,10 @@ 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)?;
-            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]))?;
+            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]))?;
 
-            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());
         }
@@ -7336,14 +7260,10 @@ 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));
         }
 
@@ -7417,7 +7337,6 @@ 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]))?;
@@ -7429,9 +7348,6 @@ 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);
         }
 
@@ -7457,7 +7373,6 @@ 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]))?;
@@ -7483,9 +7398,6 @@ 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, ScannerPublicationCommitScope},
+    object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader},
 };
 use futures::future::join_all;
 use http::HeaderMap;
@@ -522,48 +522,6 @@ 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.
@@ -1451,29 +1409,9 @@ 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()));
@@ -1484,102 +1422,6 @@ 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,9 +1553,6 @@ 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.
@@ -1566,7 +1563,6 @@ 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.
@@ -1574,7 +1570,6 @@ 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,9 +30,6 @@ 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,17 +60,16 @@ pub const REPLICATION_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &[
     "Destination.ReplicationTime",
 ];
 
-// 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;
+// 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;
 
 pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
     "sourcebucket",
     "endpoint",
     "credentials.accessKey",
     "credentials.secretKey",
-    "credentials.sessionToken",
-    "credentials.expiration",
     "targetbucket",
     "secure",
     "path",
@@ -92,12 +91,7 @@ pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
     "disableProxy",
 ];
 
-/// 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_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &["credentials.sessionToken", "credentials.expiration"];
 
 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,
-    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,
+    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,7 +55,6 @@ use storage_api::owner::{
     ecstore_new_disk,
 };
 use tokio_util::sync::CancellationToken;
-use uuid::Uuid;
 
 pub mod data_usage_define;
 pub mod error;
@@ -753,32 +752,6 @@ 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,
@@ -789,7 +762,10 @@ pub(crate) async fn delete_config_with_publication_admission_for_epoch<S>(
 where
     S: ScannerObjectIO + ScannerConfigObjectDelete,
 {
-    delete_config_with_publication_scope_for_epoch(api, bucket, object, opts, expected_epoch, None).await
+    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
 }
 
 /// Capture the storage-owned publication epoch without retaining the read
@@ -820,14 +796,13 @@ where
     Some(admission)
 }
 
-pub(crate) async fn save_config_shared_with_preconditions_and_lease_fence_and_scope<S>(
+pub(crate) async fn save_config_shared_with_preconditions_and_lease_fence<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,
@@ -847,7 +822,6 @@ where
         &ScannerObjectOptions {
             max_parity: true,
             http_preconditions: Some(preconditions),
-            scanner_publication_commit_scope,
             user_defined,
             ..Default::default()
         },
@@ -912,27 +886,6 @@ 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 {
@@ -976,32 +929,6 @@ 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]
@@ -1019,32 +946,6 @@ 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,7 +16,6 @@ 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};
@@ -63,8 +62,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,
@@ -73,7 +72,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_and_scope,
+    replace_bucket_usage_memory_from_info, save_config, save_config_shared_with_preconditions_and_lease_fence,
     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,
 };
@@ -455,7 +454,6 @@ 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>,
@@ -463,34 +461,12 @@ 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 {
@@ -555,48 +531,15 @@ async fn sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence_and_s
                 }
                 return Err(EcstoreError::other(SCANNER_PUBLICATION_EPOCH_CHANGED));
             };
-            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(
+            save_config_shared_with_preconditions_and_lease_fence(
                 storeapi.clone(),
                 &backup_path,
                 primary.clone(),
                 sha256hex,
                 revision.preconditions(),
                 scanner_publication_lease_fence,
-                publication_scope.clone(),
             )
-            .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
-            }
+            .await
         };
 
         match save_result {
@@ -1262,6 +1205,10 @@ 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)]
@@ -1546,24 +1493,21 @@ 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() {
-        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) => {
+        match notification_system
+            .acquire_scanner_publication_leases(remote_publication_lease_targets.clone())
+            .await
+        {
+            Ok(grants) => {
                 remote_publication_leases = Some((notification_system, grants));
                 None
             }
-            ScannerPublicationProofWait::Rejected(_) | ScannerPublicationProofWait::Cancelled => {
-                Some(ScannerCycleDeferReason::ActivityBaselineUnavailable)
-            }
+            Err(_) => Some(ScannerCycleDeferReason::ActivityBaselineUnavailable),
         }
     } else {
         Some(ScannerCycleDeferReason::ActivityBaselineUnavailable)
@@ -1604,16 +1548,26 @@ 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);
@@ -1627,11 +1581,6 @@ 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,
@@ -1643,9 +1592,7 @@ 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();
@@ -1710,16 +1657,7 @@ 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 !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() {
+    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,149 +108,6 @@ 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,7 +352,6 @@ 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>>,
@@ -400,9 +399,6 @@ 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;
@@ -3536,47 +3532,6 @@ 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() {
@@ -4628,6 +4583,7 @@ fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() {
     for reason in [
         ScannerCycleDeferReason::DataMovement,
         ScannerCycleDeferReason::ActivityBaselineUnavailable,
+        ScannerCycleDeferReason::PublicationLeaseBudgetExceeded,
         ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded,
         ScannerCycleDeferReason::PublicationLeaseReleaseFailed,
     ] {
@@ -4799,6 +4755,29 @@ 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();
@@ -4828,29 +4807,6 @@ 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();
@@ -5076,140 +5032,6 @@ 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,10 +13,7 @@
 // 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 {
@@ -37,16 +34,6 @@ 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>,
@@ -139,8 +126,6 @@ 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 {
@@ -153,20 +138,8 @@ 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)]
@@ -317,8 +290,6 @@ 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;
@@ -609,54 +580,25 @@ where
 
             let done_save = Metrics::time(Metric::SaveUsage);
             let save_result = {
-                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
+                let Some(_publication_admission) =
+                    scanner_publication_admission_for_epoch(storeapi.clone(), publication_epoch_for_save).await
+                else {
+                    done_save();
+                    break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement);
                 };
                 if remote_lease_expired(remote_lease_deadline) {
                     done_save();
                     break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded);
                 }
-                let save_result = crate::save_config_shared_with_preconditions_and_lease_fence_and_scope(
+                save_config_shared_with_preconditions_and_lease_fence(
                     storeapi.clone(),
                     target_path,
                     data.clone(),
                     sha256hex.clone(),
                     revision.preconditions(),
                     scanner_publication_lease_fence.as_deref(),
-                    publication_scope.clone(),
                 )
-                .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
-                }
+                .await
             };
             done_save();
 
@@ -754,8 +696,6 @@ 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 {
@@ -779,8 +719,6 @@ 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 {
@@ -823,8 +761,6 @@ 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 {
@@ -842,14 +778,12 @@ 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_and_scope(
+            let backup_result = sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence(
                 &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();
@@ -883,8 +817,6 @@ 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;
@@ -953,15 +885,7 @@ async fn cleanup_observed_data_usage_snapshot_for_epoch_and_lease(
         return false;
     }
 
-    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(
+    let result = delete_config_with_publication_admission_for_epoch(
         storeapi,
         RUSTFS_META_BUCKET,
         DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(),
@@ -980,23 +904,9 @@ 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,6 +574,10 @@ 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.
@@ -588,6 +592,7 @@ 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,10 +92,7 @@ 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::notification::scanner_peer_transport_error_message_is_retryable;
-pub(crate) use rustfs_ecstore::api::object::{
-    SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, ScannerPublicationCommitScope, ScannerPublicationCommitState,
-};
+pub(crate) use rustfs_ecstore::api::object::SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY;
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::rebalance::{
     RebalStatus as EcstoreRebalStatus, RebalanceInfo as EcstoreRebalanceInfo, RebalanceMeta as EcstoreRebalanceMeta,
@@ -109,9 +106,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 {
@@ -128,7 +125,7 @@ pub(crate) mod owner {
         EcstoreNsScannerOpenRequest, EcstoreObjectLockConfiguration, EcstoreObjectOpts, EcstoreReplicationConfigurationExt,
         EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore,
         EcstoreVersioningApi, EcstoreVersioningConfiguration, SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY,
-        ScannerPublicationCommitScope, ScannerPublicationCommitState, ScannerReplicationHealObject, ScannerReplicationHealResult,
+        SCANNER_PUBLICATION_LEASE_TTL_MS, 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,6 +57,13 @@ 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
@@ -243,7 +250,7 @@ impl RemoteTargetRequest {
 
     fn into_bucket_target(self) -> S3Result<BucketTarget> {
         self.validate_connection_fields()?;
-        self.into_bucket_target_common(true)
+        self.into_bucket_target_common()
     }
 
     /// Partial-update parse: only the field groups named by `ops` are validated;
@@ -252,18 +259,43 @@ impl RemoteTargetRequest {
         if self.arn.trim().is_empty() {
             return Err(s3_error!(InvalidRequest, "arn is required for update"));
         }
-        let replacing_credentials = ops.contains(&TargetUpdateOp::Credentials);
-        if replacing_credentials {
+        if ops.contains(&TargetUpdateOp::Credentials) {
             self.validate_connection_fields()?;
         }
-        self.into_bucket_target_common(replacing_credentials)
+        self.into_bucket_target_common()
     }
 
-    fn into_bucket_target_common(self, validate_credentials: bool) -> S3Result<BucketTarget> {
+    fn into_bucket_target_common(self) -> 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,
@@ -285,17 +317,9 @@ impl RemoteTargetRequest {
         }
 
         let mut credentials = TargetCredentials::from(self.credentials);
-        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"));
-        }
+        // Past the check above the expiration can only be the zero-value
+        // sentinel, i.e. "no expiration" — never persist it.
+        credentials.expiration = None;
 
         Ok(BucketTarget {
             source_bucket: self.source_bucket,
@@ -2080,13 +2104,27 @@ mod tests {
 
     #[test]
     fn remote_target_request_rejects_unimplemented_fields() {
-        for (field, value) in [
-            ("api", serde_json::json!("s3v2")),
-            ("edge", serde_json::json!(true)),
-            ("edgeSyncBeforeExpiry", serde_json::json!(true)),
+        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),
         ] {
             let mut request = valid_remote_target_request();
-            request[field] = value;
+            if let Some((credential_field, credential_name)) = field.split_once('.') {
+                request[credential_field][credential_name] = value;
+            } else {
+                request[field] = value;
+            }
             let request: RemoteTargetRequest =
                 serde_json::from_value(request).expect("unsupported field should still deserialize");
             let err = request
@@ -2095,86 +2133,15 @@ 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 =
@@ -2225,32 +2192,6 @@ 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
@@ -2297,6 +2238,22 @@ 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
@@ -2557,10 +2514,6 @@ 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);
-        // v4: temporary-credential fields moved from historical-only to writable.
-        assert_eq!(response.replication.remote_targets.contract_version, 4);
+        // v3: temporary-credential fields are explicitly historical-only.
+        assert_eq!(response.replication.remote_targets.contract_version, 3);
         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::Supported),
-                "remote target field {name} must be advertised as writable"
+                    .any(|field| field.name == name && field.state == super::ReplicationFieldState::ReadOnlyHistorical),
+                "remote target field {name} must be advertised as historical-only"
             );
         }
         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"], 4);
+        assert_eq!(value["replication"]["remote_targets"]["contract_version"], 3);
         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"] == "supported"),
-                "serialized remote target field {name} must be writable"
+                    .any(|field| field["name"] == name && field["state"] == "read_only_historical"),
+                "serialized remote target field {name} must be historical-only"
             );
         }
         assert_eq!(value["manual_transition_jobs"]["contract_version"], 1);

rustfs/src/app/object/put.rs

@@ -25,15 +25,14 @@ const ENV_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: &str = "RUSTFS_ZERO_COPY_EAGER_PUT
 
 /// Maximum body size materialized by the ordinary eager PUT path.
 ///
-/// Bodies above this boundary stay streaming so a 1 MiB request does not
-/// reserve a full request-sized buffer while the EC writer is consuming it.
-/// The environment override keeps the boundary reversible for workload A/B
-/// tests and for deployments whose measured workload favors eager ingestion.
+/// Bodies above this boundary stay streaming while the EC writer consumes them.
+/// The environment override keeps the boundary reversible for workload A/B tests
+/// and for deployments whose measured workload favors streaming ingestion.
 const ENV_SMALL_EAGER_PUT_MAX_SIZE_BYTES: &str = "RUSTFS_SMALL_EAGER_PUT_MAX_SIZE_BYTES";
 
 const DEFAULT_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: usize = 16 * 1024 * 1024;
 
-const DEFAULT_SMALL_EAGER_PUT_MAX_SIZE_BYTES: usize = 512 * 1024;
+const DEFAULT_SMALL_EAGER_PUT_MAX_SIZE_BYTES: usize = 1024 * 1024;
 
 /// Keep the eager buffer bounded as concurrent PUTs rise. The thresholds are
 /// deliberately conservative: tiny objects remain eager, while bursty
@@ -2375,22 +2374,25 @@ mod tests {
         assert!(should_use_small_eager_put_path(1024, &headers, false, false, false));
         assert!(should_use_small_eager_put_path(128 * 1024, &headers, false, false, false));
         assert!(should_use_small_eager_put_path(512 * 1024, &headers, false, false, false));
-        assert!(!should_use_small_eager_put_path(512 * 1024 + 1, &headers, false, false, false));
-        assert!(!should_use_small_eager_put_path(1024 * 1024, &headers, false, false, false));
+        assert!(should_use_small_eager_put_path(1024 * 1024, &headers, false, false, false));
+        assert!(!should_use_small_eager_put_path(1024 * 1024 + 1, &headers, false, false, false));
     }
 
     #[test]
     fn select_put_path_switches_at_small_eager_boundary() {
         let headers = HeaderMap::new();
 
-        let (small_path, _, use_zero_copy, use_small_eager) = select_put_path(512 * 1024, &headers, false, false, false);
+        let (small_path, _, use_zero_copy, use_small_eager) = select_put_path(1024 * 1024, &headers, false, false, false);
         assert_eq!(small_path, "small_eager");
         assert!(!use_zero_copy);
         assert!(use_small_eager);
 
-        let (streaming_path, _, use_zero_copy, use_small_eager) = select_put_path(512 * 1024 + 1, &headers, false, false, false);
-        assert_eq!(streaming_path, "streaming");
-        assert!(!use_zero_copy);
+        // The small-eager boundary is crossed first, then the existing
+        // zero-copy precedence applies to objects just above 1 MiB.
+        let (zero_copy_path, _, use_zero_copy, use_small_eager) =
+            select_put_path(1024 * 1024 + 1, &headers, false, false, false);
+        assert_eq!(zero_copy_path, "zero_copy_eager");
+        assert!(use_zero_copy);
         assert!(!use_small_eager);
     }
 
@@ -2472,10 +2474,10 @@ mod tests {
 
     #[test]
     fn dynamic_small_eager_threshold_sheds_memory_only_above_concurrency_limits() {
-        assert_eq!(dynamic_small_eager_put_max_size_bytes(0), 512 * 1024);
-        assert_eq!(dynamic_small_eager_put_max_size_bytes(SMALL_EAGER_CONCURRENCY_SOFT_LIMIT), 512 * 1024);
-        assert_eq!(dynamic_small_eager_put_max_size_bytes(SMALL_EAGER_CONCURRENCY_SOFT_LIMIT + 1), 256 * 1024);
-        assert_eq!(dynamic_small_eager_put_max_size_bytes(SMALL_EAGER_CONCURRENCY_HARD_LIMIT + 1), 128 * 1024);
+        assert_eq!(dynamic_small_eager_put_max_size_bytes(0), 1024 * 1024);
+        assert_eq!(dynamic_small_eager_put_max_size_bytes(SMALL_EAGER_CONCURRENCY_SOFT_LIMIT), 1024 * 1024);
+        assert_eq!(dynamic_small_eager_put_max_size_bytes(SMALL_EAGER_CONCURRENCY_SOFT_LIMIT + 1), 512 * 1024);
+        assert_eq!(dynamic_small_eager_put_max_size_bytes(SMALL_EAGER_CONCURRENCY_HARD_LIMIT + 1), 256 * 1024);
     }
 
     #[test]
@@ -2486,7 +2488,7 @@ mod tests {
         assert_eq!(path, "small_eager");
         assert!(small_eager);
 
-        let (path, _, _, small_eager) = select_put_path_with_concurrency(256 * 1024, &headers, false, false, false, 256);
+        let (path, _, _, small_eager) = select_put_path_with_concurrency(256 * 1024 + 1, &headers, false, false, false, 256);
         assert_eq!(path, "streaming");
         assert!(!small_eager);
     }

rustfs/src/storage/rpc/node_service.rs

@@ -123,21 +123,6 @@ 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!(
@@ -2349,7 +2334,11 @@ impl Node for NodeService {
                 success: true,
                 error_info: None,
             })),
-            Err(err) => Ok(Response::new(start_decommission_failure_response(err))),
+            Err(err) => Ok(Response::new(StartDecommissionResponse {
+                error_code: None,
+                success: false,
+                error_info: Some(err.to_string()),
+            })),
         }
     }
 
@@ -2467,7 +2456,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, start_decommission_failure_response, stop_rebalance_response,
+        scanner_activity_response_v7, stop_rebalance_response,
     };
     use crate::storage::rpc::node_service::heal::heal_topology_fingerprint;
     use crate::storage::storage_api::rpc_consumer::node_service::{DiskError, HealBucketInfo};
@@ -2490,14 +2479,14 @@ mod tests {
     use rustfs_protos::models::PingBodyBuilder;
     use rustfs_protos::proto_gen::node_service::{
         BackgroundHealStatusRequest, BatchGenerallyLockRequest, CancelDecommissionRequest, CheckPartsRequest,
-        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,
+        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,
         LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, MakeBucketRequest, MakeVolumeRequest,
         MakeVolumesRequest, Mss, PingRequest, PreparePartTransactionRequest, ReadAllRequest, ReadAtRequest, ReadMultipleRequest,
         ReadVersionRequest, ReadXlRequest, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, RenameDataRequest,
@@ -2587,20 +2576,6 @@ 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,7 +33,6 @@ 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;
@@ -1231,40 +1230,37 @@ 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: Option<Arc<dyn Send + Sync>> =
-                if let Some(token) = scanner_publication_lease_token {
-                    let Some(store) = self.resolve_object_store() else {
+            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) => {
                         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()),
+                            error: Some(DiskError::other(err.to_string()).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_borrowed_with_fence_and_guard(
+                .rename_data(
                     &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
             {
@@ -1645,36 +1641,26 @@ 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: Option<Arc<dyn Send + Sync>> =
-                if let Some(token) = scanner_publication_lease_token {
-                    let Some(store) = self.resolve_object_store() else {
+            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) => {
                         return Ok(Response::new(DeleteResponse {
                             success: false,
-                            error: Some(DiskError::other("scanner publication lease owner is unavailable").into()),
+                            error: Some(DiskError::other(err.to_string()).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_with_scanner_publication_lease_and_guard(
-                    &request.volume,
-                    &request.path,
-                    options,
-                    scanner_publication_lease_token,
-                    scanner_publication_lease_guard,
-                )
-                .await
-            {
+                }
+            } else {
+                None
+            };
+            match disk.delete(&request.volume, &request.path, options).await {
                 Ok(_) => Ok(Response::new(DeleteResponse {
                     success: true,
                     error: None,

rustfs/src/storage/storage_api.rs

@@ -1301,6 +1301,14 @@ 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>;
@@ -1444,6 +1452,17 @@ 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
     }