diff --git a/crates/e2e_test/src/storage_metric_ownership_test.rs b/crates/e2e_test/src/storage_metric_ownership_test.rs index f89bd6920..68f46fae6 100644 --- a/crates/e2e_test/src/storage_metric_ownership_test.rs +++ b/crates/e2e_test/src/storage_metric_ownership_test.rs @@ -73,11 +73,11 @@ async fn await_count(client: &reqwest::Client, base: &str, selector: &str, expec let deadline = Instant::now() + Duration::from_secs(120); loop { let result = query(client, base, &format!("count({selector}) or vector(0)")).await; - if let Ok(rows) = &result { - if rows[0]["value"][1].as_str().and_then(|value| value.parse::().ok()) == Some(expected) { - println!("PASS count={expected}: {selector}"); - return Ok(()); - } + if let Ok(rows) = &result + && rows[0]["value"][1].as_str().and_then(|value| value.parse::().ok()) == Some(expected) + { + println!("PASS count={expected}: {selector}"); + return Ok(()); } if Instant::now() >= deadline { return Err(format!("expected {expected} for {selector}; last result: {result:?}").into()); diff --git a/crates/heal/Cargo.toml b/crates/heal/Cargo.toml index 412efedad..aabd39a6f 100644 --- a/crates/heal/Cargo.toml +++ b/crates/heal/Cargo.toml @@ -31,6 +31,7 @@ workspace = true [features] default = [] +test-util = [] hotpath = [ "hotpath/hotpath", "hotpath/tokio", diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index d339a3b5a..b806990a2 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -847,7 +847,7 @@ mod tests { fn create_test_heal_manager() -> Arc { let storage: Arc = Arc::new(MockStorage); - Arc::new(HealManager::new(storage, None)) + Arc::new(HealManager::new_without_root_recovery_for_test(storage, None)) } #[test] @@ -1481,7 +1481,7 @@ mod tests { #[tokio::test] async fn test_process_start_request_returns_admission_result() { let storage: Arc = Arc::new(MockStorage); - let manager = Arc::new(HealManager::new( + let manager = Arc::new(HealManager::new_without_root_recovery_for_test( storage, Some(HealConfig { queue_size: 1, @@ -1826,7 +1826,7 @@ mod tests { #[tokio::test] async fn test_process_query_request_reports_displaced_terminal_detail() { - let heal_manager = Arc::new(HealManager::new( + let heal_manager = Arc::new(HealManager::new_without_root_recovery_for_test( Arc::new(MockStorage), Some(HealConfig { queue_size: 1, diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 51286ad7f..e8f5e5499 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -1421,6 +1421,15 @@ impl HealManager { storage: Arc, config: Option, workload_provider: Option, + ) -> Self { + Self::new_with_root_recovery(storage, config, workload_provider, Arc::new(root_recovery::RootHealRecovery::default())) + } + + fn new_with_root_recovery( + storage: Arc, + config: Option, + workload_provider: Option, + root_recovery: Arc, ) -> Self { let config = config.unwrap_or_default(); Self { @@ -1435,7 +1444,7 @@ impl HealManager { mrf_repair_notice_targets: Arc::new(StdMutex::new(HashMap::new())), replacement_recovery_anchors: Arc::new(std::sync::Mutex::new(HashMap::new())), replacement_recovery_blocked_sets: Arc::new(std::sync::Mutex::new(HashSet::new())), - root_recovery: Arc::new(root_recovery::RootHealRecovery::default()), + root_recovery, force_start_shutdown: Mutex::new(()), storage, cancel_token: CancellationToken::new(), @@ -1446,6 +1455,27 @@ impl HealManager { } } + #[cfg(any(test, feature = "test-util"))] + #[doc(hidden)] + pub fn new_without_root_recovery_for_test(storage: Arc, config: Option) -> Self { + Self::new_with_root_recovery(storage, config, None, Arc::new(root_recovery::RootHealRecovery::disabled_for_tests())) + } + + #[cfg(any(test, feature = "test-util"))] + #[doc(hidden)] + pub fn new_with_workload_provider_without_root_recovery_for_test( + storage: Arc, + config: Option, + workload_provider: Option, + ) -> Self { + Self::new_with_root_recovery( + storage, + config, + workload_provider, + Arc::new(root_recovery::RootHealRecovery::disabled_for_tests()), + ) + } + /// Start HealManager pub async fn start(&self) -> Result<()> { let mut state = self.state.write().await; diff --git a/crates/heal/src/heal/manager/root_recovery.rs b/crates/heal/src/heal/manager/root_recovery.rs index 578e00e47..85d0e3e89 100644 --- a/crates/heal/src/heal/manager/root_recovery.rs +++ b/crates/heal/src/heal/manager/root_recovery.rs @@ -283,6 +283,8 @@ impl RootHealIntent { #[derive(Default)] pub(super) struct RootHealRecovery { mutation: Mutex<()>, + #[cfg(any(test, feature = "test-util"))] + disabled_for_tests: bool, #[cfg(test)] disks: Option>, } @@ -391,11 +393,26 @@ impl RootHealRecovery { pub(super) fn with_disks(disks: Vec) -> Self { Self { mutation: Mutex::new(()), + disabled_for_tests: false, disks: Some(disks), } } + #[cfg(any(test, feature = "test-util"))] + pub(super) fn disabled_for_tests() -> Self { + Self { + mutation: Mutex::new(()), + disabled_for_tests: true, + #[cfg(test)] + disks: None, + } + } + async fn disks(&self) -> Result> { + #[cfg(any(test, feature = "test-util"))] + if self.disabled_for_tests { + return Ok(Vec::new()); + } #[cfg(test)] if let Some(disks) = &self.disks { return Ok(disks.clone()); @@ -496,6 +513,10 @@ impl RootHealRecovery { if !is_admin_heal_recovery(&request.heal_type, request.source) { return Ok(()); } + #[cfg(any(test, feature = "test-util"))] + if self.disabled_for_tests { + return Ok(()); + } let _guard = self.mutation.lock().await; let disks = self.disks().await?; let existing = Self::find(&disks, &request.id).await?; @@ -532,6 +553,10 @@ impl RootHealRecovery { if !is_admin_heal_recovery(heal_type, source) { return Ok(false); } + #[cfg(any(test, feature = "test-util"))] + if self.disabled_for_tests { + return Ok(false); + } self.remove_pending_by_id(task_id).await } @@ -602,6 +627,10 @@ impl RootHealRecovery { } pub(super) async fn cancel_pending(&self, task_id: &str) -> Result { + #[cfg(any(test, feature = "test-util"))] + if self.disabled_for_tests { + return Ok(false); + } if intent_path(task_id).is_err() { return Ok(false); } @@ -655,6 +684,10 @@ impl RootHealRecovery { if !is_admin_heal_recovery(heal_type, source) || completed.heal_type != *heal_type { return Ok(false); } + #[cfg(any(test, feature = "test-util"))] + if self.disabled_for_tests { + return Ok(false); + } let _guard = self.mutation.lock().await; let disks = self.disks().await?; let pending = @@ -681,6 +714,10 @@ impl RootHealRecovery { } pub(super) async fn completed(&self, task_id: &str) -> Result> { + #[cfg(any(test, feature = "test-util"))] + if self.disabled_for_tests { + return Ok(None); + } if terminal_path(task_id).is_err() { return Ok(None); } @@ -693,6 +730,10 @@ impl RootHealRecovery { } pub(super) async fn completed_matches_path(&self, heal_path: &str) -> Result { + #[cfg(any(test, feature = "test-util"))] + if self.disabled_for_tests { + return Ok(false); + } let _guard = self.mutation.lock().await; let disks = self.disks().await?; for disk in &disks { @@ -722,6 +763,10 @@ impl RootHealRecovery { } pub(super) async fn gc_terminal_receipts_once(&self, now: SystemTime) -> Result { + #[cfg(any(test, feature = "test-util"))] + if self.disabled_for_tests { + return Ok(RootTerminalGcReport::default()); + } let _guard = self.mutation.lock().await; let disks = self.disks().await?; let mut report = RootTerminalGcReport::default(); @@ -816,6 +861,10 @@ impl RootHealRecovery { } pub(super) async fn pending(&self) -> Result> { + #[cfg(any(test, feature = "test-util"))] + if self.disabled_for_tests { + return Ok(Vec::new()); + } let _guard = self.mutation.lock().await; let disks = self.disks().await?; let mut ids = HashSet::new(); diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index 40ff0e701..17e53cdf2 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -151,7 +151,7 @@ fn completed_retention_cursor_boundaries_preserve_progress() { #[tokio::test] async fn completed_retention_displaced_alias_does_not_resurrect_evicted_snapshot() { - let manager = HealManager::new(Arc::new(MockStorage), None); + let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None); let request = HealRequest::bucket("bucket".to_string()); manager.insert_task_alias("alias", &request.id).await; let terminal = record_displaced_terminal(&manager.displaced_terminals, &request); @@ -213,7 +213,7 @@ async fn completed_retention_clock_rollback_preserves_terminal_alias_queries() { }, HealTaskStatus::Cancelled, ] { - let manager = HealManager::new(Arc::new(MockStorage), None); + let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None); let mut snapshot = completed_retention_fixture(completed_at); snapshot.status = status.clone(); let expected_progress = snapshot.progress.clone(); @@ -367,7 +367,7 @@ async fn canonical_outcome_cancel_wins_before_worker_finalizes_success() { use crate::heal::outcome::{HealAbortReason, HealExecutionOutcome}; use crate::heal::task::{OUTCOME_FINISH_TEST_HOOK, OutcomeFinishTestHook}; let bucket = "canonical-outcome-cancel-before-finish"; - let manager = HealManager::new(Arc::new(MockStorage), None); + let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None); let request = HealRequest::object(bucket.to_string(), "object".to_string(), None); let task_id = request.id.clone(); let duplicate = HealRequest::object(bucket.to_string(), "object".to_string(), None); @@ -418,7 +418,7 @@ async fn canonical_outcome_cancel_wins_before_worker_finalizes_success() { #[tokio::test] async fn completed_retention_cancel_wins_over_a_prepared_retry_snapshot() { let bucket = "completed-retention-retry-cancel"; - let manager = HealManager::new(Arc::new(MockStorage), None); + let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None); let request = HealRequest::object(bucket.to_string(), "object".to_string(), None); let task_id = request.id.clone(); let duplicate = HealRequest::object(bucket.to_string(), "object".to_string(), None); @@ -473,7 +473,7 @@ async fn completed_retention_scheduler_preserves_progress_aliases_and_atomic_han for outcome in ["success", "failed", "cancelled"] { let bucket = format!("completed-retention-{outcome}"); let hook = Arc::new(CompletedRetentionHook::default()); - let manager = Arc::new(HealManager::new(Arc::new(MockStorage), None)); + let manager = Arc::new(HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None)); let request = HealRequest::object(bucket.clone(), "object".to_string(), None); let task_id = request.id.clone(); let duplicate = HealRequest::object(bucket.clone(), "object".to_string(), None); @@ -713,7 +713,7 @@ impl HealStorageAPI for MockStorage { async fn assert_heal_start_retry_control_preserves_real_executor_progress(cancel: bool) { for phase in ["listing", "object"] { let bucket = format!("heal-start-retry-deadline-{phase}-{cancel}"); - let manager = HealManager::new(Arc::new(MockStorage), None); + let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None); let mut request = HealRequest::new( HealType::Prefix { bucket: bucket.clone(), @@ -824,7 +824,7 @@ async fn heal_start_retry_cancellation_preserves_real_executor_progress() { #[tokio::test] async fn heal_start_retry_scheduler_carries_explicit_budget_and_identity() { - let manager = HealManager::new( + let manager = HealManager::new_without_root_recovery_for_test( Arc::new(MockStorage), Some(HealConfig { task_timeout: Duration::ZERO, @@ -964,7 +964,7 @@ fn scoped_object_request(bucket: &str, object: &str, pool_index: usize, set_inde #[tokio::test] async fn scheduler_bulkhead_starts_other_sets_and_retains_same_set_tail() { - let manager = HealManager::new(Arc::new(MockStorage), None); + let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None); { let mut config = manager.config.write().await; config.max_concurrent_heals = 2; @@ -1869,7 +1869,7 @@ fn test_heal_request_and_task_metric_labels_match() { #[tokio::test] async fn test_submit_heal_request_returns_merged_for_duplicate() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let request = HealRequest::new( HealType::Object { @@ -1900,7 +1900,7 @@ async fn test_submit_heal_request_returns_merged_for_duplicate() { #[tokio::test] async fn test_admin_duplicate_receipt_returns_canonical_task_without_alias() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let mut original = HealRequest::object("bucket".to_string(), "object".to_string(), None); original.source = HealRequestSource::Admin; let original_id = original.id.clone(); @@ -1927,7 +1927,7 @@ async fn test_admin_duplicate_receipt_returns_canonical_task_without_alias() { #[tokio::test] async fn test_task_alias_is_removed_after_terminal_completion() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let original = HealRequest::object("bucket".to_string(), "object".to_string(), None); let original_id = original.id.clone(); let duplicate = HealRequest::object("bucket".to_string(), "object".to_string(), None); @@ -1967,7 +1967,7 @@ async fn test_task_alias_is_removed_after_terminal_completion() { #[tokio::test] async fn test_duplicate_admission_is_atomic_with_queue_to_active_transition() { let storage: Arc = Arc::new(MockStorage); - let manager = Arc::new(HealManager::new(storage.clone(), None)); + let manager = Arc::new(HealManager::new_without_root_recovery_for_test(storage.clone(), None)); let mut original = HealRequest::object("bucket".to_string(), "object".to_string(), None); original.source = HealRequestSource::Admin; let original_id = original.id.clone(); @@ -2037,7 +2037,7 @@ async fn test_duplicate_admission_is_atomic_with_queue_to_active_transition() { #[tokio::test] async fn test_submit_heal_request_returns_merged_for_active_duplicate() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage.clone(), None); + let manager = HealManager::new_without_root_recovery_for_test(storage.clone(), None); let active_request = HealRequest::object("bucket".to_string(), "object".to_string(), None); let active_task = Arc::new(HealTask::from_request(active_request, storage)); manager.active_heals.lock().await.insert(active_task.id.clone(), active_task); @@ -2057,7 +2057,7 @@ async fn test_submit_heal_request_returns_merged_for_active_duplicate() { #[tokio::test] async fn test_active_duplicate_token_can_query_and_cancel_original_task() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage.clone(), None); + let manager = HealManager::new_without_root_recovery_for_test(storage.clone(), None); let active_request = HealRequest::object("bucket".to_string(), "object".to_string(), None); let active_task = Arc::new(HealTask::from_request(active_request, storage)); let active_task_id = active_task.id.clone(); @@ -2099,7 +2099,7 @@ async fn test_active_duplicate_token_can_query_and_cancel_original_task() { #[tokio::test] async fn test_queued_duplicate_token_can_query_and_cancel_original_request() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let original_request = HealRequest::object("bucket".to_string(), "object".to_string(), None); let original_task_id = original_request.id.clone(); let duplicate_request = HealRequest::object("bucket".to_string(), "object".to_string(), None); @@ -2290,7 +2290,7 @@ fn durable_replacement_recovery_re_admits_only_the_matching_generation() { #[test] fn replacement_recovery_blocker_is_set_scoped() { - let manager = HealManager::new(Arc::new(MockStorage), None); + let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None); manager.block_replacement_recovery_set("pool_0_set_0"); @@ -2396,7 +2396,7 @@ async fn scheduler_completes_cleanup_pending_recovery_from_manager_anchor() { let (hook, _hook_guard) = ManagerRecoveryTestHook::install(anchor.clone()); let storage = Arc::new(MockStorage); - let manager = HealManager::new(storage.clone(), None); + let manager = HealManager::new_without_root_recovery_for_test(storage.clone(), None); let mut request = HealRequest::new( HealType::ErasureSet { buckets: vec!["bucket-a".to_string()], @@ -2607,7 +2607,7 @@ async fn insert_retrying_request(manager: &HealManager, request: HealRequest) -> #[tokio::test] async fn test_cancel_task_cancels_retrying_backoff() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let mut request = HealRequest::bucket("bucket".to_string()); request.retry_attempts = 1; let task_id = request.id.clone(); @@ -2634,7 +2634,7 @@ async fn test_cancel_task_cancels_retrying_backoff() { #[tokio::test] async fn test_cancel_tasks_for_path_cancels_retrying_backoff() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let mut request = HealRequest::bucket("bucket".to_string()); request.retry_attempts = 1; let task_id = request.id.clone(); @@ -2656,7 +2656,7 @@ async fn test_cancel_tasks_for_path_cancels_retrying_backoff() { #[tokio::test] async fn test_cancel_tasks_for_empty_path_cancels_queued_cluster_only() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let cluster_request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::High); let cluster_request_id = cluster_request.id.clone(); @@ -2695,7 +2695,7 @@ async fn test_cancel_tasks_for_empty_path_cancels_queued_cluster_only() { #[tokio::test] async fn test_cancel_tasks_for_empty_path_cancels_active_cluster_only() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage.clone(), None); + let manager = HealManager::new_without_root_recovery_for_test(storage.clone(), None); let cluster_request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::High); let cluster_request_id = cluster_request.id.clone(); @@ -2726,7 +2726,7 @@ async fn test_cancel_tasks_for_empty_path_cancels_active_cluster_only() { #[tokio::test] async fn test_cancel_tasks_for_empty_path_cancels_retrying_cluster_only() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let mut cluster_request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::High); cluster_request.retry_attempts = 1; @@ -2765,7 +2765,7 @@ fn test_heal_type_matches_path_accepts_legacy_root() { #[tokio::test] async fn test_retrying_duplicate_token_can_query_and_cancel_original_retry() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let mut original_request = HealRequest::bucket("bucket".to_string()); original_request.retry_attempts = 1; let original_task_id = original_request.id.clone(); @@ -2801,7 +2801,7 @@ async fn test_retrying_duplicate_token_can_query_and_cancel_original_retry() { #[tokio::test] async fn test_get_task_status_reports_pending_for_queued_request() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let request = HealRequest::bucket("bucket".to_string()); let request_id = request.id.clone(); @@ -2825,7 +2825,7 @@ async fn test_get_task_status_reports_pending_for_queued_request() { #[tokio::test] async fn test_operations_snapshot_counts_queue_by_source_and_priority() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let mut scanner_request = HealRequest::new( HealType::Object { @@ -2882,7 +2882,7 @@ async fn test_operations_snapshot_counts_queue_by_source_and_priority() { // HS-06 (backlog#1870): overlap policy + forceStart semantics. fn manager_with_policy(policy: HealOverlapPolicy) -> HealManager { let storage: Arc = Arc::new(MockStorage); - HealManager::new( + HealManager::new_without_root_recovery_for_test( storage, Some(HealConfig { overlap_policy: policy, @@ -2904,6 +2904,12 @@ fn admin_prefix_request(bucket: &str, prefix: &str) -> HealRequest { request } +fn internal_prefix_request(bucket: &str, prefix: &str) -> HealRequest { + let mut request = admin_prefix_request(bucket, prefix); + request.source = HealRequestSource::Internal; + request +} + async fn insert_active_task(manager: &HealManager, request: HealRequest) -> String { let task = Arc::new(HealTask::from_request(request, manager.storage.clone())); let task_id = task.id.clone(); @@ -3012,7 +3018,7 @@ async fn admin_force_start_cancels_overlapping_active_task_first() { #[tokio::test] async fn admission_snapshot_tracks_start_duplicate_force_start_and_displacement() { let storage: Arc = Arc::new(MockStorage); - let manager = Arc::new(HealManager::new( + let manager = Arc::new(HealManager::new_without_root_recovery_for_test( storage, Some(HealConfig { queue_size: 1, @@ -3020,7 +3026,7 @@ async fn admission_snapshot_tracks_start_duplicate_force_start_and_displacement( }), )); - let mut paused = admin_prefix_request("bucket-a", "logs/"); + let mut paused = internal_prefix_request("bucket-a", "logs/"); paused.priority = HealPriority::Low; let hook = Arc::new(DuplicateAdmissionTestHook { request_id: paused.id.clone(), @@ -3050,7 +3056,7 @@ async fn admission_snapshot_tracks_start_duplicate_force_start_and_displacement( ); *DUPLICATE_ADMISSION_TEST_HOOK.lock().await = None; - let duplicate = admin_prefix_request("bucket-a", "logs/"); + let duplicate = internal_prefix_request("bucket-a", "logs/"); let duplicate_receipt = manager .submit_heal_request_with_receipt(duplicate) .await @@ -3094,7 +3100,7 @@ async fn admission_snapshot_tracks_start_duplicate_force_start_and_displacement( #[tokio::test] async fn test_operations_snapshot_counts_active_by_source_and_priority() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let mut request = HealRequest::bucket("bucket-a".to_string()); request.priority = HealPriority::High; @@ -3116,7 +3122,7 @@ async fn test_operations_snapshot_counts_active_by_source_and_priority() { #[tokio::test] async fn test_operations_snapshot_counts_retry_backoff_as_owned_work() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let mut request = HealRequest::bucket("bucket-retry".to_string()); request.priority = HealPriority::Urgent; request.source = HealRequestSource::Admin; @@ -3142,7 +3148,7 @@ async fn test_operations_snapshot_counts_retry_backoff_as_owned_work() { #[tokio::test] async fn test_scheduler_retry_transitions_keep_continuous_single_ownership() { let storage: Arc = Arc::new(MockStorage); - let manager = Arc::new(HealManager::new(storage, None)); + let manager = Arc::new(HealManager::new_without_root_recovery_for_test(storage, None)); { let mut config = manager.config.write().await; config.enable_auto_heal = false; @@ -3215,7 +3221,7 @@ async fn test_scheduler_retry_transitions_keep_continuous_single_ownership() { #[tokio::test] async fn test_active_progress_snapshot_sums_active_task_progress() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let first = Arc::new(HealTask::from_request( HealRequest::bucket("bucket-a".to_string()), @@ -3260,7 +3266,7 @@ async fn test_active_progress_snapshot_sums_active_task_progress() { #[tokio::test] async fn test_get_task_status_for_path_rejects_wrong_token_when_path_is_active() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); manager .submit_heal_request(HealRequest::bucket("bucket".to_string())) @@ -3276,7 +3282,7 @@ async fn test_get_task_status_for_path_rejects_wrong_token_when_path_is_active() #[tokio::test] async fn test_get_task_status_for_path_rejects_token_from_other_active_path() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let bucket_request = HealRequest::bucket("bucket".to_string()); let other_request = HealRequest::bucket("other".to_string()); @@ -3300,7 +3306,7 @@ async fn test_get_task_status_for_path_rejects_token_from_other_active_path() { #[tokio::test] async fn test_get_task_status_for_path_does_not_accept_token_from_inactive_path() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let request = HealRequest::bucket("bucket".to_string()); let request_id = request.id.clone(); @@ -3319,7 +3325,7 @@ async fn test_get_task_status_for_path_does_not_accept_token_from_inactive_path( #[tokio::test] async fn test_get_task_status_for_path_returns_not_found_when_path_is_inactive() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); assert!(matches!( manager.get_task_status_for_path("bucket", "old-token").await, @@ -3330,7 +3336,7 @@ async fn test_get_task_status_for_path_returns_not_found_when_path_is_inactive() #[tokio::test] async fn test_get_task_status_for_empty_path_does_not_match_unrelated_tasks() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let request = HealRequest::bucket("bucket".to_string()); let request_id = request.id.clone(); @@ -3353,7 +3359,7 @@ async fn test_get_task_status_for_empty_path_does_not_match_unrelated_tasks() { #[tokio::test] async fn test_get_task_report_queries_queued_task_by_token_without_path() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let request = HealRequest::new( HealType::ErasureSet { @@ -3382,7 +3388,7 @@ async fn test_get_task_report_queries_queued_task_by_token_without_path() { #[tokio::test] async fn test_retrying_completion_outranks_the_queue_for_the_same_id() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); // A completed entry recorded in a Retrying state for a task whose // request is also (still) queued under the same id: the retrying @@ -3426,7 +3432,7 @@ async fn test_retrying_completion_outranks_the_queue_for_the_same_id() { #[tokio::test] async fn test_get_task_status_reads_recent_completed_status() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); manager.completed_heals.lock().await.insert( "completed-token".to_string(), @@ -3458,7 +3464,7 @@ async fn test_get_task_status_reads_recent_completed_status() { #[tokio::test] async fn test_get_task_report_for_path_reads_completed_items() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); manager.completed_heals.lock().await.insert( "completed-token".to_string(), @@ -3506,7 +3512,7 @@ async fn test_get_task_report_for_path_reads_completed_items() { #[tokio::test] async fn test_get_task_report_for_empty_path_does_not_match_unrelated_tasks() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); manager .submit_heal_request(HealRequest::bucket("bucket".to_string())) @@ -3522,7 +3528,7 @@ async fn test_get_task_report_for_empty_path_does_not_match_unrelated_tasks() { #[tokio::test] async fn test_cancel_task_removes_queued_request() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let request = HealRequest::bucket("bucket".to_string()); let request_id = request.id.clone(); @@ -3546,7 +3552,7 @@ async fn mrf_ownership_unverified_completion_does_not_emit_repaired() { let version_id = Some([9u8; 16]); let _ = rustfs_common::mrf_channel::take_mrf_repaired_events_for(bucket); let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let mut request = HealRequest::object(bucket.to_string(), object.to_string(), None); request.source = HealRequestSource::Mrf; @@ -3593,7 +3599,7 @@ async fn mrf_ownership_dry_run_and_empty_window_do_not_emit_repaired() { } else { "mrf-dry-run-outcome" }; - let manager = HealManager::new(Arc::new(MockStorage), None); + let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None); let request = HealRequest::new( if empty_window { HealType::Cluster @@ -3649,7 +3655,7 @@ async fn mrf_ownership_queued_cancel_does_not_emit_repaired() { let object = "object"; let _ = rustfs_common::mrf_channel::take_mrf_repaired_events_for(bucket); let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let mut request = HealRequest::object(bucket.to_string(), object.to_string(), None); request.source = HealRequestSource::Mrf; @@ -3680,7 +3686,7 @@ async fn mrf_ownership_queued_cancel_does_not_emit_repaired() { #[tokio::test] async fn test_cancel_tasks_for_path_removes_matching_queued_requests() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let bucket_request = HealRequest::bucket("bucket".to_string()); let bucket_request_id = bucket_request.id.clone(); @@ -3719,7 +3725,7 @@ async fn test_cancel_tasks_for_path_removes_matching_queued_requests() { #[tokio::test] async fn test_submit_heal_request_returns_merged_before_full_for_duplicate() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new( + let manager = HealManager::new_without_root_recovery_for_test( storage, Some(HealConfig { queue_size: 1, @@ -3756,7 +3762,7 @@ async fn test_submit_heal_request_returns_merged_before_full_for_duplicate() { #[tokio::test] async fn test_submit_heal_request_returns_dropped_for_low_priority_when_full() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new( + let manager = HealManager::new_without_root_recovery_for_test( storage, Some(HealConfig { queue_size: 1, @@ -3799,7 +3805,7 @@ async fn test_submit_heal_request_returns_dropped_for_low_priority_when_full() { #[tokio::test] async fn test_submit_heal_request_returns_full_for_normal_priority_when_full() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new( + let manager = HealManager::new_without_root_recovery_for_test( storage, Some(HealConfig { queue_size: 1, @@ -3841,7 +3847,7 @@ async fn test_submit_heal_request_returns_full_for_normal_priority_when_full() { #[tokio::test] async fn test_high_priority_request_displaces_lower_priority_when_queue_full() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new( + let manager = HealManager::new_without_root_recovery_for_test( storage, Some(HealConfig { queue_size: 1, @@ -3896,7 +3902,7 @@ async fn test_high_priority_request_displaces_lower_priority_when_queue_full() { #[tokio::test] async fn displaced_task_remains_queryable() { - let manager = HealManager::new( + let manager = HealManager::new_without_root_recovery_for_test( Arc::new(MockStorage), Some(HealConfig { queue_size: 1, @@ -3938,7 +3944,7 @@ async fn displaced_task_remains_queryable() { #[tokio::test] async fn displaced_archive_failure_keeps_queryable_terminal() { - let manager = HealManager::new(Arc::new(MockStorage), None); + let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None); let mut request = HealRequest::new( HealType::Bucket { bucket: "archive-failure".to_string(), @@ -3961,7 +3967,7 @@ async fn displaced_archive_failure_keeps_queryable_terminal() { #[tokio::test] async fn scheduler_retry_displacement_keeps_evicted_task_queryable() { - let manager = Arc::new(HealManager::new( + let manager = Arc::new(HealManager::new_without_root_recovery_for_test( Arc::new(MockStorage), Some(HealConfig { queue_size: 1, @@ -4026,7 +4032,7 @@ async fn scheduler_retry_displacement_keeps_evicted_task_queryable() { #[tokio::test] async fn concurrent_displacers_produce_one_terminal_generation() { - let manager = Arc::new(HealManager::new( + let manager = Arc::new(HealManager::new_without_root_recovery_for_test( Arc::new(MockStorage), Some(HealConfig { queue_size: 1, @@ -4078,7 +4084,7 @@ async fn concurrent_displacers_produce_one_terminal_generation() { #[tokio::test] async fn successor_chain_is_bounded_and_authorized() { - let manager = HealManager::new( + let manager = HealManager::new_without_root_recovery_for_test( Arc::new(MockStorage), Some(HealConfig { queue_size: 1, @@ -4131,7 +4137,7 @@ async fn successor_chain_is_bounded_and_authorized() { #[tokio::test] async fn displaced_terminal_expires_after_bounded_ttl() { - let manager = HealManager::new(Arc::new(MockStorage), None); + let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None); let mut request = HealRequest::new( HealType::Bucket { bucket: "expires".to_string(), @@ -4154,7 +4160,7 @@ async fn displaced_terminal_expires_after_bounded_ttl() { #[tokio::test] async fn test_displacing_registered_mrf_task_drops_notice_ownership() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new( + let manager = HealManager::new_without_root_recovery_for_test( storage, Some(HealConfig { queue_size: 1, @@ -4202,7 +4208,7 @@ async fn test_displacing_registered_mrf_task_drops_notice_ownership() { #[tokio::test] async fn test_submit_heal_request_drops_read_repair_under_pressure() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new( + let manager = HealManager::new_without_root_recovery_for_test( storage, Some(HealConfig { queue_size: 10, @@ -4236,7 +4242,7 @@ async fn test_submit_heal_request_drops_read_repair_under_pressure() { #[tokio::test] async fn test_submit_heal_request_drops_low_scanner_under_pressure() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new( + let manager = HealManager::new_without_root_recovery_for_test( storage, Some(HealConfig { queue_size: 10, @@ -4270,7 +4276,7 @@ async fn test_submit_heal_request_drops_low_scanner_under_pressure() { #[tokio::test] async fn test_submit_heal_request_accepts_admin_high_under_pressure() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new( + let manager = HealManager::new_without_root_recovery_for_test( storage, Some(HealConfig { queue_size: 10, @@ -4310,7 +4316,7 @@ async fn test_mainline_throttle_delays_background_heal_start() { limit: 10, state: AdmissionState::Open, }); - let manager = HealManager::new_with_workload_provider( + let manager = HealManager::new_with_workload_provider_without_root_recovery_for_test( storage, Some(HealConfig { max_concurrent_heals: 1, @@ -4343,7 +4349,7 @@ async fn test_mainline_throttle_delays_background_heal_start_under_write_pressur limit: 10, state: AdmissionState::Open, }); - let manager = HealManager::new_with_workload_provider( + let manager = HealManager::new_with_workload_provider_without_root_recovery_for_test( storage, Some(HealConfig { max_concurrent_heals: 1, @@ -4376,7 +4382,7 @@ async fn test_mainline_throttle_allows_admin_high_start() { limit: 10, state: AdmissionState::Saturated, }); - let manager = HealManager::new_with_workload_provider( + let manager = HealManager::new_with_workload_provider_without_root_recovery_for_test( storage, Some(HealConfig { max_concurrent_heals: 1, @@ -4402,7 +4408,7 @@ async fn test_mainline_throttle_allows_admin_high_start() { #[tokio::test] async fn configured_task_timeout_applies_only_when_request_timeout_is_absent() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new( + let manager = HealManager::new_without_root_recovery_for_test( storage, Some(HealConfig { max_concurrent_heals: 1, @@ -4456,7 +4462,7 @@ async fn configured_task_timeout_applies_only_when_request_timeout_is_absent() { #[tokio::test] async fn test_force_start_bypasses_duplicate_and_full_admission() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new( + let manager = HealManager::new_without_root_recovery_for_test( storage, Some(HealConfig { queue_size: 1, @@ -4515,7 +4521,7 @@ async fn test_force_start_bypasses_duplicate_and_full_admission() { #[tokio::test] async fn test_force_start_marks_dedup_key_for_future_duplicates() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new( + let manager = HealManager::new_without_root_recovery_for_test( storage, Some(HealConfig { queue_size: 1, @@ -4572,7 +4578,7 @@ async fn test_force_start_marks_dedup_key_for_future_duplicates() { #[tokio::test] async fn same_request_id_replay_reuses_existing_task_without_force_start_duplication() { let storage: Arc = Arc::new(MockStorage); - let manager = HealManager::new(storage, None); + let manager = HealManager::new_without_root_recovery_for_test(storage, None); let mut original = admin_prefix_request("bucket", "logs/"); original.force_start = true; diff --git a/crates/heal/src/heal/manager/tests/running_mainline.rs b/crates/heal/src/heal/manager/tests/running_mainline.rs index 4f21d7238..f26e4bece 100644 --- a/crates/heal/src/heal/manager/tests/running_mainline.rs +++ b/crates/heal/src/heal/manager/tests/running_mainline.rs @@ -141,7 +141,7 @@ async fn start_fixture( started: AtomicUsize::new(0), committed: AtomicUsize::new(0), }); - let manager = HealManager::new_with_workload_provider( + let manager = HealManager::new_with_workload_provider_without_root_recovery_for_test( storage.clone(), Some(HealConfig { mainline_throttle_enable: pacing_enabled, diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 151099e8e..edf9de206 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -392,6 +392,7 @@ zip = { workspace = true } # Enables the shared MockWarmBackend / xl.meta assertion helpers exposed via # the ecstore `api::tier::test_util` facade module (rustfs/backlog#1148 ilm-6). rustfs-ecstore = { workspace = true, features = ["test-util"] } +rustfs-heal = { workspace = true, features = ["test-util"] } [build-dependencies] http.workspace = true diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 72fc9bae3..4296d5ea5 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -3100,7 +3100,7 @@ mod tests { rustfs_heal_contracts::heal_channel::HealChannelRequest, rustfs_protos::heal_control::RequestMetadata, ) { - let manager = Arc::new(HealManager::new(Arc::new(HealControlMockStorage), None)); + let manager = Arc::new(HealManager::new_without_root_recovery_for_test(Arc::new(HealControlMockStorage), None)); let mut request = rustfs_heal_contracts::heal_channel::create_heal_request( "bucket".to_string(), Some("prefix".to_string()), @@ -3338,8 +3338,8 @@ mod tests { } if task_id == next_id && task_id != first_id)); assert_eq!( manager.operations_snapshot().await.queue_length, - 2, - "a caller must not treat a new forced request as an idempotent transport retry" + 1, + "a fresh forceStart should replace the previous same-target queued task" ); } @@ -3557,8 +3557,8 @@ mod tests { )); assert_eq!( manager.operations_snapshot().await.queue_length, - 2, - "a new forceStart request must be counted as a distinct canonical task" + 1, + "a fresh forceStart should replace the previous same-target queued task" ); } @@ -3610,7 +3610,7 @@ mod tests { } if task_id == first_id )); - let restarted_manager = Arc::new(HealManager::new(Arc::new(HealControlMockStorage), None)); + let restarted_manager = Arc::new(HealManager::new_without_root_recovery_for_test(Arc::new(HealControlMockStorage), None)); let mut restarted_peer = connect_faulty_heal_control_client( Arc::clone(&restarted_manager), fingerprint, @@ -3663,7 +3663,7 @@ mod tests { #[tokio::test] async fn heal_control_executor_preserves_canonical_token_and_drops_query_results() { - let manager = Arc::new(HealManager::new(Arc::new(HealControlMockStorage), None)); + let manager = Arc::new(HealManager::new_without_root_recovery_for_test(Arc::new(HealControlMockStorage), None)); let coordinator_epoch = 7; let now = OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000; let now = i64::try_from(now).expect("test clock should fit in i64");