merge sync_reset_at and epoch_unannounced into opening

3 files changed, 67 insertions, 56 deletions

diff --git a/src/state.rs b/src/state.rs
index 24ae2fb..eaa6562 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -89,6 +89,10 @@ 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.
+    pub(crate) leads: bool,
+
     pub(crate) standing: Standing,
     pub(crate) hello_state: HelloState,
 
@@ -116,16 +120,10 @@ pub(crate) struct PeerState {
 
     /// Epoch the peer advertised in its most recent message.
     pub(crate) epoch: Option<Epoch>,
-
-    /// We switched epochs, and this 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. Our sync
-    /// opening does that, so we open without waiting for the pair's leader, and the
-    /// epoch announcement PING is skipped. Cleared once a sync message is generated.
-    pub(crate) epoch_unannounced: bool,
 }
 
 /// One Automerge sync with a peer, from a reset to the next. Everything a reset must
-/// forget lives here, so a reset is `SyncSession::new()` and can't miss a field.
+/// forget lives here, so a reset replaces it whole and can't miss a field.
 #[derive(Clone)]
 pub(crate) struct SyncSession {
     pub(crate) state: sync::State,
@@ -138,13 +136,30 @@ pub(crate) struct SyncSession {
     /// How long to wait for it; doubles while answers stay out.
     pub(crate) reply_timeout: Duration,
 
-    /// When this session started.
-    pub(crate) started_at: Instant,
+    /// Who sends the first message of this session.
+    pub(crate) opening: Opening,
 
     /// 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, or 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). A
+    /// reset before the opening goes out falls back to the grace period below.
+    Ours,
+    /// 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 },
+}
+
 /// 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.
@@ -166,17 +181,37 @@ const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300);
 const SYNC_OPENING_GRACE: Duration = Duration::from_secs(3);
 
 impl SyncSession {
-    pub(crate) fn new() -> Self {
+    pub(crate) fn new(leads: bool) -> Self {
         Self {
             state: sync::State::new(),
             outbound: None,
             reply: None,
             reply_timeout: SYNC_REPLY_TIMEOUT,
-            started_at: Instant::now(),
+            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 => None,
+            Opening::Theirs { until } => until
+                .checked_duration_since(now)
+                .filter(|wait| !wait.is_zero()),
+        }
+    }
+
     /// A sync message went out; `sent` are the changes it carried for the first time.
     pub(crate) fn expect_reply(&mut self, sent: Vec<automerge::ChangeHash>, now: Instant) {
         match &mut self.reply {
@@ -215,14 +250,15 @@ impl SyncSession {
     }
 }
 
-impl Default for PeerState {
-    fn default() -> Self {
+impl PeerState {
+    pub(crate) fn new(leads: bool) -> Self {
         Self {
+            leads,
             standing: Standing::default(),
             hello_state: HelloState::default(),
             reply_owed: false,
             pending_inbound_sync: Vec::new(),
-            sync: SyncSession::new(),
+            sync: SyncSession::new(leads),
             liveness: Liveness::Active,
             added_at: Instant::now(),
             last_seen: None,
@@ -230,34 +266,13 @@ impl Default for PeerState {
             last_sync_rx: None,
             last_resync_at: None,
             epoch: None,
-            epoch_unannounced: false,
         }
     }
-}
 
-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) {
-        self.sync = SyncSession::new();
-    }
-
-    /// After a reset, usually both sides do it and both open the sync 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. Only the
-    /// side that leads the pair (`swarm::pair`) opens; the other waits for its opening,
-    /// up to `SYNC_OPENING_GRACE`. Returns how much longer to wait, if at all.
-    ///
-    /// After our epoch switch, we open anyway (`epoch_unannounced`): the peer can't, as
-    /// it still thinks we are on another epoch.
-    pub(crate) fn opening_wait(&self, we_open: bool, now: Instant) -> Option<Duration> {
-        if we_open || self.epoch_unannounced || self.sync.state.their_heads.is_some() {
-            return None;
-        }
-        SYNC_OPENING_GRACE
-            .checked_sub(now.saturating_duration_since(self.sync.started_at))
-            .filter(|wait| !wait.is_zero())
+        self.sync = SyncSession::new(self.leads);
     }
 
     /// An overdue sync answer from an established, active peer (see
@@ -309,19 +324,14 @@ impl PeerState {
     /// True (and records the ping) if an established, quiet peer needs a keepalive.
     ///
     /// One PING and its PONG tell both sides that the other is alive, so only one side of
-    /// a pair needs to ping: the one leading the pair (`we_ping`, `swarm::pair`). Each of its PINGs
+    /// a pair needs to ping: the one leading the pair (`leads`). Each of its PINGs
     /// makes the other side's peer look recently heard from, so the other side never
     /// pings while it does. If the pings stop, the other side pings once it has heard
     /// nothing for half an interval more. Without that, both sides would get due at about
     /// the same time (each one's quiet time starts with the same exchange) and their
     /// PINGs would often cross.
-    pub(crate) fn take_keepalive_due(
-        &mut self,
-        timing: &Timing,
-        we_ping: bool,
-        now: Instant,
-    ) -> bool {
-        let quiet_enough = if we_ping {
+    pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool {
+        let quiet_enough = if self.leads {
             timing.keepalive
         } else {
             timing.keepalive + timing.keepalive / 2
@@ -572,10 +582,16 @@ impl AppState {
         Arc::clone(
             peers
                 .entry(pk.to_owned())
-                .or_insert_with(|| Arc::new(Mutex::new(PeerState::default()))),
+                .or_insert_with(|| self.new_peer_state(pk)),
         )
     }
 
+    /// State for a peer we haven't known so far (not inserted).
+    pub(crate) fn new_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> {
+        let leads = crate::swarm::pair::leads(self.local_id(), pk);
+        Arc::new(Mutex::new(PeerState::new(leads)))
+    }
+
     pub(crate) fn notify_store_changed(&self) {
         self.change_tx.send_modify(|version| {
             *version = version.wrapping_add(1);
@@ -693,8 +709,9 @@ impl AppState {
             let syncs_with_us = peer_state.liveness == Liveness::Active
                 && matches!(peer_state.standing, Standing::Member { .. })
                 && peer_state.epoch == Some(local);
-            peer_state.epoch_unannounced = syncs_with_us;
-            if !syncs_with_us {
+            if syncs_with_us {
+                peer_state.sync.opening = Opening::Ours;
+            } else {
                 announce.push(id);
             }
         }
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index adde444..836a967 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -171,7 +171,7 @@ pub(crate) async fn remember_peer(state: &AppState, peer: &str) {
     let mut peer_states = state.peer_states.write().await;
 
     if !peer_states.contains_key(peer) {
-        peer_states.entry(peer.to_owned()).or_default();
+        peer_states.insert(peer.to_owned(), state.new_peer_state(peer));
         info!(
             peer = %peer,
             peer_count = peer_states.len(),
@@ -410,8 +410,7 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             let (became_dormant, ping_due, unanswered) = {
                 let mut peer_state = peer_state_mutex.lock().await;
                 let became_dormant = ready && peer_state.check_dormancy(&state.timing, now);
-                let we_ping = crate::swarm::pair::leads(state.local_id(), peer);
-                let ping_due = peer_state.take_keepalive_due(&state.timing, we_ping, now);
+                let ping_due = peer_state.take_keepalive_due(&state.timing, now);
                 let unanswered = peer_state.take_overdue_sync_reply(now);
                 (became_dormant, ping_due, unanswered)
             };
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 91c6eee..1335361 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -63,8 +63,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
         };
 
         if !retry_delivery {
-            let we_open = crate::swarm::pair::leads(state.local_id(), peer);
-            if let Some(wait) = peer_state.opening_wait(we_open, Instant::now()) {
+            if let Some(wait) = peer_state.sync.opening_wait(Instant::now()) {
                 trace!(
                     wait_ms = wait.as_millis(),
                     "Leaving the opening sync message to the peer"
@@ -149,10 +148,6 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
                                 .collect();
                             peer_state.sync.expect_reply(sent, Instant::now());
                         }
-                        if message.is_some() {
-                            // It carries our epoch.
-                            peer_state.epoch_unannounced = false;
-                        }
                         (epoch, message)
                     };
                     let Some(message) = message else {