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.

3 files changed, 28 insertions, 42 deletions

diff --git a/src/swarm/pair.rs b/src/swarm/pair.rs
index 7200a91..c1a196b 100644
--- a/src/swarm/pair.rs
+++ b/src/swarm/pair.rs
@@ -1,5 +1,5 @@
-//! Which side of a pair of nodes leads, for duties only one side needs to do (opening
-//! the sync after a reset, sending keepalive PINGs).
+//! Which side of a pair of nodes leads, for duties only one side needs to do (sending
+//! keepalive PINGs).
 //!
 //! Both sides decide on their own, from the two keys alone, and agree. The leader is
 //! chosen per pair by one bit of a hash of both keys, not by comparing the keys: a
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;
 
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 8a85b8c..c918fe8 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -41,6 +41,8 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
         return Ok(());
     };
 
+    let we_ready = state.is_ready().await;
+
     let validated_swarm = {
         let peer_state = peer_state_mutex.lock().await;
 
@@ -80,27 +82,14 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
                 return Ok(());
             }
 
-            let now = Instant::now();
-            if let Some(wait) = peer_state.sync.opening_wait(now) {
-                trace!(
-                    wait_ms = wait.as_millis(),
-                    "Leaving the opening sync message to the peer"
-                );
-                let tx = state.sync_events.clone();
-                let peer = peer.to_owned();
-                tokio::spawn(async move {
-                    tokio::time::sleep(wait).await;
-                    let _ = tx.send(SyncEvent::PeerNeedsSync(peer));
-                });
+            if peer_state
+                .sync
+                .leave_opening(we_ready, peer_state.reports_ready)
+            {
+                // Its own opening (or anything else from it) brings us back here.
+                trace!("Leaving the opening sync message to the peer, which isn't ready");
                 return Ok(());
             }
-            if peer_state.sync.opening_overdue(now) {
-                debug!(
-                    cause = ?peer_state.sync.cause,
-                    likely = peer_state.sync.cause.missing_opening_hint(),
-                    "The peer didn't open the sync within the grace period; opening ourselves"
-                );
-            }
         }
 
         swarm_id.clone()