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..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 @@ -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; @@ -18,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; @@ -40,7 +42,8 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep ChainInventoryMessage chainInventoryMessage = (ChainInventoryMessage) msg; - check(peer, chainInventoryMessage); + Pair, Long> requested = peer.getSyncChainRequested(); + check(peer, requested, chainInventoryMessage); peer.setFetchAble(false); @@ -51,6 +54,7 @@ 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); peer.setNeedSyncFromPeer(false); return; @@ -98,11 +102,17 @@ 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(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"); @@ -112,7 +122,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 +135,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,11 +148,20 @@ 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); } } + + 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/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..73f053664e3 --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/PeerSyncTestSupport.java @@ -0,0 +1,45 @@ +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.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 { + + private PeerSyncTestSupport() { + } + + 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); + 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/ChainInventoryMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java index 56853c3dbb7..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 @@ -1,24 +1,56 @@ 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 static org.tron.core.net.PeerSyncTestSupport.helloMessage; + 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.P2pEventHandlerImpl; +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; +import org.tron.protos.Protocol.ReasonCode; 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 +61,361 @@ 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.setHelloMessageReceive(helloMessage(100)); + 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()); + + request(); + Assert.assertEquals("blockIds is empty", assertRejected(msg).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.setHelloMessageReceive(helloMessage(height)); + 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); + } + } } + } - peer.setSyncChainRequested(new Pair<>(new LinkedList<>(), System.currentTimeMillis())); + @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)); - try { - handler.processMessage(peer, msg); - } catch (P2pException e) { - Assert.assertEquals("blockIds is empty", e.getMessage()); + assertRejected(new ChainInventoryMessage(Collections.singletonList(blockId(height)), 0L), + TypeEnum.SYNC_FAILED); + Assert.assertFalse(peer.isSyncFinish()); + } } + } - long size = NetConstants.SYNC_FETCH_BATCH_NUM + 2; - for (int i = 0; i < size; i++) { - blockIds.add(new BlockId()); + @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)); + + 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. + peer.setHelloMessageReceive(helloMessage(50)); + 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 { + peer.setHelloMessageReceive(helloMessage(50)); + 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) { + 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, expectedType, exception.getType()); + Assert.assertSame(context, requested, peer.getSyncChainRequested()); + Assert.assertTrue(context, peer.isNeedSyncFromPeer()); + 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()); + 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..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 @@ -1,27 +1,86 @@ 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 static org.tron.core.net.PeerSyncTestSupport.helloMessage; +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 +100,154 @@ 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 { + peer.setHelloMessageReceive(helloMessage(20)); + 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()); + } }