diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index d5d3af617..9edae753e 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1088,29 +1088,14 @@ pub async fn store_data_usage_in_backend( } }; - // Save a backup every 10th update - 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; - } + let backup_data = (attempts > 10).then(|| data.clone()); // 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>>, + fail_put_number: Mutex>, + put_counts: Mutex>, } fn memory_config_key(bucket: &str, object: &str) -> String { @@ -1228,7 +1234,19 @@ mod tests { ) -> EcstoreResult { 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::(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() {