diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index da392d363..e38fb583c 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -228,6 +228,92 @@ impl Metric { } } +const SCANNER_CHECKPOINT_EVENT_SET: &str = "set"; +const SCANNER_CHECKPOINT_EVENT_USED: &str = "used"; +const SCANNER_CHECKPOINT_EVENT_IGNORED: &str = "ignored"; +const SCANNER_CHECKPOINT_EVENT_STALE: &str = "stale"; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ScannerWorkSource { + Usage, + Lifecycle, + BucketReplication, + SiteReplication, + Heal, + Bitrot, + Alerts, +} + +impl ScannerWorkSource { + const ALL: [Self; 7] = [ + Self::Usage, + Self::Lifecycle, + Self::BucketReplication, + Self::SiteReplication, + Self::Heal, + Self::Bitrot, + Self::Alerts, + ]; + + pub fn as_str(self) -> &'static str { + match self { + Self::Usage => "usage", + Self::Lifecycle => "lifecycle", + Self::BucketReplication => "bucket_replication", + Self::SiteReplication => "site_replication", + Self::Heal => "heal", + Self::Bitrot => "bitrot", + Self::Alerts => "alerts", + } + } + + fn all() -> &'static [Self] { + &Self::ALL + } + + fn index(self) -> usize { + match self { + Self::Usage => 0, + Self::Lifecycle => 1, + Self::BucketReplication => 2, + Self::SiteReplication => 3, + Self::Heal => 4, + Self::Bitrot => 5, + Self::Alerts => 6, + } + } +} + +#[derive(Debug, Default)] +struct ScannerSourceWorkCounters { + checked: AtomicU64, + queued: AtomicU64, + executed: AtomicU64, + failed: AtomicU64, + skipped: AtomicU64, +} + +impl ScannerSourceWorkCounters { + fn add(&self, checked: u64, queued: u64, executed: u64, failed: u64, skipped: u64) { + self.checked.fetch_add(checked, Ordering::Relaxed); + self.queued.fetch_add(queued, Ordering::Relaxed); + self.executed.fetch_add(executed, Ordering::Relaxed); + self.failed.fetch_add(failed, Ordering::Relaxed); + self.skipped.fetch_add(skipped, Ordering::Relaxed); + } + + fn snapshot(&self, source: ScannerWorkSource) -> ScannerSourceWorkSnapshot { + ScannerSourceWorkSnapshot { + source: source.as_str().to_string(), + checked: self.checked.load(Ordering::Relaxed), + queued: self.queued.load(Ordering::Relaxed), + executed: self.executed.load(Ordering::Relaxed), + failed: self.failed.load(Ordering::Relaxed), + skipped: self.skipped.load(Ordering::Relaxed), + } + } +} + // --------------------------------------------------------------------------- // LockedLastMinuteLatency // --------------------------------------------------------------------------- @@ -423,6 +509,12 @@ pub struct Metrics { scanner_cycle_max_directories: AtomicU64, scanner_bitrot_cycle_enabled: AtomicBool, scanner_bitrot_cycle_millis: AtomicU64, + scanner_checkpoint: Mutex>, + scanner_checkpoint_used: AtomicU64, + scanner_checkpoint_cleared: AtomicU64, + scanner_checkpoint_ignored: AtomicU64, + scanner_checkpoint_stale: AtomicU64, + scanner_source_work: Vec, partial_scan_cycles: AtomicU64, } @@ -495,6 +587,24 @@ pub struct ScannerTimedAction { pub bytes: u64, } +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct ScannerCheckpointReport { + pub version: u16, + pub resume_after: String, + pub reason: String, + pub last_event: String, +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct ScannerSourceWorkSnapshot { + pub source: String, + pub checked: u64, + pub queued: u64, + pub executed: u64, + pub failed: u64, + pub skipped: u64, +} + #[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct ScannerLastMinute { pub actions: HashMap, @@ -614,6 +724,19 @@ pub struct ScannerMetricsReport { #[serde(default)] pub bitrot_cycle_seconds: f64, #[serde(default)] + #[serde(skip_serializing_if = "Option::is_none")] + pub scan_checkpoint: Option, + #[serde(default)] + pub scan_checkpoint_used: u64, + #[serde(default)] + pub scan_checkpoint_cleared: u64, + #[serde(default)] + pub scan_checkpoint_ignored: u64, + #[serde(default)] + pub scan_checkpoint_stale: u64, + #[serde(default)] + pub source_work: Vec, + #[serde(default)] pub partial_cycles: u64, } @@ -661,6 +784,10 @@ fn duration_millis_saturated(duration: Duration) -> u64 { duration.as_millis().min(u64::MAX as u128) as u64 } +fn usize_to_u64_saturated(value: usize) -> u64 { + u64::try_from(value).unwrap_or(u64::MAX) +} + fn scaled_f64_to_u64_saturated(value: f64, scale: f64) -> u64 { if !value.is_finite() || value <= 0.0 { return 0; @@ -786,6 +913,15 @@ impl Metrics { scanner_cycle_max_directories: AtomicU64::new(0), scanner_bitrot_cycle_enabled: AtomicBool::new(false), scanner_bitrot_cycle_millis: AtomicU64::new(0), + scanner_checkpoint: Mutex::new(None), + scanner_checkpoint_used: AtomicU64::new(0), + scanner_checkpoint_cleared: AtomicU64::new(0), + scanner_checkpoint_ignored: AtomicU64::new(0), + scanner_checkpoint_stale: AtomicU64::new(0), + scanner_source_work: ScannerWorkSource::all() + .iter() + .map(|_| ScannerSourceWorkCounters::default()) + .collect(), partial_scan_cycles: AtomicU64::new(0), } } @@ -806,6 +942,7 @@ impl Metrics { move |_custom: &HashMap| { let duration = SystemTime::now().duration_since(start).unwrap_or_default(); global_metrics().operations[metric_idx].fetch_add(1, Ordering::Relaxed); + global_metrics().record_source_work_for_metric(metric, 1); emit_otel_counter(metric_idx, 1); if metric_idx < Metric::LastRealtime as usize { global_metrics().latency[metric_idx].add(duration); @@ -821,6 +958,7 @@ impl Metrics { move |size: u64| { let duration = SystemTime::now().duration_since(start).unwrap_or_default(); global_metrics().operations[metric_idx].fetch_add(1, Ordering::Relaxed); + global_metrics().record_source_work_for_metric(metric, 1); emit_otel_counter(metric_idx, 1); if metric_idx < Metric::LastRealtime as usize { global_metrics().latency[metric_idx].add_size(duration, size); @@ -836,6 +974,7 @@ impl Metrics { move || { let duration = SystemTime::now().duration_since(start).unwrap_or_default(); global_metrics().operations[metric_idx].fetch_add(1, Ordering::Relaxed); + global_metrics().record_source_work_for_metric(metric, 1); emit_otel_counter(metric_idx, 1); if metric_idx < Metric::LastRealtime as usize { global_metrics().latency[metric_idx].add(duration); @@ -851,8 +990,10 @@ impl Metrics { Box::new(move |count: usize| { Box::new(move || { let duration = SystemTime::now().duration_since(start).unwrap_or_default(); - global_metrics().operations[metric_idx].fetch_add(count as u64, Ordering::Relaxed); - emit_otel_counter(metric_idx, count as u64); + let count = usize_to_u64_saturated(count); + global_metrics().operations[metric_idx].fetch_add(count, Ordering::Relaxed); + global_metrics().record_source_work_for_metric(metric, count); + emit_otel_counter(metric_idx, count); if metric_idx < Metric::LastRealtime as usize { global_metrics().latency[metric_idx].add(duration); } @@ -885,6 +1026,7 @@ impl Metrics { pub fn inc_time(metric: Metric, duration: Duration) { let metric_idx = metric as usize; global_metrics().operations[metric_idx].fetch_add(1, Ordering::Relaxed); + global_metrics().record_source_work_for_metric(metric, 1); emit_otel_counter(metric_idx, 1); if metric_idx < Metric::LastRealtime as usize { global_metrics().latency[metric_idx].add(duration); @@ -921,6 +1063,94 @@ impl Metrics { pub fn record_scanner_ilm_action(&self, count: u64) { self.scanner_ilm_actions.fetch_add(count, Ordering::Relaxed); + self.record_scanner_source_work(ScannerWorkSource::Lifecycle, 0, 0, count, 0, 0); + } + + pub fn record_scanner_checkpoint_set(&self, version: u16, resume_after: impl Into, reason: impl Into) { + let checkpoint = ScannerCheckpointReport { + version, + resume_after: resume_after.into(), + reason: reason.into(), + last_event: SCANNER_CHECKPOINT_EVENT_SET.to_string(), + }; + match self.scanner_checkpoint.lock() { + Ok(mut current) => *current = Some(checkpoint), + Err(poisoned) => *poisoned.into_inner() = Some(checkpoint), + } + } + + pub fn record_scanner_checkpoint_used(&self) { + self.scanner_checkpoint_used.fetch_add(1, Ordering::Relaxed); + self.update_scanner_checkpoint_event(SCANNER_CHECKPOINT_EVENT_USED); + } + + pub fn record_scanner_checkpoint_cleared(&self) { + self.scanner_checkpoint_cleared.fetch_add(1, Ordering::Relaxed); + match self.scanner_checkpoint.lock() { + Ok(mut current) => *current = None, + Err(poisoned) => *poisoned.into_inner() = None, + } + } + + pub fn record_scanner_checkpoint_ignored(&self) { + self.scanner_checkpoint_ignored.fetch_add(1, Ordering::Relaxed); + self.update_scanner_checkpoint_event(SCANNER_CHECKPOINT_EVENT_IGNORED); + } + + pub fn record_scanner_checkpoint_stale(&self) { + self.scanner_checkpoint_stale.fetch_add(1, Ordering::Relaxed); + self.update_scanner_checkpoint_event(SCANNER_CHECKPOINT_EVENT_STALE); + } + + pub fn record_scanner_source_work( + &self, + source: ScannerWorkSource, + checked: u64, + queued: u64, + executed: u64, + failed: u64, + skipped: u64, + ) { + if let Some(counters) = self.scanner_source_work.get(source.index()) { + counters.add(checked, queued, executed, failed, skipped); + } + } + + fn update_scanner_checkpoint_event(&self, event: &str) { + match self.scanner_checkpoint.lock() { + Ok(mut current) => { + if let Some(checkpoint) = current.as_mut() { + checkpoint.last_event = event.to_string(); + } + } + Err(poisoned) => { + let mut current = poisoned.into_inner(); + if let Some(checkpoint) = current.as_mut() { + checkpoint.last_event = event.to_string(); + } + } + } + } + + fn record_source_work_for_metric(&self, metric: Metric, count: u64) { + match metric { + Metric::ScanObject | Metric::ScanFolder => { + self.record_scanner_source_work(ScannerWorkSource::Usage, count, 0, 0, 0, 0); + } + Metric::SaveUsage => { + self.record_scanner_source_work(ScannerWorkSource::Usage, 0, 0, count, 0, 0); + } + Metric::CheckReplication => { + self.record_scanner_source_work(ScannerWorkSource::BucketReplication, count, 0, 0, 0, 0); + } + Metric::HealCheck => { + self.record_scanner_source_work(ScannerWorkSource::Heal, count, 0, 0, 0, 0); + } + Metric::HealAbandonedObject => { + self.record_scanner_source_work(ScannerWorkSource::Heal, 0, 0, count, 0, 0); + } + _ => {} + } } pub fn record_scan_bucket_drive_start(&self) { @@ -1327,6 +1557,22 @@ impl Metrics { m.cycle_max_directories = self.scanner_cycle_max_directories.load(Ordering::Relaxed); m.bitrot_cycle_enabled = self.scanner_bitrot_cycle_enabled.load(Ordering::Relaxed); m.bitrot_cycle_seconds = self.scanner_bitrot_cycle_millis.load(Ordering::Relaxed) as f64 / 1000.0; + m.scan_checkpoint = match self.scanner_checkpoint.lock() { + Ok(checkpoint) => checkpoint.clone(), + Err(poisoned) => poisoned.into_inner().clone(), + }; + m.scan_checkpoint_used = self.scanner_checkpoint_used.load(Ordering::Relaxed); + m.scan_checkpoint_cleared = self.scanner_checkpoint_cleared.load(Ordering::Relaxed); + m.scan_checkpoint_ignored = self.scanner_checkpoint_ignored.load(Ordering::Relaxed); + m.scan_checkpoint_stale = self.scanner_checkpoint_stale.load(Ordering::Relaxed); + m.source_work = ScannerWorkSource::all() + .iter() + .filter_map(|source| { + self.scanner_source_work + .get(source.index()) + .map(|counters| counters.snapshot(*source)) + }) + .collect(); m.partial_cycles = self.partial_scan_cycles.load(Ordering::Relaxed); // Lifetime operation counts @@ -1539,6 +1785,94 @@ mod tests { assert_eq!(report.current_disk_bucket_scans_active, 0); } + #[tokio::test] + async fn report_includes_scanner_checkpoint_status() { + let metrics = Metrics::new(); + metrics.record_scanner_checkpoint_set(1, "bucket/child-a", "directories"); + metrics.record_scanner_checkpoint_used(); + + let report = metrics.report().await; + + let checkpoint = report.scan_checkpoint.as_ref().expect("scanner checkpoint should be visible"); + assert_eq!(checkpoint.version, 1); + assert_eq!(checkpoint.resume_after, "bucket/child-a"); + assert_eq!(checkpoint.reason, "directories"); + assert_eq!(checkpoint.last_event, "used"); + assert_eq!(report.scan_checkpoint_used, 1); + assert_eq!(report.scan_checkpoint_cleared, 0); + + metrics.record_scanner_checkpoint_stale(); + metrics.record_scanner_checkpoint_cleared(); + let report = metrics.report().await; + + assert!(report.scan_checkpoint.is_none()); + assert_eq!(report.scan_checkpoint_used, 1); + assert_eq!(report.scan_checkpoint_stale, 1); + assert_eq!(report.scan_checkpoint_cleared, 1); + } + + #[tokio::test] + async fn report_includes_scanner_source_work() { + let metrics = Metrics::new(); + metrics.record_scanner_source_work(ScannerWorkSource::Usage, 3, 0, 1, 0, 0); + metrics.record_scanner_source_work(ScannerWorkSource::Lifecycle, 0, 2, 1, 1, 0); + + let report = metrics.report().await; + + let usage = report + .source_work + .iter() + .find(|work| work.source == ScannerWorkSource::Usage.as_str()) + .expect("usage source work should be visible"); + assert_eq!(usage.checked, 3); + assert_eq!(usage.executed, 1); + + let lifecycle = report + .source_work + .iter() + .find(|work| work.source == ScannerWorkSource::Lifecycle.as_str()) + .expect("lifecycle source work should be visible"); + assert_eq!(lifecycle.queued, 2); + assert_eq!(lifecycle.executed, 1); + assert_eq!(lifecycle.failed, 1); + } + + #[tokio::test] + async fn scanner_metrics_update_source_work() { + let metrics = Metrics::new(); + metrics.record_source_work_for_metric(Metric::ScanObject, 3); + metrics.record_source_work_for_metric(Metric::ScanFolder, 2); + metrics.record_source_work_for_metric(Metric::SaveUsage, 1); + metrics.record_source_work_for_metric(Metric::CheckReplication, 4); + metrics.record_source_work_for_metric(Metric::HealCheck, 5); + metrics.record_source_work_for_metric(Metric::HealAbandonedObject, 6); + + let report = metrics.report().await; + + let usage = report + .source_work + .iter() + .find(|work| work.source == ScannerWorkSource::Usage.as_str()) + .expect("usage source work should be visible"); + assert_eq!(usage.checked, 5); + assert_eq!(usage.executed, 1); + + let bucket_replication = report + .source_work + .iter() + .find(|work| work.source == ScannerWorkSource::BucketReplication.as_str()) + .expect("bucket replication source work should be visible"); + assert_eq!(bucket_replication.checked, 4); + + let heal = report + .source_work + .iter() + .find(|work| work.source == ScannerWorkSource::Heal.as_str()) + .expect("heal source work should be visible"); + assert_eq!(heal.checked, 5); + assert_eq!(heal.executed, 6); + } + #[tokio::test] async fn report_preserves_current_cycle_started_time() { let previous_init_time = *crate::globals::GLOBAL_INIT_TIME.read().await; diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 43c3c8a60..1f8603694 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -220,6 +220,17 @@ pub enum DataUsageScanCheckpointReason { Unknown, } +impl DataUsageScanCheckpointReason { + pub fn as_str(self) -> &'static str { + match self { + Self::Runtime => "runtime", + Self::Objects => "objects", + Self::Directories => "directories", + Self::Unknown => "unknown", + } + } +} + #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] pub struct DataUsageScanCheckpoint { pub version: u16, diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 7c1440903..c7ecc0932 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -20,8 +20,8 @@ use std::time::{Duration, Instant, SystemTime}; use crate::ReplTargetSizeSummary; use crate::data_usage_define::{ - DataUsageCache, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageScanCheckpoint, DataUsageScanCheckpointReason, - SizeSummary, hash_path, + DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageScanCheckpoint, + DataUsageScanCheckpointReason, SizeSummary, hash_path, }; use crate::error::ScannerError; use crate::runtime_config::{ @@ -35,7 +35,9 @@ use rustfs_common::heal_channel::{ HEAL_DELETE_DANGLING, HealAdmissionResult, HealChannelPriority, HealChannelRequest, HealScanMode, send_heal_request_with_admission, }; -use rustfs_common::metrics::{IlmAction, Metric, Metrics, UpdateCurrentPathFn, current_path_updater, global_metrics}; +use rustfs_common::metrics::{ + IlmAction, Metric, Metrics, ScannerWorkSource, UpdateCurrentPathFn, current_path_updater, global_metrics, +}; use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc; use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::{GLOBAL_ExpiryState, apply_expiry_rule}; use rustfs_ecstore::bucket::lifecycle::evaluator::Evaluator; @@ -151,6 +153,13 @@ enum FolderResumeMatch { Descendant, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum FolderResumeOrder { + NoHint, + Used, + Stale, +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum FolderScanSource { New, @@ -173,14 +182,14 @@ fn folder_resume_match(folder_name: &str, resume_after: &str) -> Option(items: &mut [T], resume_after: Option<&str>, name: F) +fn order_items_for_resume(items: &mut [T], resume_after: Option<&str>, name: F) -> FolderResumeOrder where F: Fn(&T) -> &str, { items.sort_by(|left, right| name(left).cmp(name(right))); let Some(resume_after) = resume_after.filter(|resume_after| !resume_after.is_empty()) else { - return; + return FolderResumeOrder::NoHint; }; let Some((resume_index, resume_match)) = items @@ -188,7 +197,7 @@ where .enumerate() .find_map(|(index, item)| folder_resume_match(name(item), resume_after).map(|resume_match| (index, resume_match))) else { - return; + return FolderResumeOrder::Stale; }; let rotate_by = match resume_match { @@ -198,15 +207,16 @@ where if rotate_by < items.len() { items.rotate_left(rotate_by); } + FolderResumeOrder::Used } #[cfg(test)] -fn order_folders_for_resume(folders: &mut [CachedFolder], resume_after: Option<&str>) { - order_items_for_resume(folders, resume_after, |folder| folder.name.as_str()); +fn order_folders_for_resume(folders: &mut [CachedFolder], resume_after: Option<&str>) -> FolderResumeOrder { + order_items_for_resume(folders, resume_after, |folder| folder.name.as_str()) } -fn order_queued_folders_for_resume(folders: &mut [QueuedFolder], resume_after: Option<&str>) { - order_items_for_resume(folders, resume_after, |folder| folder.folder.name.as_str()); +fn order_queued_folders_for_resume(folders: &mut [QueuedFolder], resume_after: Option<&str>) -> FolderResumeOrder { + order_items_for_resume(folders, resume_after, |folder| folder.folder.name.as_str()) } fn checkpoint_reason_from_budget(reason: Option) -> DataUsageScanCheckpointReason { @@ -228,7 +238,13 @@ fn set_scan_checkpoint(cache: &mut DataUsageCache, reason: DataUsageScanCheckpoi }); if let Some(resume_after) = resume_after { - cache.info.scan_checkpoint = Some(DataUsageScanCheckpoint::new(resume_after, reason)); + let checkpoint = DataUsageScanCheckpoint::new(resume_after, reason); + global_metrics().record_scanner_checkpoint_set( + checkpoint.version, + checkpoint.resume_after.clone(), + checkpoint.reason.as_str(), + ); + cache.info.scan_checkpoint = Some(checkpoint); } else { cache.info.scan_checkpoint = None; } @@ -740,6 +756,7 @@ impl ScannerItem { ensure_scanner_alert_metrics_registered(); let (too_many_versions, too_large_versions) = should_alert_excessive_versions(remaining_versions, cumulative_size); if too_many_versions { + global_metrics().record_scanner_source_work(ScannerWorkSource::Alerts, 0, 0, 1, 0, 0); counter!( METRIC_SCANNER_EXCESS_OBJECT_VERSIONS_TOTAL, "bucket" => self.bucket.clone() @@ -754,6 +771,7 @@ impl ScannerItem { ); } if too_large_versions { + global_metrics().record_scanner_source_work(ScannerWorkSource::Alerts, 0, 0, 1, 0, 0); counter!( METRIC_SCANNER_EXCESS_OBJECT_VERSION_SIZE_TOTAL, "bucket" => self.bucket.clone() @@ -877,17 +895,23 @@ impl FolderScanner { self.new_cache.info.scan_resume_after = Some(folder.to_string()); self.update_cache.info.scan_resume_after = Some(folder.to_string()); let checkpoint = DataUsageScanCheckpoint::new(folder.to_string(), DataUsageScanCheckpointReason::Unknown); + global_metrics().record_scanner_checkpoint_set( + checkpoint.version, + checkpoint.resume_after.clone(), + checkpoint.reason.as_str(), + ); self.new_cache.info.scan_checkpoint = Some(checkpoint.clone()); self.update_cache.info.scan_checkpoint = Some(checkpoint); } fn alert_excessive_folders(&self, folder: &str, total_folders: usize) { let threshold = scanner_excess_folders_threshold(); - if total_folders as u64 <= threshold { + if u64::try_from(total_folders).unwrap_or(u64::MAX) <= threshold { return; } ensure_scanner_alert_metrics_registered(); + global_metrics().record_scanner_source_work(ScannerWorkSource::Alerts, 0, 0, 1, 0, 0); counter!( METRIC_SCANNER_EXCESS_FOLDERS_TOTAL, "root" => self.root.clone() @@ -1255,13 +1279,29 @@ impl FolderScanner { } } - let scan_resume_after = self - .old_cache - .info - .scan_checkpoint - .as_ref() - .map(|checkpoint| checkpoint.resume_after.as_str()) - .or(self.old_cache.info.scan_resume_after.as_deref()); + let is_scan_root = folder.name == self.old_cache.info.name; + let scan_checkpoint = self.old_cache.info.scan_checkpoint.as_ref(); + let checkpoint_resume_after = scan_checkpoint.and_then(|checkpoint| { + if is_scan_root { + global_metrics().record_scanner_checkpoint_set( + checkpoint.version, + checkpoint.resume_after.clone(), + checkpoint.reason.as_str(), + ); + } + if checkpoint.version != DATA_USAGE_SCAN_CHECKPOINT_VERSION || checkpoint.resume_after.is_empty() { + if is_scan_root { + global_metrics().record_scanner_checkpoint_ignored(); + } + None + } else { + Some(checkpoint.resume_after.as_str()) + } + }); + let checkpoint_tracks_child_order = checkpoint_resume_after + .and_then(|resume_after| folder_resume_match(&folder.name, resume_after)) + .is_some_and(|resume_match| matches!(resume_match, FolderResumeMatch::Descendant)); + let scan_resume_after = checkpoint_resume_after.or(self.old_cache.info.scan_resume_after.as_deref()); let mut queued_folders = Vec::with_capacity(new_folders.len() + existing_folders.len()); queued_folders.extend(new_folders.into_iter().map(|folder| QueuedFolder { folder, @@ -1271,7 +1311,15 @@ impl FolderScanner { folder, source: FolderScanSource::Existing, })); - order_queued_folders_for_resume(&mut queued_folders, scan_resume_after); + let has_queued_folders = !queued_folders.is_empty(); + let resume_order = order_queued_folders_for_resume(&mut queued_folders, scan_resume_after); + if checkpoint_tracks_child_order && has_queued_folders { + match resume_order { + FolderResumeOrder::Used => global_metrics().record_scanner_checkpoint_used(), + FolderResumeOrder::Stale => global_metrics().record_scanner_checkpoint_stale(), + FolderResumeOrder::NoHint => {} + } + } // Scan child folders in the combined resume order. for queued_folder in queued_folders { @@ -1767,8 +1815,12 @@ pub async fn scan_data_folder( new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN); new_cache.info.last_update = Some(SystemTime::now()); new_cache.info.next_cycle = cache.info.next_cycle; + let had_scan_checkpoint = cache.info.scan_checkpoint.is_some() || new_cache.info.scan_checkpoint.is_some(); new_cache.info.scan_resume_after = None; new_cache.info.scan_checkpoint = None; + if had_scan_checkpoint { + global_metrics().record_scanner_checkpoint_cleared(); + } close_disk().await; Ok(new_cache.clone()) @@ -2037,9 +2089,10 @@ mod tests { }, ]; - order_folders_for_resume(&mut folders, Some("bucket/child-b")); + let outcome = order_folders_for_resume(&mut folders, Some("bucket/child-b")); let names = folders.into_iter().map(|folder| folder.name).collect::>(); + assert_eq!(outcome, FolderResumeOrder::Used); assert_eq!( names, vec![ @@ -2070,9 +2123,10 @@ mod tests { }, ]; - order_folders_for_resume(&mut folders, Some("bucket/child-b/grandchild")); + let outcome = order_folders_for_resume(&mut folders, Some("bucket/child-b/grandchild")); let names = folders.into_iter().map(|folder| folder.name).collect::>(); + assert_eq!(outcome, FolderResumeOrder::Used); assert_eq!( names, vec![ @@ -2083,6 +2137,28 @@ mod tests { ); } + #[test] + fn test_order_folders_for_resume_reports_stale_hint() { + let mut folders = vec![ + CachedFolder { + name: "bucket/child-c".to_string(), + parent: None, + object_heal_prob_div: 1, + }, + CachedFolder { + name: "bucket/child-a".to_string(), + parent: None, + object_heal_prob_div: 1, + }, + ]; + + let outcome = order_folders_for_resume(&mut folders, Some("bucket/child-b")); + + let names = folders.into_iter().map(|folder| folder.name).collect::>(); + assert_eq!(outcome, FolderResumeOrder::Stale); + assert_eq!(names, vec!["bucket/child-a".to_string(), "bucket/child-c".to_string()]); + } + #[tokio::test] #[serial] async fn test_record_failed_prunes_to_max_entries() { @@ -2466,6 +2542,51 @@ mod tests { assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Directories)); } + #[tokio::test] + #[serial] + async fn test_scan_data_folder_reports_invalid_checkpoint_ignored_once() { + let (scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard { + temp_dir: Some(temp_dir.clone()), + }; + + tokio::fs::create_dir_all(temp_dir.join("bucket").join("child-a").join("grandchild")) + .await + .expect("failed to create nested child directory"); + + let before = global_metrics().report().await.scan_checkpoint_ignored; + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new(&parent, Default::default()); + let cache = DataUsageCache { + info: crate::data_usage_define::DataUsageCacheInfo { + name: "bucket".to_string(), + scan_checkpoint: Some(crate::data_usage_define::DataUsageScanCheckpoint { + version: crate::data_usage_define::DATA_USAGE_SCAN_CHECKPOINT_VERSION + 1, + resume_after: "bucket/child-a".to_string(), + reason: crate::data_usage_define::DataUsageScanCheckpointReason::Unknown, + }), + ..Default::default() + }, + ..Default::default() + }; + + let result = scan_data_folder( + budget.token(), + budget.clone(), + vec![scanner.local_disk.clone()], + scanner.local_disk.clone(), + cache, + None, + HealScanMode::Normal, + SCANNER_SLEEPER.clone(), + ) + .await; + + assert!(result.is_ok(), "scan should complete with an ignored checkpoint"); + let after = global_metrics().report().await.scan_checkpoint_ignored; + assert_eq!(after.saturating_sub(before), 1); + } + #[tokio::test] #[serial] async fn test_scan_data_folder_resume_hint_prioritizes_next_existing_folder() {