[Access] Use local event for event streaming API

cmd/access/node_builder/access_node_builder.go

@@ -818,6 +818,7 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 				builder.stateStreamConf,
 				node.State,
 				node.Storage.Headers,
+				node.Storage.Events,
 				node.Storage.Seals,
 				node.Storage.Results,
 				builder.ExecutionDataStore,

engine/access/integration_unsecure_grpc_server_test.go

@@ -63,6 +63,7 @@ type SameGRPCPortTestSuite struct {
 	// storage
 	blocks       *storagemock.Blocks
 	headers      *storagemock.Headers
+	events       *storagemock.Events
 	collections  *storagemock.Collections
 	transactions *storagemock.Transactions
 	receipts     *storagemock.ExecutionReceipts
@@ -101,6 +102,7 @@ func (suite *SameGRPCPortTestSuite) SetupTest() {
 	suite.snapshot.On("Epochs").Return(suite.epochQuery).Maybe()
 	suite.blocks = new(storagemock.Blocks)
 	suite.headers = new(storagemock.Headers)
+	suite.events = new(storagemock.Events)
 	suite.transactions = new(storagemock.Transactions)
 	suite.collections = new(storagemock.Collections)
 	suite.receipts = new(storagemock.ExecutionReceipts)
@@ -241,6 +243,7 @@ func (suite *SameGRPCPortTestSuite) SetupTest() {
 		conf,
 		suite.state,
 		suite.headers,
+		suite.events,
 		suite.seals,
 		suite.results,
 		nil,

engine/access/state_stream/backend/backend.go

@@ -1,4 +1,4 @@
 package backend
 
 import (
 	"context"
@@ -90,6 +90,7 @@
 	config Config,
 	state protocol.State,
 	headers storage.Headers,
+	events storage.Events,
 	seals storage.Seals,
 	results storage.ExecutionResults,
 	execDataStore execution_data.ExecutionDataStore,
@@ -137,6 +138,8 @@
 	b.EventsBackend = EventsBackend{
 		log:              logger,
 		broadcaster:      broadcaster,
+		headers:          headers,
+		events:           events,
 		sendTimeout:      config.ClientSendTimeout,
 		responseLimit:    config.ResponseLimit,
 		sendBufferSize:   int(config.ClientSendBufferSize),

engine/access/state_stream/backend/backend_events.go

@@ -10,6 +10,7 @@ import (
 	"github.com/onflow/flow-go/engine"
 	"github.com/onflow/flow-go/engine/access/state_stream"
 	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/storage"
 	"github.com/onflow/flow-go/utils/logging"
 )
 
@@ -25,9 +26,13 @@ type EventsBackend struct {
 	sendTimeout    time.Duration
 	responseLimit  float64
 	sendBufferSize int
+	events         storage.Events
+	headers        storage.Headers
 
 	getExecutionData GetExecutionDataFunc
 	getStartHeight   GetStartHeightFunc
+
+	useIndex bool
 }
 
 func (b EventsBackend) SubscribeEvents(ctx context.Context, startBlockID flow.Identifier, startHeight uint64, filter state_stream.EventFilter) state_stream.Subscription {
@@ -36,14 +41,21 @@ func (b EventsBackend) SubscribeEvents(ctx context.Context, startBlockID flow.Id
 		return NewFailedSubscription(err, "could not get start height")
 	}
 
-	sub := NewHeightBasedSubscription(b.sendBufferSize, nextHeight, b.getResponseFactory(filter))
+	var responseFactory GetDataByHeightFunc
+	if b.useIndex {
+		responseFactory = b.getStorageResponseFactory(filter)
+	} else {
+		responseFactory = b.getExecutionDataResponseFactory(filter)
+	}
+
+	sub := NewHeightBasedSubscription(b.sendBufferSize, nextHeight, responseFactory)
 
 	go NewStreamer(b.log, b.broadcaster, b.sendTimeout, b.responseLimit, sub).Stream(ctx)
 
 	return sub
 }
 
-func (b EventsBackend) getResponseFactory(filter state_stream.EventFilter) GetDataByHeightFunc {
+func (b EventsBackend) getExecutionDataResponseFactory(filter state_stream.EventFilter) GetDataByHeightFunc {
 	return func(ctx context.Context, height uint64) (interface{}, error) {
 		executionData, err := b.getExecutionData(ctx, height)
 		if err != nil {
@@ -67,3 +79,31 @@ func (b EventsBackend) getResponseFactory(filter state_stream.EventFilter) GetDa
 		}, nil
 	}
 }
+
+func (b EventsBackend) getStorageResponseFactory(filter state_stream.EventFilter) GetDataByHeightFunc {
+	return func(ctx context.Context, height uint64) (interface{}, error) {
+		header, err := b.headers.ByHeight(height)
+		if err != nil {
+			return nil, fmt.Errorf("could not get header for height %d: %w", height, err)
+		}
+		blockID := header.ID()
+
+		events, err := b.events.ByBlockID(blockID)
+		if err != nil {
+			return nil, fmt.Errorf("could not get events for block %d: %w", height, err)
+		}
+
+		events = filter.Filter(events)
+
+		b.log.Trace().
+			Hex("block_id", logging.ID(blockID)).
+			Uint64("height", height).
+			Msgf("sending %d events", len(events))
+
+		return &EventsResponse{
+			BlockID: blockID,
+			Height:  height,
+			Events:  events,
+		}, nil
+	}
+}

engine/access/state_stream/backend/backend_executiondata_test.go

@@ -75,6 +75,7 @@
 	s.snapshot = protocolmock.NewSnapshot(s.T())
 	s.params = protocolmock.NewParams(s.T())
 	s.headers = storagemock.NewHeaders(s.T())
+	s.events = storagemock.NewEvents(s.T())
 	s.seals = storagemock.NewSeals(s.T())
 	s.results = storagemock.NewExecutionResults(s.T())
 
@@ -253,6 +254,7 @@
 		conf,
 		s.state,
 		s.headers,
+		s.events,
 		s.seals,
 		s.results,
 		s.eds,