fix: 1227 subscriber plugin does not close failed streams

* The Helidon plugin throws a `RuntimeException` to indicate a stream that was closed (even if _orderly_).
    * This change handles that exception by closing the session and ending the thread.
* Also removed the "subscription" that is not used in "server streaming" plugins.

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

Author: jsync-swirlds

block-node/stream-subscriber/src/main/java/org/hiero/block/node/stream/subscriber/BlockStreamSubscriberSession.java

@@ -17,7 +17,6 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Flow.Subscription;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -78,8 +77,6 @@ public class BlockStreamSubscriberSession implements Callable<BlockStreamSubscri
     private final CountDownLatch sessionReadyLatch;
     /** A flag indicating if the session should be interrupted */
     private final AtomicBoolean interruptedStream = new AtomicBoolean(false);
-    /** The subscription for the GRPC connection with client */
-    private Subscription subscription;
     /** The current block being sent to the client */
     private long nextBlockToSend;
     /** The latest block received from the live stream. */
@@ -481,17 +478,20 @@ synchronized void close(final SubscribeStreamResponse.Code endStreamResponseCode
             sessionReadyLatch.countDown();
             LOGGER.log(Level.DEBUG, "Session ready latch was not counted down on close, releasing now");
         }
-        if (!interruptedStream.get()) {
-            // unregister us from the block messaging system, if we are not registered then this is noop
-            context.blockMessaging().unregisterBlockItemHandler(liveBlockHandler);
+        // unregister us from the block messaging system, if we are not registered then this is noop
+        context.blockMessaging().unregisterBlockItemHandler(liveBlockHandler);
+        // send an end stream response, if we have a code to set and are not interrupted.
+        if (!interruptedStream.get() && endStreamResponseCode != null) {
             final Builder response =
                     SubscribeStreamResponseUnparsed.newBuilder().status(endStreamResponseCode);
             responsePipeline.onNext(response.build());
-            responsePipeline.onComplete();
         }
-        if (subscription != null) {
-            subscription.cancel();
-            subscription = null;
+        try {
+            responsePipeline.onComplete();
+        } catch (RuntimeException e) {
+            // If the pipeline cannot be completed, log and suppress this exception.
+            final String message = "Suppressed client error when \"completing\" stream for client %d%n%s";
+            LOGGER.log(Level.DEBUG, message.formatted(clientId, e.getMessage()), e);
         }
         // Break out of the loop that sends blocks to the client, so the thread completes.
         interruptedStream.set(true);
@@ -518,7 +518,17 @@ private void sendOneBlockItemSet(final BlockItems nextBatch) {
     private void sendOneBlockItemSet(final List<BlockItemUnparsed> blockItems) {
         final BlockItemSetUnparsed dataToSend = new BlockItemSetUnparsed(blockItems);
         final OneOf<ResponseOneOfType> responseOneOf = new OneOf<>(ResponseOneOfType.BLOCK_ITEMS, dataToSend);
-        responsePipeline.onNext(new SubscribeStreamResponseUnparsed(responseOneOf));
+        try {
+            responsePipeline.onNext(new SubscribeStreamResponseUnparsed(responseOneOf));
+        } catch (RuntimeException e) {
+            // If the pipeline is in an error state; close this session.
+            // Unfortunately this is the "standard" way to end a stream, so log
+            // at debug rather than emitting noise in the logs.
+            final String message =
+                    "Client error sending block items for client %d: %s".formatted(clientId, e.getMessage());
+            LOGGER.log(Level.DEBUG, message, e);
+            close(null); // cannot send the end stream response, just close the stream.
+        }
     }
 
     // ==== Block Item Handler Class ===========================================

block-node/stream-subscriber/src/test/java/org/hiero/block/node/stream/subscriber/BlockStreamSubscriberSessionTest.java

@@ -22,6 +22,7 @@
 import org.hiero.block.api.SubscribeStreamRequest;
 import org.hiero.block.api.SubscribeStreamResponse;
 import org.hiero.block.internal.SubscribeStreamResponseUnparsed;
+import org.hiero.block.internal.SubscribeStreamResponseUnparsed.ResponseOneOfType;
 import org.hiero.block.node.app.fixtures.plugintest.SimpleInMemoryHistoricalBlockFacility;
 import org.hiero.block.node.app.fixtures.plugintest.TestBlockMessagingFacility;
 import org.hiero.block.node.spi.BlockNodeContext;
@@ -78,7 +79,6 @@ void setUp() {
     @Nested
     @DisplayName("Streaming Functionality Tests")
     class StreamingTests {
-
         /**
          * Tests the complete flow of streaming both historical and live blocks.
          * Verifies that the session can handle:
@@ -107,7 +107,6 @@ void shouldStreamHistoricalAndLiveBlocksSuccessfully()
                 sessionReadyLatch.await();
                 // Phase 1: Historical Block Streaming
                 final int expectedResponseCount = END_BLOCK - START_BLOCK;
-
                 // Phase 2: Live Block Streaming
                 BlockItems[] liveBatches = createNumberOfSimpleBlockBatches(MAX_AVAILABLE_BLOCK, END_BLOCK + 1);
                 for (BlockItems next : liveBatches) {
@@ -116,12 +115,8 @@ void shouldStreamHistoricalAndLiveBlocksSuccessfully()
                 // Wait for everything to complete.
                 // Note, don't try to wait before this; there are no execution guarantees until
                 // `get` is called; before that the thread may not run or may be parked indefinitely.
-                sessionFuture.get(
-                        1L,
-                        TimeUnit.SECONDS); // The timeout doesn't work, for some reason, but it's here because we don't
-                // have a better alternative.
+                sessionFuture.get();
             }
-
             // Verify final pipeline state
             assertThat(responsePipeline.getReceivedResponses()).hasSize(21);
             assertThat(responsePipeline.getCompletionCount()).isEqualTo(1);
@@ -148,14 +143,90 @@ void shouldStreamHistoricalBlocksSuccessfully() {
             setupHistoricalBlockProvider(MIN_AVAILABLE_BLOCK, MAX_AVAILABLE_BLOCK);
             // Execute session
             session.call();
-
             // Verify pipeline interactions
             assertThat(responsePipeline.getReceivedResponses()).hasSize(11);
             assertThat(responsePipeline.getCompletionCount()).isEqualTo(1);
             assertThat(responsePipeline.getPipelineErrors()).isEmpty();
         }
     }
 
+    /**
+     * Tests related to streams ending for the BlockStreamSubscriberSession.
+     */
+    @Nested
+    @DisplayName("Stream End Tests")
+    class StreamEndTests {
+        /**
+         * Tests the complete flow of streaming both historical and live blocks.
+         * Verifies that the session can handle:
+         * 1. Historical block streaming
+         * 2. Transition to live block streaming
+         * 3. Proper pipeline interactions
+         */
+        @Test
+        @DisplayName("Should complete a combined stream when the client exits")
+        void shouldCompleteWhenStreamClosed() throws InterruptedException, ExecutionException, TimeoutException {
+            // Setup test parameters
+            final int MIN_AVAILABLE_BLOCK = 0;
+            final int MAX_AVAILABLE_BLOCK = 10;
+            final int START_BLOCK = 0;
+            final int END_BLOCK = 19;
+            final ResponsePipeline disconnectedPipeline = new ResponsePipeline(true);
+
+            // Initialize session
+            final SubscribeStreamRequest subscribeStreamRequest = createRequest(START_BLOCK, END_BLOCK);
+            setupHistoricalBlockProvider(MIN_AVAILABLE_BLOCK, MAX_AVAILABLE_BLOCK + 1);
+            session = new BlockStreamSubscriberSession(
+                    CLIENT_ID, subscribeStreamRequest, disconnectedPipeline, context, sessionReadyLatch);
+            try (final ExecutorService sessionExecutor = Executors.newVirtualThreadPerTaskExecutor()) {
+                Future<BlockStreamSubscriberSession> sessionFuture = sessionExecutor.submit(session);
+                sessionReadyLatch.await();
+                // Phase 1: Historical Block Streaming
+                final int expectedResponseCount = END_BLOCK - START_BLOCK;
+                // Phase 2: Live Block Streaming
+                BlockItems[] liveBatches = createNumberOfSimpleBlockBatches(MAX_AVAILABLE_BLOCK, END_BLOCK + 1);
+                for (BlockItems next : liveBatches) {
+                    blockMessagingFacility.sendBlockItems(next);
+                }
+                // Wait for everything to complete.
+                // Note, don't try to wait before this; there are no execution guarantees until
+                // `get` is called; before that the thread may not run or may be parked indefinitely.
+                sessionFuture.get();
+            }
+            // Verify final pipeline state
+            assertThat(disconnectedPipeline.getReceivedResponses()).hasSize(0);
+            assertThat(disconnectedPipeline.getCompletionCount()).isEqualTo(1);
+            assertThat(disconnectedPipeline.getPipelineErrors()).isEmpty();
+        }
+
+        /**
+         * Tests the historical block streaming functionality in isolation.
+         * Verifies that the session can properly stream blocks from the historical provider.
+         */
+        @Test
+        @DisplayName("Should complete a stream of historical blocks when the client exits")
+        void shouldCompleteWhenStreamFails() {
+            // Setup test parameters
+            final int MIN_AVAILABLE_BLOCK = 0;
+            final int MAX_AVAILABLE_BLOCK = 20;
+            final long START_BLOCK = 1L;
+            final long END_BLOCK = 10L;
+            final ResponsePipeline disconnectedPipeline = new ResponsePipeline(true);
+
+            // Initialize session
+            final SubscribeStreamRequest subscribeStreamRequest = createRequest(START_BLOCK, END_BLOCK);
+            session = new BlockStreamSubscriberSession(
+                    CLIENT_ID, subscribeStreamRequest, disconnectedPipeline, context, sessionReadyLatch);
+            setupHistoricalBlockProvider(MIN_AVAILABLE_BLOCK, MAX_AVAILABLE_BLOCK);
+            // Execute session
+            session.call();
+            // Verify pipeline interactions
+            assertThat(disconnectedPipeline.getReceivedResponses()).hasSize(0);
+            assertThat(disconnectedPipeline.getCompletionCount()).isEqualTo(1);
+            assertThat(disconnectedPipeline.getPipelineErrors()).isEmpty();
+        }
+    }
+
     /**
      * Sets up the historical block provider with the specified block range.
      *
@@ -195,8 +266,17 @@ private static class ResponsePipeline implements Pipeline<SubscribeStreamRespons
         private final List<SubscribeStreamResponse> receivedResponses = new ArrayList<>();
 
         private final List<Throwable> pipelineErrors = new ArrayList<>();
+        private final boolean throwOnNext;
         private int completionCount = 0;
 
+        public ResponsePipeline() {
+            this.throwOnNext = false;
+        }
+
+        public ResponsePipeline(final boolean throwOnNext) {
+            this.throwOnNext = throwOnNext;
+        }
+
         public List<SubscribeStreamResponse> getReceivedResponses() {
             return receivedResponses;
         }
@@ -217,6 +297,9 @@ public void onSubscribe(Flow.Subscription subscription) {}
 
         @Override
         public void onNext(SubscribeStreamResponseUnparsed item) {
+            if (throwOnNext && item.response().kind() != ResponseOneOfType.STATUS) {
+                throw new RuntimeException("Simulated \"closed\" stream.");
+            }
             try {
                 var binary = SubscribeStreamResponseUnparsed.PROTOBUF.toBytes(item);
                 var response = SubscribeStreamResponse.PROTOBUF.parse(binary);