fix(heal): defer scoped repair on suspended pools (#5876)

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<Arc<SetDisks>> {
+    pub(crate) fn get_disks_for_heal_object(&self, key: &str, opts: &HealOpts) -> Result<Arc<SetDisks>> {
         match opts.set {
             Some(set_idx) => self.disk_set.get(set_idx).cloned().ok_or_else(|| {
                 StorageError::InvalidArgument(

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);

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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = Arc::new(MockStorage);

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");