diff options
| author | tv <tv@krebsco.de> | 2026-11-05 16:38:01 +0100 |
|---|---|---|
| committer | tv <tv@krebsco.de> | 2026-11-05 16:38:01 +0100 |
| commit | b3e102076666e515c77c50f0eb1150f44f0874e5 (patch) | |
| tree | 947882a9313721749bd2b303bf7cca1d5a31307c /src/document | |
| parent | 719d1e562eefb770ba87cc65d7e89aa5427eb9c7 (diff) | |
late messages keep newer heads
Diffstat (limited to 'src/document')
| -rw-r--r-- | src/document/sync.rs | 73 |
1 files changed, 73 insertions, 0 deletions
diff --git a/src/document/sync.rs b/src/document/sync.rs index b3f8221..f9f0295 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -187,6 +187,10 @@ pub(crate) struct SyncSession { /// builds on them (`with_predecessors`). known_elsewhere: std::collections::HashMap<crate::document::ChangeHash, Instant>, + /// The heads the peer said last, in the newest of its messages so far (one that + /// comes late doesn't count: `older_keeps_newer`). + said_heads: Option<Vec<crate::document::ChangeHash>>, + /// 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. @@ -230,6 +234,7 @@ impl SyncSession { before_receive: None, before_generate: (Vec::new(), false), known_elsewhere: std::collections::HashMap::new(), + said_heads: None, reply: None, reply_backoff, } @@ -305,11 +310,43 @@ impl SyncSession { ) -> Result<(), AutomergeError> { self.delta_offered = false; self.before_receive = Some(self.peer_view()); + let said = message.0.heads.clone(); doc.receive_sync_message_log_patches(&mut self.state, message.0, log)?; + self.older_keeps_newer(doc, &said); self.known_last_sync(doc); Ok(()) } + /// A message older than one the peer sent before (it came late: overtaken, or a + /// copy) names heads in the history of those it said then: Automerge takes them as + /// the peer's now, and what came since would seem to lack to it, the changes it + /// sent us among them (sent back to it, by a push above all: `push` sends what lies + /// beyond its heads). So the peer's heads, and those we share, are what it said then + /// (`said_heads`). Only by what it said: heads we took for its (a message of ours + /// taken as delivered, `assume_delivered`) may be more than it has, and a peer + /// back from away says older heads than those. Not where it says it has nothing (no + /// heads: it lost everything). + fn older_keeps_newer(&mut self, doc: &Document, said: &[crate::document::ChangeHash]) { + let newer = self.said_heads.clone(); + let older = newer.as_ref().is_some_and(|newer| { + // Older: all its heads in our history, and in that of the heads said before + // (a message whose changes wait for predecessors names heads we lack: + // newer). + !said.is_empty() + && said != newer.as_slice() + && crate::document::changes::holds_all(doc, said) + && crate::document::changes::holds_all(doc, newer) + && crate::document::changes::covers(doc, newer, said) + }); + match newer { + Some(newer) if older => { + self.state.their_heads = Some(newer.clone()); + self.state.shared_heads = newer; + } + _ => self.said_heads = Some(said.to_vec()), + } + } + /// The peer's `have` names what it last shared with us by heads (`last_sync`), and /// one that took a message to us as delivered names the heads that message brought. /// Where the next message overtook it, we don't have those heads yet: Automerge @@ -1293,6 +1330,42 @@ mod tests { assert!(b.have_all_theirs(&theirs)); } + /// A message that comes late (a copy of one before, or overtaken) names heads older + /// than the peer said since: they stay its heads, and what it sent us since doesn't + /// go back to it (as it did by a push: beyond the older heads, and no longer counted + /// as sent once the peer's heads showed it). + #[test] + fn a_late_message_takes_the_peer_back_nowhere() { + 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()); + + change(&mut ours, |doc| doc.put(ROOT, "first", 2).unwrap()); + let first = a.generate(&ours, now, bytes).expect("the first"); + a.send_succeeded(a.id()); + change(&mut ours, |doc| doc.put(ROOT, "then", 3).unwrap()); + let then = ours.get_heads(); + let second = a.generate(&ours, now, bytes).expect("the second"); + a.send_succeeded(a.id()); + + // Both arrive, the second twice (so it no longer counts as sent: the peer's + // heads show it); then the first again, late. + for message in [&first, &second, &second] { + let before = theirs.get_heads(); + b.receive(&mut theirs, decode(message).unwrap()).unwrap(); + b.came_from_peer(&theirs, &before); + } + b.receive(&mut theirs, decode(&first).unwrap()).unwrap(); + assert_eq!(b.state.their_heads.as_deref(), Some(&then[..])); + assert!( + b.push(&theirs, 1 << 20).is_none(), + "nothing of a's goes back to it" + ); + } + /// 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()); |
