immediately probe dormant members found in hellos

Instead of waiting for its regular timer, reprobe dormant members found
in hello reply messages on the next worker tick.  With this a partition
gets reconnected quickly as soon as the first pair reconnects.

1 files changed, 37 insertions, 4 deletions

diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs
index 3e8fbf4..a3290fa 100644
--- a/src/swarm/peer/handshake.rs
+++ b/src/swarm/peer/handshake.rs
@@ -260,6 +260,11 @@ pub(super) async fn send_hello_reply(
 /// (it says whom we greet; a listed peer that isn't one would get nothing from us, so
 /// it isn't remembered either). Where the sender reaches them is a hint to the
 /// transport (at most `MAX_LISTED_ADDRESSES` each).
+///
+/// A listed peer we can't reach (dormant, or unanswered HELLOs) is probed at once
+/// rather than when its probe is due (`probe_now`): the sender reaches it, so it's likely
+/// back for us too, as after a partition, whose sides would otherwise find each other
+/// pair by pair, each at its own probe (up to `dormant-probe-max` apart).
 pub(super) async fn discover_peers(
     state: &AppState,
     from_peer: &str,
@@ -272,9 +277,6 @@ pub(super) async fn discover_peers(
         if discovered_peer == from_peer || !state.transport.is_peer_id(discovered_peer) {
             continue;
         }
-        if view_known && !topology::is_neighbour(state, discovered_peer) {
-            continue;
-        }
         if taken == MAX_LISTED_PEERS {
             debug!(
                 from = %from_peer,
@@ -283,9 +285,18 @@ pub(super) async fn discover_peers(
             );
             break;
         }
+        let addresses = &listed.addresses[..listed.addresses.len().min(MAX_LISTED_ADDRESSES)];
+        if probe_now(state, discovered_peer).await {
+            taken += 1;
+            state.transport.learn_addresses(discovered_peer, addresses);
+            debug!(peer = %discovered_peer, by = %from_peer, "A peer we can't reach is reached by another; probing it now");
+            continue;
+        }
+        if view_known && !topology::is_neighbour(state, discovered_peer) {
+            continue;
+        }
         taken += 1;
 
-        let addresses = &listed.addresses[..listed.addresses.len().min(MAX_LISTED_ADDRESSES)];
         state.transport.learn_addresses(discovered_peer, addresses);
         remember_peer(state, discovered_peer).await;
         if let Err(error) = send_hello_once(state, discovered_peer).await {
@@ -298,6 +309,28 @@ pub(super) async fn discover_peers(
     }
 }
 
+/// Brings `peer`'s next probe forward to now, if it is a member we can't reach (dormant,
+/// or unanswered HELLOs) whose probe is pending: the peer worker sends it (its next
+/// tick). Whether it was.
+async fn probe_now(state: &AppState, peer: &str) -> bool {
+    let Some(peer_state) = state.get_peer_state(peer).await else {
+        return false;
+    };
+    let mut peer_state = peer_state.lock().await;
+    if !matches!(peer_state.standing, Standing::Member { .. })
+        || !(peer_state.liveness.is_dormant() || peer_state.is_unreachable())
+    {
+        return false;
+    }
+    match &mut peer_state.hello_state {
+        HelloState::AwaitingReply { next_attempt } => {
+            *next_attempt = Instant::now();
+            true
+        }
+        _ => false,
+    }
+}
+
 /// New neighbours get our HELLO (we may not have known them before: the member list is
 /// how a partial mesh discovers them) and a sync opening. The opening matters for a
 /// neighbour whose handshake is already complete, e.g. one that returned from dormancy