From 32c4593587f7d4e2a7639a89b0c62aa577778fb5 Mon Sep 17 00:00:00 2001 From: Chris An Date: Sun, 25 Oct 2026 03:58:46 +0800 Subject: [PATCH] fix(heal): retain MRF state when bucket identity is unresolved (#8905) * fix(heal): retain MRF state when bucket identity is unresolved * test(heal): give real-storage replay fixture a bounded stack --- crates/heal/src/error.rs | 3 + crates/heal/src/heal/manager.rs | 20 +- crates/heal/src/heal/mrf_queue.rs | 175 +++++++-- .../heal/src/heal/mrf_queue/partial_write.rs | 141 +++++-- .../tests/unknown_source_tests.rs | 370 ++++++++++++++++++ .../mrf_queue/tests/empty_events_tests.rs | 4 +- .../mrf_queue/tests/lifecycle_replay_tests.rs | 258 ++++++++++++ crates/heal/src/heal/storage.rs | 12 +- crates/heal/tests/mrf_partial_write_test.rs | 2 +- docs/README.md | 2 + docs/operations/mrf-responsibilities.md | 33 ++ rustfs/src/admin/handlers/heal.rs | 33 +- 12 files changed, 970 insertions(+), 83 deletions(-) create mode 100644 crates/heal/src/heal/mrf_queue/partial_write/tests/unknown_source_tests.rs create mode 100644 crates/heal/src/heal/mrf_queue/tests/lifecycle_replay_tests.rs create mode 100644 docs/operations/mrf-responsibilities.md diff --git a/crates/heal/src/error.rs b/crates/heal/src/error.rs index ff366f1b4..be6d91406 100644 --- a/crates/heal/src/error.rs +++ b/crates/heal/src/error.rs @@ -85,6 +85,9 @@ pub enum Error { #[error("stale_bucket_incarnation: bucket {bucket} no longer belongs to this heal admission ({expected:?})")] StaleBucketIncarnation { bucket: String, expected: Option }, + #[error("Bucket identity lookup returned a nil incarnation")] + InvalidBucketIncarnation, + /// The current page already exhausted its local retry budget. Retrying /// the enclosing bucket would replay pages whose results were counted. #[error("Heal listing failed for bucket {bucket}: {source}")] diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 3bcd3cfd7..8ca858b72 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -1990,15 +1990,25 @@ impl HealManager { pub(crate) async fn durable_mrf_repair_anchor( &self, intent: &rustfs_common::mrf_channel::MrfIntent, - ) -> Option { + ) -> Result> { if let Some(purge) = intent.delete_marker_purge.as_ref() { - return rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(intent, purge.bucket_incarnation_id); + return Ok(rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent( + intent, + purge.bucket_incarnation_id, + )); } - match self.storage.mrf_bucket_incarnation_id(intent.bucket.as_ref()).await { - Ok(Some(bucket_incarnation_id)) => { + Ok(match self.mrf_bucket_incarnation_id(intent.bucket.as_ref()).await? { + Some(bucket_incarnation_id) => { rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(intent, bucket_incarnation_id) } - Ok(None) | Err(_) => None, + None => None, + }) + } + + pub(crate) async fn mrf_bucket_incarnation_id(&self, bucket: &str) -> Result> { + match self.storage.mrf_bucket_incarnation_id(bucket).await? { + Some(incarnation) if incarnation.is_nil() => Err(Error::InvalidBucketIncarnation), + identity => Ok(identity), } } diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs index 896150ad3..4a3fafac1 100644 --- a/crates/heal/src/heal/mrf_queue.rs +++ b/crates/heal/src/heal/mrf_queue.rs @@ -58,12 +58,18 @@ static GLOBAL_MRF_LIFECYCLE_SENDER: std::sync::OnceLock), #[error("operator action is invalid: {0}")] InvalidAction(String), #[error("the lifecycle checkpoint could not be persisted")] @@ -115,7 +121,7 @@ pub async fn list_legacy_responsibilities( .send(MrfLifecycleCommand::List { after, limit, response }) .await .map_err(|_| MrfLifecycleControlError::Unavailable)?; - receiver.await.map_err(|_| MrfLifecycleControlError::Unavailable) + receiver.await.map_err(|_| MrfLifecycleControlError::Unavailable)? } pub async fn recheck_legacy_responsibility( @@ -172,7 +178,7 @@ enum MrfLifecycleCommand { List { after: Option, limit: usize, - response: tokio::sync::oneshot::Sender, + response: tokio::sync::oneshot::Sender>, }, Recheck { responsibility_id: Uuid, @@ -615,6 +621,16 @@ struct MrfLifecyclePayload { records: Vec, } +#[derive(Debug, thiserror::Error)] +enum MrfReplayError { + #[error(transparent)] + Snapshot(#[from] snapshot::SnapshotError), + #[error("MRF lifecycle checkpoint is invalid")] + InvalidLifecycle, + #[error("MRF lifecycle checkpoint could not be decoded")] + Lifecycle(#[source] serde_json::Error), +} + fn encode_mrf_lifecycle_checkpoint( owner: Uuid, sequence: u64, @@ -660,24 +676,25 @@ fn encode_mrf_lifecycle_checkpoint( Some(encoded) } -fn decode_mrf_lifecycle_checkpoint(data: &[u8], byte_limit: usize) -> Option { +fn decode_mrf_lifecycle_checkpoint(data: &[u8], byte_limit: usize) -> Result { const HEADER_LEN: usize = 8 + 1 + 8 + 32; if data.len() < HEADER_LEN || data.len() > byte_limit || &data[..8] != MRF_LIFECYCLE_MAGIC { - return None; + return Err(MrfReplayError::InvalidLifecycle); } if data[8] != MRF_LIFECYCLE_FORMAT_VERSION { - return None; + return Err(snapshot::SnapshotError::Unsupported.into()); } - let payload_len = usize::try_from(u64::from_le_bytes(data[9..17].try_into().ok()?)).ok()?; + let payload_len = usize::try_from(u64::from_le_bytes(data[9..17].try_into().map_err(|_| MrfReplayError::InvalidLifecycle)?)) + .map_err(|_| MrfReplayError::InvalidLifecycle)?; if payload_len != data.len().saturating_sub(HEADER_LEN) { - return None; + return Err(MrfReplayError::InvalidLifecycle); } let payload = &data[HEADER_LEN..]; let digest: [u8; 32] = sha2::Sha256::digest(payload).into(); if digest != data[17..49] { - return None; + return Err(MrfReplayError::InvalidLifecycle); } - let decoded: MrfLifecyclePayload = serde_json::from_slice(payload).ok()?; + let decoded: MrfLifecyclePayload = serde_json::from_slice(payload).map_err(MrfReplayError::Lifecycle)?; if decoded.format_version != MRF_LIFECYCLE_FORMAT_VERSION || decoded.checkpoint_owner.is_nil() || decoded.checkpoint_sequence == 0 @@ -698,9 +715,9 @@ fn decode_mrf_lifecycle_checkpoint(data: &[u8], byte_limit: usize) -> Option Option<[u8; 32]> { @@ -754,8 +771,8 @@ async fn read_mrf_lifecycle_checkpoint( sequence: u64, checkpoint_max_bytes: usize, byte_limit: usize, -) -> Option> { - let bytes = snapshot::read_committed_companion( +) -> Result>, MrfReplayError> { + let Some(bytes) = snapshot::read_committed_companion( &journal_disks().await, owner, sequence, @@ -763,10 +780,15 @@ async fn read_mrf_lifecycle_checkpoint( checkpoint_max_bytes, byte_limit, ) - .await - .ok()??; + .await? + else { + return Ok(None); + }; let decoded = decode_mrf_lifecycle_checkpoint(&bytes, byte_limit)?; - (decoded.checkpoint_owner == owner && decoded.checkpoint_sequence == sequence).then_some(decoded.records) + if decoded.checkpoint_owner != owner || decoded.checkpoint_sequence != sequence { + return Err(MrfReplayError::InvalidLifecycle); + } + Ok(Some(decoded.records)) } /// Write the snapshot to every local disk; returns true when at least one @@ -944,6 +966,7 @@ struct MrfRuntime { /// True when replay observed a responsibility that cannot be discharged by /// a complete verified repair proof in this process. retain_replay_journal: bool, + replay_readable: bool, /// Durable responsibilities accepted from replay and waiting for an /// exact storage-owned proof before the startup journal can be deleted. durable_replay_anchors: Vec, @@ -999,6 +1022,9 @@ impl MrfRuntime { intent: MrfIntent, source_bucket_incarnation_id: Option, ) -> Result<(), MrfDurableAdmissionError> { + if !self.replay_readable { + return Err(MrfDurableAdmissionError::Persistence); + } self.partial_writes.admit_with_source_incarnation( intent, source_bucket_incarnation_id, @@ -1009,10 +1035,17 @@ impl MrfRuntime { Ok(()) } - fn list_unverified_legacy(&self, after: Option, limit: usize) -> MrfLegacyResponsibilitySnapshot { + fn list_unverified_legacy( + &self, + after: Option, + limit: usize, + ) -> Result { + if !self.replay_readable { + return Err(MrfLifecycleControlError::CheckpointUnreadable); + } let (responsibilities, next_cursor) = self.partial_writes.list_unverified_legacy(after, limit); let lifecycle = self.partial_writes.lifecycle_summary(); - MrfLegacyResponsibilitySnapshot { + Ok(MrfLegacyResponsibilitySnapshot { contract_version: 2, node_local: true, checkpoint_owner: self.runtime_checkpoint.map(|(owner, _)| owner), @@ -1023,7 +1056,7 @@ impl MrfRuntime { operator_accepted_count: lifecycle.operator_accepted_count, responsibilities, next_cursor, - } + }) } async fn verify_legacy_generation_incarnation( @@ -1032,6 +1065,9 @@ impl MrfRuntime { responsibility_id: Uuid, expected_bucket_incarnation_id: Uuid, ) -> Result { + if !self.replay_readable { + return Err(MrfLifecycleControlError::CheckpointUnreadable); + } let (intent, state, source_bucket_incarnation_id) = self .partial_writes .intent_for_responsibility(responsibility_id) @@ -1047,7 +1083,8 @@ impl MrfRuntime { let anchor = manager .durable_mrf_repair_anchor(&intent) .await - .ok_or(MrfLifecycleControlError::IncarnationChanged)?; + .map_err(|error| MrfLifecycleControlError::BucketIdentityUnavailable(Box::new(error)))? + .ok_or(MrfLifecycleControlError::BucketIdentityUnresolved)?; if anchor.bucket_incarnation_id != expected_bucket_incarnation_id { return Err(MrfLifecycleControlError::IncarnationChanged); } @@ -1060,6 +1097,9 @@ impl MrfRuntime { responsibility_id: Uuid, expected_bucket_incarnation_id: Uuid, ) -> Result<(), MrfLifecycleControlError> { + if !self.replay_readable { + return Err(MrfLifecycleControlError::CheckpointUnreadable); + } if self.partial_writes.is_active_responsibility(responsibility_id) { return Ok(()); } @@ -1088,6 +1128,9 @@ impl MrfRuntime { responsibility_id: Uuid, expected_bucket_incarnation_id: Option, ) -> Result<(), MrfLifecycleControlError> { + if !self.replay_readable { + return Err(MrfLifecycleControlError::CheckpointUnreadable); + } let (intent, state, source_incarnation) = self .partial_writes .intent_for_responsibility(responsibility_id) @@ -1096,14 +1139,15 @@ impl MrfRuntime { return Err(MrfLifecycleControlError::StaleGeneration); } if expected_bucket_incarnation_id.is_none() - && (source_incarnation.is_some() || !matches!(state, partial_write::ResponsibilityState::Active)) + && (source_incarnation.is_some() || !matches!(state, partial_write::ResponsibilityState::Active {})) { return Err(MrfLifecycleControlError::StaleGeneration); } let anchor = manager .durable_mrf_repair_anchor(&intent) .await - .ok_or(MrfLifecycleControlError::IncarnationChanged)?; + .map_err(|error| MrfLifecycleControlError::BucketIdentityUnavailable(Box::new(error)))? + .ok_or(MrfLifecycleControlError::BucketIdentityUnresolved)?; if Some(anchor.bucket_incarnation_id) == expected_bucket_incarnation_id { return Ok(()); } @@ -1133,6 +1177,9 @@ impl MrfRuntime { manager: &HealManager, request: MrfLegacyRiskAcceptanceRequest, ) -> Result<(), MrfLifecycleControlError> { + if !self.replay_readable { + return Err(MrfLifecycleControlError::CheckpointUnreadable); + } let responsibility_id = request.responsibility_id; if self .partial_writes @@ -1186,6 +1233,11 @@ impl MrfRuntime { } async fn flush(&mut self) -> bool { + // A new snapshot or compatibility mirror must never replace an owner + // whose lifecycle records this process could not read completely. + if !self.replay_readable { + return false; + } let (authoritative, legacy) = self.snapshot(); let lifecycle_byte_limit = self.config.journal_max_bytes.saturating_mul(4); let lifecycle_checkpoint = if self.partial_writes.depth() == 0 { @@ -1286,6 +1338,9 @@ impl MrfRuntime { } async fn dispatch(&mut self, manager: &HealManager) { + if !self.replay_readable { + return; + } self.dispatch_partial_writes(manager).await; if let Some(until) = self.backoff_until { if tokio::time::Instant::now() < until { @@ -1357,6 +1412,9 @@ impl MrfRuntime { } async fn delete_idle_recovery_anchors(&mut self) -> bool { + if !self.replay_readable { + return false; + } let runtime_deleted = match self.runtime_checkpoint { Some((owner, sequence)) => { match snapshot::delete_committed_snapshots_through(owner, sequence, self.config.journal_max_bytes).await { @@ -1522,6 +1580,7 @@ pub async fn replay_journal_once(manager: &Arc) -> usize { dirty: false, journal_on_disk: replay.journal_on_disk, retain_replay_journal: replay.retain_journal_for_replay, + replay_readable: replay.source_readable, durable_replay_anchors: replay.durable_replay_anchors, replay_cleanup: replay.cleanup, runtime_checkpoint: None, @@ -1537,6 +1596,7 @@ struct ReplayOutcome { replayed: usize, journal_on_disk: bool, retain_journal_for_replay: bool, + source_readable: bool, durable_replay_anchors: Vec, partial_writes: Vec<(MrfIntent, Option)>, cleanup: Option, @@ -1559,7 +1619,8 @@ enum ReplayCleanup { } struct ReplaySource { - data: Vec, + intents: Vec, + truncated_bytes: usize, cleanup: ReplayCleanup, lifecycle: HashMap<[u8; 32], VecDeque>, } @@ -1576,22 +1637,38 @@ fn take_lifecycle_record( record } -async fn read_replay_source( - max_bytes: usize, - lifecycle_max_bytes: usize, -) -> Result, snapshot::SnapshotError> { +async fn read_replay_source(max_bytes: usize, lifecycle_max_bytes: usize) -> Result, MrfReplayError> { if let Some(committed) = snapshot::inspect_local_committed_snapshot(max_bytes).await? { let owner = committed.owner(); let sequence = committed.sequence(); - let records = read_mrf_lifecycle_checkpoint(owner, sequence, max_bytes, lifecycle_max_bytes) - .await - .unwrap_or_default(); + let (intents, truncated_bytes) = decode_journal(committed.payload()); + let records = read_mrf_lifecycle_checkpoint(owner, sequence, max_bytes, lifecycle_max_bytes).await?; + if let Some(records) = records.as_ref() { + // Validate the complete owner before rearming or submitting any + // intent. Different source generations may share an intent digest. + let mut counts = HashMap::new(); + for intent in intents.iter().filter(|intent| intent.kind.is_durable()) { + let digest = intent_digest(intent).ok_or(MrfReplayError::InvalidLifecycle)?; + *counts.entry(digest).or_insert(0usize) += 1; + } + for record in records { + match counts.get_mut(&record.intent_digest) { + Some(count) if *count > 0 => *count -= 1, + _ => return Err(MrfReplayError::InvalidLifecycle), + } + } + if counts.values().any(|count| *count != 0) { + return Err(MrfReplayError::InvalidLifecycle); + } + } + let records = records.unwrap_or_default(); let mut lifecycle: HashMap<[u8; 32], VecDeque> = HashMap::with_capacity(records.len()); for record in records { lifecycle.entry(record.intent_digest).or_default().push_back(record); } return Ok(Some(ReplaySource { - data: committed.payload().to_vec(), + intents, + truncated_bytes, cleanup: ReplayCleanup::Committed { owner, sequence }, lifecycle, })); @@ -1606,8 +1683,10 @@ async fn read_replay_source( None => return Ok(None), }, }; + let (intents, truncated_bytes) = decode_journal(&data); Ok(Some(ReplaySource { - data, + intents, + truncated_bytes, cleanup: ReplayCleanup::Legacy, lifecycle: HashMap::new(), })) @@ -1649,6 +1728,7 @@ async fn replay_into( replayed: 0, journal_on_disk: false, retain_journal_for_replay: false, + source_readable: true, durable_replay_anchors: Vec::new(), partial_writes: Vec::new(), cleanup: None, @@ -1665,6 +1745,7 @@ async fn replay_into( replayed: 0, journal_on_disk: true, retain_journal_for_replay: true, + source_readable: false, durable_replay_anchors: Vec::new(), partial_writes: Vec::new(), cleanup: None, @@ -1677,11 +1758,10 @@ async fn replay_into( ReplayCleanup::Legacy => 1, ReplayCleanup::Committed { sequence, .. } => sequence.saturating_add(1), }; - let data = source.data; let mut lifecycle = source.lifecycle; - let (decoded, truncated) = decode_journal(&data); - let replayed = decoded.len(); - let intents = decoded; + let intents = source.intents; + let truncated = source.truncated_bytes; + let replayed = intents.len(); if truncated > 0 { tracing::warn!( target: "rustfs::heal::mrf", @@ -1712,7 +1792,7 @@ async fn replay_into( rustfs_common::mrf_channel::release_mrf_intent(&intent); continue; } - if let Some(anchor) = manager.durable_mrf_repair_anchor(&intent).await { + if let Ok(Some(anchor)) = manager.durable_mrf_repair_anchor(&intent).await { durable_replay_anchors.push(anchor); } let restored = intent_digest(&intent).and_then(|digest| take_lifecycle_record(&mut lifecycle, digest)); @@ -1750,7 +1830,7 @@ async fn replay_into( // Adopt durable replay into its single checkpointed owner. The // runtime dispatches it after publishing the successor, avoiding // a duplicate task in the startup-replay and steady-state paths. - if let Some(anchor) = manager.durable_mrf_repair_anchor(&intent).await { + if let Ok(Some(anchor)) = manager.durable_mrf_repair_anchor(&intent).await { durable_replay_anchors.push(anchor.clone()); } let restored = intent_digest(&intent).and_then(|digest| take_lifecycle_record(&mut lifecycle, digest)); @@ -1759,7 +1839,7 @@ async fn replay_into( } match submit_mrf_heal_request(manager, &intent, None).await { Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => { - if let Some(anchor) = manager.durable_mrf_repair_anchor(&intent).await { + if let Ok(Some(anchor)) = manager.durable_mrf_repair_anchor(&intent).await { durable_replay_anchors.push(anchor); } else { accepted_without_durable_anchor = true; @@ -1811,6 +1891,7 @@ async fn replay_into( replayed, journal_on_disk, retain_journal_for_replay, + source_readable: true, durable_replay_anchors, partial_writes, cleanup: journal_on_disk.then_some(cleanup), @@ -1837,6 +1918,7 @@ async fn run_mrf_consumer( dirty: false, journal_on_disk: false, retain_replay_journal: false, + replay_readable: true, durable_replay_anchors: Vec::new(), replay_cleanup: None, runtime_checkpoint: None, @@ -1848,6 +1930,7 @@ async fn run_mrf_consumer( let replay = replay_into(&manager, &mut runtime.queue, &mut runtime.backoff_until).await; runtime.journal_on_disk = replay.journal_on_disk; runtime.retain_replay_journal = replay.retain_journal_for_replay; + runtime.replay_readable = replay.source_readable; runtime.durable_replay_anchors = replay.durable_replay_anchors; runtime.replay_cleanup = replay.cleanup; runtime.next_checkpoint_sequence = replay.next_checkpoint_sequence; @@ -2064,6 +2147,7 @@ mod tests { use std::time::{Duration as StdDuration, Instant}; mod empty_events_tests; + mod lifecycle_replay_tests; #[tokio::test(start_paused = true)] async fn journal_disk_io_bounds_fanout_and_collects_every_result() { @@ -2232,8 +2316,8 @@ mod tests { let mut torn = encoded.clone(); *torn.last_mut().expect("non-empty checkpoint") ^= 0xff; - assert!(decode_mrf_lifecycle_checkpoint(&torn, 8192).is_none()); - assert!(decode_mrf_lifecycle_checkpoint(&encoded, encoded.len() - 1).is_none()); + assert!(decode_mrf_lifecycle_checkpoint(&torn, 8192).is_err()); + assert!(decode_mrf_lifecycle_checkpoint(&encoded, encoded.len() - 1).is_err()); assert!(encode_mrf_lifecycle_checkpoint(Uuid::nil(), 42, Vec::new(), 8192).is_none()); } @@ -2362,6 +2446,7 @@ mod tests { dirty: false, journal_on_disk: replay.journal_on_disk, retain_replay_journal: replay.retain_journal_for_replay, + replay_readable: replay.source_readable, durable_replay_anchors: replay.durable_replay_anchors, replay_cleanup: replay.cleanup, runtime_checkpoint: None, @@ -2501,6 +2586,7 @@ mod tests { dirty: true, journal_on_disk: false, retain_replay_journal: false, + replay_readable: true, durable_replay_anchors: Vec::new(), replay_cleanup: None, runtime_checkpoint: None, @@ -3093,6 +3179,7 @@ mod tests { dirty: false, journal_on_disk: true, retain_replay_journal: false, + replay_readable: true, durable_replay_anchors: vec![anchor], replay_cleanup: Some(cleanup), runtime_checkpoint: None, @@ -3144,6 +3231,7 @@ mod tests { dirty: false, journal_on_disk: true, retain_replay_journal: false, + replay_readable: true, durable_replay_anchors: Vec::new(), replay_cleanup: None, runtime_checkpoint: None, @@ -3192,6 +3280,7 @@ mod tests { dirty: false, journal_on_disk: true, retain_replay_journal: false, + replay_readable: true, durable_replay_anchors: Vec::new(), replay_cleanup: Some(ReplayCleanup::Committed { owner: replay_owner, @@ -3306,7 +3395,7 @@ mod tests { responsibility_id: Uuid::new_v4(), source_bucket_incarnation_id: Some(source_bucket_incarnation_id), last_operator_acceptance: None, - state: partial_write::ResponsibilityState::Active, + state: partial_write::ResponsibilityState::Active {}, }], lifecycle_limit, ) @@ -3366,6 +3455,7 @@ mod tests { dirty: false, journal_on_disk: replay.journal_on_disk, retain_replay_journal: replay.retain_journal_for_replay, + replay_readable: replay.source_readable, durable_replay_anchors: replay.durable_replay_anchors, replay_cleanup: replay.cleanup, runtime_checkpoint: None, @@ -3613,6 +3703,7 @@ mod tests { dirty: true, journal_on_disk: false, retain_replay_journal: false, + replay_readable: true, durable_replay_anchors: Vec::new(), replay_cleanup: None, runtime_checkpoint: None, diff --git a/crates/heal/src/heal/mrf_queue/partial_write.rs b/crates/heal/src/heal/mrf_queue/partial_write.rs index 08028e5e5..ca1c8ab17 100644 --- a/crates/heal/src/heal/mrf_queue/partial_write.rs +++ b/crates/heal/src/heal/mrf_queue/partial_write.rs @@ -34,6 +34,7 @@ const MAX_OPERATOR_REFERENCE_BYTES: usize = 256; const LIFECYCLE_CHECKPOINT_RECORD_OVERHEAD_BYTES: usize = 256; const MAX_UNVERIFIED_RETRY_BACKOFF: Duration = Duration::from_secs(60); const MRF_RETIREMENT_PROOF_RECHECK_INTERVAL: Duration = Duration::from_secs(5); +const MRF_IDENTITY_PROBE_TIMEOUT: Duration = Duration::from_secs(1); #[derive(Clone, Debug, PartialEq, Eq, Hash)] struct PartialWriteKey { @@ -67,7 +68,7 @@ impl PartialWriteKey { #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)] pub(super) enum ResponsibilityState { - Active, + Active {}, HeldUnverifiedLegacy { bucket_incarnation_id: Uuid, since_ms: u64, @@ -95,12 +96,12 @@ pub(super) enum ResponsibilityState { impl ResponsibilityState { fn is_parked(&self) -> bool { - !matches!(self, Self::Active) + !matches!(self, Self::Active {}) } fn estimated_bytes(&self) -> usize { match self { - Self::Active => 0, + Self::Active {} => 0, Self::HeldUnverifiedLegacy { .. } => std::mem::size_of::() + std::mem::size_of::(), Self::LegacyGenerationUnknown { .. } => std::mem::size_of::() + std::mem::size_of::(), Self::BucketIncarnationChanged { .. } => 2 * std::mem::size_of::() + std::mem::size_of::(), @@ -115,7 +116,7 @@ impl ResponsibilityState { pub(super) fn bucket_incarnation_id(&self) -> Option { match self { - Self::Active => None, + Self::Active {} => None, Self::HeldUnverifiedLegacy { bucket_incarnation_id, .. } @@ -156,7 +157,7 @@ impl ResponsibilityCheckpoint { return false; } match &self.state { - ResponsibilityState::Active => true, + ResponsibilityState::Active {} => true, ResponsibilityState::HeldUnverifiedLegacy { bucket_incarnation_id, .. } => !bucket_incarnation_id.is_nil() && self.source_bucket_incarnation_id == Some(*bucket_incarnation_id), @@ -207,11 +208,13 @@ struct Responsibility { // Process-local pacing does not alter the durable proof obligation. attempt_in_flight: bool, consecutive_unverified_attempts: u32, + last_identity_probe_result: Option<&'static str>, + last_identity_probe_at_ms: Option, } impl Responsibility { fn generation_unobserved(&self) -> bool { - self.source_bucket_incarnation_id.is_none() && matches!(self.state, ResponsibilityState::Active) + self.source_bucket_incarnation_id.is_none() && matches!(self.state, ResponsibilityState::Active {}) } fn is_parked(&self) -> bool { @@ -255,21 +258,24 @@ pub(super) struct PartialWrites { entries: HashMap, retry_order: VecDeque, retry_index: HashSet, + identity_probe_order: VecDeque, + identity_probe_index: HashSet, retirement_checks: HashMap<(Arc, Uuid), (Instant, bool)>, bytes: usize, } impl PartialWrites { pub(super) fn cost(intent: &MrfIntent) -> usize { - Self::cost_with_state(intent, &ResponsibilityState::Active) + Self::cost_with_state(intent, &ResponsibilityState::Active {}) } pub(super) fn cost_with_state(intent: &MrfIntent, state: &ResponsibilityState) -> usize { intent.estimated_bytes() + std::mem::size_of::() - + 2 * std::mem::size_of::() + + 4 * std::mem::size_of::() + LIFECYCLE_CHECKPOINT_RECORD_OVERHEAD_BYTES - + state.estimated_bytes() + // Reserve the observation transition before admitting an intent. + + state.estimated_bytes().max(std::mem::size_of::() + std::mem::size_of::()) } pub(super) fn cost_with_state_and_audit( @@ -326,6 +332,9 @@ impl PartialWrites { } retry_queued = true; } + if source_bucket_incarnation_id.is_none() && self.identity_probe_index.insert(key.clone()) { + self.identity_probe_order.push_back(key.clone()); + } self.entries.insert( key, Responsibility { @@ -335,11 +344,18 @@ impl PartialWrites { responsibility_id: Uuid::new_v4(), source_bucket_incarnation_id, last_operator_acceptance: None, - state: ResponsibilityState::Active, + state: ResponsibilityState::Active {}, retry_queued, - next_attempt: Instant::now(), + next_attempt: Instant::now() + + if source_bucket_incarnation_id.is_none() { + MAX_UNVERIFIED_RETRY_BACKOFF + } else { + Duration::ZERO + }, attempt_in_flight: false, consecutive_unverified_attempts: 0, + last_identity_probe_result: None, + last_identity_probe_at_ms: None, }, ); self.bytes = next_bytes; @@ -450,7 +466,7 @@ impl PartialWrites { let mut summary = PartialWriteLifecycleSummary::default(); for entry in self.entries.values() { match &entry.state { - ResponsibilityState::Active => { + ResponsibilityState::Active {} => { if entry.generation_unobserved() { summary.legacy_generation_unknown_count += 1; } @@ -501,7 +517,7 @@ impl PartialWrites { let Some(entry) = self.entries.get_mut(&key) else { return false; }; - if !matches!(&entry.state, ResponsibilityState::Active) { + if !matches!(&entry.state, ResponsibilityState::Active {}) { return false; } let old_cost = Self::entry_cost(entry); @@ -599,7 +615,7 @@ impl PartialWrites { ResponsibilityState::OperatorAcceptedUnverified { .. } => { return Err("a different operator disposition already exists"); } - ResponsibilityState::Active => return Err("responsibility is not held as unverified legacy"), + ResponsibilityState::Active {} => return Err("responsibility is not held as unverified legacy"), }; if current_incarnation != expected_bucket_incarnation_id { return Err("bucket incarnation changed; refresh the responsibility listing"); @@ -677,7 +693,7 @@ impl PartialWrites { } let previous = entry.state.clone(); let old_cost = Self::entry_cost(entry); - entry.state = ResponsibilityState::Active; + entry.state = ResponsibilityState::Active {}; self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry)); entry.persisted = false; entry.next_attempt = Instant::now(); @@ -725,8 +741,8 @@ impl PartialWrites { let mut selected = BTreeMap::new(); for entry in self.entries.values() { let (status, bucket_incarnation_id, held_since_ms) = match &entry.state { - ResponsibilityState::Active if entry.generation_unobserved() => ("legacy_generation_unobserved", None, None), - ResponsibilityState::Active => continue, + ResponsibilityState::Active {} if entry.generation_unobserved() => ("legacy_generation_unobserved", None, None), + ResponsibilityState::Active {} => continue, ResponsibilityState::HeldUnverifiedLegacy { bucket_incarnation_id, since_ms, @@ -772,6 +788,14 @@ impl PartialWrites { status, held_since_ms, accepted: entry.last_operator_acceptance.clone(), + last_identity_probe_result: entry.last_identity_probe_result, + last_identity_probe_at_ms: entry.last_identity_probe_at_ms, + next_identity_probe_at_ms: entry.generation_unobserved().then(|| { + unix_now_ms().saturating_add( + u64::try_from(entry.next_attempt.saturating_duration_since(Instant::now()).as_millis()) + .unwrap_or(u64::MAX), + ) + }), }; selected.insert(entry.responsibility_id, row); if selected.len() > limit + 1 { @@ -849,10 +873,10 @@ impl PartialWrites { config: &MrfConsumerConfig, ) -> bool { let now = Instant::now(); - let mut lifecycle_changed = false; + let mut lifecycle_changed = self.probe_unknown_generation(manager, config).await; let ready = self.ready_keys(now, config.replay_batch); if ready.is_empty() { - return false; + return lifecycle_changed; } let in_flight = manager.try_in_flight_durable_mrf_anchors(); for key in ready { @@ -873,7 +897,7 @@ impl PartialWrites { } // The bucket can be deleted and recreated while an obligation is // parked, so every dispatch must compare against a fresh identity. - if let Some(anchor) = manager.durable_mrf_repair_anchor(&entry.intent).await { + if let Ok(Some(anchor)) = manager.durable_mrf_repair_anchor(&entry.intent).await { entry.anchor = Some(anchor.clone()); if let Some(source_bucket_incarnation_id) = entry.source_bucket_incarnation_id && source_bucket_incarnation_id != anchor.bucket_incarnation_id @@ -909,6 +933,57 @@ impl PartialWrites { lifecycle_changed } + async fn probe_unknown_generation(&mut self, manager: &HealManager, config: &MrfConsumerConfig) -> bool { + if config.replay_batch == 0 { + return false; + } + let now = Instant::now(); + // At most one bounded identity lookup per dispatch round. These keys + // never enter the object-heal retry path or acquire a repair anchor. + for _ in 0..self.identity_probe_order.len().min(config.replay_batch) { + let Some(key) = self.identity_probe_order.pop_front() else { + break; + }; + self.identity_probe_index.remove(&key); + let Some(entry) = self.entries.get_mut(&key) else { + continue; + }; + if !entry.generation_unobserved() { + continue; + } + self.identity_probe_index.insert(key.clone()); + self.identity_probe_order.push_back(key); + if !entry.persisted || entry.next_attempt > now { + continue; + } + // Advance before awaiting so cancellation cannot cause a tight + // retry loop. Restart begins with the same conservative delay. + entry.defer_unverified_attempt(now, config.admission_backoff.max(MRF_RETIREMENT_PROOF_RECHECK_INTERVAL)); + let result = tokio::time::timeout( + MRF_IDENTITY_PROBE_TIMEOUT, + manager.mrf_bucket_incarnation_id(entry.intent.bucket.as_ref()), + ) + .await; + entry.last_identity_probe_at_ms = Some(unix_now_ms()); + match result { + Ok(Ok(Some(observed_bucket_incarnation_id))) => { + let old_cost = Self::entry_cost(entry); + entry.state = ResponsibilityState::LegacyGenerationUnknown { + observed_bucket_incarnation_id, + detected_at_ms: unix_now_ms(), + }; + entry.last_identity_probe_result = Some("known"); + self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry)); + return true; + } + Ok(Ok(None)) => entry.last_identity_probe_result = Some("unresolved"), + Ok(Err(_)) | Err(_) => entry.last_identity_probe_result = Some("unavailable"), + } + return false; + } + false + } + fn ready_keys(&mut self, now: Instant, limit: usize) -> Vec { let mut ready = Vec::with_capacity(limit.min(self.entries.len())); // Rotation prevents a permanently failing prefix from starving the @@ -956,6 +1031,8 @@ impl PartialWrites { self.bytes = self.entries.values().map(Self::entry_cost).sum(); self.retry_order.retain(|key| self.entries.contains_key(key)); self.retry_index.retain(|key| self.entries.contains_key(key)); + self.identity_probe_order.retain(|key| self.entries.contains_key(key)); + self.identity_probe_index.retain(|key| self.entries.contains_key(key)); true } } @@ -988,6 +1065,13 @@ pub struct MrfLegacyResponsibility { pub status: &'static str, pub held_since_ms: Option, pub accepted: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_identity_probe_result: Option<&'static str>, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_identity_probe_at_ms: Option, + /// Process-local schedule; replay restarts the conservative probe delay. + #[serde(skip_serializing_if = "Option::is_none")] + pub next_identity_probe_at_ms: Option, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -1021,6 +1105,7 @@ mod tests { use uuid::Uuid; mod retirement_tests; + mod unknown_source_tests; fn intent(object: &str) -> MrfIntent { let mut intent = MrfIntent { @@ -1148,7 +1233,7 @@ mod tests { fn lifecycle_summary_counts_mixed_states_without_changing_responsibilities() { let incarnation = Uuid::new_v4(); let writes = lifecycle_fixture(vec![ - ResponsibilityState::Active, + ResponsibilityState::Active {}, ResponsibilityState::HeldUnverifiedLegacy { bucket_incarnation_id: incarnation, since_ms: u64::MAX, @@ -1228,7 +1313,7 @@ mod tests { assert!(writes.retry_order.is_empty(), "unobserved legacy work must leave the hot retry queue"); let record = writes.checkpoint_records(|_| Some([7; 32])).remove(0); assert!(record.is_valid()); - assert!(matches!(record.state, ResponsibilityState::Active), "keep the legacy disk encoding"); + assert!(matches!(record.state, ResponsibilityState::Active {}), "keep the legacy disk encoding"); assert_eq!(record.source_bucket_incarnation_id, None); let (rows, next) = writes.list_unverified_legacy(None, 1); assert_eq!(rows.len(), 1); @@ -2009,7 +2094,7 @@ mod tests { .expect("current responsibility should be resident"); let current_id = current_entry.responsibility_id; assert_ne!(old_id, current_id); - assert!(matches!(current_entry.state, ResponsibilityState::Active)); + assert!(matches!(current_entry.state, ResponsibilityState::Active {})); assert!(current_entry.last_operator_acceptance.is_none()); assert!( !writes.park_unverified_legacy(&old_anchor), @@ -2017,7 +2102,7 @@ mod tests { ); assert!(matches!( writes.intent_for_responsibility(current_id), - Some((_, ResponsibilityState::Active, Some(source))) if source == current_incarnation + Some((_, ResponsibilityState::Active {}, Some(source))) if source == current_incarnation )); let mut journal = Vec::new(); @@ -2071,6 +2156,7 @@ mod tests { dirty: false, journal_on_disk: replay.journal_on_disk, retain_replay_journal: replay.retain_journal_for_replay, + replay_readable: replay.source_readable, durable_replay_anchors: replay.durable_replay_anchors, replay_cleanup: replay.cleanup, runtime_checkpoint: None, @@ -2085,7 +2171,7 @@ mod tests { )); assert!(matches!( runtime.partial_writes.intent_for_responsibility(current_id), - Some((_, ResponsibilityState::Active, Some(source))) if source == current_incarnation + Some((_, ResponsibilityState::Active {}, Some(source))) if source == current_incarnation )); assert!(runtime.partial_writes.last_operator_acceptance(current_id).is_none()); @@ -2178,7 +2264,7 @@ mod tests { assert_eq!(writes.lifecycle_summary().legacy_generation_unknown_count, 1); let original_unknown = writes.checkpoint_records(super::super::intent_digest).remove(0); assert_eq!(original_unknown.source_bucket_incarnation_id, None); - assert!(matches!(original_unknown.state, ResponsibilityState::Active)); + assert!(matches!(original_unknown.state, ResponsibilityState::Active {})); assert_eq!(writes.list_unverified_legacy(None, 10).0[0].status, "legacy_generation_unobserved"); let current = intent("same-object-unknown-source"); @@ -2232,6 +2318,7 @@ mod tests { dirty: false, journal_on_disk: replay.journal_on_disk, retain_replay_journal: replay.retain_journal_for_replay, + replay_readable: replay.source_readable, durable_replay_anchors: replay.durable_replay_anchors, replay_cleanup: replay.cleanup, runtime_checkpoint: None, @@ -2341,6 +2428,7 @@ mod tests { dirty: false, journal_on_disk: true, retain_replay_journal: false, + replay_readable: true, durable_replay_anchors: Vec::new(), replay_cleanup: None, runtime_checkpoint: None, @@ -2417,6 +2505,7 @@ mod tests { dirty: false, journal_on_disk: true, retain_replay_journal: false, + replay_readable: true, durable_replay_anchors: vec![anchor], replay_cleanup: None, runtime_checkpoint: None, diff --git a/crates/heal/src/heal/mrf_queue/partial_write/tests/unknown_source_tests.rs b/crates/heal/src/heal/mrf_queue/partial_write/tests/unknown_source_tests.rs new file mode 100644 index 000000000..4ecb93fe8 --- /dev/null +++ b/crates/heal/src/heal/mrf_queue/partial_write/tests/unknown_source_tests.rs @@ -0,0 +1,370 @@ +// Copyright 2026 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use super::*; +use crate::heal::mrf_queue::{MrfLifecycleControlError, MrfQueue, MrfRuntime}; +use crate::heal::storage::{HealListItem, HealObjectInfo}; +use crate::heal::storage_api::storage::{BucketInfo, DiskSetSelector}; +use crate::heal::{DiskStore, EcstoreError}; +use crate::{Error, Result}; +use rustfs_heal_contracts::heal_channel::HealOpts; +use rustfs_madmin::heal_commands::HealResultItem; +use std::sync::Mutex; +use std::sync::atomic::{AtomicUsize, Ordering}; + +#[derive(Clone, Copy)] +enum Lookup { + Identity(Option), + PermissionDenied, + Quorum, + Transport, + Pending, +} + +struct ProbeStorage { + lookup: Mutex, + calls: AtomicUsize, +} + +#[async_trait::async_trait] +impl HealStorageAPI for ProbeStorage { + async fn mrf_bucket_incarnation_id(&self, _: &str) -> Result> { + self.calls.fetch_add(1, Ordering::Relaxed); + let lookup = *self.lookup.lock().expect("lookup outcome"); + match lookup { + Lookup::Identity(value) => Ok(value), + Lookup::PermissionDenied => Err(Error::Io(std::io::ErrorKind::PermissionDenied.into())), + Lookup::Quorum => Err(Error::Storage(EcstoreError::ErasureReadQuorum)), + Lookup::Transport => Err(Error::Io(std::io::ErrorKind::ConnectionReset.into())), + Lookup::Pending => std::future::pending().await, + } + } + async fn get_object_meta(&self, _: &str, _: &str) -> Result> { + panic!("identity probing must not read objects") + } + async fn ec_decode_rebuild(&self, _: &str, _: &str) -> Result> { + panic!("identity probing must not rebuild objects") + } + async fn get_bucket_info(&self, _: &str) -> Result> { + panic!("identity probing must use the incarnation lookup") + } + async fn list_buckets(&self) -> Result> { + panic!("identity probing must not enumerate buckets") + } + async fn list_buckets_for_admin_heal(&self, _: Option) -> Result> { + panic!("identity probing must not start admin healing") + } + async fn list_objects_for_heal_page( + &self, + _: &str, + _: &str, + _: Option<&str>, + _: bool, + ) -> Result<(Vec, Option, bool)> { + panic!("identity probing must not enumerate objects") + } + async fn object_exists(&self, _: &str, _: &str) -> Result { + panic!("identity probing must not probe objects") + } + async fn heal_object(&self, _: &str, _: &str, _: Option<&str>, _: &HealOpts) -> Result<(HealResultItem, Option)> { + panic!("identity probing must not heal objects") + } + async fn heal_bucket(&self, _: &str, _: &HealOpts) -> Result { + panic!("identity probing must not heal buckets") + } + async fn heal_format(&self, _: bool) -> Result<(HealResultItem, Option)> { + panic!("identity probing must not heal formats") + } + async fn get_disk_for_resume(&self, _: &str) -> Result { + panic!("identity probing must not open resume disks") + } +} + +fn runtime() -> MrfRuntime { + let config = MrfConsumerConfig::default(); + MrfRuntime { + partial_writes: PartialWrites::default(), + queue: MrfQueue::new(config.queue_capacity, config.journal_max_bytes), + config, + checkpoint_owner: Uuid::new_v4(), + next_checkpoint_sequence: 1, + new_since_flush: 0, + dirty: false, + journal_on_disk: false, + retain_replay_journal: false, + replay_readable: true, + durable_replay_anchors: Vec::new(), + replay_cleanup: None, + runtime_checkpoint: None, + backoff_until: None, + } +} + +#[tokio::test(start_paused = true)] +async fn unknown_source_deleted_bucket_probes_preserve_errors_and_bound_retries() { + let storage = Arc::new(ProbeStorage { + lookup: Mutex::new(Lookup::Identity(None)), + calls: AtomicUsize::new(0), + }); + let manager = HealManager::new_without_root_recovery_for_test(storage.clone(), None); + let mut runtime = runtime(); + let original = intent("identity-probe"); + let key = PartialWriteKey::new(&original, None); + runtime + .admit_partial_write(original, None) + .expect("admit a retained obligation"); + runtime.partial_writes.mark_persisted(); + let checkpoint = runtime.partial_writes.checkpoint_records(super::super::super::intent_digest); + let id = checkpoint[0].responsibility_id; + let bytes = runtime.partial_writes.bytes(); + for outcome in [ + Lookup::Identity(None), + Lookup::PermissionDenied, + Lookup::Quorum, + Lookup::Transport, + Lookup::Identity(Some(Uuid::nil())), + Lookup::Pending, + ] { + *storage.lookup.lock().expect("set lookup outcome") = outcome; + runtime + .partial_writes + .entries + .get_mut(&key) + .expect("retained obligation") + .next_attempt = Instant::now(); + let calls = storage.calls.load(Ordering::Relaxed); + let start = Instant::now(); + assert!( + !runtime + .partial_writes + .dispatch_collecting_lifecycle_change(&manager, &runtime.config) + .await + ); + assert_eq!(storage.calls.load(Ordering::Relaxed), calls + 1); + assert!(start.elapsed() <= MRF_IDENTITY_PROBE_TIMEOUT); + let entry = runtime.partial_writes.entries.get(&key).expect("probe must retain its owner"); + assert!(entry.next_attempt > Instant::now()); + assert!(entry.next_attempt <= Instant::now() + MAX_UNVERIFIED_RETRY_BACKOFF); + assert!(entry.anchor.is_none(), "lookup failures cannot create a repair proof"); + assert_eq!(runtime.partial_writes.bytes(), bytes); + assert_eq!(runtime.partial_writes.checkpoint_records(super::super::super::intent_digest), checkpoint); + assert!( + !runtime + .partial_writes + .dispatch_collecting_lifecycle_change(&manager, &runtime.config) + .await + ); + assert_eq!( + storage.calls.load(Ordering::Relaxed), + calls + 1, + "a consumer tick cannot bypass probe backoff" + ); + let row = runtime + .list_unverified_legacy(None, 1) + .expect("list retained obligation") + .responsibilities + .remove(0); + assert_eq!(row.bucket_incarnation_id, None); + assert_eq!(row.source_bucket_incarnation_id, None); + assert_eq!( + row.last_identity_probe_result, + Some(if matches!(outcome, Lookup::Identity(None)) { + "unresolved" + } else { + "unavailable" + }) + ); + assert!(row.next_identity_probe_at_ms.is_some()); + + if !matches!(outcome, Lookup::Pending) { + let error = runtime + .refresh_legacy_generation(&manager, id, None) + .await + .expect_err("refresh needs a known identity"); + match (outcome, error) { + (Lookup::Identity(None), MrfLifecycleControlError::BucketIdentityUnresolved) => {} + (Lookup::PermissionDenied, MrfLifecycleControlError::BucketIdentityUnavailable(source)) => { + assert!(matches!(*source, Error::Io(ref error) if error.kind() == std::io::ErrorKind::PermissionDenied)); + } + (Lookup::Quorum, MrfLifecycleControlError::BucketIdentityUnavailable(source)) => { + assert!(matches!(*source, Error::Storage(EcstoreError::ErasureReadQuorum))); + } + (Lookup::Transport, MrfLifecycleControlError::BucketIdentityUnavailable(source)) => { + assert!(matches!(*source, Error::Io(ref error) if error.kind() == std::io::ErrorKind::ConnectionReset)); + } + (Lookup::Identity(Some(_)), MrfLifecycleControlError::BucketIdentityUnavailable(source)) => { + assert!(matches!(*source, Error::InvalidBucketIncarnation)); + } + (_, error) => panic!("lookup error lost its type: {error:?}"), + } + } + } + assert_eq!(manager.get_queue_length().await, 0); +} + +#[tokio::test(start_paused = true)] +async fn unknown_source_deleted_bucket_probe_rotation_keeps_count_and_byte_budgets() { + let current = Uuid::new_v4(); + let storage = Arc::new(ProbeStorage { + lookup: Mutex::new(Lookup::Identity(Some(current))), + calls: AtomicUsize::new(0), + }); + let manager = HealManager::new_without_root_recovery_for_test(storage.clone(), None); + let mut writes = PartialWrites::default(); + let config = MrfConsumerConfig { + replay_batch: 1, + ..Default::default() + }; + for object in ["rotation-a", "rotation-b", "rotation-c"] { + let item = intent(object); + let budget = writes.bytes() + PartialWrites::cost(&item); + writes.admit(item, 3, budget).expect("admit at the exact byte boundary"); + } + let bytes = writes.bytes(); + assert!(writes.admit(intent("overflow"), 3, bytes).is_err()); + writes.mark_persisted(); + for entry in writes.entries.values_mut() { + entry.next_attempt = Instant::now(); + } + for expected_calls in 1..=3 { + assert!(writes.dispatch_collecting_lifecycle_change(&manager, &config).await); + assert_eq!(storage.calls.load(Ordering::Relaxed), expected_calls); + assert_eq!(writes.bytes(), bytes, "observation must fit the budget reserved at admission"); + } + let (rows, _) = writes.list_unverified_legacy(None, 10); + assert_eq!(rows.len(), 3); + assert!( + rows.iter() + .all(|row| row.source_bucket_incarnation_id.is_none() && row.bucket_incarnation_id == Some(current)) + ); + for _ in 0..3 { + assert!(!writes.dispatch_collecting_lifecycle_change(&manager, &config).await); + } + assert_eq!(storage.calls.load(Ordering::Relaxed), 3, "observed unknown sources stay parked"); + assert_eq!(manager.get_queue_length().await, 0); +} + +#[test] +#[serial] +fn unknown_source_deleted_bucket_replay_probes_recreation_without_rebinding() { + // Match the bounded stack used by other deep real-storage heal fixtures. + const STACK_SIZE: usize = 8 * 1024 * 1024; + std::thread::Builder::new() + .name("mrf-unknown-source-replay".to_owned()) + .stack_size(STACK_SIZE) + .spawn(|| { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .thread_stack_size(STACK_SIZE) + .enable_all() + .build() + .expect("MRF replay test runtime"); + runtime.block_on(check_unknown_source_deleted_bucket_replay()); + }) + .expect("MRF replay test thread") + .join() + .expect("MRF replay test must finish"); +} + +async fn check_unknown_source_deleted_bucket_replay() { + let env = rustfs_test_utils::TestECStoreEnv::builder() + .prefix("rustfs_mrf_unknown_deleted_bucket") + .build() + .await; + let bucket = "partial-write-retention"; + env.make_bucket(bucket, false).await; + let storage: Arc = Arc::new(ECStoreHealStorage::new(env.ecstore.clone())); + let manager = Arc::new(HealManager::new_without_root_recovery_for_test(storage, None)); + let mut original = runtime(); + for index in 0..3 { + original + .admit_partial_write(intent(&format!("deleted-bucket-{index}")), None) + .expect("admit each unknown-source responsibility"); + } + assert!(original.flush().await, "commit responsibilities before deleting the bucket"); + let before = original.partial_writes.checkpoint_records(super::super::super::intent_digest); + env.ecstore + .delete_bucket(bucket, &DeleteBucketOptions::default()) + .await + .expect("delete the bucket through its supported API"); + let mut restarted = runtime(); + let replay = super::super::super::replay_into(&manager, &mut restarted.queue, &mut None).await; + assert_eq!(replay.partial_writes.len(), before.len()); + restarted.next_checkpoint_sequence = replay.next_checkpoint_sequence; + restarted.replay_cleanup = replay.cleanup; + restarted.retain_replay_journal = replay.retain_journal_for_replay; + restarted.durable_replay_anchors = replay.durable_replay_anchors; + restarted.adopt_replayed_partial_writes(replay.partial_writes); + restarted.dirty = true; + assert!(restarted.flush().await, "publish the retained owner after replay"); + restarted.dispatch(&manager).await; + + let mut visible = Vec::new(); + let mut cursor = None; + loop { + let page = restarted + .list_unverified_legacy(cursor, 1) + .expect("list readable responsibilities"); + assert_eq!(page.legacy_generation_unknown_count, before.len()); + visible.extend(page.responsibilities.into_iter().map(|row| { + assert_eq!(row.source_bucket_incarnation_id, None); + assert_eq!(row.bucket_incarnation_id, None); + row.responsibility_id + })); + cursor = page.next_cursor; + if cursor.is_none() { + break; + } + } + assert_eq!( + visible.into_iter().collect::>(), + before.iter().map(|row| row.responsibility_id).collect() + ); + + env.make_bucket(bucket, false).await; + let observed = env + .ecstore + .bucket_incarnation_id(bucket) + .await + .expect("recreated incarnation"); + for entry in restarted.partial_writes.entries.values_mut() { + entry.next_attempt = Instant::now(); + } + for _ in 0..before.len() { + restarted.dispatch(&manager).await; + } + let rows = restarted + .list_unverified_legacy(None, 10) + .expect("list observed responsibilities") + .responsibilities; + assert_eq!(rows.len(), before.len(), "identity probing cannot discharge a responsibility"); + for row in rows { + assert_eq!( + row.status, "legacy_generation_unknown", + "recreated buckets must be observed automatically" + ); + assert_eq!(row.bucket_incarnation_id, Some(observed)); + assert_eq!( + row.source_bucket_incarnation_id, None, + "current identity is never evidence of the old source" + ); + assert!( + restarted + .recheck_legacy_generation(&manager, row.responsibility_id, observed) + .await + .is_err() + ); + } + assert_eq!(manager.get_queue_length().await, 0, "identity probes must not enqueue object healing"); +} diff --git a/crates/heal/src/heal/mrf_queue/tests/empty_events_tests.rs b/crates/heal/src/heal/mrf_queue/tests/empty_events_tests.rs index 10ea382fe..41727d6d9 100644 --- a/crates/heal/src/heal/mrf_queue/tests/empty_events_tests.rs +++ b/crates/heal/src/heal/mrf_queue/tests/empty_events_tests.rs @@ -52,6 +52,7 @@ fn runtime() -> MrfRuntime { dirty: false, journal_on_disk: false, retain_replay_journal: false, + replay_readable: true, durable_replay_anchors: Vec::new(), replay_cleanup: None, runtime_checkpoint: None, @@ -425,7 +426,8 @@ fn empty_events_real_consumer_stop_retains_unresolved_journal() { let listing = tokio::time::timeout(Duration::from_secs(10), listed) .await .expect("consumer startup finishes") - .expect("listing receipt"); + .expect("listing receipt") + .expect("readable lifecycle snapshot"); assert!( listing .responsibilities diff --git a/crates/heal/src/heal/mrf_queue/tests/lifecycle_replay_tests.rs b/crates/heal/src/heal/mrf_queue/tests/lifecycle_replay_tests.rs new file mode 100644 index 000000000..6a94d95b8 --- /dev/null +++ b/crates/heal/src/heal/mrf_queue/tests/lifecycle_replay_tests.rs @@ -0,0 +1,258 @@ +// Copyright 2026 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use super::*; + +fn lifecycle_frame(payload: &Value) -> Vec { + let payload = serde_json::to_vec(payload).expect("encode lifecycle fixture"); + let mut encoded = Vec::from(MRF_LIFECYCLE_MAGIC.as_slice()); + encoded.push(MRF_LIFECYCLE_FORMAT_VERSION); + encoded.extend_from_slice(&u64::try_from(payload.len()).expect("fixture size").to_le_bytes()); + encoded.extend_from_slice(&sha2::Sha256::digest(&payload)); + encoded.extend_from_slice(&payload); + encoded +} + +#[test] +fn unknown_source_deleted_bucket_lifecycle_rejects_unknown_state_version_and_fields() { + assert_eq!( + serde_json::to_string(&partial_write::ResponsibilityState::Active {}).expect("encode active state"), + r#"{"state":"active"}"#, + "strict field validation must preserve the legacy encoding" + ); + let owner = Uuid::new_v4(); + let intent_digest = [7_u8; 32]; + let mut payload = json!({ + "format_version": MRF_LIFECYCLE_FORMAT_VERSION, + "checkpoint_owner": owner, + "checkpoint_sequence": 1, + "records": [{ + "intent_digest": intent_digest, + "responsibility_id": Uuid::new_v4(), + "source_bucket_incarnation_id": null, + "last_operator_acceptance": null, + "state": {"state": "active"} + }] + }); + let valid = lifecycle_frame(&payload); + assert!(decode_mrf_lifecycle_checkpoint(&valid, 8192).is_ok()); + let mut future_version = valid; + future_version[8] += 1; + assert!(matches!( + decode_mrf_lifecycle_checkpoint(&future_version, 8192), + Err(MrfReplayError::Snapshot(snapshot::SnapshotError::Unsupported)) + )); + payload["records"][0]["state"] = json!({"state": "source_and_current_generation_unknown", "since_ms": 7}); + assert!(matches!( + decode_mrf_lifecycle_checkpoint(&lifecycle_frame(&payload), 8192), + Err(MrfReplayError::Lifecycle(_)) + )); + payload["records"][0]["state"] = json!({"state": "active", "hidden_source_incarnation": Uuid::new_v4()}); + assert!(matches!( + decode_mrf_lifecycle_checkpoint(&lifecycle_frame(&payload), 8192), + Err(MrfReplayError::Lifecycle(_)) + )); +} + +fn durable_files(roots: &[PathBuf]) -> Vec>> { + roots + .iter() + .flat_map(|root| { + [ + ".heal-mrf-snapshot.0.bin", + ".heal-mrf-snapshot.1.bin", + ".heal-mrf-commit.0.bin", + ".heal-mrf-commit.1.bin", + MRF_LIFECYCLE_PATHS[0], + MRF_LIFECYCLE_PATHS[1], + MRF_SCOPED_JOURNAL_PATH, + MRF_JOURNAL_PATH, + ] + .map(|path| fs::read(root.join(RUSTFS_META_BUCKET).join(path)).ok()) + }) + .collect() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[serial] +async fn unknown_source_deleted_bucket_unsupported_lifecycle_freezes_all_writes() { + assert_unreadable_lifecycle_is_retained(|payload| { + payload["records"][0]["state"] = json!({"state": "future_held", "since_ms": 17}); + }) + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[serial] +async fn unknown_source_deleted_bucket_mismatched_lifecycle_freezes_all_writes() { + assert_unreadable_lifecycle_is_retained(|payload| { + let wrong_digest = [7_u8; 32]; + payload["records"][0]["intent_digest"] = json!(wrong_digest); + }) + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[serial] +async fn unknown_source_deleted_bucket_missing_lifecycle_record_freezes_all_writes() { + assert_unreadable_lifecycle_is_retained(|payload| payload["records"] = json!([])).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[serial] +async fn unknown_source_deleted_bucket_extra_lifecycle_record_freezes_all_writes() { + assert_unreadable_lifecycle_is_retained(|payload| { + let mut extra = payload["records"][0].clone(); + extra["responsibility_id"] = json!(Uuid::new_v4()); + extra["source_bucket_incarnation_id"] = json!(Uuid::new_v4()); + payload["records"].as_array_mut().expect("fixture records").push(extra); + }) + .await; +} + +async fn assert_unreadable_lifecycle_is_retained(corrupt: fn(&mut Value)) { + let env = rustfs_test_utils::TestECStoreEnv::builder() + .prefix("rustfs_mrf_unknown_lifecycle") + .build() + .await; + let config = MrfConsumerConfig::default(); + let owner = Uuid::new_v4(); + let mut record = intent("retained-bucket", "retained-object", 0); + record.kind = MrfKind::PartialWrite; + let bytes = encoded_payload(&record); + let mut lifecycle = json!({ + "format_version": MRF_LIFECYCLE_FORMAT_VERSION, + "checkpoint_owner": owner, + "checkpoint_sequence": 1, + "records": [{ + "intent_digest": intent_digest(&record).expect("fixture digest"), + "responsibility_id": Uuid::new_v4(), + "source_bucket_incarnation_id": Uuid::new_v4(), + "last_operator_acceptance": null, + "state": {"state": "active"} + }] + }); + corrupt(&mut lifecycle); + let invalid_lifecycle = lifecycle_frame(&lifecycle); + snapshot::publish_committed_snapshot_with_companion( + &journal_disks().await, + owner, + 1, + &bytes, + config.journal_max_bytes, + Some((&MRF_LIFECYCLE_PATHS, &invalid_lifecycle, config.journal_max_bytes)), + ) + .await + .expect("commit an unreadable companion with the retained intent"); + let before = durable_files(&env.disk_paths); + assert!(before.iter().any(Option::is_some), "fixture must contain actual committed files"); + assert!( + read_replay_source(config.journal_max_bytes, config.journal_max_bytes * 4) + .await + .is_err(), + "an unreadable companion must not be treated as an absent legacy companion" + ); + + let storage: Arc = Arc::new(ECStoreHealStorage::new(env.ecstore.clone())); + let manager = Arc::new(HealManager::new_without_root_recovery_for_test(storage, None)); + let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes); + let replay = replay_into(&manager, &mut queue, &mut None).await; + assert!(replay.retain_journal_for_replay); + assert!(replay.partial_writes.is_empty(), "do not synthesize source-less active responsibilities"); + let mut runtime = MrfRuntime { + partial_writes: PartialWrites::default(), + queue, + config, + checkpoint_owner: Uuid::new_v4(), + next_checkpoint_sequence: replay.next_checkpoint_sequence, + new_since_flush: 0, + dirty: true, + journal_on_disk: replay.journal_on_disk, + retain_replay_journal: replay.retain_journal_for_replay, + replay_readable: replay.source_readable, + durable_replay_anchors: replay.durable_replay_anchors, + replay_cleanup: replay.cleanup, + runtime_checkpoint: None, + backoff_until: None, + }; + assert!(!runtime.flush().await, "an unreadable owner must fence every writer"); + assert!(matches!( + runtime.admit_partial_write(record.clone(), Some(Uuid::new_v4())), + Err(MrfDurableAdmissionError::Persistence) + )); + assert!(!runtime.flush().await, "new admission must not overwrite the unreadable owner"); + assert!(!runtime.delete_idle_recovery_anchors().await, "idle cleanup must also remain fenced"); + assert!(matches!( + runtime.list_unverified_legacy(None, 10), + Err(MrfLifecycleControlError::CheckpointUnreadable) + )); + assert_eq!( + durable_files(&env.disk_paths), + before, + "payloads, manifests, companions and mirrors must stay byte-identical" + ); + + let (incoming, receiver) = mpsc::channel(4); + let (durable, durable_receiver) = mpsc::channel(4); + let (lifecycle, lifecycle_receiver) = mpsc::channel(4); + let consumer = tokio::spawn(run_mrf_consumer(manager.clone(), receiver, durable_receiver, lifecycle_receiver)); + let (response, received) = tokio::sync::oneshot::channel(); + durable + .send(MrfDurableSubmission { + intent: record, + source_bucket_incarnation_id: Some(Uuid::new_v4()), + response, + }) + .await + .expect("submit to the actual consumer"); + assert!(matches!( + tokio::time::timeout(Duration::from_secs(5), received) + .await + .expect("admission deadline") + .expect("admission response"), + Err(MrfDurableAdmissionError::Persistence) + )); + let (response, listed) = tokio::sync::oneshot::channel(); + lifecycle + .send(MrfLifecycleCommand::List { + after: None, + limit: 10, + response, + }) + .await + .expect("query the actual consumer"); + assert!(matches!( + tokio::time::timeout(Duration::from_secs(5), listed) + .await + .expect("listing deadline") + .expect("listing response"), + Err(MrfLifecycleControlError::CheckpointUnreadable) + )); + incoming + .send(intent("retained-bucket", "new-best-effort", 0)) + .await + .expect("queue a periodic flush"); + tokio::time::sleep(Duration::from_millis(600)).await; + drop(incoming); + tokio::time::timeout(Duration::from_secs(5), consumer) + .await + .expect("consumer shutdown deadline") + .expect("consumer shutdown"); + assert_eq!(manager.get_queue_length().await, 0, "unreadable recovery must not dispatch object work"); + assert_eq!( + durable_files(&env.disk_paths), + before, + "startup, admission, periodic flush and shutdown must preserve every file" + ); +} diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index 13035d396..b11c185a5 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -1289,11 +1289,13 @@ impl HealStorageAPI for ECStoreHealStorage { } async fn mrf_bucket_incarnation_id(&self, bucket: &str) -> Result> { - self.ecstore - .bucket_incarnation_id(bucket) - .await - .map(Some) - .map_err(Error::Storage) + match self.ecstore.bucket_incarnation_id(bucket).await { + Ok(incarnation) => Ok(Some(incarnation)), + // A missing bucket leaves the current identity unresolved. It is + // not a retirement receipt for an unknown source generation. + Err(StorageError::BucketNotFound(_)) => Ok(None), + Err(error) => Err(Error::Storage(error)), + } } async fn mrf_bucket_incarnation_retired(&self, bucket: &str, expected: Uuid) -> Result { diff --git a/crates/heal/tests/mrf_partial_write_test.rs b/crates/heal/tests/mrf_partial_write_test.rs index 99cbe5e57..1ee09b82a 100644 --- a/crates/heal/tests/mrf_partial_write_test.rs +++ b/crates/heal/tests/mrf_partial_write_test.rs @@ -173,7 +173,7 @@ async fn legacy_unbound_generation_is_parked_and_requires_explicit_risk_acceptan assert!( matches!( mrf_queue::refresh_legacy_responsibility(entry.responsibility_id, None).await, - Err(rustfs_heal::heal::mrf_queue::MrfLifecycleControlError::IncarnationChanged) + Err(rustfs_heal::heal::mrf_queue::MrfLifecycleControlError::BucketIdentityUnresolved) ), "an absent bucket is not retirement proof" ); diff --git a/docs/README.md b/docs/README.md index ac212dbd2..427929e1e 100644 --- a/docs/README.md +++ b/docs/README.md @@ -8,6 +8,8 @@ collection: ## Operations +- [Retained MRF responsibilities](operations/mrf-responsibilities.md) — unknown bucket identities, bounded probes, and unreadable checkpoint recovery. + - [Internode HTTP streaming transport](operations/internode-http-transport.md) — protocol selection, pooled connection isolation, and distributed read validation. - [Persisted config compatibility](operations/persisted-config-compatibility.md) — ignored MinIO settings, bounded diagnostics, alias retention, and safe resets. diff --git a/docs/operations/mrf-responsibilities.md b/docs/operations/mrf-responsibilities.md new file mode 100644 index 000000000..daad2a28b --- /dev/null +++ b/docs/operations/mrf-responsibilities.md @@ -0,0 +1,33 @@ +# Retained MRF responsibilities + +**Use this when:** a legacy partial-write responsibility has no source bucket identity, or the MRF controller reports an unreadable checkpoint. + +**Source of truth:** `crates/heal/src/heal/mrf_queue.rs` (`read_replay_source`, `MrfRuntime::flush`), `crates/heal/src/heal/mrf_queue/partial_write.rs` (`probe_unknown_generation`, `list_unverified_legacy`), and `rustfs/src/admin/handlers/heal.rs` (`MrfLegacyResponsibilitiesHandler`). + +## Unknown bucket identity + +`GET /rustfs/admin/v4/heal/mrf/responsibilities` lists retained responsibilities on the receiving node. Follow `nextCursor` until it is absent; the maximum page size is 256. Compare responsibility IDs and original pool/set scopes, rather than matching only bucket and object names. + +`legacy_generation_unobserved` means the original source identity is unknown and no current identity has been observed. Both incarnation fields remain null. A missing bucket is not evidence that this responsibility can be acknowledged or removed. + +These entries stay outside object-heal dispatch. After an initial 60-second delay, the consumer rotates through a bounded probe queue, attempts at most one identity lookup per dispatch round, and limits each lookup to one second. Failed or unresolved attempts use the existing bounded backoff, with a five-second minimum and a 60-second ceiling. The queue examination budget is `RUSTFS_HEAL_MRF_REPLAY_BATCH`. + +The listing exposes `lastIdentityProbeResult` (`unresolved`, `unavailable`, or `known`), `lastIdentityProbeAtMs`, and `nextIdentityProbeAtMs` when available. These are process-local observations. Restart resets the probe delay; these timestamps do not establish when a responsibility was first retained. + +When a same-name bucket becomes visible, its identity is recorded as the observed current incarnation in `legacy_generation_unknown`. The source stays null. This transition does not enqueue object repair or create a repair proof. Targeted recheck remains rejected because the original generation cannot be established. + +An explicit refresh distinguishes a successful lookup with no identity from a lookup failure. The former returns an unresolved-identity error; permission, quorum, transport, and invalid-identity failures return service unavailable without exposing internal storage details. Neither case is reported as a proven incarnation change. + +## Unreadable checkpoint + +A committed lifecycle companion with an unknown state, unsupported version, invalid contents, mismatched owner/sequence, conflicting replicas, or no readable replica must not be treated as an absent legacy companion. A genuinely absent companion remains compatible with older snapshots. + +If replay cannot read the committed owner, the consumer retains the original payloads, manifests, lifecycle companions, and compatibility mirrors. It rejects durable admission, lifecycle queries/actions, snapshot publication, and cleanup. It also stops object dispatch. The read failure is sticky for that process: restore a compatible reader or repair the storage fault, then restart and inspect the retained files again. + +Do not delete a companion, journal, or commit manifest to clear this condition. An empty API response would be misleading; the API returns service unavailable while the owner is unreadable. + +## Format and rollout boundary + +The writer continues to emit lifecycle format version 1. A source-less `Active` record still means a parked legacy responsibility, and observing a current incarnation uses the existing `LegacyGenerationUnknown` state. No new persistent state is activated by identity probing. + +A durable state for unknown source and current generations, including the first-retained time and probe schedule across restarts, needs a separate reader-first rollout. The existing heal-control and cross-pool protocol capabilities do not prove support for a new lifecycle state. New writers require evidence for every reader and the minimum supported rollback reader; do not downgrade past that reader or silently convert an unknown state to `Active`. diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 29d9fd8a1..9501b07fe 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -1733,6 +1733,10 @@ struct MrfLegacyResponsibilityActionResponse { fn map_mrf_lifecycle_control_error(error: rustfs_heal::heal::mrf_queue::MrfLifecycleControlError) -> s3s::S3Error { use rustfs_heal::heal::mrf_queue::MrfLifecycleControlError; match error { + MrfLifecycleControlError::CheckpointUnreadable => admin_error( + S3ErrorCode::ServiceUnavailable, + "MRF checkpoint is unreadable; retain its files and restore a compatible reader", + ), MrfLifecycleControlError::Unavailable | MrfLifecycleControlError::Persistence => { admin_error(S3ErrorCode::ServiceUnavailable, "MRF lifecycle operation could not be completed") } @@ -1746,6 +1750,14 @@ fn map_mrf_lifecycle_control_error(error: rustfs_heal::heal::mrf_queue::MrfLifec S3ErrorCode::InvalidRequest, "bucket incarnation changed; refresh the MRF responsibility listing", ), + MrfLifecycleControlError::BucketIdentityUnresolved => admin_error( + S3ErrorCode::InvalidRequest, + "current bucket identity is unresolved; the MRF responsibility remains held", + ), + MrfLifecycleControlError::BucketIdentityUnavailable(_) => admin_error( + S3ErrorCode::ServiceUnavailable, + "current bucket identity is unavailable; retry after storage recovers", + ), MrfLifecycleControlError::InvalidAction(reason) => admin_error(S3ErrorCode::InvalidRequest, reason), } } @@ -1960,9 +1972,9 @@ mod tests { build_replacement_recovery_status_response, encode_background_heal_status, encode_heal_control_path, encode_heal_start_success, encode_heal_task_status, execute_after_heal_start_preflight, heal_channel_response_items, heal_channel_response_progress, heal_channel_response_summary, heal_control_response_id, json_response, - map_heal_response, merge_peer_heal_statuses, parse_mrf_legacy_responsibility_query, peer_topology_complete, - query_peer_heal_status, query_peer_replacement_recovery_status, read_cluster_heal_status, reject_heal_admission, - validate_heal_request_mode, validate_heal_target, validate_mrf_legacy_responsibility_action, + map_heal_response, map_mrf_lifecycle_control_error, merge_peer_heal_statuses, parse_mrf_legacy_responsibility_query, + peer_topology_complete, query_peer_heal_status, query_peer_replacement_recovery_status, read_cluster_heal_status, + reject_heal_admission, validate_heal_request_mode, validate_heal_target, validate_mrf_legacy_responsibility_action, }; use crate::storage::rpc::node_service::heal::{ NodeHealProgress, NodeHealStatusSnapshot, NodeReplacementRecoveryStatusSnapshot, encode_node_replacement_recovery_status, @@ -2087,6 +2099,21 @@ mod tests { assert!(parse_mrf_legacy_responsibility_query(&uri).is_err(), "accepted invalid query {query}"); } } + + #[test] + fn mrf_lifecycle_errors_distinguish_unresolved_identity_and_storage_failure() { + use rustfs_heal::heal::mrf_queue::MrfLifecycleControlError; + + let unresolved = map_mrf_lifecycle_control_error(MrfLifecycleControlError::BucketIdentityUnresolved); + assert_eq!(unresolved.code(), &S3ErrorCode::InvalidRequest); + let unavailable = map_mrf_lifecycle_control_error(MrfLifecycleControlError::BucketIdentityUnavailable(Box::new( + rustfs_heal::Error::Io(std::io::Error::new(std::io::ErrorKind::PermissionDenied, "private storage detail")), + ))); + assert_eq!(unavailable.code(), &S3ErrorCode::ServiceUnavailable); + assert!(!unavailable.to_string().contains("private storage detail")); + let unreadable = map_mrf_lifecycle_control_error(MrfLifecycleControlError::CheckpointUnreadable); + assert_eq!(unreadable.code(), &S3ErrorCode::ServiceUnavailable); + } use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use tokio::sync::mpsc; use tokio::time::Duration;