Merge pull request #5130 from wubin01/sync_block_prco

xxo1shine · web-flow · commit 54c05e54bbb1 · 2023-04-17T11:25:17.000+08:00
feat(net): optimize sync block processing logic
diff --git a/common/src/main/java/org/tron/core/exception/BadBlockException.java b/common/src/main/java/org/tron/core/exception/BadBlockException.java
@@ -2,11 +2,37 @@
 
 public class BadBlockException extends TronException {
 
+  private TypeEnum type = TypeEnum.DEFAULT;
+
   public BadBlockException() {
     super();
   }
 
   public BadBlockException(String message) {
     super(message);
   }
+
+  public BadBlockException(TypeEnum type, String message) {
+    super(message);
+    this.type = type;
+  }
+
+  public TypeEnum getType() {
+    return type;
+  }
+
+  public enum TypeEnum {
+    CALC_MERKLE_ROOT_FAILED(1),
+    DEFAULT(100);
+
+    private Integer value;
+
+    TypeEnum(Integer value) {
+      this.value = value;
+    }
+
+    public Integer getValue() {
+      return value;
+    }
+  }
 }
diff --git a/common/src/main/java/org/tron/core/exception/P2pException.java b/common/src/main/java/org/tron/core/exception/P2pException.java
@@ -50,6 +50,8 @@ public enum TypeEnum {
     TRX_EXE_FAILED(12, "trx exe failed"),
     DB_ITEM_NOT_FOUND(13, "DB item not found"),
     PROTOBUF_ERROR(14, "protobuf inconsistent"),
+    BLOCK_SIGN_ERROR(15, "block sign error"),
+    BLOCK_MERKLE_ERROR(16, "block merkle error"),
 
     DEFAULT(100, "default exception");
 
diff --git a/framework/src/main/java/org/tron/core/db/Manager.java b/framework/src/main/java/org/tron/core/db/Manager.java
@@ -1,6 +1,7 @@
 package org.tron.core.db;
 
 import static org.tron.common.utils.Commons.adjustBalance;
+import static org.tron.core.exception.BadBlockException.TypeEnum.CALC_MERKLE_ROOT_FAILED;
 import static org.tron.protos.Protocol.Transaction.Contract.ContractType.TransferContract;
 import static org.tron.protos.Protocol.Transaction.Result.contractResult.SUCCESS;
 
@@ -1213,8 +1214,8 @@ public void pushBlock(final BlockCapsule block)
             if (!block.calcMerkleRoot().equals(block.getMerkleRoot())) {
               logger.warn("Num: {}, the merkle root doesn't match, expect is {} , actual is {}.",
                   block.getNum(), block.getMerkleRoot(), block.calcMerkleRoot());
-              throw new BadBlockException(String.format("The merkle hash is not validated for %d",
-                  block.getNum()));
+              throw new BadBlockException(CALC_MERKLE_ROOT_FAILED,
+                      String.format("The merkle hash is not validated for %d", block.getNum()));
             }
             consensus.receiveBlock(block);
           }
diff --git a/framework/src/main/java/org/tron/core/net/TronNetDelegate.java b/framework/src/main/java/org/tron/core/net/TronNetDelegate.java
@@ -1,6 +1,7 @@
 package org.tron.core.net;
 
 import static org.tron.core.config.Parameter.ChainConstant.BLOCK_PRODUCED_INTERVAL;
+import static org.tron.core.exception.BadBlockException.TypeEnum.CALC_MERKLE_ROOT_FAILED;
 
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
@@ -282,7 +283,12 @@ public void processBlock(BlockCapsule block, boolean isSync) throws P2pException
           | EventBloomException e) {
         metricsService.failProcessBlock(block.getNum(), e.getMessage());
         logger.error("Process block failed, {}, reason: {}", blockId.getString(), e.getMessage());
-        throw new P2pException(TypeEnum.BAD_BLOCK, e);
+        if (e instanceof BadBlockException
+                && ((BadBlockException) e).getType().equals(CALC_MERKLE_ROOT_FAILED)) {
+          throw new P2pException(TypeEnum.BLOCK_MERKLE_ERROR, e);
+        } else {
+          throw new P2pException(TypeEnum.BAD_BLOCK, e);
+        }
       }
     }
   }
@@ -312,10 +318,10 @@ public void validSignature(BlockCapsule block) throws P2pException {
     try {
       if (!block.validateSignature(dbManager.getDynamicPropertiesStore(),
               dbManager.getAccountStore())) {
-        throw new P2pException(TypeEnum.BAD_BLOCK, "valid signature failed.");
+        throw new P2pException(TypeEnum.BLOCK_SIGN_ERROR, "valid signature failed.");
       }
     } catch (ValidateSignatureException e) {
-      throw new P2pException(TypeEnum.BAD_BLOCK, e);
+      throw new P2pException(TypeEnum.BLOCK_SIGN_ERROR, e);
     }
   }
 
diff --git a/framework/src/main/java/org/tron/core/net/service/sync/SyncService.java b/framework/src/main/java/org/tron/core/net/service/sync/SyncService.java
@@ -23,6 +23,7 @@
 import org.tron.core.capsule.BlockCapsule.BlockId;
 import org.tron.core.config.Parameter.NetConstants;
 import org.tron.core.config.args.Args;
+import org.tron.core.exception.BadBlockException;
 import org.tron.core.exception.P2pException;
 import org.tron.core.exception.P2pException.TypeEnum;
 import org.tron.core.net.TronNetDelegate;
@@ -263,33 +264,46 @@ private synchronized void handleSyncBlock() {
           tronNetDelegate.getActivePeer().stream()
               .filter(peer -> msg.getBlockId().equals(peer.getSyncBlockToFetch().peek()))
               .forEach(peer -> {
-                peer.getSyncBlockToFetch().pop();
-                peer.getSyncBlockInProcess().add(msg.getBlockId());
                 isFound[0] = true;
               });
           if (isFound[0]) {
             blockWaitToProcess.remove(msg);
             isProcessed[0] = true;
-            processSyncBlock(msg.getBlockCapsule());
+            processSyncBlock(msg.getBlockCapsule(), peerConnection);
           }
         }
       });
     }
   }
 
-  private void processSyncBlock(BlockCapsule block) {
+  private void processSyncBlock(BlockCapsule block, PeerConnection peerConnection) {
     boolean flag = true;
+    boolean attackFlag = false;
     BlockId blockId = block.getBlockId();
     try {
       tronNetDelegate.validSignature(block);
       tronNetDelegate.processBlock(block, true);
       pbftDataSyncHandler.processPBFTCommitData(block);
+    } catch (P2pException p2pException) {
+      logger.error("Process sync block {} failed, type: {}",
+              blockId.getString(), p2pException.getType());
+      attackFlag = p2pException.getType().equals(TypeEnum.BLOCK_SIGN_ERROR)
+              || p2pException.getType().equals(TypeEnum.BLOCK_MERKLE_ERROR);
+      flag = false;
     } catch (Exception e) {
       logger.error("Process sync block {} failed", blockId.getString(), e);
       flag = false;
     }
+
+    if (attackFlag) {
+      invalid(blockId, peerConnection);
+      peerConnection.disconnect(ReasonCode.BAD_BLOCK);
+      return;
+    }
+
     for (PeerConnection peer : tronNetDelegate.getActivePeer()) {
-      if (peer.getSyncBlockInProcess().remove(blockId)) {
+      if (blockId.equals(peer.getSyncBlockToFetch().peek())) {
+        peer.getSyncBlockToFetch().pop();
         if (flag) {
           peer.setBlockBothHave(blockId);
           if (peer.getSyncBlockToFetch().isEmpty() && peer.isFetchAble()) {
diff --git a/framework/src/test/java/org/tron/core/db/ManagerTest.java b/framework/src/test/java/org/tron/core/db/ManagerTest.java
@@ -4,6 +4,7 @@
 import static org.tron.common.utils.Commons.adjustBalance;
 import static org.tron.common.utils.Commons.adjustTotalShieldedPoolValue;
 import static org.tron.common.utils.Commons.getExchangeStoreFinal;
+import static org.tron.core.exception.BadBlockException.TypeEnum.CALC_MERKLE_ROOT_FAILED;
 
 import com.google.common.collect.Maps;
 import com.google.protobuf.ByteString;
@@ -392,6 +393,7 @@ public void pushBlockInvalidMerkelRoot() {
       Assert.assertTrue(false);
     } catch (BadBlockException e) {
       Assert.assertTrue(e instanceof BadBlockException);
+      Assert.assertTrue(e.getType().equals(CALC_MERKLE_ROOT_FAILED));
       Assert.assertEquals("The merkle hash is not validated for "
               + blockCapsule2.getNum(), e.getMessage());
     } catch (Exception e) {

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,7 @@`
`1`	`1`	`package org.tron.core.net;`
`2`	`2`
`3`	`3`	`import static org.tron.core.config.Parameter.ChainConstant.BLOCK_PRODUCED_INTERVAL;`
	`4`	`+import static org.tron.core.exception.BadBlockException.TypeEnum.CALC_MERKLE_ROOT_FAILED;`
`4`	`5`
`5`	`6`	`import com.google.common.cache.Cache;`
`6`	`7`	`import com.google.common.cache.CacheBuilder;`
`@@ -282,7 +283,12 @@ public void processBlock(BlockCapsule block, boolean isSync) throws P2pException`
`282`	`283`	`\| EventBloomException e) {`
`283`	`284`	`metricsService.failProcessBlock(block.getNum(), e.getMessage());`
`284`	`285`	`logger.error("Process block failed, {}, reason: {}", blockId.getString(), e.getMessage());`
`285`		`- throw new P2pException(TypeEnum.BAD_BLOCK, e);`
	`286`	`+ if (e instanceof BadBlockException`
	`287`	`+ && ((BadBlockException) e).getType().equals(CALC_MERKLE_ROOT_FAILED)) {`
	`288`	`+ throw new P2pException(TypeEnum.BLOCK_MERKLE_ERROR, e);`
	`289`	`+ } else {`
	`290`	`+ throw new P2pException(TypeEnum.BAD_BLOCK, e);`
	`291`	`+ }`
`286`	`292`	`}`
`287`	`293`	`}`
`288`	`294`	`}`
`@@ -312,10 +318,10 @@ public void validSignature(BlockCapsule block) throws P2pException {`
`312`	`318`	`try {`
`313`	`319`	`if (!block.validateSignature(dbManager.getDynamicPropertiesStore(),`
`314`	`320`	`dbManager.getAccountStore())) {`
`315`		`- throw new P2pException(TypeEnum.BAD_BLOCK, "valid signature failed.");`
	`321`	`+ throw new P2pException(TypeEnum.BLOCK_SIGN_ERROR, "valid signature failed.");`
`316`	`322`	`}`
`317`	`323`	`} catch (ValidateSignatureException e) {`
`318`		`- throw new P2pException(TypeEnum.BAD_BLOCK, e);`
	`324`	`+ throw new P2pException(TypeEnum.BLOCK_SIGN_ERROR, e);`
`319`	`325`	`}`
`320`	`326`	`}`
`321`	`327`