pull missing changes

Nodes push changes to their neighbours but don't relay them directly.
In a partial mesh, nodes can learn about missing changes via keepalive
messages and can pull them from their neighbours.

7 files changed, 162 insertions, 5 deletions

diff --git a/src/main.rs b/src/main.rs
index a688e04..527a730 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -193,6 +193,7 @@ async fn main() -> Result<()> {
         run_id: uuid::Uuid::new_v4().into_bytes(),
         neighbours_k: args.neighbours.max(2),
         neighbours: Default::default(),
+        pulling: Default::default(),
         stale: Default::default(),
         catch_up_source: Default::default(),
     };
diff --git a/src/state.rs b/src/state.rs
index 7574e35..10471df 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -536,6 +536,8 @@ pub(crate) struct AppState {
     /// Target number of neighbours (`--neighbours`); up to this many + 1 members the
     /// swarm is a full mesh.
     pub(crate) neighbours_k: usize,
+    /// The neighbour we are pulling changes from, and since when (`swarm::relay`).
+    pub(crate) pulling: Arc<std::sync::Mutex<Option<(String, Instant)>>>,
     /// Our neighbours, once known (`swarm::topology`).
     pub(crate) neighbours: Arc<std::sync::RwLock<Option<crate::swarm::topology::NeighbourView>>>,
 
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index 0997f7c..5616fe8 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -6,6 +6,7 @@ pub(crate) mod bootstrap;
 pub(crate) mod members;
 pub(crate) mod pair;
 pub(crate) mod peer;
+pub(crate) mod relay;
 pub(crate) mod source;
 pub(crate) mod sync;
 pub(crate) mod topology;
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index a42f5a4..2fd248e 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -133,6 +133,7 @@ async fn note_keepalive(state: &AppState, peer: &str, remote: &transport::Keepal
         peer_state.syncs_with_us = Some(remote.syncs_with_you);
         peer_state.reported_heads = remote.heads.heads().map(<[_]>::len);
         drop(peer_state);
+        crate::swarm::relay::pull_if_behind(state, peer, remote).await;
         if became_ready {
             // We may have left the opening to it while it wasn't ready, and nothing else
             // would make us look again.
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
new file mode 100644
index 0000000..a0fa6e1
--- /dev/null
+++ b/src/swarm/relay.rs
@@ -0,0 +1,126 @@
+//! Passing changes on in a partial mesh (`swarm::topology`).
+//!
+//! A node pushes its own changes to its neighbours, but nothing it receives goes any
+//! further by itself. So after applying changes from one neighbour, a node *announces*
+//! its new heads to its other neighbours with a PING (a keepalive: heads, about 100
+//! bytes, not the changes). A neighbour that sees heads it lacks *pulls*: it enters the
+//! announced heads as the peer's heads in its sync state, Automerge asks for what's
+//! missing (`need`), and the announcer answers with exactly those changes. Each node
+//! receives a change about once; the rest is announcements. Regular keepalives work the
+//! same way, so any gap is noticed within a keepalive interval.
+//!
+//! In a full mesh the writer reaches everyone itself, so nothing is announced.
+
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+
+use tokio::time::Instant;
+use tracing::debug;
+
+use crate::state::{AppState, Standing};
+use crate::swarm::peer::local_keepalive;
+use crate::swarm::sync::SyncEvent;
+use crate::swarm::topology;
+use crate::transport::{self, HeadsSummary, ParsedMessage};
+
+/// A pull that brought nothing for this long no longer blocks pulling from another peer.
+const PULL_TIMEOUT: Duration = Duration::from_secs(15);
+
+/// After changes from `from` were applied and our heads changed: tell our other
+/// neighbours.
+pub(crate) async fn announce(state: &AppState, from: &str) {
+    if !state.initial_sync_done.load(Ordering::Acquire) || topology::is_full_mesh(state) {
+        return;
+    }
+    let neighbours: Vec<String> = state
+        .peer_states
+        .read()
+        .await
+        .keys()
+        .filter(|peer| peer.as_str() != from && topology::is_neighbour(state, peer))
+        .cloned()
+        .collect();
+    for peer in neighbours {
+        let keepalive = local_keepalive(state, &peer).await;
+        if let Err(error) = state
+            .transport
+            .send_message(&peer, &ParsedMessage::Ping(keepalive))
+            .await
+        {
+            debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed");
+        }
+    }
+}
+
+/// A keepalive from `peer`: if it has heads we lack, pull them (one peer at a time).
+pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transport::Keepalive) {
+    if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) {
+        return;
+    }
+    let missing = match &remote.heads {
+        HeadsSummary::Heads(heads) => {
+            // AutoCommit's lookups take &mut self.
+            let mut replica = state.doc.write().await;
+            let missing = heads
+                .iter()
+                .any(|head| replica.doc.get_change_by_hash(head).is_none());
+            if !missing {
+                return;
+            }
+            Some(heads.clone())
+        }
+        // Can't tell which heads are missing; the digest differs from ours, though.
+        HeadsSummary::Digest(_) => None,
+    };
+
+    let now = Instant::now();
+    {
+        let mut pulling = state
+            .pulling
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        let busy = matches!(
+            &*pulling,
+            Some((other, since)) if other != peer && now < *since + PULL_TIMEOUT
+        );
+        if busy {
+            return;
+        }
+        *pulling = Some((peer.to_owned(), now));
+    }
+
+    let Some(peer_state) = state.get_peer_state(peer).await else {
+        return;
+    };
+    {
+        let mut peer_state = peer_state.lock().await;
+        if !matches!(peer_state.standing, Standing::Member { .. }) {
+            return;
+        }
+        match missing {
+            Some(heads) => {
+                // Automerge asks for whatever these heads need that we don't have.
+                peer_state.sync.state.their_heads = Some(heads);
+                peer_state.sync.state.in_flight = false;
+            }
+            // A fresh session's opening carries our heads and bloom filter; the answer
+            // brings what we lack.
+            None => peer_state.reset_sync(),
+        }
+    }
+    debug!(peer = %peer, "Neighbour has changes we lack; pulling");
+    let _ = state
+        .sync_events
+        .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
+}
+
+/// A sync message from `peer` was applied: a pull from it is answered.
+pub(crate) fn pulled(state: &AppState, peer: &str) {
+    let mut pulling = state
+        .pulling
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    if pulling.as_ref().is_some_and(|(p, _)| p == peer) {
+        *pulling = None;
+    }
+}
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index b8d16aa..383cdaf 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -310,8 +310,9 @@ pub(crate) async fn handle_sync(
         let message = envelope.message;
         let peer = peer.to_owned();
 
-        // Some(caught_up) when applied, None when the message belongs to another epoch.
-        tokio::task::spawn_blocking(move || -> Result<Option<bool>> {
+        // Some((caught_up, heads changed)) when applied, None when the message belongs
+        // to another epoch.
+        tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool)>> {
             let mut replica = doc.blocking_write();
 
             // Authoritative check, under the same lock as the apply: the document may
@@ -322,6 +323,7 @@ pub(crate) async fn handle_sync(
 
             let mut peer_state = peer_state_mutex.blocking_lock();
             peer_state.reports_ready = Some(from_ready);
+            let heads_before = replica.doc.get_heads();
 
             replica
                 .doc
@@ -332,8 +334,9 @@ pub(crate) async fn handle_sync(
             peer_state.sync.reply_settled();
 
             resend_if_needed(&peer, &mut peer_state.sync.state);
+            let changed = replica.doc.get_heads() != heads_before;
 
-            Ok(Some(
+            Ok(Some((
                 // Only from a ready peer: a peer still catching up has empty heads, which
                 // every document "contains".
                 check_caught_up
@@ -348,13 +351,14 @@ pub(crate) async fn handle_sync(
                                 .iter()
                                 .all(|h| replica.doc.get_change_by_hash(h).is_some())
                         }),
-            ))
+                changed,
+            )))
         })
         .await
         .context("Sync task panicked")??
     };
 
-    let Some(caught_up) = applied else {
+    let Some((caught_up, changed)) = applied else {
         debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch");
         return Ok(());
     };
@@ -368,6 +372,11 @@ pub(crate) async fn handle_sync(
 
     state.notify_store_changed();
 
+    crate::swarm::relay::pulled(state, peer);
+    if changed {
+        crate::swarm::relay::announce(state, peer).await;
+    }
+
     state
         .sync_events
         .send(SyncEvent::MessageReceived(peer.to_owned()))
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index b5880cf..dddf81b 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -33,6 +33,18 @@ pub(crate) struct NeighbourView {
     k: usize,
     unreachable: BTreeSet<String>,
     neighbours: BTreeSet<String>,
+    /// Every reachable member is a neighbour.
+    full_mesh: bool,
+}
+
+/// Whether every reachable member is our neighbour (also while the view is unknown).
+pub(crate) fn is_full_mesh(state: &AppState) -> bool {
+    state
+        .neighbours
+        .read()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+        .as_ref()
+        .is_none_or(|view| view.full_mesh)
 }
 
 /// Whether we initiate contact with `peer`: HELLOs, keepalives, dormancy checks, sync
@@ -113,10 +125,15 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         removed = change.removed.len(),
         "Neighbours computed from the member list"
     );
+    let reachable_others = current
+        .iter()
+        .filter(|m| m.as_str() != state.local_id() && !unreachable.contains(*m))
+        .count();
     *state
         .neighbours
         .write()
         .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(NeighbourView {
+        full_mesh: computed.len() >= reachable_others,
         members: current,
         k,
         unreachable,