diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index 9d749b6a4..947c329b5 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -262,7 +262,13 @@ impl HealChannelProcessor { "Heal query received" ); - let (summary, detail, items) = match self.heal_manager.get_task_report_for_path(&heal_path, &client_token).await { + let report = if heal_path.trim_matches('/').is_empty() { + self.heal_manager.get_task_report(&client_token).await + } else { + self.heal_manager.get_task_report_for_path(&heal_path, &client_token).await + }; + + let (summary, detail, items) = match report { Ok(HealTaskReport { status: HealTaskStatus::Pending | HealTaskStatus::Running, result_items, @@ -291,7 +297,9 @@ impl HealChannelProcessor { status: HealTaskStatus::Failed { error }, result_items, }) => ("stopped".to_string(), Some(error), result_items), - Err(crate::Error::TaskNotFound { .. }) => ("finished".to_string(), None, Vec::new()), + Err(crate::Error::TaskNotFound { .. }) => { + ("notFound".to_string(), Some("heal task not found or expired".to_string()), Vec::new()) + } Err(crate::Error::InvalidClientToken) => { let response = HealChannelResponse { request_id: client_token, @@ -908,7 +916,7 @@ mod tests { } #[tokio::test] - async fn test_process_query_request_reports_finished_when_task_is_not_active() { + async fn test_process_query_request_reports_not_found_when_task_is_unknown() { let heal_manager = create_test_heal_manager(); let processor = HealChannelProcessor::new(heal_manager); let (tx, rx) = oneshot::channel(); @@ -924,10 +932,11 @@ mod tests { .expect("query response should be returned"); assert!(response.success); assert_eq!(response.request_id, "completed-token"); + assert_eq!(response.error.as_deref(), Some("heal task not found or expired")); let payload: serde_json::Value = serde_json::from_slice(response.data.as_deref().expect("status payload should be present")) .expect("status payload should be json"); - assert_eq!(payload["summary"], "finished"); + assert_eq!(payload["summary"], "notFound"); assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0); } @@ -1018,7 +1027,46 @@ mod tests { let payload: serde_json::Value = serde_json::from_slice(response.data.as_deref().expect("status payload should be present")) .expect("status payload should be json"); - assert_eq!(payload["summary"], "finished"); + assert_eq!(payload["summary"], "notFound"); + assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0); + } + + #[tokio::test] + async fn test_process_query_request_empty_path_uses_client_token_directly() { + let heal_manager = create_test_heal_manager(); + let request = HealRequest::new( + HealType::ErasureSet { + buckets: vec![], + set_disk_id: "pool_0_set_1".to_string(), + }, + HealOptions::default(), + HealPriority::High, + ); + let task_id = request.id.clone(); + heal_manager + .submit_heal_request(request) + .await + .expect("request should be accepted"); + + let processor = HealChannelProcessor::new(heal_manager); + let (tx, rx) = oneshot::channel(); + + processor + .process_query_request(String::new(), task_id.clone(), tx) + .await + .expect("query should process"); + + let response = rx + .await + .expect("oneshot should resolve") + .expect("query response should be returned"); + assert!(response.success); + assert_eq!(response.request_id, task_id); + assert!(response.error.is_none()); + let payload: serde_json::Value = + serde_json::from_slice(response.data.as_deref().expect("status payload should be present")) + .expect("status payload should be json"); + assert_eq!(payload["summary"], "running"); assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0); } diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index aebe31906..d8fed1446 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -1200,6 +1200,65 @@ impl HealManager { }) } + pub async fn get_task_report(&self, task_id: &str) -> Result { + let canonical_task_id = self.canonical_task_id(task_id).await; + { + let active_heals = self.active_heals.lock().await; + if let Some(task) = active_heals.get(&canonical_task_id) { + return Ok(HealTaskReport { + status: task.get_status().await, + result_items: task.get_result_items().await, + }); + } + } + + { + let retrying_heals = self.retrying_heals.lock().await; + if let Some(retrying) = retrying_heals.get(&canonical_task_id) { + return Ok(HealTaskReport { + status: retrying.status(), + result_items: Vec::new(), + }); + } + } + + { + let mut completed_heals = self.completed_heals.lock().await; + prune_completed_heal_statuses(&mut completed_heals); + if let Some(completed) = completed_heals.get(&canonical_task_id) + && completed_status_is_retrying(&completed.status) + { + return Ok(HealTaskReport { + status: completed.status.clone(), + result_items: completed.result_items.clone(), + }); + } + } + + { + let queue = self.heal_queue.lock().await; + if queue.contains_request_id(&canonical_task_id) { + return Ok(HealTaskReport { + status: HealTaskStatus::Pending, + result_items: Vec::new(), + }); + } + } + + let mut completed_heals = self.completed_heals.lock().await; + prune_completed_heal_statuses(&mut completed_heals); + if let Some(completed) = completed_heals.get(&canonical_task_id) { + return Ok(HealTaskReport { + status: completed.status.clone(), + result_items: completed.result_items.clone(), + }); + } + + Err(Error::TaskNotFound { + task_id: task_id.to_string(), + }) + } + pub async fn get_task_report_for_path(&self, heal_path: &str, task_id: &str) -> Result { let canonical_task_id = self.canonical_task_id(task_id).await; { @@ -3245,6 +3304,35 @@ mod tests { )); } + #[tokio::test] + async fn test_get_task_report_queries_queued_task_by_token_without_path() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + let request = HealRequest::new( + HealType::ErasureSet { + buckets: vec![], + set_disk_id: "pool_0_set_1".to_string(), + }, + HealOptions::default(), + HealPriority::High, + ); + let request_id = request.id.clone(); + + manager + .submit_heal_request(request) + .await + .expect("request should be accepted"); + + let report = manager + .get_task_report(&request_id) + .await + .expect("queued task should be queryable by token"); + + assert_eq!(report.status, HealTaskStatus::Pending); + assert!(report.result_items.is_empty()); + } + #[tokio::test] async fn test_get_task_status_reads_recent_completed_status() { let storage: Arc = Arc::new(MockStorage); diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index f87481761..664017a15 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -26,6 +26,7 @@ use hyper::{Method, StatusCode}; use matchit::Params; use rustfs_common::heal_channel::{HealChannelPriority, HealChannelRequest, HealOpts, HealRequestSource}; use rustfs_config::MAX_HEAL_REQUEST_SIZE; +use rustfs_heal::heal::utils::format_set_disk_id; use rustfs_policy::policy::action::{Action, AdminAction}; use rustfs_scanner::scanner::{BackgroundHealInfo, read_background_heal_info}; use rustfs_storage_api::HealOperations as _; @@ -247,7 +248,9 @@ fn encode_heal_task_status( } fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest { - let recursive = if !hip.bucket.is_empty() && hip.obj_prefix.is_empty() { + let root_erasure_set_target = + hip.bucket.is_empty() && hip.obj_prefix.is_empty() && matches!((hip.hs.pool, hip.hs.set), (Some(_), Some(_))); + let recursive = if (!hip.bucket.is_empty() && hip.obj_prefix.is_empty()) || root_erasure_set_target { true } else { hip.hs.recursive @@ -265,6 +268,9 @@ fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest { heal_request.pool_index = hip.hs.pool; heal_request.set_index = hip.hs.set; + if root_erasure_set_target && let (Some(pool), Some(set)) = (hip.hs.pool, hip.hs.set) { + heal_request.disk = Some(format_set_disk_id(pool, set)); + } heal_request.scan_mode = Some(hip.hs.scan_mode); heal_request.remove_corrupted = Some(hip.hs.remove); heal_request.recreate_missing = Some(hip.hs.recreate); @@ -335,14 +341,25 @@ fn encode_background_heal_status( fn validate_heal_request_mode(hip: &HealInitParams) -> S3Result<()> { if hip.bucket.is_empty() && hip.client_token.is_empty() && !hip.force_stop { - return Err(s3_error!(InvalidRequest, "starting heal without a bucket target is not supported")); + return match (hip.hs.pool, hip.hs.set) { + (Some(_), Some(_)) => Ok(()), + (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")), + }; } 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.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 map_root_heal_status(heal_err: Option) -> S3Result<()> { @@ -836,6 +853,33 @@ mod tests { assert_eq!(request.set_index, Some(2)); } + #[test] + fn test_root_heal_channel_request_with_pool_set_targets_erasure_set() { + let hip = HealInitParams { + hs: HealOpts { + scan_mode: HealScanMode::Deep, + recreate: true, + pool: Some(1), + set: Some(2), + ..Default::default() + }, + force_start: true, + ..Default::default() + }; + + let request = build_heal_channel_request(&hip); + + assert_eq!(request.bucket, ""); + assert_eq!(request.disk.as_deref(), Some("pool_1_set_2")); + assert_eq!(request.priority, HealChannelPriority::High); + assert_eq!(request.source, HealRequestSource::Admin); + assert_eq!(request.pool_index, Some(1)); + assert_eq!(request.set_index, Some(2)); + 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 { @@ -895,6 +939,37 @@ mod tests { ); } + #[test] + fn test_validate_heal_request_mode_allows_root_erasure_set_target() { + validate_heal_request_mode(&HealInitParams { + hs: HealOpts { + pool: Some(1), + set: Some(2), + ..Default::default() + }, + ..Default::default() + }) + .expect("root heal with pool and set should start tracked erasure-set rebuild"); + } + + #[test] + fn test_validate_heal_request_mode_rejects_partial_root_erasure_set_target() { + let err = validate_heal_request_mode(&HealInitParams { + hs: HealOpts { + pool: Some(1), + ..Default::default() + }, + ..Default::default() + }) + .expect_err("root heal must provide both pool and set"); + + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + assert!( + err.to_string() + .contains("root heal erasure-set target requires both pool and set") + ); + } + #[test] fn test_should_handle_root_heal_directly_for_root_start_modes() { assert!(should_handle_root_heal_directly(&HealInitParams::default())); @@ -918,6 +993,14 @@ mod tests { bucket: "bucket".to_string(), ..Default::default() })); + assert!(!should_handle_root_heal_directly(&HealInitParams { + hs: HealOpts { + pool: Some(1), + set: Some(2), + ..Default::default() + }, + ..Default::default() + })); } #[test]