Merge branch 'master' into gregor/callbacks/process-fallback

Author: devbugging

engine/execution/computation/computer/computer.go

@@ -38,10 +38,16 @@ type collectionInfo struct {
 
 	collectionIndex int
 	*entity.CompleteCollection
-
-	isSystemTransaction bool
 }
 
+type ComputerTransactionType uint
+
+const (
+	ComputerTransactionTypeUser ComputerTransactionType = iota
+	ComputerTransactionTypeSystem
+	ComputerTransactionTypeScheduled
+)
+
 type TransactionRequest struct {
 	collectionInfo
 
@@ -50,6 +56,7 @@ type TransactionRequest struct {
 
 	txnIndex uint32
 
+	transactionType             ComputerTransactionType
 	lastTransactionInCollection bool
 
 	ctx fvm.Context
@@ -62,6 +69,7 @@ func newTransactionRequest(
 	collectionLogger zerolog.Logger,
 	txnIndex uint32,
 	txnBody *flow.TransactionBody,
+	transactionType ComputerTransactionType,
 	lastTransactionInCollection bool,
 ) TransactionRequest {
 	txnId := txnBody.ID()
@@ -83,6 +91,7 @@ func newTransactionRequest(
 			txnId,
 			txnIndex,
 			txnBody),
+		transactionType:             transactionType,
 		lastTransactionInCollection: lastTransactionInCollection,
 	}
 }
@@ -281,12 +290,11 @@ func (e *blockComputer) queueUserTransactions(
 			Logger()
 
 		collectionInfo := collectionInfo{
-			blockId:             blockId,
-			blockIdStr:          blockIdStr,
-			blockHeight:         blockHeader.Height,
-			collectionIndex:     idx,
-			CompleteCollection:  collection,
-			isSystemTransaction: false,
+			blockId:            blockId,
+			blockIdStr:         blockIdStr,
+			blockHeight:        blockHeader.Height,
+			collectionIndex:    idx,
+			CompleteCollection: collection,
 		}
 
 		for i, txnBody := range collection.Collection.Transactions {
@@ -296,7 +304,9 @@ func (e *blockComputer) queueUserTransactions(
 				collectionLogger,
 				txnIndex,
 				txnBody,
-				i == len(collection.Collection.Transactions)-1)
+				ComputerTransactionTypeUser,
+				i == len(collection.Collection.Transactions)-1,
+			)
 			txnIndex += 1
 		}
 	}
@@ -307,49 +317,50 @@ func (e *blockComputer) queueUserTransactions(
 func (e *blockComputer) queueSystemTransactions(
 	callbackCtx fvm.Context,
 	systemChunkCtx fvm.Context,
-	systemColection collectionInfo,
+	systemCollection collectionInfo,
 	systemTxn *flow.TransactionBody,
 	executeCallbackTxs []*flow.TransactionBody,
 	txnIndex uint32,
 	systemLogger zerolog.Logger,
 ) chan TransactionRequest {
-	var logger zerolog.Logger
 
-	systemChunkTxLogger := systemLogger.With().
-		Uint32("num_txs", uint32(len(systemColection.CompleteCollection.Collection.Transactions))).
+	systemTxLogger := systemLogger.With().
+		Uint32("num_txs", uint32(len(systemCollection.CompleteCollection.Collection.Transactions))).
+		Bool("system_transaction", true).
 		Logger()
 
-	scheduledTxLogger := systemChunkTxLogger.With().
+	scheduledTxLogger := systemLogger.With().
+		Uint32("num_txs", uint32(len(systemCollection.CompleteCollection.Collection.Transactions))).
 		Bool("scheduled_transaction", true).
-		Bool("critical_error", false).
 		Logger()
 
-	allTxs := append(executeCallbackTxs, systemTxn)
-	txQueue := make(chan TransactionRequest, len(allTxs))
+	txQueue := make(chan TransactionRequest, len(executeCallbackTxs)+1)
 	defer close(txQueue)
 
-	for i, txBody := range allTxs {
-		systemChunkTx := i == len(allTxs)-1 // last tx in collection is system chunk
-		ctx := callbackCtx
-		logger = scheduledTxLogger
-
-		if systemChunkTx {
-			ctx = systemChunkCtx
-			logger = systemChunkTxLogger
-		}
-
+	for _, txBody := range executeCallbackTxs {
 		txQueue <- newTransactionRequest(
-			systemColection,
-			ctx,
-			logger,
+			systemCollection,
+			callbackCtx,
+			scheduledTxLogger,
 			txnIndex,
 			txBody,
-			systemChunkTx,
+			ComputerTransactionTypeScheduled,
+			false,
 		)
 
 		txnIndex++
 	}
 
+	txQueue <- newTransactionRequest(
+		systemCollection,
+		systemChunkCtx,
+		systemTxLogger,
+		txnIndex,
+		systemTxn,
+		ComputerTransactionTypeSystem,
+		true,
+	)
+
 	return txQueue
 }
 
@@ -495,20 +506,17 @@ func (e *blockComputer) executeSystemTransactions(
 	)
 
 	systemCollectionInfo := collectionInfo{
-		blockId:             block.BlockID(),
-		blockIdStr:          block.BlockID().String(),
-		blockHeight:         block.Block.Height,
-		collectionIndex:     userCollectionCount,
-		CompleteCollection:  nil, // We do not yet know all the scheduled callbacks, so postpone construction of the collection.
-		isSystemTransaction: true,
+		blockId:            block.BlockID(),
+		blockIdStr:         block.BlockID().String(),
+		blockHeight:        block.Block.Height,
+		collectionIndex:    userCollectionCount,
+		CompleteCollection: nil, // We do not yet know all the scheduled callbacks, so postpone construction of the collection.
 	}
 
-	systemLogger := callbackCtx.Logger.With().
+	systemChunkLogger := callbackCtx.Logger.With().
 		Str("block_id", block.BlockID().String()).
 		Uint64("height", block.Block.Height).
 		Bool("system_chunk", true).
-		Bool("system_transaction", true).
-		Bool("critical_error", true).
 		Int("num_collections", userCollectionCount).
 		Logger()
 
@@ -528,7 +536,7 @@ func (e *blockComputer) executeSystemTransactions(
 			database,
 			blockSpan,
 			txIndex,
-			systemLogger,
+			systemChunkLogger,
 		)
 		if err != nil {
 			return err
@@ -565,7 +573,7 @@ func (e *blockComputer) executeSystemTransactions(
 		e.systemTxn,
 		callbackTxs,
 		txIndex,
-		systemLogger,
+		systemChunkLogger,
 	)
 
 	e.executeQueue(blockSpan, database, txQueue)
@@ -616,7 +624,7 @@ func (e *blockComputer) executeProcessCallback(
 	systemLogger zerolog.Logger,
 ) ([]*flow.TransactionBody, uint32, error) {
 	callbackLogger := systemLogger.With().
-		Bool("scheduled_transaction", true).
+		Bool("system_transaction", true).
 		Logger()
 
 	request := newTransactionRequest(
@@ -625,6 +633,7 @@ func (e *blockComputer) executeProcessCallback(
 		callbackLogger,
 		txnIndex,
 		e.processCallbackTxn,
+		ComputerTransactionTypeSystem,
 		false)
 
 	txnIndex++
@@ -730,7 +739,7 @@ func (e *blockComputer) executeTransaction(
 		attempt)
 	if err != nil {
 		prefix := ""
-		if request.isSystemTransaction {
+		if request.transactionType == ComputerTransactionTypeSystem {
 			prefix = "system "
 		}
 

engine/execution/computation/computer/computer_test.go

@@ -588,6 +588,7 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 		assert.LessOrEqual(t, vm.CallCount(), (1+totalTransactionCount)/2*totalTransactionCount)
 	})
 
+	// TODO: this test is flaky with a low probability of failing
 	t.Run(
 		"service events are emitted", func(t *testing.T) {
 			execCtx := fvm.NewContext(
@@ -1524,15 +1525,15 @@ func testScheduledCallbackWithError(
 		exemetrics.On("ExecutionTransactionExecuted",
 			mock.Anything,
 			mock.MatchedBy(func(arg module.TransactionExecutionResultStats) bool {
-				return arg.Failed && arg.SystemTransaction
+				return arg.Failed && (arg.SystemTransaction || arg.ScheduledTransaction)
 			}),
 			mock.Anything).
 			Return(nil).
 			Times(1)
 		exemetrics.On("ExecutionTransactionExecuted",
 			mock.Anything,
 			mock.MatchedBy(func(arg module.TransactionExecutionResultStats) bool {
-				return !arg.Failed && arg.SystemTransaction
+				return !arg.Failed && (arg.SystemTransaction || arg.ScheduledTransaction)
 			}),
 			mock.Anything).
 			Return(nil).
@@ -1541,7 +1542,7 @@ func testScheduledCallbackWithError(
 		exemetrics.On("ExecutionTransactionExecuted",
 			mock.Anything,
 			mock.MatchedBy(func(arg module.TransactionExecutionResultStats) bool {
-				return !arg.Failed && arg.SystemTransaction
+				return !arg.Failed && (arg.SystemTransaction || arg.ScheduledTransaction)
 			}),
 			mock.Anything).
 			Return(nil).

engine/execution/computation/computer/result_collector.go

@@ -240,14 +240,11 @@ func (collector *resultCollector) processTransactionResult(
 			Logger()
 		logger.Info().Msg("transaction execution failed")
 
-		if txn.isSystemTransaction {
+		if txn.transactionType == ComputerTransactionTypeSystem {
 			// This log is used as the data source for an alert on grafana.
-			// The system_chunk_error field must not be changed without adding
+			// The critical_error field must not be changed without adding
 			// the corresponding changes in grafana.
-			// https://github.com/dapperlabs/flow-internal/issues/1546
 			logger.Error().
-				Bool("system_chunk_error", true).
-				Bool("system_transaction_error", true).
 				Bool("critical_error", true).
 				Msg("error executing system chunk transaction")
 		}
@@ -314,7 +311,8 @@ func (collector *resultCollector) handleTransactionExecutionMetrics(
 		ComputationIntensities:     output.ComputationIntensities,
 		NumberOfTxnConflictRetries: numConflictRetries,
 		Failed:                     output.Err != nil,
-		SystemTransaction:          txn.isSystemTransaction,
+		ScheduledTransaction:       txn.transactionType == ComputerTransactionTypeScheduled,
+		SystemTransaction:          txn.transactionType == ComputerTransactionTypeSystem,
 	}
 	for _, entry := range txnExecutionSnapshot.UpdatedRegisters() {
 		transactionExecutionStats.NumberOfBytesWrittenToRegisters += len(entry.Value)

engine/execution/computation/computer/transaction_coordinator_test.go

@@ -122,6 +122,7 @@ func (db *testCoordinator) newTransaction(txnIndex uint32) (
 			zerolog.Nop(),
 			txnIndex,
 			&flow.TransactionBody{},
+			ComputerTransactionTypeUser,
 			false),
 		0)
 }