diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index 6dbc9314c..0ce25d848 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -219,35 +219,62 @@ impl PeerRestClient { recovery_running: Arc::new(AtomicBool::new(false)), } } - pub async fn new_clients(eps: EndpointServerPools) -> (Vec>, Vec>) { - if !runtime_sources::setup_is_dist_erasure().await { - return (Vec::new(), Vec::new()); - } - let hosts = eps.peer_grid_hosts_sorted(); - let mut remote = Vec::with_capacity(hosts.len()); - let mut all = vec![None; hosts.len()]; - for (i, peer) in hosts.into_iter().enumerate() { - if let Some((peer_host_port, grid_host)) = peer { - let host = match XHost::try_from(peer_host_port.clone()) { - Ok(host) => host, + fn build_clients_from_slots( + slots: Vec<(String, Option, bool)>, + ) -> (Vec>, Vec>, Vec) { + let mut remote = Vec::with_capacity(slots.len().saturating_sub(1)); + let mut all = vec![None; slots.len()]; + let mut remote_topology_hosts = Vec::with_capacity(slots.len().saturating_sub(1)); + + for (idx, (peer_host_port, grid_host, is_local)) in slots.into_iter().enumerate() { + if is_local { + continue; + } + + let client = match grid_host { + Some(grid_host) => match XHost::try_from(peer_host_port.clone()) { + Ok(host) => Some(PeerRestClient::new(host, grid_host)), Err(err) => { warn!(peer = %peer_host_port, "Xhost parse failed while constructing peer client: {err:?}"); - continue; + None } - }; - let client = PeerRestClient::new(host, grid_host); + }, + None => { + warn!(peer = %peer_host_port, "grid host is missing while constructing peer client"); + None + } + }; - all[i] = Some(client.clone()); - remote.push(Some(client)); - } + all[idx] = client.clone(); + remote.push(client); + remote_topology_hosts.push(peer_host_port); } + (remote, all, remote_topology_hosts) + } + + pub async fn new_clients(eps: EndpointServerPools) -> (Vec>, Vec>) { + let (remote, all, _) = Self::new_clients_with_topology(eps).await; + (remote, all) + } + + pub async fn new_clients_with_topology(eps: EndpointServerPools) -> (Vec>, Vec>, Vec) { + if !runtime_sources::setup_is_dist_erasure().await { + return (Vec::new(), Vec::new(), Vec::new()); + } + + let (remote, all, remote_topology_hosts) = Self::build_clients_from_slots(eps.peer_grid_host_slots_sorted()); + if all.len() != remote.len() + 1 { - warn!("Expected number of all hosts ({}) to be remote +1 ({})", all.len(), remote.len()); + warn!( + all_hosts = all.len(), + remote_slots = remote.len(), + "Expected number of all hosts to be remote slots + local node" + ); } - (remote, all) + (remote, all, remote_topology_hosts) } pub async fn get_client(&self) -> Result>> { @@ -1540,6 +1567,36 @@ mod tests { ) } + #[test] + fn build_clients_from_slots_preserves_missing_remote_topology_slots() { + let slots = vec![ + ("127.0.0.1:9000".to_string(), None, true), + ("127.0.0.1:9001".to_string(), Some("http://127.0.0.1:9001".to_string()), false), + ("127.0.0.1:notaport".to_string(), Some("http://127.0.0.1:notaport".to_string()), false), + ("127.0.0.1:9003".to_string(), None, false), + ]; + + let (remote, all, remote_topology_hosts) = PeerRestClient::build_clients_from_slots(slots); + + assert_eq!(remote.len(), 3, "local node is excluded but remote slots are not compacted away"); + assert_eq!(all.len(), 4, "all slots preserve the sorted cluster topology shape"); + assert_eq!( + remote_topology_hosts, + vec![ + "127.0.0.1:9001".to_string(), + "127.0.0.1:notaport".to_string(), + "127.0.0.1:9003".to_string() + ] + ); + assert!(remote[0].is_some(), "valid remote peer should get a client"); + assert!(remote[1].is_none(), "unparseable remote peer should remain observable as a missing slot"); + assert!(remote[2].is_none(), "missing grid host should remain observable as a missing slot"); + assert!(all[0].is_none(), "local node is represented by the local server_info row"); + assert!(all[1].is_some()); + assert!(all[2].is_none()); + assert!(all[3].is_none()); + } + #[test] fn scanner_activity_requires_restart_safe_peer_identity() { let missing_instance = ScannerActivityResponse { diff --git a/crates/ecstore/src/layout/endpoints.rs b/crates/ecstore/src/layout/endpoints.rs index 0d989217d..b8e380270 100644 --- a/crates/ecstore/src/layout/endpoints.rs +++ b/crates/ecstore/src/layout/endpoints.rs @@ -1058,9 +1058,21 @@ impl EndpointServerPools { } pub fn peer_grid_hosts_sorted(&self) -> Vec> { + self.peer_grid_host_slots_sorted() + .into_iter() + .map(|(peer, grid_host, is_local)| { + if is_local { + None + } else { + grid_host.map(|grid_host| (peer, grid_host)) + } + }) + .collect() + } + + pub fn peer_grid_host_slots_sorted(&self) -> Vec<(String, Option, bool)> { let (mut peers, local) = self.peers(); let mut grid_hosts = HashMap::with_capacity(peers.len()); - let mut ret = vec![None; peers.len()]; for ep in self.0.iter() { for endpoint in ep.endpoints.0.iter() { @@ -1073,17 +1085,14 @@ impl EndpointServerPools { peers.sort(); - for (i, peer) in peers.into_iter().enumerate() { - if local == peer { - continue; - } - let Some(grid_host) = grid_hosts.get(&peer) else { - continue; - }; - ret[i] = Some((peer, grid_host.clone())); - } - - ret + peers + .into_iter() + .map(|peer| { + let is_local = local == peer; + let grid_host = if is_local { None } else { grid_hosts.get(&peer).cloned() }; + (peer, grid_host, is_local) + }) + .collect() } pub fn peers(&self) -> (Vec, String) { diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index 67316c623..107f45ff2 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -95,16 +95,18 @@ pub fn get_global_notification_sys() -> Option> { pub struct NotificationSys { pub peer_clients: Vec>, pub all_peer_clients: Vec>, + peer_topology_hosts: Vec, peer_admin_caches: Vec>, } impl NotificationSys { pub async fn new(eps: EndpointServerPools) -> Self { - let (peer_clients, all_peer_clients) = PeerRestClient::new_clients(eps).await; + let (peer_clients, all_peer_clients, peer_topology_hosts) = PeerRestClient::new_clients_with_topology(eps).await; let peer_admin_caches = (0..peer_clients.len()).map(|_| Mutex::new(PeerAdminCache::new())).collect(); Self { peer_clients, all_peer_clients, + peer_topology_hosts, peer_admin_caches, } } @@ -379,22 +381,19 @@ impl NotificationSys { let peer_timeout = Duration::from_secs(5); for (idx, client) in self.peer_clients.iter().enumerate() { - let endpoints = endpoints.clone(); - let cache = self.peer_admin_caches.get(idx); + let host = self + .peer_topology_hosts + .get(idx) + .cloned() + .or_else(|| client.as_ref().map(|client| client.host.to_string())) + .unwrap_or_default(); futures.push(async move { - // `peer_clients` comes from `new_clients`, which only ever pushes - // `Some(client)` (local hosts are excluded, not slotted as - // `None`), so this branch is unreachable in practice. Kept as a - // defensive fallback: report an explicit `unknown` state rather - // than a blank `default()` entry, so it can never contribute a - // hollow row to `servers[]` (rustfs/backlog#1049 P3). let Some(client) = client else { - return ServerProperties { - state: ItemState::Unknown.to_string().to_owned(), - ..Default::default() + return PeerServerInfoProbe { + host, + result: Err(PeerServerInfoProbeFailure::NoClient), }; }; - let host = client.host.to_string(); // First attempt. A single evicted or half-open internode channel // is enough to fail one probe and, before retrying, would drop @@ -403,8 +402,7 @@ impl NotificationSys { // before falling back (rustfs/backlog#1049, P1-B). match timeout(peer_timeout, client.server_info()).await { Ok(Ok(info)) => { - update_server_info_cache(cache, &host, &info); - return info; + return PeerServerInfoProbe { host, result: Ok(info) }; } Ok(Err(err)) => debug!("peer {host} server_info failed (attempt 1/2): {err}"), Err(_) => debug!("peer {host} server_info timed out (attempt 1/2) after {peer_timeout:?}"), @@ -420,26 +418,29 @@ impl NotificationSys { // Second and final attempt on the fresh channel. match timeout(peer_timeout, client.server_info()).await { - Ok(Ok(info)) => { - update_server_info_cache(cache, &host, &info); - info - } + Ok(Ok(info)) => PeerServerInfoProbe { host, result: Ok(info) }, Ok(Err(err)) => { warn!("peer {host} server_info failed after retry: {err}"); let health = peer_disk_health(&host).await; - handle_server_info_failure(cache, &host, &endpoints, health.as_ref()) + PeerServerInfoProbe { + host, + result: Err(PeerServerInfoProbeFailure::Rpc { health }), + } } Err(_) => { warn!("peer {host} server_info timed out after retry ({peer_timeout:?})"); client.evict_connection().await; let health = peer_disk_health(&host).await; - handle_server_info_failure(cache, &host, &endpoints, health.as_ref()) + PeerServerInfoProbe { + host, + result: Err(PeerServerInfoProbeFailure::Rpc { health }), + } } } }); } - join_all(futures).await + publish_server_info_probe_round(&self.peer_admin_caches, &endpoints, join_all(futures).await) } pub async fn load_user(&self, access_key: &str, temp: bool) -> Vec { @@ -1263,6 +1264,16 @@ struct PeerDiskHealth { disks: Vec, } +struct PeerServerInfoProbe { + host: String, + result: std::result::Result, +} + +enum PeerServerInfoProbeFailure { + Rpc { health: Option }, + NoClient, +} + /// Consult the local disk-health state for `host` without issuing any RPC. /// /// On the aggregating node a peer's drives are remote-disk handles whose @@ -1424,6 +1435,30 @@ fn handle_server_info_failure( unknown_server_properties(host, endpoints) } +fn publish_server_info_probe_round( + caches: &[Mutex], + endpoints: &EndpointServerPools, + probes: Vec, +) -> Vec { + probes + .into_iter() + .enumerate() + .map(|(idx, probe)| { + let cache = caches.get(idx); + match probe.result { + Ok(info) => { + update_server_info_cache(cache, &probe.host, &info); + info + } + Err(PeerServerInfoProbeFailure::Rpc { health }) => { + handle_server_info_failure(cache, &probe.host, endpoints, health.as_ref()) + } + Err(PeerServerInfoProbeFailure::NoClient) => unknown_server_properties(&probe.host, endpoints), + } + }) + .collect() +} + fn update_server_info_cache(cache: Option<&Mutex>, host: &str, info: &ServerProperties) { let Some(cache) = cache else { return; @@ -1634,6 +1669,7 @@ mod tests { "127.0.0.1:9000".to_string().try_into().expect("peer host should parse"), "http://127.0.0.1:9000".to_string(), ))], + peer_topology_hosts: Vec::new(), peer_admin_caches: Vec::new(), }; @@ -1677,6 +1713,7 @@ mod tests { let sys = NotificationSys { peer_clients: vec![None], all_peer_clients: Vec::new(), + peer_topology_hosts: vec!["node-a:9000".to_string()], peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], }; @@ -1696,6 +1733,7 @@ mod tests { let sys = NotificationSys { peer_clients: vec![None], all_peer_clients: vec![None, None], + peer_topology_hosts: vec!["node-a:9000".to_string()], peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], }; @@ -1712,6 +1750,7 @@ mod tests { let sys = NotificationSys { peer_clients: Vec::new(), all_peer_clients: Vec::new(), + peer_topology_hosts: Vec::new(), peer_admin_caches: Vec::new(), }; @@ -1732,6 +1771,7 @@ mod tests { let sys = NotificationSys { peer_clients: vec![Some(client)], all_peer_clients: vec![None], + peer_topology_hosts: vec!["127.0.0.1:9000".to_string()], peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], }; @@ -1743,6 +1783,51 @@ mod tests { assert!(err.to_string().contains("peer topology is incomplete")); } + #[tokio::test] + async fn server_info_no_client_slot_uses_topology_host_without_counting_rpc_failure() { + let sys = NotificationSys { + peer_clients: vec![None], + all_peer_clients: vec![None, None], + peer_topology_hosts: vec!["node-a:9000".to_string()], + peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], + }; + + let servers = sys.server_info().await; + + assert_eq!(servers.len(), 1); + assert_eq!(servers[0].endpoint, "node-a:9000"); + assert_eq!(servers[0].state, ItemState::Unknown.to_string()); + let cache = sys.peer_admin_caches[0].lock().expect("cache mutex should not be poisoned"); + assert_eq!(cache.server_failures, 0, "construction-only missing slots are not failed RPC attempts"); + assert!(cache.last_server_info.is_none()); + } + + #[test] + fn server_info_failure_cache_stays_aligned_with_topology_slot() { + let cache_a = Mutex::new(PeerAdminCache { + last_server_info: Some(build_props("cached-a")), + last_server_success: Some(SystemTime::now()), + server_failures: 1, + storage_failures: 0, + last_storage_info: None, + }); + let cache_b = Mutex::new(PeerAdminCache { + last_server_info: Some(build_props("cached-b")), + last_server_success: Some(SystemTime::now()), + server_failures: 1, + storage_failures: 0, + last_storage_info: None, + }); + let caches = [cache_a, cache_b]; + let endpoints = EndpointServerPools::from(Vec::new()); + + let rendered = handle_server_info_failure(Some(&caches[1]), "node-b:9000", &endpoints, None); + + assert_eq!(rendered.endpoint, "cached-b"); + assert_eq!(caches[0].lock().expect("cache mutex should not be poisoned").server_failures, 1); + assert_eq!(caches[1].lock().expect("cache mutex should not be poisoned").server_failures, 2); + } + #[tokio::test] async fn scanner_activity_probe_times_out() { let err = scanner_activity_with_timeout( @@ -1762,6 +1847,7 @@ mod tests { let sys = NotificationSys { peer_clients: vec![None], all_peer_clients: Vec::new(), + peer_topology_hosts: vec!["node-a:9000".to_string()], peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], }; @@ -1781,6 +1867,7 @@ mod tests { let sys = NotificationSys { peer_clients: vec![None], all_peer_clients: Vec::new(), + peer_topology_hosts: vec!["node-a:9000".to_string()], peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], }; @@ -1796,6 +1883,7 @@ mod tests { let sys = NotificationSys { peer_clients: vec![None], all_peer_clients: Vec::new(), + peer_topology_hosts: vec!["node-a:9000".to_string()], peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], }; let mutation_id = Uuid::from_u128(1); @@ -1996,6 +2084,60 @@ mod tests { assert!(!cached_snapshot_is_fresh(Some(stale)), "an old success is stale"); } + #[test] + fn server_info_probe_round_commits_failures_only_when_published() { + let caches = vec![Mutex::new(PeerAdminCache::new())]; + let endpoints = EndpointServerPools::default(); + let probes = vec![PeerServerInfoProbe { + host: "peer-1".to_string(), + result: Err(PeerServerInfoProbeFailure::Rpc { health: None }), + }]; + + assert_eq!( + caches[0] + .lock() + .expect("peer cache should lock before publish") + .server_failures, + 0 + ); + + let replies = publish_server_info_probe_round(&caches, &endpoints, probes); + + assert_eq!(replies.len(), 1); + assert_eq!(replies[0].endpoint, "peer-1"); + assert_eq!(replies[0].state, ItemState::Unknown.to_string()); + assert_eq!( + caches[0] + .lock() + .expect("peer cache should lock after publish") + .server_failures, + 1 + ); + } + + #[test] + fn server_info_probe_round_does_not_count_no_client_slots_as_rpc_failures() { + let caches = vec![Mutex::new(PeerAdminCache::new())]; + let endpoints = EndpointServerPools::default(); + let probes = vec![PeerServerInfoProbe { + host: "node-a:9000".to_string(), + result: Err(PeerServerInfoProbeFailure::NoClient), + }]; + + let replies = publish_server_info_probe_round(&caches, &endpoints, probes); + + assert_eq!(replies.len(), 1); + assert_eq!(replies[0].endpoint, "node-a:9000"); + assert_eq!(replies[0].state, ItemState::Unknown.to_string()); + assert_eq!( + caches[0] + .lock() + .expect("peer cache should lock after no-client publish") + .server_failures, + 0 + ); + } + #[test] fn endpoint_host_matches_direct_and_canonicalized() { // Direct match (IP deployment): peer host already equals host_port.