gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-21 03:46:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(heal): split manager.rs queue/scheduler/scan children (#6303)

Browse Source 
Split the 6723-line manager.rs (44% inline tests) into a canonical
manager.rs + manager/ module tree with zero behavior change:

- manager.rs (~1830): HealManager and HealState, HealConfig, task
  report/snapshot types, overlap policy, admission classification and
  queue admission, submit paths, task-state queries, and the
  statistics surface
- manager/queue.rs (~450): the priority heal queue, its per-key dedup
  index, and the queue bookkeeping structs
- manager/scheduler.rs (~620): start_scheduler and the
  process_heal_queue consumption loop with its skip/metric helpers
- manager/auto_scan.rs (~550): the automatic disk scanner
- manager/unclean_shutdown.rs (~390): unclean-shutdown recovery and
  its durable replacement-intent helpers
- manager/tests.rs (~2970): the inline test module as a child module

All module paths are unchanged. The queue structs' fields and the
cross-module helpers gain pub(super), whose scope equals the old
single-module privacy domain; HealManager's private fields stay in the
root and remain reachable from child impl blocks. Code is moved
verbatim apart from those markers, heal-level super:: path fixes for
the unclean-shutdown move, per-module import headers, and rustfmt
re-wraps.

The logging-guardrail rule for the manager demote_to_debug_when! count
now sums manager.rs with its manager/*.rs children, since one
scheduler site moved with process_heal_queue; the >= 6 threshold is
unchanged and the forbidden admission info!/warn! pattern check keeps
targeting the root admission code.

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-20 19:12:07 +08:00
committed by GitHub
parent 129677f0b3
commit 2b9dcc646f
7 changed files with 4998 additions and 4902 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/manager.rs
+8 -4900
View File
File diff suppressed because it is too large Load Diff
crates/heal/src/heal/manager/auto_scan.rs
+555
View File
@@ -0,0 +1,555 @@
// Copyright 2024 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.
/// The automatic disk scanner: replacement discovery and unformatted-disk enqueue.
use super::*;
impl HealManager {
/// Start background task to auto scan local disks and enqueue erasure set heal requests
pub(super) async fn start_auto_disk_scanner(&self) -> Result<()> {
let config = self.config.clone();
let heal_queue = self.heal_queue.clone();
let active_heals = self.active_heals.clone();
let storage = self.storage.clone();
let replacement_recovery_anchors = self.replacement_recovery_anchors.clone();
let replacement_recovery_blocked_sets = self.replacement_recovery_blocked_sets.clone();
let cancel_token = self.cancel_token.clone();
let notify = self.notify.clone();
let mut duration = {
let config = config.read().await;
config.heal_interval
};
if duration < Duration::from_secs(10) {
duration = Duration::from_secs(10);
}
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
state = "started",
interval = ?duration,
"Heal auto disk scanner started"
);
tokio::spawn(async move {
let mut interval = interval(duration);
loop {
let mut candidate_count = 0usize;
let mut skipped_duplicate_count = 0usize;
let mut skipped_invalid_count = 0usize;
let mut enqueued_count = 0usize;
let mut not_enqueued_count = 0usize;
let mut dropped_count = 0usize;
let mut full_count = 0usize;
tokio::select! {
_ = cancel_token.cancelled() => {
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
state = "shutdown",
"Heal auto disk scanner stopped"
);
break;
}
_ = interval.tick() => {
// Build list of endpoints that need healing
let mut endpoints = HashMap::<String, Vec<Endpoint>>::new();
let mut durable_recoveries = HashMap::<String, (String, Vec<Endpoint>, Vec<String>, String)>::new();
let mut conflicted_recovery_sets = HashSet::<String>::new();
let mut deferred_replacement_endpoints = HashSet::<String>::new();
let local_disks = {
let local_disk_map = local_disk_map_read().await;
local_disk_map.values().flatten().cloned().collect::<Vec<_>>()
};
let local_endpoints = local_disks.iter().map(|disk| disk.endpoint()).collect::<Vec<_>>();
let blocked_sets = replacement_recovery_blocked_sets
.lock()
.expect("replacement recovery blocked set lock poisoned")
.clone();
if !blocked_sets.is_empty() {
let mut retry_succeeded = HashSet::new();
let mut retry_failed = HashSet::new();
for disk in &local_disks {
let endpoint = disk.endpoint();
let Some(set_disk_id) =
crate::heal::utils::format_set_disk_id_from_i32(endpoint.pool_idx, endpoint.set_idx)
else {
continue;
};
if !blocked_sets.contains(&set_disk_id) {
continue;
}
match Self::validate_replacement_recovery_records(disk).await {
Ok(()) => {
retry_succeeded.insert(set_disk_id);
}
Err(error) => {
retry_failed.insert(set_disk_id.clone());
conflicted_recovery_sets.insert(set_disk_id);
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint = %endpoint,
error = %error,
"Replacement recovery retry failed"
);
}
}
}
let mut blocked = replacement_recovery_blocked_sets
.lock()
.expect("replacement recovery blocked set lock poisoned");
unblock_replacement_recovery_sets_after_validation(&mut blocked, retry_succeeded, &retry_failed);
}
for disk in &local_disks {
let endpoint = disk.endpoint();
let runtime_state = disk.runtime_state();
let set_disk_id =
crate::heal::utils::format_set_disk_id_from_i32(endpoint.pool_idx, endpoint.set_idx);
if set_disk_id.as_ref().is_some_and(|set_disk_id| {
replacement_recovery_blocked_sets
.lock()
.expect("replacement recovery blocked set lock poisoned")
.contains(set_disk_id)
}) {
skipped_invalid_count += 1;
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_DISK,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint = %endpoint,
set_disk_id = set_disk_id.as_deref().unwrap_or_default(),
disk_state = "replacement_recovery_blocked",
"Heal auto-scan replacement deferred because durable recovery is blocked"
);
continue;
}
// detect unformatted disk via get_disk_id()
match disk.get_disk_id().await {
Err(DiskError::UnformattedDisk) => {
if !super::super::replacement_readiness::auto_replacement_target_ready(disk, &local_disks)
.await
{
deferred_replacement_endpoints.insert(endpoint.to_string());
skipped_invalid_count += 1;
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_DISK,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint = %endpoint,
disk_state = "replacement_path_unavailable",
"Heal auto-scan replacement deferred"
);
continue;
}
let Some(set_disk_id) = set_disk_id else {
skipped_invalid_count += 1;
continue;
};
candidate_count += 1;
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_DISK,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint = %endpoint,
disk_state = "unformatted",
"Heal auto-scan candidate detected"
);
endpoints.entry(set_disk_id).or_default().push(endpoint);
}
Err(e) => {
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_DISK,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint = %endpoint,
disk_state = "check_failed",
error = ?e,
"Heal auto-scan disk inspection failed"
);
}
Ok(_) => {
if runtime_state.as_str() == "returning" && let Some(set_disk_id) = set_disk_id {
candidate_count += 1;
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_DISK,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint = %endpoint,
set_disk_id,
disk_state = "returning",
"Heal auto-scan returning disk candidate detected"
);
endpoints.entry(set_disk_id).or_default().push(endpoint);
}
}
}
}
// Once formatting succeeds a replacement is no longer
// discoverable as UnformattedDisk. Re-admit exactly one
// incomplete durable generation per set after bounded
// scheduler retries are exhausted, or re-admit its
// verified terminal cleanup. Multiple generations are a
// durable conflict: leave every marker/state intact and
// require reconciliation rather than choosing one.
for disk in &local_disks {
let endpoint = disk.endpoint();
let disk_set_disk_id =
crate::heal::utils::format_set_disk_id_from_i32(endpoint.pool_idx, endpoint.set_idx);
let replacement_task_ids = match ResumeUtils::get_replacement_intent_tasks(disk).await {
Ok(task_ids) => task_ids,
Err(error) => {
let endpoint_string = endpoint.to_string();
if replacement_discovery_error_is_expected_for_deferred_endpoint(
&error,
&endpoint_string,
&deferred_replacement_endpoints,
) {
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint = %endpoint,
disk_state = "replacement_path_unavailable",
result = "recovery_records_unavailable",
"Replacement recovery discovery skipped for deferred replacement"
);
continue;
}
if let Some(set_disk_id) = &disk_set_disk_id {
conflicted_recovery_sets.insert(set_disk_id.clone());
}
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint = %endpoint,
error = %error,
"Replacement recovery discovery failed"
);
continue;
}
};
for task_id in replacement_task_ids {
let resume_manager = match ResumeManager::load_replacement_intent(disk.clone(), &task_id).await {
Ok(resume_manager) => resume_manager,
Err(error) => {
if let Some(set_disk_id) = &disk_set_disk_id {
conflicted_recovery_sets.insert(set_disk_id.clone());
}
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint = %endpoint,
task_id,
error = %error,
"Replacement recovery intent load failed"
);
continue;
}
};
let state = resume_manager.get_state().await;
if !durable_replacement_recovery_is_due(&state, &task_id) {
continue;
}
if !matches!(state.replacement_phase, ReplacementPhase::CleanupPending) {
let Ok(identities) = storage.replacement_target_identities(&state.replacement_targets).await else {
continue;
};
if identities != state.replacement_target_identities {
continue;
}
}
let targets = state
.replacement_targets
.iter()
.filter_map(|target| {
local_endpoints
.iter()
.find(|endpoint| endpoint.to_string() == *target)
.cloned()
})
.collect::<Vec<_>>();
if targets.len() != state.replacement_targets.len() {
continue;
}
let Some(set_disk_id) = crate::heal::utils::format_set_disk_id_from_i32(
targets[0].pool_idx,
targets[0].set_idx,
) else {
continue;
};
if targets.iter().any(|target| {
crate::heal::utils::format_set_disk_id_from_i32(target.pool_idx, target.set_idx)
.as_deref()
!= Some(set_disk_id.as_str())
}) {
continue;
}
let resume_endpoint = disk.endpoint().to_string();
match durable_recoveries.get(&set_disk_id) {
Some((existing_task_id, _, _, existing_anchor))
if existing_task_id != &task_id || existing_anchor != &resume_endpoint => {
replacement_recovery_blocked_sets
.lock()
.expect("replacement recovery blocked set lock poisoned")
.insert(set_disk_id.clone());
conflicted_recovery_sets.insert(set_disk_id);
}
Some(_) => {}
None => {
durable_recoveries.insert(
set_disk_id,
(task_id, targets, state.replacement_buckets, resume_endpoint),
);
}
}
}
}
for set_disk_id in &conflicted_recovery_sets {
durable_recoveries.remove(set_disk_id);
endpoints.remove(set_disk_id);
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
set_disk_id,
result = "durable_generation_conflict",
"Replacement recovery deferred because multiple durable generations exist"
);
}
for (set_disk_id, (_, targets, _, _)) in &durable_recoveries {
let expected = targets.iter().map(ToString::to_string).collect::<HashSet<_>>();
let observed = endpoints
.get(set_disk_id)
.map(|endpoints| endpoints.iter().map(ToString::to_string).collect::<HashSet<_>>())
.unwrap_or_default();
if !observed.is_subset(&expected) {
replacement_recovery_blocked_sets
.lock()
.expect("replacement recovery blocked set lock poisoned")
.insert(set_disk_id.clone());
conflicted_recovery_sets.insert(set_disk_id.clone());
continue;
}
endpoints.entry(set_disk_id.clone()).or_default().extend(targets.clone());
}
for set_disk_id in &conflicted_recovery_sets {
durable_recoveries.remove(set_disk_id);
endpoints.remove(set_disk_id);
}
for target_endpoints in endpoints.values_mut() {
target_endpoints.sort_by_key(ToString::to_string);
target_endpoints.dedup_by(|left, right| left.to_string() == right.to_string());
}
if endpoints.is_empty() {
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
state = "idle",
"Heal auto disk scanner idle"
);
continue;
}
// Admit one set task with every ready replacement target. Queue deduplication is
// set-scoped, so admitting endpoints independently would silently drop later targets.
for (set_disk_id, endpoints) in endpoints {
if replacement_recovery_blocked_sets
.lock()
.expect("replacement recovery blocked set lock poisoned")
.contains(&set_disk_id)
{
skipped_invalid_count += 1;
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
set_disk_id,
result = "replacement_recovery_blocked",
"Heal auto-scan replacement admission deferred because durable recovery is blocked"
);
continue;
}
// skip if already queued or healing
// Use consistent lock order: queue first, then active_heals to avoid deadlock
let mut skip = false;
{
let queue = heal_queue.lock().await;
if queue.contains_erasure_set(&set_disk_id) {
skip = true;
}
}
if !skip {
let active = active_heals.lock().await;
if active.values().any(|task| {
matches!(
&task.heal_type,
crate::heal::task::HealType::ErasureSet { set_disk_id: active_id, .. }
if active_id == &set_disk_id
)
}) {
skip = true;
}
}
if skip {
skipped_duplicate_count += 1;
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint_count = endpoints.len(),
set_disk_id,
result = "skipped_duplicate",
"Heal auto-scan duplicate skipped"
);
continue;
}
// enqueue erasure set heal request for all ready replacements in this set
let recovery = durable_recoveries.remove(&set_disk_id);
let mut req = HealRequest::new(
HealType::ErasureSet {
buckets: recovery
.as_ref()
.map(|(_, _, buckets, _)| buckets.clone())
.unwrap_or_default(),
set_disk_id: set_disk_id.clone(),
},
HealOptions {
pool_index: endpoints
.first()
.and_then(|endpoint| usize::try_from(endpoint.pool_idx).ok()),
set_index: endpoints
.first()
.and_then(|endpoint| usize::try_from(endpoint.set_idx).ok()),
timeout: None,
..HealOptions::default()
},
HealPriority::Low,
);
let recovery_anchor = recovery.as_ref().map(|(_, _, _, anchor)| anchor.clone());
if let Some((task_id, _, _, _)) = recovery {
req.id = task_id;
}
req.source = HealRequestSource::AutoHeal;
req.heal_endpoints = endpoints.iter().map(ToString::to_string).collect();
let request_id = req.id.clone();
let endpoint_count = req.heal_endpoints.len();
let config = config.read().await;
let mut queue = heal_queue.lock().await;
let admission = Self::admit_request_to_queue(&mut queue, req, &config, "auto_scan");
let should_notify =
matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
if matches!(admission, HealAdmissionResult::Accepted)
&& let Some(anchor) = recovery_anchor
{
replacement_recovery_anchors
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.insert(request_id, anchor);
}
drop(queue);
drop(config);
if matches!(admission, HealAdmissionResult::Accepted) {
if should_notify {
notify.notify_one();
}
enqueued_count += 1;
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint_count,
set_disk_id,
bucket_count = 0,
result = "enqueued",
"Heal auto-scan task enqueued"
);
} else {
if matches!(admission, HealAdmissionResult::Merged) {
skipped_duplicate_count += 1;
} else {
not_enqueued_count += 1;
}
if matches!(admission, HealAdmissionResult::Full) {
full_count += 1;
}
if matches!(admission, HealAdmissionResult::Dropped(_)) {
dropped_count += 1;
}
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint_count,
set_disk_id,
bucket_count = 0,
admission = admission.result_label(),
reason = admission.reason_label(),
result = "not_enqueued",
"Heal auto-scan task not enqueued"
);
}
}
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
state = "cycle_completed",
candidate_count,
enqueued_count,
not_enqueued_count,
dropped_count,
full_count,
skipped_duplicate_count,
skipped_invalid_count,
"Heal auto-scan cycle completed"
);
}
}
}
});
Ok(())
}
}
crates/heal/src/heal/manager/queue.rs
+452
View File
@@ -0,0 +1,452 @@
// Copyright 2024 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.
/// The priority heal queue and its per-key dedup index.
use super::*;
/// 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.
#[derive(Debug)]
pub(super) struct DedupKeyEntry {
pub(super) refcount: usize,
pub(super) representative_request_id: String,
}
/// Priority queue wrapper for heal requests
/// Uses BinaryHeap for priority-based ordering while maintaining FIFO for same-priority items
#[derive(Debug)]
pub(super) struct PriorityHealQueue {
/// Heap of (priority, sequence, request) tuples
pub(super) heap: BinaryHeap<PriorityQueueItem>,
/// Sequence counter for FIFO ordering within same priority
pub(super) sequence: u64,
/// Deduplication index for queued requests
pub(super) dedup_keys: HashMap<String, DedupKeyEntry>,
}
/// Wrapper for heap items to implement proper ordering
#[derive(Debug)]
pub(super) struct PriorityQueueItem {
pub(super) priority: HealPriority,
pub(super) sequence: u64,
pub(super) request: HealRequest,
}
impl Eq for PriorityQueueItem {}
impl PartialEq for PriorityQueueItem {
fn eq(&self, other: &Self) -> bool {
self.priority == other.priority && self.sequence == other.sequence
}
}
impl Ord for PriorityQueueItem {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
// First compare by priority (higher priority first)
match self.priority.cmp(&other.priority) {
std::cmp::Ordering::Equal => {
// If priorities are equal, use sequence for FIFO (lower sequence first)
other.sequence.cmp(&self.sequence)
}
ordering => ordering,
}
}
}
impl PartialOrd for PriorityQueueItem {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum QueuePushOutcome {
Accepted,
Merged,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct ForegroundPressure {
pub(super) class: WorkloadClass,
pub(super) usage_pct: usize,
pub(super) threshold_pct: usize,
}
impl ForegroundPressure {
pub(super) const fn reason(self) -> &'static str {
match self.class {
WorkloadClass::ForegroundRead => "foreground_read_pressure",
WorkloadClass::ForegroundWrite => "foreground_write_pressure",
_ => "foreground_pressure",
}
}
}
#[derive(Debug, Clone)]
pub(super) struct CompletedHealStatus {
pub(super) heal_type: HealType,
pub(super) status: HealTaskStatus,
pub(super) result_items_truncated: bool,
pub(super) completed_at: SystemTime,
/// Sequence-stamped retained window, archived with the completion so
/// incremental consumers keep their cursor across the transition (HS-06).
/// The un-stamped legacy view is derived from it on demand.
pub(super) seqed_items: Vec<(u64, HealResultItem)>,
pub(super) next_seq: u64,
pub(super) min_seq: u64,
}
#[derive(Debug, Clone)]
pub(super) struct HealTaskAlias {
pub(super) task_id: String,
}
#[derive(Debug, Clone)]
pub(super) struct RetryingHeal {
pub(super) request: HealRequest,
pub(super) error: String,
pub(super) cancel_token: CancellationToken,
}
impl PriorityHealQueue {
pub(super) fn new() -> Self {
Self {
heap: BinaryHeap::new(),
sequence: 0,
dedup_keys: HashMap::new(),
}
}
pub(super) fn len(&self) -> usize {
self.heap.len()
}
pub(super) fn pop_next(&mut self) -> Option<HealRequest> {
self.heap.pop().map(|item| {
let key = Self::make_dedup_key(&item.request);
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
item.request
})
}
pub(super) fn is_empty(&self) -> bool {
self.heap.is_empty()
}
pub(super) fn push(&mut self, request: HealRequest) -> QueuePushOutcome {
let key = Self::make_dedup_key(&request);
// Check for duplicates unless the caller explicitly forces admission.
if self.dedup_keys.contains_key(&key) && !request.force_start {
return QueuePushOutcome::Merged;
}
// Track dedup keys for both normal and forced requests so queued forced work
// 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)
.or_insert_with(|| DedupKeyEntry {
refcount: 0,
representative_request_id: request.id.clone(),
})
.refcount += 1;
self.sequence += 1;
self.heap.push(PriorityQueueItem {
priority: request.priority,
sequence: self.sequence,
request,
});
QueuePushOutcome::Accepted
}
pub(super) fn can_displace_lower_priority(&self, priority: HealPriority) -> bool {
self.heap.iter().any(|item| item.priority < priority)
}
pub(super) fn push_displacing_lower_priority(&mut self, request: HealRequest) -> Option<HealRequest> {
let mut retained = BinaryHeap::new();
let mut displaced: Option<PriorityQueueItem> = None;
while let Some(item) = self.heap.pop() {
if item.priority < request.priority {
let should_displace = displaced
.as_ref()
.map(|current| {
item.priority < current.priority
|| (item.priority == current.priority && item.sequence > current.sequence)
})
.unwrap_or(true);
if should_displace {
if let Some(current) = displaced.replace(item) {
retained.push(current);
}
} else {
retained.push(item);
}
} else {
retained.push(item);
}
}
self.heap = retained;
let displaced = displaced.map(|item| {
let key = Self::make_dedup_key(&item.request);
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
self.refresh_dedup_representative(&key);
item.request
});
if displaced.is_some() {
// The enqueue side effect must run in ALL builds. Do NOT fold `self.push(request)`
// into `debug_assert_eq!` — in release builds (`debug_assertions` off) the whole
// macro, including its argument expression, is compiled out, which would silently
// drop the new high-priority request after having already evicted a queued item.
let outcome = self.push(request);
debug_assert_eq!(outcome, QueuePushOutcome::Accepted);
}
displaced
}
/// Get statistics about queue contents by priority
pub(super) fn get_priority_stats(&self) -> HashMap<HealPriority, usize> {
let mut stats = HashMap::new();
for item in &self.heap {
*stats.entry(item.priority).or_insert(0) += 1;
}
stats
}
pub(super) fn operation_counts(&self) -> (HealPriorityCounts, HealSourceCounts) {
let mut priority = HealPriorityCounts::default();
let mut source = HealSourceCounts::default();
for item in &self.heap {
priority.increment(item.request.priority);
source.increment(item.request.source);
}
(priority, source)
}
#[cfg(test)]
pub(super) fn pop(&mut self) -> Option<HealRequest> {
self.heap.pop().map(|item| {
let key = Self::make_dedup_key(&item.request);
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
item.request
})
}
#[cfg(test)]
pub(super) fn pop_runnable<F>(&mut self, can_run: F) -> Option<HealRequest>
where
F: Fn(&HealRequest) -> bool,
{
self.pop_runnable_with_skips(can_run, |_| None).0
}
pub(super) fn pop_runnable_with_skips<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<HealRequest>, Vec<String>)
where
F: Fn(&HealRequest) -> bool,
G: Fn(&HealRequest) -> Option<String>,
{
let mut deferred = Vec::new();
let mut selected = None;
let mut skipped = Vec::new();
while let Some(item) = self.heap.pop() {
if can_run(&item.request) {
selected = Some(item);
break;
}
if let Some(label) = skip_label(&item.request) {
skipped.push(label);
}
deferred.push(item);
}
for item in deferred {
self.heap.push(item);
}
(
selected.map(|item| {
let key = Self::make_dedup_key(&item.request);
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
item.request
}),
skipped,
)
}
/// Create a deduplication key from a heal request
pub(super) fn make_dedup_key(request: &HealRequest) -> String {
Self::make_dedup_key_for_type(&request.heal_type)
}
pub(super) fn make_dedup_key_for_type(heal_type: &HealType) -> String {
match heal_type {
HealType::Cluster => "cluster".to_string(),
HealType::Object {
bucket,
object,
version_id,
} => {
format!("object:{}:{}:{}", bucket, object, version_id.as_deref().unwrap_or(""))
}
HealType::Bucket { bucket } => {
format!("bucket:{bucket}")
}
HealType::Prefix { bucket, prefix } => {
format!("prefix:{bucket}/{prefix}")
}
HealType::ErasureSet { set_disk_id, .. } => {
format!("erasure_set:{set_disk_id}")
}
HealType::Metadata { bucket, object } => {
format!("metadata:{bucket}:{object}")
}
HealType::ECDecode {
bucket,
object,
version_id,
} => {
format!("ecdecode:{}:{}:{}", bucket, object, version_id.as_deref().unwrap_or(""))
}
}
}
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) {
if entry.refcount <= 1 {
dedup_keys.remove(key);
} else {
entry.refcount -= 1;
}
}
}
/// Check if an erasure set heal request for a specific set_disk_id exists
pub(super) fn contains_erasure_set(&self, set_disk_id: &str) -> bool {
let key = format!("erasure_set:{set_disk_id}");
self.dedup_keys.contains_key(&key)
}
/// Iterate queued requests (used by the admin overlap check).
pub(super) fn requests(&self) -> impl Iterator<Item = &HealRequest> {
self.heap.iter().map(|item| &item.request)
}
pub(super) fn contains_request_id(&self, request_id: &str) -> bool {
self.heap.iter().any(|item| item.request.id == request_id)
}
pub(super) fn contains_request_id_matching_path(&self, request_id: &str, heal_path: &str) -> bool {
self.heap
.iter()
.any(|item| item.request.id == request_id && heal_type_matches_path(&item.request.heal_type, heal_path))
}
pub(super) fn queued_request_id_for_dedup_key(&self, key: &str) -> Option<&str> {
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 refresh_dedup_representative(&mut self, key: &str) {
if !self.dedup_keys.contains_key(key) {
return;
}
if let Some(id) = self
.heap
.iter()
.find(|item| Self::make_dedup_key(&item.request) == key)
.map(|item| item.request.id.clone())
&& let Some(entry) = self.dedup_keys.get_mut(key)
{
entry.representative_request_id = id;
}
}
pub(super) fn contains_matching<F>(&self, mut matches: F) -> bool
where
F: FnMut(&HealRequest) -> bool,
{
self.heap.iter().any(|item| matches(&item.request))
}
pub(super) fn remove_request_id(&mut self, request_id: &str) -> Option<HealRequest> {
let mut retained = BinaryHeap::new();
let mut removed = None;
while let Some(item) = self.heap.pop() {
if removed.is_none() && item.request.id == request_id {
let key = Self::make_dedup_key(&item.request);
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
removed = Some(item.request);
} else {
retained.push(item);
}
}
self.heap = retained;
if let Some(removed) = removed.as_ref() {
self.refresh_dedup_representative(&Self::make_dedup_key(removed));
}
removed
}
pub(super) fn remove_matching<F>(&mut self, mut should_remove: F) -> Vec<HealRequest>
where
F: FnMut(&HealRequest) -> bool,
{
let mut retained = BinaryHeap::new();
let mut removed = Vec::new();
let mut affected_keys = Vec::new();
while let Some(item) = self.heap.pop() {
if should_remove(&item.request) {
let key = Self::make_dedup_key(&item.request);
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
affected_keys.push(key);
removed.push(item.request);
} else {
retained.push(item);
}
}
self.heap = retained;
for key in &affected_keys {
self.refresh_dedup_representative(key);
}
removed
}
}
impl RetryingHeal {
pub(super) fn status(&self) -> HealTaskStatus {
HealTaskStatus::Retrying {
error: self.error.clone(),
retry_attempt: self.request.retry_attempts,
}
}
}
crates/heal/src/heal/manager/scheduler.rs
+618
View File
@@ -0,0 +1,618 @@
// Copyright 2024 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.
/// The heal scheduler: queue consumption loop and its skip/metric helpers.
use super::*;
impl HealManager {
/// Start scheduler
pub(super) async fn start_scheduler(&self) -> Result<()> {
let config = self.config.clone();
let heal_queue = self.heal_queue.clone();
let active_heals = self.active_heals.clone();
let completed_heals = self.completed_heals.clone();
let retrying_heals = self.retrying_heals.clone();
let replacement_recovery_anchors = self.replacement_recovery_anchors.clone();
let cancel_token = self.cancel_token.clone();
let statistics = self.statistics.clone();
let storage = self.storage.clone();
let notify = self.notify.clone();
let workload_provider = self.workload_provider.clone();
tokio::spawn(async move {
let mut interval = interval(config.read().await.heal_interval);
loop {
let event_driven_scheduler_enable = config.read().await.event_driven_scheduler_enable;
tokio::select! {
_ = cancel_token.cancelled() => {
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_SCHEDULER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
state = "shutdown",
"Heal scheduler stopped"
);
break;
}
_ = notify.notified(), if event_driven_scheduler_enable => {
Self::process_heal_queue(HealQueueContext {
heal_queue: &heal_queue,
active_heals: &active_heals,
completed_heals: &completed_heals,
retrying_heals: &retrying_heals,
replacement_recovery_anchors: &replacement_recovery_anchors,
config: &config,
statistics: &statistics,
storage: &storage,
notify: &notify,
cancel_token: &cancel_token,
workload_provider: &workload_provider,
})
.await;
}
_ = interval.tick() => {
Self::process_heal_queue(HealQueueContext {
heal_queue: &heal_queue,
active_heals: &active_heals,
completed_heals: &completed_heals,
retrying_heals: &retrying_heals,
replacement_recovery_anchors: &replacement_recovery_anchors,
config: &config,
statistics: &statistics,
storage: &storage,
notify: &notify,
cancel_token: &cancel_token,
workload_provider: &workload_provider,
})
.await;
}
}
}
});
Ok(())
}
/// Process heal queue
/// Processes multiple tasks per cycle when capacity allows and queue has high-priority items
pub(super) async fn process_heal_queue(context: HealQueueContext<'_>) {
let HealQueueContext {
heal_queue,
active_heals,
completed_heals,
retrying_heals,
replacement_recovery_anchors,
config,
statistics,
storage,
notify,
cancel_token,
workload_provider,
} = context;
let config = config.read().await;
let mainline_pressure = Self::mainline_throttle_active(&config, workload_provider);
let mut active_heals_guard = active_heals.lock().await;
publish_active_heal_count(&active_heals_guard);
// Check if new heal tasks can be started
let active_count = active_heals_guard.len();
if active_count >= config.max_concurrent_heals {
return;
}
// Calculate how many tasks we can start this cycle
let available_slots = config.max_concurrent_heals - active_count;
let mut queue = heal_queue.lock().await;
let queue_len = queue.len();
publish_heal_queue_length(&queue);
if queue_len == 0 {
return;
}
let mut running_per_set = running_heal_set_counts(&active_heals_guard);
let mut tasks_started = 0usize;
let mut delayed_by_mainline_throttle = false;
for _ in 0..available_slots {
let selected_request = if config.set_bulkhead_enable || mainline_pressure.is_some() {
let max_concurrent_per_set = config.max_concurrent_per_set;
let (selected_request, skipped_sets) = queue.pop_runnable_with_skips(
|request| {
let set_allowed = !config.set_bulkhead_enable
|| can_schedule_request(request, &running_per_set, max_concurrent_per_set);
let mainline_allowed = mainline_pressure.is_none() || Self::request_bypasses_mainline_throttle(request);
set_allowed && mainline_allowed
},
|request| heal_request_set_key(request).map(|_| heal_request_set_metric_label(request)),
);
for skipped_set in skipped_sets {
record_scheduler_skip(&skipped_set);
}
selected_request
} else {
queue.pop_next()
};
if let Some(mut request) = selected_request {
request.options.timeout.get_or_insert(config.task_timeout);
let task_priority = request.priority;
let task_type_label = heal_request_type_label(&request).to_string();
let task_set_label = heal_request_set_metric_label(&request);
if config.set_bulkhead_enable
&& let Some(set_key) = heal_request_set_key(&request)
{
*running_per_set.entry(set_key).or_insert(0) += 1;
}
let replacement_resume_endpoint = replacement_recovery_anchors
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.get(&request.id)
.cloned();
let task = Arc::new(HealTask::from_replacement_recovery_request(
request,
storage.clone(),
replacement_resume_endpoint,
));
let task_id = task.id.clone();
active_heals_guard.insert(task_id.clone(), task.clone());
publish_active_heal_count(&active_heals_guard);
update_task_running_metric_for_task(&active_heals_guard, task.as_ref());
let active_heals_clone = active_heals.clone();
let heal_queue_clone = heal_queue.clone();
let completed_heals_clone = completed_heals.clone();
let retrying_heals_clone = retrying_heals.clone();
let replacement_recovery_anchors_clone = replacement_recovery_anchors.clone();
let statistics_clone = statistics.clone();
let notify_clone = notify.clone();
let manager_cancel_token = cancel_token.clone();
let task_type_label_for_spawn = task_type_label.clone();
let task_set_label_for_spawn = task_set_label.clone();
let config_for_spawn = config.clone();
// start heal task
tokio::spawn(async move {
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_SCHEDULER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
task_id,
priority = ?task_priority,
heal_type = %task_type_label_for_spawn,
set = %task_set_label_for_spawn,
state = "task_started",
"Heal scheduler task started"
);
let result = task.execute().await;
let retry_request = retry_request_for_result_with_budget(task.as_ref(), &result).await;
match &result {
Ok(_) => {
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_SCHEDULER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
task_id,
heal_type = %task_type_label_for_spawn,
set = %task_set_label_for_spawn,
state = "task_completed",
"Heal scheduler task completed"
);
}
Err(e) => {
let will_retry = retry_request.is_some();
if will_retry {
demote_to_debug_when!(task.heal_type.is_per_object(), warn, target: "rustfs::heal::manager", {
event = EVENT_HEAL_SCHEDULER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
task_id,
heal_type = %task_type_label_for_spawn,
set = %task_set_label_for_spawn,
state = "task_retrying",
retry_attempt = task.retry_attempts.saturating_add(1),
error = %e,
"Heal scheduler task retrying"
});
} else {
error!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_SCHEDULER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
task_id,
heal_type = %task_type_label_for_spawn,
set = %task_set_label_for_spawn,
state = "task_failed",
error = %e,
"Heal scheduler task failed"
);
}
}
}
let retry_request_for_status = retry_request.as_ref().map(|(request, _, error)| HealTaskStatus::Retrying {
error: error.clone(),
retry_attempt: request.retry_attempts,
});
let retry_request_for_queue = retry_request;
let retry_cancel_token = retry_request_for_queue.as_ref().map(|_| CancellationToken::new());
if retry_request_for_queue.is_none() {
replacement_recovery_anchors_clone
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.remove(&task_id);
}
let mut active_heals_guard = active_heals_clone.lock().await;
// Keep retry ownership continuous: status snapshots acquire
// these locks in the same active -> retrying order.
let mut retrying_heals_guard = if let (Some((request, _, error)), Some(cancel_token)) =
(retry_request_for_queue.as_ref(), retry_cancel_token.as_ref())
{
let mut retrying = retrying_heals_clone.lock().await;
if active_heals_guard.contains_key(&task_id) {
retrying.insert(
request.id.clone(),
RetryingHeal {
request: request.clone(),
error: error.clone(),
cancel_token: cancel_token.clone(),
},
);
#[cfg(test)]
pause_retry_ownership_transition(&task_id, false).await;
}
Some(retrying)
} else {
None
};
let completed_task = active_heals_guard.remove(&task_id);
if let Some(completed_task) = completed_task.as_ref() {
publish_active_heal_count(&active_heals_guard);
update_task_running_metric_for_task(&active_heals_guard, completed_task.as_ref());
}
let active_count = active_heals_guard.len();
drop(retrying_heals_guard.take());
drop(active_heals_guard);
if let Some(completed_task) = completed_task {
let completed_status = if let Some(status) = retry_request_for_status {
status
} else {
completed_task.get_status().await
};
let completed_progress = completed_task.get_progress().await;
// Single snapshot of the retained window: the task is
// finished and already off the active map, so there is
// no concurrent writer to race with.
let seqed_items = completed_task.get_seqed_result_items().await;
let (next_seq, min_seq) = completed_task.result_seq_cursors();
let completed_status_entry = CompletedHealStatus {
heal_type: completed_task.heal_type.clone(),
status: completed_status.clone(),
result_items_truncated: completed_task.result_items_truncated(),
completed_at: SystemTime::now(),
seqed_items,
next_seq,
min_seq,
};
let mut completed_heals_guard = completed_heals_clone.lock().await;
prune_completed_heal_statuses(&mut completed_heals_guard);
completed_heals_guard.insert(task_id.clone(), Arc::new(completed_status_entry));
// update statistics
let mut stats = statistics_clone.write().await;
match completed_status {
HealTaskStatus::Completed => {
stats.update_task_completion(true);
stats.add_healed_objects(completed_progress.objects_healed, completed_progress.bytes_processed);
}
HealTaskStatus::Retrying { .. } => {}
_ => {
stats.update_task_completion(false);
}
}
stats.update_running_tasks(usize_to_u64_saturated(active_count));
}
if let (Some((retry_request, retry_delay, retry_error)), Some(retry_cancel_token)) =
(retry_request_for_queue, retry_cancel_token)
{
let retry_request_id = retry_request.id.clone();
let retry_attempt = retry_request.retry_attempts;
let retry_key = PriorityHealQueue::make_dedup_key(&retry_request);
let retry_priority = retry_request.priority;
let retry_active_heals = active_heals_clone.clone();
let retry_heal_queue = heal_queue_clone.clone();
let retrying_heals_for_spawn = retrying_heals_clone.clone();
let retry_completed_heals = completed_heals_clone.clone();
let retry_notify = notify_clone.clone();
let retry_manager_cancel_token = manager_cancel_token.clone();
let retry_config = config_for_spawn.clone();
tokio::spawn(async move {
loop {
tokio::select! {
_ = retry_cancel_token.cancelled() => {
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %retry_request_id,
priority = ?retry_priority,
retry_attempt,
result = "retry_cancelled",
"Heal retry admission decided"
);
return;
}
_ = retry_manager_cancel_token.cancelled() => {
retrying_heals_for_spawn.lock().await.remove(&retry_request_id);
return;
}
_ = sleep(retry_delay) => {}
}
{
let retrying_heals_guard = retrying_heals_for_spawn.lock().await;
if !retrying_heals_guard.contains_key(&retry_request_id) {
return;
}
}
let active_duplicate = {
let active_heals_guard = retry_active_heals.lock().await;
active_heals_contains_dedup_key(&active_heals_guard, &retry_key)
};
if active_duplicate {
retrying_heals_for_spawn.lock().await.remove(&retry_request_id);
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %retry_request_id,
priority = ?retry_priority,
retry_attempt,
result = "retry_merged_active_duplicate",
"Heal retry admission decided"
);
return;
}
let mut queue = retry_heal_queue.lock().await;
let admission =
Self::admit_request_to_queue(&mut queue, retry_request.clone(), &retry_config, "retry");
let should_notify = matches!(admission, HealAdmissionResult::Accepted)
&& retry_config.event_driven_scheduler_enable;
match admission {
HealAdmissionResult::Accepted => {
// Transfer ownership while holding queue -> retrying,
// matching operations_snapshot's lock order.
#[cfg(test)]
pause_retry_ownership_transition(&retry_request_id, true).await;
retrying_heals_for_spawn.lock().await.remove(&retry_request_id);
drop(queue);
retry_completed_heals.lock().await.remove(&retry_request_id);
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %retry_request_id,
priority = ?retry_priority,
retry_attempt,
retry_delay_ms = retry_delay.as_millis(),
error = %retry_error,
result = "retry_enqueued",
"Heal retry admission decided"
);
if should_notify {
retry_notify.notify_one();
}
return;
}
HealAdmissionResult::Merged => {
retrying_heals_for_spawn.lock().await.remove(&retry_request_id);
drop(queue);
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %retry_request_id,
priority = ?retry_priority,
retry_attempt,
result = "retry_merged_duplicate",
"Heal retry admission decided"
);
return;
}
HealAdmissionResult::Full => {
// admit_request_to_queue already logged the
// rejection (context = "retry"); this repeats
// every backoff cycle while the queue stays
// full, so keep it at debug!.
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %retry_request_id,
priority = ?retry_priority,
retry_attempt,
result = "retry_rejected_full",
"Heal retry admission decided"
);
}
HealAdmissionResult::Dropped(reason) => {
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %retry_request_id,
priority = ?retry_priority,
retry_attempt,
reason = reason.as_str(),
result = "retry_dropped",
"Heal retry admission decided"
);
}
}
}
});
}
notify_clone.notify_one();
});
tasks_started += 1;
} else {
delayed_by_mainline_throttle = mainline_pressure.is_some();
break;
}
}
// Update statistics for all started tasks
let mut stats = statistics.write().await;
stats.total_tasks += tasks_started as u64;
stats.update_running_tasks(active_heals_guard.len() as u64);
publish_active_heal_count(&active_heals_guard);
publish_heal_queue_length(&queue);
if delayed_by_mainline_throttle && let Some(pressure) = mainline_pressure {
Self::record_mainline_throttle_delay(pressure, &config);
Self::schedule_mainline_throttle_recheck(notify.clone(), config.mainline_max_sleep);
}
// Log queue status if items remain
if !queue.is_empty() {
let remaining = queue.len();
if remaining > 10 {
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
queue_len = remaining,
active_tasks = active_heals_guard.len(),
state = "backlog_high",
"Heal queue backlog high"
);
}
}
}
}
pub(super) fn heal_request_set_key(request: &HealRequest) -> Option<String> {
match &request.heal_type {
HealType::ErasureSet { set_disk_id, .. } => Some(set_disk_id.clone()),
HealType::Object { .. } => heal_options_set_key(&request.options),
_ => None,
}
}
pub(super) fn heal_options_set_key(options: &HealOptions) -> Option<String> {
match (options.pool_index, options.set_index) {
(Some(pool), Some(set)) => Some(format!("pool_{pool}_set_{set}")),
_ => None,
}
}
pub(super) fn heal_request_type_label(request: &HealRequest) -> &'static str {
match &request.heal_type {
HealType::Cluster => "cluster",
HealType::Object { .. } => "object",
HealType::Bucket { .. } => "bucket",
HealType::Prefix { .. } => "prefix",
HealType::ErasureSet { .. } => "erasure_set",
HealType::Metadata { .. } => "metadata",
HealType::ECDecode { .. } => "ec_decode",
}
}
pub(super) fn heal_request_set_metric_label(request: &HealRequest) -> String {
heal_request_set_key(request).unwrap_or_else(|| match (request.options.pool_index, request.options.set_index) {
(Some(pool), Some(set)) => format!("pool_{pool}_set_{set}"),
_ => "global".to_string(),
})
}
pub(super) fn record_scheduler_skip(set_label: &str) {
counter!(
"rustfs_heal_scheduler_skip_total",
"reason" => "set_limit".to_string(),
"set" => set_label.to_string()
)
.increment(1);
}
pub(super) fn update_task_running_metric_for_task(active_heals: &HashMap<String, Arc<HealTask>>, task: &HealTask) {
let type_label = task.metric_type_label();
let set_label = task.metric_set_label();
let count = active_heals
.values()
.filter(|active_task| active_task.metric_type_label() == type_label && active_task.metric_set_label() == set_label)
.count();
gauge!(
"rustfs_heal_task_running",
"type" => type_label.to_string(),
"set" => set_label
)
.set(count as f64);
}
pub(super) fn running_heal_set_counts(active_heals: &HashMap<String, Arc<HealTask>>) -> HashMap<String, usize> {
let mut running = HashMap::new();
for task in active_heals.values() {
if let Some(set_key) = heal_request_set_key_for_task(task) {
*running.entry(set_key).or_insert(0) += 1;
}
}
running
}
pub(super) fn heal_request_set_key_for_task(task: &HealTask) -> Option<String> {
match &task.heal_type {
HealType::ErasureSet { set_disk_id, .. } => Some(set_disk_id.clone()),
HealType::Object { .. } => heal_options_set_key(&task.options),
_ => None,
}
}
pub(super) fn prune_completed_heal_statuses(completed_heals: &mut HashMap<String, Arc<CompletedHealStatus>>) {
let Ok(now) = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) else {
return;
};
completed_heals.retain(|_, completed| {
completed
.completed_at
.duration_since(SystemTime::UNIX_EPOCH)
.map(|completed_at| now.saturating_sub(completed_at) <= KEEP_HEAL_TASK_STATUS_DURATION)
.unwrap_or(false)
});
}
pub(super) fn can_schedule_request(
request: &HealRequest,
running_per_set: &HashMap<String, usize>,
max_concurrent_per_set: usize,
) -> bool {
match heal_request_set_key(request) {
Some(set_key) => running_per_set.get(&set_key).copied().unwrap_or(0) < max_concurrent_per_set,
None => true,
}
}
crates/heal/src/heal/manager/tests.rs
+2971
View File
File diff suppressed because it is too large Load Diff
crates/heal/src/heal/manager/unclean_shutdown.rs
+390
View File
@@ -0,0 +1,390 @@
// Copyright 2024 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.
/// Unclean-shutdown recovery: durable replacement-intent discovery and healing-marker rewrite.
use super::*;
pub(super) fn durable_replacement_recovery_is_due(state: &ResumeState, task_id: &str) -> bool {
state.replacement_generation.as_deref() == Some(task_id)
&& !state.replacement_targets.is_empty()
&& ((!state.completed
&& matches!(state.replacement_phase, ReplacementPhase::Intent | ReplacementPhase::Rebuilding)
&& state.retry_count >= state.max_retries)
|| (state.completed
&& matches!(state.replacement_phase, ReplacementPhase::Verified | ReplacementPhase::CleanupPending)))
}
pub(super) fn replacement_discovery_error_is_expected_for_deferred_endpoint(
error: &Error,
endpoint: &str,
deferred_replacement_endpoints: &HashSet<String>,
) -> bool {
matches!(error, Error::Disk(DiskError::UnformattedDisk)) && deferred_replacement_endpoints.contains(endpoint)
}
pub(super) fn unblock_replacement_recovery_sets_after_validation(
blocked_sets: &mut HashSet<String>,
retry_succeeded: HashSet<String>,
retry_failed: &HashSet<String>,
) {
for set_disk_id in retry_succeeded {
if !retry_failed.contains(&set_disk_id) {
blocked_sets.remove(&set_disk_id);
}
}
}
impl HealManager {
/// Detect whether the previous run ended without a clean shutdown and, if so,
/// enqueue a full erasure-set heal for every local set. Also (re)writes the
/// marker for the current run; [`super::super::clear_unclean_shutdown_markers`]
/// removes it again during graceful shutdown. Best-effort: failures only log.
pub(super) async fn process_unclean_shutdown(&self) {
let mut unclean = false;
let mut set_disk_ids = HashSet::new();
let mut replacement_intents = HashMap::<String, (String, Vec<String>, Vec<String>, String)>::new();
let mut replacement_restarts = HashMap::<String, (String, Vec<String>)>::new();
let mut conflicted_replacement_sets = HashSet::new();
{
let local_disks = {
let local_disk_map = local_disk_map_read().await;
local_disk_map.values().flatten().cloned().collect::<Vec<_>>()
};
for disk in &local_disks {
let endpoint = disk.endpoint();
match disk
.read_all(super::super::RUSTFS_META_BUCKET, super::super::UNCLEAN_SHUTDOWN_MARKER_PATH)
.await
{
Ok(_) => unclean = true,
Err(DiskError::FileNotFound) | Err(DiskError::VolumeNotFound) => {}
Err(err) => {
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
endpoint = %endpoint,
error = ?err,
"Unclean-shutdown marker check failed"
);
}
}
let marker = SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs().to_string())
.unwrap_or_default();
if let Err(err) = disk
.write_all(
super::super::RUSTFS_META_BUCKET,
super::super::UNCLEAN_SHUTDOWN_MARKER_PATH,
marker.into(),
)
.await
{
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
endpoint = %endpoint,
error = ?err,
"Unclean-shutdown marker write failed"
);
}
let disk_set_disk_id = crate::heal::utils::format_set_disk_id_from_i32(endpoint.pool_idx, endpoint.set_idx);
if let Some(set_disk_id) = &disk_set_disk_id {
set_disk_ids.insert(set_disk_id.clone());
}
// Legacy flat records are inspected only while starting. The
// periodic scanner lists the dedicated replacement directory.
if let Err(error) = ResumeUtils::migrate_legacy_replacement_records(disk).await {
if let Some(set_disk_id) = &disk_set_disk_id {
self.block_replacement_recovery_set(set_disk_id);
}
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
endpoint = %endpoint,
error = %error,
"Legacy replacement recovery migration failed"
);
}
let replacement_task_ids = match ResumeUtils::get_replacement_intent_tasks(disk).await {
Ok(task_ids) => task_ids,
Err(error) => {
if let Some(set_disk_id) = &disk_set_disk_id {
self.block_replacement_recovery_set(set_disk_id);
}
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
endpoint = %endpoint,
error = %error,
"Replacement recovery discovery failed"
);
continue;
}
};
for task_id in replacement_task_ids {
let manager = match ResumeManager::load_replacement_intent(disk.clone(), &task_id).await {
Ok(manager) => manager,
Err(error) => {
if let Some(set_disk_id) = &disk_set_disk_id {
self.block_replacement_recovery_set(set_disk_id);
}
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
endpoint = %endpoint,
task_id,
error = %error,
"Replacement recovery intent load failed"
);
continue;
}
};
let state = manager.get_state().await;
let active_replacement = !state.completed
&& matches!(state.replacement_phase, ReplacementPhase::Intent | ReplacementPhase::Rebuilding);
let verified_replacement = state.completed
&& matches!(state.replacement_phase, ReplacementPhase::Verified | ReplacementPhase::CleanupPending);
if (active_replacement || verified_replacement)
&& state.replacement_generation.as_deref() == Some(task_id.as_str())
&& !state.replacement_targets.is_empty()
{
if matches!(state.replacement_phase, ReplacementPhase::CleanupPending) {
replacement_intents.entry(task_id).or_insert((
state.set_disk_id,
state.replacement_targets,
state.replacement_buckets,
endpoint.to_string(),
));
continue;
}
match self.storage.replacement_target_identities(&state.replacement_targets).await {
Ok(identities) if identities == state.replacement_target_identities => {
let resume_endpoint = endpoint.to_string();
match replacement_intents.entry(task_id) {
std::collections::hash_map::Entry::Vacant(entry) => {
entry.insert((
state.set_disk_id,
state.replacement_targets,
state.replacement_buckets,
resume_endpoint,
));
}
std::collections::hash_map::Entry::Occupied(entry) => {
let (existing_set_disk_id, existing_targets, existing_buckets, existing_anchor) =
entry.get();
if existing_set_disk_id != &state.set_disk_id
|| existing_targets != &state.replacement_targets
|| existing_buckets != &state.replacement_buckets
|| existing_anchor != &resume_endpoint
{
conflicted_replacement_sets.insert(state.set_disk_id.clone());
self.block_replacement_recovery_set(&state.set_disk_id);
}
}
}
}
Ok(_) => {
if manager.abandon_replacement_intent().await.is_ok() {
replacement_restarts
.entry(task_id)
.or_insert((state.set_disk_id, state.replacement_targets));
}
}
Err(_) => {}
}
}
}
}
}
if !unclean && replacement_intents.is_empty() && replacement_restarts.is_empty() {
return;
}
let mut recovery_by_set = HashMap::<String, Vec<(Option<String>, Vec<String>, Vec<String>, Option<String>)>>::new();
for (task_id, (set_disk_id, heal_endpoints, buckets, resume_endpoint)) in replacement_intents {
recovery_by_set
.entry(set_disk_id)
.or_default()
.push((Some(task_id), heal_endpoints, buckets, Some(resume_endpoint)));
}
for (_abandoned_task_id, (set_disk_id, heal_endpoints)) in replacement_restarts {
recovery_by_set
.entry(set_disk_id)
.or_default()
.push((None, heal_endpoints, Vec::new(), None));
}
for (set_disk_id, mut recoveries) in recovery_by_set {
let Ok((pool_index, set_index)) = crate::heal::utils::parse_set_disk_id(&set_disk_id) else {
continue;
};
if self.replacement_recovery_set_is_blocked(&set_disk_id) {
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
set_disk_id,
recovery_count = recoveries.len(),
"Replacement recovery deferred because durable recovery validation is blocked"
);
continue;
}
if conflicted_replacement_sets.contains(&set_disk_id) || recoveries.len() != 1 {
self.block_replacement_recovery_set(&set_disk_id);
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
set_disk_id,
recovery_count = recoveries.len(),
"Replacement recovery deferred because multiple durable generations exist"
);
continue;
}
let reuse_single_generation = recoveries.len() == 1 && recoveries[0].0.is_some();
let mut heal_endpoints = recoveries
.iter_mut()
.flat_map(|(_, targets, _, _)| std::mem::take(targets))
.collect::<Vec<_>>();
heal_endpoints.sort_unstable();
heal_endpoints.dedup();
let buckets = if reuse_single_generation {
std::mem::take(&mut recoveries[0].2)
} else {
Vec::new()
};
let mut req = HealRequest::new(
HealType::ErasureSet {
buckets,
set_disk_id: set_disk_id.clone(),
},
HealOptions {
pool_index: Some(pool_index),
set_index: Some(set_index),
timeout: None,
..HealOptions::default()
},
HealPriority::Low,
);
if reuse_single_generation && let Some(task_id) = recoveries[0].0.take() {
req.id = task_id;
}
let recovery_anchor = reuse_single_generation.then(|| recoveries[0].3.take()).flatten();
req.source = HealRequestSource::AutoHeal;
req.heal_endpoints = heal_endpoints;
let request_id = req.id.clone();
if let Some(anchor) = &recovery_anchor {
self.replacement_recovery_anchors
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.insert(request_id.clone(), anchor.clone());
}
match self.submit_heal_request(req).await {
Ok(HealAdmissionResult::Accepted) => {}
Ok(_) => {
self.replacement_recovery_anchors
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.remove(&request_id);
}
Err(err) => {
self.replacement_recovery_anchors
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.remove(&request_id);
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
set_disk_id,
error = %err,
"Replacement recovery enqueue failed"
);
}
}
}
if !unclean || set_disk_ids.is_empty() {
return;
}
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
set_count = set_disk_ids.len(),
"Unclean shutdown detected; scheduling erasure-set heal for local sets"
);
let buckets = match self.storage.list_buckets().await {
Ok(buckets) => buckets.iter().map(|b| b.name.clone()).collect::<Vec<String>>(),
Err(err) => {
error!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
error = %err,
"Unclean-shutdown heal skipped: bucket listing failed"
);
return;
}
};
for set_disk_id in set_disk_ids {
let mut req = HealRequest::new(
HealType::ErasureSet {
buckets: buckets.clone(),
set_disk_id: set_disk_id.clone(),
},
HealOptions {
timeout: None,
..HealOptions::default()
},
HealPriority::Low,
);
req.source = HealRequestSource::AutoHeal;
if let Err(err) = self.submit_heal_request(req).await {
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_UNCLEAN_SHUTDOWN,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
set_disk_id,
error = %err,
"Unclean-shutdown heal enqueue failed"
);
}
}
}
}
scripts/check_logging_guardrails.sh
+4 -2
View File
@@ -1017,9 +1017,11 @@ if rg -n -F 'target: "rustfs::server::http"' rustfs/src/server/layer.rs >/dev/nu
exit 1
fi
demoted_admission_sites="$(rg -c -F 'demote_to_debug_when!(' crates/heal/src/heal/manager.rs || echo 0)"
# manager.rs and its manager/ child modules are one logical module tree since
# the queue/scheduler split; count the demoted sites across the whole tree.
demoted_admission_sites="$(cat crates/heal/src/heal/manager.rs crates/heal/src/heal/manager/*.rs 2>/dev/null | rg -c -F 'demote_to_debug_when!(' || echo 0)"
if [[ "$demoted_admission_sites" -lt 6 ]]; then
echo "❌ logging guardrail violation: heal queue admission/scheduler warns for per-object requests must stay level-split via demote_to_debug_when! (expected >= 6 sites in crates/heal/src/heal/manager.rs, found $demoted_admission_sites)" >&2
echo "❌ logging guardrail violation: heal queue admission/scheduler warns for per-object requests must stay level-split via demote_to_debug_when! (expected >= 6 total sites in the manager module tree, found $demoted_admission_sites)" >&2
exit 1
fi