refresh peers after stall

3 files changed, 57 insertions, 1 deletions

diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs
index 25ecf89..e4981dd 100644
--- a/src/simtest/timing.rs
+++ b/src/simtest/timing.rs
@@ -146,6 +146,44 @@ async fn stalled_catch_up_source_is_replaced() {
     assert!(sources >= 2, "{sources} catch-up sources");
 }
 
+/// A joining node whose only known peer stalls as its catch-up source asks it again
+/// (a fresh HELLO): the reply lists the peers that joined meanwhile, and one of them
+/// takes over.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn stalled_source_refreshes_the_peer_list() {
+    let sim = Sim::new(7);
+    let a = sim.start("a", &[], Start::Genesis);
+    let a = &a;
+    sim.wait_ready(&[a], SOON).await;
+    write(a, "data", json!("x")).await;
+
+    // c knows only a, which lists nobody: there is nobody else yet.
+    sim.net.set_link(
+        "a",
+        "c",
+        LinkParams {
+            drop: vec![MessageKind::Sync],
+            ..LinkParams::default()
+        },
+    );
+    let c = sim.start("c", &["a"], Start::Join);
+    let c = &c;
+    sim.wait_until("c knows a", SOON, move || async move {
+        c.get_peer_state("a").await.is_some()
+    })
+    .await;
+
+    // b joins meanwhile; c hears of it only by asking a again (after giving it up, or
+    // earlier if the keepalive heal re-handshakes first).
+    let b = sim.start("b", &["a"], Start::Join);
+    let b = &b;
+    sim.wait_ready(&[b], SOON).await;
+
+    sim.wait_ready(&[c], LOSSY).await;
+    assert_eq!(read(c, "data").await, Some(json!("x")));
+    assert!(c.get_peer_state("b").await.is_some());
+}
+
 /// A peer that hears nothing for `dormant_after` is dormant; once messages flow again
 /// it is active, re-handshakes, and syncs what it missed.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index e8248bf..1240966 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -221,6 +221,19 @@ pub(crate) async fn remember_peer(state: &AppState, peer: &str) {
     }
 }
 
+/// Starts a new handshake with `peer` even though one is complete: its reply resets
+/// sync on both sides and brings its current peer list (`swarm::source`, after giving
+/// it up as the catch-up source).
+pub(crate) async fn rehandshake(state: &AppState, peer: &str) {
+    if let Some(peer_state) = state.get_peer_state(peer).await {
+        let mut peer_state = peer_state.lock().await;
+        if peer_state.hello_state == HelloState::Established {
+            peer_state.hello_state = HelloState::NotSent;
+        }
+    }
+    ensure_handshake(state, peer).await;
+}
+
 pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()> {
     let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
         trace!(peer = %peer, "Peer was forgotten; not sending HELLO");
diff --git a/src/swarm/source.rs b/src/swarm/source.rs
index 9eac445..b1701b3 100644
--- a/src/swarm/source.rs
+++ b/src/swarm/source.rs
@@ -23,7 +23,10 @@
 //! `Timing::delivery_limit`, after which Mycelium has delivered or dropped anything it
 //! sent before. It isn't chosen again for another `delivery_limit`, so another peer
 //! takes over rather than the same one stalling again; if it is the only candidate,
-//! catching up waits that long before trying it again. Meanwhile, every minute without a sync message from it is logged as a
+//! catching up waits that long before trying it again. It also gets a fresh HELLO: a
+//! joining node learns of other peers only from HELLO_REPLYs, and the one it got from
+//! this peer may have listed none (a peer lists only its neighbours, which a new member
+//! isn't until its registration has spread). Meanwhile, every minute without a sync message from it is logged as a
 //! warning.
 
 use std::sync::atomic::Ordering;
@@ -203,6 +206,8 @@ pub(crate) fn reply_timed_out(state: &AppState, peer: &str) {
             "Catch-up source doesn't answer our sync messages; catching up from another peer"
         );
         resync_everyone(state);
+        let (state, peer) = (state.clone(), peer.to_owned());
+        tokio::spawn(async move { crate::swarm::peer::rehandshake(&state, &peer).await });
     }
 }