From 645ea5c150df5fd55493d91fd5d8ada614bd1252 Mon Sep 17 00:00:00 2001 From: tv Date: Fri, 6 Nov 2026 04:33:08 +0100 Subject: sent hashes trimmed --- src/document/sync.rs | 83 ++++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 78 insertions(+), 5 deletions(-) (limited to 'src/document/sync.rs') 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, + /// 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)>, } /// 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, + /// 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>, /// 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 = self + .assumed_lately + .iter() + .flatten() + .chain(self.known_elsewhere.keys()) + .copied() + .collect(); + let trimmed: Vec = 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`). -- cgit v1.3.1