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..ce58369b89d 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,39 @@ 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()) + .max() + .orElse(lastNum); + if (lastNum < maxAdvInvNum) { + 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..6cdddcae5c4 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,8 @@ 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; //when node is isolated, retention percent peers will not be disconnected public static final double retentionPercent = 0.8; @@ -68,13 +69,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() { 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..3a0acbff866 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,124 @@ 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 testTerminalResponseBelowPriorBlockInventoryIsRejected() throws Exception { + for (List ids : Arrays.asList(Collections.singletonList(blockId(100)), + Arrays.asList(blockId(99), blockId(100)), + Arrays.asList(blockId(100), blockId(101)))) { + 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 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 +302,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 +319,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 +553,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 +603,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