diff --git a/crates/ecstore/src/core/sets.rs b/crates/ecstore/src/core/sets.rs index 896033c36..b8804963e 100644 --- a/crates/ecstore/src/core/sets.rs +++ b/crates/ecstore/src/core/sets.rs @@ -286,7 +286,7 @@ impl Sets { self.get_disks(self.get_hashed_set_index(key)) } - fn get_disks_for_heal_object(&self, key: &str, opts: &HealOpts) -> Result> { + pub(crate) fn get_disks_for_heal_object(&self, key: &str, opts: &HealOpts) -> Result> { match opts.set { Some(set_idx) => self.disk_set.get(set_idx).cloned().ok_or_else(|| { StorageError::InvalidArgument( diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index fb1b19d6d..fdfba048a 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -129,7 +129,30 @@ impl ECStore { let mut futures = Vec::with_capacity(pools.len()); for pool in pools.iter() { - if self.is_suspended(pool.pool_idx).await { + let suspended_complete = { + let pool_meta = self.pool_meta.read().await; + pool_meta.is_suspended(pool.pool_idx).then(|| { + pool_meta + .pools + .get(pool.pool_idx) + .and_then(|status| status.decommission.as_ref()) + .is_some_and(|decommission| decommission.complete) + }) + }; + if let Some(complete) = suspended_complete { + if opts.pool.is_some() { + let _ = pool.get_disks_for_heal_object(&object, opts)?; + let err = if complete { + StorageError::InvalidArgument( + "heal".to_string(), + "pool".to_string(), + format!("heal pool {} has completed decommission", pool.pool_idx), + ) + } else { + Error::SlowDown + }; + return Ok((HealResultItem::default(), Some(err))); + } continue; } futures.push(pool.heal_object(bucket, &object, version_id, opts)); @@ -196,6 +219,7 @@ impl ECStore { #[cfg(test)] mod tests { use super::*; + use crate::core::pools::{PoolDecommissionInfo, PoolStatus}; use crate::disk::{DiskOption, format::FormatV3, new_disk}; use crate::layout::endpoints::{Endpoints, PoolEndpoints}; use crate::store::init_format::{load_format_erasure, save_format_file}; @@ -276,6 +300,134 @@ mod tests { ); } + #[tokio::test] + async fn scoped_heal_object_defers_when_requested_pool_is_suspended() { + let mut store = minimal_heal_store().await; + store.pool_meta = RwLock::new(PoolMeta { + pools: vec![ + PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: None, + }, + PoolStatus { + id: 1, + cmd_line: "pool-1".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(PoolDecommissionInfo { + start_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }), + }, + ], + ..Default::default() + }); + + let (_, err) = store + .handle_heal_object( + "bucket", + "object", + "", + &HealOpts { + pool: Some(1), + set: Some(0), + ..Default::default() + }, + ) + .await + .expect("suspended pool should return a deferred heal result"); + + assert!(matches!(err, Some(StorageError::SlowDown))); + + let (_, err) = store + .handle_heal_object( + "bucket", + "object", + "", + &HealOpts { + set: Some(1), + ..Default::default() + }, + ) + .await + .expect("unscoped heal should return the active pool result"); + + assert!(matches!(err, Some(StorageError::InvalidArgument(_, ref field, _)) if field == "set")); + + let err = store + .handle_heal_object( + "bucket", + "object", + "", + &HealOpts { + pool: Some(1), + set: Some(1), + ..Default::default() + }, + ) + .await + .expect_err("invalid set scope should fail before suspended pool deferral"); + + assert!(matches!(err, StorageError::InvalidArgument(_, ref field, _) if field == "set")); + + { + let mut pool_meta = store.pool_meta.write().await; + let decommission = pool_meta.pools[1] + .decommission + .as_mut() + .expect("test pool should have decommission state"); + decommission.complete = true; + } + let (_, err) = store + .handle_heal_object( + "bucket", + "object", + "", + &HealOpts { + pool: Some(1), + set: Some(0), + ..Default::default() + }, + ) + .await + .expect("completed pool should return a terminal heal result"); + + assert!(matches!( + err, + Some(StorageError::InvalidArgument(_, ref field, ref reason)) + if field == "pool" && reason.contains("completed decommission") + )); + + for canceled in [false, true] { + { + let mut pool_meta = store.pool_meta.write().await; + let decommission = pool_meta.pools[1] + .decommission + .as_mut() + .expect("test pool should have decommission state"); + decommission.complete = false; + decommission.failed = !canceled; + decommission.canceled = canceled; + } + let (_, err) = store + .handle_heal_object( + "bucket", + "object", + "", + &HealOpts { + pool: Some(1), + set: Some(0), + ..Default::default() + }, + ) + .await + .expect("clearable terminal pool should return a deferred heal result"); + + assert!(matches!(err, Some(StorageError::SlowDown))); + } + } + #[tokio::test] async fn handle_heal_format_continues_after_a_pool_error() { let canonical_format = FormatV3::new(1, 3); diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 595943575..4e1151656 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -3842,6 +3842,36 @@ mod tests { assert!(retry_error.contains("Storage resources are insufficient")); } + #[test] + fn test_retry_request_for_scoped_slowdown_preserves_scope() { + let storage: Arc = Arc::new(MockStorage); + let task = HealTask::from_request( + HealRequest::new( + HealType::Object { + bucket: "bucket".to_string(), + object: "object".to_string(), + version_id: None, + }, + HealOptions { + pool_index: Some(0), + set_index: Some(1), + ..Default::default() + }, + HealPriority::Normal, + ), + storage, + ); + let result = Err(Error::Storage(EcstoreError::SlowDown)); + + let (retry_request, retry_delay, _) = + retry_request_for_result(&task, &result).expect("SlowDown should defer scoped heal"); + + assert_eq!(retry_request.options.pool_index, Some(0)); + assert_eq!(retry_request.options.set_index, Some(1)); + assert_eq!(retry_request.retry_attempts, 1); + assert!(retry_delay > Duration::ZERO); + } + #[test] fn test_retry_request_for_typed_not_found_error_is_not_retryable() { let storage: Arc = Arc::new(MockStorage); diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 81b8ece16..fcdbd03ed 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -2517,6 +2517,7 @@ mod tests { OkWithOtherError(&'static str), ErrOther(&'static str), RetryableReadQuorum, + RetryableSlowDown, PermanentOther(&'static str), } @@ -2650,6 +2651,9 @@ mod tests { bucket.to_string(), object.to_string(), ))), + MockHealObjectOutcome::RetryableSlowDown => { + Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::SlowDown)))) + } MockHealObjectOutcome::PermanentOther(message) => Err(Error::other(message)), MockHealObjectOutcome::OkWithOtherError(message) => { Ok((HealResultItem::default(), Some(Error::other(message)))) @@ -2669,6 +2673,9 @@ mod tests { bucket.to_string(), object.to_string(), ))), + MockHealObjectOutcome::RetryableSlowDown => { + Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::SlowDown)))) + } }; } if bucket == RUSTFS_META_BUCKET && object == format!("{BUCKET_META_PREFIX}/{DATA_USAGE_CACHE_NAME}") { @@ -2761,6 +2768,42 @@ mod tests { } } + #[tokio::test] + async fn scoped_object_heal_slowdown_is_not_treated_as_deleted() { + let storage = Arc::new(MockStorage { + object_exists: Mutex::new(Some(true)), + heal_object_outcome: Mutex::new(Some(MockHealObjectOutcome::RetryableSlowDown)), + ..Default::default() + }); + let task = HealTask::from_request( + HealRequest::new( + HealType::Object { + bucket: "bucket".to_string(), + object: "object".to_string(), + version_id: None, + }, + HealOptions { + pool_index: Some(0), + set_index: Some(1), + ..Default::default() + }, + HealPriority::Normal, + ), + storage.clone(), + ); + + let err = task.execute().await.expect_err("SlowDown must fail the current heal attempt"); + + assert!(matches!(err, Error::Storage(EcstoreError::SlowDown))); + assert!(matches!(task.get_status().await, HealTaskStatus::Failed { .. })); + let opts = storage + .object_heal_opts + .lock() + .expect("heal options lock should be available"); + assert_eq!(opts[0].pool, Some(0)); + assert_eq!(opts[0].set, Some(1)); + } + async fn make_resume_disk(temp: &TempDir) -> DiskStore { let disk_path = temp.path().join("test_disk"); std::fs::create_dir_all(&disk_path).expect("test disk directory should be created");