From 6430e7540ff5e83e777bf3c80bf609f3f85bcfe8 Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 4 Nov 2026 06:51:21 +0100 Subject: 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. --- src/swarm/peer/worker.rs | 1 + src/swarm/relay.rs | 23 ++++++++++++++++++++++- 2 files changed, 23 insertions(+), 1 deletion(-) (limited to 'src') 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 { -- cgit v1.3.1