7311: Refactor peer task system usage in DownloadReceiptsStep to bett…

…er match old system Signed-off-by: Matilda Clerke <[email protected]>
hyperledger · Oct 7, 2024 · 1a30174 · 1a30174
1 parent 493ac91
commit 1a30174
Showing 1 changed file with 31 additions and 34 deletions.
diff --git a/...h/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/DownloadReceiptsStep.java b/...h/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/DownloadReceiptsStep.java
@@ -37,9 +37,6 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 public class DownloadReceiptsStep
     implements Function<List<Block>, CompletableFuture<List<BlockWithReceipts>>> {
 
@@ -65,37 +62,8 @@ public CompletableFuture<List<BlockWithReceipts>> apply(final List<Block> blocks
     if (synchronizerConfiguration.isPeerTaskSystemEnabled()) {
       return ethContext
           .getScheduler()
-          .scheduleServiceTask(
-              () -> {
-                Map<BlockHeader, List<TransactionReceipt>> getReceipts = new HashMap<>();
-                do {
-                  GetReceiptsFromPeerTask task =
-                      new GetReceiptsFromPeerTask(headers, new BodyValidator());
-                  PeerTaskExecutorResult<Map<BlockHeader, List<TransactionReceipt>>>
-                      getReceiptsResult = peerTaskExecutor.execute(task);
-                  if (getReceiptsResult.responseCode() == PeerTaskExecutorResponseCode.SUCCESS
-                      && getReceiptsResult.result().isPresent()) {
-                    Map<BlockHeader, List<TransactionReceipt>> taskResult =
-                        getReceiptsResult.result().get();
-                    taskResult
-                        .keySet()
-                        .forEach(
-                            (blockHeader) ->
-                                getReceipts.merge(
-                                    blockHeader,
-                                    taskResult.get(blockHeader),
-                                    (initialReceipts, newReceipts) -> {
-                                      throw new IllegalStateException(
-                                          "Unexpectedly got receipts for block header already populated!");
-                                    }));
-                  }
-                  // remove all the headers we found receipts for
-                  headers.removeAll(getReceipts.keySet());
-                  // repeat until all headers have receipts
-                } while (!headers.isEmpty());
-                return CompletableFuture.completedFuture(
-                    combineBlocksAndReceipts(blocks, getReceipts));
-              });
+          .scheduleServiceTask(() -> getReceiptsWithPeerTaskSystem(headers))
+          .thenApply((receipts) -> combineBlocksAndReceipts(blocks, receipts));
 
     } else {
       return GetReceiptsForHeadersTask.forHeaders(ethContext, headers, metricsSystem)
@@ -104,6 +72,35 @@ public CompletableFuture<List<BlockWithReceipts>> apply(final List<Block> blocks
     }
   }
 
+  private CompletableFuture<Map<BlockHeader, List<TransactionReceipt>>>
+      getReceiptsWithPeerTaskSystem(final List<BlockHeader> headers) {
+    Map<BlockHeader, List<TransactionReceipt>> getReceipts = new HashMap<>();
+    do {
+      GetReceiptsFromPeerTask task = new GetReceiptsFromPeerTask(headers, new BodyValidator());
+      PeerTaskExecutorResult<Map<BlockHeader, List<TransactionReceipt>>> getReceiptsResult =
+          peerTaskExecutor.execute(task);
+      if (getReceiptsResult.responseCode() == PeerTaskExecutorResponseCode.SUCCESS
+          && getReceiptsResult.result().isPresent()) {
+        Map<BlockHeader, List<TransactionReceipt>> taskResult = getReceiptsResult.result().get();
+        taskResult
+            .keySet()
+            .forEach(
+                (blockHeader) ->
+                    getReceipts.merge(
+                        blockHeader,
+                        taskResult.get(blockHeader),
+                        (initialReceipts, newReceipts) -> {
+                          throw new IllegalStateException(
+                              "Unexpectedly got receipts for block header already populated!");
+                        }));
+      }
+      // remove all the headers we found receipts for
+      headers.removeAll(getReceipts.keySet());
+      // repeat until all headers have receipts
+    } while (!headers.isEmpty());
+    return CompletableFuture.completedFuture(getReceipts);
+  }
+
   private List<BlockWithReceipts> combineBlocksAndReceipts(
       final List<Block> blocks, final Map<BlockHeader, List<TransactionReceipt>> receiptsByHeader) {
     return blocks.stream()