feat: Block Verification Feature

server/docs/configuration.md

@@ -9,16 +9,20 @@ The default configuration allows users to quickly get up and running without hav
 ease of use at the trade-off of some insecure default configuration. Most configuration settings have appropriate
 defaults and can be left unchanged. It is recommended to browse the properties below and adjust to your needs.
 
-| Environment Variable | Description | Default Value |
-|:---|:---|---:|
-| PERSISTENCE_STORAGE_LIVE_ROOT_PATH | The root path for the live storage. | |
-| PERSISTENCE_STORAGE_ARCHIVE_ROOT_PATH | The root path for the archive storage. | |
-| PERSISTENCE_STORAGE_TYPE | Type of the persistence storage | BLOCK_AS_LOCAL_FILE |
-| PERSISTENCE_STORAGE_COMPRESSION | Compression algorithm used during persistence (could be none as well) | ZSTD |
-| PERSISTENCE_STORAGE_COMPRESSION_LEVEL | Compression level to be used by the compression algorithm | 3 |
-| CONSUMER_TIMEOUT_THRESHOLD_MILLIS | Time to wait for subscribers before disconnecting in milliseconds | 1500 |
-| SERVICE_DELAY_MILLIS | Service shutdown delay in milliseconds | 500 |
-| MEDIATOR_RING_BUFFER_SIZE | Size of the ring buffer used by the mediator (must be a power of 2) | 67108864 |
-| NOTIFIER_RING_BUFFER_SIZE | Size of the ring buffer used by the notifier (must be a power of 2) | 2048 |
-| SERVER_PORT | The port the server will listen on | 8080 |
-| SERVER_MAX_MESSAGE_SIZE_BYTES | The maximum size of a message frame in bytes | 1048576 |
+| Environment Variable                   | Description                                                              |       Default Value |
+|:---------------------------------------|:-------------------------------------------------------------------------|--------------------:|
+| PERSISTENCE_STORAGE_LIVE_ROOT_PATH     | The root path for the live storage.                                      |                     |
+| PERSISTENCE_STORAGE_ARCHIVE_ROOT_PATH  | The root path for the archive storage.                                   |                     |
+| PERSISTENCE_STORAGE_TYPE               | Type of the persistence storage                                          | BLOCK_AS_LOCAL_FILE |
+| PERSISTENCE_STORAGE_COMPRESSION        | Compression algorithm used during persistence (could be none as well)    |                ZSTD |
+| PERSISTENCE_STORAGE_COMPRESSION_LEVEL  | Compression level to be used by the compression algorithm                |                   3 |
+| CONSUMER_TIMEOUT_THRESHOLD_MILLIS      | Time to wait for subscribers before disconnecting in milliseconds        |                1500 |
+| SERVICE_DELAY_MILLIS                   | Service shutdown delay in milliseconds                                   |                 500 |
+| MEDIATOR_RING_BUFFER_SIZE              | Size of the ring buffer used by the mediator (must be a power of 2)      |            67108864 |
+| NOTIFIER_RING_BUFFER_SIZE              | Size of the ring buffer used by the notifier (must be a power of 2)      |                2048 |
+| SERVER_PORT                            | The port the server will listen on                                       |                8080 |
+| SERVER_MAX_MESSAGE_SIZE_BYTES          | The maximum size of a message frame in bytes                             |             1048576 |
+| VERIFICATION_ENABLED                   | Enables or disables the block verification process                       |                true |
+| VERIFICATION_SESSION_TYPE              | The type of BlockVerificationSession to use, either `ASYNC` or `SYNC`    |               ASYNC |
+| VERIFICATION_HASH_COMBINE_BATCH_SIZE   | The number of hashes to combine into a single hash during verification   |                  32 |
+

server/src/main/java/com/hedera/block/server/BlockNodeAppInjectionComponent.java

@@ -24,6 +24,7 @@
 import com.hedera.block.server.pbj.PbjInjectionModule;
 import com.hedera.block.server.persistence.PersistenceInjectionModule;
 import com.hedera.block.server.service.ServiceInjectionModule;
+import com.hedera.block.server.verification.VerificationInjectionModule;
 import com.swirlds.config.api.Configuration;
 import dagger.BindsInstance;
 import dagger.Component;
@@ -42,6 +43,7 @@
             ConfigInjectionModule.class,
             MetricsInjectionModule.class,
             PbjInjectionModule.class,
+            VerificationInjectionModule.class,
         })
 public interface BlockNodeAppInjectionComponent {
     /**

server/src/main/java/com/hedera/block/server/config/BlockNodeConfigExtension.java

@@ -24,6 +24,7 @@
 import com.hedera.block.server.persistence.storage.PersistenceStorageConfig;
 import com.hedera.block.server.producer.ProducerConfig;
 import com.hedera.block.server.service.ServiceConfig;
+import com.hedera.block.server.verification.VerificationConfig;
 import com.swirlds.common.metrics.config.MetricsConfig;
 import com.swirlds.common.metrics.platform.prometheus.PrometheusConfig;
 import com.swirlds.config.api.ConfigurationExtension;
@@ -56,6 +57,7 @@ public Set<Class<? extends Record>> getConfigDataTypes() {
                 ProducerConfig.class,
                 ConsumerConfig.class,
                 PersistenceStorageConfig.class,
-                ServerConfig.class);
+                ServerConfig.class,
+                VerificationConfig.class);
     }
 }

server/src/main/java/com/hedera/block/server/config/ConfigInjectionModule.java

@@ -22,6 +22,7 @@
 import com.hedera.block.server.notifier.NotifierConfig;
 import com.hedera.block.server.persistence.storage.PersistenceStorageConfig;
 import com.hedera.block.server.producer.ProducerConfig;
+import com.hedera.block.server.verification.VerificationConfig;
 import com.swirlds.common.metrics.config.MetricsConfig;
 import com.swirlds.common.metrics.platform.prometheus.PrometheusConfig;
 import com.swirlds.config.api.Configuration;
@@ -131,4 +132,16 @@ static ProducerConfig provideProducerConfig(Configuration configuration) {
     static ServerConfig provideServerConfig(Configuration configuration) {
         return configuration.getConfigData(ServerConfig.class);
     }
+
+    /**
+     * Provides a verification configuration singleton using the configuration.
+     *
+     * @param configuration is the configuration singleton
+     * @return a verification configuration singleton
+     */
+    @Singleton
+    @Provides
+    static VerificationConfig provideVerificationConfig(Configuration configuration) {
+        return configuration.getConfigData(VerificationConfig.class);
+    }
 }

server/src/main/java/com/hedera/block/server/config/ServerMappedConfigSourceInitializer.java

@@ -41,7 +41,10 @@ public final class ServerMappedConfigSourceInitializer {
             new ConfigMapping("server.maxMessageSizeBytes", "SERVER_MAX_MESSAGE_SIZE_BYTES"),
             new ConfigMapping("server.port", "SERVER_PORT"),
             new ConfigMapping("prometheus.endpointEnabled", "PROMETHEUS_ENDPOINT_ENABLED"),
-            new ConfigMapping("prometheus.endpointPortNumber", "PROMETHEUS_ENDPOINT_PORT_NUMBER"));
+            new ConfigMapping("prometheus.endpointPortNumber", "PROMETHEUS_ENDPOINT_PORT_NUMBER"),
+            new ConfigMapping("verification.enabled", "VERIFICATION_ENABLED"),
+            new ConfigMapping("verification.sessionType", "VERIFICATION_SESSION_TYPE"),
+            new ConfigMapping("verification.hashCombineBatchSize", "VERIFICATION_HASH_COMBINE_BATCH_SIZE"));
 
     private ServerMappedConfigSourceInitializer() {}
 

server/src/main/java/com/hedera/block/server/metrics/BlockNodeMetricTypes.java

@@ -59,6 +59,23 @@ public enum Counter implements MetricMetadata {
         /** The number of single blocks not found via the singleBlock rpc service. */
         SingleBlocksNotFound("single_blocks_not_found", "Single Blocks Not Found"),
 
+        // Verification counters
+
+        /** The number of blocks received for verification. */
+        VerificationBlocksReceived("verification_blocks_received", "Blocks Received for Verification"),
+
+        /** The number of blocks verified successfully. */
+        VerificationBlocksVerified("verification_blocks_verified", "Blocks Verified"),
+
+        /** The number of blocks that failed verification. */
+        VerificationBlocksFailed("verification_blocks_failed", "Blocks Failed Verification"),
+
+        /** The number of blocks that failed verification due to an error. */
+        VerificationBlocksError("verification_blocks_error", "Blocks Verification Error"),
+
+        /** The time in nanoseconds taken to verify a block */
+        VerificationBlockTime("verification_block_time", "Block Verification Time"),
+
         // Error counters
 
         /** The number of errors encountered by the live block stream mediator. */

server/src/main/java/com/hedera/block/server/notifier/NotifierImpl.java

@@ -138,6 +138,11 @@ public void publish(@NonNull List<BlockItemUnparsed> blockItems) {
         }
     }
 
+    /**
+     * Builds an error stream response.
+     *
+     * @return the error stream response
+     */
     @NonNull
     static PublishStreamResponse buildErrorStreamResponse() {
         // TODO: Replace this with a real error enum.

server/src/main/java/com/hedera/block/server/pbj/PbjBlockAccessServiceProxy.java

@@ -95,6 +95,12 @@ public Pipeline<? super Bytes> open(
         }
     }
 
+    /**
+     * Executes the unary singleBlock gRPC method.
+     *
+     * @param singleBlockRequest the single block request
+     * @return the single block response
+     */
     SingleBlockResponseUnparsed singleBlock(SingleBlockRequest singleBlockRequest) {
 
         LOGGER.log(DEBUG, "Executing Unary singleBlock gRPC method");

server/src/main/java/com/hedera/block/server/pbj/PbjBlockStreamServiceProxy.java

@@ -29,6 +29,7 @@
 import com.hedera.block.server.producer.ProducerBlockItemObserver;
 import com.hedera.block.server.producer.ProducerConfig;
 import com.hedera.block.server.service.ServiceStatus;
+import com.hedera.block.server.verification.StreamVerificationHandlerImpl;
 import com.hedera.hapi.block.BlockItemUnparsed;
 import com.hedera.hapi.block.PublishStreamRequestUnparsed;
 import com.hedera.hapi.block.PublishStreamResponse;
@@ -65,6 +66,7 @@ public class PbjBlockStreamServiceProxy implements PbjBlockStreamService {
      * @param streamMediator the live stream mediator
      * @param serviceStatus the service status
      * @param streamPersistenceHandler the stream persistence handler
+     * @param streamVerificationHandler the stream verification handler
      * @param notifier the notifier
      * @param blockNodeContext the block node context
      */
@@ -73,13 +75,15 @@ public PbjBlockStreamServiceProxy(
             @NonNull final LiveStreamMediator streamMediator,
             @NonNull final ServiceStatus serviceStatus,
             @NonNull final BlockNodeEventHandler<ObjectEvent<SubscribeStreamResponseUnparsed>> streamPersistenceHandler,
+            @NonNull final StreamVerificationHandlerImpl streamVerificationHandler,
             @NonNull final Notifier notifier,
             @NonNull final BlockNodeContext blockNodeContext) {
         this.serviceStatus = serviceStatus;
         this.notifier = notifier;
         this.blockNodeContext = blockNodeContext;
 
         streamMediator.subscribe(streamPersistenceHandler);
+        streamMediator.subscribe(streamVerificationHandler);
         this.streamMediator = streamMediator;
     }
 
@@ -119,6 +123,12 @@ public Pipeline<? super Bytes> open(
         }
     }
 
+    /**
+     * Publishes the block stream.
+     *
+     * @param helidonProducerObserver the helidon producer observer
+     * @return the pipeline
+     */
     Pipeline<List<BlockItemUnparsed>> publishBlockStream(
             Pipeline<? super PublishStreamResponse> helidonProducerObserver) {
         LOGGER.log(DEBUG, "Executing bidirectional publishBlockStream gRPC method");
@@ -153,6 +163,12 @@ Pipeline<List<BlockItemUnparsed>> publishBlockStream(
         }
     }
 
+    /**
+     * Subscribes to the block stream.
+     *
+     * @param subscribeStreamRequest the subscribe stream request
+     * @param subscribeStreamResponseObserver the subscribe stream response observer
+     */
     void subscribeBlockStream(
             SubscribeStreamRequest subscribeStreamRequest,
             Pipeline<? super SubscribeStreamResponseUnparsed> subscribeStreamResponseObserver) {

server/src/main/java/com/hedera/block/server/persistence/storage/PersistenceStorageConfig.java

@@ -173,11 +173,25 @@ public enum CompressionType {
         private final int minCompressionLevel;
         private final int maxCompressionLevel;
 
+        /**
+         * Constructs a new instance of {@link CompressionType}.
+         *
+         * @param minCompressionLevel the minimum compression level
+         * @param maxCompressionLevel the maximum compression level
+         */
         CompressionType(final int minCompressionLevel, final int maxCompressionLevel) {
             this.minCompressionLevel = minCompressionLevel;
             this.maxCompressionLevel = maxCompressionLevel;
         }
 
+        /**
+         * This method verifies that the compression level is within the
+         * acceptable range for the given compression type.
+         *
+         * @param levelToCheck the compression level to check
+         * @throws IllegalArgumentException if the compression level is not within
+         * the acceptable range
+         */
         public void verifyCompressionLevel(final int levelToCheck) {
             Preconditions.requireInRange(levelToCheck, minCompressionLevel, maxCompressionLevel);
         }

server/src/main/java/com/hedera/block/server/persistence/storage/write/LocalBlockWriter.java

@@ -19,5 +19,7 @@
 /**
  * A marker interface that groups all writers that operate on a local file
  * system.
+ *
+ * @param <V> the type of the value to be written
  */
 interface LocalBlockWriter<V> extends BlockWriter<V> {}

server/src/main/java/com/hedera/block/server/verification/BlockVerificationStatus.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright (C) 2024 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.
+ * 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.hedera.block.server.verification;
+
+/**
+ * An enum representing the status of block verification.
+ */
+public enum BlockVerificationStatus {
+    /**
+     * The Block has been verified.
+     */
+    VERIFIED,
+    /**
+     * The Block failed verification, either due to an invalid signature or an invalid hash.
+     */
+    SIGNATURE_INVALID
+}