feat: stale event detector

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

@@ -23,6 +23,8 @@
 import com.swirlds.platform.state.signed.ReservedSignedState;
 import com.swirlds.platform.system.status.PlatformStatus;
 import com.swirlds.platform.system.status.PlatformStatusNexus;
+import com.swirlds.platform.system.transaction.ConsensusTransactionImpl;
+import com.swirlds.platform.system.transaction.StateSignatureTransaction;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
 import java.util.Objects;
@@ -59,7 +61,7 @@ public StateSigner(@NonNull final PlatformSigner signer, @NonNull final Platform
      * @param reservedSignedState the state to sign
      * @return a {@link StateSignaturePayload} containing the signature, or null if the state should not be signed
      */
-    public @Nullable StateSignaturePayload signState(@NonNull final ReservedSignedState reservedSignedState) {
+    public @Nullable ConsensusTransactionImpl signState(@NonNull final ReservedSignedState reservedSignedState) {
         try (reservedSignedState) {
             if (statusNexus.getCurrentStatus() == PlatformStatus.REPLAYING_EVENTS) {
                 // the only time we don't want to submit signatures is during PCES replay
@@ -71,11 +73,12 @@ public StateSigner(@NonNull final PlatformSigner signer, @NonNull final Platform
             final Bytes signature = signer.signImmutable(stateHash);
             Objects.requireNonNull(signature);
 
-            return StateSignaturePayload.newBuilder()
+            final StateSignaturePayload payload = StateSignaturePayload.newBuilder()
                     .round(reservedSignedState.get().getRound())
                     .signature(signature)
                     .hash(stateHash.getBytes())
                     .build();
+            return new StateSignatureTransaction(payload);
         }
     }
 }

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

@@ -67,7 +67,6 @@
 import com.swirlds.platform.components.appcomm.LatestCompleteStateNotifier;
 import com.swirlds.platform.config.StateConfig;
 import com.swirlds.platform.config.TransactionConfig;
-import com.swirlds.platform.consensus.ConsensusSnapshot;
 import com.swirlds.platform.consensus.EventWindow;
 import com.swirlds.platform.crypto.KeysAndCerts;
 import com.swirlds.platform.crypto.PlatformSigner;
@@ -80,7 +79,6 @@
 import com.swirlds.platform.event.validation.AddressBookUpdate;
 import com.swirlds.platform.eventhandling.ConsensusRoundHandler;
 import com.swirlds.platform.eventhandling.EventConfig;
-import com.swirlds.platform.eventhandling.TransactionPool;
 import com.swirlds.platform.gossip.SyncGossip;
 import com.swirlds.platform.gossip.shadowgraph.Shadowgraph;
 import com.swirlds.platform.listeners.ReconnectCompleteNotification;
@@ -89,6 +87,7 @@
 import com.swirlds.platform.metrics.SwirldStateMetrics;
 import com.swirlds.platform.metrics.SyncMetrics;
 import com.swirlds.platform.metrics.TransactionMetrics;
+import com.swirlds.platform.pool.TransactionPoolNexus;
 import com.swirlds.platform.publisher.DefaultPlatformPublisher;
 import com.swirlds.platform.publisher.PlatformPublisher;
 import com.swirlds.platform.recovery.EmergencyRecoveryManager;
@@ -326,10 +325,7 @@ public SwirldsPlatform(@NonNull final PlatformComponentBuilder builder) {
             emergencyState.setState(initialState.reserve("emergency state nexus"));
         }
 
-        final Consumer<GossipEvent> preconsensusEventConsumer = blocks.preconsensusEventConsumer();
-        final Consumer<ConsensusSnapshot> snapshotOverrideConsumer = blocks.snapshotOverrideConsumer();
-        platformWiring = thingsToStart.add(new PlatformWiring(
-                platformContext, preconsensusEventConsumer != null, snapshotOverrideConsumer != null));
+        platformWiring = thingsToStart.add(new PlatformWiring(platformContext, blocks.applicationCallbacks()));
 
         thingsToStart.add(Objects.requireNonNull(recycleBin));
 
@@ -467,8 +463,6 @@ public SwirldsPlatform(@NonNull final PlatformComponentBuilder builder) {
         blocks.intakeQueueSizeSupplierSupplier().set(intakeQueueSizeSupplier);
         blocks.isInFreezePeriodReference().set(swirldStateManager::isInFreezePeriod);
 
-        platformWiring.wireExternalComponents(blocks.transactionPool());
-
         final IssHandler issHandler =
                 new IssHandler(stateConfig, this::haltRequested, SystemExitUtils::handleFatalError, issScratchpad);
 
@@ -482,8 +476,7 @@ public SwirldsPlatform(@NonNull final PlatformComponentBuilder builder) {
 
         final AppNotifier appNotifier = new DefaultAppNotifier(notificationEngine);
 
-        final PlatformPublisher publisher =
-                new DefaultPlatformPublisher(preconsensusEventConsumer, snapshotOverrideConsumer);
+        final PlatformPublisher publisher = new DefaultPlatformPublisher(blocks.applicationCallbacks());
 
         platformWiring.bind(
                 builder,
@@ -528,11 +521,11 @@ public SwirldsPlatform(@NonNull final PlatformComponentBuilder builder) {
             platformWiring.getPcesMinimumGenerationToStoreInput().inject(minimumGenerationNonAncientForOldestState);
         }
 
-        final TransactionPool transactionPool = blocks.transactionPool();
+        final TransactionPoolNexus transactionPoolNexus = blocks.transactionPoolNexus();
         transactionSubmitter = new SwirldTransactionSubmitter(
                 statusNexus,
                 transactionConfig,
-                transaction -> transactionPool.submitTransaction(transaction, false),
+                transaction -> transactionPoolNexus.submitTransaction(transaction, false),
                 new TransactionMetrics(metrics));
 
         final boolean startedFromGenesis = initialState.isGenesisState();
@@ -578,6 +571,7 @@ public SwirldsPlatform(@NonNull final PlatformComponentBuilder builder) {
         if (startedFromGenesis) {
             initialAncientThreshold = 0;
             startingRound = 0;
+            platformWiring.updateEventWindow(EventWindow.getGenesisEventWindow(ancientMode));
         } else {
             initialAncientThreshold = initialState.getState().getPlatformState().getAncientThreshold();
             startingRound = initialState.getRound();
@@ -614,10 +608,7 @@ public SwirldsPlatform(@NonNull final PlatformComponentBuilder builder) {
 
         clearAllPipelines = new LoggingClearables(
                 RECONNECT.getMarker(),
-                List.of(
-                        Pair.of(platformWiring, "platformWiring"),
-                        Pair.of(shadowGraph, "shadowGraph"),
-                        Pair.of(transactionPool, "transactionPool")));
+                List.of(Pair.of(platformWiring, "platformWiring"), Pair.of(shadowGraph, "shadowGraph")));
 
         blocks.latestImmutableStateProviderReference().set(latestImmutableStateNexus::getState);
     }

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright (C) 2024 Hedera Hashgraph, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.swirlds.platform.builder;
+
+import com.swirlds.platform.consensus.ConsensusSnapshot;
+import com.swirlds.platform.event.GossipEvent;
+import edu.umd.cs.findbugs.annotations.Nullable;
+import java.util.function.Consumer;
+
+/**
+ * A collection of callbacks that the application can provide to the platform to be notified of certain events.
+ *
+ * @param preconsensusEventConsumer a consumer that will be called when a preconsensus event is created
+ * @param snapshotOverrideConsumer  a consumer that will be called when a consensus snapshot is created
+ * @param staleEventConsumer        a consumer that will be called when a stale event is detected
+ */
+public record ApplicationCallbacks(
+        @Nullable Consumer<GossipEvent> preconsensusEventConsumer,
+        @Nullable Consumer<ConsensusSnapshot> snapshotOverrideConsumer,
+        @Nullable Consumer<GossipEvent> staleEventConsumer) {}

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

@@ -58,11 +58,11 @@
 import com.swirlds.platform.event.preconsensus.PcesFileReader;
 import com.swirlds.platform.event.preconsensus.PcesFileTracker;
 import com.swirlds.platform.eventhandling.EventConfig;
-import com.swirlds.platform.eventhandling.TransactionPool;
 import com.swirlds.platform.gossip.DefaultIntakeEventCounter;
 import com.swirlds.platform.gossip.IntakeEventCounter;
 import com.swirlds.platform.gossip.NoOpIntakeEventCounter;
 import com.swirlds.platform.gossip.sync.config.SyncConfig;
+import com.swirlds.platform.pool.TransactionPoolNexus;
 import com.swirlds.platform.recovery.EmergencyRecoveryManager;
 import com.swirlds.platform.state.State;
 import com.swirlds.platform.state.address.AddressBookInitializer;
@@ -114,6 +114,7 @@ public final class PlatformBuilder {
 
     private Consumer<GossipEvent> preconsensusEventConsumer;
     private Consumer<ConsensusSnapshot> snapshotOverrideConsumer;
+    private Consumer<GossipEvent> staleEventConsumer;
 
     /**
      * False if this builder has not yet been used to build a platform (or platform component builder), true if it has.
@@ -310,6 +311,22 @@ public PlatformBuilder withConsensusSnapshotOverrideCallback(
         return this;
     }
 
+    /**
+     * Register a callback that is called when a stale self event is detected (i.e. an event that will never reach
+     * consensus). Depending on the use case, it may be a good idea to resubmit the transactions in the stale event.
+     * <p>
+     * Stale event detection is guaranteed to catch all stale self events as long as the node remains online. However,
+     * if the node restarts or reconnects, any event that went stale "in the gap" may not be detected.
+     *
+     * @param staleEventConsumer the callback to register
+     * @return this
+     */
+    public PlatformBuilder withStaleEventCallback(@NonNull final Consumer<GossipEvent> staleEventConsumer) {
+        throwIfAlreadyUsed();
+        this.staleEventConsumer = Objects.requireNonNull(staleEventConsumer);
+        return this;
+    }
+
     /**
      * Build the configuration for the node.
      *
@@ -505,6 +522,9 @@ public PlatformComponentBuilder buildComponentBuilder() {
             throw new UncheckedIOException(e);
         }
 
+        final ApplicationCallbacks callbacks =
+                new ApplicationCallbacks(preconsensusEventConsumer, snapshotOverrideConsumer, staleEventConsumer);
+
         final PlatformBuildingBlocks buildingBlocks = new PlatformBuildingBlocks(
                 platformContext,
                 keysAndCerts.get(selfId),
@@ -515,11 +535,10 @@ public PlatformComponentBuilder buildComponentBuilder() {
                 softwareVersion,
                 initialState,
                 emergencyRecoveryManager,
-                preconsensusEventConsumer,
-                snapshotOverrideConsumer,
+                callbacks,
                 intakeEventCounter,
                 new RandomBuilder(),
-                new TransactionPool(platformContext),
+                new TransactionPoolNexus(platformContext),
                 new AtomicReference<>(),
                 new AtomicReference<>(),
                 new AtomicReference<>(),

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

@@ -19,21 +19,17 @@
 import com.swirlds.common.context.PlatformContext;
 import com.swirlds.common.io.utility.RecycleBin;
 import com.swirlds.common.platform.NodeId;
-import com.swirlds.platform.consensus.ConsensusSnapshot;
 import com.swirlds.platform.crypto.KeysAndCerts;
-import com.swirlds.platform.event.GossipEvent;
 import com.swirlds.platform.event.preconsensus.PcesFileTracker;
-import com.swirlds.platform.eventhandling.TransactionPool;
 import com.swirlds.platform.gossip.IntakeEventCounter;
+import com.swirlds.platform.pool.TransactionPoolNexus;
 import com.swirlds.platform.recovery.EmergencyRecoveryManager;
 import com.swirlds.platform.state.signed.ReservedSignedState;
 import com.swirlds.platform.system.SoftwareVersion;
 import com.swirlds.platform.util.RandomBuilder;
 import edu.umd.cs.findbugs.annotations.NonNull;
-import edu.umd.cs.findbugs.annotations.Nullable;
 import java.time.Instant;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.LongSupplier;
 import java.util.function.Predicate;
@@ -52,14 +48,11 @@
  * @param appVersion                            the current version of the running application
  * @param initialState                          the initial state of the platform
  * @param emergencyRecoveryManager              used in emergency recovery.
- * @param preconsensusEventConsumer             the consumer for preconsensus events, null if publishing this data has
- *                                              not been enabled
- * @param snapshotOverrideConsumer              the consumer for snapshot overrides, null if publishing this data has
- *                                              not been enabled
+ * @param applicationCallbacks                  the callbacks that the platform will call when certain events happen
  * @param intakeEventCounter                    counts events that have been received by gossip but not yet inserted
  *                                              into gossip event storage, per peer
  * @param randomBuilder                         a builder for creating random number generators
- * @param transactionPool                       provides transactions to be added to new events
+ * @param transactionPoolNexus                  provides transactions to be added to new events
  * @param intakeQueueSizeSupplierSupplier       supplies a method which supplies the size of the intake queue. This hack
  *                                              is required due to the lack of a platform health monitor.
  * @param isInFreezePeriodReference             a reference to a predicate that determines if a timestamp is in the
@@ -83,11 +76,10 @@ public record PlatformBuildingBlocks(
         @NonNull SoftwareVersion appVersion,
         @NonNull ReservedSignedState initialState,
         @NonNull EmergencyRecoveryManager emergencyRecoveryManager,
-        @Nullable Consumer<GossipEvent> preconsensusEventConsumer,
-        @Nullable Consumer<ConsensusSnapshot> snapshotOverrideConsumer,
+        @NonNull ApplicationCallbacks applicationCallbacks,
         @NonNull IntakeEventCounter intakeEventCounter,
         @NonNull RandomBuilder randomBuilder,
-        @NonNull TransactionPool transactionPool,
+        @NonNull TransactionPoolNexus transactionPoolNexus,
         @NonNull AtomicReference<LongSupplier> intakeQueueSizeSupplierSupplier,
         @NonNull AtomicReference<Predicate<Instant>> isInFreezePeriodReference,
         @NonNull AtomicReference<Function<String, ReservedSignedState>> latestImmutableStateProviderReference,