fix(scanner): resume committed cleanup when scanning is disabled (#7281)

Run one supervised cleanup attempt for an existing operator reset, with
strict phase and revision checks under the original leader lock. Keep v3
reset authorization and responses unchanged, report deferred status, and
bound probe and shutdown waits without aborting in-flight reset ownership.

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

crates/scanner/src/lib.rs

@@ -85,8 +85,8 @@ pub use rustfs_scanner_metrics::last_minute;
 pub use scanner::{
     ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, ScannerCycleScheduleStatus, ScannerPauseBacklogAlertReason,
     ScannerPauseBacklogPhase, ScannerPauseBacklogStatus, ScannerPauseBacklogThresholds, ScannerUsageStateResetResult,
-    init_data_scanner, reset_scanner_cycle_recovery, reset_scanner_usage_state_for_full_rebuild, scanner_cycle_recovery_status,
-    scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_topology_digest,
+    init_data_scanner, init_scanner_with_recovery, reset_scanner_cycle_recovery, reset_scanner_usage_state_for_full_rebuild,
+    scanner_cycle_recovery_status, scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_topology_digest,
 };
 pub use scanner_io::{
     ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState,

crates/scanner/src/scanner.rs

@@ -948,6 +948,33 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
     init_data_scanner_with_storage(ctx, storeapi).await;
 }
 
+/// Start normal scanning when enabled, or one resume-only cleanup attempt.
+/// The disabled branch returns a finite task for the startup owner to join;
+/// it never enables ordinary namespace scanning or accepts a new reset intent.
+pub async fn init_scanner_with_recovery(
+    ctx: CancellationToken,
+    storeapi: Arc<ECStore>,
+    enabled: bool,
+) -> Option<tokio::task::JoinHandle<()>> {
+    if enabled {
+        init_data_scanner(ctx, storeapi).await;
+        return None;
+    }
+    Some(tokio::spawn(async move {
+        if let Err(error) = resume_scanner_cycle_cleanup(ctx, storeapi).await {
+            warn!(
+                target: "rustfs::scanner",
+                event = EVENT_SCANNER_PERSIST_STATE,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_RUNTIME,
+                state = "disabled_cleanup_deferred",
+                error = %error,
+                "Disabled scanner cleanup remains pending for an operator retry"
+            );
+        }
+    }))
+}
+
 async fn init_data_scanner_with_storage<S>(ctx: CancellationToken, storeapi: Arc<S>)
 where
     S: ScannerStorage,

crates/scanner/src/scanner/cycle_state.rs

@@ -18,6 +18,7 @@ use crate::data_usage_define::{
     DATA_USAGE_BLOOM_RECOVERY_PATH, DATA_USAGE_RECOVERY_PATH, usage_floor_primary_read_error_allows_backup,
 };
 use crate::storage_api::owner::ObjectIO as _;
+use std::sync::atomic::AtomicU64;
 use tokio::io::AsyncReadExt as _;
 
 const SCANNER_CYCLE_RECOVERY_SCHEMA_VERSION: u16 = 1;
@@ -34,6 +35,86 @@ const CACHE_CYCLE_AHEAD: &str = "cache_cycle_ahead";
 
 const SCANNER_USAGE_STATE_RESET_MODE_FULL_REBUILD: &str = "full-rebuild";
 
+#[cfg(test)]
+pub(super) mod cleanup_io_fault {
+    use super::*;
+
+    #[derive(Clone, Copy, PartialEq, Eq)]
+    pub(in crate::scanner) enum Stage {
+        PrimaryRead,
+        PrimaryWrite,
+        UsageFence,
+    }
+
+    struct Injection {
+        store: std::sync::Weak<ECStore>,
+        stage: Stage,
+        fired: AtomicBool,
+        owned: AtomicBool,
+        newer_completion: bool,
+    }
+
+    static INJECTION: StdMutex<Option<Arc<Injection>>> = StdMutex::new(None);
+    pub(in crate::scanner) struct Guard(Arc<Injection>);
+
+    impl Guard {
+        pub(in crate::scanner) fn fired_while_owned(&self) -> bool {
+            self.0.fired.load(Ordering::Relaxed) && self.0.owned.load(Ordering::Relaxed)
+        }
+    }
+
+    impl Drop for Guard {
+        fn drop(&mut self) {
+            let mut slot = INJECTION.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+            if slot.as_ref().is_some_and(|current| Arc::ptr_eq(current, &self.0)) {
+                *slot = None;
+            }
+        }
+    }
+
+    pub(in crate::scanner) fn install(store: &Arc<ECStore>, stage: Stage, newer_completion: bool) -> Guard {
+        let injection = Arc::new(Injection {
+            store: Arc::downgrade(store),
+            stage,
+            fired: AtomicBool::new(false),
+            owned: AtomicBool::new(false),
+            newer_completion,
+        });
+        let mut slot = INJECTION.lock().expect("cleanup injection slot");
+        assert!(slot.is_none(), "only one cleanup I/O injection may be installed");
+        *slot = Some(injection.clone());
+        Guard(injection)
+    }
+
+    pub(super) fn check(store: &Arc<ECStore>, stage: Stage, owned: bool) -> Result<(), ScannerError> {
+        let injection = {
+            let mut slot = INJECTION.lock().expect("cleanup injection slot");
+            if slot
+                .as_ref()
+                .is_some_and(|injection| injection.stage == stage && injection.store.ptr_eq(&Arc::downgrade(store)))
+            {
+                slot.take()
+            } else {
+                None
+            }
+        };
+        let Some(injection) = injection else {
+            return Ok(());
+        };
+        injection.fired.store(true, Ordering::Relaxed);
+        injection.owned.store(owned, Ordering::Relaxed);
+        if injection.newer_completion {
+            set_scanner_cycle_recovery_status(recovery_status("healthy", None, false));
+        }
+        let reason = match stage {
+            Stage::PrimaryRead => "injected primary read failure",
+            Stage::PrimaryWrite => "injected primary write failure",
+            Stage::UsageFence => "injected usage fence failure",
+        };
+        Err(ScannerError::Io(std::io::Error::other(reason)))
+    }
+}
+
 #[derive(Clone, Debug, Default, Serialize)]
 pub struct ScannerCycleRecoveryStatus {
     /// The immutable primary object whose revision is being guarded.
@@ -81,6 +162,8 @@ static SCANNER_CYCLE_RECOVERY_STATUS: LazyLock<RwLock<ScannerCycleRecoveryStatus
         ..Default::default()
     })
 });
+// An old startup observation must not overwrite a newer explicit reset status.
+static SCANNER_CYCLE_RECOVERY_STATUS_VERSION: AtomicU64 = AtomicU64::new(0);
 
 pub fn scanner_cycle_recovery_status() -> ScannerCycleRecoveryStatus {
     SCANNER_CYCLE_RECOVERY_STATUS
@@ -90,6 +173,24 @@ pub fn scanner_cycle_recovery_status() -> ScannerCycleRecoveryStatus {
 }
 
 fn set_scanner_cycle_recovery_status(status: ScannerCycleRecoveryStatus) {
+    let _ = publish_scanner_cleanup_status(status, None);
+}
+
+pub(super) fn scanner_cleanup_status_version() -> u64 {
+    let _status = SCANNER_CYCLE_RECOVERY_STATUS
+        .read()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    SCANNER_CYCLE_RECOVERY_STATUS_VERSION.load(Ordering::Relaxed)
+}
+
+pub(super) fn publish_scanner_cleanup_status(status: ScannerCycleRecoveryStatus, expected: Option<u64>) -> Option<u64> {
+    let mut current = SCANNER_CYCLE_RECOVERY_STATUS
+        .write()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    let version = SCANNER_CYCLE_RECOVERY_STATUS_VERSION.load(Ordering::Relaxed);
+    if expected.is_some_and(|expected| expected == u64::MAX || expected != version) {
+        return None;
+    }
     let recovery_required = if matches!(
         status.state.as_str(),
         "blocked"
@@ -106,9 +207,27 @@ fn set_scanner_cycle_recovery_status(status: ScannerCycleRecoveryStatus) {
     };
     metrics::gauge!(METRIC_SCANNER_CYCLE_RECOVERY_REQUIRED).set(recovery_required);
     metrics::gauge!(METRIC_SCANNER_CYCLE_RECOVERY_RETRY_COUNT).set(status.retry_count as f64);
-    *SCANNER_CYCLE_RECOVERY_STATUS
-        .write()
-        .unwrap_or_else(|poisoned| poisoned.into_inner()) = status;
+    *current = status;
+    let next = version.saturating_add(1);
+    SCANNER_CYCLE_RECOVERY_STATUS_VERSION.store(next, Ordering::Relaxed);
+    Some(next)
+}
+
+fn publish_scanner_cleanup_failure(reason: String, expected: u64) {
+    let mut status = {
+        let current = SCANNER_CYCLE_RECOVERY_STATUS
+            .read()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        if SCANNER_CYCLE_RECOVERY_STATUS_VERSION.load(Ordering::Relaxed) != expected {
+            return;
+        }
+        current.clone()
+    };
+    // Preserve the latest core progress, including a newly written primary's
+    // revision and epoch. The original marker may predate that durable write.
+    status.reason = Some(reason);
+    status.last_attempt_at_unix_secs = Some(unix_now_secs());
+    let _ = publish_scanner_cleanup_status(status, Some(expected));
 }
 
 pub(super) fn record_scanner_usage_floor_failure(reason: String) {
@@ -932,10 +1051,81 @@ pub(crate) async fn load_scanner_cycle_state_for_startup(
     }
 }
 
-/// Reset a blocked cycle state after an operator has explicitly requested a full
-/// usage rebuild. The primary object is changed first with its observed ETag;
-/// the recovery marker is removed only when its own ETag still matches.
+/// Reset a blocked cycle state after an explicit full-rescan request. Durable
+/// cleanup state fences primary rewrites; marker removal retains its ETag and
+/// usage-epoch checks.
 pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<ECStore>) -> Result<(), ScannerError> {
+    reset_scanner_cycle_recovery_for_intent(ctx, storeapi, None, None).await
+}
+
+/// Resume only an operator reset whose cleanup phase is already durable.
+/// Missing or merely blocked markers never authorize an automatic reset.
+pub(super) async fn resume_scanner_cycle_cleanup(ctx: CancellationToken, storeapi: Arc<ECStore>) -> Result<(), ScannerError> {
+    let status_version = scanner_cleanup_status_version();
+    let (marker, revision) = match read_scanner_cleanup_marker(storeapi.clone(), &ctx).await {
+        Ok(Some(marker)) => marker,
+        Ok(None) => return Ok(()),
+        Err(error) => {
+            let _ =
+                publish_scanner_cleanup_status(recovery_status("blocked", Some(&error.to_string()), false), Some(status_version));
+            return Err(error);
+        }
+    };
+    let mut observation =
+        publish_scanner_cleanup_status(recovery_status_from_marker(&marker, &marker.state), Some(status_version));
+    if marker.state != "cleanup-pending" {
+        return Ok(());
+    }
+    let result = reset_scanner_cycle_recovery_for_intent(ctx, storeapi, Some(revision), Some(&mut observation)).await;
+    if let Err(error) = &result
+        && let Some(observation) = observation
+    {
+        publish_scanner_cleanup_failure(error.to_string(), observation);
+    }
+    result
+}
+
+pub(super) async fn read_scanner_cleanup_marker(
+    storeapi: Arc<impl ScannerObjectIO>,
+    ctx: &CancellationToken,
+) -> Result<Option<(ScannerCycleRecoveryMarker, DataUsageCacheRevision)>, ScannerError> {
+    let (data, revision) = tokio::select! {
+        biased;
+        _ = ctx.cancelled() => return Err(ScannerError::Other("scanner cleanup recovery was cancelled".to_string())),
+        result = tokio::time::timeout(data_usage_persist_timeout(), read_cycle_recovery_marker_bytes(storeapi)) => {
+            result.map_err(|_| ScannerError::Other("scanner cleanup marker inspection timed out".to_string()))?
+                .map_err(|err| ScannerError::Other(format!("failed to inspect pending scanner cleanup: {err}")))?
+        }
+    };
+    let Some(data) = data else {
+        return Ok(None);
+    };
+    let marker: ScannerCycleRecoveryMarker = serde_json::from_slice(&data)
+        .map_err(|err| ScannerError::Other(format!("pending scanner cleanup marker is invalid: {err}")))?;
+    validate_recovery_marker(&marker)
+        .map_err(|err| ScannerError::Other(format!("pending scanner cleanup marker is invalid: {err}")))?;
+    Ok(Some((marker, revision)))
+}
+
+pub(super) async fn reset_scanner_cycle_recovery_for_intent(
+    ctx: CancellationToken,
+    storeapi: Arc<ECStore>,
+    expected_cleanup_revision: Option<DataUsageCacheRevision>,
+    mut observation: Option<&mut Option<u64>>,
+) -> Result<(), ScannerError> {
+    #[cfg(test)]
+    let resume_only = expected_cleanup_revision.is_some();
+    // The outer Some distinguishes a tracked resume whose version may be
+    // invalidated from an explicit v3 reset with no observation owner.
+    let mut publish_status = |status| {
+        if let Some(version) = observation.as_deref_mut() {
+            if let Some(expected) = *version {
+                *version = publish_scanner_cleanup_status(status, Some(expected));
+            }
+        } else {
+            set_scanner_cycle_recovery_status(status);
+        }
+    };
     let lock = storeapi
         .new_ns_lock(RUSTFS_META_BUCKET, "leader.lock")
         .await
@@ -964,6 +1154,21 @@ 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}"))),
     };
+    if let Some(expected) = expected_cleanup_revision {
+        if marker_revision != expected || !owns_reset() {
+            return Err(ScannerError::Other(
+                "pending scanner cleanup changed before recovery acquired ownership".to_string(),
+            ));
+        }
+        let marker: ScannerCycleRecoveryMarker = marker_data
+            .as_deref()
+            .and_then(|data| serde_json::from_slice(data).ok())
+            .filter(|marker| validate_recovery_marker(marker).is_ok() && marker.state == "cleanup-pending")
+            .ok_or_else(|| {
+                ScannerError::Other("scanner cleanup recovery requires an unchanged cleanup-pending marker".to_string())
+            })?;
+        publish_status(recovery_status_from_marker(&marker, "cleanup-pending"));
+    }
     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.
@@ -981,7 +1186,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
                 "scanner cycle recovery marker is absent without a completed reset fence".to_string(),
             ));
         }
-        set_scanner_cycle_recovery_status(recovery_status("healthy", None, false));
+        publish_status(recovery_status("healthy", None, false));
         super::notify_scanner_cycle_recovery_wake();
         return Ok(());
     };
@@ -997,6 +1202,10 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
         ));
     }
 
+    #[cfg(test)]
+    if resume_only {
+        cleanup_io_fault::check(&storeapi, cleanup_io_fault::Stage::PrimaryRead, owns_reset())?;
+    }
     let (mut primary_reader, primary_revision) = match storeapi
         .get_object_reader(
             RUSTFS_META_BUCKET,
@@ -1062,7 +1271,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
             let (cleanup_marker, cleanup_marker_revision) =
                 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"));
+            publish_status(recovery_status_from_marker(&cleanup_marker, "cleanup-pending"));
             let usage_floor = persisted_usage_floor(storeapi.clone()).await?;
             let fence_epoch = primary_epoch
                 .max(usage_floor.leader_epoch)
@@ -1082,6 +1291,10 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
                 ));
             }
             verify_cycle_reset_intent(storeapi.clone(), &cleanup_marker_revision, &owns_reset).await?;
+            #[cfg(test)]
+            if resume_only {
+                cleanup_io_fault::check(&storeapi, cleanup_io_fault::Stage::PrimaryWrite, owns_reset())?;
+            }
             let preserved_info = save_reset_config(
                 storeapi.clone(),
                 DATA_USAGE_BLOOM_NAME_PATH.as_str(),
@@ -1155,7 +1368,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
                     ScannerError::Other(format!("failed to clear stale cycle recovery marker: {err}"))
                 }
             })?;
-            set_scanner_cycle_recovery_status(recovery_status("healthy", None, false));
+            publish_status(recovery_status("healthy", None, false));
             super::notify_scanner_cycle_recovery_wake();
             return Ok(());
         } else if !force_full_rescan && !marker_cleanup_pending {
@@ -1228,11 +1441,23 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
         ));
     }
     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, &owns_reset)
-            .await
-    {
-        set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus {
+    let usage_fence = fence_scanner_usage_epoch_with_expected_epoch(
+        &ctx,
+        storeapi.clone(),
+        leader_epoch,
+        Some(reset_epoch),
+        false,
+        &owns_reset,
+    );
+    #[cfg(test)]
+    let usage_fence = async {
+        if resume_only {
+            cleanup_io_fault::check(&storeapi, cleanup_io_fault::Stage::UsageFence, owns_reset())?;
+        }
+        usage_fence.await
+    };
+    if let Err(err) = usage_fence.await {
+        publish_status(ScannerCycleRecoveryStatus {
             path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
             quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()),
             state: "cleanup-pending".to_string(),
@@ -1260,7 +1485,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
         }
     };
     if current_revision != rebuilt_revision {
-        set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus {
+        publish_status(ScannerCycleRecoveryStatus {
             path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
             quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()),
             state: "cleanup-pending".to_string(),
@@ -1281,7 +1506,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
     }
 
     if guard.is_lock_lost() {
-        set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus {
+        publish_status(ScannerCycleRecoveryStatus {
             path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
             quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()),
             state: "cleanup-pending".to_string(),
@@ -1318,7 +1543,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
     .await
     {
         if scanner_publication_epoch_changed(&err) {
-            set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus {
+            publish_status(ScannerCycleRecoveryStatus {
                 path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
                 quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()),
                 state: "cleanup-pending".to_string(),
@@ -1336,7 +1561,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
                 "scanner recovery reset deferred by a movement epoch change".to_string(),
             ));
         }
-        set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus {
+        publish_status(ScannerCycleRecoveryStatus {
             path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
             quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()),
             state: "cleanup-pending".to_string(),
@@ -1352,7 +1577,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc<
         });
         return Err(ScannerError::Other(format!("failed to clear cycle recovery marker: {err}")));
     }
-    set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus {
+    publish_status(ScannerCycleRecoveryStatus {
         path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
         quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()),
         state: "healthy".to_string(),

crates/scanner/src/scanner/tests.rs

@@ -36,6 +36,8 @@ use tokio::time::{Duration, advance};
 
 const TEST_DEFAULT_SCANNER_CYCLE_SECS: u64 = 24 * 60 * 60;
 
+mod recovery_control;
+
 async fn setup_scanner_cycle_store() -> (tempfile::TempDir, Arc<ECStore>) {
     setup_scanner_cycle_store_with_usage_baseline(true).await
 }

crates/scanner/src/scanner/tests/recovery_control.rs

@@ -0,0 +1,475 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::super::cycle_state::cleanup_io_fault;
+use super::*;
+use crate::storage_api::owner::{EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats};
+
+async fn seed_cleanup(store: &Arc<ECStore>, state: &str) -> ScannerCycleRecoveryMarker {
+    let cycle = CurrentCycle {
+        current: 3,
+        next: 42,
+        ..Default::default()
+    };
+    save_config(
+        store.clone(),
+        DATA_USAGE_BLOOM_NAME_PATH.as_str(),
+        encode_scanner_cycle_state(&cycle, 7).expect("cycle encoding"),
+    )
+    .await
+    .expect("persist cycle");
+    let usage = DataUsageInfo {
+        scanner_epoch: Some(7),
+        scanner_cycle: Some(41),
+        ..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("usage encoding"),
+    )
+    .await
+    .expect("persist usage floor");
+    let marker = ScannerCycleRecoveryMarker {
+        schema_version: 1,
+        primary_revision: "previous-primary".to_string(),
+        generation: 41,
+        leader_epoch: 7,
+        classification: "corrupt".to_string(),
+        first_detected_at_unix_secs: 1,
+        last_attempt_at_unix_secs: 2,
+        retry_count: 1,
+        reason: "operator reset in progress".to_string(),
+        path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
+        quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(),
+        state: state.to_string(),
+    };
+    save_config(
+        store.clone(),
+        DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
+        serde_json::to_vec(&marker).expect("marker encoding"),
+    )
+    .await
+    .expect("persist operator marker");
+    marker
+}
+
+async fn persisted_state(store: &Arc<ECStore>) -> Vec<(Option<Vec<u8>>, DataUsageCacheRevision)> {
+    let mut state = Vec::new();
+    for path in [
+        DATA_USAGE_BLOOM_NAME_PATH.as_str(),
+        DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
+        DATA_USAGE_OBJ_NAME_PATH.as_str(),
+    ] {
+        state.push(
+            read_config_with_revision(store.clone(), path)
+                .await
+                .expect("read exact metadata revision"),
+        );
+    }
+    state
+}
+
+async fn run_disabled_startup(ctx: CancellationToken, store: Arc<ECStore>) {
+    let initialized_before = crate::scanner_runtime_initialized();
+    let cleanup = init_scanner_with_recovery(ctx, store, false).await;
+    if let Some(cleanup) = cleanup {
+        tokio::time::timeout(Duration::from_secs(15), cleanup)
+            .await
+            .expect("finite disabled cleanup attempt")
+            .expect("cleanup task should not panic");
+    }
+    assert_eq!(
+        crate::scanner_runtime_initialized(),
+        initialized_before,
+        "disabled recovery must not start the normal scanner runtime"
+    );
+}
+
+async fn assert_reset_fences(store: &Arc<ECStore>) {
+    let data = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
+        .await
+        .expect("cycle remains durable");
+    let (cycle, epoch) = decode_scanner_cycle_state(&data).expect("valid preserved cycle");
+    assert_eq!((cycle.current, cycle.next, epoch), (3, 42, 8));
+    let usage: DataUsageInfo = serde_json::from_slice(
+        &read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+            .await
+            .expect("durable usage fence"),
+    )
+    .expect("valid usage");
+    assert_eq!(usage.scanner_epoch, Some(8));
+    assert_eq!(usage.scanner_cycle, Some(41));
+    assert!(matches!(
+        read_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
+        Err(EcstoreError::ConfigNotFound)
+    ));
+}
+
+#[tokio::test]
+#[serial]
+async fn disabled_cleanup_reopens_persisted_intent_without_starting_scanner() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    seed_cleanup(&store, "cleanup-pending").await;
+    let restarted = restart_scanner_cycle_store_from(&store).await;
+    run_disabled_startup(CancellationToken::new(), restarted.clone()).await;
+    assert_reset_fences(&restarted).await;
+    let completed = persisted_state(&restarted).await;
+    run_disabled_startup(CancellationToken::new(), restarted.clone()).await;
+    assert_eq!(
+        persisted_state(&restarted).await,
+        completed,
+        "a later startup without an intent must not reset again"
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn disabled_cleanup_does_not_authorize_blocked_unknown_or_corrupt_markers() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    for kind in ["blocked", "unknown-phase", "future-version", "unknown-field", "corrupt"] {
+        let marker = seed_cleanup(&store, "blocked").await;
+        let mut value = serde_json::to_value(marker).expect("marker value");
+        match kind {
+            "unknown-phase" => value["state"] = "future-phase".into(),
+            "future-version" => value["schema_version"] = 99.into(),
+            "unknown-field" => value["future_hint"] = true.into(),
+            _ => {}
+        }
+        let bytes = if kind == "corrupt" {
+            b"{broken".to_vec()
+        } else {
+            serde_json::to_vec(&value).expect("marker JSON")
+        };
+        save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), bytes)
+            .await
+            .expect("persist rejected marker");
+        let before = persisted_state(&store).await;
+        run_disabled_startup(CancellationToken::new(), store.clone()).await;
+        assert_eq!(persisted_state(&store).await, before, "{kind} must not become an automatic full rescan");
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn disabled_cleanup_rechecks_revision_after_waiting_for_leader_lock() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    let mut marker = seed_cleanup(&store, "cleanup-pending").await;
+    let expected = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
+        .await
+        .expect("intent revision")
+        .1;
+    let lock = store
+        .new_ns_lock(RUSTFS_META_BUCKET, "leader.lock")
+        .await
+        .expect("leader lock");
+    let guard = lock
+        .get_write_lock_quiet(Duration::from_secs(1))
+        .await
+        .expect("hold leader ownership");
+    let mut recovery = Box::pin(reset_scanner_cycle_recovery_for_intent(
+        CancellationToken::new(),
+        store.clone(),
+        Some(expected),
+        None,
+    ));
+    assert!(matches!(futures::poll!(&mut recovery), Poll::Pending));
+    marker.state = "blocked".to_string();
+    save_config(
+        store.clone(),
+        DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
+        serde_json::to_vec(&marker).expect("replacement marker"),
+    )
+    .await
+    .expect("replace intent while the fixture owns leader lock");
+    let replaced = persisted_state(&store).await;
+    drop(guard);
+    let error = recovery.await.expect_err("old preflight cannot authorize replacement marker");
+    assert!(error.to_string().contains("changed before recovery acquired ownership"));
+    assert_eq!(persisted_state(&store).await, replaced);
+}
+
+#[tokio::test]
+#[serial]
+async fn disabled_cleanup_requires_phase_even_when_revision_matches() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    seed_cleanup(&store, "blocked").await;
+    let before = persisted_state(&store).await;
+    let error = reset_scanner_cycle_recovery_for_intent(CancellationToken::new(), store.clone(), Some(before[1].1.clone()), None)
+        .await
+        .expect_err("a matching ETag alone is not operator cleanup authorization");
+    assert!(error.to_string().contains("unchanged cleanup-pending"));
+    assert_eq!(persisted_state(&store).await, before);
+    reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
+        .await
+        .expect("explicit v3 core retains full reset authorization");
+    assert_reset_fences(&store).await;
+}
+
+#[tokio::test]
+#[serial]
+async fn disabled_cleanup_lock_busy_preserves_intent_without_force_unlock() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    seed_cleanup(&store, "cleanup-pending").await;
+    let before = persisted_state(&store).await;
+    let lock = store
+        .new_ns_lock(RUSTFS_META_BUCKET, "leader.lock")
+        .await
+        .expect("leader lock");
+    let guard = lock
+        .get_write_lock_quiet(Duration::from_secs(1))
+        .await
+        .expect("hold live leader");
+    let error = resume_scanner_cycle_cleanup(CancellationToken::new(), store.clone())
+        .await
+        .expect_err("busy leader must block recovery");
+    assert!(error.to_string().contains("leader lock is busy"));
+    let status = scanner_cycle_recovery_status();
+    assert_eq!(status.state, "cleanup-pending");
+    assert!(
+        status
+            .reason
+            .as_deref()
+            .is_some_and(|reason| reason.contains("leader lock is busy"))
+    );
+    assert!(!status.retryable, "disabled startup makes one attempt, not an automatic retry loop");
+    assert!(!guard.is_lock_lost(), "recovery must not revoke the live owner");
+    assert_eq!(persisted_state(&store).await, before);
+    drop(guard);
+    run_disabled_startup(CancellationToken::new(), store.clone()).await;
+    assert_reset_fences(&store).await;
+}
+
+#[tokio::test]
+#[serial]
+async fn disabled_cleanup_movement_pause_preserves_intent_for_later_startup() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    seed_cleanup(&store, "cleanup-pending").await;
+    let before = persisted_state(&store).await;
+    *store.rebalance_meta.write().await = Some(EcstoreRebalanceMeta {
+        id: "cleanup-movement".to_string(),
+        pool_stats: vec![EcstoreRebalanceStats {
+            participating: true,
+            info: EcstoreRebalanceInfo {
+                start_time: Some(time::OffsetDateTime::now_utc()),
+                status: EcstoreRebalStatus::Started,
+                ..Default::default()
+            },
+            ..Default::default()
+        }],
+        ..Default::default()
+    });
+    let error = resume_scanner_cycle_cleanup(CancellationToken::new(), store.clone())
+        .await
+        .expect_err("movement must block reset publication");
+    assert!(error.to_string().contains("blocked by data movement"));
+    let status = scanner_cycle_recovery_status();
+    assert_eq!(status.state, "cleanup-pending");
+    assert!(
+        status
+            .reason
+            .as_deref()
+            .is_some_and(|reason| reason.contains("blocked by data movement"))
+    );
+    assert_eq!(persisted_state(&store).await, before);
+    *store.rebalance_meta.write().await = None;
+    run_disabled_startup(CancellationToken::new(), store.clone()).await;
+    assert_reset_fences(&store).await;
+}
+
+#[tokio::test]
+#[serial]
+async fn disabled_cleanup_cancelled_startup_preserves_persisted_work() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    seed_cleanup(&store, "cleanup-pending").await;
+    let before = persisted_state(&store).await;
+    let ctx = CancellationToken::new();
+    ctx.cancel();
+    run_disabled_startup(ctx, store.clone()).await;
+    assert_eq!(persisted_state(&store).await, before);
+}
+
+#[tokio::test(start_paused = true)]
+#[serial]
+async fn disabled_cleanup_probe_obeys_cancellation_and_existing_io_deadline() {
+    for cancel in [false, true] {
+        let store = Arc::new(MemoryConfigStore::default());
+        store.delayed_gets.lock().await.insert(
+            memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()),
+            data_usage_persist_timeout().saturating_add(Duration::from_secs(60)),
+        );
+        let ctx = CancellationToken::new();
+        let mut probe = Box::pin(read_scanner_cleanup_marker(store.clone(), &ctx));
+        assert!(matches!(futures::poll!(&mut probe), Poll::Pending));
+        if cancel {
+            ctx.cancel();
+        } else {
+            advance(data_usage_persist_timeout()).await;
+        }
+        let error = probe.await.expect_err("pending read must be bounded");
+        assert!(error.to_string().contains(if cancel { "cancelled" } else { "timed out" }));
+        assert!(store.put_counts.lock().await.is_empty(), "probe must remain read-only");
+    }
+}
+
+#[test]
+#[serial]
+fn disabled_cleanup_old_observation_cannot_overwrite_a_new_completion() {
+    let original = scanner_cycle_recovery_status();
+    let healthy = ScannerCycleRecoveryStatus {
+        state: "healthy".to_string(),
+        ..Default::default()
+    };
+    publish_scanner_cleanup_status(healthy.clone(), None).expect("first status version");
+    let old = scanner_cleanup_status_version();
+    publish_scanner_cleanup_status(healthy, None).expect("a newer completion may have identical fields");
+    assert!(
+        publish_scanner_cleanup_status(
+            ScannerCycleRecoveryStatus {
+                state: "cleanup-pending".to_string(),
+                reason: Some("old lock wait failed".to_string()),
+                ..Default::default()
+            },
+            Some(old)
+        )
+        .is_none()
+    );
+    assert_eq!(scanner_cycle_recovery_status().state, "healthy");
+    publish_scanner_cleanup_status(original, None).expect("restore prior observation");
+}
+
+#[tokio::test]
+#[serial]
+async fn disabled_cleanup_owned_read_and_write_failures_keep_specific_status() {
+    for (stage, newer_completion) in [
+        (cleanup_io_fault::Stage::PrimaryRead, false),
+        (cleanup_io_fault::Stage::PrimaryWrite, false),
+        (cleanup_io_fault::Stage::PrimaryWrite, true),
+    ] {
+        let (_dir, store) = setup_scanner_cycle_store().await;
+        seed_cleanup(&store, "cleanup-pending").await;
+        let before = persisted_state(&store).await;
+        let injection = cleanup_io_fault::install(&store, stage, newer_completion);
+        let error = resume_scanner_cycle_cleanup(CancellationToken::new(), store.clone())
+            .await
+            .expect_err("injected owned I/O boundary");
+        assert!(
+            injection.fired_while_owned(),
+            "fault must occur after real leader ownership and marker validation"
+        );
+        let expected_error = match stage {
+            cleanup_io_fault::Stage::PrimaryRead => "injected primary read failure",
+            cleanup_io_fault::Stage::PrimaryWrite => "injected primary write failure",
+            cleanup_io_fault::Stage::UsageFence => "injected usage fence failure",
+        };
+        assert!(error.to_string().contains(expected_error));
+        let status = scanner_cycle_recovery_status();
+        if newer_completion {
+            assert_eq!(status.state, "healthy", "old error cannot overwrite a newer completion observation");
+            assert!(status.reason.is_none());
+        } else {
+            assert_eq!(status.state, "cleanup-pending");
+            assert!(
+                status.reason.as_deref().is_some_and(|reason| reason.contains(expected_error)),
+                "{status:?}"
+            );
+        }
+        let after = persisted_state(&store).await;
+        assert_eq!(after[0], before[0], "failed primary I/O must preserve its prior revision");
+        assert_eq!(after[2], before[2], "failed primary I/O must not advance the usage fence");
+        let marker: ScannerCycleRecoveryMarker =
+            serde_json::from_slice(after[1].0.as_deref().expect("durable marker retained")).expect("valid cleanup marker");
+        assert_eq!(marker.state, "cleanup-pending");
+        drop(injection);
+        run_disabled_startup(CancellationToken::new(), store.clone()).await;
+        assert_reset_fences(&store).await;
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn disabled_cleanup_invalidated_observation_never_becomes_unconditional() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    seed_cleanup(&store, "cleanup-pending").await;
+    let revision = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
+        .await
+        .expect("marker revision")
+        .1;
+    let newer = ScannerCycleRecoveryStatus {
+        state: "healthy".to_string(),
+        reason: Some("newer completion owner".to_string()),
+        ..Default::default()
+    };
+    publish_scanner_cleanup_status(newer.clone(), None).expect("newer observation");
+    let mut invalidated = None;
+    reset_scanner_cycle_recovery_for_intent(CancellationToken::new(), store.clone(), Some(revision), Some(&mut invalidated))
+        .await
+        .expect("metadata cleanup may complete without owning the newest status observation");
+    assert!(invalidated.is_none());
+    assert_eq!(
+        serde_json::to_value(scanner_cycle_recovery_status()).expect("observed status"),
+        serde_json::to_value(newer).expect("newer status")
+    );
+    assert_reset_fences(&store).await;
+}
+
+#[tokio::test]
+#[serial]
+async fn disabled_cleanup_later_failure_preserves_rebuilt_primary_status_identity() {
+    for newer_completion in [false, true] {
+        let (_dir, store) = setup_scanner_cycle_store().await;
+        seed_cleanup(&store, "cleanup-pending").await;
+        save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), b"corrupt-cycle".to_vec())
+            .await
+            .expect("force the full reconstruction branch");
+        let before = persisted_state(&store).await;
+        let injection = cleanup_io_fault::install(&store, cleanup_io_fault::Stage::UsageFence, newer_completion);
+        let error = resume_scanner_cycle_cleanup(CancellationToken::new(), store.clone())
+            .await
+            .expect_err("fail after primary publication");
+        assert!(injection.fired_while_owned());
+        assert!(error.to_string().contains("injected usage fence failure"));
+        let after = persisted_state(&store).await;
+        assert_ne!(after[0].1, before[0].1, "the primary write must actually commit before this failure");
+        let (cycle, epoch) =
+            decode_scanner_cycle_state(after[0].0.as_deref().expect("rebuilt primary")).expect("valid durable reconstruction");
+        assert_eq!((cycle.current, cycle.next, epoch), (0, 42, 8));
+        assert_eq!(after[2], before[2], "usage fence publication was rejected");
+        let marker: ScannerCycleRecoveryMarker =
+            serde_json::from_slice(after[1].0.as_deref().expect("cleanup marker retained")).expect("valid cleanup marker");
+        assert_eq!(marker.state, "cleanup-pending");
+        let status = scanner_cycle_recovery_status();
+        if newer_completion {
+            assert_eq!(status.state, "healthy");
+            assert!(
+                status.reason.is_none(),
+                "old core and outer failure must both retain invalidated ownership"
+            );
+        } else {
+            let DataUsageCacheRevision::Etag(etag) = &after[0].1 else {
+                panic!("rebuilt primary must have a revision");
+            };
+            assert_eq!(status.state, "cleanup-pending");
+            assert_eq!(status.primary_revision.as_deref(), Some(etag.as_str()));
+            assert_eq!(status.generation, Some(cycle.next));
+            assert_eq!(status.leader_epoch, Some(epoch));
+            assert!(
+                status
+                    .reason
+                    .as_deref()
+                    .is_some_and(|reason| reason.contains("injected usage fence failure"))
+            );
+        }
+    }
+}