diff options
Diffstat (limited to 'src/document')
| -rw-r--r-- | src/document/sync.rs | 90 |
1 files changed, 90 insertions, 0 deletions
diff --git a/src/document/sync.rs b/src/document/sync.rs index 5d139e2..5669ce3 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -825,6 +825,56 @@ impl SyncSession { true } + /// The peer said its `heads` outside the sync (a keepalive STATUS): it has every + /// change in their history. Automerge forgets changes as sent only when the peer's + /// sync message says it has them, and a neighbour another parent brings the changes + /// to (`known_to_peer`) sends us none while it has nothing to write: counted as sent, + /// every change would stay so. So those in the history of `heads` count as sent no + /// more, and the peer as having all up to `heads`: the next message to it carries + /// what lies beyond them, not those. Only forward: where `heads` aren't all ours, or + /// don't cover what the peer said before or what we take it to have (a message of + /// ours still on its way, `assume_delivered`), nothing changes. Returns how many + /// changes count as sent no more. + pub(crate) fn heard_heads( + &mut self, + doc: &Document, + heads: &[crate::document::ChangeHash], + ) -> usize { + use crate::document::changes::{covers, holds_all}; + if heads.is_empty() || !holds_all(doc, heads) { + return 0; + } + let behind = |older: &[crate::document::ChangeHash]| !covers(doc, heads, older); + if self.said_heads.as_deref().is_some_and(behind) + || self.state.their_heads.as_deref().is_some_and(behind) + || self + .state + .their_have + .iter() + .flatten() + .any(|have| behind(&have.last_sync)) + { + return 0; + } + let beyond: HashSet<crate::document::ChangeHash> = doc + .get_changes_meta(heads) + .iter() + .map(|change| change.hash) + .collect(); + let before = self.state.sent_hashes.len(); + self.state + .sent_hashes + .retain(|hash| beyond.contains(hash) || doc.get_change_by_hash(hash).is_none()); + self.state.their_heads = Some(heads.to_vec()); + self.state.shared_heads = heads.to_vec(); + self.state.their_have = Some(vec![sync::Have { + last_sync: heads.to_vec(), + bloom: sync::BloomFilter::default(), + }]); + self.said_heads = Some(heads.to_vec()); + before - self.state.sent_hashes.len() + } + /// The peer has heads we lack (its keepalive said so): ask for what they need, /// even though Automerge thinks a message is still in flight. pub(crate) fn pull_from(&mut self, heads: Vec<crate::document::ChangeHash>) { @@ -1340,6 +1390,46 @@ mod tests { assert_eq!(a.sent_hashes_len(), before); } + /// A neighbour another parent brings our changes to sends us no sync message: what + /// counts as sent to it is forgotten once its keepalive says its heads, and nothing + /// goes to it again. + #[test] + fn heads_heard_forget_what_counts_as_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()); + for i in 0..100 { + change(&mut ours, |doc| doc.put(ROOT, "key", i).unwrap()); + a.known_to_peer(&ours.get_heads(), now); + } + theirs.merge(&mut ours.clone()).unwrap(); + assert_eq!(a.sent_hashes_len(), 100); + + // Heads it said before we had them all: nothing changes. + assert_eq!( + a.heard_heads(&ours, &[crate::document::ChangeHash([7; 32])]), + 0 + ); + assert_eq!(a.heard_heads(&ours, &theirs.get_heads()), 100); + assert_eq!(a.sent_hashes_len(), 0); + let carries = |message: Option<SyncPayload>| { + message.is_some_and(|message| decode(&message).unwrap().carries_changes()) + }; + assert!(!carries(a.generate(&ours, now, bytes))); + + // Older heads than those, said later (a STATUS overtaken): nothing changes. + change(&mut ours, |doc| doc.put(ROOT, "key", 100).unwrap()); + a.known_to_peer(&ours.get_heads(), now); + let older = theirs.get_heads(); + theirs.merge(&mut ours.clone()).unwrap(); + assert_eq!(a.heard_heads(&ours, &theirs.get_heads()), 1); + assert_eq!(a.heard_heads(&ours, &older), 0); + assert!(!carries(a.generate(&ours, now, bytes))); + } + /// 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`). |