[Access] Enable Event streaming on REST API

cmd/access/node_builder/access_node_builder.go

@@ -35,13 +35,15 @@ import (
 	"github.com/onflow/flow-go/consensus/hotstuff/verification"
 	recovery "github.com/onflow/flow-go/consensus/recovery/protocol"
 	"github.com/onflow/flow-go/crypto"
+	"github.com/onflow/flow-go/engine"
 	"github.com/onflow/flow-go/engine/access/ingestion"
 	pingeng "github.com/onflow/flow-go/engine/access/ping"
 	"github.com/onflow/flow-go/engine/access/rpc"
 	"github.com/onflow/flow-go/engine/access/rpc/backend"
 	"github.com/onflow/flow-go/engine/access/state_stream"
 	followereng "github.com/onflow/flow-go/engine/common/follower"
 	"github.com/onflow/flow-go/engine/common/requester"
+	common_state_stream "github.com/onflow/flow-go/engine/common/state_stream"
 	synceng "github.com/onflow/flow-go/engine/common/synchronization"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/model/flow/filter"
@@ -115,7 +117,8 @@ type AccessNodeConfig struct {
 	apiRatelimits                map[string]int
 	apiBurstlimits               map[string]int
 	rpcConf                      rpc.Config
-	stateStreamConf              state_stream.Config
+	stateStreamConf              common_state_stream.Config
+	stateStreamBackend           common_state_stream.API
 	stateStreamFilterConf        map[string]int
 	ExecutionNodeAddress         string // deprecated
 	HistoricalAccessRPCs         []access.AccessAPIClient
@@ -161,15 +164,16 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 			ArchiveAddressList:        nil,
 			MaxMsgSize:                grpcutils.DefaultMaxMsgSize,
 		},
-		stateStreamConf: state_stream.Config{
+		stateStreamConf: common_state_stream.Config{
 			MaxExecutionDataMsgSize: grpcutils.DefaultMaxMsgSize,
-			ExecutionDataCacheSize:  state_stream.DefaultCacheSize,
-			ClientSendTimeout:       state_stream.DefaultSendTimeout,
-			ClientSendBufferSize:    state_stream.DefaultSendBufferSize,
-			MaxGlobalStreams:        state_stream.DefaultMaxGlobalStreams,
-			EventFilterConfig:       state_stream.DefaultEventFilterConfig,
-			ResponseLimit:           state_stream.DefaultResponseLimit,
+			ExecutionDataCacheSize:  common_state_stream.DefaultCacheSize,
+			ClientSendTimeout:       common_state_stream.DefaultSendTimeout,
+			ClientSendBufferSize:    common_state_stream.DefaultSendBufferSize,
+			MaxGlobalStreams:        common_state_stream.DefaultMaxGlobalStreams,
+			EventFilterConfig:       common_state_stream.DefaultEventFilterConfig,
+			ResponseLimit:           common_state_stream.DefaultResponseLimit,
 		},
+		stateStreamBackend:           nil,
 		stateStreamFilterConf:        nil,
 		ExecutionNodeAddress:         "localhost:9000",
 		logTxTimeToFinalized:         false,
@@ -600,20 +604,35 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionDataRequester() *FlowAccessN
 				return nil, fmt.Errorf("could not get highest consecutive height: %w", err)
 			}
 
-			stateStreamEng, err := state_stream.NewEng(
+			broadcaster := engine.NewBroadcaster()
+
+			backend, err := common_state_stream.New(
 				node.Logger,
 				builder.stateStreamConf,
-				builder.ExecutionDataStore,
-				executionDataCache,
 				node.State,
 				node.Storage.Headers,
 				node.Storage.Seals,
 				node.Storage.Results,
-				node.RootChainID,
+				builder.ExecutionDataStore,
+				executionDataCache,
+				broadcaster,
 				builder.executionDataConfig.InitialBlockHeight,
 				highestAvailableHeight,
+			)
+			if err != nil {
+				return nil, fmt.Errorf("could not create state stream backend: %w", err)
+			}
+
+			stateStreamEng, err := state_stream.NewEng(
+				node.Logger,
+				builder.stateStreamConf,
+				executionDataCache,
+				node.Storage.Headers,
+				node.RootChainID,
 				builder.apiRatelimits,
 				builder.apiBurstlimits,
+				backend,
+				broadcaster,
 			)
 			if err != nil {
 				return nil, fmt.Errorf("could not create state stream engine: %w", err)
@@ -892,6 +911,10 @@ func (builder *FlowAccessNodeBuilder) enqueueRelayNetwork() {
 }
 
 func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
+	if builder.executionDataSyncEnabled {
+		builder.BuildExecutionDataRequester()
+	}
+
 	builder.
 		BuildConsensusFollower().
 		Module("collection node client", func(node *cmd.NodeConfig) error {
@@ -1008,6 +1031,9 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				builder.apiRatelimits,
 				builder.apiBurstlimits,
 				builder.Me,
+				builder.stateStreamBackend,
+				builder.stateStreamConf.EventFilterConfig,
+				builder.stateStreamConf.MaxGlobalStreams,
 			)
 			if err != nil {
 				return nil, err
@@ -1093,10 +1119,6 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 		})
 	}
 
-	if builder.executionDataSyncEnabled {
-		builder.BuildExecutionDataRequester()
-	}
-
 	builder.Component("ping engine", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 		ping, err := pingeng.New(
 			node.Logger,

cmd/observer/node_builder/observer_builder.go

@@ -31,6 +31,7 @@ import (
 	"github.com/onflow/flow-go/engine/access/rpc"
 	"github.com/onflow/flow-go/engine/access/rpc/backend"
 	"github.com/onflow/flow-go/engine/common/follower"
+	"github.com/onflow/flow-go/engine/common/state_stream"
 	synceng "github.com/onflow/flow-go/engine/common/synchronization"
 	"github.com/onflow/flow-go/engine/protocol"
 	"github.com/onflow/flow-go/model/encodable"
@@ -862,6 +863,9 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 			builder.apiRatelimits,
 			builder.apiBurstlimits,
 			builder.Me,
+			nil,
+			state_stream.DefaultEventFilterConfig,
+			0,
 		)
 		if err != nil {
 			return nil, err

engine/access/rest/events.go

@@ -3,10 +3,9 @@ package rest
 import (
 	"fmt"
 
+	"github.com/onflow/flow-go/access"
 	"github.com/onflow/flow-go/engine/access/rest/models"
 	"github.com/onflow/flow-go/engine/access/rest/request"
-
-	"github.com/onflow/flow-go/access"
 )
 
 const blockQueryParam = "block_ids"

engine/access/rest/handler.go

@@ -1,26 +1,17 @@
 package rest
 
 import (
-	"encoding/json"
-	"errors"
-	"fmt"
 	"net/http"
 
+	"github.com/rs/zerolog"
+
+	"github.com/onflow/flow-go/access"
 	"github.com/onflow/flow-go/engine/access/rest/models"
 	"github.com/onflow/flow-go/engine/access/rest/request"
 	"github.com/onflow/flow-go/engine/access/rest/util"
-	fvmErrors "github.com/onflow/flow-go/fvm/errors"
 	"github.com/onflow/flow-go/model/flow"
-
-	"github.com/rs/zerolog"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
-
-	"github.com/onflow/flow-go/access"
 )
 
-const MaxRequestSize = 2 << 20 // 2MB
-
 // ApiHandlerFunc is a function that contains endpoint handling logic,
 // it fetches necessary resources and returns an error or response model.
 type ApiHandlerFunc func(
@@ -33,11 +24,10 @@ type ApiHandlerFunc func(
 // Handler function allows easier handling of errors and responses as it
 // wraps functionality for handling error and responses outside of endpoint handling.
 type Handler struct {
-	logger         zerolog.Logger
+	*HttpHandler
 	backend        access.API
 	linkGenerator  models.LinkGenerator
 	apiHandlerFunc ApiHandlerFunc
-	chain          flow.Chain
 }
 
 func NewHandler(
@@ -47,31 +37,26 @@ func NewHandler(
 	generator models.LinkGenerator,
 	chain flow.Chain,
 ) *Handler {
-	return &Handler{
-		logger:         logger,
+	handler := &Handler{
 		backend:        backend,
 		apiHandlerFunc: handlerFunc,
 		linkGenerator:  generator,
-		chain:          chain,
 	}
+	handler.HttpHandler = NewHttpHandler(logger, chain)
+	return handler
 }
 
 // ServerHTTP function acts as a wrapper to each request providing common handling functionality
 // such as logging, error handling, request decorators
 func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	// create a logger
-	errLog := h.logger.With().Str("request_url", r.URL.String()).Logger()
+	errLog := h.Logger.With().Str("request_url", r.URL.String()).Logger()
 
-	// limit requested body size
-	r.Body = http.MaxBytesReader(w, r.Body, MaxRequestSize)
-	err := r.ParseForm()
+	err := h.VerifyRequest(w, r)
 	if err != nil {
-		h.errorHandler(w, err, errLog)
 		return
 	}
-
-	// create request decorator with parsed values
-	decoratedRequest := request.Decorate(r, h.chain)
+	decoratedRequest := request.Decorate(r, h.Chain)
 
 	// execute handler function and check for error
 	response, err := h.apiHandlerFunc(decoratedRequest, h.backend, h.linkGenerator)
@@ -90,80 +75,3 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	// write response to response stream
 	h.jsonResponse(w, http.StatusOK, response, errLog)
 }
-
-func (h *Handler) errorHandler(w http.ResponseWriter, err error, errorLogger zerolog.Logger) {
-	// rest status type error should be returned with status and user message provided
-	var statusErr StatusError
-	if errors.As(err, &statusErr) {
-		h.errorResponse(w, statusErr.Status(), statusErr.UserMessage(), errorLogger)
-		return
-	}
-
-	// handle cadence errors
-	cadenceError := fvmErrors.Find(err, fvmErrors.ErrCodeCadenceRunTimeError)
-	if cadenceError != nil {
-		msg := fmt.Sprintf("Cadence error: %s", cadenceError.Error())
-		h.errorResponse(w, http.StatusBadRequest, msg, errorLogger)
-		return
-	}
-
-	// handle grpc status error returned from the backend calls, we are forwarding the message to the client
-	if se, ok := status.FromError(err); ok {
-		if se.Code() == codes.NotFound {
-			msg := fmt.Sprintf("Flow resource not found: %s", se.Message())
-			h.errorResponse(w, http.StatusNotFound, msg, errorLogger)
-			return
-		}
-		if se.Code() == codes.InvalidArgument {
-			msg := fmt.Sprintf("Invalid Flow argument: %s", se.Message())
-			h.errorResponse(w, http.StatusBadRequest, msg, errorLogger)
-			return
-		}
-		if se.Code() == codes.Internal {
-			msg := fmt.Sprintf("Invalid Flow request: %s", se.Message())
-			h.errorResponse(w, http.StatusBadRequest, msg, errorLogger)
-			return
-		}
-	}
-
-	// stop going further - catch all error
-	msg := "internal server error"
-	errorLogger.Error().Err(err).Msg(msg)
-	h.errorResponse(w, http.StatusInternalServerError, msg, errorLogger)
-}
-
-// jsonResponse builds a JSON response and send it to the client
-func (h *Handler) jsonResponse(w http.ResponseWriter, code int, response interface{}, errLogger zerolog.Logger) {
-	w.Header().Set("Content-Type", "application/json; charset=UTF-8")
-
-	// serialize response to JSON and handler errors
-	encodedResponse, err := json.MarshalIndent(response, "", "\t")
-	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		errLogger.Error().Err(err).Str("response", string(encodedResponse)).Msg("failed to indent response")
-		return
-	}
-
-	w.WriteHeader(code)
-	// write response to response stream
-	_, err = w.Write(encodedResponse)
-	if err != nil {
-		errLogger.Error().Err(err).Str("response", string(encodedResponse)).Msg("failed to write http response")
-	}
-}
-
-// errorResponse sends an HTTP error response to the client with the given return code
-// and a model error with the given response message in the response body
-func (h *Handler) errorResponse(
-	w http.ResponseWriter,
-	returnCode int,
-	responseMessage string,
-	logger zerolog.Logger,
-) {
-	// create error response model
-	modelError := models.ModelError{
-		Code:    int32(returnCode),
-		Message: responseMessage,
-	}
-	h.jsonResponse(w, returnCode, modelError, logger)
-}