feat: add support to the block stream simulator to fail sending mid-block (#523)

Signed-off-by: Alex Kehayov <aleks.kehayov@limechain.tech>

Author: AlexKehayov

simulator/src/main/java/org/hiero/block/simulator/config/SimulatorMappedConfigSourceInitializer.java

@@ -24,6 +24,8 @@ public final class SimulatorMappedConfigSourceInitializer {
             new ConfigMapping("blockStream.streamingMode", "BLOCK_STREAM_STREAMING_MODE"),
             new ConfigMapping("blockStream.millisecondsPerBlock", "BLOCK_STREAM_MILLISECONDS_PER_BLOCK"),
             new ConfigMapping("blockStream.blockItemsBatchSize", "BLOCK_STREAM_BLOCK_ITEMS_BATCH_SIZE"),
+            new ConfigMapping("blockStream.midBlockFailType", "BLOCK_STREAM_MID_BLOCK_FAIL_TYPE"),
+            new ConfigMapping("blockStream.midBlockFailOffset", "BLOCK_STREAM_MID_BLOCK_FAIL_OFFSET"),
 
             // Block consumer configuration
             new ConfigMapping("consumer.startBlockNumber", "CONSUMER_START_BLOCK_NUMBER"),

simulator/src/main/java/org/hiero/block/simulator/config/data/BlockStreamConfig.java

@@ -4,6 +4,7 @@
 import com.swirlds.config.api.ConfigData;
 import com.swirlds.config.api.ConfigProperty;
 import org.hiero.block.simulator.config.logging.Loggable;
+import org.hiero.block.simulator.config.types.MidBlockFailType;
 import org.hiero.block.simulator.config.types.SimulatorMode;
 import org.hiero.block.simulator.config.types.StreamingMode;
 
@@ -26,7 +27,9 @@ public record BlockStreamConfig(
         @Loggable @ConfigProperty(defaultValue = "100_000") int maxBlockItemsToStream,
         @Loggable @ConfigProperty(defaultValue = "MILLIS_PER_BLOCK") StreamingMode streamingMode,
         @Loggable @ConfigProperty(defaultValue = "1000") int millisecondsPerBlock,
-        @Loggable @ConfigProperty(defaultValue = "1000") int blockItemsBatchSize) {
+        @Loggable @ConfigProperty(defaultValue = "1000") int blockItemsBatchSize,
+        @Loggable @ConfigProperty(defaultValue = "NONE") MidBlockFailType midBlockFailType,
+        @Loggable @ConfigProperty(defaultValue = "0") long midBlockFailOffset) {
 
     /**
      * Creates a new {@link Builder} instance for constructing a {@code BlockStreamConfig}.
@@ -48,6 +51,8 @@ public static class Builder {
         private StreamingMode streamingMode = StreamingMode.MILLIS_PER_BLOCK;
         private int millisecondsPerBlock = 1000;
         private int blockItemsBatchSize = 1000;
+        private MidBlockFailType midBlockFailType = MidBlockFailType.NONE;
+        private long midBlockFailOffset = 0;
 
         /**
          * Creates a new instance of the {@code Builder} class with default configuration values.
@@ -133,6 +138,28 @@ public Builder blockItemsBatchSize(int blockItemsBatchSize) {
             return this;
         }
 
+        /**
+         * Sets a failure type to occur while streaming.
+         *
+         * @param midBlockFailType the failure type
+         * @return this {@code Builder} instance
+         */
+        public Builder midBlockFailType(MidBlockFailType midBlockFailType) {
+            this.midBlockFailType = midBlockFailType;
+            return this;
+        }
+
+        /**
+         * Sets the index of the failing block.
+         *
+         * @param midBlockFailOffset the index of the failing block
+         * @return this {@code Builder} instance
+         */
+        public Builder midBlockFailOffset(long midBlockFailOffset) {
+            this.midBlockFailOffset = midBlockFailOffset;
+            return this;
+        }
+
         /**
          * Builds a new {@link BlockStreamConfig} instance with the configured values.
          *
@@ -146,7 +173,9 @@ public BlockStreamConfig build() {
                     maxBlockItemsToStream,
                     streamingMode,
                     millisecondsPerBlock,
-                    blockItemsBatchSize);
+                    blockItemsBatchSize,
+                    midBlockFailType,
+                    midBlockFailOffset);
         }
     }
 }

simulator/src/main/java/org/hiero/block/simulator/config/types/MidBlockFailType.java

@@ -0,0 +1,21 @@
+// SPDX-License-Identifier: Apache-2.0
+package org.hiero.block.simulator.config.types;
+
+/** The MidBlockFailType enum defines the type of failure in the process of block streaming.
+ * Failure will occur randomly between the header and the proof of the block. */
+public enum MidBlockFailType {
+    /**
+     * The NONE value indicates no failure will be simulated during streaming.
+     */
+    NONE,
+    /**
+     * The ABRUPT value indicates that an abrupt disconnection will occur while streaming
+     * (without closing the connection or sending an EndOfStream message).
+     */
+    ABRUPT,
+    /**
+     * The EOS value indicates that an EndOfStream message will be sent before the final item of the block.
+     * Currently, onError is called, as the client is not able to send actual EOS message
+     */
+    EOS
+}

simulator/src/main/java/org/hiero/block/simulator/grpc/PublishStreamGrpcClient.java

@@ -2,7 +2,6 @@
 package org.hiero.block.simulator.grpc;
 
 import com.hedera.hapi.block.stream.protoc.Block;
-import com.hedera.hapi.block.stream.protoc.BlockItem;
 import java.util.List;
 
 /**
@@ -14,14 +13,6 @@ public interface PublishStreamGrpcClient {
      */
     void init();
 
-    /**
-     * Streams the block item.
-     *
-     * @param blockItems list of the block item to be streamed
-     * @return true if the block item is streamed successfully, false otherwise
-     */
-    boolean streamBlockItem(List<BlockItem> blockItems);
-
     /**
      * Streams the block.
      *

simulator/src/main/java/org/hiero/block/simulator/grpc/impl/PublishStreamGrpcClientImpl.java

@@ -3,7 +3,6 @@
 
 import static java.lang.System.Logger.Level.DEBUG;
 import static java.lang.System.Logger.Level.ERROR;
-import static java.lang.System.Logger.Level.INFO;
 import static java.util.Objects.requireNonNull;
 import static org.hiero.block.simulator.metrics.SimulatorMetricTypes.Counter.LiveBlockItemsSent;
 import static org.hiero.block.simulator.metrics.SimulatorMetricTypes.Counter.LiveBlocksSent;
@@ -20,11 +19,13 @@
 import java.util.ArrayDeque;
 import java.util.Deque;
 import java.util.List;
+import java.util.Random;
 import java.util.concurrent.atomic.AtomicBoolean;
 import javax.inject.Inject;
 import org.hiero.block.common.utils.ChunkUtils;
 import org.hiero.block.simulator.config.data.BlockStreamConfig;
 import org.hiero.block.simulator.config.data.GrpcConfig;
+import org.hiero.block.simulator.config.types.MidBlockFailType;
 import org.hiero.block.simulator.grpc.PublishStreamGrpcClient;
 import org.hiero.block.simulator.metrics.MetricsService;
 import org.hiero.block.simulator.startup.SimulatorStartupData;
@@ -57,11 +58,11 @@ public class PublishStreamGrpcClientImpl implements PublishStreamGrpcClient {
     /**
      * Creates a new PublishStreamGrpcClientImpl with the specified dependencies.
      *
-     * @param grpcConfig The configuration for gRPC connection settings
+     * @param grpcConfig        The configuration for gRPC connection settings
      * @param blockStreamConfig The configuration for block streaming parameters
-     * @param metricsService The service for recording publication metrics
-     * @param streamEnabled Flag controlling stream state
-     * @param startupData The startup data for the simulator
+     * @param metricsService    The service for recording publication metrics
+     * @param streamEnabled     Flag controlling stream state
+     * @param startupData       The startup data for the simulator
      * @throws NullPointerException if any parameter is null
      */
     @Inject
@@ -95,33 +96,6 @@ public void init() {
         lastKnownStatuses.clear();
     }
 
-    /**
-     * Streams a list of block items to the server.
-     *
-     * @param blockItems The list of block items to stream
-     * @return true if streaming should continue, false if streaming should stop
-     */
-    @Override
-    public boolean streamBlockItem(List<BlockItem> blockItems) {
-        if (streamEnabled.get()) {
-            requestStreamObserver.onNext(PublishStreamRequest.newBuilder()
-                    .setBlockItems(BlockItemSet.newBuilder()
-                            .addAllBlockItems(blockItems)
-                            .build())
-                    .build());
-
-            metricsService.get(LiveBlockItemsSent).add(blockItems.size());
-            LOGGER.log(
-                    INFO,
-                    "Number of block items sent: "
-                            + metricsService.get(LiveBlockItemsSent).get());
-        } else {
-            LOGGER.log(ERROR, "Not allowed to send next batch of block items");
-        }
-
-        return streamEnabled.get();
-    }
-
     /**
      * Streams a complete block to the server, chunking it if necessary based on configuration.
      *
@@ -134,6 +108,7 @@ public boolean streamBlock(Block block) {
                 ChunkUtils.chunkify(block.getItemsList(), blockStreamConfig.blockItemsBatchSize());
         for (List<BlockItem> streamingBatch : streamingBatches) {
             if (streamEnabled.get()) {
+                handleMidBlockFailIfSet(streamingBatch);
                 requestStreamObserver.onNext(PublishStreamRequest.newBuilder()
                         .setBlockItems(BlockItemSet.newBuilder()
                                 .addAllBlockItems(streamingBatch)
@@ -192,4 +167,28 @@ public void shutdown() throws InterruptedException {
         completeStreaming();
         channel.shutdown();
     }
+
+    private void handleMidBlockFailIfSet(List<BlockItem> streamingBatch) {
+        if (blockStreamConfig.midBlockFailType() != MidBlockFailType.NONE
+                && blockStreamConfig.midBlockFailOffset()
+                        == metricsService.get(LiveBlocksSent).get()) {
+            // streaming fails after the BlockHeader, but before the last BlockItem or the BlockProof
+            // the block is considered to have a proper structure
+            if (streamingBatch.size() < 3) {
+                throw new IllegalArgumentException("Block items batch size is too small for mid block fail simulation");
+            }
+            int failIndex = new Random().nextInt(1, streamingBatch.size() - 1);
+            List<BlockItem> streamingBatchBeforeFail = streamingBatch.subList(0, failIndex);
+            requestStreamObserver.onNext(PublishStreamRequest.newBuilder()
+                    .setBlockItems(BlockItemSet.newBuilder()
+                            .addAllBlockItems(streamingBatchBeforeFail)
+                            .build())
+                    .build());
+            if (blockStreamConfig.midBlockFailType() == MidBlockFailType.ABRUPT) {
+                throw new RuntimeException("Configured abrupt disconnection occurred");
+            } else if (blockStreamConfig.midBlockFailType() == MidBlockFailType.EOS) {
+                requestStreamObserver.onError(new Exception("Configured failure occurred, calling onError()"));
+            }
+        }
+    }
 }

simulator/src/main/resources/app.properties

@@ -36,6 +36,10 @@ prometheus.endpointPortNumber=9998
 #blockStream.millisecondsPerBlock=500
 #blockStream.blockItemsBatchSize=1_000
 #blockStream.delayBetweenBlockItems=3_000_000
+# When this property is active it causes an abrupt disconnect or sending EndOfStream message (NONE/ABRUPT/EOS)
+#blockStream.midBlockFailType=NONE
+# The number of blocks streamed before the failure
+#blockStream.midBlockFailOffset=0
 
 # ----------------------------------------------
 

simulator/src/test/java/org/hiero/block/simulator/config/SimulatorMappedConfigSourceInitializerTest.java

@@ -47,6 +47,8 @@ class SimulatorMappedConfigSourceInitializerTest {
         new ConfigMapping("blockStream.streamingMode", "BLOCK_STREAM_STREAMING_MODE"),
         new ConfigMapping("blockStream.millisecondsPerBlock", "BLOCK_STREAM_MILLISECONDS_PER_BLOCK"),
         new ConfigMapping("blockStream.blockItemsBatchSize", "BLOCK_STREAM_BLOCK_ITEMS_BATCH_SIZE"),
+        new ConfigMapping("blockStream.midBlockFailType", "BLOCK_STREAM_MID_BLOCK_FAIL_TYPE"),
+        new ConfigMapping("blockStream.midBlockFailOffset", "BLOCK_STREAM_MID_BLOCK_FAIL_OFFSET"),
 
         // Block consumer configuration
         new ConfigMapping("consumer.startBlockNumber", "CONSUMER_START_BLOCK_NUMBER"),

simulator/src/test/java/org/hiero/block/simulator/config/logging/SimulatorConfigurationLoggerTest.java

@@ -26,7 +26,7 @@ void testCurrentAppProperties() throws IOException {
         final SimulatorConfigurationLogger configurationLogging = new SimulatorConfigurationLogger(configuration);
         final Map<String, Object> config = configurationLogging.collectConfig(configuration);
         assertNotNull(config);
-        assertEquals(36, config.size());
+        assertEquals(38, config.size());
         for (final Map.Entry<String, Object> entry : config.entrySet()) {
             String value = entry.getValue().toString();
             if (value.contains("*")) {
@@ -42,7 +42,7 @@ void testWithMockedSensitiveProperty() throws IOException {
         final SimulatorConfigurationLogger configurationLogging = new SimulatorConfigurationLogger(configuration);
         final Map<String, Object> config = configurationLogging.collectConfig(configuration);
         assertNotNull(config);
-        assertEquals(38, config.size());
+        assertEquals(40, config.size());
         assertEquals("*****", config.get("test.secret").toString());
         assertEquals("", config.get("test.emptySecret").toString());
     }