1 files changed, 179 insertions, 0 deletions
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
new file mode 100644
index 0000000..703bc3f
--- /dev/null
+++ b/src/swarm/relay.rs
@@ -0,0 +1,179 @@
+//! 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.
+//!
+//! Announcements are coalesced: the first applied change schedules one announcement
+//! round `ANNOUNCE_DELAY` later, and every change applied until then rides along. A
+//! round costs one PING and one PONG per neighbour (plus the pulls), whatever the
+//! write rate; the delay is the price per hop. Without coalescing, every applied sync
+//! message would announce to every neighbour, and announcements would outnumber the
+//! data they announce (about 15 to 1 at 250 nodes).
+//!
+//! In a full mesh the writer reaches everyone itself, so nothing is announced.
+
+use std::collections::HashSet;
+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::send_ping;
+use crate::swarm::sync::SyncEvent;
+use crate::swarm::topology;
+use crate::transport::{self, HeadsSummary};
+
+/// A pull that brought nothing for this long no longer blocks pulling from another peer.
+const PULL_TIMEOUT: Duration = Duration::from_secs(15);
+
+/// How long an announcement waits for further changes to ride along. Added once per
+/// hop to propagation; well below a round trip over Mycelium.
+const ANNOUNCE_DELAY: Duration = Duration::from_millis(200);
+
+/// After changes from `from` were applied and our heads changed: tell our other
+/// neighbours, in the next announcement round.
+pub(crate) async fn announce(state: &AppState, from: &str) {
+    if !state.initial_sync_done.load(Ordering::Acquire) || topology::is_full_mesh(state) {
+        return;
+    }
+    let schedule = {
+        let mut pending = state
+            .announce_pending
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        match pending.as_mut() {
+            Some(senders) => {
+                senders.insert(from.to_owned());
+                false
+            }
+            None => {
+                *pending = Some(HashSet::from([from.to_owned()]));
+                true
+            }
+        }
+    };
+    if !schedule {
+        return;
+    }
+    let state = state.clone();
+    tokio::spawn(async move {
+        tokio::time::sleep(ANNOUNCE_DELAY).await;
+        let senders = state
+            .announce_pending
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .take()
+            .unwrap_or_default();
+        announce_now(&state, &senders).await;
+    });
+}
+
+/// One announcement round. The peer whose changes we announce has them already, so a
+/// round with a single sender skips it; with several, each lacks the others' changes,
+/// so every neighbour is told.
+async fn announce_now(state: &AppState, senders: &HashSet<String>) {
+    let skip = match senders.len() {
+        1 => senders.iter().next().cloned(),
+        _ => None,
+    };
+    let neighbours: Vec<String> = state
+        .peer_states
+        .read()
+        .await
+        .keys()
+        .filter(|peer| {
+            skip.as_deref() != Some(peer.as_str()) && topology::is_neighbour(state, peer)
+        })
+        .cloned()
+        .collect();
+    for peer in neighbours {
+        if let Err(error) = send_ping(state, &peer).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;
+    }
+    // The common case, the same heads, needs only the cached heads (read lock).
+    if HeadsSummary::of(state.doc.read().await.heads().to_vec()) == remote.heads {
+        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;
+    }
+}