[Issue-6301] Add bad block events (hyperledger#6848)

Signed-off-by: mbaxter <mbaxter.dev@gmail.com>

CHANGELOG.md

@@ -38,6 +38,7 @@
 - Add `tx-pool-blob-price-bump` option to configure the price bump percentage required to replace blob transactions (by default 100%) [#6874](https://github.com/hyperledger/besu/pull/6874)
 - Log detailed timing of block creation steps [#6880](https://github.com/hyperledger/besu/pull/6880)
 - Expose transaction count by type metrics for the layered txpool [#6903](https://github.com/hyperledger/besu/pull/6903)
+- Expose bad block events via the BesuEvents plugin API  [#6848](https://github.com/hyperledger/besu/pull/6848)
 
 ### Bug fixes
 - Fix txpool dump/restore race condition [#6665](https://github.com/hyperledger/besu/pull/6665)

acceptance-tests/dsl/src/main/java/org/hyperledger/besu/tests/acceptance/dsl/node/ThreadBesuNodeRunner.java

@@ -313,7 +313,8 @@ public void startNode(final BesuNode node) {
             besuController.getProtocolContext().getBlockchain(),
             besuController.getProtocolManager().getBlockBroadcaster(),
             besuController.getTransactionPool(),
-            besuController.getSyncState()));
+            besuController.getSyncState(),
+            besuController.getProtocolContext().getBadBlockManager()));
     besuPluginContext.startPlugins();
 
     runner.startEthereumMainLoop();

besu/src/main/java/org/hyperledger/besu/cli/BesuCommand.java

@@ -1293,7 +1293,8 @@ private void startPlugins() {
             besuController.getProtocolContext().getBlockchain(),
             besuController.getProtocolManager().getBlockBroadcaster(),
             besuController.getTransactionPool(),
-            besuController.getSyncState()));
+            besuController.getSyncState(),
+            besuController.getProtocolContext().getBadBlockManager()));
     besuPluginContext.addService(MetricsSystem.class, getMetricsSystem());
 
     besuPluginContext.addService(

besu/src/main/java/org/hyperledger/besu/services/BesuEventsImpl.java

@@ -18,6 +18,7 @@
 
 import org.hyperledger.besu.datatypes.Address;
 import org.hyperledger.besu.ethereum.api.query.LogsQuery;
+import org.hyperledger.besu.ethereum.chain.BadBlockManager;
 import org.hyperledger.besu.ethereum.chain.Blockchain;
 import org.hyperledger.besu.ethereum.core.BlockBody;
 import org.hyperledger.besu.ethereum.core.Difficulty;
@@ -44,6 +45,7 @@ public class BesuEventsImpl implements BesuEvents {
   private final BlockBroadcaster blockBroadcaster;
   private final TransactionPool transactionPool;
   private final SyncState syncState;
+  private final BadBlockManager badBlockManager;
 
   /**
    * Constructor for BesuEventsImpl
@@ -52,16 +54,19 @@ public class BesuEventsImpl implements BesuEvents {
    * @param blockBroadcaster An instance of BlockBroadcaster
    * @param transactionPool An instance of TransactionPool
    * @param syncState An instance of SyncState
+   * @param badBlockManager A cache of bad blocks encountered on the network
    */
   public BesuEventsImpl(
       final Blockchain blockchain,
       final BlockBroadcaster blockBroadcaster,
       final TransactionPool transactionPool,
-      final SyncState syncState) {
+      final SyncState syncState,
+      final BadBlockManager badBlockManager) {
     this.blockchain = blockchain;
     this.blockBroadcaster = blockBroadcaster;
     this.transactionPool = transactionPool;
     this.syncState = syncState;
+    this.badBlockManager = badBlockManager;
   }
 
   @Override
@@ -166,6 +171,16 @@ public void removeLogListener(final long listenerIdentifier) {
     blockchain.removeObserver(listenerIdentifier);
   }
 
+  @Override
+  public long addBadBlockListener(final BadBlockListener listener) {
+    return badBlockManager.subscribeToBadBlocks(listener);
+  }
+
+  @Override
+  public void removeBadBlockListener(final long listenerIdentifier) {
+    badBlockManager.unsubscribeFromBadBlocks(listenerIdentifier);
+  }
+
   private static PropagatedBlockContext blockPropagatedContext(
       final Supplier<BlockHeader> blockHeaderSupplier,
       final Supplier<BlockBody> blockBodySupplier,

besu/src/test/java/org/hyperledger/besu/services/BesuEventsImplTest.java

@@ -27,6 +27,8 @@
 import org.hyperledger.besu.datatypes.Transaction;
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.ProtocolContext;
+import org.hyperledger.besu.ethereum.chain.BadBlockCause;
+import org.hyperledger.besu.ethereum.chain.BadBlockManager;
 import org.hyperledger.besu.ethereum.chain.DefaultBlockchain;
 import org.hyperledger.besu.ethereum.chain.MutableBlockchain;
 import org.hyperledger.besu.ethereum.core.Block;
@@ -113,6 +115,7 @@ public class BesuEventsImplTest {
   private BesuEventsImpl serviceImpl;
   private MutableBlockchain blockchain;
   private final BlockDataGenerator gen = new BlockDataGenerator();
+  private final BadBlockManager badBlockManager = new BadBlockManager();
 
   @BeforeEach
   public void setUp() {
@@ -171,7 +174,9 @@ public void setUp() {
             new BlobCache(),
             MiningParameters.newDefault());
 
-    serviceImpl = new BesuEventsImpl(blockchain, blockBroadcaster, transactionPool, syncState);
+    serviceImpl =
+        new BesuEventsImpl(
+            blockchain, blockBroadcaster, transactionPool, syncState, badBlockManager);
   }
 
   @Test
@@ -504,6 +509,85 @@ public void logEventDoesNotFireAfterUnsubscribe() {
     assertThat(result).isEmpty();
   }
 
+  @Test
+  public void badBlockEventFiresAfterSubscribe_badBlockAdded() {
+    // Track bad block events
+    final AtomicReference<org.hyperledger.besu.plugin.data.BlockHeader> badBlockResult =
+        new AtomicReference<>();
+    final AtomicReference<org.hyperledger.besu.plugin.data.BadBlockCause> badBlockCauseResult =
+        new AtomicReference<>();
+
+    serviceImpl.addBadBlockListener(
+        (badBlock, cause) -> {
+          badBlockResult.set(badBlock);
+          badBlockCauseResult.set(cause);
+        });
+
+    // Add bad block
+    final BadBlockCause blockCause = BadBlockCause.fromValidationFailure("failed");
+    final Block block = gen.block(new BlockDataGenerator.BlockOptions());
+    badBlockManager.addBadBlock(block, blockCause);
+
+    // Check we caught the bad block
+    assertThat(badBlockResult.get()).isEqualTo(block.getHeader());
+    assertThat(badBlockCauseResult.get()).isEqualTo(blockCause);
+  }
+
+  @Test
+  public void badBlockEventFiresAfterSubscribe_badBlockHeaderAdded() {
+    // Track bad block events
+    final AtomicReference<org.hyperledger.besu.plugin.data.BlockHeader> badBlockResult =
+        new AtomicReference<>();
+    final AtomicReference<org.hyperledger.besu.plugin.data.BadBlockCause> badBlockCauseResult =
+        new AtomicReference<>();
+
+    serviceImpl.addBadBlockListener(
+        (badBlock, cause) -> {
+          badBlockResult.set(badBlock);
+          badBlockCauseResult.set(cause);
+        });
+
+    // Add bad block header
+    final BadBlockCause cause = BadBlockCause.fromValidationFailure("oops");
+    final Block badBlock = gen.block(new BlockDataGenerator.BlockOptions());
+    badBlockManager.addBadHeader(badBlock.getHeader(), cause);
+
+    // Check we caught the bad block
+    assertThat(badBlockResult.get()).isEqualTo(badBlock.getHeader());
+    assertThat(badBlockCauseResult.get()).isEqualTo(cause);
+  }
+
+  @Test
+  public void badBlockEventDoesNotFireAfterUnsubscribe() {
+    // Track bad block events
+    final AtomicReference<org.hyperledger.besu.plugin.data.BlockHeader> badBlockResult =
+        new AtomicReference<>();
+    final AtomicReference<org.hyperledger.besu.plugin.data.BadBlockCause> badBlockCauseResult =
+        new AtomicReference<>();
+
+    final long listenerId =
+        serviceImpl.addBadBlockListener(
+            (badBlock, cause) -> {
+              badBlockResult.set(badBlock);
+              badBlockCauseResult.set(cause);
+            });
+    // Unsubscribe
+    serviceImpl.removeBadBlockListener(listenerId);
+
+    // Add bad block
+    final BadBlockCause blockCause = BadBlockCause.fromValidationFailure("failed");
+    final Block block = gen.block(new BlockDataGenerator.BlockOptions());
+    badBlockManager.addBadBlock(block, blockCause);
+    // Add bad block header
+    final BadBlockCause headerCause = BadBlockCause.fromValidationFailure("oops");
+    final Block headerBlock = gen.block(new BlockDataGenerator.BlockOptions());
+    badBlockManager.addBadHeader(headerBlock.getHeader(), headerCause);
+
+    // Check we did not process any events
+    assertThat(badBlockResult.get()).isNull();
+    assertThat(badBlockCauseResult.get()).isNull();
+  }
+
   private Block generateBlock() {
     final BlockBody body = new BlockBody(Collections.emptyList(), Collections.emptyList());
     return new Block(new BlockHeaderTestFixture().buildHeader(), body);

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/chain/BadBlockCause.java

@@ -15,14 +15,14 @@
  */
 package org.hyperledger.besu.ethereum.chain;
 
-import static org.hyperledger.besu.ethereum.chain.BadBlockReason.DESCENDS_FROM_BAD_BLOCK;
-import static org.hyperledger.besu.ethereum.chain.BadBlockReason.SPEC_VALIDATION_FAILURE;
+import static org.hyperledger.besu.plugin.data.BadBlockCause.BadBlockReason.DESCENDS_FROM_BAD_BLOCK;
+import static org.hyperledger.besu.plugin.data.BadBlockCause.BadBlockReason.SPEC_VALIDATION_FAILURE;
 
 import org.hyperledger.besu.ethereum.core.Block;
 
 import com.google.common.base.MoreObjects;
 
-public class BadBlockCause {
+public class BadBlockCause implements org.hyperledger.besu.plugin.data.BadBlockCause {
 
   private final BadBlockReason reason;
   private final String description;
@@ -42,10 +42,12 @@ private BadBlockCause(final BadBlockReason reason, final String description) {
     this.description = description;
   }
 
+  @Override
   public BadBlockReason getReason() {
     return reason;
   }
 
+  @Override
   public String getDescription() {
     return description;
   }

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/chain/BadBlockManager.java

@@ -17,6 +17,8 @@
 import org.hyperledger.besu.datatypes.Hash;
 import org.hyperledger.besu.ethereum.core.Block;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
+import org.hyperledger.besu.plugin.services.BesuEvents.BadBlockListener;
+import org.hyperledger.besu.util.Subscribers;
 
 import java.util.Collection;
 import java.util.Optional;
@@ -37,6 +39,7 @@ public class BadBlockManager {
       CacheBuilder.newBuilder().maximumSize(MAX_BAD_BLOCKS_SIZE).concurrencyLevel(1).build();
   private final Cache<Hash, Hash> latestValidHashes =
       CacheBuilder.newBuilder().maximumSize(MAX_BAD_BLOCKS_SIZE).concurrencyLevel(1).build();
+  private final Subscribers<BadBlockListener> badBlockSubscribers = Subscribers.create(true);
 
   /**
    * Add a new invalid block.
@@ -45,9 +48,9 @@ public class BadBlockManager {
    * @param cause the cause detailing why the block is considered invalid
    */
   public void addBadBlock(final Block badBlock, final BadBlockCause cause) {
-    // TODO(#6301) Expose bad block with cause through BesuEvents
     LOG.debug("Register bad block {} with cause: {}", badBlock.toLogString(), cause);
     this.badBlocks.put(badBlock.getHash(), badBlock);
+    badBlockSubscribers.forEach(s -> s.onBadBlockAdded(badBlock.getHeader(), cause));
   }
 
   public void reset() {
@@ -81,9 +84,9 @@ public Optional<Block> getBadBlock(final Hash hash) {
   }
 
   public void addBadHeader(final BlockHeader header, final BadBlockCause cause) {
-    // TODO(#6301) Expose bad block header with cause through BesuEvents
     LOG.debug("Register bad block header {} with cause: {}", header.toLogString(), cause);
     badHeaders.put(header.getHash(), header);
+    badBlockSubscribers.forEach(s -> s.onBadBlockAdded(header, cause));
   }
 
   public boolean isBadBlock(final Hash blockHash) {
@@ -97,4 +100,12 @@ public void addLatestValidHash(final Hash blockHash, final Hash latestValidHash)
   public Optional<Hash> getLatestValidHash(final Hash blockHash) {
     return Optional.ofNullable(latestValidHashes.getIfPresent(blockHash));
   }
+
+  public long subscribeToBadBlocks(final BadBlockListener listener) {
+    return badBlockSubscribers.subscribe(listener);
+  }
+
+  public void unsubscribeFromBadBlocks(final long id) {
+    badBlockSubscribers.unsubscribe(id);
+  }
 }

ethereum/core/src/test/java/org/hyperledger/besu/ethereum/chain/BadBlockManagerTest.java

@@ -20,12 +20,16 @@
 import org.hyperledger.besu.ethereum.core.Block;
 import org.hyperledger.besu.ethereum.core.BlockchainSetupUtil;
 
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
 import org.junit.jupiter.api.Test;
 
 public class BadBlockManagerTest {
 
   final BlockchainSetupUtil chainUtil = BlockchainSetupUtil.forMainnet();
   final Block block = chainUtil.getBlock(1);
+  final Block block2 = chainUtil.getBlock(2);
   final BadBlockManager badBlockManager = new BadBlockManager();
 
   @Test
@@ -66,4 +70,64 @@ public void isBadBlock_trueForBadBlock() {
 
     assertThat(badBlockManager.isBadBlock(block.getHash())).isTrue();
   }
+
+  @Test
+  public void subscribeToBadBlocks_listenerReceivesBadBlockEvent() {
+
+    final AtomicReference<org.hyperledger.besu.plugin.data.BlockHeader> badBlockResult =
+        new AtomicReference<>();
+    final AtomicReference<org.hyperledger.besu.plugin.data.BadBlockCause> badBlockCauseResult =
+        new AtomicReference<>();
+
+    badBlockManager.subscribeToBadBlocks(
+        (badBlock, cause) -> {
+          badBlockResult.set(badBlock);
+          badBlockCauseResult.set(cause);
+        });
+
+    final BadBlockCause cause = BadBlockCause.fromValidationFailure("fail");
+    badBlockManager.addBadBlock(block, cause);
+
+    // Check event was emitted
+    assertThat(badBlockResult.get()).isEqualTo(block.getHeader());
+    assertThat(badBlockCauseResult.get()).isEqualTo(cause);
+  }
+
+  @Test
+  public void subscribeToBadBlocks_listenerReceivesBadHeaderEvent() {
+
+    final AtomicReference<org.hyperledger.besu.plugin.data.BlockHeader> badBlockResult =
+        new AtomicReference<>();
+    final AtomicReference<org.hyperledger.besu.plugin.data.BadBlockCause> badBlockCauseResult =
+        new AtomicReference<>();
+
+    badBlockManager.subscribeToBadBlocks(
+        (badBlock, cause) -> {
+          badBlockResult.set(badBlock);
+          badBlockCauseResult.set(cause);
+        });
+
+    final BadBlockCause cause = BadBlockCause.fromValidationFailure("fail");
+    badBlockManager.addBadHeader(block.getHeader(), cause);
+
+    // Check event was emitted
+    assertThat(badBlockResult.get()).isEqualTo(block.getHeader());
+    assertThat(badBlockCauseResult.get()).isEqualTo(cause);
+  }
+
+  @Test
+  public void unsubscribeFromBadBlocks_listenerReceivesNoEvents() {
+
+    final AtomicInteger eventCount = new AtomicInteger(0);
+    final long subscribeId =
+        badBlockManager.subscribeToBadBlocks((block, cause) -> eventCount.incrementAndGet());
+    badBlockManager.unsubscribeFromBadBlocks(subscribeId);
+
+    final BadBlockCause cause = BadBlockCause.fromValidationFailure("fail");
+    badBlockManager.addBadBlock(block, cause);
+    badBlockManager.addBadHeader(block2.getHeader(), cause);
+
+    // Check no events fired
+    assertThat(eventCount.get()).isEqualTo(0);
+  }
 }

plugin-api/build.gradle

@@ -69,7 +69,7 @@ Calculated : ${currentHash}
 tasks.register('checkAPIChanges', FileStateChecker) {
   description = "Checks that the API for the Plugin-API project does not change without deliberate thought"
   files = sourceSets.main.allJava.files
-  knownHash = 'lsBecdCyK9rIi5FIjURF2uPwKzXgqHCayMcLyOOl4fE='
+  knownHash = '0mJiCGsToqx5aAJEvwnT3V0R8o4PXBYWiB0wT6CMpuo='
 }
 check.dependsOn('checkAPIChanges')