mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-27 06:18:43 +00:00
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 <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
@@ -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<u64>,
|
||||
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())
|
||||
}
|
||||
|
||||
@@ -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 '-'"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -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<FolderRe
|
||||
.map(|_| FolderResumeMatch::Descendant)
|
||||
}
|
||||
|
||||
fn folder_is_covered_by_frontier(folder: &str, frontier: &str) -> bool {
|
||||
folder <= frontier && !matches!(folder_resume_match(folder, frontier), Some(FolderResumeMatch::Descendant))
|
||||
}
|
||||
|
||||
fn order_items_for_resume<T, F>(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<ScannerCycleBudgetReason>) -> 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<String> {
|
||||
let mut paths = cache
|
||||
.cache
|
||||
.iter()
|
||||
.filter(|(_, entry)| entry.failed_objects > 0)
|
||||
.map(|(path, _)| path.clone())
|
||||
.collect::<Vec<_>>();
|
||||
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<String>,
|
||||
resume_frontier: Option<String>,
|
||||
resume_failed_paths: Vec<String>,
|
||||
coverage_gap: bool,
|
||||
raw_enumeration_progress: Vec<RawEnumerationProgress>,
|
||||
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
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user