diff --git a/Cargo.lock b/Cargo.lock index b297135af..83749d574 100644 --- a/Cargo.lock +++ b/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", diff --git a/crates/common/src/heal_channel.rs b/crates/common/src/heal_channel.rs index b5a907e4e..f8162dee6 100644 --- a/crates/common/src/heal_channel.rs +++ b/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, response_tx: oneshot::Sender>, }, /// 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 { + 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, +) -> Result { let (response_tx, response_rx) = oneshot::channel(); send_heal_command(HealChannelCommand::Query { heal_path, client_token, + since_seq, response_tx, }) .await?; diff --git a/crates/config/src/constants/heal.rs b/crates/config/src/constants/heal.rs index b8cf3630a..9c27a663f 100644 --- a/crates/config/src/constants/heal.rs +++ b/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"; diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index 1e31056b2..8f4ab9e8a 100644 --- a/crates/heal/src/heal/channel.rs +++ b/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, progress: Option<&HealProgress>, mut truncated: bool, + next_seq: u64, + min_seq: u64, ) -> Result<(Vec, 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, truncated: bool, + next_seq: u64, + min_seq: u64, ) -> Result<(Vec, Option)> { - 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 { + 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, + ) -> Result { 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, response_tx: oneshot::Sender>, ) -> 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"); diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 216e17068..3a243c88e 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -220,6 +220,11 @@ struct CompletedHealStatus { result_items: Vec, 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, pub result_items_truncated: bool, pub progress: Option, + /// 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) -> 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) -> 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 { + 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 { + // 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 = { + 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 { + 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) -> Result { let canonical_task_id = self.canonical_task_id(task_id).await; { let active_heals = self.active_heals.lock().await; if let Some(task) = active_heals.get(&canonical_task_id) { - return Ok(HealTaskReport { - status: task.get_status().await, - result_items: task.get_result_items().await, - result_items_truncated: task.result_items_truncated(), - progress: Some(task.get_progress().await), - }); + return Ok(active_task_report(task, since).await); } } { let retrying_heals = self.retrying_heals.lock().await; if let Some(retrying) = retrying_heals.get(&canonical_task_id) { - return Ok(HealTaskReport { - status: retrying.status(), - result_items: Vec::new(), - result_items_truncated: false, - progress: None, - }); + return Ok(empty_task_report(retrying.status())); } } @@ -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 { - status: completed.status.clone(), - result_items: completed.result_items.clone(), - result_items_truncated: completed.result_items_truncated, - progress: None, - }); + return Ok(completed_task_report(completed, since)); } } { let queue = self.heal_queue.lock().await; if queue.contains_request_id(&canonical_task_id) { - return Ok(HealTaskReport { - status: HealTaskStatus::Pending, - result_items: Vec::new(), - result_items_truncated: false, - progress: None, - }); + return Ok(empty_task_report(HealTaskStatus::Pending)); } } let mut completed_heals = self.completed_heals.lock().await; prune_completed_heal_statuses(&mut completed_heals); if let Some(completed) = completed_heals.get(&canonical_task_id) { - return Ok(HealTaskReport { - status: completed.status.clone(), - result_items: completed.result_items.clone(), - result_items_truncated: completed.result_items_truncated, - progress: None, - }); + return Ok(completed_task_report(completed, since)); } 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 { + 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, + ) -> Result { let canonical_task_id = self.canonical_task_id(task_id).await; { let active_heals = self.active_heals.lock().await; if let Some(task) = active_heals.get(&canonical_task_id) && heal_type_matches_path(&task.heal_type, heal_path) { - 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), - }); + return Ok(active_task_report(task, since).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 { - status: retrying.status(), - result_items: Vec::new(), - result_items_truncated: false, - progress: None, - }); + return Ok(empty_task_report(retrying.status())); } } @@ -2004,24 +2249,14 @@ impl HealManager { && heal_type_matches_path(&completed.heal_type, heal_path) && completed_status_is_retrying(&completed.status) { - return Ok(HealTaskReport { - status: completed.status.clone(), - result_items: completed.result_items.clone(), - result_items_truncated: completed.result_items_truncated, - progress: None, - }); + return Ok(completed_task_report(completed, since)); } } { let queue = self.heal_queue.lock().await; if queue.contains_request_id_matching_path(&canonical_task_id, heal_path) { - return Ok(HealTaskReport { - status: HealTaskStatus::Pending, - result_items: Vec::new(), - result_items_truncated: false, - progress: None, - }); + return Ok(empty_task_report(HealTaskStatus::Pending)); } } @@ -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 { - status: completed.status.clone(), - result_items: completed.result_items.clone(), - result_items_truncated: completed.result_items_truncated, - progress: None, - }); + return Ok(completed_task_report(completed, since)); } } @@ -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 = 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 = 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(), }, ); diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index f6472eb65..2c0626d8f 100644 --- a/crates/heal/src/heal/task.rs +++ b/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, + 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>, /// Progress tracking pub progress: Arc>, - /// Result items collected from storage heal calls. - pub result_items: Arc>>, + /// Result items collected from storage heal calls, each stamped with a + /// 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>>, + /// Next sequence number to assign; starts at 1. + next_item_seq: Arc, + /// Sequence number of the oldest item still inside the retention window; + /// equals `next_item_seq` while the window is empty. + min_available_seq: Arc, result_items_truncated: Arc, batch_failure: Arc>>, batch_failure_recorded: Arc, @@ -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 { + 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) -> 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::>(), + Some(cursor) => { + if cursor + 1 < min_seq { + lagged = true; + } + result_items + .iter() + .filter(|(seq, _)| *seq > cursor) + .map(|(_, item)| item.clone()) + .collect::>() + } + }; + 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![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 { diff --git a/crates/protos/src/heal_control.rs b/crates/protos/src/heal_control.rs index df89126d2..515c0db75 100644 --- a/crates/protos/src/heal_control.rs +++ b/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 }, - Query { heal_path: String, client_token: String }, - Cancel { heal_path: String, client_token: String }, + Start { + request: StartCommand, + }, + 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, + }, + Cancel { + heal_path: String, + client_token: String, + }, } #[derive(Debug)] pub enum ExecutableCommand { - Start { request: HealChannelRequest }, - Query { heal_path: String, client_token: String }, - Cancel { heal_path: String, client_token: String }, + Start { + request: HealChannelRequest, + }, + Query { + heal_path: String, + client_token: String, + since_seq: Option, + }, + 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::new(request_id, metadata, Command::Query { heal_path, client_token }) + pub fn query( + request_id: String, + metadata: RequestMetadata, + heal_path: String, + client_token: String, + since_seq: Option, + ) -> Result { + 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 for Admission { @@ -315,6 +364,8 @@ impl From 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(); diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 030c55fc9..5766385e8 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/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); + } } } diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 52e9fc152..3d98e966a 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/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, } fn extract_heal_init_params(body: &Bytes, uri: &Uri, params: Params<'_, '_>) -> S3Result { @@ -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::() + .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?; diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index a39ca6c3a..352bfbd58 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/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 } => { - let response = timeout(remaining, processor.execute_query_request(heal_path, client_token)) + rustfs_protos::heal_control::ExecutableCommand::Query { + 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)) diff --git a/scripts/check_s3s_footprint.sh b/scripts/check_s3s_footprint.sh index 56fb60d35..1ebd59f5b 100755 --- a/scripts/check_s3s_footprint.sh +++ b/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/**'