fix: preserve scanner usage backups (#4179)

crates/scanner/src/scanner.rs

@@ -1088,29 +1088,14 @@ pub async fn store_data_usage_in_backend(
             }
         };
 
-        // Save a backup every 10th update
+        let backup_data = (attempts > 10).then(|| data.clone());
-        if attempts > 10 {
-            let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
-            let done_save = Metrics::time(Metric::SaveUsage);
-            if let Err(e) = save_config(storeapi.clone(), &backup_path, data.clone()).await {
-                warn!(
-                    target: "rustfs::scanner",
-                    event = EVENT_SCANNER_PERSIST_STATE,
-                    component = LOG_COMPONENT_SCANNER,
-                    subsystem = LOG_SUBSYSTEM_RUNTIME,
-                    path = %backup_path,
-                    state = "backup_save_failed",
-                    error = %e,
-                    "Scanner data usage backup save failed"
-                );
-            }
-            done_save();
-            attempts = 1;
-        }
 
         // Save main configuration
         let done_save = Metrics::time(Metric::SaveUsage);
-        if let Err(e) = save_config(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), data).await {
+        let save_result = save_config(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), data).await;
+        done_save();
+
+        if let Err(e) = save_result {
             error!(
                 target: "rustfs::scanner",
                 event = EVENT_SCANNER_PERSIST_STATE,
@@ -1125,8 +1110,27 @@ pub async fn store_data_usage_in_backend(
         } else {
             replace_bucket_usage_memory_from_info(&data_usage_info).await;
             global_metrics().record_scanner_usage_save_result(ScannerUsageSaveResult::Success);
+
+            // Save a backup only after the primary usage object is durable.
+            if let Some(data) = backup_data {
+                let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
+                let done_save = Metrics::time(Metric::SaveUsage);
+                if let Err(e) = save_config(storeapi.clone(), &backup_path, data).await {
+                    warn!(
+                        target: "rustfs::scanner",
+                        event = EVENT_SCANNER_PERSIST_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_RUNTIME,
+                        path = %backup_path,
+                        state = "backup_save_failed",
+                        error = %e,
+                        "Scanner data usage backup save failed"
+                    );
+                }
+                done_save();
+                attempts = 1;
+            }
         }
-        done_save();
 
         attempts += 1;
     }
@@ -1182,6 +1186,8 @@ mod tests {
     #[derive(Debug, Default)]
     struct MemoryConfigStore {
         objects: Mutex<HashMap<String, Vec<u8>>>,
+        fail_put_number: Mutex<HashMap<String, usize>>,
+        put_counts: Mutex<HashMap<String, usize>>,
     }
 
     fn memory_config_key(bucket: &str, object: &str) -> String {
@@ -1228,7 +1234,19 @@ mod tests {
         ) -> EcstoreResult<ObjectInfo> {
             let mut buf = Vec::new();
             data.stream.read_to_end(&mut buf).await?;
-            self.objects.lock().await.insert(memory_config_key(bucket, object), buf);
+            let key = memory_config_key(bucket, object);
+            let put_count = {
+                let mut put_counts = self.put_counts.lock().await;
+                let put_count = put_counts.entry(key.clone()).or_insert(0);
+                *put_count += 1;
+                *put_count
+            };
+
+            if self.fail_put_number.lock().await.get(&key) == Some(&put_count) {
+                return Err(EcstoreError::other("injected put failure"));
+            }
+
+            self.objects.lock().await.insert(key, buf);
             Ok(ObjectInfo::default())
         }
     }
@@ -1526,6 +1544,49 @@ mod tests {
         assert_eq!(saved.last_update, Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)));
     }
 
+    #[tokio::test]
+    async fn test_store_data_usage_in_backend_keeps_backup_when_primary_save_fails() {
+        let store = Arc::new(MemoryConfigStore::default());
+        let (sender, receiver) = mpsc::channel(11);
+        let ctx = CancellationToken::new();
+
+        let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
+        let main_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str());
+        let backup_key = memory_config_key(RUSTFS_META_BUCKET, &backup_path);
+        let old_backup = b"old-backup".to_vec();
+
+        store.objects.lock().await.insert(backup_key.clone(), old_backup.clone());
+        store.fail_put_number.lock().await.insert(main_key.clone(), 11);
+
+        for idx in 1_u64..=11 {
+            sender
+                .send(DataUsageInfo {
+                    last_update: Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(idx)),
+                    buckets_count: idx,
+                    ..Default::default()
+                })
+                .await
+                .expect("usage snapshot should enqueue");
+        }
+        drop(sender);
+
+        store_data_usage_in_backend(ctx, store.clone(), receiver).await;
+
+        let objects = store.objects.lock().await;
+        assert_eq!(
+            objects.get(&backup_key),
+            Some(&old_backup),
+            "primary save failure must not overwrite the previous backup"
+        );
+
+        let saved = objects
+            .get(&main_key)
+            .expect("last successful primary usage snapshot should remain saved");
+        let saved = serde_json::from_slice::<DataUsageInfo>(saved).expect("saved usage snapshot should decode");
+        assert_eq!(saved.buckets_count, 10);
+        assert_eq!(saved.last_update, Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(10)));
+    }
+
     #[test]
     #[serial]
     fn test_cycle_interval_prefers_explicit_cycle_override() {