Get bodies peer task

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

@@ -714,6 +714,7 @@ public BesuController build() {
     final DefaultSynchronizer synchronizer =
         createSynchronizer(
             protocolSchedule,
+            currentProtocolSpecSupplier,
             worldStateStorageCoordinator,
             protocolContext,
             ethContext,
@@ -842,6 +843,7 @@ private TrieLogPruner createTrieLogPruner(
    * Create synchronizer synchronizer.
    *
    * @param protocolSchedule the protocol schedule
+   * @param currentProtocolSpecSupplier the protocol spec supplier
    * @param worldStateStorageCoordinator the world state storage
    * @param protocolContext the protocol context
    * @param ethContext the eth context
@@ -853,6 +855,7 @@ private TrieLogPruner createTrieLogPruner(
    */
   protected DefaultSynchronizer createSynchronizer(
       final ProtocolSchedule protocolSchedule,
+      final Supplier<ProtocolSpec> currentProtocolSpecSupplier,
       final WorldStateStorageCoordinator worldStateStorageCoordinator,
       final ProtocolContext protocolContext,
       final EthContext ethContext,
@@ -864,6 +867,7 @@ protected DefaultSynchronizer createSynchronizer(
     return new DefaultSynchronizer(
         syncConfig,
         protocolSchedule,
+        currentProtocolSpecSupplier,
         protocolContext,
         worldStateStorageCoordinator,
         ethProtocolManager.getBlockBroadcaster(),

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

@@ -50,6 +50,7 @@
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.forkid.ForkIdManager;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
+import org.hyperledger.besu.ethereum.mainnet.ProtocolSpec;
 import org.hyperledger.besu.ethereum.storage.StorageProvider;
 import org.hyperledger.besu.ethereum.worldstate.DataStorageConfiguration;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
@@ -65,6 +66,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.Consumer;
+import java.util.function.Supplier;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -220,6 +222,7 @@ protected PluginServiceFactory createAdditionalPluginServices(
   @Override
   protected DefaultSynchronizer createSynchronizer(
       final ProtocolSchedule protocolSchedule,
+      final Supplier<ProtocolSpec> currentProtocolSpecSupplier,
       final WorldStateStorageCoordinator worldStateStorageCoordinator,
       final ProtocolContext protocolContext,
       final EthContext ethContext,
@@ -231,6 +234,7 @@ protected DefaultSynchronizer createSynchronizer(
     DefaultSynchronizer sync =
         super.createSynchronizer(
             protocolSchedule,
+            currentProtocolSpecSupplier,
             worldStateStorageCoordinator,
             protocolContext,
             ethContext,

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

@@ -0,0 +1,139 @@
+/*
+ * 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.eth.manager.peertask.task;
+
+import org.hyperledger.besu.ethereum.core.Block;
+import org.hyperledger.besu.ethereum.core.BlockBody;
+import org.hyperledger.besu.ethereum.core.BlockHeader;
+import org.hyperledger.besu.ethereum.eth.EthProtocol;
+import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.InvalidPeerTaskResponseException;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTask;
+import org.hyperledger.besu.ethereum.eth.messages.BlockBodiesMessage;
+import org.hyperledger.besu.ethereum.eth.messages.GetBlockBodiesMessage;
+import org.hyperledger.besu.ethereum.mainnet.BodyValidation;
+import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
+import org.hyperledger.besu.ethereum.p2p.rlpx.wire.MessageData;
+import org.hyperledger.besu.ethereum.p2p.rlpx.wire.SubProtocol;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Predicate;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class GetBodiesFromPeerTask implements PeerTask<List<Block>> {
+
+  private static final Logger LOG = LoggerFactory.getLogger(GetBodiesFromPeerTask.class);
+
+  private final List<BlockHeader> blockHeaders;
+  private final ProtocolSchedule protocolSchedule;
+
+  private final long requiredBlockchainHeight;
+  private final List<Block> blocks = new ArrayList<>();
+  private final boolean isPoS;
+
+  public GetBodiesFromPeerTask(
+      final List<BlockHeader> blockHeaders, final ProtocolSchedule protocolSchedule) {
+    if (blockHeaders == null || blockHeaders.isEmpty()) {
+      throw new IllegalArgumentException("Block headers must not be empty");
+    }
+
+    this.blockHeaders = blockHeaders;
+    this.protocolSchedule = protocolSchedule;
+
+    this.requiredBlockchainHeight =
+        blockHeaders.stream()
+            .mapToLong(BlockHeader::getNumber)
+            .max()
+            .orElse(BlockHeader.GENESIS_BLOCK_NUMBER);
+    this.isPoS = protocolSchedule.getByBlockHeader(blockHeaders.getLast()).isPoS();
+  }
+
+  @Override
+  public SubProtocol getSubProtocol() {
+    return EthProtocol.get();
+  }
+
+  @Override
+  public MessageData getRequestMessage() {
+    return GetBlockBodiesMessage.create(
+        blockHeaders.stream().map(BlockHeader::getBlockHash).toList());
+  }
+
+  @Override
+  public List<Block> processResponse(final MessageData messageData)
+      throws InvalidPeerTaskResponseException {
+    // Blocks returned by this method are in the same order as the headers, but might not be
+    // complete
+    if (messageData == null) {
+      throw new InvalidPeerTaskResponseException();
+    }
+    final BlockBodiesMessage blocksMessage = BlockBodiesMessage.readFrom(messageData);
+    final List<BlockBody> blockBodies = blocksMessage.bodies(protocolSchedule);
+    if (blockBodies.isEmpty() || blockBodies.size() > blockHeaders.size()) {
+      throw new InvalidPeerTaskResponseException();
+    }
+
+    for (int i = 0; i < blockBodies.size(); i++) {
+      final BlockBody blockBody = blockBodies.get(i);
+      final BlockHeader blockHeader = blockHeaders.get(i);
+      if (!blockBodyMatchesBlockHeader(blockBody, blockHeader)) {
+        LOG.atDebug().setMessage("Received block body does not match block header").log();
+        throw new InvalidPeerTaskResponseException();
+      }
+
+      blocks.add(new Block(blockHeader, blockBody));
+    }
+    return blocks;
+  }
+
+  private boolean blockBodyMatchesBlockHeader(
+      final BlockBody blockBody, final BlockHeader blockHeader) {
+    // this method validates that the block body matches the block header by calculating the roots
+    // of the block body and comparing them to the roots in the block header
+    if (!BodyValidation.transactionsRoot(blockBody.getTransactions())
+        .equals(blockHeader.getTransactionsRoot())) {
+      return false;
+    }
+    if (!BodyValidation.ommersHash(blockBody.getOmmers()).equals(blockHeader.getOmmersHash())) {
+      return false;
+    }
+    if (blockBody.getWithdrawals().isPresent()) {
+      if (blockHeader.getWithdrawalsRoot().isEmpty()) {
+        return false;
+      }
+      if (!BodyValidation.withdrawalsRoot(blockBody.getWithdrawals().get())
+          .equals(blockHeader.getWithdrawalsRoot().get())) {
+        return false;
+      }
+    } else if (blockHeader.getWithdrawalsRoot().isPresent()) {
+      return false;
+    }
+    return true;
+  }
+
+  @Override
+  public Predicate<EthPeer> getPeerRequirementFilter() {
+    return (ethPeer) ->
+        isPoS || ethPeer.chainState().getEstimatedHeight() >= requiredBlockchainHeight;
+  }
+
+  @Override
+  public boolean isSuccess(final List<Block> result) {
+    return !result.isEmpty();
+  }
+}

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/DefaultSynchronizer.java

@@ -35,6 +35,7 @@
 import org.hyperledger.besu.ethereum.eth.sync.state.PendingBlocksManager;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
+import org.hyperledger.besu.ethereum.mainnet.ProtocolSpec;
 import org.hyperledger.besu.ethereum.storage.StorageProvider;
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.BonsaiWorldStateProvider;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateStorageCoordinator;
@@ -79,6 +80,7 @@ public class DefaultSynchronizer implements Synchronizer, UnverifiedForkchoiceLi
   public DefaultSynchronizer(
       final SynchronizerConfiguration syncConfig,
       final ProtocolSchedule protocolSchedule,
+      final Supplier<ProtocolSpec> currentProtocolSpecSupplier,
       final ProtocolContext protocolContext,
       final WorldStateStorageCoordinator worldStateStorageCoordinator,
       final BlockBroadcaster blockBroadcaster,
@@ -136,6 +138,7 @@ public DefaultSynchronizer(
                     syncState,
                     metricsSystem,
                     terminationCondition,
+                    peerTaskExecutor,
                     syncDurationMetrics));
 
     if (SyncMode.FAST.equals(syncConfig.getSyncMode())) {
@@ -146,6 +149,7 @@ public DefaultSynchronizer(
                   syncConfig,
                   dataDirectory,
                   protocolSchedule,
+                  currentProtocolSpecSupplier,
                   protocolContext,
                   metricsSystem,
                   ethContext,
@@ -163,6 +167,7 @@ public DefaultSynchronizer(
                   syncConfig,
                   dataDirectory,
                   protocolSchedule,
+                  currentProtocolSpecSupplier,
                   protocolContext,
                   metricsSystem,
                   ethContext,
@@ -180,6 +185,7 @@ public DefaultSynchronizer(
                   syncConfig,
                   dataDirectory,
                   protocolSchedule,
+                  currentProtocolSpecSupplier,
                   protocolContext,
                   metricsSystem,
                   ethContext,

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/DownloadBodiesStep.java

@@ -17,7 +17,9 @@
 import org.hyperledger.besu.ethereum.core.Block;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutor;
 import org.hyperledger.besu.ethereum.eth.sync.tasks.CompleteBlocksTask;
+import org.hyperledger.besu.ethereum.eth.sync.tasks.CompleteBlocksWithPeerTask;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
 
@@ -31,19 +33,41 @@ public class DownloadBodiesStep
   private final ProtocolSchedule protocolSchedule;
   private final EthContext ethContext;
   private final MetricsSystem metricsSystem;
+  private final SynchronizerConfiguration synchronizerConfiguration;
+  private final PeerTaskExecutor peerTaskExecutor;
 
   public DownloadBodiesStep(
       final ProtocolSchedule protocolSchedule,
       final EthContext ethContext,
+      final PeerTaskExecutor peerTaskExecutor,
+      final SynchronizerConfiguration synchronizerConfiguration,
       final MetricsSystem metricsSystem) {
     this.protocolSchedule = protocolSchedule;
     this.ethContext = ethContext;
+    this.peerTaskExecutor = peerTaskExecutor;
+    this.synchronizerConfiguration = synchronizerConfiguration;
     this.metricsSystem = metricsSystem;
   }
 
   @Override
   public CompletableFuture<List<Block>> apply(final List<BlockHeader> blockHeaders) {
-    return CompleteBlocksTask.forHeaders(protocolSchedule, ethContext, blockHeaders, metricsSystem)
-        .run();
+    if (synchronizerConfiguration.isPeerTaskSystemEnabled()) {
+      return ethContext
+          .getScheduler()
+          .scheduleServiceTask(() -> getBodiesWithPeerTaskSystem(blockHeaders));
+    } else {
+      return CompleteBlocksTask.forHeaders(
+              protocolSchedule, ethContext, blockHeaders, metricsSystem)
+          .run();
+    }
+  }
+
+  private CompletableFuture<List<Block>> getBodiesWithPeerTaskSystem(
+      final List<BlockHeader> headers) {
+
+    final CompleteBlocksWithPeerTask completeBlocksWithPeerTask =
+        new CompleteBlocksWithPeerTask(protocolSchedule, headers, peerTaskExecutor);
+    final List<Block> blocks = completeBlocksWithPeerTask.getBlocks();
+    return CompletableFuture.completedFuture(blocks);
   }
 }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/checkpointsync/CheckpointDownloaderFactory.java

@@ -34,6 +34,7 @@
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
 import org.hyperledger.besu.ethereum.eth.sync.worldstate.WorldStateDownloader;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
+import org.hyperledger.besu.ethereum.mainnet.ProtocolSpec;
 import org.hyperledger.besu.ethereum.mainnet.ScheduleBasedBlockHeaderFunctions;
 import org.hyperledger.besu.ethereum.trie.CompactEncoding;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateStorageCoordinator;
@@ -44,6 +45,7 @@
 import java.nio.file.Path;
 import java.time.Clock;
 import java.util.Optional;
+import java.util.function.Supplier;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -59,6 +61,7 @@ public static Optional<FastSyncDownloader<?>> createCheckpointDownloader(
       final SynchronizerConfiguration syncConfig,
       final Path dataDirectory,
       final ProtocolSchedule protocolSchedule,
+      final Supplier<ProtocolSpec> currentProtocolSpecSupplier,
       final ProtocolContext protocolContext,
       final MetricsSystem metricsSystem,
       final EthContext ethContext,
@@ -110,6 +113,7 @@ public static Optional<FastSyncDownloader<?>> createCheckpointDownloader(
               syncConfig,
               worldStateStorageCoordinator,
               protocolSchedule,
+              currentProtocolSpecSupplier,
               protocolContext,
               ethContext,
               peerTaskExecutor,
@@ -128,6 +132,7 @@ public static Optional<FastSyncDownloader<?>> createCheckpointDownloader(
               syncConfig,
               worldStateStorageCoordinator,
               protocolSchedule,
+              currentProtocolSpecSupplier,
               protocolContext,
               ethContext,
               peerTaskExecutor,

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/checkpointsync/CheckpointSyncActions.java

@@ -24,15 +24,19 @@
 import org.hyperledger.besu.ethereum.eth.sync.fastsync.FastSyncState;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
+import org.hyperledger.besu.ethereum.mainnet.ProtocolSpec;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateStorageCoordinator;
 import org.hyperledger.besu.metrics.SyncDurationMetrics;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
 
+import java.util.function.Supplier;
+
 public class CheckpointSyncActions extends FastSyncActions {
   public CheckpointSyncActions(
       final SynchronizerConfiguration syncConfig,
       final WorldStateStorageCoordinator worldStateStorageCoordinator,
       final ProtocolSchedule protocolSchedule,
+      final Supplier<ProtocolSpec> currentProtocolSpecSupplier,
       final ProtocolContext protocolContext,
       final EthContext ethContext,
       final PeerTaskExecutor peerTaskExecutor,
@@ -43,6 +47,7 @@ public CheckpointSyncActions(
         syncConfig,
         worldStateStorageCoordinator,
         protocolSchedule,
+        currentProtocolSpecSupplier,
         protocolContext,
         ethContext,
         peerTaskExecutor,