open sync only from one side

If both sides open, the openings cross, and an empty node's answer to
the other opening (empty heads) makes the other side send the whole
document a second time.

Now after a sync reset, only the side with the smaller public key opens
the sync; the other answers it, or opens itself if nothing arrives.

2 files changed, 45 insertions, 3 deletions

diff --git a/src/state.rs b/src/state.rs
index 9c3b569..3fa01c2 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -110,6 +110,9 @@ pub(crate) struct PeerState {
     pub(crate) sync_reply: Option<SyncReplyWait>,
     /// How long to wait for it; doubles while answers stay out.
     pub(crate) sync_reply_timeout: Duration,
+
+    /// When sync with this peer last started from scratch.
+    pub(crate) sync_reset_at: Instant,
 }
 
 /// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery:
@@ -128,6 +131,10 @@ 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 Default for PeerState {
     fn default() -> Self {
         Self {
@@ -148,6 +155,7 @@ impl Default for PeerState {
             authenticated: false,
             sync_reply: None,
             sync_reply_timeout: SYNC_REPLY_TIMEOUT,
+            sync_reset_at: Instant::now(),
             epoch: None,
         }
     }
@@ -162,6 +170,22 @@ impl PeerState {
         self.needs_generate = false;
         self.sync_reply = None;
         self.sync_reply_timeout = SYNC_REPLY_TIMEOUT;
+        self.sync_reset_at = Instant::now();
+    }
+
+    /// 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 with the smaller key opens; the other waits for its opening, up to
+    /// `SYNC_OPENING_GRACE`. Returns how much longer to wait, if at all.
+    pub(crate) fn opening_wait(&self, we_open: bool, now: Instant) -> Option<Duration> {
+        if we_open || self.sync_state.their_heads.is_some() {
+            return None;
+        }
+        SYNC_OPENING_GRACE
+            .checked_sub(now.saturating_duration_since(self.sync_reset_at))
+            .filter(|wait| !wait.is_zero())
     }
 
     /// A sync message went out; `sent` are the changes it carried for the first time.
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 47c3b7f..d82238e 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -58,10 +58,28 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
             return Ok(());
         }
 
-        match &peer_state.membership {
-            PeerMembership::Unknown => return Ok(()),
-            PeerMembership::Member { swarm_id } => swarm_id.clone(),
+        let PeerMembership::Member { swarm_id } = &peer_state.membership else {
+            return Ok(());
+        };
+
+        if !retry_delivery {
+            let we_open = state.local_id() < peer;
+            if let Some(wait) = peer_state.opening_wait(we_open, Instant::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));
+                });
+                return Ok(());
+            }
         }
+
+        swarm_id.clone()
     };
 
     let swarm_id = state