diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index 4563e7ff7..e69576ba0 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/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!( diff --git a/crates/heal/src/heal/resume.rs b/crates/heal/src/heal/resume.rs index 3edeefeda..757cf1378 100644 --- a/crates/heal/src/heal/resume.rs +++ b/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 { + 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();