[DataAvailability] Implement fork-aware Execution Data Endpoints

.mockery.yaml

@@ -75,6 +75,8 @@ packages:
       all: False
     interfaces:
       Core:
+      Snapshot:
+      BlockExecutionDataReader:
   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

@@ -375,8 +375,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
@@ -650,6 +652,21 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 
 			return nil
 		}).
+		Module("execution state cache", func(node *cmd.NodeConfig) error {
+			// 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,
+				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(
@@ -1084,6 +1101,8 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 					builder.stateStreamConf.ClientSendBufferSize,
 				),
 				executionDataTracker,
+				notNil(builder.executionResultInfoProvider),
+				builder.executionStateCache,
 			)
 			if err != nil {
 				return nil, fmt.Errorf("could not create state stream backend: %w", err)
@@ -1714,10 +1733,50 @@ func (builder *FlowAccessNodeBuilder) enqueueRelayNetwork() {
 	})
 }
 
+// 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)
+		}
+
+		execNodeSelector := execution_result.NewExecutionNodeSelector(
+			preferredENIdentifiers,
+			fixedENIdentifiers,
+		)
+
+		builder.executionResultInfoProvider = execution_result.NewExecutionResultInfoProvider(
+			node.Logger,
+			node.State,
+			node.Storage.Receipts,
+			execNodeSelector,
+			optimistic_sync.DefaultCriteria,
+		)
+
+		return nil
+	})
+	return builder
+}
+
 func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 	var processedFinalizedBlockHeight storage.ConsumerProgressInitializer
 	var processedTxErrorMessagesBlockHeight storage.ConsumerProgressInitializer
 
+	builder.Module("events storage", func(node *cmd.NodeConfig) error {
+		builder.events = store.NewEvents(node.Metrics.Cache, node.ProtocolDB)
+		return nil
+	})
+	builder.buildExecutionResultInfoProvider()
+
 	if builder.executionDataSyncEnabled {
 		builder.BuildExecutionSyncComponents()
 	}
@@ -1924,10 +1983,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
@@ -2113,30 +2168,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,
-				optimistic_sync.DefaultCriteria,
-			)
-
-			// 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
@@ -2177,8 +2208,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,
 				OperatorCriteria:            optimistic_sync.DefaultCriteria,
 				MaxScriptAndArgumentSize:    config.BackendConfig.AccessConfig.MaxRequestMsgSize,
 				ScheduledCallbacksEnabled:   builder.scheduledCallbacksEnabled,

cmd/observer/node_builder/observer_builder.go

@@ -327,7 +327,9 @@ type ObserverServiceBuilder struct {
 	unsecureGrpcServer    *grpcserver.GrpcServer
 	stateStreamGrpcServer *grpcserver.GrpcServer
 
-	stateStreamBackend *statestreambackend.StateStreamBackend
+	stateStreamBackend          *statestreambackend.StateStreamBackend
+	executionResultInfoProvider optimistic_sync.ExecutionResultInfoProvider
+	executionStateCache         optimistic_sync.ExecutionStateCache
 }
 
 // deriveBootstrapPeerIdentities derives the Flow Identity of the bootstrap peers from the parameters.
@@ -1092,10 +1094,45 @@ func (builder *ObserverServiceBuilder) initObserverLocal() func(node *cmd.NodeCo
 	}
 }
 
+// buildExecutionResultInfoProvider registers a module that wires the
+// optimistic_sync.ExecutionResultInfoProvider on the builder.
+func (builder *ObserverServiceBuilder) buildExecutionResultInfoProvider() *ObserverServiceBuilder {
+	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)
+		}
+
+		execNodeSelector := execution_result.NewExecutionNodeSelector(
+			preferredENIdentifiers,
+			fixedENIdentifiers,
+		)
+
+		builder.executionResultInfoProvider = execution_result.NewExecutionResultInfoProvider(
+			node.Logger,
+			node.State,
+			node.Storage.Receipts,
+			execNodeSelector,
+			optimistic_sync.DefaultCriteria,
+		)
+
+		return nil
+	})
+	return builder
+}
+
 // Build enqueues the sync engine and the follower engine for the observer.
 // Currently, the observer only runs the follower engine.
 func (builder *ObserverServiceBuilder) Build() (cmd.Node, error) {
 	builder.BuildConsensusFollower()
+	builder.buildExecutionResultInfoProvider()
 
 	if builder.executionDataSyncEnabled {
 		builder.BuildExecutionSyncComponents()
@@ -1206,6 +1243,25 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 
 			return nil
 		}).
+		Module("events storage", func(node *cmd.NodeConfig) error {
+			builder.events = store.NewEvents(node.Metrics.Cache, node.ProtocolDB)
+			return nil
+		}).
+		Module("execution state cache", func(node *cmd.NodeConfig) error {
+			// TODO: use real objects instead of mocks once they're implemented
+			snapshot := osyncsnapshot.NewSnapshotMock(
+				builder.events,
+				nil,
+				nil,
+				builder.lightTransactionResults,
+				nil,
+				nil,
+				executionDataStoreCache,
+			)
+			builder.executionStateCache = execution_state.NewExecutionStateCacheMock(snapshot)
+
+			return nil
+		}).
 		Component("public execution data service", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 			opts := []network.BlobServiceOption{
 				blob.WithBitswapOptions(
@@ -1593,6 +1649,8 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 					builder.stateStreamConf.ClientSendBufferSize,
 				),
 				executionDataTracker,
+				builder.executionResultInfoProvider,
+				builder.executionStateCache,
 			)
 			if err != nil {
 				return nil, fmt.Errorf("could not create state stream backend: %w", err)
@@ -1813,10 +1871,6 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 		builder.RegistersAsyncStore = execution.NewRegistersAsyncStore()
 		return nil
 	})
-	builder.Module("events storage", func(node *cmd.NodeConfig) error {
-		builder.events = store.NewEvents(node.Metrics.Cache, node.ProtocolDB)
-		return nil
-	})
 	builder.Module("reporter", func(node *cmd.NodeConfig) error {
 		builder.Reporter = index.NewReporter()
 		return nil
@@ -1972,30 +2026,6 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 			fixedENIdentifiers,
 		)
 
-		execNodeSelector := execution_result.NewExecutionNodeSelector(
-			preferredENIdentifiers,
-			fixedENIdentifiers,
-		)
-
-		execResultInfoProvider := execution_result.NewExecutionResultInfoProvider(
-			node.Logger,
-			node.State,
-			node.Storage.Receipts,
-			execNodeSelector,
-			optimistic_sync.DefaultCriteria,
-		)
-
-		// TODO: use real objects instead of mocks once they're implemented
-		snapshot := osyncsnapshot.NewSnapshotMock(
-			builder.events,
-			nil,
-			nil,
-			builder.lightTransactionResults,
-			nil,
-			nil,
-		)
-		execStateCache := execution_state.NewExecutionStateCacheMock(snapshot)
-
 		backendParams := backend.Params{
 			State:                node.State,
 			Blocks:               node.Storage.Blocks,
@@ -2027,8 +2057,8 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 			VersionControl:              builder.VersionControl,
 			ExecNodeIdentitiesProvider:  execNodeIdentitiesProvider,
 			MaxScriptAndArgumentSize:    config.BackendConfig.AccessConfig.MaxRequestMsgSize,
-			ExecutionResultInfoProvider: execResultInfoProvider,
-			ExecutionStateCache:         execStateCache,
+			ExecutionResultInfoProvider: builder.executionResultInfoProvider,
+			ExecutionStateCache:         builder.executionStateCache,
 			OperatorCriteria:            optimistic_sync.DefaultCriteria,
 		}
 

engine/access/integration_unsecure_grpc_server_test.go

@@ -297,6 +297,8 @@ func (suite *SameGRPCPortTestSuite) SetupTest() {
 		state_stream.DefaultRegisterIDsRequestLimit,
 		subscriptionHandler,
 		suite.executionDataTracker,
+		suite.executionResultInfoProvider,
+		suite.executionStateCache,
 	)
 	assert.NoError(suite.T(), err)
 

engine/access/state_stream/backend/backend.go

@@ -18,6 +18,7 @@ import (
 	"github.com/onflow/flow-go/module/execution"
 	"github.com/onflow/flow-go/module/executiondatasync/execution_data"
 	"github.com/onflow/flow-go/module/executiondatasync/execution_data/cache"
+	"github.com/onflow/flow-go/module/executiondatasync/optimistic_sync"
 	"github.com/onflow/flow-go/state/protocol"
 	"github.com/onflow/flow-go/storage"
 )
@@ -95,6 +96,8 @@ func New(
 	registerIDsRequestLimit int,
 	subscriptionHandler *subscription.SubscriptionHandler,
 	executionDataTracker tracker.ExecutionDataTracker,
+	executionResultProvider optimistic_sync.ExecutionResultInfoProvider,
+	executionStateCache optimistic_sync.ExecutionStateCache,
 ) (*StateStreamBackend, error) {
 	logger := log.With().Str("module", "state_stream_api").Logger()
 
@@ -113,11 +116,13 @@ func New(
 	}
 
 	b.ExecutionDataBackend = ExecutionDataBackend{
-		log:                  logger,
-		headers:              headers,
-		subscriptionHandler:  subscriptionHandler,
-		getExecutionData:     b.getExecutionData,
-		executionDataTracker: executionDataTracker,
+		log:                     logger,
+		headers:                 headers,
+		subscriptionHandler:     subscriptionHandler,
+		getExecutionData:        b.getExecutionData,
+		executionDataTracker:    executionDataTracker,
+		executionResultProvider: executionResultProvider,
+		executionStateCache:     executionStateCache,
 	}
 
 	eventsProvider := EventsProvider{