From e73b17aff66766688d9eb65336480bbece473bee Mon Sep 17 00:00:00 2001 From: LoganZ2 <103290230+LoganZ2@users.noreply.github.com> Date: Fri, 27 Feb 2026 22:25:52 +0800 Subject: [PATCH] fix(scanner): skip recent IO-error objects (#1860) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: LoganZ2 <103290230+LoganZ2@users.noreply.github.com> Co-authored-by: houseme Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: heihutu <30542132+heihutu@users.noreply.github.com> Co-authored-by: loverustfs Co-authored-by: 安正超 --- crates/scanner/src/data_usage_define.rs | 39 +++ crates/scanner/src/scanner_folder.rs | 316 +++++++++++++++++++++++- crates/scanner/src/scanner_io.rs | 4 +- rustfs/src/admin/console_test.rs | 6 + rustfs/src/config/config_test.rs | 8 + 5 files changed, 362 insertions(+), 11 deletions(-) diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 090c1f137..aeecbaf7b 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -504,6 +504,9 @@ pub struct DataUsageEntry { pub obj_versions: VersionsHistogram, pub replication_stats: Option, pub compacted: bool, + /// Number of objects that failed to scan (e.g., IO errors) + #[serde(default)] + pub failed_objects: usize, } impl DataUsageEntry { @@ -541,6 +544,7 @@ impl DataUsageEntry { self.versions += other.versions; self.delete_markers += other.delete_markers; self.size += other.size; + self.failed_objects += other.failed_objects; if let Some(o_rep) = &other.replication_stats { if self.replication_stats.is_none() { @@ -590,6 +594,8 @@ pub struct DataUsageCacheInfo { pub skip_healing: bool, pub lifecycle: Option>, pub replication: Option>, + #[serde(default)] + pub failed_objects: HashMap, } /// Data usage cache @@ -1541,6 +1547,7 @@ impl SizeSummary { #[cfg(test)] mod tests { use super::*; + use serde_json::Value; #[test] fn test_data_usage_info_creation() { @@ -1587,4 +1594,36 @@ mod tests { assert_eq!(summary1.total_size, 300); assert_eq!(summary1.versions, 15); } + + #[test] + fn test_data_usage_entry_merge_sums_failed_objects() { + let mut left = DataUsageEntry { + failed_objects: 2, + ..Default::default() + }; + + let right = DataUsageEntry { + failed_objects: 3, + ..Default::default() + }; + + left.merge(&right); + + assert_eq!(left.failed_objects, 5); + } + + #[test] + fn test_data_usage_entry_deserialize_defaults_failed_objects() { + let entry = DataUsageEntry::default(); + let mut value = serde_json::to_value(&entry).expect("Failed to serialize entry"); + + let Value::Object(ref mut map) = value else { + panic!("Expected entry to serialize into an object"); + }; + + map.remove("failed_objects"); + + let decoded: DataUsageEntry = serde_json::from_value(value).expect("Failed to deserialize entry"); + assert_eq!(decoded.failed_objects, 0); + } } diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 9cc48e258..00c2604d7 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -60,6 +60,10 @@ const DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS: usize = 250_000; const DEFAULT_HEAL_OBJECT_SELECT_PROB: u32 = 1024; const ENV_DATA_USAGE_UPDATE_DIR_CYCLES: &str = "RUSTFS_DATA_USAGE_UPDATE_DIR_CYCLES"; const ENV_HEAL_OBJECT_SELECT_PROB: &str = "RUSTFS_HEAL_OBJECT_SELECT_PROB"; +const ENV_FAILED_OBJECT_TTL_SECS: &str = "RUSTFS_DATA_USAGE_FAILED_OBJECT_TTL_SECS"; +const ENV_FAILED_OBJECTS_MAX: &str = "RUSTFS_DATA_USAGE_FAILED_OBJECTS_MAX"; +const DEFAULT_FAILED_OBJECT_TTL_SECS: u32 = 86_400; +const DEFAULT_FAILED_OBJECTS_MAX: u32 = 10_000; pub fn data_usage_update_dir_cycles() -> u32 { rustfs_utils::get_env_u32(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, DATA_USAGE_UPDATE_DIR_CYCLES) @@ -390,6 +394,9 @@ pub struct FolderScanner { heal_object_select: u32, scan_mode: HealScanMode, + failed_object_ttl_secs: u64, + failed_objects_max: usize, + we_sleep: Box bool + Send + Sync>, // should_heal: Arc bool + Send + Sync>, disks: Vec>, @@ -405,6 +412,78 @@ pub struct FolderScanner { } impl FolderScanner { + fn now_secs() -> u64 { + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() + } + + fn should_skip_failed(&self, path: &str) -> bool { + let ttl = self.failed_object_ttl_secs; + if ttl == 0 { + return false; + } + + let Some(last_failed) = self.new_cache.info.failed_objects.get(path) else { + return false; + }; + + let now = Self::now_secs(); + now.saturating_sub(*last_failed) < ttl + } + + fn record_failed(&mut self, path: &str) { + let ttl = self.failed_object_ttl_secs; + if ttl == 0 { + return; + } + + let now = Self::now_secs(); + self.new_cache.info.failed_objects.insert(path.to_string(), now); + + let max_entries = self.failed_objects_max; + if max_entries > 0 && self.new_cache.info.failed_objects.len() > max_entries { + self.prune_failed_objects(now, ttl); + } + } + + fn prune_failed_objects_cache(&mut self) { + let ttl = self.failed_object_ttl_secs; + if ttl == 0 { + return; + } + + let now = Self::now_secs(); + self.prune_failed_objects(now, ttl); + } + + fn prune_failed_objects(&mut self, now: u64, ttl: u64) { + let max_entries = self.failed_objects_max; + let failed = &mut self.new_cache.info.failed_objects; + if failed.is_empty() { + return; + } + + failed.retain(|_, ts| now.saturating_sub(*ts) < ttl); + + if max_entries == 0 { + return; + } + + if failed.len() <= max_entries { + return; + } + + let mut entries: Vec<(String, u64)> = failed.iter().map(|(k, v)| (k.clone(), *v)).collect(); + entries.sort_by(|(k1, ts1), (k2, ts2)| ts1.cmp(ts2).then_with(|| k1.cmp(k2))); + + let remove_count = failed.len().saturating_sub(max_entries); + for (key, _) in entries.into_iter().take(remove_count) { + failed.remove(&key); + } + } + pub async fn should_heal(&self) -> bool { if self.skip_heal.load(std::sync::atomic::Ordering::Relaxed) { return false; @@ -489,6 +568,8 @@ impl FolderScanner { return Err(ScannerError::Other("Operation cancelled".to_string())); } + self.prune_failed_objects_cache(); + let mut abandoned_children: DataUsageHashMap = HashSet::new(); if !into.compacted { abandoned_children = self.old_cache.find_children_copy(this_hash.clone()); @@ -622,20 +703,36 @@ 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) { + continue; + } + let sz = match self.local_disk.get_size(item.clone()).await { Ok(sz) => sz, Err(e) => { - warn!("scan_folder: failed to get size for item {}: {}", item.path, e); - // TODO: check error type - if let Some(t) = wait - && let Ok(elapsed) = t.elapsed() - { - tokio::time::sleep(elapsed).await; + let is_skip_file = matches!(e, StorageError::Io(ref io) if io.to_string() == "skip file"); + + if !is_skip_file { + // Track failed objects to prevent infinite retry loops + into.failed_objects += 1; + self.record_failed(&item.path); + + // Only log non-skip errors to avoid noise + warn!("scan_folder: failed to get size for item {}: {}", item.path, e); + + // Apply sleep if configured + if let Some(t) = wait + && let Ok(elapsed) = t.elapsed() + { + tokio::time::sleep(elapsed).await; + } } - if e != StorageError::other("skip file".to_string()) { - warn!("scan_folder: failed to get size for item {}: {}", item.path, e); - } continue; } }; @@ -1107,6 +1204,9 @@ pub async fn scan_data_folder( let disks_quorum = disks.len() / 2; + let failed_object_ttl = rustfs_utils::get_env_u32(ENV_FAILED_OBJECT_TTL_SECS, DEFAULT_FAILED_OBJECT_TTL_SECS) as u64; + let failed_objects_max = rustfs_utils::get_env_u32(ENV_FAILED_OBJECTS_MAX, DEFAULT_FAILED_OBJECTS_MAX) as usize; + // Create folder scanner let mut scanner = FolderScanner { root: base_path, @@ -1122,6 +1222,8 @@ pub async fn scan_data_folder( data_usage_scanner_debug: false, heal_object_select, scan_mode, + failed_object_ttl_secs: failed_object_ttl, + failed_objects_max, we_sleep, disks, disks_quorum, @@ -1164,3 +1266,199 @@ pub async fn scan_data_folder( } } } + +#[cfg(test)] +mod tests { + use super::*; + use rustfs_ecstore::disk::{DiskOption, endpoint::Endpoint, new_disk}; + use serial_test::serial; + use std::sync::atomic::AtomicBool; + use uuid::Uuid; + + async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) { + let temp_dir = std::env::temp_dir().join(format!("rustfs-scanner-test-{}", Uuid::new_v4())); + tokio::fs::create_dir_all(&temp_dir) + .await + .expect("failed to create test directory"); + + let endpoint = Endpoint::try_from(temp_dir.to_string_lossy().as_ref()).expect("failed to create endpoint"); + let disk = new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("failed to create disk"); + + let update_current_path: UpdateCurrentPathFn = Arc::new(|_: &str| Box::pin(async {})); + + let scanner = FolderScanner { + root: temp_dir.to_string_lossy().to_string(), + old_cache: DataUsageCache::default(), + new_cache: DataUsageCache::default(), + update_cache: DataUsageCache::default(), + data_usage_scanner_debug: false, + heal_object_select: 0, + scan_mode: HealScanMode::Normal, + failed_object_ttl_secs: u64::MAX, + failed_objects_max: usize::MAX, + we_sleep: Box::new(|| false), + disks: Vec::new(), + disks_quorum: 0, + updates: None, + last_update: SystemTime::UNIX_EPOCH, + update_current_path, + skip_heal: Arc::new(AtomicBool::new(false)), + local_disk: disk, + }; + + (scanner, temp_dir) + } + + struct TestGuard { + temp_dir: Option, + } + + impl TestGuard { + fn new(ttl: u64, max: usize, scanner: &mut FolderScanner, temp_dir: std::path::PathBuf) -> Self { + scanner.failed_object_ttl_secs = ttl; + scanner.failed_objects_max = max; + Self { + temp_dir: Some(temp_dir), + } + } + } + + impl Drop for TestGuard { + fn drop(&mut self) { + if let Some(temp_dir) = self.temp_dir.take() { + let _ = std::fs::remove_dir_all(&temp_dir); + } + } + } + + #[tokio::test] + #[serial] + async fn test_should_skip_failed_respects_ttl() { + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone()); + let now = FolderScanner::now_secs(); + + scanner + .new_cache + .info + .failed_objects + .insert("recent".to_string(), now.saturating_sub(10)); + scanner + .new_cache + .info + .failed_objects + .insert("expired".to_string(), now.saturating_sub(120)); + + assert!(scanner.should_skip_failed("recent")); + assert!(!scanner.should_skip_failed("expired")); + } + + #[tokio::test] + #[serial] + async fn test_record_failed_ttl_zero_noop() { + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(0, 100, &mut scanner, temp_dir.clone()); + + scanner.record_failed("path1"); + assert!(scanner.new_cache.info.failed_objects.is_empty()); + + let now = FolderScanner::now_secs(); + scanner.new_cache.info.failed_objects.insert("path2".to_string(), now); + assert!(!scanner.should_skip_failed("path2")); + } + + #[tokio::test] + #[serial] + async fn test_record_failed_prunes_to_max_entries() { + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(1000, 2, &mut scanner, temp_dir.clone()); + let now = FolderScanner::now_secs(); + + scanner + .new_cache + .info + .failed_objects + .insert("old1".to_string(), now.saturating_sub(50)); + scanner + .new_cache + .info + .failed_objects + .insert("old2".to_string(), now.saturating_sub(40)); + scanner + .new_cache + .info + .failed_objects + .insert("old3".to_string(), now.saturating_sub(30)); + + scanner.record_failed("new"); + + assert_eq!(scanner.new_cache.info.failed_objects.len(), 2); + assert!(scanner.new_cache.info.failed_objects.contains_key("new")); + assert!(scanner.new_cache.info.failed_objects.contains_key("old3")); + assert!(!scanner.new_cache.info.failed_objects.contains_key("old1")); + assert!(!scanner.new_cache.info.failed_objects.contains_key("old2")); + } + + #[tokio::test] + #[serial] + async fn test_prune_failed_objects_cache_drops_expired() { + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(5, 10, &mut scanner, temp_dir.clone()); + let now = FolderScanner::now_secs(); + + scanner + .new_cache + .info + .failed_objects + .insert("expired".to_string(), now.saturating_sub(10)); + scanner + .new_cache + .info + .failed_objects + .insert("fresh".to_string(), now.saturating_sub(2)); + + scanner.prune_failed_objects_cache(); + + assert_eq!(scanner.new_cache.info.failed_objects.len(), 1); + assert!(scanner.new_cache.info.failed_objects.contains_key("fresh")); + } + + #[tokio::test] + #[serial] + async fn test_prune_failed_objects_max_zero_keeps_fresh() { + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(60, 0, &mut scanner, temp_dir.clone()); + let now = FolderScanner::now_secs(); + + scanner + .new_cache + .info + .failed_objects + .insert("fresh1".to_string(), now.saturating_sub(5)); + scanner + .new_cache + .info + .failed_objects + .insert("fresh2".to_string(), now.saturating_sub(10)); + scanner + .new_cache + .info + .failed_objects + .insert("expired".to_string(), now.saturating_sub(120)); + + scanner.prune_failed_objects_cache(); + + assert_eq!(scanner.new_cache.info.failed_objects.len(), 2); + assert!(scanner.new_cache.info.failed_objects.contains_key("fresh1")); + assert!(scanner.new_cache.info.failed_objects.contains_key("fresh2")); + assert!(!scanner.new_cache.info.failed_objects.contains_key("expired")); + } +} diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 0067fed16..f5e6c9390 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -499,7 +499,7 @@ impl ScannerIODisk for Disk { &item.object_path() ); - return Err(StorageError::other("skip file".to_string())); + return Err(StorageError::other("failed to read metadata".to_string())); } }; @@ -510,7 +510,7 @@ impl ScannerIODisk for Disk { Ok(versions) => versions, Err(e) => { error!("Failed to get file info versions: {}", e); - return Err(StorageError::other("skip file".to_string())); + return Err(StorageError::other("failed to get file info".to_string())); } }; diff --git a/rustfs/src/admin/console_test.rs b/rustfs/src/admin/console_test.rs index 9977b0713..65e58e518 100644 --- a/rustfs/src/admin/console_test.rs +++ b/rustfs/src/admin/console_test.rs @@ -16,8 +16,10 @@ mod tests { use crate::config::Opt; use clap::Parser; + use serial_test::serial; #[tokio::test] + #[serial] async fn test_console_cors_configuration() { // Test CORS configuration parsing use crate::admin::console::parse_cors_origins; @@ -42,6 +44,7 @@ mod tests { } #[tokio::test] + #[serial] async fn test_console_tls_configuration() { // Test TLS configuration options (now uses shared tls_path) let args = vec!["rustfs", "/tmp/test", "--tls-path", "/path/to/tls"]; @@ -51,6 +54,7 @@ mod tests { } #[tokio::test] + #[serial] async fn test_console_health_check_endpoint() { // Test that console health check can be called // This test would need a running server to be comprehensive @@ -63,6 +67,7 @@ mod tests { } #[tokio::test] + #[serial] async fn test_console_separate_logging_target() { // Test that console uses separate logging targets use tracing::info; @@ -77,6 +82,7 @@ mod tests { } #[tokio::test] + #[serial] async fn test_console_configuration_validation() { // Test configuration validation let args = vec![ diff --git a/rustfs/src/config/config_test.rs b/rustfs/src/config/config_test.rs index 44ee41799..86d1286ee 100644 --- a/rustfs/src/config/config_test.rs +++ b/rustfs/src/config/config_test.rs @@ -57,6 +57,7 @@ mod tests { } #[test] + #[serial] fn test_default_console_configuration() { // Test that default console configuration is correct let args = vec!["rustfs", "/test/volume"]; @@ -68,6 +69,7 @@ mod tests { } #[test] + #[serial] fn test_custom_console_configuration() { // Test custom console configuration let args = vec![ @@ -88,6 +90,7 @@ mod tests { } #[test] + #[serial] fn test_console_and_endpoint_ports_different() { // Ensure console and endpoint use different default ports let args = vec!["rustfs", "/test/volume"]; @@ -107,6 +110,7 @@ mod tests { } #[test] + #[serial] fn test_volumes_and_disk_layout_parsing() { use rustfs_ecstore::disks_layout::DisksLayout; @@ -436,6 +440,7 @@ mod tests { /// Test error handling for invalid ellipses patterns. #[test] + #[serial] fn test_invalid_ellipses_patterns() { // Test case 1: Invalid ellipses format (letters instead of numbers) let args = vec!["rustfs", "/data/vol{a...z}"]; @@ -458,6 +463,7 @@ mod tests { } #[test] + #[serial] fn test_server_domains_parsing() { // Test case 1: server domains without ports let args = vec![ @@ -525,6 +531,7 @@ mod tests { } #[test] + #[serial] fn test_access_key_arguments_mutually_exclusive_cli() { // Test that CLI args configuration fails on conflict let args = vec![ @@ -564,6 +571,7 @@ mod tests { } #[test] + #[serial] fn test_secret_key_arguments_mutually_exclusive_cli() { // Test that CLI args configuration fails on conflict let args = vec![