diff --git a/app/src/main/java/org/hyperledger/besu/cli/options/SynchronizerOptions.java b/app/src/main/java/org/hyperledger/besu/cli/options/SynchronizerOptions.java
index df1623cf3c3..607b8304097 100644
--- a/app/src/main/java/org/hyperledger/besu/cli/options/SynchronizerOptions.java
+++ b/app/src/main/java/org/hyperledger/besu/cli/options/SynchronizerOptions.java
@@ -68,6 +68,8 @@ public class SynchronizerOptions implements CLIOptions<SynchronizerConfiguration
       "--Xsynchronizer-world-state-min-millis-before-stalling";
   private static final String WORLD_STATE_TASK_CACHE_SIZE_FLAG =
       "--Xsynchronizer-world-state-task-cache-size";
+  private static final String RECEIPTS_DOWNLOAD_STEP_TIMEOUT_MILLIS_FLAG =
+      "--Xsynchronizer-receipts-download-step-timeout-millis";
 
   // Regular (stable) flag
   private static final String SNAP_SERVER_ENABLED_FLAG = "--snapsync-server-enabled";
@@ -258,6 +260,15 @@ public void parseBlockPropagationRange(final String arg) {
   private int worldStateTaskCacheSize =
       SynchronizerConfiguration.DEFAULT_WORLD_STATE_TASK_CACHE_SIZE;
 
+  @CommandLine.Option(
+      names = RECEIPTS_DOWNLOAD_STEP_TIMEOUT_MILLIS_FLAG,
+      hidden = true,
+      paramLabel = "<LONG>",
+      description =
+          "Maximum time in milliseconds to wait for receipts download step including all retries (default: ${DEFAULT-VALUE})")
+  private long receiptsDownloadStepTimeoutMillis =
+      SynchronizerConfiguration.DEFAULT_RECEIPTS_DOWNLOAD_STEP_TIMEOUT_MILLIS;
+
   @CommandLine.Option(
       names = SNAP_PIVOT_BLOCK_WINDOW_VALIDITY_FLAG,
       hidden = true,
@@ -509,6 +520,7 @@ public SynchronizerConfiguration.Builder toDomainObject() {
     builder.isPeerTaskSystemEnabled(isPeerTaskSystemEnabled);
     builder.snapSyncSavePreCheckpointHeadersOnlyEnabled(
         snapSyncSavePreCheckpointHeadersOnlyEnabled);
+    builder.receiptsDownloadStepTimeoutMillis(receiptsDownloadStepTimeoutMillis);
     builder.era1ImportPrepipelineEnabled(era1ImportPrepipelineEnabled);
     builder.era1DataUri(era1DataUri);
     builder.era1ImportPrepipelineConcurrency(era1ImportPrepipelineConcurrency);
@@ -553,6 +565,8 @@ public List<String> getCLIOptions() {
             OptionParser.format(worldStateMinMillisBeforeStalling),
             WORLD_STATE_TASK_CACHE_SIZE_FLAG,
             OptionParser.format(worldStateTaskCacheSize),
+            RECEIPTS_DOWNLOAD_STEP_TIMEOUT_MILLIS_FLAG,
+            OptionParser.format(receiptsDownloadStepTimeoutMillis),
             SNAP_PIVOT_BLOCK_WINDOW_VALIDITY_FLAG,
             OptionParser.format(snapsyncPivotBlockWindowValidity),
             SNAP_PIVOT_BLOCK_DISTANCE_BEFORE_CACHING_FLAG,
diff --git a/ethereum/core/src/test-support/java/org/hyperledger/besu/ethereum/core/BlockDataGenerator.java b/ethereum/core/src/test-support/java/org/hyperledger/besu/ethereum/core/BlockDataGenerator.java
index 11040833d08..25b1f2d5f2b 100644
--- a/ethereum/core/src/test-support/java/org/hyperledger/besu/ethereum/core/BlockDataGenerator.java
+++ b/ethereum/core/src/test-support/java/org/hyperledger/besu/ethereum/core/BlockDataGenerator.java
@@ -373,7 +373,7 @@ public Block block(final BlockOptions options) {
   }
 
   public Block block() {
-    return block(new BlockOptions());
+    return block(blockOptionsSupplier.get());
   }
 
   /**
@@ -477,10 +477,11 @@ public BlockBody body(final BlockOptions options) {
         ommers.add(ommer());
       }
     }
-    final List<Transaction> defaultTxs = new ArrayList<>();
+    final List<Transaction> defaultTxs = new ArrayList<>(options.transactionCount());
     if (options.hasTransactions()) {
-      defaultTxs.add(transaction(options.getTransactionTypes()));
-      defaultTxs.add(transaction(options.getTransactionTypes()));
+      for (int i = 0; i < options.transactionCount(); i++) {
+        defaultTxs.add(transaction(options.getTransactionTypes()));
+      }
     }
 
     return new BlockBody(
@@ -901,6 +902,7 @@ public static class BlockOptions {
     private Optional<Long> timestamp = Optional.empty();
     private boolean hasOmmers = true;
     private boolean hasTransactions = true;
+    private int transactionCount = 2;
     private TransactionType[] transactionTypes = {
       TransactionType.FRONTIER, TransactionType.ACCESS_LIST, TransactionType.EIP1559
     };
@@ -997,6 +999,10 @@ public boolean hasTransactions() {
       return hasTransactions;
     }
 
+    public int transactionCount() {
+      return transactionCount;
+    }
+
     public boolean hasOmmers() {
       return hasOmmers;
     }
@@ -1096,6 +1102,11 @@ public BlockOptions hasTransactions(final boolean hasTransactions) {
       return this;
     }
 
+    public BlockOptions transactionCount(final int transactionCount) {
+      this.transactionCount = transactionCount;
+      return this;
+    }
+
     public BlockOptions transactionTypes(final TransactionType... transactionTypes) {
       this.transactionTypes = transactionTypes;
       return this;
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutor.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutor.java
index 986af42df5c..af51135f3c3 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutor.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutor.java
@@ -141,7 +141,7 @@ public <T> PeerTaskExecutorResult<T> executeAgainstPeer(
               requestSender.sendRequest(peerTaskSubProtocol, requestMessageData, peer);
 
           if (responseMessageData == null) {
-            throw new InvalidPeerTaskResponseException();
+            throw new InvalidPeerTaskResponseException("Null response");
           }
 
           result = peerTask.processResponse(responseMessageData);
@@ -158,7 +158,10 @@ public <T> PeerTaskExecutorResult<T> executeAgainstPeer(
           peerTask.postProcessResult(executorResult);
         } else {
           LOG.debug(
-              "Invalid response found for {} from peer {}", taskClassName, peer.getLoggableId());
+              "Invalid response {} found for {} from peer {}",
+              validationResponse,
+              taskClassName,
+              peer.getLoggableId());
           if (validationResponse.recordUselessResponse()) {
             peer.recordUselessResponse(taskClassName);
           }
@@ -212,11 +215,13 @@ public <T> PeerTaskExecutorResult<T> executeAgainstPeer(
                 PeerTaskExecutorResponseCode.INTERNAL_SERVER_ERROR,
                 List.of(peer));
       }
-      LOG.debug(
-          "Executed peer task against {}, response code {}, retries remaining {}",
-          peer.getLoggableId(),
-          executorResult.responseCode(),
-          retriesRemaining);
+      LOG.atDebug()
+          .setMessage("Executed peer task {} against {}, response code {}, retries remaining {}")
+          .addArgument(taskClassName)
+          .addArgument(peer::getLoggableId)
+          .addArgument(executorResult::responseCode)
+          .addArgument(retriesRemaining)
+          .log();
     } while (retriesRemaining-- > 0
         && executorResult.responseCode() != PeerTaskExecutorResponseCode.SUCCESS
         && executorResult.responseCode() != PeerTaskExecutorResponseCode.PEER_DISCONNECTED
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/task/GetSyncReceiptsFromPeerTask.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/task/GetSyncReceiptsFromPeerTask.java
index c8f7b9e4914..62cc2971e47 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/task/GetSyncReceiptsFromPeerTask.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/task/GetSyncReceiptsFromPeerTask.java
@@ -14,8 +14,11 @@
  */
 package org.hyperledger.besu.ethereum.eth.manager.peertask.task;
 
+import static com.google.common.base.Preconditions.checkArgument;
+
 import org.hyperledger.besu.datatypes.Hash;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
+import org.hyperledger.besu.ethereum.core.SyncBlock;
 import org.hyperledger.besu.ethereum.core.SyncTransactionReceipt;
 import org.hyperledger.besu.ethereum.core.Util;
 import org.hyperledger.besu.ethereum.core.encoding.receipt.SyncTransactionReceiptEncoder;
@@ -32,51 +35,35 @@
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.SubProtocol;
 import org.hyperledger.besu.ethereum.rlp.RLPException;
 
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Predicate;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.tuweni.bytes.Bytes;
 
 public class GetSyncReceiptsFromPeerTask
-    implements PeerTask<Map<BlockHeader, List<SyncTransactionReceipt>>> {
+    implements PeerTask<Map<SyncBlock, List<SyncTransactionReceipt>>> {
 
-  private final Collection<BlockHeader> blockHeaders;
-  private final Map<BlockHeader, List<SyncTransactionReceipt>> receiptsByBlockHeader =
-      new HashMap<>();
-  private final Map<Hash, List<BlockHeader>> headersByReceiptsRoot = new HashMap<>();
+  private final List<SyncBlock> requestedBlocks;
+  private final List<BlockHeader> requestedHeaders;
   private final long requiredBlockchainHeight;
   private final boolean isPoS;
   private final SyncTransactionReceiptEncoder syncTransactionReceiptEncoder;
 
   public GetSyncReceiptsFromPeerTask(
-      final Collection<BlockHeader> blockHeaders,
+      final List<SyncBlock> blocks,
       final ProtocolSchedule protocolSchedule,
       final SyncTransactionReceiptEncoder syncTransactionReceiptEncoder) {
-    this.blockHeaders = new ArrayList<>(blockHeaders);
-
-    // pre-fill any headers with an empty receipts root into the result map
-    this.blockHeaders.stream()
-        .filter(header -> header.getReceiptsRoot().equals(Hash.EMPTY_TRIE_HASH))
-        .forEach(header -> receiptsByBlockHeader.put(header, Collections.emptyList()));
-    this.blockHeaders.removeAll(receiptsByBlockHeader.keySet());
-
-    // group headers by their receipts root hash to reduce total number of receipts hashes requested
-    // for
-    this.blockHeaders.forEach(
-        header ->
-            headersByReceiptsRoot
-                .computeIfAbsent(header.getReceiptsRoot(), key -> new ArrayList<>())
-                .add(header));
+    checkArgument(!blocks.isEmpty(), "Requested block list must not be empty");
+    this.requestedBlocks = blocks;
+    this.requestedHeaders = blocks.stream().map(SyncBlock::getHeader).toList();
 
     // calculate the minimum required blockchain height a peer will need to be able to fulfil this
     // request
     requiredBlockchainHeight =
-        this.blockHeaders.stream()
+        requestedHeaders.stream()
             .mapToLong(BlockHeader::getNumber)
             .max()
             .orElse(BlockHeader.GENESIS_BLOCK_NUMBER);
@@ -92,62 +79,31 @@ public SubProtocol getSubProtocol() {
 
   @Override
   public MessageData getRequestMessage() {
-    // Since we have to match up the data by receipt root, we only need to request receipts
-    // for one of the headers with each unique receipt root.
-    final List<Hash> blockHashes =
-        headersByReceiptsRoot.values().stream()
-            .map(headers -> headers.getFirst().getHash())
-            .toList();
-    return GetReceiptsMessage.create(blockHashes);
+    return GetReceiptsMessage.create(requestedHeaders.stream().map(BlockHeader::getHash).toList());
   }
 
   @Override
-  public Map<BlockHeader, List<SyncTransactionReceipt>> processResponse(
-      final MessageData messageData)
+  public Map<SyncBlock, List<SyncTransactionReceipt>> processResponse(final MessageData messageData)
       throws InvalidPeerTaskResponseException, MalformedRlpFromPeerException {
     if (messageData == null) {
-      throw new InvalidPeerTaskResponseException();
+      throw new InvalidPeerTaskResponseException("Null message data");
     }
     final ReceiptsMessage receiptsMessage = ReceiptsMessage.readFrom(messageData);
-    final List<List<SyncTransactionReceipt>> receiptsByBlock;
     try {
-      receiptsByBlock = receiptsMessage.syncReceipts();
+      final List<List<SyncTransactionReceipt>> receivedBlocks = receiptsMessage.syncReceipts();
+      if (receivedBlocks.size() > requestedBlocks.size()) {
+        throw new InvalidPeerTaskResponseException("Too many result returned");
+      }
+      final Map<SyncBlock, List<SyncTransactionReceipt>> response =
+          HashMap.newHashMap(receivedBlocks.size());
+      for (int i = 0; i < receivedBlocks.size(); i++) {
+        response.put(requestedBlocks.get(i), receivedBlocks.get(i));
+      }
+      return response;
     } catch (RLPException e) {
       // indicates a malformed or unexpected RLP result from the peer
       throw new MalformedRlpFromPeerException(e, messageData.getData());
     }
-    // take a copy of the pre-filled receiptsByBlockHeader, to ensure idempotency of subsequent
-    // calls to processResponse
-    final Map<BlockHeader, List<SyncTransactionReceipt>> receiptsByHeader =
-        new HashMap<>(receiptsByBlockHeader);
-    if (!blockHeaders.isEmpty()) {
-      if (receiptsByBlock.isEmpty() || receiptsByBlock.size() > blockHeaders.size()) {
-        throw new InvalidPeerTaskResponseException();
-      }
-
-      for (final List<SyncTransactionReceipt> receiptsInBlock : receiptsByBlock) {
-        final List<BlockHeader> blockHeaders =
-            headersByReceiptsRoot.get(
-                Util.getRootFromListOfBytes(
-                    receiptsInBlock.stream()
-                        .map(
-                            (r) -> {
-                              Bytes rlp =
-                                  r.isFormattedForRootCalculation()
-                                      ? r.getRlpBytes()
-                                      : syncTransactionReceiptEncoder.encodeForRootCalculation(r);
-                              r.clearSubVariables();
-                              return rlp;
-                            })
-                        .toList()));
-        if (blockHeaders == null) {
-          // Contains receipts that we didn't request, so mustn't be the response we're looking for.
-          throw new InvalidPeerTaskResponseException();
-        }
-        blockHeaders.forEach(header -> receiptsByHeader.put(header, receiptsInBlock));
-      }
-    }
-    return receiptsByHeader;
   }
 
   @Override
@@ -157,11 +113,50 @@ public Predicate<EthPeerImmutableAttributes> getPeerRequirementFilter() {
 
   @Override
   public PeerTaskValidationResponse validateResult(
-      final Map<BlockHeader, List<SyncTransactionReceipt>> result) {
+      final Map<SyncBlock, List<SyncTransactionReceipt>> result) {
+    if (result.isEmpty()) {
+      return PeerTaskValidationResponse.NO_RESULTS_RETURNED;
+    }
+
+    for (final Map.Entry<SyncBlock, List<SyncTransactionReceipt>> entry : result.entrySet()) {
+      final SyncBlock requestedBlock = entry.getKey();
+      final List<SyncTransactionReceipt> receivedReceiptsForBlock = entry.getValue();
+
+      // verify that the receipts count is within bounds for every received block
+      if (receivedReceiptsForBlock.size() > requestedBlock.getBody().getTransactionCount()) {
+        return PeerTaskValidationResponse.TOO_MANY_RESULTS_RETURNED;
+      }
+
+      // ensure the calculated receipts root matches the one in the requested block header
+      if (!receiptsRootMatches(requestedBlock.getHeader(), receivedReceiptsForBlock)) {
+        return PeerTaskValidationResponse.RESULTS_DO_NOT_MATCH_QUERY;
+      }
+    }
+
     return PeerTaskValidationResponse.RESULTS_VALID_AND_GOOD;
   }
 
-  public Collection<BlockHeader> getBlockHeaders() {
-    return blockHeaders;
+  private boolean receiptsRootMatches(
+      final BlockHeader blockHeader, final List<SyncTransactionReceipt> receipts) {
+    final Hash calculatedReceiptsRoot =
+        Util.getRootFromListOfBytes(
+            receipts.stream()
+                .map(
+                    (r) -> {
+                      Bytes rlp =
+                          r.isFormattedForRootCalculation()
+                              ? r.getRlpBytes()
+                              : syncTransactionReceiptEncoder.encodeForRootCalculation(r);
+                      r.clearSubVariables();
+                      return rlp;
+                    })
+                .toList());
+
+    return calculatedReceiptsRoot.equals(blockHeader.getReceiptsRoot());
+  }
+
+  @VisibleForTesting
+  public List<SyncBlock> getRequestedBlocks() {
+    return requestedBlocks;
   }
 }
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/messages/GetReceiptsMessage.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/messages/GetReceiptsMessage.java
index d9c6d6198f5..a8a0b7e601f 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/messages/GetReceiptsMessage.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/messages/GetReceiptsMessage.java
@@ -22,7 +22,7 @@
 import org.hyperledger.besu.ethereum.rlp.RLPInput;
 
 import java.util.ArrayList;
-import java.util.Collection;
+import java.util.List;
 
 import org.apache.tuweni.bytes.Bytes;
 
@@ -57,10 +57,10 @@ public int getCode() {
     return EthProtocolMessages.GET_RECEIPTS;
   }
 
-  public Iterable<Hash> hashes() {
+  public List<Hash> hashes() {
     final RLPInput input = new BytesValueRLPInput(data, false);
     input.enterList();
-    final Collection<Hash> hashes = new ArrayList<>();
+    final List<Hash> hashes = new ArrayList<>();
     while (!input.isEndOfCurrentList()) {
       hashes.add(Hash.wrap(input.readBytes32()));
     }
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/SynchronizerConfiguration.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/SynchronizerConfiguration.java
index d57a2850337..31ad024949c 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/SynchronizerConfiguration.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/SynchronizerConfiguration.java
@@ -50,6 +50,8 @@ public class SynchronizerConfiguration {
   public static final int DEFAULT_COMPUTATION_PARALLELISM = 2;
   public static final int DEFAULT_WORLD_STATE_TASK_CACHE_SIZE =
       CachingTaskCollection.DEFAULT_CACHE_SIZE;
+  public static final long DEFAULT_RECEIPTS_DOWNLOAD_STEP_TIMEOUT_MILLIS =
+      TimeUnit.SECONDS.toMillis(60);
   public static final Boolean DEFAULT_ERA1_IMPORT_PREPIPELINE_ENABLED = Boolean.FALSE;
   public static final URI DEFAULT_ERA1_DATA_URI = URI.create("https://mainnet.era1.nimbus.team/");
   public static final Integer DEFAULT_ERA1_IMPORT_PREPIPELINE_CONCURRENCY = 1;
@@ -85,6 +87,7 @@ public class SynchronizerConfiguration {
   private final int computationParallelism;
   private final int maxTrailingPeers;
   private final long worldStateMinMillisBeforeStalling;
+  private final long receiptsDownloadStepTimeoutMillis;
   private final boolean isPeerTaskSystemEnabled;
   private final boolean snapSyncSavePreCheckpointHeadersOnlyEnabled;
 
@@ -116,6 +119,7 @@ private SynchronizerConfiguration(
       final int transactionsParallelism,
       final int computationParallelism,
       final int maxTrailingPeers,
+      final long receiptsDownloadStepTimeoutMillis,
       final boolean isPeerTaskSystemEnabled,
       final boolean snapSyncSavePreCheckpointHeadersOnlyEnabled,
       final boolean era1ImportPrepipelineEnabled,
@@ -143,6 +147,7 @@ private SynchronizerConfiguration(
     this.transactionsParallelism = transactionsParallelism;
     this.computationParallelism = computationParallelism;
     this.maxTrailingPeers = maxTrailingPeers;
+    this.receiptsDownloadStepTimeoutMillis = receiptsDownloadStepTimeoutMillis;
     this.isPeerTaskSystemEnabled = isPeerTaskSystemEnabled;
     this.snapSyncSavePreCheckpointHeadersOnlyEnabled = snapSyncSavePreCheckpointHeadersOnlyEnabled;
     this.era1ImportPrepipelineEnabled = era1ImportPrepipelineEnabled;
@@ -273,6 +278,10 @@ public int getMaxTrailingPeers() {
     return maxTrailingPeers;
   }
 
+  public long getReceiptsDownloadStepTimeoutMillis() {
+    return receiptsDownloadStepTimeoutMillis;
+  }
+
   public boolean isPeerTaskSystemEnabled() {
     return isPeerTaskSystemEnabled;
   }
@@ -325,6 +334,7 @@ public static class Builder {
     private boolean era1ImportPrepipelineEnabled = DEFAULT_ERA1_IMPORT_PREPIPELINE_ENABLED;
     private URI era1DataUri = DEFAULT_ERA1_DATA_URI;
     private int era1ImportPrepipelineConcurrency = DEFAULT_ERA1_IMPORT_PREPIPELINE_CONCURRENCY;
+    private long receiptsDownloadStepTimeoutMillis = DEFAULT_RECEIPTS_DOWNLOAD_STEP_TIMEOUT_MILLIS;
 
     public Builder syncPivotDistance(final int distance) {
       syncPivotDistance = distance;
@@ -446,6 +456,11 @@ public Builder maxTrailingPeers(final int maxTailingPeers) {
       return this;
     }
 
+    public Builder receiptsDownloadStepTimeoutMillis(final long receiptsDownloadStepTimeoutMillis) {
+      this.receiptsDownloadStepTimeoutMillis = receiptsDownloadStepTimeoutMillis;
+      return this;
+    }
+
     public Builder isPeerTaskSystemEnabled(final boolean isPeerTaskSystemEnabled) {
       this.isPeerTaskSystemEnabled = isPeerTaskSystemEnabled;
       return this;
@@ -497,6 +512,7 @@ public SynchronizerConfiguration build() {
           transactionsParallelism,
           computationParallelism,
           maxTrailingPeers,
+          receiptsDownloadStepTimeoutMillis,
           isPeerTaskSystemEnabled,
           snapSyncSavePreCheckpointHeadersOnlyEnabled,
           era1ImportPrepipelineEnabled,
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/AbstractDownloadReceiptsStep.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/AbstractDownloadReceiptsStep.java
deleted file mode 100644
index ce22817e644..00000000000
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/AbstractDownloadReceiptsStep.java
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- * Copyright ConsenSys AG.
- *
- * 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.sync.fastsync;
-
-import static java.util.stream.Collectors.toList;
-
-import org.hyperledger.besu.ethereum.core.BlockHeader;
-import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.function.Function;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public abstract class AbstractDownloadReceiptsStep<B, TR, BWR>
-    implements Function<List<B>, CompletableFuture<List<BWR>>> {
-  // B is the type of block being processed (Block, SyncBlock),
-  // TR is the type of transaction receipt (TransactionReceipt, SyncTransactionReceipt),
-  // BWR is the type of block with receipts (BlockWithReceipts, SyncBlockWithReceipts)
-
-  private static final Logger LOG = LoggerFactory.getLogger(AbstractDownloadReceiptsStep.class);
-
-  private final EthScheduler ethScheduler;
-
-  public AbstractDownloadReceiptsStep(final EthScheduler ethScheduler) {
-    this.ethScheduler = ethScheduler;
-  }
-
-  @Override
-  public CompletableFuture<List<BWR>> apply(final List<B> blocks) {
-    final List<BlockHeader> headers = blocks.stream().map(this::getBlockHeader).collect(toList());
-    final List<BlockHeader> originalBlockHeaders =
-        LOG.isTraceEnabled() ? List.copyOf(headers) : null;
-    return ethScheduler
-        .scheduleServiceTask(
-            () -> {
-              Map<BlockHeader, List<TR>> receiptsByBlockHeader = new HashMap<>();
-              while (!headers.isEmpty()) {
-                Map<BlockHeader, List<TR>> receipts = getReceipts(headers);
-                headers.removeAll(receipts.keySet());
-                for (BlockHeader blockHeader : receipts.keySet()) {
-                  receiptsByBlockHeader.put(blockHeader, receipts.get(blockHeader));
-                }
-              }
-              if (LOG.isTraceEnabled()) {
-                for (BlockHeader blockHeader : originalBlockHeaders) {
-                  final List<TR> transactionReceipts = receiptsByBlockHeader.get(blockHeader);
-                  LOG.atTrace()
-                      .setMessage("{} receipts received for header {}")
-                      .addArgument(transactionReceipts == null ? 0 : transactionReceipts.size())
-                      .addArgument(blockHeader.getBlockHash())
-                      .log();
-                }
-              }
-              return CompletableFuture.completedFuture(receiptsByBlockHeader);
-            })
-        .thenApply((receipts) -> combineBlocksAndReceipts(blocks, receipts));
-  }
-
-  abstract BlockHeader getBlockHeader(final B b);
-
-  abstract List<BWR> combineBlocksAndReceipts(
-      final List<B> blocks, final Map<BlockHeader, List<TR>> receiptsByHeader);
-
-  /**
-   * Retrieves transaction receipts for as many of the supplied headers as possible. Repeat calls
-   * may be made after removing headers no longer needing transaction receipts.
-   *
-   * @param headers A list of headers to retrieve transaction receipts for
-   * @return transaction receipts for as many of the supplied headers as possible
-   */
-  abstract Map<BlockHeader, List<TR>> getReceipts(final List<BlockHeader> headers);
-}
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/DownloadSyncReceiptsStep.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/DownloadSyncReceiptsStep.java
index 19a90b2e742..72e79a25ec7 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/DownloadSyncReceiptsStep.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/DownloadSyncReceiptsStep.java
@@ -15,68 +15,212 @@
 package org.hyperledger.besu.ethereum.eth.sync.fastsync;
 
 import static java.util.Collections.emptyList;
+import static org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutorResponseCode.SUCCESS;
 
-import org.hyperledger.besu.ethereum.core.BlockHeader;
+import org.hyperledger.besu.datatypes.Hash;
 import org.hyperledger.besu.ethereum.core.SyncBlock;
 import org.hyperledger.besu.ethereum.core.SyncBlockWithReceipts;
 import org.hyperledger.besu.ethereum.core.SyncTransactionReceipt;
 import org.hyperledger.besu.ethereum.core.encoding.receipt.SyncTransactionReceiptEncoder;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutor;
 import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutorResponseCode;
 import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutorResult;
 import org.hyperledger.besu.ethereum.eth.manager.peertask.task.GetSyncReceiptsFromPeerTask;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
-import org.hyperledger.besu.ethereum.rlp.BytesValueRLPOutput;
 
-import java.util.Collections;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class DownloadSyncReceiptsStep
-    extends AbstractDownloadReceiptsStep<SyncBlock, SyncTransactionReceipt, SyncBlockWithReceipts> {
+    implements Function<List<SyncBlock>, CompletableFuture<List<SyncBlockWithReceipts>>> {
 
   private static final Logger LOG = LoggerFactory.getLogger(DownloadSyncReceiptsStep.class);
+  private static final Duration RETRY_DELAY = Duration.ofSeconds(1);
+  private static final AtomicInteger taskSequence = new AtomicInteger(0);
 
   private final ProtocolSchedule protocolSchedule;
+  private final EthScheduler ethScheduler;
   private final PeerTaskExecutor peerTaskExecutor;
   private final SyncTransactionReceiptEncoder syncTransactionReceiptEncoder;
+  private final Duration timeoutDuration;
 
   public DownloadSyncReceiptsStep(
       final ProtocolSchedule protocolSchedule,
       final EthContext ethContext,
-      final SyncTransactionReceiptEncoder syncTransactionReceiptEncoder) {
-    super(ethContext.getScheduler());
+      final SyncTransactionReceiptEncoder syncTransactionReceiptEncoder,
+      final Duration timeoutDuration) {
     this.protocolSchedule = protocolSchedule;
+    this.ethScheduler = ethContext.getScheduler();
     this.peerTaskExecutor = ethContext.getPeerTaskExecutor();
     this.syncTransactionReceiptEncoder = syncTransactionReceiptEncoder;
+    this.timeoutDuration = timeoutDuration;
   }
 
   @Override
-  BlockHeader getBlockHeader(final SyncBlock syncBlock) {
-    return syncBlock.getHeader();
+  public CompletableFuture<List<SyncBlockWithReceipts>> apply(final List<SyncBlock> blocks) {
+    final int currTaskId = taskSequence.incrementAndGet();
+    final List<SyncBlock> blocksToRequest = prepareRequest(blocks);
+    final Map<Hash, List<SyncTransactionReceipt>> receiptsByRootHash =
+        HashMap.newHashMap(blocksToRequest.size());
+
+    return ethScheduler
+        .scheduleServiceTask(
+            () -> downloadReceipts(currTaskId, 0, blocksToRequest, receiptsByRootHash))
+        .thenApply(receipts -> combineBlocksAndReceipts(blocks, receipts))
+        .orTimeout(timeoutDuration.toMillis(), TimeUnit.MILLISECONDS)
+        .whenComplete(
+            (unused, throwable) -> {
+              if (throwable instanceof TimeoutException) {
+                LOG.trace(
+                    "[{}] Timed out after {} ms while downloading receipts for {} blocks",
+                    currTaskId,
+                    timeoutDuration.toMillis(),
+                    blocks.size());
+              }
+            });
+  }
+
+  private List<SyncBlock> prepareRequest(final List<SyncBlock> blocks) {
+    // filter any headers with an empty receipts root, for which we do not need to request anything
+    final List<SyncBlock> blocksWithTxs =
+        blocks.stream()
+            .filter(
+                sb -> {
+                  if (sb.getHeader()
+                      .getReceiptsRoot()
+                      .getBytes()
+                      .equals(Hash.EMPTY_TRIE_HASH.getBytes())) {
+                    LOG.trace(
+                        "Skipping receipts retrieval for empty block {}",
+                        sb.getHeader().getNumber());
+                    return false;
+                  }
+                  return true;
+                })
+            .collect(Collectors.toCollection(ArrayList::new));
+
+    // dedup by grouping blocks by same receipt root hash
+    final List<SyncBlock> blocksToRequest =
+        blocksWithTxs.stream()
+            .collect(
+                Collectors.groupingBy(
+                    syncBlock -> syncBlock.getHeader().getReceiptsRoot(),
+                    LinkedHashMap::new,
+                    Collectors.toList()))
+            .values()
+            .stream()
+            .map(List::getFirst)
+            .collect(Collectors.toCollection(ArrayList::new));
+
+    LOG.trace("Saving by dedup {}", blocksWithTxs.size() - blocksToRequest.size());
+    return blocksToRequest;
+  }
+
+  private CompletableFuture<Map<Hash, List<SyncTransactionReceipt>>> downloadReceipts(
+      final int currTaskId,
+      final int prevIterations,
+      final List<SyncBlock> blocksToRequest,
+      final Map<Hash, List<SyncTransactionReceipt>> receiptsByRootHash) {
+
+    final int initialBlockCount = blocksToRequest.size();
+    int iteration = prevIterations;
+    while (!blocksToRequest.isEmpty()) {
+      ++iteration;
+
+      LOG.atTrace()
+          .setMessage("[{}:{}] Requesting receipts for {} blocks (initial {}): {}")
+          .addArgument(currTaskId)
+          .addArgument(iteration)
+          .addArgument(blocksToRequest::size)
+          .addArgument(initialBlockCount)
+          .addArgument(() -> formatBlockDetails(blocksToRequest))
+          .log();
+
+      final GetSyncReceiptsFromPeerTask task =
+          new GetSyncReceiptsFromPeerTask(
+              blocksToRequest, protocolSchedule, syncTransactionReceiptEncoder);
+
+      final PeerTaskExecutorResult<Map<SyncBlock, List<SyncTransactionReceipt>>> getReceiptsResult =
+          peerTaskExecutor.execute(task);
+
+      final PeerTaskExecutorResponseCode responseCode = getReceiptsResult.responseCode();
+
+      if (responseCode == SUCCESS) {
+        final Map<SyncBlock, List<SyncTransactionReceipt>> receiptsByBlock =
+            getReceiptsResult
+                .result()
+                .orElseThrow(
+                    () ->
+                        new IllegalStateException(
+                            "Task validation failure, it must flag empty result as failure"));
+
+        LOG.atTrace()
+            .setMessage("[{}:{}] Received response for {} blocks (requested {}, initial {}): {}")
+            .addArgument(currTaskId)
+            .addArgument(iteration)
+            .addArgument(receiptsByBlock::size)
+            .addArgument(blocksToRequest::size)
+            .addArgument(initialBlockCount)
+            .addArgument(() -> formatBlockDetails(receiptsByBlock.keySet()))
+            .log();
+
+        receiptsByBlock.forEach(
+            (syncBlock, syncTransactionReceipts) ->
+                receiptsByRootHash.put(
+                    syncBlock.getHeader().getReceiptsRoot(), syncTransactionReceipts));
+
+        blocksToRequest.removeAll(receiptsByBlock.keySet());
+      } else {
+        LOG.atTrace()
+            .setMessage(
+                "[{}:{}] Failed with {} to retrieve receipts for {} blocks (initial {}): {}")
+            .addArgument(currTaskId)
+            .addArgument(iteration)
+            .addArgument(responseCode)
+            .addArgument(blocksToRequest::size)
+            .addArgument(initialBlockCount)
+            .addArgument(() -> formatBlockDetails(blocksToRequest))
+            .log();
+
+        LOG.trace("[{}:{}] Waiting for {} before retrying", currTaskId, iteration, RETRY_DELAY);
+        final int passIterations = iteration;
+        return ethScheduler.scheduleFutureTask(
+            () ->
+                ethScheduler.scheduleServiceTask(
+                    () ->
+                        downloadReceipts(
+                            currTaskId, passIterations, blocksToRequest, receiptsByRootHash)),
+            RETRY_DELAY);
+      }
+    }
+    return CompletableFuture.completedFuture(receiptsByRootHash);
   }
 
-  @Override
   List<SyncBlockWithReceipts> combineBlocksAndReceipts(
       final List<SyncBlock> blocks,
-      final Map<BlockHeader, List<SyncTransactionReceipt>> receiptsByHeader) {
+      final Map<Hash, List<SyncTransactionReceipt>> receiptsByRootHash) {
     return blocks.stream()
         .map(
             block -> {
               final List<SyncTransactionReceipt> receipts =
-                  receiptsByHeader.getOrDefault(block.getHeader(), emptyList());
+                  receiptsByRootHash.getOrDefault(block.getHeader().getReceiptsRoot(), emptyList());
               if (block.getBody().getTransactionCount() != receipts.size()) {
-                final BytesValueRLPOutput headerRlpOutput = new BytesValueRLPOutput();
-                block.getHeader().writeTo(headerRlpOutput);
-                LOG.atTrace()
-                    .setMessage("Header RLP: {}")
-                    .addArgument(headerRlpOutput.encoded())
-                    .log();
-                LOG.atTrace().setMessage("Body: {}").addArgument(block.getBody().getRlp()).log();
                 throw new IllegalStateException(
                     "PeerTask response code was success, but incorrect number of receipts returned. Block hash: "
                         + block.getHeader().getHash()
@@ -90,16 +234,9 @@ List<SyncBlockWithReceipts> combineBlocksAndReceipts(
         .toList();
   }
 
-  @Override
-  Map<BlockHeader, List<SyncTransactionReceipt>> getReceipts(final List<BlockHeader> headers) {
-    GetSyncReceiptsFromPeerTask task =
-        new GetSyncReceiptsFromPeerTask(headers, protocolSchedule, syncTransactionReceiptEncoder);
-    PeerTaskExecutorResult<Map<BlockHeader, List<SyncTransactionReceipt>>> getReceiptsResult =
-        peerTaskExecutor.execute(task);
-    if (getReceiptsResult.responseCode() == PeerTaskExecutorResponseCode.SUCCESS
-        && getReceiptsResult.result().isPresent()) {
-      return getReceiptsResult.result().get();
-    }
-    return Collections.emptyMap();
+  private String formatBlockDetails(final Collection<SyncBlock> blocks) {
+    return blocks.stream()
+        .map(sb -> sb.getHeader().getNumber() + "(" + sb.getBody().getTransactionCount() + ")")
+        .collect(Collectors.joining(","));
   }
 }
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/FastSyncDownloadPipelineFactory.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/FastSyncDownloadPipelineFactory.java
index 10b64bcbcf4..d8105b0b617 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/FastSyncDownloadPipelineFactory.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/FastSyncDownloadPipelineFactory.java
@@ -49,6 +49,7 @@
 import org.hyperledger.besu.services.pipeline.PipelineBuilder;
 import org.hyperledger.besu.util.InvalidConfigurationException;
 
+import java.time.Duration;
 import java.util.concurrent.CompletionStage;
 
 import org.slf4j.Logger;
@@ -146,7 +147,8 @@ public Pipeline<SyncTargetRange> createDownloadPipelineForSyncTarget(
         new DownloadSyncReceiptsStep(
             protocolSchedule,
             ethContext,
-            new SyncTransactionReceiptEncoder(new SimpleNoCopyRlpEncoder()));
+            new SyncTransactionReceiptEncoder(new SimpleNoCopyRlpEncoder()),
+            Duration.ofMillis(syncConfig.getReceiptsDownloadStepTimeoutMillis()));
     final BlockHeader pivotBlockHeader =
         fastSyncState
             .getPivotBlockHeader()
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapSyncChainDownloadPipelineFactory.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapSyncChainDownloadPipelineFactory.java
index bb22d78def7..183d3524874 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapSyncChainDownloadPipelineFactory.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapSyncChainDownloadPipelineFactory.java
@@ -45,6 +45,7 @@
 import org.hyperledger.besu.services.pipeline.Pipeline;
 import org.hyperledger.besu.services.pipeline.PipelineBuilder;
 
+import java.time.Duration;
 import java.util.List;
 
 import org.slf4j.Logger;
@@ -191,7 +192,8 @@ public Pipeline<List<BlockHeader>> createForwardBodiesAndReceiptsDownloadPipelin
         new DownloadSyncReceiptsStep(
             protocolSchedule,
             ethContext,
-            new SyncTransactionReceiptEncoder(new SimpleNoCopyRlpEncoder()));
+            new SyncTransactionReceiptEncoder(new SimpleNoCopyRlpEncoder()),
+            Duration.ofMillis(syncConfig.getReceiptsDownloadStepTimeoutMillis()));
 
     final ImportSyncBlocksStep importBlocksStep =
         new ImportSyncBlocksStep(
diff --git a/ethereum/eth/src/test-support/java/org/hyperledger/besu/ethereum/eth/core/Utils.java b/ethereum/eth/src/test-support/java/org/hyperledger/besu/ethereum/eth/core/Utils.java
new file mode 100644
index 00000000000..2e5088a2907
--- /dev/null
+++ b/ethereum/eth/src/test-support/java/org/hyperledger/besu/ethereum/eth/core/Utils.java
@@ -0,0 +1,89 @@
+/*
+ * Copyright contributors to 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.core;
+
+import org.hyperledger.besu.ethereum.core.Block;
+import org.hyperledger.besu.ethereum.core.SyncBlock;
+import org.hyperledger.besu.ethereum.core.SyncBlockBody;
+import org.hyperledger.besu.ethereum.core.SyncTransactionReceipt;
+import org.hyperledger.besu.ethereum.core.TransactionReceipt;
+import org.hyperledger.besu.ethereum.core.encoding.receipt.SyncTransactionReceiptEncoder;
+import org.hyperledger.besu.ethereum.core.encoding.receipt.TransactionReceiptEncoder;
+import org.hyperledger.besu.ethereum.core.encoding.receipt.TransactionReceiptEncodingConfiguration;
+import org.hyperledger.besu.ethereum.mainnet.DefaultProtocolSchedule;
+import org.hyperledger.besu.ethereum.rlp.BytesValueRLPInput;
+import org.hyperledger.besu.ethereum.rlp.BytesValueRLPOutput;
+import org.hyperledger.besu.ethereum.rlp.SimpleNoCopyRlpEncoder;
+
+import java.math.BigInteger;
+import java.util.List;
+import java.util.Optional;
+
+public class Utils {
+  private static final SyncTransactionReceiptEncoder SYNC_RECEIPT_ENCODER =
+      new SyncTransactionReceiptEncoder(new SimpleNoCopyRlpEncoder());
+
+  public static SyncTransactionReceipt receiptToSyncReceipt(
+      final TransactionReceipt receipt,
+      final TransactionReceiptEncodingConfiguration receiptEncodingConfiguration) {
+    BytesValueRLPOutput rlpOutput = new BytesValueRLPOutput();
+    TransactionReceiptEncoder.writeTo(receipt, rlpOutput, receiptEncodingConfiguration);
+    return new SyncTransactionReceipt(rlpOutput.encoded());
+  }
+
+  public static List<SyncTransactionReceipt> receiptsToSyncReceipts(
+      final List<TransactionReceipt> receipts,
+      final TransactionReceiptEncodingConfiguration receiptEncodingConfiguration) {
+
+    return receipts.stream()
+        .map(receipt -> Utils.receiptToSyncReceipt(receipt, receiptEncodingConfiguration))
+        .toList();
+  }
+
+  public static SyncBlock blockToSyncBlock(final Block block) {
+    BytesValueRLPOutput rlpOutput = new BytesValueRLPOutput();
+    block.getBody().writeWrappedBodyTo(rlpOutput);
+    final BytesValueRLPInput input = new BytesValueRLPInput(rlpOutput.encoded(), false);
+    final SyncBlockBody syncBlockBody =
+        SyncBlockBody.readWrappedBodyFrom(
+            input, false, new DefaultProtocolSchedule(Optional.of(BigInteger.ONE)));
+    return new SyncBlock(block.getHeader(), syncBlockBody);
+  }
+
+  public static List<SyncBlock> blocksToSyncBlocks(final List<Block> blocks) {
+    return blocks.stream().map(Utils::blockToSyncBlock).toList();
+  }
+
+  public static int compareSyncReceipts(
+      final SyncTransactionReceipt receipt1, final SyncTransactionReceipt receipt2) {
+    if (receipt1.isFormattedForRootCalculation()) {
+      if (receipt2.isFormattedForRootCalculation()) {
+        return receipt1.getRlpBytes().compareTo(receipt2.getRlpBytes());
+      }
+      return receipt1
+          .getRlpBytes()
+          .compareTo(SYNC_RECEIPT_ENCODER.encodeForRootCalculation(receipt2));
+    } else {
+      if (receipt2.isFormattedForRootCalculation()) {
+        return SYNC_RECEIPT_ENCODER
+            .encodeForRootCalculation(receipt1)
+            .compareTo(receipt2.getRlpBytes());
+      }
+      return SYNC_RECEIPT_ENCODER
+          .encodeForRootCalculation(receipt1)
+          .compareTo(SYNC_RECEIPT_ENCODER.encodeForRootCalculation(receipt2));
+    }
+  }
+}
diff --git a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/peertask/task/GetSyncReceiptsFromPeerTaskTest.java b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/peertask/task/GetSyncReceiptsFromPeerTaskTest.java
new file mode 100644
index 00000000000..3340edb10a3
--- /dev/null
+++ b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/peertask/task/GetSyncReceiptsFromPeerTaskTest.java
@@ -0,0 +1,402 @@
+/*
+ * Copyright contributors to 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.hyperledger.besu.ethereum.eth.core.Utils.receiptToSyncReceipt;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.when;
+
+import org.hyperledger.besu.datatypes.Hash;
+import org.hyperledger.besu.ethereum.core.BlockHeader;
+import org.hyperledger.besu.ethereum.core.Difficulty;
+import org.hyperledger.besu.ethereum.core.SyncBlock;
+import org.hyperledger.besu.ethereum.core.SyncBlockBody;
+import org.hyperledger.besu.ethereum.core.SyncTransactionReceipt;
+import org.hyperledger.besu.ethereum.core.TransactionReceipt;
+import org.hyperledger.besu.ethereum.core.encoding.receipt.SyncTransactionReceiptEncoder;
+import org.hyperledger.besu.ethereum.core.encoding.receipt.TransactionReceiptEncodingConfiguration;
+import org.hyperledger.besu.ethereum.eth.EthProtocol;
+import org.hyperledger.besu.ethereum.eth.core.Utils;
+import org.hyperledger.besu.ethereum.eth.manager.ChainState;
+import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
+import org.hyperledger.besu.ethereum.eth.manager.EthPeerImmutableAttributes;
+import org.hyperledger.besu.ethereum.eth.manager.PeerReputation;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.InvalidPeerTaskResponseException;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.MalformedRlpFromPeerException;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskValidationResponse;
+import org.hyperledger.besu.ethereum.eth.messages.EthProtocolMessages;
+import org.hyperledger.besu.ethereum.eth.messages.GetReceiptsMessage;
+import org.hyperledger.besu.ethereum.eth.messages.ReceiptsMessage;
+import org.hyperledger.besu.ethereum.mainnet.BodyValidation;
+import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
+import org.hyperledger.besu.ethereum.mainnet.ProtocolSpec;
+import org.hyperledger.besu.ethereum.p2p.rlpx.connections.PeerConnection;
+import org.hyperledger.besu.ethereum.rlp.SimpleNoCopyRlpEncoder;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import org.apache.commons.lang3.StringUtils;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mockito;
+
+public class GetSyncReceiptsFromPeerTaskTest {
+  private static ProtocolSchedule protocolSchedule;
+
+  @BeforeAll
+  public static void setup() {
+    protocolSchedule = mock(ProtocolSchedule.class);
+    final ProtocolSpec protocolSpec = mock(ProtocolSpec.class);
+    when(protocolSpec.isPoS()).thenReturn(false);
+    when(protocolSchedule.getByBlockHeader(Mockito.any())).thenReturn(protocolSpec);
+    when(protocolSchedule.anyMatch(Mockito.any())).thenReturn(false);
+  }
+
+  @Test
+  public void testGetSubProtocol() {
+    final BlockHeader blockHeader = mockBlockHeader(1);
+    final TransactionReceipt receipt =
+        new TransactionReceipt(1, 123, Collections.emptyList(), Optional.empty());
+    when(blockHeader.getReceiptsRoot()).thenReturn(BodyValidation.receiptsRoot(List.of(receipt)));
+    final SyncBlockBody syncBlockBody = mock(SyncBlockBody.class);
+    when(syncBlockBody.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock = new SyncBlock(blockHeader, syncBlockBody);
+
+    final var task = createTask(List.of(syncBlock), protocolSchedule);
+    assertEquals(EthProtocol.get(), task.getSubProtocol());
+  }
+
+  @Test
+  public void testGetRequestMessage() {
+    final BlockHeader blockHeader1 = mockBlockHeader(1);
+    final TransactionReceipt receipt1_forBlock1 =
+        new TransactionReceipt(1, 123, Collections.emptyList(), Optional.empty());
+    final TransactionReceipt receipt2_forBlock1 =
+        new TransactionReceipt(1, 321, Collections.emptyList(), Optional.empty());
+    when(blockHeader1.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receipt1_forBlock1, receipt2_forBlock1)));
+    final SyncBlockBody syncBlockBody1 = mock(SyncBlockBody.class);
+    when(syncBlockBody1.getTransactionCount()).thenReturn(2);
+    final SyncBlock syncBlock1 = new SyncBlock(blockHeader1, syncBlockBody1);
+
+    BlockHeader blockHeader2 = mockBlockHeader(2);
+    TransactionReceipt receiptForBlock2 =
+        new TransactionReceipt(1, 456, Collections.emptyList(), Optional.empty());
+    when(blockHeader2.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receiptForBlock2)));
+    final SyncBlockBody syncBlockBody2 = mock(SyncBlockBody.class);
+    when(syncBlockBody2.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock2 = new SyncBlock(blockHeader2, syncBlockBody2);
+
+    BlockHeader blockHeader3 = mockBlockHeader(3);
+    TransactionReceipt receiptForBlock3 =
+        new TransactionReceipt(1, 789, Collections.emptyList(), Optional.empty());
+    when(blockHeader3.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receiptForBlock3)));
+    final SyncBlockBody syncBlockBody3 = mock(SyncBlockBody.class);
+    when(syncBlockBody3.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock3 = new SyncBlock(blockHeader3, syncBlockBody3);
+
+    final List<SyncBlock> blocks = List.of(syncBlock1, syncBlock2, syncBlock3);
+
+    final var task = createTask(blocks, protocolSchedule);
+
+    final var messageData = task.getRequestMessage();
+    final var getReceiptsMessage = GetReceiptsMessage.readFrom(messageData);
+
+    assertEquals(EthProtocolMessages.GET_RECEIPTS, getReceiptsMessage.getCode());
+
+    List<Hash> hashesInMessage = getReceiptsMessage.hashes();
+    List<Hash> expectedHashes = blocks.stream().map(SyncBlock::getHash).toList();
+
+    assertThat(expectedHashes).containsExactlyElementsOf(hashesInMessage);
+  }
+
+  @Test
+  public void testParseResponseWithNullResponseMessage() {
+    final BlockHeader blockHeader = mockBlockHeader(1);
+    final TransactionReceipt receipt =
+        new TransactionReceipt(1, 123, Collections.emptyList(), Optional.empty());
+    when(blockHeader.getReceiptsRoot()).thenReturn(BodyValidation.receiptsRoot(List.of(receipt)));
+    final SyncBlockBody syncBlockBody = mock(SyncBlockBody.class);
+    when(syncBlockBody.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock = new SyncBlock(blockHeader, syncBlockBody);
+
+    final var task = createTask(List.of(syncBlock), protocolSchedule);
+    Assertions.assertThrows(
+        InvalidPeerTaskResponseException.class, () -> task.processResponse(null));
+  }
+
+  @Test
+  public void testParseResponse()
+      throws InvalidPeerTaskResponseException, MalformedRlpFromPeerException {
+    BlockHeader blockHeader1 = mockBlockHeader(1);
+    TransactionReceipt receiptForBlock1 =
+        new TransactionReceipt(1, 123, Collections.emptyList(), Optional.empty());
+    when(blockHeader1.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receiptForBlock1)));
+    final SyncBlockBody syncBlockBody1 = mock(SyncBlockBody.class);
+    when(syncBlockBody1.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock1 = new SyncBlock(blockHeader1, syncBlockBody1);
+
+    BlockHeader blockHeader2 = mockBlockHeader(2);
+    TransactionReceipt receiptForBlock2 =
+        new TransactionReceipt(1, 456, Collections.emptyList(), Optional.empty());
+    when(blockHeader2.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receiptForBlock2)));
+    final SyncBlockBody syncBlockBody2 = mock(SyncBlockBody.class);
+    when(syncBlockBody2.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock2 = new SyncBlock(blockHeader2, syncBlockBody2);
+
+    BlockHeader blockHeader3 = mockBlockHeader(3);
+    TransactionReceipt receiptForBlock3 =
+        new TransactionReceipt(1, 789, Collections.emptyList(), Optional.empty());
+    when(blockHeader3.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receiptForBlock3)));
+    final SyncBlockBody syncBlockBody3 = mock(SyncBlockBody.class);
+    when(syncBlockBody3.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock3 = new SyncBlock(blockHeader3, syncBlockBody3);
+
+    final var task = createTask(List.of(syncBlock1, syncBlock2, syncBlock3), protocolSchedule);
+
+    ReceiptsMessage receiptsMessage =
+        ReceiptsMessage.create(
+            List.of(
+                List.of(receiptForBlock1), List.of(receiptForBlock2), List.of(receiptForBlock3)),
+            TransactionReceiptEncodingConfiguration.DEFAULT_NETWORK_CONFIGURATION);
+
+    final var response = task.processResponse(receiptsMessage);
+
+    assertThat(response).hasSize(3);
+    assertThat(response.get(syncBlock1))
+        .usingElementComparator(Utils::compareSyncReceipts)
+        .containsExactly(toResponseReceipt(receiptForBlock1));
+    assertThat(response.get(syncBlock2))
+        .usingElementComparator(Utils::compareSyncReceipts)
+        .containsExactly(toResponseReceipt(receiptForBlock2));
+    assertThat(response.get(syncBlock3))
+        .usingElementComparator(Utils::compareSyncReceipts)
+        .containsExactly(toResponseReceipt(receiptForBlock3));
+  }
+
+  @ParameterizedTest
+  @ValueSource(booleans = {false, true})
+  public void testGetPeerRequirementFilter(final boolean isPoS) {
+    reset(protocolSchedule);
+    when(protocolSchedule.anyMatch(any())).thenReturn(isPoS);
+
+    BlockHeader blockHeader1 = mockBlockHeader(1);
+    TransactionReceipt receiptForBlock1 =
+        new TransactionReceipt(1, 123, Collections.emptyList(), Optional.empty());
+    when(blockHeader1.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receiptForBlock1)));
+    final SyncBlockBody syncBlockBody1 = mock(SyncBlockBody.class);
+    when(syncBlockBody1.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock1 = new SyncBlock(blockHeader1, syncBlockBody1);
+
+    BlockHeader blockHeader2 = mockBlockHeader(2);
+    TransactionReceipt receiptForBlock2 =
+        new TransactionReceipt(1, 456, Collections.emptyList(), Optional.empty());
+    when(blockHeader2.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receiptForBlock2)));
+    final SyncBlockBody syncBlockBody2 = mock(SyncBlockBody.class);
+    when(syncBlockBody2.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock2 = new SyncBlock(blockHeader2, syncBlockBody2);
+
+    final var task = createTask(List.of(syncBlock1, syncBlock2), protocolSchedule);
+
+    EthPeer failForShortChainHeight = mockPeer(1);
+    EthPeer successfulCandidate = mockPeer(5);
+
+    assertThat(
+            task.getPeerRequirementFilter()
+                .test(EthPeerImmutableAttributes.from(failForShortChainHeight)))
+        .isEqualTo(isPoS);
+    Assertions.assertTrue(
+        task.getPeerRequirementFilter().test(EthPeerImmutableAttributes.from(successfulCandidate)));
+  }
+
+  @Test
+  public void validateResultFailsWhenNoResultAreReturned() {
+    final BlockHeader blockHeader = mockBlockHeader(1);
+    final TransactionReceipt receiptForBlock =
+        new TransactionReceipt(1, 123, Collections.emptyList(), Optional.empty());
+    when(blockHeader.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receiptForBlock)));
+    final SyncBlockBody syncBlockBody = mock(SyncBlockBody.class);
+    when(syncBlockBody.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock = new SyncBlock(blockHeader, syncBlockBody);
+
+    final var task = createTask(List.of(syncBlock), protocolSchedule);
+
+    assertEquals(PeerTaskValidationResponse.NO_RESULTS_RETURNED, task.validateResult(Map.of()));
+  }
+
+  @Test
+  public void testValidateResultForFullSuccess() {
+    final BlockHeader blockHeader = mockBlockHeader(1);
+    final TransactionReceipt receiptForBlock =
+        new TransactionReceipt(1, 123, Collections.emptyList(), Optional.empty());
+    when(blockHeader.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receiptForBlock)));
+    final SyncBlockBody syncBlockBody = mock(SyncBlockBody.class);
+    when(syncBlockBody.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock = new SyncBlock(blockHeader, syncBlockBody);
+
+    final var task = createTask(List.of(syncBlock), protocolSchedule);
+
+    assertEquals(
+        PeerTaskValidationResponse.RESULTS_VALID_AND_GOOD,
+        task.validateResult(Map.of(syncBlock, List.of(toResponseReceipt(receiptForBlock)))));
+  }
+
+  @Test
+  public void testParseResponseForInvalidResponse() {
+    // Too many block-lists in the response (4 for 3 requested blocks) must be rejected at parse
+    // time by processResponse, not deferred to validateResult.
+    final BlockHeader blockHeader1 = mockBlockHeader(1);
+    final TransactionReceipt receiptForBlock1 =
+        new TransactionReceipt(1, 123, Collections.emptyList(), Optional.empty());
+    when(blockHeader1.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receiptForBlock1)));
+    final SyncBlockBody syncBlockBody1 = mock(SyncBlockBody.class);
+    when(syncBlockBody1.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock1 = new SyncBlock(blockHeader1, syncBlockBody1);
+
+    final BlockHeader blockHeader2 = mockBlockHeader(2);
+    final TransactionReceipt receiptForBlock2 =
+        new TransactionReceipt(1, 456, Collections.emptyList(), Optional.empty());
+    when(blockHeader2.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receiptForBlock2)));
+    final SyncBlockBody syncBlockBody2 = mock(SyncBlockBody.class);
+    when(syncBlockBody2.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock2 = new SyncBlock(blockHeader2, syncBlockBody2);
+
+    final BlockHeader blockHeader3 = mockBlockHeader(3);
+    final TransactionReceipt receiptForBlock3 =
+        new TransactionReceipt(1, 789, Collections.emptyList(), Optional.empty());
+    when(blockHeader3.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receiptForBlock3)));
+    final SyncBlockBody syncBlockBody3 = mock(SyncBlockBody.class);
+    when(syncBlockBody3.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock3 = new SyncBlock(blockHeader3, syncBlockBody3);
+
+    final var task = createTask(List.of(syncBlock1, syncBlock2, syncBlock3), protocolSchedule);
+
+    final ReceiptsMessage receiptsMessage =
+        ReceiptsMessage.create(
+            List.of(
+                List.of(receiptForBlock1),
+                List.of(receiptForBlock2),
+                List.of(receiptForBlock3),
+                List.of(
+                    new TransactionReceipt(1, 101112, Collections.emptyList(), Optional.empty()))),
+            TransactionReceiptEncodingConfiguration.DEFAULT_NETWORK_CONFIGURATION);
+
+    Assertions.assertThrows(
+        InvalidPeerTaskResponseException.class, () -> task.processResponse(receiptsMessage));
+  }
+
+  @Test
+  public void validateResultFailsWhenTooManyBlocksReturned() {
+    // A block with 1 transaction that receives 2 receipts must be rejected.
+    final BlockHeader blockHeader1 = mockBlockHeader(1);
+    final TransactionReceipt receiptForBlock1 =
+        new TransactionReceipt(1, 123, Collections.emptyList(), Optional.empty());
+    when(blockHeader1.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receiptForBlock1)));
+    final SyncBlockBody syncBlockBody1 = mock(SyncBlockBody.class);
+    when(syncBlockBody1.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock1 = new SyncBlock(blockHeader1, syncBlockBody1);
+
+    final TransactionReceipt extraReceipt =
+        new TransactionReceipt(1, 321, Collections.emptyList(), Optional.empty());
+
+    final var task = createTask(List.of(syncBlock1), protocolSchedule);
+
+    assertEquals(
+        PeerTaskValidationResponse.TOO_MANY_RESULTS_RETURNED,
+        task.validateResult(
+            Map.of(
+                syncBlock1,
+                List.of(toResponseReceipt(receiptForBlock1), toResponseReceipt(extraReceipt)))));
+  }
+
+  @Test
+  public void validateResultFailsReceiptRootDoesNotMatch() {
+    final BlockHeader blockHeader1 = mockBlockHeader(1);
+    final TransactionReceipt receiptForBlock1 =
+        new TransactionReceipt(1, 123, Collections.emptyList(), Optional.empty());
+    when(blockHeader1.getReceiptsRoot())
+        .thenReturn(BodyValidation.receiptsRoot(List.of(receiptForBlock1)));
+    final SyncBlockBody syncBlockBody1 = mock(SyncBlockBody.class);
+    when(syncBlockBody1.getTransactionCount()).thenReturn(1);
+    final SyncBlock syncBlock1 = new SyncBlock(blockHeader1, syncBlockBody1);
+
+    final TransactionReceipt returnedReceiptForBlock1 =
+        new TransactionReceipt(1, 321, Collections.emptyList(), Optional.empty());
+
+    final var task = createTask(List.of(syncBlock1), protocolSchedule);
+
+    assertEquals(
+        PeerTaskValidationResponse.RESULTS_DO_NOT_MATCH_QUERY,
+        task.validateResult(
+            Map.of(syncBlock1, List.of(toResponseReceipt(returnedReceiptForBlock1)))));
+  }
+
+  private static BlockHeader mockBlockHeader(final long blockNumber) {
+    BlockHeader blockHeader = mock(BlockHeader.class);
+    when(blockHeader.getNumber()).thenReturn(blockNumber);
+    // second to last hex digit indicates the blockNumber, last hex digit indicates the usage of the
+    // hash
+    when(blockHeader.getHash())
+        .thenReturn(Hash.fromHexString(StringUtils.repeat("00", 31) + blockNumber + "1"));
+
+    return blockHeader;
+  }
+
+  private EthPeer mockPeer(final long chainHeight) {
+    EthPeer ethPeer = mock(EthPeer.class);
+    ChainState chainState = mock(ChainState.class);
+
+    when(ethPeer.chainState()).thenReturn(chainState);
+    when(chainState.getEstimatedHeight()).thenReturn(chainHeight);
+    when(chainState.getEstimatedTotalDifficulty()).thenReturn(Difficulty.of(0));
+    when(ethPeer.getReputation()).thenReturn(new PeerReputation());
+    PeerConnection connection = mock(PeerConnection.class);
+    when(ethPeer.getConnection()).thenReturn(connection);
+    return ethPeer;
+  }
+
+  private GetSyncReceiptsFromPeerTask createTask(
+      final List<SyncBlock> blocks, final ProtocolSchedule protocolSchedule) {
+    return new GetSyncReceiptsFromPeerTask(
+        blocks, protocolSchedule, new SyncTransactionReceiptEncoder(new SimpleNoCopyRlpEncoder()));
+  }
+
+  private SyncTransactionReceipt toResponseReceipt(final TransactionReceipt receipt) {
+    return receiptToSyncReceipt(receipt, TransactionReceiptEncodingConfiguration.DEFAULT);
+  }
+}
diff --git a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/DownloadSyncReceiptsStepTest.java b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/DownloadSyncReceiptsStepTest.java
new file mode 100644
index 00000000000..fadb29bdbac
--- /dev/null
+++ b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/DownloadSyncReceiptsStepTest.java
@@ -0,0 +1,565 @@
+/*
+ * Copyright contributors to 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.sync.fastsync;
+
+import static java.util.Collections.emptyList;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.hyperledger.besu.ethereum.core.encoding.receipt.TransactionReceiptEncodingConfiguration.ETH69_RECEIPT_CONFIGURATION;
+import static org.hyperledger.besu.ethereum.eth.core.Utils.blocksToSyncBlocks;
+import static org.hyperledger.besu.ethereum.eth.core.Utils.receiptsToSyncReceipts;
+import static org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutorResponseCode.NO_PEER_AVAILABLE;
+import static org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutorResponseCode.PEER_DISCONNECTED;
+import static org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutorResponseCode.SUCCESS;
+import static org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutorResponseCode.TIMEOUT;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.hyperledger.besu.datatypes.Hash;
+import org.hyperledger.besu.ethereum.core.Block;
+import org.hyperledger.besu.ethereum.core.BlockDataGenerator;
+import org.hyperledger.besu.ethereum.core.BlockDataGenerator.BlockOptions;
+import org.hyperledger.besu.ethereum.core.SyncBlock;
+import org.hyperledger.besu.ethereum.core.SyncBlockWithReceipts;
+import org.hyperledger.besu.ethereum.core.SyncTransactionReceipt;
+import org.hyperledger.besu.ethereum.core.encoding.receipt.SyncTransactionReceiptEncoder;
+import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutor;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerTaskExecutorResult;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.task.GetSyncReceiptsFromPeerTask;
+import org.hyperledger.besu.ethereum.mainnet.DefaultProtocolSchedule;
+import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
+import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
+import org.hyperledger.besu.testutil.DeterministicEthScheduler;
+
+import java.math.BigInteger;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+public class DownloadSyncReceiptsStepTest {
+
+  @Mock private EthContext ethContext;
+  @Mock private PeerTaskExecutor peerTaskExecutor;
+  @Mock private SyncTransactionReceiptEncoder syncTransactionReceiptEncoder;
+  private ProtocolSchedule protocolSchedule;
+  private final BlockDataGenerator gen = new BlockDataGenerator();
+
+  private DownloadSyncReceiptsStep downloadSyncReceiptsStep;
+
+  @BeforeEach
+  public void setUp() {
+    protocolSchedule = new DefaultProtocolSchedule(Optional.of(BigInteger.ONE));
+    when(ethContext.getScheduler()).thenReturn(new DeterministicEthScheduler());
+    when(ethContext.getPeerTaskExecutor()).thenReturn(peerTaskExecutor);
+
+    downloadSyncReceiptsStep =
+        new DownloadSyncReceiptsStep(
+            protocolSchedule, ethContext, syncTransactionReceiptEncoder, Duration.ofMinutes(1));
+  }
+
+  @Test
+  public void shouldDownloadReceiptsForBlocksWithTransactions()
+      throws ExecutionException, InterruptedException {
+
+    // skip genesis block, since we do not need to retrieve receipt for it
+    final List<Block> blockWithTxs = gen.blockSequence(3).subList(1, 3);
+
+    final var receiptsPerBlock =
+        blockWithTxs.stream()
+            .map(gen::receipts)
+            .map(rs -> receiptsToSyncReceipts(rs, ETH69_RECEIPT_CONFIGURATION))
+            .toList();
+    final var syncBlocks = blocksToSyncBlocks(blockWithTxs);
+
+    final Map<SyncBlock, List<SyncTransactionReceipt>> returnedReceiptsByBlock = new HashMap<>();
+    for (int i = 0; i < syncBlocks.size(); i++) {
+      returnedReceiptsByBlock.put(syncBlocks.get(i), receiptsPerBlock.get(i));
+    }
+
+    // Mock the peer task executor to return receipts for both blocks
+    final var executorResult =
+        new PeerTaskExecutorResult<>(Optional.of(returnedReceiptsByBlock), SUCCESS, emptyList());
+    when(peerTaskExecutor.execute(any(GetSyncReceiptsFromPeerTask.class)))
+        .thenReturn(executorResult);
+
+    // When: downloading receipts
+    final CompletableFuture<List<SyncBlockWithReceipts>> result =
+        downloadSyncReceiptsStep.apply(syncBlocks);
+
+    // Then: should return all blocks with receipts
+    final List<SyncBlockWithReceipts> blocksWithReceipts = result.get();
+    assertThat(blocksWithReceipts).hasSize(2);
+    for (int i = 0; i < blocksWithReceipts.size(); i++) {
+      assertThat(blocksWithReceipts.get(i).getBlock()).isEqualTo(syncBlocks.get(i));
+      assertThat(blocksWithReceipts.get(i).getReceipts()).isEqualTo(receiptsPerBlock.get(i));
+    }
+
+    // Verify the task was executed once
+    verify(peerTaskExecutor, times(1)).execute(any(GetSyncReceiptsFromPeerTask.class));
+  }
+
+  @Test
+  public void shouldSkipDownloadForBlocksWithEmptyReceiptsRoot()
+      throws ExecutionException, InterruptedException {
+
+    final Block block1_withTxs = gen.blockSequence(gen.genesisBlock(), 1).getFirst();
+
+    gen.setBlockOptionsSupplier(
+        () -> BlockOptions.create().hasTransactions(false).setReceiptsRoot(Hash.EMPTY_TRIE_HASH));
+    final Block block2_withoutTxs = gen.blockSequence(block1_withTxs, 1).getFirst();
+
+    gen.setBlockOptionsSupplier(() -> BlockOptions.create().hasTransactions(true));
+    final Block block3_withTxs = gen.blockSequence(block2_withoutTxs, 1).getFirst();
+
+    final var blocks = List.of(block1_withTxs, block2_withoutTxs, block3_withTxs);
+
+    // we must not request receipt for block2, so only return receipts for the 2 blocks with txs
+    final var receiptsForBlock1 =
+        receiptsToSyncReceipts(gen.receipts(block1_withTxs), ETH69_RECEIPT_CONFIGURATION);
+    final var receiptsForBlock3 =
+        receiptsToSyncReceipts(gen.receipts(block3_withTxs), ETH69_RECEIPT_CONFIGURATION);
+
+    final var syncBlocks = blocksToSyncBlocks(blocks);
+
+    final Map<SyncBlock, List<SyncTransactionReceipt>> returnedReceiptsByBlock =
+        Map.of(syncBlocks.get(0), receiptsForBlock1, syncBlocks.get(2), receiptsForBlock3);
+
+    final var executorResult =
+        new PeerTaskExecutorResult<>(Optional.of(returnedReceiptsByBlock), SUCCESS, emptyList());
+    when(peerTaskExecutor.execute(any(GetSyncReceiptsFromPeerTask.class)))
+        .thenReturn(executorResult);
+
+    // When: downloading receipts
+    final CompletableFuture<List<SyncBlockWithReceipts>> result =
+        downloadSyncReceiptsStep.apply(syncBlocks);
+
+    // Then: should also return block with empty receipts
+    final List<SyncBlockWithReceipts> blocksWithReceipts = result.get();
+    assertThat(blocksWithReceipts).hasSize(3);
+
+    assertThat(blocksWithReceipts.get(0).getBlock()).isEqualTo(syncBlocks.get(0));
+    assertThat(blocksWithReceipts.get(0).getReceipts()).isEqualTo(receiptsForBlock1);
+
+    assertThat(blocksWithReceipts.get(1).getBlock()).isEqualTo(syncBlocks.get(1));
+    assertThat(blocksWithReceipts.get(1).getReceipts()).isEmpty();
+
+    assertThat(blocksWithReceipts.get(2).getBlock()).isEqualTo(syncBlocks.get(2));
+    assertThat(blocksWithReceipts.get(2).getReceipts()).isEqualTo(receiptsForBlock3);
+
+    // Verify the task was executed once
+    verify(peerTaskExecutor, times(1)).execute(any(GetSyncReceiptsFromPeerTask.class));
+  }
+
+  @Test
+  public void shouldRetryUntilAllReceiptsDownloaded()
+      throws ExecutionException, InterruptedException {
+    // Given: 3 blocks with transactions
+    final List<Block> blocks = gen.blockSequence(4).subList(1, 4);
+    final List<SyncBlock> syncBlocks = blocksToSyncBlocks(blocks);
+
+    final var receiptsPerBlock =
+        blocks.stream()
+            .map(gen::receipts)
+            .map(rs -> receiptsToSyncReceipts(rs, ETH69_RECEIPT_CONFIGURATION))
+            .toList();
+
+    // First call returns first block only
+    final var firstExecutorResult =
+        new PeerTaskExecutorResult<>(
+            Optional.of(Map.of(syncBlocks.get(0), receiptsPerBlock.get(0))), SUCCESS, emptyList());
+
+    // Second call returns second block only
+    final var secondExecutorResult =
+        new PeerTaskExecutorResult<>(
+            Optional.of(Map.of(syncBlocks.get(1), receiptsPerBlock.get(1))), SUCCESS, emptyList());
+
+    // Third call returns third block only
+    final var thirdExecutorResult =
+        new PeerTaskExecutorResult<>(
+            Optional.of(Map.of(syncBlocks.get(2), receiptsPerBlock.get(2))), SUCCESS, emptyList());
+
+    when(peerTaskExecutor.execute(any(GetSyncReceiptsFromPeerTask.class)))
+        .thenReturn(firstExecutorResult)
+        .thenReturn(secondExecutorResult)
+        .thenReturn(thirdExecutorResult);
+
+    // When: downloading receipts
+    final CompletableFuture<List<SyncBlockWithReceipts>> result =
+        downloadSyncReceiptsStep.apply(syncBlocks);
+
+    // Then: should return all blocks with receipts
+    final List<SyncBlockWithReceipts> blocksWithReceipts = result.get();
+    assertThat(blocksWithReceipts).hasSize(3);
+    for (int i = 0; i < blocksWithReceipts.size(); i++) {
+      assertThat(blocksWithReceipts.get(i).getReceipts()).isEqualTo(receiptsPerBlock.get(i));
+    }
+
+    // Verify the task was executed three times
+    verify(peerTaskExecutor, times(3)).execute(any(GetSyncReceiptsFromPeerTask.class));
+  }
+
+  @Test
+  public void combineBlocksAndReceiptsShouldThrowWhenReceiptCountMismatch() {
+    // Given: a block with transactions and fewer receipts than transactions
+    gen.setBlockOptionsSupplier(
+        () -> BlockOptions.create().hasTransactions(true).transactionCount(3));
+    final Block block = gen.block();
+    final List<SyncBlock> syncBlocks = blocksToSyncBlocks(List.of(block));
+
+    final List<SyncTransactionReceipt> allReceipts =
+        receiptsToSyncReceipts(gen.receipts(block), ETH69_RECEIPT_CONFIGURATION);
+
+    // Add fewer receipts than transactions
+    final Map<Hash, List<SyncTransactionReceipt>> receiptsByRootHash =
+        Map.of(
+            syncBlocks.get(0).getHeader().getReceiptsRoot(),
+            allReceipts.subList(0, allReceipts.size() - 1));
+
+    // When/Then: should throw IllegalStateException
+    assertThatThrownBy(
+            () -> downloadSyncReceiptsStep.combineBlocksAndReceipts(syncBlocks, receiptsByRootHash))
+        .isInstanceOf(IllegalStateException.class)
+        .hasMessageContaining("incorrect number of receipts returned");
+  }
+
+  @Test
+  public void combineBlocksAndReceiptsShouldReturnEmptyReceiptsWhenNotInMap() {
+    // Given: blocks without transactions and without receipts in map
+    gen.setBlockOptionsSupplier(() -> BlockOptions.create().hasTransactions(false));
+    final Block blockWithoutTxs = gen.block();
+    final List<SyncBlock> syncBlocks = blocksToSyncBlocks(List.of(blockWithoutTxs));
+
+    final Map<Hash, List<SyncTransactionReceipt>> receiptsByRootHash = Map.of();
+
+    // When: combining blocks and receipts
+    final List<SyncBlockWithReceipts> result =
+        downloadSyncReceiptsStep.combineBlocksAndReceipts(syncBlocks, receiptsByRootHash);
+
+    // Then: should return blocks with empty receipts
+    assertThat(result).hasSize(1);
+    assertThat(result.getFirst().getReceipts()).isEmpty();
+  }
+
+  @Test
+  public void shouldRetryAfterSingleFailureAndEventuallySucceed()
+      throws ExecutionException, InterruptedException {
+    // Given: 2 blocks with transactions
+    final List<Block> blocks = gen.blockSequence(3).subList(1, 3);
+    final List<SyncBlock> syncBlocks = blocksToSyncBlocks(blocks);
+
+    final var receiptsPerBlock =
+        blocks.stream()
+            .map(gen::receipts)
+            .map(rs -> receiptsToSyncReceipts(rs, ETH69_RECEIPT_CONFIGURATION))
+            .toList();
+
+    final Map<SyncBlock, List<SyncTransactionReceipt>> returnedReceiptsByBlock = new HashMap<>();
+    for (int i = 0; i < syncBlocks.size(); i++) {
+      returnedReceiptsByBlock.put(syncBlocks.get(i), receiptsPerBlock.get(i));
+    }
+
+    // First call returns failure (e.g., peer disconnected)
+    final var failureResult =
+        new PeerTaskExecutorResult<Map<SyncBlock, List<SyncTransactionReceipt>>>(
+            Optional.empty(), PEER_DISCONNECTED, emptyList());
+
+    // Second call returns success with all receipts
+    final var successResult =
+        new PeerTaskExecutorResult<>(Optional.of(returnedReceiptsByBlock), SUCCESS, emptyList());
+
+    when(peerTaskExecutor.execute(any(GetSyncReceiptsFromPeerTask.class)))
+        .thenReturn(failureResult)
+        .thenReturn(successResult);
+
+    // When: downloading receipts
+    final CompletableFuture<List<SyncBlockWithReceipts>> result =
+        downloadSyncReceiptsStep.apply(syncBlocks);
+
+    // Then: should eventually succeed after retry
+    final List<SyncBlockWithReceipts> blocksWithReceipts = result.get();
+    assertThat(blocksWithReceipts).hasSize(2);
+    for (int i = 0; i < blocksWithReceipts.size(); i++) {
+      assertThat(blocksWithReceipts.get(i).getBlock()).isEqualTo(syncBlocks.get(i));
+      assertThat(blocksWithReceipts.get(i).getReceipts()).isEqualTo(receiptsPerBlock.get(i));
+    }
+
+    // Verify the task was executed twice (1 failure + 1 success)
+    verify(peerTaskExecutor, times(2)).execute(any(GetSyncReceiptsFromPeerTask.class));
+  }
+
+  @Test
+  public void shouldRetryMultipleTimesAfterConsecutiveFailures()
+      throws ExecutionException, InterruptedException {
+    // Given: 1 block with transactions
+    final List<Block> blocks = gen.blockSequence(2).subList(1, 2);
+    final List<SyncBlock> syncBlocks = blocksToSyncBlocks(blocks);
+
+    final var receiptsPerBlock =
+        blocks.stream()
+            .map(gen::receipts)
+            .map(rs -> receiptsToSyncReceipts(rs, ETH69_RECEIPT_CONFIGURATION))
+            .toList();
+
+    // First three calls return different failures
+    final var timeoutResult =
+        new PeerTaskExecutorResult<Map<SyncBlock, List<SyncTransactionReceipt>>>(
+            Optional.empty(), TIMEOUT, emptyList());
+    final var noPeerResult =
+        new PeerTaskExecutorResult<Map<SyncBlock, List<SyncTransactionReceipt>>>(
+            Optional.empty(), NO_PEER_AVAILABLE, emptyList());
+    final var disconnectedResult =
+        new PeerTaskExecutorResult<Map<SyncBlock, List<SyncTransactionReceipt>>>(
+            Optional.empty(), PEER_DISCONNECTED, emptyList());
+
+    // Fourth call returns success
+    final var successResult =
+        new PeerTaskExecutorResult<>(
+            Optional.of(Map.of(syncBlocks.getFirst(), receiptsPerBlock.getFirst())),
+            SUCCESS,
+            emptyList());
+
+    when(peerTaskExecutor.execute(any(GetSyncReceiptsFromPeerTask.class)))
+        .thenReturn(timeoutResult)
+        .thenReturn(noPeerResult)
+        .thenReturn(disconnectedResult)
+        .thenReturn(successResult);
+
+    // When: downloading receipts
+    final CompletableFuture<List<SyncBlockWithReceipts>> result =
+        downloadSyncReceiptsStep.apply(syncBlocks);
+
+    // Then: should eventually succeed after multiple retries
+    final List<SyncBlockWithReceipts> blocksWithReceipts = result.get();
+    assertThat(blocksWithReceipts).hasSize(1);
+    assertThat(blocksWithReceipts.getFirst().getBlock()).isEqualTo(syncBlocks.getFirst());
+    assertThat(blocksWithReceipts.getFirst().getReceipts()).isEqualTo(receiptsPerBlock.getFirst());
+
+    // Verify the task was executed 4 times (3 failures + 1 success)
+    verify(peerTaskExecutor, times(4)).execute(any(GetSyncReceiptsFromPeerTask.class));
+  }
+
+  @Test
+  public void shouldThrowIllegalStateExceptionWhenSuccessWithEmptyResult() {
+    // Given: 1 block with transactions
+    final List<Block> blocks = gen.blockSequence(2).subList(1, 2);
+    final List<SyncBlock> syncBlocks = blocksToSyncBlocks(blocks);
+
+    // Mock returns SUCCESS but with empty Optional (should never happen in practice)
+    final var invalidResult =
+        new PeerTaskExecutorResult<Map<SyncBlock, List<SyncTransactionReceipt>>>(
+            Optional.empty(), SUCCESS, emptyList());
+
+    when(peerTaskExecutor.execute(any(GetSyncReceiptsFromPeerTask.class)))
+        .thenReturn(invalidResult);
+
+    // When: downloading receipts
+    final CompletableFuture<List<SyncBlockWithReceipts>> result =
+        downloadSyncReceiptsStep.apply(syncBlocks);
+
+    // Then: should throw IllegalStateException (wrapped in ExecutionException and
+    // CompletionException)
+    assertThatThrownBy(result::get)
+        .isInstanceOf(ExecutionException.class)
+        .hasRootCauseInstanceOf(IllegalStateException.class)
+        .hasRootCauseMessage("Task validation failure, it must flag empty result as failure");
+
+    // Verify the task was executed once
+    verify(peerTaskExecutor, times(1)).execute(any(GetSyncReceiptsFromPeerTask.class));
+  }
+
+  @Test
+  public void shouldDeduplicateBlocksWithSameReceiptRoot()
+      throws ExecutionException, InterruptedException {
+    // Given: Create 3 blocks where block2 has the same receipt root as block1.
+    // The dedup in prepareRequest sends only one request per unique receipt root, so block2 is
+    // excluded from the request. combineBlocksAndReceipts looks up by receipt root hash, so block2
+    // still gets block1's receipts.
+    final List<Block> allBlocks = gen.blockSequence(4).subList(1, 4);
+    final Block block1 = allBlocks.get(0);
+    final Block block3 = allBlocks.get(2);
+
+    gen.setBlockOptionsSupplier(
+        () ->
+            BlockOptions.create()
+                .hasTransactions(true)
+                .setReceiptsRoot(block1.getHeader().getReceiptsRoot()));
+    final Block block2WithSameRoot = gen.blockSequence(block1, 1).getFirst();
+
+    final List<Block> blocks = List.of(block1, block2WithSameRoot, block3);
+    final List<SyncBlock> syncBlocks = blocksToSyncBlocks(blocks);
+
+    // Only 2 unique receipt requests should be made (for block1 and block3)
+    final var receiptsForBlock1 =
+        receiptsToSyncReceipts(gen.receipts(block1), ETH69_RECEIPT_CONFIGURATION);
+    final var receiptsForBlock3 =
+        receiptsToSyncReceipts(gen.receipts(block3), ETH69_RECEIPT_CONFIGURATION);
+
+    final Map<SyncBlock, List<SyncTransactionReceipt>> returnedReceiptsByBlock =
+        Map.of(syncBlocks.get(0), receiptsForBlock1, syncBlocks.get(2), receiptsForBlock3);
+
+    final var executorResult =
+        new PeerTaskExecutorResult<>(Optional.of(returnedReceiptsByBlock), SUCCESS, emptyList());
+    when(peerTaskExecutor.execute(any(GetSyncReceiptsFromPeerTask.class)))
+        .thenReturn(executorResult);
+
+    // When: downloading receipts
+    final CompletableFuture<List<SyncBlockWithReceipts>> result =
+        downloadSyncReceiptsStep.apply(syncBlocks);
+
+    // Then: should return all 3 blocks with receipts
+    final List<SyncBlockWithReceipts> blocksWithReceipts = result.get();
+    assertThat(blocksWithReceipts).hasSize(3);
+
+    // block1 should have its receipts
+    assertThat(blocksWithReceipts.get(0).getBlock()).isEqualTo(syncBlocks.get(0));
+    assertThat(blocksWithReceipts.get(0).getReceipts()).isEqualTo(receiptsForBlock1);
+
+    // block2 was deduped out of the request; combineBlocksAndReceipts finds its receipts via the
+    // shared receipt root, so it gets the same receipts as block1
+    assertThat(blocksWithReceipts.get(1).getBlock()).isEqualTo(syncBlocks.get(1));
+    assertThat(blocksWithReceipts.get(1).getReceipts()).isEqualTo(receiptsForBlock1);
+
+    // block3 should have its own receipts
+    assertThat(blocksWithReceipts.get(2).getBlock()).isEqualTo(syncBlocks.get(2));
+    assertThat(blocksWithReceipts.get(2).getReceipts()).isEqualTo(receiptsForBlock3);
+
+    // Verify the task was executed only once (deduplication worked)
+    verify(peerTaskExecutor, times(1)).execute(any(GetSyncReceiptsFromPeerTask.class));
+  }
+
+  @Test
+  public void shouldHandlePartialSuccessThenFailureThenSuccess()
+      throws ExecutionException, InterruptedException {
+    // Given: 4 blocks with transactions
+    final List<Block> blocks = gen.blockSequence(5).subList(1, 5);
+    final List<SyncBlock> syncBlocks = blocksToSyncBlocks(blocks);
+
+    final var allReceiptsPerBlock =
+        blocks.stream()
+            .map(gen::receipts)
+            .map(rs -> receiptsToSyncReceipts(rs, ETH69_RECEIPT_CONFIGURATION))
+            .toList();
+
+    // First call returns partial success (first 2 blocks only)
+    final var firstSuccessResult =
+        new PeerTaskExecutorResult<>(
+            Optional.of(
+                Map.of(
+                    syncBlocks.get(0), allReceiptsPerBlock.get(0),
+                    syncBlocks.get(1), allReceiptsPerBlock.get(1))),
+            SUCCESS,
+            emptyList());
+
+    // Second call for remaining blocks returns failure
+    final var failureResult =
+        new PeerTaskExecutorResult<Map<SyncBlock, List<SyncTransactionReceipt>>>(
+            Optional.empty(), TIMEOUT, emptyList());
+
+    // Third call (retry) returns the remaining 2 blocks successfully
+    final var secondSuccessResult =
+        new PeerTaskExecutorResult<>(
+            Optional.of(
+                Map.of(
+                    syncBlocks.get(2), allReceiptsPerBlock.get(2),
+                    syncBlocks.get(3), allReceiptsPerBlock.get(3))),
+            SUCCESS,
+            emptyList());
+
+    when(peerTaskExecutor.execute(any(GetSyncReceiptsFromPeerTask.class)))
+        .thenReturn(firstSuccessResult) // Iteration 1: SUCCESS with 2 blocks
+        .thenReturn(failureResult) // Iteration 2: FAILURE on remaining 2 blocks
+        .thenReturn(secondSuccessResult); // Iteration 3: SUCCESS with remaining 2 blocks
+
+    // When: downloading receipts
+    final CompletableFuture<List<SyncBlockWithReceipts>> result =
+        downloadSyncReceiptsStep.apply(syncBlocks);
+
+    // Then: should return all 4 blocks with receipts
+    final List<SyncBlockWithReceipts> blocksWithReceipts = result.get();
+    assertThat(blocksWithReceipts).hasSize(4);
+
+    // Verify all blocks have correct receipts
+    for (int i = 0; i < blocksWithReceipts.size(); i++) {
+      assertThat(blocksWithReceipts.get(i).getBlock()).isEqualTo(syncBlocks.get(i));
+      assertThat(blocksWithReceipts.get(i).getReceipts()).isEqualTo(allReceiptsPerBlock.get(i));
+    }
+
+    // Verify the task was executed 3 times:
+    // 1. Partial success (2 blocks)
+    // 2. Failure on remaining blocks
+    // 3. Success on retry for remaining blocks
+    verify(peerTaskExecutor, times(3)).execute(any(GetSyncReceiptsFromPeerTask.class));
+  }
+
+  @Test
+  public void shouldTimeoutAfterConfiguredDuration() throws Exception {
+    // Given: 1 block with transactions and a very short timeout
+    final List<Block> blocks = gen.blockSequence(2).subList(1, 2);
+    final List<SyncBlock> syncBlocks = blocksToSyncBlocks(blocks);
+
+    // Create a real EthScheduler (not deterministic) to test timeout behavior
+    final EthScheduler realScheduler = new EthScheduler(1, 1, 1, new NoOpMetricsSystem());
+    final EthContext realEthContext = mock(EthContext.class);
+    when(realEthContext.getScheduler()).thenReturn(realScheduler);
+    when(realEthContext.getPeerTaskExecutor()).thenReturn(peerTaskExecutor);
+
+    try {
+      // Create a new instance with a very short timeout for testing (100ms)
+      final DownloadSyncReceiptsStep shortTimeoutStep =
+          new DownloadSyncReceiptsStep(
+              protocolSchedule,
+              realEthContext,
+              syncTransactionReceiptEncoder,
+              Duration.ofMillis(100));
+
+      // Mock continuous failures that would retry indefinitely without timeout
+      final var failureResult =
+          new PeerTaskExecutorResult<Map<SyncBlock, List<SyncTransactionReceipt>>>(
+              Optional.empty(), TIMEOUT, emptyList());
+
+      when(peerTaskExecutor.execute(any(GetSyncReceiptsFromPeerTask.class)))
+          .thenReturn(failureResult);
+
+      // When: downloading receipts with short timeout
+      final CompletableFuture<List<SyncBlockWithReceipts>> result =
+          shortTimeoutStep.apply(syncBlocks);
+
+      // Then: should timeout quickly with TimeoutException (wrapped in ExecutionException)
+      assertThatThrownBy(result::get)
+          .isInstanceOf(ExecutionException.class)
+          .hasCauseInstanceOf(java.util.concurrent.TimeoutException.class);
+    } finally {
+      // Clean up the real scheduler
+      realScheduler.stop();
+      realScheduler.awaitStop();
+    }
+  }
+}
diff --git a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/FastSyncChainDownloaderTest.java b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/FastSyncChainDownloaderTest.java
index 66b44de9f92..960ff23f0ce 100644
--- a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/FastSyncChainDownloaderTest.java
+++ b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/FastSyncChainDownloaderTest.java
@@ -23,7 +23,6 @@
 import org.hyperledger.besu.ethereum.chain.Blockchain;
 import org.hyperledger.besu.ethereum.chain.MutableBlockchain;
 import org.hyperledger.besu.ethereum.core.Block;
-import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.BlockchainSetupUtil;
 import org.hyperledger.besu.ethereum.core.SyncBlock;
 import org.hyperledger.besu.ethereum.core.SyncBlockBody;
@@ -145,14 +144,17 @@ public void setup(final DataStorageFormat storageFormat) {
             (invocationOnMock) -> {
               GetSyncReceiptsFromPeerTask task =
                   invocationOnMock.getArgument(0, GetSyncReceiptsFromPeerTask.class);
-              Map<BlockHeader, List<SyncTransactionReceipt>> getReceiptsFromPeerTaskResult =
+              Map<SyncBlock, List<SyncTransactionReceipt>> getReceiptsFromPeerTaskResult =
                   new HashMap<>();
-              task.getBlockHeaders()
+              task.getRequestedBlocks()
                   .forEach(
-                      (bh) ->
+                      (block) ->
                           getReceiptsFromPeerTaskResult.put(
-                              bh,
-                              otherBlockchain.getTxReceipts(bh.getHash()).get().stream()
+                              block,
+                              otherBlockchain
+                                  .getTxReceipts(block.getHeader().getHash())
+                                  .get()
+                                  .stream()
                                   .map(
                                       (tr) ->
                                           syncTransactionReceiptDecoder.decode(