chore: Add an API to submit Otter transactions to individual nodes (#21951)

Signed-off-by: Kelly Greco <[email protected]>

Author: poulok

platform-sdk/consensus-otter-docker-app/src/testFixtures/java/org/hiero/consensus/otter/docker/app/platform/NodeCommunicationService.java

@@ -7,6 +7,7 @@
 import static java.util.Objects.requireNonNull;
 import static org.hiero.otter.fixtures.internal.helpers.Utils.createConfiguration;
 
+import com.google.protobuf.ByteString;
 import com.google.protobuf.Empty;
 import com.hedera.hapi.node.base.SemanticVersion;
 import com.hedera.hapi.node.state.roster.Roster;
@@ -205,10 +206,15 @@ public synchronized void submitTransaction(
         }
 
         wrapWithErrorHandling(responseObserver, () -> {
-            final boolean result =
-                    consensusNodeManager.submitTransaction(request.getPayload().toByteArray());
-            responseObserver.onNext(
-                    TransactionRequestAnswer.newBuilder().setResult(result).build());
+            int numFailed = 0;
+            for (final ByteString payload : request.getPayloadList()) {
+                if (!consensusNodeManager.submitTransaction(payload.toByteArray())) {
+                    numFailed++;
+                }
+            }
+            responseObserver.onNext(TransactionRequestAnswer.newBuilder()
+                    .setNumFailed(numFailed)
+                    .build());
             responseObserver.onCompleted();
         });
     }

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

@@ -15,6 +15,7 @@
 import java.util.Set;
 import org.hiero.consensus.model.quiescence.QuiescenceCommand;
 import org.hiero.consensus.model.status.PlatformStatus;
+import org.hiero.otter.fixtures.app.OtterTransaction;
 import org.hiero.otter.fixtures.internal.helpers.Utils;
 import org.hiero.otter.fixtures.network.Partition;
 import org.hiero.otter.fixtures.network.Topology;
@@ -281,6 +282,22 @@ default Partition createNetworkPartition(@NonNull final Node node0, @NonNull fin
      */
     void freeze();
 
+    /**
+     * Submits a single transaction to the first active node found in the network.
+     *
+     * @param transaction the transaction to submit
+     */
+    default void submitTransaction(@NonNull final OtterTransaction transaction) {
+        submitTransactions(List.of(transaction));
+    }
+
+    /**
+     * Submits the transactions to the first active node found in the network.
+     *
+     * @param transactions the transactions to submit
+     */
+    void submitTransactions(@NonNull List<OtterTransaction> transactions);
+
     /**
      * Triggers a catastrophic ISS. All nodes in the network will calculate different hashes for an upcoming round.
      */

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

@@ -6,10 +6,12 @@
 import edu.umd.cs.findbugs.annotations.Nullable;
 import java.nio.file.Path;
 import java.time.Duration;
+import java.util.List;
 import org.hiero.consensus.model.node.KeysAndCerts;
 import org.hiero.consensus.model.node.NodeId;
 import org.hiero.consensus.model.quiescence.QuiescenceCommand;
 import org.hiero.consensus.model.status.PlatformStatus;
+import org.hiero.otter.fixtures.app.OtterTransaction;
 import org.hiero.otter.fixtures.result.SingleNodeConsensusResult;
 import org.hiero.otter.fixtures.result.SingleNodeEventStreamResult;
 import org.hiero.otter.fixtures.result.SingleNodeLogResult;
@@ -102,7 +104,23 @@ default void startSyntheticBottleneck() {
     void sendQuiescenceCommand(@NonNull QuiescenceCommand command);
 
     /**
-     * Allows to override the default timeout for node operations.
+     * Submits a transaction to the node.
+     *
+     * @param transaction the transaction to submit
+     */
+    default void submitTransaction(@NonNull OtterTransaction transaction) {
+        submitTransactions(List.of(transaction));
+    }
+
+    /**
+     * Submits transactions to the node.
+     *
+     * @param transactions the list of transactions to submit
+     */
+    void submitTransactions(@NonNull List<OtterTransaction> transactions);
+
+    /**
+     * Overrides the default timeout for node operations.
      *
      * @param timeout the duration to wait before considering the operation as failed
      * @return an instance of {@link AsyncNodeActions} that can be used to perform node actions

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

@@ -34,6 +34,7 @@
 import java.security.SecureRandom;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.List;
 import java.util.Random;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -329,22 +330,21 @@ protected void doSendQuiescenceCommand(@NonNull final QuiescenceCommand command,
      * {@inheritDoc}
      */
     @Override
-    public void submitTransaction(@NonNull final OtterTransaction transaction) {
+    public void submitTransactions(@NonNull final List<OtterTransaction> transactions) {
         throwIfInLifecycle(INIT, "Node has not been started yet.");
         throwIfInLifecycle(SHUTDOWN, "Node has been shut down.");
         throwIfInLifecycle(DESTROYED, "Node has been destroyed.");
 
         try {
-            final TransactionRequest request = TransactionRequest.newBuilder()
-                    .setPayload(transaction.toByteString())
-                    .build();
-
-            final TransactionRequestAnswer answer = nodeCommBlockingStub.submitTransaction(request);
-            if (!answer.getResult()) {
-                fail("Failed to submit transaction for node %d.".formatted(selfId.id()));
+            final TransactionRequest.Builder builder = TransactionRequest.newBuilder();
+            transactions.forEach(t -> builder.addPayload(t.toByteString()));
+            final TransactionRequestAnswer answer = nodeCommBlockingStub.submitTransaction(builder.build());
+            if (answer.getNumFailed() > 0) {
+                fail("%d out of %d transaction(s) failed to submit for node %d."
+                        .formatted(answer.getNumFailed(), transactions.size(), selfId.id()));
             }
         } catch (final Exception e) {
-            fail("Failed to submit transaction to node %d".formatted(selfId.id()), e);
+            fail("Failed to submit transaction(s) to node %d".formatted(selfId.id()), e);
         }
     }
 
@@ -372,8 +372,8 @@ public ContainerNodeConfiguration configuration() {
     }
 
     /**
-     * Gets the container instance for this node. This allows direct access to the underlying
-     * Testcontainers container for operations like retrieving console logs.
+     * Gets the container instance for this node. This allows direct access to the underlying Testcontainers container
+     * for operations like retrieving console logs.
      *
      * @return the container instance
      */

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

@@ -596,17 +596,15 @@ private boolean allNodesInCheckingOrCatastrophicFailure() {
     }
 
     /**
-     * Submits the transaction to the first active node found in the network.
-     *
-     * @param transaction the transaction to submit
+     * {@inheritDoc}
      */
-    private void submitTransaction(@NonNull final OtterTransaction transaction) {
+    public void submitTransactions(@NonNull final List<OtterTransaction> transactions) {
         nodes().stream()
                 .filter(Node::isActive)
                 .findFirst()
                 .map(node -> (AbstractNode) node)
                 .orElseThrow(() -> new AssertionError("No active node found to send transaction to."))
-                .submitTransaction(transaction);
+                .submitTransactions(transactions);
     }
 
     /**

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

@@ -293,13 +293,6 @@ public void killImmediately() {
      */
     protected abstract void doKillImmediately(@NonNull Duration timeout);
 
-    /**
-     * Submit a transaction to the node.
-     *
-     * @param transaction the transaction to submit
-     */
-    protected abstract void submitTransaction(@NonNull OtterTransaction transaction);
-
     /**
      * {@inheritDoc}
      */

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

@@ -41,6 +41,7 @@
 import java.nio.file.Path;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.List;
 import java.util.Objects;
 import java.util.Random;
 import java.util.function.Consumer;
@@ -371,7 +372,7 @@ protected void doSendQuiescenceCommand(@NonNull final QuiescenceCommand command,
      * {@inheritDoc}
      */
     @Override
-    public void submitTransaction(@NonNull final OtterTransaction transaction) {
+    public void submitTransactions(@NonNull final List<OtterTransaction> transactions) {
         try (final LoggingContextScope ignored = installNodeContext()) {
             throwIsNotInLifecycle(RUNNING, "Cannot submit transaction when the network is not running.");
             assert executionLayer != null; // executionLayer must be initialized if lifeCycle is STARTED
@@ -381,7 +382,7 @@ public void submitTransaction(@NonNull final OtterTransaction transaction) {
                 return;
             }
 
-            executionLayer.submitApplicationTransaction(transaction.toByteArray());
+            transactions.forEach(tx -> executionLayer.submitApplicationTransaction(tx.toByteArray()));
         }
     }
 

platform-sdk/consensus-otter-tests/src/testFixtures/proto/node_communication.proto

@@ -177,13 +177,13 @@ message EventMessage {
 // Wrapper for a transaction submission request.
 message TransactionRequest {
   // Serialized transaction data.
-  bytes payload = 1;
+  repeated bytes payload = 1;
 }
 
 // Response to a transaction submission request.
 message TransactionRequestAnswer {
-  // indicator if the platform accepted the transaction
-  bool result = 1;
+  // The number of transactions that failed submission to the node
+  uint32 numFailed = 1;
 }
 
 // Request to set the synthetic bottleneck.