//! The member registry: every node that has joined the swarm, under //! `_continuum/members/`, 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), and an `alive` stamp the node renews now and then: the members //! renewed within `--electorate-window-days` are the electorate of a remake. use std::collections::HashMap; use std::time::Duration; use automerge::AutoCommit; use serde_json::{Value as JsonValue, json}; use tracing::{debug, info, warn}; use crate::document::store::{self, StoreError}; use crate::state::{AppState, unix_millis}; const META_ROOT: &str = "_continuum"; fn members_dir() -> Vec { vec![META_ROOT.to_owned(), "members".to_owned()] } pub(crate) fn member_path(pk: &str) -> Vec { let mut path = members_dir(); path.push(pk.to_owned()); path } /// Public keys of all registered members, sorted. pub(crate) fn members(doc: &AutoCommit) -> Vec { 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 { 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: &AutoCommit, 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 AutoCommit, pk: &str, tier: Tier) -> Result<(), StoreError> { store::write_path(doc, &tier_path(pk), &json!(tier.as_str())) } fn alias_path(pk: &str) -> Vec { 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: &AutoCommit, pk: &str) -> Option { 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: &AutoCommit) -> HashMap { members(doc) .into_iter() .filter_map(|pk| alias(doc, &pk).map(|alias| (pk, alias))) .collect() } fn set_alias(doc: &mut AutoCommit, 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 alive_path(pk: &str) -> Vec { let mut path = member_path(pk); path.push("alive".to_owned()); path } fn millis_at(doc: &AutoCommit, path: &[String]) -> Option { 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: &AutoCommit, pk: &str) -> Option { 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`). pub(crate) fn electorate(doc: &AutoCommit, now_ms: u64, window: Duration) -> Vec { let oldest = now_ms.saturating_sub(window.as_millis() as u64); members(doc) .into_iter() .filter(|pk| alive_at(doc, pk).is_some_and(|at| at >= oldest)) .collect() } /// How often a node renews its `alive` stamp: often enough that a live node never /// drops out of the electorate (a seventh of the window), but at most hourly and at /// least daily, so idle swarms stay quiet. fn alive_renewal(window: Duration) -> Duration { (window / 7).clamp(Duration::from_secs(3600), Duration::from_secs(24 * 3600)) } /// Registers this node if it is ready and its entry is missing, keeps its alias as /// configured (set, changed or removed after a restart with another `--alias`), 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 (registered, recorded_alias, alive) = { let replica = state.doc.read().await; let registered = store::read_path(&replica.doc, &own) .ok() .flatten() .is_some(); ( registered, alias(&replica.doc, &local), millis_at(&replica.doc, &alive_path(&local)), ) }; let wanted = state.alias.clone(); let now = unix_millis(); if !registered { let mut entry = json!({ "since": now, "alive": now }); if let Some(alias) = &wanted { entry["alias"] = json!(alias); } 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; } let renewal = alive_renewal(state.electorate_window).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_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 AutoCommit, 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::*; #[test] fn electorate_by_alive_stamp() { let day = 24 * 3600 * 1000; let now = 100 * day; let mut doc = AutoCommit::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 })), ]; for (pk, entry) in entries { store::write_path(&mut doc, &member_path(pk), &entry).unwrap(); } assert_eq!( electorate(&doc, now, Duration::from_secs(7 * 24 * 3600)), ["fresh", "new"] ); } #[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); } }