diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index b806990a2..4681d4a44 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -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, 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, - progress: Option<&HealProgress>, +#[derive(Default)] +struct HealStatusResponseContext<'a> { + progress: Option<&'a HealProgress>, detail: Option, 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, + context: HealStatusResponseContext<'_>, ) -> Result<(Vec, Option)> { - 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] diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index b269c446c..4540aec88 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -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, pub outcome: Option>, pub status: HealTaskStatus, pub result_items: Vec, @@ -294,6 +297,7 @@ pub struct HealTaskReport { async fn active_task_report(task: &HealTask, since: Option) -> 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) -> 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) -> HealTaskReport { let mut lagged = false; let result_items = match since { @@ -336,6 +348,7 @@ fn completed_task_report(completed: &CompletedHealStatus, since: Option) -> } }; 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), - Retrying(HealTaskStatus), + Retrying(HealTaskStatus, HealOptions), Completed(Arc), - 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 { 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::>() }; 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) diff --git a/crates/heal/src/heal/manager/queue.rs b/crates/heal/src/heal/manager/queue.rs index 3e2883089..2fa9962ee 100644 --- a/crates/heal/src/heal/manager/queue.rs +++ b/crates/heal/src/heal/manager/queue.rs @@ -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, pub(super) outcome: Option>, @@ -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> { diff --git a/crates/heal/src/heal/manager/root_recovery.rs b/crates/heal/src/heal/manager/root_recovery.rs index 7643908ed..dc4b004b5 100644 --- a/crates/heal/src/heal/manager/root_recovery.rs +++ b/crates/heal/src/heal/manager/root_recovery.rs @@ -208,6 +208,8 @@ struct RootHealTerminal { task_id: String, heal_type: RecoveryHealType, status: HealTaskStatus, + #[serde(default, deserialize_with = "decode_options")] + options: HealOptions, progress: Option, 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(), diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index 6b04bad40..d340daded 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -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 = 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 = 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 = 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![( diff --git a/crates/heal/src/heal/manager/tests/root_recovery.rs b/crates/heal/src/heal/manager/tests/root_recovery.rs index 88bd55c2a..93cded92e 100644 --- a/crates/heal/src/heal/manager/tests/root_recovery.rs +++ b/crates/heal/src/heal/manager/tests/root_recovery.rs @@ -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] diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 22dfcd438..3be240b78 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -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, + set: Option, +} + +impl From 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, + #[serde(default, rename = "settings", skip_serializing)] + heal_settings: Option, summary: String, #[serde(default, skip_serializing_if = "Vec::is_empty")] items: Vec, @@ -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> { 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(