initial commit of test isolated

317787106 · 317787106 · commit 3b3b6686b9e7 · 2024-07-18T16:30:20.000+08:00
diff --git a/chainbase/src/main/java/org/tron/core/ChainBaseManager.java b/chainbase/src/main/java/org/tron/core/ChainBaseManager.java
@@ -244,6 +244,10 @@ public class ChainBaseManager {
   @Setter
   private long lowestBlockNum = -1; // except num = 0.
 
+  @Getter
+  @Setter
+  private long latestSaveBlockTime;
+
   // for test only
   public List<ByteString> getWitnesses() {
     return witnessScheduleStore.getActiveWitnesses();
@@ -381,6 +385,7 @@ private void init() {
     this.lowestBlockNum = this.blockIndexStore.getLimitNumber(1, 1).stream()
             .map(BlockId::getNum).findFirst().orElse(0L);
     this.nodeType = getLowestBlockNum() > 1 ? NodeType.LITE : NodeType.FULL;
+    this.latestSaveBlockTime = System.currentTimeMillis();
   }
 
   public void shutdown() {
diff --git a/common/src/main/java/org/tron/common/parameter/CommonParameter.java b/common/src/main/java/org/tron/common/parameter/CommonParameter.java
@@ -333,6 +333,9 @@ public class CommonParameter {
   public boolean isOpenFullTcpDisconnect;
   @Getter
   @Setter
+  public int inactiveThreshold;
+  @Getter
+  @Setter
   public boolean nodeDetectEnable;
   @Getter
   @Setter
diff --git a/common/src/main/java/org/tron/core/Constant.java b/common/src/main/java/org/tron/core/Constant.java
@@ -198,6 +198,8 @@ public class Constant {
 
   public static final String NODE_IS_OPEN_FULL_TCP_DISCONNECT = "node.isOpenFullTcpDisconnect";
 
+  public static final String NODE_INACTIVE_THRESHOLD = "node.inactiveThreshold";
+
   public static final String NODE_DETECT_ENABLE = "node.nodeDetectEnable";
 
   public static final String NODE_MAX_TRANSACTION_PENDING_SIZE = "node.maxTransactionPendingSize";
diff --git a/framework/src/main/java/org/tron/core/config/args/Args.java b/framework/src/main/java/org/tron/core/config/args/Args.java
@@ -173,6 +173,7 @@ public static void clearParam() {
     PARAMETER.receiveTcpMinDataLength = 2048;
     PARAMETER.isOpenFullTcpDisconnect = false;
     PARAMETER.nodeDetectEnable = false;
+    PARAMETER.inactiveThreshold = 600;
     PARAMETER.supportConstant = false;
     PARAMETER.debug = false;
     PARAMETER.minTimeRatio = 0.0;
@@ -845,6 +846,9 @@ public static void setParam(final String[] args, final String confFileName) {
     PARAMETER.nodeDetectEnable = config.hasPath(Constant.NODE_DETECT_ENABLE)
           && config.getBoolean(Constant.NODE_DETECT_ENABLE);
 
+    PARAMETER.inactiveThreshold = config.hasPath(Constant.NODE_INACTIVE_THRESHOLD)
+        ? config.getInt(Constant.NODE_INACTIVE_THRESHOLD) : 600;
+
     PARAMETER.maxTransactionPendingSize = config.hasPath(Constant.NODE_MAX_TRANSACTION_PENDING_SIZE)
         ? config.getInt(Constant.NODE_MAX_TRANSACTION_PENDING_SIZE) : 2000;
 
diff --git a/framework/src/main/java/org/tron/core/db/Manager.java b/framework/src/main/java/org/tron/core/db/Manager.java
@@ -1384,6 +1384,7 @@ public void updateDynamicProperties(BlockCapsule block) {
         (chainBaseManager.getDynamicPropertiesStore().getLatestBlockHeaderNumber()
             - chainBaseManager.getDynamicPropertiesStore().getLatestSolidifiedBlockNum()
             + 1));
+    chainBaseManager.setLatestSaveBlockTime(System.currentTimeMillis());
     Metrics.gaugeSet(MetricKeys.Gauge.HEADER_HEIGHT, block.getNum());
     Metrics.gaugeSet(MetricKeys.Gauge.HEADER_TIME, block.getTimeStamp());
   }
diff --git a/framework/src/main/java/org/tron/core/net/TronNetService.java b/framework/src/main/java/org/tron/core/net/TronNetService.java
@@ -22,6 +22,7 @@
 import org.tron.core.net.peer.PeerStatusCheck;
 import org.tron.core.net.service.adv.AdvService;
 import org.tron.core.net.service.effective.EffectiveCheckService;
+import org.tron.core.net.service.effective.ResilienceService;
 import org.tron.core.net.service.fetchblock.FetchBlockService;
 import org.tron.core.net.service.nodepersist.NodePersistService;
 import org.tron.core.net.service.relay.RelayService;
@@ -50,6 +51,9 @@ public class TronNetService {
   @Autowired
   private PeerStatusCheck peerStatusCheck;
 
+  @Autowired
+  private ResilienceService resilienceService;
+
   @Autowired
   private TransactionsMsgHandler transactionsMsgHandler;
 
@@ -88,6 +92,7 @@ public void start() {
       advService.init();
       syncService.init();
       peerStatusCheck.init();
+      resilienceService.init();
       transactionsMsgHandler.init();
       fetchBlockService.init();
       nodePersistService.init();
@@ -110,6 +115,7 @@ public void close() {
     nodePersistService.close();
     advService.close();
     syncService.close();
+    resilienceService.close();
     peerStatusCheck.close();
     transactionsMsgHandler.close();
     fetchBlockService.close();
@@ -177,7 +183,7 @@ private P2pConfig updateConfig(P2pConfig config) {
     config.setMaxConnectionsWithSameIp(parameter.getMaxConnectionsWithSameIp());
     config.setPort(parameter.getNodeListenPort());
     config.setNetworkId(parameter.getNodeP2pVersion());
-    config.setDisconnectionPolicyEnable(parameter.isOpenFullTcpDisconnect());
+    config.setDisconnectionPolicyEnable(false);
     config.setNodeDetectEnable(parameter.isNodeDetectEnable());
     config.setDiscoverEnable(parameter.isNodeDiscoveryEnable());
     if (StringUtils.isEmpty(config.getIp()) && hasIpv4Stack(NetUtil.getAllLocalAddress())) {
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
@@ -76,6 +76,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
     if (!fastForward && !peer.isRelayPeer()) {
       check(peer, blockMessage);
     }
+    peer.setLastActiveTime(System.currentTimeMillis());
 
     if (peer.getSyncBlockRequested().containsKey(blockId)) {
       peer.getSyncBlockRequested().remove(blockId);
diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java
@@ -142,6 +142,7 @@ private void check(PeerConnection peer, ChainInventoryMessage msg) throws P2pExc
             + msg.getRemainNum() + " > futureMaxNum: " + maxFutureNum);
       }
     }
+    peer.setLastActiveTime(System.currentTimeMillis());
   }
 
 }
diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/FetchInvDataMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/FetchInvDataMsgHandler.java
@@ -171,6 +171,7 @@ private void check(PeerConnection peer, FetchInvDataMessage fetchInvDataMsg) thr
           peer.getSyncBlockIdCache().put(hash, System.currentTimeMillis());
         }
       }
+      peer.setLastActiveTime(System.currentTimeMillis());
     }
   }
 
diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/InventoryMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/InventoryMsgHandler.java
@@ -39,6 +39,9 @@ public void processMessage(PeerConnection peer, TronMessage msg) {
       Item item = new Item(id, type);
       peer.getAdvInvReceive().put(item, System.currentTimeMillis());
       advService.addInv(item);
+      if (type.equals(InventoryType.BLOCK) && peer.getAdvInvSpread().getIfPresent(item) == null) {
+        peer.setLastActiveTime(System.currentTimeMillis());
+      }
     }
   }
 
diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/SyncBlockChainMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/SyncBlockChainMsgHandler.java
@@ -33,7 +33,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
       peer.disconnect(Protocol.ReasonCode.BAD_PROTOCOL);
       return;
     }
-
+    peer.setLastActiveTime(System.currentTimeMillis());
     long remainNum = 0;
 
     List<BlockId> summaryChainIds = syncBlockChainMessage.getBlockIds();
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
@@ -79,6 +79,10 @@ public class PeerConnection {
   @Setter
   private ByteString address;
 
+  @Getter
+  @Setter
+  private long lastActiveTime;
+
   @Getter
   @Setter
   private TronState tronState = TronState.INIT;
@@ -159,6 +163,7 @@ public void setChannel(Channel channel) {
       this.isRelayPeer = true;
     }
     this.nodeStatistics = TronStatsManager.getNodeStatistics(channel.getInetAddress());
+    lastActiveTime = System.currentTimeMillis();
   }
 
   public void setBlockBothHave(BlockId blockId) {
diff --git a/framework/src/main/java/org/tron/core/net/service/effective/ResilienceService.java b/framework/src/main/java/org/tron/core/net/service/effective/ResilienceService.java
@@ -0,0 +1,170 @@
+package org.tron.core.net.service.effective;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.Optional;
+import java.util.Random;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.tron.common.es.ExecutorServiceManager;
+import org.tron.common.parameter.CommonParameter;
+import org.tron.core.ChainBaseManager;
+import org.tron.core.config.args.Args;
+import org.tron.core.net.TronNetDelegate;
+import org.tron.core.net.peer.PeerConnection;
+import org.tron.protos.Protocol.ReasonCode;
+
+@Slf4j(topic = "net")
+@Component
+public class ResilienceService {
+
+  private final long inactiveThreshold =
+      CommonParameter.getInstance().getInactiveThreshold() * 1000L;
+  public static final long blockNotChangeThreshold = 90 * 1000L;
+
+  //when node is isolated, retention percent peers will not be disconnected
+  public static final double retentionPercent = 0.8;
+  private static final int initialDelay = 300;
+  private final String esName = "resilience-service";
+
+  @Autowired
+  private TronNetDelegate tronNetDelegate;
+
+  @Autowired
+  private ChainBaseManager chainBaseManager;
+
+  private ScheduledExecutorService executor;
+
+  public void init() {
+    executor = ExecutorServiceManager.newSingleThreadScheduledExecutor(esName);
+
+    if (Args.getInstance().isOpenFullTcpDisconnect) {
+      executor.scheduleWithFixedDelay(() -> {
+        try {
+          disconnectRandom();
+        } catch (Exception e) {
+          logger.error("DisconnectRandom node failed", e);
+        }
+      }, initialDelay, 60, TimeUnit.SECONDS);
+    } else {
+      logger.info("OpenFullTcpDisconnect is disabled");
+    }
+
+    executor.scheduleWithFixedDelay(() -> {
+      try {
+        disconnectLan();
+      } catch (Exception e) {
+        logger.error("DisconnectLan node failed", e);
+      }
+    }, initialDelay, 10, TimeUnit.SECONDS);
+
+    executor.scheduleWithFixedDelay(() -> {
+      try {
+        disconnectIsolated2();
+      } catch (Exception e) {
+        logger.error("DisconnectIsolated node failed", e);
+      }
+    }, initialDelay, 30, TimeUnit.SECONDS);
+  }
+
+  private void disconnectRandom() {
+    int peerSize = tronNetDelegate.getActivePeer().size();
+    if (peerSize >= CommonParameter.getInstance().getMaxConnections()) {
+      long now = System.currentTimeMillis();
+      List<PeerConnection> peers = tronNetDelegate.getActivePeer().stream()
+          .filter(peer -> !peer.isDisconnect())
+          .filter(peer -> now - peer.getLastActiveTime() >= inactiveThreshold)
+          .filter(peer -> !peer.getChannel().isTrustPeer())
+          .collect(Collectors.toList());
+      if (!peers.isEmpty()) {
+        int index = new Random().nextInt(peers.size());
+        disconnectFromPeer(peers.get(index), ReasonCode.RANDOM_ELIMINATION);
+      }
+    }
+  }
+
+  private void disconnectLan() {
+    if (isLanNode()) {
+      // disconnect from the node that has keep inactive for more than inactiveThreshold
+      // and its lastActiveTime is smallest
+      int peerSize = tronNetDelegate.getActivePeer().size();
+      if (peerSize >= CommonParameter.getInstance().getMinConnections()) {
+        long now = System.currentTimeMillis();
+        Optional<PeerConnection> one = tronNetDelegate.getActivePeer().stream()
+            .filter(peer -> !peer.isDisconnect())
+            .filter(peer -> now - peer.getLastActiveTime() >= inactiveThreshold)
+            .filter(peer -> !peer.getChannel().isTrustPeer())
+            .min(Comparator.comparing(PeerConnection::getLastActiveTime, Long::compareTo));
+
+        one.ifPresent(peer -> disconnectFromPeer(peer, ReasonCode.BAD_PROTOCOL));
+      }
+    }
+  }
+
+  private void disconnectIsolated2() {
+    if (isIsolateLand2()) {
+      logger.info("Node is isolated, try to disconnect from peers");
+      int peerSize = tronNetDelegate.getActivePeer().size();
+
+      //disconnect from the node whose lastActiveTime is smallest
+      if (peerSize >= CommonParameter.getInstance().getMinActiveConnections()) {
+        Optional<PeerConnection> one = tronNetDelegate.getActivePeer().stream()
+            .filter(peer -> !peer.isDisconnect())
+            .filter(peer -> !peer.getChannel().isTrustPeer())
+            .filter(peer -> peer.getChannel().isActive())
+            .min(Comparator.comparing(PeerConnection::getLastActiveTime, Long::compareTo));
+
+        one.ifPresent(peer -> disconnectFromPeer(peer, ReasonCode.BAD_PROTOCOL));
+      }
+
+      //disconnect from some passive nodes, make sure retention nodes' num <= 0.8 * maxConnection
+      peerSize = tronNetDelegate.getActivePeer().size();
+      int threshold = (int) (CommonParameter.getInstance().getMaxConnections() * retentionPercent);
+      if (peerSize > threshold) {
+        int disconnectSize = peerSize - threshold;
+        List<PeerConnection> peers = tronNetDelegate.getActivePeer().stream()
+            .filter(peer -> !peer.isDisconnect())
+            .filter(peer -> !peer.getChannel().isTrustPeer())
+            .filter(peer -> !peer.getChannel().isActive())
+            .sorted(Comparator.comparing(PeerConnection::getLastActiveTime, Long::compareTo))
+            .collect(Collectors.toList());
+
+        if (peers.size() > disconnectSize) {
+          peers = peers.subList(0, disconnectSize);
+        }
+        peers.forEach(peer -> disconnectFromPeer(peer, ReasonCode.BAD_PROTOCOL));
+      }
+    }
+  }
+
+  private boolean isLanNode() {
+    int peerSize = tronNetDelegate.getActivePeer().size();
+    int activePeerSize = (int) tronNetDelegate.getActivePeer().stream()
+        .filter(peer -> peer.getChannel().isActive())
+        .count();
+    return peerSize > 0 && peerSize == activePeerSize;
+  }
+
+  private boolean isIsolateLand2() {
+    int advPeerCount = (int) tronNetDelegate.getActivePeer().stream()
+        .filter(peer -> !peer.isNeedSyncFromPeer() && !peer.isNeedSyncFromUs())
+        .count();
+    long diff = System.currentTimeMillis() - chainBaseManager.getLatestSaveBlockTime();
+    return advPeerCount >= 1 && diff >= blockNotChangeThreshold;
+  }
+
+  private void disconnectFromPeer(PeerConnection peer, ReasonCode reasonCode) {
+    int inactiveSeconds = (int) ((System.currentTimeMillis() - peer.getLastActiveTime()) / 1000);
+    logger.info("Disconnect from peer {}, inactive seconds {}", peer.getInetSocketAddress(),
+        inactiveSeconds);
+    peer.disconnect(reasonCode);
+  }
+
+  public void close() {
+    ExecutorServiceManager.shutdownAndAwaitTermination(executor, esName);
+  }
+}
diff --git a/framework/src/main/resources/config-localtest.conf b/framework/src/main/resources/config-localtest.conf
@@ -99,6 +99,7 @@ node {
 
   # check the peer data transfer ,disconnect factor
   isOpenFullTcpDisconnect = true
+  inactiveThreshold = 600
 
   p2p {
     version = 333 # 11111: mainnet; 20180622: testnet
diff --git a/framework/src/main/resources/config.conf b/framework/src/main/resources/config.conf
@@ -180,6 +180,7 @@ node {
   minParticipationRate = 15
 
   isOpenFullTcpDisconnect = false
+  inactiveThreshold = 600
 
   p2p {
     version = 11111 # 11111: mainnet; 20180622: testnet
diff --git a/framework/src/test/java/org/tron/core/net/services/ResilienceServiceTest.java b/framework/src/test/java/org/tron/core/net/services/ResilienceServiceTest.java
diff --git a/framework/src/test/resources/args-test.conf b/framework/src/test/resources/args-test.conf
diff --git a/framework/src/test/resources/config-localtest.conf b/framework/src/test/resources/config-localtest.conf
diff --git a/framework/src/test/resources/config-test.conf b/framework/src/test/resources/config-test.conf

Original file line number	Diff line number	Diff line change
`@@ -1384,6 +1384,7 @@ public void updateDynamicProperties(BlockCapsule block) {`
`1384`	`1384`	`(chainBaseManager.getDynamicPropertiesStore().getLatestBlockHeaderNumber()`
`1385`	`1385`	`- chainBaseManager.getDynamicPropertiesStore().getLatestSolidifiedBlockNum()`
`1386`	`1386`	`+ 1));`
	`1387`	`+ chainBaseManager.setLatestSaveBlockTime(System.currentTimeMillis());`
`1387`	`1388`	`Metrics.gaugeSet(MetricKeys.Gauge.HEADER_HEIGHT, block.getNum());`
`1388`	`1389`	`Metrics.gaugeSet(MetricKeys.Gauge.HEADER_TIME, block.getTimeStamp());`
`1389`	`1390`	`}`
Original file line number	Diff line number	Diff line change
`@@ -76,6 +76,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep`
`76`	`76`	`if (!fastForward && !peer.isRelayPeer()) {`
`77`	`77`	`check(peer, blockMessage);`
`78`	`78`	`}`
	`79`	`+ peer.setLastActiveTime(System.currentTimeMillis());`
`79`	`80`
`80`	`81`	`if (peer.getSyncBlockRequested().containsKey(blockId)) {`
`81`	`82`	`peer.getSyncBlockRequested().remove(blockId);`
Original file line number	Diff line number	Diff line change
`@@ -142,6 +142,7 @@ private void check(PeerConnection peer, ChainInventoryMessage msg) throws P2pExc`
`142`	`142`	`+ msg.getRemainNum() + " > futureMaxNum: " + maxFutureNum);`
`143`	`143`	`}`
`144`	`144`	`}`
	`145`	`+ peer.setLastActiveTime(System.currentTimeMillis());`
`145`	`146`	`}`
`146`	`147`
`147`	`148`	`}`
Original file line number	Diff line number	Diff line change
`@@ -171,6 +171,7 @@ private void check(PeerConnection peer, FetchInvDataMessage fetchInvDataMsg) thr`
`171`	`171`	`peer.getSyncBlockIdCache().put(hash, System.currentTimeMillis());`
`172`	`172`	`}`
`173`	`173`	`}`
	`174`	`+ peer.setLastActiveTime(System.currentTimeMillis());`
`174`	`175`	`}`
`175`	`176`	`}`
`176`	`177`
Original file line number	Diff line number	Diff line change
`@@ -39,6 +39,9 @@ public void processMessage(PeerConnection peer, TronMessage msg) {`
`39`	`39`	`Item item = new Item(id, type);`
`40`	`40`	`peer.getAdvInvReceive().put(item, System.currentTimeMillis());`
`41`	`41`	`advService.addInv(item);`
	`42`	`+ if (type.equals(InventoryType.BLOCK) && peer.getAdvInvSpread().getIfPresent(item) == null) {`
	`43`	`+ peer.setLastActiveTime(System.currentTimeMillis());`
	`44`	`+ }`
`42`	`45`	`}`
`43`	`46`	`}`
`44`	`47`