From 5c6104071bbd4b7509c7192014004fa0a1d09388 Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Thu, 12 Nov 2026 21:44:15 +0800 Subject: [PATCH 1/2] fix(net): improve block source selection and disconnect recovery --- .../tron/core/net/peer/PeerConnection.java | 6 + .../tron/core/net/service/adv/AdvService.java | 88 +++- .../service/fetchblock/FetchBlockService.java | 40 +- .../net/PeerDisconnectConcurrencyTest.java | 2 + .../fetchblock/BlockFetchSchedulingTest.java | 455 ++++++++++++++++++ .../core/net/services/AdvServiceTest.java | 3 + .../net/services/BlockFetchFailureTest.java | 12 + .../net/services/FetchBlockServiceTest.java | 12 + 8 files changed, 596 insertions(+), 22 deletions(-) create mode 100644 framework/src/test/java/org/tron/core/net/service/fetchblock/BlockFetchSchedulingTest.java diff --git a/framework/src/main/java/org/tron/core/net/peer/PeerConnection.java b/framework/src/main/java/org/tron/core/net/peer/PeerConnection.java index e5f7e54a19a..48fe85378fe 100644 --- a/framework/src/main/java/org/tron/core/net/peer/PeerConnection.java +++ b/framework/src/main/java/org/tron/core/net/peer/PeerConnection.java @@ -256,6 +256,12 @@ public boolean isIdle() { return advInvRequest.isEmpty() && isSyncIdle(); } + public boolean isBlockFetchIdle() { + return advInvRequest.keySet().stream() + .noneMatch(item -> item.getType() == Protocol.Inventory.InventoryType.BLOCK) + && isSyncIdle() && syncBlockInProcess.isEmpty(); + } + public boolean isSyncIdle() { return syncBlockRequested.isEmpty() && syncChainRequested == null; } diff --git a/framework/src/main/java/org/tron/core/net/service/adv/AdvService.java b/framework/src/main/java/org/tron/core/net/service/adv/AdvService.java index f11b4d83bd9..922d8aade71 100644 --- a/framework/src/main/java/org/tron/core/net/service/adv/AdvService.java +++ b/framework/src/main/java/org/tron/core/net/service/adv/AdvService.java @@ -8,12 +8,15 @@ import com.google.common.cache.CacheBuilder; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.Comparator; import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map.Entry; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -253,22 +256,59 @@ public void fastForward(BlockMessage msg) { */ public void onDisconnect(PeerConnection peer) { - // Queue retries for the fetch executor; disconnect callbacks may hold lifecycle locks. - if (!peer.getAdvInvRequest().isEmpty()) { - peer.getAdvInvRequest().keySet().forEach(item -> { - if (tronNetDelegate.getActivePeer().stream() - .anyMatch(p -> !p.equals(peer) && p.getAdvInvReceive().getIfPresent(item) != null)) { - invToFetch.put(item, System.currentTimeMillis()); - } else { - invToFetchCache.invalidate(item); + List blocks = new ArrayList<>(); + peer.getAdvInvRequest().keySet().forEach(item -> { + if (item.getType() == InventoryType.BLOCK) { + // PeerConnection clears its requests after this callback. + blocks.add(item); + } else if (tronNetDelegate.getActivePeer().stream() + .anyMatch(p -> !p.equals(peer) && p.getAdvInvReceive().getIfPresent(item) != null)) { + invToFetch.put(item, System.currentTimeMillis()); + } else { + invToFetchCache.invalidate(item); + } + }); + // Disconnect callbacks may hold lifecycle locks; recover only on the fetch executor. + try { + fetchExecutor.execute(() -> { + try { + fetchBlockService.onDisconnect(peer); + blocks.forEach(item -> recoverFetchOnDisconnect(peer, item)); + } catch (Exception e) { + logger.error("Recover disconnected peer fetches failed", e); } }); + } catch (RejectedExecutionException e) { + // Transport disconnect callbacks can outlive the advertisement service during shutdown. + logger.debug("Skip fetch recovery after executor shutdown", e); } } - private void consumerInvToFetch() { + private synchronized void recoverFetchOnDisconnect(PeerConnection peer, Item item) { Collection peers = tronNetDelegate.getActivePeer().stream() - .filter(peer -> peer.isIdle()) + .filter(p -> !p.equals(peer) && !p.isDisconnect()).collect(Collectors.toList()); + if (blockCache.getIfPresent(item) != null + || tronNetDelegate.containBlock(new BlockId(item.getHash()))) { + return; + } + Optional pending = peers.stream() + .filter(p -> p.getAdvInvRequest().containsKey(item)).findFirst(); + if (pending.isPresent()) { + // Reuse the original request and deadline without sending it again. + fetchBlockService.fetchBlock(Collections.singletonList(item.getHash()), pending.get()); + return; + } + if (peers.stream().anyMatch(p -> p.getAdvInvReceive().getIfPresent(item) != null)) { + invToFetch.put(item, System.currentTimeMillis()); + } else { + invToFetch.remove(item); + invToFetchCache.invalidate(item); + } + } + + private void consumerInvToFetch() { + Collection peers = tronNetDelegate.getActivePeer(); + Collection trxPeers = peers.stream().filter(PeerConnection::isIdle) .collect(Collectors.toList()); InvSender invSender = new InvSender(); synchronized (this) { @@ -276,6 +316,7 @@ private void consumerInvToFetch() { return; } long now = System.currentTimeMillis(); + List blocks = new ArrayList<>(); invToFetch.forEach((item, time) -> { if (time < now - TIMEOUT) { logger.info("This obj is too late to fetch, type: {} hash: {}", item.getType(), @@ -284,7 +325,11 @@ private void consumerInvToFetch() { invToFetchCache.invalidate(item); return; } - peers.stream().filter(peer -> { + if (item.getType() == InventoryType.BLOCK) { + blocks.add(item); + return; + } + trxPeers.stream().filter(peer -> { Long t = peer.getAdvInvReceive().getIfPresent(item); return t != null && now - t < TIMEOUT && invSender.getSize(peer) < MAX_TRX_FETCH_PER_PEER; }).sorted(Comparator.comparingInt(peer -> invSender.getSize(peer))) @@ -295,6 +340,27 @@ private void consumerInvToFetch() { invToFetch.remove(item); }); }); + + // Reserve sources for earlier blocks before a later block makes a source busy. + blocks.sort(Comparator.comparingLong(item -> new BlockId(item.getHash()).getNum())); + blocks.forEach(item -> { + if (blockCache.getIfPresent(item) != null + || tronNetDelegate.containBlock(new BlockId(item.getHash())) + || peers.stream().anyMatch(peer -> peer.getAdvInvRequest().containsKey(item))) { + invToFetch.remove(item); + return; + } + peers.stream() + .filter(peer -> fetchBlockService.canFetchBlock(peer, item, now)) + .min(Comparator.comparing((PeerConnection peer) -> !peer.isIdle()) + .thenComparingInt(invSender::getSize)) + .ifPresent(peer -> { + if (peer.checkAndPutAdvInvRequest(item, now)) { + invSender.add(item, peer); + invToFetch.remove(item); + } + }); + }); } invSender.sendFetch(); diff --git a/framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java b/framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java index d4a89dc846d..d3e08e3583b 100644 --- a/framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java +++ b/framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java @@ -18,6 +18,7 @@ import org.tron.common.utils.Sha256Hash; import org.tron.core.ChainBaseManager; import org.tron.core.capsule.BlockCapsule; +import org.tron.core.config.Parameter.NetConstants; import org.tron.core.net.TronNetDelegate; import org.tron.core.net.message.adv.FetchInvDataMessage; import org.tron.core.net.peer.Item; @@ -59,7 +60,7 @@ public void close() { ExecutorServiceManager.shutdownAndAwaitTermination(fetchBlockWorkerExecutor, esName); } - public void fetchBlock(List sha256HashList, PeerConnection peer) { + public synchronized void fetchBlock(List sha256HashList, PeerConnection peer) { if (sha256HashList.size() > 0) { logger.info("Begin fetch block {} from {}", new BlockCapsule.BlockId(sha256HashList.get(0)).getString(), @@ -71,16 +72,18 @@ public void fetchBlock(List sha256HashList, PeerConnection peer) { sha256HashList.stream().filter(sha256Hash -> new BlockCapsule.BlockId(sha256Hash).getNum() == chainBaseManager.getHeadBlockNum() + 1) .findFirst().ifPresent(sha256Hash -> { - long now = System.currentTimeMillis(); - fetchBlockInfo = new FetchBlockInfo(sha256Hash, peer, now); - Metrics.counterInc(MetricKeys.Counter.BLOCK_FETCH_ARMED, 1); - logger.info("Set fetchBlockInfo, block: {}, peer: {}, time: {}", sha256Hash, - peer.getInetAddress(), now); + Long requestTime = peer.getAdvInvRequest().get(new Item(sha256Hash, InventoryType.BLOCK)); + if (requestTime != null) { + fetchBlockInfo = new FetchBlockInfo(sha256Hash, peer, requestTime); + Metrics.counterInc(MetricKeys.Counter.BLOCK_FETCH_ARMED, 1); + logger.info("Set fetchBlockInfo, block: {}, peer: {}, time: {}", sha256Hash, + peer.getInetAddress(), requestTime); + } }); } - public void blockFetchSuccess(Sha256Hash sha256Hash) { + public synchronized void blockFetchSuccess(Sha256Hash sha256Hash) { logger.info("Fetch block success, {}", new BlockCapsule.BlockId(sha256Hash).getString()); FetchBlockInfo fetchBlockInfoTemp = this.fetchBlockInfo; if (null == fetchBlockInfoTemp || !fetchBlockInfoTemp.getHash().equals(sha256Hash)) { @@ -89,15 +92,30 @@ public void blockFetchSuccess(Sha256Hash sha256Hash) { this.fetchBlockInfo = null; } - private void fetchBlockProcess(FetchBlockInfo fetchBlock) { - if (null == fetchBlock) { + public synchronized void onDisconnect(PeerConnection peer) { + if (fetchBlockInfo != null && fetchBlockInfo.getPeer().equals(peer)) { + fetchBlockInfo = null; + } + } + + public boolean canFetchBlock(PeerConnection peer, Item item, long now) { + if (peer.isDisconnect() || peer.isNeedSyncFromPeer() || peer.isNeedSyncFromUs() + || !peer.isBlockFetchIdle()) { + return false; + } + Long received = peer.getAdvInvReceive().getIfPresent(item); + return received != null && received >= now - NetConstants.ADV_TIME_OUT; + } + + private synchronized void fetchBlockProcess(FetchBlockInfo fetchBlock) { + if (null == fetchBlock || fetchBlockInfo != fetchBlock) { return; } Item item = new Item(fetchBlock.getHash(), InventoryType.BLOCK); + long now = System.currentTimeMillis(); Optional optionalPeerConnection = tronNetDelegate.getActivePeer().stream() - .filter(PeerConnection::isIdle) .filter(filterPeer -> !filterPeer.equals(fetchBlock.getPeer())) - .filter(filterPeer -> filterPeer.getAdvInvReceive().getIfPresent(item) != null) + .filter(filterPeer -> canFetchBlock(filterPeer, item, now)) // Seeded estimates are clamped to the fetch timeout; the unseeded channel-latency // fallback is not, but min() ordering and the saturation gate keep it safe. .min(Comparator.comparingDouble(this::getPeerLatency)); diff --git a/framework/src/test/java/org/tron/core/net/PeerDisconnectConcurrencyTest.java b/framework/src/test/java/org/tron/core/net/PeerDisconnectConcurrencyTest.java index 18355ee9b1a..0c6340a1b59 100644 --- a/framework/src/test/java/org/tron/core/net/PeerDisconnectConcurrencyTest.java +++ b/framework/src/test/java/org/tron/core/net/PeerDisconnectConcurrencyTest.java @@ -40,6 +40,7 @@ import org.tron.core.net.peer.PeerManager; import org.tron.core.net.service.adv.AdvService; import org.tron.core.net.service.effective.EffectiveCheckService; +import org.tron.core.net.service.fetchblock.FetchBlockService; import org.tron.core.net.service.handshake.HandshakeService; import org.tron.p2p.connection.Channel; import org.tron.protos.Protocol.Inventory.InventoryType; @@ -219,6 +220,7 @@ public void sendMessage(Message message) { Mockito.when(delegate.getActivePeer()).thenReturn(Collections.singletonList(replacement)); AdvService adv = new AdvService(); ReflectUtils.setFieldValue(adv, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(adv, "fetchBlockService", mock(FetchBlockService.class)); ExecutorService worker = Executors.newSingleThreadExecutor(); try { synchronized (adv) { diff --git a/framework/src/test/java/org/tron/core/net/service/fetchblock/BlockFetchSchedulingTest.java b/framework/src/test/java/org/tron/core/net/service/fetchblock/BlockFetchSchedulingTest.java new file mode 100644 index 00000000000..1c5c65df0b1 --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/service/fetchblock/BlockFetchSchedulingTest.java @@ -0,0 +1,455 @@ +package org.tron.core.net.service.fetchblock; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Method; +import java.net.InetSocketAddress; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.tron.common.utils.Pair; +import org.tron.common.utils.ReflectUtils; +import org.tron.common.utils.Sha256Hash; +import org.tron.core.ChainBaseManager; +import org.tron.core.capsule.BlockCapsule; +import org.tron.core.capsule.BlockCapsule.BlockId; +import org.tron.core.config.Parameter.NetConstants; +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.peer.Item; +import org.tron.core.net.peer.PeerConnection; +import org.tron.core.net.service.adv.AdvService; +import org.tron.core.net.service.sync.SyncService; +import org.tron.p2p.connection.Channel; +import org.tron.protos.Protocol.Inventory.InventoryType; + +public class BlockFetchSchedulingTest { + + private FetchBlockService fetch; + private AdvService adv; + private TronNetDelegate delegate; + private PeerConnection first; + private PeerConnection second; + private List peers; + private Item item; + + @Before + public void setUp() { + fetch = new FetchBlockService(); + adv = new AdvService(); + delegate = mock(TronNetDelegate.class); + ChainBaseManager chain = mock(ChainBaseManager.class); + when(chain.getHeadBlockNum()).thenReturn(100L); + first = peer(18888); + second = peer(18889); + peers = new ArrayList<>(Arrays.asList(first, second)); + item = blockItem(101); + when(delegate.getActivePeer()).thenReturn(peers); + ReflectUtils.setFieldValue(fetch, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(fetch, "chainBaseManager", chain); + ReflectUtils.setFieldValue(adv, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(adv, "fetchBlockService", fetch); + } + + @After + public void tearDown() { + adv.close(); + fetch.close(); + } + + @Test + public void pendingTransactionsAllowInitialBlockFetch() { + first.getAdvInvRequest().put(transaction(), System.currentTimeMillis()); + announce(first, item); + + Assert.assertTrue(adv.addInv(item)); + + Assert.assertTrue(first.getAdvInvRequest().containsKey(transaction())); + Assert.assertTrue(first.getAdvInvRequest().containsKey(item)); + verify(first).sendMessage(any(FetchInvDataMessage.class)); + } + + @Test + public void initialFetchPrefersFullyIdleSourceOverFasterTransactionSource() { + first.getAdvInvRequest().put(transaction(), System.currentTimeMillis()); + when(first.getChannel().getAvgLatency()).thenReturn(1L); + when(second.getChannel().getAvgLatency()).thenReturn(500L); + announce(first, item); + announce(second, item); + + adv.addInv(item); + + verify(first, never()).sendMessage(any()); + verify(second).sendMessage(any(FetchInvDataMessage.class)); + } + + @Test + public void backupPrefersLowerLatencyEvenWithPendingTransactions() throws Exception { + PeerConnection slower = peer(18890); + peers.add(slower); + when(second.getChannel().getAvgLatency()).thenReturn(10L); + when(slower.getChannel().getAvgLatency()).thenReturn(100L); + second.getAdvInvRequest().put(transaction(), System.currentTimeMillis()); + announce(second, item); + announce(slower, item); + beginFetch(first, System.currentTimeMillis() - 2_000); + + tick(state()); + + verify(second).sendMessage(any(FetchInvDataMessage.class)); + verify(slower, never()).sendMessage(any()); + Assert.assertNull("Keep the baseline single-backup policy", state()); + } + + @Test + public void blockFetchExcludesPendingBlocksAndSyncWork() { + announce(first, item); + long now = System.currentTimeMillis(); + first.getAdvInvRequest().put(transaction(), now); + Assert.assertFalse(first.isIdle()); + Assert.assertTrue(fetch.canFetchBlock(first, item, now)); + first.getAdvInvRequest().put(blockItem(102), now); + Assert.assertFalse(first.isBlockFetchIdle()); + first.getAdvInvRequest().remove(blockItem(102)); + first.getSyncBlockRequested().put(new BlockId(), now); + Assert.assertFalse(fetch.canFetchBlock(first, item, now)); + first.getSyncBlockRequested().clear(); + first.setSyncChainRequested(new Pair<>(new ArrayDeque<>(), now)); + Assert.assertFalse(fetch.canFetchBlock(first, item, now)); + first.setSyncChainRequested(null); + first.getSyncBlockInProcess().add(new BlockId()); + Assert.assertFalse(fetch.canFetchBlock(first, item, now)); + first.getSyncBlockInProcess().clear(); + first.setNeedSyncFromPeer(true); + Assert.assertFalse(fetch.canFetchBlock(first, item, now)); + first.setNeedSyncFromPeer(false); + first.setNeedSyncFromUs(true); + Assert.assertFalse(fetch.canFetchBlock(first, item, now)); + first.setNeedSyncFromUs(false); + when(first.getChannel().isDisconnect()).thenReturn(true); + Assert.assertFalse(fetch.canFetchBlock(first, item, now)); + } + + @Test + public void blockFetchRequiresRecentAnnouncement() { + long now = System.currentTimeMillis(); + Assert.assertFalse(fetch.canFetchBlock(first, item, now)); + first.getAdvInvReceive().put(item, now - NetConstants.ADV_TIME_OUT - 1); + Assert.assertFalse(fetch.canFetchBlock(first, item, now)); + } + + @Test + public void queuedBlocksReserveSourcesInHeightOrder() throws Exception { + Item higher = blockItem(102); + announce(first, higher); + announce(first, item); + pending().put(higher, System.currentTimeMillis()); + pending().put(item, System.currentTimeMillis()); + + consume(); + + Assert.assertTrue(first.getAdvInvRequest().containsKey(item)); + Assert.assertFalse(first.getAdvInvRequest().containsKey(higher)); + Assert.assertTrue(pending().containsKey(higher)); + Assert.assertFalse(pending().containsKey(item)); + } + + @Test + public void unavailableEarlierBlockDoesNotPreventLaterFetch() throws Exception { + Item higher = blockItem(102); + announce(first, higher); + pending().put(item, System.currentTimeMillis()); + pending().put(higher, System.currentTimeMillis()); + + consume(); + + Assert.assertTrue(first.getAdvInvRequest().containsKey(higher)); + Assert.assertTrue(pending().containsKey(item)); + } + + @Test + public void transactionRequestsStillBatchTogether() throws Exception { + Item other = new Item(new BlockId(Sha256Hash.ZERO_HASH, 1), InventoryType.TRX); + announce(first, transaction()); + announce(first, other); + adv.addInv(transaction()); + adv.addInv(other); + + consume(); + + Assert.assertTrue(first.getAdvInvRequest().containsKey(transaction())); + Assert.assertTrue(first.getAdvInvRequest().containsKey(other)); + verify(first).sendMessage(any(FetchInvDataMessage.class)); + } + + @Test + public void disconnectReusesPendingBackupAndItsOriginalDeadline() throws Exception { + beginFetch(first, System.currentTimeMillis() - 2_000); + long backupTime = System.currentTimeMillis() - 1_000; + second.getAdvInvRequest().put(item, backupTime); + + disconnect(first); + consume(); + + Assert.assertSame(second, ReflectUtils.getFieldObject(state(), "peer")); + Assert.assertEquals(backupTime, ReflectUtils.getFieldObject(state(), "time")); + Assert.assertEquals(Long.valueOf(backupTime), second.getAdvInvRequest().get(item)); + Assert.assertFalse(pending().containsKey(item)); + verify(second, never()).sendMessage(any()); + } + + @Test + public void backupDisconnectRestoresOriginalPendingRequest() throws Exception { + beginFetch(first, System.currentTimeMillis() - 2_000); + long originalTime = first.getAdvInvRequest().get(item); + announce(second, item); + tick(state()); + Assert.assertNull(state()); + + disconnect(second); + + Assert.assertSame(first, ReflectUtils.getFieldObject(state(), "peer")); + Assert.assertEquals(originalTime, ReflectUtils.getFieldObject(state(), "time")); + verify(first, never()).sendMessage(any()); + } + + @Test + public void disconnectWaitsForBusyAnnouncerWithoutLosingRecovery() throws Exception { + announce(first, item); + adv.addInv(item); + announce(second, item); + second.getAdvInvRequest().put(blockItem(102), System.currentTimeMillis()); + + disconnect(first); + consume(); + Assert.assertTrue(pending().containsKey(item)); + verify(second, never()).sendMessage(any()); + second.getAdvInvRequest().clear(); + consume(); + + verify(second).sendMessage(any(FetchInvDataMessage.class)); + Assert.assertTrue(second.getAdvInvRequest().containsKey(item)); + } + + @Test + public void disconnectWithoutSourcesReleasesDeduplication() throws Exception { + announce(first, item); + adv.addInv(item); + + disconnect(first); + announce(second, item); + + Assert.assertNull(state()); + Assert.assertTrue(adv.addInv(item)); + verify(second).sendMessage(any(FetchInvDataMessage.class)); + } + + @Test + public void disconnectClearsTrackingAfterOutstandingRequestWasRemoved() throws Exception { + beginFetch(first, System.currentTimeMillis()); + first.getAdvInvRequest().clear(); + + disconnect(first); + + Assert.assertNull(state()); + } + + @Test + public void disconnectDoesNotRefetchStoredBlock() throws Exception { + beginFetch(first, System.currentTimeMillis()); + announce(second, item); + when(delegate.containBlock(new BlockId(item.getHash()))).thenReturn(true); + + disconnect(first); + consume(); + + Assert.assertNull(state()); + Assert.assertFalse(pending().containsKey(item)); + verify(second, never()).sendMessage(any()); + } + + @Test + public void disconnectDoesNotRefetchCachedBlock() throws Exception { + BlockCapsule block = new BlockCapsule(101, new BlockId(Sha256Hash.ZERO_HASH, 100), + System.currentTimeMillis(), Sha256Hash.ZERO_HASH.getByteString()); + item = new Item(block.getBlockId(), InventoryType.BLOCK); + beginFetch(first, System.currentTimeMillis()); + announce(second, item); + adv.broadcast(new BlockMessage(block)); + + disconnect(first); + consume(); + + Assert.assertNull(state()); + Assert.assertFalse(pending().containsKey(item)); + Assert.assertFalse(second.getAdvInvRequest().containsKey(item)); + verify(second, never()).sendMessage(any(FetchInvDataMessage.class)); + } + + @Test(timeout = 10_000) + public void disconnectSnapshotsRequestsWithoutWaitingForFetchExecutor() throws Exception { + CountDownLatch blocked = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + ScheduledExecutorService executor = fetchExecutor(); + executor.submit(() -> { + blocked.countDown(); + release.await(5, TimeUnit.SECONDS); + return null; + }); + ExecutorService callback = Executors.newSingleThreadExecutor(); + try { + Assert.assertTrue(blocked.await(5, TimeUnit.SECONDS)); + beginFetch(first, System.currentTimeMillis()); + announce(second, item); + peers.remove(first); + synchronized (adv) { + callback.submit(() -> adv.onDisconnect(first)).get(2, TimeUnit.SECONDS); + } + first.getAdvInvRequest().clear(); + Assert.assertTrue(pending().isEmpty()); + release.countDown(); + awaitRecovery(); + Assert.assertTrue(pending().containsKey(item)); + verify(second, never()).sendMessage(any()); + consume(); + verify(second).sendMessage(any(FetchInvDataMessage.class)); + } finally { + release.countDown(); + callback.shutdownNow(); + Assert.assertTrue(callback.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test + public void disconnectAfterServiceShutdownStillClearsPeerRequests() { + ReflectUtils.setFieldValue(first, "advService", adv); + ReflectUtils.setFieldValue(first, "syncService", mock(SyncService.class)); + beginFetch(first, System.currentTimeMillis()); + adv.close(); + + first.onDisconnect(); + + Assert.assertTrue(first.getAdvInvRequest().isEmpty()); + } + + @Test + public void completedRequestCannotBeRestoredByRecovery() { + beginFetch(first, System.currentTimeMillis()); + first.getAdvInvRequest().remove(item); + fetch.blockFetchSuccess(item.getHash()); + + fetch.fetchBlock(Collections.singletonList(item.getHash()), first); + + Assert.assertNull(state()); + } + + @Test + public void staleWorkerCannotClearRecoveredRequest() throws Exception { + beginFetch(first, System.currentTimeMillis() - 2_000); + Object oldState = state(); + second.getAdvInvRequest().put(item, System.currentTimeMillis()); + disconnect(first); + Object recovered = state(); + + tick(oldState); + + Assert.assertNotNull(recovered); + Assert.assertSame(recovered, state()); + verify(second, never()).sendMessage(any()); + } + + @Test + public void shortTimeoutStillClearsTrackingWithoutAlternative() throws Exception { + beginFetch(first, System.currentTimeMillis() - 2_000); + + tick(state()); + + Assert.assertNull(state()); + Assert.assertTrue(first.getAdvInvRequest().containsKey(item)); + } + + private PeerConnection peer(int port) { + PeerConnection peer = spy(new PeerConnection()); + Channel channel = mock(Channel.class); + InetSocketAddress address = new InetSocketAddress("127.0.0.1", port); + when(channel.getInetSocketAddress()).thenReturn(address); + when(channel.getInetAddress()).thenReturn(address.getAddress()); + ReflectUtils.setFieldValue(peer, "channel", channel); + peer.setNeedSyncFromPeer(false); + peer.setNeedSyncFromUs(false); + doNothing().when(peer).sendMessage(any()); + return peer; + } + + private Item blockItem(long height) { + return new Item(new BlockId(Sha256Hash.ZERO_HASH, height), InventoryType.BLOCK); + } + + private Item transaction() { + return new Item(Sha256Hash.ZERO_HASH, InventoryType.TRX); + } + + private void announce(PeerConnection peer, Item announced) { + peer.getAdvInvReceive().put(announced, System.currentTimeMillis()); + } + + private void beginFetch(PeerConnection peer, long time) { + peer.getAdvInvRequest().put(item, time); + fetch.fetchBlock(Collections.singletonList(item.getHash()), peer); + Assert.assertNotNull(state()); + } + + private void disconnect(PeerConnection peer) throws Exception { + peers.remove(peer); + when(peer.getChannel().isDisconnect()).thenReturn(true); + adv.onDisconnect(peer); + peer.getAdvInvRequest().clear(); + awaitRecovery(); + } + + private void awaitRecovery() throws Exception { + fetchExecutor().submit(() -> { }).get(5, TimeUnit.SECONDS); + } + + private ScheduledExecutorService fetchExecutor() { + return ReflectUtils.getFieldValue(adv, "fetchExecutor"); + } + + private Map pending() { + return ReflectUtils.getFieldValue(adv, "invToFetch"); + } + + private Object state() { + return ReflectUtils.getFieldObject(fetch, "fetchBlockInfo"); + } + + private void consume() { + ReflectUtils.invokeMethod(adv, "consumerInvToFetch"); + } + + private void tick(Object snapshot) throws Exception { + Method method = FetchBlockService.class.getDeclaredMethod("fetchBlockProcess", + Class.forName(FetchBlockService.class.getName() + "$FetchBlockInfo")); + method.setAccessible(true); + method.invoke(fetch, snapshot); + } +} diff --git a/framework/src/test/java/org/tron/core/net/services/AdvServiceTest.java b/framework/src/test/java/org/tron/core/net/services/AdvServiceTest.java index e4e66956236..ee0f97bd87d 100644 --- a/framework/src/test/java/org/tron/core/net/services/AdvServiceTest.java +++ b/framework/src/test/java/org/tron/core/net/services/AdvServiceTest.java @@ -44,6 +44,7 @@ import org.tron.core.net.peer.PeerManager; import org.tron.core.net.service.adv.AdvService; import org.tron.core.net.service.effective.EffectiveCheckService; +import org.tron.core.net.service.fetchblock.FetchBlockService; import org.tron.core.net.service.sync.SyncService; import org.tron.p2p.connection.Channel; import org.tron.protos.Protocol; @@ -218,6 +219,7 @@ public void sendMessage(Message message) { Mockito.when(delegate.getActivePeer()).thenReturn(Arrays.asList(first, second)); AdvService isolated = new AdvService(); ReflectUtils.setFieldValue(isolated, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(isolated, "fetchBlockService", mock(FetchBlockService.class)); try { BlockCapsule block = new BlockCapsule(1, Sha256Hash.ZERO_HASH, System.currentTimeMillis(), Sha256Hash.ZERO_HASH.getByteString()); @@ -254,6 +256,7 @@ public void sendMessage(Message message) { @Test(timeout = 20_000) public void inventorySendAllowsConcurrentDisconnectCleanup() throws Exception { AdvService isolated = new AdvService(); + ReflectUtils.setFieldValue(isolated, "fetchBlockService", mock(FetchBlockService.class)); P2pEventHandlerImpl handler = new P2pEventHandlerImpl(); ReflectUtils.setFieldValue(handler, "effectiveCheckService", mock(EffectiveCheckService.class)); SyncService syncService = mock(SyncService.class); diff --git a/framework/src/test/java/org/tron/core/net/services/BlockFetchFailureTest.java b/framework/src/test/java/org/tron/core/net/services/BlockFetchFailureTest.java index 3611b68ce5c..5e244ebdab6 100644 --- a/framework/src/test/java/org/tron/core/net/services/BlockFetchFailureTest.java +++ b/framework/src/test/java/org/tron/core/net/services/BlockFetchFailureTest.java @@ -2,8 +2,10 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.doCallRealMethod; import static org.mockito.Mockito.doNothing; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.doThrow; @@ -276,6 +278,7 @@ private void assertInactiveWitnessTiming(long[] times, boolean[] disconnect, public void advertisedScheduleFailureDoesNotDisconnectRelayingPeers() throws Exception { TronNetDelegate delegate = mock(TronNetDelegate.class); FetchBlockService fetch = mock(FetchBlockService.class); + doCallRealMethod().when(fetch).canFetchBlock(any(), any(), anyLong()); AdvService adv = new AdvService(); ReflectUtils.setFieldValue(adv, "tronNetDelegate", delegate); ReflectUtils.setFieldValue(adv, "fetchBlockService", fetch); @@ -285,6 +288,10 @@ public void advertisedScheduleFailureDoesNotDisconnectRelayingPeers() throws Exc ReflectUtils.setFieldValue(handler, "advService", adv); PeerConnection sender = newPeer(10001); PeerConnection other = newPeer(10002); + sender.setNeedSyncFromPeer(false); + sender.setNeedSyncFromUs(false); + other.setNeedSyncFromPeer(false); + other.setNeedSyncFromUs(false); BlockCapsule block = newBlock(); Item item = new Item(block.getBlockId(), InventoryType.BLOCK); when(delegate.getHeadBlockId()).thenReturn(new BlockId(Sha256Hash.ZERO_HASH, 0)); @@ -315,6 +322,7 @@ public void advertisedScheduleFailureDoesNotDisconnectRelayingPeers() throws Exc private void assertAdvertisedBlockDoesNotRetry(TypeEnum type) throws Exception { TronNetDelegate delegate = mock(TronNetDelegate.class); FetchBlockService fetch = mock(FetchBlockService.class); + doCallRealMethod().when(fetch).canFetchBlock(any(), any(), anyLong()); AdvService adv = new AdvService(); ReflectUtils.setFieldValue(adv, "tronNetDelegate", delegate); ReflectUtils.setFieldValue(adv, "fetchBlockService", fetch); @@ -323,6 +331,10 @@ private void assertAdvertisedBlockDoesNotRetry(TypeEnum type) throws Exception { ReflectUtils.setFieldValue(handler, "fetchBlockService", fetch); PeerConnection sender = newPeer(10001); PeerConnection other = newPeer(10002); + sender.setNeedSyncFromPeer(false); + sender.setNeedSyncFromUs(false); + other.setNeedSyncFromPeer(false); + other.setNeedSyncFromUs(false); BlockCapsule block = newBlock(); Item item = new Item(block.getBlockId(), InventoryType.BLOCK); when(delegate.getHeadBlockId()).thenReturn(new BlockId(Sha256Hash.ZERO_HASH, 0)); diff --git a/framework/src/test/java/org/tron/core/net/services/FetchBlockServiceTest.java b/framework/src/test/java/org/tron/core/net/services/FetchBlockServiceTest.java index b7a6ecdfcdc..7d451dd2eac 100644 --- a/framework/src/test/java/org/tron/core/net/services/FetchBlockServiceTest.java +++ b/framework/src/test/java/org/tron/core/net/services/FetchBlockServiceTest.java @@ -67,6 +67,8 @@ public void testSelectLowestLatencyPeer() throws Exception { oldPeer.updateFetchLatency(200L); PeerConnection newPeer = context.getBean(PeerConnection.class); newPeer.setChannel(newChannel); + newPeer.setNeedSyncFromPeer(false); + newPeer.setNeedSyncFromUs(false); newPeer.updateFetchLatency(50L); Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); @@ -132,6 +134,8 @@ public void testSwitchOnOldPeerTimeout() throws Exception { PeerConnection newPeer = context.getBean(PeerConnection.class); newPeer.setChannel(newChannel); + newPeer.setNeedSyncFromPeer(false); + newPeer.setNeedSyncFromUs(false); newPeer.updateFetchLatency(50L); Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); @@ -172,6 +176,8 @@ public void testSwitchOnHardTimeoutWhenOldPeerUnseeded() throws Exception { when(candidateChannel.getAvgLatency()).thenReturn(50L); ReflectUtils.setFieldValue(oldPeer, "channel", oldChannel); ReflectUtils.setFieldValue(candidate, "channel", candidateChannel); + candidate.setNeedSyncFromPeer(false); + candidate.setNeedSyncFromUs(false); candidate.updateFetchLatency(50L); doNothing().when(candidateChannel).send(any(byte[].class)); @@ -225,6 +231,8 @@ public void testNoSwitchWhenOldPeerLatencyUnknown() throws Exception { PeerConnection newPeer = context.getBean(PeerConnection.class); newPeer.setChannel(newChannel); + newPeer.setNeedSyncFromPeer(false); + newPeer.setNeedSyncFromUs(false); // first real sample replaces the unsampled state, then clamps 999 to 500 newPeer.updateFetchLatency(999L); @@ -286,6 +294,8 @@ public void testSwitchWhenBothPeersUnseededReadsChannelFallback() throws Excepti oldPeer.setChannel(oldChannel); PeerConnection newPeer = context.getBean(PeerConnection.class); newPeer.setChannel(newChannel); + newPeer.setNeedSyncFromPeer(false); + newPeer.setNeedSyncFromUs(false); Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); Item item = new Item(hash, InventoryType.BLOCK); @@ -332,6 +342,8 @@ public void testSwitchToUnseededCandidateWhenOldPeerSeeded() throws Exception { ReflectUtils.setFieldValue(oldPeer, "channel", oldChannel); oldPeer.updateFetchLatency(200L); ReflectUtils.setFieldValue(candidate, "channel", candidateChannel); + candidate.setNeedSyncFromPeer(false); + candidate.setNeedSyncFromUs(false); doNothing().when(candidateChannel).send(any(byte[].class)); Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); From d07048a634489f93fa9727ca86cfce880001f0f9 Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Thu, 12 Nov 2026 22:26:22 +0800 Subject: [PATCH 2/2] fix(net): validate block number and parent before fetch completion --- .../net/messagehandler/BlockMsgHandler.java | 26 +++--- .../BlockAlreadyKnownCounterTest.java | 2 +- .../messagehandler/BlockMsgHandlerTest.java | 79 +++++++++++++++++++ 3 files changed, 97 insertions(+), 10 deletions(-) diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/BlockMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/BlockMsgHandler.java index 3aca66362e5..0c0cc70a152 100644 --- a/framework/src/main/java/org/tron/core/net/messagehandler/BlockMsgHandler.java +++ b/framework/src/main/java/org/tron/core/net/messagehandler/BlockMsgHandler.java @@ -9,6 +9,7 @@ import org.springframework.stereotype.Component; import org.tron.common.prometheus.MetricKeys; import org.tron.common.prometheus.Metrics; +import org.tron.common.utils.Sha256Hash; import org.tron.core.Constant; import org.tron.core.capsule.BlockCapsule; import org.tron.core.capsule.BlockCapsule.BlockId; @@ -79,9 +80,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep throw new P2pException(TypeEnum.BAD_MESSAGE, "block time error"); } - if (!fastForward && !peer.isRelayPeer()) { - check(peer, blockMessage); - } + check(peer, blockMessage); blockMessage.sanitize(); @@ -131,12 +130,21 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep } private void check(PeerConnection peer, BlockMessage msg) throws P2pException { - Item item = new Item(msg.getBlockId(), InventoryType.BLOCK); - if (!peer.getSyncBlockRequested().containsKey(msg.getBlockId()) && !peer.getAdvInvRequest() - .containsKey(item)) { - logger.error("Receive bad block {} from peer {}, with no request", - msg.getBlockId(), peer.getInetSocketAddress()); - throw new P2pException(TypeEnum.BAD_MESSAGE, "no request"); + if (!fastForward && !peer.isRelayPeer()) { + Item item = new Item(msg.getBlockId(), InventoryType.BLOCK); + if (!peer.getSyncBlockRequested().containsKey(msg.getBlockId()) && !peer.getAdvInvRequest() + .containsKey(item)) { + logger.error("Receive bad block {} from peer {}, with no request", + msg.getBlockId(), peer.getInetSocketAddress()); + throw new P2pException(TypeEnum.BAD_MESSAGE, "no request"); + } + } + + BlockCapsule blockCapsule = msg.getBlockCapsule(); + if (blockCapsule.getNum() <= 0 + || blockCapsule.getParentHashStr().size() != Sha256Hash.LENGTH + || new BlockId(blockCapsule.getParentHash()).getNum() != blockCapsule.getNum() - 1) { + throw new P2pException(TypeEnum.BAD_BLOCK, "block number does not follow parent"); } } diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/BlockAlreadyKnownCounterTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/BlockAlreadyKnownCounterTest.java index c153a73b2e8..013ed1d6bba 100644 --- a/framework/src/test/java/org/tron/core/net/messagehandler/BlockAlreadyKnownCounterTest.java +++ b/framework/src/test/java/org/tron/core/net/messagehandler/BlockAlreadyKnownCounterTest.java @@ -163,7 +163,7 @@ public void testMatchedRequestHeadDuplicateIncrements() throws P2pException { } private BlockMessage newBlockMessage(long number) { - BlockCapsule capsule = new BlockCapsule(number, Sha256Hash.ZERO_HASH, + BlockCapsule capsule = new BlockCapsule(number, new BlockId(Sha256Hash.ZERO_HASH, number - 1), System.currentTimeMillis() - 60_000L, ByteString.copyFrom(ECKey.fromPrivate(BigInteger.TEN).getAddress())); return new BlockMessage(capsule); diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/BlockMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/BlockMsgHandlerTest.java index f8af0ea4c6e..02b1849cf5a 100644 --- a/framework/src/test/java/org/tron/core/net/messagehandler/BlockMsgHandlerTest.java +++ b/framework/src/test/java/org/tron/core/net/messagehandler/BlockMsgHandlerTest.java @@ -149,6 +149,85 @@ public void testProcessMessage() { } } + @Test + public void testValidFastForwardAndRelayBlocksDoNotRequireRequest() throws Exception { + for (boolean fastForward : new boolean[]{false, true}) { + BlockCapsule block = signedWitnessBlock(); + TronNetDelegate delegate = witnessDelegate(false, block.getWitnessAddress()); + BlockMsgHandler ingress = witnessHandler(delegate, Mockito.mock(AdvService.class), + Mockito.mock(SyncService.class)); + ReflectUtils.setFieldValue(ingress, "fastForward", fastForward); + ReflectUtils.setFieldValue(ingress, "relayService", Mockito.mock(RelayService.class)); + PeerConnection sender = witnessPeer(); + sender.setRelayPeer(!fastForward); + + ingress.processMessage(sender, new BlockMessage(block)); + + Mockito.verify(delegate).processBlock(any(BlockCapsule.class), Mockito.eq(false)); + Assert.assertTrue(sender.getAdvInvRequest().isEmpty()); + Assert.assertTrue(sender.getSyncBlockRequested().isEmpty()); + } + } + + @Test + public void testInvalidBlockNumberRejectedBeforeRequestCompletion() throws Exception { + for (long number : new long[]{-1, 0, 2}) { + assertInvalidBlockHeader(number, Sha256Hash.ZERO_HASH.getByteString()); + } + assertInvalidBlockHeader(1, new BlockId(Sha256Hash.ZERO_HASH, 1).getByteString()); + } + + @Test + public void testInvalidParentHashLengthRejectedBeforeRequestCompletion() throws Exception { + for (int size : new int[]{0, 7, Sha256Hash.LENGTH - 1, Sha256Hash.LENGTH + 1}) { + assertInvalidBlockHeader(1, ByteString.copyFrom(new byte[size])); + } + } + + private void assertInvalidBlockHeader(long number, ByteString parentHash) throws Exception { + BlockCapsule canonical = signedWitnessBlock(); + BlockCapsule invalid = new BlockCapsule(canonical.getInstance().toBuilder() + .setBlockHeader(canonical.getInstance().getBlockHeader().toBuilder() + .setRawData(canonical.getInstance().getBlockHeader().getRawData().toBuilder() + .setNumber(number).setParentHash(parentHash))).build()); + // Cover requested advertisement/sync responses and unrequested fast-forward/relay ingress. + for (int mode = 0; mode < 4; mode++) { + TronNetDelegate delegate = Mockito.mock(TronNetDelegate.class); + AdvService adv = Mockito.mock(AdvService.class); + SyncService sync = Mockito.mock(SyncService.class); + RelayService relay = Mockito.mock(RelayService.class); + BlockMsgHandler ingress = witnessHandler(delegate, adv, sync); + ReflectUtils.setFieldValue(ingress, "fastForward", mode == 2); + ReflectUtils.setFieldValue(ingress, "relayService", relay); + PeerConnection sender = witnessPeer(); + sender.setRelayPeer(mode == 3); + BlockMessage message = Mockito.spy(new BlockMessage(invalid)); + Item requested = new Item(message.getBlockId(), InventoryType.BLOCK); + long requestTime = System.currentTimeMillis() - 1_000; + if (mode == 0) { + sender.getAdvInvRequest().put(requested, requestTime); + } else if (mode == 1) { + sender.getSyncBlockRequested().put(message.getBlockId(), requestTime); + } + + P2pException error = Assert.assertThrows(P2pException.class, + () -> ingress.processMessage(sender, message)); + + Assert.assertEquals(P2pException.TypeEnum.BAD_BLOCK, error.getType()); + Assert.assertEquals("block number does not follow parent", error.getMessage()); + if (mode == 0) { + Assert.assertEquals(Long.valueOf(requestTime), sender.getAdvInvRequest().get(requested)); + } else if (mode == 1) { + Assert.assertEquals(Long.valueOf(requestTime), + sender.getSyncBlockRequested().get(message.getBlockId())); + } + Assert.assertTrue(sender.getSyncBlockInProcess().isEmpty()); + Mockito.verify(message, Mockito.never()).sanitize(); + Mockito.verifyNoInteractions(delegate, adv, sync, relay, + ReflectUtils.getFieldObject(ingress, "fetchBlockService")); + } + } + @Test public void testProcessBlock() { TronNetDelegate tronNetDelegate = Mockito.mock(TronNetDelegate.class);