feat: Add final Consistency functionality to Otter App (#21334)

Signed-off-by: Michael Heinrichs <[email protected]>
Signed-off-by: Kelly Greco <[email protected]>
Co-authored-by: Michael Heinrichs <[email protected]>

Author: poulok

platform-sdk/consensus-otter-tests/src/testFixtures/java/org/hiero/otter/fixtures/TransactionGenerator.java

@@ -5,11 +5,13 @@
  * Interface representing a transaction generator.
  *
  * <p>A {@link TransactionGenerator} generates a steady flow of transaction to all nodes in the
- * network. The generator sends 100 transactions to each node per second, which ensures there is
- * always at least one transaction waiting to be processed by the event creator.
+ * network. The generator sends a constant number of transactions to each node per second, which ensures there is
+ * always at least one transaction waiting to be processed by the event creator. The number of transactions per
+ * second is defined by the {@link #TPS} constant.
  */
 public interface TransactionGenerator {
 
+    /** The number of transactions to generate per second, per node. */
     int TPS = 100;
 
     /**

platform-sdk/consensus-otter-tests/src/testFixtures/java/org/hiero/otter/fixtures/app/OtterApp.java

@@ -6,6 +6,7 @@
 import com.hedera.hapi.node.base.SemanticVersion;
 import com.hedera.hapi.platform.event.StateSignatureTransaction;
 import com.swirlds.common.context.PlatformContext;
+import com.swirlds.config.api.Configuration;
 import com.swirlds.platform.state.ConsensusStateEventHandler;
 import com.swirlds.platform.system.InitTrigger;
 import com.swirlds.platform.system.Platform;
@@ -99,8 +100,8 @@ public void onPreHandle(
         for (final OtterService service : allServices) {
             service.preHandleEvent(event);
         }
-        for (final Iterator<Transaction> transactionIterator = event.transactionIterator();
-                transactionIterator.hasNext(); ) {
+        final Iterator<Transaction> transactionIterator = event.transactionIterator();
+        while (transactionIterator.hasNext()) {
             try {
                 final OtterTransaction transaction = OtterTransaction.parseFrom(
                         transactionIterator.next().getApplicationTransaction().toInputStream());
@@ -125,16 +126,15 @@ public void onHandleConsensusRound(
             @NonNull final OtterAppState state,
             @NonNull final Consumer<ScopedSystemTransaction<StateSignatureTransaction>> callback) {
         for (final OtterService service : allServices) {
-            service.handleRound(state.getWritableStates(service.name()), round);
+            service.onRoundStart(state.getWritableStates(service.name()), round);
         }
 
         for (final ConsensusEvent consensusEvent : round) {
             for (final OtterService service : allServices) {
-                service.handleEvent(state.getWritableStates(service.name()), consensusEvent);
+                service.onEventStart(state.getWritableStates(service.name()), consensusEvent);
             }
-            for (final Iterator<ConsensusTransaction> transactionIterator =
-                            consensusEvent.consensusTransactionIterator();
-                    transactionIterator.hasNext(); ) {
+            final Iterator<ConsensusTransaction> transactionIterator = consensusEvent.consensusTransactionIterator();
+            while (transactionIterator.hasNext()) {
                 try {
                     final OtterTransaction transaction = OtterTransaction.parseFrom(transactionIterator
                             .next()
@@ -151,6 +151,14 @@ public void onHandleConsensusRound(
                             ex);
                 }
             }
+
+            for (final OtterService service : allServices) {
+                service.onEventComplete(consensusEvent);
+            }
+        }
+
+        for (final OtterService service : allServices) {
+            service.onRoundComplete(round);
         }
 
         state.commitState();
@@ -190,12 +198,14 @@ public void onStateInitialized(
             @NonNull final Platform platform,
             @NonNull final InitTrigger trigger,
             @Nullable final SemanticVersion previousVersion) {
+        final Configuration configuration = platform.getContext().getConfiguration();
         if (state.getReadableStates(ConsistencyService.NAME).isEmpty()) {
-            OtterStateInitializer.initOtterAppState(
-                    platform.getContext().getConfiguration(), state, version, appServices);
+            OtterStateInitializer.initOtterAppState(configuration, state, version, appServices);
         }
 
-        consistencyService.initialize();
+        for (final OtterService service : allServices) {
+            service.initialize(trigger, platform.getSelfId(), configuration, state);
+        }
     }
 
     /**
@@ -225,4 +235,10 @@ public void onNewRecoveredState(@NonNull final OtterAppState recoveredState) {
     public void updateSyntheticBottleneck(final long millisToSleepPerRound) {
         this.syntheticBottleneckMillis.set(millisToSleepPerRound);
     }
+
+    public void destroy() {
+        for (final OtterService service : allServices) {
+            service.destroy();
+        }
+    }
 }

platform-sdk/consensus-otter-tests/src/testFixtures/java/org/hiero/otter/fixtures/app/OtterService.java

@@ -3,14 +3,16 @@
 
 import com.hedera.hapi.node.base.SemanticVersion;
 import com.hedera.hapi.platform.event.StateSignatureTransaction;
+import com.swirlds.config.api.Configuration;
+import com.swirlds.platform.system.InitTrigger;
 import com.swirlds.state.lifecycle.Schema;
 import com.swirlds.state.spi.WritableStates;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.util.function.Consumer;
 import org.hiero.consensus.model.event.Event;
 import org.hiero.consensus.model.hashgraph.Round;
+import org.hiero.consensus.model.node.NodeId;
 import org.hiero.consensus.model.transaction.ScopedSystemTransaction;
-import org.hiero.consensus.model.transaction.Transaction;
 
 /**
  * This interface defines a service of the Otter application.
@@ -34,15 +36,48 @@ public interface OtterService {
     @NonNull
     Schema<SemanticVersion> genesisSchema(@NonNull SemanticVersion version);
 
+    /**
+     * Called when the service is initialized. This is called once when the application starts up.
+     *
+     * @param trigger the trigger that caused the initialization
+     * @param selfId the ID of this node
+     * @param configuration the configuration to use
+     * @param state the current state at the time of initialization
+     */
+    default void initialize(
+            @NonNull final InitTrigger trigger,
+            @NonNull final NodeId selfId,
+            @NonNull final Configuration configuration,
+            @NonNull final OtterAppState state) {
+        // Default implementation does nothing
+    }
+
+    /**
+     * Called when the service is being shut down. This is called once when the application is shutting down.
+     */
+    default void destroy() {
+        // Default implementation does nothing
+    }
+
     /**
      * Called when a new round of consensus has been received. The service should only do actions for the whole round in
-     * this method. For actions on individual events, use {@link #handleEvent(WritableStates, Event)}. For actions on
-     * individual transactions, use {@link #handleTransaction(WritableStates, Event, Transaction, Consumer)}.
+     * this method. For actions on individual events, use {@link #onEventStart(WritableStates, Event)}. For actions on
+     * individual transactions, use {@link #handleTransaction(WritableStates, Event, OtterTransaction, Consumer)} .
      *
      * @param writableStates the {@link WritableStates} to use to modify state
      * @param round the round to handle
      */
-    default void handleRound(@NonNull final WritableStates writableStates, @NonNull final Round round) {
+    default void onRoundStart(@NonNull final WritableStates writableStates, @NonNull final Round round) {
+        // Default implementation does nothing
+    }
+
+    /**
+     * Called when a round of consensus has been completely handled. This is called after all events and transactions in
+     * the round have been handled.
+     *
+     * @param round the round that was completed
+     */
+    default void onRoundComplete(@NonNull final Round round) {
         // Default implementation does nothing
     }
 
@@ -54,7 +89,17 @@ default void handleRound(@NonNull final WritableStates writableStates, @NonNull
      * @param writableStates the {@link WritableStates} to use to modify state
      * @param event the event to handle
      */
-    default void handleEvent(@NonNull final WritableStates writableStates, @NonNull final Event event) {
+    default void onEventStart(@NonNull final WritableStates writableStates, @NonNull final Event event) {
+        // Default implementation does nothing
+    }
+
+    /**
+     * Called when an event has been completely handled. This is called after all transactions in the event have been
+     * handled.
+     *
+     * @param event the event that was completed
+     */
+    default void onEventComplete(@NonNull final Event event) {
         // Default implementation does nothing
     }
 

platform-sdk/consensus-otter-tests/src/testFixtures/java/org/hiero/otter/fixtures/app/services/consistency/ConsistencyService.java

@@ -5,29 +5,41 @@
 
 import com.hedera.hapi.node.base.SemanticVersion;
 import com.hedera.hapi.platform.event.StateSignatureTransaction;
+import com.swirlds.common.config.StateCommonConfig;
+import com.swirlds.config.api.Configuration;
+import com.swirlds.platform.system.InitTrigger;
 import com.swirlds.state.lifecycle.Schema;
 import com.swirlds.state.spi.WritableStates;
 import edu.umd.cs.findbugs.annotations.NonNull;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Consumer;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.hiero.consensus.model.event.Event;
+import org.hiero.consensus.model.hashgraph.ConsensusConstants;
 import org.hiero.consensus.model.hashgraph.Round;
+import org.hiero.consensus.model.node.NodeId;
 import org.hiero.consensus.model.transaction.ScopedSystemTransaction;
+import org.hiero.otter.fixtures.app.OtterAppState;
 import org.hiero.otter.fixtures.app.OtterService;
 import org.hiero.otter.fixtures.app.OtterTransaction;
+import org.jetbrains.annotations.NotNull;
 
 /**
  * A service that ensures the consistency of rounds and transactions sent by the platform to the execution layer for
  * handling. It checks these aspects of consistency:
  * <ol>
  *     <li>Consensus rounds increase in number monotonically</li>
  *     <li>Consensus rounds are received only once</li>
- *     <li>Differences in rounds or transactions sent to {@link #recordRound(Round)} on different nodes will cause an ISS</li>
- *     <li>Consensus transactions were previous received in preHandle</li>
- *     <li>After a restart, any rounds that reach consensus in PCES replay exactly match the rounds calculated previously.</li>
+ *     <li>Differences in rounds or transactions recorded in the {@link ConsistencyServiceRoundHistory} on different nodes will cause an ISS</li>
+ *     <li>Transactions are pre-handled only once</li>
+ *     <li>Consensus transactions were previously received in pre-handle</li>
+ *     <li>After a restart, any rounds that reach consensus during PCES replay exactly match the rounds calculated previously.</li>
  * </ol>
  */
 public class ConsistencyService implements OtterService {
@@ -40,23 +52,86 @@ public class ConsistencyService implements OtterService {
     /** A set of transaction nonce values seen in pre-handle that have not yet been handled. */
     private final Set<Long> transactionsAwaitingHandle = ConcurrentHashMap.newKeySet();
 
+    /** A history of all rounds and transaction nonce values contained within. */
+    private final ConsistencyServiceRoundHistory roundHistory = new ConsistencyServiceRoundHistory();
+
+    /** The round number of the previous round handled. */
+    private long previousRoundHandled = ConsensusConstants.ROUND_UNDEFINED;
+
+    /**
+     * {@inheritDoc}
+     */
+    public void initialize(
+            @NonNull final InitTrigger trigger,
+            @NonNull final NodeId selfId,
+            @NonNull final Configuration configuration,
+            @NonNull final OtterAppState state) {
+        if (trigger != InitTrigger.GENESIS && trigger != InitTrigger.RESTART) {
+            return;
+        }
+        final StateCommonConfig stateConfig = configuration.getConfigData(StateCommonConfig.class);
+        final ConsistencyServiceConfig consistencyServiceConfig =
+                configuration.getConfigData(ConsistencyServiceConfig.class);
+
+        final Path historyFileDirectory = stateConfig
+                .savedStateDirectory()
+                .resolve(consistencyServiceConfig.historyFileDirectory())
+                .resolve(Long.toString(selfId.id()));
+        try {
+            Files.createDirectories(historyFileDirectory);
+        } catch (final IOException e) {
+            log.error(EXCEPTION.getMarker(), "Unable to create log file directory", e);
+            throw new UncheckedIOException("unable to set up file system for consistency data", e);
+        }
+
+        final Path historyFilePath = historyFileDirectory.resolve(consistencyServiceConfig.historyFileName());
+        roundHistory.init(historyFilePath);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void destroy() {
+        roundHistory.close();
+    }
+
     /**
-     * Records the contents of all rounds, even empty ones. This method calculates a running hash that includes the
+     * Records the contents of all rounds, even empty ones. This method calculates a running checksum that includes the
      * round number and all transactions and stores the number of rounds handled in the state.
      *
      * @param writableStates the writable states used to modify the consistency state
      * @param round the round to handle
      */
     @Override
-    public void handleRound(@NonNull final WritableStates writableStates, @NonNull final Round round) {
-        new WritableConsistencyStateStore(writableStates)
+    public void onRoundStart(@NonNull final WritableStates writableStates, @NonNull final Round round) {
+        verifyRoundIncreases(round);
+
+        final WritableConsistencyStateStore store = new WritableConsistencyStateStore(writableStates)
                 .accumulateRunningChecksum(round.getRoundNum())
                 .incrementRoundsHandled();
+        roundHistory.onRoundStart(round, store.getRunningChecksum());
+    }
+
+    private void verifyRoundIncreases(@NonNull final Round round) {
+        if (previousRoundHandled == ConsensusConstants.ROUND_UNDEFINED) {
+            previousRoundHandled = round.getRoundNum();
+            return;
+        }
+
+        final long newRoundNumber = round.getRoundNum();
+
+        // make sure round numbers always increase
+        if (newRoundNumber <= previousRoundHandled) {
+            final String error = "Round " + newRoundNumber + " is not greater than round " + previousRoundHandled;
+            log.error(EXCEPTION.getMarker(), error);
+        }
+
+        previousRoundHandled = round.getRoundNum();
     }
 
     /**
-     * This method updates the running hash that includes the contents of all
-     * transactions.
+     * This method updates the running hash that includes the contents of all transactions.
      */
     @Override
     public void handleTransaction(
@@ -67,13 +142,14 @@ public void handleTransaction(
         final long transactionNonce = transaction.getNonce();
         new WritableConsistencyStateStore(writableStates).accumulateRunningChecksum(transactionNonce);
         if (!transactionsAwaitingHandle.remove(transactionNonce)) {
-            log.error(EXCEPTION.getMarker(), "Transaction {} was not prehandled.", transactionNonce);
+            log.error(EXCEPTION.getMarker(), "Transaction {} was not pre-handled.", transactionNonce);
         }
+        roundHistory.onTransaction(transaction);
     }
 
     /**
-     * This method records the checksum of all transactions that are pre-handled, so that we can verify
-     * that all consensus transactions were previously pre-handled.
+     * This method records the checksum of all transactions that are pre-handled, so that we can verify that all
+     * consensus transactions were previously pre-handled.
      *
      * @param event the event that contains the transaction
      * @param transaction the transaction being pre-handled
@@ -90,23 +166,12 @@ public void preHandleTransaction(
         }
     }
 
-    private void recordRound(@NonNull final Round round) {
-        // FUTURE WORK: Write the round data to in-memory structure and disk. Write to in-memory structure
-        // so we can verify that rounds increase monotonically (no rounds are repeated or skipped). Write to
-        // disk so that we can verify that the same rounds reach consensus after a restart during PCES replay.
-
-        // FUTURE WORK: Compare the round to rounds previous recorded in memory and do basic validations, like
-        // checking that the round number is one greater than the previous round number, and that all transactions
-        // were previously received in prehandle.
-    }
-
-    public void initialize() {
-        // FUTURE WORK: Read round data from disk (written in recordRound()) into in-memory structure.
-    }
-
-    public void recordPreHandleTransactions(@NonNull final Event event) {
-        // FUTURE WORK: Record the prehandle transactions so that we can verify all
-        // consensus transactions were previously sent to prehandle.
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void onRoundComplete(@NotNull final Round round) {
+        roundHistory.onRoundComplete();
     }
 
     /**

platform-sdk/consensus-otter-tests/src/testFixtures/java/org/hiero/otter/fixtures/app/services/consistency/ConsistencyServiceConfig.java

@@ -0,0 +1,16 @@
+// SPDX-License-Identifier: Apache-2.0
+package org.hiero.otter.fixtures.app.services.consistency;
+
+import com.swirlds.config.api.ConfigData;
+import com.swirlds.config.api.ConfigProperty;
+
+/**
+ * Configuration for the Consistency Service
+ *
+ * @param historyFileDirectory the directory where the history file will be stored
+ * @param historyFileName      the name of the history file
+ */
+@ConfigData("consistencyTestingTool")
+public record ConsistencyServiceConfig(
+        @ConfigProperty(defaultValue = "consistency-test") String historyFileDirectory,
+        @ConfigProperty(defaultValue = "ConsistencyTestLog.csv") String historyFileName) {}