diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index ab92ec0fb..bab33270b 100644 --- a/crates/config/src/constants/object.rs +++ b/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" + ); + } +} diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 25ea74b09..e2b25871b 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/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, }; } diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index 9be63f6a4..17c343e78 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/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 { + 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, @@ -235,6 +250,7 @@ pub struct PeerLiveEventsBatch { pub struct PeerRestClient { pub host: XHost, pub grid_host: String, + topology_member: String, offline: Arc, recovery_running: Arc, } @@ -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, diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index 13ef3e09a..13f453f6b 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/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> = OnceLock::new(); } +#[derive(Clone)] +struct RemoteVersionStateFleetProof { + topology_fingerprint: String, + peer_epochs: Arc>, + 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>, +} + +#[derive(Default)] +struct RemoteVersionStateFleetProofState { + proof: Option, + topology_conflict: bool, +} + +static REMOTE_VERSION_STATE_FLEET_PROOF: OnceLock> = OnceLock::new(); +static REMOTE_VERSION_STATE_PROBE_TOPOLOGY: OnceLock = OnceLock::new(); + +fn remote_version_state_fleet_proof_slot() -> &'static std::sync::RwLock { + REMOTE_VERSION_STATE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(RemoteVersionStateFleetProofState::default())) +} + +fn replace_remote_version_state_fleet_proof(proof: Option) { + 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, + proof: Option, +) { + slot.write().unwrap_or_else(std::sync::PoisonError::into_inner).proof = proof; +} + +fn publish_remote_version_state_probe_result( + slot: &std::sync::RwLock, + topology_fingerprint: &str, + result: Result>, + observed_at: Instant, +) -> Option { + 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 { + 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 { + 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(¤t.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, 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::>(); 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> { + 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(), diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index c15ba65c7..0f22d07ad 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/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, 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 { + 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, diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index 165ac3c52..ba4f53d07 100644 --- a/crates/protos/src/lib.rs +++ b/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 { + 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, 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(); diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index a7d2b272c..28e5e15cd 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -49,7 +49,7 @@ use std::{ collections::HashMap, io::Cursor, pin::Pin, - sync::{Arc, OnceLock}, + sync::{Arc, LazyLock, OnceLock}, }; use time::OffsetDateTime; use tokio::spawn; @@ -116,6 +116,7 @@ fn remove_heal_control_replay( } static HEAL_CONTROL_REPLAY_CACHE: OnceLock>>> = OnceLock::new(); +static NODE_CAPABILITY_SERVER_EPOCH: LazyLock = LazyLock::new(Uuid::new_v4); fn admit_heal_control_replay( replay_cache: &mut HashMap>, @@ -448,6 +449,19 @@ impl HealControlRpcService { pub(crate) async fn initialize_heal_topology_fingerprint( cache: Arc>, endpoint_pools: EndpointServerPools, +) -> Result<(), String> { + initialize_heal_topology_fingerprint_with_probe( + cache, + endpoint_pools, + crate::storage::storage_api::start_remote_version_state_fleet_probe, + ) + .await +} + +async fn initialize_heal_topology_fingerprint_with_probe( + cache: Arc>, + endpoint_pools: EndpointServerPools, + start_probe: impl FnOnce(String), ) -> Result<(), String> { if cache.get().is_some() { return Ok(()); @@ -455,7 +469,8 @@ pub(crate) async fn initialize_heal_topology_fingerprint( let fingerprint = tokio::task::spawn_blocking(move || heal::heal_topology_fingerprint(&endpoint_pools)) .await .map_err(|_| "heal control topology calculation task failed".to_string())??; - let _ = cache.set(fingerprint); + let _ = cache.set(fingerprint.clone()); + start_probe(fingerprint); Ok(()) } @@ -789,6 +804,35 @@ impl heal_control_service_server::HealControlService for HealControlRpcService { response_proof: Bytes::new(), })); } + if rustfs_protos::is_remote_version_state_capability_probe(&request.get_ref().command) { + let topology_member = self + .endpoint_pools() + .await + .ok_or_else(|| Status::failed_precondition("heal control topology is not initialized"))? + .peers() + .1; + if topology_member.is_empty() { + return Err(Status::failed_precondition("local topology member identity is unavailable")); + } + let result = + rustfs_protos::encode_remote_version_state_capability(&topology_member, NODE_CAPABILITY_SERVER_EPOCH.as_bytes()) + .map_err(|_| Status::internal("remote version state capability length cannot be represented"))?; + let canonical_response = rustfs_protos::canonical_heal_control_response_body( + request.get_ref().version, + &request.get_ref().topology_fingerprint, + &request.get_ref().command, + &result, + ) + .map_err(|_| Status::internal("heal control response length cannot be represented"))?; + let response_proof = sign_tonic_rpc_response_proof(&canonical_response) + .map_err(|_| Status::internal("heal control response proof is unavailable"))?; + return Ok(Response::new(HealControlResponse { + success: true, + result: result.into(), + error_info: None, + response_proof: response_proof.into(), + })); + } let endpoints = self .endpoint_pools() .await @@ -2027,10 +2071,10 @@ mod tests { PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, STORAGE_CLASS_SUB_SYS, admit_heal_control_replay, background_rebalance_start_error_message, - execute_heal_control_envelope_with_manager, initialize_heal_topology_fingerprint, legacy_scanner_activity_response, - make_heal_control_server, make_heal_control_server_with_cache, make_server, make_server_for_context, - make_tier_mutation_control_server_for_context, previous_scanner_activity_response, remove_heal_control_replay, - scanner_activity_response, stop_rebalance_response, + execute_heal_control_envelope_with_manager, initialize_heal_topology_fingerprint, + initialize_heal_topology_fingerprint_with_probe, legacy_scanner_activity_response, make_heal_control_server, + make_heal_control_server_with_cache, make_server, make_server_for_context, make_tier_mutation_control_server_for_context, + previous_scanner_activity_response, remove_heal_control_replay, scanner_activity_response, stop_rebalance_response, }; use crate::storage::rpc::node_service::heal::heal_topology_fingerprint; use crate::storage::storage_api::rpc_consumer::node_service::{HealBucketInfo, HealEndpoint}; @@ -2078,6 +2122,7 @@ mod tests { use tokio::time::Duration; use tokio_stream::wrappers::TcpListenerStream; use tonic::{Request, Response, Status}; + use uuid::Uuid; struct HealControlMockStorage; @@ -2958,6 +3003,60 @@ mod tests { assert_eq!(non_coordinator.code(), tonic::Code::FailedPrecondition); } + #[tokio::test] + async fn remote_version_state_probe_authenticates_topology_challenge_and_process_epoch() { + let _ = rustfs_credentials::set_global_rpc_secret("remote-version-state-node-service-test-secret".to_string()); + let endpoints = heal_control_test_endpoints_with_coordinator("node-d", true); + let fingerprint = heal_topology_fingerprint(&endpoints).expect("test topology should hash"); + let (service, source) = super::make_heal_control_server_for_source(); + *source.write().await = Some(endpoints); + let probe_command = rustfs_protos::remote_version_state_capability_probe(&[7; 16]); + let mut request = Request::new(HealControlRequest { + version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + topology_fingerprint: fingerprint.clone(), + command: Bytes::from(probe_command.clone()), + }); + let body = rustfs_protos::canonical_heal_control_request_body( + request.get_ref().version, + &request.get_ref().topology_fingerprint, + &request.get_ref().command, + ) + .expect("probe should encode"); + set_tonic_canonical_body_digest(&mut request, &body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut request); + let response = service + .heal_control(request) + .await + .expect("matching topology should be acknowledged") + .into_inner(); + + let (topology_member, process_epoch) = + rustfs_protos::decode_remote_version_state_capability(&response.result).expect("capability response should decode"); + assert_eq!(topology_member, "node-a:9000"); + let server_epoch = Uuid::from_slice(process_epoch).expect("server epoch should be a UUID"); + assert!(!server_epoch.is_nil()); + let canonical_response = rustfs_protos::canonical_heal_control_response_body( + rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + &fingerprint, + &probe_command, + &response.result, + ) + .expect("response should encode"); + crate::storage::storage_api::verify_tonic_rpc_response_proof(&canonical_response, &response.response_proof) + .expect("outer proof should bind the response to the request"); + + let different_probe = rustfs_protos::remote_version_state_capability_probe(&[8; 16]); + let different_response = rustfs_protos::canonical_heal_control_response_body( + rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + &fingerprint, + &different_probe, + &response.result, + ) + .expect("different response should encode"); + crate::storage::storage_api::verify_tonic_rpc_response_proof(&different_response, &response.response_proof) + .expect_err("proof from one challenge must not be reusable"); + } + #[tokio::test] async fn heal_control_coordinator_rejects_expired_and_non_admin_starts() { let _ = rustfs_credentials::set_global_rpc_secret("heal-control-node-service-test-secret".to_string()); @@ -3020,10 +3119,15 @@ mod tests { let topology = heal_control_test_endpoints("node-d"); let expected = heal_topology_fingerprint(&topology).expect("test topology should hash"); let cache = Arc::new(tokio::sync::OnceCell::new()); - initialize_heal_topology_fingerprint(Arc::clone(&cache), topology) - .await - .expect("valid topology should initialize"); + let started_probe = Arc::new(std::sync::Mutex::new(None)); + let started_probe_capture = Arc::clone(&started_probe); + initialize_heal_topology_fingerprint_with_probe(Arc::clone(&cache), topology, move |fingerprint| { + *started_probe_capture.lock().expect("probe capture should not poison") = Some(fingerprint); + }) + .await + .expect("valid topology should initialize"); assert_eq!(cache.get(), Some(&expected)); + assert_eq!(started_probe.lock().expect("probe capture should not poison").as_ref(), Some(&expected)); let mut invalid = heal_control_test_endpoints("node-d"); invalid.as_mut()[0].endpoints.as_mut()[0].pool_idx = -1; diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 1735ce2b4..cf0074d3d 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -472,7 +472,7 @@ pub(crate) mod ecstore_metrics { #[allow(unused_imports)] pub(crate) mod ecstore_notification { pub(crate) use rustfs_ecstore::api::notification::{ - NotificationSys, get_global_notification_sys, new_global_notification_sys, + NotificationSys, get_global_notification_sys, new_global_notification_sys, start_remote_version_state_fleet_probe, }; } @@ -962,6 +962,10 @@ pub(crate) async fn new_global_notification_sys(endpoint_pools: EndpointServerPo ecstore_notification::new_global_notification_sys(endpoint_pools).await } +pub(crate) fn start_remote_version_state_fleet_probe(topology_fingerprint: String) { + ecstore_notification::start_remote_version_state_fleet_probe(topology_fingerprint); +} + pub(crate) async fn read_config(api: Arc, file: &str) -> Result> { ecstore_config::com::read_config(api, file).await }