add pull queue

3 files changed, 84 insertions, 34 deletions

diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index 4ce326e..a593ac9 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -66,7 +66,7 @@ pub(super) async fn note_keepalive(state: &AppState, peer: &str, remote: &transp
         {
             let (state, peer, remote) = (state.clone(), peer.to_owned(), remote.clone());
             tokio::spawn(async move {
-                crate::swarm::relay::pull_if_behind(&state, &peer, &remote).await
+                crate::swarm::relay::pull_if_behind(&state, &peer, &remote).await;
             });
         }
         if became_ready {
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 34451c9..7c56c90 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -18,7 +18,7 @@
 //!
 //! In a full mesh the writer reaches everyone itself, so nothing is announced.
 
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::sync::atomic::Ordering;
 use std::time::Duration;
 
@@ -42,7 +42,11 @@ 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) {
+    // While the view is unknown everyone counts as a neighbour, but that isn't a full
+    // mesh: the others may not hear from the writer.
+    if !state.initial_sync_done.load(Ordering::Acquire)
+        || (topology::view_known(state) && topology::is_full_mesh(state))
+    {
         return;
     }
     let schedule = {
@@ -98,15 +102,31 @@ async fn announce_now(state: &AppState, senders: &HashSet<String>) {
     }
 }
 
-/// 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) {
+/// Pulls in progress: one neighbour at a time, so each change arrives about once. The
+/// neighbours that announced heads we lack meanwhile wait their turn, with what they
+/// announced, instead of until their next keepalive.
+#[derive(Default)]
+pub(crate) struct Pulling {
+    /// The neighbour we are pulling from, and since when.
+    current: Option<(String, Instant)>,
+    /// The latest keepalive of each neighbour waiting to be pulled from.
+    waiting: HashMap<String, transport::Keepalive>,
+}
+
+/// A keepalive from `peer`: if it has heads we lack, pull them (one peer at a time; a
+/// busy slot queues the peer). Returns whether a pull started.
+pub(crate) async fn pull_if_behind(
+    state: &AppState,
+    peer: &str,
+    remote: &transport::Keepalive,
+) -> bool {
     if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) {
-        return;
+        return false;
     }
     // The common case, the same heads, needs only the published heads.
     let ours = HeadsSummary::of(state.stamp.borrow().heads.to_vec());
     if ours == remote.heads {
-        return;
+        return false;
     }
     let missing = match &remote.heads {
         HeadsSummary::Heads(heads) => {
@@ -116,7 +136,7 @@ pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transp
                 .iter()
                 .any(|head| replica.doc.get_change_by_hash(head).is_none());
             if !missing {
-                return;
+                return false;
             }
             Some(heads.clone())
         }
@@ -124,47 +144,75 @@ pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transp
         HeadsSummary::Digest(_) => None,
     };
 
+    let Some(peer_state) = state.get_peer_state(peer).await else {
+        return false;
+    };
+    if !matches!(peer_state.lock().await.standing, Standing::Member { .. }) {
+        return false;
+    }
+
     let now = Instant::now();
     {
         let mut pulling = state.pulling.lock_or_recover();
         let busy = matches!(
-            &*pulling,
+            &pulling.current,
             Some((other, since)) if other != peer && now < *since + PULL_TIMEOUT
         );
         if busy {
-            return;
+            pulling.waiting.insert(peer.to_owned(), remote.clone());
+            return false;
         }
-        *pulling = Some((peer.to_owned(), now));
+        pulling.current = Some((peer.to_owned(), now));
+        pulling.waiting.remove(peer);
     }
 
-    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.pull_from(heads);
-            }
-            // A fresh session's opening carries our heads and bloom filter; the answer
-            // brings what we lack.
-            None => peer_state.reset_sync(),
-        }
+    match missing {
+        // Automerge asks for whatever these heads need that we don't have.
+        Some(heads) => peer_state.lock().await.sync.pull_from(heads),
+        // A fresh session's opening carries our heads and bloom filter; the answer
+        // brings what we lack.
+        None => peer_state.lock().await.reset_sync(),
     }
     debug!(peer = %peer, "Neighbour has changes we lack; pulling");
     let _ = state
         .sync_events
         .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
+    true
 }
 
-/// A sync message from `peer` was applied: a pull from it is answered.
+/// A sync message from `peer` was applied: a pull from it is answered, and the next
+/// waiting neighbour's turn comes.
 pub(crate) fn pulled(state: &AppState, peer: &str) {
-    let mut pulling = state.pulling.lock_or_recover();
-    if pulling.as_ref().is_some_and(|(p, _)| p == peer) {
-        *pulling = None;
+    {
+        let mut pulling = state.pulling.lock_or_recover();
+        if !pulling.current.as_ref().is_some_and(|(p, _)| p == peer) {
+            return;
+        }
+        pulling.current = None;
+        if pulling.waiting.is_empty() {
+            return;
+        }
+    }
+    let state = state.clone();
+    tokio::spawn(async move { pull_next(&state).await });
+}
+
+/// Pulls from the first waiting neighbour that still has heads we lack.
+async fn pull_next(state: &AppState) {
+    loop {
+        let (peer, remote) = {
+            let mut pulling = state.pulling.lock_or_recover();
+            if pulling.current.is_some() {
+                return;
+            }
+            let Some(peer) = pulling.waiting.keys().next().cloned() else {
+                return;
+            };
+            let remote = pulling.waiting.remove(&peer).expect("just listed");
+            (peer, remote)
+        };
+        if pull_if_behind(state, &peer, &remote).await {
+            return;
+        }
     }
 }
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 868fead..0fbff05 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -400,7 +400,8 @@ pub(crate) async fn handle_sync(
 
     source::reported(state, peer, from_ready, None);
 
-    if caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel) {
+    let just_caught_up = caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel);
+    if just_caught_up {
         info!(peer = %peer, "Initial sync complete");
         state
             .stats
@@ -411,7 +412,8 @@ pub(crate) async fn handle_sync(
     state.notify_store_changed();
 
     crate::swarm::relay::pulled(state, peer);
-    if changed {
+    // Caught up: our other neighbours may lack what the source had.
+    if changed || just_caught_up {
         crate::swarm::relay::announce(state, peer).await;
     }