From 7a4a3d27c6c8fc8814f5e44817022bf336c7e22e Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Wed, 12 Aug 2026 20:46:35 +0800 Subject: [PATCH] fix(heal): cancel cluster tasks from root stop (#5978) Co-authored-by: Henry Guo --- crates/heal/src/heal/channel.rs | 49 +++++++++++++++++++++- crates/heal/src/heal/manager.rs | 14 ++++++- rustfs/src/admin/handlers/heal.rs | 68 +++++++++++++++++++++---------- 3 files changed, 106 insertions(+), 25 deletions(-) diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index a11b1b425..9603b81c0 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -1778,9 +1778,22 @@ mod tests { } #[tokio::test] - async fn test_process_cancel_request_treats_unknown_path_as_stopped() { + async fn test_process_cancel_request_cancels_cluster_task_for_legacy_root_path() { let heal_manager = create_test_heal_manager(); - let processor = HealChannelProcessor::new(heal_manager); + let cluster_request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::High); + let cluster_task_id = cluster_request.id.clone(); + let bucket_request = HealRequest::bucket("bucket".to_string()); + let bucket_task_id = bucket_request.id.clone(); + heal_manager + .submit_heal_request(cluster_request) + .await + .expect("cluster request should be accepted"); + heal_manager + .submit_heal_request(bucket_request) + .await + .expect("bucket request should be accepted"); + + let processor = HealChannelProcessor::new(heal_manager.clone()); let (tx, rx) = oneshot::channel(); processor @@ -1796,6 +1809,38 @@ mod tests { assert_eq!(response.request_id, "."); assert_eq!(response.data.as_deref(), Some("stopped".as_bytes())); assert!(response.error.is_none()); + assert!(matches!( + heal_manager.get_task_status(&cluster_task_id).await, + Err(crate::Error::TaskNotFound { .. }) + )); + assert_eq!( + heal_manager + .get_task_status(&bucket_task_id) + .await + .expect("bucket request should not match the root path"), + HealTaskStatus::Pending + ); + } + + #[tokio::test] + async fn test_process_cancel_request_treats_unknown_path_as_stopped() { + let heal_manager = create_test_heal_manager(); + let processor = HealChannelProcessor::new(heal_manager); + let (tx, rx) = oneshot::channel(); + + processor + .process_cancel_request("missing".to_string(), String::new(), tx) + .await + .expect("cancel should process"); + + let response = rx + .await + .expect("oneshot should resolve") + .expect("cancel response should be returned"); + assert!(response.success); + assert_eq!(response.request_id, "missing"); + assert_eq!(response.data.as_deref(), Some("stopped".as_bytes())); + assert!(response.error.is_none()); } #[tokio::test] diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index fbdb4b35c..fa831b129 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -52,6 +52,7 @@ const EVENT_HEAL_MAINLINE_THROTTLE: &str = "heal_mainline_throttle"; const EVENT_HEAL_SCHEDULER_STATE: &str = "heal_scheduler_state"; const EVENT_HEAL_QUEUE_STATE: &str = "heal_queue_state"; const EVENT_HEAL_UNCLEAN_SHUTDOWN: &str = "heal_unclean_shutdown"; +const LEGACY_ROOT_HEAL_PATH: &str = "."; const MAX_RECOVERABLE_HEAL_RETRIES: u32 = 3; const MAX_RECOVERABLE_HEAL_RETRY_DELAY: Duration = Duration::from_secs(30); @@ -601,7 +602,7 @@ 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() { + if heal_path.is_empty() || heal_path == LEGACY_ROOT_HEAL_PATH { return matches!(heal_type, HealType::Cluster); } @@ -5110,6 +5111,17 @@ mod tests { assert!(manager.retrying_heals.lock().await.get(&bucket_request_id).is_some()); } + #[test] + fn test_heal_type_matches_path_accepts_legacy_root() { + assert!(heal_type_matches_path(&HealType::Cluster, LEGACY_ROOT_HEAL_PATH)); + assert!(!heal_type_matches_path( + &HealType::Bucket { + bucket: "bucket".to_string(), + }, + LEGACY_ROOT_HEAL_PATH, + )); + } + #[tokio::test] async fn test_retrying_duplicate_token_can_query_and_cancel_original_retry() { let storage: Arc = Arc::new(MockStorage); diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 79dd0f228..d9e7d1f7e 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -53,6 +53,7 @@ const LOG_SUBSYSTEM_HEAL_ADMIN: &str = "heal_admin"; const EVENT_ADMIN_REQUEST_REJECTED: &str = "admin_request_rejected"; const EVENT_ADMIN_REQUEST_FAILED: &str = "admin_request_failed"; const EVENT_ADMIN_RESPONSE_EMITTED: &str = "admin_response_emitted"; +const LEGACY_ROOT_HEAL_RESPONSE_ID: &str = "."; const PEER_HEAL_STATUS_TIMEOUT: Duration = Duration::from_secs(5); pub(crate) const REPLACEMENT_RECOVERY_STATUS_ROUTE_SUFFIX: &str = "/v4/heal/replacement-recovery"; const REPLACEMENT_RECOVERY_STATUS_CONTRACT_VERSION: u32 = 2; @@ -150,6 +151,26 @@ fn validate_heal_target(bucket: &str, obj_prefix: &str) -> S3Result<()> { Ok(()) } +fn encode_heal_control_path(bucket: &str, obj_prefix: &str) -> String { + if bucket.is_empty() && obj_prefix.is_empty() { + return String::new(); + } + + path_join(&[PathBuf::from(bucket), PathBuf::from(obj_prefix)]) + .to_string_lossy() + .into_owned() +} + +fn heal_control_response_id(heal_path: &str, client_token: &str) -> String { + if !client_token.is_empty() { + return client_token.to_string(); + } + if heal_path.is_empty() { + return LEGACY_ROOT_HEAL_RESPONSE_ID.to_string(); + } + heal_path.to_string() +} + pub fn register_heal_route(r: &mut S3Router) -> std::io::Result<()> { // Some APIs are only available in EC mode // if is_dist_erasure().await || is_erasure().await { @@ -1368,9 +1389,8 @@ impl Operation for HealHandler { "start_heal" }; - let heal_path = path_join(&[PathBuf::from(hip.bucket.clone()), PathBuf::from(hip.obj_prefix.clone())]); + let heal_path = encode_heal_control_path(&hip.bucket, &hip.obj_prefix); if !hip.client_token.is_empty() && !hip.force_start && !hip.force_stop { - let heal_path_str = heal_path.to_str().unwrap_or_default().to_string(); let client_token = hip.client_token.clone(); let request_id = uuid::Uuid::new_v4().to_string(); let context = app_context @@ -1380,7 +1400,7 @@ impl Operation for HealHandler { let envelope = rustfs_protos::heal_control::Envelope::query( request_id.clone(), new_heal_control_metadata(&route)?, - heal_path_str, + heal_path, client_token.clone(), ) .map_err(|err| s3_error!(InternalError, "encode heal control query failed: {err}"))?; @@ -1412,7 +1432,6 @@ impl Operation for HealHandler { ); return Ok(json_response(StatusCode::OK, body)); } else if hip.force_stop { - let heal_path_str = heal_path.to_str().unwrap_or_default().to_string(); let client_token = hip.client_token.clone(); let request_id = uuid::Uuid::new_v4().to_string(); let context = app_context @@ -1422,22 +1441,12 @@ impl Operation for HealHandler { let envelope = rustfs_protos::heal_control::Envelope::cancel( request_id.clone(), new_heal_control_metadata(&route)?, - heal_path_str, + heal_path.clone(), client_token.clone(), ) .map_err(|err| s3_error!(InternalError, "encode heal control cancel failed: {err}"))?; - let response = submit_cluster_heal_channel_command( - context, - route, - envelope, - &request_id, - if client_token.is_empty() { - heal_path.to_string_lossy().into_owned() - } else { - client_token.clone() - }, - ) - .await?; + let response_id = heal_control_response_id(&heal_path, &client_token); + let response = submit_cluster_heal_channel_command(context, route, envelope, &request_id, response_id).await?; if !response.success { return Err(s3_error!( InternalError, @@ -1579,11 +1588,12 @@ mod tests { use super::{ BackgroundHealProgress, HealInitParams, HealResp, HealRuntimeState, aggregate_cluster_heal_status, aggregate_replacement_recovery_cluster_status, background_heal_runtime_state, build_heal_channel_request, - build_replacement_recovery_status_response, encode_background_heal_status, encode_heal_start_success, - encode_heal_task_status, execute_after_heal_control_capability, heal_channel_response_items, - heal_channel_response_progress, heal_channel_response_summary, json_response, map_heal_response, map_root_heal_status, - merge_peer_heal_statuses, peer_topology_complete, query_peer_heal_status, query_peer_replacement_recovery_status, - reject_heal_admission, should_handle_root_heal_directly, validate_heal_request_mode, validate_heal_target, + build_replacement_recovery_status_response, encode_background_heal_status, encode_heal_control_path, + encode_heal_start_success, encode_heal_task_status, execute_after_heal_control_capability, heal_channel_response_items, + heal_channel_response_progress, heal_channel_response_summary, heal_control_response_id, json_response, + map_heal_response, map_root_heal_status, merge_peer_heal_statuses, peer_topology_complete, query_peer_heal_status, + query_peer_replacement_recovery_status, reject_heal_admission, should_handle_root_heal_directly, + validate_heal_request_mode, validate_heal_target, }; use crate::admin::storage_api::error::StorageError; use crate::storage::rpc::node_service::heal::{ @@ -2112,6 +2122,20 @@ mod tests { .expect("root heal cancel should be accepted"); } + #[test] + fn test_encode_heal_control_path_keeps_root_empty() { + assert_eq!(encode_heal_control_path("", ""), ""); + assert_eq!(encode_heal_control_path("bucket", ""), "bucket"); + assert_eq!(encode_heal_control_path("bucket", "prefix"), "bucket/prefix"); + } + + #[test] + fn test_heal_control_response_id_preserves_existing_contract() { + assert_eq!(heal_control_response_id("", ""), "."); + assert_eq!(heal_control_response_id("bucket", ""), "bucket"); + assert_eq!(heal_control_response_id("", "task-id"), "task-id"); + } + #[test] fn test_extract_heal_init_params_rejects_prefix_without_bucket() { let err = validate_heal_target("", "prefix").expect_err("must reject empty bucket");