From 5eda332feeeeb66029025f1ce4aee13431c2ac63 Mon Sep 17 00:00:00 2001 From: cxymds Date: Wed, 24 Jun 2026 12:51:16 +0800 Subject: [PATCH] fix(heal): recover shards after node rejoin (#3814) --- .../src/heal_erasure_disk_rebuild_test.rs | 57 ++++- crates/heal/src/heal/channel.rs | 124 ++++++++-- crates/heal/src/heal/manager.rs | 156 ++++++++++++- crates/heal/src/heal/progress.rs | 17 ++ crates/heal/src/heal/task.rs | 217 +++++++++++++++--- rustfs/src/admin/handlers/heal.rs | 153 +++++++++--- 6 files changed, 638 insertions(+), 86 deletions(-) diff --git a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs index e5435f3d5..576331ed9 100644 --- a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs +++ b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs @@ -16,8 +16,12 @@ #[cfg(test)] mod tests { - use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, execute_awscurl, init_logging}; + use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging, local_http_client}; use aws_sdk_s3::primitives::ByteStream; + use http::header::{CONTENT_TYPE, HOST}; + use rustfs_signer::constants::UNSIGNED_PAYLOAD; + use rustfs_signer::sign_v4; + use s3s::Body; use serial_test::serial; use std::collections::HashSet; use std::error::Error; @@ -61,6 +65,45 @@ mod tests { assert_eq!(body.as_ref(), expected, "object body changed for {key}"); } + async fn signed_admin_post( + url: &str, + body: Option<&str>, + access_key: &str, + secret_key: &str, + ) -> Result> { + let uri = url.parse::()?; + let authority = uri.authority().ok_or("request URL missing authority")?.to_string(); + let mut request = http::Request::builder() + .method(http::Method::POST) + .uri(uri) + .header(HOST, authority) + .header("x-amz-content-sha256", UNSIGNED_PAYLOAD); + + if body.is_some() { + request = request.header(CONTENT_TYPE, "application/json"); + } + + let content_len = body.map(str::len).unwrap_or_default() as i64; + let signed = sign_v4(request.body(Body::empty())?, content_len, access_key, secret_key, "", "us-east-1"); + + let mut request_builder = local_http_client().post(url); + for (name, value) in signed.headers() { + request_builder = request_builder.header(name, value); + } + if let Some(body) = body { + request_builder = request_builder.body(body.to_string()); + } + + let response = request_builder.send().await?; + let status = response.status(); + let body = response.text().await?; + if !status.is_success() { + return Err(format!("admin POST failed: {status} {body}").into()); + } + + Ok(body) + } + #[tokio::test] #[serial] async fn test_auto_heal_rebuilds_runtime_wiped_disk_without_restart() { @@ -292,7 +335,7 @@ mod tests { let heal_body = r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#; let heal_url = format!("{}/rustfs/admin/v3/heal/{}?forceStart=true", env.url, bucket); - execute_awscurl(&heal_url, "POST", Some(heal_body), &env.access_key, &env.secret_key) + signed_admin_post(&heal_url, Some(heal_body), &env.access_key, &env.secret_key) .await .expect("admin deep heal should be accepted"); @@ -332,9 +375,9 @@ mod tests { #[tokio::test(flavor = "multi_thread")] #[serial] - async fn test_cluster_admin_heal_rebuilds_replaced_remote_disk() -> Result<(), Box> { + async fn test_cluster_root_heal_rebuilds_replaced_remote_disk() -> Result<(), Box> { init_logging(); - info!("Admin deep heal should rebuild data on a remote node after its disk is replaced and the node rejoins"); + info!("Root recursive heal should rebuild data on a remote node after its disk is replaced and the node rejoins"); let mut cluster = RustFSTestClusterEnvironment::new(4).await?; cluster.set_env("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", "true"); @@ -383,15 +426,15 @@ mod tests { cluster.start_node(1).await?; let status_url = format!("{}/rustfs/admin/v3/background-heal/status", cluster.nodes[0].url); - let status_body = execute_awscurl(&status_url, "POST", None, &cluster.access_key, &cluster.secret_key).await?; + let status_body = signed_admin_post(&status_url, None, &cluster.access_key, &cluster.secret_key).await?; assert!( !status_body.contains("MissingContentLength"), "background heal status should not fail without an explicit Content-Length: {status_body}" ); let heal_body = r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#; - let heal_url = format!("{}/rustfs/admin/v3/heal/{}?forceStart=true", cluster.nodes[0].url, bucket); - execute_awscurl(&heal_url, "POST", Some(heal_body), &cluster.access_key, &cluster.secret_key).await?; + let heal_url = format!("{}/rustfs/admin/v3/heal/?forceStart=true", cluster.nodes[0].url); + signed_admin_post(&heal_url, Some(heal_body), &cluster.access_key, &cluster.secret_key).await?; let expected_objects = [(online_key, online_body.as_slice()), (outage_key, outage_body.as_slice())]; let mut remaining_rebuild_keys: HashSet<&str> = expected_objects.iter().map(|(key, _)| *key).collect(); diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index bedc8ca40..7ba2bcc51 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -14,6 +14,7 @@ use crate::heal::{ manager::{HealManager, HealTaskReport}, + progress::HealProgress, task::{HealOptions, HealPriority, HealRequest, HealTaskStatus, HealType}, utils, }; @@ -48,6 +49,8 @@ pub struct HealChannelProcessor { struct HealTaskStatusPayload { summary: String, items: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + progress: Option, } impl HealChannelProcessor { @@ -268,38 +271,48 @@ impl HealChannelProcessor { self.heal_manager.get_task_report_for_path(&heal_path, &client_token).await }; - let (summary, detail, items) = match report { + let (summary, detail, items, progress) = match report { Ok(HealTaskReport { status: HealTaskStatus::Pending | HealTaskStatus::Running, result_items, - }) => ("running".to_string(), None, result_items), + progress, + }) => ("running".to_string(), None, result_items, progress), Ok(HealTaskReport { status: HealTaskStatus::Retrying { error, retry_attempt }, result_items, + progress, }) => ( "running".to_string(), Some(format!("heal task retrying after recoverable failure, attempt {retry_attempt}: {error}")), result_items, + progress, ), Ok(HealTaskReport { status: HealTaskStatus::Completed, result_items, - }) => ("finished".to_string(), None, result_items), + progress, + }) => ("finished".to_string(), None, result_items, progress), Ok(HealTaskReport { status: HealTaskStatus::Cancelled, result_items, - }) => ("stopped".to_string(), Some("heal task cancelled".to_string()), result_items), + progress, + }) => ("stopped".to_string(), Some("heal task cancelled".to_string()), result_items, progress), Ok(HealTaskReport { status: HealTaskStatus::Timeout, result_items, - }) => ("stopped".to_string(), Some("heal task timed out".to_string()), result_items), + progress, + }) => ("stopped".to_string(), Some("heal task timed out".to_string()), result_items, progress), Ok(HealTaskReport { status: HealTaskStatus::Failed { error }, result_items, - }) => ("stopped".to_string(), Some(error), result_items), - Err(crate::Error::TaskNotFound { .. }) => { - ("notFound".to_string(), Some("heal task not found or expired".to_string()), Vec::new()) - } + progress, + }) => ("stopped".to_string(), Some(error), result_items, progress), + Err(crate::Error::TaskNotFound { .. }) => ( + "notFound".to_string(), + Some("heal task not found or expired".to_string()), + Vec::new(), + None, + ), Err(crate::Error::InvalidClientToken) => { let response = HealChannelResponse { request_id: client_token, @@ -325,8 +338,12 @@ impl HealChannelProcessor { } }; - let data = serde_json::to_vec(&HealTaskStatusPayload { summary, items }) - .map_err(|e| crate::Error::Serialization(format!("failed to serialize heal task status: {e}")))?; + let data = serde_json::to_vec(&HealTaskStatusPayload { + summary, + items, + progress, + }) + .map_err(|e| crate::Error::Serialization(format!("failed to serialize heal task status: {e}")))?; let response = HealChannelResponse { request_id: client_token, @@ -400,6 +417,7 @@ impl HealChannelProcessor { /// Convert channel request to heal request fn convert_to_heal_request(&self, request: HealChannelRequest) -> Result { + let recursive = request.recursive.unwrap_or(false); let heal_type = if let Some(disk_id) = &request.disk { let set_disk_id = utils::normalize_set_disk_id(disk_id).ok_or_else(|| Error::InvalidHealType { heal_type: format!("erasure-set({disk_id})"), @@ -408,12 +426,21 @@ impl HealChannelProcessor { buckets: vec![], set_disk_id, } + } else if request.bucket.is_empty() { + HealType::Cluster } else if let Some(prefix) = &request.object_prefix { if !prefix.is_empty() { - HealType::Object { - bucket: request.bucket.clone(), - object: prefix.clone(), - version_id: request.object_version_id.clone(), + if recursive { + HealType::Prefix { + bucket: request.bucket.clone(), + prefix: prefix.clone(), + } + } else { + HealType::Object { + bucket: request.bucket.clone(), + object: prefix.clone(), + version_id: request.object_version_id.clone(), + } } } else { HealType::Bucket { @@ -439,7 +466,7 @@ impl HealChannelProcessor { remove_corrupted: request.remove_corrupted.unwrap_or(false), recreate_missing: request.recreate_missing.unwrap_or(true), update_parity: request.update_parity.unwrap_or(true), - recursive: request.recursive.unwrap_or(false), + recursive, dry_run: request.dry_run.unwrap_or(false), timeout: request.timeout_seconds.map(std::time::Duration::from_secs), pool_index: request.pool_index, @@ -633,6 +660,37 @@ mod tests { assert_eq!(heal_request.priority, HealPriority::Normal); } + #[tokio::test] + async fn test_convert_to_heal_request_cluster() { + let heal_manager = create_test_heal_manager(); + let processor = HealChannelProcessor::new(heal_manager); + + let channel_request = HealChannelRequest { + id: "test-id".to_string(), + bucket: String::new(), + object_prefix: None, + object_version_id: None, + disk: None, + priority: HealChannelPriority::High, + scan_mode: Some(HealScanMode::Normal), + remove_corrupted: Some(false), + recreate_missing: Some(true), + update_parity: Some(true), + recursive: Some(true), + dry_run: Some(false), + timeout_seconds: None, + pool_index: None, + set_index: None, + force_start: false, + source: HealRequestSource::Admin, + }; + + let heal_request = processor.convert_to_heal_request(channel_request).unwrap(); + + assert!(matches!(heal_request.heal_type, HealType::Cluster)); + assert!(heal_request.options.recursive); + } + #[tokio::test] async fn test_convert_to_heal_request_object() { let heal_manager = create_test_heal_manager(); @@ -667,6 +725,40 @@ mod tests { assert!(heal_request.options.recreate_missing); } + #[tokio::test] + async fn test_convert_to_heal_request_prefix_when_recursive() { + let heal_manager = create_test_heal_manager(); + let processor = HealChannelProcessor::new(heal_manager); + + let channel_request = HealChannelRequest { + id: "test-id".to_string(), + bucket: "test-bucket".to_string(), + object_prefix: Some("logs/".to_string()), + object_version_id: None, + disk: None, + priority: HealChannelPriority::High, + scan_mode: Some(HealScanMode::Normal), + remove_corrupted: Some(false), + recreate_missing: Some(true), + update_parity: Some(true), + recursive: Some(true), + dry_run: Some(false), + timeout_seconds: None, + pool_index: None, + set_index: None, + force_start: false, + source: HealRequestSource::Admin, + }; + + let heal_request = processor.convert_to_heal_request(channel_request).unwrap(); + + assert!(matches!( + heal_request.heal_type, + HealType::Prefix { ref bucket, ref prefix } if bucket == "test-bucket" && prefix == "logs/" + )); + assert!(heal_request.options.recursive); + } + #[tokio::test] async fn test_convert_to_heal_request_erasure_set() { let heal_manager = create_test_heal_manager(); diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 0b05c9b2b..0d3c07920 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -22,7 +22,7 @@ use metrics::{counter, gauge}; use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult, HealRequestSource}; use rustfs_madmin::heal_commands::HealResultItem; use std::{ - collections::{BinaryHeap, HashMap}, + collections::{BinaryHeap, HashMap, HashSet}, sync::Arc, time::{Duration, SystemTime}, }; @@ -126,6 +126,7 @@ struct RetryingHeal { pub struct HealTaskReport { pub status: HealTaskStatus, pub result_items: Vec, + pub progress: Option, } #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize)] @@ -348,6 +349,7 @@ impl PriorityHealQueue { fn make_dedup_key_for_type(heal_type: &HealType) -> String { match heal_type { + HealType::Cluster => "cluster".to_string(), HealType::Object { bucket, object, @@ -358,6 +360,9 @@ impl PriorityHealQueue { HealType::Bucket { bucket } => { format!("bucket:{bucket}") } + HealType::Prefix { bucket, prefix } => { + format!("prefix:{bucket}/{prefix}") + } HealType::ErasureSet { set_disk_id, .. } => { format!("erasure_set:{set_disk_id}") } @@ -475,14 +480,16 @@ impl RetryingHeal { fn heal_type_matches_path(heal_type: &HealType, heal_path: &str) -> bool { let heal_path = heal_path.trim_matches('/'); if heal_path.is_empty() { - return false; + return matches!(heal_type, HealType::Cluster); } match heal_type { + HealType::Cluster => false, HealType::Object { bucket, object, .. } | HealType::Metadata { bucket, object } | HealType::ECDecode { bucket, object, .. } => heal_path == bucket || heal_path == format!("{bucket}/{object}"), HealType::Bucket { bucket } => heal_path == bucket, + HealType::Prefix { bucket, prefix } => heal_path == bucket || heal_path == format!("{bucket}/{prefix}"), HealType::ErasureSet { set_disk_id, .. } => heal_path == set_disk_id, HealType::MRF { meta_path } => heal_path == meta_path.trim_matches('/'), } @@ -1208,6 +1215,7 @@ impl HealManager { return Ok(HealTaskReport { status: task.get_status().await, result_items: task.get_result_items().await, + progress: Some(task.get_progress().await), }); } } @@ -1218,6 +1226,7 @@ impl HealManager { return Ok(HealTaskReport { status: retrying.status(), result_items: Vec::new(), + progress: None, }); } } @@ -1231,6 +1240,7 @@ impl HealManager { return Ok(HealTaskReport { status: completed.status.clone(), result_items: completed.result_items.clone(), + progress: None, }); } } @@ -1241,6 +1251,7 @@ impl HealManager { return Ok(HealTaskReport { status: HealTaskStatus::Pending, result_items: Vec::new(), + progress: None, }); } } @@ -1251,6 +1262,7 @@ impl HealManager { return Ok(HealTaskReport { status: completed.status.clone(), result_items: completed.result_items.clone(), + progress: None, }); } @@ -1269,6 +1281,7 @@ impl HealManager { return Ok(HealTaskReport { status: task.get_status().await, result_items: task.get_result_items().await, + progress: Some(task.get_progress().await), }); } } @@ -1281,6 +1294,7 @@ impl HealManager { return Ok(HealTaskReport { status: retrying.status(), result_items: Vec::new(), + progress: None, }); } } @@ -1295,6 +1309,7 @@ impl HealManager { return Ok(HealTaskReport { status: completed.status.clone(), result_items: completed.result_items.clone(), + progress: None, }); } } @@ -1305,6 +1320,7 @@ impl HealManager { return Ok(HealTaskReport { status: HealTaskStatus::Pending, result_items: Vec::new(), + progress: None, }); } } @@ -1318,6 +1334,7 @@ impl HealManager { return Ok(HealTaskReport { status: completed.status.clone(), result_items: completed.result_items.clone(), + progress: None, }); } } @@ -1762,8 +1779,14 @@ impl HealManager { _ = interval.tick() => { // Build list of endpoints that need healing let mut endpoints = Vec::new(); + let mut seen_returning_sets = HashSet::new(); for (_, disk_opt) in GLOBAL_LOCAL_DISK_MAP.read().await.iter() { if let Some(disk) = disk_opt { + let endpoint = disk.endpoint(); + let runtime_state = disk.runtime_state(); + let set_disk_id = + crate::heal::utils::format_set_disk_id_from_i32(endpoint.pool_idx, endpoint.set_idx); + // detect unformatted disk via get_disk_id() match disk.get_disk_id().await { Err(DiskError::UnformattedDisk) => { @@ -1773,11 +1796,11 @@ impl HealManager { event = EVENT_HEAL_AUTO_SCAN_DISK, component = LOG_COMPONENT_HEAL, subsystem = LOG_SUBSYSTEM_DISK_SCANNER, - endpoint = %disk.endpoint(), + endpoint = %endpoint, disk_state = "unformatted", "Heal auto-scan candidate detected" ); - endpoints.push(disk.endpoint()); + endpoints.push(endpoint); } Err(e) => { warn!( @@ -1785,14 +1808,30 @@ impl HealManager { event = EVENT_HEAL_AUTO_SCAN_DISK, component = LOG_COMPONENT_HEAL, subsystem = LOG_SUBSYSTEM_DISK_SCANNER, - endpoint = %disk.endpoint(), + endpoint = %endpoint, disk_state = "check_failed", error = ?e, "Heal auto-scan disk inspection failed" ); } Ok(_) => { - // Disk is formatted, no action needed + if runtime_state.as_str() == "returning" + && let Some(set_disk_id) = set_disk_id + && seen_returning_sets.insert(set_disk_id.clone()) + { + candidate_count += 1; + debug!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_AUTO_SCAN_DISK, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_DISK_SCANNER, + endpoint = %endpoint, + set_disk_id, + disk_state = "returning", + "Heal auto-scan returning disk candidate detected" + ); + endpoints.push(endpoint); + } } } } @@ -1887,7 +1926,10 @@ impl HealManager { buckets: buckets.clone(), set_disk_id: set_disk_id.clone(), }, - HealOptions::default(), + HealOptions { + timeout: None, + ..HealOptions::default() + }, HealPriority::Normal, ); req.source = HealRequestSource::AutoHeal; @@ -2266,8 +2308,10 @@ fn heal_request_set_key(request: &HealRequest) -> Option { fn heal_request_type_label(request: &HealRequest) -> &'static str { match &request.heal_type { + HealType::Cluster => "cluster", HealType::Object { .. } => "object", HealType::Bucket { .. } => "bucket", + HealType::Prefix { .. } => "prefix", HealType::ErasureSet { .. } => "erasure_set", HealType::Metadata { .. } => "metadata", HealType::MRF { .. } => "mrf", @@ -3072,6 +3116,104 @@ mod tests { assert!(matches!(manager.get_task_status(&task_id).await, Err(Error::TaskNotFound { .. }))); } + #[tokio::test] + async fn test_cancel_tasks_for_empty_path_cancels_queued_cluster_only() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + let cluster_request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::High); + let cluster_request_id = cluster_request.id.clone(); + let bucket_request = HealRequest::bucket("bucket".to_string()); + let bucket_request_id = bucket_request.id.clone(); + + manager + .submit_heal_request(cluster_request) + .await + .expect("cluster request should be accepted"); + manager + .submit_heal_request(bucket_request) + .await + .expect("bucket request should be accepted"); + + assert_eq!( + manager + .cancel_tasks_for_path("") + .await + .expect("root path should cancel queued cluster task"), + 1 + ); + assert!(matches!( + manager.get_task_status(&cluster_request_id).await, + Err(Error::TaskNotFound { .. }) + )); + assert_eq!( + manager + .get_task_status(&bucket_request_id) + .await + .expect("bucket request should not match root path"), + HealTaskStatus::Pending + ); + } + + #[tokio::test] + async fn test_cancel_tasks_for_empty_path_cancels_active_cluster_only() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage.clone(), None); + + let cluster_request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::High); + let cluster_request_id = cluster_request.id.clone(); + let bucket_request = HealRequest::bucket("bucket".to_string()); + let bucket_request_id = bucket_request.id.clone(); + + manager.active_heals.lock().await.insert( + cluster_request_id.clone(), + Arc::new(HealTask::from_request(cluster_request, storage.clone())), + ); + manager + .active_heals + .lock() + .await + .insert(bucket_request_id.clone(), Arc::new(HealTask::from_request(bucket_request, storage))); + + assert_eq!( + manager + .cancel_tasks_for_path("") + .await + .expect("root path should cancel active cluster task"), + 1 + ); + assert!(manager.active_heals.lock().await.get(&cluster_request_id).is_none()); + assert!(manager.active_heals.lock().await.get(&bucket_request_id).is_some()); + } + + #[tokio::test] + async fn test_cancel_tasks_for_empty_path_cancels_retrying_cluster_only() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + let mut cluster_request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::High); + cluster_request.retry_attempts = 1; + let cluster_request_id = cluster_request.id.clone(); + let cluster_cancel_token = insert_retrying_request(&manager, cluster_request).await; + + let mut bucket_request = HealRequest::bucket("bucket".to_string()); + bucket_request.retry_attempts = 1; + let bucket_request_id = bucket_request.id.clone(); + let bucket_cancel_token = insert_retrying_request(&manager, bucket_request).await; + + assert_eq!( + manager + .cancel_tasks_for_path("") + .await + .expect("root path should cancel retrying cluster task"), + 1 + ); + assert!(cluster_cancel_token.is_cancelled()); + assert!(!bucket_cancel_token.is_cancelled()); + assert!(manager.retrying_heals.lock().await.get(&cluster_request_id).is_none()); + assert!(manager.retrying_heals.lock().await.get(&bucket_request_id).is_some()); + } + #[tokio::test] async fn test_retrying_duplicate_token_can_query_and_cancel_original_retry() { let storage: Arc = Arc::new(MockStorage); diff --git a/crates/heal/src/heal/progress.rs b/crates/heal/src/heal/progress.rs index 81ce941da..e96664586 100644 --- a/crates/heal/src/heal/progress.rs +++ b/crates/heal/src/heal/progress.rs @@ -16,6 +16,7 @@ use serde::{Deserialize, Serialize}; use std::time::SystemTime; #[derive(Debug, Default, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] pub struct HealProgress { /// Objects scanned pub objects_scanned: u64, @@ -223,6 +224,22 @@ mod tests { assert!(progress.current_object.is_none()); } + #[test] + fn test_heal_progress_serializes_camel_case_fields() { + let mut progress = HealProgress::new(); + progress.update_progress(10, 8, 2, 1024); + progress.set_current_object(Some("test-bucket/test-object".to_string())); + + let json = serde_json::to_value(&progress).expect("progress should serialize"); + + assert_eq!(json["objectsScanned"], 10); + assert_eq!(json["objectsHealed"], 8); + assert_eq!(json["objectsFailed"], 2); + assert_eq!(json["bytesProcessed"], 1024); + assert_eq!(json["currentObject"], "test-bucket/test-object"); + assert!(json["progressPercentage"].is_number()); + } + #[test] fn test_heal_progress_is_completed_by_percentage() { let mut progress = HealProgress::new(); diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 1823bf4be..1dbf53647 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -51,6 +51,8 @@ const EVENT_HEAL_ERASURE_SET_RESULT: &str = "heal_erasure_set_result"; /// Heal type #[derive(Debug, Clone)] pub enum HealType { + /// Cluster heal + Cluster, /// Object heal Object { bucket: String, @@ -59,6 +61,8 @@ pub enum HealType { }, /// Bucket heal Bucket { bucket: String }, + /// Prefix heal + Prefix { bucket: String, prefix: String }, /// Erasure Set heal (includes disk format repair) ErasureSet { buckets: Vec, set_disk_id: String }, /// Metadata heal @@ -76,8 +80,10 @@ pub enum HealType { impl HealType { fn log_kind(&self) -> &'static str { match self { + Self::Cluster => "cluster", Self::Object { .. } => "object", Self::Bucket { .. } => "bucket", + Self::Prefix { .. } => "prefix", Self::ErasureSet { .. } => "erasure_set", Self::Metadata { .. } => "metadata", Self::MRF { .. } => "mrf", @@ -304,8 +310,10 @@ impl HealTask { pub fn metric_type_label(&self) -> &'static str { match &self.heal_type { + HealType::Cluster => "cluster", HealType::Object { .. } => "object", HealType::Bucket { .. } => "bucket", + HealType::Prefix { .. } => "prefix", HealType::ErasureSet { .. } => "erasure_set", HealType::Metadata { .. } => "metadata", HealType::MRF { .. } => "mrf", @@ -413,6 +421,13 @@ impl HealTask { Self::is_data_usage_cache_object(bucket, object) && Self::is_transient_lock_or_timeout_error(err) } + fn is_no_heal_required_error(err: &Error) -> bool { + match err { + Error::Other(message) => matches!(message.as_str(), "No heal required" | "No healing is required"), + _ => matches!(err.to_string().as_str(), "No heal required" | "No healing is required"), + } + } + async fn skip_data_usage_cache_heal_error(&self, bucket: &str, object: &str, err: &Error) -> bool { if !Self::should_skip_data_usage_cache_heal_error(bucket, object, err) { return false; @@ -478,12 +493,14 @@ impl HealTask { ); let result = match &self.heal_type { + HealType::Cluster => self.heal_cluster().await, HealType::Object { bucket, object, version_id, } => self.heal_object(bucket, object, version_id.as_deref()).await, HealType::Bucket { bucket } => self.heal_bucket(bucket).await, + 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, @@ -1104,7 +1121,7 @@ impl HealTask { self.record_result_item(result).await; if self.options.recursive { - self.heal_bucket_objects(bucket).await?; + self.heal_bucket_objects(bucket, "").await?; } if !self.options.recursive { @@ -1138,7 +1155,43 @@ impl HealTask { } } - async fn heal_bucket_objects(&self, bucket: &str) -> Result<()> { + async fn heal_cluster(&self) -> Result<()> { + debug!( + target: "rustfs::heal::task", + event = EVENT_HEAL_BUCKET_STAGE, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_TASK, + task_id = %self.id, + stage = "cluster_recursive", + "Heal cluster started" + ); + + let bucket_infos = self.await_with_control(self.storage.list_buckets()).await?; + for bucket_info in bucket_infos { + self.check_control_flags().await?; + self.heal_bucket(&bucket_info.name).await?; + } + + Ok(()) + } + + async fn heal_prefix(&self, bucket: &str, prefix: &str) -> Result<()> { + debug!( + target: "rustfs::heal::task", + event = EVENT_HEAL_BUCKET_STAGE, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_TASK, + task_id = %self.id, + bucket, + prefix, + stage = "prefix_recursive", + "Heal prefix started" + ); + + self.heal_bucket_objects(bucket, prefix).await + } + + async fn heal_bucket_objects(&self, bucket: &str, prefix: &str) -> Result<()> { let mut continuation_token: Option = None; let mut scanned = 0u64; let mut healed = 0u64; @@ -1162,7 +1215,7 @@ impl HealTask { let (objects, next_token, is_truncated) = self .await_with_control( self.storage - .list_objects_for_heal_page(bucket, "", continuation_token.as_deref()), + .list_objects_for_heal_page(bucket, prefix, continuation_token.as_deref()), ) .await?; @@ -1305,6 +1358,7 @@ impl HealTask { subsystem = LOG_SUBSYSTEM_TASK, task_id = %self.id, bucket, + prefix, scanned, healed, failed, @@ -1823,37 +1877,50 @@ impl HealTask { match format_result { Ok((result, error)) => { if let Some(e) = error { - error!( + if Self::is_no_heal_required_error(&e) { + debug!( + target: "rustfs::heal::task", + event = EVENT_HEAL_ERASURE_SET_RESULT, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_TASK, + task_id = %self.id, + set_disk_id, + result = "format_noop", + "Heal erasure set format repair skipped because no format heal was required" + ); + } else { + error!( + target: "rustfs::heal::task", + event = EVENT_HEAL_ERASURE_SET_RESULT, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_TASK, + task_id = %self.id, + set_disk_id, + result = "format_failed", + error = %e, + "Heal erasure set failed" + ); + { + let mut progress = self.progress.write().await; + progress.update_progress(4, 4, 0, 0); + } + return Err(Error::TaskExecutionFailed { + message: format!("Failed to heal disk format for {set_disk_id}: {e}"), + }); + } + } else { + debug!( target: "rustfs::heal::task", event = EVENT_HEAL_ERASURE_SET_RESULT, component = LOG_COMPONENT_HEAL, subsystem = LOG_SUBSYSTEM_TASK, task_id = %self.id, set_disk_id, - result = "format_failed", - error = %e, - "Heal erasure set failed" + drives_healed = result.after.drives.len(), + result = "format_ok", + "Heal erasure set format repaired" ); - { - let mut progress = self.progress.write().await; - progress.update_progress(4, 4, 0, 0); - } - return Err(Error::TaskExecutionFailed { - message: format!("Failed to heal disk format for {set_disk_id}: {e}"), - }); } - - debug!( - target: "rustfs::heal::task", - event = EVENT_HEAL_ERASURE_SET_RESULT, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_TASK, - task_id = %self.id, - set_disk_id, - drives_healed = result.after.drives.len(), - result = "format_ok", - "Heal erasure set format repaired" - ); } Err(Error::TaskCancelled) => return Err(Error::TaskCancelled), Err(Error::TaskTimeout) => return Err(Error::TaskTimeout), @@ -2060,6 +2127,8 @@ mod tests { healed_objects: Mutex>, bucket_heal_opts: Mutex>, object_heal_opts: Mutex>, + format_no_heal_required: Mutex, + listed_prefixes: Mutex>, } #[async_trait::async_trait] @@ -2108,7 +2177,10 @@ mod tests { } async fn list_buckets(&self) -> Result> { - Ok(Vec::new()) + Ok(vec![BucketInfo { + name: "bucket-a".to_string(), + ..Default::default() + }]) } async fn object_exists(&self, _bucket: &str, _object: &str) -> Result { @@ -2155,7 +2227,12 @@ mod tests { } async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option)> { - Ok((HealResultItem::default(), None)) + let no_heal_required = *self.format_no_heal_required.lock().unwrap(); + if no_heal_required { + Ok((HealResultItem::default(), Some(Error::other("No heal required")))) + } else { + Ok((HealResultItem::default(), None)) + } } async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result> { @@ -2165,9 +2242,10 @@ mod tests { async fn list_objects_for_heal_page( &self, bucket: &str, - _prefix: &str, + prefix: &str, continuation_token: Option<&str>, ) -> Result<(Vec, Option, bool)> { + self.listed_prefixes.lock().unwrap().push(prefix.to_string()); let mut listed = self.listed.lock().unwrap(); if continuation_token.is_none() && !*listed { *listed = true; @@ -2176,6 +2254,8 @@ mod tests { format!("{BUCKET_META_PREFIX}/{DATA_USAGE_CACHE_NAME}"), format!("{BUCKET_META_PREFIX}/bucket-metadata.bin"), ] + } else if prefix == "logs/" { + vec!["logs/object-a".to_string(), "logs/object-b".to_string()] } else { vec!["object-a".to_string(), "object-b".to_string()] }; @@ -2222,6 +2302,29 @@ mod tests { assert_eq!(result_items.iter().filter(|item| item.object_size == 1).count(), 2); } + #[tokio::test] + async fn test_cluster_heal_visits_bucket_objects() { + let storage = Arc::new(MockStorage::default()); + let request = HealRequest::new( + HealType::Cluster, + HealOptions { + recursive: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage.clone()); + + task.execute().await.expect("cluster heal should visit bucket objects"); + + assert_eq!( + storage.healed_objects.lock().unwrap().as_slice(), + ["object-a".to_string(), "object-b".to_string()] + ); + assert!(matches!(task.get_status().await, HealTaskStatus::Completed)); + } + #[tokio::test] async fn test_recursive_bucket_heal_does_not_remove_bucket_metadata() { let storage = Arc::new(MockStorage::default()); @@ -2258,6 +2361,34 @@ mod tests { assert!(object_opts.iter().all(|opts| opts.scan_mode == HealScanMode::Deep)); } + #[tokio::test] + async fn test_prefix_heal_lists_and_repairs_objects_under_prefix() { + let storage = Arc::new(MockStorage::default()); + let request = HealRequest::new( + HealType::Prefix { + bucket: "bucket-a".to_string(), + prefix: "logs/".to_string(), + }, + HealOptions { + recursive: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage.clone()); + + task.execute() + .await + .expect("prefix heal should scan and repair objects under the prefix"); + + assert_eq!(storage.listed_prefixes.lock().unwrap().as_slice(), ["logs/".to_string()]); + assert_eq!( + storage.healed_objects.lock().unwrap().as_slice(), + ["logs/object-a".to_string(), "logs/object-b".to_string()] + ); + } + #[tokio::test] async fn test_data_usage_cache_lock_timeout_does_not_fail_object_heal() { let storage = Arc::new(MockStorage::default()); @@ -2308,4 +2439,32 @@ mod tests { assert_eq!(progress.objects_scanned, 2); assert_eq!(progress.objects_failed, 0); } + + #[tokio::test] + async fn test_erasure_set_heal_continues_after_format_no_heal_required() { + let storage = Arc::new(MockStorage::default()); + *storage.format_no_heal_required.lock().unwrap() = true; + let request = HealRequest::new( + HealType::ErasureSet { + buckets: Vec::new(), + set_disk_id: "pool_0_set_0".to_string(), + }, + HealOptions { + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage); + + let err = task + .heal_erasure_set(Vec::new(), "pool_0_set_0".to_string()) + .await + .expect_err("test mock should fail after format when resolving resume disk"); + + assert!( + err.to_string().contains("not implemented in tests"), + "erasure-set heal should continue past NoHealRequired format result, got: {err}" + ); + } } diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index c8e50c094..662505a3f 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -194,6 +194,8 @@ struct HealTaskStatus { heal_settings: HealOpts, #[serde(skip_serializing_if = "Vec::is_empty")] items: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + progress: Option, } #[derive(Debug, Serialize)] @@ -211,6 +213,8 @@ struct HealTaskStatusPayload { summary: String, #[serde(default)] items: Vec, + #[serde(default)] + progress: Option, } fn map_heal_response(result: Option) -> S3Result<(StatusCode, Vec)> { @@ -253,6 +257,7 @@ fn encode_heal_task_status( failure_detail: String, heal_settings: HealOpts, items: Vec, + progress: Option, ) -> S3Result> { encode_json(&HealTaskStatus { summary, @@ -260,12 +265,14 @@ fn encode_heal_task_status( start_time: current_rfc3339_time()?, heal_settings, items, + progress, }) } fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest { let root_erasure_set_target = hip.bucket.is_empty() && hip.obj_prefix.is_empty() && matches!((hip.hs.pool, hip.hs.set), (Some(_), Some(_))); + let root_cluster_target = hip.bucket.is_empty() && hip.obj_prefix.is_empty() && hip.hs.pool.is_none() && hip.hs.set.is_none(); let recursive = if (!hip.bucket.is_empty() && hip.obj_prefix.is_empty()) || root_erasure_set_target { true } else { @@ -291,7 +298,7 @@ fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest { heal_request.remove_corrupted = Some(hip.hs.remove); heal_request.recreate_missing = Some(hip.hs.recreate); heal_request.update_parity = Some(hip.hs.update_parity); - heal_request.recursive = Some(recursive); + heal_request.recursive = Some(recursive || root_cluster_target); heal_request.dry_run = Some(hip.hs.dry_run); heal_request.source = HealRequestSource::Admin; heal_request @@ -299,15 +306,15 @@ fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest { fn heal_channel_response_status( response: &rustfs_common::heal_channel::HealChannelResponse, -) -> (String, Vec) { +) -> (String, Vec, Option) { let Some(data) = response.data.as_deref() else { - return ("running".to_string(), Vec::new()); + return ("running".to_string(), Vec::new(), None); }; if let Ok(payload) = serde_json::from_slice::(data) && !payload.summary.is_empty() { - return (payload.summary, payload.items); + return (payload.summary, payload.items, payload.progress); } let summary = std::str::from_utf8(data) @@ -315,7 +322,7 @@ fn heal_channel_response_status( .filter(|summary| !summary.is_empty()) .unwrap_or("running") .to_string(); - (summary, Vec::new()) + (summary, Vec::new(), None) } #[cfg(test)] @@ -330,6 +337,11 @@ fn heal_channel_response_items( heal_channel_response_status(response).1 } +#[cfg(test)] +fn heal_channel_response_progress(response: &rustfs_common::heal_channel::HealChannelResponse) -> Option { + heal_channel_response_status(response).2 +} + fn encode_background_heal_status( info: &BackgroundHealInfo, heal_operations: rustfs_heal::HealOperationsSnapshot, @@ -362,20 +374,16 @@ fn validate_heal_request_mode(hip: &HealInitParams) -> S3Result<()> { (Some(_), None) | (None, Some(_)) => { Err(s3_error!(InvalidRequest, "root heal erasure-set target requires both pool and set")) } - (None, None) => Err(s3_error!(InvalidRequest, "starting heal without a bucket target is not supported")), + (None, None) if hip.hs.recursive => Ok(()), + (None, None) => Err(s3_error!(InvalidRequest, "root heal requires recursive=true or a bucket target")), }; } Ok(()) } -fn should_handle_root_heal_directly(hip: &HealInitParams) -> bool { - hip.bucket.is_empty() - && hip.obj_prefix.is_empty() - && hip.client_token.is_empty() - && !hip.force_stop - && hip.hs.pool.is_none() - && hip.hs.set.is_none() +fn should_handle_root_heal_directly(_hip: &HealInitParams) -> bool { + false } fn map_root_heal_status(heal_err: Option) -> S3Result<()> { @@ -529,9 +537,14 @@ impl Operation for HealHandler { spawn_traced(async move { match rustfs_common::heal_channel::query_heal_status(heal_path_str, client_token).await { Ok(response) if response.success => { - let (summary, items) = heal_channel_response_status(&response); - let resp_bytes = - encode_heal_task_status(summary, response.error.unwrap_or_default(), HealOpts::default(), items); + let (summary, items, progress) = heal_channel_response_status(&response); + let resp_bytes = encode_heal_task_status( + summary, + response.error.unwrap_or_default(), + HealOpts::default(), + items, + progress, + ); match resp_bytes { Ok(resp_bytes) => { let _ = tx_clone @@ -582,8 +595,8 @@ impl Operation for HealHandler { let resp_bytes = if client_token.is_empty() { encode_heal_start_success(response.request_id, client_address) } else { - let (summary, items) = heal_channel_response_status(&response); - encode_heal_task_status(summary, response.error.unwrap_or_default(), heal_settings, items) + let (summary, items, progress) = heal_channel_response_status(&response); + encode_heal_task_status(summary, response.error.unwrap_or_default(), heal_settings, items, progress) }; match resp_bytes { Ok(resp_bytes) => { @@ -734,8 +747,9 @@ mod tests { use super::extract_heal_init_params; use super::{ HealInitParams, HealResp, build_heal_channel_request, encode_background_heal_status, encode_heal_start_success, - encode_heal_task_status, heal_channel_response_items, heal_channel_response_summary, json_response, map_heal_response, - map_root_heal_status, should_handle_root_heal_directly, validate_heal_request_mode, validate_heal_target, + encode_heal_task_status, heal_channel_response_items, heal_channel_response_progress, heal_channel_response_summary, + json_response, map_heal_response, map_root_heal_status, should_handle_root_heal_directly, validate_heal_request_mode, + validate_heal_target, }; use bytes::Bytes; use http::StatusCode; @@ -918,6 +932,31 @@ mod tests { assert_eq!(request.recreate_missing, Some(true)); } + #[test] + fn test_root_recursive_heal_channel_request_targets_cluster() { + let hip = HealInitParams { + hs: HealOpts { + recursive: true, + scan_mode: HealScanMode::Deep, + recreate: true, + ..Default::default() + }, + force_start: true, + ..Default::default() + }; + + let request = build_heal_channel_request(&hip); + + assert_eq!(request.bucket, ""); + assert_eq!(request.disk, None); + assert_eq!(request.object_prefix, None); + assert_eq!(request.priority, HealChannelPriority::High); + assert_eq!(request.source, HealRequestSource::Admin); + assert_eq!(request.scan_mode, Some(HealScanMode::Deep)); + assert_eq!(request.recursive, Some(true)); + assert_eq!(request.recreate_missing, Some(true)); + } + #[test] fn test_bucket_heal_channel_request_defaults_to_recursive() { let hip = HealInitParams { @@ -973,10 +1012,22 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); assert!( err.to_string() - .contains("starting heal without a bucket target is not supported") + .contains("root heal requires recursive=true or a bucket target") ); } + #[test] + fn test_validate_heal_request_mode_allows_root_recursive_heal_start() { + validate_heal_request_mode(&HealInitParams { + hs: HealOpts { + recursive: true, + ..Default::default() + }, + ..Default::default() + }) + .expect("root recursive heal should start tracked cluster heal"); + } + #[test] fn test_validate_heal_request_mode_allows_root_erasure_set_target() { validate_heal_request_mode(&HealInitParams { @@ -1009,9 +1060,9 @@ mod tests { } #[test] - fn test_should_handle_root_heal_directly_for_root_start_modes() { - assert!(should_handle_root_heal_directly(&HealInitParams::default())); - assert!(should_handle_root_heal_directly(&HealInitParams { + fn test_should_handle_root_heal_directly_is_disabled_for_root_start_modes() { + assert!(!should_handle_root_heal_directly(&HealInitParams::default())); + assert!(!should_handle_root_heal_directly(&HealInitParams { force_start: true, ..Default::default() })); @@ -1169,9 +1220,14 @@ mod tests { #[test] fn test_encode_heal_task_status_uses_client_wire_shape() { - let encoded = - encode_heal_task_status("Heal status query accepted".to_string(), String::new(), HealOpts::default(), Vec::new()) - .expect("status response should serialize"); + let encoded = encode_heal_task_status( + "Heal status query accepted".to_string(), + String::new(), + HealOpts::default(), + Vec::new(), + None, + ) + .expect("status response should serialize"); let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize"); assert_eq!(json["summary"], "Heal status query accepted"); @@ -1182,6 +1238,26 @@ mod tests { OffsetDateTime::parse(start_time, &Rfc3339).expect("startTime should be RFC3339"); } + #[test] + fn test_encode_heal_task_status_preserves_progress() { + let progress = json!({ + "objectsScanned": 7, + "objectsHealed": 3, + "currentObject": "bucket-a/object-a" + }); + let encoded = encode_heal_task_status( + "running".to_string(), + String::new(), + HealOpts::default(), + Vec::new(), + Some(progress.clone()), + ) + .expect("status response should serialize"); + let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize"); + + assert_eq!(json["progress"], progress); + } + #[test] fn test_build_heal_channel_request_preserves_client_options() { let hip = HealInitParams { @@ -1264,6 +1340,29 @@ mod tests { assert_eq!(items[0].object_size, 1024); } + #[test] + fn test_heal_channel_response_status_preserves_progress() { + let progress = serde_json::json!({ + "objectsScanned": 4, + "objectsHealed": 2, + "currentObject": "bucket-a/object-a" + }); + let payload = serde_json::json!({ + "summary": "running", + "items": [], + "progress": progress + }); + let response = rustfs_common::heal_channel::create_heal_response( + "token".to_string(), + true, + Some(serde_json::to_vec(&payload).expect("payload should serialize")), + None, + ); + + assert_eq!(heal_channel_response_summary(&response), "running"); + assert_eq!(heal_channel_response_progress(&response), Some(progress)); + } + #[test] fn test_json_response_sets_application_json_content_type() { let response = json_response(StatusCode::OK, b"{}".to_vec());