HIP-1056 Link scheduled transactions and its trigger

common/src/main/java/org/hiero/mirror/common/domain/transaction/BlockTransaction.java

@@ -77,6 +77,7 @@ public class BlockTransaction implements StreamItem {
     private final SignedTransaction signedTransaction;
     private final byte[] signedTransactionBytes;
     private final TopicID topicId; // for consensus submit message transaction
+    private final BlockTransaction trigger;
 
     @Getter(AccessLevel.NONE)
     private final AtomicReference<TopicMessage> topicMessage = new AtomicReference<>();
@@ -106,7 +107,8 @@ public BlockTransaction(
             List<TraceData> traceData,
             TransactionBody transactionBody,
             TransactionResult transactionResult,
-            Map<TransactionCase, TransactionOutput> transactionOutputs) {
+            Map<TransactionCase, TransactionOutput> transactionOutputs,
+            BlockTransaction trigger) {
         this.previous = previous;
         this.signedTransaction = signedTransaction;
         this.signedTransactionBytes = signedTransactionBytes;
@@ -115,6 +117,7 @@ public BlockTransaction(
         this.transactionBody = transactionBody;
         this.transactionResult = transactionResult;
         this.transactionOutputs = transactionOutputs;
+        this.trigger = trigger;
 
         consensusTimestamp = DomainUtils.timestampInNanosMax(transactionResult.getConsensusTimestamp());
         parentConsensusTimestamp = transactionResult.hasParentConsensusTimestamp()
@@ -209,6 +212,10 @@ private StateChangeContext createStateChangeContext() {
             return parent.getStateChangeContext();
         }
 
+        if (trigger != null) {
+            return trigger.getStateChangeContext();
+        }
+
         return !CollectionUtils.isEmpty(stateChanges)
                 ? new StateChangeContext(stateChanges)
                 : StateChangeContext.EMPTY_CONTEXT;

importer/src/main/java/org/hiero/mirror/importer/reader/block/BlockStreamReaderImpl.java

@@ -14,22 +14,27 @@
 import static com.hedera.hapi.block.stream.protoc.BlockItem.ItemCase.TRANSACTION_RESULT;
 
 import com.google.protobuf.InvalidProtocolBufferException;
+import com.hedera.hapi.block.stream.output.protoc.CreateScheduleOutput;
+import com.hedera.hapi.block.stream.output.protoc.SignScheduleOutput;
 import com.hedera.hapi.block.stream.output.protoc.StateChanges;
 import com.hedera.hapi.block.stream.output.protoc.TransactionOutput;
+import com.hedera.hapi.block.stream.output.protoc.TransactionOutput.TransactionCase;
 import com.hedera.hapi.block.stream.protoc.BlockItem;
 import com.hedera.hapi.block.stream.trace.protoc.TraceData;
 import com.hederahashgraph.api.proto.java.AtomicBatchTransactionBody;
 import com.hederahashgraph.api.proto.java.BlockHashAlgorithm;
 import com.hederahashgraph.api.proto.java.SignedTransaction;
 import com.hederahashgraph.api.proto.java.TransactionBody;
+import com.hederahashgraph.api.proto.java.TransactionID;
 import jakarta.annotation.Nonnull;
 import jakarta.inject.Named;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.EnumMap;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
-import lombok.CustomLog;
 import lombok.Setter;
 import lombok.Value;
 import lombok.experimental.NonFinal;
@@ -39,7 +44,6 @@
 import org.hiero.mirror.common.util.DomainUtils;
 import org.hiero.mirror.importer.exception.InvalidStreamFileException;
 
-@CustomLog
 @Named
 public final class BlockStreamReaderImpl implements BlockStreamReader {
 
@@ -137,8 +141,7 @@
                             "Missing transaction result in block " + context.getFilename());
                 }
 
-                var transactionOutputs = new EnumMap<TransactionOutput.TransactionCase, TransactionOutput>(
-                        TransactionOutput.TransactionCase.class);
+                var transactionOutputs = new EnumMap<TransactionCase, TransactionOutput>(TransactionCase.class);
                 while ((protoBlockItem = context.readBlockItemFor(TRANSACTION_OUTPUT)) != null) {
                     var transactionOutput = protoBlockItem.getTransactionOutput();
                     transactionOutputs.put(transactionOutput.getTransactionCase(), transactionOutput);
@@ -165,13 +168,14 @@
 
                 var blockTransaction = BlockTransaction.builder()
                         .previous(context.getLastBlockTransaction())
+                        .signedTransaction(signedTransaction)
+                        .signedTransactionBytes(signedTransactionInfo.signedTransaction())
                         .stateChanges(Collections.unmodifiableList(stateChangesList))
                         .traceData(Collections.unmodifiableList(traceDataList))
                         .transactionBody(transactionBody)
                         .transactionResult(transactionResult)
                         .transactionOutputs(Collections.unmodifiableMap(transactionOutputs))
-                        .signedTransaction(signedTransaction)
-                        .signedTransactionBytes(signedTransactionInfo.signedTransaction())
+                        .trigger(context.getScheduledTransactionTriggers().get(transactionBody.getTransactionID()))
                         .build();
                 context.getBlockFile().item(blockTransaction);
                 context.setLastBlockTransaction(blockTransaction, signedTransactionInfo.userTransactionInBatch());
@@ -211,6 +215,7 @@
 
     @Value
     private static class ReaderContext {
+
         private BlockFile.BlockFileBuilder blockFile;
         private List<BlockItem> blockItems;
         private BlockRootHashDigest blockRootHashDigest;
@@ -238,6 +243,8 @@
         @Setter
         private Long lastMetaTimestamp; // The last consensus timestamp from metadata
 
+        private Map<TransactionID, BlockTransaction> scheduledTransactionTriggers = new HashMap<>();
+
         ReaderContext(@Nonnull List<BlockItem> blockItems, @Nonnull String filename) {
             this.blockFile = BlockFile.builder();
             this.blockItems = blockItems;
@@ -280,7 +287,7 @@
             return blockItem;
         }
 
-        void setLastBlockTransaction(BlockTransaction lastBlockTransaction, boolean userTransactionInBatch) {
+        void setLastBlockTransaction(@Nonnull BlockTransaction lastBlockTransaction, boolean userTransactionInBatch) {
             if (userTransactionInBatch) {
                 if (lastUserTransactionInBatch != null
                         && batchBody != null
@@ -296,12 +303,27 @@
             }
 
             this.lastBlockTransaction = lastBlockTransaction;
-            if (lastBlockTransaction != null
-                    && lastBlockTransaction.getTransactionBody().hasAtomicBatch()) {
+            if (lastBlockTransaction.getTransactionBody().hasAtomicBatch()) {
                 this.batchIndex = 0;
                 this.batchBody = lastBlockTransaction.getTransactionBody().getAtomicBatch();
                 this.lastUserTransactionInBatch = null;
             }
+
+            // A short-term scheduled transaction and its triggering transaction (either a schedule create or a schedule
+            // sign) belong to one transactional unit, thus the statechanges are attached to the triggering transaction.
+            // Build the map to link a short-term scheduled transaction to its trigger
+            lastBlockTransaction
+                    .getTransactionOutput(TransactionCase.CREATE_SCHEDULE)
+                    .map(TransactionOutput::getCreateSchedule)
+                    .filter(CreateScheduleOutput::hasScheduledTransactionId)
+                    .map(CreateScheduleOutput::getScheduledTransactionId)
+                    .or(() -> lastBlockTransaction
+                            .getTransactionOutput(TransactionCase.SIGN_SCHEDULE)
+                            .map(TransactionOutput::getSignSchedule)
+                            .filter(SignScheduleOutput::hasScheduledTransactionId)
+                            .map(SignScheduleOutput::getScheduledTransactionId))
+                    .ifPresent(scheduledTransactionId ->
+                            scheduledTransactionTriggers.put(scheduledTransactionId, lastBlockTransaction));
         }
     }
 

importer/src/test/java/org/hiero/mirror/importer/downloader/block/transformer/ConsensusTransformerTest.java

@@ -14,8 +14,10 @@
 import com.hedera.hapi.block.stream.trace.protoc.TraceData;
 import com.hederahashgraph.api.proto.java.ResponseCodeEnum;
 import com.hederahashgraph.api.proto.java.Topic;
+import com.hederahashgraph.api.proto.java.TransactionID;
 import com.hederahashgraph.api.proto.java.TransactionReceipt;
 import java.util.List;
+import org.hiero.mirror.common.domain.transaction.RecordItem;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
@@ -213,4 +215,67 @@
         // then
         assertRecordFile(recordFile, blockFile, items -> assertThat(items).containsExactly(expectedRecordItem));
     }
+
+    @Test
+    void scheduledConsensusSubmitMessageTransform() {
+        // given
+        var scheduledTransactionId = TransactionID.newBuilder()
+                .setAccountID(recordItemBuilder.accountId())
+                .setScheduled(true)
+                .setTransactionValidStart(recordItemBuilder.timestamp())
+                .build();
+        var topicId = recordItemBuilder.topicId();
+        var runningHash = recordItemBuilder.bytes(48);
+        var scheduleSignRecordItem = recordItemBuilder
+                .scheduleCreate()
+                .receipt(r -> r.setScheduledTransactionID(scheduledTransactionId))
+                .customize(this::finalize)
+                .build();
+        var scheduleSignBlockTransaction = blockTransactionBuilder
+                .scheduleCreate(scheduleSignRecordItem)
+                .stateChanges(s -> {
+                    var stateChanges = s.getFirst().toBuilder()
+                            .addStateChanges(StateChange.newBuilder()
+                                    .setStateId(StateIdentifier.STATE_ID_TOPICS_VALUE)
+                                    .setMapUpdate(MapUpdateChange.newBuilder()
+                                            .setKey(MapChangeKey.newBuilder().setTopicIdKey(topicId))
+                                            .setValue(MapChangeValue.newBuilder()
+                                                    .setTopicValue(Topic.newBuilder()
+                                                            .setTopicId(topicId)
+                                                            .setRunningHash(runningHash)
+                                                            .setSequenceNumber(10)))))
+                            .build();
+                    s.clear();
+                    s.add(stateChanges);
+                })
+                .build();
+        var consensusSubmitMessageRecordItem = recordItemBuilder
+                .consensusSubmitMessage()
+                .clearIncrementer()
+                .receipt(r -> r.setTopicRunningHash(runningHash).setTopicSequenceNumber(10))
+                .transactionBody(b -> b.setTopicID(topicId))
+                .transactionBodyWrapper(w -> w.setTransactionID(scheduledTransactionId))
+                .recordItem(r -> r.transactionIndex(1))
+                .customize(this::finalize)
+                .build();
+        var consensusSubmitMessageBlockTransaction = blockTransactionBuilder
+                .consensusSubmitMessage(consensusSubmitMessageRecordItem)
+                .previous(scheduleSignBlockTransaction)
+                .stateChanges(List::clear)
+                .trigger(scheduleSignBlockTransaction)
+                .build();
+        var blockFile = blockFileBuilder
+                .items(List.of(scheduleSignBlockTransaction, consensusSubmitMessageBlockTransaction))
+                .build();
+
+        // when
+        var recordFile = blockFileTransformer.transform(blockFile);
+
+        // then
+        var expected = List.of(scheduleSignRecordItem, consensusSubmitMessageRecordItem);
+        assertRecordFile(recordFile, blockFile, items -> {
+            assertRecordItems(items, expected);
+            assertThat(items).map(RecordItem::getParent).containsOnlyNulls();
+        });
+    }
 }

importer/src/test/java/org/hiero/mirror/importer/parser/domain/BlockTransactionBuilder.java

@@ -894,13 +894,15 @@
     }
 
     public static class Builder {
+
+        private BlockTransaction previous;
         private final SignedTransaction signedTransaction;
         private final byte[] signedTransactionBytes;
-        private final Map<TransactionCase, TransactionOutput> transactionOutputs;
-        private final TransactionResult.Builder transactionResultBuilder;
         private final List<StateChanges> stateChanges;
         private final List<TraceData> traceDataList;
-        private BlockTransaction previous;
+        private final Map<TransactionCase, TransactionOutput> transactionOutputs;
+        private final TransactionResult.Builder transactionResultBuilder;
+        private BlockTransaction trigger;
 
         @SneakyThrows
         @SuppressWarnings({"java:S1640", "deprecation"})
@@ -941,6 +943,7 @@
                     .transactionBody(TransactionBody.parseFrom(signedTransaction.getBodyBytes()))
                     .transactionResult(transactionResultBuilder.build())
                     .transactionOutputs(transactionOutputs)
+                    .trigger(trigger)
                     .build();
         }
 
@@ -968,6 +971,11 @@
             consumer.accept(transactionResultBuilder);
             return this;
         }
+
+        public Builder trigger(BlockTransaction trigger) {
+            this.trigger = trigger;
+            return this;
+        }
     }
 
     private record EvmTraceDataBuilder(

importer/src/test/java/org/hiero/mirror/importer/parser/domain/RecordItemBuilder.java

@@ -1472,7 +1472,6 @@ public class Builder<T extends GeneratedMessage.Builder<T>> {
 
         private static final BiConsumer<TransactionBody.Builder, TransactionRecord.Builder> NOOP_INCREMENTER =
                 (b, r) -> {};
-
         private final TransactionType type;
         private final T transactionBody;
         private final SignatureMap.Builder signatureMap;
@@ -1552,6 +1551,11 @@ public RecordItem build() {
                     .build();
         }
 
+        public Builder<T> clearIncrementer() {
+            this.incrementer = NOOP_INCREMENTER;
+            return this;
+        }
+
         public Builder<T> customize(Consumer<Builder<T>> consumer) {
             consumer.accept(this);
             return this;
@@ -1562,11 +1566,6 @@ public Builder<T> entityTransactionPredicate(Predicate<EntityId> entityTransacti
             return this;
         }
 
-        public Builder<T> clearIncrementer() {
-            this.incrementer = NOOP_INCREMENTER;
-            return this;
-        }
-
         public Builder<T> contractTransactionPredicate(Predicate<EntityId> contractTransactionPredicate) {
             this.contractTransactionPredicate = contractTransactionPredicate;
             return this;