fix(test): stabilize release CI heal fixtures (#7589)

Keep heal admission-only tests off durable root recovery so mock storage does not require local metadata disks. Align forceStart retry assertions with replacement semantics and fix the e2e metric clippy collapse.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

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::<u64>().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::<u64>().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());

crates/heal/Cargo.toml

@@ -31,6 +31,7 @@ workspace = true
 
 [features]
 default = []
+test-util = []
 hotpath = [
     "hotpath/hotpath",
     "hotpath/tokio",

crates/heal/src/heal/channel.rs

@@ -847,7 +847,7 @@ mod tests {
 
     fn create_test_heal_manager() -> Arc<HealManager> {
         let storage: Arc<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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,

crates/heal/src/heal/manager.rs

@@ -1421,6 +1421,15 @@ impl HealManager {
         storage: Arc<dyn HealStorageAPI>,
         config: Option<HealConfig>,
         workload_provider: Option<WorkloadSnapshotProviderRef>,
+    ) -> Self {
+        Self::new_with_root_recovery(storage, config, workload_provider, Arc::new(root_recovery::RootHealRecovery::default()))
+    }
+
+    fn new_with_root_recovery(
+        storage: Arc<dyn HealStorageAPI>,
+        config: Option<HealConfig>,
+        workload_provider: Option<WorkloadSnapshotProviderRef>,
+        root_recovery: Arc<root_recovery::RootHealRecovery>,
     ) -> 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<dyn HealStorageAPI>, config: Option<HealConfig>) -> 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<dyn HealStorageAPI>,
+        config: Option<HealConfig>,
+        workload_provider: Option<WorkloadSnapshotProviderRef>,
+    ) -> 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;

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<Vec<DiskStore>>,
 }
@@ -391,11 +393,26 @@ impl RootHealRecovery {
     pub(super) fn with_disks(disks: Vec<DiskStore>) -> 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<Vec<DiskStore>> {
+        #[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<bool> {
+        #[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<Option<CompletedHealStatus>> {
+        #[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<bool> {
+        #[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<RootTerminalGcReport> {
+        #[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<Vec<HealRequest>> {
+        #[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();

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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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;

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,

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

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");