gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-19 19:16:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(heal): remove the dead MRF heal-type path (#6275)

Browse Source 
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 <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-19 22:52:36 +08:00
committed by GitHub
parent 3bde70d5b4
commit be7f684718
6 changed files with 223 additions and 965 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/event.rs
-683
View File
@@ -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<String>,
corruption_type: CorruptionType,
severity: Severity,
},
/// Object missing event
ObjectMissing {
bucket: String,
object: String,
version_id: Option<String>,
expected_locations: Vec<usize>,
available_locations: Vec<usize>,
},
/// 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<String>,
missing_shards: Vec<usize>,
available_shards: Vec<usize>,
},
/// Checksum mismatch event
ChecksumMismatch {
bucket: String,
object: String,
version_id: Option<String>,
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<HealRequest> {
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<HealEvent>,
/// 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);
}
}
crates/heal/src/heal/manager.rs
+105 -24
View File
@@ -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<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>;
/// 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<PriorityQueueItem>,
/// Sequence counter for FIFO ordering within same priority
sequence: u64,
/// Deduplication key reference counts for queued requests
dedup_keys: HashMap<String, usize>,
/// Deduplication index for queued requests
dedup_keys: HashMap<String, DedupKeyEntry>,
}
/// 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<String, usize>, key: &str) {
if let Some(count) = dedup_keys.get_mut(key) {
if *count <= 1 {
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 {
*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<F>(&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();
crates/heal/src/heal/mod.rs
-1
View File
@@ -14,7 +14,6 @@
pub mod channel;
pub mod erasure_healer;
pub mod event;
pub mod manager;
pub mod mrf_queue;
pub mod progress;
crates/heal/src/heal/mrf_queue.rs
+113 -23
View File
@@ -276,20 +276,26 @@ async fn read_journal() -> Option<Vec<u8>> {
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<u8> {
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<HealManager>, 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<HealManager>, 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<HealManager>, 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<HealManager>, 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<HealManager>, 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<HealManager>, 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);
crates/heal/src/heal/task.rs
+4 -153
View File
@@ -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<String>, 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(),
crates/heal/tests/heal_bug_fixes_test.rs
+1 -81
View File
@@ -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() {