1 files changed, 43 insertions, 14 deletions

diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index b6f55cd..9ddf177 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -93,14 +93,15 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             // Only neighbours are kept alive and judged; others we only answer.
             let neighbour = topology::is_neighbour(&state, peer);
 
-            let (became_dormant, keepalive, unanswered, ask_again) = {
+            let (became_dormant, keepalive, unanswered, ask_again, hello_due, sending) = {
                 let mut peer_state = peer_state_mutex.lock().await;
+                let sending = peer_state.worker_sending;
                 let became_dormant = ready
                     && neighbour
                     && topology::change_peer(&state, &mut peer_state, |p| {
                         p.check_dormancy(&state.timing, now)
                     });
-                let keepalive = if neighbour {
+                let keepalive = if neighbour && !sending {
                     peer_state.take_keepalive_due(&state.timing, now)
                 } else {
                     None
@@ -112,7 +113,18 @@ pub(crate) async fn run_peer_worker(state: AppState) {
                 let ask_again = peer_state.take_ask_again_due(now);
                 reach_changed |= peer_state.answer_overdue(now)
                     && crate::swarm::reach::reached_stops_answering(&state, peer);
-                (became_dormant, keepalive, unanswered, ask_again)
+                let hello_due = matches!(
+                    peer_state.hello_state,
+                    crate::state::HelloState::AwaitingReply { next_attempt } if now >= next_attempt
+                );
+                (
+                    became_dormant,
+                    keepalive,
+                    unanswered,
+                    ask_again,
+                    hello_due,
+                    sending,
+                )
             };
 
             // Unanswered: the path in use may drop what it takes. Its others first, and
@@ -146,17 +158,34 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             };
             // Nothing answered us for a while: once more before it counts as not reached.
             let keepalive = keepalive.or(ask_again.then_some(Keepalive::Ping));
-            let sent = match keepalive {
-                Some(Keepalive::Ping) => send_ping(&state, peer).await,
-                Some(Keepalive::Status) => send_status(&state, peer).await,
-                None => Ok(()),
-            };
-            if let Err(error) = sent {
-                debug!(peer = %peer, kind = ?keepalive, error = %format!("{error:#}"), "Keepalive failed");
-            }
-
-            if let Err(error) = retry_hello_if_due(&state, peer).await {
-                warn!(peer = %peer, error = %format!("{error:#}"), "HELLO retry failed");
+            // Off the pass: a send may take a dial (seconds on libp2p, a minute and more
+            // on Tor, to a peer that is gone), and every other peer's keepalives would
+            // wait for it. One at a time per peer, so a dial that hangs doesn't pile up.
+            if !sending && (keepalive.is_some() || hello_due) {
+                peer_state_mutex.lock().await.worker_sending = true;
+                let (state, peer, peer_state_mutex) = (
+                    state.clone(),
+                    peer.clone(),
+                    std::sync::Arc::clone(peer_state_mutex),
+                );
+                tokio::spawn(async move {
+                    // The HELLO first: it is marked as being sent at once, so it isn't
+                    // due again meanwhile.
+                    if let Err(error) = retry_hello_if_due(&state, &peer).await {
+                        warn!(peer = %peer, error = %format!("{error:#}"), "HELLO retry failed");
+                    }
+                    let sent = match keepalive {
+                        Some(Keepalive::Ping) => send_ping(&state, &peer).await,
+                        Some(Keepalive::Status) => send_status(&state, &peer).await,
+                        None => Ok(()),
+                    };
+                    if let Err(error) = sent {
+                        debug!(peer = %peer, kind = ?keepalive, error = %format!("{error:#}"), "Keepalive failed");
+                    }
+                    peer_state_mutex.lock().await.worker_sending = false;
+                    // What fell due meanwhile.
+                    state.peers_due.now();
+                });
             }
 
             let due = peer_state_mutex