fix(net): harden shared state concurrency

Network workers, synchronization tasks, metrics collectors, and backup services access several mutable states across threads without consistent visibility or synchronization guarantees.

Use concurrent collections and synchronized state transitions where compound access must remain atomic, publish cross-thread fields safely, serialize peer cleanup with existing manager operations, and prevent duplicate-witness metrics from exposing incomplete companion state.

Author: wb

framework/src/main/java/org/tron/common/backup/BackupManager.java

@@ -58,7 +58,7 @@ public class BackupManager implements EventHandler {
   private MessageHandler messageHandler;
 
   @Getter
-  private BackupStatusEnum status = MASTER;
+  private volatile BackupStatusEnum status = MASTER;
 
   private volatile long lastKeepAliveTime;
 

framework/src/main/java/org/tron/core/metrics/blockchain/BlockChainMetricManager.java

@@ -138,10 +138,10 @@ public void applyBlock(BlockCapsule block) {
       BlockCapsule oldBlock = witnessInfo.get(witnessAddress);
       if ((!oldBlock.getBlockId().equals(block.getBlockId()))
           && oldBlock.getTimeStamp() == block.getTimeStamp()) {
+        dupWitnessBlockNum.put(witnessAddress, block.getNum());
         MetricsUtil.counterInc(MetricsKey.BLOCKCHAIN_DUP_WITNESS + witnessAddress);
         Metrics.counterInc(MetricKeys.Counter.MINER, 1,
             StringUtil.encode58Check(address), MetricLabels.Counter.MINE_DUP);
-        dupWitnessBlockNum.put(witnessAddress, block.getNum());
       }
     }
     witnessInfo.put(witnessAddress, block);
@@ -204,7 +204,7 @@ private List<DupWitnessInfo> getDupWitness() {
     for (Map.Entry<String, Counter> entry : dupWitnessMap.entrySet()) {
       DupWitnessInfo dupWitness = new DupWitnessInfo();
       String witness = entry.getKey().substring(MetricsKey.BLOCKCHAIN_DUP_WITNESS.length());
-      long blockNum = dupWitnessBlockNum.get(witness);
+      long blockNum = dupWitnessBlockNum.getOrDefault(witness, 0L);
       dupWitness.setAddress(witness);
       dupWitness.setBlockNum(blockNum);
       dupWitness.setCount((int) entry.getValue().getCount());

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

@@ -86,15 +86,15 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
         peer.setFetchAble(true);
         return;
       }
-    }
 
-    peer.setFetchAble(true);
-    if ((chainInventoryMessage.getRemainNum() == 0 && !peer.getSyncBlockToFetch().isEmpty())
-        || (chainInventoryMessage.getRemainNum() != 0
-        && peer.getSyncBlockToFetch().size() > syncFetchBatchNum)) {
-      syncService.setFetchFlag(true);
-    } else {
-      syncService.syncNext(peer);
+      peer.setFetchAble(true);
+      if ((chainInventoryMessage.getRemainNum() == 0 && !peer.getSyncBlockToFetch().isEmpty())
+          || (chainInventoryMessage.getRemainNum() != 0
+          && peer.getSyncBlockToFetch().size() > syncFetchBatchNum)) {
+        syncService.setFetchFlag(true);
+      } else {
+        syncService.syncNext(peer);
+      }
     }
   }
 

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

@@ -10,7 +10,6 @@
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.Deque;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -158,7 +157,7 @@ public class PeerConnection {
   private volatile Pair<Deque<BlockId>, Long> syncChainRequested = null;
   @Setter
   @Getter
-  private Set<BlockId> syncBlockInProcess = new HashSet<>();
+  private Set<BlockId> syncBlockInProcess = ConcurrentHashMap.newKeySet();
   @Setter
   @Getter
   private volatile boolean needSyncFromPeer = true;

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

@@ -84,13 +84,16 @@ public static synchronized PeerConnection remove(Channel channel) {
     return peerConnection;
   }
 
-  private static void remove(PeerConnection peerConnection) {
-    peers.remove(peerConnection);
+  private static synchronized boolean remove(PeerConnection peerConnection) {
+    if (!peers.remove(peerConnection)) {
+      return false;
+    }
     if (peerConnection.getChannel().isActive()) {
       activePeersCount.decrementAndGet();
     } else {
       passivePeersCount.decrementAndGet();
     }
+    return true;
   }
 
   public static synchronized void sortPeers() {
@@ -126,13 +129,9 @@ private static void check() {
       long disconnectTime = peer.getChannel().getDisconnectTime();
       if (disconnectTime != 0 && now - disconnectTime > DISCONNECTION_TIME_OUT) {
         logger.warn("Notify disconnect peer {}.", peer.getInetSocketAddress());
-        peers.remove(peer);
-        if (peer.getChannel().isActive()) {
-          activePeersCount.decrementAndGet();
-        } else {
-          passivePeersCount.decrementAndGet();
+        if (remove(peer)) {
+          peer.onDisconnect();
         }
-        peer.onDisconnect();
       }
     }
   }

framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java

@@ -34,7 +34,7 @@ public class FetchBlockService {
   @Autowired
   private ChainBaseManager chainBaseManager;
 
-  private FetchBlockInfo fetchBlockInfo = null;
+  private volatile FetchBlockInfo fetchBlockInfo = null;
 
   private final long fetchTimeOut = CommonParameter.getInstance().fetchBlockTimeout;
 
@@ -159,4 +159,4 @@ public FetchBlockInfo(Sha256Hash hash, PeerConnection peer, long time) {
 
   }
 
-}
+}

framework/src/main/java/org/tron/core/net/service/statistics/MessageCount.java

@@ -15,7 +15,7 @@ public class MessageCount {
 
   private long totalCount = 0;
 
-  private void update() {
+  private synchronized void update() {
     long time = System.currentTimeMillis() / 1000;
     long gap = time - indexTime;
     int k = gap > SIZE ? SIZE : (int) gap;
@@ -28,19 +28,19 @@ private void update() {
     }
   }
 
-  public void add() {
+  public synchronized void add() {
     update();
     szCount[index]++;
     totalCount++;
   }
 
-  public void add(int count) {
+  public synchronized void add(int count) {
     update();
     szCount[index] += count;
     totalCount += count;
   }
 
-  public int getCount(int interval) {
+  public synchronized int getCount(int interval) {
     if (interval > SIZE) {
       logger.warn("Param interval({}) is gt SIZE({})", interval, SIZE);
       return 0;
@@ -53,16 +53,16 @@ public int getCount(int interval) {
     return count;
   }
 
-  public long getTotalCount() {
+  public synchronized long getTotalCount() {
     return totalCount;
   }
 
-  public void reset() {
+  public synchronized void reset() {
     totalCount = 0;
   }
 
   @Override
-  public String toString() {
+  public synchronized String toString() {
     return String.valueOf(totalCount);
   }
 

framework/src/main/java/org/tron/core/services/WitnessProductBlockService.java

@@ -3,10 +3,10 @@
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import java.util.Arrays;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -20,7 +20,7 @@ public class WitnessProductBlockService {
   private Cache<Long, BlockCapsule> historyBlockCapsuleCache = CacheBuilder.newBuilder()
       .initialCapacity(200).maximumSize(200).build();
 
-  private Map<String, CheatWitnessInfo> cheatWitnessInfoMap = new HashMap<>();
+  private Map<String, CheatWitnessInfo> cheatWitnessInfoMap = new ConcurrentHashMap<>();
 
   public void validWitnessProductTwoBlock(BlockCapsule block) {
     try {

framework/src/test/java/org/tron/common/backup/BackupManagerTest.java

@@ -2,6 +2,7 @@
 
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
@@ -52,6 +53,13 @@ public void tearDown() {
     Args.clearParam();
   }
 
+  @Test
+  public void statusIsVolatileForCrossThreadVisibility() throws Exception {
+    Field status = BackupManager.class.getDeclaredField("status");
+
+    Assert.assertTrue(Modifier.isVolatile(status.getModifiers()));
+  }
+
   @Test
   public void test() throws Exception {
     CommonParameter.getInstance().setBackupPriority(8);

framework/src/test/java/org/tron/core/metrics/blockchain/BlockChainMetricManagerTest.java

@@ -0,0 +1,38 @@
+package org.tron.core.metrics.blockchain;
+
+import java.lang.reflect.Method;
+import java.util.List;
+import org.junit.Assert;
+import org.junit.Test;
+import org.tron.common.parameter.CommonParameter;
+import org.tron.core.metrics.MetricsKey;
+import org.tron.core.metrics.MetricsUtil;
+
+public class BlockChainMetricManagerTest {
+
+  @Test
+  @SuppressWarnings("unchecked")
+  public void missingDuplicateWitnessBlockNumberDefaultsToZero() throws Exception {
+    CommonParameter parameter = CommonParameter.getInstance();
+    boolean nodeMetricsEnabled = parameter.isNodeMetricsEnable();
+    String witness = "missing-block-number-" + System.nanoTime();
+    parameter.setNodeMetricsEnable(true);
+    try {
+      MetricsUtil.counterInc(MetricsKey.BLOCKCHAIN_DUP_WITNESS + witness);
+
+      Method getDupWitness = BlockChainMetricManager.class.getDeclaredMethod("getDupWitness");
+      getDupWitness.setAccessible(true);
+      List<DupWitnessInfo> dupWitnesses = (List<DupWitnessInfo>) getDupWitness.invoke(
+          new BlockChainMetricManager());
+
+      DupWitnessInfo dupWitness = dupWitnesses.stream()
+          .filter(info -> witness.equals(info.getAddress()))
+          .findFirst()
+          .orElse(null);
+      Assert.assertNotNull(dupWitness);
+      Assert.assertEquals(0L, dupWitness.getBlockNum());
+    } finally {
+      parameter.setNodeMetricsEnable(nodeMetricsEnabled);
+    }
+  }
+}