diff --git a/framework/src/main/java/org/tron/core/Wallet.java b/framework/src/main/java/org/tron/core/Wallet.java index ac54cb2b7ff..ab42639bc4d 100755 --- a/framework/src/main/java/org/tron/core/Wallet.java +++ b/framework/src/main/java/org/tron/core/Wallet.java @@ -572,7 +572,16 @@ public GrpcAPI.Return broadcastTransaction(Transaction signedTransaction) { throw new ContractValidateException(ActuatorConstant.CONTRACT_NOT_EXIST); } trx.checkExpiration(chainBaseManager.getNextBlockSlotTime()); - dbManager.pushTransaction(trx); + if (!dbManager.pushTransaction(trx)) { + if (trxCacheEnable) { + dbManager.getTransactionIdCache().invalidate(txID); + } + logger.info("Broadcast transaction {} has failed, Shielded pending pool is full.", txID); + return builder.setResult(false).setCode(response_code.SERVER_BUSY) + .setMessage(ByteString.copyFromUtf8( + "Shielded transaction pending pool is full.")) + .build(); + } TransactionMessage message = new TransactionMessage(trx.getInstance().toByteArray()); int num = tronNetService.fastBroadcastTransaction(message); if (num == 0 && minEffectiveConnection != 0) { diff --git a/framework/src/main/java/org/tron/core/net/TronNetDelegate.java b/framework/src/main/java/org/tron/core/net/TronNetDelegate.java index 23050f5218d..9d15886f3ab 100644 --- a/framework/src/main/java/org/tron/core/net/TronNetDelegate.java +++ b/framework/src/main/java/org/tron/core/net/TronNetDelegate.java @@ -322,10 +322,10 @@ public void processBlock(BlockCapsule block, boolean isSync) throws P2pException } } - public void pushTransaction(TransactionCapsule trx) throws P2pException { + public boolean pushTransaction(TransactionCapsule trx) throws P2pException { try { trx.setTime(System.currentTimeMillis()); - dbManager.pushTransaction(trx); + return dbManager.pushTransaction(trx); } catch (ContractSizeNotEqualToOneException | VMIllegalException e) { throw new P2pException(TypeEnum.BAD_TRX, e); diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/TransactionsMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/TransactionsMsgHandler.java index 52137c5881c..d5be67c50bd 100644 --- a/framework/src/main/java/org/tron/core/net/messagehandler/TransactionsMsgHandler.java +++ b/framework/src/main/java/org/tron/core/net/messagehandler/TransactionsMsgHandler.java @@ -183,7 +183,11 @@ private void handleTransaction(PeerConnection peer, TransactionMessage trx) { try { trx.getTransactionCapsule().checkExpiration(chainBaseManager.getNextBlockSlotTime()); - tronNetDelegate.pushTransaction(trx.getTransactionCapsule()); + if (!tronNetDelegate.pushTransaction(trx.getTransactionCapsule())) { + logger.debug("Drop trx {} from {}, Shielded pending pool is full", + trx.getMessageId(), peer.getInetAddress()); + return; + } advService.broadcast(trx); } catch (P2pException e) { logger.warn("Trx {} from peer {} process failed. type: {}, reason: {}", @@ -216,4 +220,4 @@ public TrxEvent(PeerConnection peer, TransactionMessage msg) { this.time = System.currentTimeMillis(); } } -} \ No newline at end of file +} diff --git a/framework/src/test/java/org/tron/core/WalletMockTest.java b/framework/src/test/java/org/tron/core/WalletMockTest.java index 2f4c08d8f9f..57f52f28c8a 100644 --- a/framework/src/test/java/org/tron/core/WalletMockTest.java +++ b/framework/src/test/java/org/tron/core/WalletMockTest.java @@ -65,6 +65,7 @@ import org.tron.core.exception.ValidateSignatureException; import org.tron.core.exception.ZksnarkException; import org.tron.core.net.TronNetDelegate; +import org.tron.core.net.TronNetService; import org.tron.core.net.message.adv.TransactionMessage; import org.tron.core.net.peer.PeerConnection; import org.tron.core.store.AbiStore; @@ -309,6 +310,55 @@ public void testBroadcastTransactionTooManyPending() throws Exception { assertEquals(GrpcAPI.Return.response_code.SERVER_BUSY, ret.getCode()); } + @Test + public void testBroadcastTransactionShieldedPendingPoolFull() throws Exception { + long now = System.currentTimeMillis(); + BalanceContract.TransferContract transferContract = + BalanceContract.TransferContract.newBuilder() + .setAmount(10) + .setOwnerAddress(ByteString.copyFromUtf8("aaa")) + .setToAddress(ByteString.copyFromUtf8("bbb")) + .build(); + Protocol.Transaction transaction = Protocol.Transaction.newBuilder() + .setRawData(Protocol.Transaction.raw.newBuilder() + .setExpiration(now + 60_000) + .addContract(Protocol.Transaction.Contract.newBuilder() + .setParameter(Any.pack(transferContract)) + .setType(Protocol.Transaction.Contract.ContractType.TransferContract))) + .build(); + Sha256Hash txId = new TransactionCapsule(transaction).getTransactionId(); + + Wallet wallet = new Wallet(); + TronNetDelegate tronNetDelegate = mock(TronNetDelegate.class); + TronNetService tronNetService = mock(TronNetService.class); + Manager manager = mock(Manager.class); + ChainBaseManager chainBaseManager = mock(ChainBaseManager.class); + DynamicPropertiesStore dynamicPropertiesStore = mock(DynamicPropertiesStore.class); + Cache transactionIdCache = CacheBuilder.newBuilder().build(); + + when(tronNetDelegate.isBlockUnsolidified()).thenReturn(false); + when(manager.isTooManyPending()).thenReturn(false); + when(manager.getTransactionIdCache()).thenReturn(transactionIdCache); + when(manager.pushTransaction(any())).thenReturn(false); + when(chainBaseManager.getDynamicPropertiesStore()).thenReturn(dynamicPropertiesStore); + when(chainBaseManager.getNextBlockSlotTime()).thenReturn(now); + when(dynamicPropertiesStore.supportVM()).thenReturn(false); + + setField(wallet, "tronNetDelegate", tronNetDelegate); + setField(wallet, "tronNetService", tronNetService); + setField(wallet, "dbManager", manager); + setField(wallet, "chainBaseManager", chainBaseManager); + setField(wallet, "trxCacheEnable", true); + + GrpcAPI.Return result = wallet.broadcastTransaction(transaction); + + assertEquals(GrpcAPI.Return.response_code.SERVER_BUSY, result.getCode()); + assertEquals("Shielded transaction pending pool is full.", + result.getMessage().toStringUtf8()); + assertNull(transactionIdCache.getIfPresent(txId)); + Mockito.verify(tronNetService, Mockito.never()).fastBroadcastTransaction(any()); + } + @Test public void testBroadcastTransactionAlreadyExists() throws Exception { Wallet wallet = new Wallet(); @@ -397,6 +447,7 @@ public void testBroadcastTransactionOtherException() throws Exception { = mock(DynamicPropertiesStore.class); when(tronNetDelegateMock.isBlockUnsolidified()).thenReturn(false); when(managerMock.isTooManyPending()).thenReturn(false); + when(managerMock.pushTransaction(any())).thenReturn(true); when(chainBaseManagerMock.getDynamicPropertiesStore()) .thenReturn(dynamicPropertiesStoreMock); when(dynamicPropertiesStoreMock.supportVM()).thenReturn(false); @@ -443,6 +494,12 @@ private Protocol.Transaction getExampleTrans() { .build(); } + private void setField(Object target, String fieldName, Object value) throws Exception { + Field field = target.getClass().getDeclaredField(fieldName); + field.setAccessible(true); + field.set(target, value); + } + private void mockEnv(Wallet wallet, TronException tronException) throws Exception { TronNetDelegate tronNetDelegateMock = mock(TronNetDelegate.class); Manager managerMock = mock(Manager.class); @@ -459,21 +516,10 @@ private void mockEnv(Wallet wallet, TronException tronException) throws Exceptio doThrow(tronException).when(managerMock).pushTransaction(any()); - Field field = wallet.getClass().getDeclaredField("tronNetDelegate"); - field.setAccessible(true); - field.set(wallet, tronNetDelegateMock); - - Field field2 = wallet.getClass().getDeclaredField("dbManager"); - field2.setAccessible(true); - field2.set(wallet, managerMock); - - Field field4 = wallet.getClass().getDeclaredField("chainBaseManager"); - field4.setAccessible(true); - field4.set(wallet, chainBaseManagerMock); - - Field field3 = wallet.getClass().getDeclaredField("trxCacheEnable"); - field3.setAccessible(true); - field3.set(wallet, false); + setField(wallet, "tronNetDelegate", tronNetDelegateMock); + setField(wallet, "dbManager", managerMock); + setField(wallet, "chainBaseManager", chainBaseManagerMock); + setField(wallet, "trxCacheEnable", false); } @Test diff --git a/framework/src/test/java/org/tron/core/net/TronNetDelegateTest.java b/framework/src/test/java/org/tron/core/net/TronNetDelegateTest.java index 4c16f28930c..38738f78a6d 100644 --- a/framework/src/test/java/org/tron/core/net/TronNetDelegateTest.java +++ b/framework/src/test/java/org/tron/core/net/TronNetDelegateTest.java @@ -134,6 +134,20 @@ public void testPushVerifiedBlockPushesBlock() throws Exception { Mockito.verify(dbManager, Mockito.times(1)).pushBlock(Mockito.any()); } + @Test + public void testPushTransactionReturnsAdmissionResult() throws Exception { + TronNetDelegate tronNetDelegate = new TronNetDelegate(); + Manager dbManager = Mockito.mock(Manager.class); + TransactionCapsule transaction = new TransactionCapsule( + TransferContract.getDefaultInstance(), ContractType.TransferContract); + setField(tronNetDelegate, "dbManager", dbManager); + + Mockito.when(dbManager.pushTransaction(transaction)).thenReturn(false, true); + + Assert.assertFalse(tronNetDelegate.pushTransaction(transaction)); + Assert.assertTrue(tronNetDelegate.pushTransaction(transaction)); + } + private static void setField(Object obj, String name, Object value) throws Exception { Field f = obj.getClass().getDeclaredField(name); f.setAccessible(true); diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java index ed2121d360f..3681722bf81 100644 --- a/framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java +++ b/framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java @@ -277,9 +277,17 @@ public void testHandleTransaction() throws Exception { // happy path → push and broadcast Mockito.when(chainBaseManager.getNextBlockSlotTime()).thenReturn(now); + Mockito.when(tronNetDelegate.pushTransaction(Mockito.any())).thenReturn(true); handleTx.invoke(handler, peer, trxMsg); Mockito.verify(advService).broadcast(trxMsg); + // local capacity rejection → do not broadcast or penalize the peer + Mockito.when(tronNetDelegate.pushTransaction(Mockito.any())).thenReturn(false); + handleTx.invoke(handler, peer, trxMsg); + Mockito.verify(advService, Mockito.times(1)).broadcast(trxMsg); + Mockito.verify(peer, Mockito.never()).setBadPeer(true); + Mockito.verify(peer, Mockito.never()).disconnect(Mockito.any()); + // P2pException BAD_TRX → disconnect Mockito.doThrow(new P2pException(TypeEnum.BAD_TRX, "bad")) .when(tronNetDelegate).pushTransaction(Mockito.any());