fix(heal): reuse same admission request id (#7451)

Keep a retried heal start with the same request id from bypassing admission deduplication when the transport replay cache is unavailable but the manager still owns the task.

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

crates/heal/src/heal/manager.rs

@@ -506,6 +506,25 @@ fn active_heal_for_dedup_key(active_heals: &HashMap<String, Arc<HealTask>>, key:
         .map(|(task_id, task)| (task_id.clone(), task.heal_type.clone()))
 }
 
+fn request_matches_task(request: &HealRequest, task: &HealTask) -> bool {
+    request.heal_type == task.heal_type
+        && request.options == task.options
+        && request.priority == task.priority
+        && request.source == task.source
+        && request.retry_attempts == task.retry_attempts
+        && request.heal_endpoints == task.heal_endpoints
+}
+
+fn request_matches_request(request: &HealRequest, existing: &HealRequest) -> bool {
+    request.heal_type == existing.heal_type
+        && request.options == existing.options
+        && request.priority == existing.priority
+        && request.source == existing.source
+        && request.force_start == existing.force_start
+        && request.retry_attempts == existing.retry_attempts
+        && request.heal_endpoints == existing.heal_endpoints
+}
+
 fn retrying_heal_for_dedup_key(retrying_heals: &HashMap<String, RetryingHeal>, key: &str) -> Option<(String, HealType)> {
     retrying_heals
         .iter()
@@ -1613,6 +1632,64 @@ impl HealManager {
         pause_duplicate_admission_after_active_lock(&request.id).await;
         let mut queue = self.heal_queue.lock().await;
         let retrying_heals = self.retrying_heals.lock().await;
+
+        let request_id_admission = active_heals
+            .get(&request.id)
+            .map(|task| (request_matches_task(&request, task), "active"))
+            .or_else(|| {
+                queue
+                    .requests()
+                    .find(|queued| queued.id == request.id)
+                    .map(|queued| (request_matches_request(&request, queued), "queued"))
+            })
+            .or_else(|| {
+                retrying_heals
+                    .get(&request.id)
+                    .map(|retrying| (request_matches_request(&request, &retrying.request), "retrying"))
+            });
+        if let Some((matches_existing, duplicate_state)) = request_id_admission {
+            let admission = if matches_existing {
+                HealAdmissionResult::Accepted
+            } else {
+                HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
+            };
+            if matches!(admission, HealAdmissionResult::Accepted | HealAdmissionResult::Merged)
+                && let Some(target) = mrf_notice_target
+            {
+                let mut targets = lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets);
+                Self::insert_mrf_repair_notice_target(&mut targets, &request.id, target);
+            }
+            drop(retrying_heals);
+            drop(queue);
+            drop(active_heals);
+            let lock_phase = lock_phase_start.elapsed();
+            Self::record_admission_metric(request.source, admission, "duplicate");
+            self.record_admission_observation(HealAdmissionObservation {
+                source,
+                result: admission,
+                context: "duplicate",
+                force_start,
+                displaced: false,
+                start_duration: admission_start.elapsed(),
+                lock_phase,
+            });
+            debug!(
+                target: "rustfs::heal::manager",
+                event = EVENT_HEAL_QUEUE_ADMISSION,
+                component = LOG_COMPONENT_HEAL,
+                subsystem = LOG_SUBSYSTEM_MANAGER,
+                request_id = %request.id,
+                duplicate_state,
+                result = admission.result_label(),
+                reason = admission.reason_label(),
+                "Heal queue admission reused an existing request id"
+            );
+            return Ok(HealAdmissionReceipt {
+                result: admission,
+                task_id: request.id,
+            });
+        }
+
         let duplicate = (!request.force_start).then(|| {
             active_heal_for_dedup_key(&active_heals, &dedup_key)
                 .map(|(task_id, _)| (task_id, "active"))

crates/heal/src/heal/manager/tests.rs

@@ -4537,6 +4537,48 @@ 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 mut original = admin_prefix_request("bucket", "logs/");
+    original.force_start = true;
+    let original_id = original.id.clone();
+    let accepted = manager
+        .submit_heal_request_with_receipt(original.clone())
+        .await
+        .expect("original forceStart request should queue");
+    assert_eq!(accepted.result, HealAdmissionResult::Accepted);
+    assert_eq!(accepted.task_id, original_id);
+
+    let replayed = manager
+        .submit_heal_request_with_receipt(original.clone())
+        .await
+        .expect("same request id and payload should reuse the existing task");
+    assert_eq!(replayed.result, HealAdmissionResult::Accepted);
+    assert_eq!(replayed.task_id, original_id);
+    assert_eq!(
+        manager.get_queue_length().await,
+        1,
+        "exact forceStart replay must not create a second queued task"
+    );
+
+    let mut changed = original;
+    changed.options.remove_corrupted = true;
+    let changed = manager
+        .submit_heal_request_with_receipt(changed)
+        .await
+        .expect("same request id with a changed payload should fail closed");
+    assert_eq!(changed.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning));
+    assert_eq!(changed.task_id, original_id);
+    assert_eq!(
+        manager.get_queue_length().await,
+        1,
+        "same-id conflict must not displace or duplicate the original task"
+    );
+}
+
 #[test]
 fn test_running_heal_set_counts_groups_set_scoped_tasks() {
     let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);

crates/heal/src/heal/task.rs

@@ -106,7 +106,7 @@ const EVENT_HEAL_ERASURE_SET_STAGE: &str = "heal_erasure_set_stage";
 const EVENT_HEAL_ERASURE_SET_RESULT: &str = "heal_erasure_set_result";
 
 /// Heal type
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub enum HealType {
     /// Cluster heal
     Cluster,
@@ -209,7 +209,7 @@ impl HealPriority {
 }
 
 /// Heal options
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct HealOptions {
     /// Scan mode
     pub scan_mode: HealScanMode,

rustfs/src/storage/rpc/node_service.rs

@@ -3475,6 +3475,9 @@ mod tests {
             1,
             "post-admission response loss must leave exactly one canonical task"
         );
+        if let Some(cache) = super::HEAL_CONTROL_REPLAY_CACHE.get() {
+            cache.lock().await.clear();
+        }
 
         let mut retry = connect_faulty_heal_control_client(
             Arc::clone(&manager),