first chunk by the fastest

1 files changed, 33 insertions, 14 deletions

diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index 8c47c98..82586b8 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -72,8 +72,9 @@ 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), 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`.
+/// round-trip time), until chunks from it or another source 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.
@@ -438,10 +439,12 @@ impl Source {
     /// 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 {
+    /// link isn't given up on for being slow. Before its first chunk came, by `fastest`,
+    /// how long the fastest other source's took (chunks are alike in size): a source
+    /// that never sends doesn't hold its window for `timeout`. Only where another peer
+    /// could take them (`fetch`).
+    fn timeout_now(&self, fastest: Option<Duration>) -> Duration {
+        match self.took.or(fastest) {
             None => self.timeout,
             Some(took) => (took * 3).clamp(CHUNK_TIMEOUT_MIN + self.rtt * 2, self.timeout),
         }
@@ -709,9 +712,10 @@ pub(crate) async fn fetch(
         // 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();
+        let fastest = sources.iter().filter_map(|source| source.took).min();
         for source in &mut sources {
             let timeout = if others {
-                source.timeout_now()
+                source.timeout_now(fastest)
             } else {
                 source.timeout
             };
@@ -814,8 +818,9 @@ 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.
+    /// A source's chunk timeout follows how long its chunks took, within bounds: before
+    /// its first, how long the fastest other source's took; the long one until a chunk
+    /// came from any.
     #[test]
     fn chunk_timeouts_follow_the_source() {
         let mut source = Source {
@@ -827,20 +832,34 @@ mod tests {
             rtt: Duration::from_millis(100),
             took: None,
         };
-        assert_eq!(source.timeout_now(), CHUNK_TIMEOUT);
+        assert_eq!(source.timeout_now(None), CHUNK_TIMEOUT);
+        assert_eq!(
+            source.timeout_now(Some(Duration::from_secs(2))),
+            Duration::from_secs(6),
+            "by the fastest other source's chunks"
+        );
         source.came(Duration::from_secs(4));
-        assert_eq!(source.timeout_now(), Duration::from_secs(12));
+        assert_eq!(source.timeout_now(None), Duration::from_secs(12));
+        assert_eq!(
+            source.timeout_now(Some(Duration::from_secs(1))),
+            Duration::from_secs(12),
+            "its own once it has some"
+        );
         source.came(Duration::from_millis(100));
-        assert!(source.timeout_now() < Duration::from_secs(12));
+        assert!(source.timeout_now(None) < Duration::from_secs(12));
         for _ in 0..50 {
             source.came(Duration::from_millis(100));
         }
         assert_eq!(
-            source.timeout_now(),
+            source.timeout_now(None),
             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");
+        assert_eq!(
+            source.timeout_now(None),
+            CHUNK_TIMEOUT,
+            "at most the long one"
+        );
     }
 
     /// One delta at a time: offers meanwhile wait, and are named when it ends.