Fix RocksDBException: Busy during snapsync (#7746)

Signed-off-by: Karim Taam <karim.t2am@gmail.com>
Signed-off-by: Simon Dudley <simon.dudley@consensys.net>
Co-authored-by: Karim Taam <karim.t2am@gmail.com>

CHANGELOG.md

@@ -28,7 +28,8 @@
 - Corrects a regression where custom plugin services are not initialized correctly. [#7625](https://github.com/hyperledger/besu/pull/7625)
 - Fix for IBFT2 chains using the BONSAI DB format [#7631](https://github.com/hyperledger/besu/pull/7631)
 - Fix reading `tx-pool-min-score` option from configuration file [#7623](https://github.com/hyperledger/besu/pull/7623)
-- Fix an undhandled exception. [#7733](https://github.com/hyperledger/besu/issues/7733)
+- Fix an unhandled PeerTable exception [#7733](https://github.com/hyperledger/besu/issues/7733)
+- Fix RocksDBException: Busy leading to MerkleTrieException: Unable to load trie node value [#7745](https://github.com/hyperledger/besu/pull/7745)
 
 ## 24.9.1
 

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/heal/TrieNodeHealingRequest.java

@@ -32,6 +32,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Stream;
 
 import org.apache.tuweni.bytes.Bytes;
@@ -44,7 +45,7 @@ public abstract class TrieNodeHealingRequest extends SnapDataRequest
   private final Bytes location;
   protected Bytes data;
 
-  protected boolean requiresPersisting = true;
+  protected AtomicBoolean requiresPersisting = new AtomicBoolean(true);
 
   protected TrieNodeHealingRequest(final Hash nodeHash, final Hash rootHash, final Bytes location) {
     super(TRIE_NODE, rootHash);
@@ -65,7 +66,7 @@ public abstract class TrieNodeHealingRequest extends SnapDataRequest
       return 0;
     }
     int saved = 0;
-    if (requiresPersisting) {
+    if (requiresPersisting.getAndSet(false)) {
       checkNotNull(data, "Must set data before node can be persisted.");
       saved =
           doPersist(
@@ -143,7 +144,7 @@ public abstract class TrieNodeHealingRequest extends SnapDataRequest
   }
 
   public boolean isRequiresPersisting() {
-    return requiresPersisting;
+    return requiresPersisting.get();
   }
 
   public Bytes32 getNodeHash() {
@@ -173,7 +174,7 @@ public abstract class TrieNodeHealingRequest extends SnapDataRequest
   }
 
   public void setRequiresPersisting(final boolean requiresPersisting) {
-    this.requiresPersisting = requiresPersisting;
+    this.requiresPersisting.set(requiresPersisting);
   }
 
   private boolean nodeIsHashReferencedDescendant(final Node<Bytes> node) {

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/PersistDataStepTest.java

@@ -17,6 +17,9 @@ package org.hyperledger.besu.ethereum.eth.sync.snapsync;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import org.hyperledger.besu.datatypes.Hash;
@@ -25,12 +28,15 @@ import org.hyperledger.besu.ethereum.eth.sync.snapsync.request.AccountRangeDataR
 import org.hyperledger.besu.ethereum.eth.sync.snapsync.request.BytecodeRequest;
 import org.hyperledger.besu.ethereum.eth.sync.snapsync.request.SnapDataRequest;
 import org.hyperledger.besu.ethereum.eth.sync.snapsync.request.StorageRangeDataRequest;
+import org.hyperledger.besu.ethereum.eth.sync.snapsync.request.heal.AccountTrieNodeHealingRequest;
 import org.hyperledger.besu.ethereum.trie.diffbased.bonsai.storage.BonsaiWorldStateKeyValueStorage;
 import org.hyperledger.besu.ethereum.trie.patricia.StoredMerklePatriciaTrie;
 import org.hyperledger.besu.ethereum.worldstate.DataStorageConfiguration;
+import org.hyperledger.besu.ethereum.worldstate.WorldStateKeyValueStorage;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateStorageCoordinator;
 import org.hyperledger.besu.services.tasks.Task;
 
+import java.util.Collections;
 import java.util.List;
 
 import org.apache.tuweni.bytes.Bytes;
@@ -39,9 +45,12 @@ import org.junit.jupiter.api.Test;
 
 public class PersistDataStepTest {
 
+  private final WorldStateKeyValueStorage worldStateKeyValueStorage =
+      spy(
+          new InMemoryKeyValueStorageProvider()
+              .createWorldStateStorage(DataStorageConfiguration.DEFAULT_CONFIG));
   private final WorldStateStorageCoordinator worldStateStorageCoordinator =
-      new InMemoryKeyValueStorageProvider()
+      new WorldStateStorageCoordinator(worldStateKeyValueStorage);
-          .createWorldStateStorageCoordinator(DataStorageConfiguration.DEFAULT_CONFIG);
 
   private final SnapSyncProcessState snapSyncState = mock(SnapSyncProcessState.class);
   private final SnapWorldDownloadState downloadState = mock(SnapWorldDownloadState.class);
@@ -67,6 +76,33 @@ public class PersistDataStepTest {
     assertDataPersisted(tasks);
   }
 
+  @Test
+  public void shouldPersistTrieNodeHealDataOnlyOnce() {
+
+    final Bytes stateTrieNode =
+        Bytes.fromHexString(
+            "0xe2a0310e2d527612073b26eecdfd717e6a320cf44b4afac2b0732d9fcbe2b7fa0cf602");
+    final Hash hash = Hash.hash(stateTrieNode);
+    final Bytes location = Bytes.of(0x02);
+    final AccountTrieNodeHealingRequest accountTrieNodeDataRequest =
+        SnapDataRequest.createAccountTrieNodeDataRequest(hash, location, Collections.emptySet());
+    accountTrieNodeDataRequest.setData(stateTrieNode);
+
+    final BonsaiWorldStateKeyValueStorage.Updater updater =
+        (BonsaiWorldStateKeyValueStorage.Updater) spy(worldStateKeyValueStorage.updater());
+    when(worldStateKeyValueStorage.updater())
+        .thenReturn(updater)
+        .thenReturn(mock(BonsaiWorldStateKeyValueStorage.Updater.class));
+
+    List<Task<SnapDataRequest>> result =
+        persistDataStep.persist(List.of(new StubTask(accountTrieNodeDataRequest)));
+
+    persistDataStep.persist(List.of(new StubTask(accountTrieNodeDataRequest)));
+
+    verify(updater, times(1)).putAccountStateTrieNode(location, hash, stateTrieNode);
+    assertDataPersisted(result);
+  }
+
   @Test
   public void shouldSkipPersistDataWhenNoData() {
     final List<Task<SnapDataRequest>> tasks = TaskGenerator.createAccountRequest(false);
@@ -110,6 +146,14 @@ public class PersistDataStepTest {
                         .getStrategy(BonsaiWorldStateKeyValueStorage.class)
                         .getCode(Hash.wrap(data.getCodeHash()), Hash.wrap(data.getAccountHash())))
                 .isPresent();
+          } else if (task.getData() instanceof AccountTrieNodeHealingRequest) {
+            final AccountTrieNodeHealingRequest data =
+                (AccountTrieNodeHealingRequest) task.getData();
+            assertThat(
+                    worldStateStorageCoordinator
+                        .getStrategy(BonsaiWorldStateKeyValueStorage.class)
+                        .getTrieNodeUnsafe(data.getLocation()))
+                .isPresent();
           } else {
             fail("not expected message");
           }