fix(net): correct block validation, fetching, and peer contribution

framework/src/main/java/org/tron/core/net/P2pEventHandlerImpl.java

@@ -259,7 +259,6 @@ private void updateLastInteractiveTime(PeerConnection peer, TronMessage msg) {
     switch (type) {
       case SYNC_BLOCK_CHAIN:
       case BLOCK_CHAIN_INVENTORY:
-      case BLOCK:
         flag = true;
         break;
       case FETCH_INV_DATA:
@@ -269,7 +268,7 @@ private void updateLastInteractiveTime(PeerConnection peer, TronMessage msg) {
         break;
     }
     if (flag) {
-      peer.setLastInteractiveTime(System.currentTimeMillis());
+      peer.updateLastInteractiveTime(System.currentTimeMillis());
     }
   }
 

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;
@@ -75,6 +76,12 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
       check(peer, blockMessage);
     }
 
+    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");
+    }
+
     blockMessage.sanitize();
 
     if (peer.getSyncBlockRequested().containsKey(blockId)) {
@@ -87,27 +94,22 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
       if (peer.isRelayPeer()) {
         peer.getAdvInvSpread().put(item, now);
       }
-      Long time = peer.getAdvInvRequest().remove(item);
+      Long time = peer.getAdvInvRequest().get(item);
       if (null != time) {
         peer.updateFetchLatency(now - time);
         Metrics.histogramObserve(MetricKeys.Histogram.BLOCK_FETCH_LATENCY,
             (now - time) / Metrics.MILLISECONDS_PER_SECOND);
-        // Best-effort duplicate signal: only responses matched to an outstanding adv
-        // request whose exact block id was already known before this response is
-        // processed (a concurrent or redundant arrival) are counted. The lookup is
-        // exact-id (block store + khaos), not a height comparison: an unknown fork
-        // block below head must not count. Concurrency can still let a simultaneous
-        // arrival slip through, and a duplicate is not attributed to a secondary
-        // fetch, so this is a lower-bound indicator rather than an exact count.
         if (tronNetDelegate.containBlock(blockId)) {
           Metrics.counterInc(MetricKeys.Counter.BLOCK_ALREADY_KNOWN, 1);
         }
       }
       Metrics.histogramObserve(MetricKeys.Histogram.BLOCK_RECEIVE_DELAY,
           (now - blockMessage.getBlockCapsule().getTimeStamp()) / Metrics.MILLISECONDS_PER_SECOND);
-      fetchBlockService.blockFetchSuccess(blockId);
       long interval = blockId.getNum() - tronNetDelegate.getHeadBlockId().getNum();
-      processBlock(peer, blockMessage.getBlockCapsule());
+      BlockResult result = processBlock(peer, blockMessage.getBlockCapsule());
+      if (result == BlockResult.ACCEPTED || result == BlockResult.SYNC_REQUIRED) {
+        peer.setBlockRcvTime(System.currentTimeMillis());
+      }
       logger.info(
               "Receive block/interval {}/{} from {} fetch/delay {}/{}ms, "
                       + "txs/process {}/{}ms, witness: {}",
@@ -132,46 +134,62 @@ private void check(PeerConnection peer, BlockMessage msg) throws P2pException {
     }
   }
 
-  private void processBlock(PeerConnection peer, BlockCapsule block) throws P2pException {
+  private BlockResult processBlock(PeerConnection peer, BlockCapsule block) throws P2pException {
     BlockId blockId = block.getBlockId();
-    boolean flag = tronNetDelegate.validBlock(block);
-    if (!flag) {
-      logger.warn("Receive a bad block from {}, {}, {}",
+    boolean activeWitness = tronNetDelegate.validBlock(block);
+    // Retain pending fetch/request state if validation throws so disconnect can retry.
+    // Otherwise, complete the fetch; inactive witnesses trigger sync recovery below.
+    fetchBlockService.blockFetchSuccess(blockId);
+    peer.getAdvInvRequest().remove(new Item(blockId, InventoryType.BLOCK));
+    if (!activeWitness) {
+      logger.warn("Receive a block from an inactive witness, peer {}, block {}, witness {}",
           peer.getInetSocketAddress(), blockId.getString(),
           Hex.toHexString(block.getWitnessAddress().toByteArray()));
-      return;
+      syncService.startSync(peer);
+      return BlockResult.STATE_FAILED;
+    }
+
+    peer.updateLastInteractiveTime(System.currentTimeMillis());
+    long headNum = tronNetDelegate.getHeadBlockId().getNum();
+    if (block.getNum() < headNum || tronNetDelegate.containBlock(blockId)) {
+      logger.warn("Receive a low block {}, head {}", blockId.getString(), headNum);
+      return BlockResult.IGNORED;
     }
 
     if (!tronNetDelegate.containBlock(block.getParentBlockId())) {
       logger.warn("Get unlink block {} from {}, head is {}", blockId.getString(),
               peer.getInetAddress(), tronNetDelegate.getHeadBlockId().getString());
       syncService.startSync(peer);
-      return;
-    }
-
-    long headNum = tronNetDelegate.getHeadBlockId().getNum();
-    if (block.getNum() < headNum) {
-      logger.warn("Receive a low block {}, head {}", blockId.getString(), headNum);
-      return;
+      return BlockResult.SYNC_REQUIRED;
     }
 
     broadcast(new BlockMessage(block));
 
     try {
       tronNetDelegate.processBlock(block, false);
-      peer.setBlockRcvTime(System.currentTimeMillis());
-      witnessProductBlockService.validWitnessProductTwoBlock(block);
-
-      Item item = new Item(blockId, InventoryType.BLOCK);
-      tronNetDelegate.getActivePeer().forEach(p -> {
-        if (p.getAdvInvReceive().getIfPresent(item) != null) {
-          p.setBlockBothHave(blockId);
-        }
-      });
     } catch (Exception e) {
       logger.warn("Process adv block {} from peer {} failed. reason: {}",
               blockId, peer.getInetAddress(), e.getMessage());
+      syncService.startSync(peer);
+      return BlockResult.STATE_FAILED;
     }
+    if (tronNetDelegate.isHitDown()) {
+      return BlockResult.IGNORED;
+    }
+    advService.confirmBlockInventory(blockId);
+    witnessProductBlockService.validWitnessProductTwoBlock(block);
+
+    Item item = new Item(blockId, InventoryType.BLOCK);
+    tronNetDelegate.getActivePeer().forEach(p -> {
+      if (p.getAdvInvReceive().getIfPresent(item) != null) {
+        p.setBlockBothHave(blockId);
+      }
+    });
+    return BlockResult.ACCEPTED;
+  }
+
+  private enum BlockResult {
+    ACCEPTED, SYNC_REQUIRED, IGNORED, STATE_FAILED
   }
 
   private void broadcast(BlockMessage blockMessage) {

framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java

@@ -53,6 +53,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
     if (blockIdWeGet.size() == 1 && tronNetDelegate.containBlock(blockIdWeGet.peek())) {
       peer.setTronState(TronState.SYNC_COMPLETED);
       peer.setNeedSyncFromPeer(false);
+      syncService.setHandleFlag(true);
       return;
     }
 
@@ -88,6 +89,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
       }
 
       peer.setFetchAble(true);
+      syncService.setHandleFlag(true);
       if ((chainInventoryMessage.getRemainNum() == 0 && !peer.getSyncBlockToFetch().isEmpty())
           || (chainInventoryMessage.getRemainNum() != 0
           && peer.getSyncBlockToFetch().size() > syncFetchBatchNum)) {

framework/src/main/java/org/tron/core/net/messagehandler/InventoryMsgHandler.java

@@ -6,7 +6,6 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.tron.common.utils.Sha256Hash;
-import org.tron.core.capsule.BlockCapsule.BlockId;
 import org.tron.core.config.args.Args;
 import org.tron.core.exception.P2pException;
 import org.tron.core.exception.P2pException.TypeEnum;
@@ -42,14 +41,8 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
 
     for (Sha256Hash id : inventoryMessage.getHashList()) {
       Item item = new Item(id, type);
-      peer.getAdvInvReceive().put(item, System.currentTimeMillis());
+      advService.recordInventory(peer, item, System.currentTimeMillis());
       advService.addInv(item);
-      if (type.equals(InventoryType.BLOCK) && peer.getAdvInvSpread().getIfPresent(item) == null) {
-        long headNum = tronNetDelegate.getHeadBlockId().getNum();
-        if (new BlockId(id).getNum() > headNum) {
-          peer.setLastInteractiveTime(System.currentTimeMillis());
-        }
-      }
     }
   }
 

framework/src/main/java/org/tron/core/net/peer/PeerConnection.java

@@ -103,8 +103,8 @@ public class PeerConnection {
    * clamped to fetchBlock.timeout only afterwards, so sustained degradation saturates
    * within 1-3 samples (issue #6923). Accepted cost: one transient outlier (~10x a
    * healthy estimate) can jump the estimate straight to the clamp, needing ~9 healthy
-   * samples to decay; hard-timeout fallback and the saturation gate (one secondary
-   * fetch per tick) bound the impact meanwhile.
+   * samples to decay; backup dispatch still requires an elapsed timeout or sufficient
+   * expected benefit.
    */
   private static final int FETCH_LATENCY_EWMA_FACTOR = 10;
 
@@ -142,6 +142,11 @@ public class PeerConnection {
   private Cache<Item, Long> advInvReceive = CacheBuilder.newBuilder().maximumSize(invCacheSize)
       .expireAfterWrite(1, TimeUnit.HOURS).recordStats().build();
 
+  // Eligible block INV receipt times, pending successful block processing before activity updates.
+  @Getter
+  private Cache<Item, Long> advBlockInvReceive = CacheBuilder.newBuilder().maximumSize(100)
+      .expireAfterWrite(1, TimeUnit.MINUTES).build();
+
   @Setter
   @Getter
   private Cache<Item, Long> advInvSpread = CacheBuilder.newBuilder().maximumSize(invCacheSize)
@@ -195,7 +200,7 @@ public void setChannel(Channel channel) {
       this.isRelayPeer = true;
     }
     this.nodeStatistics = TronStatsManager.getNodeStatistics(channel.getInetAddress());
-    lastInteractiveTime = System.currentTimeMillis();
+    updateLastInteractiveTime(System.currentTimeMillis());
     p2pRateLimiter.register(SYNC_BLOCK_CHAIN.asByte(),
         Args.getInstance().getRateLimiterSyncBlockChain());
     p2pRateLimiter.register(FETCH_INV_DATA.asByte(),
@@ -238,7 +243,7 @@ public long getFetchLatency() {
   }
 
   private long clampFetchLatency(long latency) {
-    // Saturation intentionally makes the >= timeout gate in FetchBlockService trigger.
+    // Bound measured latency without treating saturation as an elapsed fetch timeout.
     return StrictMathWrapper.max(0,
         StrictMathWrapper.min(CommonParameter.getInstance().fetchBlockTimeout, latency));
   }
@@ -248,10 +253,39 @@ public void setBlockBothHave(BlockId blockId) {
     this.blockBothHaveUpdateTime = System.currentTimeMillis();
   }
 
+  public synchronized void updateLastInteractiveTime(long time) {
+    if (time > lastInteractiveTime) {
+      lastInteractiveTime = time;
+    }
+  }
+
+  /**
+   * Returns whether there are no outstanding inventory or sync requests.
+   *
+   * <p>Sync blocks already being processed and sync direction flags are not checked.
+   */
   public boolean isIdle() {
     return advInvRequest.isEmpty() && isSyncIdle();
   }
 
+  /**
+   * Returns whether there are no outstanding block or sync requests and no sync blocks
+   * being processed. Outstanding transaction inventory requests do not make this check fail.
+   *
+   * <p>Callers must check connection state and sync direction flags separately before fetching.
+   */
+  public boolean isBlockFetchIdle() {
+    return advInvRequest.keySet().stream()
+        .noneMatch(item -> item.getType() == Protocol.Inventory.InventoryType.BLOCK)
+        && isSyncIdle() && syncBlockInProcess.isEmpty();
+  }
+
+  /**
+   * Returns whether there are no outstanding sync block or chain-summary requests.
+   *
+   * <p>This does not mean synchronization is complete: received blocks may still be processing,
+   * and sync direction flags are not checked.
+   */
   public boolean isSyncIdle() {
     return syncBlockRequested.isEmpty() && syncChainRequested == null;
   }
@@ -284,6 +318,7 @@ public void onDisconnect() {
     syncService.onDisconnect(this);
     advService.onDisconnect(this);
     advInvReceive.invalidateAll();
+    advBlockInvReceive.invalidateAll();
     advInvSpread.invalidateAll();
     advInvRequest.clear();
     syncBlockIdCache.invalidateAll();