From a8d9cfd33f98d19afabc1966fd8657a7bde54294 Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Mon, 2 Nov 2026 18:18:18 +0800 Subject: [PATCH 01/10] fix(net): reject inconsistent sync replies and retry another peer Restore terminal chain inventory checks against the HELLO head and requested summary, and reject replies below recent block announcements received before the request. Resume synchronization with a random eligible peer through the existing fetch worker after a syncing peer disconnects. Cover source selection, retries, and recovery of previously discarded blocks. --- .../ChainInventoryMsgHandler.java | 33 +++ .../core/net/service/sync/SyncService.java | 32 +++ .../ChainInventoryMsgHandlerTest.java | 188 ++++++++---- .../core/net/services/SyncServiceTest.java | 268 ++++++++++++++++++ 4 files changed, 464 insertions(+), 57 deletions(-) diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java index b7992745bb1..9a7b05765bb 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,36 @@ 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. + long advertisedNum = 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 < advertisedNum) { + throw new P2pException(TypeEnum.SYNC_FAILED, + "lastNum " + lastNum + " (remainNum=0) is below advertised block " + advertisedNum); + } + } } } diff --git a/framework/src/main/java/org/tron/core/net/service/sync/SyncService.java b/framework/src/main/java/org/tron/core/net/service/sync/SyncService.java index 8ccfb1e6886..9098e1ed85d 100644 --- a/framework/src/main/java/org/tron/core/net/service/sync/SyncService.java +++ b/framework/src/main/java/org/tron/core/net/service/sync/SyncService.java @@ -13,7 +13,9 @@ import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; @@ -74,6 +76,8 @@ public class SyncService { private volatile long maxRequestedBlockNum = 0; + private final AtomicReference syncPeerToRetry = new AtomicReference<>(); + public void init() { ExecutorServiceManager.scheduleWithFixedDelay(fetchExecutor, () -> { try { @@ -159,6 +163,11 @@ public void onDisconnect(PeerConnection peer) { if (!peer.getSyncBlockRequested().isEmpty()) { peer.getSyncBlockRequested().keySet().forEach(blockId -> invalid(blockId, peer)); } + if (peer.getTronState() == TronState.SYNCING) { + // A rejected chain summary can interrupt a download before any block was requested. + syncPeerToRetry.set(peer); + fetchFlag = true; + } } private void invalid(BlockId blockId, PeerConnection peerConnection) { @@ -233,8 +242,31 @@ private BlockId getBlockIdByNum(long num) throws P2pException { return tronNetDelegate.getBlockIdByNum(num); } + private void retrySync(Collection activePeers) { + PeerConnection failedPeer = syncPeerToRetry.get(); + if (failedPeer != null) { + List candidates = new ArrayList<>(); + for (PeerConnection peer : activePeers) { + if (peer != failedPeer && !peer.isDisconnect() && peer.getHelloMessageReceive() != null + && peer.getTronState() != TronState.SYNCING && peer.isSyncIdle()) { + candidates.add(peer); + } + } + if (!candidates.isEmpty() && syncPeerToRetry.compareAndSet(failedPeer, null)) { + PeerConnection peer = candidates.get( + ThreadLocalRandom.current().nextInt(candidates.size())); + logger.info("Retry sync from {} after disconnecting {}", + peer.getInetSocketAddress(), failedPeer.getInetSocketAddress()); + startSync(peer); + } else { + fetchFlag = true; + } + } + } + private void startFetchSyncBlock() { Collection activePeers = tronNetDelegate.getActivePeer(); + retrySync(activePeers); int reqNum = activePeers.stream() .mapToInt(p -> p.getSyncBlockRequested().size()).sum(); int remainNum; 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 diff --git a/framework/src/test/java/org/tron/core/net/services/SyncServiceTest.java b/framework/src/test/java/org/tron/core/net/services/SyncServiceTest.java index d518d7dcc7b..4bcd29f88b0 100644 --- a/framework/src/test/java/org/tron/core/net/services/SyncServiceTest.java +++ b/framework/src/test/java/org/tron/core/net/services/SyncServiceTest.java @@ -1,40 +1,79 @@ package org.tron.core.net.services; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.doAnswer; 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 static org.tron.core.net.PeerSyncTestSupport.helloMessage; import com.google.common.cache.Cache; +import com.google.protobuf.ByteString; import java.lang.reflect.Field; import java.lang.reflect.Method; import java.net.InetSocketAddress; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedList; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.junit.Assert; import org.junit.Test; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import org.springframework.context.ApplicationContext; import org.tron.common.BaseMethodTest; 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; +import org.tron.core.capsule.BlockCapsule.BlockId; 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.BlockMessage; import org.tron.core.net.message.adv.FetchInvDataMessage; +import org.tron.core.net.message.sync.ChainInventoryMessage; +import org.tron.core.net.message.sync.SyncBlockChainMessage; +import org.tron.core.net.messagehandler.BlockMsgHandler; +import org.tron.core.net.messagehandler.ChainInventoryMsgHandler; import org.tron.core.net.messagehandler.PbftDataSyncHandler; +import org.tron.core.net.peer.Item; import org.tron.core.net.peer.PeerConnection; import org.tron.core.net.peer.PeerManager; import org.tron.core.net.peer.TronState; +import org.tron.core.net.service.adv.AdvService; +import org.tron.core.net.service.fetchblock.FetchBlockService; import org.tron.core.net.service.sync.SyncService; import org.tron.core.net.service.sync.UnparsedBlock; import org.tron.p2p.connection.Channel; import org.tron.protos.Protocol; +import org.tron.protos.Protocol.Inventory.InventoryType; +import org.tron.protos.Protocol.ReasonCode; public class SyncServiceTest extends BaseMethodTest { private SyncService service; + private SyncService recoveryService; + private TronNetDelegate delegate; + private PeerConnection failed; + private PeerConnection replacement; + private List peers; + private final AtomicReference head = new AtomicReference<>(); + private final Set stored = new HashSet<>(); + private final Map chainIds = new HashMap<>(); + private PeerConnection peer; private P2pEventHandlerImpl p2pEventHandler; private ApplicationContext ctx; @@ -55,6 +94,9 @@ protected void afterInit() { @Override protected void beforeDestroy() { + if (recoveryService != null) { + recoveryService.close(); + } for (PeerConnection p : PeerManager.getPeers()) { PeerManager.remove(p.getChannel()); } @@ -404,4 +446,230 @@ public void testHandleSyncBlock() throws Exception { Assert.assertTrue(requestBlockIds.getIfPresent(blockId) == null); } + + @Test + public void testRejectedSummaryRecoversPreviouslyDiscardedBlocks() throws Exception { + initRecovery(); + BlockCapsule parent = block(101, head.get()); + BlockCapsule child = block(102, parent.getBlockId()); + BlockMsgHandler blocks = new BlockMsgHandler(); + ReflectUtils.setFieldValue(blocks, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(blocks, "syncService", recoveryService); + ReflectUtils.setFieldValue(blocks, "fetchBlockService", mock(FetchBlockService.class)); + ReflectUtils.setFieldValue(blocks, "fastForward", false); + Item childItem = new Item(child.getBlockId(), InventoryType.BLOCK); + failed.getAdvInvReceive().put(childItem, System.currentTimeMillis() - 1); + failed.getAdvInvRequest().put(childItem, System.currentTimeMillis()); + + blocks.processMessage(failed, new BlockMessage(child)); + Assert.assertFalse(failed.getAdvInvRequest().containsKey(childItem)); + Assert.assertFalse(stored.contains(child.getBlockId())); + ChainInventoryMsgHandler inventory = new ChainInventoryMsgHandler(); + ReflectUtils.setFieldValue(inventory, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(inventory, "syncService", recoveryService); + P2pEventHandlerImpl events = new P2pEventHandlerImpl(); + ReflectUtils.setFieldValue(events, "chainInventoryMsgHandler", inventory); + ChainInventoryMessage echo = new ChainInventoryMessage( + Collections.singletonList(head.get()), 0L); + ReflectUtils.invokeMethod(events, "processMessage", + new Class[]{PeerConnection.class, byte[].class}, failed, echo.getSendBytes()); + verify(failed).disconnect(ReasonCode.SYNC_FAIL); + + peers.remove(failed); + when(failed.getChannel().isDisconnect()).thenReturn(true); + ReflectUtils.setFieldValue(failed, "syncService", recoveryService); + ReflectUtils.setFieldValue(failed, "advService", mock(AdvService.class)); + failed.onDisconnect(); + runFetchTask(); + verify(replacement).sendMessage(any(SyncBlockChainMessage.class)); + + inventory.processMessage(replacement, new ChainInventoryMessage( + Arrays.asList(blockId(100), parent.getBlockId(), child.getBlockId()), 0L)); + runFetchTask(); + ArgumentCaptor requests = ArgumentCaptor.forClass(Message.class); + verify(replacement, times(2)).sendMessage(requests.capture()); + Assert.assertEquals(Arrays.asList(parent.getBlockId(), child.getBlockId()), + ((FetchInvDataMessage) requests.getValue()).getHashList()); + + blocks.processMessage(replacement, new BlockMessage(parent)); + blocks.processMessage(replacement, new BlockMessage(child)); + ReflectUtils.invokeMethod(recoveryService, "handleSyncBlock"); + inventory.processMessage(replacement, new ChainInventoryMessage( + Collections.singletonList(child.getBlockId()), 0L)); + + Assert.assertEquals(child.getBlockId(), head.get()); + Assert.assertTrue(stored.contains(parent.getBlockId())); + Assert.assertTrue(stored.contains(child.getBlockId())); + Assert.assertEquals(TronState.SYNC_COMPLETED, replacement.getTronState()); + Assert.assertFalse(replacement.isNeedSyncFromPeer()); + verify(replacement, never()).disconnect(any()); + } + + @Test + public void testRecoveryFiltersCandidatesAndAllowsPendingTransactions() throws Exception { + initRecovery(); + PeerConnection unhandshaken = readyPeer(18890); + unhandshaken.setHelloMessageReceive(null); + PeerConnection disconnected = readyPeer(18891); + when(disconnected.getChannel().isDisconnect()).thenReturn(true); + PeerConnection syncing = readyPeer(18892); + syncing.setTronState(TronState.SYNCING); + PeerConnection pendingSummary = readyPeer(18893); + pendingSummary.setSyncChainRequested(new Pair<>(new LinkedList<>( + Collections.singletonList(blockId(100))), System.currentTimeMillis())); + PeerConnection pendingBlock = readyPeer(18894); + pendingBlock.getSyncBlockRequested().put(blockId(101), System.currentTimeMillis()); + peers.addAll(Arrays.asList(unhandshaken, disconnected, syncing, pendingSummary, pendingBlock)); + replacement.getAdvInvRequest().put(new Item(Sha256Hash.ZERO_HASH, InventoryType.TRX), + System.currentTimeMillis()); + Assert.assertFalse(replacement.isIdle()); + queueRecovery(failed); + // The failed connection must remain excluded even if its state changes during cleanup. + failed.setTronState(TronState.SYNC_COMPLETED); + + runFetchTask(); + + verify(replacement).sendMessage(any(SyncBlockChainMessage.class)); + for (PeerConnection peer : Arrays.asList(failed, unhandshaken, disconnected, syncing, + pendingSummary, pendingBlock)) { + verify(peer, never()).sendMessage(any()); + } + } + + @Test + public void testRecoveryWaitsForAnEligibleConnection() throws Exception { + initRecovery(); + peers.clear(); + queueRecovery(failed); + runFetchTask(); + Assert.assertTrue((boolean) ReflectUtils.getFieldObject(recoveryService, "fetchFlag")); + PeerConnection unhandshaken = readyPeer(18890); + unhandshaken.setHelloMessageReceive(null); + peers.add(unhandshaken); + runFetchTask(); + Assert.assertTrue((boolean) ReflectUtils.getFieldObject(recoveryService, "fetchFlag")); + peers.add(replacement); + + runFetchTask(); + runFetchTask(); + + Assert.assertFalse((boolean) ReflectUtils.getFieldObject(recoveryService, "fetchFlag")); + verify(replacement, times(1)).sendMessage(any(SyncBlockChainMessage.class)); + verify(unhandshaken, never()).sendMessage(any()); + } + + @Test + public void testCompletedPeerDisconnectDoesNotStartRecovery() throws Exception { + initRecovery(); + recoveryService.onDisconnect(failed); + + runFetchTask(); + + verify(replacement, never()).sendMessage(any()); + } + + @Test + public void testRecoveryStartsOnlyOneRandomCandidate() throws Exception { + initRecovery(); + PeerConnection other = readyPeer(18890); + peers.add(other); + AtomicInteger requests = new AtomicInteger(); + for (PeerConnection peer : Arrays.asList(replacement, other)) { + doAnswer(call -> { + requests.incrementAndGet(); + return null; + }).when(peer).sendMessage(any(SyncBlockChainMessage.class)); + } + queueRecovery(failed); + + runFetchTask(); + runFetchTask(); + + Assert.assertEquals(1, requests.get()); + } + + @Test + public void testAnotherFailedSourceCanTriggerRecovery() throws Exception { + initRecovery(); + queueRecovery(failed); + runFetchTask(); + PeerConnection next = readyPeer(18890); + peers.remove(replacement); + peers.add(next); + + recoveryService.onDisconnect(replacement); + runFetchTask(); + + verify(next).sendMessage(any(SyncBlockChainMessage.class)); + } + + private void initRecovery() throws Exception { + recoveryService = new SyncService(); + delegate = mock(TronNetDelegate.class); + failed = readyPeer(18888); + replacement = readyPeer(18889); + peers = new ArrayList<>(Arrays.asList(failed, replacement)); + head.set(blockId(100)); + stored.add(blockId(0)); + stored.add(head.get()); + chainIds.put(0L, blockId(0)); + chainIds.put(100L, head.get()); + when(delegate.getActivePeer()).thenReturn(peers); + when(delegate.getHeadBlockId()).thenAnswer(call -> head.get()); + when(delegate.getKhaosDbHeadBlockId()).thenAnswer(call -> head.get()); + when(delegate.getGenesisBlockId()).thenReturn(blockId(0)); + when(delegate.getSolidBlockId()).thenReturn(blockId(0)); + when(delegate.getBlockTime(any())).thenReturn(System.currentTimeMillis() - 30_000_000L); + when(delegate.getBlockIdByNum(anyLong())).thenAnswer(call -> { + long number = call.getArgument(0); + return chainIds.getOrDefault(number, blockId(number)); + }); + when(delegate.containBlock(any())).thenAnswer(call -> stored.contains(call.getArgument(0))); + when(delegate.containBlockInMainChain(any())) + .thenAnswer(call -> stored.contains(call.getArgument(0))); + when(delegate.getBlockLock()).thenReturn(new Object()); + when(delegate.getForkLock()).thenReturn(new Object()); + when(delegate.validAdvBlock(any(), any())).thenReturn(true); + doAnswer(call -> { + BlockCapsule block = call.getArgument(0); + stored.add(block.getBlockId()); + chainIds.put(block.getNum(), block.getBlockId()); + head.set(block.getBlockId()); + return null; + }).when(delegate).processBlock(any(), anyBoolean()); + ReflectUtils.setFieldValue(recoveryService, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(recoveryService, "pbftDataSyncHandler", + mock(PbftDataSyncHandler.class)); + ReflectUtils.setFieldValue(recoveryService, "maxPendingBlockSize", 10); + } + + private void queueRecovery(PeerConnection peer) { + peer.setTronState(TronState.SYNCING); + peer.setNeedSyncFromPeer(true); + recoveryService.onDisconnect(peer); + Assert.assertTrue((boolean) ReflectUtils.getFieldObject(recoveryService, "fetchFlag")); + } + + private void runFetchTask() { + if ((boolean) ReflectUtils.getFieldObject(recoveryService, "fetchFlag")) { + recoveryService.setFetchFlag(false); + ReflectUtils.invokeMethod(recoveryService, "startFetchSyncBlock"); + } + } + + private static PeerConnection readyPeer(int port) throws Exception { + PeerConnection peer = PeerSyncTestSupport.peer(port); + peer.setHelloMessageReceive(helloMessage(100)); + peer.setTronState(TronState.SYNC_COMPLETED); + peer.setNeedSyncFromPeer(false); + peer.setNeedSyncFromUs(false); + return peer; + } + + private static BlockCapsule block(long number, BlockId parent) { + BlockCapsule block = new BlockCapsule(number, parent, System.currentTimeMillis() - 1_000, + ByteString.copyFromUtf8("witness")); + block.setMerkleRoot(); + return block; + } } From 6c90efcb4616a6be3dfc84c69656be299c384688 Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Tue, 3 Nov 2026 11:53:38 +0800 Subject: [PATCH 02/10] revert(net): remove proactive sync peer retries Keep terminal chain inventory validation and its tests focused in this PR. Restore SyncService and SyncServiceTest to release_v4.8.3 so proactive source recovery can be reviewed separately. --- .../core/net/service/sync/SyncService.java | 32 --- .../core/net/services/SyncServiceTest.java | 268 ------------------ 2 files changed, 300 deletions(-) diff --git a/framework/src/main/java/org/tron/core/net/service/sync/SyncService.java b/framework/src/main/java/org/tron/core/net/service/sync/SyncService.java index 9098e1ed85d..8ccfb1e6886 100644 --- a/framework/src/main/java/org/tron/core/net/service/sync/SyncService.java +++ b/framework/src/main/java/org/tron/core/net/service/sync/SyncService.java @@ -13,9 +13,7 @@ import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; @@ -76,8 +74,6 @@ public class SyncService { private volatile long maxRequestedBlockNum = 0; - private final AtomicReference syncPeerToRetry = new AtomicReference<>(); - public void init() { ExecutorServiceManager.scheduleWithFixedDelay(fetchExecutor, () -> { try { @@ -163,11 +159,6 @@ public void onDisconnect(PeerConnection peer) { if (!peer.getSyncBlockRequested().isEmpty()) { peer.getSyncBlockRequested().keySet().forEach(blockId -> invalid(blockId, peer)); } - if (peer.getTronState() == TronState.SYNCING) { - // A rejected chain summary can interrupt a download before any block was requested. - syncPeerToRetry.set(peer); - fetchFlag = true; - } } private void invalid(BlockId blockId, PeerConnection peerConnection) { @@ -242,31 +233,8 @@ private BlockId getBlockIdByNum(long num) throws P2pException { return tronNetDelegate.getBlockIdByNum(num); } - private void retrySync(Collection activePeers) { - PeerConnection failedPeer = syncPeerToRetry.get(); - if (failedPeer != null) { - List candidates = new ArrayList<>(); - for (PeerConnection peer : activePeers) { - if (peer != failedPeer && !peer.isDisconnect() && peer.getHelloMessageReceive() != null - && peer.getTronState() != TronState.SYNCING && peer.isSyncIdle()) { - candidates.add(peer); - } - } - if (!candidates.isEmpty() && syncPeerToRetry.compareAndSet(failedPeer, null)) { - PeerConnection peer = candidates.get( - ThreadLocalRandom.current().nextInt(candidates.size())); - logger.info("Retry sync from {} after disconnecting {}", - peer.getInetSocketAddress(), failedPeer.getInetSocketAddress()); - startSync(peer); - } else { - fetchFlag = true; - } - } - } - private void startFetchSyncBlock() { Collection activePeers = tronNetDelegate.getActivePeer(); - retrySync(activePeers); int reqNum = activePeers.stream() .mapToInt(p -> p.getSyncBlockRequested().size()).sum(); int remainNum; diff --git a/framework/src/test/java/org/tron/core/net/services/SyncServiceTest.java b/framework/src/test/java/org/tron/core/net/services/SyncServiceTest.java index 4bcd29f88b0..d518d7dcc7b 100644 --- a/framework/src/test/java/org/tron/core/net/services/SyncServiceTest.java +++ b/framework/src/test/java/org/tron/core/net/services/SyncServiceTest.java @@ -1,79 +1,40 @@ package org.tron.core.net.services; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyBoolean; -import static org.mockito.ArgumentMatchers.anyLong; -import static org.mockito.Mockito.doAnswer; 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 static org.tron.core.net.PeerSyncTestSupport.helloMessage; import com.google.common.cache.Cache; -import com.google.protobuf.ByteString; import java.lang.reflect.Field; import java.lang.reflect.Method; import java.net.InetSocketAddress; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; -import java.util.HashMap; -import java.util.HashSet; -import java.util.LinkedList; import java.util.List; import java.util.Map; -import java.util.Set; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; import org.junit.Assert; import org.junit.Test; -import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import org.springframework.context.ApplicationContext; import org.tron.common.BaseMethodTest; 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; -import org.tron.core.capsule.BlockCapsule.BlockId; 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.BlockMessage; import org.tron.core.net.message.adv.FetchInvDataMessage; -import org.tron.core.net.message.sync.ChainInventoryMessage; -import org.tron.core.net.message.sync.SyncBlockChainMessage; -import org.tron.core.net.messagehandler.BlockMsgHandler; -import org.tron.core.net.messagehandler.ChainInventoryMsgHandler; import org.tron.core.net.messagehandler.PbftDataSyncHandler; -import org.tron.core.net.peer.Item; import org.tron.core.net.peer.PeerConnection; import org.tron.core.net.peer.PeerManager; import org.tron.core.net.peer.TronState; -import org.tron.core.net.service.adv.AdvService; -import org.tron.core.net.service.fetchblock.FetchBlockService; import org.tron.core.net.service.sync.SyncService; import org.tron.core.net.service.sync.UnparsedBlock; import org.tron.p2p.connection.Channel; import org.tron.protos.Protocol; -import org.tron.protos.Protocol.Inventory.InventoryType; -import org.tron.protos.Protocol.ReasonCode; public class SyncServiceTest extends BaseMethodTest { private SyncService service; - private SyncService recoveryService; - private TronNetDelegate delegate; - private PeerConnection failed; - private PeerConnection replacement; - private List peers; - private final AtomicReference head = new AtomicReference<>(); - private final Set stored = new HashSet<>(); - private final Map chainIds = new HashMap<>(); - private PeerConnection peer; private P2pEventHandlerImpl p2pEventHandler; private ApplicationContext ctx; @@ -94,9 +55,6 @@ protected void afterInit() { @Override protected void beforeDestroy() { - if (recoveryService != null) { - recoveryService.close(); - } for (PeerConnection p : PeerManager.getPeers()) { PeerManager.remove(p.getChannel()); } @@ -446,230 +404,4 @@ public void testHandleSyncBlock() throws Exception { Assert.assertTrue(requestBlockIds.getIfPresent(blockId) == null); } - - @Test - public void testRejectedSummaryRecoversPreviouslyDiscardedBlocks() throws Exception { - initRecovery(); - BlockCapsule parent = block(101, head.get()); - BlockCapsule child = block(102, parent.getBlockId()); - BlockMsgHandler blocks = new BlockMsgHandler(); - ReflectUtils.setFieldValue(blocks, "tronNetDelegate", delegate); - ReflectUtils.setFieldValue(blocks, "syncService", recoveryService); - ReflectUtils.setFieldValue(blocks, "fetchBlockService", mock(FetchBlockService.class)); - ReflectUtils.setFieldValue(blocks, "fastForward", false); - Item childItem = new Item(child.getBlockId(), InventoryType.BLOCK); - failed.getAdvInvReceive().put(childItem, System.currentTimeMillis() - 1); - failed.getAdvInvRequest().put(childItem, System.currentTimeMillis()); - - blocks.processMessage(failed, new BlockMessage(child)); - Assert.assertFalse(failed.getAdvInvRequest().containsKey(childItem)); - Assert.assertFalse(stored.contains(child.getBlockId())); - ChainInventoryMsgHandler inventory = new ChainInventoryMsgHandler(); - ReflectUtils.setFieldValue(inventory, "tronNetDelegate", delegate); - ReflectUtils.setFieldValue(inventory, "syncService", recoveryService); - P2pEventHandlerImpl events = new P2pEventHandlerImpl(); - ReflectUtils.setFieldValue(events, "chainInventoryMsgHandler", inventory); - ChainInventoryMessage echo = new ChainInventoryMessage( - Collections.singletonList(head.get()), 0L); - ReflectUtils.invokeMethod(events, "processMessage", - new Class[]{PeerConnection.class, byte[].class}, failed, echo.getSendBytes()); - verify(failed).disconnect(ReasonCode.SYNC_FAIL); - - peers.remove(failed); - when(failed.getChannel().isDisconnect()).thenReturn(true); - ReflectUtils.setFieldValue(failed, "syncService", recoveryService); - ReflectUtils.setFieldValue(failed, "advService", mock(AdvService.class)); - failed.onDisconnect(); - runFetchTask(); - verify(replacement).sendMessage(any(SyncBlockChainMessage.class)); - - inventory.processMessage(replacement, new ChainInventoryMessage( - Arrays.asList(blockId(100), parent.getBlockId(), child.getBlockId()), 0L)); - runFetchTask(); - ArgumentCaptor requests = ArgumentCaptor.forClass(Message.class); - verify(replacement, times(2)).sendMessage(requests.capture()); - Assert.assertEquals(Arrays.asList(parent.getBlockId(), child.getBlockId()), - ((FetchInvDataMessage) requests.getValue()).getHashList()); - - blocks.processMessage(replacement, new BlockMessage(parent)); - blocks.processMessage(replacement, new BlockMessage(child)); - ReflectUtils.invokeMethod(recoveryService, "handleSyncBlock"); - inventory.processMessage(replacement, new ChainInventoryMessage( - Collections.singletonList(child.getBlockId()), 0L)); - - Assert.assertEquals(child.getBlockId(), head.get()); - Assert.assertTrue(stored.contains(parent.getBlockId())); - Assert.assertTrue(stored.contains(child.getBlockId())); - Assert.assertEquals(TronState.SYNC_COMPLETED, replacement.getTronState()); - Assert.assertFalse(replacement.isNeedSyncFromPeer()); - verify(replacement, never()).disconnect(any()); - } - - @Test - public void testRecoveryFiltersCandidatesAndAllowsPendingTransactions() throws Exception { - initRecovery(); - PeerConnection unhandshaken = readyPeer(18890); - unhandshaken.setHelloMessageReceive(null); - PeerConnection disconnected = readyPeer(18891); - when(disconnected.getChannel().isDisconnect()).thenReturn(true); - PeerConnection syncing = readyPeer(18892); - syncing.setTronState(TronState.SYNCING); - PeerConnection pendingSummary = readyPeer(18893); - pendingSummary.setSyncChainRequested(new Pair<>(new LinkedList<>( - Collections.singletonList(blockId(100))), System.currentTimeMillis())); - PeerConnection pendingBlock = readyPeer(18894); - pendingBlock.getSyncBlockRequested().put(blockId(101), System.currentTimeMillis()); - peers.addAll(Arrays.asList(unhandshaken, disconnected, syncing, pendingSummary, pendingBlock)); - replacement.getAdvInvRequest().put(new Item(Sha256Hash.ZERO_HASH, InventoryType.TRX), - System.currentTimeMillis()); - Assert.assertFalse(replacement.isIdle()); - queueRecovery(failed); - // The failed connection must remain excluded even if its state changes during cleanup. - failed.setTronState(TronState.SYNC_COMPLETED); - - runFetchTask(); - - verify(replacement).sendMessage(any(SyncBlockChainMessage.class)); - for (PeerConnection peer : Arrays.asList(failed, unhandshaken, disconnected, syncing, - pendingSummary, pendingBlock)) { - verify(peer, never()).sendMessage(any()); - } - } - - @Test - public void testRecoveryWaitsForAnEligibleConnection() throws Exception { - initRecovery(); - peers.clear(); - queueRecovery(failed); - runFetchTask(); - Assert.assertTrue((boolean) ReflectUtils.getFieldObject(recoveryService, "fetchFlag")); - PeerConnection unhandshaken = readyPeer(18890); - unhandshaken.setHelloMessageReceive(null); - peers.add(unhandshaken); - runFetchTask(); - Assert.assertTrue((boolean) ReflectUtils.getFieldObject(recoveryService, "fetchFlag")); - peers.add(replacement); - - runFetchTask(); - runFetchTask(); - - Assert.assertFalse((boolean) ReflectUtils.getFieldObject(recoveryService, "fetchFlag")); - verify(replacement, times(1)).sendMessage(any(SyncBlockChainMessage.class)); - verify(unhandshaken, never()).sendMessage(any()); - } - - @Test - public void testCompletedPeerDisconnectDoesNotStartRecovery() throws Exception { - initRecovery(); - recoveryService.onDisconnect(failed); - - runFetchTask(); - - verify(replacement, never()).sendMessage(any()); - } - - @Test - public void testRecoveryStartsOnlyOneRandomCandidate() throws Exception { - initRecovery(); - PeerConnection other = readyPeer(18890); - peers.add(other); - AtomicInteger requests = new AtomicInteger(); - for (PeerConnection peer : Arrays.asList(replacement, other)) { - doAnswer(call -> { - requests.incrementAndGet(); - return null; - }).when(peer).sendMessage(any(SyncBlockChainMessage.class)); - } - queueRecovery(failed); - - runFetchTask(); - runFetchTask(); - - Assert.assertEquals(1, requests.get()); - } - - @Test - public void testAnotherFailedSourceCanTriggerRecovery() throws Exception { - initRecovery(); - queueRecovery(failed); - runFetchTask(); - PeerConnection next = readyPeer(18890); - peers.remove(replacement); - peers.add(next); - - recoveryService.onDisconnect(replacement); - runFetchTask(); - - verify(next).sendMessage(any(SyncBlockChainMessage.class)); - } - - private void initRecovery() throws Exception { - recoveryService = new SyncService(); - delegate = mock(TronNetDelegate.class); - failed = readyPeer(18888); - replacement = readyPeer(18889); - peers = new ArrayList<>(Arrays.asList(failed, replacement)); - head.set(blockId(100)); - stored.add(blockId(0)); - stored.add(head.get()); - chainIds.put(0L, blockId(0)); - chainIds.put(100L, head.get()); - when(delegate.getActivePeer()).thenReturn(peers); - when(delegate.getHeadBlockId()).thenAnswer(call -> head.get()); - when(delegate.getKhaosDbHeadBlockId()).thenAnswer(call -> head.get()); - when(delegate.getGenesisBlockId()).thenReturn(blockId(0)); - when(delegate.getSolidBlockId()).thenReturn(blockId(0)); - when(delegate.getBlockTime(any())).thenReturn(System.currentTimeMillis() - 30_000_000L); - when(delegate.getBlockIdByNum(anyLong())).thenAnswer(call -> { - long number = call.getArgument(0); - return chainIds.getOrDefault(number, blockId(number)); - }); - when(delegate.containBlock(any())).thenAnswer(call -> stored.contains(call.getArgument(0))); - when(delegate.containBlockInMainChain(any())) - .thenAnswer(call -> stored.contains(call.getArgument(0))); - when(delegate.getBlockLock()).thenReturn(new Object()); - when(delegate.getForkLock()).thenReturn(new Object()); - when(delegate.validAdvBlock(any(), any())).thenReturn(true); - doAnswer(call -> { - BlockCapsule block = call.getArgument(0); - stored.add(block.getBlockId()); - chainIds.put(block.getNum(), block.getBlockId()); - head.set(block.getBlockId()); - return null; - }).when(delegate).processBlock(any(), anyBoolean()); - ReflectUtils.setFieldValue(recoveryService, "tronNetDelegate", delegate); - ReflectUtils.setFieldValue(recoveryService, "pbftDataSyncHandler", - mock(PbftDataSyncHandler.class)); - ReflectUtils.setFieldValue(recoveryService, "maxPendingBlockSize", 10); - } - - private void queueRecovery(PeerConnection peer) { - peer.setTronState(TronState.SYNCING); - peer.setNeedSyncFromPeer(true); - recoveryService.onDisconnect(peer); - Assert.assertTrue((boolean) ReflectUtils.getFieldObject(recoveryService, "fetchFlag")); - } - - private void runFetchTask() { - if ((boolean) ReflectUtils.getFieldObject(recoveryService, "fetchFlag")) { - recoveryService.setFetchFlag(false); - ReflectUtils.invokeMethod(recoveryService, "startFetchSyncBlock"); - } - } - - private static PeerConnection readyPeer(int port) throws Exception { - PeerConnection peer = PeerSyncTestSupport.peer(port); - peer.setHelloMessageReceive(helloMessage(100)); - peer.setTronState(TronState.SYNC_COMPLETED); - peer.setNeedSyncFromPeer(false); - peer.setNeedSyncFromUs(false); - return peer; - } - - private static BlockCapsule block(long number, BlockId parent) { - BlockCapsule block = new BlockCapsule(number, parent, System.currentTimeMillis() - 1_000, - ByteString.copyFromUtf8("witness")); - block.setMerkleRoot(); - return block; - } } From daa5bd4e1857265fe504bf1d37d1df98178c4b39 Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Tue, 3 Nov 2026 14:09:24 +0800 Subject: [PATCH 03/10] refactor(net): clarify advertised inventory block number Rename advertisedNum to maxAdvInvNum and adjust the inventory validation comment and stream formatting without changing validation behavior. --- .../net/messagehandler/ChainInventoryMsgHandler.java | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java index 9a7b05765bb..1dc56f308b5 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 @@ -168,16 +168,17 @@ private void check(PeerConnection peer, ChainInventoryMessage msg) throws P2pExc } long requestedAt = requested.getValue(); // Only use recent announcements received before this request, not a newer head - // announced while the peer was preparing its response. - long advertisedNum = peer.getAdvInvReceive().asMap().entrySet().stream() + // 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 < advertisedNum) { + .max() + .orElse(lastNum); + if (lastNum < maxAdvInvNum) { throw new P2pException(TypeEnum.SYNC_FAILED, - "lastNum " + lastNum + " (remainNum=0) is below advertised block " + advertisedNum); + "lastNum " + lastNum + " (remainNum=0) is below advertised block " + maxAdvInvNum); } } } From 3916a58ae13afef9b2d8a96427f53b5f967983a8 Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Tue, 3 Nov 2026 16:56:03 +0800 Subject: [PATCH 04/10] format the code --- .../tron/core/net/messagehandler/ChainInventoryMsgHandler.java | 2 ++ 1 file changed, 2 insertions(+) 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 1dc56f308b5..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 @@ -160,12 +160,14 @@ private void check(PeerConnection peer, ChainInventoryMessage msg) throws P2pExc 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 From 722061c2d868c4f082f32b3dba8ec3b3b5554c32 Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Tue, 3 Nov 2026 18:55:50 +0800 Subject: [PATCH 05/10] adjust executuion timeout of VerifyTransferProof from 80ms to 200ms --- .../common/runtime/vm/PrecompiledContractsVerifyProofTest.java | 2 +- .../src/test/java/org/tron/core/ShieldedTRC20BuilderTest.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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; } From 9b52e54d30bcf5c8abc599d3cf928bdaf170bbc2 Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Wed, 4 Nov 2026 17:12:46 +0800 Subject: [PATCH 06/10] fix(net): shorten isolation recovery delay --- .../tron/core/net/service/effective/ResilienceService.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) 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() { From 0c3d791465f5e4abea0c827f097373da9ca3754c Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Wed, 11 Nov 2026 17:05:17 +0800 Subject: [PATCH 07/10] fix(net): allow terminal sync replies with missing blocks Reject replies below recent advertisements only when every returned BlockId is already known locally. Preserve fetching for missing blocks, including fork blocks, while retaining the hello-head and summary checks. Cover known multi-block replies and missing canonical or fork blocks. --- .../ChainInventoryMsgHandler.java | 4 +- .../ChainInventoryMsgHandlerTest.java | 58 ++++++++++++++++++- 2 files changed, 58 insertions(+), 4 deletions(-) diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java index ce58369b89d..ffdf0b2519d 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 @@ -178,7 +178,9 @@ private void check(PeerConnection peer, ChainInventoryMessage msg) throws P2pExc .mapToLong(entry -> new BlockId(entry.getKey().getHash()).getNum()) .max() .orElse(lastNum); - if (lastNum < maxAdvInvNum) { + // Allow the existing fetch path to try any missing blocks before declaring no progress. + if (lastNum < maxAdvInvNum + && blockIds.stream().allMatch(tronNetDelegate::containBlock)) { throw new P2pException(TypeEnum.SYNC_FAILED, "lastNum " + lastNum + " (remainNum=0) is below advertised block " + maxAdvInvNum); } 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 3a0acbff866..1f46ce756cb 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 @@ -193,10 +193,9 @@ public void testSingleBlockBelowHelloHeadIsRejected() throws Exception { } @Test - public void testTerminalResponseBelowPriorBlockInventoryIsRejected() throws Exception { + public void testKnownTerminalResponseBelowPriorBlockInventoryIsRejected() throws Exception { for (List ids : Arrays.asList(Collections.singletonList(blockId(100)), - Arrays.asList(blockId(99), blockId(100)), - Arrays.asList(blockId(100), blockId(101)))) { + Arrays.asList(blockId(99), blockId(100)))) { resetFixture(); request(blockId(0), blockId(50), ids.get(0)); peer.getAdvInvReceive().put(new Item(blockId(102), InventoryType.BLOCK), @@ -210,6 +209,59 @@ public void testTerminalResponseBelowPriorBlockInventoryIsRejected() throws Exce } } + @Test + public void testTerminalResponseBelowPriorInventoryFetchesMissingBlocks() throws Exception { + peer.getAdvInvReceive().put(new Item(blockId(102), 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 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), From f74542d978f9b3d89eae7b680373b693af6f2fac Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Wed, 11 Nov 2026 17:16:25 +0800 Subject: [PATCH 08/10] fix(net): simplify terminal sync block check Check only the terminal BlockId when rejecting replies below prior block advertisements. Include locally known fork blocks and preserve fetching when the terminal block is unknown. Cover replies with unknown intermediate blocks and known terminal blocks on the main chain or a fork. --- .../ChainInventoryMsgHandler.java | 5 ++-- .../ChainInventoryMsgHandlerTest.java | 23 +++++++++++++++++++ 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java index ffdf0b2519d..a9a54283ae1 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 @@ -178,9 +178,8 @@ private void check(PeerConnection peer, ChainInventoryMessage msg) throws P2pExc .mapToLong(entry -> new BlockId(entry.getKey().getHash()).getNum()) .max() .orElse(lastNum); - // Allow the existing fetch path to try any missing blocks before declaring no progress. - if (lastNum < maxAdvInvNum - && blockIds.stream().allMatch(tronNetDelegate::containBlock)) { + // Allow fetching when the terminal block is still unknown locally, including on a fork. + if (lastNum < maxAdvInvNum && tronNetDelegate.containBlock(lastBlockId)) { throw new P2pException(TypeEnum.SYNC_FAILED, "lastNum " + lastNum + " (remainNum=0) is below advertised block " + maxAdvInvNum); } 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 1f46ce756cb..1bb06ea291b 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 @@ -242,6 +242,29 @@ public void testKnownTerminalBlockOutsideSummaryBelowPriorInventoryIsRejected() 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); From dfa9b98cf1a4194fbb151334901798c0e59fe203 Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Thu, 12 Nov 2026 13:34:28 +0800 Subject: [PATCH 09/10] fix(net): separate isolation recovery from long bans Use SYNC_FAIL after 30 seconds without block progress. Apply BAD_PROTOCOL only after 60 seconds and limit long-ban rounds to one every 30 seconds. Record the last ban time only after disconnecting a candidate. Cover cooldown, ban cadence, empty rounds and recovery after block progress. --- .../service/effective/ResilienceService.java | 26 ++- .../net/services/ResilienceServiceTest.java | 177 +++++++++++++++++- 2 files changed, 198 insertions(+), 5 deletions(-) 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 6cdddcae5c4..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 @@ -33,6 +33,9 @@ public class ResilienceService { CommonParameter.getInstance().getInactiveThreshold() * 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; @@ -41,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; @@ -160,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()) @@ -171,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, @@ -197,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/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; + } +} From e7222476c55aac68479653f8003fba6e66bfd30d Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Thu, 12 Nov 2026 17:30:06 +0800 Subject: [PATCH 10/10] fix(net): tolerate one block of lag in terminal sync replies Allow locally known terminal blocks to trail recent block advertisements by one block. Keep rejecting gaps of two or more blocks and preserve fetching for unknown terminal blocks. Ignore negative advertised heights to avoid subtraction overflow. Cover singleton and multi-block boundaries, invalid heights and missing-block fetching. --- .../ChainInventoryMsgHandler.java | 5 ++- .../ChainInventoryMsgHandlerTest.java | 45 ++++++++++++++++++- 2 files changed, 47 insertions(+), 3 deletions(-) diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java index a9a54283ae1..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 @@ -176,10 +176,11 @@ private void check(PeerConnection peer, ChainInventoryMessage msg) throws P2pExc .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); - // Allow fetching when the terminal block is still unknown locally, including on a fork. - if (lastNum < maxAdvInvNum && tronNetDelegate.containBlock(lastBlockId)) { + // 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/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java index 1bb06ea291b..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 @@ -209,9 +209,52 @@ public void testKnownTerminalResponseBelowPriorBlockInventoryIsRejected() throws } } + @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(102), InventoryType.BLOCK), + peer.getAdvInvReceive().put(new Item(blockId(103), InventoryType.BLOCK), peer.getSyncChainRequested().getValue() - 1); respond(0, blockId(100), blockId(101));