recompute neighbours whenever topology changes

6 files changed, 47 insertions, 5 deletions

diff --git a/src/node.rs b/src/node.rs
index b3e7463..82b99fc 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -103,6 +103,7 @@ impl Node {
             files: config.files,
             config: Arc::new(config.config_report),
             neighbours: Default::default(),
+            topology_changed: Default::default(),
             pulling: Default::default(),
             announce_pending: Default::default(),
             rtt: Default::default(),
@@ -155,6 +156,9 @@ impl Node {
         tokio::spawn(remake::run_worker(state.clone()));
         trace!("Remake worker task spawned");
 
+        tokio::spawn(swarm::topology::run_worker(state.clone()));
+        trace!("Topology worker task spawned");
+
         state
     }
 }
diff --git a/src/state.rs b/src/state.rs
index af01cf0..ee7fa7a 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -632,6 +632,9 @@ pub(crate) struct AppState {
     pub(crate) announce_pending: Arc<std::sync::Mutex<Option<std::collections::HashSet<String>>>>,
     /// Our neighbours, once known (`swarm::topology`).
     pub(crate) neighbours: Arc<std::sync::RwLock<Option<crate::swarm::topology::NeighbourView>>>,
+    /// Something the neighbours are computed from may have changed: the member list or a
+    /// peer's reachability (`swarm::topology::run_worker`).
+    pub(crate) topology_changed: Arc<tokio::sync::Notify>,
 
     /// While rejoining a newer epoch, the previous document, served read-only (and
     /// marked stale) instead of answering 503.
diff --git a/src/swarm/members.rs b/src/swarm/members.rs
index 221437e..895e0cc 100644
--- a/src/swarm/members.rs
+++ b/src/swarm/members.rs
@@ -25,7 +25,7 @@ use crate::state::{AppState, unix_millis};
 
 const META_ROOT: &str = "_continuum";
 
-fn members_dir() -> Vec<String> {
+pub(crate) fn members_dir() -> Vec<String> {
     vec![META_ROOT.to_owned(), "members".to_owned()]
 }
 
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index 7385ae2..e8248bf 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -299,7 +299,11 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
         let mut peer_state = peer_state_mutex.lock().await;
 
         if peer_state.hello_state == HelloState::Sending {
+            let was_unreachable = peer_state.unreachable_since.is_some();
             peer_state.hello_sent(Instant::now());
+            if !was_unreachable && peer_state.unreachable_since.is_some() {
+                state.topology_changed.notify_one();
+            }
             let wait = match peer_state.liveness {
                 Liveness::Dormant { .. } => state.timing.dormant_probe,
                 Liveness::Active => peer_state.hello_backoff.next(),
@@ -413,7 +417,7 @@ async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[S
 /// neighbours get no more HELLOs, keepalives or sync openings from us; their session is
 /// reset, so we don't keep pushing changes to them either. They are still answered if
 /// they contact us.
-async fn apply_neighbour_change(state: &AppState, change: topology::NeighbourChange) {
+pub(crate) async fn apply_neighbour_change(state: &AppState, change: topology::NeighbourChange) {
     for peer in &change.added {
         remember_peer(state, peer).await;
         if let Err(error) = send_hello_once(state, peer).await {
@@ -492,6 +496,7 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             }
 
             if became_dormant {
+                state.topology_changed.notify_one();
                 warn!(
                     peer = %peer,
                     silent_secs = state.timing.dormant_after.as_secs(),
@@ -529,9 +534,8 @@ pub(crate) async fn run_peer_worker(state: AppState) {
                     .unwrap_or_else(|poisoned| poisoned.into_inner())
                     .retain(|peer| known.contains_key(peer));
             }
-            if let Some(change) = topology::refresh(&state).await {
-                apply_neighbour_change(&state, change).await;
-            }
+            // Fallback for what doesn't notify (`topology::run_worker`).
+            state.topology_changed.notify_one();
             last_forget_check = Some(Instant::now());
         }
     }
@@ -802,6 +806,10 @@ pub(crate) async fn handle_message(
         let peer_state_mutex = state.get_or_create_peer_state(&peer).await;
         let mut peer_state = peer_state_mutex.lock().await;
         let previously_seen = peer_state.last_seen;
+        if peer_state.unreachable_since.is_some() {
+            // Reachable again (`mark_seen` clears it).
+            state.topology_changed.notify_one();
+        }
         (peer_state.mark_seen(Instant::now()), previously_seen)
     };
     crate::swarm::source::heard_from(state, &peer);
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 93af6f7..d4503bb 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -608,6 +608,13 @@ pub(crate) async fn run_change_publisher(state: AppState) {
             warn!(path = ?path, "Concurrent writes produced a conflict");
         }
 
+        if paths
+            .iter()
+            .any(|path| crate::swarm::topology::affects_neighbours(path))
+        {
+            state.topology_changed.notify_one();
+        }
+
         // Fails only when nobody is subscribed.
         let _ = state.change_events.send(Arc::new(ChangeSet { paths }));
     }
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index 57856b6..b40b120 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -107,6 +107,26 @@ pub(crate) struct NeighbourChange {
     pub(crate) removed: Vec<String>,
 }
 
+/// Recomputes the neighbours whenever something they depend on may have changed
+/// (`AppState::topology_changed`): the member list (`swarm::sync::run_change_publisher`),
+/// a peer becoming unreachable or reachable again (`swarm::peer`), and every 5 s as a
+/// fallback, which also covers unreachability expiring after `dormant_probe`. A burst of
+/// notifications coalesces into one recomputation.
+pub(crate) async fn run_worker(state: AppState) {
+    loop {
+        state.topology_changed.notified().await;
+        if let Some(change) = refresh(&state).await {
+            crate::swarm::peer::apply_neighbour_change(&state, change).await;
+        }
+    }
+}
+
+/// Whether a change at `path` may change the neighbours: one in the member list, or the
+/// root (the whole document replaced, as after an epoch switch).
+pub(crate) fn affects_neighbours(path: &[String]) -> bool {
+    path.is_empty() || path.starts_with(&members::members_dir())
+}
+
 /// Recomputes the neighbours once ready, if the member list, the reachable members or
 /// `k` changed.
 pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {