fix(heal): retry unavailable recreate targets

crates/heal/src/heal/task.rs

@@ -24,7 +24,7 @@ use crate::heal::{
 use crate::{Error, Result};
 use metrics::{counter, histogram};
 use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit};
-use rustfs_heal_contracts::heal_channel::{HealOpts, HealRequestSource, HealScanMode};
+use rustfs_heal_contracts::heal_channel::{DriveState, HealOpts, HealRequestSource, HealScanMode};
 use rustfs_madmin::heal_commands::HealResultItem;
 use rustfs_utils::path::SLASH_SEPARATOR;
 use serde::{Deserialize, Serialize};

crates/heal/src/heal/task/heal_bucket.rs

@@ -16,6 +16,22 @@ use super::*;
 use crate::heal::progress::{add_bytes, increment_counter, stable_generation};
 use crate::heal::utils::format_set_disk_id;
 
+fn unavailable_recreate_error(result: &HealResultItem, opts: &HealOpts) -> Option<Error> {
+    if opts.dry_run || !opts.recreate {
+        return None;
+    }
+
+    let mut offline = false;
+    for drive in &result.after.drives {
+        if drive.state == DriveState::Faulty.to_str() {
+            return Some(Error::Disk(DiskError::FaultyDisk));
+        }
+        offline |= drive.state == DriveState::Offline.to_str();
+    }
+
+    offline.then_some(Error::Disk(DiskError::DiskNotFound))
+}
+
 impl HealTask {
     pub(super) async fn heal_bucket(&self, bucket: &str) -> Result<()> {
         debug!(
@@ -335,13 +351,16 @@ impl HealTask {
                             )
                             .await
                         {
-                            Ok((result, None)) => {
-                                telemetry_unknown |= !increment_counter(&mut healed);
-                                telemetry_unknown |=
-                                    !add_bytes(&mut bytes, u64::try_from(result.object_size).unwrap_or(u64::MAX));
-                                self.record_result_item(result).await;
-                                None
-                            }
+                            Ok((result, None)) => match unavailable_recreate_error(&result, &heal_opts) {
+                                Some(error) => Some(error),
+                                None => {
+                                    telemetry_unknown |= !increment_counter(&mut healed);
+                                    telemetry_unknown |=
+                                        !add_bytes(&mut bytes, u64::try_from(result.object_size).unwrap_or(u64::MAX));
+                                    self.record_result_item(result).await;
+                                    None
+                                }
+                            },
                             Ok((_, Some(err))) if is_missing_object_dir_heal_result(object, &err) => {
                                 telemetry_unknown |= !increment_counter(&mut healed);
                                 debug!(

crates/heal/src/heal/task/tests.rs

@@ -706,11 +706,28 @@ enum MockHealObjectOutcome {
     OkWithOtherError(&'static str),
     ErrOther(&'static str),
     DanglingGraceDeferred,
+    UnavailableDrive(DriveState),
     RetryableReadQuorum,
     RetryableSlowDown,
     PermanentOther(&'static str),
 }
 
+fn unavailable_drive_heal_result(state: DriveState) -> (HealResultItem, Option<Error>) {
+    (
+        HealResultItem {
+            after: Infos {
+                drives: vec![HealDriveInfo {
+                    endpoint: "remote-target".to_string(),
+                    state: state.to_string(),
+                    ..Default::default()
+                }],
+            },
+            ..Default::default()
+        },
+        None,
+    )
+}
+
 #[derive(Clone, Copy)]
 enum MockObjectExists {
     Exists(bool),
@@ -813,6 +830,7 @@ impl HealStorageAPI for MockStorage {
                         "dangling object deletion deferred by heal grace window; retry_after_secs=3599; grace_secs=3600",
                     ))),
                 )),
+                MockHealObjectOutcome::UnavailableDrive(state) => Ok(unavailable_drive_heal_result(state)),
                 MockHealObjectOutcome::RetryableReadQuorum => Err(Error::Storage(EcstoreError::InsufficientReadQuorum(
                     bucket.to_string(),
                     object.to_string(),
@@ -833,6 +851,7 @@ impl HealStorageAPI for MockStorage {
                         "dangling object deletion deferred by heal grace window; retry_after_secs=3599; grace_secs=3600",
                     ))),
                 )),
+                MockHealObjectOutcome::UnavailableDrive(state) => Ok(unavailable_drive_heal_result(state)),
                 MockHealObjectOutcome::OkWithOtherError(message) => Ok((HealResultItem::default(), Some(Error::other(message)))),
                 MockHealObjectOutcome::ErrOther(message) | MockHealObjectOutcome::PermanentOther(message) => {
                     Err(Error::other(message))
@@ -1449,6 +1468,46 @@ async fn test_recursive_bucket_heal_retries_only_retryable_objects() {
     assert_eq!(progress.objects_failed, 0);
 }
 
+#[tokio::test(start_paused = true)]
+async fn recursive_bucket_heal_retries_when_recreate_target_is_unavailable() {
+    for state in [DriveState::Offline, DriveState::Faulty] {
+        let state_name = state.to_string();
+        let storage = Arc::new(MockStorage::default());
+        storage
+            .heal_object_outcomes
+            .lock()
+            .unwrap()
+            .insert("object-a".to_string(), VecDeque::from([MockHealObjectOutcome::UnavailableDrive(state)]));
+        let request = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket-a".to_string(),
+            },
+            HealOptions {
+                recursive: true,
+                recreate_missing: true,
+                timeout: None,
+                ..Default::default()
+            },
+            HealPriority::Normal,
+        );
+        let task = HealTask::from_request(request, storage.clone());
+
+        task.heal_bucket("bucket-a")
+            .await
+            .expect("an unavailable recreate target should be retried after it returns");
+
+        assert_eq!(
+            storage.heal_object_calls.lock().unwrap().as_slice(),
+            ["object-a".to_string(), "object-b".to_string(), "object-a".to_string()],
+            "unexpected calls for unavailable state {state_name}"
+        );
+        let progress = task.get_progress().await;
+        assert_eq!(progress.objects_scanned, 2, "unexpected scanned count for state {state_name}");
+        assert_eq!(progress.objects_healed, 2, "unexpected healed count for state {state_name}");
+        assert_eq!(progress.objects_failed, 0, "unexpected failed count for state {state_name}");
+    }
+}
+
 #[tokio::test(start_paused = true)]
 async fn recursive_bucket_heal_skips_dangling_delete_grace_without_batch_failure() {
     let storage = Arc::new(MockStorage::default());