feat(net): optimize transaction rate limiting with accurate cache size check

1. Add Manager.getCachedTransactionSize() = pushTransactionQueue +
   pendingTransactions + rePushTransactions to expose the true cached
   transaction count across all three queues.

2. Fix isTooManyPending() to include pushTransactionQueue, which was
   previously omitted, causing the pending threshold to be underestimated.

3. Update TransactionsMsgHandler.isBusy() to factor in the Manager
   cache size via TronNetDelegate.getCachedTransactionSize(), so the
   node stops accepting TRX INV messages when the full pipeline is busy.

4. Make the busy threshold configurable via node.maxTrxCacheSize
   (default: 50000), replacing the hardcoded MAX_TRX_SIZE constant.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: 2 people

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

@@ -468,6 +468,9 @@ public class CommonParameter {
   public long pendingTransactionTimeout;
   @Getter
   @Setter
+  public int maxTrxCacheSize;
+  @Getter
+  @Setter
   public boolean nodeMetricsEnable = false;
   @Getter
   @Setter

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

@@ -80,6 +80,7 @@ public class NodeConfig {
   private ChannelConfig channel = new ChannelConfig();
   private int maxTransactionPendingSize = 2000;
   private long pendingTransactionTimeout = 60000;
+  private int maxTrxCacheSize = 50_000;
   private int agreeNodeCount = 0;
   private boolean openHistoryQueryWhenLiteFN = false;
   private boolean unsolidifiedBlockCheck = false;
@@ -472,6 +473,11 @@ private void postProcess() {
     if (dynamicConfig.checkInterval <= 0) {
       dynamicConfig.checkInterval = 600;
     }
+
+    // maxTrxCacheSize: minimum 2000
+    if (maxTrxCacheSize < 2000) {
+      maxTrxCacheSize = 2000;
+    }
   }
 
   // ===========================================================================

common/src/main/resources/reference.conf

@@ -351,6 +351,8 @@ node {
   receiveTcpMinDataLength = 2048
   maxTransactionPendingSize = 2000
   pendingTransactionTimeout = 60000
+  # total cached trx across handler queues + pending + rePush
+  maxTrxCacheSize = 50000
 
   # Consensus agreement
   agreeNodeCount = 0

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

@@ -643,6 +643,7 @@ private static void applyNodeConfig(NodeConfig nc) {
 
     PARAMETER.maxTransactionPendingSize = nc.getMaxTransactionPendingSize();
     PARAMETER.pendingTransactionTimeout = nc.getPendingTransactionTimeout();
+    PARAMETER.maxTrxCacheSize = nc.getMaxTrxCacheSize();
 
     PARAMETER.validContractProtoThreadNum = nc.getValidContractProtoThreads();
 

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

@@ -2065,9 +2065,13 @@ public NullifierStore getNullifierStore() {
     return chainBaseManager.getNullifierStore();
   }
 
+  public int getCachedTransactionSize() {
+    return pushTransactionQueue.size() + getPendingTransactions().size()
+        + getRePushTransactions().size();
+  }
+
   public boolean isTooManyPending() {
-    return getPendingTransactions().size() + getRePushTransactions().size()
-        > maxTransactionPendingSize;
+    return getCachedTransactionSize() > maxTransactionPendingSize;
   }
 
   private void preValidateTransactionSign(List<TransactionCapsule> txs)

framework/src/main/java/org/tron/core/net/TronNetDelegate.java

@@ -384,4 +384,8 @@ public boolean isBlockUnsolidified() {
     return headNum - solidNum >= maxUnsolidifiedBlocks;
   }
 
+  public int getCachedTransactionSize() {
+    return dbManager.getCachedTransactionSize();
+  }
+
 }

framework/src/main/java/org/tron/core/net/messagehandler/TransactionsMsgHandler.java

@@ -31,7 +31,6 @@
 @Component
 public class TransactionsMsgHandler implements TronMsgHandler {
 
-  private static int MAX_TRX_SIZE = 50_000;
   private static int MAX_SMART_CONTRACT_SUBMIT_SIZE = 100;
   @Autowired
   private TronNetDelegate tronNetDelegate;
@@ -40,7 +39,8 @@ public class TransactionsMsgHandler implements TronMsgHandler {
   @Autowired
   private ChainBaseManager chainBaseManager;
 
-  private BlockingQueue<TrxEvent> smartContractQueue = new LinkedBlockingQueue(MAX_TRX_SIZE);
+  private BlockingQueue<TrxEvent> smartContractQueue = new LinkedBlockingQueue(
+      Args.getInstance().getMaxTrxCacheSize());
 
   private BlockingQueue<Runnable> queue = new LinkedBlockingQueue();
 
@@ -63,7 +63,8 @@ public void close() {
   }
 
   public boolean isBusy() {
-    return queue.size() + smartContractQueue.size() > MAX_TRX_SIZE;
+    return queue.size() + smartContractQueue.size()
+        + tronNetDelegate.getCachedTransactionSize() > Args.getInstance().getMaxTrxCacheSize();
   }
 
   @Override

framework/src/main/resources/config.conf

@@ -163,6 +163,10 @@ node {
   fetchBlock.timeout = 200
   # syncFetchBatchNum = 2000
 
+  # Maximum total number of cached transactions (handler queues + pending + rePush).
+  # When exceeded, the node stops accepting TRX INV messages from peers.
+  # maxTrxCacheSize = 50000
+
   # Number of validate sign thread, default availableProcessors
   # validateSignThreadNum = 16
 

framework/src/test/java/org/tron/core/db/ManagerTest.java

@@ -16,12 +16,15 @@
 import com.google.common.collect.Sets;
 import com.google.protobuf.Any;
 import com.google.protobuf.ByteString;
+import java.lang.reflect.Field;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -1292,6 +1295,56 @@ public void blockTrigger() {
     Assert.assertEquals(TronError.ErrCode.EVENT_SUBSCRIBE_ERROR, thrown.getErrCode());
   }
 
+  @Test
+  public void testGetCachedTransactionSize() throws Exception {
+    BlockingQueue<TransactionCapsule> pushQ = new LinkedBlockingQueue<>();
+    pushQ.add(new TransactionCapsule(Protocol.Transaction.getDefaultInstance()));
+    Field pushField = Manager.class.getDeclaredField("pushTransactionQueue");
+    pushField.setAccessible(true);
+    pushField.set(dbManager, pushQ);
+
+    dbManager.getPendingTransactions().clear();
+    dbManager.getPendingTransactions().add(
+        new TransactionCapsule(Protocol.Transaction.getDefaultInstance()));
+    dbManager.getPendingTransactions().add(
+        new TransactionCapsule(Protocol.Transaction.getDefaultInstance()));
+
+    dbManager.getRePushTransactions().clear();
+
+    // 1 (push) + 2 (pending) + 0 (rePush) = 3
+    Assert.assertEquals(3, dbManager.getCachedTransactionSize());
+
+    // cleanup
+    pushQ.clear();
+    dbManager.getPendingTransactions().clear();
+  }
+
+  @Test
+  public void testIsTooManyPendingIncludesPushQueue() throws Exception {
+    int threshold = Args.getInstance().getMaxTransactionPendingSize();
+
+    BlockingQueue<TransactionCapsule> pushQ = new LinkedBlockingQueue<>();
+    Field pushField = Manager.class.getDeclaredField("pushTransactionQueue");
+    pushField.setAccessible(true);
+    pushField.set(dbManager, pushQ);
+
+    dbManager.getPendingTransactions().clear();
+    dbManager.getRePushTransactions().clear();
+
+    for (int i = 0; i < threshold; i++) {
+      dbManager.getPendingTransactions().add(
+          new TransactionCapsule(Protocol.Transaction.getDefaultInstance()));
+    }
+    Assert.assertFalse(dbManager.isTooManyPending());
+
+    pushQ.add(new TransactionCapsule(Protocol.Transaction.getDefaultInstance()));
+    Assert.assertTrue(dbManager.isTooManyPending());
+
+    // cleanup
+    dbManager.getPendingTransactions().clear();
+    pushQ.clear();
+  }
+
   public void adjustBalance(AccountStore accountStore, byte[] accountAddress, long amount)
       throws BalanceInsufficientException {
     Commons.adjustBalance(accountStore, accountAddress, amount,

framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java

@@ -42,8 +42,6 @@ public static void init() {
   public void testProcessMessage() {
     TransactionsMsgHandler transactionsMsgHandler = new TransactionsMsgHandler();
     try {
-      Assert.assertFalse(transactionsMsgHandler.isBusy());
-
       transactionsMsgHandler.init();
 
       PeerConnection peer = Mockito.mock(PeerConnection.class);
@@ -54,6 +52,8 @@ public void testProcessMessage() {
       field.setAccessible(true);
       field.set(transactionsMsgHandler, tronNetDelegate);
 
+      Assert.assertFalse(transactionsMsgHandler.isBusy());
+
       BalanceContract.TransferContract transferContract = BalanceContract.TransferContract
           .newBuilder()
           .setAmount(10)
@@ -132,6 +132,23 @@ public void testProcessMessage() {
     }
   }
 
+  @Test
+  public void testIsBusyWithCachedTransactions() throws Exception {
+    TransactionsMsgHandler handler = new TransactionsMsgHandler();
+
+    TronNetDelegate tronNetDelegateMock = Mockito.mock(TronNetDelegate.class);
+    Mockito.when(tronNetDelegateMock.getCachedTransactionSize()).thenReturn(50_001);
+    Field field = TransactionsMsgHandler.class.getDeclaredField("tronNetDelegate");
+    field.setAccessible(true);
+    field.set(handler, tronNetDelegateMock);
+
+    // queue and smartContractQueue are empty, but cached size > threshold
+    Assert.assertTrue(handler.isBusy());
+
+    Mockito.when(tronNetDelegateMock.getCachedTransactionSize()).thenReturn(0);
+    Assert.assertFalse(handler.isBusy());
+  }
+
   class TrxEvent {
 
     @Getter