From eb67cb2bf5b0389403b190f2345d21db5091ece6 Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Mon, 14 Sep 2026 22:42:23 +0800 Subject: [PATCH 1/4] fix the bug of marking sync complete when blockIdWeGet don't equal requested.getKey().peekLast() --- .../ChainInventoryMsgHandler.java | 24 +- .../tron/core/net/peer/PeerStatusCheck.java | 12 + .../tron/core/net/PeerSyncTestSupport.java | 36 +++ .../ChainInventoryCompletionTest.java | 298 ++++++++++++++++++ .../net/peer/PeerSyncChainTimeoutTest.java | 219 +++++++++++++ 5 files changed, 582 insertions(+), 7 deletions(-) create mode 100644 framework/src/test/java/org/tron/core/net/PeerSyncTestSupport.java create mode 100644 framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryCompletionTest.java create mode 100644 framework/src/test/java/org/tron/core/net/peer/PeerSyncChainTimeoutTest.java diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java index bd2e428418c..29bf83d1d88 100644 --- a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java +++ b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java @@ -10,6 +10,7 @@ import org.apache.commons.collections4.CollectionUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; +import org.tron.common.utils.Pair; import org.tron.core.capsule.BlockCapsule.BlockId; import org.tron.core.config.Parameter.ChainConstant; import org.tron.core.config.Parameter.NetConstants; @@ -40,7 +41,8 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep ChainInventoryMessage chainInventoryMessage = (ChainInventoryMessage) msg; - check(peer, chainInventoryMessage); + Pair, Long> requested = peer.getSyncChainRequested(); + check(requested, chainInventoryMessage); peer.setFetchAble(false); @@ -51,6 +53,12 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep Deque blockIdWeGet = new LinkedList<>(chainInventoryMessage.getBlockIds()); if (blockIdWeGet.size() == 1 && tronNetDelegate.containBlock(blockIdWeGet.peek())) { + // A single earlier summary block means the peer is behind the request snapshot. + // Completing our download must not mark both sync directions as complete. + if (!blockIdWeGet.peek().equals(requested.getKey().peekLast())) { + peer.setNeedSyncFromUs(true); + } + peer.setRemainNum(0); peer.setTronState(TronState.SYNC_COMPLETED); peer.setNeedSyncFromPeer(false); return; @@ -98,8 +106,9 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep } } - private void check(PeerConnection peer, ChainInventoryMessage msg) throws P2pException { - if (peer.getSyncChainRequested() == null) { + private void check(Pair, Long> requested, ChainInventoryMessage msg) + throws P2pException { + if (requested == null) { throw new P2pException(TypeEnum.BAD_MESSAGE, "not send syncBlockChainMsg"); } @@ -112,7 +121,8 @@ private void check(PeerConnection peer, ChainInventoryMessage msg) throws P2pExc throw new P2pException(TypeEnum.BAD_MESSAGE, "big blockIds size: " + blockIds.size()); } - if (msg.getRemainNum() != 0 && blockIds.size() < NetConstants.SYNC_FETCH_BATCH_NUM) { + if (msg.getRemainNum() < 0 + || (msg.getRemainNum() != 0 && blockIds.size() < NetConstants.SYNC_FETCH_BATCH_NUM)) { throw new P2pException(TypeEnum.BAD_MESSAGE, "remain: " + msg.getRemainNum() + ", blockIds size: " + blockIds.size()); } @@ -124,9 +134,9 @@ private void check(PeerConnection peer, ChainInventoryMessage msg) throws P2pExc } } - if (!peer.getSyncChainRequested().getKey().contains(blockIds.get(0))) { + if (!requested.getKey().contains(blockIds.get(0))) { throw new P2pException(TypeEnum.BAD_MESSAGE, "unlinked block, my head: " - + peer.getSyncChainRequested().getKey().getLast().getString() + + requested.getKey().getLast().getString() + ", peer: " + blockIds.get(0).getString()); } @@ -137,7 +147,7 @@ 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) { + if (lastNum > maxFutureNum || msg.getRemainNum() > maxFutureNum - lastNum) { throw new P2pException(TypeEnum.BAD_MESSAGE, "lastNum: " + lastNum + " + remainNum: " + msg.getRemainNum() + " > futureMaxNum: " + maxFutureNum); } diff --git a/framework/src/main/java/org/tron/core/net/peer/PeerStatusCheck.java b/framework/src/main/java/org/tron/core/net/peer/PeerStatusCheck.java index 04eac202484..201eb6800cf 100644 --- a/framework/src/main/java/org/tron/core/net/peer/PeerStatusCheck.java +++ b/framework/src/main/java/org/tron/core/net/peer/PeerStatusCheck.java @@ -1,11 +1,14 @@ package org.tron.core.net.peer; +import java.util.Deque; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.tron.common.es.ExecutorServiceManager; +import org.tron.common.utils.Pair; +import org.tron.core.capsule.BlockCapsule.BlockId; import org.tron.core.config.Parameter.NetConstants; import org.tron.core.net.TronNetDelegate; import org.tron.protos.Protocol.ReasonCode; @@ -64,6 +67,15 @@ public void statusCheck() { } } + if (!isDisconnected) { + Pair, Long> requested = peer.getSyncChainRequested(); + isDisconnected = requested != null + && requested.getValue() < now - NetConstants.SYNC_TIME_OUT; + if (isDisconnected) { + logger.warn("Peer {} get chain inventory timeout", peer.getInetAddress()); + } + } + if (!isDisconnected) { isDisconnected = peer.getSyncBlockRequested().values().stream() .anyMatch(time -> time < now - NetConstants.SYNC_TIME_OUT); diff --git a/framework/src/test/java/org/tron/core/net/PeerSyncTestSupport.java b/framework/src/test/java/org/tron/core/net/PeerSyncTestSupport.java new file mode 100644 index 00000000000..b3009b67142 --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/PeerSyncTestSupport.java @@ -0,0 +1,36 @@ +package org.tron.core.net; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; + +import java.net.InetSocketAddress; +import org.tron.common.utils.ReflectUtils; +import org.tron.common.utils.Sha256Hash; +import org.tron.core.capsule.BlockCapsule.BlockId; +import org.tron.core.net.peer.PeerConnection; +import org.tron.p2p.connection.Channel; + +public final class PeerSyncTestSupport { + + private PeerSyncTestSupport() { + } + + public static BlockId blockId(long number) { + return new BlockId(Sha256Hash.ZERO_HASH, number); + } + + public static PeerConnection peer(int port) { + PeerConnection peer = spy(new PeerConnection()); + Channel channel = mock(Channel.class); + InetSocketAddress address = new InetSocketAddress("127.0.0.1", port); + when(channel.getInetSocketAddress()).thenReturn(address); + when(channel.getInetAddress()).thenReturn(address.getAddress()); + ReflectUtils.setFieldValue(peer, "channel", channel); + doNothing().when(peer).sendMessage(any()); + doNothing().when(peer).disconnect(any()); + return peer; + } +} diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryCompletionTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryCompletionTest.java new file mode 100644 index 00000000000..099bf482d79 --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryCompletionTest.java @@ -0,0 +1,298 @@ +package org.tron.core.net.messagehandler; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.tron.core.net.PeerSyncTestSupport.blockId; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Deque; +import java.util.LinkedList; +import java.util.List; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +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.exception.P2pException; +import org.tron.core.exception.P2pException.TypeEnum; +import org.tron.core.net.PeerSyncTestSupport; +import org.tron.core.net.TronNetDelegate; +import org.tron.core.net.message.sync.ChainInventoryMessage; +import org.tron.core.net.message.sync.SyncBlockChainMessage; +import org.tron.core.net.peer.PeerConnection; +import org.tron.core.net.peer.TronState; +import org.tron.core.net.service.sync.SyncService; + +public class ChainInventoryCompletionTest { + + private ChainInventoryMsgHandler handler; + private TronNetDelegate delegate; + private SyncService sync; + private PeerConnection peer; + + @Before + public void setUp() throws Exception { + handler = new ChainInventoryMsgHandler(); + delegate = mock(TronNetDelegate.class); + sync = mock(SyncService.class); + ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(handler, "syncService", sync); + ReflectUtils.setFieldValue(handler, "syncFetchBatchNum", 100L); + when(delegate.getBlockLock()).thenReturn(new Object()); + when(delegate.getForkLock()).thenReturn(new Object()); + when(delegate.getHeadBlockId()).thenReturn(blockId(100)); + when(delegate.getKhaosDbHeadBlockId()).thenReturn(blockId(100)); + when(delegate.getSolidBlockId()).thenReturn(blockId(0)); + when(delegate.getGenesisBlockId()).thenReturn(blockId(0)); + when(delegate.getBlockTime(blockId(0))).thenReturn(System.currentTimeMillis() - 30_000_000L); + when(delegate.containBlock(any())).thenAnswer(invocation -> + ((BlockId) invocation.getArgument(0)).getNum() <= 100); + when(delegate.getBlockIdByNum(org.mockito.ArgumentMatchers.anyLong())) + .thenAnswer(invocation -> blockId(invocation.getArgument(0))); + peer = PeerSyncTestSupport.peer(10001); + peer.setTronState(TronState.SYNCING); + peer.setNeedSyncFromPeer(true); + peer.setNeedSyncFromUs(false); + peer.setFetchAble(true); + peer.setRemainNum(99); + peer.setBlockRcvTime(7); + request(blockId(0), blockId(50), blockId(100)); + } + + @Test + public void testSummaryTailCompletesDownload() throws Exception { + respond(0, blockId(100)); + + assertDownloadCompleted(); + Assert.assertTrue(peer.isSyncFinish()); + } + + @Test + public void testEarlierSummaryBlockMarksPeerBehind() throws Exception { + respond(0, blockId(50)); + + assertDownloadCompleted(); + Assert.assertTrue(peer.isNeedSyncFromUs()); + Assert.assertFalse(peer.isSyncFinish()); + } + + @Test + public void testEarliestSummaryBlockCannotCompleteBothDirections() throws Exception { + respond(0, blockId(0)); + + assertDownloadCompleted(); + Assert.assertTrue(peer.isNeedSyncFromUs()); + Assert.assertFalse(peer.isSyncFinish()); + } + + @Test + public void testTailComparisonUsesRequestSnapshotAfterHeadAdvances() throws Exception { + when(delegate.getHeadBlockId()).thenReturn(blockId(101)); + + respond(0, blockId(100)); + + assertDownloadCompleted(); + Assert.assertFalse(peer.isNeedSyncFromUs()); + } + + @Test + public void testTailDoesNotEraseExistingUploadDirection() throws Exception { + peer.setNeedSyncFromUs(true); + + respond(0, blockId(100)); + + assertDownloadCompleted(); + Assert.assertTrue(peer.isNeedSyncFromUs()); + Assert.assertFalse(peer.isSyncFinish()); + } + + @Test + public void testPeerBehindCanStartAnotherDownloadLater() throws Exception { + respond(0, blockId(50)); + SyncService service = new SyncService(); + ReflectUtils.setFieldValue(service, "tronNetDelegate", delegate); + try { + service.startSync(peer); + + Assert.assertEquals(TronState.SYNCING, peer.getTronState()); + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertNotNull(peer.getSyncChainRequested()); + Assert.assertEquals(blockId(100), peer.getSyncChainRequested().getKey().peekLast()); + verify(peer).sendMessage(any(SyncBlockChainMessage.class)); + verify(peer, never()).disconnect(any()); + } finally { + service.close(); + } + } + + @Test + public void testUnknownQueuedTailStillRequiresBlockDownload() throws Exception { + request(blockId(0), blockId(100), blockId(102)); + peer.getSyncBlockToFetch().addAll(Arrays.asList(blockId(101), blockId(102))); + + respond(0, blockId(102)); + + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertEquals(TronState.SYNCING, peer.getTronState()); + Assert.assertEquals(2, peer.getSyncBlockToFetch().size()); + Assert.assertTrue(peer.isFetchAble()); + Assert.assertNull(peer.getSyncChainRequested()); + Assert.assertEquals(7, peer.getBlockRcvTime()); + verify(sync).setFetchFlag(true); + verify(sync, never()).syncNext(any()); + } + + @Test + public void testFinalMultiBlockResponseSchedulesFetch() throws Exception { + respond(0, blockId(100), blockId(101), blockId(102)); + + Assert.assertEquals(Arrays.asList(blockId(101), blockId(102)), + new ArrayList<>(peer.getSyncBlockToFetch())); + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertEquals(TronState.SYNCING, peer.getTronState()); + Assert.assertEquals(0, peer.getRemainNum()); + Assert.assertEquals(7, peer.getBlockRcvTime()); + verify(sync).setFetchFlag(true); + verify(sync, never()).syncNext(any()); + } + + @Test + public void testKnownMultiBlockResponseRequestsNextSummary() throws Exception { + respond(0, blockId(50), blockId(51), blockId(52)); + + Assert.assertTrue(peer.getSyncBlockToFetch().isEmpty()); + Assert.assertEquals(blockId(52), peer.getBlockBothHave()); + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertEquals(7, peer.getBlockRcvTime()); + verify(sync).syncNext(peer); + } + + @Test + public void testPagedResponsePreservesPendingBlocks() throws Exception { + List firstPage = range(100, (int) NetConstants.SYNC_FETCH_BATCH_NUM + 1); + handler.processMessage(peer, new ChainInventoryMessage(firstPage, 2L)); + Assert.assertEquals(NetConstants.SYNC_FETCH_BATCH_NUM, peer.getSyncBlockToFetch().size()); + Assert.assertEquals(2, peer.getRemainNum()); + verify(sync).setFetchFlag(true); + + BlockId tail = firstPage.get(firstPage.size() - 1); + request(blockId(0), blockId(100), tail); + respond(0, tail, blockId(tail.getNum() + 1), blockId(tail.getNum() + 2)); + + Assert.assertEquals(NetConstants.SYNC_FETCH_BATCH_NUM + 2, + peer.getSyncBlockToFetch().size()); + Assert.assertEquals(blockId(101), peer.getSyncBlockToFetch().peekFirst()); + Assert.assertEquals(blockId(tail.getNum() + 2), peer.getSyncBlockToFetch().peekLast()); + Assert.assertEquals(0, peer.getRemainNum()); + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertEquals(7, peer.getBlockRcvTime()); + } + + @Test + public void testSingleBlockWithRemainingBlocksIsRejected() throws Exception { + assertRejected(new ChainInventoryMessage(Arrays.asList(blockId(50)), 1L)); + } + + @Test + public void testNegativeRemainingCountIsRejectedForFullBatch() throws Exception { + assertRejected(new ChainInventoryMessage(range(100, + (int) NetConstants.SYNC_FETCH_BATCH_NUM), -1L)); + } + + @Test + public void testOverflowingRemainingCountCannotBypassFutureLimit() throws Exception { + assertRejected(new ChainInventoryMessage(range(100, + (int) NetConstants.SYNC_FETCH_BATCH_NUM), Long.MAX_VALUE)); + } + + @Test + public void testFutureRangeIsRejected() throws Exception { + assertRejected(new ChainInventoryMessage(range(100, + (int) NetConstants.SYNC_FETCH_BATCH_NUM), 100_000L)); + } + + @Test + public void testKnownBlockOutsideSummaryIsRejected() throws Exception { + assertRejected(new ChainInventoryMessage(Arrays.asList(blockId(75)), 0L)); + } + + @Test + public void testSameHeightDifferentHashIsRejected() throws Exception { + byte[] hash = new byte[Sha256Hash.LENGTH]; + hash[31] = 1; + assertRejected(new ChainInventoryMessage( + Arrays.asList(new BlockId(Sha256Hash.wrap(hash), 100)), 0L)); + } + + @Test + public void testDiscontinuousResponseIsRejected() throws Exception { + assertRejected(new ChainInventoryMessage(Arrays.asList(blockId(100), blockId(102)), 0L)); + } + + @Test + public void testEmptyAndOversizedResponsesAreRejected() throws Exception { + assertRejected(new ChainInventoryMessage(new ArrayList<>(), 0L)); + assertRejected(new ChainInventoryMessage(range(100, + (int) NetConstants.SYNC_FETCH_BATCH_NUM + 2), 0L)); + } + + @Test + public void testUnsolicitedResponseIsRejected() throws Exception { + peer.setSyncChainRequested(null); + assertRejected(new ChainInventoryMessage(Arrays.asList(blockId(100)), 0L)); + } + + private void request(BlockId... ids) { + peer.setSyncChainRequested(new Pair<>(new LinkedList<>(Arrays.asList(ids)), + System.currentTimeMillis())); + } + + private void respond(long remain, BlockId... ids) throws Exception { + handler.processMessage(peer, new ChainInventoryMessage(Arrays.asList(ids), remain)); + } + + private void assertDownloadCompleted() { + Assert.assertFalse(peer.isNeedSyncFromPeer()); + Assert.assertEquals(TronState.SYNC_COMPLETED, peer.getTronState()); + Assert.assertNull(peer.getSyncChainRequested()); + Assert.assertFalse(peer.isFetchAble()); + Assert.assertEquals(0, peer.getRemainNum()); + Assert.assertEquals(7, peer.getBlockRcvTime()); + verify(sync, never()).syncNext(any()); + verify(sync, never()).setFetchFlag(true); + } + + private void assertRejected(ChainInventoryMessage message) throws Exception { + Pair, Long> requested = peer.getSyncChainRequested(); + try { + handler.processMessage(peer, message); + Assert.fail("Invalid chain inventory must be rejected before changing peer state"); + } catch (P2pException e) { + Assert.assertEquals(TypeEnum.BAD_MESSAGE, e.getType()); + } + Assert.assertSame(requested, peer.getSyncChainRequested()); + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertFalse(peer.isNeedSyncFromUs()); + Assert.assertTrue(peer.isFetchAble()); + Assert.assertEquals(TronState.SYNCING, peer.getTronState()); + Assert.assertEquals(99, peer.getRemainNum()); + Assert.assertEquals(7, peer.getBlockRcvTime()); + verify(sync, never()).syncNext(any()); + verify(sync, never()).setFetchFlag(true); + } + + private List range(long first, int size) { + List ids = new ArrayList<>(); + for (int i = 0; i < size; i++) { + ids.add(blockId(first + i)); + } + return ids; + } +} diff --git a/framework/src/test/java/org/tron/core/net/peer/PeerSyncChainTimeoutTest.java b/framework/src/test/java/org/tron/core/net/peer/PeerSyncChainTimeoutTest.java new file mode 100644 index 00000000000..c8308f019b1 --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/peer/PeerSyncChainTimeoutTest.java @@ -0,0 +1,219 @@ +package org.tron.core.net.peer; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.tron.core.net.PeerSyncTestSupport.blockId; + +import java.util.Arrays; +import java.util.Collections; +import java.util.Deque; +import java.util.LinkedList; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.tron.common.utils.Pair; +import org.tron.common.utils.ReflectUtils; +import org.tron.core.capsule.BlockCapsule.BlockId; +import org.tron.core.config.Parameter.NetConstants; +import org.tron.core.net.P2pEventHandlerImpl; +import org.tron.core.net.PeerSyncTestSupport; +import org.tron.core.net.TronNetDelegate; +import org.tron.core.net.message.TronMessage; +import org.tron.core.net.message.keepalive.PingMessage; +import org.tron.core.net.message.keepalive.PongMessage; +import org.tron.core.net.message.sync.ChainInventoryMessage; +import org.tron.core.net.message.sync.SyncBlockChainMessage; +import org.tron.core.net.messagehandler.ChainInventoryMsgHandler; +import org.tron.core.net.messagehandler.SyncBlockChainMsgHandler; +import org.tron.core.net.service.keepalive.KeepAliveService; +import org.tron.core.net.service.sync.SyncService; +import org.tron.protos.Protocol.Inventory.InventoryType; +import org.tron.protos.Protocol.ReasonCode; + +public class PeerSyncChainTimeoutTest { + + private PeerStatusCheck check; + private TronNetDelegate delegate; + private PeerConnection peer; + + @Before + public void setUp() throws Exception { + check = new PeerStatusCheck(); + delegate = mock(TronNetDelegate.class); + ReflectUtils.setFieldValue(check, "tronNetDelegate", delegate); + peer = PeerSyncTestSupport.peer(11001); + peer.setNeedSyncFromPeer(false); + peer.setNeedSyncFromUs(false); + peer.setLastInteractiveTime(1); + when(delegate.getActivePeer()).thenReturn(Collections.singletonList(peer)); + when(delegate.getHeadBlockId()).thenReturn(blockId(20)); + when(delegate.getKhaosDbHeadBlockId()).thenReturn(blockId(20)); + when(delegate.getSolidBlockId()).thenReturn(blockId(0)); + when(delegate.getBlockTime(blockId(0))).thenReturn(System.currentTimeMillis() - 60_000); + when(delegate.getForkLock()).thenReturn(new Object()); + when(delegate.getBlockIdByNum(anyLong())) + .thenAnswer(invocation -> blockId(invocation.getArgument(0))); + request(peer, System.currentTimeMillis() - 6_000); + } + + @After + public void tearDown() { + check.close(); + } + + @Test + public void testExpiredSummaryDisconnectsOnlyResponsiblePeer() { + PeerConnection other = PeerSyncTestSupport.peer(11002); + other.setNeedSyncFromPeer(false); + other.setNeedSyncFromUs(false); + request(other, System.currentTimeMillis()); + when(delegate.getActivePeer()).thenReturn(Arrays.asList(peer, other)); + + check.statusCheck(); + + verify(peer).disconnect(ReasonCode.TIME_OUT); + verify(other, never()).disconnect(any()); + } + + @Test + public void testRecentSyncProgressAndInteractionDoNotExtendDeadline() { + Pair, Long> requested = peer.getSyncChainRequested(); + peer.setNeedSyncFromPeer(true); + peer.setBlockBothHave(blockId(20)); + peer.setLastInteractiveTime(System.currentTimeMillis()); + peer.setBlockRcvTime(System.currentTimeMillis()); + + check.statusCheck(); + + Assert.assertSame(requested, peer.getSyncChainRequested()); + verify(peer).disconnect(ReasonCode.TIME_OUT); + } + + @Test + public void testPingAndPongDoNotExtendDeadline() { + Pair, Long> requested = peer.getSyncChainRequested(); + P2pEventHandlerImpl events = new P2pEventHandlerImpl(); + ReflectUtils.setFieldValue(events, "keepAliveService", new KeepAliveService()); + + dispatch(events, new PingMessage()); + dispatch(events, new PongMessage()); + + verify(peer).sendMessage(any(PongMessage.class)); + verify(peer, never()).disconnect(any()); + Assert.assertSame(requested, peer.getSyncChainRequested()); + Assert.assertEquals(1, peer.getLastInteractiveTime()); + check.statusCheck(); + verify(peer).disconnect(ReasonCode.TIME_OUT); + } + + @Test + public void testValidIncomingSyncRequestDoesNotExtendOutgoingDeadline() throws Exception { + Pair, Long> requested = peer.getSyncChainRequested(); + SyncBlockChainMsgHandler handler = new SyncBlockChainMsgHandler(); + ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate); + when(delegate.containBlockInMainChain(any())).thenReturn(true); + P2pEventHandlerImpl events = new P2pEventHandlerImpl(); + ReflectUtils.setFieldValue(events, "syncBlockChainMsgHandler", handler); + + dispatch(events, new SyncBlockChainMessage( + new LinkedList<>(Arrays.asList(blockId(0), blockId(20))))); + + verify(peer).sendMessage(any(ChainInventoryMessage.class)); + verify(peer, never()).disconnect(any()); + Assert.assertTrue(peer.getLastInteractiveTime() > 1); + Assert.assertSame(requested, peer.getSyncChainRequested()); + check.statusCheck(); + verify(peer).disconnect(ReasonCode.TIME_OUT); + } + + @Test + public void testRequestWithinFiveSecondsDoesNotTimeOut() { + request(peer, System.currentTimeMillis() - 1_000); + + check.statusCheck(); + + verify(peer, never()).disconnect(any()); + } + + @Test + public void testCompletedResponseClearsDeadlineWithoutContribution() throws Exception { + ChainInventoryMsgHandler handler = new ChainInventoryMsgHandler(); + ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(handler, "syncService", mock(SyncService.class)); + when(delegate.containBlock(blockId(20))).thenReturn(true); + peer.setNeedSyncFromPeer(true); + peer.setBlockRcvTime(7); + + handler.processMessage(peer, new ChainInventoryMessage(Collections.singletonList(blockId(20)), + 0L)); + check.statusCheck(); + + Assert.assertNull(peer.getSyncChainRequested()); + Assert.assertTrue(peer.isSyncFinish()); + Assert.assertEquals(7, peer.getBlockRcvTime()); + verify(peer, never()).disconnect(any()); + } + + @Test + public void testNewRequestUsesItsOwnDeadline() { + peer.setSyncChainRequested(null); + check.statusCheck(); + request(peer, System.currentTimeMillis()); + check.statusCheck(); + + verify(peer, never()).disconnect(any()); + } + + @Test + public void testRepeatedSyncNextDoesNotReplacePendingRequest() { + SyncService service = new SyncService(); + ReflectUtils.setFieldValue(service, "tronNetDelegate", delegate); + peer.setSyncChainRequested(null); + try { + service.syncNext(peer); + Pair, Long> requested = peer.getSyncChainRequested(); + Assert.assertNotNull(requested); + Assert.assertEquals(blockId(20), requested.getKey().peekLast()); + + service.syncNext(peer); + + Assert.assertSame(requested, peer.getSyncChainRequested()); + verify(peer, times(1)).sendMessage(any(SyncBlockChainMessage.class)); + verify(peer, never()).disconnect(any()); + } finally { + service.close(); + } + } + + @Test + public void testExistingInventoryAndSyncBlockTimeoutsRemainEffective() { + peer.setSyncChainRequested(null); + peer.getAdvInvRequest().put(new Item(blockId(21), InventoryType.BLOCK), + System.currentTimeMillis() - NetConstants.ADV_TIME_OUT - 1_000); + PeerConnection other = PeerSyncTestSupport.peer(11002); + other.setNeedSyncFromPeer(false); + other.getSyncBlockRequested().put(blockId(21), System.currentTimeMillis() - 6_000); + when(delegate.getActivePeer()).thenReturn(Arrays.asList(peer, other)); + + check.statusCheck(); + + verify(peer).disconnect(ReasonCode.TIME_OUT); + verify(other).disconnect(ReasonCode.TIME_OUT); + } + + private void request(PeerConnection target, long time) { + target.setSyncChainRequested(new Pair<>( + new LinkedList<>(Arrays.asList(blockId(0), blockId(20))), time)); + } + + private void dispatch(P2pEventHandlerImpl events, TronMessage message) { + ReflectUtils.invokeMethod(events, "processMessage", + new Class[]{PeerConnection.class, byte[].class}, peer, message.getSendBytes()); + } +} From c2081411278b9cd40a807db1dcc79c3b511ed68f Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Wed, 23 Sep 2026 18:44:56 +0800 Subject: [PATCH 2/4] don't setNeedSyncFromUs = false when last blockid from peer belows than me --- .../ChainInventoryMsgHandler.java | 8 +- .../ChainInventoryCompletionTest.java | 78 +++++++++++++++++-- 2 files changed, 74 insertions(+), 12 deletions(-) diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java index 29bf83d1d88..f3ac1f02fdd 100644 --- a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java +++ b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java @@ -53,13 +53,11 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep Deque blockIdWeGet = new LinkedList<>(chainInventoryMessage.getBlockIds()); if (blockIdWeGet.size() == 1 && tronNetDelegate.containBlock(blockIdWeGet.peek())) { - // A single earlier summary block means the peer is behind the request snapshot. - // Completing our download must not mark both sync directions as complete. - if (!blockIdWeGet.peek().equals(requested.getKey().peekLast())) { - peer.setNeedSyncFromUs(true); - } peer.setRemainNum(0); peer.setTronState(TronState.SYNC_COMPLETED); + // This trusts the peer's claim that no blocks remain. A peer can echo an earlier + // known summary block while withholding newer blocks, so ending this download + // does not prove that we have caught up with the peer's actual chain. todo fix peer.setNeedSyncFromPeer(false); return; } diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryCompletionTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryCompletionTest.java index 099bf482d79..3cc78ed19f9 100644 --- a/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryCompletionTest.java +++ b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryCompletionTest.java @@ -9,12 +9,15 @@ import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.Deque; import java.util.LinkedList; import java.util.List; import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.tron.common.overlay.message.Message; import org.tron.common.utils.Pair; import org.tron.common.utils.ReflectUtils; import org.tron.common.utils.Sha256Hash; @@ -24,11 +27,15 @@ import org.tron.core.exception.P2pException.TypeEnum; import org.tron.core.net.PeerSyncTestSupport; import org.tron.core.net.TronNetDelegate; +import org.tron.core.net.message.adv.InventoryMessage; import org.tron.core.net.message.sync.ChainInventoryMessage; import org.tron.core.net.message.sync.SyncBlockChainMessage; +import org.tron.core.net.peer.Item; import org.tron.core.net.peer.PeerConnection; import org.tron.core.net.peer.TronState; +import org.tron.core.net.service.adv.AdvService; import org.tron.core.net.service.sync.SyncService; +import org.tron.protos.Protocol.Inventory.InventoryType; public class ChainInventoryCompletionTest { @@ -75,25 +82,25 @@ public void testSummaryTailCompletesDownload() throws Exception { } @Test - public void testEarlierSummaryBlockMarksPeerBehind() throws Exception { + public void testEarlierSummaryBlockPreservesNoUploadRequirement() throws Exception { respond(0, blockId(50)); assertDownloadCompleted(); - Assert.assertTrue(peer.isNeedSyncFromUs()); - Assert.assertFalse(peer.isSyncFinish()); + Assert.assertFalse(peer.isNeedSyncFromUs()); + Assert.assertTrue(peer.isSyncFinish()); } @Test - public void testEarliestSummaryBlockCannotCompleteBothDirections() throws Exception { + public void testEarliestSummaryBlockPreservesNoUploadRequirement() throws Exception { respond(0, blockId(0)); assertDownloadCompleted(); - Assert.assertTrue(peer.isNeedSyncFromUs()); - Assert.assertFalse(peer.isSyncFinish()); + Assert.assertFalse(peer.isNeedSyncFromUs()); + Assert.assertTrue(peer.isSyncFinish()); } @Test - public void testTailComparisonUsesRequestSnapshotAfterHeadAdvances() throws Exception { + public void testKnownResponseCompletesDownloadAfterHeadAdvances() throws Exception { when(delegate.getHeadBlockId()).thenReturn(blockId(101)); respond(0, blockId(100)); @@ -113,6 +120,63 @@ public void testTailDoesNotEraseExistingUploadDirection() throws Exception { Assert.assertFalse(peer.isSyncFinish()); } + @Test + public void testEarlierSummaryBlockPreservesExistingUploadRequirement() throws Exception { + peer.setNeedSyncFromUs(true); + + respond(0, blockId(50)); + + assertDownloadCompleted(); + Assert.assertTrue(peer.isNeedSyncFromUs()); + Assert.assertFalse(peer.isSyncFinish()); + } + + @Test + public void testSingleBlockReplyFromExistingPeerKeepsInventoryFlow() throws Exception { + // The remote peer still believes it is ahead based on Hello. Our summary now includes + // newer blocks learned from another connection; replying does not start a remote download. + PeerConnection remotePeer = PeerSyncTestSupport.peer(10002); + remotePeer.setNeedSyncFromPeer(false); + remotePeer.setNeedSyncFromUs(true); + TronNetDelegate remoteDelegate = mock(TronNetDelegate.class); + when(remoteDelegate.getHeadBlockId()).thenReturn(blockId(50)); + when(remoteDelegate.containBlockInMainChain(any())).thenAnswer(invocation -> + ((BlockId) invocation.getArgument(0)).getNum() <= 50); + SyncBlockChainMsgHandler remoteHandler = new SyncBlockChainMsgHandler(); + ReflectUtils.setFieldValue(remoteHandler, "tronNetDelegate", remoteDelegate); + + remoteHandler.processMessage(remotePeer, + new SyncBlockChainMessage(new ArrayList<>(peer.getSyncChainRequested().getKey()))); + + ArgumentCaptor reply = ArgumentCaptor.forClass(Message.class); + verify(remotePeer).sendMessage(reply.capture()); + Assert.assertTrue(reply.getValue() instanceof ChainInventoryMessage); + ChainInventoryMessage response = new ChainInventoryMessage(reply.getValue().getData()); + Assert.assertEquals(Collections.singletonList(blockId(50)), response.getBlockIds()); + Assert.assertEquals(Long.valueOf(0), response.getRemainNum()); + handler.processMessage(peer, response); + + assertDownloadCompleted(); + Assert.assertTrue(peer.isSyncFinish()); + Assert.assertTrue(remotePeer.isSyncFinish()); + verify(remotePeer, never()).disconnect(any()); + + InventoryMsgHandler inventoryHandler = new InventoryMsgHandler(); + AdvService adv = mock(AdvService.class); + ReflectUtils.setFieldValue(inventoryHandler, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(inventoryHandler, "advService", adv); + ReflectUtils.setFieldValue(inventoryHandler, "transactionsMsgHandler", + mock(TransactionsMsgHandler.class)); + for (InventoryType type : Arrays.asList(InventoryType.BLOCK, InventoryType.TRX)) { + Item item = new Item(blockId(101), type); + inventoryHandler.processMessage(peer, + new InventoryMessage(Collections.singletonList(item.getHash()), type)); + Assert.assertNotNull(peer.getAdvInvReceive().getIfPresent(item)); + verify(adv).addInv(item); + } + verify(peer, never()).disconnect(any()); + } + @Test public void testPeerBehindCanStartAnotherDownloadLater() throws Exception { respond(0, blockId(50)); From b578da1dc297fe2330bb9572a9539047c863236f Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Thu, 24 Sep 2026 17:36:56 +0800 Subject: [PATCH 3/4] simplify test cases --- .../ChainInventoryCompletionTest.java | 362 ------------------ .../ChainInventoryMsgHandlerTest.java | 337 ++++++++++++++-- .../net/peer/PeerStatusCheckMockTest.java | 213 ++++++++++- .../net/peer/PeerSyncChainTimeoutTest.java | 219 ----------- 4 files changed, 513 insertions(+), 618 deletions(-) delete mode 100644 framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryCompletionTest.java delete mode 100644 framework/src/test/java/org/tron/core/net/peer/PeerSyncChainTimeoutTest.java diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryCompletionTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryCompletionTest.java deleted file mode 100644 index 3cc78ed19f9..00000000000 --- a/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryCompletionTest.java +++ /dev/null @@ -1,362 +0,0 @@ -package org.tron.core.net.messagehandler; - -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; -import static org.tron.core.net.PeerSyncTestSupport.blockId; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collections; -import java.util.Deque; -import java.util.LinkedList; -import java.util.List; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Test; -import org.mockito.ArgumentCaptor; -import org.tron.common.overlay.message.Message; -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.exception.P2pException; -import org.tron.core.exception.P2pException.TypeEnum; -import org.tron.core.net.PeerSyncTestSupport; -import org.tron.core.net.TronNetDelegate; -import org.tron.core.net.message.adv.InventoryMessage; -import org.tron.core.net.message.sync.ChainInventoryMessage; -import org.tron.core.net.message.sync.SyncBlockChainMessage; -import org.tron.core.net.peer.Item; -import org.tron.core.net.peer.PeerConnection; -import org.tron.core.net.peer.TronState; -import org.tron.core.net.service.adv.AdvService; -import org.tron.core.net.service.sync.SyncService; -import org.tron.protos.Protocol.Inventory.InventoryType; - -public class ChainInventoryCompletionTest { - - private ChainInventoryMsgHandler handler; - private TronNetDelegate delegate; - private SyncService sync; - private PeerConnection peer; - - @Before - public void setUp() throws Exception { - handler = new ChainInventoryMsgHandler(); - delegate = mock(TronNetDelegate.class); - sync = mock(SyncService.class); - ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate); - ReflectUtils.setFieldValue(handler, "syncService", sync); - ReflectUtils.setFieldValue(handler, "syncFetchBatchNum", 100L); - when(delegate.getBlockLock()).thenReturn(new Object()); - when(delegate.getForkLock()).thenReturn(new Object()); - when(delegate.getHeadBlockId()).thenReturn(blockId(100)); - when(delegate.getKhaosDbHeadBlockId()).thenReturn(blockId(100)); - when(delegate.getSolidBlockId()).thenReturn(blockId(0)); - when(delegate.getGenesisBlockId()).thenReturn(blockId(0)); - when(delegate.getBlockTime(blockId(0))).thenReturn(System.currentTimeMillis() - 30_000_000L); - when(delegate.containBlock(any())).thenAnswer(invocation -> - ((BlockId) invocation.getArgument(0)).getNum() <= 100); - when(delegate.getBlockIdByNum(org.mockito.ArgumentMatchers.anyLong())) - .thenAnswer(invocation -> blockId(invocation.getArgument(0))); - peer = PeerSyncTestSupport.peer(10001); - peer.setTronState(TronState.SYNCING); - peer.setNeedSyncFromPeer(true); - peer.setNeedSyncFromUs(false); - peer.setFetchAble(true); - peer.setRemainNum(99); - peer.setBlockRcvTime(7); - request(blockId(0), blockId(50), blockId(100)); - } - - @Test - public void testSummaryTailCompletesDownload() throws Exception { - respond(0, blockId(100)); - - assertDownloadCompleted(); - Assert.assertTrue(peer.isSyncFinish()); - } - - @Test - public void testEarlierSummaryBlockPreservesNoUploadRequirement() throws Exception { - respond(0, blockId(50)); - - assertDownloadCompleted(); - Assert.assertFalse(peer.isNeedSyncFromUs()); - Assert.assertTrue(peer.isSyncFinish()); - } - - @Test - public void testEarliestSummaryBlockPreservesNoUploadRequirement() throws Exception { - respond(0, blockId(0)); - - assertDownloadCompleted(); - Assert.assertFalse(peer.isNeedSyncFromUs()); - Assert.assertTrue(peer.isSyncFinish()); - } - - @Test - public void testKnownResponseCompletesDownloadAfterHeadAdvances() throws Exception { - when(delegate.getHeadBlockId()).thenReturn(blockId(101)); - - respond(0, blockId(100)); - - assertDownloadCompleted(); - Assert.assertFalse(peer.isNeedSyncFromUs()); - } - - @Test - public void testTailDoesNotEraseExistingUploadDirection() throws Exception { - peer.setNeedSyncFromUs(true); - - respond(0, blockId(100)); - - assertDownloadCompleted(); - Assert.assertTrue(peer.isNeedSyncFromUs()); - Assert.assertFalse(peer.isSyncFinish()); - } - - @Test - public void testEarlierSummaryBlockPreservesExistingUploadRequirement() throws Exception { - peer.setNeedSyncFromUs(true); - - respond(0, blockId(50)); - - assertDownloadCompleted(); - Assert.assertTrue(peer.isNeedSyncFromUs()); - Assert.assertFalse(peer.isSyncFinish()); - } - - @Test - public void testSingleBlockReplyFromExistingPeerKeepsInventoryFlow() throws Exception { - // The remote peer still believes it is ahead based on Hello. Our summary now includes - // newer blocks learned from another connection; replying does not start a remote download. - PeerConnection remotePeer = PeerSyncTestSupport.peer(10002); - remotePeer.setNeedSyncFromPeer(false); - remotePeer.setNeedSyncFromUs(true); - TronNetDelegate remoteDelegate = mock(TronNetDelegate.class); - when(remoteDelegate.getHeadBlockId()).thenReturn(blockId(50)); - when(remoteDelegate.containBlockInMainChain(any())).thenAnswer(invocation -> - ((BlockId) invocation.getArgument(0)).getNum() <= 50); - SyncBlockChainMsgHandler remoteHandler = new SyncBlockChainMsgHandler(); - ReflectUtils.setFieldValue(remoteHandler, "tronNetDelegate", remoteDelegate); - - remoteHandler.processMessage(remotePeer, - new SyncBlockChainMessage(new ArrayList<>(peer.getSyncChainRequested().getKey()))); - - ArgumentCaptor reply = ArgumentCaptor.forClass(Message.class); - verify(remotePeer).sendMessage(reply.capture()); - Assert.assertTrue(reply.getValue() instanceof ChainInventoryMessage); - ChainInventoryMessage response = new ChainInventoryMessage(reply.getValue().getData()); - Assert.assertEquals(Collections.singletonList(blockId(50)), response.getBlockIds()); - Assert.assertEquals(Long.valueOf(0), response.getRemainNum()); - handler.processMessage(peer, response); - - assertDownloadCompleted(); - Assert.assertTrue(peer.isSyncFinish()); - Assert.assertTrue(remotePeer.isSyncFinish()); - verify(remotePeer, never()).disconnect(any()); - - InventoryMsgHandler inventoryHandler = new InventoryMsgHandler(); - AdvService adv = mock(AdvService.class); - ReflectUtils.setFieldValue(inventoryHandler, "tronNetDelegate", delegate); - ReflectUtils.setFieldValue(inventoryHandler, "advService", adv); - ReflectUtils.setFieldValue(inventoryHandler, "transactionsMsgHandler", - mock(TransactionsMsgHandler.class)); - for (InventoryType type : Arrays.asList(InventoryType.BLOCK, InventoryType.TRX)) { - Item item = new Item(blockId(101), type); - inventoryHandler.processMessage(peer, - new InventoryMessage(Collections.singletonList(item.getHash()), type)); - Assert.assertNotNull(peer.getAdvInvReceive().getIfPresent(item)); - verify(adv).addInv(item); - } - verify(peer, never()).disconnect(any()); - } - - @Test - public void testPeerBehindCanStartAnotherDownloadLater() throws Exception { - respond(0, blockId(50)); - SyncService service = new SyncService(); - ReflectUtils.setFieldValue(service, "tronNetDelegate", delegate); - try { - service.startSync(peer); - - Assert.assertEquals(TronState.SYNCING, peer.getTronState()); - Assert.assertTrue(peer.isNeedSyncFromPeer()); - Assert.assertNotNull(peer.getSyncChainRequested()); - Assert.assertEquals(blockId(100), peer.getSyncChainRequested().getKey().peekLast()); - verify(peer).sendMessage(any(SyncBlockChainMessage.class)); - verify(peer, never()).disconnect(any()); - } finally { - service.close(); - } - } - - @Test - public void testUnknownQueuedTailStillRequiresBlockDownload() throws Exception { - request(blockId(0), blockId(100), blockId(102)); - peer.getSyncBlockToFetch().addAll(Arrays.asList(blockId(101), blockId(102))); - - respond(0, blockId(102)); - - Assert.assertTrue(peer.isNeedSyncFromPeer()); - Assert.assertEquals(TronState.SYNCING, peer.getTronState()); - Assert.assertEquals(2, peer.getSyncBlockToFetch().size()); - Assert.assertTrue(peer.isFetchAble()); - Assert.assertNull(peer.getSyncChainRequested()); - Assert.assertEquals(7, peer.getBlockRcvTime()); - verify(sync).setFetchFlag(true); - verify(sync, never()).syncNext(any()); - } - - @Test - public void testFinalMultiBlockResponseSchedulesFetch() throws Exception { - respond(0, blockId(100), blockId(101), blockId(102)); - - Assert.assertEquals(Arrays.asList(blockId(101), blockId(102)), - new ArrayList<>(peer.getSyncBlockToFetch())); - Assert.assertTrue(peer.isNeedSyncFromPeer()); - Assert.assertEquals(TronState.SYNCING, peer.getTronState()); - Assert.assertEquals(0, peer.getRemainNum()); - Assert.assertEquals(7, peer.getBlockRcvTime()); - verify(sync).setFetchFlag(true); - verify(sync, never()).syncNext(any()); - } - - @Test - public void testKnownMultiBlockResponseRequestsNextSummary() throws Exception { - respond(0, blockId(50), blockId(51), blockId(52)); - - Assert.assertTrue(peer.getSyncBlockToFetch().isEmpty()); - Assert.assertEquals(blockId(52), peer.getBlockBothHave()); - Assert.assertTrue(peer.isNeedSyncFromPeer()); - Assert.assertEquals(7, peer.getBlockRcvTime()); - verify(sync).syncNext(peer); - } - - @Test - public void testPagedResponsePreservesPendingBlocks() throws Exception { - List firstPage = range(100, (int) NetConstants.SYNC_FETCH_BATCH_NUM + 1); - handler.processMessage(peer, new ChainInventoryMessage(firstPage, 2L)); - Assert.assertEquals(NetConstants.SYNC_FETCH_BATCH_NUM, peer.getSyncBlockToFetch().size()); - Assert.assertEquals(2, peer.getRemainNum()); - verify(sync).setFetchFlag(true); - - BlockId tail = firstPage.get(firstPage.size() - 1); - request(blockId(0), blockId(100), tail); - respond(0, tail, blockId(tail.getNum() + 1), blockId(tail.getNum() + 2)); - - Assert.assertEquals(NetConstants.SYNC_FETCH_BATCH_NUM + 2, - peer.getSyncBlockToFetch().size()); - Assert.assertEquals(blockId(101), peer.getSyncBlockToFetch().peekFirst()); - Assert.assertEquals(blockId(tail.getNum() + 2), peer.getSyncBlockToFetch().peekLast()); - Assert.assertEquals(0, peer.getRemainNum()); - Assert.assertTrue(peer.isNeedSyncFromPeer()); - Assert.assertEquals(7, peer.getBlockRcvTime()); - } - - @Test - public void testSingleBlockWithRemainingBlocksIsRejected() throws Exception { - assertRejected(new ChainInventoryMessage(Arrays.asList(blockId(50)), 1L)); - } - - @Test - public void testNegativeRemainingCountIsRejectedForFullBatch() throws Exception { - assertRejected(new ChainInventoryMessage(range(100, - (int) NetConstants.SYNC_FETCH_BATCH_NUM), -1L)); - } - - @Test - public void testOverflowingRemainingCountCannotBypassFutureLimit() throws Exception { - assertRejected(new ChainInventoryMessage(range(100, - (int) NetConstants.SYNC_FETCH_BATCH_NUM), Long.MAX_VALUE)); - } - - @Test - public void testFutureRangeIsRejected() throws Exception { - assertRejected(new ChainInventoryMessage(range(100, - (int) NetConstants.SYNC_FETCH_BATCH_NUM), 100_000L)); - } - - @Test - public void testKnownBlockOutsideSummaryIsRejected() throws Exception { - assertRejected(new ChainInventoryMessage(Arrays.asList(blockId(75)), 0L)); - } - - @Test - public void testSameHeightDifferentHashIsRejected() throws Exception { - byte[] hash = new byte[Sha256Hash.LENGTH]; - hash[31] = 1; - assertRejected(new ChainInventoryMessage( - Arrays.asList(new BlockId(Sha256Hash.wrap(hash), 100)), 0L)); - } - - @Test - public void testDiscontinuousResponseIsRejected() throws Exception { - assertRejected(new ChainInventoryMessage(Arrays.asList(blockId(100), blockId(102)), 0L)); - } - - @Test - public void testEmptyAndOversizedResponsesAreRejected() throws Exception { - assertRejected(new ChainInventoryMessage(new ArrayList<>(), 0L)); - assertRejected(new ChainInventoryMessage(range(100, - (int) NetConstants.SYNC_FETCH_BATCH_NUM + 2), 0L)); - } - - @Test - public void testUnsolicitedResponseIsRejected() throws Exception { - peer.setSyncChainRequested(null); - assertRejected(new ChainInventoryMessage(Arrays.asList(blockId(100)), 0L)); - } - - private void request(BlockId... ids) { - peer.setSyncChainRequested(new Pair<>(new LinkedList<>(Arrays.asList(ids)), - System.currentTimeMillis())); - } - - private void respond(long remain, BlockId... ids) throws Exception { - handler.processMessage(peer, new ChainInventoryMessage(Arrays.asList(ids), remain)); - } - - private void assertDownloadCompleted() { - Assert.assertFalse(peer.isNeedSyncFromPeer()); - Assert.assertEquals(TronState.SYNC_COMPLETED, peer.getTronState()); - Assert.assertNull(peer.getSyncChainRequested()); - Assert.assertFalse(peer.isFetchAble()); - Assert.assertEquals(0, peer.getRemainNum()); - Assert.assertEquals(7, peer.getBlockRcvTime()); - verify(sync, never()).syncNext(any()); - verify(sync, never()).setFetchFlag(true); - } - - private void assertRejected(ChainInventoryMessage message) throws Exception { - Pair, Long> requested = peer.getSyncChainRequested(); - try { - handler.processMessage(peer, message); - Assert.fail("Invalid chain inventory must be rejected before changing peer state"); - } catch (P2pException e) { - Assert.assertEquals(TypeEnum.BAD_MESSAGE, e.getType()); - } - Assert.assertSame(requested, peer.getSyncChainRequested()); - Assert.assertTrue(peer.isNeedSyncFromPeer()); - Assert.assertFalse(peer.isNeedSyncFromUs()); - Assert.assertTrue(peer.isFetchAble()); - Assert.assertEquals(TronState.SYNCING, peer.getTronState()); - Assert.assertEquals(99, peer.getRemainNum()); - Assert.assertEquals(7, peer.getBlockRcvTime()); - verify(sync, never()).syncNext(any()); - verify(sync, never()).setFetchFlag(true); - } - - private List range(long first, int size) { - List ids = new ArrayList<>(); - for (int i = 0; i < size; i++) { - ids.add(blockId(first + i)); - } - return ids; - } -} diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java index 56853c3dbb7..66a60304718 100644 --- a/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java +++ b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java @@ -1,24 +1,53 @@ package org.tron.core.net.messagehandler; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.tron.core.net.PeerSyncTestSupport.blockId; + import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Deque; import java.util.LinkedList; import java.util.List; import org.junit.AfterClass; import org.junit.Assert; +import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; +import org.mockito.ArgumentCaptor; import org.tron.common.TestConstants; +import org.tron.common.overlay.message.Message; 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.exception.P2pException.TypeEnum; +import org.tron.core.net.PeerSyncTestSupport; +import org.tron.core.net.TronNetDelegate; +import org.tron.core.net.message.adv.InventoryMessage; import org.tron.core.net.message.sync.ChainInventoryMessage; +import org.tron.core.net.message.sync.SyncBlockChainMessage; +import org.tron.core.net.peer.Item; import org.tron.core.net.peer.PeerConnection; +import org.tron.core.net.peer.TronState; +import org.tron.core.net.service.adv.AdvService; +import org.tron.core.net.service.sync.SyncService; +import org.tron.protos.Protocol.Inventory.InventoryType; public class ChainInventoryMsgHandlerTest { + private ChainInventoryMsgHandler handler; + private TronNetDelegate delegate; + private SyncService sync; + private PeerConnection peer; + @BeforeClass public static void init() { Args.setParam(new String[]{}, TestConstants.TEST_CONF); @@ -29,53 +58,293 @@ public static void destroy() { Args.clearParam(); } - private ChainInventoryMsgHandler handler = new ChainInventoryMsgHandler(); - private PeerConnection peer = new PeerConnection(); - private ChainInventoryMessage msg = new ChainInventoryMessage(new ArrayList<>(), 0L); - private List blockIds = new ArrayList<>(); + @Before + public void setUp() throws Exception { + resetFixture(); + } + + private void resetFixture() throws Exception { + handler = new ChainInventoryMsgHandler(); + delegate = mock(TronNetDelegate.class); + sync = mock(SyncService.class); + ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(handler, "syncService", sync); + ReflectUtils.setFieldValue(handler, "syncFetchBatchNum", 100L); + when(delegate.getBlockLock()).thenReturn(new Object()); + when(delegate.getForkLock()).thenReturn(new Object()); + when(delegate.getHeadBlockId()).thenReturn(blockId(100)); + when(delegate.getKhaosDbHeadBlockId()).thenReturn(blockId(100)); + when(delegate.getSolidBlockId()).thenReturn(blockId(0)); + when(delegate.getGenesisBlockId()).thenReturn(blockId(0)); + when(delegate.getBlockTime(blockId(0))).thenReturn(System.currentTimeMillis() - 30_000_000L); + when(delegate.containBlock(any())).thenAnswer(invocation -> + ((BlockId) invocation.getArgument(0)).getNum() <= 100); + when(delegate.getBlockIdByNum(org.mockito.ArgumentMatchers.anyLong())) + .thenAnswer(invocation -> blockId(invocation.getArgument(0))); + peer = PeerSyncTestSupport.peer(10001); + peer.setTronState(TronState.SYNCING); + peer.setNeedSyncFromPeer(true); + peer.setNeedSyncFromUs(false); + peer.setFetchAble(true); + peer.setRemainNum(99); + peer.setBlockRcvTime(7); + request(blockId(0), blockId(50), blockId(100)); + } @Test public void testProcessMessage() throws Exception { - try { - handler.processMessage(peer, msg); - } catch (P2pException e) { - Assert.assertEquals("not send syncBlockChainMsg", e.getMessage()); - } + ChainInventoryMessage msg = new ChainInventoryMessage(Collections.emptyList(), 0L); + peer.setSyncChainRequested(null); + Assert.assertEquals("not send syncBlockChainMsg", assertRejected(msg).getMessage()); + Assert.assertEquals("not send syncBlockChainMsg", assertRejected( + new ChainInventoryMessage(Collections.singletonList(blockId(100)), 0L)).getMessage()); - peer.setSyncChainRequested(new Pair<>(new LinkedList<>(), System.currentTimeMillis())); + request(); + Assert.assertEquals("blockIds is empty", assertRejected(msg).getMessage()); - try { - handler.processMessage(peer, msg); - } catch (P2pException e) { - Assert.assertEquals("blockIds is empty", e.getMessage()); + int size = (int) NetConstants.SYNC_FETCH_BATCH_NUM + 2; + msg = new ChainInventoryMessage(range(100, size), 0L); + Assert.assertEquals("big blockIds size: " + size, assertRejected(msg).getMessage()); + + size = (int) NetConstants.SYNC_FETCH_BATCH_NUM / 100; + msg = new ChainInventoryMessage(range(100, size), 100L); + Assert.assertEquals("remain: 100, blockIds size: " + size, assertRejected(msg).getMessage()); + Assert.assertNotNull(msg.toString()); + Assert.assertNull(msg.getAnswerMessage()); + } + + @Test + public void testKnownSummaryBlockCompletesDownloadAndPreservesUploadDirection() throws Exception { + for (long height : new long[]{0, 50, 100}) { + for (boolean needSyncFromUs : new boolean[]{false, true}) { + resetFixture(); + peer.setNeedSyncFromUs(needSyncFromUs); + try { + respond(0, blockId(height)); + + assertDownloadCompleted(); + Assert.assertEquals(needSyncFromUs, peer.isNeedSyncFromUs()); + Assert.assertEquals(!needSyncFromUs, peer.isSyncFinish()); + } catch (Exception | AssertionError e) { + throw new AssertionError("height=" + height + ", needSyncFromUs=" + needSyncFromUs, e); + } + } } + } - long size = NetConstants.SYNC_FETCH_BATCH_NUM + 2; - for (int i = 0; i < size; i++) { - blockIds.add(new BlockId()); + @Test + public void testKnownResponseCompletesDownloadAfterHeadAdvances() throws Exception { + when(delegate.getHeadBlockId()).thenReturn(blockId(101)); + + respond(0, blockId(100)); + + assertDownloadCompleted(); + Assert.assertFalse(peer.isNeedSyncFromUs()); + } + + @Test + public void testSingleBlockReplyFromExistingPeerKeepsInventoryFlow() throws Exception { + // The remote peer still believes it is ahead based on Hello. Our summary now includes + // newer blocks learned from another connection; replying does not start a remote download. + PeerConnection remotePeer = PeerSyncTestSupport.peer(10002); + remotePeer.setNeedSyncFromPeer(false); + remotePeer.setNeedSyncFromUs(true); + TronNetDelegate remoteDelegate = mock(TronNetDelegate.class); + when(remoteDelegate.getHeadBlockId()).thenReturn(blockId(50)); + when(remoteDelegate.containBlockInMainChain(any())).thenAnswer(invocation -> + ((BlockId) invocation.getArgument(0)).getNum() <= 50); + SyncBlockChainMsgHandler remoteHandler = new SyncBlockChainMsgHandler(); + ReflectUtils.setFieldValue(remoteHandler, "tronNetDelegate", remoteDelegate); + + remoteHandler.processMessage(remotePeer, + new SyncBlockChainMessage(new ArrayList<>(peer.getSyncChainRequested().getKey()))); + + ArgumentCaptor reply = ArgumentCaptor.forClass(Message.class); + verify(remotePeer).sendMessage(reply.capture()); + Assert.assertTrue(reply.getValue() instanceof ChainInventoryMessage); + ChainInventoryMessage response = new ChainInventoryMessage(reply.getValue().getData()); + Assert.assertEquals(Collections.singletonList(blockId(50)), response.getBlockIds()); + Assert.assertEquals(Long.valueOf(0), response.getRemainNum()); + handler.processMessage(peer, response); + + assertDownloadCompleted(); + Assert.assertTrue(peer.isSyncFinish()); + Assert.assertTrue(remotePeer.isSyncFinish()); + verify(remotePeer, never()).disconnect(any()); + + InventoryMsgHandler inventoryHandler = new InventoryMsgHandler(); + AdvService adv = mock(AdvService.class); + ReflectUtils.setFieldValue(inventoryHandler, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(inventoryHandler, "advService", adv); + ReflectUtils.setFieldValue(inventoryHandler, "transactionsMsgHandler", + mock(TransactionsMsgHandler.class)); + for (InventoryType type : Arrays.asList(InventoryType.BLOCK, InventoryType.TRX)) { + Item item = new Item(blockId(101), type); + inventoryHandler.processMessage(peer, + new InventoryMessage(Collections.singletonList(item.getHash()), type)); + Assert.assertNotNull(peer.getAdvInvReceive().getIfPresent(item)); + verify(adv).addInv(item); } - msg = new ChainInventoryMessage(blockIds, 0L); + verify(peer, never()).disconnect(any()); + } + @Test + public void testPeerBehindCanStartAnotherDownloadLater() throws Exception { + respond(0, blockId(50)); + SyncService service = new SyncService(); + ReflectUtils.setFieldValue(service, "tronNetDelegate", delegate); try { - handler.processMessage(peer, msg); - } catch (P2pException e) { - Assert.assertEquals(e.getMessage(), "big blockIds size: " + size); - } + service.startSync(peer); - blockIds.clear(); - size = NetConstants.SYNC_FETCH_BATCH_NUM / 100; - for (int i = 0; i < size; i++) { - blockIds.add(new BlockId()); + Assert.assertEquals(TronState.SYNCING, peer.getTronState()); + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertNotNull(peer.getSyncChainRequested()); + Assert.assertEquals(blockId(100), peer.getSyncChainRequested().getKey().peekLast()); + verify(peer).sendMessage(any(SyncBlockChainMessage.class)); + verify(peer, never()).disconnect(any()); + } finally { + service.close(); } - msg = new ChainInventoryMessage(blockIds, 100L); + } - try { - handler.processMessage(peer, msg); - } catch (P2pException e) { - Assert.assertEquals(e.getMessage(), "remain: 100, blockIds size: " + size); + @Test + public void testUnknownQueuedTailStillRequiresBlockDownload() throws Exception { + request(blockId(0), blockId(100), blockId(102)); + peer.getSyncBlockToFetch().addAll(Arrays.asList(blockId(101), blockId(102))); + + respond(0, blockId(102)); + + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertEquals(TronState.SYNCING, peer.getTronState()); + Assert.assertEquals(2, peer.getSyncBlockToFetch().size()); + Assert.assertTrue(peer.isFetchAble()); + Assert.assertNull(peer.getSyncChainRequested()); + Assert.assertEquals(7, peer.getBlockRcvTime()); + verify(sync).setFetchFlag(true); + verify(sync, never()).syncNext(any()); + } + + @Test + public void testFinalMultiBlockResponseSchedulesFetch() throws Exception { + respond(0, blockId(100), blockId(101), blockId(102)); + + Assert.assertEquals(Arrays.asList(blockId(101), blockId(102)), + new ArrayList<>(peer.getSyncBlockToFetch())); + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertEquals(TronState.SYNCING, peer.getTronState()); + Assert.assertEquals(0, peer.getRemainNum()); + Assert.assertEquals(7, peer.getBlockRcvTime()); + verify(sync).setFetchFlag(true); + verify(sync, never()).syncNext(any()); + } + + @Test + public void testKnownMultiBlockResponseRequestsNextSummary() throws Exception { + respond(0, blockId(50), blockId(51), blockId(52)); + + Assert.assertTrue(peer.getSyncBlockToFetch().isEmpty()); + Assert.assertEquals(blockId(52), peer.getBlockBothHave()); + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertEquals(7, peer.getBlockRcvTime()); + verify(sync).syncNext(peer); + } + + @Test + public void testPagedResponsePreservesPendingBlocks() throws Exception { + List firstPage = range(100, (int) NetConstants.SYNC_FETCH_BATCH_NUM + 1); + handler.processMessage(peer, new ChainInventoryMessage(firstPage, 2L)); + Assert.assertEquals(NetConstants.SYNC_FETCH_BATCH_NUM, peer.getSyncBlockToFetch().size()); + Assert.assertEquals(2, peer.getRemainNum()); + verify(sync).setFetchFlag(true); + + BlockId tail = firstPage.get(firstPage.size() - 1); + request(blockId(0), blockId(100), tail); + respond(0, tail, blockId(tail.getNum() + 1), blockId(tail.getNum() + 2)); + + Assert.assertEquals(NetConstants.SYNC_FETCH_BATCH_NUM + 2, + peer.getSyncBlockToFetch().size()); + Assert.assertEquals(blockId(101), peer.getSyncBlockToFetch().peekFirst()); + Assert.assertEquals(blockId(tail.getNum() + 2), peer.getSyncBlockToFetch().peekLast()); + Assert.assertEquals(0, peer.getRemainNum()); + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertEquals(7, peer.getBlockRcvTime()); + } + + @Test + public void testSingleBlockWithRemainingBlocksIsRejected() throws Exception { + assertRejected(new ChainInventoryMessage(Arrays.asList(blockId(50)), 1L)); + } + + @Test + public void testInvalidRemainingCountsAreRejectedForFullBatch() throws Exception { + for (long remain : new long[]{-1, Long.MAX_VALUE, 100_000}) { + resetFixture(); + assertRejected(new ChainInventoryMessage(range(100, + (int) NetConstants.SYNC_FETCH_BATCH_NUM), remain)); } - Assert.assertNotNull(msg.toString()); - Assert.assertNull(msg.getAnswerMessage()); } + @Test + public void testKnownBlockOutsideSummaryIsRejected() throws Exception { + assertRejected(new ChainInventoryMessage(Arrays.asList(blockId(75)), 0L)); + } + + @Test + public void testSameHeightDifferentHashIsRejected() throws Exception { + byte[] hash = new byte[Sha256Hash.LENGTH]; + hash[31] = 1; + assertRejected(new ChainInventoryMessage( + Arrays.asList(new BlockId(Sha256Hash.wrap(hash), 100)), 0L)); + } + + @Test + public void testDiscontinuousResponseIsRejected() throws Exception { + assertRejected(new ChainInventoryMessage(Arrays.asList(blockId(100), blockId(102)), 0L)); + } + + private void request(BlockId... ids) { + peer.setSyncChainRequested(new Pair<>(new LinkedList<>(Arrays.asList(ids)), + System.currentTimeMillis())); + } + + private void respond(long remain, BlockId... ids) throws Exception { + handler.processMessage(peer, new ChainInventoryMessage(Arrays.asList(ids), remain)); + } + + private void assertDownloadCompleted() { + Assert.assertFalse(peer.isNeedSyncFromPeer()); + Assert.assertEquals(TronState.SYNC_COMPLETED, peer.getTronState()); + Assert.assertNull(peer.getSyncChainRequested()); + Assert.assertFalse(peer.isFetchAble()); + Assert.assertEquals(0, peer.getRemainNum()); + Assert.assertEquals(7, peer.getBlockRcvTime()); + verify(sync, never()).syncNext(any()); + verify(sync, never()).setFetchFlag(true); + } + + private P2pException assertRejected(ChainInventoryMessage message) { + String context = message.toString(); + Pair, Long> requested = peer.getSyncChainRequested(); + P2pException exception = Assert.assertThrows(context, P2pException.class, + () -> handler.processMessage(peer, message)); + Assert.assertEquals(context, TypeEnum.BAD_MESSAGE, exception.getType()); + Assert.assertSame(context, requested, peer.getSyncChainRequested()); + Assert.assertTrue(context, peer.isNeedSyncFromPeer()); + Assert.assertFalse(context, peer.isNeedSyncFromUs()); + Assert.assertTrue(context, peer.isFetchAble()); + Assert.assertEquals(context, TronState.SYNCING, peer.getTronState()); + Assert.assertEquals(context, 99, peer.getRemainNum()); + Assert.assertEquals(context, 7, peer.getBlockRcvTime()); + verify(sync, never()).syncNext(any()); + verify(sync, never()).setFetchFlag(true); + return exception; + } + + private List range(long first, int size) { + List ids = new ArrayList<>(); + for (int i = 0; i < size; i++) { + ids.add(blockId(first + i)); + } + return ids; + } } diff --git a/framework/src/test/java/org/tron/core/net/peer/PeerStatusCheckMockTest.java b/framework/src/test/java/org/tron/core/net/peer/PeerStatusCheckMockTest.java index d2ee4be5b87..bd38cde5e4e 100644 --- a/framework/src/test/java/org/tron/core/net/peer/PeerStatusCheckMockTest.java +++ b/framework/src/test/java/org/tron/core/net/peer/PeerStatusCheckMockTest.java @@ -1,27 +1,85 @@ package org.tron.core.net.peer; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.tron.core.net.PeerSyncTestSupport.blockId; +import java.util.Arrays; +import java.util.Collections; +import java.util.Deque; +import java.util.LinkedList; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import org.junit.After; +import org.junit.Assert; +import org.junit.Before; import org.junit.Test; import org.mockito.Mockito; +import org.tron.common.utils.Pair; import org.tron.common.utils.ReflectUtils; +import org.tron.core.capsule.BlockCapsule.BlockId; +import org.tron.core.config.Parameter.NetConstants; +import org.tron.core.net.P2pEventHandlerImpl; +import org.tron.core.net.PeerSyncTestSupport; +import org.tron.core.net.TronNetDelegate; +import org.tron.core.net.message.TronMessage; +import org.tron.core.net.message.keepalive.PingMessage; +import org.tron.core.net.message.keepalive.PongMessage; +import org.tron.core.net.message.sync.ChainInventoryMessage; +import org.tron.core.net.message.sync.SyncBlockChainMessage; +import org.tron.core.net.messagehandler.ChainInventoryMsgHandler; +import org.tron.core.net.messagehandler.SyncBlockChainMsgHandler; +import org.tron.core.net.service.keepalive.KeepAliveService; +import org.tron.core.net.service.sync.SyncService; +import org.tron.protos.Protocol.Inventory.InventoryType; +import org.tron.protos.Protocol.ReasonCode; public class PeerStatusCheckMockTest { + + private PeerStatusCheck check; + private TronNetDelegate delegate; + private PeerConnection peer; + + @Before + public void setUp() throws Exception { + check = new PeerStatusCheck(); + delegate = mock(TronNetDelegate.class); + ReflectUtils.setFieldValue(check, "tronNetDelegate", delegate); + peer = PeerSyncTestSupport.peer(11001); + peer.setNeedSyncFromPeer(false); + peer.setNeedSyncFromUs(false); + peer.setLastInteractiveTime(1); + when(delegate.getActivePeer()).thenReturn(Collections.singletonList(peer)); + when(delegate.getHeadBlockId()).thenReturn(blockId(20)); + when(delegate.getKhaosDbHeadBlockId()).thenReturn(blockId(20)); + when(delegate.getSolidBlockId()).thenReturn(blockId(0)); + when(delegate.getBlockTime(blockId(0))).thenReturn(System.currentTimeMillis() - 60_000); + when(delegate.getForkLock()).thenReturn(new Object()); + when(delegate.getBlockIdByNum(anyLong())) + .thenAnswer(invocation -> blockId(invocation.getArgument(0))); + request(peer, System.currentTimeMillis() - 6_000); + } + @After - public void clearMocks() { - Mockito.framework().clearInlineMocks(); + public void tearDown() { + try { + check.close(); + } finally { + Mockito.framework().clearInlineMocks(); + } } @Test public void testInitException() { - PeerStatusCheck peerStatusCheck = spy(new PeerStatusCheck()); + PeerStatusCheck peerStatusCheck = spy(check); ScheduledExecutorService executor = mock(ScheduledExecutorService.class); ReflectUtils.setFieldValue(peerStatusCheck, "peerStatusCheckExecutor", executor); doThrow(new RuntimeException("test exception")).when(peerStatusCheck).statusCheck(); @@ -41,4 +99,153 @@ public void testInitException() { Mockito.verify(peerStatusCheck).statusCheck(); } + @Test + public void testExpiredSummaryDisconnectsOnlyResponsiblePeer() { + PeerConnection other = PeerSyncTestSupport.peer(11002); + other.setNeedSyncFromPeer(false); + other.setNeedSyncFromUs(false); + request(other, System.currentTimeMillis()); + when(delegate.getActivePeer()).thenReturn(Arrays.asList(peer, other)); + + check.statusCheck(); + + verify(peer).disconnect(ReasonCode.TIME_OUT); + verify(other, never()).disconnect(any()); + } + + @Test + public void testRecentSyncProgressAndInteractionDoNotExtendDeadline() { + Pair, Long> requested = peer.getSyncChainRequested(); + peer.setNeedSyncFromPeer(true); + peer.setBlockBothHave(blockId(20)); + peer.setLastInteractiveTime(System.currentTimeMillis()); + peer.setBlockRcvTime(System.currentTimeMillis()); + + check.statusCheck(); + + Assert.assertSame(requested, peer.getSyncChainRequested()); + verify(peer).disconnect(ReasonCode.TIME_OUT); + } + + @Test + public void testPingAndPongDoNotExtendDeadline() { + Pair, Long> requested = peer.getSyncChainRequested(); + P2pEventHandlerImpl events = new P2pEventHandlerImpl(); + ReflectUtils.setFieldValue(events, "keepAliveService", new KeepAliveService()); + + dispatch(events, new PingMessage()); + dispatch(events, new PongMessage()); + + verify(peer).sendMessage(any(PongMessage.class)); + verify(peer, never()).disconnect(any()); + Assert.assertSame(requested, peer.getSyncChainRequested()); + Assert.assertEquals(1, peer.getLastInteractiveTime()); + check.statusCheck(); + verify(peer).disconnect(ReasonCode.TIME_OUT); + } + + @Test + public void testValidIncomingSyncRequestDoesNotExtendOutgoingDeadline() throws Exception { + Pair, Long> requested = peer.getSyncChainRequested(); + SyncBlockChainMsgHandler handler = new SyncBlockChainMsgHandler(); + ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate); + when(delegate.containBlockInMainChain(any())).thenReturn(true); + P2pEventHandlerImpl events = new P2pEventHandlerImpl(); + ReflectUtils.setFieldValue(events, "syncBlockChainMsgHandler", handler); + + dispatch(events, new SyncBlockChainMessage( + new LinkedList<>(Arrays.asList(blockId(0), blockId(20))))); + + verify(peer).sendMessage(any(ChainInventoryMessage.class)); + verify(peer, never()).disconnect(any()); + Assert.assertTrue(peer.getLastInteractiveTime() > 1); + Assert.assertSame(requested, peer.getSyncChainRequested()); + check.statusCheck(); + verify(peer).disconnect(ReasonCode.TIME_OUT); + } + + @Test + public void testRequestWithinFiveSecondsDoesNotTimeOut() { + request(peer, System.currentTimeMillis() - 1_000); + + check.statusCheck(); + + verify(peer, never()).disconnect(any()); + } + + @Test + public void testCompletedResponseClearsDeadlineWithoutContribution() throws Exception { + ChainInventoryMsgHandler handler = new ChainInventoryMsgHandler(); + ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(handler, "syncService", mock(SyncService.class)); + when(delegate.containBlock(blockId(20))).thenReturn(true); + peer.setNeedSyncFromPeer(true); + peer.setBlockRcvTime(7); + + handler.processMessage(peer, new ChainInventoryMessage(Collections.singletonList(blockId(20)), + 0L)); + check.statusCheck(); + + Assert.assertNull(peer.getSyncChainRequested()); + Assert.assertTrue(peer.isSyncFinish()); + Assert.assertEquals(7, peer.getBlockRcvTime()); + verify(peer, never()).disconnect(any()); + } + + @Test + public void testNewRequestUsesItsOwnDeadline() { + peer.setSyncChainRequested(null); + check.statusCheck(); + request(peer, System.currentTimeMillis()); + check.statusCheck(); + + verify(peer, never()).disconnect(any()); + } + + @Test + public void testRepeatedSyncNextDoesNotReplacePendingRequest() { + SyncService service = new SyncService(); + ReflectUtils.setFieldValue(service, "tronNetDelegate", delegate); + peer.setSyncChainRequested(null); + try { + service.syncNext(peer); + Pair, Long> requested = peer.getSyncChainRequested(); + Assert.assertNotNull(requested); + Assert.assertEquals(blockId(20), requested.getKey().peekLast()); + + service.syncNext(peer); + + Assert.assertSame(requested, peer.getSyncChainRequested()); + verify(peer, times(1)).sendMessage(any(SyncBlockChainMessage.class)); + verify(peer, never()).disconnect(any()); + } finally { + service.close(); + } + } + + @Test + public void testExistingInventoryAndSyncBlockTimeoutsRemainEffective() { + peer.setSyncChainRequested(null); + peer.getAdvInvRequest().put(new Item(blockId(21), InventoryType.BLOCK), + System.currentTimeMillis() - NetConstants.ADV_TIME_OUT - 1_000); + PeerConnection other = PeerSyncTestSupport.peer(11002); + other.setNeedSyncFromPeer(false); + other.getSyncBlockRequested().put(blockId(21), System.currentTimeMillis() - 6_000); + when(delegate.getActivePeer()).thenReturn(Arrays.asList(peer, other)); + + check.statusCheck(); + + verify(peer).disconnect(ReasonCode.TIME_OUT); + verify(other).disconnect(ReasonCode.TIME_OUT); + } + + private void request(PeerConnection target, long time) { + target.setSyncChainRequested(new Pair<>( + new LinkedList<>(Arrays.asList(blockId(0), blockId(20))), time)); + } + + private void dispatch(P2pEventHandlerImpl events, TronMessage message) { + ReflectUtils.invokeMethod(events, "processMessage", + new Class[]{PeerConnection.class, byte[].class}, peer, message.getSendBytes()); + } } diff --git a/framework/src/test/java/org/tron/core/net/peer/PeerSyncChainTimeoutTest.java b/framework/src/test/java/org/tron/core/net/peer/PeerSyncChainTimeoutTest.java deleted file mode 100644 index c8308f019b1..00000000000 --- a/framework/src/test/java/org/tron/core/net/peer/PeerSyncChainTimeoutTest.java +++ /dev/null @@ -1,219 +0,0 @@ -package org.tron.core.net.peer; - -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyLong; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; -import static org.tron.core.net.PeerSyncTestSupport.blockId; - -import java.util.Arrays; -import java.util.Collections; -import java.util.Deque; -import java.util.LinkedList; -import org.junit.After; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Test; -import org.tron.common.utils.Pair; -import org.tron.common.utils.ReflectUtils; -import org.tron.core.capsule.BlockCapsule.BlockId; -import org.tron.core.config.Parameter.NetConstants; -import org.tron.core.net.P2pEventHandlerImpl; -import org.tron.core.net.PeerSyncTestSupport; -import org.tron.core.net.TronNetDelegate; -import org.tron.core.net.message.TronMessage; -import org.tron.core.net.message.keepalive.PingMessage; -import org.tron.core.net.message.keepalive.PongMessage; -import org.tron.core.net.message.sync.ChainInventoryMessage; -import org.tron.core.net.message.sync.SyncBlockChainMessage; -import org.tron.core.net.messagehandler.ChainInventoryMsgHandler; -import org.tron.core.net.messagehandler.SyncBlockChainMsgHandler; -import org.tron.core.net.service.keepalive.KeepAliveService; -import org.tron.core.net.service.sync.SyncService; -import org.tron.protos.Protocol.Inventory.InventoryType; -import org.tron.protos.Protocol.ReasonCode; - -public class PeerSyncChainTimeoutTest { - - private PeerStatusCheck check; - private TronNetDelegate delegate; - private PeerConnection peer; - - @Before - public void setUp() throws Exception { - check = new PeerStatusCheck(); - delegate = mock(TronNetDelegate.class); - ReflectUtils.setFieldValue(check, "tronNetDelegate", delegate); - peer = PeerSyncTestSupport.peer(11001); - peer.setNeedSyncFromPeer(false); - peer.setNeedSyncFromUs(false); - peer.setLastInteractiveTime(1); - when(delegate.getActivePeer()).thenReturn(Collections.singletonList(peer)); - when(delegate.getHeadBlockId()).thenReturn(blockId(20)); - when(delegate.getKhaosDbHeadBlockId()).thenReturn(blockId(20)); - when(delegate.getSolidBlockId()).thenReturn(blockId(0)); - when(delegate.getBlockTime(blockId(0))).thenReturn(System.currentTimeMillis() - 60_000); - when(delegate.getForkLock()).thenReturn(new Object()); - when(delegate.getBlockIdByNum(anyLong())) - .thenAnswer(invocation -> blockId(invocation.getArgument(0))); - request(peer, System.currentTimeMillis() - 6_000); - } - - @After - public void tearDown() { - check.close(); - } - - @Test - public void testExpiredSummaryDisconnectsOnlyResponsiblePeer() { - PeerConnection other = PeerSyncTestSupport.peer(11002); - other.setNeedSyncFromPeer(false); - other.setNeedSyncFromUs(false); - request(other, System.currentTimeMillis()); - when(delegate.getActivePeer()).thenReturn(Arrays.asList(peer, other)); - - check.statusCheck(); - - verify(peer).disconnect(ReasonCode.TIME_OUT); - verify(other, never()).disconnect(any()); - } - - @Test - public void testRecentSyncProgressAndInteractionDoNotExtendDeadline() { - Pair, Long> requested = peer.getSyncChainRequested(); - peer.setNeedSyncFromPeer(true); - peer.setBlockBothHave(blockId(20)); - peer.setLastInteractiveTime(System.currentTimeMillis()); - peer.setBlockRcvTime(System.currentTimeMillis()); - - check.statusCheck(); - - Assert.assertSame(requested, peer.getSyncChainRequested()); - verify(peer).disconnect(ReasonCode.TIME_OUT); - } - - @Test - public void testPingAndPongDoNotExtendDeadline() { - Pair, Long> requested = peer.getSyncChainRequested(); - P2pEventHandlerImpl events = new P2pEventHandlerImpl(); - ReflectUtils.setFieldValue(events, "keepAliveService", new KeepAliveService()); - - dispatch(events, new PingMessage()); - dispatch(events, new PongMessage()); - - verify(peer).sendMessage(any(PongMessage.class)); - verify(peer, never()).disconnect(any()); - Assert.assertSame(requested, peer.getSyncChainRequested()); - Assert.assertEquals(1, peer.getLastInteractiveTime()); - check.statusCheck(); - verify(peer).disconnect(ReasonCode.TIME_OUT); - } - - @Test - public void testValidIncomingSyncRequestDoesNotExtendOutgoingDeadline() throws Exception { - Pair, Long> requested = peer.getSyncChainRequested(); - SyncBlockChainMsgHandler handler = new SyncBlockChainMsgHandler(); - ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate); - when(delegate.containBlockInMainChain(any())).thenReturn(true); - P2pEventHandlerImpl events = new P2pEventHandlerImpl(); - ReflectUtils.setFieldValue(events, "syncBlockChainMsgHandler", handler); - - dispatch(events, new SyncBlockChainMessage( - new LinkedList<>(Arrays.asList(blockId(0), blockId(20))))); - - verify(peer).sendMessage(any(ChainInventoryMessage.class)); - verify(peer, never()).disconnect(any()); - Assert.assertTrue(peer.getLastInteractiveTime() > 1); - Assert.assertSame(requested, peer.getSyncChainRequested()); - check.statusCheck(); - verify(peer).disconnect(ReasonCode.TIME_OUT); - } - - @Test - public void testRequestWithinFiveSecondsDoesNotTimeOut() { - request(peer, System.currentTimeMillis() - 1_000); - - check.statusCheck(); - - verify(peer, never()).disconnect(any()); - } - - @Test - public void testCompletedResponseClearsDeadlineWithoutContribution() throws Exception { - ChainInventoryMsgHandler handler = new ChainInventoryMsgHandler(); - ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate); - ReflectUtils.setFieldValue(handler, "syncService", mock(SyncService.class)); - when(delegate.containBlock(blockId(20))).thenReturn(true); - peer.setNeedSyncFromPeer(true); - peer.setBlockRcvTime(7); - - handler.processMessage(peer, new ChainInventoryMessage(Collections.singletonList(blockId(20)), - 0L)); - check.statusCheck(); - - Assert.assertNull(peer.getSyncChainRequested()); - Assert.assertTrue(peer.isSyncFinish()); - Assert.assertEquals(7, peer.getBlockRcvTime()); - verify(peer, never()).disconnect(any()); - } - - @Test - public void testNewRequestUsesItsOwnDeadline() { - peer.setSyncChainRequested(null); - check.statusCheck(); - request(peer, System.currentTimeMillis()); - check.statusCheck(); - - verify(peer, never()).disconnect(any()); - } - - @Test - public void testRepeatedSyncNextDoesNotReplacePendingRequest() { - SyncService service = new SyncService(); - ReflectUtils.setFieldValue(service, "tronNetDelegate", delegate); - peer.setSyncChainRequested(null); - try { - service.syncNext(peer); - Pair, Long> requested = peer.getSyncChainRequested(); - Assert.assertNotNull(requested); - Assert.assertEquals(blockId(20), requested.getKey().peekLast()); - - service.syncNext(peer); - - Assert.assertSame(requested, peer.getSyncChainRequested()); - verify(peer, times(1)).sendMessage(any(SyncBlockChainMessage.class)); - verify(peer, never()).disconnect(any()); - } finally { - service.close(); - } - } - - @Test - public void testExistingInventoryAndSyncBlockTimeoutsRemainEffective() { - peer.setSyncChainRequested(null); - peer.getAdvInvRequest().put(new Item(blockId(21), InventoryType.BLOCK), - System.currentTimeMillis() - NetConstants.ADV_TIME_OUT - 1_000); - PeerConnection other = PeerSyncTestSupport.peer(11002); - other.setNeedSyncFromPeer(false); - other.getSyncBlockRequested().put(blockId(21), System.currentTimeMillis() - 6_000); - when(delegate.getActivePeer()).thenReturn(Arrays.asList(peer, other)); - - check.statusCheck(); - - verify(peer).disconnect(ReasonCode.TIME_OUT); - verify(other).disconnect(ReasonCode.TIME_OUT); - } - - private void request(PeerConnection target, long time) { - target.setSyncChainRequested(new Pair<>( - new LinkedList<>(Arrays.asList(blockId(0), blockId(20))), time)); - } - - private void dispatch(P2pEventHandlerImpl events, TronMessage message) { - ReflectUtils.invokeMethod(events, "processMessage", - new Class[]{PeerConnection.class, byte[].class}, peer, message.getSendBytes()); - } -} From 4e46755acb88b1dac56419d153e1fa8a1b61344c Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Fri, 2 Oct 2026 11:51:40 +0800 Subject: [PATCH 4/4] add more check on ChainInventoryMsgHandler --- .../ChainInventoryMsgHandler.java | 24 ++++-- .../tron/core/net/PeerSyncTestSupport.java | 9 +++ .../ChainInventoryMsgHandlerTest.java | 75 ++++++++++++++++++- .../net/peer/PeerStatusCheckMockTest.java | 2 + 4 files changed, 102 insertions(+), 8 deletions(-) diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java index f3ac1f02fdd..d7813316732 100644 --- a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java +++ b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java @@ -19,6 +19,7 @@ import org.tron.core.exception.P2pException.TypeEnum; import org.tron.core.net.TronNetDelegate; import org.tron.core.net.message.TronMessage; +import org.tron.core.net.message.handshake.HelloMessage; import org.tron.core.net.message.sync.ChainInventoryMessage; import org.tron.core.net.peer.PeerConnection; import org.tron.core.net.peer.TronState; @@ -42,7 +43,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep ChainInventoryMessage chainInventoryMessage = (ChainInventoryMessage) msg; Pair, Long> requested = peer.getSyncChainRequested(); - check(requested, chainInventoryMessage); + check(peer, requested, chainInventoryMessage); peer.setFetchAble(false); @@ -55,9 +56,6 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep if (blockIdWeGet.size() == 1 && tronNetDelegate.containBlock(blockIdWeGet.peek())) { peer.setRemainNum(0); peer.setTronState(TronState.SYNC_COMPLETED); - // This trusts the peer's claim that no blocks remain. A peer can echo an earlier - // known summary block while withholding newer blocks, so ending this download - // does not prove that we have caught up with the peer's actual chain. todo fix peer.setNeedSyncFromPeer(false); return; } @@ -104,12 +102,17 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep } } - private void check(Pair, Long> requested, ChainInventoryMessage msg) - throws P2pException { + private void check(PeerConnection peer, Pair, Long> requested, + ChainInventoryMessage msg) throws P2pException { if (requested == null) { throw new P2pException(TypeEnum.BAD_MESSAGE, "not send syncBlockChainMsg"); } + HelloMessage hello = peer.getHelloMessageReceive(); + if (hello == null) { + throw new P2pException(TypeEnum.BAD_MESSAGE, "hello message not received"); + } + List blockIds = msg.getBlockIds(); if (CollectionUtils.isEmpty(blockIds)) { throw new P2pException(TypeEnum.BAD_MESSAGE, "blockIds is empty"); @@ -150,6 +153,15 @@ private void check(Pair, Long> requested, ChainInventoryMessage m + msg.getRemainNum() + " > futureMaxNum: " + maxFutureNum); } } + + if (blockIds.size() == 1) { + long lastNum = blockIds.get(0).getNum(); + long helloHeadNum = hello.getHeadBlockId().getNum(); + if (lastNum < helloHeadNum) { + throw new P2pException(TypeEnum.SYNC_FAILED, + "Single-block response height " + lastNum + " is below hello head " + helloHeadNum); + } + } } } diff --git a/framework/src/test/java/org/tron/core/net/PeerSyncTestSupport.java b/framework/src/test/java/org/tron/core/net/PeerSyncTestSupport.java index b3009b67142..73f053664e3 100644 --- a/framework/src/test/java/org/tron/core/net/PeerSyncTestSupport.java +++ b/framework/src/test/java/org/tron/core/net/PeerSyncTestSupport.java @@ -10,8 +10,10 @@ import org.tron.common.utils.ReflectUtils; import org.tron.common.utils.Sha256Hash; import org.tron.core.capsule.BlockCapsule.BlockId; +import org.tron.core.net.message.handshake.HelloMessage; import org.tron.core.net.peer.PeerConnection; import org.tron.p2p.connection.Channel; +import org.tron.protos.Protocol; public final class PeerSyncTestSupport { @@ -22,6 +24,13 @@ public static BlockId blockId(long number) { return new BlockId(Sha256Hash.ZERO_HASH, number); } + public static HelloMessage helloMessage(long headNum) throws Exception { + return new HelloMessage(Protocol.HelloMessage.newBuilder() + .setHeadBlockId(Protocol.HelloMessage.BlockId.newBuilder() + .setHash(blockId(headNum).getByteString()).setNumber(headNum)) + .build().toByteArray()); + } + public static PeerConnection peer(int port) { PeerConnection peer = spy(new PeerConnection()); Channel channel = mock(Channel.class); diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java index 66a60304718..87b8856d3f0 100644 --- a/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java +++ b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java @@ -6,6 +6,7 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.tron.core.net.PeerSyncTestSupport.blockId; +import static org.tron.core.net.PeerSyncTestSupport.helloMessage; import java.util.ArrayList; import java.util.Arrays; @@ -29,6 +30,7 @@ import org.tron.core.config.args.Args; import org.tron.core.exception.P2pException; import org.tron.core.exception.P2pException.TypeEnum; +import org.tron.core.net.P2pEventHandlerImpl; import org.tron.core.net.PeerSyncTestSupport; import org.tron.core.net.TronNetDelegate; import org.tron.core.net.message.adv.InventoryMessage; @@ -40,6 +42,7 @@ import org.tron.core.net.service.adv.AdvService; import org.tron.core.net.service.sync.SyncService; import org.tron.protos.Protocol.Inventory.InventoryType; +import org.tron.protos.Protocol.ReasonCode; public class ChainInventoryMsgHandlerTest { @@ -82,6 +85,7 @@ private void resetFixture() throws Exception { when(delegate.getBlockIdByNum(org.mockito.ArgumentMatchers.anyLong())) .thenAnswer(invocation -> blockId(invocation.getArgument(0))); peer = PeerSyncTestSupport.peer(10001); + peer.setHelloMessageReceive(helloMessage(100)); peer.setTronState(TronState.SYNCING); peer.setNeedSyncFromPeer(true); peer.setNeedSyncFromUs(false); @@ -118,6 +122,7 @@ public void testKnownSummaryBlockCompletesDownloadAndPreservesUploadDirection() for (long height : new long[]{0, 50, 100}) { for (boolean needSyncFromUs : new boolean[]{false, true}) { resetFixture(); + peer.setHelloMessageReceive(helloMessage(height)); peer.setNeedSyncFromUs(needSyncFromUs); try { respond(0, blockId(height)); @@ -132,6 +137,63 @@ public void testKnownSummaryBlockCompletesDownloadAndPreservesUploadDirection() } } + @Test + public void testSingleBlockBelowHelloHeadIsRejected() throws Exception { + for (long height : new long[]{50, 100, 199}) { + for (boolean needSyncFromUs : new boolean[]{false, true}) { + resetFixture(); + peer.setHelloMessageReceive(helloMessage(200)); + peer.setNeedSyncFromUs(needSyncFromUs); + request(blockId(0), blockId(50), blockId(100), blockId(199)); + + assertRejected(new ChainInventoryMessage(Collections.singletonList(blockId(height)), 0L), + TypeEnum.SYNC_FAILED); + Assert.assertFalse(peer.isSyncFinish()); + } + } + } + + @Test + public void testSingleBlockBelowHelloHeadDisconnectsWithSyncFail() throws Exception { + peer.setHelloMessageReceive(helloMessage(200)); + Pair, Long> requested = peer.getSyncChainRequested(); + P2pEventHandlerImpl events = new P2pEventHandlerImpl(); + ReflectUtils.setFieldValue(events, "chainInventoryMsgHandler", handler); + ChainInventoryMessage response = new ChainInventoryMessage( + Collections.singletonList(blockId(100)), 0L); + + ReflectUtils.invokeMethod(events, "processMessage", + new Class[]{PeerConnection.class, byte[].class}, peer, response.getSendBytes()); + + verify(peer).disconnect(ReasonCode.SYNC_FAIL); + verify(peer, never()).disconnect(ReasonCode.BAD_PROTOCOL); + Assert.assertSame(requested, peer.getSyncChainRequested()); + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertEquals(TronState.SYNCING, peer.getTronState()); + verify(sync, never()).syncNext(any()); + verify(sync, never()).setFetchFlag(true); + } + + @Test + public void testSingleBlockAboveHelloHeadCompletesDownload() throws Exception { + peer.setHelloMessageReceive(helloMessage(99)); + + respond(0, blockId(100)); + + assertDownloadCompleted(); + Assert.assertTrue(peer.isSyncFinish()); + } + + @Test + public void testResponseWithoutHelloIsRejected() { + peer.setHelloMessageReceive(null); + + P2pException exception = assertRejected(new ChainInventoryMessage( + Collections.singletonList(blockId(100)), 0L)); + + Assert.assertEquals("hello message not received", exception.getMessage()); + } + @Test public void testKnownResponseCompletesDownloadAfterHeadAdvances() throws Exception { when(delegate.getHeadBlockId()).thenReturn(blockId(101)); @@ -146,6 +208,7 @@ public void testKnownResponseCompletesDownloadAfterHeadAdvances() throws Excepti public void testSingleBlockReplyFromExistingPeerKeepsInventoryFlow() throws Exception { // The remote peer still believes it is ahead based on Hello. Our summary now includes // newer blocks learned from another connection; replying does not start a remote download. + peer.setHelloMessageReceive(helloMessage(50)); PeerConnection remotePeer = PeerSyncTestSupport.peer(10002); remotePeer.setNeedSyncFromPeer(false); remotePeer.setNeedSyncFromUs(true); @@ -190,6 +253,7 @@ public void testSingleBlockReplyFromExistingPeerKeepsInventoryFlow() throws Exce @Test public void testPeerBehindCanStartAnotherDownloadLater() throws Exception { + peer.setHelloMessageReceive(helloMessage(50)); respond(0, blockId(50)); SyncService service = new SyncService(); ReflectUtils.setFieldValue(service, "tronNetDelegate", delegate); @@ -323,14 +387,21 @@ private void assertDownloadCompleted() { } private P2pException assertRejected(ChainInventoryMessage message) { + return assertRejected(message, TypeEnum.BAD_MESSAGE); + } + + private P2pException assertRejected(ChainInventoryMessage message, TypeEnum expectedType) { String context = message.toString(); Pair, Long> requested = peer.getSyncChainRequested(); + boolean needSyncFromUs = peer.isNeedSyncFromUs(); + long blockBothHaveUpdateTime = peer.getBlockBothHaveUpdateTime(); P2pException exception = Assert.assertThrows(context, P2pException.class, () -> handler.processMessage(peer, message)); - Assert.assertEquals(context, TypeEnum.BAD_MESSAGE, exception.getType()); + Assert.assertEquals(context, expectedType, exception.getType()); Assert.assertSame(context, requested, peer.getSyncChainRequested()); Assert.assertTrue(context, peer.isNeedSyncFromPeer()); - Assert.assertFalse(context, peer.isNeedSyncFromUs()); + Assert.assertEquals(context, needSyncFromUs, peer.isNeedSyncFromUs()); + Assert.assertEquals(context, blockBothHaveUpdateTime, peer.getBlockBothHaveUpdateTime()); Assert.assertTrue(context, peer.isFetchAble()); Assert.assertEquals(context, TronState.SYNCING, peer.getTronState()); Assert.assertEquals(context, 99, peer.getRemainNum()); diff --git a/framework/src/test/java/org/tron/core/net/peer/PeerStatusCheckMockTest.java b/framework/src/test/java/org/tron/core/net/peer/PeerStatusCheckMockTest.java index bd38cde5e4e..8f2a6bb076a 100644 --- a/framework/src/test/java/org/tron/core/net/peer/PeerStatusCheckMockTest.java +++ b/framework/src/test/java/org/tron/core/net/peer/PeerStatusCheckMockTest.java @@ -11,6 +11,7 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.tron.core.net.PeerSyncTestSupport.blockId; +import static org.tron.core.net.PeerSyncTestSupport.helloMessage; import java.util.Arrays; import java.util.Collections; @@ -175,6 +176,7 @@ public void testRequestWithinFiveSecondsDoesNotTimeOut() { @Test public void testCompletedResponseClearsDeadlineWithoutContribution() throws Exception { + peer.setHelloMessageReceive(helloMessage(20)); ChainInventoryMsgHandler handler = new ChainInventoryMsgHandler(); ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate); ReflectUtils.setFieldValue(handler, "syncService", mock(SyncService.class));