mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-14 00:20:12 +00:00
fix(storage): default new bucket durability to strict (#8253)
* fix(storage): align new bucket durability with strict defaults Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(scanner): defer failure recovery until scan completion Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
@@ -718,6 +718,8 @@ pub struct FolderScanner {
|
||||
|
||||
failed_object_ttl_secs: u64,
|
||||
failed_objects_max: usize,
|
||||
failed_object_paths_seen: HashSet<String>,
|
||||
resolved_failed_object_paths: HashSet<String>,
|
||||
|
||||
sleeper: DynamicSleeper,
|
||||
// should_heal: Arc<dyn Fn() -> bool + Send + Sync>,
|
||||
@@ -927,7 +929,7 @@ impl FolderScanner {
|
||||
into.add_child(child_hash);
|
||||
}
|
||||
|
||||
fn should_skip_failed(&self, path: &str) -> bool {
|
||||
fn failed_retry_suppressed(&self, path: &str) -> bool {
|
||||
let ttl = self.failed_object_ttl_secs;
|
||||
if ttl == 0 {
|
||||
return false;
|
||||
@@ -947,6 +949,7 @@ impl FolderScanner {
|
||||
return;
|
||||
}
|
||||
|
||||
self.resolved_failed_object_paths.remove(path);
|
||||
let now = Self::now_secs();
|
||||
self.new_cache.info.failed_objects.insert(path.to_string(), now);
|
||||
|
||||
@@ -954,6 +957,43 @@ impl FolderScanner {
|
||||
if max_entries > 0 && self.new_cache.info.failed_objects.len() > max_entries {
|
||||
self.prune_failed_objects(now, ttl);
|
||||
}
|
||||
if self.new_cache.info.failed_objects.contains_key(path) {
|
||||
self.failed_object_paths_seen.insert(path.to_string());
|
||||
}
|
||||
}
|
||||
|
||||
fn mark_failed_path_seen(&mut self, path: &str) -> bool {
|
||||
if self.new_cache.info.failed_objects.contains_key(path) {
|
||||
return !self.failed_object_paths_seen.insert(path.to_string());
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
fn clear_failed_path(&mut self, path: &str) {
|
||||
if self.new_cache.info.failed_objects.contains_key(path) {
|
||||
self.resolved_failed_object_paths.insert(path.to_string());
|
||||
}
|
||||
self.failed_object_paths_seen.remove(path);
|
||||
}
|
||||
|
||||
fn reconcile_failed_objects_after_full_scan(&mut self) {
|
||||
self.new_cache
|
||||
.info
|
||||
.failed_objects
|
||||
.retain(|path, _| !self.resolved_failed_object_paths.contains(path));
|
||||
self.resolved_failed_object_paths.clear();
|
||||
|
||||
let mixed_coverage = self
|
||||
.new_cache
|
||||
.info
|
||||
.scan_progress
|
||||
.is_some_and(|progress| progress.started_plan != progress.requested_plan);
|
||||
if self.prefix_scan_scope.is_some() || self.resume_frontier.is_some() || self.coverage_gap || mixed_coverage {
|
||||
return;
|
||||
}
|
||||
|
||||
let seen = &self.failed_object_paths_seen;
|
||||
self.new_cache.info.failed_objects.retain(|path, _| seen.contains(path));
|
||||
}
|
||||
|
||||
fn prune_failed_objects_cache(&mut self) {
|
||||
@@ -1795,12 +1835,14 @@ impl FolderScanner {
|
||||
file_type: entry_type,
|
||||
};
|
||||
|
||||
// If this path is already known as failed, just skip it.
|
||||
// We intentionally do NOT call `record_failed` or bump `failed_objects` here,
|
||||
// because the failure was recorded when the original error occurred
|
||||
// (e.g. in the get_size error branch below). This branch only accounts
|
||||
// for subsequent skips of already-failed paths.
|
||||
if self.should_skip_failed(&item.path) {
|
||||
// Keep failed paths visible to this scan so repair success can
|
||||
// clear stale failure state and a complete walk can discard
|
||||
// paths that were removed since the previous cycle.
|
||||
let repeated_failed_path = self.mark_failed_path_seen(&item.path);
|
||||
if repeated_failed_path && self.failed_retry_suppressed(&item.path) {
|
||||
// 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();
|
||||
continue;
|
||||
}
|
||||
@@ -1813,14 +1855,17 @@ impl FolderScanner {
|
||||
Ok(sz) => sz,
|
||||
Err(e) => {
|
||||
let failure_action = classify_get_size_failure(&item, &e);
|
||||
let retry_suppressed = self.failed_retry_suppressed(&item.path);
|
||||
|
||||
if failure_action != GetSizeFailureAction::Skip {
|
||||
self.resolved_failed_object_paths.remove(&item.path);
|
||||
self.coverage_gap |= self.old_cache.info.scan_progress.is_some();
|
||||
// Track failed objects to prevent infinite retry loops
|
||||
into.failed_objects += 1;
|
||||
self.record_failed(&item.path);
|
||||
if !retry_suppressed {
|
||||
into.failed_objects += 1;
|
||||
self.record_failed(&item.path);
|
||||
}
|
||||
|
||||
if should_log_failed_object(into.failed_objects) {
|
||||
if !retry_suppressed && should_log_failed_object(into.failed_objects) {
|
||||
if let GetSizeFailureAction::HealMetadata { object } = &failure_action {
|
||||
error!(
|
||||
target: "rustfs::scanner::folder",
|
||||
@@ -1850,9 +1895,13 @@ impl FolderScanner {
|
||||
);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
self.clear_failed_path(&item.path);
|
||||
}
|
||||
|
||||
if let GetSizeFailureAction::HealMetadata { object } = failure_action {
|
||||
if let GetSizeFailureAction::HealMetadata { object } = failure_action
|
||||
&& !retry_suppressed
|
||||
{
|
||||
// Single-flight (backlog#1894 axis A) — the
|
||||
// recording mode and its guarantees are pinned by
|
||||
// corrupt_metadata_recording below.
|
||||
@@ -1901,6 +1950,7 @@ impl FolderScanner {
|
||||
}
|
||||
};
|
||||
|
||||
self.clear_failed_path(&item.path);
|
||||
found_object_metadata = true;
|
||||
|
||||
item.transform_meta_dir();
|
||||
@@ -1967,8 +2017,10 @@ impl FolderScanner {
|
||||
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);
|
||||
|
||||
if !self.should_skip_failed(&metadata_path) {
|
||||
if !retry_suppressed {
|
||||
into.failed_objects = into.failed_objects.saturating_add(1);
|
||||
self.record_failed(&metadata_path);
|
||||
|
||||
@@ -1985,7 +2037,9 @@ impl FolderScanner {
|
||||
"Scanner found erasure object data without metadata"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if !retry_suppressed {
|
||||
let (bucket, object) = path2_bucket_object_with_base_path(&self.root, &folder.name);
|
||||
if !bucket.is_empty() && !object.is_empty() {
|
||||
self.send_required_scanner_heal_request(
|
||||
@@ -2769,6 +2823,8 @@ pub(crate) async fn scan_data_folder_scoped(
|
||||
prefix_scan_scope,
|
||||
failed_object_ttl_secs: failed_object_ttl,
|
||||
failed_objects_max,
|
||||
failed_object_paths_seen: HashSet::new(),
|
||||
resolved_failed_object_paths: HashSet::new(),
|
||||
sleeper,
|
||||
disks,
|
||||
disks_quorum,
|
||||
@@ -2823,6 +2879,7 @@ pub(crate) async fn scan_data_folder_scoped(
|
||||
Ok(()) => {
|
||||
// Get the new cache and finalize it
|
||||
let coverage_gap = scanner.coverage_gap;
|
||||
scanner.reconcile_failed_objects_after_full_scan();
|
||||
let new_cache = scanner.as_mut_new_cache();
|
||||
new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
|
||||
new_cache.info.last_update = Some(SystemTime::now());
|
||||
|
||||
@@ -336,6 +336,8 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
|
||||
prefix_scan_scope: None,
|
||||
failed_object_ttl_secs: u64::MAX,
|
||||
failed_objects_max: usize::MAX,
|
||||
failed_object_paths_seen: HashSet::new(),
|
||||
resolved_failed_object_paths: HashSet::new(),
|
||||
sleeper: SCANNER_SLEEPER.clone(),
|
||||
disks: Vec::new(),
|
||||
disks_quorum: 0,
|
||||
@@ -454,7 +456,7 @@ impl Drop for TestGuard {
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_should_skip_failed_respects_ttl() {
|
||||
async fn test_failed_retry_suppression_respects_ttl() {
|
||||
let (mut scanner, temp_dir) = build_test_scanner().await;
|
||||
let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir);
|
||||
let now = FolderScanner::now_secs();
|
||||
@@ -470,8 +472,15 @@ async fn test_should_skip_failed_respects_ttl() {
|
||||
.failed_objects
|
||||
.insert("expired".to_string(), now.saturating_sub(120));
|
||||
|
||||
assert!(scanner.should_skip_failed("recent"));
|
||||
assert!(!scanner.should_skip_failed("expired"));
|
||||
assert!(!scanner.mark_failed_path_seen("recent"), "first observation gets one retry");
|
||||
assert!(
|
||||
scanner.mark_failed_path_seen("recent"),
|
||||
"duplicate observation is eligible for TTL suppression"
|
||||
);
|
||||
assert!(!scanner.mark_failed_path_seen("expired"), "expired failure is observed for a fresh retry");
|
||||
assert!(scanner.mark_failed_path_seen("expired"), "duplicate expired entry is marked in this pass");
|
||||
assert!(scanner.failed_retry_suppressed("recent"));
|
||||
assert!(!scanner.failed_retry_suppressed("expired"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -485,7 +494,7 @@ async fn test_record_failed_ttl_zero_noop() {
|
||||
|
||||
let now = FolderScanner::now_secs();
|
||||
scanner.new_cache.info.failed_objects.insert("path2".to_string(), now);
|
||||
assert!(!scanner.should_skip_failed("path2"));
|
||||
assert!(!scanner.failed_retry_suppressed("path2"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1105,6 +1114,72 @@ async fn test_prune_failed_objects_cache_drops_expired() {
|
||||
assert!(scanner.new_cache.info.failed_objects.contains_key("fresh"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_failed_objects_clear_on_recovery_and_full_scan_reconciliation() {
|
||||
let (mut scanner, temp_dir) = build_test_scanner().await;
|
||||
let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir);
|
||||
let now = FolderScanner::now_secs();
|
||||
scanner
|
||||
.new_cache
|
||||
.info
|
||||
.failed_objects
|
||||
.insert("bucket/recovered/xl.meta".to_string(), now);
|
||||
scanner
|
||||
.new_cache
|
||||
.info
|
||||
.failed_objects
|
||||
.insert("bucket/removed/xl.meta".to_string(), now);
|
||||
scanner
|
||||
.new_cache
|
||||
.info
|
||||
.failed_objects
|
||||
.insert("bucket/still-failed/xl.meta".to_string(), now);
|
||||
|
||||
scanner.clear_failed_path("bucket/recovered/xl.meta");
|
||||
assert!(
|
||||
scanner.new_cache.info.failed_objects.contains_key("bucket/recovered/xl.meta"),
|
||||
"recovery is committed only after the scan completes"
|
||||
);
|
||||
scanner.mark_failed_path_seen("bucket/still-failed/xl.meta");
|
||||
scanner.reconcile_failed_objects_after_full_scan();
|
||||
|
||||
assert_eq!(
|
||||
scanner
|
||||
.new_cache
|
||||
.info
|
||||
.failed_objects
|
||||
.keys()
|
||||
.map(String::as_str)
|
||||
.collect::<Vec<_>>(),
|
||||
["bucket/still-failed/xl.meta"]
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_checkpoint_resume_preserves_failed_paths_outside_visited_suffix() {
|
||||
let (mut scanner, temp_dir) = build_test_scanner().await;
|
||||
let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir);
|
||||
scanner.resume_frontier = Some("bucket/resume-after".to_string());
|
||||
scanner
|
||||
.new_cache
|
||||
.info
|
||||
.failed_objects
|
||||
.insert("bucket/prefix-failure/xl.meta".to_string(), FolderScanner::now_secs());
|
||||
|
||||
scanner.reconcile_failed_objects_after_full_scan();
|
||||
|
||||
assert!(
|
||||
scanner
|
||||
.new_cache
|
||||
.info
|
||||
.failed_objects
|
||||
.contains_key("bucket/prefix-failure/xl.meta"),
|
||||
"a resumed suffix scan cannot prune failure state for its unvisited prefix"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_prune_failed_objects_max_zero_keeps_fresh() {
|
||||
@@ -2813,6 +2888,57 @@ async fn test_scan_data_folder_returns_partial_cache_on_budget_cancel() {
|
||||
assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Directories));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn full_scan_clears_failed_cache_for_manually_removed_object() {
|
||||
let (scanner, temp_dir) = build_test_scanner().await;
|
||||
let _guard = TestGuard {
|
||||
temp_dir: Some(temp_dir.clone()),
|
||||
};
|
||||
tokio::fs::create_dir_all(temp_dir.join("bucket"))
|
||||
.await
|
||||
.expect("bucket directory should be created");
|
||||
write_test_object_metadata(&temp_dir, "bucket", "repaired").await;
|
||||
|
||||
let mut info = crate::data_usage_define::DataUsageCacheInfo {
|
||||
name: "bucket".to_string(),
|
||||
..Default::default()
|
||||
};
|
||||
let canonical_root = temp_dir
|
||||
.canonicalize()
|
||||
.expect("test disk root should resolve to the path used by scanner");
|
||||
info.failed_objects.insert(
|
||||
canonical_root.join("bucket/removed/xl.meta").to_string_lossy().into_owned(),
|
||||
FolderScanner::now_secs(),
|
||||
);
|
||||
info.failed_objects.insert(
|
||||
canonical_root.join("bucket/repaired/xl.meta").to_string_lossy().into_owned(),
|
||||
FolderScanner::now_secs(),
|
||||
);
|
||||
let cache = DataUsageCache {
|
||||
info,
|
||||
..Default::default()
|
||||
};
|
||||
let parent = CancellationToken::new();
|
||||
let budget = ScannerCycleBudget::new(&parent, Default::default());
|
||||
|
||||
let scanned = scan_data_folder(
|
||||
budget.token(),
|
||||
budget,
|
||||
vec![scanner.local_disk.clone()],
|
||||
scanner.local_disk.clone(),
|
||||
cache,
|
||||
None,
|
||||
HealScanMode::Normal,
|
||||
SCANNER_SLEEPER.clone(),
|
||||
)
|
||||
.await
|
||||
.expect("complete bucket scan should succeed");
|
||||
|
||||
assert!(scanned.info.snapshot_complete);
|
||||
assert!(scanned.info.failed_objects.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_scan_data_folder_returns_raw_cursor_on_enumeration_cancel_without_root_progress() {
|
||||
@@ -3486,11 +3612,12 @@ async fn test_scan_data_folder_keeps_unresolved_objects_partial() {
|
||||
let _guard = TestGuard {
|
||||
temp_dir: Some(temp_dir.clone()),
|
||||
};
|
||||
write_test_object_metadata(&temp_dir, "bucket", "object").await;
|
||||
write_test_object_metadata_bytes(&temp_dir, "bucket", "object", &[]).await;
|
||||
|
||||
let failed_path = temp_dir
|
||||
.join("bucket")
|
||||
.join("object")
|
||||
.canonicalize()
|
||||
.expect("test disk root should resolve to the path used by scanner")
|
||||
.join("bucket/object")
|
||||
.join(STORAGE_FORMAT_FILE)
|
||||
.to_string_lossy()
|
||||
.into_owned();
|
||||
@@ -3502,7 +3629,10 @@ async fn test_scan_data_folder_keeps_unresolved_objects_partial() {
|
||||
},
|
||||
..Default::default()
|
||||
};
|
||||
cache.info.failed_objects.insert(failed_path, FolderScanner::now_secs());
|
||||
cache
|
||||
.info
|
||||
.failed_objects
|
||||
.insert(failed_path.clone(), FolderScanner::now_secs());
|
||||
|
||||
let parent = CancellationToken::new();
|
||||
let budget = ScannerCycleBudget::new(&parent, Default::default());
|
||||
@@ -3523,7 +3653,11 @@ async fn test_scan_data_folder_keeps_unresolved_objects_partial() {
|
||||
other => panic!("expected unresolved object to keep the cache partial, got {other:?}"),
|
||||
};
|
||||
assert!(!partial.info.snapshot_complete);
|
||||
assert!(!partial.info.failed_objects.is_empty());
|
||||
assert!(
|
||||
partial.info.failed_objects.contains_key(&failed_path),
|
||||
"corrupt object failure should remain recorded: {:?}",
|
||||
partial.info.failed_objects
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -3654,10 +3654,9 @@ async fn get_size_marks_corrupt_metadata_for_heal() {
|
||||
tokio::fs::create_dir_all(&object_dir)
|
||||
.await
|
||||
.expect("failed to create object directory");
|
||||
tokio::fs::write(&metadata_path, b"not-valid-filemeta")
|
||||
tokio::fs::write(&metadata_path, b"")
|
||||
.await
|
||||
.expect("failed to write corrupt metadata");
|
||||
|
||||
.expect("empty metadata fixture should be created");
|
||||
let endpoint = Endpoint::try_from(temp_dir.to_string_lossy().as_ref()).expect("failed to create endpoint");
|
||||
let disk = new_disk(
|
||||
&endpoint,
|
||||
@@ -3690,11 +3689,16 @@ async fn get_size_marks_corrupt_metadata_for_heal() {
|
||||
debug: false,
|
||||
};
|
||||
|
||||
let err = disk
|
||||
.get_size(item)
|
||||
.await
|
||||
.expect_err("corrupt metadata should be surfaced as scanner-heal work");
|
||||
assert!(is_scanner_metadata_corrupt_error(&err));
|
||||
for contents in [b"".as_slice(), b"not-valid-filemeta".as_slice()] {
|
||||
tokio::fs::write(&metadata_path, contents)
|
||||
.await
|
||||
.expect("corrupt metadata fixture should be written");
|
||||
let err = disk
|
||||
.get_size(item.clone())
|
||||
.await
|
||||
.expect_err("corrupt metadata should be surfaced as scanner-heal work");
|
||||
assert!(is_scanner_metadata_corrupt_error(&err));
|
||||
}
|
||||
|
||||
let _ = tokio::fs::remove_dir_all(&temp_dir).await;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user