Merge pull request #7579 from devbugging/gregor/callbacks/04-contract-integration

[Scheduled Callbacks] Integrate callback scheduler contract

Author: janezpodhostnik

engine/execution/computation/computer/computer.go

@@ -111,6 +111,7 @@ type blockComputer struct {
 	tracer                module.Tracer
 	log                   zerolog.Logger
 	systemTxn             *flow.TransactionBody
+	processCallbackTxn    *flow.TransactionBody
 	committer             ViewCommitter
 	executionDataProvider provider.Provider
 	signer                module.Local
@@ -184,6 +185,8 @@ func NewBlockComputer(
 		return nil, fmt.Errorf("could not build system chunk transaction: %w", err)
 	}
 
+	processCallbackTxn := blueprints.ProcessCallbacksTransaction(vmCtx.Chain)
+
 	return &blockComputer{
 		vm:                    vm,
 		vmCtx:                 vmCtx,
@@ -193,6 +196,7 @@ func NewBlockComputer(
 		tracer:                tracer,
 		log:                   logger,
 		systemTxn:             systemTxn,
+		processCallbackTxn:    processCallbackTxn,
 		committer:             committer,
 		executionDataProvider: executionDataProvider,
 		signer:                signer,
@@ -241,8 +245,11 @@ func (e *blockComputer) queueUserTransactions(
 	blockId flow.Identifier,
 	blockHeader *flow.Header,
 	rawCollections []*entity.CompleteCollection,
-	requestQueue chan TransactionRequest,
-) {
+	userTxCount int,
+) chan TransactionRequest {
+	txQueue := make(chan TransactionRequest, userTxCount)
+	defer close(txQueue)
+
 	txnIndex := uint32(0)
 	blockIdStr := blockId.String()
 
@@ -270,7 +277,7 @@ func (e *blockComputer) queueUserTransactions(
 		}
 
 		for i, txnBody := range collection.Transactions {
-			requestQueue <- newTransactionRequest(
+			txQueue <- newTransactionRequest(
 				collectionInfo,
 				collectionCtx,
 				collectionLogger,
@@ -280,6 +287,8 @@ func (e *blockComputer) queueUserTransactions(
 			txnIndex += 1
 		}
 	}
+
+	return txQueue
 }
 
 func (e *blockComputer) queueSystemTransactions(
@@ -288,16 +297,18 @@ func (e *blockComputer) queueSystemTransactions(
 	systemColection collectionInfo,
 	systemTxn *flow.TransactionBody,
 	executeCallbackTxs []*flow.TransactionBody,
-	requestQueue chan TransactionRequest,
 	txnIndex uint32,
 	systemLogger zerolog.Logger,
-) {
+) chan TransactionRequest {
 	allTxs := append(executeCallbackTxs, systemTxn)
 	// add execute callback transactions to the system collection info along to existing process transaction
 	systemTxs := systemColection.CompleteCollection.Transactions
 	systemColection.CompleteCollection.Transactions = append(systemTxs, allTxs...)
 	systemLogger = systemLogger.With().Uint32("num_txs", uint32(len(systemTxs))).Logger()
 
+	txQueue := make(chan TransactionRequest, len(allTxs))
+	defer close(txQueue)
+
 	for i, txBody := range allTxs {
 		last := i == len(allTxs)-1
 		ctx := callbackCtx
@@ -306,7 +317,7 @@ func (e *blockComputer) queueSystemTransactions(
 			ctx = systemChunkCtx
 		}
 
-		requestQueue <- newTransactionRequest(
+		txQueue <- newTransactionRequest(
 			systemColection,
 			ctx,
 			systemLogger,
@@ -317,6 +328,8 @@ func (e *blockComputer) queueSystemTransactions(
 
 		txnIndex++
 	}
+
+	return txQueue
 }
 
 func (e *blockComputer) executeBlock(
@@ -415,17 +428,13 @@ func (e *blockComputer) executeUserTransactions(
 	rawCollections []*entity.CompleteCollection,
 	userTxCount int,
 ) {
-	txQueue := make(chan TransactionRequest, userTxCount)
-
-	e.queueUserTransactions(
+	txQueue := e.queueUserTransactions(
 		block.ID(),
 		block.Block.Header,
 		rawCollections,
-		txQueue,
+		userTxCount,
 	)
 
-	close(txQueue)
-
 	e.executeQueue(blockSpan, database, txQueue)
 }
 
@@ -479,10 +488,9 @@ func (e *blockComputer) executeSystemTransactions(
 	}
 
 	var callbackTxs []*flow.TransactionBody
-	var err error
 
 	if e.vmCtx.ScheduleCallbacksEnabled {
-		callbackTxs, err = e.executeProcessCallback(
+		callbacks, updatedTxnIndex, err := e.executeProcessCallback(
 			callbackCtx,
 			systemCollectionInfo,
 			database,
@@ -494,25 +502,20 @@ func (e *blockComputer) executeSystemTransactions(
 			return err
 		}
 
-		txIndex++
+		callbackTxs = callbacks
+		txIndex = updatedTxnIndex
 	}
 
-	// queue size for callback transactions + 1 system transaction (process callback already executed)
-	txQueue := make(chan TransactionRequest, len(callbackTxs)+1)
-
-	e.queueSystemTransactions(
+	txQueue := e.queueSystemTransactions(
 		callbackCtx,
 		systemChunkCtx,
 		systemCollectionInfo,
 		e.systemTxn,
 		callbackTxs,
-		txQueue,
 		txIndex,
 		systemLogger,
 	)
 
-	close(txQueue)
-
 	e.executeQueue(blockSpan, database, txQueue)
 
 	return nil
@@ -549,28 +552,28 @@ func (e *blockComputer) executeProcessCallback(
 	blockSpan otelTrace.Span,
 	txnIndex uint32,
 	systemLogger zerolog.Logger,
-) ([]*flow.TransactionBody, error) {
-	processTxn := blueprints.ProcessCallbacksTransaction(e.vmCtx.Chain)
-
+) ([]*flow.TransactionBody, uint32, error) {
 	// add process callback transaction to the system collection info
-	systemCollectionInfo.CompleteCollection.Transactions = append(systemCollectionInfo.CompleteCollection.Transactions, processTxn)
+	systemCollectionInfo.CompleteCollection.Transactions = append(systemCollectionInfo.CompleteCollection.Transactions, e.processCallbackTxn)
 
 	request := newTransactionRequest(
 		systemCollectionInfo,
 		systemCtx,
 		systemLogger,
 		txnIndex,
-		processTxn,
+		e.processCallbackTxn,
 		false)
 
+	txnIndex++
+
 	txn, err := e.executeTransactionInternal(blockSpan, database, request, 0)
 	if err != nil {
 		snapshotTime := logical.Time(0)
 		if txn != nil {
 			snapshotTime = txn.SnapshotTime()
 		}
 
-		return nil, fmt.Errorf(
+		return nil, 0, fmt.Errorf(
 			"failed to execute %s transaction %v (%d@%d) for block %s at height %v: %w",
 			"system",
 			request.txnIdStr,
@@ -582,13 +585,18 @@ func (e *blockComputer) executeProcessCallback(
 	}
 
 	if txn.Output().Err != nil {
-		return nil, fmt.Errorf(
+		return nil, 0, fmt.Errorf(
 			"process callback transaction %s error: %v",
 			request.txnIdStr,
 			txn.Output().Err)
 	}
 
-	return blueprints.ExecuteCallbacksTransactions(e.vmCtx.Chain, txn.Output().Events)
+	callbackTxs, err := blueprints.ExecuteCallbacksTransactions(e.vmCtx.Chain, txn.Output().Events)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	return callbackTxs, txnIndex, nil
 }
 
 func (e *blockComputer) executeTransactions(

engine/execution/computation/computer/computer_test.go

@@ -1636,16 +1636,19 @@ func (executor *callbackTestExecutor) Output() fvm.ProcedureOutput {
 
 	switch {
 	// scheduled callbacks process transaction
-	case strings.Contains(string(txBody.Script), "CallbackScheduler.process"):
+	case strings.Contains(string(txBody.Script), "scheduler.process"):
 		executor.vm.executedTransactions[txID] = "process_callback"
+		env := systemcontracts.SystemContractsForChain(flow.Mainnet.Chain().ChainID()).AsTemplateEnv()
+		eventTypeString := fmt.Sprintf("A.%v.CallbackScheduler.CallbackProcessed", env.FlowCallbackSchedulerAddress)
 
 		// return events for each scheduled callback
 		events := make([]flow.Event, len(executor.vm.eventPayloads))
 		for i, payload := range executor.vm.eventPayloads {
 			events[i] = flow.Event{
 				// TODO: we shouldn't hardcode this event types, refactor after the scheduler contract is done
-				Type:             flow.EventType("A.0x0000000000000000.CallbackScheduler.CallbackProcessed"),
-				TransactionID:    txProc.ID,
+				Type:          flow.EventType(eventTypeString),
+				TransactionID: txProc.ID,
+
 				TransactionIndex: txProc.TxIndex,
 				EventIndex:       uint32(i),
 				Payload:          payload,
@@ -1656,7 +1659,7 @@ func (executor *callbackTestExecutor) Output() fvm.ProcedureOutput {
 			Events: events,
 		}
 	// scheduled callbacks execute transaction
-	case strings.Contains(string(txBody.Script), "CallbackScheduler.executeCallback"):
+	case strings.Contains(string(txBody.Script), "scheduler.executeCallback"):
 		// extract the callback ID from the arguments
 		if len(txBody.Arguments) == 0 {
 			return fvm.ProcedureOutput{}

fvm/blueprints/scheduled_callback.go

@@ -1,38 +1,23 @@
 package blueprints
 
 import (
-	_ "embed"
 	"fmt"
-	"strings"
 
 	"github.com/onflow/cadence"
 	"github.com/onflow/cadence/encoding/ccf"
+	"github.com/rs/zerolog/log"
 
+	"github.com/onflow/flow-core-contracts/lib/go/templates"
+
+	"github.com/onflow/flow-go/fvm/systemcontracts"
 	"github.com/onflow/flow-go/model/flow"
 )
 
-// processScheduledCallbacksTransaction calls scheduled callback contract
-// and process new callbacks that should be executed.
-//
-//go:embed scripts/processScheduledCallbacksTransaction.cdc
-var processCallbacksTransaction string
-
-// executeCallbacksTransaction calls scheduled callback contract
-// to execute the provided callback by ID.
-//
-//go:embed scripts/executeScheduledCallbackTransaction.cdc
-var executeCallbacksTransaction string
-
-const (
-	placeholderScheduledContract     = "import \"CallbackScheduler\""
-	processedCallbackIDFieldName     = "ID"
-	processedCallbackEffortFieldName = "executionEffort"
-	processedEventTypeTemplate       = "A.%v.CallbackScheduler.CallbackProcessed"
-	callbackTransactionGasLimit      = flow.DefaultMaxTransactionGasLimit
-)
+const callbackTransactionGasLimit = flow.DefaultMaxTransactionGasLimit
 
 func ProcessCallbacksTransaction(chain flow.Chain) *flow.TransactionBody {
-	script := prepareScheduledContractTransaction(chain, processCallbacksTransaction)
+	sc := systemcontracts.SystemContractsForChain(chain.ChainID())
+	script := templates.GenerateProcessCallbackScript(sc.AsTemplateEnv())
 
 	return flow.NewTransactionBody().
 		SetScript(script).
@@ -41,22 +26,24 @@ func ProcessCallbacksTransaction(chain flow.Chain) *flow.TransactionBody {
 
 func ExecuteCallbacksTransactions(chainID flow.Chain, processEvents flow.EventsList) ([]*flow.TransactionBody, error) {
 	txs := make([]*flow.TransactionBody, 0, len(processEvents))
+	env := systemcontracts.SystemContractsForChain(chainID.ChainID()).AsTemplateEnv()
 
 	for _, event := range processEvents {
-		id, effort, err := callbackArgsFromEvent(event)
+		id, effort, err := callbackArgsFromEvent(env, event)
 		if err != nil {
 			return nil, fmt.Errorf("failed to get callback args from event: %w", err)
 		}
 
-		tx := executeCallbackTransaction(chainID, id, effort)
+		tx := executeCallbackTransaction(env, id, effort)
 		txs = append(txs, tx)
 	}
 
 	return txs, nil
 }
 
-func executeCallbackTransaction(chain flow.Chain, id []byte, effort uint64) *flow.TransactionBody {
-	script := prepareScheduledContractTransaction(chain, executeCallbacksTransaction)
+func executeCallbackTransaction(env templates.Environment, id []byte, effort uint64) *flow.TransactionBody {
+	script := templates.GenerateExecuteCallbackScript(env)
+
 	return flow.NewTransactionBody().
 		SetScript(script).
 		AddArgument(id).
@@ -68,9 +55,17 @@ func executeCallbackTransaction(chain flow.Chain, id []byte, effort uint64) *flo
 // The event for processed callback event is emitted by the process callback transaction from
 // callback scheduler contract and has the following signature:
 // event CallbackProcessed(ID: UInt64, executionEffort: UInt64)
-func callbackArgsFromEvent(event flow.Event) ([]byte, uint64, error) {
-	scheduledContractAddress := "0x0000000000000000" // todo use contract addr
-	if string(event.Type) != fmt.Sprintf(processedEventTypeTemplate, scheduledContractAddress) {
+func callbackArgsFromEvent(env templates.Environment, event flow.Event) ([]byte, uint64, error) {
+	const (
+		processedCallbackIDFieldName     = "ID"
+		processedCallbackEffortFieldName = "executionEffort"
+		processedEventTypeTemplate       = "A.%v.CallbackScheduler.CallbackProcessed"
+	)
+
+	scheduledContractAddress := env.FlowCallbackSchedulerAddress
+	processedEventType := flow.EventType(fmt.Sprintf(processedEventTypeTemplate, scheduledContractAddress))
+
+	if event.Type != processedEventType {
 		return nil, 0, fmt.Errorf("wrong event type is passed")
 	}
 
@@ -99,29 +94,22 @@ func callbackArgsFromEvent(event flow.Event) ([]byte, uint64, error) {
 		return nil, 0, fmt.Errorf("id is not uint64")
 	}
 
-	effort, ok := effortValue.(cadence.UInt64)
+	cadenceEffort, ok := effortValue.(cadence.UInt64)
 	if !ok {
 		return nil, 0, fmt.Errorf("effort is not uint64")
 	}
 
+	effort := uint64(cadenceEffort)
+
+	if effort > flow.DefaultMaxTransactionGasLimit {
+		log.Warn().Uint64("effort", effort).Msg("effort is greater than max transaction gas limit, setting to max")
+		effort = flow.DefaultMaxTransactionGasLimit
+	}
+
 	encodedID, err := ccf.Encode(id)
 	if err != nil {
 		return nil, 0, fmt.Errorf("failed to encode id: %w", err)
 	}
 
-	return encodedID, uint64(effort), nil
-}
-
-func prepareScheduledContractTransaction(_ flow.Chain, txScript string) []byte {
-	// todo use this instead of palceholder address
-	// _ = systemcontracts.SystemContractsForChain(chain.ChainID())
-	scheduledContractAddress := "0x0000000000000000"
-
-	code := strings.ReplaceAll(
-		txScript,
-		placeholderScheduledContract,
-		fmt.Sprintf("%s from %s", placeholderScheduledContract, scheduledContractAddress),
-	)
-
-	return []byte(code)
+	return encodedID, effort, nil
 }