From 97bd32e9efa39c95dd4f346a18412c2fc279294a Mon Sep 17 00:00:00 2001 From: matkt Date: Tue, 24 Oct 2023 04:09:38 +0200 Subject: [PATCH] add retry logic on sync pipeline for rocksdb issue (#6004) * add retry logic for sync pipeline with rocksdb issue Signed-off-by: Karim TAAM (cherry picked from commit c839a3b02298c6b00ba0867cc0155d760fe7f43a) Signed-off-by: Justin Florentine --- ethereum/eth/build.gradle | 1 + .../eth/sync/StorageExceptionManager.java | 64 +++++++++++++++ .../fastsync/worldstate/PersistDataStep.java | 63 ++++++++++----- .../eth/sync/snapsync/LoadLocalDataStep.java | 50 +++++++++--- .../eth/sync/snapsync/PersistDataStep.java | 80 ++++++++++++------- .../SnapWorldStateDownloadProcess.java | 16 ++++ .../ethereum/eth/sync/snapsync/StackTrie.java | 15 +++- .../request/AccountRangeDataRequest.java | 6 ++ .../snapsync/request/BytecodeRequest.java | 5 ++ .../request/StorageRangeDataRequest.java | 6 ++ .../request/heal/TrieNodeHealingRequest.java | 5 ++ 11 files changed, 251 insertions(+), 60 deletions(-) create mode 100644 ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/StorageExceptionManager.java diff --git a/ethereum/eth/build.gradle b/ethereum/eth/build.gradle index a40b49afffb..1919efdd4f2 100644 --- a/ethereum/eth/build.gradle +++ b/ethereum/eth/build.gradle @@ -58,6 +58,7 @@ dependencies { implementation 'io.tmio:tuweni-bytes' implementation 'io.tmio:tuweni-units' implementation 'io.tmio:tuweni-rlp' + implementation 'org.rocksdb:rocksdbjni' annotationProcessor "org.immutables:value" implementation "org.immutables:value-annotations" diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/StorageExceptionManager.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/StorageExceptionManager.java new file mode 100644 index 00000000000..97d1506cf3e --- /dev/null +++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/StorageExceptionManager.java @@ -0,0 +1,64 @@ +/* + * 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; + +import org.hyperledger.besu.plugin.services.exception.StorageException; + +import java.util.EnumSet; +import java.util.Optional; + +import org.rocksdb.RocksDBException; +import org.rocksdb.Status; + +public final class StorageExceptionManager { + + private static final EnumSet RETRYABLE_STATUS_CODES = + EnumSet.of(Status.Code.TimedOut, Status.Code.TryAgain, Status.Code.Busy); + + private static final long ERROR_THRESHOLD = 1000; + + private static long retryableErrorCounter; + /** + * Determines if an operation can be retried based on the error received. This method checks if + * the cause of the StorageException is a RocksDBException. If it is, it retrieves the status code + * of the RocksDBException and checks if it is contained in the list of retryable {@link + * StorageExceptionManager.RETRYABLE_STATUS_CODES} status codes. + * + * @param e the StorageException to check + * @return true if the operation can be retried, false otherwise + */ + public static boolean canRetryOnError(final StorageException e) { + return Optional.of(e.getCause()) + .filter(z -> z instanceof RocksDBException) + .map(RocksDBException.class::cast) + .map(RocksDBException::getStatus) + .map(Status::getCode) + .map(RETRYABLE_STATUS_CODES::contains) + .map( + result -> { + retryableErrorCounter++; + return result; + }) + .orElse(false); + } + + public static long getRetryableErrorCounter() { + return retryableErrorCounter; + } + + public static boolean errorCountAtThreshold() { + return retryableErrorCounter % ERROR_THRESHOLD == 1; + } +} diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/worldstate/PersistDataStep.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/worldstate/PersistDataStep.java index 94592b98d81..1ab202ee6ce 100644 --- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/worldstate/PersistDataStep.java +++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/worldstate/PersistDataStep.java @@ -14,15 +14,26 @@ */ package org.hyperledger.besu.ethereum.eth.sync.fastsync.worldstate; +import static org.hyperledger.besu.ethereum.eth.sync.StorageExceptionManager.canRetryOnError; +import static org.hyperledger.besu.ethereum.eth.sync.StorageExceptionManager.errorCountAtThreshold; +import static org.hyperledger.besu.ethereum.eth.sync.StorageExceptionManager.getRetryableErrorCounter; + import org.hyperledger.besu.ethereum.core.BlockHeader; import org.hyperledger.besu.ethereum.eth.sync.worldstate.WorldDownloadState; import org.hyperledger.besu.ethereum.worldstate.WorldStateStorage; import org.hyperledger.besu.ethereum.worldstate.WorldStateStorage.Updater; +import org.hyperledger.besu.plugin.services.exception.StorageException; import org.hyperledger.besu.services.tasks.Task; import java.util.List; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + public class PersistDataStep { + + private static final Logger LOG = LoggerFactory.getLogger(PersistDataStep.class); + private final WorldStateStorage worldStateStorage; public PersistDataStep(final WorldStateStorage worldStateStorage) { @@ -33,24 +44,40 @@ public List> persist( final List> tasks, final BlockHeader blockHeader, final WorldDownloadState downloadState) { - final Updater updater = worldStateStorage.updater(); - tasks.stream() - .map( - task -> { - enqueueChildren(task, downloadState); - return task; - }) - .map(Task::getData) - .filter(request -> request.getData() != null) - .forEach( - request -> { - if (isRootState(blockHeader, request)) { - downloadState.setRootNodeData(request.getData()); - } else { - request.persist(updater); - } - }); - updater.commit(); + try { + final Updater updater = worldStateStorage.updater(); + tasks.stream() + .map( + task -> { + enqueueChildren(task, downloadState); + return task; + }) + .map(Task::getData) + .filter(request -> request.getData() != null) + .forEach( + request -> { + if (isRootState(blockHeader, request)) { + downloadState.setRootNodeData(request.getData()); + } else { + request.persist(updater); + } + }); + updater.commit(); + } catch (StorageException storageException) { + if (canRetryOnError(storageException)) { + // We reset the task by setting it to null. This way, it is considered as failed by the + // pipeline, and it will attempt to execute it again later. + if (errorCountAtThreshold()) { + LOG.info( + "Encountered {} retryable RocksDB errors, latest error message {}", + getRetryableErrorCounter(), + storageException.getMessage()); + } + tasks.forEach(nodeDataRequestTask -> nodeDataRequestTask.getData().setData(null)); + } else { + throw storageException; + } + } return tasks; } diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/LoadLocalDataStep.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/LoadLocalDataStep.java index 5264eac2b50..c24dbf6037d 100644 --- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/LoadLocalDataStep.java +++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/LoadLocalDataStep.java @@ -14,11 +14,16 @@ */ package org.hyperledger.besu.ethereum.eth.sync.snapsync; +import static org.hyperledger.besu.ethereum.eth.sync.StorageExceptionManager.canRetryOnError; +import static org.hyperledger.besu.ethereum.eth.sync.StorageExceptionManager.errorCountAtThreshold; +import static org.hyperledger.besu.ethereum.eth.sync.StorageExceptionManager.getRetryableErrorCounter; + import org.hyperledger.besu.ethereum.eth.sync.snapsync.request.SnapDataRequest; import org.hyperledger.besu.ethereum.eth.sync.snapsync.request.heal.TrieNodeHealingRequest; import org.hyperledger.besu.ethereum.worldstate.WorldStateStorage; import org.hyperledger.besu.metrics.BesuMetricCategory; import org.hyperledger.besu.plugin.services.MetricsSystem; +import org.hyperledger.besu.plugin.services.exception.StorageException; import org.hyperledger.besu.plugin.services.metrics.Counter; import org.hyperledger.besu.services.pipeline.Pipe; import org.hyperledger.besu.services.tasks.Task; @@ -27,9 +32,12 @@ import java.util.stream.Stream; import org.apache.tuweni.bytes.Bytes; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class LoadLocalDataStep { + private static final Logger LOG = LoggerFactory.getLogger(LoadLocalDataStep.class); private final WorldStateStorage worldStateStorage; private final SnapWorldDownloadState downloadState; private final SnapSyncProcessState snapSyncState; @@ -58,19 +66,35 @@ public Stream> loadLocalDataTrieNode( final Task task, final Pipe> completedTasks) { final TrieNodeHealingRequest request = (TrieNodeHealingRequest) task.getData(); // check if node is already stored in the worldstate - if (snapSyncState.hasPivotBlockHeader()) { - Optional existingData = request.getExistingData(downloadState, worldStateStorage); - if (existingData.isPresent()) { - existingNodeCounter.inc(); - request.setData(existingData.get()); - request.setRequiresPersisting(false); - final WorldStateStorage.Updater updater = worldStateStorage.updater(); - request.persist( - worldStateStorage, updater, downloadState, snapSyncState, snapSyncConfiguration); - updater.commit(); - downloadState.enqueueRequests(request.getRootStorageRequests(worldStateStorage)); - completedTasks.put(task); - return Stream.empty(); + try { + if (snapSyncState.hasPivotBlockHeader()) { + Optional existingData = request.getExistingData(downloadState, worldStateStorage); + if (existingData.isPresent()) { + existingNodeCounter.inc(); + request.setData(existingData.get()); + request.setRequiresPersisting(false); + final WorldStateStorage.Updater updater = worldStateStorage.updater(); + request.persist( + worldStateStorage, updater, downloadState, snapSyncState, snapSyncConfiguration); + updater.commit(); + downloadState.enqueueRequests(request.getRootStorageRequests(worldStateStorage)); + completedTasks.put(task); + return Stream.empty(); + } + } + } catch (StorageException storageException) { + if (canRetryOnError(storageException)) { + // We reset the task by setting it to null. This way, it is considered as failed by the + // pipeline, and it will attempt to execute it again later. + if (errorCountAtThreshold()) { + LOG.info( + "Encountered {} retryable RocksDB errors, latest error message {}", + getRetryableErrorCounter(), + storageException.getMessage()); + } + task.getData().clear(); + } else { + throw storageException; } } return Stream.of(task); diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/PersistDataStep.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/PersistDataStep.java index 6a39f648716..df3696ccdf7 100644 --- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/PersistDataStep.java +++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/PersistDataStep.java @@ -14,16 +14,25 @@ */ package org.hyperledger.besu.ethereum.eth.sync.snapsync; +import static org.hyperledger.besu.ethereum.eth.sync.StorageExceptionManager.canRetryOnError; +import static org.hyperledger.besu.ethereum.eth.sync.StorageExceptionManager.errorCountAtThreshold; +import static org.hyperledger.besu.ethereum.eth.sync.StorageExceptionManager.getRetryableErrorCounter; + import org.hyperledger.besu.ethereum.bonsai.storage.BonsaiWorldStateKeyValueStorage; import org.hyperledger.besu.ethereum.eth.sync.snapsync.request.SnapDataRequest; import org.hyperledger.besu.ethereum.eth.sync.snapsync.request.heal.TrieNodeHealingRequest; import org.hyperledger.besu.ethereum.worldstate.WorldStateStorage; +import org.hyperledger.besu.plugin.services.exception.StorageException; import org.hyperledger.besu.services.tasks.Task; import java.util.List; import java.util.stream.Stream; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + public class PersistDataStep { + private static final Logger LOG = LoggerFactory.getLogger(PersistDataStep.class); private final SnapSyncProcessState snapSyncState; private final WorldStateStorage worldStateStorage; @@ -43,41 +52,58 @@ public PersistDataStep( } public List> persist(final List> tasks) { - final WorldStateStorage.Updater updater = worldStateStorage.updater(); - for (Task task : tasks) { - if (task.getData().isResponseReceived()) { - // enqueue child requests - final Stream childRequests = - task.getData().getChildRequests(downloadState, worldStateStorage, snapSyncState); - if (!(task.getData() instanceof TrieNodeHealingRequest)) { - enqueueChildren(childRequests); - } else { - if (!task.getData().isExpired(snapSyncState)) { + try { + final WorldStateStorage.Updater updater = worldStateStorage.updater(); + for (Task task : tasks) { + if (task.getData().isResponseReceived()) { + // enqueue child requests + final Stream childRequests = + task.getData().getChildRequests(downloadState, worldStateStorage, snapSyncState); + if (!(task.getData() instanceof TrieNodeHealingRequest)) { enqueueChildren(childRequests); } else { - continue; + if (!task.getData().isExpired(snapSyncState)) { + enqueueChildren(childRequests); + } else { + continue; + } } - } - // persist nodes - final int persistedNodes = - task.getData() - .persist( - worldStateStorage, - updater, - downloadState, - snapSyncState, - snapSyncConfiguration); - if (persistedNodes > 0) { - if (task.getData() instanceof TrieNodeHealingRequest) { - downloadState.getMetricsManager().notifyTrieNodesHealed(persistedNodes); - } else { - downloadState.getMetricsManager().notifyNodesGenerated(persistedNodes); + // persist nodes + final int persistedNodes = + task.getData() + .persist( + worldStateStorage, + updater, + downloadState, + snapSyncState, + snapSyncConfiguration); + if (persistedNodes > 0) { + if (task.getData() instanceof TrieNodeHealingRequest) { + downloadState.getMetricsManager().notifyTrieNodesHealed(persistedNodes); + } else { + downloadState.getMetricsManager().notifyNodesGenerated(persistedNodes); + } } } } + updater.commit(); + } catch (StorageException storageException) { + if (canRetryOnError(storageException)) { + // We reset the task by setting it to null. This way, it is considered as failed by the + // pipeline, and it will attempt to execute it again later. not display all the retryable + // issues + if (errorCountAtThreshold()) { + LOG.info( + "Encountered {} retryable RocksDB errors, latest error message {}", + getRetryableErrorCounter(), + storageException.getMessage()); + } + tasks.forEach(task -> task.getData().clear()); + } else { + throw storageException; + } } - updater.commit(); return tasks; } 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 c8afc582b72..c19ae6facc7 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 @@ -231,6 +231,22 @@ public SnapWorldStateDownloadProcess build() { "step", "action"); + /* + The logic and intercommunication of different pipelines can be summarized as follows: + + 1. Account Data Pipeline (fetchAccountDataPipeline): This process starts with downloading the leaves of the account tree in ranges, with multiple ranges being processed simultaneously. + If the downloaded accounts are smart contracts, tasks are created in the storage pipeline to download the storage tree of the smart contract, and in the code download pipeline for the smart contract. + + 2. Storage Data Pipeline (fetchStorageDataPipeline): Running parallel to the account data pipeline, this pipeline downloads the storage of smart contracts. + If all slots cannot be downloaded at once, tasks are created in the fetchLargeStorageDataPipeline to download the storage by range, allowing parallelization of large account downloads. + + 3. Code Data Pipeline (fetchCodePipeline): This pipeline, running concurrently with the account and storage data pipelines, is responsible for downloading the code of the smart contracts. + + 4. Large Storage Data Pipeline (fetchLargeStorageDataPipeline): This pipeline is used when the storage data for a smart contract is too large to be downloaded at once. + It enables the storage data to be downloaded in ranges, similar to the account data. + + 5. Healing Phase: Initiated after all other pipelines have completed their tasks, this phase ensures the integrity and completeness of the downloaded data. + */ final Pipeline> completionPipeline = PipelineBuilder.>createPipeline( "requestDataAvailable", bufferCapacity, outputCounter, true, "node_data_request") diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/StackTrie.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/StackTrie.java index fd6acde3754..01f17eb79b4 100644 --- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/StackTrie.java +++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/StackTrie.java @@ -51,8 +51,8 @@ public class StackTrie { private final AtomicInteger nbSegments; private final int maxSegments; private final Bytes32 startKeyHash; - private final Map elements; - private final AtomicLong elementsCount; + private Map elements; + private AtomicLong elementsCount; public StackTrie(final Hash rootHash, final Bytes32 startKeyHash) { this(rootHash, 1, 1, startKeyHash); @@ -78,6 +78,12 @@ public void addElement( taskIdentifier, ImmutableTaskElement.builder().proofs(proofs).keys(keys).build()); } + public void removeElement(final Bytes32 taskIdentifier) { + if (this.elements.containsKey(taskIdentifier)) { + this.elementsCount.addAndGet(-this.elements.remove(taskIdentifier).keys().size()); + } + } + public TaskElement getElement(final Bytes32 taskIdentifier) { return this.elements.get(taskIdentifier); } @@ -142,6 +148,11 @@ public void maybeStoreNode(final Bytes location, final Node node) { } } + public void clear() { + this.elements = new LinkedHashMap<>(); + this.elementsCount = new AtomicLong(); + } + public boolean addSegment() { if (nbSegments.get() > maxSegments) { return false; diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/AccountRangeDataRequest.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/AccountRangeDataRequest.java index a1a6bc63da4..06181fd09f1 100644 --- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/AccountRangeDataRequest.java +++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/AccountRangeDataRequest.java @@ -216,6 +216,12 @@ public TreeMap getAccounts() { return stackTrie.getElement(startKeyHash).keys(); } + @Override + public void clear() { + stackTrie.clear(); + isProofValid = Optional.of(false); + } + public Bytes serialize() { return RLP.encode( out -> { diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/BytecodeRequest.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/BytecodeRequest.java index 96673d6f874..5db5ec0211e 100644 --- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/BytecodeRequest.java +++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/BytecodeRequest.java @@ -88,6 +88,11 @@ public Bytes32 getAccountHash() { return accountHash; } + @Override + public void clear() { + setCode(Bytes.EMPTY); + } + public Bytes32 getCodeHash() { return codeHash; } diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/StorageRangeDataRequest.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/StorageRangeDataRequest.java index c18d063d74d..14839f0ad6f 100644 --- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/StorageRangeDataRequest.java +++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/StorageRangeDataRequest.java @@ -205,6 +205,12 @@ public Bytes32 getEndKeyHash() { return endKeyHash; } + @Override + public void clear() { + this.isProofValid = Optional.of(false); + this.stackTrie.removeElement(startKeyHash); + } + @VisibleForTesting public void setProofValid(final boolean isProofValid) { this.isProofValid = Optional.of(isProofValid); diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/heal/TrieNodeHealingRequest.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/heal/TrieNodeHealingRequest.java index c04066141d8..ef7191a0167 100644 --- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/heal/TrieNodeHealingRequest.java +++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/heal/TrieNodeHealingRequest.java @@ -123,6 +123,11 @@ public boolean isResponseReceived() { return !data.isEmpty() && Hash.hash(data).equals(getNodeHash()); } + @Override + public void clear() { + setData(Bytes.EMPTY); + } + @Override public boolean isExpired(final SnapSyncProcessState snapSyncState) { return snapSyncState.isExpired(this);