From a159f312f04099f55548ec8b11a1088e939bcac6 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 8 Sep 2026 10:14:34 +0800 Subject: [PATCH] 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 --- crates/heal/src/heal/manager.rs | 77 ++++++++++++++++++++++++++ crates/heal/src/heal/manager/tests.rs | 42 ++++++++++++++ crates/heal/src/heal/task.rs | 4 +- rustfs/src/storage/rpc/node_service.rs | 3 + 4 files changed, 124 insertions(+), 2 deletions(-) diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index b54049a14..3cd199a9f 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -506,6 +506,25 @@ fn active_heal_for_dedup_key(active_heals: &HashMap>, 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, 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")) diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index 16e71eb79..37d5553af 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/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 = 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 = Arc::new(MockStorage); diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index bb2c659b1..42ad9eeb7 100644 --- a/crates/heal/src/heal/task.rs +++ b/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, diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 306933145..72fc9bae3 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/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),