1 files changed, 67 insertions, 0 deletions

diff --git a/src/state.rs b/src/state.rs
index 8462f88..ede5950 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 {