protocols/kad/query/peers/closest: Make at_capacity use max (#1548)

When not making progress for `parallelism` time a `ClosestPeersIter`
becomes `State::Stalled`. When stalled an iterator is allowed to make
more parallel requests up to `num_results`. If `num_results` is smaller
than `parallelism` make sure to still allow up to `parallelism` requests
in-flight.

Co-Authored-By: Roman Borschel <romanb@users.noreply.github.com>

protocols/kad/src/query/peers/closest.rs

@@ -369,7 +369,9 @@ impl ClosestPeersIter {
     /// k closest nodes it has not already queried".
     fn at_capacity(&self) -> bool {
         match self.state {
-            State::Stalled => self.num_waiting >= self.config.num_results,
+            State::Stalled => self.num_waiting >= usize::max(
+                self.config.num_results, self.config.parallelism
+            ),
             State::Iterating { .. } => self.num_waiting >= self.config.parallelism,
             State::Finished => true
         }
@@ -691,4 +693,29 @@ mod tests {
 
         QuickCheck::new().tests(10).quickcheck(prop as fn(_) -> _)
     }
+
+    #[test]
+    fn stalled_at_capacity() {
+        fn prop(mut iter: ClosestPeersIter) {
+            iter.state = State::Stalled;
+
+            for i in 0..usize::max(iter.config.parallelism, iter.config.num_results) {
+                iter.num_waiting = i;
+                assert!(
+                    !iter.at_capacity(),
+                    "Iterator should not be at capacity if less than \
+                     `max(parallelism, num_results)` requests are waiting.",
+                )
+            }
+
+            iter.num_waiting = usize::max(iter.config.parallelism, iter.config.num_results);
+            assert!(
+                iter.at_capacity(),
+                "Iterator should be at capacity if `max(parallelism, num_results)` requests are \
+                 waiting.",
+            )
+        }
+
+        QuickCheck::new().tests(10).quickcheck(prop as fn(_))
+    }
 }