one delta at a time

1 files changed, 38 insertions, 7 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();