aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm/relay.rs
diff options
context:
space:
mode:
authortv <tv@krebsco.de>2026-11-04 06:51:21 +0100
committertv <tv@krebsco.de>2026-11-04 06:51:21 +0100
commit6430e7540ff5e83e777bf3c80bf609f3f85bcfe8 (patch)
tree6eeed53073d288e029e684a603a6a1656376d72d /src/swarm/relay.rs
parent9bda1116b34c7bc3dcf38c570ce21b21670f16e8 (diff)
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.
Diffstat (limited to 'src/swarm/relay.rs')
-rw-r--r--src/swarm/relay.rs23
1 files changed, 22 insertions, 1 deletions
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 {