1 files changed, 116 insertions, 5 deletions
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index ca54fae..70ce28e 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -124,6 +124,11 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
             // The retry timer enqueues the peer once it's due.
             return Ok(());
         }
+        if peer_state.sync.delta_offered {
+            // Its answer, or the reply timeout, brings us back here.
+            trace!("A delta is offered; waiting for the peer");
+            return Ok(());
+        }
     }
 
     let (session, encoded) = {
@@ -134,13 +139,14 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
         let ready = state.is_ready().await;
         // Before this message is counted.
         let bytes_before = SyncBytes::of(&state.traffic, peer);
+        let delta_threshold = state.blobs.delta_threshold();
 
         // The session the message comes from, for its send's outcome.
         let (session, generated) = tokio::task::spawn_blocking(
             move || -> (u64, Result<Option<transport::EncodedMessage>>) {
                 // Label the message with the epoch of the document it was generated
                 // from, read under the same lock.
-                let (session, epoch, epoch_seq, message) = {
+                let (session, epoch, epoch_seq, body) = {
                     let mut replica = doc.blocking_write();
                     let mut peer_state = psm.blocking_lock();
                     let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
@@ -150,16 +156,33 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                         peer_state
                             .sync
                             .generate(&mut replica.doc, Instant::now(), bytes_before);
-                    (peer_state.sync.id(), epoch, epoch_seq, message)
+                    let body = message.map(|message| {
+                        let payload = transport::SyncPayload::encode(message);
+                        if payload.0.len() <= delta_threshold {
+                            return transport::SyncBody::Message(payload);
+                        }
+                        // Too large for one message: all or nothing, and on a slow link
+                        // within Mycelium's deadline. The peer fetches it in chunks.
+                        let from = peer_state.sync.withdraw_for_delta();
+                        let spec = transport::DeltaSpec::new(from, replica.heads().to_vec());
+                        info!(
+                            message_bytes = payload.0.len(),
+                            from = spec.from.len(),
+                            to = spec.to.len(),
+                            "Sync message too large; offering the changes as a blob"
+                        );
+                        transport::SyncBody::Delta(spec)
+                    });
+                    (peer_state.sync.id(), epoch, epoch_seq, body)
                 };
-                let Some(message) = message else {
+                let Some(body) = body else {
                     return (session, Ok(None));
                 };
                 let envelope = transport::SyncEnvelope {
                     swarm_id,
                     epoch,
                     epoch_seq,
-                    message: transport::SyncPayload::encode(message),
+                    body,
                     ready,
                 };
                 let encoded = transport
@@ -316,6 +339,14 @@ pub(crate) async fn handle_sync(
 
     let epoch_label = envelope.epoch;
     let from_ready = envelope.ready;
+    let payload = match envelope.body {
+        transport::SyncBody::Message(payload) => payload,
+        transport::SyncBody::Delta(spec) => {
+            fetch_delta(state, peer, envelope.epoch, spec);
+            return Ok(());
+        }
+    };
+    state.stats.sync_received(payload.0.len());
 
     let applied = {
         let doc = Arc::clone(&state.doc);
@@ -329,7 +360,7 @@ pub(crate) async fn handle_sync(
         // While catching up, everything arrives late; that says nothing about propagation.
         let measure_propagation = !check_caught_up;
         let epoch = envelope.epoch;
-        let message = envelope.message;
+        let message = payload;
         let peer = peer.to_owned();
         let clock = Arc::clone(&state.clock);
 
@@ -427,6 +458,86 @@ pub(crate) async fn handle_sync(
     Ok(())
 }
 
+/// The peer offered the changes we lack as a blob (`SyncBody::Delta`): fetch them, from
+/// it and any other peer that has them, apply them, and sync on. One download per peer
+/// at a time: the offer is repeated while it runs.
+fn fetch_delta(state: &AppState, peer: &str, epoch: transport::Epoch, spec: transport::DeltaSpec) {
+    if !state.blobs.start_delta(peer) {
+        trace!(peer = %peer, "Delta offered again while it downloads");
+        return;
+    }
+    let state = state.clone();
+    let peer = peer.to_owned();
+    tokio::spawn(async move {
+        let applied = download_and_apply_delta(&state, &peer, epoch, spec).await;
+        state.blobs.end_delta(&peer);
+        match applied {
+            Ok(changed) => {
+                state.notify_store_changed();
+                if changed {
+                    crate::swarm::relay::announce(&state, &peer).await;
+                }
+            }
+            Err(error) => {
+                warn!(peer = %peer, error = %format!("{error:#}"), "Delta download failed; syncing on");
+            }
+        }
+        // Our next message says what we have now; the peer answers with the rest.
+        let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(peer));
+    });
+}
+
+/// Returns whether our heads changed.
+async fn download_and_apply_delta(
+    state: &AppState,
+    peer: &str,
+    epoch: transport::Epoch,
+    spec: transport::DeltaSpec,
+) -> Result<bool> {
+    let mut peers = vec![peer.to_owned()];
+    peers.extend(
+        state
+            .active_peers_on_local_epoch()
+            .await
+            .into_iter()
+            .filter(|other| other != peer),
+    );
+    info!(peer = %peer, to = spec.to.len(), "Fetching changes offered as a blob");
+    let bytes = crate::swarm::blob::fetch_delta(state, &spec, peers).await?;
+
+    let doc = Arc::clone(&state.doc);
+    let peer_state = state.get_peer_state(peer).await;
+    let stats = Arc::clone(&state.stats);
+    tokio::task::spawn_blocking(move || -> Result<bool> {
+        let mut replica = doc.blocking_write();
+        if replica.epoch != epoch {
+            bail!("the epoch changed during the download");
+        }
+        let heads_before = replica.heads().to_vec();
+        replica
+            .doc
+            .load_incremental(&bytes)
+            .context("Loading the downloaded changes")?;
+        replica.refresh_heads();
+        let missing = spec
+            .to
+            .iter()
+            .filter(|head| replica.doc.get_change_by_hash(head).is_none())
+            .count();
+        if missing > 0 {
+            bail!("{missing} of the offered heads are still missing after loading");
+        }
+        stats.heads_seen(replica.heads().len());
+        if let Some(peer_state) = peer_state {
+            // Sync from it made progress: no heal for a while (`keepalive`).
+            peer_state.blocking_lock().last_sync_rx = Some(Instant::now());
+        }
+        Ok(replica.heads() != heads_before.as_slice())
+    })
+    .await
+    .context("Delta apply task panicked")?
+}
+
 pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedReceiver<SyncEvent>) {
     let mut dirty_peers = HashSet::<String>::new();
     let mut document_dirty = false;