From 4cdc6ae63f6f5768aa44ae5d7145350546374664 Mon Sep 17 00:00:00 2001 From: cxymds Date: Thu, 25 Jun 2026 21:08:55 +0800 Subject: [PATCH] fix(admin): refresh persisted pool states (#3861) --- crates/ecstore/src/pools.rs | 216 +++++++++++++++++++++-- crates/ecstore/src/rebalance/control.rs | 219 +++++++++++++++++++++++- rustfs/src/app/admin_usecase.rs | 22 +++ 3 files changed, 436 insertions(+), 21 deletions(-) diff --git a/crates/ecstore/src/pools.rs b/crates/ecstore/src/pools.rs index 779f47397..98c7c2fbe 100644 --- a/crates/ecstore/src/pools.rs +++ b/crates/ecstore/src/pools.rs @@ -752,6 +752,44 @@ fn apply_decommission_status_space_info(mut pool_info: PoolStatus, space_info: P pool_info } +fn should_replace_pool_status_for_status_refresh( + current: Option<&PoolStatus>, + persisted: &PoolStatus, + has_active_worker: bool, +) -> bool { + let Some(current) = current else { + return true; + }; + + !has_active_worker && persisted.last_update > current.last_update +} + +fn merge_pool_status_refresh(current: &mut PoolMeta, persisted: PoolMeta, active_workers: &[bool]) { + if persisted.pools.is_empty() { + return; + } + + if current.pools.is_empty() { + *current = persisted; + return; + } + + for (idx, persisted_pool) in persisted.pools.into_iter().enumerate() { + if persisted_pool.id != idx { + continue; + } + + let has_active_worker = active_workers.get(idx).copied().unwrap_or(false); + if idx < current.pools.len() { + if should_replace_pool_status_for_status_refresh(current.pools.get(idx), &persisted_pool, has_active_worker) { + current.pools[idx] = persisted_pool; + } + } else if idx == current.pools.len() && !has_active_worker { + current.pools.push(persisted_pool); + } + } +} + fn resolve_start_decommission_pool_meta_reload_result(result: Result<()>) -> Result<()> { resolve_decommission_pool_meta_reload_result(result, "start_decommission") } @@ -2217,6 +2255,26 @@ impl ECStore { Ok(apply_decommission_status_space_info(pool_info, space_info)) } + #[tracing::instrument(skip_all)] + pub async fn refresh_pool_status_meta(&self) -> Result<()> { + let pool = self + .pools + .first() + .cloned() + .ok_or_else(|| Error::other("refresh_pool_status_meta: no pools available"))?; + let mut persisted = PoolMeta::default(); + persisted.load(pool, self.pools.clone()).await?; + + let active_workers = { + let cancelers = self.decommission_cancelers.read().await; + cancelers.iter().map(Option::is_some).collect::>() + }; + + let mut pool_meta = self.pool_meta.write().await; + merge_pool_status_refresh(&mut pool_meta, persisted, &active_workers); + Ok(()) + } + async fn get_decommission_pool_space_info(&self, idx: usize) -> Result { if let Some(sets) = self.pools.get(idx) { let mut info = sets.storage_info_snapshot().await; @@ -4808,22 +4866,23 @@ mod pools_tests { ensure_local_decommission_pool_leaders, ensure_valid_decommission_pool_index, first_resumable_decommission_queue_indices, get_by_index, has_active_decommission_canceler, is_decommission_active, is_decommission_cancel_requested, load_decommission_entry_versions, local_decommission_queue_prefix, mark_decommission_bucket_done, - missing_decommission_worker_prefix, observe_decommission_terminal_reload_result, pool_meta_has_active_decommission, - require_decommission_store, resolve_decommission_bucket_done_save_result, resolve_decommission_bucket_state, - resolve_decommission_check_after_list_result, resolve_decommission_entry_cleanup_delete_result, - resolve_decommission_entry_exact_versions, resolve_decommission_entry_reload_result, - resolve_decommission_listing_worker_result, resolve_decommission_optional_bucket_config_result, - resolve_decommission_pool_meta_reload_result, resolve_decommission_preflight_heal_result, - resolve_decommission_progress_save_result, resolve_decommission_spawn_failure_result, - resolve_decommission_terminal_mark_after_error_result, resolve_decommission_terminal_mark_result, - resolve_decommission_update_after_result, resolve_start_decommission_pool_meta_reload_result, - rollback_start_decommission_pool_meta, run_decommission_buckets_bounded, should_cleanup_decommission_source_entry, - should_continue_decommission_queue, should_count_decommission_version_complete, - should_preserve_decommission_canceled_state, should_reject_decommission_cancel_as_terminal, - should_retry_decommission_cancel_reload, should_skip_canceled_decommission_routine, split_decommission_buckets, - take_and_cancel_decommission_canceler, take_decommission_canceler, touch_decommission_progress, - track_decommission_current_object, track_decommission_current_object_stage, validate_start_decommission_request, - wait_decommission_worker_drain, with_decommission_entry_context, + merge_pool_status_refresh, missing_decommission_worker_prefix, observe_decommission_terminal_reload_result, + pool_meta_has_active_decommission, require_decommission_store, resolve_decommission_bucket_done_save_result, + resolve_decommission_bucket_state, resolve_decommission_check_after_list_result, + resolve_decommission_entry_cleanup_delete_result, resolve_decommission_entry_exact_versions, + resolve_decommission_entry_reload_result, resolve_decommission_listing_worker_result, + resolve_decommission_optional_bucket_config_result, resolve_decommission_pool_meta_reload_result, + resolve_decommission_preflight_heal_result, resolve_decommission_progress_save_result, + resolve_decommission_spawn_failure_result, resolve_decommission_terminal_mark_after_error_result, + resolve_decommission_terminal_mark_result, resolve_decommission_update_after_result, + resolve_start_decommission_pool_meta_reload_result, rollback_start_decommission_pool_meta, + run_decommission_buckets_bounded, should_cleanup_decommission_source_entry, should_continue_decommission_queue, + should_count_decommission_version_complete, should_preserve_decommission_canceled_state, + should_reject_decommission_cancel_as_terminal, should_retry_decommission_cancel_reload, + should_skip_canceled_decommission_routine, split_decommission_buckets, take_and_cancel_decommission_canceler, + take_decommission_canceler, touch_decommission_progress, track_decommission_current_object, + track_decommission_current_object_stage, validate_start_decommission_request, wait_decommission_worker_drain, + with_decommission_entry_context, }; use crate::data_movement; use crate::disk::endpoint::Endpoint; @@ -4910,6 +4969,131 @@ mod pools_tests { assert_eq!(decommission.current_size, 25); } + #[test] + fn test_merge_pool_status_refresh_uses_persisted_terminal_decommission() { + let older = OffsetDateTime::from_unix_timestamp(1_000).expect("test timestamp should be valid"); + let newer = OffsetDateTime::from_unix_timestamp(2_000).expect("test timestamp should be valid"); + let mut current = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: older, + decommission: Some(PoolDecommissionInfo { + start_time: Some(older), + ..Default::default() + }), + }], + ..Default::default() + }; + let persisted = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: newer, + decommission: Some(PoolDecommissionInfo { + complete: true, + ..Default::default() + }), + }], + ..Default::default() + }; + + merge_pool_status_refresh(&mut current, persisted, &[false]); + + let info = current.pools[0] + .decommission + .as_ref() + .expect("decommission info should be present"); + assert!(info.complete); + assert!(!info.failed); + assert!(!info.canceled); + } + + #[test] + fn test_merge_pool_status_refresh_keeps_newer_local_active_progress() { + let older = OffsetDateTime::from_unix_timestamp(1_000).expect("test timestamp should be valid"); + let newer = OffsetDateTime::from_unix_timestamp(2_000).expect("test timestamp should be valid"); + let mut current = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: newer, + decommission: Some(PoolDecommissionInfo { + start_time: Some(older), + items_decommissioned: 10, + bytes_done: 1_024, + ..Default::default() + }), + }], + ..Default::default() + }; + let persisted = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: older, + decommission: Some(PoolDecommissionInfo { + start_time: Some(older), + items_decommissioned: 1, + bytes_done: 128, + ..Default::default() + }), + }], + ..Default::default() + }; + + merge_pool_status_refresh(&mut current, persisted, &[true]); + + let info = current.pools[0] + .decommission + .as_ref() + .expect("local decommission info should remain present"); + assert_eq!(info.items_decommissioned, 10); + assert_eq!(info.bytes_done, 1_024); + } + + #[test] + fn test_merge_pool_status_refresh_keeps_newer_local_active_over_older_terminal() { + let older = OffsetDateTime::from_unix_timestamp(1_000).expect("test timestamp should be valid"); + let newer = OffsetDateTime::from_unix_timestamp(2_000).expect("test timestamp should be valid"); + let mut current = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: newer, + decommission: Some(PoolDecommissionInfo { + start_time: Some(older), + items_decommissioned: 10, + bytes_done: 1_024, + ..Default::default() + }), + }], + ..Default::default() + }; + let persisted = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: older, + decommission: Some(PoolDecommissionInfo { + failed: true, + ..Default::default() + }), + }], + ..Default::default() + }; + + merge_pool_status_refresh(&mut current, persisted, &[true]); + + let info = current.pools[0] + .decommission + .as_ref() + .expect("local active decommission info should remain present"); + assert!(!info.failed); + assert_eq!(info.items_decommissioned, 10); + assert_eq!(info.bytes_done, 1_024); + } + #[test] fn test_dedup_indices_removes_duplicates_preserving_order() { assert_eq!(dedup_indices(&[0, 2, 1, 2, 3, 0]), vec![0, 2, 1, 3]); diff --git a/crates/ecstore/src/rebalance/control.rs b/crates/ecstore/src/rebalance/control.rs index 58b9b62bf..abc467b69 100644 --- a/crates/ecstore/src/rebalance/control.rs +++ b/crates/ecstore/src/rebalance/control.rs @@ -1,10 +1,11 @@ use super::meta::{ RebalanceMetaMergeOutcome, clone_first_arc, clone_rebalance_pool_stats, defer_bucket_in_rebalance_queue, - ensure_valid_rebalance_pool_index, invalid_rebalance_pool_index_error, is_rebalance_conflicting_with_decommission, - mark_rebalance_bucket_done, merge_rebalance_meta, percent_free_ratio, rebalance_metadata_not_initialized_error, - record_rebalance_cleanup_warning_in_meta, record_rebalance_stop_propagation_snapshot, resolve_next_rebalance_bucket, - rollback_rebalance_start_meta_snapshot_for_id, should_accept_rebalance_stats_update, should_pool_participate, - stop_rebalance_meta_snapshot_for_id, validate_init_rebalance_state, + ensure_valid_rebalance_pool_index, invalid_rebalance_pool_index_error, is_rebalance_actively_running, + is_rebalance_conflicting_with_decommission, mark_rebalance_bucket_done, merge_rebalance_meta, percent_free_ratio, + rebalance_metadata_not_initialized_error, record_rebalance_cleanup_warning_in_meta, + record_rebalance_stop_propagation_snapshot, resolve_next_rebalance_bucket, rollback_rebalance_start_meta_snapshot_for_id, + should_accept_rebalance_stats_update, should_pool_participate, stop_rebalance_meta_snapshot_for_id, + validate_init_rebalance_state, }; use super::worker::{ rebalance_meta_lock_error, resolve_load_rebalance_stats_update_result, resolve_rebalance_meta_load_result, @@ -46,6 +47,32 @@ fn pool_rebalance_status_from_meta(meta: Option<&RebalanceMeta>, pool_index: usi .unwrap_or_default() } +fn merge_rebalance_status_refresh(current: &mut Option, persisted: RebalanceMeta) { + if persisted.id.is_empty() && persisted.pool_stats.is_empty() { + clear_rebalance_status_refresh(current); + return; + } + + match current.as_mut() { + Some(current_meta) => { + if merge_rebalance_meta(current_meta, &persisted) == RebalanceMetaMergeOutcome::RejectedActiveConflict + && !is_rebalance_actively_running(current_meta) + { + *current = Some(persisted); + } + } + None => { + *current = Some(persisted); + } + } +} + +fn clear_rebalance_status_refresh(current: &mut Option) { + if current.as_ref().is_none_or(|meta| !is_rebalance_actively_running(meta)) { + *current = None; + } +} + impl ECStore { pub(super) async fn save_rebalance_meta_with_merge( &self, @@ -128,6 +155,27 @@ impl ECStore { Ok(()) } + #[tracing::instrument(skip_all)] + pub async fn refresh_rebalance_status_meta(&self) -> Result<()> { + let pool = clone_first_arc(&self.pools, "refresh_rebalance_status_meta: no pools available")?; + let mut persisted = RebalanceMeta::new(); + match persisted.load(pool).await { + Ok(()) => { + let mut rebalance_meta = self.rebalance_meta.write().await; + merge_rebalance_status_refresh(&mut rebalance_meta, persisted); + } + Err(Error::ConfigNotFound) => { + let mut rebalance_meta = self.rebalance_meta.write().await; + clear_rebalance_status_refresh(&mut rebalance_meta); + } + Err(err) => { + return Err(Error::other(format!("rebalance metadata refresh failed during pool status: {err}"))); + } + } + + Ok(()) + } + #[tracing::instrument(skip_all)] pub async fn update_rebalance_stats(&self) -> Result<()> { let mut ok = false; @@ -567,4 +615,165 @@ mod tests { assert_eq!(pool_rebalance_status_from_meta(Some(&meta), 1), (RebalStatus::Started, false)); assert_eq!(pool_rebalance_status_from_meta(Some(&meta), 2), (RebalStatus::None, false)); } + + #[test] + fn rebalance_status_refresh_applies_persisted_terminal_state() { + let rebalance_id = "rebalance-id".to_string(); + let now = OffsetDateTime::from_unix_timestamp(1_000).expect("test timestamp should be valid"); + let mut current = Some(RebalanceMeta { + id: rebalance_id.clone(), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + start_time: Some(now), + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }); + let persisted = RebalanceMeta { + id: rebalance_id, + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Completed, + start_time: Some(now), + end_time: Some(now), + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + + merge_rebalance_status_refresh(&mut current, persisted); + + let refreshed = current.as_ref().expect("refresh should keep rebalance metadata"); + assert_eq!(refreshed.pool_stats[0].info.status, RebalStatus::Completed); + assert_eq!(refreshed.pool_stats[0].info.end_time, Some(now)); + } + + #[test] + fn rebalance_status_refresh_preserves_runtime_cancel_token() { + let rebalance_id = "rebalance-id".to_string(); + let now = OffsetDateTime::from_unix_timestamp(1_000).expect("test timestamp should be valid"); + let mut current = Some(RebalanceMeta { + id: rebalance_id.clone(), + cancel: Some(tokio_util::sync::CancellationToken::new()), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + start_time: Some(now), + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }); + let persisted = RebalanceMeta { + id: rebalance_id, + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + start_time: Some(now), + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + + merge_rebalance_status_refresh(&mut current, persisted); + + assert!( + current.as_ref().and_then(|meta| meta.cancel.as_ref()).is_some(), + "status refresh must not drop the runtime cancellation token" + ); + } + + #[test] + fn rebalance_status_refresh_preserves_local_active_different_id_conflict() { + let now = OffsetDateTime::from_unix_timestamp(1_000).expect("test timestamp should be valid"); + let mut current = Some(RebalanceMeta { + id: "old-active-id".to_string(), + cancel: Some(tokio_util::sync::CancellationToken::new()), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + start_time: Some(now), + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }); + let persisted = RebalanceMeta { + id: "new-terminal-id".to_string(), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Completed, + start_time: Some(now), + end_time: Some(now), + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + + merge_rebalance_status_refresh(&mut current, persisted); + + let refreshed = current.as_ref().expect("local active metadata should remain visible"); + assert_eq!(refreshed.id, "old-active-id"); + assert_eq!(refreshed.pool_stats[0].info.status, RebalStatus::Started); + assert!( + refreshed.cancel.is_some(), + "status refresh must not drop a live runtime cancellation token" + ); + } + + #[test] + fn rebalance_status_refresh_replaces_stale_memory_without_runtime_token() { + let now = OffsetDateTime::from_unix_timestamp(1_000).expect("test timestamp should be valid"); + let mut current = Some(RebalanceMeta { + id: "old-stale-id".to_string(), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + start_time: Some(now), + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }); + let persisted = RebalanceMeta { + id: "new-terminal-id".to_string(), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Completed, + start_time: Some(now), + end_time: Some(now), + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + + merge_rebalance_status_refresh(&mut current, persisted); + + let refreshed = current.as_ref().expect("persisted metadata should replace stale memory"); + assert_eq!(refreshed.id, "new-terminal-id"); + assert_eq!(refreshed.pool_stats[0].info.status, RebalStatus::Completed); + assert!(refreshed.cancel.is_none()); + } } diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index 769b13cbf..be60b32d4 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -205,6 +205,20 @@ impl DefaultAdminUsecase { Self::app_error(code, message) } + async fn refresh_rebalance_status_snapshot(store: &ECStore) -> AdminUsecaseResult<()> { + store.refresh_rebalance_status_meta().await.map_err(|err| { + error!("refresh rebalance metadata for pool status failed: {:?}", err); + ApiError::from(err) + }) + } + + async fn refresh_pool_status_snapshot(store: &ECStore) -> AdminUsecaseResult<()> { + store.refresh_pool_status_meta().await.map_err(|err| { + error!("refresh pool metadata for pool status failed: {:?}", err); + ApiError::from(err) + }) + } + pub async fn execute_query_server_info(&self, req: QueryServerInfoRequest) -> AdminUsecaseResult { let info = get_server_info(req.include_pools).await; Ok(QueryServerInfoResponse { info }) @@ -310,6 +324,8 @@ impl DefaultAdminUsecase { return Err(Self::app_error_default(S3ErrorCode::NotImplemented)); } + Self::refresh_pool_status_snapshot(store.as_ref()).await?; + let mut pool_statuses = Vec::new(); for (idx, _) in endpoints.as_ref().iter().enumerate() { let state = store.status(idx).await.map_err(ApiError::from)?; @@ -324,6 +340,7 @@ impl DefaultAdminUsecase { return Err(Self::app_error(S3ErrorCode::InternalError, "Not init")); }; let pool_statuses = self.execute_list_pool_statuses().await?; + Self::refresh_rebalance_status_snapshot(store.as_ref()).await?; let mut items = Vec::with_capacity(pool_statuses.len()); for status in pool_statuses { let rebalance_status = store.pool_rebalance_status(status.id).await; @@ -362,7 +379,9 @@ impl DefaultAdminUsecase { return Err(Self::app_error(S3ErrorCode::InternalError, "Not init")); }; + Self::refresh_pool_status_snapshot(store.as_ref()).await?; let status = store.status(idx).await.map_err(ApiError::from)?; + Self::refresh_rebalance_status_snapshot(store.as_ref()).await?; let rebalance_status = store.pool_rebalance_status(idx).await; Ok(Self::pool_list_item_from_status(status, rebalance_status)) } @@ -372,6 +391,7 @@ impl DefaultAdminUsecase { return Err(Self::app_error(S3ErrorCode::InternalError, "Not init")); }; let pool_statuses = self.execute_list_pool_statuses().await?; + Self::refresh_rebalance_status_snapshot(store.as_ref()).await?; let mut pools = Vec::with_capacity(pool_statuses.len()); for status in pool_statuses { let rebalance_status = store.pool_rebalance_status(status.id).await; @@ -398,7 +418,9 @@ impl DefaultAdminUsecase { return Err(Self::app_error(S3ErrorCode::InternalError, "Not init")); }; + Self::refresh_pool_status_snapshot(store.as_ref()).await?; let status = store.status(idx).await.map_err(ApiError::from)?; + Self::refresh_rebalance_status_snapshot(store.as_ref()).await?; let rebalance_status = store.pool_rebalance_status(idx).await; Ok(Self::decommission_pool_status_from_status(status, rebalance_status)) }