feat: implement the AtomicBatchHandler.handle() method

hapi/hedera-protobufs/services/response_code.proto

@@ -1726,4 +1726,9 @@ enum ResponseCodeEnum {
     * The list of batch transactions contains null values
     */
     BATCH_LIST_CONTAINS_NULL_VALUES = 390;
+
+    /**
+     * The inner transaction failed
+     */
+    INNER_TRANSACTION_FAILED = 391;
 }

hedera-node/hedera-app-spi/src/main/java/com/hedera/node/app/spi/workflows/DispatchOptions.java

@@ -312,4 +312,31 @@ public static <T extends StreamBuilder> DispatchOptions<T> stepDispatch(
                 transactionCustomizer,
                 metaData);
     }
+
+    /**
+     * returns options for a dispatch for atomic batch transaction
+     * @param payerId the account to pay for the dispatch
+     * @param body the transaction to dispatch
+     * @param streamBuilderType the type of stream builder to use for the dispatch
+     * @return the options for the atomic batch
+     * @param <T> the type of stream builder to use for the dispatch
+     */
+    public static <T extends StreamBuilder> DispatchOptions<T> atomicBatchDispatch(
+            @NonNull final AccountID payerId,
+            @NonNull final TransactionBody body,
+            @NonNull final Class<T> streamBuilderType) {
+        return new DispatchOptions<>(
+                Commit.WITH_PARENT,
+                payerId,
+                body,
+                UsePresetTxnId.NO,
+                PREAUTHORIZED_KEYS,
+                emptySet(),
+                TransactionCategory.CHILD,
+                ConsensusThrottling.ON,
+                streamBuilderType,
+                ReversingBehavior.REVERSIBLE,
+                NOOP_TRANSACTION_CUSTOMIZER,
+                EMPTY_METADATA);
+    }
 }

hedera-node/hedera-app-spi/src/main/java/com/hedera/node/app/spi/workflows/HandleContext.java

@@ -18,6 +18,7 @@
 
 import com.hedera.hapi.node.base.AccountID;
 import com.hedera.hapi.node.base.HederaFunctionality;
+import com.hedera.hapi.node.base.Transaction;
 import com.hedera.hapi.node.transaction.TransactionBody;
 import com.hedera.node.app.spi.authorization.SystemPrivilege;
 import com.hedera.node.app.spi.fees.ExchangeRateInfo;
@@ -35,6 +36,7 @@
 import com.swirlds.state.lifecycle.info.NetworkInfo;
 import com.swirlds.state.lifecycle.info.NodeInfo;
 import edu.umd.cs.findbugs.annotations.NonNull;
+import edu.umd.cs.findbugs.annotations.Nullable;
 import java.time.Instant;
 import java.util.Map;
 
@@ -402,4 +404,11 @@ enum ConsensusThrottling {
      * be used to pass additional information to the targeted handlers.
      */
     DispatchMetadata dispatchMetadata();
+
+    /**
+     * Returns the TransactionBogy from the given transaction.
+     * @return the TransactionBogy
+     */
+    @Nullable
+    TransactionBody bodyFromTransaction(@NonNull final Transaction tx);
 }

hedera-node/hedera-app/src/main/java/com/hedera/node/app/workflows/handle/DispatchHandleContext.java

@@ -29,6 +29,7 @@
 import com.hedera.hapi.node.base.SignatureMap;
 import com.hedera.hapi.node.base.SubType;
 import com.hedera.hapi.node.base.Timestamp;
+import com.hedera.hapi.node.base.Transaction;
 import com.hedera.hapi.node.base.TransactionID;
 import com.hedera.hapi.node.transaction.TransactionBody;
 import com.hedera.hapi.util.UnknownHederaFunctionality;
@@ -413,4 +414,15 @@
     public DispatchMetadata dispatchMetadata() {
         return dispatchMetaData;
     }
+
+    @Nullable
+    @Override
+    public TransactionBody bodyFromTransaction(@NonNull final Transaction tx) throws HandleException {
+        try {
+            final var transactionInfo = transactionChecker.check(tx, null);
+            return transactionInfo.txBody();
+        } catch (PreCheckException e) {
+            throw new HandleException(e.responseCode());
+        }
+    }
 }

hedera-node/hedera-app/src/main/java/com/hedera/node/app/workflows/handle/DispatchProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2024 Hedera Hashgraph, LLC
+ * Copyright (C) 2024-2025 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.

hedera-node/hedera-config/src/testFixtures/java/com/hedera/node/config/testfixtures/HederaTestConfigBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2023-2024 Hedera Hashgraph, LLC
+ * Copyright (C) 2023-2025 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.
@@ -39,6 +39,7 @@
 import com.hedera.node.config.converter.SemanticVersionConverter;
 import com.hedera.node.config.data.AccountsConfig;
 import com.hedera.node.config.data.ApiPermissionConfig;
+import com.hedera.node.config.data.AtomicBatchConfig;
 import com.hedera.node.config.data.AutoCreationConfig;
 import com.hedera.node.config.data.AutoRenew2Config;
 import com.hedera.node.config.data.AutoRenewConfig;
@@ -188,6 +189,7 @@ public static TestConfigBuilder create() {
                 .withConfigDataType(NodesConfig.class)
                 .withConfigDataType(TssConfig.class)
                 .withConfigDataType(BlockStreamConfig.class)
+                .withConfigDataType(AtomicBatchConfig.class)
                 .withConverter(CongestionMultipliers.class, new CongestionMultipliersConverter())
                 .withConverter(EntityScaleFactors.class, new EntityScaleFactorsConverter())
                 .withConverter(KnownBlockValues.class, new KnownBlockValuesConverter())

hedera-node/hedera-util-service-impl/src/main/java/com/hedera/node/app/service/util/impl/handlers/AtomicBatchHandler.java

@@ -16,10 +16,14 @@
 
 package com.hedera.node.app.service.util.impl.handlers;
 
+import static com.hedera.hapi.node.base.ResponseCodeEnum.INNER_TRANSACTION_FAILED;
+import static com.hedera.hapi.node.base.ResponseCodeEnum.SUCCESS;
+import static com.hedera.node.app.spi.workflows.DispatchOptions.atomicBatchDispatch;
 import static java.util.Objects.requireNonNull;
 
 import com.hedera.hapi.node.base.HederaFunctionality;
 import com.hedera.hapi.node.base.SubType;
+import com.hedera.hapi.node.transaction.TransactionBody;
 import com.hedera.node.app.spi.fees.FeeContext;
 import com.hedera.node.app.spi.fees.Fees;
 import com.hedera.node.app.spi.workflows.HandleContext;
@@ -28,6 +32,7 @@
 import com.hedera.node.app.spi.workflows.PreHandleContext;
 import com.hedera.node.app.spi.workflows.PureChecksContext;
 import com.hedera.node.app.spi.workflows.TransactionHandler;
+import com.hedera.node.app.spi.workflows.record.StreamBuilder;
 import com.hedera.node.config.data.AtomicBatchConfig;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import javax.inject.Inject;
@@ -42,9 +47,7 @@ public class AtomicBatchHandler implements TransactionHandler {
      * Constructs a {@link AtomicBatchHandler}
      */
     @Inject
-    public AtomicBatchHandler() {
-        // exists for Dagger injection
-    }
+    public AtomicBatchHandler() {}
 
     /**
      * Performs checks independent of state or context.
@@ -69,14 +72,28 @@ public void preHandle(@NonNull final PreHandleContext context) throws PreCheckEx
     }
 
     @Override
-    public void handle(@NonNull final HandleContext handleContext) throws HandleException {
-        requireNonNull(handleContext);
-        // TODO
-        if (!handleContext
-                .configuration()
-                .getConfigData(AtomicBatchConfig.class)
-                .isEnabled()) {
-            return;
+    public void handle(@NonNull final HandleContext context) throws HandleException {
+        requireNonNull(context);
+        final var batchConfig = context.configuration().getConfigData(AtomicBatchConfig.class);
+        final var op = context.body().atomicBatchOrThrow();
+        if (batchConfig.isEnabled()) {
+            final var transactions = op.transactions();
+            for (final var transaction : transactions) {
+                TransactionBody body;
+                try {
+                    body = context.bodyFromTransaction(transaction);
+                } catch (HandleException e) {
+                    // we should have validated the inner transaction in preChecks already, so this should not happen.
+                    throw new HandleException(INNER_TRANSACTION_FAILED);
+                }
+                final var payerId = body.transactionIDOrThrow().accountIDOrThrow();
+                // all the inner transactions' keys are verified in PreHandleWorkflow
+                final var dispatchOptions = atomicBatchDispatch(payerId, body, StreamBuilder.class);
+                final var streamBuilder = context.dispatch(dispatchOptions);
+                if (streamBuilder == null || streamBuilder.status() != SUCCESS) {
+                    throw new HandleException(INNER_TRANSACTION_FAILED);
+                }
+            }
         }
     }
 

hedera-node/hedera-util-service-impl/src/test/java/com/hedera/node/app/service/util/impl/test/handlers/AtomicBatchHandlerTest.java

@@ -0,0 +1,180 @@
+/*
+ * Copyright (C) 2023-2025 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.hedera.node.app.service.util.impl.test.handlers;
+
+import static com.hedera.hapi.node.base.ResponseCodeEnum.INNER_TRANSACTION_FAILED;
+import static com.hedera.hapi.node.base.ResponseCodeEnum.INVALID_TRANSACTION_BODY;
+import static com.hedera.hapi.node.base.ResponseCodeEnum.SUCCESS;
+import static com.hedera.node.app.spi.workflows.DispatchOptions.atomicBatchDispatch;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.willThrow;
+import static org.mockito.Mock.Strictness.LENIENT;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import com.hedera.hapi.node.base.AccountID;
+import com.hedera.hapi.node.base.Key;
+import com.hedera.hapi.node.base.Timestamp;
+import com.hedera.hapi.node.base.Transaction;
+import com.hedera.hapi.node.base.TransactionID;
+import com.hedera.hapi.node.consensus.ConsensusCreateTopicTransactionBody;
+import com.hedera.hapi.node.consensus.ConsensusDeleteTopicTransactionBody;
+import com.hedera.hapi.node.transaction.TransactionBody;
+import com.hedera.hapi.node.util.AtomicBatchTransactionBody;
+import com.hedera.node.app.service.util.impl.handlers.AtomicBatchHandler;
+import com.hedera.node.app.spi.records.BlockRecordInfo;
+import com.hedera.node.app.spi.workflows.HandleContext;
+import com.hedera.node.app.spi.workflows.HandleException;
+import com.hedera.node.app.spi.workflows.record.StreamBuilder;
+import com.hedera.node.config.testfixtures.HederaTestConfigBuilder;
+import com.hedera.pbj.runtime.io.buffer.Bytes;
+import java.time.Instant;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class AtomicBatchHandlerTest {
+    @Mock(strictness = LENIENT)
+    private HandleContext handleContext;
+
+    @Mock
+    private StreamBuilder recordBuilder;
+
+    @Mock
+    private BlockRecordInfo blockRecordInfo;
+
+    private AtomicBatchHandler subject;
+
+    private Timestamp consensusTimestamp =
+            Timestamp.newBuilder().seconds(1_234_567L).build();
+    private static final Key SIMPLE_KEY_A = Key.newBuilder()
+            .ed25519(Bytes.wrap("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".getBytes()))
+            .build();
+    private AccountID payerId1 = AccountID.newBuilder().accountNum(1001).build();
+    private AccountID payerId2 = AccountID.newBuilder().accountNum(1002).build();
+    private AccountID payerId3 = AccountID.newBuilder().accountNum(1003).build();
+
+    @BeforeEach
+    void setUp() {
+        final var config = HederaTestConfigBuilder.create()
+                .withValue("atomicBatch.isEnabled", true)
+                .getOrCreateConfig();
+        given(handleContext.configuration()).willReturn(config);
+
+        subject = new AtomicBatchHandler();
+    }
+
+    @Test
+    void cannotParseInnerTransactionFailed() {
+        final var batchKey = SIMPLE_KEY_A;
+        final var transaction = mock(Transaction.class);
+        final var txnBody = newAtomicBatch(payerId1, consensusTimestamp, batchKey, transaction);
+        given(handleContext.body()).willReturn(txnBody);
+        given(handleContext.consensusNow()).willReturn(Instant.ofEpochSecond(1_234_567L));
+        willThrow(new HandleException(INVALID_TRANSACTION_BODY))
+                .given(handleContext)
+                .bodyFromTransaction(transaction);
+
+        final var msg = assertThrows(HandleException.class, () -> subject.handle(handleContext));
+        assertEquals(INNER_TRANSACTION_FAILED, msg.getStatus());
+    }
+
+    @Test
+    void innerTransactionFailed() {
+        final var batchKey = SIMPLE_KEY_A;
+        final var transaction = mock(Transaction.class);
+        final var txnBody = newAtomicBatch(payerId1, consensusTimestamp, batchKey, transaction);
+        final var innerTxnBody = newTxnBodyBuilder(payerId2, consensusTimestamp)
+                .consensusCreateTopic(
+                        ConsensusCreateTopicTransactionBody.newBuilder().build())
+                .build();
+        given(handleContext.body()).willReturn(txnBody);
+        given(handleContext.consensusNow()).willReturn(Instant.ofEpochSecond(1_234_567L));
+        given(handleContext.bodyFromTransaction(transaction)).willReturn(innerTxnBody);
+        final var msg = assertThrows(HandleException.class, () -> subject.handle(handleContext));
+        assertEquals(INNER_TRANSACTION_FAILED, msg.getStatus());
+    }
+
+    @Test
+    void handleDispatched() {
+        final var batchKey = SIMPLE_KEY_A;
+        final var transaction = mock(Transaction.class);
+        final var txnBody = newAtomicBatch(payerId1, consensusTimestamp, batchKey, transaction);
+        final var innerTxnBody = newTxnBodyBuilder(payerId2, consensusTimestamp)
+                .consensusCreateTopic(
+                        ConsensusCreateTopicTransactionBody.newBuilder().build())
+                .build();
+        given(handleContext.body()).willReturn(txnBody);
+        given(handleContext.consensusNow()).willReturn(Instant.ofEpochSecond(1_234_567L));
+        given(handleContext.bodyFromTransaction(transaction)).willReturn(innerTxnBody);
+        final var dispatchOptions = atomicBatchDispatch(payerId2, innerTxnBody, StreamBuilder.class);
+        given(handleContext.dispatch(dispatchOptions)).willReturn(recordBuilder);
+        given(recordBuilder.status()).willReturn(SUCCESS);
+        subject.handle(handleContext);
+        verify(handleContext).dispatch(dispatchOptions);
+    }
+
+    @Test
+    void handleMultipleDispatched() {
+        final var batchKey = SIMPLE_KEY_A;
+        final var transaction1 = mock(Transaction.class);
+        final var transaction2 = mock(Transaction.class);
+        final var txnBody = newAtomicBatch(payerId1, consensusTimestamp, batchKey, transaction1, transaction2);
+        final var innerTxnBody1 = newTxnBodyBuilder(payerId2, consensusTimestamp)
+                .consensusCreateTopic(
+                        ConsensusCreateTopicTransactionBody.newBuilder().build())
+                .build();
+        final var innerTxnBody2 = newTxnBodyBuilder(payerId3, consensusTimestamp)
+                .consensusDeleteTopic(
+                        ConsensusDeleteTopicTransactionBody.newBuilder().build())
+                .build();
+        given(handleContext.body()).willReturn(txnBody);
+        given(handleContext.consensusNow()).willReturn(Instant.ofEpochSecond(1_234_567L));
+        given(handleContext.bodyFromTransaction(transaction1)).willReturn(innerTxnBody1);
+        given(handleContext.bodyFromTransaction(transaction2)).willReturn(innerTxnBody2);
+        final var dispatchOptions1 = atomicBatchDispatch(payerId2, innerTxnBody1, StreamBuilder.class);
+        final var dispatchOptions2 = atomicBatchDispatch(payerId3, innerTxnBody2, StreamBuilder.class);
+        given(handleContext.dispatch(dispatchOptions1)).willReturn(recordBuilder);
+        given(handleContext.dispatch(dispatchOptions2)).willReturn(recordBuilder);
+        given(recordBuilder.status()).willReturn(SUCCESS);
+        subject.handle(handleContext);
+        verify(handleContext).dispatch(dispatchOptions1);
+        verify(handleContext).dispatch(dispatchOptions2);
+    }
+
+    private TransactionBody newAtomicBatch(
+            AccountID payerId, Timestamp consensusTimestamp, Key batchKey, Transaction... transactions) {
+        final var atomicBatchBuilder = AtomicBatchTransactionBody.newBuilder().transactions(transactions);
+        return newTxnBodyBuilder(payerId, consensusTimestamp)
+                .batchKey(batchKey)
+                .atomicBatch(atomicBatchBuilder)
+                .build();
+    }
+
+    private TransactionBody.Builder newTxnBodyBuilder(AccountID payerId, Timestamp consensusTimestamp) {
+        final var txnId = TransactionID.newBuilder()
+                .accountID(payerId)
+                .transactionValidStart(consensusTimestamp)
+                .build();
+        return TransactionBody.newBuilder().transactionID(txnId);
+    }
+}