fix(net): improve block source selection and disconnect recovery

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");
     }
   }
 

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;
   }

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,29 +256,67 @@ 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<Item> 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<PeerConnection> 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<PeerConnection> 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<PeerConnection> peers = tronNetDelegate.getActivePeer();
+    Collection<PeerConnection> trxPeers = peers.stream().filter(PeerConnection::isIdle)
         .collect(Collectors.toList());
     InvSender invSender = new InvSender();
     synchronized (this) {
       if (invToFetch.isEmpty() || peers.isEmpty()) {
         return;
       }
       long now = System.currentTimeMillis();
+      List<Item> 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();

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<Sha256Hash> sha256HashList, PeerConnection peer) {
+  public synchronized void fetchBlock(List<Sha256Hash> 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<Sha256Hash> 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<PeerConnection> 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));

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) {

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);

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);