sent hashes trimmed

1 files changed, 78 insertions, 5 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index 7123138..eed0024 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -120,9 +120,10 @@ pub(crate) struct Sending {
     /// A sync pass was requested meanwhile. Once this one is handed over, the peer must
     /// be enqueued again so the deferred changes go out.
     pub(crate) regenerate: bool,
-    /// Taken as delivered (`assume_delivered`): what we knew of the peer before, put
-    /// back if the send fails.
-    pub(crate) assumed: Option<PeerView>,
+    /// Taken as delivered (`assume_delivered`): what we knew of the peer before, and
+    /// the changes no longer counted as sent then (they are in the heads it is taken to
+    /// have), put back if the send fails.
+    pub(crate) assumed: Option<(PeerView, Vec<crate::document::ChangeHash>)>,
 }
 
 /// What the sync state knows of the peer: its heads, the heads we share, what it has
@@ -186,6 +187,10 @@ pub(crate) struct SyncSession {
     /// and when: still in flight perhaps, so they go along with a change of ours that
     /// builds on them (`with_predecessors`).
     known_elsewhere: std::collections::HashMap<crate::document::ChangeHash, Instant>,
+    /// What the last few messages taken as delivered carried for the first time
+    /// (`ASSUMED_KEPT`): still counted as sent when the rest is no longer
+    /// (`assume_delivered`), as they may still be on their way.
+    assumed_lately: std::collections::VecDeque<Vec<crate::document::ChangeHash>>,
 
     /// The heads the peer said last, in the newest of its messages so far (one that
     /// comes late doesn't count: `older_keeps_newer`).
@@ -203,6 +208,11 @@ const KNOWN_ELSEWHERE_FOR: Duration = Duration::from_secs(2);
 /// How long such a change is remembered as counted so: to tell, should the peer ask for
 /// it, how long after (`SyncSession::resend_if_needed`).
 const KNOWN_ELSEWHERE_KEPT: Duration = Duration::from_secs(30);
+/// Messages taken as delivered whose changes stay counted as sent once the peer is taken
+/// to have them (`SyncSession::assume_delivered`): should one of them be lost, the peer
+/// asks for what it lacks, and those still counted as sent stay so where they may only
+/// be late (`forget_sent_for_need`).
+const ASSUMED_KEPT: usize = 8;
 
 /// 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
@@ -234,6 +244,7 @@ impl SyncSession {
             before_receive: None,
             before_generate: (Vec::new(), false),
             known_elsewhere: std::collections::HashMap::new(),
+            assumed_lately: std::collections::VecDeque::new(),
             said_heads: None,
             reply: None,
             reply_backoff,
@@ -570,7 +581,34 @@ impl SyncSession {
         let Some(sending) = &mut self.sending else {
             return;
         };
-        sending.assumed = Some(view);
+        // Taken to have all up to `heads`, the peer has every change sent before: they
+        // need not count as sent any more. Automerge forgets them only when the peer's
+        // message says it has them, and along a tree none comes back: kept, they would
+        // grow by every change passed on. But for those of the last messages taken as
+        // delivered (they may still be on their way), and those another parent brings
+        // (`with_predecessors` asks whether they count so).
+        self.assumed_lately.push_back(sending.sent.clone());
+        if self.assumed_lately.len() > ASSUMED_KEPT {
+            self.assumed_lately.pop_front();
+        }
+        let kept: HashSet<crate::document::ChangeHash> = self
+            .assumed_lately
+            .iter()
+            .flatten()
+            .chain(self.known_elsewhere.keys())
+            .copied()
+            .collect();
+        let trimmed: Vec<crate::document::ChangeHash> = self
+            .state
+            .sent_hashes
+            .iter()
+            .filter(|hash| !kept.contains(hash))
+            .copied()
+            .collect();
+        for hash in &trimmed {
+            self.state.sent_hashes.remove(hash);
+        }
+        sending.assumed = Some((view, trimmed));
         self.state.their_heads = Some(heads.to_vec());
         self.state.shared_heads = heads.to_vec();
         // And it has all up to our heads, needing nothing: the next message carries
@@ -884,11 +922,12 @@ impl SyncSession {
         for hash in &sending.sent {
             self.state.sent_hashes.remove(hash);
         }
-        if let Some(assumed) = sending.assumed {
+        if let Some((assumed, trimmed)) = sending.assumed {
             self.state.their_heads = assumed.their_heads;
             self.state.shared_heads = assumed.shared_heads;
             self.state.their_have = assumed.their_have;
             self.state.their_need = assumed.their_need;
+            self.state.sent_hashes.extend(trimmed);
         }
         if let Some(wait) = &mut self.reply {
             wait.sent.retain(|hash| !sending.sent.contains(hash));
@@ -1252,6 +1291,40 @@ mod tests {
         assert!(!a.state.sent_hashes.contains(&news));
     }
 
+    /// Messages taken as delivered, one after the other, and never an answer (a tree's
+    /// edge, with one writer): what counts as sent stays at the last few messages', not
+    /// all ever sent; and a send that fails puts back what it no longer counted.
+    #[test]
+    fn a_one_way_session_doesnt_grow() {
+        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, mut b) = in_step(&mut ours, &mut theirs);
+        let (now, bytes) = (Instant::now(), SyncBytes::default());
+        for i in 0..100 {
+            change(&mut ours, |doc| doc.put(ROOT, "key", i).unwrap());
+            let message = a.generate(&ours, now, bytes).expect("the change");
+            a.assume_delivered(&ours.get_heads());
+            a.send_succeeded(a.id());
+            b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
+        }
+        assert_eq!(ours.get_heads(), theirs.get_heads());
+        assert!(
+            a.sent_hashes_len() <= ASSUMED_KEPT,
+            "{} counted as sent",
+            a.sent_hashes_len()
+        );
+
+        // One more, refused by the transport: all counted as before.
+        let before = a.sent_hashes_len();
+        change(&mut ours, |doc| doc.put(ROOT, "key", 100).unwrap());
+        a.generate(&ours, now, bytes).expect("the change");
+        a.assume_delivered(&ours.get_heads());
+        a.send_failed(a.id(), now);
+        assert_eq!(a.sent_hashes_len(), before);
+    }
+
     /// Automerge, as we rely on it: a message with no heads says the peer lost all it
     /// had, and what we counted as its goes again (why a push never goes without heads:
     /// `Received::pushed`, and an opening never goes empty: `leave_opening`).