diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index c5114ef4d..83c24682f 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -625,6 +625,37 @@ fn request_matches_request(request: &HealRequest, existing: &HealRequest) -> boo && request.heal_endpoints == existing.heal_endpoints } +fn lookup_request_id_admission( + request: &HealRequest, + active_heals: &HashMap>, + queue: &PriorityHealQueue, + retrying_heals: &HashMap, + durable_owners: &[HealRequest], +) -> Option<(bool, &'static str)> { + 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")) + }) + // A durable-only owner must be resumed by recovery, not overwritten + // with a new execution budget by a replay before recovery completes. + .or_else(|| { + durable_owners + .iter() + .find(|owner| owner.id == request.id) + .map(|_| (false, "durable")) + }) +} + fn retrying_heal_for_dedup_key(retrying_heals: &HashMap, key: &str) -> Option<(String, HealType)> { retrying_heals .iter() @@ -2119,28 +2150,59 @@ impl HealManager { let mut queue = self.heal_queue.lock().await; let mut 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")) - }) - // A durable-only owner must be resumed by recovery, not overwritten - // with a new execution budget by a replay before recovery completes. - .or_else(|| { - durable_owners - .iter() - .find(|owner| owner.id == request.id) - .map(|_| (false, "durable")) - }); + let mut request_id_admission = + lookup_request_id_admission(&request, &active_heals, &queue, &retrying_heals, &durable_owners); + + if request_id_admission.is_none() && source == HealRequestSource::Admin && force_start { + let overlapping = active_heals + .values() + .map(|task| (&task.id, &task.heal_type, &task.options, task.source)) + .chain( + queue + .requests() + .chain(retrying_heals.values().map(|retrying| &retrying.request)) + .chain(durable_owners.iter()) + .map(|owner| (&owner.id, &owner.heal_type, &owner.options, owner.source)), + ) + .filter(|(id, heal_type, options, source)| { + **id != request.id + && *source == HealRequestSource::Admin + && heal_scopes_overlap(&request.heal_type, &request.options, heal_type, options) + != OverlapVerdict::Disjoint + }) + .map(|(id, _, _, _)| id.clone()) + .collect::>(); + drop(retrying_heals); + drop(queue); + drop(active_heals); + active_lock_hold += active_lock_acquired.elapsed(); + for task_id in &overlapping { + match self.cancel_task(task_id).await { + Ok(_) => info!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + request_id = %request.id, + cancelled_task_id = %task_id, + result = "force_start_cancelled_overlap", + "Admin forceStart cancelled an overlapping heal task" + ), + Err(Error::TaskNotFound { .. }) => {} // The owner completed before cancellation acquired its state. + Err(err) => return Err(err), + } + } + durable_owners.retain(|owner| !overlapping.contains(&owner.id)); + let reacquire_start = Instant::now(); + active_heals = self.active_heals.lock().await; + active_lock_acquired = Instant::now(); + active_lock_wait += active_lock_acquired.duration_since(reacquire_start); + queue = self.heal_queue.lock().await; + retrying_heals = self.retrying_heals.lock().await; + // Non-admin admission can claim this id while cancellation drops the state locks. + request_id_admission = lookup_request_id_admission(&request, &active_heals, &queue, &retrying_heals, &durable_owners); + } + if let Some((matches_existing, duplicate_state)) = request_id_admission { let admission = if matches_existing { if accept_same_request_id_replay { @@ -2189,54 +2251,6 @@ impl HealManager { }); } - if source == HealRequestSource::Admin && force_start { - let overlapping = active_heals - .values() - .map(|task| (&task.id, &task.heal_type, &task.options, task.source)) - .chain( - queue - .requests() - .chain(retrying_heals.values().map(|retrying| &retrying.request)) - .chain(durable_owners.iter()) - .map(|owner| (&owner.id, &owner.heal_type, &owner.options, owner.source)), - ) - .filter(|(id, heal_type, options, source)| { - **id != request.id - && *source == HealRequestSource::Admin - && heal_scopes_overlap(&request.heal_type, &request.options, heal_type, options) - != OverlapVerdict::Disjoint - }) - .map(|(id, _, _, _)| id.clone()) - .collect::>(); - drop(retrying_heals); - drop(queue); - drop(active_heals); - active_lock_hold += active_lock_acquired.elapsed(); - for task_id in &overlapping { - match self.cancel_task(task_id).await { - Ok(_) => info!( - target: "rustfs::heal::manager", - event = EVENT_HEAL_QUEUE_ADMISSION, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_MANAGER, - request_id = %request.id, - cancelled_task_id = %task_id, - result = "force_start_cancelled_overlap", - "Admin forceStart cancelled an overlapping heal task" - ), - Err(Error::TaskNotFound { .. }) => {} // The owner completed before cancellation acquired its state. - Err(err) => return Err(err), - } - } - durable_owners.retain(|owner| !overlapping.contains(&owner.id)); - let reacquire_start = Instant::now(); - active_heals = self.active_heals.lock().await; - active_lock_acquired = Instant::now(); - active_lock_wait += active_lock_acquired.duration_since(reacquire_start); - queue = self.heal_queue.lock().await; - retrying_heals = self.retrying_heals.lock().await; - } - let rejection = if source == HealRequestSource::Admin { let live_owners = active_heals .values() @@ -2282,13 +2296,22 @@ impl HealManager { None }; + let scanner_queued_owner_required = request.source == HealRequestSource::Scanner + && mrf_notice_target.is_none() + && request.expected_mrf_bucket_incarnation_id.is_none(); let duplicate = (!request.force_start && rejection.is_none()).then(|| { active_heal_for_dedup_key(&active_heals, &dedup_key) .map(|(task_id, _)| (task_id, "active")) .or_else(|| { - queue - .queued_request_id_for_dedup_key(&dedup_key) - .map(|queued_id| (queued_id.to_string(), "queued")) + let queued_id = if scanner_queued_owner_required { + queue.scanner_queued_request_id_for_dedup_key(&dedup_key) + } else { + queue.queued_request_id_for_dedup_key(&dedup_key) + }; + queued_id.map(|queued_id| (queued_id.to_string(), "queued")).or_else(|| { + (scanner_queued_owner_required && queue.dedup_keys.contains_key(&dedup_key)) + .then(|| (request.id.clone(), "queued_owner_unknown")) + }) }) .or_else(|| retrying_heal_for_dedup_key(&retrying_heals, &dedup_key).map(|(task_id, _)| (task_id, "retrying"))) }); @@ -2296,12 +2319,12 @@ impl HealManager { // HS-06: under the minio_error overlap policy an exact duplicate // admin start reports the typed AlreadyRunning rejection instead // of the silent merge (MinIO's ErrHealAlreadyRunning). - let mut admission = if (request.source == HealRequestSource::Admin - && config.overlap_policy == HealOverlapPolicy::MinioError) + let mut admission = if duplicate_state == "queued_owner_unknown" + || (request.source == HealRequestSource::Admin && config.overlap_policy == HealOverlapPolicy::MinioError) || (duplicate_state != "queued" && mrf_notice_target.as_ref().is_some_and(|target| target.kind.is_durable())) { - // A running/retrying task may have observed the object before - // this write committed. Its receipt cannot prove the new lease. + // An unknown queued owner cannot certify a receipt. A running or + // retrying task may have observed the object before this write. HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning) } else { Self::duplicate_admission_for_request(&request, &config) @@ -2514,10 +2537,17 @@ impl HealManager { .remove(&request_id, &request_heal_type, request_source) .await?; } - if admission == HealAdmissionResult::Merged - && let Some(queued_id) = queue.queued_request_id_for_dedup_key(&dedup_key) - { - task_id = queued_id.to_owned(); + if admission == HealAdmissionResult::Merged { + let queued_id = if scanner_queued_owner_required { + queue.scanner_queued_request_id_for_dedup_key(&dedup_key) + } else { + queue.queued_request_id_for_dedup_key(&dedup_key) + }; + if let Some(queued_id) = queued_id { + task_id = queued_id.to_owned(); + } else if scanner_queued_owner_required { + return Err(Error::Other("Scanner queued owner is unavailable".to_owned())); + } } let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable; let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned); diff --git a/crates/heal/src/heal/manager/queue.rs b/crates/heal/src/heal/manager/queue.rs index f50bb49a5..2c8e9abea 100644 --- a/crates/heal/src/heal/manager/queue.rs +++ b/crates/heal/src/heal/manager/queue.rs @@ -13,15 +13,23 @@ // limitations under the License. /// The priority heal queue and its per-key dedup index. use super::*; +use std::collections::BTreeMap; -/// Per-key bookkeeping for the queued-request dedup index: how many queued -/// requests hold the key, and the id of the first request that opened it — -/// the O(1) stand-in for the former heap scan when a merge receipt needs to -/// name a queued representative. +/// Per-key bookkeeping for the queued-request dedup index. The legacy +/// representative is retained for existing non-scanner callers; scanner +/// receipts use the separately tracked queued holders. #[derive(Debug)] pub(super) struct DedupKeyEntry { pub(super) refcount: usize, pub(super) representative_request_id: String, + pub(super) scanner_queued_owners: ScannerQueuedOwners, +} + +#[derive(Debug)] +pub(super) enum ScannerQueuedOwners { + Single(u64), + // Once a key has forced duplicates, keep every live holder until it is released. + Indexed(BTreeMap), } /// Priority queue wrapper for heal requests @@ -306,11 +314,7 @@ impl PriorityHealQueue { #[cfg(test)] pub(super) fn pop_next(&mut self) -> Option { - self.heap.pop().map(|item| { - Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key); - self.remove_best_effort_count(item.request.source); - item.request - }) + self.heap.pop().map(|item| self.remove_queued_item(item)) } pub(super) fn is_empty(&self) -> bool { @@ -332,14 +336,32 @@ impl PriorityHealQueue { // also reserves the dedup key for later non-forced duplicates. The first // request that opens the key becomes the named representative for merge // receipts (taken before `request` moves into the heap). - self.dedup_keys - .entry(key.clone()) - .or_insert_with(|| DedupKeyEntry { - refcount: 0, - representative_request_id: request.id.clone(), - }) - .refcount += 1; self.sequence += 1; + if let Some(entry) = self.dedup_keys.get_mut(&key) { + match &mut entry.scanner_queued_owners { + ScannerQueuedOwners::Single(first_sequence) => { + let mut owners = BTreeMap::new(); + if entry.refcount == 1 { + owners.insert(*first_sequence, entry.representative_request_id.clone()); + } + owners.insert(self.sequence, request.id.clone()); + entry.scanner_queued_owners = ScannerQueuedOwners::Indexed(owners); + } + ScannerQueuedOwners::Indexed(owners) => { + owners.insert(self.sequence, request.id.clone()); + } + } + entry.refcount += 1; + } else { + self.dedup_keys.insert( + key.clone(), + DedupKeyEntry { + refcount: 1, + representative_request_id: request.id.clone(), + scanner_queued_owners: ScannerQueuedOwners::Single(self.sequence), + }, + ); + } let source = request.source; self.heap.push(PriorityQueueItem { priority: request.priority, @@ -402,7 +424,7 @@ impl PriorityHealQueue { self.heap = retained; let displaced = displaced.map(|item| { - Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key); + Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item); self.remove_best_effort_count(item.request.source); self.refresh_dedup_representative(&item.dedup_key); item.request @@ -446,11 +468,7 @@ impl PriorityHealQueue { #[cfg(test)] pub(super) fn pop(&mut self) -> Option { - self.heap.pop().map(|item| { - Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key); - self.remove_best_effort_count(item.request.source); - item.request - }) + self.heap.pop().map(|item| self.remove_queued_item(item)) } pub(super) fn pop_runnable_with_skips(&mut self, can_run: F, skip_label: G) -> (Option, Vec) @@ -475,14 +493,7 @@ impl PriorityHealQueue { self.restore_deferred_items(deferred); - ( - selected.map(|item| { - Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key); - self.remove_best_effort_count(item.request.source); - item.request - }), - skipped, - ) + (selected.map(|item| self.remove_queued_item(item)), skipped) } pub(super) fn pop_runnable_with_fairness(&mut self, can_run: F, skip_label: G) -> (Option, Vec) @@ -554,15 +565,19 @@ impl PriorityHealQueue { self.heap = BinaryHeap::from(items); let chosen = chosen.map(|item| { - let request = item.request; - Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key); - self.remove_best_effort_count(request.source); + let request = self.remove_queued_item(item); self.record_dispatch(&request); request }); (chosen, skipped) } + fn remove_queued_item(&mut self, item: PriorityQueueItem) -> HealRequest { + Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item); + self.remove_best_effort_count(item.request.source); + item.request + } + fn record_dispatch(&mut self, request: &HealRequest) { if is_best_effort_source(request.source) { self.mrf_normal_dispatches_since_best_effort = 0; @@ -667,10 +682,19 @@ impl PriorityHealQueue { } } - pub(super) fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap, key: &str) { - if let Some(entry) = dedup_keys.get_mut(key) { + fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap, item: &PriorityQueueItem) { + if let Some(entry) = dedup_keys.get_mut(&item.dedup_key) { + match &mut entry.scanner_queued_owners { + ScannerQueuedOwners::Single(sequence) if entry.refcount != 1 || *sequence != item.sequence => { + entry.scanner_queued_owners = ScannerQueuedOwners::Indexed(BTreeMap::new()); + } + ScannerQueuedOwners::Single(_) => {} + ScannerQueuedOwners::Indexed(owners) => { + owners.remove(&item.sequence); + } + } if entry.refcount <= 1 { - dedup_keys.remove(key); + dedup_keys.remove(&item.dedup_key); } else { entry.refcount -= 1; } @@ -747,15 +771,19 @@ impl PriorityHealQueue { self.dedup_keys.get(key).map(|entry| entry.representative_request_id.as_str()) } - /// Re-elect the representative for `key` from the queue entries holding - /// it. Needed after a holder leaves the queue *without* becoming active - /// (canceled by id, or displaced): the former opener may be the request - /// that just left, and a merge receipt must never name an id that - /// resolves nowhere. The scheduler pop path does not need this — the - /// popped request surfaces in `active_heals` under the same id and the - /// duplicate pre-check consults active heals before the queue. No-op for - /// released keys; the survivor scan only runs when a key still has - /// holders, which under forced duplicates is the rare admin path. + pub(super) fn scanner_queued_request_id_for_dedup_key(&self, key: &str) -> Option<&str> { + let entry = self.dedup_keys.get(key)?; + match &entry.scanner_queued_owners { + ScannerQueuedOwners::Single(_) if entry.refcount == 1 => Some(entry.representative_request_id.as_str()), + ScannerQueuedOwners::Indexed(owners) if owners.len() == entry.refcount => { + owners.first_key_value().map(|(_, id)| id.as_str()) + } + _ => None, + } + } + + /// Re-elect the legacy representative from the heap after cancellation or + /// displacement. Dequeue leaves it unchanged for existing callers. pub(super) fn refresh_dedup_representative(&mut self, key: &str) { if !self.dedup_keys.contains_key(key) { return; @@ -786,7 +814,7 @@ impl PriorityHealQueue { while let Some(item) = self.heap.pop() { if removed.is_none() && item.request.id == request_id { let key = item.dedup_key.clone(); - Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key); + Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item); self.remove_best_effort_count(item.request.source); affected_key = Some(key); removed = Some(item.request); @@ -812,7 +840,7 @@ impl PriorityHealQueue { while let Some(item) = self.heap.pop() { if should_remove(&item.request) { - Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key); + Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item); self.remove_best_effort_count(item.request.source); affected_keys.push(item.dedup_key); removed.push(item.request); diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index 1b168726e..d9c67b524 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -34,6 +34,7 @@ mod auto_scan; mod completion; mod disposition; mod legacy_ambiguity_logging; +mod queued_representative; mod running_mainline; use super::super::{DiskOption, DiskStore, Endpoint, new_disk, storage_api::status::BucketInfo}; diff --git a/crates/heal/src/heal/manager/tests/admin_overlap.rs b/crates/heal/src/heal/manager/tests/admin_overlap.rs index 703e74f0a..e39c51b52 100644 --- a/crates/heal/src/heal/manager/tests/admin_overlap.rs +++ b/crates/heal/src/heal/manager/tests/admin_overlap.rs @@ -450,6 +450,176 @@ async fn admin_overlap_normal_start_cannot_enter_force_start_cancellation_window assert_eq!(manager.heal_queue.lock().await.len(), 1); } +#[tokio::test] +async fn admin_force_start_rechecks_request_id_after_cancelling_overlaps() { + for source in [HealRequestSource::Scanner, HealRequestSource::Internal] { + let manager = manager_with_policy(HealOverlapPolicy::Merge); + let previous_id = install_owner(&manager, admin_prefix_request("bucket", "scope/child/"), OwnerState::Active).await; + let aliases = manager.task_aliases.lock().await; + let mut replacement = admin_prefix_request("bucket", "scope/"); + replacement.force_start = true; + let request_id = replacement.id.clone(); + let mut forced = Box::pin(manager.submit_heal_request_with_receipt(replacement)); + + tokio::time::timeout( + Duration::from_secs(5), + std::future::poll_fn(|cx| { + assert!( + std::pin::pin!(tokio::task::unconstrained(forced.as_mut())) + .poll(cx) + .is_pending() + ); + if manager.active_heals.try_lock().is_ok() { + Poll::Ready(()) + } else { + cx.waker().wake_by_ref(); + Poll::Pending + } + }), + ) + .await + .expect("forceStart reaches cancellation with registry locks released"); + + let mut competing = HealRequest::object("other-bucket".to_string(), "object".to_string(), None); + competing.id = request_id.clone(); + competing.source = source; + competing.options.recreate_missing = false; + let receipt = manager + .submit_heal_request_with_receipt(competing) + .await + .expect("non-admin admission during forceStart cancellation"); + assert_eq!(receipt.result, HealAdmissionResult::Accepted); + assert_eq!(receipt.task_id, request_id); + + drop(aliases); + let receipt = forced.await.expect("forceStart returns a conflict"); + assert_eq!(receipt.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)); + assert_eq!(receipt.task_id, request_id); + let queue = manager.heal_queue.lock().await; + let same_id: Vec<_> = queue.requests().filter(|request| request.id == request_id).collect(); + assert_eq!(same_id.len(), 1, "conflicting requests must not share an id"); + assert_eq!(same_id[0].source, source); + drop(queue); + assert!( + matches!(manager.get_task_status(&previous_id).await, Ok(HealTaskStatus::Cancelled)), + "overlapping admin cancellation precedes the conflict recheck" + ); + } +} + +#[tokio::test] +async fn forced_id_conflict_preserves_scanner_queued_owner() { + for competing_source in [HealRequestSource::Scanner, HealRequestSource::Internal] { + let manager = manager_with_policy(HealOverlapPolicy::Merge); + let previous_id = install_owner(&manager, admin_object_request("object"), OwnerState::Active).await; + let mut queued_ids = Vec::new(); + for source in [ + HealRequestSource::Scanner, + HealRequestSource::Internal, + HealRequestSource::Scanner, + ] { + let mut request = HealRequest::object("bucket".to_string(), "object".to_string(), None); + request.source = source; + request.force_start = true; + let id = request.id.clone(); + let receipt = manager + .submit_heal_request_with_receipt(request) + .await + .expect("queue forced work"); + assert_eq!(receipt.result, HealAdmissionResult::Accepted); + assert_eq!(receipt.task_id, id); + queued_ids.push(id); + } + + // The first holder leaves the queue, leaving two forced tasks queued. + let selected = { + let mut queue = manager.heal_queue.lock().await; + let (selected, skipped) = queue.pop_runnable_with_skips(|_| true, |_| None); + assert!(skipped.is_empty()); + let selected = selected.expect("first queued task should run"); + assert_eq!(selected.id, queued_ids[0]); + assert_eq!(queue.len(), 2); + selected + }; + let completed_task = HealTask::from_request(selected, manager.storage.clone()); + let mut completed = CompletedHealStatus::snapshot(&completed_task, HealTaskStatus::Completed).await; + Arc::make_mut(completed.outcome.as_mut().expect("completed task outcome")).finish(None); + manager + .completed_heals + .lock() + .await + .insert(queued_ids[0].clone(), Arc::new(completed)); + + let aliases = manager.task_aliases.lock().await; + let mut replacement = admin_object_request("object"); + replacement.force_start = true; + let competing_id = replacement.id.clone(); + let mut forced = Box::pin(manager.submit_heal_request_with_receipt(replacement)); + tokio::time::timeout( + Duration::from_secs(5), + std::future::poll_fn(|cx| { + assert!( + std::pin::pin!(tokio::task::unconstrained(forced.as_mut())) + .poll(cx) + .is_pending() + ); + if manager.active_heals.try_lock().is_ok() { + Poll::Ready(()) + } else { + cx.waker().wake_by_ref(); + Poll::Pending + } + }), + ) + .await + .expect("forceStart reaches cancellation with registry locks released"); + + let mut competing = HealRequest::object("bucket".to_string(), "object".to_string(), None); + competing.id = competing_id.clone(); + competing.source = competing_source; + competing.force_start = true; + let receipt = manager + .submit_heal_request_with_receipt(competing) + .await + .expect("competing same-key forced work"); + assert_eq!(receipt.result, HealAdmissionResult::Accepted); + assert_eq!(receipt.task_id, competing_id); + + drop(aliases); + let receipt = forced.await.expect("forceStart returns an ID conflict"); + assert_eq!(receipt.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)); + assert_eq!(receipt.task_id, competing_id); + assert!(!manager.active_heals.lock().await.contains_key(&competing_id)); + assert!(!manager.retrying_heals.lock().await.contains_key(&competing_id)); + assert!(matches!(manager.get_task_status(&previous_id).await, Ok(HealTaskStatus::Cancelled))); + { + let queue = manager.heal_queue.lock().await; + let same_id: Vec<_> = queue.requests().filter(|request| request.id == competing_id).collect(); + assert_eq!(same_id.len(), 1, "the competing ID must have one live owner"); + assert_eq!(same_id[0].source, competing_source); + assert_eq!(queue.len(), 3); + } + + let mut scanner = HealRequest::object("bucket".to_string(), "object".to_string(), None); + scanner.source = HealRequestSource::Scanner; + let receipt = manager + .submit_heal_request_with_receipt(scanner) + .await + .expect("scanner should merge into queued work"); + assert_eq!(receipt.result, HealAdmissionResult::Merged); + assert_ne!(receipt.task_id, queued_ids[0], "the completed holder cannot own a queued receipt"); + assert!( + manager + .heal_queue + .lock() + .await + .requests() + .any(|request| request.id == receipt.task_id), + "the canonical receipt must name a live queued task" + ); + } +} + #[tokio::test] async fn admin_overlap_cancel_cannot_miss_a_queued_to_active_transition() { let manager = manager_with_policy(HealOverlapPolicy::Merge); diff --git a/crates/heal/src/heal/manager/tests/queued_representative.rs b/crates/heal/src/heal/manager/tests/queued_representative.rs new file mode 100644 index 000000000..fbcf0430c --- /dev/null +++ b/crates/heal/src/heal/manager/tests/queued_representative.rs @@ -0,0 +1,398 @@ +// Copyright 2026 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use super::*; + +fn object_request(source: HealRequestSource, priority: HealPriority, force_start: bool) -> HealRequest { + let mut request = HealRequest::new( + HealType::Object { + bucket: "canonical-owner".to_owned(), + object: "object".to_owned(), + version_id: None, + }, + HealOptions::default(), + priority, + ); + request.source = source; + request.force_start = force_start; + request +} + +async fn scanner_receipt_tracks_queued_owner(use_fairness: bool) { + let manager = HealManager::new_without_root_recovery_for_test( + Arc::new(MockStorage), + Some(HealConfig { + queue_size: 16, + low_priority_merge_enable: true, + overlap_policy: HealOverlapPolicy::Merge, + ..HealConfig::default() + }), + ); + + if use_fairness { + let mut queue = manager.heal_queue.lock().await; + for index in 0..4 { + let request = bucket_request(&format!("internal-{index}"), HealPriority::Normal, HealRequestSource::Internal); + assert_eq!(queue.push(request), QueuePushOutcome::Accepted); + } + for _ in 0..4 { + let (request, _) = queue.pop_runnable_with_fairness(|_| true, |_| None); + assert_eq!(request.expect("internal work should run").source, HealRequestSource::Internal); + } + } + + let first = object_request(HealRequestSource::Scanner, HealPriority::High, false); + let first_id = first.id.clone(); + let key = PriorityHealQueue::make_dedup_key(&first); + let second = object_request(HealRequestSource::Admin, HealPriority::High, true); + let second_id = second.id.clone(); + for request in [first, second] { + assert_eq!( + manager + .submit_heal_request_with_receipt(request) + .await + .expect("admit task") + .result, + HealAdmissionResult::Accepted + ); + } + + let first = { + let mut queue = manager.heal_queue.lock().await; + let (selected, skipped) = if use_fairness { + queue.pop_runnable_with_fairness(|_| true, |_| None) + } else { + queue.pop_runnable_with_skips(|_| true, |_| None) + }; + assert!(skipped.is_empty()); + let selected = selected.expect("the first same-priority task should run"); + assert_eq!(selected.id, first_id); + assert!(queue.contains_request_id(&second_id)); + assert_eq!( + queue.queued_request_id_for_dedup_key(&key), + Some(first_id.as_str()), + "the existing representative for other sources must remain unchanged" + ); + selected + }; + manager + .active_heals + .lock() + .await + .insert(first_id.clone(), Arc::new(HealTask::from_request(first, Arc::new(MockStorage)))); + + let active = manager + .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::Low, false)) + .await + .expect("scanner admission while the first task is active"); + assert_eq!(active.result, HealAdmissionResult::Merged); + assert_eq!(active.task_id, first_id, "active ownership keeps precedence"); + + let completed_task = manager + .active_heals + .lock() + .await + .remove(&first_id) + .expect("first task should leave active ownership"); + let mut retained = CompletedHealStatus::snapshot(&completed_task, HealTaskStatus::Completed).await; + Arc::make_mut(retained.outcome.as_mut().expect("retained task outcome")).finish(None); + manager + .completed_heals + .lock() + .await + .insert(first_id.clone(), Arc::new(retained)); + let queued = manager + .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::Low, false)) + .await + .expect("scanner admission after the first task leaves active state"); + assert_eq!(queued.result, HealAdmissionResult::Merged); + assert_eq!(queued.task_id, second_id, "the receipt must name the remaining queued task"); +} + +#[tokio::test] +async fn scanner_receipt_tracks_queued_owner_after_normal_dequeue() { + scanner_receipt_tracks_queued_owner(false).await; +} + +#[tokio::test] +async fn scanner_receipt_tracks_queued_owner_after_fair_dequeue() { + scanner_receipt_tracks_queued_owner(true).await; +} + +#[tokio::test] +async fn scanner_receipt_keeps_queued_owner_when_a_different_forced_task_runs() { + let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None); + let first = object_request(HealRequestSource::Scanner, HealPriority::Low, false); + let first_id = first.id.clone(); + let second = object_request(HealRequestSource::Admin, HealPriority::High, true); + let second_id = second.id.clone(); + { + let mut queue = manager.heal_queue.lock().await; + assert_eq!(queue.push(first), QueuePushOutcome::Accepted); + assert_eq!(queue.push(second), QueuePushOutcome::Accepted); + let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None); + assert_eq!(selected.expect("higher-priority forced task should run").id, second_id); + } + let receipt = manager + .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false)) + .await + .expect("scanner admission after a non-representative task leaves the queue"); + assert_eq!(receipt.result, HealAdmissionResult::Merged); + assert_eq!(receipt.task_id, first_id); +} + +#[tokio::test] +async fn scanner_receipt_reselects_after_the_queued_owner_is_cancelled() { + let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None); + let first = object_request(HealRequestSource::Scanner, HealPriority::High, false); + let first_id = first.id.clone(); + let second = object_request(HealRequestSource::Admin, HealPriority::High, true); + let second_id = second.id.clone(); + let third = object_request(HealRequestSource::Internal, HealPriority::High, true); + let third_id = third.id.clone(); + { + let mut queue = manager.heal_queue.lock().await; + for request in [first, second, third] { + assert_eq!(queue.push(request), QueuePushOutcome::Accepted); + } + let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None); + assert_eq!(selected.expect("first task should run").id, first_id); + } + let first_receipt = manager + .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false)) + .await + .expect("first scanner admission"); + assert_eq!(first_receipt.result, HealAdmissionResult::Merged); + assert!(first_receipt.task_id == second_id || first_receipt.task_id == third_id); + let remaining = if first_receipt.task_id == second_id { + third_id + } else { + second_id + }; + manager.heal_queue.lock().await.remove_request_id(&first_receipt.task_id); + + let next_receipt = manager + .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false)) + .await + .expect("scanner admission after the queued owner is removed"); + assert_eq!(next_receipt.result, HealAdmissionResult::Merged); + assert_eq!(next_receipt.task_id, remaining); +} + +#[tokio::test] +async fn scanner_receipt_reselects_after_multiple_forced_dequeues() { + let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None); + let first = object_request(HealRequestSource::Scanner, HealPriority::High, false); + let first_id = first.id.clone(); + let second = object_request(HealRequestSource::Admin, HealPriority::High, true); + let second_id = second.id.clone(); + let third = object_request(HealRequestSource::Internal, HealPriority::High, true); + let third_id = third.id.clone(); + { + let mut queue = manager.heal_queue.lock().await; + for request in [first, second, third] { + assert_eq!(queue.push(request), QueuePushOutcome::Accepted); + } + let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None); + assert_eq!(selected.expect("first task should run").id, first_id); + } + let first_receipt = manager + .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false)) + .await + .expect("first scanner admission"); + assert_eq!(first_receipt.result, HealAdmissionResult::Merged); + assert!(first_receipt.task_id == second_id || first_receipt.task_id == third_id); + let remaining = if first_receipt.task_id == second_id { + third_id + } else { + second_id + }; + { + let mut queue = manager.heal_queue.lock().await; + let (selected, _) = queue.pop_runnable_with_skips(|request| request.id == first_receipt.task_id, |_| None); + assert_eq!(selected.expect("selected queued owner should run").id, first_receipt.task_id); + } + let next_receipt = manager + .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false)) + .await + .expect("scanner admission after another queued owner starts"); + assert_eq!(next_receipt.result, HealAdmissionResult::Merged); + assert_eq!(next_receipt.task_id, remaining); +} + +#[test] +fn scanner_owner_index_survives_a_single_remaining_holder_and_new_forced_work() { + let mut queue = PriorityHealQueue::new(); + let first = object_request(HealRequestSource::Scanner, HealPriority::High, false); + let first_id = first.id.clone(); + let key = PriorityHealQueue::make_dedup_key(&first); + let second = object_request(HealRequestSource::Admin, HealPriority::High, true); + let second_id = second.id.clone(); + let third = object_request(HealRequestSource::Internal, HealPriority::High, true); + let third_id = third.id.clone(); + + assert_eq!(queue.push(first), QueuePushOutcome::Accepted); + assert!(matches!(&queue.dedup_keys[&key].scanner_queued_owners, ScannerQueuedOwners::Single(_))); + assert_eq!(queue.push(second), QueuePushOutcome::Accepted); + assert_eq!(queue.pop().expect("first queued task").id, first_id); + assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(first_id.as_str())); + assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str())); + + assert_eq!(queue.push(third), QueuePushOutcome::Accepted); + assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str())); + assert_eq!(queue.pop().expect("second queued task").id, second_id); + assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(third_id.as_str())); +} + +#[test] +fn scanner_owner_index_distinguishes_holders_with_the_same_request_id() { + let mut queue = PriorityHealQueue::new(); + let first = object_request(HealRequestSource::Scanner, HealPriority::High, false); + let key = PriorityHealQueue::make_dedup_key(&first); + let shared_id = first.id.clone(); + let mut second = object_request(HealRequestSource::Admin, HealPriority::High, true); + second.id.clone_from(&shared_id); + let third = object_request(HealRequestSource::Internal, HealPriority::High, true); + let third_id = third.id.clone(); + for request in [first, second, third] { + assert_eq!(queue.push(request), QueuePushOutcome::Accepted); + } + + assert_eq!(queue.pop().expect("first holder").id, shared_id); + assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(shared_id.as_str())); + assert_eq!(queue.remove_request_id(&shared_id).expect("second holder").id, shared_id); + assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(third_id.as_str())); + assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(third_id.as_str())); +} + +#[test] +fn scanner_owner_index_tracks_bulk_removal_and_key_release() { + let mut queue = PriorityHealQueue::new(); + let first = object_request(HealRequestSource::Scanner, HealPriority::High, false); + let first_id = first.id.clone(); + let key = PriorityHealQueue::make_dedup_key(&first); + let second = object_request(HealRequestSource::Admin, HealPriority::Normal, true); + let second_id = second.id.clone(); + let third = object_request(HealRequestSource::Internal, HealPriority::Low, true); + let third_id = third.id.clone(); + for request in [first, second, third] { + assert_eq!(queue.push(request), QueuePushOutcome::Accepted); + } + + let removed = queue.remove_matching(|request| request.id == first_id || request.id == second_id); + assert_eq!(removed.len(), 2); + assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(third_id.as_str())); + assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(third_id.as_str())); + assert_eq!(queue.remove_matching(|request| request.id == third_id).len(), 1); + assert!(!queue.dedup_keys.contains_key(&key)); + + let next = object_request(HealRequestSource::Scanner, HealPriority::High, false); + let next_id = next.id.clone(); + assert_eq!(queue.push(next), QueuePushOutcome::Accepted); + assert!(matches!(&queue.dedup_keys[&key].scanner_queued_owners, ScannerQueuedOwners::Single(_))); + assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(next_id.as_str())); +} + +#[test] +fn scanner_owner_index_tracks_displacement_without_changing_legacy_reselection() { + let mut queue = PriorityHealQueue::new(); + let first = object_request(HealRequestSource::Scanner, HealPriority::Low, false); + let first_id = first.id.clone(); + let key = PriorityHealQueue::make_dedup_key(&first); + let second = object_request(HealRequestSource::Admin, HealPriority::Normal, true); + let second_id = second.id.clone(); + assert_eq!(queue.push(first), QueuePushOutcome::Accepted); + assert_eq!(queue.push(second), QueuePushOutcome::Accepted); + + let incoming = bucket_request("displacement", HealPriority::Urgent, HealRequestSource::Internal); + let displaced = queue + .push_displacing_lower_priority(incoming) + .expect("lower priority task should be displaced"); + assert_eq!(displaced.id, first_id); + assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str())); + assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(second_id.as_str())); +} + +#[tokio::test] +async fn scanner_receipt_rejects_a_missing_or_empty_holder_index() { + for missing in [true, false] { + let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None); + let first = object_request(HealRequestSource::Scanner, HealPriority::High, false); + let key = PriorityHealQueue::make_dedup_key(&first); + let second = object_request(HealRequestSource::Admin, HealPriority::High, true); + { + let mut queue = manager.heal_queue.lock().await; + assert_eq!(queue.push(first), QueuePushOutcome::Accepted); + assert_eq!(queue.push(second), QueuePushOutcome::Accepted); + let entry = queue.dedup_keys.get_mut(&key).expect("dedup key should exist"); + entry.scanner_queued_owners = if missing { + ScannerQueuedOwners::Single(0) + } else { + ScannerQueuedOwners::Indexed(std::collections::BTreeMap::new()) + }; + } + + for leave_one in [false, true] { + if leave_one { + manager.heal_queue.lock().await.pop().expect("one holder should leave"); + } + let request = object_request(HealRequestSource::Scanner, HealPriority::High, false); + let request_id = request.id.clone(); + let receipt = manager + .submit_heal_request_with_receipt(request) + .await + .expect("admission should fail closed"); + assert_eq!(receipt.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)); + assert_eq!(receipt.task_id, request_id); + } + } +} + +#[test] +fn scanner_and_legacy_owners_keep_their_distinct_reselection_orders() { + let mut queue = PriorityHealQueue::new(); + let first = object_request(HealRequestSource::Scanner, HealPriority::High, false); + let first_id = first.id.clone(); + let key = PriorityHealQueue::make_dedup_key(&first); + let second = object_request(HealRequestSource::Admin, HealPriority::Low, true); + let second_id = second.id.clone(); + let third = object_request(HealRequestSource::Internal, HealPriority::Normal, true); + let third_id = third.id.clone(); + for request in [first, second, third] { + assert_eq!(queue.push(request), QueuePushOutcome::Accepted); + } + + assert_eq!(queue.remove_request_id(&first_id).expect("first holder should be removed").id, first_id); + assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str())); + assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(third_id.as_str())); +} + +#[test] +fn scanner_owner_index_tracks_a_series_of_forced_holders() { + let mut queue = PriorityHealQueue::new(); + let first = object_request(HealRequestSource::Scanner, HealPriority::High, false); + let key = PriorityHealQueue::make_dedup_key(&first); + let mut ids = vec![first.id.clone()]; + assert_eq!(queue.push(first), QueuePushOutcome::Accepted); + for _ in 0..128 { + let request = object_request(HealRequestSource::Internal, HealPriority::High, true); + ids.push(request.id.clone()); + assert_eq!(queue.push(request), QueuePushOutcome::Accepted); + } + for id in &ids { + assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(id.as_str())); + assert_eq!(&queue.pop().expect("indexed holder should remain queued").id, id); + } + assert!(!queue.dedup_keys.contains_key(&key)); +}