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.

1 files changed, 126 insertions, 0 deletions

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;
+    }
+}