From 8c51c6f18cbc17034d00c4849271e21e44b37a41 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A9=AC=E7=99=BB=E5=B1=B1?= Date: Sat, 22 Aug 2026 16:29:28 +0800 Subject: [PATCH] fix(ecstore): fence decommission routine generations --- crates/ecstore/src/core/pools.rs | 119 ++++++++++++++++++++++--------- 1 file changed, 84 insertions(+), 35 deletions(-) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 0d96899c5..1d3aa84b6 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -1247,6 +1247,7 @@ fn is_decommission_cancel_requested(cancel_signal: bool, pool: Option<&PoolStatu .is_some_and(|info| info.canceled) } +#[cfg(test)] fn should_skip_canceled_decommission_routine(cancel_signal: bool, pool: Option<&PoolStatus>) -> bool { cancel_signal && pool @@ -2854,13 +2855,28 @@ impl ECStore { #[tracing::instrument(skip(self))] pub async fn decommission_cancel(&self, idx: usize) -> Result<()> { + self.decommission_cancel_with_generation(idx, None).await + } + + async fn decommission_cancel_with_generation(&self, idx: usize, generation: Option) -> Result<()> { ensure_decommission_terminal_operation_supported(self.single_pool(), "cancel decommission")?; let _start_guard = self.start_gate.lock().await; - let canceled_worker = self.cancel_decommission_routine_and_wait(idx).await; + let canceled_worker = if generation.is_some() { + false + } else { + self.cancel_decommission_routine_and_wait(idx).await + }; let (should_save_pool_meta, should_reload_pool_meta, already_canceled, previous_pool_meta) = { let mut lock = self.pool_meta.write().await; + if let Some(generation) = generation { + match ensure_decommission_generation(&lock, idx, generation) { + Ok(()) => {} + Err(Error::OperationCanceled) => return Ok(()), + Err(err) => return Err(err), + } + } let mut already_canceled = false; let (pool_present, decommission_present, terminal) = if let Some(pool) = lock.pools.get(idx) { if let Some(info) = pool.decommission.as_ref() { @@ -2888,6 +2904,12 @@ impl ECStore { ) }; + if generation.is_some() && should_save_pool_meta { + let mut cancelers = self.decommission_cancelers.write().await; + take_and_cancel_decommission_canceler(cancelers.as_mut_slice(), idx); + } + let canceled_worker = canceled_worker || (generation.is_some() && should_save_pool_meta); + if !canceled_worker && !already_canceled { warn!( event = EVENT_DECOMMISSION_STATE, @@ -2960,21 +2982,48 @@ impl ECStore { Ok(()) } - async fn promote_queued_decommission(&self, idx: usize) -> Result<()> { - let promoted = { + async fn promote_queued_decommission(&self, idx: usize) -> Result { + // Serialize promotion and generation capture with clear/restart transitions. + let (promoted, generation, save_error) = { + let _start_guard = self.start_gate.lock().await; let mut pool_meta = self.pool_meta.write().await; - pool_meta.promote_queued_decommission(idx) + let promoted = pool_meta.promote_queued_decommission(idx); + drop(pool_meta); + + let save_error = if promoted { + self.save_current_pool_meta().await.err() + } else { + None + }; + + let generation = self.active_decommission_generation(idx).await?; + (promoted, generation, save_error) }; - if promoted { - self.save_current_pool_meta().await?; - if let Some(notification_sys) = runtime_sources::notification_sys() { - let stage = format!("promote_queued_decommission for pool {idx}"); - resolve_decommission_pool_meta_reload_result(notification_sys.reload_pool_meta().await, stage.as_str())?; + if let Some(err) = save_error { + resolve_decommission_terminal_mark_after_error_result( + self.decommission_failed_with_generation(idx, Some(generation)).await, + idx, + &err, + )?; + return Err(err); + } + + if promoted && let Some(notification_sys) = runtime_sources::notification_sys() { + let stage = format!("promote_queued_decommission for pool {idx}"); + if let Err(err) = + resolve_decommission_pool_meta_reload_result(notification_sys.reload_pool_meta().await, stage.as_str()) + { + resolve_decommission_terminal_mark_after_error_result( + self.decommission_failed_with_generation(idx, Some(generation)).await, + idx, + &err, + )?; + return Err(err); } } - Ok(()) + Ok(generation) } async fn record_decommission_terminal_reload_failure(&self, idx: usize, stage: &str, err: Error) -> Result<()> { @@ -4158,36 +4207,26 @@ impl ECStore { idx: usize, entry_budget: Arc, ) -> Result<()> { - if let Err(err) = self.promote_queued_decommission(idx).await { - resolve_decommission_terminal_mark_after_error_result(self.decommission_failed(idx).await, idx, &err)?; - return Err(err); - } + let generation = match self.promote_queued_decommission(idx).await { + Ok(generation) => generation, + Err(Error::OperationCanceled) => return Ok(()), + Err(err) => return Err(err), + }; + defer!(|| async { + self.clear_decommission_canceler_for_generation(idx, generation).await; + }); + if rx.is_cancelled() { - let already_canceled = { - let pool_meta = self.pool_meta.read().await; - should_skip_canceled_decommission_routine(true, pool_meta.pools.get(idx)) - }; - if already_canceled { - warn!( - event = EVENT_DECOMMISSION_STATE, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_POOLS, - pool_index = idx, - state = "canceled_preserved", - "Decommission routine skipped because pool is already canceled" - ); - return Ok(()); - } - if let Err(err) = self.decommission_cancel(idx).await { - resolve_decommission_terminal_mark_after_error_result(self.decommission_failed(idx).await, idx, &err)?; + if let Err(err) = self.decommission_cancel_with_generation(idx, Some(generation)).await { + resolve_decommission_terminal_mark_after_error_result( + self.decommission_failed_with_generation(idx, Some(generation)).await, + idx, + &err, + )?; return Err(err); } return Ok(()); } - let generation = self.active_decommission_generation(idx).await?; - defer!(|| async { - self.clear_decommission_canceler_for_generation(idx, generation).await; - }); let result = self.decommission_in_background(rx.clone(), idx, entry_budget).await; let (final_state, canceled, cmd_line) = { @@ -7878,6 +7917,16 @@ mod pools_tests { .expect("decommission metadata should exist") .queued = true; assert!(ensure_decommission_generation(&meta, 0, generation).is_err()); + + let replacement_generation = generation + Duration::seconds(2); + let info = meta.pools[0] + .decommission + .as_mut() + .expect("decommission metadata should exist"); + info.queued = false; + info.start_time = Some(replacement_generation); + assert!(ensure_decommission_generation(&meta, 0, generation).is_err()); + assert!(ensure_decommission_generation(&meta, 0, replacement_generation).is_ok()); } #[test]