don't choose stale neighbours from member list

1 files changed, 23 insertions, 6 deletions

diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index 2c6b867..c9d1266 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -143,10 +143,18 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
     if !state.is_ready().await {
         return None;
     }
-    let (current, stable, transports) = {
+    let (current, stable, transports, alive) = {
         let replica = state.doc.read().await;
         members::learn_addresses(state, &replica.doc);
         let current = members::members(&replica.doc);
+        // Those that said they are alive within the electorate window.
+        let alive: BTreeSet<String> = members::electorate(
+            &replica.doc,
+            state.clock.now_millis().0,
+            state.electorate_window,
+        )
+        .into_iter()
+        .collect();
         let stable: BTreeSet<String> = current
             .iter()
             .filter(|m| members::tier(&replica.doc, m) == members::Tier::Stable)
@@ -156,11 +164,13 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
             .iter()
             .map(|m| (m.clone(), schemes(&members::addresses(&replica.doc, m))))
             .collect();
-        (current, stable, transports)
+        (current, stable, transports, alive)
     };
     let local_transports: BTreeSet<String> = state.transport.transports().into_iter().collect();
     let k = state.neighbours_k;
-    // Those that don't answer, and those on none of our transports.
+    // Those that don't answer, those on none of our transports, and those that haven't
+    // said they are alive for longer than the electorate window, unless we heard from
+    // them since we started (one that comes back says so to us first: its HELLO).
     let mut unreachable = BTreeSet::new();
     for member in &current {
         if member == state.local_id() {
@@ -168,9 +178,16 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         }
         if !state.transport.can_reach(member) {
             unreachable.insert(member.clone());
-        } else if let Some(peer_state) = state.get_peer_state(member).await
-            && peer_state.lock().await.is_unreachable()
-        {
+            continue;
+        }
+        let (gone, heard) = match state.get_peer_state(member).await {
+            Some(peer_state) => {
+                let peer_state = peer_state.lock().await;
+                (peer_state.is_unreachable(), peer_state.last_seen.is_some())
+            }
+            None => (false, false),
+        };
+        if gone || (!heard && !alive.contains(member)) {
             unreachable.insert(member.clone());
         }
     }