worker sends off the pass

2 files changed, 49 insertions, 15 deletions

diff --git a/src/state/peer.rs b/src/state/peer.rs
index 857c29d..2f027ba 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -104,6 +104,9 @@ pub(crate) struct PeerState {
     pub(crate) last_seen: Option<Instant>,
     /// Our last keepalive to it (a PING or a STATUS).
     pub(crate) last_keepalive_at: Option<Instant>,
+    /// The peer worker's sends to it (a keepalive, a HELLO again) are on their way, off
+    /// the worker's pass (`swarm::peer::run_peer_worker`): none more until they are.
+    pub(crate) worker_sending: bool,
     /// Its last keepalive to us (a PING or a STATUS): after it, the next is ours.
     pub(crate) keepalive_received_at: Option<Instant>,
     /// The last STATUS between us, either way (`status_refresh`).
@@ -288,6 +291,7 @@ impl PeerState {
             added_at: Instant::now(),
             last_seen: None,
             last_keepalive_at: None,
+            worker_sending: false,
             keepalive_received_at: None,
             last_status_at: None,
             awaiting_answer: None,
@@ -640,7 +644,8 @@ impl PeerState {
     ) -> Option<Instant> {
         let dormancy = (ready && neighbour && !self.liveness.is_dormant())
             .then(|| self.last_seen.unwrap_or(self.added_at) + timing.dormant_after);
-        let keepalive = if neighbour {
+        // Not while the worker's sends to it are on their way: after them (they wake it).
+        let keepalive = if neighbour && !self.worker_sending {
             self.keepalive_due_at(timing)
         } else {
             None
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