one delta at a time

2 files changed, 53 insertions, 9 deletions

diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index 98dee66..694fff0 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -78,8 +78,9 @@ pub(crate) struct Blobs {
     cache: Mutex<Option<(BlobId, Arc<Prepared>)>>,
     /// Answers for downloads in progress, by blob.
     fetches: Mutex<HashMap<BlobId, mpsc::UnboundedSender<(String, BlobBody)>>>,
-    /// Peers whose offered delta is being downloaded (`swarm::sync`).
-    deltas: Mutex<HashSet<String>>,
+    /// The peer whose offered delta is being downloaded (`swarm::sync`), and the peers
+    /// whose offers came meanwhile (`start_delta`).
+    deltas: Mutex<(Option<String>, HashSet<String>)>,
 }
 
 /// A sync message larger than this many chunks is sent as a delta blob instead
@@ -110,7 +111,7 @@ impl Blobs {
             dropped: Mutex::new((0, None)),
             cache: Mutex::new(None),
             fetches: Mutex::new(HashMap::new()),
-            deltas: Mutex::new(HashSet::new()),
+            deltas: Mutex::new((None, HashSet::new())),
         }
     }
 
@@ -138,13 +139,29 @@ impl Blobs {
         threshold.max(chunk)
     }
 
-    /// Starts downloading a delta offered by `peer`; false if one from it is running.
+    /// Starts downloading a delta offered by `peer`; false if one is downloading (from it
+    /// or another), and `peer`'s offer waits for the end (`end_delta`). One at a time:
+    /// while a large one comes slowly, the changes in it are what other neighbours
+    /// announce and offer too (it lacks them), and they would come once per offer.
     pub(crate) fn start_delta(&self, peer: &str) -> bool {
-        self.deltas.lock_or_recover().insert(peer.to_owned())
+        let mut deltas = self.deltas.lock_or_recover();
+        if deltas.0.is_some() {
+            deltas.1.insert(peer.to_owned());
+            return false;
+        }
+        deltas.0 = Some(peer.to_owned());
+        true
     }
 
-    pub(crate) fn end_delta(&self, peer: &str) {
-        self.deltas.lock_or_recover().remove(peer);
+    /// The delta from `peer` is in (or failed): the peers whose offers waited, to sync
+    /// with now, which then send what is still missing (`peer` among them if it offered
+    /// again meanwhile).
+    pub(crate) fn end_delta(&self, peer: &str) -> Vec<String> {
+        let mut deltas = self.deltas.lock_or_recover();
+        if deltas.0.as_deref() == Some(peer) {
+            deltas.0 = None;
+        }
+        deltas.1.drain().collect()
     }
 
     fn fetches(
@@ -627,6 +644,20 @@ pub(crate) async fn fetch_delta(
 mod tests {
     use super::*;
 
+    /// One delta at a time: offers meanwhile wait, and are named when it ends.
+    #[test]
+    fn one_delta_at_a_time() {
+        let blobs = Blobs::new(64 * 1024);
+        assert!(blobs.start_delta("b"));
+        assert!(!blobs.start_delta("c"), "c's offer waits");
+        assert!(!blobs.start_delta("b"), "b's again too");
+        let mut waited = blobs.end_delta("b");
+        waited.sort();
+        assert_eq!(waited, ["b", "c"]);
+        assert!(blobs.start_delta("c"), "the next may start");
+        assert!(blobs.end_delta("c").is_empty());
+    }
+
     #[test]
     fn chunking() {
         let bytes: Vec<u8> = (0..1000u32).map(|n| n as u8).collect();
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;