From be7f684718976c62bf3bc845778d9828aa670737 Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 19 Aug 2026 22:52:36 +0800 Subject: [PATCH] refactor(heal): remove the dead MRF heal-type path (#6275) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit HealType::MRF (a #1664-era "metadata repair file" task kind) had no production construction site left: its only builder lived in the HealEvent -> HealRequest converter, and the HealEvent/HealEventHandler queue itself had zero production references — both were superseded by the MrfIntent pipeline (mrf_queue.rs), which produces Object/Metadata/ ECDecode requests and never an MRF task. The dead path nevertheless carried ~700 lines: the whole event.rs module, the heal_mrf executor, a dedup-key arm, an overlap arm with the "\u{0}mrf" sentinel bucket hack, per-kind labels, and an empty MrfRuntime::record_accept shell. Deleting the variant is compile-time safe: HealType has no Serialize derive, the protos wire enums carry no heal-type discriminant (the receiver rebuilds it from HealChannelRequest fields), the MRF journal encodes MrfKind (1/2/3), and the scanner pending-heal ledger uses its own kind enum — none of them can name an MRF task. Also resolves the in-crate naming clash where "MRF" denoted both the dead task kind and the live mission-repair-feed loop; the loop stays, the task kind goes. Co-authored-by: heihutu --- crates/heal/src/heal/event.rs | 683 ----------------------- crates/heal/src/heal/manager.rs | 129 ++++- crates/heal/src/heal/mod.rs | 1 - crates/heal/src/heal/mrf_queue.rs | 136 ++++- crates/heal/src/heal/task.rs | 157 +----- crates/heal/tests/heal_bug_fixes_test.rs | 82 +-- 6 files changed, 223 insertions(+), 965 deletions(-) delete mode 100644 crates/heal/src/heal/event.rs diff --git a/crates/heal/src/heal/event.rs b/crates/heal/src/heal/event.rs deleted file mode 100644 index 0d46b859b..000000000 --- a/crates/heal/src/heal/event.rs +++ /dev/null @@ -1,683 +0,0 @@ -// 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. - -use crate::heal::{HealOptions, HealPriority, HealRequest, HealType}; -use crate::{Error, Result}; -use serde::{Deserialize, Serialize}; -use std::time::SystemTime; - -use super::Endpoint; - -/// Corruption type -#[derive(Debug, Clone, Serialize, Deserialize)] -pub enum CorruptionType { - /// Data corruption - DataCorruption, - /// Metadata corruption - MetadataCorruption, - /// Partial corruption - PartialCorruption, - /// Complete corruption - CompleteCorruption, -} - -/// Severity level -#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] -pub enum Severity { - /// Low severity - Low = 0, - /// Medium severity - Medium = 1, - /// High severity - High = 2, - /// Critical severity - Critical = 3, -} - -/// Heal event -#[derive(Debug, Clone)] -pub enum HealEvent { - /// Object corruption event - ObjectCorruption { - bucket: String, - object: String, - version_id: Option, - corruption_type: CorruptionType, - severity: Severity, - }, - /// Object missing event - ObjectMissing { - bucket: String, - object: String, - version_id: Option, - expected_locations: Vec, - available_locations: Vec, - }, - /// Metadata corruption event - MetadataCorruption { - bucket: String, - object: String, - corruption_type: CorruptionType, - }, - /// Disk status change event - DiskStatusChange { - endpoint: Endpoint, - old_status: String, - new_status: String, - }, - /// EC decode failure event - ECDecodeFailure { - bucket: String, - object: String, - version_id: Option, - missing_shards: Vec, - available_shards: Vec, - }, - /// Checksum mismatch event - ChecksumMismatch { - bucket: String, - object: String, - version_id: Option, - expected_checksum: String, - actual_checksum: String, - }, - /// Bucket metadata corruption event - BucketMetadataCorruption { - bucket: String, - corruption_type: CorruptionType, - }, - /// MRF metadata corruption event - MRFMetadataCorruption { - meta_path: String, - corruption_type: CorruptionType, - }, -} - -impl HealEvent { - /// Convert HealEvent to HealRequest - pub fn to_heal_request(&self) -> Result { - match self { - HealEvent::ObjectCorruption { - bucket, - object, - version_id, - severity, - .. - } => Ok(HealRequest::new( - HealType::Object { - bucket: bucket.clone(), - object: object.clone(), - version_id: version_id.clone(), - }, - HealOptions::default(), - Self::severity_to_priority(severity), - )), - HealEvent::ObjectMissing { - bucket, - object, - version_id, - .. - } => Ok(HealRequest::new( - HealType::Object { - bucket: bucket.clone(), - object: object.clone(), - version_id: version_id.clone(), - }, - HealOptions::default(), - HealPriority::High, - )), - HealEvent::MetadataCorruption { bucket, object, .. } => Ok(HealRequest::new( - HealType::Metadata { - bucket: bucket.clone(), - object: object.clone(), - }, - HealOptions::default(), - HealPriority::High, - )), - HealEvent::DiskStatusChange { endpoint, .. } => { - // Convert disk status change to erasure set heal - // Note: This requires access to storage to get bucket list, which is not available here - // The actual bucket list will need to be provided by the caller or retrieved differently - let set_disk_id = crate::heal::utils::format_set_disk_id_from_i32(endpoint.pool_idx, endpoint.set_idx) - .ok_or_else(|| Error::InvalidHealType { - heal_type: format!("erasure-set(pool={}, set={})", endpoint.pool_idx, endpoint.set_idx), - })?; - Ok(HealRequest::new( - HealType::ErasureSet { - buckets: vec![], // Empty bucket list - caller should populate this - set_disk_id, - }, - HealOptions::default(), - HealPriority::High, - )) - } - HealEvent::ECDecodeFailure { - bucket, - object, - version_id, - .. - } => Ok(HealRequest::new( - HealType::ECDecode { - bucket: bucket.clone(), - object: object.clone(), - version_id: version_id.clone(), - }, - HealOptions::default(), - HealPriority::Urgent, - )), - HealEvent::ChecksumMismatch { - bucket, - object, - version_id, - .. - } => Ok(HealRequest::new( - HealType::Object { - bucket: bucket.clone(), - object: object.clone(), - version_id: version_id.clone(), - }, - HealOptions::default(), - HealPriority::High, - )), - HealEvent::BucketMetadataCorruption { bucket, .. } => Ok(HealRequest::new( - HealType::Bucket { bucket: bucket.clone() }, - HealOptions::default(), - HealPriority::High, - )), - HealEvent::MRFMetadataCorruption { meta_path, .. } => Ok(HealRequest::new( - HealType::MRF { - meta_path: meta_path.clone(), - }, - HealOptions::default(), - HealPriority::High, - )), - } - } - - /// Convert severity to priority - fn severity_to_priority(severity: &Severity) -> HealPriority { - match severity { - Severity::Low => HealPriority::Low, - Severity::Medium => HealPriority::Normal, - Severity::High => HealPriority::High, - Severity::Critical => HealPriority::Urgent, - } - } - - /// Get event description - pub fn description(&self) -> String { - match self { - HealEvent::ObjectCorruption { - bucket, - object, - corruption_type, - .. - } => { - format!("Object corruption detected: {bucket}/{object} - {corruption_type:?}") - } - HealEvent::ObjectMissing { bucket, object, .. } => { - format!("Object missing: {bucket}/{object}") - } - HealEvent::MetadataCorruption { - bucket, - object, - corruption_type, - .. - } => { - format!("Metadata corruption: {bucket}/{object} - {corruption_type:?}") - } - HealEvent::DiskStatusChange { - endpoint, - old_status, - new_status, - .. - } => { - format!("Disk status changed: {endpoint:?} {old_status} -> {new_status}") - } - HealEvent::ECDecodeFailure { - bucket, - object, - missing_shards, - .. - } => { - format!("EC decode failure: {bucket}/{object} - missing shards: {missing_shards:?}") - } - HealEvent::ChecksumMismatch { - bucket, - object, - expected_checksum, - actual_checksum, - .. - } => { - format!("Checksum mismatch: {bucket}/{object} - expected: {expected_checksum}, actual: {actual_checksum}") - } - HealEvent::BucketMetadataCorruption { - bucket, corruption_type, .. - } => { - format!("Bucket metadata corruption: {bucket} - {corruption_type:?}") - } - HealEvent::MRFMetadataCorruption { - meta_path, - corruption_type, - .. - } => { - format!("MRF metadata corruption: {meta_path} - {corruption_type:?}") - } - } - } - - /// Get event severity - pub fn severity(&self) -> Severity { - match self { - HealEvent::ObjectCorruption { severity, .. } => severity.clone(), - HealEvent::ObjectMissing { .. } => Severity::High, - HealEvent::MetadataCorruption { .. } => Severity::High, - HealEvent::DiskStatusChange { .. } => Severity::High, - HealEvent::ECDecodeFailure { .. } => Severity::Critical, - HealEvent::ChecksumMismatch { .. } => Severity::High, - HealEvent::BucketMetadataCorruption { .. } => Severity::High, - HealEvent::MRFMetadataCorruption { .. } => Severity::High, - } - } - - /// Get event timestamp - pub fn timestamp(&self) -> SystemTime { - SystemTime::now() - } -} - -/// Heal event handler -pub struct HealEventHandler { - /// Event queue - events: Vec, - /// Maximum number of events - max_events: usize, -} - -impl HealEventHandler { - pub fn new(max_events: usize) -> Self { - Self { - events: Vec::new(), - max_events, - } - } - - /// Add event - pub fn add_event(&mut self, event: HealEvent) { - if self.events.len() >= self.max_events { - // Remove oldest event - self.events.remove(0); - } - self.events.push(event); - } - - /// Get all events - pub fn get_events(&self) -> &[HealEvent] { - &self.events - } - - /// Clear events - pub fn clear_events(&mut self) { - self.events.clear(); - } - - /// Get event count - pub fn event_count(&self) -> usize { - self.events.len() - } - - /// Filter events by severity - pub fn filter_by_severity(&self, min_severity: Severity) -> Vec<&HealEvent> { - self.events.iter().filter(|event| event.severity() >= min_severity).collect() - } - - /// Filter events by type - pub fn filter_by_type(&self, event_type: &str) -> Vec<&HealEvent> { - self.events - .iter() - .filter(|event| match event { - HealEvent::ObjectCorruption { .. } => event_type == "ObjectCorruption", - HealEvent::ObjectMissing { .. } => event_type == "ObjectMissing", - HealEvent::MetadataCorruption { .. } => event_type == "MetadataCorruption", - HealEvent::DiskStatusChange { .. } => event_type == "DiskStatusChange", - HealEvent::ECDecodeFailure { .. } => event_type == "ECDecodeFailure", - HealEvent::ChecksumMismatch { .. } => event_type == "ChecksumMismatch", - HealEvent::BucketMetadataCorruption { .. } => event_type == "BucketMetadataCorruption", - HealEvent::MRFMetadataCorruption { .. } => event_type == "MRFMetadataCorruption", - }) - .collect() - } -} - -impl Default for HealEventHandler { - fn default() -> Self { - Self::new(1000) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::heal::task::{HealPriority, HealType}; - - #[test] - fn test_heal_event_object_corruption_to_request() { - let event = HealEvent::ObjectCorruption { - bucket: "test-bucket".to_string(), - object: "test-object".to_string(), - version_id: None, - corruption_type: CorruptionType::DataCorruption, - severity: Severity::High, - }; - - let request = event.to_heal_request().unwrap(); - assert!(matches!(request.heal_type, HealType::Object { .. })); - assert_eq!(request.priority, HealPriority::High); - } - - #[test] - fn test_heal_event_object_missing_to_request() { - let event = HealEvent::ObjectMissing { - bucket: "test-bucket".to_string(), - object: "test-object".to_string(), - version_id: Some("v1".to_string()), - expected_locations: vec![0, 1], - available_locations: vec![2, 3], - }; - - let request = event.to_heal_request().unwrap(); - assert!(matches!(request.heal_type, HealType::Object { .. })); - assert_eq!(request.priority, HealPriority::High); - } - - #[test] - fn test_heal_event_metadata_corruption_to_request() { - let event = HealEvent::MetadataCorruption { - bucket: "test-bucket".to_string(), - object: "test-object".to_string(), - corruption_type: CorruptionType::MetadataCorruption, - }; - - let request = event.to_heal_request().unwrap(); - assert!(matches!(request.heal_type, HealType::Metadata { .. })); - assert_eq!(request.priority, HealPriority::High); - } - - #[test] - fn test_heal_event_ec_decode_failure_to_request() { - let event = HealEvent::ECDecodeFailure { - bucket: "test-bucket".to_string(), - object: "test-object".to_string(), - version_id: None, - missing_shards: vec![0, 1], - available_shards: vec![2, 3, 4], - }; - - let request = event.to_heal_request().unwrap(); - assert!(matches!(request.heal_type, HealType::ECDecode { .. })); - assert_eq!(request.priority, HealPriority::Urgent); - } - - #[test] - fn test_heal_event_checksum_mismatch_to_request() { - let event = HealEvent::ChecksumMismatch { - bucket: "test-bucket".to_string(), - object: "test-object".to_string(), - version_id: None, - expected_checksum: "abc123".to_string(), - actual_checksum: "def456".to_string(), - }; - - let request = event.to_heal_request().unwrap(); - assert!(matches!(request.heal_type, HealType::Object { .. })); - assert_eq!(request.priority, HealPriority::High); - } - - #[test] - fn test_heal_event_bucket_metadata_corruption_to_request() { - let event = HealEvent::BucketMetadataCorruption { - bucket: "test-bucket".to_string(), - corruption_type: CorruptionType::MetadataCorruption, - }; - - let request = event.to_heal_request().unwrap(); - assert!(matches!(request.heal_type, HealType::Bucket { .. })); - assert_eq!(request.priority, HealPriority::High); - } - - #[test] - fn test_heal_event_mrf_metadata_corruption_to_request() { - let event = HealEvent::MRFMetadataCorruption { - meta_path: "test-bucket/test-object".to_string(), - corruption_type: CorruptionType::MetadataCorruption, - }; - - let request = event.to_heal_request().unwrap(); - assert!(matches!(request.heal_type, HealType::MRF { .. })); - assert_eq!(request.priority, HealPriority::High); - } - - #[test] - fn test_heal_event_severity_to_priority() { - let event_low = HealEvent::ObjectCorruption { - bucket: "test".to_string(), - object: "test".to_string(), - version_id: None, - corruption_type: CorruptionType::DataCorruption, - severity: Severity::Low, - }; - let request = event_low.to_heal_request().unwrap(); - assert_eq!(request.priority, HealPriority::Low); - - let event_medium = HealEvent::ObjectCorruption { - bucket: "test".to_string(), - object: "test".to_string(), - version_id: None, - corruption_type: CorruptionType::DataCorruption, - severity: Severity::Medium, - }; - let request = event_medium.to_heal_request().unwrap(); - assert_eq!(request.priority, HealPriority::Normal); - - let event_high = HealEvent::ObjectCorruption { - bucket: "test".to_string(), - object: "test".to_string(), - version_id: None, - corruption_type: CorruptionType::DataCorruption, - severity: Severity::High, - }; - let request = event_high.to_heal_request().unwrap(); - assert_eq!(request.priority, HealPriority::High); - - let event_critical = HealEvent::ObjectCorruption { - bucket: "test".to_string(), - object: "test".to_string(), - version_id: None, - corruption_type: CorruptionType::DataCorruption, - severity: Severity::Critical, - }; - let request = event_critical.to_heal_request().unwrap(); - assert_eq!(request.priority, HealPriority::Urgent); - } - - #[test] - fn test_heal_event_description() { - let event = HealEvent::ObjectCorruption { - bucket: "test-bucket".to_string(), - object: "test-object".to_string(), - version_id: None, - corruption_type: CorruptionType::DataCorruption, - severity: Severity::High, - }; - - let desc = event.description(); - assert!(desc.contains("Object corruption detected")); - assert!(desc.contains("test-bucket/test-object")); - assert!(desc.contains("DataCorruption")); - } - - #[test] - fn test_heal_event_severity() { - let event = HealEvent::ECDecodeFailure { - bucket: "test".to_string(), - object: "test".to_string(), - version_id: None, - missing_shards: vec![], - available_shards: vec![], - }; - assert_eq!(event.severity(), Severity::Critical); - - let event = HealEvent::ObjectMissing { - bucket: "test".to_string(), - object: "test".to_string(), - version_id: None, - expected_locations: vec![], - available_locations: vec![], - }; - assert_eq!(event.severity(), Severity::High); - } - - #[test] - fn test_heal_event_handler_new() { - let handler = HealEventHandler::new(10); - assert_eq!(handler.event_count(), 0); - assert_eq!(handler.max_events, 10); - } - - #[test] - fn test_heal_event_handler_default() { - let handler = HealEventHandler::default(); - assert_eq!(handler.max_events, 1000); - } - - #[test] - fn test_heal_event_handler_add_event() { - let mut handler = HealEventHandler::new(3); - let event = HealEvent::ObjectCorruption { - bucket: "test".to_string(), - object: "test".to_string(), - version_id: None, - corruption_type: CorruptionType::DataCorruption, - severity: Severity::High, - }; - - handler.add_event(event.clone()); - assert_eq!(handler.event_count(), 1); - - handler.add_event(event.clone()); - handler.add_event(event); - assert_eq!(handler.event_count(), 3); - } - - #[test] - fn test_heal_event_handler_max_events() { - let mut handler = HealEventHandler::new(2); - let event = HealEvent::ObjectCorruption { - bucket: "test".to_string(), - object: "test".to_string(), - version_id: None, - corruption_type: CorruptionType::DataCorruption, - severity: Severity::High, - }; - - handler.add_event(event.clone()); - handler.add_event(event.clone()); - handler.add_event(event); // Should remove oldest - - assert_eq!(handler.event_count(), 2); - } - - #[test] - fn test_heal_event_handler_get_events() { - let mut handler = HealEventHandler::new(10); - let event = HealEvent::ObjectCorruption { - bucket: "test".to_string(), - object: "test".to_string(), - version_id: None, - corruption_type: CorruptionType::DataCorruption, - severity: Severity::High, - }; - - handler.add_event(event.clone()); - handler.add_event(event); - - let events = handler.get_events(); - assert_eq!(events.len(), 2); - } - - #[test] - fn test_heal_event_handler_clear_events() { - let mut handler = HealEventHandler::new(10); - let event = HealEvent::ObjectCorruption { - bucket: "test".to_string(), - object: "test".to_string(), - version_id: None, - corruption_type: CorruptionType::DataCorruption, - severity: Severity::High, - }; - - handler.add_event(event); - assert_eq!(handler.event_count(), 1); - - handler.clear_events(); - assert_eq!(handler.event_count(), 0); - } - - #[test] - fn test_heal_event_handler_filter_by_severity() { - let mut handler = HealEventHandler::new(10); - handler.add_event(HealEvent::ObjectCorruption { - bucket: "test".to_string(), - object: "test".to_string(), - version_id: None, - corruption_type: CorruptionType::DataCorruption, - severity: Severity::Low, - }); - handler.add_event(HealEvent::ECDecodeFailure { - bucket: "test".to_string(), - object: "test".to_string(), - version_id: None, - missing_shards: vec![], - available_shards: vec![], - }); - - let high_severity = handler.filter_by_severity(Severity::High); - assert_eq!(high_severity.len(), 1); // Only ECDecodeFailure is Critical >= High - } - - #[test] - fn test_heal_event_handler_filter_by_type() { - let mut handler = HealEventHandler::new(10); - handler.add_event(HealEvent::ObjectCorruption { - bucket: "test".to_string(), - object: "test".to_string(), - version_id: None, - corruption_type: CorruptionType::DataCorruption, - severity: Severity::High, - }); - handler.add_event(HealEvent::ObjectMissing { - bucket: "test".to_string(), - object: "test".to_string(), - version_id: None, - expected_locations: vec![], - available_locations: vec![], - }); - - let corruption_events = handler.filter_by_type("ObjectCorruption"); - assert_eq!(corruption_events.len(), 1); - - let missing_events = handler.filter_by_type("ObjectMissing"); - assert_eq!(missing_events.len(), 1); - } -} diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 4c3c622c7..2f03b30d4 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -86,7 +86,7 @@ fn unblock_replacement_recovery_sets_after_validation( } } -// Admission/scheduler outcomes for per-object requests (Object/Metadata/MRF/ +// Admission/scheduler outcomes for per-object requests (Object/Metadata/ // ECDecode) log via demote_to_debug_when! — MRF, autoheal, and scanner // recovery loops submit those per object, so a full queue or a retry storm // would otherwise emit one warn! per object (rustfs/rustfs#5716). The @@ -143,6 +143,16 @@ async fn pause_duplicate_admission_after_active_lock(request_id: &str) { type WorkloadSnapshotProviderRef = Arc; +/// 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)] +struct DedupKeyEntry { + refcount: usize, + 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)] @@ -151,8 +161,8 @@ struct PriorityHealQueue { heap: BinaryHeap, /// Sequence counter for FIFO ordering within same priority sequence: u64, - /// Deduplication key reference counts for queued requests - dedup_keys: HashMap, + /// Deduplication index for queued requests + dedup_keys: HashMap, } /// Wrapper for heap items to implement proper ordering @@ -402,8 +412,16 @@ impl PriorityHealQueue { 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. - *self.dedup_keys.entry(key).or_insert(0) += 1; + // 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, @@ -447,6 +465,7 @@ impl PriorityHealQueue { 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 }); @@ -559,9 +578,6 @@ impl PriorityHealQueue { HealType::Metadata { bucket, object } => { format!("metadata:{bucket}:{object}") } - HealType::MRF { meta_path } => { - format!("mrf:{meta_path}") - } HealType::ECDecode { bucket, object, @@ -572,12 +588,12 @@ impl PriorityHealQueue { } } - fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap, key: &str) { - if let Some(count) = dedup_keys.get_mut(key) { - if *count <= 1 { + fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap, key: &str) { + if let Some(entry) = dedup_keys.get_mut(key) { + if entry.refcount <= 1 { dedup_keys.remove(key); } else { - *count -= 1; + entry.refcount -= 1; } } } @@ -610,10 +626,32 @@ impl PriorityHealQueue { .any(|item| item.request.id == request_id && heal_type_matches_path(&item.request.heal_type, heal_path)) } - fn request_for_dedup_key(&self, key: &str) -> Option<&HealRequest> { - self.heap + 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. + fn refresh_dedup_representative(&mut self, key: &str) { + if !self.dedup_keys.contains_key(key) { + return; + } + if let Some(id) = self + .heap .iter() - .find_map(|item| (Self::make_dedup_key(&item.request) == key).then_some(&item.request)) + .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; + } } fn contains_matching(&self, mut matches: F) -> bool @@ -638,6 +676,9 @@ impl PriorityHealQueue { } self.heap = retained; + if let Some(removed) = removed.as_ref() { + self.refresh_dedup_representative(&Self::make_dedup_key(removed)); + } removed } @@ -647,11 +688,13 @@ impl PriorityHealQueue { { 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); @@ -659,6 +702,9 @@ impl PriorityHealQueue { } self.heap = retained; + for key in &affected_keys { + self.refresh_dedup_representative(key); + } removed } } @@ -686,7 +732,6 @@ fn heal_type_matches_path(heal_type: &HealType, heal_path: &str) -> bool { HealType::Bucket { bucket } => heal_path == bucket, HealType::Prefix { bucket, prefix } => heal_path_matches_bucket_child(heal_path, bucket, prefix), HealType::ErasureSet { set_disk_id, .. } => heal_path == set_disk_id, - HealType::MRF { meta_path } => heal_path == meta_path.trim_matches('/'), } } @@ -781,9 +826,6 @@ fn heal_type_path_view(heal_type: &HealType) -> (Option<&str>, &str) { HealType::Object { bucket, object, .. } | HealType::Metadata { bucket, object } | HealType::ECDecode { bucket, object, .. } => (Some(bucket), object), - // MRF/MetaPath heal keys on a meta path; treat the whole set of - // buckets as one namespace so it only overlaps itself exactly. - HealType::MRF { meta_path } => (Some("\u{0}mrf"), meta_path), // Erasure-set heal: the set id is the overlap dimension. HealType::ErasureSet { set_disk_id, .. } => (Some("\u{0}set"), set_disk_id), } @@ -1974,8 +2016,8 @@ impl HealManager { .map(|(task_id, _)| (task_id, "active")) .or_else(|| { queue - .request_for_dedup_key(&dedup_key) - .map(|queued| (queued.id.clone(), "queued")) + .queued_request_id_for_dedup_key(&dedup_key) + .map(|queued_id| (queued_id.to_string(), "queued")) }) .or_else(|| retrying_heal_for_dedup_key(&retrying_heals, &dedup_key).map(|(task_id, _)| (task_id, "retrying"))) }); @@ -2093,9 +2135,9 @@ impl HealManager { let mut task_id = request.id.clone(); let admission = Self::admit_request_to_queue(&mut queue, request, &config, "submit"); if admission == HealAdmissionResult::Merged - && let Some(queued) = queue.request_for_dedup_key(&dedup_key) + && let Some(queued_id) = queue.queued_request_id_for_dedup_key(&dedup_key) { - task_id.clone_from(&queued.id); + task_id = queued_id.to_owned(); } let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable; drop(retrying_heals); @@ -3712,7 +3754,6 @@ fn heal_request_type_label(request: &HealRequest) -> &'static str { HealType::Prefix { .. } => "prefix", HealType::ErasureSet { .. } => "erasure_set", HealType::Metadata { .. } => "metadata", - HealType::MRF { .. } => "mrf", HealType::ECDecode { .. } => "ec_decode", } } @@ -4104,6 +4145,46 @@ mod tests { assert_eq!(queue.len(), 0); } + #[test] + fn queued_request_id_for_dedup_key_tracks_the_representative() { + let mut queue = PriorityHealQueue::new(); + + let first = HealRequest::object("bucket".to_string(), "object".to_string(), None); + let first_id = first.id.clone(); + let first_key = PriorityHealQueue::make_dedup_key(&first); + assert_eq!(queue.push(first), QueuePushOutcome::Accepted); + + // A forced duplicate of the same target opens a second entry under + // the same key; the representative stays the request that opened it. + let mut second = HealRequest::object("bucket".to_string(), "object".to_string(), None); + second.force_start = true; + let second_id = second.id.clone(); + assert_eq!(queue.push(second), QueuePushOutcome::Accepted); + + let representative = queue + .queued_request_id_for_dedup_key(&first_key) + .expect("key must be reserved while either request is queued"); + assert_eq!(representative, first_id); + + // A holder leaving WITHOUT becoming active (canceled by id) must + // re-elect the representative to the surviving queued request, or a + // later merge receipt would name an id that resolves nowhere. The + // scheduler pop path needs no re-election: the popped request + // surfaces in active_heals under the same id and the duplicate + // pre-check consults active heals before the queue. + queue.remove_request_id(&first_id); + assert_eq!( + queue.queued_request_id_for_dedup_key(&first_key), + Some(second_id.as_str()), + "canceling the opener must re-elect the surviving queued holder" + ); + + // Pop the last holder: the key is released entirely. + let last = queue.pop_next().expect("second request must be queued"); + assert_eq!(last.id, second_id); + assert!(queue.queued_request_id_for_dedup_key(&first_key).is_none()); + } + #[test] fn test_priority_queue_ordering() { let mut queue = PriorityHealQueue::new(); diff --git a/crates/heal/src/heal/mod.rs b/crates/heal/src/heal/mod.rs index ff910ed70..35080e8e9 100644 --- a/crates/heal/src/heal/mod.rs +++ b/crates/heal/src/heal/mod.rs @@ -14,7 +14,6 @@ pub mod channel; pub mod erasure_healer; -pub mod event; pub mod manager; pub mod mrf_queue; pub mod progress; diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs index c1144435e..5101a5e12 100644 --- a/crates/heal/src/heal/mrf_queue.rs +++ b/crates/heal/src/heal/mrf_queue.rs @@ -276,20 +276,26 @@ async fn read_journal() -> Option> { None } -async fn write_journal(data: &[u8]) { +/// Write the snapshot to every local disk; returns true when at least one +/// disk accepted it, so a total write failure keeps the runtime dirty and +/// the next tick retries the persist. +async fn write_journal(data: &[u8]) -> bool { let payload = bytes::Bytes::copy_from_slice(data); + let mut any_persisted = false; for disk in journal_disks().await { - if let Err(err) = disk + match disk .write_all(super::RUSTFS_META_BUCKET, MRF_JOURNAL_PATH, payload.clone()) .await { - warn_mrf_journal_write(&err); + Ok(()) => any_persisted = true, + Err(err) => warn_mrf_journal_write(&err), } } if !data.is_empty() { counter!("rustfs_heal_mrf_journal_fsync_total").increment(1); } gauge!("rustfs_heal_mrf_journal_bytes").set(data.len() as f64); + any_persisted } async fn delete_journal() { @@ -351,6 +357,12 @@ struct MrfRuntime { queue: MrfQueue, config: MrfConsumerConfig, new_since_flush: usize, + /// True while the in-memory pending set has changed since the last + /// journal flush (push, pop, or an attempts bump that alters the encoded + /// bytes). Only a dirty state rewrites the snapshot: a steady backlog + /// waiting out an admission backoff must not re-fsync every local disk + /// twice a second. + dirty: bool, /// True while a journal snapshot exists on disk that no longer reflects /// an all-consumed pending set; the next idle tick removes it (MinIO /// deletes its `list.bin` after replay for the same reason). @@ -360,11 +372,6 @@ struct MrfRuntime { } impl MrfRuntime { - fn record_accept(&mut self) { - // Accepted intents leave the pending set; the next flush persists the - // smaller snapshot, which is the journal's compaction. - } - fn snapshot(&self) -> Vec { let mut buf = Vec::new(); for intent in self.queue.intents() { @@ -374,8 +381,14 @@ impl MrfRuntime { } async fn flush(&mut self) { - write_journal(&self.snapshot()).await; + let persisted = write_journal(&self.snapshot()).await; self.new_since_flush = 0; + // Keep the dirty flag when every disk write failed: a clean backlog + // would otherwise never rewrite, losing the periodic persist retry a + // non-empty queue used to provide. + if persisted { + self.dirty = false; + } self.journal_on_disk = true; } @@ -389,9 +402,15 @@ impl MrfRuntime { self.backoff_until = None; } while let Some(mut intent) = self.queue.pop_front() { + // Leaving the pending set (consumed or re-queued with a bumped + // attempts counter) changes the encoded snapshot; mark it dirty + // either way. + self.dirty = true; let request = build_heal_request(&intent); match manager.submit_heal_request(request).await { - Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => self.record_accept(), + // Accepted intents leave the pending set; the next flush persists the + // smaller snapshot, which is the journal's compaction. + Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {} Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => { intent.attempts = intent.attempts.saturating_add(1); if intent.attempts >= MRF_MAX_ATTEMPTS { @@ -519,6 +538,7 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive queue: MrfQueue::new(config.queue_capacity, config.journal_max_bytes), config: config.clone(), new_since_flush: 0, + dirty: false, journal_on_disk: false, backoff_until: None, }; @@ -526,6 +546,10 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive // Replay: read the journal, re-arm intents (duplicates are merged by the // manager's dedup key), then drop the file so the next flush starts clean. replay_into(&manager, &mut runtime.queue, &mut runtime.backoff_until).await; + // The replay deleted the journal file; anything still pending (e.g. the + // manager was full and backoff armed) must be re-persisted by the next + // flush or a crash before it would lose those intents. + runtime.dirty = runtime.queue.depth() > 0; let mut flush_tick = tokio::time::interval(runtime.config.flush_interval); flush_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); @@ -535,8 +559,13 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive tokio::select! { received = receiver.recv_many(&mut batch, runtime.config.replay_batch) => { if received == 0 { - // Channel closed: flush once more and stop. - runtime.flush().await; + // Channel closed: flush once more unless the snapshot is + // provably current AND idle (a dirty or pending state + // gets one last persist attempt, matching the shutdown + // retry the unconditional flush used to provide). + if runtime.dirty || runtime.queue.depth() > 0 { + runtime.flush().await; + } tracing::info!( target: "rustfs::heal::mrf", "MRF channel closed; consumer stopped after final flush" @@ -544,8 +573,10 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive return; } for intent in batch.drain(..) { - runtime.queue.try_push(intent); - runtime.new_since_flush += 1; + if runtime.queue.try_push(intent) { + runtime.new_since_flush += 1; + runtime.dirty = true; + } } runtime.dispatch(manager.as_ref()).await; if runtime.new_since_flush >= runtime.config.flush_threshold { @@ -553,15 +584,26 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive } } _ = flush_tick.tick() => { - if runtime.new_since_flush > 0 || runtime.queue.depth() > 0 { - runtime.flush().await; - runtime.dispatch(manager.as_ref()).await; - } else if runtime.journal_on_disk { - // All intents consumed: remove the journal so a restart - // replays nothing (mirrors MinIO's post-replay unlink). - delete_journal().await; - runtime.journal_on_disk = false; - gauge!("rustfs_heal_mrf_journal_bytes").set(0.0); + match tick_action(runtime.dirty, runtime.queue.depth(), runtime.journal_on_disk) { + TickAction::Flush => { + runtime.flush().await; + runtime.dispatch(manager.as_ref()).await; + } + TickAction::Retry => { + // Pending set unchanged since the last flush (a + // backlog waiting out an admission backoff): skip the + // rewrite but keep dispatching so the retry fires on + // time. + runtime.dispatch(manager.as_ref()).await; + } + TickAction::DeleteJournal => { + // All intents consumed: remove the journal so a restart + // replays nothing (mirrors MinIO's post-replay unlink). + delete_journal().await; + runtime.journal_on_disk = false; + gauge!("rustfs_heal_mrf_journal_bytes").set(0.0); + } + TickAction::Idle => {} } gauge!("rustfs_heal_mrf_queue_depth").set(runtime.queue.depth() as f64); } @@ -569,6 +611,33 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive } } +/// What the periodic tick should do, as a pure function of the runtime state +/// so the decision table is unit-testable. +enum TickAction { + /// The pending set changed since the last snapshot: rewrite it, then + /// drain. + Flush, + /// Pending intents exist but the snapshot is current: only drain (an + /// admission backoff may have expired). + Retry, + /// Nothing pending and a stale journal file remains: remove it. + DeleteJournal, + /// Quiescent: nothing to do. + Idle, +} + +fn tick_action(dirty: bool, depth: usize, journal_on_disk: bool) -> TickAction { + if dirty { + TickAction::Flush + } else if depth > 0 { + TickAction::Retry + } else if journal_on_disk { + TickAction::DeleteJournal + } else { + TickAction::Idle + } +} + #[cfg(test)] mod tests { use super::*; @@ -586,6 +655,27 @@ mod tests { } } + #[test] + fn tick_action_table() { + use TickAction::*; + + // Dirty dominates: a changed pending set flushes even when idle + // otherwise. + assert!(matches!(tick_action(true, 0, false), Flush)); + assert!(matches!(tick_action(true, 3, true), Flush)); + + // Clean backlog: no rewrite, but keep draining so an expired + // admission backoff retries on time. + assert!(matches!(tick_action(false, 1, false), Retry)); + assert!(matches!(tick_action(false, 2, true), Retry)); + + // Quiescent with a stale journal file on disk: remove it. + assert!(matches!(tick_action(false, 0, true), DeleteJournal)); + + // Fully quiescent: nothing to do. + assert!(matches!(tick_action(false, 0, false), Idle)); + } + #[test] fn queue_enforces_count_and_byte_ceilings() { let mut queue = MrfQueue::new(2, usize::MAX); diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 7b1d33348..4db62f271 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -54,7 +54,7 @@ const MAX_BUCKET_OBJECT_HEAL_RETRIES: u32 = 3; const MAX_BUCKET_FAILURE_LOG_SAMPLES: u64 = 5; /// Emits at `$level`, demoted to `debug!` when `$demote` is true. Keeps -/// per-object heal work — Object/Metadata/MRF/ECDecode tasks queued per +/// per-object heal work — Object/Metadata/ECDecode tasks queued per /// object by MRF/autoheal/scanner loops, and per-object sweep failures past /// a sample cap — from amplifying into one info!/warn!/error! line per /// object during mass recovery (rustfs/rustfs#5716). Aggregate task kinds @@ -75,8 +75,6 @@ const EVENT_HEAL_BUCKET_STAGE: &str = "heal_bucket_stage"; const EVENT_HEAL_BUCKET_RESULT: &str = "heal_bucket_result"; const EVENT_HEAL_METADATA_STAGE: &str = "heal_metadata_stage"; const EVENT_HEAL_METADATA_RESULT: &str = "heal_metadata_result"; -const EVENT_HEAL_MRF_STAGE: &str = "heal_mrf_stage"; -const EVENT_HEAL_MRF_RESULT: &str = "heal_mrf_result"; const EVENT_HEAL_EC_DECODE_STAGE: &str = "heal_ec_decode_stage"; const EVENT_HEAL_EC_DECODE_RESULT: &str = "heal_ec_decode_result"; const EVENT_HEAL_ERASURE_SET_STAGE: &str = "heal_erasure_set_stage"; @@ -101,8 +99,6 @@ pub enum HealType { ErasureSet { buckets: Vec, set_disk_id: String }, /// Metadata heal Metadata { bucket: String, object: String }, - /// MRF heal - MRF { meta_path: String }, /// EC decode heal ECDecode { bucket: String, @@ -120,21 +116,18 @@ impl HealType { Self::Prefix { .. } => "prefix", Self::ErasureSet { .. } => "erasure_set", Self::Metadata { .. } => "metadata", - Self::MRF { .. } => "mrf", Self::ECDecode { .. } => "ec_decode", } } /// Task kinds enqueued at per-object granularity (MRF, autoheal, scanner, - /// read-repair loops). Their lifecycle and admission logs stay at `debug!` + /// read-repair loops; the MRF loop queues Object/ECDecode/Metadata + /// tasks). Their lifecycle and admission logs stay at `debug!` /// so a recovery loop queuing hundreds of thousands of object heal tasks /// cannot amplify into per-object `info!`/`warn!` lines; aggregate kinds /// (cluster/bucket/prefix/erasure-set) keep operator-visible levels. pub(crate) fn is_per_object(&self) -> bool { - matches!( - self, - Self::Object { .. } | Self::Metadata { .. } | Self::MRF { .. } | Self::ECDecode { .. } - ) + matches!(self, Self::Object { .. } | Self::Metadata { .. } | Self::ECDecode { .. }) } } @@ -504,7 +497,6 @@ impl HealTask { HealType::Prefix { .. } => "prefix", HealType::ErasureSet { .. } => "erasure_set", HealType::Metadata { .. } => "metadata", - HealType::MRF { .. } => "mrf", HealType::ECDecode { .. } => "ec_decode", } } @@ -579,7 +571,6 @@ impl HealTask { None => event, } } - HealType::MRF { meta_path } => event.with_object(meta_path.as_str()), }; match error { @@ -821,7 +812,6 @@ impl HealTask { HealType::Prefix { bucket, prefix } => self.heal_prefix(bucket, prefix).await, HealType::Metadata { bucket, object } => self.heal_metadata(bucket, object).await, - HealType::MRF { meta_path } => self.heal_mrf(meta_path).await, HealType::ECDecode { bucket, object, @@ -2020,139 +2010,6 @@ impl HealTask { } } - async fn heal_mrf(&self, meta_path: &str) -> Result<()> { - debug!( - target: "rustfs::heal::task", - event = EVENT_HEAL_MRF_STAGE, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_TASK, - task_id = %self.id, - meta_path, - stage = "start", - "Heal MRF started" - ); - - // update progress - { - let mut progress = self.progress.write().await; - progress.set_current_object(Some(format!("mrf: {meta_path}"))); - progress.update_progress(0, 2, 0, 0); - } - - // Parse meta_path to extract bucket and object - let parts: Vec<&str> = meta_path.split('/').collect(); - if parts.len() < 2 { - return Err(Error::TaskExecutionFailed { - message: format!("Invalid meta path format: {meta_path}"), - }); - } - - let bucket = parts[0]; - let object = parts[1..].join("/"); - - // Step 1: Perform MRF heal using ecstore - debug!( - target: "rustfs::heal::task", - event = EVENT_HEAL_MRF_STAGE, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_TASK, - task_id = %self.id, - meta_path, - bucket, - object = %object, - stage = "heal_with_ecstore", - "Heal MRF stage entered" - ); - let heal_opts = HealOpts { - recursive: true, - dry_run: self.options.dry_run, - remove: self.options.remove_corrupted, - recreate: self.options.recreate_missing, - scan_mode: HealScanMode::Deep, - update_parity: true, - no_lock: self.options.no_lock, - pool: None, - set: None, - }; - - let heal_result = self - .await_with_control(self.storage.heal_object(bucket, &object, None, &heal_opts)) - .await; - - match heal_result { - Ok((result, error)) => { - if let Some(e) = error { - error!( - target: "rustfs::heal::task", - event = EVENT_HEAL_MRF_RESULT, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_TASK, - task_id = %self.id, - meta_path, - bucket, - object = %object, - result = "failed", - error = %e, - "Heal MRF failed" - ); - { - let mut progress = self.progress.write().await; - progress.update_progress(2, 2, 0, 0); - } - return Err(Error::TaskExecutionFailed { - message: format!("Failed to heal MRF {meta_path}: {e}"), - }); - } - - debug!( - target: "rustfs::heal::task", - event = EVENT_HEAL_MRF_RESULT, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_TASK, - task_id = %self.id, - meta_path, - bucket, - object = %object, - drives_healed = result.drives_healed(), - drives_total = result.drives_reported(), - result = "ok", - "Heal MRF repaired" - ); - - { - let mut progress = self.progress.write().await; - progress.update_progress(2, 2, 0, 0); - } - self.record_result_item(result).await; - Ok(()) - } - Err(Error::TaskCancelled) => Err(Error::TaskCancelled), - Err(Error::TaskTimeout) => Err(Error::TaskTimeout), - Err(e) => { - error!( - target: "rustfs::heal::task", - event = EVENT_HEAL_MRF_RESULT, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_TASK, - task_id = %self.id, - meta_path, - bucket, - object = %object, - result = "failed", - error = %e, - "Heal MRF failed" - ); - { - let mut progress = self.progress.write().await; - progress.update_progress(2, 2, 0, 0); - } - Err(Error::TaskExecutionFailed { - message: format!("Failed to heal MRF {meta_path}: {e}"), - }) - } - } - } - async fn heal_ec_decode(&self, bucket: &str, object: &str, version_id: Option<&str>) -> Result<()> { debug!( target: "rustfs::heal::task", @@ -3389,12 +3246,6 @@ mod tests { } .is_per_object() ); - assert!( - HealType::MRF { - meta_path: "p".to_string(), - } - .is_per_object() - ); assert!( HealType::ECDecode { bucket: "b".to_string(), diff --git a/crates/heal/tests/heal_bug_fixes_test.rs b/crates/heal/tests/heal_bug_fixes_test.rs index 1bdb00348..59ff1bb3f 100644 --- a/crates/heal/tests/heal_bug_fixes_test.rs +++ b/crates/heal/tests/heal_bug_fixes_test.rs @@ -13,93 +13,13 @@ // limitations under the License. use rustfs_heal::heal::{ - event::{HealEvent, Severity}, task::{HealPriority, HealType}, utils, }; mod storage_api; -use storage_api::bug_fixes::{BucketInfo, DiskStore, Endpoint}; - -#[test] -fn test_heal_event_to_heal_request_no_panic() { - // Test that invalid pool/set indices don't cause panic - // Create endpoint using try_from or similar method - let endpoint_result = Endpoint::try_from("http://localhost:9000"); - if let Ok(mut endpoint) = endpoint_result { - endpoint.pool_idx = -1; - endpoint.set_idx = -1; - endpoint.disk_idx = 0; - - let event = HealEvent::DiskStatusChange { - endpoint, - old_status: "ok".to_string(), - new_status: "offline".to_string(), - }; - - // Should return error instead of panicking - let result = event.to_heal_request(); - assert!(result.is_err()); - assert!(result.unwrap_err().to_string().contains("Invalid heal type")); - } -} - -#[test] -fn test_heal_event_to_heal_request_valid_indices() { - // Test that valid indices work correctly - let endpoint_result = Endpoint::try_from("http://localhost:9000"); - if let Ok(mut endpoint) = endpoint_result { - endpoint.pool_idx = 0; - endpoint.set_idx = 1; - endpoint.disk_idx = 0; - - let event = HealEvent::DiskStatusChange { - endpoint, - old_status: "ok".to_string(), - new_status: "offline".to_string(), - }; - - let result = event.to_heal_request(); - assert!(result.is_ok()); - let request = result.unwrap(); - assert!(matches!(request.heal_type, HealType::ErasureSet { .. })); - } -} - -#[test] -fn test_heal_event_object_corruption() { - let event = HealEvent::ObjectCorruption { - bucket: "test-bucket".to_string(), - object: "test-object".to_string(), - version_id: None, - corruption_type: rustfs_heal::heal::event::CorruptionType::DataCorruption, - severity: Severity::High, - }; - - let result = event.to_heal_request(); - assert!(result.is_ok()); - let request = result.unwrap(); - assert!(matches!(request.heal_type, HealType::Object { .. })); - assert_eq!(request.priority, HealPriority::High); -} - -#[test] -fn test_heal_event_ec_decode_failure() { - let event = HealEvent::ECDecodeFailure { - bucket: "test-bucket".to_string(), - object: "test-object".to_string(), - version_id: None, - missing_shards: vec![0, 1], - available_shards: vec![2, 3], - }; - - let result = event.to_heal_request(); - assert!(result.is_ok()); - let request = result.unwrap(); - assert!(matches!(request.heal_type, HealType::ECDecode { .. })); - assert_eq!(request.priority, HealPriority::Urgent); -} +use storage_api::bug_fixes::{BucketInfo, DiskStore}; #[test] fn test_format_set_disk_id_from_i32_negative() {