handed over stays sent

1 files changed, 35 insertions, 2 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index c703300..1fc5589 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -882,13 +882,19 @@ impl SyncSession {
 
     /// 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.
+    /// changes that concerns. Those of a message still being handed over stay sent: they
+    /// are on their way (and taken back if its send fails, `send_failed`).
     pub(crate) fn take_overdue_reply(&mut self, now: Instant) -> Option<usize> {
         if self.reply.as_ref().is_none_or(|wait| now < wait.deadline) {
             return None;
         }
         let wait = self.reply.take()?;
-        for hash in &wait.sent {
+        let handing_over: HashSet<_> = self
+            .sending
+            .iter()
+            .flat_map(|sending| sending.sent.iter().copied())
+            .collect();
+        for hash in wait.sent.iter().filter(|hash| !handing_over.contains(hash)) {
             self.state.sent_hashes.remove(hash);
         }
         self.state.in_flight = false;
@@ -1154,6 +1160,33 @@ mod tests {
         assert!(carried_again > 4_000, "sent again in {carried_again} bytes");
     }
 
+    /// A reply overdue while its message is still being handed over: what that message
+    /// carries stays sent (it is on its way), so the next pass doesn't send it twice.
+    #[test]
+    fn what_is_being_handed_over_stays_sent() {
+        use crate::document::testing::change;
+        use automerge::{ROOT, transaction::Transactable};
+        let (mut ours, mut theirs) = (Document::new(), Document::new());
+        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
+        let (mut a, _b) = in_step(&mut ours, &mut theirs);
+        let (now, bytes) = (Instant::now(), SyncBytes::default());
+        change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap());
+        a.generate(&ours, now, bytes).expect("news");
+        assert!(a.reply.is_some(), "an answer awaited");
+        let news = ours.get_heads()[0];
+        assert!(a.state.sent_hashes.contains(&news));
+
+        // Overdue while still in the transport's hands: still sent.
+        assert!(
+            a.take_overdue_reply(now + Duration::from_secs(3600))
+                .is_some()
+        );
+        assert!(a.state.sent_hashes.contains(&news), "on its way");
+        // The send fails: taken back, to go again.
+        a.send_failed(a.id(), now);
+        assert!(!a.state.sent_hashes.contains(&news));
+    }
+
     /// Two sessions exchanging messages until neither has anything to say.
     fn in_step(ours: &mut Document, theirs: &mut Document) -> (SyncSession, SyncSession) {
         let (mut a, mut b) = (session(), session());