From 189d634889f1c063e51ee368bb816d2f94a69521 Mon Sep 17 00:00:00 2001
From: Jason Frame <jason.frame@consensys.net>
Date: Tue, 7 May 2024 11:36:04 +1000
Subject: [PATCH] World state halt and chain halt fixes (#7027) (#7059)

Signed-off-by: Jason Frame <jason.frame@consensys.net>
Signed-off-by: stefan.pingel@consensys.net <stefan.pingel@consensys.net>
---
 CHANGELOG.md                                  |  1 +
 .../options/unstable/SynchronizerOptions.java | 13 ++---
 .../cli/options/SynchronizerOptionsTest.java  |  2 +-
 .../eth/sync/SynchronizerConfiguration.java   | 20 ++++----
 .../FastSyncDownloadPipelineFactory.java      |  2 +-
 .../fastsync/PivotSelectorFromSafeBlock.java  |  3 +-
 .../FullSyncDownloadPipelineFactory.java      |  2 +-
 .../eth/sync/range/SyncTargetRangeSource.java | 37 +++++++++-----
 .../snapsync/DynamicPivotBlockSelector.java   |  2 +-
 .../sync/snapsync/SnapWorldDownloadState.java | 48 ++++++++++-------
 .../SnapWorldStateDownloadProcess.java        | 51 +++++++++++++++++++
 .../snapsync/SnapWorldStateDownloader.java    |  4 +-
 .../snapsync/SnapWorldDownloadStateTest.java  | 17 ++++++-
 13 files changed, 145 insertions(+), 57 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7bcabc7592..3130fa01fb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -55,6 +55,7 @@
 - Snap client fixes discovered during snap server testing [#6847](https://github.com/hyperledger/besu/pull/6847)
 - Correctly initialize the txpool as disabled on creation [#6890](https://github.com/hyperledger/besu/pull/6890)
 - Fix worldstate download halt when using snap sync during initial sync [#6981](https://github.com/hyperledger/besu/pull/6981)
+- Fix chain halt due to peers only partially responding with headers. And worldstate halts caused by a halt in the chain sync [#7027](https://github.com/hyperledger/besu/pull/7027)
 
 ### Download Links
 
diff --git a/besu/src/main/java/org/hyperledger/besu/cli/options/unstable/SynchronizerOptions.java b/besu/src/main/java/org/hyperledger/besu/cli/options/unstable/SynchronizerOptions.java
index da5199f85d..1adf708058 100644
--- a/besu/src/main/java/org/hyperledger/besu/cli/options/unstable/SynchronizerOptions.java
+++ b/besu/src/main/java/org/hyperledger/besu/cli/options/unstable/SynchronizerOptions.java
@@ -39,6 +39,8 @@ public class SynchronizerOptions implements CLIOptions<SynchronizerConfiguration
       "--Xsynchronizer-downloader-header-request-size";
   private static final String DOWNLOADER_CHECKPOINT_TIMEOUTS_PERMITTED_FLAG =
       "--Xsynchronizer-downloader-checkpoint-timeouts-permitted";
+  private static final String DOWNLOADER_CHECKPOINT_RETRIES_FLAG =
+      "--Xsynchronizer-downloader-checkpoint-RETRIES";
   private static final String DOWNLOADER_CHAIN_SEGMENT_SIZE_FLAG =
       "--Xsynchronizer-downloader-chain-segment-size";
   private static final String DOWNLOADER_PARALLELISM_FLAG =
@@ -132,12 +134,12 @@ public class SynchronizerOptions implements CLIOptions<SynchronizerConfiguration
       SynchronizerConfiguration.DEFAULT_DOWNLOADER_HEADER_REQUEST_SIZE;
 
   @CommandLine.Option(
-      names = DOWNLOADER_CHECKPOINT_TIMEOUTS_PERMITTED_FLAG,
+      names = {DOWNLOADER_CHECKPOINT_RETRIES_FLAG, DOWNLOADER_CHECKPOINT_TIMEOUTS_PERMITTED_FLAG},
       hidden = true,
       paramLabel = "<INTEGER>",
       description =
           "Number of tries to attempt to download checkpoints before stopping (default: ${DEFAULT-VALUE})")
-  private int downloaderCheckpointTimeoutsPermitted =
+  private int downloaderCheckpointRetries =
       SynchronizerConfiguration.DEFAULT_DOWNLOADER_CHECKPOINT_TIMEOUTS_PERMITTED;
 
   @CommandLine.Option(
@@ -354,8 +356,7 @@ public class SynchronizerOptions implements CLIOptions<SynchronizerConfiguration
         config.getDownloaderChangeTargetThresholdByHeight();
     options.downloaderChangeTargetThresholdByTd = config.getDownloaderChangeTargetThresholdByTd();
     options.downloaderHeaderRequestSize = config.getDownloaderHeaderRequestSize();
-    options.downloaderCheckpointTimeoutsPermitted =
-        config.getDownloaderCheckpointTimeoutsPermitted();
+    options.downloaderCheckpointRetries = config.getDownloaderCheckpointRetries();
     options.downloaderChainSegmentSize = config.getDownloaderChainSegmentSize();
     options.downloaderParallelism = config.getDownloaderParallelism();
     options.transactionsParallelism = config.getTransactionsParallelism();
@@ -394,7 +395,7 @@ public class SynchronizerOptions implements CLIOptions<SynchronizerConfiguration
     builder.downloaderChangeTargetThresholdByHeight(downloaderChangeTargetThresholdByHeight);
     builder.downloaderChangeTargetThresholdByTd(downloaderChangeTargetThresholdByTd);
     builder.downloaderHeadersRequestSize(downloaderHeaderRequestSize);
-    builder.downloaderCheckpointTimeoutsPermitted(downloaderCheckpointTimeoutsPermitted);
+    builder.downloaderCheckpointRetries(downloaderCheckpointRetries);
     builder.downloaderChainSegmentSize(downloaderChainSegmentSize);
     builder.downloaderParallelism(downloaderParallelism);
     builder.transactionsParallelism(transactionsParallelism);
@@ -436,7 +437,7 @@ public class SynchronizerOptions implements CLIOptions<SynchronizerConfiguration
             DOWNLOADER_HEADER_REQUEST_SIZE_FLAG,
             OptionParser.format(downloaderHeaderRequestSize),
             DOWNLOADER_CHECKPOINT_TIMEOUTS_PERMITTED_FLAG,
-            OptionParser.format(downloaderCheckpointTimeoutsPermitted),
+            OptionParser.format(downloaderCheckpointRetries),
             DOWNLOADER_CHAIN_SEGMENT_SIZE_FLAG,
             OptionParser.format(downloaderChainSegmentSize),
             DOWNLOADER_PARALLELISM_FLAG,
diff --git a/besu/src/test/java/org/hyperledger/besu/cli/options/SynchronizerOptionsTest.java b/besu/src/test/java/org/hyperledger/besu/cli/options/SynchronizerOptionsTest.java
index 9585fadf0d..ce2a4d0d55 100644
--- a/besu/src/test/java/org/hyperledger/besu/cli/options/SynchronizerOptionsTest.java
+++ b/besu/src/test/java/org/hyperledger/besu/cli/options/SynchronizerOptionsTest.java
@@ -60,7 +60,7 @@ public class SynchronizerOptionsTest
             SynchronizerConfiguration.DEFAULT_DOWNLOADER_CHANGE_TARGET_THRESHOLD_BY_TD.add(2L))
         .downloaderHeadersRequestSize(
             SynchronizerConfiguration.DEFAULT_DOWNLOADER_HEADER_REQUEST_SIZE + 2)
-        .downloaderCheckpointTimeoutsPermitted(
+        .downloaderCheckpointRetries(
             SynchronizerConfiguration.DEFAULT_DOWNLOADER_CHECKPOINT_TIMEOUTS_PERMITTED + 2)
         .downloaderChainSegmentSize(
             SynchronizerConfiguration.DEFAULT_DOWNLOADER_CHAIN_SEGMENT_SIZE + 2)
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 2a6ec11ed6..28f8bee6db 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
@@ -77,7 +77,7 @@ public class SynchronizerConfiguration {
   private final long downloaderChangeTargetThresholdByHeight;
   private final UInt256 downloaderChangeTargetThresholdByTd;
   private final int downloaderHeaderRequestSize;
-  private final int downloaderCheckpointTimeoutsPermitted;
+  private final int downloaderCheckpointRetries;
   private final int downloaderChainSegmentSize;
   private final int downloaderParallelism;
   private final int transactionsParallelism;
@@ -101,7 +101,7 @@ public class SynchronizerConfiguration {
       final long downloaderChangeTargetThresholdByHeight,
       final UInt256 downloaderChangeTargetThresholdByTd,
       final int downloaderHeaderRequestSize,
-      final int downloaderCheckpointTimeoutsPermitted,
+      final int downloaderCheckpointRetries,
       final int downloaderChainSegmentSize,
       final int downloaderParallelism,
       final int transactionsParallelism,
@@ -123,7 +123,7 @@ public class SynchronizerConfiguration {
     this.downloaderChangeTargetThresholdByHeight = downloaderChangeTargetThresholdByHeight;
     this.downloaderChangeTargetThresholdByTd = downloaderChangeTargetThresholdByTd;
     this.downloaderHeaderRequestSize = downloaderHeaderRequestSize;
-    this.downloaderCheckpointTimeoutsPermitted = downloaderCheckpointTimeoutsPermitted;
+    this.downloaderCheckpointRetries = downloaderCheckpointRetries;
     this.downloaderChainSegmentSize = downloaderChainSegmentSize;
     this.downloaderParallelism = downloaderParallelism;
     this.transactionsParallelism = transactionsParallelism;
@@ -191,8 +191,8 @@ public class SynchronizerConfiguration {
     return downloaderHeaderRequestSize;
   }
 
-  public int getDownloaderCheckpointTimeoutsPermitted() {
-    return downloaderCheckpointTimeoutsPermitted;
+  public int getDownloaderCheckpointRetries() {
+    return downloaderCheckpointRetries;
   }
 
   public int getDownloaderChainSegmentSize() {
@@ -264,8 +264,7 @@ public class SynchronizerConfiguration {
     private UInt256 downloaderChangeTargetThresholdByTd =
         DEFAULT_DOWNLOADER_CHANGE_TARGET_THRESHOLD_BY_TD;
     private int downloaderHeaderRequestSize = DEFAULT_DOWNLOADER_HEADER_REQUEST_SIZE;
-    private int downloaderCheckpointTimeoutsPermitted =
-        DEFAULT_DOWNLOADER_CHECKPOINT_TIMEOUTS_PERMITTED;
+    private int downloaderCheckpointRetries = DEFAULT_DOWNLOADER_CHECKPOINT_TIMEOUTS_PERMITTED;
     private SnapSyncConfiguration snapSyncConfiguration = SnapSyncConfiguration.getDefault();
     private int downloaderChainSegmentSize = DEFAULT_DOWNLOADER_CHAIN_SEGMENT_SIZE;
     private int downloaderParallelism = DEFAULT_DOWNLOADER_PARALLELISM;
@@ -327,9 +326,8 @@ public class SynchronizerConfiguration {
       return this;
     }
 
-    public Builder downloaderCheckpointTimeoutsPermitted(
-        final int downloaderCheckpointTimeoutsPermitted) {
-      this.downloaderCheckpointTimeoutsPermitted = downloaderCheckpointTimeoutsPermitted;
+    public Builder downloaderCheckpointRetries(final int downloaderCheckpointRetries) {
+      this.downloaderCheckpointRetries = downloaderCheckpointRetries;
       return this;
     }
 
@@ -422,7 +420,7 @@ public class SynchronizerConfiguration {
           downloaderChangeTargetThresholdByHeight,
           downloaderChangeTargetThresholdByTd,
           downloaderHeaderRequestSize,
-          downloaderCheckpointTimeoutsPermitted,
+          downloaderCheckpointRetries,
           downloaderChainSegmentSize,
           downloaderParallelism,
           transactionsParallelism,
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 a458e9bd51..07e964426c 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
@@ -125,7 +125,7 @@ public class FastSyncDownloadPipelineFactory implements DownloadPipelineFactory
             ethContext.getScheduler(),
             target.peer(),
             getCommonAncestor(target),
-            syncConfig.getDownloaderCheckpointTimeoutsPermitted(),
+            syncConfig.getDownloaderCheckpointRetries(),
             SyncTerminationCondition.never());
     final DownloadHeadersStep downloadHeadersStep =
         new DownloadHeadersStep(
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/PivotSelectorFromSafeBlock.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/PivotSelectorFromSafeBlock.java
index 052bee29fa..21150c8c1c 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/PivotSelectorFromSafeBlock.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/PivotSelectorFromSafeBlock.java
@@ -28,6 +28,7 @@ import org.hyperledger.besu.plugin.services.MetricsSystem;
 
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
 import org.slf4j.Logger;
@@ -127,7 +128,7 @@ public class PivotSelectorFromSafeBlock implements PivotBlockSelector {
                                           maybeCachedHeadBlockHeader = Optional.of(blockHeader);
                                           return blockHeader.getNumber();
                                         })
-                                    .get();
+                                    .get(20, TimeUnit.SECONDS);
                               } catch (Throwable t) {
                                 LOG.debug(
                                     "Error trying to download chain head block header by hash {}",
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fullsync/FullSyncDownloadPipelineFactory.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fullsync/FullSyncDownloadPipelineFactory.java
index 27f50eead8..b822236b8f 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fullsync/FullSyncDownloadPipelineFactory.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fullsync/FullSyncDownloadPipelineFactory.java
@@ -91,7 +91,7 @@ public class FullSyncDownloadPipelineFactory implements DownloadPipelineFactory
             ethContext.getScheduler(),
             target.peer(),
             target.commonAncestor(),
-            syncConfig.getDownloaderCheckpointTimeoutsPermitted(),
+            syncConfig.getDownloaderCheckpointRetries(),
             fullSyncTerminationCondition);
     final DownloadHeadersStep downloadHeadersStep =
         new DownloadHeadersStep(
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/range/SyncTargetRangeSource.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/range/SyncTargetRangeSource.java
index fbeac02f40..8c6186448e 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/range/SyncTargetRangeSource.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/range/SyncTargetRangeSource.java
@@ -22,6 +22,7 @@ import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
 import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.sync.fullsync.SyncTerminationCondition;
+import org.hyperledger.besu.ethereum.p2p.rlpx.wire.messages.DisconnectMessage;
 
 import java.time.Duration;
 import java.util.ArrayDeque;
@@ -44,7 +45,7 @@ public class SyncTargetRangeSource implements Iterator<SyncTargetRange> {
   private final SyncTargetChecker syncTargetChecker;
   private final EthPeer peer;
   private final EthScheduler ethScheduler;
-  private final int rangeTimeoutsPermitted;
+  private final int retriesPermitted;
   private final Duration newHeaderWaitDuration;
   private final SyncTerminationCondition terminationCondition;
 
@@ -52,7 +53,7 @@ public class SyncTargetRangeSource implements Iterator<SyncTargetRange> {
   private BlockHeader lastRangeEnd;
   private boolean reachedEndOfRanges = false;
   private Optional<CompletableFuture<List<BlockHeader>>> pendingRequests = Optional.empty();
-  private int requestFailureCount = 0;
+  private int retryCount = 0;
 
   public SyncTargetRangeSource(
       final RangeHeadersFetcher fetcher,
@@ -60,7 +61,7 @@ public class SyncTargetRangeSource implements Iterator<SyncTargetRange> {
       final EthScheduler ethScheduler,
       final EthPeer peer,
       final BlockHeader commonAncestor,
-      final int rangeTimeoutsPermitted,
+      final int retriesPermitted,
       final SyncTerminationCondition terminationCondition) {
     this(
         fetcher,
@@ -68,7 +69,7 @@ public class SyncTargetRangeSource implements Iterator<SyncTargetRange> {
         ethScheduler,
         peer,
         commonAncestor,
-        rangeTimeoutsPermitted,
+        retriesPermitted,
         Duration.ofSeconds(5),
         terminationCondition);
   }
@@ -79,7 +80,7 @@ public class SyncTargetRangeSource implements Iterator<SyncTargetRange> {
       final EthScheduler ethScheduler,
       final EthPeer peer,
       final BlockHeader commonAncestor,
-      final int rangeTimeoutsPermitted,
+      final int retriesPermitted,
       final Duration newHeaderWaitDuration,
       final SyncTerminationCondition terminationCondition) {
     this.fetcher = fetcher;
@@ -87,7 +88,7 @@ public class SyncTargetRangeSource implements Iterator<SyncTargetRange> {
     this.ethScheduler = ethScheduler;
     this.peer = peer;
     this.lastRangeEnd = commonAncestor;
-    this.rangeTimeoutsPermitted = rangeTimeoutsPermitted;
+    this.retriesPermitted = retriesPermitted;
     this.newHeaderWaitDuration = newHeaderWaitDuration;
     this.terminationCondition = terminationCondition;
   }
@@ -96,7 +97,7 @@ public class SyncTargetRangeSource implements Iterator<SyncTargetRange> {
   public boolean hasNext() {
     return terminationCondition.shouldContinueDownload()
         && (!retrievedRanges.isEmpty()
-            || (requestFailureCount < rangeTimeoutsPermitted
+            || (retryCount < retriesPermitted
                 && syncTargetChecker.shouldContinueDownloadingFromSyncTarget(peer, lastRangeEnd)
                 && !reachedEndOfRanges));
   }
@@ -148,13 +149,21 @@ public class SyncTargetRangeSource implements Iterator<SyncTargetRange> {
           pendingRequest.get(newHeaderWaitDuration.toMillis(), MILLISECONDS);
       this.pendingRequests = Optional.empty();
       if (newHeaders.isEmpty()) {
-        requestFailureCount++;
+        retryCount++;
+        if (retryCount >= retriesPermitted) {
+          LOG.atDebug()
+              .setMessage(
+                  "Disconnecting target peer for providing useless or empty range header: {}.")
+              .addArgument(peer)
+              .log();
+          peer.disconnect(DisconnectMessage.DisconnectReason.USELESS_PEER_USELESS_RESPONSES);
+        }
       } else {
-        requestFailureCount = 0;
-      }
-      for (final BlockHeader header : newHeaders) {
-        retrievedRanges.add(new SyncTargetRange(peer, lastRangeEnd, header));
-        lastRangeEnd = header;
+        retryCount = 0;
+        for (final BlockHeader header : newHeaders) {
+          retrievedRanges.add(new SyncTargetRange(peer, lastRangeEnd, header));
+          lastRangeEnd = header;
+        }
       }
       return retrievedRanges.poll();
     } catch (final InterruptedException e) {
@@ -163,7 +172,7 @@ public class SyncTargetRangeSource implements Iterator<SyncTargetRange> {
     } catch (final ExecutionException e) {
       LOG.debug("Failed to retrieve new range headers", e);
       this.pendingRequests = Optional.empty();
-      requestFailureCount++;
+      retryCount++;
       return null;
     } catch (final TimeoutException e) {
       return null;
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/DynamicPivotBlockSelector.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/DynamicPivotBlockSelector.java
index 1e3d6bb2a5..3961f72b3e 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/DynamicPivotBlockSelector.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/DynamicPivotBlockSelector.java
@@ -161,7 +161,7 @@ public class DynamicPivotBlockSelector {
                   .addArgument(this::logLastPivotBlockFound)
                   .log();
             })
-        .orTimeout(5, TimeUnit.MINUTES);
+        .orTimeout(20, TimeUnit.SECONDS);
   }
 
   private boolean isSamePivotBlock(final FastSyncState fss) {
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldDownloadState.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldDownloadState.java
index 1b6567588b..df5ac18f45 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldDownloadState.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldDownloadState.java
@@ -21,6 +21,7 @@ import static org.hyperledger.besu.ethereum.worldstate.WorldStateStorageCoordina
 import org.hyperledger.besu.ethereum.chain.BlockAddedObserver;
 import org.hyperledger.besu.ethereum.chain.Blockchain;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
+import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.sync.snapsync.context.SnapSyncStatePersistenceManager;
 import org.hyperledger.besu.ethereum.eth.sync.snapsync.request.AccountRangeDataRequest;
 import org.hyperledger.besu.ethereum.eth.sync.snapsync.request.BytecodeRequest;
@@ -86,6 +87,7 @@ public class SnapWorldDownloadState extends WorldDownloadState<SnapDataRequest>
   // blockchain
   private final Blockchain blockchain;
   private final Long blockObserverId;
+  private final EthContext ethContext;
 
   // metrics around the snapsync
   private final SnapSyncMetricsManager metricsManager;
@@ -99,7 +101,8 @@ public class SnapWorldDownloadState extends WorldDownloadState<SnapDataRequest>
       final int maxRequestsWithoutProgress,
       final long minMillisBeforeStalling,
       final SnapSyncMetricsManager metricsManager,
-      final Clock clock) {
+      final Clock clock,
+      final EthContext ethContext) {
     super(
         worldStateStorageCoordinator,
         pendingRequests,
@@ -111,6 +114,7 @@ public class SnapWorldDownloadState extends WorldDownloadState<SnapDataRequest>
     this.snapSyncState = snapSyncState;
     this.metricsManager = metricsManager;
     this.blockObserverId = blockchain.observeBlockAdded(createBlockchainObserver());
+    this.ethContext = ethContext;
 
     metricsManager
         .getMetricsSystem()
@@ -423,23 +427,29 @@ public class SnapWorldDownloadState extends WorldDownloadState<SnapDataRequest>
   }
 
   public BlockAddedObserver createBlockchainObserver() {
-    return addedBlockContext -> {
-      final AtomicBoolean foundNewPivotBlock = new AtomicBoolean(false);
-      pivotBlockSelector.check(
-          (____, isNewPivotBlock) -> {
-            if (isNewPivotBlock) {
-              foundNewPivotBlock.set(true);
-            }
-          });
-
-      final boolean isNewPivotBlockFound = foundNewPivotBlock.get();
-      final boolean isBlockchainCaughtUp =
-          snapSyncState.isWaitingBlockchain() && !pivotBlockSelector.isBlockchainBehind();
-
-      if (snapSyncState.isHealTrieInProgress() && (isNewPivotBlockFound || isBlockchainCaughtUp)) {
-        snapSyncState.setWaitingBlockchain(false);
-        reloadTrieHeal();
-      }
-    };
+    return addedBlockContext ->
+        ethContext
+            .getScheduler()
+            .executeServiceTask(
+                () -> {
+                  final AtomicBoolean foundNewPivotBlock = new AtomicBoolean(false);
+                  pivotBlockSelector.check(
+                      (____, isNewPivotBlock) -> {
+                        if (isNewPivotBlock) {
+                          foundNewPivotBlock.set(true);
+                        }
+                      });
+
+                  final boolean isNewPivotBlockFound = foundNewPivotBlock.get();
+                  final boolean isBlockchainCaughtUp =
+                      snapSyncState.isWaitingBlockchain()
+                          && !pivotBlockSelector.isBlockchainBehind();
+
+                  if (snapSyncState.isHealTrieInProgress()
+                      && (isNewPivotBlockFound || isBlockchainCaughtUp)) {
+                    snapSyncState.setWaitingBlockchain(false);
+                    reloadTrieHeal();
+                  }
+                });
   }
 }
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldStateDownloadProcess.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldStateDownloadProcess.java
index ddab968804..c19ae6facc 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldStateDownloadProcess.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldStateDownloadProcess.java
@@ -15,6 +15,7 @@
 package org.hyperledger.besu.ethereum.eth.sync.snapsync;
 
 import static com.google.common.base.Preconditions.checkNotNull;
+import static org.hyperledger.besu.ethereum.eth.sync.snapsync.DynamicPivotBlockSelector.doNothingOnPivotChange;
 import static org.hyperledger.besu.services.pipeline.PipelineBuilder.createPipelineFrom;
 
 import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
@@ -158,12 +159,19 @@ public class SnapWorldStateDownloadProcess implements WorldStateDownloadProcess
     private SnapSyncProcessState snapSyncState;
     private PersistDataStep persistDataStep;
     private CompleteTaskStep completeTaskStep;
+    private DynamicPivotBlockSelector pivotBlockManager;
 
     public Builder configuration(final SnapSyncConfiguration snapSyncConfiguration) {
       this.snapSyncConfiguration = snapSyncConfiguration;
       return this;
     }
 
+    public Builder dynamicPivotBlockSelector(
+        final DynamicPivotBlockSelector dynamicPivotBlockSelector) {
+      this.pivotBlockManager = dynamicPivotBlockSelector;
+      return this;
+    }
+
     public Builder maxOutstandingRequests(final int maxOutstandingRequests) {
       this.maxOutstandingRequests = maxOutstandingRequests;
       return this;
@@ -257,6 +265,12 @@ public class SnapWorldStateDownloadProcess implements WorldStateDownloadProcess
                   outputCounter,
                   true,
                   "world_state_download")
+              .thenProcess(
+                  "checkNewPivotBlock-Account",
+                  tasks -> {
+                    pivotBlockManager.check(doNothingOnPivotChange);
+                    return tasks;
+                  })
               .thenProcessAsync(
                   "batchDownloadAccountData",
                   requestTask -> requestDataStep.requestAccount(requestTask),
@@ -274,6 +288,12 @@ public class SnapWorldStateDownloadProcess implements WorldStateDownloadProcess
                   true,
                   "world_state_download")
               .inBatches(snapSyncConfiguration.getStorageCountPerRequest())
+              .thenProcess(
+                  "checkNewPivotBlock-Storage",
+                  tasks -> {
+                    pivotBlockManager.check(doNothingOnPivotChange);
+                    return tasks;
+                  })
               .thenProcessAsyncOrdered(
                   "batchDownloadStorageData",
                   requestTask -> requestDataStep.requestStorage(requestTask),
@@ -294,6 +314,12 @@ public class SnapWorldStateDownloadProcess implements WorldStateDownloadProcess
                   outputCounter,
                   true,
                   "world_state_download")
+              .thenProcess(
+                  "checkNewPivotBlock-LargeStorage",
+                  tasks -> {
+                    pivotBlockManager.check(doNothingOnPivotChange);
+                    return tasks;
+                  })
               .thenProcessAsyncOrdered(
                   "batchDownloadLargeStorageData",
                   requestTask -> requestDataStep.requestStorage(List.of(requestTask)),
@@ -328,6 +354,14 @@ public class SnapWorldStateDownloadProcess implements WorldStateDownloadProcess
                                   .map(BytecodeRequest::getCodeHash)
                                   .distinct()
                                   .count())
+              .thenProcess(
+                  "checkNewPivotBlock-Code",
+                  tasks -> {
+                    pivotBlockManager.check(
+                        (blockHeader, newBlockFound) ->
+                            reloadHealWhenNeeded(snapSyncState, downloadState, newBlockFound));
+                    return tasks;
+                  })
               .thenProcessAsyncOrdered(
                   "batchDownloadCodeData",
                   tasks -> requestDataStep.requestCode(tasks),
@@ -356,6 +390,14 @@ public class SnapWorldStateDownloadProcess implements WorldStateDownloadProcess
                   3,
                   bufferCapacity)
               .inBatches(snapSyncConfiguration.getTrienodeCountPerRequest())
+              .thenProcess(
+                  "checkNewPivotBlock-TrieNode",
+                  tasks -> {
+                    pivotBlockManager.check(
+                        (blockHeader, newBlockFound) ->
+                            reloadHealWhenNeeded(snapSyncState, downloadState, newBlockFound));
+                    return tasks;
+                  })
               .thenProcessAsync(
                   "batchDownloadTrieNodeData",
                   tasks -> requestDataStep.requestTrieNodeByPath(tasks),
@@ -419,4 +461,13 @@ public class SnapWorldStateDownloadProcess implements WorldStateDownloadProcess
           requestsToComplete);
     }
   }
+
+  private static void reloadHealWhenNeeded(
+      final SnapSyncProcessState snapSyncState,
+      final SnapWorldDownloadState downloadState,
+      final boolean newBlockFound) {
+    if (snapSyncState.isHealTrieInProgress() && newBlockFound) {
+      downloadState.reloadTrieHeal();
+    }
+  }
 }
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldStateDownloader.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldStateDownloader.java
index 0e0e1b1ac3..cf2f1a4bf6 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldStateDownloader.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldStateDownloader.java
@@ -147,7 +147,8 @@ public class SnapWorldStateDownloader implements WorldStateDownloader {
               maxNodeRequestsWithoutProgress,
               minMillisBeforeStalling,
               snapsyncMetricsManager,
-              clock);
+              clock,
+              ethContext);
 
       final Map<Bytes32, Bytes32> ranges = RangeManager.generateAllRanges(16);
       snapsyncMetricsManager.initRange(ranges);
@@ -214,6 +215,7 @@ public class SnapWorldStateDownloader implements WorldStateDownloader {
           SnapWorldStateDownloadProcess.builder()
               .configuration(snapSyncConfiguration)
               .maxOutstandingRequests(maxOutstandingRequests)
+              .dynamicPivotBlockSelector(dynamicPivotBlockManager)
               .loadLocalDataStep(
                   new LoadLocalDataStep(
                       worldStateStorageCoordinator,
diff --git a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldDownloadStateTest.java b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldDownloadStateTest.java
index 35c5de5dac..011027813d 100644
--- a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldDownloadStateTest.java
+++ b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/SnapWorldDownloadStateTest.java
@@ -33,6 +33,8 @@ import org.hyperledger.besu.ethereum.core.BlockBody;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.BlockHeaderTestFixture;
 import org.hyperledger.besu.ethereum.core.InMemoryKeyValueStorageProvider;
+import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.manager.task.EthTask;
 import org.hyperledger.besu.ethereum.eth.sync.snapsync.context.SnapSyncStatePersistenceManager;
 import org.hyperledger.besu.ethereum.eth.sync.snapsync.request.BytecodeRequest;
@@ -90,6 +92,7 @@ public class SnapWorldDownloadStateTest {
   private final Blockchain blockchain = mock(Blockchain.class);
   private final DynamicPivotBlockSelector dynamicPivotBlockManager =
       mock(DynamicPivotBlockSelector.class);
+  private final EthContext ethContext = mock(EthContext.class);
 
   private final TestClock clock = new TestClock();
   private SnapWorldDownloadState downloadState;
@@ -132,7 +135,8 @@ public class SnapWorldDownloadStateTest {
             MAX_REQUESTS_WITHOUT_PROGRESS,
             MIN_MILLIS_BEFORE_STALLING,
             metricsManager,
-            clock);
+            clock,
+            ethContext);
     final DynamicPivotBlockSelector dynamicPivotBlockManager =
         mock(DynamicPivotBlockSelector.class);
     doAnswer(
@@ -147,6 +151,17 @@ public class SnapWorldDownloadStateTest {
     downloadState.setRootNodeData(ROOT_NODE_DATA);
     future = downloadState.getDownloadFuture();
     assertThat(downloadState.isDownloading()).isTrue();
+
+    final EthScheduler ethScheduler = mock(EthScheduler.class);
+    when(ethContext.getScheduler()).thenReturn(ethScheduler);
+    doAnswer(
+            invocation -> {
+              Runnable runnable = invocation.getArgument(0);
+              runnable.run();
+              return null;
+            })
+        .when(ethScheduler)
+        .executeServiceTask(any(Runnable.class));
   }
 
   @ParameterizedTest