feat: rewire and remove PlatformSchedulers.java (#13625)

Signed-off-by: Cody Littley <[email protected]>

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/SwirldsPlatform.java

@@ -41,8 +41,6 @@
 import com.swirlds.platform.components.DefaultSavedStateController;
 import com.swirlds.platform.components.EventWindowManager;
 import com.swirlds.platform.components.SavedStateController;
-import com.swirlds.platform.components.appcomm.DefaultLatestCompleteStateNotifier;
-import com.swirlds.platform.components.appcomm.LatestCompleteStateNotifier;
 import com.swirlds.platform.config.StateConfig;
 import com.swirlds.platform.consensus.EventWindow;
 import com.swirlds.platform.crypto.KeysAndCerts;
@@ -232,8 +230,6 @@ public SwirldsPlatform(@NonNull final PlatformComponentBuilder builder) {
         final StateSignatureCollector stateSignatureCollector =
                 new DefaultStateSignatureCollector(platformContext, signedStateMetrics);
 
-        final LatestCompleteStateNotifier latestCompleteStateNotifier = new DefaultLatestCompleteStateNotifier();
-
         blocks.statusActionSubmitterReference()
                 .set(x -> platformWiring.getStatusActionSubmitter().submitStatusAction(x));
 
@@ -285,7 +281,6 @@ public SwirldsPlatform(@NonNull final PlatformComponentBuilder builder) {
                 stateSignatureCollector,
                 eventWindowManager,
                 birthRoundMigrationShim,
-                latestCompleteStateNotifier,
                 latestImmutableStateNexus,
                 latestCompleteStateNexus,
                 savedStateController,

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/builder/PlatformComponentBuilder.java

@@ -24,6 +24,8 @@
 import com.swirlds.common.merkle.utility.SerializableLong;
 import com.swirlds.common.threading.manager.AdHocThreadManager;
 import com.swirlds.platform.SwirldsPlatform;
+import com.swirlds.platform.components.appcomm.DefaultLatestCompleteStateNotifier;
+import com.swirlds.platform.components.appcomm.LatestCompleteStateNotifier;
 import com.swirlds.platform.components.consensus.ConsensusEngine;
 import com.swirlds.platform.components.consensus.DefaultConsensusEngine;
 import com.swirlds.platform.config.StateConfig;
@@ -152,6 +154,7 @@ public class PlatformComponentBuilder {
     private BranchReporter branchReporter;
     private StateSigner stateSigner;
     private TransactionHandler transactionHandler;
+    private LatestCompleteStateNotifier latestCompleteStateNotifier;
 
     private boolean metricsDocumentationEnabled = true;
 
@@ -1269,4 +1272,37 @@ public TransactionHandler buildTransactionHandler() {
         }
         return transactionHandler;
     }
+
+    /**
+     * Provide a latest complete state notifier in place of the platform's default latest complete state notifier.
+     *
+     * @param latestCompleteStateNotifier the latest complete state notifier to use
+     * @return this builder
+     */
+    @NonNull
+    public PlatformComponentBuilder withLatestCompleteStateNotifier(
+            @NonNull final LatestCompleteStateNotifier latestCompleteStateNotifier) {
+        throwIfAlreadyUsed();
+        if (this.latestCompleteStateNotifier != null) {
+            throw new IllegalStateException("Latest complete state notifier has already been set");
+        }
+        this.latestCompleteStateNotifier = Objects.requireNonNull(latestCompleteStateNotifier);
+        return this;
+    }
+
+    /**
+     * Build the latest complete state notifier if it has not yet been built. If one has been provided via
+     * {@link #withLatestCompleteStateNotifier(LatestCompleteStateNotifier)}, that notifier will be used. If this method
+     * is called more than once, only the first call will build the latest complete state notifier. Otherwise, the
+     * default notifier will be created and returned.
+     *
+     * @return the latest complete state notifier
+     */
+    @NonNull
+    public LatestCompleteStateNotifier buildLatestCompleteStateNotifier() {
+        if (latestCompleteStateNotifier == null) {
+            latestCompleteStateNotifier = new DefaultLatestCompleteStateNotifier();
+        }
+        return latestCompleteStateNotifier;
+    }
 }

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/state/nexus/SignedStateNexus.java

@@ -17,6 +17,7 @@
 package com.swirlds.platform.state.nexus;
 
 import com.swirlds.common.utility.Clearable;
+import com.swirlds.common.wiring.component.SchedulerLabel;
 import com.swirlds.platform.consensus.ConsensusConstants;
 import com.swirlds.platform.state.signed.ReservedSignedState;
 import edu.umd.cs.findbugs.annotations.NonNull;
@@ -25,6 +26,7 @@
 /**
  * A thread-safe container that also manages reservations for a single signed state.
  */
+@SchedulerLabel("LatestImmutableStateNexus")
 public interface SignedStateNexus extends Clearable {
     /**
      * Returns the current signed state and reserves it. If the current signed state is null, or cannot be reserved,

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/wiring/PlatformSchedulersConfig.java

@@ -24,43 +24,43 @@
 /**
  * Contains configuration values for the platform schedulers.
  *
- * @param eventHasherUnhandledCapacity           number of unhandled tasks allowed in the event hasher scheduler
- * @param internalEventValidator                 configuration for the internal event validator scheduler
- * @param eventDeduplicator                      configuration for the event deduplicator scheduler
- * @param eventSignatureValidator                configuration for the event signature validator scheduler
- * @param orphanBuffer                           configuration for the orphan buffer scheduler
- * @param consensusEngine                        configuration for the consensus engine scheduler
- * @param eventCreationManager                   configuration for the event creation manager scheduler
- * @param selfEventSigner                        configuration for the self event signer scheduler
- * @param stateSigner                            configuration for the state signer scheduler
- * @param pcesWriter                             configuration for the preconsensus event writer scheduler
- * @param pcesSequencer                          configuration for the preconsensus event sequencer scheduler
- * @param applicationTransactionPrehandler       configuration for the application transaction prehandler scheduler
- * @param stateSignatureCollector                configuration for the state signature collector scheduler
- * @param transactionHandler                     configuration for the transaction handler scheduler
- * @param issDetector                            configuration for the ISS detector scheduler
- * @param issHandler                             configuration for the ISS handler scheduler
- * @param hashLogger                             configuration for the hash logger scheduler
- * @param completeStateNotifierUnhandledCapacity number of unhandled tasks allowed for the state completion notifier
- * @param stateHasher                            configuration for the state hasher scheduler
- * @param stateGarbageCollector                  configuration for the state garbage collector scheduler
- * @param stateGarbageCollectorHeartbeatPeriod   the frequency that heartbeats should be sent to the state garbage
- *                                               collector
- * @param platformPublisher                      configuration for the platform publisher scheduler
- * @param consensusEventStream                   configuration for the consensus event stream scheduler
- * @param roundDurabilityBuffer                  configuration for the round durability buffer scheduler
- * @param signedStateSentinel                    configuration for the signed state sentinel scheduler
- * @param signedStateSentinelHeartbeatPeriod     the frequency that heartbeats should be sent to the signed state
- *                                               sentinel
- * @param statusStateMachine                     configuration for the status state machine scheduler
- * @param staleEventDetector                     configuration for the stale event detector scheduler
- * @param transactionResubmitter                 configuration for the transaction resubmitter scheduler
- * @param transactionPool                        configuration for the transaction pool scheduler
- * @param gossip                                 configuration for the gossip scheduler
- * @param eventHasher                            configuration for the event hasher scheduler
- * @param postHashCollector                      configuration for the post hash collector scheduler
- * @param branchDetector                         configuration for the branch detector scheduler
- * @param branchReporter                         configuration for the branch reporter scheduler
+ * @param eventHasherUnhandledCapacity         number of unhandled tasks allowed in the event hasher scheduler
+ * @param internalEventValidator               configuration for the internal event validator scheduler
+ * @param eventDeduplicator                    configuration for the event deduplicator scheduler
+ * @param eventSignatureValidator              configuration for the event signature validator scheduler
+ * @param orphanBuffer                         configuration for the orphan buffer scheduler
+ * @param consensusEngine                      configuration for the consensus engine scheduler
+ * @param eventCreationManager                 configuration for the event creation manager scheduler
+ * @param selfEventSigner                      configuration for the self event signer scheduler
+ * @param stateSigner                          configuration for the state signer scheduler
+ * @param pcesWriter                           configuration for the preconsensus event writer scheduler
+ * @param pcesSequencer                        configuration for the preconsensus event sequencer scheduler
+ * @param applicationTransactionPrehandler     configuration for the application transaction prehandler scheduler
+ * @param stateSignatureCollector              configuration for the state signature collector scheduler
+ * @param transactionHandler                   configuration for the transaction handler scheduler
+ * @param issDetector                          configuration for the ISS detector scheduler
+ * @param issHandler                           configuration for the ISS handler scheduler
+ * @param hashLogger                           configuration for the hash logger scheduler
+ * @param latestCompleteStateNotifier          configuration for the latest complete state notifier scheduler
+ * @param stateHasher                          configuration for the state hasher scheduler
+ * @param stateGarbageCollector                configuration for the state garbage collector scheduler
+ * @param stateGarbageCollectorHeartbeatPeriod the frequency that heartbeats should be sent to the state garbage
+ *                                             collector
+ * @param platformPublisher                    configuration for the platform publisher scheduler
+ * @param consensusEventStream                 configuration for the consensus event stream scheduler
+ * @param roundDurabilityBuffer                configuration for the round durability buffer scheduler
+ * @param signedStateSentinel                  configuration for the signed state sentinel scheduler
+ * @param signedStateSentinelHeartbeatPeriod   the frequency that heartbeats should be sent to the signed state
+ *                                             sentinel
+ * @param statusStateMachine                   configuration for the status state machine scheduler
+ * @param staleEventDetector                   configuration for the stale event detector scheduler
+ * @param transactionResubmitter               configuration for the transaction resubmitter scheduler
+ * @param transactionPool                      configuration for the transaction pool scheduler
+ * @param gossip                               configuration for the gossip scheduler
+ * @param eventHasher                          configuration for the event hasher scheduler
+ * @param postHashCollector                    configuration for the post hash collector scheduler
+ * @param branchDetector                       configuration for the branch detector scheduler
+ * @param branchReporter                       configuration for the branch reporter scheduler
  */
 @ConfigData("platformSchedulers")
 public record PlatformSchedulersConfig(
@@ -100,7 +100,8 @@ public record PlatformSchedulersConfig(
         @ConfigProperty(defaultValue = "DIRECT") TaskSchedulerConfiguration issHandler,
         @ConfigProperty(defaultValue = "SEQUENTIAL CAPACITY(100) UNHANDLED_TASK_METRIC")
                 TaskSchedulerConfiguration hashLogger,
-        @ConfigProperty(defaultValue = "1000") int completeStateNotifierUnhandledCapacity,
+        @ConfigProperty(defaultValue = "SEQUENTIAL CAPACITY(5) UNHANDLED_TASK_METRIC")
+                TaskSchedulerConfiguration latestCompleteStateNotifier,
         @ConfigProperty(
                         defaultValue =
                                 "SEQUENTIAL_THREAD CAPACITY(5) FLUSHABLE UNHANDLED_TASK_METRIC BUSY_FRACTION_METRIC")