Terminating at target and not a block later

Signed-off-by: Jiri Peinlich <jiri.peinlich@gmail.com>

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fullsync/FlexibleBlockHashTerminalCondition.java

@@ -1,3 +1,17 @@
+/*
+ * 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.fullsync;
 
 import org.hyperledger.besu.datatypes.Hash;

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fullsync/FullImportBlockStep.java

@@ -35,18 +35,25 @@ public class FullImportBlockStep implements Consumer<Block> {
  private final EthContext ethContext;
  private long gasAccumulator = 0;
  private long lastReportMillis = 0;
+ private final FullSyncTerminationCondition fullSyncTerminationCondition;
 
  public FullImportBlockStep(
  final ProtocolSchedule protocolSchedule,
  final ProtocolContext protocolContext,
- final EthContext ethContext) {
+ final EthContext ethContext,
+ final FullSyncTerminationCondition fullSyncTerminationCondition) {
  this.protocolSchedule = protocolSchedule;
  this.protocolContext = protocolContext;
  this.ethContext = ethContext;
+ this.fullSyncTerminationCondition = fullSyncTerminationCondition;
  }
 
  @Override
  public void accept(final Block block) {
+ if (fullSyncTerminationCondition.shouldStopDownload()) {
+ LOG.info("Not importing another block, because terminal condition was reached.");
+ return;
+ }
  final long blockNumber = block.getHeader().getNumber();
  final String blockHash = block.getHash().toHexString();
  final BlockImporter importer =

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fullsync/FullSyncChainDownloader.java

@@ -48,7 +48,12 @@ public static ChainDownloader create(
  syncState,
  syncTargetManager,
  new FullSyncDownloadPipelineFactory(
- config, protocolSchedule, protocolContext, ethContext, metricsSystem),
+ config,
+ protocolSchedule,
+ protocolContext,
+ ethContext,
+ metricsSystem,
+ terminationCondition),
  ethContext.getScheduler(),
  metricsSystem);
  }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fullsync/FullSyncDownloadPipelineFactory.java

@@ -48,18 +48,21 @@ public class FullSyncDownloadPipelineFactory implements DownloadPipelineFactory
  private final ValidationPolicy detachedValidationPolicy =
  () -> HeaderValidationMode.DETACHED_ONLY;
  private final BetterSyncTargetEvaluator betterSyncTargetEvaluator;
+ private final FullSyncTerminationCondition fullSyncTerminationCondition;
 
  public FullSyncDownloadPipelineFactory(
  final SynchronizerConfiguration syncConfig,
  final ProtocolSchedule protocolSchedule,
  final ProtocolContext protocolContext,
  final EthContext ethContext,
- final MetricsSystem metricsSystem) {
+ final MetricsSystem metricsSystem,
+ final FullSyncTerminationCondition fullSyncTerminationCondition) {
  this.syncConfig = syncConfig;
  this.protocolSchedule = protocolSchedule;
  this.protocolContext = protocolContext;
  this.ethContext = ethContext;
  this.metricsSystem = metricsSystem;
+ this.fullSyncTerminationCondition = fullSyncTerminationCondition;
  betterSyncTargetEvaluator = new BetterSyncTargetEvaluator(syncConfig, ethContext.getEthPeers());
  }
 
@@ -91,7 +94,8 @@ public Pipeline<?> createDownloadPipelineForSyncTarget(final SyncTarget target)
  new DownloadBodiesStep(protocolSchedule, ethContext, metricsSystem);
  final ExtractTxSignaturesStep extractTxSignaturesStep = new ExtractTxSignaturesStep();
  final FullImportBlockStep importBlockStep =
- new FullImportBlockStep(protocolSchedule, protocolContext, ethContext);
+ new FullImportBlockStep(
+ protocolSchedule, protocolContext, ethContext, fullSyncTerminationCondition);
 
  return PipelineBuilder.createPipelineFrom(
  "fetchCheckpoints",
@@ -115,18 +119,19 @@ public Pipeline<?> createDownloadPipelineForSyncTarget(final SyncTarget target)
 
  private boolean shouldContinueDownloadingFromPeer(
  final EthPeer peer, final BlockHeader lastCheckpointHeader) {
+ final boolean shouldTerminate = fullSyncTerminationCondition.getAsBoolean();
  final boolean caughtUpToPeer =
  peer.chainState().getEstimatedHeight() <= lastCheckpointHeader.getNumber();
  final boolean isDisconnected = peer.isDisconnected();
  final boolean shouldSwitchSyncTarget = betterSyncTargetEvaluator.shouldSwitchSyncTarget(peer);
-
  LOG.debug(
- "shouldContinueDownloadingFromPeer? {}, disconnected {}, caughtUp {}, shouldSwitchSyncTarget {}",
+ "shouldTerminate {}, shouldContinueDownloadingFromPeer? {}, disconnected {}, caughtUp {}, shouldSwitchSyncTarget {}",
+ shouldTerminate,
  peer,
  isDisconnected,
  caughtUpToPeer,
  shouldSwitchSyncTarget);
 
- return !isDisconnected && !caughtUpToPeer && !shouldSwitchSyncTarget;
+ return !shouldTerminate && !isDisconnected && !caughtUpToPeer && !shouldSwitchSyncTarget;
  }
 }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fullsync/FullSyncTargetManager.java

@@ -108,6 +108,6 @@ private boolean isSyncTargetReached(final EthPeer peer) {
 
  @Override
  public boolean shouldContinueDownloading() {
- return terminationCondition.getAsBoolean();
+ return terminationCondition.shouldContinueDownload();
  }
 }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fullsync/FullSyncTerminationCondition.java

@@ -22,21 +22,60 @@
 
 import org.apache.tuweni.units.bigints.UInt256;
 
+/** return true when termination condition is fullfilled and the full sync should stop */
 public interface FullSyncTerminationCondition extends BooleanSupplier {
+
+ default boolean shouldContinueDownload() {
+ return !shouldStopDownload();
+ }
+
+ default boolean shouldStopDownload() {
+ return getAsBoolean();
+ }
+
+ /**
+ * When we want full sync to continue forever (for instance when we don't want to merge)
+ *
+ * @return always false therefore continues forever *
+ */
  static FullSyncTerminationCondition never() {
- return () -> true;
+ return () -> false;
  }
 
+ /**
+ * When we want full sync to finish after reaching a difficulty. For instance when we merge on
+ * total terminal difficulty.
+ *
+ * @param targetDifficulty target difficulty to reach
+ * @param blockchain blockchain to reach the difficulty on
+ * @return true when blockchain reaches difficulty
+ */
  static FullSyncTerminationCondition difficulty(
- final UInt256 difficulty, final Blockchain blockchain) {
- return difficulty(Difficulty.of(difficulty), blockchain);
+ final UInt256 targetDifficulty, final Blockchain blockchain) {
+ return difficulty(Difficulty.of(targetDifficulty), blockchain);
  }
 
+ /**
+ * When we want full sync to finish after reaching a difficulty. For instance when we merge on
+ * total terminal difficulty.
+ *
+ * @param targetDifficulty target difficulty to reach
+ * @param blockchain blockchain to reach the difficulty on*
+ * @return true when blockchain reaches difficulty
+ */
  static FullSyncTerminationCondition difficulty(
- final Difficulty difficulty, final Blockchain blockchain) {
- return () -> difficulty.greaterThan(blockchain.getChainHead().getTotalDifficulty());
+ final Difficulty targetDifficulty, final Blockchain blockchain) {
+ return () -> blockchain.getChainHead().getTotalDifficulty().greaterThan(targetDifficulty);
  }
 
+ /**
+ * When we want the full sync to finish on a target hash. For instance when we reach a merge
+ * checkpoint.
+ *
+ * @param blockHash target hash to look for
+ * @param blockchain blockchain to reach the difficulty on
+ * @return true when blockchain contains target hash (target hash can be changed)
+ */
  static FlexibleBlockHashTerminalCondition blockHash(
  final Hash blockHash, final Blockchain blockchain) {
  return new FlexibleBlockHashTerminalCondition(blockHash, blockchain);

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/PipelineChainDownloaderTest.java

@@ -158,7 +158,6 @@ public void shouldBeCompleteWhenPipelineCompletesAndSyncTargetManagerShouldNotCo
 
  verify(syncTargetManager).findSyncTarget();
 
- when(syncTargetManager.shouldContinueDownloading()).thenReturn(false);
  pipelineFuture.complete(null);
 
  verify(syncTargetManager, Mockito.times(2)).shouldContinueDownloading();

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/fullsync/FullImportBlockStepTest.java

@@ -50,7 +50,9 @@ public void setUp() {
  when(protocolSchedule.getByBlockNumber(anyLong())).thenReturn(protocolSpec);
  when(protocolSpec.getBlockImporter()).thenReturn(blockImporter);
 
- importBlocksStep = new FullImportBlockStep(protocolSchedule, protocolContext, null);
+ importBlocksStep =
+ new FullImportBlockStep(
+ protocolSchedule, protocolContext, null, FullSyncTerminationCondition.never());
  }
 
  @Test