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 1/6] 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 2/6] 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 3/6] 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 4/6] 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 5/6] 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 6/6] 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() {