mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-19 11:06:17 +00:00
Fix: Implement priority-based heal queue with comprehensive diagnostic logging (#884)
* Initial plan * Implement priority-based heal queue with deduplication Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> * Apply cargo fmt formatting fixes Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> * Add queue monitoring, better error handling, and adaptive processing - Add priority-based queue statistics tracking - Implement queue capacity warnings (>80% full) - Process multiple tasks per cycle when capacity allows - Add proper error logging for failed heal request submissions - Add Hash trait to HealPriority for HashMap support - Improve observability with detailed queue status logs Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> * Add comprehensive diagnostic logging to heal flow - Add detailed logging for shard availability checks during reads - Log successful heal request submissions - Add detailed logging in heal_object for disk checking - Log each disk's heal decision with reason and endpoint - Add logging for heal writing and renaming phases - Improve variable naming (nil_count -> available_shards) - Add success confirmation messages for renamed healed data These logs will help diagnose where the heal flow is failing. Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> * fix * feat(heal): enhance robustness, safety, and observability of heal process - **Logging & Observability**: - Upgrade critical heal logs from [info](/crates/e2e_test/src/reliant/node_interact_test.rs:196:0-213:1) to `warn` for better visibility. - Implement structured logging with `tracing` fields for machine readability. - Add `#[tracing::instrument]` to [HealTask](c/crates/ahm/src/heal/task.rs:182:0-205:1) and [SetDisks](/crates/ecstore/src/set_disk.rs:120:0-131:1) methods for automatic context propagation. - **Robustness**: - Add exponential backoff retry (3 attempts) for acquiring write locks in [heal_object](/crates/ahm/src/heal/storage.rs:438:4-460:5) to handle contention. - Handle [rename_data](/crates/ecstore/src/set_disk.rs:392:4-516:5) failures gracefully by preserving temporary files instead of forcing deletion, preventing potential data loss. - **Data Safety**: - Fix [object_exists](/crates/ahm/src/heal/storage.rs:395:4-412:5) to propagate IO errors instead of treating them as "object not found". - Update [ErasureSetHealer](/crates/ahm/src/heal/erasure_healer.rs:28:0-33:1) to mark objects as failed rather than skipped when existence checks error, ensuring they are tracked for retry. * fix * fmt * improve code for heal_object * fix * fix * fix --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
Copilot
@@ -49,8 +49,9 @@ impl ErasureSetHealer {
|
||||
}
|
||||
|
||||
/// execute erasure set heal with resume
|
||||
#[tracing::instrument(skip(self, buckets), fields(set_disk_id = %set_disk_id, bucket_count = buckets.len()))]
|
||||
pub async fn heal_erasure_set(&self, buckets: &[String], set_disk_id: &str) -> Result<()> {
|
||||
info!("Starting erasure set heal for {} buckets on set disk {}", buckets.len(), set_disk_id);
|
||||
info!("Starting erasure set heal");
|
||||
|
||||
// 1. generate or get task id
|
||||
let task_id = self.get_or_create_task_id(set_disk_id).await?;
|
||||
@@ -231,6 +232,7 @@ impl ErasureSetHealer {
|
||||
|
||||
/// heal single bucket with resume
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
#[tracing::instrument(skip(self, current_object_index, processed_objects, successful_objects, failed_objects, _skipped_objects, resume_manager, checkpoint_manager), fields(bucket = %bucket, bucket_index = bucket_index))]
|
||||
async fn heal_bucket_with_resume(
|
||||
&self,
|
||||
bucket: &str,
|
||||
@@ -243,7 +245,7 @@ impl ErasureSetHealer {
|
||||
resume_manager: &ResumeManager,
|
||||
checkpoint_manager: &CheckpointManager,
|
||||
) -> Result<()> {
|
||||
info!(target: "rustfs:ahm:heal_bucket_with_resume" ,"Starting heal for bucket: {} from object index {}", bucket, current_object_index);
|
||||
info!(target: "rustfs:ahm:heal_bucket_with_resume" ,"Starting heal for bucket from object index {}", current_object_index);
|
||||
|
||||
// 1. get bucket info
|
||||
let _bucket_info = match self.storage.get_bucket_info(bucket).await? {
|
||||
@@ -273,7 +275,9 @@ impl ErasureSetHealer {
|
||||
let object_exists = match self.storage.object_exists(bucket, object).await {
|
||||
Ok(exists) => exists,
|
||||
Err(e) => {
|
||||
warn!("Failed to check existence of {}/{}: {}, skipping", bucket, object, e);
|
||||
warn!("Failed to check existence of {}/{}: {}, marking as failed", bucket, object, e);
|
||||
*failed_objects += 1;
|
||||
checkpoint_manager.add_failed_object(object.clone()).await?;
|
||||
*current_object_index = obj_idx + 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
+592
-53
@@ -22,7 +22,7 @@ use rustfs_ecstore::disk::DiskAPI;
|
||||
use rustfs_ecstore::disk::error::DiskError;
|
||||
use rustfs_ecstore::global::GLOBAL_LOCAL_DISK_MAP;
|
||||
use std::{
|
||||
collections::{HashMap, VecDeque},
|
||||
collections::{BinaryHeap, HashMap, HashSet},
|
||||
sync::Arc,
|
||||
time::{Duration, SystemTime},
|
||||
};
|
||||
@@ -33,6 +33,151 @@ use tokio::{
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{error, info, warn};
|
||||
|
||||
/// Priority queue wrapper for heal requests
|
||||
/// Uses BinaryHeap for priority-based ordering while maintaining FIFO for same-priority items
|
||||
#[derive(Debug)]
|
||||
struct PriorityHealQueue {
|
||||
/// Heap of (priority, sequence, request) tuples
|
||||
heap: BinaryHeap<PriorityQueueItem>,
|
||||
/// Sequence counter for FIFO ordering within same priority
|
||||
sequence: u64,
|
||||
/// Set of request keys to prevent duplicates
|
||||
dedup_keys: HashSet<String>,
|
||||
}
|
||||
|
||||
/// Wrapper for heap items to implement proper ordering
|
||||
#[derive(Debug)]
|
||||
struct PriorityQueueItem {
|
||||
priority: HealPriority,
|
||||
sequence: u64,
|
||||
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))
|
||||
}
|
||||
}
|
||||
|
||||
impl PriorityHealQueue {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
heap: BinaryHeap::new(),
|
||||
sequence: 0,
|
||||
dedup_keys: HashSet::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn len(&self) -> usize {
|
||||
self.heap.len()
|
||||
}
|
||||
|
||||
fn is_empty(&self) -> bool {
|
||||
self.heap.is_empty()
|
||||
}
|
||||
|
||||
fn push(&mut self, request: HealRequest) -> bool {
|
||||
let key = Self::make_dedup_key(&request);
|
||||
|
||||
// Check for duplicates
|
||||
if self.dedup_keys.contains(&key) {
|
||||
return false; // Duplicate request, don't add
|
||||
}
|
||||
|
||||
self.dedup_keys.insert(key);
|
||||
self.sequence += 1;
|
||||
self.heap.push(PriorityQueueItem {
|
||||
priority: request.priority,
|
||||
sequence: self.sequence,
|
||||
request,
|
||||
});
|
||||
true
|
||||
}
|
||||
|
||||
/// Get statistics about queue contents by priority
|
||||
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
|
||||
}
|
||||
|
||||
fn pop(&mut self) -> Option<HealRequest> {
|
||||
self.heap.pop().map(|item| {
|
||||
let key = Self::make_dedup_key(&item.request);
|
||||
self.dedup_keys.remove(&key);
|
||||
item.request
|
||||
})
|
||||
}
|
||||
|
||||
/// Create a deduplication key from a heal request
|
||||
fn make_dedup_key(request: &HealRequest) -> String {
|
||||
match &request.heal_type {
|
||||
HealType::Object {
|
||||
bucket,
|
||||
object,
|
||||
version_id,
|
||||
} => {
|
||||
format!("object:{}:{}:{}", bucket, object, version_id.as_deref().unwrap_or(""))
|
||||
}
|
||||
HealType::Bucket { bucket } => {
|
||||
format!("bucket:{}", bucket)
|
||||
}
|
||||
HealType::ErasureSet { set_disk_id, .. } => {
|
||||
format!("erasure_set:{}", set_disk_id)
|
||||
}
|
||||
HealType::Metadata { bucket, object } => {
|
||||
format!("metadata:{}:{}", bucket, object)
|
||||
}
|
||||
HealType::MRF { meta_path } => {
|
||||
format!("mrf:{}", meta_path)
|
||||
}
|
||||
HealType::ECDecode {
|
||||
bucket,
|
||||
object,
|
||||
version_id,
|
||||
} => {
|
||||
format!("ecdecode:{}:{}:{}", bucket, object, version_id.as_deref().unwrap_or(""))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Check if a request with the same key already exists in the queue
|
||||
#[allow(dead_code)]
|
||||
fn contains_key(&self, request: &HealRequest) -> bool {
|
||||
let key = Self::make_dedup_key(request);
|
||||
self.dedup_keys.contains(&key)
|
||||
}
|
||||
|
||||
/// Check if an erasure set heal request for a specific set_disk_id exists
|
||||
fn contains_erasure_set(&self, set_disk_id: &str) -> bool {
|
||||
let key = format!("erasure_set:{}", set_disk_id);
|
||||
self.dedup_keys.contains(&key)
|
||||
}
|
||||
}
|
||||
|
||||
/// Heal config
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct HealConfig {
|
||||
@@ -85,8 +230,8 @@ pub struct HealManager {
|
||||
state: Arc<RwLock<HealState>>,
|
||||
/// Active heal tasks
|
||||
active_heals: Arc<Mutex<HashMap<String, Arc<HealTask>>>>,
|
||||
/// Heal queue
|
||||
heal_queue: Arc<Mutex<VecDeque<HealRequest>>>,
|
||||
/// Heal queue (priority-based)
|
||||
heal_queue: Arc<Mutex<PriorityHealQueue>>,
|
||||
/// Storage layer interface
|
||||
storage: Arc<dyn HealStorageAPI>,
|
||||
/// Cancel token
|
||||
@@ -103,7 +248,7 @@ impl HealManager {
|
||||
config: Arc::new(RwLock::new(config)),
|
||||
state: Arc::new(RwLock::new(HealState::default())),
|
||||
active_heals: Arc::new(Mutex::new(HashMap::new())),
|
||||
heal_queue: Arc::new(Mutex::new(VecDeque::new())),
|
||||
heal_queue: Arc::new(Mutex::new(PriorityHealQueue::new())),
|
||||
storage,
|
||||
cancel_token: CancellationToken::new(),
|
||||
statistics: Arc::new(RwLock::new(HealStatistics::new())),
|
||||
@@ -161,17 +306,54 @@ impl HealManager {
|
||||
let config = self.config.read().await;
|
||||
let mut queue = self.heal_queue.lock().await;
|
||||
|
||||
if queue.len() >= config.queue_size {
|
||||
let queue_len = queue.len();
|
||||
let queue_capacity = config.queue_size;
|
||||
|
||||
if queue_len >= queue_capacity {
|
||||
return Err(Error::ConfigurationError {
|
||||
message: "Heal queue is full".to_string(),
|
||||
message: format!("Heal queue is full ({}/{})", queue_len, queue_capacity),
|
||||
});
|
||||
}
|
||||
|
||||
// Warn when queue is getting full (>80% capacity)
|
||||
let capacity_threshold = (queue_capacity as f64 * 0.8) as usize;
|
||||
if queue_len >= capacity_threshold {
|
||||
warn!(
|
||||
"Heal queue is {}% full ({}/{}). Consider increasing queue size or processing capacity.",
|
||||
(queue_len * 100) / queue_capacity,
|
||||
queue_len,
|
||||
queue_capacity
|
||||
);
|
||||
}
|
||||
|
||||
let request_id = request.id.clone();
|
||||
queue.push_back(request);
|
||||
let priority = request.priority;
|
||||
|
||||
// Try to push the request; if it's a duplicate, still return the request_id
|
||||
let is_new = queue.push(request);
|
||||
|
||||
// Log queue statistics periodically (when adding high/urgent priority items)
|
||||
if matches!(priority, HealPriority::High | HealPriority::Urgent) {
|
||||
let stats = queue.get_priority_stats();
|
||||
info!(
|
||||
"Heal queue stats after adding {:?} priority request: total={}, urgent={}, high={}, normal={}, low={}",
|
||||
priority,
|
||||
queue_len + 1,
|
||||
stats.get(&HealPriority::Urgent).unwrap_or(&0),
|
||||
stats.get(&HealPriority::High).unwrap_or(&0),
|
||||
stats.get(&HealPriority::Normal).unwrap_or(&0),
|
||||
stats.get(&HealPriority::Low).unwrap_or(&0)
|
||||
);
|
||||
}
|
||||
|
||||
drop(queue);
|
||||
|
||||
info!("Submitted heal request: {}", request_id);
|
||||
if is_new {
|
||||
info!("Submitted heal request: {} with priority: {:?}", request_id, priority);
|
||||
} else {
|
||||
info!("Heal request already queued (duplicate): {}", request_id);
|
||||
}
|
||||
|
||||
Ok(request_id)
|
||||
}
|
||||
|
||||
@@ -321,13 +503,7 @@ impl HealManager {
|
||||
let mut skip = false;
|
||||
{
|
||||
let queue = heal_queue.lock().await;
|
||||
if queue.iter().any(|req| {
|
||||
matches!(
|
||||
&req.heal_type,
|
||||
crate::heal::task::HealType::ErasureSet { set_disk_id: queued_id, .. }
|
||||
if queued_id == &set_disk_id
|
||||
)
|
||||
}) {
|
||||
if queue.contains_erasure_set(&set_disk_id) {
|
||||
skip = true;
|
||||
}
|
||||
}
|
||||
@@ -358,7 +534,7 @@ impl HealManager {
|
||||
HealPriority::Normal,
|
||||
);
|
||||
let mut queue = heal_queue.lock().await;
|
||||
queue.push_back(req);
|
||||
queue.push(req);
|
||||
info!("Enqueued auto erasure set heal for endpoint: {} (set_disk_id: {})", ep, set_disk_id);
|
||||
}
|
||||
}
|
||||
@@ -369,8 +545,9 @@ impl HealManager {
|
||||
}
|
||||
|
||||
/// Process heal queue
|
||||
/// Processes multiple tasks per cycle when capacity allows and queue has high-priority items
|
||||
async fn process_heal_queue(
|
||||
heal_queue: &Arc<Mutex<VecDeque<HealRequest>>>,
|
||||
heal_queue: &Arc<Mutex<PriorityHealQueue>>,
|
||||
active_heals: &Arc<Mutex<HashMap<String, Arc<HealTask>>>>,
|
||||
config: &Arc<RwLock<HealConfig>>,
|
||||
statistics: &Arc<RwLock<HealStatistics>>,
|
||||
@@ -379,51 +556,83 @@ impl HealManager {
|
||||
let config = config.read().await;
|
||||
let mut active_heals_guard = active_heals.lock().await;
|
||||
|
||||
// check if new heal tasks can be started
|
||||
if active_heals_guard.len() >= config.max_concurrent_heals {
|
||||
// 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;
|
||||
if let Some(request) = queue.pop_front() {
|
||||
let task = Arc::new(HealTask::from_request(request, storage.clone()));
|
||||
let task_id = task.id.clone();
|
||||
active_heals_guard.insert(task_id.clone(), task.clone());
|
||||
drop(active_heals_guard);
|
||||
let active_heals_clone = active_heals.clone();
|
||||
let statistics_clone = statistics.clone();
|
||||
let queue_len = queue.len();
|
||||
|
||||
// start heal task
|
||||
tokio::spawn(async move {
|
||||
info!("Starting heal task: {}", task_id);
|
||||
let result = task.execute().await;
|
||||
match result {
|
||||
Ok(_) => {
|
||||
info!("Heal task completed successfully: {}", task_id);
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Heal task failed: {} - {}", task_id, e);
|
||||
}
|
||||
}
|
||||
let mut active_heals_guard = active_heals_clone.lock().await;
|
||||
if let Some(completed_task) = active_heals_guard.remove(&task_id) {
|
||||
// update statistics
|
||||
let mut stats = statistics_clone.write().await;
|
||||
match completed_task.get_status().await {
|
||||
HealTaskStatus::Completed => {
|
||||
stats.update_task_completion(true);
|
||||
if queue_len == 0 {
|
||||
return;
|
||||
}
|
||||
|
||||
// Process multiple tasks if:
|
||||
// 1. We have available slots
|
||||
// 2. Queue is not empty
|
||||
// Prioritize urgent/high priority tasks by processing up to 2 tasks per cycle if available
|
||||
let tasks_to_process = if queue_len > 0 {
|
||||
std::cmp::min(available_slots, std::cmp::min(2, queue_len))
|
||||
} else {
|
||||
0
|
||||
};
|
||||
|
||||
for _ in 0..tasks_to_process {
|
||||
if let Some(request) = queue.pop() {
|
||||
let task_priority = request.priority;
|
||||
let task = Arc::new(HealTask::from_request(request, storage.clone()));
|
||||
let task_id = task.id.clone();
|
||||
active_heals_guard.insert(task_id.clone(), task.clone());
|
||||
let active_heals_clone = active_heals.clone();
|
||||
let statistics_clone = statistics.clone();
|
||||
|
||||
// start heal task
|
||||
tokio::spawn(async move {
|
||||
info!("Starting heal task: {} with priority: {:?}", task_id, task_priority);
|
||||
let result = task.execute().await;
|
||||
match result {
|
||||
Ok(_) => {
|
||||
info!("Heal task completed successfully: {}", task_id);
|
||||
}
|
||||
_ => {
|
||||
stats.update_task_completion(false);
|
||||
Err(e) => {
|
||||
error!("Heal task failed: {} - {}", task_id, e);
|
||||
}
|
||||
}
|
||||
stats.update_running_tasks(active_heals_guard.len() as u64);
|
||||
}
|
||||
});
|
||||
let mut active_heals_guard = active_heals_clone.lock().await;
|
||||
if let Some(completed_task) = active_heals_guard.remove(&task_id) {
|
||||
// update statistics
|
||||
let mut stats = statistics_clone.write().await;
|
||||
match completed_task.get_status().await {
|
||||
HealTaskStatus::Completed => {
|
||||
stats.update_task_completion(true);
|
||||
}
|
||||
_ => {
|
||||
stats.update_task_completion(false);
|
||||
}
|
||||
}
|
||||
stats.update_running_tasks(active_heals_guard.len() as u64);
|
||||
}
|
||||
});
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// update statistics
|
||||
let mut stats = statistics.write().await;
|
||||
stats.total_tasks += 1;
|
||||
// Update statistics for all started tasks
|
||||
let mut stats = statistics.write().await;
|
||||
stats.total_tasks += tasks_to_process as u64;
|
||||
|
||||
// Log queue status if items remain
|
||||
if !queue.is_empty() {
|
||||
let remaining = queue.len();
|
||||
if remaining > 10 {
|
||||
info!("Heal queue has {} pending requests, {} tasks active", remaining, active_heals_guard.len());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -438,3 +647,333 @@ impl std::fmt::Debug for HealManager {
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::heal::task::{HealOptions, HealPriority, HealRequest, HealType};
|
||||
|
||||
#[test]
|
||||
fn test_priority_queue_ordering() {
|
||||
let mut queue = PriorityHealQueue::new();
|
||||
|
||||
// Add requests with different priorities
|
||||
let low_req = HealRequest::new(
|
||||
HealType::Bucket {
|
||||
bucket: "bucket1".to_string(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Low,
|
||||
);
|
||||
|
||||
let normal_req = HealRequest::new(
|
||||
HealType::Bucket {
|
||||
bucket: "bucket2".to_string(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Normal,
|
||||
);
|
||||
|
||||
let high_req = HealRequest::new(
|
||||
HealType::Bucket {
|
||||
bucket: "bucket3".to_string(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::High,
|
||||
);
|
||||
|
||||
let urgent_req = HealRequest::new(
|
||||
HealType::Bucket {
|
||||
bucket: "bucket4".to_string(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Urgent,
|
||||
);
|
||||
|
||||
// Add in random order: low, high, normal, urgent
|
||||
assert!(queue.push(low_req));
|
||||
assert!(queue.push(high_req));
|
||||
assert!(queue.push(normal_req));
|
||||
assert!(queue.push(urgent_req));
|
||||
|
||||
assert_eq!(queue.len(), 4);
|
||||
|
||||
// Should pop in priority order: urgent, high, normal, low
|
||||
let popped1 = queue.pop().unwrap();
|
||||
assert_eq!(popped1.priority, HealPriority::Urgent);
|
||||
|
||||
let popped2 = queue.pop().unwrap();
|
||||
assert_eq!(popped2.priority, HealPriority::High);
|
||||
|
||||
let popped3 = queue.pop().unwrap();
|
||||
assert_eq!(popped3.priority, HealPriority::Normal);
|
||||
|
||||
let popped4 = queue.pop().unwrap();
|
||||
assert_eq!(popped4.priority, HealPriority::Low);
|
||||
|
||||
assert_eq!(queue.len(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_priority_queue_fifo_same_priority() {
|
||||
let mut queue = PriorityHealQueue::new();
|
||||
|
||||
// Add multiple requests with same priority
|
||||
let req1 = HealRequest::new(
|
||||
HealType::Bucket {
|
||||
bucket: "bucket1".to_string(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Normal,
|
||||
);
|
||||
|
||||
let req2 = HealRequest::new(
|
||||
HealType::Bucket {
|
||||
bucket: "bucket2".to_string(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Normal,
|
||||
);
|
||||
|
||||
let req3 = HealRequest::new(
|
||||
HealType::Bucket {
|
||||
bucket: "bucket3".to_string(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Normal,
|
||||
);
|
||||
|
||||
let id1 = req1.id.clone();
|
||||
let id2 = req2.id.clone();
|
||||
let id3 = req3.id.clone();
|
||||
|
||||
assert!(queue.push(req1));
|
||||
assert!(queue.push(req2));
|
||||
assert!(queue.push(req3));
|
||||
|
||||
// Should maintain FIFO order for same priority
|
||||
let popped1 = queue.pop().unwrap();
|
||||
assert_eq!(popped1.id, id1);
|
||||
|
||||
let popped2 = queue.pop().unwrap();
|
||||
assert_eq!(popped2.id, id2);
|
||||
|
||||
let popped3 = queue.pop().unwrap();
|
||||
assert_eq!(popped3.id, id3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_priority_queue_deduplication() {
|
||||
let mut queue = PriorityHealQueue::new();
|
||||
|
||||
let req1 = HealRequest::new(
|
||||
HealType::Object {
|
||||
bucket: "bucket1".to_string(),
|
||||
object: "object1".to_string(),
|
||||
version_id: None,
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Normal,
|
||||
);
|
||||
|
||||
let req2 = HealRequest::new(
|
||||
HealType::Object {
|
||||
bucket: "bucket1".to_string(),
|
||||
object: "object1".to_string(),
|
||||
version_id: None,
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::High,
|
||||
);
|
||||
|
||||
// First request should be added
|
||||
assert!(queue.push(req1));
|
||||
assert_eq!(queue.len(), 1);
|
||||
|
||||
// Second request with same object should be rejected (duplicate)
|
||||
assert!(!queue.push(req2));
|
||||
assert_eq!(queue.len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_priority_queue_contains_erasure_set() {
|
||||
let mut queue = PriorityHealQueue::new();
|
||||
|
||||
let req = HealRequest::new(
|
||||
HealType::ErasureSet {
|
||||
buckets: vec!["bucket1".to_string()],
|
||||
set_disk_id: "pool_0_set_1".to_string(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Normal,
|
||||
);
|
||||
|
||||
assert!(queue.push(req));
|
||||
assert!(queue.contains_erasure_set("pool_0_set_1"));
|
||||
assert!(!queue.contains_erasure_set("pool_0_set_2"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_priority_queue_dedup_key_generation() {
|
||||
// Test different heal types generate different keys
|
||||
let obj_req = HealRequest::new(
|
||||
HealType::Object {
|
||||
bucket: "bucket1".to_string(),
|
||||
object: "object1".to_string(),
|
||||
version_id: None,
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Normal,
|
||||
);
|
||||
|
||||
let bucket_req = HealRequest::new(
|
||||
HealType::Bucket {
|
||||
bucket: "bucket1".to_string(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Normal,
|
||||
);
|
||||
|
||||
let erasure_req = HealRequest::new(
|
||||
HealType::ErasureSet {
|
||||
buckets: vec!["bucket1".to_string()],
|
||||
set_disk_id: "pool_0_set_1".to_string(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Normal,
|
||||
);
|
||||
|
||||
let obj_key = PriorityHealQueue::make_dedup_key(&obj_req);
|
||||
let bucket_key = PriorityHealQueue::make_dedup_key(&bucket_req);
|
||||
let erasure_key = PriorityHealQueue::make_dedup_key(&erasure_req);
|
||||
|
||||
// All keys should be different
|
||||
assert_ne!(obj_key, bucket_key);
|
||||
assert_ne!(obj_key, erasure_key);
|
||||
assert_ne!(bucket_key, erasure_key);
|
||||
|
||||
assert!(obj_key.starts_with("object:"));
|
||||
assert!(bucket_key.starts_with("bucket:"));
|
||||
assert!(erasure_key.starts_with("erasure_set:"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_priority_queue_mixed_priorities_and_types() {
|
||||
let mut queue = PriorityHealQueue::new();
|
||||
|
||||
// Add various requests
|
||||
let requests = vec![
|
||||
(
|
||||
HealType::Object {
|
||||
bucket: "b1".to_string(),
|
||||
object: "o1".to_string(),
|
||||
version_id: None,
|
||||
},
|
||||
HealPriority::Low,
|
||||
),
|
||||
(
|
||||
HealType::Bucket {
|
||||
bucket: "b2".to_string(),
|
||||
},
|
||||
HealPriority::Urgent,
|
||||
),
|
||||
(
|
||||
HealType::ErasureSet {
|
||||
buckets: vec!["b3".to_string()],
|
||||
set_disk_id: "pool_0_set_1".to_string(),
|
||||
},
|
||||
HealPriority::Normal,
|
||||
),
|
||||
(
|
||||
HealType::Object {
|
||||
bucket: "b4".to_string(),
|
||||
object: "o4".to_string(),
|
||||
version_id: None,
|
||||
},
|
||||
HealPriority::High,
|
||||
),
|
||||
];
|
||||
|
||||
for (heal_type, priority) in requests {
|
||||
let req = HealRequest::new(heal_type, HealOptions::default(), priority);
|
||||
queue.push(req);
|
||||
}
|
||||
|
||||
assert_eq!(queue.len(), 4);
|
||||
|
||||
// Check they come out in priority order
|
||||
let priorities: Vec<HealPriority> = (0..4).filter_map(|_| queue.pop().map(|r| r.priority)).collect();
|
||||
|
||||
assert_eq!(
|
||||
priorities,
|
||||
vec![
|
||||
HealPriority::Urgent,
|
||||
HealPriority::High,
|
||||
HealPriority::Normal,
|
||||
HealPriority::Low,
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_priority_queue_stats() {
|
||||
let mut queue = PriorityHealQueue::new();
|
||||
|
||||
// Add requests with different priorities
|
||||
for _ in 0..3 {
|
||||
queue.push(HealRequest::new(
|
||||
HealType::Bucket {
|
||||
bucket: format!("bucket-low-{}", queue.len()),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Low,
|
||||
));
|
||||
}
|
||||
|
||||
for _ in 0..2 {
|
||||
queue.push(HealRequest::new(
|
||||
HealType::Bucket {
|
||||
bucket: format!("bucket-normal-{}", queue.len()),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Normal,
|
||||
));
|
||||
}
|
||||
|
||||
queue.push(HealRequest::new(
|
||||
HealType::Bucket {
|
||||
bucket: "bucket-high".to_string(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::High,
|
||||
));
|
||||
|
||||
let stats = queue.get_priority_stats();
|
||||
|
||||
assert_eq!(*stats.get(&HealPriority::Low).unwrap_or(&0), 3);
|
||||
assert_eq!(*stats.get(&HealPriority::Normal).unwrap_or(&0), 2);
|
||||
assert_eq!(*stats.get(&HealPriority::High).unwrap_or(&0), 1);
|
||||
assert_eq!(*stats.get(&HealPriority::Urgent).unwrap_or(&0), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_priority_queue_is_empty() {
|
||||
let mut queue = PriorityHealQueue::new();
|
||||
|
||||
assert!(queue.is_empty());
|
||||
|
||||
queue.push(HealRequest::new(
|
||||
HealType::Bucket {
|
||||
bucket: "test".to_string(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Normal,
|
||||
));
|
||||
|
||||
assert!(!queue.is_empty());
|
||||
|
||||
queue.pop();
|
||||
|
||||
assert!(queue.is_empty());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -400,13 +400,13 @@ impl HealStorageAPI for ECStoreHealStorage {
|
||||
match self.ecstore.get_object_info(bucket, object, &Default::default()).await {
|
||||
Ok(_) => Ok(true), // Object exists
|
||||
Err(e) => {
|
||||
// Map ObjectNotFound to false, other errors to false as well for safety
|
||||
// Map ObjectNotFound to false, other errors must be propagated!
|
||||
if matches!(e, rustfs_ecstore::error::StorageError::ObjectNotFound(_, _)) {
|
||||
debug!("Object not found: {}/{}", bucket, object);
|
||||
Ok(false)
|
||||
} else {
|
||||
debug!("Error checking object existence {}/{}: {}", bucket, object, e);
|
||||
Ok(false) // Treat errors as non-existence to be safe
|
||||
error!("Error checking object existence {}/{}: {}", bucket, object, e);
|
||||
Err(Error::other(e))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+12
-12
@@ -51,7 +51,7 @@ pub enum HealType {
|
||||
}
|
||||
|
||||
/// Heal priority
|
||||
#[derive(Debug, Default, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
|
||||
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
|
||||
pub enum HealPriority {
|
||||
/// Low priority
|
||||
Low = 0,
|
||||
@@ -272,6 +272,7 @@ impl HealTask {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self), fields(task_id = %self.id, heal_type = ?self.heal_type))]
|
||||
pub async fn execute(&self) -> Result<()> {
|
||||
// update status and timestamps atomically to avoid race conditions
|
||||
let now = SystemTime::now();
|
||||
@@ -285,7 +286,7 @@ impl HealTask {
|
||||
*task_start_instant = Some(start_instant);
|
||||
}
|
||||
|
||||
info!("Starting heal task: {} with type: {:?}", self.id, self.heal_type);
|
||||
info!("Task started");
|
||||
|
||||
let result = match &self.heal_type {
|
||||
HealType::Object {
|
||||
@@ -315,7 +316,7 @@ impl HealTask {
|
||||
Ok(_) => {
|
||||
let mut status = self.status.write().await;
|
||||
*status = HealTaskStatus::Completed;
|
||||
info!("Heal task completed successfully: {}", self.id);
|
||||
info!("Task completed successfully");
|
||||
}
|
||||
Err(Error::TaskCancelled) => {
|
||||
let mut status = self.status.write().await;
|
||||
@@ -354,8 +355,9 @@ impl HealTask {
|
||||
}
|
||||
|
||||
// specific heal implementation method
|
||||
#[tracing::instrument(skip(self), fields(bucket = %bucket, object = %object, version_id = ?version_id))]
|
||||
async fn heal_object(&self, bucket: &str, object: &str, version_id: Option<&str>) -> Result<()> {
|
||||
info!("Healing object: {}/{}", bucket, object);
|
||||
info!("Starting object heal workflow");
|
||||
|
||||
// update progress
|
||||
{
|
||||
@@ -365,7 +367,7 @@ impl HealTask {
|
||||
}
|
||||
|
||||
// Step 1: Check if object exists and get metadata
|
||||
info!("Step 1: Checking object existence and metadata");
|
||||
warn!("Step 1: Checking object existence and metadata");
|
||||
self.check_control_flags().await?;
|
||||
let object_exists = self.await_with_control(self.storage.object_exists(bucket, object)).await?;
|
||||
if !object_exists {
|
||||
@@ -424,7 +426,7 @@ impl HealTask {
|
||||
|
||||
// If heal failed and remove_corrupted is enabled, delete the corrupted object
|
||||
if self.options.remove_corrupted {
|
||||
warn!("Removing corrupted object: {}/{}", bucket, object);
|
||||
info!("Removing corrupted object: {}/{}", bucket, object);
|
||||
if !self.options.dry_run {
|
||||
self.await_with_control(self.storage.delete_object(bucket, object)).await?;
|
||||
info!("Successfully deleted corrupted object: {}/{}", bucket, object);
|
||||
@@ -447,11 +449,9 @@ impl HealTask {
|
||||
info!("Step 3: Verifying heal result");
|
||||
let object_size = result.object_size as u64;
|
||||
info!(
|
||||
"Heal completed successfully: {}/{} ({} bytes, {} drives healed)",
|
||||
bucket,
|
||||
object,
|
||||
object_size,
|
||||
result.after.drives.len()
|
||||
object_size = object_size,
|
||||
drives_healed = result.after.drives.len(),
|
||||
"Heal completed successfully"
|
||||
);
|
||||
|
||||
{
|
||||
@@ -481,7 +481,7 @@ impl HealTask {
|
||||
|
||||
// If heal failed and remove_corrupted is enabled, delete the corrupted object
|
||||
if self.options.remove_corrupted {
|
||||
warn!("Removing corrupted object: {}/{}", bucket, object);
|
||||
info!("Removing corrupted object: {}/{}", bucket, object);
|
||||
if !self.options.dry_run {
|
||||
self.await_with_control(self.storage.delete_object(bucket, object)).await?;
|
||||
info!("Successfully deleted corrupted object: {}/{}", bucket, object);
|
||||
|
||||
Reference in New Issue
Block a user