fix(heal): preserve resume state and retry when heal objects fail (backlog#855) (#4308)

fix(heal): don't mark set healed / discard resume state when objects failed (backlog#855)

`execute_heal_with_resume` unconditionally called `mark_completed()` and
returned `Ok(())` after the bucket loop, even when objects failed. The caller
then treats `Ok` as success and cleans up the resume + checkpoint state, so a
heal run with per-object failures was reported as a clean completion and its
state (including the failed set) was destroyed with no retry.

Finalize based on the failure count instead:

- failed_objects == 0 -> mark completed (unchanged);
- failed_objects > 0 with retry budget left -> `schedule_retry()` (bump the
  bounded retry counter + reset per-pass progress for a full re-scan) and
  return Err, so `heal_erasure_set` preserves the resume/checkpoint state and
  the caller keeps the healing markers for the next run;
- failed_objects > 0 with retries exhausted -> drop the resume state (no
  zombie task) but still return Err so the markers survive for a later heal
  cycle / the background scanner. Never silently claim a clean completion.

The failure identities are not persisted across pages (the per-page sets are
pruned in `complete_page`), so a retry is a bounded full re-scan; healed
objects are skipped quickly on the normal-scan pass.

Adds `ResumeState::reset_for_retry`, `ResumeManager::schedule_retry`,
`ResumeCheckpoint::reset_for_retry`, `CheckpointManager::reset_for_retry`, and
regression tests. This activates the failure path the caller already documents
at task.rs ("Keep the markers on failure ... the next run re-marks and
eventually clears them"), which was previously dead because heal never
returned Err on object failures.

Refs backlog#799 (B6).

crates/heal/src/heal/erasure_healer.rs

@@ -441,7 +441,53 @@ impl ErasureSetHealer {
             current_object_index = 0;
         }
 
-        // 5. mark task completed
+        // 5. finalize. Only declare the set healed when nothing failed —
+        // otherwise the resume/checkpoint state must survive so the failures are
+        // retried instead of being silently marked "completed" and discarded
+        // (backlog#855 / #799 B6).
+        if failed_objects > 0 {
+            if resume_manager.schedule_retry().await? {
+                // Retry budget remains: state has been reset for a full re-scan.
+                // Return Err so `heal_erasure_set` preserves (does not clean up)
+                // the resume/checkpoint state and the caller keeps the healing
+                // markers for the next heal run.
+                warn!(
+                    target: "rustfs::heal::erasure_healer",
+                    event = EVENT_HEAL_ERASURE_RESUME_STATE,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
+                    set_disk_id,
+                    failed_objects,
+                    state = "retry_scheduled",
+                    "Erasure set heal pass finished with failures; scheduled full re-heal retry"
+                );
+                return Err(Error::other(format!(
+                    "Erasure set heal incomplete: {failed_objects} object(s) failed; retry scheduled"
+                )));
+            }
+
+            // Retry budget exhausted: drop the resume/checkpoint state so this
+            // task does not loop, but keep the healing markers (return Err) so a
+            // later heal cycle / the background scanner starts a fresh attempt.
+            // Never silently claim a clean completion while objects are unhealed.
+            error!(
+                target: "rustfs::heal::erasure_healer",
+                event = EVENT_HEAL_ERASURE_RESUME_STATE,
+                component = LOG_COMPONENT_HEAL,
+                subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
+                set_disk_id,
+                failed_objects,
+                state = "failed_after_retries",
+                "Erasure set heal exhausted retries with unrecovered failures"
+            );
+            let _ = resume_manager.cleanup().await;
+            let _ = checkpoint_manager.cleanup().await;
+            return Err(Error::other(format!(
+                "Erasure set heal exhausted retries with {failed_objects} unrecovered object(s)"
+            )));
+        }
+
+        // no failures — mark task completed
         resume_manager.mark_completed().await?;
 
         debug!(

crates/heal/src/heal/resume.rs

@@ -168,6 +168,19 @@ impl ResumeState {
         self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
     }
 
+    /// Reset per-pass progress so a retry re-scans the whole set from the
+    /// start. `retry_count`/`max_retries` are intentionally preserved so
+    /// retries stay bounded.
+    pub fn reset_for_retry(&mut self) {
+        self.completed_buckets.clear();
+        self.processed_objects = 0;
+        self.successful_objects = 0;
+        self.failed_objects = 0;
+        self.skipped_objects = 0;
+        self.completed = false;
+        self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
+    }
+
     pub fn set_error(&mut self, error: String) {
         self.error_message = Some(error);
         self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
@@ -331,6 +344,22 @@ impl ResumeManager {
         self.save_state().await
     }
 
+    /// Arm a bounded retry: if the retry budget remains, bump the retry
+    /// counter, reset per-pass progress (so the next resume re-scans the whole
+    /// set), persist, and return `true`. Returns `false` when retries are
+    /// exhausted, leaving the state untouched.
+    pub async fn schedule_retry(&self) -> Result<bool> {
+        let mut state = self.state.write().await;
+        if !state.can_retry() {
+            return Ok(false);
+        }
+        state.increment_retry();
+        state.reset_for_retry();
+        drop(state);
+        self.save_state().await?;
+        Ok(true)
+    }
+
     /// cleanup resume state
     pub async fn cleanup(&self) -> Result<()> {
         let state = self.state.read().await;
@@ -480,6 +509,15 @@ impl ResumeCheckpoint {
         self.skipped_objects.clear();
         self.failed_objects.clear();
     }
+
+    /// Reset the scan to the start and clear the per-object sets so a retry
+    /// re-scans the whole set.
+    pub fn reset_for_retry(&mut self) {
+        self.update_position(0, 0);
+        self.processed_objects.clear();
+        self.skipped_objects.clear();
+        self.failed_objects.clear();
+    }
 }
 
 /// resume checkpoint manager
@@ -561,6 +599,14 @@ impl CheckpointManager {
         self.save_checkpoint_throttled().await
     }
 
+    /// Reset the checkpoint to the start of the scan for a retry, then persist.
+    pub async fn reset_for_retry(&self) -> Result<()> {
+        let mut checkpoint = self.checkpoint.write().await;
+        checkpoint.reset_for_retry();
+        drop(checkpoint);
+        self.save_checkpoint_throttled().await
+    }
+
     /// Add a processed object. Called once per healed object, so persistence
     /// is batched (`PERSIST_EVERY_MUTATIONS` / `PERSIST_INTERVAL`); positions
     /// and page boundaries still persist unconditionally.
@@ -808,6 +854,56 @@ mod tests {
         assert_eq!(progress, 10.0);
     }
 
+    #[test]
+    fn reset_for_retry_clears_progress_but_keeps_retry_budget() {
+        // backlog#855 / #799 B6: a retry must re-scan from the start without
+        // spending the retry budget's identity.
+        let buckets = vec!["bucket1".to_string(), "bucket2".to_string()];
+        let mut state = ResumeState::new("t".to_string(), "erasure_set".to_string(), "pool_0_set_0".to_string(), buckets);
+        state.update_progress(10, 8, 2, 0);
+        state.complete_bucket("bucket1");
+        state.increment_retry();
+        state.mark_completed();
+
+        state.reset_for_retry();
+
+        assert!(!state.completed, "retry must un-complete the task");
+        assert_eq!(state.completed_buckets.len(), 0, "all buckets must be re-scanned");
+        assert_eq!(state.processed_objects, 0);
+        assert_eq!(state.successful_objects, 0);
+        assert_eq!(state.failed_objects, 0);
+        assert_eq!(state.skipped_objects, 0);
+        assert_eq!(state.retry_count, 1, "retry budget must be preserved");
+    }
+
+    #[test]
+    fn can_retry_is_bounded_by_max_retries() {
+        let mut state = ResumeState::new("t".to_string(), "erasure_set".to_string(), "pool_0_set_0".to_string(), vec![]);
+        assert!(state.can_retry());
+        for _ in 0..state.max_retries {
+            assert!(state.can_retry());
+            state.increment_retry();
+        }
+        assert!(!state.can_retry(), "retries must stop after max_retries");
+    }
+
+    #[test]
+    fn checkpoint_reset_for_retry_rewinds_position_and_clears_sets() {
+        let mut checkpoint = ResumeCheckpoint::new("task".to_string());
+        checkpoint.update_position(3, 42);
+        checkpoint.add_processed_object("bucket/a".to_string());
+        checkpoint.add_failed_object("bucket/b".to_string());
+        checkpoint.add_skipped_object("bucket/c".to_string());
+
+        checkpoint.reset_for_retry();
+
+        assert_eq!(checkpoint.current_bucket_index, 0);
+        assert_eq!(checkpoint.current_object_index, 0);
+        assert!(checkpoint.processed_objects.is_empty());
+        assert!(checkpoint.failed_objects.is_empty());
+        assert!(checkpoint.skipped_objects.is_empty());
+    }
+
     #[tokio::test]
     async fn test_resume_state_bucket_completion() {
         let task_id = ResumeUtils::generate_task_id();