From 33341d5fcf19bf46020590a740926c02f58f121d Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 23 Aug 2026 23:51:33 +0800 Subject: [PATCH] fix(ecstore): unify decommission resume queue (#6403) * fix(ecstore): unify decommission resume queue * fix(ecstore): clear swift clippy warnings * fix(ecstore): serialize decommission recovery --------- Co-authored-by: houseme --- crates/ecstore/src/core/pools.rs | 217 +++++++++++++++++++++++-------- 1 file changed, 160 insertions(+), 57 deletions(-) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 58f0d4241..69918647a 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -502,19 +502,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); } } @@ -607,6 +604,7 @@ fn spawn_decommission_index_cancelers( error = %err, "Decommission routine failed" ); + store.quiesce_decommission_worker_after_join_error(&canceler).await; store.retry_decommission_failed_for_operation(idx, &canceler).await; stop_queue = true; continue; @@ -3191,24 +3189,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() } } @@ -4232,6 +4216,12 @@ impl ECStore { } } + async fn quiesce_decommission_worker_after_join_error(&self, canceler: &DecommissionCanceler) { + canceler.cancel(); + let movement_gate = self.ctx.data_movement_operation_gate(); + let _movement_guard = movement_gate.write().await; + } + async fn reserve_decommission_routines( &self, rx: &CancellationToken, @@ -4245,7 +4235,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::>() @@ -4268,6 +4258,19 @@ impl ECStore { Ok(index_cancelers) } + async fn reserve_missing_local_decommission_routines( + &self, + rx: &CancellationToken, + endpoints: &EndpointServerPools, + ) -> Result> { + let indices = { + let pool_meta = self.pool_meta.read().await; + resumable_decommission_queue_indices(&pool_meta) + }; + let indices = local_decommission_queue_prefix(endpoints, &indices)?; + self.reserve_decommission_routines(rx, indices.as_slice()).await + } + pub(crate) async fn spawn_decommission_routines( &self, store: Arc, @@ -4288,17 +4291,9 @@ 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) - }; - let indices = local_decommission_queue_prefix(&self.endpoints(), &indices)?; - if indices.is_empty() { - return Ok(()); - } - let rx = CancellationToken::new(); - let index_cancelers = self.reserve_decommission_routines(&rx, indices.as_slice()).await?; + let endpoints = self.endpoints(); + let index_cancelers = self.reserve_missing_local_decommission_routines(&rx, &endpoints).await?; if index_cancelers.is_empty() { return Ok(()); } @@ -8941,23 +8936,22 @@ 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_decommission_durable_ilm_receipts, merge_pool_status_refresh, - missing_decommission_worker_prefix, observe_decommission_terminal_reload_result, pool_meta_has_active_decommission, - reconcile_decommission_meta_buckets, 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_decommission_durable_ilm_receipts, + merge_pool_status_refresh, missing_decommission_worker_prefix, observe_decommission_terminal_reload_result, + pool_meta_has_active_decommission, reconcile_decommission_meta_buckets, 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, resolve_decommission_pool_meta_reload_result, 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, should_fail_decommission_pool_after_exhausted_source_changed, + 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, 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_fail_decommission_pool_after_exhausted_source_changed, 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, @@ -9731,6 +9725,33 @@ mod pools_tests { .expect("transition task should not panic"); } + #[tokio::test] + async fn test_join_error_quiesces_side_effects_before_failure_transition() { + let canceler = DecommissionCanceler::new(CancellationToken::new()); + let store = decommission_worker_test_store(PoolMeta::default(), vec![Some(canceler.clone())]); + let operation_gate = store.ctx.data_movement_operation_gate(); + let operation_guard = operation_gate.read().await; + let transition = tokio::spawn({ + let store = store.clone(); + let canceler = canceler.clone(); + async move { store.quiesce_decommission_worker_after_join_error(&canceler).await } + }); + + tokio::time::timeout(StdDuration::from_secs(1), canceler.token().cancelled()) + .await + .expect("join error handling should cancel the detached worker"); + assert!( + !transition.is_finished(), + "failure transition must wait for the detached worker's in-flight side effect" + ); + + drop(operation_guard); + tokio::time::timeout(StdDuration::from_secs(1), transition) + .await + .expect("failure transition should finish after the side effect") + .expect("failure transition task should not panic"); + } + #[tokio::test(start_paused = true)] async fn test_run_decommission_listing_with_retry_drains_before_each_retry() { let attempts = Arc::new(AtomicUsize::new(0)); @@ -12163,7 +12184,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( @@ -12199,11 +12220,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( @@ -12216,7 +12237,7 @@ mod pools_tests { decommission_test_pool_status( 1, Some(PoolDecommissionInfo { - queued: true, + start_time: Some(OffsetDateTime::UNIX_EPOCH), ..Default::default() }), ), @@ -12231,11 +12252,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() { + #[tokio::test] + async fn test_runtime_recovery_reserves_the_startup_resumable_queue() { let meta = PoolMeta { pools: vec![ decommission_test_pool_status( @@ -12247,6 +12268,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() @@ -12255,11 +12290,79 @@ mod pools_tests { ], ..Default::default() }; + let startup_ids = meta + .return_resumable_pools() + .into_iter() + .map(|pool| pool.id) + .collect::>(); + let store = decommission_worker_test_store(meta, vec![None, None, None, None]); + let endpoints = EndpointServerPools::from(vec![ + decommission_test_pool_endpoint(0, true), + decommission_test_pool_endpoint(1, true), + decommission_test_pool_endpoint(2, true), + decommission_test_pool_endpoint(3, true), + ]); + let reserved = store + .reserve_missing_local_decommission_routines(&CancellationToken::new(), &endpoints) + .await + .expect("runtime recovery reservation should succeed"); + let runtime_indices = reserved.iter().map(|(idx, _)| *idx).collect::>(); - let resumable = meta.return_resumable_pools(); + assert_eq!(runtime_indices, vec![2, 3]); + assert_eq!(startup_ids, vec![2, 3]); + } - assert_eq!(resumable.len(), 1); - assert_eq!(resumable[0].id, 1); + #[tokio::test] + async fn test_runtime_recovery_does_not_reserve_behind_active_predecessor() { + let meta = PoolMeta { + pools: vec![ + decommission_test_pool_status( + 0, + Some(PoolDecommissionInfo { + failed: true, + ..Default::default() + }), + ), + 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() + }), + ), + ], + ..Default::default() + }; + let active = DecommissionCanceler::new(CancellationToken::new()); + let store = decommission_worker_test_store(meta, vec![None, None, Some(active.clone()), None]); + let endpoints = EndpointServerPools::from(vec![ + decommission_test_pool_endpoint(0, true), + decommission_test_pool_endpoint(1, true), + decommission_test_pool_endpoint(2, true), + decommission_test_pool_endpoint(3, true), + ]); + + let reserved = store + .reserve_missing_local_decommission_routines(&CancellationToken::new(), &endpoints) + .await + .expect("runtime recovery reservation should succeed"); + + assert!(reserved.is_empty()); + assert!(active.is_active()); } #[test]