Merge pull request #5963 from tronprotocol/release_v4.7.6

Release v4.7.6  merge to Nile_v4.7.6

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

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

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

framework/build.gradle

@@ -1,4 +1,5 @@
 plugins {
+    id "org.gradle.test-retry" version "1.5.9"
     id "org.sonarqube" version "2.6"
     id "com.gorylenko.gradle-git-properties" version "2.4.1"
 }
@@ -113,6 +114,10 @@ run {
 }
 
 test {
+    retry {
+        maxRetries = 5
+        maxFailures = 20
+    }
     testLogging {
         exceptionFormat = 'full'
     }

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,12 @@ 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;
+    if (PARAMETER.inactiveThreshold < 1) {
+      PARAMETER.inactiveThreshold = 1;
+    }
+
     PARAMETER.maxTransactionPendingSize = config.hasPath(Constant.NODE_MAX_TRANSACTION_PENDING_SIZE)
         ? config.getInt(Constant.NODE_MAX_TRANSACTION_PENDING_SIZE) : 2000;
 

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

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

@@ -19,6 +19,7 @@
 import org.tron.core.net.message.PbftMessageFactory;
 import org.tron.core.net.message.TronMessage;
 import org.tron.core.net.message.TronMessageFactory;
+import org.tron.core.net.message.adv.FetchInvDataMessage;
 import org.tron.core.net.message.adv.InventoryMessage;
 import org.tron.core.net.message.base.DisconnectMessage;
 import org.tron.core.net.message.handshake.HelloMessage;
@@ -38,7 +39,7 @@
 import org.tron.p2p.P2pEventHandler;
 import org.tron.p2p.connection.Channel;
 import org.tron.protos.Protocol;
-import org.tron.protos.Protocol.ReasonCode;
+import org.tron.protos.Protocol.Inventory.InventoryType;
 
 @Slf4j(topic = "net")
 @Component
@@ -205,6 +206,7 @@ private void processMessage(PeerConnection peer, byte[] data) {
         default:
           throw new P2pException(P2pException.TypeEnum.NO_SUCH_MESSAGE, msg.getType().toString());
       }
+      updateLastInteractiveTime(peer, msg);
     } catch (Exception e) {
       processException(peer, msg, e);
     } finally {
@@ -220,6 +222,27 @@ private void processMessage(PeerConnection peer, byte[] data) {
     }
   }
 
+  private void updateLastInteractiveTime(PeerConnection peer, TronMessage msg) {
+    MessageTypes type = msg.getType();
+
+    boolean flag = false;
+    switch (type) {
+      case SYNC_BLOCK_CHAIN:
+      case BLOCK_CHAIN_INVENTORY:
+      case BLOCK:
+        flag = true;
+        break;
+      case FETCH_INV_DATA:
+        flag = ((FetchInvDataMessage) msg).getInventoryType().equals(InventoryType.BLOCK);
+        break;
+      default:
+        break;
+    }
+    if (flag) {
+      peer.setLastInteractiveTime(System.currentTimeMillis());
+    }
+  }
+
   private void processException(PeerConnection peer, TronMessage msg, Exception ex) {
     Protocol.ReasonCode code;
 

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

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

@@ -151,6 +151,13 @@ private void processBlock(PeerConnection peer, BlockCapsule block) throws P2pExc
     try {
       tronNetDelegate.processBlock(block, false);
       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());

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

@@ -164,6 +164,10 @@ private void check(PeerConnection peer, FetchInvDataMessage fetchInvDataMsg) thr
             throw new P2pException(TypeEnum.BAD_MESSAGE,
                 "minBlockNum: " + minBlockNum + ", blockNum: " + blockNum);
           }
+          if (blockNum > peer.getLastSyncBlockId().getNum()) {
+            throw new P2pException(TypeEnum.BAD_MESSAGE,
+                "maxBlockNum: " + peer.getLastSyncBlockId().getNum() + ", blockNum: " + blockNum);
+          }
           if (peer.getSyncBlockIdCache().getIfPresent(hash) != null) {
             throw new P2pException(TypeEnum.BAD_MESSAGE,
                 new BlockId(hash).getString() + " is exist");

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.setLastInteractiveTime(System.currentTimeMillis());
+      }
     }
   }
 

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

@@ -33,7 +33,6 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
       peer.disconnect(Protocol.ReasonCode.BAD_PROTOCOL);
       return;
     }
-
     long remainNum = 0;
 
     List<BlockId> summaryChainIds = syncBlockChainMessage.getBlockIds();

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 volatile long lastInteractiveTime;
+
   @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());
+    lastInteractiveTime = System.currentTimeMillis();
   }
 
   public void setBlockBothHave(BlockId blockId) {
@@ -167,7 +172,11 @@ public void setBlockBothHave(BlockId blockId) {
   }
 
   public boolean isIdle() {
-    return advInvRequest.isEmpty() && syncBlockRequested.isEmpty() && syncChainRequested == null;
+    return advInvRequest.isEmpty() && isSyncIdle();
+  }
+
+  public boolean isSyncIdle() {
+    return syncBlockRequested.isEmpty() && syncChainRequested == null;
   }
 
   public void sendMessage(Message message) {
@@ -221,11 +230,13 @@ public String log() {
             + "syncBlockRequestedSize:%d\n"
             + "remainNum:%d\n"
             + "syncChainRequested:%d\n"
+            + "inactiveSeconds:%d\n"
             + "blockInProcess:%d\n",
         channel.getInetSocketAddress(),
         (now - channel.getStartTime()) / Constant.ONE_THOUSAND,
         channel.getAvgLatency(),
-        fastForwardBlock != null ? fastForwardBlock.getNum() : blockBothHave.getNum(),
+        fastForwardBlock != null ? fastForwardBlock.getNum() : String.format("%d [%ds]",
+            blockBothHave.getNum(), (now - blockBothHaveUpdateTime) / Constant.ONE_THOUSAND),
         isNeedSyncFromPeer(),
         isNeedSyncFromUs(),
         syncBlockToFetch.size(),
@@ -234,6 +245,7 @@ public String log() {
         remainNum,
         requested == null ? 0 : (now - requested.getValue())
                 / Constant.ONE_THOUSAND,
+        (now - lastInteractiveTime) / Constant.ONE_THOUSAND,
         syncBlockInProcess.size());
   }
 

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

@@ -94,7 +94,11 @@ private static void remove(PeerConnection peerConnection) {
   }
 
   public static synchronized void sortPeers() {
-    peers.sort(Comparator.comparingDouble(c -> c.getChannel().getAvgLatency()));
+    try {
+      peers.sort(Comparator.comparingLong(c -> c.getChannel().getAvgLatency()));
+    } catch (Exception e) {
+      logger.warn("Sort peers failed. {}", e.getMessage());
+    }
   }
 
   public static PeerConnection getPeerConnection(Channel channel) {