offered deltas lapse

2 files changed, 52 insertions, 7 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index 6eb791e..e921b22 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -267,7 +267,15 @@ impl SyncSession {
     /// message being handed over becomes the offer (it carries nothing); the answer is
     /// still awaited, and if it doesn't come, the offer is made again. Returns the
     /// peer's heads as far as we know them: where the delta starts.
-    pub(crate) fn withdraw_for_delta(&mut self) -> Vec<crate::document::ChangeHash> {
+    ///
+    /// An answer is awaited even where none was (the peer's heads were ours: `generate`),
+    /// so the offer lapses at the reply timeout at the latest: nothing else ends it but a
+    /// message from the peer, and one that has everything has nothing to say. Without
+    /// that, no message would go to it again.
+    pub(crate) fn withdraw_for_delta(&mut self, now: Instant) -> Vec<crate::document::ChangeHash> {
+        if self.reply.is_none() {
+            self.expect_reply(Vec::new(), now, SyncBytes::default(), 0);
+        }
         if let Some(sending) = &mut self.sending {
             let sent = std::mem::take(&mut sending.sent);
             for hash in &sent {
@@ -283,6 +291,15 @@ impl SyncSession {
         self.state.their_heads.clone().unwrap_or_default()
     }
 
+    /// The message just generated, withdrawn for a delta (`withdraw_for_delta`), turns
+    /// out to carry nothing the peer lacks (it has our heads): none goes, and nothing is
+    /// offered or awaited.
+    pub(crate) fn nothing_to_send(&mut self) {
+        self.sending = None;
+        self.delta_offered = false;
+        self.reply_settled();
+    }
+
     /// Whether `doc` holds everything the peer had when it last told us its heads: they
     /// are known, not empty (empty heads are "contained" in every document, so they
     /// prove nothing), and all in `doc`.
@@ -712,6 +729,24 @@ mod tests {
         assert!(new.sending.is_none());
     }
 
+    /// A delta offered where no answer was awaited (the peer's heads were ours) still
+    /// lapses at the reply timeout: else nothing ends the wait for a peer with nothing
+    /// to say, and no message goes to it again.
+    #[test]
+    fn an_offered_delta_always_lapses() {
+        let mut a = session();
+        a.sending = sending();
+        assert!(a.reply.is_none());
+        let now = Instant::now();
+        a.withdraw_for_delta(now);
+        assert!(a.delta_offered);
+        assert!(
+            a.take_overdue_reply(now + Duration::from_secs(3600))
+                .is_some()
+        );
+        assert!(!a.delta_offered);
+    }
+
     /// A message too large to send is taken back: the peer, after the offer and an
     /// overdue answer, gets the same changes in the next message.
     #[test]
@@ -743,7 +778,7 @@ mod tests {
         let carried = carrying.0.len();
         assert!(carried > 4_000, "the changes took {carried} bytes");
 
-        let from = a.withdraw_for_delta();
+        let from = a.withdraw_for_delta(now);
         assert!(from.is_empty());
         assert!(a.delta_offered);
         assert!(!a.send_succeeded(a.id()));
@@ -1045,7 +1080,7 @@ mod tests {
         b.send_succeeded(b.id());
         a.receive(&mut ours, decode(&opening).unwrap()).unwrap();
         assert!(a.generate(&ours, now, bytes).is_some());
-        a.withdraw_for_delta();
+        a.withdraw_for_delta(now);
         a.send_succeeded(a.id());
 
         // b loads the delta and tells a its heads, which are a's now.
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index de2d86d..10ebf9f 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -162,22 +162,32 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                         peer_state
                             .sync
                             .generate(&replica.doc, Instant::now(), bytes_before);
-                    let body = message.map(|payload| {
+                    let body = message.and_then(|payload| {
                         if payload.0.len() <= delta_threshold {
-                            return transport::SyncBody::Message(payload);
+                            return Some(transport::SyncBody::Message(payload));
                         }
                         // Too large for one message: all or nothing, and on a slow link
                         // within the transport's delivery limit. The peer fetches it in
                         // chunks.
-                        let from = peer_state.sync.withdraw_for_delta();
+                        let from = peer_state.sync.withdraw_for_delta(Instant::now());
                         let spec = transport::DeltaSpec::new(from, replica.heads().to_vec());
+                        if spec.to.iter().all(|head| spec.from.contains(head)) {
+                            // It has our heads: nothing to offer (Automerge thought it
+                            // lacked what it has, as from a view of it that was old).
+                            debug!(
+                                message_bytes = payload.0.len(),
+                                "Sync message too large, but the peer has our heads; not sent"
+                            );
+                            peer_state.sync.nothing_to_send();
+                            return None;
+                        }
                         info!(
                             message_bytes = payload.0.len(),
                             from = spec.from.len(),
                             to = spec.to.len(),
                             "Sync message too large; offering the changes as a blob"
                         );
-                        transport::SyncBody::Delta(spec)
+                        Some(transport::SyncBody::Delta(spec))
                     });
                     let delivered = level && matches!(body, Some(transport::SyncBody::Message(_)));
                     if delivered {