number handshakes

This prevents late hellos from resetting the session.

8 files changed, 96 insertions, 7 deletions

diff --git a/src/node.rs b/src/node.rs
index 5005d30..9200d16 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -97,6 +97,7 @@ impl Node {
             remake: Arc::new(remake::RemakeState::new(config.clock.now_millis())),
             swarm_key: config.swarm_key,
             run_id: uuid::Uuid::new_v4().into_bytes(),
+            handshakes: Default::default(),
             neighbours_k: config.neighbours_k,
             alias: config.alias,
             electorate_window: config.electorate_window,
diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs
index e4981dd..4247090 100644
--- a/src/simtest/timing.rs
+++ b/src/simtest/timing.rs
@@ -239,6 +239,53 @@ async fn is_dormant(node: &AppState, other: &str) -> bool {
         == Some(true)
 }
 
+/// R27: a resent HELLO (its reply was lost) or a late copy of one doesn't reset the
+/// session it opened; only a new handshake does.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn resent_hello_keeps_the_session() {
+    let sim = Sim::new(8);
+    let a = sim.start("a", &[], Start::Genesis);
+    let b = sim.start("b", &["a"], Start::Join);
+    let (a, b) = (&a, &b);
+    sim.wait_ready(&[a, b], SOON).await;
+    write(a, "data", json!(1)).await;
+    sim.wait_converged(&[a, b], SOON).await;
+    let opened = move || async move { peer(a, "b", |p| p.sync.has_opened()).await == Some(true) };
+    sim.wait_until("a has heard b's heads", SOON, opened).await;
+
+    let (run, handshake) = peer(a, "b", |p| p.last_hello).await.unwrap().unwrap();
+    assert_eq!(run, b.run_id);
+    let (epoch, epoch_seq) = b.epoch_stamp().await;
+    let swarm_id = b.swarm_id().await;
+    let hello = |handshake| {
+        ParsedMessage::Hello(HelloRequest {
+            swarm_id: swarm_id.clone(),
+            epoch,
+            epoch_seq,
+            auth: Vec::new(),
+            ready: true,
+            run_id: run,
+            handshake,
+        })
+    };
+    for repeated in [handshake, handshake.saturating_sub(1)] {
+        sim.net.inject("b", "a", &hello(repeated)).await.unwrap();
+        assert!(opened().await, "HELLO {repeated} reset the session");
+        assert_eq!(
+            peer(a, "b", |p| p.last_hello).await.unwrap(),
+            Some((run, handshake))
+        );
+    }
+    sim.net
+        .inject("b", "a", &hello(handshake + 1))
+        .await
+        .unwrap();
+    assert_eq!(
+        peer(a, "b", |p| p.last_hello).await.unwrap(),
+        Some((run, handshake + 1))
+    );
+}
+
 /// R17: a message sent before the peer's last epoch change and delivered late doesn't
 /// flip the epoch we recorded for it back; a new run of the peer starts its count over.
 /// The peer is a "ghost" whose messages are injected, so their order is the test's.
@@ -261,6 +308,7 @@ async fn late_epoch_does_not_flip_a_peers_epoch() {
             auth: Vec::new(),
             ready: true,
             run_id: run,
+            handshake: 1,
         })
     };
     let ping = |epoch: Epoch, epoch_seq: u32| {
diff --git a/src/state/mod.rs b/src/state/mod.rs
index 438ad52..a51339c 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -85,6 +85,8 @@ pub(crate) struct AppState {
 
     /// Random per run of this node; our HELLOs carry it and replies must echo it.
     pub(crate) run_id: transport::RunId,
+    /// Handshakes this node started in this run, with any peer (`HelloRequest::handshake`).
+    pub(crate) handshakes: Arc<std::sync::atomic::AtomicU32>,
 
     /// Target number of neighbours (`--neighbours`); up to this many + 1 members the
     /// swarm is a full mesh.
diff --git a/src/state/peer.rs b/src/state/peer.rs
index e96cd9c..2ac6b1a 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -102,6 +102,11 @@ pub(crate) struct PeerState {
     pub(crate) epoch_seq: Option<u32>,
     /// The peer's run, from its HELLO or HELLO_REPLY: a new one starts `epoch_seq` over.
     pub(crate) run_id: Option<crate::transport::RunId>,
+    /// The newest handshake the peer started with us: its run and number
+    /// (`HelloRequest::handshake`). A HELLO repeating it is a retry or a late copy.
+    pub(crate) last_hello: Option<(crate::transport::RunId, u32)>,
+    /// The number of our current handshake with the peer (`AppState::handshakes`).
+    pub(crate) handshake: u32,
 
     /// Whether the peer was ready according to its most recent sync message or keepalive.
     pub(crate) reports_ready: Option<bool>,
@@ -169,6 +174,8 @@ impl PeerState {
             epoch: None,
             epoch_seq: None,
             run_id: None,
+            last_hello: None,
+            handshake: 0,
             reports_ready: None,
             syncs_with_us: None,
             reported_heads: None,
diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs
index 038dac6..f89e371 100644
--- a/src/swarm/peer/handshake.rs
+++ b/src/swarm/peer/handshake.rs
@@ -1,6 +1,7 @@
 //! Discovering peers, the handshake (HELLO / HELLO_REPLY), and membership.
 
 use std::sync::Arc;
+use std::sync::atomic::Ordering;
 use std::time::Duration;
 
 use anyhow::{Context, Result, anyhow, bail};
@@ -67,6 +68,9 @@ pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()>
         let mut peer_state = peer_state_mutex.lock().await;
         if peer_state.hello_state == HelloState::NotSent {
             // A new handshake: retries start over.
+            // Numbered across all peers, so a peer we forgot and meet again still
+            // sees a higher number than any it had from this run.
+            peer_state.handshake = state.handshakes.fetch_add(1, Ordering::Relaxed) + 1;
             peer_state.hello_state = HelloState::Sending;
             peer_state.hello_backoff.reset();
             peer_state.hello_attempts = 0;
@@ -115,6 +119,11 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
         .map(|key| key.hello_tag(swarm_id.as_deref().unwrap_or(""), state.local_id(), peer))
         .unwrap_or_default();
     let (epoch, epoch_seq) = state.epoch_stamp().await;
+    // Retries repeat the handshake's number.
+    let handshake = match state.get_peer_state(peer).await {
+        Some(peer_state) => peer_state.lock().await.handshake,
+        None => 0,
+    };
     state.lock_rtt().hello_sent(peer, Instant::now());
     let result = state
         .transport
@@ -127,6 +136,7 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
                 auth,
                 ready: state.is_ready().await,
                 run_id: state.run_id,
+                handshake,
             }),
         )
         .await;
@@ -293,10 +303,11 @@ pub(crate) async fn apply_neighbour_change(state: &AppState, change: topology::N
 
 // --- peer membership ---------------------------------------------------------
 
-/// Makes `peer` a member of `swarm_id`, after its valid HELLO (`from_hello`) or
-/// HELLO_REPLY. Either completes the handshake: one round trip proves both sides to
-/// each other (the HELLO the sender's key, the reply the swarm ID and the responder's
-/// key), so neither side needs a HELLO of its own. `ready` is what the message said.
+/// Makes `peer` a member of `swarm_id`, after its valid HELLO or HELLO_REPLY. Either
+/// completes the handshake: one round trip proves both sides to each other (the HELLO
+/// the sender's key, the reply the swarm ID and the responder's key), so neither side
+/// needs a HELLO of its own. `from_hello`: a HELLO that starts a new handshake (a resent
+/// or late one doesn't). `ready` is what the message said.
 pub(super) async fn establish_peer(
     state: &AppState,
     peer: &str,
@@ -340,8 +351,9 @@ pub(super) async fn establish_peer(
         };
     }
 
-    // Stop our retries and sync from a clean slate. A HELLO always resets: the peer
-    // starts over with us. Duplicate replies (caused by our own retries) don't.
+    // Stop our retries and sync from a clean slate. A new handshake from the peer always
+    // resets: it starts over with us. Its resent HELLOs and duplicate replies (caused by
+    // our own retries) don't.
     peer_state.reports_ready = Some(ready);
     if from_hello || peer_state.hello_state != HelloState::Established {
         peer_state.hello_state = HelloState::Established;
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 040cafb..684a358 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -84,7 +84,20 @@ pub(crate) async fn handle_message(
                     bail!("Incompatible swarm at {peer}: ours is {swarm_id}, it claims {theirs}");
                 }
             }
-            establish_peer(state, &peer, swarm_id, true, hello.ready).await?;
+            // A resent HELLO (our reply was lost) or a late copy of one repeats a
+            // handshake we answered already: it must not reset the session again.
+            let new_handshake = {
+                let peer_state = state.get_or_create_peer_state(&peer).await;
+                let mut peer_state = peer_state.lock().await;
+                let new = peer_state.last_hello.is_none_or(|(run, handshake)| {
+                    run != hello.run_id || hello.handshake > handshake
+                });
+                if new {
+                    peer_state.last_hello = Some((hello.run_id, hello.handshake));
+                }
+                new
+            };
+            establish_peer(state, &peer, swarm_id, new_handshake, hello.ready).await?;
 
             let away = previously_seen.map(|seen| seen.elapsed());
             if let Err(error) = send_hello_reply(state, &peer, hello.run_id, away).await {
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index b2ce63f..614ac48 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -256,6 +256,10 @@ pub struct HelloRequest {
     /// Random per run of the sender; the reply echoes it.
     #[serde(with = "crate::transport::wire::codec::run_id")]
     pub run_id: RunId,
+    /// The handshake's number: the sender numbers the handshakes it starts in a run,
+    /// with all peers, from 1. A retry carries the same number, so the recipient resets
+    /// its sync session only for a new handshake, not for every resend or late copy.
+    pub handshake: u32,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 8a8cbb8..e055393 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -271,6 +271,7 @@ mod tests {
             auth: vec![1, 2, 3],
             ready: true,
             run_id: [4; 16],
+            handshake: 7,
         });
         let ParsedMessage::Hello(decoded) = decode(&encode(&hello).unwrap()).unwrap() else {
             panic!("not a HELLO");
@@ -279,6 +280,7 @@ mod tests {
         assert_eq!((decoded.epoch, decoded.epoch_seq), (epoch, 1));
         assert_eq!(decoded.auth, vec![1, 2, 3]);
         assert_eq!(decoded.run_id, [4; 16]);
+        assert_eq!(decoded.handshake, 7);
 
         let reply = ParsedMessage::HelloReply(HelloReply {
             swarm_id: "swarm".into(),