diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 19703c262..795cb8cc9 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -1832,7 +1832,12 @@ impl FolderScanner { } } FolderScanSource::Existing => { - if !forward_sweep && !into.compacted && self.old_cache.is_compacted(&h) { + // Usage sampling is not proof that a Deep check ran. + if self.scan_mode != HealScanMode::Deep + && !forward_sweep + && !into.compacted + && self.old_cache.is_compacted(&h) + { let next_cycle = self.old_cache.info.next_cycle as u32; if !h.mod_(next_cycle, data_usage_update_dir_cycles()) { // Transfer and add as child... diff --git a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs index 7052e9de4..45bddb6ab 100644 --- a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs +++ b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs @@ -20,6 +20,7 @@ use crate::{DataUsageCacheSource, DataUsageScanPlanDigest}; use std::io::Cursor; use tokio::io::AsyncReadExt; +mod deep_compacted; mod segment_observation; const CACHE_NAME: &str = "bucket/checkpoint-fixture.bin"; 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 new file mode 100644 index 000000000..cc2013dc6 --- /dev/null +++ b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/deep_compacted.rs @@ -0,0 +1,204 @@ +// Copyright 2026 RustFS Team +// Licensed under the Apache License, Version 2.0. + +use super::*; + +const PREFIX: &str = "bucket/prefix"; +const OBJECTS: u64 = 4; + +struct CompactedFixture { + disk: Arc, + root: std::path::PathBuf, + cache: DataUsageCache, + identity: crate::DataUsageScanIdentity, + store: Arc, + _cleanup: TestGuard, +} + +async fn scan( + disk: &Arc, + cache: DataUsageCache, + mode: HealScanMode, + max_objects: u64, +) -> (ScannerDiskScanOutcome, Arc) { + let budget = ScannerCycleBudget::new_with_progress_tracking( + &CancellationToken::new(), + ScannerCycleBudgetConfig { + max_objects: Some(max_objects), + ..Default::default() + }, + ); + let outcome = disk + .clone() + .nsscanner_disk(budget.token(), budget.clone(), vec![disk.clone()], cache, None, mode) + .await + .expect("bounded real disk scan"); + (outcome, budget) +} + +async fn save_reload(store: &Arc, cache: &DataUsageCache) -> DataUsageCache { + let revisions = DataUsageCache::default() + .load_with_revisions(store.clone(), CACHE_NAME) + .await + .expect("fixture save revisions"); + cache + .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0) + .await + .expect("save compacted checkpoint through real codec and revision checks"); + let loaded = store.strict_load().await; + assert_eq!(loaded.info.snapshot_complete, cache.info.snapshot_complete); + assert_eq!(loaded.info.scan_checkpoint, cache.info.scan_checkpoint); + assert_eq!(loaded.info.scan_identity, cache.info.scan_identity); + assert_eq!(loaded.info.scan_plan_digest, cache.info.scan_plan_digest); + assert_eq!( + loaded.checked_flatten("bucket").expect("reloaded root").size, + cache.checked_flatten("bucket").expect("returned root").size + ); + loaded +} + +impl CompactedFixture { + async fn new(mode: HealScanMode) -> Self { + let (scanner, root) = build_test_scanner().await; + let cleanup = TestGuard { + temp_dir: Some(root.clone()), + }; + for index in 0..OBJECTS { + write_checkpoint_object(&root, &format!("prefix/{index:04}"), &[(None, 1)]).await; + } + let identity = crate::DataUsageScanIdentity { + scan_mode: mode, + tier_registry_generation: crate::runtime_tier_registry_for_cycle(11, 7).await.generation, + ..bound_checkpoint().1 + }; + let mut cache = DataUsageCache::default(); + // The first real scan builds coverage; the second same-plan scan takes + // the normal compaction path. No synthetic compacted cache is injected. + for cycle in [11, 12] { + cache.prepare_bucket_checkpoint("bucket", cycle, 7, SOURCE, PLAN, identity); + cache.info.skip_healing = true; + let (outcome, budget) = scan(&scanner.local_disk, cache, mode, OBJECTS + 1).await; + let ScannerDiskScanOutcome::Complete(completed) = outcome else { + panic!("fixture baseline must complete") + }; + assert_eq!(budget.progress().0, OBJECTS, "baseline must read every metadata object"); + cache = completed; + } + let store = FixtureStore::new(); + cache = save_reload(&store, &cache).await; + assert!(cache.find(PREFIX).expect("baseline prefix").compacted); + assert_eq!(cache.checked_flatten("bucket").expect("complete baseline").size, 4); + assert!(cache.info.scan_progress.is_none()); + Self { + disk: scanner.local_disk, + root, + cache, + identity, + store, + _cleanup: cleanup, + } + } + + fn next_cycle(&self, sampled: bool) -> u64 { + let first = self.cache.info.next_cycle + 1; + (first..first + 16) + .find(|cycle| hash_path(PREFIX).mod_(u32::try_from(*cycle).expect("bounded cycle"), 16) == sampled) + .expect("one selected cycle and non-selected cycles exist within the fixed rotation") + } + + fn prepare(&mut self, cycle: u64) { + let state = crate::scanner_io::current_cache_root_or_prepare_with_generation( + &mut self.cache, + "bucket", + SOURCE, + cycle, + 7, + PLAN, + crate::scanner_io::DataUsageCacheReuseOptions { + checkpoint_identity: Some(self.identity), + ..Default::default() + }, + ); + assert!(matches!(state, crate::scanner_io::DataUsageCacheScanState::Prepared { .. })); + assert_eq!(self.cache.info.scan_identity, Some(self.identity)); + assert_eq!(self.cache.info.scan_plan_digest, Some(PLAN)); + assert!( + self.cache.info.scan_progress.is_none(), + "same-strength complete baseline uses the existing tree" + ); + assert!(self.cache.find(PREFIX).expect("prepared prefix").compacted); + } + + async fn change_metadata_without_activity_event(&self) { + // This models a local metadata change not announced by a segment + // producer. Deep traversal must not depend on a usage-clean signal. + // Healing is disabled: the oracle proves metadata re-entry, not repair. + write_checkpoint_object(&self.root, "prefix/0000", &[(None, 7)]).await; + } +} + +#[tokio::test] +#[serial] +async fn deep_compacted_same_plan_rechecks_unsampled_prefix() { + temp_env::async_with_vars([(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, Some("16"))], async { + let mut fixture = CompactedFixture::new(HealScanMode::Deep).await; + fixture.change_metadata_without_activity_event().await; + fixture.prepare(fixture.next_cycle(false)); + let (outcome, budget) = scan(&fixture.disk, fixture.cache.clone(), HealScanMode::Deep, OBJECTS + 1).await; + assert_eq!( + budget.progress().0, + OBJECTS, + "Deep must inspect compacted children even outside the usage sample cycle" + ); + let ScannerDiskScanOutcome::Complete(cache) = outcome else { + panic!("bounded Deep scan must complete") + }; + let loaded = save_reload(&fixture.store, &cache).await; + let root = loaded.checked_flatten("bucket").expect("Deep scan root"); + assert_eq!((root.objects, root.size), (4, 10), "Deep must observe the changed metadata"); + }) + .await; +} + +#[tokio::test] +#[serial] +async fn deep_compacted_normal_scan_preserves_periodic_sampling() { + temp_env::async_with_vars([(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, Some("16"))], async { + let mut fixture = CompactedFixture::new(HealScanMode::Normal).await; + fixture.change_metadata_without_activity_event().await; + fixture.prepare(fixture.next_cycle(false)); + let (outcome, budget) = scan(&fixture.disk, fixture.cache.clone(), HealScanMode::Normal, OBJECTS + 1).await; + assert_eq!(budget.progress().0, 0, "Normal retains its existing unsampled-subtree policy"); + let ScannerDiskScanOutcome::Complete(cache) = outcome else { panic!("normal sampling completes") }; + assert_eq!(cache.checked_flatten("bucket").expect("sampled root").size, 4); + fixture.cache = save_reload(&fixture.store, &cache).await; + fixture.prepare(fixture.next_cycle(true)); + let (outcome, budget) = scan(&fixture.disk, fixture.cache.clone(), HealScanMode::Normal, OBJECTS + 1).await; + assert_eq!(budget.progress().0, OBJECTS); + let ScannerDiskScanOutcome::Complete(cache) = outcome else { + panic!("selected Normal rotation completes") + }; + assert_eq!(cache.checked_flatten("bucket").expect("refreshed root").size, 10); + }) + .await; +} + +#[tokio::test] +#[serial] +async fn deep_compacted_budget_preserves_partial_checkpoint() { + temp_env::async_with_vars([(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, Some("16"))], async { + let mut fixture = CompactedFixture::new(HealScanMode::Deep).await; + fixture.prepare(fixture.next_cycle(false)); + let (outcome, budget) = scan(&fixture.disk, fixture.cache.clone(), HealScanMode::Deep, 2).await; + assert_eq!(budget.progress().0, 2); + assert_eq!(budget.reason(), Some(ScannerCycleBudgetReason::Objects)); + let ScannerDiskScanOutcome::Partial(cache) = outcome else { + panic!("Deep must retain budget interruption as partial") + }; + assert!(!cache.info.snapshot_complete); + let loaded = save_reload(&fixture.store, &cache).await; + assert!(!loaded.info.snapshot_complete); + assert!(loaded.checked_flatten("bucket").expect("partial root").objects > 0); + }) + .await; +}