8337 modify chain data pruner

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

@@ -856,8 +856,6 @@ public void run() {
       // set merge config on the basis of genesis config
       setMergeConfigOptions();
 
-      setIgnorableStorageSegments();
-
       instantiateSignatureAlgorithmFactory();
 
       logger.info("Starting Besu");
@@ -869,6 +867,8 @@ public void run() {
 
       configure();
 
+      setIgnorableStorageSegments();
+
       // If we're not running against a named network, or if version compat protection has been
       // explicitly enabled, perform compatibility check
       VersionMetadata.versionCompatibilityChecks(versionCompatibilityProtection, dataDir());
@@ -2393,7 +2393,8 @@ private void setMergeConfigOptions() {
 
   /** Set ignorable segments in RocksDB Storage Provider plugin. */
   public void setIgnorableStorageSegments() {
-    if (!unstableChainPruningOptions.getChainDataPruningEnabled()) {
+    if (!unstableChainPruningOptions.getChainDataPruningEnabled()
+        && !dataStorageConfiguration.getHistoryExpiryPruneEnabled()) {
       rocksDBPlugin.addIgnorableSegmentIdentifier(KeyValueSegmentIdentifier.CHAIN_PRUNER_STATE);
     }
   }

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

@@ -30,6 +30,7 @@ public class ChainPruningOptions implements CLIOptions<ChainPrunerConfiguration>
   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,6 +43,9 @@ 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 = 100;
+
   @CommandLine.Option(
       hidden = true,
       names = {CHAIN_PRUNING_ENABLED_FLAG},
@@ -55,7 +59,7 @@ public class ChainPruningOptions implements CLIOptions<ChainPrunerConfiguration>
       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 --Xhistory-expiry-prune is enabled")
   private final Long chainDataPruningBlocksRetained = CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED_LIMIT;
 
   @CommandLine.Option(
@@ -65,7 +69,7 @@ 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 --Xhistory-expiry-prune is enabled")
   private final Long chainDataPruningBlocksRetainedLimit =
       CHAIN_DATA_PRUNING_MIN_BLOCKS_RETAINED_LIMIT;
 
@@ -77,6 +81,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() {}
 
@@ -122,7 +134,8 @@ public ChainPrunerConfiguration toDomainObject() {
         chainDataPruningEnabled,
         chainDataPruningBlocksRetained,
         chainDataPruningBlocksRetainedLimit,
-        chainDataPruningBlocksFrequency.getValue());
+        chainDataPruningBlocksFrequency.getValue(),
+        preMergePruningBlocksQuantity.getValue());
   }
 
   @Override
@@ -135,6 +148,8 @@ public List<String> getCLIOptions() {
         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

@@ -116,6 +116,7 @@
 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;
 
@@ -683,14 +684,29 @@ 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()
+        || dataStorageConfiguration.getHistoryExpiryPruneEnabled()) {
+      LOG.info("Adding ChainDataPruner to observe block added events");
+      final AtomicLong chainDataPrunerObserverId = new AtomicLong();
+      final ChainDataPruner chainDataPruner =
+          createChainPruner(
+              blockchainStorage,
+              () -> blockchain.removeObserver(chainDataPrunerObserverId.get()),
+              syncState);
+      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 (dataStorageConfiguration.getHistoryExpiryPruneEnabled()) {
+        LOG.info(
+            "Pre-merge block pruning enabled with frequency: "
+                + chainPrunerConfiguration.blocksFrequency()
+                + " and quantity: "
+                + chainPrunerConfiguration.preMergePruningBlocksQuantity());
+      }
     }
 
     final TransactionPool transactionPool =
@@ -1151,16 +1167,27 @@ yield new ForestWorldStateArchive(
     };
   }
 
-  private ChainDataPruner createChainPruner(final BlockchainStorage blockchainStorage) {
+  private ChainDataPruner createChainPruner(
+      final BlockchainStorage blockchainStorage,
+      final Runnable unsubscribeRunnable,
+      final SyncState syncState) {
     return new ChainDataPruner(
         blockchainStorage,
+        unsubscribeRunnable,
         new ChainDataPrunerStorage(
             storageProvider.getStorageBySegmentIdentifier(
                 KeyValueSegmentIdentifier.CHAIN_PRUNER_STATE)),
-        chainPrunerConfiguration.getChainPruningBlocksRetained(),
-        chainPrunerConfiguration.getChainPruningBlocksFrequency(),
+        syncState.getCheckpoint().map(Checkpoint::blockNumber).orElse(0L),
+        chainPrunerConfiguration.chainPruningEnabled()
+            ? ChainDataPruner.Mode.CHAIN_PRUNING
+            : (dataStorageConfiguration.getHistoryExpiryPruneEnabled()
+                ? ChainDataPruner.Mode.PRE_MERGE_PRUNING
+                : null),
+        chainPrunerConfiguration.chainPruningBlocksRetained(),
+        chainPrunerConfiguration.blocksFrequency(),
+        chainPrunerConfiguration.preMergePruningBlocksQuantity(),
         MonitoredExecutors.newBoundedThreadPool(
-            ChainDataPruner.class.getSimpleName(),
+            EthScheduler.class.getSimpleName() + "-ChainDataPruner",
             1,
             1,
             ChainDataPruner.MAX_PRUNING_THREAD_QUEUE_SIZE,

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

@@ -16,37 +16,64 @@
 
 import org.hyperledger.besu.datatypes.Hash;
 import org.hyperledger.besu.plugin.services.storage.KeyValueStorageTransaction;
+import org.hyperledger.besu.util.log.LogUtil;
 
 import java.util.Collection;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.slf4j.Logger;
 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);
+  private static final int LOG_PRE_MERGE_PRUNING_PROGRESS_REPEAT_DELAY_SECONDS = 300;
+
+  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;
+  private final AtomicBoolean logPreMergePruningProgress = new AtomicBoolean(true);
 
   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 -> {
+        if (event.isNewCanonicalHead()) 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 +114,56 @@ public void onBlockAdded(final BlockAddedEvent event) {
         });
   }
 
+  private void preMergePruningAction() {
+    pruningExecutor.submit(
+        () -> {
+          try {
+            Thread.sleep(1000);
+            final long storedPruningMark = prunerStorage.getPruningMark().orElse(1L);
+            final long expectedNewPruningMark =
+                Math.min(storedPruningMark + pruningQuantity, mergeBlock);
+            LOG.debug(
+                "Attempting to prune blocks {} to {}", storedPruningMark, expectedNewPruningMark);
+            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);
+                        blockchainStorage
+                            .getBlockBody(blockHash)
+                            .ifPresent(
+                                blockBody ->
+                                    blockBody
+                                        .getTransactions()
+                                        .forEach(
+                                            t -> updater.removeTransactionLocation(t.getHash())));
+                      });
+            }
+            updater.commit();
+            prunerStorage.setPruningMark(pruningTransaction, expectedNewPruningMark);
+            pruningTransaction.commit();
+            LOG.debug("Pruned pre-merge blocks up to {}", expectedNewPruningMark);
+            LogUtil.throttledLog(
+                () -> LOG.info("Pruned pre-merge blocks up to {}", expectedNewPruningMark),
+                logPreMergePruningProgress,
+                LOG_PRE_MERGE_PRUNING_PROGRESS_REPEAT_DELAY_SECONDS);
+            if (expectedNewPruningMark == mergeBlock) {
+              LOG.info("Done pruning pre-merge blocks.");
+              LOG.debug("Unsubscribing from block added event observation");
+              unsubscribeRunnable.run();
+            }
+          } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+          }
+        });
+  }
+
   private void pruneChainDataAtBlock(final KeyValueStorageTransaction tx, final long blockNumber) {
     final Collection<Hash> oldForkBlocks = prunerStorage.getForkBlocks(blockNumber);
     final BlockchainStorage.Updater updater = blockchainStorage.updater();
@@ -107,4 +184,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,12 @@
  */
 package org.hyperledger.besu.ethereum.chain;
 
-public class ChainPrunerConfiguration {
+public record ChainPrunerConfiguration(
+    boolean chainPruningEnabled,
+    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, 7200, 7200, 256, 1000);
 }