cleanup and fix tests

Author: peterargue

cmd/access/node_builder/access_node_builder.go

@@ -70,7 +70,6 @@ import (
 	"github.com/onflow/flow-go/module/metrics/unstaked"
 	"github.com/onflow/flow-go/module/state_synchronization"
 	"github.com/onflow/flow-go/module/state_synchronization/indexer"
-	"github.com/onflow/flow-go/module/state_synchronization/proxies"
 	edrequester "github.com/onflow/flow-go/module/state_synchronization/requester"
 	"github.com/onflow/flow-go/network"
 	alspmgr "github.com/onflow/flow-go/network/alsp/manager"
@@ -279,8 +278,8 @@ type FlowAccessNodeBuilder struct {
 	ExecutionIndexer           *indexer.Indexer
 	ExecutionIndexerCore       *indexer.IndexerCore
 	ScriptExecutor             *backend.ScriptExecutor
-	RegistersStoreProxy        *proxies.RegistersStore
-	IndexReporterProxy         *proxies.IndexReporter
+	RegistersAsyncStore        *execution.RegistersAsyncStore
+	EventsIndex                *backend.EventsIndex
 	IndexerDependencies        *cmd.DependencyList
 
 	// The sync engine participants provider is the libp2p peer store for the access node
@@ -775,16 +774,6 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 					return nil, err
 				}
 
-				err = builder.IndexReporterProxy.Initialize(builder.ExecutionIndexer)
-				if err != nil {
-					return nil, err
-				}
-
-				err = builder.RegistersStoreProxy.Initialize(registers)
-				if err != nil {
-					return nil, err
-				}
-
 				// setup requester to notify indexer when new execution data is received
 				execDataDistributor.AddOnExecutionDataReceivedConsumer(builder.ExecutionIndexer.OnExecutionData)
 
@@ -803,7 +792,20 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 					return nil, err
 				}
 
-				builder.ScriptExecutor.InitReporter(builder.ExecutionIndexer, scripts)
+				err = builder.ScriptExecutor.Initialize(builder.ExecutionIndexer, scripts)
+				if err != nil {
+					return nil, err
+				}
+
+				err = builder.EventsIndex.Initialize(builder.ExecutionIndexer)
+				if err != nil {
+					return nil, err
+				}
+
+				err = builder.RegistersAsyncStore.Initialize(registers)
+				if err != nil {
+					return nil, err
+				}
 
 				return builder.ExecutionIndexer, nil
 			}, builder.IndexerDependencies)
@@ -844,16 +846,15 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 				builder.stateStreamConf,
 				node.State,
 				node.Storage.Headers,
-				node.Storage.Events,
 				node.Storage.Seals,
 				node.Storage.Results,
 				builder.ExecutionDataStore,
 				executionDataStoreCache,
 				broadcaster,
 				builder.executionDataConfig.InitialBlockHeight,
 				highestAvailableHeight,
-				builder.RegistersStoreProxy,
-				builder.IndexReporterProxy,
+				builder.RegistersAsyncStore,
+				builder.EventsIndex,
 				useIndex,
 			)
 			if err != nil {
@@ -1445,18 +1446,18 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 			builder.ScriptExecutor = backend.NewScriptExecutor(builder.Logger, builder.scriptExecMinBlock, builder.scriptExecMaxBlock)
 			return nil
 		}).
-		Module("register store proxy", func(node *cmd.NodeConfig) error {
-			builder.RegistersStoreProxy = proxies.NewRegistersStore()
-			return nil
-		}).
-		Module("index reporter proxy", func(node *cmd.NodeConfig) error {
-			builder.IndexReporterProxy = proxies.NewIndexReporter()
+		Module("async register store", func(node *cmd.NodeConfig) error {
+			builder.RegistersAsyncStore = execution.NewRegistersAsyncStore()
 			return nil
 		}).
 		Module("events storage", func(node *cmd.NodeConfig) error {
 			builder.Storage.Events = bstorage.NewEvents(node.Metrics.Cache, node.DB)
 			return nil
 		}).
+		Module("events index", func(node *cmd.NodeConfig) error {
+			builder.EventsIndex = backend.NewEventsIndex(builder.Storage.Events)
+			return nil
+		}).
 		Component("RPC engine", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 			config := builder.rpcConf
 			backendConfig := config.BackendConfig
@@ -1508,7 +1509,6 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				HistoricalAccessNodes:     builder.HistoricalAccessRPCs,
 				Blocks:                    node.Storage.Blocks,
 				Headers:                   node.Storage.Headers,
-				Events:                    node.Storage.Events,
 				Collections:               node.Storage.Collections,
 				Transactions:              node.Storage.Transactions,
 				ExecutionReceipts:         node.Storage.Receipts,
@@ -1528,7 +1528,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				ScriptExecutor:            builder.ScriptExecutor,
 				ScriptExecutionMode:       scriptExecMode,
 				EventQueryMode:            eventQueryMode,
-				IndexReporter:             builder.IndexReporterProxy,
+				EventsIndex:               builder.EventsIndex,
 			})
 			if err != nil {
 				return nil, fmt.Errorf("could not initialize backend: %w", err)

engine/access/integration_unsecure_grpc_server_test.go

@@ -243,7 +243,6 @@ func (suite *SameGRPCPortTestSuite) SetupTest() {
 		conf,
 		suite.state,
 		suite.headers,
-		suite.events,
 		suite.seals,
 		suite.results,
 		nil,
@@ -252,6 +251,7 @@ func (suite *SameGRPCPortTestSuite) SetupTest() {
 		rootBlock.Header.Height,
 		rootBlock.Header.Height,
 		suite.registers,
+		backend.NewEventsIndex(suite.events),
 		false,
 	)
 	assert.NoError(suite.T(), err)

engine/access/rpc/backend/backend.go

@@ -18,7 +18,6 @@ import (
 	"github.com/onflow/flow-go/model/flow/filter"
 	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/execution"
-	"github.com/onflow/flow-go/module/state_synchronization"
 	"github.com/onflow/flow-go/state/protocol"
 	"github.com/onflow/flow-go/storage"
 )
@@ -84,7 +83,6 @@ type Params struct {
 	HistoricalAccessNodes     []accessproto.AccessAPIClient
 	Blocks                    storage.Blocks
 	Headers                   storage.Headers
-	Events                    storage.Events
 	Collections               storage.Collections
 	Transactions              storage.Transactions
 	ExecutionReceipts         storage.ExecutionReceipts
@@ -105,7 +103,7 @@ type Params struct {
 	ScriptExecutor            execution.ScriptExecutor
 	ScriptExecutionMode       IndexQueryMode
 	EventQueryMode            IndexQueryMode
-	IndexReporter             state_synchronization.IndexReporter
+	EventsIndex               *EventsIndex
 }
 
 // New creates backend instance
@@ -185,13 +183,12 @@ func New(params Params) (*Backend, error) {
 			chain:             params.ChainID.Chain(),
 			state:             params.State,
 			headers:           params.Headers,
-			events:            params.Events,
 			executionReceipts: params.ExecutionReceipts,
 			connFactory:       params.ConnFactory,
 			maxHeightRange:    params.MaxHeightRange,
 			nodeCommunicator:  params.Communicator,
 			queryMode:         params.EventQueryMode,
-			indexReporter:     params.IndexReporter,
+			eventsIndex:       params.EventsIndex,
 		},
 		backendBlockHeaders: backendBlockHeaders{
 			headers: params.Headers,

engine/access/rpc/backend/backend_accounts_test.go

@@ -15,7 +15,6 @@ import (
 	connectionmock "github.com/onflow/flow-go/engine/access/rpc/connection/mock"
 	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/flow"
-	"github.com/onflow/flow-go/module/execution"
 	execmock "github.com/onflow/flow-go/module/execution/mock"
 	"github.com/onflow/flow-go/module/irrecoverable"
 	protocol "github.com/onflow/flow-go/state/protocol/mock"
@@ -222,7 +221,7 @@ func (s *BackendAccountsSuite) TestGetAccountFromStorage_Fails() {
 		statusCode codes.Code
 	}{
 		{
-			err:        execution.ErrDataNotAvailable,
+			err:        storage.ErrHeightNotIndexed,
 			statusCode: codes.OutOfRange,
 		},
 		{
@@ -267,7 +266,7 @@ func (s *BackendAccountsSuite) TestGetAccountFromFailover_HappyPath() {
 	backend.scriptExecMode = IndexQueryModeFailover
 	backend.scriptExecutor = scriptExecutor
 
-	for _, errToReturn := range []error{execution.ErrDataNotAvailable, storage.ErrNotFound} {
+	for _, errToReturn := range []error{storage.ErrHeightNotIndexed, storage.ErrNotFound} {
 		scriptExecutor.On("GetAccountAtBlockHeight", mock.Anything, s.account.Address, s.block.Header.Height).
 			Return(nil, errToReturn).Times(3)
 
@@ -299,7 +298,7 @@ func (s *BackendAccountsSuite) TestGetAccountFromFailover_ReturnsENErrors() {
 
 	scriptExecutor := execmock.NewScriptExecutor(s.T())
 	scriptExecutor.On("GetAccountAtBlockHeight", mock.Anything, s.failingAddress, s.block.Header.Height).
-		Return(nil, execution.ErrDataNotAvailable)
+		Return(nil, storage.ErrHeightNotIndexed)
 
 	backend := s.defaultBackend()
 	backend.scriptExecMode = IndexQueryModeFailover

engine/access/rpc/backend/backend_events.go

@@ -20,16 +20,14 @@ import (
 	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/events"
 	"github.com/onflow/flow-go/model/flow"
-	"github.com/onflow/flow-go/module/execution"
 	"github.com/onflow/flow-go/module/irrecoverable"
-	"github.com/onflow/flow-go/module/state_synchronization"
+	"github.com/onflow/flow-go/module/state_synchronization/indexer"
 	"github.com/onflow/flow-go/state/protocol"
 	"github.com/onflow/flow-go/storage"
 )
 
 type backendEvents struct {
 	headers           storage.Headers
-	events            storage.Events
 	executionReceipts storage.ExecutionReceipts
 	state             protocol.State
 	chain             flow.Chain
@@ -38,7 +36,7 @@ type backendEvents struct {
 	maxHeightRange    uint
 	nodeCommunicator  Communicator
 	queryMode         IndexQueryMode
-	indexReporter     state_synchronization.IndexReporter
+	eventsIndex       *EventsIndex
 }
 
 // blockMetadata is used to capture information about requested blocks to avoid repeated blockID
@@ -230,25 +228,16 @@ func (b *backendEvents) getBlockEventsFromStorage(
 	missing := make([]blockMetadata, 0)
 	resp := make([]flow.BlockEvents, 0)
 
-	lowestHeight, highestHeight, err := b.getIndexerHeights()
-	if err != nil {
-		return nil, nil, err
-	}
-
 	for _, blockInfo := range blockInfos {
 		if ctx.Err() != nil {
 			return nil, nil, rpc.ConvertError(ctx.Err(), "failed to get events from storage", codes.Canceled)
 		}
 
-		if blockInfo.Height < lowestHeight || blockInfo.Height > highestHeight {
-			missing = append(missing, blockInfo)
-			continue
-		}
-
-		events, err := b.events.ByBlockID(blockInfo.ID)
+		events, err := b.eventsIndex.GetEvents(blockInfo.ID, blockInfo.Height)
 		if err != nil {
-			// Note: if there are no events for a block, an empty slice is returned
-			if errors.Is(err, storage.ErrNotFound) {
+			if errors.Is(err, storage.ErrNotFound) ||
+				errors.Is(err, storage.ErrHeightNotIndexed) ||
+				errors.Is(err, indexer.ErrIndexNotInitialized) {
 				missing = append(missing, blockInfo)
 				continue
 			}
@@ -443,29 +432,3 @@ func (b *backendEvents) tryGetEvents(ctx context.Context,
 
 	return execRPCClient.GetEventsForBlockIDs(ctx, req)
 }
-
-// getIndexerHeights returns the lowest and highest indexed block heights
-// Expected errors during normal operation:
-// - codes.FailedPrecondition: if the index reporter is not ready yet.
-// - codes.Internal: if there was any other error getting the heights.
-func (b *backendEvents) getIndexerHeights() (uint64, uint64, error) {
-	lowestHeight, err := b.indexReporter.LowestIndexedHeight()
-	if err != nil {
-		if errors.Is(err, execution.ErrDataNotAvailable) {
-			// if the index is not ready yet, but likely will be eventually
-			return 0, 0, status.Errorf(codes.FailedPrecondition, "failed to get lowest indexed height: %v", err)
-		}
-		return 0, 0, rpc.ConvertError(err, "failed to get lowest indexed height", codes.Internal)
-	}
-
-	highestHeight, err := b.indexReporter.HighestIndexedHeight()
-	if err != nil {
-		if errors.Is(err, execution.ErrDataNotAvailable) {
-			// if the index is not ready yet, but likely will be eventually
-			return 0, 0, status.Errorf(codes.FailedPrecondition, "failed to get highest indexed height: %v", err)
-		}
-		return 0, 0, rpc.ConvertError(err, "failed to get highest indexed height", codes.Internal)
-	}
-
-	return lowestHeight, highestHeight, nil
-}