fix(scanner): skip recent IO-error objects (#1860)

Signed-off-by: LoganZ2 <103290230+LoganZ2@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: heihutu <30542132+heihutu@users.noreply.github.com>
Co-authored-by: loverustfs <hello@rustfs.com>
Co-authored-by: 安正超 <anzhengchao@gmail.com>

crates/scanner/src/data_usage_define.rs

@@ -504,6 +504,9 @@ pub struct DataUsageEntry {
     pub obj_versions: VersionsHistogram,
     pub replication_stats: Option<ReplicationAllStats>,
     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<Arc<BucketLifecycleConfiguration>>,
     pub replication: Option<Arc<ReplicationConfig>>,
+    #[serde(default)]
+    pub failed_objects: HashMap<String, u64>,
 }
 
 /// 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);
+    }
 }

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<dyn Fn() -> bool + Send + Sync>,
     // should_heal: Arc<dyn Fn() -> bool + Send + Sync>,
     disks: Vec<Arc<Disk>>,
@@ -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<std::path::PathBuf>,
+    }
+
+    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"));
+    }
+}

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()));
             }
         };
 

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![

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![