merge pruning block for fleet mode

Signed-off-by: Karim TAAM <karim.t2am@gmail.com>

Author: matkt

besu/src/main/java/org/hyperledger/besu/cli/options/unstable/ChainPruningOptions.java

@@ -31,7 +31,7 @@ public class ChainPruningOptions implements CLIOptions<ChainPrunerConfiguration>
       "--Xchain-pruning-blocks-retained";
   private static final String CHAIN_PRUNING_FREQUENCY_FLAG = "--Xchain-pruning-frequency";
   /** The constant DEFAULT_CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED. */
-  public static final long DEFAULT_CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED = 7200;
+  public static final long DEFAULT_CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED = 512;
   /** The constant DEFAULT_CHAIN_DATA_PRUNING_FREQUENCY. */
   public static final int DEFAULT_CHAIN_DATA_PRUNING_FREQUENCY = 256;
 

besu/src/main/java/org/hyperledger/besu/controller/BesuControllerBuilder.java

@@ -629,22 +629,6 @@ public BesuController build() {
             transactionSelectorFactory);
     validateContext(protocolContext);
 
-    if (chainPrunerConfiguration.getChainPruningEnabled()) {
-      protocolContext
-          .safeConsensusContext(MergeContext.class)
-          .ifPresent(
-              mergeContext -> {
-                mergeContext.setIsChainPruningEnabled(true);
-              });
-      final ChainDataPruner chainDataPruner = createChainPruner(blockchainStorage);
-      blockchain.observeBlockAdded(chainDataPruner);
-      LOG.info(
-          "Chain data pruning enabled with recent blocks retained to be: "
-              + chainPrunerConfiguration.getChainPruningBlocksRetained()
-              + " and frequency to be: "
-              + chainPrunerConfiguration.getChainPruningBlocksFrequency());
-    }
-
     protocolSchedule.setPublicWorldStateArchiveForPrivacyBlockProcessor(
         protocolContext.getWorldStateArchive());
 
@@ -715,6 +699,22 @@ public BesuController build() {
     final boolean fullSyncDisabled = !SyncMode.isFullSync(syncConfig.getSyncMode());
     final SyncState syncState = new SyncState(blockchain, ethPeers, fullSyncDisabled, checkpoint);
 
+    if (chainPrunerConfiguration.getChainPruningEnabled()) {
+      protocolContext
+          .safeConsensusContext(MergeContext.class)
+          .ifPresent(
+              mergeContext -> {
+                mergeContext.setIsChainPruningEnabled(true);
+              });
+      final ChainDataPruner chainDataPruner = createChainPruner(syncState, blockchainStorage);
+      blockchain.observeBlockAdded(chainDataPruner);
+      LOG.info(
+          "Chain data pruning enabled with recent blocks retained to be: "
+              + chainPrunerConfiguration.getChainPruningBlocksRetained()
+              + " and frequency to be: "
+              + chainPrunerConfiguration.getChainPruningBlocksFrequency());
+    }
+
     final TransactionPool transactionPool =
         TransactionPoolFactory.createTransactionPool(
             protocolSchedule,
@@ -1089,14 +1089,16 @@ WorldStateArchive createWorldStateArchive(
     }
   }
 
-  private ChainDataPruner createChainPruner(final BlockchainStorage blockchainStorage) {
+  private ChainDataPruner createChainPruner(
+      final SyncState syncState, final BlockchainStorage blockchainStorage) {
     return new ChainDataPruner(
         blockchainStorage,
         new ChainDataPrunerStorage(
             storageProvider.getStorageBySegmentIdentifier(
                 KeyValueSegmentIdentifier.CHAIN_PRUNER_STATE)),
         chainPrunerConfiguration.getChainPruningBlocksRetained(),
         chainPrunerConfiguration.getChainPruningBlocksFrequency(),
+        syncState::isInitialSyncPhaseDone,
         MonitoredExecutors.newBoundedThreadPool(
             ChainDataPruner.class.getSimpleName(),
             1,

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/chain/ChainDataPruner.java

@@ -20,6 +20,7 @@
 
 import java.util.Collection;
 import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -32,22 +33,29 @@ public class ChainDataPruner implements BlockAddedObserver {
   private final long blocksToRetain;
   private final long pruningFrequency;
   private final ExecutorService pruningExecutor;
+  private final Supplier<Boolean> isInitialSyncPhaseDoneSupplier;
 
   public ChainDataPruner(
       final BlockchainStorage blockchainStorage,
       final ChainDataPrunerStorage prunerStorage,
       final long blocksToRetain,
       final long pruningFrequency,
+      final Supplier<Boolean> isInitialSyncPhaseDoneSupplier,
       final ExecutorService pruningExecutor) {
     this.blockchainStorage = blockchainStorage;
     this.prunerStorage = prunerStorage;
     this.blocksToRetain = blocksToRetain;
     this.pruningFrequency = pruningFrequency;
+    this.isInitialSyncPhaseDoneSupplier = isInitialSyncPhaseDoneSupplier;
     this.pruningExecutor = pruningExecutor;
   }
 
   @Override
   public void onBlockAdded(final BlockAddedEvent event) {
+    if (!isInitialSyncPhaseDoneSupplier.get()) {
+      // not run block pruning because the initial synchronization is still in progress.
+      return;
+    }
     final long blockNumber = event.getBlock().getHeader().getNumber();
     final long storedPruningMark = prunerStorage.getPruningMark().orElse(blockNumber);
     if (blockNumber < storedPruningMark) {

ethereum/core/src/test/java/org/hyperledger/besu/ethereum/chain/ChainDataPrunerTest.java

@@ -48,6 +48,8 @@ public void singleChainPruning() {
             new ChainDataPrunerStorage(new InMemoryKeyValueStorage()),
             512,
             0,
+            () -> true, // Set to 'true' to indicate that the initial synchronization phase has
+            // completed
             new BlockingExecutor());
     Block genesisBlock = gen.genesisBlock();
     final MutableBlockchain blockchain =
@@ -86,6 +88,8 @@ public void forkPruning() {
             new ChainDataPrunerStorage(new InMemoryKeyValueStorage()),
             512,
             0,
+            () -> true, // Set to 'true' to indicate that the initial synchronization phase has
+            // completed
             new BlockingExecutor());
     Block genesisBlock = gen.genesisBlock();
     final MutableBlockchain blockchain =
@@ -117,6 +121,51 @@ public void forkPruning() {
     }
   }
 
+  @Test
+  public void disablePruningWhenInitialSyncPhaseRunning() {
+    final BlockDataGenerator gen = new BlockDataGenerator();
+    final BlockchainStorage blockchainStorage =
+        new KeyValueStoragePrefixedKeyBlockchainStorage(
+            new InMemoryKeyValueStorage(),
+            new VariablesKeyValueStorage(new InMemoryKeyValueStorage()),
+            new MainnetBlockHeaderFunctions());
+    final ChainDataPruner chainDataPruner =
+        new ChainDataPruner(
+            blockchainStorage,
+            new ChainDataPrunerStorage(new InMemoryKeyValueStorage()),
+            512,
+            0,
+            () -> false, // Set to 'false' to indicate that the initial synchronization phase is
+            // running.
+            new BlockingExecutor());
+    Block genesisBlock = gen.genesisBlock();
+    final MutableBlockchain blockchain =
+        DefaultBlockchain.createMutable(
+            genesisBlock, blockchainStorage, new NoOpMetricsSystem(), 0);
+    blockchain.observeBlockAdded(chainDataPruner);
+
+    List<Block> canonicalChain = gen.blockSequence(genesisBlock, 1000);
+    List<Block> forkChain = gen.blockSequence(genesisBlock, 16);
+    for (Block blk : forkChain) {
+      blockchain.storeBlock(blk, gen.receipts(blk));
+    }
+    for (int i = 0; i < 512; i++) {
+      Block blk = canonicalChain.get(i);
+      blockchain.appendBlock(blk, gen.receipts(blk));
+    }
+    // No prune happened
+    assertThat(blockchain.getBlockByHash(canonicalChain.get(0).getHash())).isPresent();
+    assertThat(blockchain.getBlockByHash(forkChain.get(0).getHash())).isPresent();
+    for (int i = 512; i < 527; i++) {
+      Block blk = canonicalChain.get(i);
+      blockchain.appendBlock(blk, gen.receipts(blk));
+      assertThat(blockchain.getBlockByHash(canonicalChain.get(i - 512).getHash())).isPresent();
+      assertThat(blockchain.getBlockByHash(canonicalChain.get(i - 511).getHash())).isPresent();
+      assertThat(blockchain.getBlockByHash(canonicalChain.get(i - 512).getHash())).isPresent();
+      assertThat(blockchain.getBlockByHash(forkChain.get(i - 511).getHash())).isPresent();
+    }
+  }
+
   protected static class BlockingExecutor extends AbstractExecutorService {
     @Override
     public void shutdown() {}