mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-26 22:07:53 +00:00
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 <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * 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 <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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::<serde_json::Value>(line).expect("scanner JSON event"))
|
||||
.collect::<Vec<_>>();
|
||||
let admissions = all_events
|
||||
.iter()
|
||||
.filter(|event| event["fields"]["state"] == "bucket_checkpoint_prepared")
|
||||
.collect::<Vec<_>>();
|
||||
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::<Vec<_>>();
|
||||
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);
|
||||
|
||||
@@ -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::<serde_json::Value>(line).expect("scanner JSON event"))
|
||||
.filter(|event| event["fields"]["state"] == "bucket_sweep_incomplete")
|
||||
.collect::<Vec<_>>();
|
||||
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]
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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])
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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:
|
||||
|
||||
|
||||
Reference in New Issue
Block a user