Adding a new plugin

 * Instead of beating on the old implementation, implement a new subscriber plugin
   that uses a single client thread and feeds that alternately from messaging or history
   This avoids the countdown latch and state transition logic.
   Also avoids a state machine, just take the next block from one source or the other,
   always getting just the next block to send the client.
   Does require a secondary queue from the messaging, but it's small and simple.
* Adjust the subscriber "method" class to be reused, so we track sessions properly
* Added a latch so the pipeline thread holds until the session is registered with
  messaging before returning. This has minimal impact in production, and makes
  it possible to test reliably.
* Added live block push to queue, live blocks stream correctly.
* Added handling for request ahead of live condition
    * Session just skips over live blocks until live reaches the start block.
* Added proper shutdown of all sessions if the plugin is stopped.
* History and Live both work, and the test disables history to prove it
    * This is necessary because otherwise the test "history" plugin is too fast and we always serve history.

* TODO:
    * Add more unit tests
    * Add testing that forces a client to fall behind and catch up
    * Add some end-to-end test suites to exercise this plugin better
    * Add more and better javadoc
    * Clean up the (excessive) trace logging
    * Add better metrics, the "transition" metrics aren't super helpful.
    * Suggested metrics
        * Live batches sent to client (would be good if we can tag this with a client ID)
        * History blocks sent to client (would be good if we can tag this with a client ID)
        * Count of times the client thread must wait for live blocks, and time spent waiting
        * Count of times the client cannot read from history, but subsequently reads from live
        * Client connection and time from connect to session established

Signed-off-by: Joseph Sinclair <121976561+jsync-swirlds@users.noreply.github.com>

Author: jsync-swirlds

block-node/app/src/testFixtures/java/org/hiero/block/node/app/fixtures/plugintest/GrpcPluginTestBase.java

@@ -60,7 +60,6 @@ public void clientEndStreamReceived() {
             @Override
             public void onNext(Bytes item) throws RuntimeException {
                 fromPluginBytes.add(item);
-                LOGGER.log(TRACE, "onNext: %d".formatted(fromPluginBytes.size()));
             }
 
             @Override

block-node/app/src/testFixtures/java/org/hiero/block/node/app/fixtures/plugintest/SimpleInMemoryHistoricalBlockFacility.java

@@ -30,6 +30,7 @@ public class SimpleInMemoryHistoricalBlockFacility implements HistoricalBlockFac
     private final AtomicLong currentBlockNumber = new AtomicLong(UNKNOWN_BLOCK_NUMBER);
     private final List<BlockItems> partialBlock = new ArrayList<>();
     private final AtomicBoolean delayResponses = new AtomicBoolean(false);
+    private final AtomicBoolean disablePlugin = new AtomicBoolean(false);
     private BlockNodeContext blockNodeContext;
 
     @Override
@@ -42,28 +43,30 @@ public void init(BlockNodeContext context, ServiceBuilder serviceBuilder) {
      */
     @Override
     public void handleBlockItemsReceived(BlockItems blockItems) {
-        if (blockItems.isStartOfNewBlock()) {
-            if (!partialBlock.isEmpty()) {
-                throw new RuntimeException("Something went wrong, partitionedBlock is not empty. So we never got a end "
-                        + "block for current block");
+        if (!disablePlugin.get()) {
+            if (blockItems.isStartOfNewBlock()) {
+                if (!partialBlock.isEmpty()) {
+                    throw new RuntimeException(
+                            "Something went wrong, partitionedBlock is not empty. So we never got a end block for current block");
+                }
+                currentBlockNumber.set(blockItems.newBlockNumber());
             }
-            currentBlockNumber.set(blockItems.newBlockNumber());
-        }
-        partialBlock.add(blockItems);
-        if (blockItems.isEndOfBlock()) {
-            final long blockNumber = currentBlockNumber.getAndSet(UNKNOWN_BLOCK_NUMBER);
-            List<BlockItem> bi = new ArrayList<>();
-            for (BlockItems items : partialBlock) {
-                bi.addAll(toBlockItems(items.blockItems()));
+            partialBlock.add(blockItems);
+            if (blockItems.isEndOfBlock()) {
+                final long blockNumber = currentBlockNumber.getAndSet(UNKNOWN_BLOCK_NUMBER);
+                List<BlockItem> bi = new ArrayList<>();
+                for (BlockItems items : partialBlock) {
+                    bi.addAll(toBlockItems(items.blockItems()));
+                }
+                Block block = new Block(bi);
+                blockStorage.put(blockNumber, block);
+                availableBlocks.add(blockNumber);
+                partialBlock.clear();
+                // send block persisted message
+                blockNodeContext
+                        .blockMessaging()
+                        .sendBlockPersisted(new PersistedNotification(blockNumber, blockNumber, 2000));
             }
-            Block block = new Block(bi);
-            blockStorage.put(blockNumber, block);
-            availableBlocks.add(blockNumber);
-            partialBlock.clear();
-            // send block persisted message
-            blockNodeContext
-                    .blockMessaging()
-                    .sendBlockPersisted(new PersistedNotification(blockNumber, blockNumber, 2000));
         }
     }
 
@@ -97,6 +100,14 @@ public void clearDelayResponses() {
         delayResponses.compareAndSet(true, false);
     }
 
+    public void setDisablePlugin() {
+        disablePlugin.compareAndSet(false, true);
+    }
+
+    public void clearDisablePlugin() {
+        disablePlugin.compareAndSet(true, false);
+    }
+
     /**
      * {@inheritDoc}
      */

block-node/app/src/testFixtures/java/org/hiero/block/node/app/fixtures/plugintest/TestBlockMessagingFacility.java

@@ -134,8 +134,8 @@ public void sendBlockItems(BlockItems blockItems) {
         final int handlerCount = blockItemHandlers.size() + nonBackpressureBlockItemHandlers.size();
         LOGGER.log(
                 Level.TRACE,
-                "Sending next %d block items to %d handlers."
-                        .formatted(blockItems.blockItems().size(), handlerCount));
+                "Sending next %d block items for block %d to %d handlers."
+                        .formatted(blockItems.blockItems().size(), blockItems.newBlockNumber(), handlerCount));
         sentBlockBlockItems.add(blockItems);
         boolean handlerHasBackpressure = false;
         for (BlockItemHandler handler : blockItemHandlers) {

block-node/block-access/src/main/java/org/hiero/block/node/access/service/BlockAccessServicePlugin.java

@@ -18,6 +18,7 @@
 import org.hiero.block.api.BlockRequest;
 import org.hiero.block.api.BlockResponse;
 import org.hiero.block.api.BlockResponse.Code;
+// PBJ doesn't generate GRPC stubs for some reason, also the proto file is broken when PBJ compiles it...
 import org.hiero.block.api.protoc.BlockAccessServiceGrpc;
 import org.hiero.block.node.spi.BlockNodeContext;
 import org.hiero.block.node.spi.BlockNodePlugin;

block-node/stream-subscriber/build.gradle.kts

@@ -0,0 +1,21 @@
+// SPDX-License-Identifier: Apache-2.0
+plugins { id("org.hiero.gradle.module.library") }
+
+description = "Hiero Block Node Subscriber Service"
+
+// Remove the following line to enable all 'javac' lint checks that we have turned on by default
+// and then fix the reported issues.
+tasks.withType<JavaCompile>().configureEach { options.compilerArgs.add("-Xlint:-exports") }
+
+mainModuleInfo {
+    runtimeOnly("com.swirlds.config.impl")
+    runtimeOnly("org.apache.logging.log4j.slf4j2.impl")
+    runtimeOnly("io.helidon.logging.jul")
+    runtimeOnly("com.hedera.pbj.grpc.helidon.config")
+}
+
+testModuleInfo {
+    requires("org.junit.jupiter.api")
+    requires("org.hiero.block.node.app.test.fixtures")
+    requires("org.assertj.core.api")
+}

block-node/stream-subscriber/src/main/java/module-info.java

@@ -0,0 +1,24 @@
+// SPDX-License-Identifier: Apache-2.0
+import org.hiero.block.node.stream.subscriber.SubscriberServicePlugin;
+
+// SPDX-License-Identifier: Apache-2.0
+module org.hiero.block.node.stream.subscriber {
+    uses com.swirlds.config.api.spi.ConfigurationBuilderFactory;
+
+    // export configuration classes to the config module and app
+    exports org.hiero.block.node.stream.subscriber to
+            com.swirlds.config.impl,
+            com.swirlds.config.extensions,
+            org.hiero.block.node.app;
+
+    requires transitive com.hedera.pbj.runtime;
+    requires transitive com.swirlds.config.api;
+    requires transitive org.hiero.block.node.spi;
+    requires transitive org.hiero.block.protobuf;
+    requires com.swirlds.metrics.api;
+    requires org.hiero.block.node.base;
+    requires com.github.spotbugs.annotations;
+
+    provides org.hiero.block.node.spi.BlockNodePlugin with
+            SubscriberServicePlugin;
+}