path: root/src/swarm/links.rs

//! 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<Inner>,
}

#[derive(Default)]
struct Inner {
    /// Our LINKS out, by member: its nonce.
    asked: HashMap<String, u64>,
    /// The latest LINKS_REPLY of each member: when, and its links.
    replies: HashMap<String, (Instant, Vec<Link>)>,
    /// When we last answered each member.
    answered: HashMap<String, Instant>,
}

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<Link>)> {
        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<Link> {
    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<String> {
    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<OtherPath> {
    let mut others: Vec<OtherPath> = 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<Duration> {
    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));
}