Logs Index RPC API

besu/build.gradle

@@ -39,7 +39,6 @@ dependencies {
   implementation project(':ethereum:core')
   implementation project(':ethereum:eth')
   implementation project(':ethereum:api')
-  implementation project(':ethereum:api')
   implementation project(':ethereum:permissioning')
   implementation project(':ethereum:p2p')
   implementation project(':ethereum:retesteth')

besu/src/main/java/org/hyperledger/besu/RunnerBuilder.java

@@ -358,7 +358,11 @@ public Runner build() {
     final MiningCoordinator miningCoordinator = besuController.getMiningCoordinator();
 
     final BlockchainQueries blockchainQueries =
-        new BlockchainQueries(context, Optional.of(dataDir.resolve(CACHE_PATH)));
+        new BlockchainQueries(
+            context.getBlockchain(),
+            context.getWorldStateArchive(),
+            Optional.of(dataDir.resolve(CACHE_PATH)),
+            Optional.of(besuController.getProtocolManager().ethContext().getScheduler()));
 
     final PrivacyParameters privacyParameters = besuController.getPrivacyParameters();
     final FilterManager filterManager =

besu/src/main/java/org/hyperledger/besu/cli/subcommands/operator/GenerateLogBloomCache.java

@@ -16,21 +16,19 @@
 
 package org.hyperledger.besu.cli.subcommands.operator;
 
-import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 import static org.hyperledger.besu.cli.DefaultCommandValues.MANDATORY_LONG_FORMAT_HELP;
+import static org.hyperledger.besu.ethereum.api.query.TransactionLogsIndexer.BLOCKS_PER_BLOOM_CACHE;
 
 import org.hyperledger.besu.controller.BesuController;
-import org.hyperledger.besu.ethereum.chain.Blockchain;
-import org.hyperledger.besu.plugin.data.BlockHeader;
+import org.hyperledger.besu.ethereum.api.query.TransactionLogsIndexer;
+import org.hyperledger.besu.ethereum.chain.MutableBlockchain;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
+import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
 
-import java.io.File;
-import java.io.FileOutputStream;
 import java.nio.file.Path;
 
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 import picocli.CommandLine.Command;
 import picocli.CommandLine.Option;
 import picocli.CommandLine.ParentCommand;
@@ -40,90 +38,56 @@
     description = "Generate cached values of block log bloom filters.",
     mixinStandardHelpOptions = true)
 public class GenerateLogBloomCache implements Runnable {
-  private static final Logger LOG = LogManager.getLogger();
-
-  public static final int BLOCKS_PER_FILE = 100_000;
-  public static final String CACHE_DIRECTORY_NAME = "caches";
 
   @Option(
       names = "--start-block",
       paramLabel = MANDATORY_LONG_FORMAT_HELP,
       description =
           "The block to start generating indexes.  Must be an increment of "
-              + BLOCKS_PER_FILE
+              + BLOCKS_PER_BLOOM_CACHE
               + " (default: ${DEFAULT-VALUE})",
       arity = "1..1")
   private final Long startBlock = 0L;
 
   @Option(
       names = "--end-block",
       paramLabel = MANDATORY_LONG_FORMAT_HELP,
-      description =
-          "The block to start generating indexes (exclusive). (default: last block divisible by "
-              + BLOCKS_PER_FILE
-              + ")",
+      description = "The block to stop generating indexes (default is last block of the chain).",
       arity = "1..1")
-  private final Long endBlock = -1L;
+  private final Long endBlock = Long.MAX_VALUE;
 
   @ParentCommand private OperatorSubCommand parentCommand;
 
+  @SuppressWarnings("ResultOfMethodCallIgnored")
   @Override
   public void run() {
     checkPreconditions();
-    generateLogBloomCache();
-  }
-
-  private void checkPreconditions() {
-    checkNotNull(parentCommand.parentCommand.dataDir());
-    checkState(
-        startBlock % BLOCKS_PER_FILE == 0,
-        "Start block must be an even increment of %s",
-        BLOCKS_PER_FILE);
-  }
-
-  @SuppressWarnings("ResultOfMethodCallIgnored")
-  private void generateLogBloomCache() {
-    final Path cacheDir = parentCommand.parentCommand.dataDir().resolve(CACHE_DIRECTORY_NAME);
+    final Path cacheDir = parentCommand.parentCommand.dataDir().resolve(BesuController.CACHE_PATH);
     cacheDir.toFile().mkdirs();
-    generateLogBloomCache(
-        startBlock,
-        endBlock,
-        cacheDir,
-        createBesuController().getProtocolContext().getBlockchain());
-  }
-
-  public static void generateLogBloomCache(
-      final long start, final long stop, final Path cacheDir, final Blockchain blockchain) {
-    final long stopBlock =
-        stop < 0
-            ? (blockchain.getChainHeadBlockNumber() / BLOCKS_PER_FILE) * BLOCKS_PER_FILE
-            : stop;
-    LOG.debug("Start block: {} Stop block: {} Path: {}", start, stopBlock, cacheDir);
-    checkArgument(start % BLOCKS_PER_FILE == 0, "Start block must be at the beginning of a file");
-    checkArgument(stopBlock % BLOCKS_PER_FILE == 0, "End block must be at the beginning of a file");
+    final MutableBlockchain blockchain =
+        createBesuController().getProtocolContext().getBlockchain();
+    final EthScheduler scheduler = new EthScheduler(1, 1, 1, 1, new NoOpMetricsSystem());
     try {
-      FileOutputStream fos = null;
-      for (long blockNum = start; blockNum < stopBlock; blockNum++) {
-        if (blockNum % BLOCKS_PER_FILE == 0 || fos == null) {
-          LOG.info("Indexing block {}", blockNum);
-          if (fos != null) {
-            fos.close();
-          }
-          fos = new FileOutputStream(createCacheFile(blockNum, cacheDir));
-        }
-        final BlockHeader header = blockchain.getBlockHeader(blockNum).orElseThrow();
-        final byte[] logs = header.getLogsBloom().getByteArray();
-        checkNotNull(logs);
-        checkState(logs.length == 256, "BloomBits are not the correct length");
-        fos.write(logs);
+      final long finalBlock = Math.min(blockchain.getChainHeadBlockNumber(), endBlock);
+      final TransactionLogsIndexer indexer =
+          new TransactionLogsIndexer(blockchain, cacheDir, scheduler);
+      indexer.generateLogBloomCache(startBlock, finalBlock);
+    } finally {
+      scheduler.stop();
+      try {
+        scheduler.awaitStop();
+      } catch (final InterruptedException e) {
+        // ignore
       }
-    } catch (final Exception e) {
-      LOG.error("Unhandled indexing exception", e);
     }
   }
 
-  private static File createCacheFile(final long blockNumber, final Path cacheDir) {
-    return cacheDir.resolve("logBloom-" + (blockNumber / BLOCKS_PER_FILE) + ".index").toFile();
+  private void checkPreconditions() {
+    checkNotNull(parentCommand.parentCommand.dataDir());
+    checkState(
+        startBlock % BLOCKS_PER_BLOOM_CACHE == 0,
+        "Start block must be an even increment of %s",
+        BLOCKS_PER_BLOOM_CACHE);
   }
 
   private BesuController<?> createBesuController() {

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/RpcMethod.java

@@ -23,6 +23,7 @@ public enum RpcMethod {
   ADMIN_PEERS("admin_peers"),
   ADMIN_REMOVE_PEER("admin_removePeer"),
   ADMIN_CHANGE_LOG_LEVEL("admin_changeLogLevel"),
+  ADMIN_GENERATE_LOG_BLOOM_CACHE("admin_generateLogBloomCache"),
   CLIQUE_DISCARD("clique_discard"),
   CLIQUE_GET_SIGNERS("clique_getSigners"),
   CLIQUE_GET_SIGNERS_AT_HASH("clique_getSignersAtHash"),

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/methods/AdminGenerateLogBloomCache.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright ConsenSys AG.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods;
+
+import org.hyperledger.besu.ethereum.api.jsonrpc.RpcMethod;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestContext;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.parameters.BlockParameter;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponse;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcSuccessResponse;
+import org.hyperledger.besu.ethereum.api.query.BlockchainQueries;
+
+import java.util.Optional;
+
+public class AdminGenerateLogBloomCache implements JsonRpcMethod {
+
+  private final BlockchainQueries blockchainQueries;
+
+  public AdminGenerateLogBloomCache(final BlockchainQueries blockchainQueries) {
+    this.blockchainQueries = blockchainQueries;
+  }
+
+  @Override
+  public String getName() {
+    return RpcMethod.ADMIN_GENERATE_LOG_BLOOM_CACHE.getMethodName();
+  }
+
+  @Override
+  public JsonRpcResponse response(final JsonRpcRequestContext requestContext) {
+    final Optional<BlockParameter> startBlockParam =
+        requestContext.getOptionalParameter(0, BlockParameter.class);
+    final long startBlock;
+    if (startBlockParam.isEmpty() || startBlockParam.get().isEarliest()) {
+      startBlock = 0;
+    } else if (startBlockParam.get().getNumber().isPresent()) {
+      startBlock = startBlockParam.get().getNumber().getAsLong();
+    } else {
+      // latest, pending
+      startBlock = Long.MAX_VALUE;
+    }
+
+    final Optional<BlockParameter> stopBlockParam =
+        requestContext.getOptionalParameter(1, BlockParameter.class);
+    final long stopBlock;
+    if (stopBlockParam.isEmpty()) {
+      if (startBlockParam.isEmpty()) {
+        stopBlock = -1L;
+      } else {
+        stopBlock = Long.MAX_VALUE;
+      }
+    } else if (stopBlockParam.get().isEarliest()) {
+      stopBlock = 0;
+    } else if (stopBlockParam.get().getNumber().isPresent()) {
+      stopBlock = stopBlockParam.get().getNumber().getAsLong();
+    } else {
+      // latest, pending
+      stopBlock = Long.MAX_VALUE;
+    }
+
+    return new JsonRpcSuccessResponse(
+        requestContext.getRequest().getId(),
+        blockchainQueries
+            .getTransactionLogsIndexer()
+            .map(indexer -> indexer.requestIndexing(startBlock, stopBlock))
+            .orElse(null));
+  }
+}

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/methods/AdminJsonRpcMethods.java

@@ -19,6 +19,7 @@
 import org.hyperledger.besu.ethereum.api.jsonrpc.RpcApis;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.AdminAddPeer;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.AdminChangeLogLevel;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.AdminGenerateLogBloomCache;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.AdminNodeInfo;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.AdminPeers;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.AdminRemovePeer;
@@ -63,6 +64,7 @@ protected Map<String, JsonRpcMethod> create() {
         new AdminNodeInfo(
             clientVersion, networkId, genesisConfigOptions, p2pNetwork, blockchainQueries),
         new AdminPeers(p2pNetwork),
-        new AdminChangeLogLevel());
+        new AdminChangeLogLevel(),
+        new AdminGenerateLogBloomCache(blockchainQueries));
   }
 }

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/query/BlockchainQueries.java

@@ -15,8 +15,8 @@
 package org.hyperledger.besu.ethereum.api.query;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static org.hyperledger.besu.ethereum.api.query.TransactionLogsIndexer.BLOCKS_PER_BLOOM_CACHE;
 
-import org.hyperledger.besu.ethereum.ProtocolContext;
 import org.hyperledger.besu.ethereum.chain.Blockchain;
 import org.hyperledger.besu.ethereum.chain.TransactionLocation;
 import org.hyperledger.besu.ethereum.core.Account;
@@ -32,6 +32,7 @@
 import org.hyperledger.besu.ethereum.core.TransactionReceipt;
 import org.hyperledger.besu.ethereum.core.Wei;
 import org.hyperledger.besu.ethereum.core.WorldState;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
 import org.hyperledger.besu.util.bytes.BytesValue;
 import org.hyperledger.besu.util.uint.UInt256;
@@ -51,33 +52,40 @@
 import java.util.stream.IntStream;
 import java.util.stream.LongStream;
 
-import com.google.common.annotations.VisibleForTesting;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 public class BlockchainQueries {
   private static final Logger LOG = LogManager.getLogger();
-  @VisibleForTesting static final long BLOCKS_PER_BLOOM_CACHE = 100_000;
 
   private final WorldStateArchive worldStateArchive;
   private final Blockchain blockchain;
   private final Optional<Path> cachePath;
+  private final Optional<TransactionLogsIndexer> transactionLogsIndexer;
 
   public BlockchainQueries(final Blockchain blockchain, final WorldStateArchive worldStateArchive) {
-    this(blockchain, worldStateArchive, Optional.empty());
+    this(blockchain, worldStateArchive, Optional.empty(), Optional.empty());
   }
 
-  public BlockchainQueries(final ProtocolContext<?> context, final Optional<Path> cachePath) {
-    this(context.getBlockchain(), context.getWorldStateArchive(), cachePath);
+  public BlockchainQueries(
+      final Blockchain blockchain,
+      final WorldStateArchive worldStateArchive,
+      final EthScheduler scheduler) {
+    this(blockchain, worldStateArchive, Optional.empty(), Optional.ofNullable(scheduler));
   }
 
-  private BlockchainQueries(
+  public BlockchainQueries(
       final Blockchain blockchain,
       final WorldStateArchive worldStateArchive,
-      final Optional<Path> cachePath) {
+      final Optional<Path> cachePath,
+      final Optional<EthScheduler> scheduler) {
     this.blockchain = blockchain;
     this.worldStateArchive = worldStateArchive;
     this.cachePath = cachePath;
+    this.transactionLogsIndexer =
+        (cachePath.isPresent() && scheduler.isPresent())
+            ? Optional.of(new TransactionLogsIndexer(blockchain, cachePath.get(), scheduler.get()))
+            : Optional.empty();
   }
 
   public Blockchain getBlockchain() {
@@ -88,6 +96,10 @@ public WorldStateArchive getWorldStateArchive() {
     return worldStateArchive;
   }
 
+  public Optional<TransactionLogsIndexer> getTransactionLogsIndexer() {
+    return transactionLogsIndexer;
+  }
+
   /**
    * Retrieves the header hash of the block at the given height in the canonical chain.
    *