diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index e256b1c18..b643261cb 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -65,6 +65,18 @@ fn durable_replacement_recovery_is_due(state: &ResumeState, task_id: &str) -> bo && matches!(state.replacement_phase, ReplacementPhase::Verified | ReplacementPhase::CleanupPending))) } +fn unblock_replacement_recovery_sets_after_validation( + blocked_sets: &mut HashSet, + retry_succeeded: HashSet, + retry_failed: &HashSet, +) { + for set_disk_id in retry_succeeded { + if !retry_failed.contains(&set_disk_id) { + blocked_sets.remove(&set_disk_id); + } + } +} + // Admission/scheduler outcomes for per-object requests (Object/Metadata/MRF/ // ECDecode) log via demote_to_debug_when! — MRF, autoheal, and scanner // recovery loops submit those per object, so a full queue or a retry storm @@ -2532,11 +2544,7 @@ impl HealManager { let mut blocked = replacement_recovery_blocked_sets .lock() .expect("replacement recovery blocked set lock poisoned"); - for set_disk_id in retry_succeeded { - if !retry_failed.contains(&set_disk_id) { - blocked.remove(&set_disk_id); - } - } + unblock_replacement_recovery_sets_after_validation(&mut blocked, retry_succeeded, &retry_failed); } for disk in &local_disks { let endpoint = disk.endpoint(); @@ -3491,11 +3499,13 @@ fn can_schedule_request(request: &HealRequest, running_per_set: &HashMap Result> { + if let Some(hook) = manager_recovery_test_hook() { + *hook.listed.lock().expect("manager recovery listed lock should not poison") = true; + } Ok(Vec::new()) } @@ -3638,6 +3651,12 @@ mod tests { _version_id: Option<&str>, _opts: &HealOpts, ) -> Result<(HealResultItem, Option)> { + if let Some(hook) = manager_recovery_test_hook() { + *hook + .heal_object_calls + .lock() + .expect("manager recovery object call lock should not poison") += 1; + } if bucket == "retry-transition" { return Ok(( HealResultItem::default(), @@ -3651,10 +3670,38 @@ mod tests { } async fn heal_bucket(&self, _bucket: &str, _opts: &HealOpts) -> Result { + if let Some(hook) = manager_recovery_test_hook() { + *hook + .bucket_heal_calls + .lock() + .expect("manager recovery bucket call lock should not poison") += 1; + } Ok(HealResultItem::default()) } async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option)> { + if let Some(hook) = manager_recovery_test_hook() { + *hook + .global_format_calls + .lock() + .expect("manager recovery global format call lock should not poison") += 1; + } + Ok((HealResultItem::default(), None)) + } + + async fn heal_replacement_format( + &self, + _dry_run: bool, + _pool_index: usize, + _set_index: usize, + _targets: &[String], + ) -> Result<(HealResultItem, Option)> { + if let Some(hook) = manager_recovery_test_hook() { + *hook + .replacement_format_calls + .lock() + .expect("manager recovery replacement format call lock should not poison") += 1; + } Ok((HealResultItem::default(), None)) } @@ -3674,6 +3721,89 @@ mod tests { async fn get_disk_for_resume(&self, _set_disk_id: &str) -> Result { Err(Error::other("not implemented in tests")) } + + async fn get_replacement_resume_disk( + &self, + _set_disk_id: &str, + _task_id: &str, + _excluded_targets: &[String], + ) -> Result { + let Some(hook) = manager_recovery_test_hook() else { + return Err(Error::other("not implemented in tests")); + }; + Ok(crate::heal::storage::ReplacementResumeDisk::Existing( + hook.replacement_resume_disk.clone(), + )) + } + } + + struct ManagerRecoveryTestHook { + replacement_resume_disk: DiskStore, + listed: StdMutex, + global_format_calls: StdMutex, + replacement_format_calls: StdMutex, + bucket_heal_calls: StdMutex, + heal_object_calls: StdMutex, + } + + static MANAGER_RECOVERY_TEST_HOOK: LazyLock>>> = + LazyLock::new(|| StdMutex::new(None)); + + struct ManagerRecoveryTestHookGuard; + + impl ManagerRecoveryTestHook { + fn install(replacement_resume_disk: DiskStore) -> (Arc, ManagerRecoveryTestHookGuard) { + let hook = Arc::new(Self { + replacement_resume_disk, + listed: StdMutex::new(false), + global_format_calls: StdMutex::new(0), + replacement_format_calls: StdMutex::new(0), + bucket_heal_calls: StdMutex::new(0), + heal_object_calls: StdMutex::new(0), + }); + let previous = MANAGER_RECOVERY_TEST_HOOK + .lock() + .expect("manager recovery hook lock should not poison") + .replace(hook.clone()); + assert!(previous.is_none(), "manager recovery hook already installed"); + (hook, ManagerRecoveryTestHookGuard) + } + } + + impl Drop for ManagerRecoveryTestHookGuard { + fn drop(&mut self) { + *MANAGER_RECOVERY_TEST_HOOK + .lock() + .expect("manager recovery hook lock should not poison") = None; + } + } + + fn manager_recovery_test_hook() -> Option> { + MANAGER_RECOVERY_TEST_HOOK + .lock() + .expect("manager recovery hook lock should not poison") + .clone() + } + + async fn make_manager_resume_disk(temp: &TempDir, name: &str) -> DiskStore { + let disk_path = temp.path().join(name); + std::fs::create_dir_all(&disk_path).expect("manager recovery disk directory should be created"); + let endpoint = Endpoint::try_from(disk_path.to_string_lossy().as_ref()).expect("manager recovery endpoint should parse"); + let disk = new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("manager recovery disk should initialize"); + let metadata_volume = disk.make_volume(super::super::RUSTFS_META_BUCKET).await; + assert!( + matches!(metadata_volume, Ok(()) | Err(DiskError::VolumeExists)), + "manager recovery metadata volume should exist: {metadata_volume:?}" + ); + disk } fn bucket_request(bucket: &str, priority: HealPriority, source: HealRequestSource) -> HealRequest { @@ -4525,6 +4655,143 @@ mod tests { )); } + #[test] + fn replacement_recovery_retry_barrier_requires_all_set_records_to_validate() { + let mut blocked = HashSet::from(["pool_0_set_0".to_string(), "pool_0_set_1".to_string()]); + let retry_succeeded = HashSet::from(["pool_0_set_0".to_string(), "pool_0_set_1".to_string()]); + let retry_failed = HashSet::from(["pool_0_set_0".to_string()]); + + unblock_replacement_recovery_sets_after_validation(&mut blocked, retry_succeeded, &retry_failed); + + assert!( + blocked.contains("pool_0_set_0"), + "one failed disk record must keep the whole replacement set blocked" + ); + assert!( + !blocked.contains("pool_0_set_1"), + "a blocked set may resume only after every retried record validates" + ); + } + + #[tokio::test] + async fn scheduler_completes_cleanup_pending_recovery_from_manager_anchor() { + let temp = TempDir::new().expect("temporary manager recovery directory should be created"); + let anchor = make_manager_resume_disk(&temp, "anchor").await; + let task_id = ResumeUtils::generate_task_id(); + let target = "replacement-a".to_string(); + let identity = ReplacementTargetIdentity { + endpoint: target.clone(), + canonical_path: "/replacement/replacement-a".to_string(), + physical_device_ids: vec!["replacement-a".to_string()], + filesystem_identity: "identity-replacement-a".to_string(), + }; + let resume_manager = ResumeManager::new_replacement_intent( + anchor.clone(), + task_id.clone(), + "pool_0_set_0".to_string(), + vec!["bucket-a".to_string()], + vec![target.clone()], + vec![identity], + ) + .await + .expect("cleanup-pending replacement state should persist on the survivor anchor"); + resume_manager + .mark_replacement_completed_and_verified() + .await + .expect("completion proof should persist before cleanup"); + resume_manager + .mark_replacement_cleanup_pending() + .await + .expect("cleanup-pending state should persist before restart"); + CheckpointManager::new(anchor.clone(), task_id.clone()) + .await + .expect("checkpoint fixture should persist"); + + let (hook, _hook_guard) = ManagerRecoveryTestHook::install(anchor.clone()); + let storage = Arc::new(MockStorage); + let manager = HealManager::new(storage.clone(), None); + let mut request = HealRequest::new( + HealType::ErasureSet { + buckets: vec!["bucket-a".to_string()], + set_disk_id: "pool_0_set_0".to_string(), + }, + HealOptions { + pool_index: Some(0), + set_index: Some(0), + ..HealOptions::default() + }, + HealPriority::Low, + ); + request.id = task_id.clone(); + request.source = HealRequestSource::AutoHeal; + request.heal_endpoints = vec![target]; + assert_eq!( + manager + .submit_heal_request(request) + .await + .expect("durable recovery request should be admitted"), + HealAdmissionResult::Accepted + ); + manager + .replacement_recovery_anchors + .lock() + .expect("replacement recovery anchor lock should not poison") + .insert(task_id.clone(), anchor.endpoint().to_string()); + + process_manager_queue_once(&manager).await; + tokio::time::timeout(Duration::from_secs(2), async { + loop { + let resume_removed = !ResumeManager::has_resume_state(&anchor, &task_id).await; + let checkpoint_removed = !CheckpointManager::has_checkpoint(&anchor, &task_id).await; + let anchor_removed = !manager + .replacement_recovery_anchors + .lock() + .expect("replacement recovery anchor lock should not poison") + .contains_key(&task_id); + if resume_removed && checkpoint_removed && anchor_removed { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("cleanup-pending recovery should finish through the manager scheduler"); + + assert!( + !*hook.listed.lock().expect("manager recovery listed lock should not poison"), + "cleanup-pending recovery must not list buckets or restart object healing" + ); + assert_eq!( + *hook + .global_format_calls + .lock() + .expect("manager recovery global format call lock should not poison"), + 0 + ); + assert_eq!( + *hook + .replacement_format_calls + .lock() + .expect("manager recovery replacement format call lock should not poison"), + 0, + "manager-resumed terminal cleanup must not format replacement targets" + ); + assert_eq!( + *hook + .bucket_heal_calls + .lock() + .expect("manager recovery bucket call lock should not poison"), + 0 + ); + assert_eq!( + *hook + .heal_object_calls + .lock() + .expect("manager recovery object call lock should not poison"), + 0 + ); + } + #[test] fn test_retry_request_for_scoped_slowdown_preserves_scope() { let storage: Arc = Arc::new(MockStorage); diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index ddf83646e..6937f69e7 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -2997,6 +2997,147 @@ mod tests { assert!(!*storage.listed.lock().unwrap()); } + #[tokio::test] + async fn cleanup_pending_recovery_removes_checkpoint_without_rebuild_work() { + let temp = TempDir::new().expect("temporary resume disk directory should be created"); + let anchor = make_resume_disk(&temp).await; + let task_id = crate::heal::resume::ResumeUtils::generate_task_id(); + let identity = replacement_identity("replacement-a", "device-a", "filesystem-a"); + let resume_manager = ResumeManager::new_replacement_intent( + anchor.clone(), + task_id.clone(), + "pool_0_set_0".to_string(), + vec!["bucket-a".to_string()], + vec!["replacement-a".to_string()], + vec![identity], + ) + .await + .expect("terminal replacement state should persist on the survivor anchor"); + resume_manager + .mark_replacement_completed_and_verified() + .await + .expect("terminal replacement proof should persist before cleanup"); + resume_manager + .mark_replacement_cleanup_pending() + .await + .expect("failed cleanup must retain a cleanup-pending state"); + CheckpointManager::new(anchor.clone(), task_id.clone()) + .await + .expect("checkpoint fixture should persist"); + assert!( + CheckpointManager::has_checkpoint(&anchor, &task_id).await, + "checkpoint fixture must exist before restart cleanup" + ); + + let storage = Arc::new(MockStorage { + replacement_resume_disk: Mutex::new(Some(anchor.clone())), + ..Default::default() + }); + let mut request = HealRequest::new( + HealType::ErasureSet { + buckets: vec!["bucket-a".to_string()], + set_disk_id: "pool_0_set_0".to_string(), + }, + HealOptions { + pool_index: Some(0), + set_index: Some(0), + ..HealOptions::default() + }, + HealPriority::Low, + ); + request.id = task_id.clone(); + request.source = HealRequestSource::AutoHeal; + request.heal_endpoints = vec!["replacement-a".to_string()]; + + HealTask::from_replacement_recovery_request(request, storage.clone(), Some(anchor.endpoint().to_string())) + .execute() + .await + .expect("cleanup-pending recovery must finish terminal cleanup"); + + assert!( + !CheckpointManager::has_checkpoint(&anchor, &task_id).await, + "terminal cleanup must remove the retained checkpoint" + ); + assert!( + !ResumeManager::has_resume_state(&anchor, &task_id).await, + "terminal cleanup must remove the retained resume state" + ); + assert_eq!(*storage.global_format_calls.lock().unwrap(), 0); + assert!( + storage.replacement_format_calls.lock().unwrap().is_empty(), + "terminal checkpoint cleanup must not format replacement targets" + ); + assert!(storage.bucket_heal_calls.lock().unwrap().is_empty()); + assert!(storage.heal_object_calls.lock().unwrap().is_empty()); + assert!(!*storage.listed.lock().unwrap()); + } + + #[tokio::test] + async fn verified_recovery_keeps_state_when_marker_clear_fails() { + let temp = TempDir::new().expect("temporary resume disk directory should be created"); + let anchor = make_resume_disk(&temp).await; + let task_id = crate::heal::resume::ResumeUtils::generate_task_id(); + let target = format!("replacement-marker-missing-{task_id}"); + let identity = replacement_identity(&target, &target, &format!("identity-{target}")); + let resume_manager = ResumeManager::new_replacement_intent( + anchor.clone(), + task_id.clone(), + "pool_0_set_0".to_string(), + vec!["bucket-a".to_string()], + vec![target.clone()], + vec![identity], + ) + .await + .expect("verified replacement state should persist on the survivor anchor"); + resume_manager + .mark_replacement_completed_and_verified() + .await + .expect("verified state must persist proof before marker cleanup"); + + let storage = Arc::new(MockStorage { + replacement_resume_disk: Mutex::new(Some(anchor.clone())), + replacement_targets_ready: Mutex::new(true), + ..Default::default() + }); + let mut request = HealRequest::new( + HealType::ErasureSet { + buckets: vec!["bucket-a".to_string()], + set_disk_id: "pool_0_set_0".to_string(), + }, + HealOptions { + pool_index: Some(0), + set_index: Some(0), + ..HealOptions::default() + }, + HealPriority::Low, + ); + request.id = task_id.clone(); + request.source = HealRequestSource::AutoHeal; + request.heal_endpoints = vec![target]; + + let error = HealTask::from_replacement_recovery_request(request, storage.clone(), Some(anchor.endpoint().to_string())) + .execute() + .await + .expect_err("marker clear failure must keep the durable terminal state retryable"); + + assert!(error.to_string().contains("healing marker target is unavailable")); + let state = ResumeManager::load_replacement_intent(anchor.clone(), &task_id) + .await + .expect("verified state must remain for retry after marker clear failure") + .get_state() + .await; + assert!(state.completed); + assert_eq!(state.replacement_phase, ReplacementPhase::Verified); + assert_eq!(*storage.global_format_calls.lock().unwrap(), 0); + assert!( + storage.replacement_format_calls.lock().unwrap().is_empty(), + "marker cleanup retry must not format replacement targets again" + ); + assert!(storage.bucket_heal_calls.lock().unwrap().is_empty()); + assert!(storage.heal_object_calls.lock().unwrap().is_empty()); + assert!(!*storage.listed.lock().unwrap()); + } + #[derive(Default)] struct MockStorage { listed: Mutex,