both sides may open sync after reset

Ready nodes will leave the first message to the pear that isn't ready.
Hello and hello reply now carry ready, so readiness is known from the
very first contact, before any sync.  Together with the sync messages
and keepalives, reports_ready is always current.

After epoch switch, pings go to everyone except ready members already
on the new epoch, these learn from the lagging peer's opening.

1 files changed, 18 insertions, 21 deletions

diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index bf552e9..0ef42bf 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -12,7 +12,7 @@ use tracing::{debug, info, instrument, trace, warn};
 
 use crate::document::replica::{Epoch, EpochRelation};
 use crate::document::store::{self, StoreError};
-use crate::state::{AppState, HelloState, Liveness, PeerState, SessionCause, Standing};
+use crate::state::{AppState, HelloState, Liveness, PeerState, Standing};
 use crate::swarm::sync::{SyncEvent, handle_sync};
 use crate::transport::{self, ParsedMessage};
 
@@ -267,6 +267,7 @@ async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result
                 swarm_id,
                 epoch: state.local_epoch(),
                 auth,
+                ready: state.is_ready().await,
             }),
         )
         .await;
@@ -327,15 +328,15 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> {
         .as_ref()
         .map(|key| key.hello_reply_tag(&swarm_id, state.local_id(), peer))
         .unwrap_or_default();
+    let ready = state.is_ready().await;
     let reply = transport::HelloReply {
         swarm_id,
         peers,
         epoch: state.local_epoch(),
         auth,
+        ready,
     };
 
-    let ready = state.is_ready().await;
-
     debug!(to = %peer, ready, "Sending HELLO_REPLY");
 
     state
@@ -445,15 +446,16 @@ pub(crate) async fn run_peer_worker(state: AppState) {
 
 // --- peer membership ---------------------------------------------------------
 
-/// Makes `peer` a member of `swarm_id`, after its valid HELLO (`cause: PeerHello`) or
-/// HELLO_REPLY (`HandshakeComplete`). 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.
+/// 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.
 async fn establish_peer(
     state: &AppState,
     peer: &str,
     swarm_id: String,
-    cause: SessionCause,
+    from_hello: bool,
+    ready: bool,
 ) -> Result<()> {
     let expected_swarm = state.swarm_id().await;
 
@@ -493,10 +495,11 @@ 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.
-    if cause == SessionCause::PeerHello || peer_state.hello_state != HelloState::Established {
+    peer_state.reports_ready = Some(ready);
+    if from_hello || peer_state.hello_state != HelloState::Established {
         peer_state.hello_state = HelloState::Established;
-        peer_state.reset_sync(cause);
-        info!(peer = %peer, ?cause, "Handshake complete; sync state reset");
+        peer_state.reset_sync();
+        info!(peer = %peer, from_hello, "Handshake complete; sync state reset");
     }
 
     info!(peer = %peer, swarm_id = %swarm_id, "Peer membership established");
@@ -529,7 +532,7 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool
             // The peer switched documents; our sync state describes its old one. Count
             // the restart as sync activity, so the keepalive heal doesn't re-handshake
             // (resetting both sides again) while this fresh sync is under way.
-            peer_state.reset_sync(SessionCause::PeerEpochChanged);
+            peer_state.reset_sync();
             peer_state.last_sync_rx = Some(Instant::now());
             // Whether it would sync with us was said about its old epoch.
             peer_state.syncs_with_us = None;
@@ -620,7 +623,7 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool {
                 }
                 peer_state.standing = Standing::Stranger;
                 peer_state.hello_state = HelloState::NotSent;
-                peer_state.reset_sync(SessionCause::KeyRejected);
+                peer_state.reset_sync();
             }
             // The one place an operator can see why a node can't join: rate limited,
             // since anyone can send us handshakes.
@@ -695,7 +698,7 @@ pub(crate) async fn handle_message(
                     bail!("Incompatible swarm at {peer}: ours is {swarm_id}, it claims {theirs}");
                 }
             }
-            establish_peer(state, &peer, swarm_id, SessionCause::PeerHello).await?;
+            establish_peer(state, &peer, swarm_id, true, hello.ready).await?;
 
             if let Err(error) = send_hello_reply(state, &peer).await {
                 // The peer retries its HELLO.
@@ -707,13 +710,7 @@ pub(crate) async fn handle_message(
                 .send(SyncEvent::PeerValidated(peer.clone()));
         }
         transport::ParsedMessage::HelloReply(reply) => {
-            establish_peer(
-                state,
-                &peer,
-                reply.swarm_id.clone(),
-                SessionCause::HandshakeComplete,
-            )
-            .await?;
+            establish_peer(state, &peer, reply.swarm_id.clone(), false, reply.ready).await?;
 
             discover_peers(state, &peer, &reply.peers).await;