Dealing with duplication trx

Author: xxo1shine

src/main/java/org/tron/core/net/node/NodeImpl.java

@@ -476,25 +476,19 @@ private void consumerAdvObjToFetch() {
     long now = Time.getCurrentMillis();
     advObjToFetch.values().stream().sorted(PriorItem::compareTo).forEach(idToFetch -> {
       Sha256Hash hash = idToFetch.getHash();
-      if ( (idToFetch.getType().equals(InventoryType.TRX) && TrxCache.getIfPresent(hash) != null) ||
-              (idToFetch.getType().equals(InventoryType.BLOCK) && BlockCache.getIfPresent(hash) != null) ){
-        logger.info("{} {}  Already exist.", idToFetch.getType(), hash);
-        advObjToFetch.remove(hash);
-        return;
-      }
       if (idToFetch.getTime() < now - MSG_CACHE_DURATION_IN_BLOCKS * BLOCK_PRODUCED_INTERVAL) {
         logger.info("This obj is too late to fetch: " + idToFetch);
         advObjToFetch.remove(hash);
-      }else {
-        filterActivePeer.stream()
-                .filter(peer -> peer.getAdvObjSpreadToUs().containsKey(hash) && sendPackage.getSize(peer) < MAX_TRX_PER_PEER)
-                .sorted(Comparator.comparingInt(peer -> sendPackage.getSize(peer)))
-                .findFirst().ifPresent(peer -> {
-          sendPackage.add(idToFetch, peer);
-          peer.getAdvObjWeRequested().put(idToFetch.getItem(), now);
-          advObjToFetch.remove(hash);
-        });
+        return;
       }
+      filterActivePeer.stream()
+              .filter(peer -> peer.getAdvObjSpreadToUs().containsKey(hash) && sendPackage.getSize(peer) < MAX_TRX_PER_PEER)
+              .sorted(Comparator.comparingInt(peer -> sendPackage.getSize(peer)))
+              .findFirst().ifPresent(peer -> {
+        sendPackage.add(idToFetch, peer);
+        peer.getAdvObjWeRequested().put(idToFetch.getItem(), now);
+        advObjToFetch.remove(hash);
+      });
     });
 
     sendPackage.sendFetch();
@@ -653,8 +647,7 @@ public synchronized void disconnectInactive() {
 
   private void onHandleInventoryMessage(PeerConnection peer, InventoryMessage msg) {
     for (Sha256Hash id : msg.getHashList()){
-      if ( (msg.getInventoryType().equals(InventoryType.TRX) && TrxCache.getIfPresent(id) != null) ||
-              (msg.getInventoryType().equals(InventoryType.BLOCK) && BlockCache.getIfPresent(id) != null) ){
+      if (msg.getInventoryType().equals(InventoryType.TRX) && TrxCache.getIfPresent(id) != null) {
         logger.info("{} {} from peer {} Already exist.", msg.getInventoryType(), id, peer.getNode().getHost());
         continue;
       }
@@ -859,7 +852,18 @@ private void cleanUpSyncPeer(PeerConnection peer, ReasonCode reasonCode) {
     disconnectPeer(peer, reasonCode);
   }
 
+  synchronized boolean isExist(TransactionMessage trxMsg){
+    if (TrxCache.getIfPresent(trxMsg.getMessageId()) != null){
+      return true;
+    }
+    TrxCache.put(trxMsg.getMessageId(), trxMsg);
+    return false;
+  }
+
   private void onHandleTransactionMessage(PeerConnection peer, TransactionMessage trxMsg) {
+    if (isExist(trxMsg)){
+      return;
+    }
     //logger.info("on handle transaction message");
     try {
       Item item = new Item(trxMsg.getMessageId(), InventoryType.TRX);