resend sync messages that haven't been replied

3 files changed, 90 insertions, 6 deletions

diff --git a/src/state.rs b/src/state.rs
index e24c882..81a3e6a 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -105,8 +105,29 @@ pub(crate) struct PeerState {
 
     /// Consecutive sync exchanges that changed nothing (livelock detection).
     pub(crate) stalled_exchanges: u32,
+
+    /// The answer we expect to our last sync message(s), if any.
+    pub(crate) sync_reply: Option<SyncReplyWait>,
+    /// How long to wait for it; doubles while answers stay out.
+    pub(crate) sync_reply_timeout: Duration,
+}
+
+/// 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.
+#[derive(Clone, Debug)]
+pub(crate) struct SyncReplyWait {
+    pub(crate) deadline: Instant,
+    /// Changes first sent in the unanswered messages.
+    pub(crate) sent: Vec<automerge::ChangeHash>,
 }
 
+/// First wait for a sync answer; generous, since Mycelium delivers large messages in
+/// many small packets.
+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);
+
 impl Default for PeerState {
     fn default() -> Self {
         Self {
@@ -124,6 +145,8 @@ impl Default for PeerState {
             last_sync_rx: None,
             last_resync_at: None,
             stalled_exchanges: 0,
+            sync_reply: None,
+            sync_reply_timeout: SYNC_REPLY_TIMEOUT,
             epoch: None,
         }
     }
@@ -136,6 +159,50 @@ impl PeerState {
         self.sync_state = sync::State::new();
         self.pending_outbound_sync = None;
         self.needs_generate = false;
+        self.sync_reply = None;
+        self.sync_reply_timeout = SYNC_REPLY_TIMEOUT;
+    }
+
+    /// A sync message went out; `sent` are the changes it carried for the first time.
+    pub(crate) fn expect_sync_reply(&mut self, sent: Vec<automerge::ChangeHash>, now: Instant) {
+        match &mut self.sync_reply {
+            // Still waiting for an earlier one: keep its deadline, remember these too.
+            Some(wait) => wait.sent.extend(sent),
+            None => {
+                self.sync_reply = Some(SyncReplyWait {
+                    deadline: now + self.sync_reply_timeout,
+                    sent,
+                });
+            }
+        }
+    }
+
+    /// The peer answered.
+    pub(crate) fn sync_reply_received(&mut self) {
+        self.sync_reply = None;
+        self.sync_reply_timeout = SYNC_REPLY_TIMEOUT;
+    }
+
+    /// If the answer is overdue: forgets that the unanswered changes were sent and that
+    /// a message is in flight, so the next sync pass sends them again. Returns how many
+    /// changes that concerns.
+    pub(crate) fn take_overdue_sync_reply(&mut self, now: Instant) -> Option<usize> {
+        if self.liveness != Liveness::Active
+            || self.hello_state != HelloState::Established
+            || self
+                .sync_reply
+                .as_ref()
+                .is_none_or(|wait| now < wait.deadline)
+        {
+            return None;
+        }
+        let wait = self.sync_reply.take()?;
+        for hash in &wait.sent {
+            self.sync_state.sent_hashes.remove(hash);
+        }
+        self.sync_state.in_flight = false;
+        self.sync_reply_timeout = (self.sync_reply_timeout * 2).min(SYNC_REPLY_TIMEOUT_MAX);
+        Some(wait.sent.len())
     }
 
     fn quiet_for(&self, now: Instant) -> Duration {
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index b092b03..6184427 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -387,13 +387,21 @@ pub(crate) async fn run_peer_worker(state: AppState) {
         for (peer, peer_state_mutex) in &peers {
             let now = Instant::now();
 
-            let (became_dormant, ping_due) = {
+            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 ping_due = peer_state.take_keepalive_due(&state.timing, now);
-                (became_dormant, ping_due)
+                let unanswered = peer_state.take_overdue_sync_reply(now);
+                (became_dormant, ping_due, unanswered)
             };
 
+            if let Some(changes) = unanswered {
+                debug!(peer = %peer, changes, "No answer to our sync message; sending again");
+                let _ = state
+                    .sync_events
+                    .send(SyncEvent::PeerNeedsSync(peer.clone()));
+            }
+
             if became_dormant {
                 warn!(
                     peer = %peer,
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 115d5c0..04d5b15 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -109,11 +109,19 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
                         let mut peer_state = psm.blocking_lock();
                         peer_state.needs_generate = false;
                         let epoch = replica.epoch;
+                        let sent_before = peer_state.sync_state.sent_hashes.clone();
                         let sync_doc = replica.doc.sync();
-                        (
-                            epoch,
-                            sync_doc.generate_sync_message(&mut peer_state.sync_state),
-                        )
+                        let message = sync_doc.generate_sync_message(&mut peer_state.sync_state);
+                        if message.is_some() {
+                            let sent = peer_state
+                                .sync_state
+                                .sent_hashes
+                                .difference(&sent_before)
+                                .copied()
+                                .collect();
+                            peer_state.expect_sync_reply(sent, Instant::now());
+                        }
+                        (epoch, message)
                     };
                     let Some(message) = message else {
                         return Ok(None);
@@ -290,6 +298,7 @@ pub(crate) async fn handle_sync(
                 .receive_sync_message(&mut peer_state.sync_state, message)
                 .context("Automerge synchronization failed")?;
             peer_state.last_sync_rx = Some(Instant::now());
+            peer_state.sync_reply_received();
 
             let heads_after = replica.doc.get_heads();
             let unproductive = heads_before == heads_after