1 files changed, 74 insertions, 6 deletions

diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index 6efa5ca..c2b397c 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -21,7 +21,8 @@
 //!   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
-//!   that doesn't arrive in time is asked of another peer. The caller checks the whole
+//!   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
 //!   (for a change: its hash).
 //!
 //! The chunk size is the serving node's (`--blob-chunk`); the manifest carries it.
@@ -68,8 +69,10 @@ 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).
+/// round-trip time), until chunks from it show how long they take: then three times as
+/// long as they took lately (`Source::timeout_now`), but at least `CHUNK_TIMEOUT_MIN`.
 const CHUNK_TIMEOUT: Duration = Duration::from_secs(60);
+const CHUNK_TIMEOUT_MIN: Duration = Duration::from_secs(5);
 /// A download gives up after this long without any chunk arriving.
 const STALL_TIMEOUT: Duration = Duration::from_secs(180);
 /// Bad chunks after which a peer is no longer asked.
@@ -410,7 +413,34 @@ struct Source {
     in_flight: HashMap<u32, Instant>,
     window: usize,
     bad: u32,
+    /// The longest a chunk is waited for (`CHUNK_TIMEOUT`, raised by the round trip).
     timeout: Duration,
+    rtt: Duration,
+    /// How long its chunks took lately, from the request (smoothed), once one came.
+    took: Option<Duration>,
+}
+
+impl Source {
+    /// How long a chunk from it is waited for now: three times as long as its chunks
+    /// took lately (they wait in its queue too, as long as the window is), within
+    /// `CHUNK_TIMEOUT_MIN` (and two round trips) and `timeout`. So a source that stops
+    /// sending hands its chunks on within seconds where they came quickly, and a slow
+    /// link isn't given up on for being slow. Only where another peer could take them
+    /// (`fetch`).
+    fn timeout_now(&self) -> Duration {
+        match self.took {
+            None => self.timeout,
+            Some(took) => (took * 3).clamp(CHUNK_TIMEOUT_MIN + self.rtt * 2, self.timeout),
+        }
+    }
+
+    /// A chunk asked for `took` ago came.
+    fn came(&mut self, took: Duration) {
+        self.took = Some(match self.took {
+            None => took,
+            Some(before) => (before * 7 + took) / 8,
+        });
+    }
 }
 
 /// Downloads blob `id` from `peers` (all thought to hold it): of `expected_size` bytes
@@ -525,14 +555,15 @@ pub(crate) async fn fetch(
             let Some(peer) = candidates.pop_front() else {
                 break;
             };
-            let timeout =
-                CHUNK_TIMEOUT + rtts.get(&peer).copied().flatten().unwrap_or_default() * 4;
+            let rtt = rtts.get(&peer).copied().flatten().unwrap_or_default();
             sources.push(Source {
                 peer,
                 in_flight: HashMap::new(),
                 window: WINDOW,
                 bad: 0,
-                timeout,
+                timeout: CHUNK_TIMEOUT + rtt * 4,
+                rtt,
+                took: None,
             });
         }
         if sources.is_empty() {
@@ -588,6 +619,7 @@ pub(crate) async fn fetch(
                 }
                 if let Some(sent_at) = sent_at {
                     debug!(peer = %from, blob = %id, index, ms = sent_at.elapsed().as_millis() as u64, "Blob chunk");
+                    source.came(sent_at.elapsed());
                 }
                 chunks[slot] = Some(data);
                 done += 1;
@@ -606,8 +638,15 @@ pub(crate) async fn fetch(
         // Chunks that took too long go to another peer; the slow peer gets a smaller
         // window. Peers that sent bad chunks are dropped.
         let now = Instant::now();
+        // The quick timeout only where another peer could take the chunks: with one
+        // source, asking it again would only bring them twice over a slow link.
+        let others = sources.len() > 1 || !candidates.is_empty();
         for source in &mut sources {
-            let timeout = source.timeout;
+            let timeout = if others {
+                source.timeout_now()
+            } else {
+                source.timeout
+            };
             let before = source.in_flight.len();
             source
                 .in_flight
@@ -690,6 +729,35 @@ pub(crate) async fn fetch_delta(
 mod tests {
     use super::*;
 
+    /// A source's chunk timeout follows how long its chunks took, within bounds: the
+    /// long one until a chunk came.
+    #[test]
+    fn chunk_timeouts_follow_the_source() {
+        let mut source = Source {
+            peer: "p".into(),
+            in_flight: HashMap::new(),
+            window: WINDOW,
+            bad: 0,
+            timeout: CHUNK_TIMEOUT,
+            rtt: Duration::from_millis(100),
+            took: None,
+        };
+        assert_eq!(source.timeout_now(), CHUNK_TIMEOUT);
+        source.came(Duration::from_secs(4));
+        assert_eq!(source.timeout_now(), Duration::from_secs(12));
+        source.came(Duration::from_millis(100));
+        assert!(source.timeout_now() < Duration::from_secs(12));
+        for _ in 0..50 {
+            source.came(Duration::from_millis(100));
+        }
+        assert_eq!(
+            source.timeout_now(),
+            CHUNK_TIMEOUT_MIN + Duration::from_millis(200)
+        );
+        source.came(Duration::from_secs(600));
+        assert_eq!(source.timeout_now(), CHUNK_TIMEOUT, "at most the long one");
+    }
+
     /// One delta at a time: offers meanwhile wait, and are named when it ends.
     #[test]
     fn one_delta_at_a_time() {