From 784c6e7c9088dde2f23cc1d02381a0c5fe8b010b Mon Sep 17 00:00:00 2001 From: tv Date: Thu, 5 Nov 2026 04:50:17 +0100 Subject: pushed heads as tips --- src/document/sync.rs | 66 ++++++++++++++++++++++++++++++++++++++++++++++----- src/transport/mod.rs | 3 ++- src/transport/wire.rs | 2 +- 3 files changed, 63 insertions(+), 8 deletions(-) diff --git a/src/document/sync.rs b/src/document/sync.rs index e7a2665..9415bd0 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -6,6 +6,7 @@ //! forgetting what was sent when the peer still needs it) live here with it, so an //! Automerge upgrade that changes the state touches this file only. +use std::collections::HashSet; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; @@ -22,6 +23,29 @@ use crate::transport::SyncPayload; /// A sync message from a peer, decoded (`decode`), to apply (`SyncSession::receive`). pub(crate) struct Received(sync::Message); +/// The changes none of the others depend on, sorted as heads are (`get_heads`). +fn tips<'a>( + changes: impl Iterator< + Item = ( + crate::document::ChangeHash, + &'a [crate::document::ChangeHash], + ), + >, +) -> Vec { + let changes: Vec<_> = changes.collect(); + let depended: HashSet = changes + .iter() + .flat_map(|(_, deps)| deps.iter().copied()) + .collect(); + let mut tips: Vec = changes + .into_iter() + .map(|(hash, _)| hash) + .filter(|hash| !depended.contains(hash)) + .collect(); + tips.sort_unstable(); + tips +} + impl Received { /// Whether it carries changes (it may carry only heads and what the peer needs). pub(crate) fn carries_changes(&self) -> bool { @@ -32,11 +56,23 @@ impl Received { /// message that would have brought them: the sender's `heads`, nothing needed, and /// as what it has, everything we had (`ours`, our heads before: it takes the /// message as delivered, so it had all of them). + /// + /// No heads stand for the tips of the changes (`SyncSession::push`). Changes that + /// don't decode leave the heads out; applying them fails as it would have anyway. pub(crate) fn pushed( heads: Vec, changes: Vec>, ours: Vec, ) -> Self { + let heads = if heads.is_empty() { + let decoded: Vec = changes + .iter() + .filter_map(|bytes| automerge::Change::try_from(bytes.as_slice()).ok()) + .collect(); + tips(decoded.iter().map(|change| (change.hash(), change.deps()))) + } else { + heads + }; Self(sync::Message { heads, need: Vec::new(), @@ -352,8 +388,11 @@ impl SyncSession { regenerate: false, assumed: None, }); + // Our heads are mostly the tips of the changes pushed: then they go as none, and + // the receiver finds them in the changes (`Received::pushed`). + let tips = tips(changes.iter().map(|change| (change.hash(), change.deps()))); Some(( - heads, + if tips == heads { Vec::new() } else { heads }, changes .iter() .map(|change| change.raw_bytes().to_vec()) @@ -1453,16 +1492,18 @@ mod tests { change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap()); let (heads, changes) = a.push(&ours, 1 << 20).expect("news to push"); assert_eq!(changes.len(), 1); + assert!( + heads.is_empty(), + "our heads are the change's: they go as none" + ); a.assume_delivered(&ours.get_heads()); a.send_succeeded(a.id()); assert!(a.push(&ours, 1 << 20).is_none(), "sent already"); let before = theirs.get_heads(); - b.receive( - &mut theirs, - Received::pushed(heads, changes, before.clone()), - ) - .unwrap(); + let message = Received::pushed(heads, changes, before.clone()); + assert_eq!(message.heads(), ours.get_heads(), "found in the changes"); + b.receive(&mut theirs, message).unwrap(); b.came_from_peer(&theirs, &before); assert_eq!(theirs.get_heads(), ours.get_heads()); assert!(b.answered_as_delivered(&theirs, &theirs.get_heads())); @@ -1472,6 +1513,19 @@ mod tests { ); } + /// Tips are the changes none of the others depend on, sorted as heads are. + #[test] + fn tips_are_what_nothing_depends_on() { + let hash = |n: u8| crate::document::ChangeHash([n; 32]); + let (a, b, c, d) = (hash(1), hash(2), hash(3), hash(4)); + let deps = [vec![], vec![a], vec![a], vec![b, c, hash(9)]]; + let all = [a, b, c, d]; + let tips_of = |which: &[usize]| tips(which.iter().map(|&i| (all[i], deps[i].as_slice()))); + assert_eq!(tips_of(&[0, 1, 2, 3]), vec![d]); + assert_eq!(tips_of(&[2, 0, 1]), vec![b, c]); + assert_eq!(tips_of(&[]), Vec::new()); + } + /// Changes larger than the limit aren't pushed: they go as `generate` sends them. #[test] fn large_changes_are_not_pushed() { diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 7e5d41b..2595257 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -746,7 +746,8 @@ pub(crate) enum SyncBody { /// Changes passed on along a writer's tree as they are, with the sender's heads, /// instead of an Automerge sync message (`AppState::push_changes`): where the sender /// takes the message as delivered, it has everything the receiver has, and needs - /// none of the rest of one (`SyncSession::push`). + /// none of the rest of one (`SyncSession::push`). No heads stand for the tips of + /// the changes, as the sender's heads mostly are. Changes { #[serde(with = "crate::transport::wire::codec::change_hashes")] heads: Vec, diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 70ac3ce..580712e 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage}; /// The wire format's version, the first byte of every payload. Bump it with every change /// to a message type or a codec (`golden_bytes` fails then). -pub(crate) const VERSION: u8 = 33; +pub(crate) const VERSION: u8 = 34; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { -- cgit v1.3.1