fix(heal): coordinate cluster-wide control operations (#5003)

crates/heal/src/heal/channel.rs

@@ -34,6 +34,7 @@ const LOG_SUBSYSTEM_CHANNEL: &str = "channel";
 const EVENT_HEAL_CHANNEL_STATE: &str = "heal_channel_state";
 const EVENT_HEAL_CHANNEL_REQUEST: &str = "heal_channel_request";
 const EVENT_HEAL_CHANNEL_RESPONSE: &str = "heal_channel_response";
+const MAX_HEAL_STATUS_PAYLOAD_SIZE: usize = 8 * 1024 * 1024;
 
 fn admission_response(request_id: String, admission: HealAdmissionResult) -> HealChannelResponse {
     let (success, error) = match admission {
@@ -61,11 +62,56 @@ pub struct HealChannelProcessor {
 }
 
 #[derive(Serialize)]
-struct HealTaskStatusPayload {
-    summary: String,
-    items: Vec<HealResultItem>,
+struct HealTaskStatusPayload<'a> {
+    summary: &'a str,
+    items: &'a [HealResultItem],
+    truncated: bool,
     #[serde(skip_serializing_if = "Option::is_none")]
-    progress: Option<HealProgress>,
+    progress: Option<&'a HealProgress>,
+}
+
+fn encode_heal_task_status_payload(
+    summary: &str,
+    mut items: Vec<HealResultItem>,
+    progress: Option<&HealProgress>,
+    mut truncated: bool,
+) -> Result<(Vec<u8>, bool)> {
+    loop {
+        let data = serde_json::to_vec(&HealTaskStatusPayload {
+            summary,
+            items: &items,
+            truncated,
+            progress,
+        })
+        .map_err(|e| Error::Serialization(format!("failed to serialize heal task status: {e}")))?;
+        if data.len() <= MAX_HEAL_STATUS_PAYLOAD_SIZE {
+            return Ok((data, truncated));
+        }
+        if items.is_empty() {
+            return Err(Error::Serialization("heal task status metadata exceeds size limit".to_string()));
+        }
+        truncated = true;
+        items.truncate(items.len() / 2);
+    }
+}
+
+fn heal_status_detail(detail: Option<String>, truncated: bool) -> Option<String> {
+    if !truncated {
+        return detail;
+    }
+    let truncation = "heal result items were truncated";
+    Some(detail.map_or_else(|| truncation.to_string(), |detail| format!("{detail}; {truncation}")))
+}
+
+fn encode_heal_status_response(
+    summary: &str,
+    items: Vec<HealResultItem>,
+    progress: Option<&HealProgress>,
+    detail: Option<String>,
+    truncated: bool,
+) -> Result<(Vec<u8>, Option<String>)> {
+    let (data, truncated) = encode_heal_task_status_payload(summary, items, progress, truncated)?;
+    Ok((data, heal_status_detail(detail, truncated)))
 }
 
 impl HealChannelProcessor {
@@ -79,6 +125,37 @@ impl HealChannelProcessor {
         }
     }
 
+    /// Execute a start directly against the manager without entering the
+    /// process-global unbounded command queue.
+    pub async fn execute_start_request(&self, request: HealChannelRequest) -> Result<HealAdmissionReceipt> {
+        let (response_tx, response_rx) = oneshot::channel();
+        self.process_start_request(request, false, true, response_tx).await?;
+        response_rx
+            .await
+            .map_err(|err| Error::other(format!("heal receipt channel closed: {err}")))?
+            .map_err(Error::other)
+    }
+
+    /// Execute a token query directly against the manager.
+    pub async fn execute_query_request(&self, heal_path: String, client_token: String) -> Result<HealChannelResponse> {
+        let (response_tx, response_rx) = oneshot::channel();
+        self.process_query_request(heal_path, client_token, response_tx).await?;
+        response_rx
+            .await
+            .map_err(|err| Error::other(format!("heal query channel closed: {err}")))?
+            .map_err(Error::other)
+    }
+
+    /// Execute cancellation directly against the manager.
+    pub async fn execute_cancel_request(&self, heal_path: String, client_token: String) -> Result<HealChannelResponse> {
+        let (response_tx, response_rx) = oneshot::channel();
+        self.process_cancel_request(heal_path, client_token, response_tx).await?;
+        response_rx
+            .await
+            .map_err(|err| Error::other(format!("heal cancel channel closed: {err}")))?
+            .map_err(Error::other)
+    }
+
     /// Start processing legacy heal channel requests.
     pub async fn start(&mut self, receiver: HealChannelReceiver) -> Result<()> {
         let (receipt_sender, receipt_receiver) = mpsc::unbounded_channel();
@@ -326,46 +403,66 @@ impl HealChannelProcessor {
             self.heal_manager.get_task_report_for_path(&heal_path, &client_token).await
         };
 
-        let (summary, detail, items, progress) = match report {
+        let (summary, detail, items, truncated, progress) = match report {
             Ok(HealTaskReport {
                 status: HealTaskStatus::Pending | HealTaskStatus::Running,
                 result_items,
+                result_items_truncated,
                 progress,
-            }) => ("running".to_string(), None, result_items, progress),
+            }) => ("running".to_string(), None, result_items, result_items_truncated, progress),
             Ok(HealTaskReport {
                 status: HealTaskStatus::Retrying { error, retry_attempt },
                 result_items,
+                result_items_truncated,
                 progress,
             }) => (
                 "running".to_string(),
                 Some(format!("heal task retrying after recoverable failure, attempt {retry_attempt}: {error}")),
                 result_items,
+                result_items_truncated,
                 progress,
             ),
             Ok(HealTaskReport {
                 status: HealTaskStatus::Completed,
                 result_items,
+                result_items_truncated,
                 progress,
-            }) => ("finished".to_string(), None, result_items, progress),
+            }) => ("finished".to_string(), None, result_items, result_items_truncated, progress),
             Ok(HealTaskReport {
                 status: HealTaskStatus::Cancelled,
                 result_items,
+                result_items_truncated,
                 progress,
-            }) => ("stopped".to_string(), Some("heal task cancelled".to_string()), result_items, progress),
+            }) => (
+                "stopped".to_string(),
+                Some("heal task cancelled".to_string()),
+                result_items,
+                result_items_truncated,
+                progress,
+            ),
             Ok(HealTaskReport {
                 status: HealTaskStatus::Timeout,
                 result_items,
+                result_items_truncated,
                 progress,
-            }) => ("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,
+                result_items_truncated,
+                progress,
+            ),
             Ok(HealTaskReport {
                 status: HealTaskStatus::Failed { error },
                 result_items,
+                result_items_truncated,
                 progress,
-            }) => ("stopped".to_string(), Some(error), result_items, progress),
+            }) => ("stopped".to_string(), Some(error), result_items, result_items_truncated, progress),
             Err(crate::Error::TaskNotFound { .. }) => (
                 "notFound".to_string(),
                 Some("heal task not found or expired".to_string()),
                 Vec::new(),
+                false,
                 None,
             ),
             Err(crate::Error::InvalidClientToken) => {
@@ -393,12 +490,7 @@ impl HealChannelProcessor {
             }
         };
 
-        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 (data, detail) = encode_heal_status_response(&summary, items, progress.as_ref(), detail, truncated)?;
 
         let response = HealChannelResponse {
             request_id: client_token,
@@ -697,6 +789,22 @@ mod tests {
         // If we can get the sender, processor was created correctly
     }
 
+    #[test]
+    fn oversized_status_items_are_truncated_before_transport() {
+        let items = vec![HealResultItem {
+            detail: "x".repeat(MAX_HEAL_STATUS_PAYLOAD_SIZE + 1),
+            ..Default::default()
+        }];
+
+        let (data, detail) = encode_heal_status_response("running", items, None, None, false).unwrap();
+
+        assert!(data.len() <= MAX_HEAL_STATUS_PAYLOAD_SIZE);
+        let payload: serde_json::Value = serde_json::from_slice(&data).unwrap();
+        assert_eq!(payload["truncated"], true);
+        assert!(payload["items"].as_array().unwrap().is_empty());
+        assert_eq!(detail.as_deref(), Some("heal result items were truncated"));
+    }
+
     #[test]
     fn admission_response_preserves_all_admission_outcomes() {
         let cases = [
@@ -1339,6 +1447,70 @@ mod tests {
             .expect("receipt processor should stop cleanly");
     }
 
+    #[tokio::test]
+    async fn direct_control_execution_preserves_target_dedup_and_token_ownership() {
+        let manager = create_test_heal_manager();
+        let processor = HealChannelProcessor::new(manager);
+        let original_id = uuid::Uuid::new_v4().to_string();
+        let request = HealChannelRequest {
+            id: original_id.clone(),
+            bucket: "bucket-a".to_string(),
+            object_prefix: Some("object".to_string()),
+            priority: HealChannelPriority::High,
+            source: HealRequestSource::Admin,
+            ..Default::default()
+        };
+
+        let accepted = processor
+            .execute_start_request(request.clone())
+            .await
+            .expect("first direct start should be admitted");
+        assert_eq!(accepted.result, HealAdmissionResult::Accepted);
+        assert_eq!(accepted.task_id, original_id);
+
+        let mut duplicate = request;
+        duplicate.id = uuid::Uuid::new_v4().to_string();
+        let merged = processor
+            .execute_start_request(duplicate)
+            .await
+            .expect("duplicate direct start should merge");
+        assert_eq!(merged.result, HealAdmissionResult::Merged);
+        assert_eq!(merged.task_id, accepted.task_id);
+
+        let other = processor
+            .execute_start_request(HealChannelRequest {
+                id: uuid::Uuid::new_v4().to_string(),
+                bucket: "bucket-b".to_string(),
+                object_prefix: Some("object".to_string()),
+                priority: HealChannelPriority::High,
+                source: HealRequestSource::Admin,
+                ..Default::default()
+            })
+            .await
+            .expect("different target should be admitted independently");
+        assert_eq!(other.result, HealAdmissionResult::Accepted);
+        assert_ne!(other.task_id, accepted.task_id);
+
+        let status = processor
+            .execute_query_request(String::new(), accepted.task_id.clone())
+            .await
+            .expect("canonical token should be queryable");
+        assert!(status.success);
+        let cancelled = processor
+            .execute_cancel_request(String::new(), accepted.task_id.clone())
+            .await
+            .expect("canonical token should be cancellable");
+        assert!(cancelled.success);
+        let stopped = processor
+            .execute_query_request(String::new(), accepted.task_id)
+            .await
+            .expect("cancelled task status should remain queryable");
+        assert!(stopped.success);
+        assert_eq!(stopped.error.as_deref(), Some("heal task not found or expired"));
+        let status: serde_json::Value = serde_json::from_slice(stopped.data.as_deref().unwrap()).unwrap();
+        assert_eq!(status["summary"], "notFound");
+    }
+
     #[tokio::test]
     async fn test_process_start_request_returns_error_on_invalid_request() {
         let heal_manager = create_test_heal_manager();

crates/heal/src/heal/manager.rs

@@ -179,6 +179,7 @@ struct CompletedHealStatus {
     heal_type: HealType,
     status: HealTaskStatus,
     result_items: Vec<HealResultItem>,
+    result_items_truncated: bool,
     completed_at: SystemTime,
 }
 
@@ -198,6 +199,7 @@ struct RetryingHeal {
 pub struct HealTaskReport {
     pub status: HealTaskStatus,
     pub result_items: Vec<HealResultItem>,
+    pub result_items_truncated: bool,
     pub progress: Option<HealProgress>,
 }
 
@@ -1600,6 +1602,7 @@ impl HealManager {
                 return Ok(HealTaskReport {
                     status: task.get_status().await,
                     result_items: task.get_result_items().await,
+                    result_items_truncated: task.result_items_truncated(),
                     progress: Some(task.get_progress().await),
                 });
             }
@@ -1611,6 +1614,7 @@ impl HealManager {
                 return Ok(HealTaskReport {
                     status: retrying.status(),
                     result_items: Vec::new(),
+                    result_items_truncated: false,
                     progress: None,
                 });
             }
@@ -1625,6 +1629,7 @@ impl HealManager {
                 return Ok(HealTaskReport {
                     status: completed.status.clone(),
                     result_items: completed.result_items.clone(),
+                    result_items_truncated: completed.result_items_truncated,
                     progress: None,
                 });
             }
@@ -1636,6 +1641,7 @@ impl HealManager {
                 return Ok(HealTaskReport {
                     status: HealTaskStatus::Pending,
                     result_items: Vec::new(),
+                    result_items_truncated: false,
                     progress: None,
                 });
             }
@@ -1647,6 +1653,7 @@ impl HealManager {
             return Ok(HealTaskReport {
                 status: completed.status.clone(),
                 result_items: completed.result_items.clone(),
+                result_items_truncated: completed.result_items_truncated,
                 progress: None,
             });
         }
@@ -1666,6 +1673,7 @@ impl HealManager {
                 return Ok(HealTaskReport {
                     status: task.get_status().await,
                     result_items: task.get_result_items().await,
+                    result_items_truncated: task.result_items_truncated(),
                     progress: Some(task.get_progress().await),
                 });
             }
@@ -1679,6 +1687,7 @@ impl HealManager {
                 return Ok(HealTaskReport {
                     status: retrying.status(),
                     result_items: Vec::new(),
+                    result_items_truncated: false,
                     progress: None,
                 });
             }
@@ -1694,6 +1703,7 @@ impl HealManager {
                 return Ok(HealTaskReport {
                     status: completed.status.clone(),
                     result_items: completed.result_items.clone(),
+                    result_items_truncated: completed.result_items_truncated,
                     progress: None,
                 });
             }
@@ -1705,6 +1715,7 @@ impl HealManager {
                 return Ok(HealTaskReport {
                     status: HealTaskStatus::Pending,
                     result_items: Vec::new(),
+                    result_items_truncated: false,
                     progress: None,
                 });
             }
@@ -1719,6 +1730,7 @@ impl HealManager {
                 return Ok(HealTaskReport {
                     status: completed.status.clone(),
                     result_items: completed.result_items.clone(),
+                    result_items_truncated: completed.result_items_truncated,
                     progress: None,
                 });
             }
@@ -2618,6 +2630,7 @@ impl HealManager {
                             heal_type: completed_task.heal_type.clone(),
                             status: completed_status.clone(),
                             result_items: completed_task.get_result_items().await,
+                            result_items_truncated: completed_task.result_items_truncated(),
                             completed_at: SystemTime::now(),
                         };
                         let mut completed_heals_guard = completed_heals_clone.lock().await;
@@ -3848,6 +3861,7 @@ mod tests {
                     retry_attempt: request.retry_attempts,
                 },
                 result_items: Vec::new(),
+                result_items_truncated: false,
                 completed_at: SystemTime::now(),
             },
         );
@@ -4411,6 +4425,7 @@ mod tests {
                 },
                 status: HealTaskStatus::Completed,
                 result_items: Vec::new(),
+                result_items_truncated: false,
                 completed_at: SystemTime::now(),
             },
         );
@@ -4444,6 +4459,7 @@ mod tests {
                     object_size: 1024,
                     ..Default::default()
                 }],
+                result_items_truncated: true,
                 completed_at: SystemTime::now(),
             },
         );
@@ -4452,6 +4468,7 @@ mod tests {
             .get_task_report_for_path("bucket/object", "completed-token")
             .await
             .expect("recent completed task report should be queryable");
+        assert!(report.result_items_truncated);
 
         assert_eq!(report.status, HealTaskStatus::Completed);
         assert_eq!(report.result_items.len(), 1);

crates/heal/src/heal/task.rs

@@ -43,6 +43,7 @@ const LOG_SUBSYSTEM_OBJECT: &str = "object";
 const EVENT_HEAL_TASK_STATE: &str = "heal_task_state";
 const EVENT_HEAL_OBJECT_STAGE: &str = "heal_object_stage";
 const EVENT_HEAL_OBJECT_MISSING: &str = "heal_object_missing";
+const MAX_RETAINED_HEAL_RESULT_ITEMS: usize = 1024;
 const EVENT_HEAL_OBJECT_RESULT: &str = "heal_object_result";
 const MAX_BUCKET_OBJECT_HEAL_RETRIES: u32 = 3;
 const MAX_BUCKET_FAILURE_LOG_SAMPLES: u64 = 5;
@@ -306,6 +307,7 @@ pub struct HealTask {
     pub progress: Arc<RwLock<HealProgress>>,
     /// Result items collected from storage heal calls.
     pub result_items: Arc<RwLock<Vec<HealResultItem>>>,
+    result_items_truncated: Arc<AtomicBool>,
     batch_failure: Arc<RwLock<Option<BatchHealFailure>>>,
     batch_failure_recorded: Arc<AtomicBool>,
     /// Created time
@@ -337,6 +339,7 @@ impl HealTask {
             status: Arc::new(RwLock::new(HealTaskStatus::Pending)),
             progress: Arc::new(RwLock::new(HealProgress::new())),
             result_items: Arc::new(RwLock::new(Vec::new())),
+            result_items_truncated: Arc::new(AtomicBool::new(false)),
             batch_failure: Arc::new(RwLock::new(None)),
             batch_failure_recorded: Arc::new(AtomicBool::new(false)),
             created_at: request.created_at,
@@ -741,8 +744,17 @@ impl HealTask {
         self.result_items.read().await.clone()
     }
 
+    pub fn result_items_truncated(&self) -> bool {
+        self.result_items_truncated.load(Ordering::Relaxed)
+    }
+
     async fn record_result_item(&self, result: HealResultItem) {
-        self.result_items.write().await.push(result);
+        let mut result_items = self.result_items.write().await;
+        if result_items.len() < MAX_RETAINED_HEAL_RESULT_ITEMS {
+            result_items.push(result);
+        } else {
+            self.result_items_truncated.store(true, Ordering::Relaxed);
+        }
     }
 
     // specific heal implementation method
@@ -2656,6 +2668,19 @@ mod tests {
         assert_eq!(result_items.iter().filter(|item| item.object_size == 1).count(), 2);
     }
 
+    #[tokio::test]
+    async fn result_items_are_bounded_and_report_truncation() {
+        let storage = Arc::new(MockStorage::default());
+        let task = HealTask::from_request(HealRequest::bucket("bucket-a".to_string()), storage);
+
+        for _ in 0..=MAX_RETAINED_HEAL_RESULT_ITEMS {
+            task.record_result_item(HealResultItem::default()).await;
+        }
+
+        assert_eq!(task.get_result_items().await.len(), MAX_RETAINED_HEAL_RESULT_ITEMS);
+        assert!(task.result_items_truncated());
+    }
+
     #[tokio::test]
     async fn test_recursive_bucket_heal_skips_object_dir_candidates() {
         let storage = Arc::new(MockStorage {