From 975983abdd752f2d9d8e411fa402249584dffd74 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Mon, 7 Sep 2026 05:26:56 +0800 Subject: [PATCH] feat(scanner): reuse clean local bucket prefixes (#7208) --- crates/scanner/src/lib.rs | 4 +- crates/scanner/src/remote_scanner/stream.rs | 18 +- crates/scanner/src/scanner_folder.rs | 97 ++++++++++- crates/scanner/src/scanner_folder/tests.rs | 110 +++++++++++++ .../tests/checkpoint_fixture.rs | 38 ++++- .../checkpoint_fixture/deep_compacted.rs | 2 +- crates/scanner/src/scanner_io.rs | 47 +++++- crates/scanner/src/scanner_io/dirty_usage.rs | 155 +++++++++++++++++- crates/scanner/src/scanner_io/io_cache.rs | 12 +- crates/scanner/src/scanner_io/io_cycle.rs | 1 + crates/scanner/src/scanner_io/io_disk.rs | 23 ++- crates/scanner/src/scanner_io/tests.rs | 103 +++++++++++- rustfs/src/app/multipart_usecase.rs | 2 +- rustfs/src/app/object/copy.rs | 2 +- rustfs/src/app/object/delete.rs | 2 +- rustfs/src/app/object/internal_put.rs | 2 +- rustfs/src/app/object/put.rs | 2 +- rustfs/src/app/object/restore.rs | 4 +- 18 files changed, 577 insertions(+), 47 deletions(-) diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 7cc6d22e8..539e0af94 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -91,8 +91,8 @@ pub use scanner::{ pub use scanner_io::{ ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState, acknowledge_dirty_usage_generation, acknowledge_scoped_dirty_usage, clear_dirty_usage_bucket, record_dirty_usage_bucket, - record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state, - scanner_maintenance_generation, + record_dirty_usage_object, record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_snapshot, + scanner_dirty_usage_state, scanner_maintenance_generation, }; pub use sleeper::{DynamicSleeper, SCANNER_IDLE_MODE, SCANNER_SLEEPER}; use std::sync::atomic::{AtomicU64, Ordering}; diff --git a/crates/scanner/src/remote_scanner/stream.rs b/crates/scanner/src/remote_scanner/stream.rs index 567490244..3e31d81f8 100644 --- a/crates/scanner/src/remote_scanner/stream.rs +++ b/crates/scanner/src/remote_scanner/stream.rs @@ -16,8 +16,9 @@ use crate::RUSTFS_META_BUCKET; use crate::scanner_budget::{ScannerCycleBudget, ScannerCycleBudgetConfig}; use crate::scanner_io::{ - DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerDiskScanOutcome, ScannerIODisk, acquire_scanner_cache_locks, - cache_root_entry_info, current_cache_root_or_prepare_with_generation, scanner_set_disk_inventory, + DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerDiskScanOptions, ScannerDiskScanOutcome, ScannerIODisk, + acquire_scanner_cache_locks, cache_root_entry_info, current_cache_root_or_prepare_with_generation, + scanner_set_disk_inventory, }; use crate::storage_api::owner::NS_SCANNER_PROTOCOL_VERSION; use crate::{ @@ -779,7 +780,18 @@ async fn scan_and_persist_local_bucket( let set_disks = scanner_set_disk_inventory(set.as_ref()).await; let scan_ctx = ctx.child_token(); - let scan = ScannerIODisk::nsscanner_disk(disk.clone(), scan_ctx.clone(), budget, set_disks, cache, None, scan_mode); + let scan = ScannerIODisk::nsscanner_disk( + disk.clone(), + scan_ctx.clone(), + budget, + set_disks, + cache, + None, + ScannerDiskScanOptions { + scan_mode, + prefix_scan_scope: None, + }, + ); tokio::pin!(scan); let fence_watch = watch_remote_scanner_request_fence(next_cycle, leader_epoch, store.clone(), NS_SCANNER_FENCE_POLL_INTERVAL); tokio::pin!(fence_watch); diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 795cb8cc9..05e04b0ac 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -675,6 +675,28 @@ fn partial_cache_is_useful(root: &DataUsageEntry, pending_heals_changed: bool) - data_usage_root_has_progress(root) || pending_heals_changed } +/// Process-local hint that narrows a dirty bucket scan to known changed direct children. +/// +/// The hint is used only while rebuilding a complete bucket cache. It never +/// changes usage publication semantics and callers must discard it when the +/// mutation source cannot be verified. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ScannerBucketPrefixScanScope { + selected_top_level_entries: Arc>, +} + +impl ScannerBucketPrefixScanScope { + pub(crate) fn from_dirty_top_level_entries(entries: HashSet) -> Option { + (!entries.is_empty()).then(|| Self { + selected_top_level_entries: Arc::new(entries), + }) + } + + fn contains(&self, entry: &str) -> bool { + self.selected_top_level_entries.contains(entry) + } +} + /// Folder scanner for scanning directory structures pub struct FolderScanner { root: String, @@ -686,6 +708,7 @@ pub struct FolderScanner { heal_object_select: u32, scan_mode: HealScanMode, is_erasure_mode: bool, + prefix_scan_scope: Option, failed_object_ttl_secs: u64, failed_objects_max: usize, @@ -781,6 +804,47 @@ impl FolderScanner { .as_secs() } + fn should_reuse_clean_root_child( + &self, + folder: &CachedFolder, + into: &DataUsageEntry, + child: &CachedFolder, + child_hash: &DataUsageHash, + abandoned_children: &DataUsageHashMap, + ) -> bool { + let Some(prefix_scan_scope) = &self.prefix_scan_scope else { + return false; + }; + // Erasure-mode usage scans also perform probabilistic object-health + // work. Reusing a clean subtree here would silently suppress that + // independent maintenance path, so prefix reuse is limited to the + // non-erasure data-usage scanner. + if self.is_erasure_mode + || folder.parent.is_some() + || folder.name != self.old_cache.info.name + || into.compacted + || !abandoned_children.contains(&child_hash.key()) + { + return false; + } + + let Some(entry) = child + .name + .strip_prefix(folder.name.as_str()) + .and_then(|entry| entry.strip_prefix('/')) + else { + return false; + }; + !entry.is_empty() && !entry.contains('/') && !prefix_scan_scope.contains(entry) + } + + fn reuse_clean_root_child(&mut self, child: &CachedFolder, child_hash: &DataUsageHash, into: &mut DataUsageEntry) { + self.new_cache.copy_with_children(&self.old_cache, child_hash, &child.parent); + self.update_cache + .copy_with_children(&self.old_cache, child_hash, &child.parent); + into.add_child(child_hash); + } + fn should_skip_failed(&self, path: &str) -> bool { let ttl = self.failed_object_ttl_secs; if ttl == 0 { @@ -1451,13 +1515,16 @@ impl FolderScanner { continue; } - abandoned_children.remove(&h.key()); - - if exists { + if exists && self.should_reuse_clean_root_child(&folder, into, &this, &h, &abandoned_children) { + abandoned_children.remove(&h.key()); + self.reuse_clean_root_child(&this, &h, into); + } else if exists { + abandoned_children.remove(&h.key()); existing_folders.push(this); self.update_cache .copy_with_children(&self.old_cache, &h, &Some(this_hash.clone())); } else { + abandoned_children.remove(&h.key()); new_folders.push(this); } continue; @@ -1633,11 +1700,15 @@ impl FolderScanner { if !found_object_metadata && !found_erasure_data_directory { for (candidate, exists, _) in erasure_data_directory_candidates { let h = hash_path(&candidate.name); - abandoned_children.remove(&h.key()); - if exists { + if exists && self.should_reuse_clean_root_child(&folder, into, &candidate, &h, &abandoned_children) { + abandoned_children.remove(&h.key()); + self.reuse_clean_root_child(&candidate, &h, into); + } else if exists { + abandoned_children.remove(&h.key()); self.update_cache.copy_with_children(&self.old_cache, &h, &candidate.parent); existing_folders.push(candidate); } else { + abandoned_children.remove(&h.key()); new_folders.push(candidate); } } @@ -2377,6 +2448,21 @@ pub async fn scan_data_folder( updates: Option>, scan_mode: HealScanMode, sleeper: DynamicSleeper, +) -> Result { + scan_data_folder_scoped(ctx, budget, disks, local_disk, cache, updates, scan_mode, sleeper, None).await +} + +#[allow(clippy::too_many_arguments)] +pub(crate) async fn scan_data_folder_scoped( + ctx: CancellationToken, + budget: Arc, + disks: Vec>, + local_disk: Arc, + cache: DataUsageCache, + updates: Option>, + scan_mode: HealScanMode, + sleeper: DynamicSleeper, + prefix_scan_scope: Option, ) -> Result { use crate::data_usage_define::DATA_USAGE_ROOT; @@ -2428,6 +2514,7 @@ pub async fn scan_data_folder( heal_object_select, scan_mode, is_erasure_mode, + prefix_scan_scope, failed_object_ttl_secs: failed_object_ttl, failed_objects_max, sleeper, diff --git a/crates/scanner/src/scanner_folder/tests.rs b/crates/scanner/src/scanner_folder/tests.rs index 723c79207..962090a00 100644 --- a/crates/scanner/src/scanner_folder/tests.rs +++ b/crates/scanner/src/scanner_folder/tests.rs @@ -332,6 +332,7 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) { heal_object_select: 0, scan_mode: HealScanMode::Normal, is_erasure_mode: false, + prefix_scan_scope: None, failed_object_ttl_secs: u64::MAX, failed_objects_max: usize::MAX, sleeper: SCANNER_SLEEPER.clone(), @@ -2384,6 +2385,115 @@ async fn test_scan_folder_non_erasure_metadata_keeps_namespace_descent() { assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Directories)); } +#[tokio::test] +#[serial] +async fn scoped_root_scan_reuses_clean_top_level_entries_and_rescans_dirty_entries() { + 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("clean")) + .await + .expect("failed to create clean top-level directory"); + tokio::fs::create_dir_all(bucket_dir.join("dirty")) + .await + .expect("failed to create dirty top-level 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.old_cache.replace("bucket", "", DataUsageEntry::default()); + scanner.old_cache.replace( + "bucket/clean", + "bucket", + DataUsageEntry { + size: 17, + objects: 3, + ..Default::default() + }, + ); + scanner.old_cache.replace( + "bucket/dirty", + "bucket", + DataUsageEntry { + size: 23, + objects: 4, + ..Default::default() + }, + ); + scanner.prefix_scan_scope = ScannerBucketPrefixScanScope::from_dirty_top_level_entries(HashSet::from(["dirty".to_string()])); + + let folder = CachedFolder { + name: "bucket".to_string(), + parent: None, + object_heal_prob_div: 1, + }; + let mut root = DataUsageEntry::default(); + scanner + .scan_folder(CancellationToken::new(), folder, &mut root) + .await + .expect("scoped root scan should finish successfully"); + + let clean = scanner + .new_cache + .size_recursive("bucket/clean") + .expect("clean entry should be copied from the complete cache"); + assert_eq!((clean.size, clean.objects), (17, 3)); + let dirty = scanner + .new_cache + .size_recursive("bucket/dirty") + .expect("dirty entry should be rescanned"); + assert_eq!((dirty.size, dirty.objects), (0, 0)); + let bucket = scanner + .new_cache + .size_recursive("bucket") + .expect("bucket root should include reused and rescanned entries"); + assert_eq!((bucket.size, bucket.objects), (17, 3)); +} + +#[tokio::test] +#[serial] +async fn scoped_root_scan_preserves_erasure_health_walks() { + 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("clean")) + .await + .expect("failed to create clean top-level directory"); + + scanner.is_erasure_mode = true; + 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.old_cache.replace("bucket", "", DataUsageEntry::default()); + scanner.old_cache.replace( + "bucket/clean", + "bucket", + DataUsageEntry { + size: 17, + objects: 3, + ..Default::default() + }, + ); + scanner.prefix_scan_scope = ScannerBucketPrefixScanScope::from_dirty_top_level_entries(HashSet::from(["dirty".to_string()])); + + let folder = CachedFolder { + name: "bucket".to_string(), + parent: None, + object_heal_prob_div: 1, + }; + let mut root = DataUsageEntry::default(); + scanner + .scan_folder(CancellationToken::new(), folder, &mut root) + .await + .expect("erasure root scan should finish successfully"); + + let clean = scanner + .new_cache + .size_recursive("bucket/clean") + .expect("erasure scan should visit the clean entry"); + assert_eq!((clean.size, clean.objects), (0, 0)); +} + #[tokio::test] #[serial] async fn test_scan_folder_compacted_parent_sends_partial_update() { diff --git a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs index 45bddb6ab..a253ba84e 100644 --- a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs +++ b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs @@ -14,7 +14,7 @@ use super::*; use crate::scanner_budget::ScannerCycleBudgetConfig; -use crate::scanner_io::{ScannerDiskScanOutcome, ScannerIODisk}; +use crate::scanner_io::{ScannerDiskScanOptions, ScannerDiskScanOutcome, ScannerIODisk}; use crate::storage_api::scanner_io::ObjectIO; use crate::{DataUsageCacheSource, DataUsageScanPlanDigest}; use std::io::Cursor; @@ -29,6 +29,13 @@ const MAX_CACHE_BYTES: u64 = 1024 * 1024; const SOURCE: DataUsageCacheSource = DataUsageCacheSource::new(0, 0); const PLAN: DataUsageScanPlanDigest = DataUsageScanPlanDigest([17; 32]); +fn scan_options(scan_mode: HealScanMode) -> ScannerDiskScanOptions { + ScannerDiskScanOptions { + scan_mode, + prefix_scan_scope: None, + } +} + /// Real cache persistence codec and CAS calls, backed by two bounded local files. #[derive(Debug)] struct FixtureStore { @@ -538,7 +545,7 @@ async fn checkpoint_fixture_existing_uncovered_cursor_cannot_skip_to_complete() vec![scanner.local_disk.clone()], loaded, None, - HealScanMode::Normal, + scan_options(HealScanMode::Normal), ) .await .expect("scan must revisit the prefix"); @@ -594,7 +601,7 @@ async fn checkpoint_fixture_failed_child_prevents_receipt_advancing_past_gap() { vec![scanner.local_disk.clone()], cache, None, - HealScanMode::Normal, + scan_options(HealScanMode::Normal), ) .await .expect("scan with a known failed child"); @@ -645,7 +652,7 @@ async fn check_complete_sampling_resumption(resume_mode: HealScanMode) { vec![scanner.local_disk.clone()], cache, None, - HealScanMode::Normal, + scan_options(HealScanMode::Normal), ) .await .expect("initial complete baseline"); @@ -690,7 +697,7 @@ async fn check_complete_sampling_resumption(resume_mode: HealScanMode) { vec![scanner.local_disk.clone()], cache, None, - HealScanMode::Normal, + scan_options(HealScanMode::Normal), ) .await .expect("sampling interruption"); @@ -737,7 +744,14 @@ async fn check_complete_sampling_resumption(resume_mode: HealScanMode) { let result = scanner .local_disk .clone() - .nsscanner_disk(budget.token(), budget, vec![scanner.local_disk.clone()], loaded, None, resume_mode) + .nsscanner_disk( + budget.token(), + budget, + vec![scanner.local_disk.clone()], + loaded, + None, + scan_options(resume_mode), + ) .await .expect("bounded recovery scan"); let (cache, complete) = match result { @@ -853,7 +867,10 @@ async fn run_checkpoint_fixture(change_digest: bool) { vec![scanner.local_disk.clone()], cache, None, - HealScanMode::Normal, + ScannerDiskScanOptions { + scan_mode: HealScanMode::Normal, + prefix_scan_scope: None, + }, ) .await .expect("budgeted local disk scan returns partial cache"); @@ -931,7 +948,10 @@ async fn run_checkpoint_fixture(change_digest: bool) { vec![scanner.local_disk.clone()], loaded.clone(), None, - HealScanMode::Normal, + ScannerDiskScanOptions { + scan_mode: HealScanMode::Normal, + prefix_scan_scope: None, + }, ) .await; assert!(result.is_err(), "pre-scan cancellation must not produce a complete root"); @@ -982,7 +1002,7 @@ async fn run_checkpoint_fixture(change_digest: bool) { vec![scanner.local_disk.clone()], cache, None, - HealScanMode::Normal, + scan_options(HealScanMode::Normal), ) .await .expect("bounded sweep outcome"); diff --git a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/deep_compacted.rs b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/deep_compacted.rs index cc2013dc6..7a2d6ae9c 100644 --- a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/deep_compacted.rs +++ b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/deep_compacted.rs @@ -30,7 +30,7 @@ async fn scan( ); let outcome = disk .clone() - .nsscanner_disk(budget.token(), budget.clone(), vec![disk.clone()], cache, None, mode) + .nsscanner_disk(budget.token(), budget.clone(), vec![disk.clone()], cache, None, super::scan_options(mode)) .await .expect("bounded real disk scan"); (outcome, budget) diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index b51e5d2f8..a0da35913 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -14,7 +14,7 @@ use crate::data_usage_define::{DATA_USAGE_CACHE_KEY_FORMAT, DataUsageCacheRevisions}; use crate::scanner_budget::ScannerCycleBudget; -use crate::scanner_folder::{ScannerItem, scan_data_folder}; +use crate::scanner_folder::{ScannerBucketPrefixScanScope, ScannerItem, scan_data_folder_scoped}; use crate::sleeper::SCANNER_SLEEPER; use crate::{ DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, DataUsageCache, DataUsageCacheInfo, DataUsageCachePrepareOutcome, @@ -103,6 +103,7 @@ pub type DirtyUsageBuckets = HashMap; #[derive(Clone, Debug)] struct DirtyUsageSnapshot { buckets: Arc, + scopes: Arc, generation: u64, covers_all_pending: bool, } @@ -110,6 +111,7 @@ struct DirtyUsageSnapshot { #[derive(Clone, Debug, Default)] pub(crate) struct ScannerBucketScanScope { selected_buckets: Option>>, + selected_bucket_prefixes: Option>>, baseline_scan_plan_digest: Option, } @@ -120,15 +122,24 @@ impl ScannerBucketScanScope { } fn is_default(&self) -> bool { - self.selected_buckets.is_none() && self.baseline_scan_plan_digest.is_none() + self.selected_buckets.is_none() && self.selected_bucket_prefixes.is_none() && self.baseline_scan_plan_digest.is_none() } - fn from_dirty_buckets(selected_buckets: HashSet, baseline_scan_plan_digest: DataUsageScanPlanDigest) -> Self { + fn from_dirty_buckets( + selected_buckets: HashSet, + selected_bucket_prefixes: HashMap, + baseline_scan_plan_digest: DataUsageScanPlanDigest, + ) -> Self { Self { selected_buckets: Some(Arc::new(selected_buckets)), + selected_bucket_prefixes: (!selected_bucket_prefixes.is_empty()).then(|| Arc::new(selected_bucket_prefixes)), baseline_scan_plan_digest: Some(baseline_scan_plan_digest), } } + + pub(crate) fn prefix_scope_for(&self, bucket: &str) -> Option { + self.selected_bucket_prefixes.as_ref()?.get(bucket).cloned() + } } #[derive(Clone, Copy)] @@ -248,6 +259,7 @@ fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof< fn scoped_scan_scope_from_dirty_buckets( requested_scope: ScannerBucketScanScope, dirty_buckets: HashSet, + dirty_scopes: Option<&DirtyUsageBucketScopes>, dirty_snapshot_complete: bool, all_buckets: &[BucketInfo], baseline_proof: ScannerCacheBaselineProof<'_>, @@ -269,7 +281,22 @@ fn scoped_scan_scope_from_dirty_buckets( return requested_scope; }; - ScannerBucketScanScope::from_dirty_buckets(selected_buckets, baseline_scan_plan_digest) + let selected_bucket_prefixes = dirty_scopes + .into_iter() + .flat_map(|dirty_scopes| { + selected_buckets + .iter() + .filter_map(|bucket| dirty_scopes.get(bucket).map(|scope| (bucket.clone(), scope))) + }) + .filter_map(|(bucket, scope)| match scope { + DirtyUsageBucketScope::WholeBucket => None, + DirtyUsageBucketScope::TopLevelEntries(entries) => { + ScannerBucketPrefixScanScope::from_dirty_top_level_entries(entries.clone()).map(|scope| (bucket, scope)) + } + }) + .collect(); + + ScannerBucketScanScope::from_dirty_buckets(selected_buckets, selected_bucket_prefixes, baseline_scan_plan_digest) } pub(crate) fn is_scanner_metadata_corrupt_error(err: &StorageError) -> bool { @@ -765,6 +792,12 @@ pub trait ScannerIOCache: Send + Sync + Debug + 'static { ) -> Result<()>; } +#[derive(Debug)] +pub struct ScannerDiskScanOptions { + pub scan_mode: HealScanMode, + pub prefix_scan_scope: Option, +} + #[async_trait::async_trait] pub trait ScannerIODisk: Send + Sync + Debug + 'static { async fn nsscanner_disk( @@ -774,7 +807,7 @@ pub trait ScannerIODisk: Send + Sync + Debug + 'static { set_disks: Vec>, cache: DataUsageCache, updates: Option>, - scan_mode: HealScanMode, + options: ScannerDiskScanOptions, ) -> Result; async fn get_size(&self, item: ScannerItem) -> Result; @@ -1065,8 +1098,8 @@ pub(crate) use cache::{ pub use dirty_usage::{ ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState, acknowledge_dirty_usage_generation, acknowledge_scoped_dirty_usage, clear_dirty_usage_bucket, record_dirty_usage_bucket, - record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state, - scanner_maintenance_generation, + record_dirty_usage_object, record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_snapshot, + scanner_dirty_usage_state, scanner_maintenance_generation, }; #[cfg(test)] pub(crate) use dirty_usage::{clear_dirty_usage_buckets_for_tests, dirty_usage_buckets_for_tests}; diff --git a/crates/scanner/src/scanner_io/dirty_usage.rs b/crates/scanner/src/scanner_io/dirty_usage.rs index 18db78a81..6457f9585 100644 --- a/crates/scanner/src/scanner_io/dirty_usage.rs +++ b/crates/scanner/src/scanner_io/dirty_usage.rs @@ -16,6 +16,12 @@ use super::*; pub(super) static DIRTY_USAGE_BUCKET_GENERATION: AtomicU64 = AtomicU64::new(0); pub(super) static DIRTY_USAGE_BUCKETS: LazyLock> = LazyLock::new(|| StdMutex::new(HashMap::new())); +// Lock order when both dirty maps are needed is `DIRTY_USAGE_BUCKETS` followed +// by `DIRTY_USAGE_BUCKET_SCOPES`. Both are held only for synchronous map +// updates, so no scanner task can observe a bucket generation without its +// matching scope. +pub(super) static DIRTY_USAGE_BUCKET_SCOPES: LazyLock> = + LazyLock::new(|| StdMutex::new(HashMap::new())); pub(super) static DIRTY_USAGE_BUCKET_NOTIFY: LazyLock = LazyLock::new(Notify::new); pub(super) static SCANNER_ACTIVITY_EPOCH: LazyLock = LazyLock::new(|| format!("{:032x}", rand::random::())); pub(super) static SCANNER_MAINTENANCE_GENERATION: AtomicU64 = AtomicU64::new(0); @@ -33,6 +39,21 @@ pub struct ScannerDirtyUsageBucket { pub generation: u64, } +/// A non-durable optimization hint for a dirty bucket. +/// +/// A whole-bucket marker always wins over narrow path hints. The scanner never +/// publishes a prefix-only result as authoritative usage; this only controls +/// whether a complete per-bucket cache can reuse known-clean direct children. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) enum DirtyUsageBucketScope { + WholeBucket, + TopLevelEntries(HashSet), +} + +pub(super) type DirtyUsageBucketScopes = HashMap; + +const MAX_DIRTY_USAGE_TOP_LEVEL_ENTRIES_PER_BUCKET: usize = 128; + /// A point-in-time view of the local dirty bucket generations. /// /// `complete == false` is an all-or-nothing overflow signal: `buckets` is @@ -67,6 +88,7 @@ pub fn acknowledge_scoped_dirty_usage( // No await or storage operation occurs while the dirty map is locked. let (cleared, pending) = { let mut dirty = dirty_usage_buckets(); + let mut dirty_scopes = dirty_usage_bucket_scopes(); let checked = entries .iter() .map(|(guard, generation)| { @@ -81,6 +103,7 @@ pub fn acknowledge_scoped_dirty_usage( scanner_activity_epoch(), DIRTY_USAGE_BUCKET_GENERATION.load(Ordering::Acquire), &mut dirty, + &mut dirty_scopes, &checked, probe_only, )?; @@ -100,6 +123,7 @@ fn apply_scoped_dirty_usage_ack( current_instance: &str, current_generation: u64, dirty: &mut DirtyUsageBuckets, + dirty_scopes: &mut DirtyUsageBucketScopes, entries: &[(&str, u64)], probe_only: bool, ) -> std::result::Result { @@ -118,6 +142,7 @@ fn apply_scoped_dirty_usage_ack( for (bucket, generation) in entries { if dirty.get(*bucket) == Some(generation) { dirty.remove(*bucket); + dirty_scopes.remove(*bucket); cleared += 1; } } @@ -132,30 +157,60 @@ mod scoped_dirty_usage_tests { #[test] fn scoped_dirty_usage_preserves_uncovered_newer_and_replayed_generations() { let mut dirty = HashMap::from([("hot".to_string(), 7), ("cold".to_string(), 8)]); - assert_eq!(apply_scoped_dirty_usage_ack("p", "p", 8, &mut dirty, &[("cold", 8)], true), Ok(0)); + let mut scopes = HashMap::from([ + ("hot".to_string(), DirtyUsageBucketScope::WholeBucket), + ( + "cold".to_string(), + DirtyUsageBucketScope::TopLevelEntries(HashSet::from(["first".to_string()])), + ), + ]); + assert_eq!( + apply_scoped_dirty_usage_ack("p", "p", 8, &mut dirty, &mut scopes, &[("cold", 8)], true), + Ok(0) + ); assert_eq!(dirty.len(), 2); - assert_eq!(apply_scoped_dirty_usage_ack("p", "p", 8, &mut dirty, &[("cold", 8)], false), Ok(1)); + assert!(scopes.contains_key("cold")); + assert_eq!( + apply_scoped_dirty_usage_ack("p", "p", 8, &mut dirty, &mut scopes, &[("cold", 8)], false), + Ok(1) + ); assert_eq!(dirty.get("hot"), Some(&7)); - assert_eq!(apply_scoped_dirty_usage_ack("p", "p", 8, &mut dirty, &[("cold", 8)], false), Ok(0)); + assert!(!scopes.contains_key("cold")); + assert_eq!( + apply_scoped_dirty_usage_ack("p", "p", 8, &mut dirty, &mut scopes, &[("cold", 8)], false), + Ok(0) + ); dirty.insert("cold".to_string(), 9); - assert_eq!(apply_scoped_dirty_usage_ack("p", "p", 9, &mut dirty, &[("cold", 8)], false), Ok(0)); + scopes.insert("cold".to_string(), DirtyUsageBucketScope::WholeBucket); + assert_eq!( + apply_scoped_dirty_usage_ack("p", "p", 9, &mut dirty, &mut scopes, &[("cold", 8)], false), + Ok(0) + ); assert_eq!(dirty.get("cold"), Some(&9)); + assert!(scopes.contains_key("cold")); } #[test] fn scoped_dirty_usage_rejects_restart_and_invalid_batch_before_clearing() { let original = HashMap::from([("hot".to_string(), 7), ("cold".to_string(), 8)]); let mut dirty = original.clone(); + let original_scopes = HashMap::from([ + ("hot".to_string(), DirtyUsageBucketScope::WholeBucket), + ("cold".to_string(), DirtyUsageBucketScope::WholeBucket), + ]); + let mut scopes = original_scopes.clone(); assert_eq!( - apply_scoped_dirty_usage_ack("old", "new", 8, &mut dirty, &[("cold", 8)], false), + apply_scoped_dirty_usage_ack("old", "new", 8, &mut dirty, &mut scopes, &[("cold", 8)], false), Err(ScannerDirtyUsageAckError::ProcessChanged) ); + assert_eq!(scopes, original_scopes); for generation in [0, 9, u64::MAX] { assert_eq!( - apply_scoped_dirty_usage_ack("p", "p", 8, &mut dirty, &[("cold", 8), ("hot", generation)], false), + apply_scoped_dirty_usage_ack("p", "p", 8, &mut dirty, &mut scopes, &[("cold", 8), ("hot", generation)], false,), Err(ScannerDirtyUsageAckError::InvalidGeneration) ); assert_eq!(dirty, original); + assert_eq!(scopes, original_scopes); } } } @@ -164,6 +219,12 @@ pub(super) fn dirty_usage_buckets() -> MutexGuard<'static, DirtyUsageBuckets> { DIRTY_USAGE_BUCKETS.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) } +fn dirty_usage_bucket_scopes() -> MutexGuard<'static, DirtyUsageBucketScopes> { + DIRTY_USAGE_BUCKET_SCOPES + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + pub(super) fn usize_to_u64_saturated(value: usize) -> u64 { u64::try_from(value).unwrap_or(u64::MAX) } @@ -181,8 +242,10 @@ pub fn record_dirty_usage_bucket(bucket: &str) { let pending_buckets = { let mut dirty_buckets = dirty_usage_buckets(); + let mut dirty_scopes = dirty_usage_bucket_scopes(); let generation = advance_generation(&DIRTY_USAGE_BUCKET_GENERATION); dirty_buckets.insert(bucket.to_string(), generation); + dirty_scopes.insert(bucket.to_string(), DirtyUsageBucketScope::WholeBucket); dirty_buckets.len() }; global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending_buckets)); @@ -193,6 +256,56 @@ pub fn record_dirty_usage_bucket(bucket: &str) { DIRTY_USAGE_BUCKET_NOTIFY.notify_one(); } +/// Record a mutation whose affected top-level namespace entry is known. +/// +/// Object names that cannot be represented as one safe direct child retain the +/// conservative whole-bucket marker. This journal is intentionally process +/// local: after restart or any unverified distributed path the scanner falls +/// back to its ordinary bucket scan. +pub fn record_dirty_usage_object(bucket: &str, object: &str) { + let Some(top_level_entry) = dirty_usage_top_level_entry(object) else { + record_dirty_usage_bucket(bucket); + return; + }; + if bucket.is_empty() { + return; + } + + let pending_buckets = { + let mut dirty_buckets = dirty_usage_buckets(); + let mut dirty_scopes = dirty_usage_bucket_scopes(); + let generation = advance_generation(&DIRTY_USAGE_BUCKET_GENERATION); + dirty_buckets.insert(bucket.to_string(), generation); + let scope = dirty_scopes + .entry(bucket.to_string()) + .or_insert_with(|| DirtyUsageBucketScope::TopLevelEntries(HashSet::new())); + let overflowed = match scope { + DirtyUsageBucketScope::WholeBucket => false, + DirtyUsageBucketScope::TopLevelEntries(entries) => { + entries.insert(top_level_entry); + entries.len() > MAX_DIRTY_USAGE_TOP_LEVEL_ENTRIES_PER_BUCKET + } + }; + if overflowed { + *scope = DirtyUsageBucketScope::WholeBucket; + } + dirty_buckets.len() + }; + global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending_buckets)); + crate::prefix_usage::invalidate_prefix_usage_cache(bucket); + DIRTY_USAGE_BUCKET_NOTIFY.notify_one(); +} + +fn dirty_usage_top_level_entry(object: &str) -> Option { + let (top_level_entry, _) = object.split_once('/').unwrap_or((object, "")); + (!top_level_entry.is_empty() + && top_level_entry != "." + && top_level_entry != ".." + && !object.starts_with('/') + && !top_level_entry.contains(['\\', '\0'])) + .then(|| top_level_entry.to_string()) +} + pub fn record_scanner_maintenance_change(bucket: &str) { if bucket.is_empty() { return; @@ -262,6 +375,7 @@ pub fn acknowledge_dirty_usage_generation( let (cleared_buckets, pending_buckets) = { let mut dirty_buckets = dirty_usage_buckets(); + let mut dirty_scopes = dirty_usage_bucket_scopes(); let current_generation = DIRTY_USAGE_BUCKET_GENERATION.load(Ordering::Acquire); if generation == 0 || generation == u64::MAX || current_generation == u64::MAX || generation > current_generation { return Err(ScannerDirtyUsageAckError::InvalidGeneration); @@ -269,6 +383,7 @@ pub fn acknowledge_dirty_usage_generation( let before = dirty_buckets.len(); dirty_buckets.retain(|_, dirty_generation| *dirty_generation > generation); + dirty_scopes.retain(|bucket, _| dirty_buckets.contains_key(bucket)); let cleared_buckets = before.saturating_sub(dirty_buckets.len()); if cleared_buckets > 0 { advance_generation(&DIRTY_USAGE_BUCKET_GENERATION); @@ -291,7 +406,9 @@ pub fn clear_dirty_usage_bucket(bucket: &str) { let pending_buckets = { let mut dirty_buckets = dirty_usage_buckets(); + let mut dirty_scopes = dirty_usage_bucket_scopes(); dirty_buckets.remove(bucket); + dirty_scopes.remove(bucket); advance_generation(&DIRTY_USAGE_BUCKET_GENERATION); dirty_buckets.len() }; @@ -299,8 +416,9 @@ pub fn clear_dirty_usage_bucket(bucket: &str) { } pub(super) fn snapshot_dirty_usage_buckets(buckets: &[BucketInfo], absent_generation_cutoff: u64) -> DirtyUsageSnapshot { - let (snapshot, generation, covers_all_pending) = { + let (snapshot, scopes, generation, covers_all_pending) = { let dirty_buckets = dirty_usage_buckets(); + let dirty_scopes = dirty_usage_bucket_scopes(); let listed_buckets = dirty_buckets .values() .any(|generation| *generation > absent_generation_cutoff) @@ -315,13 +433,26 @@ pub(super) fn snapshot_dirty_usage_buckets(buckets: &[BucketInfo], absent_genera }) .map(|(bucket, generation)| (bucket.clone(), *generation)) .collect::(); + let scopes = snapshot + .keys() + .map(|bucket| { + ( + bucket.clone(), + dirty_scopes + .get(bucket) + .cloned() + .unwrap_or(DirtyUsageBucketScope::WholeBucket), + ) + }) + .collect::(); let generation = DIRTY_USAGE_BUCKET_GENERATION.load(Ordering::Acquire); let covers_all_pending = generation == absent_generation_cutoff && snapshot.len() == dirty_buckets.len(); - (snapshot, generation, covers_all_pending) + (snapshot, scopes, generation, covers_all_pending) }; global_metrics().record_scanner_dirty_usage_cycle_snapshot(usize_to_u64_saturated(snapshot.len())); DirtyUsageSnapshot { buckets: Arc::new(snapshot), + scopes: Arc::new(scopes), generation, covers_all_pending, } @@ -338,10 +469,12 @@ pub(crate) async fn dirty_usage_bucket_notified() { pub(super) fn clear_dirty_usage_buckets(snapshot: &DirtyUsageBuckets) { let (cleared_buckets, pending_buckets) = { let mut dirty_buckets = dirty_usage_buckets(); + let mut dirty_scopes = dirty_usage_bucket_scopes(); let mut cleared_buckets = 0usize; for (bucket, generation) in snapshot { if dirty_buckets.get(bucket).is_some_and(|current| current == generation) { dirty_buckets.remove(bucket); + dirty_scopes.remove(bucket); cleared_buckets += 1; } } @@ -443,9 +576,15 @@ pub(super) fn dirty_usage_bucket_count() -> usize { #[cfg(test)] pub(crate) fn clear_dirty_usage_buckets_for_tests() { dirty_usage_buckets().clear(); + dirty_usage_bucket_scopes().clear(); } #[cfg(test)] pub(crate) fn dirty_usage_buckets_for_tests() -> DirtyUsageBuckets { dirty_usage_buckets().clone() } + +#[cfg(test)] +pub(crate) fn dirty_usage_bucket_scopes_for_tests() -> DirtyUsageBucketScopes { + dirty_usage_bucket_scopes().clone() +} diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index aad6403c6..df28c55f0 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -585,6 +585,7 @@ impl ScannerIOCache for SetDisks { let partial_dirty_buckets_clone = bucket_failures.partial.clone(); let pending_maintenance_work_clone = pending_maintenance_work.clone(); let dirty_usage_buckets_clone = dirty_usage_buckets.clone(); + let scope_clone = scope.clone(); let cache_cycle_floor_clone = cache_cycle_floor.clone(); let expected_publication_epoch_clone = expected_publication_epoch; let remote_server_epoch = match worker_mode { @@ -619,6 +620,12 @@ impl ScannerIOCache for SetDisks { }; let mut work_guard = BucketWorkGuard::new(remaining_bucket_work_clone.clone(), bucket_work_complete_clone.clone()); + // Prefix hints are process-local. Never hand one to a + // remote or legacy-coordinator disk path. + let prefix_scan_scope = disk_clone + .is_local() + .then(|| scope_clone.prefix_scope_for(&bucket.name)) + .flatten(); metrics::histogram!( METRIC_SCANNER_DISK_SCAN_WAIT_SECONDS, @@ -1054,7 +1061,10 @@ impl ScannerIOCache for SetDisks { set_disk_inventory_clone.as_ref().clone(), cache.clone(), None, - scan_mode, + ScannerDiskScanOptions { + scan_mode, + prefix_scan_scope, + }, ); tokio::pin!(scan); let mut lock_watch = tokio::time::interval(SCANNER_CACHE_LOCK_POLL_INTERVAL); diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index d4094a311..13a2d7b23 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -168,6 +168,7 @@ where scoped_scan_scope_from_dirty_buckets( resolution.requested_scope, dirty_buckets, + (!distributed).then_some(resolution.dirty_usage_snapshot.scopes.as_ref()), true, resolution.all_buckets, resolution.baseline_proof, diff --git a/crates/scanner/src/scanner_io/io_disk.rs b/crates/scanner/src/scanner_io/io_disk.rs index 2bb55b55c..2c6510788 100644 --- a/crates/scanner/src/scanner_io/io_disk.rs +++ b/crates/scanner/src/scanner_io/io_disk.rs @@ -146,7 +146,7 @@ impl ScannerIODisk for Disk { Ok(size_summary) } - #[tracing::instrument(skip(self, budget, updates, cache, set_disks))] + #[tracing::instrument(skip(self, budget, updates, cache, set_disks, options), fields(scan_mode = ?options.scan_mode))] async fn nsscanner_disk( self: Arc, ctx: CancellationToken, @@ -154,8 +154,12 @@ impl ScannerIODisk for Disk { set_disks: Vec>, cache: DataUsageCache, updates: Option>, - scan_mode: HealScanMode, + options: ScannerDiskScanOptions, ) -> Result { + let ScannerDiskScanOptions { + scan_mode, + prefix_scan_scope, + } = options; let done_drive = Metrics::time(Metric::ScanBucketDrive); let drive_start = std::time::Instant::now(); let bucket = cache.info.name.clone(); @@ -198,7 +202,19 @@ impl ScannerIODisk for Disk { cache.info.object_lock = Some(Arc::new(object_lock_config)); } - let result = scan_data_folder( + // Prefix reuse never crosses semantic maintenance boundaries. A + // lifecycle, replication, Object Lock, or erasure health walk can + // make a clean data subtree require scanner-side work even without a + // direct object mutation in the local journal. The folder scanner + // separately rejects scopes in erasure mode. + let prefix_scan_scope = (scan_mode == HealScanMode::Normal + && cache.info.lifecycle.is_none() + && cache.info.replication.is_none() + && cache.info.object_lock.is_none()) + .then_some(prefix_scan_scope) + .flatten(); + + let result = scan_data_folder_scoped( ctx.clone(), budget, set_disks, @@ -207,6 +223,7 @@ impl ScannerIODisk for Disk { updates, scan_mode, SCANNER_SLEEPER.clone(), + prefix_scan_scope, ) .await; diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 1d58dc88e..06441f92b 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -12,7 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::dirty_usage::{clear_dirty_usage_buckets_for_tests, dirty_usage_buckets_for_tests}; +use super::dirty_usage::{ + DirtyUsageBucketScope, clear_dirty_usage_buckets_for_tests, dirty_usage_bucket_scopes_for_tests, + dirty_usage_buckets_for_tests, +}; use super::io_disk::tier_stats_template; use super::*; use crate::scanner_budget::ScannerCycleBudgetConfig; @@ -374,6 +377,7 @@ async fn scoped_scan_production_entry_preserves_deep_and_full_maintenance_work() let requested_scope = if explicit_scope { ScannerBucketScanScope::from_dirty_buckets( HashSet::from(["hot-bucket".to_string()]), + HashMap::new(), DataUsageScanPlanDigest([7; 32]), ) } else { @@ -919,6 +923,47 @@ fn dirty_usage_snapshot_is_sorted_and_reports_its_cutoff() { clear_dirty_usage_buckets_for_tests(); } +#[test] +#[serial] +fn dirty_usage_object_marks_only_its_top_level_entry_until_the_scope_becomes_ambiguous() { + clear_dirty_usage_buckets_for_tests(); + + record_dirty_usage_object("photos", "2026/january/object-a"); + record_dirty_usage_object("photos", "archive/object-b"); + let scopes = dirty_usage_bucket_scopes_for_tests(); + assert_eq!( + scopes.get("photos"), + Some(&DirtyUsageBucketScope::TopLevelEntries(HashSet::from([ + "2026".to_string(), + "archive".to_string(), + ]))) + ); + drop(scopes); + + record_dirty_usage_object("photos", "../ambiguous"); + assert_eq!( + dirty_usage_bucket_scopes_for_tests().get("photos"), + Some(&DirtyUsageBucketScope::WholeBucket) + ); + clear_dirty_usage_buckets_for_tests(); +} + +#[test] +#[serial] +fn dirty_usage_object_expands_an_overfull_prefix_journal_to_the_whole_bucket() { + clear_dirty_usage_buckets_for_tests(); + + for index in 0..129 { + record_dirty_usage_object("photos", &format!("prefix-{index}/object")); + } + + assert_eq!( + dirty_usage_bucket_scopes_for_tests().get("photos"), + Some(&DirtyUsageBucketScope::WholeBucket) + ); + clear_dirty_usage_buckets_for_tests(); +} + #[test] #[serial] fn dirty_usage_snapshot_marks_truncated_results_incomplete() { @@ -1500,6 +1545,7 @@ fn scoped_scan_selects_only_current_dirty_buckets_after_baseline_validation() { let scope = scoped_scan_scope_from_dirty_buckets( ScannerBucketScanScope::default(), HashSet::from(["photos".to_string(), "deleted".to_string()]), + None, true, &[bucket_info("photos")], ScannerCacheBaselineProof { @@ -1554,6 +1600,56 @@ fn scoped_scan_baseline_work_proof_requires_uniform_known_set_identity() { } } +#[test] +fn scoped_scan_uses_only_locally_verified_prefix_hints() { + let source = DataUsageCacheSource::new(1, 2); + let expected_sources = HashSet::from([source]); + let scan_plan_digest = DataUsageScanPlanDigest([6; 32]); + let baseline = complete_usage_baseline(source, scan_plan_digest, 7, 11); + let dirty_scopes = HashMap::from([ + ( + "photos".to_string(), + DirtyUsageBucketScope::TopLevelEntries(HashSet::from(["2026".to_string()])), + ), + ("videos".to_string(), DirtyUsageBucketScope::WholeBucket), + ]); + + let locally_scoped = scoped_scan_scope_from_dirty_buckets( + ScannerBucketScanScope::default(), + HashSet::from(["photos".to_string(), "videos".to_string()]), + Some(&dirty_scopes), + true, + &[bucket_info("photos"), bucket_info("videos")], + ScannerCacheBaselineProof { + authoritative_data: Some(&baseline), + observed_candidate_data: None, + expected_sources: &expected_sources, + leader_epoch: 11, + want_cycle: 8, + scan_plan_digest, + }, + ); + assert!(locally_scoped.prefix_scope_for("photos").is_some()); + assert!(locally_scoped.prefix_scope_for("videos").is_none()); + + let distributed_scope = scoped_scan_scope_from_dirty_buckets( + ScannerBucketScanScope::default(), + HashSet::from(["photos".to_string(), "videos".to_string()]), + None, + true, + &[bucket_info("photos"), bucket_info("videos")], + ScannerCacheBaselineProof { + authoritative_data: Some(&baseline), + observed_candidate_data: None, + expected_sources: &expected_sources, + leader_epoch: 11, + want_cycle: 8, + scan_plan_digest, + }, + ); + assert!(distributed_scope.prefix_scope_for("photos").is_none()); +} + fn peer_dirty_usage_snapshot( instance_id: &str, generation: u64, @@ -1669,6 +1765,7 @@ fn scoped_set_scan_rebuilds_selected_buckets_and_drops_deleted_buckets() { &all_buckets, &ScannerBucketScanScope { selected_buckets: Some(selected_buckets), + selected_bucket_prefixes: None, baseline_scan_plan_digest: Some(baseline_digest), }, ScannerSetCacheGeneration { @@ -1707,6 +1804,7 @@ fn scoped_set_scan_rejects_unbound_bucket_incarnations() { let old_cache = complete_set_usage_cache(&[("stable", 10), ("dirty", 20)], baseline_digest); let scope = ScannerBucketScanScope { selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))), + selected_bucket_prefixes: None, baseline_scan_plan_digest: Some(baseline_digest), }; let generation = ScannerSetCacheGeneration { @@ -1744,6 +1842,7 @@ fn scoped_set_scan_falls_back_when_an_unselected_bucket_has_no_baseline() { &all_buckets, &ScannerBucketScanScope { selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))), + selected_bucket_prefixes: None, baseline_scan_plan_digest: Some(baseline_digest), }, ScannerSetCacheGeneration { @@ -1764,6 +1863,7 @@ fn scoped_set_scan_requires_an_exact_complete_baseline() { let all_buckets = vec![bucket_info_with_created_time("dirty")]; let scope = ScannerBucketScanScope { selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))), + selected_bucket_prefixes: None, baseline_scan_plan_digest: Some(baseline_digest), }; let generation = ScannerSetCacheGeneration { @@ -1792,6 +1892,7 @@ fn scoped_set_scan_requires_an_exact_complete_baseline() { let empty_scope = ScannerBucketScanScope { selected_buckets: Some(Arc::new(HashSet::new())), + selected_bucket_prefixes: None, baseline_scan_plan_digest: Some(baseline_digest), }; let complete = complete_set_usage_cache(&[("dirty", 10)], baseline_digest); diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index fc81b3530..501fdf83a 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -532,7 +532,7 @@ impl DefaultMultipartUsecase { .await { Ok(_) => { - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_object(&bucket, &key); Ok(S3Response::new(AbortMultipartUploadOutput { ..Default::default() })) } Err(err) => { diff --git a/rustfs/src/app/object/copy.rs b/rustfs/src/app/object/copy.rs index 13393f797..76192ca9d 100644 --- a/rustfs/src/app/object/copy.rs +++ b/rustfs/src/app/object/copy.rs @@ -814,7 +814,7 @@ impl DefaultObjectUsecase { } } - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_object(&bucket, &key); Ok::<_, S3Error>((oi, dest_versioned)) } }); diff --git a/rustfs/src/app/object/delete.rs b/rustfs/src/app/object/delete.rs index c797335fb..ad7851590 100644 --- a/rustfs/src/app/object/delete.rs +++ b/rustfs/src/app/object/delete.rs @@ -1143,7 +1143,7 @@ impl DefaultObjectUsecase { let manager = get_capacity_manager(); manager.record_write_operation().await; let _ = helper.complete(&result); - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_object(&bucket, &key); result } } diff --git a/rustfs/src/app/object/internal_put.rs b/rustfs/src/app/object/internal_put.rs index eb93d2610..7f24c6e5c 100644 --- a/rustfs/src/app/object/internal_put.rs +++ b/rustfs/src/app/object/internal_put.rs @@ -689,7 +689,7 @@ impl DefaultObjectUsecase { schedule_object_replication(obj_info.clone(), store, completion_replication_decision).await; } - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_object(&bucket, &key); Ok::<_, ApiError>(obj_info) } }); diff --git a/rustfs/src/app/object/put.rs b/rustfs/src/app/object/put.rs index f1873beea..f5eb8d0f4 100644 --- a/rustfs/src/app/object/put.rs +++ b/rustfs/src/app/object/put.rs @@ -2033,7 +2033,7 @@ impl DefaultObjectUsecase { schedule_object_replication(obj_info.clone(), store, dsc).await; } - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_object(&bucket, &key); rustfs_io_metrics::record_put_object_stage_duration_from("app_post_store_bookkeeping", post_store_stage_start); let capacity_update_stage_start = put_stage_metrics_enabled.then(Instant::now); diff --git a/rustfs/src/app/object/restore.rs b/rustfs/src/app/object/restore.rs index 96201a670..506826281 100644 --- a/rustfs/src/app/object/restore.rs +++ b/rustfs/src/app/object/restore.rs @@ -420,7 +420,7 @@ impl DefaultObjectUsecase { ) .await .map_err(ApiError::from)?; - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_object(&bucket, &object); #[cfg(test)] maybe_pause_after_restore_status_commit(&bucket, &object).await; drop(superseded_worker_guard.take()); @@ -494,7 +494,7 @@ impl DefaultObjectUsecase { err.to_string() ); } else { - rustfs_scanner::record_dirty_usage_bucket(&bucket_clone); + rustfs_scanner::record_dirty_usage_object(&bucket_clone, &object_clone); debug!(bucket = %bucket_clone, object = %object_clone, "Transitioned object restored"); } });