diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 484e56c89..23f302b55 100644 --- a/crates/scanner/src/scanner.rs +++ b/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) -> 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) -> 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() { diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 44ca8a601..917b29835 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/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> = LazyLock::new(|| StdMutex::new(HashMap::new())); +static DIRTY_USAGE_BUCKET_NOTIFY: LazyLock = 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(); diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index ec9ccac80..263403ccf 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/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 {})) } diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 34934acb6..d25a231d2 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/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(_)) { diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 5eb5459f1..291169bcc 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/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"); } }); diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index f887cbd10..b6ffc3e12 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/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