From 6681dd12e39fda867c54cc4e576ecf7f4257aec4 Mon Sep 17 00:00:00 2001 From: Hauser Date: Sun, 18 Oct 2026 16:50:42 +0800 Subject: [PATCH] fix(scanner): expose mixed sweep checkpoint transitions (#8657) * fix(scanner): expose mixed sweep checkpoint transitions Correlate checkpoint admission, incomplete sweeps, and successful terminal Partial persistence by bucket/source and cycle/leader epoch. Share the traversal-position predicate and local/remote save diagnostic without changing certification, acknowledgement, or persistence contracts. Extend real-disk cache regressions across cycles and leader changes and verify that failed terminal saves retain an older periodic frontier until a successful terminal save permits fresh certification. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(scanner): report the aggregate incomplete sweep gate Expose the existing unresolved_objects decision in the incomplete-sweep event. Nested cache-entry failures can keep a sweep partial while both the root counter and the capped diagnostic failure map are empty. Extend the real-disk regression with diagnostics disabled and distinguish this case from mixed-only coverage without adding another cache walk. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: heihutu Co-authored-by: zhi22915 --- crates/scanner/src/data_usage_define.rs | 52 +++++--- crates/scanner/src/remote_scanner/stream.rs | 6 +- crates/scanner/src/scanner_folder.rs | 5 + .../tests/checkpoint_fixture.rs | 64 ++++++++- .../checkpoint_fixture/periodic_snapshot.rs | 121 +++++++++++++++++- crates/scanner/src/scanner_io.rs | 3 +- crates/scanner/src/scanner_io/cache.rs | 18 +++ crates/scanner/src/scanner_io/io_cache.rs | 11 +- .../architecture/scanner-usage-publication.md | 20 ++- 9 files changed, 261 insertions(+), 39 deletions(-) diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 7c033b6ca..1a94614d7 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -899,6 +899,14 @@ impl DataUsageCache { } } + pub(crate) fn has_scan_position(&self) -> bool { + self.info.scan_resume_after.is_some() + || self.info.scan_checkpoint.is_some() + || self.info.scan_raw_enumeration_cursor.is_some() + || self.info.scan_raw_enumeration_page_index.is_some() + || self.info.scan_coverage_receipt.is_some() + } + pub(crate) fn prepare_bucket_checkpoint( &mut self, name: &str, @@ -914,6 +922,8 @@ impl DataUsageCache { if self.info.leader_epoch > leader_epoch { return DataUsageCachePrepareOutcome::RejectedNewerLeader; } + let previous_cycle = self.info.next_cycle; + let previous_leader_epoch = self.info.leader_epoch; // A checkpoint is a durable traversal frontier, not a publication // result. A newer leader may adopt it only when the frontier is // independently fenced by the bucket identity and has a validated @@ -942,22 +952,13 @@ impl DataUsageCache { if reusable && self.info.snapshot_complete && self.info.scan_progress.is_none() - && self.info.scan_checkpoint.is_none() - && self.info.scan_raw_enumeration_cursor.is_none() - && self.info.scan_raw_enumeration_page_index.is_none() - && self.info.scan_resume_after.is_none() - && self.info.scan_coverage_receipt.is_none() + && !self.has_scan_position() && self.info.scan_plan_digest == Some(scan_plan_digest) { return self.prepare_for_scan(name, next_cycle, leader_epoch, source, scan_plan_digest, true); } if !reusable { - let had_resume_state = self.info.scan_checkpoint.is_some() - || self.info.scan_resume_after.is_some() - || self.info.scan_coverage_receipt.is_some() - || self.info.scan_raw_enumeration_cursor.is_some() - || self.info.scan_raw_enumeration_page_index.is_some(); - if had_resume_state { + if self.has_scan_position() { let reason = if name == DATA_USAGE_ROOT { "invalid_bucket_scope" } else if !identity.is_valid() { @@ -1011,11 +1012,7 @@ impl DataUsageCache { if self.validated_raw_enumeration_page_index().is_none() { self.info.scan_raw_enumeration_page_index = None; } - let cursor_is_valid = (self.info.scan_checkpoint.is_none() - && self.info.scan_raw_enumeration_cursor.is_none() - && self.info.scan_raw_enumeration_page_index.is_none() - && self.info.scan_resume_after.is_none() - && self.info.scan_coverage_receipt.is_none()) + let cursor_is_valid = !self.has_scan_position() || self.validated_scan_frontier().is_some() || self.info.scan_raw_enumeration_cursor.is_some() || self.info.scan_raw_enumeration_page_index.is_some(); @@ -1043,11 +1040,7 @@ impl DataUsageCache { // so it observes nothing under the plan that finished the previous // sweep. Keeping that plan would mark every sweep of a bucket written // between cycles as mixed, and the bucket could never certify. - let has_position = self.info.scan_resume_after.is_some() - || self.info.scan_checkpoint.is_some() - || self.info.scan_raw_enumeration_cursor.is_some() - || self.info.scan_raw_enumeration_page_index.is_some() - || self.info.scan_coverage_receipt.is_some(); + let has_position = self.has_scan_position(); if let Some(progress) = &mut self.info.scan_progress && has_position { @@ -1066,6 +1059,23 @@ impl DataUsageCache { // Old readers do not understand coverage sweeps. An absent plan makes // their existing prepare path rebuild instead of promoting mixed data. self.info.scan_plan_digest = None; + debug!( + target: "rustfs::scanner::data_usage", + event = EVENT_SCANNER_CACHE_LOAD_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_CACHE, + state = "bucket_checkpoint_prepared", + bucket = %name, + pool = source.pool_index, + set = source.set_index, + cycle = next_cycle, + leader_epoch, + previous_cycle, + previous_leader_epoch, + resume_position = self.has_scan_position(), + mixed_coverage = self.info.scan_progress.is_some_and(|progress| progress.started_plan != progress.requested_plan), + "Scanner bucket checkpoint prepared for a coverage sweep" + ); if reusable { DataUsageCachePrepareOutcome::Reused } else { diff --git a/crates/scanner/src/remote_scanner/stream.rs b/crates/scanner/src/remote_scanner/stream.rs index 8bb3224e6..3ce3740fd 100644 --- a/crates/scanner/src/remote_scanner/stream.rs +++ b/crates/scanner/src/remote_scanner/stream.rs @@ -18,7 +18,8 @@ use crate::scanner_budget::{ScannerCycleBudget, ScannerCycleBudgetConfig}; use crate::scanner_io::{ DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerCheckpointPersistContext, ScannerCheckpointPersistResult, ScannerDiskScanOptions, ScannerDiskScanOutcome, ScannerIODisk, acquire_scanner_cache_locks, cache_root_entry_info, - current_cache_root_or_prepare_with_generation, persist_scanner_checkpoint, scanner_set_disk_inventory, + current_cache_root_or_prepare_with_generation, log_saved_partial_bucket_cache, persist_scanner_checkpoint, + scanner_set_disk_inventory, }; use crate::storage_api::owner::NS_SCANNER_PROTOCOL_VERSION; use crate::{ @@ -957,6 +958,9 @@ async fn scan_and_persist_local_bucket( } if let Some(partial_result) = partial_result { + if matches!(partial_result, RemoteScannerFrameResult::Partial) { + log_saved_partial_bucket_cache(&cache, &cache_name); + } return Ok(partial_result); } if cache.info.source != Some(source) || !cache.info.snapshot_complete { diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 726c4fdf1..9a3e9a829 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -2955,6 +2955,7 @@ pub(crate) async fn scan_data_folder_scoped( } let bucket_name = cache.info.name.clone(); let next_cycle = cache.info.next_cycle; + let resume_position_on_entry = cache.has_scan_position(); let had_scan_checkpoint = cache.info.scan_checkpoint.is_some(); let new_cache = DataUsageCache { info: cache.info.clone(), @@ -3089,7 +3090,11 @@ pub(crate) async fn scan_data_folder_scoped( pool = new_cache.info.source.map(|source| source.pool_index), set = new_cache.info.source.map(|source| source.set_index), cycle = next_cycle, + leader_epoch = new_cache.info.leader_epoch, + resume_position_on_entry, + resume_position = new_cache.has_scan_position(), coverage_gap, + unresolved_objects, root_failed_objects = root.failed_objects, failed_objects = new_cache.info.failed_objects.len(), size_reconciliation_entries = new_cache.info.size_reconciliation.len(), diff --git a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs index 305fe3e77..e1645fd55 100644 --- a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs +++ b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs @@ -886,6 +886,16 @@ async fn checkpoint_fixture_hot_digest_retains_partial_progress() { #[tokio::test] #[serial] async fn checkpoint_fixture_hot_bucket_certifies_first_full_sweep_after_mixed_end() { + hot_bucket_certifies_first_full_sweep_after_mixed_end(false).await; +} + +#[tokio::test] +#[serial] +async fn checkpoint_fixture_hot_bucket_certifies_after_mixed_end_and_leader_handoff() { + hot_bucket_certifies_first_full_sweep_after_mixed_end(true).await; +} + +async fn hot_bucket_certifies_first_full_sweep_after_mixed_end(advance_epoch: bool) { let logs = CapturedLogs::default(); let subscriber = tracing_subscriber::fmt() .json() @@ -915,6 +925,8 @@ async fn checkpoint_fixture_hot_bucket_certifies_first_full_sweep_after_mixed_en // tail that ends mixed, then fresh unbounded sweeps under new plans. let rounds = [(Some(4), false), (None, false), (None, true), (None, true)]; for (round, (max_objects, expect_complete)) in (0_u64..).zip(rounds) { + let cycle = 11 + round; + let leader_epoch = if advance_epoch { 7 + round } else { 7 }; write_checkpoint_object(&root, "hot/current", &[(None, 1)]).await; let plan = crate::scanner_io::checkpoint_fixture_bucket_digest(PLAN, Some(round)); let mut cache = DataUsageCache::default(); @@ -926,8 +938,8 @@ async fn checkpoint_fixture_hot_bucket_certifies_first_full_sweep_after_mixed_en &mut cache, "bucket", SOURCE, - 11, - 7, + cycle, + leader_epoch, plan, crate::scanner_io::DataUsageCacheReuseOptions { require_source: true, @@ -970,6 +982,18 @@ async fn checkpoint_fixture_hot_bucket_certifies_first_full_sweep_after_mixed_en let saved = store.strict_load().await; assert_eq!(complete, expect_complete, "round {round} outcome"); assert_eq!(saved.info.snapshot_complete, expect_complete, "round {round} saved completion"); + assert_eq!(saved.info.next_cycle, cycle); + assert_eq!(saved.info.leader_epoch, leader_epoch); + if round == 1 { + assert!(saved.info.scan_resume_after.is_none()); + assert!(saved.info.scan_checkpoint.is_none()); + assert!(saved.info.scan_raw_enumeration_cursor.is_none()); + assert!(saved.info.scan_raw_enumeration_page_index.is_none()); + assert!(saved.info.scan_coverage_receipt.is_none()); + let progress = saved.info.scan_progress.expect("mixed tail retains a verification plan"); + assert_eq!(progress.started_plan, plan); + assert_eq!(progress.requested_plan, plan); + } if expect_complete { assert_eq!(saved.info.scan_plan_digest, Some(plan), "round {round} certifies its own plan"); assert!(saved.info.scan_progress.is_none()); @@ -979,10 +1003,37 @@ async fn checkpoint_fixture_hot_bucket_certifies_first_full_sweep_after_mixed_en assert!(saved.info.scan_plan_digest.is_none(), "round {round} must not certify"); } } - let events = logs - .contents() + let captured = logs.contents(); + let all_events = captured .lines() .map(|line| serde_json::from_str::(line).expect("scanner JSON event")) + .collect::>(); + let admissions = all_events + .iter() + .filter(|event| event["fields"]["state"] == "bucket_checkpoint_prepared") + .collect::>(); + assert_eq!(admissions.len(), 4, "every sweep must identify fresh versus resumed admission"); + for (round, event) in admissions.iter().enumerate() { + let fields = &event["fields"]; + let round_cycle = 11 + u64::try_from(round).expect("round fits u64"); + assert_eq!(fields["cycle"], round_cycle); + assert_eq!(fields["leader_epoch"], if advance_epoch { round_cycle - 4 } else { 7 }); + assert_eq!(fields["previous_cycle"], if round == 0 { 0 } else { round_cycle - 1 }); + assert_eq!( + fields["previous_leader_epoch"], + if round == 0 { + 0 + } else if advance_epoch { + round_cycle - 5 + } else { + 7 + } + ); + assert_eq!(fields["resume_position"], round == 1); + assert_eq!(fields["mixed_coverage"], round == 1); + } + let events = all_events + .iter() .filter(|event| event["fields"]["state"] == "bucket_sweep_incomplete") .collect::>(); assert_eq!(events.len(), 1, "only the mixed tail needs a verification-sweep diagnostic"); @@ -990,7 +1041,12 @@ async fn checkpoint_fixture_hot_bucket_certifies_first_full_sweep_after_mixed_en assert_eq!(fields["bucket"], "bucket"); assert_eq!(fields["pool"], 0); assert_eq!(fields["set"], 0); + assert_eq!(fields["cycle"], 12); + assert_eq!(fields["leader_epoch"], if advance_epoch { 8 } else { 7 }); + assert_eq!(fields["resume_position_on_entry"], true); + assert_eq!(fields["resume_position"], false); assert_eq!(fields["mixed_coverage"], true); + assert_eq!(fields["unresolved_objects"], false); assert_eq!(fields["coverage_gap"], false); assert_eq!(fields["failed_objects"], 0); assert_eq!(fields["size_reconciliation_entries"], 0); 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 43ee7b21c..93355f7d2 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 @@ -53,7 +53,10 @@ async fn nested_periodic_snapshot( if corrupt_first { write_test_object_metadata_bytes(&root, "bucket", "prefix/a/0000", b"not-valid-filemeta").await; } - let (_, identity) = bound_checkpoint(); + let identity = crate::DataUsageScanIdentity { + tier_registry_generation: crate::runtime_tier_registry_for_cycle(11, 7).await.generation, + ..bound_checkpoint().1 + }; let mut cache = DataUsageCache::default(); cache.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, identity); prepare_scanner(&mut scanner, cache); @@ -202,6 +205,96 @@ async fn finish_resumed_walk(cache: DataUsageCache, original: &FolderScanner) -> (scanner.new_cache, visits, guard) } +#[tokio::test] +#[serial] +async fn periodic_mixed_tail_requires_terminal_persistence_before_fresh_certification() { + let (periodic, identity, scanner, _guard) = nested_periodic_snapshot(false, (0, 0)).await; + assert!(periodic.has_scan_position(), "real periodic snapshot retains traversal position"); + let store = FixtureStore::new(); + let revisions = DataUsageCache::default() + .load_with_revisions(store.clone(), CACHE_NAME) + .await + .expect("initial periodic fixture revisions"); + periodic + .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0) + .await + .expect("seed durable periodic snapshot"); + + for round in 0..3 { + let cycle = 12 + round; + let leader_epoch = 9 + round; + let plan = crate::scanner_io::checkpoint_fixture_bucket_digest(PLAN, Some(cycle)); + write_checkpoint_object(std::path::Path::new(&scanner.root), "prefix/b/0003", &[(None, 1)]).await; + let mut cache = DataUsageCache::default(); + let revisions = cache + .load_with_revisions(store.clone(), CACHE_NAME) + .await + .expect("reload last successful save"); + cache.prepare_bucket_checkpoint("bucket", cycle, leader_epoch, SOURCE, plan, identity); + assert_eq!( + cache.has_scan_position(), + round < 2, + "only a saved terminal clears the inherited position" + ); + assert_eq!( + cache + .info + .scan_progress + .is_some_and(|progress| progress.started_plan != progress.requested_plan), + round < 2, + "a surviving periodic prefix must remain mixed under a changed plan" + ); + cache.info.skip_healing = true; + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new_with_progress_tracking(&parent, Default::default()); + let outcome = scanner + .local_disk + .clone() + .nsscanner_disk( + budget.token(), + budget.clone(), + vec![scanner.local_disk.clone()], + cache, + None, + scan_options(HealScanMode::Normal), + ) + .await + .expect("finish the periodic tail or fresh verification sweep"); + let (terminal, complete) = match outcome { + ScannerDiskScanOutcome::Partial(cache) => (cache, false), + ScannerDiskScanOutcome::Complete(cache) => (cache, true), + ScannerDiskScanOutcome::NamespaceNotFound(_) => panic!("fixture namespace exists"), + }; + assert!(!budget.budget_elapsed()); + assert_eq!(complete, round == 2); + assert!(!terminal.has_scan_position(), "completed traversal clears every position before saving"); + let totals = terminal.checked_flatten("bucket").expect("connected terminal root"); + assert_eq!(totals.objects, usize::try_from(OBJECTS).expect("fixture count")); + assert_eq!(totals.size, usize::try_from(OBJECTS).expect("fixture bytes")); + + store.reject_save.store(round == 0, Ordering::SeqCst); + let saved = terminal + .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0) + .await; + if round == 0 { + assert!(saved.is_err(), "injected terminal persistence failure must be observable"); + let retained = store.strict_load().await; + assert_eq!(retained.info.scan_coverage_receipt, periodic.info.scan_coverage_receipt); + assert_eq!(retained.info.scan_checkpoint, periodic.info.scan_checkpoint); + assert!( + retained.has_scan_position(), + "failed terminal save must leave periodic recovery state intact" + ); + } else { + saved.expect("persist terminal sweep"); + let retained = store.strict_load().await; + assert!(!retained.has_scan_position()); + assert_eq!(retained.info.snapshot_complete, round == 2); + } + store.reject_save.store(false, Ordering::SeqCst); + } +} + #[tokio::test] #[serial] async fn periodic_nested_snapshot_resumes_after_leader_handoff_with_exact_totals() { @@ -905,6 +998,15 @@ async fn periodic_resume_uses_traversal_order_for_punctuation_siblings() { #[tokio::test] #[serial] async fn periodic_failed_nested_entries_cannot_certify_a_bucket_with_diagnostics_disabled() { + let logs = CapturedLogs::default(); + let subscriber = tracing_subscriber::fmt() + .json() + .with_max_level(tracing::Level::DEBUG) + .with_writer(logs.clone()) + .with_ansi(false) + .without_time() + .finish(); + let _subscriber_guard = tracing::subscriber::set_default(subscriber); 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; @@ -937,6 +1039,23 @@ async fn periodic_failed_nested_entries_cannot_certify_a_bucket_with_diagnostics .failed_objects, 1 ); + let events = logs + .contents() + .lines() + .map(|line| serde_json::from_str::(line).expect("scanner JSON event")) + .filter(|event| event["fields"]["state"] == "bucket_sweep_incomplete") + .collect::>(); + assert_eq!(events.len(), 1, "the completed sweep must identify its uncapped failure debt"); + let fields = &events[0]["fields"]; + assert_eq!(fields["coverage_gap"], false); + assert_eq!(fields["root_failed_objects"], 0); + assert_eq!(fields["failed_objects"], 0); + assert_eq!(fields["size_reconciliation_entries"], 0); + assert_eq!(fields["mixed_coverage"], false); + assert_eq!( + fields["unresolved_objects"], true, + "child-entry failure debt remains a certification blocker" + ); } #[tokio::test] diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index d570ca6bd..8a1421a4d 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -1809,7 +1809,8 @@ use guards::*; pub(crate) use cache::{ DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerCheckpointPersistContext, ScannerCheckpointPersistResult, - acquire_scanner_cache_locks, current_cache_root_or_prepare_with_generation, persist_scanner_checkpoint, + acquire_scanner_cache_locks, current_cache_root_or_prepare_with_generation, log_saved_partial_bucket_cache, + persist_scanner_checkpoint, }; pub use dirty_usage::{ ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageClearObserver, ScannerDirtyUsageMutationObserver, diff --git a/crates/scanner/src/scanner_io/cache.rs b/crates/scanner/src/scanner_io/cache.rs index 21d2eb286..d11c2d000 100644 --- a/crates/scanner/src/scanner_io/cache.rs +++ b/crates/scanner/src/scanner_io/cache.rs @@ -15,6 +15,24 @@ use super::*; use crate::{ScannerConfigObjectDelete, ScannerObjectIO}; +pub(crate) fn log_saved_partial_bucket_cache(cache: &DataUsageCache, cache_name: &str) { + debug!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_CACHE_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + state = "partial_saved_not_published", + bucket = %cache.info.name, + pool = cache.info.source.map(|source| source.pool_index), + set = cache.info.source.map(|source| source.set_index), + cycle = cache.info.next_cycle, + leader_epoch = cache.info.leader_epoch, + cache_name, + resume_position = cache.has_scan_position(), + "Scanner partial bucket cache saved without publishing usage aggregate" + ); +} + pub(crate) fn scanner_cache_lock_resource(cache_name: &str, source: DataUsageCacheSource) -> String { let lock_name = format!("{SCANNER_CACHE_LOCK_SUFFIX}.pool-{}.set-{}", source.pool_index, source.set_index); path_join_buf(&[crate::BUCKET_META_PREFIX, cache_name, &lock_name]) diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index 4ff5ba70e..07eddd249 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -1401,16 +1401,7 @@ impl ScannerIOCache for SetDisks { record_failed_dirty_bucket(&failed_dirty_buckets_clone, &bucket.name).await; } if partial_saved { - debug!( - target: "rustfs::scanner::io", - event = EVENT_SCANNER_CACHE_PERSIST_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_IO, - bucket = %bucket.name, - cache_name = %cache_name, - state = "partial_saved_not_published", - "Scanner partial bucket cache saved without publishing usage aggregate" - ); + log_saved_partial_bucket_cache(&partial_cache, &cache_name); } continue; diff --git a/docs/architecture/scanner-usage-publication.md b/docs/architecture/scanner-usage-publication.md index 42cf6ba0d..67c152eb5 100644 --- a/docs/architecture/scanner-usage-publication.md +++ b/docs/architecture/scanner-usage-publication.md @@ -230,7 +230,9 @@ worker finishing does not certify its namespace coverage. | Signal | Meaning and next evidence | | --- | --- | -| `bucket_sweep_incomplete` | A bucket reached the end of a sweep but needs another verification pass. Inspect `coverage_gap`, `root_failed_objects`, `failed_objects`, `size_reconciliation_entries`, and `mixed_coverage` with its bucket, pool, set, and cycle. Multiple blockers can coexist. | +| `bucket_checkpoint_prepared` | Checkpoint admission identifies the bucket, pool/set, cycle/leader epoch, and previous cycle/epoch. `resume_position` distinguishes retained traversal position from a fresh sweep; `mixed_coverage` reports whether that retained prefix began under another plan. | +| `bucket_sweep_incomplete` | A bucket reached the end of a sweep but needs another verification pass. `unresolved_objects` reports the complete coverage/health gate, including failed child-cache entries that can be absent from the root counter and diagnostic failure map. Inspect it with `coverage_gap`, `root_failed_objects`, `failed_objects`, `size_reconciliation_entries`, and `mixed_coverage`, using bucket, pool/set, and cycle/leader epoch. `resume_position_on_entry` identifies a sweep entered with traversal position; `resume_position` reports the returned cache position after terminal cleanup. Multiple blockers can coexist. | +| `partial_saved_not_published` | The local or remote worker successfully saved a terminal Partial cache. Correlate its bucket, pool/set, cycle/leader epoch and `resume_position` with admission and sweep events. This confirms persistence, not complete coverage or authoritative publication. | | `remote_scan_partial` | The coordinator received a Partial terminal response for the named bucket and disk. Correlate the worker's bucket-sweep or checkpoint events; Partial is not a complete bucket root. | | `bucket_result_rejected` | A bucket result did not match the cycle's frozen tier-registry generation and was not applied. | | `set_cache_publish_skipped` | `completed_buckets` counts distinct bucket roots successfully applied to this set cache. Use `first_incomplete_bucket` with pool/set/cycle to locate one actual gap in this set. Inspect `unprocessed_buckets` and budget/cancellation state, then correlate that bucket's worker events. | @@ -245,6 +247,22 @@ under a newer plan; failed objects and size reconciliation require different follow-up. A global dirty-generation change alone does not establish which bucket failed to produce a complete root. +A mixed tail clears traversal positions before returning Partial. If that +terminal cache is saved, the next fresh sweep binds both coverage plans to its +new request, even if ordinary writes continue between cycles. A periodic +checkpoint saved before terminal cleanup still contains traversal position; +resuming it under a changed plan can legitimately be mixed again. Correlate +admission, sweep completion, and terminal persistence for the same bucket and +pool/set across consecutive cycles. Samples from different sets cannot establish +a repeated mixed loop. Check save failures, remote interruption, and leadership +changes if only periodic progress survives. + +An unknown bucket cannot be quarantined as zero in an authoritative total. +Existing pending-bucket semantics still require complete structural coverage. +A separately labelled subtotal would need proof for each included bucket across +every source, with missing buckets explicitly unavailable or stale; it must not +advance full-census authority or acknowledge their dirty work. + Use a bounded DEBUG window on the capacity-scanner leader and the workers serving the affected bucket: