one delta at a time

1 files changed, 15 insertions, 2 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 94153ff..24b867b 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -575,14 +575,20 @@ pub(crate) async fn handle_sync(
 /// 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");
+        debug!(peer = %peer, "Delta offered while another downloads; after it");
         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);
+        // Those that offered meanwhile send what is still missing: our next message
+        // says what we have now.
+        for other in state.blobs.end_delta(&peer) {
+            if other != peer {
+                state.request_sync(&other);
+            }
+        }
         match applied {
             Ok(changed) => {
                 state.notify_store_changed();
@@ -615,7 +621,14 @@ async fn download_and_apply_delta(
             .filter(|other| other != peer),
     );
     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?;
+    info!(
+        peer = %peer,
+        bytes = bytes.len(),
+        took_ms = started.elapsed().as_millis() as u64,
+        "Fetched changes offered as a blob"
+    );
 
     let doc = Arc::clone(&state.doc);
     let peer_state = state.get_peer_state(peer).await;