diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index 4a2afbbe7..24f35678b 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -1114,6 +1114,8 @@ pub struct ScannerLastMinute { pub struct ScannerMetricsReport { pub collected_at: DateTime, pub current_cycle: u64, + #[serde(default)] + pub current_cycle_active: bool, pub current_started: DateTime, pub cycles_completed_at: Vec>, pub ongoing_buckets: usize, @@ -2051,7 +2053,7 @@ impl Metrics { pub fn record_scanner_transition_failed(&self, count: u64) { self.scanner_transition_failed.fetch_add(count, Ordering::Relaxed); self.record_scanner_source_failed(ScannerWorkSource::Lifecycle, count); - if !self.current_scan_cycle_work_active.load(Ordering::Relaxed) { + if !self.current_scan_cycle_work_active.load(Ordering::Acquire) { self.record_last_cycle_scanner_source_work(ScannerWorkSource::Lifecycle, ScannerSourceWorkUpdate::failed(count)); } } @@ -2336,6 +2338,21 @@ impl Metrics { *self.cycle_info.write().await = cycle; } + /// Publish a scanner cycle and its work-accounting baseline as one state transition. + pub async fn start_scan_cycle_work_with_cycle(&self, cycle: CurrentCycle) -> ScanCycleWorkSnapshot { + let mut current_cycle = self.cycle_info.write().await; + let snapshot = self.start_scan_cycle_work(); + *current_cycle = Some(cycle); + snapshot + } + + /// Publish the completed work snapshot and idle cycle state as one state transition. + pub async fn finish_scan_cycle_work_with_cycle(&self, start: ScanCycleWorkSnapshot, cycle: CurrentCycle) { + let mut current_cycle = self.cycle_info.write().await; + self.finish_scan_cycle_work(start); + *current_cycle = Some(cycle); + } + /// Read the current cycle record. pub async fn get_cycle(&self) -> Option { self.cycle_info.read().await.clone() @@ -2464,7 +2481,7 @@ impl Metrics { &self.current_scan_cycle_replication_repair_work_start, &replication_repair_snapshot, ); - self.current_scan_cycle_work_active.store(true, Ordering::Relaxed); + self.current_scan_cycle_work_active.store(true, Ordering::Release); snapshot } @@ -2476,11 +2493,11 @@ impl Metrics { self.record_scan_cycle_work(work); self.record_scan_cycle_source_work(&source_work); self.record_scan_cycle_replication_repair_work(&replication_repair_work); - self.current_scan_cycle_work_active.store(false, Ordering::Relaxed); + self.current_scan_cycle_work_active.store(false, Ordering::Release); } pub fn current_scan_cycle_has_unresolved_heal_work(&self) -> bool { - if !self.current_scan_cycle_work_active.load(Ordering::Relaxed) { + if !self.current_scan_cycle_work_active.load(Ordering::Acquire) { return false; } @@ -2746,13 +2763,41 @@ impl Metrics { pub async fn report(&self) -> ScannerMetricsReport { let mut m = ScannerMetricsReport::default(); - let has_cycle = if let Some(cycle) = self.get_cycle().await { - m.current_cycle = cycle.current; - m.cycles_completed_at = cycle.cycle_completed; - m.current_started = cycle.started; - true - } else { - false + let has_cycle = { + let cycle = self.cycle_info.read().await; + let has_cycle = if let Some(cycle) = cycle.as_ref() { + m.current_cycle = cycle.current; + m.cycles_completed_at = cycle.cycle_completed.clone(); + m.current_started = cycle.started; + true + } else { + false + }; + m.current_cycle_active = self.current_scan_cycle_work_active.load(Ordering::Acquire); + if m.current_cycle_active { + let current_work = self.scan_cycle_work_since(self.current_scan_cycle_work_start()); + let current_source_work = self.scanner_source_work_since(&self.current_scan_cycle_source_work_start_values()); + let current_replication_repair_work = + self.scanner_replication_repair_work_since(&self.current_scan_cycle_replication_repair_work_start_values()); + m.current_cycle_objects_scanned = current_work.objects_scanned; + m.current_cycle_directories_scanned = current_work.directories_scanned; + m.current_cycle_bucket_drive_scans = current_work.bucket_drive_scans; + m.current_cycle_bucket_drive_failures = current_work.bucket_drive_failures; + m.current_cycle_yield_events = current_work.yield_events; + m.current_cycle_yield_duration_seconds = current_work.yield_duration_millis as f64 / 1000.0; + m.current_cycle_throttle_sleep_events = current_work.throttle_sleep_events; + m.current_cycle_throttle_sleep_duration_seconds = current_work.throttle_sleep_duration_millis as f64 / 1000.0; + m.current_cycle_ilm_actions = current_work.ilm_actions; + m.current_cycle_lifecycle_expiry_actions = current_work.lifecycle_expiry_actions; + m.current_cycle_lifecycle_transition_actions = current_work.lifecycle_transition_actions; + m.current_cycle_heal_objects = current_work.heal_objects; + m.current_cycle_replication_checks = current_work.replication_checks; + m.current_cycle_usage_saves = current_work.usage_saves; + m.current_cycle_source_work = self.scanner_source_work_snapshots(¤t_source_work); + m.current_cycle_replication_repair = + self.scanner_replication_repair_work_snapshots(¤t_replication_repair_work); + } + has_cycle }; if !has_cycle && let Some(init_time) = crate::get_global_init_time().await { @@ -2793,28 +2838,6 @@ impl Metrics { m.current_disk_scan_concurrency_limit = disk_scan_concurrency_limit; m.current_disk_bucket_scans_queued = disk_bucket_scans_queued; m.current_disk_bucket_scans_active = disk_bucket_scans_active; - if self.current_scan_cycle_work_active.load(Ordering::Relaxed) { - let current_work = self.scan_cycle_work_since(self.current_scan_cycle_work_start()); - let current_source_work = self.scanner_source_work_since(&self.current_scan_cycle_source_work_start_values()); - let current_replication_repair_work = - self.scanner_replication_repair_work_since(&self.current_scan_cycle_replication_repair_work_start_values()); - m.current_cycle_objects_scanned = current_work.objects_scanned; - m.current_cycle_directories_scanned = current_work.directories_scanned; - m.current_cycle_bucket_drive_scans = current_work.bucket_drive_scans; - m.current_cycle_bucket_drive_failures = current_work.bucket_drive_failures; - m.current_cycle_yield_events = current_work.yield_events; - m.current_cycle_yield_duration_seconds = current_work.yield_duration_millis as f64 / 1000.0; - m.current_cycle_throttle_sleep_events = current_work.throttle_sleep_events; - m.current_cycle_throttle_sleep_duration_seconds = current_work.throttle_sleep_duration_millis as f64 / 1000.0; - m.current_cycle_ilm_actions = current_work.ilm_actions; - m.current_cycle_lifecycle_expiry_actions = current_work.lifecycle_expiry_actions; - m.current_cycle_lifecycle_transition_actions = current_work.lifecycle_transition_actions; - m.current_cycle_heal_objects = current_work.heal_objects; - m.current_cycle_replication_checks = current_work.replication_checks; - m.current_cycle_usage_saves = current_work.usage_saves; - m.current_cycle_source_work = self.scanner_source_work_snapshots(¤t_source_work); - m.current_cycle_replication_repair = self.scanner_replication_repair_work_snapshots(¤t_replication_repair_work); - } let last_cycle_result = self.last_scan_cycle_result.load(Ordering::Relaxed); m.last_cycle_result = scan_cycle_result_label(last_cycle_result).to_string(); m.last_cycle_result_code = last_cycle_result as u64; @@ -4142,6 +4165,8 @@ mod tests { let report = metrics.report().await; + assert!(report.current_cycle_active); + assert_eq!(report.current_cycle, 0); assert_eq!(report.current_cycle_objects_scanned, 7); assert_eq!(report.current_cycle_directories_scanned, 3); assert_eq!(report.current_cycle_bucket_drive_scans, 2); @@ -4158,6 +4183,8 @@ mod tests { metrics.finish_scan_cycle_work(start); let report = metrics.report().await; + assert!(!report.current_cycle_active); + assert_eq!(report.current_cycle, 0); assert_eq!(report.current_cycle_objects_scanned, 0); assert_eq!(report.current_cycle_directories_scanned, 0); assert_eq!(report.current_cycle_bucket_drive_scans, 0); @@ -4184,6 +4211,91 @@ mod tests { assert_eq!(report.last_cycle_usage_saves, 2); } + #[tokio::test] + async fn scan_cycle_activity_and_cycle_state_publish_together() { + let metrics = Metrics::new(); + let cycle_started = Utc::now() - chrono::Duration::seconds(5); + let active_cycle = CurrentCycle { + current: 12, + next: 13, + started: cycle_started, + ..Default::default() + }; + + let cycle_state = metrics.cycle_info.read().await; + let mut start_transition = Box::pin(metrics.start_scan_cycle_work_with_cycle(active_cycle)); + let waker = std::task::Waker::noop(); + let mut context = std::task::Context::from_waker(waker); + assert!(start_transition.as_mut().poll(&mut context).is_pending()); + assert!(!metrics.current_scan_cycle_work_active.load(Ordering::Acquire)); + drop(cycle_state); + + let start = start_transition.await; + let active = metrics.report().await; + assert!(active.current_cycle_active); + assert_eq!(active.current_cycle, 12); + assert_eq!(active.current_started, cycle_started); + + let idle_cycle = CurrentCycle { + current: 0, + next: 13, + started: cycle_started, + ..Default::default() + }; + let cycle_state = metrics.cycle_info.read().await; + let mut finish_transition = Box::pin(metrics.finish_scan_cycle_work_with_cycle(start, idle_cycle)); + assert!(finish_transition.as_mut().poll(&mut context).is_pending()); + assert!(metrics.current_scan_cycle_work_active.load(Ordering::Acquire)); + drop(cycle_state); + + finish_transition.await; + let idle = metrics.report().await; + assert!(!idle.current_cycle_active); + assert_eq!(idle.current_cycle, 0); + } + + #[tokio::test] + async fn report_keeps_cycle_identity_and_work_in_one_snapshot() { + let metrics = Metrics::new(); + let cycle_ten = CurrentCycle { + current: 10, + next: 11, + started: Utc::now() - chrono::Duration::seconds(10), + ..Default::default() + }; + let cycle_ten_start = metrics.start_scan_cycle_work_with_cycle(cycle_ten.clone()).await; + metrics.operations[Metric::ScanObject as usize].store(1, Ordering::Relaxed); + + let paths = metrics.current_paths.write().await; + let mut report = Box::pin(metrics.report()); + let waker = std::task::Waker::noop(); + let mut context = std::task::Context::from_waker(waker); + assert!(report.as_mut().poll(&mut context).is_pending()); + + metrics + .finish_scan_cycle_work_with_cycle(cycle_ten_start, CurrentCycle { current: 0, ..cycle_ten }) + .await; + let cycle_eleven_start = metrics + .start_scan_cycle_work_with_cycle(CurrentCycle { + current: 11, + next: 12, + started: Utc::now(), + ..Default::default() + }) + .await; + metrics.operations[Metric::ScanObject as usize].store(101, Ordering::Relaxed); + + drop(paths); + let snapshot = report.await; + + assert_eq!(snapshot.current_cycle, 10); + assert_eq!(snapshot.current_cycle_objects_scanned, 1); + + metrics + .finish_scan_cycle_work_with_cycle(cycle_eleven_start, CurrentCycle::default()) + .await; + } + #[tokio::test] async fn scanner_cycle_ilm_actions_ignore_global_ilm_work() { let metrics = Metrics::new(); diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index 78ac568fb..c3f4f4f68 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -207,10 +207,6 @@ pub(crate) async fn ensure_boot_time() { GLOBAL_BOOT_TIME.get_or_init(|| async { SystemTime::now() }).await; } -pub(crate) async fn scanner_init_time() -> Option> { - rustfs_common::get_global_init_time().await -} - pub(crate) async fn root_disk_threshold_for_erasure_disk() -> Option { if is_erasure_sd().await { None diff --git a/crates/ecstore/src/services/metrics_realtime.rs b/crates/ecstore/src/services/metrics_realtime.rs index f85169483..f99e018c3 100644 --- a/crates/ecstore/src/services/metrics_realtime.rs +++ b/crates/ecstore/src/services/metrics_realtime.rs @@ -71,6 +71,7 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo MadminScannerMetrics { collected_at: metrics.collected_at, current_cycle: metrics.current_cycle, + current_cycle_active: Some(metrics.current_cycle_active), current_started: metrics.current_started, cycles_completed_at: metrics.cycles_completed_at, ongoing_buckets: metrics.ongoing_buckets, @@ -398,10 +399,7 @@ pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts) if types.contains(&MetricType::SCANNER) { debug!("start get scanner metrics"); - let mut metrics = global_metrics().report().await; - if let Some(init_time) = runtime_sources::scanner_init_time().await { - metrics.current_started = init_time; - } + let metrics = global_metrics().report().await; real_time_metrics.aggregated.scanner = Some(to_madmin_scanner_metrics(metrics)); } @@ -540,7 +538,9 @@ async fn collect_local_disks_metrics(disks: &HashSet) -> HashMap, #[serde(rename = "current_cycle")] pub current_cycle: u64, + #[serde(rename = "current_cycle_active", default, skip_serializing_if = "Option::is_none")] + pub current_cycle_active: Option, #[serde(rename = "current_started")] pub current_started: DateTime, #[serde(rename = "cycle_complete_times")] @@ -718,7 +720,31 @@ pub struct ScannerMetrics { } impl ScannerMetrics { + pub fn is_current_cycle_active(&self) -> bool { + self.current_cycle_active.unwrap_or(self.current_cycle > 0) + } + pub fn merge(&mut self, other: &Self) { + // Legacy nodes omit the activity field and use a non-zero cycle as + // their active signal. New nodes publish explicit first-cycle and idle + // states, including cycle zero. + let self_cycle_active = self.is_current_cycle_active(); + let other_cycle_active = other.is_current_cycle_active(); + let self_cycle_authority = ( + self_cycle_active, + self.current_cycle, + self.cycles_completed_at.len(), + self.cycles_completed_at.as_slice(), + self.current_started, + ); + let other_cycle_authority = ( + other_cycle_active, + other.current_cycle, + other.cycles_completed_at.len(), + other.cycles_completed_at.as_slice(), + other.current_started, + ); + let other_cycle_is_authoritative = self_cycle_authority < other_cycle_authority; let other_is_newer = self.collected_at < other.collected_at; if other_is_newer { self.collected_at = other.collected_at; @@ -854,15 +880,12 @@ impl ScannerMetrics { self.ongoing_buckets = other.ongoing_buckets; } - if self.current_cycle < other.current_cycle { + if other_cycle_is_authoritative { self.current_cycle = other.current_cycle; self.cycles_completed_at = other.cycles_completed_at.clone(); self.current_started = other.current_started; } - - if other.cycles_completed_at.len() > self.cycles_completed_at.len() { - self.cycles_completed_at = other.cycles_completed_at.clone(); - } + self.current_cycle_active = Some(self_cycle_active || other_cycle_active); if !other.life_time_ops.is_empty() && self.life_time_ops.is_empty() { self.life_time_ops = other.life_time_ops.clone(); @@ -931,7 +954,13 @@ impl Metrics { if let Some(scanner) = other.scanner.as_ref() { match self.scanner { Some(ref mut s_scanner) => s_scanner.merge(scanner), - None => self.scanner = Some(scanner.clone()), + None => { + let mut scanner = scanner.clone(); + if scanner.current_cycle_active.is_none() { + scanner.current_cycle_active = Some(scanner.is_current_cycle_active()); + } + self.scanner = Some(scanner); + } } } @@ -1389,6 +1418,280 @@ pub struct Operations { mod tests { use super::*; + #[test] + fn scanner_metrics_serializes_cycle_active_presence() { + let missing_value = serde_json::to_value(ScannerMetrics::default()).expect("scanner metrics should serialize"); + assert!(missing_value.get("current_cycle_active").is_none()); + let missing: ScannerMetrics = + serde_json::from_value(missing_value).expect("older scanner metrics without cycle-active should decode"); + assert_eq!(missing.current_cycle_active, None); + + let explicit_false_value = serde_json::to_value(ScannerMetrics { + current_cycle_active: Some(false), + ..Default::default() + }) + .expect("scanner metrics with explicit cycle-active should serialize"); + assert_eq!(explicit_false_value["current_cycle_active"], serde_json::Value::Bool(false)); + let explicit_false: ScannerMetrics = + serde_json::from_value(explicit_false_value).expect("explicit cycle-active should decode"); + assert_eq!(explicit_false.current_cycle_active, Some(false)); + } + + #[test] + fn scanner_metrics_merge_prefers_an_active_first_cycle() { + let collected_at = Utc::now(); + let idle_started = collected_at - chrono::Duration::hours(1); + let active_started = collected_at - chrono::Duration::seconds(5); + let mut scanner = ScannerMetrics { + collected_at, + current_cycle: 0, + current_cycle_active: Some(false), + current_started: idle_started, + ..Default::default() + }; + + scanner.merge(&ScannerMetrics { + collected_at: collected_at + chrono::Duration::seconds(1), + current_cycle: 0, + current_cycle_active: Some(true), + current_started: active_started, + ..Default::default() + }); + + assert_eq!(scanner.current_cycle_active, Some(true)); + assert_eq!(scanner.current_cycle, 0); + assert_eq!(scanner.current_started, active_started); + } + + #[test] + fn metrics_merge_preserves_explicit_active_first_cycle() { + let mut aggregated = Metrics::default(); + aggregated.merge(&Metrics { + scanner: Some(ScannerMetrics { + current_cycle_active: Some(true), + ..Default::default() + }), + ..Default::default() + }); + + let scanner = aggregated.scanner.expect("aggregated scanner metrics"); + assert_eq!(scanner.current_cycle_active, Some(true)); + assert_eq!(scanner.current_cycle, 0); + } + + #[test] + fn scanner_metrics_merge_preserves_legacy_nonzero_active_signal() { + let collected_at = Utc::now(); + let mut scanner = ScannerMetrics { + collected_at, + ..Default::default() + }; + + scanner.merge(&ScannerMetrics { + collected_at: collected_at + chrono::Duration::seconds(1), + current_cycle: 7, + ..Default::default() + }); + + assert_eq!(scanner.current_cycle_active, Some(true)); + assert_eq!(scanner.current_cycle, 7); + } + + #[test] + fn metrics_merge_normalizes_first_legacy_scanner_snapshot() { + let legacy = Metrics { + scanner: Some(ScannerMetrics { + current_cycle: 7, + ..Default::default() + }), + ..Default::default() + }; + let mut aggregated = Metrics::default(); + + aggregated.merge(&legacy); + + let scanner = aggregated.scanner.expect("aggregated scanner metrics"); + assert_eq!(scanner.current_cycle_active, Some(true)); + assert_eq!(scanner.current_cycle, 7); + } + + #[test] + fn scanner_metrics_merge_preserves_explicit_inactive_nonzero_cycle() { + let collected_at = Utc::now(); + let mut scanner = ScannerMetrics::default(); + + scanner.merge(&ScannerMetrics { + collected_at, + current_cycle: 7, + current_cycle_active: Some(false), + ..Default::default() + }); + + assert_eq!(scanner.current_cycle_active, Some(false)); + assert_eq!(scanner.current_cycle, 7); + } + + #[test] + fn scanner_metrics_merge_cycle_active_is_order_independent() { + let collected_at = Utc::now(); + let legacy_active = ScannerMetrics { + collected_at, + current_cycle: 7, + current_started: collected_at - chrono::Duration::seconds(10), + ..Default::default() + }; + let explicit_idle = ScannerMetrics { + collected_at, + current_cycle: 0, + current_cycle_active: Some(false), + current_started: collected_at - chrono::Duration::hours(1), + ..Default::default() + }; + + let mut legacy_first = legacy_active.clone(); + legacy_first.merge(&explicit_idle); + let mut legacy_second = explicit_idle.clone(); + legacy_second.merge(&legacy_active); + assert_eq!(legacy_first.current_cycle_active, Some(true)); + assert_eq!(legacy_second.current_cycle_active, Some(true)); + assert_eq!(legacy_first.current_cycle, 7); + assert_eq!(legacy_second.current_cycle, 7); + + let explicit_inactive_nonzero = ScannerMetrics { + collected_at, + current_cycle: 7, + current_cycle_active: Some(false), + ..Default::default() + }; + let mut inactive_first = explicit_inactive_nonzero.clone(); + inactive_first.merge(&explicit_idle); + let mut inactive_second = explicit_idle; + inactive_second.merge(&explicit_inactive_nonzero); + assert_eq!(inactive_first.current_cycle_active, Some(false)); + assert_eq!(inactive_second.current_cycle_active, Some(false)); + assert_eq!(inactive_first.current_cycle, 7); + assert_eq!(inactive_second.current_cycle, 7); + } + + #[test] + fn scanner_metrics_merge_cycle_authority_is_order_independent() { + let collected_at = Utc::now(); + let completion = collected_at - chrono::Duration::minutes(1); + let earlier_active = ScannerMetrics { + collected_at, + current_cycle: 7, + current_cycle_active: Some(true), + current_started: collected_at - chrono::Duration::seconds(10), + cycles_completed_at: vec![completion], + ..Default::default() + }; + let later_active = ScannerMetrics { + collected_at: collected_at + chrono::Duration::seconds(1), + current_cycle: 7, + current_cycle_active: Some(true), + current_started: collected_at - chrono::Duration::seconds(5), + cycles_completed_at: vec![completion], + ..Default::default() + }; + + let mut earlier_first = earlier_active.clone(); + earlier_first.merge(&later_active); + let mut later_first = later_active.clone(); + later_first.merge(&earlier_active); + + assert_eq!(earlier_first.current_started, later_active.current_started); + assert_eq!(later_first.current_started, later_active.current_started); + assert_eq!(earlier_first.cycles_completed_at, later_active.cycles_completed_at); + assert_eq!(later_first.cycles_completed_at, later_active.cycles_completed_at); + + let stale_idle = ScannerMetrics { + collected_at, + current_cycle_active: Some(false), + current_started: collected_at - chrono::Duration::hours(1), + ..Default::default() + }; + let completed_idle = ScannerMetrics { + collected_at: collected_at + chrono::Duration::seconds(1), + current_cycle_active: Some(false), + current_started: collected_at - chrono::Duration::seconds(5), + cycles_completed_at: vec![completion], + ..Default::default() + }; + + let mut stale_first = stale_idle.clone(); + stale_first.merge(&completed_idle); + let mut completed_first = completed_idle.clone(); + completed_first.merge(&stale_idle); + + assert_eq!(stale_first.current_started, completed_idle.current_started); + assert_eq!(completed_first.current_started, completed_idle.current_started); + assert_eq!(stale_first.cycles_completed_at, completed_idle.cycles_completed_at); + assert_eq!(completed_first.cycles_completed_at, completed_idle.cycles_completed_at); + } + + #[test] + fn scanner_metrics_merge_cycle_authority_is_associative() { + let collected_at = Utc::now(); + let older_completion = collected_at - chrono::Duration::minutes(3); + let last_completion = collected_at - chrono::Duration::minutes(1); + let cycle_seven = ScannerMetrics { + collected_at, + current_cycle: 7, + current_cycle_active: Some(true), + current_started: collected_at - chrono::Duration::seconds(10), + cycles_completed_at: vec![older_completion, last_completion], + ..Default::default() + }; + let cycle_eight = ScannerMetrics { + collected_at: collected_at + chrono::Duration::seconds(1), + current_cycle: 8, + current_cycle_active: Some(true), + current_started: collected_at - chrono::Duration::seconds(5), + cycles_completed_at: vec![last_completion], + ..Default::default() + }; + let newer_idle = ScannerMetrics { + collected_at: collected_at + chrono::Duration::hours(1), + current_cycle_active: Some(false), + current_started: collected_at - chrono::Duration::hours(1), + ..Default::default() + }; + + let mut left_associative = cycle_seven.clone(); + left_associative.merge(&cycle_eight); + left_associative.merge(&newer_idle); + + let mut right_group = cycle_eight.clone(); + right_group.merge(&newer_idle); + let mut right_associative = cycle_seven.clone(); + right_associative.merge(&right_group); + + assert_eq!(left_associative.current_cycle, 8); + assert_eq!(right_associative.current_cycle, 8); + assert_eq!(left_associative.current_started, cycle_eight.current_started); + assert_eq!(right_associative.current_started, cycle_eight.current_started); + assert_eq!(left_associative.cycles_completed_at, cycle_eight.cycles_completed_at); + assert_eq!(right_associative.cycles_completed_at, cycle_eight.cycles_completed_at); + + for order in [ + [&cycle_seven, &cycle_eight, &newer_idle], + [&cycle_seven, &newer_idle, &cycle_eight], + [&cycle_eight, &cycle_seven, &newer_idle], + [&cycle_eight, &newer_idle, &cycle_seven], + [&newer_idle, &cycle_seven, &cycle_eight], + [&newer_idle, &cycle_eight, &cycle_seven], + ] { + let mut merged = ScannerMetrics::default(); + for scanner in order { + merged.merge(scanner); + } + assert_eq!(merged.current_cycle_active, Some(true)); + assert_eq!(merged.current_cycle, 8); + assert_eq!(merged.current_started, cycle_eight.current_started); + assert_eq!(merged.cycles_completed_at, cycle_eight.cycles_completed_at); + } + } + #[test] fn scanner_metrics_merge_aggregates_partial_cycles_by_source() { let collected_at = Utc::now(); diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 0770b877b..743722921 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -262,11 +262,11 @@ async fn obs_site_replication_stats() -> ReplicationStats { } fn current_scanner_cycle_age_seconds( - current_cycle: u64, + current_cycle_active: bool, current_started: chrono::DateTime, now: chrono::DateTime, ) -> u64 { - if current_cycle == 0 { + if !current_cycle_active { 0 } else { now.signed_duration_since(current_started).num_seconds().max(0) as u64 @@ -1090,7 +1090,7 @@ pub async fn collect_scanner_metric_stats() -> Option { let reference_time = metrics.cycles_completed_at.last().copied().unwrap_or(metrics.current_started); let last_activity_seconds = now.signed_duration_since(reference_time).num_seconds().max(0) as u64; let active_paths = metrics.active_scan_paths as u64; - let current_cycle_age_seconds = current_scanner_cycle_age_seconds(metrics.current_cycle, metrics.current_started, now); + let current_cycle_age_seconds = current_scanner_cycle_age_seconds(metrics.current_cycle_active, metrics.current_started, now); let current_scan_mode = scanner_scan_mode_code(&metrics.current_scan_mode); let current_cycle_age = current_cycle_age_seconds as f64; let last_cycle_duration = metrics.last_cycle_duration_seconds; @@ -1464,21 +1464,21 @@ mod tests { fn current_scanner_cycle_age_seconds_returns_zero_when_idle() { let now = Utc::now(); - assert_eq!(current_scanner_cycle_age_seconds(0, now - chrono::Duration::seconds(30), now), 0); + assert_eq!(current_scanner_cycle_age_seconds(false, now - chrono::Duration::seconds(30), now), 0); } #[test] fn current_scanner_cycle_age_seconds_clamps_future_start() { let now = Utc::now(); - assert_eq!(current_scanner_cycle_age_seconds(4, now + chrono::Duration::seconds(30), now), 0); + assert_eq!(current_scanner_cycle_age_seconds(true, now + chrono::Duration::seconds(30), now), 0); } #[test] - fn current_scanner_cycle_age_seconds_reports_active_elapsed_time() { + fn current_scanner_cycle_age_seconds_reports_active_first_cycle_elapsed_time() { let now = Utc::now(); - assert_eq!(current_scanner_cycle_age_seconds(4, now - chrono::Duration::seconds(45), now), 45); + assert_eq!(current_scanner_cycle_age_seconds(true, now - chrono::Duration::seconds(45), now), 45); } #[test] diff --git a/crates/scanner/Cargo.toml b/crates/scanner/Cargo.toml index 490a1d9c6..33dde7715 100644 --- a/crates/scanner/Cargo.toml +++ b/crates/scanner/Cargo.toml @@ -47,7 +47,7 @@ rmp-serde = { workspace = true } hmac = { workspace = true } sha2 = { workspace = true } rustfs-filemeta = { workspace = true } -tokio-util = { workspace = true, features = ["io", "compat"] } +tokio-util = { workspace = true, features = ["io", "compat", "rt"] } rustfs-ecstore = { workspace = true } rustfs-storage-api = { workspace = true } http = { workspace = true } diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index c1c61a3e9..91f9e1597 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -14,6 +14,8 @@ use std::collections::BTreeMap; use std::future::Future; +#[cfg(test)] +use std::sync::Mutex as StdMutex; use std::sync::{Arc, LazyLock, RwLock}; use crate::ScannerObjectIO; @@ -38,8 +40,8 @@ use bytes::Bytes; use chrono::{DateTime, Utc}; use rustfs_common::heal_channel::HealScanMode; use rustfs_common::metrics::{ - CurrentCycle, Metric, Metrics, ScanCyclePartialReason, ScannerUsageSaveResult, ScannerWorkSource, emit_scan_cycle_complete, - emit_scan_cycle_partial_with_source, emit_scan_cycle_superseded, global_metrics, + CurrentCycle, Metric, Metrics, ScanCyclePartialReason, ScanCycleWorkSnapshot, ScannerUsageSaveResult, ScannerWorkSource, + emit_scan_cycle_complete, emit_scan_cycle_partial_with_source, emit_scan_cycle_superseded, global_metrics, }; use rustfs_config::ScannerSpeed; #[cfg(test)] @@ -50,9 +52,12 @@ use rustfs_config::{ use rustfs_config::{ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS}; use serde::{Deserialize, Serialize}; use sha2::{Digest as _, Sha256}; +#[cfg(test)] +use tokio::sync::Notify; use tokio::sync::mpsc; use tokio::time::{Duration, Instant}; use tokio_util::sync::CancellationToken; +use tokio_util::task::AbortOnDropHandle; use tracing::{debug, error, info, instrument, warn}; use crate::storage_api::scan::{ @@ -93,6 +98,44 @@ const SCANNER_CYCLE_STATE_MAGIC: &[u8; 8] = b"RSCYC001"; const SCANNER_CYCLE_STATE_HEADER_LEN: usize = 24; #[cfg(test)] const ENV_SCANNER_START_DELAY_SECS_DEPRECATED: &str = "RUSTFS_DATA_SCANNER_START_DELAY_SECS"; +#[cfg(test)] +type ScannerCycleStatePersistTestHook = (u64, Arc); +#[cfg(test)] +static SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK: LazyLock>> = + LazyLock::new(|| StdMutex::new(None)); + +#[cfg(test)] +struct ScannerCycleStatePersistTestHookGuard; + +#[cfg(test)] +impl Drop for ScannerCycleStatePersistTestHookGuard { + fn drop(&mut self) { + *SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = None; + } +} + +#[cfg(test)] +fn set_scanner_cycle_state_persist_test_hook(leader_epoch: u64, reached: Arc) -> ScannerCycleStatePersistTestHookGuard { + *SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some((leader_epoch, reached)); + ScannerCycleStatePersistTestHookGuard +} + +#[cfg(test)] +fn notify_scanner_cycle_state_persist_test_hook(leader_epoch: u64) { + let reached = SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .as_ref() + .filter(|(expected_epoch, _)| *expected_epoch == leader_epoch) + .map(|(_, reached)| reached.clone()); + if let Some(reached) = reached { + reached.notify_one(); + } +} #[derive(Debug, thiserror::Error)] enum ScannerCycleStateError { @@ -1691,10 +1734,10 @@ fn data_usage_persist_timeout() -> Duration { DataUsageCache::persistence_timeout() } -async fn mark_scan_cycle_idle(cycle_info: &mut CurrentCycle) { +async fn mark_scan_cycle_idle(cycle_info: &mut CurrentCycle, cycle_metrics_guard: &mut ScannerCycleMetricsGuard) { cycle_info.current = 0; global_metrics().clear_current_scan_mode(); - global_metrics().set_cycle(Some(cycle_info.clone())).await; + cycle_metrics_guard.finish(cycle_info.clone()).await; } fn encode_scanner_cycle_state(cycle_info: &CurrentCycle, leader_epoch: u64) -> Result, ScannerCycleStateError> { @@ -2218,6 +2261,8 @@ async fn persist_scanner_cycle_state( return false; } + #[cfg(test)] + notify_scanner_cycle_state_persist_test_hook(leader_epoch); match save_config_with_preconditions(storeapi.clone(), &DATA_USAGE_BLOOM_NAME_PATH, buf.clone(), revision.preconditions()) .await { @@ -2340,7 +2385,6 @@ async fn persist_scanner_cycle_state( if persisted_cycle.next >= cycle_info.next { *cycle_info = persisted_cycle; - global_metrics().set_cycle(Some(cycle_info.clone())).await; debug!( target: "rustfs::scanner", event = EVENT_SCANNER_PERSIST_STATE, @@ -2406,6 +2450,7 @@ async fn finalize_partial_scan_cycle( cycle_info: &mut CurrentCycle, revision: &mut DataUsageCacheRevision, leader_epoch: u64, + cycle_metrics_guard: &mut ScannerCycleMetricsGuard, ) -> bool { // A budget-limited cycle is deliberate pacing, not a failure. The cycle counter // must still advance (and persist) because per-bucket next_cycle is stamped from @@ -2422,11 +2467,14 @@ async fn finalize_partial_scan_cycle( error = %err, "Scanner partial cycle could not advance" ); - mark_scan_cycle_idle(cycle_info).await; + mark_scan_cycle_idle(cycle_info, cycle_metrics_guard).await; return false; } - mark_scan_cycle_idle(cycle_info).await; - persist_scanner_cycle_state(ctx, storeapi, cycle_info, revision, leader_epoch).await + cycle_info.current = 0; + global_metrics().clear_current_scan_mode(); + let persisted = persist_scanner_cycle_state(ctx, storeapi, cycle_info, revision, leader_epoch).await; + cycle_metrics_guard.finish(cycle_info.clone()).await; + persisted } async fn persist_required_scanner_cycle_floor( @@ -2436,6 +2484,7 @@ async fn persist_required_scanner_cycle_floor( revision: &mut DataUsageCacheRevision, leader_epoch: u64, required_cycle: u64, + cycle_metrics_guard: &mut ScannerCycleMetricsGuard, ) -> bool { if required_cycle <= cycle_info.current || required_cycle == u64::MAX { error!( @@ -2448,13 +2497,16 @@ async fn persist_required_scanner_cycle_floor( state = "invalid_cache_cycle_floor", "Scanner cache cycle floor is invalid" ); - mark_scan_cycle_idle(cycle_info).await; + mark_scan_cycle_idle(cycle_info, cycle_metrics_guard).await; return false; } cycle_info.next = cycle_info.next.max(required_cycle); - mark_scan_cycle_idle(cycle_info).await; - persist_scanner_cycle_state(ctx, storeapi, cycle_info, revision, leader_epoch).await + cycle_info.current = 0; + global_metrics().clear_current_scan_mode(); + let persisted = persist_scanner_cycle_state(ctx, storeapi, cycle_info, revision, leader_epoch).await; + cycle_metrics_guard.finish(cycle_info.clone()).await; + persisted } async fn await_scanner_cycle_with_lock_fence( @@ -2513,7 +2565,7 @@ async fn run_data_scanner_cycle( let now = Instant::now(); cycle_info.started = Utc::now(); - global_metrics().set_cycle(Some(cycle_info.clone())).await; + let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; let mut background_heal_info = read_background_heal_info(storeapi.clone()).await; @@ -2564,14 +2616,14 @@ async fn run_data_scanner_cycle( "Scanner cycle could not capture the data usage persistence baseline" ); emit_scan_cycle_complete(false, cycle_start.elapsed()); - mark_scan_cycle_idle(cycle_info).await; + mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await; return ScannerCycleOutcome::Failed; } }; let (sender, receiver) = mpsc::channel::(1); let storeapi_clone = storeapi.clone(); let ctx_clone = ctx.clone(); - let mut usage_persist_task = tokio::spawn(async move { + let mut usage_persist_task = AbortOnDropHandle::new(tokio::spawn(async move { store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( ctx_clone, storeapi_clone, @@ -2580,10 +2632,9 @@ async fn run_data_scanner_cycle( Some(usage_persist_baseline), ) .await - }); + })); let done_cycle = Metrics::time(Metric::ScanCycle); - let cycle_work_start = global_metrics().start_scan_cycle_work(); let cycle_budget = ScannerCycleBudget::new(ctx, cycle_budget_config); let scan_result = storeapi .clone() @@ -2640,7 +2691,6 @@ async fn run_data_scanner_cycle( } }; let unresolved_heal_work = global_metrics().current_scan_cycle_has_unresolved_heal_work(); - global_metrics().finish_scan_cycle_work(cycle_work_start); let scan_cycle_result = match scan_result { Ok(result) => result, @@ -2663,7 +2713,7 @@ async fn run_data_scanner_cycle( { save_background_heal_info(storeapi.clone(), new_heal_info).await; } - mark_scan_cycle_idle(cycle_info).await; + mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await; return ScannerCycleOutcome::Failed; } }; @@ -2678,7 +2728,7 @@ async fn run_data_scanner_cycle( "Scanner cycle stopped before committing cycle state" ); emit_scan_cycle_complete(false, cycle_start.elapsed()); - mark_scan_cycle_idle(cycle_info).await; + mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await; return ScannerCycleOutcome::Failed; } if let Some(required_cycle) = scan_cycle_result.required_cycle_floor() { @@ -2700,6 +2750,7 @@ async fn run_data_scanner_cycle( cycle_revision, leader_epoch, required_cycle, + &mut cycle_metrics_guard, ) .await { @@ -2719,7 +2770,7 @@ async fn run_data_scanner_cycle( "Scanner cycle completed without a durable data usage snapshot" ); emit_scan_cycle_complete(false, cycle_start.elapsed()); - mark_scan_cycle_idle(cycle_info).await; + mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await; return ScannerCycleOutcome::Failed; } if budget_elapsed { @@ -2743,7 +2794,16 @@ async fn run_data_scanner_cycle( scan_cycle_partial_reason(budget_reason), scan_cycle_partial_source(budget_reason), ); - return if finalize_partial_scan_cycle(ctx, storeapi.clone(), cycle_info, cycle_revision, leader_epoch).await { + return if finalize_partial_scan_cycle( + ctx, + storeapi.clone(), + cycle_info, + cycle_revision, + leader_epoch, + &mut cycle_metrics_guard, + ) + .await + { ScannerCycleOutcome::Partial } else { ScannerCycleOutcome::Failed @@ -2792,7 +2852,7 @@ async fn run_data_scanner_cycle( "Scanner cycle completed without a durable data usage snapshot" ); emit_scan_cycle_complete(false, cycle_start.elapsed()); - mark_scan_cycle_idle(cycle_info).await; + mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await; return ScannerCycleOutcome::Failed; } ScannerCycleOutcome::Partial => { @@ -2818,7 +2878,16 @@ async fn run_data_scanner_cycle( ); } emit_scan_cycle_partial_with_source(cycle_start.elapsed(), ScanCyclePartialReason::Unknown, None); - return if finalize_partial_scan_cycle(ctx, storeapi.clone(), cycle_info, cycle_revision, leader_epoch).await { + return if finalize_partial_scan_cycle( + ctx, + storeapi.clone(), + cycle_info, + cycle_revision, + leader_epoch, + &mut cycle_metrics_guard, + ) + .await + { ScannerCycleOutcome::Partial } else { ScannerCycleOutcome::Failed @@ -2834,7 +2903,16 @@ async fn run_data_scanner_cycle( state = "superseded", "Scanner cycle usage snapshot was superseded by concurrent namespace activity" ); - if finalize_partial_scan_cycle(ctx, storeapi.clone(), cycle_info, cycle_revision, leader_epoch).await { + if finalize_partial_scan_cycle( + ctx, + storeapi.clone(), + cycle_info, + cycle_revision, + leader_epoch, + &mut cycle_metrics_guard, + ) + .await + { emit_scan_cycle_superseded(cycle_start.elapsed()); return ScannerCycleOutcome::Superseded; } @@ -2853,7 +2931,7 @@ async fn run_data_scanner_cycle( error = %err, "Scanner completed cycle could not advance" ); - mark_scan_cycle_idle(cycle_info).await; + mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await; emit_scan_cycle_complete(false, cycle_start.elapsed()); return ScannerCycleOutcome::Failed; } @@ -2862,9 +2940,8 @@ async fn run_data_scanner_cycle( global_metrics().clear_current_scan_mode(); retain_recent_cycle_completions(&mut cycle_info.cycle_completed); - global_metrics().set_cycle(Some(cycle_info.clone())).await; if !persist_scanner_cycle_state(ctx, storeapi.clone(), cycle_info, cycle_revision, leader_epoch).await { - mark_scan_cycle_idle(cycle_info).await; + cycle_metrics_guard.finish(cycle_info.clone()).await; emit_scan_cycle_complete(false, cycle_start.elapsed()); return ScannerCycleOutcome::Failed; } @@ -2888,9 +2965,37 @@ async fn run_data_scanner_cycle( "Scanner cycle completed" ); + cycle_metrics_guard.finish(cycle_info.clone()).await; scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, pending_maintenance_work) } +struct ScannerCycleMetricsGuard { + start: Option, +} + +impl ScannerCycleMetricsGuard { + async fn new(cycle: CurrentCycle) -> Self { + Self { + start: Some(global_metrics().start_scan_cycle_work_with_cycle(cycle).await), + } + } + + async fn finish(&mut self, cycle: CurrentCycle) { + if let Some(start) = self.start { + global_metrics().finish_scan_cycle_work_with_cycle(start, cycle).await; + self.start = None; + } + } +} + +impl Drop for ScannerCycleMetricsGuard { + fn drop(&mut self) { + if let Some(start) = self.start.take() { + global_metrics().finish_scan_cycle_work(start); + } + } +} + async fn record_scanner_leader_lock_lost(message: &'static str) { reset_scanner_cycle_schedule(); record_scanner_leader_lock_state("lost"); @@ -3443,7 +3548,7 @@ enum DataUsagePersistTaskResult { async fn wait_for_data_usage_persist_task( ctx: &CancellationToken, - task: &mut tokio::task::JoinHandle, + task: &mut AbortOnDropHandle, timeout: Duration, ) -> DataUsagePersistTaskResult { tokio::select! { @@ -3840,8 +3945,9 @@ mod tests { use super::*; use crate::EcstoreResult; use crate::{ - ScannerGetObjectReader as GetObjectReader, ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, - ScannerPutObjReader as PutObjReader, + Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerGetObjectReader as GetObjectReader, + ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, ScannerPutObjReader as PutObjReader, + init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests, init_local_disks_with_instance_ctx, }; use serial_test::serial; use std::collections::HashMap; @@ -3853,6 +3959,47 @@ mod tests { const TEST_DEFAULT_SCANNER_CYCLE_SECS: u64 = 24 * 60 * 60; + async fn setup_scanner_cycle_store() -> (tempfile::TempDir, Arc) { + init_ecstore_config_for_scanner_tests(); + let temp_dir = tempfile::tempdir().expect("scanner cycle test directory should be created"); + let mut endpoints = Vec::new(); + for disk_index in 0..4 { + let disk_path = temp_dir.path().join(format!("disk{disk_index}")); + tokio::fs::create_dir_all(&disk_path) + .await + .expect("scanner cycle test disk should be created"); + let mut endpoint = + Endpoint::try_from(disk_path.to_str().expect("disk path should be utf8")).expect("endpoint should parse"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(disk_index); + endpoints.push(endpoint); + } + let endpoint_pools = EndpointServerPools::from(vec![PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 4, + endpoints: Endpoints::from(endpoints), + cmd_line: "scanner-cycle-metrics".to_string(), + platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), + }]); + let instance_ctx = Arc::new(InstanceContext::new()); + init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone()) + .await + .expect("scanner cycle test disks should initialize"); + let store = ECStore::new_with_instance_ctx( + "127.0.0.1:0".parse().expect("test address should parse"), + endpoint_pools, + CancellationToken::new(), + instance_ctx, + ) + .await + .expect("scanner cycle test ECStore should initialize"); + init_bucket_metadata_sys_for_scanner_tests(store.clone()).await; + + (temp_dir, store) + } + fn assert_run_data_scanner_signature(_run: F) where F: Fn(CancellationToken, Arc) -> Fut, @@ -4312,9 +4459,9 @@ mod tests { }; global_metrics().set_current_scan_mode(HealScanMode::Deep); - global_metrics().set_cycle(Some(cycle_info.clone())).await; + let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; - mark_scan_cycle_idle(&mut cycle_info).await; + mark_scan_cycle_idle(&mut cycle_info, &mut cycle_metrics_guard).await; let published = global_metrics() .get_cycle() @@ -4330,6 +4477,123 @@ mod tests { global_metrics().set_cycle(None).await; } + #[tokio::test] + #[serial] + async fn scanner_cycle_metrics_guard_covers_published_first_cycle_lifetime() { + let cycle_started = Utc::now() - chrono::Duration::seconds(5); + let mut cycle_info = CurrentCycle { + current: 0, + next: 1, + started: cycle_started, + ..Default::default() + }; + let mut guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; + let setup_report = global_metrics().report().await; + assert!(setup_report.current_cycle_active); + assert_eq!(setup_report.current_cycle, 0); + assert_eq!(setup_report.current_started, cycle_started); + + mark_scan_cycle_idle(&mut cycle_info, &mut guard).await; + let idle_report = global_metrics().report().await; + assert!(!idle_report.current_cycle_active); + + global_metrics().set_cycle(None).await; + } + + #[tokio::test] + #[serial] + async fn scanner_cycle_metrics_guard_keeps_active_cycle_published_during_finalization() { + let mut cycle_info = CurrentCycle { + current: 12, + next: 13, + started: Utc::now(), + ..Default::default() + }; + let mut guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; + + cycle_info.current = 0; + tokio::task::yield_now().await; + let finalizing_report = global_metrics().report().await; + assert!(finalizing_report.current_cycle_active); + assert_eq!(finalizing_report.current_cycle, 12); + + guard.finish(cycle_info).await; + let idle_report = global_metrics().report().await; + assert!(!idle_report.current_cycle_active); + assert_eq!(idle_report.current_cycle, 0); + + global_metrics().set_cycle(None).await; + } + + #[tokio::test] + #[serial] + async fn scanner_cycle_metrics_guard_drop_clears_activity() { + let guard = ScannerCycleMetricsGuard::new(CurrentCycle { + current: 12, + next: 13, + started: Utc::now(), + ..Default::default() + }) + .await; + assert!(global_metrics().report().await.current_cycle_active); + + drop(guard); + + assert!(!global_metrics().report().await.current_cycle_active); + global_metrics().set_cycle(None).await; + } + + #[tokio::test] + #[serial] + async fn run_data_scanner_cycle_publishes_activity_for_owner_lifetime() { + let (_temp_dir, store) = setup_scanner_cycle_store().await; + let ctx = CancellationToken::new(); + let mut cycle_info = CurrentCycle::default(); + let mut revision = DataUsageCacheRevision::Missing; + let leader_epoch = u64::MAX - 1; + let state_persist_reached = Arc::new(Notify::new()); + let _state_persist_hook = set_scanner_cycle_state_persist_test_hook(leader_epoch, state_persist_reached.clone()); + let state_lock = store + .new_ns_lock(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()) + .await + .expect("scanner cycle state lock should be created"); + let state_guard = state_lock + .get_write_lock(Duration::from_secs(1)) + .await + .expect("scanner cycle state lock should be acquired"); + let mut cycle = Box::pin(run_data_scanner_cycle(&ctx, &store, &mut cycle_info, &mut revision, leader_epoch)); + let waker = std::task::Waker::noop(); + let mut context = std::task::Context::from_waker(waker); + + assert!(cycle.as_mut().poll(&mut context).is_pending()); + let active = global_metrics().report().await; + assert!(active.current_cycle_active); + assert_eq!(active.current_cycle, 0); + + tokio::time::timeout(Duration::from_secs(30), async { + tokio::select! { + outcome = &mut cycle => panic!("scanner cycle finished before state persistence was released: {outcome:?}"), + _ = state_persist_reached.notified() => {} + } + }) + .await + .expect("scanner cycle should reach state persistence"); + let finalizing = global_metrics().report().await; + assert!(finalizing.current_cycle_active); + + drop(state_guard); + let outcome = tokio::time::timeout(Duration::from_secs(30), cycle) + .await + .expect("scanner cycle should finish"); + assert!(matches!( + outcome, + ScannerCycleOutcome::Completed | ScannerCycleOutcome::CompletedWithPendingMaintenance + )); + assert!(!global_metrics().report().await.current_cycle_active); + + global_metrics().set_cycle(None).await; + } + #[tokio::test] #[serial] async fn test_finalize_partial_scan_cycle_advances_and_persists_counter() { @@ -4342,8 +4606,11 @@ mod tests { cycle_completed: vec![], started: Utc::now(), }; + let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; - assert!(finalize_partial_scan_cycle(&ctx, store.clone(), &mut cycle_info, &mut revision, 1).await); + assert!( + finalize_partial_scan_cycle(&ctx, store.clone(), &mut cycle_info, &mut revision, 1, &mut cycle_metrics_guard,).await + ); assert_eq!(cycle_info.next, 13); assert_eq!(cycle_info.current, 0); @@ -4377,8 +4644,20 @@ mod tests { cycle_completed: vec![], started: Utc::now(), }; + let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; - assert!(persist_required_scanner_cycle_floor(&ctx, store.clone(), &mut cycle_info, &mut revision, 7, 19).await); + assert!( + persist_required_scanner_cycle_floor( + &ctx, + store.clone(), + &mut cycle_info, + &mut revision, + 7, + 19, + &mut cycle_metrics_guard, + ) + .await + ); assert_eq!(cycle_info.current, 0); assert_eq!(cycle_info.next, 19); @@ -4405,22 +4684,37 @@ mod tests { cycle_completed: vec![], started: Utc::now(), }; + let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; - assert!(!persist_required_scanner_cycle_floor(&ctx, store.clone(), &mut cycle_info, &mut revision, 7, 12).await); - assert_eq!(cycle_info.next, 12); - assert_eq!(revision, DataUsageCacheRevision::Missing); assert!( !persist_required_scanner_cycle_floor( &ctx, store.clone(), - &mut CurrentCycle { - current: 12, - next: 12, - ..Default::default() - }, + &mut cycle_info, + &mut revision, + 7, + 12, + &mut cycle_metrics_guard, + ) + .await + ); + assert_eq!(cycle_info.next, 12); + assert_eq!(revision, DataUsageCacheRevision::Missing); + let mut max_cycle_info = CurrentCycle { + current: 12, + next: 12, + ..Default::default() + }; + let mut max_cycle_metrics_guard = ScannerCycleMetricsGuard::new(max_cycle_info.clone()).await; + assert!( + !persist_required_scanner_cycle_floor( + &ctx, + store.clone(), + &mut max_cycle_info, &mut revision, 7, u64::MAX, + &mut max_cycle_metrics_guard, ) .await ); @@ -4685,8 +4979,9 @@ mod tests { cycle_completed: vec![], started: Utc::now(), }; + let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; - assert!(!finalize_partial_scan_cycle(&ctx, store, &mut cycle_info, &mut revision, 1).await); + assert!(!finalize_partial_scan_cycle(&ctx, store, &mut cycle_info, &mut revision, 1, &mut cycle_metrics_guard,).await); assert_eq!(cycle_info.next, 13); assert_eq!(cycle_info.current, 0); assert_eq!(revision, DataUsageCacheRevision::Missing); @@ -5943,10 +6238,10 @@ mod tests { #[tokio::test] async fn data_usage_persist_wait_aborts_when_scanner_is_cancelled() { let ctx = CancellationToken::new(); - let mut task = tokio::spawn(async { + let mut task = AbortOnDropHandle::new(tokio::spawn(async { std::future::pending::<()>().await; DataUsagePersistOutcome::Saved - }); + })); ctx.cancel(); let result = wait_for_data_usage_persist_task(&ctx, &mut task, Duration::from_secs(60)).await; @@ -5958,10 +6253,10 @@ mod tests { #[tokio::test(start_paused = true)] async fn data_usage_persist_wait_aborts_after_timeout() { let ctx = CancellationToken::new(); - let mut task = tokio::spawn(async { + let mut task = AbortOnDropHandle::new(tokio::spawn(async { std::future::pending::<()>().await; DataUsagePersistOutcome::Saved - }); + })); let result = wait_for_data_usage_persist_task(&ctx, &mut task, Duration::from_secs(30)).await;