sync with neighbours

1 files changed, 38 insertions, 3 deletions

diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index 3db4710..7ab704f 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -15,6 +15,7 @@ use crate::document::replica::{Epoch, EpochRelation};
 use crate::document::store::{self, StoreError};
 use crate::state::{AppState, HelloState, Liveness, PeerState, Standing};
 use crate::swarm::sync::{SyncEvent, handle_sync};
+use crate::swarm::topology;
 use crate::transport::{self, ParsedMessage};
 
 const PEER_TICK: Duration = Duration::from_millis(500);
@@ -150,7 +151,8 @@ async fn heal_if_diverged(
     remote: &transport::Keepalive,
 ) {
     // Heads of documents from different epochs are never equal and never converge.
-    if local.epoch != remote.epoch {
+    // And a peer that isn't our neighbour isn't ours to keep in step.
+    if local.epoch != remote.epoch || !topology::is_neighbour(state, peer) {
         return;
     }
 
@@ -320,6 +322,7 @@ async fn send_hello_reply(state: &AppState, peer: &str, run_id: transport::RunId
         let peer_state = peer_state_mutex.lock().await;
         if peer_state.liveness == Liveness::Active
             && peer_state.hello_state == HelloState::Established
+            && topology::is_neighbour(state, &known)
         {
             peers.push(known);
         }
@@ -361,6 +364,9 @@ async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[S
 
         remember_peer(state, discovered_peer).await;
 
+        if !topology::is_neighbour(state, discovered_peer) {
+            continue;
+        }
         if let Err(error) = send_hello_once(state, discovered_peer).await {
             warn!(
                 peer = %discovered_peer,
@@ -371,6 +377,26 @@ async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[S
     }
 }
 
+/// New neighbours get our HELLO (we may not have known them before: the member list is
+/// how a partial mesh discovers them). Former 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) {
+    for peer in &change.added {
+        remember_peer(state, peer).await;
+        if let Err(error) = send_hello_once(state, peer).await {
+            warn!(peer = %peer, error = %format!("{error:#}"), "HELLO to new neighbour failed; will retry");
+        }
+    }
+    for peer in &change.removed {
+        if let Some(peer_state) = state.get_peer_state(peer).await {
+            let mut peer_state = peer_state.lock().await;
+            peer_state.stop_hello_retries();
+            peer_state.reset_sync();
+        }
+    }
+}
+
 // --- peer worker -------------------------------------------------------------
 
 pub(crate) async fn run_peer_worker(state: AppState) {
@@ -397,11 +423,17 @@ pub(crate) async fn run_peer_worker(state: AppState) {
 
         for (peer, peer_state_mutex) in &peers {
             let now = Instant::now();
+            // Only neighbours are kept alive and judged; others we only answer.
+            let neighbour = topology::is_neighbour(&state, peer);
 
             let (became_dormant, ping_due, unanswered) = {
                 let mut peer_state = peer_state_mutex.lock().await;
-                let became_dormant = ready && peer_state.check_dormancy(&state.timing, now);
-                let ping_due = peer_state.take_keepalive_due(&state.timing, now);
+                let became_dormant =
+                    ready && neighbour && peer_state.check_dormancy(&state.timing, now);
+                let ping_due = neighbour && peer_state.take_keepalive_due(&state.timing, now);
+                if !neighbour {
+                    peer_state.stop_hello_retries();
+                }
                 let unanswered = peer_state.take_overdue_sync_reply(now);
                 (became_dormant, ping_due, unanswered)
             };
@@ -443,6 +475,9 @@ pub(crate) async fn run_peer_worker(state: AppState) {
         if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) {
             apply_forget_records(&state).await;
             crate::swarm::members::ensure_registered(&state).await;
+            if let Some(change) = topology::refresh(&state).await {
+                apply_neighbour_change(&state, change).await;
+            }
             last_forget_check = Some(Instant::now());
         }
     }