fix(tier): replay committed mutation intents (#5107)

Keep recovered tier mutation blocks installed until the local publish transition has atomically established draining for the affected tiers, so old-generation leases cannot slip in after peer commit replay and before local publish.

Delay committed intent cleanup until local publish succeeds. If peer replay succeeds but local publish fails, the durable committed intent remains available for retry and the runtime block stays fail-closed.

Co-authored-by: heihutu <heihutu@gmail.com>

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

@@ -21,10 +21,12 @@ use uuid::Uuid;
 use crate::config::com;
 use crate::disk::RUSTFS_META_BUCKET;
 use crate::error::{Error, Result as EcstoreResult};
-use crate::object_api::ObjectOptions;
+use crate::object_api::{ObjectInfo, ObjectOptions};
 use crate::services::tier::tier::TierDestinationId;
-use crate::storage_api_contracts::{list::ListOperations as _, object::HTTPPreconditions};
-use crate::store::ECStore;
+use crate::storage_api_contracts::{
+    list::{ListOperations, StorageListObjectsV2Info},
+    object::{EcstoreObjectIO, EcstoreObjectOperations, HTTPPreconditions},
+};
 
 pub(crate) const TIER_MUTATION_INTENT_SCHEMA: &str = "rustfs-tier-mutation-intent-v1";
 pub(crate) const MAX_TIER_MUTATION_INTENT_SIZE: usize = rustfs_protos::TIER_MUTATION_RPC_MAX_PREPARE_PAYLOAD_SIZE;
@@ -340,16 +342,19 @@ pub(crate) fn tier_mutation_intent_id_from_record_object_name(object: &str) -> R
     Uuid::parse_str(mutation_key).map_err(|_| TierMutationIntentError::Corrupt("intent record path has invalid uuid"))
 }
 
-pub(crate) async fn save_tier_mutation_intent_record(api: Arc<ECStore>, intent: &TierMutationIntent) -> EcstoreResult<()> {
+pub(crate) async fn save_tier_mutation_intent_record<S>(api: Arc<S>, intent: &TierMutationIntent) -> EcstoreResult<()>
+where
+    S: EcstoreObjectIO,
+{
     let object = tier_mutation_intent_record_object_name(intent.mutation_id).map_err(tier_mutation_intent_store_error)?;
     let data = intent.encode().map_err(tier_mutation_intent_store_error)?;
     com::save_config(api, &object, data).await
 }
 
-pub(crate) async fn save_tier_mutation_intent_record_if_absent(
-    api: Arc<ECStore>,
-    intent: &TierMutationIntent,
-) -> EcstoreResult<()> {
+pub(crate) async fn save_tier_mutation_intent_record_if_absent<S>(api: Arc<S>, intent: &TierMutationIntent) -> EcstoreResult<()>
+where
+    S: EcstoreObjectIO,
+{
     let object = tier_mutation_intent_record_object_name(intent.mutation_id).map_err(tier_mutation_intent_store_error)?;
     let data = intent.encode().map_err(tier_mutation_intent_store_error)?;
     com::save_config_with_opts(
@@ -368,15 +373,21 @@ pub(crate) async fn save_tier_mutation_intent_record_if_absent(
     .await
 }
 
-pub(crate) async fn load_tier_mutation_intent_record(api: Arc<ECStore>, mutation_id: Uuid) -> EcstoreResult<TierMutationIntent> {
+pub(crate) async fn load_tier_mutation_intent_record<S>(api: Arc<S>, mutation_id: Uuid) -> EcstoreResult<TierMutationIntent>
+where
+    S: EcstoreObjectIO,
+{
     let (intent, _) = load_tier_mutation_intent_record_with_etag(api, mutation_id).await?;
     Ok(intent)
 }
 
-pub(crate) async fn load_tier_mutation_intent_record_with_etag(
-    api: Arc<ECStore>,
+pub(crate) async fn load_tier_mutation_intent_record_with_etag<S>(
+    api: Arc<S>,
     mutation_id: Uuid,
-) -> EcstoreResult<(TierMutationIntent, String)> {
+) -> EcstoreResult<(TierMutationIntent, String)>
+where
+    S: EcstoreObjectIO,
+{
     let object = tier_mutation_intent_record_object_name(mutation_id).map_err(tier_mutation_intent_store_error)?;
     let (data, object_info) = com::read_config_with_metadata(api, &object, &ObjectOptions::default()).await?;
     let etag = object_info
@@ -387,11 +398,14 @@ pub(crate) async fn load_tier_mutation_intent_record_with_etag(
     Ok((intent, etag))
 }
 
-pub(crate) async fn save_tier_mutation_intent_record_if_current(
-    api: Arc<ECStore>,
+pub(crate) async fn save_tier_mutation_intent_record_if_current<S>(
+    api: Arc<S>,
     intent: &TierMutationIntent,
     current_etag: &str,
-) -> EcstoreResult<()> {
+) -> EcstoreResult<()>
+where
+    S: EcstoreObjectIO,
+{
     if current_etag.trim().is_empty() {
         return Err(Error::other("tier mutation intent current ETag is empty"));
     }
@@ -413,7 +427,10 @@ pub(crate) async fn save_tier_mutation_intent_record_if_current(
     .await
 }
 
-pub(crate) async fn delete_tier_mutation_intent_record(api: Arc<ECStore>, mutation_id: Uuid) -> EcstoreResult<()> {
+pub(crate) async fn delete_tier_mutation_intent_record<S>(api: Arc<S>, mutation_id: Uuid) -> EcstoreResult<()>
+where
+    S: EcstoreObjectOperations,
+{
     let object = tier_mutation_intent_record_object_name(mutation_id).map_err(tier_mutation_intent_store_error)?;
     match com::delete_config(api, &object).await {
         Ok(()) | Err(Error::ConfigNotFound) => Ok(()),
@@ -421,12 +438,15 @@ pub(crate) async fn delete_tier_mutation_intent_record(api: Arc<ECStore>, mutati
     }
 }
 
-pub(crate) async fn advance_tier_mutation_intent_record_idempotent(
-    api: Arc<ECStore>,
+pub(crate) async fn advance_tier_mutation_intent_record_idempotent<S>(
+    api: Arc<S>,
     mutation_id: Uuid,
     next: TierMutationIntentState,
     committed_config_etag: Option<String>,
-) -> EcstoreResult<(TierMutationIntent, bool)> {
+) -> EcstoreResult<(TierMutationIntent, bool)>
+where
+    S: EcstoreObjectIO,
+{
     let (mut intent, current_etag) = load_tier_mutation_intent_record_with_etag(api.clone(), mutation_id).await?;
     let advanced = intent
         .advance_idempotent(next, committed_config_etag)
@@ -437,11 +457,14 @@ pub(crate) async fn advance_tier_mutation_intent_record_idempotent(
     Ok((intent, advanced))
 }
 
-pub(crate) async fn list_tier_mutation_intent_records(
-    api: Arc<ECStore>,
+pub(crate) async fn list_tier_mutation_intent_records<S>(
+    api: Arc<S>,
     limit: usize,
     marker: Option<String>,
-) -> EcstoreResult<TierMutationIntentRecordScan> {
+) -> EcstoreResult<TierMutationIntentRecordScan>
+where
+    S: EcstoreObjectIO + ListOperations<Error = Error, ListObjectsV2Info = StorageListObjectsV2Info<ObjectInfo>>,
+{
     if limit == 0 {
         return Err(Error::other("tier mutation intent scan limit must be greater than zero"));
     }

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;
+use super::tier::{TierConfigMgr, 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,
@@ -140,13 +140,27 @@ async fn handle_commit(
 ) -> TierMutationPeerResult<TierMutationPeerOutcome> {
     let committed_config_etag = parse_commit_etag(canonical_payload)?;
     let tier_config_mgr = api.tier_config_mgr();
-    let (intent, applied) = advance_tier_mutation_intent_record_idempotent(
-        api,
+    let (intent, applied) = match advance_tier_mutation_intent_record_idempotent(
+        api.clone(),
         mutation_id,
         TierMutationIntentState::Committed,
-        Some(committed_config_etag),
+        Some(committed_config_etag.clone()),
     )
-    .await?;
+    .await
+    {
+        Ok(result) => result,
+        Err(Error::ConfigNotFound)
+            if tier_config_etag_matches(api, &committed_config_etag)
+                .await
+                .map_err(Error::other)? =>
+        {
+            return Ok(TierMutationPeerOutcome {
+                state: TierMutationPeerState::Committed,
+                applied: false,
+            });
+        }
+        Err(err) => return Err(err.into()),
+    };
     if intent.state == TierMutationIntentState::Committed {
         TierConfigMgr::clear_prepared_mutation_intent_block(&tier_config_mgr, mutation_id).await;
     }

crates/ecstore/src/store/init.rs

@@ -546,7 +546,7 @@ mod tests {
         runtime::{global::set_object_store_resolver, sources as runtime_sources},
         services::tier::{
             test_util::{MockWarmBackend, TransitionCleanupStoreBarrier, register_mock_tier},
-            tier::TierConfigMgr,
+            tier::{TIER_CONFIG_FILE, TierConfigMgr},
             tier_mutation_intent::{
                 TIER_MUTATION_INTENT_RECORD_PREFIX, TierMutationIntent, TierMutationIntentKind, TierMutationIntentState,
                 TierMutationIntentTarget, advance_tier_mutation_intent_record_idempotent, delete_tier_mutation_intent_record,
@@ -1589,6 +1589,47 @@ mod tests {
         assert_eq!(loaded.state, TierMutationIntentState::Committed);
         assert_eq!(loaded.committed_config_etag.as_deref(), Some("new-etag"));
 
+        store
+            .tier_config_mgr()
+            .read()
+            .await
+            .save_tiering_config(store.clone())
+            .await
+            .expect("tier config should persist for cleaned intent commit proof");
+        let tier_config_info = store
+            .get_object_info(
+                RUSTFS_META_BUCKET,
+                &format!("{}/{}", com::CONFIG_PREFIX, TIER_CONFIG_FILE),
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("tier config object info should load");
+        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");
+        let cleaned_commit_retry = handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Commit,
+            mutation_id,
+            tier_config_etag.as_bytes(),
+        )
+        .await
+        .expect("commit retry after durable cleanup should be idempotently terminal");
+        assert!(!cleaned_commit_retry.applied);
+        assert_eq!(cleaned_commit_retry.state, TierMutationPeerState::Committed);
+        let mismatched_cleaned_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("missing intent without a matching committed config ETag must fail closed");
+        assert!(matches!(mismatched_cleaned_commit, TierMutationPeerError::Store(Error::ConfigNotFound)));
+
         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");