Backward sync exception improvements

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Additions and Improvements
 - Add a block to the bad blocks if it did not descend from the terminal block [#4080](https://github.com/hyperledger/besu/pull/4080)
+- Backward sync exception improvements [#4092](https://github.com/hyperledger/besu/pull/4092)
 
 ### Bug Fixes
 - Return the correct latest valid hash in case of bad block when calling engine methods [#4056](https://github.com/hyperledger/besu/pull/4056)

consensus/merge/src/main/java/org/hyperledger/besu/consensus/merge/blockcreation/MergeCoordinator.java

@@ -217,7 +217,9 @@ public Optional<BlockHeader> getOrSyncHeaderByHash(final Hash blockHash) {
       debugLambda(LOG, "BlockHeader {} is already present", () -> optHeader.get().toLogString());
     } else {
       debugLambda(LOG, "appending block hash {} to backward sync", blockHash::toHexString);
-      backwardSyncContext.syncBackwardsUntil(blockHash);
+      backwardSyncContext
+          .syncBackwardsUntil(blockHash)
+          .exceptionally(e -> logSyncException(blockHash, e));
     }
     return optHeader;
   }
@@ -233,11 +235,18 @@ public Optional<BlockHeader> getOrSyncHeaderByHash(
     } else {
       debugLambda(LOG, "appending block hash {} to backward sync", blockHash::toHexString);
       backwardSyncContext.updateHeads(blockHash, finalizedBlockHash);
-      backwardSyncContext.syncBackwardsUntil(blockHash);
+      backwardSyncContext
+          .syncBackwardsUntil(blockHash)
+          .exceptionally(e -> logSyncException(blockHash, e));
     }
     return optHeader;
   }
 
+  private Void logSyncException(final Hash blockHash, final Throwable exception) {
+    LOG.warn("Sync to block hash " + blockHash.toHexString() + " failed", exception);
+    return null;
+  }
+
   @Override
   public Result validateBlock(final Block block) {
 

consensus/merge/src/test/java/org/hyperledger/besu/consensus/merge/blockcreation/MergeCoordinatorTest.java

@@ -23,7 +23,6 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -50,6 +49,7 @@
 import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
 
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.tuweni.bytes.Bytes32;
@@ -280,10 +280,12 @@ public void assertGetOrSyncForBlockAlreadyPresent() {
   public void assertGetOrSyncForBlockNotPresent() {
     BlockHeader mockHeader =
         headerGenerator.parentHash(Hash.fromHexStringLenient("0xbeef")).buildHeader();
+    when(backwardSyncContext.syncBackwardsUntil(mockHeader.getBlockHash()))
+        .thenReturn(CompletableFuture.completedFuture(null));
+
     var res = coordinator.getOrSyncHeaderByHash(mockHeader.getHash());
 
     assertThat(res).isNotPresent();
-    verify(backwardSyncContext, times(1)).syncBackwardsUntil(mockHeader.getHash());
   }
 
   @Test

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/methods/engine/EngineNewPayload.java

@@ -165,7 +165,13 @@ public JsonRpcResponse syncResponse(final JsonRpcRequestContext requestContext)
         new Block(newBlockHeader, new BlockBody(transactions, Collections.emptyList()));
 
     if (mergeContext.isSyncing() || parentHeader.isEmpty()) {
-      mergeCoordinator.appendNewPayloadToSync(block);
+      mergeCoordinator
+          .appendNewPayloadToSync(block)
+          .exceptionally(
+              exception -> {
+                LOG.warn("Sync to block " + block.toLogString() + " failed", exception);
+                return null;
+              });
       return respondWith(reqId, blockParam, null, SYNCING);
     }
 

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/methods/engine/EngineNewPayloadTest.java

@@ -52,6 +52,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 
 import io.vertx.core.Vertx;
 import org.apache.tuweni.bytes.Bytes32;
@@ -263,7 +264,8 @@ public void shouldRespondWithSyncingDuringForwardSync() {
     BlockHeader mockHeader = new BlockHeaderTestFixture().baseFeePerGas(Wei.ONE).buildHeader();
     when(blockchain.getBlockByHash(any())).thenReturn(Optional.empty());
     when(mergeContext.isSyncing()).thenReturn(Boolean.TRUE);
-
+    when(mergeCoordinator.appendNewPayloadToSync(any()))
+        .thenReturn(CompletableFuture.completedFuture(null));
     var resp = resp(mockPayload(mockHeader, Collections.emptyList()));
 
     EnginePayloadStatusResult res = fromSuccessResp(resp);
@@ -275,7 +277,8 @@ public void shouldRespondWithSyncingDuringForwardSync() {
   @Test
   public void shouldRespondWithSyncingDuringBackwardsSync() {
     BlockHeader mockHeader = new BlockHeaderTestFixture().baseFeePerGas(Wei.ONE).buildHeader();
-
+    when(mergeCoordinator.appendNewPayloadToSync(any()))
+        .thenReturn(CompletableFuture.completedFuture(null));
     var resp = resp(mockPayload(mockHeader, Collections.emptyList()));
 
     EnginePayloadStatusResult res = fromSuccessResp(resp);

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/backwardsync/BackwardChain.java

@@ -97,27 +97,27 @@ public synchronized void prependAncestorsHeader(final BlockHeader blockHeader) {
     BlockHeader firstHeader = firstStoredAncestor.get();
     if (firstHeader.getNumber() != blockHeader.getNumber() + 1) {
       throw new BackwardSyncException(
-          "Wrong height of header "
-              + blockHeader.getHash().toHexString()
-              + " is "
-              + blockHeader.getNumber()
-              + " when we were expecting "
-              + (firstHeader.getNumber() - 1));
+          "Block "
+              + firstHeader.toLogString()
+              + " has a wrong height, we were expecting "
+              + (blockHeader.getNumber() + 1));
     }
     if (!firstHeader.getParentHash().equals(blockHeader.getHash())) {
       throw new BackwardSyncException(
-          "Hash of header does not match our expectations, was "
-              + blockHeader.toLogString()
-              + " when we expected "
-              + firstHeader.getParentHash().toHexString());
+          "For block "
+              + firstHeader.toLogString()
+              + " we were expecting the parent with hash "
+              + firstHeader.getParentHash().toHexString()
+              + " while as parent we found "
+              + blockHeader.toLogString());
     }
     headers.put(blockHeader.getHash(), blockHeader);
     chainStorage.put(blockHeader.getHash(), firstStoredAncestor.get().getHash());
     firstStoredAncestor = Optional.of(blockHeader);
     debugLambda(
         LOG,
         "Added header {} on height {} to backward chain led by pivot {} on height {}",
-        () -> blockHeader.toLogString(),
+        blockHeader::toLogString,
         blockHeader::getNumber,
         () -> lastStoredPivot.orElseThrow().toLogString(),
         firstHeader::getNumber);

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/backwardsync/BackwardSyncContext.java

@@ -15,7 +15,6 @@
 package org.hyperledger.besu.ethereum.eth.sync.backwardsync;
 
 import static org.hyperledger.besu.util.Slf4jLambdaHelper.debugLambda;
-import static org.hyperledger.besu.util.Slf4jLambdaHelper.infoLambda;
 import static org.hyperledger.besu.util.Slf4jLambdaHelper.traceLambda;
 
 import org.hyperledger.besu.datatypes.Hash;
@@ -96,27 +95,33 @@ public synchronized void updateHeads(final Hash head, final Hash finalizedBlockH
   }
 
   public synchronized CompletableFuture<Void> syncBackwardsUntil(final Hash newBlockHash) {
-    final CompletableFuture<Void> future = this.currentBackwardSyncFuture.get();
-    if (isTrusted(newBlockHash)) return future;
-    backwardChain.addNewHash(newBlockHash);
-    if (future != null) {
-      return future;
+    Optional<CompletableFuture<Void>> maybeFuture =
+        Optional.ofNullable(this.currentBackwardSyncFuture.get());
+    if (isTrusted(newBlockHash)) {
+      return maybeFuture.orElseGet(() -> CompletableFuture.completedFuture(null));
     }
-    infoLambda(LOG, "Starting new backward sync towards a pivot {}", newBlockHash::toHexString);
-    this.currentBackwardSyncFuture.set(prepareBackwardSyncFutureWithRetry());
-    return this.currentBackwardSyncFuture.get();
+    backwardChain.addNewHash(newBlockHash);
+    return maybeFuture.orElseGet(
+        () -> {
+          CompletableFuture<Void> future = prepareBackwardSyncFutureWithRetry();
+          this.currentBackwardSyncFuture.set(future);
+          return future;
+        });
   }
 
   public synchronized CompletableFuture<Void> syncBackwardsUntil(final Block newPivot) {
-    final CompletableFuture<Void> future = this.currentBackwardSyncFuture.get();
-    if (isTrusted(newPivot.getHash())) return future;
-    backwardChain.appendTrustedBlock(newPivot);
-    if (future != null) {
-      return future;
+    Optional<CompletableFuture<Void>> maybeFuture =
+        Optional.ofNullable(this.currentBackwardSyncFuture.get());
+    if (isTrusted(newPivot.getHash())) {
+      return maybeFuture.orElseGet(() -> CompletableFuture.completedFuture(null));
     }
-    infoLambda(LOG, "Starting new backward sync towards a pivot {}", newPivot::toLogString);
-    this.currentBackwardSyncFuture.set(prepareBackwardSyncFutureWithRetry());
-    return this.currentBackwardSyncFuture.get();
+    backwardChain.appendTrustedBlock(newPivot);
+    return maybeFuture.orElseGet(
+        () -> {
+          CompletableFuture<Void> future = prepareBackwardSyncFutureWithRetry();
+          this.currentBackwardSyncFuture.set(future);
+          return future;
+        });
   }
 
   private boolean isTrusted(final Hash hash) {
@@ -149,33 +154,44 @@ private CompletableFuture<Void> prepareBackwardSyncFutureWithRetry() {
         (unused, throwable) -> {
           this.currentBackwardSyncFuture.set(null);
           if (throwable != null) {
-            throw new BackwardSyncException(throwable);
+            throw extractBackwardSyncException(throwable)
+                .orElse(new BackwardSyncException(throwable));
           }
           return null;
         });
   }
 
   @VisibleForTesting
   protected void processException(final Throwable throwable) {
+    extractBackwardSyncException(throwable)
+        .ifPresentOrElse(
+            backwardSyncException -> {
+              if (backwardSyncException.shouldRestart()) {
+                LOG.info(
+                    "Backward sync failed ({}). Current Peers: {}. Retrying in few seconds...",
+                    backwardSyncException.getMessage(),
+                    ethContext.getEthPeers().peerCount());
+                return;
+              } else {
+                throw backwardSyncException;
+              }
+            },
+            () ->
+                LOG.warn(
+                    "There was an uncaught exception during Backwards Sync. Retrying in few seconds...",
+                    throwable));
+  }
+
+  private Optional<BackwardSyncException> extractBackwardSyncException(final Throwable throwable) {
     Throwable currentCause = throwable;
 
     while (currentCause != null) {
       if (currentCause instanceof BackwardSyncException) {
-        if (((BackwardSyncException) currentCause).shouldRestart()) {
-          LOG.info(
-              "Backward sync failed ({}). Current Peers: {}. Retrying in few seconds... ",
-              currentCause.getMessage(),
-              ethContext.getEthPeers().peerCount());
-          return;
-        } else {
-          throw new BackwardSyncException(throwable);
-        }
+        return Optional.of((BackwardSyncException) currentCause);
       }
       currentCause = currentCause.getCause();
     }
-    LOG.warn(
-        "There was an uncaught exception during Backwards Sync... Retrying in few seconds...",
-        throwable);
+    return Optional.empty();
   }
 
   private CompletableFuture<Void> prepareBackwardSyncFuture() {

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/backwardsync/InMemoryBackwardChainTest.java

@@ -103,7 +103,7 @@ public void shouldNotSaveHeadersWhenWrongHeight() {
     assertThatThrownBy(
             () -> backwardChain.prependAncestorsHeader(blocks.get(blocks.size() - 5).getHeader()))
         .isInstanceOf(BackwardSyncException.class)
-        .hasMessageContaining("Wrong height of header");
+        .hasMessageContaining("has a wrong height");
     BlockHeader firstHeader = backwardChain.getFirstAncestorHeader().orElseThrow();
     assertThat(firstHeader).isEqualTo(blocks.get(blocks.size() - 3).getHeader());
   }
@@ -116,7 +116,7 @@ public void shouldNotSaveHeadersWhenWrongHash() {
     BlockHeader wrongHashHeader = prepareWrongParentHash(blocks.get(blocks.size() - 4).getHeader());
     assertThatThrownBy(() -> backwardChain.prependAncestorsHeader(wrongHashHeader))
         .isInstanceOf(BackwardSyncException.class)
-        .hasMessageContaining("Hash of header does not match our expectations");
+        .hasMessageContaining("we were expecting the parent with hash");
     BlockHeader firstHeader = backwardChain.getFirstAncestorHeader().orElseThrow();
     assertThat(firstHeader).isEqualTo(blocks.get(blocks.size() - 3).getHeader());
   }