feat(heal): read retained MRF lifecycle v2 checkpoints (#8913)

2026-10-27 06:18:43 +00:00 · 2026-10-25 14:34:57 +08:00
parent 3aa7964c58
commit 124cbfc00a
7 changed files with 672 additions and 28 deletions
@@ -212,6 +212,7 @@ pub(crate) const MRF_SCOPED_JOURNAL_PATH: &str = "buckets/.heal/mrf/journal-scop
 const MRF_LIFECYCLE_PATHS: [&str; 2] = [".heal-mrf-lifecycle.0.bin", ".heal-mrf-lifecycle.1.bin"];
 const MRF_LIFECYCLE_MAGIC: &[u8; 8] = b"RFMFLC01";
 const MRF_LIFECYCLE_FORMAT_VERSION: u8 = 1;
+const MRF_LIFECYCLE_FORMAT_VERSION_V2: u8 = 2;

 /// Record format tag.
 const MRF_JOURNAL_FORMAT: u8 = 1;
@@ -632,15 +633,21 @@ enum MrfReplayError {
 }

 fn encode_mrf_lifecycle_checkpoint(
+    format_version: u8,
    owner: Uuid,
    sequence: u64,
    records: Vec<ResponsibilityCheckpoint>,
    byte_limit: usize,
 ) -> Option<Vec<u8>> {
-    if owner.is_nil()
+    if !matches!(format_version, MRF_LIFECYCLE_FORMAT_VERSION | MRF_LIFECYCLE_FORMAT_VERSION_V2)
+        || owner.is_nil()
        || sequence == 0
        || sequence == u64::MAX
        || records.iter().any(|record| !record.is_valid())
+        || (format_version == MRF_LIFECYCLE_FORMAT_VERSION
+            && records.iter().any(|record| {
+                matches!(record.state, partial_write::ResponsibilityState::SourceAndCurrentGenerationUnknown { .. })
+            }))
        || records
            .iter()
            .map(|record| record.responsibility_id)
@@ -657,7 +664,7 @@ fn encode_mrf_lifecycle_checkpoint(
        return None;
    }
    let payload = MrfLifecyclePayload {
-        format_version: MRF_LIFECYCLE_FORMAT_VERSION,
+        format_version,
        checkpoint_owner: owner,
        checkpoint_sequence: sequence,
        records,
@@ -669,7 +676,7 @@ fn encode_mrf_lifecycle_checkpoint(
    let digest: [u8; 32] = sha2::Sha256::digest(&payload).into();
    let mut encoded = Vec::with_capacity(8 + 1 + 8 + 32 + payload.len());
    encoded.extend_from_slice(MRF_LIFECYCLE_MAGIC);
-    encoded.push(MRF_LIFECYCLE_FORMAT_VERSION);
+    encoded.push(format_version);
    encoded.extend_from_slice(&u64::try_from(payload.len()).ok()?.to_le_bytes());
    encoded.extend_from_slice(&digest);
    encoded.extend_from_slice(&payload);
@@ -681,7 +688,7 @@ fn decode_mrf_lifecycle_checkpoint(data: &[u8], byte_limit: usize) -> Result<Mrf
    if data.len() < HEADER_LEN || data.len() > byte_limit || &data[..8] != MRF_LIFECYCLE_MAGIC {
        return Err(MrfReplayError::InvalidLifecycle);
    }
-    if data[8] != MRF_LIFECYCLE_FORMAT_VERSION {
+    if !matches!(data[8], MRF_LIFECYCLE_FORMAT_VERSION | MRF_LIFECYCLE_FORMAT_VERSION_V2) {
        return Err(snapshot::SnapshotError::Unsupported.into());
    }
    let payload_len = usize::try_from(u64::from_le_bytes(data[9..17].try_into().map_err(|_| MrfReplayError::InvalidLifecycle)?))
@@ -695,11 +702,15 @@ fn decode_mrf_lifecycle_checkpoint(data: &[u8], byte_limit: usize) -> Result<Mrf
        return Err(MrfReplayError::InvalidLifecycle);
    }
    let decoded: MrfLifecyclePayload = serde_json::from_slice(payload).map_err(MrfReplayError::Lifecycle)?;
-    if decoded.format_version != MRF_LIFECYCLE_FORMAT_VERSION
+    if decoded.format_version != data[8]
        || decoded.checkpoint_owner.is_nil()
        || decoded.checkpoint_sequence == 0
        || decoded.checkpoint_sequence == u64::MAX
        || decoded.records.iter().any(|record| !record.is_valid())
+        || (decoded.format_version == MRF_LIFECYCLE_FORMAT_VERSION
+            && decoded.records.iter().any(|record| {
+                matches!(record.state, partial_write::ResponsibilityState::SourceAndCurrentGenerationUnknown { .. })
+            }))
        || decoded
            .records
            .iter()
@@ -771,7 +782,7 @@ async fn read_mrf_lifecycle_checkpoint(
    sequence: u64,
    checkpoint_max_bytes: usize,
    byte_limit: usize,
-) -> Result<Option<Vec<ResponsibilityCheckpoint>>, MrfReplayError> {
+) -> Result<Option<MrfLifecyclePayload>, MrfReplayError> {
    let Some(bytes) = snapshot::read_committed_companion(
        &journal_disks().await,
        owner,
@@ -788,7 +799,7 @@ async fn read_mrf_lifecycle_checkpoint(
    if decoded.checkpoint_owner != owner || decoded.checkpoint_sequence != sequence {
        return Err(MrfReplayError::InvalidLifecycle);
    }
-    Ok(Some(decoded.records))
+    Ok(Some(decoded))
 }

 /// Write the snapshot to every local disk; returns true when at least one
@@ -953,6 +964,9 @@ struct MrfRuntime {
    config: MrfConsumerConfig,
    checkpoint_owner: Uuid,
    next_checkpoint_sequence: u64,
+    // Preserve an observed format across checkpoint ownership changes.
+    // Without a V2 checkpoint, keep writing V1 until writer rollout is qualified.
+    lifecycle_format_version: u8,
    new_since_flush: usize,
    /// True while the in-memory pending set has changed since the last
    /// journal flush (push, pop, or an attempts bump that alters the encoded
@@ -1072,6 +1086,9 @@ impl MrfRuntime {
            .partial_writes
            .intent_for_responsibility(responsibility_id)
            .ok_or(MrfLifecycleControlError::StaleGeneration)?;
+        if matches!(state, partial_write::ResponsibilityState::SourceAndCurrentGenerationUnknown { .. }) {
+            return Err(MrfLifecycleControlError::BucketIdentityUnresolved);
+        }
        if state.bucket_incarnation_id() != Some(expected_bucket_incarnation_id) {
            return Err(MrfLifecycleControlError::IncarnationChanged);
        }
@@ -1139,7 +1156,12 @@ 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 {}
+                        | partial_write::ResponsibilityState::SourceAndCurrentGenerationUnknown { .. }
+                ))
        {
            return Err(MrfLifecycleControlError::StaleGeneration);
        }
@@ -1247,6 +1269,7 @@ impl MrfRuntime {
            (records.len() == self.partial_writes.depth())
                .then(|| {
                    encode_mrf_lifecycle_checkpoint(
+                        self.lifecycle_format_version,
                        self.checkpoint_owner,
                        self.next_checkpoint_sequence,
                        records,
@@ -1576,6 +1599,7 @@ pub async fn replay_journal_once(manager: &Arc<HealManager>) -> usize {
        config,
        checkpoint_owner: Uuid::new_v4(),
        next_checkpoint_sequence: replay.next_checkpoint_sequence,
+        lifecycle_format_version: replay.lifecycle_format_version,
        new_since_flush: 0,
        dirty: false,
        journal_on_disk: replay.journal_on_disk,
@@ -1601,6 +1625,7 @@ struct ReplayOutcome {
    partial_writes: Vec<(MrfIntent, Option<ResponsibilityCheckpoint>)>,
    cleanup: Option<ReplayCleanup>,
    next_checkpoint_sequence: u64,
+    lifecycle_format_version: u8,
 }

 fn replay_must_retain_journal(
@@ -1623,6 +1648,7 @@ struct ReplaySource {
    truncated_bytes: usize,
    cleanup: ReplayCleanup,
    lifecycle: HashMap<[u8; 32], VecDeque<ResponsibilityCheckpoint>>,
+    lifecycle_format_version: u8,
 }

 fn take_lifecycle_record(
@@ -1642,8 +1668,8 @@ async fn read_replay_source(max_bytes: usize, lifecycle_max_bytes: usize) -> Res
        let owner = committed.owner();
        let sequence = committed.sequence();
        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() {
+        let checkpoint = read_mrf_lifecycle_checkpoint(owner, sequence, max_bytes, lifecycle_max_bytes).await?;
+        if let Some(checkpoint) = checkpoint.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();
@@ -1651,7 +1677,7 @@ async fn read_replay_source(max_bytes: usize, lifecycle_max_bytes: usize) -> Res
                let digest = intent_digest(intent).ok_or(MrfReplayError::InvalidLifecycle)?;
                *counts.entry(digest).or_insert(0usize) += 1;
            }
-            for record in records {
+            for record in &checkpoint.records {
                match counts.get_mut(&record.intent_digest) {
                    Some(count) if *count > 0 => *count -= 1,
                    _ => return Err(MrfReplayError::InvalidLifecycle),
@@ -1661,7 +1687,10 @@ async fn read_replay_source(max_bytes: usize, lifecycle_max_bytes: usize) -> Res
                return Err(MrfReplayError::InvalidLifecycle);
            }
        }
-        let records = records.unwrap_or_default();
+        let (lifecycle_format_version, records) = match checkpoint {
+            Some(checkpoint) => (checkpoint.format_version, checkpoint.records),
+            None => (MRF_LIFECYCLE_FORMAT_VERSION, Vec::new()),
+        };
        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);
@@ -1671,6 +1700,7 @@ async fn read_replay_source(max_bytes: usize, lifecycle_max_bytes: usize) -> Res
            truncated_bytes,
            cleanup: ReplayCleanup::Committed { owner, sequence },
            lifecycle,
+            lifecycle_format_version,
        }));
    }
    // The scoped file is a complete authoritative legacy snapshot. Fall back
@@ -1689,6 +1719,7 @@ async fn read_replay_source(max_bytes: usize, lifecycle_max_bytes: usize) -> Res
        truncated_bytes,
        cleanup: ReplayCleanup::Legacy,
        lifecycle: HashMap::new(),
+        lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
    }))
 }

@@ -1733,6 +1764,7 @@ async fn replay_into(
                partial_writes: Vec::new(),
                cleanup: None,
                next_checkpoint_sequence: 1,
+                lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
            };
        }
        Err(err) => {
@@ -1750,6 +1782,7 @@ async fn replay_into(
                partial_writes: Vec::new(),
                cleanup: None,
                next_checkpoint_sequence: 1,
+                lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
            };
        }
    };
@@ -1896,6 +1929,7 @@ async fn replay_into(
        partial_writes,
        cleanup: journal_on_disk.then_some(cleanup),
        next_checkpoint_sequence,
+        lifecycle_format_version: source.lifecycle_format_version,
    }
 }

@@ -1914,6 +1948,7 @@ async fn run_mrf_consumer(
        config: config.clone(),
        checkpoint_owner: Uuid::new_v4(),
        next_checkpoint_sequence: 1,
+        lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
        new_since_flush: 0,
        dirty: false,
        journal_on_disk: false,
@@ -1934,6 +1969,7 @@ async fn run_mrf_consumer(
    runtime.durable_replay_anchors = replay.durable_replay_anchors;
    runtime.replay_cleanup = replay.cleanup;
    runtime.next_checkpoint_sequence = replay.next_checkpoint_sequence;
+    runtime.lifecycle_format_version = replay.lifecycle_format_version;
    runtime.adopt_replayed_partial_writes(replay.partial_writes);
    // Anything still pending (e.g. the manager was full and backoff armed)
    // must be re-persisted by the next flush before replay can delete the
@@ -2148,6 +2184,7 @@ mod tests {

    mod empty_events_tests;
    mod lifecycle_replay_tests;
+    mod lifecycle_v2_reader_tests;

    #[tokio::test(start_paused = true)]
    async fn journal_disk_io_bounds_fanout_and_collects_every_result() {
@@ -2307,7 +2344,7 @@ mod tests {
            },
        };
        let records = vec![record, accepted];
-        let encoded = encode_mrf_lifecycle_checkpoint(owner, 42, records.clone(), 8192)
+        let encoded = encode_mrf_lifecycle_checkpoint(MRF_LIFECYCLE_FORMAT_VERSION, owner, 42, records.clone(), 8192)
            .expect("lifecycle checkpoint should fit the configured bound");
        let decoded = decode_mrf_lifecycle_checkpoint(&encoded, 8192).expect("valid lifecycle checkpoint");
        assert_eq!(decoded.checkpoint_owner, owner);
@@ -2318,7 +2355,7 @@ mod tests {
        *torn.last_mut().expect("non-empty checkpoint") ^= 0xff;
        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());
+        assert!(encode_mrf_lifecycle_checkpoint(MRF_LIFECYCLE_FORMAT_VERSION, Uuid::nil(), 42, Vec::new(), 8192).is_none());
    }

    fn w13_timestamp() -> String {
@@ -2442,6 +2479,7 @@ mod tests {
            config,
            checkpoint_owner: Uuid::new_v4(),
            next_checkpoint_sequence: replay.next_checkpoint_sequence,
+            lifecycle_format_version: replay.lifecycle_format_version,
            new_since_flush: 0,
            dirty: false,
            journal_on_disk: replay.journal_on_disk,
@@ -2582,6 +2620,7 @@ mod tests {
            config: MrfConsumerConfig::default(),
            checkpoint_owner: Uuid::new_v4(),
            next_checkpoint_sequence: 1,
+            lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
            new_since_flush: 0,
            dirty: true,
            journal_on_disk: false,
@@ -3175,6 +3214,7 @@ mod tests {
            config: MrfConsumerConfig::default(),
            checkpoint_owner: Uuid::new_v4(),
            next_checkpoint_sequence: 1,
+            lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
            new_since_flush: 0,
            dirty: false,
            journal_on_disk: true,
@@ -3227,6 +3267,7 @@ mod tests {
            config: MrfConsumerConfig::default(),
            checkpoint_owner: Uuid::new_v4(),
            next_checkpoint_sequence: 1,
+            lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
            new_since_flush: 0,
            dirty: false,
            journal_on_disk: true,
@@ -3276,6 +3317,7 @@ mod tests {
            config,
            checkpoint_owner: Uuid::new_v4(),
            next_checkpoint_sequence: 9,
+            lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
            new_since_flush: 0,
            dirty: false,
            journal_on_disk: true,
@@ -3388,6 +3430,7 @@ mod tests {
            .expect("the fixture bucket must have a persisted incarnation");
        let lifecycle_limit = config.journal_max_bytes.saturating_mul(4);
        let lifecycle = encode_mrf_lifecycle_checkpoint(
+            MRF_LIFECYCLE_FORMAT_VERSION,
            replay_owner,
            11,
            vec![ResponsibilityCheckpoint {
@@ -3451,6 +3494,7 @@ mod tests {
            config,
            checkpoint_owner: Uuid::new_v4(),
            next_checkpoint_sequence: replay.next_checkpoint_sequence,
+            lifecycle_format_version: replay.lifecycle_format_version,
            new_since_flush: 0,
            dirty: false,
            journal_on_disk: replay.journal_on_disk,
@@ -3699,6 +3743,7 @@ mod tests {
            config: MrfConsumerConfig::default(),
            checkpoint_owner: Uuid::new_v4(),
            next_checkpoint_sequence: 1,
+            lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
            new_since_flush: 0,
            dirty: true,
            journal_on_disk: false,
@@ -69,6 +69,11 @@ impl PartialWriteKey {
 #[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
 pub(super) enum ResponsibilityState {
    Active {},
+    SourceAndCurrentGenerationUnknown {
+        since_ms: u64,
+        next_probe_at_ms: u64,
+        probe_attempts: u32,
+    },
    HeldUnverifiedLegacy {
        bucket_incarnation_id: Uuid,
        since_ms: u64,
@@ -102,6 +107,7 @@ impl ResponsibilityState {
    fn estimated_bytes(&self) -> usize {
        match self {
            Self::Active {} => 0,
+            Self::SourceAndCurrentGenerationUnknown { .. } => 2 * std::mem::size_of::<u64>() + std::mem::size_of::<u32>(),
            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>(),
@@ -116,7 +122,7 @@ impl ResponsibilityState {

    pub(super) fn bucket_incarnation_id(&self) -> Option<Uuid> {
        match self {
-            Self::Active {} => None,
+            Self::Active {} | Self::SourceAndCurrentGenerationUnknown { .. } => None,
            Self::HeldUnverifiedLegacy {
                bucket_incarnation_id, ..
            }
@@ -158,6 +164,7 @@ impl ResponsibilityCheckpoint {
        }
        match &self.state {
            ResponsibilityState::Active {} => true,
+            ResponsibilityState::SourceAndCurrentGenerationUnknown { .. } => self.source_bucket_incarnation_id.is_none(),
            ResponsibilityState::HeldUnverifiedLegacy {
                bucket_incarnation_id, ..
            } => !bucket_incarnation_id.is_nil() && self.source_bucket_incarnation_id == Some(*bucket_incarnation_id),
@@ -214,7 +221,11 @@ struct Responsibility {

 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 {} | ResponsibilityState::SourceAndCurrentGenerationUnknown { .. }
+            )
    }

    fn is_parked(&self) -> bool {
@@ -479,7 +490,8 @@ impl PartialWrites {
                            .map_or(*since_ms, |oldest| oldest.min(*since_ms)),
                    );
                }
-                ResponsibilityState::LegacyGenerationUnknown { .. } => summary.legacy_generation_unknown_count += 1,
+                ResponsibilityState::LegacyGenerationUnknown { .. }
+                | ResponsibilityState::SourceAndCurrentGenerationUnknown { .. } => summary.legacy_generation_unknown_count += 1,
                ResponsibilityState::BucketIncarnationChanged { .. } => summary.bucket_incarnation_changed_count += 1,
                ResponsibilityState::OperatorAcceptedUnverified { accepted_at_ms, .. } => {
                    summary.operator_accepted_count += 1;
@@ -525,6 +537,17 @@ impl PartialWrites {
        entry.source_bucket_incarnation_id = record.source_bucket_incarnation_id;
        entry.last_operator_acceptance = record.last_operator_acceptance.clone();
        entry.state = record.state.clone();
+        if let ResponsibilityState::SourceAndCurrentGenerationUnknown {
+            next_probe_at_ms,
+            probe_attempts,
+            ..
+        } = &entry.state
+        {
+            let delay = Duration::from_millis(next_probe_at_ms.saturating_sub(unix_now_ms()))
+                .clamp(MRF_RETIREMENT_PROOF_RECHECK_INTERVAL, MAX_UNVERIFIED_RETRY_BACKOFF);
+            entry.next_attempt = Instant::now() + delay;
+            entry.consecutive_unverified_attempts = *probe_attempts;
+        }
        self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry));
        if entry.is_parked() {
            entry.retry_queued = false;
@@ -616,6 +639,9 @@ impl PartialWrites {
                return Err("a different operator disposition already exists");
            }
            ResponsibilityState::Active {} => return Err("responsibility is not held as unverified legacy"),
+            ResponsibilityState::SourceAndCurrentGenerationUnknown { .. } => {
+                return Err("current bucket incarnation is unresolved");
+            }
        };
        if current_incarnation != expected_bucket_incarnation_id {
            return Err("bucket incarnation changed; refresh the responsibility listing");
@@ -743,6 +769,9 @@ impl PartialWrites {
            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::SourceAndCurrentGenerationUnknown { since_ms, .. } => {
+                    ("source_and_current_generation_unknown", None, Some(*since_ms))
+                }
                ResponsibilityState::HeldUnverifiedLegacy {
                    bucket_incarnation_id,
                    since_ms,
@@ -851,9 +880,17 @@ impl PartialWrites {
            entry.retry_queued = false;
        } else if source_bucket_incarnation_id.is_none() {
            let old_cost = Self::entry_cost(entry);
+            let detected_at_ms = match entry.state {
+                ResponsibilityState::SourceAndCurrentGenerationUnknown { since_ms, .. }
+                | ResponsibilityState::LegacyGenerationUnknown {
+                    detected_at_ms: since_ms,
+                    ..
+                } => since_ms,
+                _ => unix_now_ms(),
+            };
            entry.state = ResponsibilityState::LegacyGenerationUnknown {
                observed_bucket_incarnation_id,
-                detected_at_ms: unix_now_ms(),
+                detected_at_ms,
            };
            self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry));
            entry.retry_queued = false;
@@ -957,8 +994,27 @@ impl PartialWrites {
                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));
+            // retry loop. V2 checkpoints also retain this pacing on restart.
+            entry.defer_unverified_attempt(
+                now,
+                config
+                    .admission_backoff
+                    .clamp(MRF_RETIREMENT_PROOF_RECHECK_INTERVAL, MAX_UNVERIFIED_RETRY_BACKOFF),
+            );
+            let (lifecycle_changed, detected_at_ms) = match &mut entry.state {
+                ResponsibilityState::SourceAndCurrentGenerationUnknown {
+                    since_ms,
+                    next_probe_at_ms,
+                    probe_attempts,
+                } => {
+                    *next_probe_at_ms = unix_now_ms().saturating_add(
+                        u64::try_from(entry.next_attempt.saturating_duration_since(now).as_millis()).unwrap_or(u64::MAX),
+                    );
+                    *probe_attempts = entry.consecutive_unverified_attempts;
+                    (true, *since_ms)
+                }
+                _ => (false, unix_now_ms()),
+            };
            let result = tokio::time::timeout(
                MRF_IDENTITY_PROBE_TIMEOUT,
                manager.mrf_bucket_incarnation_id(entry.intent.bucket.as_ref()),
@@ -970,7 +1026,7 @@ impl PartialWrites {
                    let old_cost = Self::entry_cost(entry);
                    entry.state = ResponsibilityState::LegacyGenerationUnknown {
                        observed_bucket_incarnation_id,
-                        detected_at_ms: unix_now_ms(),
+                        detected_at_ms,
                    };
                    entry.last_identity_probe_result = Some("known");
                    self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry));
@@ -979,7 +1035,7 @@ impl PartialWrites {
                Ok(Ok(None)) => entry.last_identity_probe_result = Some("unresolved"),
                Ok(Err(_)) | Err(_) => entry.last_identity_probe_result = Some("unavailable"),
            }
-            return false;
+            return lifecycle_changed;
        }
        false
    }
@@ -2127,8 +2183,14 @@ mod tests {
        let records = writes.checkpoint_records(super::super::intent_digest);
        assert_eq!(records.len(), 2);
        assert_eq!(records[0].intent_digest, records[1].intent_digest);
-        let lifecycle = super::super::encode_mrf_lifecycle_checkpoint(owner, sequence, records, lifecycle_limit)
-            .expect("same-key generations need distinct lifecycle records");
+        let lifecycle = super::super::encode_mrf_lifecycle_checkpoint(
+            crate::heal::mrf_queue::MRF_LIFECYCLE_FORMAT_VERSION,
+            owner,
+            sequence,
+            records,
+            lifecycle_limit,
+        )
+        .expect("same-key generations need distinct lifecycle records");
        super::super::snapshot::publish_committed_snapshot_with_companion(
            &disks,
            owner,
@@ -2152,6 +2214,7 @@ mod tests {
            config,
            checkpoint_owner: Uuid::new_v4(),
            next_checkpoint_sequence: replay.next_checkpoint_sequence,
+            lifecycle_format_version: replay.lifecycle_format_version,
            new_since_flush: 0,
            dirty: false,
            journal_on_disk: replay.journal_on_disk,
@@ -2288,6 +2351,7 @@ mod tests {
        let lifecycle_limit = config.journal_max_bytes.saturating_mul(4);
        let owner = Uuid::new_v4();
        let lifecycle = super::super::encode_mrf_lifecycle_checkpoint(
+            crate::heal::mrf_queue::MRF_LIFECYCLE_FORMAT_VERSION,
            owner,
            7,
            writes.checkpoint_records(super::super::intent_digest),
@@ -2314,6 +2378,7 @@ mod tests {
            config,
            checkpoint_owner: Uuid::new_v4(),
            next_checkpoint_sequence: replay.next_checkpoint_sequence,
+            lifecycle_format_version: replay.lifecycle_format_version,
            new_since_flush: 0,
            dirty: false,
            journal_on_disk: replay.journal_on_disk,
@@ -2424,6 +2489,7 @@ mod tests {
            config,
            checkpoint_owner: Uuid::new_v4(),
            next_checkpoint_sequence: 1,
+            lifecycle_format_version: crate::heal::mrf_queue::MRF_LIFECYCLE_FORMAT_VERSION,
            new_since_flush: 0,
            dirty: false,
            journal_on_disk: true,
@@ -2501,6 +2567,7 @@ mod tests {
            config,
            checkpoint_owner: Uuid::new_v4(),
            next_checkpoint_sequence: 1,
+            lifecycle_format_version: crate::heal::mrf_queue::MRF_LIFECYCLE_FORMAT_VERSION,
            new_since_flush: 0,
            dirty: false,
            journal_on_disk: true,
@@ -99,6 +99,7 @@ fn runtime() -> MrfRuntime {
        config,
        checkpoint_owner: Uuid::new_v4(),
        next_checkpoint_sequence: 1,
+        lifecycle_format_version: crate::heal::mrf_queue::MRF_LIFECYCLE_FORMAT_VERSION,
        new_since_flush: 0,
        dirty: false,
        journal_on_disk: false,
@@ -368,3 +369,169 @@ async fn check_unknown_source_deleted_bucket_replay() {
    }
    assert_eq!(manager.get_queue_length().await, 0, "identity probes must not enqueue object healing");
 }
+
+fn restored_v2_runtime(next_probe_at_ms: u64, probe_attempts: u32) -> (MrfRuntime, PartialWriteKey) {
+    let mut runtime = runtime();
+    runtime.lifecycle_format_version = 2;
+    let original = intent("v2-unknown-generation");
+    let key = PartialWriteKey::new(&original, None);
+    runtime
+        .admit_partial_write(original.clone(), None)
+        .expect("admit retained owner");
+    let record = ResponsibilityCheckpoint {
+        intent_digest: super::super::super::intent_digest(&original).expect("intent identity"),
+        responsibility_id: Uuid::new_v4(),
+        source_bucket_incarnation_id: None,
+        last_operator_acceptance: None,
+        state: ResponsibilityState::SourceAndCurrentGenerationUnknown {
+            since_ms: 7,
+            next_probe_at_ms,
+            probe_attempts,
+        },
+    };
+    assert!(runtime.partial_writes.restore_state(&original, &record));
+    runtime.partial_writes.mark_persisted();
+    (runtime, key)
+}
+
+#[tokio::test(start_paused = true)]
+async fn lifecycle_v2_reader_restores_bounded_probe_schedule_and_attempt_count() {
+    for (deadline, attempts, expected_delay) in [
+        (0, 0, MRF_RETIREMENT_PROOF_RECHECK_INTERVAL),
+        (u64::MAX, u32::MAX, MAX_UNVERIFIED_RETRY_BACKOFF),
+    ] {
+        let now = Instant::now();
+        let (runtime, key) = restored_v2_runtime(deadline, attempts);
+        let entry = runtime.partial_writes.entries.get(&key).expect("retained owner");
+        assert_eq!(entry.next_attempt, now + expected_delay);
+        assert_eq!(entry.consecutive_unverified_attempts, attempts);
+        assert!(entry.anchor.is_none());
+        assert!(!entry.retry_queued);
+        assert!(entry.is_parked());
+        assert_eq!(entry.source_bucket_incarnation_id, None);
+        assert!(
+            matches!(entry.state, ResponsibilityState::SourceAndCurrentGenerationUnknown { since_ms: 7, next_probe_at_ms, probe_attempts } if next_probe_at_ms == deadline && probe_attempts == attempts)
+        );
+        assert_eq!(runtime.partial_writes.bytes(), PartialWrites::entry_cost(entry));
+        assert_eq!(runtime.partial_writes.lifecycle_summary().legacy_generation_unknown_count, 1);
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn lifecycle_v2_reader_persists_failed_probe_pacing_without_object_dispatch() {
+    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, key) = restored_v2_runtime(0, u32::MAX);
+    let id = runtime.partial_writes.entries[&key].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 outcome") = outcome;
+        runtime.partial_writes.entries.get_mut(&key).expect("owner").next_attempt = Instant::now();
+        let calls = storage.calls.load(Ordering::Relaxed);
+        runtime.dirty = false;
+        runtime.dispatch(&manager).await;
+        assert!(runtime.dirty, "failed V2 probes still change the durable retry schedule");
+        assert_eq!(storage.calls.load(Ordering::Relaxed), calls + 1);
+        let entry = &runtime.partial_writes.entries[&key];
+        assert_eq!(entry.responsibility_id, id);
+        assert_eq!(entry.source_bucket_incarnation_id, None);
+        assert!(entry.anchor.is_none());
+        assert!(!entry.retry_queued);
+        assert!(entry.next_attempt > Instant::now());
+        assert!(entry.next_attempt <= Instant::now() + MAX_UNVERIFIED_RETRY_BACKOFF);
+        assert!(
+            matches!(entry.state, ResponsibilityState::SourceAndCurrentGenerationUnknown { since_ms: 7, probe_attempts: u32::MAX, next_probe_at_ms } if next_probe_at_ms > unix_now_ms() && next_probe_at_ms <= unix_now_ms() + 60_000)
+        );
+        assert_eq!(runtime.partial_writes.bytes(), bytes);
+        runtime.dirty = false;
+        runtime.dispatch(&manager).await;
+        assert!(!runtime.dirty);
+        assert_eq!(storage.calls.load(Ordering::Relaxed), calls + 1);
+    }
+    let observed = Uuid::new_v4();
+    *storage.lookup.lock().expect("known outcome") = Lookup::Identity(Some(observed));
+    runtime.partial_writes.entries.get_mut(&key).expect("owner").next_attempt = Instant::now();
+    runtime.dispatch(&manager).await;
+    assert!(runtime.dirty);
+    assert_eq!(runtime.partial_writes.entries[&key].source_bucket_incarnation_id, None);
+    assert!(
+        matches!(runtime.partial_writes.entries[&key].state, ResponsibilityState::LegacyGenerationUnknown { observed_bucket_incarnation_id, detected_at_ms: 7 } if observed_bucket_incarnation_id == observed)
+    );
+    assert_eq!(runtime.partial_writes.bytes(), bytes);
+    assert_eq!(manager.get_queue_length().await, 0);
+}
+
+#[tokio::test(start_paused = true)]
+async fn lifecycle_v2_reader_rejects_risk_acceptance_and_recheck_without_current_identity() {
+    let storage = Arc::new(ProbeStorage {
+        lookup: Mutex::new(Lookup::Identity(Some(Uuid::new_v4()))),
+        calls: AtomicUsize::new(0),
+    });
+    let manager = HealManager::new_without_root_recovery_for_test(storage.clone(), None);
+    let (mut runtime, key) = restored_v2_runtime(0, 3);
+    let id = runtime.partial_writes.entries[&key].responsibility_id;
+    let before = runtime.partial_writes.checkpoint_records(super::super::super::intent_digest);
+    let expected = Uuid::new_v4();
+    assert!(matches!(
+        runtime.recheck_legacy_generation(&manager, id, expected).await,
+        Err(MrfLifecycleControlError::BucketIdentityUnresolved)
+    ));
+    assert!(matches!(
+        runtime
+            .accept_legacy_risk(
+                &manager,
+                risk_acceptance(id, expected, true, true, "Reviewed the retained owner", "INC-1234")
+            )
+            .await,
+        Err(MrfLifecycleControlError::BucketIdentityUnresolved)
+    ));
+    assert_eq!(runtime.partial_writes.checkpoint_records(super::super::super::intent_digest), before);
+    assert_eq!(
+        storage.calls.load(Ordering::Relaxed),
+        0,
+        "a guessed identity cannot bypass explicit refresh"
+    );
+}
+
+#[test]
+fn lifecycle_v2_reader_keeps_first_hold_across_repeated_incarnation_refresh() {
+    let (mut runtime, key) = restored_v2_runtime(0, 3);
+    let id = runtime.partial_writes.entries[&key].responsibility_id;
+    let first = Uuid::new_v4();
+    let second = Uuid::new_v4();
+    runtime
+        .partial_writes
+        .replace_observed_incarnation(id, None, first, None)
+        .expect("first observation");
+    assert!(matches!(
+        runtime.partial_writes.entries[&key].state,
+        ResponsibilityState::LegacyGenerationUnknown { detected_at_ms: 7, .. }
+    ));
+    runtime
+        .partial_writes
+        .replace_observed_incarnation(id, Some(first), second, None)
+        .expect("second observation of the same retained responsibility");
+    let row = runtime
+        .list_unverified_legacy(None, 1)
+        .expect("list the retained responsibility")
+        .responsibilities
+        .remove(0);
+    assert_eq!(row.responsibility_id, id);
+    assert_eq!(row.source_bucket_incarnation_id, None);
+    assert_eq!(row.bucket_incarnation_id, Some(second));
+    assert_eq!(
+        row.held_since_ms,
+        Some(7),
+        "changing the observed bucket cannot reset when this responsibility was first retained"
+    );
+}
@@ -48,6 +48,7 @@ fn runtime() -> MrfRuntime {
        config,
        checkpoint_owner: Uuid::new_v4(),
        next_checkpoint_sequence: 1,
+        lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
        new_since_flush: 0,
        dirty: false,
        journal_on_disk: false,
@@ -48,7 +48,7 @@ fn unknown_source_deleted_bucket_lifecycle_rejects_unknown_state_version_and_fie
    let valid = lifecycle_frame(&payload);
    assert!(decode_mrf_lifecycle_checkpoint(&valid, 8192).is_ok());
    let mut future_version = valid;
-    future_version[8] += 1;
+    future_version[8] = MRF_LIFECYCLE_FORMAT_VERSION_V2 + 1;
    assert!(matches!(
        decode_mrf_lifecycle_checkpoint(&future_version, 8192),
        Err(MrfReplayError::Snapshot(snapshot::SnapshotError::Unsupported))
@@ -176,6 +176,7 @@ async fn assert_unreadable_lifecycle_is_retained(corrupt: fn(&mut Value)) {
        config,
        checkpoint_owner: Uuid::new_v4(),
        next_checkpoint_sequence: replay.next_checkpoint_sequence,
+        lifecycle_format_version: replay.lifecycle_format_version,
        new_since_flush: 0,
        dirty: true,
        journal_on_disk: replay.journal_on_disk,
@@ -0,0 +1,359 @@
+// 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_reader_frame(payload: &Value, envelope_version: u8) -> Vec<u8> {
+    let payload = serde_json::to_vec(payload).expect("encode reader fixture");
+    let mut encoded = Vec::from(MRF_LIFECYCLE_MAGIC.as_slice());
+    encoded.push(envelope_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
+}
+
+fn unknown_generation_payload() -> Value {
+    let intent_digest = [7_u8; 32];
+    json!({
+        "format_version": 2,
+        "checkpoint_owner": Uuid::new_v4(),
+        "checkpoint_sequence": 1,
+        "records": [{
+            "intent_digest": intent_digest,
+            "responsibility_id": Uuid::new_v4(),
+            "source_bucket_incarnation_id": null,
+            "last_operator_acceptance": null,
+            "state": {
+                "state": "source_and_current_generation_unknown",
+                "since_ms": 7,
+                "next_probe_at_ms": 60_007,
+                "probe_attempts": 3
+            }
+        }]
+    })
+}
+
+#[test]
+fn lifecycle_v2_reader_accepts_retained_unknown_generation_without_a_source_identity() {
+    let payload = unknown_generation_payload();
+    let decoded = decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&payload, 2), 8192)
+        .expect("the reader must understand a valid retained V2 responsibility");
+    assert_eq!(decoded.format_version, 2);
+    assert_eq!(decoded.records.len(), 1);
+    assert!(decoded.records[0].is_valid());
+    assert_eq!(decoded.records[0].source_bucket_incarnation_id, None);
+    assert_eq!(
+        serde_json::to_value(&decoded.records[0].state).expect("serialize restored state"),
+        payload["records"][0]["state"],
+        "reading must preserve the first hold and bounded retry information"
+    );
+}
+
+#[test]
+fn lifecycle_v2_reader_rejects_mislabeled_and_unsupported_formats() {
+    let payload = unknown_generation_payload();
+    for (payload_version, envelope_version) in [(1, 1), (1, 2), (2, 1), (3, 3)] {
+        let mut invalid = payload.clone();
+        invalid["format_version"] = json!(payload_version);
+        assert!(decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&invalid, envelope_version), 8192).is_err());
+    }
+    let decoded = decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&payload, 2), 8192).expect("valid V2 fixture");
+    for version in [0, 1, 3] {
+        assert!(
+            encode_mrf_lifecycle_checkpoint(version, decoded.checkpoint_owner, 1, decoded.records.clone(), 8192).is_none(),
+            "V2 state cannot be written under format {version}"
+        );
+    }
+    let encoded = encode_mrf_lifecycle_checkpoint(2, decoded.checkpoint_owner, 1, decoded.records.clone(), 8192)
+        .expect("preserve an observed V2 owner");
+    assert_eq!(encoded[8], 2);
+    assert_eq!(
+        decode_mrf_lifecycle_checkpoint(&encoded, 8192)
+            .expect("read V2 round trip")
+            .records,
+        decoded.records
+    );
+}
+
+#[test]
+fn lifecycle_v2_reader_rejects_invalid_identity_retry_fields_and_torn_frames() {
+    let payload = unknown_generation_payload();
+    for source in [Uuid::nil(), Uuid::new_v4()] {
+        let mut invalid = payload.clone();
+        invalid["records"][0]["source_bucket_incarnation_id"] = json!(source);
+        assert!(decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&invalid, 2), 8192).is_err());
+    }
+    for field in ["since_ms", "next_probe_at_ms", "probe_attempts"] {
+        for replacement in [None, Some(json!(-1)), Some(json!("unknown"))] {
+            let mut invalid = payload.clone();
+            let state = invalid["records"][0]["state"].as_object_mut().expect("state object");
+            match replacement {
+                Some(value) => {
+                    state.insert(field.to_string(), value);
+                }
+                None => {
+                    state.remove(field);
+                }
+            }
+            assert!(
+                decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&invalid, 2), 8192).is_err(),
+                "reject {field}"
+            );
+        }
+    }
+    let mut invalid = payload.clone();
+    invalid["records"][0]["state"]["probe_attempts"] = json!(u64::MAX);
+    assert!(decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&invalid, 2), 8192).is_err());
+    invalid = payload.clone();
+    invalid["records"][0]["state"]["current_incarnation"] = json!(Uuid::new_v4());
+    assert!(decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&invalid, 2), 8192).is_err());
+    let encoded = lifecycle_reader_frame(&payload, 2);
+    for length in [0, 48, encoded.len() - 1] {
+        assert!(decode_mrf_lifecycle_checkpoint(&encoded[..length], 8192).is_err());
+    }
+    assert!(decode_mrf_lifecycle_checkpoint(&encoded, encoded.len() - 1).is_err());
+    let mut corrupt = encoded;
+    *corrupt.last_mut().expect("frame body") ^= 1;
+    assert!(decode_mrf_lifecycle_checkpoint(&corrupt, 8192).is_err());
+}
+
+#[test]
+#[serial]
+fn lifecycle_v2_reader_replays_real_files_without_promoting_v1_or_downgrading_v2() {
+    const STACK_SIZE: usize = 8 * 1024 * 1024;
+    std::thread::Builder::new()
+        .name("mrf-lifecycle-reader".to_owned())
+        .stack_size(STACK_SIZE)
+        .spawn(|| {
+            tokio::runtime::Builder::new_multi_thread()
+                .worker_threads(2)
+                .thread_stack_size(STACK_SIZE)
+                .enable_all()
+                .build()
+                .expect("reader fixture runtime")
+                .block_on(async {
+                    for version in [1, 2] {
+                        check_reader_restart(version).await;
+                    }
+                });
+        })
+        .expect("reader fixture thread")
+        .join()
+        .expect("reader fixture must finish");
+}
+
+async fn check_reader_restart(version: u8) {
+    let env = rustfs_test_utils::TestECStoreEnv::builder()
+        .prefix("rustfs_mrf_lifecycle_reader")
+        .build()
+        .await;
+    let config = MrfConsumerConfig::default();
+    let owner = Uuid::new_v4();
+    let bucket = "reader-unknown-bucket";
+    let state = if version == 2 {
+        unknown_generation_payload()["records"][0]["state"].clone()
+    } else {
+        json!({"state": "active"})
+    };
+    let mut payload = Vec::new();
+    let mut records = Vec::new();
+    for index in 0..3 {
+        let mut original = intent(bucket, &format!("retained-{index}"), 0);
+        original.kind = MrfKind::PartialWrite;
+        original.scope = Some(rustfs_common::mrf_channel::MrfScope {
+            pool_index: 0,
+            set_index: 0,
+        });
+        payload.extend(encoded_payload(&original));
+        records.push(json!({
+            "intent_digest": intent_digest(&original).expect("fixture identity"),
+            "responsibility_id": Uuid::new_v4(),
+            "source_bucket_incarnation_id": null,
+            "last_operator_acceptance": null,
+            "state": state
+        }));
+    }
+    let lifecycle = lifecycle_reader_frame(
+        &json!({
+            "format_version": version,
+            "checkpoint_owner": owner,
+            "checkpoint_sequence": 1,
+            "records": records
+        }),
+        version,
+    );
+    snapshot::publish_committed_snapshot_with_companion(
+        &journal_disks().await,
+        owner,
+        1,
+        &payload,
+        config.journal_max_bytes,
+        Some((&MRF_LIFECYCLE_PATHS, &lifecycle, config.journal_max_bytes * 4)),
+    )
+    .await
+    .expect("publish the reader fixture on real disks");
+    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 ids: HashSet<_> = records
+        .iter()
+        .map(|record| serde_json::from_value::<Uuid>(record["responsibility_id"].clone()).expect("fixture id"))
+        .collect();
+    let (incoming, receiver) = mpsc::channel(4);
+    let (_durable, durable_receiver) = mpsc::channel(4);
+    let (commands, lifecycle_receiver) = mpsc::channel(4);
+    let consumer = tokio::spawn(run_mrf_consumer(manager.clone(), receiver, durable_receiver, lifecycle_receiver));
+    let (response, listed) = tokio::sync::oneshot::channel();
+    commands
+        .send(MrfLifecycleCommand::List {
+            after: None,
+            limit: 3,
+            response,
+        })
+        .await
+        .expect("query the real consumer after its startup replay");
+    let listing = tokio::time::timeout(Duration::from_secs(5), listed)
+        .await
+        .expect("consumer listing deadline")
+        .expect("consumer listing response")
+        .expect("V1 and V2 owners must be readable");
+    let successor_owner = listing.checkpoint_owner.expect("the real consumer must publish a successor");
+    let successor_sequence = listing.checkpoint_sequence.expect("successor sequence");
+    assert_ne!(successor_owner, owner);
+    assert!(successor_sequence > 1);
+    assert_eq!(
+        listing
+            .responsibilities
+            .iter()
+            .map(|row| row.responsibility_id)
+            .collect::<HashSet<_>>(),
+        ids
+    );
+    drop(incoming);
+    tokio::time::timeout(Duration::from_secs(5), consumer)
+        .await
+        .expect("consumer shutdown deadline")
+        .expect("consumer shutdown");
+    let published = read_mrf_lifecycle_checkpoint(
+        successor_owner,
+        successor_sequence,
+        config.journal_max_bytes,
+        config.journal_max_bytes * 4,
+    )
+    .await
+    .expect("read the actual consumer's successor")
+    .expect("retained lifecycle companion");
+    assert_eq!(published.format_version, version);
+    assert_eq!(published.records.len(), 3);
+    let mut last_records = None;
+    for restart in 0..3 {
+        let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes);
+        let replay = replay_into(&manager, &mut queue, &mut None).await;
+        assert!(replay.source_readable);
+        assert_eq!(replay.partial_writes.len(), 3);
+        assert_eq!(replay.lifecycle_format_version, version);
+        let mut runtime = MrfRuntime {
+            queue,
+            partial_writes: PartialWrites::default(),
+            config: config.clone(),
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: replay.next_checkpoint_sequence,
+            lifecycle_format_version: replay.lifecycle_format_version,
+            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,
+        };
+        runtime.adopt_replayed_partial_writes(replay.partial_writes);
+        if let Some(previous) = &last_records {
+            let current: HashMap<_, _> = runtime
+                .partial_writes
+                .checkpoint_records(intent_digest)
+                .into_iter()
+                .map(|record| (record.responsibility_id, record))
+                .collect();
+            assert_eq!(&current, previous, "replay must preserve every state and audit field");
+        }
+        assert!(runtime.flush().await);
+        runtime.dispatch(&manager).await;
+        let mut cursor = None;
+        let mut visible = HashSet::new();
+        loop {
+            let page = runtime.list_unverified_legacy(cursor, 1).expect("list retained records");
+            assert_eq!(page.legacy_generation_unknown_count, 3);
+            assert_eq!(page.responsibilities.len(), 1);
+            let row = &page.responsibilities[0];
+            assert!(visible.insert(row.responsibility_id));
+            assert_eq!(row.source_bucket_incarnation_id, None);
+            assert_eq!((row.pool_index, row.set_index), (Some(0), Some(0)));
+            assert_eq!(row.version_id, Some(Uuid::from_bytes([7; 16]).to_string()));
+            if restart < 2 {
+                assert_eq!(row.bucket_incarnation_id, None);
+                assert_eq!(
+                    row.status,
+                    if version == 2 {
+                        "source_and_current_generation_unknown"
+                    } else {
+                        "legacy_generation_unobserved"
+                    }
+                );
+                assert_eq!(row.held_since_ms, (version == 2).then_some(7));
+                assert!(row.next_identity_probe_at_ms.is_some());
+            } else {
+                assert_eq!(row.status, "legacy_generation_unknown");
+                assert!(row.bucket_incarnation_id.is_some());
+                if version == 2 {
+                    assert_eq!(row.held_since_ms, Some(7));
+                }
+            }
+            cursor = page.next_cursor;
+            if cursor.is_none() {
+                break;
+            }
+        }
+        assert_eq!(visible, ids);
+        assert_eq!(manager.get_queue_length().await, 0, "unknown source never authorizes object healing");
+        if restart == 1 {
+            env.make_bucket(bucket, false).await;
+            for id in &ids {
+                runtime
+                    .refresh_legacy_generation(&manager, *id, None)
+                    .await
+                    .expect("record current identity without rebinding source");
+            }
+        }
+        let (checkpoint_owner, sequence) = runtime.runtime_checkpoint.expect("successor checkpoint");
+        let stored =
+            read_mrf_lifecycle_checkpoint(checkpoint_owner, sequence, config.journal_max_bytes, config.journal_max_bytes * 4)
+                .await
+                .expect("read actual successor files")
+                .expect("retained companion");
+        assert_eq!(
+            stored.format_version, version,
+            "owner transfer must preserve the format even after observing a bucket"
+        );
+        last_records = Some(
+            stored
+                .records
+                .into_iter()
+                .map(|record| (record.responsibility_id, record))
+                .collect::<HashMap<_, _>>(),
+        );
+    }
+}
@@ -12,7 +12,9 @@

 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.
+The listing exposes `lastIdentityProbeResult` (`unresolved`, `unavailable`, or `known`), `lastIdentityProbeAtMs`, and `nextIdentityProbeAtMs` when available. The result and last-probe timestamp are process-local observations. For lifecycle V1, restart resets the probe delay; these timestamps do not establish when a responsibility was first retained.
+
+A lifecycle V2 checkpoint can contain `source_and_current_generation_unknown`. It retains the first-held timestamp, next-probe timestamp, and attempt count. Both incarnation fields remain null, and the listing exposes the first-held timestamp as `heldSinceMs`. Restart restores the remaining delay within five to 60 seconds, including after clock changes or an expired deadline. Identity probes retain their schedule after unsuccessful lookups. Risk acceptance and targeted recheck require a known current incarnation; a guessed identity cannot bypass refresh.

 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.

@@ -28,6 +30,8 @@ Do not delete a companion, journal, or commit manifest to clear this condition.

 ## 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.
+Owners that have not read a V2 checkpoint continue 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 admission or 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`.
+The reader accepts lifecycle formats 1 and 2. Replaying a V2 owner preserves that format in successor checkpoints, even after the current bucket becomes visible; it never writes the new state inside a V1 envelope. A valid V2 unknown-generation state cannot contain a source incarnation. Unknown fields, missing retry fields, unsupported versions, and mismatched envelope/payload versions remain unreadable.
+
+Creating the new state requires a separate writer rollout. The existing heal-control and cross-pool protocol capabilities do not prove support for it. New writers require pinned read/restart evidence for every supported reader and an explicit minimum rollback reader. Reader support alone does not qualify writer activation; do not downgrade past that reader or silently convert an unknown state to `Active`.