feat(scanner): prioritize dirty usage refresh work (#3538)

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};
 

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<String, u64>;
+
+#[derive(Clone)]
+pub struct ScannerBucketScanPlan {
+    buckets: Vec<BucketInfo>,
+    dirty_usage_buckets: Arc<DirtyUsageBuckets>,
+}
+
+impl ScannerBucketScanPlan {
+    fn new(buckets: Vec<BucketInfo>, dirty_usage_buckets: Arc<DirtyUsageBuckets>) -> Self {
+        Self {
+            buckets,
+            dirty_usage_buckets,
+        }
+    }
+}
+
+static DIRTY_USAGE_BUCKET_GENERATION: AtomicU64 = AtomicU64::new(0);
+static DIRTY_USAGE_BUCKETS: LazyLock<StdMutex<DirtyUsageBuckets>> = 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<BucketInfo> {
+    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<Self>,
         ctx: CancellationToken,
         budget: Arc<ScannerCycleBudget>,
-        buckets: Vec<BucketInfo>,
+        scan_plan: ScannerBucketScanPlan,
         updates: mpsc::Sender<DataUsageCache>,
         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<Self>,
         ctx: CancellationToken,
         budget: Arc<ScannerCycleBudget>,
-        buckets: Vec<BucketInfo>,
+        scan_plan: ScannerBucketScanPlan,
         updates: mpsc::Sender<DataUsageCache>,
         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<Mutex<DataUsageCache>> = 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::<Vec<_>>();
+
+        assert_eq!(names, vec!["dirty", "missing", "cached"]);
+    }
+
     #[test]
     fn record_set_scan_failure_preserves_first_error() {
         let mut first = None;

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");

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
     }
 

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::<request_context::RequestContext>().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();