Merge pull request #7672 from devbugging/gregor/callbacks/e2e

[Scheduled Callbacks] E2E Tests

Author: janezpodhostnik

cmd/execution_builder.go

@@ -575,7 +575,9 @@ func (exeNode *ExecutionNode) LoadProviderEngine(
 	opts = append(opts, computation.DefaultFVMOptions(
 		node.RootChainID,
 		exeNode.exeConf.computationConfig.CadenceTracing,
-		exeNode.exeConf.computationConfig.ExtensiveTracing)...)
+		exeNode.exeConf.computationConfig.ExtensiveTracing,
+		exeNode.exeConf.scheduleCallbacksEnabled)...)
+
 	vmCtx := fvm.NewContext(opts...)
 
 	var collector module.ExecutionMetrics

cmd/execution_config.go

@@ -60,6 +60,7 @@ type ExecutionConfig struct {
 	transactionExecutionMetricsEnabled    bool
 	transactionExecutionMetricsBufferSize uint
 	executionDataDBMode                   string
+	scheduleCallbacksEnabled              bool
 
 	computationConfig        computation.ComputationConfig
 	receiptRequestWorkers    uint   // common provider engine workers
@@ -139,6 +140,7 @@ func (exeConf *ExecutionConfig) SetupFlags(flags *pflag.FlagSet) {
 	flags.BoolVar(&exeConf.onflowOnlyLNs, "temp-onflow-only-lns", false, "do not use unless required. forces node to only request collections from onflow collection nodes")
 	flags.BoolVar(&exeConf.enableStorehouse, "enable-storehouse", false, "enable storehouse to store registers on disk, default is false")
 	flags.BoolVar(&exeConf.enableChecker, "enable-checker", true, "enable checker to check the correctness of the execution result, default is true")
+	flags.BoolVar(&exeConf.scheduleCallbacksEnabled, "scheduled-callbacks-enabled", false, "enable execution of scheduled callbacks")
 	// deprecated. Retain it to prevent nodes that previously had this configuration from crashing.
 	var deprecatedEnableNewIngestionEngine bool
 	flags.BoolVar(&deprecatedEnableNewIngestionEngine, "enable-new-ingestion-engine", true, "enable new ingestion engine, default is true")

cmd/util/cmd/run-script/cmd.go

@@ -139,7 +139,7 @@ func run(*cobra.Command, []string) {
 		registersByAccount.AccountCount(),
 	)
 
-	options := computation.DefaultFVMOptions(chainID, false, false)
+	options := computation.DefaultFVMOptions(chainID, false, false, false)
 	options = append(
 		options,
 		fvm.WithContractDeploymentRestricted(false),

cmd/util/ledger/migrations/transaction_migration.go

@@ -19,7 +19,7 @@ func NewTransactionBasedMigration(
 ) RegistersMigration {
 	return func(registersByAccount *registers.ByAccount) error {
 
-		options := computation.DefaultFVMOptions(chainID, false, false)
+		options := computation.DefaultFVMOptions(chainID, false, false, false)
 		options = append(options,
 			fvm.WithContractDeploymentRestricted(false),
 			fvm.WithContractRemovalRestricted(false),

cmd/verification_builder.go

@@ -55,7 +55,8 @@ type VerificationConfig struct {
 	blockWorkers uint64 // number of blocks processed in parallel.
 	chunkWorkers uint64 // number of chunks processed in parallel.
 
-	stopAtHeight uint64 // height to stop the node on
+	stopAtHeight             uint64 // height to stop the node on
+	scheduleCallbacksEnabled bool   // enable execution of scheduled callbacks
 }
 
 type VerificationNodeBuilder struct {
@@ -83,6 +84,7 @@ func (v *VerificationNodeBuilder) LoadFlags() {
 			flags.Uint64Var(&v.verConf.blockWorkers, "block-workers", blockconsumer.DefaultBlockWorkers, "maximum number of blocks being processed in parallel")
 			flags.Uint64Var(&v.verConf.chunkWorkers, "chunk-workers", chunkconsumer.DefaultChunkWorkers, "maximum number of execution nodes a chunk data pack request is dispatched to")
 			flags.Uint64Var(&v.verConf.stopAtHeight, "stop-at-height", 0, "height to stop the node at (0 to disable)")
+			flags.BoolVar(&v.verConf.scheduleCallbacksEnabled, "scheduled-callbacks-enabled", false, "enable execution of scheduled callbacks")
 		})
 }
 
@@ -211,7 +213,7 @@ func (v *VerificationNodeBuilder) LoadComponentsAndModules() {
 			)
 
 			// TODO(JanezP): cleanup creation of fvm context github.com/onflow/flow-go/issues/5249
-			fvmOptions = append(fvmOptions, computation.DefaultFVMOptions(node.RootChainID, false, false)...)
+			fvmOptions = append(fvmOptions, computation.DefaultFVMOptions(node.RootChainID, false, false, v.verConf.scheduleCallbacksEnabled)...)
 			vmCtx := fvm.NewContext(fvmOptions...)
 
 			chunkVerifier := chunks.NewChunkVerifier(vm, vmCtx, node.Logger)

engine/execution/computation/computer/computer.go

@@ -324,7 +324,8 @@ func (e *blockComputer) queueSystemTransactions(
 	defer close(txQueue)
 
 	for i, txBody := range allTxs {
-		last := i == len(allTxs)-1
+		last := i == len(allTxs)-1 // Is this last tx in collection
+
 		ctx := callbackCtx
 		// last transaction is system chunk and has own context
 		if last {
@@ -383,7 +384,8 @@ func (e *blockComputer) executeBlock(
 		e.receiptHasher,
 		parentBlockExecutionResultID,
 		block,
-		e.maxConcurrency*2, // we add some buffer just in case result collection becomes slower than the execution
+		// Add buffer just in case result collection becomes slower than the execution
+		e.maxConcurrency*2,
 		e.colResCons,
 		baseSnapshot,
 	)
@@ -454,14 +456,16 @@ func (e *blockComputer) executeUserTransactions(
 
 // executeSystemTransactions executes all system transactions in the block as part of the system collection.
 //
-// System transactions are executed in the following order:
-// 1. system transaction that processes the scheduled callbacks which is a blocking transaction and
-// the result is used for the next system transaction
-// 2. system transactions that each execute a single scheduled callback by the ID obtained from events
-// of the previous system transaction
-// 3. system transaction that executes the system chunk
+// When scheduled callbacks are enabled, system transactions are executed in the following order:
+//  1. Process callback transaction - queries the scheduler contract to identify ready callbacks
+//     and emits events containing callback IDs and execution effort requirements
+//  2. Callback execution transactions - one transaction per callback ID from step 1 events,
+//     each executing a single scheduled callback with its specified effort limit
+//  3. System chunk transaction - performs standard system operations
+//
+// When scheduled callbacks are disabled, only the system chunk transaction is executed.
 //
-// An error can be returned if the process callback transaction fails. This is a fatal error.
+// All errors are indicators of bugs or corrupted internal state (continuation impossible)
 func (e *blockComputer) executeSystemTransactions(
 	block *entity.ExecutableBlock,
 	blockSpan otelTrace.Span,
@@ -496,7 +500,7 @@ func (e *blockComputer) executeSystemTransactions(
 		blockId:         block.BlockID(),
 		blockIdStr:      block.BlockID().String(),
 		blockHeight:     block.Block.Height,
-		collectionIndex: len(rawCollections),
+		collectionIndex: userCollectionCount,
 		CompleteCollection: &entity.CompleteCollection{
 			Collection: flow.NewEmptyCollection(), // TODO(7749)
 		},
@@ -522,6 +526,12 @@ func (e *blockComputer) executeSystemTransactions(
 		txIndex = updatedTxnIndex
 	}
 
+	// Update logger with number of transactions once they've become known
+	// (user tx + callbacks + 2 (process, system)
+	systemLogger = systemLogger.With().
+		Uint32("num_txs", uint32(userTxCount+len(callbackTxs)+2)).
+		Logger()
+
 	txQueue := e.queueSystemTransactions(
 		callbackCtx,
 		systemChunkCtx,
@@ -557,10 +567,20 @@ func (e *blockComputer) executeQueue(
 	wg.Wait()
 }
 
-// executeProcessCallback executes a transaction that calls callback scheduler contract process method.
-// The execution result contains events that are emitted for each callback which is ready for execution.
-// We use these events to prepare callback execution transactions, which are later executed as part of the system collection.
-// An error can be returned if the process callback transaction fails. This is a fatal error.
+// executeProcessCallback submits a transaction that invokes the `process` method
+// on the callback scheduler contract.
+//
+// The `process` method scans for scheduled callbacks and emits an event for each that should
+// be executed. These emitted events are used to construct callback execution transactions,
+// which are then added to the system transaction collection.
+//
+// If the `process` transaction fails, a fatal error is returned.
+//
+// Note: this transaction is executed serially and not concurrently with the system transaction.
+// This is because it's unclear whether the callback executions triggered by `process`
+// will result in additional system transactions.
+// In theory, if no additional transactions are emitted, concurrent execution could be optimized.
+// However, due to the added complexity, this optimization was deferred.
 func (e *blockComputer) executeProcessCallback(
 	systemCtx fvm.Context,
 	systemCollectionInfo collectionInfo,
@@ -591,8 +611,7 @@ func (e *blockComputer) executeProcessCallback(
 		}
 
 		return nil, 0, fmt.Errorf(
-			"failed to execute %s transaction %v (%d@%d) for block %s at height %v: %w",
-			"system",
+			"failed to execute system process transaction %v (%d@%d) for block %s at height %v: %w",
 			request.txnIdStr,
 			request.txnIndex,
 			snapshotTime,