Scheduled callbacks

cmd/emulator/start/start.go

@@ -80,6 +80,7 @@ type Config struct {
 	CheckpointPath               string        `default:"" flag:"checkpoint-dir" info:"checkpoint directory to load the emulator state from"`
 	StateHash                    string        `default:"" flag:"state-hash" info:"state hash of the checkpoint to load the emulator state from"`
 	ComputationReportingEnabled  bool          `default:"false" flag:"computation-reporting" info:"enable Cadence computation reporting"`
+	ScheduledCallbacksEnabled    bool          `default:"false" flag:"scheduled-callbacks" info:"enable Cadence scheduled callbacks"`
 	SetupEVMEnabled              bool          `default:"true" flag:"setup-evm" info:"enable EVM setup for the emulator, this will deploy the EVM contracts"`
 	SetupVMBridgeEnabled         bool          `default:"true" flag:"setup-vm-bridge" info:"enable VM Bridge setup for the emulator, this will deploy the VM Bridge contracts"`
 }
@@ -190,39 +191,40 @@ func Cmd(config StartConfig) *cobra.Command {
 				RESTPort:     conf.RestPort,
 				RESTDebug:    conf.RESTDebug,
 				// TODO: allow headers to be parsed from environment
-				HTTPHeaders:                 nil,
-				BlockTime:                   conf.BlockTime,
-				ServicePublicKey:            servicePublicKey,
-				ServicePrivateKey:           servicePrivateKey,
-				ServiceKeySigAlgo:           serviceKeySigAlgo,
-				ServiceKeyHashAlgo:          serviceKeyHashAlgo,
-				Persist:                     conf.Persist,
-				Snapshot:                    conf.Snapshot,
-				DBPath:                      conf.DBPath,
-				GenesisTokenSupply:          parseCadenceUFix64(conf.TokenSupply, "token-supply"),
-				TransactionMaxGasLimit:      uint64(conf.TransactionMaxGasLimit),
-				ScriptGasLimit:              uint64(conf.ScriptGasLimit),
-				TransactionExpiry:           uint(conf.TransactionExpiry),
-				StorageLimitEnabled:         conf.StorageLimitEnabled,
-				StorageMBPerFLOW:            storageMBPerFLOW,
-				MinimumStorageReservation:   minimumStorageReservation,
-				TransactionFeesEnabled:      conf.TransactionFeesEnabled,
-				WithContracts:               conf.Contracts,
-				SkipTransactionValidation:   conf.SkipTxValidation,
-				SimpleAddressesEnabled:      conf.SimpleAddresses,
-				Host:                        conf.Host,
-				ChainID:                     flowChainID,
-				RedisURL:                    conf.RedisURL,
-				ContractRemovalEnabled:      conf.ContractRemovalEnabled,
-				SqliteURL:                   conf.SqliteURL,
-				CoverageReportingEnabled:    conf.CoverageReportingEnabled,
-				StartBlockHeight:            conf.StartBlockHeight,
-				RPCHost:                     conf.RPCHost,
-				CheckpointPath:              conf.CheckpointPath,
-				StateHash:                   conf.StateHash,
-				ComputationReportingEnabled: conf.ComputationReportingEnabled,
-				SetupEVMEnabled:             conf.SetupEVMEnabled,
-				SetupVMBridgeEnabled:        conf.SetupVMBridgeEnabled,
+				HTTPHeaders:                  nil,
+				BlockTime:                    conf.BlockTime,
+				ServicePublicKey:             servicePublicKey,
+				ServicePrivateKey:            servicePrivateKey,
+				ServiceKeySigAlgo:            serviceKeySigAlgo,
+				ServiceKeyHashAlgo:           serviceKeyHashAlgo,
+				Persist:                      conf.Persist,
+				Snapshot:                     conf.Snapshot,
+				DBPath:                       conf.DBPath,
+				GenesisTokenSupply:           parseCadenceUFix64(conf.TokenSupply, "token-supply"),
+				TransactionMaxGasLimit:       uint64(conf.TransactionMaxGasLimit),
+				ScriptGasLimit:               uint64(conf.ScriptGasLimit),
+				TransactionExpiry:            uint(conf.TransactionExpiry),
+				StorageLimitEnabled:          conf.StorageLimitEnabled,
+				StorageMBPerFLOW:             storageMBPerFLOW,
+				MinimumStorageReservation:    minimumStorageReservation,
+				TransactionFeesEnabled:       conf.TransactionFeesEnabled,
+				WithContracts:                conf.Contracts,
+				SkipTransactionValidation:    conf.SkipTxValidation,
+				SimpleAddressesEnabled:       conf.SimpleAddresses,
+				Host:                         conf.Host,
+        ChainID:                      flowChainID,
+				RedisURL:                     conf.RedisURL,
+				ContractRemovalEnabled:       conf.ContractRemovalEnabled,
+				SqliteURL:                    conf.SqliteURL,
+				CoverageReportingEnabled:     conf.CoverageReportingEnabled,
+				StartBlockHeight:             conf.StartBlockHeight,
+				RPCHost:                      conf.RPCHost,
+				CheckpointPath:               conf.CheckpointPath,
+				StateHash:                    conf.StateHash,
+				ComputationReportingEnabled:  conf.ComputationReportingEnabled,
+				ScheduledCallbacksEnabled:    conf.ScheduledCallbacksEnabled,
+				SetupEVMEnabled:              conf.SetupEVMEnabled,
+				SetupVMBridgeEnabled:         conf.SetupVMBridgeEnabled,
 			}
 
 			emu := server.NewEmulatorServer(logger, serverConf)

emulator/blockchain.go

@@ -313,6 +313,12 @@ func WithComputationReporting(enabled bool) Option {
 	}
 }
 
+func WithScheduledCallbacks(enabled bool) Option {
+	return func(c *config) {
+		c.ScheduledCallbacksEnabled = enabled
+  }
+}
+
 // WithSetupEVMEnabled enables/disables the EVM setup.
 func WithSetupEVMEnabled(enabled bool) Option {
 	return func(c *config) {
@@ -397,6 +403,7 @@ type config struct {
 	AutoMine                     bool
 	Contracts                    []ContractDescription
 	ComputationReportingEnabled  bool
+	ScheduledCallbacksEnabled    bool
 	SetupEVMEnabled              bool
 	SetupVMBridgeEnabled         bool
 }
@@ -1276,6 +1283,23 @@ func (b *Blockchain) executeBlock() ([]*types.TransactionResult, error) {
 		results = append(results, result)
 	}
 
+	// lastly execute any scheduled callbacks if the feature is enabled
+	if b.conf.ScheduledCallbacksEnabled {
+		// todo refactor after bootstrap deploys CallbackScheduler
+		// this is a temporary workaround since deployment of CallbackScheduler is not
+		// yet part of bootstrap procedure, so it must be deployed in the first block
+		// during emulator startup, so we shouldn't support scheduling in these first blocks
+		if b.pendingBlock.height < 2 {
+			return results, nil
+		}
+
+		callbacks, err := b.executeScheduledCallbacks(blockContext)
+		if err != nil {
+			return results, err
+		}
+		results = append(results, callbacks...)
+	}
+
 	return results, nil
 }
 
@@ -1877,6 +1901,44 @@ func (b *Blockchain) executeSystemChunkTransaction() error {
 	return nil
 }
 
+func (b *Blockchain) executeScheduledCallbacks(blockContext fvm.Context) ([]*types.TransactionResult, error) {
+	var results []*types.TransactionResult
+
+	serviceAddress := b.GetChain().ServiceAddress()
+	parentID := b.pendingBlock.parentID
+	// disable checks for signatures and keys since we are executing a system transaction
+	ctx := fvm.NewContextFromParent(
+		blockContext,
+		fvm.WithAuthorizationChecksEnabled(false),
+		fvm.WithSequenceNumberCheckAndIncrementEnabled(false),
+	)
+
+	// add transaction to schedule callbacks
+	b.pendingBlock.AddTransaction(processCallbackTransaction(serviceAddress, parentID))
+	result, err := b.executeNextTransaction(ctx)
+	if err != nil {
+		return results, err
+	}
+	results = append(results, result)
+
+	// execute callbacks we receive from events of the schedule transaction
+	executeTxs, err := executeCallbackTransactions(result.Events, serviceAddress, parentID)
+	if err != nil {
+		return results, err
+	}
+
+	for _, tx := range executeTxs {
+		b.pendingBlock.AddTransaction(tx)
+		result, err := b.executeNextTransaction(ctx)
+		if err != nil {
+			return results, err
+		}
+		results = append(results, result)
+	}
+
+	return results, nil
+}
+
 func (b *Blockchain) GetRegisterValues(registerIDs flowgo.RegisterIDs, height uint64) (values []flowgo.RegisterValue, err error) {
 	ledger, err := b.storage.LedgerByHeight(context.Background(), height)
 	if err != nil {

emulator/callbacks.go

@@ -0,0 +1,136 @@
+/*
+ * Flow Emulator
+ *
+ * Copyright Flow Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package emulator
+
+import (
+	_ "embed"
+	"fmt"
+	"strings"
+
+	"github.com/onflow/cadence"
+	"github.com/onflow/cadence/common"
+	jsoncdc "github.com/onflow/cadence/encoding/json"
+	flowsdk "github.com/onflow/flow-go-sdk"
+	"github.com/onflow/flow-go/model/flow"
+)
+
+//go:embed templates/executeCallbackTransaction.cdc
+var executeCallbackScript []byte
+
+//go:embed templates/processCallbackTransaction.cdc
+var processCallbackScript []byte
+
+const (
+	contractName           = "UnsafeCallbackScheduler"
+	callbackProcessedEvent = "CallbackProcessed"
+)
+
+// todo: replace all the functions bellow with flow-go implementation once it's done
+// issue: https://github.com/onflow/flow-emulator/issues/829
+
+func processCallbackTransaction(
+	serviceAddress flow.Address,
+	parentID flow.Identifier,
+) flow.TransactionBody {
+	script := replaceSchedulerAddress(processCallbackScript, serviceAddress)
+
+	return *flow.NewTransactionBody().
+		SetScript(script).
+		SetComputeLimit(defaultTransactionMaxGasLimit).
+		SetPayer(serviceAddress).
+		SetReferenceBlockID(parentID)
+}
+
+func executeCallbackTransactions(
+	scheduleEvent []flowsdk.Event,
+	serviceAddress flow.Address,
+	parentID flow.Identifier,
+) ([]flow.TransactionBody, error) {
+	var transactions []flow.TransactionBody
+	script := replaceSchedulerAddress(executeCallbackScript, serviceAddress)
+
+	for _, e := range scheduleEvent {
+		limit, id, err := parseSchedulerProcessedEvent(e, serviceAddress)
+		if err != nil {
+			return nil, err
+		}
+
+		tx := flow.NewTransactionBody().
+			SetScript(script).
+			AddArgument(id).
+			SetPayer(serviceAddress).
+			SetReferenceBlockID(parentID).
+			SetComputeLimit(limit)
+
+		transactions = append(transactions, *tx)
+	}
+
+	return transactions, nil
+}
+
+// parseSchedulerProcessedEvent parses flow event that is emitted during scheduler
+// marking the callback as scheduled.
+// Returns:
+// - execution effort
+// - ID of the event encoded as bytes
+// - error in case the event type is not correct
+func parseSchedulerProcessedEvent(event flowsdk.Event, serviceAddress flow.Address) (uint64, []byte, error) {
+	contractLocation := common.AddressLocation{
+		Address: common.Address(serviceAddress),
+		Name:    contractName,
+	}
+	callbackScheduledEvent := contractLocation.TypeID(nil, fmt.Sprintf("%s.%s", contractName, callbackProcessedEvent))
+
+	const (
+		IDField        = "ID"
+		executionField = "executionEffort"
+	)
+
+	if event.Type != string(callbackScheduledEvent) {
+		return 0, nil, fmt.Errorf("invalid event type, got: %s, expected: %s", event.Type, callbackScheduledEvent)
+	}
+
+	id, ok := event.Value.SearchFieldByName(IDField).(cadence.UInt64)
+	if !ok {
+		return 0, nil, fmt.Errorf("invalid ID value type: %v", id)
+	}
+
+	encodedID, err := jsoncdc.Encode(id)
+	if err != nil {
+		return 0, nil, err
+	}
+
+	effort, ok := event.Value.SearchFieldByName(executionField).(cadence.UInt64)
+	if !ok {
+		return 0, nil, fmt.Errorf("invalid effort value type: %v", effort)
+	}
+	computeLimit := uint64(effort)
+
+	return computeLimit, encodedID, nil
+}
+
+func replaceSchedulerAddress(script []byte, serviceAddress flow.Address) []byte {
+	s := strings.ReplaceAll(
+		string(script),
+		`import "UnsafeCallbackScheduler"`,
+		fmt.Sprintf("import UnsafeCallbackScheduler from %s", serviceAddress.HexWithPrefix()),
+	)
+
+	return []byte(s)
+}