Use retry switching peer for world state download tasks

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

@@ -226,6 +226,10 @@ public EthPeer peer(final PeerConnection connection) {
     return ethPeer != null ? ethPeer : completeConnections.get(connection.getPeer().getId());
   }
 
+  public EthPeer peer(final Bytes peerId) {
+    return completeConnections.get(peerId);
+  }
+
   public PendingPeerRequest executePeerRequest(
       final PeerRequest request, final long minimumBlockNumber, final Optional<EthPeer> peer) {
     final long actualMinBlockNumber;

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

@@ -316,6 +316,8 @@ public void processMessage(final Capability cap, final Message message) {
       return;
     }
 
+    // TODO: stefan: find out whether ethMessage is a response or a request
+    // There are some annoying TRACE messages in the following call if this is not a response
     // This will handle responses
     ethPeers.dispatchMessage(ethPeer, ethMessage, getSupportedProtocol());
 

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/snap/RetryingGetAccountRangeFromPeerTask.java

@@ -15,20 +15,22 @@
 package org.hyperledger.besu.ethereum.eth.manager.snap;
 
 import org.hyperledger.besu.ethereum.core.BlockHeader;
+import org.hyperledger.besu.ethereum.eth.SnapProtocol;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
-import org.hyperledger.besu.ethereum.eth.manager.task.AbstractRetryingPeerTask;
+import org.hyperledger.besu.ethereum.eth.manager.task.AbstractPeerTask.PeerTaskResult;
+import org.hyperledger.besu.ethereum.eth.manager.task.AbstractRetryingSwitchingPeerTask;
 import org.hyperledger.besu.ethereum.eth.manager.task.EthTask;
 import org.hyperledger.besu.ethereum.eth.messages.snap.AccountRangeMessage;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
 
-import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Predicate;
 
 import org.apache.tuweni.bytes.Bytes32;
 
 public class RetryingGetAccountRangeFromPeerTask
-    extends AbstractRetryingPeerTask<AccountRangeMessage.AccountRangeData> {
+    extends AbstractRetryingSwitchingPeerTask<AccountRangeMessage.AccountRangeData> {
 
   private final EthContext ethContext;
   private final Bytes32 startKeyHash;
@@ -41,9 +43,9 @@ private RetryingGetAccountRangeFromPeerTask(
       final Bytes32 startKeyHash,
       final Bytes32 endKeyHash,
       final BlockHeader blockHeader,
-      final MetricsSystem metricsSystem) {
-    super(
-        ethContext, 4, data -> data.accounts().isEmpty() && data.proofs().isEmpty(), metricsSystem);
+      final MetricsSystem metricsSystem,
+      final int maxRetries) {
+    super(ethContext, metricsSystem, maxRetries);
     this.ethContext = ethContext;
     this.startKeyHash = startKeyHash;
     this.endKeyHash = endKeyHash;
@@ -56,23 +58,41 @@ public static EthTask<AccountRangeMessage.AccountRangeData> forAccountRange(
       final Bytes32 startKeyHash,
       final Bytes32 endKeyHash,
       final BlockHeader blockHeader,
-      final MetricsSystem metricsSystem) {
+      final MetricsSystem metricsSystem,
+      final int maxRetries) {
     return new RetryingGetAccountRangeFromPeerTask(
-        ethContext, startKeyHash, endKeyHash, blockHeader, metricsSystem);
+        ethContext, startKeyHash, endKeyHash, blockHeader, metricsSystem, maxRetries);
   }
 
   @Override
-  protected CompletableFuture<AccountRangeMessage.AccountRangeData> executePeerTask(
-      final Optional<EthPeer> assignedPeer) {
+  protected CompletableFuture<AccountRangeMessage.AccountRangeData> executeTaskOnCurrentPeer(
+      final EthPeer assignedPeer) {
     final GetAccountRangeFromPeerTask task =
         GetAccountRangeFromPeerTask.forAccountRange(
             ethContext, startKeyHash, endKeyHash, blockHeader, metricsSystem);
-    assignedPeer.ifPresent(task::assignPeer);
-    return executeSubTask(task::run)
-        .thenApply(
-            peerResult -> {
-              result.complete(peerResult.getResult());
-              return peerResult.getResult();
-            });
+    task.assignPeer(assignedPeer);
+    return executeSubTask(task::run).thenApply(PeerTaskResult::getResult);
+  }
+
+  @Override
+  protected boolean emptyResult(final AccountRangeMessage.AccountRangeData data) {
+    return data.accounts().isEmpty() && data.proofs().isEmpty();
+  }
+
+  @Override
+  protected boolean reportUselessIfEmptyResponse() {
+    return false;
+  }
+
+  @Override
+  protected boolean successfulResult(final AccountRangeMessage.AccountRangeData peerResult) {
+    return !emptyResult(peerResult);
+  }
+
+  @Override
+  protected Predicate<EthPeer> getPeerFilter() {
+    return (peer) ->
+        peer.getConnection().getAgreedCapabilities().stream()
+            .anyMatch((c) -> c.getName().equals(SnapProtocol.NAME));
   }
 }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/snap/RetryingGetBytecodeFromPeerTask.java

@@ -15,21 +15,24 @@
 package org.hyperledger.besu.ethereum.eth.manager.snap;
 
 import org.hyperledger.besu.ethereum.core.BlockHeader;
+import org.hyperledger.besu.ethereum.eth.SnapProtocol;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
-import org.hyperledger.besu.ethereum.eth.manager.task.AbstractRetryingPeerTask;
+import org.hyperledger.besu.ethereum.eth.manager.task.AbstractPeerTask.PeerTaskResult;
+import org.hyperledger.besu.ethereum.eth.manager.task.AbstractRetryingSwitchingPeerTask;
 import org.hyperledger.besu.ethereum.eth.manager.task.EthTask;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
 
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Predicate;
 
 import org.apache.tuweni.bytes.Bytes;
 import org.apache.tuweni.bytes.Bytes32;
 
-public class RetryingGetBytecodeFromPeerTask extends AbstractRetryingPeerTask<Map<Bytes32, Bytes>> {
+public class RetryingGetBytecodeFromPeerTask
+    extends AbstractRetryingSwitchingPeerTask<Map<Bytes32, Bytes>> {
 
   private final EthContext ethContext;
   private final List<Bytes32> codeHashes;
@@ -40,8 +43,9 @@ private RetryingGetBytecodeFromPeerTask(
       final EthContext ethContext,
       final List<Bytes32> codeHashes,
       final BlockHeader blockHeader,
-      final MetricsSystem metricsSystem) {
-    super(ethContext, 4, Map::isEmpty, metricsSystem);
+      final MetricsSystem metricsSystem,
+      final int maxRetries) {
+    super(ethContext, metricsSystem, maxRetries);
     this.ethContext = ethContext;
     this.codeHashes = codeHashes;
     this.blockHeader = blockHeader;
@@ -52,21 +56,39 @@ public static EthTask<Map<Bytes32, Bytes>> forByteCode(
       final EthContext ethContext,
       final List<Bytes32> codeHashes,
       final BlockHeader blockHeader,
-      final MetricsSystem metricsSystem) {
-    return new RetryingGetBytecodeFromPeerTask(ethContext, codeHashes, blockHeader, metricsSystem);
+      final MetricsSystem metricsSystem,
+      final int maxRetries) {
+    return new RetryingGetBytecodeFromPeerTask(
+        ethContext, codeHashes, blockHeader, metricsSystem, maxRetries);
   }
 
   @Override
-  protected CompletableFuture<Map<Bytes32, Bytes>> executePeerTask(
-      final Optional<EthPeer> assignedPeer) {
+  protected CompletableFuture<Map<Bytes32, Bytes>> executeTaskOnCurrentPeer(final EthPeer peer) {
     final GetBytecodeFromPeerTask task =
         GetBytecodeFromPeerTask.forBytecode(ethContext, codeHashes, blockHeader, metricsSystem);
-    assignedPeer.ifPresent(task::assignPeer);
-    return executeSubTask(task::run)
-        .thenApply(
-            peerResult -> {
-              result.complete(peerResult.getResult());
-              return peerResult.getResult();
-            });
+    task.assignPeer(peer);
+    return executeSubTask(task::run).thenApply(PeerTaskResult::getResult);
+  }
+
+  @Override
+  protected boolean emptyResult(final Map<Bytes32, Bytes> peerResult) {
+    return peerResult.isEmpty();
+  }
+
+  @Override
+  protected boolean reportUselessIfEmptyResponse() {
+    return false;
+  }
+
+  @Override
+  protected boolean successfulResult(final Map<Bytes32, Bytes> peerResult) {
+    return !emptyResult(peerResult);
+  }
+
+  @Override
+  protected Predicate<EthPeer> getPeerFilter() {
+    return (peer) ->
+        peer.getConnection().getAgreedCapabilities().stream()
+            .anyMatch((c) -> c.getName().equals(SnapProtocol.NAME));
   }
 }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/snap/RetryingGetStorageRangeFromPeerTask.java

@@ -15,21 +15,23 @@
 package org.hyperledger.besu.ethereum.eth.manager.snap;
 
 import org.hyperledger.besu.ethereum.core.BlockHeader;
+import org.hyperledger.besu.ethereum.eth.SnapProtocol;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
-import org.hyperledger.besu.ethereum.eth.manager.task.AbstractRetryingPeerTask;
+import org.hyperledger.besu.ethereum.eth.manager.task.AbstractPeerTask.PeerTaskResult;
+import org.hyperledger.besu.ethereum.eth.manager.task.AbstractRetryingSwitchingPeerTask;
 import org.hyperledger.besu.ethereum.eth.manager.task.EthTask;
 import org.hyperledger.besu.ethereum.eth.messages.snap.StorageRangeMessage;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
 
 import java.util.List;
-import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Predicate;
 
 import org.apache.tuweni.bytes.Bytes32;
 
 public class RetryingGetStorageRangeFromPeerTask
-    extends AbstractRetryingPeerTask<StorageRangeMessage.SlotRangeData> {
+    extends AbstractRetryingSwitchingPeerTask<StorageRangeMessage.SlotRangeData> {
 
   private final EthContext ethContext;
   private final List<Bytes32> accountHashes;
@@ -44,8 +46,9 @@ private RetryingGetStorageRangeFromPeerTask(
       final Bytes32 startKeyHash,
       final Bytes32 endKeyHash,
       final BlockHeader blockHeader,
-      final MetricsSystem metricsSystem) {
-    super(ethContext, 4, data -> data.proofs().isEmpty() && data.slots().isEmpty(), metricsSystem);
+      final MetricsSystem metricsSystem,
+      final int maxRetries) {
+    super(ethContext, metricsSystem, maxRetries);
     this.ethContext = ethContext;
     this.accountHashes = accountHashes;
     this.startKeyHash = startKeyHash;
@@ -60,23 +63,47 @@ public static EthTask<StorageRangeMessage.SlotRangeData> forStorageRange(
       final Bytes32 startKeyHash,
       final Bytes32 endKeyHash,
       final BlockHeader blockHeader,
-      final MetricsSystem metricsSystem) {
+      final MetricsSystem metricsSystem,
+      final int maxRetries) {
     return new RetryingGetStorageRangeFromPeerTask(
-        ethContext, accountHashes, startKeyHash, endKeyHash, blockHeader, metricsSystem);
+        ethContext,
+        accountHashes,
+        startKeyHash,
+        endKeyHash,
+        blockHeader,
+        metricsSystem,
+        maxRetries);
   }
 
   @Override
-  protected CompletableFuture<StorageRangeMessage.SlotRangeData> executePeerTask(
-      final Optional<EthPeer> assignedPeer) {
+  protected CompletableFuture<StorageRangeMessage.SlotRangeData> executeTaskOnCurrentPeer(
+      final EthPeer peer) {
     final GetStorageRangeFromPeerTask task =
         GetStorageRangeFromPeerTask.forStorageRange(
             ethContext, accountHashes, startKeyHash, endKeyHash, blockHeader, metricsSystem);
-    assignedPeer.ifPresent(task::assignPeer);
-    return executeSubTask(task::run)
-        .thenApply(
-            peerResult -> {
-              result.complete(peerResult.getResult());
-              return peerResult.getResult();
-            });
+    task.assignPeer(peer);
+    return executeSubTask(task::run).thenApply(PeerTaskResult::getResult);
+  }
+
+  @Override
+  protected boolean emptyResult(final StorageRangeMessage.SlotRangeData peerResult) {
+    return peerResult.proofs().isEmpty() && peerResult.slots().isEmpty();
+  }
+
+  @Override
+  protected boolean reportUselessIfEmptyResponse() {
+    return false;
+  }
+
+  @Override
+  protected boolean successfulResult(final StorageRangeMessage.SlotRangeData peerResult) {
+    return !emptyResult(peerResult);
+  }
+
+  @Override
+  protected Predicate<EthPeer> getPeerFilter() {
+    return (peer) ->
+        peer.getConnection().getAgreedCapabilities().stream()
+            .anyMatch((c) -> c.getName().equals(SnapProtocol.NAME));
   }
 }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/snap/RetryingGetTrieNodeFromPeerTask.java

@@ -15,20 +15,23 @@
 package org.hyperledger.besu.ethereum.eth.manager.snap;
 
 import org.hyperledger.besu.ethereum.core.BlockHeader;
+import org.hyperledger.besu.ethereum.eth.SnapProtocol;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
-import org.hyperledger.besu.ethereum.eth.manager.task.AbstractRetryingPeerTask;
+import org.hyperledger.besu.ethereum.eth.manager.task.AbstractPeerTask.PeerTaskResult;
+import org.hyperledger.besu.ethereum.eth.manager.task.AbstractRetryingSwitchingPeerTask;
 import org.hyperledger.besu.ethereum.eth.manager.task.EthTask;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
 
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Predicate;
 
 import org.apache.tuweni.bytes.Bytes;
 
-public class RetryingGetTrieNodeFromPeerTask extends AbstractRetryingPeerTask<Map<Bytes, Bytes>> {
+public class RetryingGetTrieNodeFromPeerTask
+    extends AbstractRetryingSwitchingPeerTask<Map<Bytes, Bytes>> {
 
   private final EthContext ethContext;
   private final Map<Bytes, List<Bytes>> paths;
@@ -39,8 +42,9 @@ private RetryingGetTrieNodeFromPeerTask(
       final EthContext ethContext,
       final Map<Bytes, List<Bytes>> paths,
       final BlockHeader blockHeader,
-      final MetricsSystem metricsSystem) {
-    super(ethContext, 4, Map::isEmpty, metricsSystem);
+      final MetricsSystem metricsSystem,
+      final int maxRetries) {
+    super(ethContext, metricsSystem, maxRetries);
     this.ethContext = ethContext;
     this.paths = paths;
     this.blockHeader = blockHeader;
@@ -51,21 +55,39 @@ public static EthTask<Map<Bytes, Bytes>> forTrieNodes(
       final EthContext ethContext,
       final Map<Bytes, List<Bytes>> paths,
       final BlockHeader blockHeader,
-      final MetricsSystem metricsSystem) {
-    return new RetryingGetTrieNodeFromPeerTask(ethContext, paths, blockHeader, metricsSystem);
+      final MetricsSystem metricsSystem,
+      final int maxRetries) {
+    return new RetryingGetTrieNodeFromPeerTask(
+        ethContext, paths, blockHeader, metricsSystem, maxRetries);
   }
 
   @Override
-  protected CompletableFuture<Map<Bytes, Bytes>> executePeerTask(
-      final Optional<EthPeer> assignedPeer) {
+  protected CompletableFuture<Map<Bytes, Bytes>> executeTaskOnCurrentPeer(final EthPeer peer) {
     final GetTrieNodeFromPeerTask task =
         GetTrieNodeFromPeerTask.forTrieNodes(ethContext, paths, blockHeader, metricsSystem);
-    assignedPeer.ifPresent(task::assignPeer);
-    return executeSubTask(task::run)
-        .thenApply(
-            peerResult -> {
-              result.complete(peerResult.getResult());
-              return peerResult.getResult();
-            });
+    task.assignPeer(peer);
+    return executeSubTask(task::run).thenApply(PeerTaskResult::getResult);
+  }
+
+  @Override
+  protected boolean emptyResult(final Map<Bytes, Bytes> peerResult) {
+    return peerResult.isEmpty();
+  }
+
+  @Override
+  protected boolean reportUselessIfEmptyResponse() {
+    return false;
+  }
+
+  @Override
+  protected boolean successfulResult(final Map<Bytes, Bytes> peerResult) {
+    return !emptyResult(peerResult);
+  }
+
+  @Override
+  protected Predicate<EthPeer> getPeerFilter() {
+    return (peer) ->
+        peer.getConnection().getAgreedCapabilities().stream()
+            .anyMatch((c) -> c.getName().equals(SnapProtocol.NAME));
   }
 }