feat: Suppress child records

hapi/hedera-protobufs/block/stream/output/consensus_service.proto

@@ -23,6 +23,8 @@ syntax = "proto3";
 
 package com.hedera.hapi.block.stream.output;
 
+import "custom_fees.proto";
+
 /*
  * Copyright (C) 2024 Hedera Hashgraph, LLC
  *
@@ -75,7 +77,15 @@ message DeleteTopicOutput {}
  * The actual topic running hash SHALL be present in a `StateChanges` block
  * item, and is not duplicated here.
  */
-message SubmitMessageOutput {}
+message SubmitMessageOutput {
+
+    /**
+     * Custom fees assessed during a SubmitMessage.
+     * <p>
+     * These fees SHALL be present in the full transfer list for the transaction.
+     */
+    repeated proto.AssessedCustomFee assessed_custom_fees = 1;
+}
 
 /**
  * A version of the topic running hash.

hapi/hedera-protobufs/block/stream/output/transaction_output.proto

@@ -40,6 +40,7 @@ import "stream/output/util_service.proto";
 import "stream/output/crypto_service.proto";
 import "stream/output/token_service.proto";
 import "stream/output/smart_contract_service.proto";
+import "stream/output/consensus_service.proto";
 
 /**
  * Output from a transaction.
@@ -155,5 +156,10 @@ message TransactionOutput {
          * Output from a token airdrop transaction.
          */
         TokenAirdropOutput token_airdrop = 8;
+
+        /**
+         * Output from a consensus submit message transaction.
+         */
+        SubmitMessageOutput submit_message = 9;
     }
 }

hedera-node/hedera-app/src/main/java/com/hedera/node/app/blocks/BlockItemsTranslator.java

@@ -182,6 +182,13 @@
                             recordBuilder.assessedCustomFees(cryptoOutput.assessedCustomFees());
                         }
                     }
+                    case CONSENSUS_SUBMIT_MESSAGE -> {
+                        final var submitMessageOutput = outputValueIfPresent(
+                                TransactionOutput::hasSubmitMessage, TransactionOutput::submitMessageOrThrow, outputs);
+                        if (submitMessageOutput != null) {
+                            recordBuilder.assessedCustomFees(submitMessageOutput.assessedCustomFees());
+                        }
+                    }
                     case CRYPTO_CREATE, CRYPTO_UPDATE -> recordBuilder.evmAddress(
                             ((CryptoOpContext) context).evmAddress());
                     case TOKEN_AIRDROP -> recordBuilder.newPendingAirdrops(

hedera-node/hedera-app/src/main/java/com/hedera/node/app/blocks/impl/BlockStreamBuilder.java

@@ -16,6 +16,7 @@
 
 package com.hedera.node.app.blocks.impl;
 
+import static com.hedera.hapi.node.base.HederaFunctionality.CONSENSUS_SUBMIT_MESSAGE;
 import static com.hedera.hapi.node.base.HederaFunctionality.CRYPTO_TRANSFER;
 import static com.hedera.hapi.node.base.HederaFunctionality.TOKEN_AIRDROP;
 import static com.hedera.hapi.node.base.ResponseCodeEnum.IDENTICAL_SCHEDULE_ALREADY_CREATED;
@@ -35,6 +36,7 @@
 import com.hedera.hapi.block.stream.output.SignScheduleOutput;
 import com.hedera.hapi.block.stream.output.StateChange;
 import com.hedera.hapi.block.stream.output.StateChanges;
+import com.hedera.hapi.block.stream.output.SubmitMessageOutput;
 import com.hedera.hapi.block.stream.output.TokenAirdropOutput;
 import com.hedera.hapi.block.stream.output.TransactionOutput;
 import com.hedera.hapi.block.stream.output.TransactionResult;
@@ -1124,6 +1126,9 @@
         } else if (functionality == TOKEN_AIRDROP && hasAssessedCustomFees) {
             items.add(
                     itemWith(TransactionOutput.newBuilder().tokenAirdrop(new TokenAirdropOutput(assessedCustomFees))));
+        } else if (functionality == CONSENSUS_SUBMIT_MESSAGE && hasAssessedCustomFees) {
+            items.add(itemWith(
+                    TransactionOutput.newBuilder().submitMessage(new SubmitMessageOutput(assessedCustomFees))));
         }
     }
 

hedera-node/hedera-consensus-service-impl/src/main/java/com/hedera/node/app/service/consensus/impl/handlers/ConsensusSubmitMessageHandler.java

@@ -36,7 +36,7 @@
 import static com.hedera.node.app.spi.workflows.HandleException.validateTrue;
 import static com.hedera.node.app.spi.workflows.PreCheckException.validateFalsePreCheck;
 import static com.hedera.node.app.spi.workflows.PreCheckException.validateTruePreCheck;
-import static com.hedera.node.app.spi.workflows.record.StreamBuilder.TransactionCustomizer.NOOP_TRANSACTION_CUSTOMIZER;
+import static com.hedera.node.app.spi.workflows.record.StreamBuilder.TransactionCustomizer.SUPPRESSING_TRANSACTION_CUSTOMIZER;
 import static java.util.Objects.requireNonNull;
 
 import com.hedera.hapi.node.base.AccountID;
@@ -48,6 +48,8 @@
 import com.hedera.hapi.node.base.TransactionID;
 import com.hedera.hapi.node.consensus.ConsensusSubmitMessageTransactionBody;
 import com.hedera.hapi.node.state.consensus.Topic;
+import com.hedera.hapi.node.token.CryptoTransferTransactionBody;
+import com.hedera.hapi.node.transaction.AssessedCustomFee;
 import com.hedera.hapi.node.transaction.ConsensusCustomFee;
 import com.hedera.hapi.node.transaction.FixedFee;
 import com.hedera.hapi.node.transaction.TransactionBody;
@@ -76,6 +78,7 @@
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -157,6 +160,9 @@
         final var config = handleContext.configuration().getConfigData(ConsensusConfig.class);
         validateTransaction(txn, config, topic);
 
+        final var streamBuilder =
+                handleContext.savepointStack().getBaseBuilder(ConsensusSubmitMessageStreamBuilder.class);
+
         /* handle custom fees */
         if (!topic.customFees().isEmpty() && !isFeeExempted(topic.feeExemptKeyList(), handleContext.keyVerifier())) {
             // check payer limits or throw
@@ -165,16 +171,31 @@
             }
             // create synthetic body and dispatch crypto transfer
             final var syntheticBodies = customFeeAssessor.assessCustomFee(topic, handleContext);
+            // build a list with the top level custom fees
+            final var assessedCustomFees =
+                    new ArrayList<>(buildTopLevelAssessedCustomFees(handleContext.payer(), syntheticBodies));
+
+            // dispatch transfers to pay the fees, but suppress any child records. All assessed fees will be
+            // externalized in to the top level txn record/stream
             for (final var syntheticBody : syntheticBodies) {
-                final var record = handleContext.dispatch(stepDispatch(
+                final var dispatchedStreamBuilder = handleContext.dispatch(stepDispatch(
                         handleContext.payer(),
                         TransactionBody.newBuilder()
                                 .cryptoTransfer(syntheticBody)
                                 .build(),
                         CryptoTransferStreamBuilder.class,
-                        NOOP_TRANSACTION_CUSTOMIZER));
-                validateTrue(record.status().equals(SUCCESS), record.status());
+                        SUPPRESSING_TRANSACTION_CUSTOMIZER));
+
+                validateTrue(dispatchedStreamBuilder.status().equals(SUCCESS), dispatchedStreamBuilder.status());
+
+                // check if there is nested custom fees
+                if (!dispatchedStreamBuilder.getAssessedCustomFees().isEmpty()) {
+                    assessedCustomFees.addAll(dispatchedStreamBuilder.getAssessedCustomFees());
+                }
             }
+
+            // externalize all custom fees
+            streamBuilder.assessedCustomFees(assessedCustomFees);
         }
 
         try {
@@ -184,9 +205,7 @@
             It will not be committed to state until commit is called on the state.--- */
             topicStore.put(updatedTopic);
 
-            final var recordBuilder =
-                    handleContext.savepointStack().getBaseBuilder(ConsensusSubmitMessageStreamBuilder.class);
-            recordBuilder
+            streamBuilder
                     .topicRunningHash(updatedTopic.runningHash())
                     .topicSequenceNumber(updatedTopic.sequenceNumber())
                     .topicRunningHashVersion(RUNNING_HASH_VERSION);
@@ -439,4 +458,35 @@
                 .addNetworkRamByteSeconds((LONG_SIZE + TX_HASH_SIZE) * RECEIPT_STORAGE_TIME_SEC)
                 .calculate();
     }
+
+    private List<AssessedCustomFee> buildTopLevelAssessedCustomFees(
+            AccountID payer, List<CryptoTransferTransactionBody> bodies) {
+        final var assessedCustomFees = new ArrayList<AssessedCustomFee>();
+        for (final var body : bodies) {
+            final var customFeeBuilder = AssessedCustomFee.newBuilder().effectivePayerAccountId(payer);
+            if (body.tokenTransfers().isEmpty()) {
+                final var aa = body.transfers().accountAmounts();
+                for (final var amount : aa) {
+                    if (amount.amount() > 0) {
+                        customFeeBuilder.amount(amount.amount());
+                        customFeeBuilder.feeCollectorAccountId(amount.accountID());
+                    }
+                }
+            } else {
+                final var tokenTransferLists = body.tokenTransfers();
+                for (final var tokenTransferList : tokenTransferLists) {
+                    customFeeBuilder.tokenId(tokenTransferList.token());
+                    for (final var transfer : tokenTransferList.transfers()) {
+                        if (transfer.amount() > 0) {
+                            customFeeBuilder.amount(transfer.amount());
+                            customFeeBuilder.feeCollectorAccountId(transfer.accountID());
+                        }
+                    }
+                }
+            }
+            assessedCustomFees.add(customFeeBuilder.build());
+        }
+
+        return assessedCustomFees;
+    }
 }

hedera-node/hedera-consensus-service-impl/src/main/java/com/hedera/node/app/service/consensus/impl/records/ConsensusSubmitMessageStreamBuilder.java

@@ -16,9 +16,12 @@
 
 package com.hedera.node.app.service.consensus.impl.records;
 
+import com.hedera.hapi.node.transaction.AssessedCustomFee;
+import com.hedera.node.app.service.token.records.CryptoTransferStreamBuilder;
 import com.hedera.node.app.spi.workflows.record.StreamBuilder;
 import com.hedera.pbj.runtime.io.buffer.Bytes;
 import edu.umd.cs.findbugs.annotations.NonNull;
+import java.util.List;
 
 /**
  * A {@code StreamBuilder} specialization for tracking the side effects of a {@code ConsensusSubmitMessage}
@@ -51,4 +54,12 @@ public interface ConsensusSubmitMessageStreamBuilder extends StreamBuilder {
      */
     @NonNull
     ConsensusSubmitMessageStreamBuilder topicRunningHashVersion(long topicRunningHashVersion);
+
+    /**
+     * Tracks the total custom fees assessed in the transaction.
+     * @param assessedCustomFees the total custom fees assessed in the transaction
+     * @return this builder
+     */
+    @NonNull
+    CryptoTransferStreamBuilder assessedCustomFees(@NonNull List<AssessedCustomFee> assessedCustomFees);
 }

hedera-node/hedera-token-service/src/main/java/com/hedera/node/app/service/token/records/CryptoTransferStreamBuilder.java

@@ -53,6 +53,14 @@ public interface CryptoTransferStreamBuilder extends StreamBuilder {
     @NonNull
     CryptoTransferStreamBuilder tokenTransferLists(@NonNull List<TokenTransferList> tokenTransferLists);
 
+    /**
+     * Returns all assessed custom fees for this call.
+     *
+     * @return the assessed custom fees
+     */
+    @NonNull
+    List<AssessedCustomFee> getAssessedCustomFees();
+
     /**
      * Tracks the total custom fees assessed in the transaction.
      * @param assessedCustomFees the total custom fees assessed in the transaction

hedera-node/test-clients/src/main/java/com/hedera/services/bdd/junit/support/translators/impl/SubmitMessageTranslator.java

@@ -19,6 +19,7 @@
 import static com.hedera.hapi.node.base.ResponseCodeEnum.SUCCESS;
 
 import com.hedera.hapi.block.stream.output.StateChange;
+import com.hedera.hapi.block.stream.output.TransactionOutput;
 import com.hedera.node.app.state.SingleTransactionRecord;
 import com.hedera.services.bdd.junit.support.translators.BaseTranslator;
 import com.hedera.services.bdd.junit.support.translators.BlockTransactionPartsTranslator;
@@ -45,6 +46,9 @@ public SingleTransactionRecord translate(
             @NonNull final List<StateChange> remainingStateChanges) {
         return baseTranslator.recordFrom(parts, (receiptBuilder, recordBuilder) -> {
             if (parts.status() == SUCCESS) {
+                parts.outputIfPresent(TransactionOutput.TransactionOneOfType.SUBMIT_MESSAGE)
+                        .ifPresent(output -> recordBuilder.assessedCustomFees(
+                                output.submitMessageOrThrow().assessedCustomFees()));
                 receiptBuilder.topicRunningHashVersion(RUNNING_HASH_VERSION);
                 final var iter = remainingStateChanges.listIterator();
                 while (iter.hasNext()) {