From bf3a3a618975221f7c422da3b0598338f44887df Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 14 Jun 2026 19:50:26 +0800 Subject: [PATCH] fix(rebalance): ignore source cleanup failures (#3440) --- crates/ecstore/src/rebalance.rs | 327 ++++++++++++++++++++++--- rustfs/src/admin/handlers/rebalance.rs | 21 +- rustfs/src/app/context/compat.rs | 4 + 3 files changed, 311 insertions(+), 41 deletions(-) diff --git a/crates/ecstore/src/rebalance.rs b/crates/ecstore/src/rebalance.rs index dc8c9dedd..1f312083e 100644 --- a/crates/ecstore/src/rebalance.rs +++ b/crates/ecstore/src/rebalance.rs @@ -82,6 +82,8 @@ pub struct RebalanceStats { pub participating: bool, // Whether the pool is participating in rebalance #[serde(rename = "inf")] pub info: RebalanceInfo, // Rebalance operation info + #[serde(rename = "cw", default)] + pub cleanup_warnings: RebalanceCleanupWarnings, } impl RebalanceStats { @@ -514,6 +516,20 @@ pub struct RebalanceInfo { pub status: RebalStatus, // Current state of rebalance operation } +#[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct RebalanceCleanupWarnings { + #[serde(rename = "count", default)] + pub count: u64, + #[serde(rename = "lastMsg", default)] + pub last_message: Option, + #[serde(rename = "lastBucket", default)] + pub last_bucket: Option, + #[serde(rename = "lastObject", default)] + pub last_object: Option, + #[serde(rename = "lastAt", default)] + pub last_at: Option, +} + #[allow(dead_code)] #[derive(Debug, Clone, Default)] pub struct DiskStat { @@ -924,6 +940,24 @@ impl ECStore { mark_rebalance_bucket_done(rebalance_meta.as_mut(), pool_index, &bucket) } + async fn record_rebalance_cleanup_warning( + &self, + pool_index: usize, + bucket: &str, + object: &str, + message: String, + ) -> Result<()> { + let mut rebalance_meta = self.rebalance_meta.write().await; + record_rebalance_cleanup_warning_in_meta( + rebalance_meta.as_mut(), + pool_index, + bucket, + object, + message, + OffsetDateTime::now_utc(), + ) + } + async fn defer_rebalance_bucket(&self, pool_index: usize, bucket: String, last_error: String) -> Result<()> { let mut rebalance_meta = self.rebalance_meta.write().await; let Some(meta) = rebalance_meta.as_mut() else { @@ -1678,6 +1712,32 @@ fn mark_rebalance_bucket_done(meta: Option<&mut RebalanceMeta>, pool_index: usiz } } +fn record_rebalance_cleanup_warning_in_meta( + meta: Option<&mut RebalanceMeta>, + pool_index: usize, + bucket: &str, + object: &str, + message: String, + now: OffsetDateTime, +) -> Result<()> { + let Some(meta) = meta else { + return Err(rebalance_metadata_not_initialized_error("record rebalance cleanup warning")); + }; + + ensure_valid_rebalance_pool_index(meta.pool_stats.len(), pool_index)?; + let Some(pool_stat) = meta.pool_stats.get_mut(pool_index) else { + return Err(invalid_rebalance_pool_index_error(pool_index, meta.pool_stats.len())); + }; + + pool_stat.cleanup_warnings.count = pool_stat.cleanup_warnings.count.saturating_add(1); + pool_stat.cleanup_warnings.last_message = Some(message); + pool_stat.cleanup_warnings.last_bucket = Some(bucket.to_string()); + pool_stat.cleanup_warnings.last_object = Some(object.to_string()); + pool_stat.cleanup_warnings.last_at = Some(now); + meta.last_refreshed_at = Some(now); + Ok(()) +} + fn take_bucket_from_rebalance_queue(pool_stat: &mut RebalanceStats, bucket: &str) -> bool { let mut found = false; pool_stat.buckets.retain(|name| { @@ -1871,11 +1931,15 @@ where result.map_err(|err| Error::other(format!("rebalance file_info_versions failed for {bucket}/{object_name}: {err}"))) } -fn resolve_rebalance_entry_cleanup_delete_result(result: Result, bucket: &str, object_name: &str) -> Result<()> { +fn resolve_rebalance_entry_cleanup_delete_result( + result: Result, + bucket: &str, + object_name: &str, +) -> Result> { match result { - Ok(_) => Ok(()), - Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => Ok(()), - Err(err) => Err(Error::other(format!("rebalance cleanup delete failed for {bucket}/{object_name}: {err}"))), + Ok(_) => Ok(None), + Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => Ok(None), + Err(err) => Ok(Some(format!("rebalance cleanup delete failed for {bucket}/{object_name}: {err}"))), } } @@ -2278,6 +2342,25 @@ fn remove_rebalanced_buckets_from_queue(pool_stat: &mut RebalanceStats) { pool_stat.buckets.retain(|bucket| !rebalanced_buckets.contains(bucket)); } +fn merge_rebalance_cleanup_warnings(remote: &mut RebalanceCleanupWarnings, local: &RebalanceCleanupWarnings) { + remote.count = remote.count.max(local.count); + + if should_replace_rebalance_cleanup_warning(remote.last_at, local.last_at) { + remote.last_message = local.last_message.clone(); + remote.last_bucket = local.last_bucket.clone(); + remote.last_object = local.last_object.clone(); + remote.last_at = local.last_at; + } +} + +fn should_replace_rebalance_cleanup_warning(remote_at: Option, local_at: Option) -> bool { + match (remote_at, local_at) { + (_, None) => false, + (None, Some(_)) => true, + (Some(remote), Some(local)) => local >= remote, + } +} + fn merge_rebalance_pool_stats(remote: &mut RebalanceStats, local: &RebalanceStats) { remote.init_free_space = remote.init_free_space.max(local.init_free_space); remote.init_capacity = remote.init_capacity.max(local.init_capacity); @@ -2291,6 +2374,7 @@ fn merge_rebalance_pool_stats(remote: &mut RebalanceStats, local: &RebalanceStat remote.num_objects = remote.num_objects.max(local.num_objects); remote.num_versions = remote.num_versions.max(local.num_versions); remote.bytes = remote.bytes.max(local.bytes); + merge_rebalance_cleanup_warnings(&mut remote.cleanup_warnings, &local.cleanup_warnings); if local_is_newer { remote.bucket = local.bucket.clone(); @@ -2620,7 +2704,7 @@ impl ECStore { )?; if should_cleanup_rebalance_source_entry(rebalanced, fivs.versions.len()) { - resolve_rebalance_entry_cleanup_delete_result( + let cleanup_warning = resolve_rebalance_entry_cleanup_delete_result( set.delete_object( bucket.as_str(), &encode_dir_object(&entry.name), @@ -2634,18 +2718,48 @@ impl ECStore { .await, bucket.as_str(), entry.name.as_str(), - ) - .map_err(|err| with_rebalance_entry_context("cleanup_source", bucket.as_str(), entry.name.as_str(), err))?; - debug!( - event = EVENT_REBALANCE_ENTRY, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REBALANCE, - pool_index, - bucket = %bucket, - object = %entry.name, - state = "source_deleted", - "Deleted rebalance source entry" - ); + )?; + if let Some(message) = cleanup_warning { + warn!( + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + object = %entry.name, + stage = "cleanup_source", + cleanup_status = "failed_ignored", + error = %message, + "Ignored rebalance source cleanup failure" + ); + if let Err(err) = self + .record_rebalance_cleanup_warning(pool_index, bucket.as_str(), entry.name.as_str(), message) + .await + { + error!( + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + object = %entry.name, + stage = "cleanup_source", + error = ?err, + "Failed to record rebalance source cleanup warning" + ); + } + } else { + debug!( + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + object = %entry.name, + state = "source_deleted", + "Deleted rebalance source entry" + ); + } } Ok(RebalanceEntryOutcome::Completed) @@ -3049,27 +3163,27 @@ mod rebalance_unit_tests { use super::rebalance_goal_reached; use super::{ DiskError, GetObjectReader, HTTPRangeSpec, MigrationBackend, MigrationVersionResult, ObjectInfo, ObjectOptions, - RebalSaveOpt, RebalStatus, RebalanceBucketOutcome, RebalanceEntryOutcome, RebalanceInfo, RebalanceMeta, RebalanceStats, - RebalanceTerminalEvent, apply_rebalance_save_option, apply_rebalance_terminal_event, apply_stopped_at, - classify_rebalance_terminal_event, clone_arc_by_index, clone_first_arc, clone_rebalance_pool_stats, + RebalSaveOpt, RebalStatus, RebalanceBucketOutcome, RebalanceCleanupWarnings, RebalanceEntryOutcome, RebalanceInfo, + RebalanceMeta, RebalanceStats, RebalanceTerminalEvent, apply_rebalance_save_option, apply_rebalance_terminal_event, + apply_stopped_at, classify_rebalance_terminal_event, clone_arc_by_index, clone_first_arc, clone_rebalance_pool_stats, complete_rebalance_pools_at_goal, complete_rebalance_pools_with_empty_queue, defer_bucket_in_rebalance_queue, ensure_rebalance_listing_disks_available, ensure_rebalance_not_decommissioning, ensure_valid_rebalance_pool_index, has_deferred_rebalance_error, is_rebalance_stopped_terminal_event, is_transient_rebalance_error, load_rebalance_bucket_configs, mark_rebalance_bucket_done, merge_rebalance_meta, migrate_entry_version, migrate_entry_version_with_retry_wait, next_rebal_bucket_from_stat, rebalance_delete_marker_opts, rebalance_listing_retry_delay, rebalance_meta_load_no_data_error, rebalance_meta_load_unknown_format_error, - rebalance_meta_load_unknown_version_error, rebalance_migration_retry_delay, resolve_load_rebalance_stats_update_result, - resolve_next_rebalance_bucket, resolve_rebalance_bucket_error, resolve_rebalance_bucket_result, - resolve_rebalance_entry_cleanup_delete_result, resolve_rebalance_file_info_versions_result, - resolve_rebalance_meta_load_result, resolve_rebalance_meta_save_result, resolve_rebalance_migrate_result_error, - resolve_rebalance_optional_bucket_config_result, resolve_rebalance_participants, resolve_rebalance_save_task_result, - resolve_rebalance_stats_update_result, resolve_rebalance_terminal_error, resolve_rebalance_worker_result, - send_rebalance_done_signal, should_accept_rebalance_stats_update, should_cleanup_rebalance_source_entry, - should_count_rebalance_version_complete, should_defer_rebalance_entry_failure, should_ignore_rebalance_data_usage_cache, - should_pool_participate, should_preserve_rebalance_stopped_state, should_retry_rebalance_listing, - should_skip_rebalance_delete_marker, should_skip_start_rebalance, stop_rebalance_meta_snapshot, stop_rebalance_state, - take_bucket_from_rebalance_queue, validate_start_rebalance_state, wait_rebalance_listing_retry, - with_rebalance_entry_context, + rebalance_meta_load_unknown_version_error, rebalance_migration_retry_delay, record_rebalance_cleanup_warning_in_meta, + resolve_load_rebalance_stats_update_result, resolve_next_rebalance_bucket, resolve_rebalance_bucket_error, + resolve_rebalance_bucket_result, resolve_rebalance_entry_cleanup_delete_result, + resolve_rebalance_file_info_versions_result, resolve_rebalance_meta_load_result, resolve_rebalance_meta_save_result, + resolve_rebalance_migrate_result_error, resolve_rebalance_optional_bucket_config_result, resolve_rebalance_participants, + resolve_rebalance_save_task_result, resolve_rebalance_stats_update_result, resolve_rebalance_terminal_error, + resolve_rebalance_worker_result, send_rebalance_done_signal, should_accept_rebalance_stats_update, + should_cleanup_rebalance_source_entry, should_count_rebalance_version_complete, should_defer_rebalance_entry_failure, + should_ignore_rebalance_data_usage_cache, should_pool_participate, should_preserve_rebalance_stopped_state, + should_retry_rebalance_listing, should_skip_rebalance_delete_marker, should_skip_start_rebalance, + stop_rebalance_meta_snapshot, stop_rebalance_state, take_bucket_from_rebalance_queue, validate_start_rebalance_state, + wait_rebalance_listing_retry, with_rebalance_entry_context, }; use crate::data_movement; use crate::data_usage::DATA_USAGE_CACHE_NAME; @@ -3079,6 +3193,7 @@ mod rebalance_unit_tests { use rustfs_filemeta::TRANSITION_COMPLETE; use rustfs_rio::Index; use s3s::dto::ReplicationConfiguration; + use serde::Serialize; use std::io::Cursor; use std::sync::Arc; use std::sync::Mutex; @@ -3088,6 +3203,44 @@ mod rebalance_unit_tests { use tokio::time::Duration; use tokio_util::sync::CancellationToken; + #[derive(Debug, Default, Serialize)] + struct LegacyRebalanceStats { + #[serde(rename = "ifs")] + init_free_space: u64, + #[serde(rename = "ic")] + init_capacity: u64, + #[serde(rename = "bus")] + buckets: Vec, + #[serde(rename = "rbs")] + rebalanced_buckets: Vec, + #[serde(rename = "bu")] + bucket: String, + #[serde(rename = "ob")] + object: String, + #[serde(rename = "no")] + num_objects: u64, + #[serde(rename = "nv")] + num_versions: u64, + #[serde(rename = "bs")] + bytes: u64, + #[serde(rename = "par")] + participating: bool, + #[serde(rename = "inf")] + info: RebalanceInfo, + } + + #[derive(Debug, Default, Serialize)] + struct LegacyRebalanceMeta { + #[serde(rename = "stopTs")] + stopped_at: Option, + #[serde(rename = "id")] + id: String, + #[serde(rename = "pf")] + percent_free_goal: f64, + #[serde(rename = "rss")] + pool_stats: Vec, + } + struct MigrationBackendSpy { get_object_reader: Mutex>>, delete_object: Mutex>>, @@ -4117,6 +4270,7 @@ mod rebalance_unit_tests { #[test] fn test_merge_rebalance_meta_preserves_updates_from_multiple_pools() { let start_time = OffsetDateTime::from_unix_timestamp(1_000).unwrap(); + let warning_at = OffsetDateTime::from_unix_timestamp(1_500).unwrap(); let mut remote = RebalanceMeta { id: "rebal-1".to_string(), percent_free_goal: 0.5, @@ -4175,6 +4329,13 @@ mod rebalance_unit_tests { bytes: 700, bucket: "bucket-a".to_string(), object: "local-object".to_string(), + cleanup_warnings: RebalanceCleanupWarnings { + count: 1, + last_message: Some("cleanup failed".to_string()), + last_bucket: Some("bucket-a".to_string()), + last_object: Some("local-object".to_string()), + last_at: Some(warning_at), + }, ..Default::default() }, ], @@ -4189,6 +4350,9 @@ mod rebalance_unit_tests { assert_eq!(remote.pool_stats[1].object, "local-object"); assert!(remote.pool_stats[1].buckets.is_empty()); assert_eq!(remote.pool_stats[1].rebalanced_buckets, vec!["bucket-a"]); + assert_eq!(remote.pool_stats[1].cleanup_warnings.count, 1); + assert_eq!(remote.pool_stats[1].cleanup_warnings.last_message.as_deref(), Some("cleanup failed")); + assert_eq!(remote.pool_stats[1].cleanup_warnings.last_at, Some(warning_at)); } #[test] @@ -4212,6 +4376,7 @@ mod rebalance_unit_tests { let local = RebalanceMeta { id: "rebal-1".to_string(), pool_stats: vec![RebalanceStats { + participating: true, info: RebalanceInfo { status: RebalStatus::Started, ..Default::default() @@ -4374,6 +4539,33 @@ mod rebalance_unit_tests { assert!(rendered.contains("unknown version 3"), "{rendered}"); } + #[test] + fn test_rebalance_meta_deserializes_legacy_stats_without_cleanup_warnings() { + let legacy = LegacyRebalanceMeta { + id: "rebal-legacy".to_string(), + percent_free_goal: 0.35, + pool_stats: vec![LegacyRebalanceStats { + buckets: vec!["bucket-a".to_string()], + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + let data = rmp_serde::to_vec(&legacy).expect("legacy rebalance metadata should serialize"); + + let decoded: RebalanceMeta = + rmp_serde::from_slice(data.as_slice()).expect("legacy rebalance metadata should deserialize"); + + assert_eq!(decoded.id, "rebal-legacy"); + assert_eq!(decoded.pool_stats.len(), 1); + assert_eq!(decoded.pool_stats[0].cleanup_warnings, RebalanceCleanupWarnings::default()); + assert_eq!(decoded.pool_stats[0].info.status, RebalStatus::Started); + } + #[test] fn test_resolve_rebalance_stats_update_result_passthrough() { assert!(resolve_rebalance_stats_update_result(Ok(()), 0, "bucket", "object").is_ok()); @@ -4440,7 +4632,7 @@ mod rebalance_unit_tests { #[test] fn test_resolve_rebalance_entry_cleanup_delete_result_passthrough() { let result = resolve_rebalance_entry_cleanup_delete_result(Ok(ObjectInfo::default()), "bucket-a", "obj.txt"); - assert!(result.is_ok()); + assert_eq!(result.expect("successful cleanup should pass through"), None); } #[test] @@ -4450,14 +4642,15 @@ mod rebalance_unit_tests { "bucket-a", "obj.txt", ); - assert!(result.is_ok()); + assert_eq!(result.expect("missing cleanup source should be ignored"), None); } #[test] - fn test_resolve_rebalance_entry_cleanup_delete_result_wraps_error_context() { - let err = resolve_rebalance_entry_cleanup_delete_result(Err(Error::SlowDown), "bucket-a", "obj.txt") - .expect_err("unexpected cleanup errors should be wrapped"); - let message = err.to_string(); + fn test_resolve_rebalance_entry_cleanup_delete_result_returns_warning_for_failures() { + let warning = resolve_rebalance_entry_cleanup_delete_result(Err(Error::SlowDown), "bucket-a", "obj.txt") + .expect("cleanup delete failures should be downgraded to warnings") + .expect("cleanup delete failure should return warning"); + let message = warning.as_str(); assert!(message.contains("rebalance cleanup delete failed for bucket-a/obj.txt")); } @@ -5765,6 +5958,64 @@ mod rebalance_unit_tests { assert!(meta.pool_stats[0].info.last_error.is_none()); } + #[test] + fn test_record_rebalance_cleanup_warning_in_meta_preserves_last_error() { + let now = OffsetDateTime::from_unix_timestamp(10_000).unwrap(); + let mut meta = RebalanceMeta { + pool_stats: vec![RebalanceStats { + info: RebalanceInfo { + last_error: Some("old-error".to_string()), + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + + record_rebalance_cleanup_warning_in_meta(Some(&mut meta), 0, "bucket-a", "obj.txt", "cleanup failed".to_string(), now) + .expect("cleanup warning should be recorded"); + + assert_eq!(meta.pool_stats[0].info.last_error.as_deref(), Some("old-error")); + assert_eq!(meta.pool_stats[0].cleanup_warnings.count, 1); + assert_eq!(meta.pool_stats[0].cleanup_warnings.last_message.as_deref(), Some("cleanup failed")); + assert_eq!(meta.pool_stats[0].cleanup_warnings.last_bucket.as_deref(), Some("bucket-a")); + assert_eq!(meta.pool_stats[0].cleanup_warnings.last_object.as_deref(), Some("obj.txt")); + assert_eq!(meta.pool_stats[0].cleanup_warnings.last_at, Some(now)); + assert_eq!(meta.last_refreshed_at, Some(now)); + } + + #[test] + fn test_complete_rebalance_pools_with_empty_queue_preserves_cleanup_warnings() { + let warning_at = OffsetDateTime::from_unix_timestamp(9_000).unwrap(); + let completed_at = OffsetDateTime::from_unix_timestamp(10_000).unwrap(); + let mut meta = RebalanceMeta { + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + cleanup_warnings: RebalanceCleanupWarnings { + count: 1, + last_message: Some("cleanup failed".to_string()), + last_bucket: Some("bucket-a".to_string()), + last_object: Some("obj.txt".to_string()), + last_at: Some(warning_at), + }, + ..Default::default() + }], + ..Default::default() + }; + + assert!(complete_rebalance_pools_with_empty_queue(&mut meta, completed_at)); + + assert_eq!(meta.pool_stats[0].info.status, RebalStatus::Completed); + assert!(meta.pool_stats[0].info.last_error.is_none()); + assert_eq!(meta.pool_stats[0].cleanup_warnings.count, 1); + assert_eq!(meta.pool_stats[0].cleanup_warnings.last_message.as_deref(), Some("cleanup failed")); + assert_eq!(meta.pool_stats[0].cleanup_warnings.last_at, Some(warning_at)); + } + #[test] fn test_apply_stopped_at_transitions_started_pools_only() { let now = OffsetDateTime::now_utc(); diff --git a/rustfs/src/admin/handlers/rebalance.rs b/rustfs/src/admin/handlers/rebalance.rs index 8f8476d1d..42b4eeb51 100644 --- a/rustfs/src/admin/handlers/rebalance.rs +++ b/rustfs/src/admin/handlers/rebalance.rs @@ -24,11 +24,10 @@ use crate::{ use http::{HeaderMap, HeaderValue, StatusCode}; use hyper::Method; use matchit::Params; -use rustfs_ecstore::rebalance::RebalanceMeta; use rustfs_ecstore::{ error::StorageError, notification_sys::get_global_notification_sys, - rebalance::{DiskStat, RebalSaveOpt}, + rebalance::{DiskStat, RebalSaveOpt, RebalanceCleanupWarnings, RebalanceMeta}, store_api::BucketOperations, }; use rustfs_policy::policy::action::{Action, AdminAction}; @@ -104,6 +103,8 @@ pub struct RebalancePoolStatus { pub used: f64, // Fraction of used space in range 0.0..=1.0 #[serde(rename = "lastError")] pub last_error: Option, // Last rebalance error message for this pool + #[serde(rename = "cleanupWarnings")] + pub cleanup_warnings: RebalanceCleanupWarnings, #[serde(rename = "progress")] pub progress: Option, // None when rebalance is not running } @@ -205,6 +206,7 @@ fn build_rebalance_pool_statuses( status: ps.info.status.to_string(), used: rebalance_pool_used(disk_stats, i), last_error: ps.info.last_error.clone(), + cleanup_warnings: ps.cleanup_warnings.clone(), progress: None, }; @@ -562,7 +564,7 @@ mod rebalance_handler_tests { RebalPoolProgress, RebalanceAdminStatus, RebalancePoolStatus, build_rebalance_pool_statuses, rebalance_pool_used, rebalance_remaining_buckets, rebalance_used_pct, }; - use rustfs_ecstore::rebalance::{DiskStat, RebalStatus, RebalanceInfo, RebalanceStats}; + use rustfs_ecstore::rebalance::{DiskStat, RebalStatus, RebalanceCleanupWarnings, RebalanceInfo, RebalanceStats}; use time::OffsetDateTime; #[test] @@ -668,6 +670,7 @@ mod rebalance_handler_tests { start_time: Some(OffsetDateTime::from_unix_timestamp(1_000).unwrap()), ..Default::default() }, + ..Default::default() }; let progress = build_rebalance_pool_progress(OffsetDateTime::from_unix_timestamp(1_050).unwrap(), None, 0.3, &ps) @@ -772,6 +775,7 @@ mod rebalance_handler_tests { start_time: Some(OffsetDateTime::from_unix_timestamp(1_000).unwrap()), ..Default::default() }, + ..Default::default() }, RebalanceStats { participating: false, @@ -847,6 +851,7 @@ mod rebalance_handler_tests { start_time: Some(OffsetDateTime::from_unix_timestamp(2_000).unwrap()), ..Default::default() }, + ..Default::default() }, ]; @@ -883,6 +888,13 @@ mod rebalance_handler_tests { status: "Started".to_string(), used: 0.5, last_error: Some("temporary error".to_string()), + cleanup_warnings: RebalanceCleanupWarnings { + count: 1, + last_message: Some("cleanup warning".to_string()), + last_bucket: Some("bucket-a".to_string()), + last_object: Some("obj".to_string()), + last_at: Some(OffsetDateTime::from_unix_timestamp(1_001).unwrap()), + }, progress: Some(RebalPoolProgress { num_objects: 3, num_versions: 5, @@ -899,6 +911,8 @@ mod rebalance_handler_tests { let json = serde_json::to_string(&status).unwrap(); assert!(json.contains("\"remainingBuckets\"")); assert!(json.contains("\"lastError\"")); + assert!(json.contains("\"cleanupWarnings\"")); + assert!(json.contains("\"lastMsg\":\"cleanup warning\"")); assert!(json.contains("\"stoppedAt\":null")); } @@ -913,6 +927,7 @@ mod rebalance_handler_tests { status: "Stopped".to_string(), used: 0.3, last_error: None, + cleanup_warnings: RebalanceCleanupWarnings::default(), progress: None, }], }; diff --git a/rustfs/src/app/context/compat.rs b/rustfs/src/app/context/compat.rs index fe35990e8..6883861a5 100644 --- a/rustfs/src/app/context/compat.rs +++ b/rustfs/src/app/context/compat.rs @@ -245,6 +245,10 @@ mod tests { }; let endpoint_pools = EndpointServerPools(vec![pool_endpoints]); + if let Some(store) = new_object_layer_fn() { + return (temp_dir, store, endpoint_pools); + } + init_local_disks(endpoint_pools.clone()).await.expect("test local disks"); let store = ECStore::new( "127.0.0.1:0".parse().expect("test addr"),