test(e2e): restore distributed workflow contract

.config/nextest.toml

@@ -906,6 +906,11 @@ test-group = 'e2e-reliability'
 filter = 'package(e2e_test) & test(/^inline_fast_path_cluster_test::/)'
 test-group = 'e2e-inline-boundaries'
 
+[[profile.e2e-full.overrides]]
+filter = 'package(e2e_test) & test(=inline_fast_path_cluster_test::four_node_add_tier_converges)'
+success-output = "immediate"
+junit.store-success-output = true
+
 # Heal interruption cases start whole multi-node clusters. Bound their overlap
 # just as the nightly lane does, without changing membership or retry policy.
 [[profile.e2e-full.overrides]]

.github/workflows/ci.yml

@@ -990,6 +990,7 @@ jobs:
       # debug binary; each test spawns its own rustfs server on a random port.
       - name: Run e2e full suite
         env:
+          RUSTFS_E2E_ADD_TIER_LOG: "rustfs=info,rustfs_notify=debug,rustfs_ecstore::object_lock_diag=trace,rustfs_lock=debug,rustfs_ecstore::cluster::rpc::remote_locker=debug"
           CARGO_BIN_EXE_rustfs: ${{ runner.temp }}/rustfs-startup-cas-input/rustfs
           RUSTFS_E2E_STARTUP_CAS_BINARY: ${{ runner.temp }}/rustfs-startup-cas-input/rustfs
           RUSTFS_E2E_STARTUP_CAS_BUILD_MANIFEST: ${{ runner.temp }}/rustfs-startup-cas-input/rustfs.e2e-startup-cas-build.json

crates/config/src/constants/heal.rs

@@ -197,6 +197,10 @@ pub const ENV_HEAL_MRF_REPLAY_BATCH: &str = "RUSTFS_HEAL_MRF_REPLAY_BATCH";
 /// Default behavior keeps the MRF intent pipeline enabled.
 pub const DEFAULT_HEAL_MRF_ENABLE: bool = true;
 
+/// Opt in to lifecycle V2 transitions after qualifying every reader and the rollback binary.
+pub const ENV_MRF_LIFECYCLE_V2_ENABLE: &str = "RUSTFS_MRF_LIFECYCLE_V2_ENABLE";
+pub const DEFAULT_MRF_LIFECYCLE_V2_ENABLE: bool = false;
+
 /// Default MRF queue capacity (matches MinIO's 100k MRF list ceiling).
 pub const DEFAULT_HEAL_MRF_QUEUE_SIZE: usize = 100_000;
 

crates/e2e_test/src/inline_fast_path_cluster_test.rs

@@ -2147,6 +2147,12 @@ async fn four_node_add_tier_converges() -> TestResult {
 
     let mut hot = RustFSTestClusterEnvironment::new(4).await?;
     hot.set_env(ALLOW_LOOPBACK_TIER_ENDPOINT_ENV, "true");
+    if let Ok(log_filter) = std::env::var("RUSTFS_E2E_ADD_TIER_LOG") {
+        hot.set_env("RUST_LOG", log_filter);
+        hot.set_env("RUSTFS_OBJECT_LOCK_DIAG_ENABLE", "true");
+        hot.set_env("RUSTFS_OBJECT_LOCK_DIAG_SLOW_ACQUIRE_MS", "1");
+        hot.set_env("RUSTFS_OBJECT_LOCK_DIAG_SLOW_HOLD_MS", "1");
+    }
     hot.start().await?;
 
     let tier_name = unique_tier_name();

crates/ecstore/src/api/mod.rs

@@ -513,13 +513,16 @@ pub mod metrics {
 pub mod notification {
     pub use crate::services::notification_sys::{
         ClusterTierDailyStats, CrossPoolFenceFleetProofToken, IlmRecoveryExportFleetProofToken,
-        LegacyTransitionStateReconcileFleetProofToken, NotificationPeerErr, NotificationSys, ScannerPublicationLeaseGrant,
-        acquire_cross_pool_fence_fleet_proof, acquire_ilm_recovery_export_fleet_proof,
-        acquire_legacy_transition_state_reconcile_fleet_proof, configured_fleet_topology_fingerprint,
-        cross_pool_fence_fleet_proof_matches, get_global_notification_sys, ilm_recovery_export_fleet_proof_matches,
-        ilm_recovery_export_local_process_epoch, ilm_recovery_export_member_epochs_sha256,
-        ilm_recovery_export_topology_generation, legacy_transition_state_reconcile_fleet_proof_matches,
-        new_global_notification_sys, scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe,
+        LegacyTransitionStateReconcileFleetProofToken, MrfLifecycleWriterFleetProofToken, NotificationPeerErr, NotificationSys,
+        ScannerPublicationLeaseGrant, acquire_cross_pool_fence_fleet_proof, acquire_ilm_recovery_export_fleet_proof,
+        acquire_legacy_transition_state_reconcile_fleet_proof, acquire_mrf_lifecycle_writer_fleet_proof,
+        configured_fleet_topology_fingerprint, cross_pool_fence_fleet_proof_matches, get_global_notification_sys,
+        ilm_recovery_export_fleet_proof_matches, ilm_recovery_export_local_process_epoch,
+        ilm_recovery_export_member_epochs_sha256, ilm_recovery_export_topology_generation,
+        legacy_transition_state_reconcile_fleet_proof_matches, mrf_lifecycle_reader_local_process_epoch,
+        mrf_lifecycle_writer_fleet_proof_current, mrf_lifecycle_writer_fleet_proof_matches, new_global_notification_sys,
+        scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe,
+        with_mrf_lifecycle_writer_fleet_proof,
     };
     #[cfg(any(test, feature = "test-util"))]
     pub use crate::services::notification_sys::{

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -631,6 +631,17 @@ fn decode_cross_pool_fence_capability(expected_member: &str, result: &[u8]) -> R
     Ok((version, epoch))
 }
 
+fn decode_mrf_lifecycle_reader_capability(expected_member: &str, result: &[u8]) -> Result<(u8, Uuid)> {
+    let (&maximum_format_version, identity) = result
+        .split_first()
+        .ok_or_else(|| Error::other("peer returned a truncated MRF lifecycle reader capability"))?;
+    if maximum_format_version == 0 || expected_member.trim().is_empty() {
+        return Err(Error::other("peer returned an invalid MRF lifecycle reader capability"));
+    }
+    let epoch = decode_remote_version_state_capability(expected_member, identity)?;
+    Ok((maximum_format_version, epoch))
+}
+
 #[derive(Clone, Debug)]
 pub struct PeerLiveEventsBatch {
     pub events: Vec<u8>,
@@ -1964,6 +1975,16 @@ impl PeerRestClient {
         Ok((self.topology_member.clone(), supported_version, epoch))
     }
 
+    /// Reports one peer's reader support without authorizing a lifecycle writer or rollback.
+    pub async fn probe_mrf_lifecycle_reader(&self, topology_fingerprint: String) -> Result<(String, u8, Uuid)> {
+        let probe = rustfs_protos::mrf_lifecycle_reader_capability_probe(Uuid::new_v4().as_bytes());
+        let result = self
+            .heal_control(rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, topology_fingerprint, probe)
+            .await?;
+        let (maximum_format_version, epoch) = decode_mrf_lifecycle_reader_capability(&self.topology_member, &result)?;
+        Ok((self.topology_member.clone(), maximum_format_version, epoch))
+    }
+
     pub async fn probe_ilm_recovery_export(&self, topology_fingerprint: String) -> Result<(String, Uuid)> {
         let probe = rustfs_protos::ilm_recovery_export_capability_probe(Uuid::new_v4().as_bytes());
         let result = self
@@ -4333,6 +4354,49 @@ mod tests {
         }
     }
 
+    #[test]
+    fn mrf_lifecycle_reader_capability_decoder_preserves_the_advertised_version() {
+        let epoch = Uuid::new_v4();
+        for version in [1, 2, u8::MAX] {
+            let result = rustfs_protos::encode_mrf_lifecycle_reader_capability(version, "node-a:9000", epoch.as_bytes())
+                .expect("capability frame");
+            assert_eq!(
+                decode_mrf_lifecycle_reader_capability("node-a:9000", &result).expect("reported capability"),
+                (version, epoch)
+            );
+        }
+    }
+
+    #[test]
+    fn mrf_lifecycle_reader_capability_decoder_rejects_unbound_or_malformed_identity() {
+        let epoch = Uuid::new_v4();
+        let valid =
+            rustfs_protos::encode_mrf_lifecycle_reader_capability(2, "node-a:9000", epoch.as_bytes()).expect("capability frame");
+        let mut invalid = vec![Vec::new(), vec![2], valid[..valid.len() - 1].to_vec()];
+        let mut zero = valid.clone();
+        zero[0] = 0;
+        invalid.push(zero);
+        let mut trailing = valid.clone();
+        trailing.push(0);
+        invalid.push(trailing);
+        let mut length_overflow = valid.clone();
+        length_overflow[1..9].copy_from_slice(&u64::MAX.to_be_bytes());
+        invalid.push(length_overflow);
+        let mut invalid_utf8 = valid.clone();
+        invalid_utf8[9] = 0xff;
+        invalid.push(invalid_utf8);
+        invalid.push(
+            rustfs_protos::encode_mrf_lifecycle_reader_capability(2, "node-a:9000", Uuid::nil().as_bytes())
+                .expect("nil epoch fixture"),
+        );
+        for result in invalid {
+            assert!(decode_mrf_lifecycle_reader_capability("node-a:9000", &result).is_err());
+        }
+        assert!(decode_mrf_lifecycle_reader_capability("node-b:9000", &valid).is_err());
+        let empty = rustfs_protos::encode_mrf_lifecycle_reader_capability(2, "", epoch.as_bytes()).expect("empty member fixture");
+        assert!(decode_mrf_lifecycle_reader_capability("", &empty).is_err());
+    }
+
     #[test]
     fn remote_version_state_capability_decoder_fails_closed() {
         let epoch = Uuid::new_v4();

crates/ecstore/src/disk/local.rs

@@ -11348,6 +11348,19 @@ impl DiskAPI for LocalDisk {
         expected: Option<Bytes>,
         replacement: Option<Bytes>,
     ) -> Result<ConditionalFileUpdate> {
+        let authority = crate::services::notification_sys::current_mrf_lifecycle_publication();
+        let commit_allowed = move || {
+            if authority
+                .as_ref()
+                .is_some_and(|proof| !crate::services::notification_sys::mrf_lifecycle_writer_fleet_proof_current(proof))
+            {
+                return Err(DiskError::conditional_file_not_committed(std::io::Error::new(
+                    ErrorKind::PermissionDenied,
+                    "MRF lifecycle writer authority expired or was revoked",
+                )));
+            }
+            Ok(())
+        };
         #[cfg(unix)]
         {
             use rustix::fs::{FlockOperation, flock};
@@ -11357,6 +11370,7 @@ impl DiskAPI for LocalDisk {
             let path = path.to_string();
             let sync_metadata = effective_durability(volume).syncs_commit_metadata();
             return Ok(tokio::task::spawn_blocking(move || {
+                commit_allowed()?;
                 // A persistent directory lock bounds metadata growth. Removing
                 // per-target lock files can split flock ownership across inodes.
                 let lock_path = file_path
@@ -11436,6 +11450,14 @@ impl DiskAPI for LocalDisk {
                                 if sync_metadata {
                                     staged.sync_all().map_err(not_committed)?;
                                 }
+                                #[cfg(any(test, feature = "test-util"))]
+                                os::prepared_publication_test_hooks::run(
+                                    os::prepared_publication_test_hooks::Stage::Rename,
+                                    &file_path,
+                                );
+                                // The file CAS lock precedes the brief proof-slot
+                                // read; proof updates never acquire disk locks.
+                                commit_allowed()?;
                                 std::fs::rename(&temporary, &file_path)?;
                                 Ok(())
                             })();
@@ -11449,6 +11471,7 @@ impl DiskAPI for LocalDisk {
                             }
                         }
                         None => {
+                            commit_allowed()?;
                             std::fs::remove_file(&file_path)?;
                             if sync_metadata
                                 && let Some(parent) = file_path.parent()
@@ -11480,6 +11503,7 @@ impl DiskAPI for LocalDisk {
                     replacement.as_deref(),
                     sync_metadata,
                     &publication_root,
+                    commit_allowed,
                 )
             })
             .await
@@ -11488,7 +11512,7 @@ impl DiskAPI for LocalDisk {
 
         #[cfg(not(any(unix, windows)))]
         {
-            let _ = (volume, path, expected, replacement);
+            let _ = (volume, path, expected, replacement, commit_allowed);
             Err(DiskError::MethodNotAllowed)
         }
     }

crates/ecstore/src/disk/os.rs

@@ -4162,6 +4162,17 @@ impl RenameDestinationPathGuard {
         data: &[u8],
         sync_file: bool,
         sync_parent: bool,
+    ) -> io::Result<()> {
+        self.write_file_for_path_access_guarded(file_path, data, sync_file, sync_parent, || Ok(()))
+    }
+
+    fn write_file_for_path_access_guarded(
+        &self,
+        file_path: &Path,
+        data: &[u8],
+        sync_file: bool,
+        sync_parent: bool,
+        before_publish: impl Fn() -> io::Result<()>,
     ) -> io::Result<()> {
         #[cfg(windows)]
         {
@@ -4198,7 +4209,9 @@ impl RenameDestinationPathGuard {
             // the separate publication handle as the identity anchor and close
             // the writer before issuing the handle-relative rename.
             drop(writer);
-            if let Err(rename_err) = rename_windows_prepared(file_path, &self._directory_guard, &publication, 0) {
+            if let Err(rename_err) =
+                before_publish().and_then(|()| rename_windows_prepared(file_path, &self._directory_guard, &publication, 0))
+            {
                 if let Err(cleanup_err) = set_windows_file_delete_on_close(&publication, true) {
                     return Err(io::Error::new(
                         rename_err.kind(),
@@ -4217,6 +4230,7 @@ impl RenameDestinationPathGuard {
         #[cfg(not(windows))]
         {
             let _ = self;
+            before_publish()?;
             let mut file = std::fs::OpenOptions::new()
                 .create(true)
                 .write(true)
@@ -4935,6 +4949,7 @@ pub(crate) fn compare_and_update_control_file(
     replacement: Option<&[u8]>,
     sync_metadata: bool,
     publication_root: &PublicationRoot,
+    before_publish: impl Fn() -> io::Result<()>,
 ) -> io::Result<ConditionalFileUpdate> {
     use windows_sys::{
         Wdk::Storage::FileSystem::{
@@ -4984,7 +4999,7 @@ pub(crate) fn compare_and_update_control_file(
             directory: parent.to_path_buf(),
             _directory_guard: parent_guard,
         }
-        .write_file_for_path_access(file_path, replacement, sync_metadata, sync_metadata)?,
+        .write_file_for_path_access_guarded(file_path, replacement, sync_metadata, sync_metadata, before_publish)?,
         None => {
             let file_name = file_path
                 .file_name()
@@ -5000,6 +5015,7 @@ pub(crate) fn compare_and_update_control_file(
                 true,
             )?;
             validate_windows_owned_file(&file)?;
+            before_publish()?;
             set_windows_file_delete_on_close(&file, true)?;
             drop(file);
             if sync_metadata {

crates/ecstore/src/services/notification_sys.rs

@@ -61,6 +61,7 @@ const TIER_CONFIG_RELOAD_RETRY_CAP: Duration = Duration::from_secs(5);
 const REMOTE_VERSION_STATE_PROBE_INTERVAL: Duration = Duration::from_secs(10);
 const REMOTE_VERSION_STATE_PROBE_TIMEOUT: Duration = Duration::from_secs(5);
 const REMOTE_VERSION_STATE_PROOF_TTL: Duration = Duration::from_secs(30);
+const MRF_LIFECYCLE_WRITER_FORMAT_VERSION: u8 = 2;
 /// Fallback poll cadence while startup has not yet published the notification
 /// system. IAM finalization starts the probe before `init_notification_runtime`
 /// runs, so the first pass always fails closed; waiting a full probe interval
@@ -357,6 +358,8 @@ static DECOMMISSION_TARGET_FENCE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCa
 static LEGACY_TRANSITION_STATE_RECONCILE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static ILM_RECOVERY_EXPORT_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static TRANSITION_TRANSACTION_COMPACTION_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
+static MRF_LIFECYCLE_WRITER_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
+static MRF_LIFECYCLE_READER_LOCAL_PROCESS_EPOCH: LazyLock<Uuid> = LazyLock::new(Uuid::new_v4);
 static REMOTE_VERSION_STATE_PROBE_TOPOLOGY: OnceLock<String> = OnceLock::new();
 /// Signalled once `GLOBAL_NOTIFICATION_SYS` is published so the fleet probe can
 /// run its first real pass without waiting for the bootstrap poll.
@@ -391,7 +394,7 @@ fn transition_transaction_compaction_fleet_proof_slot() -> &'static std::sync::R
     TRANSITION_TRANSACTION_COMPACTION_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
 }
 
-fn all_fleet_capability_proof_slots() -> [&'static std::sync::RwLock<FleetCapabilityProofState>; 7] {
+fn all_fleet_capability_proof_slots() -> [&'static std::sync::RwLock<FleetCapabilityProofState>; 8] {
     [
         remote_version_state_fleet_proof_slot(),
         cross_pool_fence_fleet_proof_slot(),
@@ -400,6 +403,7 @@ fn all_fleet_capability_proof_slots() -> [&'static std::sync::RwLock<FleetCapabi
         legacy_transition_state_reconcile_fleet_proof_slot(),
         ilm_recovery_export_fleet_proof_slot(),
         transition_transaction_compaction_fleet_proof_slot(),
+        mrf_lifecycle_writer_fleet_proof_slot(),
     ]
 }
 
@@ -1286,6 +1290,161 @@ fn insert_remote_version_state_peer(peer_epochs: &mut BTreeMap<String, Uuid>, pe
     Ok(())
 }
 
+/// Authority for one MRF lifecycle format transition. The permit remains
+/// owned by the publisher until every admitted disk operation has finished.
+pub struct MrfLifecycleWriterFleetProofToken {
+    token: FleetCapabilityProofToken,
+    _permit: FleetCapabilityProofPermit,
+    local_reader_max_format: u8,
+    expires_at: Instant,
+}
+
+tokio::task_local! {
+    static MRF_LIFECYCLE_PUBLICATION: Arc<MrfLifecycleWriterFleetProofToken>;
+}
+
+/// Carry one publication's authority into local disk executors. Native CAS
+/// workers retain the same permit after an async timeout or cancellation.
+pub async fn with_mrf_lifecycle_writer_fleet_proof<F: std::future::Future>(
+    proof: Arc<MrfLifecycleWriterFleetProofToken>,
+    future: F,
+) -> F::Output {
+    MRF_LIFECYCLE_PUBLICATION.scope(proof, future).await
+}
+
+pub(crate) fn current_mrf_lifecycle_publication() -> Option<Arc<MrfLifecycleWriterFleetProofToken>> {
+    MRF_LIFECYCLE_PUBLICATION.try_with(Arc::clone).ok()
+}
+
+pub fn mrf_lifecycle_writer_fleet_proof_current(proof: &MrfLifecycleWriterFleetProofToken) -> bool {
+    let Some(topology) = configured_fleet_topology_fingerprint() else {
+        return false;
+    };
+    let state = mrf_lifecycle_writer_fleet_proof_slot()
+        .read()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    mrf_lifecycle_writer_fleet_proof_matches_at(&state, proof, topology, Instant::now())
+}
+
+fn mrf_lifecycle_writer_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
+    MRF_LIFECYCLE_WRITER_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
+}
+
+pub fn mrf_lifecycle_reader_local_process_epoch() -> Uuid {
+    *MRF_LIFECYCLE_READER_LOCAL_PROCESS_EPOCH
+}
+
+/// The caller supplies its decoder's actual maximum, not a deployment setting.
+/// Operator opt-in and rollback qualification remain the caller's responsibility.
+pub async fn acquire_mrf_lifecycle_writer_fleet_proof(local_reader_max_format: u8) -> Option<MrfLifecycleWriterFleetProofToken> {
+    let expected_topology = configured_fleet_topology_fingerprint()?;
+    let observed_at = Instant::now();
+    let observed = observe_mrf_lifecycle_reader_fleet(expected_topology, local_reader_max_format).await;
+    if publish_fleet_capability_probe_result(mrf_lifecycle_writer_fleet_proof_slot(), expected_topology, observed, observed_at)
+        .is_some()
+    {
+        return None;
+    }
+    let state = mrf_lifecycle_writer_fleet_proof_slot()
+        .read()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    acquire_mrf_lifecycle_writer_fleet_proof_from(&state, expected_topology, local_reader_max_format, Instant::now())
+}
+
+fn acquire_mrf_lifecycle_writer_fleet_proof_from(
+    state: &FleetCapabilityProofState,
+    expected_topology: &str,
+    local_reader_max_format: u8,
+    now: Instant,
+) -> Option<MrfLifecycleWriterFleetProofToken> {
+    if local_reader_max_format < MRF_LIFECYCLE_WRITER_FORMAT_VERSION {
+        return None;
+    }
+    let token = acquire_fleet_capability_proof_from(state, expected_topology, now)?;
+    let permit = state.proof.as_ref()?.generation.try_acquire()?;
+    Some(MrfLifecycleWriterFleetProofToken {
+        token,
+        _permit: permit,
+        local_reader_max_format,
+        expires_at: state.proof.as_ref()?.expires_at,
+    })
+}
+
+fn mrf_lifecycle_writer_fleet_proof_matches_at(
+    state: &FleetCapabilityProofState,
+    proof: &MrfLifecycleWriterFleetProofToken,
+    expected_topology: &str,
+    now: Instant,
+) -> bool {
+    now < proof.expires_at
+        && proof._permit.generation.is_accepting()
+        && fleet_capability_proof_matches_at(state, &proof.token, expected_topology, now)
+        && state
+            .proof
+            .as_ref()
+            .is_some_and(|current| Arc::ptr_eq(&current.generation, &proof._permit.generation))
+}
+
+/// Reobserve every member before a manifest commit. No proof-slot lock is held
+/// during network I/O, and any failed or changed observation revokes admission.
+pub async fn mrf_lifecycle_writer_fleet_proof_matches(proof: &MrfLifecycleWriterFleetProofToken) -> bool {
+    let Some(expected_topology) = configured_fleet_topology_fingerprint() else {
+        return false;
+    };
+    {
+        let state = mrf_lifecycle_writer_fleet_proof_slot()
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        if !mrf_lifecycle_writer_fleet_proof_matches_at(&state, proof, expected_topology, Instant::now()) {
+            return false;
+        }
+    }
+    let observed = observe_mrf_lifecycle_reader_fleet(expected_topology, proof.local_reader_max_format).await;
+    let mut state = mrf_lifecycle_writer_fleet_proof_slot()
+        .write()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    if !observed.is_ok_and(|observed| observed == *proof.token.peer_epochs) {
+        revoke_fleet_capability_proof_state(&mut state);
+        return false;
+    }
+    mrf_lifecycle_writer_fleet_proof_matches_at(&state, proof, expected_topology, Instant::now())
+}
+
+async fn observe_mrf_lifecycle_reader_fleet(
+    topology_fingerprint: &str,
+    local_reader_max_format: u8,
+) -> Result<BTreeMap<String, Uuid>> {
+    let notification_sys =
+        get_global_notification_sys().ok_or_else(|| Error::other("MRF lifecycle reader fleet is not initialized"))?;
+    timeout(
+        REMOTE_VERSION_STATE_PROBE_TIMEOUT,
+        notification_sys.probe_mrf_lifecycle_reader_fleet(topology_fingerprint, local_reader_max_format),
+    )
+    .await
+    .map_err(|_| Error::other("MRF lifecycle reader fleet probe timed out"))?
+}
+
+fn mrf_lifecycle_reader_epoch_map(
+    expected_remote_members: &[String],
+    local_member: String,
+    local_reader_max_format: u8,
+    remote_readers: Vec<(String, u8, Uuid)>,
+) -> Result<BTreeMap<String, Uuid>> {
+    if local_member.trim().is_empty() || local_reader_max_format < MRF_LIFECYCLE_WRITER_FORMAT_VERSION {
+        return Err(Error::other("MRF lifecycle local reader does not support the writer format"));
+    }
+    let mut peer_epochs = BTreeMap::new();
+    insert_remote_version_state_peer(&mut peer_epochs, local_member.clone(), mrf_lifecycle_reader_local_process_epoch())?;
+    for (member, maximum_format, epoch) in remote_readers {
+        if maximum_format < MRF_LIFECYCLE_WRITER_FORMAT_VERSION {
+            return Err(Error::other("MRF lifecycle peer reader does not support the writer format"));
+        }
+        insert_remote_version_state_peer(&mut peer_epochs, member, epoch)?;
+    }
+    validate_capability_fleet_members(expected_remote_members, &local_member, &peer_epochs)?;
+    Ok(peer_epochs)
+}
+
 pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
     if REMOTE_VERSION_STATE_PROBE_TOPOLOGY.set(topology_fingerprint.clone()).is_err() {
         if REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get() != Some(&topology_fingerprint) {
@@ -1606,6 +1765,25 @@ impl NotificationSys {
         Ok((peer_epochs, minimum_version))
     }
 
+    async fn probe_mrf_lifecycle_reader_fleet(
+        &self,
+        topology_fingerprint: &str,
+        local_reader_max_format: u8,
+    ) -> Result<BTreeMap<String, Uuid>> {
+        if self.peer_clients.len() != self.peer_topology_hosts.len() {
+            return Err(Error::other("MRF lifecycle reader fleet membership is incomplete"));
+        }
+        let local_member = runtime_sources::local_node_name().await;
+        let probes = self.peer_clients.iter().map(|client| async {
+            let client = client
+                .as_ref()
+                .ok_or_else(|| Error::other("MRF lifecycle reader peer is unreachable"))?;
+            client.probe_mrf_lifecycle_reader(topology_fingerprint.to_string()).await
+        });
+        let readers = join_all(probes).await.into_iter().collect::<Result<Vec<_>>>()?;
+        mrf_lifecycle_reader_epoch_map(&self.peer_topology_hosts, local_member, local_reader_max_format, readers)
+    }
+
     async fn probe_ilm_recovery_export_fleet(&self, topology_fingerprint: &str) -> Result<BTreeMap<String, Uuid>> {
         if self.peer_clients.len() != self.peer_topology_hosts.len() {
             return Err(Error::other("ILM recovery export capability fleet membership is incomplete"));
@@ -1626,12 +1804,12 @@ impl NotificationSys {
             let (peer, epoch) = result?;
             insert_remote_version_state_peer(&mut peer_epochs, peer, epoch)?;
         }
-        validate_ilm_recovery_export_members(&self.peer_topology_hosts, &local_member, &peer_epochs)?;
+        validate_capability_fleet_members(&self.peer_topology_hosts, &local_member, &peer_epochs)?;
         Ok(peer_epochs)
     }
 }
 
-fn validate_ilm_recovery_export_members(
+fn validate_capability_fleet_members(
     expected_remote_members: &[String],
     local_member: &str,
     observed: &BTreeMap<String, Uuid>,
@@ -1642,7 +1820,7 @@ fn validate_ilm_recovery_export_members(
         .chain(std::iter::once(local_member.to_string()))
         .collect::<BTreeSet<_>>();
     if expected.len() != expected_remote_members.len().saturating_add(1) || observed.keys().ne(expected.iter()) {
-        return Err(Error::other("ILM recovery export capability fleet membership does not match topology"));
+        return Err(Error::other("fleet capability membership does not match topology"));
     }
     Ok(())
 }
@@ -4112,6 +4290,228 @@ mod tests {
         );
     }
 
+    #[test]
+    fn mrf_lifecycle_writer_requires_every_reader_and_the_local_decoder() {
+        let expected = vec!["peer-a".to_owned(), "peer-b".to_owned()];
+        let a = Uuid::new_v4();
+        let b = Uuid::new_v4();
+        let readers = vec![("peer-a".to_owned(), 2, a), ("peer-b".to_owned(), 3, b)];
+        let observed = mrf_lifecycle_reader_epoch_map(&expected, "local".to_owned(), 2, readers.clone())
+            .expect("all members have compatible readers");
+        assert_eq!(observed.len(), 3);
+        assert_eq!(observed["local"], mrf_lifecycle_reader_local_process_epoch());
+        assert_eq!(observed["peer-a"], a);
+        for version in [0, 1] {
+            assert!(mrf_lifecycle_reader_epoch_map(&expected, "local".to_owned(), version, readers.clone()).is_err());
+            let mut old_peer = readers.clone();
+            old_peer[0].1 = version;
+            assert!(mrf_lifecycle_reader_epoch_map(&expected, "local".to_owned(), 2, old_peer).is_err());
+        }
+        for invalid in [
+            vec![readers[0].clone()],
+            vec![("unknown".to_owned(), 2, a), readers[1].clone()],
+            vec![readers[0].clone(), ("peer-a".to_owned(), 2, b)],
+            vec![readers[0].clone(), ("peer-b".to_owned(), 2, a)],
+            vec![readers[0].clone(), ("peer-b".to_owned(), 2, Uuid::nil())],
+            vec![
+                readers[0].clone(),
+                ("peer-b".to_owned(), 2, mrf_lifecycle_reader_local_process_epoch()),
+            ],
+        ] {
+            assert!(mrf_lifecycle_reader_epoch_map(&expected, "local".to_owned(), 2, invalid).is_err());
+        }
+        assert!(mrf_lifecycle_reader_epoch_map(&expected, "".to_owned(), 2, readers).is_err());
+        let single = mrf_lifecycle_reader_epoch_map(&[], "local".to_owned(), 2, Vec::new())
+            .expect("a single node must still prove its local decoder and epoch");
+        assert_eq!(single.len(), 1);
+    }
+
+    #[test]
+    fn mrf_lifecycle_writer_rejects_expired_failed_and_conflicting_proofs() {
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let now = Instant::now();
+        let peers = BTreeMap::from([("local".to_owned(), Uuid::new_v4())]);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology", Ok(peers), now).is_none());
+        let token = {
+            let state = slot.read().expect("proof state");
+            assert!(acquire_mrf_lifecycle_writer_fleet_proof_from(&state, "topology", 1, now).is_none());
+            assert!(acquire_mrf_lifecycle_writer_fleet_proof_from(&state, "other-topology", 2, now).is_none());
+            let token = acquire_mrf_lifecycle_writer_fleet_proof_from(&state, "topology", 2, now)
+                .expect("compatible current observation");
+            assert!(mrf_lifecycle_writer_fleet_proof_matches_at(&state, &token, "topology", now));
+            assert!(!mrf_lifecycle_writer_fleet_proof_matches_at(
+                &state,
+                &token,
+                "topology",
+                now + REMOTE_VERSION_STATE_PROOF_TTL,
+            ));
+            token
+        };
+        assert!(publish_fleet_capability_probe_result(&slot, "topology", Err(Error::other("old reader")), now).is_some());
+        let state = slot.read().expect("revoked proof state");
+        assert!(!mrf_lifecycle_writer_fleet_proof_matches_at(&state, &token, "topology", now));
+        assert!(acquire_mrf_lifecycle_writer_fleet_proof_from(&state, "topology", 2, now).is_none());
+        assert!(state.draining_generation.is_some());
+        drop(state);
+        drop(token);
+        mark_fleet_capability_topology_conflict(&slot);
+        let state = slot.read().expect("conflicting topology state");
+        assert!(acquire_mrf_lifecycle_writer_fleet_proof_from(&state, "topology", 2, now).is_none());
+    }
+
+    #[test]
+    fn mrf_lifecycle_writer_refresh_cannot_extend_an_existing_publication() {
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let now = Instant::now();
+        let peers = BTreeMap::from([("local".to_owned(), Uuid::new_v4())]);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology", Ok(peers.clone()), now).is_none());
+        let old = acquire_mrf_lifecycle_writer_fleet_proof_from(&slot.read().expect("proof state"), "topology", 2, now)
+            .expect("first publication");
+        let refreshed_at = now + REMOTE_VERSION_STATE_PROOF_TTL;
+        assert!(publish_fleet_capability_probe_result(&slot, "topology", Ok(peers), refreshed_at).is_none());
+        let state = slot.read().expect("refreshed proof state");
+        let current = acquire_mrf_lifecycle_writer_fleet_proof_from(&state, "topology", 2, refreshed_at)
+            .expect("a new publication can use the fresh observation");
+        assert!(mrf_lifecycle_writer_fleet_proof_matches_at(&state, &current, "topology", refreshed_at));
+        assert!(!mrf_lifecycle_writer_fleet_proof_matches_at(&state, &old, "topology", refreshed_at));
+    }
+
+    #[test]
+    fn mrf_lifecycle_writer_restart_waits_for_the_previous_publication() {
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let now = Instant::now();
+        let peers = BTreeMap::from([("local".to_owned(), Uuid::new_v4())]);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology", Ok(peers), now).is_none());
+        let token = acquire_mrf_lifecycle_writer_fleet_proof_from(&slot.read().expect("proof state"), "topology", 2, now)
+            .expect("first publication permit");
+        let restarted = BTreeMap::from([("local".to_owned(), Uuid::new_v4())]);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology", Ok(restarted.clone()), now).is_some());
+        {
+            let state = slot.read().expect("draining proof state");
+            assert!(!mrf_lifecycle_writer_fleet_proof_matches_at(&state, &token, "topology", now));
+            assert!(acquire_mrf_lifecycle_writer_fleet_proof_from(&state, "topology", 2, now).is_none());
+        }
+        drop(token);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology", Ok(restarted.clone()), now).is_none());
+        let state = slot.read().expect("successor proof state");
+        let successor = acquire_mrf_lifecycle_writer_fleet_proof_from(&state, "topology", 2, now)
+            .expect("successor may be admitted after the old disk operations finish");
+        assert_eq!(*successor.token.peer_epochs, restarted);
+        assert!(mrf_lifecycle_writer_fleet_proof_matches_at(&state, &successor, "topology", now));
+    }
+
+    #[tokio::test]
+    async fn mrf_lifecycle_writer_probe_rejects_missing_or_unreachable_members() {
+        for peer_clients in [Vec::new(), vec![None]] {
+            let notification_sys = NotificationSys {
+                peer_clients,
+                all_peer_clients: vec![None],
+                peer_topology_hosts: vec!["peer-a".to_owned()],
+                peer_admin_caches: Vec::new(),
+                tier_config_reload_workers: Default::default(),
+            };
+            assert!(
+                notification_sys
+                    .probe_mrf_lifecycle_reader_fleet("topology", 2)
+                    .await
+                    .is_err()
+            );
+        }
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn mrf_lifecycle_writer_native_cas_retains_authority_after_timeout_or_cancellation() {
+        use crate::disk::os::prepared_publication_test_hooks as hooks;
+        use crate::disk::{RUSTFS_META_BUCKET, disk_store::LocalDiskWrapper, local::LocalDisk};
+        use crate::layout::endpoint::Endpoint;
+
+        temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("60"))], async {
+            let topology = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get_or_init(|| "mrf-cas-test".to_owned());
+            let slot = mrf_lifecycle_writer_fleet_proof_slot();
+            for cancel_waiter in [true, false] {
+                let root = tempfile::tempdir().expect("isolated CAS fixture");
+                let endpoint = Endpoint::try_from(root.path().to_str().expect("fixture path")).expect("local endpoint");
+                let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk"));
+                let volume = disk.make_volume(RUSTFS_META_BUCKET).await;
+                assert!(
+                    matches!(volume, Ok(()) | Err(crate::disk::error::Error::VolumeExists)),
+                    "metadata volume: {volume:?}"
+                );
+                let relative = "mrf-authority";
+                let path = disk
+                    .get_object_path_for_io(RUSTFS_META_BUCKET, relative)
+                    .expect("native CAS publication path");
+                std::fs::write(&path, b"old").expect("old committed file");
+                let peers = BTreeMap::from([("local".to_owned(), Uuid::new_v4())]);
+                assert!(publish_fleet_capability_probe_result(slot, topology, Ok(peers), Instant::now()).is_none());
+                let proof = Arc::new(
+                    acquire_mrf_lifecycle_writer_fleet_proof_from(
+                        &slot.read().expect("proof state"),
+                        topology,
+                        2,
+                        Instant::now(),
+                    )
+                    .expect("publication permit"),
+                );
+                let weak = Arc::downgrade(&proof);
+                let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
+                let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
+                let _hook = hooks::install_at(hooks::Stage::Rename, &path, move || {
+                    let _ = entered_tx.send(());
+                    let _ = release_rx.recv();
+                });
+                let wrapper = LocalDiskWrapper::new(disk, false);
+                let mut write = Box::pin(with_mrf_lifecycle_writer_fleet_proof(
+                    proof,
+                    wrapper.compare_and_update_file(
+                        RUSTFS_META_BUCKET,
+                        relative,
+                        Some(Bytes::from_static(b"old")),
+                        Some(Bytes::from_static(b"new")),
+                    ),
+                ));
+                timeout(Duration::from_secs(10), async {
+                    tokio::select! {
+                        result = entered_rx => result.expect("native CAS reached publication"),
+                        result = write.as_mut() => panic!("CAS returned before publication: {result:?}"),
+                    }
+                })
+                .await
+                .expect("native CAS boundary");
+                if cancel_waiter {
+                    drop(write);
+                } else {
+                    tokio::time::pause();
+                    tokio::time::advance(Duration::from_secs(61)).await;
+                    tokio::time::resume();
+                    let error = timeout(Duration::from_secs(5), write)
+                        .await
+                        .expect("disk deadline is bounded")
+                        .expect_err("blocked CAS must time out");
+                    assert_eq!(error, crate::disk::error::Error::Timeout);
+                }
+                assert!(weak.upgrade().is_some(), "the native worker still owns the permit");
+                let restarted = BTreeMap::from([("local".to_owned(), Uuid::new_v4())]);
+                assert!(publish_fleet_capability_probe_result(slot, topology, Ok(restarted.clone()), Instant::now()).is_some());
+                assert_eq!(std::fs::read(&path).expect("old file retained while blocked"), b"old");
+                drop(release_tx);
+                timeout(Duration::from_secs(5), async {
+                    while weak.upgrade().is_some() {
+                        tokio::time::sleep(Duration::from_millis(5)).await;
+                    }
+                })
+                .await
+                .expect("native worker drains");
+                assert_eq!(std::fs::read(&path).expect("revoked writer cannot publish"), b"old");
+                assert!(publish_fleet_capability_probe_result(slot, topology, Ok(restarted), Instant::now()).is_none());
+                revoke_fleet_capability_proof(slot);
+            }
+        })
+        .await;
+    }
+
     #[test]
     fn remote_version_state_fleet_proof_rejects_stale_or_mismatched_membership() {
         let now = Instant::now();
@@ -4193,12 +4593,12 @@ mod tests {
             (local.to_string(), Uuid::new_v4()),
             (expected_remote[0].clone(), Uuid::new_v4()),
         ]);
-        assert!(validate_ilm_recovery_export_members(&expected_remote, local, &complete).is_ok());
+        assert!(validate_capability_fleet_members(&expected_remote, local, &complete).is_ok());
 
         let unexpected = BTreeMap::from([(local.to_string(), Uuid::new_v4()), ("node-c".to_string(), Uuid::new_v4())]);
-        assert!(validate_ilm_recovery_export_members(&expected_remote, local, &unexpected).is_err());
+        assert!(validate_capability_fleet_members(&expected_remote, local, &unexpected).is_err());
         assert!(
-            validate_ilm_recovery_export_members(&[local.to_string()], local, &complete).is_err(),
+            validate_capability_fleet_members(&[local.to_string()], local, &complete).is_err(),
             "the configured remote set cannot repeat the local member"
         );
     }

crates/heal/src/heal/mrf_queue.rs

@@ -27,6 +27,11 @@
 
 use super::{DiskStore, HealDiskExt as _, local_disk_map_read};
 use crate::heal::manager::{HealManager, MrfRepairNoticeTarget};
+use crate::heal::storage_api::{
+    EcstoreMrfLifecycleWriterFleetProofToken, ecstore_acquire_mrf_lifecycle_writer_fleet_proof,
+    ecstore_mrf_lifecycle_writer_fleet_proof_current, ecstore_mrf_lifecycle_writer_fleet_proof_matches,
+    ecstore_with_mrf_lifecycle_writer_fleet_proof,
+};
 use futures::{StreamExt, stream};
 use metrics::{counter, gauge};
 use rustfs_common::mrf_channel::{
@@ -212,6 +217,10 @@ 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;
+
+/// Highest supported lifecycle format; every format from 1 through this value is readable.
+pub const MRF_LIFECYCLE_READER_MAX_FORMAT_VERSION: u8 = MRF_LIFECYCLE_FORMAT_VERSION_V2;
 
 /// Record format tag.
 const MRF_JOURNAL_FORMAT: u8 = 1;
@@ -252,6 +261,8 @@ pub(crate) struct MrfConsumerConfig {
     pub flush_threshold: usize,
     /// Backoff after the heal manager reports a full admission.
     pub admission_backoff: Duration,
+    /// Permit new V2 states only after an authenticated reader-fleet observation.
+    pub lifecycle_v2_writer_enabled: bool,
 }
 
 impl Default for MrfConsumerConfig {
@@ -272,6 +283,10 @@ impl Default for MrfConsumerConfig {
             flush_interval: Duration::from_millis(500),
             flush_threshold: 1000,
             admission_backoff: Duration::from_secs(5),
+            lifecycle_v2_writer_enabled: rustfs_utils::get_env_bool(
+                rustfs_config::ENV_MRF_LIFECYCLE_V2_ENABLE,
+                rustfs_config::DEFAULT_MRF_LIFECYCLE_V2_ENABLE,
+            ),
         }
     }
 }
@@ -632,15 +647,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 +678,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 +690,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 +702,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 +716,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 +796,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 +813,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 +978,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
@@ -1022,7 +1050,7 @@ impl MrfRuntime {
         intent: MrfIntent,
         source_bucket_incarnation_id: Option<Uuid>,
     ) -> Result<(), MrfDurableAdmissionError> {
-        if !self.replay_readable {
+        if !self.replay_readable || self.partial_writes.pending_lifecycle_transition.is_some() {
             return Err(MrfDurableAdmissionError::Persistence);
         }
         self.partial_writes.admit_with_source_incarnation(
@@ -1072,6 +1100,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);
         }
@@ -1100,6 +1131,9 @@ impl MrfRuntime {
         if !self.replay_readable {
             return Err(MrfLifecycleControlError::CheckpointUnreadable);
         }
+        if self.partial_writes.pending_lifecycle_transition.is_some() {
+            return Err(MrfLifecycleControlError::Persistence);
+        }
         if self.partial_writes.is_active_responsibility(responsibility_id) {
             return Ok(());
         }
@@ -1131,6 +1165,9 @@ impl MrfRuntime {
         if !self.replay_readable {
             return Err(MrfLifecycleControlError::CheckpointUnreadable);
         }
+        if self.partial_writes.pending_lifecycle_transition.is_some() {
+            return Err(MrfLifecycleControlError::Persistence);
+        }
         let (intent, state, source_incarnation) = self
             .partial_writes
             .intent_for_responsibility(responsibility_id)
@@ -1139,7 +1176,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);
         }
@@ -1180,6 +1222,9 @@ impl MrfRuntime {
         if !self.replay_readable {
             return Err(MrfLifecycleControlError::CheckpointUnreadable);
         }
+        if self.partial_writes.pending_lifecycle_transition.is_some() {
+            return Err(MrfLifecycleControlError::Persistence);
+        }
         let responsibility_id = request.responsibility_id;
         if self
             .partial_writes
@@ -1216,6 +1261,39 @@ impl MrfRuntime {
         Ok(())
     }
 
+    async fn handle_lifecycle_command(&mut self, manager: &HealManager, command: MrfLifecycleCommand) {
+        match command {
+            MrfLifecycleCommand::List { after, limit, response } => {
+                let _ = response.send(self.list_unverified_legacy(after, limit));
+            }
+            MrfLifecycleCommand::Recheck {
+                responsibility_id,
+                expected_bucket_incarnation_id,
+                response,
+            } => {
+                let result = self
+                    .recheck_legacy_generation(manager, responsibility_id, expected_bucket_incarnation_id)
+                    .await;
+                let _ = response.send(result);
+            }
+            MrfLifecycleCommand::Refresh {
+                responsibility_id,
+                expected_bucket_incarnation_id,
+                response,
+            } => {
+                let result = self
+                    .refresh_legacy_generation(manager, responsibility_id, expected_bucket_incarnation_id)
+                    .await;
+                let _ = response.send(result);
+            }
+            MrfLifecycleCommand::AcceptUnverifiedRisk { request, response } => {
+                let result = self.accept_legacy_risk(manager, request).await;
+                let _ = response.send(result);
+            }
+        }
+        self.publish_metrics();
+    }
+
     fn snapshot(&self) -> (Vec<u8>, Vec<u8>) {
         let mut authoritative = Vec::new();
         let mut legacy = Vec::new();
@@ -1238,6 +1316,9 @@ impl MrfRuntime {
         if !self.replay_readable {
             return false;
         }
+        if self.partial_writes.pending_lifecycle_transition.is_some() {
+            return self.flush_pending_lifecycle(None).await;
+        }
         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 {
@@ -1247,6 +1328,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,
@@ -1289,7 +1371,179 @@ impl MrfRuntime {
                 }
             }
         };
-        let authoritative_persisted = write_journal(MRF_SCOPED_JOURNAL_PATH, &authoritative).await;
+        self.finish_flush(
+            &authoritative,
+            &legacy,
+            committed_persisted,
+            committed_on_disk,
+            lifecycle_checkpoint.is_some(),
+        )
+        .await
+    }
+
+    async fn pending_lifecycle_commit_matches(
+        &self,
+        authoritative: &[u8],
+        lifecycle: &[u8],
+    ) -> Result<bool, snapshot::SnapshotError> {
+        let Some(checkpoint) = snapshot::inspect_local_committed_snapshot(self.config.journal_max_bytes).await? else {
+            return Ok(false);
+        };
+        if checkpoint.owner() != self.checkpoint_owner
+            || checkpoint.sequence() != self.next_checkpoint_sequence
+            || checkpoint.payload() != authoritative
+        {
+            return Ok(false);
+        }
+        let companion = snapshot::read_committed_companion(
+            &journal_disks().await,
+            self.checkpoint_owner,
+            self.next_checkpoint_sequence,
+            &MRF_LIFECYCLE_PATHS,
+            self.config.journal_max_bytes,
+            self.config.journal_max_bytes.saturating_mul(4),
+        )
+        .await?;
+        Ok(companion.as_deref() == Some(lifecycle))
+    }
+
+    async fn flush_pending_lifecycle(&mut self, proof: Option<EcstoreMrfLifecycleWriterFleetProofToken>) -> bool {
+        if !self.replay_readable {
+            return false;
+        }
+        let Some(pending) = self.partial_writes.pending_lifecycle_transition.as_mut() else {
+            return false;
+        };
+        let now = tokio::time::Instant::now();
+        if now < pending.next_attempt {
+            return false;
+        }
+        pending.next_attempt = now
+            + self
+                .config
+                .admission_backoff
+                .clamp(Duration::from_secs(5), Duration::from_secs(60));
+        let intent = pending.intent.clone();
+        let record = pending.record.clone();
+        let mut records = self.partial_writes.checkpoint_records(intent_digest);
+        let Some(current) = records.iter_mut().find(|current| {
+            current.responsibility_id == record.responsibility_id && current.intent_digest == record.intent_digest
+        }) else {
+            return false;
+        };
+        if !matches!(current.state, partial_write::ResponsibilityState::Active {})
+            || current.source_bucket_incarnation_id.is_some()
+        {
+            return false;
+        }
+        *current = record.clone();
+        let lifecycle_limit = self.config.journal_max_bytes.saturating_mul(4);
+        let Some(lifecycle) = encode_mrf_lifecycle_checkpoint(
+            MRF_LIFECYCLE_FORMAT_VERSION_V2,
+            self.checkpoint_owner,
+            self.next_checkpoint_sequence,
+            records,
+            lifecycle_limit,
+        ) else {
+            return false;
+        };
+        let (authoritative, legacy) = self.snapshot();
+        // A lost publication result must not regenerate the body or downgrade
+        // the owner. An exact committed readback resolves it without new writes.
+        let already_committed = match self.pending_lifecycle_commit_matches(&authoritative, &lifecycle).await {
+            Ok(committed) => committed,
+            Err(_) => return false,
+        };
+        let (authoritative, lifecycle) = if already_committed {
+            (authoritative, lifecycle)
+        } else {
+            let proof = match proof {
+                Some(proof) => proof,
+                None => {
+                    let Some(proof) =
+                        ecstore_acquire_mrf_lifecycle_writer_fleet_proof(MRF_LIFECYCLE_READER_MAX_FORMAT_VERSION).await
+                    else {
+                        counter!("rustfs_heal_mrf_lifecycle_writer_blocked_total", "reason" => "reader_fleet").increment(1);
+                        return false;
+                    };
+                    proof
+                }
+            };
+            let disks = journal_disks().await;
+            let owner = self.checkpoint_owner;
+            let sequence = self.next_checkpoint_sequence;
+            let limit = self.config.journal_max_bytes;
+            // The task and native CAS workers share one permit. A dropped
+            // waiter or disk timeout cannot release a still-running worker.
+            let published = tokio::spawn(async move {
+                let proof = Arc::new(proof);
+                // All replicas share one fresh pre-commit observation; native
+                // workers still check expiry and revocation independently.
+                let observed = tokio::sync::OnceCell::new();
+                let result = ecstore_with_mrf_lifecycle_writer_fleet_proof(
+                    Arc::clone(&proof),
+                    snapshot::publish_committed_snapshot_with_companion_guarded(
+                        &disks,
+                        owner,
+                        sequence,
+                        &authoritative,
+                        limit,
+                        Some((&MRF_LIFECYCLE_PATHS, &lifecycle, lifecycle_limit)),
+                        |stage| {
+                            let proof = &proof;
+                            let observed = &observed;
+                            async move {
+                                if matches!(stage, snapshot::SnapshotPublicationStage::Manifest)
+                                    && !*observed
+                                        .get_or_init(|| ecstore_mrf_lifecycle_writer_fleet_proof_matches(proof))
+                                        .await
+                                {
+                                    return false;
+                                }
+                                ecstore_mrf_lifecycle_writer_fleet_proof_current(proof)
+                            }
+                        },
+                    ),
+                )
+                .await;
+                (result, authoritative, lifecycle)
+            })
+            .await;
+            let Ok((result, authoritative, lifecycle)) = published else {
+                return false;
+            };
+            if result.is_err()
+                && !self
+                    .pending_lifecycle_commit_matches(&authoritative, &lifecycle)
+                    .await
+                    .unwrap_or(false)
+            {
+                return false;
+            }
+            (authoritative, lifecycle)
+        };
+        if !self.partial_writes.restore_state(&intent, &record) {
+            return false;
+        }
+        self.partial_writes.pending_lifecycle_transition = None;
+        self.lifecycle_format_version = MRF_LIFECYCLE_FORMAT_VERSION_V2;
+        self.runtime_checkpoint = Some((self.checkpoint_owner, self.next_checkpoint_sequence));
+        self.next_checkpoint_sequence = self.next_checkpoint_sequence.saturating_add(1);
+        self.journal_on_disk = true;
+        self.dirty = true;
+        self.finish_flush(&authoritative, &legacy, true, true, !lifecycle.is_empty())
+            .await
+    }
+
+    async fn finish_flush(
+        &mut self,
+        authoritative: &[u8],
+        legacy: &[u8],
+        committed_persisted: bool,
+        committed_on_disk: bool,
+        lifecycle_ready: bool,
+    ) -> bool {
+        let authoritative_persisted = write_journal(MRF_SCOPED_JOURNAL_PATH, authoritative).await;
         if !authoritative.is_empty() {
             counter!("rustfs_heal_mrf_journal_fsync_total").increment(1);
         }
@@ -1298,7 +1552,7 @@ impl MrfRuntime {
         // snapshot has reached at least one disk. This ordering prevents an
         // old reader from observing a newer epoch that a new reader cannot
         // see when the canonical write is unavailable.
-        let legacy_persisted = authoritative_persisted && write_journal(MRF_JOURNAL_PATH, &legacy).await;
+        let legacy_persisted = authoritative_persisted && write_journal(MRF_JOURNAL_PATH, legacy).await;
         let lifecycle_persisted = if self.partial_writes.depth() == 0 {
             let mut all_deleted = true;
             for path in MRF_LIFECYCLE_PATHS {
@@ -1306,7 +1560,7 @@ impl MrfRuntime {
             }
             all_deleted
         } else {
-            lifecycle_checkpoint.is_some() && committed_persisted
+            lifecycle_ready && committed_persisted
         };
         // Keep dirty until the committed checkpoint, authoritative snapshot,
         // and compatibility mirror have all been accepted; otherwise a
@@ -1327,21 +1581,38 @@ impl MrfRuntime {
     /// Drain pending intents into the heal manager until it is full, the
     /// queue empties, or attempts are exhausted.
     async fn dispatch_partial_writes(&mut self, manager: &HealManager) {
-        if self
+        if self.partial_writes.pending_lifecycle_transition.is_some() {
+            return;
+        }
+        let changed = self
             .partial_writes
             .dispatch_collecting_lifecycle_change(manager, &self.config)
-            .await
+            .await;
+        if self.config.lifecycle_v2_writer_enabled
+            && let Some(candidate) = self.partial_writes.take_unresolved_generation_candidate()
         {
+            if let Some(proof) = ecstore_acquire_mrf_lifecycle_writer_fleet_proof(MRF_LIFECYCLE_READER_MAX_FORMAT_VERSION).await {
+                self.partial_writes.pending_lifecycle_transition = Some(candidate);
+                self.dirty = true;
+                self.flush_pending_lifecycle(Some(proof)).await;
+                return;
+            }
+            counter!("rustfs_heal_mrf_lifecycle_writer_blocked_total", "reason" => "reader_fleet").increment(1);
+        }
+        if changed {
             self.dirty = true;
             self.flush().await;
         }
     }
 
     async fn dispatch(&mut self, manager: &HealManager) {
-        if !self.replay_readable {
+        if !self.replay_readable || self.partial_writes.pending_lifecycle_transition.is_some() {
             return;
         }
         self.dispatch_partial_writes(manager).await;
+        if self.partial_writes.pending_lifecycle_transition.is_some() {
+            return;
+        }
         if let Some(until) = self.backoff_until {
             if tokio::time::Instant::now() < until {
                 return;
@@ -1357,6 +1628,9 @@ impl MrfRuntime {
                 if self.flush().await {
                     self.dispatch_partial_writes(manager).await;
                 }
+                if self.partial_writes.pending_lifecycle_transition.is_some() {
+                    return;
+                }
                 continue;
             }
             // Leaving the pending set (consumed or re-queued with a bumped
@@ -1576,6 +1850,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 +1876,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 +1899,7 @@ struct ReplaySource {
     truncated_bytes: usize,
     cleanup: ReplayCleanup,
     lifecycle: HashMap<[u8; 32], VecDeque<ResponsibilityCheckpoint>>,
+    lifecycle_format_version: u8,
 }
 
 fn take_lifecycle_record(
@@ -1642,8 +1919,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 +1928,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 +1938,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 +1951,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 +1970,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 +2015,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 +2033,7 @@ async fn replay_into(
                 partial_writes: Vec::new(),
                 cleanup: None,
                 next_checkpoint_sequence: 1,
+                lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
             };
         }
     };
@@ -1896,6 +2180,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 +2199,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 +2220,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
@@ -1955,6 +2242,29 @@ async fn run_mrf_consumer(
     let mut durable_batch = Vec::with_capacity(runtime.config.replay_batch);
 
     loop {
+        if runtime.partial_writes.pending_lifecycle_transition.is_some() {
+            tokio::select! {
+                received = durable_receiver.recv_many(&mut durable_batch, runtime.config.replay_batch) => {
+                    if received == 0 {
+                        return;
+                    }
+                    for submission in durable_batch.drain(..) {
+                        let _ = submission.response.send(Err(MrfDurableAdmissionError::Persistence));
+                    }
+                }
+                command = lifecycle_receiver.recv() => {
+                    let Some(command) = command else {
+                        return;
+                    };
+                    runtime.handle_lifecycle_command(manager.as_ref(), command).await;
+                }
+                _ = flush_tick.tick() => {
+                    runtime.flush().await;
+                    runtime.publish_metrics();
+                }
+            }
+            continue;
+        }
         tokio::select! {
             received = durable_receiver.recv_many(&mut durable_batch, runtime.config.replay_batch) => {
                 if received == 0 {
@@ -1979,44 +2289,7 @@ async fn run_mrf_consumer(
                 let Some(command) = command else {
                     return;
                 };
-                match command {
-                    MrfLifecycleCommand::List { after, limit, response } => {
-                        let _ = response.send(runtime.list_unverified_legacy(after, limit));
-                    }
-                    MrfLifecycleCommand::Recheck {
-                        responsibility_id,
-                        expected_bucket_incarnation_id,
-                        response,
-                    } => {
-                        let result = runtime
-                            .recheck_legacy_generation(
-                                manager.as_ref(),
-                                responsibility_id,
-                                expected_bucket_incarnation_id,
-                            )
-                            .await;
-                        let _ = response.send(result);
-                    }
-                    MrfLifecycleCommand::Refresh {
-                        responsibility_id,
-                        expected_bucket_incarnation_id,
-                        response,
-                    } => {
-                        let result = runtime
-                            .refresh_legacy_generation(
-                                manager.as_ref(),
-                                responsibility_id,
-                                expected_bucket_incarnation_id,
-                            )
-                            .await;
-                        let _ = response.send(result);
-                    }
-                    MrfLifecycleCommand::AcceptUnverifiedRisk { request, response } => {
-                        let result = runtime.accept_legacy_risk(manager.as_ref(), request).await;
-                        let _ = response.send(result);
-                    }
-                }
-                runtime.publish_metrics();
+                runtime.handle_lifecycle_command(manager.as_ref(), command).await;
             }
             received = receiver.recv_many(&mut batch, runtime.config.replay_batch) => {
                 if received == 0 {
@@ -2148,6 +2421,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 +2581,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 +2592,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 +2716,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 +2857,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 +3451,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 +3504,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 +3554,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 +3667,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 +3731,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 +3980,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,

crates/heal/src/heal/mrf_queue/partial_write.rs

@@ -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),
@@ -210,11 +217,16 @@ struct Responsibility {
     consecutive_unverified_attempts: u32,
     last_identity_probe_result: Option<&'static str>,
     last_identity_probe_at_ms: Option<u64>,
+    first_unresolved_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 {} | ResponsibilityState::SourceAndCurrentGenerationUnknown { .. }
+            )
     }
 
     fn is_parked(&self) -> bool {
@@ -251,6 +263,12 @@ pub(super) struct PartialWriteLifecycleSummary {
     pub oldest_operator_accepted_at_ms: Option<u64>,
 }
 
+pub(super) struct PendingLifecycleTransition {
+    pub intent: MrfIntent,
+    pub record: ResponsibilityCheckpoint,
+    pub next_attempt: Instant,
+}
+
 /// Durable storage responsibilities have no guaranteed rediscovery producer.
 /// Admission, task failure and retry exhaustion cannot release their records.
 #[derive(Default)]
@@ -261,6 +279,8 @@ pub(super) struct PartialWrites {
     identity_probe_order: VecDeque<PartialWriteKey>,
     identity_probe_index: HashSet<PartialWriteKey>,
     retirement_checks: HashMap<(Arc<str>, Uuid), (Instant, bool)>,
+    unresolved_generation_candidate: Option<PartialWriteKey>,
+    pub(super) pending_lifecycle_transition: Option<PendingLifecycleTransition>,
     bytes: usize,
 }
 
@@ -356,6 +376,7 @@ impl PartialWrites {
                 consecutive_unverified_attempts: 0,
                 last_identity_probe_result: None,
                 last_identity_probe_at_ms: None,
+                first_unresolved_at_ms: None,
             },
         );
         self.bytes = next_bytes;
@@ -479,7 +500,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 +547,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 +649,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 +779,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 +890,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;
@@ -861,6 +908,36 @@ impl PartialWrites {
         Ok((previous, previous_acceptance))
     }
 
+    pub(super) fn take_unresolved_generation_candidate(&mut self) -> Option<PendingLifecycleTransition> {
+        let key = self.unresolved_generation_candidate.take()?;
+        let entry = self.entries.get(&key)?;
+        if !entry.persisted
+            || entry.source_bucket_incarnation_id.is_some()
+            || !matches!(entry.state, ResponsibilityState::Active {})
+        {
+            return None;
+        }
+        let since_ms = entry.first_unresolved_at_ms?;
+        let next_probe_at_ms = unix_now_ms().saturating_add(
+            u64::try_from(entry.next_attempt.saturating_duration_since(Instant::now()).as_millis()).unwrap_or(u64::MAX),
+        );
+        Some(PendingLifecycleTransition {
+            intent: entry.intent.clone(),
+            record: ResponsibilityCheckpoint {
+                intent_digest: super::intent_digest(&entry.intent)?,
+                responsibility_id: entry.responsibility_id,
+                source_bucket_incarnation_id: None,
+                last_operator_acceptance: entry.last_operator_acceptance.clone(),
+                state: ResponsibilityState::SourceAndCurrentGenerationUnknown {
+                    since_ms,
+                    next_probe_at_ms,
+                    probe_attempts: entry.consecutive_unverified_attempts,
+                },
+            },
+            next_attempt: Instant::now(),
+        })
+    }
+
     pub(super) fn mark_persisted(&mut self) {
         for entry in self.entries.values_mut() {
             entry.persisted = true;
@@ -934,6 +1011,7 @@ impl PartialWrites {
     }
 
     async fn probe_unknown_generation(&mut self, manager: &HealManager, config: &MrfConsumerConfig) -> bool {
+        self.unresolved_generation_candidate = None;
         if config.replay_batch == 0 {
             return false;
         }
@@ -952,34 +1030,60 @@ impl PartialWrites {
                 continue;
             }
             self.identity_probe_index.insert(key.clone());
-            self.identity_probe_order.push_back(key);
+            self.identity_probe_order.push_back(key.clone());
             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));
+            // 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()),
             )
             .await;
-            entry.last_identity_probe_at_ms = Some(unix_now_ms());
+            let observed_at_ms = unix_now_ms();
+            entry.last_identity_probe_at_ms = Some(observed_at_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(),
+                        detected_at_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(Ok(None)) => {
+                    entry.last_identity_probe_result = Some("unresolved");
+                    if config.lifecycle_v2_writer_enabled && matches!(entry.state, ResponsibilityState::Active {}) {
+                        entry.first_unresolved_at_ms.get_or_insert(observed_at_ms);
+                        self.unresolved_generation_candidate = Some(key);
+                    }
+                }
                 Ok(Err(_)) | Err(_) => entry.last_identity_probe_result = Some("unavailable"),
             }
-            return false;
+            return lifecycle_changed;
         }
         false
     }
@@ -2127,8 +2231,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 +2262,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 +2399,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 +2426,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 +2537,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 +2615,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,

crates/heal/src/heal/mrf_queue/partial_write/tests/unknown_source_tests.rs

@@ -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,643 @@ 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"
+    );
+}
+
+fn with_writer_fixture<F, Fut>(test: F)
+where
+    F: FnOnce() -> Fut + Send + 'static,
+    Fut: std::future::Future<Output = ()>,
+{
+    temp_env::with_var("RUSTFS_MRF_LIFECYCLE_V2_ENABLE", Some("true"), || {
+        const STACK_SIZE: usize = 8 * 1024 * 1024;
+        std::thread::Builder::new()
+            .name("mrf-lifecycle-writer".to_owned())
+            .stack_size(STACK_SIZE)
+            .spawn(move || {
+                tokio::runtime::Builder::new_multi_thread()
+                    .worker_threads(2)
+                    .thread_stack_size(STACK_SIZE)
+                    .enable_all()
+                    .build()
+                    .expect("writer fixture runtime")
+                    .block_on(async {
+                        let env = rustfs_test_utils::TestECStoreEnv::builder()
+                            .prefix("rustfs_mrf_lifecycle_writer")
+                            .build()
+                            .await;
+                        rustfs_common::set_global_local_node_name("mrf-writer-test:9000").await;
+                        crate::heal::storage_api::ecstore_new_global_notification_sys(env.endpoint_pools.clone())
+                            .await
+                            .expect("publish the single-node notification system");
+                        crate::heal::storage_api::ecstore_start_remote_version_state_fleet_probe(
+                            "mrf-writer-test-topology".to_owned(),
+                        );
+                        test().await;
+                    });
+            })
+            .expect("writer fixture thread")
+            .join()
+            .expect("writer fixture must finish");
+    });
+}
+
+#[test]
+#[serial]
+fn unknown_source_deleted_bucket_writer_commits_only_the_retained_identity() {
+    with_writer_fixture(|| async {
+        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 mut original = intent("writer-unresolved");
+        original.scope = Some(rustfs_common::mrf_channel::MrfScope {
+            pool_index: 0,
+            set_index: 0,
+        });
+        let key = PartialWriteKey::new(&original, None);
+        runtime
+            .admit_partial_write(original.clone(), None)
+            .expect("retain the legacy responsibility");
+        assert!(runtime.flush().await, "commit the old-format responsibility first");
+        let before = runtime.partial_writes.checkpoint_records(super::super::super::intent_digest);
+        assert_eq!(before.len(), 1);
+        assert!(matches!(before[0].state, ResponsibilityState::Active {}));
+        runtime
+            .partial_writes
+            .entries
+            .get_mut(&key)
+            .expect("retained entry")
+            .next_attempt = Instant::now();
+        runtime.dispatch_partial_writes(&manager).await;
+        let checkpoint = super::super::super::read_mrf_lifecycle_checkpoint(
+            runtime.checkpoint_owner,
+            runtime.runtime_checkpoint.expect("committed owner").1,
+            runtime.config.journal_max_bytes,
+            runtime.config.journal_max_bytes * 4,
+        )
+        .await
+        .expect("read the committed lifecycle")
+        .expect("lifecycle companion");
+        assert_eq!(checkpoint.format_version, 2, "an authorized unresolved observation must persist V2");
+        assert_eq!(checkpoint.records.len(), 1);
+        let retained = &checkpoint.records[0];
+        assert_eq!(retained.responsibility_id, before[0].responsibility_id);
+        assert_eq!(retained.intent_digest, before[0].intent_digest);
+        assert_eq!(retained.source_bucket_incarnation_id, None);
+        assert!(matches!(retained.state, ResponsibilityState::SourceAndCurrentGenerationUnknown { .. }));
+        assert_eq!(runtime.partial_writes.depth(), 1);
+        assert_eq!(runtime.partial_writes.intents().next().expect("retained intent").scope, original.scope);
+        assert_eq!(storage.calls.load(Ordering::Relaxed), 1);
+        let (listed, _) = runtime.partial_writes.list_unverified_legacy(None, 1);
+        assert_eq!(listed.len(), 1);
+        assert_eq!(listed[0].responsibility_id, before[0].responsibility_id);
+        assert_eq!(listed[0].bucket_incarnation_id, None);
+        assert_eq!(listed[0].source_bucket_incarnation_id, None);
+        assert!(listed[0].held_since_ms.is_some());
+    });
+}
+
+#[tokio::test(start_paused = true)]
+async fn unknown_source_deleted_bucket_writer_does_not_promote_errors_or_disabled_observations() {
+    for (enabled, lookup) in [
+        (false, Lookup::Identity(None)),
+        (true, Lookup::PermissionDenied),
+        (true, Lookup::Quorum),
+        (true, Lookup::Transport),
+        (true, Lookup::Identity(Some(Uuid::nil()))),
+        (true, Lookup::Pending),
+    ] {
+        let storage = Arc::new(ProbeStorage {
+            lookup: Mutex::new(lookup),
+            calls: AtomicUsize::new(0),
+        });
+        let manager = HealManager::new_without_root_recovery_for_test(storage.clone(), None);
+        let mut runtime = runtime();
+        runtime.config.lifecycle_v2_writer_enabled = enabled;
+        let original = intent("writer-ineligible");
+        let key = PartialWriteKey::new(&original, None);
+        runtime
+            .admit_partial_write(original, None)
+            .expect("retain legacy responsibility");
+        runtime.partial_writes.mark_persisted();
+        runtime
+            .partial_writes
+            .entries
+            .get_mut(&key)
+            .expect("legacy entry")
+            .next_attempt = Instant::now();
+        let before = runtime.partial_writes.checkpoint_records(super::super::super::intent_digest);
+        assert!(
+            !runtime
+                .partial_writes
+                .dispatch_collecting_lifecycle_change(&manager, &runtime.config)
+                .await
+        );
+        assert!(runtime.partial_writes.take_unresolved_generation_candidate().is_none());
+        assert_eq!(runtime.partial_writes.checkpoint_records(super::super::super::intent_digest), before);
+        assert!(runtime.partial_writes.entries[&key].next_attempt > Instant::now());
+        assert_eq!(storage.calls.load(Ordering::Relaxed), 1);
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn unknown_source_deleted_bucket_writer_without_fleet_proof_keeps_v1_visible_and_backed_off() {
+    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();
+    runtime.config.lifecycle_v2_writer_enabled = true;
+    let original = intent("writer-without-fleet");
+    let key = PartialWriteKey::new(&original, None);
+    runtime
+        .admit_partial_write(original, None)
+        .expect("retain legacy responsibility");
+    runtime.partial_writes.mark_persisted();
+    runtime
+        .partial_writes
+        .entries
+        .get_mut(&key)
+        .expect("legacy entry")
+        .next_attempt = Instant::now();
+    let before = runtime.partial_writes.checkpoint_records(super::super::super::intent_digest);
+    for _ in 0..20 {
+        runtime.dispatch_partial_writes(&manager).await;
+    }
+    assert_eq!(storage.calls.load(Ordering::Relaxed), 1);
+    assert_eq!(runtime.lifecycle_format_version, 1);
+    assert!(runtime.partial_writes.pending_lifecycle_transition.is_none());
+    assert_eq!(runtime.partial_writes.checkpoint_records(super::super::super::intent_digest), before);
+    let page = runtime
+        .list_unverified_legacy(None, 1)
+        .expect("legacy responsibility remains visible");
+    assert_eq!(page.responsibilities.len(), 1);
+    assert_eq!(page.responsibilities[0].responsibility_id, before[0].responsibility_id);
+    assert_eq!(page.responsibilities[0].bucket_incarnation_id, None);
+
+    // Earlier unresolved observations remain the retention origin even when
+    // fleet unavailability forces another identity lookup before promotion.
+    runtime
+        .partial_writes
+        .entries
+        .get_mut(&key)
+        .expect("legacy entry")
+        .first_unresolved_at_ms = Some(7);
+    tokio::time::advance(Duration::from_secs(61)).await;
+    runtime.dispatch_partial_writes(&manager).await;
+    assert_eq!(runtime.partial_writes.entries[&key].first_unresolved_at_ms, Some(7));
+    tokio::time::advance(Duration::from_secs(61)).await;
+    runtime
+        .partial_writes
+        .dispatch_collecting_lifecycle_change(&manager, &runtime.config)
+        .await;
+    let candidate = runtime
+        .partial_writes
+        .take_unresolved_generation_candidate()
+        .expect("a later unresolved observation");
+    assert!(matches!(
+        candidate.record.state,
+        ResponsibilityState::SourceAndCurrentGenerationUnknown { since_ms: 7, .. }
+    ));
+    assert_eq!(storage.calls.load(Ordering::Relaxed), 3);
+}
+
+#[test]
+#[serial]
+fn unknown_source_deleted_bucket_writer_failed_companion_keeps_owner_until_bounded_retry() {
+    with_writer_fixture(|| async {
+        use crate::heal::mrf_queue::{MRF_LIFECYCLE_PATHS, journal_disks, read_mrf_lifecycle_checkpoint};
+        use crate::heal::storage_api::EcstoreDiskAPI;
+
+        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("writer-failed-companion");
+        let key = PartialWriteKey::new(&original, None);
+        runtime
+            .admit_partial_write(original, None)
+            .expect("retain legacy responsibility");
+        assert!(runtime.flush().await);
+        let owner = runtime.runtime_checkpoint.expect("old checkpoint");
+        let before = runtime.partial_writes.checkpoint_records(super::super::super::intent_digest);
+        let mut obstacles = Vec::new();
+        for disk in journal_disks().await {
+            let path = EcstoreDiskAPI::path(disk.as_ref())
+                .join(crate::heal::RUSTFS_META_BUCKET)
+                .join(MRF_LIFECYCLE_PATHS[1]);
+            std::fs::create_dir(&path).expect("inject an unwritable successor companion");
+            obstacles.push(path);
+        }
+        runtime
+            .partial_writes
+            .entries
+            .get_mut(&key)
+            .expect("legacy entry")
+            .next_attempt = Instant::now();
+        runtime.dispatch_partial_writes(&manager).await;
+        assert!(runtime.partial_writes.pending_lifecycle_transition.is_some());
+        assert_eq!(runtime.runtime_checkpoint, Some(owner));
+        assert_eq!(runtime.lifecycle_format_version, 1);
+        assert_eq!(runtime.partial_writes.checkpoint_records(super::super::super::intent_digest), before);
+        let old = read_mrf_lifecycle_checkpoint(
+            owner.0,
+            owner.1,
+            runtime.config.journal_max_bytes,
+            runtime.config.journal_max_bytes * 4,
+        )
+        .await
+        .expect("read predecessor")
+        .expect("predecessor companion");
+        assert_eq!(old.format_version, 1);
+        assert_eq!(old.records, before);
+        assert_eq!(
+            runtime
+                .list_unverified_legacy(None, 1)
+                .expect("pending owner remains queryable")
+                .responsibilities[0]
+                .responsibility_id,
+            before[0].responsibility_id
+        );
+        assert!(matches!(
+            runtime.admit_partial_write(intent("later"), None),
+            Err(MrfDurableAdmissionError::Persistence)
+        ));
+        assert!(matches!(
+            runtime
+                .recheck_legacy_generation(&manager, before[0].responsibility_id, Uuid::new_v4())
+                .await,
+            Err(MrfLifecycleControlError::Persistence)
+        ));
+        runtime.dispatch_partial_writes(&manager).await;
+        assert_eq!(storage.calls.load(Ordering::Relaxed), 1);
+        assert!(!runtime.flush().await, "a consumer tick cannot bypass the retry deadline");
+        for path in obstacles {
+            std::fs::remove_dir(path).expect("remove only the empty injected obstacle");
+        }
+        runtime
+            .partial_writes
+            .pending_lifecycle_transition
+            .as_mut()
+            .expect("same pending transition")
+            .next_attempt = Instant::now();
+        assert!(runtime.flush().await, "the same owner retries after storage recovery");
+        assert!(runtime.partial_writes.pending_lifecycle_transition.is_none());
+        assert_eq!(runtime.runtime_checkpoint, Some((owner.0, owner.1 + 1)));
+        let committed = read_mrf_lifecycle_checkpoint(
+            owner.0,
+            owner.1 + 1,
+            runtime.config.journal_max_bytes,
+            runtime.config.journal_max_bytes * 4,
+        )
+        .await
+        .expect("read retry commit")
+        .expect("retry companion");
+        assert_eq!(committed.format_version, 2);
+        assert_eq!(committed.records[0].responsibility_id, before[0].responsibility_id);
+        assert_eq!(committed.records[0].source_bucket_incarnation_id, None);
+    });
+}
+
+#[test]
+#[serial]
+fn unknown_source_deleted_bucket_writer_lost_result_is_resolved_by_exact_commit_readback() {
+    with_writer_fixture(|| async {
+        use crate::heal::mrf_queue::{
+            MRF_LIFECYCLE_PATHS, encode_mrf_lifecycle_checkpoint, intent_digest, journal_disks, snapshot,
+        };
+
+        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, None);
+        let mut runtime = runtime();
+        let original = intent("writer-lost-result");
+        let key = PartialWriteKey::new(&original, None);
+        runtime
+            .admit_partial_write(original, None)
+            .expect("retain legacy responsibility");
+        for object in ["writer-lost-result-other-a", "writer-lost-result-other-b"] {
+            runtime
+                .admit_partial_write(intent(object), None)
+                .expect("retain another responsibility");
+        }
+        assert!(runtime.flush().await);
+        runtime
+            .partial_writes
+            .entries
+            .get_mut(&key)
+            .expect("legacy entry")
+            .next_attempt = Instant::now();
+        runtime
+            .partial_writes
+            .dispatch_collecting_lifecycle_change(&manager, &runtime.config)
+            .await;
+        let pending = runtime
+            .partial_writes
+            .take_unresolved_generation_candidate()
+            .expect("unresolved proposal");
+        let record = pending.record.clone();
+        runtime.partial_writes.pending_lifecycle_transition = Some(pending);
+        let sequence = runtime.next_checkpoint_sequence;
+        let (payload, _) = runtime.snapshot();
+        let lifecycle = encode_mrf_lifecycle_checkpoint(
+            2,
+            runtime.checkpoint_owner,
+            sequence,
+            {
+                let mut records = runtime.partial_writes.checkpoint_records(intent_digest);
+                *records
+                    .iter_mut()
+                    .find(|item| item.responsibility_id == record.responsibility_id)
+                    .expect("pending responsibility") = record.clone();
+                records
+            },
+            runtime.config.journal_max_bytes * 4,
+        )
+        .expect("exact candidate body");
+        // Commit the proposal without delivering the result to its runtime.
+        snapshot::publish_committed_snapshot_with_companion(
+            &journal_disks().await,
+            runtime.checkpoint_owner,
+            sequence,
+            &payload,
+            runtime.config.journal_max_bytes,
+            Some((&MRF_LIFECYCLE_PATHS, &lifecycle, runtime.config.journal_max_bytes * 4)),
+        )
+        .await
+        .expect("commit before the simulated lost acknowledgement");
+        crate::heal::storage_api::ecstore_start_remote_version_state_fleet_probe("changed-test-topology".to_owned());
+        assert!(
+            crate::heal::storage_api::ecstore_acquire_mrf_lifecycle_writer_fleet_proof(2)
+                .await
+                .is_none()
+        );
+        assert!(runtime.flush().await, "an exact committed result needs no new writer authority");
+        assert_eq!(runtime.lifecycle_format_version, 2);
+        assert_eq!(runtime.next_checkpoint_sequence, sequence + 1);
+        let restored = runtime.partial_writes.checkpoint_records(intent_digest);
+        assert_eq!(restored.len(), 3);
+        assert_eq!(
+            restored
+                .iter()
+                .find(|item| item.responsibility_id == record.responsibility_id),
+            Some(&record)
+        );
+        assert_eq!(
+            restored
+                .iter()
+                .filter(|item| matches!(item.state, ResponsibilityState::Active {}))
+                .count(),
+            2
+        );
+        assert!(runtime.partial_writes.pending_lifecycle_transition.is_none());
+        let committed = snapshot::inspect_local_committed_snapshot(runtime.config.journal_max_bytes)
+            .await
+            .expect("inspect readback")
+            .expect("committed snapshot");
+        assert_eq!(committed.sequence(), sequence, "resolving a lost result must not publish a new sequence");
+    });
+}
+
+#[test]
+#[serial]
+fn unknown_source_deleted_bucket_writer_failed_mirror_cannot_downgrade_a_committed_owner() {
+    with_writer_fixture(|| async {
+        use crate::heal::mrf_queue::{MRF_SCOPED_JOURNAL_PATH, journal_disks, read_mrf_lifecycle_checkpoint};
+        use crate::heal::storage_api::EcstoreDiskAPI;
+
+        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, None);
+        let mut runtime = runtime();
+        let original = intent("writer-failed-mirror");
+        let key = PartialWriteKey::new(&original, None);
+        runtime.admit_partial_write(original, None).expect("retain responsibility");
+        assert!(runtime.flush().await);
+        let owner = runtime.runtime_checkpoint.expect("old checkpoint");
+        let mut obstacles = Vec::new();
+        for disk in journal_disks().await {
+            let path = EcstoreDiskAPI::path(disk.as_ref())
+                .join(crate::heal::RUSTFS_META_BUCKET)
+                .join(MRF_SCOPED_JOURNAL_PATH);
+            std::fs::remove_file(&path).expect("remove the isolated fixture mirror");
+            std::fs::create_dir(&path).expect("prevent only compatibility-mirror publication");
+            obstacles.push(path);
+        }
+        runtime
+            .partial_writes
+            .entries
+            .get_mut(&key)
+            .expect("legacy entry")
+            .next_attempt = Instant::now();
+        runtime.dispatch_partial_writes(&manager).await;
+        assert_eq!(runtime.runtime_checkpoint, Some((owner.0, owner.1 + 1)));
+        assert_eq!(runtime.lifecycle_format_version, 2);
+        assert!(runtime.partial_writes.pending_lifecycle_transition.is_none());
+        assert!(runtime.dirty, "the failed mirror must be retried");
+        let committed = read_mrf_lifecycle_checkpoint(
+            owner.0,
+            owner.1 + 1,
+            runtime.config.journal_max_bytes,
+            runtime.config.journal_max_bytes * 4,
+        )
+        .await
+        .expect("read committed owner")
+        .expect("committed lifecycle");
+        assert_eq!(committed.format_version, 2);
+        assert!(matches!(
+            committed.records[0].state,
+            ResponsibilityState::SourceAndCurrentGenerationUnknown { .. }
+        ));
+        for path in obstacles {
+            std::fs::remove_dir(path).expect("remove the empty mirror obstacle");
+        }
+        assert!(runtime.flush().await);
+        assert!(!runtime.dirty);
+        let retried = read_mrf_lifecycle_checkpoint(
+            owner.0,
+            owner.1 + 2,
+            runtime.config.journal_max_bytes,
+            runtime.config.journal_max_bytes * 4,
+        )
+        .await
+        .expect("read repaired mirror owner")
+        .expect("repaired lifecycle");
+        assert_eq!(retried.format_version, 2);
+        assert_eq!(retried.records, committed.records);
+    });
+}

crates/heal/src/heal/mrf_queue/snapshot.rs

@@ -61,6 +61,8 @@ pub enum SnapshotError {
     Conflict,
     #[error("MRF checkpoint has no writable replica")]
     NoWritableReplica,
+    #[error("MRF lifecycle writer capability is unavailable")]
+    WriterCapabilityUnavailable,
     #[error("MRF checkpoint storage is unavailable")]
     Disk(#[source] EcstoreDiskError),
     #[error("MRF checkpoint body could not be read")]
@@ -373,6 +375,37 @@ pub async fn publish_committed_snapshot_with_companion(
     limit: usize,
     companion: Option<(&[&str; 2], &[u8], usize)>,
 ) -> Result<SnapshotPublication, SnapshotError> {
+    publish_committed_snapshot_with_companion_guarded(disks, owner, sequence, payload, limit, companion, |_| {
+        std::future::ready(true)
+    })
+    .await
+}
+
+#[derive(Clone, Copy)]
+pub(super) enum SnapshotPublicationStage {
+    Staging,
+    Manifest,
+}
+
+/// Revalidate format authority before staging and before each manifest CAS.
+/// The caller must retain that authority until all admitted disk I/O finishes,
+/// including when its own waiter is cancelled.
+pub(super) async fn publish_committed_snapshot_with_companion_guarded<G, F>(
+    disks: &[EcstoreDiskStore],
+    owner: Uuid,
+    sequence: u64,
+    payload: &[u8],
+    limit: usize,
+    companion: Option<(&[&str; 2], &[u8], usize)>,
+    commit_allowed: G,
+) -> Result<SnapshotPublication, SnapshotError>
+where
+    G: Fn(SnapshotPublicationStage) -> F + Sync,
+    F: std::future::Future<Output = bool>,
+{
+    if !commit_allowed(SnapshotPublicationStage::Staging).await {
+        return Err(SnapshotError::WriterCapabilityUnavailable);
+    }
     if disks.is_empty() {
         return Err(SnapshotError::NoWritableReplica);
     }
@@ -400,6 +433,7 @@ pub async fn publish_committed_snapshot_with_companion(
     let manifest = &manifest;
     let payload = &payload;
     let companion = &companion;
+    let commit_allowed = &commit_allowed;
     let results = journal_disk_io(disks.into_iter().map(|disk| async move {
         let mut payload_written = false;
         let manifest_written: Result<bool, SnapshotError> = async {
@@ -415,6 +449,9 @@ pub async fn publish_committed_snapshot_with_companion(
                     EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch => return Ok(false),
                 }
             }
+            if !commit_allowed(SnapshotPublicationStage::Manifest).await {
+                return Err(SnapshotError::WriterCapabilityUnavailable);
+            }
             Ok(matches!(
                 cas_replace_expected(
                     &disk,
@@ -1559,6 +1596,122 @@ mod tests {
         assert_eq!(recovered.manifest.sequence, 1);
     }
 
+    #[tokio::test]
+    async fn lifecycle_writer_revocation_before_staging_or_manifest_retains_the_old_commit() {
+        const COMPANIONS: [&str; 2] = ["writer-lifecycle.0", "writer-lifecycle.1"];
+        for allowed_checks in [0, 1] {
+            let root = TempDir::new().expect("test directory");
+            let store = disk(&root, "disk").await;
+            let owner = Uuid::new_v4();
+            let old = payload("old");
+            let next = payload("next");
+            publish_committed_snapshot_with_companion(
+                std::slice::from_ref(&store),
+                owner,
+                1,
+                &old,
+                4096,
+                Some((&COMPANIONS, b"old-state", 4096)),
+            )
+            .await
+            .expect("old complete checkpoint");
+            let calls = std::sync::atomic::AtomicUsize::new(0);
+            let result = publish_committed_snapshot_with_companion_guarded(
+                std::slice::from_ref(&store),
+                owner,
+                2,
+                &next,
+                4096,
+                Some((&COMPANIONS, b"new-state", 4096)),
+                |_| std::future::ready(calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) < allowed_checks),
+            )
+            .await;
+            assert!(matches!(result, Err(SnapshotError::WriterCapabilityUnavailable)));
+            let reopened = disk(&root, "disk").await;
+            let recovered = read_committed(std::slice::from_ref(&reopened), 4096)
+                .await
+                .expect("read after interruption")
+                .expect("old commit survives");
+            assert_eq!(recovered.sequence(), 1);
+            assert_eq!(recovered.payload(), old.as_slice());
+            assert_eq!(
+                read_committed_companion(std::slice::from_ref(&reopened), owner, 1, &COMPANIONS, 4096, 4096,)
+                    .await
+                    .expect("old matching companion"),
+                Some(b"old-state".to_vec())
+            );
+            assert!(
+                EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[1])
+                    .await
+                    .is_err()
+            );
+            if allowed_checks == 1 {
+                assert_eq!(
+                    EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, COMPANIONS[1])
+                        .await
+                        .expect("staged companion remains evidence")
+                        .as_ref(),
+                    b"new-state"
+                );
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn lifecycle_writer_partial_manifest_success_reports_only_committed_replicas() {
+        const COMPANIONS: [&str; 2] = ["writer-lifecycle.0", "writer-lifecycle.1"];
+        let root = TempDir::new().expect("test directory");
+        let disks = [disk(&root, "first").await, disk(&root, "second").await];
+        let owner = Uuid::new_v4();
+        let old = payload("old");
+        let next = payload("next");
+        publish_committed_snapshot_with_companion(&disks, owner, 1, &old, 4096, Some((&COMPANIONS, b"old-state", 4096)))
+            .await
+            .expect("old replicated checkpoint");
+        let calls = std::sync::atomic::AtomicUsize::new(0);
+        let published = publish_committed_snapshot_with_companion_guarded(
+            &disks,
+            owner,
+            2,
+            &next,
+            4096,
+            Some((&COMPANIONS, b"new-state", 4096)),
+            |_| std::future::ready(calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) < 2),
+        )
+        .await
+        .expect("one complete replica preserves the owner's established contract");
+        assert_eq!(published.payload_replicas, 2);
+        assert_eq!(published.manifest_replicas, 1);
+        let mut actual_manifests = 0;
+        for store in &disks {
+            actual_manifests += usize::from(
+                EcstoreDiskAPI::read_all(store.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[1])
+                    .await
+                    .is_ok(),
+            );
+            assert_eq!(
+                EcstoreDiskAPI::read_all(store.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0])
+                    .await
+                    .expect("predecessor retained")
+                    .as_ref(),
+                manifest(owner, 1, &old)
+            );
+        }
+        assert_eq!(actual_manifests, 1);
+        let recovered = read_committed(&disks, 4096)
+            .await
+            .expect("recover partial publication")
+            .expect("complete successor");
+        assert_eq!(recovered.sequence(), 2);
+        assert_eq!(recovered.payload(), next.as_slice());
+        assert_eq!(
+            read_committed_companion(&disks, owner, 2, &COMPANIONS, 4096, 4096)
+                .await
+                .expect("companion from the committed replica"),
+            Some(b"new-state".to_vec())
+        );
+    }
+
     #[tokio::test]
     async fn manifest_cas_failure_after_payload_write_keeps_previous_anchor() {
         let root = TempDir::new().expect("test directory");

crates/heal/src/heal/mrf_queue/tests/empty_events_tests.rs

@@ -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,

crates/heal/src/heal/mrf_queue/tests/lifecycle_replay_tests.rs

@@ -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,

crates/heal/src/heal/mrf_queue/tests/lifecycle_v2_reader_tests.rs

@@ -0,0 +1,373 @@
+// 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_reader_capability_matches_supported_formats() {
+    let mut payload = unknown_generation_payload();
+    payload["records"][0]["state"] = json!({"state": "active"});
+    for version in 0..=MRF_LIFECYCLE_READER_MAX_FORMAT_VERSION + 1 {
+        payload["format_version"] = json!(version);
+        assert_eq!(
+            decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&payload, version), 8192).is_ok(),
+            (1..=MRF_LIFECYCLE_READER_MAX_FORMAT_VERSION).contains(&version),
+            "advertised reader support must match the actual decoder for format {version}"
+        );
+    }
+}
+
+#[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<_, _>>(),
+        );
+    }
+}

crates/heal/src/heal/storage_api.rs

@@ -31,6 +31,18 @@ pub(crate) use rustfs_ecstore::api::error::{Error as EcstoreErrorType, StorageEr
 pub(crate) use rustfs_ecstore::api::layout::{
     EndpointServerPools as EcstoreEndpointServerPools, Endpoints as EcstoreEndpoints, PoolEndpoints as EcstorePoolEndpoints,
 };
+pub(crate) use rustfs_ecstore::api::notification::{
+    MrfLifecycleWriterFleetProofToken as EcstoreMrfLifecycleWriterFleetProofToken,
+    acquire_mrf_lifecycle_writer_fleet_proof as ecstore_acquire_mrf_lifecycle_writer_fleet_proof,
+    mrf_lifecycle_writer_fleet_proof_current as ecstore_mrf_lifecycle_writer_fleet_proof_current,
+    mrf_lifecycle_writer_fleet_proof_matches as ecstore_mrf_lifecycle_writer_fleet_proof_matches,
+    with_mrf_lifecycle_writer_fleet_proof as ecstore_with_mrf_lifecycle_writer_fleet_proof,
+};
+#[cfg(test)]
+pub(crate) use rustfs_ecstore::api::notification::{
+    new_global_notification_sys as ecstore_new_global_notification_sys,
+    start_remote_version_state_fleet_probe as ecstore_start_remote_version_state_fleet_probe,
+};
 pub(crate) use rustfs_ecstore::api::runtime::local_disk_map_read as ecstore_local_disk_map_read;
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::storage::init_local_disks as ecstore_init_local_disks;

crates/protos/src/lib.rs

@@ -217,6 +217,7 @@ pub const BACKGROUND_HEAL_STATUS_PROTOCOL_VERSION: u32 = 2;
 pub const HEAL_CONTROL_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-heal-control-capability-v4\0";
 pub const REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-tier-remote-version-state-capability-v1\0";
 pub const CROSS_POOL_FENCE_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-cross-pool-fence-capability-v1\0";
+pub const MRF_LIFECYCLE_READER_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-mrf-lifecycle-reader-capability-v1\0";
 pub const ILM_RECOVERY_EXPORT_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-ilm-recovery-export-capability-v1\0";
 pub const TRANSITION_TRANSACTION_COMPACTION_CAPABILITY_PROBE_PREFIX: &[u8] =
     b"rustfs-transition-transaction-compaction-capability-v1\0";
@@ -288,6 +289,30 @@ pub fn is_transition_transaction_compaction_capability_probe(command: &[u8]) ->
         && command.starts_with(TRANSITION_TRANSACTION_COMPACTION_CAPABILITY_PROBE_PREFIX)
 }
 
+pub fn mrf_lifecycle_reader_capability_probe(nonce: &[u8; 16]) -> Vec<u8> {
+    let mut probe = Vec::with_capacity(MRF_LIFECYCLE_READER_CAPABILITY_PROBE_PREFIX.len() + nonce.len());
+    probe.extend_from_slice(MRF_LIFECYCLE_READER_CAPABILITY_PROBE_PREFIX);
+    probe.extend_from_slice(nonce);
+    probe
+}
+
+pub fn is_mrf_lifecycle_reader_capability_probe(command: &[u8]) -> bool {
+    command.len() == MRF_LIFECYCLE_READER_CAPABILITY_PROBE_PREFIX.len() + 16
+        && command.starts_with(MRF_LIFECYCLE_READER_CAPABILITY_PROBE_PREFIX)
+}
+
+pub fn encode_mrf_lifecycle_reader_capability(
+    maximum_format_version: u8,
+    topology_member: &str,
+    process_epoch: &[u8; 16],
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let identity = encode_remote_version_state_capability(topology_member, process_epoch)?;
+    let mut result = Vec::with_capacity(1 + identity.len());
+    result.push(maximum_format_version);
+    result.extend_from_slice(&identity);
+    Ok(result)
+}
+
 pub fn encode_remote_version_state_capability(
     topology_member: &str,
     process_epoch: &[u8; 16],
@@ -2478,6 +2503,39 @@ mod heal_control_tests {
     use crate::heal_control;
     use std::time::Duration;
 
+    #[test]
+    fn mrf_lifecycle_reader_capability_probe_is_distinct_and_strict() {
+        let probe = super::mrf_lifecycle_reader_capability_probe(&[7; 16]);
+        let mut expected = b"rustfs-mrf-lifecycle-reader-capability-v1\0".to_vec();
+        expected.extend_from_slice(&[7; 16]);
+        assert_eq!(probe, expected);
+        assert!(super::is_mrf_lifecycle_reader_capability_probe(&probe));
+        assert!(!is_heal_control_capability_probe(&probe));
+        assert!(!is_cross_pool_fence_capability_probe(&probe));
+        assert!(!is_remote_version_state_capability_probe(&probe));
+        for length in 0..probe.len() {
+            assert!(!super::is_mrf_lifecycle_reader_capability_probe(&probe[..length]));
+        }
+        let mut trailing = probe.clone();
+        trailing.push(0);
+        assert!(!super::is_mrf_lifecycle_reader_capability_probe(&trailing));
+        let mut unknown = probe;
+        unknown[0] ^= 1;
+        assert!(!super::is_mrf_lifecycle_reader_capability_probe(&unknown));
+        assert!(!super::is_mrf_lifecycle_reader_capability_probe(&heal_control_capability_probe(&[7; 16])));
+    }
+
+    #[test]
+    fn mrf_lifecycle_reader_capability_carries_the_reported_format_and_identity() {
+        let epoch = [9; 16];
+        let result = super::encode_mrf_lifecycle_reader_capability(2, "node-a:9000", &epoch).expect("reader capability");
+        assert_eq!(result[0], 2);
+        assert_eq!(result.len(), 1 + 8 + "node-a:9000".len() + 16);
+        let (member, decoded_epoch) = decode_remote_version_state_capability(&result[1..]).expect("capability identity");
+        assert_eq!(member, "node-a:9000");
+        assert_eq!(decoded_epoch, &epoch);
+    }
+
     #[test]
     fn canonical_heal_control_body_binds_every_field_and_boundary() {
         let baseline = canonical_heal_control_request_body(1, "ab", b"c").expect("small request should encode");

docs/operations/mrf-responsibilities.md

@@ -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,18 @@ 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.
+The writer is disabled by default. With `RUSTFS_MRF_LIFECYCLE_V2_ENABLE=false`, 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.
 
-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.
+
+Peers can query a dedicated MRF lifecycle reader capability through the authenticated heal-control RPC. Its response binds the maximum readable lifecycle format, topology member and process epoch to the requested topology and fresh challenge. Old peers that cannot answer this probe return an error; the general heal-control and cross-pool capabilities are not substitutes.
+
+Set `RUSTFS_MRF_LIFECYCLE_V2_ENABLE=true` only after deploying compatible readers to every member and qualifying the rollback binary. For each new unresolved-generation state, the writer requires a fresh authenticated observation of every configured remote member and its own decoder. Missing, unreachable, duplicate, unknown, or older members prevent promotion. Each observation has a five-second deadline. One observation establishes the publication permit; a second, shared by all replicas, rechecks the fleet before the first manifest commit. The permit has a fixed 30-second deadline that later probe refreshes cannot extend. Every manifest commit checks the current permit locally. Local disk workers retain the publication permit through native I/O, including after an async timeout or cancellation, and recheck it before the publishing rename.
+
+The capability probe does not persist a cluster version floor. The minimum rollback reader must support lifecycle format 2 and preserve unknown-generation records on restart and subsequent flush. The initial qualified rollback candidate is [d8fe88ad4cb57c74f5061fccf6ffc1b4932368aa](https://github.com/rustfs/rustfs/commit/d8fe88ad4cb57c74f5061fccf6ffc1b4932368aa), which includes the V2 reader and the dedicated capability response. Pin and verify the deployed binary or image digest; this commit is not a released-version guarantee. Disabling the writer prevents new promotions but does not downgrade an existing V2 owner. Keep a compatible reader on every member and never convert the retained state to `Active` or remove its companion to force a rollback.
+
+## Interrupted writer publication
+
+The consumer keeps the original visible responsibility until a matching payload, lifecycle companion, and manifest have committed. It preserves the responsibility ID, intent digest, scope, and unknown source identity. The first successful unresolved lookup in the current process establishes the first-held timestamp; denied writer admission does not reset it. V1 data cannot establish observations made before a previous restart. A compatibility-mirror failure after that commit leaves the owner in V2 and retries the mirror; it cannot undo the committed state.
+
+An uncertain publication freezes that owner's checkpoint body. Listing remains available, while new durable admissions and lifecycle actions return a persistence error. The consumer retries the same proposal within the existing five-to-60-second backoff. An exact owner/sequence/body readback can confirm a lost successful result without requesting new writer authority. Conflicting or unreadable evidence remains retained. Restore reader connectivity or the storage fault and let this retry path proceed; do not delete journals, companions, or manifests.

rustfs/src/storage/rpc/node_service.rs

@@ -1230,11 +1230,13 @@ impl heal_control_service_server::HealControlService for HealControlRpcService {
         }
         let remote_version_state_probe = rustfs_protos::is_remote_version_state_capability_probe(&request.get_ref().command);
         let cross_pool_fence_probe = rustfs_protos::is_cross_pool_fence_capability_probe(&request.get_ref().command);
+        let mrf_lifecycle_reader_probe = rustfs_protos::is_mrf_lifecycle_reader_capability_probe(&request.get_ref().command);
         let recovery_export_probe = rustfs_protos::is_ilm_recovery_export_capability_probe(&request.get_ref().command);
         let transition_transaction_compaction_probe =
             rustfs_protos::is_transition_transaction_compaction_capability_probe(&request.get_ref().command);
         if remote_version_state_probe
             || cross_pool_fence_probe
+            || mrf_lifecycle_reader_probe
             || recovery_export_probe
             || transition_transaction_compaction_probe
         {
@@ -1247,12 +1249,21 @@ impl heal_control_service_server::HealControlService for HealControlRpcService {
             if topology_member.is_empty() {
                 return Err(Status::failed_precondition("local topology member identity is unavailable"));
             }
-            let result = encode_heal_capability_response(
-                &topology_member,
-                remote_version_state_probe,
-                recovery_export_probe,
-                transition_transaction_compaction_probe,
-            )?;
+            let result = if mrf_lifecycle_reader_probe {
+                rustfs_protos::encode_mrf_lifecycle_reader_capability(
+                    rustfs_heal::heal::mrf_queue::MRF_LIFECYCLE_READER_MAX_FORMAT_VERSION,
+                    &topology_member,
+                    crate::storage::storage_api::mrf_lifecycle_reader_local_process_epoch().as_bytes(),
+                )
+                .map_err(|_| Status::internal("MRF lifecycle reader capability length cannot be represented"))?
+            } else {
+                encode_heal_capability_response(
+                    &topology_member,
+                    remote_version_state_probe,
+                    recovery_export_probe,
+                    transition_transaction_compaction_probe,
+                )?
+            };
             let canonical_response = rustfs_protos::canonical_heal_control_response_body(
                 request.get_ref().version,
                 &request.get_ref().topology_fingerprint,
@@ -5846,6 +5857,159 @@ mod tests {
         assert_eq!(non_coordinator.code(), tonic::Code::FailedPrecondition);
     }
 
+    #[tokio::test]
+    async fn mrf_lifecycle_reader_probe_authenticates_supported_format_and_identity() {
+        let _ = rustfs_credentials::set_global_rpc_secret("mrf-lifecycle-reader-test-secret".to_string());
+        let endpoints = heal_control_test_endpoints_with_coordinator("node-0", true);
+        assert!(
+            !super::heal::heal_control_coordinator(&endpoints)
+                .expect("test coordinator")
+                .is_local
+        );
+        let fingerprint = heal_topology_fingerprint(&endpoints).expect("test topology");
+        let (service, source) = super::make_heal_control_server_for_source();
+        *source.write().await = Some(endpoints);
+        let mut command = b"rustfs-mrf-lifecycle-reader-capability-v1\0".to_vec();
+        command.extend_from_slice(&[7; 16]);
+        let mut request = Request::new(HealControlRequest {
+            version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION,
+            topology_fingerprint: fingerprint.clone(),
+            command: Bytes::from(command.clone()),
+        });
+        let body = rustfs_protos::canonical_heal_control_request_body(
+            request.get_ref().version,
+            &request.get_ref().topology_fingerprint,
+            &request.get_ref().command,
+        )
+        .expect("probe body");
+        set_tonic_canonical_body_digest(&mut request, &body).expect("request digest");
+        mark_v2_authenticated(&mut request);
+        let response = service
+            .heal_control(request)
+            .await
+            .expect("every topology member must report its MRF reader capability")
+            .into_inner();
+        assert!(response.success);
+        assert_eq!(response.error_info, None);
+        assert_eq!(response.result[0], 2);
+        let (member, epoch) =
+            rustfs_protos::decode_remote_version_state_capability(&response.result[1..]).expect("reader capability identity");
+        assert_eq!(member, "node-a:9000");
+        assert!(!Uuid::from_slice(epoch).expect("process epoch").is_nil());
+        assert_eq!(
+            Uuid::from_slice(epoch).expect("shared reader epoch"),
+            crate::storage::storage_api::mrf_lifecycle_reader_local_process_epoch(),
+        );
+        let canonical = rustfs_protos::canonical_heal_control_response_body(
+            rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION,
+            &fingerprint,
+            &command,
+            &response.result,
+        )
+        .expect("signed response");
+        crate::storage::storage_api::verify_tonic_rpc_response_proof(&canonical, &response.response_proof)
+            .expect("proof must bind the advertised reader and process identity");
+        let mut altered_version = response.result.to_vec();
+        altered_version[0] = 1;
+        let mut altered_member = response.result.to_vec();
+        altered_member[9] ^= 1;
+        let mut altered_epoch = response.result.to_vec();
+        *altered_epoch.last_mut().expect("epoch byte") ^= 1;
+        for result in [altered_version, altered_member, altered_epoch] {
+            let changed = rustfs_protos::canonical_heal_control_response_body(
+                rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION,
+                &fingerprint,
+                &command,
+                &result,
+            )
+            .expect("changed response");
+            assert!(crate::storage::storage_api::verify_tonic_rpc_response_proof(&changed, &response.response_proof).is_err());
+        }
+        for (version, topology, challenge) in [
+            (rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION + 1, fingerprint.clone(), command.clone()),
+            (rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, "other-topology".to_owned(), command),
+            (
+                rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION,
+                fingerprint,
+                rustfs_protos::mrf_lifecycle_reader_capability_probe(&[8; 16]),
+            ),
+        ] {
+            let changed = rustfs_protos::canonical_heal_control_response_body(version, &topology, &challenge, &response.result)
+                .expect("different request");
+            assert!(crate::storage::storage_api::verify_tonic_rpc_response_proof(&changed, &response.response_proof).is_err());
+        }
+    }
+
+    #[tokio::test]
+    async fn mrf_lifecycle_reader_probe_rejects_missing_auth_wrong_topology_and_malformed_command() {
+        let _ = rustfs_credentials::set_global_rpc_secret("mrf-lifecycle-reader-boundary-test-secret".to_string());
+        let endpoints = heal_control_test_endpoints_with_coordinator("node-0", true);
+        assert!(
+            !super::heal::heal_control_coordinator(&endpoints)
+                .expect("coordinator")
+                .is_local
+        );
+        let fingerprint = heal_topology_fingerprint(&endpoints).expect("topology");
+        let (service, source) = super::make_heal_control_server_for_source();
+        *source.write().await = Some(endpoints);
+        let command = rustfs_protos::mrf_lifecycle_reader_capability_probe(&[7; 16]);
+        let request = |topology: &str, command: Vec<u8>, authenticated: bool| {
+            let mut request = Request::new(HealControlRequest {
+                version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION,
+                topology_fingerprint: topology.to_owned(),
+                command: Bytes::from(command),
+            });
+            if authenticated {
+                let body = rustfs_protos::canonical_heal_control_request_body(
+                    request.get_ref().version,
+                    &request.get_ref().topology_fingerprint,
+                    &request.get_ref().command,
+                )
+                .expect("request body");
+                set_tonic_canonical_body_digest(&mut request, &body).expect("body digest");
+                mark_v2_authenticated(&mut request);
+            }
+            request
+        };
+        let unauthenticated = service
+            .heal_control(request(&fingerprint, command.clone(), false))
+            .await
+            .expect_err("unsigned probe must fail");
+        assert_eq!(unauthenticated.code(), tonic::Code::PermissionDenied);
+        let divergent = service
+            .heal_control(request("other-topology", command.clone(), true))
+            .await
+            .expect_err("divergent topology must fail");
+        assert_eq!(divergent.code(), tonic::Code::FailedPrecondition);
+        let mut tampered = request(&fingerprint, command.clone(), true);
+        let mut changed_command = command.clone();
+        changed_command[0] ^= 1;
+        tampered.get_mut().command = Bytes::from(changed_command);
+        assert_eq!(
+            service
+                .heal_control(tampered)
+                .await
+                .expect_err("body tampering must fail")
+                .code(),
+            tonic::Code::PermissionDenied
+        );
+        let mut malformed = command.clone();
+        malformed.push(0);
+        assert!(
+            service.heal_control(request(&fingerprint, malformed, true)).await.is_err(),
+            "a malformed probe must never receive a capability acknowledgement"
+        );
+        let (uninitialized, _) = super::make_heal_control_server_for_source();
+        assert_eq!(
+            uninitialized
+                .heal_control(request(&fingerprint, command, true))
+                .await
+                .expect_err("missing topology must fail")
+                .code(),
+            tonic::Code::FailedPrecondition
+        );
+    }
+
     #[tokio::test]
     async fn remote_version_state_probe_authenticates_topology_challenge_and_process_epoch() {
         let _ = rustfs_credentials::set_global_rpc_secret("remote-version-state-node-service-test-secret".to_string());

rustfs/src/storage/storage_api.rs

@@ -544,7 +544,8 @@ pub(crate) mod ecstore_notification {
     pub(crate) use rustfs_ecstore::api::notification::{
         ClusterTierDailyStats, CrossPoolFenceFleetProofToken, NotificationSys, acquire_cross_pool_fence_fleet_proof,
         configured_fleet_topology_fingerprint, cross_pool_fence_fleet_proof_matches, get_global_notification_sys,
-        ilm_recovery_export_local_process_epoch, new_global_notification_sys, start_remote_version_state_fleet_probe,
+        ilm_recovery_export_local_process_epoch, mrf_lifecycle_reader_local_process_epoch, new_global_notification_sys,
+        start_remote_version_state_fleet_probe,
     };
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::notification::{
@@ -1329,6 +1330,10 @@ pub(crate) fn ilm_recovery_export_local_process_epoch() -> uuid::Uuid {
     ecstore_notification::ilm_recovery_export_local_process_epoch()
 }
 
+pub(crate) fn mrf_lifecycle_reader_local_process_epoch() -> uuid::Uuid {
+    ecstore_notification::mrf_lifecycle_reader_local_process_epoch()
+}
+
 pub(crate) async fn read_config(api: Arc<ECStore>, file: &str) -> Result<Vec<u8>> {
     ecstore_config::com::read_config(api, file).await
 }