feat: add connection/client for BN service API

Signed-off-by: Tim Farber-Newman <[email protected]>

Author: timfn-hg

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

@@ -26,11 +26,11 @@ enum ConnectionType {
         /**
          * Denotes a connection that intends to stream block data to a block node.
          */
-        BLOCK_STREAMING("B"), // 'B' for block
+        BLOCK_STREAMING("STR"), // block STReaming
         /**
          * Denotes a connection that intends to query server information from a block node.
          */
-        SERVER_STATUS("S"); // 'S' for status
+        SERVER_STATUS("SVC"); // block node SerViCe
 
         private final String key;
 
@@ -83,7 +83,7 @@ enum ConnectionType {
         requireNonNull(type, "type is required");
 
         connectionId =
-                String.format("%s:%05d", type.key, connIdCtrByType.get(type).incrementAndGet());
+                String.format("%s.%06d", type.key, connIdCtrByType.get(type).incrementAndGet());
         stateRef = new AtomicReference<>(ConnectionState.UNINITIALIZED);
     }
 
@@ -118,7 +118,7 @@ final boolean updateConnectionState(
 
         if (!latestState.canTransitionTo(newState)) {
             logger.warn(
-                    "{} Attempted to downgrade state from {} to {}, but this is not allowed; ignoring update",
+                    "{} Attempted to downgrade state from {} to {}, but this is not allowed; this is not allowed",
                     this,
                     latestState,
                     newState);

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

@@ -85,7 +85,8 @@ public BlockStreamPublishServiceClient createStreamingClient(
      * @param timeout the timeout to use
      * @return a new {@link BlockNodeServiceClient} instance
      */
-    public BlockNodeServiceClient createServiceClient(@NonNull final BlockNodeConfiguration config, @NonNull final Duration timeout) {
+    public BlockNodeServiceClient createServiceClient(
+            @NonNull final BlockNodeConfiguration config, @NonNull final Duration timeout) {
         final PbjGrpcClient client = buildPbjClient(config, timeout);
         return new BlockNodeServiceClient(client, new DefaultRequestOptions());
     }

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

@@ -1,3 +1,4 @@
+// SPDX-License-Identifier: Apache-2.0
 package com.hedera.node.app.blocks.impl.streaming;
 
 import static java.util.Objects.requireNonNull;
@@ -6,47 +7,55 @@
 import com.hedera.node.config.ConfigProvider;
 import com.hedera.node.config.data.BlockNodeConnectionConfig;
 import edu.umd.cs.findbugs.annotations.NonNull;
+import edu.umd.cs.findbugs.annotations.Nullable;
 import java.time.Duration;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.hiero.block.api.BlockNodeServiceInterface.BlockNodeServiceClient;
 import org.hiero.block.api.ServerStatusRequest;
 import org.hiero.block.api.ServerStatusResponse;
 
+/**
+ * Connection that can be used to query the Block Node Service API (e.g. to retrieve server status).
+ */
 public class BlockNodeServiceConnection extends AbstractBlockNodeConnection {
 
     private static final Logger logger = LogManager.getLogger(BlockNodeServiceConnection.class);
 
-    private final AtomicReference<BlockNodeServiceClient> clientRef = new AtomicReference<>();
+    private static final AtomicLong clientCtr = new AtomicLong(0);
+
+    record ServiceClientHolder(long clientId, BlockNodeServiceClient client) {}
+
+    private final AtomicReference<ServiceClientHolder> clientRef = new AtomicReference<>();
     private final ExecutorService executorService;
     private final BlockNodeClientFactory clientFactory;
     private final BlockNodeConnectionConfig bncConfig;
 
-    public BlockNodeServiceConnection(@NonNull final ConfigProvider configProvider,
+    /**
+     * Create a new instance.
+     *
+     * @param configProvider the configuration provider to use
+     * @param nodeConfig the block node configuration to use for this connection
+     * @param executorService the executor service to use for executing tasks
+     * @param clientFactory the factory to use for creating clients to the block node
+     */
+    public BlockNodeServiceConnection(
+            @NonNull final ConfigProvider configProvider,
             @NonNull final BlockNodeConfiguration nodeConfig,
             @NonNull final ExecutorService executorService,
             @NonNull final BlockNodeClientFactory clientFactory) {
         super(ConnectionType.SERVER_STATUS, nodeConfig, configProvider);
         this.executorService = requireNonNull(executorService, "pipeline executor is required");
         this.clientFactory = requireNonNull(clientFactory, "client factory is required");
 
-        bncConfig = configProvider()
-                .getConfiguration()
-                .getConfigData(BlockNodeConnectionConfig.class);
-    }
-
-    private Duration timeout() {
-        return configProvider()
-                .getConfiguration()
-                .getConfigData(BlockNodeConnectionConfig.class)
-                .grpcOverallTimeout();
+        bncConfig = configProvider().getConfiguration().getConfigData(BlockNodeConnectionConfig.class);
     }
 
     @Override
@@ -56,69 +65,124 @@ void initialize() {
             return;
         }
 
-        final Future<?> future = executorService.submit(new CreateClientTask());
+        Future<?> future = null;
 
         try {
+            future = executorService.submit(new CreateClientTask());
             future.get(bncConfig.pipelineOperationTimeout().toMillis(), TimeUnit.MILLISECONDS);
-        } catch (final TimeoutException e) {
-            logger.warn("{} Client initialization timed out (timeout={})", this, bncConfig.pipelineOperationTimeout());
-            future.cancel(true);
-            throw new RuntimeException("Error initializing client", e);
-        } catch (final InterruptedException e) {
-            logger.warn("{} Client initialization interrupted", this);
-            Thread.currentThread().interrupt();
-            throw new RuntimeException("Error initializing client", e);
-        } catch (final ExecutionException e) {
-            logger.warn("{} Error initializing client", this, e.getCause());
-            throw new RuntimeException("Error initializing client", e);
+        } catch (final Exception e) {
+            logger.warn("{} Error initializing connection", this, e);
+
+            if (future != null) {
+                future.cancel(true);
+            }
+
+            if (e instanceof InterruptedException) {
+                Thread.currentThread().interrupt();
+            }
+
+            final Throwable error;
+            if (e instanceof final ExecutionException ee) {
+                error = ee.getCause();
+            } else {
+                error = e;
+            }
+
+            throw new RuntimeException("Error initializing connection", error);
         }
     }
 
+    /**
+     * Task used to create a Block Node service client for this connection.
+     */
     class CreateClientTask implements Runnable {
         @Override
         public void run() {
-            final BlockNodeServiceClient client = clientFactory.createServiceClient(configuration(), timeout());
-            if (clientRef.compareAndSet(null, client)) {
-                updateConnectionState(ConnectionState.UNINITIALIZED, ConnectionState.ACTIVE); // jump to active
-                logger.debug("{} Client initialized successfully", BlockNodeServiceConnection.this);
+            final Duration timeout = configProvider()
+                    .getConfiguration()
+                    .getConfigData(BlockNodeConnectionConfig.class)
+                    .grpcOverallTimeout();
+
+            final long clientId = clientCtr.incrementAndGet();
+            logger.debug("{} Creating new client (clientId: {})", BlockNodeServiceConnection.this, clientId);
+            final BlockNodeServiceClient client = clientFactory.createServiceClient(configuration(), timeout);
+            if (clientRef.compareAndSet(null, new BlockNodeServiceConnection.ServiceClientHolder(clientId, client))) {
+                // unlike the streaming connection, these connections don't really have an intermediate state between
+                // UNINITIALIZED and ACTIVE, so just set the state to ACTIVE
+                updateConnectionState(ConnectionState.UNINITIALIZED, ConnectionState.ACTIVE);
+                logger.info(
+                        "{} Client initialized successfully (clientId: {})", BlockNodeServiceConnection.this, clientId);
             } else {
-                close();
+                logger.debug(
+                        "{} Another thread has created the client and applied it to this connection; ignoring this attempt",
+                        BlockNodeServiceConnection.this);
+                closeSilently(new ServiceClientHolder(clientId, client));
             }
         }
-    }
 
-    class CloseTask implements Runnable {
-        private final BlockNodeServiceClient client;
+        /**
+         * Silently close specified client. This is used for cases where another thread has won initializing the client
+         * to use for this connection, and thus we want to close the client created by the losing thread. Any errors
+         * that result from closing the client will be suppressed.
+         *
+         * @param holder the client to close
+         */
+        private void closeSilently(@NonNull final ServiceClientHolder holder) {
+            logger.debug("{} Silently closing client (clientId: {})", BlockNodeServiceConnection.this, holder.clientId);
+            final Future<?> future = executorService.submit(new CloseClientTask(holder));
+            try {
+                future.get(bncConfig.pipelineOperationTimeout().toMillis(), TimeUnit.MILLISECONDS);
+            } catch (final Exception e) {
+                logger.debug(
+                        "{} Attempted to close a client (clientId: {}), but it failed; ignoring failure",
+                        BlockNodeServiceConnection.this,
+                        holder.clientId,
+                        e);
+            }
+        }
+    }
 
-        CloseTask(@NonNull final BlockNodeServiceClient client) {
-            this.client = requireNonNull(client);
+    /**
+     * Task to close a specific client.
+     */
+    class CloseClientTask implements Runnable {
+        /**
+         * The client to close
+         */
+        private final ServiceClientHolder clientHolder;
+
+        CloseClientTask(@NonNull final ServiceClientHolder clientHolder) {
+            this.clientHolder = requireNonNull(clientHolder);
         }
 
         @Override
         public void run() {
-            client.close();
+            logger.debug("{} Closing client (clientId: {})", BlockNodeServiceConnection.this, clientHolder.clientId);
+            clientHolder.client.close();
         }
     }
 
     @Override
     void close() {
-        final BlockNodeServiceClient client = clientRef.get();
+        final ServiceClientHolder clientHolder = clientRef.get();
 
-        if (client == null || currentState().isTerminal()) {
+        if (clientHolder == null || currentState().isTerminal()) {
             // either close has already been called or close was called while the connection wasn't initialized
             return;
         }
 
-        logger.info("{} Closing connection", this);
-        updateConnectionState(ConnectionState.CLOSING);
-
-        if (!clientRef.compareAndSet(client, null)) {
+        if (!clientRef.compareAndSet(clientHolder, null)) {
             logger.debug("{} Another thread has closed the connection", this);
+            return;
         }
 
-        final Future<?> future = executorService.submit(new CloseTask(client));
+        logger.info("{} Closing connection", this);
+        updateConnectionState(ConnectionState.CLOSING);
+
+        Future<?> future = null;
 
         try {
+            future = executorService.submit(new CloseClientTask(clientHolder));
             future.get(bncConfig.pipelineOperationTimeout().toMillis(), TimeUnit.MILLISECONDS);
         } catch (final Exception e) {
             // the connection is being closed... don't propagate the exception
@@ -127,50 +191,71 @@ void close() {
             if (e instanceof InterruptedException) {
                 Thread.currentThread().interrupt();
             }
-            future.cancel(true);
+
+            if (future != null) {
+                future.cancel(true);
+            }
         } finally {
             // regardless of outcome, mark this connection as closed
             updateConnectionState(ConnectionState.CLOSED);
         }
     }
 
-    public BlockNodeStatus getBlockNodeStatus() {
-        final BlockNodeServiceClient client = clientRef.get();
-
-        if (client == null || currentState() != ConnectionState.ACTIVE) {
+    /**
+     * Retrieves the server status of the Block Node associated with this connection. If there are any errors
+     * experienced during this process, the node will be considered unreachable and as such an "unreachable" response
+     * will be returned.
+     *
+     * @return null if this connection is not active, else the Block Node's status
+     */
+    public @Nullable BlockNodeStatus getBlockNodeStatus() {
+        final ServiceClientHolder clientHolder = clientRef.get();
+
+        if (clientHolder == null || currentState() != ConnectionState.ACTIVE) {
             logger.debug("{} Tried to retrieve block node status, but this connection is not active", this);
             return null;
         }
 
         final long startMillis = System.currentTimeMillis();
-        final Future<ServerStatusResponse> future = executorService.submit(new GetBlockNodeStatusTask(client));
+        Future<ServerStatusResponse> future = null;
         final ServerStatusResponse response;
         final long durationMillis;
 
         try {
+            future = executorService.submit(new GetBlockNodeStatusTask(clientHolder.client));
             response = future.get(bncConfig.pipelineOperationTimeout().toMillis(), TimeUnit.MILLISECONDS);
             durationMillis = System.currentTimeMillis() - startMillis;
-        } catch (final TimeoutException e) {
-            logger.warn("{} Timed out trying to retrieve server status (timeout={})", this, bncConfig.pipelineOperationTimeout());
-            future.cancel(true);
-            return BlockNodeStatus.notReachable();
-        } catch (final InterruptedException e) {
-            logger.warn("{} Interrupted while retrieving server status", this);
-            Thread.currentThread().interrupt();
-            return BlockNodeStatus.notReachable();
-        } catch (final ExecutionException e) {
-            logger.warn("{} Error occurred while retrieving server status", this, e);
+        } catch (final Exception e) {
+            logger.warn("{} Error retrieving block node status", this, e);
+
+            if (future != null) {
+                future.cancel(true);
+            }
+
+            if (e instanceof InterruptedException) {
+                Thread.currentThread().interrupt();
+            }
+
             return BlockNodeStatus.notReachable();
         }
 
-        logger.debug("{} Received the following block node server status: lastAvailableBlock={} (latency={}ms)",
-                this, response.lastAvailableBlock(), durationMillis);
+        logger.debug(
+                "{} Received the following block node server status: lastAvailableBlock={} (latency: {}ms)",
+                this,
+                response.lastAvailableBlock(),
+                durationMillis);
 
         return BlockNodeStatus.reachable(durationMillis, response.lastAvailableBlock());
     }
 
+    /**
+     * Task to get the server status.
+     */
     class GetBlockNodeStatusTask implements Callable<ServerStatusResponse> {
 
+        /**
+         * The client to use
+         */
         private final BlockNodeServiceClient client;
 
         private GetBlockNodeStatusTask(@NonNull final BlockNodeServiceClient client) {

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

@@ -1,11 +1,33 @@
+// SPDX-License-Identifier: Apache-2.0
 package com.hedera.node.app.blocks.impl.streaming;
 
+/**
+ * Result of a status check for a specific Block Node.
+ *
+ * @param wasReachable true if the Block Node was reachable, else false
+ * @param latencyMillis the duration the status check took to complete, or -1 if unreachable
+ * @param latestBlockAvailable the latest block available on the Block Node, or -1 if unreachable
+ */
 public record BlockNodeStatus(boolean wasReachable, long latencyMillis, long latestBlockAvailable) {
 
+    private static final BlockNodeStatus NOT_REACHABLE = new BlockNodeStatus(false, -1L, -1L);
+
+    /**
+     * Convenience method for creating a status of unreachable.
+     *
+     * @return a status marked as unreachable
+     */
     public static BlockNodeStatus notReachable() {
-        return new BlockNodeStatus(false, -1L, -1L);
+        return NOT_REACHABLE;
     }
 
+    /**
+     * Convenience method for creating a status of reachable.
+     *
+     * @param latencyMillis the latency (in milliseconds) the status check took
+     * @param latestBlockAvailable the latest block available
+     * @return a status marked as reachable
+     */
     public static BlockNodeStatus reachable(final long latencyMillis, final long latestBlockAvailable) {
         return new BlockNodeStatus(true, latencyMillis, latestBlockAvailable);
     }