From 46e200c22b99faa57db7e7066018207f2d535e1a Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Thu, 18 Jun 2026 11:03:23 +0800 Subject: [PATCH] feat(scanner): prioritize dirty usage refresh work (#3538) --- crates/scanner/src/lib.rs | 1 + crates/scanner/src/scanner_io.rs | 227 ++++++++++++++++++++++++---- rustfs/src/app/bucket_usecase.rs | 2 + rustfs/src/app/multipart_usecase.rs | 1 + rustfs/src/app/object_usecase.rs | 18 ++- 5 files changed, 214 insertions(+), 35 deletions(-) diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 04e83426f..614118f25 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -34,6 +34,7 @@ pub use error::ScannerError; pub use runtime_config::{apply_scanner_runtime_config, scanner_runtime_config_status, validate_scanner_runtime_config}; pub use rustfs_common::last_minute; pub use scanner::init_data_scanner; +pub use scanner_io::{clear_dirty_usage_bucket, record_dirty_usage_bucket}; pub use sleeper::{DynamicSleeper, SCANNER_IDLE_MODE, SCANNER_SLEEPER}; use std::sync::atomic::{AtomicU64, Ordering}; diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index a020cc9b0..0a530223f 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -49,7 +49,8 @@ use rustfs_utils::path::path_join_buf; use s3s::dto::{BucketLifecycleConfiguration, ReplicationConfiguration}; use std::collections::{HashMap, HashSet}; use std::path::Path; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::{LazyLock, Mutex as StdMutex, MutexGuard}; use std::time::{Instant, SystemTime}; use std::{fmt::Debug, sync::Arc}; use time::OffsetDateTime; @@ -75,6 +76,110 @@ const METRIC_SCANNER_SET_SCANS_QUEUED: &str = "rustfs_scanner_set_scans_queued"; const METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE: &str = "rustfs_scanner_disk_bucket_scans_active"; const METRIC_SCANNER_DISK_BUCKET_SCANS_QUEUED: &str = "rustfs_scanner_disk_bucket_scans_queued"; +pub type DirtyUsageBuckets = HashMap; + +#[derive(Clone)] +pub struct ScannerBucketScanPlan { + buckets: Vec, + dirty_usage_buckets: Arc, +} + +impl ScannerBucketScanPlan { + fn new(buckets: Vec, dirty_usage_buckets: Arc) -> Self { + Self { + buckets, + dirty_usage_buckets, + } + } +} + +static DIRTY_USAGE_BUCKET_GENERATION: AtomicU64 = AtomicU64::new(0); +static DIRTY_USAGE_BUCKETS: LazyLock> = LazyLock::new(|| StdMutex::new(HashMap::new())); + +fn dirty_usage_buckets() -> MutexGuard<'static, DirtyUsageBuckets> { + DIRTY_USAGE_BUCKETS.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +pub fn record_dirty_usage_bucket(bucket: &str) { + if bucket.is_empty() { + return; + } + + let generation = DIRTY_USAGE_BUCKET_GENERATION.fetch_add(1, Ordering::AcqRel) + 1; + dirty_usage_buckets().insert(bucket.to_string(), generation); +} + +pub fn clear_dirty_usage_bucket(bucket: &str) { + if bucket.is_empty() { + return; + } + + dirty_usage_buckets().remove(bucket); +} + +fn snapshot_dirty_usage_buckets(buckets: &[BucketInfo]) -> DirtyUsageBuckets { + let dirty_buckets = dirty_usage_buckets(); + buckets + .iter() + .filter_map(|bucket| { + dirty_buckets + .get(&bucket.name) + .map(|generation| (bucket.name.clone(), *generation)) + }) + .collect() +} + +fn clear_dirty_usage_buckets(snapshot: &DirtyUsageBuckets) { + if snapshot.is_empty() { + return; + } + + let mut dirty_buckets = dirty_usage_buckets(); + for (bucket, generation) in snapshot { + if dirty_buckets.get(bucket).is_some_and(|current| current == generation) { + dirty_buckets.remove(bucket); + } + } +} + +#[cfg(test)] +fn dirty_usage_bucket_count() -> usize { + dirty_usage_buckets().len() +} + +#[cfg(test)] +fn clear_dirty_usage_buckets_for_tests() { + dirty_usage_buckets().clear(); +} + +fn bucket_usage_scan_order( + buckets: &[BucketInfo], + old_cache: &DataUsageCache, + dirty_buckets: &DirtyUsageBuckets, +) -> Vec { + let mut ordered = Vec::with_capacity(buckets.len()); + + for bucket in buckets { + if dirty_buckets.contains_key(&bucket.name) { + ordered.push(bucket.clone()); + } + } + + for bucket in buckets { + if !dirty_buckets.contains_key(&bucket.name) && old_cache.find(&bucket.name).is_none() { + ordered.push(bucket.clone()); + } + } + + for bucket in buckets { + if !dirty_buckets.contains_key(&bucket.name) && old_cache.find(&bucket.name).is_some() { + ordered.push(bucket.clone()); + } + } + + ordered +} + fn record_set_scan_concurrency_limit(limit: usize) { metrics::gauge!(METRIC_SCANNER_SET_SCAN_CONCURRENCY_LIMIT).set(limit as f64); global_metrics().record_scanner_set_scan_state(Some(limit), None, None); @@ -358,7 +463,7 @@ pub trait ScannerIOCache: Send + Sync + Debug + 'static { self: Arc, ctx: CancellationToken, budget: Arc, - buckets: Vec, + scan_plan: ScannerBucketScanPlan, updates: mpsc::Sender, want_cycle: u64, scan_mode: HealScanMode, @@ -435,6 +540,7 @@ impl ScannerIO for ECStore { } let set_scan_limit = scanner_max_concurrent_set_scans(total_results); + let dirty_usage_buckets = Arc::new(snapshot_dirty_usage_buckets(&all_buckets)); record_set_scan_concurrency_limit(set_scan_limit); debug!( target: "rustfs::scanner::io", @@ -489,7 +595,7 @@ impl ScannerIO for ECStore { }); wait_futs.push(receiver_fut); - let all_buckets_clone = all_buckets.clone(); + let scan_plan = ScannerBucketScanPlan::new(all_buckets.clone(), dirty_usage_buckets.clone()); // Spawn task to run the scanner let scanner_fut = tokio::spawn(async move { let permit_wait = child_token_clone.clone(); @@ -515,7 +621,7 @@ impl ScannerIO for ECStore { .nsscanner_cache( child_token_clone.clone(), budget_clone, - all_buckets_clone, + scan_plan, tx, want_cycle_clone, scan_mode_clone, @@ -644,22 +750,31 @@ impl ScannerIO for ECStore { let first_err = first_err_mutex.lock().await.take(); let results = results_mutex.lock().await.clone(); - finalize_nsscanner_result(&results, first_err) + let completed_all_sets = results.iter().all(|result| result.info.last_update.is_some()); + let result = finalize_nsscanner_result(&results, first_err); + if result.is_ok() && completed_all_sets && !budget.budget_elapsed() { + clear_dirty_usage_buckets(&dirty_usage_buckets); + } + result } } #[async_trait::async_trait] impl ScannerIOCache for SetDisks { - #[tracing::instrument(skip(self, budget, updates))] + #[tracing::instrument(skip(self, budget, scan_plan, updates))] async fn nsscanner_cache( self: Arc, ctx: CancellationToken, budget: Arc, - buckets: Vec, + scan_plan: ScannerBucketScanPlan, updates: mpsc::Sender, want_cycle: u64, scan_mode: HealScanMode, ) -> Result<()> { + let ScannerBucketScanPlan { + buckets, + dirty_usage_buckets, + } = scan_plan; let pool_label = self.pool_index.to_string(); let set_label = self.set_index.to_string(); @@ -720,12 +835,18 @@ impl ScannerIOCache for SetDisks { let mut permutes = buckets.clone(); permutes.shuffle(&mut rand::rng()); + let scan_order = bucket_usage_scan_order(&permutes, &old_cache, &dirty_usage_buckets); let mut preloaded_published_buckets = HashSet::new(); - for bucket in permutes.iter() { - if old_cache.find(&bucket.name).is_none() - && let Err(e) = bucket_tx.send(bucket.clone()).await - { + for bucket in scan_order.iter() { + if let Some(c) = old_cache.find(&bucket.name) { + cache.replace(&bucket.name, DATA_USAGE_ROOT, c.clone()); + if old_cache.info.last_update.is_some() { + preloaded_published_buckets.insert(bucket.name.clone()); + } + } + + if let Err(e) = bucket_tx.send(bucket.clone()).await { error!( target: "rustfs::scanner::io", event = EVENT_SCANNER_SET_STATE, @@ -739,28 +860,6 @@ impl ScannerIOCache for SetDisks { } } - for bucket in permutes.iter() { - if let Some(c) = old_cache.find(&bucket.name) { - cache.replace(&bucket.name, DATA_USAGE_ROOT, c.clone()); - if old_cache.info.last_update.is_some() { - preloaded_published_buckets.insert(bucket.name.clone()); - } - - if let Err(e) = bucket_tx.send(bucket.clone()).await { - error!( - target: "rustfs::scanner::io", - event = EVENT_SCANNER_SET_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_IO, - bucket = %bucket.name, - state = "send_bucket_failed", - error = %e, - "Scanner bucket dispatch failed" - ); - } - } - } - drop(bucket_tx); let cache_mutex: Arc> = Arc::new(Mutex::new(cache)); @@ -1435,6 +1534,68 @@ mod tests { use temp_env::with_var; use uuid::Uuid; + fn bucket_info(name: &str) -> BucketInfo { + BucketInfo { + name: name.to_string(), + created: None, + deleted: None, + versioning: false, + object_locking: false, + } + } + + #[test] + #[serial] + fn dirty_usage_snapshot_clear_preserves_newer_generation() { + clear_dirty_usage_buckets_for_tests(); + record_dirty_usage_bucket("photos"); + let buckets = vec![bucket_info("photos")]; + let snapshot = snapshot_dirty_usage_buckets(&buckets); + + record_dirty_usage_bucket("photos"); + clear_dirty_usage_buckets(&snapshot); + + assert_eq!(dirty_usage_bucket_count(), 1); + clear_dirty_usage_buckets_for_tests(); + } + + #[test] + #[serial] + fn clear_dirty_usage_bucket_removes_deleted_bucket_marker() { + clear_dirty_usage_buckets_for_tests(); + record_dirty_usage_bucket("photos"); + record_dirty_usage_bucket("videos"); + + clear_dirty_usage_bucket("photos"); + + let buckets = vec![bucket_info("photos"), bucket_info("videos")]; + let snapshot = snapshot_dirty_usage_buckets(&buckets); + assert!(!snapshot.contains_key("photos")); + assert!(snapshot.contains_key("videos")); + assert_eq!(dirty_usage_bucket_count(), 1); + clear_dirty_usage_buckets_for_tests(); + } + + #[test] + fn bucket_usage_scan_order_prioritizes_dirty_buckets() { + let buckets = vec![bucket_info("missing"), bucket_info("cached"), bucket_info("dirty")]; + let mut old_cache = DataUsageCache { + info: DataUsageCacheInfo { + name: DATA_USAGE_ROOT.to_string(), + ..Default::default() + }, + ..Default::default() + }; + old_cache.replace("cached", DATA_USAGE_ROOT, DataUsageEntry::default()); + old_cache.replace("dirty", DATA_USAGE_ROOT, DataUsageEntry::default()); + + let dirty_buckets = HashMap::from([("dirty".to_string(), 1)]); + let ordered = bucket_usage_scan_order(&buckets, &old_cache, &dirty_buckets); + let names = ordered.iter().map(|bucket| bucket.name.as_str()).collect::>(); + + assert_eq!(names, vec!["dirty", "missing", "cached"]); + } + #[test] fn record_set_scan_failure_preserves_first_error() { let mut first = None; diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index e6f9c3b2e..2bc886d27 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -800,6 +800,7 @@ impl DefaultBucketUsecase { counter!("rustfs_create_bucket_total").increment(1); let result = Ok(S3Response::new(output)); let _ = helper.complete(&result); + rustfs_scanner::record_dirty_usage_bucket(&bucket); result } @@ -832,6 +833,7 @@ impl DefaultBucketUsecase { ) .await .map_err(ApiError::from)?; + rustfs_scanner::clear_dirty_usage_bucket(&input.bucket); if let Err(err) = site_replication_delete_bucket_hook(&input.bucket, force).await { warn!(bucket = %input.bucket, error = ?err, "site replication delete bucket hook failed"); diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 4bb5fb421..1b850ebc0 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -564,6 +564,7 @@ impl DefaultMultipartUsecase { let result = Ok(S3Response::new(output)); let _ = helper.complete(&result); + rustfs_scanner::record_dirty_usage_bucket(&bucket); result } diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 09eb8112e..6150844a9 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -2399,6 +2399,7 @@ impl DefaultObjectUsecase { let result = Ok(S3Response::new(output)); let _ = helper.complete(&result); + rustfs_scanner::record_dirty_usage_bucket(&bucket); // Record write operation for capacity management (inline to avoid per-request tokio::spawn overhead) let manager = get_capacity_manager(); @@ -3329,6 +3330,7 @@ impl DefaultObjectUsecase { let result = Ok(S3Response::new(output)); let _ = helper.complete(&result); + rustfs_scanner::record_dirty_usage_bucket(&bucket); result } @@ -3644,6 +3646,8 @@ impl DefaultObjectUsecase { .map(|context| context.notify()) .unwrap_or_else(default_notify_interface); let request_context = req.extensions.get::().cloned(); + let deleted_any = delete_results.iter().any(|result| result.delete_object.is_some()); + let notify_bucket = bucket.clone(); spawn_background_with_context(request_context, async move { let _activity_guard = DeleteTailActivityGuard::new(DeleteTailStage::Notify); for res in delete_results { @@ -3651,10 +3655,10 @@ impl DefaultObjectUsecase { let event_name = delete_event_name_for_marker(dobj.delete_marker); let event_args = EventArgsBuilder::new( event_name, - bucket.clone(), + notify_bucket.clone(), ObjectInfo { name: dobj.object_name.clone(), - bucket: bucket.clone(), + bucket: notify_bucket.clone(), ..Default::default() }, ) @@ -3672,6 +3676,9 @@ impl DefaultObjectUsecase { let result = Ok(S3Response::new(output)); let _ = helper.complete(&result); + if deleted_any { + rustfs_scanner::record_dirty_usage_bucket(&bucket); + } // Record write operation for capacity management (inline to avoid per-request tokio::spawn overhead) let manager = get_capacity_manager(); manager.record_write_operation().await; @@ -3840,6 +3847,7 @@ impl DefaultObjectUsecase { let manager = get_capacity_manager(); manager.record_write_operation().await; let _ = helper.complete(&result); + rustfs_scanner::record_dirty_usage_bucket(&bucket); return result; } @@ -3908,6 +3916,7 @@ impl DefaultObjectUsecase { let manager = get_capacity_manager(); manager.record_write_operation().await; let _ = helper.complete(&result); + rustfs_scanner::record_dirty_usage_bucket(&bucket); result } @@ -4623,6 +4632,7 @@ impl DefaultObjectUsecase { let host = get_request_host(&req.headers); let port = get_request_port(&req.headers); let user_agent = get_request_user_agent(&req.headers); + let mut wrote_any_entry = false; while let Some(entry) = entries.next().await { let mut f = match entry { @@ -4789,6 +4799,10 @@ impl DefaultObjectUsecase { return Err(ApiError::from(e).into()); } }; + if !wrote_any_entry { + rustfs_scanner::record_dirty_usage_bucket(&bucket); + wrote_any_entry = true; + } let _manager = get_concurrency_manager(); let _fpath_clone = fpath.clone();