Skip to content

fix(net): harden shared state concurrency #6970

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
xxo1shine wants to merge 1 commit into
base: release_v4.8.3
Choose a base branch
from
+349 −32
Open

fix(net): harden shared state concurrency #6970

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion framework/src/main/java/org/tron/common/backup/BackupManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
4 changes: 2 additions & 2 deletions framework/src/main/java/org/tron/core/metrics/blockchain/BlockChainMetricManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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());
Expand Down
16 changes: 8 additions & 8 deletions framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

@lxcmyf lxcmyf Sep 17, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[MUST] Holding blockLock only on this call path does not make the check-then-set in syncNext() atomic; SyncService.processBlock() still calls syncNext() without that lock. Two threads can both observe syncChainRequested == null and send requests, after which a valid second response may be treated as BAD_MESSAGE. Put the check, summary creation, state update, and send under a lock shared by every caller, and add a same-peer concurrency test.

@xxo1shine xxo1shine Sep 18, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In the current workflow, processBlock() requests the next chain summary once the pending sync queue size falls within the batch threshold, usually before background processing drains the queue. This change already coordinates ChainInventory response handling and background block processing using the same blockLock. Given the added complexity of expanding the synchronization scope, we will retain the current approach in this PR. If a duplicate-request scenario can be reproduced, we can address that specific path in a follow-up.

}
}
}

Expand Down
3 changes: 1 addition & 2 deletions framework/src/main/java/org/tron/core/net/peer/PeerConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down
15 changes: 7 additions & 8 deletions framework/src/main/java/org/tron/core/net/peer/PeerManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down Expand Up @@ -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();
}
}
}
Expand Down
4 changes: 2 additions & 2 deletions framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -159,4 +159,4 @@ public FetchBlockInfo(Sha256Hash hash, PeerConnection peer, long time) {

}

}
}
14 changes: 7 additions & 7 deletions framework/src/main/java/org/tron/core/net/service/statistics/MessageCount.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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);
}

Expand Down
4 changes: 2 additions & 2 deletions framework/src/main/java/org/tron/core/services/WitnessProductBlockService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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<>();

@lxcmyf lxcmyf Sep 17, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[MUST] Replacing the outer map with ConcurrentHashMap only makes individual map operations safe. The cache get→put at lines 27–39 can let two conflicting blocks for the same witness and height both observe null and overwrite each other; containsKey→put→clear/add can also lose counts, while NodeInfoService may iterate the inner HashSet concurrently. Use one atomic section for detection and update, expose an immutable or synchronized snapshot to readers, and add a real concurrency test.

@xxo1shine xxo1shine Sep 18, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Block processing itself is serialized. Although the subsequent double-production detection runs outside that lock and therefore still has a theoretical interleaving window, triggering it requires a particular thread scheduling sequence. This functionality provides auxiliary records and reporting of witness double production; it does not participate in consensus decisions. Considering the scope of the impact and the maintenance cost, this PR will retain the change that makes the outer map safe for concurrent access without introducing additional synchronization or snapshot mechanisms.


public void validWitnessProductTwoBlock(BlockCapsule block) {
try {
Expand Down
8 changes: 8 additions & 0 deletions framework/src/test/java/org/tron/common/backup/BackupManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down
38 changes: 38 additions & 0 deletions framework/src/test/java/org/tron/core/metrics/blockchain/BlockChainMetricManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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);
}
}
}
77 changes: 77 additions & 0 deletions framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,21 +1,28 @@
package org.tron.core.net.messagehandler;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.Mockito;
import org.tron.common.TestConstants;
import org.tron.common.utils.Pair;
import org.tron.common.utils.ReflectUtils;
import org.tron.common.utils.Sha256Hash;
import org.tron.core.capsule.BlockCapsule.BlockId;
import org.tron.core.config.Parameter.NetConstants;
import org.tron.core.config.args.Args;
import org.tron.core.exception.P2pException;
import org.tron.core.net.TronNetDelegate;
import org.tron.core.net.message.keepalive.PingMessage;
import org.tron.core.net.message.sync.ChainInventoryMessage;
import org.tron.core.net.peer.PeerConnection;
import org.tron.core.net.service.sync.SyncService;

public class ChainInventoryMsgHandlerTest {

Expand Down Expand Up @@ -78,4 +85,74 @@ public void testProcessMessage() throws Exception {
Assert.assertNull(msg.getAnswerMessage());
}

@Test
public void testFetchFlagDecisionIsMadeUnderBlockLock() throws Exception {
Object blockLock = new Object();
TronNetDelegate tronNetDelegate = Mockito.mock(TronNetDelegate.class);
SyncService syncService = Mockito.mock(SyncService.class);
PeerConnection testPeer = Mockito.spy(new PeerConnection());
AtomicBoolean fetchAbleSetUnderLock = new AtomicBoolean();
AtomicBoolean fetchFlagSetUnderLock = new AtomicBoolean();
BlockId firstBlock = new BlockId(Sha256Hash.ZERO_HASH, 1);
BlockId secondBlock = new BlockId(Sha256Hash.ZERO_HASH, 2);

Mockito.when(tronNetDelegate.getBlockLock()).thenReturn(blockLock);
Mockito.when(tronNetDelegate.getHeadBlockId()).thenReturn(new BlockId(Sha256Hash.ZERO_HASH, 0));
Mockito.when(tronNetDelegate.containBlock(Mockito.any())).thenReturn(false);
Mockito.doAnswer(invocation -> {
if (invocation.getArgument(0)) {
fetchAbleSetUnderLock.set(Thread.holdsLock(blockLock));
}
return invocation.callRealMethod();
}).when(testPeer).setFetchAble(Mockito.anyBoolean());
Mockito.doAnswer(invocation -> {
fetchFlagSetUnderLock.set(Thread.holdsLock(blockLock));
return null;
}).when(syncService).setFetchFlag(true);
ReflectUtils.setFieldValue(handler, "tronNetDelegate", tronNetDelegate);
ReflectUtils.setFieldValue(handler, "syncService", syncService);
testPeer.setSyncChainRequested(new Pair<>(new LinkedList<>(Arrays.asList(firstBlock)),
System.currentTimeMillis()));

handler.processMessage(testPeer,
new ChainInventoryMessage(Arrays.asList(firstBlock, secondBlock), 0L));

Assert.assertTrue(fetchAbleSetUnderLock.get());
Assert.assertTrue(fetchFlagSetUnderLock.get());
}

@Test
public void testSyncNextDecisionIsMadeUnderBlockLock() throws Exception {
Object blockLock = new Object();
TronNetDelegate tronNetDelegate = Mockito.mock(TronNetDelegate.class);
SyncService syncService = Mockito.mock(SyncService.class);
PeerConnection testPeer = Mockito.spy(new PeerConnection());
AtomicBoolean fetchAbleSetUnderLock = new AtomicBoolean();
AtomicBoolean syncNextCalledUnderLock = new AtomicBoolean();
BlockId firstBlock = new BlockId(Sha256Hash.ZERO_HASH, 1);

Mockito.when(tronNetDelegate.getBlockLock()).thenReturn(blockLock);
Mockito.when(tronNetDelegate.getHeadBlockId()).thenReturn(new BlockId(Sha256Hash.ZERO_HASH, 0));
Mockito.when(tronNetDelegate.containBlock(Mockito.any())).thenReturn(false);
Mockito.doAnswer(invocation -> {
if (invocation.getArgument(0)) {
fetchAbleSetUnderLock.set(Thread.holdsLock(blockLock));
}
return invocation.callRealMethod();
}).when(testPeer).setFetchAble(Mockito.anyBoolean());
Mockito.doAnswer(invocation -> {
syncNextCalledUnderLock.set(Thread.holdsLock(blockLock));
return null;
}).when(syncService).syncNext(testPeer);
ReflectUtils.setFieldValue(handler, "tronNetDelegate", tronNetDelegate);
ReflectUtils.setFieldValue(handler, "syncService", syncService);
testPeer.setSyncChainRequested(new Pair<>(new LinkedList<>(Arrays.asList(firstBlock)),
System.currentTimeMillis()));

handler.processMessage(testPeer, new ChainInventoryMessage(Arrays.asList(firstBlock), 0L));

Assert.assertTrue(fetchAbleSetUnderLock.get());
Assert.assertTrue(syncNextCalledUnderLock.get());
}

}
9 changes: 9 additions & 0 deletions framework/src/test/java/org/tron/core/net/peer/PeerConnectionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

import org.junit.AfterClass;
import org.junit.Assert;
Expand Down Expand Up @@ -56,6 +57,14 @@ public void testVariableDefaultValue() {
Assert.assertTrue(!peerConnection.isSyncFinish());
}

@Test
public void testSyncBlockInProcessUsesConcurrentSet() {
PeerConnection peerConnection = new PeerConnection();

Assert.assertTrue(peerConnection.getSyncBlockInProcess()
instanceof ConcurrentHashMap.KeySetView);
}

@Test
public void testOnDisconnect() {
PeerConnection peerConnection = new PeerConnection();
Expand Down
Loading
Loading