fix(scanner): stabilize data usage cache persistence under slow metadata I/O (#2594)

crates/config/README.md

@@ -58,6 +58,7 @@ Current guidance:
 - `RUSTFS_SCANNER_START_DELAY_SECS` (canonical)
 - `RUSTFS_DATA_SCANNER_START_DELAY_SECS` (deprecated alias for compatibility)
 - `RUSTFS_SCANNER_IDLE_MODE` (canonical)
+- `RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS` (canonical)
 
 ## Drive timeout environment variables
 

crates/config/src/constants/scanner.rs

@@ -45,6 +45,12 @@ pub const DEFAULT_SCANNER_SPEED: &str = "default";
 /// - Example: `export RUSTFS_SCANNER_IDLE_MODE=false`
 pub const ENV_SCANNER_IDLE_MODE: &str = "RUSTFS_SCANNER_IDLE_MODE";
 
+/// Environment variable that controls scanner cache save timeout in seconds.
+/// The scanner enforces a minimum value of `1`.
+/// - Unit: seconds (u64).
+/// - Example: `export RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS=30`
+pub const ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS: &str = "RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS";
+
 /// Default scanner idle mode.
 pub const DEFAULT_SCANNER_IDLE_MODE: bool = true;
 

crates/scanner/src/data_usage_define.rs

@@ -17,13 +17,16 @@ use s3s::dto::BucketLifecycleConfiguration;
 use serde::{Deserialize, Serialize};
 use std::{
     collections::{HashMap, HashSet},
+    future::Future,
     hash::{DefaultHasher, Hash, Hasher},
     path::Path,
-    sync::{Arc, LazyLock},
+    sync::{Arc, LazyLock, Once},
     time::SystemTime,
 };
 
 use http::HeaderMap;
+use metrics::{counter, describe_counter, describe_histogram, histogram};
+use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
 use rustfs_ecstore::{
     StorageAPI,
     bucket::{lifecycle::lifecycle::TRANSITION_COMPLETE, replication::ReplicationConfig},
@@ -33,8 +36,8 @@ use rustfs_ecstore::{
     store_api::{ObjectInfo, ObjectOptions},
 };
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
-use tokio::time::{Duration, sleep, timeout};
-use tracing::{error, warn};
+use tokio::time::{Duration, Instant, sleep, timeout};
+use tracing::warn;
 
 // Data usage constants
 pub const DATA_USAGE_ROOT: &str = SLASH_SEPARATOR;
@@ -44,6 +47,15 @@ const DATA_USAGE_OBJ_NAME: &str = ".usage.json";
 const DATA_USAGE_BLOOM_NAME: &str = ".bloomcycle.bin";
 
 pub const DATA_USAGE_CACHE_NAME: &str = ".usage-cache.bin";
+const DATA_USAGE_CACHE_SAVE_TIMEOUT_SECS_DEFAULT: u64 = 30;
+const DATA_USAGE_CACHE_SAVE_RETRIES: u32 = 2;
+const DATA_USAGE_CACHE_BACKUP_SAVE_TIMEOUT_SECS_MAX: u64 = 5;
+const DATA_USAGE_CACHE_BACKUP_SAVE_RETRIES: u32 = 0;
+const METRIC_CACHE_SAVE_ATTEMPT_TOTAL: &str = "rustfs_scanner_cache_save_attempt_total";
+const METRIC_CACHE_SAVE_TIMEOUT_TOTAL: &str = "rustfs_scanner_cache_save_timeout_total";
+const METRIC_CACHE_SAVE_RETRY_TOTAL: &str = "rustfs_scanner_cache_save_retry_total";
+const METRIC_CACHE_SAVE_DURATION_SECONDS: &str = "rustfs_scanner_cache_save_duration_seconds";
+static CACHE_SAVE_METRICS_ONCE: Once = Once::new();
 
 // Data usage paths (computed at runtime)
 pub static DATA_USAGE_BUCKET: LazyLock<String> =
@@ -595,6 +607,31 @@ pub struct DataUsageCache {
 }
 
 impl DataUsageCache {
+    fn ensure_cache_save_metrics_registered() {
+        CACHE_SAVE_METRICS_ONCE.call_once(|| {
+            describe_counter!(
+                METRIC_CACHE_SAVE_ATTEMPT_TOTAL,
+                "Total scanner data usage cache save attempts by result and cache type."
+            );
+            describe_counter!(
+                METRIC_CACHE_SAVE_TIMEOUT_TOTAL,
+                "Total scanner data usage cache save timeouts by cache type."
+            );
+            describe_counter!(
+                METRIC_CACHE_SAVE_RETRY_TOTAL,
+                "Total scanner data usage cache save retries by cache type."
+            );
+            describe_histogram!(
+                METRIC_CACHE_SAVE_DURATION_SECONDS,
+                "Duration of scanner data usage cache save attempts in seconds."
+            );
+        });
+    }
+
+    fn cache_path_type(path: &str) -> &'static str {
+        if path.ends_with(".bkp") { "backup" } else { "main" }
+    }
+
     pub fn replace(&mut self, path: &str, parent: &str, e: DataUsageEntry) {
         let hash = hash_path(path);
         self.cache.insert(hash.key(), e);
@@ -1094,39 +1131,110 @@ impl DataUsageCache {
         }
     }
 
+    fn cache_save_timeout() -> Duration {
+        Duration::from_secs(
+            rustfs_utils::get_env_u64(ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, DATA_USAGE_CACHE_SAVE_TIMEOUT_SECS_DEFAULT).max(1),
+        )
+    }
+
+    fn backup_cache_save_timeout(timeout_duration: Duration) -> Duration {
+        timeout_duration.min(Duration::from_secs(DATA_USAGE_CACHE_BACKUP_SAVE_TIMEOUT_SECS_MAX))
+    }
+
+    fn record_save_attempt(path_type: &'static str, result: &'static str, duration: Duration) {
+        histogram!(METRIC_CACHE_SAVE_DURATION_SECONDS, "cache" => path_type).record(duration.as_secs_f64());
+        counter!(
+            METRIC_CACHE_SAVE_ATTEMPT_TOTAL,
+            "cache" => path_type,
+            "result" => result
+        )
+        .increment(1);
+        if result == "timeout" {
+            counter!(METRIC_CACHE_SAVE_TIMEOUT_TOTAL, "cache" => path_type).increment(1);
+        }
+    }
+
+    async fn retry_save_op<F, Fut>(
+        path_type: &'static str,
+        timeout_duration: Duration,
+        max_retries: u32,
+        mut save_op: F,
+    ) -> StorageResult<()>
+    where
+        F: FnMut() -> Fut,
+        Fut: Future<Output = StorageResult<()>>,
+    {
+        let mut last_err: Option<StorageError> = None;
+
+        for attempt in 0..=max_retries {
+            let attempt_start = Instant::now();
+            let timeout_res = timeout(timeout_duration, save_op()).await;
+            let duration = attempt_start.elapsed();
+
+            match timeout_res {
+                Ok(Ok(())) => {
+                    Self::record_save_attempt(path_type, "success", duration);
+                    return Ok(());
+                }
+                Err(e) => {
+                    Self::record_save_attempt(path_type, "timeout", duration);
+                    last_err = Some(StorageError::other(format!("{e} after {timeout_duration:?}")));
+                }
+                Ok(Err(e)) => {
+                    Self::record_save_attempt(path_type, "error", duration);
+                    last_err = Some(e);
+                }
+            }
+
+            if last_err.is_some() && attempt < max_retries {
+                counter!(METRIC_CACHE_SAVE_RETRY_TOTAL, "cache" => path_type).increment(1);
+                let backoff_ms = 50_u64 * (1_u64 << attempt) + (rand::random::<u64>() % 100);
+                sleep(Duration::from_millis(backoff_ms)).await;
+            }
+        }
+
+        Err(last_err.unwrap_or_else(|| StorageError::other("Failed to save data usage cache".to_string())))
+    }
+
+    async fn save_path_with_retry<S: StorageAPI>(
+        store: Arc<S>,
+        path: &str,
+        buf: &[u8],
+        timeout_duration: Duration,
+        max_retries: u32,
+    ) -> StorageResult<()> {
+        Self::ensure_cache_save_metrics_registered();
+        let path_type = Self::cache_path_type(path);
+        let path = path.to_string();
+
+        Self::retry_save_op(path_type, timeout_duration, max_retries, move || {
+            let store_clone = store.clone();
+            let path_clone = path.clone();
+            let buf_clone = buf.to_vec();
+            async move {
+                save_config(store_clone, &path_clone, buf_clone).await?;
+                Ok::<(), StorageError>(())
+            }
+        })
+        .await
+    }
+
     pub async fn save<S: StorageAPI>(&self, store: Arc<S>, name: &str) -> StorageResult<()> {
         let mut buf = Vec::new();
         self.serialize(&mut rmp_serde::Serializer::new(&mut buf))?;
+        let timeout_duration = Self::cache_save_timeout();
 
         let path = path_join_buf(&[BUCKET_META_PREFIX, name]);
+        Self::save_path_with_retry(store.clone(), &path, &buf, timeout_duration, DATA_USAGE_CACHE_SAVE_RETRIES).await?;
 
-        let store_clone = store.clone();
-        let buf_clone = buf.clone();
-        let path_clone = path.clone();
-        let res = timeout(Duration::from_secs(5), async move {
-            save_config(store_clone, &path_clone, buf_clone).await?;
-            Ok::<(), StorageError>(())
-        })
-        .await
-        .map_err(|e| StorageError::other(format!("Failed to save data usage cache: {e}")))?;
-
-        if let Err(e) = res {
-            error!("Failed to save data usage cache: {e}");
-            return Err(e);
-        }
-
-        let store_clone = store.clone();
         let backup_name = format!("{name}.bkp");
         let backup_path = path_join_buf(&[BUCKET_META_PREFIX, &backup_name]);
-        let res = timeout(Duration::from_secs(5), async move {
-            save_config(store_clone, &backup_path, buf).await?;
-            Ok::<(), StorageError>(())
-        })
-        .await
-        .map_err(|e| StorageError::other(format!("Failed to save data usage cache: {e}")))?;
-        if let Err(e) = res {
-            error!("Failed to save data usage cache backup: {e}");
-            return Err(e);
+        let backup_timeout_duration = Self::backup_cache_save_timeout(timeout_duration);
+        if let Err(e) =
+            Self::save_path_with_retry(store, &backup_path, &buf, backup_timeout_duration, DATA_USAGE_CACHE_BACKUP_SAVE_RETRIES)
+                .await
+        {
+            warn!("Failed to save data usage cache backup: {e}");
         }
         Ok(())
     }
@@ -1541,6 +1649,9 @@ impl SizeSummary {
 mod tests {
     use super::*;
     use serde_json::Value;
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use temp_env::{with_var, with_var_unset};
 
     #[test]
     fn test_data_usage_info_creation() {
@@ -1619,4 +1730,72 @@ mod tests {
         let decoded: DataUsageEntry = serde_json::from_value(value).expect("Failed to deserialize entry");
         assert_eq!(decoded.failed_objects, 0);
     }
+
+    #[test]
+    fn test_cache_path_type_distinguishes_main_and_backup() {
+        assert_eq!(DataUsageCache::cache_path_type("buckets/.usage-cache.bin"), "main");
+        assert_eq!(DataUsageCache::cache_path_type("buckets/.usage-cache.bin.bkp"), "backup");
+    }
+
+    #[test]
+    fn test_cache_save_timeout_uses_default_when_env_missing() {
+        with_var_unset(ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, || {
+            assert_eq!(
+                DataUsageCache::cache_save_timeout(),
+                Duration::from_secs(DATA_USAGE_CACHE_SAVE_TIMEOUT_SECS_DEFAULT)
+            );
+        });
+    }
+
+    #[test]
+    fn test_cache_save_timeout_respects_env_and_minimum_bound() {
+        with_var(ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, Some("7"), || {
+            assert_eq!(DataUsageCache::cache_save_timeout(), Duration::from_secs(7));
+        });
+
+        with_var(ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, Some("0"), || {
+            assert_eq!(DataUsageCache::cache_save_timeout(), Duration::from_secs(1));
+        });
+    }
+
+    #[tokio::test]
+    async fn test_retry_save_op_retries_on_error_then_succeeds() {
+        let attempts = Arc::new(AtomicUsize::new(0));
+        let attempts_clone = attempts.clone();
+
+        let result =
+            DataUsageCache::retry_save_op("main", Duration::from_millis(200), DATA_USAGE_CACHE_SAVE_RETRIES, move || {
+                let attempts = attempts_clone.clone();
+                async move {
+                    let current = attempts.fetch_add(1, Ordering::SeqCst);
+                    if current < 2 {
+                        return Err(StorageError::other("transient".to_string()));
+                    }
+                    Ok(())
+                }
+            })
+            .await;
+
+        assert!(result.is_ok());
+        assert_eq!(attempts.load(Ordering::SeqCst), 3);
+    }
+
+    #[tokio::test]
+    async fn test_retry_save_op_times_out_and_returns_error_after_retries() {
+        let attempts = Arc::new(AtomicUsize::new(0));
+        let attempts_clone = attempts.clone();
+
+        let result = DataUsageCache::retry_save_op("main", Duration::from_millis(10), DATA_USAGE_CACHE_SAVE_RETRIES, move || {
+            let attempts = attempts_clone.clone();
+            async move {
+                attempts.fetch_add(1, Ordering::SeqCst);
+                tokio::time::sleep(Duration::from_millis(50)).await;
+                Ok(())
+            }
+        })
+        .await;
+
+        assert!(result.is_err());
+        assert_eq!(attempts.load(Ordering::SeqCst), (DATA_USAGE_CACHE_SAVE_RETRIES + 1) as usize);
+    }
 }

crates/scanner/src/scanner_io.rs

@@ -69,6 +69,24 @@ fn finalize_nsscanner_result(results: &[DataUsageCache], first_err: Option<Error
     Ok(())
 }
 
+async fn persist_and_publish_cache_snapshot<S: StorageAPI>(
+    store: Arc<S>,
+    updates: &mpsc::Sender<DataUsageCache>,
+    cache_snapshot: DataUsageCache,
+) -> Option<SystemTime> {
+    let last_update = cache_snapshot.info.last_update;
+
+    if let Err(e) = cache_snapshot.save(store, DATA_USAGE_CACHE_NAME).await {
+        error!("Failed to save data usage cache: {}", e);
+    }
+
+    if let Err(e) = updates.send(cache_snapshot).await {
+        error!("Failed to send data usage cache: {}", e);
+    }
+
+    last_update
+}
+
 #[async_trait::async_trait]
 pub trait ScannerIO: Send + Sync + Debug + 'static {
     async fn nsscanner(
@@ -339,22 +357,20 @@ impl ScannerIOCache for SetDisks {
                         break;
                     }
                     _ = ticker.tick() => {
+                        let cache_snapshot = {
+                            let cache = cache_mutex_clone.lock().await;
+                            if cache.info.last_update == last_update {
+                                None
+                            } else {
+                                Some(cache.clone())
+                            }
+                        };
 
-                       let cache = cache_mutex_clone.lock().await;
-                       if cache.info.last_update == last_update {
-                        continue;
-                       }
-
-                       if let Err(e) = cache.save(store_clone.clone(), DATA_USAGE_CACHE_NAME).await {
-                           error!("Failed to save data usage cache: {}", e);
-                       }
-
-                       if let Err(e) = updates.send(cache.clone()).await {
-                           error!("Failed to send data usage cache: {}", e);
-
-                       }
-
-                       last_update = cache.info.last_update;
+                        let Some(cache_snapshot) = cache_snapshot else {
+                            continue;
+                        };
+                        last_update =
+                            persist_and_publish_cache_snapshot(store_clone.clone(), &updates, cache_snapshot).await;
                     }
                     res =  bucket_result_rx.recv() => {
                         if let Some(result) = res {
@@ -363,18 +379,13 @@ impl ScannerIOCache for SetDisks {
                             cache.info.last_update = Some(SystemTime::now());
 
                         } else {
-                            let mut cache = cache_mutex_clone.lock().await;
-                            cache.info.next_cycle =want_cycle;
-                            cache.info.last_update = Some(SystemTime::now());
-
-                            if let Err(e) = cache.save(store_clone.clone(), DATA_USAGE_CACHE_NAME).await {
-                                error!("Failed to save data usage cache: {}", e);
-                            }
-
-                            if let Err(e) = updates.send(cache.clone()).await {
-                                error!("Failed to send data usage cache: {}", e);
-
-                            }
+                            let cache_snapshot = {
+                                let mut cache = cache_mutex_clone.lock().await;
+                                cache.info.next_cycle = want_cycle;
+                                cache.info.last_update = Some(SystemTime::now());
+                                cache.clone()
+                            };
+                            let _ = persist_and_publish_cache_snapshot(store_clone.clone(), &updates, cache_snapshot).await;
 
                             return;
                         }

crates/scanner/tests/lifecycle_integration_test.rs

@@ -1219,7 +1219,7 @@ mod serial_tests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     #[serial]
     async fn test_scanner_expires_zero_day_current_version() {
-        let (disk_paths, ecstore) = setup_isolated_test_env(true).await;
+        let (disk_paths, ecstore) = setup_isolated_test_env(false).await;
 
         let bucket_name = format!("test-zero-day-expire-{}", &Uuid::new_v4().simple().to_string()[..8]);
         let object_name = "test/object.txt";
@@ -1232,6 +1232,7 @@ mod serial_tests {
 
         assert!(object_exists(&ecstore, bucket_name.as_str(), object_name).await);
 
+        rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::init_background_expiry(ecstore.clone()).await;
         scan_object_with_lifecycle(&disk_paths[0], bucket_name.as_str(), object_name).await;
 
         assert!(

helm/README.md

@@ -36,6 +36,7 @@ RustFS helm chart supports **standalone and distributed mode**. For standalone m
 | config.rustfs.scanner.speed | string | `""` | Scanner speed preset: `fastest`, `fast`, `default`, `slow`, `slowest`. |
 | config.rustfs.scanner.start_delay_secs | string | `""` | Override scanner cycle interval in seconds with `RUSTFS_SCANNER_START_DELAY_SECS`. |
 | config.rustfs.scanner.idle_mode | string | `""` | Override scanner idle throttling flag (`RUSTFS_SCANNER_IDLE_MODE`). |
+| config.rustfs.scanner.cache_save_timeout_secs | string | `""` | Override scanner cache save timeout in seconds with `RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS` (minimum `1`). |
 | config.rustfs.obs_endpoint.enabled | bool | `false` | Whether to send metrics/logs/traces/profilings to remote endpoint, eg, OLTP. |
 | config.rustfs.obs_endpoint.base_endpoint | string | `""` | Root OTLP/HTTP endpoint, e.g. http://otel-collector:4318. |
 | config.rustfs.obs_endpoint.use_stdout | bool | `false` | Whether to output logs to stdout in addition the OLTP. |

helm/rustfs/templates/configmap.yaml

@@ -81,6 +81,9 @@ data:
     {{- if .idle_mode }}
   RUSTFS_SCANNER_IDLE_MODE: {{ .idle_mode | quote }}
     {{- end }}
+    {{- if .cache_save_timeout_secs }}
+  RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS: {{ .cache_save_timeout_secs | quote }}
+    {{- end }}
   {{- end }}
   {{- if .Values.config.rustfs.kms.enabled }}
     {{- if eq .Values.config.rustfs.kms.type "vault" }}

helm/rustfs/values.yaml

@@ -81,6 +81,8 @@ config:
       start_delay_secs: ""
       # Enable/disable scanner sleeps for throttling
       idle_mode: ""
+      # Timeout for scanner cache saves in seconds (minimum 1 second)
+      cache_save_timeout_secs: ""
     obs_endpoint:
       enabled: false # If true, rustfs will export metrics, traces, logs and profiling data to the specified OTLP endpoints. If false, the individual settings for metrics, traces, logs and profiling endpoints will be ignored and all data will not be exported.
       base_endpoint: "" #Root OTLP/HTTP endpoint, e.g. http://otel-collector:4318