Record empty responses when retrying a peer task

Signed-off-by: Fabio Di Fabio <fabio.difabio@consensys.net>

Author: fab-10

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

@@ -16,6 +16,7 @@
 
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
+import org.hyperledger.besu.ethereum.eth.manager.exceptions.IncompleteResultsException;
 import org.hyperledger.besu.ethereum.eth.manager.exceptions.MaxRetriesReachedException;
 import org.hyperledger.besu.ethereum.eth.manager.exceptions.NoAvailablePeersException;
 import org.hyperledger.besu.ethereum.eth.manager.exceptions.PeerBreachedProtocolException;
@@ -93,9 +94,19 @@ protected void executeTask() {
               if (error != null) {
                 handleTaskError(error);
               } else {
-                // If we get a partial success, reset the retry counter.
-                if (!isEmptyResponse.test(peerResult)) {
+                // If we get a partial success, reset the retry counter, otherwise demote the peer
+                if (isEmptyResponse.test(peerResult)) {
+                  // record this empty response, so that the peer will be disconnected if there were
+                  // too many
+                  assignedPeer.ifPresent(
+                      peer -> {
+                        peer.recordUselessResponse(getClass().getSimpleName());
+                        throw new IncompleteResultsException(
+                            "Empty response from peer " + peer.getShortNodeId());
+                      });
+                } else {
                   retryCount = 0;
+                  assignedPeer.ifPresent(EthPeer::recordUsefulResponse);
                 }
                 executeTaskTimed();
               }
@@ -140,7 +151,9 @@ protected void handleTaskError(final Throwable error) {
   }
 
   protected boolean isRetryableError(final Throwable error) {
-    return error instanceof TimeoutException || (!assignedPeer.isPresent() && isPeerFailure(error));
+    return error instanceof IncompleteResultsException
+        || error instanceof TimeoutException
+        || (!assignedPeer.isPresent() && isPeerFailure(error));
   }
 
   protected boolean isPeerFailure(final Throwable error) {

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

@@ -25,7 +25,6 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeoutException;
 import java.util.function.Predicate;
 import java.util.stream.Stream;
 
@@ -106,11 +105,6 @@ protected void handleTaskError(final Throwable error) {
     super.handleTaskError(error);
   }
 
-  @Override
-  protected boolean isRetryableError(final Throwable error) {
-    return error instanceof TimeoutException || isPeerFailure(error);
-  }
-
   private Optional<EthPeer> selectNextPeer() {
     final Optional<EthPeer> maybeNextPeer = remainingPeersToTry().findFirst();
 
@@ -136,7 +130,7 @@ private void refreshPeers() {
     // If we are at max connections, then refresh peers disconnecting one of the failed peers,
     // or the least useful
 
-    if (peers.peerCount() >= peers.getMaxPeers()) {
+    if (peers.peerCount() >= peers.getPeerLowerBound()) {
       failedPeers.stream()
           .filter(peer -> !peer.isDisconnected())
           .findAny()

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

@@ -18,7 +18,6 @@
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
-import org.hyperledger.besu.ethereum.eth.manager.exceptions.IncompleteResultsException;
 import org.hyperledger.besu.ethereum.eth.manager.task.AbstractPeerTask.PeerTaskResult;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
@@ -44,7 +43,11 @@ protected RetryingGetBlocksFromPeersTask(
       final MetricsSystem metricsSystem,
       final int maxRetries,
       final List<BlockHeader> headers) {
-    super(ethContext, metricsSystem, Objects::isNull, maxRetries);
+    super(
+        ethContext,
+        metricsSystem,
+        res -> Objects.isNull(res) || res.getResult().isEmpty(),
+        maxRetries);
     this.protocolSchedule = protocolSchedule;
     this.headers = headers;
   }
@@ -77,22 +80,11 @@ protected CompletableFuture<PeerTaskResult<List<Block>>> executeTaskOnCurrentPee
                   .addArgument(this::getRetryCount)
                   .log();
 
-              if (peerResult.getResult().isEmpty()) {
-                currentPeer.recordUselessResponse("GetBodiesFromPeerTask");
-                throw new IncompleteResultsException(
-                    "No blocks returned by peer " + currentPeer.getShortNodeId());
-              }
-
               result.complete(peerResult);
               return peerResult;
             });
   }
 
-  @Override
-  protected boolean isRetryableError(final Throwable error) {
-    return super.isRetryableError(error) || error instanceof IncompleteResultsException;
-  }
-
   @Override
   protected void handleTaskError(final Throwable error) {
     if (getRetryCount() < getMaxRetries()) {

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

@@ -21,7 +21,6 @@
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
-import org.hyperledger.besu.ethereum.eth.manager.exceptions.IncompleteResultsException;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
 
@@ -98,21 +97,8 @@ protected CompletableFuture<List<BlockHeader>> executeTaskOnCurrentPeer(
                   referenceHash,
                   currentPeer,
                   peerResult.getResult());
-              if (peerResult.getResult().isEmpty()) {
-                currentPeer.recordUselessResponse("GetHeadersFromPeerByHashTask");
-                throw new IncompleteResultsException(
-                    "No block headers for hash "
-                        + referenceHash
-                        + " returned by peer "
-                        + currentPeer.getShortNodeId());
-              }
               result.complete(peerResult.getResult());
               return peerResult.getResult();
             });
   }
-
-  @Override
-  protected boolean isRetryableError(final Throwable error) {
-    return super.isRetryableError(error) || error instanceof IncompleteResultsException;
-  }
 }