[DataAvailability] Implement fork-aware Execution Data Endpoints

.mockery.yaml

@@ -71,10 +71,6 @@ packages:
   github.com/onflow/flow-go/module/execution:
   github.com/onflow/flow-go/module/executiondatasync/execution_data:
   github.com/onflow/flow-go/module/executiondatasync/optimistic_sync:
-    config:
-      all: False
-    interfaces:
-      Core:
   github.com/onflow/flow-go/module/executiondatasync/tracker:
   github.com/onflow/flow-go/module/forest:
   github.com/onflow/flow-go/module/mempool:

cmd/access/node_builder/access_node_builder.go

@@ -379,8 +379,10 @@ type FlowAccessNodeBuilder struct {
 	unsecureGrpcServer    *grpcserver.GrpcServer
 	stateStreamGrpcServer *grpcserver.GrpcServer
 
-	stateStreamBackend *statestreambackend.StateStreamBackend
-	nodeBackend        *backend.Backend
+	stateStreamBackend          *statestreambackend.StateStreamBackend
+	nodeBackend                 *backend.Backend
+	executionResultInfoProvider optimistic_sync.ExecutionResultInfoProvider
+	executionStateCache         optimistic_sync.ExecutionStateCache
 
 	ExecNodeIdentitiesProvider   *commonrpc.ExecutionNodeIdentitiesProvider
 	TxResultErrorMessagesCore    *tx_error_messages.TxErrorMessagesCore
@@ -654,6 +656,39 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 
 			return nil
 		}).
+		Module("execution state cache", func(node *cmd.NodeConfig) error {
+
+			// check if all the storages are initialized before passing them to Snapshot
+			if builder.events == nil {
+				return fmt.Errorf("events store not initialized: ensure 'events storage' module runs before 'execution state cache'")
+			}
+			if builder.collections == nil {
+				return fmt.Errorf("collections store not initialized: ensure 'collections storage' module runs before 'execution state cache'")
+			}
+			if builder.transactions == nil {
+				return fmt.Errorf("transactions store not initialized: ensure 'transactions storage' module runs before 'execution state cache'")
+			}
+			if builder.lightTransactionResults == nil {
+				return fmt.Errorf("lightTransactionResults store not initialized: ensure 'lightTransactionResults storage' module runs before 'execution state cache'")
+			}
+			if builder.transactionResultErrorMessages == nil {
+				return fmt.Errorf("transactionResultErrorMessages store not initialized: ensure 'transactionResultErrorMessages storage' module runs before 'execution state cache'")
+			}
+
+			// TODO: use real objects instead of mocks once they're implemented
+			snapshot := osyncsnapshot.NewSnapshotMock(
+				builder.events,
+				builder.collections,
+				builder.transactions,
+				builder.lightTransactionResults,
+				builder.transactionResultErrorMessages,
+				nil,
+				notNil(executionDataStoreCache),
+			)
+			builder.executionStateCache = execution_state.NewExecutionStateCacheMock(snapshot)
+
+			return nil
+		}).
 		Component("execution data service", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 			opts := []network.BlobServiceOption{
 				blob.WithBitswapOptions(
@@ -858,10 +893,6 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 				indexedBlockHeight = store.NewConsumerProgress(builder.ProtocolDB, module.ConsumeProgressExecutionDataIndexerBlockHeight)
 				return nil
 			}).
-			Module("transaction results storage", func(node *cmd.NodeConfig) error {
-				builder.lightTransactionResults = store.NewLightTransactionResults(node.Metrics.Cache, node.ProtocolDB, bstorage.DefaultCacheSize)
-				return nil
-			}).
 			DependableComponent("execution data indexer", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 				// Note: using a DependableComponent here to ensure that the indexer does not block
 				// other components from starting while bootstrapping the register db since it may
@@ -1088,6 +1119,8 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 					builder.stateStreamConf.ClientSendBufferSize,
 				),
 				executionDataTracker,
+				notNil(builder.executionResultInfoProvider),
+				builder.executionStateCache, // might be nil
 			)
 			if err != nil {
 				return nil, fmt.Errorf("could not create state stream backend: %w", err)
@@ -1729,10 +1762,76 @@ func (builder *FlowAccessNodeBuilder) enqueueRelayNetwork() {
 	})
 }
 
+// buildStoragesData registers modules that initialize core read-only stores used by
+// the access node (events, lightTransactionResults, and transactionResultErrorMessages).
+func (builder *FlowAccessNodeBuilder) buildStoragesData() *FlowAccessNodeBuilder {
+	builder.Module("events storage", func(node *cmd.NodeConfig) error {
+		builder.events = store.NewEvents(node.Metrics.Cache, node.ProtocolDB)
+		return nil
+	})
+	builder.Module("transaction result error messages storage", func(node *cmd.NodeConfig) error {
+		builder.transactionResultErrorMessages = store.NewTransactionResultErrorMessages(node.Metrics.Cache, node.ProtocolDB, bstorage.DefaultCacheSize)
+		return nil
+	})
+	builder.Module("transaction results storage", func(node *cmd.NodeConfig) error {
+		builder.lightTransactionResults = store.NewLightTransactionResults(node.Metrics.Cache, node.ProtocolDB, bstorage.DefaultCacheSize)
+		return nil
+	})
+
+	return builder
+}
+
+// buildExecutionResultInfoProvider registers a module that wires the
+// optimistic_sync.ExecutionResultInfoProvider on the builder.
+func (builder *FlowAccessNodeBuilder) buildExecutionResultInfoProvider() *FlowAccessNodeBuilder {
+	builder.Module("execution result info provider", func(node *cmd.NodeConfig) error {
+		backendConfig := builder.rpcConf.BackendConfig
+
+		preferredENIdentifiers, err := flow.IdentifierListFromHex(backendConfig.PreferredExecutionNodeIDs)
+		if err != nil {
+			return fmt.Errorf("failed to convert node id string to Flow Identifier for preferred EN map: %w", err)
+		}
+
+		fixedENIdentifiers, err := flow.IdentifierListFromHex(backendConfig.FixedExecutionNodeIDs)
+		if err != nil {
+			return fmt.Errorf("failed to convert node id string to Flow Identifier for fixed EN map: %w", err)
+		}
+
+		requiredENIdentifiers, err := flow.IdentifierListFromHex(builder.executionResultRequiredExecutors)
+		if err != nil {
+			return fmt.Errorf("failed to convert node id string to Flow Identifier for required EN list: %w", err)
+		}
+
+		execNodeSelector := execution_result.NewExecutionNodeSelector(
+			preferredENIdentifiers,
+			fixedENIdentifiers,
+		)
+
+		operatorCriteria := optimistic_sync.Criteria{
+			AgreeingExecutorsCount: builder.executionResultAgreeingExecutorsCount,
+			RequiredExecutors:      requiredENIdentifiers,
+		}
+
+		builder.executionResultInfoProvider = execution_result.NewExecutionResultInfoProvider(
+			node.Logger,
+			node.State,
+			node.Storage.Receipts,
+			execNodeSelector,
+			operatorCriteria,
+		)
+
+		return nil
+	})
+	return builder
+}
+
 func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 	var processedFinalizedBlockHeight storage.ConsumerProgressInitializer
 	var processedTxErrorMessagesBlockHeight storage.ConsumerProgressInitializer
 
+	builder.buildStoragesData()
+	builder.buildExecutionResultInfoProvider()
+
 	if builder.executionDataSyncEnabled {
 		builder.BuildExecutionSyncComponents()
 	}
@@ -1939,10 +2038,6 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 			builder.RegistersAsyncStore = execution.NewRegistersAsyncStore()
 			return nil
 		}).
-		Module("events storage", func(node *cmd.NodeConfig) error {
-			builder.events = store.NewEvents(node.Metrics.Cache, node.ProtocolDB)
-			return nil
-		}).
 		Module("reporter", func(node *cmd.NodeConfig) error {
 			builder.Reporter = index.NewReporter()
 			return nil
@@ -1974,13 +2069,6 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 
 			return nil
 		}).
-		Module("transaction result error messages storage", func(node *cmd.NodeConfig) error {
-			if builder.storeTxResultErrorMessages {
-				builder.transactionResultErrorMessages = store.NewTransactionResultErrorMessages(node.Metrics.Cache, node.ProtocolDB, bstorage.DefaultCacheSize)
-			}
-
-			return nil
-		}).
 		Component("version control", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 			if !builder.versionControlEnabled {
 				noop := &module.NoopReadyDoneAware{}
@@ -2110,15 +2198,6 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				return nil, fmt.Errorf("failed to convert node id string to Flow Identifier for fixed EN list: %w", err)
 			}
 
-			requiredENIdentifiers, err := flow.IdentifierListFromHex(builder.executionResultRequiredExecutors)
-			if err != nil {
-				return nil, fmt.Errorf("failed to convert node id string to Flow Identifier for required EN list: %w", err)
-			}
-			operatorCriteria := optimistic_sync.Criteria{
-				AgreeingExecutorsCount: builder.executionResultAgreeingExecutorsCount,
-				RequiredExecutors:      requiredENIdentifiers,
-			}
-
 			builder.ExecNodeIdentitiesProvider = commonrpc.NewExecutionNodeIdentitiesProvider(
 				node.Logger,
 				node.State,
@@ -2137,30 +2216,6 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				notNil(builder.ExecNodeIdentitiesProvider),
 			)
 
-			execNodeSelector := execution_result.NewExecutionNodeSelector(
-				preferredENIdentifiers,
-				fixedENIdentifiers,
-			)
-
-			execResultInfoProvider := execution_result.NewExecutionResultInfoProvider(
-				node.Logger,
-				node.State,
-				node.Storage.Receipts,
-				execNodeSelector,
-				operatorCriteria,
-			)
-
-			// TODO: use real objects instead of mocks once they're implemented
-			snapshot := osyncsnapshot.NewSnapshotMock(
-				builder.events,
-				builder.collections,
-				builder.transactions,
-				builder.lightTransactionResults,
-				builder.transactionResultErrorMessages,
-				nil,
-			)
-			execStateCache := execution_state.NewExecutionStateCacheMock(snapshot)
-
 			builder.nodeBackend, err = backend.New(backend.Params{
 				State:                 node.State,
 				CollectionRPC:         builder.CollectionRPC, // might be nil
@@ -2201,8 +2256,8 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				VersionControl:              notNil(builder.VersionControl),
 				ExecNodeIdentitiesProvider:  notNil(builder.ExecNodeIdentitiesProvider),
 				TxErrorMessageProvider:      notNil(builder.txResultErrorMessageProvider),
-				ExecutionResultInfoProvider: execResultInfoProvider,
-				ExecutionStateCache:         execStateCache,
+				ExecutionResultInfoProvider: notNil(builder.executionResultInfoProvider),
+				ExecutionStateCache:         builder.executionStateCache, // might be nil
 				MaxScriptAndArgumentSize:    config.BackendConfig.AccessConfig.MaxRequestMsgSize,
 				ScheduledCallbacksEnabled:   builder.scheduledCallbacksEnabled,
 			})
@@ -2534,7 +2589,7 @@ func (builder *FlowAccessNodeBuilder) initPublicLibp2pNode(networkKey crypto.Pri
 // notNil ensures that the input is not nil and returns it
 // the usage is to ensure the dependencies are initialized before initializing a module.
 // for instance, the IngestionEngine depends on storage.Collections, which is initialized in a
-// different function, so we need to ensure that the storage.Collections is initialized before
+// different function, so we need to ensure that the storage.Collections were initialized before
 // creating the IngestionEngine.
 func notNil[T any](dep T) T {
 	if any(dep) == nil {