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

@@ -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<uuid::Uuid> },
 
+    #[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}")]

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<rustfs_common::mrf_channel::MrfDurableRepairAnchor> {
+    ) -> Result<Option<rustfs_common::mrf_channel::MrfDurableRepairAnchor>> {
         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<Option<Uuid>> {
+        match self.storage.mrf_bucket_incarnation_id(bucket).await? {
+            Some(incarnation) if incarnation.is_nil() => Err(Error::InvalidBucketIncarnation),
+            identity => Ok(identity),
         }
     }
 

crates/heal/src/heal/mrf_queue.rs

@@ -58,12 +58,18 @@ static GLOBAL_MRF_LIFECYCLE_SENDER: std::sync::OnceLock<mpsc::Sender<MrfLifecycl
 pub enum MrfLifecycleControlError {
     #[error("MRF lifecycle controller is unavailable")]
     Unavailable,
+    #[error("the MRF checkpoint is unreadable; retain its files and restore a compatible reader")]
+    CheckpointUnreadable,
     #[error("the requested responsibility changed or does not exist")]
     StaleGeneration,
     #[error("the responsibility is not held as unverified legacy")]
     NotHeld,
     #[error("the bucket incarnation changed")]
     IncarnationChanged,
+    #[error("the current bucket identity is unresolved; the responsibility remains held")]
+    BucketIdentityUnresolved,
+    #[error("the current bucket identity is unavailable; retry after storage recovers")]
+    BucketIdentityUnavailable(#[source] Box<crate::Error>),
     #[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<Uuid>,
         limit: usize,
-        response: tokio::sync::oneshot::Sender<MrfLegacyResponsibilitySnapshot>,
+        response: tokio::sync::oneshot::Sender<Result<MrfLegacyResponsibilitySnapshot, MrfLifecycleControlError>>,
     },
     Recheck {
         responsibility_id: Uuid,
@@ -615,6 +621,16 @@ struct MrfLifecyclePayload {
     records: Vec<ResponsibilityCheckpoint>,
 }
 
+#[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<MrfLifecyclePayload> {
+fn decode_mrf_lifecycle_checkpoint(data: &[u8], byte_limit: usize) -> Result<MrfLifecyclePayload, MrfReplayError> {
     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<Mrf
             .len()
             != decoded.records.len()
     {
-        return None;
+        return Err(MrfReplayError::InvalidLifecycle);
     }
-    Some(decoded)
+    Ok(decoded)
 }
 
 fn intent_digest(intent: &MrfIntent) -> Option<[u8; 32]> {
@@ -754,8 +771,8 @@ async fn read_mrf_lifecycle_checkpoint(
     sequence: u64,
     checkpoint_max_bytes: usize,
     byte_limit: usize,
-) -> Option<Vec<ResponsibilityCheckpoint>> {
-    let bytes = snapshot::read_committed_companion(
+) -> Result<Option<Vec<ResponsibilityCheckpoint>>, 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<MrfDurableRepairAnchor>,
@@ -999,6 +1022,9 @@ impl MrfRuntime {
         intent: MrfIntent,
         source_bucket_incarnation_id: Option<Uuid>,
     ) -> 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<Uuid>, limit: usize) -> MrfLegacyResponsibilitySnapshot {
+    fn list_unverified_legacy(
+        &self,
+        after: Option<Uuid>,
+        limit: usize,
+    ) -> Result<MrfLegacyResponsibilitySnapshot, MrfLifecycleControlError> {
+        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<MrfIntent, MrfLifecycleControlError> {
+        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<Uuid>,
     ) -> 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<HealManager>) -> 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<MrfDurableRepairAnchor>,
     partial_writes: Vec<(MrfIntent, Option<ResponsibilityCheckpoint>)>,
     cleanup: Option<ReplayCleanup>,
@@ -1559,7 +1619,8 @@ enum ReplayCleanup {
 }
 
 struct ReplaySource {
-    data: Vec<u8>,
+    intents: Vec<MrfIntent>,
+    truncated_bytes: usize,
     cleanup: ReplayCleanup,
     lifecycle: HashMap<[u8; 32], VecDeque<ResponsibilityCheckpoint>>,
 }
@@ -1576,22 +1637,38 @@ fn take_lifecycle_record(
     record
 }
 
-async fn read_replay_source(
-    max_bytes: usize,
-    lifecycle_max_bytes: usize,
-) -> Result<Option<ReplaySource>, snapshot::SnapshotError> {
+async fn read_replay_source(max_bytes: usize, lifecycle_max_bytes: usize) -> Result<Option<ReplaySource>, 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<ResponsibilityCheckpoint>> = 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,

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::<Uuid>() + std::mem::size_of::<u64>(),
             Self::LegacyGenerationUnknown { .. } => std::mem::size_of::<Uuid>() + std::mem::size_of::<u64>(),
             Self::BucketIncarnationChanged { .. } => 2 * std::mem::size_of::<Uuid>() + std::mem::size_of::<u64>(),
@@ -115,7 +116,7 @@ impl ResponsibilityState {
 
     pub(super) fn bucket_incarnation_id(&self) -> Option<Uuid> {
         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<u64>,
 }
 
 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<PartialWriteKey, Responsibility>,
     retry_order: VecDeque<PartialWriteKey>,
     retry_index: HashSet<PartialWriteKey>,
+    identity_probe_order: VecDeque<PartialWriteKey>,
+    identity_probe_index: HashSet<PartialWriteKey>,
     retirement_checks: HashMap<(Arc<str>, 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::<Responsibility>()
-            + 2 * std::mem::size_of::<PartialWriteKey>()
+            + 4 * std::mem::size_of::<PartialWriteKey>()
             + LIFECYCLE_CHECKPOINT_RECORD_OVERHEAD_BYTES
-            + state.estimated_bytes()
+            // Reserve the observation transition before admitting an intent.
+            + state.estimated_bytes().max(std::mem::size_of::<Uuid>() + std::mem::size_of::<u64>())
     }
 
     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<PartialWriteKey> {
         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<u64>,
     pub accepted: Option<MrfOperatorAcceptance>,
+    #[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<u64>,
+    /// Process-local schedule; replay restarts the conservative probe delay.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub next_identity_probe_at_ms: Option<u64>,
 }
 
 #[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,

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<Uuid>),
+    PermissionDenied,
+    Quorum,
+    Transport,
+    Pending,
+}
+
+struct ProbeStorage {
+    lookup: Mutex<Lookup>,
+    calls: AtomicUsize,
+}
+
+#[async_trait::async_trait]
+impl HealStorageAPI for ProbeStorage {
+    async fn mrf_bucket_incarnation_id(&self, _: &str) -> Result<Option<Uuid>> {
+        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<Option<HealObjectInfo>> {
+        panic!("identity probing must not read objects")
+    }
+    async fn ec_decode_rebuild(&self, _: &str, _: &str) -> Result<Vec<u8>> {
+        panic!("identity probing must not rebuild objects")
+    }
+    async fn get_bucket_info(&self, _: &str) -> Result<Option<BucketInfo>> {
+        panic!("identity probing must use the incarnation lookup")
+    }
+    async fn list_buckets(&self) -> Result<Vec<BucketInfo>> {
+        panic!("identity probing must not enumerate buckets")
+    }
+    async fn list_buckets_for_admin_heal(&self, _: Option<DiskSetSelector>) -> Result<Vec<BucketInfo>> {
+        panic!("identity probing must not start admin healing")
+    }
+    async fn list_objects_for_heal_page(
+        &self,
+        _: &str,
+        _: &str,
+        _: Option<&str>,
+        _: bool,
+    ) -> Result<(Vec<HealListItem>, Option<String>, bool)> {
+        panic!("identity probing must not enumerate objects")
+    }
+    async fn object_exists(&self, _: &str, _: &str) -> Result<bool> {
+        panic!("identity probing must not probe objects")
+    }
+    async fn heal_object(&self, _: &str, _: &str, _: Option<&str>, _: &HealOpts) -> Result<(HealResultItem, Option<Error>)> {
+        panic!("identity probing must not heal objects")
+    }
+    async fn heal_bucket(&self, _: &str, _: &HealOpts) -> Result<HealResultItem> {
+        panic!("identity probing must not heal buckets")
+    }
+    async fn heal_format(&self, _: bool) -> Result<(HealResultItem, Option<Error>)> {
+        panic!("identity probing must not heal formats")
+    }
+    async fn get_disk_for_resume(&self, _: &str) -> Result<DiskStore> {
+        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<dyn HealStorageAPI> = 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::<HashSet<_>>(),
+        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");
+}

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

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<u8> {
+    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<Option<Vec<u8>>> {
+    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<dyn HealStorageAPI> = 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"
+    );
+}

crates/heal/src/heal/storage.rs

@@ -1289,11 +1289,13 @@ impl HealStorageAPI for ECStoreHealStorage {
     }
 
     async fn mrf_bucket_incarnation_id(&self, bucket: &str) -> Result<Option<Uuid>> {
-        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<bool> {

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"
         );

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.

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`.

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;