fix: guard operator candidate delete by exact version

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

crates/ecstore/src/api/mod.rs

@@ -67,6 +67,14 @@ pub mod bucket {
             };
         }
 
+        pub mod transition_transaction {
+            pub use crate::bucket::lifecycle::transition_transaction::{
+                TransitionOperatorDeleteResult, TransitionOperatorError, TransitionOperatorProbe, TransitionOperatorStatus,
+                delete_transition_candidate_for_operator, finalize_missing_transition_transaction_for_operator,
+                inspect_transition_transaction_for_operator,
+            };
+        }
+
         pub mod evaluator {
             pub use crate::bucket::lifecycle::evaluator::Evaluator;
         }
@@ -377,6 +385,7 @@ pub mod metrics {
 pub mod notification {
     pub use crate::services::notification_sys::{
         NotificationPeerErr, NotificationSys, get_global_notification_sys, new_global_notification_sys,
+        start_remote_version_state_fleet_probe,
     };
 }
 

crates/ecstore/src/bucket/lifecycle/transition_transaction.rs

@@ -23,6 +23,7 @@ use uuid::Uuid;
 use crate::bucket::lifecycle::config_boundary;
 use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
 use crate::bucket::lifecycle::tier_sweeper::{
+    delete_confirmed_transition_candidate_exact_with_lease_idempotent,
     delete_confirmed_transition_candidate_exact_with_manager_and_identity,
     delete_object_from_remote_tier_idempotent_with_manager_and_identity,
 };
@@ -616,6 +617,199 @@ pub enum TransitionTransactionRecoveryOutcome {
     Retained,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum TransitionOperatorProbe {
+    Missing,
+    UnversionedPresent,
+    VersionedPresent(String),
+    Ambiguous,
+    Unsupported,
+}
+
+impl From<TransitionCandidateProbe> for TransitionOperatorProbe {
+    fn from(value: TransitionCandidateProbe) -> Self {
+        match value {
+            TransitionCandidateProbe::Missing => Self::Missing,
+            TransitionCandidateProbe::UnversionedPresent => Self::UnversionedPresent,
+            TransitionCandidateProbe::VersionedPresent(version_id) => Self::VersionedPresent(version_id),
+            TransitionCandidateProbe::Ambiguous => Self::Ambiguous,
+            TransitionCandidateProbe::Unsupported => Self::Unsupported,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct TransitionOperatorStatus {
+    pub transaction_id: Uuid,
+    pub state: TransitionTransactionState,
+    pub tier_name: String,
+    pub remote_object: String,
+    pub not_after_unix_nanos: i64,
+    pub probe: TransitionOperatorProbe,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct TransitionOperatorDeleteResult {
+    pub status: TransitionOperatorStatus,
+    pub journal_observed_after_delete: bool,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum TransitionOperatorError {
+    #[error("transition transaction was not found")]
+    NotFound,
+    #[error("transition transaction is still inside its active ownership window")]
+    NotExpired,
+    #[error("transition transaction state is not eligible for operator reconciliation: {0:?}")]
+    InvalidState(TransitionTransactionState),
+    #[error("an exact non-empty remote version is required")]
+    RemoteVersionRequired,
+    #[error("remote candidate is not proven missing: {0:?}")]
+    CandidateNotMissing(TransitionOperatorProbe),
+    #[error("remote candidate version does not match requested exact version: expected {expected}, observed {actual:?}")]
+    CandidateVersionMismatch {
+        expected: String,
+        actual: TransitionOperatorProbe,
+    },
+    #[error("transition transaction store failed: {0}")]
+    Store(#[source] Error),
+    #[error("remote tier reconciliation failed: {0}")]
+    Remote(#[source] std::io::Error),
+}
+
+type TransitionOperatorResult<T> = std::result::Result<T, TransitionOperatorError>;
+
+fn validate_operator_reconcile_transaction(
+    transaction: &TransitionTransaction,
+    now_unix_nanos: i128,
+) -> TransitionOperatorResult<()> {
+    transaction
+        .validate()
+        .map_err(|err| TransitionOperatorError::Store(Error::other(err)))?;
+    if transaction.state != TransitionTransactionState::UploadOutcomeUnknown {
+        return Err(TransitionOperatorError::InvalidState(transaction.state));
+    }
+    if now_unix_nanos < i128::from(transaction.not_after_unix_nanos) {
+        return Err(TransitionOperatorError::NotExpired);
+    }
+    Ok(())
+}
+
+async fn load_operator_reconcile_transaction(
+    api: Arc<ECStore>,
+    transaction_id: Uuid,
+) -> TransitionOperatorResult<TransitionTransaction> {
+    match load_transition_transaction_record(api, transaction_id).await {
+        Ok(transaction) => Ok(transaction),
+        Err(Error::ConfigNotFound) => Err(TransitionOperatorError::NotFound),
+        Err(err) => Err(TransitionOperatorError::Store(err)),
+    }
+}
+
+async fn operator_probe_transition_candidate(
+    api: Arc<ECStore>,
+    transaction: &TransitionTransaction,
+) -> TransitionOperatorResult<TransitionOperatorProbe> {
+    let lease = TierConfigMgr::acquire_operation_lease_for_backend_identity(
+        &api.tier_config_mgr(),
+        &transaction.tier_name,
+        transaction.backend_fingerprint,
+    )
+    .await
+    .map_err(|err| TransitionOperatorError::Remote(std::io::Error::other(err)))?;
+    lease
+        .probe_transition_candidate_for(&transaction.remote_object, transaction.transaction_id)
+        .await
+        .map(TransitionOperatorProbe::from)
+        .map_err(TransitionOperatorError::Remote)
+}
+
+pub async fn inspect_transition_transaction_for_operator(
+    api: Arc<ECStore>,
+    transaction_id: Uuid,
+) -> TransitionOperatorResult<TransitionOperatorStatus> {
+    let transaction = load_operator_reconcile_transaction(api.clone(), transaction_id).await?;
+    validate_operator_reconcile_transaction(&transaction, time::OffsetDateTime::now_utc().unix_timestamp_nanos())?;
+    let probe = operator_probe_transition_candidate(api, &transaction).await?;
+    Ok(TransitionOperatorStatus {
+        transaction_id,
+        state: transaction.state,
+        tier_name: transaction.tier_name,
+        remote_object: transaction.remote_object,
+        not_after_unix_nanos: transaction.not_after_unix_nanos,
+        probe,
+    })
+}
+
+pub async fn delete_transition_candidate_for_operator(
+    api: Arc<ECStore>,
+    transaction_id: Uuid,
+    remote_version_id: &str,
+) -> TransitionOperatorResult<TransitionOperatorDeleteResult> {
+    if remote_version_id.is_empty() {
+        return Err(TransitionOperatorError::RemoteVersionRequired);
+    }
+    let transaction = load_operator_reconcile_transaction(api.clone(), transaction_id).await?;
+    validate_operator_reconcile_transaction(&transaction, time::OffsetDateTime::now_utc().unix_timestamp_nanos())?;
+    let lease = TierConfigMgr::acquire_operation_lease_for_backend_identity(
+        &api.tier_config_mgr(),
+        &transaction.tier_name,
+        transaction.backend_fingerprint,
+    )
+    .await
+    .map_err(|err| TransitionOperatorError::Remote(std::io::Error::other(err)))?;
+    lease
+        .validate_remote_version_id(remote_version_id)
+        .map_err(TransitionOperatorError::Remote)?;
+    let before_delete_probe = lease
+        .probe_transition_candidate_for(&transaction.remote_object, transaction.transaction_id)
+        .await
+        .map(TransitionOperatorProbe::from)
+        .map_err(TransitionOperatorError::Remote)?;
+    if !matches!(&before_delete_probe, TransitionOperatorProbe::VersionedPresent(version_id) if version_id == remote_version_id) {
+        return Err(TransitionOperatorError::CandidateVersionMismatch {
+            expected: remote_version_id.to_string(),
+            actual: before_delete_probe,
+        });
+    }
+    delete_confirmed_transition_candidate_exact_with_lease_idempotent(&transaction.remote_object, remote_version_id, &lease)
+        .await
+        .map_err(TransitionOperatorError::Remote)?;
+    let probe = operator_probe_transition_candidate(api.clone(), &transaction).await?;
+    let journal_observed_after_delete = match load_transition_transaction_record(api, transaction_id).await {
+        Ok(_) => true,
+        Err(Error::ConfigNotFound) => false,
+        Err(err) => return Err(TransitionOperatorError::Store(err)),
+    };
+    Ok(TransitionOperatorDeleteResult {
+        status: TransitionOperatorStatus {
+            transaction_id,
+            state: transaction.state,
+            tier_name: transaction.tier_name,
+            remote_object: transaction.remote_object,
+            not_after_unix_nanos: transaction.not_after_unix_nanos,
+            probe,
+        },
+        journal_observed_after_delete,
+    })
+}
+
+pub async fn finalize_missing_transition_transaction_for_operator(
+    api: Arc<ECStore>,
+    transaction_id: Uuid,
+) -> TransitionOperatorResult<()> {
+    let transaction = load_operator_reconcile_transaction(api.clone(), transaction_id).await?;
+    validate_operator_reconcile_transaction(&transaction, time::OffsetDateTime::now_utc().unix_timestamp_nanos())?;
+    let probe = operator_probe_transition_candidate(api.clone(), &transaction).await?;
+    if probe != TransitionOperatorProbe::Missing {
+        return Err(TransitionOperatorError::CandidateNotMissing(probe));
+    }
+    delete_transition_transaction_record(api, transaction_id)
+        .await
+        .map_err(TransitionOperatorError::Store)
+}
+
 pub(crate) fn decode_transition_transaction_record(object: &str, data: &[u8]) -> Result<TransitionTransaction> {
     let transaction_id = transition_transaction_id_from_record_object_name(object)?;
     TransitionTransaction::decode(transaction_id, data)
@@ -700,7 +894,7 @@ async fn recover_unknown_upload_outcome(
     .map_err(Error::other)?;
 
     match lease
-        .probe_transition_candidate(&transaction.remote_object)
+        .probe_transition_candidate_for(&transaction.remote_object, transaction.transaction_id)
         .await
         .map_err(Error::other)?
     {
@@ -1097,6 +1291,27 @@ mod tests {
             .expect("upload state change should succeed")
     }
 
+    #[test]
+    fn operator_reconcile_requires_expired_unknown_upload_outcome() {
+        let mut transaction = new_transaction();
+        let active_deadline = transaction.not_after_unix_nanos;
+
+        assert!(matches!(
+            validate_operator_reconcile_transaction(&transaction, i128::from(active_deadline) + 1),
+            Err(TransitionOperatorError::InvalidState(TransitionTransactionState::UploadStarted))
+        ));
+
+        transaction
+            .advance(transaction.fence(), TransitionTransactionState::UploadOutcomeUnknown, None)
+            .expect("unknown upload outcome should be recorded");
+        assert!(matches!(
+            validate_operator_reconcile_transaction(&transaction, i128::from(active_deadline) - 1),
+            Err(TransitionOperatorError::NotExpired)
+        ));
+        validate_operator_reconcile_transaction(&transaction, i128::from(active_deadline))
+            .expect("expired unknown upload outcome should be eligible");
+    }
+
     fn cleanup_proof(transaction: &TransitionTransaction, decision: TransitionCleanupDecision) -> TransitionCleanupProof {
         TransitionCleanupProof {
             transaction_id: transaction.transaction_id,

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

@@ -222,6 +222,21 @@ fn validate_heal_control_response_proof(canonical_response: &[u8], proof: &[u8])
         .map_err(|_| Error::other("peer returned an invalid heal control response proof"))
 }
 
+fn decode_remote_version_state_capability(expected_member: &str, result: &[u8]) -> Result<Uuid> {
+    let (topology_member, process_epoch) = rustfs_protos::decode_remote_version_state_capability(result).map_err(Error::other)?;
+    if topology_member != expected_member {
+        return Err(Error::other(
+            "peer returned a remote version state capability for a different topology member",
+        ));
+    }
+    let server_epoch =
+        Uuid::from_slice(process_epoch).map_err(|_| Error::other("peer returned an invalid remote version state epoch"))?;
+    if server_epoch.is_nil() {
+        return Err(Error::other("peer returned a nil remote version state epoch"));
+    }
+    Ok(server_epoch)
+}
+
 #[derive(Clone, Debug)]
 pub struct PeerLiveEventsBatch {
     pub events: Vec<u8>,
@@ -233,6 +248,7 @@ pub struct PeerLiveEventsBatch {
 pub struct PeerRestClient {
     pub host: XHost,
     pub grid_host: String,
+    topology_member: String,
     offline: Arc<AtomicBool>,
     recovery_running: Arc<AtomicBool>,
 }
@@ -325,9 +341,11 @@ impl PeerRestClient {
     }
 
     pub fn new(host: XHost, grid_host: String) -> Self {
+        let topology_member = host.to_string();
         Self {
             host,
             grid_host,
+            topology_member,
             offline: Arc::new(AtomicBool::new(false)),
             recovery_running: Arc::new(AtomicBool::new(false)),
         }
@@ -347,7 +365,11 @@ impl PeerRestClient {
 
             let client = match grid_host {
                 Some(grid_host) => match XHost::try_from(peer_host_port.clone()) {
-                    Ok(host) => Some(PeerRestClient::new(host, grid_host)),
+                    Ok(host) => {
+                        let mut client = PeerRestClient::new(host, grid_host);
+                        client.topology_member = peer_host_port.clone();
+                        Some(client)
+                    }
                     Err(err) => {
                         warn!(peer = %peer_host_port, "Xhost parse failed while constructing peer client: {err:?}");
                         None
@@ -1154,6 +1176,15 @@ impl PeerRestClient {
         validate_heal_control_capability_proof(&canonical_ack, &proof)
     }
 
+    pub async fn probe_remote_version_state(&self, topology_fingerprint: String) -> Result<(String, Uuid)> {
+        let probe = rustfs_protos::remote_version_state_capability_probe(Uuid::new_v4().as_bytes());
+        let result = self
+            .heal_control(rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, topology_fingerprint, probe)
+            .await?;
+        let epoch = decode_remote_version_state_capability(&self.topology_member, &result)?;
+        Ok((self.topology_member.clone(), epoch))
+    }
+
     pub async fn load_bucket_metadata(&self, bucket: &str, scanner_maintenance_change: bool) -> Result<()> {
         self.finalize_result(
             async {
@@ -2470,6 +2501,22 @@ mod tests {
         }
     }
 
+    #[test]
+    fn remote_version_state_capability_decoder_fails_closed() {
+        let epoch = Uuid::new_v4();
+        let result = rustfs_protos::encode_remote_version_state_capability("node-a:9000", epoch.as_bytes())
+            .expect("small capability response should encode");
+        assert_eq!(
+            decode_remote_version_state_capability("node-a:9000", &result).expect("valid epoch should decode"),
+            epoch
+        );
+        assert!(decode_remote_version_state_capability("node-b:9000", &result).is_err());
+        assert!(decode_remote_version_state_capability("node-a:9000", &result[..result.len() - 1]).is_err());
+        let nil = rustfs_protos::encode_remote_version_state_capability("node-a:9000", Uuid::nil().as_bytes())
+            .expect("small capability response should encode");
+        assert!(decode_remote_version_state_capability("node-a:9000", &nil).is_err());
+    }
+
     struct TierMutationResponseFixture<'a> {
         version: u32,
         phase: TierMutationRpcPhase,

crates/ecstore/src/services/notification_sys.rs

@@ -29,11 +29,11 @@ use rustfs_madmin::metrics::RealtimeMetrics;
 use rustfs_madmin::net::NetInfo;
 use rustfs_madmin::{ItemState, ServerProperties, StorageInfo};
 use rustfs_utils::XHost;
-use std::collections::{HashMap, hash_map::DefaultHasher};
+use std::collections::{BTreeMap, HashMap, hash_map::DefaultHasher};
 use std::future::Future;
 use std::hash::{Hash, Hasher};
 use std::sync::{Arc, Mutex, OnceLock};
-use std::time::{Duration, SystemTime};
+use std::time::{Duration, Instant, SystemTime};
 use tokio::time::{sleep, timeout};
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, warn};
@@ -47,6 +47,9 @@ const EVENT_NOTIFICATION_PEER_PROPAGATION: &str = "notification_peer_propagation
 const SCANNER_ACTIVITY_PROBE_TIMEOUT: Duration = Duration::from_secs(5);
 const TIER_CONFIG_RELOAD_RETRY_BASE: Duration = Duration::from_millis(100);
 const TIER_CONFIG_RELOAD_RETRY_CAP: Duration = Duration::from_secs(5);
+const REMOTE_VERSION_STATE_PROBE_INTERVAL: Duration = Duration::from_secs(10);
+const REMOTE_VERSION_STATE_PROBE_TIMEOUT: Duration = Duration::from_secs(5);
+const REMOTE_VERSION_STATE_PROOF_TTL: Duration = Duration::from_secs(30);
 
 /// Cached result from the last successful admin call to a peer.
 struct PeerAdminCache {
@@ -91,6 +94,180 @@ lazy_static! {
     pub static ref GLOBAL_NOTIFICATION_SYS: OnceLock<Arc<NotificationSys>> = OnceLock::new();
 }
 
+#[derive(Clone)]
+struct RemoteVersionStateFleetProof {
+    topology_fingerprint: String,
+    peer_epochs: Arc<BTreeMap<String, Uuid>>,
+    expires_at: Instant,
+}
+
+impl RemoteVersionStateFleetProof {
+    fn token(&self) -> RemoteVersionStateFleetProofToken {
+        RemoteVersionStateFleetProofToken {
+            topology_fingerprint: self.topology_fingerprint.clone(),
+            peer_epochs: self.peer_epochs.clone(),
+        }
+    }
+}
+
+#[derive(Clone, PartialEq, Eq)]
+pub(crate) struct RemoteVersionStateFleetProofToken {
+    topology_fingerprint: String,
+    peer_epochs: Arc<BTreeMap<String, Uuid>>,
+}
+
+#[derive(Default)]
+struct RemoteVersionStateFleetProofState {
+    proof: Option<RemoteVersionStateFleetProof>,
+    topology_conflict: bool,
+}
+
+static REMOTE_VERSION_STATE_FLEET_PROOF: OnceLock<std::sync::RwLock<RemoteVersionStateFleetProofState>> = OnceLock::new();
+static REMOTE_VERSION_STATE_PROBE_TOPOLOGY: OnceLock<String> = OnceLock::new();
+
+fn remote_version_state_fleet_proof_slot() -> &'static std::sync::RwLock<RemoteVersionStateFleetProofState> {
+    REMOTE_VERSION_STATE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(RemoteVersionStateFleetProofState::default()))
+}
+
+fn replace_remote_version_state_fleet_proof(proof: Option<RemoteVersionStateFleetProof>) {
+    replace_remote_version_state_fleet_proof_in(remote_version_state_fleet_proof_slot(), proof);
+}
+
+fn replace_remote_version_state_fleet_proof_in(
+    slot: &std::sync::RwLock<RemoteVersionStateFleetProofState>,
+    proof: Option<RemoteVersionStateFleetProof>,
+) {
+    slot.write().unwrap_or_else(std::sync::PoisonError::into_inner).proof = proof;
+}
+
+fn publish_remote_version_state_probe_result(
+    slot: &std::sync::RwLock<RemoteVersionStateFleetProofState>,
+    topology_fingerprint: &str,
+    result: Result<BTreeMap<String, Uuid>>,
+    observed_at: Instant,
+) -> Option<Error> {
+    match result {
+        Ok(peer_epochs) => {
+            let mut state = slot.write().unwrap_or_else(std::sync::PoisonError::into_inner);
+            let peer_epochs = state
+                .proof
+                .as_ref()
+                .filter(|proof| proof.topology_fingerprint == topology_fingerprint && proof.peer_epochs.as_ref() == &peer_epochs)
+                .map(|proof| Arc::clone(&proof.peer_epochs))
+                .unwrap_or_else(|| Arc::new(peer_epochs));
+            state.proof = Some(RemoteVersionStateFleetProof {
+                topology_fingerprint: topology_fingerprint.to_string(),
+                peer_epochs,
+                expires_at: observed_at + REMOTE_VERSION_STATE_PROOF_TTL,
+            });
+            None
+        }
+        Err(err) => {
+            replace_remote_version_state_fleet_proof_in(slot, None);
+            Some(err)
+        }
+    }
+}
+
+pub(crate) fn acquire_remote_version_state_fleet_proof() -> Option<RemoteVersionStateFleetProofToken> {
+    let expected_topology = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get()?;
+    let state = remote_version_state_fleet_proof_slot()
+        .read()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    acquire_remote_version_state_fleet_proof_from(&state, expected_topology, Instant::now())
+}
+
+fn acquire_remote_version_state_fleet_proof_from(
+    state: &RemoteVersionStateFleetProofState,
+    expected_topology: &str,
+    now: Instant,
+) -> Option<RemoteVersionStateFleetProofToken> {
+    if state.topology_conflict || !remote_version_state_fleet_proof_valid_at(state.proof.as_ref(), expected_topology, now) {
+        return None;
+    }
+    state.proof.as_ref().map(RemoteVersionStateFleetProof::token)
+}
+
+pub(crate) fn remote_version_state_fleet_proof_matches(proof: &RemoteVersionStateFleetProofToken) -> bool {
+    let Some(expected_topology) = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get() else {
+        return false;
+    };
+    let state = remote_version_state_fleet_proof_slot()
+        .read()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    if state.topology_conflict {
+        return false;
+    }
+    state.proof.as_ref().is_some_and(|current| {
+        current.topology_fingerprint == *expected_topology
+            && current.topology_fingerprint == proof.topology_fingerprint
+            && Arc::ptr_eq(&current.peer_epochs, &proof.peer_epochs)
+            && Instant::now() < current.expires_at
+    })
+}
+
+fn remote_version_state_fleet_proof_valid_at(
+    proof: Option<&RemoteVersionStateFleetProof>,
+    expected_topology: &str,
+    now: Instant,
+) -> bool {
+    proof.is_some_and(|proof| proof.topology_fingerprint == expected_topology && now < proof.expires_at)
+}
+
+fn insert_remote_version_state_peer(peer_epochs: &mut BTreeMap<String, Uuid>, peer: String, epoch: Uuid) -> Result<()> {
+    if epoch.is_nil() || peer_epochs.values().any(|existing| *existing == epoch) || peer_epochs.insert(peer, epoch).is_some() {
+        return Err(Error::other("remote version state capability peer identity is invalid"));
+    }
+    Ok(())
+}
+
+pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
+    if REMOTE_VERSION_STATE_PROBE_TOPOLOGY.set(topology_fingerprint.clone()).is_err() {
+        if REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get() != Some(&topology_fingerprint) {
+            let mut state = remote_version_state_fleet_proof_slot()
+                .write()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            state.topology_conflict = true;
+            state.proof = None;
+        }
+        return;
+    }
+
+    tokio::spawn(async move {
+        loop {
+            let result = match get_global_notification_sys() {
+                Some(notification_sys) => {
+                    match timeout(
+                        REMOTE_VERSION_STATE_PROBE_TIMEOUT,
+                        notification_sys.probe_remote_version_state_fleet(&topology_fingerprint),
+                    )
+                    .await
+                    {
+                        Ok(result) => result,
+                        Err(_) => Err(Error::other("remote version state fleet capability probe timed out")),
+                    }
+                }
+                None => Err(Error::other("remote version state fleet capability notification system is unavailable")),
+            };
+            let topology_conflict = remote_version_state_fleet_proof_slot()
+                .read()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .topology_conflict;
+            if topology_conflict {
+                replace_remote_version_state_fleet_proof(None);
+            } else if let Some(err) = publish_remote_version_state_probe_result(
+                remote_version_state_fleet_proof_slot(),
+                &topology_fingerprint,
+                result,
+                Instant::now(),
+            ) {
+                debug!(error = %err, "remote version state fleet capability probe failed closed");
+            }
+            sleep(REMOTE_VERSION_STATE_PROBE_INTERVAL).await;
+        }
+    });
+}
+
 pub async fn new_global_notification_sys(eps: EndpointServerPools) -> Result<()> {
     let _ = GLOBAL_NOTIFICATION_SYS
         .set(Arc::new(NotificationSys::new(eps).await))
@@ -115,7 +292,17 @@ pub struct NotificationSys {
 
 impl NotificationSys {
     pub async fn new(eps: EndpointServerPools) -> Self {
+        let expected_remote_hosts = eps
+            .peer_grid_host_slots_sorted()
+            .into_iter()
+            .filter_map(|(peer, _, is_local)| (!is_local).then_some(peer))
+            .collect::<Vec<_>>();
         let (peer_clients, all_peer_clients, peer_topology_hosts) = PeerRestClient::new_clients_with_topology(eps).await;
+        let peer_topology_hosts = if peer_topology_hosts.is_empty() {
+            expected_remote_hosts
+        } else {
+            peer_topology_hosts
+        };
         let peer_admin_caches = (0..peer_clients.len()).map(|_| Mutex::new(PeerAdminCache::new())).collect();
         Self {
             peer_clients,
@@ -125,6 +312,24 @@ impl NotificationSys {
             tier_config_reload_workers: Default::default(),
         }
     }
+
+    async fn probe_remote_version_state_fleet(&self, topology_fingerprint: &str) -> Result<BTreeMap<String, Uuid>> {
+        if self.peer_clients.len() != self.peer_topology_hosts.len() {
+            return Err(Error::other("remote version state capability fleet membership is incomplete"));
+        }
+        let probes = self.peer_clients.iter().map(|client| async {
+            let client = client
+                .as_ref()
+                .ok_or_else(|| Error::other("remote version state capability peer is unreachable"))?;
+            client.probe_remote_version_state(topology_fingerprint.to_string()).await
+        });
+        let mut peer_epochs = BTreeMap::new();
+        for result in join_all(probes).await {
+            let (peer, epoch) = result?;
+            insert_remote_version_state_peer(&mut peer_epochs, peer, epoch)?;
+        }
+        Ok(peer_epochs)
+    }
 }
 
 pub struct NotificationPeerErr {
@@ -1890,6 +2095,183 @@ fn aggregate_scanner_dirty_usage_acknowledgement_results(
 mod tests {
     use super::*;
 
+    #[test]
+    fn remote_version_state_fleet_proof_rejects_stale_or_mismatched_membership() {
+        let now = Instant::now();
+        let mut peer_epochs = BTreeMap::new();
+        peer_epochs.insert("peer-a".to_string(), Uuid::new_v4());
+        let proof = RemoteVersionStateFleetProof {
+            topology_fingerprint: "topology-a".to_string(),
+            peer_epochs: Arc::new(peer_epochs),
+            expires_at: now + Duration::from_secs(1),
+        };
+
+        assert!(remote_version_state_fleet_proof_valid_at(Some(&proof), "topology-a", now));
+        assert!(!remote_version_state_fleet_proof_valid_at(Some(&proof), "topology-b", now));
+        assert!(!remote_version_state_fleet_proof_valid_at(Some(&proof), "topology-a", proof.expires_at));
+        assert!(!remote_version_state_fleet_proof_valid_at(None, "topology-a", now));
+    }
+
+    #[test]
+    fn remote_version_state_fleet_proof_rejects_nil_process_epoch() {
+        let mut peer_epochs = BTreeMap::new();
+
+        assert!(insert_remote_version_state_peer(&mut peer_epochs, "peer-a".to_string(), Uuid::nil()).is_err());
+        assert!(peer_epochs.is_empty());
+    }
+
+    #[test]
+    fn remote_version_state_fleet_proof_accepts_single_node_membership() {
+        let now = Instant::now();
+        let proof = RemoteVersionStateFleetProof {
+            topology_fingerprint: "topology-a".to_string(),
+            peer_epochs: Arc::new(BTreeMap::new()),
+            expires_at: now + Duration::from_secs(1),
+        };
+
+        assert!(remote_version_state_fleet_proof_valid_at(Some(&proof), "topology-a", now));
+    }
+
+    #[test]
+    fn remote_version_state_fleet_proof_token_changes_with_process_epoch() {
+        let now = Instant::now();
+        let proof = RemoteVersionStateFleetProof {
+            topology_fingerprint: "topology-a".to_string(),
+            peer_epochs: Arc::new(BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())])),
+            expires_at: now + Duration::from_secs(1),
+        };
+        let captured = proof.token();
+        let restarted = RemoteVersionStateFleetProof {
+            topology_fingerprint: proof.topology_fingerprint.clone(),
+            peer_epochs: Arc::new(BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())])),
+            expires_at: proof.expires_at,
+        };
+
+        assert!(captured != restarted.token());
+    }
+
+    #[test]
+    fn remote_version_state_fleet_proof_renewal_preserves_only_same_epoch_token() {
+        let slot = std::sync::RwLock::new(RemoteVersionStateFleetProofState::default());
+        let now = Instant::now();
+        let epoch = Uuid::new_v4();
+        let peers = BTreeMap::from([("peer-a".to_string(), epoch)]);
+        assert!(publish_remote_version_state_probe_result(&slot, "topology-a", Ok(peers.clone()), now).is_none());
+        let original = slot
+            .read()
+            .expect("proof slot should not poison")
+            .proof
+            .as_ref()
+            .expect("successful probe should publish proof")
+            .token();
+
+        assert!(
+            publish_remote_version_state_probe_result(&slot, "topology-a", Ok(peers), now + Duration::from_millis(1)).is_none()
+        );
+        let renewed = slot
+            .read()
+            .expect("proof slot should not poison")
+            .proof
+            .as_ref()
+            .expect("renewal should retain proof")
+            .token();
+        assert!(Arc::ptr_eq(&original.peer_epochs, &renewed.peer_epochs));
+
+        let restarted = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+        assert!(
+            publish_remote_version_state_probe_result(&slot, "topology-a", Ok(restarted), now + Duration::from_millis(2))
+                .is_none()
+        );
+        let replaced = slot
+            .read()
+            .expect("proof slot should not poison")
+            .proof
+            .as_ref()
+            .expect("restarted peer should publish a new proof")
+            .token();
+        assert!(!Arc::ptr_eq(&original.peer_epochs, &replaced.peer_epochs));
+    }
+
+    #[test]
+    fn remote_version_state_fleet_proof_conflict_revokes_atomic_snapshot() {
+        let now = Instant::now();
+        let mut state = RemoteVersionStateFleetProofState {
+            proof: Some(RemoteVersionStateFleetProof {
+                topology_fingerprint: "topology-a".to_string(),
+                peer_epochs: Arc::new(BTreeMap::new()),
+                expires_at: now + Duration::from_secs(1),
+            }),
+            topology_conflict: false,
+        };
+        assert!(acquire_remote_version_state_fleet_proof_from(&state, "topology-a", now).is_some());
+
+        state.topology_conflict = true;
+        assert!(acquire_remote_version_state_fleet_proof_from(&state, "topology-a", now).is_none());
+    }
+
+    #[test]
+    fn remote_version_state_fleet_probe_rejects_duplicate_member_or_process_epoch() {
+        let epoch = Uuid::new_v4();
+        let mut peer_epochs = BTreeMap::new();
+        insert_remote_version_state_peer(&mut peer_epochs, "node-a:9000".to_string(), epoch)
+            .expect("first member should be admitted");
+        assert!(insert_remote_version_state_peer(&mut peer_epochs, "node-b:9000".to_string(), epoch).is_err());
+        assert!(insert_remote_version_state_peer(&mut peer_epochs, "node-a:9000".to_string(), Uuid::new_v4()).is_err());
+        assert!(insert_remote_version_state_peer(&mut peer_epochs, "node-c:9000".to_string(), Uuid::nil()).is_err());
+    }
+
+    #[test]
+    fn remote_version_state_fleet_probe_failure_revokes_previous_proof() {
+        let slot = std::sync::RwLock::new(RemoteVersionStateFleetProofState::default());
+        let now = Instant::now();
+        let peer_epochs = BTreeMap::from([("node-a:9000".to_string(), Uuid::new_v4())]);
+        assert!(publish_remote_version_state_probe_result(&slot, "topology-a", Ok(peer_epochs), now).is_none());
+        assert!(slot.read().expect("proof slot should not poison").proof.is_some());
+
+        assert!(
+            publish_remote_version_state_probe_result(&slot, "topology-a", Err(Error::other("peer unavailable")), now,).is_some()
+        );
+        assert!(slot.read().expect("proof slot should not poison").proof.is_none());
+
+        let peer_epochs = BTreeMap::from([("node-a:9000".to_string(), Uuid::new_v4())]);
+        assert!(publish_remote_version_state_probe_result(&slot, "topology-a", Ok(peer_epochs), now).is_none());
+        assert!(slot.read().expect("proof slot should not poison").proof.is_some());
+    }
+
+    #[tokio::test]
+    async fn remote_version_state_fleet_probe_rejects_unreachable_member() {
+        let notification_sys = NotificationSys {
+            peer_clients: vec![None],
+            all_peer_clients: vec![None, None],
+            peer_topology_hosts: vec!["peer-a".to_string()],
+            peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())],
+            tier_config_reload_workers: Default::default(),
+        };
+
+        let err = notification_sys
+            .probe_remote_version_state_fleet("topology-a")
+            .await
+            .expect_err("an unreachable configured member must fail the fleet proof");
+        assert!(err.to_string().contains("unreachable"));
+    }
+
+    #[tokio::test]
+    async fn remote_version_state_fleet_probe_rejects_missing_member_slot() {
+        let notification_sys = NotificationSys {
+            peer_clients: Vec::new(),
+            all_peer_clients: vec![None],
+            peer_topology_hosts: vec!["peer-a".to_string()],
+            peer_admin_caches: Vec::new(),
+            tier_config_reload_workers: Default::default(),
+        };
+
+        let err = notification_sys
+            .probe_remote_version_state_fleet("topology-a")
+            .await
+            .expect_err("a missing configured member slot must fail the fleet proof");
+        assert!(err.to_string().contains("incomplete"));
+    }
+
     fn build_props(endpoint: &str) -> ServerProperties {
         ServerProperties {
             endpoint: endpoint.to_string(),

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

@@ -236,6 +236,7 @@ struct TierPublishTransition {
 
 struct PreparedTierDriver {
     tier_name: String,
+    tier_config: TierConfig,
     config_fingerprint: TierDriverFingerprint,
     backend_identity: TierDestinationId,
     exact_get_delete: bool,
@@ -1342,6 +1343,7 @@ fn tier_exact_get_delete(config: &TierConfig) -> bool {
 
 struct TierDriverGeneration {
     tier_name: Arc<str>,
+    tier_config: TierConfig,
     generation: DriverRevision,
     // Process-local only: this may reflect credential changes and must never be persisted or logged.
     config_fingerprint: TierDriverFingerprint,
@@ -1349,6 +1351,9 @@ struct TierDriverGeneration {
     backend_identity: TierDestinationId,
     exact_get_delete: bool,
     driver: SharedWarmBackend,
+    reconciler: tokio::sync::OnceCell<
+        Option<Arc<dyn crate::services::tier::warm_backend::TransitionCandidateReconciler + Send + Sync + 'static>>,
+    >,
     accepting: AtomicBool,
     active_leases: AtomicUsize,
     drained: tokio::sync::Notify,
@@ -1473,6 +1478,35 @@ impl TierOperationLease {
         self.inner.backend_identity
     }
 
+    pub(crate) async fn probe_transition_candidate_for(
+        &self,
+        object: &str,
+        transaction_id: uuid::Uuid,
+    ) -> io::Result<TransitionCandidateProbe> {
+        let Some(reconciler) = self
+            .inner
+            .reconciler
+            .get_or_try_init(|| async {
+                crate::services::tier::warm_backend::new_transition_candidate_reconciler(&self.inner.tier_config)
+                    .await
+                    .map(|reconciler| reconciler.map(Arc::from))
+            })
+            .await
+            .map_err(|err| io::Error::other(err.message))?
+        else {
+            return self.inner.driver.probe_transition_candidate(object).await;
+        };
+        reconciler
+            .probe_transition_candidate_for(
+                object,
+                crate::services::tier::warm_backend::TransitionCandidateIdentity {
+                    transaction_id,
+                    destination_id: self.backend_identity(),
+                },
+            )
+            .await
+    }
+
     pub(crate) fn validate_remote_version_id(&self, remote_version_id: &str) -> io::Result<()> {
         self.inner.driver.validate_remote_version_id(remote_version_id)?;
         if !remote_version_id.is_empty() && !self.inner.exact_get_delete {
@@ -2833,6 +2867,7 @@ impl TierConfigMgr {
                     let exact_get_delete = tier_exact_get_delete(config);
                     Some(PreparedTierDriver {
                         tier_name: tier_name.to_string(),
+                        tier_config: config.clone(),
                         config_fingerprint,
                         backend_identity,
                         exact_get_delete,
@@ -2861,11 +2896,13 @@ impl TierConfigMgr {
             })?;
             let entry = Arc::new(TierDriverGeneration {
                 tier_name: Arc::from(prepared.tier_name.as_str()),
+                tier_config: prepared.tier_config.clone(),
                 generation,
                 config_fingerprint: prepared.config_fingerprint,
                 backend_identity: prepared.backend_identity,
                 exact_get_delete: prepared.exact_get_delete,
                 driver: prepared.driver.clone(),
+                reconciler: tokio::sync::OnceCell::new(),
                 accepting: AtomicBool::new(true),
                 active_leases: AtomicUsize::new(0),
                 drained: tokio::sync::Notify::new(),
@@ -3609,11 +3646,13 @@ impl TierConfigMgr {
         let driver: SharedWarmBackend = Arc::from(driver);
         let entry = Arc::new(TierDriverGeneration {
             tier_name: Arc::from(tier_name),
+            tier_config: config.clone(),
             generation,
             config_fingerprint,
             backend_identity,
             exact_get_delete,
             driver: driver.clone(),
+            reconciler: tokio::sync::OnceCell::new(),
             accepting: AtomicBool::new(true),
             active_leases: AtomicUsize::new(0),
             drained: tokio::sync::Notify::new(),

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

@@ -73,6 +73,21 @@ pub enum TransitionCandidateProbe {
     Unsupported,
 }
 
+#[derive(Clone, Copy)]
+pub(crate) struct TransitionCandidateIdentity {
+    pub transaction_id: uuid::Uuid,
+    pub destination_id: [u8; 32],
+}
+
+#[async_trait::async_trait]
+pub(crate) trait TransitionCandidateReconciler {
+    async fn probe_transition_candidate_for(
+        &self,
+        object: &str,
+        identity: TransitionCandidateIdentity,
+    ) -> Result<TransitionCandidateProbe, std::io::Error>;
+}
+
 #[async_trait::async_trait]
 pub trait WarmBackend {
     async fn validate(&self) -> Result<(), std::io::Error> {
@@ -189,6 +204,20 @@ pub fn build_transition_put_options(storage_class: String, mut metadata: HashMap
         metadata.remove(key);
     }
 
+    for suffix in [
+        rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TRANSACTION_ID,
+        rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID,
+    ] {
+        for key in [
+            rustfs_utils::http::metadata_compat::internal_key_rustfs(suffix),
+            format!("{}{}", rustfs_utils::http::metadata_compat::MINIO_INTERNAL_PREFIX, suffix),
+        ] {
+            if let Some(value) = metadata.remove(&key) {
+                metadata.insert(format!("x-amz-meta-{key}"), value);
+            }
+        }
+    }
+
     opts.user_metadata = metadata;
     opts
 }
@@ -446,6 +475,51 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBack
     Ok(d)
 }
 
+pub(crate) async fn new_transition_candidate_reconciler(
+    tier: &TierConfig,
+) -> Result<Option<Box<dyn TransitionCandidateReconciler + Send + Sync + 'static>>, AdminError> {
+    let reconciler: Box<dyn TransitionCandidateReconciler + Send + Sync + 'static> = match tier.tier_type {
+        TierType::S3 => Box::new(
+            WarmBackendS3::new(tier.s3.as_ref().ok_or_else(|| ERR_TIER_INVALID_CONFIG.clone())?, &tier.name)
+                .await
+                .map_err(|err| {
+                    let mut admin_err = ERR_TIER_INVALID_CONFIG.clone();
+                    admin_err.message = err.to_string();
+                    admin_err
+                })?,
+        ),
+        TierType::MinIO => Box::new(
+            WarmBackendMinIO::new(tier.minio.as_ref().ok_or_else(|| ERR_TIER_INVALID_CONFIG.clone())?, &tier.name)
+                .await
+                .map_err(|err| {
+                    let mut admin_err = ERR_TIER_INVALID_CONFIG.clone();
+                    admin_err.message = err.to_string();
+                    admin_err
+                })?,
+        ),
+        TierType::RustFS => Box::new(
+            WarmBackendRustFS::new(tier.rustfs.as_ref().ok_or_else(|| ERR_TIER_INVALID_CONFIG.clone())?, &tier.name)
+                .await
+                .map_err(|err| {
+                    let mut admin_err = ERR_TIER_INVALID_CONFIG.clone();
+                    admin_err.message = err.to_string();
+                    admin_err
+                })?,
+        ),
+        TierType::R2 => Box::new(
+            WarmBackendR2::new(tier.r2.as_ref().ok_or_else(|| ERR_TIER_INVALID_CONFIG.clone())?, &tier.name)
+                .await
+                .map_err(|err| {
+                    let mut admin_err = ERR_TIER_INVALID_CONFIG.clone();
+                    admin_err.message = err.to_string();
+                    admin_err
+                })?,
+        ),
+        _ => return Ok(None),
+    };
+    Ok(Some(reconciler))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -775,6 +849,37 @@ mod tests {
         assert!(!opts.user_metadata.contains_key(X_AMZ_REPLICATION_STATUS.as_str()));
     }
 
+    #[test]
+    fn build_transition_put_options_persists_both_candidate_identity_keys_as_s3_metadata() {
+        let mut metadata = HashMap::new();
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut metadata,
+            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TRANSACTION_ID,
+            "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa".to_string(),
+        );
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut metadata,
+            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID,
+            "5a".repeat(32),
+        );
+
+        let opts = build_transition_put_options("COLD".to_string(), metadata);
+
+        for suffix in [
+            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TRANSACTION_ID,
+            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID,
+        ] {
+            assert!(opts.user_metadata.contains_key(&format!(
+                "x-amz-meta-{}",
+                rustfs_utils::http::metadata_compat::internal_key_rustfs(suffix)
+            )));
+            assert!(opts.user_metadata.contains_key(&format!(
+                "x-amz-meta-{}{suffix}",
+                rustfs_utils::http::metadata_compat::MINIO_INTERNAL_PREFIX
+            )));
+        }
+    }
+
     #[test]
     fn build_transition_put_options_requests_no_checksum_and_content_md5() {
         // Regression for rustfs/rustfs#4811: transition uploads must leave the

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

@@ -135,6 +135,20 @@ impl WarmBackend for WarmBackendMinIO {
     }
 }
 
+#[async_trait::async_trait]
+impl crate::services::tier::warm_backend::TransitionCandidateReconciler for WarmBackendMinIO {
+    async fn probe_transition_candidate_for(
+        &self,
+        object: &str,
+        identity: crate::services::tier::warm_backend::TransitionCandidateIdentity,
+    ) -> Result<TransitionCandidateProbe, std::io::Error> {
+        crate::services::tier::warm_backend::TransitionCandidateReconciler::probe_transition_candidate_for(
+            &self.0, object, identity,
+        )
+        .await
+    }
+}
+
 fn optimal_part_size(object_size: i64) -> Result<i64, std::io::Error> {
     let mut object_size = object_size;
     if object_size == -1 {

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

@@ -135,6 +135,20 @@ impl WarmBackend for WarmBackendR2 {
     }
 }
 
+#[async_trait::async_trait]
+impl crate::services::tier::warm_backend::TransitionCandidateReconciler for WarmBackendR2 {
+    async fn probe_transition_candidate_for(
+        &self,
+        object: &str,
+        identity: crate::services::tier::warm_backend::TransitionCandidateIdentity,
+    ) -> Result<TransitionCandidateProbe, std::io::Error> {
+        crate::services::tier::warm_backend::TransitionCandidateReconciler::probe_transition_candidate_for(
+            &self.0, object, identity,
+        )
+        .await
+    }
+}
+
 fn optimal_part_size(object_size: i64) -> Result<i64, std::io::Error> {
     let mut object_size = object_size;
     if object_size == -1 {

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

@@ -132,6 +132,20 @@ impl WarmBackend for WarmBackendRustFS {
     }
 }
 
+#[async_trait::async_trait]
+impl crate::services::tier::warm_backend::TransitionCandidateReconciler for WarmBackendRustFS {
+    async fn probe_transition_candidate_for(
+        &self,
+        object: &str,
+        identity: crate::services::tier::warm_backend::TransitionCandidateIdentity,
+    ) -> Result<TransitionCandidateProbe, std::io::Error> {
+        crate::services::tier::warm_backend::TransitionCandidateReconciler::probe_transition_candidate_for(
+            &self.0, object, identity,
+        )
+        .await
+    }
+}
+
 fn optimal_part_size(object_size: i64) -> Result<i64, std::io::Error> {
     let mut object_size = object_size;
     if object_size == -1 {

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

@@ -37,7 +37,10 @@ use crate::error::ErrorResponse;
 use crate::error::error_resp_to_object_err;
 use crate::services::tier::{
     tier_config::TierS3,
-    warm_backend::{TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options},
+    warm_backend::{
+        TransitionCandidateIdentity, TransitionCandidateProbe, TransitionCandidateReconciler, WarmBackend, WarmBackendGetOpts,
+        build_transition_put_options,
+    },
 };
 use http::HeaderMap;
 use rustfs_utils::egress::validate_outbound_url;
@@ -219,6 +222,92 @@ impl WarmBackendS3 {
             advance_version_markers(&mut key_marker, &mut version_id_marker, &versions)?;
         }
     }
+
+    async fn probe_transition_candidate_identity(
+        &self,
+        object: &str,
+        identity: TransitionCandidateIdentity,
+        bucket_versioning: RemoteBucketVersioning,
+    ) -> Result<TransitionCandidateProbe, std::io::Error> {
+        let remote_object = self.get_dest(object);
+        let mut opts = ListObjectsOptions::default();
+        opts.set("prefix", &remote_object);
+        opts.set("max-keys", "1000");
+        let mut key_marker = String::new();
+        let mut version_id_marker = String::new();
+        let mut matched_version = None;
+        let mut saw_unproven_candidate = false;
+
+        loop {
+            let versions = self
+                .client
+                .list_object_versions_query(&self.bucket, &opts, &key_marker, &version_id_marker, "")
+                .await?;
+            for version in versions.versions.iter().filter(|version| version.key == remote_object) {
+                let mut stat_opts = GetObjectOptions::default();
+                stat_opts.version_id.clone_from(&version.version_id);
+                let info = self.client.stat_object(&self.bucket, &remote_object, &stat_opts).await?;
+                let mut metadata = info.user_metadata;
+                for (name, value) in &info.metadata {
+                    if (name
+                        .as_str()
+                        .starts_with(rustfs_utils::http::metadata_compat::RUSTFS_INTERNAL_PREFIX)
+                        || name
+                            .as_str()
+                            .starts_with(rustfs_utils::http::metadata_compat::MINIO_INTERNAL_PREFIX))
+                        && let Ok(value) = value.to_str()
+                    {
+                        metadata.insert(name.as_str().to_string(), value.to_string());
+                    }
+                }
+                if transition_candidate_metadata_matches(&metadata, identity)? {
+                    if matched_version.is_some() {
+                        return Ok(TransitionCandidateProbe::Ambiguous);
+                    }
+                    matched_version = Some(version.version_id.clone());
+                } else {
+                    saw_unproven_candidate = true;
+                }
+            }
+            if !versions.is_truncated {
+                if matched_version.is_none() && saw_unproven_candidate {
+                    return Ok(TransitionCandidateProbe::Unsupported);
+                }
+                let candidates = TransitionCandidateVersions {
+                    version_id: matched_version,
+                    ambiguous: false,
+                };
+                return classify_transition_candidates(candidates, bucket_versioning);
+            }
+            advance_version_markers(&mut key_marker, &mut version_id_marker, &versions)?;
+        }
+    }
+}
+
+fn transition_candidate_metadata_matches(
+    metadata: &HashMap<String, String>,
+    identity: TransitionCandidateIdentity,
+) -> Result<bool, std::io::Error> {
+    use rustfs_utils::http::metadata_compat::{
+        SUFFIX_TRANSITION_TIER_DESTINATION_ID, SUFFIX_TRANSITION_TRANSACTION_ID, contains_key_str, get_consistent_str,
+    };
+
+    let transaction_id = get_consistent_str(metadata, SUFFIX_TRANSITION_TRANSACTION_ID);
+    let destination_id = get_consistent_str(metadata, SUFFIX_TRANSITION_TIER_DESTINATION_ID);
+    if transaction_id.is_none() || destination_id.is_none() {
+        if contains_key_str(metadata, SUFFIX_TRANSITION_TRANSACTION_ID)
+            || contains_key_str(metadata, SUFFIX_TRANSITION_TIER_DESTINATION_ID)
+        {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                "transition candidate identity metadata is empty or conflicting",
+            ));
+        }
+        return Ok(false);
+    }
+    let expected_transaction_id = identity.transaction_id.to_string();
+    let expected_destination_id = rustfs_utils::crypto::hex(identity.destination_id);
+    Ok(transaction_id == Some(expected_transaction_id.as_str()) && destination_id == Some(expected_destination_id.as_str()))
 }
 
 fn classify_transition_candidates(
@@ -342,6 +431,60 @@ mod tests {
         candidates.classify(bucket_versioning)
     }
 
+    fn candidate_identity() -> TransitionCandidateIdentity {
+        TransitionCandidateIdentity {
+            transaction_id: uuid::Uuid::parse_str("aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa").unwrap(),
+            destination_id: [0x5a; 32],
+        }
+    }
+
+    fn candidate_metadata(identity: TransitionCandidateIdentity) -> HashMap<String, String> {
+        let mut metadata = HashMap::new();
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut metadata,
+            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TRANSACTION_ID,
+            identity.transaction_id.to_string(),
+        );
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut metadata,
+            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID,
+            rustfs_utils::crypto::hex(identity.destination_id),
+        );
+        metadata
+    }
+
+    #[test]
+    fn transition_candidate_identity_requires_exact_compatible_metadata() {
+        let identity = candidate_identity();
+        let metadata = candidate_metadata(identity);
+        assert!(transition_candidate_metadata_matches(&metadata, identity).unwrap());
+
+        let mut adjacent = metadata;
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut adjacent,
+            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TRANSACTION_ID,
+            uuid::Uuid::parse_str("bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb")
+                .unwrap()
+                .to_string(),
+        );
+        assert!(!transition_candidate_metadata_matches(&adjacent, identity).unwrap());
+
+        assert!(!transition_candidate_metadata_matches(&HashMap::new(), identity).unwrap());
+    }
+
+    #[test]
+    fn transition_candidate_identity_rejects_conflicting_compatibility_keys() {
+        let identity = candidate_identity();
+        let mut metadata = candidate_metadata(identity);
+        metadata.insert(
+            rustfs_utils::http::metadata_compat::internal_key_rustfs(
+                rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TRANSACTION_ID,
+            ),
+            uuid::Uuid::new_v4().to_string(),
+        );
+        assert!(transition_candidate_metadata_matches(&metadata, identity).is_err());
+    }
+
     #[test]
     fn transition_candidate_probe_classifier_is_fail_closed() {
         assert_eq!(
@@ -503,3 +646,16 @@ impl WarmBackend for WarmBackendS3 {
         Ok(result.common_prefixes.len() > 0 || result.contents.len() > 0)
     }
 }
+
+#[async_trait::async_trait]
+impl TransitionCandidateReconciler for WarmBackendS3 {
+    async fn probe_transition_candidate_for(
+        &self,
+        object: &str,
+        identity: TransitionCandidateIdentity,
+    ) -> Result<TransitionCandidateProbe, std::io::Error> {
+        let bucket_versioning = self.remote_bucket_versioning().await?;
+        self.probe_transition_candidate_identity(object, identity, bucket_versioning)
+            .await
+    }
+}

crates/ecstore/src/set_disk/ops/object.rs

@@ -2320,13 +2320,25 @@ async fn pause_transition_commit(bucket: &str, object: &str, pause: TransitionCo
     }
 }
 
+#[cfg(test)]
 fn persisted_transition_version(
     remote_version: &str,
 ) -> std::io::Result<(Option<String>, rustfs_filemeta::TransitionVersionState)> {
     persisted_transition_version_with_gate(remote_version, remote_version_state_writer_enabled())
 }
 
+#[cfg(test)]
 fn remote_version_state_writer_enabled() -> bool {
+    remote_version_state_writer_fleet_proof().is_some()
+}
+
+fn remote_version_state_writer_fleet_proof() -> Option<crate::services::notification_sys::RemoteVersionStateFleetProofToken> {
+    remote_version_state_writer_requested()
+        .then(crate::services::notification_sys::acquire_remote_version_state_fleet_proof)
+        .flatten()
+}
+
+fn remote_version_state_writer_requested() -> bool {
     remote_version_state_writer_enabled_for(
         rustfs_utils::get_env_bool(
             rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_WRITE,
@@ -2336,11 +2348,25 @@ fn remote_version_state_writer_enabled() -> bool {
             rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED,
             rustfs_config::DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED,
         ),
+        true,
     )
 }
 
-fn remote_version_state_writer_enabled_for(requested: bool, fleet_confirmed: bool) -> bool {
-    requested && fleet_confirmed
+fn remote_version_state_writer_fleet_proof_matches(
+    proof: &crate::services::notification_sys::RemoteVersionStateFleetProofToken,
+) -> bool {
+    remote_version_state_writer_fleet_proof_matches_for(
+        remote_version_state_writer_requested(),
+        crate::services::notification_sys::remote_version_state_fleet_proof_matches(proof),
+    )
+}
+
+fn remote_version_state_writer_fleet_proof_matches_for(requested: bool, fleet_proof_matches: bool) -> bool {
+    requested && fleet_proof_matches
+}
+
+fn remote_version_state_writer_enabled_for(requested: bool, fleet_confirmed: bool, fleet_proof_valid: bool) -> bool {
+    requested && fleet_confirmed && fleet_proof_valid
 }
 
 fn persisted_transition_version_with_gate(
@@ -2359,7 +2385,7 @@ fn persisted_transition_version_with_gate(
         Ok(_) => Ok((Some(remote_version.to_string()), rustfs_filemeta::TransitionVersionState::Exact)),
         Err(_) if !remote_version_state_writer_enabled => Err(std::io::Error::new(
             std::io::ErrorKind::Unsupported,
-            "opaque remote tier versions require the operator-attested fleet gate",
+            "opaque remote tier versions require the operator-attested live fleet capability gate",
         )),
         Err(_) if remote_version == "null" => {
             Ok((Some(remote_version.to_string()), rustfs_filemeta::TransitionVersionState::SuspendedNull))
@@ -2605,7 +2631,7 @@ mod transition_upload_completion_tests {
 mod transition_version_id_tests {
     use super::{
         TransitionUploadCandidate, persisted_transition_version, persisted_transition_version_with_gate,
-        remote_version_state_writer_enabled_for,
+        remote_version_state_writer_enabled_for, remote_version_state_writer_fleet_proof_matches_for,
     };
     use rustfs_filemeta::TransitionVersionState;
     use uuid::Uuid;
@@ -2648,15 +2674,35 @@ mod transition_version_id_tests {
 
     #[test]
     fn remote_version_state_writer_requires_request_and_fleet_confirmation() {
-        for (case, requested, fleet_confirmed, expected) in [
-            ("old defaults", false, false, false),
-            ("missing fleet confirmation", true, false, false),
-            ("missing local opt-in", false, true, false),
-            ("explicitly unconfirmed fleet", true, false, false),
-            ("rolled-back writer", false, true, false),
-            ("fully upgraded fleet", true, true, true),
+        for (case, requested, fleet_confirmed, fleet_proof_valid, expected) in [
+            ("old defaults", false, false, false, false),
+            ("missing fleet confirmation", true, false, true, false),
+            ("missing local opt-in", false, true, true, false),
+            ("missing fleet proof", true, true, false, false),
+            ("explicitly unconfirmed fleet", true, false, true, false),
+            ("rolled-back writer", false, true, true, false),
+            ("fully upgraded fleet", true, true, true, true),
         ] {
-            assert_eq!(remote_version_state_writer_enabled_for(requested, fleet_confirmed), expected, "{case}");
+            assert_eq!(
+                remote_version_state_writer_enabled_for(requested, fleet_confirmed, fleet_proof_valid),
+                expected,
+                "{case}"
+            );
+        }
+    }
+
+    #[test]
+    fn remote_version_state_commit_rechecks_operator_gate_and_live_proof() {
+        for (case, requested, fleet_proof_matches, expected) in [
+            ("operator gate closed", false, true, false),
+            ("fleet proof changed", true, false, false),
+            ("current authorization", true, true, true),
+        ] {
+            assert_eq!(
+                remote_version_state_writer_fleet_proof_matches_for(requested, fleet_proof_matches),
+                expected,
+                "{case}"
+            );
         }
     }
 
@@ -3841,6 +3887,16 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         let dest_obj = transaction.remote_object.clone();
         let mut transition_meta = (*oi.user_defined).clone();
         transition_meta.insert("name".to_string(), object.to_string());
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut transition_meta,
+            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TRANSACTION_ID,
+            transaction.transaction_id.to_string(),
+        );
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut transition_meta,
+            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID,
+            rustfs_utils::crypto::hex(transaction.backend_fingerprint),
+        );
 
         if let Some(content_type) = oi.content_type.as_ref().filter(|value| !value.is_empty()) {
             transition_meta.insert(CONTENT_TYPE.to_ascii_lowercase(), content_type.clone());
@@ -3951,20 +4007,24 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object).await;
             return Err(err.into());
         }
-        let (transition_version_id, transition_version_state) = match persisted_transition_version(candidate.remote_version()) {
-            Ok(version) => version,
-            Err(err) => {
-                let cleanup_api = transition_cleanup_store(&self.ctx).await;
-                if let Err(cleanup_err) = upload_cleanup.cleanup_rejected_upload(cleanup_api).await {
-                    return Err(StorageError::Io(std::io::Error::other(format!(
-                        "{err}; rejected remote upload cleanup failed: {cleanup_err}"
-                    ))));
+        let fleet_proof = remote_version_state_writer_fleet_proof();
+        let remote_version_requires_fleet_proof =
+            !candidate.remote_version().is_empty() && Uuid::parse_str(candidate.remote_version()).is_err();
+        let (transition_version_id, transition_version_state) =
+            match persisted_transition_version_with_gate(candidate.remote_version(), fleet_proof.is_some()) {
+                Ok(version) => version,
+                Err(err) => {
+                    let cleanup_api = transition_cleanup_store(&self.ctx).await;
+                    if let Err(cleanup_err) = upload_cleanup.cleanup_rejected_upload(cleanup_api).await {
+                        return Err(StorageError::Io(std::io::Error::other(format!(
+                            "{err}; rejected remote upload cleanup failed: {cleanup_err}"
+                        ))));
+                    }
+                    delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
+                        .await;
+                    return Err(err.into());
                 }
-                delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
-                    .await;
-                return Err(err.into());
-            }
-        };
+            };
         if let Err(err) = advance_and_save_transition_transaction(
             transaction_api.as_ref(),
             &mut transaction,
@@ -4080,6 +4140,21 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         }
         #[cfg(test)]
         pause_transition_commit(bucket, object, TransitionCommitPause::AfterLeaseValidation).await;
+        // This check is the fleet-proof lease linearization point. Revocation
+        // blocks later commits; an already-authorized local quorum commit is
+        // allowed to finish without holding a synchronous lock across I/O.
+        if remote_version_requires_fleet_proof
+            && !fleet_proof
+                .as_ref()
+                .is_some_and(remote_version_state_writer_fleet_proof_matches)
+        {
+            drop(transition_lock_guard);
+            if upload_cleanup.cleanup().await.is_ok() {
+                delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
+                    .await;
+            }
+            return Err(Error::other("remote version state fleet capability changed during transition"));
+        }
         if let Err(err) = advance_and_save_transition_transaction(
             transaction_api.as_ref(),
             &mut transaction,

crates/ecstore/src/store/init.rs

@@ -562,10 +562,12 @@ mod tests {
             },
             tier_sweeper::Jentry,
             transition_transaction::{
-                TRANSITION_TRANSACTION_RECORD_PREFIX, TransitionCleanupDecision, TransitionCleanupProof, TransitionRemoteVersion,
-                TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction, TransitionTransactionInit,
-                TransitionTransactionState, load_transition_transaction_record, recover_transition_transaction_records,
-                save_transition_transaction_record,
+                TRANSITION_TRANSACTION_RECORD_PREFIX, TransitionCleanupDecision, TransitionCleanupProof, TransitionOperatorError,
+                TransitionOperatorProbe, TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode,
+                TransitionTransaction, TransitionTransactionInit, TransitionTransactionState,
+                delete_transition_candidate_for_operator, finalize_missing_transition_transaction_for_operator,
+                inspect_transition_transaction_for_operator, load_transition_transaction_record,
+                recover_transition_transaction_records, save_transition_transaction_record,
             },
         },
         client::transition_api::ReaderImpl,
@@ -575,6 +577,7 @@ mod tests {
         services::tier::{
             test_util::{MockWarmBackend, MockWarmOp, TransitionCleanupStoreBarrier, register_mock_tier},
             tier::{TIER_CONFIG_FILE, TierConfigMgr},
+            tier_config::{TierConfig, TierType, TierWasabi},
             tier_mutation_intent::{
                 TIER_MUTATION_INTENT_RECORD_PREFIX, TierMutationIntent, TierMutationIntentKind, TierMutationIntentState,
                 TierMutationIntentTarget, advance_tier_mutation_intent_record_idempotent, delete_tier_mutation_intent_record,
@@ -1163,6 +1166,37 @@ mod tests {
             .len()
     }
 
+    #[cfg(feature = "test-util")]
+    async fn register_operator_reconcile_test_tier(
+        handle: &Arc<tokio::sync::RwLock<TierConfigMgr>>,
+        tier_name: &str,
+    ) -> MockWarmBackend {
+        let backend = MockWarmBackend::new();
+        let mut manager = handle.write().await;
+        manager.tiers.insert(
+            tier_name.to_string(),
+            TierConfig {
+                version: "v1".to_string(),
+                tier_type: TierType::Wasabi,
+                name: tier_name.to_string(),
+                wasabi: Some(TierWasabi {
+                    name: tier_name.to_string(),
+                    endpoint: "https://s3.wasabisys.com".to_string(),
+                    access_key: "test-access-key".to_string(),
+                    secret_key: "test-secret-key".to_string(),
+                    bucket: "mock-tier".to_string(),
+                    prefix: format!("mock/{}/", uuid::Uuid::new_v4()),
+                    region: "us-east-1".to_string(),
+                }),
+                ..Default::default()
+            },
+        );
+        manager
+            .install_test_driver(tier_name, Box::new(backend.clone()))
+            .expect("mock fallback tier driver should install");
+        backend
+    }
+
     #[cfg(feature = "test-util")]
     async fn wait_for_tier_delete_journal_recovery(
         store: Arc<crate::store::ECStore>,
@@ -2809,6 +2843,157 @@ mod tests {
         }
     }
 
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn operator_reconcile_deletes_exact_candidate_before_finalizing_record() {
+        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(), "transition-operator-reconcile", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+
+        let tier_name = "TXOPERATOR";
+        let backend = register_operator_reconcile_test_tier(&ctx.tier_config_mgr(), tier_name).await;
+        let backend_identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name)
+            .await
+            .expect("tier lease should resolve")
+            .backend_identity();
+        let mut transaction = TransitionTransaction::new(TransitionTransactionInit {
+            deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"),
+            transaction_id: uuid::Uuid::new_v4(),
+            owner_epoch: uuid::Uuid::new_v4(),
+            write_id: uuid::Uuid::new_v4(),
+            source: TransitionSourceIdentity {
+                bucket: "source-bucket".to_string(),
+                object: "source-object".to_string(),
+                version_id: None,
+                data_dir: uuid::Uuid::new_v4(),
+                mod_time_unix_nanos: 1_770_000_000_000_000_000,
+                size: 42,
+                etag: "source-etag".to_string(),
+                version_mode: TransitionSourceVersionMode::Unversioned,
+            },
+            tier_name: tier_name.to_string(),
+            backend_fingerprint: backend_identity,
+            not_after_unix_nanos: 1,
+        })
+        .expect("transaction should build");
+        transaction
+            .advance(transaction.fence(), TransitionTransactionState::UploadOutcomeUnknown, None)
+            .expect("transaction should enter unknown upload outcome state");
+
+        let remote_version = uuid::Uuid::new_v4().to_string();
+        backend.set_put_remote_version(Some(remote_version.clone())).await;
+        let candidate = bytes::Bytes::from_static(b"operator-confirmed transition candidate");
+        backend
+            .put(
+                &transaction.remote_object,
+                ReaderImpl::Body(candidate.clone()),
+                i64::try_from(candidate.len()).expect("test candidate length should fit i64"),
+            )
+            .await
+            .expect("mock backend should accept candidate");
+        save_transition_transaction_record(store.clone(), &transaction)
+            .await
+            .expect("transaction record should persist");
+
+        let status = inspect_transition_transaction_for_operator(store.clone(), transaction.transaction_id)
+            .await
+            .expect("operator inspection should probe the candidate");
+        assert_eq!(status.probe, TransitionOperatorProbe::VersionedPresent(remote_version.clone()));
+
+        let wrong_version = uuid::Uuid::new_v4().to_string();
+        let err = delete_transition_candidate_for_operator(store.clone(), transaction.transaction_id, &wrong_version)
+            .await
+            .expect_err("a mismatched exact version must fail before deleting a candidate");
+        assert!(matches!(
+            err,
+            TransitionOperatorError::CandidateVersionMismatch {
+                expected,
+                actual: TransitionOperatorProbe::VersionedPresent(ref observed),
+            } if expected == wrong_version && observed == &remote_version
+        ));
+        assert!(backend.contains(&transaction.remote_object).await);
+        assert_eq!(backend.exact_remove_count(), 0);
+        load_transition_transaction_record(store.clone(), transaction.transaction_id)
+            .await
+            .expect("an incorrect exact version must retain the transaction journal");
+
+        let result = delete_transition_candidate_for_operator(store.clone(), transaction.transaction_id, &remote_version)
+            .await
+            .expect("operator-confirmed exact candidate should be deleted");
+        assert_eq!(result.status.probe, TransitionOperatorProbe::Missing);
+        assert!(result.journal_observed_after_delete);
+        assert_eq!(backend.exact_remove_count(), 1);
+        assert_eq!(backend.remove_versions().await, vec![(transaction.remote_object.clone(), remote_version)]);
+        load_transition_transaction_record(store.clone(), transaction.transaction_id)
+            .await
+            .expect("candidate deletion must retain the transaction journal");
+
+        finalize_missing_transition_transaction_for_operator(store.clone(), transaction.transaction_id)
+            .await
+            .expect("a separately confirmed missing candidate should permit finalization");
+        assert!(matches!(
+            load_transition_transaction_record(store, transaction.transaction_id).await,
+            Err(Error::ConfigNotFound)
+        ));
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn operator_finalize_retains_record_without_missing_proof() {
+        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(), "transition-operator-fail-closed", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+
+        let tier_name = "TXOPERATORFAIL";
+        let backend = register_operator_reconcile_test_tier(&ctx.tier_config_mgr(), tier_name).await;
+        let backend_identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name)
+            .await
+            .expect("tier lease should resolve")
+            .backend_identity();
+        let mut transaction = TransitionTransaction::new(TransitionTransactionInit {
+            deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"),
+            transaction_id: uuid::Uuid::new_v4(),
+            owner_epoch: uuid::Uuid::new_v4(),
+            write_id: uuid::Uuid::new_v4(),
+            source: TransitionSourceIdentity {
+                bucket: "source-bucket".to_string(),
+                object: "source-object".to_string(),
+                version_id: None,
+                data_dir: uuid::Uuid::new_v4(),
+                mod_time_unix_nanos: 1_770_000_000_000_000_000,
+                size: 42,
+                etag: "source-etag".to_string(),
+                version_mode: TransitionSourceVersionMode::Unversioned,
+            },
+            tier_name: tier_name.to_string(),
+            backend_fingerprint: backend_identity,
+            not_after_unix_nanos: 1,
+        })
+        .expect("transaction should build");
+        transaction
+            .advance(transaction.fence(), TransitionTransactionState::UploadOutcomeUnknown, None)
+            .expect("transaction should enter unknown upload outcome state");
+        save_transition_transaction_record(store.clone(), &transaction)
+            .await
+            .expect("transaction record should persist");
+        backend
+            .set_transition_candidate_probe_override(Some(TransitionCandidateProbe::Unsupported))
+            .await;
+
+        assert!(matches!(
+            finalize_missing_transition_transaction_for_operator(store.clone(), transaction.transaction_id).await,
+            Err(TransitionOperatorError::CandidateNotMissing(TransitionOperatorProbe::Unsupported))
+        ));
+        load_transition_transaction_record(store, transaction.transaction_id)
+            .await
+            .expect("an unsupported probe must retain the transaction journal");
+        assert_eq!(backend.remove_count().await, 0);
+    }
+
     #[cfg(feature = "test-util")]
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]

crates/protos/src/lib.rs

@@ -171,6 +171,7 @@ pub const HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE: usize = heal_control::RESULT_MAX_SI
 pub const HEAL_CONTROL_PROTOCOL_VERSION: u32 = 2;
 pub const DYNAMIC_CONFIG_PROTOCOL_VERSION: u32 = 1;
 pub const HEAL_CONTROL_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-heal-control-capability-v2\0";
+pub const REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-tier-remote-version-state-capability-v1\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;
@@ -197,6 +198,49 @@ pub fn is_heal_control_capability_probe(command: &[u8]) -> bool {
     command.len() == HEAL_CONTROL_CAPABILITY_PROBE_PREFIX.len() + 16 && command.starts_with(HEAL_CONTROL_CAPABILITY_PROBE_PREFIX)
 }
 
+pub fn remote_version_state_capability_probe(nonce: &[u8; 16]) -> Vec<u8> {
+    let mut probe = Vec::with_capacity(REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX.len() + nonce.len());
+    probe.extend_from_slice(REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX);
+    probe.extend_from_slice(nonce);
+    probe
+}
+
+pub fn is_remote_version_state_capability_probe(command: &[u8]) -> bool {
+    command.len() == REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX.len() + 16
+        && command.starts_with(REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX)
+}
+
+pub fn encode_remote_version_state_capability(
+    topology_member: &str,
+    process_epoch: &[u8; 16],
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let topology_member = topology_member.as_bytes();
+    let mut result = Vec::with_capacity(8 + topology_member.len() + process_epoch.len());
+    result.extend_from_slice(&u64::try_from(topology_member.len())?.to_be_bytes());
+    result.extend_from_slice(topology_member);
+    result.extend_from_slice(process_epoch);
+    Ok(result)
+}
+
+pub fn decode_remote_version_state_capability(result: &[u8]) -> Result<(&str, &[u8; 16]), &'static str> {
+    let member_len = result
+        .get(..8)
+        .and_then(|value| value.try_into().ok())
+        .map(u64::from_be_bytes)
+        .ok_or("remote version state capability is truncated")?;
+    let member_len = usize::try_from(member_len).map_err(|_| "remote version state member length cannot be represented")?;
+    let member_end = 8_usize
+        .checked_add(member_len)
+        .ok_or("remote version state member length overflow")?;
+    let topology_member = std::str::from_utf8(result.get(8..member_end).ok_or("remote version state member is truncated")?)
+        .map_err(|_| "remote version state member is not UTF-8")?;
+    let process_epoch = result
+        .get(member_end..)
+        .and_then(|value| value.try_into().ok())
+        .ok_or("remote version state process epoch has an invalid length")?;
+    Ok((topology_member, process_epoch))
+}
+
 /// Builds the stable byte representation authenticated for a heal-control request.
 ///
 /// This deliberately does not reuse protobuf encoding: mixed-version peers may
@@ -1656,10 +1700,12 @@ mod scanner_activity_tests {
 #[cfg(test)]
 mod heal_control_tests {
     use super::{
-        HEAL_CONTROL_CAPABILITY_PROBE_PREFIX, HEAL_CONTROL_PROTOCOL_VERSION, canonical_heal_control_capability_ack,
-        canonical_heal_control_request_body, canonical_heal_control_response_body, heal_control_capability_probe,
+        HEAL_CONTROL_CAPABILITY_PROBE_PREFIX, HEAL_CONTROL_PROTOCOL_VERSION, REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX,
+        canonical_heal_control_capability_ack, canonical_heal_control_request_body, canonical_heal_control_response_body,
+        decode_remote_version_state_capability, encode_remote_version_state_capability, heal_control_capability_probe,
         heal_control_coordinator_epoch, heal_control_execution_timeout, heal_control_execution_timeout_for,
-        internode_rpc_timeout, is_heal_control_capability_probe, normalize_internode_rpc_timeout,
+        internode_rpc_timeout, is_heal_control_capability_probe, is_remote_version_state_capability_probe,
+        normalize_internode_rpc_timeout, remote_version_state_capability_probe,
     };
     use crate::heal_control;
     use std::time::Duration;
@@ -1716,6 +1762,29 @@ mod heal_control_tests {
         assert!(!is_heal_control_capability_probe(HEAL_CONTROL_CAPABILITY_PROBE_PREFIX));
     }
 
+    #[test]
+    fn remote_version_state_capability_probe_requires_exact_nonce() {
+        let probe = remote_version_state_capability_probe(&[7; 16]);
+        assert!(is_remote_version_state_capability_probe(&probe));
+        assert!(!is_remote_version_state_capability_probe(REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX));
+    }
+
+    #[test]
+    fn remote_version_state_capability_binds_member_and_process_epoch() {
+        let encoded =
+            encode_remote_version_state_capability("node-a:9000", &[7; 16]).expect("small capability response should encode");
+        assert_eq!(
+            decode_remote_version_state_capability(&encoded).expect("capability response should decode"),
+            ("node-a:9000", &[7; 16])
+        );
+        assert!(decode_remote_version_state_capability(&encoded[..encoded.len() - 1]).is_err());
+
+        let mut invalid_utf8 =
+            encode_remote_version_state_capability("node-a", &[7; 16]).expect("small capability response should encode");
+        invalid_utf8[8] = 0xff;
+        assert!(decode_remote_version_state_capability(&invalid_utf8).is_err());
+    }
+
     #[test]
     fn canonical_response_binds_request_and_result() {
         let baseline = canonical_heal_control_response_body(2, "abcdef", b"query", b"result").unwrap();

crates/utils/src/http/metadata_compat.rs

@@ -40,6 +40,7 @@ pub const SUFFIX_TRANSITIONED_VERSION_ID: &str = "transitioned-versionID";
 pub const SUFFIX_TRANSITIONED_VERSION_STATE: &str = "transitioned-version-state";
 pub const SUFFIX_TRANSITION_TIER: &str = "transition-tier";
 pub const SUFFIX_TRANSITION_TIER_DESTINATION_ID: &str = "transition-tier-destination-id";
+pub const SUFFIX_TRANSITION_TRANSACTION_ID: &str = "transition-transaction-id";
 pub const SUFFIX_RESTORE_OPERATION_ID: &str = "restore-operation-id";
 pub const SUFFIX_FREE_VERSION: &str = "free-version";
 pub const SUFFIX_PURGESTATUS: &str = "purgestatus";

docs/operations/tier-ilm-debugging.md

@@ -98,6 +98,39 @@ rc admin ilm transition run local/mybucket --prefix logs/ --tier cold --max-obje
 
 Inspect the aggregate counters before widening scope. Full object-key lists are intentionally not returned by the admin response. If `RUSTFS_RPC_SECRET` or other credentials were pasted into an issue, chat, log, or ticket while debugging tiering, rotate them on every node, restart the cluster with the new value, and redact the exposed copy before sharing more diagnostics.
 
+## Reconcile an unknown transition upload
+
+Historical transition transactions in `upload_outcome_unknown` state can use an explicit two-stage operator workflow when the tier probe is ambiguous and the provider supports exact version deletion. The endpoint refuses transactions that are still inside their ownership window or are in any other state.
+
+First inspect the transaction without changing it:
+
+```text
+GET /rustfs/admin/v3/ilm/transition/reconcile/<transaction-id>
+```
+
+If independent provider evidence identifies the exact remote version to remove, submit that opaque version identifier with explicit confirmation:
+
+```json
+POST /rustfs/admin/v3/ilm/transition/reconcile/<transaction-id>
+{
+  "action": "delete_candidate",
+  "confirm": true,
+  "remote_version_id": "<exact-provider-version>"
+}
+```
+
+This operation performs only an exact version delete. Its response reports whether the transaction journal was still observed after the delete; background recovery may have finalized the same transaction concurrently. If the journal remains, inspect the transaction again and finalize it only after the live provider probe proves that the candidate is missing:
+
+```json
+POST /rustfs/admin/v3/ilm/transition/reconcile/<transaction-id>
+{
+  "action": "finalize_missing",
+  "confirm": true
+}
+```
+
+`finalize_missing` re-runs the provider probe and fails closed for `unversioned_present`, `versioned_present`, `ambiguous`, `unsupported`, or probe errors. It never accepts an operator assertion in place of a live `missing` result. Providers without an authoritative probe or exact version deletion remain pending; this endpoint does not infer provider capabilities, accept external absence assertions, or select a candidate automatically.
+
 ## Historical fixes (for context, already merged)
 
 - Expire/GET race (`NoSuchVersion` during expiry of a tiered object):

rustfs/src/admin/handlers/ilm_transition.rs

@@ -20,8 +20,10 @@ use crate::admin::storage_api::error::StorageError;
 use crate::admin::storage_api::lifecycle::{
     ManualTransitionCancelCheck, ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionProgressSink,
     ManualTransitionQueueSnapshot, ManualTransitionRunOptions, ManualTransitionRunReport, ManualTransitionScopeAdmission,
-    ManualTransitionScopeAdmissionClaim, claim_manual_transition_scope_admission,
-    delete_manual_transition_scope_admission_if_current, enqueue_transition_for_existing_objects_scoped,
+    ManualTransitionScopeAdmissionClaim, TransitionOperatorDeleteResult, TransitionOperatorError,
+    claim_manual_transition_scope_admission, delete_manual_transition_scope_admission_if_current,
+    delete_transition_candidate_for_operator, enqueue_transition_for_existing_objects_scoped,
+    finalize_missing_transition_transaction_for_operator, inspect_transition_transaction_for_operator,
     load_manual_transition_job_record, load_manual_transition_job_record_with_etag, load_manual_transition_scope_admission,
     manual_transition_job_lease_expired, manual_transition_queue_snapshot, manual_transition_scope_admission_lease_expired,
     persist_manual_transition_job_progress, renew_manual_transition_job_lease, request_manual_transition_job_cancel,
@@ -33,6 +35,7 @@ use crate::server::{ADMIN_PREFIX, RemoteAddr};
 use http::{HeaderMap, HeaderValue};
 use hyper::{Method, StatusCode};
 use matchit::Params;
+use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
 use rustfs_policy::policy::action::{Action, AdminAction};
 use rustfs_utils::{
     MaskedAccessKey,
@@ -56,6 +59,7 @@ const MAX_MANUAL_TRANSITION_DURATION_SECONDS: u64 = 3600;
 const LOG_COMPONENT_ADMIN: &str = "admin";
 const LOG_SUBSYSTEM_ILM_TRANSITION: &str = "ilm_transition";
 const EVENT_ADMIN_ILM_TRANSITION_STATE: &str = "admin_ilm_transition_state";
+const EVENT_ADMIN_ILM_TRANSITION_RECONCILE: &str = "admin_ilm_transition_reconcile";
 
 static ACTIVE_MANUAL_TRANSITION_SCOPES: OnceLock<Mutex<Vec<ManualTransitionRunScope>>> = OnceLock::new();
 #[cfg(feature = "e2e-test-hooks")]
@@ -218,6 +222,16 @@ pub fn register_ilm_transition_route(r: &mut S3Router<AdminOperation>) -> std::i
         format!("{ADMIN_PREFIX}/v3/ilm/transition/jobs/{{job_id}}").as_str(),
         AdminOperation(&ManualTransitionJobCancelHandler {}),
     )?;
+    r.insert(
+        Method::GET,
+        format!("{ADMIN_PREFIX}/v3/ilm/transition/reconcile/{{transaction_id}}").as_str(),
+        AdminOperation(&TransitionReconcileInspectHandler {}),
+    )?;
+    r.insert(
+        Method::POST,
+        format!("{ADMIN_PREFIX}/v3/ilm/transition/reconcile/{{transaction_id}}").as_str(),
+        AdminOperation(&TransitionReconcileApplyHandler {}),
+    )?;
     Ok(())
 }
 
@@ -389,6 +403,10 @@ fn log_manual_transition_completed(
 }
 
 async fn authorize_manual_transition_request(req: &S3Request<Body>) -> S3Result<String> {
+    authorize_transition_admin_request(req, AdminAction::SetTierAction).await
+}
+
+async fn authorize_transition_admin_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<String> {
     let Some(input_cred) = req.credentials.as_ref() else {
         return Err(s3_error!(InvalidRequest, "authentication required"));
     };
@@ -401,19 +419,122 @@ async fn authorize_manual_transition_request(req: &S3Request<Body>) -> S3Result<
         .get::<Option<RemoteAddr>>()
         .and_then(|opt| opt.map(|addr| addr.0));
 
-    validate_admin_request(
-        &req.headers,
-        &cred,
-        owner,
-        false,
-        vec![Action::AdminAction(AdminAction::SetTierAction)],
-        remote_addr,
-    )
-    .await?;
+    validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(action)], remote_addr).await?;
 
     Ok(actor)
 }
 
+fn transition_transaction_id_from_params(params: &Params<'_, '_>) -> S3Result<Uuid> {
+    Uuid::parse_str(params.get("transaction_id").unwrap_or(""))
+        .map_err(|_| s3_error!(InvalidArgument, "invalid transition transaction id"))
+}
+
+fn map_transition_operator_error(err: TransitionOperatorError) -> S3Error {
+    match err {
+        TransitionOperatorError::NotFound => s3_error!(NoSuchKey, "transition transaction not found"),
+        TransitionOperatorError::NotExpired => {
+            s3_error!(OperationAborted, "transition transaction is still inside its active ownership window")
+        }
+        TransitionOperatorError::InvalidState(_) => {
+            s3_error!(OperationAborted, "transition transaction is not eligible for operator reconciliation")
+        }
+        TransitionOperatorError::RemoteVersionRequired => {
+            s3_error!(InvalidArgument, "an exact non-empty remote version is required")
+        }
+        TransitionOperatorError::CandidateNotMissing(_) => {
+            s3_error!(OperationAborted, "remote candidate is not proven missing")
+        }
+        TransitionOperatorError::CandidateVersionMismatch { .. } => {
+            s3_error!(OperationAborted, "remote candidate version does not match requested exact version")
+        }
+        TransitionOperatorError::Store(_) | TransitionOperatorError::Remote(_) => {
+            s3_error!(InternalError, "transition reconciliation failed")
+        }
+    }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "snake_case")]
+enum TransitionReconcileAction {
+    DeleteCandidate,
+    FinalizeMissing,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct TransitionReconcileRequest {
+    action: TransitionReconcileAction,
+    confirm: bool,
+    #[serde(default)]
+    remote_version_id: Option<String>,
+}
+
+enum ValidatedTransitionReconcileAction<'a> {
+    DeleteCandidate(&'a str),
+    FinalizeMissing,
+}
+
+fn validate_transition_reconcile_request(
+    request: &TransitionReconcileRequest,
+) -> S3Result<ValidatedTransitionReconcileAction<'_>> {
+    if !request.confirm {
+        return Err(s3_error!(
+            InvalidRequest,
+            "transition reconciliation requires confirm=true; use GET to inspect without changes"
+        ));
+    }
+    match request.action {
+        TransitionReconcileAction::DeleteCandidate => request
+            .remote_version_id
+            .as_deref()
+            .filter(|version_id| !version_id.is_empty())
+            .map(ValidatedTransitionReconcileAction::DeleteCandidate)
+            .ok_or_else(|| s3_error!(InvalidArgument, "delete_candidate requires remote_version_id")),
+        TransitionReconcileAction::FinalizeMissing if request.remote_version_id.is_none() => {
+            Ok(ValidatedTransitionReconcileAction::FinalizeMissing)
+        }
+        TransitionReconcileAction::FinalizeMissing => {
+            Err(s3_error!(InvalidArgument, "finalize_missing must not include remote_version_id"))
+        }
+    }
+}
+
+#[derive(Debug, Serialize)]
+struct TransitionCandidateDeleteResponse {
+    outcome: &'static str,
+    result: TransitionOperatorDeleteResult,
+}
+
+#[derive(Debug, Serialize)]
+struct TransitionFinalizeMissingResponse {
+    outcome: &'static str,
+    journal_retained: bool,
+    transaction_id: Uuid,
+}
+
+fn log_transition_reconcile_applied(
+    transaction_id: Uuid,
+    action: &str,
+    outcome: &str,
+    request_id: &str,
+    actor: &str,
+    remote_addr: &str,
+) {
+    info!(
+        event = EVENT_ADMIN_ILM_TRANSITION_RECONCILE,
+        component = LOG_COMPONENT_ADMIN,
+        subsystem = LOG_SUBSYSTEM_ILM_TRANSITION,
+        operation = "transition_operator_reconcile",
+        transaction_id = %transaction_id,
+        action,
+        outcome,
+        request_id = %request_id,
+        actor = %actor,
+        remote_addr = %remote_addr,
+        "admin transition reconciliation applied"
+    );
+}
+
 fn response_state(report: &ManualTransitionRunReport) -> &'static str {
     if report.was_truncated() || report.has_partial_enqueue() || report.tier_failure > 0 || report.transition_failed > 0 {
         "partial"
@@ -902,6 +1023,81 @@ impl Operation for ManualTransitionJobCancelHandler {
     }
 }
 
+pub struct TransitionReconcileInspectHandler {}
+
+#[async_trait::async_trait]
+impl Operation for TransitionReconcileInspectHandler {
+    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_transition_admin_request(&req, AdminAction::ListTierAction).await?;
+        let transaction_id = transition_transaction_id_from_params(&params)?;
+        let Some(store) = object_store_from_extensions(&req.extensions) else {
+            return Err(s3_error!(InternalError, "object store is not initialized"));
+        };
+        let status = inspect_transition_transaction_for_operator(store, transaction_id)
+            .await
+            .map_err(map_transition_operator_error)?;
+        json_response(&status, StatusCode::OK)
+    }
+}
+
+pub struct TransitionReconcileApplyHandler {}
+
+#[async_trait::async_trait]
+impl Operation for TransitionReconcileApplyHandler {
+    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let request_id = admin_request_id(&req.headers).unwrap_or_default().to_string();
+        let remote_addr = admin_remote_addr(&req).unwrap_or_default();
+        let actor = authorize_transition_admin_request(&req, AdminAction::SetTierAction).await?;
+        let transaction_id = transition_transaction_id_from_params(&params)?;
+        let Some(store) = object_store_from_extensions(&req.extensions) else {
+            return Err(s3_error!(InternalError, "object store is not initialized"));
+        };
+        let mut input = req.input;
+        let body = input
+            .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE)
+            .await
+            .map_err(|_| s3_error!(InvalidRequest, "transition reconciliation body is too large or unreadable"))?;
+        let request: TransitionReconcileRequest = serde_json::from_slice(&body)
+            .map_err(|_| s3_error!(InvalidRequest, "transition reconciliation request must be valid JSON"))?;
+
+        match validate_transition_reconcile_request(&request)? {
+            ValidatedTransitionReconcileAction::DeleteCandidate(remote_version_id) => {
+                let result = delete_transition_candidate_for_operator(store, transaction_id, remote_version_id)
+                    .await
+                    .map_err(map_transition_operator_error)?;
+                let outcome = if result.journal_observed_after_delete {
+                    "exact_delete_completed_journal_observed"
+                } else {
+                    "exact_delete_completed_journal_already_finalized"
+                };
+                log_transition_reconcile_applied(transaction_id, "delete_candidate", outcome, &request_id, &actor, &remote_addr);
+                json_response(&TransitionCandidateDeleteResponse { outcome, result }, StatusCode::OK)
+            }
+            ValidatedTransitionReconcileAction::FinalizeMissing => {
+                finalize_missing_transition_transaction_for_operator(store, transaction_id)
+                    .await
+                    .map_err(map_transition_operator_error)?;
+                log_transition_reconcile_applied(
+                    transaction_id,
+                    "finalize_missing",
+                    "journal_deleted_after_missing_probe",
+                    &request_id,
+                    &actor,
+                    &remote_addr,
+                );
+                json_response(
+                    &TransitionFinalizeMissingResponse {
+                        outcome: "journal_finalized",
+                        journal_retained: false,
+                        transaction_id,
+                    },
+                    StatusCode::OK,
+                )
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -931,6 +1127,65 @@ mod tests {
         }
     }
 
+    #[test]
+    fn transition_reconcile_request_is_explicit_and_fail_closed() {
+        let unconfirmed: TransitionReconcileRequest =
+            serde_json::from_slice(br#"{"action":"delete_candidate","confirm":false,"remote_version_id":"v1"}"#)
+                .expect("request should decode");
+        assert!(validate_transition_reconcile_request(&unconfirmed).is_err());
+
+        let missing_version: TransitionReconcileRequest =
+            serde_json::from_slice(br#"{"action":"delete_candidate","confirm":true}"#).expect("request should decode");
+        assert!(validate_transition_reconcile_request(&missing_version).is_err());
+
+        let unsafe_finalize: TransitionReconcileRequest =
+            serde_json::from_slice(br#"{"action":"finalize_missing","confirm":true,"remote_version_id":"v1"}"#)
+                .expect("request should decode");
+        assert!(validate_transition_reconcile_request(&unsafe_finalize).is_err());
+
+        let delete: TransitionReconcileRequest =
+            serde_json::from_slice(br#"{"action":"delete_candidate","confirm":true,"remote_version_id":"opaque-v1"}"#)
+                .expect("request should decode");
+        assert!(matches!(
+            validate_transition_reconcile_request(&delete),
+            Ok(ValidatedTransitionReconcileAction::DeleteCandidate("opaque-v1"))
+        ));
+
+        let finalize: TransitionReconcileRequest =
+            serde_json::from_slice(br#"{"action":"finalize_missing","confirm":true}"#).expect("request should decode");
+        assert!(matches!(
+            validate_transition_reconcile_request(&finalize),
+            Ok(ValidatedTransitionReconcileAction::FinalizeMissing)
+        ));
+
+        assert!(
+            serde_json::from_slice::<TransitionReconcileRequest>(
+                br#"{"action":"finalize_missing","confirm":true,"unexpected":true}"#
+            )
+            .is_err()
+        );
+    }
+
+    #[test]
+    fn transition_reconcile_routes_use_read_and_write_tier_actions() {
+        let src = include_str!("ilm_transition.rs");
+        let inspect = src
+            .split("impl Operation for TransitionReconcileInspectHandler")
+            .nth(1)
+            .and_then(|block| block.split("impl Operation for TransitionReconcileApplyHandler").next())
+            .expect("inspect handler block");
+        assert!(inspect.contains("AdminAction::ListTierAction"));
+        assert!(!inspect.contains("AdminAction::SetTierAction"));
+
+        let apply = src
+            .split("impl Operation for TransitionReconcileApplyHandler")
+            .nth(1)
+            .and_then(|block| block.split("#[cfg(test)]").next())
+            .expect("apply handler block");
+        assert!(apply.contains("AdminAction::SetTierAction"));
+        assert!(!apply.contains("AdminAction::ListTierAction"));
+    }
+
     #[test]
     fn manual_transition_query_defaults_to_bounded_run() {
         let (bucket, options, run_mode) =

rustfs/src/admin/route_policy.rs

@@ -432,6 +432,18 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
         SET_TIER,
         RouteRiskLevel::High,
     ),
+    admin(
+        HttpMethod::Get,
+        "/rustfs/admin/v3/ilm/transition/reconcile/{transaction_id}",
+        LIST_TIER,
+        RouteRiskLevel::High,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/ilm/transition/reconcile/{transaction_id}",
+        SET_TIER,
+        RouteRiskLevel::High,
+    ),
     admin(
         HttpMethod::Get,
         "/rustfs/admin/v3/audit/target/list",
@@ -1825,13 +1837,17 @@ mod tests {
     }
 
     #[test]
-    fn route_policy_requires_set_tier_for_manual_transition_routes() {
+    fn route_policy_uses_tier_actions_for_transition_routes() {
         assert_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/run", SET_TIER);
         assert_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/transition/jobs/{job_id}", SET_TIER);
         assert_action(HttpMethod::Delete, "/rustfs/admin/v3/ilm/transition/jobs/{job_id}", SET_TIER);
+        assert_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/transition/reconcile/{transaction_id}", LIST_TIER);
+        assert_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/reconcile/{transaction_id}", SET_TIER);
         assert_not_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/run", SERVER_INFO);
         assert_not_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/transition/jobs/{job_id}", SERVER_INFO);
         assert_not_action(HttpMethod::Delete, "/rustfs/admin/v3/ilm/transition/jobs/{job_id}", SERVER_INFO);
+        assert_not_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/transition/reconcile/{transaction_id}", SET_TIER);
+        assert_not_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/reconcile/{transaction_id}", LIST_TIER);
     }
 
     #[test]

rustfs/src/admin/route_registration_test.rs

@@ -202,6 +202,16 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
             "/v3/ilm/transition/jobs/{job_id}",
             "/v3/ilm/transition/jobs/11111111-1111-4111-8111-111111111111",
         ),
+        admin_route_sample(
+            Method::GET,
+            "/v3/ilm/transition/reconcile/{transaction_id}",
+            "/v3/ilm/transition/reconcile/11111111-1111-4111-8111-111111111111",
+        ),
+        admin_route_sample(
+            Method::POST,
+            "/v3/ilm/transition/reconcile/{transaction_id}",
+            "/v3/ilm/transition/reconcile/11111111-1111-4111-8111-111111111111",
+        ),
         admin_route_sample(
             Method::DELETE,
             "/v3/ilm/transition/jobs/{job_id}",
@@ -856,6 +866,16 @@ fn test_register_routes_cover_representative_admin_paths() {
         Method::DELETE,
         &admin_path("/v3/ilm/transition/jobs/11111111-1111-4111-8111-111111111111"),
     );
+    assert_route(
+        &router,
+        Method::GET,
+        &admin_path("/v3/ilm/transition/reconcile/11111111-1111-4111-8111-111111111111"),
+    );
+    assert_route(
+        &router,
+        Method::POST,
+        &admin_path("/v3/ilm/transition/reconcile/11111111-1111-4111-8111-111111111111"),
+    );
 
     assert_route(&router, Method::GET, &table_catalog_path("/config"));
     assert_route(&router, Method::PUT, &table_catalog_path("/buckets/analytics"));

rustfs/src/admin/storage_api.rs

@@ -210,6 +210,10 @@ pub(crate) mod lifecycle {
     pub(crate) type ManualTransitionRunOptions =
         super::ecstore_bucket::lifecycle::bucket_lifecycle_ops::ManualTransitionRunOptions;
     pub(crate) type ManualTransitionRunReport = super::ecstore_bucket::lifecycle::bucket_lifecycle_ops::ManualTransitionRunReport;
+    pub(crate) use super::ecstore_bucket::lifecycle::transition_transaction::{
+        TransitionOperatorDeleteResult, TransitionOperatorError, delete_transition_candidate_for_operator,
+        finalize_missing_transition_transaction_for_operator, inspect_transition_transaction_for_operator,
+    };
 
     pub(crate) async fn enqueue_transition_for_existing_objects_scoped(
         api: std::sync::Arc<super::ECStore>,

rustfs/src/storage/rpc/node_service.rs

@@ -52,7 +52,7 @@ use std::{
     collections::HashMap,
     io::Cursor,
     pin::Pin,
-    sync::{Arc, OnceLock},
+    sync::{Arc, LazyLock, OnceLock},
 };
 use time::OffsetDateTime;
 use tokio::spawn;
@@ -128,6 +128,7 @@ fn remove_heal_control_replay(
 }
 
 static HEAL_CONTROL_REPLAY_CACHE: OnceLock<tokio::sync::Mutex<HashMap<String, Arc<HealControlReplayEntry>>>> = OnceLock::new();
+static NODE_CAPABILITY_SERVER_EPOCH: LazyLock<Uuid> = LazyLock::new(Uuid::new_v4);
 
 fn admit_heal_control_replay(
     replay_cache: &mut HashMap<String, Arc<HealControlReplayEntry>>,
@@ -465,6 +466,19 @@ impl HealControlRpcService {
 pub(crate) async fn initialize_heal_topology_fingerprint(
     cache: Arc<tokio::sync::OnceCell<String>>,
     endpoint_pools: EndpointServerPools,
+) -> Result<(), String> {
+    initialize_heal_topology_fingerprint_with_probe(
+        cache,
+        endpoint_pools,
+        crate::storage::storage_api::start_remote_version_state_fleet_probe,
+    )
+    .await
+}
+
+async fn initialize_heal_topology_fingerprint_with_probe(
+    cache: Arc<tokio::sync::OnceCell<String>>,
+    endpoint_pools: EndpointServerPools,
+    start_probe: impl FnOnce(String),
 ) -> Result<(), String> {
     if cache.get().is_some() {
         return Ok(());
@@ -472,7 +486,8 @@ pub(crate) async fn initialize_heal_topology_fingerprint(
     let fingerprint = tokio::task::spawn_blocking(move || heal::heal_topology_fingerprint(&endpoint_pools))
         .await
         .map_err(|_| "heal control topology calculation task failed".to_string())??;
-    let _ = cache.set(fingerprint);
+    let _ = cache.set(fingerprint.clone());
+    start_probe(fingerprint);
     Ok(())
 }
 
@@ -806,6 +821,35 @@ impl heal_control_service_server::HealControlService for HealControlRpcService {
                 response_proof: Bytes::new(),
             }));
         }
+        if rustfs_protos::is_remote_version_state_capability_probe(&request.get_ref().command) {
+            let topology_member = self
+                .endpoint_pools()
+                .await
+                .ok_or_else(|| Status::failed_precondition("heal control topology is not initialized"))?
+                .peers()
+                .1;
+            if topology_member.is_empty() {
+                return Err(Status::failed_precondition("local topology member identity is unavailable"));
+            }
+            let result =
+                rustfs_protos::encode_remote_version_state_capability(&topology_member, NODE_CAPABILITY_SERVER_EPOCH.as_bytes())
+                    .map_err(|_| Status::internal("remote version state capability length cannot be represented"))?;
+            let canonical_response = rustfs_protos::canonical_heal_control_response_body(
+                request.get_ref().version,
+                &request.get_ref().topology_fingerprint,
+                &request.get_ref().command,
+                &result,
+            )
+            .map_err(|_| Status::internal("heal control response length cannot be represented"))?;
+            let response_proof = sign_tonic_rpc_response_proof(&canonical_response)
+                .map_err(|_| Status::internal("heal control response proof is unavailable"))?;
+            return Ok(Response::new(HealControlResponse {
+                success: true,
+                result: result.into(),
+                error_info: None,
+                response_proof: response_proof.into(),
+            }));
+        }
         let endpoints = self
             .endpoint_pools()
             .await
@@ -2090,10 +2134,10 @@ mod tests {
         PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION,
         SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC,
         STORAGE_CLASS_SUB_SYS, admit_heal_control_replay, background_rebalance_start_error_message,
-        execute_heal_control_envelope_with_manager, initialize_heal_topology_fingerprint, legacy_scanner_activity_response,
-        make_heal_control_server, make_heal_control_server_with_cache, make_server, make_server_for_context,
-        make_tier_mutation_control_server_for_context, previous_scanner_activity_response, remove_heal_control_replay,
-        scanner_activity_response, stop_rebalance_response,
+        execute_heal_control_envelope_with_manager, initialize_heal_topology_fingerprint,
+        initialize_heal_topology_fingerprint_with_probe, legacy_scanner_activity_response, make_heal_control_server,
+        make_heal_control_server_with_cache, make_server, make_server_for_context, make_tier_mutation_control_server_for_context,
+        previous_scanner_activity_response, remove_heal_control_replay, scanner_activity_response, stop_rebalance_response,
     };
     use crate::storage::rpc::node_service::heal::heal_topology_fingerprint;
     use crate::storage::storage_api::rpc_consumer::node_service::{HealBucketInfo, HealEndpoint};
@@ -2147,6 +2191,7 @@ mod tests {
     use tokio::time::Duration;
     use tokio_stream::wrappers::TcpListenerStream;
     use tonic::{Request, Response, Status};
+    use uuid::Uuid;
 
     const DISK_MUTATION_RPC_METHODS: [&str; 18] = [
         "renamedata",
@@ -3132,6 +3177,60 @@ mod tests {
         assert_eq!(non_coordinator.code(), tonic::Code::FailedPrecondition);
     }
 
+    #[tokio::test]
+    async fn remote_version_state_probe_authenticates_topology_challenge_and_process_epoch() {
+        let _ = rustfs_credentials::set_global_rpc_secret("remote-version-state-node-service-test-secret".to_string());
+        let endpoints = heal_control_test_endpoints_with_coordinator("node-d", true);
+        let fingerprint = heal_topology_fingerprint(&endpoints).expect("test topology should hash");
+        let (service, source) = super::make_heal_control_server_for_source();
+        *source.write().await = Some(endpoints);
+        let probe_command = rustfs_protos::remote_version_state_capability_probe(&[7; 16]);
+        let mut request = Request::new(HealControlRequest {
+            version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION,
+            topology_fingerprint: fingerprint.clone(),
+            command: Bytes::from(probe_command.clone()),
+        });
+        let body = rustfs_protos::canonical_heal_control_request_body(
+            request.get_ref().version,
+            &request.get_ref().topology_fingerprint,
+            &request.get_ref().command,
+        )
+        .expect("probe should encode");
+        set_tonic_canonical_body_digest(&mut request, &body).expect("digest metadata should encode");
+        mark_v2_authenticated(&mut request);
+        let response = service
+            .heal_control(request)
+            .await
+            .expect("matching topology should be acknowledged")
+            .into_inner();
+
+        let (topology_member, process_epoch) =
+            rustfs_protos::decode_remote_version_state_capability(&response.result).expect("capability response should decode");
+        assert_eq!(topology_member, "node-a:9000");
+        let server_epoch = Uuid::from_slice(process_epoch).expect("server epoch should be a UUID");
+        assert!(!server_epoch.is_nil());
+        let canonical_response = rustfs_protos::canonical_heal_control_response_body(
+            rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION,
+            &fingerprint,
+            &probe_command,
+            &response.result,
+        )
+        .expect("response should encode");
+        crate::storage::storage_api::verify_tonic_rpc_response_proof(&canonical_response, &response.response_proof)
+            .expect("outer proof should bind the response to the request");
+
+        let different_probe = rustfs_protos::remote_version_state_capability_probe(&[8; 16]);
+        let different_response = rustfs_protos::canonical_heal_control_response_body(
+            rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION,
+            &fingerprint,
+            &different_probe,
+            &response.result,
+        )
+        .expect("different response should encode");
+        crate::storage::storage_api::verify_tonic_rpc_response_proof(&different_response, &response.response_proof)
+            .expect_err("proof from one challenge must not be reusable");
+    }
+
     #[tokio::test]
     async fn heal_control_coordinator_rejects_expired_and_non_admin_starts() {
         let _ = rustfs_credentials::set_global_rpc_secret("heal-control-node-service-test-secret".to_string());
@@ -3194,10 +3293,15 @@ mod tests {
         let topology = heal_control_test_endpoints("node-d");
         let expected = heal_topology_fingerprint(&topology).expect("test topology should hash");
         let cache = Arc::new(tokio::sync::OnceCell::new());
-        initialize_heal_topology_fingerprint(Arc::clone(&cache), topology)
-            .await
-            .expect("valid topology should initialize");
+        let started_probe = Arc::new(std::sync::Mutex::new(None));
+        let started_probe_capture = Arc::clone(&started_probe);
+        initialize_heal_topology_fingerprint_with_probe(Arc::clone(&cache), topology, move |fingerprint| {
+            *started_probe_capture.lock().expect("probe capture should not poison") = Some(fingerprint);
+        })
+        .await
+        .expect("valid topology should initialize");
         assert_eq!(cache.get(), Some(&expected));
+        assert_eq!(started_probe.lock().expect("probe capture should not poison").as_ref(), Some(&expected));
 
         let mut invalid = heal_control_test_endpoints("node-d");
         invalid.as_mut()[0].endpoints.as_mut()[0].pool_idx = -1;

rustfs/src/storage/storage_api.rs

@@ -472,7 +472,7 @@ pub(crate) mod ecstore_metrics {
 #[allow(unused_imports)]
 pub(crate) mod ecstore_notification {
     pub(crate) use rustfs_ecstore::api::notification::{
-        NotificationSys, get_global_notification_sys, new_global_notification_sys,
+        NotificationSys, get_global_notification_sys, new_global_notification_sys, start_remote_version_state_fleet_probe,
     };
 }
 
@@ -963,6 +963,10 @@ pub(crate) async fn new_global_notification_sys(endpoint_pools: EndpointServerPo
     ecstore_notification::new_global_notification_sys(endpoint_pools).await
 }
 
+pub(crate) fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
+    ecstore_notification::start_remote_version_state_fleet_probe(topology_fingerprint);
+}
+
 pub(crate) async fn read_config(api: Arc<ECStore>, file: &str) -> Result<Vec<u8>> {
     ecstore_config::com::read_config(api, file).await
 }