Merge pull request #5787 from onflow/v0.33-new-ingestion-engine

v0.33 add new execution ingestion engine

cmd/execution_builder.go

@@ -150,7 +150,6 @@ type ExecutionNode struct {
 	followerEng            *followereng.ComplianceEngine // to sync blocks from consensus nodes
 	computationManager     *computation.Manager
 	collectionRequester    *requester.Engine
-	ingestionEng           *ingestion.Engine
 	scriptsEng             *scripts.Engine
 	followerDistributor    *pubsub.FollowerDistributor
 	checkAuthorizedAtBlock func(blockID flow.Identifier) (bool, error)
@@ -1096,14 +1095,35 @@ func (exeNode *ExecutionNode) LoadIngestionEngine(
 	}
 
 	fetcher := fetcher.NewCollectionFetcher(node.Logger, exeNode.collectionRequester, node.State, exeNode.exeConf.onflowOnlyLNs)
+	if exeNode.exeConf.enableNewIngestionEngine {
+		_, core, err := ingestion.NewMachine(
+			node.Logger,
+			node.ProtocolEvents,
+			exeNode.collectionRequester,
+			fetcher,
+			node.Storage.Headers,
+			node.Storage.Blocks,
+			node.Storage.Collections,
+			exeNode.executionState,
+			node.State,
+			exeNode.collector,
+			exeNode.computationManager,
+			exeNode.providerEngine,
+			exeNode.blockDataUploader,
+			exeNode.stopControl,
+		)
+
+		return core, err
+	}
+
 	var blockLoader ingestion.BlockLoader
 	if exeNode.exeConf.enableStorehouse {
 		blockLoader = loader.NewUnfinalizedLoader(node.Logger, node.State, node.Storage.Headers, exeNode.executionState)
 	} else {
 		blockLoader = loader.NewUnexecutedLoader(node.Logger, node.State, node.Storage.Headers, exeNode.executionState)
 	}
 
-	exeNode.ingestionEng, err = ingestion.New(
+	ingestionEng, err := ingestion.New(
 		exeNode.ingestionUnit,
 		node.Logger,
 		node.EngineRegistry,
@@ -1126,11 +1146,11 @@ func (exeNode *ExecutionNode) LoadIngestionEngine(
 
 	// TODO: we should solve these mutual dependencies better
 	// => https://github.com/dapperlabs/flow-go/issues/4360
-	exeNode.collectionRequester = exeNode.collectionRequester.WithHandle(exeNode.ingestionEng.OnCollection)
+	exeNode.collectionRequester.WithHandle(ingestionEng.OnCollection)
 
-	node.ProtocolEvents.AddConsumer(exeNode.ingestionEng)
+	node.ProtocolEvents.AddConsumer(ingestionEng)
 
-	return exeNode.ingestionEng, err
+	return ingestionEng, err
 }
 
 // create scripts engine for handling script execution

cmd/execution_config.go

@@ -63,9 +63,10 @@ type ExecutionConfig struct {
 	// It works around an issue where some collection nodes are not configured with enough
 	// this works around an issue where some collection nodes are not configured with enough
 	// file descriptors causing connection failures.
-	onflowOnlyLNs    bool
-	enableStorehouse bool
-	enableChecker    bool
+	onflowOnlyLNs            bool
+	enableStorehouse         bool
+	enableChecker            bool
+	enableNewIngestionEngine bool
 }
 
 func (exeConf *ExecutionConfig) SetupFlags(flags *pflag.FlagSet) {
@@ -120,6 +121,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.enableNewIngestionEngine, "enable-new-ingestion-engine", false, "enable new ingestion engine, default is false")
 
 }
 

engine/common/requester/engine.go

@@ -133,9 +133,8 @@ func New(log zerolog.Logger, metrics module.EngineMetrics, net network.EngineReg
 // function. It is done in a separate call so that the requester can be injected
 // into engines upon construction, and then provide a handle function to the
 // requester from that engine itself.
-func (e *Engine) WithHandle(handle HandleFunc) *Engine {
+func (e *Engine) WithHandle(handle HandleFunc) {
 	e.handle = handle
-	return e
 }
 
 // Ready returns a ready channel that is closed once the engine has fully