Merge pull request #572 from streamingfast/feature/max-tier1-requests

Added max tier1 request

Author: maoueh

app/tier1.go

@@ -7,11 +7,6 @@ import (
 	"time"
 
 	"connectrpc.com/connect"
-	"github.com/streamingfast/substreams/pb/sf/substreams/rpc/v2/pbsubstreamsrpcconnect"
-	ssconnect "github.com/streamingfast/substreams/pb/sf/substreams/rpc/v2/pbsubstreamsrpcconnect"
-	"github.com/streamingfast/substreams/reqctx"
-	"github.com/streamingfast/substreams/wasm/wazero"
-
 	"github.com/streamingfast/bstream"
 	"github.com/streamingfast/bstream/blockstream"
 	"github.com/streamingfast/bstream/hub"
@@ -23,8 +18,12 @@ import (
 	"github.com/streamingfast/shutter"
 	"github.com/streamingfast/substreams/client"
 	"github.com/streamingfast/substreams/metrics"
+	"github.com/streamingfast/substreams/pb/sf/substreams/rpc/v2/pbsubstreamsrpcconnect"
+	ssconnect "github.com/streamingfast/substreams/pb/sf/substreams/rpc/v2/pbsubstreamsrpcconnect"
+	"github.com/streamingfast/substreams/reqctx"
 	"github.com/streamingfast/substreams/service"
 	"github.com/streamingfast/substreams/wasm"
+	"github.com/streamingfast/substreams/wasm/wazero"
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
 )
@@ -56,11 +55,13 @@ type Tier1Config struct {
 	BlockExecutionTimeout   time.Duration
 	TmpDir                  string
 
-	StateStoreURL        string
-	StateStoreDefaultTag string
-	BlockType            string
-	StateBundleSize      uint64
-	EnforceCompression   bool // refuse incoming requests that do not accept gzip compression (ConnectRPC or GRPC)
+	StateStoreURL           string
+	StateStoreDefaultTag    string
+	BlockType               string
+	StateBundleSize         uint64
+	EnforceCompression      bool // refuse incoming requests that do not accept gzip compression (ConnectRPC or GRPC)
+	ActiveRequestsSoftLimit int  // maximum number of active requests a tier1 app can have with external clients before starting to advertise itself as unready in the health check
+	ActiveRequestsHardLimit int  // maximum number of active requests a tier1 app can have with external clients, refuse with CodeUnavailable if reached
 
 	MaxSubrequests       uint64
 	SubrequestsEndpoint  string
@@ -199,9 +200,12 @@ func (a *Tier1App) Run() error {
 		a.config.MaxSubrequests,
 		a.config.StateBundleSize,
 		a.config.BlockType,
+		a.setIsReady,
 		subrequestsClientConfig,
 		tier2RequestParameters,
 		a.config.EnforceCompression,
+		a.config.ActiveRequestsSoftLimit,
+		a.config.ActiveRequestsHardLimit,
 		opts...,
 	)
 	if err != nil {
@@ -230,14 +234,14 @@ func (a *Tier1App) Run() error {
 			a.logger.Info("waiting until hub is real-time synced")
 			select {
 			case <-forkableHub.Ready:
-				metrics.AppReadinessTier1.SetReady()
+				// Wait until the hub is ready
 			case <-a.Terminating():
 				return
 			}
 		}
 
 		a.logger.Info("launching gRPC server", zap.Bool("live_support", withLive))
-		a.isReady.CompareAndSwap(false, true)
+		a.setIsReady(true)
 
 		err := service.ListenTier1(a.config.GRPCListenAddr, svc, infoServer, a.modules.Authenticator, a.logger, a.HealthCheck)
 		a.Shutdown(err)
@@ -267,6 +271,16 @@ func (a *Tier1App) IsReady(ctx context.Context) bool {
 	return a.isReady.Load()
 }
 
+func (a *Tier1App) setIsReady(ready bool) {
+	if ready {
+		a.isReady.Store(true)
+		metrics.AppReadinessTier1.SetReady()
+	} else {
+		a.isReady.Store(false)
+		metrics.AppReadinessTier1.SetNotReady()
+	}
+}
+
 // Validate inspects itself to determine if the current config is valid according to
 // substreams rules.
 func (config *Tier1Config) Validate() error {

app/tier2.go

@@ -80,7 +80,7 @@ func (a *Tier2App) Run() error {
 		opts = append(opts, service.WithBlockExecutionTimeout(a.config.BlockExecutionTimeout))
 	}
 
-	opts = append(opts, service.WithReadinessFunc(a.setReadiness))
+	opts = append(opts, service.WithReadinessFunc(a.setIsReady))
 
 	if a.config.TmpDir != "" {
 		wazero.SetTempDir(a.config.TmpDir)
@@ -103,12 +103,11 @@ func (a *Tier2App) Run() error {
 		return fmt.Errorf("failed to setup trust authenticator: %w", err)
 	}
 
-	a.OnTerminating(func(_ error) { metrics.AppReadinessTier2.SetNotReady() })
+	a.OnTerminating(func(_ error) { a.setIsReady(false) })
 
 	go func() {
 		a.logger.Info("launching gRPC server")
-		a.isReady.CompareAndSwap(false, true)
-		metrics.AppReadinessTier2.SetReady()
+		a.setIsReady(false)
 
 		err := service.ListenTier2(a.config.GRPCListenAddr, a.config.ServiceDiscoveryURL, svc, trustAuth, a.logger, a.HealthCheck)
 		a.Shutdown(err)
@@ -135,8 +134,14 @@ func (a *Tier2App) IsReady(ctx context.Context) bool {
 	return a.isReady.Load()
 }
 
-func (a *Tier2App) setReadiness(ready bool) {
+func (a *Tier2App) setIsReady(ready bool) {
 	a.isReady.Store(ready)
+
+	if ready {
+		metrics.AppReadinessTier2.SetReady()
+	} else {
+		metrics.AppReadinessTier2.SetNotReady()
+	}
 }
 
 // Validate inspects itself to determine if the current config is valid according to

docs/release-notes/change-log.md

@@ -17,6 +17,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 * Fix detection of accepted gzip compression when multiple values are sent in the `Grpc-Accept-Encoding` header (ex: Python library)
 * Properly accept and compress responses with `gzip` for browser HTTP clients using ConnectWeb with `Accept-Encoding` header
 * Allow setting subscription channel max capacity via `SOURCE_CHAN_SIZE` env var (default: 100)
+* Added tier1 app configuration option to limit max active requests a single instance can accept before starting to reject them with 'Unavailable' gRPC code.
 
 ### Client-side
 

metrics/metrics.go

@@ -9,7 +9,7 @@ import (
 
 var MetricSet = dmetrics.NewSet()
 
-var ActiveSubstreams = MetricSet.NewGauge("substreams_active_requests", "Number of active Substreams requests")
+var ActiveRequests = MetricSet.NewGauge("substreams_active_requests", "Number of active Substreams requests")
 var SubstreamsCounter = MetricSet.NewCounter("substreams_counter", "Substreams requests count")
 
 var BlockBeginProcess = MetricSet.NewCounter("substreams_block_process_start_counter", "Counter for total block processes started, used for rate")

service/options.go

@@ -67,7 +67,7 @@ func WithReadinessFunc(f func(bool)) Option {
 		case *Tier1Service:
 			// not used
 		case *Tier2Service:
-			s.setReadyFunc = f
+			s.appSetIsReadyState = f
 		}
 	}
 }

service/tier1.go

@@ -13,24 +13,22 @@ import (
 	"sync"
 	"time"
 
-	"github.com/streamingfast/substreams/metering"
-
+	"connectrpc.com/connect"
 	"github.com/streamingfast/bstream"
-	"github.com/streamingfast/dgrpc"
-
 	"github.com/streamingfast/bstream/hub"
 	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"
 	bsstream "github.com/streamingfast/bstream/stream"
 	"github.com/streamingfast/dauth"
+	"github.com/streamingfast/dgrpc"
 	"github.com/streamingfast/dmetering"
+	"github.com/streamingfast/dmetrics"
 	"github.com/streamingfast/dstore"
 	"github.com/streamingfast/logging"
 	tracing "github.com/streamingfast/sf-tracing"
 	"github.com/streamingfast/shutter"
-
-	"connectrpc.com/connect"
 	"github.com/streamingfast/substreams"
 	"github.com/streamingfast/substreams/client"
+	"github.com/streamingfast/substreams/metering"
 	"github.com/streamingfast/substreams/metrics"
 	"github.com/streamingfast/substreams/orchestrator/plan"
 	"github.com/streamingfast/substreams/orchestrator/work"
@@ -69,12 +67,17 @@ type Tier1Service struct {
 	tracer                ttrace.Tracer
 	logger                *zap.Logger
 
+	// You can call this function to switch the parent app to be ready or not ready influencing the health check,
+	// it's provided by [app.Tier1App] and tied to the health check endpoint.
+	appSetIsReadyState  func(isReady bool)
 	getRecentFinalBlock func() (uint64, error)
 	resolveCursor       pipeline.CursorResolver
 	getHeadBlock        func() (uint64, error)
 
-	enforceCompression     bool
-	tier2RequestParameters reqctx.Tier2RequestParameters
+	enforceCompression      bool
+	activeRequestsSoftLimit int
+	activeRequestsHardLimit int
+	tier2RequestParameters  reqctx.Tier2RequestParameters
 }
 
 func getBlockTypeFromStreamFactory(sf *StreamFactory) (string, error) {
@@ -129,9 +132,12 @@ func NewTier1(
 	stateBundleSize uint64,
 	blockType string,
 
+	appSetIsReadyState func(isReady bool),
 	substreamsClientConfig *client.SubstreamsClientConfig,
 	tier2RequestParameters reqctx.Tier2RequestParameters,
 	enforceCompression bool,
+	activeRequestsSoftLimit int,
+	activeRequestsHardLimit int,
 	opts ...Option,
 ) (*Tier1Service, error) {
 
@@ -168,16 +174,19 @@ func NewTier1(
 
 	logger.Info("launching tier1 service", zap.Reflect("client_config", substreamsClientConfig), zap.String("block_type", blockType), zap.Bool("with_live", hub != nil))
 	s := &Tier1Service{
-		Shutter:                shutter.New(),
-		runtimeConfig:          runtimeConfig,
-		blockType:              blockType,
-		tracer:                 tracing.GetTracer(),
-		failedRequests:         make(map[string]*recordedFailure),
-		resolveCursor:          pipeline.NewCursorResolver(hub, mergedBlocksStore, forkedBlocksStore),
-		logger:                 logger,
-		tier2RequestParameters: tier2RequestParameters,
-		blockExecutionTimeout:  3 * time.Minute,
-		enforceCompression:     enforceCompression,
+		Shutter:                 shutter.New(),
+		runtimeConfig:           runtimeConfig,
+		blockType:               blockType,
+		tracer:                  tracing.GetTracer(),
+		failedRequests:          make(map[string]*recordedFailure),
+		resolveCursor:           pipeline.NewCursorResolver(hub, mergedBlocksStore, forkedBlocksStore),
+		logger:                  logger,
+		appSetIsReadyState:      appSetIsReadyState,
+		tier2RequestParameters:  tier2RequestParameters,
+		blockExecutionTimeout:   3 * time.Minute,
+		enforceCompression:      enforceCompression,
+		activeRequestsSoftLimit: activeRequestsSoftLimit,
+		activeRequestsHardLimit: activeRequestsHardLimit,
 	}
 
 	s.streamFactoryFunc = sf.New
@@ -218,7 +227,7 @@ func (s *Tier1Service) Blocks(
 		compressed = true
 	}
 	if s.enforceCompression && !compressed {
-		return connect.NewError(connect.CodeInvalidArgument, fmt.Errorf("Your client does not accept gzip- or zstd-compressed streams. Check how to enable it on your GRPC or ConnectRPC client"))
+		return connect.NewError(connect.CodeInvalidArgument, fmt.Errorf("your client does not accept gzip- or zstd-compressed streams. Check how to enable it on your gRPC or ConnectRPC client"))
 	}
 
 	request := req.Msg
@@ -284,10 +293,28 @@ func (s *Tier1Service) Blocks(
 		}
 	}
 
+	status := s.getOverloadedStatus()
+
+	// Set us as unready if the soft limit would be reached by this request
+	if status.softLimitWouldBeReached() {
+		s.appSetIsReadyState(false)
+	}
+
+	// Refuse the request if the hard limit is currently reached by this instance
+	if status.hardLimitReached() {
+		return connect.NewError(connect.CodeUnavailable, fmt.Errorf("service is currently not accepting new requests, re-connect back right away to be balanced to a non-full node"))
+	}
+
 	logger.Info("incoming Substreams Blocks request", fields...)
 	metrics.SubstreamsCounter.Inc()
-	metrics.ActiveSubstreams.Inc()
-	defer metrics.ActiveSubstreams.Dec()
+	metrics.ActiveRequests.Inc()
+	defer func() {
+		metrics.ActiveRequests.Dec()
+
+		if status := s.getOverloadedStatus(); status.canAcceptUpcomingRequests() {
+			s.appSetIsReadyState(true)
+		}
+	}()
 
 	requestID := fmt.Sprintf("%s:%d:%d:%s:%t:%t:%s",
 		outputModuleHash,
@@ -737,3 +764,56 @@ func matchHeader(header http.Header) bool {
 	}
 	return false
 }
+
+type overloadingStatus struct {
+	// set only if either soft or hard limit set is > 0
+	activeRequestCount int
+	softLimit          int
+	hardLimit          int
+}
+
+// softLimitWouldBeReached returns true if the soft limit would be reached if one more request was added.
+func (s *overloadingStatus) softLimitWouldBeReached() bool {
+	return s.softLimit > 0 && s.activeRequestCount+1 >= s.softLimit
+}
+
+// hardLimitReached returns true if the hard limit is actually reached from the active request count.
+func (s *overloadingStatus) hardLimitReached() bool {
+	return s.hardLimit > 0 && s.activeRequestCount >= s.hardLimit
+}
+
+// canAcceptUpcomingRequests returns true if the service can accept upcoming new requests.
+func (s *overloadingStatus) canAcceptUpcomingRequests() bool {
+	if s.softLimit <= 0 && s.hardLimit <= 0 {
+		return true
+	}
+
+	if s.softLimit > 0 && s.activeRequestCount >= s.softLimit {
+		return false
+	}
+
+	if s.hardLimit > 0 && s.activeRequestCount >= s.hardLimit {
+		return false
+	}
+
+	return true
+}
+
+func (s *Tier1Service) getOverloadedStatus() (status overloadingStatus) {
+	// Never overloaded if both soft & hard limit are 0, -1 or anything less
+	if s.activeRequestsSoftLimit <= 0 && s.activeRequestsHardLimit <= 0 {
+		return
+	}
+
+	activeRequestCount := s.getActiveRequestCount()
+
+	return overloadingStatus{
+		activeRequestCount: activeRequestCount,
+		softLimit:          s.activeRequestsSoftLimit,
+		hardLimit:          s.activeRequestsHardLimit,
+	}
+}
+
+func (s *Tier1Service) getActiveRequestCount() int {
+	return int(dmetrics.NewValueFromMetric(metrics.ActiveRequests, "requests").ValueUint())
+}