add global stream limit

Author: peterargue

cmd/access/node_builder/access_node_builder.go

@@ -112,6 +112,8 @@ type AccessNodeConfig struct {
 	apiRatelimits                map[string]int
 	apiBurstlimits               map[string]int
 	rpcConf                      rpc.Config
+	stateStreamConf              state_stream.Config
+	stateStreamFilterConf        map[string]int
 	ExecutionNodeAddress         string // deprecated
 	HistoricalAccessRPCs         []access.AccessAPIClient
 	logTxTimeToFinalized         bool
@@ -143,7 +145,6 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 		rpcConf: rpc.Config{
 			UnsecureGRPCListenAddr:    "0.0.0.0:9000",
 			SecureGRPCListenAddr:      "0.0.0.0:9001",
-			StateStreamListenAddr:     "",
 			HTTPListenAddr:            "0.0.0.0:8000",
 			RESTListenAddr:            "",
 			CollectionAddr:            "",
@@ -154,9 +155,17 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 			MaxHeightRange:            backend.DefaultMaxHeightRange,
 			PreferredExecutionNodeIDs: nil,
 			FixedExecutionNodeIDs:     nil,
-			MaxExecutionDataMsgSize:   grpcutils.DefaultMaxMsgSize,
 			MaxMsgSize:                grpcutils.DefaultMaxMsgSize,
 		},
+		stateStreamConf: 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,
+		},
+		stateStreamFilterConf:        nil,
 		ExecutionNodeAddress:         "localhost:9000",
 		logTxTimeToFinalized:         false,
 		logTxTimeToExecuted:          false,
@@ -553,27 +562,33 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionDataRequester() *FlowAccessN
 			return builder.ExecutionDataRequester, nil
 		})
 
-	if builder.rpcConf.StateStreamListenAddr != "" {
+	if builder.stateStreamConf.ListenAddr != "" {
 		builder.Component("exec state stream engine", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
-			conf := state_stream.Config{
-				ListenAddr:              builder.rpcConf.StateStreamListenAddr,
-				MaxExecutionDataMsgSize: builder.rpcConf.MaxExecutionDataMsgSize,
-				RpcMetricsEnabled:       builder.rpcMetricsEnabled,
+			for key, value := range builder.stateStreamFilterConf {
+				switch key {
+				case "EventTypes":
+					builder.stateStreamConf.MaxEventTypes = value
+				case "Addresses":
+					builder.stateStreamConf.MaxAddresses = value
+				case "Contracts":
+					builder.stateStreamConf.MaxContracts = value
+				}
 			}
+			builder.stateStreamConf.RpcMetricsEnabled = builder.rpcMetricsEnabled
 
 			var heroCacheCollector module.HeroCacheMetrics = metrics.NewNoopCollector()
 			if builder.HeroCacheMetricsEnable {
 				heroCacheCollector = metrics.AccessNodeExecutionDataCacheMetrics(builder.MetricsRegisterer)
 			}
 
 			stateStreamEng, err := state_stream.NewEng(
-				conf,
+				node.Logger,
+				builder.stateStreamConf,
 				builder.ExecutionDataStore,
 				node.State,
 				node.Storage.Headers,
 				node.Storage.Seals,
 				node.Storage.Results,
-				node.Logger,
 				node.RootChainID,
 				builder.apiRatelimits,
 				builder.apiBurstlimits,
@@ -620,7 +635,7 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 		flags.UintVar(&builder.executionGRPCPort, "execution-ingress-port", defaultConfig.executionGRPCPort, "the grpc ingress port for all execution nodes")
 		flags.StringVarP(&builder.rpcConf.UnsecureGRPCListenAddr, "rpc-addr", "r", defaultConfig.rpcConf.UnsecureGRPCListenAddr, "the address the unsecured gRPC server listens on")
 		flags.StringVar(&builder.rpcConf.SecureGRPCListenAddr, "secure-rpc-addr", defaultConfig.rpcConf.SecureGRPCListenAddr, "the address the secure gRPC server listens on")
-		flags.StringVar(&builder.rpcConf.StateStreamListenAddr, "state-stream-addr", defaultConfig.rpcConf.StateStreamListenAddr, "the address the state stream server listens on (if empty the server will not be started)")
+		flags.StringVar(&builder.stateStreamConf.ListenAddr, "state-stream-addr", defaultConfig.stateStreamConf.ListenAddr, "the address the state stream server listens on (if empty the server will not be started)")
 		flags.StringVarP(&builder.rpcConf.HTTPListenAddr, "http-addr", "h", defaultConfig.rpcConf.HTTPListenAddr, "the address the http proxy server listens on")
 		flags.StringVar(&builder.rpcConf.RESTListenAddr, "rest-addr", defaultConfig.rpcConf.RESTListenAddr, "the address the REST server listens on (if empty the REST server will not be started)")
 		flags.StringVarP(&builder.rpcConf.CollectionAddr, "static-collection-ingress-addr", "", defaultConfig.rpcConf.CollectionAddr, "the address (of the collection node) to send transactions to")
@@ -631,7 +646,6 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 		flags.UintVar(&builder.rpcConf.ConnectionPoolSize, "connection-pool-size", defaultConfig.rpcConf.ConnectionPoolSize, "maximum number of connections allowed in the connection pool, size of 0 disables the connection pooling, and anything less than the default size will be overridden to use the default size")
 		flags.UintVar(&builder.rpcConf.MaxMsgSize, "rpc-max-message-size", grpcutils.DefaultMaxMsgSize, "the maximum message size in bytes for messages sent or received over grpc")
 		flags.UintVar(&builder.rpcConf.MaxHeightRange, "rpc-max-height-range", defaultConfig.rpcConf.MaxHeightRange, "maximum size for height range requests")
-		flags.UintVar(&builder.rpcConf.MaxExecutionDataMsgSize, "max-block-msg-size", defaultConfig.rpcConf.MaxExecutionDataMsgSize, "maximum size for a gRPC message containing block execution data")
 		flags.StringSliceVar(&builder.rpcConf.PreferredExecutionNodeIDs, "preferred-execution-node-ids", defaultConfig.rpcConf.PreferredExecutionNodeIDs, "comma separated list of execution nodes ids to choose from when making an upstream call e.g. b4a4dbdcd443d...,fb386a6a... etc.")
 		flags.StringSliceVar(&builder.rpcConf.FixedExecutionNodeIDs, "fixed-execution-node-ids", defaultConfig.rpcConf.FixedExecutionNodeIDs, "comma separated list of execution nodes ids to choose from when making an upstream call if no matching preferred execution id is found e.g. b4a4dbdcd443d...,fb386a6a... etc.")
 		flags.BoolVar(&builder.logTxTimeToFinalized, "log-tx-time-to-finalized", defaultConfig.logTxTimeToFinalized, "log transaction time to finalized")
@@ -655,6 +669,14 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 		flags.DurationVar(&builder.executionDataConfig.MaxFetchTimeout, "execution-data-max-fetch-timeout", defaultConfig.executionDataConfig.MaxFetchTimeout, "maximum timeout to use when fetching execution data from the network e.g. 300s")
 		flags.DurationVar(&builder.executionDataConfig.RetryDelay, "execution-data-retry-delay", defaultConfig.executionDataConfig.RetryDelay, "initial delay for exponential backoff when fetching execution data fails e.g. 10s")
 		flags.DurationVar(&builder.executionDataConfig.MaxRetryDelay, "execution-data-max-retry-delay", defaultConfig.executionDataConfig.MaxRetryDelay, "maximum delay for exponential backoff when fetching execution data fails e.g. 5m")
+
+		// Execution State Streaming API
+		flags.Uint32Var(&builder.stateStreamConf.ExecutionDataCacheSize, "execution-data-cache-size", defaultConfig.stateStreamConf.ExecutionDataCacheSize, "block execution data cache size")
+		flags.Uint32Var(&builder.stateStreamConf.MaxGlobalStreams, "state-stream-global-max-streams", defaultConfig.stateStreamConf.MaxGlobalStreams, "global maximum number of concurrent streams")
+		flags.UintVar(&builder.stateStreamConf.MaxExecutionDataMsgSize, "state-stream-max-message-size", defaultConfig.stateStreamConf.MaxExecutionDataMsgSize, "maximum size for a gRPC message containing block execution data")
+		flags.DurationVar(&builder.stateStreamConf.ClientSendTimeout, "state-stream-send-timeout", defaultConfig.stateStreamConf.ClientSendTimeout, "maximum wait before timing out while sending a response to a streaming client e.g. 30s")
+		flags.UintVar(&builder.stateStreamConf.ClientSendBufferSize, "state-stream-send-buffer-size", defaultConfig.stateStreamConf.ClientSendBufferSize, "maximum number of responses to buffer within a stream")
+		flags.StringToIntVar(&builder.stateStreamFilterConf, "state-stream-event-filter-limits", defaultConfig.stateStreamFilterConf, "event filter limits for ExecutionData SubscribeEvents API e.g. EventTypes=100,Addresses=100,Contracts=100 etc.")
 	}).ValidateFlags(func() error {
 		if builder.supportsObserver && (builder.PublicNetworkConfig.BindAddress == cmd.NotSet || builder.PublicNetworkConfig.BindAddress == "") {
 			return errors.New("public-network-address must be set if supports-observer is true")
@@ -676,6 +698,27 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 				return errors.New("execution-data-max-search-ahead must be greater than 0")
 			}
 		}
+		if builder.stateStreamConf.ListenAddr != "" {
+			if builder.stateStreamConf.ExecutionDataCacheSize == 0 {
+				return errors.New("execution-data-cache-size must be greater than 0")
+			}
+			if builder.stateStreamConf.ClientSendBufferSize == 0 {
+				return errors.New("state-stream-send-buffer-size must be greater than 0")
+			}
+			if len(builder.stateStreamFilterConf) > 3 {
+				return errors.New("state-stream-event-filter-limits must have at most 3 keys (EventTypes, Addresses, Contracts)")
+			}
+			for key, value := range builder.stateStreamFilterConf {
+				switch key {
+				case "EventTypes", "Addresses", "Contracts":
+					if value <= 0 {
+						return fmt.Errorf("state-stream-event-filter-limits %s must be greater than 0", key)
+					}
+				default:
+					return errors.New("state-stream-event-filter-limits may only contain the keys EventTypes, Addresses, Contracts")
+				}
+			}
+		}
 
 		return nil
 	})

engine/access/rpc/engine.go

@@ -32,14 +32,12 @@ import (
 type Config struct {
 	UnsecureGRPCListenAddr    string                           // the non-secure GRPC server address as ip:port
 	SecureGRPCListenAddr      string                           // the secure GRPC server address as ip:port
-	StateStreamListenAddr     string                           // the state stream GRPC server address as ip:port
 	TransportCredentials      credentials.TransportCredentials // the secure GRPC credentials
 	HTTPListenAddr            string                           // the HTTP web proxy address as ip:port
 	RESTListenAddr            string                           // the REST server address as ip:port (if empty the REST server will not be started)
 	CollectionAddr            string                           // the address of the upstream collection node
 	HistoricalAccessAddrs     string                           // the list of all access nodes from previous spork
 	MaxMsgSize                uint                             // GRPC max message size
-	MaxExecutionDataMsgSize   uint                             // GRPC max message size for block execution data
 	ExecutionClientTimeout    time.Duration                    // execution API GRPC client timeout
 	CollectionClientTimeout   time.Duration                    // collection API GRPC client timeout
 	ConnectionPoolSize        uint                             // size of the cache for storing collection and execution connections

engine/access/state_stream/backend.go

@@ -20,6 +20,10 @@ import (
 )
 
 const (
+	// DefaultMaxGlobalStreams defines the default max number of streams that can be open at the same time.
+	DefaultMaxGlobalStreams = 1000
+
+	// DefaultCacheSize defines the default max number of objects for the execution data cache.
 	DefaultCacheSize = 100
 
 	// DefaultSendTimeout is the default timeout for sending a message to the client. After the timeout
@@ -48,11 +52,11 @@ type StateStreamBackend struct {
 	execDataStore execution_data.ExecutionDataStore
 	execDataCache *herocache.Cache
 	broadcaster   *engine.Broadcaster
-	sendTimeout   time.Duration
 }
 
 func New(
 	log zerolog.Logger,
+	config Config,
 	state protocol.State,
 	headers storage.Headers,
 	seals storage.Seals,
@@ -72,14 +76,14 @@ func New(
 		execDataStore: execDataStore,
 		execDataCache: execDataCache,
 		broadcaster:   broadcaster,
-		sendTimeout:   DefaultSendTimeout,
 	}
 
 	b.ExecutionDataBackend = ExecutionDataBackend{
 		log:              logger,
 		headers:          headers,
 		broadcaster:      broadcaster,
-		sendTimeout:      DefaultSendTimeout,
+		sendTimeout:      config.ClientSendTimeout,
+		sendBufferSize:   int(config.ClientSendBufferSize),
 		getExecutionData: b.getExecutionData,
 		getStartHeight:   b.getStartHeight,
 	}
@@ -88,7 +92,8 @@ func New(
 		log:              logger,
 		headers:          headers,
 		broadcaster:      broadcaster,
-		sendTimeout:      DefaultSendTimeout,
+		sendTimeout:      config.ClientSendTimeout,
+		sendBufferSize:   int(config.ClientSendBufferSize),
 		getExecutionData: b.getExecutionData,
 		getStartHeight:   b.getStartHeight,
 	}

engine/access/state_stream/backend_events.go

@@ -21,10 +21,11 @@ type EventsResponse struct {
 }
 
 type EventsBackend struct {
-	log         zerolog.Logger
-	headers     storage.Headers
-	broadcaster *engine.Broadcaster
-	sendTimeout time.Duration
+	log            zerolog.Logger
+	headers        storage.Headers
+	broadcaster    *engine.Broadcaster
+	sendTimeout    time.Duration
+	sendBufferSize int
 
 	getExecutionData GetExecutionDataFunc
 	getStartHeight   GetStartHeightFunc
@@ -33,7 +34,7 @@ type EventsBackend struct {
 func (b EventsBackend) SubscribeEvents(ctx context.Context, startBlockID flow.Identifier, startHeight uint64, filter EventFilter) Subscription {
 	nextHeight, err := b.getStartHeight(startBlockID, startHeight)
 	if err != nil {
-		sub := NewSubscription()
+		sub := NewSubscription(b.sendBufferSize)
 		if st, ok := status.FromError(err); ok {
 			sub.Fail(status.Errorf(st.Code(), "could not get start height: %s", st.Message()))
 			return sub
@@ -43,7 +44,7 @@ func (b EventsBackend) SubscribeEvents(ctx context.Context, startBlockID flow.Id
 		return sub
 	}
 
-	sub := NewHeightBasedSubscription(nextHeight, b.getResponseFactory(filter))
+	sub := NewHeightBasedSubscription(b.sendBufferSize, nextHeight, b.getResponseFactory(filter))
 
 	go NewStreamer(b.log, b.broadcaster, b.sendTimeout, sub).Stream(ctx)
 

engine/access/state_stream/backend_events_test.go

@@ -77,13 +77,13 @@ func (s *BackendEventsSuite) TestSubscribeEvents() {
 
 		t2 := test
 		t2.name = fmt.Sprintf("%s - some events", test.name)
-		t2.filters, err = NewEventFilter(chain, []string{string(testEventTypes[0])}, nil, nil)
+		t2.filters, err = NewEventFilter(DefaultEventFilterConfig, chain, []string{string(testEventTypes[0])}, nil, nil)
 		require.NoError(s.T(), err)
 		tests = append(tests, t2)
 
 		t3 := test
 		t3.name = fmt.Sprintf("%s - no events", test.name)
-		t3.filters, err = NewEventFilter(chain, []string{"A.0x1.NonExistent.Event"}, nil, nil)
+		t3.filters, err = NewEventFilter(DefaultEventFilterConfig, chain, []string{"A.0x1.NonExistent.Event"}, nil, nil)
 		require.NoError(s.T(), err)
 		tests = append(tests, t3)
 	}

engine/access/state_stream/backend_executiondata.go

@@ -23,10 +23,11 @@ type ExecutionDataResponse struct {
 }
 
 type ExecutionDataBackend struct {
-	log         zerolog.Logger
-	headers     storage.Headers
-	broadcaster *engine.Broadcaster
-	sendTimeout time.Duration
+	log            zerolog.Logger
+	headers        storage.Headers
+	broadcaster    *engine.Broadcaster
+	sendTimeout    time.Duration
+	sendBufferSize int
 
 	getExecutionData GetExecutionDataFunc
 	getStartHeight   GetStartHeightFunc
@@ -50,7 +51,7 @@ func (b *ExecutionDataBackend) GetExecutionDataByBlockID(ctx context.Context, bl
 func (b *ExecutionDataBackend) SubscribeExecutionData(ctx context.Context, startBlockID flow.Identifier, startHeight uint64) Subscription {
 	nextHeight, err := b.getStartHeight(startBlockID, startHeight)
 	if err != nil {
-		sub := NewSubscription()
+		sub := NewSubscription(b.sendBufferSize)
 		if st, ok := status.FromError(err); ok {
 			sub.Fail(status.Errorf(st.Code(), "could not get start height: %s", st.Message()))
 			return sub
@@ -60,7 +61,7 @@ func (b *ExecutionDataBackend) SubscribeExecutionData(ctx context.Context, start
 		return sub
 	}
 
-	sub := NewHeightBasedSubscription(nextHeight, b.getResponse)
+	sub := NewHeightBasedSubscription(b.sendBufferSize, nextHeight, b.getResponse)
 
 	go NewStreamer(b.log, b.broadcaster, b.sendTimeout, sub).Stream(ctx)
 

engine/access/state_stream/backend_executiondata_test.go

@@ -92,9 +92,15 @@ func (s *BackendExecutionDataSuite) SetupTest() {
 		metrics.NewNoopCollector(),
 	)
 
+	conf := Config{
+		ClientSendTimeout:    DefaultSendTimeout,
+		ClientSendBufferSize: DefaultSendBufferSize,
+	}
+
 	var err error
 	s.backend, err = New(
 		logger,
+		conf,
 		s.state,
 		s.headers,
 		s.seals,

engine/access/state_stream/engine.go

@@ -3,6 +3,7 @@ package state_stream
 import (
 	"fmt"
 	"net"
+	"time"
 
 	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
 	access "github.com/onflow/flow/protobuf/go/flow/executiondata"
@@ -25,9 +26,29 @@ import (
 
 // Config defines the configurable options for the ingress server.
 type Config struct {
-	ListenAddr              string
-	MaxExecutionDataMsgSize uint // in bytes
-	RpcMetricsEnabled       bool // enable GRPC metrics
+	EventFilterConfig
+
+	// ListenAddr is the address the GRPC server will listen on as host:port
+	ListenAddr string
+
+	// MaxExecutionDataMsgSize is the max message size for block execution data API
+	MaxExecutionDataMsgSize uint
+
+	// RpcMetricsEnabled specifies whether to enable the GRPC metrics
+	RpcMetricsEnabled bool
+
+	// MaxGlobalStreams defines the global max number of streams that can be open at the same time.
+	MaxGlobalStreams uint32
+
+	// ExecutionDataCacheSize is the max number of objects for the execution data cache.
+	ExecutionDataCacheSize uint32
+
+	// ClientSendTimeout is the timeout for sending a message to the client. After the timeout,
+	// the stream is closed with an error.
+	ClientSendTimeout time.Duration
+
+	// ClientSendBufferSize is the size of the response buffer for sending messages to the client.
+	ClientSendBufferSize uint
 }
 
 // Engine exposes the server with the state stream API.
@@ -48,15 +69,15 @@ type Engine struct {
 	stateStreamGrpcAddress net.Addr
 }
 
-// New returns a new ingress server.
+// NewEng returns a new ingress server.
 func NewEng(
+	log zerolog.Logger,
 	config Config,
 	execDataStore execution_data.ExecutionDataStore,
 	state protocol.State,
 	headers storage.Headers,
 	seals storage.Seals,
 	results storage.ExecutionResults,
-	log zerolog.Logger,
 	chainID flow.ChainID,
 	apiRatelimits map[string]int, // the api rate limit (max calls per second) for each of the gRPC API e.g. Ping->100, GetExecutionDataByBlockID->300
 	apiBurstLimits map[string]int, // the api burst limit (max calls at the same time) for each of the gRPC API e.g. Ping->50, GetExecutionDataByBlockID->10
@@ -93,7 +114,7 @@ func NewEng(
 	server := grpc.NewServer(grpcOpts...)
 
 	execDataCache := herocache.NewCache(
-		DefaultCacheSize,
+		config.ExecutionDataCacheSize,
 		herocache.DefaultOversizeFactor,
 		heropool.LRUEjection,
 		logger,
@@ -102,7 +123,7 @@ func NewEng(
 
 	broadcaster := engine.NewBroadcaster()
 
-	backend, err := New(logger, state, headers, seals, results, execDataStore, execDataCache, broadcaster)
+	backend, err := New(logger, config, state, headers, seals, results, execDataStore, execDataCache, broadcaster)
 	if err != nil {
 		return nil, fmt.Errorf("could not create state stream backend: %w", err)
 	}
@@ -113,7 +134,7 @@ func NewEng(
 		server:              server,
 		chain:               chainID.Chain(),
 		config:              config,
-		handler:             NewHandler(backend, chainID.Chain()),
+		handler:             NewHandler(backend, chainID.Chain(), config.EventFilterConfig, config.MaxGlobalStreams),
 		execDataBroadcaster: broadcaster,
 		execDataCache:       execDataCache,
 	}
@@ -127,6 +148,7 @@ func NewEng(
 	return e, nil
 }
 
+// OnExecutionData is called to notify the engine when a new execution data is received.
 func (e *Engine) OnExecutionData(executionData *execution_data.BlockExecutionDataEntity) {
 	e.log.Trace().
 		Hex("block_id", logging.ID(executionData.BlockID)).