Skip to content

fix(net): harden sync message resource controls #6966

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
xxo1shine wants to merge 1 commit into
base: release_v4.8.3
Choose a base branch
from
+156 −7
Open

fix(net): harden sync message resource controls #6966

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
14 changes: 11 additions & 3 deletions framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,11 @@ private void check(PeerConnection peer, ChainInventoryMessage msg) throws P2pExc
throw new P2pException(TypeEnum.BAD_MESSAGE, "blockIds is empty");
}

if (msg.getRemainNum() < 0) {
throw new P2pException(TypeEnum.BAD_MESSAGE,
"remainNum is negative: " + msg.getRemainNum());
}

if (blockIds.size() > NetConstants.SYNC_FETCH_BATCH_NUM + 1) {
throw new P2pException(TypeEnum.BAD_MESSAGE, "big blockIds size: " + blockIds.size());
}
Expand Down Expand Up @@ -137,9 +142,12 @@ private void check(PeerConnection peer, ChainInventoryMessage msg) throws P2pExc
long maxFutureNum =
maxRemainTime / BLOCK_PRODUCED_INTERVAL + tronNetDelegate.getSolidBlockId().getNum();
long lastNum = blockIds.get(blockIds.size() - 1).getNum();
if (lastNum + msg.getRemainNum() > maxFutureNum) {
throw new P2pException(TypeEnum.BAD_MESSAGE, "lastNum: " + lastNum + " + remainNum: "
+ msg.getRemainNum() + " > futureMaxNum: " + maxFutureNum);
long declaredHighestNum = lastNum + msg.getRemainNum();

@lxcmyf lxcmyf Sep 17, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[MUST] The overflow check is still inside the headBlockId > 0 branch. At height 0, a message containing genesis plus continuous heights 1..1999 and remainNum = Long.MAX_VALUE skips the check, so the original overflow bypass remains. Always perform a checked addition or overflow check; only the futureMaxNum bound may remain conditional. Add a height=0 test.

@xxo1shine xxo1shine Sep 18, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This change addresses addition overflow that makes the upper-bound comparison ineffective when height validation is enabled. When headBlockId <= 0, this validation was already skipped as part of the existing initial-sync behavior; it is not an overflow bypass introduced by this PR. We will preserve that branch behavior in this change. Whether initial sync should have an independent check on the declared highest block number can be evaluated separately.

if (declaredHighestNum < 0 || declaredHighestNum > maxFutureNum) {
throw new P2pException(TypeEnum.BAD_MESSAGE,
"Invalid declared highest block number: " + declaredHighestNum
+ ", lastNum: " + lastNum + ", remainNum: " + msg.getRemainNum()
+ ", futureMaxNum: " + maxFutureNum);
}
}
}
Expand Down
3 changes: 1 addition & 2 deletions framework/src/main/java/org/tron/core/net/messagehandler/SyncBlockChainMsgHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,8 +58,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
}

private boolean check(PeerConnection peer, SyncBlockChainMessage msg) throws P2pException {
if (peer.getRemainNum() > 0
&& !peer.getP2pRateLimiter().tryAcquire(msg.getType().asByte())) {
if (!peer.getP2pRateLimiter().tryAcquire(msg.getType().asByte())) {
// Discard messages that exceed the rate limit
logger.warn("{} message from peer {} exceeds the rate limit",
msg.getType(), peer.getInetSocketAddress());
Expand Down
3 changes: 2 additions & 1 deletion framework/src/main/java/org/tron/core/net/peer/PeerConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,8 @@ public class PeerConnection {
private volatile long remainNum;
@Getter
private Cache<Sha256Hash, Long> syncBlockIdCache = CacheBuilder.newBuilder()
.maximumSize(2 * NetConstants.SYNC_FETCH_BATCH_NUM).recordStats().build();
.concurrencyLevel(1)
.maximumSize(2 * NetConstants.SYNC_FETCH_BATCH_NUM + 1).recordStats().build();

@lxcmyf lxcmyf Sep 17, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[MUST] Raising maximumSize to 4001 still does not guarantee that all IDs in the active window remain cached: entries survive across moving windows, and Guava segmented eviction can evict entries before the total size reaches the limit. An evicted block still inside [last-4000,last] can be requested again. Maintain an exact height-aware window and test all 4001 IDs plus advancing, out-of-order windows.

@xxo1shine xxo1shine Sep 18, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The early eviction caused by segmentation is a valid issue. I will add concurrencyLevel(1), retain the capacity of 4001, and add tests covering retention of every ID within a fixed window, out-of-order requests, and rejection of duplicate requests. The goal of this change is to prevent repeated fetching within a fixed window caused by premature cache eviction. Strict deduplication across advancing windows is outside the scope of this change. This path also has request rate limits and a per-request block count limit, so we will not introduce exact height-aware window management at this stage.

@Setter
@Getter
private Deque<BlockId> syncBlockToFetch = new ConcurrentLinkedDeque<>();
Expand Down
55 changes: 54 additions & 1 deletion framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,19 +1,24 @@
package org.tron.core.net.messagehandler;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.Mockito;
import org.tron.common.TestConstants;
import org.tron.common.utils.Pair;
import org.tron.common.utils.ReflectUtils;
import org.tron.common.utils.Sha256Hash;
import org.tron.core.capsule.BlockCapsule.BlockId;
import org.tron.core.config.Parameter.NetConstants;
import org.tron.core.config.args.Args;
import org.tron.core.exception.P2pException;
import org.tron.core.net.message.keepalive.PingMessage;
import org.tron.core.net.TronNetDelegate;
import org.tron.core.net.message.sync.ChainInventoryMessage;
import org.tron.core.net.peer.PeerConnection;

Expand Down Expand Up @@ -78,4 +83,52 @@ public void testProcessMessage() throws Exception {
Assert.assertNull(msg.getAnswerMessage());
}

@Test
public void testNegativeRemainNumRejected() throws Exception {
assertCheckRejects(createContinuousBlockIds(0L), -1L);
}

@Test
public void testRemainNumOverflowRejected() throws Exception {
assertCheckRejects(createContinuousBlockIds(
Long.MAX_VALUE - NetConstants.SYNC_FETCH_BATCH_NUM + 1), 1L);
}

private void assertCheckRejects(List<BlockId> ids, long remainNum) throws Exception {
ChainInventoryMsgHandler messageHandler = new ChainInventoryMsgHandler();
TronNetDelegate tronNetDelegate = Mockito.mock(TronNetDelegate.class);
ReflectUtils.setFieldValue(messageHandler, "tronNetDelegate", tronNetDelegate);
Mockito.when(tronNetDelegate.getHeadBlockId()).thenReturn(
new BlockId(Sha256Hash.ZERO_HASH, 1L));
Mockito.when(tronNetDelegate.getSolidBlockId()).thenReturn(
new BlockId(Sha256Hash.ZERO_HASH, 1L));
Mockito.when(tronNetDelegate.getBlockTime(Mockito.any())).thenReturn(0L);

PeerConnection connection = Mockito.mock(PeerConnection.class);
LinkedList<BlockId> requestedIds = new LinkedList<>();
requestedIds.add(ids.get(0));
Mockito.when(connection.getSyncChainRequested()).thenReturn(
new Pair<>(requestedIds, System.currentTimeMillis()));

Method check = ChainInventoryMsgHandler.class.getDeclaredMethod(
"check", PeerConnection.class, ChainInventoryMessage.class);
check.setAccessible(true);
try {
check.invoke(messageHandler, connection, new ChainInventoryMessage(ids, remainNum));
Assert.fail("Expected invalid remainNum to be rejected");
} catch (InvocationTargetException e) {
Assert.assertTrue(e.getCause() instanceof P2pException);
Assert.assertEquals(P2pException.TypeEnum.BAD_MESSAGE,
((P2pException) e.getCause()).getType());
}
}

private List<BlockId> createContinuousBlockIds(long firstNum) {
List<BlockId> ids = new ArrayList<>();
for (int i = 0; i < NetConstants.SYNC_FETCH_BATCH_NUM; i++) {
ids.add(new BlockId(Sha256Hash.ZERO_HASH, firstNum + i));
}
return ids;
}

}
66 changes: 66 additions & 0 deletions framework/src/test/java/org/tron/core/net/messagehandler/FetchInvDataMsgHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,13 @@
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import org.junit.Assert;
import org.junit.Test;
Expand Down Expand Up @@ -68,6 +73,67 @@ public void testProcessMessage() throws Exception {
Assert.assertNotNull(syncBlockIdCache.getIfPresent(blockId));
}

@Test
public void testSyncBlockIdCacheRetainsFullWindowAndRejectsDuplicates() throws Exception {
assertFullWindowRetainedAndDuplicatesRejected(false);
}

@Test
public void testSyncBlockIdCacheRetainsShuffledWindowAndRejectsDuplicates() throws Exception {
assertFullWindowRetainedAndDuplicatesRejected(true);
}

private void assertFullWindowRetainedAndDuplicatesRejected(boolean shuffled) throws Exception {
PeerConnection peer = new PeerConnection();
int windowSize = 2 * (int) Parameter.NetConstants.SYNC_FETCH_BATCH_NUM + 1;
long lastHeight = 10000L;
peer.setNeedSyncFromUs(true);
peer.setLastSyncBlockId(new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, lastHeight));

FetchInvDataMsgHandler handler = new FetchInvDataMsgHandler();
Method check = FetchInvDataMsgHandler.class.getDeclaredMethod(
"check", PeerConnection.class, FetchInvDataMessage.class, boolean.class);
check.setAccessible(true);
List<Sha256Hash> hashes = new ArrayList<>();
for (int i = 0; i < windowSize; i++) {
hashes.add(new BlockCapsule.BlockId(createHash(i), lastHeight - windowSize + 1 + i));
}
if (shuffled) {
Collections.shuffle(hashes, new Random(6966L));
}

for (Sha256Hash hash : hashes) {
check.invoke(handler, peer, new FetchInvDataMessage(Collections.singletonList(hash),
Protocol.Inventory.InventoryType.BLOCK), false);
}
peer.getSyncBlockIdCache().cleanUp();

for (Sha256Hash hash : hashes) {
Assert.assertNotNull("Missing block " + new BlockCapsule.BlockId(hash).getNum(),
peer.getSyncBlockIdCache().getIfPresent(hash));
}
for (Sha256Hash hash : hashes) {
FetchInvDataMessage message = new FetchInvDataMessage(Collections.singletonList(hash),
Protocol.Inventory.InventoryType.BLOCK);
InvocationTargetException exception = Assert.assertThrows(InvocationTargetException.class,
() -> check.invoke(handler, peer, message, false));
Assert.assertTrue(exception.getCause() instanceof P2pException);
P2pException cause = (P2pException) exception.getCause();
Assert.assertEquals(P2pException.TypeEnum.BAD_MESSAGE, cause.getType());
Assert.assertEquals(new BlockCapsule.BlockId(hash).getString() + " is exist",
cause.getMessage());
}
}

private Sha256Hash createHash(int value) {
byte[] bytes = new byte[Sha256Hash.LENGTH];
bytes[28] = (byte) (value >>> 24);
bytes[29] = (byte) (value >>> 16);
bytes[30] = (byte) (value >>> 8);
bytes[31] = (byte) value;
return Sha256Hash.wrap(bytes);
}

@Test
public void testIsAdvInv() {
FetchInvDataMsgHandler fetchInvDataMsgHandler = new FetchInvDataMsgHandler();
Expand Down
22 changes: 22 additions & 0 deletions framework/src/test/java/org/tron/core/net/messagehandler/SyncBlockChainMsgHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
package org.tron.core.net.messagehandler;

import static org.tron.core.net.message.MessageTypes.SYNC_BLOCK_CHAIN;

import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
Expand All @@ -14,6 +16,7 @@
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.mockito.Mockito;
import org.tron.common.TestConstants;
import org.tron.common.application.TronApplicationContext;
import org.tron.common.utils.Sha256Hash;
Expand All @@ -22,6 +25,7 @@
import org.tron.core.config.DefaultConfig;
import org.tron.core.config.args.Args;
import org.tron.core.exception.P2pException;
import org.tron.core.net.P2pRateLimiter;
import org.tron.core.net.TronNetDelegate;
import org.tron.core.net.message.sync.BlockInventoryMessage;
import org.tron.core.net.message.sync.SyncBlockChainMessage;
Expand Down Expand Up @@ -160,6 +164,24 @@ public void testBlockIdsAtLimit() throws Exception {
}
}

@Test
public void testRemainNumZeroStillConsumesSyncBlockChainRateLimit() throws Exception {
PeerConnection rateLimitedPeer = Mockito.mock(PeerConnection.class);
P2pRateLimiter rateLimiter = new P2pRateLimiter();
rateLimiter.register(SYNC_BLOCK_CHAIN.asByte(), 0.0001D);
Mockito.when(rateLimitedPeer.getP2pRateLimiter()).thenReturn(rateLimiter);

BlockId genesis = context.getBean(TronNetDelegate.class).getGenesisBlockId();
SyncBlockChainMessage message = new SyncBlockChainMessage(
java.util.Collections.singletonList(genesis));
Method checkMethod = SyncBlockChainMsgHandler.class
.getDeclaredMethod("check", PeerConnection.class, SyncBlockChainMessage.class);
checkMethod.setAccessible(true);

Assert.assertTrue((boolean) checkMethod.invoke(handler, rateLimitedPeer, message));
Assert.assertFalse((boolean) checkMethod.invoke(handler, rateLimitedPeer, message));
}

@AfterClass
public static void destroy() {
for (PeerConnection p : PeerManager.getPeers()) {
Expand Down
Loading