feat(scanner): wake cycles for dirty usage refresh (#3617)

crates/scanner/src/scanner.rs

@@ -23,7 +23,7 @@ use crate::runtime_config::{
 };
 use crate::scanner_budget::{ScannerCycleBudget, ScannerCycleBudgetConfig, ScannerCycleBudgetReason};
 use crate::scanner_folder::{data_usage_update_dir_cycles, heal_object_select_prob};
-use crate::scanner_io::ScannerIO;
+use crate::scanner_io::{ScannerIO, dirty_usage_bucket_notified, dirty_usage_buckets_pending};
 use crate::sleeper::{SCANNER_SLEEPER, set_scanner_default_speed};
 use crate::{DataUsageInfo, ScannerActivityGuard, ScannerError};
 use chrono::{DateTime, Utc};
@@ -131,6 +131,30 @@ fn randomized_cycle_delay_for(interval: Duration) -> Duration {
     delay.max(Duration::from_secs(1))
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum ScannerCycleWakeReason {
+    Timer,
+    DirtyUsage,
+    Cancelled,
+}
+
+async fn wait_for_next_scanner_cycle(ctx: &CancellationToken, delay: Duration) -> ScannerCycleWakeReason {
+    let sleep = tokio::time::sleep(delay);
+    tokio::pin!(sleep);
+
+    loop {
+        tokio::select! {
+            _ = ctx.cancelled() => return ScannerCycleWakeReason::Cancelled,
+            _ = &mut sleep => return ScannerCycleWakeReason::Timer,
+            _ = dirty_usage_bucket_notified() => {
+                if dirty_usage_buckets_pending() {
+                    return ScannerCycleWakeReason::DirtyUsage;
+                }
+            }
+        }
+    }
+}
+
 fn initial_scanner_delay_for(start_delay_secs: Option<u64>) -> Duration {
     start_delay_secs
         .map(|secs| randomized_cycle_delay_for(Duration::from_secs(secs)))
@@ -932,12 +956,22 @@ pub async fn run_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) ->
             break;
         }
 
-        // Randomized inter-cycle delay
-        tokio::select! {
-            _ = ctx.cancelled() => break,
-            _ = tokio::time::sleep(randomized_cycle_delay()) => {
+        match wait_for_next_scanner_cycle(&ctx, randomized_cycle_delay()).await {
+            ScannerCycleWakeReason::Cancelled => break,
+            ScannerCycleWakeReason::Timer => {
                 run_data_scanner_cycle(&ctx, &storeapi, &mut cycle_info).await;
-            },
+            }
+            ScannerCycleWakeReason::DirtyUsage => {
+                debug!(
+                    target: "rustfs::scanner",
+                    event = EVENT_SCANNER_CYCLE_STATE,
+                    component = LOG_COMPONENT_SCANNER,
+                    subsystem = LOG_SUBSYSTEM_RUNTIME,
+                    state = "dirty_usage_wakeup",
+                    "Scanner cycle woke for dirty usage work"
+                );
+                run_data_scanner_cycle(&ctx, &storeapi, &mut cycle_info).await;
+            }
         }
     }
 
@@ -1590,6 +1624,21 @@ mod tests {
         assert!(delay < Duration::from_secs(2), "expected delay < 2s");
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn test_wait_for_next_scanner_cycle_wakes_for_dirty_usage() {
+        crate::scanner_io::clear_dirty_usage_bucket("photos");
+        crate::scanner_io::record_dirty_usage_bucket("photos");
+
+        let ctx = CancellationToken::new();
+        let reason = tokio::time::timeout(Duration::from_secs(1), wait_for_next_scanner_cycle(&ctx, Duration::from_secs(60)))
+            .await
+            .expect("dirty usage should wake scanner before timer");
+
+        assert_eq!(reason, ScannerCycleWakeReason::DirtyUsage);
+        crate::scanner_io::clear_dirty_usage_bucket("photos");
+    }
+
     #[test]
     #[serial]
     fn test_get_cycle_scan_mode_runs_deep_until_selection_window_completes() {

crates/scanner/src/scanner_io.rs

@@ -37,7 +37,7 @@ use std::sync::{LazyLock, Mutex as StdMutex, MutexGuard};
 use std::time::{Instant, SystemTime};
 use std::{fmt::Debug, sync::Arc};
 use time::OffsetDateTime;
-use tokio::sync::{Mutex, Semaphore, mpsc};
+use tokio::sync::{Mutex, Notify, Semaphore, mpsc};
 use tokio::time::Duration;
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, warn};
@@ -86,6 +86,7 @@ impl ScannerBucketScanPlan {
 
 static DIRTY_USAGE_BUCKET_GENERATION: AtomicU64 = AtomicU64::new(0);
 static DIRTY_USAGE_BUCKETS: LazyLock<StdMutex<DirtyUsageBuckets>> = LazyLock::new(|| StdMutex::new(HashMap::new()));
+static DIRTY_USAGE_BUCKET_NOTIFY: LazyLock<Notify> = LazyLock::new(Notify::new);
 
 fn dirty_usage_buckets() -> MutexGuard<'static, DirtyUsageBuckets> {
     DIRTY_USAGE_BUCKETS.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
@@ -107,6 +108,7 @@ pub fn record_dirty_usage_bucket(bucket: &str) {
         dirty_buckets.len()
     };
     global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending_buckets));
+    DIRTY_USAGE_BUCKET_NOTIFY.notify_one();
 }
 
 pub fn clear_dirty_usage_bucket(bucket: &str) {
@@ -138,6 +140,14 @@ fn snapshot_dirty_usage_buckets(buckets: &[BucketInfo]) -> DirtyUsageBuckets {
     snapshot
 }
 
+pub(crate) fn dirty_usage_buckets_pending() -> bool {
+    !dirty_usage_buckets().is_empty()
+}
+
+pub(crate) async fn dirty_usage_bucket_notified() {
+    DIRTY_USAGE_BUCKET_NOTIFY.notified().await;
+}
+
 fn clear_dirty_usage_buckets(snapshot: &DirtyUsageBuckets) {
     let (cleared_buckets, pending_buckets) = {
         let mut dirty_buckets = dirty_usage_buckets();

rustfs/src/app/bucket_usecase.rs

@@ -1027,6 +1027,7 @@ impl DefaultBucketUsecase {
             warn!(bucket = %bucket, error = ?err, "site replication bucket lifecycle delete hook failed");
         }
 
+        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         Ok(S3Response::new(DeleteBucketLifecycleOutput::default()))
     }
 
@@ -1096,6 +1097,7 @@ impl DefaultBucketUsecase {
 
         info!(bucket = %bucket, "deleted bucket replication config");
 
+        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         Ok(S3Response::new(DeleteBucketReplicationOutput::default()))
     }
 
@@ -1115,6 +1117,7 @@ impl DefaultBucketUsecase {
             warn!(bucket = %bucket, error = ?err, "site replication bucket tagging delete hook failed");
         }
 
+        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         Ok(S3Response::new(DeleteBucketTaggingOutput {}))
     }
 
@@ -1676,6 +1679,7 @@ impl DefaultBucketUsecase {
             });
         }
 
+        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         Ok(S3Response::new(PutBucketLifecycleConfigurationOutput::default()))
     }
 
@@ -1889,6 +1893,7 @@ impl DefaultBucketUsecase {
             warn!(bucket = %bucket, error = ?err, "site replication bucket replication-config hook failed");
         }
 
+        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         Ok(S3Response::new(PutBucketReplicationOutput::default()))
     }
 
@@ -1951,6 +1956,7 @@ impl DefaultBucketUsecase {
             warn!(bucket = %bucket, error = ?err, "site replication bucket tagging hook failed");
         }
 
+        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         Ok(S3Response::new(PutBucketTaggingOutput::default()))
     }
 
@@ -1981,6 +1987,7 @@ impl DefaultBucketUsecase {
             warn!(bucket = %bucket, error = ?err, "site replication bucket versioning hook failed");
         }
 
+        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         Ok(S3Response::new(PutBucketVersioningOutput {}))
     }
 

rustfs/src/app/multipart_usecase.rs

@@ -314,7 +314,10 @@ impl DefaultMultipartUsecase {
             .abort_multipart_upload(bucket.as_str(), key.as_str(), upload_id.as_str(), opts)
             .await
         {
-            Ok(_) => Ok(S3Response::new(AbortMultipartUploadOutput { ..Default::default() })),
+            Ok(_) => {
+                rustfs_scanner::record_dirty_usage_bucket(&bucket);
+                Ok(S3Response::new(AbortMultipartUploadOutput { ..Default::default() }))
+            }
             Err(err) => {
                 // Convert MalformedUploadID to NoSuchUpload for S3 API compatibility
                 if matches!(err, StorageError::MalformedUploadID(_)) {

rustfs/src/app/object_usecase.rs

@@ -4379,6 +4379,7 @@ impl DefaultObjectUsecase {
                 )
                 .await
                 .map_err(|_| S3Error::with_message(S3ErrorCode::Custom("ErrCopyObject".into()), "restore object failed."))?;
+            rustfs_scanner::record_dirty_usage_bucket(&bucket);
 
             if already_restored {
                 let output = RestoreObjectOutput {
@@ -4435,6 +4436,7 @@ impl DefaultObjectUsecase {
                     err.to_string()
                 );
             } else {
+                rustfs_scanner::record_dirty_usage_bucket(&bucket_clone);
                 debug!(bucket = %bucket_clone, object = %object_clone, "Transitioned object restored");
             }
         });

rustfs/src/storage/ecfs.rs

@@ -462,6 +462,7 @@ impl S3 for FS {
 
         let result = Ok(S3Response::new(DeleteObjectTaggingOutput { version_id }));
         let _ = helper.complete(&result);
+        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         let duration = start_time.elapsed();
         histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "delete").record(duration.as_secs_f64());
         result
@@ -1296,6 +1297,7 @@ impl S3 for FS {
 
         let result = Ok(S3Response::new(output));
         let _ = helper.complete(&result);
+        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         result
     }
 
@@ -1364,6 +1366,7 @@ impl S3 for FS {
                 .map_err(ApiError::from)?;
         }
 
+        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         Ok(S3Response::new(PutObjectLockConfigurationOutput::default()))
     }
 
@@ -1442,6 +1445,7 @@ impl S3 for FS {
 
         let result = Ok(S3Response::new(output));
         let _ = helper.complete(&result);
+        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         result
     }
 
@@ -1519,6 +1523,7 @@ impl S3 for FS {
             version_id: req.input.version_id.clone(),
         }));
         let _ = helper.complete(&result);
+        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         let duration = start_time.elapsed();
         histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "put").record(duration.as_secs_f64());
         result