From 52e307d7b95dda735d1afbc4fe2dcffaa87512a4 Mon Sep 17 00:00:00 2001 From: tv Date: Thu, 5 Nov 2026 00:29:56 +0100 Subject: delta from held heads --- docs/peers.md | 5 +-- src/document/changes.rs | 21 +++++++------ src/document/sync.rs | 83 ++++++++++++++++++++++++++++++++++++++++--------- src/swarm/sync.rs | 5 ++- src/transport/mod.rs | 4 +-- 5 files changed, 88 insertions(+), 30 deletions(-) diff --git a/docs/peers.md b/docs/peers.md index eb1eb5c..70437ab 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -162,8 +162,9 @@ A sync message larger than 16 chunks (16 × `--blob-chunk`, 1 MiB by default) is sent: it would be all or nothing, sent again whole after a lost copy, and on a slow link may not arrive within the transport's delivery limit (Mycelium's five minutes). The sender takes it back and offers a *delta* instead: the changes in the history of its heads but not of the peer's (as far -as it knows them). The peer fetches them as a blob, in chunks, from the sender and any -other peer holding those heads (every such peer produces the same bytes), loads them, +as it knows them; of those, the ones it holds, and for any it lacks, the heads they +share). The peer fetches them as a blob, in chunks, from the sender and any other peer +holding both sets of heads (every such peer produces the same bytes), loads them, and syncs on; the rest goes in ordinary messages. Until the peer answers, the sender sends it nothing; if the answer is overdue, it offers again (the same offer, without generating the message again, while neither side's heads moved). This is how a join without a diff --git a/src/document/changes.rs b/src/document/changes.rs index 345f151..0409c50 100644 --- a/src/document/changes.rs +++ b/src/document/changes.rs @@ -67,24 +67,20 @@ pub(crate) fn change_from_bytes(bytes: Vec, hash: ChangeHash) -> Result Option> { - if !missing(doc, to).is_empty() { + if !missing(doc, to).is_empty() || !missing(doc, from).is_empty() { return None; } - // Heads of `from` we don't have can't hide anything we have. - let from: Vec = from - .iter() - .copied() - .filter(|head| doc.get_change_by_hash(head).is_some()) - .collect(); let since: HashMap, Vec)> = doc - .get_changes(&from) + .get_changes(from) .iter() .map(|change| { ( @@ -181,7 +177,12 @@ mod tests { fn no_delta_without_its_heads() { let mut doc = Document::new(); change(&mut doc, |doc| doc.put(ROOT, "k", 1_i64).unwrap()); + let heads = doc.get_heads(); assert!(delta_bytes(&doc, &[], &[ChangeHash([7; 32])]).is_none()); + assert!( + delta_bytes(&doc, &[ChangeHash([7; 32])], &heads).is_none(), + "nor without where it starts" + ); } /// A change loads from its bytes only as the change it was asked for. diff --git a/src/document/sync.rs b/src/document/sync.rs index 8265a02..0aeec83 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -273,14 +273,18 @@ impl SyncSession { /// The message just generated is too large to send in one piece: take back what it /// carried, as if it had never been generated, and offer a delta instead. The /// message being handed over becomes the offer (it carries nothing); the answer is - /// still awaited, and if it doesn't come, the offer is made again. Returns the - /// peer's heads as far as we know them: where the delta starts. + /// still awaited, and if it doesn't come, the offer is made again. Returns where + /// the delta starts (`delta_from`). /// /// An answer is awaited even where none was (the peer's heads were ours: `generate`), /// so the offer lapses at the reply timeout at the latest: nothing else ends it but a /// message from the peer, and one that has everything has nothing to say. Without /// that, no message would go to it again. - pub(crate) fn withdraw_for_delta(&mut self, now: Instant) -> Vec { + pub(crate) fn withdraw_for_delta( + &mut self, + doc: &Document, + now: Instant, + ) -> Vec { if self.reply.is_none() { self.expect_reply(Vec::new(), now, SyncBytes::default(), 0); } @@ -296,7 +300,24 @@ impl SyncSession { self.state.in_flight = false; (self.state.last_sent_heads, self.state.have_responded) = self.before_generate.clone(); self.delta_offered = true; - self.state.their_heads.clone().unwrap_or_default() + self.delta_from(doc) + } + + /// Where a delta to the peer starts, sorted: its heads as far as we know them, of + /// those we hold. Every node that serves the delta must hold all of them + /// (`document::changes::delta_bytes`), so the same bytes come from each; we do. In + /// place of heads we lack (changes it has and we don't), the heads we share: the + /// delta may then carry some it has, but nothing it lacks. + fn delta_from(&self, doc: &Document) -> Vec { + let known = |hash: &&crate::document::ChangeHash| doc.get_change_by_hash(hash).is_some(); + let theirs = self.state.their_heads.as_deref().unwrap_or_default(); + let mut from: Vec<_> = theirs.iter().filter(known).copied().collect(); + if from.len() < theirs.len() { + from.extend(self.state.shared_heads.iter().filter(known).copied()); + } + from.sort_unstable(); + from.dedup(); + from } /// We offered the delta from `from` to `to` in place of a message of `bytes`. @@ -310,11 +331,12 @@ impl SyncSession { } /// The delta to offer again, if nothing changed since the last offer (`offered`): - /// the peer's heads as we know them, our `heads`, and the message too large for + /// where it starts (`delta_from`), our `heads`, and the message too large for /// `threshold` still. Then the message would be the same, and so the delta: its /// `from`, offered as by `withdraw_for_delta`, without generating the message. pub(crate) fn offer_again( &mut self, + doc: &Document, heads: &[crate::document::ChangeHash], threshold: usize, now: Instant, @@ -326,8 +348,7 @@ impl SyncSession { heads.dedup(); heads }; - let theirs = sorted(self.state.their_heads.as_deref().unwrap_or_default()); - if *bytes <= threshold || theirs != *from || sorted(heads) != *to { + if *bytes <= threshold || self.delta_from(doc) != *from || sorted(heads) != *to { return None; } let from = from.clone(); @@ -827,7 +848,7 @@ mod tests { a.sending = sending(); assert!(a.reply.is_none()); let now = Instant::now(); - a.withdraw_for_delta(now); + a.withdraw_for_delta(&Document::new(), now); assert!(a.delta_offered); assert!( a.take_overdue_reply(now + Duration::from_secs(3600)) @@ -849,24 +870,32 @@ mod tests { a.state.their_heads = Some(Vec::new()); let now = Instant::now(); assert_eq!( - a.offer_again(&heads, 1000, now), + a.offer_again(&ours, &heads, 1000, now), None, "nothing offered yet" ); a.offered(&[], &heads, 5000); - assert_eq!(a.offer_again(&heads, 1000, now), Some(Vec::new())); + assert_eq!(a.offer_again(&ours, &heads, 1000, now), Some(Vec::new())); assert!(a.delta_offered && a.reply.is_some() && a.sending.is_some()); - assert_eq!(a.offer_again(&heads, 5000, now), None, "it would fit now"); + assert_eq!( + a.offer_again(&ours, &heads, 5000, now), + None, + "it would fit now" + ); change(&mut ours, |doc| doc.put(ROOT, "k", 2).unwrap()); assert_eq!( - a.offer_again(&ours.get_heads(), 1000, now), + a.offer_again(&ours, &ours.get_heads(), 1000, now), None, "ours moved" ); a.state.their_heads = Some(heads.clone()); - assert_eq!(a.offer_again(&heads, 1000, now), None, "theirs moved"); + assert_eq!( + a.offer_again(&ours, &heads, 1000, now), + None, + "theirs moved" + ); } /// A message too large to send is taken back: the peer, after the offer and an @@ -900,7 +929,7 @@ mod tests { let carried = carrying.0.len(); assert!(carried > 4_000, "the changes took {carried} bytes"); - let from = a.withdraw_for_delta(now); + let from = a.withdraw_for_delta(&ours, now); assert!(from.is_empty()); assert!(a.delta_offered); assert!(!a.send_succeeded(a.id())); @@ -1157,6 +1186,30 @@ mod tests { } } + /// A delta starts at the peer's heads where we hold them all; where it has one we + /// lack, at those we hold and the heads we share, so every server of the delta + /// holds where it starts. + #[test] + fn a_delta_starts_where_we_hold_its_start() { + use crate::document::testing::change; + use automerge::{ROOT, transaction::Transactable}; + let mut doc = Document::new(); + change(&mut doc, |doc| doc.put(ROOT, "k", 1).unwrap()); + let shared = doc.get_heads(); + change(&mut doc, |doc| doc.put(ROOT, "k", 2).unwrap()); + let known = doc.get_heads(); + let unknown = crate::document::ChangeHash([7; 32]); + let mut a = session(); + a.state.shared_heads = shared.clone(); + + a.state.their_heads = Some(known.clone()); + assert_eq!(a.delta_from(&doc), known); + a.state.their_heads = Some(vec![known[0], unknown]); + let mut expected = vec![known[0], shared[0]]; + expected.sort_unstable(); + assert_eq!(a.delta_from(&doc), expected); + } + /// A delta offered from heads we since moved past, by changes from the offerer, /// would bring those again; our own, which it lacks, don't count. #[test] @@ -1228,7 +1281,7 @@ mod tests { b.send_succeeded(b.id()); a.receive(&mut ours, decode(&opening).unwrap()).unwrap(); assert!(a.generate(&ours, now, bytes).is_some()); - a.withdraw_for_delta(now); + a.withdraw_for_delta(&ours, now); a.send_succeeded(a.id()); // b loads the delta and tells a its heads, which are a's now. diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 3cbead1..2a08aae 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -162,6 +162,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { // (it didn't answer in time): offered again without generating the // message it replaces, which may take seconds when large. let again = peer_state.sync.offer_again( + &replica.doc, replica.heads(), delta_threshold, Instant::now(), @@ -186,7 +187,9 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { // Too large for one message: all or nothing, and on a slow link // within the transport's delivery limit. The peer fetches it in // chunks. - let from = peer_state.sync.withdraw_for_delta(Instant::now()); + let from = peer_state + .sync + .withdraw_for_delta(&replica.doc, Instant::now()); let spec = transport::DeltaSpec::new(from, replica.heads().to_vec()); if spec.to.iter().all(|head| spec.from.contains(head)) { // It has our heads: nothing to offer (Automerge thought it diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 4ddfe32..0f009cb 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -690,8 +690,8 @@ pub(crate) enum SyncBody { } /// The changes in the history of `to` but not of `from`: what a peer at `from` lacks of -/// a document at `to`. Every node holding `to` produces the same bytes for it -/// (`document::changes::delta_bytes`). +/// a document at `to`. Every node holding `to` and `from` produces the same bytes for +/// it; others serve none (`document::changes::delta_bytes`). #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub(crate) struct DeltaSpec { /// Sorted. -- cgit v1.3.1