Skip to content

fix(net): reduce memory usage in transaction queues #7011

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
317787106 wants to merge 2 commits into
base: release_v4.8.3
Choose a base branch
from
Open

fix(net): reduce memory usage in transaction queues #7011

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f49e9ef
drop oversized transaction and delete unknown fields of transaction
317787106 Oct 7, 2026
bdafbed
reuse txid results of check in TransactionsMsgHandler
317787106 Oct 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 26 additions & 9 deletions framework/src/main/java/org/tron/core/net/messagehandler/TransactionsMsgHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -88,31 +90,41 @@ 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<Sha256Hash> 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.removeRedundantRet();
if (capsule.getSerializedSize() > Constant.TRANSACTION_MAX_BYTE_SIZE) {
oversizedCount++;
continue;
Comment thread
3for marked this conversation as resolved.
}
// 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)))) {
Comment thread
3for marked this conversation as resolved.
smartContractQueueSize = smartContractQueue.size();
trxHandlePoolQueueSize = queue.size();
dropSmartContractCount++;
}
} 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;
Expand All @@ -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<Sha256Hash> check(PeerConnection peer, TransactionsMessage msg)
throws P2pException {
List<Transaction> list = msg.getTransactions().getTransactionsList();
Set<Sha256Hash> 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);
Expand All @@ -151,6 +167,7 @@ private void check(PeerConnection peer, TransactionsMessage msg) throws P2pExcep
}
}
}
return seen;
}

private void handleSmartContract() {
Expand Down Expand Up @@ -216,4 +233,4 @@ public TrxEvent(PeerConnection peer, TransactionMessage msg) {
this.time = System.currentTimeMillis();
}
}
}
}
152 changes: 152 additions & 0 deletions framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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<Protocol.Transaction> 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<Item, Long> 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();
Expand Down Expand Up @@ -492,6 +528,122 @@ 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 testRedundantResultsTrimmedBeforeQueueAdmission() throws Exception {
for (boolean smart : new boolean[] {false, true}) {
Protocol.Transaction expected = admissionTransaction(smart).toBuilder()
.addRet(Protocol.Transaction.Result.newBuilder()
.setContractRet(Protocol.Transaction.Result.contractResult.SUCCESS)).build();
Protocol.Transaction padded = expected.toBuilder().addRet(
Protocol.Transaction.Result.newBuilder().setUnknownFields(padding(600 * 1024))).build();
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);
Mockito.when(pool.submit(Mockito.any(Runnable.class))).thenAnswer(call -> {
((Runnable) call.getArgument(0)).run();
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 {
org.mockito.ArgumentCaptor<TransactionMessage> captured =
org.mockito.ArgumentCaptor.forClass(TransactionMessage.class);
Mockito.verify(adv).broadcast(captured.capture());
admitted = captured.getValue();
}
Assert.assertEquals(expected, admitted.getTransactionCapsule().getInstance());
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();
Expand Down
Loading