//! How members are linked, for a map of the swarm (the dashboard, `api::dashboard`): a //! node asks a member with LINKS, and the member answers with a LINKS_REPLY listing its //! neighbours, each with the transport its messages go over, whether only through a //! relay, the round trip and whether the neighbour is dormant: as the member measures //! them, which no other node can. //! //! Every node answers, whatever its build: one with a dashboard asks any member, which //! may have none. Only members (in the member list) are answered, as the links name //! members and where they're reached; and each at most every `ANSWER_EVERY`. Nothing //! is written to the document: asked for while someone looks, forgotten after. use std::collections::HashMap; use std::sync::Mutex; use std::time::Duration; use tokio::time::Instant; use tracing::debug; use crate::locks::MutexExt; use crate::state::AppState; use crate::swarm::{members, topology}; use crate::transport::{ Connection, Destination, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage, }; /// Links one reply lists at most (a member has about `--neighbours`, more if others /// prefer it). pub(crate) const MAX_LINKS: usize = 256; /// Other transports in a link at most. pub(crate) const MAX_OTHERS: usize = 8; /// A member answers each other one at most this often. const ANSWER_EVERY: Duration = Duration::from_secs(5); /// Replies kept from members at most (one each), and requests out. const MAX_KEPT: usize = 4096; /// The requests out and the replies in, and when each asker was last answered. #[derive(Default)] pub(crate) struct Links { inner: Mutex, } #[derive(Default)] struct Inner { /// Our LINKS out, by member: its nonce. asked: HashMap, /// The latest LINKS_REPLY of each member: when, and its links. replies: HashMap)>, /// When we last answered each member. answered: HashMap, } impl Links { /// The latest links each member replied with, and how long ago. #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] pub(crate) fn replies(&self) -> Vec<(String, Duration, Vec)> { let now = Instant::now(); self.inner .lock_or_recover() .replies .iter() .map(|(member, (at, links))| { ( member.clone(), now.saturating_duration_since(*at), links.clone(), ) }) .collect() } } /// This node's links: its neighbours, as it measures them. pub(crate) async fn own_links(state: &AppState) -> Vec { let mut links = Vec::new(); for (peer, peer_state) in state.peers_snapshot().await { if !topology::is_neighbour(state, &peer) { continue; } let dormant = peer_state.lock().await.liveness.is_dormant(); let destination = state.transport.destination(&peer); let transport = destination .as_ref() .and_then(|destination| scheme_of(&destination.address)); links.push(Link { others: others(destination.as_ref(), transport.as_deref()), transport, relayed: destination .as_ref() .is_some_and(|destination| destination.connection == Some(Connection::Relayed)), rtt_us: rtt_of(destination.as_ref()) .or_else(|| state.transport.rtt(&peer)) .map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)), dormant, peer, }); } links.sort_by(|a, b| a.peer.cmp(&b.peer)); links.truncate(MAX_LINKS); links } /// The scheme of `address` (`mycelium:…`): its transport. fn scheme_of(address: &str) -> Option { address.split_once(':').map(|(scheme, _)| scheme.to_owned()) } /// The transports besides `used` the peer proved itself at, with a path up: each with /// the round trip of its fastest path, if any is measured. fn others(destination: Option<&Destination>, used: Option<&str>) -> Vec { let mut others: Vec = Vec::new(); for path in destination.map_or(&[][..], |destination| &destination.paths[..]) { let Some(transport) = scheme_of(&path.address) else { continue; }; if path.down || used == Some(transport.as_str()) { continue; } let rtt_us = path .rtt_ms .filter(|ms| ms.is_finite() && *ms >= 0.0) .map(|ms| (ms * 1000.0).min(f64::from(u32::MAX)) as u32); match others.iter_mut().find(|other| other.transport == transport) { Some(other) => { other.rtt_us = match (other.rtt_us, rtt_us) { (Some(a), Some(b)) => Some(a.min(b)), (a, b) => a.or(b), }; } None => others.push(OtherPath { transport, rtt_us }), } } others.sort_by(|a, b| a.transport.cmp(&b.transport)); others.truncate(MAX_OTHERS); others } /// The round-trip time of the path messages to the peer go over (`destination`'s /// first path, best first), if measured: `Transport::rtt` takes the first address the /// peer proved itself at, which needn't be the one in use. fn rtt_of(destination: Option<&Destination>) -> Option { let path = destination?.paths.first()?; path.rtt_ms .filter(|ms| ms.is_finite() && *ms >= 0.0) .map(|ms| Duration::from_secs_f64(ms / 1000.0)) } /// Asks `member` for its links (a LINKS); its reply lands in `AppState::links`. #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] pub(crate) async fn ask(state: &AppState, member: &str) { let nonce = crate::transport::paths::random_nonce(); { let mut inner = state.links.inner.lock_or_recover(); if inner.asked.len() >= MAX_KEPT && !inner.asked.contains_key(member) { inner.asked.clear(); } inner.asked.insert(member.to_owned(), nonce); } let (epoch, epoch_seq) = state.epoch_stamp(); let request = ParsedMessage::Links(LinksRequest { epoch, epoch_seq, nonce, }); if let Err(error) = state.transport.send_message(member, &request).await { debug!(member = %member, error = %format!("{error:#}"), "LINKS failed"); } } /// A LINKS from `peer`: answered if it's a member, and not answered lately. pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) { let member = { let replica = state.doc.read().await; members::members(&replica.doc) .iter() .any(|member| member == peer) }; if !member { debug!(peer = %peer, "Not answering LINKS from a peer that isn't a member"); return; } { let now = Instant::now(); let mut inner = state.links.inner.lock_or_recover(); if inner .answered .get(peer) .is_some_and(|at| now.saturating_duration_since(*at) < ANSWER_EVERY) { return; } if inner.answered.len() >= MAX_KEPT { inner .answered .retain(|_, at| now.saturating_duration_since(*at) < ANSWER_EVERY); } inner.answered.insert(peer.to_owned(), now); } let (epoch, epoch_seq) = state.epoch_stamp(); let reply = ParsedMessage::LinksReply(LinksReply { epoch, epoch_seq, nonce: request.nonce, links: own_links(state).await, }); if let Err(error) = state.transport.send_message(peer, &reply).await { debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed"); } } /// A LINKS_REPLY from `peer`: kept if it answers our LINKS to it. pub(crate) fn received(state: &AppState, peer: &str, mut reply: LinksReply) { let mut inner = state.links.inner.lock_or_recover(); if inner.asked.get(peer) != Some(&reply.nonce) { return; } inner.asked.remove(peer); reply.links.truncate(MAX_LINKS); for link in &mut reply.links { link.others.truncate(MAX_OTHERS); } if inner.replies.len() >= MAX_KEPT && !inner.replies.contains_key(peer) { inner.replies.clear(); } inner .replies .insert(peer.to_owned(), (Instant::now(), reply.links)); }