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.

6 files changed, 64 insertions, 163 deletions

diff --git a/src/state.rs b/src/state.rs
index 64fa8ea..3abfd85 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -89,8 +89,8 @@ pub(crate) struct PendingOutbound {
 
 #[derive(Clone)]
 pub(crate) struct PeerState {
-    /// Whether we lead the pair (`swarm::pair`): we open sync after a reset and send the
-    /// keepalive PINGs. Fixed for the pair, so computed once.
+    /// Whether we lead the pair (`swarm::pair`): we send the keepalive PINGs. Fixed for
+    /// the pair, so computed once.
     pub(crate) leads: bool,
 
     pub(crate) standing: Standing,
@@ -138,79 +138,10 @@ pub(crate) struct SyncSession {
     /// How long to wait for it; doubles while answers stay out.
     pub(crate) reply_timeout: Duration,
 
-    /// Who sends the first message of this session.
-    pub(crate) opening: Opening,
-
-    /// What started this session.
-    pub(crate) cause: SessionCause,
-
     /// Consecutive exchanges that changed nothing (livelock detection).
     pub(crate) stalled_exchanges: u32,
 }
 
-/// Who opens a sync session. After a reset, usually both sides do it and both would
-/// open at once. The openings cross, and each side answers the other's: harmless between
-/// peers that share history, but an empty peer answers with empty heads, which Automerge
-/// takes for "lost everything", so the whole document goes out a second time. So only
-/// the side that leads the pair opens.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub(crate) enum Opening {
-    /// We open: we lead the pair.
-    Ours,
-    /// We open, leader or not: we just switched epochs, and the peer, already on our new
-    /// epoch, still has our old one on record. It won't sync with us until a message
-    /// tells it otherwise, and our opening does (no epoch announcement PING needed).
-    /// Survives resets until the peer has answered, since only then do we know it
-    /// learned our epoch.
-    Announce,
-    /// The peer opens. If nothing came by `until`, we open anyway: the peer may not have
-    /// reset, or its opening got lost.
-    Theirs { until: Instant },
-}
-
-/// What started a sync session. Kept for the log: when the peer doesn't open within the
-/// grace period, the cause usually tells why.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub(crate) enum SessionCause {
-    /// First session with a peer we just learned about.
-    NewPeer,
-    /// The peer sent HELLO: it (re)started its handshake with us.
-    PeerHello,
-    /// Its HELLO_REPLY completed our handshake.
-    HandshakeComplete,
-    /// The peer advertised another epoch than before.
-    PeerEpochChanged,
-    /// We switched epochs.
-    OwnEpochSwitch,
-    /// The peer went dormant.
-    PeerDormant,
-    /// The peer failed the swarm key check.
-    KeyRejected,
-}
-
-impl SessionCause {
-    /// The likely reason the peer didn't open a session started this way.
-    pub(crate) fn missing_opening_hint(self) -> &'static str {
-        match self {
-            Self::NewPeer | Self::PeerHello | Self::HandshakeComplete => {
-                "the peer may not count us as a member yet (our HELLO_REPLY hasn't reached it), \
-                 or its opening was lost"
-            }
-            Self::PeerEpochChanged => {
-                "the peer switched epochs but doesn't know we're on its new one yet, or its \
-                 opening was lost"
-            }
-            Self::OwnEpochSwitch => {
-                "we switched epochs; the peer opens only once it learns that, and it may not \
-                 have yet"
-            }
-            Self::PeerDormant | Self::KeyRejected => {
-                "only our side reset, so the peer had no reason to open"
-            }
-        }
-    }
-}
-
 /// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery:
 /// a lost message leaves both sides waiting (`in_flight`), and changes it carried are
 /// never sent again (`sent_hashes`). We resend after a timeout instead.
@@ -227,47 +158,27 @@ const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(10);
 /// Upper bound (Mycelium gives up on a message after five minutes).
 const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300);
 
-/// How long the side that doesn't open a fresh sync waits for the other's opening
-/// before opening itself (the other one may not have reset, or its message got lost).
-const SYNC_OPENING_GRACE: Duration = Duration::from_secs(3);
-
 impl SyncSession {
-    pub(crate) fn new(leads: bool, cause: SessionCause) -> Self {
+    pub(crate) fn new() -> Self {
         Self {
-            cause,
             state: sync::State::new(),
             outbound: None,
             reply: None,
             reply_timeout: SYNC_REPLY_TIMEOUT,
-            opening: if leads {
-                Opening::Ours
-            } else {
-                Opening::Theirs {
-                    until: Instant::now() + SYNC_OPENING_GRACE,
-                }
-            },
             stalled_exchanges: 0,
         }
     }
 
-    /// How much longer to hold back our opening, if at all. Only the first message
-    /// counts: once the peer has said anything, we answer as usual.
-    pub(crate) fn opening_wait(&self, now: Instant) -> Option<Duration> {
-        if self.state.their_heads.is_some() {
-            return None;
-        }
-        match self.opening {
-            Opening::Ours | Opening::Announce => None,
-            Opening::Theirs { until } => until
-                .checked_duration_since(now)
-                .filter(|wait| !wait.is_zero()),
-        }
-    }
-
-    /// The peer should have opened by now but hasn't said anything: we open ourselves.
-    pub(crate) fn opening_overdue(&self, now: Instant) -> bool {
-        self.state.their_heads.is_none()
-            && matches!(self.opening, Opening::Theirs { until } if until <= now)
+    /// Whether we leave the first message of this session to the peer.
+    ///
+    /// After a reset, both sides usually open at once, and the openings cross. Between
+    /// peers that share history that costs nothing. But a peer with an empty document
+    /// answers our opening with empty heads, which Automerge takes for "lost
+    /// everything", so the whole document would go out a second time. So a ready node
+    /// never opens towards a peer that isn't ready (one that is still catching up); that
+    /// peer opens itself.
+    pub(crate) fn leave_opening(&self, we_ready: bool, peer_ready: Option<bool>) -> bool {
+        self.state.their_heads.is_none() && we_ready && peer_ready == Some(false)
     }
 
     /// A sync message went out; `sent` are the changes it carried for the first time.
@@ -314,7 +225,7 @@ impl PeerState {
             leads,
             standing: Standing::default(),
             hello_state: HelloState::default(),
-            sync: SyncSession::new(leads, SessionCause::NewPeer),
+            sync: SyncSession::new(),
             liveness: Liveness::Active,
             added_at: Instant::now(),
             last_seen: None,
@@ -330,13 +241,8 @@ impl PeerState {
 
     /// Start Automerge sync with this peer from scratch. Any envelope generated against
     /// the old state is discarded; a stale RetryDelivery event is then a no-op.
-    pub(crate) fn reset_sync(&mut self, cause: SessionCause) {
-        let announce =
-            self.sync.opening == Opening::Announce && self.sync.state.their_heads.is_none();
-        self.sync = SyncSession::new(self.leads, cause);
-        if announce {
-            self.sync.opening = Opening::Announce;
-        }
+    pub(crate) fn reset_sync(&mut self) {
+        self.sync = SyncSession::new();
     }
 
     /// An overdue sync answer from an established, active peer (see
@@ -374,7 +280,7 @@ impl PeerState {
         }
 
         self.liveness = Liveness::Dormant { since: now };
-        self.reset_sync(SessionCause::PeerDormant);
+        self.reset_sync();
 
         if !matches!(self.hello_state, HelloState::Sending { .. }) {
             self.hello_state = HelloState::AwaitingReply {
@@ -782,22 +688,21 @@ impl AppState {
         self.publish_change();
 
         // Tell peers about the new epoch now instead of with the next keepalive (up to a
-        // keepalive interval later). Peers already on it learn it from our sync opening,
-        // which goes out right away; everyone else gets a PING, so peers that never saw
-        // the commit start catching up (we don't sync with them, so nothing else would
-        // tell them).
+        // keepalive interval later). Ready peers already on it learn it from our sync
+        // opening, which goes out right away; everyone else gets a PING, so peers that
+        // never saw the commit start catching up (we don't sync with them, so nothing
+        // else would tell them).
         let local = self.local_epoch();
         let mut announce = Vec::new();
         for (id, peer_state_mutex) in &peers {
             let mut peer_state = peer_state_mutex.lock().await;
             // What it said about syncing with us was about our old epoch.
             peer_state.syncs_with_us = None;
-            let syncs_with_us = peer_state.liveness == Liveness::Active
+            let gets_our_opening = peer_state.liveness == Liveness::Active
                 && matches!(peer_state.standing, Standing::Member { .. })
-                && peer_state.epoch == Some(local);
-            if syncs_with_us {
-                peer_state.sync.opening = Opening::Announce;
-            } else {
+                && peer_state.epoch == Some(local)
+                && peer_state.reports_ready == Some(true);
+            if !gets_our_opening {
                 announce.push(id);
             }
         }
@@ -1067,7 +972,7 @@ impl Host for AppState {
                     let now = Instant::now();
                     for peer in state.peer_states.blocking_read().values() {
                         let mut peer = peer.blocking_lock();
-                        peer.reset_sync(SessionCause::OwnEpochSwitch);
+                        peer.reset_sync();
                         // A fresh start, not an idle sync: no heal re-handshake for it.
                         peer.last_sync_rx = Some(now);
                     }
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()
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index d96b059..83c4e8d 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -118,6 +118,8 @@ pub struct HelloRequest {
     pub epoch: Epoch,
     /// Proof of the swarm key (`swarm::auth`); empty without one.
     pub auth: Vec<u8>,
+    /// The sender is ready (same meaning as `SyncEnvelope::ready`).
+    pub ready: bool,
 }
 
 #[derive(Debug, Clone)]
@@ -127,6 +129,8 @@ pub struct HelloReply {
     pub epoch: Epoch,
     /// Proof of the swarm key (`swarm::auth`); empty without one.
     pub auth: Vec<u8>,
+    /// The sender is ready.
+    pub ready: bool,
 }
 
 #[derive(Debug, Clone)]
diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
index 2406312..4a8a5a5 100644
--- a/src/transport/mycelium.rs
+++ b/src/transport/mycelium.rs
@@ -830,6 +830,7 @@ pub struct HelloRequestShadow {
     swarm_id: Option<String>,
     epoch: EpochShadow,
     auth: ByteBuf,
+    ready: bool,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -842,6 +843,7 @@ pub struct HelloReplyShadow<'a> {
     epoch: EpochShadow,
     #[serde(default)]
     auth: ByteBuf,
+    ready: bool,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -891,6 +893,7 @@ pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::
         swarm_id: msg.swarm_id.clone(),
         epoch: epoch_to_shadow(&msg.epoch),
         auth: ByteBuf::from(msg.auth.clone()),
+        ready: msg.ready,
     })
 }
 pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> {
@@ -899,6 +902,7 @@ pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportErr
         swarm_id: shadow.swarm_id,
         epoch: epoch_from_shadow(shadow.epoch)?,
         auth: shadow.auth.into_vec(),
+        ready: shadow.ready,
     })
 }
 
@@ -909,6 +913,7 @@ pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::enco
         peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching
         epoch: epoch_to_shadow(&msg.epoch),
         auth: ByteBuf::from(msg.auth.clone()),
+        ready: msg.ready,
     };
     rmp_serde::to_vec(&shadow)
 }
@@ -919,6 +924,7 @@ pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError>
         peers: shadow.peers,
         epoch: epoch_from_shadow(shadow.epoch)?,
         auth: shadow.auth.into_vec(),
+        ready: shadow.ready,
     })
 }