fix(tier): lock tier config mutations (#5080)

* fix(tier): lock tier config mutations

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(tier): add mutation RPC auth contract (#5082)

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

* fix(tier): add peer mutation handler core (#5084)

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

* fix(tier): add mutation control rpc service (#5087)

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

* fix(tier): recover prepared mutation drains (#5093)

Recover prepared tier mutation intent records into the local tier runtime so a restarted peer fails closed before issuing new remote-tier operation leases or conflicting admin publishes.

Reconcile the recovered block map on each scan so committed, aborted, or removed intents clear stale local blocks instead of wedging the peer until process restart.

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

* fix(tier): prove zero references before tier removal (#5092)

Signed-off-by: houseme <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>

* fix(tier): clear peer mutation runtime blocks (#5094)

Install prepared mutation runtime blocks when peer prepare requests are applied or replayed so followers fail closed immediately before restart recovery.

Clear the in-memory block once peer commit or abort reaches a durable terminal state, including delayed duplicate prepare requests that observe a committed or aborted record.

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

---------

Signed-off-by: houseme <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/api/mod.rs

@@ -437,6 +437,13 @@ pub mod tier {
         };
     }
 
+    pub mod tier_mutation_peer {
+        pub use crate::services::tier::tier_mutation_peer::{
+            MAX_TIER_MUTATION_PEER_COMMIT_ETAG_SIZE, TierMutationPeerError, TierMutationPeerOutcome, TierMutationPeerResult,
+            TierMutationPeerState, handle_tier_mutation_peer_request,
+        };
+    }
+
     pub mod warm_backend {
         pub use crate::services::tier::warm_backend::{
             WarmBackend, WarmBackendGetOpts, WarmBackendImpl, build_transition_put_options, check_warm_backend, new_warm_backend,

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

@@ -1144,6 +1144,56 @@ mod tests {
         assert_eq!(tampered.to_string(), "RPC content SHA-256 mismatch");
     }
 
+    #[test]
+    fn tier_mutation_rpc_contract_requires_method_bound_v2_body_digest() {
+        ensure_test_rpc_secret();
+        let mutation_id = uuid::uuid!("12345678-1234-5678-9abc-def012345678");
+        let body = rustfs_protos::canonical_tier_mutation_rpc_body(
+            rustfs_protos::TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            rustfs_protos::TierMutationRpcPhase::Prepare,
+            mutation_id,
+            b"canonical-tier-mutation-prepare",
+        )
+        .expect("small tier mutation body should encode");
+        let mut request = tonic::Request::new(());
+        set_tonic_canonical_body_digest(&mut request, &body).expect("canonical body digest should be attached");
+        let content_sha256 = request
+            .metadata()
+            .get(RPC_CONTENT_SHA256_HEADER)
+            .and_then(|value| value.to_str().ok());
+        let headers = gen_tonic_signature_headers(
+            "node-a:9000",
+            "node_service.TierMutationControlService",
+            "PrepareTierMutation",
+            content_sha256,
+        )
+        .expect("body-bound tier mutation auth headers should build");
+        request.metadata_mut().as_mut().extend(headers.clone());
+
+        assert!(
+            verify_tonic_rpc_signature("node-a:9000", "/node_service.TierMutationControlService/PrepareTierMutation", &headers)
+                .is_ok(),
+            "tier mutation RPC signature must bind destination, service, method, nonce, and body digest"
+        );
+        let method_replay =
+            verify_tonic_rpc_signature("node-a:9000", "/node_service.TierMutationControlService/CommitTierMutation", &headers)
+                .expect_err("prepare auth must not replay to commit");
+        assert_eq!(method_replay.to_string(), "Invalid RPC v2 signature");
+        let service_replay = verify_tonic_rpc_signature("node-a:9000", "/node_service.NodeService/PrepareTierMutation", &headers)
+            .expect_err("tier mutation auth must not replay to the legacy node service path");
+        assert_eq!(service_replay.to_string(), "Invalid RPC v2 signature");
+        let tampered_body = rustfs_protos::canonical_tier_mutation_rpc_body(
+            rustfs_protos::TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            rustfs_protos::TierMutationRpcPhase::Commit,
+            mutation_id,
+            b"canonical-tier-mutation-prepare",
+        )
+        .expect("small tier mutation body should encode");
+        let tampered =
+            verify_tonic_canonical_body_digest(&request, &tampered_body).expect_err("commit body must not match prepare digest");
+        assert_eq!(tampered.to_string(), "RPC content SHA-256 mismatch");
+    }
+
     #[test]
     fn partial_v2_metadata_fails_closed() {
         ensure_test_rpc_secret();

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

@@ -20,6 +20,7 @@ pub mod tier_config;
 pub mod tier_gen;
 pub mod tier_handlers;
 pub(crate) mod tier_mutation_intent;
+pub mod tier_mutation_peer;
 pub mod warm_backend;
 pub mod warm_backend_aliyun;
 pub mod warm_backend_azure;

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

@@ -27,7 +27,7 @@ use crate::storage_api_contracts::{list::ListOperations as _, object::HTTPPrecon
 use crate::store::ECStore;
 
 pub(crate) const TIER_MUTATION_INTENT_SCHEMA: &str = "rustfs-tier-mutation-intent-v1";
-pub(crate) const MAX_TIER_MUTATION_INTENT_SIZE: usize = 64 * 1024;
+pub(crate) const MAX_TIER_MUTATION_INTENT_SIZE: usize = rustfs_protos::TIER_MUTATION_RPC_MAX_PREPARE_PAYLOAD_SIZE;
 pub(crate) const TIER_MUTATION_INTENT_RECORD_PREFIX: &str = "tier/mutation-intents/records";
 pub(crate) type TierMutationDigest = [u8; 32];
 
@@ -346,6 +346,28 @@ pub(crate) async fn save_tier_mutation_intent_record(api: Arc<ECStore>, intent:
     com::save_config(api, &object, data).await
 }
 
+pub(crate) async fn save_tier_mutation_intent_record_if_absent(
+    api: Arc<ECStore>,
+    intent: &TierMutationIntent,
+) -> EcstoreResult<()> {
+    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(
+        api,
+        &object,
+        data,
+        &ObjectOptions {
+            max_parity: true,
+            http_preconditions: Some(HTTPPreconditions {
+                if_none_match: Some("*".to_string()),
+                ..Default::default()
+            }),
+            ..Default::default()
+        },
+    )
+    .await
+}
+
 pub(crate) async fn load_tier_mutation_intent_record(api: Arc<ECStore>, mutation_id: Uuid) -> EcstoreResult<TierMutationIntent> {
     let (intent, _) = load_tier_mutation_intent_record_with_etag(api, mutation_id).await?;
     Ok(intent)

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

@@ -0,0 +1,278 @@
+// Copyright 2024 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 std::sync::Arc;
+
+use rustfs_protos::{TIER_MUTATION_RPC_PROTOCOL_VERSION, TierMutationRpcPhase};
+use uuid::Uuid;
+
+use super::tier::TierConfigMgr;
+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,
+};
+use crate::client::admin_handler_utils::AdminError;
+use crate::error::{Error, StorageError};
+use crate::store::ECStore;
+
+pub const MAX_TIER_MUTATION_PEER_COMMIT_ETAG_SIZE: usize = rustfs_protos::TIER_MUTATION_RPC_MAX_COMMIT_PAYLOAD_SIZE;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TierMutationPeerState {
+    Prepared,
+    Committed,
+    Aborted,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct TierMutationPeerOutcome {
+    pub state: TierMutationPeerState,
+    pub applied: bool,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum TierMutationPeerError {
+    #[error("unsupported tier mutation peer protocol version: {0}")]
+    UnsupportedProtocolVersion(u32),
+    #[error("tier mutation peer mutation_id is nil")]
+    NilMutationId,
+    #[error("tier mutation peer payload is too large: {len}/{max}")]
+    PayloadTooLarge { len: usize, max: usize },
+    #[error("tier mutation peer payload is invalid: {0}")]
+    InvalidPayload(String),
+    #[error("tier mutation peer intent conflicts with existing record")]
+    ConflictingIntent,
+    #[error("tier mutation peer runtime error: {0}")]
+    Runtime(#[source] AdminError),
+    #[error("tier mutation peer store error: {0}")]
+    Store(#[source] StorageError),
+}
+
+impl From<Error> for TierMutationPeerError {
+    fn from(error: Error) -> Self {
+        Self::Store(error)
+    }
+}
+
+pub type TierMutationPeerResult<T> = std::result::Result<T, TierMutationPeerError>;
+
+pub async fn handle_tier_mutation_peer_request(
+    api: Arc<ECStore>,
+    protocol_version: u32,
+    phase: TierMutationRpcPhase,
+    mutation_id: Uuid,
+    canonical_payload: &[u8],
+) -> TierMutationPeerResult<TierMutationPeerOutcome> {
+    validate_peer_request_envelope(protocol_version, mutation_id, canonical_payload)?;
+    match phase {
+        TierMutationRpcPhase::Prepare => handle_prepare(api, mutation_id, canonical_payload).await,
+        TierMutationRpcPhase::Commit => handle_commit(api, mutation_id, canonical_payload).await,
+        TierMutationRpcPhase::Abort => handle_abort(api, mutation_id, canonical_payload).await,
+        _ => Err(TierMutationPeerError::InvalidPayload(
+            "tier mutation rpc phase is unsupported".to_string(),
+        )),
+    }
+}
+
+async fn handle_prepare(
+    api: Arc<ECStore>,
+    mutation_id: Uuid,
+    canonical_payload: &[u8],
+) -> TierMutationPeerResult<TierMutationPeerOutcome> {
+    let intent = TierMutationIntent::decode(mutation_id, canonical_payload)
+        .map_err(|err| TierMutationPeerError::InvalidPayload(err.to_string()))?;
+    if intent.state != TierMutationIntentState::Prepared {
+        return Err(TierMutationPeerError::InvalidPayload(
+            "prepare intent must be in prepared state".to_string(),
+        ));
+    }
+    let tier_config_mgr = api.tier_config_mgr();
+
+    match save_tier_mutation_intent_record_if_absent(api.clone(), &intent).await {
+        Ok(()) => {
+            TierConfigMgr::apply_prepared_mutation_intent_block(&tier_config_mgr, &intent)
+                .await
+                .map_err(TierMutationPeerError::Runtime)?;
+            Ok(TierMutationPeerOutcome {
+                state: TierMutationPeerState::Prepared,
+                applied: true,
+            })
+        }
+        Err(Error::PreconditionFailed) => {
+            let existing = load_tier_mutation_intent_record(api, mutation_id).await?;
+            if !same_mutation_identity(&existing, &intent) {
+                return Err(TierMutationPeerError::ConflictingIntent);
+            }
+            match existing.state {
+                TierMutationIntentState::Prepared => {
+                    TierConfigMgr::apply_prepared_mutation_intent_block(&tier_config_mgr, &existing)
+                        .await
+                        .map_err(TierMutationPeerError::Runtime)?;
+                }
+                TierMutationIntentState::Committed | TierMutationIntentState::Aborted => {
+                    TierConfigMgr::clear_prepared_mutation_intent_block(&tier_config_mgr, mutation_id).await;
+                }
+            }
+            Ok(TierMutationPeerOutcome {
+                state: peer_state_from_intent(existing.state),
+                applied: false,
+            })
+        }
+        Err(err) => Err(err.into()),
+    }
+}
+
+async fn handle_commit(
+    api: Arc<ECStore>,
+    mutation_id: Uuid,
+    canonical_payload: &[u8],
+) -> 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,
+        mutation_id,
+        TierMutationIntentState::Committed,
+        Some(committed_config_etag),
+    )
+    .await?;
+    if intent.state == TierMutationIntentState::Committed {
+        TierConfigMgr::clear_prepared_mutation_intent_block(&tier_config_mgr, mutation_id).await;
+    }
+    Ok(TierMutationPeerOutcome {
+        state: peer_state_from_intent(intent.state),
+        applied,
+    })
+}
+
+async fn handle_abort(
+    api: Arc<ECStore>,
+    mutation_id: Uuid,
+    canonical_payload: &[u8],
+) -> TierMutationPeerResult<TierMutationPeerOutcome> {
+    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 (intent, applied) =
+        advance_tier_mutation_intent_record_idempotent(api, mutation_id, TierMutationIntentState::Aborted, None).await?;
+    if intent.state == TierMutationIntentState::Aborted {
+        TierConfigMgr::clear_prepared_mutation_intent_block(&tier_config_mgr, mutation_id).await;
+    }
+    Ok(TierMutationPeerOutcome {
+        state: peer_state_from_intent(intent.state),
+        applied,
+    })
+}
+
+fn validate_peer_request_envelope(
+    protocol_version: u32,
+    mutation_id: Uuid,
+    canonical_payload: &[u8],
+) -> TierMutationPeerResult<()> {
+    if protocol_version != TIER_MUTATION_RPC_PROTOCOL_VERSION {
+        return Err(TierMutationPeerError::UnsupportedProtocolVersion(protocol_version));
+    }
+    if mutation_id.is_nil() {
+        return Err(TierMutationPeerError::NilMutationId);
+    }
+    if canonical_payload.len() > MAX_TIER_MUTATION_INTENT_SIZE {
+        return Err(TierMutationPeerError::PayloadTooLarge {
+            len: canonical_payload.len(),
+            max: MAX_TIER_MUTATION_INTENT_SIZE,
+        });
+    }
+    Ok(())
+}
+
+fn parse_commit_etag(canonical_payload: &[u8]) -> TierMutationPeerResult<String> {
+    if canonical_payload.len() > MAX_TIER_MUTATION_PEER_COMMIT_ETAG_SIZE {
+        return Err(TierMutationPeerError::PayloadTooLarge {
+            len: canonical_payload.len(),
+            max: MAX_TIER_MUTATION_PEER_COMMIT_ETAG_SIZE,
+        });
+    }
+    let etag = std::str::from_utf8(canonical_payload)
+        .map_err(|err| TierMutationPeerError::InvalidPayload(err.to_string()))?
+        .trim();
+    if etag.is_empty() {
+        return Err(TierMutationPeerError::InvalidPayload(
+            "commit payload must carry a committed config etag".to_string(),
+        ));
+    }
+    Ok(etag.to_string())
+}
+
+fn peer_state_from_intent(state: TierMutationIntentState) -> TierMutationPeerState {
+    match state {
+        TierMutationIntentState::Prepared => TierMutationPeerState::Prepared,
+        TierMutationIntentState::Committed => TierMutationPeerState::Committed,
+        TierMutationIntentState::Aborted => TierMutationPeerState::Aborted,
+    }
+}
+
+fn same_mutation_identity(existing: &TierMutationIntent, expected: &TierMutationIntent) -> bool {
+    existing.mutation_id == expected.mutation_id
+        && existing.kind == expected.kind
+        && existing.old_config_etag == expected.old_config_etag
+        && existing.candidate_digest == expected.candidate_digest
+        && existing.affected_targets == expected.affected_targets
+        && existing.expires_at_unix_nanos == expected.expires_at_unix_nanos
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn peer_request_envelope_fails_closed_on_old_version_nil_id_and_large_payload() {
+        let mutation_id = Uuid::new_v4();
+        assert!(matches!(
+            validate_peer_request_envelope(TIER_MUTATION_RPC_PROTOCOL_VERSION + 1, mutation_id, b"payload"),
+            Err(TierMutationPeerError::UnsupportedProtocolVersion(_))
+        ));
+        assert!(matches!(
+            validate_peer_request_envelope(TIER_MUTATION_RPC_PROTOCOL_VERSION, Uuid::nil(), b"payload"),
+            Err(TierMutationPeerError::NilMutationId)
+        ));
+
+        let oversized = vec![0; MAX_TIER_MUTATION_INTENT_SIZE + 1];
+        assert!(matches!(
+            validate_peer_request_envelope(TIER_MUTATION_RPC_PROTOCOL_VERSION, mutation_id, &oversized),
+            Err(TierMutationPeerError::PayloadTooLarge { .. })
+        ));
+    }
+
+    #[test]
+    fn commit_payload_requires_small_non_empty_utf8_etag() {
+        assert_eq!(
+            parse_commit_etag(b"  committed-etag  ").expect("etag payload should parse"),
+            "committed-etag"
+        );
+        assert!(matches!(
+            parse_commit_etag(b"   "),
+            Err(TierMutationPeerError::InvalidPayload(message)) if message.contains("committed config etag")
+        ));
+        assert!(matches!(
+            parse_commit_etag(&[0xff]),
+            Err(TierMutationPeerError::InvalidPayload(message)) if message.contains("utf-8")
+        ));
+
+        let oversized = vec![b'a'; MAX_TIER_MUTATION_PEER_COMMIT_ETAG_SIZE + 1];
+        assert!(matches!(
+            parse_commit_etag(&oversized),
+            Err(TierMutationPeerError::PayloadTooLarge { .. })
+        ));
+    }
+}

crates/ecstore/src/store/init.rs

@@ -553,6 +553,7 @@ mod tests {
                 list_tier_mutation_intent_records, load_tier_mutation_intent_record, load_tier_mutation_intent_record_with_etag,
                 save_tier_mutation_intent_record, save_tier_mutation_intent_record_if_current,
             },
+            tier_mutation_peer::{TierMutationPeerError, TierMutationPeerState, handle_tier_mutation_peer_request},
             warm_backend::WarmBackend,
         },
         storage_api_contracts::{
@@ -572,6 +573,8 @@ mod tests {
     };
     use http::HeaderMap;
     use rustfs_config::server_config::KVS;
+    #[cfg(feature = "test-util")]
+    use rustfs_protos::{TIER_MUTATION_RPC_PROTOCOL_VERSION, TierMutationRpcPhase};
     use std::{
         future::Future,
         io::Cursor,
@@ -1459,6 +1462,225 @@ mod tests {
         assert_eq!(aborted_retry, aborted);
     }
 
+    #[cfg(feature = "test-util")]
+    fn tier_mutation_peer_test_intent(
+        mutation_id: uuid::Uuid,
+        tier_name: &str,
+        candidate_digest: [u8; 32],
+    ) -> TierMutationIntent {
+        TierMutationIntent {
+            mutation_id,
+            revision: 1,
+            kind: TierMutationIntentKind::Edit,
+            state: TierMutationIntentState::Prepared,
+            old_config_etag: Some("old-etag".to_string()),
+            committed_config_etag: None,
+            candidate_digest,
+            affected_targets: vec![TierMutationIntentTarget {
+                tier_name: tier_name.to_string(),
+                old_backend_identity: Some([1; 32]),
+                new_backend_identity: Some([2; 32]),
+            }],
+            expires_at_unix_nanos: 1_780_000_000_000_000_000,
+        }
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn tier_mutation_peer_handler_applies_prepare_commit_and_abort_idempotently() {
+        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 prepared = handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Prepare,
+            mutation_id,
+            &prepare_payload,
+        )
+        .await
+        .expect("first prepare should create the peer intent");
+        assert!(prepared.applied);
+        assert_eq!(prepared.state, TierMutationPeerState::Prepared);
+        let blocked = match TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-A").await {
+            Ok(_) => panic!("prepared peer mutation should block new tier operation leases"),
+            Err(err) => err,
+        };
+        assert!(
+            blocked.message.contains("being replaced"),
+            "prepared peer mutation should reuse the existing blocked-tier error: {blocked}"
+        );
+
+        let retried_prepare = handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Prepare,
+            mutation_id,
+            &prepare_payload,
+        )
+        .await
+        .expect("same prepare retry should be idempotent");
+        assert!(!retried_prepare.applied);
+        assert_eq!(retried_prepare.state, TierMutationPeerState::Prepared);
+        let retried_blocked = match TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-A").await {
+            Ok(_) => panic!("prepared retry should keep blocking new tier operation leases"),
+            Err(err) => err,
+        };
+        assert!(
+            retried_blocked.message.contains("being replaced"),
+            "prepared retry should keep the existing blocked-tier error: {retried_blocked}"
+        );
+
+        let committed = handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Commit,
+            mutation_id,
+            b"new-etag",
+        )
+        .await
+        .expect("commit should advance the prepared peer intent");
+        assert!(committed.applied);
+        assert_eq!(committed.state, TierMutationPeerState::Committed);
+        drop(
+            TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-A")
+                .await
+                .expect("committed peer mutation should clear the prepared runtime block"),
+        );
+
+        let retried_commit = handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Commit,
+            mutation_id,
+            b"new-etag",
+        )
+        .await
+        .expect("same commit retry should be idempotent");
+        assert!(!retried_commit.applied);
+        assert_eq!(retried_commit.state, TierMutationPeerState::Committed);
+
+        let delayed_prepare_retry = handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Prepare,
+            mutation_id,
+            &prepare_payload,
+        )
+        .await
+        .expect("delayed duplicate prepare should report the durable committed state");
+        assert!(!delayed_prepare_retry.applied);
+        assert_eq!(delayed_prepare_retry.state, TierMutationPeerState::Committed);
+        drop(
+            TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-A")
+                .await
+                .expect("delayed committed prepare retry must not recreate a runtime block"),
+        );
+
+        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"));
+
+        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;
+        handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Prepare,
+            abort_id,
+            &abort_prepare_payload,
+        )
+        .await
+        .expect("abort target prepare should create the peer intent");
+        let abort_blocked = match TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-B").await {
+            Ok(_) => panic!("abort target prepare should block new tier operation leases"),
+            Err(err) => err,
+        };
+        assert!(
+            abort_blocked.message.contains("being replaced"),
+            "abort target prepare should reuse the existing blocked-tier error: {abort_blocked}"
+        );
+
+        let aborted = handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Abort,
+            abort_id,
+            b"",
+        )
+        .await
+        .expect("abort should advance the prepared peer intent");
+        assert!(aborted.applied);
+        assert_eq!(aborted.state, TierMutationPeerState::Aborted);
+        drop(
+            TierConfigMgr::acquire_operation_lease(&store.tier_config_mgr(), "COLD-B")
+                .await
+                .expect("aborted peer mutation should clear the prepared runtime block"),
+        );
+
+        let retried_abort = handle_tier_mutation_peer_request(
+            store,
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Abort,
+            abort_id,
+            b"",
+        )
+        .await
+        .expect("same abort retry should be idempotent");
+        assert!(!retried_abort.applied);
+        assert_eq!(retried_abort.state, TierMutationPeerState::Aborted);
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn tier_mutation_peer_handler_rejects_conflicting_prepare_without_overwrite() {
+        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-conflict", &[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");
+        handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Prepare,
+            mutation_id,
+            &prepare_payload,
+        )
+        .await
+        .expect("first prepare should create the peer intent");
+
+        let conflicting = tier_mutation_peer_test_intent(mutation_id, "COLD-A", [4; 32]);
+        let conflicting_payload = conflicting.encode().expect("conflicting intent should encode");
+        let conflict = handle_tier_mutation_peer_request(
+            store.clone(),
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Prepare,
+            mutation_id,
+            &conflicting_payload,
+        )
+        .await
+        .expect_err("conflicting prepare must fail closed");
+        assert!(matches!(conflict, TierMutationPeerError::ConflictingIntent));
+
+        let loaded = load_tier_mutation_intent_record(store, mutation_id)
+            .await
+            .expect("conflicting prepare must not overwrite the first record");
+        assert_eq!(loaded.candidate_digest, [3; 32]);
+        assert_eq!(loaded.state, TierMutationIntentState::Prepared);
+    }
+
     #[cfg(feature = "test-util")]
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]

crates/protos/src/generated/proto_gen/node_service.rs

@@ -1223,6 +1223,46 @@ pub struct LoadTransitionTierConfigResponse {
     #[prost(string, optional, tag = "2")]
     pub error_info: ::core::option::Option<::prost::alloc::string::String>,
 }
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct TierMutationPrepareRequest {
+    #[prost(uint32, tag = "1")]
+    pub version: u32,
+    #[prost(string, tag = "2")]
+    pub mutation_id: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "3")]
+    pub canonical_payload: ::prost::bytes::Bytes,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct TierMutationCommitRequest {
+    #[prost(uint32, tag = "1")]
+    pub version: u32,
+    #[prost(string, tag = "2")]
+    pub mutation_id: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "3")]
+    pub canonical_payload: ::prost::bytes::Bytes,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct TierMutationAbortRequest {
+    #[prost(uint32, tag = "1")]
+    pub version: u32,
+    #[prost(string, tag = "2")]
+    pub mutation_id: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "3")]
+    pub canonical_payload: ::prost::bytes::Bytes,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct TierMutationControlResponse {
+    #[prost(bool, tag = "1")]
+    pub success: bool,
+    #[prost(enumeration = "TierMutationPeerState", tag = "2")]
+    pub state: i32,
+    #[prost(bool, tag = "3")]
+    pub applied: bool,
+    #[prost(string, optional, tag = "4")]
+    pub error_info: ::core::option::Option<::prost::alloc::string::String>,
+    #[prost(bytes = "bytes", tag = "5")]
+    pub response_proof: ::prost::bytes::Bytes,
+}
 #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct GetLiveEventsRequest {
     #[prost(uint64, tag = "1")]
@@ -1243,6 +1283,38 @@ pub struct GetLiveEventsResponse {
     #[prost(string, optional, tag = "5")]
     pub error_info: ::core::option::Option<::prost::alloc::string::String>,
 }
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+#[repr(i32)]
+pub enum TierMutationPeerState {
+    Unspecified = 0,
+    Prepared = 1,
+    Committed = 2,
+    Aborted = 3,
+}
+impl TierMutationPeerState {
+    /// String value of the enum field names used in the ProtoBuf definition.
+    ///
+    /// The values are not transformed in any way and thus are considered stable
+    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+    pub fn as_str_name(&self) -> &'static str {
+        match self {
+            Self::Unspecified => "TIER_MUTATION_PEER_STATE_UNSPECIFIED",
+            Self::Prepared => "TIER_MUTATION_PEER_STATE_PREPARED",
+            Self::Committed => "TIER_MUTATION_PEER_STATE_COMMITTED",
+            Self::Aborted => "TIER_MUTATION_PEER_STATE_ABORTED",
+        }
+    }
+    /// Creates an enum from field names used in the ProtoBuf definition.
+    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+        match value {
+            "TIER_MUTATION_PEER_STATE_UNSPECIFIED" => Some(Self::Unspecified),
+            "TIER_MUTATION_PEER_STATE_PREPARED" => Some(Self::Prepared),
+            "TIER_MUTATION_PEER_STATE_COMMITTED" => Some(Self::Committed),
+            "TIER_MUTATION_PEER_STATE_ABORTED" => Some(Self::Aborted),
+            _ => None,
+        }
+    }
+}
 /// Generated client implementations.
 pub mod node_service_client {
     #![allow(unused_variables, dead_code, missing_docs, clippy::wildcard_imports, clippy::let_unit_value)]
@@ -5590,3 +5662,334 @@ pub mod heal_control_service_server {
         const NAME: &'static str = SERVICE_NAME;
     }
 }
+/// Generated client implementations.
+pub mod tier_mutation_control_service_client {
+    #![allow(unused_variables, dead_code, missing_docs, clippy::wildcard_imports, clippy::let_unit_value)]
+    use tonic::codegen::http::Uri;
+    use tonic::codegen::*;
+    #[derive(Debug, Clone)]
+    pub struct TierMutationControlServiceClient<T> {
+        inner: tonic::client::Grpc<T>,
+    }
+    impl TierMutationControlServiceClient<tonic::transport::Channel> {
+        /// Attempt to create a new client by connecting to a given endpoint.
+        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
+        where
+            D: TryInto<tonic::transport::Endpoint>,
+            D::Error: Into<StdError>,
+        {
+            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
+            Ok(Self::new(conn))
+        }
+    }
+    impl<T> TierMutationControlServiceClient<T>
+    where
+        T: tonic::client::GrpcService<tonic::body::Body>,
+        T::Error: Into<StdError>,
+        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
+        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
+    {
+        pub fn new(inner: T) -> Self {
+            let inner = tonic::client::Grpc::new(inner);
+            Self { inner }
+        }
+        pub fn with_origin(inner: T, origin: Uri) -> Self {
+            let inner = tonic::client::Grpc::with_origin(inner, origin);
+            Self { inner }
+        }
+        pub fn with_interceptor<F>(inner: T, interceptor: F) -> TierMutationControlServiceClient<InterceptedService<T, F>>
+        where
+            F: tonic::service::Interceptor,
+            T::ResponseBody: Default,
+            T: tonic::codegen::Service<
+                    http::Request<tonic::body::Body>,
+                    Response = http::Response<<T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody>,
+                >,
+            <T as tonic::codegen::Service<http::Request<tonic::body::Body>>>::Error:
+                Into<StdError> + std::marker::Send + std::marker::Sync,
+        {
+            TierMutationControlServiceClient::new(InterceptedService::new(inner, interceptor))
+        }
+        /// Compress requests with the given encoding.
+        ///
+        /// This requires the server to support it otherwise it might respond with an
+        /// error.
+        #[must_use]
+        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.inner = self.inner.send_compressed(encoding);
+            self
+        }
+        /// Enable decompressing responses.
+        #[must_use]
+        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.inner = self.inner.accept_compressed(encoding);
+            self
+        }
+        /// Limits the maximum size of a decoded message.
+        ///
+        /// Default: `4MB`
+        #[must_use]
+        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
+            self.inner = self.inner.max_decoding_message_size(limit);
+            self
+        }
+        /// Limits the maximum size of an encoded message.
+        ///
+        /// Default: `usize::MAX`
+        #[must_use]
+        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
+            self.inner = self.inner.max_encoding_message_size(limit);
+            self
+        }
+        pub async fn prepare_tier_mutation(
+            &mut self,
+            request: impl tonic::IntoRequest<super::TierMutationPrepareRequest>,
+        ) -> std::result::Result<tonic::Response<super::TierMutationControlResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.TierMutationControlService/PrepareTierMutation");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.TierMutationControlService", "PrepareTierMutation"));
+            self.inner.unary(req, path, codec).await
+        }
+        pub async fn commit_tier_mutation(
+            &mut self,
+            request: impl tonic::IntoRequest<super::TierMutationCommitRequest>,
+        ) -> std::result::Result<tonic::Response<super::TierMutationControlResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.TierMutationControlService/CommitTierMutation");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.TierMutationControlService", "CommitTierMutation"));
+            self.inner.unary(req, path, codec).await
+        }
+        pub async fn abort_tier_mutation(
+            &mut self,
+            request: impl tonic::IntoRequest<super::TierMutationAbortRequest>,
+        ) -> std::result::Result<tonic::Response<super::TierMutationControlResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.TierMutationControlService/AbortTierMutation");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.TierMutationControlService", "AbortTierMutation"));
+            self.inner.unary(req, path, codec).await
+        }
+    }
+}
+/// Generated server implementations.
+pub mod tier_mutation_control_service_server {
+    #![allow(unused_variables, dead_code, missing_docs, clippy::wildcard_imports, clippy::let_unit_value)]
+    use tonic::codegen::*;
+    /// Generated trait containing gRPC methods that should be implemented for use with TierMutationControlServiceServer.
+    #[async_trait]
+    pub trait TierMutationControlService: std::marker::Send + std::marker::Sync + 'static {
+        async fn prepare_tier_mutation(
+            &self,
+            request: tonic::Request<super::TierMutationPrepareRequest>,
+        ) -> std::result::Result<tonic::Response<super::TierMutationControlResponse>, tonic::Status>;
+        async fn commit_tier_mutation(
+            &self,
+            request: tonic::Request<super::TierMutationCommitRequest>,
+        ) -> std::result::Result<tonic::Response<super::TierMutationControlResponse>, tonic::Status>;
+        async fn abort_tier_mutation(
+            &self,
+            request: tonic::Request<super::TierMutationAbortRequest>,
+        ) -> std::result::Result<tonic::Response<super::TierMutationControlResponse>, tonic::Status>;
+    }
+    #[derive(Debug)]
+    pub struct TierMutationControlServiceServer<T> {
+        inner: Arc<T>,
+        accept_compression_encodings: EnabledCompressionEncodings,
+        send_compression_encodings: EnabledCompressionEncodings,
+        max_decoding_message_size: Option<usize>,
+        max_encoding_message_size: Option<usize>,
+    }
+    impl<T> TierMutationControlServiceServer<T> {
+        pub fn new(inner: T) -> Self {
+            Self::from_arc(Arc::new(inner))
+        }
+        pub fn from_arc(inner: Arc<T>) -> Self {
+            Self {
+                inner,
+                accept_compression_encodings: Default::default(),
+                send_compression_encodings: Default::default(),
+                max_decoding_message_size: None,
+                max_encoding_message_size: None,
+            }
+        }
+        pub fn with_interceptor<F>(inner: T, interceptor: F) -> InterceptedService<Self, F>
+        where
+            F: tonic::service::Interceptor,
+        {
+            InterceptedService::new(Self::new(inner), interceptor)
+        }
+        /// Enable decompressing requests with the given encoding.
+        #[must_use]
+        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.accept_compression_encodings.enable(encoding);
+            self
+        }
+        /// Compress responses with the given encoding, if the client supports it.
+        #[must_use]
+        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.send_compression_encodings.enable(encoding);
+            self
+        }
+        /// Limits the maximum size of a decoded message.
+        ///
+        /// Default: `4MB`
+        #[must_use]
+        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
+            self.max_decoding_message_size = Some(limit);
+            self
+        }
+        /// Limits the maximum size of an encoded message.
+        ///
+        /// Default: `usize::MAX`
+        #[must_use]
+        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
+            self.max_encoding_message_size = Some(limit);
+            self
+        }
+    }
+    impl<T, B> tonic::codegen::Service<http::Request<B>> for TierMutationControlServiceServer<T>
+    where
+        T: TierMutationControlService,
+        B: Body + std::marker::Send + 'static,
+        B::Error: Into<StdError> + std::marker::Send + 'static,
+    {
+        type Response = http::Response<tonic::body::Body>;
+        type Error = std::convert::Infallible;
+        type Future = BoxFuture<Self::Response, Self::Error>;
+        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+        fn call(&mut self, req: http::Request<B>) -> Self::Future {
+            match req.uri().path() {
+                "/node_service.TierMutationControlService/PrepareTierMutation" => {
+                    #[allow(non_camel_case_types)]
+                    struct PrepareTierMutationSvc<T: TierMutationControlService>(pub Arc<T>);
+                    impl<T: TierMutationControlService> tonic::server::UnaryService<super::TierMutationPrepareRequest> for PrepareTierMutationSvc<T> {
+                        type Response = super::TierMutationControlResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::TierMutationPrepareRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut =
+                                async move { <T as TierMutationControlService>::prepare_tier_mutation(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = PrepareTierMutationSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                "/node_service.TierMutationControlService/CommitTierMutation" => {
+                    #[allow(non_camel_case_types)]
+                    struct CommitTierMutationSvc<T: TierMutationControlService>(pub Arc<T>);
+                    impl<T: TierMutationControlService> tonic::server::UnaryService<super::TierMutationCommitRequest> for CommitTierMutationSvc<T> {
+                        type Response = super::TierMutationControlResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::TierMutationCommitRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut =
+                                async move { <T as TierMutationControlService>::commit_tier_mutation(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = CommitTierMutationSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                "/node_service.TierMutationControlService/AbortTierMutation" => {
+                    #[allow(non_camel_case_types)]
+                    struct AbortTierMutationSvc<T: TierMutationControlService>(pub Arc<T>);
+                    impl<T: TierMutationControlService> tonic::server::UnaryService<super::TierMutationAbortRequest> for AbortTierMutationSvc<T> {
+                        type Response = super::TierMutationControlResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::TierMutationAbortRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut =
+                                async move { <T as TierMutationControlService>::abort_tier_mutation(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = AbortTierMutationSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                _ => Box::pin(async move {
+                    let mut response = http::Response::new(tonic::body::Body::default());
+                    let headers = response.headers_mut();
+                    headers.insert(tonic::Status::GRPC_STATUS, (tonic::Code::Unimplemented as i32).into());
+                    headers.insert(http::header::CONTENT_TYPE, tonic::metadata::GRPC_CONTENT_TYPE);
+                    Ok(response)
+                }),
+            }
+        }
+    }
+    impl<T> Clone for TierMutationControlServiceServer<T> {
+        fn clone(&self) -> Self {
+            let inner = self.inner.clone();
+            Self {
+                inner,
+                accept_compression_encodings: self.accept_compression_encodings,
+                send_compression_encodings: self.send_compression_encodings,
+                max_decoding_message_size: self.max_decoding_message_size,
+                max_encoding_message_size: self.max_encoding_message_size,
+            }
+        }
+    }
+    /// Generated gRPC service name
+    pub const SERVICE_NAME: &str = "node_service.TierMutationControlService";
+    impl<T> tonic::server::NamedService for TierMutationControlServiceServer<T> {
+        const NAME: &'static str = SERVICE_NAME;
+    }
+}

crates/protos/src/lib.rs

@@ -35,6 +35,7 @@ use tonic::{
     transport::{Certificate, Channel, ClientTlsConfig, Endpoint},
 };
 use tracing::{debug, info, warn};
+use uuid::Uuid;
 
 // Type alias for the complex client type
 pub type NodeServiceClientType = NodeServiceClient<
@@ -165,6 +166,9 @@ pub fn internode_rpc_max_message_size() -> usize {
 pub const HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE: usize = heal_control::RESULT_MAX_SIZE + 1024;
 pub const HEAL_CONTROL_PROTOCOL_VERSION: u32 = 2;
 pub const HEAL_CONTROL_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-heal-control-capability-v2\0";
+pub const TIER_MUTATION_RPC_MAX_PREPARE_PAYLOAD_SIZE: usize = 64 * 1024;
+pub const TIER_MUTATION_RPC_MAX_COMMIT_PAYLOAD_SIZE: usize = 1024;
+pub const TIER_MUTATION_RPC_MAX_MESSAGE_SIZE: usize = TIER_MUTATION_RPC_MAX_PREPARE_PAYLOAD_SIZE + 4096;
 
 pub fn heal_control_coordinator_epoch(topology_fingerprint: &str) -> Result<u64, &'static str> {
     let prefix = topology_fingerprint
@@ -251,6 +255,100 @@ pub fn canonical_heal_control_response_body(
     Ok(body)
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum TierMutationRpcPhase {
+    Prepare,
+    Commit,
+    Abort,
+}
+
+impl TierMutationRpcPhase {
+    fn as_wire_str(self) -> &'static str {
+        match self {
+            Self::Prepare => "prepare",
+            Self::Commit => "commit",
+            Self::Abort => "abort",
+        }
+    }
+}
+
+pub const TIER_MUTATION_RPC_PROTOCOL_VERSION: u32 = 1;
+
+pub fn canonical_tier_mutation_rpc_body(
+    version: u32,
+    phase: TierMutationRpcPhase,
+    mutation_id: Uuid,
+    canonical_payload: &[u8],
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    const DOMAIN: &[u8] = b"rustfs-tier-mutation-rpc-v1\0";
+
+    let phase = phase.as_wire_str().as_bytes();
+    let mutation_id = mutation_id.as_bytes();
+    let mut body = Vec::with_capacity(DOMAIN.len() + 4 + 8 + phase.len() + mutation_id.len() + 8 + canonical_payload.len());
+    body.extend_from_slice(DOMAIN);
+    body.extend_from_slice(&version.to_be_bytes());
+    body.extend_from_slice(&u64::try_from(phase.len())?.to_be_bytes());
+    body.extend_from_slice(phase);
+    body.extend_from_slice(mutation_id);
+    body.extend_from_slice(&u64::try_from(canonical_payload.len())?.to_be_bytes());
+    body.extend_from_slice(canonical_payload);
+    Ok(body)
+}
+
+pub struct TierMutationRpcResponseProofInput<'a> {
+    pub version: u32,
+    pub phase: TierMutationRpcPhase,
+    pub mutation_id: Uuid,
+    pub canonical_payload: &'a [u8],
+    pub success: bool,
+    pub state: i32,
+    pub applied: bool,
+    pub error_info: Option<&'a str>,
+}
+
+pub fn canonical_tier_mutation_rpc_response_body(
+    input: TierMutationRpcResponseProofInput<'_>,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    const DOMAIN: &[u8] = b"rustfs-tier-mutation-rpc-response-v1\0";
+
+    let phase = input.phase.as_wire_str().as_bytes();
+    let mutation_id = input.mutation_id.as_bytes();
+    let error_info = input.error_info.map(str::as_bytes);
+    let error_info_len = error_info.map_or(0, <[u8]>::len);
+    let mut body = Vec::with_capacity(
+        DOMAIN.len()
+            + 4
+            + 8
+            + phase.len()
+            + mutation_id.len()
+            + 8
+            + input.canonical_payload.len()
+            + 1
+            + 4
+            + 1
+            + 1
+            + 8
+            + error_info_len,
+    );
+    body.extend_from_slice(DOMAIN);
+    body.extend_from_slice(&input.version.to_be_bytes());
+    body.extend_from_slice(&u64::try_from(phase.len())?.to_be_bytes());
+    body.extend_from_slice(phase);
+    body.extend_from_slice(mutation_id);
+    body.extend_from_slice(&u64::try_from(input.canonical_payload.len())?.to_be_bytes());
+    body.extend_from_slice(input.canonical_payload);
+    body.push(u8::from(input.success));
+    body.extend_from_slice(&input.state.to_be_bytes());
+    body.push(u8::from(input.applied));
+    body.push(u8::from(error_info.is_some()));
+    body.extend_from_slice(&u64::try_from(error_info_len)?.to_be_bytes());
+    if let Some(error_info) = error_info {
+        body.extend_from_slice(error_info);
+    }
+    Ok(body)
+}
+
 #[cfg(test)]
 mod heal_control_tests {
     use super::{
@@ -363,6 +461,179 @@ mod heal_control_tests {
     }
 }
 
+#[cfg(test)]
+mod tier_mutation_rpc_tests {
+    use super::{
+        TIER_MUTATION_RPC_PROTOCOL_VERSION, TierMutationRpcPhase, TierMutationRpcResponseProofInput,
+        canonical_tier_mutation_rpc_body, canonical_tier_mutation_rpc_response_body,
+    };
+    use crate::proto_gen::node_service::TierMutationPeerState;
+    use uuid::uuid;
+
+    #[test]
+    fn canonical_tier_mutation_body_binds_phase_id_and_payload() {
+        let mutation_id = uuid!("12345678-1234-5678-9abc-def012345678");
+        let payload = b"canonical-intent-record";
+        let baseline = canonical_tier_mutation_rpc_body(
+            TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            TierMutationRpcPhase::Prepare,
+            mutation_id,
+            payload,
+        )
+        .expect("small mutation body should encode");
+        let mut golden = b"rustfs-tier-mutation-rpc-v1\0".to_vec();
+        golden.extend_from_slice(&1_u32.to_be_bytes());
+        golden.extend_from_slice(&7_u64.to_be_bytes());
+        golden.extend_from_slice(b"prepare");
+        golden.extend_from_slice(mutation_id.as_bytes());
+        golden.extend_from_slice(&u64::try_from(payload.len()).expect("payload length should fit").to_be_bytes());
+        golden.extend_from_slice(payload);
+        assert_eq!(baseline, golden);
+
+        assert_ne!(
+            baseline,
+            canonical_tier_mutation_rpc_body(2, TierMutationRpcPhase::Prepare, mutation_id, payload)
+                .expect("small mutation body should encode")
+        );
+        assert_ne!(
+            baseline,
+            canonical_tier_mutation_rpc_body(
+                TIER_MUTATION_RPC_PROTOCOL_VERSION,
+                TierMutationRpcPhase::Commit,
+                mutation_id,
+                payload,
+            )
+            .expect("small mutation body should encode")
+        );
+        assert_ne!(
+            baseline,
+            canonical_tier_mutation_rpc_body(
+                TIER_MUTATION_RPC_PROTOCOL_VERSION,
+                TierMutationRpcPhase::Prepare,
+                uuid!("22345678-1234-5678-9abc-def012345678"),
+                payload,
+            )
+            .expect("small mutation body should encode")
+        );
+        assert_ne!(
+            baseline,
+            canonical_tier_mutation_rpc_body(
+                TIER_MUTATION_RPC_PROTOCOL_VERSION,
+                TierMutationRpcPhase::Prepare,
+                mutation_id,
+                b"canonical-intent-record-tampered",
+            )
+            .expect("small mutation body should encode")
+        );
+    }
+
+    #[test]
+    fn canonical_tier_mutation_response_binds_request_state_and_error() {
+        let mutation_id = uuid!("12345678-1234-5678-9abc-def012345678");
+        let payload = b"canonical-intent-record";
+        let baseline = canonical_tier_mutation_rpc_response_body(TierMutationRpcResponseProofInput {
+            version: TIER_MUTATION_RPC_PROTOCOL_VERSION,
+            phase: TierMutationRpcPhase::Prepare,
+            mutation_id,
+            canonical_payload: payload,
+            success: true,
+            state: TierMutationPeerState::Prepared as i32,
+            applied: true,
+            error_info: None,
+        })
+        .expect("small mutation response should encode");
+
+        let cases = [
+            TierMutationRpcResponseProofInput {
+                version: 2,
+                phase: TierMutationRpcPhase::Prepare,
+                mutation_id,
+                canonical_payload: payload,
+                success: true,
+                state: TierMutationPeerState::Prepared as i32,
+                applied: true,
+                error_info: None,
+            },
+            TierMutationRpcResponseProofInput {
+                version: TIER_MUTATION_RPC_PROTOCOL_VERSION,
+                phase: TierMutationRpcPhase::Commit,
+                mutation_id,
+                canonical_payload: payload,
+                success: true,
+                state: TierMutationPeerState::Prepared as i32,
+                applied: true,
+                error_info: None,
+            },
+            TierMutationRpcResponseProofInput {
+                version: TIER_MUTATION_RPC_PROTOCOL_VERSION,
+                phase: TierMutationRpcPhase::Prepare,
+                mutation_id: uuid!("22345678-1234-5678-9abc-def012345678"),
+                canonical_payload: payload,
+                success: true,
+                state: TierMutationPeerState::Prepared as i32,
+                applied: true,
+                error_info: None,
+            },
+            TierMutationRpcResponseProofInput {
+                version: TIER_MUTATION_RPC_PROTOCOL_VERSION,
+                phase: TierMutationRpcPhase::Prepare,
+                mutation_id,
+                canonical_payload: b"tampered-intent-record",
+                success: true,
+                state: TierMutationPeerState::Prepared as i32,
+                applied: true,
+                error_info: None,
+            },
+            TierMutationRpcResponseProofInput {
+                version: TIER_MUTATION_RPC_PROTOCOL_VERSION,
+                phase: TierMutationRpcPhase::Prepare,
+                mutation_id,
+                canonical_payload: payload,
+                success: false,
+                state: TierMutationPeerState::Prepared as i32,
+                applied: true,
+                error_info: None,
+            },
+            TierMutationRpcResponseProofInput {
+                version: TIER_MUTATION_RPC_PROTOCOL_VERSION,
+                phase: TierMutationRpcPhase::Prepare,
+                mutation_id,
+                canonical_payload: payload,
+                success: true,
+                state: TierMutationPeerState::Committed as i32,
+                applied: true,
+                error_info: None,
+            },
+            TierMutationRpcResponseProofInput {
+                version: TIER_MUTATION_RPC_PROTOCOL_VERSION,
+                phase: TierMutationRpcPhase::Prepare,
+                mutation_id,
+                canonical_payload: payload,
+                success: true,
+                state: TierMutationPeerState::Prepared as i32,
+                applied: false,
+                error_info: None,
+            },
+            TierMutationRpcResponseProofInput {
+                version: TIER_MUTATION_RPC_PROTOCOL_VERSION,
+                phase: TierMutationRpcPhase::Prepare,
+                mutation_id,
+                canonical_payload: payload,
+                success: true,
+                state: TierMutationPeerState::Prepared as i32,
+                applied: true,
+                error_info: Some("error"),
+            },
+        ];
+        for case in cases {
+            assert_ne!(
+                baseline,
+                canonical_tier_mutation_rpc_response_body(case).expect("small mutation response should encode")
+            );
+        }
+    }
+}
+
 /// Whether internode metadata RPCs should send only the msgpack `_bin` payloads and leave the JSON
 /// compatibility strings empty (grpc-optimization P2-1). Shared by the client (`remote_disk`) and
 /// server (`node_service`) send paths. Defaults to `false` (dual-write); see

crates/protos/src/node.proto

@@ -865,6 +865,39 @@ message LoadTransitionTierConfigResponse {
   optional string error_info = 2;
 }
 
+message TierMutationPrepareRequest {
+  uint32 version = 1;
+  string mutation_id = 2;
+  bytes canonical_payload = 3;
+}
+
+message TierMutationCommitRequest {
+  uint32 version = 1;
+  string mutation_id = 2;
+  bytes canonical_payload = 3;
+}
+
+message TierMutationAbortRequest {
+  uint32 version = 1;
+  string mutation_id = 2;
+  bytes canonical_payload = 3;
+}
+
+enum TierMutationPeerState {
+  TIER_MUTATION_PEER_STATE_UNSPECIFIED = 0;
+  TIER_MUTATION_PEER_STATE_PREPARED = 1;
+  TIER_MUTATION_PEER_STATE_COMMITTED = 2;
+  TIER_MUTATION_PEER_STATE_ABORTED = 3;
+}
+
+message TierMutationControlResponse {
+  bool success = 1;
+  TierMutationPeerState state = 2;
+  bool applied = 3;
+  optional string error_info = 4;
+  bytes response_proof = 5;
+}
+
 message GetLiveEventsRequest {
   uint64 after_sequence = 1;
   uint32 limit = 2;
@@ -983,3 +1016,9 @@ service NodeService {
 service HealControlService {
   rpc HealControl(HealControlRequest) returns (HealControlResponse) {};
 }
+
+service TierMutationControlService {
+  rpc PrepareTierMutation(TierMutationPrepareRequest) returns (TierMutationControlResponse) {};
+  rpc CommitTierMutation(TierMutationCommitRequest) returns (TierMutationControlResponse) {};
+  rpc AbortTierMutation(TierMutationAbortRequest) returns (TierMutationControlResponse) {};
+}