From 5ace906873fc5d1757a86b016b5e731ab5decf1a Mon Sep 17 00:00:00 2001 From: tv Date: Thu, 5 Nov 2026 00:10:24 +0100 Subject: stale delta offers --- docs/peers.md | 3 +++ src/document/sync.rs | 60 ++++++++++++++++++++++++++++++++++++++++++++++++++++ src/swarm/sync.rs | 19 +++++++++++++++++ 3 files changed, 82 insertions(+) diff --git a/docs/peers.md b/docs/peers.md index 4775e85..fc4b6f1 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -172,6 +172,9 @@ other neighbours offer them too, and they would come once per offer. Offers that meanwhile wait; once the delta is in, the node syncs with those that offered, which then send only what is still missing. What a delta brought counts as known to its sender, as what its sync messages bring does: none of it goes back, by message or as a delta. +An offer made before its sender heard what the node has now (one that came behind the +chunks of the delta before it, say) would bring part of that again: such an offer isn't +fetched; the node's next message tells the sender, which offers afresh if still needed. Towards a peer on a slow link the limit is lower: a message is offered as a delta if it would take more than 5 s at the throughput measured to that peer (`syncThroughput`, diff --git a/src/document/sync.rs b/src/document/sync.rs index 2921bb9..e93fa17 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -366,6 +366,40 @@ impl SyncSession { } } + /// Of the changes in `doc` beyond `from`, how many the peer has too, as far as we + /// know (its heads' history, and what went either way since, `sent_hashes`). A delta + /// it offers from `from` brings those again: offered before it heard what we have + /// now (while we downloaded its last one, say, and it waited for our answer). + pub(crate) fn known_beyond( + &self, + doc: &Document, + from: &[crate::document::ChangeHash], + ) -> usize { + let from: Vec = from + .iter() + .copied() + .filter(|hash| doc.get_change_by_hash(hash).is_some()) + .collect(); + let beyond: std::collections::HashMap<_, _> = doc + .get_changes_meta(&from) + .into_iter() + .map(|change| (change.hash, change.deps)) + .collect(); + // A change's predecessors outside `beyond` are in the history of `from`. + let mut known = std::collections::HashSet::new(); + let mut pending: Vec = + self.state.their_heads.iter().flatten().copied().collect(); + pending.extend(self.state.sent_hashes.iter().copied()); + while let Some(hash) = pending.pop() { + if let Some(deps) = beyond.get(&hash) + && known.insert(hash) + { + pending.extend(deps.iter().copied()); + } + } + known.len() + } + /// What the sync state knows of the peer (`PeerView`). fn peer_view(&self) -> PeerView { PeerView { @@ -1037,6 +1071,32 @@ mod tests { } } + /// 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] + fn offers_from_before_count_what_came_since() { + use crate::document::changes::{delta_bytes, load_delta}; + 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 (_a, mut b) = in_step(&mut ours, &mut theirs); + + let before = theirs.get_heads(); + change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap()); + let delta = delta_bytes(&ours, &before, &ours.get_heads()).unwrap(); + load_delta(&mut theirs, &delta).unwrap(); + b.came_from_peer(&theirs, &before); + change(&mut theirs, |doc| doc.put(ROOT, "own", 3).unwrap()); + + assert_eq!( + b.known_beyond(&theirs, &before), + 1, + "the delta's change again" + ); + assert_eq!(b.known_beyond(&theirs, &theirs.get_heads()), 0); + } + /// Our transport refused a message taken as delivered: the peer's heads are as /// before, so the next message carries the change again. #[test] diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 8059fe2..aa6a8f9 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -620,6 +620,25 @@ async fn download_and_apply_delta( .into_iter() .filter(|other| other != peer), ); + if let Some(peer_state) = state.get_peer_state(peer).await { + let doc = Arc::clone(&state.doc); + let from = spec.from.clone(); + let known = tokio::task::spawn_blocking(move || { + let replica = doc.blocking_read(); + peer_state + .blocking_lock() + .sync + .known_beyond(&replica.doc, &from) + }) + .await + .context("Delta check task panicked")?; + if known > 0 { + // Offered before it heard what we have: it would bring those again. Our + // next message tells it (`fetch_delta`); it offers afresh if still needed. + info!(peer = %peer, known, "Offered changes partly here already; not fetched"); + return Ok(false); + } + } info!(peer = %peer, to = spec.to.len(), "Fetching changes offered as a blob"); let started = Instant::now(); let bytes = crate::swarm::blob::fetch_delta(state, &spec, peers).await?; -- cgit v1.3.1