1 files changed, 262 insertions, 0 deletions
diff --git a/src/swarm/members.rs b/src/swarm/members.rs
new file mode 100644
index 0000000..221437e
--- /dev/null
+++ b/src/swarm/members.rs
@@ -0,0 +1,262 @@
+//! 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), 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<String> {
+    vec![META_ROOT.to_owned(), "members".to_owned()]
+}
+
+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: &AutoCommit) -> 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: &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<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: &AutoCommit, 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: &AutoCommit) -> HashMap<String, String> {
+    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<String> {
+    let mut path = member_path(pk);
+    path.push("alive".to_owned());
+    path
+}
+
+fn millis_at(doc: &AutoCommit, 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: &AutoCommit, 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`).
+pub(crate) fn electorate(doc: &AutoCommit, now_ms: u64, window: Duration) -> Vec<String> {
+    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);
+    }
+}