feat(scanner): replay recovery intents at startup (#7354)

Discover durable scanner recovery intents during startup and sequentially re-drive accepted or running usage full-rebuild work. Disabled scanner startup also replays existing durable intents without enabling the normal scanner loop.

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

crates/scanner/src/scanner.rs

@@ -70,12 +70,13 @@ use crate::storage_api::scan::{
     SCANNER_ACTIVITY_PROTOCOL_VERSION,
 };
 use crate::{
-    ECStore, EcstoreError, RUSTFS_META_BUCKET, SCANNER_PUBLICATION_EPOCH_CHANGED, ScannerLifecycleConfigExt as _,
-    ScannerReplicationConfigExt as _, delete_config_with_publication_admission_for_epoch, get_lifecycle_config,
-    get_replication_config, invalidate_admin_data_usage_snapshot_cache, invalidate_data_usage_snapshot_cache, read_config,
-    replace_bucket_usage_memory_from_info, save_config, save_config_shared_with_preconditions_and_lease_fence_and_scope,
-    save_config_with_preconditions, save_config_with_publication_admission_for_epoch, scanner_publication_admission_for_epoch,
-    scanner_publication_epoch, scanner_publication_epoch_changed,
+    DiskError, ECStore, EcstoreError, ListPathRawOptions, RUSTFS_META_BUCKET, SCANNER_PUBLICATION_EPOCH_CHANGED,
+    ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _, delete_config_with_publication_admission_for_epoch,
+    get_lifecycle_config, get_replication_config, invalidate_admin_data_usage_snapshot_cache,
+    invalidate_data_usage_snapshot_cache, list_path_raw, read_config, replace_bucket_usage_memory_from_info, save_config,
+    save_config_shared_with_preconditions_and_lease_fence_and_scope, save_config_with_preconditions,
+    save_config_with_publication_admission_for_epoch, scanner_publication_admission_for_epoch, scanner_publication_epoch,
+    scanner_publication_epoch_changed,
 };
 
 const LOG_COMPONENT_SCANNER: &str = "scanner";
@@ -947,6 +948,22 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
     init_data_scanner_with_storage(ctx, storeapi).await;
 }
 
+async fn run_scanner_usage_recovery_intents_for_startup(
+    ctx: CancellationToken,
+    storeapi: Arc<ECStore>,
+) -> Result<usize, ScannerError> {
+    let intent_ids = scanner_usage_recovery_intents_for_startup(&ctx, storeapi.clone()).await?;
+    let mut attempted = 0usize;
+    for intent_id in intent_ids {
+        if ctx.is_cancelled() {
+            break;
+        }
+        run_scanner_usage_recovery_intent(ctx.child_token(), storeapi.clone(), intent_id).await?;
+        attempted = attempted.saturating_add(1);
+    }
+    Ok(attempted)
+}
+
 /// 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.
@@ -956,10 +973,32 @@ pub async fn init_scanner_with_recovery(
     enabled: bool,
 ) -> Option<tokio::task::JoinHandle<()>> {
     if enabled {
+        if let Err(error) = run_scanner_usage_recovery_intents_for_startup(ctx.clone(), storeapi.clone()).await {
+            warn!(
+                target: "rustfs::scanner",
+                event = EVENT_SCANNER_PERSIST_STATE,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_RUNTIME,
+                state = "recovery_intent_startup_discovery_failed",
+                error = %error,
+                "Scanner recovery intent startup discovery failed"
+            );
+        }
         init_data_scanner(ctx, storeapi).await;
         return None;
     }
     Some(tokio::spawn(async move {
+        if let Err(error) = run_scanner_usage_recovery_intents_for_startup(ctx.clone(), storeapi.clone()).await {
+            warn!(
+                target: "rustfs::scanner",
+                event = EVENT_SCANNER_PERSIST_STATE,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_RUNTIME,
+                state = "recovery_intent_startup_discovery_failed",
+                error = %error,
+                "Disabled scanner recovery intent startup discovery failed"
+            );
+        }
         if let Err(error) = resume_scanner_cycle_cleanup(ctx, storeapi).await {
             warn!(
                 target: "rustfs::scanner",

crates/scanner/src/scanner/cycle_state.rs

@@ -18,6 +18,8 @@ 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 rustfs_filemeta::MetaCacheEntry;
+use std::collections::BTreeSet;
 use std::sync::atomic::AtomicU64;
 use tokio::io::AsyncReadExt as _;
 
@@ -42,6 +44,9 @@ const SCANNER_RECOVERY_INTENT_STATE_RUNNING: &str = "running";
 const SCANNER_RECOVERY_INTENT_STATE_COMPLETED: &str = "completed";
 const SCANNER_RECOVERY_INTENT_STATE_FAILED: &str = "failed";
 pub const SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD: &str = "scanner-usage-full-rebuild";
+const MAX_SCANNER_RECOVERY_INTENT_STARTUP_CANDIDATES: usize = 4096;
+const SCANNER_RECOVERY_INTENT_STARTUP_PAGE_SIZE: usize = 128;
+const MAX_SCANNER_RECOVERY_INTENT_STARTUP_REPLAY: usize = 64;
 
 #[cfg(test)]
 pub(super) mod cleanup_io_fault {
@@ -465,6 +470,14 @@ fn scanner_recovery_intent_path(intent_id: &str) -> Result<String, ScannerError>
     Ok(format!("{SCANNER_RECOVERY_INTENT_PREFIX}/{intent_id}.json"))
 }
 
+fn scanner_recovery_intent_id_from_entry_name(entry_name: &str) -> Option<String> {
+    let relative = entry_name
+        .strip_prefix(SCANNER_RECOVERY_INTENT_PREFIX)
+        .and_then(|name| name.strip_prefix('/'))?;
+    let intent_id = relative.strip_suffix(".json")?;
+    (!intent_id.contains('/') && is_canonical_sha256(intent_id)).then(|| intent_id.to_string())
+}
+
 pub fn scanner_recovery_actor_sha256(actor: &str) -> String {
     sha256_hex(&[b"scanner-recovery-actor-v1", actor.as_bytes()])
 }
@@ -623,6 +636,117 @@ pub async fn get_scanner_usage_recovery_intent(
     read_recovery_intent_record(storeapi, &path).await
 }
 
+fn scanner_recovery_intent_is_replayable(record: &ScannerRecoveryIntentRecord) -> bool {
+    matches!(
+        record.state.as_str(),
+        SCANNER_RECOVERY_INTENT_STATE_ACCEPTED | SCANNER_RECOVERY_INTENT_STATE_RUNNING
+    )
+}
+
+pub(super) async fn scanner_usage_recovery_intents_for_startup(
+    ctx: &CancellationToken,
+    storeapi: Arc<ECStore>,
+) -> Result<Vec<String>, ScannerError> {
+    let discovered = Arc::new(StdMutex::new(BTreeSet::<String>::new()));
+    for set in storeapi.all_set_disks() {
+        if ctx.is_cancelled() {
+            break;
+        }
+        let disks = set.get_local_disks().await;
+        if disks.is_empty() {
+            continue;
+        }
+        let read_quorum = disks.len().saturating_sub(set.default_parity_count).clamp(1, disks.len());
+        let mut forward_to = None;
+        loop {
+            let page_entries = Arc::new(StdMutex::new(Vec::<String>::new()));
+            let page_entries_for_set = page_entries.clone();
+            let list_result = list_path_raw(
+                ctx.child_token(),
+                ListPathRawOptions {
+                    disks: disks.clone(),
+                    bucket: RUSTFS_META_BUCKET.to_string(),
+                    path: SCANNER_RECOVERY_INTENT_PREFIX.to_string(),
+                    recursive: true,
+                    skip_hidden_prefix_check: true,
+                    forward_to: forward_to.clone(),
+                    min_disks: read_quorum,
+                    report_not_found: true,
+                    per_disk_limit: i32::try_from(SCANNER_RECOVERY_INTENT_STARTUP_PAGE_SIZE).unwrap_or(i32::MAX),
+                    agreed: Some(Box::new(move |entry: MetaCacheEntry| {
+                        let page_entries = page_entries_for_set.clone();
+                        Box::pin(async move {
+                            if scanner_recovery_intent_id_from_entry_name(&entry.name).is_some() {
+                                page_entries
+                                    .lock()
+                                    .unwrap_or_else(|poisoned| poisoned.into_inner())
+                                    .push(entry.name);
+                            }
+                        })
+                    })),
+                    ..Default::default()
+                },
+            )
+            .await;
+            match list_result {
+                Ok(()) => {}
+                Err(DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::VolumeNotFound) => break,
+                Err(err) => {
+                    return Err(ScannerError::Other(format!("failed to list scanner recovery intents for startup: {err}")));
+                }
+            }
+
+            let page = {
+                let mut guard = page_entries.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+                guard.sort();
+                guard.dedup();
+                std::mem::take(&mut *guard)
+            };
+            if page.is_empty() {
+                break;
+            }
+
+            {
+                let mut all = discovered.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+                for entry_name in &page {
+                    if let Some(intent_id) = scanner_recovery_intent_id_from_entry_name(entry_name) {
+                        all.insert(intent_id);
+                    }
+                    if all.len() >= MAX_SCANNER_RECOVERY_INTENT_STARTUP_CANDIDATES {
+                        break;
+                    }
+                }
+                if all.len() >= MAX_SCANNER_RECOVERY_INTENT_STARTUP_CANDIDATES {
+                    break;
+                }
+            }
+
+            if page.len() < SCANNER_RECOVERY_INTENT_STARTUP_PAGE_SIZE {
+                break;
+            }
+            forward_to = page.last().cloned();
+        }
+    }
+
+    let intent_ids = {
+        let guard = discovered.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        guard.iter().cloned().collect::<Vec<_>>()
+    };
+    let mut replayable = Vec::new();
+    for intent_id in intent_ids {
+        let Some(record) = get_scanner_usage_recovery_intent(storeapi.clone(), &intent_id).await? else {
+            continue;
+        };
+        if scanner_recovery_intent_is_replayable(&record) {
+            replayable.push(record.intent_id);
+            if replayable.len() >= MAX_SCANNER_RECOVERY_INTENT_STARTUP_REPLAY {
+                break;
+            }
+        }
+    }
+    Ok(replayable)
+}
+
 pub async fn accept_scanner_usage_recovery_intent(
     storeapi: Arc<impl ScannerObjectIO>,
     request: ScannerRecoveryIntentRequest,

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

@@ -106,6 +106,20 @@ fn recovery_intent_request(key: &str, actor: &str) -> ScannerRecoveryIntentReque
     }
 }
 
+fn synthetic_recovery_intent_record(intent_id: String, state: &str) -> ScannerRecoveryIntentRecord {
+    ScannerRecoveryIntentRecord {
+        schema_version: 1,
+        intent_id,
+        action: SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD.to_string(),
+        mode: "full-rebuild".to_string(),
+        state: state.to_string(),
+        actor_sha256: "a".repeat(64),
+        idempotency_key_sha256: "b".repeat(64),
+        request_sha256: "c".repeat(64),
+        accepted_at_unix_secs: 1,
+    }
+}
+
 async fn assert_reset_fences(store: &Arc<ECStore>) {
     let data = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
         .await
@@ -311,6 +325,162 @@ async fn scanner_recovery_intent_executor_persists_failed_progress() {
     assert_eq!(failed.intent_id, record.intent_id);
 }
 
+#[tokio::test]
+#[serial]
+async fn scanner_recovery_intent_startup_discovers_only_non_terminal_records() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    let accepted = match accept_scanner_usage_recovery_intent(
+        store.clone(),
+        recovery_intent_request("intent-key-0001-startup-accepted", "operator-a"),
+    )
+    .await
+    .expect("accepted startup intent should persist")
+    {
+        ScannerRecoveryIntentAcceptResult::Accepted { record } => record,
+        other => panic!("first request must create accepted startup intent: {other:?}"),
+    };
+    let mut running = match accept_scanner_usage_recovery_intent(
+        store.clone(),
+        recovery_intent_request("intent-key-0001-startup-running", "operator-a"),
+    )
+    .await
+    .expect("running startup intent should persist")
+    {
+        ScannerRecoveryIntentAcceptResult::Accepted { record } => record,
+        other => panic!("first request must create running startup intent: {other:?}"),
+    };
+    running.state = "running".to_string();
+    let running_path = format!(".usage.v2.recovery-intents/{}.json", running.intent_id);
+    save_config(
+        store.clone(),
+        &running_path,
+        serde_json::to_vec(&running).expect("running intent should encode"),
+    )
+    .await
+    .expect("running intent override should persist");
+
+    for (key, state) in [
+        ("intent-key-0001-startup-completed", "completed"),
+        ("intent-key-0001-startup-failed", "failed"),
+    ] {
+        let mut terminal = match accept_scanner_usage_recovery_intent(store.clone(), recovery_intent_request(key, "operator-a"))
+            .await
+            .expect("terminal startup intent should persist")
+        {
+            ScannerRecoveryIntentAcceptResult::Accepted { record } => record,
+            other => panic!("first request must create terminal startup intent: {other:?}"),
+        };
+        terminal.state = state.to_string();
+        let path = format!(".usage.v2.recovery-intents/{}.json", terminal.intent_id);
+        save_config(
+            store.clone(),
+            &path,
+            serde_json::to_vec(&terminal).expect("terminal intent should encode"),
+        )
+        .await
+        .expect("terminal intent override should persist");
+    }
+    save_config(store.clone(), ".usage.v2.recovery-intents/not-a-sha.json", b"{}".to_vec())
+        .await
+        .expect("foreign startup key should persist");
+
+    let restarted = restart_scanner_cycle_store_from(&store).await;
+    let replayable = scanner_usage_recovery_intents_for_startup(&CancellationToken::new(), restarted)
+        .await
+        .expect("startup discovery should tolerate terminal and foreign records");
+    let replayable = replayable.into_iter().collect::<std::collections::BTreeSet<_>>();
+    assert_eq!(
+        replayable,
+        [accepted.intent_id, running.intent_id].into_iter().collect(),
+        "startup discovery must only re-drive accepted/running durable intents"
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_recovery_intent_startup_pages_past_terminal_records() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    for index in 0..70 {
+        let intent_id = format!("{index:064x}");
+        let path = format!(".usage.v2.recovery-intents/{intent_id}.json");
+        let terminal = synthetic_recovery_intent_record(intent_id, "completed");
+        save_config(
+            store.clone(),
+            &path,
+            serde_json::to_vec(&terminal).expect("terminal paging fixture should encode"),
+        )
+        .await
+        .expect("terminal paging fixture should persist");
+    }
+    let pending_id = "f".repeat(64);
+    let pending = synthetic_recovery_intent_record(pending_id.clone(), "accepted");
+    save_config(
+        store.clone(),
+        &format!(".usage.v2.recovery-intents/{pending_id}.json"),
+        serde_json::to_vec(&pending).expect("pending paging fixture should encode"),
+    )
+    .await
+    .expect("pending paging fixture should persist");
+
+    let restarted = restart_scanner_cycle_store_from(&store).await;
+    let replayable = scanner_usage_recovery_intents_for_startup(&CancellationToken::new(), restarted)
+        .await
+        .expect("startup discovery should page past terminal records");
+    assert_eq!(replayable, vec![pending_id]);
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_recovery_intent_startup_rejects_corrupt_pending_record() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    let record = match accept_scanner_usage_recovery_intent(
+        store.clone(),
+        recovery_intent_request("intent-key-0001-startup-corrupt", "operator-a"),
+    )
+    .await
+    .expect("corrupt startup fixture intent should persist")
+    {
+        ScannerRecoveryIntentAcceptResult::Accepted { record } => record,
+        other => panic!("first request must create corrupt startup fixture: {other:?}"),
+    };
+    let path = format!(".usage.v2.recovery-intents/{}.json", record.intent_id);
+    save_config(store.clone(), &path, b"{corrupt".to_vec())
+        .await
+        .expect("corrupt intent payload should persist");
+
+    let restarted = restart_scanner_cycle_store_from(&store).await;
+    let error = scanner_usage_recovery_intents_for_startup(&CancellationToken::new(), restarted)
+        .await
+        .expect_err("startup discovery must not silently drop corrupt pending intent records");
+    assert!(error.to_string().contains("scanner recovery intent is invalid"), "{error}");
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_recovery_intent_disabled_startup_executes_persisted_non_terminal_intent() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    let record = match accept_scanner_usage_recovery_intent(
+        store.clone(),
+        recovery_intent_request("intent-key-0001-startup-exec", "operator-a"),
+    )
+    .await
+    .expect("startup execution intent should persist")
+    {
+        ScannerRecoveryIntentAcceptResult::Accepted { record } => record,
+        other => panic!("first request must create startup execution intent: {other:?}"),
+    };
+
+    let restarted = restart_scanner_cycle_store_from(&store).await;
+    run_disabled_startup(CancellationToken::new(), restarted.clone()).await;
+
+    let completed = get_scanner_usage_recovery_intent(restarted, &record.intent_id)
+        .await
+        .expect("startup-executed intent should read")
+        .expect("startup-executed intent should remain durable");
+    assert_eq!(completed.state, "completed");
+    assert_eq!(completed.intent_id, record.intent_id);
+}
+
 #[tokio::test]
 #[serial]
 async fn scanner_recovery_intent_rejects_same_namespace_conflict() {