Add WebSocketBlockchainListener to fetch latest block without polling the blockchain network (#747)

Author: jbern0rd

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
 
 ## [[NEXT]](https://github.com/iExecBlockchainComputing/iexec-core/releases/tag/vNEXT) 2025
 
+### New Features
+
+- Add WebSocketBlockchainListener to fetch latest block without polling the blockchain network. (#747)
+
 ### Bug Fixes
 
 - Remove TaskResultUploadTimeout detector, covered by FinalDeadlineTaskDetector. (#732)

docker-compose.yml

@@ -0,0 +1,18 @@
+services:
+
+  chain:
+    image: docker-regis.iex.ec/poco-chain:1.0.0-poco-v5.5.0-voucher-v1.0.0-nethermind
+    expose:
+      - "8545"
+
+  config-server:
+    image: wiremock/wiremock:3.3.1
+    expose:
+      - "8080"
+    volumes:
+      - "./src/test/resources/wiremock/mappings:/home/wiremock/mappings"
+
+  mongo:
+    image: library/mongo:7.0.15-jammy
+    expose:
+      - "27017"

iexec-task-api/src/main/java/com/iexec/core/metric/PlatformMetric.java

@@ -35,6 +35,7 @@ public class PlatformMetric {
     int aliveComputingGpu;
     int aliveRegisteredGpu;
     LinkedHashMap<TaskStatus, Long> currentTaskStatusesCount;
+    LatestBlockMetric latestBlockMetric;
     long dealEventsCount;
     long dealsCount;
     long replayDealsCount;
@@ -66,4 +67,7 @@ public class PlatformMetric {
     @JsonPOJOBuilder(withPrefix = "")
     public static class PlatformMetricBuilder {
     }
+
+    public record LatestBlockMetric(long blockNumber, String blockHash, long blockTimestamp) {
+    }
 }

src/main/java/com/iexec/core/chain/BlockchainConnectionHealthIndicator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023-2024 IEXEC BLOCKCHAIN TECH
+ * Copyright 2023-2025 IEXEC BLOCKCHAIN TECH
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@
 
 import com.iexec.core.chain.event.ChainConnectedEvent;
 import com.iexec.core.chain.event.ChainDisconnectedEvent;
+import com.iexec.core.chain.event.LatestBlockEvent;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -31,6 +32,7 @@
 
 import java.time.Clock;
 import java.time.Duration;
+import java.time.Instant;
 import java.time.LocalDateTime;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -41,7 +43,6 @@
 public class BlockchainConnectionHealthIndicator implements HealthIndicator {
 
     private final ApplicationEventPublisher applicationEventPublisher;
-    private final Web3jService web3jService;
     /**
      * Interval between 2 requests onto the chain.
      */
@@ -57,6 +58,8 @@ public class BlockchainConnectionHealthIndicator implements HealthIndicator {
     private int consecutiveFailures = 1;
     @Getter
     private LocalDateTime firstFailure = null;
+    private long latestBlockTimestamp;
+    private long latestBlockNumber;
 
     /**
      * Required for test purposes.
@@ -66,11 +69,9 @@ public class BlockchainConnectionHealthIndicator implements HealthIndicator {
     @Autowired
     public BlockchainConnectionHealthIndicator(
             ApplicationEventPublisher applicationEventPublisher,
-            Web3jService web3jService,
             ChainConfig chainConfig) {
         this(
                 applicationEventPublisher,
-                web3jService,
                 chainConfig,
                 Executors.newSingleThreadScheduledExecutor(),
                 Clock.systemDefaultZone()
@@ -79,12 +80,10 @@ public BlockchainConnectionHealthIndicator(
 
     BlockchainConnectionHealthIndicator(
             ApplicationEventPublisher applicationEventPublisher,
-            Web3jService web3jService,
             ChainConfig chainConfig,
             ScheduledExecutorService monitoringExecutor,
             Clock clock) {
         this.applicationEventPublisher = applicationEventPublisher;
-        this.web3jService = web3jService;
         this.pollingInterval = chainConfig.getBlockTime();
         this.monitoringExecutor = monitoringExecutor;
         this.clock = clock;
@@ -107,12 +106,13 @@ void scheduleMonitoring() {
      * blockchain node is restored.
      */
     void checkConnection() {
-        final long latestBlockNumber = web3jService.getLatestBlockNumber();
         log.debug("Latest on-chain block number [block:{}]", latestBlockNumber);
-        if (latestBlockNumber == 0) {
+        final Instant now = Instant.now();
+        final Instant threshold = Instant.ofEpochSecond(latestBlockTimestamp).plusSeconds(pollingInterval.toSeconds());
+        if (now.isAfter(threshold)) {
             connectionFailed();
         } else {
-            connectionSucceeded(latestBlockNumber);
+            connectionSucceeded();
         }
     }
 
@@ -145,7 +145,7 @@ private void connectionFailed() {
      * <li>If {@link Status#OUT_OF_SERVICE}, publish a {@link ChainConnectedEvent} event
      * </ul>
      */
-    private void connectionSucceeded(long latestBlockNumber) {
+    private void connectionSucceeded() {
         if (isOutOfService()) {
             log.info("Blockchain connection is now restored after a period of unavailability." +
                             " [block:{}, unavailabilityPeriod:{}]",
@@ -190,4 +190,10 @@ public boolean isOutOfService() {
     public boolean isUp() {
         return health().getStatus() == Status.UP;
     }
+
+    @EventListener
+    private void onLatestBlockEvent(final LatestBlockEvent event) {
+        latestBlockNumber = event.getBlockNumber();
+        latestBlockTimestamp = event.getBlockTimestamp();
+    }
 }

src/main/java/com/iexec/core/chain/Web3jService.java

@@ -17,11 +17,15 @@
 package com.iexec.core.chain;
 
 import com.iexec.commons.poco.chain.Web3jAbstractService;
+import com.iexec.core.chain.event.LatestBlockEvent;
+import org.springframework.context.event.EventListener;
 import org.springframework.stereotype.Service;
 
 @Service
 public class Web3jService extends Web3jAbstractService {
 
+    private long latestBlockNumber;
+
     public Web3jService(ChainConfig chainConfig) {
         super(
                 chainConfig.getChainId(),
@@ -33,4 +37,14 @@ public Web3jService(ChainConfig chainConfig) {
         );
     }
 
+    @EventListener
+    private void onLatestBlockEvent(final LatestBlockEvent event) {
+        this.latestBlockNumber = event.getBlockNumber();
+    }
+
+    @Override
+    public long getLatestBlockNumber() {
+        return latestBlockNumber;
+    }
+
 }

src/main/java/com/iexec/core/chain/WebSocketBlockchainListener.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2025 IEXEC BLOCKCHAIN TECH
+ *
+ * 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
+ *
+ *     http://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.
+ */
+
+package com.iexec.core.chain;
+
+import com.iexec.core.chain.event.LatestBlockEvent;
+import io.micrometer.core.instrument.Metrics;
+import io.reactivex.Flowable;
+import io.reactivex.disposables.Disposable;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+import org.web3j.protocol.core.Request;
+import org.web3j.protocol.core.methods.response.EthSubscribe;
+import org.web3j.protocol.websocket.WebSocketService;
+import org.web3j.protocol.websocket.events.NewHeadsNotification;
+import org.web3j.utils.Numeric;
+
+import java.net.ConnectException;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+@Slf4j
+@Service
+public class WebSocketBlockchainListener {
+    static final String LATEST_BLOCK_METRIC_NAME = "iexec.chain.block.latest";
+
+    private static final String SUBSCRIBE_METHOD = "eth_subscribe";
+    private static final String UNSUBSCRIBE_METHOD = "eth_unsubscribe";
+
+    private final ApplicationEventPublisher applicationEventPublisher;
+    private final WebSocketService webSocketService;
+
+    private final AtomicLong lastSeenBlock;
+    private Disposable newHeads;
+
+    public WebSocketBlockchainListener(final ApplicationEventPublisher applicationEventPublisher,
+                                       final ChainConfig chainConfig) {
+        this.applicationEventPublisher = applicationEventPublisher;
+        final String wsUrl = chainConfig.getNodeAddress().replace("http", "ws");
+        this.webSocketService = new WebSocketService(wsUrl, false);
+        lastSeenBlock = Metrics.gauge(LATEST_BLOCK_METRIC_NAME, new AtomicLong(0));
+    }
+
+    @Scheduled(fixedRate = 5000)
+    private void run() throws ConnectException {
+        if (newHeads != null && !newHeads.isDisposed()) {
+            return;
+        }
+
+        log.warn("web socket disconnection detected");
+        webSocketService.connect();
+
+        final Request<?, EthSubscribe> newHeadsRequest = new Request<>(
+                SUBSCRIBE_METHOD,
+                List.of("newHeads", Map.of()),
+                webSocketService,
+                EthSubscribe.class
+        );
+
+        final Flowable<NewHeadsNotification> newHeadsEvents = webSocketService.subscribe(
+                newHeadsRequest, UNSUBSCRIBE_METHOD, NewHeadsNotification.class);
+        newHeads = newHeadsEvents.subscribe(this::processHead, this::handleError);
+    }
+
+    private void processHead(final NewHeadsNotification event) {
+        final long blockNumber = Numeric.toBigInt(event.getParams().getResult().getNumber()).longValue();
+        final String blockHash = event.getParams().getResult().getHash();
+        final long blockTimestamp = Numeric.toBigInt(event.getParams().getResult().getTimestamp()).longValue();
+        final Instant blockTimestampInstant = Instant.ofEpochSecond(blockTimestamp);
+        log.info("Last seen block [number:{}, hash:{}, timestamp:{}, instant:{}]",
+                blockNumber, blockHash, blockTimestamp, blockTimestampInstant);
+        lastSeenBlock.set(blockNumber);
+        applicationEventPublisher.publishEvent(new LatestBlockEvent(this, blockNumber, blockHash, blockTimestamp));
+    }
+
+    private void handleError(final Throwable t) {
+        log.error("An error happened during subscription", t);
+    }
+}

src/main/java/com/iexec/core/chain/event/LatestBlockEvent.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2025 IEXEC BLOCKCHAIN TECH
+ *
+ * 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
+ *
+ *     http://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.
+ */
+
+package com.iexec.core.chain.event;
+
+import lombok.EqualsAndHashCode;
+import lombok.Value;
+import org.springframework.context.ApplicationEvent;
+
+@Value
+@EqualsAndHashCode(callSuper = true)
+public class LatestBlockEvent extends ApplicationEvent {
+    long blockNumber;
+    String blockHash;
+    long blockTimestamp;
+
+    public LatestBlockEvent(final Object source, final long blockNumber, final String blockHash, final long blockTimestamp) {
+        super(source);
+        this.blockNumber = blockNumber;
+        this.blockHash = blockHash;
+        this.blockTimestamp = blockTimestamp;
+    }
+}

src/main/java/com/iexec/core/metric/MetricService.java

@@ -17,6 +17,7 @@
 package com.iexec.core.metric;
 
 import com.iexec.core.chain.DealWatcherService;
+import com.iexec.core.chain.event.LatestBlockEvent;
 import com.iexec.core.task.TaskStatus;
 import com.iexec.core.task.event.TaskStatusesCountUpdatedEvent;
 import com.iexec.core.worker.AliveWorkerMetrics;
@@ -31,6 +32,7 @@ public class MetricService {
     private final DealWatcherService dealWatcherService;
     private final WorkerService workerService;
     private LinkedHashMap<TaskStatus, Long> currentTaskStatusesCount;
+    private PlatformMetric.LatestBlockMetric latestBlock;
 
     public MetricService(final DealWatcherService dealWatcherService,
                          final WorkerService workerService) {
@@ -49,6 +51,7 @@ public PlatformMetric getPlatformMetrics() {
                 .aliveComputingGpu(aliveWorkerMetrics.aliveComputingGpu())
                 .aliveRegisteredGpu(aliveWorkerMetrics.aliveRegisteredGpu())
                 .currentTaskStatusesCount(currentTaskStatusesCount)
+                .latestBlockMetric(latestBlock)
                 .dealEventsCount(dealWatcherService.getDealEventsCount())
                 .dealsCount(dealWatcherService.getDealsCount())
                 .replayDealsCount(dealWatcherService.getReplayDealsCount())
@@ -60,6 +63,11 @@ public PlatformMetric getPlatformMetrics() {
                 .build();
     }
 
+    @EventListener
+    private void onLatestBlockEvent(final LatestBlockEvent event) {
+        latestBlock = new PlatformMetric.LatestBlockMetric(event.getBlockNumber(), event.getBlockHash(), event.getBlockTimestamp());
+    }
+
     @EventListener
     void onTaskStatusesCountUpdateEvent(final TaskStatusesCountUpdatedEvent event) {
         this.currentTaskStatusesCount = event.getCurrentTaskStatusesCount();