diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 76aff02aa..593fbb782 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -118,16 +118,90 @@ fn randomized_cycle_delay_for(interval: Duration) -> Duration { delay.max(Duration::from_secs(1)) } -fn initial_scanner_delay() -> Duration { - initial_scanner_delay_for(scanner_start_delay().map(|duration| duration.as_secs())) -} - fn initial_scanner_delay_for(start_delay_secs: Option) -> Duration { start_delay_secs .map(|secs| randomized_cycle_delay_for(Duration::from_secs(secs))) .unwrap_or_else(randomized_cycle_delay) } +fn initial_scanner_delay_for_startup(start_delay_secs: Option, usage_cache_is_cold: bool, has_buckets: bool) -> Duration { + if usage_cache_is_cold && has_buckets { + Duration::ZERO + } else { + initial_scanner_delay_for(start_delay_secs) + } +} + +fn data_usage_info_is_cold(info: &DataUsageInfo) -> bool { + info.last_update.is_none() || (info.buckets_usage.is_empty() && info.bucket_sizes.is_empty()) +} + +async fn read_data_usage_config_for_startup(storeapi: &Arc) -> Result>, EcstoreError> { + match read_config(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()).await { + Ok(data) => Ok(Some(data)), + Err(EcstoreError::ConfigNotFound) => { + let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()); + match read_config(storeapi.clone(), backup_path.as_str()).await { + Ok(data) => Ok(Some(data)), + Err(EcstoreError::ConfigNotFound) => Ok(None), + Err(err) => Err(err), + } + } + Err(err) => Err(err), + } +} + +async fn persisted_usage_cache_is_cold_for_startup(storeapi: &Arc) -> bool { + let Some(data) = (match read_data_usage_config_for_startup(storeapi).await { + Ok(data) => data, + Err(err) => { + warn!( + error = %err, + "Failed to inspect persisted data usage cache; keeping configured scanner startup delay" + ); + return false; + } + }) else { + return true; + }; + + match serde_json::from_slice::(&data) { + Ok(info) => data_usage_info_is_cold(&info), + Err(err) => { + warn!( + error = %err, + "Failed to decode persisted data usage cache; running initial scanner cycle without startup delay" + ); + true + } + } +} + +async fn initial_scanner_startup_usage_state(storeapi: &Arc) -> (bool, bool) { + let has_buckets = match storeapi + .list_bucket(&BucketOptions { + no_metadata: true, + ..Default::default() + }) + .await + { + Ok(buckets) => !buckets.is_empty(), + Err(err) => { + warn!( + error = %err, + "Failed to inspect buckets for scanner startup delay; keeping configured scanner startup delay" + ); + false + } + }; + + if !has_buckets { + return (false, false); + } + + (persisted_usage_cache_is_cold_for_startup(storeapi).await, true) +} + pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { configure_scanner_defaults(&storeapi).await; // Force init global sleeper so config is read once at startup. @@ -139,8 +213,17 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { let ctx_clone = ctx; let storeapi_clone = storeapi; tokio::spawn(async move { - let sleep_time = initial_scanner_delay(); - tokio::time::sleep(sleep_time).await; + let (usage_cache_is_cold, has_buckets) = initial_scanner_startup_usage_state(&storeapi_clone).await; + let sleep_time = initial_scanner_delay_for_startup( + scanner_start_delay().map(|duration| duration.as_secs()), + usage_cache_is_cold, + has_buckets, + ); + if sleep_time.is_zero() { + info!("Skipping initial data scanner delay because persisted data usage cache is cold"); + } else { + tokio::time::sleep(sleep_time).await; + } loop { if ctx_clone.is_cancelled() { @@ -780,6 +863,29 @@ mod tests { crate::runtime_config::refresh_scanner_runtime_config_for_tests(); } + #[test] + #[serial] + fn test_initial_scanner_delay_skips_for_cold_usage_cache_with_buckets() { + let delay = initial_scanner_delay_for_startup(Some(120), true, true); + assert_eq!(delay, Duration::ZERO); + } + + #[test] + #[serial] + fn test_initial_scanner_delay_keeps_configured_delay_for_warm_usage_cache() { + let delay = initial_scanner_delay_for_startup(Some(120), false, true); + assert!(delay >= Duration::from_secs(108)); + assert!(delay <= Duration::from_secs(132)); + } + + #[test] + #[serial] + fn test_initial_scanner_delay_keeps_configured_delay_without_buckets() { + let delay = initial_scanner_delay_for_startup(Some(120), true, false); + assert!(delay >= Duration::from_secs(108)); + assert!(delay <= Duration::from_secs(132)); + } + #[test] #[serial] fn test_scanner_cycle_max_duration_uses_env() { diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index d72c8eb88..dfed36381 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -1024,14 +1024,18 @@ impl FolderScanner { /// Send update if enough time has passed /// Should be called on a regular basis when the new_cache contains more recent total than previously. /// May or may not send an update upstream. - pub async fn send_update(&mut self) { - // Send at most an update every minute. + fn should_send_update(&self) -> bool { if self.updates.is_none() { - return; + return false; } let elapsed = self.last_update.elapsed().unwrap_or(Duration::from_secs(0)); - if elapsed < Duration::from_secs(60) { + elapsed >= Duration::from_secs(60) + } + + pub async fn send_update(&mut self) { + // Send at most an update every minute. + if !self.should_send_update() { return; } @@ -1046,6 +1050,15 @@ impl FolderScanner { } } + async fn send_update_for_entry(&mut self, hash: &DataUsageHash, parent: &Option, entry: &DataUsageEntry) { + if !self.should_send_update() { + return; + } + + self.update_cache.replace_hashed(hash, parent, entry); + self.send_update().await; + } + /// Scan a folder recursively /// Files found in the folders will be added to new_cache. #[allow(clippy::never_loop)] @@ -1300,6 +1313,7 @@ impl FolderScanner { } if should_yield_after_object(object_count, yield_every_objects) { + self.send_update_for_entry(&this_hash, &folder.parent, into).await; let yield_start = Instant::now(); tokio::task::yield_now().await; global_metrics().record_scanner_yield(yield_start.elapsed()); @@ -1445,6 +1459,7 @@ impl FolderScanner { let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), into)); fut.await.map_err(|e| ScannerError::Other(e.to_string()))?; self.record_scan_resume_hint(&folder_item.name); + self.send_update_for_entry(&this_hash, &folder.parent, into).await; tokio::task::yield_now().await; } else { let mut dst = DataUsageEntry::default(); @@ -1694,6 +1709,7 @@ impl FolderScanner { // In compacted mode child totals are accumulated directly into the parent entry. let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), into)); fut.await.map_err(|e| ScannerError::Other(e.to_string()))?; + self.send_update_for_entry(&this_hash, &folder.parent, into).await; tokio::task::yield_now().await; } else { let mut dst = DataUsageEntry::default(); @@ -2601,6 +2617,48 @@ mod tests { assert!(budget.token().is_cancelled()); } + #[tokio::test] + #[serial] + async fn test_scan_folder_compacted_parent_sends_partial_update() { + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone()); + + let bucket_dir = temp_dir.join("bucket"); + tokio::fs::create_dir_all(bucket_dir.join("child")) + .await + .expect("failed to create child directory"); + + scanner.old_cache.info.name = "bucket".to_string(); + scanner.new_cache.info.name = "bucket".to_string(); + scanner.update_cache.info.name = "bucket".to_string(); + scanner.last_update = SystemTime::UNIX_EPOCH; + + let (tx, mut rx) = mpsc::channel(1); + scanner.updates = Some(tx); + + let folder = CachedFolder { + name: "bucket".to_string(), + parent: None, + object_heal_prob_div: 1, + }; + let mut into = DataUsageEntry { + compacted: true, + ..Default::default() + }; + + scanner + .scan_folder(CancellationToken::new(), folder, &mut into) + .await + .expect("compacted scan should finish successfully"); + + let update = tokio::time::timeout(Duration::from_secs(1), rx.recv()) + .await + .expect("compacted scan should send a partial update") + .expect("partial update channel should remain open"); + + assert!(update.compacted, "partial update should preserve compacted state"); + } + #[tokio::test] #[serial] async fn test_scan_data_folder_returns_partial_cache_on_budget_cancel() { diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index db716e41f..2666179e7 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -45,7 +45,7 @@ use rustfs_ecstore::{StorageAPI, error::Result, store::ECStore}; use rustfs_filemeta::FileMeta; use rustfs_utils::path::path_join_buf; use s3s::dto::{BucketLifecycleConfiguration, ReplicationConfiguration}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::path::Path; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::{Instant, SystemTime}; @@ -269,6 +269,22 @@ fn cache_root_entry_info(cache: &DataUsageCache) -> DataUsageEntryInfo { } } +fn apply_bucket_result_to_cache( + cache: &mut DataUsageCache, + result: DataUsageEntryInfo, + update_time: SystemTime, + publish_immediately: bool, +) -> bool { + cache.replace(&result.name, &result.parent, result.entry); + cache.info.last_update = Some(update_time); + + publish_immediately +} + +fn bucket_result_should_publish_immediately(published_buckets: &mut HashSet, bucket_name: &str) -> bool { + published_buckets.insert(bucket_name.to_string()) +} + async fn send_cache_root_entry_info( bucket_result_tx: &Arc>>, cache: &DataUsageCache, @@ -620,6 +636,7 @@ impl ScannerIOCache for SetDisks { let mut permutes = buckets.clone(); permutes.shuffle(&mut rand::rng()); + let mut preloaded_published_buckets = HashSet::new(); for bucket in permutes.iter() { if old_cache.find(&bucket.name).is_none() @@ -632,6 +649,9 @@ 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!("Failed to send bucket info: {}", e); @@ -677,10 +697,21 @@ impl ScannerIOCache for SetDisks { } res = bucket_result_rx.recv() => { if let Some(result) = res { - let mut cache = cache_mutex_clone.lock().await; - cache.replace(&result.name, &result.parent, result.entry); - cache.info.last_update = Some(SystemTime::now()); + let cache_snapshot = { + let mut cache = cache_mutex_clone.lock().await; + let publish_immediately = + bucket_result_should_publish_immediately(&mut preloaded_published_buckets, &result.name); + if apply_bucket_result_to_cache(&mut cache, result, SystemTime::now(), publish_immediately) { + Some(cache.clone()) + } else { + None + } + }; + if let Some(cache_snapshot) = cache_snapshot { + last_update = + persist_and_publish_cache_snapshot(store_clone.clone(), &updates, cache_snapshot).await; + } } else { let cache_snapshot = { let mut cache = cache_mutex_clone.lock().await; @@ -1260,4 +1291,126 @@ mod tests { assert_eq!(info.entry.size, 10); assert_eq!(info.entry.objects, 1); } + + #[test] + fn apply_bucket_result_requests_immediate_publish_for_missing_bucket() { + let mut cache = DataUsageCache { + info: DataUsageCacheInfo { + name: DATA_USAGE_ROOT.to_string(), + ..Default::default() + }, + ..Default::default() + }; + + let should_publish = apply_bucket_result_to_cache( + &mut cache, + DataUsageEntryInfo { + name: "bucket".to_string(), + parent: DATA_USAGE_ROOT.to_string(), + entry: DataUsageEntry { + size: 10, + objects: 1, + ..Default::default() + }, + }, + SystemTime::now(), + true, + ); + + assert!(should_publish); + assert!(cache.info.last_update.is_some()); + let entry = cache.find("bucket").expect("bucket entry should be inserted"); + assert_eq!(entry.size, 10); + assert_eq!(entry.objects, 1); + } + + #[test] + fn apply_bucket_result_defers_publish_for_existing_published_bucket() { + let mut cache = DataUsageCache { + info: DataUsageCacheInfo { + name: DATA_USAGE_ROOT.to_string(), + last_update: Some(SystemTime::now()), + ..Default::default() + }, + ..Default::default() + }; + cache.replace( + "bucket", + DATA_USAGE_ROOT, + DataUsageEntry { + size: 5, + objects: 1, + ..Default::default() + }, + ); + + let should_publish = apply_bucket_result_to_cache( + &mut cache, + DataUsageEntryInfo { + name: "bucket".to_string(), + parent: DATA_USAGE_ROOT.to_string(), + entry: DataUsageEntry { + size: 10, + objects: 2, + ..Default::default() + }, + }, + SystemTime::now(), + false, + ); + + assert!(!should_publish); + let entry = cache.find("bucket").expect("bucket entry should remain present"); + assert_eq!(entry.size, 10); + assert_eq!(entry.objects, 2); + } + + #[test] + fn apply_bucket_result_defers_publish_for_preloaded_published_bucket() { + let mut cache = DataUsageCache { + info: DataUsageCacheInfo { + name: DATA_USAGE_ROOT.to_string(), + ..Default::default() + }, + ..Default::default() + }; + cache.replace( + "bucket", + DATA_USAGE_ROOT, + DataUsageEntry { + size: 5, + objects: 1, + ..Default::default() + }, + ); + + let should_publish = apply_bucket_result_to_cache( + &mut cache, + DataUsageEntryInfo { + name: "bucket".to_string(), + parent: DATA_USAGE_ROOT.to_string(), + entry: DataUsageEntry { + size: 10, + objects: 2, + ..Default::default() + }, + }, + SystemTime::now(), + false, + ); + + assert!(!should_publish); + let entry = cache.find("bucket").expect("bucket entry should remain present"); + assert_eq!(entry.size, 10); + assert_eq!(entry.objects, 2); + } + + #[test] + fn bucket_result_immediate_publish_tracks_preloaded_and_current_results() { + let mut published_buckets = HashSet::from(["existing".to_string()]); + + assert!(!bucket_result_should_publish_immediately(&mut published_buckets, "existing")); + assert!(bucket_result_should_publish_immediately(&mut published_buckets, "missing")); + assert!(!bucket_result_should_publish_immediately(&mut published_buckets, "missing")); + } }