8337 modify chain data pruner

besu/src/main/java/org/hyperledger/besu/cli/BesuCommand.java

@@ -2662,7 +2662,8 @@ private void setMergeConfigOptions() {
 
   /** Set ignorable segments in RocksDB Storage Provider plugin. */
   public void setIgnorableStorageSegments() {
-    if (!unstableChainPruningOptions.getChainDataPruningEnabled()) {
+    if (!unstableChainPruningOptions.getChainDataPruningEnabled()
+        && !unstableChainPruningOptions.getPreMergePruningEnabled()) {
       rocksDBPlugin.addIgnorableSegmentIdentifier(KeyValueSegmentIdentifier.CHAIN_PRUNER_STATE);
     }
   }

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

@@ -25,11 +25,13 @@
 /** The Chain pruning CLI options. */
 public class ChainPruningOptions implements CLIOptions<ChainPrunerConfiguration> {
   private static final String CHAIN_PRUNING_ENABLED_FLAG = "--Xchain-pruning-enabled";
+  private static final String PRE_MERGE_PRUNING_ENABLED_FLAG = "--Xpre-merge-pruning-enabled";
   private static final String CHAIN_PRUNING_BLOCKS_RETAINED_FLAG =
       "--Xchain-pruning-blocks-retained";
   private static final String CHAIN_PRUNING_BLOCKS_RETAINED_LIMIT_FLAG =
       "--Xchain-pruning-blocks-retained-limit";
   private static final String CHAIN_PRUNING_FREQUENCY_FLAG = "--Xchain-pruning-frequency";
+  private static final String PRE_MERGE_PRUNING_QUANTITY_FLAG = "--Xpre-merge-pruning-quantity";
 
   /**
    * The "CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED_LIMIT" field sets the minimum limit for the
@@ -42,20 +44,32 @@ public class ChainPruningOptions implements CLIOptions<ChainPrunerConfiguration>
   /** The constant DEFAULT_CHAIN_DATA_PRUNING_FREQUENCY. */
   public static final int DEFAULT_CHAIN_DATA_PRUNING_FREQUENCY = 256;
 
+  /** The constant DEFAULT_PRE_MERGE_PRUNING_QUANTITY. */
+  public static final int DEFAULT_PRE_MERGE_PRUNING_QUANTITY = 256;
+
   @CommandLine.Option(
       hidden = true,
       names = {CHAIN_PRUNING_ENABLED_FLAG},
       description =
           "Enable the chain pruner to actively prune old chain data (default: ${DEFAULT-VALUE})")
   private final Boolean chainDataPruningEnabled = Boolean.FALSE;
 
+  @CommandLine.Option(
+      hidden = true,
+      names = {PRE_MERGE_PRUNING_ENABLED_FLAG},
+      description =
+          "Enable the chain pruner to actively prune pre-merge blocks, but not headers (default: ${DEFAULT-VALUE})")
+  private final Boolean preMergePruningEnabled = Boolean.FALSE;
+
   @CommandLine.Option(
       hidden = true,
       names = {CHAIN_PRUNING_BLOCKS_RETAINED_FLAG},
       description =
           "The number of recent blocks for which to keep the chain data. Should be >= "
               + CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED_LIMIT
-              + " (default: ${DEFAULT-VALUE})")
+              + " (default: ${DEFAULT-VALUE}). Unused if "
+              + PRE_MERGE_PRUNING_ENABLED_FLAG
+              + " is enabled")
   private final Long chainDataPruningBlocksRetained = CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED_LIMIT;
 
   @CommandLine.Option(
@@ -65,7 +79,9 @@ public class ChainPruningOptions implements CLIOptions<ChainPrunerConfiguration>
           "Allows setting the limit below which no more blocks can be pruned. This prevents setting a value lower than this for "
               + CHAIN_PRUNING_BLOCKS_RETAINED_FLAG
               + ". This flag should be used with caution as reducing the limit may have unintended side effects."
-              + " (default: ${DEFAULT-VALUE})")
+              + " (default: ${DEFAULT-VALUE}). Unused if "
+              + PRE_MERGE_PRUNING_ENABLED_FLAG
+              + " is enabled")
   private final Long chainDataPruningBlocksRetainedLimit =
       CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED_LIMIT;
 
@@ -77,6 +93,14 @@ public class ChainPruningOptions implements CLIOptions<ChainPrunerConfiguration>
   private final PositiveNumber chainDataPruningBlocksFrequency =
       PositiveNumber.fromInt(DEFAULT_CHAIN_DATA_PRUNING_FREQUENCY);
 
+  @CommandLine.Option(
+      hidden = true,
+      names = {PRE_MERGE_PRUNING_QUANTITY_FLAG},
+      description =
+          "The number of pre-merge blocks to prune per pruning operation. Must be non-negative (default: ${DEFAULT-VALUE})")
+  private final PositiveNumber preMergePruningBlocksQuantity =
+      PositiveNumber.fromInt(DEFAULT_PRE_MERGE_PRUNING_QUANTITY);
+
   /** Default Constructor. */
   ChainPruningOptions() {}
 
@@ -98,6 +122,15 @@ public Boolean getChainDataPruningEnabled() {
     return chainDataPruningEnabled;
   }
 
+  /**
+   * Gets pre-merge pruning enabled
+   *
+   * @return the pre-merge pruning enabled
+   */
+  public Boolean getPreMergePruningEnabled() {
+    return preMergePruningEnabled;
+  }
+
   /**
    * Gets chain data pruning blocks retained.
    *
@@ -120,21 +153,27 @@ public Long getChainDataPruningBlocksRetainedLimit() {
   public ChainPrunerConfiguration toDomainObject() {
     return new ChainPrunerConfiguration(
         chainDataPruningEnabled,
+        preMergePruningEnabled,
         chainDataPruningBlocksRetained,
         chainDataPruningBlocksRetainedLimit,
-        chainDataPruningBlocksFrequency.getValue());
+        chainDataPruningBlocksFrequency.getValue(),
+        preMergePruningBlocksQuantity.getValue());
   }
 
   @Override
   public List<String> getCLIOptions() {
     return Arrays.asList(
         CHAIN_PRUNING_ENABLED_FLAG,
         chainDataPruningEnabled.toString(),
+        PRE_MERGE_PRUNING_ENABLED_FLAG,
+        preMergePruningEnabled.toString(),
         CHAIN_PRUNING_BLOCKS_RETAINED_FLAG,
         chainDataPruningBlocksRetained.toString(),
         CHAIN_PRUNING_BLOCKS_RETAINED_LIMIT_FLAG,
         chainDataPruningBlocksRetainedLimit.toString(),
         CHAIN_PRUNING_FREQUENCY_FLAG,
-        chainDataPruningBlocksFrequency.toString());
+        chainDataPruningBlocksFrequency.toString(),
+        PRE_MERGE_PRUNING_QUANTITY_FLAG,
+        preMergePruningBlocksQuantity.toString());
   }
 }

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

@@ -16,6 +16,7 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import org.hyperledger.besu.cli.config.NetworkName;
 import org.hyperledger.besu.components.BesuComponent;
 import org.hyperledger.besu.config.CheckpointConfigOptions;
 import org.hyperledger.besu.config.GenesisConfig;
@@ -115,8 +116,10 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.OptionalLong;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
+import java.util.stream.Stream;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -125,6 +128,9 @@
 public abstract class BesuControllerBuilder implements MiningParameterOverrides {
   private static final Logger LOG = LoggerFactory.getLogger(BesuControllerBuilder.class);
 
+  private static final long MAINNET_MERGE_BLOCK_NUMBER = 15_537_393;
+  private static final long SEPOLIA_MERGE_BLOCK_NUMBER = 1_735_371;
+
   /** The genesis file */
   protected GenesisConfig genesisConfig;
 
@@ -695,14 +701,27 @@ public BesuController build() {
     final boolean fullSyncDisabled = !SyncMode.isFullSync(syncConfig.getSyncMode());
     final SyncState syncState = new SyncState(blockchain, ethPeers, fullSyncDisabled, checkpoint);
 
-    if (chainPrunerConfiguration.getChainPruningEnabled()) {
-      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());
+    if (chainPrunerConfiguration.chainPruningEnabled()
+        || chainPrunerConfiguration.preMergePruningEnabled()) {
+      LOG.info("Adding ChainDataPruner to observe block added events");
+      final AtomicLong chainDataPrunerObserverId = new AtomicLong();
+      final ChainDataPruner chainDataPruner =
+          createChainPruner(
+              blockchainStorage, () -> blockchain.removeObserver(chainDataPrunerObserverId.get()));
+      chainDataPrunerObserverId.set(blockchain.observeBlockAdded(chainDataPruner));
+      if (chainPrunerConfiguration.chainPruningEnabled()) {
+        LOG.info(
+            "Chain data pruning enabled with recent blocks retained to be: "
+                + chainPrunerConfiguration.chainPruningBlocksRetained()
+                + " and frequency to be: "
+                + chainPrunerConfiguration.blocksFrequency());
+      } else if (chainPrunerConfiguration.preMergePruningEnabled()) {
+        LOG.info(
+            "Pre-merge block pruning enabled with frequency: "
+                + chainPrunerConfiguration.blocksFrequency()
+                + " and quantity: "
+                + chainPrunerConfiguration.preMergePruningBlocksQuantity());
+      }
     }
 
     final TransactionPool transactionPool =
@@ -1159,14 +1178,32 @@ yield new ForestWorldStateArchive(
     };
   }
 
-  private ChainDataPruner createChainPruner(final BlockchainStorage blockchainStorage) {
+  private ChainDataPruner createChainPruner(
+      final BlockchainStorage blockchainStorage, final Runnable unsubscribeRunnable) {
+    NetworkName network =
+        Stream.of(NetworkName.values())
+            .filter((n) -> n.getNetworkId().equals(networkId))
+            .findAny()
+            .orElseThrow(() -> new RuntimeException("Unrecognised network"));
     return new ChainDataPruner(
         blockchainStorage,
+        unsubscribeRunnable,
         new ChainDataPrunerStorage(
             storageProvider.getStorageBySegmentIdentifier(
                 KeyValueSegmentIdentifier.CHAIN_PRUNER_STATE)),
-        chainPrunerConfiguration.getChainPruningBlocksRetained(),
-        chainPrunerConfiguration.getChainPruningBlocksFrequency(),
+        switch (network) {
+          case MAINNET -> MAINNET_MERGE_BLOCK_NUMBER;
+          case SEPOLIA -> SEPOLIA_MERGE_BLOCK_NUMBER;
+          default -> 0;
+        },
+        chainPrunerConfiguration.chainPruningEnabled()
+            ? ChainDataPruner.Mode.CHAIN_PRUNING
+            : (chainPrunerConfiguration.preMergePruningEnabled()
+                ? ChainDataPruner.Mode.PRE_MERGE_PRUNING
+                : null),
+        chainPrunerConfiguration.chainPruningBlocksRetained(),
+        chainPrunerConfiguration.blocksFrequency(),
+        chainPrunerConfiguration.preMergePruningBlocksQuantity(),
         MonitoredExecutors.newBoundedThreadPool(
             ChainDataPruner.class.getSimpleName(),
             1,

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

@@ -24,29 +24,50 @@
 import org.slf4j.LoggerFactory;
 
 public class ChainDataPruner implements BlockAddedObserver {
-  public static final int MAX_PRUNING_THREAD_QUEUE_SIZE = 16;
   private static final Logger LOG = LoggerFactory.getLogger(ChainDataPruner.class);
+
+  public static final int MAX_PRUNING_THREAD_QUEUE_SIZE = 16;
+
   private final BlockchainStorage blockchainStorage;
+  private final Runnable unsubscribeRunnable;
   private final ChainDataPrunerStorage prunerStorage;
+  private final long mergeBlock;
+  private final Mode mode;
   private final long blocksToRetain;
   private final long pruningFrequency;
+  private final long pruningQuantity;
   private final ExecutorService pruningExecutor;
 
   public ChainDataPruner(
       final BlockchainStorage blockchainStorage,
+      final Runnable unsubscribeRunnable,
       final ChainDataPrunerStorage prunerStorage,
+      final long mergeBlock,
+      final Mode mode,
       final long blocksToRetain,
       final long pruningFrequency,
+      final long pruningQuantity,
       final ExecutorService pruningExecutor) {
     this.blockchainStorage = blockchainStorage;
+    this.unsubscribeRunnable = unsubscribeRunnable;
     this.prunerStorage = prunerStorage;
+    this.mergeBlock = mergeBlock;
+    this.mode = mode;
     this.blocksToRetain = blocksToRetain;
     this.pruningFrequency = pruningFrequency;
     this.pruningExecutor = pruningExecutor;
+    this.pruningQuantity = pruningQuantity;
   }
 
   @Override
   public void onBlockAdded(final BlockAddedEvent event) {
+    switch (mode) {
+      case CHAIN_PRUNING -> chainPrunerAction(event);
+      case PRE_MERGE_PRUNING -> preMergePruningAction();
+    }
+  }
+
+  private void chainPrunerAction(final BlockAddedEvent event) {
     final long blockNumber = event.getBlock().getHeader().getNumber();
     final long storedPruningMark = prunerStorage.getPruningMark().orElse(blockNumber);
     if (blockNumber < storedPruningMark) {
@@ -87,6 +108,46 @@ public void onBlockAdded(final BlockAddedEvent event) {
         });
   }
 
+  private void preMergePruningAction() {
+    pruningExecutor.submit(
+        () -> {
+          final long storedPruningMark = prunerStorage.getPruningMark().orElse(1L);
+          final long expectedNewPruningMark =
+              Math.min(storedPruningMark + pruningQuantity, mergeBlock);
+          final KeyValueStorageTransaction pruningTransaction = prunerStorage.startTransaction();
+          final BlockchainStorage.Updater updater = blockchainStorage.updater();
+          for (long blockNumber = storedPruningMark;
+              blockNumber < expectedNewPruningMark;
+              blockNumber++) {
+            blockchainStorage
+                .getBlockHash(blockNumber)
+                .ifPresent(
+                    (blockHash) -> {
+                      updater.removeBlockBody(blockHash);
+                      updater.removeTransactionReceipts(blockHash);
+                      updater.removeTotalDifficulty(blockHash);
+                      blockchainStorage
+                          .getBlockBody(blockHash)
+                          .ifPresent(
+                              blockBody ->
+                                  blockBody
+                                      .getTransactions()
+                                      .forEach(
+                                          t -> updater.removeTransactionLocation(t.getHash())));
+                    });
+          }
+          updater.commit();
+          prunerStorage.setPruningMark(pruningTransaction, expectedNewPruningMark);
+          pruningTransaction.commit();
+          LOG.info("Pruned blocks {} to {}", storedPruningMark, expectedNewPruningMark);
+          if (expectedNewPruningMark == mergeBlock) {
+            LOG.info(
+                "Done pruning pre-merge blocks. Unsubscribing from block added event observation");
+            unsubscribeRunnable.run();
+          }
+        });
+  }
+
   private void pruneChainDataAtBlock(final KeyValueStorageTransaction tx, final long blockNumber) {
     final Collection<Hash> oldForkBlocks = prunerStorage.getForkBlocks(blockNumber);
     final BlockchainStorage.Updater updater = blockchainStorage.updater();
@@ -107,4 +168,9 @@ private void pruneChainDataAtBlock(final KeyValueStorageTransaction tx, final lo
     updater.commit();
     prunerStorage.removeForkBlocks(tx, blockNumber);
   }
+
+  public enum Mode {
+    CHAIN_PRUNING,
+    PRE_MERGE_PRUNING
+  }
 }

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

@@ -14,38 +14,13 @@
  */
 package org.hyperledger.besu.ethereum.chain;
 
-public class ChainPrunerConfiguration {
+public record ChainPrunerConfiguration(
+    boolean chainPruningEnabled,
+    boolean preMergePruningEnabled,
+    long chainPruningBlocksRetained,
+    long blocksFrequency,
+    long chainPruningBlocksRetainedLimit,
+    int preMergePruningBlocksQuantity) {
   public static final ChainPrunerConfiguration DEFAULT =
-      new ChainPrunerConfiguration(false, 7200, 7200, 256);
-  private final boolean enabled;
-  private final long blocksRetained;
-  private final long blocksFrequency;
-  private final long blocksRetainedLimit;
-
-  public ChainPrunerConfiguration(
-      final boolean enabled,
-      final long blocksRetained,
-      final long blocksRetainedLimit,
-      final long blocksFrequency) {
-    this.enabled = enabled;
-    this.blocksRetained = blocksRetained;
-    this.blocksRetainedLimit = blocksRetainedLimit;
-    this.blocksFrequency = blocksFrequency;
-  }
-
-  public long getChainPruningBlocksRetained() {
-    return blocksRetained;
-  }
-
-  public long getBlocksRetainedLimit() {
-    return blocksRetainedLimit;
-  }
-
-  public boolean getChainPruningEnabled() {
-    return enabled;
-  }
-
-  public long getChainPruningBlocksFrequency() {
-    return blocksFrequency;
-  }
+      new ChainPrunerConfiguration(false, false, 7200, 7200, 256, 1000);
 }