feat: disentangle consensus and gossip (#12428)

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

Author: cody-littley

platform-sdk/docs/core/wiring-diagram.svg

@@ -22,7 +22,6 @@
 import com.swirlds.platform.internal.ConsensusRound;
 import com.swirlds.platform.internal.EventImpl;
 import edu.umd.cs.findbugs.annotations.NonNull;
-import edu.umd.cs.findbugs.annotations.Nullable;
 import java.util.List;
 
 /** An interface for classes that calculate consensus of events */
@@ -33,15 +32,15 @@ public interface Consensus extends GraphGenerations, RoundNumberProvider {
      *
      * @param event the event to be added
      * @return A list of consensus rounds, each with a list of consensus events (that can be empty). The rounds are
-     * stored in consensus order (round at index 0 occurs before the round at index 1 in consensus time). Returns null
-     * if no consensus was reached
+     * stored in consensus order (round at index 0 occurs before the round at index 1 in consensus time). Returns an
+     * empty list if no rounds reached consensus.
      */
-    @Nullable
+    @NonNull
     List<ConsensusRound> addEvent(@NonNull EventImpl event);
 
     /**
-     * Load consensus from a snapshot. This will continue consensus from the round of the snapshot
-     * once all the required events are provided. This method is called at restart and reconnect boundaries.
+     * Load consensus from a snapshot. This will continue consensus from the round of the snapshot once all the required
+     * events are provided. This method is called at restart and reconnect boundaries.
      */
     void loadSnapshot(@NonNull ConsensusSnapshot snapshot);
 }

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

@@ -251,23 +251,24 @@ private void reset() {
      * round.
      *
      * @param event the event to be added
-     * @return A list of consensus rounds, or null if no consensus was reached
+     * @return A list of consensus rounds or an empty list if no consensus was reached
      */
+    @NonNull
     @Override
-    public @Nullable List<ConsensusRound> addEvent(@NonNull final EventImpl event) {
+    public List<ConsensusRound> addEvent(@NonNull final EventImpl event) {
         recentEvents.add(event);
-        final List<ConsensusRound> toReturn = new ArrayList<>();
+        final List<ConsensusRound> rounds = new ArrayList<>();
         // set its round to undefined so that it gets calculated
         event.setRoundCreated(ConsensusConstants.ROUND_UNDEFINED);
         checkInitJudges(event);
         ConsensusRound consensusRound = calculateAndVote(event);
 
         while (consensusRound != null) {
-            toReturn.add(consensusRound);
+            rounds.add(consensusRound);
 
             consensusRound = recalculateAndVote();
         }
-        return toReturn.isEmpty() ? null : toReturn;
+        return rounds;
     }
 
     /**

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

@@ -62,15 +62,15 @@
 import com.swirlds.platform.builder.PlatformBuildingBlocks;
 import com.swirlds.platform.builder.PlatformComponentBuilder;
 import com.swirlds.platform.components.AppNotifier;
-import com.swirlds.platform.components.ConsensusEngine;
 import com.swirlds.platform.components.DefaultAppNotifier;
-import com.swirlds.platform.components.DefaultConsensusEngine;
 import com.swirlds.platform.components.DefaultEventWindowManager;
 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.components.consensus.ConsensusEngine;
+import com.swirlds.platform.components.consensus.DefaultConsensusEngine;
 import com.swirlds.platform.config.StateConfig;
 import com.swirlds.platform.config.TransactionConfig;
 import com.swirlds.platform.consensus.ConsensusSnapshot;
@@ -82,6 +82,7 @@
 import com.swirlds.platform.event.EventCounter;
 import com.swirlds.platform.event.GossipEvent;
 import com.swirlds.platform.event.creation.EventCreationManager;
+import com.swirlds.platform.event.linking.GossipLinker;
 import com.swirlds.platform.event.linking.InOrderLinker;
 import com.swirlds.platform.event.orphan.OrphanBuffer;
 import com.swirlds.platform.event.preconsensus.DefaultPcesSequencer;
@@ -109,8 +110,6 @@
 import com.swirlds.platform.listeners.PlatformStatusChangeNotification;
 import com.swirlds.platform.listeners.ReconnectCompleteNotification;
 import com.swirlds.platform.listeners.StateLoadedFromDiskNotification;
-import com.swirlds.platform.metrics.ConsensusMetrics;
-import com.swirlds.platform.metrics.ConsensusMetricsImpl;
 import com.swirlds.platform.metrics.RuntimeMetrics;
 import com.swirlds.platform.metrics.SwirldStateMetrics;
 import com.swirlds.platform.metrics.SyncMetrics;
@@ -174,7 +173,6 @@
 import java.util.Objects;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.LongSupplier;
 import org.apache.logging.log4j.LogManager;
@@ -197,12 +195,6 @@ public class SwirldsPlatform implements Platform {
      */
     private final Shadowgraph shadowGraph;
 
-    /**
-     * the object used to calculate consensus. it is volatile because the whole object is replaced when reading a state
-     * from disk or getting it through reconnect
-     */
-    private final AtomicReference<Consensus> consensusRef = new AtomicReference<>();
-
     /**
      * the current nodes in the network and their information
      */
@@ -394,12 +386,10 @@ public SwirldsPlatform(@NonNull final PlatformComponentBuilder builder) {
 
         registerAddressBookMetrics(metrics, currentAddressBook, selfId);
 
-        final ConsensusMetrics consensusMetrics = new ConsensusMetricsImpl(selfId, metrics);
-
         final SyncMetrics syncMetrics = new SyncMetrics(metrics);
         RuntimeMetrics.setup(metrics);
 
-        shadowGraph = new Shadowgraph(platformContext, currentAddressBook);
+        shadowGraph = new Shadowgraph(platformContext, currentAddressBook, blocks.intakeEventCounter());
 
         final EventConfig eventConfig = platformContext.getConfiguration().getConfigData(EventConfig.class);
 
@@ -591,10 +581,11 @@ public SwirldsPlatform(@NonNull final PlatformComponentBuilder builder) {
                 initialState.getState().getPlatformState().getPreviousAddressBook(),
                 currentAddressBook,
                 blocks.intakeEventCounter());
+
         final OrphanBuffer orphanBuffer = new OrphanBuffer(platformContext, blocks.intakeEventCounter());
-        final InOrderLinker inOrderLinker = new InOrderLinker(platformContext, time, blocks.intakeEventCounter());
-        final ConsensusEngine consensusEngine = new DefaultConsensusEngine(
-                platformContext, selfId, consensusRef::get, shadowGraph, blocks.intakeEventCounter(), e -> {});
+        final InOrderLinker inOrderLinker = new GossipLinker(platformContext, blocks.intakeEventCounter());
+
+        final ConsensusEngine consensusEngine = new DefaultConsensusEngine(platformContext, currentAddressBook, selfId);
 
         final LongSupplier intakeQueueSizeSupplier =
                 oldStyleIntakeQueue == null ? platformWiring.getIntakeQueueSizeSupplier() : oldStyleIntakeQueue::size;
@@ -717,8 +708,6 @@ public SwirldsPlatform(@NonNull final PlatformComponentBuilder builder) {
                 blocks.intakeEventCounter(),
                 () -> emergencyState.getState("emergency reconnect")) {};
 
-        consensusRef.set(new ConsensusImpl(platformContext, consensusMetrics, getAddressBook()));
-
         latestImmutableStateNexus.setState(initialState.reserve("set latest immutable to initial state"));
 
         if (startedFromGenesis) {

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

@@ -34,39 +34,39 @@ public interface AppNotifier {
      *
      * @param notification the notification
      */
-    @InputWireLabel("StateWriteToDiskCompleteNotification")
+    @InputWireLabel("state written notification")
     void sendStateWrittenToDiskNotification(@NonNull final StateWriteToDiskCompleteNotification notification);
 
     /**
      * Send a notification to the app that the state has been loaded from disk.
      *
      * @param notification the notification
      */
-    @InputWireLabel("StateLoadedFromDiskNotification")
+    @InputWireLabel("state loaded notification")
     void sendStateLoadedFromDiskNotification(@NonNull final StateLoadedFromDiskNotification notification);
 
     /**
      * Send a notification to the app that a reconnect has completed.
      *
      * @param notification the notification
      */
-    @InputWireLabel("ReconnectCompleteNotification")
+    @InputWireLabel("reconnect notification")
     void sendReconnectCompleteNotification(@NonNull final ReconnectCompleteNotification notification);
 
     /**
      * Send a notification to the app that the platform status has changed.
      *
      * @param notification the notification
      */
-    @InputWireLabel("PlatformStatusChangeNotification")
+    @InputWireLabel("platform status notification")
     void sendPlatformStatusChangeNotification(@NonNull final PlatformStatusChangeNotification notification);
 
     /**
      * Send a notification to the app with the latest complete state.
      *
      * @param notificationWithCleanup the notification, with required cleanup
      */
-    @InputWireLabel("CompleteStateNotificationWithCleanup")
+    @InputWireLabel("complete state notification")
     void sendLatestCompleteStateNotification(
             @NonNull final CompleteStateNotificationWithCleanup notificationWithCleanup);
 
@@ -75,6 +75,6 @@ void sendLatestCompleteStateNotification(
      *
      * @param notification the notification
      */
-    @InputWireLabel("IssNotification")
+    @InputWireLabel("Iss Notification")
     void sendIssNotification(@NonNull final IssNotification notification);
 }

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/components/AppNotifier.java

@@ -14,11 +14,12 @@
  * limitations under the License.
  */
 
-package com.swirlds.platform.components;
+package com.swirlds.platform.components.consensus;
 
 import com.swirlds.common.wiring.component.InputWireLabel;
 import com.swirlds.platform.Consensus;
 import com.swirlds.platform.consensus.ConsensusSnapshot;
+import com.swirlds.platform.event.GossipEvent;
 import com.swirlds.platform.internal.ConsensusRound;
 import com.swirlds.platform.internal.EventImpl;
 import edu.umd.cs.findbugs.annotations.NonNull;
@@ -35,8 +36,8 @@ public interface ConsensusEngine {
      * @return a list of rounds that came to consensus as a result of adding the event
      */
     @NonNull
-    @InputWireLabel("EventImpl")
-    List<ConsensusRound> addEvent(@NonNull EventImpl event);
+    @InputWireLabel("GossipEvent")
+    List<ConsensusRound> addEvent(@NonNull GossipEvent event);
 
     /**
      * Perform an out-of-band snapshot update. This happens at restart/reconnect boundaries.