|
|
|
|
@ -47,7 +47,7 @@ public class BackwardSyncContext { |
|
|
|
|
private static final Logger LOG = LoggerFactory.getLogger(BackwardSyncContext.class); |
|
|
|
|
public static final int BATCH_SIZE = 200; |
|
|
|
|
private static final int DEFAULT_MAX_RETRIES = 20; |
|
|
|
|
|
|
|
|
|
private static final long MILLIS_DELAY_BETWEEN_PROGRESS_LOG = 10_000L; |
|
|
|
|
private static final long DEFAULT_MILLIS_BETWEEN_RETRIES = 5000; |
|
|
|
|
|
|
|
|
|
protected final ProtocolContext protocolContext; |
|
|
|
|
@ -55,9 +55,7 @@ public class BackwardSyncContext { |
|
|
|
|
private final EthContext ethContext; |
|
|
|
|
private final MetricsSystem metricsSystem; |
|
|
|
|
private final SyncState syncState; |
|
|
|
|
|
|
|
|
|
private final AtomicReference<CompletableFuture<Void>> currentBackwardSyncFuture = |
|
|
|
|
new AtomicReference<>(); |
|
|
|
|
private final AtomicReference<Status> currentBackwardSyncStatus = new AtomicReference<>(); |
|
|
|
|
private final BackwardChain backwardChain; |
|
|
|
|
private int batchSize = BATCH_SIZE; |
|
|
|
|
private Optional<Hash> maybeFinalized = Optional.empty(); |
|
|
|
|
@ -105,8 +103,8 @@ public class BackwardSyncContext { |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public synchronized boolean isSyncing() { |
|
|
|
|
return Optional.ofNullable(currentBackwardSyncFuture.get()) |
|
|
|
|
.map(CompletableFuture::isDone) |
|
|
|
|
return Optional.ofNullable(currentBackwardSyncStatus.get()) |
|
|
|
|
.map(status -> status.currentFuture.isDone()) |
|
|
|
|
.orElse(Boolean.FALSE); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -124,33 +122,51 @@ public class BackwardSyncContext { |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public synchronized CompletableFuture<Void> syncBackwardsUntil(final Hash newBlockHash) { |
|
|
|
|
Optional<CompletableFuture<Void>> maybeFuture = |
|
|
|
|
Optional.ofNullable(this.currentBackwardSyncFuture.get()); |
|
|
|
|
Optional<Status> maybeCurrentStatus = Optional.ofNullable(this.currentBackwardSyncStatus.get()); |
|
|
|
|
if (isTrusted(newBlockHash)) { |
|
|
|
|
return maybeFuture.orElseGet(() -> CompletableFuture.completedFuture(null)); |
|
|
|
|
return maybeCurrentStatus |
|
|
|
|
.map( |
|
|
|
|
status -> { |
|
|
|
|
backwardChain |
|
|
|
|
.getBlock(newBlockHash) |
|
|
|
|
.ifPresent(block -> status.updateTargetHeight(block.getHeader().getNumber())); |
|
|
|
|
return status.currentFuture; |
|
|
|
|
}) |
|
|
|
|
.orElseGet(() -> CompletableFuture.completedFuture(null)); |
|
|
|
|
} |
|
|
|
|
backwardChain.addNewHash(newBlockHash); |
|
|
|
|
return maybeFuture.orElseGet( |
|
|
|
|
() -> { |
|
|
|
|
CompletableFuture<Void> future = prepareBackwardSyncFutureWithRetry(); |
|
|
|
|
this.currentBackwardSyncFuture.set(future); |
|
|
|
|
return future; |
|
|
|
|
}); |
|
|
|
|
return maybeCurrentStatus |
|
|
|
|
.map(Status::getCurrentFuture) |
|
|
|
|
.orElseGet( |
|
|
|
|
() -> { |
|
|
|
|
LOG.info("Starting a new backward sync session"); |
|
|
|
|
Status status = new Status(prepareBackwardSyncFutureWithRetry()); |
|
|
|
|
this.currentBackwardSyncStatus.set(status); |
|
|
|
|
return status.currentFuture; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public synchronized CompletableFuture<Void> syncBackwardsUntil(final Block newPivot) { |
|
|
|
|
Optional<CompletableFuture<Void>> maybeFuture = |
|
|
|
|
Optional.ofNullable(this.currentBackwardSyncFuture.get()); |
|
|
|
|
Optional<Status> maybeCurrentStatus = Optional.ofNullable(this.currentBackwardSyncStatus.get()); |
|
|
|
|
if (isTrusted(newPivot.getHash())) { |
|
|
|
|
return maybeFuture.orElseGet(() -> CompletableFuture.completedFuture(null)); |
|
|
|
|
return maybeCurrentStatus |
|
|
|
|
.map(Status::getCurrentFuture) |
|
|
|
|
.orElseGet(() -> CompletableFuture.completedFuture(null)); |
|
|
|
|
} |
|
|
|
|
backwardChain.appendTrustedBlock(newPivot); |
|
|
|
|
return maybeFuture.orElseGet( |
|
|
|
|
() -> { |
|
|
|
|
CompletableFuture<Void> future = prepareBackwardSyncFutureWithRetry(); |
|
|
|
|
this.currentBackwardSyncFuture.set(future); |
|
|
|
|
return future; |
|
|
|
|
}); |
|
|
|
|
return maybeCurrentStatus |
|
|
|
|
.map(Status::getCurrentFuture) |
|
|
|
|
.orElseGet( |
|
|
|
|
() -> { |
|
|
|
|
LOG.info("Starting a new backward sync session"); |
|
|
|
|
LOG.info("Backward sync target block is {}", newPivot.toLogString()); |
|
|
|
|
Status status = new Status(prepareBackwardSyncFutureWithRetry()); |
|
|
|
|
status.setSyncRange( |
|
|
|
|
getProtocolContext().getBlockchain().getChainHeadBlockNumber(), |
|
|
|
|
newPivot.getHeader().getNumber()); |
|
|
|
|
this.currentBackwardSyncStatus.set(status); |
|
|
|
|
return status.currentFuture; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private boolean isTrusted(final Hash hash) { |
|
|
|
|
@ -168,8 +184,9 @@ public class BackwardSyncContext { |
|
|
|
|
return prepareBackwardSyncFutureWithRetry(maxRetries) |
|
|
|
|
.handle( |
|
|
|
|
(unused, throwable) -> { |
|
|
|
|
this.currentBackwardSyncFuture.set(null); |
|
|
|
|
this.currentBackwardSyncStatus.set(null); |
|
|
|
|
if (throwable != null) { |
|
|
|
|
LOG.info("Current backward sync session failed, it will be restarted"); |
|
|
|
|
throw extractBackwardSyncException(throwable) |
|
|
|
|
.orElse(new BackwardSyncException(throwable)); |
|
|
|
|
} |
|
|
|
|
@ -201,8 +218,8 @@ public class BackwardSyncContext { |
|
|
|
|
.ifPresentOrElse( |
|
|
|
|
backwardSyncException -> { |
|
|
|
|
if (backwardSyncException.shouldRestart()) { |
|
|
|
|
LOG.info( |
|
|
|
|
"Backward sync failed ({}). Current Peers: {}. Retrying in {} milliseconds...", |
|
|
|
|
LOG.debug( |
|
|
|
|
"Backward sync failed ({}). Current Peers: {}. Retrying in {} milliseconds", |
|
|
|
|
throwable.getMessage(), |
|
|
|
|
ethContext.getEthPeers().peerCount(), |
|
|
|
|
millisBetweenRetries); |
|
|
|
|
@ -213,8 +230,8 @@ public class BackwardSyncContext { |
|
|
|
|
} |
|
|
|
|
}, |
|
|
|
|
() -> { |
|
|
|
|
LOG.warn( |
|
|
|
|
"Backward sync failed ({}). Current Peers: {}. Retrying in {} milliseconds...", |
|
|
|
|
LOG.debug( |
|
|
|
|
"Backward sync failed ({}). Current Peers: {}. Retrying in {} milliseconds", |
|
|
|
|
throwable.getMessage(), |
|
|
|
|
ethContext.getEthPeers().peerCount(), |
|
|
|
|
millisBetweenRetries); |
|
|
|
|
@ -278,10 +295,6 @@ public class BackwardSyncContext { |
|
|
|
|
&& syncState.isInitialSyncPhaseDone(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public CompletableFuture<Void> stop() { |
|
|
|
|
return currentBackwardSyncFuture.get(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public void subscribeBadChainListener(final BadChainListener badChainListener) { |
|
|
|
|
badChainListeners.subscribe(badChainListener); |
|
|
|
|
} |
|
|
|
|
@ -316,6 +329,7 @@ public class BackwardSyncContext { |
|
|
|
|
.getBlockchain() |
|
|
|
|
.appendBlock(block, optResult.getYield().get().getReceipts()); |
|
|
|
|
possiblyMoveHead(block); |
|
|
|
|
logBlockImportProgress(block.getHeader().getNumber()); |
|
|
|
|
} else { |
|
|
|
|
emitBadChainEvent(block); |
|
|
|
|
throw new BackwardSyncException( |
|
|
|
|
@ -365,6 +379,10 @@ public class BackwardSyncContext { |
|
|
|
|
.findFirst(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public Status getStatus() { |
|
|
|
|
return currentBackwardSyncStatus.get(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private void emitBadChainEvent(final Block badBlock) { |
|
|
|
|
final List<Block> badBlockDescendants = new ArrayList<>(); |
|
|
|
|
final List<BlockHeader> badBlockHeaderDescendants = new ArrayList<>(); |
|
|
|
|
@ -385,4 +403,75 @@ public class BackwardSyncContext { |
|
|
|
|
badChainListeners.forEach( |
|
|
|
|
listener -> listener.onBadChain(badBlock, badBlockDescendants, badBlockHeaderDescendants)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private void logBlockImportProgress(final long currImportedHeight) { |
|
|
|
|
final Status currentStatus = getStatus(); |
|
|
|
|
final long targetHeight = currentStatus.getTargetChainHeight(); |
|
|
|
|
final long initialHeight = currentStatus.getInitialChainHeight(); |
|
|
|
|
final long estimatedTotal = targetHeight - initialHeight; |
|
|
|
|
final long imported = currImportedHeight - initialHeight; |
|
|
|
|
|
|
|
|
|
final float completedPercentage = 100.0f * imported / estimatedTotal; |
|
|
|
|
|
|
|
|
|
if (completedPercentage < 100.0f) { |
|
|
|
|
if (currentStatus.progressLogDue()) { |
|
|
|
|
LOG.info( |
|
|
|
|
String.format( |
|
|
|
|
"Backward sync phase 2 of 2, %.2f%% completed, imported %d blocks of at least %d (current head %d, target head %d). Peers: %d", |
|
|
|
|
completedPercentage, |
|
|
|
|
imported, |
|
|
|
|
estimatedTotal, |
|
|
|
|
currImportedHeight, |
|
|
|
|
currentStatus.getTargetChainHeight(), |
|
|
|
|
getEthContext().getEthPeers().peerCount())); |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
LOG.info( |
|
|
|
|
String.format( |
|
|
|
|
"Backward sync phase 2 of 2 completed, imported a total of %d blocks. Peers: %d", |
|
|
|
|
imported, getEthContext().getEthPeers().peerCount())); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static class Status { |
|
|
|
|
private final CompletableFuture<Void> currentFuture; |
|
|
|
|
private long targetChainHeight; |
|
|
|
|
private long initialChainHeight; |
|
|
|
|
|
|
|
|
|
private static long lastLogAt = 0; |
|
|
|
|
|
|
|
|
|
public Status(final CompletableFuture<Void> currentFuture) { |
|
|
|
|
this.currentFuture = currentFuture; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public void setSyncRange(final long initialHeight, final long targetHeight) { |
|
|
|
|
initialChainHeight = initialHeight; |
|
|
|
|
targetChainHeight = targetHeight; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public void updateTargetHeight(final long newTargetHeight) { |
|
|
|
|
targetChainHeight = newTargetHeight; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public boolean progressLogDue() { |
|
|
|
|
final long now = System.currentTimeMillis(); |
|
|
|
|
if (now - lastLogAt > MILLIS_DELAY_BETWEEN_PROGRESS_LOG) { |
|
|
|
|
lastLogAt = now; |
|
|
|
|
return true; |
|
|
|
|
} |
|
|
|
|
return false; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public CompletableFuture<Void> getCurrentFuture() { |
|
|
|
|
return currentFuture; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public long getTargetChainHeight() { |
|
|
|
|
return targetChainHeight; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public long getInitialChainHeight() { |
|
|
|
|
return initialChainHeight; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
Fabio Di Fabio
2 years ago
committed by
GitHub