fix(heal): cancel cluster tasks from root stop (#5978)

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>

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]

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<dyn HealStorageAPI> = Arc::new(MockStorage);

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<AdminOperation>) -> 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");