fix(tier): add operator transition reconciliation (#5410)

* fix(tier): add operator transition reconciliation

* fix(tiering): require live fleet capability proof (#5423)

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,183 @@ 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("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)?;
+    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)
@@ -1097,6 +1275,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/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}"
+            );
         }
     }
 
@@ -3961,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,
@@ -4090,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,154 @@ 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 result = delete_transition_candidate_for_operator(store.clone(), transaction.transaction_id, &wrong_version)
+            .await
+            .expect("an exact delete of an absent version should be idempotent");
+        assert_eq!(
+            result.status.probe,
+            TransitionOperatorProbe::VersionedPresent(remote_version.clone()),
+            "an incorrect exact version must not delete the provider-confirmed candidate"
+        );
+        assert!(result.journal_observed_after_delete);
+        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(), 2);
+        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();