Fix EthStats integration with updating blocks (besu-eth#9220)

* Fix EthStats not updating blocks

Signed-off-by: Jason Frame <jason.frame@consensys.net>

* Update gradle to include jackson-datatype-jdk8 for ethstats module

Signed-off-by: Jason Frame <jason.frame@consensys.net>

* changelog

Signed-off-by: Jason Frame <jason.frame@consensys.net>

* test for sending ethstats block message

Signed-off-by: Jason Frame <jason.frame@consensys.net>

* rename test

Signed-off-by: Jason Frame <jason.frame@consensys.net>

* rename test

Signed-off-by: Jason Frame <jason.frame@consensys.net>

---------

Signed-off-by: Jason Frame <jason.frame@consensys.net>
Co-authored-by: Fabio Di Fabio <fabio.difabio@consensys.net>
Signed-off-by: jflo <justin+github@florentine.us>

Author: 2 people

CHANGELOG.md

@@ -18,6 +18,8 @@
 - Implement optional sender balance checks in the layered txpool [#9176](https://github.com/hyperledger/besu/pull/9176)
 
 ### Bug fixes
+- Fix eth_subscribe RPC failing returning a block response [#9212](https://github.com/hyperledger/besu/pull/9212)
+- Fix ethstats integration failing to provide block updates to ethstats server [#9220](https://github.com/hyperledger/besu/pull/9220)
 
 ## 25.9.0
 

ethereum/ethstats/build.gradle

@@ -32,6 +32,7 @@ dependencies {
   api 'org.slf4j:slf4j-api'
 
   implementation 'com.fasterxml.jackson.core:jackson-databind'
+  implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jdk8'
   implementation 'com.google.guava:guava'
   implementation 'com.squareup.okhttp3:okhttp'
   implementation 'io.vertx:vertx-core'

ethereum/ethstats/src/main/java/org/hyperledger/besu/ethstats/EthStatsService.java

@@ -71,6 +71,7 @@
 import java.util.stream.LongStream;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.annotations.VisibleForTesting;
 import io.vertx.core.Vertx;
 import io.vertx.core.http.WebSocket;
 import io.vertx.core.http.WebSocketClientOptions;
@@ -366,7 +367,8 @@ private void sendLatencyReport() {
   }
 
   /** Sends a block report concerning the last block */
-  private void sendBlockReport() {
+  @VisibleForTesting
+  protected void sendBlockReport() {
     blockchainQueries
         .latestBlock()
         .map(tx -> blockResultFactory.transactionComplete(tx, false))

ethereum/ethstats/src/main/java/org/hyperledger/besu/ethstats/request/EthStatsRequest.java

@@ -22,6 +22,7 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
 
 /**
  * This class represents an Ethereum statistics request. It provides methods to get the type of the
@@ -30,7 +31,7 @@
 public class EthStatsRequest {
 
   /** The constant MAPPER. */
-  public static final ObjectMapper MAPPER = new ObjectMapper();
+  public static final ObjectMapper MAPPER = new ObjectMapper().registerModule(new Jdk8Module());
 
   /** The constant EMIT_FIELD. */
   public static final String EMIT_FIELD = "emit";

ethereum/ethstats/src/test/java/org/hyperledger/besu/ethstats/EthStatsServiceTest.java

@@ -22,8 +22,14 @@
 import static org.mockito.Mockito.when;
 
 import org.hyperledger.besu.config.GenesisConfigOptions;
+import org.hyperledger.besu.datatypes.Hash;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.BlockResultFactory;
+import org.hyperledger.besu.ethereum.api.query.BlockWithMetadata;
 import org.hyperledger.besu.ethereum.api.query.BlockchainQueries;
+import org.hyperledger.besu.ethereum.api.query.TransactionWithMetadata;
 import org.hyperledger.besu.ethereum.blockcreation.MiningCoordinator;
+import org.hyperledger.besu.ethereum.core.Block;
+import org.hyperledger.besu.ethereum.core.BlockDataGenerator;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
 import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
@@ -43,6 +49,9 @@
 import java.util.NoSuchElementException;
 import java.util.Optional;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
@@ -226,6 +235,67 @@ public void shouldSendFullReportIfHelloMessageSucceeded() {
     verify(ethScheduler, times(1)).scheduleFutureTaskWithFixedDelay(any(), any(), any());
   }
 
+  @Test
+  public void shouldSendBlockMessage() throws Exception {
+    ethStatsService =
+        new EthStatsService(
+            ethStatsConnectOptions,
+            blockchainQueries,
+            ethProtocolManager,
+            transactionPool,
+            miningCoordinator,
+            syncState,
+            vertx,
+            "clientVersion",
+            genesisConfigOptions,
+            p2PNetwork);
+
+    final BlockDataGenerator blockDataGenerator = new BlockDataGenerator();
+    final Block testBlock = blockDataGenerator.block();
+    final BlockWithMetadata<TransactionWithMetadata, Hash> blockWithMetadata =
+        new BlockWithMetadata<>(
+            testBlock.getHeader(),
+            List.of(),
+            List.of(),
+            testBlock.getHeader().getDifficulty(),
+            testBlock.getSize(),
+            Optional.empty());
+
+    when(p2PNetwork.getLocalEnode()).thenReturn(Optional.of(node));
+    when(blockchainQueries.latestBlock()).thenReturn(Optional.of(blockWithMetadata));
+
+    final ArgumentCaptor<Handler<AsyncResult<WebSocket>>> webSocketCaptor =
+        ArgumentCaptor.forClass(Handler.class);
+
+    ethStatsService.start();
+
+    verify(webSocketClient, times(1))
+        .connect(any(WebSocketConnectOptions.class), webSocketCaptor.capture());
+    webSocketCaptor.getValue().handle(succeededWebSocketEvent(Optional.of(webSocket)));
+
+    // send block message
+    ethStatsService.sendBlockReport();
+    final ArgumentCaptor<String> messagesCaptor = ArgumentCaptor.forClass(String.class);
+    verify(webSocket, times(2)).writeTextMessage(messagesCaptor.capture(), any(Handler.class));
+
+    final List<String> sentMessages = messagesCaptor.getAllValues();
+    assertThat(sentMessages.get(0)).contains("hello");
+
+    final String blockMessage = sentMessages.get(1);
+    assertThat(blockMessage).contains("\"block\"");
+
+    // verify block message
+    final ObjectMapper objectMapper = new ObjectMapper().registerModule(new Jdk8Module());
+    final var jsonNode = objectMapper.readTree(blockMessage);
+    final var blockReportData = jsonNode.get("emit").get(1);
+    final var blockDataNode = blockReportData.get("block");
+
+    final var expectedBlockResult = new BlockResultFactory().transactionComplete(blockWithMetadata);
+    final JsonNode expectedBlockResultNode = objectMapper.valueToTree(expectedBlockResult);
+
+    assertThat(blockDataNode).isEqualTo(expectedBlockResultNode);
+  }
+
   private <T> AsyncResult<T> succeededWebSocketEvent(final Optional<T> object) {
     return new AsyncResult<>() {
       @Override