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

Author: 317787106

common/src/main/java/org/tron/common/es/ExecutorServiceManager.java

@@ -4,11 +4,14 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinWorkerThread;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import lombok.extern.slf4j.Slf4j;
 import org.tron.common.exit.ExitManager;
 
@@ -35,6 +38,22 @@ public static ScheduledExecutorService newSingleThreadScheduledExecutor(String n
         new ThreadFactoryBuilder().setNameFormat(name).setDaemon(isDaemon).build());
   }
 
+  public static ForkJoinPool newForkJoinPool(String name, int parallelism) {
+    return newForkJoinPool(name, parallelism, false);
+  }
+
+  public static ForkJoinPool newForkJoinPool(String name, int parallelism, boolean isDaemon) {
+    AtomicInteger counter = new AtomicInteger(0);
+    ForkJoinPool.ForkJoinWorkerThreadFactory factory = pool -> {
+      ForkJoinWorkerThread thread =
+          ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
+      thread.setName(name + "-" + counter.getAndIncrement());
+      thread.setDaemon(isDaemon);
+      return thread;
+    };
+    return new ForkJoinPool(parallelism, factory, null, false);
+  }
+
   public static ExecutorService newFixedThreadPool(String name, int fixThreads) {
     return newFixedThreadPool(name, fixThreads, false);
   }

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

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

common/src/main/java/org/tron/core/config/args/NodeConfig.java

@@ -311,6 +311,7 @@ public void setHttpPBFTPort(int v) {
     private int maxBlockRange = 5000;
     private int maxSubTopics = 1000;
     private int maxBlockFilterNum = 50000;
+    private int maxLogFilterNum = 20000;
     private long maxMessageSize = 4194304;
   }
 

common/src/main/resources/reference.conf

@@ -386,9 +386,12 @@ node {
     # Maximum topics within a topic criteria, >0 otherwise no limit
     maxSubTopics = 1000
 
-    # Maximum number for blockFilter
+    # Maximum number for blockFilter. >0 otherwise no limit
     maxBlockFilterNum = 50000
 
+    # Maximum number of concurrent eth_newFilter registrations, >0 otherwise no limit
+    maxLogFilterNum = 20000
+
     # Maximum JSON-RPC request body size, default 4MB. Independent from rpc.maxMessageSize.
     maxMessageSize = 4M
   }

framework/src/main/java/org/tron/common/logsfilter/capsule/BlockFilterCapsule.java

@@ -1,7 +1,5 @@
 package org.tron.common.logsfilter.capsule;
 
-import static org.tron.core.services.jsonrpc.TronJsonRpcImpl.handleBLockFilter;
-
 import lombok.Getter;
 import lombok.Setter;
 import lombok.ToString;
@@ -20,19 +18,12 @@ public class BlockFilterCapsule extends FilterTriggerCapsule {
   private boolean solidified;
 
   public BlockFilterCapsule(BlockCapsule block, boolean solidified) {
-    blockHash = block.getBlockId().toString();
-    this.solidified = solidified;
+    this(block.getBlockId().toString(), solidified);
   }
 
   public BlockFilterCapsule(String blockHash, boolean solidified) {
     this.blockHash = blockHash;
     this.solidified = solidified;
   }
 
-  @Override
-  public void processFilterTrigger() {
-    handleBLockFilter(this);
-  }
-
 }
-

framework/src/main/java/org/tron/common/logsfilter/capsule/FilterTriggerCapsule.java

@@ -1,8 +1,5 @@
 package org.tron.common.logsfilter.capsule;
 
-public class FilterTriggerCapsule extends TriggerCapsule {
+public class FilterTriggerCapsule {
 
-  public void processFilterTrigger() {
-    throw new UnsupportedOperationException();
-  }
 }

framework/src/main/java/org/tron/common/logsfilter/capsule/LogsFilterCapsule.java

@@ -1,7 +1,5 @@
 package org.tron.common.logsfilter.capsule;
 
-import static org.tron.core.services.jsonrpc.TronJsonRpcImpl.handleLogsFilter;
-
 import java.util.List;
 import lombok.Getter;
 import lombok.Setter;
@@ -43,8 +41,4 @@ public LogsFilterCapsule(long blockNumber, String blockHash, Bloom bloom,
     this.removed = removed;
   }
 
-  @Override
-  public void processFilterTrigger() {
-    handleLogsFilter(this);
-  }
-}
+}

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

@@ -559,6 +559,7 @@ private static void applyNodeConfig(NodeConfig nc) {
     PARAMETER.jsonRpcMaxBlockRange = jsonrpc.getMaxBlockRange();
     PARAMETER.jsonRpcMaxSubTopics = jsonrpc.getMaxSubTopics();
     PARAMETER.jsonRpcMaxBlockFilterNum = jsonrpc.getMaxBlockFilterNum();
+    PARAMETER.jsonRpcMaxLogFilterNum = jsonrpc.getMaxLogFilterNum();
     PARAMETER.jsonRpcMaxMessageSize = jsonrpc.getMaxMessageSize();
 
     // ---- P2P sub-bean ----

framework/src/main/java/org/tron/core/db/Manager.java

@@ -48,6 +48,7 @@
 import org.apache.commons.collections4.CollectionUtils;
 import org.bouncycastle.util.encoders.Hex;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Component;
 import org.tron.api.GrpcAPI;
 import org.tron.api.GrpcAPI.TransactionInfoList;
@@ -142,6 +143,7 @@
 import org.tron.core.service.MortgageService;
 import org.tron.core.service.RewardViCalService;
 import org.tron.core.services.event.exception.EventException;
+import org.tron.core.services.jsonrpc.TronJsonRpcImpl;
 import org.tron.core.store.AccountAssetStore;
 import org.tron.core.store.AccountIdIndexStore;
 import org.tron.core.store.AccountIndexStore;
@@ -277,6 +279,10 @@ public class Manager {
   @Autowired
   private RewardViCalService rewardViCalService;
 
+  @Lazy
+  @Autowired
+  private TronJsonRpcImpl tronJsonRpcImpl;
+
   /**
    * Cycle thread to rePush Transactions
    */
@@ -333,8 +339,10 @@ public class Manager {
         while (isRunFilterProcessThread) {
           try {
             FilterTriggerCapsule filterCapsule = filterCapsuleQueue.poll(1, TimeUnit.SECONDS);
-            if (filterCapsule != null) {
-              filterCapsule.processFilterTrigger();
+            if (filterCapsule instanceof LogsFilterCapsule) {
+              tronJsonRpcImpl.handleLogsFilter((LogsFilterCapsule) filterCapsule);
+            } else if (filterCapsule instanceof BlockFilterCapsule) {
+              tronJsonRpcImpl.handleBLockFilter((BlockFilterCapsule) filterCapsule);
             }
           } catch (InterruptedException e) {
             logger.error("FilterProcessLoop get InterruptedException, error is {}.",
@@ -2268,7 +2276,8 @@ private void reOrgLogsFilter() {
   }
 
   private void postBlockFilter(final BlockCapsule blockCapsule, boolean solidified) {
-    BlockFilterCapsule blockFilterCapsule = new BlockFilterCapsule(blockCapsule, solidified);
+    BlockFilterCapsule blockFilterCapsule =
+        new BlockFilterCapsule(blockCapsule, solidified);
     if (!filterCapsuleQueue.offer(blockFilterCapsule)) {
       logger.info("Too many filters, block filter lost: {}.", blockCapsule.getBlockId());
     }

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

@@ -61,6 +61,8 @@ public class JsonRpcApiUtil {
   public static final String TAG_SAFE_SUPPORT_ERROR = "TAG safe not supported";
   public static final String BLOCK_NUM_ERROR = "invalid block number";
 
+  private static final SecureRandom random = new SecureRandom();
+
   public static byte[] convertToTronAddress(byte[] address) {
     byte[] newAddress = new byte[21];
     byte[] temp = new byte[] {Wallet.getAddressPreFixByte()};
@@ -647,7 +649,6 @@ public static long parseBlockNumber(String blockNumOrTag, Wallet wallet)
   }
 
   public static String generateFilterId() {
-    SecureRandom random = new SecureRandom();
     byte[] uid = new byte[16]; // 128 bits are converted to 16 bytes
     random.nextBytes(uid);
     return ByteArray.toHexString(uid);