fix(admin): refresh persisted pool states (#3861)

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::<Vec<_>>()
+        };
+
+        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<PoolSpaceInfo> {
         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]);

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<RebalanceMeta>, 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<RebalanceMeta>) {
+    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<S>(
         &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());
+    }
 }

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<QueryServerInfoResponse> {
         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))
     }