announce epochs by sync instead of ping

2 files changed, 33 insertions, 4 deletions

diff --git a/src/state.rs b/src/state.rs
index e1e375a..ccc8253 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -113,6 +113,12 @@ pub(crate) struct PeerState {
 
     /// When sync with this peer last started from scratch.
     pub(crate) sync_reset_at: Instant,
+
+    /// 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,
 }
 
 /// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery:
@@ -156,6 +162,7 @@ impl Default for PeerState {
             sync_reply: None,
             sync_reply_timeout: SYNC_REPLY_TIMEOUT,
             sync_reset_at: Instant::now(),
+            epoch_unannounced: false,
             epoch: None,
         }
     }
@@ -179,8 +186,11 @@ impl PeerState {
     /// 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.sync_state.their_heads.is_some() {
+        if we_open || self.epoch_unannounced || self.sync_state.their_heads.is_some() {
             return None;
         }
         SYNC_OPENING_GRACE
@@ -649,10 +659,25 @@ impl AppState {
         self.publish_change();
 
         // Tell peers about the new epoch now instead of with the next keepalive (up to a
-        // keepalive interval later): sync with peers already on it resumes right away,
-        // and peers that never saw the commit start catching up.
+        // 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).
+        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;
+            let syncs_with_us = peer_state.liveness == Liveness::Active
+                && matches!(peer_state.membership, PeerMembership::Member { .. })
+                && peer_state.epoch == Some(local);
+            peer_state.epoch_unannounced = syncs_with_us;
+            if !syncs_with_us {
+                announce.push(id);
+            }
+        }
+
         let keepalive = crate::swarm::peer::local_keepalive(self).await;
-        for (id, _) in &peers {
+        for id in announce {
             if let Err(error) = self
                 .transport
                 .send_message(id, &transport::ParsedMessage::Ping(keepalive.clone()))
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 2c2079a..567b1ea 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -149,6 +149,10 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
                                 .collect();
                             peer_state.expect_sync_reply(sent, Instant::now());
                         }
+                        if message.is_some() {
+                            // It carries our epoch.
+                            peer_state.epoch_unannounced = false;
+                        }
                         (epoch, message)
                     };
                     let Some(message) = message else {