feat: Http2ClientProtocolConfig and GrpcClientProtocolConfig in internal proto

hapi/hapi/src/main/proto/network/block_node_connections.proto

@@ -18,6 +18,8 @@ option java_package = "com.hedera.node.internal.network.legacy";
 // <<<pbj.java_package = "com.hedera.node.internal.network">>> This comment is special code for setting PBJ Compiler java package
 option java_multiple_files = true;
 
+import "google/protobuf/wrappers.proto";
+
 /**
  * A single block node connection configuration.<br/>
  *
@@ -57,6 +59,100 @@ message BlockNodeConfig {
    * node SHALL connect to a node in the next-highest available priority group.
    */
     int32 priority = 3;
+
+    /**
+     * Optional HTTP/2 client protocol configuration for this block node connection.
+     * If present, these settings SHALL override defaults when creating the WebClient.
+     */
+    Http2ClientProtocolConfig http2ClientProtocolConfig = 4;
+
+    /**
+     * Optional Grpc client protocol configuration for this block node connection.
+     * If present, these settings SHALL override defaults when creating the WebClient.
+     */
+    GrpcClientProtocolConfig grpcClientProtocolConfig = 5;
+}
+
+
+/**
+ * Optional HTTP/2 client protocol configuration for a specific block node connection.
+ * If present, these values SHALL be used to configure the HTTP/2 protocol for the WebClient connecting
+ * to the corresponding block node.
+ */
+message Http2ClientProtocolConfig {
+  /**
+   * Timeout for blocking while waiting for window update when window is depleted.
+   * ISO-8601 duration string (e.g., "PT15S").
+   */
+  google.protobuf.StringValue flow_control_block_timeout = 1;
+
+  /**
+   * Configure INITIAL_WINDOW_SIZE setting for new HTTP/2 connections.
+   */
+  google.protobuf.Int32Value initial_window_size = 2;
+
+  /**
+   * Configure initial MAX_FRAME_SIZE setting for new HTTP/2 connections.
+   */
+  google.protobuf.Int32Value max_frame_size = 3;
+
+  /**
+   * Configure initial MAX_HEADER_LIST_SIZE setting for new HTTP/2 connections.
+   */
+  google.protobuf.Int64Value max_header_list_size = 4;
+
+  /**
+   * Name of this HTTP/2 protocol configuration. Default is "h2".
+   */
+  google.protobuf.StringValue name = 5;
+
+  /**
+   * Check healthiness of cached connections with HTTP/2.0 ping frame.
+   */
+  google.protobuf.BoolValue ping = 6;
+
+  /**
+   * Timeout for ping probe used for checking healthiness of cached connections.
+   * ISO-8601 duration string (e.g., "PT0.5S").
+   */
+  google.protobuf.StringValue ping_timeout = 7;
+
+  /**
+   * Prior knowledge of HTTP/2 capabilities of the server.
+   */
+  google.protobuf.BoolValue prior_knowledge = 8;
+}
+
+message GrpcClientProtocolConfig {
+  /**
+   * Whether to continue retrying after a poll wait timeout expired or not. If a read operation timeouts out and this
+   * flag is set to false, the event is logged and the client will retry. Otherwise, an exception is thrown.
+   */
+  google.protobuf.BoolValue abort_poll_time_expired = 1;
+
+  /**
+   * How often to send a heartbeat (HTTP/2 ping) to check if the connection is still alive. This is useful for
+   * long-running, streaming gRPC calls. It is turned off by default but can be enabled by setting the period to a value
+   * greater than 0. ISO-8601 duration string (e.g., "PT0S").
+   */
+  google.protobuf.StringValue heartbeat_period = 2;
+
+  /**
+   * Initial buffer size used to serialize gRPC request payloads. Buffers shall grow according to the payload size, but
+   * setting this initial buffer size to a larger value may improve performance for certain applications.
+   */
+  google.protobuf.Int32Value init_buffer_size = 3;
+
+  /**
+   * Name identifying this client protocol. Defaults to type.
+   */
+  google.protobuf.StringValue name = 4;
+
+  /**
+   * How long to wait for the next HTTP/2 data frame to arrive in underlying stream. Whether this is a fatal error or
+   * not is controlled by abort_poll_time_expired(). ISO-8601 duration string (e.g., "PT10S").
+   */
+  google.protobuf.StringValue poll_wait_time = 5;
 }
 
 /**

hedera-node/hedera-app/build.gradle.kts

@@ -17,6 +17,7 @@ mainModuleInfo {
     runtimeOnly("io.helidon.grpc.core")
     runtimeOnly("io.helidon.webclient")
     runtimeOnly("io.helidon.webclient.grpc")
+    runtimeOnly("io.helidon.webclient.http2")
     runtimeOnly("com.hedera.pbj.grpc.client.helidon")
     runtimeOnly("com.hedera.pbj.grpc.helidon")
 }

hedera-node/hedera-app/src/main/java/com/hedera/node/app/blocks/impl/streaming/BlockNodeConnection.java

@@ -25,8 +25,10 @@
 import io.helidon.common.tls.Tls;
 import io.helidon.webclient.api.WebClient;
 import io.helidon.webclient.grpc.GrpcClientProtocolConfig;
+import io.helidon.webclient.spi.ProtocolConfig;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
@@ -235,13 +237,27 @@ public synchronized void createRequestPipeline() {
         final PbjGrpcClientConfig grpcConfig =
                 new PbjGrpcClientConfig(timeoutDuration, tls, Optional.of(""), "application/grpc");
 
+        var extractedGrpcConfig =
+                blockNodeConnectionManager.getGrpcClientProtocolConfigs().get(blockNodeConfig);
+        final var grpcProtocolConfig = (extractedGrpcConfig != null)
+                ? extractedGrpcConfig
+                : GrpcClientProtocolConfig.builder()
+                        .abortPollTimeExpired(false)
+                        .pollWaitTime(timeoutDuration)
+                        .build();
+
+        final List<ProtocolConfig> protocolConfigs = new ArrayList<>();
+        protocolConfigs.add(grpcProtocolConfig);
+        var http2ClientProtocolConfig =
+                blockNodeConnectionManager.getHttp2ClientProtocolConfigs().get(blockNodeConfig);
+        if (http2ClientProtocolConfig != null) {
+            protocolConfigs.add(http2ClientProtocolConfig);
+        }
+
         final WebClient webClient = WebClient.builder()
                 .baseUri("http://" + blockNodeConfig.address() + ":" + blockNodeConfig.port())
                 .tls(tls)
-                .protocolConfigs(List.of(GrpcClientProtocolConfig.builder()
-                        .abortPollTimeExpired(false)
-                        .pollWaitTime(timeoutDuration)
-                        .build()))
+                .protocolConfigs(protocolConfigs)
                 .connectTimeout(timeoutDuration)
                 .build();
         logWithContext(

hedera-node/hedera-app/src/main/java/com/hedera/node/app/blocks/impl/streaming/BlockNodeConnectionManager.java

@@ -25,6 +25,8 @@
 import com.hedera.pbj.runtime.io.buffer.Bytes;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
+import io.helidon.webclient.grpc.GrpcClientProtocolConfig;
+import io.helidon.webclient.http2.Http2ClientProtocolConfig;
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.net.InetAddress;
@@ -39,6 +41,7 @@
 import java.nio.file.WatchService;
 import java.time.Duration;
 import java.time.Instant;
+import java.time.format.DateTimeParseException;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
@@ -110,6 +113,9 @@ public class BlockNodeConnectionManager {
      * startup from the configuration file(s) on disk.
      */
     private final List<BlockNodeConfig> availableBlockNodes = new ArrayList<>();
+
+    private Map<BlockNodeConfig, Http2ClientProtocolConfig> http2ClientProtocolConfigs = new ConcurrentHashMap<>();
+    private Map<BlockNodeConfig, GrpcClientProtocolConfig> grpcClientProtocolConfigs = new ConcurrentHashMap<>();
     /**
      * Flag that indicates if this connection manager is active or not. In this case, being active means it is actively
      * processing blocks and attempting to send them to a block node.
@@ -298,11 +304,9 @@ private List<BlockNodeConfig> extractBlockNodesConfigurations(@NonNull final Str
 
             final byte[] jsonConfig = Files.readAllBytes(configPath);
             final BlockNodeConnectionInfo protoConfig = BlockNodeConnectionInfo.JSON.parse(Bytes.wrap(jsonConfig));
-
-            // Convert proto config to internal config objects
-            return protoConfig.nodes().stream()
-                    .map(node -> new BlockNodeConfig(node.address(), node.port(), node.priority()))
-                    .toList();
+            extractOptionalHttp2ClientProtocolConfig(protoConfig);
+            extractOptionalGrpcClientProtocolConfig(protoConfig);
+            return protoConfig.nodes();
         } catch (final IOException | ParseException e) {
             logWithContext(
                     logger,
@@ -314,6 +318,97 @@ private List<BlockNodeConfig> extractBlockNodesConfigurations(@NonNull final Str
         }
     }
 
+    private void extractOptionalHttp2ClientProtocolConfig(BlockNodeConnectionInfo protoConfig) {
+        for (BlockNodeConfig config : protoConfig.nodes()) {
+            if (config.hasHttp2ClientProtocolConfig()) {
+                com.hedera.node.internal.network.Http2ClientProtocolConfig protocolConfig =
+                        config.http2ClientProtocolConfig();
+                Http2ClientProtocolConfig.Builder builder = Http2ClientProtocolConfig.builder();
+                if (protocolConfig != null) {
+                    if (protocolConfig.flowControlBlockTimeout() != null) {
+                        try {
+                            Duration flowControlBlockTimeout = Duration.parse(protocolConfig.flowControlBlockTimeout());
+                            builder.flowControlBlockTimeout(flowControlBlockTimeout);
+                        } catch (DateTimeParseException e) {
+                            logger.info(
+                                    "Unable to parse Http2ClientProtocolConfig flowControlBlockTimeout: {}",
+                                    protocolConfig.flowControlBlockTimeout());
+                        }
+                    }
+                    if (protocolConfig.initialWindowSize() != null) {
+                        builder.initialWindowSize(protocolConfig.initialWindowSize());
+                    }
+                    if (protocolConfig.maxFrameSize() != null) {
+                        builder.maxFrameSize(protocolConfig.maxFrameSize());
+                    }
+                    if (protocolConfig.maxHeaderListSize() != null) {
+                        builder.maxHeaderListSize(protocolConfig.maxHeaderListSize());
+                    }
+                    if (protocolConfig.name() != null) {
+                        builder.name(protocolConfig.name());
+                    }
+                    if (protocolConfig.ping() != null) {
+                        builder.ping(protocolConfig.ping());
+                    }
+                    if (protocolConfig.pingTimeout() != null) {
+                        try {
+                            Duration pingTimeout = Duration.parse(protocolConfig.pingTimeout());
+                            builder.pingTimeout(pingTimeout);
+                        } catch (DateTimeParseException e) {
+                            logger.info(
+                                    "Unable to parse Http2ClientProtocolConfig pingTimeout: {}",
+                                    protocolConfig.pingTimeout());
+                        }
+                    }
+                    if (protocolConfig.priorKnowledge() != null) {
+                        builder.priorKnowledge(protocolConfig.priorKnowledge());
+                    }
+                }
+                http2ClientProtocolConfigs.put(config, builder.build());
+            }
+        }
+    }
+
+    private void extractOptionalGrpcClientProtocolConfig(BlockNodeConnectionInfo protoConfig) {
+        for (BlockNodeConfig config : protoConfig.nodes()) {
+            if (config.hasGrpcClientProtocolConfig()) {
+                com.hedera.node.internal.network.GrpcClientProtocolConfig protocolConfig =
+                        config.grpcClientProtocolConfig();
+                GrpcClientProtocolConfig.Builder builder = GrpcClientProtocolConfig.builder();
+                if (protocolConfig != null) {
+                    if (protocolConfig.abortPollTimeExpired() != null) {
+                        builder.abortPollTimeExpired(protocolConfig.abortPollTimeExpired());
+                    }
+                    if (protocolConfig.heartbeatPeriod() != null) {
+                        try {
+                            builder.heartbeatPeriod(Duration.parse(protocolConfig.heartbeatPeriod()));
+                        } catch (DateTimeParseException e) {
+                            logger.info(
+                                    "Unable to parse GrpcClientProtocolConfig heartbeatPeriod: {}",
+                                    protocolConfig.heartbeatPeriod());
+                        }
+                    }
+                    if (protocolConfig.initBufferSize() != null) {
+                        builder.initBufferSize(protocolConfig.initBufferSize());
+                    }
+                    if (protocolConfig.name() != null) {
+                        builder.name(protocolConfig.name());
+                    }
+                    if (protocolConfig.pollWaitTime() != null) {
+                        try {
+                            builder.pollWaitTime(Duration.parse(protocolConfig.pollWaitTime()));
+                        } catch (DateTimeParseException e) {
+                            logger.info(
+                                    "Unable to parse GrpcClientProtocolConfig pollWaitTime: {}",
+                                    protocolConfig.pollWaitTime());
+                        }
+                    }
+                }
+                grpcClientProtocolConfigs.put(config, builder.build());
+            }
+        }
+    }
+
     /**
      * Checks if there is only one block node configured.
      * @return whether there is only one block node configured
@@ -516,6 +611,8 @@ private void clearManagerMetadata() {
         activeConnectionRef.set(null);
         nodeStats.clear();
         availableBlockNodes.clear();
+        http2ClientProtocolConfigs.clear();
+        grpcClientProtocolConfigs.clear();
     }
 
     private void closeAllConnections() {
@@ -1433,4 +1530,20 @@ public BlockNodeStats.HighLatencyResult recordBlockAckAndCheckLatency(
 
         return result;
     }
+
+    /**
+     * Gets the HTTP/2 client protocol configurations for all block nodes.
+     * @return a map of BlockNodeConfig to Http2ClientProtocolConfig
+     */
+    public Map<BlockNodeConfig, Http2ClientProtocolConfig> getHttp2ClientProtocolConfigs() {
+        return http2ClientProtocolConfigs;
+    }
+
+    /**
+     * Gets the gRPC client protocol configurations for all block nodes.
+     * @return a map of BlockNodeConfig to GrpcClientProtocolConfig
+     */
+    public Map<BlockNodeConfig, GrpcClientProtocolConfig> getGrpcClientProtocolConfigs() {
+        return grpcClientProtocolConfigs;
+    }
 }

hedera-node/hedera-app/src/main/java/module-info.java

@@ -37,6 +37,8 @@
     requires transitive io.grpc;
     requires transitive io.helidon.grpc.core;
     requires transitive io.helidon.webclient.api;
+    requires transitive io.helidon.webclient.grpc;
+    requires transitive io.helidon.webclient.http2;
     requires transitive javax.inject;
     requires transitive org.apache.logging.log4j;
     requires transitive org.hyperledger.besu.datatypes;
@@ -57,7 +59,6 @@
     requires com.google.common;
     requires io.grpc.netty;
     requires io.helidon.common.tls;
-    requires io.helidon.webclient.grpc;
     requires io.netty.handler;
     requires io.netty.transport.classes.epoll;
     requires io.netty.transport;