unreached sources hand over

1 files changed, 26 insertions, 1 deletions

diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index c2b397c..454687f 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -22,7 +22,8 @@
 //!   [`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
 //!   that doesn't arrive in time (three times as long as its chunks took lately, where
-//!   another peer could send it) is asked of another peer. The caller checks the whole
+//!   another peer could send it) is asked of another peer, as are all of a peer's once
+//!   we no longer reach it (`swarm::reach`). The caller checks the whole
 //!   (for a change: its hash).
 //!
 //! The chunk size is the serving node's (`--blob-chunk`); the manifest carries it.
@@ -378,6 +379,13 @@ impl Drop for Registration<'_> {
     }
 }
 
+/// Whether we no longer reach `peer_state`'s peer, as `swarm::reach` judges it.
+fn unreached(peer_state: &crate::state::PeerState, now: Instant) -> bool {
+    peer_state.liveness.is_dormant()
+        || peer_state.is_unreachable()
+        || peer_state.answer_overdue(now)
+}
+
 async fn request(
     state: &AppState,
     peer: &str,
@@ -663,6 +671,23 @@ pub(crate) async fn fetch(
             }
             keep
         });
+        // A source we no longer reach (dormant, unreachable, or what we sent it went
+        // unanswered: `swarm::reach`) hands its chunks to the others at once, rather than
+        // after its chunks' timeout; but only where there are others.
+        if sources.len() > 1 || !candidates.is_empty() {
+            let mut gone = Vec::new();
+            for source in &sources {
+                if let Some(peer_state) = state.get_peer_state(&source.peer).await
+                    && unreached(&*peer_state.lock().await, now)
+                {
+                    gone.push(source.peer.clone());
+                }
+            }
+            if !gone.is_empty() {
+                debug!(blob = %id, peers = ?gone, "Blob sources no longer reached; asking others");
+                sources.retain(|source| !gone.contains(&source.peer));
+            }
+        }
         if last_progress.elapsed() > STALL_TIMEOUT {
             bail!(
                 "download of {id} stalled ({done} of {total} chunks, nothing for {}s)",