refactor(backfill): Use of PBJ gRPC Client Helidon instead of Vanilla Helidon. (#1499)

Signed-off-by: Alfredo Gutierrez Grajeda <alfredo@hashgraph.com>

Author: AlfredoG87

block-node/backfill/build.gradle.kts

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 plugins {
     id("org.hiero.gradle.module.library")
-    id("com.hedera.pbj.pbj-compiler") version "0.11.13"
+    id("com.hedera.pbj.pbj-compiler") version "0.11.14"
 }
 
 description = "Hiero Block Node Backfill Plugin"
@@ -25,6 +25,8 @@ mainModuleInfo {
     runtimeOnly("org.apache.logging.log4j.slf4j2.impl")
     runtimeOnly("io.helidon.logging.jul")
     runtimeOnly("com.hedera.pbj.grpc.helidon.config")
+    runtimeOnly("com.hedera.pbj.grpc.client.helidon")
+    runtimeOnly("com.hedera.pbj.grpc.helidon")
 }
 
 testModuleInfo {

block-node/backfill/src/main/java/module-info.java

@@ -20,13 +20,11 @@
     requires transitive com.swirlds.metrics.api;
     requires transitive org.hiero.block.node.spi;
     requires transitive org.hiero.block.protobuf.pbj;
-    requires transitive io.grpc.stub;
-    requires transitive io.grpc;
-    requires transitive io.helidon.grpc.core;
-    requires transitive io.helidon.webclient.grpc;
+    requires com.hedera.pbj.grpc.client.helidon;
     requires org.hiero.block.node.base;
     requires io.helidon.common.tls;
     requires io.helidon.webclient.api;
+    requires io.helidon.webclient.grpc;
     requires org.antlr.antlr4.runtime;
     requires static transitive com.github.spotbugs.annotations;
 

block-node/backfill/src/main/java/org/hiero/block/node/backfill/BackfillConfiguration.java

@@ -21,6 +21,8 @@
  * @param initialDelay Initial delay in seconds before starting the backfill process, to give time for the system to stabilize
  * @param perBlockProcessingTimeout Timeout in milliseconds for processing each block, to avoid blocking the backfill
  *                                  process indefinitely in case something unexpected happens, this would allow for self-recovery
+ * @param grpcOverallTimeout single timeout configuration for gRPC Client construction, connectTimeout, readTimeout and pollWaitTime
+ * @param enableTLS if enabled will assume block-node client supports tls connection.
  */
 @ConfigData("backfill")
 public record BackfillConfiguration(
@@ -33,4 +35,6 @@ public record BackfillConfiguration(
         @Loggable @ConfigProperty(defaultValue = "25") @Min(1) @Max(10_000) int fetchBatchSize,
         @Loggable @ConfigProperty(defaultValue = "1000") @Min(100) int delayBetweenBatches,
         @Loggable @ConfigProperty(defaultValue = "15000") @Min(5) int initialDelay,
-        @Loggable @ConfigProperty(defaultValue = "1000") @Min(500) int perBlockProcessingTimeout) {}
+        @Loggable @ConfigProperty(defaultValue = "1000") @Min(500) int perBlockProcessingTimeout,
+        @Loggable @ConfigProperty(defaultValue = "30000") @Min(10000) int grpcOverallTimeout,
+        @Loggable @ConfigProperty(defaultValue = "false") boolean enableTLS) {}

block-node/backfill/src/main/java/org/hiero/block/node/backfill/BackfillGrpcClient.java

@@ -15,6 +15,8 @@
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
+import org.hiero.block.api.ServerStatusRequest;
+import org.hiero.block.api.ServerStatusResponse;
 import org.hiero.block.internal.BlockUnparsed;
 import org.hiero.block.node.backfill.client.BackfillSource;
 import org.hiero.block.node.backfill.client.BackfillSourceConfig;
@@ -50,7 +52,10 @@ public class BackfillGrpcClient {
      * This is used for exponential backoff in case of failures.
      */
     private final int initialRetryDelayMs;
-
+    /** Connection timeout in milliseconds for gRPC calls to block nodes. */
+    private final int connectionTimeoutSeconds;
+    /** Enable TLS for secure connections to block nodes. */
+    private final boolean enableTls;
     /** Current status of the Block Node Clients */
     private ConcurrentHashMap<BackfillSourceConfig, Status> nodeStatusMap = new ConcurrentHashMap<>();
     /**
@@ -65,12 +70,19 @@ public class BackfillGrpcClient {
      * @param blockNodePreferenceFilePath the path to the block node preference file
      */
     public BackfillGrpcClient(
-            Path blockNodePreferenceFilePath, int maxRetries, Counter backfillRetriesCounter, int retryInitialDelayMs)
+            Path blockNodePreferenceFilePath,
+            int maxRetries,
+            Counter backfillRetriesCounter,
+            int retryInitialDelayMs,
+            int connectionTimeoutSeconds,
+            boolean enableTls)
             throws IOException, ParseException {
         this.blockNodeSource = BackfillSource.JSON.parse(Bytes.wrap(Files.readAllBytes(blockNodePreferenceFilePath)));
         this.maxRetries = maxRetries;
         this.initialRetryDelayMs = retryInitialDelayMs;
         this.backfillRetries = backfillRetriesCounter;
+        this.connectionTimeoutSeconds = connectionTimeoutSeconds;
+        this.enableTls = enableTls;
 
         for (BackfillSourceConfig node : blockNodeSource.nodes()) {
             LOGGER.log(INFO, "Address: {0}, Port: {1}, Priority: {2}", node.address(), node.port(), node.priority());
@@ -84,10 +96,11 @@ public BackfillGrpcClient(
      * @return a LongRange representing the intersection of the block range and the available blocks in the node.
      */
     private LongRange getAvailableRangeInNode(BlockNodeClient node, LongRange blockRange) {
-        long firstAvailableBlock =
-                node.getBlockNodeServerStatusClient().getServerStatus().firstAvailableBlock();
-        long lastAvailableBlock =
-                node.getBlockNodeServerStatusClient().getServerStatus().lastAvailableBlock();
+
+        final ServerStatusResponse nodeStatus =
+                node.getBlockNodeServiceClient().serverStatus(new ServerStatusRequest());
+        long firstAvailableBlock = nodeStatus.firstAvailableBlock();
+        long lastAvailableBlock = nodeStatus.lastAvailableBlock();
 
         long start = blockRange.start();
         long end = blockRange.end();
@@ -148,7 +161,7 @@ public List<BlockUnparsed> fetchBlocks(LongRange blockRange) {
 
                     // Try to fetch blocks from this node
                     return currentNodeClient
-                            .getBlockNodeSubscribeClient()
+                            .getBlockstreamSubscribeUnparsedClient()
                             .getBatchOfBlocks(actualRange.start(), actualRange.end());
                 } catch (Exception e) {
                     if (attempt == maxRetries) {
@@ -195,7 +208,8 @@ public List<BlockUnparsed> fetchBlocks(LongRange blockRange) {
      * @return a BlockNodeClient for the specified node
      */
     private BlockNodeClient getNodeClient(BackfillSourceConfig node) {
-        return nodeClientMap.computeIfAbsent(node, BlockNodeClient::new);
+        return nodeClientMap.computeIfAbsent(
+                node, BlockNodeClient -> new BlockNodeClient(node, connectionTimeoutSeconds, enableTls));
     }
 
     /**

block-node/backfill/src/main/java/org/hiero/block/node/backfill/BackfillPlugin.java

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 package org.hiero.block.node.backfill;
 
+import static java.lang.System.Logger.Level.DEBUG;
 import static java.lang.System.Logger.Level.INFO;
 import static java.lang.System.Logger.Level.TRACE;
 
@@ -180,7 +181,9 @@ public void init(BlockNodeContext context, ServiceBuilder serviceBuilder) {
                     blockNodeSourcesPath,
                     backfillConfiguration.maxRetries(),
                     this.backfillRetries,
-                    backfillConfiguration.initialRetryDelay());
+                    backfillConfiguration.initialRetryDelay(),
+                    backfillConfiguration.grpcOverallTimeout(),
+                    backfillConfiguration.enableTLS());
             LOGGER.log(TRACE, "Initialized gRPC client with sources path: {0}", blockNodeSourcesPath);
         } catch (Exception e) {
             LOGGER.log(INFO, "Failed to initialize gRPC client: {0}", e.getMessage());
@@ -206,7 +209,8 @@ public void start() {
                 TRACE,
                 "Scheduling backfill process to start in {0} milliseconds",
                 backfillConfiguration.initialDelay());
-        scheduler = Executors.newSingleThreadScheduledExecutor();
+        scheduler = Executors.newScheduledThreadPool(
+                2); // Two threads: one for autonomous backfill, one for on-demand backfill
         scheduler.scheduleAtFixedRate(
                 this::detectGaps,
                 backfillConfiguration.initialDelay(),
@@ -331,6 +335,11 @@ private void backfillGap(LongRange gap, BackfillType backfillType) throws Interr
             // to avoid deadlocks, since blocks that fail verification are not persisted
             getLatch(backfillType).set(new CountDownLatch(batchOfBlocks.size()));
 
+            if (batchOfBlocks.isEmpty()) {
+                LOGGER.log(DEBUG, "No blocks fetched for gap {0}, skipping", chunk);
+                continue; // Skip empty batches
+            }
+
             // Process each fetched block
             for (BlockUnparsed blockUnparsed : batchOfBlocks) {
                 long blockNumber = extractBlockNumber(blockUnparsed);
@@ -347,21 +356,25 @@ private void backfillGap(LongRange gap, BackfillType backfillType) throws Interr
             boolean backfillFinished = getLatch(backfillType).get().await(timeout, TimeUnit.MILLISECONDS);
 
             // Check if the backfill finished successfully
-            if (!backfillFinished) {
+            if (backfillFinished) {
+                // just log a victory message for each chunk
+                LOGGER.log(TRACE, "Successfully backfilled gap {0}", chunk);
+            } else {
                 LOGGER.log(TRACE, "Backfill for gap {0} did not finish in time", chunk);
                 backfillFetchErrors.increment();
                 // If it didn't finish, we will retry it later but move on to next chunk
-            } else {
-                // just log a victory message for each chunk
-                LOGGER.log(TRACE, "Successfully backfilled gap {0}", chunk);
             }
 
             // Cooldown between batches
             Thread.sleep(backfillConfiguration.delayBetweenBatches());
         }
 
         LOGGER.log(
-                TRACE, "Completed backfilling of type {0} gap from {1} to {2}", backfillType, gap.start(), gap.end());
+                TRACE,
+                "Completed backfilling task (completion only) of type {0} gap from {1} to {2} ",
+                backfillType,
+                gap.start(),
+                gap.end());
         if (backfillType.equals(BackfillType.ON_DEMAND)) {
             onDemandBackfillStartBlock.set(-1); // Reset on-demand start block after backfill
         }

block-node/backfill/src/main/java/org/hiero/block/node/backfill/client/BlockNodeClient.java

@@ -1,37 +1,74 @@
 // SPDX-License-Identifier: Apache-2.0
 package org.hiero.block.node.backfill.client;
 
+import com.hedera.pbj.grpc.client.helidon.PbjGrpcClient;
+import com.hedera.pbj.grpc.client.helidon.PbjGrpcClientConfig;
+import com.hedera.pbj.runtime.grpc.ServiceInterface;
 import io.helidon.common.tls.Tls;
-import io.helidon.webclient.grpc.GrpcClient;
+import io.helidon.webclient.api.WebClient;
 import io.helidon.webclient.grpc.GrpcClientProtocolConfig;
 import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
+import org.hiero.block.api.BlockNodeServiceInterface;
 
 public class BlockNodeClient {
-    private final BlockNodeServerStatusClient blockNodeServerStatusClient;
-    private final BlockNodeSubscribeClient blockNodeSubscribeClient;
+    // Options definition for all gRPC services in the block node client
+    private static record Options(Optional<String> authority, String contentType)
+            implements ServiceInterface.RequestOptions {}
 
-    public BlockNodeClient(BackfillSourceConfig blockNodeConfig) {
+    private static final BlockNodeClient.Options OPTIONS =
+            new BlockNodeClient.Options(Optional.empty(), ServiceInterface.RequestOptions.APPLICATION_GRPC);
 
-        // Initialize gRPC client with the block node configuration
-        GrpcClient grpcClient = GrpcClient.builder()
-                .tls(Tls.builder().enabled(false).build())
+    // block node services
+    private final BlockStreamSubscribeUnparsedClient blockStreamSubscribeUnparsedClient;
+    private final BlockNodeServiceInterface.BlockNodeServiceClient blockNodeServiceClient;
+
+    /**
+     * Constructs a BlockNodeClient using the provided configuration.
+     *
+     * @param blockNodeConfig the configuration for the block node, including address and port
+     */
+    public BlockNodeClient(BackfillSourceConfig blockNodeConfig, int timeoutMs, boolean enableTls) {
+
+        final Duration timeoutDuration = Duration.ofMillis(timeoutMs);
+
+        final Tls tls = Tls.builder().enabled(enableTls).build();
+        final PbjGrpcClientConfig grpcConfig =
+                new PbjGrpcClientConfig(timeoutDuration, tls, Optional.of(""), "application/grpc");
+
+        final WebClient webClient = WebClient.builder()
                 .baseUri("http://" + blockNodeConfig.address() + ":" + blockNodeConfig.port())
-                .protocolConfig(GrpcClientProtocolConfig.builder()
+                .tls(tls)
+                .protocolConfigs(List.of(GrpcClientProtocolConfig.builder()
                         .abortPollTimeExpired(false)
-                        .pollWaitTime(Duration.ofSeconds(30))
-                        .build())
-                .keepAlive(true)
+                        .pollWaitTime(timeoutDuration)
+                        .build()))
+                .connectTimeout(timeoutDuration)
                 .build();
-        // Initialize clients for server status and block subscription
-        this.blockNodeServerStatusClient = new BlockNodeServerStatusClient(grpcClient);
-        this.blockNodeSubscribeClient = new BlockNodeSubscribeClient(grpcClient);
+
+        PbjGrpcClient pbjGrpcClient = new PbjGrpcClient(webClient, grpcConfig);
+
+        // we reuse the host connection with many services.
+        blockNodeServiceClient = new BlockNodeServiceInterface.BlockNodeServiceClient(pbjGrpcClient, OPTIONS);
+        this.blockStreamSubscribeUnparsedClient = new BlockStreamSubscribeUnparsedClient(pbjGrpcClient);
     }
 
-    public BlockNodeServerStatusClient getBlockNodeServerStatusClient() {
-        return blockNodeServerStatusClient;
+    /**
+     * Returns the BlockStreamSubscribeUnparsedClient for subscribing to block streams.
+     *
+     * @return the BlockStreamSubscribeUnparsedClient
+     */
+    public BlockStreamSubscribeUnparsedClient getBlockstreamSubscribeUnparsedClient() {
+        return blockStreamSubscribeUnparsedClient;
     }
 
-    public BlockNodeSubscribeClient getBlockNodeSubscribeClient() {
-        return blockNodeSubscribeClient;
+    /**
+     * Returns the BlockNodeServiceClient for accessing block node services.
+     *
+     * @return the BlockNodeServiceClient
+     */
+    public BlockNodeServiceInterface.BlockNodeServiceClient getBlockNodeServiceClient() {
+        return blockNodeServiceClient;
     }
 }