diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index f2337c3ea..b8c16f994 100644 --- a/crates/scanner/src/scanner.rs +++ b/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) { init_data_scanner_with_storage(ctx, storeapi).await; } +async fn run_scanner_usage_recovery_intents_for_startup( + ctx: CancellationToken, + storeapi: Arc, +) -> Result { + 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> { 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", diff --git a/crates/scanner/src/scanner/cycle_state.rs b/crates/scanner/src/scanner/cycle_state.rs index 646fbeead..f20374ae0 100644 --- a/crates/scanner/src/scanner/cycle_state.rs +++ b/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 Ok(format!("{SCANNER_RECOVERY_INTENT_PREFIX}/{intent_id}.json")) } +fn scanner_recovery_intent_id_from_entry_name(entry_name: &str) -> Option { + 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, +) -> Result, ScannerError> { + let discovered = Arc::new(StdMutex::new(BTreeSet::::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::::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::>() + }; + 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, request: ScannerRecoveryIntentRequest, diff --git a/crates/scanner/src/scanner/tests/recovery_control.rs b/crates/scanner/src/scanner/tests/recovery_control.rs index 5f927011a..dacd83a64 100644 --- a/crates/scanner/src/scanner/tests/recovery_control.rs +++ b/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) { 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::>(); + 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() {