unanswered pull frees the slot

When the answer to a pull is overdue (the sync reply timeout: 1 s in the
simtests, 10 s in production), the pull gives up the slot, and the next
waiting neighbour is pulled at once.

3 files changed, 27 insertions, 4 deletions

diff --git a/docs/peers.md b/docs/peers.md
index a521db3..be04b16 100644
--- a/docs/peers.md
+++ b/docs/peers.md
@@ -289,9 +289,10 @@ own announcement or STATUS brought them, are this node's: announcements wait 200
 the other skips it), and a neighbour that lacks those heads pulls them (it sets them as the peer's heads in its sync state, so Automerge asks
 for what's missing), if it still lacks them after a round trip to the announcer (at
 most 1 s): while writes go on, the newest are usually on their way to it along a
-writer's tree, and would otherwise come twice. It pulls from one neighbour at a time (15 s timeout); neighbours
-that announce heads it lacks meanwhile wait their turn and are pulled from as soon as
-the current pull is answered, if they still have something it lacks. The regular
+writer's tree, and would otherwise come twice. It pulls from one neighbour at a time
+(until the answer is overdue, by the sync reply timeout, or 15 s); neighbours that
+announce heads it lacks meanwhile wait their turn and are pulled from as soon as the
+current pull is answered or overdue, if they still have something it lacks. The regular
 STATUS (every third keepalive) triggers the same pull, so gaps close within a few
 keepalive intervals. Changes that pass on along a writer's tree (below) go at once,
 without announcement or pull; the announcements catch what the trees miss. A node
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index 67213e7..399741c 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -114,6 +114,7 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             if let Some(changes) = unanswered {
                 debug!(peer = %peer, changes, "No answer to our sync message; sending again");
                 crate::swarm::source::reply_timed_out(&state, peer);
+                crate::swarm::relay::pull_unanswered(&state, peer);
                 state.request_sync(peer);
             }
 
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 95abab8..f141dbb 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -50,7 +50,8 @@ use crate::swarm::peer::announce_heads;
 use crate::swarm::topology;
 use crate::transport::{self, HeadsSummary};
 
-/// A pull that brought nothing for this long no longer blocks pulling from another peer.
+/// A pull that brought nothing for this long no longer blocks pulling from another peer
+/// (sooner once its answer is overdue: `pull_unanswered`).
 const PULL_TIMEOUT: Duration = Duration::from_secs(15);
 
 /// The longest a STATUS's heads we lack wait before we pull them (`pull_after_grace`: a
@@ -336,6 +337,26 @@ pub(crate) fn pulled(state: &AppState, peer: &str) {
     tokio::spawn(async move { pull_next(&state).await });
 }
 
+/// Our sync message to `peer` went unanswered for the reply timeout: a pull from it no
+/// longer blocks the others, and the next waiting neighbour's turn comes. Otherwise
+/// they would wait for `PULL_TIMEOUT`, and then for their next keepalive, as nothing
+/// else brings them back.
+pub(crate) fn pull_unanswered(state: &AppState, peer: &str) {
+    {
+        let mut pulling = state.pulling.lock_or_recover();
+        if pulling.current.as_ref().is_none_or(|(p, _)| p != peer) {
+            return;
+        }
+        debug!(peer = %peer, "Pull unanswered; pulling from the next neighbour");
+        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 {