feat: Initial consensus node to block node communication

hapi/internal-protobufs/network/block_node_connections.proto

@@ -0,0 +1,65 @@
+/**
+ * # Block Node Connections
+ * A representation of the configuration for connecting to a set of block nodes.
+ *
+ * ### Keywords
+ * The key words "MUST", "MUST NOT", "REQUIRED", "SHALL", "SHALL NOT",
+ * "SHOULD", "SHOULD NOT", "RECOMMENDED", "MAY", and "OPTIONAL" in this
+ * document are to be interpreted as described in
+ * [RFC2119](https://www.ietf.org/rfc/rfc2119) and clarified in
+ * [RFC8174](https://www.ietf.org/rfc/rfc8174).
+ */
+
+syntax = "proto3";
+
+package com.hedera.hapi.network;
+
+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;
+
+/**
+ * A single block node connection configuration.<br/>
+ *
+ * This message provides the necessary details for a consensus node to
+ * establish a connection and stream blocks to a block node.</br>
+ * Each instance of this message represents one block node connection.
+ * Multiple instances may be included in {@link BlockNodeConnectionInfo}.
+ * - Each instance SHALL contain a valid address and port.<br/>
+ * - A consensus node SHALL connect to all listed block nodes.<br/>
+ * - The address MUST be a valid DNS hostname or IP address.<br/>
+ * - The port SHALL be within the valid TCP/UDP port range.<br/>
+ */
+message BlockNodeConfig {
+    /**
+     * The network address of the block node.<br/>
+     * This MAY be an IP address or a DNS hostname.
+     */
+    string address = 1;
+
+    /**
+     * The port used to communicate with the block node.<br/>
+     * Valid port range is 1 to 65535.
+     */
+    int32 port = 2;
+}
+
+/**
+ * Overall connection configuration for block nodes.<br/>
+ *
+ * The BlockNodeConnectionInfo message SHOULD encapsulate all the information needed
+ * for a consensus node to connect to a set of block nodes and stream blocks.
+ */
+message BlockNodeConnectionInfo {
+    /**
+     * A list of block node configurations.<br/>
+     * This field SHALL list all block node configurations
+     * to which the consensus node SHOULD connect.
+     */
+    repeated BlockNodeConfig nodes = 1;
+
+    /**
+     * The size of the batches of block items to be streamed to block nodes.<br/>
+     */
+    int32 block_item_batch_size = 2;
+} 

hedera-node/configuration/dev/application.properties

@@ -16,6 +16,7 @@ ledger.id=0x03
 blockStream.streamMode=BOTH
 nodes.enableDAB=true
 blockStream.blockFileDir=data/blockStreams
+blockStream.blockNodeConnectionFileDir=data/config
 grpc.nodeOperatorPortEnabled=true
 # For CI tests we want to override roster weights from override-network.json
 networkAdmin.preserveStateWeightsDuringOverride=false

hedera-node/configuration/dev/block-nodes.json

@@ -0,0 +1,9 @@
+{
+  "nodes": [
+    {
+      "address": "localhost",
+      "port": 8080
+    }
+  ],
+  "blockItemBatchSize": 256
+}

hedera-node/configuration/dev/node.properties

@@ -3,3 +3,4 @@ hedera.recordStream.logDir=data/recordStreams
 hedera.recordStream.sidecarDir=
 stats.executionTimesToTrack=1000
 blockStream.blockFileDir=data/blockStreams
+blockStream.blockNodeConnectionFileDir=data/config

hedera-node/hedera-app/src/main/java/com/hedera/node/app/Hedera.java

@@ -593,12 +593,30 @@ public StateLifecycles<PlatformMerkleStateRoot> newStateLifecycles() {
     public void notify(@NonNull final PlatformStatusChangeNotification notification) {
         this.platformStatus = notification.getNewStatus();
         logger.info("HederaNode#{} is {}", platform.getSelfId(), platformStatus.name());
+        final var streamToBlockNodes = configProvider
+                .getConfiguration()
+                .getConfigData(BlockStreamConfig.class)
+                .streamToBlockNodes();
         switch (platformStatus) {
-            case ACTIVE -> startGrpcServer();
-            case CATASTROPHIC_FAILURE -> shutdownGrpcServer();
+            case ACTIVE -> {
+                startGrpcServer();
+            }
             case FREEZE_COMPLETE -> {
+                logger.info("Platform status is now FREEZE_COMPLETE");
+                shutdownGrpcServer();
                 closeRecordStreams();
+                if (streamToBlockNodes && isNotEmbedded()) {
+                    logger.info("FREEZE_COMPLETE - Shutting down connections to Block Nodes");
+                    daggerApp.blockNodeConnectionManager().shutdown();
+                }
+            }
+            case CATASTROPHIC_FAILURE -> {
+                logger.error("Platform status is now CATASTROPHIC_FAILURE");
                 shutdownGrpcServer();
+                if (streamToBlockNodes && isNotEmbedded()) {
+                    logger.info("CATASTROPHIC_FAILURE - Shutting down connections to Block Nodes");
+                    daggerApp.blockNodeConnectionManager().shutdown();
+                }
             }
             case REPLAYING_EVENTS, STARTING_UP, OBSERVING, RECONNECT_COMPLETE, CHECKING, FREEZING, BEHIND -> {
                 // Nothing to do here, just enumerate for completeness
@@ -890,6 +908,9 @@ public void shutdown() {
         shutdownGrpcServer();
 
         if (daggerApp != null) {
+            logger.debug("Shutting down the Block Node Connection Manager");
+            daggerApp.blockNodeConnectionManager().shutdown();
+
             logger.debug("Shutting down the state");
             final var state = daggerApp.workingStateAccessor().getState();
             if (state instanceof MerkleStateRoot msr) {
@@ -1336,4 +1357,26 @@ private void onAdoptRoster() {
                     new ReadableHintsStoreImpl(initState.getReadableStates(HintsService.NAME)));
         }
     }
+
+    /**
+     * Initializes block node connections and waits for at least one connection to be established.
+     * This should be called before platform.start() to ensure we don't miss any blocks.
+     *
+     * @param timeout maximum time to wait for a connection
+     */
+    public void initializeBlockNodeConnections(java.time.Duration timeout) {
+        final var blockStreamConfig = configProvider.getConfiguration().getConfigData(BlockStreamConfig.class);
+        if (!blockStreamConfig.streamToBlockNodes()) {
+            logger.info("Block stream to Block Nodes is disabled, skipping block node connection initialization");
+            return;
+        }
+
+        logger.info("Initializing block node connections with timeout {}", timeout);
+        boolean connected = daggerApp.blockNodeConnectionManager().waitForConnection(timeout);
+        if (blockStreamConfig.shutdownNodeOnNoBlockNodes() && !connected) {
+            logger.error("No block node connections established within timeout, shutting down");
+            this.shutdown();
+            System.exit(1);
+        }
+    }
 }

hedera-node/hedera-app/src/main/java/com/hedera/node/app/HederaInjectionComponent.java

@@ -26,6 +26,7 @@
 import com.hedera.node.app.blocks.InitialStateHash;
 import com.hedera.node.app.blocks.impl.BoundaryStateChangeListener;
 import com.hedera.node.app.blocks.impl.KVStateChangeListener;
+import com.hedera.node.app.blocks.impl.streaming.BlockNodeConnectionManager;
 import com.hedera.node.app.components.IngestInjectionComponent;
 import com.hedera.node.app.config.BootstrapConfigProviderImpl;
 import com.hedera.node.app.config.ConfigProviderImpl;
@@ -138,6 +139,8 @@ public interface HederaInjectionComponent {
 
     BlockRecordManager blockRecordManager();
 
+    BlockNodeConnectionManager blockNodeConnectionManager();
+
     BlockStreamManager blockStreamManager();
 
     FeeManager feeManager();

hedera-node/hedera-app/src/main/java/com/hedera/node/app/ServicesMain.java

@@ -56,6 +56,7 @@
 import com.hedera.node.app.state.StateLifecyclesImpl;
 import com.hedera.node.app.tss.TssBlockHashSigner;
 import com.hedera.node.app.version.ServicesSoftwareVersion;
+import com.hedera.node.config.data.BlockStreamConfig;
 import com.hedera.node.internal.network.Network;
 import com.hedera.pbj.runtime.io.buffer.Bytes;
 import com.swirlds.base.time.Time;
@@ -101,6 +102,7 @@
 import com.swirlds.state.State;
 import com.swirlds.state.merkle.MerkleStateRoot;
 import edu.umd.cs.findbugs.annotations.NonNull;
+import java.time.Duration;
 import java.time.InstantSource;
 import java.util.List;
 import java.util.Optional;
@@ -371,6 +373,12 @@ public static void main(final String... args) throws Exception {
                 .withSystemTransactionEncoderCallback(hedera::encodeSystemTransaction);
         final var platform = platformBuilder.build();
         hedera.init(platform, selfId);
+
+        // Initialize block node connections before starting the platform
+        final var waitPeriodForActiveConnection =
+                platformConfig.getConfigData(BlockStreamConfig.class).waitPeriodForActiveConnection();
+        hedera.initializeBlockNodeConnections(Duration.ofMinutes(waitPeriodForActiveConnection));
+
         platform.start();
         hedera.run();
     }

hedera-node/hedera-app/src/main/java/com/hedera/node/app/blocks/BlockStreamModule.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2024 Hedera Hashgraph, LLC
+ * Copyright (C) 2024-2025 Hedera Hashgraph, LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,12 +17,12 @@
 package com.hedera.node.app.blocks;
 
 import com.hedera.node.app.blocks.impl.BlockStreamManagerImpl;
-import com.hedera.node.app.blocks.impl.FileBlockItemWriter;
-import com.hedera.node.app.blocks.impl.GrpcBlockItemWriter;
+import com.hedera.node.app.blocks.impl.streaming.BlockNodeConnectionManager;
+import com.hedera.node.app.blocks.impl.streaming.FileBlockItemWriter;
+import com.hedera.node.app.blocks.impl.streaming.GrpcBlockItemWriter;
 import com.hedera.node.config.ConfigProvider;
 import com.hedera.node.config.data.BlockStreamConfig;
 import com.swirlds.state.lifecycle.info.NodeInfo;
-import dagger.Binds;
 import dagger.Module;
 import dagger.Provides;
 import edu.umd.cs.findbugs.annotations.NonNull;
@@ -31,23 +31,32 @@
 import javax.inject.Singleton;
 
 @Module
-public interface BlockStreamModule {
-    @Binds
+public class BlockStreamModule {
+
+    @Provides
     @Singleton
-    BlockStreamManager bindBlockStreamManager(BlockStreamManagerImpl blockStreamManager);
+    public BlockNodeConnectionManager provideBlockNodeConnectionManager(ConfigProvider configProvider) {
+        return new BlockNodeConnectionManager(configProvider);
+    }
+
+    @Provides
+    @Singleton
+    public BlockStreamManager provideBlockStreamManager(BlockStreamManagerImpl impl) {
+        return impl;
+    }
 
     @Provides
     @Singleton
     static Supplier<BlockItemWriter> bindBlockItemWriterSupplier(
             @NonNull final ConfigProvider configProvider,
             @NonNull final NodeInfo selfNodeInfo,
-            @NonNull final FileSystem fileSystem) {
+            @NonNull final FileSystem fileSystem,
+            @NonNull final BlockNodeConnectionManager blockNodeConnectionManager) {
         final var config = configProvider.getConfiguration();
         final var blockStreamConfig = config.getConfigData(BlockStreamConfig.class);
         return switch (blockStreamConfig.writerMode()) {
             case FILE -> () -> new FileBlockItemWriter(configProvider, selfNodeInfo, fileSystem);
-            case GRPC -> () -> new GrpcBlockItemWriter(blockStreamConfig);
+            case FILE_AND_GRPC -> () -> new GrpcBlockItemWriter(blockNodeConnectionManager);
         };
     }
-    ;
 }

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

@@ -468,7 +468,8 @@ private synchronized void finishProofWithSignature(
                     .siblingHashes(siblingHashes.stream().flatMap(List::stream).toList());
             final var proofItem = BlockItem.newBuilder().blockProof(proof).build();
             block.writer().writePbjItem(BlockItem.PROTOBUF.toBytes(proofItem));
-            if (streamWriterType == BlockStreamWriterMode.FILE) {
+            if (streamWriterType == BlockStreamWriterMode.FILE
+                    || streamWriterType == BlockStreamWriterMode.FILE_AND_GRPC) {
                 block.writer().closeBlock();
             }
             if (block.number() != blockNumber) {