fix: consumers will only start receiving blockitems at the beginning of the next full block

Signed-off-by: Matt Peterson <[email protected]>

Author: mattp-swirldslabs

protos/src/main/protobuf/blockstream.proto

@@ -48,13 +48,10 @@ message SubscribeStreamResponse {
  * It will be replaced with a PBJ implementation in the future.
  */
 message BlockItem {
-    /**
-     * The id of the block.  Each block id should be unique.
-     */
-    int64 id = 1;
-
-    /**
-     * The value of the block.  The value can be any string.
-     */
-    string value = 2;
+
+    int64 block_header = 1;
+
+    uint64 id = 2;
+
+    string value = 3;
 }

server/src/main/java/com/hedera/block/server/consumer/ConsumerBlockItemObserver.java

@@ -39,14 +39,15 @@ public class ConsumerBlockItemObserver implements BlockItemEventHandler<ObjectEv
     private final StreamObserver<BlockItem> subscribeStreamResponseObserver;
 
     private final long timeoutThresholdMillis;
-
     private final InstantSource producerLivenessClock;
     private long producerLivenessMillis;
 
     private final CountDownLatch shutdownLatch = new CountDownLatch(1);
 
     private final StreamMediator<ObjectEvent<BlockItem>, SubscribeStreamRequest> streamMediator;
 
+    private boolean isReachedFirstBlock;
+
     /**
      * Constructor for the LiveStreamObserverImpl class.
      *
@@ -91,16 +92,23 @@ public void onEvent(final ObjectEvent<BlockItem> event, final long l, final bool
 
         // Refresh the producer liveness and pass the block to the observer.
         producerLivenessMillis = producerLivenessClock.millis();
-        subscribeStreamResponseObserver.onNext(event.get());
+
+        final BlockItem blockItem = event.get();
+        if (!isReachedFirstBlock && blockItem.getBlockHeader() > 0) {
+            isReachedFirstBlock = true;
+        }
+
+        if (isReachedFirstBlock) {
+            subscribeStreamResponseObserver.onNext(blockItem);
+        }
     }
 
     /**
      * The onNext() method is triggered by Helidon when a consumer sends a blockResponse via the bidirectional stream.
      *
-     * @param blockItemResponse the BlockItemResponse passed back to the server via the bidirectional stream to the downstream consumer.
      */
     @Override
-    public void onNext(final SubscribeStreamRequest blockItemResponse) {
+    public void onNext(final SubscribeStreamRequest subscribeStreamRequest) {
 
         // Check if the producer has timed out.  If so, unsubscribe the observer from the mediator.
         if (isThresholdExceeded(producerLivenessMillis)) {

server/src/main/java/com/hedera/block/server/mediator/LiveStreamMediatorImpl.java

@@ -72,7 +72,7 @@ public void publishEvent(BlockItem blockItem) {
         ringBuffer.publishEvent((event, sequence) -> event.set(blockItem));
 
         // Block persistence
-        blockPersistenceHandler.persist(blockItem);
+//        blockPersistenceHandler.persist(blockItem);
     }
 
     @Override

server/src/test/resources/producer.sh

@@ -43,15 +43,27 @@ trap cleanup SIGINT
 (
   iter=$1
   while true; do
-    echo "{\"block_item\": {\"id\": $iter,\"value\": \"est dolor nulla\"}}"
 
-    if [ $iter -eq $2 ]; then
-      exit 0
-    fi
+    # Generate 10 BlockItems per Block
+    for ((i=1; i<=10; i++))
+    do
+
+      if [[ i -eq 1 ]]; then
+        echo "{\"block_item\": {\"block_header\": $iter,\"id\": $i,\"value\": \"Lorem ipsum dolor sit amet\"}}"
+      else
+        echo "{\"block_item\": {\"block_header\": -1,\"id\": $i,\"value\": \"est dolor nulla\"}}"
+      fi
+
+
+      if [ $iter -eq $2 ]; then
+        exit 0
+      fi
+
+      sleep 0.5
+    done
 
     ((iter++))
 
-    sleep 1
   done
 ) | grpcurl -vv -plaintext -proto $PATH_TO_PROTO -d @ $GRPC_SERVER $GRPC_METHOD &