Merge pull request #6031 from halibobo1205/master_merge_to_develop

merge master to develop

build.gradle

@@ -37,7 +37,6 @@ subprojects {
         implementation group: 'org.slf4j', name: 'slf4j-api', version: '1.7.25'
         implementation group: 'org.slf4j', name: 'jcl-over-slf4j', version: '1.7.25'
         implementation group: 'ch.qos.logback', name: 'logback-classic', version: '1.2.9'
-        implementation group: 'com.google.guava', name: 'guava', version: '30.1-jre'
         implementation "com.google.code.findbugs:jsr305:3.0.0"
         implementation group: 'org.springframework', name: 'spring-context', version: '5.3.18'
         implementation group: 'org.springframework', name: 'spring-tx', version: '5.3.18'
@@ -68,12 +67,6 @@ subprojects {
         reproducibleFileOrder = true
         duplicatesStrategy = DuplicatesStrategy.INCLUDE // allow duplicates
     }
-
-    configurations.all {
-        resolutionStrategy {
-            force group: 'com.google.guava', name: 'guava', version: '30.1-jre'
-        }
-    }
 }
 
 task copyToParent(type: Copy) {

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

chainbase/src/main/java/org/tron/core/capsule/TransactionCapsule.java

@@ -59,6 +59,7 @@
 import org.tron.core.exception.P2pException;
 import org.tron.core.exception.PermissionException;
 import org.tron.core.exception.SignatureFormatException;
+import org.tron.core.exception.TransactionExpirationException;
 import org.tron.core.exception.ValidateSignatureException;
 import org.tron.core.store.AccountStore;
 import org.tron.core.store.DynamicPropertiesStore;
@@ -869,4 +870,12 @@ public void removeRedundantRet() {
       this.transaction = transactionBuilder.build();
     }
   }
+
+  public void checkExpiration(long nextSlotTime) throws TransactionExpirationException {
+    if (getExpiration() < nextSlotTime) {
+      throw new TransactionExpirationException(String.format(
+          "Transaction expiration time is %d, but next slot time is %d",
+          getExpiration(), nextSlotTime));
+    }
+  }
 }

common/build.gradle

@@ -46,7 +46,7 @@ dependencies {
     api 'org.aspectj:aspectjrt:1.8.13'
     api 'org.aspectj:aspectjweaver:1.8.13'
     api 'org.aspectj:aspectjtools:1.8.13'
-    api group: 'io.github.tronprotocol', name: 'libp2p', version: '2.2.1',{
+    api group: 'io.github.tronprotocol', name: 'libp2p', version: '2.2.4',{
         exclude group: 'io.grpc', module: 'grpc-context'
         exclude group: 'io.grpc', module: 'grpc-core'
         exclude group: 'io.grpc', module: 'grpc-netty'

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/src/main/java/org/tron/core/Wallet.java

@@ -549,6 +549,7 @@ public GrpcAPI.Return broadcastTransaction(Transaction signedTransaction) {
         throw new ContractValidateException(ActuatorConstant.CONTRACT_NOT_EXIST);
       }
       TransactionMessage message = new TransactionMessage(trx.getInstance().toByteArray());
+      trx.checkExpiration(tronNetDelegate.getNextBlockSlotTime());
       dbManager.pushTransaction(trx);
       int num = tronNetService.fastBroadcastTransaction(message);
       if (num == 0 && minEffectiveConnection != 0) {

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

@@ -1391,6 +1391,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());
   }
@@ -1575,6 +1576,7 @@ public BlockCapsule generateBlock(Miner miner, long blockTime, long timeout) {
     List<TransactionCapsule> toBePacked = new ArrayList<>();
     long currentSize = blockCapsule.getInstance().getSerializedSize();
     boolean isSort = Args.getInstance().isOpenTransactionSort();
+    int[] logSize = new int[] {pendingTransactions.size(), rePushTransactions.size(), 0, 0};
     while (pendingTransactions.size() > 0 || rePushTransactions.size() > 0) {
       boolean fromPending = false;
       TransactionCapsule trx;
@@ -1650,6 +1652,11 @@ public BlockCapsule generateBlock(Miner miner, long blockTime, long timeout) {
         tmpSession.merge();
         toBePacked.add(trx);
         currentSize += trxPackSize;
+        if (fromPending) {
+          logSize[2] += 1;
+        } else {
+          logSize[3] += 1;
+        }
       } catch (Exception e) {
         logger.warn("Process trx {} failed when generating block {}, {}.", trx.getTransactionId(),
             blockCapsule.getNum(), e.getMessage());
@@ -1666,11 +1673,14 @@ public BlockCapsule generateBlock(Miner miner, long blockTime, long timeout) {
     BlockCapsule capsule = new BlockCapsule(blockCapsule.getInstance());
     capsule.generatedByMyself = true;
     Metrics.histogramObserve(timer);
-    logger.info("Generate block {} success, trxs:{}, pendingCount: {}, rePushCount: {},"
-                    + " postponedCount: {}, blockSize: {} B",
-            capsule.getNum(), capsule.getTransactions().size(),
-            pendingTransactions.size(), rePushTransactions.size(), postponedTrxCount,
-            capsule.getSerializedSize());
+    logger.info("Generate block {} success, trxs:{}, before pendingCount: {}, rePushCount: {}, "
+            + "from pending: {}, rePush: {}, after pendingCount: {}, rePushCount: {}, "
+            + "postponedCount: {}, blockSize: {} B",
+        capsule.getNum(), capsule.getTransactions().size(),
+        logSize[0], logSize[1], logSize[2], logSize[3],
+        pendingTransactions.size(), rePushTransactions.size(), postponedTrxCount,
+        capsule.getSerializedSize());
+
     return capsule;
   }
 

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/TronNetDelegate.java

@@ -380,4 +380,12 @@ public boolean isBlockUnsolidified() {
     return headNum - solidNum >= maxUnsolidifiedBlocks;
   }
 
+  public long getNextBlockSlotTime() {
+    long slotCount = 1;
+    if (chainBaseManager.getDynamicPropertiesStore().getStateFlag() == 1) {
+      slotCount += chainBaseManager.getDynamicPropertiesStore().getMaintenanceSkipSlots();
+    }
+    return chainBaseManager.getDynamicPropertiesStore().getLatestBlockHeaderTimestamp()
+        + slotCount * BLOCK_PRODUCED_INTERVAL;
+  }
 }

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/messagehandler/TransactionsMsgHandler.java

@@ -13,6 +13,7 @@
 import org.tron.core.config.args.Args;
 import org.tron.core.exception.P2pException;
 import org.tron.core.exception.P2pException.TypeEnum;
+import org.tron.core.exception.TransactionExpirationException;
 import org.tron.core.net.TronNetDelegate;
 import org.tron.core.net.message.TronMessage;
 import org.tron.core.net.message.adv.TransactionMessage;
@@ -128,6 +129,7 @@ private void handleTransaction(PeerConnection peer, TransactionMessage trx) {
     }
 
     try {
+      trx.getTransactionCapsule().checkExpiration(tronNetDelegate.getNextBlockSlotTime());
       tronNetDelegate.pushTransaction(trx.getTransactionCapsule());
       advService.broadcast(trx);
     } catch (P2pException e) {
@@ -137,6 +139,9 @@ private void handleTransaction(PeerConnection peer, TransactionMessage trx) {
         peer.setBadPeer(true);
         peer.disconnect(ReasonCode.BAD_TX);
       }
+    } catch (TransactionExpirationException e) {
+      logger.warn("{}. trx: {}, peer: {}",
+          e.getMessage(), trx.getMessageId(), peer.getInetAddress());
     } catch (Exception e) {
       logger.error("Trx {} from peer {} process failed", trx.getMessageId(), peer.getInetAddress(),
           e);

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