diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index 580b01246..d339a3b5a 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -76,6 +76,8 @@ struct HealTaskStatusPayload<'a> { min_seq: u64, #[serde(skip_serializing_if = "Option::is_none")] progress: Option<&'a HealProgress>, + #[serde(skip_serializing_if = "Option::is_none")] + outcome: Option<&'a super::outcome::HealTaskOutcome>, } fn u64_is_zero(value: &u64) -> bool { @@ -87,17 +89,18 @@ fn encode_heal_task_status_payload( mut items: Vec, progress: Option<&HealProgress>, mut truncated: bool, - next_seq: u64, - min_seq: u64, + sequence: (u64, u64), + outcome: Option<&super::outcome::HealTaskOutcome>, ) -> Result<(Vec, bool)> { loop { let data = serde_json::to_vec(&HealTaskStatusPayload { summary, items: &items, truncated, - next_seq, - min_seq, + next_seq: sequence.0, + min_seq: sequence.1, progress, + outcome, }) .map_err(|e| Error::Serialization(format!("failed to serialize heal task status: {e}")))?; if data.len() <= MAX_HEAL_STATUS_PAYLOAD_SIZE { @@ -111,25 +114,21 @@ fn encode_heal_task_status_payload( } } -fn heal_status_detail(detail: Option, truncated: bool) -> Option { - if !truncated { - return detail; - } - let truncation = "heal result items were truncated"; - Some(detail.map_or_else(|| truncation.to_string(), |detail| format!("{detail}; {truncation}"))) -} - fn encode_heal_status_response( summary: &str, items: Vec, progress: Option<&HealProgress>, detail: Option, truncated: bool, - next_seq: u64, - min_seq: u64, + sequence: (u64, u64), + outcome: Option<&super::outcome::HealTaskOutcome>, ) -> Result<(Vec, Option)> { - let (data, truncated) = encode_heal_task_status_payload(summary, items, progress, truncated, next_seq, min_seq)?; - Ok((data, heal_status_detail(detail, truncated))) + let (summary, detail) = match outcome { + Some(outcome) => outcome.legacy_status(summary, detail), + None => (summary, detail), + }; + let (data, truncated) = encode_heal_task_status_payload(summary, items, progress, truncated, sequence, outcome)?; + Ok((data, super::outcome::heal_status_detail(detail, truncated))) } impl HealChannelProcessor { @@ -439,6 +438,7 @@ impl HealChannelProcessor { .await }; + let outcome = report.as_ref().ok().and_then(|report| report.outcome.clone()); let (summary, detail, items, truncated, progress, next_seq, min_seq) = match report { Ok(HealTaskReport { status: HealTaskStatus::Pending | HealTaskStatus::Running, @@ -576,8 +576,15 @@ impl HealChannelProcessor { } }; - let (data, detail) = - encode_heal_status_response(&summary, items, progress.as_ref(), detail, truncated, next_seq, min_seq)?; + let (data, detail) = encode_heal_status_response( + &summary, + items, + progress.as_ref(), + detail, + truncated, + (next_seq, min_seq), + outcome.as_deref(), + )?; let response = HealChannelResponse { request_id: client_token, @@ -866,7 +873,7 @@ mod tests { ..Default::default() }]; - let (data, detail) = encode_heal_status_response("running", items, None, None, false, 0, 0).unwrap(); + let (data, detail) = encode_heal_status_response("running", items, None, None, false, (0, 0), None).unwrap(); assert!(data.len() <= MAX_HEAL_STATUS_PAYLOAD_SIZE); let payload: serde_json::Value = serde_json::from_slice(&data).unwrap(); @@ -875,6 +882,135 @@ mod tests { assert_eq!(detail.as_deref(), Some("heal result items were truncated")); } + #[test] + fn outcome_v3_fixture_matches_canonical_owner_and_preserves_legacy_terminals() { + use crate::heal::outcome::*; + let cases: serde_json::Value = serde_json::from_str(include_str!("../../../madmin/tests/fixtures/heal-outcome-v3.json")) + .expect("shared client fixtures"); + for case in cases.as_array().expect("fixture cases") { + if case.get("remoteResponse").is_some() { + continue; + } + let mut outcome = HealTaskOutcome::default(); + let name = case["name"].as_str().expect("case name"); + if matches!(name, "unknown" | "completed_with_errors") { + let disposition = if name == "unknown" { + HealObjectDisposition::Unknown + } else { + outcome.attempt_failed(); + HealObjectDisposition::Failed(HealFailureClass::RetryExhausted) + }; + outcome.record(HealObjectOutcome { + identity: HealObjectIdentity { + kind: HealObjectKind::Object, + bucket: "bucket".into(), + object: "object".into(), + version_id: None, + bucket_incarnation_id: None, + pool_index: None, + set_index: None, + }, + disposition, + detail: None, + }); + } + let abort = match name { + "cancelled" => Some(HealAbortReason::Cancelled), + "deadline" => Some(HealAbortReason::Deadline), + "untraversable" => Some(HealAbortReason::Untraversable), + _ => None, + }; + outcome.finish(abort); + let expected = &case["response"]; + let initial_detail = abort.map(|reason| { + match reason { + HealAbortReason::Cancelled => "heal task cancelled", + HealAbortReason::Deadline => "heal task timed out", + HealAbortReason::Untraversable => "heal listing is untraversable", + } + .to_string() + }); + let (bytes, detail) = encode_heal_status_response( + if abort.is_some() { "stopped" } else { "finished" }, + Vec::new(), + None, + initial_detail, + true, + (9, 4), + Some(&outcome), + ) + .expect("canonical owner encoding"); + let decoded: serde_json::Value = serde_json::from_slice(&bytes).expect("wire payload"); + assert_eq!(decoded["summary"], expected["summary"], "{name}"); + assert_eq!(detail.unwrap_or_default(), expected["detail"].as_str().expect("detail"), "{name}"); + assert_eq!(decoded["outcome"], expected["outcome"], "{name}"); + assert_eq!((decoded["next_seq"].as_u64(), decoded["min_seq"].as_u64()), (Some(9), Some(4))); + assert!(decoded["outcome"].get("retainedObjectBytes").is_none()); + assert!(decoded["outcome"].get("untraversable").is_none()); + } + } + + #[test] + fn outcome_v3_abort_cannot_be_hidden_by_a_finished_status() { + use crate::heal::outcome::{HealAbortReason, HealTaskOutcome}; + for reason in [ + HealAbortReason::Cancelled, + HealAbortReason::Deadline, + HealAbortReason::Untraversable, + ] { + let mut outcome = HealTaskOutcome::default(); + outcome.finish(Some(reason)); + let (data, detail) = encode_heal_status_response("finished", Vec::new(), None, None, false, (0, 0), Some(&outcome)) + .expect("canonical abort adapter"); + let json: serde_json::Value = serde_json::from_slice(&data).expect("public state"); + assert_eq!(json["summary"], "stopped"); + assert_eq!(json["outcome"]["execution"]["state"], "aborted"); + assert!(detail.is_some()); + } + } + + #[test] + fn outcome_v3_payload_bound_keeps_cumulative_outcome_and_cursors() { + use crate::heal::outcome::*; + let mut outcome = HealTaskOutcome::default(); + outcome.start(); + for index in 0..256 { + outcome.record(HealObjectOutcome { + identity: HealObjectIdentity { + kind: HealObjectKind::Object, + bucket: "bucket".into(), + object: format!("object-{index}"), + version_id: None, + bucket_incarnation_id: None, + pool_index: None, + set_index: None, + }, + disposition: HealObjectDisposition::Unknown, + detail: Some("\"".repeat(1024)), + }); + } + let retained = outcome.objects.len(); + let items = vec![ + HealResultItem::default(), + HealResultItem { + detail: "x".repeat(MAX_HEAL_STATUS_PAYLOAD_SIZE + 1), + ..Default::default() + }, + ]; + let (bytes, detail) = encode_heal_status_response("running", items, None, None, false, (9, 4), Some(&outcome)) + .expect("bounded status with cumulative outcome"); + assert!(bytes.len() <= MAX_HEAL_STATUS_PAYLOAD_SIZE); + let wire: serde_json::Value = serde_json::from_slice(&bytes).expect("bounded payload"); + assert_eq!(wire["items"].as_array().expect("items").len(), 1); + assert_eq!(wire["truncated"], true); + assert_eq!((wire["next_seq"].as_u64(), wire["min_seq"].as_u64()), (Some(9), Some(4))); + assert_eq!(wire["outcome"]["counters"]["processed"], 256); + assert_eq!(wire["outcome"]["counters"]["healed"], 0); + assert_eq!(wire["outcome"]["objects"].as_array().expect("outcome window").len(), retained); + assert!(retained < 256 && outcome.objects_truncated); + assert_eq!(detail.as_deref(), Some("heal result items were truncated")); + } + #[test] fn admission_response_preserves_all_admission_outcomes() { let cases = [ diff --git a/crates/heal/src/heal/outcome.rs b/crates/heal/src/heal/outcome.rs index ada74376d..a283b0117 100644 --- a/crates/heal/src/heal/outcome.rs +++ b/crates/heal/src/heal/outcome.rs @@ -15,6 +15,7 @@ //! Execution results are separate from repair responsibility. A legacy //! successful storage call supplies no authoritative repair receipt. +use serde::{Deserialize, Serialize}; use std::{collections::VecDeque, time::SystemTime}; use uuid::Uuid; @@ -22,14 +23,16 @@ const MAX_OUTCOME_ITEMS: usize = 128; const MAX_OUTCOME_BYTES: usize = 64 * 1024; const MAX_OUTCOME_DETAIL_BYTES: usize = 1024; -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] pub enum HealObjectKind { Object, Metadata, Decode, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] pub struct HealObjectIdentity { pub kind: HealObjectKind, pub bucket: String, @@ -41,7 +44,8 @@ pub struct HealObjectIdentity { pub set_index: Option, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] pub enum HealDeferredReason { DanglingDeleteGrace, TransientUsageCache, @@ -49,14 +53,21 @@ pub enum HealDeferredReason { Deadline, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] pub enum HealFailureClass { Recoverable, RetryExhausted, Permanent, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde( + tag = "state", + content = "details", + rename_all = "snake_case", + rename_all_fields = "camelCase" +)] pub enum HealObjectDisposition { /// The legacy storage response does not prove the requested check or commit. Unknown, @@ -72,7 +83,8 @@ pub enum HealObjectDisposition { DryRunObserved, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] pub struct HealObjectOutcome { pub identity: HealObjectIdentity, pub disposition: HealObjectDisposition, @@ -89,7 +101,8 @@ impl HealObjectOutcome { } } -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] pub enum HealTraversalCoverage { #[default] Unknown, @@ -97,14 +110,16 @@ pub enum HealTraversalCoverage { Complete, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] pub enum HealAbortReason { Cancelled, Deadline, Untraversable, } -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "state", content = "reason", rename_all = "snake_case")] pub enum HealExecutionOutcome { #[default] Pending, @@ -114,7 +129,8 @@ pub enum HealExecutionOutcome { Aborted(HealAbortReason), } -#[derive(Debug, Clone, Default, PartialEq, Eq)] +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] pub struct HealOutcomeCounters { pub processed: u64, pub healed: u64, @@ -127,7 +143,8 @@ pub struct HealOutcomeCounters { pub overflowed: bool, } -#[derive(Debug, Clone, Default, PartialEq, Eq)] +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] pub struct HealTaskOutcome { pub execution: HealExecutionOutcome, pub coverage: HealTraversalCoverage, @@ -135,11 +152,99 @@ pub struct HealTaskOutcome { /// A bounded diagnostic window, not a complete responsibility ledger. pub objects: VecDeque, pub objects_truncated: bool, + #[serde(skip)] retained_object_bytes: usize, + #[serde(skip)] untraversable: bool, } +#[derive(Debug, thiserror::Error)] +pub enum HealOutcomeWireError { + #[error("heal outcome is missing execution or counters")] + MissingFields, + #[error("heal outcome has invalid or unsupported execution fields")] + InvalidFields(#[from] serde_json::Error), + #[error("finished heal summary contradicts its canonical outcome")] + ContradictoryCompletion, +} + +/// Reconcile a peer's successful legacy summary using the canonical owner types. +/// A running retry may legitimately retain the preceding attempt's outcome. +pub fn legacy_wire_status<'a>( + summary: &'a str, + wire: &serde_json::Value, + truncated: bool, +) -> Result<(&'a str, Option), HealOutcomeWireError> { + if summary != "finished" { + return Ok((summary, None)); + } + let execution = HealExecutionOutcome::deserialize(wire.get("execution").ok_or(HealOutcomeWireError::MissingFields)?)?; + let counters = HealOutcomeCounters::deserialize(wire.get("counters").ok_or(HealOutcomeWireError::MissingFields)?)?; + if matches!(execution, HealExecutionOutcome::Pending | HealExecutionOutcome::Running) + || (execution == HealExecutionOutcome::Completed && counters.failed > 0) + { + return Err(HealOutcomeWireError::ContradictoryCompletion); + } + let (adapted, detail) = legacy_execution_status(summary, None, execution, &counters); + Ok(( + adapted, + if adapted != summary { + heal_status_detail(detail, truncated) + } else { + None + }, + )) +} + +pub(crate) fn heal_status_detail(detail: Option, truncated: bool) -> Option { + if !truncated { + return detail; + } + let truncation = "heal result items were truncated"; + Some(detail.map_or_else(|| truncation.to_string(), |detail| format!("{detail}; {truncation}"))) +} + +fn legacy_execution_status<'a>( + summary: &'a str, + detail: Option, + execution: HealExecutionOutcome, + counters: &HealOutcomeCounters, +) -> (&'a str, Option) { + if summary != "finished" { + return (summary, detail); + } + match execution { + HealExecutionOutcome::CompletedWithErrors => ( + "stopped", + Some(format!("heal traversal completed with errors: {} failed objects", counters.failed)), + ), + HealExecutionOutcome::Aborted(reason) => { + let reason = match reason { + HealAbortReason::Cancelled => "cancelled", + HealAbortReason::Deadline => "timed out", + HealAbortReason::Untraversable => "untraversable", + }; + ("stopped", Some(format!("heal task {reason}"))) + } + HealExecutionOutcome::Completed if counters.unknown > 0 => ( + summary, + Some(format!( + "heal traversal completed; authoritative storage proof is unavailable for {} objects", + counters.unknown + )), + ), + HealExecutionOutcome::Pending | HealExecutionOutcome::Running => { + ("running", Some("heal execution has not reached a terminal outcome".to_string())) + } + HealExecutionOutcome::Completed => (summary, detail), + } +} + impl HealTaskOutcome { + pub(crate) fn legacy_status<'a>(&self, summary: &'a str, detail: Option) -> (&'a str, Option) { + legacy_execution_status(summary, detail, self.execution, &self.counters) + } + pub(crate) fn start(&mut self) { if self.execution != HealExecutionOutcome::Aborted(HealAbortReason::Cancelled) { self.execution = HealExecutionOutcome::Running; @@ -292,6 +397,69 @@ mod canonical_outcome_tests { assert!(outcome.objects.len() < MAX_OUTCOME_ITEMS); } + #[test] + fn outcome_v3_serialization_keeps_unverified_dispositions_and_window_bounds() { + let mut outcome = HealTaskOutcome::default(); + for disposition in [ + HealObjectDisposition::Unknown, + HealObjectDisposition::Deferred { + reason: HealDeferredReason::DanglingDeleteGrace, + retry_not_before: None, + }, + HealObjectDisposition::DryRunObserved, + ] { + outcome.record(item(disposition)); + } + outcome.finish(None); + let wire = serde_json::to_value(&outcome).expect("canonical wire view"); + assert_eq!(wire["execution"]["state"], "completed"); + assert_eq!(wire["counters"]["healed"], 0); + assert_eq!(wire["counters"]["skipped"], 3); + assert_eq!(wire["objects"][1]["disposition"]["details"]["reason"], "dangling_delete_grace"); + assert!(wire["objects"][1]["identity"]["bucketIncarnationId"].is_null()); + for _ in 0..MAX_OUTCOME_ITEMS + 1 { + let mut result = item(HealObjectDisposition::Unknown); + result.detail = Some("\"".repeat(MAX_OUTCOME_DETAIL_BYTES)); + outcome.record(result); + } + let bytes = serde_json::to_vec(&outcome).expect("bounded canonical samples"); + assert!( + bytes.len() < 8 * MAX_OUTCOME_BYTES, + "JSON escaping remains bounded independently of object count" + ); + assert!(outcome.objects_truncated); + } + + #[test] + fn outcome_v3_wire_consistency_rejects_unknown_success_without_rejecting_extensions() { + let mut outcome = HealTaskOutcome::default(); + outcome.finish(None); + let mut wire = serde_json::to_value(&outcome).expect("canonical snapshot"); + wire["execution"]["futureField"] = serde_json::json!({"new": true}); + wire["counters"]["futureCounter"] = serde_json::json!(42); + assert_eq!( + legacy_wire_status("finished", &wire, false).expect("unknown extension fields"), + ("finished", None) + ); + wire["execution"]["state"] = serde_json::json!("future_execution"); + assert!(legacy_wire_status("finished", &wire, false).is_err()); + assert_eq!( + legacy_wire_status("running", &wire, false).expect("unknown nonterminal outcome"), + ("running", None) + ); + wire["execution"] = serde_json::json!({"state":"completed"}); + wire["counters"]["failed"] = serde_json::json!(1); + assert!(matches!( + legacy_wire_status("finished", &wire, false), + Err(HealOutcomeWireError::ContradictoryCompletion) + )); + wire.as_object_mut().expect("object").remove("execution"); + assert!(matches!( + legacy_wire_status("finished", &wire, false), + Err(HealOutcomeWireError::MissingFields) + )); + } + #[test] fn canonical_outcome_counter_overflow_cannot_claim_complete_coverage() { let mut outcome = HealTaskOutcome::default(); diff --git a/crates/madmin/src/client.rs b/crates/madmin/src/client.rs index b49fde280..982304308 100644 --- a/crates/madmin/src/client.rs +++ b/crates/madmin/src/client.rs @@ -167,6 +167,13 @@ pub struct HealTaskStatus { /// Live progress snapshot; the exact shape is owned by the heal runtime. #[serde(default)] pub progress: Option, + /// Canonical heal-owner result. Missing or future states are not repair proof. + #[serde(default)] + pub outcome: Option, + #[serde(default, alias = "next_seq")] + pub next_seq: Option, + #[serde(default, alias = "min_seq")] + pub min_seq: Option, } /// `POST /v3/background-heal/status` response. Known top-level fields are @@ -358,8 +365,22 @@ impl AdminClient { prefix: Option<&str>, client_token: &str, ) -> Result { - self.post_json(&heal_path(bucket, prefix), &[("clientToken", client_token.to_string())], Vec::new()) - .await + self.heal_status_since(bucket, prefix, client_token, None).await + } + + /// Query a retained result window. Missing cursors and outcome remain unknown. + pub async fn heal_status_since( + &self, + bucket: Option<&str>, + prefix: Option<&str>, + client_token: &str, + since_seq: Option, + ) -> Result { + let mut query = vec![("clientToken", client_token.to_string())]; + if let Some(since_seq) = since_seq { + query.push(("sinceSeq", since_seq.to_string())); + } + self.post_json(&heal_path(bucket, prefix), &query, Vec::new()).await } /// Stop a heal: with a `client_token` only that task is cancelled and its @@ -378,7 +399,7 @@ impl AdminClient { match client_token { Some(_) => { let status: HealTaskStatus = self.post_json(&heal_path(bucket, prefix), &query, Vec::new()).await?; - Ok(HealStopOutcome::Stopped(status)) + Ok(HealStopOutcome::Stopped(Box::new(status))) } None => { let success: HealStartSuccess = self.post_json(&heal_path(bucket, prefix), &query, Vec::new()).await?; @@ -533,7 +554,7 @@ impl AdminClient { /// start-success-shaped receipt. #[derive(Debug, Clone)] pub enum HealStopOutcome { - Stopped(HealTaskStatus), + Stopped(Box), PathStopped(HealStartSuccess), } @@ -635,6 +656,24 @@ mod tests { assert!(status.progress.is_none()); } + #[test] + fn outcome_v3_decoder_preserves_canonical_unknown_and_future_fields() { + let cases: serde_json::Value = + serde_json::from_str(include_str!("../tests/fixtures/heal-outcome-v3.json")).expect("shared fixtures"); + for case in cases.as_array().expect("cases") { + let status: HealTaskStatus = serde_json::from_value(case["response"].clone()).expect("optional outcome response"); + assert_eq!(status.outcome.as_ref(), Some(&case["response"]["outcome"])); + assert_eq!((status.next_seq, status.min_seq), (Some(9), Some(4))); + assert!(status.truncated); + } + let old: HealTaskStatus = serde_json::from_value(json!({"summary":"finished"})).expect("legacy response"); + assert!(old.outcome.is_none() && old.next_seq.is_none() && old.min_seq.is_none()); + let future = json!({"execution":{"state":"future_state"},"newField":7}); + let status: HealTaskStatus = + serde_json::from_value(json!({"summary":"running","outcome":future})).expect("future outcome remains opaque"); + assert_eq!(status.outcome, Some(future)); + } + #[test] fn background_heal_status_types_known_fields_and_passes_the_rest_through() { let raw = json!({ @@ -750,6 +789,26 @@ mod tests { assert!(!request.query.contains("forceStop")); } + #[tokio::test] + async fn outcome_v3_since_query_preserves_cursor_and_never_sends_force_start() { + let server = TestServer::spawn( + r#"{"summary":"running","nextSeq":9,"minSeq":4,"truncated":true,"outcome":{"execution":{"state":"future_state"}}}"#, + 200, + ) + .await; + let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").expect("test client"); + let status = client + .heal_status_since(Some("bucket"), None, "token-1", Some(3)) + .await + .expect("window response"); + assert_eq!((status.next_seq, status.min_seq), (Some(9), Some(4))); + assert!(status.truncated); + assert_eq!(status.outcome.expect("future state is preserved")["execution"]["state"], "future_state"); + let request = server.recorded(); + assert!(request.query.contains("sinceSeq=3") && request.query.contains("clientToken=token-1")); + assert!(!request.query.contains("forceStart") && !request.query.contains("forceStop")); + } + #[tokio::test] async fn stop_without_token_takes_the_path_cancel_branch() { let server = TestServer::spawn(r#"{"clientToken":"path","clientAddress":"c","startTime":"t"}"#, 200).await; @@ -762,6 +821,25 @@ mod tests { assert!(!request.query.contains("clientToken")); } + #[tokio::test] + async fn stop_with_token_decodes_boxed_task_status() { + let body = r#"{"summary":"stopped","detail":"","settings":{"recursive":false},"items":[],"truncated":false}"#; + let server = TestServer::spawn(body, 200).await; + let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap(); + + let outcome = client + .heal_stop(Some("bucket"), None, Some("token-1")) + .await + .expect("token stop decodes"); + let super::HealStopOutcome::Stopped(status) = outcome else { + panic!("token stop should return task status"); + }; + assert_eq!(status.summary, "stopped"); + let request = server.recorded(); + assert!(request.query.contains("forceStop=true")); + assert!(request.query.contains("clientToken=token-1")); + } + #[tokio::test] async fn background_heal_status_posts_to_the_registered_route() { let body = r#"{"state":"idle","healQueueLength":0,"healActiveTasks":0,"clusterStatusComplete":true}"#; diff --git a/crates/madmin/tests/fixtures/heal-outcome-v3.json b/crates/madmin/tests/fixtures/heal-outcome-v3.json new file mode 100644 index 000000000..9f634af80 --- /dev/null +++ b/crates/madmin/tests/fixtures/heal-outcome-v3.json @@ -0,0 +1,472 @@ +[ + { + "name": "completed", + "cliExit": 0, + "response": { + "summary": "finished", + "detail": "heal result items were truncated", + "startTime": "2026-01-01T00:00:00Z", + "settings": { + "recursive": true, + "scanMode": 1 + }, + "items": [], + "truncated": true, + "nextSeq": 9, + "minSeq": 4, + "progress": { + "objectsScanned": 11, + "objectsHealed": 7 + }, + "outcome": { + "execution": { + "state": "completed" + }, + "coverage": "complete", + "counters": { + "processed": 0, + "healed": 0, + "unchanged": 0, + "skipped": 0, + "failed": 0, + "unknown": 0, + "attemptFailures": 0, + "overflowed": false + }, + "objects": [], + "objectsTruncated": false + } + } + }, + { + "name": "unknown", + "cliExit": 0, + "response": { + "summary": "finished", + "detail": "heal traversal completed; authoritative storage proof is unavailable for 1 objects; heal result items were truncated", + "startTime": "2026-01-01T00:00:00Z", + "settings": { + "recursive": true, + "scanMode": 1 + }, + "items": [], + "truncated": true, + "nextSeq": 9, + "minSeq": 4, + "progress": { + "objectsScanned": 11, + "objectsHealed": 7 + }, + "outcome": { + "execution": { + "state": "completed" + }, + "coverage": "complete", + "counters": { + "processed": 1, + "healed": 0, + "unchanged": 0, + "skipped": 1, + "failed": 0, + "unknown": 1, + "attemptFailures": 0, + "overflowed": false + }, + "objects": [ + { + "identity": { + "kind": "object", + "bucket": "bucket", + "object": "object", + "versionId": null, + "bucketIncarnationId": null, + "poolIndex": null, + "setIndex": null + }, + "disposition": { + "state": "unknown" + }, + "detail": null + } + ], + "objectsTruncated": false + } + } + }, + { + "name": "completed_with_errors", + "cliExit": 1, + "response": { + "summary": "stopped", + "detail": "heal traversal completed with errors: 1 failed objects; heal result items were truncated", + "startTime": "2026-01-01T00:00:00Z", + "settings": { + "recursive": true, + "scanMode": 1 + }, + "items": [], + "truncated": true, + "nextSeq": 9, + "minSeq": 4, + "progress": { + "objectsScanned": 11, + "objectsHealed": 7 + }, + "outcome": { + "execution": { + "state": "completed_with_errors" + }, + "coverage": "complete", + "counters": { + "processed": 1, + "healed": 0, + "unchanged": 0, + "skipped": 0, + "failed": 1, + "unknown": 0, + "attemptFailures": 1, + "overflowed": false + }, + "objects": [ + { + "identity": { + "kind": "object", + "bucket": "bucket", + "object": "object", + "versionId": null, + "bucketIncarnationId": null, + "poolIndex": null, + "setIndex": null + }, + "disposition": { + "state": "failed", + "details": "retry_exhausted" + }, + "detail": null + } + ], + "objectsTruncated": false + } + } + }, + { + "name": "cancelled", + "cliExit": 1, + "response": { + "summary": "stopped", + "detail": "heal task cancelled; heal result items were truncated", + "startTime": "2026-01-01T00:00:00Z", + "settings": { + "recursive": true, + "scanMode": 1 + }, + "items": [], + "truncated": true, + "nextSeq": 9, + "minSeq": 4, + "progress": { + "objectsScanned": 11, + "objectsHealed": 7 + }, + "outcome": { + "execution": { + "state": "aborted", + "reason": "cancelled" + }, + "coverage": "partial", + "counters": { + "processed": 0, + "healed": 0, + "unchanged": 0, + "skipped": 0, + "failed": 0, + "unknown": 0, + "attemptFailures": 0, + "overflowed": false + }, + "objects": [], + "objectsTruncated": false + } + } + }, + { + "name": "deadline", + "cliExit": 1, + "response": { + "summary": "stopped", + "detail": "heal task timed out; heal result items were truncated", + "startTime": "2026-01-01T00:00:00Z", + "settings": { + "recursive": true, + "scanMode": 1 + }, + "items": [], + "truncated": true, + "nextSeq": 9, + "minSeq": 4, + "progress": { + "objectsScanned": 11, + "objectsHealed": 7 + }, + "outcome": { + "execution": { + "state": "aborted", + "reason": "deadline" + }, + "coverage": "partial", + "counters": { + "processed": 0, + "healed": 0, + "unchanged": 0, + "skipped": 0, + "failed": 0, + "unknown": 0, + "attemptFailures": 0, + "overflowed": false + }, + "objects": [], + "objectsTruncated": false + } + } + }, + { + "name": "untraversable", + "cliExit": 1, + "response": { + "summary": "stopped", + "detail": "heal listing is untraversable; heal result items were truncated", + "startTime": "2026-01-01T00:00:00Z", + "settings": { + "recursive": true, + "scanMode": 1 + }, + "items": [], + "truncated": true, + "nextSeq": 9, + "minSeq": 4, + "progress": { + "objectsScanned": 11, + "objectsHealed": 7 + }, + "outcome": { + "execution": { + "state": "aborted", + "reason": "untraversable" + }, + "coverage": "partial", + "counters": { + "processed": 0, + "healed": 0, + "unchanged": 0, + "skipped": 0, + "failed": 0, + "unknown": 0, + "attemptFailures": 0, + "overflowed": false + }, + "objects": [], + "objectsTruncated": false + } + } + }, + { + "name": "remote_completed_with_errors", + "cliExit": 1, + "response": { + "summary": "stopped", + "detail": "heal traversal completed with errors: 1 failed objects; heal result items were truncated", + "startTime": "2026-01-01T00:00:00Z", + "settings": { + "recursive": true, + "scanMode": 1 + }, + "items": [], + "truncated": true, + "nextSeq": 9, + "minSeq": 4, + "progress": { + "objectsScanned": 11, + "objectsHealed": 7 + }, + "outcome": { + "execution": { + "state": "completed_with_errors", + "futureExtension": { + "value": 7 + } + }, + "coverage": "complete", + "counters": { + "processed": 1, + "healed": 0, + "unchanged": 0, + "skipped": 0, + "failed": 1, + "unknown": 0, + "attemptFailures": 1, + "overflowed": false, + "futureCounter": 11 + }, + "objects": [ + { + "identity": { + "kind": "object", + "bucket": "bucket", + "object": "object", + "versionId": null, + "bucketIncarnationId": null, + "poolIndex": null, + "setIndex": null + }, + "disposition": { + "state": "failed", + "details": "retry_exhausted" + }, + "detail": null + } + ], + "objectsTruncated": false + } + }, + "remoteResponse": { + "summary": "finished", + "detail": "", + "startTime": "2026-01-01T00:00:00Z", + "settings": { + "recursive": true, + "scanMode": 1 + }, + "items": [], + "truncated": true, + "nextSeq": 9, + "minSeq": 4, + "progress": { + "objectsScanned": 11, + "objectsHealed": 7 + }, + "outcome": { + "execution": { + "state": "completed_with_errors", + "futureExtension": { + "value": 7 + } + }, + "coverage": "complete", + "counters": { + "processed": 1, + "healed": 0, + "unchanged": 0, + "skipped": 0, + "failed": 1, + "unknown": 0, + "attemptFailures": 1, + "overflowed": false, + "futureCounter": 11 + }, + "objects": [ + { + "identity": { + "kind": "object", + "bucket": "bucket", + "object": "object", + "versionId": null, + "bucketIncarnationId": null, + "poolIndex": null, + "setIndex": null + }, + "disposition": { + "state": "failed", + "details": "retry_exhausted" + }, + "detail": null + } + ], + "objectsTruncated": false + } + } + }, + { + "name": "remote_cancelled", + "cliExit": 1, + "response": { + "summary": "stopped", + "detail": "heal task cancelled; heal result items were truncated", + "startTime": "2026-01-01T00:00:00Z", + "settings": { + "recursive": true, + "scanMode": 1 + }, + "items": [], + "truncated": true, + "nextSeq": 9, + "minSeq": 4, + "progress": { + "objectsScanned": 11, + "objectsHealed": 7 + }, + "outcome": { + "execution": { + "state": "aborted", + "reason": "cancelled", + "futureExtension": { + "value": 7 + } + }, + "coverage": "partial", + "counters": { + "processed": 0, + "healed": 0, + "unchanged": 0, + "skipped": 0, + "failed": 0, + "unknown": 0, + "attemptFailures": 0, + "overflowed": false, + "futureCounter": 11 + }, + "objects": [], + "objectsTruncated": false + } + }, + "remoteResponse": { + "summary": "finished", + "detail": "", + "startTime": "2026-01-01T00:00:00Z", + "settings": { + "recursive": true, + "scanMode": 1 + }, + "items": [], + "truncated": true, + "nextSeq": 9, + "minSeq": 4, + "progress": { + "objectsScanned": 11, + "objectsHealed": 7 + }, + "outcome": { + "execution": { + "state": "aborted", + "reason": "cancelled", + "futureExtension": { + "value": 7 + } + }, + "coverage": "partial", + "counters": { + "processed": 0, + "healed": 0, + "unchanged": 0, + "skipped": 0, + "failed": 0, + "unknown": 0, + "attemptFailures": 0, + "overflowed": false, + "futureCounter": 11 + }, + "objects": [], + "objectsTruncated": false + } + } + } +] diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 7955a80a7..b0d8173c9 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -237,18 +237,13 @@ struct HealStartSuccess { #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct HealTaskStatus { - summary: String, + #[serde(flatten)] + payload: HealTaskStatusPayload, #[serde(rename = "detail")] failure_detail: String, start_time: String, #[serde(rename = "settings")] heal_settings: HealOpts, - #[serde(skip_serializing_if = "Vec::is_empty")] - items: Vec, - #[serde(skip_serializing_if = "std::ops::Not::not")] - truncated: bool, - #[serde(skip_serializing_if = "Option::is_none")] - progress: Option, } #[derive(Debug, Serialize)] @@ -1055,15 +1050,23 @@ async fn submit_cluster_heal_channel_command( } } -#[derive(Debug, Deserialize)] +#[derive(Debug, Default, Serialize, Deserialize)] struct HealTaskStatusPayload { + #[serde(skip)] + adapted_detail: Option, summary: String, - #[serde(default)] + #[serde(default, skip_serializing_if = "Vec::is_empty")] items: Vec, - #[serde(default)] + #[serde(default, skip_serializing_if = "std::ops::Not::not")] truncated: bool, - #[serde(default)] + #[serde(default, skip_serializing_if = "Option::is_none")] progress: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + outcome: Option, + #[serde(default, rename = "nextSeq", alias = "next_seq", skip_serializing_if = "Option::is_none")] + next_seq: Option, + #[serde(default, rename = "minSeq", alias = "min_seq", skip_serializing_if = "Option::is_none")] + min_seq: Option, } #[cfg(test)] @@ -1103,21 +1106,16 @@ fn encode_heal_start_success(client_token: String, client_address: String) -> S3 } fn encode_heal_task_status( - summary: String, + mut payload: HealTaskStatusPayload, failure_detail: String, heal_settings: HealOpts, - items: Vec, - truncated: bool, - progress: Option, ) -> S3Result> { + let failure_detail = payload.adapted_detail.take().unwrap_or(failure_detail); encode_json(&HealTaskStatus { - summary, + payload, failure_detail, start_time: current_rfc3339_time()?, heal_settings, - items, - truncated, - progress, }) } @@ -1162,42 +1160,63 @@ fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest { fn heal_channel_response_status( response: &rustfs_heal_contracts::heal_channel::HealChannelResponse, -) -> (String, Vec, bool, Option) { +) -> S3Result { let Some(data) = response.data.as_deref() else { - return ("running".to_string(), Vec::new(), false, None); + return Ok(HealTaskStatusPayload { + summary: "running".to_string(), + ..Default::default() + }); }; - if let Ok(payload) = serde_json::from_slice::(data) - && !payload.summary.is_empty() + if let Ok(mut payload) = serde_json::from_slice::(data) + && matches!(payload.summary.as_str(), "running" | "finished" | "stopped" | "notFound") { - return (payload.summary, payload.items, payload.truncated, payload.progress); + let adapted = payload + .outcome + .as_ref() + .map(|outcome| { + rustfs_heal::heal::outcome::legacy_wire_status(&payload.summary, outcome, payload.truncated) + .map(|(summary, detail)| (summary.to_string(), detail)) + }) + .transpose(); + if let Ok(adapted) = adapted { + if let Some((summary, detail)) = adapted { + payload.summary = summary; + payload.adapted_detail = detail; + } + return Ok(payload); + } } - let summary = std::str::from_utf8(data) - .ok() - .filter(|summary| !summary.is_empty()) - .unwrap_or("running") - .to_string(); - (summary, Vec::new(), false, None) + if let Ok(summary @ ("running" | "finished" | "stopped" | "notFound")) = std::str::from_utf8(data) { + return Ok(HealTaskStatusPayload { + summary: summary.to_string(), + ..Default::default() + }); + } + Err(s3s::S3Error::with_message( + s3s::S3ErrorCode::InternalError, + "invalid heal status payload or unsupported summary", + )) } #[cfg(test)] fn heal_channel_response_summary(response: &rustfs_heal_contracts::heal_channel::HealChannelResponse) -> String { - heal_channel_response_status(response).0 + heal_channel_response_status(response).expect("valid status fixture").summary } #[cfg(test)] fn heal_channel_response_items( response: &rustfs_heal_contracts::heal_channel::HealChannelResponse, ) -> Vec { - heal_channel_response_status(response).1 + heal_channel_response_status(response).expect("valid status fixture").items } #[cfg(test)] fn heal_channel_response_progress( response: &rustfs_heal_contracts::heal_channel::HealChannelResponse, ) -> Option { - heal_channel_response_status(response).3 + heal_channel_response_status(response).expect("valid status fixture").progress } fn encode_background_heal_status( @@ -1385,15 +1404,8 @@ impl Operation for HealHandler { response.error.unwrap_or_else(|| "query heal status failed".to_string()) )); } - let (summary, items, truncated, progress) = heal_channel_response_status(&response); - let body = encode_heal_task_status( - summary, - response.error.unwrap_or_default(), - HealOpts::default(), - items, - truncated, - progress, - )?; + let payload = heal_channel_response_status(&response)?; + let body = encode_heal_task_status(payload, response.error.unwrap_or_default(), HealOpts::default())?; info!( event = EVENT_ADMIN_RESPONSE_EMITTED, component = LOG_COMPONENT_ADMIN_API, @@ -1430,8 +1442,8 @@ impl Operation for HealHandler { let body = if client_token.is_empty() { encode_heal_start_success(response.request_id, client_address)? } else { - let (summary, items, truncated, progress) = heal_channel_response_status(&response); - encode_heal_task_status(summary, response.error.unwrap_or_default(), hip.hs, items, truncated, progress)? + let payload = heal_channel_response_status(&response)?; + encode_heal_task_status(payload, response.error.unwrap_or_default(), hip.hs)? }; info!( event = EVENT_ADMIN_RESPONSE_EMITTED, @@ -2761,12 +2773,12 @@ mod tests { #[test] fn test_encode_heal_task_status_uses_client_wire_shape() { let encoded = encode_heal_task_status( - "Heal status query accepted".to_string(), + super::HealTaskStatusPayload { + summary: "Heal status query accepted".to_string(), + ..Default::default() + }, String::new(), HealOpts::default(), - Vec::new(), - false, - None, ) .expect("status response should serialize"); let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize"); @@ -2783,12 +2795,13 @@ mod tests { #[test] fn test_encode_heal_task_status_reports_truncated_items() { let encoded = encode_heal_task_status( - "running".to_string(), + super::HealTaskStatusPayload { + summary: "running".to_string(), + truncated: true, + ..Default::default() + }, "heal result items were truncated".to_string(), HealOpts::default(), - Vec::new(), - true, - None, ) .expect("truncated status response should serialize"); let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize"); @@ -2804,12 +2817,13 @@ mod tests { "currentObject": "bucket-a/object-a" }); let encoded = encode_heal_task_status( - "running".to_string(), + super::HealTaskStatusPayload { + summary: "running".to_string(), + progress: Some(progress.clone()), + ..Default::default() + }, String::new(), HealOpts::default(), - Vec::new(), - false, - Some(progress.clone()), ) .expect("status response should serialize"); let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize"); @@ -2817,6 +2831,80 @@ mod tests { assert_eq!(json["progress"], progress); } + #[test] + fn outcome_v3_admin_forwards_outcome_cursors_and_progress_without_recounting() { + let cases: serde_json::Value = + serde_json::from_str(include_str!("../../../../crates/madmin/tests/fixtures/heal-outcome-v3.json")) + .expect("shared fixtures"); + for case in cases.as_array().expect("cases") { + let expected = &case["response"]; + let mut channel_payload = case.get("remoteResponse").unwrap_or(expected).clone(); + let payload = channel_payload.as_object_mut().expect("payload"); + let next_seq = payload.remove("nextSeq").expect("cursor"); + let min_seq = payload.remove("minSeq").expect("cursor"); + payload.insert("next_seq".to_string(), next_seq); + payload.insert("min_seq".to_string(), min_seq); + let response = rustfs_heal_contracts::heal_channel::HealChannelResponse { + request_id: "token".into(), + success: true, + data: Some(serde_json::to_vec(&channel_payload).expect("channel bytes")), + error: None, + }; + let payload = super::heal_channel_response_status(&response).expect("valid owner payload"); + let encoded = + encode_heal_task_status(payload, expected["detail"].as_str().expect("detail").into(), HealOpts::default()) + .expect("public response"); + let actual: serde_json::Value = serde_json::from_slice(&encoded).expect("public JSON"); + for key in ["summary", "detail", "outcome", "progress", "truncated", "nextSeq", "minSeq"] { + assert_eq!(actual[key], expected[key], "{}: {key}", case["name"]); + } + assert_eq!(actual["progress"]["objectsHealed"], 7); + assert_eq!(actual["outcome"]["counters"]["healed"], 0); + } + } + + #[test] + fn outcome_v3_rejects_corrupt_status_without_inventing_a_terminal() { + for data in [ + br#"{"summary":"future_state"}"#.as_slice(), + br#"{"summary":"finished","nextSeq":9,"next_seq":8}"#.as_slice(), + br#"{"outcome":{"execution":{"state":"completed"}}}"#.as_slice(), + b"future_state".as_slice(), + ] { + let response = rustfs_heal_contracts::heal_channel::HealChannelResponse { + request_id: "token".into(), + success: true, + data: Some(data.to_vec()), + error: None, + }; + assert!(super::heal_channel_response_status(&response).is_err()); + } + } + + #[test] + fn outcome_v3_admin_preserves_future_nonterminal_without_validating_success() { + let outcome = serde_json::json!({ + "execution": {"state": "future_execution", "extension": {"value": 7}}, + "futureCounter": 9 + }); + let mut wire = serde_json::json!({"summary": "running", "outcome": outcome, "next_seq": 9, "min_seq": 4}); + let mut response = rustfs_heal_contracts::heal_channel::HealChannelResponse { + request_id: "token".into(), + success: true, + data: Some(serde_json::to_vec(&wire).expect("future running wire")), + error: None, + }; + let payload = super::heal_channel_response_status(&response).expect("future nonterminal is opaque"); + let bytes = encode_heal_task_status(payload, String::new(), HealOpts::default()).expect("public nonterminal"); + let public: serde_json::Value = serde_json::from_slice(&bytes).expect("public JSON"); + assert_eq!(public["summary"], "running"); + assert_eq!(public["outcome"], outcome); + assert_eq!((public["nextSeq"].as_u64(), public["minSeq"].as_u64()), (Some(9), Some(4))); + wire["summary"] = serde_json::json!("finished"); + response.data = Some(serde_json::to_vec(&wire).expect("unprovable success wire")); + assert!(super::heal_channel_response_status(&response).is_err()); + } + #[test] fn test_build_heal_channel_request_preserves_safe_client_options() { let hip = HealInitParams { diff --git a/scripts/compat/heal-outcome/README.md b/scripts/compat/heal-outcome/README.md new file mode 100644 index 000000000..58b215910 --- /dev/null +++ b/scripts/compat/heal-outcome/README.md @@ -0,0 +1,33 @@ +# Legacy Heal Outcome Compatibility + +This fixture executes the real `madmin-go` HTTP decoder and a pinned `mc` binary against synthetic v3 responses. Rust owner, admin adapter, and SDK tests validate the same JSON cases in `crates/madmin/tests/fixtures/heal-outcome-v3.json`. It does not run a storage repair or prove distributed recovery. + +Pinned primary sources: + +- `mc` release `RELEASE.2025-08-13T08-35-41Z`, commit `7394ce0dd2a80935aded936b09fa12cbb3cb8096`: [polling implementation](https://github.com/minio/mc/blob/7394ce0dd2a80935aded936b09fa12cbb3cb8096/cmd/admin-heal-ui.go#L414). +- Its `madmin-go/v3` dependency is `v3.0.107-0.20250415152934-4b504b82db63`: [decoder and response type](https://github.com/minio/madmin-go/blob/4b504b82db633e978a57d49443b2be75824244c3/heal-commands.go#L101). + +The old decoder ignores additional JSON fields. The old poller returns success for `finished` without examining `detail`; only `stopped` returns a terminal error. Consequently `completed_with_errors` retains its canonical outcome and complete traversal coverage, but uses legacy summary `stopped`. `completed` describes execution only: unknown storage receipts remain `unknown`, not `repaired`. + +Run from the repository root with an isolated tool cache and binary directory: + +```sh +( +set -eu +compat_dir=$(mktemp -d) +trap 'rm -rf "$compat_dir"' EXIT +export GOPATH="$compat_dir/gopath" GOMODCACHE="$compat_dir/mod" GOCACHE="$compat_dir/cache" GOBIN="$compat_dir/bin" +export CGO_ENABLED=0 GOTOOLCHAIN=local GOMAXPROCS=2 +go install github.com/minio/mc@v0.0.0-20250813083541-7394ce0dd2a8 +cd scripts/compat/heal-outcome +MC_BINARY="$compat_dir/bin/mc" NO_PROXY=127.0.0.1,localhost go test -mod=readonly -p 2 -count=1 -v ./... +) +``` + +The subshell keeps the calling shell unchanged. `MC_BINARY` is mandatory and its Go build metadata must identify the pinned commit. Each subprocess gets a temporary mc configuration directory and synthetic credentials; it never edits the user's mc configuration. Loopback socket permission is required. The Go tests do not skip unavailable prerequisites. + +Six cases cover completed traversal, unknown repair proof, completed traversal with failures, cancellation, deadline, and untraversable listing. Two receiver cases carry a remote `finished` summary that contradicts an aborted or completed-with-errors outcome. The admin test applies the heal owner's wire validator to each `remoteResponse` and must produce the corresponding public `response`; the old CLI must then exit with an error. Unknown extension fields remain intact. Unknown or missing execution fields cannot validate a successful summary. + +New counters do not replace or reinterpret legacy progress. Outcome is a cumulative snapshot, not a page delta; `sinceSeq` only pages legacy result items. Result cursors and truncation markers remain separate from execution and traversal coverage. + +Two existing CLI limitations remain explicit: this mc does not terminate on `notFound`, and `-f` polling sends `forceStart` together with `clientToken`, a combination the RustFS v3 request contract rejects. The fixture uses the standard non-force polling flow. Neither limitation is hidden by emitting a new summary string or reporting a missing task as completed. diff --git a/scripts/compat/heal-outcome/compat_test.go b/scripts/compat/heal-outcome/compat_test.go new file mode 100644 index 000000000..8178ee27f --- /dev/null +++ b/scripts/compat/heal-outcome/compat_test.go @@ -0,0 +1,137 @@ +// Copyright 2026 RustFS Team +// Licensed under the Apache License, Version 2.0. + +package compat_test + +import ( + "context" + "debug/buildinfo" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + madmin "github.com/minio/madmin-go/v3" +) + +type fixture struct { + Name string `json:"name"` + CLIExit int `json:"cliExit"` + Response json.RawMessage `json:"response"` +} + +func fixtures(t *testing.T) []fixture { + t.Helper() + data, err := os.ReadFile("../../../crates/madmin/tests/fixtures/heal-outcome-v3.json") + if err != nil { + t.Fatal(err) + } + var cases []fixture + if err := json.Unmarshal(data, &cases); err != nil { + t.Fatal(err) + } + if len(cases) != 8 { + t.Fatalf("expected eight owner/receiver-validated fixtures, got %d", len(cases)) + } + return cases +} + +func fixtureServer(t *testing.T, response []byte, polls *atomic.Int32) *httptest.Server { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || !strings.HasPrefix(r.URL.Path, "/minio/admin/v3/heal/") { + t.Errorf("unexpected client request %s %s", r.Method, r.URL.Path) + http.Error(w, "unexpected request", http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "application/json") + if r.URL.Query().Get("clientToken") == "" { + fmt.Fprint(w, `{"clientToken":"fixture-token","clientAddress":"","startTime":"2026-01-01T00:00:00Z"}`) + return + } + polls.Add(1) + w.Write(response) + })) + t.Cleanup(server.Close) + return server +} + +func TestLegacyMadminDecoder(t *testing.T) { + for _, f := range fixtures(t) { + t.Run(f.Name, func(t *testing.T) { + var polls atomic.Int32 + server := fixtureServer(t, f.Response, &polls) + client, err := madmin.New(strings.TrimPrefix(server.URL, "http://"), "fixture-access", "fixture-secret", false) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, status, err := client.Heal(ctx, "bucket", "", madmin.HealOpts{}, "fixture-token", false, false) + if err != nil { + t.Fatal(err) + } + var expected struct { + Summary string `json:"summary"` + Detail string `json:"detail"` + } + if err := json.Unmarshal(f.Response, &expected); err != nil { + t.Fatal(err) + } + if status.Summary != expected.Summary || status.FailureDetail != expected.Detail || polls.Load() != 1 { + t.Fatalf("decoder changed legacy fields: %+v, polls=%d", status, polls.Load()) + } + }) + } +} + +func TestLegacyMCPoll(t *testing.T) { + binary := os.Getenv("MC_BINARY") + if binary == "" { + t.Fatal("MC_BINARY must point to the pinned mc release; this check cannot be skipped") + } + info, err := buildinfo.ReadFile(binary) + if err != nil { + t.Fatal(err) + } + if info.Main.Path != "github.com/minio/mc" || !strings.Contains(info.Main.Version, "7394ce0dd2a8") { + t.Fatalf("expected mc RELEASE.2025-08-13T08-35-41Z (7394ce0dd2a8), got %+v", info.Main) + } + for _, f := range fixtures(t) { + t.Run(f.Name, func(t *testing.T) { + var polls atomic.Int32 + server := fixtureServer(t, f.Response, &polls) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, binary, "--config-dir", filepath.Join(t.TempDir(), "mc"), "--json", "admin", "heal", "--recursive", "w23/bucket") + cmd.Env = append(os.Environ(), "MC_HOST_w23="+strings.Replace(server.URL, "http://", "http://fixture-access:fixture-secret@", 1), "MC_NO_COLOR=1") + output, err := cmd.CombinedOutput() + if ctx.Err() != nil { + t.Fatalf("legacy poll did not terminate: %s", output) + } + exit := 0 + if err != nil { + var ok bool + var status *exec.ExitError + status, ok = err.(*exec.ExitError) + if !ok { + t.Fatal(err) + } + exit = status.ExitCode() + } + if exit != f.CLIExit || polls.Load() != 1 { + t.Fatalf("exit=%d expected=%d polls=%d output=%s", exit, f.CLIExit, polls.Load(), output) + } + if f.CLIExit != 0 && !strings.Contains(string(output), "Heal had an error") { + t.Fatalf("failure was not the expected legacy terminal result: %s", output) + } + }) + } +} diff --git a/scripts/compat/heal-outcome/go.mod b/scripts/compat/heal-outcome/go.mod new file mode 100644 index 000000000..64e114ff9 --- /dev/null +++ b/scripts/compat/heal-outcome/go.mod @@ -0,0 +1,34 @@ +module rustfs.local/heal-outcome-compat + +go 1.24.0 + +require github.com/minio/madmin-go/v3 v3.0.107-0.20250415152934-4b504b82db63 + +require ( + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/dustin/go-humanize v1.0.1 // indirect + github.com/go-ini/ini v1.67.0 // indirect + github.com/goccy/go-json v0.10.5 // indirect + github.com/golang-jwt/jwt/v4 v4.5.2 // indirect + github.com/golang/protobuf v1.5.4 // indirect + github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect + github.com/minio/md5-simd v1.1.2 // indirect + github.com/minio/minio-go/v7 v7.0.90 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.63.0 // indirect + github.com/prometheus/procfs v0.16.0 // indirect + github.com/prometheus/prom2json v1.4.2 // indirect + github.com/prometheus/prometheus v0.303.0 // indirect + github.com/rs/xid v1.6.0 // indirect + github.com/secure-io/sio-go v0.3.1 // indirect + github.com/shirou/gopsutil/v3 v3.24.5 // indirect + github.com/shoenig/go-m1cpu v0.1.6 // indirect + github.com/tinylib/msgp v1.2.5 // indirect + github.com/tklauser/go-sysconf v0.3.15 // indirect + golang.org/x/crypto v0.37.0 // indirect + golang.org/x/net v0.39.0 // indirect + golang.org/x/sys v0.32.0 // indirect + google.golang.org/protobuf v1.36.6 // indirect +) diff --git a/scripts/compat/heal-outcome/go.sum b/scripts/compat/heal-outcome/go.sum new file mode 100644 index 000000000..7ffafd8ae --- /dev/null +++ b/scripts/compat/heal-outcome/go.sum @@ -0,0 +1,64 @@ +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/go-ini/ini v1.67.0 h1:z6ZrTEZqSWOTyH2FlglNbNgARyHG8oLW9gMELqKr06A= +github.com/go-ini/ini v1.67.0/go.mod h1:ByCAeIL28uOIIG0E3PJtZPDL8WnHpFKFOtgjp+3Ies8= +github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4= +github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M= +github.com/golang-jwt/jwt/v4 v4.5.2 h1:YtQM7lnr8iZ+j5q71MGKkNw9Mn7AjHM68uc9g5fXeUI= +github.com/golang-jwt/jwt/v4 v4.5.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0= +github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= +github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo= +github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4= +github.com/minio/madmin-go/v3 v3.0.107-0.20250415152934-4b504b82db63 h1:ktN/FrMuM9sjvjIbPZYRKeHEzBDOXQdpYUDiNO0CutE= +github.com/minio/madmin-go/v3 v3.0.107-0.20250415152934-4b504b82db63/go.mod h1:U0bL6ip4yKFwvo0keonUcWFQp0Hd462tOLLeVyPzWmE= +github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34= +github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM= +github.com/minio/minio-go/v7 v7.0.90 h1:TmSj1083wtAD0kEYTx7a5pFsv3iRYMsOJ6A4crjA1lE= +github.com/minio/minio-go/v7 v7.0.90/go.mod h1:uvMUcGrpgeSAAI6+sD3818508nUyMULw94j2Nxku/Go= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c h1:dAMKvw0MlJT1GshSTtih8C2gDs04w8dReiOGXrGLNoY= +github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.63.0 h1:YR/EIY1o3mEFP/kZCD7iDMnLPlGyuU2Gb3HIcXnA98k= +github.com/prometheus/common v0.63.0/go.mod h1:VVFF/fBIoToEnWRVkYoXEkq3R3paCoxG9PXP74SnV18= +github.com/prometheus/procfs v0.16.0 h1:xh6oHhKwnOJKMYiYBDWmkHqQPyiY40sny36Cmx2bbsM= +github.com/prometheus/procfs v0.16.0/go.mod h1:8veyXUu3nGP7oaCxhX6yeaM5u4stL2FeMXnCqhDthZg= +github.com/prometheus/prom2json v1.4.2 h1:PxCTM+Whqi/eykO1MKsEL0p/zMpxp9ybpsmdFamw6po= +github.com/prometheus/prom2json v1.4.2/go.mod h1:zuvPm7u3epZSbXPWHny6G+o8ETgu6eAK3oPr6yFkRWE= +github.com/prometheus/prometheus v0.303.0 h1:wsNNsbd4EycMCphYnTmNY9JASBVbp7NWwJna857cGpA= +github.com/prometheus/prometheus v0.303.0/go.mod h1:8PMRi+Fk1WzopMDeb0/6hbNs9nV6zgySkU/zds5Lu3o= +github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU= +github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0= +github.com/secure-io/sio-go v0.3.1 h1:dNvY9awjabXTYGsTF1PiCySl9Ltofk9GA3VdWlo7rRc= +github.com/secure-io/sio-go v0.3.1/go.mod h1:+xbkjDzPjwh4Axd07pRKSNriS9SCiYksWnZqdnfpQxs= +github.com/shirou/gopsutil/v3 v3.24.5 h1:i0t8kL+kQTvpAYToeuiVk3TgDeKOFioZO3Ztz/iZ9pI= +github.com/shirou/gopsutil/v3 v3.24.5/go.mod h1:bsoOS1aStSs9ErQ1WWfxllSeS1K5D+U30r2NfcubMVk= +github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM= +github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ= +github.com/tinylib/msgp v1.2.5 h1:WeQg1whrXRFiZusidTQqzETkRpGjFjcIhW6uqWH09po= +github.com/tinylib/msgp v1.2.5/go.mod h1:ykjzy2wzgrlvpDCRc4LA8UXy6D8bzMSuAF3WD57Gok0= +github.com/tklauser/go-sysconf v0.3.15 h1:VE89k0criAymJ/Os65CSn1IXaol+1wrsFHEB8Ol49K4= +github.com/tklauser/go-sysconf v0.3.15/go.mod h1:Dmjwr6tYFIseJw7a3dRLJfsHAMXZ3nEnL/aZY+0IuI4= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20200302210943-78000ba7a073/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE= +golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.39.0 h1:ZCu7HMWDxpXpaiKdhzIfaltL9Lp31x/3fCP11bc6/fY= +golang.org/x/net v0.39.0/go.mod h1:X7NRbYVEA+ewNkCNyJ513WmMdQ3BineSwVtN2zD/d+E= +golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20= +golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY= +google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=