Making sure that the resolved DNS lists get used in full (#3071)

* Checking that we try hard enough to get peers from dns-discovery

Signed-off-by: Jiri Peinlich <jiri.peinlich@gmail.com>

* Making sure that the resolved DNS lists get used in full

Previously we would always process the list from the beginning. We would
only try maximumpeers elements from the list and never try more. This PR
firstly shuffles the peers list each time before applying it and then
also takes from the list for as long as there are free peers slots
available.

Signed-off-by: Jiri Peinlich <jiri.peinlich@gmail.com>

Co-authored-by: Sally MacFarlane <sally.macfarlane@consensys.net>

ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/network/DefaultP2PNetwork.java

@@ -55,6 +55,7 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Optional;
@@ -376,6 +377,7 @@ public class DefaultP2PNetwork implements P2PNetwork {
   public Stream<DiscoveryPeer> streamDiscoveredPeers() {
     List<DiscoveryPeer> peers = dnsPeers.get();
     if (peers != null) {
+      Collections.shuffle(peers);
       return Stream.concat(peerDiscoveryAgent.streamDiscoveredPeers(), peers.stream());
     }
     return peerDiscoveryAgent.streamDiscoveredPeers();

ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/rlpx/RlpxAgent.java

@@ -173,12 +173,11 @@ public class RlpxAgent {
     if (!localNode.isReady()) {
       return;
     }
-    final int availablePeerSlots = Math.max(0, maxConnections - getConnectionCount());
     peerStream
+        .takeWhile(peer -> Math.max(0, maxConnections - getConnectionCount()) > 0)
         .filter(peer -> !connectionsById.containsKey(peer.getId()))
         .filter(peer -> peer.getEnodeURL().isListening())
         .filter(peerPermissions::allowNewOutboundConnectionTo)
-        .limit(availablePeerSlots)
         .forEach(this::connect);
   }
 

ethereum/p2p/src/test/java/org/hyperledger/besu/ethereum/p2p/rlpx/RlpxAgentTest.java

@@ -711,6 +711,22 @@ public class RlpxAgentTest {
     verify(agent, times(maxPeers)).connect(any(Peer.class));
   }
 
+  @Test
+  public void connect_largeStreamOfPeersFirstFewImpostors() {
+    final int maxPeers = 5;
+    final int impostorsCount = 5;
+    connectionInitializer.setAutoDisconnectCounter(impostorsCount);
+    final Stream<Peer> peerStream = Stream.generate(PeerTestHelper::createPeer).limit(20);
+
+    startAgentWithMaxPeers(maxPeers);
+    agent = spy(agent);
+    agent.connect(peerStream);
+
+    assertThat(agent.getConnectionCount()).isEqualTo(maxPeers);
+    // Check that stream was not fully iterated
+    verify(agent, times(maxPeers + impostorsCount)).connect(any(Peer.class));
+  }
+
   @Test
   public void disconnect() throws ExecutionException, InterruptedException {
     startAgent();

ethereum/p2p/src/test/java/org/hyperledger/besu/ethereum/p2p/rlpx/connections/MockConnectionInitializer.java

@@ -16,6 +16,7 @@ package org.hyperledger.besu.ethereum.p2p.rlpx.connections;
 
 import org.hyperledger.besu.ethereum.p2p.peers.Peer;
 import org.hyperledger.besu.ethereum.p2p.rlpx.ConnectCallback;
+import org.hyperledger.besu.ethereum.p2p.rlpx.wire.messages.DisconnectMessage;
 import org.hyperledger.besu.util.Subscribers;
 
 import java.net.InetSocketAddress;
@@ -32,6 +33,7 @@ public class MockConnectionInitializer implements ConnectionInitializer {
   private boolean autocompleteConnections = true;
   private final Map<Peer, CompletableFuture<PeerConnection>> incompleteConnections =
       new HashMap<>();
+  private int autoDisconnectCounter = 0;
 
   public MockConnectionInitializer(final PeerConnectionEventDispatcher eventDispatcher) {
     this.eventDispatcher = eventDispatcher;
@@ -72,6 +74,12 @@ public class MockConnectionInitializer implements ConnectionInitializer {
 
   @Override
   public CompletableFuture<PeerConnection> connect(final Peer peer) {
+    if (autoDisconnectCounter > 0) {
+      autoDisconnectCounter--;
+      MockPeerConnection mockPeerConnection = MockPeerConnection.create(peer, eventDispatcher);
+      mockPeerConnection.disconnect(DisconnectMessage.DisconnectReason.CLIENT_QUITTING);
+      return CompletableFuture.completedFuture(mockPeerConnection);
+    }
     if (autocompleteConnections) {
       return CompletableFuture.completedFuture(MockPeerConnection.create(peer, eventDispatcher));
     } else {
@@ -80,4 +88,8 @@ public class MockConnectionInitializer implements ConnectionInitializer {
       return future;
     }
   }
+
+  public void setAutoDisconnectCounter(final int autoDisconnectCounter) {
+    this.autoDisconnectCounter = autoDisconnectCounter;
+  }
 }