path: root/src/swarm/members.rs

//! The member registry: every node that has joined the swarm, under
//! `_continuum/members/<pk>`, replicated with the document (and carried into new
//! epochs with the rest of `_continuum`).
//!
//! A pure registry: it decides nothing about admission (that's the swarm key). Each
//! node registers itself once it is ready, and re-registers if its entry disappears
//! while it is alive. An operator's forget (`DELETE /peers/{pk}`) removes the entry
//! together with writing the forget record; a forgotten node that turns out to be alive
//! cancels the record and registers again.
//!
//! Besides its tier, an entry holds the alias a node gives itself (`--alias`), a name
//! for people reading /peers and metrics (not unique and never an identity: the public
//! key stays that), where people reach its dashboard (`--dashboard-url`, `dashboard`),
//! where it can be reached (`addresses`, as its transport gives them:
//! hints for the others' transports), and an `alive` stamp the node renews now and
//! then: the members renewed within `--electorate-window` are the electorate of a
//! remake.

use std::collections::HashMap;
use std::time::Duration;

use serde_json::{Value as JsonValue, json};
use tracing::{debug, info, warn};

use crate::document::meta;
use crate::document::store::{self, StoreError};
use crate::document::{Document, ReadDoc, Transactable};
use crate::state::AppState;

pub(crate) fn members_dir() -> Vec<String> {
    meta::path(&[meta::MEMBERS])
}

pub(crate) fn member_path(pk: &str) -> Vec<String> {
    let mut path = members_dir();
    path.push(pk.to_owned());
    path
}

/// Public keys of all registered members, sorted.
pub(crate) fn members(doc: &impl ReadDoc) -> Vec<String> {
    let mut members = store::keys_at(doc, &members_dir()).unwrap_or_default();
    members.sort();
    members
}

/// A member's stability tier, judged by itself (`swarm::stability`).
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(crate) enum Tier {
    Stable,
    #[default]
    Transient,
}

impl Tier {
    pub(crate) fn as_str(self) -> &'static str {
        match self {
            Self::Stable => "stable",
            Self::Transient => "transient",
        }
    }
}

fn tier_path(pk: &str) -> Vec<String> {
    let mut path = member_path(pk);
    path.push("tier".to_owned());
    path
}

/// `pk`'s tier; transient unless its entry says stable.
pub(crate) fn tier(doc: &Document, pk: &str) -> Tier {
    match store::read_path(doc, &tier_path(pk)) {
        Ok(Some(JsonValue::String(tier))) if tier == "stable" => Tier::Stable,
        _ => Tier::Transient,
    }
}

pub(crate) fn set_tier(
    doc: &mut impl Transactable,
    pk: &str,
    tier: Tier,
) -> Result<(), StoreError> {
    store::write_path(doc, &tier_path(pk), &json!(tier.as_str()))
}

fn alias_path(pk: &str) -> Vec<String> {
    let mut path = member_path(pk);
    path.push("alias".to_owned());
    path
}

/// `pk`'s alias, if it gave itself one.
pub(crate) fn alias(doc: &Document, pk: &str) -> Option<String> {
    match store::read_path(doc, &alias_path(pk)) {
        Ok(Some(JsonValue::String(alias))) => Some(alias),
        _ => None,
    }
}

/// The aliases of all members that have one.
pub(crate) fn aliases(doc: &Document) -> HashMap<String, String> {
    members(doc)
        .into_iter()
        .filter_map(|pk| alias(doc, &pk).map(|alias| (pk, alias)))
        .collect()
}

fn set_alias(doc: &mut impl Transactable, pk: &str, alias: Option<&str>) -> Result<(), StoreError> {
    match alias {
        Some(alias) => store::write_path(doc, &alias_path(pk), &json!(alias)),
        None => match store::delete_path(doc, &alias_path(pk), false) {
            Err(StoreError::NotFound) => Ok(()),
            result => result,
        },
    }
}

fn dashboard_path(pk: &str) -> Vec<String> {
    let mut path = member_path(pk);
    path.push("dashboard".to_owned());
    path
}

/// Where people reach `pk`'s dashboard, if it says.
pub(crate) fn dashboard(doc: &Document, pk: &str) -> Option<String> {
    match store::read_path(doc, &dashboard_path(pk)) {
        Ok(Some(JsonValue::String(url))) => Some(url),
        _ => None,
    }
}

fn set_dashboard(
    doc: &mut impl Transactable,
    pk: &str,
    url: Option<&str>,
) -> Result<(), StoreError> {
    match url {
        Some(url) => store::write_path(doc, &dashboard_path(pk), &json!(url)),
        None => match store::delete_path(doc, &dashboard_path(pk), false) {
            Err(StoreError::NotFound) => Ok(()),
            result => result,
        },
    }
}

fn addresses_path(pk: &str) -> Vec<String> {
    let mut path = member_path(pk);
    path.push("addresses".to_owned());
    path
}

/// Addresses a member lists, at most: a few per transport (libp2p's up to
/// `PUBLISHED_DIRECT` and `PUBLISHED_RELAYED`, and where it relays, again as many to
/// reserve a slot at), taken in turns by transport (`Paths::local_addresses`).
const MAX_ADDRESSES: usize = 16;

/// Where `pk` says it can be reached (at most `MAX_ADDRESSES`); none if its entry
/// doesn't say.
pub(crate) fn addresses(doc: &Document, pk: &str) -> Vec<String> {
    match store::read_path(doc, &addresses_path(pk)) {
        Ok(Some(JsonValue::Array(items))) => items
            .into_iter()
            .filter_map(|item| match item {
                JsonValue::String(address) => Some(address),
                _ => None,
            })
            .take(MAX_ADDRESSES)
            .collect(),
        _ => Vec::new(),
    }
}

/// What the member list tells this node besides who its neighbours are: where members
/// are (`learn_addresses`), what they call themselves (`set_aliases`), and the relays
/// listed (`relays`). Done whenever the neighbours are recomputed (`topology::refresh`).
pub(crate) fn apply(state: &AppState, doc: &Document) {
    learn_addresses(state, doc);
    set_aliases(state, aliases(doc));
    state.transport.learn_relays(&relays(doc));
}

/// Hands the members' `aliases` to the logger, under every name that stands for each:
/// its node ID, and its addresses (`logging`, which renames them bare too).
fn set_aliases(state: &AppState, aliases: HashMap<String, String>) {
    let mut names = HashMap::new();
    for (member, alias) in aliases {
        for address in state.transport.identities_of(&member) {
            names.insert(address, alias.clone());
        }
        names.insert(member, alias);
    }
    crate::logging::set_aliases(names);
}

/// Hands every other member's addresses to the transport (`Transport::learn_addresses`):
/// the member list is how a node finds where members it never met are.
pub(crate) fn learn_addresses(state: &AppState, doc: &Document) {
    let me = state.local_id();
    for pk in members(doc) {
        if pk != me {
            state.transport.learn_addresses(&pk, &addresses(doc, &pk));
        }
    }
}

/// Relays listed for the swarm, at most this many.
const MAX_RELAYS: usize = 32;

/// The relays listed for the swarm under `_continuum/relays` (by `--protocol-writers`):
/// addresses (`libp2p:<multiaddr>`, with or without the relay's peer ID) of relays that
/// aren't members, e.g. nodes that only relay. Every member's transport may
/// pick them (`Transport::learn_relays`), after the relays it is configured with.
pub(crate) fn relays(doc: &Document) -> Vec<String> {
    match store::read_path(doc, &meta::path(&[meta::RELAYS])) {
        Ok(Some(JsonValue::Array(items))) => items
            .into_iter()
            .filter_map(|item| match item {
                JsonValue::String(address) => Some(address),
                _ => None,
            })
            .take(MAX_RELAYS)
            .collect(),
        _ => Vec::new(),
    }
}

fn alive_path(pk: &str) -> Vec<String> {
    let mut path = member_path(pk);
    path.push("alive".to_owned());
    path
}

fn millis_at(doc: &impl ReadDoc, path: &[String]) -> Option<u64> {
    match store::read_path(doc, path) {
        Ok(Some(JsonValue::Number(number))) => number.as_u64(),
        _ => None,
    }
}

/// When `pk` last said it is alive (Unix ms): its last renewal, or its registration.
pub(crate) fn alive_at(doc: &impl ReadDoc, pk: &str) -> Option<u64> {
    let mut since = member_path(pk);
    since.push("since".to_owned());
    millis_at(doc, &alive_path(pk)).max(millis_at(doc, &since))
}

/// The members that said they are alive within `window` before `now_ms` (sorted): the
/// electorate of a remake (`remake::coordinator`). A stamp further in the future than a
/// renewal interval (`alive_renewal`) doesn't count: its node's clock is that far ahead
/// (it warns about it, `swarm::skew`), and the stamp would never age out.
pub(crate) fn electorate(doc: &impl ReadDoc, now_ms: u64, window: Duration) -> Vec<String> {
    let oldest = now_ms.saturating_sub(window.as_millis() as u64);
    let newest = now_ms.saturating_add(alive_renewal(window).as_millis() as u64);
    members(doc)
        .into_iter()
        .filter(|pk| alive_at(doc, pk).is_some_and(|at| (oldest..=newest).contains(&at)))
        .collect()
}

fn window_path(pk: &str) -> Vec<String> {
    let mut path = member_path(pk);
    path.push("window".to_owned());
    path
}

/// The shortest `--electorate-window` among this node's (`own`) and those the members
/// publish in their entries (ms, `window`): every node renews its alive stamp by it
/// (`alive_renewal`), so a coordinator with a shorter window than others' still counts
/// them all. Below `MIN_ELECTORATE_WINDOW` (which no node accepts) counts as that.
pub(crate) fn shortest_window(doc: &impl ReadDoc, own: Duration) -> Duration {
    members(doc)
        .iter()
        .filter_map(|pk| millis_at(doc, &window_path(pk)))
        .map(|ms| Duration::from_millis(ms).max(MIN_ELECTORATE_WINDOW))
        .fold(own, Duration::min)
}

/// How often a node renews its `alive` stamp: often enough that a live node never
/// drops out of the electorate (a seventh of the shortest window, `shortest_window`),
/// but at most hourly and at least daily, so idle swarms stay quiet.
fn alive_renewal(window: Duration) -> Duration {
    (window / 7).clamp(RENEWAL_MOST_OFTEN, Duration::from_secs(24 * 3600))
}

const RENEWAL_MOST_OFTEN: Duration = Duration::from_secs(3600);

/// The shortest `--electorate-window`: seven renewals at most often, so a live node
/// renews within a seventh of it, as with any longer one.
pub(crate) const MIN_ELECTORATE_WINDOW: Duration =
    Duration::from_secs(7 * RENEWAL_MOST_OFTEN.as_secs());

/// Registers this node if it is ready and its entry is missing, keeps its alias and its
/// dashboard's URL as configured (set, changed or removed after a restart with another
/// `--alias` or `--dashboard-url`) and its
/// addresses as the transport gives them, and renews its `alive` stamp now and then
/// (see `electorate`).
pub(crate) async fn ensure_registered(state: &AppState) {
    if !state.is_ready().await {
        return;
    }
    let local = state.local_id().to_owned();
    let own = member_path(&local);
    let window = state.electorate_window.as_millis() as u64;
    let (
        registered,
        recorded_alias,
        recorded_dashboard,
        recorded_addresses,
        alive,
        recorded_window,
        shortest,
    ) = {
        let replica = state.doc.read().await;
        let registered = store::read_path(&replica.doc, &own)
            .ok()
            .flatten()
            .is_some();
        (
            registered,
            alias(&replica.doc, &local),
            dashboard(&replica.doc, &local),
            addresses(&replica.doc, &local),
            millis_at(&replica.doc, &alive_path(&local)),
            millis_at(&replica.doc, &window_path(&local)),
            shortest_window(&replica.doc, state.electorate_window),
        )
    };
    let wanted = state.alias.clone();
    let mut own_addresses = state.transport.local_addresses();
    own_addresses.truncate(MAX_ADDRESSES);
    let now = state.clock.now_millis().as_u64();

    if !registered {
        let mut entry = json!({
            "since": now,
            "alive": now,
            "addresses": own_addresses,
            "window": window,
        });
        if let Some(alias) = &wanted {
            entry["alias"] = json!(alias);
        }
        if let Some(url) = &state.dashboard_url {
            entry["dashboard"] = json!(url);
        }
        match state
            .mutate_doc(move |doc| store::write_path(doc, &own, &entry))
            .await
        {
            Ok(()) => info!(
                alias = wanted.as_deref(),
                "Registered in the swarm's member list"
            ),
            Err(error) => warn!(error = %error, "Failed to register in the member list"),
        }
        return;
    }

    if recorded_window != Some(window) {
        let path = window_path(&local);
        match state
            .mutate_doc(move |doc| store::write_path(doc, &path, &json!(window)))
            .await
        {
            Ok(()) => debug!(
                window_ms = window,
                "Electorate window published in the member list"
            ),
            Err(error) => warn!(error = %error, "Failed to record our electorate window"),
        }
    }

    let renewal = alive_renewal(shortest).as_millis() as u64;
    if alive.is_none_or(|at| now.saturating_sub(at) >= renewal) {
        let path = alive_path(&local);
        match state
            .mutate_doc(move |doc| store::write_path(doc, &path, &json!(now)))
            .await
        {
            Ok(()) => debug!("Renewed our alive stamp in the member list"),
            Err(error) => warn!(error = %error, "Failed to renew our alive stamp"),
        }
    }

    if recorded_addresses != own_addresses {
        let path = addresses_path(&local);
        let value = json!(own_addresses);
        match state
            .mutate_doc(move |doc| store::write_path(doc, &path, &value))
            .await
        {
            Ok(()) => info!(addresses = ?own_addresses, "Addresses updated in the member list"),
            Err(error) => warn!(error = %error, "Failed to record our addresses"),
        }
    }

    if recorded_dashboard != state.dashboard_url {
        let url = state.dashboard_url.clone();
        let local = local.clone();
        match state
            .mutate_doc(move |doc| set_dashboard(doc, &local, url.as_deref()))
            .await
        {
            Ok(()) => info!(
                url = state.dashboard_url.as_deref(),
                "Dashboard URL updated in the member list"
            ),
            Err(error) => warn!(error = %error, "Failed to record our dashboard URL"),
        }
    }

    if recorded_alias == wanted {
        return;
    }
    let alias = wanted.clone();
    match state
        .mutate_doc(move |doc| set_alias(doc, &local, alias.as_deref()))
        .await
    {
        Ok(()) => info!(
            alias = wanted.as_deref(),
            previous = recorded_alias.as_deref(),
            "Alias updated in the member list"
        ),
        Err(error) => warn!(error = %error, "Failed to record our alias"),
    }
}

/// Removes `pk`'s entry (an operator forgot it). Absent is fine.
pub(crate) fn unregister(doc: &mut impl Transactable, pk: &str) -> Result<(), StoreError> {
    match store::delete_path(doc, &member_path(pk), false) {
        Err(StoreError::NotFound) => Ok(()),
        result => result,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::document::testing::change;

    #[test]
    fn electorate_by_alive_stamp() {
        let day = 24 * 3600 * 1000;
        let now = 100 * day;
        let mut doc = Document::new();
        let entries = [
            ("fresh", json!({ "since": 1, "alive": now - day })),
            ("stale", json!({ "since": 1, "alive": now - 8 * day })),
            ("new", json!({ "since": now - day })),
            ("old", json!({ "since": 1 })),
            // A clock a day ahead is tolerated (one renewal interval); a week isn't.
            ("ahead", json!({ "since": 1, "alive": now + day })),
            ("future", json!({ "since": 1, "alive": now + 7 * day })),
        ];
        change(&mut doc, |doc| {
            for (pk, entry) in entries {
                store::write_path(doc, &member_path(pk), &entry).unwrap();
            }
        });
        assert_eq!(
            electorate(&doc, now, Duration::from_secs(7 * 24 * 3600)),
            ["ahead", "fresh", "new"]
        );
    }

    #[test]
    fn listed_relays() {
        let mut doc = Document::new();
        assert!(relays(&doc).is_empty());
        let relay = "libp2p:/ip6/fd00::1/tcp/9661";
        change(&mut doc, |doc| {
            store::write_path(doc, &meta::path(&[meta::RELAYS]), &json!([relay, 7])).unwrap()
        });
        assert_eq!(relays(&doc), [relay]); // Anything but text is ignored.
        change(&mut doc, |doc| {
            store::write_path(doc, &meta::path(&[meta::RELAYS]), &json!(relay)).unwrap()
        });
        assert!(relays(&doc).is_empty()); // Not a list.
    }

    #[test]
    fn shortest_published_window() {
        let day = Duration::from_secs(24 * 3600);
        let mut doc = Document::new();
        assert_eq!(shortest_window(&doc, 7 * day), 7 * day);
        change(&mut doc, |doc| {
            store::write_path(
                doc,
                &member_path("a"),
                &json!({ "window": 30 * 86_400_000_u64 }),
            )
            .unwrap();
            store::write_path(doc, &member_path("b"), &json!({ "window": 86_400_000 })).unwrap();
            store::write_path(doc, &member_path("c"), &json!({ "since": 1 })).unwrap();
        });
        assert_eq!(shortest_window(&doc, 7 * day), day);
        // Below what any node accepts: as the shortest it accepts.
        change(&mut doc, |doc| {
            store::write_path(doc, &window_path("c"), &json!(1)).unwrap();
        });
        assert_eq!(shortest_window(&doc, 7 * day), MIN_ELECTORATE_WINDOW);
    }

    #[test]
    fn renewal_interval() {
        let day = Duration::from_secs(24 * 3600);
        assert_eq!(alive_renewal(7 * day), day);
        assert_eq!(alive_renewal(70 * day), day);
        assert_eq!(alive_renewal(day), day / 7);
        // The hourly floor only matters for windows shorter than the option allows.
        let hour = Duration::from_secs(3600);
        assert_eq!(alive_renewal(2 * hour), hour);
    }
}