From 26d41bef0de3a0d8f25b62776e966ed464a34063 Mon Sep 17 00:00:00 2001 From: tv Date: Thu, 5 Nov 2026 00:19:12 +0100 Subject: offer again ungenerated --- docs/peers.md | 3 +- src/document/sync.rs | 86 ++++++++++++++++++++++++++++++++++++++++++++++++++++ src/swarm/sync.rs | 85 ++++++++++++++++++++++++++++++++------------------- 3 files changed, 141 insertions(+), 33 deletions(-) diff --git a/docs/peers.md b/docs/peers.md index fc4b6f1..eb1eb5c 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -165,7 +165,8 @@ may not arrive within the transport's delivery limit (Mycelium's five minutes). 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, 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. This is how a join without a +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 recent remake, or a return after much was written, gets the bulk of its changes. A node downloads one delta at a time: while a large one comes, it lacks those changes, so its other neighbours offer them too, and they would come once per offer. Offers that come diff --git a/src/document/sync.rs b/src/document/sync.rs index e93fa17..8265a02 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -103,6 +103,13 @@ pub(crate) struct SyncSession { /// (`withdraw_for_delta`): no more messages until it answers, or the answer is /// overdue (then the offer goes out again). pub(crate) delta_offered: bool, + /// The delta last offered: its heads (sorted, as `DeltaSpec`), and the size of the + /// message it replaced (`offer_again`). + last_offer: Option<( + Vec, + Vec, + usize, + )>, /// What the sync state knew of the peer before the last message from it /// (`answered_as_delivered`). before_receive: Option, @@ -146,6 +153,7 @@ impl SyncSession { retry_at: None, send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX), delta_offered: false, + last_offer: None, before_receive: None, before_generate: (Vec::new(), false), reply: None, @@ -291,6 +299,51 @@ impl SyncSession { self.state.their_heads.clone().unwrap_or_default() } + /// We offered the delta from `from` to `to` in place of a message of `bytes`. + pub(crate) fn offered( + &mut self, + from: &[crate::document::ChangeHash], + to: &[crate::document::ChangeHash], + bytes: usize, + ) { + self.last_offer = Some((from.to_vec(), to.to_vec(), bytes)); + } + + /// 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 + /// `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, + heads: &[crate::document::ChangeHash], + threshold: usize, + now: Instant, + ) -> Option> { + let (from, to, bytes) = self.last_offer.as_ref()?; + let sorted = |heads: &[crate::document::ChangeHash]| { + let mut heads = heads.to_vec(); + heads.sort_unstable(); + heads.dedup(); + heads + }; + let theirs = sorted(self.state.their_heads.as_deref().unwrap_or_default()); + if *bytes <= threshold || theirs != *from || sorted(heads) != *to { + return None; + } + let from = from.clone(); + if self.reply.is_none() { + self.expect_reply(Vec::new(), now, SyncBytes::default(), 0); + } + // Handed over like a generated message (`send_failed` then retries). + self.sending = Some(Sending { + sent: Vec::new(), + regenerate: false, + assumed: None, + }); + self.delta_offered = true; + Some(from) + } + /// The message just generated, withdrawn for a delta (`withdraw_for_delta`), turns /// out to carry nothing the peer lacks (it has our heads): none goes, and nothing is /// offered or awaited. @@ -783,6 +836,39 @@ mod tests { assert!(!a.delta_offered); } + /// An offer that lapsed goes again as it was while nothing changed; otherwise, or + /// once the message would fit, it is generated afresh. + #[test] + fn the_same_offer_goes_again_ungenerated() { + use crate::document::testing::change; + use automerge::{ROOT, transaction::Transactable}; + let mut ours = Document::new(); + change(&mut ours, |doc| doc.put(ROOT, "k", 1).unwrap()); + let heads = ours.get_heads(); + let mut a = session(); + a.state.their_heads = Some(Vec::new()); + let now = Instant::now(); + assert_eq!( + a.offer_again(&heads, 1000, now), + None, + "nothing offered yet" + ); + + a.offered(&[], &heads, 5000); + assert_eq!(a.offer_again(&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"); + + change(&mut ours, |doc| doc.put(ROOT, "k", 2).unwrap()); + assert_eq!( + a.offer_again(&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"); + } + /// A message too large to send is taken back: the peer, after the offer and an /// overdue answer, gets the same changes in the next message. #[test] diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index aa6a8f9..3cbead1 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -158,43 +158,64 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { && peer_state.sync.have_all_theirs(&replica.doc); // Under the same locks as the epoch label: the message belongs to // this document. - let message = - peer_state - .sync - .generate(&replica.doc, Instant::now(), bytes_before); - let body = message.and_then(|payload| { - if payload.0.len() <= delta_threshold { - return Some(transport::SyncBody::Message(payload)); - } - // 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()); + // The same delta as last time, the peer's heads and ours unchanged + // (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.heads(), + delta_threshold, + Instant::now(), + ); + let body = if let Some(from) = again { 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 - // lacked what it has, as from a view of it that was old). - debug!( - message_bytes = payload.0.len(), - "Sync message too large, but the peer has our heads; not sent" - ); - peer_state.sync.nothing_to_send(); - return None; - } - // And why it is too large here: the threshold, lowered by the - // throughput measured to the peer or a message that failed whole - // to it lately (`Blobs::delta_threshold_for`). - info!( - message_bytes = payload.0.len(), - threshold = delta_threshold, - throughput = ?rate.map(|rate| rate as u64), - failed_whole = ?failed, + debug!( from = spec.from.len(), to = spec.to.len(), - "Sync message too large; offering the changes as a blob" + "Offering the same changes as a blob again" ); Some(transport::SyncBody::Delta(spec)) - }); + } else { + let message = + peer_state + .sync + .generate(&replica.doc, Instant::now(), bytes_before); + message.and_then(|payload| { + if payload.0.len() <= delta_threshold { + return Some(transport::SyncBody::Message(payload)); + } + // 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 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 + // lacked what it has, as from a view of it that was old). + debug!( + message_bytes = payload.0.len(), + "Sync message too large, but the peer has our heads; not sent" + ); + peer_state.sync.nothing_to_send(); + return None; + } + // And why it is too large here: the threshold, lowered by the + // throughput measured to the peer or a message that failed whole + // to it lately (`Blobs::delta_threshold_for`). + info!( + message_bytes = payload.0.len(), + threshold = delta_threshold, + throughput = ?rate.map(|rate| rate as u64), + failed_whole = ?failed, + from = spec.from.len(), + to = spec.to.len(), + "Sync message too large; offering the changes as a blob" + ); + peer_state + .sync + .offered(&spec.from, &spec.to, payload.0.len()); + Some(transport::SyncBody::Delta(spec)) + }) + }; let delivered = level && matches!(body, Some(transport::SyncBody::Message(_))); if delivered { peer_state.sync.assume_delivered(replica.heads()); -- cgit v1.3.1