diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 9d8e0d01a..827d62005 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -462,19 +462,16 @@ pub(crate) fn local_decommission_queue_prefix(endpoints: &EndpointServerPools, i Ok(local) } -fn first_resumable_decommission_queue_indices(meta: &PoolMeta) -> Vec { +fn resumable_decommission_queue_indices(meta: &PoolMeta) -> Vec { let mut indices = Vec::new(); for (idx, pool) in meta.pools.iter().enumerate() { if let Some(decommission) = &pool.decommission { if !decommission.has_decommission_state() { continue; } - if decommission.complete { + if decommission.complete || decommission.failed || decommission.canceled { continue; } - if decommission.failed || decommission.canceled { - break; - } indices.push(idx); } } @@ -2406,24 +2403,10 @@ impl PoolMeta { } pub fn return_resumable_pools(&self) -> Vec { - let mut new_pools = Vec::new(); - for pool in &self.pools { - if let Some(decommission) = &pool.decommission { - if !decommission.has_decommission_state() { - continue; - } - if decommission.complete || decommission.failed || decommission.canceled { - // Recovery is not required when: - // - Decommissioning completed - // - Decommissioning failed and must be explicitly restarted or cleared - // - Decommissioning was cancelled - continue; - } - // All other scenarios require recovery - new_pools.push(pool.clone()); - } - } - new_pools + resumable_decommission_queue_indices(self) + .into_iter() + .map(|idx| self.pools[idx].clone()) + .collect() } } @@ -3332,7 +3315,7 @@ impl ECStore { let _start_guard = self.start_gate.lock().await; let indices = { let pool_meta = self.pool_meta.read().await; - first_resumable_decommission_queue_indices(&pool_meta) + resumable_decommission_queue_indices(&pool_meta) .into_iter() .filter(|idx| indices.contains(idx)) .collect::>() @@ -3377,7 +3360,7 @@ impl ECStore { pub async fn spawn_missing_local_decommission_routines(self: &Arc) -> Result<()> { let indices = { let pool_meta = self.pool_meta.read().await; - first_resumable_decommission_queue_indices(&pool_meta) + resumable_decommission_queue_indices(&pool_meta) }; let indices = local_decommission_queue_prefix(&self.endpoints(), &indices)?; if indices.is_empty() { @@ -6363,12 +6346,12 @@ mod pools_tests { ensure_decommission_start_keeps_active_pool, ensure_decommission_start_local_leader, ensure_decommission_start_pool_states, ensure_decommission_start_rebalance_meta_allowed, ensure_decommission_start_target_capacity, ensure_decommission_terminal_operation_supported, - ensure_local_decommission_pool_leaders, ensure_valid_decommission_pool_index, first_resumable_decommission_queue_indices, - get_by_index, guard_decommission_cancelers, has_active_decommission_canceler, is_decommission_active, - is_decommission_cancel_requested, load_decommission_entry_versions, local_decommission_queue_prefix, - mark_decommission_bucket_done, merge_pool_status_refresh, missing_decommission_worker_prefix, - observe_decommission_terminal_reload_result, pool_meta_has_active_decommission, require_decommission_store, - reserve_decommission_start_cancelers, resolve_decommission_bucket_done_save_result, resolve_decommission_bucket_state, + ensure_local_decommission_pool_leaders, ensure_valid_decommission_pool_index, get_by_index, guard_decommission_cancelers, + has_active_decommission_canceler, is_decommission_active, is_decommission_cancel_requested, + load_decommission_entry_versions, local_decommission_queue_prefix, mark_decommission_bucket_done, + merge_pool_status_refresh, missing_decommission_worker_prefix, observe_decommission_terminal_reload_result, + pool_meta_has_active_decommission, require_decommission_store, reserve_decommission_start_cancelers, + 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, @@ -6376,9 +6359,9 @@ mod pools_tests { resolve_decommission_preflight_heal_result, resolve_decommission_progress_save_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, run_decommission_listing_with_retry, - run_decommission_listing_with_retry_and_drain, run_decommission_side_effect, should_cleanup_decommission_source_entry, - should_continue_decommission_queue, should_count_decommission_version_complete, + resumable_decommission_queue_indices, rollback_start_decommission_pool_meta, run_decommission_buckets_bounded, + run_decommission_listing_with_retry, run_decommission_listing_with_retry_and_drain, run_decommission_side_effect, + 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_retry_decommission_listing, should_skip_canceled_decommission_routine, spawn_decommission_index_cancelers, split_decommission_buckets, take_and_cancel_decommission_canceler, @@ -9387,7 +9370,7 @@ mod pools_tests { } #[test] - fn test_first_resumable_decommission_queue_indices_stops_at_failed_or_canceled_state() { + fn test_resumable_decommission_queue_indices_skip_terminal_predecessors() { let meta = PoolMeta { pools: vec![ decommission_test_pool_status( @@ -9423,11 +9406,11 @@ mod pools_tests { ..Default::default() }; - assert!(first_resumable_decommission_queue_indices(&meta).is_empty()); + assert_eq!(resumable_decommission_queue_indices(&meta), vec![3]); } #[test] - fn test_first_resumable_decommission_queue_indices_allows_after_completed_prefix() { + fn test_resumable_decommission_queue_indices_preserve_active_predecessor_order() { let meta = PoolMeta { pools: vec![ decommission_test_pool_status( @@ -9440,7 +9423,7 @@ mod pools_tests { decommission_test_pool_status( 1, Some(PoolDecommissionInfo { - queued: true, + start_time: Some(OffsetDateTime::UNIX_EPOCH), ..Default::default() }), ), @@ -9455,11 +9438,11 @@ mod pools_tests { ..Default::default() }; - assert_eq!(first_resumable_decommission_queue_indices(&meta), vec![1, 2]); + assert_eq!(resumable_decommission_queue_indices(&meta), vec![1, 2]); } #[test] - fn test_return_resumable_pools_skips_failed_decommission() { + fn test_runtime_and_startup_use_the_same_resumable_queue() { let meta = PoolMeta { pools: vec![ decommission_test_pool_status( @@ -9471,6 +9454,20 @@ mod pools_tests { ), decommission_test_pool_status( 1, + Some(PoolDecommissionInfo { + canceled: true, + ..Default::default() + }), + ), + decommission_test_pool_status( + 2, + Some(PoolDecommissionInfo { + start_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }), + ), + decommission_test_pool_status( + 3, Some(PoolDecommissionInfo { queued: true, ..Default::default() @@ -9480,10 +9477,15 @@ mod pools_tests { ..Default::default() }; - let resumable = meta.return_resumable_pools(); + let runtime_indices = resumable_decommission_queue_indices(&meta); + let startup_ids = meta + .return_resumable_pools() + .into_iter() + .map(|pool| pool.id) + .collect::>(); - assert_eq!(resumable.len(), 1); - assert_eq!(resumable[0].id, 1); + assert_eq!(runtime_indices, vec![2, 3]); + assert_eq!(startup_ids, vec![2, 3]); } #[test]