refactor: Arch changes to address performance bottlenecks

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

@@ -20,6 +20,7 @@
 import static java.lang.System.Logger.Level.INFO;
 
 import com.hedera.block.server.health.HealthService;
+import com.hedera.block.server.service.ServiceStatus;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import io.helidon.webserver.WebServer;
 import io.helidon.webserver.WebServerConfig;

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

@@ -20,7 +20,9 @@
 import com.hedera.block.server.health.HealthInjectionModule;
 import com.hedera.block.server.mediator.MediatorInjectionModule;
 import com.hedera.block.server.metrics.MetricsInjectionModule;
+import com.hedera.block.server.notifier.NotifierInjectionModule;
 import com.hedera.block.server.persistence.PersistenceInjectionModule;
+import com.hedera.block.server.service.ServiceInjectionModule;
 import com.swirlds.config.api.Configuration;
 import dagger.BindsInstance;
 import dagger.Component;
@@ -30,6 +32,8 @@
 @Singleton
 @Component(
         modules = {
+            NotifierInjectionModule.class,
+            ServiceInjectionModule.class,
             BlockNodeAppInjectionModule.class,
             HealthInjectionModule.class,
             PersistenceInjectionModule.class,

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

@@ -33,16 +33,6 @@
 @Module
 public interface BlockNodeAppInjectionModule {
 
-    /**
-     * Binds the service status to the service status implementation.
-     *
-     * @param serviceStatus needs a service status implementation
-     * @return the service status implementation
-     */
-    @Singleton
-    @Binds
-    ServiceStatus bindServiceStatus(ServiceStatusImpl serviceStatus);
-
     /**
      * Provides a block node context singleton.
      *

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

@@ -32,19 +32,21 @@
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.hedera.block.server.config.BlockNodeContext;
 import com.hedera.block.server.consumer.ConsumerStreamResponseObserver;
-import com.hedera.block.server.data.ObjectEvent;
-import com.hedera.block.server.mediator.StreamMediator;
+import com.hedera.block.server.events.BlockNodeEventHandler;
+import com.hedera.block.server.events.ObjectEvent;
+import com.hedera.block.server.mediator.LiveStreamMediator;
 import com.hedera.block.server.metrics.MetricsService;
+import com.hedera.block.server.notifier.Notifier;
 import com.hedera.block.server.persistence.storage.read.BlockReader;
 import com.hedera.block.server.producer.ProducerBlockItemObserver;
+import com.hedera.block.server.service.ServiceStatus;
 import com.hedera.hapi.block.SingleBlockRequest;
 import com.hedera.hapi.block.SingleBlockResponse;
 import com.hedera.hapi.block.SingleBlockResponseCode;
 import com.hedera.hapi.block.SubscribeStreamResponse;
 import com.hedera.hapi.block.SubscribeStreamResponseCode;
 import com.hedera.hapi.block.protoc.BlockService;
 import com.hedera.hapi.block.stream.Block;
-import com.hedera.hapi.block.stream.BlockItem;
 import com.hedera.pbj.runtime.ParseException;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import io.grpc.stub.StreamObserver;
@@ -63,12 +65,15 @@ public class BlockStreamService implements GrpcService {
 
     private final Logger LOGGER = System.getLogger(getClass().getName());
 
-    private final StreamMediator<BlockItem, ObjectEvent<SubscribeStreamResponse>> streamMediator;
+    private final LiveStreamMediator streamMediator;
     private final ServiceStatus serviceStatus;
     private final BlockReader<Block> blockReader;
+
     private final BlockNodeContext blockNodeContext;
     private final MetricsService metricsService;
 
+    private final Notifier notifier;
+
     /**
      * Constructor for the BlockStreamService class. It initializes the BlockStreamService with the
      * given parameters.
@@ -82,17 +87,22 @@ public class BlockStreamService implements GrpcService {
      */
     @Inject
     BlockStreamService(
-            @NonNull
-                    final StreamMediator<BlockItem, ObjectEvent<SubscribeStreamResponse>>
-                            streamMediator,
+            @NonNull final LiveStreamMediator streamMediator,
             @NonNull final BlockReader<Block> blockReader,
             @NonNull final ServiceStatus serviceStatus,
+            @NonNull
+                    final BlockNodeEventHandler<ObjectEvent<SubscribeStreamResponse>>
+                            streamPersistenceHandler,
+            @NonNull final Notifier notifier,
             @NonNull final BlockNodeContext blockNodeContext) {
-        this.streamMediator = streamMediator;
         this.blockReader = blockReader;
         this.serviceStatus = serviceStatus;
+        this.notifier = notifier;
         this.blockNodeContext = blockNodeContext;
         this.metricsService = blockNodeContext.metricsService();
+
+        streamMediator.subscribe(streamPersistenceHandler);
+        this.streamMediator = streamMediator;
     }
 
     /**
@@ -139,8 +149,23 @@ StreamObserver<com.hedera.hapi.block.protoc.PublishStreamRequest> protocPublishB
                             publishStreamResponseObserver) {
         LOGGER.log(DEBUG, "Executing bidirectional publishBlockStream gRPC method");
 
-        return new ProducerBlockItemObserver(
-                streamMediator, publishStreamResponseObserver, blockNodeContext, serviceStatus);
+        // Unsubscribe any expired notifiers
+        notifier.unsubscribeAllExpired();
+
+        final var producerBlockItemObserver =
+                new ProducerBlockItemObserver(
+                        Clock.systemDefaultZone(),
+                        streamMediator,
+                        notifier,
+                        publishStreamResponseObserver,
+                        blockNodeContext,
+                        serviceStatus);
+
+        // Register the producer observer with the notifier to publish responses back to the
+        // producer
+        notifier.subscribe(producerBlockItemObserver);
+
+        return producerBlockItemObserver;
     }
 
     void protocSubscribeBlockStream(
@@ -152,16 +177,18 @@ void protocSubscribeBlockStream(
                             subscribeStreamResponseObserver) {
         LOGGER.log(DEBUG, "Executing Server Streaming subscribeBlockStream gRPC method");
 
-        // Return a custom StreamObserver to handle streaming blocks from the producer.
         if (serviceStatus.isRunning()) {
-            final var streamObserver =
+            // Unsubscribe any expired notifiers
+            streamMediator.unsubscribeAllExpired();
+
+            final var consumerStreamResponseObserver =
                     new ConsumerStreamResponseObserver(
-                            blockNodeContext,
                             Clock.systemDefaultZone(),
                             streamMediator,
-                            subscribeStreamResponseObserver);
+                            subscribeStreamResponseObserver,
+                            blockNodeContext);
 
-            streamMediator.subscribe(streamObserver);
+            streamMediator.subscribe(consumerStreamResponseObserver);
         } else {
             LOGGER.log(
                     ERROR,

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

@@ -57,7 +57,7 @@ public static void main(final String[] args) throws IOException {
         Config.global(config);
 
         // Init BlockNode Configuration
-        Configuration configuration =
+        final Configuration configuration =
                 ConfigurationBuilder.create()
                         .withSource(SystemEnvironmentConfigSource.getInstance())
                         .withSource(SystemPropertiesConfigSource.getInstance())

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

@@ -18,7 +18,10 @@
 
 import com.google.auto.service.AutoService;
 import com.hedera.block.server.consumer.ConsumerConfig;
+import com.hedera.block.server.mediator.MediatorConfig;
+import com.hedera.block.server.notifier.NotifierConfig;
 import com.hedera.block.server.persistence.storage.PersistenceStorageConfig;
+import com.hedera.block.server.service.ServiceConfig;
 import com.swirlds.common.metrics.config.MetricsConfig;
 import com.swirlds.common.metrics.platform.prometheus.PrometheusConfig;
 import com.swirlds.config.api.ConfigurationExtension;
@@ -43,6 +46,9 @@ public BlockNodeConfigExtension() {
     @Override
     public Set<Class<? extends Record>> getConfigDataTypes() {
         return Set.of(
+                ServiceConfig.class,
+                MediatorConfig.class,
+                NotifierConfig.class,
                 MetricsConfig.class,
                 PrometheusConfig.class,
                 ConsumerConfig.class,

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

@@ -17,6 +17,8 @@
 package com.hedera.block.server.config;
 
 import com.hedera.block.server.consumer.ConsumerConfig;
+import com.hedera.block.server.mediator.MediatorConfig;
+import com.hedera.block.server.notifier.NotifierConfig;
 import com.hedera.block.server.persistence.storage.PersistenceStorageConfig;
 import com.swirlds.common.metrics.config.MetricsConfig;
 import com.swirlds.common.metrics.platform.prometheus.PrometheusConfig;
@@ -79,4 +81,28 @@ static PrometheusConfig providePrometheusConfig(Configuration configuration) {
     static ConsumerConfig provideConsumerConfig(Configuration configuration) {
         return configuration.getConfigData(ConsumerConfig.class);
     }
+
+    /**
+     * Provides a mediator configuration singleton using the configuration.
+     *
+     * @param configuration is the configuration singleton
+     * @return a mediator configuration singleton
+     */
+    @Singleton
+    @Provides
+    static MediatorConfig provideMediatorConfig(Configuration configuration) {
+        return configuration.getConfigData(MediatorConfig.class);
+    }
+
+    /**
+     * Provides a notifier configuration singleton using the configuration.
+     *
+     * @param configuration is the configuration singleton
+     * @return a notifier configuration singleton
+     */
+    @Singleton
+    @Provides
+    static NotifierConfig provideNotifierConfig(Configuration configuration) {
+        return configuration.getConfigData(NotifierConfig.class);
+    }
 }

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

@@ -26,4 +26,21 @@
  *     timed out and will be disconnected
  */
 @ConfigData("consumer")
-public record ConsumerConfig(@ConfigProperty(defaultValue = "1500") long timeoutThresholdMillis) {}
+public record ConsumerConfig(@ConfigProperty(defaultValue = "1500") long timeoutThresholdMillis) {
+    private static final System.Logger LOGGER = System.getLogger(ConsumerConfig.class.getName());
+
+    /**
+     * Validate the configuration.
+     *
+     * @throws IllegalArgumentException if the configuration is invalid
+     */
+    public ConsumerConfig {
+        if (timeoutThresholdMillis <= 0) {
+            throw new IllegalArgumentException("Timeout threshold must be greater than 0");
+        }
+
+        LOGGER.log(
+                System.Logger.Level.INFO,
+                "Consumer configuration timeoutThresholdMillis: " + timeoutThresholdMillis);
+    }
+}