[Exec] Adding collection result consumer type (#4049)

Author: ramtinms

engine/execution/computation/computer/computer.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/onflow/flow-go/crypto/hash"
 	"github.com/onflow/flow-go/engine/execution"
+	"github.com/onflow/flow-go/engine/execution/computation/result"
 	"github.com/onflow/flow-go/engine/execution/state/delta"
 	"github.com/onflow/flow-go/engine/execution/utils"
 	"github.com/onflow/flow-go/fvm"
@@ -126,6 +127,7 @@ type blockComputer struct {
 	signer                module.Local
 	spockHasher           hash.Hasher
 	receiptHasher         hash.Hasher
+	colResCons            []result.ExecutedCollectionConsumer
 }
 
 func SystemChunkContext(vmCtx fvm.Context, logger zerolog.Logger) fvm.Context {
@@ -152,6 +154,7 @@ func NewBlockComputer(
 	committer ViewCommitter,
 	signer module.Local,
 	executionDataProvider *provider.Provider,
+	colResCons []result.ExecutedCollectionConsumer,
 ) (BlockComputer, error) {
 	systemChunkCtx := SystemChunkContext(vmCtx, logger)
 	vmCtx = fvm.NewContextFromParent(
@@ -170,6 +173,7 @@ func NewBlockComputer(
 		signer:                signer,
 		spockHasher:           utils.NewSPOCKHasher(),
 		receiptHasher:         utils.NewExecutionReceiptHasher(),
+		colResCons:            colResCons,
 	}, nil
 }
 
@@ -303,7 +307,8 @@ func (e *blockComputer) executeBlock(
 		e.receiptHasher,
 		parentBlockExecutionResultID,
 		block,
-		len(transactions))
+		len(transactions),
+		e.colResCons)
 	defer collector.Stop()
 
 	stateView := delta.NewDeltaView(snapshot)

engine/execution/computation/computer/computer_test.go

@@ -181,7 +181,8 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 			zerolog.Nop(),
 			committer,
 			me,
-			prov)
+			prov,
+			nil)
 		require.NoError(t, err)
 
 		// create a block with 1 collection with 2 transactions
@@ -311,7 +312,8 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 			zerolog.Nop(),
 			committer,
 			me,
-			prov)
+			prov,
+			nil)
 		require.NoError(t, err)
 
 		// create an empty block
@@ -401,7 +403,8 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 			zerolog.Nop(),
 			comm,
 			me,
-			prov)
+			prov,
+			nil)
 		require.NoError(t, err)
 
 		// create an empty block
@@ -450,7 +453,8 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 			zerolog.Nop(),
 			committer,
 			me,
-			prov)
+			prov,
+			nil)
 		require.NoError(t, err)
 
 		collectionCount := 2
@@ -636,7 +640,8 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 			zerolog.Nop(),
 			committer.NewNoopViewCommitter(),
 			me,
-			prov)
+			prov,
+			nil)
 		require.NoError(t, err)
 
 		result, err := exe.ExecuteBlock(
@@ -722,7 +727,8 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 			zerolog.Nop(),
 			committer.NewNoopViewCommitter(),
 			me,
-			prov)
+			prov,
+			nil)
 		require.NoError(t, err)
 
 		const collectionCount = 2
@@ -823,7 +829,8 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 			zerolog.Nop(),
 			committer.NewNoopViewCommitter(),
 			me,
-			prov)
+			prov,
+			nil)
 		require.NoError(t, err)
 
 		block := generateBlock(collectionCount, transactionCount, rag)
@@ -1045,7 +1052,8 @@ func Test_AccountStatusRegistersAreIncluded(t *testing.T) {
 		zerolog.Nop(),
 		committer.NewNoopViewCommitter(),
 		me,
-		prov)
+		prov,
+		nil)
 	require.NoError(t, err)
 
 	block := generateBlockWithVisitor(1, 1, fag, func(txBody *flow.TransactionBody) {
@@ -1156,7 +1164,8 @@ func Test_ExecutingSystemCollection(t *testing.T) {
 		zerolog.Nop(),
 		committer,
 		me,
-		prov)
+		prov,
+		nil)
 	require.NoError(t, err)
 
 	// create empty block, it will have system collection attached while executing

engine/execution/computation/computer/result_collector.go

@@ -11,6 +11,7 @@ import (
 	"github.com/onflow/flow-go/crypto"
 	"github.com/onflow/flow-go/crypto/hash"
 	"github.com/onflow/flow-go/engine/execution"
+	"github.com/onflow/flow-go/engine/execution/computation/result"
 	"github.com/onflow/flow-go/engine/execution/state/delta"
 	"github.com/onflow/flow-go/fvm/state"
 	"github.com/onflow/flow-go/ledger"
@@ -41,6 +42,7 @@ type transactionResult struct {
 	*state.ExecutionSnapshot
 }
 
+// TODO(ramtin): move committer and other folks to consumers layer
 type resultCollector struct {
 	tracer    module.Tracer
 	blockSpan otelTrace.Span
@@ -62,7 +64,8 @@ type resultCollector struct {
 
 	parentBlockExecutionResultID flow.Identifier
 
-	result *execution.ComputationResult
+	result    *execution.ComputationResult
+	consumers []result.ExecutedCollectionConsumer
 
 	chunks                 []*flow.Chunk
 	spockSignatures        []crypto.Signature
@@ -88,6 +91,7 @@ func newResultCollector(
 	parentBlockExecutionResultID flow.Identifier,
 	block *entity.ExecutableBlock,
 	numTransactions int,
+	consumers []result.ExecutedCollectionConsumer,
 ) *resultCollector {
 	numCollections := len(block.Collections()) + 1
 	now := time.Now()
@@ -104,6 +108,7 @@ func newResultCollector(
 		executionDataProvider:        executionDataProvider,
 		parentBlockExecutionResultID: parentBlockExecutionResultID,
 		result:                       execution.NewEmptyComputationResult(block),
+		consumers:                    consumers,
 		chunks:                       make([]*flow.Chunk, 0, numCollections),
 		spockSignatures:              make([]crypto.Signature, 0, numCollections),
 		blockStartTime:               now,
@@ -226,6 +231,13 @@ func (collector *resultCollector) commitCollection(
 		NumberOfCollections: 1,
 	}
 
+	for _, consumer := range collector.consumers {
+		err = consumer.OnExecutedCollection(collector.result.CollectionResult(collection.collectionIndex))
+		if err != nil {
+			return fmt.Errorf("consumer failed: %w", err)
+		}
+	}
+
 	return nil
 }
 

engine/execution/computation/execution_verification_test.go

@@ -701,7 +701,8 @@ func executeBlockAndVerifyWithParameters(t *testing.T,
 		logger,
 		ledgerCommiter,
 		me,
-		prov)
+		prov,
+		nil)
 	require.NoError(t, err)
 
 	executableBlock := unittest.ExecutableBlockFromTransactions(chain.ChainID(), txs)

engine/execution/computation/manager.go

@@ -130,6 +130,7 @@ func New(
 		committer,
 		me,
 		executionDataProvider,
+		nil, // TODO(ramtin): update me with proper consumers
 	)
 
 	if err != nil {

engine/execution/computation/manager_benchmark_test.go

@@ -153,7 +153,8 @@ func BenchmarkComputeBlock(b *testing.B) {
 		zerolog.Nop(),
 		committer.NewNoopViewCommitter(),
 		me,
-		prov)
+		prov,
+		nil)
 	require.NoError(b, err)
 
 	derivedChainData, err := derived.NewDerivedChainData(

engine/execution/computation/manager_test.go

@@ -139,7 +139,8 @@ func TestComputeBlockWithStorage(t *testing.T) {
 		zerolog.Nop(),
 		committer.NewNoopViewCommitter(),
 		me,
-		prov)
+		prov,
+		nil)
 	require.NoError(t, err)
 
 	derivedChainData, err := derived.NewDerivedChainData(10)
@@ -777,6 +778,7 @@ func Test_EventEncodingFailsOnlyTxAndCarriesOn(t *testing.T) {
 		committer.NewNoopViewCommitter(),
 		me,
 		prov,
+		nil,
 	)
 	require.NoError(t, err)
 

engine/execution/computation/programs_test.go

@@ -135,7 +135,8 @@ func TestPrograms_TestContractUpdates(t *testing.T) {
 		zerolog.Nop(),
 		committer.NewNoopViewCommitter(),
 		me,
-		prov)
+		prov,
+		nil)
 	require.NoError(t, err)
 
 	derivedChainData, err := derived.NewDerivedChainData(10)
@@ -246,7 +247,8 @@ func TestPrograms_TestBlockForks(t *testing.T) {
 		zerolog.Nop(),
 		committer.NewNoopViewCommitter(),
 		me,
-		prov)
+		prov,
+		nil)
 	require.NoError(t, err)
 
 	derivedChainData, err := derived.NewDerivedChainData(10)

engine/execution/computation/result/consumer.go

@@ -0,0 +1,31 @@
+package result
+
+import (
+	"github.com/onflow/flow-go/model/flow"
+)
+
+// ExecutedCollection holds results of a collection execution
+type ExecutedCollection interface {
+	// BlockHeader returns the block header in which collection was included
+	BlockHeader() *flow.Header
+
+	// Collection returns the content of the collection
+	Collection() *flow.Collection
+
+	// RegisterUpdates returns all registers that were updated during collection execution
+	UpdatedRegisters() flow.RegisterEntries
+
+	// ReadRegisterIDs returns all registers that has been read during collection execution
+	ReadRegisterIDs() flow.RegisterIDs
+
+	// EmittedEvents returns a list of events emitted during collection execution
+	EmittedEvents() flow.EventsList
+
+	// TransactionResults returns a list of transaction results
+	TransactionResults() flow.TransactionResults
+}
+
+// ExecutedCollectionConsumer consumes ExecutedCollections
+type ExecutedCollectionConsumer interface {
+	OnExecutedCollection(ec ExecutedCollection) error
+}

engine/execution/messages.go

@@ -52,3 +52,61 @@ func NewEmptyComputationResult(
 		},
 	}
 }
+
+func (cr ComputationResult) transactionResultsByCollectionIndex(colIndex int) []flow.TransactionResult {
+	var startTxnIndex int
+	if colIndex > 0 {
+		startTxnIndex = cr.TransactionResultIndex[colIndex-1]
+	}
+	endTxnIndex := cr.TransactionResultIndex[colIndex]
+	return cr.TransactionResults[startTxnIndex:endTxnIndex]
+}
+
+func (cr *ComputationResult) CollectionResult(colIndex int) *ColResSnapshot {
+	if colIndex < 0 && colIndex > len(cr.CompleteCollections) {
+		return nil
+	}
+	return &ColResSnapshot{
+		blockHeader: cr.Block.Header,
+		collection: &flow.Collection{
+			Transactions: cr.CollectionAt(colIndex).Transactions,
+		},
+		updatedRegisters:   cr.StateSnapshots[colIndex].UpdatedRegisters(),
+		readRegisterIDs:    cr.StateSnapshots[colIndex].ReadRegisterIDs(),
+		emittedEvents:      cr.Events[colIndex],
+		transactionResults: cr.transactionResultsByCollectionIndex(colIndex),
+	}
+}
+
+type ColResSnapshot struct {
+	blockHeader        *flow.Header
+	collection         *flow.Collection
+	updatedRegisters   flow.RegisterEntries
+	readRegisterIDs    flow.RegisterIDs
+	emittedEvents      flow.EventsList
+	transactionResults flow.TransactionResults
+}
+
+func (c *ColResSnapshot) BlockHeader() *flow.Header {
+	return c.blockHeader
+}
+
+func (c *ColResSnapshot) Collection() *flow.Collection {
+	return c.collection
+}
+
+func (c *ColResSnapshot) UpdatedRegisters() flow.RegisterEntries {
+	return c.updatedRegisters
+}
+
+func (c *ColResSnapshot) ReadRegisterIDs() flow.RegisterIDs {
+	return c.readRegisterIDs
+}
+
+func (c *ColResSnapshot) EmittedEvents() flow.EventsList {
+	return c.emittedEvents
+}
+
+func (c *ColResSnapshot) TransactionResults() flow.TransactionResults {
+	return c.transactionResults
+}