From 2fccfdeabe9a28a66a36c8787edbc09d8bd0ebe2 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sat, 22 Aug 2026 22:38:18 +0800 Subject: [PATCH] fix(scanner): fence timed out scan cycles (#6352) * fix(scanner): fence timed out scan cycles * fix(scanner): cancel scan workers with cycle scope * fix(scanner): reject persisted timer overflow * fix(scanner): reject terminal leadership epochs * fix(scanner): reject trailing cycle state bytes --- crates/common/src/metrics.rs | 56 +++++ crates/config/README.md | 6 + crates/config/src/constants/scanner.rs | 9 +- .../ecstore/src/services/metrics_realtime.rs | 12 + crates/madmin/src/metrics.rs | 16 ++ crates/scanner/src/runtime_config.rs | 118 ++++++++-- crates/scanner/src/scanner.rs | 217 ++++++++++++++++-- crates/scanner/src/scanner/cycle_state.rs | 60 +++++ crates/scanner/src/scanner/tests.rs | 189 ++++++++++++++- crates/scanner/src/scanner_budget.rs | 161 +++++++++++-- crates/scanner/src/scanner_io.rs | 1 + crates/scanner/src/scanner_io/io_cache.rs | 8 +- crates/scanner/src/scanner_io/io_cycle.rs | 4 +- crates/utils/src/envs.rs | 4 +- docs/operations/scanner-runtime-controls.md | 21 +- 15 files changed, 803 insertions(+), 79 deletions(-) diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index 51eef967a..813e39ac3 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -901,6 +901,10 @@ pub struct Metrics { scanner_cycle_max_duration_millis: AtomicU64, scanner_cycle_max_objects: AtomicU64, scanner_cycle_max_directories: AtomicU64, + scanner_cycle_timeout_total: AtomicU64, + scanner_cycle_recovery_required_total: AtomicU64, + scanner_cycle_last_progress_age_seconds: AtomicU64, + scanner_leader_lease_without_progress: AtomicBool, scanner_bitrot_cycle_enabled: AtomicBool, scanner_bitrot_cycle_millis: AtomicU64, scanner_checkpoint: Mutex>, @@ -1370,6 +1374,14 @@ pub struct ScannerMetricsReport { #[serde(default)] pub cycle_max_directories: u64, #[serde(default)] + pub cycle_timeout_total: u64, + #[serde(default)] + pub cycle_recovery_required_total: u64, + #[serde(default)] + pub cycle_last_progress_age: u64, + #[serde(default)] + pub leader_lease_without_progress: bool, + #[serde(default)] pub bitrot_cycle_enabled: bool, #[serde(default)] pub bitrot_cycle_seconds: f64, @@ -1430,6 +1442,9 @@ const OTEL_SCANNER_BUCKETS_SCANNED: &str = "rustfs_scanner_buckets_scanned_total const OTEL_SCANNER_CYCLES: &str = "rustfs_scanner_cycles_total"; const OTEL_SCANNER_CYCLE_DURATION_SECONDS: &str = "rustfs_scanner_cycle_duration_seconds"; const OTEL_SCANNER_BUCKET_DRIVE_DURATION_SECONDS: &str = "rustfs_scanner_bucket_drive_duration_seconds"; +const OTEL_SCANNER_CYCLE_TIMEOUT_TOTAL: &str = "rustfs_scanner_cycle_timeout_total"; +const OTEL_SCANNER_CYCLE_LAST_PROGRESS_AGE: &str = "rustfs_scanner_cycle_last_progress_age"; +const OTEL_SCANNER_LEADER_LEASE_WITHOUT_PROGRESS: &str = "rustfs_scanner_leader_lease_without_progress"; fn scan_cycle_result_label(result: u8) -> &'static str { match result { @@ -1913,6 +1928,10 @@ impl Metrics { scanner_cycle_max_duration_millis: AtomicU64::new(0), scanner_cycle_max_objects: AtomicU64::new(0), scanner_cycle_max_directories: AtomicU64::new(0), + scanner_cycle_timeout_total: AtomicU64::new(0), + scanner_cycle_recovery_required_total: AtomicU64::new(0), + scanner_cycle_last_progress_age_seconds: AtomicU64::new(0), + scanner_leader_lease_without_progress: AtomicBool::new(false), scanner_bitrot_cycle_enabled: AtomicBool::new(false), scanner_bitrot_cycle_millis: AtomicU64::new(0), scanner_checkpoint: Mutex::new(None), @@ -2412,12 +2431,29 @@ impl Metrics { .store(cycle_max_objects.unwrap_or_default(), Ordering::Relaxed); self.scanner_cycle_max_directories .store(cycle_max_directories.unwrap_or_default(), Ordering::Relaxed); + self.scanner_leader_lease_without_progress.store(false, Ordering::Relaxed); + self.scanner_cycle_last_progress_age_seconds.store(0, Ordering::Relaxed); + metrics::gauge!(OTEL_SCANNER_LEADER_LEASE_WITHOUT_PROGRESS).set(0.0); + metrics::gauge!(OTEL_SCANNER_CYCLE_LAST_PROGRESS_AGE).set(0.0); self.scanner_bitrot_cycle_enabled .store(bitrot_cycle.is_some(), Ordering::Relaxed); self.scanner_bitrot_cycle_millis .store(bitrot_cycle.map(duration_millis_saturated).unwrap_or_default(), Ordering::Relaxed); } + pub fn record_scanner_cycle_timeout(&self, recovery_required: bool, progress_age: Duration) { + self.scanner_cycle_timeout_total.fetch_add(1, Ordering::Relaxed); + if recovery_required { + self.scanner_cycle_recovery_required_total.fetch_add(1, Ordering::Relaxed); + } + self.scanner_cycle_last_progress_age_seconds + .store(progress_age.as_secs(), Ordering::Relaxed); + self.scanner_leader_lease_without_progress.store(true, Ordering::Relaxed); + metrics::counter!(OTEL_SCANNER_CYCLE_TIMEOUT_TOTAL).increment(1); + metrics::gauge!(OTEL_SCANNER_CYCLE_LAST_PROGRESS_AGE).set(progress_age.as_secs_f64()); + metrics::gauge!(OTEL_SCANNER_LEADER_LEASE_WITHOUT_PROGRESS).set(1.0); + } + pub fn record_scanner_set_scan_state(&self, concurrency_limit: Option, queued: Option, active: Option) { if let Some(concurrency_limit) = concurrency_limit { self.scanner_set_scan_concurrency_limit @@ -3265,6 +3301,10 @@ impl Metrics { m.cycle_max_duration_seconds = self.scanner_cycle_max_duration_millis.load(Ordering::Relaxed) as f64 / 1000.0; m.cycle_max_objects = self.scanner_cycle_max_objects.load(Ordering::Relaxed); m.cycle_max_directories = self.scanner_cycle_max_directories.load(Ordering::Relaxed); + m.cycle_timeout_total = self.scanner_cycle_timeout_total.load(Ordering::Relaxed); + m.cycle_recovery_required_total = self.scanner_cycle_recovery_required_total.load(Ordering::Relaxed); + m.cycle_last_progress_age = self.scanner_cycle_last_progress_age_seconds.load(Ordering::Relaxed); + m.leader_lease_without_progress = self.scanner_leader_lease_without_progress.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() { @@ -4926,4 +4966,20 @@ mod tests { assert!(!report.bitrot_cycle_enabled); assert_eq!(report.bitrot_cycle_seconds, 0.0); } + + #[tokio::test] + async fn scanner_cycle_timeout_metrics_reset_for_a_new_cycle() { + let metrics = Metrics::new(); + metrics.record_scanner_cycle_timeout(true, Duration::from_secs(17)); + let timed_out = metrics.report().await; + assert_eq!(timed_out.cycle_timeout_total, 1); + assert_eq!(timed_out.cycle_last_progress_age, 17); + assert!(timed_out.leader_lease_without_progress); + + metrics.record_scanner_cycle_config(Duration::from_secs(60), None, Some(Duration::from_secs(1)), None, None); + let current = metrics.report().await; + assert_eq!(current.cycle_timeout_total, 1); + assert_eq!(current.cycle_last_progress_age, 0); + assert!(!current.leader_lease_without_progress); + } } diff --git a/crates/config/README.md b/crates/config/README.md index 02cf81d52..338b85dbf 100644 --- a/crates/config/README.md +++ b/crates/config/README.md @@ -84,6 +84,12 @@ Current guidance: - `RUSTFS_SCANNER_CYCLE_MAX_OBJECTS` (canonical) - `RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES` (canonical) +Scanner cycle budget controls: + +- When `RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS` is unset, the finite default is 1800 seconds (30 minutes), matching the scanner benchmark guidance. +- An explicit `0` preserves the compatibility behavior of an unbounded runtime budget. Object and directory budgets likewise remain unbounded when explicitly set to `0`. +- A timed-out cycle cancels cooperative scanner work, then fences its leader epoch before releasing the lease. An uncooperative I/O operation is dropped after the bounded shutdown window; its cursor is not claimed to be durable and the scanner reports `recovery-required` when the worker cannot stop cooperatively, the cycle state was not confirmed durable, or epoch fencing cannot be persisted. + ## Mmap read environment aliases - `RUSTFS_OBJECT_MMAP_READ_ENABLE` (canonical) diff --git a/crates/config/src/constants/scanner.rs b/crates/config/src/constants/scanner.rs index 8086c3229..953ef789b 100644 --- a/crates/config/src/constants/scanner.rs +++ b/crates/config/src/constants/scanner.rs @@ -143,9 +143,12 @@ pub const ENV_SCANNER_MAX_WAIT_SECS: &str = "RUSTFS_SCANNER_MAX_WAIT_SECS"; /// Default scanner speed preset. pub const DEFAULT_SCANNER_SPEED: &str = "default"; -/// Default scanner cycle runtime budget. -/// `0` keeps the existing unbounded per-cycle behavior. -pub const DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS: u64 = 0; +/// Default scanner cycle runtime budget when no override is configured. +/// +/// An explicit `0` remains the compatibility escape hatch for an unbounded +/// cycle. Keeping the unset default finite prevents a stalled scanner I/O +/// operation from holding the leader lease forever. +pub const DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS: u64 = 30 * 60; /// Default scanner per-cycle object budget. /// `0` keeps the existing unbounded per-cycle behavior. diff --git a/crates/ecstore/src/services/metrics_realtime.rs b/crates/ecstore/src/services/metrics_realtime.rs index e4fc919e6..6e2e5cd7d 100644 --- a/crates/ecstore/src/services/metrics_realtime.rs +++ b/crates/ecstore/src/services/metrics_realtime.rs @@ -256,6 +256,10 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo cycle_max_duration_seconds: metrics.cycle_max_duration_seconds, cycle_max_objects: metrics.cycle_max_objects, cycle_max_directories: metrics.cycle_max_directories, + cycle_timeout_total: metrics.cycle_timeout_total, + cycle_recovery_required_total: metrics.cycle_recovery_required_total, + cycle_last_progress_age: metrics.cycle_last_progress_age, + leader_lease_without_progress: metrics.leader_lease_without_progress, bitrot_cycle_enabled: metrics.bitrot_cycle_enabled, bitrot_cycle_seconds: metrics.bitrot_cycle_seconds, scan_checkpoint: metrics.scan_checkpoint.map(|checkpoint| MadminScannerCheckpointReport { @@ -611,6 +615,10 @@ mod test { current_started: chrono_to_jiff_timestamp(current_started), last_cycle_partial_source: "usage".to_string(), last_cycle_partial_source_code: 1, + cycle_timeout_total: 3, + cycle_recovery_required_total: 2, + cycle_last_progress_age: 17, + leader_lease_without_progress: true, partial_cycles_by_source: vec![rustfs_common::metrics::ScannerSourceCycleSnapshot { source: "usage".to_string(), cycles: 2, @@ -622,6 +630,10 @@ mod test { assert_eq!(scanner.current_started, chrono_to_jiff_timestamp(current_started)); assert_eq!(scanner.last_cycle_partial_source, "usage"); assert_eq!(scanner.last_cycle_partial_source_code, 1); + assert_eq!(scanner.cycle_timeout_total, 3); + assert_eq!(scanner.cycle_recovery_required_total, 2); + assert_eq!(scanner.cycle_last_progress_age, 17); + assert!(scanner.leader_lease_without_progress); let usage = scanner .partial_cycles_by_source .iter() diff --git a/crates/madmin/src/metrics.rs b/crates/madmin/src/metrics.rs index 49e1e5ff2..4e55563a5 100644 --- a/crates/madmin/src/metrics.rs +++ b/crates/madmin/src/metrics.rs @@ -689,6 +689,14 @@ pub struct ScannerMetrics { pub cycle_max_objects: u64, #[serde(rename = "cycle_max_directories", default)] pub cycle_max_directories: u64, + #[serde(rename = "cycle_timeout_total", default)] + pub cycle_timeout_total: u64, + #[serde(rename = "cycle_recovery_required_total", default)] + pub cycle_recovery_required_total: u64, + #[serde(rename = "cycle_last_progress_age", default)] + pub cycle_last_progress_age: u64, + #[serde(rename = "leader_lease_without_progress", default)] + pub leader_lease_without_progress: bool, #[serde(rename = "bitrot_cycle_enabled", default)] pub bitrot_cycle_enabled: bool, #[serde(rename = "bitrot_cycle_seconds", default)] @@ -764,6 +772,8 @@ impl ScannerMetrics { self.cycle_max_duration_seconds = other.cycle_max_duration_seconds; self.cycle_max_objects = other.cycle_max_objects; self.cycle_max_directories = other.cycle_max_directories; + self.cycle_last_progress_age = other.cycle_last_progress_age; + self.leader_lease_without_progress = other.leader_lease_without_progress; self.bitrot_cycle_enabled = other.bitrot_cycle_enabled; self.bitrot_cycle_seconds = other.bitrot_cycle_seconds; } @@ -857,6 +867,12 @@ impl ScannerMetrics { .saturating_add(other.last_cycle_replication_checks); self.last_cycle_usage_saves = self.last_cycle_usage_saves.saturating_add(other.last_cycle_usage_saves); self.failed_cycles = self.failed_cycles.saturating_add(other.failed_cycles); + self.cycle_timeout_total = self.cycle_timeout_total.saturating_add(other.cycle_timeout_total); + self.cycle_recovery_required_total = self + .cycle_recovery_required_total + .saturating_add(other.cycle_recovery_required_total); + self.cycle_last_progress_age = self.cycle_last_progress_age.max(other.cycle_last_progress_age); + self.leader_lease_without_progress |= other.leader_lease_without_progress; self.superseded_cycles = self.superseded_cycles.saturating_add(other.superseded_cycles); self.partial_cycles_unknown = self.partial_cycles_unknown.saturating_add(other.partial_cycles_unknown); self.partial_cycles_runtime = self.partial_cycles_runtime.saturating_add(other.partial_cycles_runtime); diff --git a/crates/scanner/src/runtime_config.rs b/crates/scanner/src/runtime_config.rs index 48fb40d2d..2fef6f9f4 100644 --- a/crates/scanner/src/runtime_config.rs +++ b/crates/scanner/src/runtime_config.rs @@ -125,7 +125,10 @@ impl Default for ScannerRuntimeConfig { cycle_interval_source: ScannerRuntimeConfigSource::Default, bitrot_cycle: Some(Duration::from_secs(DEFAULT_HEAL_BITROT_CYCLE_SECS)), bitrot_cycle_source: ScannerRuntimeConfigSource::Default, - cycle_budget: ScannerCycleBudgetConfig::default(), + cycle_budget: ScannerCycleBudgetConfig { + max_duration: Some(Duration::from_secs(DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS)), + ..Default::default() + }, cycle_max_duration_source: ScannerRuntimeConfigSource::Default, cycle_max_objects_source: ScannerRuntimeConfigSource::Default, cycle_max_directories_source: ScannerRuntimeConfigSource::Default, @@ -374,7 +377,10 @@ fn validate_persisted_scanner_runtime_config(config: &ServerConfig) -> Result<() } validate_optional_config_u64(scanner_kvs, SCANNER_START_DELAY, "")?; validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE, "")?; - validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE_MAX_DURATION, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS)?; + if let Some(value) = config_value(scanner_kvs, SCANNER_CYCLE_MAX_DURATION, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS) { + let secs = parse_config_u64(SCANNER_CYCLE_MAX_DURATION, value)?; + cycle_duration_from_secs(SCANNER_CYCLE_MAX_DURATION, secs)?; + } validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE_MAX_OBJECTS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS)?; validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE_MAX_DIRECTORIES, DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES)?; if let Some(value) = config_value(heal_kvs, HEAL_BITROT_CYCLE, DEFAULT_HEAL_BITROT_CYCLE_SECS) { @@ -436,19 +442,46 @@ fn lookup_max_wait( Ok((speed.max_sleep(), speed_source)) } -fn lookup_optional_seconds( - kvs: Option<&KVS>, - key: &'static str, - env_key: &'static str, - default: u64, -) -> Result<(Option, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> { - if let Some(secs) = rustfs_utils::get_env_opt_u64(env_key) { - return Ok((Some(Duration::from_secs(secs)), ScannerRuntimeConfigSource::Env)); +fn lookup_cycle_duration(kvs: Option<&KVS>) -> Result<(Option, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> { + match rustfs_utils::get_env_parse_outcome::(ENV_SCANNER_CYCLE_MAX_DURATION_SECS) { + rustfs_utils::EnvParseOutcome::Parsed(secs) => { + return cycle_duration_from_secs(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, secs) + .map(|duration| (duration, ScannerRuntimeConfigSource::Env)); + } + rustfs_utils::EnvParseOutcome::Invalid => { + // Do not include the raw environment value in the typed error: + // deployments occasionally put sensitive material in inherited + // environment snapshots. The key still identifies the control. + return Err(invalid_value( + ENV_SCANNER_CYCLE_MAX_DURATION_SECS, + "", + "expected unsigned integer seconds", + )); + } + rustfs_utils::EnvParseOutcome::Absent => {} } - if let Some(value) = config_value(kvs, key, default) { - return parse_config_u64(key, value).map(|secs| (Some(Duration::from_secs(secs)), ScannerRuntimeConfigSource::Config)); + + if let Some(value) = config_value(kvs, SCANNER_CYCLE_MAX_DURATION, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS) { + let secs = parse_config_u64(SCANNER_CYCLE_MAX_DURATION, value)?; + return cycle_duration_from_secs(SCANNER_CYCLE_MAX_DURATION, secs) + .map(|duration| (duration, ScannerRuntimeConfigSource::Config)); } - Ok((None, ScannerRuntimeConfigSource::Default)) + + Ok(( + Some(Duration::from_secs(DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS)), + ScannerRuntimeConfigSource::Default, + )) +} + +fn cycle_duration_from_secs(key: &'static str, secs: u64) -> Result, ScannerRuntimeConfigError> { + if secs == 0 { + return Ok(None); + } + let duration = Duration::from_secs(secs); + if std::time::Instant::now().checked_add(duration).is_none() { + return Err(invalid_value(key, "", "duration exceeds the timer range")); + } + Ok(Some(duration)) } fn lookup_start_delay(kvs: Option<&KVS>) -> Result<(Option, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> { @@ -553,12 +586,7 @@ pub(crate) fn lookup_scanner_runtime_config( (speed.cycle_interval(), speed_source) }; - let (cycle_max_duration, cycle_max_duration_source) = lookup_optional_seconds( - scanner_kvs, - SCANNER_CYCLE_MAX_DURATION, - ENV_SCANNER_CYCLE_MAX_DURATION_SECS, - DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS, - )?; + let (cycle_max_duration, cycle_max_duration_source) = lookup_cycle_duration(scanner_kvs)?; let (cycle_max_objects, cycle_max_objects_source) = lookup_count_budget( scanner_kvs, SCANNER_CYCLE_MAX_OBJECTS, @@ -863,10 +891,10 @@ mod tests { use rustfs_config::server_config::{Config as ServerConfig, KVS}; use rustfs_config::{ DEFAULT_DELIMITER, DEFAULT_HEAL_BITROT_CYCLE_SECS, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, - ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY, ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED, - HEAL_BITROT_CYCLE, HEAL_SUB_SYS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, - SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE, - SCANNER_SPEED, SCANNER_SUB_SYS, ScannerSpeed, + ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY, + ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED, HEAL_BITROT_CYCLE, HEAL_SUB_SYS, SCANNER_BITROT_CYCLE, + SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, + SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE, SCANNER_SPEED, SCANNER_SUB_SYS, ScannerSpeed, }; use std::collections::HashMap; use std::time::Duration; @@ -941,6 +969,50 @@ mod tests { }); } + #[test] + fn scanner_unset_budget_uses_safe_default_but_explicit_zero_is_unbounded() { + let config = server_config_with_scanner(&[]); + with_var_unset(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, || { + let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config"); + assert_eq!(resolved.cycle_budget.max_duration, Some(Duration::from_secs(1800))); + assert_eq!(resolved.cycle_max_duration_source, ScannerRuntimeConfigSource::Default); + }); + + let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_DURATION, "0")]); + with_var_unset(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, || { + let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config"); + assert_eq!(resolved.cycle_budget.max_duration, None); + assert_eq!(resolved.cycle_max_duration_source, ScannerRuntimeConfigSource::Config); + }); + } + + #[test] + fn cycle_budget_invalid_or_overflow_config_is_rejected() { + with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("invalid"), || { + let error = lookup_scanner_runtime_config(None).expect_err("invalid duration env must be rejected"); + assert!(error.to_string().contains(ENV_SCANNER_CYCLE_MAX_DURATION_SECS)); + assert!(error.to_string().contains("")); + assert!(!error.to_string().contains(": invalid (")); + }); + with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("18446744073709551616"), || { + assert!(lookup_scanner_runtime_config(None).is_err()); + }); + with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("18446744073709551615"), || { + assert!(lookup_scanner_runtime_config(None).is_err()); + }); + let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_DURATION, "not-a-duration")]); + assert!(lookup_scanner_runtime_config(Some(&config)).is_err()); + } + + #[test] + fn scanner_runtime_config_validation_rejects_overflow_persisted_duration() { + let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_DURATION, "18446744073709551615")]); + + let error = validate_scanner_runtime_config(&config) + .expect_err("persisted duration that exceeds the timer range must be rejected"); + assert!(error.to_string().contains(SCANNER_CYCLE_MAX_DURATION)); + } + #[test] fn scanner_runtime_config_normalizes_persisted_default_speed() { let config = server_config_with_scanner(&[(SCANNER_SPEED, "default")]); diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 7a4a482cf..059de1fb4 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -52,6 +52,7 @@ use rustfs_config::{ }; use rustfs_config::{ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS}; use rustfs_data_usage::observed_data_usage_is_newer; +use rustfs_lock::NamespaceLockGuard; use serde::{Deserialize, Serialize}; use sha2::{Digest as _, Sha256}; use tokio::sync::{Notify, mpsc}; @@ -1037,20 +1038,116 @@ fn data_usage_persist_timeout() -> Duration { DataUsageCache::persistence_timeout() } +#[cfg(not(test))] +const SCANNER_CYCLE_EPOCH_FENCE_TIMEOUT: Duration = Duration::from_secs(30); +#[cfg(test)] +const SCANNER_CYCLE_EPOCH_FENCE_TIMEOUT: Duration = Duration::from_millis(50); + +async fn fence_scanner_epoch_after_cycle_timeout( + ctx: &CancellationToken, + storeapi: Arc, + cycle_info: &mut CurrentCycle, + cycle_revision: &mut DataUsageCacheRevision, + leader_epoch: &mut u64, + lock_lost: LockLost, +) -> bool +where + Store: ScannerObjectIO, + LockLost: Future, +{ + let fence_ctx = ctx.child_token(); + let claim = claim_scanner_leadership(&fence_ctx, storeapi, cycle_info, cycle_revision, leader_epoch); + tokio::pin!(claim); + tokio::pin!(lock_lost); + tokio::select! { + biased; + _ = &mut lock_lost => { + fence_ctx.cancel(); + false + } + result = tokio::time::timeout(SCANNER_CYCLE_EPOCH_FENCE_TIMEOUT, &mut claim) => { + result.unwrap_or(false) && !fence_ctx.is_cancelled() + } + } +} + +struct ScannerCycleDeadlineState<'a> { + cycle_info: &'a mut CurrentCycle, + cycle_revision: &'a mut DataUsageCacheRevision, + leader_epoch: &'a mut u64, + cycle_budget: &'a ScannerCycleBudget, +} + +fn cycle_timeout_requires_recovery(worker_stopped: bool, cycle_state_persisted: bool, generation_fenced: bool) -> bool { + !worker_stopped || !cycle_state_persisted || !generation_fenced +} + +async fn handle_scanner_cycle_deadline( + ctx: &CancellationToken, + storeapi: Arc, + state: ScannerCycleDeadlineState<'_>, + worker_stopped: bool, + guard: &mut NamespaceLockGuard, +) where + Store: ScannerObjectIO, +{ + let fenced = fence_scanner_epoch_after_cycle_timeout( + ctx, + storeapi, + state.cycle_info, + state.cycle_revision, + state.leader_epoch, + guard.lock_lost_notified(), + ) + .await; + let cycle_state_persisted = state.cycle_budget.cycle_state_persisted(); + let recovery_required = cycle_timeout_requires_recovery(worker_stopped, cycle_state_persisted, fenced); + warn!( + target: "rustfs::scanner", + event = EVENT_SCANNER_CYCLE_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + state = "cycle_timeout", + worker_stopped, + cycle_state_persisted, + generation_fenced = fenced, + recovery_required, + "Scanner cycle deadline expired; durable cursor/generation fencing completed when possible" + ); + global_metrics().record_scanner_cycle_timeout(recovery_required, state.cycle_budget.progress_age()); + // Stop renewing before releasing the lease. A new leader can then claim the + // higher persisted generation instead of inheriting the expired worker. + guard.release(); + global_metrics().set_cycle(None).await; +} + 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(); cycle_metrics_guard.finish(cycle_info.clone()).await; } -#[instrument(skip_all)] -#[hotpath::measure] +#[cfg(test)] async fn run_data_scanner_cycle( ctx: &CancellationToken, storeapi: &Arc, cycle_info: &mut CurrentCycle, cycle_revision: &mut DataUsageCacheRevision, leader_epoch: u64, +) -> ScannerCycleOutcome { + let cycle_budget = ScannerCycleBudget::new(ctx, scanner_cycle_budget_config()); + run_data_scanner_cycle_with_budget(ctx, storeapi, cycle_info, cycle_revision, leader_epoch, cycle_budget).await +} + +#[instrument(skip_all)] +#[hotpath::measure] +async fn run_data_scanner_cycle_with_budget( + ctx: &CancellationToken, + storeapi: &Arc, + cycle_info: &mut CurrentCycle, + cycle_revision: &mut DataUsageCacheRevision, + leader_epoch: u64, + cycle_budget: Arc, ) -> ScannerCycleOutcome { let _activity_guard = ScannerActivityGuard::new(); if let Err(err) = refresh_scanner_runtime_config_from_global() { @@ -1066,7 +1163,11 @@ async fn run_data_scanner_cycle( } let configured_cycle_interval = scanner_cycle_interval(); let configured_bitrot_cycle = scanner_bitrot_cycle(); - let cycle_budget_config = scanner_cycle_budget_config(); + let cycle_budget_config = ScannerCycleBudgetConfig { + max_duration: cycle_budget.max_duration(), + max_objects: cycle_budget.max_objects(), + max_directories: cycle_budget.max_directories(), + }; let usage_persist_timeout = data_usage_persist_timeout(); global_metrics().record_scanner_cycle_config( configured_cycle_interval, @@ -1137,7 +1238,6 @@ async fn run_data_scanner_cycle( let (sender, receiver) = mpsc::channel::(1); let done_cycle = Metrics::time(Metric::ScanCycle); - let cycle_budget = ScannerCycleBudget::new(ctx, cycle_budget_config); let scan_result = storeapi .clone() .nsscanner_with_status( @@ -1277,7 +1377,7 @@ async fn run_data_scanner_cycle( "Scanner cycle is recovering to a newer durable cache generation" ); emit_scan_cycle_partial_with_source(cycle_start.elapsed(), ScanCyclePartialReason::Unknown, None); - return if persist_required_scanner_cycle_floor( + let persisted = persist_required_scanner_cycle_floor( ctx, storeapi.clone(), cycle_info, @@ -1286,8 +1386,9 @@ async fn run_data_scanner_cycle( required_cycle, &mut cycle_metrics_guard, ) - .await - { + .await; + return if persisted { + cycle_budget.mark_cycle_state_persisted(); ScannerCycleOutcome::Partial } else { ScannerCycleOutcome::Failed @@ -1345,7 +1446,7 @@ async fn run_data_scanner_cycle( scan_cycle_partial_reason(budget_reason), scan_cycle_partial_source(budget_reason), ); - return if finalize_partial_scan_cycle( + let persisted = finalize_partial_scan_cycle( ctx, storeapi.clone(), cycle_info, @@ -1353,8 +1454,9 @@ async fn run_data_scanner_cycle( leader_epoch, &mut cycle_metrics_guard, ) - .await - { + .await; + return if persisted { + cycle_budget.mark_cycle_state_persisted(); ScannerCycleOutcome::Partial } else { ScannerCycleOutcome::Failed @@ -1429,7 +1531,7 @@ async fn run_data_scanner_cycle( ); } emit_scan_cycle_partial_with_source(cycle_start.elapsed(), ScanCyclePartialReason::Unknown, None); - return if finalize_partial_scan_cycle( + let persisted = finalize_partial_scan_cycle( ctx, storeapi.clone(), cycle_info, @@ -1437,8 +1539,9 @@ async fn run_data_scanner_cycle( leader_epoch, &mut cycle_metrics_guard, ) - .await - { + .await; + return if persisted { + cycle_budget.mark_cycle_state_persisted(); ScannerCycleOutcome::Partial } else { ScannerCycleOutcome::Failed @@ -1479,6 +1582,7 @@ async fn run_data_scanner_cycle( ) .await { + cycle_budget.mark_cycle_state_persisted(); emit_scan_cycle_superseded(cycle_start.elapsed()); return ScannerCycleOutcome::Superseded; } @@ -1511,6 +1615,7 @@ async fn run_data_scanner_cycle( emit_scan_cycle_complete(false, cycle_start.elapsed()); return ScannerCycleOutcome::Failed; } + cycle_budget.mark_cycle_state_persisted(); done_cycle(); emit_scan_cycle_complete(true, cycle_start.elapsed()); @@ -1575,7 +1680,7 @@ async fn run_data_scanner_with_maintenance_state( ) -> Result<(), ScannerError> { reset_scanner_cycle_schedule(); // Acquire leader lock (write lock) to ensure only one scanner runs - let guard = match storeapi.new_ns_lock(RUSTFS_META_BUCKET, "leader.lock").await { + let mut guard = match storeapi.new_ns_lock(RUSTFS_META_BUCKET, "leader.lock").await { Ok(ns_lock) => match ns_lock.get_write_lock_quiet(get_lock_acquire_timeout()).await { Ok(guard) => { record_scanner_leader_lock_state("acquired"); @@ -1740,13 +1845,49 @@ async fn run_data_scanner_with_maintenance_state( return Ok(()); } let cycle_ctx = ctx.child_token(); - let initial_outcome = await_scanner_cycle_with_lock_fence( + let cycle_budget = ScannerCycleBudget::new_with_runtime_progress_tracking(&cycle_ctx, scanner_cycle_budget_config()); + let initial_outcome = match await_scanner_cycle_with_budget_fence( &cycle_ctx, - run_data_scanner_cycle(&cycle_ctx, &storeapi, &mut cycle_info, &mut cycle_revision, leader_epoch), + &cycle_budget, + run_data_scanner_cycle_with_budget( + &cycle_ctx, + &storeapi, + &mut cycle_info, + &mut cycle_revision, + leader_epoch, + cycle_budget.clone(), + ), guard.lock_lost_notified(), ) .await - .unwrap_or(ScannerCycleOutcome::Failed); + { + ScannerCycleWaitOutcome::Completed(outcome) => outcome, + ScannerCycleWaitOutcome::LockLost => { + record_scanner_leader_lock_lost("Scanner leader lock lost during the initial cycle").await; + global_metrics().set_cycle(None).await; + return Ok(()); + } + ScannerCycleWaitOutcome::Cancelled => { + global_metrics().set_cycle(None).await; + return Ok(()); + } + ScannerCycleWaitOutcome::Deadline { worker_stopped } => { + handle_scanner_cycle_deadline( + &ctx, + storeapi.clone(), + ScannerCycleDeadlineState { + cycle_info: &mut cycle_info, + cycle_revision: &mut cycle_revision, + leader_epoch: &mut leader_epoch, + cycle_budget: &cycle_budget, + }, + worker_stopped, + &mut guard, + ) + .await; + return Ok(()); + } + }; superseded_backoff.record_retryable_cycle(initial_outcome == ScannerCycleOutcome::Superseded); deferred_backoff.record_retryable_cycle(matches!(initial_outcome, ScannerCycleOutcome::Deferred(_))); dirty_usage_generation_seen = dirty_generation_before_cycle; @@ -1952,13 +2093,49 @@ async fn run_data_scanner_with_maintenance_state( } let dirty_generation_before_cycle = dirty_usage_generation(); let cycle_ctx = ctx.child_token(); - let outcome = await_scanner_cycle_with_lock_fence( + let cycle_budget = ScannerCycleBudget::new_with_runtime_progress_tracking(&cycle_ctx, scanner_cycle_budget_config()); + let outcome = match await_scanner_cycle_with_budget_fence( &cycle_ctx, - run_data_scanner_cycle(&cycle_ctx, &storeapi, &mut cycle_info, &mut cycle_revision, leader_epoch), + &cycle_budget, + run_data_scanner_cycle_with_budget( + &cycle_ctx, + &storeapi, + &mut cycle_info, + &mut cycle_revision, + leader_epoch, + cycle_budget.clone(), + ), guard.lock_lost_notified(), ) .await - .unwrap_or(ScannerCycleOutcome::Failed); + { + ScannerCycleWaitOutcome::Completed(outcome) => outcome, + ScannerCycleWaitOutcome::LockLost => { + record_scanner_leader_lock_lost("Scanner leader lock lost during a scanner cycle").await; + global_metrics().set_cycle(None).await; + return Ok(()); + } + ScannerCycleWaitOutcome::Cancelled => { + global_metrics().set_cycle(None).await; + return Ok(()); + } + ScannerCycleWaitOutcome::Deadline { worker_stopped } => { + handle_scanner_cycle_deadline( + &ctx, + storeapi.clone(), + ScannerCycleDeadlineState { + cycle_info: &mut cycle_info, + cycle_revision: &mut cycle_revision, + leader_epoch: &mut leader_epoch, + cycle_budget: &cycle_budget, + }, + worker_stopped, + &mut guard, + ) + .await; + return Ok(()); + } + }; superseded_backoff.record_retryable_cycle(outcome == ScannerCycleOutcome::Superseded); deferred_backoff.record_retryable_cycle(matches!(outcome, ScannerCycleOutcome::Deferred(_))); dirty_usage_generation_seen = dirty_generation_before_cycle; diff --git a/crates/scanner/src/scanner/cycle_state.rs b/crates/scanner/src/scanner/cycle_state.rs index 32c893fdf..6459ae6e5 100644 --- a/crates/scanner/src/scanner/cycle_state.rs +++ b/crates/scanner/src/scanner/cycle_state.rs @@ -1581,3 +1581,63 @@ where output = &mut cycle => Some(output), } } + +#[derive(Debug, PartialEq, Eq)] +pub(super) enum ScannerCycleWaitOutcome { + Completed(T), + LockLost, + Cancelled, + Deadline { worker_stopped: bool }, +} + +pub(super) async fn await_scanner_cycle_with_budget_fence( + cycle_ctx: &CancellationToken, + budget: &ScannerCycleBudget, + cycle: Cycle, + lock_lost: LockLost, +) -> ScannerCycleWaitOutcome +where + Cycle: Future, + LockLost: Future, +{ + tokio::pin!(cycle); + tokio::pin!(lock_lost); + let deadline = async { + if let Some(deadline) = budget.deadline() { + tokio::time::sleep_until(deadline).await; + } else { + std::future::pending::<()>().await; + } + }; + tokio::pin!(deadline); + tokio::select! { + biased; + _ = &mut lock_lost => { + cycle_ctx.cancel(); + let _ = tokio::time::timeout(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, &mut cycle).await; + ScannerCycleWaitOutcome::LockLost + } + _ = &mut deadline => { + budget.cancel_for_runtime(); + // Let the budget cancellation reach the scanner first so it can + // persist a partial cursor. Only an uncooperative worker gets the + // parent cancellation, and it is dropped after the bounded window; + // the caller fences its epoch next. + let worker_stopped = if tokio::time::timeout(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, &mut cycle) + .await + .is_ok() + { + true + } else { + cycle_ctx.cancel(); + false + }; + ScannerCycleWaitOutcome::Deadline { worker_stopped } + } + _ = cycle_ctx.cancelled() => { + let _ = tokio::time::timeout(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, &mut cycle).await; + ScannerCycleWaitOutcome::Cancelled + } + output = &mut cycle => ScannerCycleWaitOutcome::Completed(output), + } +} diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 5430e9d29..d73d63cde 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -26,6 +26,7 @@ use std::task::Poll; use temp_env::{with_var, with_var_unset}; use tokio::io::AsyncReadExt; use tokio::sync::Mutex; +use tokio::time::{Duration, advance}; const TEST_DEFAULT_SCANNER_CYCLE_SECS: u64 = 24 * 60 * 60; @@ -118,6 +119,178 @@ async fn scanner_cycle_lock_fence_bounds_uncooperative_shutdown() { assert!(cycle_ctx.is_cancelled()); } +#[tokio::test(start_paused = true)] +async fn cycle_budget_fences_late_writer_after_timeout() { + let cycle_ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new( + &cycle_ctx, + ScannerCycleBudgetConfig { + max_duration: Some(Duration::from_secs(5)), + ..Default::default() + }, + ); + let outcome = { + let cycle = std::future::pending::<()>(); + let lock_lost = std::future::pending::<()>(); + let waiter = await_scanner_cycle_with_budget_fence(&cycle_ctx, &budget, cycle, lock_lost); + tokio::pin!(waiter); + tokio::task::yield_now().await; + advance(Duration::from_secs(5)).await; + tokio::task::yield_now().await; + advance(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT).await; + waiter.await + }; + assert_eq!(outcome, ScannerCycleWaitOutcome::Deadline { worker_stopped: false }); + assert!(cycle_ctx.is_cancelled()); + assert_eq!(budget.reason(), Some(ScannerCycleBudgetReason::Runtime)); + + // A newer leadership epoch is the durable fence that rejects a late + // writer after the timed-out future has been dropped. + let store = Arc::new(MemoryConfigStore::default()); + let mut revision = DataUsageCacheRevision::Missing; + let mut cycle = CurrentCycle { + current: 0, + next: 12, + ..Default::default() + }; + let persist_ctx = CancellationToken::new(); + assert!(persist_scanner_cycle_state(&persist_ctx, store.clone(), &mut cycle, &mut revision, 1).await); + let newer = encode_scanner_cycle_state(&cycle, 2).expect("new epoch fence should encode"); + let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()); + store.interleaving_puts.lock().await.insert(key, (2, newer)); + let mut late_cycle = CurrentCycle { next: 13, ..cycle }; + assert!(!persist_scanner_cycle_state(&persist_ctx, store, &mut late_cycle, &mut revision, 1).await); +} + +#[tokio::test(start_paused = true)] +async fn cycle_budget_parent_cancellation_is_not_reported_as_timeout() { + let cycle_ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new( + &cycle_ctx, + ScannerCycleBudgetConfig { + max_duration: Some(Duration::from_secs(5)), + ..Default::default() + }, + ); + let waiter = await_scanner_cycle_with_budget_fence(&cycle_ctx, &budget, std::future::pending::<()>(), std::future::pending()); + tokio::pin!(waiter); + tokio::task::yield_now().await; + cycle_ctx.cancel(); + tokio::task::yield_now().await; + advance(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT).await; + assert_eq!(waiter.await, ScannerCycleWaitOutcome::Cancelled); +} + +#[tokio::test(start_paused = true)] +async fn cycle_budget_deadline_wins_same_tick_as_parent_cancellation() { + let cycle_ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new( + &cycle_ctx, + ScannerCycleBudgetConfig { + max_duration: Some(Duration::from_secs(5)), + ..Default::default() + }, + ); + let waiter = await_scanner_cycle_with_budget_fence(&cycle_ctx, &budget, std::future::pending::<()>(), std::future::pending()); + tokio::pin!(waiter); + tokio::task::yield_now().await; + advance(Duration::from_secs(5)).await; + cycle_ctx.cancel(); + tokio::task::yield_now().await; + advance(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT).await; + + assert_eq!(waiter.await, ScannerCycleWaitOutcome::Deadline { worker_stopped: false }); + assert_eq!(budget.reason(), Some(ScannerCycleBudgetReason::Runtime)); +} + +#[tokio::test] +async fn cycle_budget_persist_cursor_failure_is_recovery_required() { + let store = Arc::new(MemoryConfigStore::default()); + let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()); + store.fail_put_number.lock().await.insert(key, 1); + + let ctx = CancellationToken::new(); + let mut revision = DataUsageCacheRevision::Missing; + let mut cycle = CurrentCycle { + current: 12, + next: 12, + ..Default::default() + }; + let mut leader_epoch = 1; + let fenced = fence_scanner_epoch_after_cycle_timeout( + &ctx, + store, + &mut cycle, + &mut revision, + &mut leader_epoch, + std::future::pending(), + ) + .await; + assert!(!fenced, "a failed cursor/generation write must require recovery"); + let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()); + assert!(cycle_timeout_requires_recovery(true, budget.cycle_state_persisted(), fenced)); + + let metrics = Metrics::new(); + metrics.record_scanner_cycle_timeout(!fenced, Duration::from_secs(17)); + let report = metrics.report().await; + assert_eq!(report.cycle_timeout_total, 1); + assert_eq!(report.cycle_recovery_required_total, 1); + assert_eq!(report.cycle_last_progress_age, 17); + assert!(report.leader_lease_without_progress); +} + +#[tokio::test] +async fn cycle_budget_deadline_handler_fences_and_releases_guard() { + let (_temp_dir, store) = setup_scanner_cycle_store().await; + let lock = store + .new_ns_lock(RUSTFS_META_BUCKET, "leader.lock") + .await + .expect("scanner leader lock should be created"); + let mut guard = lock + .get_write_lock(Duration::from_secs(1)) + .await + .expect("scanner leader lock should be acquired"); + + let ctx = CancellationToken::new(); + let mut cycle_info = CurrentCycle { + current: 12, + next: 12, + ..Default::default() + }; + let mut cycle_revision = DataUsageCacheRevision::Missing; + let mut leader_epoch = 1; + let budget = ScannerCycleBudget::new( + &ctx, + ScannerCycleBudgetConfig { + max_duration: Some(Duration::from_secs(60)), + ..Default::default() + }, + ); + budget.mark_cycle_state_persisted(); + + handle_scanner_cycle_deadline( + &ctx, + store.clone(), + ScannerCycleDeadlineState { + cycle_info: &mut cycle_info, + cycle_revision: &mut cycle_revision, + leader_epoch: &mut leader_epoch, + cycle_budget: &budget, + }, + true, + &mut guard, + ) + .await; + + assert!(guard.is_released()); + let persisted = read_config(store, &DATA_USAGE_BLOOM_NAME_PATH) + .await + .expect("deadline handler should persist a fenced cursor"); + let (_, persisted_epoch) = decode_scanner_cycle_state(&persisted).expect("fenced cursor should decode"); + assert_eq!(persisted_epoch, 2); + global_metrics().set_cycle(None).await; +} + #[tokio::test] async fn scanner_cycle_recovery_wake_survives_wait_registration_race() { notify_scanner_cycle_recovery_wake(); @@ -428,13 +601,6 @@ fn test_scanner_cycle_max_duration_uses_env() { }); } -#[test] -fn test_scanner_cycle_max_duration_default_is_disabled() { - with_var_unset(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, || { - assert_eq!(scanner_cycle_max_duration(), None); - }); -} - #[tokio::test] async fn test_scanner_cycle_budget_cancels_after_duration() { let parent = CancellationToken::new(); @@ -2242,7 +2408,7 @@ async fn test_leadership_claim_usage_fence_rejects_old_inflight_writer() { } #[tokio::test] -async fn test_successful_old_epoch_commit_is_fenced_after_cancellation() { +async fn cycle_budget_lease_takeover_rejects_old_generation() { let store = Arc::new(MemoryConfigStore::default()); let ctx = CancellationToken::new(); let mut revision = DataUsageCacheRevision::Missing; @@ -2287,12 +2453,17 @@ async fn test_successful_old_epoch_commit_is_fenced_after_cancellation() { .await ); - let state = read_config(store, &DATA_USAGE_BLOOM_NAME_PATH) + let state = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH) .await .expect("replacement leadership claim should persist"); let (claimed_cycle, claimed_epoch) = decode_scanner_cycle_state(&state).expect("replacement cycle state should decode"); assert_eq!(claimed_cycle.next, 14); assert_eq!(claimed_epoch, 2); + + let mut stale_cycle = CurrentCycle { next: 15, ..cycle }; + let mut stale_revision = DataUsageCacheRevision::Etag("memory-2".to_string()); + let stale_ctx = CancellationToken::new(); + assert!(!persist_scanner_cycle_state(&stale_ctx, store, &mut stale_cycle, &mut stale_revision, 1,).await); } #[tokio::test] diff --git a/crates/scanner/src/scanner_budget.rs b/crates/scanner/src/scanner_budget.rs index 43ce732d5..65743439f 100644 --- a/crates/scanner/src/scanner_budget.rs +++ b/crates/scanner/src/scanner_budget.rs @@ -14,17 +14,16 @@ use std::sync::{ Arc, - atomic::{AtomicU8, AtomicU64, Ordering}, + atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering}, }; -use std::time::Instant; - -use tokio::time::Duration; +use tokio::time::{Duration, Instant}; use tokio_util::sync::CancellationToken; const BUDGET_REASON_NONE: u8 = 0; const BUDGET_REASON_RUNTIME: u8 = 1; const BUDGET_REASON_OBJECTS: u8 = 2; const BUDGET_REASON_DIRECTORIES: u8 = 3; +const PROGRESS_CLOCK_SAMPLE_INTERVAL: u64 = 128; #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub(crate) struct ScannerCycleBudgetConfig { @@ -63,29 +62,51 @@ pub struct ScannerCycleBudget { token: CancellationToken, reason: Arc, started_at: Instant, + deadline: Option, max_duration: Option, max_objects: Option, max_directories: Option, track_progress: bool, + track_unbounded_counts: bool, objects_scanned: AtomicU64, directories_started: AtomicU64, entries_visited: AtomicU64, + last_progress_millis: AtomicU64, + cycle_state_persisted: AtomicBool, } impl ScannerCycleBudget { + #[cfg(test)] pub(crate) fn new(parent: &CancellationToken, config: ScannerCycleBudgetConfig) -> Arc { - Self::new_inner(parent, config, false) + Self::new_inner(parent, config, false, false) } pub(crate) fn new_with_progress_tracking(parent: &CancellationToken, config: ScannerCycleBudgetConfig) -> Arc { - Self::new_inner(parent, config, true) + Self::new_inner(parent, config, true, true) } - fn new_inner(parent: &CancellationToken, config: ScannerCycleBudgetConfig, track_progress: bool) -> Arc { + pub(crate) fn new_with_runtime_progress_tracking(parent: &CancellationToken, config: ScannerCycleBudgetConfig) -> Arc { + let track_progress = config.max_duration.is_some(); + Self::new_inner(parent, config, track_progress, false) + } + + fn new_inner( + parent: &CancellationToken, + config: ScannerCycleBudgetConfig, + track_progress: bool, + track_unbounded_counts: bool, + ) -> Arc { let token = parent.child_token(); let reason = Arc::new(AtomicU8::new(BUDGET_REASON_NONE)); + let started_at = Instant::now(); + let deadline = config.max_duration.map(|duration| match started_at.checked_add(duration) { + Some(deadline) => deadline, + // Runtime config rejects this range, but keep programmatic callers + // fail-closed instead of panicking or silently disabling the wall clock. + None => started_at, + }); - if let Some(duration) = config.max_duration { + if let Some(deadline) = deadline { let parent = parent.clone(); let token_wait = token.clone(); let token_cancel = token.clone(); @@ -94,7 +115,7 @@ impl ScannerCycleBudget { tokio::select! { _ = parent.cancelled() => {} _ = token_wait.cancelled() => {} - _ = tokio::time::sleep(duration) => { + _ = tokio::time::sleep_until(deadline) => { Self::cancel_for_reason(&reason, &token_cancel, ScannerCycleBudgetReason::Runtime); } } @@ -104,14 +125,18 @@ impl ScannerCycleBudget { Arc::new(Self { token, reason, - started_at: Instant::now(), + started_at, + deadline, max_duration: config.max_duration, max_objects: config.max_objects, max_directories: config.max_directories, track_progress, + track_unbounded_counts, objects_scanned: AtomicU64::new(0), directories_started: AtomicU64::new(0), entries_visited: AtomicU64::new(0), + last_progress_millis: AtomicU64::new(0), + cycle_state_persisted: AtomicBool::new(false), }) } @@ -131,6 +156,14 @@ impl ScannerCycleBudget { self.max_duration } + pub(crate) fn deadline(&self) -> Option { + self.deadline + } + + pub(crate) fn cancel_for_runtime(&self) { + self.cancel_for(ScannerCycleBudgetReason::Runtime); + } + pub(crate) fn max_objects(&self) -> Option { self.max_objects } @@ -173,15 +206,43 @@ impl ScannerCycleBudget { self.entries_visited.load(Ordering::Relaxed) } + pub(crate) fn mark_cycle_state_persisted(&self) { + self.cycle_state_persisted.store(true, Ordering::Release); + } + + pub(crate) fn cycle_state_persisted(&self) -> bool { + self.cycle_state_persisted.load(Ordering::Acquire) + } + + pub(crate) fn progress_age(&self) -> Duration { + let elapsed_millis = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX); + let last_progress = self.last_progress_millis.load(Ordering::Relaxed); + Duration::from_millis(elapsed_millis.saturating_sub(last_progress)) + } + + fn record_progress_sample(&self, event: u64) { + // Clock reads are sampled at batch/count boundaries; the scanner's + // per-object path does not add a second progress atomic. + if event == 0 || (event != 1 && !event.is_multiple_of(PROGRESS_CLOCK_SAMPLE_INTERVAL)) { + return; + } + let elapsed_millis = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX); + self.last_progress_millis.store(elapsed_millis, Ordering::Relaxed); + } + pub(crate) fn record_entries_visited(&self, entries_visited: u64) { if self.track_progress { - saturating_fetch_add(&self.entries_visited, entries_visited); + let entries = saturating_fetch_add(&self.entries_visited, entries_visited); + self.record_progress_sample(entries); } } pub(crate) fn record_remote_progress(&self, objects_scanned: u64, directories_started: u64) { if self.track_progress || self.max_objects.is_some() { let objects = saturating_fetch_add(&self.objects_scanned, objects_scanned); + if self.track_progress { + self.record_progress_sample(objects); + } if self.max_objects.is_some_and(|max_objects| objects >= max_objects) { self.cancel_for(ScannerCycleBudgetReason::Objects); } @@ -189,9 +250,12 @@ impl ScannerCycleBudget { if self.track_progress || self.max_directories.is_some() { let directories = saturating_fetch_add(&self.directories_started, directories_started); + if self.track_progress { + self.record_progress_sample(directories); + } if self .max_directories - .is_some_and(|max_directories| directories > max_directories) + .is_some_and(|max_directories| directory_budget_exhausted(directories, max_directories)) { self.cancel_for(ScannerCycleBudgetReason::Directories); } @@ -207,14 +271,17 @@ impl ScannerCycleBudget { } pub(crate) fn try_start_directory(&self) -> bool { - if !self.track_progress && self.max_directories.is_none() { + if self.max_directories.is_none() && !self.track_unbounded_counts { return true; } let directories = saturating_fetch_add(&self.directories_started, 1); + if self.track_progress { + self.record_progress_sample(directories); + } if self .max_directories - .is_some_and(|max_directories| directories > max_directories) + .is_some_and(|max_directories| directory_budget_exhausted(directories, max_directories)) { self.cancel_for(ScannerCycleBudgetReason::Directories); return false; @@ -224,11 +291,14 @@ impl ScannerCycleBudget { } pub(crate) fn record_object_scanned(&self) { - if !self.track_progress && self.max_objects.is_none() { + if self.max_objects.is_none() && !self.track_unbounded_counts { return; } let objects = saturating_fetch_add(&self.objects_scanned, 1); + if self.track_progress { + self.record_progress_sample(objects); + } if self.max_objects.is_some_and(|max_objects| objects >= max_objects) { self.cancel_for(ScannerCycleBudgetReason::Objects); } @@ -259,6 +329,13 @@ fn saturating_fetch_add(value: &AtomicU64, delta: u64) -> u64 { } } +fn directory_budget_exhausted(directories: u64, max_directories: u64) -> bool { + // Saturation hides a remote max+1 update when the configured limit is the + // largest representable counter. Treat that boundary as exhausted rather + // than allowing work to continue indefinitely. + directories > max_directories || (directories == u64::MAX && max_directories == u64::MAX) +} + impl Drop for ScannerCycleBudget { fn drop(&mut self) { self.token.cancel(); @@ -401,6 +478,35 @@ mod tests { assert_eq!(directory_budget.reason(), Some(ScannerCycleBudgetReason::Directories)); } + #[test] + fn directory_budget_fails_closed_when_progress_saturates() { + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new( + &parent, + ScannerCycleBudgetConfig { + max_directories: Some(u64::MAX), + ..Default::default() + }, + ); + + budget.record_remote_progress(0, u64::MAX); + + assert_eq!(budget.reason(), Some(ScannerCycleBudgetReason::Directories)); + assert!(budget.token().is_cancelled()); + + let local_budget = ScannerCycleBudget::new( + &parent, + ScannerCycleBudgetConfig { + max_directories: Some(u64::MAX), + ..Default::default() + }, + ); + local_budget.record_remote_progress(0, u64::MAX - 1); + assert!(!local_budget.budget_elapsed()); + assert!(!local_budget.try_start_directory()); + assert_eq!(local_budget.reason(), Some(ScannerCycleBudgetReason::Directories)); + } + #[test] fn explicit_progress_tracking_counts_unbounded_remote_work_without_cancelling() { let parent = CancellationToken::new(); @@ -461,4 +567,29 @@ mod tests { assert!(object_limited.requires_serial_progress_accounting()); assert!(directory_limited.requires_serial_progress_accounting()); } + + #[tokio::test(start_paused = true)] + async fn progress_age_uses_virtual_time_and_sampled_progress() { + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new_with_runtime_progress_tracking( + &parent, + ScannerCycleBudgetConfig { + max_duration: Some(Duration::from_secs(60)), + ..Default::default() + }, + ); + + tokio::time::advance(Duration::from_secs(5)).await; + assert_eq!(budget.progress_age(), Duration::from_secs(5)); + budget.record_entries_visited(1); + assert_eq!(budget.progress_age(), Duration::ZERO); + + tokio::time::advance(Duration::from_secs(2)).await; + for _ in 0..126 { + budget.record_entries_visited(1); + } + assert_eq!(budget.progress_age(), Duration::from_secs(2)); + budget.record_entries_visited(1); + assert_eq!(budget.progress_age(), Duration::ZERO); + } } diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 677719a69..c104fb263 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -48,6 +48,7 @@ use time::OffsetDateTime; use tokio::sync::{Mutex, Notify, Semaphore, mpsc}; use tokio::time::Duration; use tokio_util::sync::CancellationToken; +use tokio_util::task::AbortOnDropHandle; use tracing::{debug, error, warn}; use crate::ScannerObjectInfo as ObjectInfo; diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index 3151b4c0c..aa0339e4b 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -314,7 +314,7 @@ impl ScannerIOCache for SetDisks { let ctx_clone = ctx.clone(); let completed_bucket_count = Arc::new(AtomicUsize::new(0)); let completed_bucket_count_clone = completed_bucket_count.clone(); - let collect_bucket_results_fut = tokio::spawn(async move { + let collect_bucket_results_fut = AbortOnDropHandle::new(tokio::spawn(async move { let mut cancelled = false; loop { @@ -333,7 +333,7 @@ impl ScannerIOCache for SetDisks { } } } - }); + })); let mut futs = Vec::new(); @@ -365,7 +365,7 @@ impl ScannerIOCache for SetDisks { NamespaceScannerWorkerMode::RemoteV4(server_epoch) => Some(server_epoch), NamespaceScannerWorkerMode::Coordinator => None, }; - futs.push(tokio::spawn(async move { + futs.push(AbortOnDropHandle::new(tokio::spawn(async move { let remote_session_id = uuid::Uuid::new_v4(); let mut remote_session_sequence = 0_u64; loop { @@ -1038,7 +1038,7 @@ impl ScannerIOCache for SetDisks { ); } } - })); + }))); } drop(bucket_tx); drop(bucket_result_tx); diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index c2a8eb253..64763655f 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -242,7 +242,7 @@ impl ScannerIOCycle for ECStore { results[results_index_clone] = result; } }); - wait_futs.push(receiver_fut); + wait_futs.push(AbortOnDropHandle::new(receiver_fut)); let scan_plan = ScannerBucketScanPlan { buckets: set_buckets, @@ -318,7 +318,7 @@ impl ScannerIOCycle for ECStore { record_set_scan_failure(&mut first_err, e); } }); - wait_futs.push(scanner_fut); + wait_futs.push(AbortOnDropHandle::new(scanner_fut)); } } diff --git a/crates/utils/src/envs.rs b/crates/utils/src/envs.rs index 2d039dec5..5e572e5c0 100644 --- a/crates/utils/src/envs.rs +++ b/crates/utils/src/envs.rs @@ -268,7 +268,7 @@ where .parse::() .map_err(|_| { log_once(&format!("env_invalid_value:{used_key}"), || { - format!("Invalid {} value for {used_key}: {value}. Treating as unset.", type_name::()) + format!("Invalid {} value for {used_key}. Treating as unset.", type_name::()) }); }) .ok() @@ -570,7 +570,7 @@ where Ok(parsed) => EnvParseOutcome::Parsed(parsed), Err(_) => { log_once(&format!("env_invalid_value:{used_key}"), || { - format!("Invalid {} value for {used_key}: {value}. Treating as unset.", type_name::()) + format!("Invalid {} value for {used_key}. Treating as unset.", type_name::()) }); EnvParseOutcome::Invalid } diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md index 33206d699..de54b8f40 100644 --- a/docs/operations/scanner-runtime-controls.md +++ b/docs/operations/scanner-runtime-controls.md @@ -52,7 +52,7 @@ The `/v3/scanner/status` response reports each effective runtime value with a | `scanner.max_wait` | `RUSTFS_SCANNER_MAX_WAIT_SECS` | seconds | preset-derived | Caps one scanner sleep. | | `scanner.cycle` | `RUSTFS_SCANNER_CYCLE` | seconds | preset-derived | Sets the interval between scanner cycles. | | `scanner.start_delay` | `RUSTFS_SCANNER_START_DELAY_SECS` | seconds | unset | Sets startup delay and, for compatibility, the cycle interval when `scanner.cycle` is unset. | -| `scanner.cycle_max_duration` | `RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS` | seconds | `0` | Caps one cycle's runtime. `0` disables this budget. | +| `scanner.cycle_max_duration` | `RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS` | seconds | `1800` | Caps one cycle's runtime. An explicit `0` disables this budget. | | `scanner.cycle_max_objects` | `RUSTFS_SCANNER_CYCLE_MAX_OBJECTS` | objects | `0` | Caps objects processed by one cycle. `0` disables this budget. | | `scanner.cycle_max_directories` | `RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES` | directories | `0` | Caps directories entered by one cycle. `0` disables this budget. | | `heal.bitrot_cycle` | `RUSTFS_SCANNER_BITROT_CYCLE_SECS` | seconds | `2592000` | Controls periodic deep bitrot scans. `false`, `off`, `no`, or `disabled` disables periodic deep scans; `0`, `true`, `on`, or `yes` runs deep mode every scanner cycle. | @@ -70,6 +70,21 @@ sleep multiplier, maximum wait, and cycle interval. Use `scanner.delay`, `scanner.max_wait`, and `scanner.cycle` when the preset is close but one axis needs a precise override. +When the cycle duration control is unset, RustFS uses a finite 1800-second +(30-minute) default, matching the scanner benchmark guidance. An explicit `0` +preserves the compatibility behavior of an unbounded cycle; object and +directory budgets likewise remain unbounded when explicitly set to `0`. Invalid +or overflowing duration environment values are configuration errors rather than +silent fallback values. + +When a finite deadline expires, RustFS cancels cooperative scanner work and +waits only for the existing bounded shutdown window. A non-yielding I/O future +is dropped after that window. RustFS then attempts a higher leadership epoch so +late cycle, usage, cache, and remote writes from the old generation fail closed. +If the worker cannot stop cooperatively, the cycle state was not confirmed +durable, or that epoch fence cannot be durably persisted, the scanner reports +`recovery-required`; it does not claim an uncooperative cursor was saved. + An explicit `scanner.cycle` or `RUSTFS_SCANNER_CYCLE` is a minimum inter-cycle cadence: dirty-usage notifications do not bypass that configured interval. The default adaptive policy continues to use dirty-usage notifications to wake @@ -144,6 +159,10 @@ metrics.maintenance_control.primary_control metrics.source_work metrics.replication_repair metrics.scan_checkpoint +metrics.cycle_timeout_total +metrics.cycle_last_progress_age +metrics.leader_lease_without_progress +metrics.cycle_recovery_required_total ``` ## Reading Pacing Pressure