7311: add GetReceiptsFromPeerTask

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

@@ -55,6 +55,10 @@
 import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.manager.MergePeerFilter;
 import org.hyperledger.besu.ethereum.eth.manager.MonitoredExecutors;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.DefaultPeerSelector;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerSelector;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutor;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskRequestSender;
 import org.hyperledger.besu.ethereum.eth.manager.snap.SnapProtocolManager;
 import org.hyperledger.besu.ethereum.eth.peervalidation.CheckpointBlocksPeerValidator;
 import org.hyperledger.besu.ethereum.eth.peervalidation.ClassicForkPeerValidator;
@@ -652,6 +656,10 @@ public BesuController build() {
     }
 
     final EthContext ethContext = new EthContext(ethPeers, ethMessages, snapMessages, scheduler);
+    final PeerSelector peerSelector = new DefaultPeerSelector(currentProtocolSpecSupplier);
+    ethPeers.streamAllPeers().forEach(peerSelector::addPeer);
+    final PeerTaskExecutor peerTaskExecutor =
+        new PeerTaskExecutor(peerSelector, new PeerTaskRequestSender(), metricsSystem);
     final boolean fullSyncDisabled = !SyncMode.isFullSync(syncConfig.getSyncMode());
     final SyncState syncState = new SyncState(blockchain, ethPeers, fullSyncDisabled, checkpoint);
 
@@ -691,7 +699,8 @@ public BesuController build() {
             scheduler,
             peerValidators,
             Optional.empty(),
-            forkIdManager);
+            forkIdManager,
+            peerSelector);
 
     final PivotBlockSelector pivotBlockSelector =
         createPivotSelector(
@@ -703,6 +712,7 @@ public BesuController build() {
             worldStateStorageCoordinator,
             protocolContext,
             ethContext,
+            peerTaskExecutor,
             syncState,
             ethProtocolManager,
             pivotBlockSelector);
@@ -825,6 +835,7 @@ private TrieLogPruner createTrieLogPruner(
    * @param worldStateStorageCoordinator the world state storage
    * @param protocolContext the protocol context
    * @param ethContext the eth context
+   * @param peerTaskExecutor the PeerTaskExecutor
    * @param syncState the sync state
    * @param ethProtocolManager the eth protocol manager
    * @param pivotBlockSelector the pivot block selector
@@ -835,6 +846,7 @@ protected DefaultSynchronizer createSynchronizer(
       final WorldStateStorageCoordinator worldStateStorageCoordinator,
       final ProtocolContext protocolContext,
       final EthContext ethContext,
+      final PeerTaskExecutor peerTaskExecutor,
       final SyncState syncState,
       final EthProtocolManager ethProtocolManager,
       final PivotBlockSelector pivotBlockSelector) {
@@ -846,6 +858,7 @@ protected DefaultSynchronizer createSynchronizer(
         worldStateStorageCoordinator,
         ethProtocolManager.getBlockBroadcaster(),
         ethContext,
+        peerTaskExecutor,
         syncState,
         dataDirectory,
         storageProvider,
@@ -1022,6 +1035,7 @@ protected String getSupportedProtocol() {
    * @param peerValidators the peer validators
    * @param mergePeerFilter the merge peer filter
    * @param forkIdManager the fork id manager
+   * @param peerSelector the PeerSelector
    * @return the eth protocol manager
    */
   protected EthProtocolManager createEthProtocolManager(
@@ -1035,7 +1049,8 @@ protected EthProtocolManager createEthProtocolManager(
       final EthScheduler scheduler,
       final List<PeerValidator> peerValidators,
       final Optional<MergePeerFilter> mergePeerFilter,
-      final ForkIdManager forkIdManager) {
+      final ForkIdManager forkIdManager,
+      final PeerSelector peerSelector) {
     return new EthProtocolManager(
         protocolContext.getBlockchain(),
         networkId,
@@ -1049,7 +1064,8 @@ protected EthProtocolManager createEthProtocolManager(
         mergePeerFilter,
         synchronizerConfiguration,
         scheduler,
-        forkIdManager);
+        forkIdManager,
+        peerSelector);
   }
 
   /**

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

@@ -42,6 +42,7 @@
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
 import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.manager.MergePeerFilter;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerSelector;
 import org.hyperledger.besu.ethereum.eth.manager.snap.SnapProtocolManager;
 import org.hyperledger.besu.ethereum.eth.peervalidation.PeerValidator;
 import org.hyperledger.besu.ethereum.eth.sync.SynchronizerConfiguration;
@@ -243,7 +244,8 @@ protected EthProtocolManager createEthProtocolManager(
       final EthScheduler scheduler,
       final List<PeerValidator> peerValidators,
       final Optional<MergePeerFilter> mergePeerFilter,
-      final ForkIdManager forkIdManager) {
+      final ForkIdManager forkIdManager,
+      final PeerSelector peerSelector) {
     return besuControllerBuilderSchedule
         .get(0L)
         .createEthProtocolManager(
@@ -257,7 +259,8 @@ protected EthProtocolManager createEthProtocolManager(
             scheduler,
             peerValidators,
             mergePeerFilter,
-            forkIdManager);
+            forkIdManager,
+            peerSelector);
   }
 
   @Override

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

@@ -34,6 +34,7 @@
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
 import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.manager.MergePeerFilter;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerSelector;
 import org.hyperledger.besu.ethereum.eth.peervalidation.PeerValidator;
 import org.hyperledger.besu.ethereum.eth.peervalidation.RequiredBlocksPeerValidator;
 import org.hyperledger.besu.ethereum.eth.sync.SynchronizerConfiguration;
@@ -99,7 +100,8 @@ protected EthProtocolManager createEthProtocolManager(
       final EthScheduler scheduler,
       final List<PeerValidator> peerValidators,
       final Optional<MergePeerFilter> mergePeerFilter,
-      final ForkIdManager forkIdManager) {
+      final ForkIdManager forkIdManager,
+      final PeerSelector peerSelector) {
 
     var mergeContext = protocolContext.getConsensusContext(MergeContext.class);
 
@@ -129,7 +131,8 @@ protected EthProtocolManager createEthProtocolManager(
             scheduler,
             peerValidators,
             filterToUse,
-            forkIdManager);
+            forkIdManager,
+            peerSelector);
 
     return ethProtocolManager;
   }

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

@@ -40,6 +40,8 @@
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
 import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.manager.MergePeerFilter;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerSelector;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutor;
 import org.hyperledger.besu.ethereum.eth.peervalidation.PeerValidator;
 import org.hyperledger.besu.ethereum.eth.sync.DefaultSynchronizer;
 import org.hyperledger.besu.ethereum.eth.sync.PivotBlockSelector;
@@ -163,7 +165,8 @@ protected EthProtocolManager createEthProtocolManager(
       final EthScheduler scheduler,
       final List<PeerValidator> peerValidators,
       final Optional<MergePeerFilter> mergePeerFilter,
-      final ForkIdManager forkIdManager) {
+      final ForkIdManager forkIdManager,
+      final PeerSelector peerSelector) {
     return mergeBesuControllerBuilder.createEthProtocolManager(
         protocolContext,
         synchronizerConfiguration,
@@ -175,7 +178,8 @@ protected EthProtocolManager createEthProtocolManager(
         scheduler,
         peerValidators,
         mergePeerFilter,
-        forkIdManager);
+        forkIdManager,
+        peerSelector);
   }
 
   @Override
@@ -225,6 +229,7 @@ protected DefaultSynchronizer createSynchronizer(
       final WorldStateStorageCoordinator worldStateStorageCoordinator,
       final ProtocolContext protocolContext,
       final EthContext ethContext,
+      final PeerTaskExecutor peerTaskExecutor,
       final SyncState syncState,
       final EthProtocolManager ethProtocolManager,
       final PivotBlockSelector pivotBlockSelector) {
@@ -235,6 +240,7 @@ protected DefaultSynchronizer createSynchronizer(
             worldStateStorageCoordinator,
             protocolContext,
             ethContext,
+            peerTaskExecutor,
             syncState,
             ethProtocolManager,
             pivotBlockSelector);

besu/src/test/java/org/hyperledger/besu/RunnerTest.java

@@ -119,7 +119,7 @@ public final class RunnerTest {
   private Vertx vertx;
 
   @BeforeEach
-  public void initVertx() {
+  public void initVertx() throws IllegalAccessException {
     vertx = Vertx.vertx();
   }
 
@@ -131,7 +131,7 @@ public void stopVertx() {
   @TempDir private Path temp;
 
   @Test
-  public void getFixedNodes() {
+  public void getFixedNodes() throws IllegalAccessException {
     final EnodeURL staticNode =
         EnodeURLImpl.fromString(
             "enode://8f4b88336cc40ef2516d8b27df812e007fb2384a61e93635f1899051311344f3dcdbb49a4fe49a79f66d2f589a9f282e8cc4f1d7381e8ef7e4fcc6b0db578c77@127.0.0.1:30301");
@@ -153,26 +153,48 @@ public void fullSyncFromGenesis() throws Exception {
     // set merge flag to false, otherwise this test can fail if a merge test runs first
     MergeConfigOptions.setMergeEnabled(false);
 
-    syncFromGenesis(SyncMode.FULL, getFastSyncGenesis());
+    syncFromGenesis(SyncMode.FULL, getFastSyncGenesis(), false);
+  }
+
+  @Test
+  public void fullSyncFromGenesisUsingPeerTaskSystem() throws Exception {
+    // set merge flag to false, otherwise this test can fail if a merge test runs first
+    MergeConfigOptions.setMergeEnabled(false);
+
+    syncFromGenesis(SyncMode.FULL, getFastSyncGenesis(), true);
   }
 
   @Test
   public void fastSyncFromGenesis() throws Exception {
     // set merge flag to false, otherwise this test can fail if a merge test runs first
     MergeConfigOptions.setMergeEnabled(false);
 
-    syncFromGenesis(SyncMode.FAST, getFastSyncGenesis());
+    syncFromGenesis(SyncMode.FAST, getFastSyncGenesis(), false);
+  }
+
+  @Test
+  public void fastSyncFromGenesisUsingPeerTaskSystem() throws Exception {
+    // set merge flag to false, otherwise this test can fail if a merge test runs first
+    MergeConfigOptions.setMergeEnabled(false);
+
+    syncFromGenesis(SyncMode.FAST, getFastSyncGenesis(), true);
   }
 
-  private void syncFromGenesis(final SyncMode mode, final GenesisConfigFile genesisConfig)
+  private void syncFromGenesis(
+      final SyncMode mode,
+      final GenesisConfigFile genesisConfig,
+      final boolean isPeerTaskSystemEnabled)
       throws Exception {
     final Path dataDirAhead = Files.createTempDirectory(temp, "db-ahead");
     final Path dbAhead = dataDirAhead.resolve("database");
     final int blockCount = 500;
     final NodeKey aheadDbNodeKey = NodeKeyUtils.createFrom(KeyPairUtil.loadKeyPair(dataDirAhead));
     final NodeKey behindDbNodeKey = NodeKeyUtils.generate();
     final SynchronizerConfiguration syncConfigAhead =
-        SynchronizerConfiguration.builder().syncMode(SyncMode.FULL).build();
+        SynchronizerConfiguration.builder()
+            .syncMode(SyncMode.FULL)
+            .isPeerTaskSystemEnabled(isPeerTaskSystemEnabled)
+            .build();
     final ObservableMetricsSystem noOpMetricsSystem = new NoOpMetricsSystem();
     final var miningParameters = MiningParameters.newDefault();
     final var dataStorageConfiguration = DataStorageConfiguration.DEFAULT_FOREST_CONFIG;

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/BodyValidator.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.ethereum.mainnet;
+
+import org.hyperledger.besu.datatypes.Hash;
+import org.hyperledger.besu.ethereum.core.BlockHeader;
+import org.hyperledger.besu.ethereum.core.Request;
+import org.hyperledger.besu.ethereum.core.Transaction;
+import org.hyperledger.besu.ethereum.core.TransactionReceipt;
+import org.hyperledger.besu.ethereum.core.Withdrawal;
+import org.hyperledger.besu.evm.log.LogsBloomFilter;
+
+import java.util.List;
+
+/** A utility class for body validation tasks. Implemented utilising BodyValidation */
+public class BodyValidator {
+
+  /**
+   * Generates the transaction root for a list of transactions
+   *
+   * @param transactions the transactions
+   * @return the transaction root
+   */
+  public Hash transactionsRoot(final List<Transaction> transactions) {
+    return BodyValidation.transactionsRoot(transactions);
+  }
+
+  /**
+   * Generates the withdrawals root for a list of withdrawals
+   *
+   * @param withdrawals the transactions
+   * @return the transaction root
+   */
+  public Hash withdrawalsRoot(final List<Withdrawal> withdrawals) {
+    return BodyValidation.withdrawalsRoot(withdrawals);
+  }
+
+  /**
+   * Generates the requests root for a list of requests
+   *
+   * @param requests list of request
+   * @return the requests root
+   */
+  public Hash requestsRoot(final List<Request> requests) {
+    return BodyValidation.requestsRoot(requests);
+  }
+
+  /**
+   * Generates the receipt root for a list of receipts
+   *
+   * @param receipts the receipts
+   * @return the receipt root
+   */
+  public Hash receiptsRoot(final List<TransactionReceipt> receipts) {
+    return BodyValidation.receiptsRoot(receipts);
+  }
+
+  /**
+   * Generates the ommers hash for a list of ommer block headers
+   *
+   * @param ommers the ommer block headers
+   * @return the ommers hash
+   */
+  public Hash ommersHash(final List<BlockHeader> ommers) {
+    return BodyValidation.ommersHash(ommers);
+  }
+
+  /**
+   * Generates the logs bloom filter for a list of transaction receipts
+   *
+   * @param receipts the transaction receipts
+   * @return the logs bloom filter
+   */
+  public LogsBloomFilter logsBloom(final List<TransactionReceipt> receipts) {
+    return BodyValidation.logsBloom(receipts);
+  }
+}