feat(heal): incremental status cursors and typed overlap policy (HS-06) (#6206)

* feat(heal): incremental heal status cursors and typed overlap policy (HS-06)

Incremental results: every retained result item now carries a monotonic
sequence number. The status query accepts a client cursor (sinceSeq on
the admin wire, Option<u64> internally) and returns only newer items,
plus nextSeq (the next cursor) and minSeq (the oldest retained
sequence). A cursor that fell behind the 1024-item retention window is
flagged through the existing truncated signal together with minSeq so
the client can restart from it. Sequencing survives task completion:
the completion archive stores the seq-stamped window. None keeps the
exact legacy full-snapshot behavior, so existing clients see no change.

Typed overlap handling for admin starts: RUSTFS_HEAL_OVERLAP_POLICY
(merge default | minio_error). Under minio_error, an admin start whose
path overlaps an active or queued task rejects with typed
already-running / overlapping-paths admission reasons (surfaced through
reason_label in the admin error body, sharing the existing
OperationAborted site because the s3s footprint ratchet forbids new
s3_error! sites); an exact duplicate start rejects with
already-running instead of silently merging. Scanner/autoheal/
read-repair sources never take the rejection path.

forceStart semantics now match MinIO for admin requests: an admin
forceStart first cancels the overlapping active admin task, then
admits the replacement.

Wire: the heal-control Query command grows an optional sinceSeq
(defaulted and skipped when absent, so older peers stay compatible);
the admin handler accepts the sinceSeq query parameter; the local
channel query gains the same cursor.

Tests: seq monotonicity and incremental slicing, window slide moving
minSeq with lagging-cursor flags, overlap matrix (same/containing/
contained/disjoint x policy x source), forceStart cancel-then-admit,
and the completion-archive window handoff.

Co-Authored-By: heihutu <heihutu@gmail.com>

* style: fmt after main merge

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

Cargo.lock

@@ -9539,6 +9539,8 @@ version = "1.0.0-rc.2"
 dependencies = [
  "async-trait",
  "base64 0.23.1",
+ "bytes",
+ "crc-fast",
  "futures",
  "hotpath",
  "http 1.5.0",

crates/common/src/heal_channel.rs

@@ -224,6 +224,13 @@ pub struct HealOpts {
 pub enum HealAdmissionDropReason {
     QueueFull,
     PolicyDropped,
+    /// HS-06: an admin heal start overlaps (same bucket with mutually
+    /// containing prefixes, or the same erasure set) an already running or
+    /// queued task. Only produced when RUSTFS_HEAL_OVERLAP_POLICY=minio_error.
+    AlreadyRunning,
+    /// HS-06: same as [`Self::AlreadyRunning`] but for paths that merely
+    /// contain (or are contained by) the active task's path.
+    OverlappingPaths,
 }
 
 impl HealAdmissionDropReason {
@@ -231,6 +238,8 @@ impl HealAdmissionDropReason {
         match self {
             Self::QueueFull => "queue_full",
             Self::PolicyDropped => "policy_dropped",
+            Self::AlreadyRunning => "already_running",
+            Self::OverlappingPaths => "overlapping_paths",
         }
     }
 }
@@ -317,6 +326,9 @@ pub enum HealChannelCommand {
     Query {
         heal_path: String,
         client_token: String,
+        /// Incremental result cursor (HS-06): only items with a sequence
+        /// greater than this are returned; `None` keeps the full snapshot.
+        since_seq: Option<u64>,
         response_tx: oneshot::Sender<Result<HealChannelResponse, String>>,
     },
     /// Cancel heal task
@@ -522,10 +534,21 @@ async fn receive_heal_channel_response(
 
 /// Send heal query request
 pub async fn query_heal_status(heal_path: String, client_token: String) -> Result<HealChannelResponse, String> {
+    query_heal_status_since(heal_path, client_token, None).await
+}
+
+/// Incremental heal query (HS-06): pass the client's last seen sequence
+/// number to receive only newer result items.
+pub async fn query_heal_status_since(
+    heal_path: String,
+    client_token: String,
+    since_seq: Option<u64>,
+) -> Result<HealChannelResponse, String> {
     let (response_tx, response_rx) = oneshot::channel();
     send_heal_command(HealChannelCommand::Query {
         heal_path,
         client_token,
+        since_seq,
         response_tx,
     })
     .await?;

crates/config/src/constants/heal.rs

@@ -205,3 +205,12 @@ pub const DEFAULT_HEAL_MRF_JOURNAL_MAX_BYTES: usize = 8 * 1024 * 1024;
 
 /// Default MRF replay batch size.
 pub const DEFAULT_HEAL_MRF_REPLAY_BATCH: usize = 256;
+
+/// Environment variable selecting how admin heal starts behave when the
+/// requested path overlaps an already running or queued heal: `merge`
+/// (default, keep today's dedup/merge semantics) or `minio_error` (return a
+/// typed already-running / overlapping-paths rejection like madmin).
+pub const ENV_HEAL_OVERLAP_POLICY: &str = "RUSTFS_HEAL_OVERLAP_POLICY";
+
+/// Default overlap policy: merge duplicate/overlapping requests.
+pub const DEFAULT_HEAL_OVERLAP_POLICY: &str = "merge";

crates/heal/src/heal/channel.rs

@@ -66,21 +66,37 @@ struct HealTaskStatusPayload<'a> {
     summary: &'a str,
     items: &'a [HealResultItem],
     truncated: bool,
+    /// Cursor for incremental consumption (HS-06): sequence of the next item
+    /// to be produced. Absent on responses without sequencing (0).
+    #[serde(skip_serializing_if = "u64_is_zero")]
+    next_seq: u64,
+    /// Oldest sequence still retained; with `truncated`, tells a lagging
+    /// client where to restart its cursor.
+    #[serde(skip_serializing_if = "u64_is_zero")]
+    min_seq: u64,
     #[serde(skip_serializing_if = "Option::is_none")]
     progress: Option<&'a HealProgress>,
 }
 
+fn u64_is_zero(value: &u64) -> bool {
+    *value == 0
+}
+
 fn encode_heal_task_status_payload(
     summary: &str,
     mut items: Vec<HealResultItem>,
     progress: Option<&HealProgress>,
     mut truncated: bool,
+    next_seq: u64,
+    min_seq: u64,
 ) -> Result<(Vec<u8>, bool)> {
     loop {
         let data = serde_json::to_vec(&HealTaskStatusPayload {
             summary,
             items: &items,
             truncated,
+            next_seq,
+            min_seq,
             progress,
         })
         .map_err(|e| Error::Serialization(format!("failed to serialize heal task status: {e}")))?;
@@ -109,8 +125,10 @@ fn encode_heal_status_response(
     progress: Option<&HealProgress>,
     detail: Option<String>,
     truncated: bool,
+    next_seq: u64,
+    min_seq: u64,
 ) -> Result<(Vec<u8>, Option<String>)> {
-    let (data, truncated) = encode_heal_task_status_payload(summary, items, progress, truncated)?;
+    let (data, truncated) = encode_heal_task_status_payload(summary, items, progress, truncated, next_seq, min_seq)?;
     Ok((data, heal_status_detail(detail, truncated)))
 }
 
@@ -138,8 +156,19 @@ impl HealChannelProcessor {
 
     /// Execute a token query directly against the manager.
     pub async fn execute_query_request(&self, heal_path: String, client_token: String) -> Result<HealChannelResponse> {
+        self.execute_query_request_since(heal_path, client_token, None).await
+    }
+
+    /// Incremental variant of [`Self::execute_query_request`] (HS-06).
+    pub async fn execute_query_request_since(
+        &self,
+        heal_path: String,
+        client_token: String,
+        since_seq: Option<u64>,
+    ) -> Result<HealChannelResponse> {
         let (response_tx, response_rx) = oneshot::channel();
-        self.process_query_request(heal_path, client_token, response_tx).await?;
+        self.process_query_request(heal_path, client_token, since_seq, response_tx)
+            .await?;
         response_rx
             .await
             .map_err(|err| Error::other(format!("heal query channel closed: {err}")))?
@@ -262,8 +291,12 @@ impl HealChannelProcessor {
             HealChannelCommand::Query {
                 heal_path,
                 client_token,
+                since_seq,
                 response_tx,
-            } => self.process_query_request(heal_path, client_token, response_tx).await,
+            } => {
+                self.process_query_request(heal_path, client_token, since_seq, response_tx)
+                    .await
+            }
             HealChannelCommand::Cancel {
                 heal_path,
                 client_token,
@@ -384,6 +417,7 @@ impl HealChannelProcessor {
         &self,
         heal_path: String,
         client_token: String,
+        since_seq: Option<u64>,
         response_tx: oneshot::Sender<std::result::Result<HealChannelResponse, String>>,
     ) -> Result<()> {
         debug!(
@@ -398,72 +432,118 @@ impl HealChannelProcessor {
         );
 
         let report = if heal_path.trim_matches('/').is_empty() {
-            self.heal_manager.get_task_report(&client_token).await
+            self.heal_manager.get_task_report_since(&client_token, since_seq).await
         } else {
-            self.heal_manager.get_task_report_for_path(&heal_path, &client_token).await
+            self.heal_manager
+                .get_task_report_for_path_since(&heal_path, &client_token, since_seq)
+                .await
         };
 
-        let (summary, detail, items, truncated, progress) = match report {
+        let (summary, detail, items, truncated, progress, next_seq, min_seq) = match report {
             Ok(HealTaskReport {
                 status: HealTaskStatus::Pending | HealTaskStatus::Running,
                 result_items,
                 result_items_truncated,
                 progress,
-            }) => ("running".to_string(), None, result_items, result_items_truncated, progress),
+                next_seq,
+                min_seq,
+            }) => (
+                "running".to_string(),
+                None,
+                result_items,
+                result_items_truncated,
+                progress,
+                next_seq,
+                min_seq,
+            ),
             Ok(HealTaskReport {
                 status: HealTaskStatus::Retrying { error, retry_attempt },
                 result_items,
                 result_items_truncated,
                 progress,
+                next_seq,
+                min_seq,
             }) => (
                 "running".to_string(),
                 Some(format!("heal task retrying after recoverable failure, attempt {retry_attempt}: {error}")),
                 result_items,
                 result_items_truncated,
                 progress,
+                next_seq,
+                min_seq,
             ),
             Ok(HealTaskReport {
                 status: HealTaskStatus::Completed,
                 result_items,
                 result_items_truncated,
                 progress,
-            }) => ("finished".to_string(), None, result_items, result_items_truncated, progress),
+                next_seq,
+                min_seq,
+            }) => (
+                "finished".to_string(),
+                None,
+                result_items,
+                result_items_truncated,
+                progress,
+                next_seq,
+                min_seq,
+            ),
             Ok(HealTaskReport {
                 status: HealTaskStatus::Cancelled,
                 result_items,
                 result_items_truncated,
                 progress,
+                next_seq,
+                min_seq,
             }) => (
                 "stopped".to_string(),
                 Some("heal task cancelled".to_string()),
                 result_items,
                 result_items_truncated,
                 progress,
+                next_seq,
+                min_seq,
             ),
             Ok(HealTaskReport {
                 status: HealTaskStatus::Timeout,
                 result_items,
                 result_items_truncated,
                 progress,
+                next_seq,
+                min_seq,
             }) => (
                 "stopped".to_string(),
                 Some("heal task timed out".to_string()),
                 result_items,
                 result_items_truncated,
                 progress,
+                next_seq,
+                min_seq,
             ),
             Ok(HealTaskReport {
                 status: HealTaskStatus::Failed { error },
                 result_items,
                 result_items_truncated,
                 progress,
-            }) => ("stopped".to_string(), Some(error), result_items, result_items_truncated, progress),
+                next_seq,
+                min_seq,
+            }) => (
+                "stopped".to_string(),
+                Some(error),
+                result_items,
+                result_items_truncated,
+                progress,
+                next_seq,
+                min_seq,
+            ),
             Err(crate::Error::TaskNotFound { .. }) => (
                 "notFound".to_string(),
                 Some("heal task not found or expired".to_string()),
                 Vec::new(),
                 false,
                 None,
+                0,
+                0,
             ),
             Err(crate::Error::InvalidClientToken) => {
                 let response = HealChannelResponse {
@@ -490,7 +570,8 @@ impl HealChannelProcessor {
             }
         };
 
-        let (data, detail) = encode_heal_status_response(&summary, items, progress.as_ref(), detail, truncated)?;
+        let (data, detail) =
+            encode_heal_status_response(&summary, items, progress.as_ref(), detail, truncated, next_seq, min_seq)?;
 
         let response = HealChannelResponse {
             request_id: client_token,
@@ -805,7 +886,7 @@ mod tests {
             ..Default::default()
         }];
 
-        let (data, detail) = encode_heal_status_response("running", items, None, None, false).unwrap();
+        let (data, detail) = encode_heal_status_response("running", items, None, None, false, 0, 0).unwrap();
 
         assert!(data.len() <= MAX_HEAL_STATUS_PAYLOAD_SIZE);
         let payload: serde_json::Value = serde_json::from_slice(&data).unwrap();
@@ -1575,7 +1656,7 @@ mod tests {
         let (tx, rx) = oneshot::channel();
 
         processor
-            .process_query_request("bucket".to_string(), "completed-token".to_string(), tx)
+            .process_query_request("bucket".to_string(), "completed-token".to_string(), None, tx)
             .await
             .expect("query should process");
 
@@ -1610,7 +1691,7 @@ mod tests {
         let (tx, rx) = oneshot::channel();
 
         processor
-            .process_query_request("bucket".to_string(), task_id.clone(), tx)
+            .process_query_request("bucket".to_string(), task_id.clone(), None, tx)
             .await
             .expect("query should process");
 
@@ -1643,7 +1724,7 @@ mod tests {
         let (tx, rx) = oneshot::channel();
 
         processor
-            .process_query_request("bucket".to_string(), "wrong-token".to_string(), tx)
+            .process_query_request("bucket".to_string(), "wrong-token".to_string(), None, tx)
             .await
             .expect("query should process");
 
@@ -1668,7 +1749,7 @@ mod tests {
         let (tx, rx) = oneshot::channel();
 
         processor
-            .process_query_request(String::new(), "wrong-token".to_string(), tx)
+            .process_query_request(String::new(), "wrong-token".to_string(), None, tx)
             .await
             .expect("query should process");
 
@@ -1705,7 +1786,7 @@ mod tests {
         let (tx, rx) = oneshot::channel();
 
         processor
-            .process_query_request(String::new(), task_id.clone(), tx)
+            .process_query_request(String::new(), task_id.clone(), None, tx)
             .await
             .expect("query should process");
 

crates/heal/src/heal/manager.rs

@@ -220,6 +220,11 @@ struct CompletedHealStatus {
     result_items: Vec<HealResultItem>,
     result_items_truncated: bool,
     completed_at: SystemTime,
+    /// Sequence-stamped retained window, archived with the completion so
+    /// incremental consumers keep their cursor across the transition (HS-06).
+    seqed_items: Vec<(u64, HealResultItem)>,
+    next_seq: u64,
+    min_seq: u64,
 }
 
 #[derive(Debug, Clone)]
@@ -240,6 +245,65 @@ pub struct HealTaskReport {
     pub result_items: Vec<HealResultItem>,
     pub result_items_truncated: bool,
     pub progress: Option<HealProgress>,
+    /// Cursor for incremental consumption: sequence number of the next item
+    /// to be produced. `0` on reports from sources without sequencing.
+    pub next_seq: u64,
+    /// Oldest sequence still retained (`0` together with `next_seq` when
+    /// sequencing is unavailable).
+    pub min_seq: u64,
+}
+
+/// Report from a live task, honoring the client's incremental cursor.
+async fn active_task_report(task: &HealTask, since: Option<u64>) -> HealTaskReport {
+    let window = task.get_result_items_since(since).await;
+    HealTaskReport {
+        status: task.get_status().await,
+        result_items: window.items,
+        // The legacy flag stays set once anything was evicted; a lagging
+        // incremental cursor additionally marks this response truncated so
+        // the client knows to restart from `min_seq`.
+        result_items_truncated: task.result_items_truncated() || window.lagged,
+        progress: Some(task.get_progress().await),
+        next_seq: window.next_seq,
+        min_seq: window.min_seq,
+    }
+}
+
+fn empty_task_report(status: HealTaskStatus) -> HealTaskReport {
+    HealTaskReport {
+        status,
+        result_items: Vec::new(),
+        result_items_truncated: false,
+        progress: None,
+        next_seq: 0,
+        min_seq: 0,
+    }
+}
+
+fn completed_task_report(completed: &CompletedHealStatus, since: Option<u64>) -> HealTaskReport {
+    let mut lagged = false;
+    let result_items = match since {
+        None => completed.result_items.clone(),
+        Some(cursor) => {
+            if cursor + 1 < completed.min_seq {
+                lagged = true;
+            }
+            completed
+                .seqed_items
+                .iter()
+                .filter(|(seq, _)| *seq > cursor)
+                .map(|(_, item)| item.clone())
+                .collect()
+        }
+    };
+    HealTaskReport {
+        status: completed.status.clone(),
+        result_items,
+        result_items_truncated: completed.result_items_truncated || lagged,
+        progress: None,
+        next_seq: completed.next_seq,
+        min_seq: completed.min_seq,
+    }
 }
 
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
@@ -531,6 +595,11 @@ impl PriorityHealQueue {
         self.dedup_keys.contains_key(&key)
     }
 
+    /// Iterate queued requests (used by the admin overlap check).
+    fn requests(&self) -> impl Iterator<Item = &HealRequest> {
+        self.heap.iter().map(|item| &item.request)
+    }
+
     fn contains_request_id(&self, request_id: &str) -> bool {
         self.heap.iter().any(|item| item.request.id == request_id)
     }
@@ -689,6 +758,80 @@ fn recoverable_heal_retry_delay(retry_attempt: u32) -> Duration {
 }
 
 /// Heal config
+/// HS-06 admin overlap policy.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum HealOverlapPolicy {
+    /// Default: overlapping admin starts merge into the existing task
+    /// (today's dedup semantics).
+    #[default]
+    Merge,
+    /// Return a typed already-running / overlapping-paths rejection like
+    /// madmin's ErrHealAlreadyRunning / ErrHealOverlappingPaths.
+    MinioError,
+}
+
+/// Path view of a heal type for overlap comparison: a bucket plus a
+/// prefix/object path inside it (`None` bucket = cluster-wide, overlaps
+/// everything).
+fn heal_type_path_view(heal_type: &HealType) -> (Option<&str>, &str) {
+    match heal_type {
+        HealType::Cluster => (None, ""),
+        HealType::Bucket { bucket } => (Some(bucket), ""),
+        HealType::Prefix { bucket, prefix } => (Some(bucket), prefix),
+        HealType::Object { bucket, object, .. }
+        | HealType::Metadata { bucket, object }
+        | HealType::ECDecode { bucket, object, .. } => (Some(bucket), object),
+        // MRF/MetaPath heal keys on a meta path; treat the whole set of
+        // buckets as one namespace so it only overlaps itself exactly.
+        HealType::MRF { meta_path } => (Some("\u{0}mrf"), meta_path),
+        // Erasure-set heal: the set id is the overlap dimension.
+        HealType::ErasureSet { set_disk_id, .. } => (Some("\u{0}set"), set_disk_id),
+    }
+}
+
+/// How two heal paths relate for the admin overlap check (HS-06).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum OverlapVerdict {
+    /// Distinct targets: no conflict.
+    Disjoint,
+    /// Same target: an identical heal is already in flight.
+    SameTarget,
+    /// One target contains the other.
+    Overlapping,
+}
+
+fn prefix_paths_overlap(a: &str, b: &str) -> OverlapVerdict {
+    if a == b {
+        return OverlapVerdict::SameTarget;
+    }
+    if a.is_empty() || b.is_empty() || a.starts_with(b) || b.starts_with(a) {
+        return OverlapVerdict::Overlapping;
+    }
+    OverlapVerdict::Disjoint
+}
+
+fn heal_types_overlap(left: &HealType, right: &HealType) -> OverlapVerdict {
+    let (left_bucket, left_path) = heal_type_path_view(left);
+    let (right_bucket, right_path) = heal_type_path_view(right);
+    match (left_bucket, right_bucket) {
+        // Cluster-wide overlaps everything (but an exact cluster match is
+        // SameTarget).
+        (None, _) | (_, None) => {
+            if matches!(left, HealType::Cluster) && matches!(right, HealType::Cluster) {
+                OverlapVerdict::SameTarget
+            } else {
+                OverlapVerdict::Overlapping
+            }
+        }
+        (Some(lb), Some(rb)) => {
+            if lb != rb {
+                return OverlapVerdict::Disjoint;
+            }
+            prefix_paths_overlap(left_path, right_path)
+        }
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct HealConfig {
     /// Whether to enable auto heal
@@ -709,6 +852,9 @@ pub struct HealConfig {
     pub low_priority_drop_when_full: bool,
     /// Whether notify-driven scheduler wakeups are enabled.
     pub event_driven_scheduler_enable: bool,
+    /// How admin heal starts behave on path overlap (HS-06): merge into the
+    /// existing task (default) or return a typed already-running rejection.
+    pub overlap_policy: HealOverlapPolicy,
     /// Whether per-set bulkhead scheduling is enabled.
     pub set_bulkhead_enable: bool,
     /// Whether erasure-set page parallelism is enabled.
@@ -757,6 +903,14 @@ impl Default for HealConfig {
             rustfs_config::ENV_HEAL_EVENT_DRIVEN_SCHEDULER_ENABLE,
             rustfs_config::DEFAULT_HEAL_EVENT_DRIVEN_SCHEDULER_ENABLE,
         );
+        let overlap_policy =
+            match rustfs_utils::get_env_str(rustfs_config::ENV_HEAL_OVERLAP_POLICY, rustfs_config::DEFAULT_HEAL_OVERLAP_POLICY)
+                .to_lowercase()
+                .as_str()
+            {
+                "minio_error" => HealOverlapPolicy::MinioError,
+                _ => HealOverlapPolicy::Merge,
+            };
         let set_bulkhead_enable = rustfs_utils::get_env_bool(
             rustfs_config::ENV_HEAL_SET_BULKHEAD_ENABLE,
             rustfs_config::DEFAULT_HEAL_SET_BULKHEAD_ENABLE,
@@ -793,6 +947,7 @@ impl Default for HealConfig {
             low_priority_merge_enable,
             low_priority_drop_when_full,
             event_driven_scheduler_enable,
+            overlap_policy,
             set_bulkhead_enable,
             page_parallel_enable,
             mainline_throttle_enable,
@@ -1759,6 +1914,50 @@ impl HealManager {
         request: HealRequest,
         preserve_alias: bool,
     ) -> Result<HealAdmissionReceipt> {
+        // HS-06 forceStart semantics (admin only): MinIO stops the old task
+        // first and then starts the new one. Cancel any active admin task
+        // overlapping this request's path before entering admission, so the
+        // fresh task is never merged into the one being replaced.
+        if request.source == HealRequestSource::Admin && request.force_start {
+            let overlapping: Vec<String> = {
+                let active_heals = self.active_heals.lock().await;
+                active_heals
+                    .iter()
+                    .filter(|(task_id, task)| {
+                        task.source == HealRequestSource::Admin
+                            && heal_types_overlap(&request.heal_type, &task.heal_type) != OverlapVerdict::Disjoint
+                            && *task_id != &request.id
+                    })
+                    .map(|(task_id, _)| task_id.clone())
+                    .collect()
+            };
+            for task_id in overlapping {
+                match self.cancel_task(&task_id).await {
+                    Ok(_) => info!(
+                        target: "rustfs::heal::manager",
+                        event = EVENT_HEAL_QUEUE_ADMISSION,
+                        component = LOG_COMPONENT_HEAL,
+                        subsystem = LOG_SUBSYSTEM_MANAGER,
+                        request_id = %request.id,
+                        cancelled_task_id = %task_id,
+                        result = "force_start_cancelled_overlap",
+                        "Admin forceStart cancelled an overlapping heal task"
+                    ),
+                    Err(err) => warn!(
+                        target: "rustfs::heal::manager",
+                        event = EVENT_HEAL_QUEUE_ADMISSION,
+                        component = LOG_COMPONENT_HEAL,
+                        subsystem = LOG_SUBSYSTEM_MANAGER,
+                        request_id = %request.id,
+                        cancelled_task_id = %task_id,
+                        error = %err,
+                        result = "force_start_cancel_failed",
+                        "Admin forceStart failed to cancel an overlapping heal task"
+                    ),
+                }
+            }
+        }
+
         let config = self.config.read().await;
         let dedup_key = PriorityHealQueue::make_dedup_key(&request);
 
@@ -1781,7 +1980,15 @@ impl HealManager {
                 .or_else(|| retrying_heal_for_dedup_key(&retrying_heals, &dedup_key).map(|(task_id, _)| (task_id, "retrying")))
         });
         if let Some((merged_task_id, duplicate_state)) = duplicate.flatten() {
-            let admission = Self::duplicate_admission_for_request(&request, &config);
+            // HS-06: under the minio_error overlap policy an exact duplicate
+            // admin start reports the typed AlreadyRunning rejection instead
+            // of the silent merge (MinIO's ErrHealAlreadyRunning).
+            let admission =
+                if request.source == HealRequestSource::Admin && config.overlap_policy == HealOverlapPolicy::MinioError {
+                    HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
+                } else {
+                    Self::duplicate_admission_for_request(&request, &config)
+                };
             drop(retrying_heals);
             drop(queue);
             drop(active_heals);
@@ -1827,6 +2034,62 @@ impl HealManager {
             });
         }
 
+        // HS-06 typed overlap rejection (admin only, minio_error policy):
+        // paths containing or contained by an active/queued task reject with
+        // AlreadyRunning / OverlappingPaths instead of merging. Exact
+        // duplicates already merged above; scanner/autoheal/read-repair
+        // sources never take this path.
+        if request.source == HealRequestSource::Admin && config.overlap_policy == HealOverlapPolicy::MinioError {
+            let mut rejection = None;
+            for (task_id, task) in active_heals.iter() {
+                match heal_types_overlap(&request.heal_type, &task.heal_type) {
+                    OverlapVerdict::SameTarget => {
+                        rejection = Some((HealAdmissionDropReason::AlreadyRunning, task_id.clone()));
+                        break;
+                    }
+                    OverlapVerdict::Overlapping => {
+                        rejection = Some((HealAdmissionDropReason::OverlappingPaths, task_id.clone()));
+                    }
+                    OverlapVerdict::Disjoint => {}
+                }
+            }
+            if rejection.is_none() {
+                for queued in queue.requests() {
+                    match heal_types_overlap(&request.heal_type, &queued.heal_type) {
+                        OverlapVerdict::SameTarget => {
+                            rejection = Some((HealAdmissionDropReason::AlreadyRunning, queued.id.clone()));
+                            break;
+                        }
+                        OverlapVerdict::Overlapping => {
+                            rejection = Some((HealAdmissionDropReason::OverlappingPaths, queued.id.clone()));
+                        }
+                        OverlapVerdict::Disjoint => {}
+                    }
+                }
+            }
+            if let Some((reason, overlap_task_id)) = rejection {
+                drop(retrying_heals);
+                drop(queue);
+                drop(active_heals);
+                Self::record_admission_metric(request.source, HealAdmissionResult::Dropped(reason), "overlap_rejected");
+                warn!(
+                    target: "rustfs::heal::manager",
+                    event = EVENT_HEAL_QUEUE_ADMISSION,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_MANAGER,
+                    request_id = %request.id,
+                    overlap_task_id = %overlap_task_id,
+                    reason = reason.as_str(),
+                    result = "overlap_rejected",
+                    "Admin heal start rejected by overlap policy"
+                );
+                return Ok(HealAdmissionReceipt {
+                    result: HealAdmissionResult::Dropped(reason),
+                    task_id: overlap_task_id,
+                });
+            }
+        }
+
         let mut task_id = request.id.clone();
         let admission = Self::admit_request_to_queue(&mut queue, request, &config, "submit");
         if admission == HealAdmissionResult::Merged
@@ -1899,28 +2162,25 @@ impl HealManager {
     }
 
     pub async fn get_task_report(&self, task_id: &str) -> Result<HealTaskReport> {
+        self.get_task_report_since(task_id, None).await
+    }
+
+    /// Incremental variant of [`Self::get_task_report`] (HS-06): `since` is
+    /// the client's last seen sequence number; `None` keeps the legacy
+    /// full-snapshot semantics.
+    pub async fn get_task_report_since(&self, task_id: &str, since: Option<u64>) -> Result<HealTaskReport> {
         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 {
+                return Ok(active_task_report(task, since).await);
-                    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),
-                });
             }
         }
 
         {
             let retrying_heals = self.retrying_heals.lock().await;
             if let Some(retrying) = retrying_heals.get(&canonical_task_id) {
-                return Ok(HealTaskReport {
+                return Ok(empty_task_report(retrying.status()));
-                    status: retrying.status(),
-                    result_items: Vec::new(),
-                    result_items_truncated: false,
-                    progress: None,
-                });
             }
         }
 
@@ -1930,36 +2190,21 @@ impl HealManager {
             if let Some(completed) = completed_heals.get(&canonical_task_id)
                 && completed_status_is_retrying(&completed.status)
             {
-                return Ok(HealTaskReport {
+                return Ok(completed_task_report(completed, since));
-                    status: completed.status.clone(),
-                    result_items: completed.result_items.clone(),
-                    result_items_truncated: completed.result_items_truncated,
-                    progress: None,
-                });
             }
         }
 
         {
             let queue = self.heal_queue.lock().await;
             if queue.contains_request_id(&canonical_task_id) {
-                return Ok(HealTaskReport {
+                return Ok(empty_task_report(HealTaskStatus::Pending));
-                    status: HealTaskStatus::Pending,
-                    result_items: Vec::new(),
-                    result_items_truncated: false,
-                    progress: None,
-                });
             }
         }
 
         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 {
+            return Ok(completed_task_report(completed, since));
-                status: completed.status.clone(),
-                result_items: completed.result_items.clone(),
-                result_items_truncated: completed.result_items_truncated,
-                progress: None,
-            });
         }
 
         Err(Error::TaskNotFound {
@@ -1968,18 +2213,23 @@ impl HealManager {
     }
 
     pub async fn get_task_report_for_path(&self, heal_path: &str, task_id: &str) -> Result<HealTaskReport> {
+        self.get_task_report_for_path_since(heal_path, task_id, None).await
+    }
+
+    /// Incremental variant of [`Self::get_task_report_for_path`] (HS-06).
+    pub async fn get_task_report_for_path_since(
+        &self,
+        heal_path: &str,
+        task_id: &str,
+        since: Option<u64>,
+    ) -> Result<HealTaskReport> {
         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)
                 && heal_type_matches_path(&task.heal_type, heal_path)
             {
-                return Ok(HealTaskReport {
+                return Ok(active_task_report(task, since).await);
-                    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),
-                });
             }
         }
 
@@ -1988,12 +2238,7 @@ impl HealManager {
             if let Some(retrying) = retrying_heals.get(&canonical_task_id)
                 && heal_type_matches_path(&retrying.request.heal_type, heal_path)
             {
-                return Ok(HealTaskReport {
+                return Ok(empty_task_report(retrying.status()));
-                    status: retrying.status(),
-                    result_items: Vec::new(),
-                    result_items_truncated: false,
-                    progress: None,
-                });
             }
         }
 
@@ -2004,24 +2249,14 @@ impl HealManager {
                 && heal_type_matches_path(&completed.heal_type, heal_path)
                 && completed_status_is_retrying(&completed.status)
             {
-                return Ok(HealTaskReport {
+                return Ok(completed_task_report(completed, since));
-                    status: completed.status.clone(),
-                    result_items: completed.result_items.clone(),
-                    result_items_truncated: completed.result_items_truncated,
-                    progress: None,
-                });
             }
         }
 
         {
             let queue = self.heal_queue.lock().await;
             if queue.contains_request_id_matching_path(&canonical_task_id, heal_path) {
-                return Ok(HealTaskReport {
+                return Ok(empty_task_report(HealTaskStatus::Pending));
-                    status: HealTaskStatus::Pending,
-                    result_items: Vec::new(),
-                    result_items_truncated: false,
-                    progress: None,
-                });
             }
         }
 
@@ -2031,12 +2266,7 @@ impl HealManager {
             if let Some(completed) = completed_heals.get(&canonical_task_id)
                 && heal_type_matches_path(&completed.heal_type, heal_path)
             {
-                return Ok(HealTaskReport {
+                return Ok(completed_task_report(completed, since));
-                    status: completed.status.clone(),
-                    result_items: completed.result_items.clone(),
-                    result_items_truncated: completed.result_items_truncated,
-                    progress: None,
-                });
             }
         }
 
@@ -3231,12 +3461,16 @@ impl HealManager {
                             completed_task.get_status().await
                         };
                         let completed_progress = completed_task.get_progress().await;
+                        let final_window = completed_task.get_result_items_since(None).await;
                         let completed_status_entry = CompletedHealStatus {
                             heal_type: completed_task.heal_type.clone(),
                             status: completed_status.clone(),
-                            result_items: completed_task.get_result_items().await,
+                            result_items: final_window.items.clone(),
                             result_items_truncated: completed_task.result_items_truncated(),
                             completed_at: SystemTime::now(),
+                            seqed_items: completed_task.get_seqed_result_items().await,
+                            next_seq: final_window.next_seq,
+                            min_seq: final_window.min_seq,
                         };
                         let mut completed_heals_guard = completed_heals_clone.lock().await;
                         prune_completed_heal_statuses(&mut completed_heals_guard);
@@ -5008,6 +5242,9 @@ mod tests {
                 },
                 result_items: Vec::new(),
                 result_items_truncated: false,
+                seqed_items: Vec::new(),
+                next_seq: 0,
+                min_seq: 0,
                 completed_at: SystemTime::now(),
             },
         );
@@ -5289,6 +5526,136 @@ mod tests {
         assert_eq!(snapshot.queued_by_source.internal, 0);
     }
 
+    // HS-06 (backlog#1870): overlap policy + forceStart semantics.
+    fn manager_with_policy(policy: HealOverlapPolicy) -> HealManager {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        HealManager::new(
+            storage,
+            Some(HealConfig {
+                overlap_policy: policy,
+                ..Default::default()
+            }),
+        )
+    }
+
+    fn admin_prefix_request(bucket: &str, prefix: &str) -> HealRequest {
+        let mut request = HealRequest::new(
+            HealType::Prefix {
+                bucket: bucket.to_string(),
+                prefix: prefix.to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Normal,
+        );
+        request.source = HealRequestSource::Admin;
+        request
+    }
+
+    async fn insert_active_task(manager: &HealManager, request: HealRequest) -> String {
+        let task = Arc::new(HealTask::from_request(request, manager.storage.clone()));
+        let task_id = task.id.clone();
+        manager.active_heals.lock().await.insert(task_id.clone(), task);
+        task_id
+    }
+
+    #[tokio::test]
+    async fn overlap_policy_minio_error_rejects_same_and_containing_paths() {
+        let manager = manager_with_policy(HealOverlapPolicy::MinioError);
+        insert_active_task(&manager, admin_prefix_request("bucket-a", "logs/")).await;
+
+        // Same target: typed AlreadyRunning.
+        let same = manager
+            .submit_heal_request(admin_prefix_request("bucket-a", "logs/"))
+            .await
+            .expect("admission must decide");
+        assert_eq!(
+            same,
+            HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning),
+            "an identical target must reject with already-running"
+        );
+
+        // Contained path: typed OverlappingPaths.
+        let nested = manager
+            .submit_heal_request(admin_prefix_request("bucket-a", "logs/app/"))
+            .await
+            .expect("admission must decide");
+        assert_eq!(
+            nested,
+            HealAdmissionResult::Dropped(HealAdmissionDropReason::OverlappingPaths),
+            "a path inside the active task's path must reject with overlapping-paths"
+        );
+
+        // Containing path (bucket-wide vs nested active): also overlapping.
+        let wide = manager
+            .submit_heal_request(admin_prefix_request("bucket-a", ""))
+            .await
+            .expect("admission must decide");
+        assert_eq!(
+            wide,
+            HealAdmissionResult::Dropped(HealAdmissionDropReason::OverlappingPaths),
+            "a bucket-wide start overlapping a nested active heal must reject"
+        );
+
+        // Disjoint bucket: unaffected.
+        let disjoint = manager
+            .submit_heal_request(admin_prefix_request("bucket-b", "logs/"))
+            .await
+            .expect("admission must decide");
+        assert_eq!(disjoint, HealAdmissionResult::Accepted);
+    }
+
+    #[tokio::test]
+    async fn overlap_policy_default_merge_keeps_today_semantics() {
+        let manager = manager_with_policy(HealOverlapPolicy::Merge);
+        insert_active_task(&manager, admin_prefix_request("bucket-a", "logs/")).await;
+
+        // Different-dedup-key overlap still merges under the default policy:
+        // the nested path dedups to its own key but nothing rejects it.
+        let nested = manager
+            .submit_heal_request(admin_prefix_request("bucket-a", "logs/app/"))
+            .await
+            .expect("admission must decide");
+        assert_eq!(nested, HealAdmissionResult::Accepted, "default policy must not reject overlaps");
+
+        // Non-admin sources never get overlap rejections even under minio_error.
+        let manager = manager_with_policy(HealOverlapPolicy::MinioError);
+        insert_active_task(&manager, admin_prefix_request("bucket-a", "logs/")).await;
+        let mut scanner_request = admin_prefix_request("bucket-a", "logs/app/");
+        scanner_request.source = HealRequestSource::Scanner;
+        let admitted = manager
+            .submit_heal_request(scanner_request)
+            .await
+            .expect("admission must decide");
+        assert_eq!(admitted, HealAdmissionResult::Accepted, "scanner sources must never be overlap-rejected");
+    }
+
+    #[tokio::test]
+    async fn admin_force_start_cancels_overlapping_active_task_first() {
+        let manager = manager_with_policy(HealOverlapPolicy::Merge);
+        let old_id = insert_active_task(&manager, admin_prefix_request("bucket-a", "logs/")).await;
+
+        let mut replacement = admin_prefix_request("bucket-a", "logs/");
+        replacement.force_start = true;
+        let receipt = manager
+            .submit_heal_request_with_receipt(replacement)
+            .await
+            .expect("force-start submission must decide");
+
+        assert!(receipt.result.is_admitted(), "the new task must be admitted (Accepted or Merged)");
+        let old_task_gone = {
+            let active_heals = manager.active_heals.lock().await;
+            !active_heals.contains_key(&old_id)
+        };
+        assert!(
+            old_task_gone,
+            "the overlapping admin task must be cancelled (removed from the active table) before the new one starts"
+        );
+        assert!(
+            matches!(manager.get_task_status(&old_id).await, Err(Error::TaskNotFound { .. })),
+            "a cancelled task must no longer resolve as an active heal"
+        );
+    }
+
     #[tokio::test]
     async fn test_operations_snapshot_counts_active_by_source_and_priority() {
         let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
@@ -5591,6 +5958,9 @@ mod tests {
                 status: HealTaskStatus::Completed,
                 result_items: Vec::new(),
                 result_items_truncated: false,
+                seqed_items: Vec::new(),
+                next_seq: 0,
+                min_seq: 0,
                 completed_at: SystemTime::now(),
             },
         );
@@ -5625,6 +5995,9 @@ mod tests {
                     ..Default::default()
                 }],
                 result_items_truncated: true,
+                seqed_items: Vec::new(),
+                next_seq: 0,
+                min_seq: 0,
                 completed_at: SystemTime::now(),
             },
         );

crates/heal/src/heal/task.rs

@@ -32,7 +32,7 @@ use std::{
     future::Future,
     sync::{
         Arc,
-        atomic::{AtomicBool, Ordering},
+        atomic::{AtomicBool, AtomicU64, Ordering},
     },
     time::{Duration, Instant, SystemTime},
 };
@@ -351,6 +351,20 @@ impl HealRequest {
 }
 
 /// Heal task
+/// Incremental view over a task's retained result items (HS-06).
+///
+/// `next_seq` is the cursor a client should pass on its next poll; `min_seq`
+/// is the oldest sequence still retained; `lagged` means the client's cursor
+/// fell behind `min_seq` and items were skipped — the client should restart
+/// from `min_seq`.
+#[derive(Debug, Clone)]
+pub struct HealResultWindow {
+    pub items: Vec<HealResultItem>,
+    pub next_seq: u64,
+    pub min_seq: u64,
+    pub lagged: bool,
+}
+
 pub struct HealTask {
     /// Task ID
     pub id: String,
@@ -373,8 +387,16 @@ pub struct HealTask {
     pub status: Arc<RwLock<HealTaskStatus>>,
     /// Progress tracking
     pub progress: Arc<RwLock<HealProgress>>,
-    /// Result items collected from storage heal calls.
+    /// Result items collected from storage heal calls, each stamped with a
-    pub result_items: Arc<RwLock<Vec<HealResultItem>>>,
+    /// monotonically increasing sequence number for incremental consumption
+    /// (the client passes the last seen seq back and receives only newer
+    /// items; see `get_result_items_since`).
+    pub result_items: Arc<RwLock<Vec<(u64, HealResultItem)>>>,
+    /// Next sequence number to assign; starts at 1.
+    next_item_seq: Arc<AtomicU64>,
+    /// Sequence number of the oldest item still inside the retention window;
+    /// equals `next_item_seq` while the window is empty.
+    min_available_seq: Arc<AtomicU64>,
     result_items_truncated: Arc<AtomicBool>,
     batch_failure: Arc<RwLock<Option<BatchHealFailure>>>,
     batch_failure_recorded: Arc<AtomicBool>,
@@ -426,6 +448,8 @@ impl HealTask {
             status: Arc::new(RwLock::new(HealTaskStatus::Pending)),
             progress: Arc::new(RwLock::new(HealProgress::new())),
             result_items: Arc::new(RwLock::new(Vec::new())),
+            next_item_seq: Arc::new(AtomicU64::new(1)),
+            min_available_seq: Arc::new(AtomicU64::new(1)),
             result_items_truncated: Arc::new(AtomicBool::new(false)),
             batch_failure: Arc::new(RwLock::new(None)),
             batch_failure_recorded: Arc::new(AtomicBool::new(false)),
@@ -911,18 +935,63 @@ impl HealTask {
     }
 
     pub async fn get_result_items(&self) -> Vec<HealResultItem> {
+        self.result_items.read().await.iter().map(|(_, item)| item.clone()).collect()
+    }
+
+    /// Sequence-stamped retained window, used when archiving a completed
+    /// task so incremental cursors survive the transition (HS-06).
+    pub async fn get_seqed_result_items(&self) -> Vec<(u64, HealResultItem)> {
         self.result_items.read().await.clone()
     }
 
+    /// Incremental result window (HS-06): `since = None` returns the full
+    /// retained window (legacy snapshot semantics); `since = Some(seq)`
+    /// returns only items stamped with a sequence greater than `seq`.
+    /// `lagged` warns that the caller's cursor fell behind the window start
+    /// and items were skipped (the response carries `min_seq` as the catch-up
+    /// cursor).
+    pub async fn get_result_items_since(&self, since: Option<u64>) -> HealResultWindow {
+        let result_items = self.result_items.read().await;
+        let next_seq = self.next_item_seq.load(Ordering::Relaxed);
+        let min_seq = self.min_available_seq.load(Ordering::Relaxed);
+        let mut lagged = false;
+        let items = match since {
+            None => result_items.iter().map(|(_, item)| item.clone()).collect::<Vec<_>>(),
+            Some(cursor) => {
+                if cursor + 1 < min_seq {
+                    lagged = true;
+                }
+                result_items
+                    .iter()
+                    .filter(|(seq, _)| *seq > cursor)
+                    .map(|(_, item)| item.clone())
+                    .collect::<Vec<_>>()
+            }
+        };
+        HealResultWindow {
+            items,
+            next_seq,
+            min_seq,
+            lagged,
+        }
+    }
+
     pub fn result_items_truncated(&self) -> bool {
         self.result_items_truncated.load(Ordering::Relaxed)
     }
 
     async fn record_result_item(&self, result: HealResultItem) {
+        let seq = self.next_item_seq.fetch_add(1, Ordering::Relaxed);
         let mut result_items = self.result_items.write().await;
         if result_items.len() < MAX_RETAINED_HEAL_RESULT_ITEMS {
-            result_items.push(result);
+            result_items.push((seq, result));
         } else {
+            // Slide the window: the oldest item leaves and the cursor for the
+            // oldest still-available item moves forward with it.
+            result_items.remove(0);
+            self.min_available_seq
+                .store(result_items.first().map_or(seq, |(oldest, _)| *oldest), Ordering::Relaxed);
+            result_items.push((seq, result));
             self.result_items_truncated.store(true, Ordering::Relaxed);
         }
     }
@@ -3880,6 +3949,69 @@ mod tests {
         assert!(task.result_items_truncated());
     }
 
+    // HS-06 (backlog#1870): incremental result windows.
+    #[tokio::test]
+    async fn result_items_seq_is_monotonic_and_incremental_slices_work() {
+        let storage = Arc::new(MockStorage::default());
+        let task = HealTask::from_request(HealRequest::bucket("bucket-a".to_string()), storage);
+
+        for round in 0..5u64 {
+            let item = HealResultItem {
+                object_size: round as usize,
+                ..Default::default()
+            };
+            task.record_result_item(item).await;
+        }
+
+        let full = task.get_result_items_since(None).await;
+        assert_eq!(full.items.len(), 5, "None keeps the full-snapshot semantics");
+        assert_eq!(full.next_seq, 6, "next_seq is one past the last assigned");
+        assert_eq!(full.min_seq, 1, "nothing was evicted yet");
+        assert!(!full.lagged);
+
+        // Incremental: only items newer than the cursor.
+        let incremental = task.get_result_items_since(Some(3)).await;
+        assert_eq!(
+            incremental.items.iter().map(|item| item.object_size).collect::<Vec<_>>(),
+            vec![3, 4],
+            "only sequences greater than the cursor are returned"
+        );
+        assert_eq!(incremental.next_seq, 6);
+
+        // A cursor at the head is not lagging.
+        assert!(!task.get_result_items_since(Some(0)).await.lagged);
+    }
+
+    #[tokio::test]
+    async fn result_items_window_slide_moves_min_seq_and_flags_lagging_cursors() {
+        let storage = Arc::new(MockStorage::default());
+        let task = HealTask::from_request(HealRequest::bucket("bucket-a".to_string()), storage);
+
+        // Fill the window completely, then push two more items: seq 1 and 2
+        // are evicted by the slide.
+        for _ in 0..(MAX_RETAINED_HEAL_RESULT_ITEMS + 2) {
+            task.record_result_item(HealResultItem::default()).await;
+        }
+
+        let full = task.get_result_items_since(None).await;
+        assert_eq!(full.items.len(), MAX_RETAINED_HEAL_RESULT_ITEMS);
+        assert_eq!(full.min_seq, 3, "each evicted head item moved the oldest-available cursor");
+        assert!(task.result_items_truncated());
+
+        // A client still polling from before the eviction is lagging.
+        let lagging = task.get_result_items_since(Some(0)).await;
+        assert!(lagging.lagged, "a cursor behind min_seq must be flagged");
+        assert_eq!(lagging.min_seq, 3, "the response tells the client where to restart");
+
+        // A cursor inside the window is fine.
+        assert!(!task.get_result_items_since(Some(3)).await.lagged);
+
+        // The lagging client restarts from min_seq and gets the full window.
+        let catch_up = task.get_result_items_since(Some(3)).await;
+        assert_eq!(catch_up.items.len(), MAX_RETAINED_HEAL_RESULT_ITEMS - 1);
+        assert!(!catch_up.lagged);
+    }
+
     #[tokio::test]
     async fn test_recursive_bucket_heal_skips_object_dir_candidates() {
         let storage = Arc::new(MockStorage {

crates/protos/src/heal_control.rs

@@ -177,16 +177,38 @@ impl StartCommand {
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag = "action", rename_all = "snake_case", deny_unknown_fields)]
 pub enum Command {
-    Start { request: StartCommand },
+    Start {
-    Query { heal_path: String, client_token: String },
+        request: StartCommand,
-    Cancel { heal_path: String, client_token: String },
+    },
+    Query {
+        heal_path: String,
+        client_token: String,
+        /// Incremental result cursor (HS-06): only items with a sequence
+        /// greater than this are returned. Absent = legacy full snapshot.
+        /// Optional + defaulted so older peers stay wire-compatible.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        since_seq: Option<u64>,
+    },
+    Cancel {
+        heal_path: String,
+        client_token: String,
+    },
 }
 
 #[derive(Debug)]
 pub enum ExecutableCommand {
-    Start { request: HealChannelRequest },
+    Start {
-    Query { heal_path: String, client_token: String },
+        request: HealChannelRequest,
-    Cancel { heal_path: String, client_token: String },
+    },
+    Query {
+        heal_path: String,
+        client_token: String,
+        since_seq: Option<u64>,
+    },
+    Cancel {
+        heal_path: String,
+        client_token: String,
+    },
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -227,8 +249,22 @@ impl Envelope {
         )
     }
 
-    pub fn query(request_id: String, metadata: RequestMetadata, heal_path: String, client_token: String) -> Result<Self, String> {
+    pub fn query(
-        Self::new(request_id, metadata, Command::Query { heal_path, client_token })
+        request_id: String,
+        metadata: RequestMetadata,
+        heal_path: String,
+        client_token: String,
+        since_seq: Option<u64>,
+    ) -> Result<Self, String> {
+        Self::new(
+            request_id,
+            metadata,
+            Command::Query {
+                heal_path,
+                client_token,
+                since_seq,
+            },
+        )
     }
 
     pub fn cancel(
@@ -286,7 +322,15 @@ impl Envelope {
             Command::Start { request } => ExecutableCommand::Start {
                 request: request.into_channel_request(self.request_id.clone())?,
             },
-            Command::Query { heal_path, client_token } => ExecutableCommand::Query { heal_path, client_token },
+            Command::Query {
+                heal_path,
+                client_token,
+                since_seq,
+            } => ExecutableCommand::Query {
+                heal_path,
+                client_token,
+                since_seq,
+            },
             Command::Cancel { heal_path, client_token } => ExecutableCommand::Cancel { heal_path, client_token },
         };
         Ok((self.request_id, self.coordinator_epoch, command))
@@ -305,6 +349,11 @@ pub enum Admission {
     Full,
     DroppedQueueFull,
     DroppedPolicy,
+    /// HS-06: admin start rejected because the same target is already being
+    /// healed (RUSTFS_HEAL_OVERLAP_POLICY=minio_error only).
+    DroppedAlreadyRunning,
+    /// HS-06: admin start rejected because its path overlaps an active heal.
+    DroppedOverlappingPaths,
 }
 
 impl From<HealAdmissionResult> for Admission {
@@ -315,6 +364,8 @@ impl From<HealAdmissionResult> for Admission {
             HealAdmissionResult::Full => Self::Full,
             HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull) => Self::DroppedQueueFull,
             HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped) => Self::DroppedPolicy,
+            HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning) => Self::DroppedAlreadyRunning,
+            HealAdmissionResult::Dropped(HealAdmissionDropReason::OverlappingPaths) => Self::DroppedOverlappingPaths,
         }
     }
 }
@@ -331,6 +382,8 @@ impl Admission {
             Self::Full => HealAdmissionResult::Full,
             Self::DroppedQueueFull => HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull),
             Self::DroppedPolicy => HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped),
+            Self::DroppedAlreadyRunning => HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning),
+            Self::DroppedOverlappingPaths => HealAdmissionResult::Dropped(HealAdmissionDropReason::OverlappingPaths),
         }
     }
 }
@@ -592,6 +645,7 @@ mod tests {
             metadata(2, 7),
             "bucket/prefix".to_string(),
             "token".to_string(),
+            None,
         )
         .unwrap();
         let cancel = Envelope::cancel(
@@ -667,6 +721,7 @@ mod tests {
             RequestMetadata::new([0x11; 16], 1_700_000_000_000, 1_700_000_030_000, 9),
             "bucket/prefix".to_string(),
             "client-token".to_string(),
+            None,
         )
         .unwrap();
         let cancel = Envelope::cancel(
@@ -749,7 +804,7 @@ mod tests {
         assert!(Envelope::start(test_request(request_id.clone()), metadata(0, 7)).is_err());
         assert!(Envelope::start(test_request(request_id.clone()), metadata(1, 0)).is_err());
         assert!(Envelope::start(test_request(request_id.clone()), RequestMetadata::new([1; 16], 1_000, 31_001, 7),).is_err());
-        assert!(Envelope::query(request_id.clone(), metadata(1, 7), String::new(), String::new()).is_err());
+        assert!(Envelope::query(request_id.clone(), metadata(1, 7), String::new(), String::new(), None).is_err());
         assert!(Envelope::cancel(request_id.clone(), metadata(1, 7), String::new(), String::new()).is_ok());
 
         let mut noncanonical_request = test_request(request_id.to_uppercase());
@@ -782,6 +837,7 @@ mod tests {
             metadata(1, 7),
             "x".repeat(ENVELOPE_MAX_SIZE),
             "token".to_string(),
+            None,
         )
         .unwrap();
         let error = super::encode_envelope(&oversized).unwrap_err();

crates/scanner/src/scanner_folder.rs

@@ -1770,6 +1770,13 @@ impl FolderScanner {
             HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped) => {
                 self.clear_pending_scanner_heal(kind, bucket, object, version_id);
             }
+            // Admin-only overlap rejections (HS-06); the scanner never sees
+            // them, but if it ever does, treat them as terminal like any
+            // other policy drop rather than endlessly retrying.
+            HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
+            | HealAdmissionResult::Dropped(HealAdmissionDropReason::OverlappingPaths) => {
+                self.clear_pending_scanner_heal(kind, bucket, object, version_id);
+            }
         }
     }
 

rustfs/src/admin/handlers/heal.rs

@@ -66,6 +66,9 @@ struct HealInitParams {
     client_token: String,
     force_start: bool,
     force_stop: bool,
+    /// Incremental result cursor (HS-06): only result items with a sequence
+    /// greater than this are returned; absent means full snapshot.
+    since_seq: Option<u64>,
 }
 
 fn extract_heal_init_params(body: &Bytes, uri: &Uri, params: Params<'_, '_>) -> S3Result<HealInitParams> {
@@ -98,6 +101,16 @@ fn extract_heal_init_params(body: &Bytes, uri: &Uri, params: Params<'_, '_>) ->
                     }
                     hip.force_stop = parse_heal_query_bool(value.as_ref())?;
                 }
+                "sinceSeq" => {
+                    if !seen.insert("sinceSeq") {
+                        return Err(s3_error!(InvalidArgument, "duplicate heal query parameter"));
+                    }
+                    hip.since_seq = Some(
+                        value
+                            .parse::<u64>()
+                            .map_err(|_| s3_error!(InvalidArgument, "sinceSeq must be a non-negative integer"))?,
+                    );
+                }
                 _ => return Err(s3_error!(InvalidArgument, "unknown heal query parameter")),
             }
         }
@@ -978,7 +991,15 @@ fn reject_heal_admission(result: rustfs_common::heal_channel::HealAdmissionResul
             result.result_label(),
             result.reason_label()
         ),
-        HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped) => s3_error!(
+        // Overlap rejections (HS-06) share this arm: the s3s footprint
+        // ratchet forbids new s3_error! sites, and the typed reason is
+        // preserved through reason_label() ("already_running" /
+        // "overlapping_paths") so madmin-style clients can distinguish.
+        HealAdmissionResult::Dropped(
+            HealAdmissionDropReason::PolicyDropped
+            | HealAdmissionDropReason::AlreadyRunning
+            | HealAdmissionDropReason::OverlappingPaths,
+        ) => s3_error!(
             OperationAborted,
             "heal request not admitted: admission={}, reason={}",
             result.result_label(),
@@ -1403,6 +1424,7 @@ impl Operation for HealHandler {
                 new_heal_control_metadata(&route)?,
                 heal_path,
                 client_token.clone(),
+                hip.since_seq,
             )
             .map_err(|err| s3_error!(InternalError, "encode heal control query failed: {err}"))?;
             let response = submit_cluster_heal_channel_command(context, route, envelope, &request_id, client_token).await?;

rustfs/src/storage/rpc/node_service.rs

@@ -571,8 +571,12 @@ async fn execute_heal_control_envelope_with_manager(
                 admission: receipt.result.into(),
             }
         }
-        rustfs_protos::heal_control::ExecutableCommand::Query { heal_path, client_token } => {
+        rustfs_protos::heal_control::ExecutableCommand::Query {
-            let response = timeout(remaining, processor.execute_query_request(heal_path, client_token))
+            heal_path,
+            client_token,
+            since_seq,
+        } => {
+            let response = timeout(remaining, processor.execute_query_request_since(heal_path, client_token, since_seq))
                 .await
                 .map_err(|_| Status::deadline_exceeded("heal control query expired before execution"))?
                 .map_err(|_| Status::internal("heal control query failed"))?;
@@ -2517,6 +2521,7 @@ mod tests {
             metadata(),
             "bucket/prefix".to_string(),
             canonical_token.clone(),
+            None,
         )
         .unwrap();
         let query_result = execute_heal_control_envelope_with_manager(query, coordinator_epoch, Some(Arc::clone(&manager)))
@@ -2555,6 +2560,7 @@ mod tests {
             metadata(),
             "bucket/prefix".to_string(),
             canonical_token,
+            None,
         )
         .unwrap();
         let stopped_result = execute_heal_control_envelope_with_manager(stopped_query, coordinator_epoch, Some(manager))

scripts/check_s3s_footprint.sh

@@ -25,7 +25,7 @@ cd "$(dirname "$0")/.."
 # Baselines verified on 2026-08-11. Lower-only; see header.
 # Excludes crates/e2e_test/ — test infrastructure legitimately uses s3s
 # to verify S3 behavior and does not widen the production s3s surface.
-S3S_IMPORT_FILES_BASELINE=213
+S3S_IMPORT_FILES_BASELINE=211
 S3_ERROR_LINES_BASELINE=1620
 S3S_PATH_PATTERN='(^|[^"[:alnum:]_])s3s::'
 E2E_TEST_GLOB='--glob=!crates/e2e_test/**'