resend sync messages that haven't been replied

2 files changed, 23 insertions, 6 deletions

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