1 files changed, 143 insertions, 32 deletions

diff --git a/src/swarm/members.rs b/src/swarm/members.rs
index 3bfb9cf..05c6501 100644
--- a/src/swarm/members.rs
+++ b/src/swarm/members.rs
@@ -18,12 +18,12 @@
 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::meta;
 use crate::document::store::{self, StoreError};
+use crate::document::{Document, ReadDoc, Transactable};
 use crate::state::AppState;
 
 pub(crate) fn members_dir() -> Vec<String> {
@@ -37,7 +37,7 @@ pub(crate) fn member_path(pk: &str) -> Vec<String> {
 }
 
 /// Public keys of all registered members, sorted.
-pub(crate) fn members(doc: &AutoCommit) -> Vec<String> {
+pub(crate) fn members(doc: &impl ReadDoc) -> Vec<String> {
     let mut members = store::keys_at(doc, &members_dir()).unwrap_or_default();
     members.sort();
     members
@@ -67,14 +67,18 @@ fn tier_path(pk: &str) -> Vec<String> {
 }
 
 /// `pk`'s tier; transient unless its entry says stable.
-pub(crate) fn tier(doc: &AutoCommit, pk: &str) -> Tier {
+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 AutoCommit, pk: &str, tier: Tier) -> Result<(), StoreError> {
+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()))
 }
 
@@ -85,7 +89,7 @@ fn alias_path(pk: &str) -> Vec<String> {
 }
 
 /// `pk`'s alias, if it gave itself one.
-pub(crate) fn alias(doc: &AutoCommit, pk: &str) -> Option<String> {
+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,
@@ -93,14 +97,14 @@ pub(crate) fn alias(doc: &AutoCommit, pk: &str) -> Option<String> {
 }
 
 /// The aliases of all members that have one.
-pub(crate) fn aliases(doc: &AutoCommit) -> HashMap<String, String> {
+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 AutoCommit, pk: &str, alias: Option<&str>) -> Result<(), StoreError> {
+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) {
@@ -116,12 +120,14 @@ fn addresses_path(pk: &str) -> Vec<String> {
     path
 }
 
-/// Addresses a member most likely has, at most: one per transport.
-const MAX_ADDRESSES: usize = 8;
+/// 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: &AutoCommit, pk: &str) -> Vec<String> {
+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()
@@ -135,9 +141,31 @@ pub(crate) fn addresses(doc: &AutoCommit, pk: &str) -> Vec<String> {
     }
 }
 
+/// 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: &AutoCommit) {
+pub(crate) fn learn_addresses(state: &AppState, doc: &Document) {
     let me = state.local_id();
     for pk in members(doc) {
         if pk != me {
@@ -153,7 +181,7 @@ const MAX_RELAYS: usize = 32;
 /// 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: &AutoCommit) -> Vec<String> {
+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()
@@ -173,7 +201,7 @@ fn alive_path(pk: &str) -> Vec<String> {
     path
 }
 
-fn millis_at(doc: &AutoCommit, path: &[String]) -> Option<u64> {
+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,
@@ -181,29 +209,57 @@ fn millis_at(doc: &AutoCommit, path: &[String]) -> Option<u64> {
 }
 
 /// 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> {
+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`).
-pub(crate) fn electorate(doc: &AutoCommit, now_ms: u64, window: Duration) -> Vec<String> {
+/// 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| at >= oldest))
+        .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 window), but at most hourly and at
-/// least daily, so idle swarms stay quiet.
+/// 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(Duration::from_secs(3600), Duration::from_secs(24 * 3600))
+    (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 as
 /// configured (set, changed or removed after a restart with another `--alias`) and its
 /// addresses as the transport gives them, and renews its `alive` stamp now and then
@@ -214,7 +270,8 @@ pub(crate) async fn ensure_registered(state: &AppState) {
     }
     let local = state.local_id().to_owned();
     let own = member_path(&local);
-    let (registered, recorded_alias, recorded_addresses, alive) = {
+    let window = state.electorate_window.as_millis() as u64;
+    let (registered, recorded_alias, recorded_addresses, alive, recorded_window, shortest) = {
         let replica = state.doc.read().await;
         let registered = store::read_path(&replica.doc, &own)
             .ok()
@@ -225,6 +282,8 @@ pub(crate) async fn ensure_registered(state: &AppState) {
             alias(&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();
@@ -233,7 +292,12 @@ pub(crate) async fn ensure_registered(state: &AppState) {
     let now = state.clock.now_millis().as_u64();
 
     if !registered {
-        let mut entry = json!({ "since": now, "alive": now, "addresses": own_addresses });
+        let mut entry = json!({
+            "since": now,
+            "alive": now,
+            "addresses": own_addresses,
+            "window": window,
+        });
         if let Some(alias) = &wanted {
             entry["alias"] = json!(alias);
         }
@@ -250,7 +314,21 @@ pub(crate) async fn ensure_registered(state: &AppState) {
         return;
     }
 
-    let renewal = alive_renewal(state.electorate_window).as_millis() as u64;
+    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
@@ -292,7 +370,7 @@ pub(crate) async fn ensure_registered(state: &AppState) {
 }
 
 /// Removes `pk`'s entry (an operator forgot it). Absent is fine.
-pub(crate) fn unregister(doc: &mut AutoCommit, pk: &str) -> Result<(), StoreError> {
+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,
@@ -302,39 +380,72 @@ pub(crate) fn unregister(doc: &mut AutoCommit, pk: &str) -> Result<(), StoreErro
 #[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 = AutoCommit::new();
+        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 })),
         ];
-        for (pk, entry) in entries {
-            store::write_path(&mut doc, &member_path(pk), &entry).unwrap();
-        }
+        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)),
-            ["fresh", "new"]
+            ["ahead", "fresh", "new"]
         );
     }
 
     #[test]
     fn listed_relays() {
-        let mut doc = AutoCommit::new();
+        let mut doc = Document::new();
         assert!(relays(&doc).is_empty());
         let relay = "libp2p:/ip6/fd00::1/tcp/9661";
-        store::write_path(&mut doc, &meta::path(&[meta::RELAYS]), &json!([relay, 7])).unwrap();
+        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.
-        store::write_path(&mut doc, &meta::path(&[meta::RELAYS]), &json!(relay)).unwrap();
+        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);