gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-06 12:09:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(heal): expose compatible canonical v3 outcomes (#7256)

Browse Source 
* feat(heal): expose compatible canonical v3 outcomes

Serialize the existing canonical heal outcome, retain the v3 summary vocabulary, and reject or conservatively adapt contradictory peer success responses. Preserve progress and bounded result cursors without treating legacy storage responses as repair proof.

Add optional SDK outcome and cursor support with shared Rust fixtures, pinned legacy Go decoder and mc polling checks, and explicit compatibility limits.

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

* fix(madmin): box heal stop task status outcome

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

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-06 16:46:42 +08:00
committed by GitHub
parent 0ee5408b94
commit 5b962b6c58
9 changed files with 1300 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/channel.rs
+155 -19
View File
@@ -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<HealResultItem>,
progress: Option<&HealProgress>,
mut truncated: bool,
next_seq: u64,
min_seq: u64,
sequence: (u64, u64),
outcome: Option<&super::outcome::HealTaskOutcome>,
) -> Result<(Vec<u8>, 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<String>, truncated: bool) -> Option<String> {
if !truncated {
return detail;
}
let truncation = "heal result items were truncated";
Some(detail.map_or_else(|| truncation.to_string(), |detail| format!("{detail}; {truncation}")))
}
fn encode_heal_status_response(
summary: &str,
items: Vec<HealResultItem>,
progress: Option<&HealProgress>,
detail: Option<String>,
truncated: bool,
next_seq: u64,
min_seq: u64,
sequence: (u64, u64),
outcome: Option<&super::outcome::HealTaskOutcome>,
) -> Result<(Vec<u8>, Option<String>)> {
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 = [
crates/heal/src/heal/outcome.rs
+179 -11
View File
@@ -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<usize>,
}
#[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<HealObjectOutcome>,
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<String>), 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<String>, truncated: bool) -> Option<String> {
if !truncated {
return detail;
}
let truncation = "heal result items were truncated";
Some(detail.map_or_else(|| truncation.to_string(), |detail| format!("{detail}; {truncation}")))
}
fn legacy_execution_status<'a>(
summary: &'a str,
detail: Option<String>,
execution: HealExecutionOutcome,
counters: &HealOutcomeCounters,
) -> (&'a str, Option<String>) {
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<String>) -> (&'a str, Option<String>) {
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();
crates/madmin/src/client.rs
+82 -4
View File
@@ -167,6 +167,13 @@ pub struct HealTaskStatus {
/// Live progress snapshot; the exact shape is owned by the heal runtime.
#[serde(default)]
pub progress: Option<serde_json::Value>,
/// Canonical heal-owner result. Missing or future states are not repair proof.
#[serde(default)]
pub outcome: Option<serde_json::Value>,
#[serde(default, alias = "next_seq")]
pub next_seq: Option<u64>,
#[serde(default, alias = "min_seq")]
pub min_seq: Option<u64>,
}
/// `POST /v3/background-heal/status` response. Known top-level fields are
@@ -358,8 +365,22 @@ impl AdminClient {
prefix: Option<&str>,
client_token: &str,
) -> Result<HealTaskStatus, AdminClientError> {
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<u64>,
) -> Result<HealTaskStatus, AdminClientError> {
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<HealTaskStatus>),
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}"#;
crates/madmin/tests/fixtures/heal-outcome-v3.json
Vendored
+472
View File
@@ -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
}
}
}
]
rustfs/src/admin/handlers/heal.rs
+144 -56
View File
@@ -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<rustfs_madmin::heal_commands::HealResultItem>,
#[serde(skip_serializing_if = "std::ops::Not::not")]
truncated: bool,
#[serde(skip_serializing_if = "Option::is_none")]
progress: Option<serde_json::Value>,
}
#[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<String>,
summary: String,
#[serde(default)]
#[serde(default, skip_serializing_if = "Vec::is_empty")]
items: Vec<rustfs_madmin::heal_commands::HealResultItem>,
#[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_json::Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
outcome: Option<serde_json::Value>,
#[serde(default, rename = "nextSeq", alias = "next_seq", skip_serializing_if = "Option::is_none")]
next_seq: Option<u64>,
#[serde(default, rename = "minSeq", alias = "min_seq", skip_serializing_if = "Option::is_none")]
min_seq: Option<u64>,
}
#[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<rustfs_madmin::heal_commands::HealResultItem>,
truncated: bool,
progress: Option<serde_json::Value>,
) -> S3Result<Vec<u8>> {
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<rustfs_madmin::heal_commands::HealResultItem>, bool, Option<serde_json::Value>) {
) -> S3Result<HealTaskStatusPayload> {
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::<HealTaskStatusPayload>(data)
&& !payload.summary.is_empty()
if let Ok(mut payload) = serde_json::from_slice::<HealTaskStatusPayload>(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<rustfs_madmin::heal_commands::HealResultItem> {
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<serde_json::Value> {
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 {
scripts/compat/heal-outcome/README.md
+33
View File
@@ -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.
scripts/compat/heal-outcome/compat_test.go
+137
View File
@@ -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)
}
})
}
}
scripts/compat/heal-outcome/go.mod
+34
View File
@@ -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
)
scripts/compat/heal-outcome/go.sum
+64
View File
@@ -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=