offer again ungenerated

1 files changed, 53 insertions, 32 deletions

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());