gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-25 05:26:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): unify decommission resume queue (#6403)

Browse Source 
* fix(ecstore): unify decommission resume queue

* fix(ecstore): clear swift clippy warnings

* fix(ecstore): serialize decommission recovery

---------

Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
Zhengchao An
2026-08-23 23:51:33 +08:00
committed by GitHub
parent 2a43e021c9
commit 33341d5fcf
1 changed files with 160 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/core/pools.rs
+160 -57
View File
@@ -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<usize> {
fn resumable_decommission_queue_indices(meta: &PoolMeta) -> Vec<usize> {
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<PoolStatus> {
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::<Vec<_>>()
@@ -4268,6 +4258,19 @@ impl ECStore {
Ok(index_cancelers)
}
async fn reserve_missing_local_decommission_routines(
&self,
rx: &CancellationToken,
endpoints: &EndpointServerPools,
) -> Result<Vec<(usize, DecommissionCancelerGuard)>> {
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<ECStore>,
@@ -4288,17 +4291,9 @@ impl ECStore {
}
pub async fn spawn_missing_local_decommission_routines(self: &Arc<Self>) -> 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::<Vec<_>>();
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::<Vec<_>>();
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]