gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-28 06:47:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(heal): preserve queued ownership under force admission

Browse Source
This commit is contained in:
overtrue
2026-10-28 07:28:47 +08:00
parent 7a3bf8cdbd
commit c96fc5cf26
5 changed files with 755 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/manager.rs
+111 -81
View File
@@ -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<String, Arc<HealTask>>,
queue: &PriorityHealQueue,
retrying_heals: &HashMap<String, RetryingHeal>,
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<String, RetryingHeal>, 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::<HashSet<_>>();
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::<HashSet<_>>();
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);
crates/heal/src/heal/manager/queue.rs
+75 -47
View File
@@ -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<u64, String>),
}
/// Priority queue wrapper for heal requests
@@ -306,11 +314,7 @@ impl PriorityHealQueue {
#[cfg(test)]
pub(super) fn pop_next(&mut self) -> Option<HealRequest> {
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<HealRequest> {
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<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<HealRequest>, Vec<String>)
@@ -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<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<HealRequest>, Vec<String>)
@@ -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<String, DedupKeyEntry>, key: &str) {
if let Some(entry) = dedup_keys.get_mut(key) {
fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap<String, DedupKeyEntry>, 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);
crates/heal/src/heal/manager/tests.rs
+1
View File
@@ -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};
crates/heal/src/heal/manager/tests/admin_overlap.rs
+170
View File
@@ -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);
crates/heal/src/heal/manager/tests/queued_representative.rs
+398
View File
@@ -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));
}