1 files changed, 75 insertions, 0 deletions

diff --git a/src/state/session.rs b/src/state/session.rs
index 618d65b..6efe538 100644
--- a/src/state/session.rs
+++ b/src/state/session.rs
@@ -52,6 +52,11 @@ pub(crate) struct SyncSession {
     /// The wait after the next refusal.
     pub(crate) send_backoff: Backoff,
 
+    /// We offered the peer a delta instead of a message too large to send
+    /// (`withdraw_for_delta`): no more messages until it answers, or the answer is
+    /// overdue (then the offer goes out again).
+    pub(crate) delta_offered: bool,
+
     /// The answer we expect to our last sync message(s), if any.
     pub(crate) reply: Option<SyncReplyWait>,
     /// How long to wait for it; doubles while answers stay out.
@@ -80,6 +85,7 @@ impl SyncSession {
             sending: None,
             retry_at: None,
             send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX),
+            delta_offered: false,
             reply: None,
             reply_backoff,
         }
@@ -134,9 +140,30 @@ impl SyncSession {
         doc: &mut AutoCommit,
         message: sync::Message,
     ) -> Result<(), AutomergeError> {
+        self.delta_offered = false;
         doc.sync().receive_sync_message(&mut self.state, message)
     }
 
+    /// The message just generated is too large to send in one piece: take back what it
+    /// carried, as if it had never been generated, and offer a delta instead. The
+    /// 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<automerge::ChangeHash> {
+        if let Some(sending) = &mut self.sending {
+            let sent = std::mem::take(&mut sending.sent);
+            for hash in &sent {
+                self.state.sent_hashes.remove(hash);
+            }
+            if let Some(wait) = &mut self.reply {
+                wait.sent.retain(|hash| !sent.contains(hash));
+            }
+        }
+        self.state.in_flight = false;
+        self.delta_offered = true;
+        self.state.their_heads.clone().unwrap_or_default()
+    }
+
     /// 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`.
@@ -282,6 +309,7 @@ impl SyncSession {
             self.state.sent_hashes.remove(hash);
         }
         self.state.in_flight = false;
+        self.delta_offered = false;
         self.reply_backoff.grow();
         Some(wait.sent.len())
     }
@@ -323,4 +351,51 @@ mod tests {
         assert!(!new.send_succeeded(new.id()));
         assert!(new.sending.is_none());
     }
+
+    /// 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]
+    fn withdrawing_for_a_delta_keeps_the_changes_for_later() {
+        use automerge::{ROOT, transaction::Transactable};
+
+        // 16 KiB of hex, about 8 KiB of information: Automerge compresses what it sends
+        // (a peer with nothing gets the document saved), so this stays above 4 KB.
+        let mut rng = crate::testrand::Rng::new(1);
+        let big: String = (0..1024)
+            .map(|_| format!("{:016x}", rng.next_u64()))
+            .collect();
+        let mut ours = AutoCommit::new();
+        ours.put(ROOT, "big", big).unwrap();
+        ours.commit();
+        let mut theirs = AutoCommit::new();
+        let (mut a, mut b) = (session(), session());
+        let now = Instant::now();
+        let bytes = SyncBytes::default();
+
+        // a opens, b answers with its (empty) heads, then a sends the changes.
+        let opening = a.generate(&mut ours, now, bytes).unwrap();
+        a.send_succeeded(a.id());
+        b.receive(&mut theirs, opening).unwrap();
+        let answer = b.generate(&mut theirs, now, bytes).unwrap();
+        b.send_succeeded(b.id());
+        a.receive(&mut ours, answer).unwrap();
+        let carrying = a.generate(&mut ours, now, bytes).unwrap();
+        let carried = carrying.clone().encode().len();
+        assert!(carried > 4_000, "the changes took {carried} bytes");
+
+        let from = a.withdraw_for_delta();
+        assert!(from.is_empty());
+        assert!(a.delta_offered);
+        assert!(!a.send_succeeded(a.id()));
+
+        // No answer to the offer: it lapses, and the changes go out again.
+        assert!(
+            a.take_overdue_reply(now + Duration::from_secs(3600))
+                .is_some()
+        );
+        assert!(!a.delta_offered);
+        let again = a.generate(&mut ours, now, bytes).unwrap();
+        let carried_again = again.encode().len();
+        assert!(carried_again > 4_000, "sent again in {carried_again} bytes");
+    }
 }