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 b7992745bb1..bc3135c19d1 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 @@ -18,10 +18,12 @@ 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; import org.tron.core.net.service.sync.SyncService; +import org.tron.protos.Protocol.Inventory.InventoryType; @Slf4j(topic = "net") @Component @@ -95,6 +97,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep } private void check(PeerConnection peer, ChainInventoryMessage msg) throws P2pException { + HelloMessage hello = peer.getHelloMessageReceive(); Pair, Long> requested = peer.getSyncChainRequested(); if (requested == null) { throw new P2pException(TypeEnum.BAD_MESSAGE, "not send syncBlockChainMsg"); @@ -147,6 +150,41 @@ private void check(PeerConnection peer, ChainInventoryMessage msg) throws P2pExc + ", futureMaxNum: " + maxFutureNum); } } + + if (msg.getRemainNum() == 0) { + BlockId lastBlockId = blockIds.get(blockIds.size() - 1); + long lastNum = lastBlockId.getNum(); + long helloHeadNum = hello.getHeadBlockId().getNum(); + if (lastNum < helloHeadNum) { + // A fork rollback can temporarily lower the head; allow a fresh HELLO on reconnect. + throw new P2pException(TypeEnum.SYNC_FAILED, + "lastNum " + lastNum + " (remainNum=0) is below hello head " + helloHeadNum); + } + + // A peer that has the terminal block in our summary should reply with that block alone. + if (blockIds.size() > 1 && requested.getKey().contains(lastBlockId)) { + // A concurrent fork switch can also produce this response; use a sync failure. + throw new P2pException(TypeEnum.SYNC_FAILED, + "terminal block already in summary: " + lastBlockId.getString()); + } + + long requestedAt = requested.getValue(); + // Only use recent announcements received before this request, not a newer head + // announced while the peer was preparing its response. A concurrent fork switch also cause it + long maxAdvInvNum = peer.getAdvInvReceive().asMap().entrySet().stream() + .filter(entry -> entry.getKey().getType() == InventoryType.BLOCK) + .filter(entry -> entry.getValue() <= requestedAt + && entry.getValue() >= requestedAt - NetConstants.ADV_TIME_OUT) + .mapToLong(entry -> new BlockId(entry.getKey().getHash()).getNum()) + .filter(height -> height >= 0) + .max() + .orElse(lastNum); + // Tolerate one block awaiting application, and allow fetching unknown terminal blocks. + if (lastNum < maxAdvInvNum - 1 && tronNetDelegate.containBlock(lastBlockId)) { + throw new P2pException(TypeEnum.SYNC_FAILED, + "lastNum " + lastNum + " (remainNum=0) is below advertised block " + maxAdvInvNum); + } + } } } diff --git a/framework/src/main/java/org/tron/core/net/service/effective/ResilienceService.java b/framework/src/main/java/org/tron/core/net/service/effective/ResilienceService.java index fca3efa3036..cb1f2c12586 100644 --- a/framework/src/main/java/org/tron/core/net/service/effective/ResilienceService.java +++ b/framework/src/main/java/org/tron/core/net/service/effective/ResilienceService.java @@ -31,7 +31,11 @@ public class ResilienceService { private static final long inactiveThreshold = CommonParameter.getInstance().getInactiveThreshold() * 1000L; - public static final long blockNotChangeThreshold = 60 * 1000L; + // Leave room for local GC pauses and disk stalls; a shorter threshold can evict healthy peers. + public static final long blockNotChangeThreshold = 30 * 1000L; + // Free slots early, but keep long bans at the previous threshold and cadence. + private static final long blockNotChangeBanThreshold = 60 * 1000L; + private static final long banRoundInterval = 30 * 1000L; //when node is isolated, retention percent peers will not be disconnected public static final double retentionPercent = 0.8; @@ -40,6 +44,7 @@ public class ResilienceService { private static final String esName = "resilience-service"; private final ScheduledExecutorService executor = ExecutorServiceManager .newSingleThreadScheduledExecutor(esName); + private long lastBanRoundTime; @Autowired private TronNetDelegate tronNetDelegate; @@ -68,13 +73,14 @@ public void init() { } }, initialDelay, 10, TimeUnit.SECONDS); + // Limit polling delay after the no-progress threshold is reached. executor.scheduleWithFixedDelay(() -> { try { disconnectIsolated2(); } catch (Exception e) { logger.error("DisconnectIsolated node failed", e); } - }, initialDelay, 30, TimeUnit.SECONDS); + }, initialDelay, 10, TimeUnit.SECONDS); } private void disconnectRandom() { @@ -158,6 +164,12 @@ private void disconnectIsolated2() { } logger.warn("Node is isolated, try to disconnect from peers"); + long now = System.currentTimeMillis(); + long stalled = now - chainBaseManager.getLatestSaveBlockTime(); + boolean ban = stalled >= blockNotChangeBanThreshold + && now - lastBanRoundTime >= banRoundInterval; + ReasonCode reason = ban ? ReasonCode.BAD_PROTOCOL : ReasonCode.SYNC_FAIL; + //disconnect from the node whose lastActiveTime is smallest int activePeerSize = (int) tronNetDelegate.getActivePeer().stream() .filter(peer -> peer.getChannel().isOutbound()) @@ -169,8 +181,12 @@ private void disconnectIsolated2() { .collect(Collectors.toList()); Optional one = getEarliestPeer(peers); - one.ifPresent(peer -> disconnectFromPeer(peer, ReasonCode.BAD_PROTOCOL, - DisconnectCause.ISOLATE2_ACTIVE)); + one.ifPresent(peer -> { + disconnectFromPeer(peer, reason, DisconnectCause.ISOLATE2_ACTIVE); + if (ban) { + lastBanRoundTime = System.currentTimeMillis(); + } + }); } //disconnect from some passive nodes, make sure retention nodes' num <= 0.8 * maxConnection, @@ -195,8 +211,12 @@ private void disconnectIsolated2() { } logger.info("All peer Size:{}, plan size:{}, candidate size:{}, real size:{}", peerSize, disconnectSize, candidateSize, peers.size()); - peers.forEach(peer -> disconnectFromPeer(peer, ReasonCode.BAD_PROTOCOL, - DisconnectCause.ISOLATE2_PASSIVE)); + peers.forEach(peer -> { + disconnectFromPeer(peer, reason, DisconnectCause.ISOLATE2_PASSIVE); + if (ban) { + lastBanRoundTime = System.currentTimeMillis(); + } + }); } } diff --git a/framework/src/test/java/org/tron/common/runtime/vm/PrecompiledContractsVerifyProofTest.java b/framework/src/test/java/org/tron/common/runtime/vm/PrecompiledContractsVerifyProofTest.java index 997b742d9ab..9a8ea0f828e 100644 --- a/framework/src/test/java/org/tron/common/runtime/vm/PrecompiledContractsVerifyProofTest.java +++ b/framework/src/test/java/org/tron/common/runtime/vm/PrecompiledContractsVerifyProofTest.java @@ -3699,7 +3699,7 @@ public void verifyBurnWrongHash() throws ZksnarkException { private Pair verifyTransfer(byte[] input) { transferContract.getEnergyForData(input); - transferContract.setVmShouldEndInUs(System.nanoTime() / 1000 + 80 * 1000); + transferContract.setVmShouldEndInUs(System.nanoTime() / 1000 + 200 * 1000); return transferContract.execute(input); } diff --git a/framework/src/test/java/org/tron/core/ShieldedTRC20BuilderTest.java b/framework/src/test/java/org/tron/core/ShieldedTRC20BuilderTest.java index a3c8cec16cd..be716b3cf90 100644 --- a/framework/src/test/java/org/tron/core/ShieldedTRC20BuilderTest.java +++ b/framework/src/test/java/org/tron/core/ShieldedTRC20BuilderTest.java @@ -2376,7 +2376,7 @@ private byte[] abiEncodeForBurn(ShieldedTRC20Parameters params, long value) { private Pair verifyTransfer(byte[] input) { transferContract.getEnergyForData(input); - transferContract.setVmShouldEndInUs(System.nanoTime() / 1000 + 80 * 1000); + transferContract.setVmShouldEndInUs(System.nanoTime() / 1000 + 200 * 1000); Pair ret = transferContract.execute(input); return ret; } 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 193571a3397..79e4c828da7 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 @@ -47,6 +47,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 { @@ -111,6 +112,7 @@ public void testQueueChangedDuringProcessing() throws Exception { BlockId next = new BlockId(Sha256Hash.ZERO_HASH, 1); PeerConnection peer = Mockito.mock(PeerConnection.class); Mockito.when(peer.getHelloMessageReceive()).thenReturn(helloMessage(0)); + Mockito.doReturn(this.peer.getAdvInvReceive()).when(peer).getAdvInvReceive(); Deque toFetch = Mockito.mock(Deque.class); LinkedList requested = new LinkedList<>(); requested.add(parent); @@ -175,37 +177,242 @@ public void testKnownSummaryBlockCompletesDownloadAndPreservesUploadDirection() } @Test - public void testSingleBlockBelowHelloHeadUsesLocalBlockAvailability() throws Exception { + 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)); - if (height > 100) { - peer.getSyncBlockToFetch().add(blockId(height)); - } - - respond(0, blockId(height)); - Assert.assertEquals(needSyncFromUs, peer.isNeedSyncFromUs()); - if (height <= 100) { - assertDownloadCompleted(); - Assert.assertEquals(!needSyncFromUs, peer.isSyncFinish()); - } else { - Assert.assertTrue(peer.isNeedSyncFromPeer()); - Assert.assertEquals(Collections.singletonList(blockId(height)), - new ArrayList<>(peer.getSyncBlockToFetch())); - verify(sync).setFetchFlag(true); - verify(sync, never()).syncNext(any()); - } - verify(peer, never()).disconnect(any()); + assertRejected(new ChainInventoryMessage(Collections.singletonList(blockId(height)), 0L), + TypeEnum.SYNC_FAILED); + Assert.assertFalse(peer.isSyncFinish()); } } } @Test - public void testTerminalMultiBlockBelowHelloHeadIsAccepted() throws Exception { + public void testKnownTerminalResponseBelowPriorBlockInventoryIsRejected() throws Exception { + for (List ids : Arrays.asList(Collections.singletonList(blockId(100)), + Arrays.asList(blockId(99), blockId(100)))) { + resetFixture(); + request(blockId(0), blockId(50), ids.get(0)); + peer.getAdvInvReceive().put(new Item(blockId(102), InventoryType.BLOCK), + peer.getSyncChainRequested().getValue() - 1); + + P2pException exception = assertRejected(new ChainInventoryMessage(ids, 0L), + TypeEnum.SYNC_FAILED); + + Assert.assertEquals("lastNum " + ids.get(ids.size() - 1).getNum() + + " (remainNum=0) is below advertised block 102", exception.getMessage()); + } + } + + @Test + public void testKnownTerminalBlockOneBelowInventoryCompletesDownload() throws Exception { + peer.getAdvInvReceive().put(new Item(blockId(101), InventoryType.BLOCK), + peer.getSyncChainRequested().getValue() - 1); + + respond(0, blockId(100)); + + assertDownloadCompleted(); + verify(sync, never()).syncNext(any()); + verify(sync, never()).setFetchFlag(true); + verify(peer, never()).disconnect(any()); + } + + @Test + public void testKnownMultiBlockOneBelowInventoryRequestsFinalSummary() throws Exception { + request(blockId(0), blockId(50), blockId(99)); + peer.getAdvInvReceive().put(new Item(blockId(101), InventoryType.BLOCK), + peer.getSyncChainRequested().getValue() - 1); + + respond(0, blockId(99), blockId(100)); + + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertTrue(peer.getSyncBlockToFetch().isEmpty()); + Assert.assertEquals(blockId(100), peer.getBlockBothHave()); + verify(sync).syncNext(peer); + verify(sync, never()).setFetchFlag(true); + verify(peer, never()).disconnect(any()); + } + + @Test + public void testTerminalResponseIgnoresNegativeBlockInventory() throws Exception { + for (long height : new long[]{-1L, Long.MIN_VALUE}) { + resetFixture(); + peer.getAdvInvReceive().put(new Item(blockId(height), InventoryType.BLOCK), + peer.getSyncChainRequested().getValue() - 1); + + respond(0, blockId(100)); + + assertDownloadCompleted(); + verify(peer, never()).disconnect(any()); + } + } + + @Test + public void testTerminalResponseBelowPriorInventoryFetchesMissingBlocks() throws Exception { + peer.getAdvInvReceive().put(new Item(blockId(103), InventoryType.BLOCK), + peer.getSyncChainRequested().getValue() - 1); + + respond(0, blockId(100), blockId(101)); + + Assert.assertEquals(Collections.singletonList(blockId(101)), + new ArrayList<>(peer.getSyncBlockToFetch())); + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertEquals(TronState.SYNCING, peer.getTronState()); + Assert.assertNull(peer.getSyncChainRequested()); + verify(sync).setFetchFlag(true); + verify(sync, never()).syncNext(any()); + verify(peer, never()).disconnect(any()); + } + + @Test + public void testKnownTerminalBlockOutsideSummaryBelowPriorInventoryIsRejected() throws Exception { + peer.setHelloMessageReceive(helloMessage(99)); + request(blockId(0), blockId(51), blockId(76), blockId(89), blockId(95), blockId(98), + blockId(100)); + peer.getAdvInvReceive().put(new Item(blockId(102), InventoryType.BLOCK), + peer.getSyncChainRequested().getValue() - 1); + + P2pException exception = assertRejected( + new ChainInventoryMessage(Arrays.asList(blockId(98), blockId(99)), 0L), + TypeEnum.SYNC_FAILED); + + Assert.assertEquals("lastNum 99 (remainNum=0) is below advertised block 102", + exception.getMessage()); + } + + @Test + public void testKnownTerminalBlockRejectsUnknownIntermediateBlock() throws Exception { + BlockId unknown99 = new BlockId(Sha256Hash.of(new byte[]{1}), 99); + BlockId fork100 = new BlockId(Sha256Hash.of(new byte[]{2}), 100); + for (BlockId lastBlockId : Arrays.asList(blockId(100), fork100)) { + resetFixture(); + when(delegate.containBlock(unknown99)).thenReturn(false); + when(delegate.containBlock(lastBlockId)).thenReturn(true); + when(delegate.containBlockInMainChain(lastBlockId)) + .thenReturn(lastBlockId.equals(blockId(100))); + request(blockId(0), blockId(50), blockId(98)); + peer.getAdvInvReceive().put(new Item(blockId(102), InventoryType.BLOCK), + peer.getSyncChainRequested().getValue() - 1); + + P2pException exception = assertRejected( + new ChainInventoryMessage(Arrays.asList(blockId(98), unknown99, lastBlockId), 0L), + TypeEnum.SYNC_FAILED); + + Assert.assertEquals("lastNum 100 (remainNum=0) is below advertised block 102", + exception.getMessage()); + } + } + + @Test + public void testTerminalResponseBelowPriorInventoryFetchesUnknownForkBlocks() throws Exception { + BlockId fork99 = new BlockId(Sha256Hash.of(new byte[]{1}), 99); + BlockId fork100 = new BlockId(Sha256Hash.of(new byte[]{2}), 100); + when(delegate.containBlock(fork99)).thenReturn(false); + when(delegate.containBlock(fork100)).thenReturn(false); + request(blockId(0), blockId(50), blockId(98)); + peer.getAdvInvReceive().put(new Item(blockId(102), InventoryType.BLOCK), + peer.getSyncChainRequested().getValue() - 1); + + respond(0, blockId(98), fork99, fork100); + + Assert.assertEquals(Arrays.asList(fork99, fork100), + new ArrayList<>(peer.getSyncBlockToFetch())); + Assert.assertTrue(peer.isNeedSyncFromPeer()); + verify(sync).setFetchFlag(true); + verify(sync, never()).syncNext(any()); + verify(peer, never()).disconnect(any()); + } + + @Test + public void testTerminalInventoryContradictionDisconnectsWithSyncFail() throws Exception { + peer.getAdvInvReceive().put(new Item(blockId(102), InventoryType.BLOCK), + peer.getSyncChainRequested().getValue() - 1); + 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 testTerminalResponseReachesPriorAdvertisedBlock() throws Exception { + peer.getAdvInvReceive().put(new Item(blockId(101), InventoryType.BLOCK), + peer.getSyncChainRequested().getValue() - 1); + + respond(0, blockId(100), blockId(101)); + + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertEquals(Collections.singletonList(blockId(101)), + new ArrayList<>(peer.getSyncBlockToFetch())); + verify(sync).setFetchFlag(true); + verify(peer, never()).disconnect(any()); + } + + @Test + public void testTerminalResponseIgnoresInventoryAfterRequest() throws Exception { + peer.getAdvInvReceive().put(new Item(blockId(102), InventoryType.BLOCK), + peer.getSyncChainRequested().getValue() + 1); + + respond(0, blockId(100)); + + assertDownloadCompleted(); + verify(peer, never()).disconnect(any()); + } + + @Test + public void testTerminalResponseIgnoresStaleBlockInventory() throws Exception { + peer.getAdvInvReceive().put(new Item(blockId(102), InventoryType.BLOCK), + peer.getSyncChainRequested().getValue() - NetConstants.ADV_TIME_OUT - 1); + + respond(0, blockId(100)); + + assertDownloadCompleted(); + verify(peer, never()).disconnect(any()); + } + + @Test + public void testTerminalResponseIgnoresTransactionInventory() throws Exception { + peer.getAdvInvReceive().put(new Item(blockId(102), InventoryType.TRX), + peer.getSyncChainRequested().getValue() - 1); + + respond(0, blockId(100)); + + assertDownloadCompleted(); + verify(peer, never()).disconnect(any()); + } + + @Test + public void testPagedResponseCanEndBelowPriorAdvertisedBlock() throws Exception { + List ids = range(100, (int) NetConstants.SYNC_FETCH_BATCH_NUM + 1); + peer.getAdvInvReceive().put(new Item(blockId(ids.get(ids.size() - 1).getNum() + 10), + InventoryType.BLOCK), peer.getSyncChainRequested().getValue() - 1); + + handler.processMessage(peer, new ChainInventoryMessage(ids, 10L)); + + Assert.assertTrue(peer.isNeedSyncFromPeer()); + Assert.assertEquals(10, peer.getRemainNum()); + verify(sync).setFetchFlag(true); + verify(peer, never()).disconnect(any()); + } + + @Test + public void testTerminalMultiBlockBelowHelloHeadIsRejected() throws Exception { for (int size : new int[]{2, 3, (int) NetConstants.SYNC_FETCH_BATCH_NUM + 1}) { resetFixture(); List ids = range(50, size); @@ -213,28 +420,16 @@ public void testTerminalMultiBlockBelowHelloHeadIsAccepted() throws Exception { peer.setBlockBothHave(blockId(50)); peer.getSyncBlockToFetch().add(blockId(101)); - handler.processMessage(peer, new ChainInventoryMessage(ids, 0L)); - - List pending = new ArrayList<>(ids); - pending.removeIf(id -> id.getNum() <= 100); - Assert.assertEquals(pending, new ArrayList<>(peer.getSyncBlockToFetch())); - Assert.assertTrue(peer.isNeedSyncFromPeer()); - Assert.assertNull(peer.getSyncChainRequested()); - Assert.assertEquals(0, peer.getRemainNum()); - if (pending.isEmpty()) { - verify(sync).syncNext(peer); - } else { - verify(sync).setFetchFlag(true); - } - verify(peer, never()).disconnect(any()); + assertRejected(new ChainInventoryMessage(ids, 0L), TypeEnum.SYNC_FAILED); } } @Test - public void testTerminalResponseBelowHelloHeadDoesNotDisconnect() throws Exception { + public void testTerminalResponseBelowHelloHeadDisconnectsWithSyncFail() throws Exception { for (int size : new int[]{1, 2, 3}) { resetFixture(); peer.setHelloMessageReceive(helloMessage(200)); + Pair, Long> requested = peer.getSyncChainRequested(); P2pEventHandlerImpl events = new P2pEventHandlerImpl(); ReflectUtils.setFieldValue(events, "chainInventoryMsgHandler", handler); ChainInventoryMessage response = new ChainInventoryMessage(range(50, size), 0L); @@ -242,15 +437,12 @@ public void testTerminalResponseBelowHelloHeadDoesNotDisconnect() throws Excepti ReflectUtils.invokeMethod(events, "processMessage", new Class[]{PeerConnection.class, byte[].class}, peer, response.getSendBytes()); - verify(peer, never()).disconnect(any()); - Assert.assertNull(peer.getSyncChainRequested()); - if (size == 1) { - assertDownloadCompleted(); - } else { - Assert.assertTrue(peer.isNeedSyncFromPeer()); - Assert.assertEquals(TronState.SYNCING, peer.getTronState()); - verify(sync).syncNext(peer); - } + 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); } } @@ -479,24 +671,26 @@ public void testKnownMultiBlockResponseClearsRemoteUploadState() throws Exceptio } @Test - public void testRepeatedKnownTerminalResponseRequestsAnotherSummary() throws Exception { + public void testRepeatedKnownTerminalResponseIsRejected() throws Exception { peer.setHelloMessageReceive(helloMessage(52)); respond(0, blockId(50), blockId(51), blockId(52)); verify(sync).syncNext(peer); request(blockId(0), blockId(50), blockId(52)); - respond(0, blockId(50), blockId(51), blockId(52)); + Pair, Long> requested = peer.getSyncChainRequested(); + long updateTime = peer.getBlockBothHaveUpdateTime(); + P2pException exception = Assert.assertThrows(P2pException.class, + () -> respond(0, blockId(50), blockId(51), blockId(52))); - Assert.assertNull(peer.getSyncChainRequested()); - Assert.assertEquals(blockId(52), peer.getBlockBothHave()); - Assert.assertTrue(peer.getSyncBlockToFetch().isEmpty()); + Assert.assertEquals(TypeEnum.SYNC_FAILED, exception.getType()); + Assert.assertSame(requested, peer.getSyncChainRequested()); + Assert.assertEquals(updateTime, peer.getBlockBothHaveUpdateTime()); Assert.assertTrue(peer.isNeedSyncFromPeer()); - verify(sync, times(2)).syncNext(peer); - verify(peer, never()).disconnect(any()); + verify(sync).syncNext(peer); } @Test - public void testReorgDuringSummaryResponseRequestsAnotherSummary() throws Exception { + public void testReorgDuringSummaryResponseDisconnectsWithSyncFail() throws Exception { request(blockId(0), blockId(51), blockId(76), blockId(89), blockId(95), blockId(98), blockId(100)); AtomicBoolean forkSwitched = new AtomicBoolean(); @@ -527,18 +721,16 @@ public void testReorgDuringSummaryResponseRequestsAnotherSummary() throws Except Assert.assertEquals(Arrays.asList(blockId(98), blockId(99), blockId(100)), response.getBlockIds()); Assert.assertEquals(Long.valueOf(0), response.getRemainNum()); + P2pException exception = assertRejected(response, TypeEnum.SYNC_FAILED); + Assert.assertTrue(exception.getMessage().startsWith("terminal block already in summary:")); P2pEventHandlerImpl events = new P2pEventHandlerImpl(); ReflectUtils.setFieldValue(events, "chainInventoryMsgHandler", handler); ReflectUtils.invokeMethod(events, "processMessage", new Class[]{PeerConnection.class, byte[].class}, peer, response.getSendBytes()); - verify(peer, never()).disconnect(any()); - Assert.assertNull(peer.getSyncChainRequested()); - Assert.assertEquals(blockId(100), peer.getBlockBothHave()); - Assert.assertTrue(peer.getSyncBlockToFetch().isEmpty()); - Assert.assertTrue(peer.isNeedSyncFromPeer()); - verify(sync).syncNext(peer); + verify(peer).disconnect(ReasonCode.SYNC_FAIL); + verify(peer, never()).disconnect(ReasonCode.BAD_PROTOCOL); } @Test diff --git a/framework/src/test/java/org/tron/core/net/services/ResilienceServiceTest.java b/framework/src/test/java/org/tron/core/net/services/ResilienceServiceTest.java index 4adaa232b87..b08b8b22f3c 100644 --- a/framework/src/test/java/org/tron/core/net/services/ResilienceServiceTest.java +++ b/framework/src/test/java/org/tron/core/net/services/ResilienceServiceTest.java @@ -6,6 +6,7 @@ import io.netty.channel.ChannelHandlerContext; import java.io.IOException; import java.net.InetSocketAddress; +import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Set; @@ -19,12 +20,15 @@ import org.tron.common.BaseTest; import org.tron.common.TestConstants; import org.tron.common.utils.ReflectUtils; +import org.tron.core.ChainBaseManager; import org.tron.core.config.args.Args; import org.tron.core.net.P2pEventHandlerImpl; +import org.tron.core.net.TronNetDelegate; import org.tron.core.net.peer.PeerConnection; import org.tron.core.net.peer.PeerManager; import org.tron.core.net.service.effective.ResilienceService; import org.tron.p2p.connection.Channel; +import org.tron.protos.Protocol.ReasonCode; public class ResilienceServiceTest extends BaseTest { @@ -246,4 +250,175 @@ public void testDisconnectIsolated2() { Assert.assertEquals((int) (maxConnection * ResilienceService.retentionPercent), PeerManager.getPeers().size()); } -} \ No newline at end of file + + @Test + public void testDisconnectIsolated2UsesCooldownBeforeBanThreshold() { + List peers = isolatedPeers(3, 23); + PeerConnection outbound = peers.get(0); + PeerConnection passive = peers.get(3); + ResilienceService isolatedService = isolatedService(peers, 35_000L); + try { + ReflectUtils.invokeMethod(isolatedService, "disconnectIsolated2"); + + Mockito.verify(outbound).disconnect(ReasonCode.SYNC_FAIL); + Mockito.verify(passive).disconnect(ReasonCode.SYNC_FAIL); + Assert.assertEquals(24, peers.size()); + Assert.assertEquals(0L, + (long) ReflectUtils.getFieldObject(isolatedService, "lastBanRoundTime")); + } finally { + isolatedService.close(); + } + } + + @Test + public void testDisconnectIsolated2LimitsBanRounds() { + List peers = isolatedPeers(3, 23); + List original = new ArrayList<>(peers); + ResilienceService isolatedService = isolatedService(peers, 65_000L); + try { + ReflectUtils.invokeMethod(isolatedService, "disconnectIsolated2"); + + Mockito.verify(original.get(0)).disconnect(ReasonCode.BAD_PROTOCOL); + Mockito.verify(original.get(3)).disconnect(ReasonCode.BAD_PROTOCOL); + long lastBanTime = (long) ReflectUtils.getFieldObject(isolatedService, "lastBanRoundTime"); + Assert.assertTrue(lastBanTime > 0); + + // Replace the two disconnected peers so the next round can free slots again. + peers.add(isolatedPeer(peers, true, 100)); + peers.add(isolatedPeer(peers, false, 100)); + ReflectUtils.invokeMethod(isolatedService, "disconnectIsolated2"); + + Mockito.verify(original.get(1)).disconnect(ReasonCode.SYNC_FAIL); + Mockito.verify(original.get(4)).disconnect(ReasonCode.SYNC_FAIL); + Assert.assertEquals(lastBanTime, + (long) ReflectUtils.getFieldObject(isolatedService, "lastBanRoundTime")); + + peers.add(isolatedPeer(peers, true, 101)); + peers.add(isolatedPeer(peers, false, 101)); + ReflectUtils.setFieldValue(isolatedService, "lastBanRoundTime", + System.currentTimeMillis() - 31_000L); + ReflectUtils.invokeMethod(isolatedService, "disconnectIsolated2"); + + Mockito.verify(original.get(2)).disconnect(ReasonCode.BAD_PROTOCOL); + Mockito.verify(original.get(5)).disconnect(ReasonCode.BAD_PROTOCOL); + Assert.assertTrue( + (long) ReflectUtils.getFieldObject(isolatedService, "lastBanRoundTime") >= lastBanTime); + } finally { + isolatedService.close(); + } + } + + @Test + public void testDisconnectIsolated2RecordsPassiveBanRound() { + List peers = isolatedPeers(0, 26); + PeerConnection first = peers.get(0); + PeerConnection second = peers.get(1); + ResilienceService isolatedService = isolatedService(peers, 65_000L); + try { + ReflectUtils.invokeMethod(isolatedService, "disconnectIsolated2"); + + Mockito.verify(first).disconnect(ReasonCode.BAD_PROTOCOL); + Mockito.verify(second).disconnect(ReasonCode.BAD_PROTOCOL); + Assert.assertEquals(24, peers.size()); + Assert.assertTrue( + (long) ReflectUtils.getFieldObject(isolatedService, "lastBanRoundTime") > 0); + } finally { + isolatedService.close(); + } + } + + @Test + public void testDisconnectIsolated2WaitsForStallThreshold() { + List peers = isolatedPeers(3, 23); + ResilienceService isolatedService = isolatedService(peers, 25_000L); + try { + ReflectUtils.invokeMethod(isolatedService, "disconnectIsolated2"); + + peers.forEach(peer -> Mockito.verify(peer, Mockito.never()).disconnect(any())); + Assert.assertEquals(0L, + (long) ReflectUtils.getFieldObject(isolatedService, "lastBanRoundTime")); + } finally { + isolatedService.close(); + } + } + + @Test + public void testDisconnectIsolated2StopsAfterBlockProgress() { + List peers = isolatedPeers(3, 23); + ResilienceService isolatedService = isolatedService(peers, 65_000L); + try { + ReflectUtils.invokeMethod(isolatedService, "disconnectIsolated2"); + long lastBanTime = (long) ReflectUtils.getFieldObject(isolatedService, "lastBanRoundTime"); + peers.add(isolatedPeer(peers, true, 100)); + peers.add(isolatedPeer(peers, false, 100)); + ChainBaseManager chain = (ChainBaseManager) + ReflectUtils.getFieldObject(isolatedService, "chainBaseManager"); + Mockito.when(chain.getLatestSaveBlockTime()).thenReturn(System.currentTimeMillis()); + + ReflectUtils.invokeMethod(isolatedService, "disconnectIsolated2"); + + peers.forEach(peer -> Mockito.verify(peer, Mockito.never()).disconnect(any())); + Assert.assertEquals(26, peers.size()); + Assert.assertEquals(lastBanTime, + (long) ReflectUtils.getFieldObject(isolatedService, "lastBanRoundTime")); + } finally { + isolatedService.close(); + } + } + + @Test + public void testDisconnectIsolated2EmptyRoundDoesNotDelayBan() { + List peers = isolatedPeers(2, 0); + PeerConnection first = peers.get(0); + ResilienceService isolatedService = isolatedService(peers, 65_000L); + try { + ReflectUtils.invokeMethod(isolatedService, "disconnectIsolated2"); + + peers.forEach(peer -> Mockito.verify(peer, Mockito.never()).disconnect(any())); + Assert.assertEquals(0L, + (long) ReflectUtils.getFieldObject(isolatedService, "lastBanRoundTime")); + + peers.add(isolatedPeer(peers, true, 100)); + ReflectUtils.invokeMethod(isolatedService, "disconnectIsolated2"); + + Mockito.verify(first).disconnect(ReasonCode.BAD_PROTOCOL); + Assert.assertTrue( + (long) ReflectUtils.getFieldObject(isolatedService, "lastBanRoundTime") > 0); + } finally { + isolatedService.close(); + } + } + + private ResilienceService isolatedService(List peers, long stalledMillis) { + ResilienceService isolatedService = new ResilienceService(); + TronNetDelegate delegate = Mockito.mock(TronNetDelegate.class); + ChainBaseManager chain = Mockito.mock(ChainBaseManager.class); + Mockito.when(delegate.getActivePeer()).thenReturn(peers); + Mockito.when(chain.getLatestSaveBlockTime()) + .thenReturn(System.currentTimeMillis() - stalledMillis); + ReflectUtils.setFieldValue(isolatedService, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(isolatedService, "chainBaseManager", chain); + return isolatedService; + } + + private List isolatedPeers(int outboundCount, int passiveCount) { + List peers = new ArrayList<>(); + for (int i = 0; i < outboundCount + passiveCount; i++) { + peers.add(isolatedPeer(peers, i < outboundCount, i)); + } + return peers; + } + + private PeerConnection isolatedPeer(List peers, boolean outbound, long order) { + PeerConnection peer = Mockito.mock(PeerConnection.class); + Channel channel = Mockito.mock(Channel.class); + Mockito.when(peer.getChannel()).thenReturn(channel); + Mockito.when(channel.isOutbound()).thenReturn(outbound); + Mockito.when(peer.getLastInteractiveTime()).thenReturn(order); + Mockito.doAnswer(invocation -> { + peers.remove(peer); + return null; + }).when(peer).disconnect(any()); + return peer; + } +}