fix(tiering): require live fleet capability proof

crates/config/src/constants/object.rs

@@ -137,18 +137,6 @@ pub const DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED: bool = false;
 const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE);
 const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED);
 
-#[cfg(test)]
-mod remote_version_state_tests {
-    #[test]
-    fn remote_version_state_gate_uses_stable_environment_names() {
-        assert_eq!(super::ENV_TIER_REMOTE_VERSION_STATE_WRITE, "RUSTFS_TIER_REMOTE_VERSION_STATE_WRITE");
-        assert_eq!(
-            super::ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED,
-            "RUSTFS_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED"
-        );
-    }
-}
-
 // =============================================================================
 // Concurrent Request Fix - Timeout and Backpressure Configuration
 // =============================================================================
@@ -650,3 +638,15 @@ pub const ENV_OBJECT_IO_RANDOM_READAHEAD_DISABLE_CONCURRENCY: &str = "RUSTFS_OBJ
 
 /// Default read-ahead disable concurrency threshold: 4.
 pub const DEFAULT_OBJECT_IO_RANDOM_READAHEAD_DISABLE_CONCURRENCY: usize = 4;
+
+#[cfg(test)]
+mod remote_version_state_tests {
+    #[test]
+    fn remote_version_state_gate_uses_stable_environment_names() {
+        assert_eq!(super::ENV_TIER_REMOTE_VERSION_STATE_WRITE, "RUSTFS_TIER_REMOTE_VERSION_STATE_WRITE");
+        assert_eq!(
+            super::ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED,
+            "RUSTFS_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED"
+        );
+    }
+}

crates/ecstore/src/api/mod.rs

@@ -384,6 +384,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/cluster/rpc/peer_rest_client.rs

@@ -224,6 +224,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>,
@@ -235,6 +250,7 @@ pub struct PeerLiveEventsBatch {
 pub struct PeerRestClient {
     pub host: XHost,
     pub grid_host: String,
+    topology_member: String,
     offline: Arc<AtomicBool>,
     recovery_running: Arc<AtomicBool>,
 }
@@ -327,9 +343,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)),
         }
@@ -349,7 +367,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
@@ -1161,6 +1183,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 {
@@ -2271,6 +2302,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 {
@@ -1886,6 +2091,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

@@ -2313,13 +2313,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,
@@ -2329,11 +2341,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(
@@ -2352,7 +2378,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))
@@ -2598,7 +2624,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;
@@ -2641,15 +2667,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}"
+            );
         }
     }
 
@@ -3954,20 +4000,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,
@@ -4083,6 +4133,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/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
@@ -1203,10 +1247,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;
@@ -1263,6 +1309,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();