1 files changed, 74 insertions, 20 deletions
diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs
index 3e8fbf4..072545b 100644
--- a/src/swarm/peer/handshake.rs
+++ b/src/swarm/peer/handshake.rs
@@ -10,7 +10,6 @@ use tokio::time::Instant;
 use tracing::{debug, info, instrument, trace, warn};
 
 use crate::state::{AppState, HelloState, PeerState, Standing};
-use crate::swarm::sync::SyncEvent;
 use crate::swarm::topology;
 use crate::transport::{self, ParsedMessage};
 
@@ -148,15 +147,11 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
         let mut peer_state = peer_state_mutex.lock().await;
 
         if peer_state.hello_state == HelloState::Sending {
-            let was_unreachable = peer_state.unreachable_since.is_some();
-            peer_state.hello_sent(Instant::now());
-            if !was_unreachable && peer_state.unreachable_since.is_some() {
-                state.topology_changed.notify_one();
-            }
+            topology::change_peer(state, &mut peer_state, |p| p.hello_sent(Instant::now()));
             let wait = peer_state.next_hello_wait();
-            peer_state.hello_state = HelloState::AwaitingReply {
-                next_attempt: Instant::now() + wait,
-            };
+            let next_attempt = Instant::now() + wait;
+            peer_state.hello_state = HelloState::AwaitingReply { next_attempt };
+            state.peers_due.by(next_attempt);
         }
     }
 
@@ -260,6 +255,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 +272,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 +280,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 +304,53 @@ pub(super) async fn discover_peers(
     }
 }
 
+/// Whenever others may reach this node where they couldn't before
+/// (`Transport::on_reachable`: a relay granted it a slot), the members it can't reach
+/// are probed at once (`probe_now`): they may have given up on it meanwhile, waiting
+/// for their next probe, up to `dormant-probe-max` away. Two nodes behind NATs that
+/// restart together each fail to reach the other until it holds a slot; the second to
+/// get one reaches the first this way. Runs for the node's life.
+pub(crate) async fn probe_when_reachable(state: AppState, reachable: Arc<tokio::sync::Notify>) {
+    loop {
+        reachable.notified().await;
+        let mut probed = 0;
+        for (peer, _) in state.peers_snapshot().await {
+            if probe_now(&state, &peer).await {
+                probed += 1;
+            }
+        }
+        if probed > 0 {
+            debug!(
+                probed,
+                "Reachable anew; probing the members we can't reach now"
+            );
+        }
+    }
+}
+
+/// 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 (woken for
+/// it). 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();
+            state.peers_due.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
@@ -314,9 +367,7 @@ pub(crate) async fn apply_neighbour_change(state: &AppState, change: topology::N
         }
         // Skipped there unless the handshake is complete; otherwise its reply restarts
         // sync anyway.
-        let _ = state
-            .sync_events
-            .send(SyncEvent::PeerNeedsSync(peer.clone()));
+        state.request_sync(peer);
     }
     for peer in &change.removed {
         if let Some(peer_state) = state.get_peer_state(peer).await {
@@ -372,11 +423,12 @@ pub(super) async fn establish_peer(
     }
 
     if !matches!(peer_state.standing, Standing::Member { .. }) {
-        peer_state.standing = Standing::Member {
-            swarm_id: swarm_id.clone(),
-        };
         // A member our list may lack yet: a neighbour then (`topology::refresh`).
-        state.topology_changed.notify_one();
+        topology::change_peer(state, &mut peer_state, |p| {
+            p.standing = Standing::Member {
+                swarm_id: swarm_id.clone(),
+            }
+        });
     }
 
     // Stop our retries and sync from a clean slate. A new handshake from the peer always
@@ -386,6 +438,8 @@ pub(super) async fn establish_peer(
     if from_hello || peer_state.hello_state != HelloState::Established {
         peer_state.hello_state = HelloState::Established;
         peer_state.reset_sync();
+        // Keepalives start.
+        state.peers_due.now();
         info!(peer = %peer, from_hello, "Handshake complete; sync state reset");
     }