7311 add get headers from peer task

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

@@ -690,7 +690,8 @@ public BesuController build() {
             besuComponent.map(BesuComponent::getBlobCache).orElse(new BlobCache()),
             miningConfiguration);
 
-    final List<PeerValidator> peerValidators = createPeerValidators(protocolSchedule);
+    final List<PeerValidator> peerValidators =
+        createPeerValidators(protocolSchedule, peerTaskExecutor);
 
     final EthProtocolManager ethProtocolManager =
         createEthProtocolManager(
@@ -708,7 +709,13 @@ public BesuController build() {
 
     final PivotBlockSelector pivotBlockSelector =
         createPivotSelector(
-            protocolSchedule, protocolContext, ethContext, syncState, metricsSystem, blockchain);
+            protocolSchedule,
+            protocolContext,
+            ethContext,
+            peerTaskExecutor,
+            syncState,
+            metricsSystem,
+            blockchain);
 
     final DefaultSynchronizer synchronizer =
         createSynchronizer(
@@ -743,6 +750,7 @@ public BesuController build() {
             protocolContext,
             transactionPool,
             miningConfiguration,
+            peerTaskExecutor,
             syncState,
             ethProtocolManager);
 
@@ -881,6 +889,7 @@ private PivotBlockSelector createPivotSelector(
       final ProtocolSchedule protocolSchedule,
       final ProtocolContext protocolContext,
       final EthContext ethContext,
+      final PeerTaskExecutor peerTaskExecutor,
       final SyncState syncState,
       final MetricsSystem metricsSystem,
       final Blockchain blockchain) {
@@ -918,6 +927,8 @@ private PivotBlockSelector createPivotSelector(
           ethContext,
           metricsSystem,
           genesisConfigOptions,
+          peerTaskExecutor,
+          syncConfig,
           unverifiedForkchoiceSupplier,
           unsubscribeForkchoiceListener);
     } else {
@@ -982,6 +993,7 @@ protected SubProtocolConfiguration createSubProtocolConfiguration(
    * @param protocolContext the protocol context
    * @param transactionPool the transaction pool
    * @param miningConfiguration the mining parameters
+   * @param peerTaskExecutor the peer task executor
    * @param syncState the sync state
    * @param ethProtocolManager the eth protocol manager
    * @return the mining coordinator
@@ -991,6 +1003,7 @@ protected abstract MiningCoordinator createMiningCoordinator(
       ProtocolContext protocolContext,
       TransactionPool transactionPool,
       MiningConfiguration miningConfiguration,
+      PeerTaskExecutor peerTaskExecutor,
       SyncState syncState,
       EthProtocolManager ethProtocolManager);
 
@@ -1162,29 +1175,42 @@ private ChainDataPruner createChainPruner(final BlockchainStorage blockchainStor
    * Create peer validators list.
    *
    * @param protocolSchedule the protocol schedule
+   * @param peerTaskExecutor the peer task executor
    * @return the list
    */
-  protected List<PeerValidator> createPeerValidators(final ProtocolSchedule protocolSchedule) {
+  protected List<PeerValidator> createPeerValidators(
+      final ProtocolSchedule protocolSchedule, final PeerTaskExecutor peerTaskExecutor) {
     final List<PeerValidator> validators = new ArrayList<>();
 
     final OptionalLong daoBlock = genesisConfigOptions.getDaoForkBlock();
     if (daoBlock.isPresent()) {
       // Setup dao validator
       validators.add(
-          new DaoForkPeerValidator(protocolSchedule, metricsSystem, daoBlock.getAsLong()));
+          new DaoForkPeerValidator(
+              protocolSchedule, peerTaskExecutor, syncConfig, metricsSystem, daoBlock.getAsLong()));
     }
 
     final OptionalLong classicBlock = genesisConfigOptions.getClassicForkBlock();
     // setup classic validator
     if (classicBlock.isPresent()) {
       validators.add(
-          new ClassicForkPeerValidator(protocolSchedule, metricsSystem, classicBlock.getAsLong()));
+          new ClassicForkPeerValidator(
+              protocolSchedule,
+              peerTaskExecutor,
+              syncConfig,
+              metricsSystem,
+              classicBlock.getAsLong()));
     }
 
     for (final Map.Entry<Long, Hash> requiredBlock : requiredBlocks.entrySet()) {
       validators.add(
           new RequiredBlocksPeerValidator(
-              protocolSchedule, metricsSystem, requiredBlock.getKey(), requiredBlock.getValue()));
+              protocolSchedule,
+              peerTaskExecutor,
+              syncConfig,
+              metricsSystem,
+              requiredBlock.getKey(),
+              requiredBlock.getValue()));
     }
 
     final CheckpointConfigOptions checkpointConfigOptions =
@@ -1193,6 +1219,8 @@ protected List<PeerValidator> createPeerValidators(final ProtocolSchedule protoc
       validators.add(
           new CheckpointBlocksPeerValidator(
               protocolSchedule,
+              peerTaskExecutor,
+              syncConfig,
               metricsSystem,
               checkpointConfigOptions.getNumber().orElseThrow(),
               checkpointConfigOptions.getHash().map(Hash::fromHexString).orElseThrow()));

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

@@ -39,6 +39,7 @@
 import org.hyperledger.besu.ethereum.core.MiningConfiguration;
 import org.hyperledger.besu.ethereum.core.Util;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutor;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
@@ -84,6 +85,7 @@ protected MiningCoordinator createMiningCoordinator(
       final ProtocolContext protocolContext,
       final TransactionPool transactionPool,
       final MiningConfiguration miningConfiguration,
+      final PeerTaskExecutor peerTaskExecutor,
       final SyncState syncState,
       final EthProtocolManager ethProtocolManager) {
     final CliqueMinerExecutor miningExecutor =

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.PeerTaskExecutor;
 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;
@@ -129,6 +130,7 @@ protected MiningCoordinator createMiningCoordinator(
       final ProtocolContext protocolContext,
       final TransactionPool transactionPool,
       final MiningConfiguration miningConfiguration,
+      final PeerTaskExecutor peerTaskExecutor,
       final SyncState syncState,
       final EthProtocolManager ethProtocolManager) {
 
@@ -144,6 +146,7 @@ protected MiningCoordinator createMiningCoordinator(
                                 protocolContext,
                                 transactionPool,
                                 miningConfiguration,
+                                peerTaskExecutor,
                                 syncState,
                                 ethProtocolManager)))
             .collect(Collectors.toList());

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

@@ -65,6 +65,7 @@
 import org.hyperledger.besu.ethereum.eth.EthProtocol;
 import org.hyperledger.besu.ethereum.eth.SnapProtocol;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutor;
 import org.hyperledger.besu.ethereum.eth.manager.snap.SnapProtocolManager;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
@@ -142,6 +143,7 @@ protected MiningCoordinator createMiningCoordinator(
       final ProtocolContext protocolContext,
       final TransactionPool transactionPool,
       final MiningConfiguration miningConfiguration,
+      final PeerTaskExecutor peerTaskExecutor,
       final SyncState syncState,
       final EthProtocolManager ethProtocolManager) {
     final MutableBlockchain blockchain = protocolContext.getBlockchain();

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

@@ -23,6 +23,7 @@
 import org.hyperledger.besu.ethereum.chain.Blockchain;
 import org.hyperledger.besu.ethereum.core.MiningConfiguration;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutor;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.EpochCalculator;
@@ -47,6 +48,7 @@ protected MiningCoordinator createMiningCoordinator(
       final ProtocolContext protocolContext,
       final TransactionPool transactionPool,
       final MiningConfiguration miningConfiguration,
+      final PeerTaskExecutor peerTaskExecutor,
       final SyncState syncState,
       final EthProtocolManager ethProtocolManager) {
 

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.PeerTaskExecutor;
 import org.hyperledger.besu.ethereum.eth.peervalidation.PeerValidator;
 import org.hyperledger.besu.ethereum.eth.peervalidation.RequiredBlocksPeerValidator;
 import org.hyperledger.besu.ethereum.eth.sync.SynchronizerConfiguration;
@@ -68,6 +69,7 @@ protected MiningCoordinator createMiningCoordinator(
       final ProtocolContext protocolContext,
       final TransactionPool transactionPool,
       final MiningConfiguration miningConfiguration,
+      final PeerTaskExecutor peerTaskExecutor,
       final SyncState syncState,
       final EthProtocolManager ethProtocolManager) {
     return createTransitionMiningCoordinator(
@@ -79,6 +81,8 @@ protected MiningCoordinator createMiningCoordinator(
         new BackwardSyncContext(
             protocolContext,
             protocolSchedule,
+            peerTaskExecutor,
+            syncConfig,
             metricsSystem,
             ethProtocolManager.ethContext(),
             syncState,
@@ -235,14 +239,17 @@ protected PluginServiceFactory createAdditionalPluginServices(
   }
 
   @Override
-  protected List<PeerValidator> createPeerValidators(final ProtocolSchedule protocolSchedule) {
-    List<PeerValidator> retval = super.createPeerValidators(protocolSchedule);
+  protected List<PeerValidator> createPeerValidators(
+      final ProtocolSchedule protocolSchedule, final PeerTaskExecutor peerTaskExecutor) {
+    List<PeerValidator> retval = super.createPeerValidators(protocolSchedule, peerTaskExecutor);
     final OptionalLong powTerminalBlockNumber = genesisConfigOptions.getTerminalBlockNumber();
     final Optional<Hash> powTerminalBlockHash = genesisConfigOptions.getTerminalBlockHash();
     if (powTerminalBlockHash.isPresent() && powTerminalBlockNumber.isPresent()) {
       retval.add(
           new RequiredBlocksPeerValidator(
               protocolSchedule,
+              peerTaskExecutor,
+              syncConfig,
               metricsSystem,
               powTerminalBlockNumber.getAsLong(),
               powTerminalBlockHash.get(),

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

@@ -74,6 +74,7 @@
 import org.hyperledger.besu.ethereum.eth.EthProtocol;
 import org.hyperledger.besu.ethereum.eth.SnapProtocol;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutor;
 import org.hyperledger.besu.ethereum.eth.manager.snap.SnapProtocolManager;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
@@ -182,6 +183,7 @@ protected MiningCoordinator createMiningCoordinator(
       final ProtocolContext protocolContext,
       final TransactionPool transactionPool,
       final MiningConfiguration miningConfiguration,
+      final PeerTaskExecutor peerTaskExecutor,
       final SyncState syncState,
       final EthProtocolManager ethProtocolManager) {
     final MutableBlockchain blockchain = protocolContext.getBlockchain();

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

@@ -103,6 +103,7 @@ protected MiningCoordinator createMiningCoordinator(
       final ProtocolContext protocolContext,
       final TransactionPool transactionPool,
       final MiningConfiguration miningConfiguration,
+      final PeerTaskExecutor peerTaskExecutor,
       final SyncState syncState,
       final EthProtocolManager ethProtocolManager) {
 
@@ -120,6 +121,8 @@ protected MiningCoordinator createMiningCoordinator(
         new TransitionBackwardSyncContext(
             protocolContext,
             transitionProtocolSchedule,
+            peerTaskExecutor,
+            syncConfig,
             metricsSystem,
             ethProtocolManager.ethContext(),
             syncState,
@@ -138,6 +141,7 @@ protected MiningCoordinator createMiningCoordinator(
                             .isMiningEnabled(false)
                             .build())
                     .build(),
+                peerTaskExecutor,
                 syncState,
                 ethProtocolManager),
             mergeBesuControllerBuilder.createTransitionMiningCoordinator(

besu/src/test/java/org/hyperledger/besu/controller/ConsensusScheduleBesuControllerBuilderTest.java

@@ -34,6 +34,7 @@
 import org.hyperledger.besu.ethereum.chain.MutableBlockchain;
 import org.hyperledger.besu.ethereum.core.MiningConfiguration;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutor;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
@@ -121,9 +122,11 @@ public void createsMigratingMiningCoordinator() {
     final Map<Long, BesuControllerBuilder> consensusSchedule =
         Map.of(0L, besuControllerBuilder1, 5L, besuControllerBuilder2);
 
-    when(besuControllerBuilder1.createMiningCoordinator(any(), any(), any(), any(), any(), any()))
+    when(besuControllerBuilder1.createMiningCoordinator(
+            any(), any(), any(), any(), any(), any(), any()))
         .thenReturn(miningCoordinator1);
-    when(besuControllerBuilder2.createMiningCoordinator(any(), any(), any(), any(), any(), any()))
+    when(besuControllerBuilder2.createMiningCoordinator(
+            any(), any(), any(), any(), any(), any(), any()))
         .thenReturn(miningCoordinator2);
     final ProtocolContext mockProtocolContext = mock(ProtocolContext.class);
     when(mockProtocolContext.getBlockchain()).thenReturn(mock(MutableBlockchain.class));
@@ -136,6 +139,7 @@ public void createsMigratingMiningCoordinator() {
             mockProtocolContext,
             mock(TransactionPool.class),
             mock(MiningConfiguration.class),
+            mock(PeerTaskExecutor.class),
             mock(SyncState.class),
             mock(EthProtocolManager.class));
 

besu/src/test/java/org/hyperledger/besu/controller/TransitionControllerBuilderTest.java

@@ -40,6 +40,7 @@
 import org.hyperledger.besu.ethereum.core.InMemoryKeyValueStorageProvider;
 import org.hyperledger.besu.ethereum.core.MiningConfiguration;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutor;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.mainnet.BlockHeaderValidator;
@@ -73,6 +74,7 @@ public class TransitionControllerBuilderTest {
   @Mock ProtocolContext protocolContext;
   @Mock MutableBlockchain mockBlockchain;
   @Mock TransactionPool transactionPool;
+  @Mock PeerTaskExecutor peerTaskExecutor;
   @Mock SyncState syncState;
 
   @Mock(answer = Answers.RETURNS_DEEP_STUBS)
@@ -274,6 +276,7 @@ TransitionCoordinator buildTransitionCoordinator(
             protocolContext,
             transactionPool,
             miningConfiguration,
+            peerTaskExecutor,
             syncState,
             ethProtocolManager);
 

consensus/merge/src/main/java/org/hyperledger/besu/consensus/merge/TransitionBackwardSyncContext.java

@@ -18,6 +18,8 @@
 import org.hyperledger.besu.ethereum.ProtocolContext;
 import org.hyperledger.besu.ethereum.core.Block;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutor;
+import org.hyperledger.besu.ethereum.eth.sync.SynchronizerConfiguration;
 import org.hyperledger.besu.ethereum.eth.sync.backwardsync.BackwardChain;
 import org.hyperledger.besu.ethereum.eth.sync.backwardsync.BackwardSyncContext;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
@@ -43,13 +45,17 @@ public class TransitionBackwardSyncContext extends BackwardSyncContext {
   public TransitionBackwardSyncContext(
       final ProtocolContext protocolContext,
       final TransitionProtocolSchedule transitionProtocolSchedule,
+      final PeerTaskExecutor peerTaskExecutor,
+      final SynchronizerConfiguration synchronizerConfiguration,
       final MetricsSystem metricsSystem,
       final EthContext ethContext,
       final SyncState syncState,
       final StorageProvider storageProvider) {
     super(
         protocolContext,
         transitionProtocolSchedule,
+        peerTaskExecutor,
+        synchronizerConfiguration,
         metricsSystem,
         ethContext,
         syncState,

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/InvalidPeerTaskResponseException.java

@@ -20,6 +20,10 @@ public InvalidPeerTaskResponseException() {
     super();
   }
 
+  public InvalidPeerTaskResponseException(final String message) {
+    super(message);
+  }
+
   public InvalidPeerTaskResponseException(final Throwable cause) {
     super(cause);
   }