manifests asked at once

1 files changed, 57 insertions, 45 deletions

diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index f06b251..6efa5ca 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -15,8 +15,9 @@
 //!   are kept, so serving a download doesn't extract them for every chunk. Serving runs
 //!   in tasks of its own, at most [`MAX_ANSWERING`] requests and [`MAX_SERVING`] chunk
 //!   sends at a time.
-//! * Fetching: the manifest from the first peer that has it (closest first, and only
-//!   from a peer asked; of content, only with the size its reference names), then the
+//! * Fetching: the manifest from the first peer that has it (closest first, asked of
+//!   [`MANIFESTS_AT_ONCE`] at a time, and only from a peer asked; of content, only with
+//!   the size its reference names), then the
 //!   chunks, a few in flight per peer at first, more as they come (up to
 //!   [`MAX_WINDOW`]), from up to [`MAX_SOURCES`] peers. Every chunk is
 //!   checked against the manifest; a peer that sends a bad one is dropped, a chunk
@@ -64,6 +65,8 @@ const MAX_ANSWERING: usize = 32;
 const DROPPED_LOG_INTERVAL: Duration = Duration::from_secs(60);
 /// How long to wait for a manifest from one peer.
 const MANIFEST_TIMEOUT: Duration = Duration::from_secs(15);
+/// Peers asked for the manifest at once.
+const MANIFESTS_AT_ONCE: usize = 3;
 /// How long to wait for a chunk before asking another peer (raised by the peer's
 /// round-trip time).
 const CHUNK_TIMEOUT: Duration = Duration::from_secs(60);
@@ -444,54 +447,63 @@ pub(crate) async fn fetch(
         bail!("no peer to download {id} from");
     }
 
-    // The manifest, from the first peer that has the blob.
+    // The manifest, from the first peer that has the blob: asked of the closest few at
+    // once (`MANIFESTS_AT_ONCE`), and of the next as one answers it hasn't the blob or
+    // doesn't answer in time, rather than of one at a time, each waited for.
     let mut manifest = None;
     let mut missing: HashSet<String> = HashSet::new();
-    for peer in &candidates {
-        request(state, peer, id, Vec::new(), delta).await;
-        let deadline = Instant::now() + MANIFEST_TIMEOUT;
-        loop {
-            let answer = tokio::time::timeout_at(deadline, answers.recv()).await;
-            match answer {
-                Ok(Some((
-                    from,
-                    BlobBody::Manifest {
-                        size,
-                        chunk_size,
-                        hashes,
-                    },
-                ))) if candidates.contains(&from) => {
-                    let size_fits = match expected_size {
-                        Some(expected) => size == expected,
-                        None => size <= MAX_SIZE,
-                    };
-                    if !size_fits
-                        || !(MIN_CHUNK..=MAX_CHUNK).contains(&chunk_size)
-                        || hashes.len() as u64 != chunk_count(size, chunk_size)
-                    {
-                        warn!(peer = %from, blob = %id, size, chunk_size, "Unusable blob manifest; asking another peer");
-                        missing.insert(from);
-                        break;
-                    }
-                    manifest = Some(Manifest {
-                        size,
-                        chunk_size,
-                        hashes,
-                    });
-                    break;
-                }
-                Ok(Some((from, BlobBody::Missing))) => {
+    let mut asked: HashMap<String, Instant> = HashMap::new();
+    let mut next = candidates.iter().cloned().collect::<VecDeque<_>>();
+    loop {
+        while asked.len() < MANIFESTS_AT_ONCE
+            && let Some(peer) = next.pop_front()
+        {
+            request(state, &peer, id, Vec::new(), delta).await;
+            asked.insert(peer, Instant::now() + MANIFEST_TIMEOUT);
+        }
+        let Some(deadline) = asked.values().min().copied() else {
+            break;
+        };
+        match tokio::time::timeout_at(deadline, answers.recv()).await {
+            Ok(Some((
+                from,
+                BlobBody::Manifest {
+                    size,
+                    chunk_size,
+                    hashes,
+                },
+            ))) if candidates.contains(&from) => {
+                let size_fits = match expected_size {
+                    Some(expected) => size == expected,
+                    None => size <= MAX_SIZE,
+                };
+                if !size_fits
+                    || !(MIN_CHUNK..=MAX_CHUNK).contains(&chunk_size)
+                    || hashes.len() as u64 != chunk_count(size, chunk_size)
+                {
+                    warn!(peer = %from, blob = %id, size, chunk_size, "Unusable blob manifest; asking another peer");
                     missing.insert(from.clone());
-                    if &from == peer {
-                        break;
-                    }
+                    asked.remove(&from);
+                    continue;
                 }
-                Ok(Some(_)) => {} // chunks of an earlier attempt
-                Ok(None) | Err(_) => break,
+                manifest = Some(Manifest {
+                    size,
+                    chunk_size,
+                    hashes,
+                });
+                break;
+            }
+            Ok(Some((from, BlobBody::Missing))) => {
+                missing.insert(from.clone());
+                asked.remove(&from);
+            }
+            Ok(Some(_)) => {} // chunks of an earlier attempt
+            Ok(None) => break,
+            // Those asked too long ago make way for the next.
+            Err(_) => {
+                let now = Instant::now();
+                asked.retain(|_, deadline| now < *deadline);
             }
-        }
-        if manifest.is_some() {
-            break;
         }
     }
     let Some(manifest) = manifest else {