1 files changed, 54 insertions, 16 deletions

diff --git a/src/state.rs b/src/state.rs
index 49f75dc..e1e375a 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -2,13 +2,13 @@
 
 use std::collections::HashMap;
 use std::future::Future;
-use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
 use std::time::Duration;
 
-use anyhow::{bail, Result};
-use automerge::{sync, AutoCommit, ChangeHash};
-use tokio::sync::{broadcast, mpsc, watch, Mutex, RwLock};
+use anyhow::{Result, bail};
+use automerge::{AutoCommit, ChangeHash, sync};
+use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch};
 use tokio::time::Instant;
 use tracing::{debug, error, warn};
 
@@ -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 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.
+    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.
@@ -178,6 +202,12 @@ impl PeerState {
         }
     }
 
+    /// Keepalives show both documents are the same: stop waiting for an answer.
+    pub(crate) fn sync_in_step(&mut self) {
+        self.sync_reply = None;
+        self.sync_reply_timeout = SYNC_REPLY_TIMEOUT;
+    }
+
     /// The peer answered.
     pub(crate) fn sync_reply_received(&mut self) {
         self.sync_reply = None;
@@ -244,11 +274,29 @@ impl PeerState {
     }
 
     /// True (and records the ping) if an established, quiet peer needs a keepalive.
-    pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool {
+    ///
+    /// 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
+    /// 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 {
+            timing.keepalive
+        } else {
+            timing.keepalive + timing.keepalive / 2
+        };
         let due = self.liveness == Liveness::Active
             && self.hello_state == HelloState::Established
             && now.saturating_duration_since(self.added_at) >= timing.keepalive
-            && self.quiet_for(now) >= timing.keepalive
+            && self.quiet_for(now) >= quiet_enough
             && self
                 .last_ping_at
                 .is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive);
@@ -759,7 +807,6 @@ impl Host for AppState {
                 summaries.push(PeerSummary {
                     id,
                     active: peer.liveness == Liveness::Active,
-                    established: peer.hello_state == HelloState::Established,
                     epoch: peer.epoch,
                 });
             }
@@ -828,15 +875,6 @@ impl Host for AppState {
         async move { state.render(f, |value| value).await }
     }
 
-    fn update_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send
-    where
-        F: FnOnce(&mut AutoCommit) -> T + Send + 'static,
-        T: Send + 'static,
-    {
-        let state = self.clone();
-        async move { state.write_doc(f).await }
-    }
-
     fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send
     where
         F: FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static,