fix(jsonrpc): enforce log filter cap and improve match efficiency

common/src/main/java/org/tron/common/parameter/CommonParameter.java

@@ -459,6 +459,9 @@ public class CommonParameter {
   @Getter
   @Setter
   public int jsonRpcMaxBlockFilterNum = 50000;
+  @Getter
+  @Setter
+  public int jsonRpcMaxLogFilterNum = 20000;
 
   @Getter
   @Setter

framework/src/main/java/org/tron/core/config/args/Args.java

@@ -265,6 +265,11 @@ public static void applyConfigParams(
           config.getInt(ConfigKey.NODE_JSONRPC_MAX_BLOCK_FILTER_NUM);
     }
 
+    if (config.hasPath(ConfigKey.NODE_JSONRPC_MAX_LOG_FILTER_NUM)) {
+      PARAMETER.jsonRpcMaxLogFilterNum =
+          config.getInt(ConfigKey.NODE_JSONRPC_MAX_LOG_FILTER_NUM);
+    }
+
     if (config.hasPath(ConfigKey.VM_MIN_TIME_RATIO)) {
       PARAMETER.minTimeRatio = config.getDouble(ConfigKey.VM_MIN_TIME_RATIO);
     }

framework/src/main/java/org/tron/core/config/args/ConfigKey.java

@@ -150,6 +150,8 @@ private ConfigKey() {
   public static final String NODE_JSONRPC_MAX_SUB_TOPICS = "node.jsonrpc.maxSubTopics";
   public static final String NODE_JSONRPC_MAX_BLOCK_FILTER_NUM =
       "node.jsonrpc.maxBlockFilterNum";
+  public static final String NODE_JSONRPC_MAX_LOG_FILTER_NUM =
+      "node.jsonrpc.maxLogFilterNum";
 
   // node - dns
   public static final String NODE_DNS_TREE_URLS = "node.dns.treeUrls";

framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpc.java

@@ -291,9 +291,10 @@ CompilationResult ethSubmitHashrate(String hashrate, String id)
   @JsonRpcErrors({
       @JsonRpcError(exception = JsonRpcMethodNotFoundException.class, code = -32601, data = "{}"),
       @JsonRpcError(exception = JsonRpcInvalidParamsException.class, code = -32602, data = "{}"),
+      @JsonRpcError(exception = JsonRpcExceedLimitException.class, code = -32005, data = "{}"),
   })
   String newFilter(FilterRequest fr) throws JsonRpcInvalidParamsException,
-      JsonRpcMethodNotFoundException;
+      JsonRpcMethodNotFoundException, JsonRpcExceedLimitException;
 
   @JsonRpcMethod("eth_newBlockFilter")
   @JsonRpcErrors({

framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java

@@ -27,6 +27,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
 import lombok.Getter;
@@ -115,6 +116,7 @@ public enum RequestSource {
   private static final String FILTER_NOT_FOUND = "filter not found";
   public static final int EXPIRE_SECONDS = 5 * 60;
   private static final int maxBlockFilterNum = Args.getInstance().getJsonRpcMaxBlockFilterNum();
+  private static final int maxLogFilterNum = Args.getInstance().getJsonRpcMaxLogFilterNum();
   private static final Cache<LogFilterElement, LogFilterElement> logElementCache =
       CacheBuilder.newBuilder()
           .maximumSize(300_000L) // 300s * tps(1000) * 1 log/tx ≈ 300_000
@@ -169,6 +171,8 @@ public enum RequestSource {
   private static final String NO_BLOCK_HEADER_BY_HASH = "header for hash not found";
 
   private static final String ERROR_SELECTOR = "08c379a0"; // Function selector for Error(string)
+  private static final int FILTER_PARALLEL_THRESHOLD = 10000;
+  private static final ForkJoinPool LOGS_FILTER_POOL = new ForkJoinPool(2);
   /**
    * thread pool of query section bloom store
    */
@@ -222,53 +226,68 @@ public static void handleBLockFilter(BlockFilterCapsule blockFilterCapsule) {
    * append LogsFilterCapsule's LogFilterElement list to each filter if matched
    */
   public static void handleLogsFilter(LogsFilterCapsule logsFilterCapsule) {
-    Iterator<Entry<String, LogFilterAndResult>> it;
+    long t1 = System.currentTimeMillis();
+    Map<String, LogFilterAndResult> eventFilterMap;
 
     if (logsFilterCapsule.isSolidified()) {
-      it = getEventFilter2ResultSolidity().entrySet().iterator();
+      eventFilterMap = getEventFilter2ResultSolidity();
     } else {
-      it = getEventFilter2ResultFull().entrySet().iterator();
+      eventFilterMap = getEventFilter2ResultFull();
     }
 
-    while (it.hasNext()) {
-      Entry<String, LogFilterAndResult> entry = it.next();
-      if (entry.getValue().isExpire()) {
-        it.remove();
-        continue;
-      }
+    if (eventFilterMap.size() <= FILTER_PARALLEL_THRESHOLD) {
+      eventFilterMap.entrySet().forEach(
+          entry -> processLogFilterEntry(entry, eventFilterMap, logsFilterCapsule));
+    } else {
+      LOGS_FILTER_POOL.submit(() -> eventFilterMap.entrySet().parallelStream()
+          .forEach(entry -> processLogFilterEntry(entry, eventFilterMap, logsFilterCapsule))
+      ).join();
+    }
+    long t2 = System.currentTimeMillis();
+    logger.debug("handleLogsFilter {} cost {}, filter size {}",
+        logsFilterCapsule.isSolidified() ? "Solidity" : "Full", t2 - t1, eventFilterMap.size());
+  }
+
+  private static void processLogFilterEntry(
+      Map.Entry<String, LogFilterAndResult> entry,
+      Map<String, LogFilterAndResult> eventFilterMap,
+      LogsFilterCapsule logsFilterCapsule) {
+    LogFilterAndResult logFilterAndResult = entry.getValue();
+    if (logFilterAndResult.isExpire()) {
+      eventFilterMap.remove(entry.getKey());
+      return;
+    }
 
-      LogFilterAndResult logFilterAndResult = entry.getValue();
-      long fromBlock = logFilterAndResult.getLogFilterWrapper().getFromBlock();
-      long toBlock = logFilterAndResult.getLogFilterWrapper().getToBlock();
-      if (!(fromBlock <= logsFilterCapsule.getBlockNumber()
-          && logsFilterCapsule.getBlockNumber() <= toBlock)) {
-        continue;
-      }
+    long blockNumber = logsFilterCapsule.getBlockNumber();
+    long fromBlock = logFilterAndResult.getLogFilterWrapper().getFromBlock();
+    long toBlock = logFilterAndResult.getLogFilterWrapper().getToBlock();
+    if (!(fromBlock <= blockNumber && blockNumber <= toBlock)) {
+      return;
+    }
 
-      if (logsFilterCapsule.getBloom() != null
-          && !logFilterAndResult.getLogFilterWrapper().getLogFilter()
-          .matchBloom(logsFilterCapsule.getBloom())) {
-        continue;
-      }
+    if (logsFilterCapsule.getBloom() != null && !logFilterAndResult.getLogFilterWrapper()
+        .getLogFilter().matchBloom(logsFilterCapsule.getBloom())) {
+      return;
+    }
 
-      LogFilter logFilter = logFilterAndResult.getLogFilterWrapper().getLogFilter();
-      List<LogFilterElement> elements =
-          LogMatch.matchBlock(logFilter, logsFilterCapsule.getBlockNumber(),
-              logsFilterCapsule.getBlockHash(), logsFilterCapsule.getTxInfoList(),
-              logsFilterCapsule.isRemoved());
+    LogFilter logFilter = logFilterAndResult.getLogFilterWrapper().getLogFilter();
+    List<LogFilterElement> elements =
+        LogMatch.matchBlock(logFilter, blockNumber, logsFilterCapsule.getBlockHash(),
+            logsFilterCapsule.getTxInfoList(), logsFilterCapsule.isRemoved());
 
-      for (LogFilterElement element : elements) {
-        LogFilterElement cachedElement;
-        try {
-          // compare with hashcode() first, then with equals(). If not exist, put it.
-          cachedElement = logElementCache.get(element, () -> element);
-        } catch (ExecutionException e) {
-          logger.error("Getting/loading LogFilterElement from cache fails", e); // never happen
-          cachedElement = element;
-        }
-        logFilterAndResult.getResult().add(cachedElement);
+    List<LogFilterElement> localResults = new ArrayList<>(elements.size());
+    for (LogFilterElement element : elements) {
+      LogFilterElement cachedElement;
+      try {
+        // compare with hashcode() first, then with equals(). If not exist, put it.
+        cachedElement = logElementCache.get(element, () -> element);
+      } catch (ExecutionException e) {
+        logger.error("Getting/loading LogFilterElement from cache fails", e); // never happen
+        cachedElement = element;
       }
+      localResults.add(cachedElement);
     }
+    logFilterAndResult.getResult().addAll(localResults);
   }
 
   @Override
@@ -1406,7 +1425,7 @@ public CompilationResult ethSubmitHashrate(String hashrate, String id)
 
   @Override
   public String newFilter(FilterRequest fr) throws JsonRpcInvalidParamsException,
-      JsonRpcMethodNotFoundException {
+      JsonRpcMethodNotFoundException, JsonRpcExceedLimitException {
     disableInPBFT("eth_newFilter");
 
     // not supports finalized as block parameter
@@ -1421,7 +1440,10 @@ public String newFilter(FilterRequest fr) throws JsonRpcInvalidParamsException,
     } else {
       eventFilter2Result = eventFilter2ResultSolidity;
     }
-
+    if (eventFilter2Result.size() >= maxLogFilterNum) {
+      throw new JsonRpcExceedLimitException(
+          "exceed max log filters: " + maxLogFilterNum + ", try again later");
+    }
     long currentMaxFullNum = wallet.getNowBlock().getBlockHeader().getRawData().getNumber();
     LogFilterAndResult logFilterAndResult = new LogFilterAndResult(fr, currentMaxFullNum, wallet);
     String filterID = generateFilterId();
@@ -1573,6 +1595,7 @@ public static Object[] getFilterResult(String filterId, Map<String, BlockFilterA
 
   @Override
   public void close() throws IOException {
+    ExecutorServiceManager.shutdownAndAwaitTermination(LOGS_FILTER_POOL, "logs-filter-pool");
     logElementCache.invalidateAll();
     blockHashCache.invalidateAll();
     ExecutorServiceManager.shutdownAndAwaitTermination(sectionExecutor, esName);

framework/src/main/java/org/tron/core/services/jsonrpc/filters/LogMatch.java

@@ -93,14 +93,14 @@ public LogFilterElement[] matchBlockOneByOne()
       String blockHash = manager.getChainBaseManager().getBlockIdByNum(blockNum).toString();
       List<LogFilterElement> matchedLog = matchBlock(logFilterWrapper.getLogFilter(), blockNum,
           blockHash, transactionInfoList, false);
+
       if (!matchedLog.isEmpty()) {
+        if (logFilterElementList.size() + matchedLog.size() > LogBlockQuery.MAX_RESULT) {
+          throw new JsonRpcTooManyResultException(
+              "query returned more than " + LogBlockQuery.MAX_RESULT + " results");
+        }
         logFilterElementList.addAll(matchedLog);
       }
-
-      if (logFilterElementList.size() > LogBlockQuery.MAX_RESULT) {
-        throw new JsonRpcTooManyResultException(
-            "query returned more than " + LogBlockQuery.MAX_RESULT + " results");
-      }
     }
 
     return logFilterElementList.toArray(new LogFilterElement[0]);

framework/src/main/resources/config.conf

@@ -375,6 +375,8 @@ node {
     maxSubTopics = 1000
     # Allowed maximum number for blockFilter
     maxBlockFilterNum = 50000
+    # Allowed maximum number for newFilter
+    maxLogFilterNum = 20000
   }
 
   # Disabled api list, it will work for http, rpc and pbft, both FullNode and SolidityNode,

framework/src/test/java/org/tron/core/jsonrpc/WalletCursorTest.java

@@ -87,6 +87,7 @@ public void testDisableInSolidity() {
     TronJsonRpcImpl tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet, dbManager);
     try {
       tronJsonRpc.buildTransaction(buildArguments);
+      tronJsonRpc.close();
     } catch (Exception e) {
       Assert.assertEquals("the method buildTransaction does not exist/is not available in "
           + "SOLIDITY", e.getMessage());
@@ -136,6 +137,7 @@ public void testEnableInFullNode() {
 
     try {
       tronJsonRpc.buildTransaction(buildArguments);
+      tronJsonRpc.close();
     } catch (Exception e) {
       Assert.fail();
     }

framework/src/test/resources/config-localtest.conf

@@ -168,6 +168,7 @@ node {
     # maxBlockRange = 5000
     # maxSubTopics = 1000
     # maxBlockFilterNum = 30000
+    # maxLogFilterNum = 20000
   }
 
 }

framework/src/test/resources/config-test-mainnet.conf

@@ -95,6 +95,7 @@ node {
     # maxBlockRange = 5000
     # maxSubTopics = 1000
     # maxBlockFilterNum = 50000
+    # maxLogFilterNum = 20000
   }
 
   rpc {

framework/src/test/resources/config-test.conf

@@ -119,6 +119,7 @@ node {
     # maxBlockRange = 5000
     # maxSubTopics = 1000
     # maxBlockFilterNum = 30000
+    # maxLogFilterNum = 20000
   }
 
   # use your ipv6 address for node discovery and tcp connection, default false