fix(ecstore): fence decommission cancellation

This commit is contained in:
overtrue
2026-08-23 05:22:29 +08:00
parent acc37b49c8
commit 12f56c9ebd
2 changed files with 26 additions and 11 deletions
+3 -2
View File
@@ -3475,6 +3475,8 @@ impl ECStore {
async fn decommission_cancel_with_owner(&self, idx: usize, owner: Option<&DecommissionCanceler>) -> Result<()> { async fn decommission_cancel_with_owner(&self, idx: usize, owner: Option<&DecommissionCanceler>) -> Result<()> {
ensure_decommission_terminal_operation_supported(self.single_pool(), "cancel decommission")?; ensure_decommission_terminal_operation_supported(self.single_pool(), "cancel decommission")?;
let _start_guard = self.start_gate.lock().await; let _start_guard = self.start_gate.lock().await;
let operation_gate = self.ctx.decommission_operation_gate();
let operation_guard = operation_gate.write().await;
// Lock order: decommission_cancelers before pool_meta. Holding both makes // Lock order: decommission_cancelers before pool_meta. Holding both makes
// owner validation and the terminal transition one atomic operation. // owner validation and the terminal transition one atomic operation.
@@ -3533,8 +3535,6 @@ impl ECStore {
); );
} }
self.wait_for_decommission_side_effects().await;
if should_save_pool_meta && let Err(err) = self.save_current_pool_meta().await { if should_save_pool_meta && let Err(err) = self.save_current_pool_meta().await {
if let Some(previous_pool_meta) = previous_pool_meta { if let Some(previous_pool_meta) = previous_pool_meta {
let mut pool_meta = self.pool_meta.write().await; let mut pool_meta = self.pool_meta.write().await;
@@ -3542,6 +3542,7 @@ impl ECStore {
} }
return Err(err); return Err(err);
} }
drop(operation_guard);
if let Some(canceler) = terminal_canceler.as_ref() { if let Some(canceler) = terminal_canceler.as_ref() {
self.release_decommission_canceler_slot(idx, canceler).await; self.release_decommission_canceler_slot(idx, canceler).await;
+23 -9
View File
@@ -4616,11 +4616,24 @@ mod tests {
async move { store.decommission_cancel(0).await } async move { store.decommission_cancel(0).await }
}); });
assert!( assert!(
tokio::time::timeout(StdDuration::from_millis(100), &mut cancel) tokio::time::timeout(Duration::from_millis(100), &mut cancel).await.is_err(),
.await
.is_err(),
"cancel must wait for the final sweep source cleanup" "cancel must wait for the final sweep source cleanup"
); );
{
let pool_meta = store.pool_meta.read().await;
let decommission = pool_meta.pools[0]
.decommission
.as_ref()
.expect("decommission state should remain present");
assert!(
!decommission.canceled,
"cancel must not publish terminal state before the final sweep drains"
);
assert!(
decommission.start_time.is_some(),
"cancel must preserve the run identity until the final sweep drains"
);
}
barrier.release(); barrier.release();
final_sweep final_sweep
@@ -4631,12 +4644,13 @@ mod tests {
.await .await
.expect("cancel task should not panic") .expect("cancel task should not panic")
.expect("cancel should complete after the final sweep releases the operation gate"); .expect("cancel should complete after the final sweep releases the operation gate");
assert!( let pool_meta = store.pool_meta.read().await;
store.pool_meta.read().await.pools[0] let decommission = pool_meta.pools[0]
.decommission .decommission
.as_ref() .as_ref()
.is_some_and(|info| info.canceled) .expect("decommission state should remain present");
); assert!(decommission.canceled);
assert!(decommission.start_time.is_none());
} }
#[cfg(feature = "test-util")] #[cfg(feature = "test-util")]