fix(tier): recover multi-committed mutation intents (#6296)

* fix(tier): recover multi-committed mutation intents

* fix(tier): recover committed mutations on standalone nodes

crates/e2e_test/src/reliant/tiering.rs

@@ -200,11 +200,19 @@ async fn add_rustfs_tier(hot: &RustFSTestEnvironment, cold: &RustFSTestEnvironme
 
 async fn remove_rustfs_tier_force(hot: &RustFSTestEnvironment) -> TestResult {
     let path = format!("/rustfs/admin/v3/tier/{TIER_NAME}?force=true");
-    let (status, resp) = signed_admin_request(&hot.url, Method::DELETE, &path, None, &hot.access_key, &hot.secret_key).await?;
+    let deadline = Instant::now() + StdDuration::from_secs(30);
-    if !status.is_success() {
+    loop {
-        return Err(format!("RemoveTier(RustFS) failed: status={status}, body={resp}").into());
+        let (status, resp) =
+            signed_admin_request(&hot.url, Method::DELETE, &path, None, &hot.access_key, &hot.secret_key).await?;
+        if status.is_success() {
+            return Ok(());
+        }
+        if !resp.contains("TierNameBackendInUse") || Instant::now() >= deadline {
+            return Err(format!("RemoveTier(RustFS) failed: status={status}, body={resp}").into());
+        }
+        // AddTier cleanup is asynchronous; wait until its committed mutation fence clears.
+        tokio::time::sleep(StdDuration::from_millis(100)).await;
     }
-    Ok(())
 }
 
 /// A current-version `Transition Days=0` rule scoped to the object's prefix.
@@ -1477,15 +1485,6 @@ async fn test_manual_transition_async_tier_failure_reports_terminal_partial() ->
     add_rustfs_tier(&hot, &cold).await?;
 
     hot_client.create_bucket().bucket(MANUAL_TIER_FAILURE_BUCKET).send().await?;
-    let due_mtime = OffsetDateTime::now_utc() - time::Duration::hours(25);
-    put_backdated_single_part_object(
-        &hot_client,
-        MANUAL_TIER_FAILURE_BUCKET,
-        MANUAL_TIER_FAILURE_KEY,
-        b"manual tier failure object",
-        due_mtime,
-    )
-    .await?;
     put_lifecycle_transition_rule(
         &hot_client,
         MANUAL_TIER_FAILURE_BUCKET,
@@ -1496,6 +1495,15 @@ async fn test_manual_transition_async_tier_failure_reports_terminal_partial() ->
     .await?;
     remove_rustfs_tier_force(&hot).await?;
 
+    let due_mtime = OffsetDateTime::now_utc() - time::Duration::hours(25);
+    put_backdated_single_part_object(
+        &hot_client,
+        MANUAL_TIER_FAILURE_BUCKET,
+        MANUAL_TIER_FAILURE_KEY,
+        b"manual tier failure object",
+        due_mtime,
+    )
+    .await?;
     let before_remote_count = cold_tier_object_count(&cold_client).await?;
     let accepted = manual_transition_async_run(&hot, MANUAL_TIER_FAILURE_BUCKET, MANUAL_TIER_FAILURE_PREFIX, false, 10).await?;
     assert_eq!(accepted.state, "accepted");

crates/ecstore/src/runtime/sources.rs

@@ -573,9 +573,7 @@ pub(crate) async fn initialize_local_disk_maps(
 pub(crate) async fn init_tier_config_mgr(store: Arc<ECStore>) -> Result<()> {
     let handle = get_global_tier_config_mgr();
     TierConfigMgr::reload_handle(&handle, store.clone()).await?;
-    if setup_is_dist_erasure().await {
+    tokio::spawn(TierConfigMgr::refresh_tier_config_handle(handle, store));
-        tokio::spawn(TierConfigMgr::refresh_tier_config_handle(handle, store));
-    }
     Ok(())
 }
 

crates/ecstore/src/services/tier/tier_mutation_peer.rs

@@ -17,7 +17,7 @@ use std::sync::Arc;
 use rustfs_protos::{TIER_MUTATION_RPC_PROTOCOL_VERSION, TierMutationRpcPhase};
 use uuid::Uuid;
 
-use super::tier::{TierConfigMgr, tier_config_etag_matches};
+use super::tier::{TierConfigMgr, tier_config_abort_matches, tier_config_commit_matches, tier_config_etag_matches};
 use super::tier_mutation_intent::{
     MAX_TIER_MUTATION_INTENT_SIZE, TierMutationIntent, TierMutationIntentState, advance_tier_mutation_intent_record_idempotent,
     load_tier_mutation_intent_record, save_tier_mutation_intent_record_if_absent,
@@ -53,6 +53,10 @@ pub enum TierMutationPeerError {
     InvalidPayload(String),
     #[error("tier mutation peer intent conflicts with existing record")]
     ConflictingIntent,
+    #[error("tier mutation peer commit proof does not match the persisted tier configuration")]
+    CommitProofMismatch,
+    #[error("tier mutation peer abort proof does not match the persisted tier configuration")]
+    AbortProofMismatch,
     #[error("tier mutation peer runtime error: {0}")]
     Runtime(#[source] AdminError),
     #[error("tier mutation peer store error: {0}")]
@@ -120,8 +124,13 @@ async fn handle_prepare(
                         .await
                         .map_err(TierMutationPeerError::Runtime)?;
                 }
-                TierMutationIntentState::Committed | TierMutationIntentState::Aborted => {
+                TierMutationIntentState::Committed => {
-                    TierConfigMgr::clear_prepared_mutation_intent_block(&tier_config_mgr, mutation_id).await;
+                    TierConfigMgr::apply_committed_mutation_intent_block(&tier_config_mgr, &existing)
+                        .await
+                        .map_err(TierMutationPeerError::Runtime)?;
+                }
+                TierMutationIntentState::Aborted => {
+                    TierConfigMgr::request_committed_mutation_refresh(&tier_config_mgr).await;
                 }
             }
             Ok(TierMutationPeerOutcome {
@@ -140,6 +149,18 @@ async fn handle_commit(
 ) -> TierMutationPeerResult<TierMutationPeerOutcome> {
     let committed_config_etag = parse_commit_etag(canonical_payload)?;
     let tier_config_mgr = api.tier_config_mgr();
+    match load_tier_mutation_intent_record(api.clone(), mutation_id).await {
+        Ok(intent) if intent.state == TierMutationIntentState::Prepared => {
+            let proof_matches = tier_config_commit_matches(api.clone(), &committed_config_etag, intent.candidate_digest)
+                .await
+                .map_err(Error::other)?;
+            if !proof_matches {
+                return Err(TierMutationPeerError::CommitProofMismatch);
+            }
+        }
+        Ok(_) | Err(Error::ConfigNotFound) => {}
+        Err(err) => return Err(err.into()),
+    }
     let (intent, applied) = match advance_tier_mutation_intent_record_idempotent(
         api.clone(),
         mutation_id,
@@ -154,6 +175,9 @@ async fn handle_commit(
                 .await
                 .map_err(Error::other)? =>
         {
+            TierConfigMgr::promote_prepared_mutation_intent_block(&tier_config_mgr, mutation_id)
+                .await
+                .map_err(TierMutationPeerError::Runtime)?;
             return Ok(TierMutationPeerOutcome {
                 state: TierMutationPeerState::Committed,
                 applied: false,
@@ -162,7 +186,9 @@ async fn handle_commit(
         Err(err) => return Err(err.into()),
     };
     if intent.state == TierMutationIntentState::Committed {
-        TierConfigMgr::clear_prepared_mutation_intent_block(&tier_config_mgr, mutation_id).await;
+        TierConfigMgr::apply_committed_mutation_intent_block(&tier_config_mgr, &intent)
+            .await
+            .map_err(TierMutationPeerError::Runtime)?;
     }
     Ok(TierMutationPeerOutcome {
         state: peer_state_from_intent(intent.state),
@@ -178,11 +204,18 @@ async fn handle_abort(
     if !canonical_payload.is_empty() {
         return Err(TierMutationPeerError::InvalidPayload("abort payload must be empty".to_string()));
     }
-    let tier_config_mgr = api.tier_config_mgr();
+    let existing = load_tier_mutation_intent_record(api.clone(), mutation_id).await?;
+    if existing.state == TierMutationIntentState::Prepared
+        && !tier_config_abort_matches(api.clone(), &existing)
+            .await
+            .map_err(Error::other)?
+    {
+        return Err(TierMutationPeerError::AbortProofMismatch);
+    }
     let (intent, applied) =
-        advance_tier_mutation_intent_record_idempotent(api, mutation_id, TierMutationIntentState::Aborted, None).await?;
+        advance_tier_mutation_intent_record_idempotent(api.clone(), mutation_id, TierMutationIntentState::Aborted, None).await?;
     if intent.state == TierMutationIntentState::Aborted {
-        TierConfigMgr::clear_prepared_mutation_intent_block(&tier_config_mgr, mutation_id).await;
+        TierConfigMgr::request_committed_mutation_refresh(&api.tier_config_mgr()).await;
     }
     Ok(TierMutationPeerOutcome {
         state: peer_state_from_intent(intent.state),

crates/ecstore/src/store/init.rs

@@ -575,7 +575,7 @@ mod tests {
         runtime::{global::set_object_store_resolver, sources as runtime_sources},
         services::tier::{
             test_util::{MockWarmBackend, MockWarmOp, TransitionCleanupStoreBarrier, register_mock_tier},
-            tier::{TIER_CONFIG_FILE, TierConfigMgr},
+            tier::{TIER_CONFIG_FILE, TierConfigMgr, tier_config_candidate_digest},
             tier_config::{TierConfig, TierType, TierWasabi},
             tier_mutation_intent::{
                 TIER_MUTATION_INTENT_RECORD_PREFIX, TierMutationIntent, TierMutationIntentKind, TierMutationIntentState,
@@ -4721,10 +4721,29 @@ mod tests {
         let temp_dir = tempfile::tempdir().expect("create temp store dir");
         let (_ctx, store, _shutdown) =
             without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-mutation-peer-handler", &[4])).await;
-        let mutation_id = uuid::Uuid::new_v4();
-        let intent = tier_mutation_peer_test_intent(mutation_id, "COLD-A", [3; 32]);
-        let prepare_payload = intent.encode().expect("prepare intent should encode");
         register_mock_tier(&store.tier_config_mgr(), "COLD-A").await;
+        let (candidate_digest, config_etag) = {
+            let tier_config_mgr = store.tier_config_mgr();
+            let manager = tier_config_mgr.read().await;
+            let candidate_digest = tier_config_candidate_digest(&manager).expect("peer commit candidate digest should build");
+            manager
+                .save_tiering_config(store.clone())
+                .await
+                .expect("peer commit config fixture should persist");
+            let config_info = store
+                .get_object_info(
+                    RUSTFS_META_BUCKET,
+                    &format!("{}/{}", com::CONFIG_PREFIX, TIER_CONFIG_FILE),
+                    &ObjectOptions::default(),
+                )
+                .await
+                .expect("peer commit config fixture should load");
+            (candidate_digest, config_info.etag.expect("peer commit config should carry an ETag"))
+        };
+        let mutation_id = uuid::Uuid::new_v4();
+        let mut intent = tier_mutation_peer_test_intent(mutation_id, "COLD-A", candidate_digest);
+        intent.old_config_etag = Some(config_etag.clone());
+        let prepare_payload = intent.encode().expect("prepare intent should encode");
 
         let prepared = handle_tier_mutation_peer_request(
             store.clone(),
@@ -4766,21 +4785,77 @@ mod tests {
             "prepared retry should keep the existing blocked-tier error: {retried_blocked}"
         );
 
+        let mismatched_commit = handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Commit,
+            mutation_id,
+            b"not-the-current-etag",
+        )
+        .await
+        .expect_err("commit with a mismatched config proof must fail closed");
+        assert!(matches!(mismatched_commit, TierMutationPeerError::CommitProofMismatch));
+
+        register_mock_tier(&store.tier_config_mgr(), "COLD-C").await;
+        let bad_digest_id = uuid::Uuid::new_v4();
+        let mut bad_digest_intent = tier_mutation_peer_test_intent(bad_digest_id, "COLD-C", [9; 32]);
+        bad_digest_intent.old_config_etag = Some(config_etag.clone());
+        handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Prepare,
+            bad_digest_id,
+            &bad_digest_intent.encode().expect("bad digest prepare intent should encode"),
+        )
+        .await
+        .expect("bad digest prepare should install a prepared intent");
+        let mismatched_digest_commit = handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Commit,
+            bad_digest_id,
+            config_etag.as_bytes(),
+        )
+        .await
+        .expect_err("a correct ETag with a mismatched candidate digest must fail closed");
+        assert!(matches!(mismatched_digest_commit, TierMutationPeerError::CommitProofMismatch));
+        let bad_digest_loaded = load_tier_mutation_intent_record(store.clone(), bad_digest_id)
+            .await
+            .expect("mismatched digest must leave the prepared intent durable");
+        assert_eq!(bad_digest_loaded.state, TierMutationIntentState::Prepared);
+        let bad_digest_blocked = match TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-C").await {
+            Ok(_) => panic!("mismatched digest must retain the prepared runtime fence"),
+            Err(err) => err,
+        };
+        assert!(bad_digest_blocked.message.contains("being replaced"), "{bad_digest_blocked}");
+        handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Abort,
+            bad_digest_id,
+            b"",
+        )
+        .await
+        .expect("the negative digest proof fixture should clean up through abort");
+
         let committed = handle_tier_mutation_peer_request(
             store.clone(),
             TIER_MUTATION_RPC_PROTOCOL_VERSION,
             TierMutationRpcPhase::Commit,
             mutation_id,
-            b"new-etag",
+            config_etag.as_bytes(),
         )
         .await
         .expect("commit should advance the prepared peer intent");
         assert!(committed.applied);
         assert_eq!(committed.state, TierMutationPeerState::Committed);
-        drop(
+        let committed_blocked = match TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-A").await {
-            TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-A")
+            Ok(_) => panic!("committed peer mutation must remain blocked until local reload publishes the config"),
-                .await
+            Err(err) => err,
-                .expect("committed peer mutation should clear the prepared runtime block"),
+        };
+        assert!(
+            committed_blocked.message.contains("being replaced"),
+            "committed peer mutation should keep the existing blocked-tier error: {committed_blocked}"
         );
 
         let retried_commit = handle_tier_mutation_peer_request(
@@ -4788,12 +4863,20 @@ mod tests {
             TIER_MUTATION_RPC_PROTOCOL_VERSION,
             TierMutationRpcPhase::Commit,
             mutation_id,
-            b"new-etag",
+            config_etag.as_bytes(),
         )
         .await
         .expect("same commit retry should be idempotent");
         assert!(!retried_commit.applied);
         assert_eq!(retried_commit.state, TierMutationPeerState::Committed);
+        let retried_commit_blocked = match TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-A").await {
+            Ok(_) => panic!("committed retry must keep the tier blocked until local reload"),
+            Err(err) => err,
+        };
+        assert!(
+            retried_commit_blocked.message.contains("being replaced"),
+            "committed retry should keep the existing blocked-tier error: {retried_commit_blocked}"
+        );
 
         let delayed_prepare_retry = handle_tier_mutation_peer_request(
             store.clone(),
@@ -4806,17 +4889,20 @@ mod tests {
         .expect("delayed duplicate prepare should report the durable committed state");
         assert!(!delayed_prepare_retry.applied);
         assert_eq!(delayed_prepare_retry.state, TierMutationPeerState::Committed);
-        drop(
+        let delayed_prepare_blocked = match TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-A").await {
-            TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-A")
+            Ok(_) => panic!("delayed committed prepare retry must preserve the committed runtime block"),
-                .await
+            Err(err) => err,
-                .expect("delayed committed prepare retry must not recreate a runtime block"),
+        };
+        assert!(
+            delayed_prepare_blocked.message.contains("being replaced"),
+            "delayed committed prepare retry should keep the existing blocked-tier error: {delayed_prepare_blocked}"
         );
 
         let loaded = load_tier_mutation_intent_record(store.clone(), mutation_id)
             .await
             .expect("committed peer intent should remain durable");
         assert_eq!(loaded.state, TierMutationIntentState::Committed);
-        assert_eq!(loaded.committed_config_etag.as_deref(), Some("new-etag"));
+        assert_eq!(loaded.committed_config_etag.as_deref(), Some(config_etag.as_str()));
 
         store
             .tier_config_mgr()
@@ -4836,7 +4922,7 @@ mod tests {
         let tier_config_etag = tier_config_info.etag.expect("tier config should carry an ETag");
         delete_tier_mutation_intent_record(store.clone(), mutation_id)
             .await
-            .expect("committed peer intent cleanup should persist");
+            .expect("simulate another node cleaning the shared committed peer intent");
         let cleaned_commit_retry = handle_tier_mutation_peer_request(
             store.clone(),
             TIER_MUTATION_RPC_PROTOCOL_VERSION,
@@ -4848,6 +4934,14 @@ mod tests {
         .expect("commit retry after durable cleanup should be idempotently terminal");
         assert!(!cleaned_commit_retry.applied);
         assert_eq!(cleaned_commit_retry.state, TierMutationPeerState::Committed);
+        let cleaned_commit_blocked = match TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-A").await {
+            Ok(_) => panic!("shared intent cleanup must not clear this node's committed runtime block"),
+            Err(err) => err,
+        };
+        assert!(
+            cleaned_commit_blocked.message.contains("being replaced"),
+            "commit retry after shared cleanup should keep the existing blocked-tier error: {cleaned_commit_blocked}"
+        );
         let mismatched_cleaned_commit = handle_tier_mutation_peer_request(
             store.clone(),
             TIER_MUTATION_RPC_PROTOCOL_VERSION,
@@ -4858,11 +4952,46 @@ mod tests {
         .await
         .expect_err("missing intent without a matching committed config ETag must fail closed");
         assert!(matches!(mismatched_cleaned_commit, TierMutationPeerError::Store(Error::ConfigNotFound)));
+        let refresh_store = store.clone();
+        let refresh_manager = store.tier_config_mgr();
+        let refresh_worker = tokio::spawn(async move {
+            TierConfigMgr::refresh_tier_config_handle_with(refresh_manager, refresh_store).await;
+        });
+        tokio::time::timeout(Duration::from_secs(5), async {
+            loop {
+                if let Ok(lease) = TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-A").await {
+                    drop(lease);
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("matching cleaned commit should wake the refresh worker and clear the committed fence");
+        refresh_worker.abort();
+        let _ = refresh_worker.await;
 
         let abort_id = uuid::Uuid::new_v4();
-        let abort_intent = tier_mutation_peer_test_intent(abort_id, "COLD-B", [4; 32]);
-        let abort_prepare_payload = abort_intent.encode().expect("abort prepare intent should encode");
         register_mock_tier(&store.tier_config_mgr(), "COLD-B").await;
+        store
+            .tier_config_mgr()
+            .read()
+            .await
+            .save_tiering_config(store.clone())
+            .await
+            .expect("abort target tier config should persist");
+        let abort_config_info = store
+            .get_object_info(
+                RUSTFS_META_BUCKET,
+                &format!("{}/{}", com::CONFIG_PREFIX, TIER_CONFIG_FILE),
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("abort target config metadata should load");
+        let abort_config_etag = abort_config_info.etag.expect("abort target config should carry an ETag");
+        let mut abort_intent = tier_mutation_peer_test_intent(abort_id, "COLD-B", [4; 32]);
+        abort_intent.old_config_etag = Some(abort_config_etag);
+        let abort_prepare_payload = abort_intent.encode().expect("abort prepare intent should encode");
         handle_tier_mutation_peer_request(
             store.clone(),
             TIER_MUTATION_RPC_PROTOCOL_VERSION,
@@ -4892,23 +5021,238 @@ mod tests {
         .expect("abort should advance the prepared peer intent");
         assert!(aborted.applied);
         assert_eq!(aborted.state, TierMutationPeerState::Aborted);
-        drop(
+        let aborted_blocked = match TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-B").await {
-            TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-B")
+            Ok(_) => panic!("aborted peer mutation must remain blocked until local recovery cleans it up"),
-                .await
+            Err(err) => err,
-                .expect("aborted peer mutation should clear the prepared runtime block"),
+        };
-        );
+        assert!(aborted_blocked.message.contains("being replaced"), "{aborted_blocked}");
 
         let retried_abort = handle_tier_mutation_peer_request(
-            store,
+            store.clone(),
             TIER_MUTATION_RPC_PROTOCOL_VERSION,
             TierMutationRpcPhase::Abort,
             abort_id,
             b"",
         )
         .await
-        .expect("same abort retry should be idempotent");
+        .expect("same abort retry should be idempotent before recovery cleanup");
         assert!(!retried_abort.applied);
         assert_eq!(retried_abort.state, TierMutationPeerState::Aborted);
+
+        let refresh_store = store.clone();
+        let refresh_manager = store.tier_config_mgr();
+        let refresh_worker = tokio::spawn(async move {
+            TierConfigMgr::refresh_tier_config_handle_with(refresh_manager, refresh_store).await;
+        });
+        tokio::time::timeout(Duration::from_secs(5), async {
+            loop {
+                if let Ok(lease) = TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-B").await {
+                    drop(lease);
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("abort notification should drive cleanup before clearing the prepared fence");
+        refresh_worker.abort();
+        let _ = refresh_worker.await;
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn late_abort_after_config_commit_keeps_fence_until_commit_recovery() {
+        let temp_dir = tempfile::tempdir().expect("create temp store dir");
+        let (_ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-mutation-late-abort", &[4])).await;
+        register_mock_tier(&store.tier_config_mgr(), "COLD-A").await;
+        store
+            .tier_config_mgr()
+            .read()
+            .await
+            .save_tiering_config(store.clone())
+            .await
+            .expect("base tier config should persist");
+        let base_config_info = store
+            .get_object_info(
+                RUSTFS_META_BUCKET,
+                &format!("{}/{}", com::CONFIG_PREFIX, TIER_CONFIG_FILE),
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("base tier config metadata should load");
+        let base_etag = base_config_info.etag.expect("base tier config should carry an ETag");
+
+        register_mock_tier(&store.tier_config_mgr(), "COLD-B").await;
+        let candidate_digest = {
+            let manager = store.tier_config_mgr();
+            let manager = manager.read().await;
+            tier_config_candidate_digest(&manager).expect("candidate digest should build")
+        };
+        let mutation_id = uuid::Uuid::new_v4();
+        let mut intent = tier_mutation_peer_test_intent(mutation_id, "COLD-B", candidate_digest);
+        intent.old_config_etag = Some(base_etag.clone());
+        handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Prepare,
+            mutation_id,
+            &intent.encode().expect("late abort prepare intent should encode"),
+        )
+        .await
+        .expect("prepare should install the runtime fence");
+
+        store
+            .tier_config_mgr()
+            .read()
+            .await
+            .save_tiering_config(store.clone())
+            .await
+            .expect("candidate tier config should persist before the late abort");
+        let committed_config_info = store
+            .get_object_info(
+                RUSTFS_META_BUCKET,
+                &format!("{}/{}", com::CONFIG_PREFIX, TIER_CONFIG_FILE),
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("committed tier config metadata should load");
+        let committed_etag = committed_config_info
+            .etag
+            .expect("committed tier config should carry an ETag");
+        assert_ne!(committed_etag, base_etag);
+
+        let late_abort = handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Abort,
+            mutation_id,
+            b"",
+        )
+        .await
+        .expect_err("an abort after the candidate config commit must fail closed");
+        assert!(matches!(late_abort, TierMutationPeerError::AbortProofMismatch));
+        let prepared = load_tier_mutation_intent_record(store.clone(), mutation_id)
+            .await
+            .expect("rejected late abort must retain the prepared intent");
+        assert_eq!(prepared.state, TierMutationIntentState::Prepared);
+        let blocked = match TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-B").await {
+            Ok(_) => panic!("rejected late abort must retain the runtime fence"),
+            Err(err) => err,
+        };
+        assert!(blocked.message.contains("being replaced"), "{blocked}");
+
+        let committed = handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Commit,
+            mutation_id,
+            committed_etag.as_bytes(),
+        )
+        .await
+        .expect("matching commit should converge the rejected late abort fixture");
+        assert!(committed.applied);
+        assert_eq!(committed.state, TierMutationPeerState::Committed);
+
+        let refresh_store = store.clone();
+        let refresh_manager = store.tier_config_mgr();
+        let refresh_worker = tokio::spawn(async move {
+            TierConfigMgr::refresh_tier_config_handle_with(refresh_manager, refresh_store).await;
+        });
+        tokio::time::timeout(Duration::from_secs(5), async {
+            loop {
+                if let Ok(lease) = TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-B").await {
+                    drop(lease);
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("commit recovery should publish before clearing the late-abort fence");
+        refresh_worker.abort();
+        let _ = refresh_worker.await;
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn missing_record_commit_promotes_prepared_fence_until_worker_publish() {
+        let temp_dir = tempfile::tempdir().expect("create temp store dir");
+        let (_ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-mutation-missing-record-commit", &[4]))
+                .await;
+        register_mock_tier(&store.tier_config_mgr(), "COLD-A").await;
+        let tier_config_mgr = store.tier_config_mgr();
+        let candidate_digest = {
+            let manager = tier_config_mgr.read().await;
+            let digest = tier_config_candidate_digest(&manager).expect("candidate digest should build");
+            manager
+                .save_tiering_config(store.clone())
+                .await
+                .expect("candidate config should persist");
+            digest
+        };
+        let config_info = store
+            .get_object_info(
+                RUSTFS_META_BUCKET,
+                &format!("{}/{}", com::CONFIG_PREFIX, TIER_CONFIG_FILE),
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("candidate config metadata should load");
+        let config_etag = config_info.etag.expect("candidate config should carry an ETag");
+        let mutation_id = uuid::Uuid::new_v4();
+        let intent = tier_mutation_peer_test_intent(mutation_id, "COLD-A", candidate_digest);
+        handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Prepare,
+            mutation_id,
+            &intent.encode().expect("prepare intent should encode"),
+        )
+        .await
+        .expect("prepare should install the runtime fence");
+        delete_tier_mutation_intent_record(store.clone(), mutation_id)
+            .await
+            .expect("simulate shared intent cleanup before the local commit arrives");
+
+        let committed = handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Commit,
+            mutation_id,
+            config_etag.as_bytes(),
+        )
+        .await
+        .expect("matching commit after shared cleanup should be terminal");
+        assert!(!committed.applied);
+        assert_eq!(committed.state, TierMutationPeerState::Committed);
+        let blocked = match TierConfigMgr::acquire_operation_lease(&tier_config_mgr, "COLD-A").await {
+            Ok(_) => panic!("the promoted committed fence must block old-generation leases"),
+            Err(err) => err,
+        };
+        assert!(blocked.message.contains("being replaced"), "{blocked}");
+
+        let refresh_store = store.clone();
+        let refresh_manager = tier_config_mgr.clone();
+        let refresh_worker = tokio::spawn(async move {
+            TierConfigMgr::refresh_tier_config_handle_with(refresh_manager, refresh_store).await;
+        });
+        tokio::time::timeout(Duration::from_secs(5), async {
+            loop {
+                if let Ok(lease) = TierConfigMgr::acquire_operation_lease(&tier_config_mgr, "COLD-A").await {
+                    drop(lease);
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("the commit notification should drive publish and clear the promoted fence");
+        refresh_worker.abort();
+        let _ = refresh_worker.await;
     }
 
     #[cfg(feature = "test-util")]