fix(scanner): run bootstrap usage rebuild promptly (#7758)

Treat non-authoritative usage floor startup as pending bootstrap rebuild work so reset-published bootstrap markers cannot sit behind clean-idle or empty pause-backlog delay.

Wire recovery wakeups into the normal scanner cycle wait and expose the pending rebuild state in scanner status.

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

crates/scanner/src/scanner.rs

@@ -333,6 +333,7 @@ pub struct ScannerCycleScheduleStatus {
     execution_role: &'static str,
     effective_interval_available: bool,
     effective_interval_seconds: u64,
+    usage_bootstrap_rebuild_pending: bool,
     clean_idle_backoff_enabled: bool,
     clean_idle_backoff_multiplier: u64,
     superseded_retry_backoff_enabled: bool,
@@ -345,6 +346,7 @@ impl Default for ScannerCycleScheduleStatus {
             execution_role: "unknown",
             effective_interval_available: false,
             effective_interval_seconds: 0,
+            usage_bootstrap_rebuild_pending: false,
             clean_idle_backoff_enabled: false,
             clean_idle_backoff_multiplier: 1,
             superseded_retry_backoff_enabled: false,
@@ -368,6 +370,7 @@ pub fn scanner_cycle_schedule_status() -> ScannerCycleScheduleStatus {
 
 fn record_scanner_cycle_schedule(
     effective_interval: Duration,
+    usage_bootstrap_rebuild_pending: bool,
     clean_idle_backoff_enabled: bool,
     clean_idle_backoff_multiplier: u64,
     superseded_retry_backoff_enabled: bool,
@@ -383,6 +386,7 @@ fn record_scanner_cycle_schedule(
         execution_role: "leader",
         effective_interval_available: true,
         effective_interval_seconds,
+        usage_bootstrap_rebuild_pending,
         clean_idle_backoff_enabled,
         clean_idle_backoff_multiplier: clean_idle_backoff_multiplier.max(1),
         superseded_retry_backoff_enabled,
@@ -1160,6 +1164,56 @@ impl ScannerMaintenanceFeatures {
     }
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+struct ScannerUsageBootstrapRebuild {
+    pending: bool,
+}
+
+impl ScannerUsageBootstrapRebuild {
+    fn from_startup(startup: PersistedUsageFloorStartup) -> Self {
+        Self {
+            pending: startup != PersistedUsageFloorStartup::Authoritative,
+        }
+    }
+
+    fn pending(self) -> bool {
+        self.pending
+    }
+
+    fn wait_plan(self, mut plan: ScannerCycleWaitPlan, convergence_retry_interval: Option<Duration>) -> ScannerCycleWaitPlan {
+        if self.pending && convergence_retry_interval.is_none() {
+            plan.delay = Duration::ZERO;
+        }
+        plan
+    }
+
+    fn clean_idle_backoff_enabled(self, enabled: bool) -> bool {
+        enabled && !self.pending
+    }
+
+    fn requires_full_scan(
+        self,
+        maintenance_features: ScannerMaintenanceFeatures,
+        observed_generation: Option<u64>,
+        current_generation: u64,
+        wake: ScannerCycleWakeReason,
+    ) -> bool {
+        self.pending || maintenance_features.requires_full_scan(observed_generation, current_generation, wake)
+    }
+
+    fn record_cycle(&mut self, outcome: ScannerCycleOutcome) -> bool {
+        if matches!(
+            outcome,
+            ScannerCycleOutcome::Completed | ScannerCycleOutcome::CompletedWithPendingMaintenance
+        ) {
+            let was_pending = self.pending;
+            self.pending = false;
+            return was_pending;
+        }
+        false
+    }
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum MaintenanceInspectionDecision {
     Accept,
@@ -2770,6 +2824,7 @@ where
     };
     let (allow_usage_floor_bootstrap_pending, usage_floor_cycle_reset_policy) =
         prepare_cycle_for_usage_floor_bootstrap(&mut cycle_info, usage_floor, usage_floor_startup);
+    let mut usage_bootstrap_rebuild = ScannerUsageBootstrapRebuild::from_startup(usage_floor_startup);
     apply_persisted_usage_floor(&mut cycle_info, &mut leader_epoch, usage_floor);
     match usage_floor_startup {
         PersistedUsageFloorStartup::Authoritative
@@ -2888,7 +2943,13 @@ where
         return Ok(());
     }
 
-    let initial_pause_backlog_attempt = pause_backlog.begin_attempt(scanner_pause_backlog_now()).await;
+    let initial_pause_backlog_attempt = if usage_bootstrap_rebuild.pending() {
+        pause_backlog
+            .begin_usage_bootstrap_rebuild_attempt(scanner_pause_backlog_now())
+            .await
+    } else {
+        pause_backlog.begin_attempt(scanner_pause_backlog_now()).await
+    };
     if !ctx.is_cancelled()
         && matches!(
             initial_pause_backlog_attempt,
@@ -2896,6 +2957,7 @@ where
         )
     {
         // Preserve previous behavior: run one cycle immediately after lock acquisition.
+        let usage_bootstrap_pending_before_cycle = usage_bootstrap_rebuild.pending();
         let dirty_generation_before_cycle = dirty_usage_generation();
         let dirty_usage_pending_before_cycle = dirty_usage_buckets_pending();
         let maintenance_generation_before_cycle = scanner_maintenance_generation();
@@ -2958,6 +3020,9 @@ where
             }
         };
         finish_scanner_pause_backlog_cycle(&mut pause_backlog, &storeapi, initial_pause_backlog_attempt, initial_outcome).await;
+        if usage_bootstrap_rebuild.record_cycle(initial_outcome) {
+            clean_idle_backoff.reset();
+        }
         superseded_backoff.record_retryable_cycle(initial_outcome == ScannerCycleOutcome::Superseded);
         deferred_backoff.record_retryable_cycle(matches!(initial_outcome, ScannerCycleOutcome::Deferred(_)));
         dirty_usage_generation_seen = dirty_generation_before_cycle;
@@ -2983,12 +3048,12 @@ where
             scanner_activity_backoff_blocked = true;
         }
         let scanner_activity_ready = !scanner_activity_backoff_blocked && scanner_activity_seen.is_some();
-        let backoff_enabled = scanner_clean_idle_backoff_enabled(
+        let backoff_enabled = usage_bootstrap_rebuild.clean_idle_backoff_enabled(scanner_clean_idle_backoff_enabled(
             clean_idle_topology_supported,
             scanner_activity_ready,
             maintenance_features,
             &runtime_config,
-        );
+        ));
         record_scanner_cycle_result(
             &mut clean_idle_backoff,
             &runtime_config,
@@ -2999,7 +3064,8 @@ where
                 dirty_usage_pending_before_cycle,
                 dirty_generation_before_cycle,
                 dirty_usage_generation(),
-            ) || maintenance_generation_before_cycle != scanner_maintenance_generation()
+            ) || usage_bootstrap_pending_before_cycle
+                || maintenance_generation_before_cycle != scanner_maintenance_generation()
                 || scanner_activity_observed_work(scanner_activity_observation),
         );
         runtime_config_generation_seen = scanner_runtime_config_generation();
@@ -3040,12 +3106,12 @@ where
             scanner_activity_seen = None;
         }
         let scanner_activity_ready = !scanner_activity_backoff_blocked && scanner_activity_seen.is_some();
-        let backoff_enabled = scanner_clean_idle_backoff_enabled(
+        let backoff_enabled = usage_bootstrap_rebuild.clean_idle_backoff_enabled(scanner_clean_idle_backoff_enabled(
             clean_idle_topology_supported,
             scanner_activity_ready,
             maintenance_features,
             &runtime_config,
-        );
+        ));
         let mut wait_plan =
             scanner_cycle_wait_plan(&runtime_config, clean_idle_backoff, backoff_enabled, randomized_cycle_delay_for);
         let superseded_retry_interval = scanner_superseded_retry_interval(superseded_backoff, &runtime_config);
@@ -3055,16 +3121,20 @@ where
             wait_plan.effective_interval = retry_interval;
             wait_plan.delay = randomized_cycle_delay_for(retry_interval).min(retry_interval);
         }
-        if let Some(pause_backlog_delay) = pause_backlog.scheduling_delay(scanner_pause_backlog_now()) {
+        if let Some(pause_backlog_delay) =
+            pause_backlog.scheduling_delay(scanner_pause_backlog_now(), usage_bootstrap_rebuild.pending())
+        {
             wait_plan.effective_interval = pause_backlog_delay.max(Duration::from_secs(1));
             wait_plan.delay = pause_backlog_delay;
             convergence_retry_interval = Some(pause_backlog_delay.max(Duration::from_secs(1)));
         }
+        wait_plan = usage_bootstrap_rebuild.wait_plan(wait_plan, convergence_retry_interval);
         let dirty_generation_before_wait = dirty_usage_generation();
         let dirty_usage_pending_before_wait = dirty_usage_buckets_pending();
         let maintenance_generation_before_wait = scanner_maintenance_generation();
         record_scanner_cycle_schedule(
             wait_plan.effective_interval,
+            usage_bootstrap_rebuild.pending(),
             backoff_enabled,
             u64::from(clean_idle_backoff.interval_multiplier),
             superseded_retry_interval.is_some(),
@@ -3079,6 +3149,7 @@ where
             effective_interval = ?wait_plan.effective_interval,
             clean_idle_max_interval = ?wait_plan.clean_idle_max_interval,
             scheduled_delay = ?wait_plan.delay,
+            usage_bootstrap_rebuild_pending = usage_bootstrap_rebuild.pending(),
             interval_multiplier = clean_idle_backoff.interval_multiplier,
             clean_idle_backoff_enabled = backoff_enabled,
             superseded_retry_backoff_enabled = superseded_retry_interval.is_some(),
@@ -3101,6 +3172,7 @@ where
             movement_changed,
             current_movement_generation: move || movement_store.scanner_data_movement_generation(),
             is_lock_lost: || guard.is_lock_lost(),
+            recovery_wake: Some(&SCANNER_CYCLE_RECOVERY_WAKE),
         };
         let wake_reason = wait_for_next_scanner_cycle_with_activity_and_movement(
             &ctx,
@@ -3149,7 +3221,8 @@ where
             ScannerCycleWakeReason::Timer
             | ScannerCycleWakeReason::DirtyUsage
             | ScannerCycleWakeReason::ClusterActivity
-            | ScannerCycleWakeReason::ClusterActivityUnavailable => {}
+            | ScannerCycleWakeReason::ClusterActivityUnavailable
+            | ScannerCycleWakeReason::Recovery => {}
         }
 
         if wake_reason == ScannerCycleWakeReason::DirtyUsage {
@@ -3191,13 +3264,20 @@ where
         if pause_backlog_observation.paused {
             continue;
         }
-        let pause_backlog_attempt = pause_backlog.begin_attempt(scanner_pause_backlog_now()).await;
+        let pause_backlog_attempt = if usage_bootstrap_rebuild.pending() {
+            pause_backlog
+                .begin_usage_bootstrap_rebuild_attempt(scanner_pause_backlog_now())
+                .await
+        } else {
+            pause_backlog.begin_attempt(scanner_pause_backlog_now()).await
+        };
         if matches!(
             pause_backlog_attempt,
             ScannerPauseBacklogAttemptDecision::RateLimited | ScannerPauseBacklogAttemptDecision::PersistenceUnavailable
         ) {
             continue;
         }
+        let usage_bootstrap_pending_before_cycle = usage_bootstrap_rebuild.pending();
         let dirty_generation_before_cycle = dirty_usage_generation();
         let cycle_ctx = ctx.child_token();
         let cycle_budget = ScannerCycleBudget::new_with_runtime_progress_tracking(&cycle_ctx, scanner_cycle_budget_config());
@@ -3212,7 +3292,8 @@ where
                 leader_epoch,
                 cycle_budget.clone(),
                 ScannerCycleScheduling {
-                    requires_full_scan: maintenance_features.requires_full_scan(
+                    requires_full_scan: usage_bootstrap_rebuild.requires_full_scan(
+                        maintenance_features,
                         maintenance_generation_seen,
                         scanner_maintenance_generation(),
                         wake_reason,
@@ -3256,6 +3337,9 @@ where
             }
         };
         finish_scanner_pause_backlog_cycle(&mut pause_backlog, &storeapi, pause_backlog_attempt, outcome).await;
+        if usage_bootstrap_rebuild.record_cycle(outcome) {
+            clean_idle_backoff.reset();
+        }
         superseded_backoff.record_retryable_cycle(outcome == ScannerCycleOutcome::Superseded);
         deferred_backoff.record_retryable_cycle(matches!(outcome, ScannerCycleOutcome::Deferred(_)));
         dirty_usage_generation_seen = dirty_generation_before_cycle;
@@ -3314,12 +3398,12 @@ where
             scanner_activity_backoff_blocked = true;
         }
         let scanner_activity_ready = !scanner_activity_backoff_blocked && scanner_activity_seen.is_some();
-        let backoff_enabled = scanner_clean_idle_backoff_enabled(
+        let backoff_enabled = usage_bootstrap_rebuild.clean_idle_backoff_enabled(scanner_clean_idle_backoff_enabled(
             clean_idle_topology_supported,
             scanner_activity_ready,
             maintenance_features,
             &runtime_config,
-        );
+        ));
         record_scanner_cycle_result(
             &mut clean_idle_backoff,
             &runtime_config,
@@ -3330,7 +3414,8 @@ where
                 dirty_usage_pending_before_wait,
                 dirty_generation_before_wait,
                 dirty_usage_generation(),
-            ) || scanner_activity_observed_work(scanner_activity_observation),
+            ) || usage_bootstrap_pending_before_cycle
+                || scanner_activity_observed_work(scanner_activity_observation),
         );
     }
 

crates/scanner/src/scanner/activity.rs

@@ -27,6 +27,7 @@ pub(super) enum ScannerCycleWakeReason {
     ClusterActivityUnavailable,
     RuntimeConfig,
     MaintenanceConfig,
+    Recovery,
     LeaderLockLost,
     Cancelled,
 }
@@ -453,11 +454,12 @@ impl ScannerCycleObservedGenerations {
 ///
 /// Keeping the movement inputs together makes it harder for callers to pair a
 /// generation with the wrong notification or lock predicate.
-pub(super) struct ScannerMovementWaitContext<G, F> {
+pub(super) struct ScannerMovementWaitContext<'a, G, F> {
     pub(super) movement_generation_seen: Option<u64>,
     pub(super) movement_changed: Arc<Notify>,
     pub(super) current_movement_generation: G,
     pub(super) is_lock_lost: F,
+    pub(super) recovery_wake: Option<&'a Notify>,
 }
 
 pub(super) const LOCAL_SCANNER_ACTIVITY_NODE: &str = "<local>";
@@ -682,6 +684,7 @@ where
         movement_changed: Arc::new(Notify::new()),
         current_movement_generation: || 0,
         is_lock_lost,
+        recovery_wake: None,
     };
     wait_for_next_scanner_cycle_with_movement(
         ctx,
@@ -701,7 +704,7 @@ pub(super) async fn wait_for_next_scanner_cycle_with_movement<G, F>(
     ctx: &CancellationToken,
     delay: Duration,
     generations: ScannerCycleObservedGenerations,
-    movement: &ScannerMovementWaitContext<G, F>,
+    movement: &ScannerMovementWaitContext<'_, G, F>,
 ) -> ScannerCycleWakeReason
 where
     F: Fn() -> bool,
@@ -747,9 +750,17 @@ where
         {
             return ScannerCycleWakeReason::MovementGeneration;
         }
+        let recovery_notification = async {
+            match movement.recovery_wake {
+                Some(wake) => wake.notified().await,
+                None => std::future::pending::<()>().await,
+            }
+        };
+        tokio::pin!(recovery_notification);
         tokio::select! {
             _ = ctx.cancelled() => return ScannerCycleWakeReason::Cancelled,
             _ = &mut sleep => return ScannerCycleWakeReason::Timer,
+            _ = &mut recovery_notification => return ScannerCycleWakeReason::Recovery,
             _ = &mut lock_poll => {
                 if (movement.is_lock_lost)() {
                     return ScannerCycleWakeReason::LeaderLockLost;
@@ -812,6 +823,7 @@ where
         movement_changed: Arc::new(Notify::new()),
         current_movement_generation: || 0,
         is_lock_lost,
+        recovery_wake: None,
     };
     wait_for_next_scanner_cycle_with_activity_and_movement(
         ctx,
@@ -831,7 +843,7 @@ pub(super) async fn wait_for_next_scanner_cycle_with_activity_and_movement<F, G,
     activity_poll_interval: Option<Duration>,
     activity_seen: &mut Option<ScannerActivitySnapshot>,
     generations: ScannerCycleObservedGenerations,
-    movement: ScannerMovementWaitContext<G, F>,
+    movement: ScannerMovementWaitContext<'_, G, F>,
     mut probe_activity: Probe,
 ) -> ScannerCycleWakeReason
 where

crates/scanner/src/scanner/backlog.rs

@@ -428,6 +428,23 @@ impl ScannerPauseBacklogLedger {
         ScannerPauseBacklogAttemptDecision::Tracked(serial)
     }
 
+    fn blocks_usage_bootstrap_rebuild(&self) -> bool {
+        match self.phase {
+            ScannerPauseBacklogPhase::Idle => false,
+            ScannerPauseBacklogPhase::Paused => true,
+            ScannerPauseBacklogPhase::CatchingUp | ScannerPauseBacklogPhase::RetryExhausted => {
+                self.pending_full_scan || self.pending_work_items() != 0 || self.has_unfinished_attempt()
+            }
+        }
+    }
+
+    fn begin_usage_bootstrap_rebuild_attempt(&mut self, now: u64) -> ScannerPauseBacklogAttemptDecision {
+        if !self.blocks_usage_bootstrap_rebuild() {
+            return ScannerPauseBacklogAttemptDecision::Untracked;
+        }
+        self.begin_attempt(now)
+    }
+
     fn finish_attempt(
         &mut self,
         serial: u64,
@@ -1613,10 +1630,13 @@ where
         controller
     }
 
-    pub(super) fn scheduling_delay(&self, now: u64) -> Option<Duration> {
+    pub(super) fn scheduling_delay(&self, now: u64, usage_bootstrap_rebuild_pending: bool) -> Option<Duration> {
         if self.persistence_disabled {
             return Some(Duration::from_secs(self.persistence_retry_at_unix_secs.saturating_sub(now)));
         }
+        if usage_bootstrap_rebuild_pending && !self.loaded.ledger.blocks_usage_bootstrap_rebuild() {
+            return None;
+        }
         match self.loaded.ledger.phase {
             ScannerPauseBacklogPhase::Idle => None,
             ScannerPauseBacklogPhase::Paused => Some(Duration::from_secs(SCANNER_PAUSE_REFRESH_INTERVAL_SECONDS)),
@@ -1638,11 +1658,24 @@ where
     }
 
     pub(super) async fn begin_attempt(&mut self, now: u64) -> ScannerPauseBacklogAttemptDecision {
+        self.begin_attempt_with(now, ScannerPauseBacklogLedger::begin_attempt).await
+    }
+
+    pub(super) async fn begin_usage_bootstrap_rebuild_attempt(&mut self, now: u64) -> ScannerPauseBacklogAttemptDecision {
+        self.begin_attempt_with(now, ScannerPauseBacklogLedger::begin_usage_bootstrap_rebuild_attempt)
+            .await
+    }
+
+    async fn begin_attempt_with(
+        &mut self,
+        now: u64,
+        begin: fn(&mut ScannerPauseBacklogLedger, u64) -> ScannerPauseBacklogAttemptDecision,
+    ) -> ScannerPauseBacklogAttemptDecision {
         if self.persistence_disabled {
             return ScannerPauseBacklogAttemptDecision::PersistenceUnavailable;
         }
         let mut candidate = self.loaded.ledger.clone();
-        let decision = candidate.begin_attempt(now);
+        let decision = begin(&mut candidate, now);
         if candidate == self.loaded.ledger {
             self.record_status(now);
             return decision;
@@ -3554,6 +3587,40 @@ mod tests {
         assert_eq!(decode_valid_ledger(&ledger).phase, ScannerPauseBacklogPhase::Idle);
     }
 
+    #[test]
+    fn usage_bootstrap_rebuild_bypasses_empty_catch_up_backlog() {
+        let mut empty_catch_up = durable_ledger(100);
+        empty_catch_up.phase = ScannerPauseBacklogPhase::CatchingUp;
+        empty_catch_up.pending_full_scan = false;
+        empty_catch_up.next_attempt_at_unix_secs = 420;
+
+        assert!(!empty_catch_up.blocks_usage_bootstrap_rebuild());
+        assert_eq!(
+            empty_catch_up.begin_usage_bootstrap_rebuild_attempt(120),
+            ScannerPauseBacklogAttemptDecision::Untracked
+        );
+        assert_eq!(
+            empty_catch_up.next_attempt_at_unix_secs, 420,
+            "usage bootstrap rebuild must not rewrite an unrelated empty catch-up ledger"
+        );
+
+        let mut movement_full_scan = empty_catch_up.clone();
+        movement_full_scan.pending_full_scan = true;
+        assert!(movement_full_scan.blocks_usage_bootstrap_rebuild());
+        assert_eq!(
+            movement_full_scan.begin_usage_bootstrap_rebuild_attempt(120),
+            ScannerPauseBacklogAttemptDecision::RateLimited
+        );
+
+        let mut known_work = empty_catch_up;
+        known_work.discovered_expiry_items = 1;
+        assert!(known_work.blocks_usage_bootstrap_rebuild());
+        assert_eq!(
+            known_work.begin_usage_bootstrap_rebuild_attempt(120),
+            ScannerPauseBacklogAttemptDecision::RateLimited
+        );
+    }
+
     #[test]
     fn fourth_completed_known_work_attempt_preserves_window_end_before_convergence() {
         let mut ledger = durable_ledger(100);

crates/scanner/src/scanner/tests.rs

@@ -8442,13 +8442,14 @@ fn scanner_cycle_wait_plan_drives_growth_resets_and_bitrot_cap() {
 #[test]
 #[serial]
 fn scanner_cycle_schedule_status_reports_effective_backoff() {
-    record_scanner_cycle_schedule(Duration::from_millis(86_400_001), true, 2_048, true, 7);
+    record_scanner_cycle_schedule(Duration::from_millis(86_400_001), true, true, 2_048, true, 7);
 
     let status = scanner_cycle_schedule_status();
 
     assert_eq!(status.execution_role, "leader");
     assert!(status.effective_interval_available);
     assert_eq!(status.effective_interval_seconds, 86_401);
+    assert!(status.usage_bootstrap_rebuild_pending);
     assert!(status.clean_idle_backoff_enabled);
     assert_eq!(status.clean_idle_backoff_multiplier, 2_048);
     assert!(status.superseded_retry_backoff_enabled);
@@ -8459,12 +8460,14 @@ fn scanner_cycle_schedule_status_reports_effective_backoff() {
     assert_eq!(status.execution_role, "follower");
     assert!(!status.effective_interval_available);
     assert_eq!(status.effective_interval_seconds, 0);
+    assert!(!status.usage_bootstrap_rebuild_pending);
 
     reset_scanner_cycle_schedule();
     let status = scanner_cycle_schedule_status();
     assert_eq!(status.execution_role, "unknown");
     assert!(!status.effective_interval_available);
     assert_eq!(status.effective_interval_seconds, 0);
+    assert!(!status.usage_bootstrap_rebuild_pending);
     assert!(!status.clean_idle_backoff_enabled);
     assert_eq!(status.clean_idle_backoff_multiplier, 1);
     assert!(!status.superseded_retry_backoff_enabled);
@@ -8712,6 +8715,57 @@ fn clean_idle_backoff_policy_preserves_explicit_and_maintenance_cycles() {
     }
 }
 
+#[test]
+fn usage_bootstrap_rebuild_bypasses_clean_idle_until_authoritative_cycle() {
+    let config = ScannerRuntimeConfig {
+        cycle_interval: Duration::from_secs(60),
+        ..Default::default()
+    };
+    let features = ScannerMaintenanceFeatures::default();
+    let generation = Some(scanner_maintenance_generation());
+    let mut clean_idle_backoff = ScannerCleanIdleBackoff { interval_multiplier: 8 };
+    let mut rebuild = ScannerUsageBootstrapRebuild::from_startup(PersistedUsageFloorStartup::BootstrapPending);
+
+    assert!(!rebuild.clean_idle_backoff_enabled(true));
+    assert!(rebuild.requires_full_scan(
+        features,
+        generation,
+        scanner_maintenance_generation(),
+        ScannerCycleWakeReason::DirtyUsage,
+    ));
+
+    let plan = rebuild.wait_plan(scanner_cycle_wait_plan(&config, clean_idle_backoff, true, std::convert::identity), None);
+    assert_eq!(plan.delay, Duration::ZERO);
+
+    assert!(!rebuild.record_cycle(ScannerCycleOutcome::Partial));
+    assert!(rebuild.pending());
+    assert!(rebuild.record_cycle(ScannerCycleOutcome::CompletedWithPendingMaintenance));
+    assert!(!rebuild.pending());
+
+    let mut rebuild = ScannerUsageBootstrapRebuild::from_startup(PersistedUsageFloorStartup::BootstrapPending);
+    assert!(rebuild.record_cycle(ScannerCycleOutcome::Completed));
+    assert!(!rebuild.pending());
+    assert!(rebuild.clean_idle_backoff_enabled(true));
+
+    clean_idle_backoff.reset();
+    record_scanner_cycle_result(
+        &mut clean_idle_backoff,
+        &config,
+        rebuild.clean_idle_backoff_enabled(true),
+        ScannerCycleWakeReason::Timer,
+        ScannerCycleOutcome::Completed,
+        true,
+    );
+    assert_eq!(
+        clean_idle_backoff.effective_interval(
+            config.cycle_interval,
+            scanner_clean_idle_max_interval(config.cycle_interval, &config),
+            true,
+        ),
+        Duration::from_secs(60)
+    );
+}
+
 #[tokio::test]
 #[serial]
 async fn scoped_scan_explicit_bitrot_keeps_dirty_planning_without_idle_backoff() {
@@ -9188,6 +9242,7 @@ async fn movement_generation_wakes_deferred_wait_without_dirty_bucket() {
         movement_changed,
         current_movement_generation: move || movement_generation.load(Ordering::Acquire),
         is_lock_lost: || false,
+        recovery_wake: None,
     };
     let reason = wait_for_next_scanner_cycle_with_movement(
         &ctx,
@@ -9205,6 +9260,39 @@ async fn movement_generation_wakes_deferred_wait_without_dirty_bucket() {
     assert_eq!(reason, ScannerCycleWakeReason::MovementGeneration);
 }
 
+#[tokio::test]
+async fn recovery_wake_interrupts_normal_cycle_wait() {
+    let ctx = CancellationToken::new();
+    let recovery_wake = Notify::new();
+    let movement = ScannerMovementWaitContext {
+        movement_generation_seen: None,
+        movement_changed: Arc::new(Notify::new()),
+        current_movement_generation: || 0,
+        is_lock_lost: || false,
+        recovery_wake: Some(&recovery_wake),
+    };
+
+    let mut wait = Box::pin(wait_for_next_scanner_cycle_with_movement(
+        &ctx,
+        Duration::from_secs(60),
+        ScannerCycleObservedGenerations {
+            dirty_usage: None,
+            runtime_config: crate::runtime_config::scanner_runtime_config_generation(),
+            maintenance: crate::scanner_io::scanner_maintenance_generation(),
+            defer_cluster_activity: false,
+        },
+        &movement,
+    ));
+    assert!(matches!(futures::poll!(&mut wait), Poll::Pending));
+
+    recovery_wake.notify_one();
+    let reason = tokio::time::timeout(Duration::from_secs(1), wait)
+        .await
+        .expect("recovery wake should interrupt the scanner cycle wait");
+
+    assert_eq!(reason, ScannerCycleWakeReason::Recovery);
+}
+
 fn scanner_node_activity(epoch: &str, namespace_generation: u64, maintenance_generation: u64) -> ScannerNodeActivity {
     ScannerNodeActivity {
         instance_id: epoch.to_string(),

docs/operations/scanner-runtime-controls.md

@@ -178,12 +178,27 @@ The reset does not delete metadata files by hand and does not publish an authori
 |---|---|
 | data movement | wait for decommission or rebalance to leave the scanner metadata path, then retry |
 | invalid scanner cycle state | run `POST /v3/scanner/cycle-state/reset` with `{"mode":"full-rescan"}` first |
+| scanner leader lock is busy | retry with an async recovery intent or wait for the current leader to restart/release the lock |
+
+When a running scanner leader already holds `leader.lock`, submit a durable async recovery intent instead of repeatedly calling the synchronous reset:
+
+```json
+{
+  "mode": "full-rebuild",
+  "async": true,
+  "idempotency_key": "<stable-operator-request-id>"
+}
+```
+
+The async form returns HTTP 202 when a new or replayable intent is accepted. Reusing the same `idempotency_key` is safe for retries of the same operator action; a different payload for the same key is rejected as a conflict. Poll `GET /v3/scanner/usage-state/recovery-intents/{intent_id}` until the intent reaches a terminal state.
+
+A successful reset leaves usage state in `bootstrap-pending` only as a fenced rebuild marker. With the scanner enabled, the leader must treat that marker as pending full-rebuild work: clean-idle backoff must not extend the wait, an empty pause-backlog catch-up ledger must not rate-limit the rebuild, and the next admitted scanner cycle must run as a full scan until an authoritative usage snapshot replaces the marker.
 
 ## Cleanup With The Scanner Disabled
 
 With `RUSTFS_SCANNER_ENABLED=false`, startup makes one controlled attempt to finish a previously persisted cycle reset whose validated recovery marker is already `cleanup-pending`. This is metadata cleanup only: it does not start the ordinary scanner loop, scan namespaces, accept a new reset request, or automatically perform a usage-state `full-rebuild`. Missing, merely `blocked`, unknown-version, unknown-phase, or corrupt markers do not authorize an automatic reset.
 
-The attempt uses the existing leader lock and revalidates the observed marker revision and phase after acquiring it. A busy leader or data-movement pause leaves the marker intact and is reported through `cycle_recovery.state` and `cycle_recovery.reason` in the existing scanner status response. There is no automatic retry loop while disabled. After resolving the blocker, explicitly retry `POST /v3/scanner/cycle-state/reset` with `{"mode":"full-rescan"}`, or restart to make another controlled attempt. The v3 reset routes remain synchronous and return their existing successful HTTP 200 responses; no asynchronous HTTP 202 acceptance is introduced.
+The attempt uses the existing leader lock and revalidates the observed marker revision and phase after acquiring it. A busy leader or data-movement pause leaves the marker intact and is reported through `cycle_recovery.state` and `cycle_recovery.reason` in the existing scanner status response. There is no automatic retry loop while disabled. After resolving the blocker, explicitly retry `POST /v3/scanner/cycle-state/reset` with `{"mode":"full-rescan"}`, or restart to make another controlled attempt. Cycle-state reset remains a synchronous HTTP 200 operation; usage-state reset also supports the async recovery-intent form described above.
 
 The startup probe is cancellation-aware and uses the existing cache persistence I/O timeout. Shutdown waits only for the existing server shutdown timeout. If the cleanup task cannot join in that window, the `scanner_cleanup_not_joined` warning means completion is unconfirmed, not drained. The task is not force-aborted or force-unlocked while its runtime remains alive; it retains its existing namespace/admission guards, and durable marker/fence state remains authoritative. Inspect status before retrying. This does not establish a hard deadline for an unresponsive storage operation or prove that I/O has drained when the process or runtime subsequently exits. Task-ownership timeout tests are not storage fsync, commit-tail, or process-crash durability evidence.
 

rustfs/src/admin/handlers/scanner.rs

@@ -827,6 +827,7 @@ mod tests {
         assert_eq!(encoded["cycle_schedule"]["execution_role"], "unknown");
         assert_eq!(encoded["cycle_schedule"]["effective_interval_available"], false);
         assert_eq!(encoded["cycle_schedule"]["effective_interval_seconds"], 0);
+        assert_eq!(encoded["cycle_schedule"]["usage_bootstrap_rebuild_pending"], false);
         assert_eq!(encoded["cycle_schedule"]["clean_idle_backoff_enabled"], false);
         assert_eq!(encoded["cycle_schedule"]["clean_idle_backoff_multiplier"], 1);
         assert_eq!(encoded["cycle_recovery"]["state"], "healthy");