Merge release v4.7.6 to Nile_v4.7.6

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

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/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/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/service/effective/ResilienceService.java

@@ -1,7 +1,9 @@
 package org.tron.core.net.service.effective;
 
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Random;
 import java.util.concurrent.ScheduledExecutorService;
@@ -29,6 +31,7 @@ public class ResilienceService {
   //when node is isolated, retention percent peers will not be disconnected
   public static final double retentionPercent = 0.8;
   private static final int initialDelay = 300;
+  public static final int minBroadcastPeerSize = 3;
   private static final String esName = "resilience-service";
   private final ScheduledExecutorService executor = ExecutorServiceManager
       .newSingleThreadScheduledExecutor(esName);
@@ -47,7 +50,7 @@ public void init() {
         } catch (Exception e) {
           logger.error("DisconnectRandom node failed", e);
         }
-      }, initialDelay, 60, TimeUnit.SECONDS);
+      }, initialDelay, 30, TimeUnit.SECONDS);
     } else {
       logger.info("OpenFullTcpDisconnect is disabled");
     }
@@ -71,21 +74,50 @@ public void init() {
 
   private void disconnectRandom() {
     int peerSize = tronNetDelegate.getActivePeer().size();
-    if (peerSize >= CommonParameter.getInstance().getMaxConnections()) {
+    if (peerSize < CommonParameter.getInstance().getMaxConnections()) {
+      return;
+    }
+    List<PeerConnection> peers = tronNetDelegate.getActivePeer().stream()
+        .filter(peer -> !peer.getChannel().isTrustPeer())
+        .filter(peer -> !peer.isNeedSyncFromUs() && !peer.isNeedSyncFromPeer())
+        .collect(Collectors.toList());
+
+    if (peers.size() >= minBroadcastPeerSize) {
       long now = System.currentTimeMillis();
-      List<PeerConnection> peers = tronNetDelegate.getActivePeer().stream()
-          .filter(peer -> now - peer.getLastInteractiveTime() >= inactiveThreshold)
+      Map<Object, Integer> weights = new HashMap<>();
+      peers.forEach(peer -> {
+        int weight = (int) Math.ceil((double) (now - peer.getLastInteractiveTime()) / 500);
+        weights.put(peer, Math.max(weight, 1));
+      });
+      WeightedRandom weightedRandom = new WeightedRandom(weights);
+      PeerConnection one = (PeerConnection) weightedRandom.next();
+      disconnectFromPeer(one, ReasonCode.RANDOM_ELIMINATION, DisconnectCause.RANDOM_ELIMINATION);
+      return;
+    }
+
+    int needSyncFromPeerCount = (int) tronNetDelegate.getActivePeer().stream()
+        .filter(peer -> !peer.getChannel().isTrustPeer())
+        .filter(PeerConnection::isNeedSyncFromPeer)
+        .count();
+    if (needSyncFromPeerCount >= 2) {
+      peers = tronNetDelegate.getActivePeer().stream()
           .filter(peer -> !peer.getChannel().isTrustPeer())
-          .filter(peer -> !peer.isNeedSyncFromUs() && !peer.isNeedSyncFromPeer())
+          .filter(peer -> peer.isNeedSyncFromUs() || peer.isNeedSyncFromPeer())
+          .collect(Collectors.toList());
+    } else {
+      peers = tronNetDelegate.getActivePeer().stream()
+          .filter(peer -> !peer.getChannel().isTrustPeer())
+          .filter(PeerConnection::isNeedSyncFromUs)
           .collect(Collectors.toList());
-      if (!peers.isEmpty()) {
-        int index = new Random().nextInt(peers.size());
-        disconnectFromPeer(peers.get(index), ReasonCode.RANDOM_ELIMINATION,
-            DisconnectCause.RANDOM_ELIMINATION);
-      }
+    }
+    if (!peers.isEmpty()) {
+      int index = new Random().nextInt(peers.size());
+      disconnectFromPeer(peers.get(index), ReasonCode.RANDOM_ELIMINATION,
+          DisconnectCause.RANDOM_ELIMINATION);
     }
   }
 
+
   private void disconnectLan() {
     if (!isLanNode()) {
       return;
@@ -198,6 +230,32 @@ private enum DisconnectCause {
     ISOLATE2_PASSIVE,
   }
 
+  static class WeightedRandom {
+
+    private final Map<Object, Integer> weights;
+    private final Random random;
+
+    public WeightedRandom(Map<Object, Integer> weights) {
+      this.weights = weights;
+      this.random = new Random();
+    }
+
+    public Object next() {
+      int totalWeight = 0;
+      for (int weight : weights.values()) {
+        totalWeight += weight;
+      }
+      int randomNum = random.nextInt(totalWeight);
+      for (Object key : weights.keySet()) {
+        randomNum -= weights.get(key);
+        if (randomNum < 0) {
+          return key;
+        }
+      }
+      throw new IllegalStateException("Sum of weights should not be negative.");
+    }
+  }
+
   public void close() {
     ExecutorServiceManager.shutdownAndAwaitTermination(executor, esName);
   }

framework/src/test/java/org/tron/core/db/TransactionExpireTest.java

@@ -50,19 +50,6 @@ public void init() throws IOException {
     context = new TronApplicationContext(DefaultConfig.class);
     wallet = context.getBean(Wallet.class);
     dbManager = context.getBean(Manager.class);
-
-    blockCapsule = new BlockCapsule(
-        1,
-        Sha256Hash.wrap(ByteString.copyFrom(
-            ByteArray.fromHexString(
-                "0304f784e4e7bae517bcab94c3e0c9214fb4ac7ff9d7d5a937d1f40031f87b81"))),
-        1,
-        ByteString.copyFromUtf8("testAddress"));
-    dbManager.getDynamicPropertiesStore().saveLatestBlockHeaderNumber(blockCapsule.getNum());
-    dbManager.getDynamicPropertiesStore()
-        .saveLatestBlockHeaderTimestamp(blockCapsule.getTimeStamp());
-    dbManager.updateRecentBlock(blockCapsule);
-    initLocalWitness();
   }
 
   private void initLocalWitness() {
@@ -81,6 +68,19 @@ public void removeDb() {
 
   @Test
   public void testExpireTransaction() {
+    blockCapsule = new BlockCapsule(
+        1,
+        Sha256Hash.wrap(ByteString.copyFrom(
+            ByteArray.fromHexString(
+                "0304f784e4e7bae517bcab94c3e0c9214fb4ac7ff9d7d5a937d1f40031f87b81"))),
+        1,
+        ByteString.copyFromUtf8("testAddress"));
+    dbManager.getDynamicPropertiesStore().saveLatestBlockHeaderNumber(blockCapsule.getNum());
+    dbManager.getDynamicPropertiesStore()
+        .saveLatestBlockHeaderTimestamp(blockCapsule.getTimeStamp());
+    dbManager.updateRecentBlock(blockCapsule);
+    initLocalWitness();
+
     TransferContract transferContract = TransferContract.newBuilder()
         .setAmount(1L)
         .setOwnerAddress(ByteString.copyFrom(Args.getLocalWitnesses()
@@ -101,8 +101,61 @@ public void testExpireTransaction() {
     Assert.assertEquals(response_code.TRANSACTION_EXPIRATION_ERROR, result.getCode());
   }
 
+  @Test
+  public void testExpireTransactionNew() {
+    blockCapsule = new BlockCapsule(
+        1,
+        Sha256Hash.wrap(ByteString.copyFrom(
+            ByteArray.fromHexString(
+                "0304f784e4e7bae517bcab94c3e0c9214fb4ac7ff9d7d5a937d1f40031f87b81"))),
+        System.currentTimeMillis(),
+        ByteString.copyFromUtf8("testAddress"));
+    dbManager.getDynamicPropertiesStore().saveLatestBlockHeaderNumber(blockCapsule.getNum());
+    dbManager.getDynamicPropertiesStore()
+        .saveLatestBlockHeaderTimestamp(blockCapsule.getTimeStamp());
+    dbManager.updateRecentBlock(blockCapsule);
+    initLocalWitness();
+    byte[] address = Args.getLocalWitnesses()
+        .getWitnessAccountAddress(CommonParameter.getInstance().isECKeyCryptoEngine());
+    ByteString addressByte = ByteString.copyFrom(address);
+    AccountCapsule accountCapsule =
+        new AccountCapsule(Protocol.Account.newBuilder().setAddress(addressByte).build());
+    accountCapsule.setBalance(1000_000_000L);
+    dbManager.getChainBaseManager().getAccountStore()
+        .put(accountCapsule.createDbKey(), accountCapsule);
+
+    TransferContract transferContract = TransferContract.newBuilder()
+        .setAmount(1L)
+        .setOwnerAddress(addressByte)
+        .setToAddress(ByteString.copyFrom(ByteArray.fromHexString(
+            (Wallet.getAddressPreFixString() + "A389132D6639FBDA4FBC8B659264E6B7C90DB086"))))
+        .build();
+    TransactionCapsule transactionCapsule =
+        new TransactionCapsule(transferContract, ContractType.TransferContract);
+    transactionCapsule.setReference(blockCapsule.getNum(), blockCapsule.getBlockId().getBytes());
+
+    transactionCapsule.setExpiration(System.currentTimeMillis());
+    transactionCapsule.sign(ByteArray.fromHexString(Args.getLocalWitnesses().getPrivateKey()));
+
+    GrpcAPI.Return result = wallet.broadcastTransaction(transactionCapsule.getInstance());
+    Assert.assertEquals(response_code.TRANSACTION_EXPIRATION_ERROR, result.getCode());
+  }
+
   @Test
   public void testTransactionApprovedList() {
+    blockCapsule = new BlockCapsule(
+        1,
+        Sha256Hash.wrap(ByteString.copyFrom(
+            ByteArray.fromHexString(
+                "0304f784e4e7bae517bcab94c3e0c9214fb4ac7ff9d7d5a937d1f40031f87b81"))),
+        1,
+        ByteString.copyFromUtf8("testAddress"));
+    dbManager.getDynamicPropertiesStore().saveLatestBlockHeaderNumber(blockCapsule.getNum());
+    dbManager.getDynamicPropertiesStore()
+        .saveLatestBlockHeaderTimestamp(blockCapsule.getTimeStamp());
+    dbManager.updateRecentBlock(blockCapsule);
+    initLocalWitness();
+
     byte[] address = Args.getLocalWitnesses()
         .getWitnessAccountAddress(CommonParameter.getInstance().isECKeyCryptoEngine());
     TransferContract transferContract = TransferContract.newBuilder()

framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java

@@ -6,13 +6,18 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import lombok.Getter;
 import org.joda.time.DateTime;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.mockito.Mockito;
 import org.tron.common.BaseTest;
+import org.tron.common.runtime.TvmTestUtils;
 import org.tron.common.utils.ByteArray;
 import org.tron.core.Constant;
 import org.tron.core.config.args.Args;
@@ -75,10 +80,48 @@ public void testProcessMessage() {
       transactionsMsgHandler.processMessage(peer, new TransactionsMessage(transactionList));
       Assert.assertNull(advInvRequest.get(item));
       //Thread.sleep(10);
+      transactionsMsgHandler.close();
+      BlockingQueue<TrxEvent> smartContractQueue =
+          new LinkedBlockingQueue(2);
+      smartContractQueue.offer(new TrxEvent(null, null));
+      smartContractQueue.offer(new TrxEvent(null, null));
+      Field field1 = TransactionsMsgHandler.class.getDeclaredField("smartContractQueue");
+      field1.setAccessible(true);
+      field1.set(transactionsMsgHandler, smartContractQueue);
+      Protocol.Transaction trx1 = TvmTestUtils.generateTriggerSmartContractAndGetTransaction(
+          ByteArray.fromHexString("121212a9cf"),
+          ByteArray.fromHexString("121212a9cf"),
+          ByteArray.fromHexString("123456"),
+          100, 100000000, 0, 0);
+      Map<Item, Long> advInvRequest1 = new ConcurrentHashMap<>();
+      Item item1 = new Item(new TransactionMessage(trx1).getMessageId(),
+          Protocol.Inventory.InventoryType.TRX);
+      advInvRequest1.put(item1, 0L);
+      Mockito.when(peer.getAdvInvRequest()).thenReturn(advInvRequest1);
+      List<Protocol.Transaction> transactionList1 = new ArrayList<>();
+      transactionList1.add(trx1);
+      transactionsMsgHandler.processMessage(peer, new TransactionsMessage(transactionList1));
+      Assert.assertNull(advInvRequest.get(item1));
     } catch (Exception e) {
       Assert.fail();
     } finally {
       transactionsMsgHandler.close();
     }
   }
+
+  class TrxEvent {
+
+    @Getter
+    private PeerConnection peer;
+    @Getter
+    private TransactionMessage msg;
+    @Getter
+    private long time;
+
+    public TrxEvent(PeerConnection peer, TransactionMessage msg) {
+      this.peer = peer;
+      this.msg = msg;
+      this.time = System.currentTimeMillis();
+    }
+  }
 }