history-expiry: Prune fast by skipping pruning the transaction locations (#8596)

- Also skip pruning the totalDifficulty, for consistency with how the history expiry sync works
- Improve UX
  - more flexible number of threads default
  - logging to show settings and remaining blocks estimate

Signed-off-by: Simon Dudley <simon.dudley@consensys.net>

Author: siladu

besu/src/main/java/org/hyperledger/besu/cli/subcommands/storage/PrunePreMergeBlockDataSubCommand.java

@@ -17,11 +17,12 @@
 import org.hyperledger.besu.cli.config.NetworkName;
 import org.hyperledger.besu.cli.util.VersionProvider;
 import org.hyperledger.besu.controller.BesuController;
-import org.hyperledger.besu.datatypes.Transaction;
+import org.hyperledger.besu.datatypes.Hash;
 import org.hyperledger.besu.ethereum.chain.BlockchainStorage;
 
 import java.nio.file.Path;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
@@ -45,7 +46,7 @@ public class PrunePreMergeBlockDataSubCommand implements Runnable {
   private static final long MAINNET_MERGE_BLOCK_NUMBER = 15_537_393;
   private static final long SEPOLIA_MERGE_BLOCK_NUMBER = 1_735_371;
 
-  private static final int DEFAULT_THREADS = 4;
+  private static final int DEFAULT_THREADS = Runtime.getRuntime().availableProcessors() - 1;
   private static final int DEFAULT_PRUNE_RANGE_SIZE = 10000;
 
   @SuppressWarnings("unused")
@@ -85,6 +86,9 @@ public void run() {
         network,
         dataPath);
     final long mergeBlockNumber = getMergeBlockNumber(network);
+    LOG.info("Parallelizing with number of threads: {}", threads);
+    LOG.info("Merge block number: {}", mergeBlockNumber);
+    LOG.info("Prune range size: {}", pruneRangeSize);
 
     try (BesuController besuController = storageSubCommand.besuCommand.buildController()) {
 
@@ -97,13 +101,22 @@ public void run() {
                   besuController.getDataStorageConfiguration());
 
       try (ExecutorService executor = Executors.newFixedThreadPool(threads)) {
+        // cannot prune genesis block so start at 1
         for (long i = 1; i < mergeBlockNumber; i += pruneRangeSize) {
-          final long headerNumber = i;
+          LOG.info(
+              "Starting pruning of block range {} to {}...",
+              i,
+              Math.min(i + pruneRangeSize, mergeBlockNumber));
+          final long startBlockNumber = i;
+          final long endBlockNumber = Math.min(startBlockNumber + pruneRangeSize, mergeBlockNumber);
           executor.submit(
-              () -> deleteBlockRange(headerNumber, mergeBlockNumber, blockchainStorage));
+              () ->
+                  deleteBlockRange(
+                      startBlockNumber, endBlockNumber, mergeBlockNumber, blockchainStorage));
         }
       }
     }
+    LOG.info("Pruning pre-merge blocks and transaction receipts completed");
   }
 
   private static long getMergeBlockNumber(final NetworkName network) {
@@ -116,31 +129,25 @@ private static long getMergeBlockNumber(final NetworkName network) {
 
   private void deleteBlockRange(
       final long startBlockNumber,
+      final long endBlockNumber,
       final long mergeBlockNumber,
       final BlockchainStorage blockchainStorage) {
     BlockchainStorage.Updater updater = blockchainStorage.updater();
     long headerNumber = startBlockNumber;
-    final long endBlockNumber = Math.min(startBlockNumber + pruneRangeSize, mergeBlockNumber);
     do {
-      blockchainStorage
-          .getBlockHash(headerNumber)
-          .filter((h) -> blockchainStorage.getBlockBody(h).isPresent())
-          .ifPresent(
-              (h) -> {
-                updater.removeTransactionReceipts(h);
-                updater.removeTotalDifficulty(h);
-                blockchainStorage
-                    .getBlockBody(h)
-                    .map((bb) -> bb.getTransactions())
-                    .ifPresent(
-                        (transactions) ->
-                            transactions.stream()
-                                .map(Transaction::getHash)
-                                .forEach((th) -> updater.removeTransactionLocation(th)));
-                updater.removeBlockBody(h);
-              });
+      final Optional<Hash> maybeBlockHash = blockchainStorage.getBlockHash(headerNumber);
+      if (maybeBlockHash.isEmpty()) {
+        continue;
+      }
+      final Hash h = maybeBlockHash.get();
+      updater.removeTransactionReceipts(h);
+      updater.removeBlockBody(h);
     } while (++headerNumber < endBlockNumber);
     updater.commit();
-    LOG.info("Completed deletion of block range {} to {}", startBlockNumber, endBlockNumber);
+    LOG.info(
+        "...completed pruning of block range {} to {}; estimated {} blocks remaining",
+        startBlockNumber,
+        endBlockNumber,
+        mergeBlockNumber - endBlockNumber);
   }
 }

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/storage/keyvalue/KeyValueSegmentIdentifier.java

@@ -25,7 +25,7 @@
 
 public enum KeyValueSegmentIdentifier implements SegmentIdentifier {
   DEFAULT("default".getBytes(StandardCharsets.UTF_8)),
-  BLOCKCHAIN(new byte[] {1}, true, true),
+  BLOCKCHAIN(new byte[] {1}, EnumSet.allOf(DataStorageFormat.class), true, true, false),
   WORLD_STATE(new byte[] {2}, EnumSet.of(FOREST), false, true, false),
   PRIVATE_TRANSACTIONS(new byte[] {3}),
   PRIVATE_STATE(new byte[] {4}),
@@ -58,16 +58,6 @@ public enum KeyValueSegmentIdentifier implements SegmentIdentifier {
     this(id, EnumSet.allOf(DataStorageFormat.class));
   }
 
-  KeyValueSegmentIdentifier(
-      final byte[] id, final boolean containsStaticData, final boolean eligibleToHighSpecFlag) {
-    this(
-        id,
-        EnumSet.allOf(DataStorageFormat.class),
-        containsStaticData,
-        eligibleToHighSpecFlag,
-        false);
-  }
-
   KeyValueSegmentIdentifier(final byte[] id, final EnumSet<DataStorageFormat> formats) {
     this(id, formats, false, false, false);
   }