From 0f37c5675e838115d695415c080aa815f67c92b9 Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 16 Jun 2026 15:11:54 +0800 Subject: [PATCH] fix(ecstore): bound decommission bucket concurrency (#3498) --- crates/ecstore/src/pools.rs | 636 +++++++++++++++++++++++++++---- crates/ecstore/src/store/init.rs | 10 +- 2 files changed, 579 insertions(+), 67 deletions(-) diff --git a/crates/ecstore/src/pools.rs b/crates/ecstore/src/pools.rs index d0ffcfb50..5577ac1c8 100644 --- a/crates/ecstore/src/pools.rs +++ b/crates/ecstore/src/pools.rs @@ -44,7 +44,7 @@ use crate::store_api::{ }; use crate::{global::GLOBAL_LifecycleSys, sets::Sets, store::ECStore}; use byteorder::{ByteOrder, LittleEndian, WriteBytesExt}; -use futures::future::BoxFuture; +use futures::{StreamExt, future::BoxFuture, stream::FuturesUnordered}; use http::HeaderMap; #[cfg(test)] use rmp_serde::Deserializer; @@ -76,6 +76,10 @@ const LOG_SUBSYSTEM_POOLS: &str = "pools"; const EVENT_DECOMMISSION_STATE: &str = "decommission_state"; const EVENT_DECOMMISSION_BUCKET: &str = "decommission_bucket"; const EVENT_DECOMMISSION_ENTRY: &str = "decommission_entry"; +const DECOMMISSION_PROGRESS_SAVE_INTERVAL: Duration = Duration::seconds(30); +const DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD: usize = 1000; +const DECOMMISSION_BUCKET_CONCURRENCY_ENV: &str = "RUSTFS_DECOMMISSION_BUCKET_CONCURRENCY"; +const DECOMMISSION_BUCKET_CONCURRENCY_DEFAULT_CAP: usize = 4; pub const POOL_META_NAME: &str = "pool.bin"; pub const POOL_META_FORMAT: u16 = 1; @@ -141,6 +145,35 @@ fn ensure_decommission_routines_scheduled(bound_count: usize, expected_count: us Ok(()) } +fn default_decommission_bucket_concurrency(cpu_count: usize) -> usize { + cpu_count.clamp(1, DECOMMISSION_BUCKET_CONCURRENCY_DEFAULT_CAP) +} + +fn decommission_bucket_concurrency_limit() -> usize { + let default_limit = default_decommission_bucket_concurrency(num_cpus::get()); + rustfs_utils::get_env_usize(DECOMMISSION_BUCKET_CONCURRENCY_ENV, default_limit).max(1) +} + +fn is_decommission_meta_bucket(bucket: &DecomBucketInfo) -> bool { + bucket.name == RUSTFS_META_BUCKET +} + +fn split_decommission_buckets(buckets: Vec) -> (Vec, Vec) { + let mut regular = Vec::with_capacity(buckets.len()); + let mut meta = Vec::new(); + + for bucket in buckets { + if is_decommission_meta_bucket(&bucket) { + meta.push(bucket); + } else { + regular.push(bucket); + } + } + + regular.shrink_to_fit(); + (regular, meta) +} + fn ensure_decommission_not_rebalancing(rebalance_running: bool) -> Result<()> { if rebalance_running { return Err(Error::RebalanceAlreadyRunning); @@ -413,6 +446,13 @@ fn should_preserve_decommission_canceled_state(meta_canceled: bool, cancel_signa meta_canceled || cancel_signal } +fn should_continue_decommission_queue(meta: &PoolMeta, idx: usize) -> bool { + meta.pools + .get(idx) + .and_then(|pool| pool.decommission.as_ref()) + .is_some_and(|info| info.complete && !info.failed && !info.canceled) +} + fn decommission_cancel_signal_result(cancel_signal: bool) -> Result<()> { if cancel_signal { Err(StorageError::OperationCanceled) @@ -421,6 +461,59 @@ fn decommission_cancel_signal_result(cancel_signal: bool) -> Result<()> { } } +async fn run_decommission_buckets_bounded( + rx: CancellationToken, + buckets: Vec, + limit: usize, + mut start_bucket: F, +) -> Result<()> +where + F: FnMut(DecomBucketInfo, CancellationToken) -> BoxFuture<'static, Result<()>>, +{ + let mut pending = buckets.into_iter(); + let mut active: FuturesUnordered>> = FuturesUnordered::new(); + let mut first_err = None; + let limit = limit.max(1); + + for _ in 0..limit { + let Some(bucket) = pending.next() else { + break; + }; + + active.push(start_bucket(bucket, rx.clone())); + } + + while let Some(result) = active.next().await { + if let Err(err) = result { + rx.cancel(); + if first_err.is_none() { + first_err = Some(err); + } + continue; + } + + if first_err.is_some() || rx.is_cancelled() { + continue; + } + + let Some(bucket) = pending.next() else { + continue; + }; + + active.push(start_bucket(bucket, rx.clone())); + } + + if first_err.is_none() && rx.is_cancelled() && pending.len() > 0 { + return decommission_cancel_signal_result(true); + } + + if let Some(err) = first_err { + return Err(err); + } + + Ok(()) +} + fn is_decommission_cancel_terminal(complete: bool, failed: bool, canceled: bool) -> bool { complete || failed || canceled } @@ -452,6 +545,12 @@ fn validate_start_decommission_request(indices: &[usize], single_pool: bool) -> return Err(Error::other("failed to start decommission: no target pools were provided")); } + if indices.len() > 1 { + return Err(Error::other( + "failed to start decommission: decommission supports one target pool at a time", + )); + } + ensure_decommission_terminal_operation_supported(single_pool, "start decommission") } @@ -589,6 +688,7 @@ impl TryFrom for PoolDecommissionInfo { items_decommission_failed: value.items_decommission_failed, bytes_done: value.bytes_done, bytes_failed: value.bytes_failed, + progress_save_item_baseline: value.items_decommissioned.saturating_add(value.items_decommission_failed), }) } } @@ -653,6 +753,7 @@ impl PoolMeta { validate_decommission_terminal_state(decommission.complete, decommission.failed, decommission.canceled)?; } } + legacy.mark_decommission_progress_saved(); Ok(legacy) } } @@ -695,6 +796,14 @@ impl PoolMeta { self.pools.get(idx).is_some_and(|pool| pool.decommission.is_some()) } + fn mark_decommission_progress_saved(&mut self) { + for pool in &mut self.pools { + if let Some(info) = pool.decommission.as_mut() { + info.mark_progress_saved(); + } + } + } + pub async fn load(&mut self, pool: Arc, _pools: Vec>) -> Result<()> { let data = match read_config(pool, POOL_META_NAME).await { Ok(data) => { @@ -919,8 +1028,11 @@ impl PoolMeta { let pool_count = self.pools.len(); ensure_valid_decommission_pool_index(pool_count, idx)?; - let last_update = match self.pools.get(idx) { - Some(pool) if pool.decommission.is_some() => pool.last_update, + let (last_update, item_threshold_reached) = match self.pools.get(idx) { + Some(pool) if let Some(info) = pool.decommission.as_ref() => ( + pool.last_update, + info.items_since_last_progress_save() >= DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD, + ), Some(_) => { return Err(decommission_metadata_not_initialized_error("update decommission metadata timestamp")); } @@ -928,12 +1040,18 @@ impl PoolMeta { }; let now = OffsetDateTime::now_utc(); - if now.unix_timestamp() - last_update.unix_timestamp() > duration.whole_seconds() { + if now.unix_timestamp() - last_update.unix_timestamp() >= duration.whole_seconds() || item_threshold_reached { let Some(pool) = self.pools.get_mut(idx) else { return Err(invalid_decommission_pool_index_error(pool_count, idx)); }; pool.last_update = now; self.save(pools).await?; + let Some(pool) = self.pools.get_mut(idx) else { + return Err(invalid_decommission_pool_index_error(pool_count, idx)); + }; + if let Some(info) = pool.decommission.as_mut() { + info.mark_progress_saved(); + } return Ok(true); } @@ -1065,9 +1183,23 @@ pub struct PoolDecommissionInfo { pub bytes_done: usize, #[serde(rename = "bytesDecommissionedFailed")] pub bytes_failed: usize, + #[serde(skip)] + pub progress_save_item_baseline: usize, } impl PoolDecommissionInfo { + fn counted_items(&self) -> usize { + self.items_decommissioned.saturating_add(self.items_decommission_failed) + } + + fn items_since_last_progress_save(&self) -> usize { + self.counted_items().saturating_sub(self.progress_save_item_baseline) + } + + fn mark_progress_saved(&mut self) { + self.progress_save_item_baseline = self.counted_items(); + } + pub fn bucket_push(&mut self, bucket: &DecomBucketInfo) { for b in self.queued_buckets.iter() { if self.is_bucket_decommissioned(b) { @@ -1370,9 +1502,26 @@ impl ECStore { ensure_decommission_routines_scheduled(index_cancelers.len(), indices.len())?; - for (idx, canceler) in index_cancelers { - let store = store.clone(); - tokio::spawn(async move { + tokio::spawn(async move { + let mut stop_queue = false; + + for (idx, canceler) in index_cancelers { + if stop_queue || rx.is_cancelled() { + canceler.cancel(); + if let Err(err) = store.decommission_cancel(idx).await { + warn!( + event = EVENT_DECOMMISSION_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + pool_index = idx, + state = "queued_cancel_failed", + error = %err, + "Failed to cancel queued decommission" + ); + } + continue; + } + if let Err(err) = store.do_decommission_in_routine(canceler, idx).await { error!( event = EVENT_DECOMMISSION_STATE, @@ -1383,9 +1532,16 @@ impl ECStore { error = %err, "Decommission routine failed" ); + stop_queue = true; + continue; } - }); - } + + stop_queue = { + let pool_meta = store.pool_meta.read().await; + !should_continue_decommission_queue(&pool_meta, idx) + }; + } + }); Ok(()) } @@ -1775,7 +1931,9 @@ impl ECStore { } let ok = match resolve_decommission_update_after_result( - pool_meta.update_after(idx, self.pools.clone(), Duration::seconds(30)).await, + pool_meta + .update_after(idx, self.pools.clone(), DECOMMISSION_PROGRESS_SAVE_INTERVAL) + .await, ) { Ok(ok) => ok, Err(err) => { @@ -2254,6 +2412,81 @@ impl ECStore { Ok(()) } + async fn decommission_pending_bucket( + self: &Arc, + rx: CancellationToken, + idx: usize, + pool: Arc, + bucket: DecomBucketInfo, + ) -> Result<()> { + let is_decommissioned = { + let pool_meta = self.pool_meta.read().await; + resolve_decommission_bucket_state(&pool_meta, idx, &bucket)? + }; + + if is_decommissioned { + warn!("decommission: already done, moving on {}", bucket.to_string()); + + { + let mut pool_meta = self.pool_meta.write().await; + if mark_decommission_bucket_done(&mut pool_meta, idx, &bucket)? { + resolve_decommission_bucket_done_save_result( + pool_meta.save(self.pools.clone()).await, + idx, + bucket.name.as_str(), + )?; + } + } + return Ok(()); + } + + warn!("decommission: currently on bucket {}", &bucket.name); + + if let Err(err) = self.decommission_pool(rx.clone(), idx, pool, bucket.clone()).await { + error!("decommission: decommission_pool err {:?}", &err); + return Err(err); + } else { + warn!("decommission: decommission_pool done {}", &bucket.name); + } + + if let Err(err) = decommission_cancel_signal_result(rx.is_cancelled()) { + warn!("decommission: cancellation observed after decommission_pool {}", &bucket.name); + return Err(err); + } + + { + let mut pool_meta = self.pool_meta.write().await; + if mark_decommission_bucket_done(&mut pool_meta, idx, &bucket)? { + resolve_decommission_bucket_done_save_result( + pool_meta.save(self.pools.clone()).await, + idx, + bucket.name.as_str(), + )?; + } + + warn!("decommission: decommission_pool bucket_done {}", &bucket.name); + } + + Ok(()) + } + + async fn decommission_buckets_concurrently( + self: &Arc, + rx: CancellationToken, + idx: usize, + pool: Arc, + buckets: Vec, + limit: usize, + ) -> Result<()> { + let store = Arc::clone(self); + run_decommission_buckets_bounded(rx, buckets, limit, move |bucket, rx| { + let store = Arc::clone(&store); + let pool = pool.clone(); + Box::pin(async move { store.decommission_pending_bucket(rx, idx, pool, bucket).await }) + }) + .await + } + #[tracing::instrument(skip(self, rx))] async fn decommission_in_background(self: &Arc, rx: CancellationToken, idx: usize) -> Result<()> { let pool = get_by_index(self.pools.as_slice(), idx, "load decommission background pool")?.clone(); @@ -2263,54 +2496,22 @@ impl ECStore { pool_meta.pending_buckets(idx) }; - for bucket in pending.iter() { - let is_decommissioned = { - let pool_meta = self.pool_meta.read().await; - resolve_decommission_bucket_state(&pool_meta, idx, bucket)? - }; - - if is_decommissioned { - warn!("decommission: already done, moving on {}", bucket.to_string()); - - { - let mut pool_meta = self.pool_meta.write().await; - if mark_decommission_bucket_done(&mut pool_meta, idx, bucket)? { - resolve_decommission_bucket_done_save_result( - pool_meta.save(self.pools.clone()).await, - idx, - bucket.name.as_str(), - )?; - } - } - continue; + let bucket_concurrency = decommission_bucket_concurrency_limit(); + if bucket_concurrency <= 1 { + for bucket in pending { + self.decommission_pending_bucket(rx.clone(), idx, pool.clone(), bucket) + .await?; } + return Ok(()); + } - warn!("decommission: currently on bucket {}", &bucket.name); + let (regular_buckets, meta_buckets) = split_decommission_buckets(pending); + self.decommission_buckets_concurrently(rx.clone(), idx, pool.clone(), regular_buckets, bucket_concurrency) + .await?; - if let Err(err) = self.decommission_pool(rx.clone(), idx, pool.clone(), bucket.clone()).await { - error!("decommission: decommission_pool err {:?}", &err); - return Err(err); - } else { - warn!("decommission: decommission_pool done {}", &bucket.name); - } - - if let Err(err) = decommission_cancel_signal_result(rx.is_cancelled()) { - warn!("decommission: cancellation observed after decommission_pool {}", &bucket.name); - return Err(err); - } - - { - let mut pool_meta = self.pool_meta.write().await; - if mark_decommission_bucket_done(&mut pool_meta, idx, bucket)? { - resolve_decommission_bucket_done_save_result( - pool_meta.save(self.pools.clone()).await, - idx, - bucket.name.as_str(), - )?; - } - - warn!("decommission: decommission_pool bucket_done {}", &bucket.name); - } + for bucket in meta_buckets { + self.decommission_pending_bucket(rx.clone(), idx, pool.clone(), bucket) + .await?; } Ok(()) @@ -2747,6 +2948,7 @@ mod tests { assert_eq!(restored_decommission.items_decommission_failed, 1); assert_eq!(restored_decommission.bytes_done, 1024); assert_eq!(restored_decommission.bytes_failed, 128); + assert_eq!(restored_decommission.items_since_last_progress_save(), 0); } #[test] @@ -2798,6 +3000,7 @@ mod tests { assert_eq!(decommission.items_decommission_failed, 2); assert_eq!(decommission.bytes_done, 2048); assert_eq!(decommission.bytes_failed, 256); + assert_eq!(decommission.items_since_last_progress_save(), 0); // These fields were skipped in legacy payload and should be defaulted. assert!(decommission.queued_buckets.is_empty()); assert!(decommission.decommissioned_buckets.is_empty()); @@ -3104,28 +3307,35 @@ pub(crate) fn fallback_free_capacity_dedup(disks: &[rustfs_madmin::Disk]) -> usi #[cfg(test)] mod pools_tests { use super::{ - DecomBucketInfo, DecommissionTerminalState, PoolDecommissionInfo, PoolMeta, PoolStatus, bind_decommission_cancelers, + DECOMMISSION_PROGRESS_SAVE_INTERVAL, DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD, DecomBucketInfo, + DecommissionTerminalState, PoolDecommissionInfo, PoolMeta, PoolStatus, bind_decommission_cancelers, cancel_decommission_canceler, classify_decommission_terminal_state, count_decommission_item, decommission_cancel_signal_result, decommission_item_size, decommission_meta_bucket_options, - decommission_start_guard_state, dedup_indices, ensure_decommission_cancel_allowed, - ensure_decommission_listing_disks_available, ensure_decommission_not_rebalancing, ensure_decommission_start_allowed, - ensure_decommission_terminal_operation_supported, ensure_valid_decommission_pool_index, get_by_index, - has_active_decommission_canceler, is_decommission_active, is_decommission_cancel_terminal, - load_decommission_entry_versions, mark_decommission_bucket_done, require_decommission_store, - resolve_decommission_bucket_done_save_result, resolve_decommission_bucket_state, + decommission_start_guard_state, dedup_indices, default_decommission_bucket_concurrency, + ensure_decommission_cancel_allowed, ensure_decommission_listing_disks_available, ensure_decommission_not_rebalancing, + ensure_decommission_start_allowed, ensure_decommission_terminal_operation_supported, + ensure_valid_decommission_pool_index, get_by_index, has_active_decommission_canceler, is_decommission_active, + is_decommission_cancel_terminal, load_decommission_entry_versions, mark_decommission_bucket_done, + require_decommission_store, 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_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_spawn_failure_result, resolve_decommission_terminal_mark_after_error_result, resolve_decommission_terminal_mark_result, - resolve_decommission_update_after_result, should_cleanup_decommission_source_entry, - should_count_decommission_version_complete, should_preserve_decommission_canceled_state, take_decommission_canceler, + resolve_decommission_update_after_result, run_decommission_buckets_bounded, should_cleanup_decommission_source_entry, + should_continue_decommission_queue, should_count_decommission_version_complete, + should_preserve_decommission_canceled_state, split_decommission_buckets, take_decommission_canceler, track_decommission_current_object, validate_start_decommission_request, with_decommission_entry_context, }; use crate::data_movement; use crate::error::Error; use rustfs_filemeta::MetaCacheEntry; use rustfs_rio::Index; + use std::sync::{ + Arc, + atomic::{AtomicBool, AtomicUsize, Ordering}, + }; + use std::time::Duration as StdDuration; use time::{Duration, OffsetDateTime}; use tokio_util::sync::CancellationToken; @@ -3140,6 +3350,160 @@ mod pools_tests { assert!(dedup_indices(&empty).is_empty()); } + #[test] + fn test_default_decommission_bucket_concurrency_is_conservative() { + assert_eq!(default_decommission_bucket_concurrency(0), 1); + assert_eq!(default_decommission_bucket_concurrency(1), 1); + assert_eq!(default_decommission_bucket_concurrency(2), 2); + assert_eq!(default_decommission_bucket_concurrency(8), 4); + } + + #[test] + fn test_split_decommission_buckets_keeps_meta_buckets_last() { + let (regular, meta) = split_decommission_buckets(vec![ + DecomBucketInfo { + name: "bucket-a".to_string(), + ..Default::default() + }, + DecomBucketInfo { + name: crate::disk::RUSTFS_META_BUCKET.to_string(), + prefix: crate::config::com::CONFIG_PREFIX.to_string(), + }, + DecomBucketInfo { + name: "bucket-b".to_string(), + ..Default::default() + }, + DecomBucketInfo { + name: crate::disk::RUSTFS_META_BUCKET.to_string(), + prefix: crate::disk::BUCKET_META_PREFIX.to_string(), + }, + ]); + + assert_eq!( + regular.iter().map(|bucket| bucket.name.as_str()).collect::>(), + vec!["bucket-a", "bucket-b",] + ); + assert_eq!( + meta.iter().map(|bucket| bucket.prefix.as_str()).collect::>(), + vec![crate::config::com::CONFIG_PREFIX, crate::disk::BUCKET_META_PREFIX,] + ); + } + + #[tokio::test] + async fn test_run_decommission_buckets_bounded_respects_limit() { + let rx = CancellationToken::new(); + let running = Arc::new(AtomicUsize::new(0)); + let max_running = Arc::new(AtomicUsize::new(0)); + let started = Arc::new(AtomicUsize::new(0)); + let buckets = (0..8) + .map(|idx| DecomBucketInfo { + name: format!("bucket-{idx}"), + ..Default::default() + }) + .collect::>(); + + run_decommission_buckets_bounded(rx, buckets, 2, { + let running = Arc::clone(&running); + let max_running = Arc::clone(&max_running); + let started = Arc::clone(&started); + move |_bucket, _rx| { + let running = Arc::clone(&running); + let max_running = Arc::clone(&max_running); + let started = Arc::clone(&started); + Box::pin(async move { + started.fetch_add(1, Ordering::SeqCst); + let current = running.fetch_add(1, Ordering::SeqCst) + 1; + max_running.fetch_max(current, Ordering::SeqCst); + tokio::time::sleep(StdDuration::from_millis(10)).await; + running.fetch_sub(1, Ordering::SeqCst); + Ok(()) + }) + } + }) + .await + .expect("bounded bucket scheduler should complete"); + + assert_eq!(started.load(Ordering::SeqCst), 8); + assert_eq!(max_running.load(Ordering::SeqCst), 2); + assert_eq!(running.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn test_run_decommission_buckets_bounded_cancels_and_stops_launching_after_failure() { + let rx = CancellationToken::new(); + let started = Arc::new(AtomicUsize::new(0)); + let observed_cancel = Arc::new(AtomicBool::new(false)); + let buckets = (0..5) + .map(|idx| DecomBucketInfo { + name: format!("bucket-{idx}"), + ..Default::default() + }) + .collect::>(); + + let err = tokio::time::timeout( + StdDuration::from_secs(2), + run_decommission_buckets_bounded(rx.clone(), buckets, 2, { + let started = Arc::clone(&started); + let observed_cancel = Arc::clone(&observed_cancel); + move |bucket, rx| { + let started = Arc::clone(&started); + let observed_cancel = Arc::clone(&observed_cancel); + Box::pin(async move { + started.fetch_add(1, Ordering::SeqCst); + if bucket.name == "bucket-0" { + while started.load(Ordering::SeqCst) < 2 { + tokio::task::yield_now().await; + } + return Err(Error::SlowDown); + } + + rx.cancelled().await; + observed_cancel.store(true, Ordering::SeqCst); + Ok(()) + }) + } + }), + ) + .await + .expect("bucket scheduler should not hang after a bucket failure") + .expect_err("first bucket failure should be returned"); + + assert!(matches!(err, Error::SlowDown)); + assert!(rx.is_cancelled()); + assert!(observed_cancel.load(Ordering::SeqCst)); + assert_eq!(started.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn test_run_decommission_buckets_bounded_external_cancel_stops_pending_buckets() { + let rx = CancellationToken::new(); + let started = Arc::new(AtomicUsize::new(0)); + let buckets = (0..4) + .map(|idx| DecomBucketInfo { + name: format!("bucket-{idx}"), + ..Default::default() + }) + .collect::>(); + + let err = run_decommission_buckets_bounded(rx.clone(), buckets, 1, { + let started = Arc::clone(&started); + move |_bucket, rx| { + let started = Arc::clone(&started); + Box::pin(async move { + started.fetch_add(1, Ordering::SeqCst); + rx.cancel(); + Ok(()) + }) + } + }) + .await + .expect_err("external cancellation with pending buckets should stop the scheduler"); + + assert!(matches!(err, Error::OperationCanceled)); + assert!(rx.is_cancelled()); + assert_eq!(started.load(Ordering::SeqCst), 1); + } + #[test] fn test_get_by_index_returns_value_when_in_range() { let values = vec!["a", "b", "c"]; @@ -3735,6 +4099,81 @@ mod pools_tests { ); } + #[tokio::test] + async fn test_pool_meta_update_after_skips_before_time_and_item_thresholds() { + let mut meta = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::now_utc(), + decommission: Some(PoolDecommissionInfo { + items_decommissioned: DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD - 1, + ..Default::default() + }), + }], + ..Default::default() + }; + + let saved = meta + .update_after(0, Vec::new(), DECOMMISSION_PROGRESS_SAVE_INTERVAL) + .await + .expect("valid decommission state should update"); + + assert!(!saved); + let info = meta.pools[0].decommission.as_ref().expect("decommission info should exist"); + assert_eq!(info.items_since_last_progress_save(), DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD - 1); + } + + #[tokio::test] + async fn test_pool_meta_update_after_saves_when_item_threshold_reached() { + let mut meta = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::now_utc(), + decommission: Some(PoolDecommissionInfo { + items_decommissioned: DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD, + ..Default::default() + }), + }], + ..Default::default() + }; + + let saved = meta + .update_after(0, Vec::new(), DECOMMISSION_PROGRESS_SAVE_INTERVAL) + .await + .expect("item threshold should save progress"); + + assert!(saved); + let info = meta.pools[0].decommission.as_ref().expect("decommission info should exist"); + assert_eq!(info.items_since_last_progress_save(), 0); + } + + #[tokio::test] + async fn test_pool_meta_update_after_saves_when_time_threshold_reached() { + let mut meta = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::now_utc() - DECOMMISSION_PROGRESS_SAVE_INTERVAL, + decommission: Some(PoolDecommissionInfo { + items_decommissioned: 1, + ..Default::default() + }), + }], + ..Default::default() + }; + + let saved = meta + .update_after(0, Vec::new(), DECOMMISSION_PROGRESS_SAVE_INTERVAL) + .await + .expect("time threshold should save progress"); + + assert!(saved); + let info = meta.pools[0].decommission.as_ref().expect("decommission info should exist"); + assert_eq!(info.items_since_last_progress_save(), 0); + } + #[test] fn test_ensure_decommission_not_rebalancing_rejects_running_rebalance() { let err = ensure_decommission_not_rebalancing(true).expect_err("rebalance running should be rejected"); @@ -3874,6 +4313,61 @@ mod pools_tests { assert!(!should_preserve_decommission_canceled_state(false, false)); } + #[test] + fn test_should_continue_decommission_queue_requires_clean_completion() { + let meta = PoolMeta { + pools: vec![ + PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(PoolDecommissionInfo { + complete: true, + ..Default::default() + }), + }, + PoolStatus { + id: 1, + cmd_line: "pool-1".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(PoolDecommissionInfo::default()), + }, + PoolStatus { + id: 2, + cmd_line: "pool-2".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(PoolDecommissionInfo { + failed: true, + ..Default::default() + }), + }, + PoolStatus { + id: 3, + cmd_line: "pool-3".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(PoolDecommissionInfo { + canceled: true, + ..Default::default() + }), + }, + PoolStatus { + id: 4, + cmd_line: "pool-4".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: None, + }, + ], + ..Default::default() + }; + + assert!(should_continue_decommission_queue(&meta, 0)); + assert!(!should_continue_decommission_queue(&meta, 1)); + assert!(!should_continue_decommission_queue(&meta, 2)); + assert!(!should_continue_decommission_queue(&meta, 3)); + assert!(!should_continue_decommission_queue(&meta, 4)); + assert!(!should_continue_decommission_queue(&meta, 5)); + } + #[test] fn test_decommission_cancel_signal_result_returns_err_when_canceled() { let err = decommission_cancel_signal_result(true).expect_err("canceled signal should return operation-canceled"); @@ -3967,8 +4461,18 @@ mod pools_tests { } #[test] - fn test_contextualized_decommission_start_request_allows_non_empty_multi_pool() { - assert!(validate_start_decommission_request(&[0, 1], false).is_ok()); + fn test_contextualized_decommission_start_request_rejects_multiple_target_pools() { + let err = validate_start_decommission_request(&[0, 1], false) + .expect_err("multiple target pools should be rejected for single active draining"); + assert!( + err.to_string() + .contains("failed to start decommission: decommission supports one target pool at a time") + ); + } + + #[test] + fn test_contextualized_decommission_start_request_allows_one_target_pool() { + assert!(validate_start_decommission_request(&[0], false).is_ok()); } #[test] diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 9df2f6e6f..859b6c6d8 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -71,7 +71,15 @@ async fn resume_local_decommission_after_init(store: Arc, rx: Cancellat return; } - match store.decommission(rx.clone(), pool_indices.clone()).await { + let result = if pool_indices.len() > 1 { + store + .spawn_decommission_routines(store.clone(), rx.clone(), pool_indices.clone()) + .await + } else { + store.decommission(rx.clone(), pool_indices.clone()).await + }; + + match result { Ok(()) => return, Err(err) if is_err_decommission_running(&err) => { if let Err(spawn_err) = store