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..872ae3f6799 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 @@ -18,6 +18,8 @@ import org.tron.common.es.ExecutorServiceManager; import org.tron.common.utils.Sha256Hash; import org.tron.core.ChainBaseManager; +import org.tron.core.Constant; +import org.tron.core.capsule.TransactionCapsule; import org.tron.core.config.args.Args; import org.tron.core.exception.P2pException; import org.tron.core.exception.P2pException.TypeEnum; @@ -88,23 +90,33 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep return; } TransactionsMessage transactionsMessage = (TransactionsMessage) msg; - check(peer, transactionsMessage); - for (Transaction trx : transactionsMessage.getTransactions().getTransactionsList()) { - Item item = new Item(new TransactionMessage(trx).getMessageId(), InventoryType.TRX); + Set ids = check(peer, transactionsMessage); + for (Sha256Hash id : ids) { + Item item = new Item(id, InventoryType.TRX); peer.getAdvInvRequest().remove(item); } int smartContractQueueSize = 0; int trxHandlePoolQueueSize = 0; int dropSmartContractCount = 0; + int oversizedCount = 0; for (Transaction trx : transactionsMessage.getTransactions().getTransactionsList()) { if (isClosed) { logger.info("TransactionsMsgHandler is closed during processing, stop submit"); break; } - int type = trx.getRawData().getContract(0).getType().getNumber(); + TransactionCapsule capsule = new TransactionCapsule(trx); + capsule.sanitize(); + capsule.resetResult(); + if (capsule.getSerializedSize() > Constant.TRANSACTION_MAX_BYTE_SIZE) { + oversizedCount++; + continue; + } + // Both queues must retain the cleaned transaction, including its serialized bytes. + Transaction cleanTrx = capsule.getInstance(); + int type = cleanTrx.getRawData().getContract(0).getType().getNumber(); if (type == ContractType.TriggerSmartContract_VALUE || type == ContractType.CreateSmartContract_VALUE) { - if (!smartContractQueue.offer(new TrxEvent(peer, new TransactionMessage(trx)))) { + if (!smartContractQueue.offer(new TrxEvent(peer, new TransactionMessage(cleanTrx)))) { smartContractQueueSize = smartContractQueue.size(); trxHandlePoolQueueSize = queue.size(); dropSmartContractCount++; @@ -112,7 +124,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep } else { try { ExecutorServiceManager.submit( - trxHandlePool, () -> handleTransaction(peer, new TransactionMessage(trx))); + trxHandlePool, () -> handleTransaction(peer, new TransactionMessage(cleanTrx))); } catch (RejectedExecutionException e) { logger.warn("Submit task to {} failed", trxEsName); break; @@ -124,13 +136,17 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep logger.warn("Add smart contract failed, drop count: {}, queueSize {}:{}", dropSmartContractCount, smartContractQueueSize, trxHandlePoolQueueSize); } + if (oversizedCount > 0) { + logger.warn("Drop oversized transactions, count: {}", oversizedCount); + } } - private void check(PeerConnection peer, TransactionsMessage msg) throws P2pException { + private Set check(PeerConnection peer, TransactionsMessage msg) + throws P2pException { List list = msg.getTransactions().getTransactionsList(); Set seen = new HashSet<>(list.size() * 2); for (Transaction trx : list) { - Sha256Hash id = new TransactionMessage(trx).getMessageId(); + Sha256Hash id = new TransactionCapsule(trx).getTransactionId(); if (!seen.add(id)) { throw new P2pException(TypeEnum.BAD_MESSAGE, "TransactionsMessage contains duplicate transaction: " + id); @@ -151,6 +167,7 @@ private void check(PeerConnection peer, TransactionsMessage msg) throws P2pExcep } } } + return seen; } private void handleSmartContract() { @@ -216,4 +233,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/net/messagehandler/TransactionsMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java index 282c80f9f6d..851e084a747 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 @@ -2,10 +2,12 @@ import com.google.protobuf.Any; import com.google.protobuf.ByteString; +import com.google.protobuf.UnknownFieldSet; import java.lang.reflect.Field; import java.lang.reflect.Method; import java.time.ZonedDateTime; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.BlockingQueue; @@ -25,7 +27,9 @@ import org.tron.common.TestConstants; import org.tron.common.runtime.TvmTestUtils; import org.tron.common.utils.ByteArray; +import org.tron.common.utils.ReflectUtils; import org.tron.core.ChainBaseManager; +import org.tron.core.Constant; import org.tron.core.config.args.Args; import org.tron.core.exception.P2pException; import org.tron.core.exception.P2pException.TypeEnum; @@ -199,6 +203,7 @@ public void testRejectedExecution() throws Exception { // 2 transfer transactions, submit throws on the first → catch + break, only called once handler.processMessage(peer, msg); + Assert.assertTrue(peer.getAdvInvRequest().isEmpty()); Mockito.verify(mockPool, Mockito.times(1)).submit(Mockito.any(Runnable.class)); } finally { closeHandlerAndOriginalPool(handler, originalPool); @@ -389,11 +394,42 @@ public void testDuplicateTransactionRejected() throws Exception { } catch (P2pException e) { Assert.assertEquals(P2pException.TypeEnum.BAD_MESSAGE, e.getType()); } + Assert.assertEquals(1, advInvRequest.size()); + Assert.assertTrue(advInvRequest.containsKey(item)); } finally { handler.close(); } } + @Test + public void testInvalidBatchPreservesRequests() throws Exception { + TransactionsMsgHandler handler = new TransactionsMsgHandler(); + ExecutorService originalPool = null; + try { + ExecutorService pool = Mockito.mock(ExecutorService.class); + originalPool = replaceTrxHandlePool(handler, pool); + List transactions = new ArrayList<>( + buildTransferMessage(2).getTransactions().getTransactionsList()); + // The first transaction passes; the second fails signature-length validation. + transactions.set(1, transactions.get(1).toBuilder() + .addSignature(ByteString.copyFrom(new byte[64])).build()); + TransactionsMessage msg = new TransactionsMessage(transactions); + PeerConnection peer = Mockito.mock(PeerConnection.class); + stubAdvInvRequest(peer, msg); + Map requests = new ConcurrentHashMap<>(peer.getAdvInvRequest()); + Assert.assertEquals(2, requests.size()); + + P2pException exception = Assert.assertThrows(P2pException.class, + () -> handler.processMessage(peer, msg)); + + Assert.assertEquals(TypeEnum.BAD_TRX, exception.getType()); + Assert.assertEquals(requests, peer.getAdvInvRequest()); + Mockito.verify(pool, Mockito.never()).submit(Mockito.any(Runnable.class)); + } finally { + closeHandlerAndOriginalPool(handler, originalPool); + } + } + @Test public void testInvalidSigLength() throws Exception { TransactionsMsgHandler handler = new TransactionsMsgHandler(); @@ -492,6 +528,149 @@ public void testInvalidSigLength() throws Exception { } } + @Test + public void testSanitizeBeforeQueueAdmission() throws Exception { + for (boolean smart : new boolean[] {false, true}) { + Protocol.Transaction base = admissionTransaction(smart); + Protocol.Transaction clean = base.toBuilder().setRawData(base.getRawData().toBuilder() + .setUnknownFields(padding(32))).build(); + Protocol.Transaction padded = clean.toBuilder().setUnknownFields(padding(600 * 1024)) + .build(); + assertAdmission(padded, clean); + } + } + + @Test + public void testOversizedTransactionDroppedBeforeQueueAdmission() throws Exception { + for (boolean smart : new boolean[] {false, true}) { + Protocol.Transaction base = admissionTransaction(smart); + Protocol.Transaction oversized = base.toBuilder().setRawData(base.getRawData().toBuilder() + .setData(ByteString.copyFrom(new byte[600 * 1024]))).build(); + assertAdmission(oversized, null); + } + } + + @Test + public void testQueueAdmissionSizeBoundary() throws Exception { + int limit = (int) Constant.TRANSACTION_MAX_BYTE_SIZE; + for (boolean smart : new boolean[] {false, true}) { + Protocol.Transaction.raw.Builder raw = admissionTransaction(smart).getRawData().toBuilder() + .setData(ByteString.copyFrom(new byte[limit])); + int overhead = Protocol.Transaction.newBuilder().setRawData(raw).build() + .getSerializedSize() - limit; + Protocol.Transaction exact = Protocol.Transaction.newBuilder().setRawData( + raw.setData(ByteString.copyFrom(new byte[limit - overhead]))).build(); + Assert.assertEquals(limit, exact.getSerializedSize()); + assertAdmission(exact, exact); + Protocol.Transaction over = exact.toBuilder().setRawData( + raw.setData(ByteString.copyFrom(new byte[limit - overhead + 1]))).build(); + Assert.assertEquals(limit + 1, over.getSerializedSize()); + assertAdmission(over, null); + } + } + + @Test + public void testResultsClearedBeforeQueueAdmission() throws Exception { + for (boolean smart : new boolean[] {false, true}) { + Protocol.Transaction expected = admissionTransaction(smart).toBuilder() + .addSignature(ByteString.copyFrom(new byte[65])).build(); + // Even a single result must be discarded, including its known and unknown fields. + Protocol.Transaction.Result result = Protocol.Transaction.Result.newBuilder() + .setFee(123456).setWithdrawAmount(77) + .setContractRet(Protocol.Transaction.Result.contractResult.REVERT) + .setUnknownFields(padding(32)).build(); + assertAdmission(expected.toBuilder().addRet(result).build(), expected); + assertAdmission(expected.toBuilder().addRet(result).addRet(result).build(), expected); + } + } + + @Test + public void testOversizedResultsClearedBeforeQueueAdmission() throws Exception { + for (boolean smart : new boolean[] {false, true}) { + Protocol.Transaction expected = admissionTransaction(smart).toBuilder() + .addSignature(ByteString.copyFrom(new byte[65])).build(); + // Padding in the first result must be removed before the transaction size check. + Protocol.Transaction padded = expected.toBuilder().addRet( + Protocol.Transaction.Result.newBuilder().setUnknownFields(padding(600 * 1024))).build(); + Assert.assertTrue(padded.getSerializedSize() > Constant.TRANSACTION_MAX_BYTE_SIZE); + assertAdmission(padded, expected); + } + } + + private void assertAdmission(Protocol.Transaction input, Protocol.Transaction expected) + throws Exception { + TransactionsMsgHandler handler = new TransactionsMsgHandler(); + ExecutorService originalPool = null; + try { + ExecutorService pool = Mockito.mock(ExecutorService.class); + List tasks = new ArrayList<>(); + Mockito.when(pool.submit(Mockito.any(Runnable.class))).thenAnswer(call -> { + tasks.add(call.getArgument(0)); + return Mockito.mock(Future.class); + }); + originalPool = replaceTrxHandlePool(handler, pool); + TronNetDelegate delegate = Mockito.mock(TronNetDelegate.class); + AdvService adv = Mockito.mock(AdvService.class); + ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(handler, "advService", adv); + ReflectUtils.setFieldValue(handler, "chainBaseManager", Mockito.mock(ChainBaseManager.class)); + PeerConnection peer = Mockito.mock(PeerConnection.class); + TransactionsMessage message = new TransactionsMessage(Collections.singletonList(input)); + stubAdvInvRequest(peer, message); + handler.processMessage(peer, message); + BlockingQueue smartQueue = (BlockingQueue) + ReflectUtils.getFieldObject(handler, "smartContractQueue"); + if (expected == null) { + Assert.assertTrue(smartQueue.isEmpty()); + Mockito.verify(pool, Mockito.never()).submit(Mockito.any(Runnable.class)); + } else { + TransactionMessage admitted; + if (input.getRawData().getContract(0).getType() + == Protocol.Transaction.Contract.ContractType.TriggerSmartContract) { + Assert.assertEquals(1, smartQueue.size()); + admitted = (TransactionMessage) ReflectUtils.getFieldObject(smartQueue.peek(), "msg"); + } else { + Assert.assertEquals(1, tasks.size()); + // Run only after admission completes, exercising the transaction retained by the task. + tasks.get(0).run(); + org.mockito.ArgumentCaptor captured = + org.mockito.ArgumentCaptor.forClass(TransactionMessage.class); + Mockito.verify(adv).broadcast(captured.capture()); + admitted = captured.getValue(); + } + Protocol.Transaction retained = admitted.getTransactionCapsule().getInstance(); + Protocol.Transaction serialized = Protocol.Transaction.parseFrom(admitted.getData()); + Assert.assertEquals(expected, retained); + Assert.assertEquals(0, retained.getRetCount()); + Assert.assertEquals(0, serialized.getRetCount()); + Assert.assertArrayEquals(input.getRawData().toByteArray(), + retained.getRawData().toByteArray()); + Assert.assertEquals(input.getSignatureList(), retained.getSignatureList()); + Assert.assertArrayEquals(expected.toByteArray(), admitted.getData()); + Assert.assertEquals(new TransactionMessage(input).getMessageId(), admitted.getMessageId()); + } + Assert.assertTrue(peer.getAdvInvRequest().isEmpty()); + Mockito.verify(peer, Mockito.never()).disconnect(Mockito.any()); + } finally { + closeHandlerAndOriginalPool(handler, originalPool); + } + } + + private Protocol.Transaction admissionTransaction(boolean smart) { + if (smart) { + return TvmTestUtils.generateTriggerSmartContractAndGetTransaction( + ByteArray.fromHexString("121212a9cf"), ByteArray.fromHexString("121212a9cf"), + ByteArray.fromHexString("123456"), 100, 100000000, 0, 0); + } + return buildTransferMessage(1).getTransactions().getTransactions(0); + } + + private UnknownFieldSet padding(int size) { + return UnknownFieldSet.newBuilder().addField(99999, + UnknownFieldSet.Field.newBuilder().addLengthDelimited(ByteString.copyFrom(new byte[size])) + .build()).build(); + } + @Test public void testIsBusyWithCachedTransactions() throws Exception { TransactionsMsgHandler handler = new TransactionsMsgHandler();