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, 26 insertions, 121 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);
                     }