Add totalDiffculty to BlockPropagated events. (#97)

* Add totalDiffculty to BlockPropagated events.
The chain head block can be contentious with many fork blocks (ommers) propagating on the network.  We should add a totalDifficulty to make it easier to see which block is most likely the current head.
- added `BlockPropagated` interface in `plugin-api`.
- updated `BesuEvents.onBlockPropagated` method to take a `BlockPropagated` instead of a `BlockHeader`.
- created `BlockPropagatedSubscriber` in `BlockBroadcaster`.
- changed type of `BlockBroadcaster.blockPropagatedSubscribers` from `Consumer<Block>` to `BlockPropagatedSubscriber`.
- updated unit tests accordingly to all changes.
- updated known hash in `build.gradle` file of `plugin-api`: new value is `4SAeaZIJMsDvUK5Wp2RzU8TlHacslALnM/4yvVhsMtY=`

Signed-off-by: Abdelhamid Bakhta <[email protected]>

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

@@ -17,8 +17,13 @@
 import org.hyperledger.besu.ethereum.eth.sync.BlockBroadcaster;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
+import org.hyperledger.besu.plugin.data.BlockHeader;
+import org.hyperledger.besu.plugin.data.PropagatedBlockContext;
+import org.hyperledger.besu.plugin.data.Quantity;
 import org.hyperledger.besu.plugin.services.BesuEvents;
 
+import java.util.function.Supplier;
+
 public class BesuEventsImpl implements BesuEvents {
  private final BlockBroadcaster blockBroadcaster;
  private final TransactionPool transactionPool;
@@ -36,7 +41,9 @@ public BesuEventsImpl(
  @Override
  public long addBlockPropagatedListener(final BlockPropagatedListener listener) {
  return blockBroadcaster.subscribePropagateNewBlocks(
- block -> listener.onBlockPropagated(block.getHeader()));
+ (block, totalDifficulty) ->
+ listener.onBlockPropagated(
+ blockPropagatedContext(block::getHeader, () -> totalDifficulty)));
  }
 
  @Override
@@ -75,4 +82,20 @@ public long addSyncStatusListener(final SyncStatusListener syncStatusListener) {
  public void removeSyncStatusListener(final long listenerIdentifier) {
  syncState.removeSyncStatusListener(listenerIdentifier);
  }
+
+ private static PropagatedBlockContext blockPropagatedContext(
+ final Supplier<BlockHeader> blockHeaderSupplier,
+ final Supplier<Quantity> totalDifficultySupplier) {
+ return new PropagatedBlockContext() {
+ @Override
+ public BlockHeader getBlockHeader() {
+ return blockHeaderSupplier.get();
+ }
+
+ @Override
+ public Quantity getTotalDifficulty() {
+ return totalDifficultySupplier.get();
+ }
+ };
+ }
 }

besu/src/test/java/org/hyperledger/besu/plugins/TestBesuEventsPlugin.java

@@ -17,6 +17,7 @@
 import org.hyperledger.besu.plugin.BesuContext;
 import org.hyperledger.besu.plugin.BesuPlugin;
 import org.hyperledger.besu.plugin.data.BlockHeader;
+import org.hyperledger.besu.plugin.data.PropagatedBlockContext;
 import org.hyperledger.besu.plugin.services.BesuEvents;
 
 import java.io.File;
@@ -66,7 +67,8 @@ public void stop() {
  LOG.info("No longer listening with ID#" + subscriptionId);
  }
 
- private void onBlockAnnounce(final BlockHeader header) {
+ private void onBlockAnnounce(final PropagatedBlockContext propagatedBlockContext) {
+ final BlockHeader header = propagatedBlockContext.getBlockHeader();
  final int blockCount = blockCounter.incrementAndGet();
  LOG.info("I got a new block! (I've seen {}) - {}", blockCount, header);
  try {

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

@@ -43,7 +43,7 @@
 import org.hyperledger.besu.ethereum.mainnet.ValidationResult;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
-import org.hyperledger.besu.plugin.data.BlockHeader;
+import org.hyperledger.besu.plugin.data.PropagatedBlockContext;
 import org.hyperledger.besu.plugin.data.SyncStatus;
 import org.hyperledger.besu.plugin.data.Transaction;
 import org.hyperledger.besu.testutil.TestClock;
@@ -144,24 +144,29 @@ public void syncStatusEventDoesNotFireAfterUnsubscribe() {
 
  @Test
  public void newBlockEventFiresAfterSubscribe() {
- final AtomicReference<BlockHeader> result = new AtomicReference<>();
+ final AtomicReference<PropagatedBlockContext> result = new AtomicReference<>();
  serviceImpl.addBlockPropagatedListener(result::set);
-
+ final Block block = generateBlock();
  assertThat(result.get()).isNull();
- blockBroadcaster.propagate(generateBlock(), UInt256.of(1));
+ blockBroadcaster.propagate(block, UInt256.of(1));
 
  assertThat(result.get()).isNotNull();
+ assertThat(result.get().getBlockHeader()).isEqualTo(block.getHeader());
+ assertThat(result.get().getTotalDifficulty()).isEqualTo(UInt256.of(1));
  }
 
  @Test
  public void newBlockEventDoesNotFireAfterUnsubscribe() {
- final AtomicReference<BlockHeader> result = new AtomicReference<>();
+ final AtomicReference<PropagatedBlockContext> result = new AtomicReference<>();
  final long id = serviceImpl.addBlockPropagatedListener(result::set);
 
  assertThat(result.get()).isNull();
- blockBroadcaster.propagate(generateBlock(), UInt256.of(1));
+ final Block block = generateBlock();
+ blockBroadcaster.propagate(block, UInt256.of(2));
 
  assertThat(result.get()).isNotNull();
+ assertThat(result.get().getBlockHeader()).isEqualTo(block.getHeader());
+ assertThat(result.get().getTotalDifficulty()).isEqualTo(UInt256.of(2));
  serviceImpl.removeBlockPropagatedListener(id);
  result.set(null);
 

build.gradle

@@ -268,16 +268,16 @@ allprojects {
 
 task deploy() {}
 
-task checkMavenCoordianteCollisions {
+task checkMavenCoordinateCollisions {
  doLast {
  def coordinates = [:]
  getAllprojects().forEach {
  if (it.properties.containsKey('publishing')) {
  def coordinate = it.publishing?.publications[0].coordinates
  if (coordinates.containsKey(coordinate)) {
  throw new GradleException("Duplicate maven coordinates detected, ${coordinate} is used by " +
-  "both ${coordinates[coordinate]} and ${it.path}.\n" +
-  "Please add a `publishing` script block to one or both subprojects.")
+ "both ${coordinates[coordinate]} and ${it.path}.\n" +
+ "Please add a `publishing` script block to one or both subprojects.")
  }
  coordinates[coordinate] = it.path
  }
@@ -287,7 +287,7 @@ task checkMavenCoordianteCollisions {
 
 tasks.register('checkPluginAPIChanges', DefaultTask) { }
 checkPluginAPIChanges.dependsOn(':plugin-api:checkAPIChanges')
-check.dependsOn('checkPluginAPIChanges', 'checkMavenCoordianteCollisions')
+check.dependsOn('checkPluginAPIChanges', 'checkMavenCoordinateCollisions')
 
 subprojects {
 

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/BlockBroadcaster.java

@@ -21,22 +21,21 @@
 import org.hyperledger.besu.util.Subscribers;
 import org.hyperledger.besu.util.uint.UInt256;
 
-import java.util.function.Consumer;
-
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 public class BlockBroadcaster {
  private static final Logger LOG = LogManager.getLogger();
 
  private final EthContext ethContext;
- private final Subscribers<Consumer<Block>> blockPropagatedSubscribers = Subscribers.create();
+ private final Subscribers<BlockPropagatedSubscriber> blockPropagatedSubscribers =
+ Subscribers.create();
 
  public BlockBroadcaster(final EthContext ethContext) {
  this.ethContext = ethContext;
  }
 
- public long subscribePropagateNewBlocks(final Consumer<Block> callback) {
+ public long subscribePropagateNewBlocks(final BlockPropagatedSubscriber callback) {
  return blockPropagatedSubscribers.subscribe(callback);
  }
 
@@ -45,7 +44,7 @@ public void unsubscribePropagateNewBlocks(final long id) {
  }
 
  public void propagate(final Block block, final UInt256 totalDifficulty) {
- blockPropagatedSubscribers.forEach(listener -> listener.accept(block));
+ blockPropagatedSubscribers.forEach(listener -> listener.accept(block, totalDifficulty));
  final NewBlockMessage newBlockMessage = NewBlockMessage.create(block, totalDifficulty);
  ethContext
  .getEthPeers()
@@ -61,4 +60,9 @@ public void propagate(final Block block, final UInt256 totalDifficulty) {
  }
  });
  }
+
+ @FunctionalInterface
+ public interface BlockPropagatedSubscriber {
+ void accept(Block block, UInt256 totalDifficulty);
+ }
 }

plugin-api/build.gradle

@@ -56,7 +56,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 = 'b8BCiNvy9vSYsTtS1NfszsVT0EMV8tiNc7igzXzTrak='
+ knownHash = 'XMBMdVNS2VAa+WN3lyEAKmXznu4ExSIHA2arYUeo0eo='
 }
 check.dependsOn('checkAPIChanges')
 

plugin-api/src/main/java/org/hyperledger/besu/plugin/data/PropagatedBlockContext.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright ConsenSys AG.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.plugin.data;
+
+import org.hyperledger.besu.plugin.Unstable;
+
+/** The minimum set of data for a PropagatedBlockContext. */
+@Unstable
+public interface PropagatedBlockContext {
+
+ /**
+ * A {@link BlockHeader} object.
+ *
+ * @return A {@link BlockHeader}
+ */
+ BlockHeader getBlockHeader();
+
+ /**
+ * A scalar value corresponding to the total difficulty.
+ *
+ * @return A scalar value corresponding to the total difficulty.
+ */
+ Quantity getTotalDifficulty();
+}

plugin-api/src/main/java/org/hyperledger/besu/plugin/services/BesuEvents.java

@@ -15,7 +15,7 @@
 package org.hyperledger.besu.plugin.services;
 
 import org.hyperledger.besu.plugin.Unstable;
-import org.hyperledger.besu.plugin.data.BlockHeader;
+import org.hyperledger.besu.plugin.data.PropagatedBlockContext;
 import org.hyperledger.besu.plugin.data.SyncStatus;
 import org.hyperledger.besu.plugin.data.Transaction;
 
@@ -109,9 +109,9 @@ interface BlockPropagatedListener {
  * <p>The block may not have been imported to the local chain yet and may fail later
  * validations.
  *
- * @param blockHeader the new block header.
+ * @param propagatedBlockContext block being propagated.
  */
- void onBlockPropagated(BlockHeader blockHeader);
+ void onBlockPropagated(PropagatedBlockContext propagatedBlockContext);
  }
 
  /** The listener interface for receiving new transaction added events. */