Peering - disconnects refactor

CHANGELOG.md

@@ -22,6 +22,7 @@
 - Default bonsai to use full-flat db and code-storage-by-code-hash [#6984](https://github.com/hyperledger/besu/pull/6894)
 - New RPC methods miner_setExtraData and miner_getExtraData [#7078](https://github.com/hyperledger/besu/pull/7078)
 - Disconnect peers that have multiple discovery ports since they give us bad neighbours [#7089](https://github.com/hyperledger/besu/pull/7089)
+- Peering - refactor disconnect logic to improve peer acquisition and retention [#6968](https://github.com/hyperledger/besu/pull/6968)
 
 ### Known Issues
 - [Frequency: occasional < 10%] Chain download halt. Only affects new syncs (new nodes syncing from scratch). Symptom: Block import halts, despite having a full set of peers and world state downloading finishing. Generally restarting besu will resolve the issue. We are tracking this in [#6884](https://github.com/hyperledger/besu/pull/6884)

besu/src/main/java/org/hyperledger/besu/controller/MergeBesuControllerBuilder.java

@@ -104,7 +104,6 @@ protected EthProtocolManager createEthProtocolManager(
     var mergeBestPeerComparator =
         new TransitionBestPeerComparator(
             genesisConfigOptions.getTerminalTotalDifficulty().map(Difficulty::of).orElseThrow());
-    ethPeers.setBestChainComparator(mergeBestPeerComparator);
     mergeContext.observeNewIsPostMergeState(mergeBestPeerComparator);
 
     Optional<MergePeerFilter> filterToUse = Optional.of(new MergePeerFilter());

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/EthPeer.java

@@ -215,7 +215,7 @@ public void recordRequestTimeout(final int requestCode) {
         .addArgument(this::getLoggableId)
         .log();
     LOG.trace("Timed out while waiting for response from peer {}", this);
-    reputation.recordRequestTimeout(requestCode, this).ifPresent(this::disconnect);
+    reputation.recordRequestTimeout(requestCode, this);
   }
 
   public void recordUselessResponse(final String requestType) {
@@ -224,7 +224,7 @@ public void recordUselessResponse(final String requestType) {
         .addArgument(requestType)
         .addArgument(this::getLoggableId)
         .log();
-    reputation.recordUselessResponse(System.currentTimeMillis(), this).ifPresent(this::disconnect);
+    reputation.recordUselessResponse(System.currentTimeMillis(), this);
   }
 
   public void recordUsefulResponse() {

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/EthPeers.java

@@ -115,7 +115,7 @@ public EthPeers(
     this.clock = clock;
     this.permissioningProviders = permissioningProviders;
     this.maxMessageSize = maxMessageSize;
-    this.bestPeerComparator = HEAVIEST_CHAIN;
+    this.bestPeerComparator = MOST_USEFUL_PEER;
     this.localNodeId = localNodeId;
     this.peerUpperBound = peerUpperBound;
     this.maxRemotelyInitiatedConnections = maxRemotelyInitiatedConnections;
@@ -335,7 +335,7 @@ public Optional<EthPeer> bestPeerMatchingCriteria(final Predicate<EthPeer> match
   }
 
   public void setBestChainComparator(final Comparator<EthPeer> comparator) {
-    LOG.info("Updating the default best peer comparator");
+    LOG.info("Updating the default best peer comparator to {}", comparator);
     bestPeerComparator = comparator;
   }
 
@@ -363,13 +363,6 @@ public Stream<PeerConnection> getAllConnections() {
 
   public boolean shouldConnect(final Peer peer, final boolean inbound) {
     final Bytes id = peer.getId();
-    if (peerCount() >= peerUpperBound && !canExceedPeerLimits(id)) {
-      LOG.atTrace()
-          .setMessage("not connecting to peer {} - too many peers")
-          .addArgument(peer.getLoggableId())
-          .log();
-      return false;
-    }
     final EthPeer ethPeer = completeConnections.get(id);
     if (ethPeer != null && !ethPeer.isDisconnected()) {
       LOG.atTrace()
@@ -392,21 +385,44 @@ public boolean shouldConnect(final Peer peer, final boolean inbound) {
     return true;
   }
 
-  public void disconnectWorstUselessPeer() {
-    streamAvailablePeers()
-        .sorted(getBestChainComparator())
-        .findFirst()
-        .ifPresent(
-            peer -> {
-              LOG.atDebug()
-                  .setMessage(
-                      "disconnecting peer {}. Waiting for better peers. Current {} of max {}")
-                  .addArgument(peer::getLoggableId)
-                  .addArgument(this::peerCount)
-                  .addArgument(this::getMaxPeers)
-                  .log();
-              peer.disconnect(DisconnectMessage.DisconnectReason.USELESS_PEER_BY_CHAIN_COMPARATOR);
-            });
+  public void disconnectWorstUselessPeerIfAtCapacityIncludingConnectingPeer(
+      final EthPeer connectingPeer) {
+    if (peerCount() >= getMaxPeers()) {
+      streamAvailablePeers()
+          .filter(p -> !canExceedPeerLimits(p.getId()))
+          .min(getBestChainComparator())
+          .ifPresent(
+              worstCurrentlyConnectedPeer -> {
+                // TODO remove this debug log
+                LOG.atDebug()
+                    .setMessage("comparing worstCurrentPeer {} with connectingPeer {}")
+                    .addArgument(worstCurrentlyConnectedPeer)
+                    .addArgument(connectingPeer)
+                    .log();
+                if (getBestChainComparator().compare(worstCurrentlyConnectedPeer, connectingPeer)
+                    < 0) {
+
+                  LOG.atDebug()
+                      .setMessage(
+                          "disconnecting current peer {}. Waiting for better peers. Current {} of max {}")
+                      .addArgument(worstCurrentlyConnectedPeer::getLoggableId)
+                      .addArgument(this::peerCount)
+                      .addArgument(this::getMaxPeers)
+                      .log();
+                  worstCurrentlyConnectedPeer.disconnect(
+                      DisconnectMessage.DisconnectReason.USELESS_PEER_BY_CHAIN_COMPARATOR);
+                } else {
+                  LOG.atDebug()
+                      .setMessage(
+                          "disconnecting connecting peer {}. Waiting for better peers. Current {} of max {}")
+                      .addArgument(connectingPeer::getLoggableId)
+                      .addArgument(this::peerCount)
+                      .addArgument(this::getMaxPeers)
+                      .log();
+                  connectingPeer.disconnect(DisconnectMessage.DisconnectReason.TOO_MANY_PEERS);
+                }
+              });
+    }
   }
 
   @FunctionalInterface
@@ -510,24 +526,10 @@ private void enforceConnectionLimits() {
             });
   }
 
-  private boolean remoteConnectionLimitReached() {
-    return shouldLimitRemoteConnections()
-        && countUntrustedRemotelyInitiatedConnections() >= maxRemotelyInitiatedConnections;
-  }
-
   private boolean shouldLimitRemoteConnections() {
     return maxRemotelyInitiatedConnections < peerUpperBound;
   }
 
-  private long countUntrustedRemotelyInitiatedConnections() {
-    return completeConnections.values().stream()
-        .map(ep -> ep.getConnection())
-        .filter(c -> c.inboundInitiated())
-        .filter(c -> !c.isDisconnected())
-        .filter(conn -> !canExceedPeerLimits(conn.getPeer().getId()))
-        .count();
-  }
-
   private void onCacheRemoval(
       final RemovalNotification<PeerConnection, EthPeer> removalNotification) {
     if (removalNotification.wasEvicted()) {
@@ -548,40 +550,62 @@ private boolean addPeerToEthPeers(final EthPeer peer) {
     }
     final PeerConnection connection = peer.getConnection();
     final Bytes id = peer.getId();
-    if (!randomPeerPriority) {
-      // Disconnect if too many peers
-      if (!canExceedPeerLimits(id) && peerCount() >= peerUpperBound) {
-        LOG.trace(
-            "Too many peers. Disconnect connection: {}, max connections {}",
-            connection,
-            peerUpperBound);
-        connection.disconnect(DisconnectMessage.DisconnectReason.TOO_MANY_PEERS);
-        return false;
-      }
-      // Disconnect if too many remotely-initiated connections
-      if (connection.inboundInitiated()
-          && !canExceedPeerLimits(id)
-          && remoteConnectionLimitReached()) {
-        LOG.trace(
-            "Too many remotely-initiated connections. Disconnect incoming connection: {}, maxRemote={}",
-            connection,
-            maxRemotelyInitiatedConnections);
-        connection.disconnect(DisconnectMessage.DisconnectReason.TOO_MANY_PEERS);
-        return false;
-      }
-      final boolean added = (completeConnections.putIfAbsent(id, peer) == null);
-      if (added) {
-        LOG.trace("Added peer {} with connection {} to completeConnections", id, connection);
-      } else {
-        LOG.trace("Did not add peer {} with connection {} to completeConnections", id, connection);
-      }
-      return added;
-    } else {
+    if (randomPeerPriority || canExceedPeerLimits(id)) {
+      // either the incoming peer is a static peer, or
       // randomPeerPriority! Add the peer and if there are too many connections fix it
-      completeConnections.putIfAbsent(id, peer);
-      enforceRemoteConnectionLimits();
-      enforceConnectionLimits();
-      return completeConnections.containsKey(id);
+      return addPeerAndEnforceConnectionLimits(peer, id, connection);
+    }
+    // otherwise, we need to compare the incoming peer to existing peers
+
+    // Disconnect if too many peers
+    if (peerCount() >= peerUpperBound) {
+      // If too many peers, disconnect one (could be the connecting peer if it does not compare
+      // favourably
+      disconnectWorstUselessPeerIfAtCapacityIncludingConnectingPeer(peer);
+    }
+    return addPeerAndEnforceConnectionLimits(peer, id, connection);
+  }
+
+  /**
+   * Returns true if this specific peer was added, after connection limits were checked
+   *
+   * @param peer the peer to add
+   * @param id the peer's id
+   * @param connection the peer's connection
+   * @return whether the peer remains added after connection limits are enforced
+   */
+  private boolean addPeerAndEnforceConnectionLimits(
+      final EthPeer peer, final Bytes id, final PeerConnection connection) {
+    completeConnections.putIfAbsent(id, peer);
+    enforceRemoteConnectionLimits();
+    enforceConnectionLimits();
+    final boolean added = completeConnections.containsKey(id);
+    if (added) {
+      LOG.trace("Added peer {} with connection {} to completeConnections", id, connection);
+    } else {
+      LOG.trace("Did not add peer {} with connection {} to completeConnections", id, connection);
+    }
+    return added;
+  }
+
+  public void disconnectWorstIncomingUselessPeer() {
+    if (peerCount() >= getMaxPeers()) {
+      streamAvailablePeers()
+          .filter(p -> p.getConnection().inboundInitiated())
+          .filter(p -> !canExceedPeerLimits(p.getId()))
+          .min(getBestChainComparator())
+          .ifPresent(
+              peer -> {
+                LOG.atDebug()
+                    .setMessage(
+                        "disconnecting peer {}. Waiting for better peers. Current {} of max {}")
+                    .addArgument(peer::getLoggableId)
+                    .addArgument(this::peerCount)
+                    .addArgument(this::getMaxPeers)
+                    .log();
+                peer.disconnect(
+                    DisconnectMessage.DisconnectReason.USELESS_PEER_BY_CHAIN_COMPARATOR);
+              });
     }
   }
 }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/EthProtocolManager.java

@@ -63,7 +63,6 @@
 
 public class EthProtocolManager implements ProtocolManager, MinedBlockObserver {
   private static final Logger LOG = LoggerFactory.getLogger(EthProtocolManager.class);
-
   private final EthScheduler scheduler;
   private final CountDownLatch shutdown;
   private final AtomicBoolean stopped = new AtomicBoolean(false);
@@ -405,9 +404,7 @@ public boolean shouldConnect(final Peer peer, final boolean incoming) {
           .setMessage("ForkId OK or not available for peer {}")
           .addArgument(peer::getLoggableId)
           .log();
-      if (ethPeers.shouldConnect(peer, incoming)) {
-        return true;
-      }
+      return (ethPeers.shouldConnect(peer, incoming));
     } else {
       LOG.atDebug()
           .setMessage("ForkId check failed for peer {} our fork id {} theirs {}")
@@ -417,7 +414,6 @@ public boolean shouldConnect(final Peer peer, final boolean incoming) {
           .log();
       return false;
     }
-    return false;
   }
 
   @Override

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/task/AbstractRetryingSwitchingPeerTask.java

@@ -132,6 +132,7 @@ private Stream<EthPeer> remainingPeersToTry() {
   }
 
   private void refreshPeers() {
+    // TODO this duplicates EthPeers.disconnectWorst
     final EthPeers peers = getEthContext().getEthPeers();
     // If we are at max connections, then refresh peers disconnecting one of the failed peers,
     // or the least useful

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/FastSyncDownloader.java

@@ -163,7 +163,7 @@ public void deleteFastSyncState() {
   protected FastSyncState updateMaxTrailingPeers(final FastSyncState state) {
     if (state.getPivotBlockNumber().isPresent()) {
       trailingPeerRequirements =
-          Optional.of(new TrailingPeerRequirements(state.getPivotBlockNumber().getAsLong(), 0));
+          Optional.of(new TrailingPeerRequirements(state.getPivotBlockNumber().getAsLong(), 5));
     } else {
       trailingPeerRequirements = Optional.empty();
     }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/SyncTargetManager.java

@@ -101,7 +101,6 @@ protected CompletableFuture<Optional<EthPeer>> selectBestAvailableSyncTarget() {
             pivotBlockHeader.getNumber(),
             ethPeers.peerCount(),
             ethPeers.getMaxPeers());
-        ethPeers.disconnectWorstUselessPeer();
         return completedFuture(Optional.empty());
       } else {
         return confirmPivotBlockHeader(bestPeer);

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/EthPeersTest.java

@@ -75,10 +75,6 @@ public void comparesPeersWithHeightAndTd() {
     assertThat(EthPeers.HEAVIEST_CHAIN.compare(peerB, peerA)).isGreaterThan(0);
     assertThat(EthPeers.HEAVIEST_CHAIN.compare(peerA, peerA)).isEqualTo(0);
     assertThat(EthPeers.HEAVIEST_CHAIN.compare(peerB, peerB)).isEqualTo(0);
-
-    assertThat(ethProtocolManager.ethContext().getEthPeers().bestPeer()).contains(peerB);
-    assertThat(ethProtocolManager.ethContext().getEthPeers().bestPeerWithHeightEstimate())
-        .contains(peerB);
   }
 
   @Test
@@ -104,7 +100,6 @@ public void comparesPeersWithTdAndNoHeight() {
     assertThat(EthPeers.HEAVIEST_CHAIN.compare(peerA, peerA)).isEqualTo(0);
     assertThat(EthPeers.HEAVIEST_CHAIN.compare(peerB, peerB)).isEqualTo(0);
 
-    assertThat(ethProtocolManager.ethContext().getEthPeers().bestPeer()).contains(peerA);
     assertThat(ethProtocolManager.ethContext().getEthPeers().bestPeerWithHeightEstimate())
         .isEmpty();
   }