From f64cb816dbf411209edefa7019dfd90c177b817c Mon Sep 17 00:00:00 2001 From: Hauser Date: Sat, 17 Oct 2026 23:33:47 +0800 Subject: [PATCH] fix(scanner): preserve checkpoint progress across object failures (#8642) Separate completed traversal from persisted object health debt so leader handoffs reuse healthy subtrees and retry failures without losing progress. Fence old readers with a new receipt digest domain and preserve exact accounting under diagnostic eviction, compaction and punctuation ordering. Co-authored-by: heihutu Co-authored-by: zhi22915 --- crates/scanner/src/data_usage_define.rs | 27 ++- crates/scanner/src/data_usage_define/tests.rs | 82 +++++-- crates/scanner/src/scanner_folder.rs | 83 ++++++-- crates/scanner/src/scanner_folder/tests.rs | 5 +- .../checkpoint_fixture/periodic_snapshot.rs | 201 +++++++++++++++++- .../architecture/scanner-usage-publication.md | 18 ++ 6 files changed, 368 insertions(+), 48 deletions(-) diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 4fe28a3c6..7c033b6ca 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -514,6 +514,14 @@ impl Serialize for DataUsageScanCoverageReceipt { } } +/// Coverage follows depth-first directory traversal, rather than raw path order. +/// A completed subtree includes its descendants but never an unfinished ancestor. +pub(crate) fn scan_frontier_covers_path(path: &str, frontier: &str) -> bool { + let ancestor = frontier.strip_prefix(path).is_some_and(|suffix| suffix.starts_with('/')); + let descendant = path.strip_prefix(frontier).is_some_and(|suffix| suffix.starts_with('/')); + (path.split('/').cmp(frontier.split('/')).is_le() && !ancestor) || descendant +} + struct CheckpointDigestWriter(Sha256); impl std::io::Write for CheckpointDigestWriter { @@ -1066,17 +1074,22 @@ impl DataUsageCache { } fn coverage_prefix_digest(&self, through: &str) -> Result<[u8; 32], serde_json::Error> { - self.coverage_prefix_digest_with_epoch(through, None) + self.coverage_prefix_digest_with_epoch(through, None, 3) + } + + fn v2_coverage_prefix_digest(&self, through: &str) -> Result<[u8; 32], serde_json::Error> { + self.coverage_prefix_digest_with_epoch(through, None, 2) } fn legacy_coverage_prefix_digest(&self, through: &str) -> Result<[u8; 32], serde_json::Error> { - self.coverage_prefix_digest_with_epoch(through, Some(self.info.leader_epoch)) + self.coverage_prefix_digest_with_epoch(through, Some(self.info.leader_epoch), 0) } fn coverage_prefix_digest_with_epoch( &self, through: &str, legacy_leader_epoch: Option, + domain_version: u8, ) -> Result<[u8; 32], serde_json::Error> { let mut writer = CheckpointDigestWriter(Sha256::new()); if let Some(leader_epoch) = legacy_leader_epoch { @@ -1100,10 +1113,12 @@ impl DataUsageCache { // process-local leader epoch is intentionally excluded because it // changes during a safe handoff. The domain version also keeps // old epoch-bound receipts from colliding with this digest. + // Domain 3 permits object-health debt behind the frontier. Older + // readers must reject it because they skip covered failures. serde_json::to_writer( &mut writer, &( - 2u8, + domain_version, &self.info.name, self.info.scan_identity, self.info.source, @@ -1117,6 +1132,11 @@ impl DataUsageCache { .cache .iter() .filter(|(key, _)| { + // Sign exactly the subtrees the reader can reuse, including + // descendants whose raw path order differs from traversal order. + if legacy_leader_epoch.is_none() && domain_version >= 3 { + return scan_frontier_covers_path(key, through); + } let ancestor = through .strip_prefix(key.as_str()) .is_some_and(|suffix| suffix.starts_with('/')); @@ -1152,6 +1172,7 @@ impl DataUsageCache { .is_some_and(|suffix| suffix.starts_with('/')) && self.find(&receipt.through).is_some() && (self.coverage_prefix_digest(&receipt.through).ok() == Some(receipt.digest) + || self.v2_coverage_prefix_digest(&receipt.through).ok() == Some(receipt.digest) || self.legacy_coverage_prefix_digest(&receipt.through).ok() == Some(receipt.digest))) .then_some(receipt.through.as_str()) } diff --git a/crates/scanner/src/data_usage_define/tests.rs b/crates/scanner/src/data_usage_define/tests.rs index 31cc51f54..52d5c119c 100644 --- a/crates/scanner/src/data_usage_define/tests.rs +++ b/crates/scanner/src/data_usage_define/tests.rs @@ -2220,22 +2220,36 @@ fn prepare_bucket_checkpoint_migrates_legacy_epoch_bound_receipt() { ..Default::default() }, ); - let legacy_digest = cache - .legacy_coverage_prefix_digest("bucket/a") - .expect("legacy receipt digest"); - cache.info.scan_coverage_receipt = Some(DataUsageScanCoverageReceipt { - through: "bucket/a".to_string(), - digest: legacy_digest, - }); - assert_eq!(cache.validated_scan_frontier(), Some("bucket/a")); - - assert_eq!( - cache.prepare_bucket_checkpoint("bucket", 9, 2, source, TEST_PLAN_DIGEST, identity), - DataUsageCachePrepareOutcome::Reused - ); - assert_eq!(cache.info.leader_epoch, 2); - assert_eq!(cache.validated_scan_frontier(), Some("bucket/a")); - assert_ne!(cache.info.scan_coverage_receipt.expect("migrated receipt").digest, legacy_digest); + for legacy_digest in [ + cache + .legacy_coverage_prefix_digest("bucket/a") + .expect("epoch-bound receipt digest"), + cache.v2_coverage_prefix_digest("bucket/a").expect("v2 receipt digest"), + ] { + let mut cache = cache.clone(); + cache.info.scan_coverage_receipt = Some(DataUsageScanCoverageReceipt { + through: "bucket/a".to_string(), + digest: legacy_digest, + }); + assert_eq!(cache.validated_scan_frontier(), Some("bucket/a")); + assert_eq!( + cache.prepare_bucket_checkpoint("bucket", 9, 2, source, TEST_PLAN_DIGEST, identity), + DataUsageCachePrepareOutcome::Reused + ); + assert_eq!(cache.info.leader_epoch, 2); + assert_eq!(cache.validated_scan_frontier(), Some("bucket/a")); + let digest = cache.info.scan_coverage_receipt.as_ref().expect("migrated receipt").digest; + assert_ne!(digest, legacy_digest); + assert_ne!(digest, cache.v2_coverage_prefix_digest("bucket/a").expect("old reader v2 digest")); + assert_ne!( + digest, + cache + .legacy_coverage_prefix_digest("bucket/a") + .expect("old reader epoch digest") + ); + cache.cache.get_mut("bucket/a").expect("covered entry").failed_objects = 1; + assert!(cache.validated_scan_frontier().is_none(), "tampered health debt invalidates coverage"); + } } #[test] @@ -2985,3 +2999,39 @@ async fn test_retry_save_op_times_out_and_returns_error_after_retries() { assert!(result.is_err()); assert_eq!(attempts.load(Ordering::SeqCst), (DATA_USAGE_CACHE_SAVE_RETRIES + 1) as usize); } + +#[test] +fn scan_frontier_receipt_binds_nested_debt_outside_raw_lexical_prefix() { + let mut cache = DataUsageCache::default(); + cache.prepare_bucket_checkpoint("bucket", 8, 1, DataUsageCacheSource::new(0, 0), TEST_PLAN_DIGEST, valid_scan_identity()); + cache.replace("bucket", "", DataUsageEntry::default()); + cache.replace("bucket/a", "bucket", DataUsageEntry::default()); + cache.replace( + "bucket/a/z", + "bucket/a", + DataUsageEntry { + failed_objects: 1, + ..Default::default() + }, + ); + cache.replace( + "bucket/a-", + "bucket", + DataUsageEntry { + objects: 1, + size: 1, + ..Default::default() + }, + ); + cache + .seal_scan_frontier(Some("bucket/a-")) + .expect("seal traversal after failed subtree"); + assert_eq!(cache.validated_scan_frontier(), Some("bucket/a-")); + let digest = cache.info.scan_coverage_receipt.as_ref().expect("new receipt").digest; + assert_ne!(digest, cache.v2_coverage_prefix_digest("bucket/a-").expect("old reader digest")); + cache.cache.get_mut("bucket/a/z").expect("nested failure").failed_objects = 0; + assert!( + cache.validated_scan_frontier().is_none(), + "nested debt must be signed even when '/' sorts after '-'" + ); +} diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 6faab80ce..64c3ea1c6 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -23,6 +23,7 @@ use crate::data_usage_define::{ DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageCacheInfo, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageRawEnumerationCursor, DataUsageScanCheckpoint, DataUsageScanCheckpointReason, MAX_DATA_USAGE_CACHE_DEPTH, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt, SizeReconciliationEntry, SizeSummary, hash_path, + scan_frontier_covers_path, }; use crate::error::ScannerError; use crate::raw_page_index::{RawEnumerationPageIndex, RawEnumerationPageWriter}; @@ -426,10 +427,6 @@ fn folder_resume_match(folder_name: &str, resume_after: &str) -> Option bool { - folder <= frontier && !matches!(folder_resume_match(folder, frontier), Some(FolderResumeMatch::Descendant)) -} - fn order_items_for_resume(items: &mut [T], resume_after: Option<&str>, name: F) -> FolderResumeOrder where F: Fn(&T) -> &str, @@ -476,6 +473,19 @@ fn checkpoint_reason_from_budget(reason: Option) -> Da } } +// Cache entry failures survive diagnostic retry-cache TTL and capacity eviction. +// Index once per scan; resumed subtree checks must not flatten histograms repeatedly. +fn failed_scan_paths(cache: &DataUsageCache) -> Vec { + let mut paths = cache + .cache + .iter() + .filter(|(_, entry)| entry.failed_objects > 0) + .map(|(path, _)| path.clone()) + .collect::>(); + paths.sort_unstable(); + paths +} + fn data_usage_entry_has_progress(entry: &DataUsageEntry) -> bool { data_usage_root_has_progress(entry) } @@ -756,6 +766,7 @@ pub struct FolderScanner { pending_heals_changed: bool, coverage_frontier: Option, resume_frontier: Option, + resume_failed_paths: Vec, coverage_gap: bool, raw_enumeration_progress: Vec, pending_heal_sync_deferred: bool, @@ -1111,16 +1122,39 @@ impl FolderScanner { self.update_cache.info.scan_checkpoint = Some(checkpoint); } - fn record_completed_child(&mut self, folder: &str, healthy: bool) { - if self.old_cache.info.scan_progress.is_some() { - self.coverage_gap |= !healthy; - if !self.coverage_gap { - self.coverage_frontier = Some(folder.to_owned()); - } + fn record_completed_child(&mut self, folder: &str) { + if self.old_cache.info.scan_progress.is_some() + && !self.coverage_gap + && !self + .coverage_frontier + .as_deref() + .is_some_and(|frontier| scan_frontier_covers_path(folder, frontier)) + { + self.coverage_frontier = Some(folder.to_owned()); } self.record_scan_resume_hint(folder); } + fn resume_subtree_has_failures(&self, folder: &str) -> bool { + if self.resume_failed_paths.is_empty() { + return false; + } + if self + .resume_failed_paths + .binary_search_by(|path| path.as_str().cmp(folder)) + .is_ok() + { + return true; + } + let prefix = format!("{folder}/"); + let index = self + .resume_failed_paths + .partition_point(|path| path.as_str() < prefix.as_str()); + self.resume_failed_paths + .get(index) + .is_some_and(|path| path.starts_with(&prefix)) + } + fn record_scan_resume_hint_if_not_ancestor(&mut self, folder: &str) { let keep_existing = self .new_cache @@ -1284,10 +1318,10 @@ impl FolderScanner { // earlier subtree, including a suspended ancestor, must not be signed // as completed prefix coverage. if self.coverage_frontier.as_deref().is_some_and(|frontier| { - folder_is_covered_by_frontier(&folder.name, frontier) + scan_frontier_covers_path(&folder.name, frontier) || self.checkpoint_ancestors[..self.checkpoint_depth] .iter() - .any(|ancestor| folder_is_covered_by_frontier(&ancestor.hash.0, frontier)) + .any(|ancestor| scan_frontier_covers_path(&ancestor.hash.0, frontier)) }) { return; } @@ -1362,7 +1396,7 @@ impl FolderScanner { if self .coverage_frontier .as_deref() - .is_some_and(|frontier| folder_is_covered_by_frontier(&child_hash.0, frontier)) + .is_some_and(|frontier| scan_frontier_covers_path(&child_hash.0, frontier)) { self.coverage_frontier = None; self.coverage_gap = true; @@ -1927,7 +1961,7 @@ impl FolderScanner { // A new FolderScanner starts with an empty seen set, so // cached paths are still rechecked once per cycle. Skip // duplicate listings within this same scanner pass. - self.coverage_gap |= self.old_cache.info.scan_progress.is_some(); + into.failed_objects = into.failed_objects.max(1); continue; } @@ -1943,9 +1977,8 @@ impl FolderScanner { if failure_action != GetSizeFailureAction::Skip { self.resolved_failed_object_paths.remove(&item.path); - self.coverage_gap |= self.old_cache.info.scan_progress.is_some(); + into.failed_objects = into.failed_objects.saturating_add(1); if !retry_suppressed { - into.failed_objects += 1; self.record_failed(&item.path); } @@ -2098,14 +2131,13 @@ impl FolderScanner { } if self.is_erasure_mode && found_erasure_data_directory && !found_object_metadata { - self.coverage_gap |= self.old_cache.info.scan_progress.is_some(); found_object_metadata = true; let metadata_path = path_join_buf(&[&dir_path, STORAGE_FORMAT_FILE]); self.mark_failed_path_seen(&metadata_path); let retry_suppressed = self.failed_retry_suppressed(&metadata_path); + into.failed_objects = into.failed_objects.saturating_add(1); if !retry_suppressed { - into.failed_objects = into.failed_objects.saturating_add(1); self.record_failed(&metadata_path); let failed_cache_entries = self.new_cache.info.failed_objects.len(); @@ -2253,8 +2285,9 @@ impl FolderScanner { && !into.compacted && forward_resume_after .as_deref() - .is_some_and(|resume| folder_is_covered_by_frontier(&folder_item.name, resume)) + .is_some_and(|resume| scan_frontier_covers_path(&folder_item.name, resume)) && self.old_cache.find(&folder_item.name).is_some() + && !self.resume_subtree_has_failures(&folder_item.name) { self.new_cache.copy_with_children(&self.old_cache, &h, &folder_item.parent); into.add_child(&h); @@ -2323,7 +2356,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.record_completed_child(&folder_item.name, into.failed_objects == 0); + self.record_completed_child(&folder_item.name); self.send_update_for_entry(&this_hash, &folder.parent, into).await; tokio::task::yield_now().await; } else { @@ -2356,7 +2389,7 @@ impl FolderScanner { tokio::task::yield_now().await; into.add_child(&h); - self.record_completed_child(&folder_item.name, dst.failed_objects == 0); + self.record_completed_child(&folder_item.name); self.maybe_send_checkpoint(&folder, &this_hash, into); // We scanned a folder, optionally send update. self.update_cache.delete_recursive(&h); @@ -2918,6 +2951,11 @@ pub(crate) async fn scan_data_folder_scoped( info: cache.info.clone(), ..Default::default() }; + let resume_failed_paths = if resume_frontier.is_some() { + failed_scan_paths(&cache) + } else { + Vec::new() + }; // Create folder scanner let mut scanner = FolderScanner { root: base_path, @@ -2952,6 +2990,7 @@ pub(crate) async fn scan_data_folder_scoped( pending_heals_changed: false, coverage_frontier: resume_frontier.clone(), resume_frontier, + resume_failed_paths, coverage_gap: false, pending_heal_sync_deferred: false, pending_heal_batch_dirty: false, @@ -2995,7 +3034,7 @@ pub(crate) async fn scan_data_folder_scoped( new_cache.info.last_update = Some(SystemTime::now()); new_cache.info.next_cycle = next_cycle; let unresolved_objects = coverage_gap - || root.failed_objects > 0 + || new_cache.cache.values().any(|entry| entry.failed_objects > 0) || !new_cache.info.failed_objects.is_empty() || !new_cache.info.size_reconciliation.is_empty(); let mixed_coverage = new_cache diff --git a/crates/scanner/src/scanner_folder/tests.rs b/crates/scanner/src/scanner_folder/tests.rs index 106169bd1..5a0925cf2 100644 --- a/crates/scanner/src/scanner_folder/tests.rs +++ b/crates/scanner/src/scanner_folder/tests.rs @@ -361,6 +361,7 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) { pending_heals_changed: false, coverage_frontier: None, resume_frontier: None, + resume_failed_paths: Vec::new(), coverage_gap: false, raw_enumeration_progress: Vec::new(), pending_heal_sync_deferred: false, @@ -2495,7 +2496,7 @@ async fn test_scan_folder_corrupt_xl_meta_stops_erasure_data_dir_descent() { .await .expect("cached metadata failure must still stop erasure data directory descent"); - assert_eq!(retry_into.failed_objects, 0, "cached failure should not be counted twice"); + assert_eq!(retry_into.failed_objects, 1, "a rebuilt entry must retain suppressed failure debt"); assert!(!retry_budget.budget_elapsed()); assert_eq!(retry_budget.reason(), None); @@ -2620,7 +2621,7 @@ async fn test_scan_folder_missing_xl_meta_stops_erasure_data_dir_descent() { .await .expect("cached missing metadata must still stop erasure data directory descent"); - assert_eq!(retry_into.failed_objects, 0, "cached failure should not be counted twice"); + assert_eq!(retry_into.failed_objects, 1, "a rebuilt entry must retain suppressed failure debt"); assert!(!retry_budget.budget_elapsed()); assert_eq!(retry_budget.reason(), None); } diff --git a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/periodic_snapshot.rs b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/periodic_snapshot.rs index c838d9fc4..43ee7b21c 100644 --- a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/periodic_snapshot.rs +++ b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/periodic_snapshot.rs @@ -22,6 +22,7 @@ const OBJECTS: u64 = 8; fn prepare_scanner(scanner: &mut FolderScanner, cache: DataUsageCache) { scanner.resume_frontier = cache.validated_scan_frontier().map(str::to_owned); scanner.coverage_frontier = scanner.resume_frontier.clone(); + scanner.resume_failed_paths = failed_scan_paths(&cache); scanner.new_cache = DataUsageCache { info: cache.info.clone(), ..Default::default() @@ -180,6 +181,8 @@ async fn finish_resumed_walk(cache: DataUsageCache, original: &FolderScanner) -> scanner.root = original.root.clone(); scanner.local_disk = original.local_disk.clone(); prepare_scanner(&mut scanner, cache); + scanner.failed_object_ttl_secs = original.failed_object_ttl_secs; + scanner.failed_objects_max = original.failed_objects_max; let parent = CancellationToken::new(); scanner.budget = ScannerCycleBudget::new_with_progress_tracking(&parent, Default::default()); let mut entry = DataUsageEntry::default(); @@ -216,20 +219,22 @@ async fn periodic_nested_snapshot_resumes_after_leader_handoff_with_exact_totals #[tokio::test] #[serial] -async fn periodic_nested_snapshot_keeps_failed_prefix_outside_resume_frontier() { +async fn periodic_nested_snapshot_resumes_healthy_progress_after_a_failed_object() { let (loaded, _, original, _guard) = nested_periodic_snapshot(true, (0, 0)).await; - assert!( - loaded.validated_scan_frontier().is_none(), - "an earlier corrupt object must block subtree skipping" + assert_eq!( + loaded.validated_scan_frontier(), + Some("bucket/prefix/a/0001"), + "object health debt must not freeze completed traversal coverage" ); assert!(!loaded.info.failed_objects.is_empty(), "persisted failure debt must survive handoff"); let (finished, visits, _unused_guard) = finish_resumed_walk(loaded, &original).await; - assert_eq!(visits, OBJECTS - 1, "every healthy object must be re-read after the coverage gap"); + assert_eq!(visits, OBJECTS - 2, "the healthy covered sibling must skip its metadata read"); let totals = finished .checked_flatten_complete_scope("bucket") .expect("failed sweep still has connected accounting"); assert_eq!(u64::try_from(totals.objects).expect("healthy count"), OBJECTS - 1); assert_eq!(u64::try_from(totals.size).expect("healthy bytes"), OBJECTS - 1); + assert_eq!(totals.failed_objects, 1, "the retried failed subtree must retain its debt"); assert!(!finished.info.failed_objects.is_empty(), "unrepaired metadata must remain pending"); } @@ -792,3 +797,189 @@ async fn periodic_nested_snapshot_does_not_publish_shared_compacted_accumulator_ "a shared compacted parent accumulator must not be snapshotted as the current child" ); } + +#[tokio::test] +#[serial] +async fn periodic_failed_object_debt_survives_diagnostic_eviction_and_clears_after_repair() { + let (mut loaded, _, mut original, _guard) = nested_periodic_snapshot(true, (0, 0)).await; + loaded.info.failed_objects.clear(); + original.failed_object_ttl_secs = 0; + let (failed, visits, _unused_guard) = finish_resumed_walk(loaded.clone(), &original).await; + let totals = failed + .checked_flatten_complete_scope("bucket") + .expect("failed resumed accounting"); + assert_eq!(visits, OBJECTS - 2); + assert_eq!(totals.objects, usize::try_from(OBJECTS - 1).expect("healthy objects")); + assert_eq!(totals.failed_objects, 1, "uncapped entry debt must survive without diagnostic entries"); + assert!(failed.info.failed_objects.is_empty(), "TTL zero must not be needed for correctness"); + + write_checkpoint_object(std::path::Path::new(&original.root), "prefix/a/0000", &[(None, 1)]).await; + let (repaired, visits, _repair_guard) = finish_resumed_walk(loaded, &original).await; + let totals = repaired + .checked_flatten_complete_scope("bucket") + .expect("repaired resumed accounting"); + assert_eq!( + visits, + OBJECTS - 1, + "repair must reread the failed object and retain the covered healthy sibling" + ); + assert_eq!(totals.objects, usize::try_from(OBJECTS).expect("all objects")); + assert_eq!(totals.size, usize::try_from(OBJECTS).expect("all bytes")); + assert_eq!(totals.failed_objects, 0); +} + +#[tokio::test] +#[serial] +async fn periodic_failed_subtree_resumes_healthy_descendants_and_compacted_debt() { + let (loaded, identity, original, _guard) = nested_periodic_snapshot(true, (0, 0)).await; + let (mut complete_walk, _, _unused_guard) = finish_resumed_walk(loaded, &original).await; + complete_walk.info.failed_objects.clear(); + complete_walk + .seal_scan_frontier(Some("bucket/prefix/a")) + .expect("seal completed unhealthy parent"); + for compacted in [false, true] { + let mut cache = complete_walk.clone(); + if compacted { + cache.reduce_children_of(&hash_path("bucket/prefix/a"), 1, true); + assert!(cache.find("bucket/prefix/a").expect("compacted parent").compacted); + cache + .seal_scan_frontier(Some("bucket/prefix/a")) + .expect("reseal compacted accounting"); + } + let store = FixtureStore::new(); + let revisions = DataUsageCache::default() + .load_with_revisions(store.clone(), CACHE_NAME) + .await + .expect("fresh revisions"); + cache + .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0) + .await + .expect("persist failure-bearing coverage"); + let mut adopted = DataUsageCache::default(); + adopted + .load(store, CACHE_NAME) + .await + .expect("reload failure-bearing coverage"); + assert_eq!( + adopted.prepare_bucket_checkpoint("bucket", 11, 9, SOURCE, PLAN, identity), + crate::DataUsageCachePrepareOutcome::Reused + ); + let (finished, visits, _resumed_guard) = finish_resumed_walk(adopted, &original).await; + assert_eq!( + visits, + if compacted { OBJECTS - 1 } else { 4 }, + "only healthy independently cached covered subtrees may skip reads" + ); + let totals = finished + .checked_flatten_complete_scope("bucket") + .expect("resumed failed parent accounting"); + assert_eq!(totals.objects, usize::try_from(OBJECTS - 1).expect("healthy objects")); + assert_eq!(totals.size, usize::try_from(OBJECTS - 1).expect("healthy bytes")); + assert_eq!(totals.failed_objects, 1); + } +} + +#[tokio::test] +#[serial] +async fn periodic_resume_uses_traversal_order_for_punctuation_siblings() { + let (loaded, _, original, _guard) = nested_periodic_snapshot(true, (0, 0)).await; + write_checkpoint_object(std::path::Path::new(&original.root), "prefix/a-", &[(None, 1)]).await; + let (mut cache, _, _walk_guard) = finish_resumed_walk(loaded, &original).await; + cache.info.failed_objects.clear(); + cache + .seal_scan_frontier(Some("bucket/prefix/a-")) + .expect("seal completed punctuation sibling"); + let (finished, visits, _resumed_guard) = finish_resumed_walk(cache, &original).await; + assert_eq!(visits, 4, "healthy children of the earlier failed sibling must remain covered"); + let totals = finished + .checked_flatten_complete_scope("bucket") + .expect("punctuation sibling accounting"); + assert_eq!( + totals.objects, + usize::try_from(OBJECTS).expect("healthy originals plus punctuation sibling") + ); + assert_eq!(totals.size, totals.objects); + assert_eq!(totals.failed_objects, 1); +} + +#[tokio::test] +#[serial] +async fn periodic_failed_nested_entries_cannot_certify_a_bucket_with_diagnostics_disabled() { + let (loaded, _, original, _guard) = nested_periodic_snapshot(true, (0, 0)).await; + let result = temp_env::async_with_vars([(ENV_FAILED_OBJECT_TTL_SECS, Some("0"))], async { + let mut cache = loaded; + cache.info.failed_objects.clear(); + cache.info.skip_healing = true; + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new_with_progress_tracking(&parent, Default::default()); + scan_data_folder( + parent, + budget, + vec![original.local_disk.clone()], + original.local_disk.clone(), + cache, + None, + HealScanMode::Normal, + DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest), + ) + .await + }) + .await; + let Err(ScannerError::PartialCache(cache)) = result else { + panic!("nested unresolved object must keep the bucket partial"); + }; + assert!(!cache.info.snapshot_complete); + assert!(cache.info.failed_objects.is_empty()); + assert_eq!( + cache + .checked_flatten_complete_scope("bucket") + .expect("partial accounting") + .failed_objects, + 1 + ); +} + +#[tokio::test] +#[serial] +async fn periodic_resume_debt_index_and_frontier_keep_subtree_boundaries() { + let (mut scanner, root) = build_test_scanner().await; + let _guard = TestGuard { temp_dir: Some(root) }; + scanner.resume_failed_paths = vec!["bucket/a-".into(), "bucket/a/z".into(), "bucket/b".into()]; + assert!( + scanner.resume_subtree_has_failures("bucket/a"), + "punctuation sibling must not hide a failing descendant" + ); + assert!(scanner.resume_subtree_has_failures("bucket/b")); + assert!(!scanner.resume_subtree_has_failures("bucket/ab")); + assert!(!scanner.resume_subtree_has_failures("bucket/a/healthy")); + let (cache, _) = bound_checkpoint(); + prepare_scanner(&mut scanner, cache); + scanner.coverage_frontier = Some("bucket/prefix/a".into()); + scanner.record_completed_child("bucket/prefix/a/0000"); + assert_eq!( + scanner.coverage_frontier.as_deref(), + Some("bucket/prefix/a"), + "debt retry must not narrow inherited coverage" + ); + scanner.record_completed_child("bucket/prefix"); + assert_eq!( + scanner.coverage_frontier.as_deref(), + Some("bucket/prefix"), + "completed ancestor widens coverage" + ); + scanner.coverage_frontier = Some("bucket/a-".into()); + assert!(scan_frontier_covers_path("bucket/a/healthy", "bucket/a-")); + scanner.record_completed_child("bucket/a/healthy"); + assert_eq!( + scanner.coverage_frontier.as_deref(), + Some("bucket/a-"), + "component order must preserve a later sibling's frontier" + ); + scanner.coverage_gap = true; + scanner.record_completed_child("bucket/z"); + assert_eq!( + scanner.coverage_frontier.as_deref(), + Some("bucket/a-"), + "structural gaps must still freeze coverage" + ); +} diff --git a/docs/architecture/scanner-usage-publication.md b/docs/architecture/scanner-usage-publication.md index 12c600d21..f712cbbbc 100644 --- a/docs/architecture/scanner-usage-publication.md +++ b/docs/architecture/scanner-usage-publication.md @@ -121,6 +121,24 @@ reuse without this identity. Older readers can ignore the added map key, but older writers do not enforce its fence; readability is not a mixed-version publication-safety guarantee. +Bucket checkpoints distinguish completed traversal from object health. A failed +object does not stop later completed subtrees from advancing the traversal +frontier. On resume, a sorted index of cached entries with `failed_objects > 0` +excludes their containing subtrees from reuse; healthy covered siblings keep +their exact cached accounting. Failure counters survive retry-cache TTL and +capacity eviction, and any remaining entry failure keeps the bucket partial. +Directory traversal errors and unresolved size reconciliation still stop the +frontier because they do not have the same per-entry coverage proof. + +Coverage follows path-component traversal order, including punctuation in sibling +names. Coverage receipts use digest domain 3 and bind every entry in each covered +subtree using the same predicate as resume, while excluding the unvisited suffix. +No unsigned nested failure can authorize subtree reuse. Current +readers accept and migrate domain 2 and legacy epoch-bound receipts. Older +binaries cannot validate a domain 3 receipt and safely reread the bucket rather +than applying their unconditional covered-subtree skip. The serialized receipt +shape, cache CAS, leader epoch and publication fences are unchanged. + ## Persisted Objects The persisted objects are part of the compatibility contract. Removing one