fix(net): improve block sync resilience and transaction handling

framework/src/main/java/org/tron/core/db/Manager.java

@@ -999,8 +999,12 @@ public void eraseBlock() {
       khaosDb.pop();
       revokingStore.fastPop();
       logger.info("End to erase block: {}.", oldHeadBlock);
-      oldHeadBlock.getTransactions().forEach(tc ->
-          poppedTransactions.add(new TransactionCapsule(tc.getInstance())));
+      oldHeadBlock.getTransactions().forEach(tc -> {
+        if (isMultiSignTransaction(tc.getInstance())) {
+          ownerAddressSet.add(ByteArray.toHexString(tc.getOwnerAddress()));
+        }
+        poppedTransactions.add(new TransactionCapsule(tc.getInstance()));
+      });
       Metrics.gaugeInc(MetricKeys.Gauge.MANAGER_QUEUE, oldHeadBlock.getTransactions().size(),
           MetricLabels.Gauge.QUEUE_POPPED);
 

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

@@ -233,9 +233,6 @@ private boolean checkInvRateLimit(PeerConnection peer, InventoryMessage msg)
       if (count + currentSize > maxCountIn10s) {
         logger.warn("Drop TRX inv from {}, window:{}, cur:{}, max:{}",
             peer.getInetAddress(), count, currentSize, maxCountIn10s);
-        if (Args.getInstance().isOpenPrintLog()) {
-          logger.warn("[overload] Drop tx list: {}", msg.getHashList());
-        }
         return false;
       }
     } else if (invType == InventoryType.BLOCK) {

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

@@ -1,7 +1,6 @@
 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;
@@ -24,6 +23,7 @@
 import org.tron.common.prometheus.MetricLabels;
 import org.tron.common.prometheus.Metrics;
 import org.tron.common.utils.Sha256Hash;
+import org.tron.common.utils.Time;
 import org.tron.core.ChainBaseManager;
 import org.tron.core.capsule.BlockCapsule;
 import org.tron.core.capsule.BlockCapsule.BlockId;
@@ -312,12 +312,7 @@ 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());
-        if (e instanceof BadBlockException
-                && ((BadBlockException) e).getType().equals(CALC_MERKLE_ROOT_FAILED)) {
-          throw new P2pException(TypeEnum.BLOCK_MERKLE_INVALID, e);
-        } else {
-          throw new P2pException(TypeEnum.BAD_BLOCK, e);
-        }
+        throw new P2pException(TypeEnum.BAD_BLOCK, e);
       }
     }
   }
@@ -356,7 +351,7 @@ public void validSignature(BlockCapsule block) throws P2pException {
     }
   }
 
-  public boolean validBlock(BlockCapsule block) throws P2pException {
+  public void validBlock(BlockCapsule block) throws P2pException {
     long time = System.currentTimeMillis();
     if (block.getTimeStamp() - time > timeout) {
       throw new P2pException(TypeEnum.BAD_BLOCK,
@@ -368,7 +363,24 @@ public boolean validBlock(BlockCapsule block) throws P2pException {
       throw new P2pException(TypeEnum.BLOCK_MERKLE_INVALID, e.getMessage());
     }
     validSignature(block);
-    return witnessScheduleStore.getActiveWitnesses().contains(block.getWitnessAddress());
+  }
+
+  public boolean validAdvBlock(PeerConnection peer, BlockCapsule block) throws P2pException {
+    boolean activeWitness = witnessScheduleStore.getActiveWitnesses()
+        .contains(block.getWitnessAddress());
+    if (!activeWitness) {
+      logger.warn("Block witness is not active, block: {}, peer: {}",
+          block.getBlockId().getString(), peer.getInetSocketAddress());
+      long now = Time.getCurrentMillis();
+      long lastTime = peer.getLastInactiveWitnessBlockTime();
+      peer.setLastInactiveWitnessBlockTime(now);
+      if (now - lastTime < getMaintenanceTimeInterval() / 2) {
+        throw new P2pException(TypeEnum.BAD_BLOCK,
+            "Repeated block from an inactive witness: " + block.getBlockId().getString());
+      }
+    }
+    validBlock(block);
+    return activeWitness;
   }
 
   public PbftSignCapsule getBlockPbftCommitData(long blockNum) {

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

@@ -134,14 +134,9 @@ private void check(PeerConnection peer, BlockMessage msg) throws P2pException {
 
   private void processBlock(PeerConnection peer, BlockCapsule block) throws P2pException {
     BlockId blockId = block.getBlockId();
-    boolean flag = tronNetDelegate.validBlock(block);
-    if (!flag) {
-      logger.warn("Receive a bad block from {}, {}, {}",
-          peer.getInetSocketAddress(), blockId.getString(),
-          Hex.toHexString(block.getWitnessAddress().toByteArray()));
+    if (!tronNetDelegate.validAdvBlock(peer, block)) {
       return;
     }
-
     if (!tronNetDelegate.containBlock(block.getParentBlockId())) {
       logger.warn("Get unlink block {} from {}, head is {}", blockId.getString(),
               peer.getInetAddress(), tronNetDelegate.getHeadBlockId().getString());

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

@@ -93,6 +93,10 @@ public class PeerConnection {
   @Getter
   private volatile long blockRcvTime;
 
+  @Getter
+  @Setter
+  private volatile long lastInactiveWitnessBlockTime;
+
   /**
    * Sliding factor of the fetch latency Exponentially Weighted Moving Average (EWMA):
    * each new sample contributes

framework/src/main/java/org/tron/core/net/service/adv/AdvService.java

@@ -169,7 +169,8 @@ public int fastBroadcastTransaction(TransactionMessage msg) {
 
     Item item = new Item(msg.getMessageId(), InventoryType.TRX);
     trxCount.add();
-    trxCache.put(item, new TransactionMessage(msg.getTransactionCapsule().getInstance()));
+    trxCache.put(item, new TransactionMessage(
+        msg.getTransactionCapsule().getInstance().toBuilder().clearRet().build()));
 
     List<Sha256Hash> list = new ArrayList<>();
     list.add(msg.getMessageId());
@@ -216,7 +217,8 @@ public void broadcast(Message msg) {
       TransactionMessage trxMsg = (TransactionMessage) msg;
       item = new Item(trxMsg.getMessageId(), InventoryType.TRX);
       trxCount.add();
-      trxCache.put(item, new TransactionMessage(trxMsg.getTransactionCapsule().getInstance()));
+      trxCache.put(item, new TransactionMessage(
+          trxMsg.getTransactionCapsule().getInstance().toBuilder().clearRet().build()));
     } else {
       logger.error("Adv item is neither block nor trx, type: {}", msg.getType());
       return;

framework/src/main/java/org/tron/core/net/service/sync/SyncService.java

@@ -339,26 +339,21 @@ private synchronized void handleSyncBlock() {
   }
 
   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);
       peerConnection.setBlockRcvTime(System.currentTimeMillis());
       pbftDataSyncHandler.processPBFTCommitData(block);
-    } catch (P2pException p2pException) {
-      logger.error("Process sync block {} failed, type: {}",
-              blockId.getString(), p2pException.getType());
-      attackFlag = p2pException.getType().equals(TypeEnum.BLOCK_SIGN_INVALID)
-              || p2pException.getType().equals(TypeEnum.BLOCK_MERKLE_INVALID);
-      flag = false;
     } catch (Exception e) {
-      logger.error("Process sync block {} failed", blockId.getString(), e);
-      flag = false;
-    }
-
-    if (attackFlag) {
+      if (e instanceof P2pException) {
+        logger.error("Process sync block {} from {} failed, type: {}, reason: {}",
+            blockId.getString(), peerConnection.getInetSocketAddress(),
+            ((P2pException) e).getType(), e.getMessage());
+      } else {
+        logger.error("Process sync block {} from {} failed",
+            blockId.getString(), peerConnection.getInetSocketAddress(), e);
+      }
       invalid(blockId, peerConnection);
       peerConnection.disconnect(ReasonCode.BAD_BLOCK);
       return;
@@ -369,12 +364,8 @@ private void processSyncBlock(BlockCapsule block, PeerConnection peerConnection)
       if (blockId.equals(bid)) {
         peer.setBlockBothHave(blockId);
         peer.getSyncBlockToFetch().remove(bid);
-        if (flag) {
-          if (peer.getSyncBlockToFetch().isEmpty() && peer.isFetchAble()) {
-            syncNext(peer);
-          }
-        } else {
-          peer.disconnect(ReasonCode.BAD_BLOCK);
+        if (peer.getSyncBlockToFetch().isEmpty() && peer.isFetchAble()) {
+          syncNext(peer);
         }
       }
     }

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

@@ -928,6 +928,39 @@ public void getVerifyTxsSkipsBlockWhenPermissionTxAlreadyConsumed() throws Excep
     }
   }
 
+  @Test
+  public void eraseBlockMarksPermissionUpdateOwnerForVerification() throws Exception {
+    byte[] owner = ByteArray.fromHexString(Wallet.getAddressPreFixString()
+        + "548794500882809695A8A687866E76D4271A1ABC");
+    AccountContract.AccountPermissionUpdateContract contract =
+        AccountContract.AccountPermissionUpdateContract.newBuilder()
+            .setOwnerAddress(ByteString.copyFrom(owner))
+            .build();
+    TransactionCapsule permissionUpdate = new TransactionCapsule(contract,
+        ContractType.AccountPermissionUpdateContract);
+    BlockCapsule rolledBackBlock = new BlockCapsule(1, Sha256Hash.ZERO_HASH.getByteString(), 0,
+        Arrays.asList(permissionUpdate.getInstance()));
+    rolledBackBlock.setMerkleRoot();
+
+    Manager manager = new Manager();
+    ChainBaseManager mockedChainManager = mock(ChainBaseManager.class);
+    DynamicPropertiesStore mockedDynamicPropertiesStore = mock(DynamicPropertiesStore.class);
+    when(mockedChainManager.getDynamicPropertiesStore()).thenReturn(mockedDynamicPropertiesStore);
+    when(mockedDynamicPropertiesStore.getLatestBlockHeaderHash())
+        .thenReturn(Sha256Hash.wrap(new byte[32]));
+    when(mockedChainManager.getBlockById(any(Sha256Hash.class))).thenReturn(rolledBackBlock);
+    ReflectUtils.setFieldValue(manager, "chainBaseManager", mockedChainManager);
+    ReflectUtils.setFieldValue(manager, "khaosDb", mock(KhaosDatabase.class));
+    ReflectUtils.setFieldValue(manager, "revokingStore", mock(RevokingDatabase.class));
+
+    manager.eraseBlock();
+
+    @SuppressWarnings("unchecked")
+    Set<String> ownerAddressSet =
+        (Set<String>) ReflectUtils.getFieldObject(manager, "ownerAddressSet");
+    Assert.assertTrue(ownerAddressSet.contains(ByteArray.toHexString(owner)));
+  }
+
   @Test
   public void doNotSwitch()
       throws ValidateSignatureException, ContractValidateException, ContractExeException,

framework/src/test/java/org/tron/core/net/TronNetDelegateTest.java

@@ -4,6 +4,8 @@
 
 import com.google.protobuf.ByteString;
 import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
+import java.util.Collections;
 import org.junit.Assert;
 import org.junit.Test;
 import org.mockito.Mockito;
@@ -17,9 +19,14 @@
 import org.tron.core.capsule.TransactionCapsule;
 import org.tron.core.config.args.Args;
 import org.tron.core.db.Manager;
+import org.tron.core.exception.BadBlockException;
 import org.tron.core.exception.P2pException;
 import org.tron.core.exception.P2pException.TypeEnum;
+import org.tron.core.exception.ValidateScheduleException;
+import org.tron.core.metrics.MetricsService;
+import org.tron.core.net.peer.PeerConnection;
 import org.tron.core.store.DynamicPropertiesStore;
+import org.tron.core.store.WitnessScheduleStore;
 import org.tron.protos.Protocol.Transaction.Contract.ContractType;
 import org.tron.protos.contract.BalanceContract.TransferContract;
 
@@ -185,5 +192,127 @@ public void testValidBlockMerkleRoot() throws Exception {
     } catch (P2pException e) {
       Assert.assertEquals(TypeEnum.BLOCK_MERKLE_INVALID, e.getType());
     }
+    PeerConnection peer = new PeerConnection();
+    WitnessScheduleStore schedule = mock(WitnessScheduleStore.class);
+    setField(tronNetDelegate, "witnessScheduleStore", schedule);
+    Mockito.when(schedule.getActiveWitnesses())
+        .thenReturn(Collections.singletonList(tampered.getWitnessAddress()));
+    try {
+      tronNetDelegate.validAdvBlock(peer, tampered);
+      Assert.fail("Invalid Merkle root must still be rejected for an active witness");
+    } catch (P2pException e) {
+      Assert.assertEquals(TypeEnum.BLOCK_MERKLE_INVALID, e.getType());
+    }
+    Assert.assertEquals(0, peer.getLastInactiveWitnessBlockTime());
+  }
+
+  @Test
+  public void testValidBlockDoesNotCheckWitnessActivity() throws Exception {
+    TronNetDelegate delegate = Mockito.spy(new TronNetDelegate());
+    WitnessScheduleStore schedule = mock(WitnessScheduleStore.class);
+    setField(delegate, "witnessScheduleStore", schedule);
+    BlockCapsule block = new BlockCapsule(1, Sha256Hash.ZERO_HASH, 0,
+        ByteString.copyFromUtf8("witness"));
+    block.setMerkleRoot();
+    Mockito.doNothing().when(delegate).validSignature(block);
+    delegate.validBlock(block);
+    Mockito.verify(schedule, Mockito.never()).getActiveWitnesses();
+  }
+
+  @Test
+  public void testValidAdvBlockChecksSignatureAfterWitnessActivity() throws Exception {
+    TronNetDelegate delegate = Mockito.spy(new TronNetDelegate());
+    WitnessScheduleStore schedule = mock(WitnessScheduleStore.class);
+    setField(delegate, "witnessScheduleStore", schedule);
+    Mockito.when(schedule.getActiveWitnesses()).thenReturn(Collections.emptyList());
+    Mockito.doReturn(21_600_000L).when(delegate).getMaintenanceTimeInterval();
+    BlockCapsule block = new BlockCapsule(1, Sha256Hash.ZERO_HASH, 0, ByteString.EMPTY);
+    block.setMerkleRoot();
+    P2pException failure = new P2pException(TypeEnum.BLOCK_SIGN_INVALID, "invalid signature");
+    Mockito.doThrow(failure).when(delegate).validSignature(block);
+    PeerConnection peer = Mockito.spy(new PeerConnection());
+    Mockito.doReturn(new InetSocketAddress("127.0.0.1", 18888))
+        .when(peer).getInetSocketAddress();
+    try {
+      delegate.validAdvBlock(peer, block);
+      Assert.fail("Invalid signature must still be rejected after witness activity handling");
+    } catch (P2pException e) {
+      Assert.assertSame(failure, e);
+    }
+    Assert.assertTrue(peer.getLastInactiveWitnessBlockTime() > 0);
+  }
+
+  @Test
+  public void testRepeatedInactiveWitnessIsRejectedBeforeBasicValidation() throws Exception {
+    TronNetDelegate delegate = Mockito.spy(new TronNetDelegate());
+    WitnessScheduleStore schedule = mock(WitnessScheduleStore.class);
+    setField(delegate, "witnessScheduleStore", schedule);
+    Mockito.when(schedule.getActiveWitnesses()).thenReturn(Collections.emptyList());
+    Mockito.doReturn(21_600_000L).when(delegate).getMaintenanceTimeInterval();
+    BlockCapsule block = new BlockCapsule(1, Sha256Hash.ZERO_HASH, 0, ByteString.EMPTY);
+    PeerConnection peer = Mockito.spy(new PeerConnection());
+    Mockito.doReturn(new InetSocketAddress("127.0.0.1", 18888))
+        .when(peer).getInetSocketAddress();
+    peer.setLastInactiveWitnessBlockTime(System.currentTimeMillis());
+
+    try {
+      delegate.validAdvBlock(peer, block);
+      Assert.fail("Repeated inactive witness must be rejected before basic validation");
+    } catch (P2pException e) {
+      Assert.assertEquals(TypeEnum.BAD_BLOCK, e.getType());
+    }
+    Mockito.verify(delegate, Mockito.never()).validBlock(block);
+  }
+
+  @Test
+  public void testValidAdvBlockAcceptsActiveWitnessWithoutChangingHistory() throws Exception {
+    TronNetDelegate delegate = Mockito.spy(new TronNetDelegate());
+    WitnessScheduleStore schedule = mock(WitnessScheduleStore.class);
+    setField(delegate, "witnessScheduleStore", schedule);
+    BlockCapsule block = new BlockCapsule(1, Sha256Hash.ZERO_HASH, 0,
+        ByteString.copyFromUtf8("witness"));
+    block.setMerkleRoot();
+    Mockito.doNothing().when(delegate).validSignature(block);
+    Mockito.when(schedule.getActiveWitnesses())
+        .thenReturn(Collections.singletonList(block.getWitnessAddress()));
+    PeerConnection peer = new PeerConnection();
+    peer.setLastInactiveWitnessBlockTime(86_400_000L);
+
+    Assert.assertTrue(delegate.validAdvBlock(peer, block));
+
+    Mockito.verify(delegate).validSignature(block);
+    Assert.assertEquals(86_400_000L, peer.getLastInactiveWitnessBlockTime());
+  }
+
+  @Test
+  public void testProcessBlockWrapsScheduleFailureWithCause() throws Exception {
+    assertProcessBlockFailure(new ValidateScheduleException("invalid scheduled witness"));
+  }
+
+  @Test
+  public void testProcessBlockWrapsMerkleFailureWithCause() throws Exception {
+    assertProcessBlockFailure(new BadBlockException(
+        BadBlockException.TypeEnum.CALC_MERKLE_ROOT_FAILED, "invalid merkle root"));
+  }
+
+  private void assertProcessBlockFailure(Exception failure) throws Exception {
+    TronNetDelegate delegate = new TronNetDelegate();
+    Manager manager = mock(Manager.class);
+    ChainBaseManager chainManager = mock(ChainBaseManager.class);
+    setField(delegate, "dbManager", manager);
+    setField(delegate, "chainBaseManager", chainManager);
+    setField(delegate, "metricsService", mock(MetricsService.class));
+    Mockito.when(manager.getBlockedTimer()).thenReturn(new ThreadLocal<>());
+    Mockito.when(chainManager.getHeadBlockId())
+        .thenReturn(new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 0));
+    BlockCapsule block = new BlockCapsule(1, Sha256Hash.ZERO_HASH, 0, ByteString.EMPTY);
+    Mockito.doThrow(failure).when(manager).pushBlock(block);
+    try {
+      delegate.processBlock(block, true);
+      Assert.fail("Expected block validation failure");
+    } catch (P2pException e) {
+      Assert.assertEquals(TypeEnum.BAD_BLOCK, e.getType());
+      Assert.assertSame(failure, e.getCause());
+    }
   }
 }