feat: stale event detector

platform-sdk/swirlds-common/src/testFixtures/java/com/swirlds/common/test/fixtures/Randotron.java

@@ -70,7 +70,7 @@ private Randotron(final long seed) {
      * @return a random string
      */
     @NonNull
-    public String randomString(final int length) {
+    public String nextString(final int length) {
         final int LEFT_LIMIT = 48; // numeral '0'
         final int RIGHT_LIMIT = 122; // letter 'z'
 
@@ -87,7 +87,7 @@ public String randomString(final int length) {
      * @return a random IP address
      */
     @NonNull
-    public String randomIp() {
+    public String nextIp() {
         return this.nextInt(256) + "." + this.nextInt(256) + "." + this.nextInt(256) + "." + this.nextInt(256);
     }
 
@@ -97,7 +97,7 @@ public String randomIp() {
      * @param maxValue the upper bound, the returned value will be smaller than this
      * @return the random long
      */
-    public long randomPositiveLong(final long maxValue) {
+    public long nextPositiveLong(final long maxValue) {
         return this.longs(1, 1, maxValue).findFirst().orElseThrow();
     }
 
@@ -106,8 +106,27 @@ public long randomPositiveLong(final long maxValue) {
      *
      * @return the random long
      */
-    public long randomPositiveLong() {
-        return randomPositiveLong(Long.MAX_VALUE);
+    public long nextPositiveLong() {
+        return nextPositiveLong(Long.MAX_VALUE);
+    }
+
+    /**
+     * Generates a random positive int that is smaller than the supplied value
+     *
+     * @param maxValue the upper bound, the returned value will be smaller than this
+     * @return the random int
+     */
+    public int nextPositiveInt(final int maxValue) {
+        return this.ints(1, 1, maxValue).findFirst().orElseThrow();
+    }
+
+    /**
+     * Generates a random positive int
+     *
+     * @return the random int
+     */
+    public int nextPositiveInt() {
+        return nextPositiveInt(Integer.MAX_VALUE);
     }
 
     /**
@@ -116,8 +135,8 @@ public long randomPositiveLong() {
      * @return a random hash
      */
     @NonNull
-    public Hash randomHash() {
-        return new Hash(randomByteArray(DigestType.SHA_384.digestLength()), DigestType.SHA_384);
+    public Hash nextHash() {
+        return new Hash(nextByteArray(DigestType.SHA_384.digestLength()), DigestType.SHA_384);
     }
 
     /**
@@ -126,8 +145,8 @@ public Hash randomHash() {
      * @return random Bytes
      */
     @NonNull
-    public Bytes randomHashBytes() {
-        return Bytes.wrap(randomByteArray(DigestType.SHA_384.digestLength()));
+    public Bytes nextHashBytes() {
+        return Bytes.wrap(nextByteArray(DigestType.SHA_384.digestLength()));
     }
 
     /**
@@ -136,8 +155,8 @@ public Bytes randomHashBytes() {
      * @return a random signature
      */
     @NonNull
-    public Signature randomSignature() {
-        return new Signature(SignatureType.RSA, randomByteArray(SignatureType.RSA.signatureLength()));
+    public Signature nextSignature() {
+        return new Signature(SignatureType.RSA, nextByteArray(SignatureType.RSA.signatureLength()));
     }
 
     /**
@@ -146,8 +165,8 @@ public Signature randomSignature() {
      * @return random signature bytes
      */
     @NonNull
-    public Bytes randomSignatureBytes() {
-        return Bytes.wrap(randomByteArray(SignatureType.RSA.signatureLength()));
+    public Bytes nextSignatureBytes() {
+        return Bytes.wrap(nextByteArray(SignatureType.RSA.signatureLength()));
     }
 
     /**
@@ -157,7 +176,7 @@ public Bytes randomSignatureBytes() {
      * @return a random byte array
      */
     @NonNull
-    public byte[] randomByteArray(final int size) {
+    public byte[] nextByteArray(final int size) {
         final byte[] bytes = new byte[size];
         this.nextBytes(bytes);
         return bytes;
@@ -169,8 +188,8 @@ public byte[] randomByteArray(final int size) {
      * @return a random instant
      */
     @NonNull
-    public Instant randomInstant() {
-        return Instant.ofEpochMilli(randomPositiveLong(2000000000000L));
+    public Instant nextInstant() {
+        return Instant.ofEpochMilli(nextPositiveLong(2000000000000L));
     }
 
     /**
@@ -179,7 +198,7 @@ public Instant randomInstant() {
      * @param trueProbability the probability of the boolean being true
      * @return a random boolean
      */
-    public boolean randomBooleanWithProbability(final double trueProbability) {
+    public boolean nextBoolean(final double trueProbability) {
         if (trueProbability < 0 || trueProbability > 1) {
             throw new IllegalArgumentException("Probability must be between 0 and 1");
         }

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,35 @@
+/*
+ * 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 on preconsensus events in topological order
+ * @param snapshotOverrideConsumer  a consumer that will be called when the current consensus snapshot is overridden
+ *                                  (i.e. at reconnect/restart boundaries)
+ * @param staleEventConsumer        a consumer that will be called when a stale self 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,23 @@ 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
+     */
+    @NonNull
+    public PlatformBuilder withStaleEventCallback(@NonNull final Consumer<GossipEvent> staleEventConsumer) {
+        throwIfAlreadyUsed();
+        this.staleEventConsumer = Objects.requireNonNull(staleEventConsumer);
+        return this;
+    }
+
     /**
      * Build the configuration for the node.
      *
@@ -505,6 +523,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 +536,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<>(),