From 934238cb486609fe56e32115f1baf34db7bd94d4 Mon Sep 17 00:00:00 2001 From: Hauser Date: Tue, 27 Oct 2026 00:53:01 +0800 Subject: [PATCH] perf(scanner): compact completed leaf entries in place Keep completed-folder eligibility and frontier handling in one compaction boundary. Borrow leaf statistics and update the cached sampling flag without cloning, deleting, or reinserting the entry; aggregate only non-leaf nodes. Pin allocation identity, complete metadata, failure guards and caller state with a regression that fails under the previous compaction implementation. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/scanner/src/scanner_folder.rs | 97 +++++++++++-------- .../completed_compaction.rs | 59 +++++++++++ 2 files changed, 114 insertions(+), 42 deletions(-) diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index b5242a98d..2f9d31ff1 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::borrow::Cow; use std::collections::{HashMap, HashSet}; use std::fs::FileType; use std::io::ErrorKind; @@ -1671,6 +1672,59 @@ impl FolderScanner { self.data_usage_scanner_debug = debug; } + fn compact_completed_folder(&mut self, folder: &CachedFolder, hash: &DataUsageHash, entry: &DataUsageEntry) { + // Completed healthy subtrees can keep aggregate usage. Coverage gaps + // and failed-object debt still need child entries for precise resume + // and to avoid rereading healthy siblings on a retry. + let forward_sweep = self.old_cache.info.scan_progress.is_some(); + if entry.compacted || self.new_cache.info.name == folder.name || (forward_sweep && self.coverage_gap) { + return; + } + let summary = if entry.children.is_empty() { + Cow::Borrowed(entry) + } else if let Some(flat) = self.new_cache.size_recursive(&hash.0) { + Cow::Owned(flat) + } else { + return; + }; + if forward_sweep && summary.failed_objects != 0 { + return; + } + let should_compact = summary.objects < DATA_SCANNER_COMPACT_LEAST_OBJECT + || entry.children.iter().all(|key| { + self.new_cache + .cache + .get(key) + .is_none_or(|child| child.children.is_empty() && child.objects <= 1) + }); + if !should_compact { + return; + } + // Heal revisits also reach this path without a parent advancing + // coverage. Keep the endpoint addressable when its completed + // subtree is replaced by an aggregate. + if self.coverage_frontier.as_deref().is_some_and(|frontier| { + frontier + .strip_prefix(&folder.name) + .is_some_and(|suffix| suffix.starts_with('/')) + }) { + self.record_completed_child(&folder.name); + } + match summary { + Cow::Borrowed(_) => { + // The leaf is already cached; only its sampling mode changes. + if let Some(cached) = self.new_cache.cache.get_mut(&hash.0) { + cached.compacted = true; + } + } + Cow::Owned(mut flat) => { + flat.compacted = true; + self.new_cache.delete_recursive(hash); + self.new_cache.replace_hashed(hash, &folder.parent, &flat); + } + } + } + fn refresh_usage_subtree(&mut self, hash: &DataUsageHash, parent: &Option) { if let Some(updates) = &mut self.usage_updates { updates.cache.delete_recursive(hash); @@ -3025,48 +3079,7 @@ impl FolderScanner { self.new_cache.replace_hashed(&this_hash, &folder.parent, into); } - // Completed healthy subtrees can keep aggregate usage. Coverage gaps - // and failed-object debt still need child entries for precise resume - // and to avoid rereading healthy siblings on a retry. - if !into.compacted - && (self.old_cache.info.scan_progress.is_none() || !self.coverage_gap) - && self.new_cache.info.name != folder.name - && let Some(mut flat) = self.new_cache.size_recursive(&this_hash.key()) - && (self.old_cache.info.scan_progress.is_none() || flat.failed_objects == 0) - { - flat.compacted = true; - let mut should_compact = false; - - if flat.objects < DATA_SCANNER_COMPACT_LEAST_OBJECT { - should_compact = true; - } else { - // Compact if we only have objects as children... - should_compact = true; - for k in &into.children { - if let Some(v) = self.new_cache.cache.get(k) - && (!v.children.is_empty() || v.objects > 1) - { - should_compact = false; - break; - } - } - } - - if should_compact { - // Heal revisits also reach this path without a parent advancing - // coverage. Keep the endpoint addressable when its completed - // subtree is replaced by an aggregate. - if self.coverage_frontier.as_deref().is_some_and(|frontier| { - frontier - .strip_prefix(&folder.name) - .is_some_and(|suffix| suffix.starts_with('/')) - }) { - self.record_completed_child(&folder.name); - } - self.new_cache.delete_recursive(&this_hash); - self.new_cache.replace_hashed(&this_hash, &folder.parent, &flat); - } - } + self.compact_completed_folder(&folder, &this_hash, into); // Compact if too many children... if !into.compacted { diff --git a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/completed_compaction.rs b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/completed_compaction.rs index 3357f28ab..9763c770c 100644 --- a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/completed_compaction.rs +++ b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/completed_compaction.rs @@ -3,6 +3,65 @@ use super::*; +#[tokio::test] +async fn completed_leaf_compaction_reuses_cached_storage() { + let (mut scanner, root) = build_test_scanner().await; + let _cleanup = TestGuard { temp_dir: Some(root) }; + scanner.old_cache = bound_checkpoint().0; + let folder = CachedFolder { + name: "bucket/object".to_string(), + parent: Some(hash_path("bucket")), + object_heal_prob_div: 1, + }; + let hash = hash_path(&folder.name); + + for (coverage_gap, failed_objects, should_compact) in [(false, 0, true), (true, 0, false), (false, 1, false)] { + scanner.coverage_gap = coverage_gap; + scanner.new_cache = DataUsageCache { + info: scanner.old_cache.info.clone(), + ..Default::default() + }; + let mut target = String::with_capacity(4096); + target.push_str("arn:test:replica"); + let allocation = target.as_ptr(); + let capacity = target.capacity(); + let mut leaf = DataUsageEntry { + size: 2048, + objects: 1, + versions: 3, + failed_objects, + replication_stats: Some(rustfs_data_usage::ReplicationAllStats { + targets: HashMap::from([(target, rustfs_data_usage::ReplicationTargetUsage::default())]), + ..Default::default() + }), + all_tier_stats: Some(rustfs_data_usage::AllTierStats::new()), + ..Default::default() + }; + leaf.obj_sizes.add(2048); + leaf.obj_versions.add(3); + let accumulator = leaf.clone(); + let mut expected = serde_json::to_value(&leaf).expect("leaf metadata"); + expected["compacted"] = serde_json::json!(should_compact); + scanner.new_cache.replace(&folder.name, "bucket", leaf); + + scanner.compact_completed_folder(&folder, &hash, &accumulator); + + let cached = scanner.new_cache.find(&folder.name).expect("retained leaf"); + assert_eq!(serde_json::to_value(cached).expect("compacted metadata"), expected); + let target = cached + .replication_stats + .as_ref() + .expect("retained replication statistics") + .targets + .keys() + .next() + .expect("retained replication target"); + assert_eq!(target.capacity(), capacity, "leaf compaction must preserve owned allocations"); + assert_eq!(target.as_ptr(), allocation); + assert!(!accumulator.compacted, "cache compaction must not change the caller's accounting mode"); + } +} + async fn scan( disk: &Arc, cache: DataUsageCache,