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..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 @@ -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,7 +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())) { + 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; } @@ -98,8 +104,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 +119,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 +132,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 +145,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..3cc78ed19f9 --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryCompletionTest.java @@ -0,0 +1,362 @@ +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/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()); + } +}