diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 23b586280..16cf3ece5 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -944,6 +944,10 @@ impl HealManager { matches!(request.source, HealRequestSource::Admin | HealRequestSource::Internal) } + fn queued_request_can_be_displaced(request: &HealRequest) -> bool { + !root_recovery::is_admin_heal_recovery(&request.heal_type, request.source) + } + fn request_bypasses_mainline_throttle(request: &HealRequest) -> bool { request.force_start || matches!(request.source, HealRequestSource::Admin | HealRequestSource::Internal) @@ -1080,11 +1084,15 @@ impl HealManager { let per_object_request = request.heal_type.is_per_object(); if queue_len >= queue_capacity && !request.force_start { - if Self::can_displace_queued_work(&request) && queue.can_displace_lower_priority(request.priority) { + if Self::can_displace_queued_work(&request) + && queue.can_displace_lower_priority_where(request.priority, Self::queued_request_can_be_displaced) + { let request_id = request.id.clone(); let priority = request.priority; let source = request.source; - if let Some(displaced) = queue.push_displacing_lower_priority(request) { + if let Some(displaced) = + queue.push_displacing_lower_priority_where(request, Self::queued_request_can_be_displaced) + { publish_heal_queue_length(queue); Self::record_admission_metric(source, HealAdmissionResult::Accepted, context); demote_to_debug_when!(per_object_request, warn, target: "rustfs::heal::manager", { @@ -1484,13 +1492,13 @@ impl HealManager { ); // Keep scheduler, cancellation, and retry ownership stable until every - // unfinished root traversal has a durable successor. A failed write - // must leave the manager running and the shutdown marker unclean. + // unfinished admin control-plane heal has a durable successor. A failed + // write must leave the manager running and the shutdown marker unclean. let mut active_heals = self.active_heals.lock().await; let queue = self.heal_queue.lock().await; let retrying = self.retrying_heals.lock().await; for task in active_heals.values() { - if root_recovery::is_root_heal(&task.heal_type, task.source) { + if root_recovery::is_admin_heal_recovery(&task.heal_type, task.source) { if task.get_status().await == HealTaskStatus::Completed { self.root_recovery.remove(&task.id, &task.heal_type, task.source).await?; } else { @@ -1676,7 +1684,9 @@ impl HealManager { .requests() .chain(retrying.values().map(|retrying| &retrying.request)) .filter(|pending| { - root_recovery::is_root_heal(&pending.heal_type, pending.source) && pending.id != request.id + root_recovery::is_admin_heal_recovery(&pending.heal_type, pending.source) + && heal_types_overlap(&request.heal_type, &pending.heal_type) != OverlapVerdict::Disjoint + && pending.id != request.id }) .map(|pending| pending.id.clone()), ); @@ -1698,9 +1708,11 @@ impl HealManager { } } // A failed or timed-out replay may have only its durable owner - // left. Root responsibility overlaps every administrator path. + // left. Cancel only records that overlap this forced start. for pending in self.root_recovery.pending().await? { - if pending.id != request.id { + if pending.id != request.id + && heal_types_overlap(&request.heal_type, &pending.heal_type) != OverlapVerdict::Disjoint + { self.cancel_task(&pending.id).await?; } } @@ -1930,9 +1942,29 @@ impl HealManager { } } + let durable_handoff = root_recovery::is_admin_heal_recovery(&request.heal_type, request.source); + let durable_handoff_required = durable_handoff + && (request.force_start + || queue.len() < config.queue_size + || (Self::can_displace_queued_work(&request) + && queue.can_displace_lower_priority_where(request.priority, Self::queued_request_can_be_displaced))); + // An admin receipt is a control-plane responsibility. Persist it before + // queue publication so a crash after admission can replay it. + if durable_handoff_required { + self.root_recovery.persist(&request).await?; + } + let mut task_id = request.id.clone(); + let request_id = request.id.clone(); + let request_heal_type = request.heal_type.clone(); + let request_source = request.source; let admission_decision = Self::admit_request_to_queue(&mut queue, request, &config, "submit"); let admission = admission_decision.result; + if durable_handoff_required && !admission.is_admitted() { + self.root_recovery + .remove(&request_id, &request_heal_type, request_source) + .await?; + } if admission == HealAdmissionResult::Merged && let Some(queued_id) = queue.queued_request_id_for_dedup_key(&dedup_key) { @@ -2003,7 +2035,7 @@ impl HealManager { /// Retrying, never Pending — then the queue, and finally a terminal /// completed entry. `heal_path` additionally constrains the map matches /// the way the `*_for_path` variants always have. - async fn lookup_task_state(&self, canonical_task_id: &str, heal_path: Option<&str>) -> TaskStateLookup { + async fn lookup_task_state(&self, canonical_task_id: &str, heal_path: Option<&str>) -> Result { let matches_path = |heal_type: &HealType| heal_path.is_none_or(|path| heal_type_matches_path(heal_type, path)); { @@ -2012,7 +2044,7 @@ impl HealManager { .get(canonical_task_id) .filter(|task| matches_path(&task.heal_type)) { - return TaskStateLookup::Active(Arc::clone(task)); + return Ok(TaskStateLookup::Active(Arc::clone(task))); } } @@ -2022,7 +2054,7 @@ impl HealManager { .get(canonical_task_id) .filter(|retrying| matches_path(&retrying.request.heal_type)) { - return TaskStateLookup::Retrying(retrying.status()); + return Ok(TaskStateLookup::Retrying(retrying.status())); } } @@ -2035,7 +2067,7 @@ impl HealManager { prune_completed_heal_statuses(&mut completed_heals); if let Some(completed) = completed_heals.get(canonical_task_id).filter(|c| matches_path(&c.heal_type)) { if completed_status_is_retrying(&completed.status) { - return TaskStateLookup::Completed(Arc::clone(completed)); + return Ok(TaskStateLookup::Completed(Arc::clone(completed))); } terminal_completed = Some(Arc::clone(completed)); } @@ -2048,7 +2080,7 @@ impl HealManager { None => queue.contains_request_id(canonical_task_id), }; if queued { - return TaskStateLookup::Queued; + return Ok(TaskStateLookup::Queued); } } @@ -2061,15 +2093,55 @@ impl HealManager { .cloned(); } - match terminal_completed { + if terminal_completed.is_none() + && let Some(completed) = self.root_recovery.completed(canonical_task_id).await? + && matches_path(&completed.heal_type) + { + terminal_completed = Some(Arc::new(completed)); + } + + Ok(match terminal_completed { Some(completed) => TaskStateLookup::Completed(completed), None => TaskStateLookup::NotFound, - } + }) + } + + async fn publish_admin_terminal( + &self, + task_id: &str, + heal_type: &HealType, + source: HealRequestSource, + completed: &CompletedHealStatus, + ) -> Result { + self.root_recovery + .persist_terminal(task_id, heal_type, source, completed) + .await + } + + async fn publish_admin_cancelled_terminal( + &self, + task_id: &str, + heal_type: &HealType, + source: HealRequestSource, + ) -> Result { + let completed = CompletedHealStatus { + outcome: None, + progress: None, + retained_bytes: std::sync::OnceLock::new(), + heal_type: heal_type.clone(), + status: HealTaskStatus::Cancelled, + result_items_truncated: false, + completed_at: SystemTime::now(), + seqed_items: Vec::new(), + next_seq: 0, + min_seq: 0, + }; + self.publish_admin_terminal(task_id, heal_type, source, &completed).await } pub async fn get_task_status(&self, task_id: &str) -> Result { let canonical_task_id = self.canonical_task_id(task_id).await; - match self.lookup_task_state(&canonical_task_id, None).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::Completed(completed) => Ok(completed.status.clone()), @@ -2089,7 +2161,7 @@ impl HealManager { /// full-snapshot semantics. pub async fn get_task_report_since(&self, task_id: &str, since: Option) -> Result { let canonical_task_id = self.canonical_task_id(task_id).await; - match self.lookup_task_state(&canonical_task_id, None).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::Completed(completed) => Ok(completed_task_report(&completed, since)), @@ -2112,7 +2184,7 @@ impl HealManager { since: Option, ) -> Result { let canonical_task_id = self.canonical_task_id(task_id).await; - match self.lookup_task_state(&canonical_task_id, Some(heal_path)).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::Completed(completed) => Ok(completed_task_report(&completed, since)), @@ -2135,7 +2207,7 @@ impl HealManager { /// recently completed task, a different token is invalid for that path. pub async fn get_task_status_for_path(&self, heal_path: &str, task_id: &str) -> Result { let canonical_task_id = self.canonical_task_id(task_id).await; - match self.lookup_task_state(&canonical_task_id, Some(heal_path)).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::Completed(completed) => Ok(completed.status.clone()), @@ -2189,11 +2261,18 @@ impl HealManager { } drop(completed_heals); - let mut displaced_terminals = lock_displaced_terminals(&self.displaced_terminals); - prune_completed_heal_statuses(&mut displaced_terminals); - displaced_terminals - .values() - .any(|terminal| heal_type_matches_path(&terminal.heal_type, heal_path)) + { + let mut displaced_terminals = lock_displaced_terminals(&self.displaced_terminals); + prune_completed_heal_statuses(&mut displaced_terminals); + if displaced_terminals + .values() + .any(|terminal| heal_type_matches_path(&terminal.heal_type, heal_path)) + { + return true; + } + } + + self.root_recovery.completed_matches_path(heal_path).await.unwrap_or(false) } /// Get task progress @@ -2205,7 +2284,7 @@ impl HealManager { pub async fn get_task_progress(&self, task_id: &str) -> Result { let canonical_task_id = self.canonical_task_id(task_id).await; - let progress = match self.lookup_task_state(&canonical_task_id, None).await { + let progress = match self.lookup_task_state(&canonical_task_id, None).await? { TaskStateLookup::Active(task) => Some(task.get_progress().await), TaskStateLookup::Completed(completed) => completed.progress.clone(), _ => None, @@ -2221,9 +2300,11 @@ impl HealManager { { let mut active_heals = self.active_heals.lock().await; if let Some(task) = active_heals.get(&canonical_task_id) { + let completed = CompletedHealStatus::snapshot(task, HealTaskStatus::Cancelled).await; + self.publish_admin_terminal(&canonical_task_id, &task.heal_type, task.source, &completed) + .await?; self.root_recovery.remove(&task.id, &task.heal_type, task.source).await?; task.cancel().await?; - let completed = CompletedHealStatus::snapshot(task, HealTaskStatus::Cancelled).await; publish_completed_heal(&self.completed_heals, &self.task_aliases, &canonical_task_id, completed, true).await; active_heals.remove(&canonical_task_id); publish_active_heal_count(&active_heals); @@ -2246,6 +2327,8 @@ 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.root_recovery .remove(&canonical_task_id, &retrying.request.heal_type, retrying.request.source) .await?; @@ -2271,6 +2354,8 @@ 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) + .await?; self.root_recovery .remove(&request.id, &request.heal_type, request.source) .await?; @@ -2314,9 +2399,11 @@ impl HealManager { for task_id in &task_ids { if let Some(task) = active_heals.get(task_id) { + let completed = CompletedHealStatus::snapshot(task, HealTaskStatus::Cancelled).await; + self.publish_admin_terminal(task_id, &task.heal_type, task.source, &completed) + .await?; self.root_recovery.remove(&task.id, &task.heal_type, task.source).await?; task.cancel().await?; - let completed = CompletedHealStatus::snapshot(task, HealTaskStatus::Cancelled).await; publish_completed_heal(&self.completed_heals, &self.task_aliases, task_id, completed, true).await; } active_heals.remove(task_id); @@ -2344,6 +2431,8 @@ 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.root_recovery .remove(task_id, &retrying.request.heal_type, retrying.request.source) .await?; @@ -2369,15 +2458,22 @@ impl HealManager { } } - let mut queue = self.heal_queue.lock().await; - for request in queue - .requests() - .filter(|request| heal_type_matches_path(&request.heal_type, heal_path)) - { + let queued_matches = { + let queue = self.heal_queue.lock().await; + queue + .requests() + .filter(|request| heal_type_matches_path(&request.heal_type, heal_path)) + .cloned() + .collect::>() + }; + for request in &queued_matches { + self.publish_admin_cancelled_terminal(&request.id, &request.heal_type, request.source) + .await?; self.root_recovery .remove(&request.id, &request.heal_type, request.source) .await?; } + let mut queue = self.heal_queue.lock().await; let queued_cancelled = queue.remove_matching(|request| heal_type_matches_path(&request.heal_type, heal_path)); if !queued_cancelled.is_empty() { publish_heal_queue_length(&queue); @@ -2389,11 +2485,15 @@ impl HealManager { self.remove_mrf_repair_notice_targets_for_task(&request.id); } - if heal_type_matches_path(&HealType::Cluster, heal_path) { - for pending in self.root_recovery.pending().await? { - if self.root_recovery.cancel_pending(&pending.id).await? { - cancelled += 1; - } + for pending in self + .root_recovery + .pending() + .await? + .into_iter() + .filter(|pending| heal_type_matches_path(&pending.heal_type, heal_path)) + { + if self.root_recovery.cancel_pending(&pending.id).await? { + cancelled += 1; } } if cancelled == 0 { diff --git a/crates/heal/src/heal/manager/queue.rs b/crates/heal/src/heal/manager/queue.rs index a47cd43e2..3e2883089 100644 --- a/crates/heal/src/heal/manager/queue.rs +++ b/crates/heal/src/heal/manager/queue.rs @@ -288,16 +288,29 @@ impl PriorityHealQueue { QueuePushOutcome::Accepted } - pub(super) fn can_displace_lower_priority(&self, priority: HealPriority) -> bool { - self.heap.iter().any(|item| item.priority < priority) + pub(super) fn can_displace_lower_priority_where(&self, priority: HealPriority, can_displace: F) -> bool + where + F: Fn(&HealRequest) -> bool, + { + self.heap + .iter() + .any(|item| item.priority < priority && can_displace(&item.request)) } - pub(super) fn push_displacing_lower_priority(&mut self, request: HealRequest) -> Option { + #[cfg(test)] + pub(super) fn can_displace_lower_priority(&self, priority: HealPriority) -> bool { + self.can_displace_lower_priority_where(priority, |_| true) + } + + pub(super) fn push_displacing_lower_priority_where(&mut self, request: HealRequest, can_displace: F) -> Option + where + F: Fn(&HealRequest) -> bool, + { let mut retained = BinaryHeap::new(); let mut displaced: Option = None; while let Some(item) = self.heap.pop() { - if item.priority < request.priority { + if item.priority < request.priority && can_displace(&item.request) { let should_displace = displaced .as_ref() .map(|current| { @@ -337,6 +350,11 @@ impl PriorityHealQueue { displaced } + #[cfg(test)] + pub(super) fn push_displacing_lower_priority(&mut self, request: HealRequest) -> Option { + self.push_displacing_lower_priority_where(request, |_| true) + } + /// Get statistics about queue contents by priority pub(super) fn get_priority_stats(&self) -> HashMap { let mut stats = HashMap::new(); diff --git a/crates/heal/src/heal/manager/root_recovery.rs b/crates/heal/src/heal/manager/root_recovery.rs index f32ff1028..4ba4bafc5 100644 --- a/crates/heal/src/heal/manager/root_recovery.rs +++ b/crates/heal/src/heal/manager/root_recovery.rs @@ -12,10 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Graceful-shutdown handoff for administrator root heals. This namespace is +//! Graceful-shutdown handoff for administrator heals. This namespace is //! separate from erasure-set checkpoints and replacement generations, which -//! cannot represent a cluster traversal. One coordinator disk owns each -//! record; never create a fallback copy after an uncertain write or deletion. +//! cannot represent an admitted admin control-plane request. One coordinator +//! disk owns each record; never create a fallback copy after an uncertain write +//! or deletion. use super::*; use crate::heal::storage_api::owner::{EcstoreConditionalFileUpdate, EcstoreDiskAPI, EcstoreDiskBytes}; @@ -25,13 +26,172 @@ use serde::{Deserialize, Serialize}; // The metadata bucket already exists and its parent is durable. Creating a // nested journal directory here would also require syncing every ancestor. const ROOT_RECOVERY_PREFIX: &str = "root-heal-"; -const ROOT_RECOVERY_SCHEMA: u32 = 1; +const ROOT_TERMINAL_PREFIX: &str = "terminal-root-heal-"; +const LEGACY_ROOT_RECOVERY_SCHEMA: u32 = 1; +const ROOT_RECOVERY_SCHEMA: u32 = 2; +const ROOT_TERMINAL_SCHEMA: u32 = 1; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +enum RecoveryHealType { + Cluster, + Bucket { + bucket: String, + }, + Object { + bucket: String, + object: String, + version_id: Option, + }, + Prefix { + bucket: String, + prefix: String, + }, + ErasureSet { + buckets: Vec, + set_disk_id: String, + }, + Metadata { + bucket: String, + object: String, + }, + EcDecode { + bucket: String, + object: String, + version_id: Option, + }, +} + +impl RecoveryHealType { + fn validate(&self) -> Result<()> { + match self { + Self::Cluster => {} + Self::Bucket { bucket } => validate_recovery_component("bucket", bucket)?, + Self::Object { + bucket, + object, + version_id, + } + | Self::EcDecode { + bucket, + object, + version_id, + } => { + validate_recovery_component("bucket", bucket)?; + validate_recovery_component("object", object)?; + if let Some(version_id) = version_id { + validate_recovery_component("version id", version_id)?; + } + } + Self::Prefix { bucket, prefix } => { + validate_recovery_component("bucket", bucket)?; + validate_recovery_component("prefix", prefix)?; + } + Self::ErasureSet { buckets, set_disk_id } => { + validate_recovery_component("set disk id", set_disk_id)?; + if buckets.is_empty() { + return Err(Error::Other("Admin heal recovery erasure set must name buckets".to_string())); + } + for bucket in buckets { + validate_recovery_component("bucket", bucket)?; + } + } + Self::Metadata { bucket, object } => { + validate_recovery_component("bucket", bucket)?; + validate_recovery_component("object", object)?; + } + } + Ok(()) + } +} + +impl From<&HealType> for RecoveryHealType { + fn from(heal_type: &HealType) -> Self { + match heal_type { + HealType::Cluster => Self::Cluster, + HealType::Bucket { bucket } => Self::Bucket { bucket: bucket.clone() }, + HealType::Object { + bucket, + object, + version_id, + } => Self::Object { + bucket: bucket.clone(), + object: object.clone(), + version_id: version_id.clone(), + }, + HealType::Prefix { bucket, prefix } => Self::Prefix { + bucket: bucket.clone(), + prefix: prefix.clone(), + }, + HealType::ErasureSet { buckets, set_disk_id } => Self::ErasureSet { + buckets: buckets.clone(), + set_disk_id: set_disk_id.clone(), + }, + HealType::Metadata { bucket, object } => Self::Metadata { + bucket: bucket.clone(), + object: object.clone(), + }, + HealType::ECDecode { + bucket, + object, + version_id, + } => Self::EcDecode { + bucket: bucket.clone(), + object: object.clone(), + version_id: version_id.clone(), + }, + } + } +} + +impl From for HealType { + fn from(heal_type: RecoveryHealType) -> Self { + match heal_type { + RecoveryHealType::Cluster => Self::Cluster, + RecoveryHealType::Bucket { bucket } => Self::Bucket { bucket }, + RecoveryHealType::Object { + bucket, + object, + version_id, + } => Self::Object { + bucket, + object, + version_id, + }, + RecoveryHealType::Prefix { bucket, prefix } => Self::Prefix { bucket, prefix }, + RecoveryHealType::ErasureSet { buckets, set_disk_id } => Self::ErasureSet { buckets, set_disk_id }, + RecoveryHealType::Metadata { bucket, object } => Self::Metadata { bucket, object }, + RecoveryHealType::EcDecode { + bucket, + object, + version_id, + } => Self::ECDecode { + bucket, + object, + version_id, + }, + } + } +} + +fn validate_recovery_component(label: &str, value: &str) -> Result<()> { + if value.is_empty() || value.contains('\0') { + return Err(Error::Other(format!("Invalid admin heal recovery {label}"))); + } + Ok(()) +} + +fn default_recovery_heal_type() -> RecoveryHealType { + RecoveryHealType::Cluster +} #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] struct RootHealIntent { schema: u32, task_id: String, + #[serde(default = "default_recovery_heal_type")] + heal_type: RecoveryHealType, #[serde(deserialize_with = "decode_options")] options: HealOptions, priority: HealPriority, @@ -39,11 +199,62 @@ struct RootHealIntent { created_at: SystemTime, } +#[derive(Debug, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct RootHealTerminal { + schema: u32, + task_id: String, + heal_type: RecoveryHealType, + status: HealTaskStatus, + progress: Option, + completed_at: SystemTime, +} + +impl RootHealTerminal { + fn from_completed(task_id: &str, completed: &CompletedHealStatus) -> Self { + Self { + schema: ROOT_TERMINAL_SCHEMA, + task_id: task_id.to_owned(), + heal_type: RecoveryHealType::from(&completed.heal_type), + status: completed.status.clone(), + progress: completed.progress.clone(), + completed_at: completed.completed_at, + } + } + + fn cancelled(task_id: &str, heal_type: &HealType) -> Self { + Self { + schema: ROOT_TERMINAL_SCHEMA, + task_id: task_id.to_owned(), + heal_type: RecoveryHealType::from(heal_type), + status: HealTaskStatus::Cancelled, + progress: None, + completed_at: SystemTime::now(), + } + } + + fn into_completed(self) -> CompletedHealStatus { + CompletedHealStatus { + outcome: None, + progress: self.progress, + retained_bytes: std::sync::OnceLock::new(), + heal_type: self.heal_type.into(), + status: self.status, + result_items_truncated: false, + completed_at: self.completed_at, + seqed_items: Vec::new(), + next_seq: 0, + min_seq: 0, + } + } +} + impl RootHealIntent { fn from_request(request: &HealRequest) -> Self { Self { schema: ROOT_RECOVERY_SCHEMA, task_id: request.id.clone(), + heal_type: RecoveryHealType::from(&request.heal_type), options: request.options.clone(), priority: request.priority, retry_attempts: request.retry_attempts, @@ -52,7 +263,7 @@ impl RootHealIntent { } fn into_request(self) -> HealRequest { - let mut request = HealRequest::new(HealType::Cluster, self.options, self.priority); + let mut request = HealRequest::new(self.heal_type.into(), self.options, self.priority); request.id = self.task_id; request.source = HealRequestSource::Admin; request.retry_attempts = self.retry_attempts; @@ -68,8 +279,18 @@ pub(super) struct RootHealRecovery { disks: Option>, } -pub(super) fn is_root_heal(heal_type: &HealType, source: HealRequestSource) -> bool { - source == HealRequestSource::Admin && matches!(heal_type, HealType::Cluster) +pub(super) fn is_admin_heal_recovery(heal_type: &HealType, source: HealRequestSource) -> bool { + source == HealRequestSource::Admin + && matches!( + heal_type, + HealType::Cluster + | HealType::Bucket { .. } + | HealType::Object { .. } + | HealType::Prefix { .. } + | HealType::ErasureSet { .. } + | HealType::Metadata { .. } + | HealType::ECDecode { .. } + ) } fn decode_options<'de, D: serde::Deserializer<'de>>(deserializer: D) -> std::result::Result { @@ -107,16 +328,41 @@ fn intent_path(task_id: &str) -> Result { Ok(format!("{ROOT_RECOVERY_PREFIX}{task_id}.json")) } +fn terminal_path(task_id: &str) -> Result { + let parsed = uuid::Uuid::parse_str(task_id).map_err(|_| Error::Other("Invalid root heal terminal task id".to_string()))?; + if parsed.to_string() != task_id { + return Err(Error::Other("Noncanonical root heal terminal task id".to_string())); + } + Ok(format!("{ROOT_TERMINAL_PREFIX}{task_id}.json")) +} + fn decode_intent(task_id: &str, bytes: &[u8]) -> Result { let _ = intent_path(task_id)?; let intent: RootHealIntent = serde_json::from_slice(bytes) .map_err(|error| Error::Other(format!("Invalid root heal recovery record {task_id}: {error}")))?; - if intent.schema != ROOT_RECOVERY_SCHEMA || intent.task_id != task_id { + if intent.task_id != task_id { return Err(Error::Other(format!("Unsupported or mismatched root heal recovery record {task_id}"))); } + match intent.schema { + LEGACY_ROOT_RECOVERY_SCHEMA if intent.heal_type == RecoveryHealType::Cluster => {} + ROOT_RECOVERY_SCHEMA => {} + _ => return Err(Error::Other(format!("Unsupported or mismatched root heal recovery record {task_id}"))), + } + intent.heal_type.validate()?; Ok(intent) } +fn decode_terminal(task_id: &str, bytes: &[u8]) -> Result { + let _ = terminal_path(task_id)?; + let terminal: RootHealTerminal = serde_json::from_slice(bytes) + .map_err(|error| Error::Other(format!("Invalid root heal terminal record {task_id}: {error}")))?; + if terminal.schema != ROOT_TERMINAL_SCHEMA || terminal.task_id != task_id { + return Err(Error::Other(format!("Unsupported or mismatched root heal terminal record {task_id}"))); + } + terminal.heal_type.validate()?; + Ok(terminal) +} + impl RootHealRecovery { #[cfg(test)] pub(super) fn with_disks(disks: Vec) -> Self { @@ -162,8 +408,55 @@ impl RootHealRecovery { Ok(found) } + async fn find_terminal(disks: &[DiskStore], task_id: &str) -> Result> { + let path = terminal_path(task_id)?; + let mut found = None; + for disk in disks { + EcstoreDiskAPI::stat_volume(disk.as_ref(), RUSTFS_META_BUCKET).await?; + match EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, &path).await { + Ok(bytes) => { + decode_terminal(task_id, &bytes)?; + if found.is_some() { + return Err(Error::Other(format!("Multiple root heal terminal owners for {task_id}"))); + } + found = Some((disk.clone(), bytes)); + } + Err(DiskError::FileNotFound) => {} + Err(error) => return Err(Error::Disk(error)), + } + } + Ok(found) + } + + async fn persist_terminal_locked( + disks: &[DiskStore], + task_id: &str, + terminal: RootHealTerminal, + ) -> Result> { + let path = terminal_path(task_id)?; + if let Some((_, bytes)) = Self::find_terminal(disks, task_id).await? { + let current = decode_terminal(task_id, &bytes)?; + if current == terminal { + return Self::find(disks, task_id).await; + } + return Err(Error::Other(format!("Root heal terminal record changed for {task_id}"))); + } + let pending = Self::find(disks, task_id).await?; + let disk = pending + .as_ref() + .map(|(disk, _)| disk.clone()) + .or_else(|| disks.first().cloned()) + .ok_or_else(|| Error::Other("No local disk available for root heal terminal receipt".to_string()))?; + let bytes = serde_json::to_vec(&terminal) + .map_err(|error| Error::Other(format!("Serialize root heal terminal receipt: {error}")))?; + match EcstoreDiskAPI::compare_and_update_file(disk.as_ref(), RUSTFS_META_BUCKET, &path, None, Some(bytes.into())).await? { + EcstoreConditionalFileUpdate::Updated => Ok(pending), + _ => Err(Error::Other(format!("Root heal terminal record changed for {task_id}"))), + } + } + pub(super) async fn persist(&self, request: &HealRequest) -> Result<()> { - if !is_root_heal(&request.heal_type, request.source) { + if !is_admin_heal_recovery(&request.heal_type, request.source) { return Ok(()); } let _guard = self.mutation.lock().await; @@ -199,9 +492,13 @@ impl RootHealRecovery { } pub(super) async fn remove(&self, task_id: &str, heal_type: &HealType, source: HealRequestSource) -> Result { - if !is_root_heal(heal_type, source) { + if !is_admin_heal_recovery(heal_type, source) { return Ok(false); } + self.remove_pending_by_id(task_id).await + } + + async fn remove_pending_by_id(&self, task_id: &str) -> Result { let _guard = self.mutation.lock().await; let Some((disk, bytes)) = Self::find(&self.disks().await?, task_id).await? else { return Ok(false); @@ -221,7 +518,7 @@ impl RootHealRecovery { } pub(super) async fn checkpoint_failed_execution(&self, task: &HealTask) -> Result<()> { - if !is_root_heal(&task.heal_type, task.source) { + if !is_admin_heal_recovery(&task.heal_type, task.source) { return Ok(()); } let remaining = match task.retry_request_with_remaining_timeout().await { @@ -235,6 +532,9 @@ impl RootHealRecovery { return Ok(()); }; let mut intent = decode_intent(&task.id, &expected)?; + if HealType::from(intent.heal_type.clone()) != task.heal_type { + return Err(Error::Other(format!("Root heal recovery owner changed for {}", task.id))); + } let mut expected_options = intent.options.clone(); expected_options.timeout = task.options.timeout; if intent.created_at != task.created_at || intent.priority != task.priority || expected_options != task.options { @@ -268,7 +568,120 @@ impl RootHealRecovery { if intent_path(task_id).is_err() { return Ok(false); } - self.remove(task_id, &HealType::Cluster, HealRequestSource::Admin).await + let _guard = self.mutation.lock().await; + let disks = self.disks().await?; + if Self::find_terminal(&disks, task_id).await?.is_some() { + if let Some((disk, bytes)) = Self::find(&disks, task_id).await? { + match EcstoreDiskAPI::compare_and_update_file( + disk.as_ref(), + RUSTFS_META_BUCKET, + &intent_path(task_id)?, + Some(bytes), + None, + ) + .await? + { + EcstoreConditionalFileUpdate::Updated => {} + _ => return Err(Error::Other(format!("Root heal recovery record changed while cancelling {task_id}"))), + } + } + return Ok(true); + } + let Some((disk, bytes)) = Self::find(&disks, task_id).await? else { + return Ok(false); + }; + let pending = decode_intent(task_id, &bytes)?; + let heal_type = HealType::from(pending.heal_type); + let terminal = RootHealTerminal::cancelled(task_id, &heal_type); + let _ = Self::persist_terminal_locked(&disks, task_id, terminal).await?; + match EcstoreDiskAPI::compare_and_update_file( + disk.as_ref(), + RUSTFS_META_BUCKET, + &intent_path(task_id)?, + Some(bytes), + None, + ) + .await? + { + EcstoreConditionalFileUpdate::Updated => Ok(true), + _ => Err(Error::Other(format!("Root heal recovery record changed while cancelling {task_id}"))), + } + } + + pub(super) async fn persist_terminal( + &self, + task_id: &str, + heal_type: &HealType, + source: HealRequestSource, + completed: &CompletedHealStatus, + ) -> Result { + if !is_admin_heal_recovery(heal_type, source) || completed.heal_type != *heal_type { + return Ok(false); + } + let _guard = self.mutation.lock().await; + let disks = self.disks().await?; + let pending = + Self::persist_terminal_locked(&disks, task_id, RootHealTerminal::from_completed(task_id, completed)).await?; + if let Some((disk, bytes)) = pending { + match EcstoreDiskAPI::compare_and_update_file( + disk.as_ref(), + RUSTFS_META_BUCKET, + &intent_path(task_id)?, + Some(bytes), + None, + ) + .await? + { + EcstoreConditionalFileUpdate::Updated => {} + _ => { + return Err(Error::Other(format!( + "Root heal recovery record changed while publishing terminal {task_id}" + ))); + } + } + } + Ok(true) + } + + pub(super) async fn completed(&self, task_id: &str) -> Result> { + if terminal_path(task_id).is_err() { + return Ok(None); + } + let _guard = self.mutation.lock().await; + let disks = self.disks().await?; + let Some((_, bytes)) = Self::find_terminal(&disks, task_id).await? else { + return Ok(None); + }; + Ok(Some(decode_terminal(task_id, &bytes)?.into_completed())) + } + + pub(super) async fn completed_matches_path(&self, heal_path: &str) -> Result { + let _guard = self.mutation.lock().await; + let disks = self.disks().await?; + for disk in &disks { + EcstoreDiskAPI::stat_volume(disk.as_ref(), RUSTFS_META_BUCKET).await?; + let entries = match EcstoreDiskAPI::list_dir(disk.as_ref(), "", RUSTFS_META_BUCKET, "", -1).await { + Ok(entries) => entries, + Err(DiskError::FileNotFound) => continue, + Err(error) => return Err(Error::Disk(error)), + }; + for entry in entries { + let Some(task_id) = entry + .strip_prefix(ROOT_TERMINAL_PREFIX) + .and_then(|entry| entry.strip_suffix(".json")) + else { + continue; + }; + let Some((_, bytes)) = Self::find_terminal(&disks, task_id).await? else { + continue; + }; + let heal_type = HealType::from(decode_terminal(task_id, &bytes)?.heal_type); + if heal_type_matches_path(&heal_type, heal_path) { + return Ok(true); + } + } + } + Ok(false) } pub(super) async fn pending(&self) -> Result> { @@ -295,6 +708,9 @@ impl RootHealRecovery { } let mut requests = Vec::new(); for task_id in ids { + if Self::find_terminal(&disks, &task_id).await?.is_some() { + continue; + } if let Some((_, bytes)) = Self::find(&disks, &task_id).await? { requests.push(decode_intent(&task_id, &bytes)?.into_request()); } diff --git a/crates/heal/src/heal/manager/scheduler.rs b/crates/heal/src/heal/manager/scheduler.rs index a3820fa2b..dfe96dfaa 100644 --- a/crates/heal/src/heal/manager/scheduler.rs +++ b/crates/heal/src/heal/manager/scheduler.rs @@ -302,38 +302,6 @@ impl HealManager { tests::pause_completed_retention_before_publish(&task_id, &completed_status).await; let mut active_heals_guard = active_heals_clone.lock().await; let owns_completion = active_heals_guard.contains_key(&task_id); - if owns_completion - && result.is_ok() - && let Err(error) = root_recovery_clone.remove(&task_id, &task.heal_type, task.source).await - { - // Keep the durable responsibility if retirement fails. - // Replaying a completed traversal is idempotent. - warn!( - target: "rustfs::heal::manager", - event = EVENT_HEAL_SCHEDULER_STATE, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_MANAGER, - task_id, - state = "root_recovery_retirement_failed", - error = %error, - "Failed to retire root heal recovery record" - ); - } - if owns_completion - && result.is_err() - && let Err(error) = root_recovery_clone.checkpoint_failed_execution(&task).await - { - warn!( - target: "rustfs::heal::manager", - event = EVENT_HEAL_SCHEDULER_STATE, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_MANAGER, - task_id, - state = "root_recovery_checkpoint_failed", - error = %error, - "Failed to checkpoint root heal recovery execution budget" - ); - } let cancelled_completion = if owns_completion { false } else { @@ -352,7 +320,47 @@ impl HealManager { completed_status_entry.status = HealTaskStatus::Cancelled; completed_status_entry.outcome = Some(Arc::new(task.get_outcome().await)); } - let terminal_completion = !matches!(completed_status, HealTaskStatus::Retrying { .. }); + let terminal_completion = matches!( + completed_status, + HealTaskStatus::Completed | HealTaskStatus::Cancelled | HealTaskStatus::Failed { .. } + ); + if owns_completion + && terminal_completion + && root_recovery::is_admin_heal_recovery(&task.heal_type, task.source) + && let Err(error) = root_recovery_clone + .persist_terminal(&task_id, &task.heal_type, task.source, &completed_status_entry) + .await + { + // Keep the durable responsibility if terminal + // publication fails. Replaying the task is preferable + // to losing the final receipt across restart. + warn!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_SCHEDULER_STATE, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + task_id, + state = "root_recovery_terminal_publish_failed", + error = %error, + "Failed to publish heal terminal receipt" + ); + } + if owns_completion + && !terminal_completion + && result.is_err() + && let Err(error) = root_recovery_clone.checkpoint_failed_execution(&task).await + { + warn!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_SCHEDULER_STATE, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + task_id, + state = "root_recovery_checkpoint_failed", + error = %error, + "Failed to checkpoint root heal recovery execution budget" + ); + } let completed_status_for_verified_events = completed_status_entry.clone(); // Keep retry ownership continuous: status snapshots acquire // these locks in the same active -> retrying order. diff --git a/crates/heal/src/heal/manager/tests/root_recovery.rs b/crates/heal/src/heal/manager/tests/root_recovery.rs index 28fb83288..1f1a65314 100644 --- a/crates/heal/src/heal/manager/tests/root_recovery.rs +++ b/crates/heal/src/heal/manager/tests/root_recovery.rs @@ -15,6 +15,7 @@ use super::super::root_recovery::RootHealRecovery; use super::*; use crate::heal::RUSTFS_META_BUCKET; +use std::collections::HashSet; async fn recovery_disk() -> (TempDir, DiskStore) { let temp = TempDir::new().expect("temporary root recovery disk"); @@ -48,7 +49,11 @@ fn recovery_manager(disks: Vec) -> HealManager { } fn root_request() -> HealRequest { - let mut request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::High); + admin_request(HealType::Cluster) +} + +fn admin_request(heal_type: HealType) -> HealRequest { + let mut request = HealRequest::new(heal_type, HealOptions::default(), HealPriority::High); request.source = HealRequestSource::Admin; request } @@ -106,6 +111,549 @@ async fn root_recovery_shutdown_restart_replays_same_id_and_success_retires_inte assert!(restarted.root_recovery.pending().await.expect("read completion").is_empty()); } +#[tokio::test] +async fn root_recovery_admin_start_persists_before_shutdown() { + let (_temp, disk) = recovery_disk().await; + let manager = recovery_manager(vec![disk.clone()]); + let mut request = root_request(); + request.options.recursive = true; + let receipt = manager + .submit_heal_request_with_receipt(request.clone()) + .await + .expect("root admission should persist"); + assert_eq!(receipt.result, HealAdmissionResult::Accepted); + assert_eq!(receipt.task_id, request.id); + let pending = manager.root_recovery.pending().await.expect("read durable admission"); + assert_eq!( + pending.iter().map(|request| request.id.as_str()).collect::>(), + [request.id.as_str()] + ); + drop(manager); + + let restarted = recovery_manager(vec![disk]); + restarted.replay_root_heals().await.expect("replay durable admission"); + let queued = restarted + .heal_queue + .lock() + .await + .requests() + .map(|request| request.id.clone()) + .collect::>(); + assert_eq!(queued, [request.id]); +} + +#[tokio::test] +async fn root_recovery_admin_non_root_types_persist_and_replay() { + for heal_type in [ + HealType::Bucket { + bucket: "bucket".to_string(), + }, + HealType::Prefix { + bucket: "bucket".to_string(), + prefix: "logs/2026".to_string(), + }, + HealType::Object { + bucket: "bucket".to_string(), + object: "object".to_string(), + version_id: Some("version-1".to_string()), + }, + HealType::Metadata { + bucket: "bucket".to_string(), + object: "object".to_string(), + }, + HealType::ECDecode { + bucket: "bucket".to_string(), + object: "object".to_string(), + version_id: None, + }, + HealType::ErasureSet { + buckets: vec!["bucket".to_string()], + set_disk_id: "pool_0_set_1".to_string(), + }, + ] { + let (_temp, disk) = recovery_disk().await; + let manager = recovery_manager(vec![disk.clone()]); + let mut request = admin_request(heal_type.clone()); + request.options.recursive = true; + let receipt = manager + .submit_heal_request_with_receipt(request.clone()) + .await + .expect("admin heal admission should persist"); + assert_eq!(receipt.result, HealAdmissionResult::Accepted); + assert_eq!( + manager + .root_recovery + .pending() + .await + .expect("read durable admin state") + .iter() + .map(|request| (&request.id, &request.heal_type)) + .collect::>(), + [(&request.id, &request.heal_type)] + ); + drop(manager); + + let restarted = recovery_manager(vec![disk]); + restarted.replay_root_heals().await.expect("replay durable admin heal"); + let queued = restarted.heal_queue.lock().await.requests().cloned().collect::>(); + assert_eq!(queued.len(), 1); + assert_eq!(queued[0].id, request.id); + assert_eq!(queued[0].heal_type, request.heal_type); + assert_eq!(queued[0].options, request.options); + assert_eq!(queued[0].source, HealRequestSource::Admin); + } +} + +#[tokio::test] +async fn root_recovery_non_admin_request_is_not_persisted() { + let (_temp, disk) = recovery_disk().await; + let manager = recovery_manager(vec![disk.clone()]); + let mut request = HealRequest::new( + HealType::Bucket { + bucket: "scanner".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + request.source = HealRequestSource::Scanner; + assert_eq!( + manager + .submit_heal_request(request) + .await + .expect("scanner request should still queue"), + HealAdmissionResult::Accepted + ); + assert!(manager.root_recovery.pending().await.expect("read durable state").is_empty()); + drop(manager); + + let restarted = recovery_manager(vec![disk]); + restarted.replay_root_heals().await.expect("replay empty durable state"); + assert_eq!(restarted.get_queue_length().await, 0); +} + +#[tokio::test] +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 { + bucket: "bucket".to_string(), + }); + manager + .root_recovery + .persist(&request) + .await + .expect("durable bucket responsibility"); + assert_eq!( + manager + .cancel_tasks_for_path("bucket") + .await + .expect("cancel durable-only bucket path"), + 1 + ); + drop(manager); + + let restarted = recovery_manager(vec![disk]); + restarted + .replay_root_heals() + .await + .expect("restart after durable path cancellation"); + assert_eq!(restarted.get_queue_length().await, 0); + assert_eq!( + restarted + .get_task_status_for_path("bucket", &request.id) + .await + .expect("durable cancellation remains queryable by path"), + HealTaskStatus::Cancelled + ); + assert_eq!( + restarted + .get_task_status(&request.id) + .await + .expect("durable cancellation remains queryable by id"), + HealTaskStatus::Cancelled + ); +} + +#[tokio::test] +async fn root_recovery_active_cancel_is_queryable_after_restart_for_scoped_admin() { + let (_temp, disk) = recovery_disk().await; + let manager = recovery_manager(vec![disk.clone()]); + let request = admin_request(HealType::Bucket { + bucket: "bucket".to_string(), + }); + active_root(&manager, request.clone()).await; + manager + .root_recovery + .persist(&request) + .await + .expect("durable active bucket responsibility"); + + manager.cancel_task(&request.id).await.expect("cancel active bucket"); + assert!( + manager + .root_recovery + .pending() + .await + .expect("active terminal retires intent") + .is_empty() + ); + drop(manager); + + let restarted = recovery_manager(vec![disk]); + restarted + .replay_root_heals() + .await + .expect("restart after active cancellation"); + assert_eq!(restarted.get_queue_length().await, 0); + assert_eq!( + restarted + .get_task_status(&request.id) + .await + .expect("active cancellation remains queryable by id"), + HealTaskStatus::Cancelled + ); + assert_eq!( + restarted + .get_task_status_for_path("bucket", &request.id) + .await + .expect("active cancellation remains queryable by path"), + HealTaskStatus::Cancelled + ); +} + +#[tokio::test] +async fn root_recovery_terminal_receipt_wins_over_stale_pending_scoped_intent_after_restart() { + let (_temp, disk) = recovery_disk().await; + let manager = recovery_manager(vec![disk.clone()]); + let request = admin_request(HealType::Bucket { + bucket: "bucket".to_string(), + }); + manager + .root_recovery + .persist(&request) + .await + .expect("durable bucket responsibility"); + manager + .publish_admin_cancelled_terminal(&request.id, &request.heal_type, request.source) + .await + .expect("publish terminal receipt"); + manager + .root_recovery + .persist(&request) + .await + .expect("restore stale pending intent after terminal publication"); + assert!( + manager + .root_recovery + .pending() + .await + .expect("terminal masks stale pending") + .is_empty() + ); + drop(manager); + + let restarted = recovery_manager(vec![disk]); + restarted + .replay_root_heals() + .await + .expect("restart with terminal and stale pending"); + assert_eq!(restarted.get_queue_length().await, 0); + assert_eq!( + restarted + .get_task_status(&request.id) + .await + .expect("terminal status survives stale pending"), + HealTaskStatus::Cancelled + ); +} + +#[tokio::test] +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 { + bucket: "bucket".to_string(), + }); + manager + .root_recovery + .persist(&request) + .await + .expect("durable bucket responsibility"); + let completed = CompletedHealStatus { + outcome: None, + heal_type: request.heal_type.clone(), + status: HealTaskStatus::Completed, + progress: Some(HealProgress { + objects_scanned: 2, + objects_healed: 2, + bytes_processed: 128, + ..Default::default() + }), + retained_bytes: std::sync::OnceLock::new(), + result_items_truncated: false, + completed_at: SystemTime::now(), + seqed_items: Vec::new(), + next_seq: 0, + min_seq: 0, + }; + assert!( + manager + .publish_admin_terminal(&request.id, &request.heal_type, request.source, &completed) + .await + .expect("publish completed terminal") + ); + assert!( + manager + .root_recovery + .pending() + .await + .expect("completed terminal retires intent") + .is_empty() + ); + drop(manager); + + let restarted = recovery_manager(vec![disk]); + restarted.replay_root_heals().await.expect("restart after completed terminal"); + assert_eq!(restarted.get_queue_length().await, 0); + assert_eq!( + restarted + .get_task_status_for_path("bucket", &request.id) + .await + .expect("completed terminal remains queryable by path"), + HealTaskStatus::Completed + ); + let progress = restarted + .get_task_progress(&request.id) + .await + .expect("completed terminal exposes progress"); + assert_eq!(progress.objects_scanned, 2); + assert_eq!(progress.objects_healed, 2); +} + +#[tokio::test] +async fn root_recovery_admin_start_fails_closed_when_owner_is_unavailable() { + let (_temp, disk) = recovery_disk().await; + let (unavailable_temp, unavailable) = recovery_disk().await; + std::fs::remove_dir_all(unavailable_temp.path().join(RUSTFS_META_BUCKET)).expect("make owner volume unavailable"); + let manager = recovery_manager(vec![unavailable, disk.clone()]); + let request = admin_request(HealType::Bucket { + bucket: "bucket".to_string(), + }); + + assert!( + manager.submit_heal_request_with_receipt(request).await.is_err(), + "admin admission must fail closed when the durable owner cannot be checked" + ); + assert_eq!(manager.get_queue_length().await, 0); + assert!( + RootHealRecovery::with_disks(vec![disk]) + .pending() + .await + .expect("other disk remains empty") + .is_empty() + ); +} + +#[tokio::test] +async fn root_recovery_force_start_cancels_only_overlapping_durable_admin_records() { + let (_temp, disk) = recovery_disk().await; + let manager = recovery_manager(vec![disk.clone()]); + let old = admin_request(HealType::Bucket { + bucket: "bucket-a".to_string(), + }); + let disjoint = admin_request(HealType::Bucket { + bucket: "bucket-b".to_string(), + }); + manager.root_recovery.persist(&old).await.expect("old bucket owner"); + manager.root_recovery.persist(&disjoint).await.expect("disjoint bucket owner"); + + let mut replacement = admin_request(HealType::Prefix { + bucket: "bucket-a".to_string(), + prefix: "logs/".to_string(), + }); + replacement.force_start = true; + assert_eq!( + manager + .submit_heal_request(replacement.clone()) + .await + .expect("forceStart should replace only the overlapping durable owner"), + HealAdmissionResult::Accepted + ); + + let mut pending = manager + .root_recovery + .pending() + .await + .expect("read durable owners") + .into_iter() + .map(|request| (request.id, request.heal_type)) + .collect::>(); + pending.sort_by(|left, right| left.0.cmp(&right.0)); + let mut expected = vec![ + (disjoint.id.clone(), disjoint.heal_type.clone()), + (replacement.id.clone(), replacement.heal_type.clone()), + ]; + expected.sort_by(|left, right| left.0.cmp(&right.0)); + assert_eq!(pending, expected); + drop(manager); + + let restarted = recovery_manager(vec![disk]); + restarted.replay_root_heals().await.expect("replay surviving owners"); + let queued_ids = restarted + .heal_queue + .lock() + .await + .requests() + .map(|request| request.id.clone()) + .collect::>(); + assert_eq!(queued_ids, HashSet::from([disjoint.id, replacement.id])); +} + +#[tokio::test] +async fn root_recovery_queued_non_root_admin_owner_is_not_priority_displaced() { + let (_temp, disk) = recovery_disk().await; + let manager = recovery_manager(vec![disk.clone()]); + manager.config.write().await.queue_size = 1; + let mut durable = admin_request(HealType::Bucket { + bucket: "bucket".to_string(), + }); + durable.priority = HealPriority::Low; + assert_eq!( + manager + .submit_heal_request(durable.clone()) + .await + .expect("low-priority admin bucket should queue durably"), + HealAdmissionResult::Accepted + ); + + let mut urgent = HealRequest::new( + HealType::Object { + bucket: "other".to_string(), + object: "object".to_string(), + version_id: None, + }, + HealOptions::default(), + HealPriority::Urgent, + ); + urgent.source = HealRequestSource::Internal; + assert_eq!( + manager + .submit_heal_request(urgent) + .await + .expect("durable admin owner cannot be displaced"), + HealAdmissionResult::Full + ); + assert_eq!( + manager + .root_recovery + .pending() + .await + .expect("read durable bucket") + .iter() + .map(|request| request.id.as_str()) + .collect::>(), + [durable.id.as_str()] + ); +} + +#[tokio::test] +async fn root_recovery_legacy_schema_replays_as_cluster() { + #[derive(serde::Serialize)] + struct LegacyRootHealIntent<'a> { + schema: u32, + task_id: &'a str, + options: &'a HealOptions, + priority: HealPriority, + retry_attempts: u32, + created_at: SystemTime, + } + + let (_temp, disk) = recovery_disk().await; + let request = root_request(); + let path = format!("root-heal-{}.json", request.id); + let bytes = serde_json::to_vec(&LegacyRootHealIntent { + schema: 1, + task_id: &request.id, + options: &request.options, + priority: request.priority, + retry_attempts: request.retry_attempts, + created_at: request.created_at, + }) + .expect("legacy root recovery JSON"); + disk.write_all(RUSTFS_META_BUCKET, &path, bytes.into()) + .await + .expect("write legacy root record"); + + let restarted = recovery_manager(vec![disk]); + restarted.replay_root_heals().await.expect("replay legacy root record"); + let queued = restarted.heal_queue.lock().await.requests().cloned().collect::>(); + assert_eq!(queued.len(), 1); + assert_eq!(queued[0].id, request.id); + assert_eq!(queued[0].heal_type, HealType::Cluster); +} + +#[tokio::test] +async fn root_recovery_rejected_admin_start_does_not_persist() { + let (_temp, disk) = recovery_disk().await; + let manager = recovery_manager(vec![disk.clone()]); + manager.config.write().await.queue_size = 0; + let request = root_request(); + + let receipt = manager + .submit_heal_request_with_receipt(request.clone()) + .await + .expect("full admission reports a receipt"); + assert_eq!(receipt.result, HealAdmissionResult::Full); + assert!(manager.root_recovery.pending().await.expect("read durable state").is_empty()); + drop(manager); + + let restarted = recovery_manager(vec![disk]); + restarted.replay_root_heals().await.expect("replay empty durable state"); + assert_eq!(restarted.get_queue_length().await, 0); +} + +#[tokio::test] +async fn root_recovery_queued_owner_is_not_priority_displaced() { + let (_temp, disk) = recovery_disk().await; + let manager = recovery_manager(vec![disk.clone()]); + manager.config.write().await.queue_size = 1; + let mut root = root_request(); + root.priority = HealPriority::Low; + assert_eq!( + manager + .submit_heal_request(root.clone()) + .await + .expect("low-priority root should queue"), + HealAdmissionResult::Accepted + ); + + let mut bucket = HealRequest::new( + HealType::Bucket { + bucket: "bucket".to_string(), + }, + HealOptions::default(), + HealPriority::Urgent, + ); + bucket.source = HealRequestSource::Admin; + assert_eq!( + manager + .submit_heal_request(bucket) + .await + .expect("durable root owner cannot be displaced"), + HealAdmissionResult::Full + ); + let pending = manager.root_recovery.pending().await.expect("read durable root"); + assert_eq!(pending.iter().map(|request| request.id.as_str()).collect::>(), [root.id.as_str()]); + let queued = manager + .heal_queue + .lock() + .await + .requests() + .map(|request| request.id.clone()) + .collect::>(); + assert_eq!(queued, [root.id]); +} + #[tokio::test] async fn root_recovery_explicit_cancel_covers_active_queued_retrying_and_durable_only() { for state in ["active", "queued", "retrying", "durable_only", "root_path"] { @@ -159,7 +707,17 @@ async fn root_recovery_force_start_cancels_durable_only_responsibility() { .expect("force start replacement"), HealAdmissionResult::Accepted ); - assert!(manager.root_recovery.pending().await.expect("old owner retired").is_empty()); + assert_eq!( + manager + .root_recovery + .pending() + .await + .expect("new owner retained") + .iter() + .map(|request| request.id.as_str()) + .collect::>(), + [new.id.as_str()] + ); manager.stop().await.expect("persist new root only"); let restarted = recovery_manager(vec![disk]); restarted.replay_root_heals().await.expect("restart replacement"); @@ -173,6 +731,58 @@ async fn root_recovery_force_start_cancels_durable_only_responsibility() { assert_eq!(ids, [new.id]); } +#[tokio::test] +async fn root_recovery_force_start_preserves_disjoint_durable_only_admin_work() { + let (_temp, disk) = recovery_disk().await; + let manager = recovery_manager(vec![disk.clone()]); + let old_overlap = admin_request(HealType::Bucket { + bucket: "overlap".to_string(), + }); + let old_disjoint = admin_request(HealType::Bucket { + bucket: "disjoint".to_string(), + }); + manager + .root_recovery + .persist(&old_overlap) + .await + .expect("durable overlapping bucket"); + manager + .root_recovery + .persist(&old_disjoint) + .await + .expect("durable disjoint bucket"); + + let mut replacement = admin_request(HealType::Bucket { + bucket: "overlap".to_string(), + }); + replacement.force_start = true; + assert_eq!( + manager + .submit_heal_request(replacement.clone()) + .await + .expect("forceStart replaces overlapping durable owner"), + HealAdmissionResult::Accepted + ); + drop(manager); + + let restarted = recovery_manager(vec![disk]); + restarted + .replay_root_heals() + .await + .expect("restart after selective forceStart"); + let mut ids = restarted + .heal_queue + .lock() + .await + .requests() + .map(|request| request.id.clone()) + .collect::>(); + ids.sort(); + let mut expected = vec![old_disjoint.id, replacement.id]; + expected.sort(); + assert_eq!(ids, expected); +} + #[tokio::test] async fn root_recovery_force_start_replaces_fresh_queued_and_retrying_admin_roots() { for retrying in [false, true] { @@ -184,7 +794,28 @@ async fn root_recovery_force_start_replaces_fresh_queued_and_retrying_admin_root } else { manager.submit_heal_request(old.clone()).await.expect("queue original root"); } - assert!(manager.root_recovery.pending().await.expect("not handed off yet").is_empty()); + if retrying { + assert!( + manager + .root_recovery + .pending() + .await + .expect("retrying not handed off yet") + .is_empty() + ); + } else { + assert_eq!( + manager + .root_recovery + .pending() + .await + .expect("queued root is durable immediately") + .iter() + .map(|request| request.id.as_str()) + .collect::>(), + [old.id.as_str()] + ); + } let mut new = root_request(); new.force_start = true; assert_eq!( @@ -222,7 +853,7 @@ async fn root_recovery_invalid_records_are_retained_without_partial_replay() { let original = disk.read_all(RUSTFS_META_BUCKET, &path).await.expect("read root record"); let mut value: serde_json::Value = serde_json::from_slice(&original).expect("record JSON"); match kind { - "schema" => value["schema"] = 2.into(), + "schema" => value["schema"] = 3.into(), "identity" => value["task_id"] = valid.id.clone().into(), "option" => value["options"]["future_delete_mode"] = true.into(), "no_lock" => value["options"]["no_lock"] = true.into(), @@ -448,18 +1079,15 @@ async fn root_recovery_terminal_timeout_updates_only_existing_journal_before_sec .await .expect("second restart after terminal timeout"); let queue = restarted.heal_queue.lock().await; - if durable { - assert_eq!( - queue - .requests() - .next() - .expect("remaining timeout responsibility") - .options - .timeout, - Some(Duration::ZERO) - ); - } else { - assert!(queue.is_empty(), "terminal failure must not create a new durable responsibility"); - } + assert_eq!( + queue + .requests() + .next() + .expect("remaining timeout responsibility") + .options + .timeout, + Some(Duration::ZERO), + "durable={durable}" + ); } }