Skip to content

Commit

Permalink
Add extra test case and tweak code accordingly
Browse files Browse the repository at this point in the history 
Signed-off-by: Simon Dudley <[email protected]>

Signed-off-by: Simon Dudley <[email protected]>
  • Loading branch information
@siladu
siladu committed Mar 25, 2024
1 parent 80fcde0 commit b20713f
Show file tree
Hide file tree
Showing 4 changed files with 55 additions and 65 deletions.
1 change: 0 additions & 1 deletion ethereum/eth/build.gradle
View file
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,6 @@ dependencies {
testImplementation 'org.mockito:mockito-core'
testImplementation 'org.mockito:mockito-junit-jupiter'
testImplementation 'org.openjdk.jol:jol-core'
testImplementation 'org.apache.logging.log4j:log4j-core'

testSupportImplementation 'org.mockito:mockito-core'
testSupportImplementation project(':testutil')
Expand Down
4 changes: 0 additions & 4 deletions .../eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/backwardsync/BackwardChain.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -254,8 +254,4 @@ public synchronized void removeFromHashToAppend(final Hash hashToRemove) {
hashesToAppend.remove(hashToRemove);
}
}

public synchronized int getHashesToAppendSize() {
return hashesToAppend.size();
}
}
59 changes: 24 additions & 35 deletions .../main/java/org/hyperledger/besu/ethereum/eth/sync/backwardsync/BackwardSyncAlgorithm.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -63,51 +63,40 @@ public CompletableFuture<Void> executeBackwardsSync(final Void unused) {
public CompletableFuture<Void> pickNextStep() {
final Optional<Hash> firstHash = context.getBackwardChain().getFirstHashToAppend();
if (firstHash.isPresent()) {
final CompletableFuture<Void> completableFuture = new CompletableFuture<>();
final CompletableFuture<Void> syncStep = new CompletableFuture<>();
executeSyncStep(firstHash.get())
.whenComplete(
(result, error) -> {
if (error != null) {
if (error instanceof CompletionException
&& error.getCause() instanceof MaxRetriesReachedException) { // &&
// context.getEthContext().getEthPeers().peerCount() >
// (context.getEthContext().getEthPeers().getMaxPeers() / 2)
if (context.getBackwardChain().getHashesToAppendSize() > 1) {
context.getBackwardChain().removeFromHashToAppend(firstHash.get());
LOG.atWarn()
.setMessage(
"Unable to retrieve block {} from any peer, with {} peers available. Could be a reorged block. Removing hash from queue to move onto next hash.")
.addArgument(firstHash.get())
.addArgument(context.getEthContext().getEthPeers().peerCount())
.addArgument(context.getBackwardChain().getFirstHashToAppend())
.log();
completableFuture.complete(null);
} else { // removing the last hash from the queue will break things
LOG.atWarn()
.setMessage(
"Unable to retrieve block {} from any peer, with {} peers available. Could be a reorged block. Waiting for another block hash from consensus client...")
.addArgument(firstHash::get)
.addArgument(() -> context.getEthContext().getEthPeers().peerCount())
.log();
completableFuture.completeExceptionally(error);
}
&& error.getCause() instanceof MaxRetriesReachedException) {
context.getBackwardChain().removeFromHashToAppend(firstHash.get());
LOG.atWarn()
.setMessage(
"Unable to retrieve block {} from any peer, with {} peers available. Could be a reorged block. Waiting for the next block from the consensus client to try again.")
.addArgument(firstHash.get())
.addArgument(context.getEthContext().getEthPeers().peerCount())
.addArgument(context.getBackwardChain().getFirstHashToAppend())
.log();
LOG.atDebug()
.setMessage("Removing hash {} from hashesToAppend")
.addArgument(firstHash.get())
.log();
syncStep.complete(null);
} else {
completableFuture.completeExceptionally(error);
syncStep.completeExceptionally(error);
}
} else {
completableFuture.complete(null);
LOG.atDebug()
.setMessage("Backward sync target block is {}")
.addArgument(result::toLogString)
.log();
context.getBackwardChain().removeFromHashToAppend(firstHash.get());
context.getStatus().updateTargetHeight(result.getHeader().getNumber());
syncStep.complete(null);
}
})
.thenAccept(
result -> {
LOG.atDebug()
.setMessage("Backward sync target block is {}")
.addArgument(result::toLogString)
.log();
context.getBackwardChain().removeFromHashToAppend(firstHash.get());
context.getStatus().updateTargetHeight(result.getHeader().getNumber());
});
return completableFuture;
return syncStep;
}
if (!context.isReady()) {
return waitForReady();
Expand Down
56 changes: 31 additions & 25 deletions ...est/java/org/hyperledger/besu/ethereum/eth/sync/backwardsync/BackwardSyncContextTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,12 +60,8 @@
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nonnull;

import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.core.config.Configurator;
import org.apache.tuweni.bytes.Bytes;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
Expand Down Expand Up @@ -448,40 +444,50 @@ public void shouldFailAfterMaxNumberOfRetries() {

@SuppressWarnings("BannedMethod")
@Test
public void
whenMaxRetriesException_shouldRemoveUnattainableBlockFromQueueAndProgressUponNextFCU() {
public void whenBlockNotFoundInPeers_shouldRemoveBlockFromQueueAndProgressInNextSession() {
// This scenario can happen due to a reorg
// The expectation is we can progress beyond the reorg block the next time we receive an FCU

Configurator.setLevel(LogManager.getRootLogger(), Level.TRACE);
// Expectation we progress beyond the reorg block upon receiving the next FCU

// choose an intermediate remote block to create a reorg block from
int reorgBlockHeight = REMOTE_HEIGHT - 1;
int reorgBlockHeight = REMOTE_HEIGHT - 1; // 49
final Hash reorgBlockParentHash = getBlockByNumber(reorgBlockHeight - 1).getHash();
final Block reorgBlock = createBlock(reorgBlockHeight, reorgBlockParentHash);

// represents first FCU with a block that will become reorged away
final CompletableFuture<Void> fcuBeforeReorg = context.syncBackwardsUntil(reorgBlock.getHash());

respondUntilFutureIsDone(fcuBeforeReorg);
assertThat(localBlockchain.getChainHeadBlockNumber()).isLessThan(reorgBlockHeight);

assertThatThrownBy(() -> fcuBeforeReorg.get(100, TimeUnit.MILLISECONDS))
.isInstanceOf(ExecutionException.class)
.hasCauseInstanceOf(BackwardSyncException.class)
.cause()
.hasMessageContaining("Max number of retries " + NUM_OF_RETRIES + " reached");
// represents subsequent FCU with successfully reorged version of the same block
final CompletableFuture<Void> fcuAfterReorg =
context.syncBackwardsUntil(getBlockByNumber(reorgBlockHeight).getHash());
respondUntilFutureIsDone(fcuAfterReorg);
assertThat(localBlockchain.getChainHeadBlock())
.isEqualTo(remoteBlockchain.getBlockByNumber(reorgBlockHeight).orElseThrow());
}

assertThat(localBlockchain.getChainHeadBlockNumber()).isLessThan(reorgBlockHeight);
@SuppressWarnings("BannedMethod")
@Test
public void
whenBlockNotFoundInPeers_shouldRemoveBlockFromQueueAndProgressWithQueueInSameSession() {
// This scenario can happen due to a reorg
// Expectation we progress beyond the reorg block due to FCU we received during the same session

// choose an intermediate remote block to create a reorg block from
int reorgBlockHeight = REMOTE_HEIGHT - 1; // 49
final Hash reorgBlockParentHash = getBlockByNumber(reorgBlockHeight - 1).getHash();
final Block reorgBlock = createBlock(reorgBlockHeight, reorgBlockParentHash);

// represents first FCU with a block that will become reorged away
final CompletableFuture<Void> fcuBeforeReorg = context.syncBackwardsUntil(reorgBlock.getHash());
// represents subsequent FCU with successfully reorged version of the same block
// received during the first FCU's BWS session
final CompletableFuture<Void> fcuAfterReorg =
context.syncBackwardsUntil(getBlockByNumber(REMOTE_HEIGHT - 1).getHash());
context.syncBackwardsUntil(getBlockByNumber(reorgBlockHeight).getHash());

respondUntilFutureIsDone(fcuBeforeReorg);
respondUntilFutureIsDone(fcuAfterReorg);
try {
fcuAfterReorg.get(100, TimeUnit.MILLISECONDS);
assertThat(localBlockchain.getChainHeadBlock())
.isEqualTo(remoteBlockchain.getBlockByNumber(reorgBlockHeight).orElseThrow());
} catch (final Throwable throwable) {
throw new AssertionError("Expected backwards sync to progress", throwable);
}
assertThat(localBlockchain.getChainHeadBlock())
.isEqualTo(remoteBlockchain.getBlockByNumber(reorgBlockHeight).orElseThrow());
}
}

0 comments on commit b20713f

Please sign in to comment.