fix(odm): bound empty pagination chains with staged tokens (#7197)

* fix(odm): add staged cross-request pagination progress budgets

* fix(ecstore): remove duplicate local rename implementation

Keep the canonical commit module after concurrent storage changes merged.
The control-write and rollback changes are already present there.

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

* fix(app): simplify absent SSE configuration matching

* fix(tests): satisfy new clippy lints

---------

Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

.config/make/tests.mak

@@ -40,6 +40,7 @@ script-tests: ## Run shell script tests
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_test_wiring.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_security_coverage.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_scheduled_validation_freshness.py --self-test
+	$(RUSTFS_PYTHON_BIN) ./scripts/test_security_workflow.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_report_compat.py
 	bash -n ./scripts/validate_object_data_cache_cold_stampede.sh
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_object_data_cache_follower_samples.py --self-test

.github/pull_request_template.md

@@ -10,16 +10,16 @@ Use N/A when there is no related issue.
 
 ## Summary of Changes
 <!--
-Briefly explain what changed and why reviewers should accept it.
-Focus on behavior, compatibility, and review-relevant context.
+Describe the concrete problem and resulting behavior. For a behavior change, name the input or state that triggers it and the expected outcome. Explain any new dependency or abstraction that the change needs.
 -->
 
 ## Verification
 <!--
-List the commands or checks you ran, for example:
-- `make pre-commit`
+Give 1–3 concrete pieces of evidence for the changed behavior: the test or command, its observed result, and the regression it catches. For a bug fix, record a failing-before/passing-after check or explain why it was unavailable.
 
-Use N/A only when verification is not applicable.
+Identify the tested commit and any local changes. When testing a prebuilt binary or external service, include its source/version and artifact identity; a successful run against a different build is not evidence for this change.
+
+List relevant checks not run and the remaining risk. Use the validation tier in AGENTS.md; do not run broader checks solely to fill this section. For documentation-only changes, list the applicable documentation checks. Use N/A only when verification is not applicable.
 -->
 
 ## Impact

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

@@ -129,6 +129,7 @@ jobs:
         run: |
           python3 ./scripts/check_test_wiring.py --self-test
           python3 ./scripts/check_scheduled_validation_freshness.py --self-test
+          python3 ./scripts/test_security_workflow.py
           python3 ./scripts/check_test_wiring.py
 
       - name: Check no planning docs committed

.github/workflows/ci.yml

@@ -167,6 +167,7 @@ jobs:
         run: |
           python3 ./scripts/check_test_wiring.py --self-test
           python3 ./scripts/check_scheduled_validation_freshness.py --self-test
+          python3 ./scripts/test_security_workflow.py
           python3 ./scripts/check_test_wiring.py
 
       - name: Check no planning docs committed

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

@@ -74,10 +74,23 @@ env:
 jobs:
   security-test:
     runs-on: smoke-testing
-    continue-on-error: true
     timeout-minutes: 360
     if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
     steps:
+      - name: Checkout repository (for the OIDC live gate script)
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+
+      - name: Initialize security evidence
+        id: evidence
+        run: |
+          set -euo pipefail
+          umask 077
+          SECURITY_ARTIFACTS_DIR="${RUNNER_TEMP}/rustfs-security-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
+          mkdir -- "${SECURITY_ARTIFACTS_DIR}"
+          printf 'SECURITY_ARTIFACTS_DIR=%s\n' "${SECURITY_ARTIFACTS_DIR}" >> "${GITHUB_ENV}"
+
       # auto-testing is private: clone it with the dedicated PF token (not
       # GITHUB_TOKEN) and retry transient GitHub/network failures.
       - name: Checkout auto-testing scripts (with retry)
@@ -98,11 +111,6 @@ jobs:
           echo "ERROR: unable to clone rustfs/auto-testing after 5 attempts" >&2
           exit 1
 
-      - name: Checkout repository (for the OIDC live gate script)
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
-        with:
-          persist-credentials: false
-
       - name: Show environment
         run: |
           uname -a
@@ -135,7 +143,8 @@ jobs:
         id: test
         continue-on-error: true
         env:
-          REPORT_FILE: /tmp/rustfs-security-report.md
+          REPORT_FILE: ${{ env.SECURITY_ARTIFACTS_DIR }}/suite-report.md
+          TMPDIR: ${{ env.SECURITY_ARTIFACTS_DIR }}
           RUSTFS_SECURITY_OIDC_LIVE_SCRIPT: ${{ github.workspace }}/scripts/test/oidc_keycloak_live.sh
         run: |
           set -euo pipefail
@@ -159,29 +168,48 @@ jobs:
           else
             ARGS+=(--package-url "${RUSTFS_NIGHTLY_PACKAGE_URL}")
           fi
-          ./auto-testing/rustfs-security-test.sh "${ARGS[@]}"
+          GITHUB_STEP_SUMMARY=/dev/null ./auto-testing/rustfs-security-test.sh "${ARGS[@]}"
 
       - name: Generate report
-        if: always()
+        id: report
+        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        env:
+          TEST_OUTCOME: ${{ steps.test.outcome }}
         run: |
           set -euo pipefail
-          if [ ! -f /tmp/rustfs-security-report.md ]; then
-            {
-              echo "# RustFS security test report"
-              echo ""
-              echo "- Run: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
-              echo "- Trigger: ${{ github.event_name }}"
-              echo "- Test Step Outcome: failure (suite did not produce a report)"
-            } > /tmp/rustfs-security-report.md
+          RESULT=failure
+          if [ "${TEST_OUTCOME}" = "success" ] && [ -s "${SECURITY_ARTIFACTS_DIR}/suite-report.md" ]; then
+            RESULT=success
           fi
-          cat /tmp/rustfs-security-report.md >> "${GITHUB_STEP_SUMMARY}"
+          {
+            echo "# RustFS security test report"
+            echo ""
+            echo "- Run: ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}"
+            echo "- Attempt: ${GITHUB_RUN_ATTEMPT}"
+            echo "- Workflow Commit: ${GITHUB_SHA}"
+            echo "- Trigger: ${GITHUB_EVENT_NAME}"
+            echo "- Test Step Outcome: ${RESULT}"
+            echo "- Suite Step Outcome: ${TEST_OUTCOME}"
+            echo ""
+            # The dashboard prioritizes case rows over the step outcome.
+            # Keep partial case results in the artifact when the suite fails.
+            if [ "${RESULT}" = "success" ]; then
+              cat "${SECURITY_ARTIFACTS_DIR}/suite-report.md"
+            elif [ -s "${SECURITY_ARTIFACTS_DIR}/suite-report.md" ]; then
+              echo "The suite did not complete successfully. See suite-report.md in this run's artifact for diagnostics."
+            else
+              echo "The suite did not produce a non-empty report."
+            fi
+          } > "${SECURITY_ARTIFACTS_DIR}/report.md"
+          cat "${SECURITY_ARTIFACTS_DIR}/report.md" >> "${GITHUB_STEP_SUMMARY}"
+          [ "${RESULT}" = "success" ]
 
       - name: Upload functional report to dashboard
-        if: always()
+        if: ${{ always() && steps.evidence.outcome == 'success' }}
         continue-on-error: true
         env:
           GH_TOKEN: ${{ env.PF_TESTING_GH_TOKEN }}
-          REPORT_FILE: /tmp/rustfs-security-report.md
+          REPORT_FILE: ${{ env.SECURITY_ARTIFACTS_DIR }}/report.md
           SUITE: security
         run: |
           set -euo pipefail
@@ -210,8 +238,9 @@ jobs:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
           SUITE: 'security'
           SUITE_LABEL: 'Security'
+          EVIDENCE_OUTCOME: ${{ steps.evidence.outcome }}
           RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
-          REPORT_FILE: '/tmp/rustfs-security-report.md'
+          REPORT_FILE: ${{ env.SECURITY_ARTIFACTS_DIR }}/report.md
           LOG_FILE: ''
         run: |
           set -euo pipefail
@@ -245,7 +274,7 @@ jobs:
             echo ""
             echo "## Report (errors and symptoms)"
             echo ""
-            if [ -s "${REPORT_FILE}" ]; then
+            if [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${REPORT_FILE}" ]; then
               redact < "${REPORT_FILE}"
             elif [ -s "${LOG_FILE:-}" ]; then
               echo "(report file missing; log tail below)"
@@ -263,14 +292,12 @@ jobs:
           echo "filed backlog issue for suite ${SUITE}"
 
       - name: Upload report and logs
-        if: always()
+        if: ${{ always() && steps.evidence.outcome == 'success' }}
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
         with:
-          name: rustfs-security-test-${{ github.run_id }}
-          path: |
-            /tmp/rustfs-security-report.md
-            /tmp/rustfs-security.*/*
-          if-no-files-found: ignore
+          name: rustfs-security-test-${{ github.run_id }}-${{ github.run_attempt }}
+          path: ${{ env.SECURITY_ARTIFACTS_DIR }}/
+          if-no-files-found: error
           retention-days: 3
 
       - name: Cleanup environment (after)

.github/workflows/scheduled-validation-freshness.yml

@@ -42,6 +42,7 @@ jobs:
       - name: Check latest scheduled runs
         env:
           GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          RUSTFS_DEFAULT_BRANCH: ${{ github.event.repository.default_branch }}
         run: |
           set +e
           python3 scripts/check_scheduled_validation_freshness.py \

crates/ecstore/src/api/mod.rs

@@ -167,9 +167,10 @@ pub mod bucket {
             idle_guarded_body,
         };
         pub use crate::bucket::on_demand_migration::{
-            FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken,
-            ListThroughTokenError, MAX_LIST_FETCHES_PER_SIDE, MergeOutcome, MergePick, MergeSide, SOURCE_LIST_MAX_RATE_WAIT,
-            SOURCE_LIST_RATE_PER_SEC, SourceListPlan, SourceListRateLimiter, decode_continuation_token, source_list_plan,
+            FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger,
+            ListThroughToken, ListThroughTokenError, MAX_LIST_FETCHES_PER_SIDE, MAX_LIST_NO_PROGRESS_PAGES, MergeOutcome,
+            MergePick, MergeSide, SOURCE_LIST_MAX_RATE_WAIT, SOURCE_LIST_RATE_PER_SEC, SourceListPlan, SourceListRateLimiter,
+            decode_continuation_token, source_list_plan,
         };
         pub mod backfill {
             pub use crate::bucket::on_demand_migration::backfill::{
@@ -196,15 +197,16 @@ pub mod bucket {
         pub use crate::bucket::metadata_sys::ConfigWriteLockProbe;
         pub use crate::bucket::metadata_sys::{
             BucketMetadataMutationGuard, BucketMetadataSys, ObjectLockConfigState, acquire_bucket_metadata_transaction_lock,
-            acquire_bucket_metadata_transaction_lock_for_incarnation, capture_bucket_metadata_incarnation, delete,
-            delete_if_incarnation, delete_under_transaction_lock, get, get_accelerate_config, get_bucket_policy,
-            get_bucket_policy_raw, get_bucket_targets_config, get_config_from_disk, get_cors_config, get_durability_config,
-            get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config, get_notification_config,
-            get_object_lock_config, get_object_lock_config_state, get_on_demand_migration_config, get_public_access_block_config,
-            get_quota_config, get_replication_config, get_request_payment_config, get_sse_config, get_tagging_config,
-            get_versioning_config, get_website_config, init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata,
-            remove_bucket_metadata, set_bucket_metadata, update, update_bucket_targets_under_transaction_lock,
-            update_config_with, update_if_incarnation, update_quota_if_incarnation, update_under_transaction_lock,
+            acquire_bucket_metadata_transaction_lock_for_incarnation, acquire_scanner_bucket_incarnation_fence,
+            capture_bucket_metadata_incarnation, delete, delete_if_incarnation, delete_under_transaction_lock, get,
+            get_accelerate_config, get_bucket_policy, get_bucket_policy_raw, get_bucket_targets_config, get_config_from_disk,
+            get_cors_config, get_durability_config, get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config,
+            get_notification_config, get_object_lock_config, get_object_lock_config_state, get_on_demand_migration_config,
+            get_public_access_block_config, get_quota_config, get_replication_config, get_request_payment_config, get_sse_config,
+            get_tagging_config, get_versioning_config, get_website_config, init_bucket_metadata_sys, list_bucket_targets,
+            reload_bucket_metadata, remove_bucket_metadata, set_bucket_metadata, update,
+            update_bucket_targets_under_transaction_lock, update_config_with, update_if_incarnation, update_quota_if_incarnation,
+            update_under_transaction_lock,
         };
     }
 

crates/ecstore/src/bucket/metadata_sys.rs

@@ -655,6 +655,12 @@ pub struct BucketMetadataMutationGuard {
 }
 
 impl BucketMetadataMutationGuard {
+    /// Returns the storage-verified identity while both incarnation fences remain valid.
+    pub fn checked_bucket_incarnation(&self) -> Result<(&str, Uuid)> {
+        self.ensure_valid(&self.bucket)?;
+        Ok((&self.bucket, self.incarnation_id))
+    }
+
     fn ensure_valid(&self, bucket: &str) -> Result<()> {
         if self.bucket != bucket {
             return Err(Error::other("bucket metadata mutation guard does not match bucket"));
@@ -674,6 +680,29 @@ async fn acquire_config_write_guard_for_incarnation(
     sys: Arc<RwLock<BucketMetadataSys>>,
     bucket: &str,
     expected_incarnation_id: Option<Uuid>,
+) -> Result<BucketMetadataMutationGuard> {
+    acquire_config_write_guard_with_migration(sys, bucket, expected_incarnation_id, true).await
+}
+
+/// Scanner probes must not create an incarnation to make a capability available.
+pub async fn acquire_scanner_bucket_incarnation_fence(
+    bucket: &str,
+    expected_incarnation_id: Uuid,
+    expected_owner_id: Uuid,
+) -> Result<BucketMetadataMutationGuard> {
+    super::utils::check_valid_bucket_name(bucket)?;
+    let sys = get_bucket_metadata_sys()?;
+    if expected_owner_id.is_nil() || sys.read().await.api.id != expected_owner_id || expected_incarnation_id.is_nil() {
+        return Err(Error::other("scanner bucket incarnation owner does not match"));
+    }
+    acquire_config_write_guard_with_migration(sys, bucket, Some(expected_incarnation_id), false).await
+}
+
+async fn acquire_config_write_guard_with_migration(
+    sys: Arc<RwLock<BucketMetadataSys>>,
+    bucket: &str,
+    expected_incarnation_id: Option<Uuid>,
+    migrate: bool,
 ) -> Result<BucketMetadataMutationGuard> {
     let metadata_sys = sys.read().await.clone();
     let lifecycle_guard = metadata_sys.api.acquire_bucket_lifecycle_read_lock(bucket).await?;
@@ -681,13 +710,15 @@ async fn acquire_config_write_guard_for_incarnation(
     // Legacy buckets are migrated while the lifecycle fence prevents a
     // same-name replacement. The second read under the write transaction is
     // the CAS source of truth for the actual rewrite.
-    await_bucket_namespace_operation(
-        Some(&lifecycle_guard),
-        bucket,
-        "bucket config incarnation migration",
-        metadata_sys.get_bucket_incarnation_id(bucket),
-    )
-    .await?;
+    if migrate {
+        await_bucket_namespace_operation(
+            Some(&lifecycle_guard),
+            bucket,
+            "bucket config incarnation migration",
+            metadata_sys.get_bucket_incarnation_id(bucket),
+        )
+        .await?;
+    }
     let transaction_guard = await_bucket_namespace_operation(
         Some(&lifecycle_guard),
         bucket,
@@ -3176,6 +3207,82 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn scoped_dirty_usage_incarnation_probe_does_not_migrate_legacy_metadata() {
+        let (dirs, store) = isolated_store_over_temp_disks().await;
+        let sys = Arc::new(RwLock::new(BucketMetadataSys::new(store.clone())));
+        let bucket = "scoped-ack-legacy";
+        for dir in &dirs {
+            std::fs::create_dir_all(dir.path().join(bucket)).expect("create legacy bucket");
+        }
+        let mut metadata = BucketMetadata::new(bucket);
+        metadata.bucket_incarnation_id = Uuid::nil();
+        sys.read()
+            .await
+            .persist_and_set(metadata)
+            .await
+            .expect("persist legacy metadata");
+        assert!(
+            acquire_config_write_guard_with_migration(sys.clone(), bucket, Some(Uuid::new_v4()), false)
+                .await
+                .is_err()
+        );
+        assert!(load_bucket_incarnation(store, bucket).await.expect("read sidecar").is_none());
+        assert!(
+            sys.read()
+                .await
+                .get_config_from_disk(bucket)
+                .await
+                .expect("read metadata")
+                .bucket_incarnation_id
+                .is_nil()
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[serial]
+    async fn scoped_dirty_usage_incarnation_rejects_deleted_and_recreated_bucket() {
+        let (_dirs, store) = isolated_store_over_temp_disks().await;
+        init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let sys = bucket_metadata_sys_of(&store.ctx).expect("metadata owner");
+        let bucket = "scoped-ack-recreated";
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+        let old = store.bucket_incarnation_id_from_disk(bucket).await.expect("old incarnation");
+        let guard = acquire_config_write_guard_with_migration(sys.clone(), bucket, Some(old), false)
+            .await
+            .expect("trusted incarnation fence");
+        assert_eq!(guard.checked_bucket_incarnation().expect("valid fences"), (bucket, old));
+        drop(guard);
+        store
+            .delete_bucket(bucket, &DeleteBucketOptions::default())
+            .await
+            .expect("delete bucket");
+        assert!(
+            acquire_config_write_guard_with_migration(sys.clone(), bucket, Some(old), false)
+                .await
+                .is_err()
+        );
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("recreate bucket");
+        let new = store.bucket_incarnation_id_from_disk(bucket).await.expect("new incarnation");
+        assert_ne!(old, new);
+        assert!(
+            acquire_config_write_guard_with_migration(sys.clone(), bucket, Some(old), false)
+                .await
+                .is_err()
+        );
+        assert!(
+            acquire_config_write_guard_with_migration(sys, bucket, Some(new), false)
+                .await
+                .is_ok()
+        );
+    }
+
     #[tokio::test]
     async fn old_node_metadata_rewrite_cannot_replace_bucket_incarnation_sidecar() {
         let (dirs, ecstore) = isolated_store_over_temp_disks().await;

crates/ecstore/src/bucket/on_demand_migration/list_through.rs

@@ -25,8 +25,13 @@ use parking_lot::Mutex;
 use serde::{Deserialize, Serialize};
 use std::time::{Duration, Instant};
 
-/// The only continuation-token envelope version this build reads and writes.
+/// The continuation-token version used by ordinary progressing pages.
 pub const LIST_THROUGH_TOKEN_VERSION: u32 = 1;
+const LIST_THROUGH_PROGRESS_TOKEN_VERSION: u32 = 2;
+
+/// The sixteenth consecutive merged page without a key or new EOF fails.
+/// This also bounds legitimate sparse listings; it is not a cycle detector.
+pub const MAX_LIST_NO_PROGRESS_PAGES: u8 = 16;
 
 /// Envelope marker. A bucket that is *not* merging hands out the local
 /// listing's own marker, so the decoder needs a positive signal before it
@@ -111,6 +116,10 @@ pub struct ListThroughToken {
     /// common prefix compares as itself, never as its members.
     #[serde(default)]
     pub last_key: Option<String>,
+    /// Consecutive empty truncated merged pages, present only in v2 tokens.
+    /// Ordinary v1 tokens retain their original serialized shape.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub no_progress: Option<u8>,
 }
 
 impl ListThroughToken {
@@ -123,6 +132,7 @@ impl ListThroughToken {
             source: source.token,
             source_done: source.done,
             last_key,
+            no_progress: None,
         }
     }
 
@@ -170,7 +180,21 @@ pub fn decode_continuation_token(decoded: &str) -> Result<ListThroughCursor, Lis
         return Ok(ListThroughCursor::Local(decoded.to_string()));
     }
     match value.get("v").and_then(serde_json::Value::as_u64) {
-        Some(version) if version == u64::from(LIST_THROUGH_TOKEN_VERSION) => {}
+        Some(version) if version == u64::from(LIST_THROUGH_TOKEN_VERSION) => {
+            // v1 readers reject this field even when it is null or zero.
+            if value.get("no_progress").is_some() {
+                return Err(ListThroughTokenError::Malformed);
+            }
+        }
+        Some(version) if version == u64::from(LIST_THROUGH_PROGRESS_TOKEN_VERSION) => {
+            if !value
+                .get("no_progress")
+                .and_then(serde_json::Value::as_u64)
+                .is_some_and(|count| (1..u64::from(MAX_LIST_NO_PROGRESS_PAGES)).contains(&count))
+            {
+                return Err(ListThroughTokenError::Malformed);
+            }
+        }
         Some(version) => return Err(ListThroughTokenError::UnsupportedVersion(version.min(u64::from(u32::MAX)) as u32)),
         None => return Err(ListThroughTokenError::Malformed),
     }
@@ -288,6 +312,8 @@ pub enum ListPageError {
     Empty,
     #[error("truncated listing repeats a continuation token")]
     Repeated,
+    #[error("listing exhausted its consecutive no-progress page budget")]
+    NoProgress(MergeSide),
 }
 
 pub(crate) fn validate_list_page(is_truncated: bool, token: Option<&str>, next_token: Option<&str>) -> Result<(), ListPageError> {
@@ -352,6 +378,7 @@ pub struct MergeOutcome {
 #[derive(Debug)]
 pub struct ListThroughMerger {
     max_keys: usize,
+    no_progress: Option<u8>,
     last_key: Option<String>,
     local: SideState,
     source: SideState,
@@ -371,6 +398,7 @@ impl ListThroughMerger {
         };
         Self {
             max_keys,
+            no_progress: token.and_then(|token| token.no_progress),
             last_key,
             local,
             source,
@@ -436,13 +464,18 @@ impl ListThroughMerger {
         Ok(())
     }
 
-    pub fn finish(self) -> MergeOutcome {
+    /// `issue_progress_tokens` allows a v1 chain to start carrying a budget.
+    /// An existing v2 budget is always enforced, including on reader-only nodes.
+    /// Borrowing lets a source failure re-merge the fetched local buffers.
+    pub fn finish(&self, issue_progress_tokens: bool) -> Result<MergeOutcome, ListPageError> {
         let Self {
             max_keys,
+            no_progress,
             last_key,
             local,
             source,
         } = self;
+        let max_keys = *max_keys;
 
         // A side with more pages behind it can only be trusted up to the last
         // key it handed over: past that horizon the other side's entries could
@@ -508,12 +541,44 @@ impl ListThroughMerger {
         let source_left = !source.disabled && (!source_cursor.done || consumed_source < source.entries.len());
         let is_truncated = local_left || source_left;
 
-        let last_key = consumed_key.or(last_key);
-        MergeOutcome {
+        let reached_eof = (!local.start.done && local_cursor.done) || (!source.start.done && source_cursor.done);
+        let next_no_progress = if !is_truncated || !picks.is_empty() || reached_eof {
+            None
+        } else if max_keys == 0 {
+            // A zero-sized request cannot consume entries. Preserve an existing
+            // budget without spending it or starting a new one.
+            *no_progress
+        } else if issue_progress_tokens || no_progress.is_some() {
+            let count = no_progress.unwrap_or(0).saturating_add(1);
+            if count >= MAX_LIST_NO_PROGRESS_PAGES {
+                // An empty truncated side closes the merge horizon. Local
+                // failure takes precedence; disabling the source cannot fix it.
+                let side = if local.more && local.entries.is_empty() {
+                    MergeSide::Local
+                } else if !source.disabled && source.more && source.entries.is_empty() {
+                    MergeSide::Source
+                } else {
+                    MergeSide::Local
+                };
+                return Err(ListPageError::NoProgress(side));
+            }
+            Some(count)
+        } else {
+            None
+        };
+        let last_key = consumed_key.or_else(|| last_key.clone());
+        Ok(MergeOutcome {
             picks,
             is_truncated,
-            next_token: is_truncated.then(|| ListThroughToken::new(local_cursor, source_cursor, last_key)),
-        }
+            next_token: is_truncated.then(|| {
+                let mut token = ListThroughToken::new(local_cursor, source_cursor, last_key);
+                if let Some(count) = next_no_progress {
+                    token.v = LIST_THROUGH_PROGRESS_TOKEN_VERSION;
+                    token.no_progress = Some(count);
+                }
+                token
+            }),
+        })
     }
 }
 
@@ -641,7 +706,7 @@ mod tests {
                     .push_page(fetch.side, kept, truncated, next)
                     .expect("reference provider pages must advance");
             }
-            let outcome = merger.finish();
+            let outcome = merger.finish(false).expect("valid merge outcome");
             assert_eq!(outcome.is_truncated, outcome.next_token.is_some());
             if outcome.is_truncated {
                 assert_ne!(outcome.next_token, token, "every truncated merged page must make progress");
@@ -724,7 +789,7 @@ mod tests {
             .push_page(MergeSide::Local, vec![ListEntryKey::object("a")], false, None)
             .expect("local EOF is valid");
         assert_eq!(merger.next_fetch(), None);
-        let outcome = merger.finish();
+        let outcome = merger.finish(false).expect("valid merge outcome");
         assert_eq!(outcome.picks.len(), 1);
         assert!(!outcome.is_truncated);
         assert!(outcome.next_token.is_none());
@@ -740,6 +805,7 @@ mod tests {
             source: Some("source-1".to_string()),
             source_done: false,
             last_key: Some("a".to_string()),
+            no_progress: None,
         };
         let mut merger = ListThroughMerger::new(1, Some(&resume));
         merger.disable_source();
@@ -751,7 +817,7 @@ mod tests {
                 Some("local-2".to_string()),
             )
             .expect("local cursor advances");
-        let outcome = merger.finish();
+        let outcome = merger.finish(false).expect("valid merge outcome");
         assert!(outcome.is_truncated);
         let token = outcome.next_token.expect("truncated page carries a token");
         assert_eq!(token.source.as_deref(), Some("source-1"), "the source cursor must not move");
@@ -830,7 +896,7 @@ mod tests {
                 .expect("opaque cursor advances regardless of sort order");
         }
         assert!(merger.next_fetch().is_none(), "two source fetches exhaust the request budget");
-        let outcome = merger.finish();
+        let outcome = merger.finish(false).expect("valid merge outcome");
         assert!(outcome.picks.is_empty());
         assert!(outcome.is_truncated);
         let token = outcome.next_token.expect("empty progressing page has a cursor");
@@ -840,7 +906,7 @@ mod tests {
         merger
             .push_page(MergeSide::Source, vec![ListEntryKey::object("result")], false, None)
             .expect("source EOF");
-        let outcome = merger.finish();
+        let outcome = merger.finish(false).expect("valid merge outcome");
         assert_eq!(
             outcome.picks,
             vec![MergePick {
@@ -887,7 +953,7 @@ mod tests {
             Err(ListPageError::Repeated)
         );
         merger.disable_source();
-        let outcome = merger.finish();
+        let outcome = merger.finish(false).expect("valid merge outcome");
         assert_eq!(
             outcome.picks,
             vec![MergePick {
@@ -979,8 +1045,8 @@ mod tests {
         let encoded = token.encode();
         assert_eq!(decode_continuation_token(&encoded), Ok(ListThroughCursor::Merged(Box::new(token))));
 
-        let bumped = encoded.replace("\"v\":1", "\"v\":2");
-        assert_eq!(decode_continuation_token(&bumped), Err(ListThroughTokenError::UnsupportedVersion(2)));
+        let bumped = encoded.replace("\"v\":1", "\"v\":3");
+        assert_eq!(decode_continuation_token(&bumped), Err(ListThroughTokenError::UnsupportedVersion(3)));
 
         let extra = encoded.replace("{", "{\"x\":1,");
         assert_eq!(decode_continuation_token(&extra), Err(ListThroughTokenError::Malformed));
@@ -992,6 +1058,257 @@ mod tests {
         assert_eq!(decode_continuation_token(no_version), Err(ListThroughTokenError::Malformed));
     }
 
+    fn progress_token(count: Option<u8>, local_done: bool, source_done: bool) -> ListThroughToken {
+        let mut token = ListThroughToken::new(
+            SideCursor {
+                token: None,
+                done: local_done,
+            },
+            SideCursor {
+                token: Some("A".into()),
+                done: source_done,
+            },
+            Some("last-key".into()),
+        );
+        if let Some(count) = count {
+            token.v = LIST_THROUGH_PROGRESS_TOKEN_VERSION;
+            token.no_progress = Some(count);
+        }
+        token
+    }
+
+    fn push_empty_pages(merger: &mut ListThroughMerger, side: MergeSide) {
+        for _ in 0..MAX_LIST_FETCHES_PER_SIDE {
+            let fetch = merger.next_fetch().expect("empty truncated side must be fetched");
+            assert_eq!(fetch.side, side);
+            let next = format!("{}:next", fetch.token.unwrap_or_default());
+            merger
+                .push_page(side, vec![], true, Some(next))
+                .expect("opaque cursor advances");
+        }
+    }
+
+    #[test]
+    fn progress_tokens_preserve_v1_bytes_and_validate_v2_counts() {
+        let token = progress_token(None, true, false);
+        assert_eq!(
+            token.encode(),
+            r#"{"t":"odm-list","v":1,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key"}"#
+        );
+        for count in 1..MAX_LIST_NO_PROGRESS_PAGES {
+            let token = progress_token(Some(count), true, false);
+            assert_eq!(decode_continuation_token(&token.encode()), Ok(ListThroughCursor::Merged(Box::new(token))));
+        }
+        for version in [1, 2] {
+            for value in ["null", "0", "16", "-1", "1.5", "256", "18446744073709551616", "\"1\""] {
+                let encoded = format!(r#"{{"t":"odm-list","v":{version},"no_progress":{value}}}"#);
+                assert_eq!(decode_continuation_token(&encoded), Err(ListThroughTokenError::Malformed), "{encoded}");
+            }
+        }
+        for encoded in [
+            r#"{"t":"odm-list","v":1,"no_progress":1}"#,
+            r#"{"t":"odm-list","v":2}"#,
+            r#"{"t":"odm-list","v":2,"no_progress":1,"extra":true}"#,
+        ] {
+            assert_eq!(decode_continuation_token(encoded), Err(ListThroughTokenError::Malformed), "{encoded}");
+        }
+    }
+
+    #[test]
+    fn reader_only_nodes_do_not_start_a_budget_but_mixed_readers_preserve_one() {
+        let mut token = progress_token(None, true, false);
+        for _ in 0..MAX_LIST_NO_PROGRESS_PAGES {
+            let mut merger = ListThroughMerger::new(2, Some(&token));
+            push_empty_pages(&mut merger, MergeSide::Source);
+            token = merger
+                .finish(false)
+                .expect("reader-only v1 behavior")
+                .next_token
+                .expect("truncated cursor");
+            assert_eq!(token.v, 1);
+            assert_eq!(token.no_progress, None);
+        }
+        for count in 1..=MAX_LIST_NO_PROGRESS_PAGES {
+            let mut merger = ListThroughMerger::new(2, Some(&token));
+            push_empty_pages(&mut merger, MergeSide::Source);
+            assert!(merger.next_fetch().is_none(), "the per-request two-fetch limit stays intact");
+            let outcome = merger.finish(count % 2 == 1);
+            if count == MAX_LIST_NO_PROGRESS_PAGES {
+                assert_eq!(outcome, Err(ListPageError::NoProgress(MergeSide::Source)));
+                break;
+            }
+            token = outcome.expect("budget not exhausted").next_token.expect("truncated cursor");
+            assert_eq!(token.no_progress, Some(count));
+            let ListThroughCursor::Merged(decoded) = decode_continuation_token(&token.encode()).expect("round-trip v2") else {
+                panic!("merged cursor expected");
+            };
+            token = *decoded;
+        }
+    }
+
+    #[test]
+    fn objects_and_common_prefixes_reset_a_budget_at_the_boundary() {
+        for entry in [ListEntryKey::object("result"), ListEntryKey::prefix("result/")] {
+            for issue_tokens in [false, true] {
+                let resume = progress_token(Some(MAX_LIST_NO_PROGRESS_PAGES - 1), true, false);
+                let mut merger = ListThroughMerger::new(2, Some(&resume));
+                merger
+                    .push_page(MergeSide::Source, vec![], true, Some("B".into()))
+                    .expect("empty advancing page");
+                merger
+                    .push_page(MergeSide::Source, vec![entry.clone()], true, Some("C".into()))
+                    .expect("real progress");
+                let outcome = merger
+                    .finish(issue_tokens)
+                    .expect("real progress does not exhaust the budget");
+                assert_eq!(
+                    outcome.picks,
+                    vec![MergePick {
+                        side: MergeSide::Source,
+                        index: 0
+                    }]
+                );
+                let next = outcome.next_token.expect("source remains truncated");
+                assert_eq!(next.last_key.as_deref(), Some(entry.name.as_str()));
+                assert_eq!(next.v, 1);
+                assert_eq!(next.no_progress, None);
+                assert!(!next.encode().contains("no_progress"));
+            }
+        }
+    }
+
+    #[test]
+    fn only_a_new_eof_transition_resets_the_empty_page_budget() {
+        for finished_side in [MergeSide::Local, MergeSide::Source] {
+            let resume = progress_token(Some(MAX_LIST_NO_PROGRESS_PAGES - 1), false, false);
+            let mut merger = ListThroughMerger::new(2, Some(&resume));
+            if finished_side == MergeSide::Local {
+                merger
+                    .push_page(MergeSide::Local, vec![], false, None)
+                    .expect("new local EOF");
+                push_empty_pages(&mut merger, MergeSide::Source);
+            } else {
+                push_empty_pages(&mut merger, MergeSide::Local);
+                merger
+                    .push_page(MergeSide::Source, vec![], false, None)
+                    .expect("new source EOF");
+            }
+            let next = merger
+                .finish(false)
+                .expect("new EOF is progress")
+                .next_token
+                .expect("other side truncated");
+            assert_eq!(next.no_progress, None);
+            assert_eq!(next.v, 1);
+            assert_eq!(next.local_done, finished_side == MergeSide::Local);
+            assert_eq!(next.source_done, finished_side == MergeSide::Source);
+            let mut merger = ListThroughMerger::new(2, Some(&next));
+            let remaining = if finished_side == MergeSide::Local {
+                MergeSide::Source
+            } else {
+                MergeSide::Local
+            };
+            push_empty_pages(&mut merger, remaining);
+            let next = merger
+                .finish(true)
+                .expect("a new budget starts")
+                .next_token
+                .expect("truncated");
+            assert_eq!(next.no_progress, Some(1), "an already-done side cannot reset every page");
+        }
+        let resume = progress_token(Some(MAX_LIST_NO_PROGRESS_PAGES - 1), true, false);
+        let mut merger = ListThroughMerger::new(2, Some(&resume));
+        merger.push_page(MergeSide::Source, vec![], false, None).expect("final EOF");
+        let outcome = merger.finish(false).expect("EOF succeeds at the budget boundary");
+        assert!(!outcome.is_truncated);
+        assert!(outcome.next_token.is_none());
+    }
+
+    #[test]
+    fn filtered_duplicates_cannot_reset_the_no_progress_budget() {
+        let resume = progress_token(Some(MAX_LIST_NO_PROGRESS_PAGES - 1), true, false);
+        let mut merger = ListThroughMerger::new(2, Some(&resume));
+        for next in ["B", "C"] {
+            let entries = [ListEntryKey::object("last-key"), ListEntryKey::object("earlier")]
+                .into_iter()
+                .filter(|entry| merger.accepts(&entry.name))
+                .collect::<Vec<_>>();
+            assert!(entries.is_empty(), "both provider entries were already consumed");
+            merger
+                .push_page(MergeSide::Source, entries, true, Some(next.into()))
+                .expect("advancing cursor");
+        }
+        assert_eq!(merger.finish(false), Err(ListPageError::NoProgress(MergeSide::Source)));
+    }
+
+    #[test]
+    fn no_progress_is_attributed_to_local_when_source_cannot_unblock_it() {
+        for source_mode in ["disabled", "done", "empty", "data"] {
+            let resume = progress_token(Some(MAX_LIST_NO_PROGRESS_PAGES - 1), false, source_mode == "done");
+            let mut merger = ListThroughMerger::new(2, Some(&resume));
+            if source_mode == "disabled" {
+                merger.disable_source();
+            }
+            push_empty_pages(&mut merger, MergeSide::Local);
+            match source_mode {
+                "empty" => push_empty_pages(&mut merger, MergeSide::Source),
+                "data" => merger
+                    .push_page(MergeSide::Source, vec![ListEntryKey::object("source")], false, None)
+                    .expect("source data"),
+                _ => {}
+            }
+            assert_eq!(merger.finish(false), Err(ListPageError::NoProgress(MergeSide::Local)), "{source_mode}");
+        }
+    }
+
+    #[test]
+    fn source_budget_failure_remerges_local_objects_and_prefixes_without_refetching() {
+        let resume = progress_token(Some(MAX_LIST_NO_PROGRESS_PAGES - 1), false, false);
+        let mut merger = ListThroughMerger::new(2, Some(&resume));
+        merger
+            .push_page(MergeSide::Local, vec![ListEntryKey::object("local")], true, Some("L1".into()))
+            .expect("local object");
+        merger
+            .push_page(MergeSide::Local, vec![ListEntryKey::prefix("prefix/")], true, Some("L2".into()))
+            .expect("local prefix");
+        push_empty_pages(&mut merger, MergeSide::Source);
+        assert_eq!(merger.finish(false), Err(ListPageError::NoProgress(MergeSide::Source)));
+        merger.disable_source();
+        assert!(merger.next_fetch().is_none(), "fallback does not perform another fetch");
+        let outcome = merger.finish(false).expect("local data makes progress");
+        assert_eq!(
+            outcome.picks,
+            vec![
+                MergePick {
+                    side: MergeSide::Local,
+                    index: 0
+                },
+                MergePick {
+                    side: MergeSide::Local,
+                    index: 1
+                }
+            ]
+        );
+        let token = outcome.next_token.expect("remaining local page");
+        assert_eq!(token.local.as_deref(), Some("L2"));
+        assert_eq!(token.source.as_deref(), Some("A"));
+        assert_eq!(token.last_key.as_deref(), Some("prefix/"));
+        assert_eq!(token.no_progress, None);
+        assert_eq!(token.v, 1);
+    }
+
+    #[test]
+    fn a_zero_sized_merge_preserves_an_existing_budget() {
+        let resume = progress_token(Some(MAX_LIST_NO_PROGRESS_PAGES - 1), true, false);
+        let mut merger = ListThroughMerger::new(0, Some(&resume));
+        merger
+            .push_page(MergeSide::Source, vec![ListEntryKey::object("result")], true, Some("B".into()))
+            .expect("source page");
+        let outcome = merger.finish(false).expect("a zero-sized request cannot consume entries");
+        assert!(outcome.picks.is_empty());
+        assert_eq!(outcome.next_token.expect("unconsumed source").no_progress, resume.no_progress);
+    }
+
     #[test]
     fn a_plain_local_marker_stays_local() {
         assert_eq!(

crates/ecstore/src/bucket/on_demand_migration/mod.rs

@@ -40,9 +40,10 @@ pub use config::{
     SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
 };
 pub use list_through::{
-    FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken,
-    ListThroughTokenError, MAX_LIST_FETCHES_PER_SIDE, MergeOutcome, MergePick, MergeSide, SOURCE_LIST_MAX_RATE_WAIT,
-    SOURCE_LIST_RATE_PER_SEC, SourceListPlan, SourceListRateLimiter, decode_continuation_token, source_list_plan,
+    FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger,
+    ListThroughToken, ListThroughTokenError, MAX_LIST_FETCHES_PER_SIDE, MAX_LIST_NO_PROGRESS_PAGES, MergeOutcome, MergePick,
+    MergeSide, SOURCE_LIST_MAX_RATE_WAIT, SOURCE_LIST_RATE_PER_SEC, SourceListPlan, SourceListRateLimiter,
+    decode_continuation_token, source_list_plan,
 };
 pub use negative_cache::{NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache};
 pub use pull::{

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

@@ -30,6 +30,7 @@ use rustfs_protos::{
     ChannelClass, create_new_channel, get_channel_for_class,
     proto_gen::node_service::{
         heal_control_service_client::HealControlServiceClient, node_service_client::NodeServiceClient,
+        scanner_control_service_client::ScannerControlServiceClient,
         tier_mutation_control_service_client::TierMutationControlServiceClient,
     },
 };
@@ -60,6 +61,24 @@ pub async fn node_service_time_out_client(
     node_service_time_out_client_for_class(addr, interceptor, ChannelClass::Control).await
 }
 
+pub(crate) async fn scanner_control_time_out_client(
+    addr: &str,
+    interceptor: TonicInterceptor,
+) -> crate::error::Result<ScannerControlServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>> {
+    let interceptor = interceptor.with_rpc_audience(addr)?;
+    let channel = match runtime_sources::cached_node_channel(addr).await {
+        Some(channel) => channel,
+        None => create_new_channel(addr)
+            .await
+            .map_err(|err| crate::error::Error::other(err.to_string()))?,
+    };
+    let channel = ReplayScopeChannel::new(channel, interceptor.replay_scope_audience());
+    let limit = rustfs_protos::scoped_dirty_usage::SCOPED_DIRTY_USAGE_MAX_REQUEST_BYTES as usize;
+    Ok(ScannerControlServiceClient::with_interceptor(channel, interceptor)
+        .max_decoding_message_size(limit)
+        .max_encoding_message_size(limit))
+}
+
 pub async fn heal_control_time_out_client(
     addr: &str,
     interceptor: TonicInterceptor,

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

@@ -2050,6 +2050,53 @@ impl PeerRestClient {
         .await
     }
 
+    /// Probe only: scoped ACK production requires a durable per-bucket proof.
+    pub async fn scanner_scoped_dirty_usage_capability(
+        &self,
+        owner_id: String,
+        instance_id: String,
+        entries: Vec<rustfs_protos::proto_gen::node_service::ScannerScopedDirtyUsageEntry>,
+    ) -> Result<bool> {
+        use rustfs_protos::scoped_dirty_usage::*;
+        let payload = rustfs_protos::proto_gen::node_service::ScannerScopedDirtyUsageAckRequest {
+            challenge: Uuid::new_v4().as_bytes().to_vec().into(),
+            protocol_version: SCOPED_DIRTY_USAGE_PROTOCOL_VERSION,
+            owner_id,
+            instance_id,
+            scope: SCOPED_DIRTY_USAGE_BUCKET_SCOPE,
+            probe_only: true,
+            entries,
+        };
+        let canonical = canonical_scoped_dirty_usage_request(&payload).map_err(|err| Error::other(err.to_string()))?;
+        self.finalize_result(
+            async {
+                let mut client = super::client::scanner_control_time_out_client(
+                    &self.grid_host,
+                    TonicInterceptor::Signature(gen_tonic_signature_interceptor()),
+                )
+                .await?;
+                let mut request = Request::new(payload.clone());
+                set_tonic_canonical_body_digest(&mut request, &canonical)?;
+                let response = client.scanner_scoped_dirty_usage_ack(request).await?.into_inner();
+                let body = canonical_scoped_dirty_usage_response(&canonical, &response)
+                    .map_err(|_| Error::other("scoped dirty usage capability response is too large"))?;
+                verify_tonic_rpc_response_proof(&body, response.response_proof.as_ref())?;
+                if response.protocol_version != SCOPED_DIRTY_USAGE_PROTOCOL_VERSION
+                    || response.owner_id != payload.owner_id
+                    || response.instance_id != payload.instance_id
+                    || response.max_entries != SCOPED_DIRTY_USAGE_MAX_ENTRIES
+                    || response.max_request_bytes != SCOPED_DIRTY_USAGE_MAX_REQUEST_BYTES
+                    || response.cleared != 0
+                {
+                    return Err(Error::other("scoped dirty usage capability response does not match request"));
+                }
+                Ok(response.supported)
+            }
+            .await,
+        )
+        .await
+    }
+
     pub async fn acknowledge_scanner_dirty_usage(&self, instance_id: String, generation: u64) -> Result<ScannerPeerActivity> {
         let result = self
             .scanner_activity_request_with_protocol(instance_id.clone(), generation, SCANNER_ACTIVITY_PROTOCOL_VERSION)

crates/heal/src/heal/manager.rs

@@ -45,6 +45,11 @@ use tracing::{debug, error, info, warn};
 use super::{DiskError, Endpoint, HealDiskExt as _, local_disk_map_read};
 
 const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60);
+// Each cache includes alias tokens in its count and byte budget. Eviction
+// removes every token sharing a snapshot; neither cache retains repair state.
+const MAX_COMPLETED_HEAL_TOKENS: usize = 1024;
+const MAX_COMPLETED_HEAL_BYTES: usize = 64 * 1024 * 1024;
+const MAX_COMPLETED_HEAL_RESULT_BYTES: usize = 1024 * 1024;
 const DISPLACED_HEAL_REASON: &str = "reason=displaced; retry_hint=submit_again";
 const LOG_COMPONENT_HEAL: &str = "heal";
 const LOG_SUBSYSTEM_DISK_SCANNER: &str = "disk_scanner";
@@ -180,6 +185,8 @@ fn record_displaced_terminal(
     request: &HealRequest,
 ) -> Arc<CompletedHealStatus> {
     let terminal = Arc::new(CompletedHealStatus {
+        progress: None,
+        retained_bytes: std::sync::OnceLock::new(),
         heal_type: request.heal_type.clone(),
         status: HealTaskStatus::Failed {
             error: format!("heal task displaced by a higher-priority request ({DISPLACED_HEAL_REASON})"),
@@ -193,6 +200,7 @@ fn record_displaced_terminal(
     let mut terminals = lock_displaced_terminals(registry);
     prune_completed_heal_statuses(&mut terminals);
     terminals.insert(request.id.clone(), Arc::clone(&terminal));
+    prune_completed_heal_statuses(&mut terminals);
     terminal
 }
 
@@ -209,9 +217,15 @@ async fn remove_displaced_task_aliases(
         .collect::<Vec<_>>();
     let mut displaced_terminals = lock_displaced_terminals(terminals);
     prune_completed_heal_statuses(&mut displaced_terminals);
-    for alias_id in alias_ids {
-        displaced_terminals.insert(alias_id, Arc::clone(terminal));
+    if displaced_terminals
+        .get(task_id)
+        .is_some_and(|current| Arc::ptr_eq(current, terminal))
+    {
+        for alias_id in alias_ids {
+            displaced_terminals.insert(alias_id, Arc::clone(terminal));
+        }
     }
+    prune_completed_heal_statuses(&mut displaced_terminals);
     aliases.retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id);
 }
 
@@ -222,6 +236,36 @@ async fn remove_task_aliases_for_task(registry: &Arc<Mutex<HashMap<String, HealT
         .retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id);
 }
 
+// Callers hold active ownership until publication. Lock order is active ->
+// retrying (when needed) -> aliases -> completed; queries release aliases
+// before looking up active state. Publishing aliases before removing their
+// mapping keeps both an already-resolved token and a new lookup valid.
+async fn publish_completed_heal(
+    completed_heals: &Mutex<HashMap<String, Arc<CompletedHealStatus>>>,
+    task_aliases: &Mutex<HashMap<String, HealTaskAlias>>,
+    task_id: &str,
+    completed: CompletedHealStatus,
+    terminal: bool,
+) {
+    let completed = Arc::new(completed);
+    completed.retained_bytes();
+    let mut aliases = task_aliases.lock().await;
+    let mut retained = completed_heals.lock().await;
+    if let Some(previous) = retained.get(task_id).cloned() {
+        for entry in retained.values_mut().filter(|entry| Arc::ptr_eq(entry, &previous)) {
+            *entry = Arc::clone(&completed);
+        }
+    }
+    retained.insert(task_id.to_owned(), Arc::clone(&completed));
+    if terminal {
+        for (alias_id, _) in aliases.iter().filter(|(_, alias)| alias.task_id == task_id) {
+            retained.insert(alias_id.clone(), Arc::clone(&completed));
+        }
+        aliases.retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id);
+    }
+    prune_completed_heal_statuses(&mut retained);
+}
+
 #[derive(Debug, Clone)]
 pub struct HealTaskReport {
     pub status: HealTaskStatus,
@@ -268,7 +312,7 @@ fn completed_task_report(completed: &CompletedHealStatus, since: Option<u64>) ->
     let result_items = match since {
         None => completed.seqed_items.iter().map(|(_, item)| item.clone()).collect(),
         Some(cursor) => {
-            if cursor + 1 < completed.min_seq {
+            if cursor.saturating_add(1) < completed.min_seq {
                 lagged = true;
             }
             completed
@@ -283,7 +327,7 @@ fn completed_task_report(completed: &CompletedHealStatus, since: Option<u64>) ->
         status: completed.status.clone(),
         result_items,
         result_items_truncated: completed.result_items_truncated || lagged,
-        progress: None,
+        progress: completed.progress.clone(),
         next_seq: completed.next_seq,
         min_seq: completed.min_seq,
     }
@@ -1847,14 +1891,14 @@ impl HealManager {
 
     pub async fn get_task_progress(&self, task_id: &str) -> Result<HealProgress> {
         let canonical_task_id = self.canonical_task_id(task_id).await;
-        let active_heals = self.active_heals.lock().await;
-        if let Some(task) = active_heals.get(&canonical_task_id) {
-            Ok(task.get_progress().await)
-        } else {
-            Err(Error::TaskNotFound {
-                task_id: task_id.to_string(),
-            })
-        }
+        let progress = match self.lookup_task_state(&canonical_task_id, None).await {
+            TaskStateLookup::Active(task) => Some(task.get_progress().await),
+            TaskStateLookup::Completed(completed) => completed.progress.clone(),
+            _ => None,
+        };
+        progress.ok_or_else(|| Error::TaskNotFound {
+            task_id: task_id.to_string(),
+        })
     }
 
     /// Cancel task
@@ -1864,6 +1908,8 @@ impl HealManager {
             let mut active_heals = self.active_heals.lock().await;
             if let Some(task) = active_heals.get(&canonical_task_id) {
                 task.cancel().await?;
+                let completed = CompletedHealStatus::snapshot(task, HealTaskStatus::Cancelled).await;
+                publish_completed_heal(&self.completed_heals, &self.task_aliases, &canonical_task_id, completed, true).await;
                 active_heals.remove(&canonical_task_id);
                 publish_active_heal_count(&active_heals);
                 info!(
@@ -1940,6 +1986,8 @@ impl HealManager {
             for task_id in &task_ids {
                 if let Some(task) = active_heals.get(task_id) {
                     task.cancel().await?;
+                    let completed = CompletedHealStatus::snapshot(task, HealTaskStatus::Cancelled).await;
+                    publish_completed_heal(&self.completed_heals, &self.task_aliases, task_id, completed, true).await;
                 }
                 active_heals.remove(task_id);
                 cancelled += 1;

crates/heal/src/heal/manager/queue.rs

@@ -82,6 +82,8 @@ pub(super) enum QueuePushOutcome {
 pub(super) struct CompletedHealStatus {
     pub(super) heal_type: HealType,
     pub(super) status: HealTaskStatus,
+    pub(super) progress: Option<HealProgress>,
+    pub(super) retained_bytes: std::sync::OnceLock<usize>,
     pub(super) result_items_truncated: bool,
     pub(super) completed_at: SystemTime,
     /// Sequence-stamped retained window, archived with the completion so
@@ -92,6 +94,133 @@ pub(super) struct CompletedHealStatus {
     pub(super) min_seq: u64,
 }
 
+impl CompletedHealStatus {
+    // Account for owned capacities, including nested drive arrays. Aliases
+    // conservatively charge the shared allocation again, keeping both token
+    // count and retained payload bounded without a second ownership index.
+    pub(super) fn retained_bytes(&self) -> usize {
+        *self.retained_bytes.get_or_init(|| self.measure_retained_bytes())
+    }
+
+    fn measure_retained_bytes(&self) -> usize {
+        let mut bytes = size_of::<Self>();
+        let mut add = |amount: usize| bytes = bytes.saturating_add(amount);
+        match &self.heal_type {
+            HealType::Cluster => {}
+            HealType::Bucket { bucket } => add(bucket.capacity()),
+            HealType::Object {
+                bucket,
+                object,
+                version_id,
+            }
+            | HealType::ECDecode {
+                bucket,
+                object,
+                version_id,
+            } => {
+                add(bucket.capacity());
+                add(object.capacity());
+                add(version_id.as_ref().map_or(0, String::capacity));
+            }
+            HealType::Prefix { bucket, prefix } => {
+                add(bucket.capacity());
+                add(prefix.capacity());
+            }
+            HealType::Metadata { bucket, object } => {
+                add(bucket.capacity());
+                add(object.capacity());
+            }
+            HealType::ErasureSet { buckets, set_disk_id } => {
+                add(buckets.capacity().saturating_mul(size_of::<String>()));
+                for bucket in buckets {
+                    add(bucket.capacity());
+                }
+                add(set_disk_id.capacity());
+            }
+        }
+        if let HealTaskStatus::Failed { error } | HealTaskStatus::Retrying { error, .. } = &self.status {
+            add(error.capacity());
+        }
+        add(self
+            .progress
+            .as_ref()
+            .and_then(|progress| progress.current_object.as_ref())
+            .map_or(0, String::capacity));
+        add(self.seqed_items.capacity().saturating_mul(size_of::<(u64, HealResultItem)>()));
+        for (_, item) in &self.seqed_items {
+            add(Self::result_item_heap_bytes(item));
+        }
+        bytes
+    }
+
+    fn result_item_heap_bytes(item: &HealResultItem) -> usize {
+        let mut bytes = 0usize;
+        let mut add = |amount: usize| bytes = bytes.saturating_add(amount);
+        for value in [
+            &item.heal_item_type,
+            &item.bucket,
+            &item.object,
+            &item.version_id,
+            &item.detail,
+        ] {
+            add(value.capacity());
+        }
+        for infos in [&item.before, &item.after] {
+            add(infos
+                .drives
+                .capacity()
+                .saturating_mul(size_of::<rustfs_madmin::heal_commands::HealDriveInfo>()));
+            for drive in &infos.drives {
+                add(drive.uuid.capacity());
+                add(drive.endpoint.capacity());
+                add(drive.state.capacity());
+            }
+        }
+        bytes
+    }
+
+    pub(super) fn bound_result_window(&mut self) {
+        let mut bytes = 0usize;
+        let retained = self
+            .seqed_items
+            .iter()
+            .rev()
+            .take_while(|(_, item)| {
+                bytes = bytes
+                    .saturating_add(size_of::<(u64, HealResultItem)>())
+                    .saturating_add(Self::result_item_heap_bytes(item));
+                bytes <= MAX_COMPLETED_HEAL_RESULT_BYTES
+            })
+            .count();
+        let truncated = retained < self.seqed_items.len();
+        if truncated {
+            self.seqed_items.drain(..self.seqed_items.len() - retained);
+            self.seqed_items.shrink_to_fit();
+            self.min_seq = self.seqed_items.first().map_or(self.next_seq, |(seq, _)| *seq);
+            self.result_items_truncated = true;
+            self.retained_bytes.take();
+        }
+    }
+
+    pub(super) async fn snapshot(task: &HealTask, status: HealTaskStatus) -> Self {
+        let seqed_items = task.get_seqed_result_items().await;
+        let (next_seq, min_seq) = task.result_seq_cursors();
+        let mut snapshot = Self {
+            heal_type: task.heal_type.clone(),
+            status,
+            progress: Some(task.get_progress().await),
+            retained_bytes: std::sync::OnceLock::new(),
+            result_items_truncated: task.result_items_truncated(),
+            completed_at: SystemTime::now(),
+            seqed_items,
+            next_seq,
+            min_seq,
+        };
+        snapshot.bound_result_window();
+        snapshot
+    }
+}
+
 #[derive(Debug, Clone)]
 pub(super) struct HealTaskAlias {
     pub(super) task_id: String,

crates/heal/src/heal/manager/scheduler.rs

@@ -264,7 +264,7 @@ impl HealManager {
                         error: error.clone(),
                         retry_attempt: request.retry_attempts,
                     });
-                    let retry_request_for_queue = retry_request;
+                    let mut retry_request_for_queue = retry_request;
                     let retry_cancel_token = retry_request_for_queue.as_ref().map(|_| CancellationToken::new());
                     if retry_request_for_queue.is_none() {
                         replacement_recovery_anchors_clone
@@ -272,7 +272,35 @@ impl HealManager {
                             .unwrap_or_else(|poisoned| poisoned.into_inner())
                             .remove(&task_id);
                     }
+                    let mut completed_status = match retry_request_for_status {
+                        Some(status) => status,
+                        None => task.get_status().await,
+                    };
+                    let mut completed_status_entry = CompletedHealStatus::snapshot(&task, completed_status.clone()).await;
+                    let completed_progress = task.get_progress().await;
+                    #[cfg(test)]
+                    tests::pause_completed_retention_before_publish(&task_id, &completed_status).await;
                     let mut active_heals_guard = active_heals_clone.lock().await;
+                    let owns_completion = active_heals_guard.contains_key(&task_id);
+                    let cancelled_completion = if owns_completion {
+                        false
+                    } else {
+                        // Cancellation can win while a finished worker waits
+                        // for active ownership. It must not resurrect a retry
+                        // or replace an acknowledged cancellation with success.
+                        retry_request_for_queue = None;
+                        completed_heals_clone
+                            .lock()
+                            .await
+                            .get(&task_id)
+                            .is_some_and(|completed| completed.status == HealTaskStatus::Cancelled)
+                    };
+                    if cancelled_completion {
+                        completed_status = HealTaskStatus::Cancelled;
+                        completed_status_entry.status = HealTaskStatus::Cancelled;
+                    }
+                    let terminal_completion = !matches!(completed_status, HealTaskStatus::Retrying { .. });
+                    let successful_completion = matches!(completed_status, HealTaskStatus::Completed);
                     // Keep retry ownership continuous: status snapshots acquire
                     // these locks in the same active -> retrying order.
                     let mut retrying_heals_guard = if let (Some((request, _, error)), Some(cancel_token)) =
@@ -295,6 +323,16 @@ impl HealManager {
                     } else {
                         None
                     };
+                    if owns_completion || cancelled_completion {
+                        publish_completed_heal(
+                            &completed_heals_clone,
+                            &task_aliases_clone,
+                            &task_id,
+                            completed_status_entry,
+                            terminal_completion,
+                        )
+                        .await;
+                    }
                     let completed_task = active_heals_guard.remove(&task_id);
                     if let Some(completed_task) = completed_task.as_ref() {
                         publish_active_heal_count(&active_heals_guard);
@@ -304,33 +342,10 @@ impl HealManager {
                     drop(retrying_heals_guard.take());
                     drop(active_heals_guard);
 
-                    if let Some(completed_task) = completed_task {
-                        let completed_status = if let Some(status) = retry_request_for_status {
-                            status
-                        } else {
-                            completed_task.get_status().await
-                        };
-                        let terminal_completion = !matches!(completed_status, HealTaskStatus::Retrying { .. });
-                        let successful_completion = matches!(completed_status, HealTaskStatus::Completed);
-                        let completed_progress = completed_task.get_progress().await;
-                        // Single snapshot of the retained window: the task is
-                        // finished and already off the active map, so there is
-                        // no concurrent writer to race with.
-                        let seqed_items = completed_task.get_seqed_result_items().await;
-                        let (next_seq, min_seq) = completed_task.result_seq_cursors();
-                        let completed_status_entry = CompletedHealStatus {
-                            heal_type: completed_task.heal_type.clone(),
-                            status: completed_status.clone(),
-                            result_items_truncated: completed_task.result_items_truncated(),
-                            completed_at: SystemTime::now(),
-                            seqed_items,
-                            next_seq,
-                            min_seq,
-                        };
-                        let mut completed_heals_guard = completed_heals_clone.lock().await;
-                        prune_completed_heal_statuses(&mut completed_heals_guard);
-                        completed_heals_guard.insert(task_id.clone(), Arc::new(completed_status_entry));
-                        drop(completed_heals_guard);
+                    #[cfg(test)]
+                    tests::pause_completed_retention_handoff(&task_id).await;
+
+                    if completed_task.is_some() {
                         // update statistics
                         let mut stats = statistics_clone.write().await;
                         match completed_status {
@@ -352,10 +367,6 @@ impl HealManager {
                             } else {
                                 release_mrf_repair_notice_targets(notice_targets);
                             }
-                            task_aliases_clone
-                                .lock()
-                                .await
-                                .retain(|alias_id, alias| alias_id != &task_id && alias.task_id != task_id);
                         }
                     }
 
@@ -718,17 +729,42 @@ pub(super) fn heal_request_set_key_for_task(task: &HealTask) -> Option<String> {
 }
 
 pub(super) fn prune_completed_heal_statuses(completed_heals: &mut HashMap<String, Arc<CompletedHealStatus>>) {
-    let Ok(now) = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) else {
-        return;
-    };
+    prune_completed_heal_statuses_at(completed_heals, SystemTime::now());
+}
 
+pub(super) fn prune_completed_heal_statuses_at(completed_heals: &mut HashMap<String, Arc<CompletedHealStatus>>, now: SystemTime) {
     completed_heals.retain(|_, completed| {
-        completed
-            .completed_at
-            .duration_since(SystemTime::UNIX_EPOCH)
-            .map(|completed_at| now.saturating_sub(completed_at) <= KEEP_HEAL_TASK_STATUS_DURATION)
+        now.duration_since(completed.completed_at)
+            .map(|age| age <= KEEP_HEAL_TASK_STATUS_DURATION)
             .unwrap_or(false)
     });
+    let entry_bytes = |key: &String, value: &Arc<CompletedHealStatus>| {
+        key.capacity()
+            .saturating_add(size_of::<(String, Arc<CompletedHealStatus>)>())
+            .saturating_add(value.retained_bytes())
+    };
+    let mut bytes = completed_heals
+        .iter()
+        .fold(0usize, |total, (key, value)| total.saturating_add(entry_bytes(key, value)));
+    while completed_heals.len() > MAX_COMPLETED_HEAL_TOKENS || bytes > MAX_COMPLETED_HEAL_BYTES {
+        let Some(oldest) = completed_heals
+            .iter()
+            .min_by(|(left_id, left), (right_id, right)| {
+                left.completed_at.cmp(&right.completed_at).then_with(|| left_id.cmp(right_id))
+            })
+            .map(|(_, value)| Arc::clone(value))
+        else {
+            break;
+        };
+        completed_heals.retain(|key, value| {
+            if Arc::ptr_eq(value, &oldest) {
+                bytes = bytes.saturating_sub(entry_bytes(key, value));
+                false
+            } else {
+                true
+            }
+        });
+    }
 }
 
 pub(super) fn can_schedule_request(

crates/heal/src/heal/manager/tests.rs

@@ -101,6 +101,326 @@ async fn process_manager_queue_once(manager: &HealManager) {
 
 struct MockStorage;
 
+fn completed_retention_fixture(completed_at: SystemTime) -> CompletedHealStatus {
+    CompletedHealStatus {
+        heal_type: HealType::Cluster,
+        status: HealTaskStatus::Completed,
+        progress: Some(HealProgress {
+            objects_scanned: 9,
+            objects_healed: 8,
+            objects_failed: 1,
+            ..Default::default()
+        }),
+        retained_bytes: std::sync::OnceLock::new(),
+        result_items_truncated: false,
+        completed_at,
+        seqed_items: vec![(3, HealResultItem::default()), (4, HealResultItem::default())],
+        next_seq: 5,
+        min_seq: 3,
+    }
+}
+
+#[test]
+fn completed_retention_cursor_boundaries_preserve_progress() {
+    let completed = completed_retention_fixture(SystemTime::now());
+    for (cursor, count, lagged) in [
+        (0, 2, true),
+        (1, 2, true),
+        (2, 2, false),
+        (3, 1, false),
+        (4, 0, false),
+        (5, 0, false),
+        (u64::MAX, 0, false),
+    ] {
+        let report = completed_task_report(&completed, Some(cursor));
+        assert_eq!(report.result_items.len(), count, "cursor={cursor}");
+        assert_eq!(report.result_items_truncated, lagged, "cursor={cursor}");
+        assert_eq!(report.progress, completed.progress);
+        assert_eq!((report.next_seq, report.min_seq), (5, 3));
+    }
+    assert_eq!(completed_task_report(&completed, None).result_items.len(), 2);
+}
+
+#[tokio::test]
+async fn completed_retention_displaced_alias_does_not_resurrect_evicted_snapshot() {
+    let manager = HealManager::new(Arc::new(MockStorage), None);
+    let request = HealRequest::bucket("bucket".to_string());
+    manager.insert_task_alias("alias", &request.id).await;
+    let terminal = record_displaced_terminal(&manager.displaced_terminals, &request);
+    lock_displaced_terminals(&manager.displaced_terminals).remove(&request.id);
+    remove_displaced_task_aliases(&manager.task_aliases, &manager.displaced_terminals, &request.id, &terminal).await;
+    for token in [&request.id, &"alias".to_string()] {
+        assert!(matches!(manager.get_task_report(token).await, Err(Error::TaskNotFound { .. })));
+    }
+    assert!(manager.task_aliases.lock().await.is_empty());
+    assert!(lock_displaced_terminals(&manager.displaced_terminals).is_empty());
+}
+
+#[test]
+fn completed_retention_count_ttl_and_alias_eviction_are_bounded() {
+    let now = SystemTime::now();
+    let mut entries = HashMap::new();
+    let oldest = Arc::new(completed_retention_fixture(now - KEEP_HEAL_TASK_STATUS_DURATION));
+    entries.insert("oldest".to_string(), Arc::clone(&oldest));
+    entries.insert("oldest-alias".to_string(), Arc::clone(&oldest));
+    for index in 2..MAX_COMPLETED_HEAL_TOKENS {
+        entries.insert(format!("task-{index}"), Arc::new(completed_retention_fixture(now)));
+    }
+    prune_completed_heal_statuses_at(&mut entries, now);
+    assert_eq!(entries.len(), MAX_COMPLETED_HEAL_TOKENS);
+    entries.insert("cap-plus-one".to_string(), Arc::new(completed_retention_fixture(now)));
+    prune_completed_heal_statuses_at(&mut entries, now);
+    assert_eq!(entries.len(), MAX_COMPLETED_HEAL_TOKENS - 1);
+    assert!(!entries.contains_key("oldest"));
+    assert!(!entries.contains_key("oldest-alias"));
+    entries.clear();
+    entries.insert("ttl-boundary".to_string(), oldest);
+    entries.insert(
+        "expired".to_string(),
+        Arc::new(completed_retention_fixture(
+            now - KEEP_HEAL_TASK_STATUS_DURATION - Duration::from_nanos(1),
+        )),
+    );
+    entries.insert("future".to_string(), Arc::new(completed_retention_fixture(now + Duration::from_nanos(1))));
+    prune_completed_heal_statuses_at(&mut entries, now);
+    assert_eq!(entries.len(), 1);
+    assert!(entries.contains_key("ttl-boundary"));
+    prune_completed_heal_statuses_at(&mut entries, now + Duration::from_nanos(1));
+    assert!(entries.is_empty());
+}
+
+#[test]
+fn completed_retention_total_byte_cap_and_cap_plus_one() {
+    let now = SystemTime::now();
+    let key = "large".to_string();
+    let mut entry = completed_retention_fixture(now);
+    let base_bytes = entry.retained_bytes() + key.capacity() + size_of::<(String, Arc<CompletedHealStatus>)>();
+    entry.retained_bytes.take();
+    entry.status = HealTaskStatus::Failed {
+        error: "x".repeat(MAX_COMPLETED_HEAL_BYTES - base_bytes),
+    };
+    assert_eq!(
+        entry.retained_bytes() + key.capacity() + size_of::<(String, Arc<CompletedHealStatus>)>(),
+        MAX_COMPLETED_HEAL_BYTES
+    );
+    let mut entries = HashMap::from([(key, Arc::new(entry))]);
+    prune_completed_heal_statuses_at(&mut entries, now);
+    assert_eq!(entries.len(), 1, "exact byte cap remains retained");
+    let mut over = Arc::try_unwrap(entries.remove("large").expect("entry retained")).expect("entry not shared");
+    over.retained_bytes.take();
+    if let HealTaskStatus::Failed { error } = &mut over.status {
+        *error = "x".repeat(error.len() + 1);
+    }
+    entries.insert("large".to_string(), Arc::new(over));
+    prune_completed_heal_statuses_at(&mut entries, now);
+    assert!(entries.is_empty(), "oversized metadata cannot escape total byte bound");
+}
+
+#[tokio::test]
+async fn completed_retention_large_window_keeps_cursors_and_progress() {
+    let task = HealTask::from_request(HealRequest::bucket("bucket".to_string()), Arc::new(MockStorage));
+    let mut snapshot = completed_retention_fixture(SystemTime::now());
+    snapshot.seqed_items[0].1.detail = "x".repeat(MAX_COMPLETED_HEAL_RESULT_BYTES);
+    snapshot.bound_result_window();
+    assert_eq!(snapshot.seqed_items.len(), 1);
+    assert_eq!((snapshot.min_seq, snapshot.next_seq), (4, 5));
+    assert!(snapshot.result_items_truncated);
+    assert!(snapshot.retained_bytes() < MAX_COMPLETED_HEAL_RESULT_BYTES);
+    let report = completed_task_report(&snapshot, Some(0));
+    assert_eq!(report.progress.expect("progress retained").objects_scanned, 9);
+    assert!(report.result_items_truncated);
+    let active_max = task.get_result_items_since(Some(u64::MAX)).await;
+    assert!(active_max.items.is_empty());
+    assert!(!active_max.lagged);
+}
+
+#[test]
+fn completed_retention_result_byte_cap_and_cap_plus_one() {
+    for extra in [0, 1] {
+        let mut snapshot = completed_retention_fixture(SystemTime::now());
+        snapshot.seqed_items = vec![(
+            4,
+            HealResultItem {
+                detail: "x".repeat(MAX_COMPLETED_HEAL_RESULT_BYTES - size_of::<(u64, HealResultItem)>() + extra),
+                ..Default::default()
+            },
+        )];
+        snapshot.min_seq = 4;
+        snapshot.bound_result_window();
+        assert_eq!(snapshot.seqed_items.len(), 1 - extra);
+        assert_eq!(snapshot.result_items_truncated, extra == 1);
+        assert_eq!(snapshot.min_seq, if extra == 0 { 4 } else { 5 });
+        assert_eq!(snapshot.next_seq, 5);
+        assert_eq!(snapshot.progress.as_ref().expect("progress retained").objects_scanned, 9);
+    }
+}
+
+#[derive(Default)]
+struct CompletedRetentionHook {
+    started: Notify,
+    execute: Notify,
+    handoff: Notify,
+    finish: Notify,
+    pause_before_publish: bool,
+    before_publish: Notify,
+    publish: Notify,
+    prepared_status: Mutex<Option<HealTaskStatus>>,
+}
+
+static COMPLETED_RETENTION_HOOKS: LazyLock<Mutex<HashMap<String, Arc<CompletedRetentionHook>>>> =
+    LazyLock::new(|| Mutex::new(HashMap::new()));
+
+pub(super) async fn pause_completed_retention_handoff(task_id: &str) {
+    let hook = COMPLETED_RETENTION_HOOKS.lock().await.get(task_id).cloned();
+    if let Some(hook) = hook {
+        hook.handoff.notify_one();
+        hook.finish.notified().await;
+    }
+}
+
+pub(super) async fn pause_completed_retention_before_publish(task_id: &str, status: &HealTaskStatus) {
+    let hook = COMPLETED_RETENTION_HOOKS.lock().await.get(task_id).cloned();
+    if let Some(hook) = hook.filter(|hook| hook.pause_before_publish) {
+        *hook.prepared_status.lock().await = Some(status.clone());
+        hook.before_publish.notify_one();
+        hook.publish.notified().await;
+    }
+}
+
+#[tokio::test]
+async fn completed_retention_cancel_wins_over_a_prepared_retry_snapshot() {
+    let bucket = "completed-retention-retry-cancel";
+    let manager = HealManager::new(Arc::new(MockStorage), None);
+    let request = HealRequest::object(bucket.to_string(), "object".to_string(), None);
+    let task_id = request.id.clone();
+    let duplicate = HealRequest::object(bucket.to_string(), "object".to_string(), None);
+    let alias = duplicate.id.clone();
+    let hook = Arc::new(CompletedRetentionHook {
+        pause_before_publish: true,
+        ..Default::default()
+    });
+    {
+        let mut hooks = COMPLETED_RETENTION_HOOKS.lock().await;
+        hooks.insert(bucket.to_string(), Arc::clone(&hook));
+        hooks.insert(task_id.clone(), Arc::clone(&hook));
+    }
+    manager.submit_heal_request(request).await.expect("admit original");
+    manager.submit_heal_request(duplicate).await.expect("admit alias");
+    process_manager_queue_once(&manager).await;
+    tokio::time::timeout(Duration::from_secs(5), hook.started.notified())
+        .await
+        .expect("scheduler starts");
+    let task = manager.active_heals.lock().await.get(&task_id).cloned().expect("active task");
+    task.progress.write().await.update_object_progress(1, 1, 0, 0, 4096);
+    hook.execute.notify_one();
+    tokio::time::timeout(Duration::from_secs(5), hook.before_publish.notified())
+        .await
+        .expect("retry snapshot prepared");
+    manager.cancel_task(&alias).await.expect("cancel wins active ownership");
+    assert!(matches!(*hook.prepared_status.lock().await, Some(HealTaskStatus::Retrying { .. })));
+    hook.publish.notify_one();
+    tokio::time::timeout(Duration::from_secs(5), hook.handoff.notified())
+        .await
+        .expect("scheduler finishes handoff");
+    for token in [&task_id, &alias] {
+        let report = manager.get_task_report(token).await.expect("cancelled token retained");
+        assert_eq!(report.status, HealTaskStatus::Cancelled);
+        assert_eq!(report.progress.expect("frozen progress").objects_scanned, 1);
+    }
+    assert!(!manager.retrying_heals.lock().await.contains_key(&task_id));
+    assert!(!manager.heal_queue.lock().await.contains_request_id(&task_id));
+    hook.finish.notify_one();
+    COMPLETED_RETENTION_HOOKS
+        .lock()
+        .await
+        .retain(|key, _| key != bucket && key != &task_id);
+}
+
+#[tokio::test]
+async fn completed_retention_scheduler_preserves_progress_aliases_and_atomic_handoff() {
+    for outcome in ["success", "failed", "cancelled"] {
+        let bucket = format!("completed-retention-{outcome}");
+        let hook = Arc::new(CompletedRetentionHook::default());
+        let manager = Arc::new(HealManager::new(Arc::new(MockStorage), None));
+        let request = HealRequest::object(bucket.clone(), "object".to_string(), None);
+        let task_id = request.id.clone();
+        let duplicate = HealRequest::object(bucket.clone(), "object".to_string(), None);
+        let alias = duplicate.id.clone();
+        {
+            let mut hooks = COMPLETED_RETENTION_HOOKS.lock().await;
+            hooks.insert(bucket.clone(), Arc::clone(&hook));
+            hooks.insert(task_id.clone(), Arc::clone(&hook));
+        }
+        manager.submit_heal_request(request).await.expect("admit original");
+        manager.submit_heal_request(duplicate).await.expect("admit alias");
+        process_manager_queue_once(&manager).await;
+        tokio::time::timeout(Duration::from_secs(5), hook.started.notified())
+            .await
+            .expect("scheduler reaches storage");
+        let task = manager
+            .active_heals
+            .lock()
+            .await
+            .get(&task_id)
+            .cloned()
+            .expect("task is active");
+        task.progress.write().await.update_object_progress(1, 1, 0, 0, 4096);
+        let before = manager.get_task_report(&alias).await.expect("alias resolves active progress");
+        assert_eq!(before.progress.as_ref().expect("active progress").objects_scanned, 1);
+        let poll_manager = Arc::clone(&manager);
+        let poll_alias = alias.clone();
+        let stop = CancellationToken::new();
+        let poll_stop = stop.clone();
+        let polling = tokio::spawn(async move {
+            while !poll_stop.is_cancelled() {
+                let report = poll_manager
+                    .get_task_report(&poll_alias)
+                    .await
+                    .expect("handoff must never return NotFound");
+                assert!(report.progress.expect("progress never disappears").objects_scanned >= 1);
+                tokio::task::yield_now().await;
+            }
+        });
+        if outcome == "cancelled" {
+            manager.cancel_task(&alias).await.expect("cancel active task by alias");
+        } else {
+            hook.execute.notify_one();
+        }
+        tokio::time::timeout(Duration::from_secs(5), hook.handoff.notified())
+            .await
+            .expect("scheduler archives terminal");
+        assert!(!manager.active_heals.lock().await.contains_key(&task_id));
+        let expected = task.get_progress().await;
+        for token in [&task_id, &alias] {
+            assert_eq!(manager.get_task_progress(token).await.expect("terminal progress query"), expected);
+            let report = manager
+                .get_task_report_for_path_since(&format!("{bucket}/object"), token, Some(u64::MAX))
+                .await
+                .expect("terminal token remains queryable at handoff");
+            assert_eq!(report.progress.as_ref(), Some(&expected));
+            assert!(report.result_items.is_empty());
+            match outcome {
+                "success" => assert_eq!(report.status, HealTaskStatus::Completed),
+                "failed" => assert!(matches!(report.status, HealTaskStatus::Failed { .. })),
+                _ => assert_eq!(report.status, HealTaskStatus::Cancelled),
+            }
+        }
+        let retained = manager.completed_heals.lock().await;
+        assert!(Arc::ptr_eq(&retained[&task_id], &retained[&alias]));
+        drop(retained);
+        stop.cancel();
+        polling.await.expect("concurrent polling succeeds");
+        // Archived progress must not alias a mutable live progress object.
+        task.progress.write().await.objects_scanned = 999;
+        assert_eq!(manager.get_task_report(&alias).await.expect("frozen report").progress, Some(expected));
+        hook.finish.notify_one();
+        COMPLETED_RETENTION_HOOKS
+            .lock()
+            .await
+            .retain(|key, _| key != &bucket && key != &task_id);
+    }
+}
+
 #[async_trait::async_trait]
 impl HealStorageAPI for MockStorage {
     async fn get_object_meta(&self, _bucket: &str, _object: &str) -> Result<Option<HealObjectInfo>> {
@@ -123,6 +443,12 @@ impl HealStorageAPI for MockStorage {
     }
 
     async fn object_exists(&self, bucket: &str, _object: &str) -> Result<bool> {
+        let hook = COMPLETED_RETENTION_HOOKS.lock().await.get(bucket).cloned();
+        if let Some(hook) = hook {
+            hook.started.notify_one();
+            hook.execute.notified().await;
+            return Ok(true);
+        }
         Ok(bucket == "retry-transition")
     }
 
@@ -133,13 +459,18 @@ impl HealStorageAPI for MockStorage {
         _version_id: Option<&str>,
         _opts: &HealOpts,
     ) -> Result<(HealResultItem, Option<Error>)> {
+        if bucket == "completed-retention-failed" {
+            return Err(Error::TaskExecutionFailed {
+                message: "retention fixture failure".to_string(),
+            });
+        }
         if let Some(hook) = manager_recovery_test_hook() {
             *hook
                 .heal_object_calls
                 .lock()
                 .expect("manager recovery object call lock should not poison") += 1;
         }
-        if bucket == "retry-transition" {
+        if matches!(bucket, "retry-transition" | "completed-retention-retry-cancel") {
             return Ok((
                 HealResultItem::default(),
                 Some(Error::Storage(EcstoreError::InsufficientReadQuorum(
@@ -1145,7 +1476,13 @@ async fn test_active_duplicate_token_can_query_and_cancel_original_task() {
         .expect("duplicate token should cancel merged active task");
 
     assert!(manager.active_heals.lock().await.get(&active_task_id).is_none());
-    assert!(matches!(manager.get_task_status(&active_task_id).await, Err(Error::TaskNotFound { .. })));
+    assert_eq!(
+        manager
+            .get_task_status(&active_task_id)
+            .await
+            .expect("cancelled task remains queryable"),
+        HealTaskStatus::Cancelled
+    );
 }
 
 #[tokio::test]
@@ -1638,6 +1975,8 @@ async fn insert_retrying_request(manager: &HealManager, request: HealRequest) ->
     manager.completed_heals.lock().await.insert(
         task_id,
         Arc::new(CompletedHealStatus {
+            progress: None,
+            retained_bytes: std::sync::OnceLock::new(),
             heal_type: request.heal_type,
             status: HealTaskStatus::Retrying {
                 error: "Lock acquisition timeout".to_string(),
@@ -2053,7 +2392,7 @@ async fn admin_force_start_cancels_overlapping_active_task_first() {
         "the overlapping admin task must be cancelled (removed from the active table) before the new one starts"
     );
     assert!(
-        matches!(manager.get_task_status(&old_id).await, Err(Error::TaskNotFound { .. })),
+        matches!(manager.get_task_status(&old_id).await, Ok(HealTaskStatus::Cancelled)),
         "a cancelled task must no longer resolve as an active heal"
     );
 }
@@ -2360,6 +2699,8 @@ async fn test_retrying_completion_outranks_the_queue_for_the_same_id() {
     manager.completed_heals.lock().await.insert(
         task_id.clone(),
         Arc::new(CompletedHealStatus {
+            progress: None,
+            retained_bytes: std::sync::OnceLock::new(),
             heal_type: request.heal_type.clone(),
             status: HealTaskStatus::Retrying {
                 error: "transient disk failure".to_string(),
@@ -2395,6 +2736,8 @@ async fn test_get_task_status_reads_recent_completed_status() {
     manager.completed_heals.lock().await.insert(
         "completed-token".to_string(),
         Arc::new(CompletedHealStatus {
+            progress: None,
+            retained_bytes: std::sync::OnceLock::new(),
             heal_type: HealType::Bucket {
                 bucket: "bucket".to_string(),
             },
@@ -2424,6 +2767,8 @@ async fn test_get_task_report_for_path_reads_completed_items() {
     manager.completed_heals.lock().await.insert(
         "completed-token".to_string(),
         Arc::new(CompletedHealStatus {
+            progress: None,
+            retained_bytes: std::sync::OnceLock::new(),
             heal_type: HealType::Object {
                 bucket: "bucket".to_string(),
                 object: "object".to_string(),

crates/heal/src/heal/task.rs

@@ -999,7 +999,7 @@ impl HealTask {
         let items = match since {
             None => result_items.iter().map(|(_, item)| item.clone()).collect::<Vec<_>>(),
             Some(cursor) => {
-                if cursor + 1 < min_seq {
+                if cursor.saturating_add(1) < min_seq {
                     lagged = true;
                 }
                 result_items

crates/madmin/src/client.rs

@@ -182,6 +182,9 @@ pub struct BackgroundHealStatus {
     pub heal_active_tasks: u64,
     #[serde(default)]
     pub cluster_status_complete: bool,
+    /// Missing on older servers; absent coverage or counts mean unknown.
+    #[serde(default)]
+    pub coverage: Option<BackgroundHealCoverage>,
     #[serde(default)]
     pub progress: Option<serde_json::Value>,
     /// Remaining wire fields (flattened `BackgroundHealInfo` plus the
@@ -190,6 +193,22 @@ pub struct BackgroundHealStatus {
     pub extra: serde_json::Map<String, serde_json::Value>,
 }
 
+/// Node coverage of a background heal status snapshot. Counters describe only
+/// nodes with usable snapshots; unknown peers may still be running heal work.
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BackgroundHealCoverage {
+    #[serde(default)]
+    pub expected: Option<usize>,
+    #[serde(default)]
+    pub responded: Option<usize>,
+    #[serde(default)]
+    pub unknown: Option<usize>,
+    /// Stable reason codes; unknown future codes are preserved verbatim.
+    #[serde(default)]
+    pub reasons: Vec<String>,
+}
+
 /// `GET /v3/scanner/status` response, typed at the fields operators branch
 /// on; everything else passes through verbatim.
 #[derive(Debug, Clone, Deserialize)]
@@ -630,9 +649,39 @@ mod tests {
         assert_eq!(status.state, "active");
         assert_eq!(status.heal_queue_length, 3);
         assert!(status.cluster_status_complete);
+        assert!(status.coverage.is_none(), "legacy payloads have unknown coverage");
         assert!(status.extra.contains_key("healOperations"), "unknown nested payloads must pass through");
     }
 
+    #[test]
+    fn background_heal_status_missing_coverage_fields_remain_unknown() {
+        for raw in [json!({"state": "degraded"}), json!({"state": "degraded", "coverage": {}})] {
+            let status: BackgroundHealStatus = serde_json::from_value(raw).expect("partial legacy payload decodes");
+            assert!(!status.cluster_status_complete);
+            if let Some(coverage) = status.coverage {
+                assert_eq!(coverage.expected, None);
+                assert_eq!(coverage.responded, None);
+                assert_eq!(coverage.unknown, None);
+            }
+        }
+    }
+
+    #[test]
+    fn background_heal_status_preserves_future_fields_and_reasons() {
+        let raw = json!({
+            "state": "degraded", "clusterStatusComplete": false,
+            "coverage": {"expected": 3, "responded": 1, "unknown": 2, "reasons": ["future_reason"], "futureCoverage": true},
+            "futureStatus": {"value": 7}
+        });
+        let status: BackgroundHealStatus = serde_json::from_value(raw).expect("future additive fields decode");
+        assert_eq!(status.extra["futureStatus"]["value"], 7);
+        let coverage = status.coverage.expect("coverage supplied");
+        assert_eq!(coverage.expected, Some(3));
+        assert_eq!(coverage.responded, Some(1));
+        assert_eq!(coverage.unknown, Some(2));
+        assert_eq!(coverage.reasons, ["future_reason"]);
+    }
+
     #[test]
     fn scanner_status_defaults_freshness_to_unknown() {
         let raw = json!({"enabled": true, "freshness": {"state": "stale"}, "metrics": {}});
@@ -721,6 +770,7 @@ mod tests {
 
         let status = client.background_heal_status().await.expect("status decodes");
         assert_eq!(status.state, "idle");
+        assert!(status.coverage.is_none(), "older HTTP responses retain unknown coverage");
         let request = server.recorded();
         // The server registers this route POST-only; a GET here answers 405.
         assert_eq!(request.method, "POST");
@@ -728,6 +778,28 @@ mod tests {
         assert_eq!(request.query, "");
     }
 
+    #[tokio::test]
+    async fn background_heal_status_decodes_partial_coverage_over_http() {
+        let body = r#"{"state":"degraded","healQueueLength":0,"healActiveTasks":0,"clusterStatusComplete":false,"coverage":{"expected":3,"responded":1,"unknown":2,"reasons":["notification_system_unavailable"]},"futureStatus":true}"#;
+        let server = TestServer::spawn(body, 200).await;
+        let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").expect("client builds");
+        let status = client
+            .background_heal_status()
+            .await
+            .expect("partial status is a successful response");
+        assert_eq!(status.state, "degraded");
+        assert!(!status.cluster_status_complete);
+        assert_eq!(status.extra["futureStatus"], true);
+        let coverage = status.coverage.expect("partial coverage supplied");
+        assert_eq!(coverage.expected, Some(3));
+        assert_eq!(coverage.responded, Some(1));
+        assert_eq!(coverage.unknown, Some(2));
+        assert_eq!(coverage.reasons, ["notification_system_unavailable"]);
+        let request = server.recorded();
+        assert_eq!(request.method, "POST");
+        assert_eq!(request.query, "", "reading status must not send heal control parameters");
+    }
+
     #[tokio::test]
     async fn http_error_status_maps_to_a_typed_error_with_body() {
         let server = TestServer::spawn(r#"{"code":"AccessDenied","message":"denied"}"#, 403).await;

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

@@ -1283,6 +1283,54 @@ pub struct ScannerDirtyUsageSnapshotResponse {
     #[prost(bytes = "bytes", tag = "7")]
     pub response_proof: ::prost::bytes::Bytes,
 }
+/// Receiver-only protocol. Producers must retain whole-cycle ACK until they
+/// have a durable per-bucket publication proof.
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct ScannerScopedDirtyUsageEntry {
+    #[prost(string, tag = "1")]
+    pub bucket: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "2")]
+    pub bucket_incarnation: ::prost::bytes::Bytes,
+    #[prost(uint64, tag = "3")]
+    pub generation: u64,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ScannerScopedDirtyUsageAckRequest {
+    #[prost(bytes = "bytes", tag = "1")]
+    pub challenge: ::prost::bytes::Bytes,
+    #[prost(uint32, tag = "2")]
+    pub protocol_version: u32,
+    #[prost(string, tag = "3")]
+    pub owner_id: ::prost::alloc::string::String,
+    #[prost(string, tag = "4")]
+    pub instance_id: ::prost::alloc::string::String,
+    /// Only scope 1 (a complete bucket) is supported; zero is invalid.
+    #[prost(uint32, tag = "5")]
+    pub scope: u32,
+    #[prost(bool, tag = "6")]
+    pub probe_only: bool,
+    #[prost(message, repeated, tag = "7")]
+    pub entries: ::prost::alloc::vec::Vec<ScannerScopedDirtyUsageEntry>,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct ScannerScopedDirtyUsageAckResponse {
+    #[prost(uint32, tag = "1")]
+    pub protocol_version: u32,
+    #[prost(string, tag = "2")]
+    pub owner_id: ::prost::alloc::string::String,
+    #[prost(string, tag = "3")]
+    pub instance_id: ::prost::alloc::string::String,
+    #[prost(bool, tag = "4")]
+    pub supported: bool,
+    #[prost(uint32, tag = "5")]
+    pub max_entries: u32,
+    #[prost(uint32, tag = "6")]
+    pub max_request_bytes: u32,
+    #[prost(uint64, tag = "7")]
+    pub cleared: u64,
+    #[prost(bytes = "bytes", tag = "8")]
+    pub response_proof: ::prost::bytes::Bytes,
+}
 /// A short-lived storage-owned read admission used only around a final
 /// scanner metadata publication.  It is intentionally separate from the
 /// ScannerActivity observation wire so v6/v7 rolling compatibility remains
@@ -6282,6 +6330,244 @@ pub mod node_service_server {
     }
 }
 /// Generated client implementations.
+pub mod scanner_control_service_client {
+    #![allow(unused_variables, dead_code, missing_docs, clippy::wildcard_imports, clippy::let_unit_value)]
+    use tonic::codegen::http::Uri;
+    use tonic::codegen::*;
+    #[derive(Debug, Clone)]
+    pub struct ScannerControlServiceClient<T> {
+        inner: tonic::client::Grpc<T>,
+    }
+    impl ScannerControlServiceClient<tonic::transport::Channel> {
+        /// Attempt to create a new client by connecting to a given endpoint.
+        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
+        where
+            D: TryInto<tonic::transport::Endpoint>,
+            D::Error: Into<StdError>,
+        {
+            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
+            Ok(Self::new(conn))
+        }
+    }
+    impl<T> ScannerControlServiceClient<T>
+    where
+        T: tonic::client::GrpcService<tonic::body::Body>,
+        T::Error: Into<StdError>,
+        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
+        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
+    {
+        pub fn new(inner: T) -> Self {
+            let inner = tonic::client::Grpc::new(inner);
+            Self { inner }
+        }
+        pub fn with_origin(inner: T, origin: Uri) -> Self {
+            let inner = tonic::client::Grpc::with_origin(inner, origin);
+            Self { inner }
+        }
+        pub fn with_interceptor<F>(inner: T, interceptor: F) -> ScannerControlServiceClient<InterceptedService<T, F>>
+        where
+            F: tonic::service::Interceptor,
+            T::ResponseBody: Default,
+            T: tonic::codegen::Service<
+                    http::Request<tonic::body::Body>,
+                    Response = http::Response<<T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody>,
+                >,
+            <T as tonic::codegen::Service<http::Request<tonic::body::Body>>>::Error:
+                Into<StdError> + std::marker::Send + std::marker::Sync,
+        {
+            ScannerControlServiceClient::new(InterceptedService::new(inner, interceptor))
+        }
+        /// Compress requests with the given encoding.
+        ///
+        /// This requires the server to support it otherwise it might respond with an
+        /// error.
+        #[must_use]
+        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.inner = self.inner.send_compressed(encoding);
+            self
+        }
+        /// Enable decompressing responses.
+        #[must_use]
+        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.inner = self.inner.accept_compressed(encoding);
+            self
+        }
+        /// Limits the maximum size of a decoded message.
+        ///
+        /// Default: `4MB`
+        #[must_use]
+        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
+            self.inner = self.inner.max_decoding_message_size(limit);
+            self
+        }
+        /// Limits the maximum size of an encoded message.
+        ///
+        /// Default: `usize::MAX`
+        #[must_use]
+        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
+            self.inner = self.inner.max_encoding_message_size(limit);
+            self
+        }
+        pub async fn scanner_scoped_dirty_usage_ack(
+            &mut self,
+            request: impl tonic::IntoRequest<super::ScannerScopedDirtyUsageAckRequest>,
+        ) -> std::result::Result<tonic::Response<super::ScannerScopedDirtyUsageAckResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.ScannerControlService/ScannerScopedDirtyUsageAck");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.ScannerControlService", "ScannerScopedDirtyUsageAck"));
+            self.inner.unary(req, path, codec).await
+        }
+    }
+}
+/// Generated server implementations.
+pub mod scanner_control_service_server {
+    #![allow(unused_variables, dead_code, missing_docs, clippy::wildcard_imports, clippy::let_unit_value)]
+    use tonic::codegen::*;
+    /// Generated trait containing gRPC methods that should be implemented for use with ScannerControlServiceServer.
+    #[async_trait]
+    pub trait ScannerControlService: std::marker::Send + std::marker::Sync + 'static {
+        async fn scanner_scoped_dirty_usage_ack(
+            &self,
+            request: tonic::Request<super::ScannerScopedDirtyUsageAckRequest>,
+        ) -> std::result::Result<tonic::Response<super::ScannerScopedDirtyUsageAckResponse>, tonic::Status>;
+    }
+    #[derive(Debug)]
+    pub struct ScannerControlServiceServer<T> {
+        inner: Arc<T>,
+        accept_compression_encodings: EnabledCompressionEncodings,
+        send_compression_encodings: EnabledCompressionEncodings,
+        max_decoding_message_size: Option<usize>,
+        max_encoding_message_size: Option<usize>,
+    }
+    impl<T> ScannerControlServiceServer<T> {
+        pub fn new(inner: T) -> Self {
+            Self::from_arc(Arc::new(inner))
+        }
+        pub fn from_arc(inner: Arc<T>) -> Self {
+            Self {
+                inner,
+                accept_compression_encodings: Default::default(),
+                send_compression_encodings: Default::default(),
+                max_decoding_message_size: None,
+                max_encoding_message_size: None,
+            }
+        }
+        pub fn with_interceptor<F>(inner: T, interceptor: F) -> InterceptedService<Self, F>
+        where
+            F: tonic::service::Interceptor,
+        {
+            InterceptedService::new(Self::new(inner), interceptor)
+        }
+        /// Enable decompressing requests with the given encoding.
+        #[must_use]
+        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.accept_compression_encodings.enable(encoding);
+            self
+        }
+        /// Compress responses with the given encoding, if the client supports it.
+        #[must_use]
+        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.send_compression_encodings.enable(encoding);
+            self
+        }
+        /// Limits the maximum size of a decoded message.
+        ///
+        /// Default: `4MB`
+        #[must_use]
+        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
+            self.max_decoding_message_size = Some(limit);
+            self
+        }
+        /// Limits the maximum size of an encoded message.
+        ///
+        /// Default: `usize::MAX`
+        #[must_use]
+        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
+            self.max_encoding_message_size = Some(limit);
+            self
+        }
+    }
+    impl<T, B> tonic::codegen::Service<http::Request<B>> for ScannerControlServiceServer<T>
+    where
+        T: ScannerControlService,
+        B: Body + std::marker::Send + 'static,
+        B::Error: Into<StdError> + std::marker::Send + 'static,
+    {
+        type Response = http::Response<tonic::body::Body>;
+        type Error = std::convert::Infallible;
+        type Future = BoxFuture<Self::Response, Self::Error>;
+        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+        fn call(&mut self, req: http::Request<B>) -> Self::Future {
+            match req.uri().path() {
+                "/node_service.ScannerControlService/ScannerScopedDirtyUsageAck" => {
+                    #[allow(non_camel_case_types)]
+                    struct ScannerScopedDirtyUsageAckSvc<T: ScannerControlService>(pub Arc<T>);
+                    impl<T: ScannerControlService> tonic::server::UnaryService<super::ScannerScopedDirtyUsageAckRequest>
+                        for ScannerScopedDirtyUsageAckSvc<T>
+                    {
+                        type Response = super::ScannerScopedDirtyUsageAckResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::ScannerScopedDirtyUsageAckRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as ScannerControlService>::scanner_scoped_dirty_usage_ack(&inner, request).await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = ScannerScopedDirtyUsageAckSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                _ => Box::pin(async move {
+                    let mut response = http::Response::new(tonic::body::Body::default());
+                    let headers = response.headers_mut();
+                    headers.insert(tonic::Status::GRPC_STATUS, (tonic::Code::Unimplemented as i32).into());
+                    headers.insert(http::header::CONTENT_TYPE, tonic::metadata::GRPC_CONTENT_TYPE);
+                    Ok(response)
+                }),
+            }
+        }
+    }
+    impl<T> Clone for ScannerControlServiceServer<T> {
+        fn clone(&self) -> Self {
+            let inner = self.inner.clone();
+            Self {
+                inner,
+                accept_compression_encodings: self.accept_compression_encodings,
+                send_compression_encodings: self.send_compression_encodings,
+                max_decoding_message_size: self.max_decoding_message_size,
+                max_encoding_message_size: self.max_encoding_message_size,
+            }
+        }
+    }
+    /// Generated gRPC service name
+    pub const SERVICE_NAME: &str = "node_service.ScannerControlService";
+    impl<T> tonic::server::NamedService for ScannerControlServiceServer<T> {
+        const NAME: &'static str = SERVICE_NAME;
+    }
+}
+/// Generated client implementations.
 pub mod heal_control_service_client {
     #![allow(unused_variables, dead_code, missing_docs, clippy::wildcard_imports, clippy::let_unit_value)]
     use tonic::codegen::http::Uri;

crates/protos/src/lib.rs

@@ -541,6 +541,8 @@ pub fn canonical_scanner_activity_v7_response_body(
     Ok(body)
 }
 
+pub mod scoped_dirty_usage;
+
 pub fn canonical_scanner_dirty_usage_snapshot_request_body(
     request: &proto_gen::node_service::ScannerDirtyUsageSnapshotRequest,
 ) -> Result<Vec<u8>, std::num::TryFromIntError> {

crates/protos/src/node.proto

@@ -903,6 +903,36 @@ message ScannerDirtyUsageSnapshotResponse {
   bytes response_proof = 7;
 }
 
+// Receiver-only protocol. Producers must retain whole-cycle ACK until they
+// have a durable per-bucket publication proof.
+message ScannerScopedDirtyUsageEntry {
+  string bucket = 1;
+  bytes bucket_incarnation = 2;
+  uint64 generation = 3;
+}
+
+message ScannerScopedDirtyUsageAckRequest {
+  bytes challenge = 1;
+  uint32 protocol_version = 2;
+  string owner_id = 3;
+  string instance_id = 4;
+  // Only scope 1 (a complete bucket) is supported; zero is invalid.
+  uint32 scope = 5;
+  bool probe_only = 6;
+  repeated ScannerScopedDirtyUsageEntry entries = 7;
+}
+
+message ScannerScopedDirtyUsageAckResponse {
+  uint32 protocol_version = 1;
+  string owner_id = 2;
+  string instance_id = 3;
+  bool supported = 4;
+  uint32 max_entries = 5;
+  uint32 max_request_bytes = 6;
+  uint64 cleared = 7;
+  bytes response_proof = 8;
+}
+
 // A short-lived storage-owned read admission used only around a final
 // scanner metadata publication.  It is intentionally separate from the
 // ScannerActivity observation wire so v6/v7 rolling compatibility remains
@@ -1245,6 +1275,10 @@ service NodeService {
   rpc GetLiveEvents(GetLiveEventsRequest) returns (GetLiveEventsResponse) {}; // auth-policy: read-only
 }
 
+service ScannerControlService {
+  rpc ScannerScopedDirtyUsageAck(ScannerScopedDirtyUsageAckRequest) returns (ScannerScopedDirtyUsageAckResponse) {}; // auth-policy: body-bound
+}
+
 service HealControlService {
   rpc HealControl(HealControlRequest) returns (HealControlResponse) {};
 }

crates/protos/src/scoped_dirty_usage.rs

@@ -0,0 +1,213 @@
+// Copyright 2024 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+//! Bounded, authenticated receiver contract for per-bucket dirty acknowledgements.
+
+use crate::CanonicalBodyBuilder;
+use crate::proto_gen::node_service::{ScannerScopedDirtyUsageAckRequest, ScannerScopedDirtyUsageAckResponse};
+use prost::Message;
+
+pub const SCOPED_DIRTY_USAGE_PROTOCOL_VERSION: u32 = 1;
+pub const SCOPED_DIRTY_USAGE_BUCKET_SCOPE: u32 = 1;
+pub const SCOPED_DIRTY_USAGE_MAX_ENTRIES: u32 = 32;
+pub const SCOPED_DIRTY_USAGE_MAX_REQUEST_BYTES: u32 = 8192;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ScopedDirtyUsageRequestError {
+    UnsupportedProtocol,
+    UnsupportedScope,
+    InvalidIdentity,
+    InvalidGeneration,
+    InvalidEntries,
+    TooLarge,
+}
+
+impl std::fmt::Display for ScopedDirtyUsageRequestError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(match self {
+            Self::UnsupportedProtocol => "unsupported scoped dirty usage protocol",
+            Self::UnsupportedScope => "unsupported scoped dirty usage scope",
+            Self::InvalidIdentity => "invalid scoped dirty usage identity",
+            Self::InvalidGeneration => "invalid scoped dirty usage generation",
+            Self::InvalidEntries => "scoped dirty usage entries must be nonempty and strictly ordered",
+            Self::TooLarge => "scoped dirty usage request exceeds its budget",
+        })
+    }
+}
+
+impl std::error::Error for ScopedDirtyUsageRequestError {}
+
+pub fn validate_scoped_dirty_usage_request(
+    request: &ScannerScopedDirtyUsageAckRequest,
+) -> Result<(), ScopedDirtyUsageRequestError> {
+    use ScopedDirtyUsageRequestError as E;
+    if request.entries.len() > SCOPED_DIRTY_USAGE_MAX_ENTRIES as usize
+        || request.encoded_len() > SCOPED_DIRTY_USAGE_MAX_REQUEST_BYTES as usize
+    {
+        return Err(E::TooLarge);
+    }
+    if request.protocol_version != SCOPED_DIRTY_USAGE_PROTOCOL_VERSION {
+        return Err(E::UnsupportedProtocol);
+    }
+    if request.scope != SCOPED_DIRTY_USAGE_BUCKET_SCOPE {
+        return Err(E::UnsupportedScope);
+    }
+    if request.challenge.len() != 16 || request.owner_id.len() != 36 || request.instance_id.len() != 32 {
+        return Err(E::InvalidIdentity);
+    }
+    if request.entries.is_empty() || request.entries.windows(2).any(|pair| pair[0].bucket >= pair[1].bucket) {
+        return Err(E::InvalidEntries);
+    }
+    for entry in &request.entries {
+        if entry.bucket.is_empty()
+            || entry.bucket.len() > 63
+            || entry.bucket_incarnation.len() != 16
+            || entry.bucket_incarnation.iter().all(|byte| *byte == 0)
+        {
+            return Err(E::InvalidIdentity);
+        }
+        if entry.generation == 0 || entry.generation == u64::MAX {
+            return Err(E::InvalidGeneration);
+        }
+    }
+    Ok(())
+}
+
+pub fn canonical_scoped_dirty_usage_request(
+    request: &ScannerScopedDirtyUsageAckRequest,
+) -> Result<Vec<u8>, ScopedDirtyUsageRequestError> {
+    validate_scoped_dirty_usage_request(request)?;
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-scoped-dirty-usage-ack-request-v1\0");
+    let encode = |_: std::num::TryFromIntError| ScopedDirtyUsageRequestError::TooLarge;
+    body.push_bytes(request.challenge.as_ref()).map_err(encode)?;
+    body.push_u32(request.protocol_version);
+    body.push_str(&request.owner_id).map_err(encode)?;
+    body.push_str(&request.instance_id).map_err(encode)?;
+    body.push_u32(request.scope);
+    body.push_bool(request.probe_only);
+    body.push_count(request.entries.len()).map_err(encode)?;
+    for entry in &request.entries {
+        body.push_str(&entry.bucket).map_err(encode)?;
+        body.push_bytes(entry.bucket_incarnation.as_ref()).map_err(encode)?;
+        body.push_u64(entry.generation);
+    }
+    Ok(body.finish())
+}
+
+pub fn canonical_scoped_dirty_usage_response(
+    request_body: &[u8],
+    response: &ScannerScopedDirtyUsageAckResponse,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-scoped-dirty-usage-ack-response-v1\0");
+    body.push_bytes(request_body)?;
+    body.push_u32(response.protocol_version);
+    body.push_str(&response.owner_id)?;
+    body.push_str(&response.instance_id)?;
+    body.push_bool(response.supported);
+    body.push_u32(response.max_entries);
+    body.push_u32(response.max_request_bytes);
+    body.push_u64(response.cleared);
+    Ok(body.finish())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::proto_gen::node_service::ScannerScopedDirtyUsageEntry;
+
+    fn request() -> ScannerScopedDirtyUsageAckRequest {
+        ScannerScopedDirtyUsageAckRequest {
+            challenge: vec![1; 16].into(),
+            protocol_version: 1,
+            owner_id: "11111111-1111-1111-1111-111111111111".into(),
+            instance_id: "a".repeat(32),
+            scope: 1,
+            probe_only: false,
+            entries: vec![ScannerScopedDirtyUsageEntry {
+                bucket: "photos".into(),
+                bucket_incarnation: vec![2; 16].into(),
+                generation: 8,
+            }],
+        }
+    }
+
+    #[test]
+    fn scoped_dirty_usage_binds_every_request_field() {
+        let base = request();
+        let baseline = canonical_scoped_dirty_usage_request(&base).expect("valid request");
+        for field in 0..9 {
+            let mut changed = base.clone();
+            match field {
+                0 => changed.challenge = vec![3; 16].into(),
+                1 => changed.protocol_version += 1,
+                2 => changed.owner_id = "22222222-2222-2222-2222-222222222222".into(),
+                3 => changed.instance_id = "b".repeat(32),
+                4 => changed.scope += 1,
+                5 => changed.probe_only = true,
+                6 => changed.entries[0].bucket = "videos".into(),
+                7 => changed.entries[0].bucket_incarnation = vec![3; 16].into(),
+                _ => changed.entries[0].generation += 1,
+            }
+            assert!(canonical_scoped_dirty_usage_request(&changed).map_or(true, |body| body != baseline));
+        }
+    }
+
+    #[test]
+    fn scoped_dirty_usage_binds_capability_and_ack_to_exact_request() {
+        let request = canonical_scoped_dirty_usage_request(&request()).expect("valid request");
+        let response = ScannerScopedDirtyUsageAckResponse {
+            protocol_version: 1,
+            owner_id: "owner".into(),
+            instance_id: "process".into(),
+            supported: true,
+            max_entries: 32,
+            max_request_bytes: 8192,
+            cleared: 1,
+            response_proof: vec![1; 32].into(),
+        };
+        let baseline = canonical_scoped_dirty_usage_response(&request, &response).expect("valid response");
+        for field in 0..7 {
+            let mut changed = response.clone();
+            match field {
+                0 => changed.protocol_version += 1,
+                1 => changed.owner_id.push('x'),
+                2 => changed.instance_id.push('x'),
+                3 => changed.supported = false,
+                4 => changed.max_entries += 1,
+                5 => changed.max_request_bytes += 1,
+                _ => changed.cleared += 1,
+            }
+            assert_ne!(
+                canonical_scoped_dirty_usage_response(&request, &changed).expect("response variant"),
+                baseline
+            );
+        }
+        assert_ne!(
+            canonical_scoped_dirty_usage_response(b"another request", &response).expect("request variant"),
+            baseline
+        );
+    }
+
+    #[test]
+    fn scoped_dirty_usage_rejects_overflow_unknown_and_duplicate_entries() {
+        let base = request();
+        let mut invalid = base.clone();
+        invalid.entries = vec![base.entries[0].clone(); SCOPED_DIRTY_USAGE_MAX_ENTRIES as usize + 1];
+        assert_eq!(validate_scoped_dirty_usage_request(&invalid), Err(ScopedDirtyUsageRequestError::TooLarge));
+        invalid = base.clone();
+        invalid.entries[0].bucket = "x".repeat(SCOPED_DIRTY_USAGE_MAX_REQUEST_BYTES as usize);
+        assert_eq!(validate_scoped_dirty_usage_request(&invalid), Err(ScopedDirtyUsageRequestError::TooLarge));
+        invalid = base.clone();
+        invalid.entries.push(base.entries[0].clone());
+        assert_eq!(
+            validate_scoped_dirty_usage_request(&invalid),
+            Err(ScopedDirtyUsageRequestError::InvalidEntries)
+        );
+        invalid = base;
+        invalid.entries[0].bucket_incarnation = vec![0; 16].into();
+        assert_eq!(
+            validate_scoped_dirty_usage_request(&invalid),
+            Err(ScopedDirtyUsageRequestError::InvalidIdentity)
+        );
+    }
+}

crates/scanner/src/lib.rs

@@ -90,8 +90,9 @@ pub use scanner::{
 };
 pub use scanner_io::{
     ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState,
-    acknowledge_dirty_usage_generation, clear_dirty_usage_bucket, record_dirty_usage_bucket, record_scanner_maintenance_change,
-    scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state, scanner_maintenance_generation,
+    acknowledge_dirty_usage_generation, acknowledge_scoped_dirty_usage, clear_dirty_usage_bucket, record_dirty_usage_bucket,
+    record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state,
+    scanner_maintenance_generation,
 };
 pub use sleeper::{DynamicSleeper, SCANNER_IDLE_MODE, SCANNER_SLEEPER};
 use std::sync::atomic::{AtomicU64, Ordering};

crates/scanner/src/scanner.rs

@@ -1703,14 +1703,18 @@ where
     let (sender, receiver) = mpsc::channel::<DataUsageInfo>(1);
 
     let done_cycle = Metrics::time(Metric::ScanCycle);
-    let scan_result = crate::scanner_io::nsscanner_with_storage_status(
+    let scan_result = crate::scanner_io::nsscanner_with_storage_status_scoped(
         storeapi.as_ref(),
-        cycle_budget.token(),
-        cycle_budget.clone(),
-        sender,
-        cycle_info.current,
-        leader_epoch,
-        scan_mode,
+        crate::scanner_io::ScannerCycleRequest {
+            ctx: cycle_budget.token(),
+            budget: cycle_budget.clone(),
+            updates: sender,
+            want_cycle: cycle_info.current,
+            leader_epoch,
+            scan_mode,
+            scan_scope: crate::scanner_io::ScannerBucketScanScope::default(),
+            persisted_usage_baseline: usage_persist_baseline.data.clone(),
+        },
     )
     .await;
     let publication_defer_reason = match &scan_result {
@@ -3424,10 +3428,13 @@ use cycle_state::*;
 use leadership::*;
 use usage_store::*;
 
+#[cfg(test)]
+pub(crate) use activity::scanner_activity_snapshot_digest;
 pub use activity::scanner_topology_digest;
 pub(crate) use activity::{
     ScannerActivitySnapshot, ScannerDirtyUsageAcknowledgement, probe_scanner_activity, scanner_activity_allows_usage_publication,
-    scanner_activity_publication_lease_targets, scanner_activity_snapshot_digest, scanner_dirty_usage_acknowledgements,
+    scanner_activity_dirty_usage_state_for_host, scanner_activity_publication_lease_targets, scanner_activity_structural_digest,
+    scanner_dirty_usage_acknowledgements,
 };
 pub(crate) use activity::{ScannerCycleOutcome, scanner_cycle_outcome_with_pending_maintenance};
 pub use backlog::{

crates/scanner/src/scanner/activity.rs

@@ -902,6 +902,7 @@ where
     observation
 }
 
+#[cfg(test)]
 pub(crate) fn scanner_activity_snapshot_digest(snapshot: &ScannerActivitySnapshot) -> [u8; 32] {
     let mut hasher = Sha256::new();
     hasher.update(u64::try_from(snapshot.len()).unwrap_or(u64::MAX).to_be_bytes());
@@ -925,6 +926,30 @@ pub(crate) fn scanner_activity_snapshot_digest(snapshot: &ScannerActivitySnapsho
     hasher.finalize().into()
 }
 
+/// Hash the activity inputs that make an existing scanner cache unsafe to
+/// reuse. Regular namespace writes and dirty-usage generations are omitted:
+/// their affected buckets are tracked separately and may be refreshed from a
+/// complete authoritative cache baseline.
+pub(crate) fn scanner_activity_structural_digest(snapshot: &ScannerActivitySnapshot) -> [u8; 32] {
+    let mut hasher = Sha256::new();
+    hasher.update(u64::try_from(snapshot.len()).unwrap_or(u64::MAX).to_be_bytes());
+    for (host, activity) in snapshot {
+        let host = host.as_bytes();
+        let instance_id = activity.instance_id.as_bytes();
+        hasher.update(u64::try_from(host.len()).unwrap_or(u64::MAX).to_be_bytes());
+        hasher.update(host);
+        hasher.update(u64::try_from(instance_id.len()).unwrap_or(u64::MAX).to_be_bytes());
+        hasher.update(instance_id);
+        hasher.update(activity.maintenance_generation.to_be_bytes());
+        hasher.update(activity.protocol_version.to_be_bytes());
+        hasher.update(activity.topology_digest);
+        hasher.update([u8::from(activity.data_movement_active)]);
+        hasher.update(activity.movement_generation.to_be_bytes());
+        hasher.update([u8::from(activity.publication_blocked)]);
+    }
+    hasher.finalize().into()
+}
+
 pub(crate) fn scanner_activity_allows_usage_publication(snapshot: &ScannerActivitySnapshot) -> bool {
     !snapshot.is_empty()
         && snapshot.values().all(|activity| {
@@ -955,6 +980,22 @@ pub(crate) fn scanner_dirty_usage_acknowledgements(snapshot: &ScannerActivitySna
         .collect()
 }
 
+pub(crate) fn scanner_activity_dirty_usage_state_for_host<'a>(
+    snapshot: &'a ScannerActivitySnapshot,
+    host: &str,
+) -> Option<(&'a str, u64, bool)> {
+    snapshot
+        .get(host)
+        .filter(|_| host != LOCAL_SCANNER_ACTIVITY_NODE)
+        .map(|activity| {
+            (
+                activity.instance_id.as_str(),
+                activity.dirty_usage_generation,
+                activity.dirty_usage_pending,
+            )
+        })
+}
+
 pub fn scanner_topology_digest(storeapi: &ECStore) -> [u8; 32] {
     let endpoint_pools = storeapi.endpoints();
     let mut hasher = Sha256::new();

crates/scanner/src/scanner/cycle_state.rs

@@ -379,6 +379,12 @@ pub(super) fn decode_recovery_marker_for_reset(
     if !matches!(marker_revision, DataUsageCacheRevision::Etag(_)) {
         return Err(ScannerError::Other("cycle recovery marker has no object revision".to_string()));
     }
+    if let Ok(value) = serde_json::from_slice::<serde_json::Value>(data)
+        && let Some(state) = value.get("state")
+        && !matches!(state.as_str(), Some("blocked" | "cleanup-pending"))
+    {
+        return Err(ScannerError::Other("cycle recovery marker state is unsupported".to_string()));
+    }
     let compat = serde_json::from_slice::<ScannerCycleRecoveryMarkerCompat>(data).ok();
     let _schema_version = compat.as_ref().and_then(|marker| marker.schema_version);
     let primary_revision = compat
@@ -406,7 +412,10 @@ pub(super) fn decode_recovery_marker_for_reset(
     };
     let state = match compat.as_ref().and_then(|marker| marker.state.as_deref()) {
         Some("cleanup-pending") => "cleanup-pending",
-        _ => "blocked",
+        Some("blocked") | None => "blocked",
+        Some(_) => {
+            return Err(ScannerError::Other("cycle recovery marker state is unsupported".to_string()));
+        }
     };
     let now = unix_now_secs();
     Ok(ScannerCycleRecoveryMarker {
@@ -721,17 +730,19 @@ async fn mark_cycle_recovery_cleanup_pending(
     mut marker: ScannerCycleRecoveryMarker,
     marker_revision: &DataUsageCacheRevision,
     expected_epoch: u64,
+    owns_reset: &(impl Fn() -> bool + Sync),
 ) -> Result<(ScannerCycleRecoveryMarker, DataUsageCacheRevision), ScannerError> {
     marker.state = "cleanup-pending".to_string();
     marker.last_attempt_at_unix_secs = unix_now_secs();
     let bytes = serde_json::to_vec(&marker)
         .map_err(|err| ScannerError::Other(format!("failed to encode cycle recovery marker: {err}")))?;
-    let info = save_config_with_publication_admission_for_epoch(
+    let info = save_reset_config(
         storeapi.clone(),
         DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
         bytes,
         marker_revision.preconditions(),
         expected_epoch,
+        owns_reset,
     )
     .await
     .map_err(|err| ScannerError::Other(format!("failed to mark cycle recovery cleanup pending: {err}")))?;
@@ -933,6 +944,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
         .get_write_lock_quiet(Duration::from_secs(5))
         .await
         .map_err(|err| ScannerError::Other(format!("scanner leader lock is busy: {err}")))?;
+    let owns_reset = || !guard.is_lock_lost() && !ctx.is_cancelled();
 
     if guard.is_lock_lost() {
         return Err(ScannerError::Other("scanner leader lock was lost before recovery reset".to_string()));
@@ -952,7 +964,27 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
         }
         Err(err) => return Err(ScannerError::Other(format!("failed to read cycle recovery marker: {err}"))),
     };
-    let marker_data = marker_data.ok_or_else(|| ScannerError::Other("scanner cycle recovery marker is absent".to_string()))?;
+    let Some(marker_data) = marker_data else {
+        // A delete may commit before its reply is lost. Confirm both durable
+        // fences before treating a retry without its marker as completed.
+        let (cycle, epoch, revision) = read_cycle_state_for_usage_reset(storeapi.clone()).await?;
+        let floor = persisted_usage_floor(storeapi.clone()).await?;
+        if !matches!(revision, DataUsageCacheRevision::Etag(_))
+            || epoch < floor.leader_epoch
+            || cycle.next < floor.next_cycle
+            || !owns_reset()
+            || scanner_publication_admission_for_epoch(storeapi.clone(), reset_epoch)
+                .await
+                .is_none()
+        {
+            return Err(ScannerError::Other(
+                "scanner cycle recovery marker is absent without a completed reset fence".to_string(),
+            ));
+        }
+        set_scanner_cycle_recovery_status(recovery_status("healthy", None, false));
+        super::notify_scanner_cycle_recovery_wake();
+        return Ok(());
+    };
     let (marker, force_full_rescan) = match serde_json::from_slice::<ScannerCycleRecoveryMarker>(&marker_data) {
         Ok(marker) if validate_recovery_marker(&marker).is_ok() => (marker, false),
         _ => (decode_recovery_marker_for_reset(&marker_data, &marker_revision)?, true),
@@ -1026,8 +1058,10 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
             }
         };
         if let Some((primary_cycle, primary_epoch)) = primary_state {
+            verify_cycle_reset_intent(storeapi.clone(), &marker_revision, &owns_reset).await?;
             let (cleanup_marker, cleanup_marker_revision) =
-                mark_cycle_recovery_cleanup_pending(storeapi.clone(), marker.clone(), &marker_revision, reset_epoch).await?;
+                mark_cycle_recovery_cleanup_pending(storeapi.clone(), marker.clone(), &marker_revision, reset_epoch, &owns_reset)
+                    .await?;
             set_scanner_cycle_recovery_status(recovery_status_from_marker(&cleanup_marker, "cleanup-pending"));
             let usage_floor = persisted_usage_floor(storeapi.clone()).await?;
             let fence_epoch = primary_epoch
@@ -1047,12 +1081,14 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
                     "preserved scanner cycle state exceeds the bounded object size".to_string(),
                 ));
             }
-            let preserved_info = save_config_with_publication_admission_for_epoch(
+            verify_cycle_reset_intent(storeapi.clone(), &cleanup_marker_revision, &owns_reset).await?;
+            let preserved_info = save_reset_config(
                 storeapi.clone(),
                 DATA_USAGE_BLOOM_NAME_PATH.as_str(),
                 preserved_data,
                 primary_revision.preconditions(),
                 reset_epoch,
+                &owns_reset,
             )
             .await
             .map_err(|err| {
@@ -1072,9 +1108,17 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
                     "scanner leader lock was lost after fencing newer cycle state".to_string(),
                 ));
             }
-            fence_scanner_usage_epoch_with_expected_epoch(&ctx, storeapi.clone(), fence_epoch, Some(reset_epoch), false)
-                .await
-                .map_err(|err| ScannerError::Other(format!("failed to fence preserved scanner usage epoch: {err}")))?;
+            verify_cycle_reset_intent(storeapi.clone(), &cleanup_marker_revision, &owns_reset).await?;
+            fence_scanner_usage_epoch_with_expected_epoch(
+                &ctx,
+                storeapi.clone(),
+                fence_epoch,
+                Some(reset_epoch),
+                false,
+                &owns_reset,
+            )
+            .await
+            .map_err(|err| ScannerError::Other(format!("failed to fence preserved scanner usage epoch: {err}")))?;
             if guard.is_lock_lost() {
                 return Err(ScannerError::Other(
                     "scanner leader lock was lost after fencing newer cycle state".to_string(),
@@ -1088,7 +1132,8 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
                     "scanner cycle state changed before recovery marker cleanup".to_string(),
                 ));
             }
-            delete_config_with_publication_admission_for_epoch(
+            verify_cycle_reset_intent(storeapi.clone(), &cleanup_marker_revision, &owns_reset).await?;
+            delete_reset_config(
                 storeapi.clone(),
                 RUSTFS_META_BUCKET,
                 DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
@@ -1100,6 +1145,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
                     ..Default::default()
                 },
                 reset_epoch,
+                &owns_reset,
             )
             .await
             .map_err(|err| {
@@ -1149,17 +1195,20 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
     // Persist the cleanup-pending phase before rewriting the primary. If the
     // process dies after the rewrite, startup still sees a durable fence and
     // cannot mistake the partially completed reset for a healthy state.
+    verify_cycle_reset_intent(storeapi.clone(), &marker_revision, &owns_reset).await?;
     let (marker, marker_revision) = if marker.state == "cleanup-pending" {
         (marker, marker_revision)
     } else {
-        mark_cycle_recovery_cleanup_pending(storeapi.clone(), marker, &marker_revision, reset_epoch).await?
+        mark_cycle_recovery_cleanup_pending(storeapi.clone(), marker, &marker_revision, reset_epoch, &owns_reset).await?
     };
-    let rebuilt_info = save_config_with_publication_admission_for_epoch(
+    verify_cycle_reset_intent(storeapi.clone(), &marker_revision, &owns_reset).await?;
+    let rebuilt_info = save_reset_config(
         storeapi.clone(),
         DATA_USAGE_BLOOM_NAME_PATH.as_str(),
         data,
         primary_revision.preconditions(),
         reset_epoch,
+        &owns_reset,
     )
     .await
     .map_err(|err| {
@@ -1178,8 +1227,10 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
             "scanner leader lock was lost after rebuilding cycle state".to_string(),
         ));
     }
+    verify_cycle_reset_intent(storeapi.clone(), &marker_revision, &owns_reset).await?;
     if let Err(err) =
-        fence_scanner_usage_epoch_with_expected_epoch(&ctx, storeapi.clone(), leader_epoch, Some(reset_epoch), false).await
+        fence_scanner_usage_epoch_with_expected_epoch(&ctx, storeapi.clone(), leader_epoch, Some(reset_epoch), false, &owns_reset)
+            .await
     {
         set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus {
             path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
@@ -1249,7 +1300,8 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
         ));
     }
 
-    if let Err(err) = delete_config_with_publication_admission_for_epoch(
+    verify_cycle_reset_intent(storeapi.clone(), &marker_revision, &owns_reset).await?;
+    if let Err(err) = delete_reset_config(
         storeapi.clone(),
         RUSTFS_META_BUCKET,
         DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
@@ -1261,6 +1313,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
             ..Default::default()
         },
         reset_epoch,
+        &owns_reset,
     )
     .await
     {
@@ -1310,6 +1363,57 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
     Ok(())
 }
 
+async fn verify_cycle_reset_intent(
+    storeapi: Arc<impl ScannerObjectIO>,
+    expected_revision: &DataUsageCacheRevision,
+    owns_reset: &(impl Fn() -> bool + Sync),
+) -> Result<(), ScannerError> {
+    let revision = read_config_revision(storeapi, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
+        .await
+        .map_err(|err| ScannerError::Other(format!("failed to verify scanner cycle reset intent: {err}")))?;
+    if &revision != expected_revision {
+        return Err(ScannerError::Other("scanner cycle reset intent changed".to_string()));
+    }
+    if !owns_reset() {
+        return Err(ScannerError::Other("scanner cycle reset ownership was lost".to_string()));
+    }
+    Ok(())
+}
+
+async fn save_reset_config(
+    storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
+    path: &str,
+    data: Vec<u8>,
+    preconditions: crate::HTTPPreconditions,
+    expected_epoch: u64,
+    owns_reset: &(impl Fn() -> bool + Sync),
+) -> Result<crate::ScannerObjectInfo, EcstoreError> {
+    let Some(_admission) = scanner_publication_admission_for_epoch(storeapi.clone(), expected_epoch).await else {
+        return Err(EcstoreError::other(SCANNER_PUBLICATION_EPOCH_CHANGED));
+    };
+    if !owns_reset() {
+        return Err(EcstoreError::other("scanner reset ownership was lost before write"));
+    }
+    save_config_with_preconditions(storeapi, path, data, preconditions).await
+}
+
+async fn delete_reset_config(
+    storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
+    bucket: &str,
+    path: &str,
+    options: ScannerObjectOptions,
+    expected_epoch: u64,
+    owns_reset: &(impl Fn() -> bool + Sync),
+) -> Result<crate::ScannerObjectInfo, EcstoreError> {
+    let Some(_admission) = scanner_publication_admission_for_epoch(storeapi.clone(), expected_epoch).await else {
+        return Err(EcstoreError::other(SCANNER_PUBLICATION_EPOCH_CHANGED));
+    };
+    if !owns_reset() {
+        return Err(EcstoreError::other("scanner reset ownership was lost before delete"));
+    }
+    storeapi.delete_config_object(bucket, path, options).await
+}
+
 fn scanner_usage_state_reset_paths() -> Vec<String> {
     vec![
         DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
@@ -1333,8 +1437,14 @@ pub(super) async fn read_usage_state_reset_slots(
     Ok(slots)
 }
 
-fn usage_state_reset_floor(slots: &[ScannerUsageStateResetSlot]) -> Result<PersistedUsageFloor, ScannerError> {
-    let mut floor = PersistedUsageFloor::default();
+enum ScannerUsageResetFloor {
+    Missing,
+    Trusted(PersistedUsageFloor),
+    Corrupt,
+}
+
+fn usage_state_reset_floor(slots: &[ScannerUsageStateResetSlot]) -> Result<ScannerUsageResetFloor, ScannerError> {
+    let mut floor = None;
     for slot in slots {
         let Some(data) = slot.data.as_deref() else {
             continue;
@@ -1342,9 +1452,21 @@ fn usage_state_reset_floor(slots: &[ScannerUsageStateResetSlot]) -> Result<Persi
         let Ok(usage) = serde_json::from_slice::<DataUsageInfo>(data) else {
             continue;
         };
-        update_persisted_usage_floor(&mut floor, &usage, &slot.path)?;
+        if !data_usage_info_has_persisted_baseline_identity(&usage)
+            && !(slot.path == DATA_USAGE_OBJ_NAME_PATH.as_str() && data_usage_info_is_bootstrap_pending(&usage))
+            && legacy_incomplete_usage_fence(data, &usage)
+                .and_then(|fence| fence.claimable_epoch())
+                .is_none()
+        {
+            continue;
+        }
+        update_persisted_usage_floor(floor.get_or_insert_with(PersistedUsageFloor::default), &usage, &slot.path)?;
     }
-    Ok(floor)
+    Ok(match floor {
+        Some(floor) => ScannerUsageResetFloor::Trusted(floor),
+        None if slots.iter().any(|slot| slot.data.is_some()) => ScannerUsageResetFloor::Corrupt,
+        None => ScannerUsageResetFloor::Missing,
+    })
 }
 
 async fn read_cycle_state_for_usage_reset(
@@ -1401,11 +1523,12 @@ async fn delete_usage_state_reset_slot(
     storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
     slot: &ScannerUsageStateResetSlot,
     expected_epoch: u64,
+    owns_reset: &(impl Fn() -> bool + Sync),
 ) -> Result<bool, ScannerError> {
     if matches!(slot.revision, DataUsageCacheRevision::Missing) {
         return Ok(false);
     }
-    let delete_result = delete_config_with_publication_admission_for_epoch(
+    let delete_result = delete_reset_config(
         storeapi.clone(),
         RUSTFS_META_BUCKET,
         &slot.path,
@@ -1415,6 +1538,7 @@ async fn delete_usage_state_reset_slot(
             ..Default::default()
         },
         expected_epoch,
+        owns_reset,
     )
     .await;
     match delete_result {
@@ -1486,21 +1610,24 @@ pub(super) async fn publish_scanner_usage_bootstrap_primary(
     expected_publication_epoch: u64,
     leader_epoch: Option<u64>,
     context: ScannerUsageBootstrapPublishContext,
+    owns_publication: impl Fn() -> bool + Sync,
 ) -> Result<(), ScannerError> {
     async fn inner(
         storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
         expected_revision: &DataUsageCacheRevision,
         expected_publication_epoch: u64,
         leader_epoch: Option<u64>,
+        owns_publication: &(impl Fn() -> bool + Sync),
     ) -> Result<(), ScannerUsageBootstrapPublishError> {
         let marker = scanner_usage_bootstrap_marker(std::time::SystemTime::now(), leader_epoch);
         let data = serde_json::to_vec(&marker).map_err(ScannerUsageBootstrapPublishError::Encode)?;
-        let save_result = save_config_with_publication_admission_for_epoch(
+        let save_result = save_reset_config(
             storeapi.clone(),
             DATA_USAGE_OBJ_NAME_PATH.as_str(),
             data.clone(),
             expected_revision.preconditions(),
             expected_publication_epoch,
+            owns_publication,
         )
         .await;
         if save_result
@@ -1524,7 +1651,7 @@ pub(super) async fn publish_scanner_usage_bootstrap_primary(
         })
     }
 
-    inner(storeapi, expected_revision, expected_publication_epoch, leader_epoch)
+    inner(storeapi, expected_revision, expected_publication_epoch, leader_epoch, &owns_publication)
         .await
         .map_err(|err| err.into_scanner_error(context))
 }
@@ -1534,32 +1661,108 @@ pub(super) async fn reset_scanner_usage_state_slots_for_full_rebuild(
     slots: &[ScannerUsageStateResetSlot],
     expected_epoch: u64,
     leader_epoch: u64,
+    owns_reset: impl Fn() -> bool + Sync,
 ) -> Result<Vec<String>, ScannerError> {
     let mut reset_paths = Vec::new();
     let primary = slots
         .iter()
         .find(|slot| slot.path == DATA_USAGE_OBJ_NAME_PATH.as_str())
         .ok_or_else(|| ScannerError::Other("scanner usage reset primary slot was not inspected".to_string()))?;
-    publish_scanner_usage_bootstrap_primary(
-        storeapi.clone(),
-        &primary.revision,
-        expected_epoch,
-        Some(leader_epoch),
-        ScannerUsageBootstrapPublishContext::Reset,
-    )
-    .await?;
+    if !owns_reset() {
+        return Err(ScannerError::Other("scanner usage reset ownership was lost".to_string()));
+    }
+    let resume_epoch = usage_state_reset_resume_epoch(slots)?;
+    match resume_epoch {
+        Some(epoch) if epoch == leader_epoch => {}
+        Some(_) => return Err(ScannerError::Other("scanner usage reset bootstrap epoch changed".to_string())),
+        None => {
+            publish_scanner_usage_bootstrap_primary(
+                storeapi.clone(),
+                &primary.revision,
+                expected_epoch,
+                Some(leader_epoch),
+                ScannerUsageBootstrapPublishContext::Reset,
+                &owns_reset,
+            )
+            .await?;
+        }
+    }
+    let (data, intent_revision) = read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+        .await
+        .map_err(|err| ScannerError::Other(format!("failed to inspect scanner usage reset intent: {err}")))?;
+    data.as_deref()
+        .and_then(|data| serde_json::from_slice::<DataUsageInfo>(data).ok())
+        .filter(|usage| data_usage_info_is_bootstrap_pending(usage) && usage.scanner_epoch == Some(leader_epoch))
+        .ok_or_else(|| ScannerError::Other("scanner usage reset intent changed before cleanup".to_string()))?;
+    if !matches!(intent_revision, DataUsageCacheRevision::Etag(_))
+        || (resume_epoch.is_some() && intent_revision != primary.revision)
+    {
+        return Err(ScannerError::Other("scanner usage reset intent revision changed".to_string()));
+    }
     reset_paths.push(DATA_USAGE_OBJ_NAME_PATH.as_str().to_string());
 
     for slot in slots.iter().filter(|slot| slot.path != DATA_USAGE_OBJ_NAME_PATH.as_str()) {
-        if delete_usage_state_reset_slot(storeapi.clone(), slot, expected_epoch).await? {
+        if let Some(usage) = slot
+            .data
+            .as_deref()
+            .and_then(|data| serde_json::from_slice::<DataUsageInfo>(data).ok())
+            && usage_epoch(&usage) >= leader_epoch
+        {
+            return Err(ScannerError::Other(format!(
+                "scanner usage reset slot is not older than its intent: {}",
+                slot.path
+            )));
+        }
+        let revision = read_config_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+            .await
+            .map_err(|err| ScannerError::Other(format!("failed to verify scanner usage reset intent: {err}")))?;
+        if revision != intent_revision {
+            return Err(ScannerError::Other("scanner usage reset intent changed during cleanup".to_string()));
+        }
+        if !owns_reset() {
+            return Err(ScannerError::Other("scanner usage reset ownership was lost".to_string()));
+        }
+        if delete_usage_state_reset_slot(storeapi.clone(), slot, expected_epoch, &owns_reset).await? {
             reset_paths.push(slot.path.clone());
         }
     }
+    let revision = read_config_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+        .await
+        .map_err(|err| ScannerError::Other(format!("failed to confirm scanner usage reset intent: {err}")))?;
+    if revision != intent_revision || !owns_reset() {
+        return Err(ScannerError::Other(
+            "scanner usage reset intent or ownership changed before completion".to_string(),
+        ));
+    }
     invalidate_admin_data_usage_snapshot_cache().await;
     invalidate_data_usage_snapshot_cache().await;
     Ok(reset_paths)
 }
 
+fn usage_state_reset_resume_epoch(slots: &[ScannerUsageStateResetSlot]) -> Result<Option<u64>, ScannerError> {
+    let primary = slots.iter().find(|slot| slot.path == DATA_USAGE_OBJ_NAME_PATH.as_str());
+    let usage = primary
+        .and_then(|slot| slot.data.as_deref())
+        .and_then(|data| serde_json::from_slice::<DataUsageInfo>(data).ok());
+    match usage {
+        Some(usage) if usage.usage_snapshot_bootstrap_pending => {
+            if !data_usage_info_is_bootstrap_pending(&usage) {
+                return Err(ScannerError::Other("scanner usage reset bootstrap is invalid".to_string()));
+            }
+            if usage.scanner_epoch.is_none() {
+                // Initial bootstrap has no reset owner yet.
+                return Ok(None);
+            }
+            usage
+                .scanner_epoch
+                .filter(|epoch| *epoch > 0 && *epoch < u64::MAX)
+                .map(Some)
+                .ok_or_else(|| ScannerError::Other("scanner usage reset bootstrap has no valid epoch".to_string()))
+        }
+        _ => Ok(None),
+    }
+}
+
 pub async fn reset_scanner_usage_state_for_full_rebuild(
     ctx: CancellationToken,
     storeapi: Arc<ECStore>,
@@ -1584,12 +1787,31 @@ pub async fn reset_scanner_usage_state_for_full_rebuild(
     };
     let (cycle, cycle_epoch, cycle_revision) = read_cycle_state_for_usage_reset(storeapi.clone()).await?;
     let slots = read_usage_state_reset_slots(storeapi.clone()).await?;
-    let usage_floor = usage_state_reset_floor(&slots)?;
-    let leader_epoch = cycle_epoch
-        .max(usage_floor.leader_epoch)
-        .checked_add(1)
-        .filter(|epoch| *epoch < u64::MAX)
-        .ok_or_else(|| ScannerError::Other("scanner leader epoch is exhausted".to_string()))?;
+    let usage_floor = match usage_state_reset_floor(&slots)? {
+        ScannerUsageResetFloor::Trusted(floor) => floor,
+        ScannerUsageResetFloor::Corrupt if matches!(cycle_revision, DataUsageCacheRevision::Missing) => {
+            return Err(ScannerError::Other("scanner usage reset has no trusted cycle or usage floor".to_string()));
+        }
+        ScannerUsageResetFloor::Missing | ScannerUsageResetFloor::Corrupt => PersistedUsageFloor {
+            next_cycle: cycle.next,
+            leader_epoch: cycle_epoch,
+        },
+    };
+    let resume_epoch = usage_state_reset_resume_epoch(&slots)?;
+    let leader_epoch = if let Some(epoch) = resume_epoch {
+        if epoch != cycle_epoch || usage_floor.leader_epoch > epoch || usage_floor.next_cycle > cycle.next {
+            return Err(ScannerError::Other(
+                "scanner usage reset bootstrap conflicts with the persisted cycle fence".to_string(),
+            ));
+        }
+        epoch
+    } else {
+        cycle_epoch
+            .max(usage_floor.leader_epoch)
+            .checked_add(1)
+            .filter(|epoch| *epoch < u64::MAX)
+            .ok_or_else(|| ScannerError::Other("scanner leader epoch is exhausted".to_string()))?
+    };
     let rebuilt_cycle = CurrentCycle {
         next: cycle.next.max(usage_floor.next_cycle),
         ..Default::default()
@@ -1602,21 +1824,24 @@ pub async fn reset_scanner_usage_state_for_full_rebuild(
             "scanner leader lock was lost before fencing usage reset cycle state".to_string(),
         ));
     }
-    save_config_with_publication_admission_for_epoch(
-        storeapi.clone(),
-        DATA_USAGE_BLOOM_NAME_PATH.as_str(),
-        cycle_data,
-        cycle_revision.preconditions(),
-        reset_epoch,
-    )
-    .await
-    .map_err(|err| {
-        if scanner_publication_epoch_changed(&err) {
-            ScannerError::Other("scanner usage reset deferred by a movement epoch change".to_string())
-        } else {
-            ScannerError::Other(format!("failed to fence scanner cycle state for usage reset: {err}"))
-        }
-    })?;
+    if resume_epoch.is_none() {
+        save_reset_config(
+            storeapi.clone(),
+            DATA_USAGE_BLOOM_NAME_PATH.as_str(),
+            cycle_data,
+            cycle_revision.preconditions(),
+            reset_epoch,
+            &|| !guard.is_lock_lost() && !ctx.is_cancelled(),
+        )
+        .await
+        .map_err(|err| {
+            if scanner_publication_epoch_changed(&err) {
+                ScannerError::Other("scanner usage reset deferred by a movement epoch change".to_string())
+            } else {
+                ScannerError::Other(format!("failed to fence scanner cycle state for usage reset: {err}"))
+            }
+        })?;
+    }
 
     if guard.is_lock_lost() {
         return Err(ScannerError::Other(
@@ -1624,7 +1849,10 @@ pub async fn reset_scanner_usage_state_for_full_rebuild(
         ));
     }
     let reset_paths =
-        reset_scanner_usage_state_slots_for_full_rebuild(storeapi.clone(), &slots, reset_epoch, leader_epoch).await?;
+        reset_scanner_usage_state_slots_for_full_rebuild(storeapi.clone(), &slots, reset_epoch, leader_epoch, || {
+            !guard.is_lock_lost() && !ctx.is_cancelled()
+        })
+        .await?;
     if guard.is_lock_lost() {
         return Err(ScannerError::Other(
             "scanner leader lock was lost after publishing usage reset marker".to_string(),
@@ -2135,6 +2363,7 @@ async fn recover_legacy_incomplete_usage_floor(
         expected_publication_epoch,
         Some(primary.epoch),
         ScannerUsageBootstrapPublishContext::Recovery,
+        || true,
     )
     .await?;
     warn!(

crates/scanner/src/scanner/leadership.rs

@@ -191,6 +191,7 @@ pub(super) async fn initialize_usage_baseline_bootstrap(
         expected_epoch,
         None,
         ScannerUsageBootstrapPublishContext::Initial,
+        || true,
     )
     .await
 }
@@ -201,9 +202,10 @@ pub(super) async fn fence_scanner_usage_epoch_with_expected_epoch(
     claimed_epoch: u64,
     expected_publication_epoch: Option<u64>,
     allow_bootstrap_pending: bool,
+    owns_fence: impl Fn() -> bool,
 ) -> Result<(), ScannerError> {
     for retry in 0..=SCANNER_PERSIST_CAS_RETRIES {
-        if ctx.is_cancelled() {
+        if ctx.is_cancelled() || !owns_fence() {
             return Err(ScannerError::Other("scanner leadership was cancelled before usage fencing".to_string()));
         }
 
@@ -264,6 +266,9 @@ pub(super) async fn fence_scanner_usage_epoch_with_expected_epoch(
                     "scanner usage epoch fence changed while preparing its conditional write".to_string(),
                 ));
             };
+            if ctx.is_cancelled() || !owns_fence() {
+                return Err(ScannerError::Other("scanner leadership was lost before usage fencing".to_string()));
+            }
             save_config_with_preconditions(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), data, revision.preconditions())
                 .await
         };
@@ -319,6 +324,7 @@ pub(super) async fn complete_scanner_leadership_claim(
         claimed_epoch,
         expected_publication_epoch,
         allow_bootstrap_pending,
+        || true,
     )
     .await
     {

crates/scanner/src/scanner/tests.rs

@@ -634,6 +634,8 @@ struct MemoryConfigStore {
     cancel_after_successful_puts: Mutex<HashMap<String, (usize, CancellationToken)>>,
     replace_after_successful_puts: Mutex<HashMap<String, (usize, Vec<u8>)>>,
     error_after_commit_deletes: Mutex<HashSet<String>>,
+    cancel_after_deletes: Mutex<HashMap<String, CancellationToken>>,
+    pause_next_publication_admission: Mutex<Option<(Arc<tokio::sync::Notify>, Arc<tokio::sync::Notify>)>>,
     put_counts: Mutex<HashMap<String, usize>>,
     publication_admission_blocked: AtomicBool,
     block_publication_after_admissions: AtomicUsize,
@@ -4081,6 +4083,9 @@ impl crate::ScannerConfigObjectDelete for MemoryConfigStore {
         revisions.remove(&key);
         drop(revisions);
         drop(objects);
+        if let Some(token) = self.cancel_after_deletes.lock().await.remove(&key) {
+            token.cancel();
+        }
         if self.error_after_commit_deletes.lock().await.remove(&key) {
             return Err(EcstoreError::other("injected delete error after commit"));
         }
@@ -4088,6 +4093,11 @@ impl crate::ScannerConfigObjectDelete for MemoryConfigStore {
     }
 
     async fn scanner_data_usage_publication_admission(&self) -> Option<crate::ScannerDataUsagePublicationAdmission> {
+        let pause = self.pause_next_publication_admission.lock().await.take();
+        if let Some((entered, resume)) = pause {
+            entered.notify_one();
+            resume.notified().await;
+        }
         if self.publication_admission_blocked.load(Ordering::Acquire) {
             return None;
         }
@@ -4589,7 +4599,7 @@ async fn scanner_legacy_usage_backup_survives_fencing_and_restart_after_real_met
         .expect("publication must also read the intact backup");
     assert_eq!(baseline.data.as_deref(), Some(data.as_slice()));
     assert_eq!(baseline.revision, DataUsageCacheRevision::Missing);
-    fence_scanner_usage_epoch_with_expected_epoch(&CancellationToken::new(), store.clone(), 7, None, false)
+    fence_scanner_usage_epoch_with_expected_epoch(&CancellationToken::new(), store.clone(), 7, None, false, || true)
         .await
         .expect("legacy backup must be fenced into v2");
     let fenced = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
@@ -4818,6 +4828,28 @@ async fn scanner_usage_state_reset_publishes_fenced_bootstrap_marker() {
         );
     }
 
+    let cycle_before_retry = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
+        .await
+        .expect("cycle should remain before retry");
+    let marker_before_retry = read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+        .await
+        .expect("bootstrap should remain before retry");
+    let retry = reset_scanner_usage_state_for_full_rebuild(CancellationToken::new(), store.clone())
+        .await
+        .expect("completed cleanup should be reentrant");
+    assert_eq!(retry.leader_epoch, result.leader_epoch);
+    assert_eq!(
+        read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
+            .await
+            .expect("cycle should remain"),
+        cycle_before_retry
+    );
+    assert_eq!(
+        read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+            .await
+            .expect("bootstrap should remain"),
+        marker_before_retry
+    );
     let (floor, state) = persisted_usage_floor_for_startup(store, false)
         .await
         .expect("reset marker should be resumable");
@@ -4973,7 +5005,7 @@ async fn scanner_usage_state_reset_slots_reject_primary_aba() {
 
     store.objects.lock().await.insert(key.clone(), b"newer-json".to_vec());
     store.revisions.lock().await.insert(key, 2);
-    let err = reset_scanner_usage_state_slots_for_full_rebuild(store, &slots, 0, 3)
+    let err = reset_scanner_usage_state_slots_for_full_rebuild(store, &slots, 0, 3, || true)
         .await
         .expect_err("stale primary revision must not be overwritten");
     assert!(
@@ -4983,6 +5015,348 @@ async fn scanner_usage_state_reset_slots_reject_primary_aba() {
     );
 }
 
+#[tokio::test]
+async fn scanner_usage_state_reset_resumes_every_cleanup_boundary_without_rewriting_intent() {
+    for completed in 0..=4 {
+        let store = Arc::new(MemoryConfigStore::default());
+        let primary_path = DATA_USAGE_OBJ_NAME_PATH.as_str();
+        let cleanup_paths = [
+            format!("{primary_path}.bkp"),
+            LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
+            format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
+            DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str().to_string(),
+        ];
+        for path in std::iter::once(primary_path).chain(cleanup_paths.iter().map(String::as_str)) {
+            let mut usage = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+            usage.scanner_epoch = Some(1);
+            save_config(store.clone(), path, serde_json::to_vec(&usage).expect("fixture should encode"))
+                .await
+                .expect("fixture should persist");
+        }
+        // These objects belong to other owners, even when reset cleanup resumes.
+        for path in ["buckets/quota-reservations/ledger", "buckets/example/incarnation"] {
+            save_config(store.clone(), path, b"retain".to_vec())
+                .await
+                .expect("unrelated state should persist");
+        }
+        let slots = read_usage_state_reset_slots(store.clone()).await.expect("slots should load");
+        let cancelled = CancellationToken::new();
+        if completed == 0 {
+            store
+                .cancel_after_successful_puts
+                .lock()
+                .await
+                .insert(memory_config_key(RUSTFS_META_BUCKET, primary_path), (2, cancelled.clone()));
+        } else {
+            store
+                .cancel_after_deletes
+                .lock()
+                .await
+                .insert(memory_config_key(RUSTFS_META_BUCKET, &cleanup_paths[completed - 1]), cancelled.clone());
+        }
+        let err = reset_scanner_usage_state_slots_for_full_rebuild(store.clone(), &slots, 0, 3, || !cancelled.is_cancelled())
+            .await
+            .expect_err("interruption should stop cleanup");
+        assert!(err.to_string().contains("ownership"), "boundary {completed}: {err}");
+        for (index, path) in cleanup_paths.iter().enumerate() {
+            assert_eq!(
+                store
+                    .objects
+                    .lock()
+                    .await
+                    .contains_key(&memory_config_key(RUSTFS_META_BUCKET, path)),
+                index >= completed,
+                "boundary {completed}, slot {index}"
+            );
+        }
+        let intent = read_config_with_revision(store.clone(), primary_path)
+            .await
+            .expect("intent should persist");
+        let slots = read_usage_state_reset_slots(store.clone())
+            .await
+            .expect("restart should reload slots");
+        reset_scanner_usage_state_slots_for_full_rebuild(store.clone(), &slots, 0, 3, || true)
+            .await
+            .expect("restart should complete the same intent");
+        assert_eq!(
+            read_config_with_revision(store.clone(), primary_path)
+                .await
+                .expect("intent should remain"),
+            intent
+        );
+        assert_eq!(store.put_counts.lock().await[&memory_config_key(RUSTFS_META_BUCKET, primary_path)], 2);
+        for path in cleanup_paths {
+            assert!(
+                !store
+                    .objects
+                    .lock()
+                    .await
+                    .contains_key(&memory_config_key(RUSTFS_META_BUCKET, &path))
+            );
+        }
+        for path in ["buckets/quota-reservations/ledger", "buckets/example/incarnation"] {
+            assert_eq!(read_config(store.clone(), path).await.expect("unrelated state should remain"), b"retain");
+        }
+    }
+}
+
+#[tokio::test]
+async fn scanner_usage_state_reset_stops_usage_fence_after_owner_loss() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let mut usage = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    usage.scanner_epoch = Some(1);
+    let bytes = serde_json::to_vec(&usage).expect("baseline should encode");
+    save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), bytes.clone())
+        .await
+        .expect("baseline should persist");
+    let checks = AtomicUsize::new(0);
+    let err = fence_scanner_usage_epoch_with_expected_epoch(&CancellationToken::new(), store.clone(), 3, Some(0), false, || {
+        checks.fetch_add(1, Ordering::SeqCst) == 0
+    })
+    .await
+    .expect_err("ownership lost during reads must prevent the write");
+    assert!(err.to_string().contains("leadership was lost"), "{err}");
+    assert_eq!(
+        read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str())
+            .await
+            .expect("baseline should remain"),
+        bytes
+    );
+}
+
+#[tokio::test]
+async fn scanner_usage_state_reset_cancels_during_publication_admission() {
+    for resuming in [false, true] {
+        let store = Arc::new(MemoryConfigStore::default());
+        let usage = if resuming {
+            scanner_usage_bootstrap_marker(std::time::SystemTime::UNIX_EPOCH, Some(3))
+        } else {
+            complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0)
+        };
+        save_config(
+            store.clone(),
+            DATA_USAGE_OBJ_NAME_PATH.as_str(),
+            serde_json::to_vec(&usage).expect("primary should encode"),
+        )
+        .await
+        .expect("primary should persist");
+        save_config(store.clone(), LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str(), b"corrupt".to_vec())
+            .await
+            .expect("cleanup target should persist");
+        let slots = read_usage_state_reset_slots(store.clone()).await.expect("slots should load");
+        let before = store.objects.lock().await.clone();
+        let revisions_before = store.revisions.lock().await.clone();
+        let entered = Arc::new(tokio::sync::Notify::new());
+        let resume = Arc::new(tokio::sync::Notify::new());
+        *store.pause_next_publication_admission.lock().await = Some((entered.clone(), resume.clone()));
+        let cancelled = CancellationToken::new();
+        let (result, ()) = tokio::join!(
+            reset_scanner_usage_state_slots_for_full_rebuild(store.clone(), &slots, 0, 3, || !cancelled.is_cancelled()),
+            async {
+                entered.notified().await;
+                cancelled.cancel();
+                resume.notify_one();
+            }
+        );
+        let err = result.expect_err("losing ownership during admission must prevent mutation");
+        assert!(err.to_string().contains("ownership was lost"), "resuming={resuming}: {err}");
+        assert_eq!(*store.objects.lock().await, before);
+        assert_eq!(*store.revisions.lock().await, revisions_before);
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_usage_state_reset_rejects_corruption_without_a_trusted_floor() {
+    let (_temp_dir, store) = setup_scanner_cycle_store_with_usage_baseline(false).await;
+    save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), b"{corrupt".to_vec())
+        .await
+        .expect("corrupt primary should persist");
+    let before = read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+        .await
+        .expect("evidence should load");
+    let err = reset_scanner_usage_state_for_full_rebuild(CancellationToken::new(), store.clone())
+        .await
+        .expect_err("corruption must not become a zero floor");
+    assert!(err.to_string().contains("no trusted cycle or usage floor"), "{err}");
+    assert_eq!(
+        read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+            .await
+            .expect("evidence should remain"),
+        before
+    );
+    assert!(matches!(
+        read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str()).await,
+        Err(EcstoreError::ConfigNotFound)
+    ));
+    let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    backup.scanner_epoch = Some(7);
+    backup.scanner_cycle = Some(40);
+    save_config(
+        store.clone(),
+        &format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()),
+        serde_json::to_vec(&backup).expect("backup should encode"),
+    )
+    .await
+    .expect("valid backup should persist");
+    let result = reset_scanner_usage_state_for_full_rebuild(CancellationToken::new(), store)
+        .await
+        .expect("valid backup should supply the recovery floor");
+    assert_eq!(result.leader_epoch, 8);
+    assert_eq!(result.next_cycle, 41);
+}
+
+#[tokio::test]
+async fn scanner_usage_state_reset_rejects_replaced_intent_and_newer_cleanup_slot() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let marker = scanner_usage_bootstrap_marker(std::time::SystemTime::UNIX_EPOCH, Some(3));
+    let bytes = serde_json::to_vec(&marker).expect("marker should encode");
+    save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), bytes.clone())
+        .await
+        .expect("intent should persist");
+    let slots = read_usage_state_reset_slots(store.clone()).await.expect("slots should load");
+    save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), bytes)
+        .await
+        .expect("another intent should persist");
+    let err = reset_scanner_usage_state_slots_for_full_rebuild(store.clone(), &slots, 0, 3, || true)
+        .await
+        .expect_err("same epoch cannot replace an intent revision");
+    assert!(err.to_string().contains("intent revision changed"), "{err}");
+
+    let mut newer = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    newer.scanner_epoch = Some(3);
+    let path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
+    let bytes = serde_json::to_vec(&newer).expect("newer snapshot should encode");
+    save_config(store.clone(), &path, bytes.clone())
+        .await
+        .expect("newer snapshot should persist");
+    let slots = read_usage_state_reset_slots(store.clone())
+        .await
+        .expect("slots should reload");
+    let err = reset_scanner_usage_state_slots_for_full_rebuild(store.clone(), &slots, 0, 3, || true)
+        .await
+        .expect_err("cleanup cannot delete same-epoch progress");
+    assert!(err.to_string().contains("not older than its intent"), "{err}");
+    assert_eq!(read_config(store, &path).await.expect("newer snapshot should remain"), bytes);
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_usage_state_reset_rejects_decodable_untrusted_floor() {
+    let (_temp_dir, store) = setup_scanner_cycle_store_with_usage_baseline(false).await;
+    let invalid_identity = DataUsageInfo {
+        usage_snapshot_complete: true,
+        buckets_count: 1,
+        last_update: Some(std::time::SystemTime::UNIX_EPOCH),
+        ..Default::default()
+    };
+    for usage in [DataUsageInfo::default(), invalid_identity] {
+        save_config(
+            store.clone(),
+            DATA_USAGE_OBJ_NAME_PATH.as_str(),
+            serde_json::to_vec(&usage).expect("fixture should encode"),
+        )
+        .await
+        .expect("untrusted primary should persist");
+        let before = read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+            .await
+            .expect("primary should load");
+        let err = reset_scanner_usage_state_for_full_rebuild(CancellationToken::new(), store.clone())
+            .await
+            .expect_err("valid JSON alone cannot prove a usage floor");
+        assert!(err.to_string().contains("no trusted cycle or usage floor"), "{err}");
+        assert_eq!(
+            read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+                .await
+                .expect("evidence should remain"),
+            before
+        );
+        assert!(matches!(
+            read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str()).await,
+            Err(EcstoreError::ConfigNotFound)
+        ));
+    }
+}
+
+#[test]
+fn full_rescan_reset_rejects_unknown_marker_phase_even_with_invalid_compat_fields() {
+    for state in [serde_json::json!("rewrite-v2"), serde_json::json!(7), serde_json::Value::Null] {
+        let marker = serde_json::json!({"state": state, "retry_count": "future-type", "schema_version": 99});
+        let err = super::cycle_state::decode_recovery_marker_for_reset(
+            &serde_json::to_vec(&marker).expect("future marker should encode"),
+            &DataUsageCacheRevision::Etag("intent-1".to_string()),
+        )
+        .expect_err("unknown persistent phases must remain fenced");
+        assert!(err.to_string().contains("state is unsupported"), "{err}");
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn full_rescan_reset_preserves_unknown_phase_and_retries_completed_cleanup() {
+    let (_temp_dir, store) = setup_scanner_cycle_store().await;
+    save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), b"corrupt".to_vec())
+        .await
+        .expect("corrupt primary should persist");
+    save_config(
+        store.clone(),
+        DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
+        br#"{"state":"future-rewrite"}"#.to_vec(),
+    )
+    .await
+    .expect("future marker should persist");
+    let primary_before = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
+        .await
+        .expect("primary should load");
+    let marker_before = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
+        .await
+        .expect("marker should load");
+    let err = reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
+        .await
+        .expect_err("unknown phase must block explicit reset");
+    assert!(err.to_string().contains("state is unsupported"), "{err}");
+    assert_eq!(
+        read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
+            .await
+            .expect("primary should remain"),
+        primary_before
+    );
+    assert_eq!(
+        read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
+            .await
+            .expect("marker should remain"),
+        marker_before
+    );
+
+    save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"{malformed".to_vec())
+        .await
+        .expect("recoverable marker should persist");
+    reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
+        .await
+        .expect("reset should complete");
+    let primary = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
+        .await
+        .expect("rebuilt primary should load");
+    let usage = read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+        .await
+        .expect("fenced usage should load");
+    reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
+        .await
+        .expect("retry after marker deletion should complete");
+    assert_eq!(
+        read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
+            .await
+            .expect("rebuilt primary should remain"),
+        primary
+    );
+    assert_eq!(
+        read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+            .await
+            .expect("fenced usage should remain"),
+        usage
+    );
+}
+
 #[tokio::test]
 async fn scanner_usage_state_reset_slots_defer_when_publication_epoch_moves() {
     let store = Arc::new(MemoryConfigStore::default());
@@ -4993,7 +5367,7 @@ async fn scanner_usage_state_reset_slots_defer_when_publication_epoch_moves() {
         .expect("usage reset slots should be inspected");
     store.publication_admission_blocked.store(true, Ordering::Release);
 
-    let err = reset_scanner_usage_state_slots_for_full_rebuild(store, &slots, 0, 3)
+    let err = reset_scanner_usage_state_slots_for_full_rebuild(store, &slots, 0, 3, || true)
         .await
         .expect_err("movement admission loss must defer reset");
     assert!(
@@ -8169,6 +8543,44 @@ fn scanner_activity_snapshot_digest_fences_dirty_usage_state() {
     assert_ne!(scanner_activity_snapshot_digest(&clean), scanner_activity_snapshot_digest(&pending));
 }
 
+#[test]
+fn scanner_activity_structural_digest_ignores_regular_bucket_writes() {
+    let baseline = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
+    let mut written = baseline.clone();
+    let activity = written.get_mut("node-2").expect("node should exist");
+    activity.namespace_generation = 8;
+    activity.dirty_usage_generation = 6;
+    activity.dirty_usage_pending = true;
+
+    assert_ne!(scanner_activity_snapshot_digest(&baseline), scanner_activity_snapshot_digest(&written));
+    assert_eq!(
+        scanner_activity_structural_digest(&baseline),
+        scanner_activity_structural_digest(&written),
+        "bucket writes are refreshed through the dirty-bucket scope rather than invalidating every cache"
+    );
+}
+
+#[test]
+fn scanner_activity_structural_digest_fences_restart_and_maintenance() {
+    let baseline = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
+    let mut restarted = baseline.clone();
+    restarted.get_mut("node-2").expect("node should exist").instance_id = "epoch-b".to_string();
+    let mut maintained = baseline.clone();
+    maintained
+        .get_mut("node-2")
+        .expect("node should exist")
+        .maintenance_generation = 4;
+
+    assert_ne!(
+        scanner_activity_structural_digest(&baseline),
+        scanner_activity_structural_digest(&restarted)
+    );
+    assert_ne!(
+        scanner_activity_structural_digest(&baseline),
+        scanner_activity_structural_digest(&maintained)
+    );
+}
+
 #[test]
 fn scanner_dirty_usage_acknowledgements_exclude_local_and_clean_nodes() {
     let snapshot = BTreeMap::from([

crates/scanner/src/scanner_io.rs

@@ -21,6 +21,7 @@ use crate::{
     DataUsageCacheSource, DataUsageEntry, DataUsageEntryInfo, DataUsageInfo, DataUsageScanPlanDigest, DataUsageSnapshotSetState,
     ScannerError, SizeSummary, TierStats,
 };
+use bytes::Bytes;
 use futures::future::join_all;
 use metrics::counter;
 use rand::seq::SliceRandom as _;
@@ -54,6 +55,7 @@ use tokio_util::task::AbortOnDropHandle;
 use tracing::{debug, error, warn};
 
 use crate::ScannerObjectInfo as ObjectInfo;
+use crate::storage_api::EcstoreScannerPeerDirtyUsageSnapshot;
 use crate::storage_api::ScannerStorage;
 use crate::storage_api::scan::NamespaceLocking as _;
 use crate::storage_api::scanner_io::{BucketInfo, BucketOptions};
@@ -111,6 +113,121 @@ pub(crate) struct ScannerBucketScanScope {
     baseline_scan_plan_digest: Option<DataUsageScanPlanDigest>,
 }
 
+impl ScannerBucketScanScope {
+    fn is_default(&self) -> bool {
+        self.selected_buckets.is_none() && self.baseline_scan_plan_digest.is_none()
+    }
+
+    fn from_dirty_buckets(selected_buckets: HashSet<String>, baseline_scan_plan_digest: DataUsageScanPlanDigest) -> Self {
+        Self {
+            selected_buckets: Some(Arc::new(selected_buckets)),
+            baseline_scan_plan_digest: Some(baseline_scan_plan_digest),
+        }
+    }
+}
+
+#[derive(Clone, Copy)]
+pub(super) struct ScannerCacheBaselineProof<'a> {
+    pub(super) data: Option<&'a Bytes>,
+    pub(super) expected_sources: &'a HashSet<DataUsageCacheSource>,
+    pub(super) leader_epoch: u64,
+    pub(super) want_cycle: u64,
+    pub(super) scan_plan_digest: DataUsageScanPlanDigest,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+struct ScannerPeerDirtyUsageExpectation {
+    instance_id: String,
+    generation: u64,
+    pending: bool,
+}
+
+fn verified_remote_dirty_usage_buckets(
+    expected_peers: &HashMap<String, ScannerPeerDirtyUsageExpectation>,
+    peer_snapshots: Vec<(String, EcstoreScannerPeerDirtyUsageSnapshot)>,
+) -> Option<HashSet<String>> {
+    if expected_peers.is_empty() || peer_snapshots.len() != expected_peers.len() {
+        return None;
+    }
+
+    let mut received_peers = HashSet::with_capacity(peer_snapshots.len());
+    let mut dirty_buckets = HashSet::new();
+    for (host, snapshot) in peer_snapshots {
+        let expected = expected_peers.get(&host)?;
+        if !received_peers.insert(host)
+            || snapshot.instance_id != expected.instance_id
+            || snapshot.generation != expected.generation
+            || snapshot.generation == u64::MAX
+            || snapshot.protocol_version != crate::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION
+            || !snapshot.complete
+            || snapshot.pending_bucket_count != u64::try_from(snapshot.buckets.len()).unwrap_or(u64::MAX)
+            || (expected.pending && snapshot.pending_bucket_count == 0)
+        {
+            return None;
+        }
+        dirty_buckets.extend(snapshot.buckets.into_keys());
+    }
+
+    (received_peers.len() == expected_peers.len()).then_some(dirty_buckets)
+}
+
+fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<'_>) -> Option<DataUsageScanPlanDigest> {
+    let data = proof.data?;
+    let baseline = serde_json::from_slice::<DataUsageInfo>(data).ok()?;
+    if !baseline.is_complete_bucket_usage_snapshot()
+        || baseline.usage_snapshot_partial
+        || baseline.usage_snapshot_converged != Some(true)
+        || baseline.scanner_epoch != Some(proof.leader_epoch)
+        || baseline.usage_snapshot_set_states.len() != proof.expected_sources.len()
+    {
+        return None;
+    }
+
+    let mut states = HashSet::with_capacity(baseline.usage_snapshot_set_states.len());
+    for state in &baseline.usage_snapshot_set_states {
+        let source = DataUsageCacheSource::new(usize::try_from(state.pool_index).ok()?, usize::try_from(state.set_index).ok()?);
+        if !proof.expected_sources.contains(&source)
+            || !states.insert(source)
+            || !state.complete
+            || state.tombstone
+            || state.scanner_epoch != Some(proof.leader_epoch)
+            || state.scanner_cycle.is_none_or(|cycle| cycle > proof.want_cycle)
+            || state.scan_plan_digest != Some(proof.scan_plan_digest.0)
+        {
+            return None;
+        }
+    }
+
+    (states == *proof.expected_sources).then_some(proof.scan_plan_digest)
+}
+
+fn scoped_scan_scope_from_dirty_buckets(
+    requested_scope: ScannerBucketScanScope,
+    dirty_buckets: HashSet<String>,
+    dirty_snapshot_complete: bool,
+    all_buckets: &[BucketInfo],
+    baseline_proof: ScannerCacheBaselineProof<'_>,
+) -> ScannerBucketScanScope {
+    if !requested_scope.is_default() || !dirty_snapshot_complete {
+        return requested_scope;
+    }
+
+    let current_buckets = all_buckets.iter().map(|bucket| bucket.name.as_str()).collect::<HashSet<_>>();
+    let selected_buckets = dirty_buckets
+        .into_iter()
+        .filter(|bucket| current_buckets.contains(bucket.as_str()))
+        .collect::<HashSet<_>>();
+    if selected_buckets.is_empty() {
+        return requested_scope;
+    }
+
+    let Some(baseline_scan_plan_digest) = complete_scanner_cache_baseline_plan_digest(baseline_proof) else {
+        return requested_scope;
+    };
+
+    ScannerBucketScanScope::from_dirty_buckets(selected_buckets, baseline_scan_plan_digest)
+}
+
 pub(crate) fn is_scanner_metadata_corrupt_error(err: &StorageError) -> bool {
     matches!(err, StorageError::Io(io) if io.to_string().starts_with(SCANNER_METADATA_CORRUPT_ERROR))
 }
@@ -749,7 +866,7 @@ mod io_cache;
 mod io_cycle;
 #[cfg(test)]
 use io_cache::{ScannerSetCacheGeneration, prepare_scoped_set_scan};
-pub(crate) use io_cycle::nsscanner_with_storage_status;
+pub(crate) use io_cycle::{ScannerCycleRequest, nsscanner_with_storage_status_scoped};
 mod io_disk;
 #[cfg(test)]
 mod publish_gate_tests;
@@ -766,8 +883,9 @@ pub(crate) use cache::{
 };
 pub use dirty_usage::{
     ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState,
-    acknowledge_dirty_usage_generation, clear_dirty_usage_bucket, record_dirty_usage_bucket, record_scanner_maintenance_change,
-    scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state, scanner_maintenance_generation,
+    acknowledge_dirty_usage_generation, acknowledge_scoped_dirty_usage, clear_dirty_usage_bucket, record_dirty_usage_bucket,
+    record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state,
+    scanner_maintenance_generation,
 };
 #[cfg(test)]
 pub(crate) use dirty_usage::{clear_dirty_usage_buckets_for_tests, dirty_usage_buckets_for_tests};

crates/scanner/src/scanner_io/cache.rs

@@ -282,9 +282,26 @@ pub(super) fn completed_data_usage_info(
         .iter()
         .map(|(bucket, usage)| (bucket.clone(), usage.size))
         .collect();
+    let mut usage_snapshot_set_states = results
+        .iter()
+        .map(|result| {
+            let source = result.info.source?;
+            Some(DataUsageSnapshotSetState {
+                pool_index: u64::try_from(source.pool_index).ok()?,
+                set_index: u64::try_from(source.set_index).ok()?,
+                scanner_cycle: Some(result.info.next_cycle),
+                scanner_epoch: Some(result.info.leader_epoch),
+                scan_plan_digest: Some(result.info.scan_plan_digest?.0),
+                complete: true,
+                tombstone: false,
+            })
+        })
+        .collect::<Option<Vec<_>>>()?;
+    usage_snapshot_set_states.sort_by_key(|state| (state.pool_index, state.set_index));
     let data_usage_info = DataUsageInfo {
         last_update: Some(merged_last_update),
         scanner_cycle: Some(results.first()?.info.next_cycle),
+        scanner_epoch: Some(results.first()?.info.leader_epoch),
         objects_total_count: u64::try_from(total.objects).ok()?,
         versions_total_count: u64::try_from(total.versions).ok()?,
         delete_markers_total_count: u64::try_from(total.delete_markers).ok()?,
@@ -295,6 +312,7 @@ pub(super) fn completed_data_usage_info(
         bucket_sizes,
         buckets_usage,
         usage_snapshot_complete: true,
+        usage_snapshot_set_states,
         ..Default::default()
     };
     Some((data_usage_info, merged_last_update))

crates/scanner/src/scanner_io/dirty_usage.rs

@@ -52,6 +52,112 @@ pub enum ScannerDirtyUsageAckError {
     ProcessChanged,
     #[error("scanner dirty usage generation cannot be acknowledged")]
     InvalidGeneration,
+    #[error("scanner dirty usage bucket incarnation fence is unavailable")]
+    IncarnationUnavailable,
+}
+
+/// A scoped ACK requires storage-owned lifecycle and incarnation fences.
+/// Callers must only send ACKs backed by durable per-bucket publication.
+pub fn acknowledge_scoped_dirty_usage(
+    instance_id: &str,
+    entries: &[(&crate::storage_api::EcstoreBucketMetadataMutationGuard, u64)],
+    probe_only: bool,
+) -> std::result::Result<u64, ScannerDirtyUsageAckError> {
+    // Lock order: sorted bucket lifecycle/metadata fences (caller), then dirty map.
+    // No await or storage operation occurs while the dirty map is locked.
+    let (cleared, pending) = {
+        let mut dirty = dirty_usage_buckets();
+        let checked = entries
+            .iter()
+            .map(|(guard, generation)| {
+                guard
+                    .checked_bucket_incarnation()
+                    .map(|(bucket, _)| (bucket, *generation))
+                    .map_err(|_| ScannerDirtyUsageAckError::IncarnationUnavailable)
+            })
+            .collect::<std::result::Result<Vec<_>, _>>()?;
+        let cleared = apply_scoped_dirty_usage_ack(
+            instance_id,
+            scanner_activity_epoch(),
+            DIRTY_USAGE_BUCKET_GENERATION.load(Ordering::Acquire),
+            &mut dirty,
+            &checked,
+            probe_only,
+        )?;
+        if cleared > 0 {
+            advance_generation(&DIRTY_USAGE_BUCKET_GENERATION);
+        }
+        (cleared, dirty.len())
+    };
+    if !probe_only {
+        global_metrics().record_scanner_dirty_usage_cycle_clear(usize_to_u64_saturated(cleared), usize_to_u64_saturated(pending));
+    }
+    Ok(usize_to_u64_saturated(cleared))
+}
+
+fn apply_scoped_dirty_usage_ack(
+    instance_id: &str,
+    current_instance: &str,
+    current_generation: u64,
+    dirty: &mut DirtyUsageBuckets,
+    entries: &[(&str, u64)],
+    probe_only: bool,
+) -> std::result::Result<usize, ScannerDirtyUsageAckError> {
+    if instance_id != current_instance {
+        return Err(ScannerDirtyUsageAckError::ProcessChanged);
+    }
+    if current_generation == u64::MAX
+        || entries
+            .iter()
+            .any(|(_, generation)| *generation == 0 || *generation == u64::MAX || *generation > current_generation)
+    {
+        return Err(ScannerDirtyUsageAckError::InvalidGeneration);
+    }
+    let mut cleared = 0;
+    if !probe_only {
+        for (bucket, generation) in entries {
+            if dirty.get(*bucket) == Some(generation) {
+                dirty.remove(*bucket);
+                cleared += 1;
+            }
+        }
+    }
+    Ok(cleared)
+}
+
+#[cfg(test)]
+mod scoped_dirty_usage_tests {
+    use super::*;
+
+    #[test]
+    fn scoped_dirty_usage_preserves_uncovered_newer_and_replayed_generations() {
+        let mut dirty = HashMap::from([("hot".to_string(), 7), ("cold".to_string(), 8)]);
+        assert_eq!(apply_scoped_dirty_usage_ack("p", "p", 8, &mut dirty, &[("cold", 8)], true), Ok(0));
+        assert_eq!(dirty.len(), 2);
+        assert_eq!(apply_scoped_dirty_usage_ack("p", "p", 8, &mut dirty, &[("cold", 8)], false), Ok(1));
+        assert_eq!(dirty.get("hot"), Some(&7));
+        assert_eq!(apply_scoped_dirty_usage_ack("p", "p", 8, &mut dirty, &[("cold", 8)], false), Ok(0));
+        dirty.insert("cold".to_string(), 9);
+        assert_eq!(apply_scoped_dirty_usage_ack("p", "p", 9, &mut dirty, &[("cold", 8)], false), Ok(0));
+        assert_eq!(dirty.get("cold"), Some(&9));
+    }
+
+    #[test]
+    fn scoped_dirty_usage_rejects_restart_and_invalid_batch_before_clearing() {
+        let original = HashMap::from([("hot".to_string(), 7), ("cold".to_string(), 8)]);
+        let mut dirty = original.clone();
+        assert_eq!(
+            apply_scoped_dirty_usage_ack("old", "new", 8, &mut dirty, &[("cold", 8)], false),
+            Err(ScannerDirtyUsageAckError::ProcessChanged)
+        );
+        for generation in [0, 9, u64::MAX] {
+            assert_eq!(
+                apply_scoped_dirty_usage_ack("p", "p", 8, &mut dirty, &[("cold", 8), ("hot", generation)], false),
+                Err(ScannerDirtyUsageAckError::InvalidGeneration)
+            );
+            assert_eq!(dirty, original);
+        }
+    }
 }
 
 pub(super) fn dirty_usage_buckets() -> MutexGuard<'static, DirtyUsageBuckets> {

crates/scanner/src/scanner_io/io_cycle.rs

@@ -71,6 +71,7 @@ where
         leader_epoch,
         scan_mode,
         scan_scope: ScannerBucketScanScope::default(),
+        persisted_usage_baseline: None,
     };
     nsscanner_with_storage_status_scoped(store, request).await
 }
@@ -83,6 +84,79 @@ pub(crate) struct ScannerCycleRequest {
     pub(crate) leader_epoch: u64,
     pub(crate) scan_mode: HealScanMode,
     pub(crate) scan_scope: ScannerBucketScanScope,
+    pub(crate) persisted_usage_baseline: Option<Bytes>,
+}
+
+struct ScannerBucketScopeResolution<'a> {
+    requested_scope: ScannerBucketScanScope,
+    baseline_proof: ScannerCacheBaselineProof<'a>,
+    activity_before: &'a crate::scanner::ScannerActivitySnapshot,
+    dirty_usage_snapshot: &'a DirtyUsageSnapshot,
+    all_buckets: &'a [BucketInfo],
+}
+
+async fn resolve_scanner_bucket_scan_scope<S>(
+    store: &S,
+    distributed: bool,
+    resolution: ScannerBucketScopeResolution<'_>,
+) -> ScannerBucketScanScope
+where
+    S: ScannerStorage,
+{
+    if !resolution.requested_scope.is_default()
+        || !resolution.dirty_usage_snapshot.covers_all_pending
+        || resolution.dirty_usage_snapshot.generation == u64::MAX
+        || resolution.dirty_usage_snapshot.buckets.len() > crate::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES
+    {
+        return resolution.requested_scope;
+    }
+
+    let mut dirty_buckets = resolution
+        .dirty_usage_snapshot
+        .buckets
+        .keys()
+        .cloned()
+        .collect::<HashSet<_>>();
+    if distributed {
+        let Some(notification_system) = store.scanner_notification_system() else {
+            return resolution.requested_scope;
+        };
+        let Ok(peer_snapshots) = notification_system.scanner_dirty_usage_snapshots().await else {
+            return resolution.requested_scope;
+        };
+        let mut expected_peers = HashMap::new();
+        for (host, lease_instance_id, _) in crate::scanner::scanner_activity_publication_lease_targets(resolution.activity_before)
+        {
+            let Some((activity_instance_id, generation, pending)) =
+                crate::scanner::scanner_activity_dirty_usage_state_for_host(resolution.activity_before, &host)
+            else {
+                return resolution.requested_scope;
+            };
+            if activity_instance_id != lease_instance_id || expected_peers.contains_key(&host) {
+                return resolution.requested_scope;
+            }
+            expected_peers.insert(
+                host,
+                ScannerPeerDirtyUsageExpectation {
+                    instance_id: activity_instance_id.to_string(),
+                    generation,
+                    pending,
+                },
+            );
+        }
+        let Some(remote_dirty_buckets) = verified_remote_dirty_usage_buckets(&expected_peers, peer_snapshots) else {
+            return resolution.requested_scope;
+        };
+        dirty_buckets.extend(remote_dirty_buckets);
+    }
+
+    scoped_scan_scope_from_dirty_buckets(
+        resolution.requested_scope,
+        dirty_buckets,
+        true,
+        resolution.all_buckets,
+        resolution.baseline_proof,
+    )
 }
 
 pub(crate) async fn nsscanner_with_storage_status_scoped<S>(store: &S, request: ScannerCycleRequest) -> Result<ScannerCycleResult>
@@ -97,6 +171,7 @@ where
         leader_epoch,
         scan_mode,
         scan_scope,
+        persisted_usage_baseline,
     } = request;
     let child_token = ctx.child_token();
     let _tier_cycle_guard = begin_tier_registry_cycle(want_cycle, leader_epoch);
@@ -186,8 +261,26 @@ where
     }
     bucket_plan_complete &= buckets_by_source.keys().copied().collect::<HashSet<_>>() == *expected_sources;
     let scan_plan_digest =
-        scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_snapshot_digest(&activity_before));
+        scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_structural_digest(&activity_before));
     let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets(&all_buckets, dirty_generation_before_bucket_list));
+    let scan_scope = resolve_scanner_bucket_scan_scope(
+        store,
+        distributed,
+        ScannerBucketScopeResolution {
+            requested_scope: scan_scope,
+            baseline_proof: ScannerCacheBaselineProof {
+                data: persisted_usage_baseline.as_ref(),
+                expected_sources: &expected_sources,
+                leader_epoch,
+                want_cycle,
+                scan_plan_digest,
+            },
+            activity_before: &activity_before,
+            dirty_usage_snapshot: &dirty_usage_snapshot,
+            all_buckets: &all_buckets,
+        },
+    )
+    .await;
     let cache_cycle_floor = Arc::new(AtomicU64::new(want_cycle));
     let tier_registry = runtime_tier_registry_for_cycle(want_cycle, leader_epoch).await;
     let tier_registry_generation = tier_registry.generation;

crates/scanner/src/scanner_io/publish_gate_tests.rs

@@ -655,9 +655,31 @@ fn completed_data_usage_info_requires_every_set_before_publish() {
         .expect("all completed sets should produce a publishable data usage snapshot");
     assert_eq!(last_update, SystemTime::UNIX_EPOCH + Duration::from_secs(20));
     assert_eq!(data_usage_info.scanner_cycle, Some(0));
+    assert_eq!(data_usage_info.scanner_epoch, Some(0));
     assert_eq!(data_usage_info.objects_total_count, 3);
     assert_eq!(data_usage_info.buckets_usage.len(), 3);
     assert!(data_usage_info.usage_snapshot_complete);
+    assert_eq!(
+        data_usage_info
+            .usage_snapshot_set_states
+            .iter()
+            .map(|state| {
+                (
+                    state.pool_index,
+                    state.set_index,
+                    state.scanner_cycle,
+                    state.scanner_epoch,
+                    state.scan_plan_digest,
+                    state.complete,
+                    state.tombstone,
+                )
+            })
+            .collect::<Vec<_>>(),
+        vec![
+            (0, 0, Some(0), Some(0), Some(TEST_PLAN_DIGEST.0), true, false),
+            (1, 0, Some(0), Some(0), Some(TEST_PLAN_DIGEST.0), true, false),
+        ]
+    );
     assert_eq!(
         data_usage_info
             .buckets_usage

crates/scanner/src/scanner_io/tests.rs

@@ -17,6 +17,7 @@ use super::io_disk::tier_stats_template;
 use super::*;
 use crate::scanner_budget::ScannerCycleBudgetConfig;
 use crate::scanner_folder::ScannerItem;
+use crate::storage_api::EcstoreScannerPeerDirtyUsageSnapshot;
 use crate::storage_api::owner::{
     EcstorePoolDecommissionInfo, EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats,
 };
@@ -796,6 +797,195 @@ fn complete_set_usage_cache(buckets: &[(&str, usize)], scan_plan_digest: DataUsa
     cache
 }
 
+fn complete_usage_baseline(
+    source: DataUsageCacheSource,
+    scan_plan_digest: DataUsageScanPlanDigest,
+    scanner_cycle: u64,
+    scanner_epoch: u64,
+) -> bytes::Bytes {
+    let baseline = DataUsageInfo {
+        last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(10)),
+        scanner_cycle: Some(scanner_cycle),
+        scanner_epoch: Some(scanner_epoch),
+        buckets_count: 1,
+        buckets_usage: HashMap::from([("photos".to_string(), Default::default())]),
+        usage_snapshot_complete: true,
+        usage_snapshot_converged: Some(true),
+        usage_snapshot_set_states: vec![DataUsageSnapshotSetState {
+            pool_index: u64::try_from(source.pool_index).expect("test pool index should fit"),
+            set_index: u64::try_from(source.set_index).expect("test set index should fit"),
+            scanner_cycle: Some(scanner_cycle),
+            scanner_epoch: Some(scanner_epoch),
+            scan_plan_digest: Some(scan_plan_digest.0),
+            complete: true,
+            tombstone: false,
+        }],
+        ..Default::default()
+    };
+    bytes::Bytes::from(serde_json::to_vec(&baseline).expect("test baseline should encode"))
+}
+
+#[test]
+fn scoped_scan_requires_a_converged_complete_baseline_with_exact_set_provenance() {
+    let source = DataUsageCacheSource::new(1, 2);
+    let expected_sources = HashSet::from([source]);
+    let scan_plan_digest = DataUsageScanPlanDigest([9; 32]);
+    let baseline = complete_usage_baseline(source, scan_plan_digest, 7, 11);
+
+    assert_eq!(
+        complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
+            data: Some(&baseline),
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 8,
+            scan_plan_digest,
+        }),
+        Some(scan_plan_digest)
+    );
+
+    let mut incomplete = serde_json::from_slice::<DataUsageInfo>(&baseline).expect("test baseline should decode");
+    incomplete.usage_snapshot_converged = Some(false);
+    let incomplete = bytes::Bytes::from(serde_json::to_vec(&incomplete).expect("test baseline should encode"));
+    assert_eq!(
+        complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
+            data: Some(&incomplete),
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 8,
+            scan_plan_digest,
+        }),
+        None
+    );
+
+    let mut wrong_provenance = serde_json::from_slice::<DataUsageInfo>(&baseline).expect("test baseline should decode");
+    wrong_provenance.usage_snapshot_set_states[0].scan_plan_digest = Some([8; 32]);
+    let wrong_provenance = bytes::Bytes::from(serde_json::to_vec(&wrong_provenance).expect("test baseline should encode"));
+    assert_eq!(
+        complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
+            data: Some(&wrong_provenance),
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 8,
+            scan_plan_digest,
+        }),
+        None
+    );
+}
+
+#[test]
+fn scoped_scan_selects_only_current_dirty_buckets_after_baseline_validation() {
+    let source = DataUsageCacheSource::new(1, 2);
+    let expected_sources = HashSet::from([source]);
+    let baseline_scan_plan_digest = DataUsageScanPlanDigest([4; 32]);
+    let current_scan_plan_digest = DataUsageScanPlanDigest([5; 32]);
+    let baseline = complete_usage_baseline(source, current_scan_plan_digest, 7, 11);
+    let scope = scoped_scan_scope_from_dirty_buckets(
+        ScannerBucketScanScope::default(),
+        HashSet::from(["photos".to_string(), "deleted".to_string()]),
+        true,
+        &[bucket_info("photos")],
+        ScannerCacheBaselineProof {
+            data: Some(&baseline),
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 8,
+            scan_plan_digest: current_scan_plan_digest,
+        },
+    );
+
+    assert_eq!(scope.baseline_scan_plan_digest, Some(current_scan_plan_digest));
+    assert_eq!(
+        scope
+            .selected_buckets
+            .as_deref()
+            .expect("validated scope should select a bucket"),
+        &HashSet::from(["photos".to_string()])
+    );
+    assert_ne!(scope.baseline_scan_plan_digest, Some(baseline_scan_plan_digest));
+}
+
+fn peer_dirty_usage_snapshot(
+    instance_id: &str,
+    generation: u64,
+    complete: bool,
+    buckets: &[(&str, u64)],
+) -> EcstoreScannerPeerDirtyUsageSnapshot {
+    EcstoreScannerPeerDirtyUsageSnapshot {
+        instance_id: instance_id.to_string(),
+        generation,
+        pending_bucket_count: u64::try_from(buckets.len()).expect("test bucket count should fit"),
+        protocol_version: crate::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION,
+        complete,
+        buckets: buckets
+            .iter()
+            .map(|(bucket, generation)| ((*bucket).to_string(), *generation))
+            .collect(),
+    }
+}
+
+#[test]
+fn verified_remote_dirty_usage_buckets_merges_only_complete_current_snapshots() {
+    let expected_peers = HashMap::from([
+        (
+            "node-a:9000".to_string(),
+            ScannerPeerDirtyUsageExpectation {
+                instance_id: "instance-a".to_string(),
+                generation: 7,
+                pending: true,
+            },
+        ),
+        (
+            "node-b:9000".to_string(),
+            ScannerPeerDirtyUsageExpectation {
+                instance_id: "instance-b".to_string(),
+                generation: 3,
+                pending: false,
+            },
+        ),
+    ]);
+
+    assert_eq!(
+        verified_remote_dirty_usage_buckets(
+            &expected_peers,
+            vec![
+                (
+                    "node-a:9000".to_string(),
+                    peer_dirty_usage_snapshot("instance-a", 7, true, &[("photos", 7)]),
+                ),
+                (
+                    "node-b:9000".to_string(),
+                    peer_dirty_usage_snapshot("instance-b", 3, true, &[("archive", 3)]),
+                ),
+            ],
+        ),
+        Some(HashSet::from(["photos".to_string(), "archive".to_string()]))
+    );
+}
+
+#[test]
+fn verified_remote_dirty_usage_buckets_rejects_incomplete_or_stale_peer_state() {
+    let expected_peers = HashMap::from([(
+        "node-a:9000".to_string(),
+        ScannerPeerDirtyUsageExpectation {
+            instance_id: "instance-a".to_string(),
+            generation: 7,
+            pending: true,
+        },
+    )]);
+
+    for snapshot in [
+        peer_dirty_usage_snapshot("instance-a", 7, false, &[("photos", 7)]),
+        peer_dirty_usage_snapshot("instance-a", 6, true, &[("photos", 6)]),
+        peer_dirty_usage_snapshot("instance-b", 7, true, &[("photos", 7)]),
+        peer_dirty_usage_snapshot("instance-a", 7, true, &[]),
+    ] {
+        assert!(
+            verified_remote_dirty_usage_buckets(&expected_peers, vec![("node-a:9000".to_string(), snapshot)]).is_none(),
+            "incomplete, stale, mismatched, or empty pending peer state must fall back to a full scan"
+        );
+    }
+}
+
 #[test]
 fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
     let baseline_digest = DataUsageScanPlanDigest([1; 32]);

crates/scanner/src/storage_api.rs

@@ -38,8 +38,8 @@ pub(crate) use rustfs_ecstore::api::bucket::lifecycle::lifecycle::object_opts_fr
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::init_bucket_metadata_sys as ecstore_init_bucket_metadata_sys;
 pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::{
-    get_lifecycle_config as ecstore_get_lifecycle_config, get_object_lock_config as ecstore_get_object_lock_config,
-    get_replication_config as ecstore_get_replication_config,
+    BucketMetadataMutationGuard as EcstoreBucketMetadataMutationGuard, get_lifecycle_config as ecstore_get_lifecycle_config,
+    get_object_lock_config as ecstore_get_object_lock_config, get_replication_config as ecstore_get_replication_config,
 };
 pub(crate) use rustfs_ecstore::api::bucket::replication::{
     ReplicateObjectInfo, ReplicationConfig as EcstoreReplicationConfig,
@@ -103,7 +103,9 @@ pub(crate) use rustfs_ecstore::api::rebalance::{
     RebalStatus as EcstoreRebalStatus, RebalanceInfo as EcstoreRebalanceInfo, RebalanceMeta as EcstoreRebalanceMeta,
     RebalanceStats as EcstoreRebalanceStats,
 };
-pub(crate) use rustfs_ecstore::api::rpc::ScannerBucketListing as EcstoreScannerBucketListing;
+pub(crate) use rustfs_ecstore::api::rpc::{
+    ScannerBucketListing as EcstoreScannerBucketListing, ScannerPeerDirtyUsageSnapshot as EcstoreScannerPeerDirtyUsageSnapshot,
+};
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::runtime::InstanceContext as EcstoreInstanceContext;
 pub(crate) use rustfs_ecstore::api::runtime::{

docs/operations/on-demand-migration.md

@@ -7,6 +7,16 @@ On-Demand Migration (ODM) attaches an external S3-compatible **source bucket** t
 
 The module is on by default (rustfs/backlog#2163); set `RUSTFS_ON_DEMAND_MIGRATION_ENABLED=false` on every node to turn it off (`rustfs/src/module_switches.rs`). With the switch off, the runtime never intervenes on a read and the admin `PUT` route refuses with `OnDemandMigrationDisabled`. Reads of the configuration and of the status endpoint keep working while the switch is off, so a disabled deployment can still be inspected. The switch only decides whether the module may act at all: a bucket with no `on-demand-migration.json` is never resolved by the runtime and makes no source call, so turning the module on changes nothing for buckets you have not configured.
 
+## List continuation token rollout
+
+`RUSTFS_ON_DEMAND_MIGRATION_LIST_V2_TOKENS` defaults to `false`; unset or invalid boolean values also keep it off. It controls only whether a v1 listing may first issue a v2 continuation token after an empty truncated merged page. Every node with this reader support accepts existing v2 tokens and continues their budget even with the switch off. Ordinary pages that consume an object or common prefix retain the original v1 token shape.
+
+Leave the switch off while deploying v2 reader support to every node that can receive a continuation request, including nodes behind other load-balancer routes. Then set it to `true` in each node's environment and restart those nodes to enable issuance. A v1-only binary rejects v2 with `400 InvalidArgument` before the source-error policy runs; neither `not_found` nor turning off list-through makes that old reader compatible. With issuance still off, a new v1 chain retains the existing limitation: an empty source cursor cycle spanning requests can continue indefinitely. The default rollout does not claim to fix that chain until issuance is enabled.
+
+An active v2 budget rejects the sixteenth consecutive merged page that consumes no new object/common prefix and reaches no new end-of-list state. The first fifteen empty pages can be resumed; with the existing two-fetch-per-side limit, that interval costs at most 32 fetches per side, including the failing request. A key, common prefix, or a newly exhausted side on the sixteenth request succeeds and resets the budget. A side that was already exhausted does not reset it again. This is a resource bound, not proof of a cursor cycle: an unusually long but valid empty source-page chain also reaches the limit. Tokens are unsigned base64 JSON, so this budget applies to clients that continue with the returned token unchanged; replaying or editing a token can reset it, and it is not a malicious-client defense or a global request quota. The two-fetch-per-side request limit and existing source rate limiter still apply. A source failure follows `policy.source_error`: `propagate` returns `424 SourceUnavailable` with `invalid_pagination`; `not_found` returns the fetched local listing with `x-rustfs-on-demand-migration-list: local_only`. A blocking local-side failure returns `InternalError`, without silently discarding local entries.
+
+For rollback, first turn issuance off on every node. Keep v2-capable readers available for outstanding v2 chains: switching issuance off does not erase their budgets, and tokens have no expiration that proves those chains have drained. Route those continuations to compatible readers or have clients explicitly restart their listings before restoring v1-only binaries. Restarting a listing is a new scan and can repeat entries. Do not roll back readers while assuming the issuance switch makes existing v2 tokens disappear.
+
 ## Positioning
 
 | Capability | Direction | What it moves | Where the authoritative copy is | When to use it instead |

rustfs/src/admin/handlers/heal.rs

@@ -39,7 +39,7 @@ use rustfs_utils::path::path_join;
 use s3s::header::{CONTENT_LENGTH, CONTENT_TYPE};
 use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
 use serde::{Deserialize, Serialize};
-use std::collections::{BTreeMap, HashSet};
+use std::collections::{BTreeMap, BTreeSet, HashSet};
 use std::future::Future;
 use std::path::PathBuf;
 use std::sync::Arc;
@@ -261,6 +261,7 @@ struct BackgroundHealStatus<'a> {
     heal_active_tasks: u64,
     heal_operations: rustfs_heal::HealOperationsSnapshot,
     cluster_status_complete: bool,
+    coverage: &'a BackgroundHealCoverage,
     #[serde(skip_serializing_if = "Option::is_none")]
     progress: Option<BackgroundHealProgress>,
 }
@@ -300,6 +301,23 @@ fn background_heal_runtime_state(
 
 type BackgroundHealProgress = rustfs_heal::HealProgress;
 
+#[derive(Debug, Serialize)]
+struct BackgroundHealCoverage {
+    expected: usize,
+    responded: usize,
+    unknown: usize,
+    reasons: BTreeSet<BackgroundHealCoverageReason>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize)]
+#[serde(rename_all = "snake_case")]
+enum BackgroundHealCoverageReason {
+    NotificationSystemUnavailable,
+    PeerTopologyIncomplete,
+    PeerStatusUnsupported,
+    PeerStatusUnavailable,
+}
+
 #[derive(Debug)]
 struct ClusterHealStatusSnapshot {
     info: BackgroundHealInfo,
@@ -307,6 +325,7 @@ struct ClusterHealStatusSnapshot {
     operations: rustfs_heal::HealOperationsSnapshot,
     progress: Option<BackgroundHealProgress>,
     complete: bool,
+    coverage: BackgroundHealCoverage,
 }
 
 fn add_priority_counts(total: &mut rustfs_heal::HealPriorityCounts, next: rustfs_heal::HealPriorityCounts) {
@@ -338,6 +357,7 @@ fn add_operations(total: &mut rustfs_heal::HealOperationsSnapshot, next: rustfs_
 }
 
 fn aggregate_cluster_heal_status(snapshots: Vec<NodeHealStatusSnapshot>) -> ClusterHealStatusSnapshot {
+    let responded = snapshots.len();
     let mut info = BackgroundHealInfo::default();
     let mut operations = rustfs_heal::HealOperationsSnapshot::default();
     let mut progress = Vec::new();
@@ -379,6 +399,12 @@ fn aggregate_cluster_heal_status(snapshots: Vec<NodeHealStatusSnapshot>) -> Clus
         operations,
         progress,
         complete: true,
+        coverage: BackgroundHealCoverage {
+            expected: responded,
+            responded,
+            unknown: 0,
+            reasons: BTreeSet::new(),
+        },
     }
 }
 
@@ -413,12 +439,14 @@ fn merge_peer_heal_statuses(
     mut snapshots: Vec<NodeHealStatusSnapshot>,
     peer_statuses: Vec<Result<Option<NodeHealStatusSnapshot>, String>>,
     expected_nodes: usize,
-    topology_complete: bool,
+    coverage_reason: Option<BackgroundHealCoverageReason>,
 ) -> S3Result<ClusterHealStatusSnapshot> {
+    let mut reasons: BTreeSet<_> = coverage_reason.into_iter().collect();
     for peer_status in peer_statuses {
         match peer_status {
             Ok(Some(snapshot)) => snapshots.push(snapshot),
             Ok(None) => {
+                reasons.insert(BackgroundHealCoverageReason::PeerStatusUnsupported);
                 warn!(
                     event = EVENT_ADMIN_REQUEST_FAILED,
                     component = LOG_COMPONENT_ADMIN_API,
@@ -430,6 +458,7 @@ fn merge_peer_heal_statuses(
                 );
             }
             Err(err) => {
+                reasons.insert(BackgroundHealCoverageReason::PeerStatusUnavailable);
                 warn!(
                     event = EVENT_ADMIN_REQUEST_FAILED,
                     component = LOG_COMPONENT_ADMIN_API,
@@ -452,9 +481,12 @@ fn merge_peer_heal_statuses(
     // so during a reconfiguration the count can equal `expected_nodes` while
     // the topology is known-incomplete. Counting alone would report a
     // definitive answer precisely when the membership itself is in doubt.
-    let complete = topology_complete && snapshots.len() == expected_nodes;
+    let complete = reasons.is_empty() && snapshots.len() == expected_nodes;
     let mut status = aggregate_cluster_heal_status(snapshots);
     status.complete = complete;
+    status.coverage.expected = expected_nodes;
+    status.coverage.unknown = expected_nodes.saturating_sub(status.coverage.responded);
+    status.coverage.reasons = reasons;
     // A partial answer must never be mistakable for a definitive verdict: an
     // unreachable peer might be mid-heal, so reporting the reachable nodes'
     // "idle" (or disabled/uninitialized) as the cluster state would falsely
@@ -497,7 +529,12 @@ async fn read_cluster_heal_status(
         return Ok(aggregate_cluster_heal_status(snapshots));
     }
     let Some(notification_system) = notification_system else {
-        return Err(cluster_heal_status_unavailable("notification_system_unavailable"));
+        return merge_peer_heal_statuses(
+            snapshots,
+            Vec::new(),
+            expected_nodes,
+            Some(BackgroundHealCoverageReason::NotificationSystemUnavailable),
+        );
     };
     // An incomplete peer topology (a down member's client slot, a rolling
     // upgrade) previously failed the whole endpoint here, before any peer was
@@ -540,7 +577,12 @@ async fn read_cluster_heal_status(
     }))
     .await;
 
-    merge_peer_heal_statuses(snapshots, peer_statuses, expected_nodes, topology_complete)
+    merge_peer_heal_statuses(
+        snapshots,
+        peer_statuses,
+        expected_nodes,
+        (!topology_complete).then_some(BackgroundHealCoverageReason::PeerTopologyIncomplete),
+    )
 }
 
 async fn query_peer_replacement_recovery_status<E>(
@@ -1164,6 +1206,7 @@ fn encode_background_heal_status(
     heal_operations: rustfs_heal::HealOperationsSnapshot,
     progress: Option<BackgroundHealProgress>,
     cluster_status_complete: bool,
+    coverage: &BackgroundHealCoverage,
 ) -> S3Result<Vec<u8>> {
     let status = BackgroundHealStatus {
         info,
@@ -1172,6 +1215,7 @@ fn encode_background_heal_status(
         heal_active_tasks: heal_operations.active_tasks,
         heal_operations,
         cluster_status_complete,
+        coverage,
         progress,
     };
     serde_json::to_vec(&status).map_err(|e| {
@@ -1461,6 +1505,7 @@ impl Operation for BackgroundHealStatusHandler {
             cluster_status.operations,
             cluster_status.progress,
             cluster_status.complete,
+            &cluster_status.coverage,
         )?;
         info!(
             event = EVENT_ADMIN_RESPONSE_EMITTED,
@@ -1515,13 +1560,14 @@ impl Operation for ReplacementRecoveryStatusHandler {
 mod tests {
     use super::extract_heal_init_params;
     use super::{
-        BackgroundHealProgress, HealInitParams, HealResp, HealRuntimeState, aggregate_cluster_heal_status,
-        aggregate_replacement_recovery_cluster_status, background_heal_runtime_state, build_heal_channel_request,
-        build_replacement_recovery_status_response, encode_background_heal_status, encode_heal_control_path,
-        encode_heal_start_success, encode_heal_task_status, execute_after_heal_control_capability, heal_channel_response_items,
-        heal_channel_response_progress, heal_channel_response_summary, heal_control_response_id, json_response,
-        map_heal_response, merge_peer_heal_statuses, peer_topology_complete, query_peer_heal_status,
-        query_peer_replacement_recovery_status, reject_heal_admission, validate_heal_request_mode, validate_heal_target,
+        BackgroundHealCoverage, BackgroundHealCoverageReason, BackgroundHealProgress, HealInitParams, HealResp, HealRuntimeState,
+        aggregate_cluster_heal_status, aggregate_replacement_recovery_cluster_status, background_heal_runtime_state,
+        build_heal_channel_request, build_replacement_recovery_status_response, encode_background_heal_status,
+        encode_heal_control_path, encode_heal_start_success, encode_heal_task_status, execute_after_heal_control_capability,
+        heal_channel_response_items, heal_channel_response_progress, heal_channel_response_summary, heal_control_response_id,
+        json_response, map_heal_response, merge_peer_heal_statuses, peer_topology_complete, query_peer_heal_status,
+        query_peer_replacement_recovery_status, read_cluster_heal_status, reject_heal_admission, validate_heal_request_mode,
+        validate_heal_target,
     };
     use crate::storage::rpc::node_service::heal::{
         NodeHealProgress, NodeHealStatusSnapshot, NodeReplacementRecoveryStatusSnapshot, encode_node_replacement_recovery_status,
@@ -2175,7 +2221,13 @@ mod tests {
             ..Default::default()
         };
 
-        let encoded = encode_background_heal_status(&info, HealRuntimeState::Active, operations, None, true)
+        let coverage = BackgroundHealCoverage {
+            expected: 1,
+            responded: 1,
+            unknown: 0,
+            reasons: Default::default(),
+        };
+        let encoded = encode_background_heal_status(&info, HealRuntimeState::Active, operations, None, true, &coverage)
             .expect("background heal info should serialize");
         let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize");
 
@@ -2225,6 +2277,12 @@ mod tests {
             rustfs_heal::HealOperationsSnapshot::default(),
             Some(progress),
             true,
+            &BackgroundHealCoverage {
+                expected: 1,
+                responded: 1,
+                unknown: 0,
+                reasons: Default::default(),
+            },
         )
         .expect("background heal info should serialize");
         let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize");
@@ -2398,6 +2456,84 @@ mod tests {
         assert!(peer_topology_complete(1, 0, 0, 1, 0));
     }
 
+    #[tokio::test]
+    async fn test_background_heal_status_without_notification_preserves_local_snapshot() {
+        let initialized = rustfs_heal::heal_runtime_initialized();
+        let info = BackgroundHealInfo {
+            bitrot_start_cycle: 37,
+            current_scan_mode: HealScanMode::Deep,
+            ..Default::default()
+        };
+        for expected in [1, 3] {
+            let status = tokio::time::timeout(Duration::from_secs(1), read_cluster_heal_status(info.clone(), None, expected))
+                .await
+                .expect("local status must not wait for remote peers")
+                .expect("missing notification must retain the local snapshot");
+            assert_eq!(status.info.bitrot_start_cycle, 37);
+            assert_eq!(status.info.current_scan_mode, HealScanMode::Deep);
+            assert_eq!(status.complete, expected == 1);
+            assert_eq!(status.coverage.expected, expected);
+            assert_eq!(status.coverage.responded, 1);
+            assert_eq!(status.coverage.unknown, expected - 1);
+            if expected == 1 {
+                assert!(status.coverage.reasons.is_empty());
+            } else {
+                assert!(matches!(status.state, HealRuntimeState::Degraded | HealRuntimeState::Active));
+                assert_eq!(
+                    status.coverage.reasons,
+                    [BackgroundHealCoverageReason::NotificationSystemUnavailable].into()
+                );
+            }
+            let encoded = encode_background_heal_status(
+                &status.info,
+                status.state,
+                status.operations,
+                status.progress,
+                status.complete,
+                &status.coverage,
+            )
+            .expect("fallback status must encode");
+            let decoded: rustfs_madmin::client::BackgroundHealStatus =
+                serde_json::from_slice(&encoded).expect("the actual madmin client must decode the server response");
+            assert_eq!(decoded.cluster_status_complete, expected == 1);
+            let coverage = decoded.coverage.expect("new server supplies coverage");
+            assert_eq!(coverage.expected, Some(expected));
+            assert_eq!(coverage.responded, Some(1));
+            assert_eq!(coverage.unknown, Some(expected - 1));
+            if expected > 1 {
+                assert_eq!(coverage.reasons, ["notification_system_unavailable"]);
+            }
+        }
+        assert_eq!(
+            rustfs_heal::heal_runtime_initialized(),
+            initialized,
+            "reading status must not initialize heal"
+        );
+    }
+
+    #[test]
+    fn test_background_heal_status_coverage_reasons_are_bounded() {
+        let local = NodeHealStatusSnapshot::for_test(true, true, BackgroundHealInfo::default(), Default::default(), None);
+        let peers = (0..100)
+            .map(|index| {
+                if index % 2 == 0 {
+                    Ok(None)
+                } else {
+                    Err("peer unavailable".to_owned())
+                }
+            })
+            .collect();
+        let status = merge_peer_heal_statuses(vec![local], peers, 101, None).expect("local status remains available");
+        assert_eq!(status.coverage.responded, 1);
+        assert_eq!(status.coverage.unknown, 100);
+        assert_eq!(status.coverage.reasons.len(), 2);
+        let encoded = serde_json::to_vec(&status.coverage).expect("coverage encodes");
+        assert!(encoded.len() < 256, "coverage must not grow with peer failures");
+        let decoded: rustfs_madmin::client::BackgroundHealCoverage =
+            serde_json::from_slice(&encoded).expect("client coverage decodes");
+        assert_eq!(decoded.reasons, ["peer_status_unsupported", "peer_status_unavailable"]);
+    }
+
     #[test]
     fn test_peer_status_merge_degrades_explicitly_and_never_claims_idle() {
         let local = || {
@@ -2413,15 +2549,20 @@ mod tests {
         // but the safety property of the previous fail-closed behaviour is
         // preserved: the partial answer is labelled Degraded, never Idle, so
         // unknown peer work cannot be mistaken for "nothing is running".
-        let partial = merge_peer_heal_statuses(vec![local()], vec![Err("peer timeout".to_string())], 2, true)
+        let partial = merge_peer_heal_statuses(vec![local()], vec![Err("peer timeout".to_string())], 2, None)
             .expect("an unreachable peer degrades the answer instead of destroying it");
         assert!(!partial.complete);
         assert_eq!(partial.state, HealRuntimeState::Degraded);
+        assert_eq!(partial.coverage.expected, 2);
+        assert_eq!(partial.coverage.responded, 1);
+        assert_eq!(partial.coverage.unknown, 1);
+        assert_eq!(partial.coverage.reasons, [BackgroundHealCoverageReason::PeerStatusUnavailable].into());
 
-        let older_peer = merge_peer_heal_statuses(vec![local()], vec![Ok(None)], 2, true)
+        let older_peer = merge_peer_heal_statuses(vec![local()], vec![Ok(None)], 2, None)
             .expect("an older peer degrades the answer instead of destroying it");
         assert!(!older_peer.complete);
         assert_eq!(older_peer.state, HealRuntimeState::Degraded);
+        assert_eq!(older_peer.coverage.reasons, [BackgroundHealCoverageReason::PeerStatusUnsupported].into());
 
         let known_active = NodeHealStatusSnapshot::for_test(
             true,
@@ -2433,12 +2574,12 @@ mod tests {
             },
             None,
         );
-        let partial_active = merge_peer_heal_statuses(vec![known_active], vec![Ok(None)], 2, true)
+        let partial_active = merge_peer_heal_statuses(vec![known_active], vec![Ok(None)], 2, None)
             .expect("known active work may be reported as an explicit partial status");
         assert!(!partial_active.complete);
         assert_eq!(partial_active.state, HealRuntimeState::Active);
 
-        merge_peer_heal_statuses(Vec::new(), vec![Err("peer timeout".to_string())], 2, true)
+        merge_peer_heal_statuses(Vec::new(), vec![Err("peer timeout".to_string())], 2, None)
             .expect_err("no snapshot at all still fails closed");
     }
 
@@ -2459,12 +2600,22 @@ mod tests {
                 None,
             )
         };
-        let full_count_incomplete_topology = merge_peer_heal_statuses(vec![snapshot()], vec![Ok(Some(snapshot()))], 2, false)
-            .expect("incomplete topology degrades the answer instead of destroying it");
+        let full_count_incomplete_topology = merge_peer_heal_statuses(
+            vec![snapshot()],
+            vec![Ok(Some(snapshot()))],
+            2,
+            Some(BackgroundHealCoverageReason::PeerTopologyIncomplete),
+        )
+        .expect("incomplete topology degrades the answer instead of destroying it");
         assert!(!full_count_incomplete_topology.complete);
         assert_eq!(full_count_incomplete_topology.state, HealRuntimeState::Degraded);
+        assert_eq!(full_count_incomplete_topology.coverage.unknown, 0);
+        assert_eq!(
+            full_count_incomplete_topology.coverage.reasons,
+            [BackgroundHealCoverageReason::PeerTopologyIncomplete].into()
+        );
 
-        let full_count_complete_topology = merge_peer_heal_statuses(vec![snapshot()], vec![Ok(Some(snapshot()))], 2, true)
+        let full_count_complete_topology = merge_peer_heal_statuses(vec![snapshot()], vec![Ok(Some(snapshot()))], 2, None)
             .expect("complete topology and full count is a definitive answer");
         assert!(full_count_complete_topology.complete);
         assert_eq!(full_count_complete_topology.state, HealRuntimeState::Idle);
@@ -2478,6 +2629,12 @@ mod tests {
             rustfs_heal::HealOperationsSnapshot::default(),
             None,
             false,
+            &BackgroundHealCoverage {
+                expected: 2,
+                responded: 1,
+                unknown: 1,
+                reasons: [BackgroundHealCoverageReason::PeerStatusUnavailable].into(),
+            },
         )
         .expect("degraded status must serialize");
         let json: serde_json::Value = serde_json::from_slice(&encoded).expect("valid json");

rustfs/src/app/bucket_list_through.rs

@@ -26,8 +26,8 @@ use super::storage_api::bucket_usecase::ECStore;
 use super::storage_api::bucket_usecase::StorageObjectInfo as ObjectInfo;
 use super::storage_api::bucket_usecase::StorageObjectOptions;
 use super::storage_api::bucket_usecase::bucket::on_demand_migration::{
-    BucketOdmState, ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError, MergeSide,
-    OnDemandMigrationSys, SOURCE_LIST_MAX_RATE_WAIT, SourceClient, SourceError, SourceErrorPolicy, SourceListPlan,
+    BucketOdmState, ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError,
+    MergeSide, OnDemandMigrationSys, SOURCE_LIST_MAX_RATE_WAIT, SourceClient, SourceError, SourceErrorPolicy, SourceListPlan,
     SourceListRequest, SourceObject, SourcePage, decode_continuation_token, source_list_plan,
 };
 use super::storage_api::bucket_usecase::bucket::versioning_sys::BucketVersioningSys;
@@ -51,6 +51,9 @@ type ListObjectsV2Info = StorageListObjectsV2Info<ObjectInfo>;
 /// yet, so the only class RustFS can vouch for is the default one.
 const SOURCE_STORAGE_CLASS: &str = "STANDARD";
 
+/// Enable only after every node serving continuation requests can read v2.
+const ENV_LIST_PROGRESS_TOKENS: &str = "RUSTFS_ON_DEMAND_MIGRATION_LIST_V2_TOKENS";
+
 /// Concurrent local metadata probes when a versioned bucket has to check
 /// source-only keys for a shadowing delete marker.
 const DELETE_MARKER_PROBE_CONCURRENCY: usize = 32;
@@ -264,7 +267,18 @@ pub(crate) async fn merged_list_objects_v2(
         buffers[usize::from(fetch.side == MergeSide::Source)].extend(kept.into_iter().map(Some));
     }
 
-    let outcome = merger.finish();
+    let issue_progress_tokens = rustfs_utils::get_env_bool(ENV_LIST_PROGRESS_TOKENS, false);
+    let outcome = match merger.finish(issue_progress_tokens) {
+        Ok(outcome) => outcome,
+        Err(ListPageError::NoProgress(MergeSide::Source)) => {
+            degrade_or_fail(&mut merger, &mut degraded, policy.source_error, "invalid_pagination")?;
+            merger
+                .finish(issue_progress_tokens)
+                .map_err(|error| S3Error::with_message(S3ErrorCode::InternalError, error.to_string()))?
+        }
+        Err(error) => return Err(S3Error::with_message(S3ErrorCode::InternalError, error.to_string())),
+    };
+    drop(merger);
     let mut objects = Vec::with_capacity(outcome.picks.len());
     let mut prefixes = Vec::new();
     let mut source_only_keys = Vec::new();
@@ -430,7 +444,8 @@ mod tests {
     use crate::app::bucket_usecase::DefaultBucketUsecase;
     use crate::app::gating_test_env::{run_large_stack_test, shared_gating_ecstore};
     use crate::app::storage_api::bucket_usecase::bucket::on_demand_migration::{
-        FilterConfig, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig, SourceCredentials, TlsConfig,
+        FilterConfig, MAX_LIST_NO_PROGRESS_PAGES, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig,
+        SourceCredentials, TlsConfig,
     };
     use crate::app::storage_api::bucket_usecase::s3::{ListObjectsV2Input, ListObjectsV2Output, S3Request, S3Response};
     use crate::app::storage_api::test::StoragePutObjReader;
@@ -448,6 +463,7 @@ mod tests {
             source: Some("source-2".to_string()),
             source_done: false,
             last_key: Some("k".to_string()),
+            no_progress: None,
         }
     }
 
@@ -515,6 +531,24 @@ mod tests {
         assert!(matches!(local_cursor(Some(&encoded), decoded.as_ref()), LocalListCursor::Exhausted));
     }
 
+    #[test]
+    fn a_v2_token_keeps_the_local_cursor_when_list_through_is_turned_off() {
+        let mut resume = token(Some("local-2"), false);
+        resume.v = 2;
+        resume.no_progress = Some(MAX_LIST_NO_PROGRESS_PAGES - 1);
+        let encoded = resume.encode();
+        let decoded = decode_list_cursor(Some(&encoded)).expect("a v2 envelope decodes");
+        assert_eq!(decoded.as_ref(), Some(&resume));
+        assert!(matches!(
+            local_cursor(Some(&encoded), decoded.as_ref()),
+            LocalListCursor::Token(Some(local)) if local == "local-2"
+        ));
+        resume.local_done = true;
+        let encoded = resume.encode();
+        let decoded = decode_list_cursor(Some(&encoded)).expect("v2 with local EOF decodes");
+        assert!(matches!(local_cursor(Some(&encoded), decoded.as_ref()), LocalListCursor::Exhausted));
+    }
+
     #[test]
     fn a_plain_local_token_is_passed_through_and_a_tampered_one_is_rejected() {
         assert!(
@@ -551,14 +585,30 @@ mod tests {
     /// Serves exactly the scripted S3 pages and joins every connection before
     /// returning. A source retry or unexpected operation fails the test.
     async fn scripted_list_source(pages: Vec<String>) -> (String, tokio_util::task::AbortOnDropHandle<Vec<String>>) {
+        let (endpoint, server, _) = list_source(pages.into_iter()).await;
+        (endpoint, server)
+    }
+
+    async fn list_source(
+        pages: impl Iterator<Item = String> + Send + 'static,
+    ) -> (
+        String,
+        tokio_util::task::AbortOnDropHandle<Vec<String>>,
+        tokio_util::sync::CancellationToken,
+    ) {
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
             .await
             .expect("bind listing source");
         let address = listener.local_addr().expect("listing source address");
+        let stop = tokio_util::sync::CancellationToken::new();
+        let server_stop = stop.clone();
         let server = tokio::spawn(async move {
             let mut requests = Vec::new();
             for body in pages {
-                let (mut stream, _) = listener.accept().await.expect("accept source listing");
+                let (mut stream, _) = tokio::select! {
+                    _ = server_stop.cancelled() => break,
+                    accepted = listener.accept() => accepted.expect("accept source listing"),
+                };
                 let mut request = Vec::new();
                 let mut chunk = [0; 4096];
                 while !request.windows(4).any(|window| window == b"\r\n\r\n") {
@@ -588,7 +638,7 @@ mod tests {
             }
             requests
         });
-        (format!("http://{address}"), tokio_util::task::AbortOnDropHandle::new(server))
+        (format!("http://{address}"), tokio_util::task::AbortOnDropHandle::new(server), stop)
     }
 
     fn source_xml(next: Option<&str>, truncated: bool, key: Option<&str>) -> String {
@@ -616,12 +666,12 @@ mod tests {
         }
     }
 
-    async fn source_policy_request(
-        pages: Vec<String>,
+    async fn source_policy_input(
+        endpoint: String,
         policy: SourceErrorPolicy,
         resume_source: Option<&str>,
         filter_prefix: Option<&str>,
-    ) -> (S3Result<S3Response<ListObjectsV2Output>>, Vec<String>) {
+    ) -> (ListThroughTestState, ListObjectsV2Input) {
         let store = shared_gating_ecstore().await;
         crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await;
         let bucket = format!("odm-list-{}", uuid::Uuid::new_v4().simple());
@@ -638,9 +688,8 @@ mod tests {
             )
             .await
             .expect("seed real local listing");
-        let (endpoint, server) = scripted_list_source(pages).await;
         let sys = OnDemandMigrationSys::get();
-        let _state_guard = ListThroughTestState {
+        let state_guard = ListThroughTestState {
             bucket: bucket.clone(),
             module_enabled: sys.is_module_enabled(),
         };
@@ -685,6 +734,7 @@ mod tests {
                 source: Some(source.into()),
                 source_done: false,
                 last_key: None,
+                no_progress: None,
             };
             base64_simd::STANDARD.encode_to_string(token.encode().as_bytes())
         });
@@ -701,6 +751,10 @@ mod tests {
             request_payer: None,
             start_after: None,
         };
+        (state_guard, input)
+    }
+
+    async fn execute_source_list(input: ListObjectsV2Input) -> S3Result<S3Response<ListObjectsV2Output>> {
         let request = S3Request {
             input,
             method: http::Method::GET,
@@ -712,12 +766,23 @@ mod tests {
             service: None,
             trailing_headers: None,
         };
-        let result = tokio::time::timeout(
+        tokio::time::timeout(
             Duration::from_secs(10),
             DefaultBucketUsecase::from_global().execute_list_objects_v2(request),
         )
         .await
-        .expect("listing must complete within its bounded source budget");
+        .expect("listing must complete within its bounded source budget")
+    }
+
+    async fn source_policy_request(
+        pages: Vec<String>,
+        policy: SourceErrorPolicy,
+        resume_source: Option<&str>,
+        filter_prefix: Option<&str>,
+    ) -> (S3Result<S3Response<ListObjectsV2Output>>, Vec<String>) {
+        let (endpoint, server) = scripted_list_source(pages).await;
+        let (_state_guard, input) = source_policy_input(endpoint, policy, resume_source, filter_prefix).await;
+        let result = execute_source_list(input).await;
         let requests = tokio::time::timeout(Duration::from_secs(5), server)
             .await
             .expect("source connections must finish")
@@ -842,6 +907,244 @@ mod tests {
         });
     }
 
+    #[test]
+    #[serial_test::serial]
+    fn list_through_cross_request_empty_cursor_cycle_obeys_policy() {
+        run_large_stack_test("list-through-cross-request-cursor-cycle", || async {
+            temp_env::async_with_vars(
+                [
+                    (ENV_LIST_PROGRESS_TOKENS, Some("true")),
+                    ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
+                    ("HTTP_PROXY", None),
+                    ("HTTPS_PROXY", None),
+                    ("ALL_PROXY", None),
+                    ("http_proxy", None),
+                    ("https_proxy", None),
+                    ("all_proxy", None),
+                    ("NO_PROXY", Some("*")),
+                    ("no_proxy", Some("*")),
+                ],
+                async {
+                    for policy in [SourceErrorPolicy::Propagate, SourceErrorPolicy::NotFound] {
+                        let pages = ["B", "C", "A"].map(|next| source_xml(Some(next), true, None));
+                        let (endpoint, server, stop) = list_source(pages.into_iter().cycle()).await;
+                        let (_state_guard, mut input) = source_policy_input(endpoint, policy, Some("A"), None).await;
+                        let mut seen = std::collections::HashSet::from([input
+                            .continuation_token
+                            .clone()
+                            .expect("the first request resumes source cursor A")]);
+                        let mut client_requests = 0;
+                        let mut empty_pages = 0;
+                        let terminal = tokio::time::timeout(Duration::from_secs(30), async {
+                            loop {
+                                client_requests += 1;
+                                let response = match execute_source_list(input.clone()).await {
+                                    Ok(response) => response,
+                                    Err(error) => break Err(error),
+                                };
+                                if response.headers.contains_key("x-rustfs-on-demand-migration-list") {
+                                    break Ok(response);
+                                }
+                                let output = response.output;
+                                assert!(output.contents.as_ref().is_none_or(Vec::is_empty));
+                                assert!(output.common_prefixes.as_ref().is_none_or(Vec::is_empty));
+                                assert_eq!(output.key_count, Some(0));
+                                assert_eq!(output.is_truncated, Some(true));
+                                let next = output
+                                    .next_continuation_token
+                                    .expect("a truncated page must carry its cursor");
+                                assert!(
+                                    seen.insert(next.clone()),
+                                    "a cross-request source cursor cycle must not return an identical empty merged token"
+                                );
+                                empty_pages += 1;
+                                input.continuation_token = Some(next);
+                            }
+                        })
+                        .await
+                        .expect("a source cursor cycle must terminate within a bounded client pagination chain");
+                        assert_eq!(empty_pages, usize::from(MAX_LIST_NO_PROGRESS_PAGES - 1));
+                        assert_eq!(client_requests, usize::from(MAX_LIST_NO_PROGRESS_PAGES));
+                        assert_source_policy_result(terminal, policy);
+                        stop.cancel();
+                        let requests = tokio::time::timeout(Duration::from_secs(5), server)
+                            .await
+                            .expect("cyclic source server must stop")
+                            .expect("cyclic source server must not panic");
+                        assert_eq!(requests.len(), 2 * client_requests, "the sixteenth empty page exhausts the budget");
+                        for (index, request) in requests.iter().enumerate() {
+                            let source_cursor = ["A", "B", "C"][index % 3];
+                            assert!(
+                                request.contains(&format!("continuation-token={source_cursor}")),
+                                "the real SDK must follow the returned source cursor: {request}"
+                            );
+                        }
+                    }
+                },
+            )
+            .await;
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn list_through_default_rollout_continues_v2_without_issuing_it_from_v1() {
+        run_large_stack_test("list-through-reader-first-rollout", || async {
+            temp_env::async_with_vars(
+                [
+                    (ENV_LIST_PROGRESS_TOKENS, None),
+                    ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
+                    ("HTTP_PROXY", None),
+                    ("HTTPS_PROXY", None),
+                    ("ALL_PROXY", None),
+                    ("http_proxy", None),
+                    ("https_proxy", None),
+                    ("all_proxy", None),
+                    ("NO_PROXY", Some("*")),
+                    ("no_proxy", Some("*")),
+                ],
+                async {
+                    for policy in [SourceErrorPolicy::Propagate, SourceErrorPolicy::NotFound] {
+                        let pages = ["B", "C", "A"].map(|next| source_xml(Some(next), true, None));
+                        let (endpoint, server, stop) = list_source(pages.into_iter().cycle()).await;
+                        let (_state_guard, mut input) = source_policy_input(endpoint, policy, Some("A"), None).await;
+                        let original = input.continuation_token.clone();
+                        for _ in 0..3 {
+                            let response = execute_source_list(input.clone()).await.expect("reader-only v1 behavior");
+                            assert!(!response.headers.contains_key("x-rustfs-on-demand-migration-list"));
+                            assert_eq!(response.output.key_count, Some(0));
+                            assert_eq!(response.output.is_truncated, Some(true));
+                            let next = response.output.next_continuation_token.expect("resumable empty page");
+                            let raw = base64_simd::STANDARD.decode_to_vec(&next).expect("base64 continuation token");
+                            let decoded = std::str::from_utf8(&raw).expect("JSON token");
+                            let token = decode_list_cursor(Some(decoded)).expect("v1 reader").expect("merged token");
+                            assert_eq!(token.v, 1, "the default rollout cannot begin issuing v2");
+                            assert_eq!(token.no_progress, None);
+                            assert!(!decoded.contains("no_progress"), "ordinary v1 wire shape stays unchanged");
+                            input.continuation_token = Some(next);
+                        }
+                        assert_eq!(input.continuation_token, original, "default rollout retains the known v1 limitation");
+
+                        let raw = base64_simd::STANDARD
+                            .decode_to_vec(input.continuation_token.as_ref().expect("v1 token"))
+                            .expect("base64 continuation token");
+                        let mut token = decode_list_cursor(Some(std::str::from_utf8(&raw).expect("JSON token")))
+                            .expect("v1 reader")
+                            .expect("merged token");
+                        token.v = 2;
+                        token.no_progress = Some(MAX_LIST_NO_PROGRESS_PAGES - 2);
+                        input.continuation_token = Some(base64_simd::STANDARD.encode_to_string(token.encode().as_bytes()));
+                        let response = execute_source_list(input.clone()).await.expect("reader-only node resumes v2");
+                        assert_eq!(response.output.key_count, Some(0));
+                        assert_eq!(response.output.is_truncated, Some(true));
+                        let next = response.output.next_continuation_token.expect("last allowed empty cursor");
+                        let raw = base64_simd::STANDARD.decode_to_vec(&next).expect("base64 continuation token");
+                        let token = decode_list_cursor(Some(std::str::from_utf8(&raw).expect("JSON token")))
+                            .expect("v2 reader")
+                            .expect("merged token");
+                        assert_eq!(token.v, 2);
+                        assert_eq!(token.no_progress, Some(MAX_LIST_NO_PROGRESS_PAGES - 1));
+                        input.continuation_token = Some(next);
+                        assert_source_policy_result(execute_source_list(input).await, policy);
+                        stop.cancel();
+                        let requests = tokio::time::timeout(Duration::from_secs(5), server)
+                            .await
+                            .expect("cyclic source server must stop")
+                            .expect("source server must not panic");
+                        assert_eq!(requests.len(), 10, "five handler requests each fetched two source pages");
+                        for (index, request) in requests.iter().enumerate() {
+                            let cursor = ["A", "B", "C"][index % 3];
+                            assert!(request.contains(&format!("continuation-token={cursor}")), "{request}");
+                        }
+                    }
+                },
+            )
+            .await;
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn list_through_empty_advancing_pages_resume_across_handler_requests() {
+        run_large_stack_test("list-through-resumable-empty-pages", || async {
+            temp_env::async_with_vars(
+                [
+                    (ENV_LIST_PROGRESS_TOKENS, Some("true")),
+                    ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
+                    ("HTTP_PROXY", None),
+                    ("HTTPS_PROXY", None),
+                    ("ALL_PROXY", None),
+                    ("http_proxy", None),
+                    ("https_proxy", None),
+                    ("all_proxy", None),
+                    ("NO_PROXY", Some("*")),
+                    ("no_proxy", Some("*")),
+                ],
+                async {
+                    for filter_prefix in [None, Some("photos/2024/")] {
+                        let source_key = filter_prefix.map_or("a-source", |_| "photos/2024/a-source");
+                        let (endpoint, server) = scripted_list_source(vec![
+                            source_xml(Some("A"), true, None),
+                            source_xml(Some("B"), true, None),
+                            source_xml(Some("C"), true, None),
+                            source_xml(None, false, Some(source_key)),
+                        ])
+                        .await;
+                        let (_state_guard, mut input) =
+                            source_policy_input(endpoint, SourceErrorPolicy::Propagate, None, filter_prefix).await;
+                        let first = execute_source_list(input.clone())
+                            .await
+                            .expect("valid empty pages must remain resumable");
+                        assert!(!first.headers.contains_key("x-rustfs-on-demand-migration-list"));
+                        assert!(first.output.contents.as_ref().is_none_or(Vec::is_empty));
+                        assert!(first.output.common_prefixes.as_ref().is_none_or(Vec::is_empty));
+                        assert_eq!(first.output.key_count, Some(0));
+                        assert_eq!(first.output.is_truncated, Some(true));
+                        input.continuation_token = Some(first.output.next_continuation_token.expect("empty advancing cursor"));
+
+                        let second = execute_source_list(input)
+                            .await
+                            .expect("a progressing empty chain must reach its data");
+                        assert!(!second.headers.contains_key("x-rustfs-on-demand-migration-list"));
+                        let output = second.output;
+                        let objects = output
+                            .contents
+                            .unwrap_or_default()
+                            .into_iter()
+                            .map(|object| object.key.expect("listed object key"))
+                            .collect::<Vec<_>>();
+                        let prefixes = output
+                            .common_prefixes
+                            .unwrap_or_default()
+                            .into_iter()
+                            .map(|prefix| prefix.prefix.expect("listed common prefix"))
+                            .collect::<Vec<_>>();
+                        if filter_prefix.is_some() {
+                            assert_eq!(objects, vec!["z-local"]);
+                            assert_eq!(prefixes, vec!["photos/"]);
+                        } else {
+                            assert_eq!(objects, vec!["a-source", "z-local"]);
+                            assert!(prefixes.is_empty());
+                        }
+                        assert_eq!(output.key_count, Some(2));
+                        assert_eq!(output.is_truncated, Some(false));
+                        assert!(output.next_continuation_token.is_none());
+                        let requests = tokio::time::timeout(Duration::from_secs(5), server)
+                            .await
+                            .expect("finite source connections must finish")
+                            .expect("finite source server must not panic");
+                        assert_eq!(requests.len(), 4);
+                        assert!(!requests[0].contains("continuation-token="));
+                        for (request, cursor) in requests[1..].iter().zip(["A", "B", "C"]) {
+                            assert!(request.contains(&format!("continuation-token={cursor}")), "{request}");
+                        }
+                    }
+                },
+            )
+            .await;
+        });
+    }
+
     fn assert_source_policy_result(result: S3Result<S3Response<ListObjectsV2Output>>, policy: SourceErrorPolicy) {
         match policy {
             SourceErrorPolicy::Propagate => {

rustfs/src/app/object/shared.rs

@@ -309,7 +309,7 @@ fn classify_bucket_default_sse_lookup(
 ) -> S3Result<Option<(ServerSideEncryptionConfiguration, OffsetDateTime)>> {
     match lookup {
         Ok(config) => Ok(Some(config)),
-        Err(err) if err == StorageError::ConfigNotFound => Ok(None),
+        Err(StorageError::ConfigNotFound) => Ok(None),
         Err(err) => {
             let api_error = ApiError::from(err);
             error!(

rustfs/src/app/storage_api.rs

@@ -634,8 +634,8 @@ pub(crate) mod bucket {
         };
         #[cfg(test)]
         pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{
-            BREAKER_FAILURE_THRESHOLD, BreakerState, FilterConfig, OnDemandMigrationConfig, PathStyle, Provider, SourceConfig,
-            SourceCredentials, TlsConfig,
+            BREAKER_FAILURE_THRESHOLD, BreakerState, FilterConfig, MAX_LIST_NO_PROGRESS_PAGES, OnDemandMigrationConfig,
+            PathStyle, Provider, SourceConfig, SourceCredentials, TlsConfig,
         };
         pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{
             BucketOdmState, HeadPolicy, OdmLookup, OdmOp, OdmOutcome, OdmStateError, OnDemandMigrationSys, PolicyConfig,
@@ -643,8 +643,8 @@ pub(crate) mod bucket {
             commit_inline, idle_guarded_body,
         };
         pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{
-            ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError, MergeSide,
-            SOURCE_LIST_MAX_RATE_WAIT, SourceListPlan, decode_continuation_token, source_list_plan,
+            ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError,
+            MergeSide, SOURCE_LIST_MAX_RATE_WAIT, SourceListPlan, decode_continuation_token, source_list_plan,
         };
     }
 

rustfs/src/server/http.rs

@@ -208,6 +208,7 @@ const EVENT_PEER_ADDR_UNAVAILABLE: &str = "peer_addr_unavailable";
 const EVENT_RPC_SIGNATURE_VERIFICATION_FAILED: &str = "rpc_signature_verification_failed";
 const EVENT_GRPC_TRACE_CONTEXT_PROPAGATION_FAILED: &str = "grpc_trace_context_propagation_failed";
 const HEAL_CONTROL_TONIC_RPC_PATH: &str = "/node_service.HealControlService/HealControl";
+const SCANNER_SCOPED_DIRTY_USAGE_ACK_TONIC_RPC_PATH: &str = "/node_service.ScannerControlService/ScannerScopedDirtyUsageAck";
 const TIER_MUTATION_PREPARE_TONIC_RPC_PATH: &str = "/node_service.TierMutationControlService/PrepareTierMutation";
 const TIER_MUTATION_COMMIT_TONIC_RPC_PATH: &str = "/node_service.TierMutationControlService/CommitTierMutation";
 const TIER_MUTATION_ABORT_TONIC_RPC_PATH: &str = "/node_service.TierMutationControlService/AbortTierMutation";
@@ -1856,6 +1857,7 @@ fn process_connection(
         );
         let rpc_service = RpcRequestPathService::new(
             Routes::new(node_service)
+                .add_service(InterceptedService::new(storage::tonic_service::make_scanner_control_server(), check_auth))
                 .add_service(heal_control_service)
                 .add_service(tier_mutation_control_service)
                 .prepare(),
@@ -2259,6 +2261,7 @@ fn check_auth(req: Request<()>) -> std::result::Result<Request<()>, Status> {
         .strip_prefix(TONIC_RPC_PREFIX)
         .and_then(|suffix| suffix.strip_prefix('/'))
         .or_else(|| (target.uri.path() == HEAL_CONTROL_TONIC_RPC_PATH).then_some("HealControl"))
+        .or_else(|| (target.uri.path() == SCANNER_SCOPED_DIRTY_USAGE_ACK_TONIC_RPC_PATH).then_some("ScannerScopedDirtyUsageAck"))
         .or_else(|| (target.uri.path() == TIER_MUTATION_PREPARE_TONIC_RPC_PATH).then_some("PrepareTierMutation"))
         .or_else(|| (target.uri.path() == TIER_MUTATION_COMMIT_TONIC_RPC_PATH).then_some("CommitTierMutation"))
         .or_else(|| (target.uri.path() == TIER_MUTATION_ABORT_TONIC_RPC_PATH).then_some("AbortTierMutation"))
@@ -3427,6 +3430,50 @@ mod tests {
         rustfs_common::set_global_local_node_name(&previous_node_name).await;
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn scoped_dirty_usage_peer_probe_reaches_handler_through_production_auth() {
+        let _ = rustfs_credentials::set_global_rpc_secret("rpc-http-test-secret".to_string());
+        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind scoped ACK auth test");
+        let addr = listener.local_addr().expect("listener address");
+        let previous_node_name = rustfs_common::get_global_local_node_name().await;
+        rustfs_common::set_global_local_node_name(&addr.to_string()).await;
+        let node = InterceptedService::new(NodeServiceServer::new(make_server()), check_auth);
+        let scanner = InterceptedService::new(storage::tonic_service::make_scanner_control_server(), check_auth);
+        let service = RpcRequestPathService::new(Routes::new(node).add_service(scanner).prepare());
+        let server = tokio::spawn(async move {
+            let (socket, _) = listener.accept().await.expect("accept test connection");
+            ConnBuilder::new(TokioExecutor::new())
+                .serve_connection(TokioIo::new(socket), TowerToHyperService::new(service))
+                .await
+                .expect("serve scoped ACK auth test");
+        });
+        let host = rustfs_utils::XHost::try_from(addr.to_string()).expect("peer address");
+        let client = storage::PeerRestClient::new(host, format!("http://{addr}"));
+        let result = client
+            .scanner_scoped_dirty_usage_capability(
+                "11111111-1111-1111-1111-111111111111".to_string(),
+                "a".repeat(32),
+                vec![rustfs_protos::proto_gen::node_service::ScannerScopedDirtyUsageEntry {
+                    bucket: "photos".into(),
+                    bucket_incarnation: vec![1; 16].into(),
+                    generation: 8,
+                }],
+            )
+            .await;
+        client.evict_connection().await;
+        server.abort();
+        let _ = server.await;
+        rustfs_common::set_global_local_node_name(&previous_node_name).await;
+        let error = result
+            .expect_err("probe must fail closed without the requested storage owner")
+            .to_string();
+        assert!(
+            error.contains("storage layer is not initialized") || error.contains("scoped dirty usage peer or process changed"),
+            "signed probe must pass production path authentication and reach owner validation: {error}"
+        );
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn peer_rest_heal_control_uses_production_auth_and_keeps_validation_errors_online() {

rustfs/src/site_replication/tests.rs

@@ -867,7 +867,6 @@ fn test_retry_drain_bounds_each_peer_round_to_one_small_request_chain() {
             r#type: "tags".to_string(),
             ..Default::default()
         }],
-        ..Default::default()
     };
     let make = RetryDrainAction::BucketOpReplay {
         operation: SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING.to_string(),
@@ -973,7 +972,7 @@ fn test_lightweight_bucket_retry_plan_orders_real_metadata_and_counts_it() {
     operator_replication.rules.push(operator_rule("operator-backup"));
     let mut bucket_with_operator_rule = bucket;
     bucket_with_operator_rule.replication_config =
-        Some(BASE64_STANDARD.encode_to_string(&serialize(&operator_replication).expect("operator replication config")));
+        Some(BASE64_STANDARD.encode_to_string(serialize(&operator_replication).expect("operator replication config")));
     let plan = site_replication_bucket_retry_plan_from_info(&bucket_with_operator_rule, false).expect("targeted retry plan");
     assert!(
         plan.bucket_items.iter().any(|item| item.r#type == "replication-config"),
@@ -1050,7 +1049,7 @@ fn test_reachable_probe_promotion_is_fenced_by_the_observed_event() {
         .peers
         .insert("remote".to_string(), peer("remote", "https://remote.example.com"));
 
-    assert_eq!(mark_reachable_deferred_retry_events(&mut state, &[recovered.clone()]), 1);
+    assert_eq!(mark_reachable_deferred_retry_events(&mut state, std::slice::from_ref(&recovered)), 1);
     assert_eq!(state.retry_queue[0].updated_at, None);
     assert!(!state.retry_queue[0].peer_unreachable);
     assert_eq!(

rustfs/src/storage/rpc/node_service.rs

@@ -493,6 +493,13 @@ impl std::fmt::Debug for NodeService {
     }
 }
 
+pub(crate) fn make_scanner_control_server() -> scanner_control_service_server::ScannerControlServiceServer<NodeService> {
+    let limit = rustfs_protos::scoped_dirty_usage::SCOPED_DIRTY_USAGE_MAX_REQUEST_BYTES as usize;
+    scanner_control_service_server::ScannerControlServiceServer::new(make_server())
+        .max_decoding_message_size(limit)
+        .max_encoding_message_size(limit)
+}
+
 pub fn make_server() -> NodeService {
     let context = runtime_sources::current_app_context();
     make_server_for_context(context)
@@ -1087,6 +1094,74 @@ impl NodeService {
     }
 }
 
+#[tonic::async_trait]
+impl scanner_control_service_server::ScannerControlService for NodeService {
+    async fn scanner_scoped_dirty_usage_ack(
+        &self,
+        request: Request<ScannerScopedDirtyUsageAckRequest>,
+    ) -> Result<Response<ScannerScopedDirtyUsageAckResponse>, Status> {
+        use rustfs_protos::scoped_dirty_usage::*;
+        static ADMISSION: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(4);
+
+        let canonical =
+            canonical_scoped_dirty_usage_request(request.get_ref()).map_err(|err| Status::invalid_argument(err.to_string()))?;
+        verify_tonic_canonical_body_digest(&request, &canonical)
+            .map_err(|_| Status::permission_denied("scoped dirty usage authentication failed"))?;
+        let _admission = ADMISSION
+            .try_acquire()
+            .map_err(|_| Status::resource_exhausted("scoped dirty usage receiver is busy"))?;
+        let request = request.into_inner();
+        let store = self
+            .resolve_object_store()
+            .ok_or_else(|| Status::unavailable("storage layer is not initialized"))?;
+        if store.id.is_nil()
+            || request.owner_id != store.id.to_string()
+            || request.instance_id != rustfs_scanner::scanner_activity_epoch()
+        {
+            return Err(Status::failed_precondition("scoped dirty usage peer or process changed"));
+        }
+        let cleared = timeout(Duration::from_secs(30), async {
+            // Strict bucket order is validated before admission. Acquire every
+            // lifecycle/metadata fence before clearing any dirty record.
+            let mut guards = Vec::with_capacity(request.entries.len());
+            for entry in &request.entries {
+                let incarnation = Uuid::from_slice(entry.bucket_incarnation.as_ref())
+                    .map_err(|_| Status::invalid_argument("invalid bucket incarnation"))?;
+                let guard =
+                    crate::storage::storage_api::acquire_scanner_bucket_incarnation_fence(&entry.bucket, incarnation, store.id)
+                        .await
+                        .map_err(|_| Status::failed_precondition("trusted bucket incarnation is unavailable"))?;
+                guards.push(guard);
+            }
+            let entries = guards
+                .iter()
+                .zip(&request.entries)
+                .map(|(guard, entry)| (guard, entry.generation))
+                .collect::<Vec<_>>();
+            rustfs_scanner::acknowledge_scoped_dirty_usage(&request.instance_id, &entries, request.probe_only)
+                .map_err(|err| Status::failed_precondition(err.to_string()))
+        })
+        .await
+        .map_err(|_| Status::deadline_exceeded("scoped dirty usage incarnation validation timed out"))??;
+        let mut response = ScannerScopedDirtyUsageAckResponse {
+            protocol_version: SCOPED_DIRTY_USAGE_PROTOCOL_VERSION,
+            owner_id: request.owner_id,
+            instance_id: request.instance_id,
+            supported: true,
+            max_entries: SCOPED_DIRTY_USAGE_MAX_ENTRIES,
+            max_request_bytes: SCOPED_DIRTY_USAGE_MAX_REQUEST_BYTES,
+            cleared,
+            response_proof: Bytes::new(),
+        };
+        let body = canonical_scoped_dirty_usage_response(&canonical, &response)
+            .map_err(|_| Status::internal("scoped dirty usage response is too large"))?;
+        response.response_proof = sign_tonic_rpc_response_proof(&body)
+            .map_err(|_| Status::unavailable("scoped dirty usage response authentication is unavailable"))?
+            .into();
+        Ok(Response::new(response))
+    }
+}
+
 #[tonic::async_trait]
 impl Node for NodeService {
     async fn ping(&self, request: Request<PingRequest>) -> Result<Response<PingResponse>, Status> {
@@ -2623,6 +2698,7 @@ mod tests {
     use rustfs_kms::KmsServiceManager;
     use rustfs_protos::CanonicalMutationBody as _;
     use rustfs_protos::models::PingBodyBuilder;
+    use rustfs_protos::proto_gen::node_service::scanner_control_service_server::ScannerControlService as _;
     use rustfs_protos::proto_gen::node_service::{
         BackgroundHealStatusRequest, BatchGenerallyLockRequest, CancelDecommissionRequest, CheckPartsRequest,
         ClearDecommissionRequest, ControlPlaneErrorCode, DeleteBucketMetadataRequest, DeleteBucketRequest, DeletePathsRequest,
@@ -5990,6 +6066,74 @@ mod tests {
         );
     }
 
+    fn scoped_dirty_usage_request() -> rustfs_protos::proto_gen::node_service::ScannerScopedDirtyUsageAckRequest {
+        rustfs_protos::proto_gen::node_service::ScannerScopedDirtyUsageAckRequest {
+            challenge: vec![7; 16].into(),
+            protocol_version: 1,
+            owner_id: "11111111-1111-1111-1111-111111111111".into(),
+            instance_id: "a".repeat(32),
+            scope: 1,
+            probe_only: false,
+            entries: vec![rustfs_protos::proto_gen::node_service::ScannerScopedDirtyUsageEntry {
+                bucket: "photos".into(),
+                bucket_incarnation: vec![1; 16].into(),
+                generation: 8,
+            }],
+        }
+    }
+
+    #[tokio::test]
+    async fn scoped_dirty_usage_authenticates_before_storage_and_rejects_tampering() {
+        use rustfs_protos::scoped_dirty_usage::canonical_scoped_dirty_usage_request;
+        let service = create_test_node_service();
+        let unsigned = service
+            .scanner_scoped_dirty_usage_ack(Request::new(scoped_dirty_usage_request()))
+            .await
+            .expect_err("unsigned ACK must not access storage");
+        assert_eq!(unsigned.code(), tonic::Code::PermissionDenied);
+        for field in 0..9 {
+            let mut signed = Request::new(scoped_dirty_usage_request());
+            let canonical = canonical_scoped_dirty_usage_request(signed.get_ref()).expect("canonical request");
+            set_tonic_canonical_body_digest(&mut signed, &canonical).expect("digest");
+            mark_v2_authenticated(&mut signed);
+            match field {
+                0 => signed.get_mut().challenge = vec![3; 16].into(),
+                1 => signed.get_mut().owner_id = "22222222-2222-2222-2222-222222222222".into(),
+                2 => signed.get_mut().instance_id = "b".repeat(32),
+                3 => signed.get_mut().probe_only = true,
+                4 => signed.get_mut().entries[0].bucket = "videos".into(),
+                5 => signed.get_mut().entries[0].bucket_incarnation = vec![2; 16].into(),
+                6 => signed.get_mut().entries[0].generation += 1,
+                7 => signed.get_mut().scope += 1,
+                _ => signed.get_mut().protocol_version += 1,
+            }
+            let error = service
+                .scanner_scoped_dirty_usage_ack(signed)
+                .await
+                .expect_err("tampered ACK must fail");
+            assert_eq!(
+                error.code(),
+                if field < 7 {
+                    tonic::Code::PermissionDenied
+                } else {
+                    tonic::Code::InvalidArgument
+                }
+            );
+        }
+        let mut signed = Request::new(scoped_dirty_usage_request());
+        let canonical = canonical_scoped_dirty_usage_request(signed.get_ref()).expect("canonical request");
+        set_tonic_canonical_body_digest(&mut signed, &canonical).expect("digest");
+        mark_v2_authenticated(&mut signed);
+        assert_eq!(
+            service
+                .scanner_scoped_dirty_usage_ack(signed)
+                .await
+                .expect_err("missing owner cannot advertise capability")
+                .code(),
+            tonic::Code::Unavailable
+        );
+    }
+
     #[tokio::test]
     async fn test_scanner_activity_requires_body_bound_auth_before_storage_lookup() {
         let service = create_test_node_service();
@@ -6485,6 +6629,62 @@ mod tests {
         )
     }
 
+    #[tokio::test]
+    async fn scoped_dirty_usage_transport_rejects_oversized_unknown_and_duplicate_fields() {
+        let listener = TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind scoped ACK transport test");
+        let addr = listener.local_addr().expect("test listener address");
+        let (shutdown, stopped) = tokio::sync::oneshot::channel();
+        let server = tokio::spawn(async move {
+            tonic::transport::Server::builder()
+                .add_service(super::make_scanner_control_server())
+                .serve_with_incoming_shutdown(TcpListenerStream::new(listener), async {
+                    let _ = stopped.await;
+                })
+                .await
+                .expect("scoped ACK transport server");
+        });
+        let client = reqwest::Client::builder()
+            .no_proxy()
+            .http2_prior_knowledge()
+            .build()
+            .expect("HTTP/2 client");
+        let limit = rustfs_protos::scoped_dirty_usage::SCOPED_DIRTY_USAGE_MAX_REQUEST_BYTES as usize;
+        for tag in [0x78, 0x0a] {
+            // Unknown varint field 15, or repeated empty singular challenge:
+            // both decode to a tiny default struct despite the large wire body.
+            for oversized in [false, true] {
+                let mut payload = [tag, 0].repeat(if oversized { (limit - 4) / 2 } else { limit / 2 });
+                if oversized {
+                    // Unknown fixed32 field 15 makes a valid cap+1 protobuf.
+                    payload.extend_from_slice(&[0x7d, 0, 0, 0, 0]);
+                }
+                assert_eq!(payload.len(), limit + usize::from(oversized));
+                let mut frame = vec![0];
+                frame.extend_from_slice(&u32::try_from(payload.len()).expect("bounded test payload").to_be_bytes());
+                frame.extend_from_slice(&payload);
+                let response = client
+                    .post(format!("http://{addr}/node_service.ScannerControlService/ScannerScopedDirtyUsageAck"))
+                    .header("content-type", "application/grpc")
+                    .header("te", "trailers")
+                    .body(frame)
+                    .send()
+                    .await
+                    .expect("send raw protobuf frame");
+                let status = response.headers().get("grpc-status").expect("gRPC failure status");
+                assert_eq!(
+                    status.to_str().expect("status text"),
+                    if oversized { "11" } else { "3" },
+                    "cap+1 must fail in the codec, while cap bytes reach request validation"
+                );
+            }
+        }
+        drop(client);
+        shutdown.send(()).expect("stop test server");
+        server.await.expect("join test server");
+    }
+
     #[tokio::test]
     async fn heal_control_transport_enforces_codec_limit_and_fails_closed() {
         let Some(mut client) = connect_test_heal_control_client().await else {

rustfs/src/storage/storage_api.rs

@@ -379,7 +379,7 @@ pub(crate) mod tonic_service_consumer {
     #[cfg(test)]
     pub(crate) use super::super::tonic_service::{heal_topology_fingerprint, make_heal_control_server_for_source};
     pub(crate) use super::super::tonic_service::{
-        make_heal_control_server_with_cache, make_server, make_tier_mutation_control_server,
+        make_heal_control_server_with_cache, make_scanner_control_server, make_server, make_tier_mutation_control_server,
     };
 }
 
@@ -1704,6 +1704,14 @@ pub(crate) async fn acquire_bucket_metadata_transaction_lock(
     ecstore_bucket::metadata_sys::acquire_bucket_metadata_transaction_lock(bucket).await
 }
 
+pub(crate) async fn acquire_scanner_bucket_incarnation_fence(
+    bucket: &str,
+    incarnation: uuid::Uuid,
+    owner_id: uuid::Uuid,
+) -> Result<ecstore_bucket::metadata_sys::BucketMetadataMutationGuard> {
+    ecstore_bucket::metadata_sys::acquire_scanner_bucket_incarnation_fence(bucket, incarnation, owner_id).await
+}
+
 pub(crate) async fn update_bucket_targets_under_transaction_lock(
     guard: &ecstore_bucket::metadata_sys::BucketMetadataMutationGuard,
     bucket: &str,

rustfs/src/storage/tonic_service.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 pub(crate) use crate::storage::rpc::node_service::make_heal_control_server_with_cache;
+pub(crate) use crate::storage::rpc::node_service::make_scanner_control_server;
 #[cfg(test)]
 pub(crate) use crate::storage::rpc::node_service::{heal::heal_topology_fingerprint, make_heal_control_server_for_source};
 pub use crate::storage::rpc::{make_heal_control_server, make_server, make_tier_mutation_control_server};

rustfs/src/storage_api.rs

@@ -176,7 +176,7 @@ pub(crate) mod server {
                 heal_topology_fingerprint, make_heal_control_server_for_source,
             };
             pub(crate) use crate::storage::storage_api::tonic_service_consumer::{
-                make_heal_control_server_with_cache, make_server, make_tier_mutation_control_server,
+                make_heal_control_server_with_cache, make_scanner_control_server, make_server, make_tier_mutation_control_server,
             };
         }
     }

scripts/check_scheduled_validation_freshness.py

@@ -1,10 +1,11 @@
 #!/usr/bin/env python3
-"""Fail when a critical scheduled validation has not started recently."""
+"""Require recent scheduled attempts and completed successes on the default branch."""
 
 from __future__ import annotations
 
 import argparse
 from datetime import datetime, timedelta, timezone
+import io
 import json
 import os
 from pathlib import Path
@@ -13,7 +14,7 @@ import sys
 import tempfile
 import unittest
 from unittest import mock
-from urllib.parse import quote, urlencode
+from urllib.parse import parse_qs, quote, urlencode, urlsplit
 from urllib.request import Request, urlopen
 
 
@@ -75,15 +76,8 @@ def stale_reason(
     run: dict[str, object] | None,
     now: datetime,
     max_age_hours: int,
-    never_ran_grace_until: datetime | None = None,
 ) -> str | None:
     if run is None:
-        # The grace deadline only covers a workflow whose first scheduled slot
-        # has not arrived yet (for example a monthly cron enabled mid-month).
-        # A recorded-but-old run proves the schedule used to fire and stopped,
-        # so the grace never masks that case.
-        if never_ran_grace_until is not None and now <= never_ran_grace_until:
-            return None
         return "no scheduled run has been recorded"
     created_at = parse_timestamp(run.get("created_at"))
     age = now - created_at
@@ -93,14 +87,23 @@ def stale_reason(
 
 
 def fetch_latest_scheduled_run(
-    repository: str, workflow: str, token: str, api_url: str
+    repository: str,
+    workflow: str,
+    token: str,
+    api_url: str,
+    default_branch: str,
+    successful: bool = False,
 ) -> dict[str, object] | None:
     owner, repo = repository.split("/", 1)
     workflow_name = Path(workflow).name
+    query = {"event": "schedule", "branch": default_branch, "per_page": 1}
+    if successful:
+        # Filter on the server: the last success may be beyond a page of failures.
+        query["status"] = "success"
     endpoint = (
         f"{api_url.rstrip('/')}/repos/{quote(owner, safe='')}/{quote(repo, safe='')}"
         f"/actions/workflows/{quote(workflow_name, safe='')}/runs?"
-        + urlencode({"event": "schedule", "per_page": 1})
+        + urlencode(query)
     )
     request = Request(
         endpoint,
@@ -110,57 +113,104 @@ def fetch_latest_scheduled_run(
             "X-GitHub-Api-Version": "2022-11-28",
         },
     )
-    with urlopen(request, timeout=30) as response:
+    # Two requests per manifest entry must fit the watchdog's ten-minute job.
+    with urlopen(request, timeout=15) as response:
         payload = json.load(response)
-    runs = payload.get("workflow_runs")
+    runs = payload.get("workflow_runs") if isinstance(payload, dict) else None
     if not isinstance(runs, list):
         raise ValueError(f"GitHub returned no workflow_runs list for {workflow}")
+    total_count = payload.get("total_count")
+    if not isinstance(total_count, int) or isinstance(total_count, bool) or total_count < len(runs):
+        raise ValueError(f"GitHub returned an invalid run count for {workflow}")
     if not runs:
+        if total_count:
+            raise ValueError(f"GitHub returned an empty first page with recorded runs for {workflow}")
         return None
-    if not isinstance(runs[0], dict):
+    run = runs[0]
+    if not isinstance(run, dict):
         raise ValueError(f"GitHub returned an invalid workflow run for {workflow}")
-    return runs[0]
+    if run.get("event") != "schedule" or run.get("head_branch") != default_branch:
+        raise ValueError(f"GitHub returned a run outside the scheduled default-branch query for {workflow}")
+    if not isinstance(run.get("status"), str) or not run["status"]:
+        raise ValueError(f"GitHub returned no run status for {workflow}")
+    conclusion = run.get("conclusion")
+    if (conclusion is not None and not isinstance(conclusion, str)) or (
+        run["status"] == "completed" and not conclusion
+    ):
+        raise ValueError(f"GitHub returned an invalid run conclusion for {workflow}")
+    if successful and (run["status"] != "completed" or conclusion != "success"):
+        raise ValueError(f"GitHub returned a run without a completed success for {workflow}")
+    parse_timestamp(run.get("created_at"))
+    if not isinstance(run.get("html_url"), str) or not run["html_url"]:
+        raise ValueError(f"GitHub returned no run URL for {workflow}")
+    return run
 
 
-def write_report(path: Path, failures: list[tuple[str, int, str, str]]) -> None:
-    lines = ["## Scheduled validation freshness"]
-    if not failures:
-        lines.append("")
-        lines.append("All critical scheduled validations have a recent scheduled run.")
-    else:
-        lines.extend(
-            [
-                "",
-                "The following critical validations are stale or could not be inspected:",
-                "",
-                "| Workflow | Limit | Result | Last run |",
-                "| --- | ---: | --- | --- |",
-            ]
-        )
-        for workflow, max_age_hours, reason, run_url in failures:
-            link = f"[open]({run_url})" if run_url else "—"
-            lines.append(f"| `{workflow}` | {max_age_hours}h | {reason} | {link} |")
+def describe_run(run: dict[str, object] | None) -> str:
+    if run is None:
+        return "No recorded run"
+    outcome = run["status"]
+    if run.get("conclusion"):
+        outcome = f"{outcome}/{run['conclusion']}"
+    return f"[{outcome}]({run['html_url']}) — created {run['created_at']}"
+
+
+def write_report(path: Path, rows: list[tuple[str, int, str, str, str]], default_branch: str) -> None:
+    lines = [
+        "## Scheduled validation freshness",
+        "",
+        f"Default branch: `{default_branch}`. Ages use scheduled-run creation time; rerunning an old commit does not refresh its evidence.",
+        "Attempt outcomes are shown independently of successful-run freshness.",
+        "Success is the GitHub workflow run conclusion; suite completeness remains the responsibility of each workflow.",
+        "",
+        "| Workflow | Limit | Freshness | Last attempt | Last completed success |",
+        "| --- | ---: | --- | --- | --- |",
+    ]
+    for workflow, max_age_hours, result, attempt, success in rows:
+        cells = [f"`{workflow}`", f"{max_age_hours}h", result, attempt, success]
+        lines.append("| " + " | ".join(cell.replace("|", "\\|").replace("\n", " ") for cell in cells) + " |")
     path.write_text("\n".join(lines) + "\n")
 
 
 def check_freshness(
-    config: Path, report: Path, repository: str, token: str, api_url: str
+    config: Path, report: Path, repository: str, token: str, api_url: str, default_branch: str
 ) -> int:
     now = datetime.now(timezone.utc)
-    failures: list[tuple[str, int, str, str]] = []
+    rows: list[tuple[str, int, str, str, str]] = []
+    failed = False
     for workflow, max_age_hours, never_ran_grace_until in load_validations(config):
-        try:
-            run = fetch_latest_scheduled_run(repository, workflow, token, api_url)
-            reason = stale_reason(run, now, max_age_hours, never_ran_grace_until)
-            if reason is not None:
-                run_url = str(run.get("html_url", "")) if run else ""
-                failures.append((workflow, max_age_hours, reason, run_url))
-        except Exception as error:
-            failures.append(
-                (workflow, max_age_hours, f"inspection failed: {error}", "")
-            )
-    write_report(report, failures)
-    return 1 if failures else 0
+        runs: dict[str, dict[str, object] | None] = {}
+        reasons: list[str] = []
+        for label, successful in (("Last attempt", False), ("Last completed success", True)):
+            try:
+                runs[label] = fetch_latest_scheduled_run(
+                    repository, workflow, token, api_url, default_branch, successful
+                )
+            except Exception as error:
+                reasons.append(f"{label}: inspection failed: {error}")
+        # A failed inspection or any recorded attempt ends first-run grace.
+        initial_grace = (
+            len(runs) == 2
+            and all(run is None for run in runs.values())
+            and never_ran_grace_until is not None
+            and now <= never_ran_grace_until
+        )
+        if not initial_grace:
+            for label, run in runs.items():
+                reason = stale_reason(run, now, max_age_hours)
+                if reason is not None:
+                    reasons.append(f"{label}: {reason}")
+        failed |= bool(reasons)
+        result = "; ".join(reasons) if reasons else "Fresh"
+        if initial_grace:
+            result = f"Initial grace until {never_ran_grace_until.isoformat()}"
+        evidence = [
+            describe_run(runs[label]) if label in runs else "Inspection failed"
+            for label in ("Last attempt", "Last completed success")
+        ]
+        rows.append((workflow, max_age_hours, result, *evidence))
+    write_report(report, rows, default_branch)
+    return 1 if failed else 0
 
 
 class SelfTests(unittest.TestCase):
@@ -173,15 +223,6 @@ class SelfTests(unittest.TestCase):
         self.assertIsNotNone(stale_reason(past_limit, self.NOW, 36))
         self.assertIsNotNone(stale_reason(None, self.NOW, 36))
 
-    def test_never_ran_grace_only_covers_missing_runs(self) -> None:
-        future_grace = self.NOW + timedelta(hours=1)
-        past_grace = self.NOW - timedelta(seconds=1)
-        self.assertIsNone(stale_reason(None, self.NOW, 36, future_grace))
-        self.assertIsNone(stale_reason(None, self.NOW, 36, self.NOW))
-        self.assertIsNotNone(stale_reason(None, self.NOW, 36, past_grace))
-        stale_run = {"created_at": "2026-08-20T23:59:59Z"}
-        self.assertIsNotNone(stale_reason(stale_run, self.NOW, 36, future_grace))
-
     def test_config_rejects_duplicate_and_invalid_entries(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             path = Path(tmp) / "validations.json"
@@ -238,63 +279,205 @@ class SelfTests(unittest.TestCase):
                 ],
             )
 
-    def test_check_reports_missing_runs(self) -> None:
+    @staticmethod
+    def run_fixture(**overrides: object) -> dict[str, object]:
+        return {
+            "status": "completed",
+            "conclusion": "success",
+            "event": "schedule",
+            "head_branch": "release/current",
+            "created_at": "2026-08-22T00:00:00Z",
+            "html_url": "https://github.test/rustfs/rustfs/actions/runs/1",
+            **overrides,
+        }
+
+    def check_payloads(
+        self, payloads: list[object], *, grace: str | None = None, workflows: int = 1
+    ) -> tuple[int, str, list]:
         with tempfile.TemporaryDirectory() as tmp:
             root = Path(tmp)
             config = root / "validations.json"
             report = root / "report.md"
-            config.write_text(
-                json.dumps(
-                    [
-                        {"workflow": ".github/workflows/ci.yml", "max_age_hours": 36},
-                        {"workflow": ".github/workflows/fuzz.yml", "max_age_hours": 36},
-                        {"workflow": ".github/workflows/mint.yml", "max_age_hours": 36},
-                    ]
-                )
-            )
-            with mock.patch(
-                __name__ + ".fetch_latest_scheduled_run",
-                side_effect=[
-                    {"created_at": "2999-01-01T00:00:00Z"},
-                    None,
-                    RuntimeError("API unavailable"),
-                ],
+            entries = [
+                {"workflow": f".github/workflows/check-{index}.yml", "max_age_hours": 36}
+                for index in range(workflows)
+            ]
+            if grace is not None:
+                entries[0]["never_ran_grace_until"] = grace
+            config.write_text(json.dumps(entries))
+            responses = []
+            for payload in payloads:
+                if isinstance(payload, dict) and isinstance(payload.get("workflow_runs"), list):
+                    payload = {"total_count": len(payload["workflow_runs"]), **payload}
+                responses.append(payload if isinstance(payload, Exception) else io.StringIO(json.dumps(payload)))
+            with (
+                mock.patch(__name__ + ".urlopen", side_effect=responses) as request,
+                mock.patch(__name__ + ".datetime", wraps=datetime) as clock,
             ):
-                self.assertEqual(
-                    check_freshness(
-                        config,
-                        report,
-                        "rustfs/rustfs",
-                        "token",
-                        "https://api.github.test",
-                    ),
-                    1,
+                clock.now.return_value = self.NOW
+                status = check_freshness(
+                    config, report, "rustfs/rustfs", "test-token",
+                    "https://api.github.test", "release/current",
                 )
-            contents = report.read_text()
-            self.assertIn(".github/workflows/fuzz.yml", contents)
-            self.assertIn("inspection failed: API unavailable", contents)
-            self.assertNotIn(".github/workflows/ci.yml`", contents)
+            return status, report.read_text(), request.call_args_list
 
-            config.write_text(
-                json.dumps(
-                    [{"workflow": ".github/workflows/ci.yml", "max_age_hours": 36}]
+    def test_requests_filter_schedule_default_branch_and_success_on_server(self) -> None:
+        attempt = self.run_fixture(status="in_progress", conclusion=None)
+        success = self.run_fixture(html_url="https://github.test/rustfs/rustfs/actions/runs/2")
+        status, report, calls = self.check_payloads([
+            {"workflow_runs": [attempt], "total_count": 1001},
+            {"workflow_runs": [success], "total_count": 1},
+        ])
+        self.assertEqual(status, 0)
+        self.assertEqual(len(calls), 2)
+        for call, successful in zip(calls, (False, True)):
+            request = call.args[0]
+            url = urlsplit(request.full_url)
+            self.assertEqual(url.path, "/repos/rustfs/rustfs/actions/workflows/check-0.yml/runs")
+            expected = {"event": ["schedule"], "branch": ["release/current"], "per_page": ["1"]}
+            if successful:
+                expected["status"] = ["success"]
+            self.assertEqual(parse_qs(url.query), expected)
+            self.assertEqual(request.get_header("Authorization"), "Bearer test-token")
+            self.assertEqual(call.kwargs, {"timeout": 15})
+        self.assertIn("[in_progress]", report)
+        self.assertIn(str(attempt["html_url"]), report)
+        self.assertIn(str(success["html_url"]), report)
+
+    def test_cancelled_attempt_cannot_refresh_expired_success(self) -> None:
+        attempt = self.run_fixture(conclusion="cancelled")
+        success = self.run_fixture(
+            created_at="2026-08-20T23:59:59Z", updated_at="2026-08-22T11:59:59Z",
+            html_url="https://github.test/rustfs/rustfs/actions/runs/2",
+        )
+        status, report, _ = self.check_payloads([
+            {"workflow_runs": [attempt]}, {"workflow_runs": [success]},
+        ])
+        self.assertEqual(status, 1)
+        self.assertIn("Last completed success: last scheduled run is", report)
+        self.assertIn("[completed/cancelled]", report)
+        for run in (attempt, success):
+            self.assertIn(str(run["html_url"]), report)
+            self.assertIn(str(run["created_at"]), report)
+
+    def test_attempt_outcome_does_not_replace_recent_success(self) -> None:
+        success = self.run_fixture(created_at="2026-08-21T00:00:00Z")
+        for state, conclusion in (
+            ("completed", "failure"), ("completed", "cancelled"),
+            ("completed", "timed_out"), ("completed", "success"),
+            ("queued", None), ("in_progress", None),
+        ):
+            with self.subTest(state=state, conclusion=conclusion):
+                status, report, _ = self.check_payloads([
+                    {"workflow_runs": [self.run_fixture(status=state, conclusion=conclusion)]},
+                    {"workflow_runs": [success]},
+                ])
+                self.assertEqual(status, 0)
+                self.assertIn(f"[{state}" + (f"/{conclusion}" if conclusion else "") + "]", report)
+                self.assertIn("Fresh", report)
+                self.assertNotIn("All critical scheduled validations", report)
+
+    def test_grace_requires_two_successful_queries_with_no_history(self) -> None:
+        for attempt, success, grace, expected in (
+            (None, None, "2026-08-22T12:00:00Z", 0),
+            (None, None, "2026-08-22T11:59:59Z", 1),
+            (self.run_fixture(conclusion="failure"), None, "2026-08-23T00:00:00Z", 1),
+            (self.run_fixture(status="queued", conclusion=None), None, "2026-08-23T00:00:00Z", 1),
+            (None, self.run_fixture(), "2026-08-23T00:00:00Z", 1),
+        ):
+            with self.subTest(attempt=attempt, success=success, grace=grace):
+                status, report, _ = self.check_payloads([
+                    {"workflow_runs": [] if attempt is None else [attempt]},
+                    {"workflow_runs": [] if success is None else [success]},
+                ], grace=grace)
+                self.assertEqual(status, expected)
+                self.assertEqual("Initial grace until" in report, expected == 0)
+
+    def test_api_failures_preserve_other_evidence_and_never_enter_grace(self) -> None:
+        good = {"workflow_runs": [self.run_fixture()]}
+        for first, second in (
+            (RuntimeError("API unavailable"), good),
+            (good, RuntimeError("API unavailable")),
+            (RuntimeError("API unavailable"), {"workflow_runs": []}),
+        ):
+            with self.subTest(first=first, second=second):
+                status, report, calls = self.check_payloads(
+                    [first, second], grace="2026-08-23T00:00:00Z"
                 )
-            )
-            with mock.patch(
-                __name__ + ".fetch_latest_scheduled_run",
-                return_value={"created_at": "2999-01-01T00:00:00Z"},
-            ):
-                self.assertEqual(
-                    check_freshness(
-                        config,
-                        report,
-                        "rustfs/rustfs",
-                        "token",
-                        "https://api.github.test",
-                    ),
-                    0,
-                )
-            self.assertIn("All critical scheduled validations", report.read_text())
+                self.assertEqual(status, 1)
+                self.assertEqual(len(calls), 2)
+                self.assertIn("inspection failed: API unavailable", report)
+                self.assertNotIn("Initial grace until", report)
+                if first is good or second is good:
+                    self.assertIn(str(self.run_fixture()["html_url"]), report)
+
+    def test_invalid_api_evidence_fails_closed(self) -> None:
+        malformed = [
+            [], {}, {"workflow_runs": {}}, {"workflow_runs": [None]},
+            {"workflow_runs": [], "total_count": 1},
+            {"workflow_runs": [], "total_count": -1},
+            {"workflow_runs": [], "total_count": None},
+            {"workflow_runs": [], "total_count": True},
+            *({"workflow_runs": [self.run_fixture(**override)]} for override in (
+                {"event": "workflow_dispatch"}, {"head_branch": "other"},
+                {"created_at": "invalid"}, {"created_at": "2026-08-22T00:00:00"},
+                {"status": None}, {"conclusion": None}, {"conclusion": 1},
+                {"html_url": ""},
+            )),
+        ]
+        for payload in malformed:
+            for index, label in enumerate(("Last attempt", "Last completed success")):
+                with self.subTest(payload=payload, label=label):
+                    payloads = [{"workflow_runs": [self.run_fixture()]} for _ in range(2)]
+                    payloads[index] = payload
+                    status, report, _ = self.check_payloads(payloads, grace="2026-08-23T00:00:00Z")
+                    self.assertEqual(status, 1)
+                    self.assertIn(f"{label}: inspection failed", report)
+                    self.assertNotIn("Initial grace until", report)
+                    self.assertIn(str(self.run_fixture()["html_url"]), report)
+        for state, conclusion in (("in_progress", "success"), ("completed", "failure"), ("completed", "skipped")):
+            with self.subTest(state=state, conclusion=conclusion):
+                status, report, _ = self.check_payloads([
+                    {"workflow_runs": [self.run_fixture()]},
+                    {"workflow_runs": [self.run_fixture(status=state, conclusion=conclusion)]},
+                ])
+                self.assertEqual(status, 1)
+                self.assertIn("without a completed success", report)
+
+    def test_report_retains_every_workflow(self) -> None:
+        status, report, calls = self.check_payloads([
+            {"workflow_runs": [self.run_fixture()]}, {"workflow_runs": [self.run_fixture()]},
+            {"workflow_runs": []}, {"workflow_runs": []},
+            RuntimeError("API unavailable"), {"workflow_runs": [self.run_fixture()]},
+        ], workflows=3)
+        self.assertEqual(status, 1)
+        self.assertEqual(len(calls), 6)
+        for index in range(3):
+            self.assertEqual(report.count(f"`.github/workflows/check-{index}.yml`"), 1)
+        self.assertIn("No recorded run", report)
+        self.assertIn("Inspection failed", report)
+
+    def test_cli_requires_the_repository_default_branch(self) -> None:
+        from check_test_wiring import yaml_block
+
+        workflow = (ROOT / ".github/workflows/scheduled-validation-freshness.yml").read_text().splitlines()
+        job = yaml_block(workflow, "check-freshness", 2)
+        self.assertIsNotNone(job)
+        start = job.index("      - name: Check latest scheduled runs")
+        end = next((index for index in range(start + 1, len(job)) if job[index].startswith("      - ")), len(job))
+        environment = yaml_block(job[start:end], "env", 8)
+        self.assertIsNotNone(environment)
+        self.assertIn("          RUSTFS_DEFAULT_BRANCH: ${{ github.event.repository.default_branch }}", environment)
+
+        with (
+            mock.patch.dict(os.environ, {"GITHUB_REPOSITORY": "rustfs/rustfs", "GH_TOKEN": "test-token"}, clear=True),
+            mock.patch.object(sys, "argv", ["checker", "--report", "unused.md"]),
+            mock.patch("sys.stderr", new=io.StringIO()) as stderr,
+            self.assertRaises(SystemExit) as error,
+        ):
+            main()
+        self.assertEqual(error.exception.code, 2)
+        self.assertIn("RUSTFS_DEFAULT_BRANCH", stderr.getvalue())
 
 
 def main() -> int:
@@ -318,11 +501,14 @@ def main() -> int:
     repository = os.environ.get("GITHUB_REPOSITORY", "")
     token = os.environ.get("GH_TOKEN", "")
     api_url = os.environ.get("GITHUB_API_URL", "https://api.github.com")
+    default_branch = os.environ.get("RUSTFS_DEFAULT_BRANCH", "")
     if not re.fullmatch(r"[^/\s]+/[^/\s]+", repository):
         parser.error("GITHUB_REPOSITORY must be owner/repository")
     if not token:
         parser.error("GH_TOKEN is required")
-    return check_freshness(args.config, args.report, repository, token, api_url)
+    if not default_branch or any(character.isspace() for character in default_branch):
+        parser.error("RUSTFS_DEFAULT_BRANCH is required and must name the repository default branch")
+    return check_freshness(args.config, args.report, repository, token, api_url, default_branch)
 
 
 if __name__ == "__main__":

scripts/test_security_workflow.py

@@ -0,0 +1,197 @@
+#!/usr/bin/env python3
+"""Run the security workflow's evidence and result steps without remote VMs."""
+
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+
+from check_test_wiring import yaml_block
+
+
+ROOT = Path(__file__).resolve().parents[1]
+WORKFLOW = ROOT / ".github/workflows/rustfs-security-test.yml"
+CASE_ROW = "| IAM-101 | user CRUD lifecycle | PASS |"
+
+
+class SecurityWorkflowTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.source = WORKFLOW.read_text()
+        self.job = yaml_block(self.source.splitlines(), "security-test", 2)
+        self.assertIsNotNone(self.job)
+        starts = [i for i, line in enumerate(self.job) if line.startswith("      - name: ")]
+        self.steps = {
+            self.job[start].split(": ", 1)[1].strip('"'): self.job[start:end]
+            for start, end in zip(starts, starts[1:] + [len(self.job)])
+        }
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.directory = Path(self.temp.name)
+        self.context = {
+            "runner.temp": self.temp.name,
+            "github.server_url": "https://github.com",
+            "github.repository": "rustfs/rustfs",
+            "github.run_id": "314159",
+            "github.run_attempt": "2",
+            "github.sha": "0123456789abcdef0123456789abcdef01234567",
+            "github.event_name": "workflow_dispatch",
+            "github.workspace": self.temp.name,
+            "inputs.package_url": "",
+            "inputs.rustfs_version": "test-version",
+            "inputs.topology": "all",
+            "inputs.oidc_live": "false",
+            "steps.evidence.outcome": "skipped",
+            "steps.test.outcome": "skipped",
+            "steps.report.outcome": "skipped",
+        }
+        self.env = {
+            **os.environ, "GITHUB_STEP_SUMMARY": str(self.directory / "summary.md"),
+            "GITHUB_ENV": str(self.directory / "github-env"), "RUNNER_TEMP": self.temp.name, "TMPDIR": self.temp.name,
+        }
+        for key in ("server_url", "repository", "run_id", "run_attempt", "sha", "event_name"):
+            self.env[f"GITHUB_{key.upper()}"] = self.context[f"github.{key}"]
+        self.context["env.SECURITY_ARTIFACTS_DIR"] = ""
+        self.artifacts = self.directory / "rustfs-security-314159-2"
+        suite = self.directory / "auto-testing/rustfs-security-test.sh"
+        suite.parent.mkdir()
+        suite.write_text(
+            '#!/usr/bin/env bash\nset -euo pipefail\n'
+            'log_dir=$(mktemp -d "$TMPDIR/rustfs-security.XXXXXX")\n'
+            'echo "CURRENT SUITE LOG" > "$log_dir/suite.log"\n'
+            'case "$FAKE_REPORT" in\n'
+            f'  present) printf "%s\\n" "CURRENT SUITE DIAGNOSTIC" "{CASE_ROW}" > "$REPORT_FILE" ;;\n'
+            '  empty) : > "$REPORT_FILE" ;;\n'
+            'esac\n'
+            'echo "UNWRAPPED SUITE SUMMARY" >> "$GITHUB_STEP_SUMMARY"\n'
+            'exit "$FAKE_EXIT"\n'
+        )
+
+    def render(self, value: str) -> str:
+        return re.sub(r"\$\{\{\s*(.*?)\s*\}\}", lambda match: self.context[match[1]], value)
+
+    def step_env(self, lines: list[str], indent: int = 8) -> dict[str, str]:
+        result = {}
+        for line in yaml_block(lines, "env", indent) or []:
+            if line.strip() and not line.lstrip().startswith("#"):
+                key, value = line.strip().split(": ", 1)
+                result[key] = self.render(value.strip("'\""))
+        return result
+
+    def run_step(self, name: str) -> subprocess.CompletedProcess[str]:
+        lines = self.steps[name]
+        start = lines.index("        run: |") + 1
+        shell_lines = []
+        for line in lines[start:]:
+            if line.strip() and not line.startswith("          "):
+                break
+            shell_lines.append(line[10:])
+        self.assertTrue(shell_lines, f"missing literal shell body: {name}")
+        result = subprocess.run(
+            ["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", self.render("\n".join(shell_lines))],
+            cwd=self.directory, env={**self.env, **self.step_env(lines)}, capture_output=True, text=True,
+        )
+        for line in lines:
+            if line.startswith("        id: "):
+                self.context[f"steps.{line.split(': ', 1)[1]}.outcome"] = "failure" if result.returncode else "success"
+        if Path(self.env["GITHUB_ENV"]).exists():
+            for line in Path(self.env["GITHUB_ENV"]).read_text().splitlines():
+                key, value = line.split("=", 1)
+                self.env[key] = value
+                self.context[f"env.{key}"] = value
+        return result
+
+    def test_workflow_wiring(self) -> None:
+        names = list(self.steps)
+        self.assertLess(names.index("Checkout repository (for the OIDC live gate script)"), names.index("Checkout auto-testing scripts (with retry)"))
+        self.assertNotIn("    continue-on-error: true", self.job)
+        self.assertIn("        continue-on-error: true", self.steps["Run security suite"])
+        for name in ("Initialize security evidence", "Generate report"):
+            self.assertNotIn("        continue-on-error: true", self.steps[name])
+        self.assertIn("        if: ${{ always() && steps.evidence.outcome == 'success' }}", self.steps["Generate report"])
+        self.assertNotIn("/tmp/rustfs-security", self.source)
+        for name in ("Upload functional report to dashboard", "File failure issue in rustfs/backlog"):
+            report = next(line for line in self.steps[name] if line.strip().startswith("REPORT_FILE:"))
+            self.assertIn("${{ env.SECURITY_ARTIFACTS_DIR }}/report.md", report)
+        for name in ("Upload functional report to dashboard", "Upload report and logs"):
+            self.assertIn("        if: ${{ always() && steps.evidence.outcome == 'success' }}", self.steps[name])
+        artifact_settings = yaml_block(self.steps["Upload report and logs"], "with", 8)
+        self.assertIn("          path: ${{ env.SECURITY_ARTIFACTS_DIR }}/", artifact_settings)
+        self.assertIn("          if-no-files-found: error", artifact_settings)
+
+    def test_suite_report_and_result_matrix(self) -> None:
+        for outcome, mode, exit_code in (
+            ("success", "present", 0), ("failure", "present", 7), ("failure", "missing", 7),
+            ("success", "missing", 0), ("success", "empty", 0),
+            ("skipped", "missing", 0), ("skipped", "present", 0),
+            ("cancelled", "missing", 0), ("cancelled", "present", 0),
+        ):
+            with self.subTest(outcome=outcome, report=mode):
+                self.setUp()
+                initialized = self.run_step("Initialize security evidence")
+                self.assertEqual(initialized.returncode, 0, initialized.stderr)
+                self.assertEqual(self.env["SECURITY_ARTIFACTS_DIR"], str(self.artifacts))
+                self.env.update(FAKE_REPORT=mode, FAKE_EXIT=str(exit_code))
+                if outcome != "skipped" or mode == "present":
+                    suite = self.run_step("Run security suite")
+                    self.assertEqual(suite.returncode, exit_code, suite.stderr)
+                    logs = list(self.artifacts.glob("rustfs-security.*/suite.log"))
+                    self.assertEqual(len(logs), 1)
+                    self.assertEqual(logs[0].read_text(), "CURRENT SUITE LOG\n")
+                self.context["steps.test.outcome"] = outcome
+                report = self.run_step("Generate report")
+                success = outcome == "success" and mode == "present"
+                self.assertEqual(report.returncode == 0, success, report.stderr)
+                contents = (self.artifacts / "report.md").read_text()
+                for expected in (
+                    "https://github.com/rustfs/rustfs/actions/runs/314159", "Attempt: 2",
+                    f"Workflow Commit: {self.context['github.sha']}", "Trigger: workflow_dispatch",
+                    f"Test Step Outcome: {'success' if success else 'failure'}", f"Suite Step Outcome: {outcome}",
+                ):
+                    self.assertIn(expected, contents)
+                self.assertEqual(CASE_ROW in contents, success)
+                self.assertEqual("CURRENT SUITE DIAGNOSTIC" in contents, success)
+                if mode == "present":
+                    raw = (self.artifacts / "suite-report.md").read_text()
+                    self.assertEqual(raw, f"CURRENT SUITE DIAGNOSTIC\n{CASE_ROW}\n")
+                summary = Path(self.env["GITHUB_STEP_SUMMARY"]).read_text()
+                self.assertEqual(summary, contents)
+                self.assertNotIn("UNWRAPPED SUITE SUMMARY", summary)
+
+    def test_existing_evidence_directory_is_rejected(self) -> None:
+        self.artifacts.mkdir()
+        stale = self.artifacts / "suite-report.md"
+        stale.write_text("OLD RUN REPORT")
+        self.assertNotEqual(self.run_step("Initialize security evidence").returncode, 0)
+        self.assertEqual(stale.read_text(), "OLD RUN REPORT")
+        self.assertFalse(Path(self.env["GITHUB_ENV"]).exists())
+        (self.artifacts / "report.md").write_text("OLD RUN REPORT")
+        self.context.update({
+            "env.SECURITY_ARTIFACTS_DIR": str(self.artifacts), "secrets.PF_TESTING_GH_TOKEN": "fake-local-token",
+        })
+        fake_bin = self.directory / "bin"
+        fake_bin.mkdir()
+        gh = fake_bin / "gh"
+        gh.write_text(
+            '#!/usr/bin/env bash\nset -euo pipefail\n'
+            'if [ "$1 $2" = "issue create" ]; then\n'
+            '  while [ "$#" -gt 0 ]; do\n'
+            '    if [ "$1" = "--body-file" ]; then cat "$2" > "$CAPTURE_BODY"; fi\n'
+            '    shift\n'
+            '  done\n'
+            'fi\n'
+        )
+        gh.chmod(0o755)
+        body = self.directory / "issue-body.md"
+        self.env.update(PATH=f"{fake_bin}{os.pathsep}{os.environ['PATH']}", CAPTURE_BODY=str(body))
+        result = self.run_step("File failure issue in rustfs/backlog")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertNotIn("OLD RUN REPORT", body.read_text())
+        self.assertIn("https://github.com/rustfs/rustfs/actions/runs/314159", body.read_text())
+
+
+if __name__ == "__main__":
+    unittest.main()