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

fix(heal): preserve settings in token status (#7691)

Browse Source 
* fix(heal): preserve settings in token status

* fix(heal): group status response context

* fix(heal): decode partial status settings
This commit is contained in:
cxymds
2026-09-12 17:22:58 +08:00
committed by GitHub
parent e3ecbf5db9
commit f90d4071a5
7 changed files with 353 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/channel.rs
+108 -26
View File
@@ -21,7 +21,8 @@ use crate::heal::{
use crate::{Error, Result};
use rustfs_heal_contracts::heal_channel::{
HealAdmissionReceipt, HealAdmissionResult, HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest,
HealChannelResponse, HealReceiptCommand, HealReceiptReceiver, HealRequestSource, HealScanMode, publish_heal_response,
HealChannelResponse, HealOpts, HealReceiptCommand, HealReceiptReceiver, HealRequestSource, HealScanMode,
publish_heal_response,
};
use rustfs_madmin::heal_commands::HealResultItem;
use serde::Serialize;
@@ -78,6 +79,8 @@ struct HealTaskStatusPayload<'a> {
progress: Option<&'a HealProgress>,
#[serde(skip_serializing_if = "Option::is_none")]
outcome: Option<&'a super::outcome::HealTaskOutcome>,
#[serde(skip_serializing_if = "Option::is_none")]
settings: Option<&'a HealOpts>,
}
fn u64_is_zero(value: &u64) -> bool {
@@ -91,6 +94,7 @@ fn encode_heal_task_status_payload(
mut truncated: bool,
sequence: (u64, u64),
outcome: Option<&super::outcome::HealTaskOutcome>,
settings: Option<&HealOpts>,
) -> Result<(Vec<u8>, bool)> {
loop {
let data = serde_json::to_vec(&HealTaskStatusPayload {
@@ -101,6 +105,7 @@ fn encode_heal_task_status_payload(
min_seq: sequence.1,
progress,
outcome,
settings,
})
.map_err(|e| Error::Serialization(format!("failed to serialize heal task status: {e}")))?;
if data.len() <= MAX_HEAL_STATUS_PAYLOAD_SIZE {
@@ -114,20 +119,34 @@ fn encode_heal_task_status_payload(
}
}
fn encode_heal_status_response(
summary: &str,
items: Vec<HealResultItem>,
progress: Option<&HealProgress>,
#[derive(Default)]
struct HealStatusResponseContext<'a> {
progress: Option<&'a HealProgress>,
detail: Option<String>,
truncated: bool,
sequence: (u64, u64),
outcome: Option<&super::outcome::HealTaskOutcome>,
outcome: Option<&'a super::outcome::HealTaskOutcome>,
settings: Option<&'a HealOpts>,
}
fn encode_heal_status_response(
summary: &str,
items: Vec<HealResultItem>,
context: HealStatusResponseContext<'_>,
) -> Result<(Vec<u8>, Option<String>)> {
let (summary, detail) = match outcome {
Some(outcome) => outcome.legacy_status(summary, detail),
None => (summary, detail),
let (summary, detail) = match context.outcome {
Some(outcome) => outcome.legacy_status(summary, context.detail),
None => (summary, context.detail),
};
let (data, truncated) = encode_heal_task_status_payload(summary, items, progress, truncated, sequence, outcome)?;
let (data, truncated) = encode_heal_task_status_payload(
summary,
items,
context.progress,
context.truncated,
context.sequence,
context.outcome,
context.settings,
)?;
Ok((data, super::outcome::heal_status_detail(detail, truncated)))
}
@@ -439,6 +458,10 @@ impl HealChannelProcessor {
};
let outcome = report.as_ref().ok().and_then(|report| report.outcome.clone());
let settings = report
.as_ref()
.ok()
.and_then(|report| report.options.as_ref().map(heal_options_to_wire));
let (summary, detail, items, truncated, progress, next_seq, min_seq) = match report {
Ok(HealTaskReport {
status: HealTaskStatus::Pending | HealTaskStatus::Running,
@@ -579,11 +602,14 @@ impl HealChannelProcessor {
let (data, detail) = encode_heal_status_response(
&summary,
items,
progress.as_ref(),
detail,
truncated,
(next_seq, min_seq),
outcome.as_deref(),
HealStatusResponseContext {
progress: progress.as_ref(),
detail,
truncated,
sequence: (next_seq, min_seq),
outcome: outcome.as_deref(),
settings: settings.as_ref(),
},
)?;
let response = HealChannelResponse {
@@ -778,6 +804,21 @@ impl HealChannelProcessor {
}
}
fn heal_options_to_wire(options: &HealOptions) -> HealOpts {
HealOpts {
recursive: options.recursive,
dry_run: options.dry_run,
remove: options.remove_corrupted,
recreate: options.recreate_missing,
scan_mode: options.scan_mode,
update_parity: options.update_parity,
no_lock: options.no_lock,
read_repair: false,
pool: options.pool_index,
set: options.set_index,
}
}
#[cfg(test)]
mod tests {
use super::super::DiskStore;
@@ -873,7 +914,7 @@ mod tests {
..Default::default()
}];
let (data, detail) = encode_heal_status_response("running", items, None, None, false, (0, 0), None).unwrap();
let (data, detail) = encode_heal_status_response("running", items, HealStatusResponseContext::default()).unwrap();
assert!(data.len() <= MAX_HEAL_STATUS_PAYLOAD_SIZE);
let payload: serde_json::Value = serde_json::from_slice(&data).unwrap();
@@ -933,11 +974,13 @@ mod tests {
let (bytes, detail) = encode_heal_status_response(
if abort.is_some() { "stopped" } else { "finished" },
Vec::new(),
None,
initial_detail,
true,
(9, 4),
Some(&outcome),
HealStatusResponseContext {
detail: initial_detail,
truncated: true,
sequence: (9, 4),
outcome: Some(&outcome),
..Default::default()
},
)
.expect("canonical owner encoding");
let decoded: serde_json::Value = serde_json::from_slice(&bytes).expect("wire payload");
@@ -960,8 +1003,15 @@ mod tests {
] {
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 (data, detail) = encode_heal_status_response(
"finished",
Vec::new(),
HealStatusResponseContext {
outcome: Some(&outcome),
..Default::default()
},
)
.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");
@@ -997,8 +1047,16 @@ mod tests {
..Default::default()
},
];
let (bytes, detail) = encode_heal_status_response("running", items, None, None, false, (9, 4), Some(&outcome))
.expect("bounded status with cumulative outcome");
let (bytes, detail) = encode_heal_status_response(
"running",
items,
HealStatusResponseContext {
sequence: (9, 4),
outcome: Some(&outcome),
..Default::default()
},
)
.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);
@@ -1881,7 +1939,23 @@ mod tests {
#[tokio::test]
async fn test_process_query_request_reports_running_for_queued_task() {
let heal_manager = create_test_heal_manager();
let request = HealRequest::bucket("bucket".to_string());
let request = HealRequest::new(
HealType::Bucket {
bucket: "bucket".to_string(),
},
HealOptions {
scan_mode: HealScanMode::Deep,
remove_corrupted: true,
recreate_missing: false,
update_parity: false,
recursive: true,
dry_run: true,
pool_index: Some(1),
set_index: Some(2),
..Default::default()
},
HealPriority::High,
);
let task_id = request.id.clone();
assert_eq!(
heal_manager
@@ -1910,6 +1984,14 @@ mod tests {
.expect("status payload should be json");
assert_eq!(payload["summary"], "running");
assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0);
assert_eq!(payload["settings"]["scanMode"], 2);
assert_eq!(payload["settings"]["dryRun"], true);
assert_eq!(payload["settings"]["remove"], true);
assert_eq!(payload["settings"]["recreate"], false);
assert_eq!(payload["settings"]["updateParity"], false);
assert_eq!(payload["settings"]["recursive"], true);
assert_eq!(payload["settings"]["pool"], 1);
assert_eq!(payload["settings"]["set"], 2);
}
#[tokio::test]
crates/heal/src/heal/manager.rs
+44 -23
View File
@@ -197,6 +197,7 @@ fn record_displaced_terminal(
progress: None,
retained_bytes: std::sync::OnceLock::new(),
heal_type: request.heal_type.clone(),
options: request.options.clone(),
status: HealTaskStatus::Failed {
error: format!("heal task displaced by a higher-priority request ({DISPLACED_HEAL_REASON})"),
},
@@ -277,6 +278,8 @@ async fn publish_completed_heal(
#[derive(Debug, Clone)]
pub struct HealTaskReport {
/// Options used by the task, when retained by the state source.
pub options: Option<HealOptions>,
pub outcome: Option<Arc<HealTaskOutcome>>,
pub status: HealTaskStatus,
pub result_items: Vec<HealResultItem>,
@@ -294,6 +297,7 @@ pub struct HealTaskReport {
async fn active_task_report(task: &HealTask, since: Option<u64>) -> HealTaskReport {
let window = task.get_result_items_since(since).await;
HealTaskReport {
options: Some(task.options.clone()),
status: task.get_status().await,
outcome: Some(Arc::new(task.get_outcome().await)),
result_items: window.items,
@@ -309,6 +313,7 @@ async fn active_task_report(task: &HealTask, since: Option<u64>) -> HealTaskRepo
fn empty_task_report(status: HealTaskStatus) -> HealTaskReport {
HealTaskReport {
options: None,
outcome: None,
status,
result_items: Vec::new(),
@@ -319,6 +324,13 @@ fn empty_task_report(status: HealTaskStatus) -> HealTaskReport {
}
}
fn empty_task_report_with_options(status: HealTaskStatus, options: HealOptions) -> HealTaskReport {
HealTaskReport {
options: Some(options),
..empty_task_report(status)
}
}
fn completed_task_report(completed: &CompletedHealStatus, since: Option<u64>) -> HealTaskReport {
let mut lagged = false;
let result_items = match since {
@@ -336,6 +348,7 @@ fn completed_task_report(completed: &CompletedHealStatus, since: Option<u64>) ->
}
};
HealTaskReport {
options: Some(completed.options.clone()),
status: completed.status.clone(),
outcome: completed.outcome.clone(),
result_items,
@@ -866,9 +879,9 @@ pub struct HealManager {
/// cascade without re-locking.
enum TaskStateLookup {
Active(Arc<HealTask>),
Retrying(HealTaskStatus),
Retrying(HealTaskStatus, HealOptions),
Completed(Arc<CompletedHealStatus>),
Queued,
Queued(HealOptions),
NotFound,
}
@@ -2131,7 +2144,7 @@ impl HealManager {
.get(canonical_task_id)
.filter(|retrying| matches_path(&retrying.request.heal_type))
{
return Ok(TaskStateLookup::Retrying(retrying.status()));
return Ok(TaskStateLookup::Retrying(retrying.status(), retrying.request.options.clone()));
}
}
@@ -2152,12 +2165,8 @@ impl HealManager {
{
let queue = self.heal_queue.lock().await;
let queued = match heal_path {
Some(path) => queue.contains_request_id_matching_path(canonical_task_id, path),
None => queue.contains_request_id(canonical_task_id),
};
if queued {
return Ok(TaskStateLookup::Queued);
if let Some(request) = queue.request_matching_id_and_path(canonical_task_id, heal_path) {
return Ok(TaskStateLookup::Queued(request.options.clone()));
}
}
@@ -2200,12 +2209,14 @@ impl HealManager {
task_id: &str,
heal_type: &HealType,
source: HealRequestSource,
options: &HealOptions,
) -> Result<bool> {
let completed = CompletedHealStatus {
outcome: None,
progress: None,
retained_bytes: std::sync::OnceLock::new(),
heal_type: heal_type.clone(),
options: options.clone(),
status: HealTaskStatus::Cancelled,
result_items_truncated: false,
completed_at: SystemTime::now(),
@@ -2220,9 +2231,9 @@ impl HealManager {
let canonical_task_id = self.canonical_task_id(task_id).await;
match self.lookup_task_state(&canonical_task_id, None).await? {
TaskStateLookup::Active(task) => Ok(task.get_status().await),
TaskStateLookup::Retrying(status) => Ok(status),
TaskStateLookup::Retrying(status, _) => Ok(status),
TaskStateLookup::Completed(completed) => Ok(completed.status.clone()),
TaskStateLookup::Queued => Ok(HealTaskStatus::Pending),
TaskStateLookup::Queued(_) => Ok(HealTaskStatus::Pending),
TaskStateLookup::NotFound => Err(Error::TaskNotFound {
task_id: task_id.to_string(),
}),
@@ -2240,9 +2251,9 @@ impl HealManager {
let canonical_task_id = self.canonical_task_id(task_id).await;
match self.lookup_task_state(&canonical_task_id, None).await? {
TaskStateLookup::Active(task) => Ok(active_task_report(&task, since).await),
TaskStateLookup::Retrying(status) => Ok(empty_task_report(status)),
TaskStateLookup::Retrying(status, options) => Ok(empty_task_report_with_options(status, options)),
TaskStateLookup::Completed(completed) => Ok(completed_task_report(&completed, since)),
TaskStateLookup::Queued => Ok(empty_task_report(HealTaskStatus::Pending)),
TaskStateLookup::Queued(options) => Ok(empty_task_report_with_options(HealTaskStatus::Pending, options)),
TaskStateLookup::NotFound => Err(Error::TaskNotFound {
task_id: task_id.to_string(),
}),
@@ -2263,9 +2274,9 @@ impl HealManager {
let canonical_task_id = self.canonical_task_id(task_id).await;
match self.lookup_task_state(&canonical_task_id, Some(heal_path)).await? {
TaskStateLookup::Active(task) => Ok(active_task_report(&task, since).await),
TaskStateLookup::Retrying(status) => Ok(empty_task_report(status)),
TaskStateLookup::Retrying(status, options) => Ok(empty_task_report_with_options(status, options)),
TaskStateLookup::Completed(completed) => Ok(completed_task_report(&completed, since)),
TaskStateLookup::Queued => Ok(empty_task_report(HealTaskStatus::Pending)),
TaskStateLookup::Queued(options) => Ok(empty_task_report_with_options(HealTaskStatus::Pending, options)),
TaskStateLookup::NotFound => {
if self.path_has_task(heal_path).await {
return Err(Error::InvalidClientToken);
@@ -2286,9 +2297,9 @@ impl HealManager {
let canonical_task_id = self.canonical_task_id(task_id).await;
match self.lookup_task_state(&canonical_task_id, Some(heal_path)).await? {
TaskStateLookup::Active(task) => Ok(task.get_status().await),
TaskStateLookup::Retrying(status) => Ok(status),
TaskStateLookup::Retrying(status, _) => Ok(status),
TaskStateLookup::Completed(completed) => Ok(completed.status.clone()),
TaskStateLookup::Queued => Ok(HealTaskStatus::Pending),
TaskStateLookup::Queued(_) => Ok(HealTaskStatus::Pending),
TaskStateLookup::NotFound => {
if self.path_has_task(heal_path).await {
return Err(Error::InvalidClientToken);
@@ -2404,8 +2415,13 @@ impl HealManager {
{
let mut retrying_heals = self.retrying_heals.lock().await;
if let Some(retrying) = retrying_heals.get(&canonical_task_id) {
self.publish_admin_cancelled_terminal(&canonical_task_id, &retrying.request.heal_type, retrying.request.source)
.await?;
self.publish_admin_cancelled_terminal(
&canonical_task_id,
&retrying.request.heal_type,
retrying.request.source,
&retrying.request.options,
)
.await?;
self.root_recovery
.remove(&canonical_task_id, &retrying.request.heal_type, retrying.request.source)
.await?;
@@ -2431,7 +2447,7 @@ impl HealManager {
let mut queue = self.heal_queue.lock().await;
if let Some(request) = queue.requests().find(|request| request.id == canonical_task_id) {
self.publish_admin_cancelled_terminal(&canonical_task_id, &request.heal_type, request.source)
self.publish_admin_cancelled_terminal(&canonical_task_id, &request.heal_type, request.source, &request.options)
.await?;
self.root_recovery
.remove(&request.id, &request.heal_type, request.source)
@@ -2508,8 +2524,13 @@ impl HealManager {
for task_id in &task_ids {
if let Some(retrying) = retrying_heals.get(task_id) {
self.publish_admin_cancelled_terminal(task_id, &retrying.request.heal_type, retrying.request.source)
.await?;
self.publish_admin_cancelled_terminal(
task_id,
&retrying.request.heal_type,
retrying.request.source,
&retrying.request.options,
)
.await?;
self.root_recovery
.remove(task_id, &retrying.request.heal_type, retrying.request.source)
.await?;
@@ -2544,7 +2565,7 @@ impl HealManager {
.collect::<Vec<_>>()
};
for request in &queued_matches {
self.publish_admin_cancelled_terminal(&request.id, &request.heal_type, request.source)
self.publish_admin_cancelled_terminal(&request.id, &request.heal_type, request.source, &request.options)
.await?;
self.root_recovery
.remove(&request.id, &request.heal_type, request.source)
crates/heal/src/heal/manager/queue.rs
+8 -4
View File
@@ -81,6 +81,8 @@ pub(super) enum QueuePushOutcome {
#[derive(Debug, Clone)]
pub(super) struct CompletedHealStatus {
pub(super) heal_type: HealType,
/// Options used to execute the task, retained for token-scoped status.
pub(super) options: HealOptions,
pub(super) status: HealTaskStatus,
pub(super) progress: Option<HealProgress>,
pub(super) outcome: Option<Arc<HealTaskOutcome>>,
@@ -209,6 +211,7 @@ impl CompletedHealStatus {
let (next_seq, min_seq) = task.result_seq_cursors();
let mut snapshot = Self {
heal_type: task.heal_type.clone(),
options: task.options.clone(),
status,
progress: Some(task.get_progress().await),
outcome: Some(Arc::new(task.get_outcome().await)),
@@ -491,14 +494,15 @@ impl PriorityHealQueue {
self.heap.iter().map(|item| &item.request)
}
#[cfg(test)]
pub(super) fn contains_request_id(&self, request_id: &str) -> bool {
self.heap.iter().any(|item| item.request.id == request_id)
}
pub(super) fn contains_request_id_matching_path(&self, request_id: &str, heal_path: &str) -> bool {
self.heap
.iter()
.any(|item| item.request.id == request_id && heal_type_matches_path(&item.request.heal_type, heal_path))
pub(super) fn request_matching_id_and_path(&self, request_id: &str, heal_path: Option<&str>) -> Option<&HealRequest> {
self.heap.iter().map(|item| &item.request).find(|request| {
request.id == request_id && heal_path.is_none_or(|path| heal_type_matches_path(&request.heal_type, path))
})
}
pub(super) fn queued_request_id_for_dedup_key(&self, key: &str) -> Option<&str> {
crates/heal/src/heal/manager/root_recovery.rs
+7 -2
View File
@@ -208,6 +208,8 @@ struct RootHealTerminal {
task_id: String,
heal_type: RecoveryHealType,
status: HealTaskStatus,
#[serde(default, deserialize_with = "decode_options")]
options: HealOptions,
progress: Option<HealProgress>,
completed_at: SystemTime,
}
@@ -219,17 +221,19 @@ impl RootHealTerminal {
task_id: task_id.to_owned(),
heal_type: RecoveryHealType::from(&completed.heal_type),
status: completed.status.clone(),
options: completed.options.clone(),
progress: completed.progress.clone(),
completed_at: completed.completed_at,
}
}
fn cancelled(task_id: &str, heal_type: &HealType) -> Self {
fn cancelled(task_id: &str, heal_type: &HealType, options: HealOptions) -> Self {
Self {
schema: ROOT_TERMINAL_SCHEMA,
task_id: task_id.to_owned(),
heal_type: RecoveryHealType::from(heal_type),
status: HealTaskStatus::Cancelled,
options,
progress: None,
completed_at: SystemTime::now(),
}
@@ -241,6 +245,7 @@ impl RootHealTerminal {
progress: self.progress,
retained_bytes: std::sync::OnceLock::new(),
heal_type: self.heal_type.into(),
options: self.options,
status: self.status,
result_items_truncated: false,
completed_at: self.completed_at,
@@ -676,7 +681,7 @@ impl RootHealRecovery {
};
let pending = decode_intent(task_id, &bytes)?;
let heal_type = HealType::from(pending.heal_type);
let terminal = RootHealTerminal::cancelled(task_id, &heal_type);
let terminal = RootHealTerminal::cancelled(task_id, &heal_type, pending.options);
let _ = Self::persist_terminal_locked(&disks, task_id, terminal).await?;
match EcstoreDiskAPI::compare_and_update_file(
disk.as_ref(),
crates/heal/src/heal/manager/tests.rs
+41 -1
View File
@@ -112,6 +112,7 @@ fn completed_retention_fixture(completed_at: SystemTime) -> CompletedHealStatus
CompletedHealStatus {
outcome: None,
heal_type: HealType::Cluster,
options: HealOptions::default(),
status: HealTaskStatus::Completed,
progress: Some(HealProgress {
objects_scanned: 9,
@@ -2697,6 +2698,7 @@ async fn insert_retrying_request(manager: &HealManager, request: HealRequest) ->
outcome: None,
retained_bytes: std::sync::OnceLock::new(),
heal_type: request.heal_type,
options: request.options.clone(),
status: HealTaskStatus::Retrying {
error: "Lock acquisition timeout".to_string(),
retry_attempt: request.retry_attempts,
@@ -3468,12 +3470,19 @@ async fn test_get_task_report_queries_queued_task_by_token_without_path() {
let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
let manager = HealManager::new_without_root_recovery_for_test(storage, None);
let options = HealOptions {
scan_mode: rustfs_heal_contracts::heal_channel::HealScanMode::Deep,
dry_run: true,
remove_corrupted: true,
recreate_missing: false,
..Default::default()
};
let request = HealRequest::new(
HealType::ErasureSet {
buckets: vec![],
set_disk_id: "pool_0_set_1".to_string(),
},
HealOptions::default(),
options.clone(),
HealPriority::High,
);
let request_id = request.id.clone();
@@ -3489,9 +3498,37 @@ async fn test_get_task_report_queries_queued_task_by_token_without_path() {
.expect("queued task should be queryable by token");
assert_eq!(report.status, HealTaskStatus::Pending);
assert_eq!(report.options, Some(options));
assert!(report.result_items.is_empty());
}
#[tokio::test]
async fn test_get_task_report_preserves_retrying_options() {
let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
let manager = HealManager::new_without_root_recovery_for_test(storage, None);
let options = HealOptions {
scan_mode: rustfs_heal_contracts::heal_channel::HealScanMode::Deep,
dry_run: true,
recreate_missing: false,
..Default::default()
};
let mut request = HealRequest::bucket("bucket-retrying-options".to_string());
request.options = options.clone();
let task_id = request.id.clone();
manager.retrying_heals.lock().await.insert(
task_id.clone(),
RetryingHeal {
request,
error: "transient".to_string(),
cancel_token: CancellationToken::new(),
},
);
let report = manager.get_task_report(&task_id).await.expect("retrying task report");
assert!(matches!(report.status, HealTaskStatus::Retrying { .. }));
assert_eq!(report.options, Some(options));
}
#[tokio::test]
async fn test_retrying_completion_outranks_the_queue_for_the_same_id() {
let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
@@ -3510,6 +3547,7 @@ async fn test_retrying_completion_outranks_the_queue_for_the_same_id() {
outcome: None,
retained_bytes: std::sync::OnceLock::new(),
heal_type: request.heal_type.clone(),
options: request.options.clone(),
status: HealTaskStatus::Retrying {
error: "transient disk failure".to_string(),
retry_attempt: 1,
@@ -3550,6 +3588,7 @@ async fn test_get_task_status_reads_recent_completed_status() {
heal_type: HealType::Bucket {
bucket: "bucket".to_string(),
},
options: HealOptions::default(),
status: HealTaskStatus::Completed,
result_items_truncated: false,
seqed_items: Vec::new(),
@@ -3584,6 +3623,7 @@ async fn test_get_task_report_for_path_reads_completed_items() {
object: "object".to_string(),
version_id: None,
},
options: HealOptions::default(),
status: HealTaskStatus::Completed,
result_items_truncated: true,
seqed_items: vec![(
crates/heal/src/heal/manager/tests/root_recovery.rs
+62 -3
View File
@@ -100,6 +100,7 @@ fn completed_admin_status(heal_type: &HealType, completed_at: SystemTime) -> Com
CompletedHealStatus {
outcome: None,
heal_type: heal_type.clone(),
options: HealOptions::default(),
status: HealTaskStatus::Completed,
progress: Some(HealProgress {
objects_scanned: 1,
@@ -381,9 +382,12 @@ async fn root_recovery_non_admin_request_is_not_persisted() {
async fn root_recovery_path_cancel_covers_durable_only_non_root_record() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
let request = admin_request(HealType::Bucket {
let mut request = admin_request(HealType::Bucket {
bucket: "bucket".to_string(),
});
request.options.scan_mode = rustfs_heal_contracts::heal_channel::HealScanMode::Deep;
request.options.dry_run = true;
request.options.recreate_missing = false;
manager
.root_recovery
.persist(&request)
@@ -418,6 +422,14 @@ async fn root_recovery_path_cancel_covers_durable_only_non_root_record() {
.expect("durable cancellation remains queryable by id"),
HealTaskStatus::Cancelled
);
assert_eq!(
restarted
.get_task_report(&request.id)
.await
.expect("durable cancellation report")
.options,
Some(request.options)
);
}
#[tokio::test]
@@ -480,7 +492,7 @@ async fn root_recovery_terminal_receipt_wins_over_stale_pending_scoped_intent_af
.await
.expect("durable bucket responsibility");
manager
.publish_admin_cancelled_terminal(&request.id, &request.heal_type, request.source)
.publish_admin_cancelled_terminal(&request.id, &request.heal_type, request.source, &request.options)
.await
.expect("publish terminal receipt");
manager
@@ -517,9 +529,12 @@ async fn root_recovery_terminal_receipt_wins_over_stale_pending_scoped_intent_af
async fn root_recovery_completed_non_root_admin_is_queryable_after_restart() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
let request = admin_request(HealType::Bucket {
let mut request = admin_request(HealType::Bucket {
bucket: "bucket".to_string(),
});
request.options.scan_mode = rustfs_heal_contracts::heal_channel::HealScanMode::Deep;
request.options.dry_run = true;
request.options.recreate_missing = false;
manager
.root_recovery
.persist(&request)
@@ -528,6 +543,7 @@ async fn root_recovery_completed_non_root_admin_is_queryable_after_restart() {
let completed = CompletedHealStatus {
outcome: None,
heal_type: request.heal_type.clone(),
options: request.options.clone(),
status: HealTaskStatus::Completed,
progress: Some(HealProgress {
objects_scanned: 2,
@@ -574,6 +590,49 @@ async fn root_recovery_completed_non_root_admin_is_queryable_after_restart() {
.expect("completed terminal exposes progress");
assert_eq!(progress.objects_scanned, 2);
assert_eq!(progress.objects_healed, 2);
assert_eq!(
restarted
.get_task_report(&request.id)
.await
.expect("completed terminal report")
.options,
Some(request.options)
);
}
#[tokio::test]
async fn root_recovery_legacy_terminal_without_options_uses_defaults() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
let request = admin_request(HealType::Bucket {
bucket: "legacy-bucket".to_string(),
});
let completed = completed_admin_status(&request.heal_type, SystemTime::now());
manager
.publish_admin_terminal(&request.id, &request.heal_type, request.source, &completed)
.await
.expect("publish terminal receipt");
let path = format!("terminal-root-heal-{}.json", request.id);
let bytes = disk.read_all(RUSTFS_META_BUCKET, &path).await.expect("read terminal receipt");
let mut value: serde_json::Value = serde_json::from_slice(&bytes).expect("decode terminal receipt");
value.as_object_mut().expect("terminal object").remove("options");
disk.write_all(
RUSTFS_META_BUCKET,
&path,
serde_json::to_vec(&value).expect("encode legacy receipt").into(),
)
.await
.expect("write legacy terminal receipt");
drop(manager);
let restarted = recovery_manager(vec![disk]);
let report = restarted
.get_task_report(&request.id)
.await
.expect("legacy terminal remains queryable");
assert_eq!(report.status, HealTaskStatus::Completed);
assert_eq!(report.options, Some(HealOptions::default()));
}
#[tokio::test]
rustfs/src/admin/handlers/heal.rs
+83 -2
View File
@@ -242,6 +242,40 @@ struct HealStartSuccess {
start_time: String,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", default)]
struct HealStatusSettings {
recursive: bool,
dry_run: bool,
remove: bool,
recreate: bool,
scan_mode: HealScanMode,
update_parity: bool,
#[serde(rename = "nolock")]
no_lock: bool,
#[serde(rename = "readRepair")]
read_repair: bool,
pool: Option<usize>,
set: Option<usize>,
}
impl From<HealOpts> for HealStatusSettings {
fn from(settings: HealOpts) -> Self {
Self {
recursive: settings.recursive,
dry_run: settings.dry_run,
remove: settings.remove,
recreate: settings.recreate,
scan_mode: settings.scan_mode,
update_parity: settings.update_parity,
no_lock: settings.no_lock,
read_repair: settings.read_repair,
pool: settings.pool,
set: settings.set,
}
}
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct HealTaskStatus {
@@ -251,7 +285,7 @@ struct HealTaskStatus {
failure_detail: String,
start_time: String,
#[serde(rename = "settings")]
heal_settings: HealOpts,
heal_settings: HealStatusSettings,
}
#[derive(Debug, Serialize)]
@@ -1076,6 +1110,8 @@ async fn submit_cluster_heal_channel_command(
struct HealTaskStatusPayload {
#[serde(skip)]
adapted_detail: Option<String>,
#[serde(default, rename = "settings", skip_serializing)]
heal_settings: Option<HealStatusSettings>,
summary: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
items: Vec<rustfs_madmin::heal_commands::HealResultItem>,
@@ -1130,9 +1166,10 @@ fn encode_heal_start_success(client_token: String, client_address: String) -> S3
fn encode_heal_task_status(
mut payload: HealTaskStatusPayload,
failure_detail: String,
heal_settings: HealOpts,
fallback_heal_settings: HealOpts,
) -> S3Result<Vec<u8>> {
let failure_detail = payload.adapted_detail.take().unwrap_or(failure_detail);
let heal_settings = payload.heal_settings.take().unwrap_or_else(|| fallback_heal_settings.into());
encode_json(&HealTaskStatus {
payload,
failure_detail,
@@ -2959,6 +2996,50 @@ mod tests {
OffsetDateTime::parse(start_time, &Rfc3339).expect("startTime should be RFC3339");
}
#[test]
fn test_encode_heal_task_status_uses_settings_from_channel_payload() {
let response = rustfs_heal_contracts::heal_channel::HealChannelResponse {
request_id: "token".into(),
success: true,
data: Some(
br#"{"summary":"running","settings":{"recursive":true,"dryRun":true,"remove":true,"recreate":false,"scanMode":2,"updateParity":false,"nolock":false,"readRepair":false,"pool":1,"set":2}}"#
.to_vec(),
),
error: None,
};
let payload = super::heal_channel_response_status(&response).expect("channel status should decode");
let encoded =
encode_heal_task_status(payload, String::new(), HealOpts::default()).expect("public status should serialize");
let json: serde_json::Value = serde_json::from_slice(&encoded).expect("public status should decode");
assert_eq!(json["settings"]["scanMode"], 2);
assert_eq!(json["settings"]["dryRun"], true);
assert_eq!(json["settings"]["remove"], true);
assert_eq!(json["settings"]["recreate"], false);
assert_eq!(json["settings"]["updateParity"], false);
assert_eq!(json["settings"]["recursive"], true);
assert_eq!(json["settings"]["pool"], 1);
assert_eq!(json["settings"]["set"], 2);
}
#[test]
fn test_encode_heal_task_status_defaults_settings_for_legacy_channel_payload() {
let response = rustfs_heal_contracts::heal_channel::HealChannelResponse {
request_id: "token".into(),
success: true,
data: Some(br#"{"summary":"running"}"#.to_vec()),
error: None,
};
let payload = super::heal_channel_response_status(&response).expect("legacy channel status should decode");
let encoded =
encode_heal_task_status(payload, String::new(), HealOpts::default()).expect("legacy public status should serialize");
let json: serde_json::Value = serde_json::from_slice(&encoded).expect("public status should decode");
assert_eq!(json["settings"]["scanMode"], 1);
assert_eq!(json["settings"]["dryRun"], false);
assert_eq!(json["settings"]["remove"], false);
}
#[test]
fn test_encode_heal_task_status_reports_truncated_items() {
let encoded = encode_heal_task_status(