From 26d41bef0de3a0d8f25b62776e966ed464a34063 Mon Sep 17 00:00:00 2001 From: tv Date: Thu, 5 Nov 2026 00:19:12 +0100 Subject: offer again ungenerated --- src/swarm/sync.rs | 85 ++++++++++++++++++++++++++++++++++--------------------- 1 file changed, 53 insertions(+), 32 deletions(-) (limited to 'src/swarm') 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