remake: decouple grace time from node's keepalive

1 files changed, 68 insertions, 5 deletions

diff --git a/src/swarm/members.rs b/src/swarm/members.rs
index 098c767..59c7571 100644
--- a/src/swarm/members.rs
+++ b/src/swarm/members.rs
@@ -225,9 +225,27 @@ pub(crate) fn electorate(doc: &impl ReadDoc, now_ms: u64, window: Duration) -> V
         .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(RENEWAL_MOST_OFTEN, Duration::from_secs(24 * 3600))
 }
@@ -249,7 +267,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()
@@ -260,6 +279,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();
@@ -268,7 +289,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);
         }
@@ -285,7 +311,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
@@ -377,6 +417,29 @@ mod tests {
     }
 
     #[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);