feat: PRT provider load rate

protocol/chainlib/common.go

@@ -30,6 +30,7 @@ const (
 	relayMsgLogMaxChars        = 200
 	RPCProviderNodeAddressHash = "Lava-Provider-Node-Address-Hash"
 	RPCProviderNodeExtension   = "Lava-Provider-Node-Extension"
+	RpcProviderLoadRateHeader  = "Lava-Provider-load-rate"
 	RpcProviderUniqueIdHeader  = "Lava-Provider-Unique-Id"
 	WebSocketExtension         = "websocket"
 )

protocol/common/cobra_common.go

@@ -42,6 +42,7 @@ const (
 	// websocket flags
 	RateLimitWebSocketFlag                       = "rate-limit-websocket-requests-per-connection"
 	BanDurationForWebsocketRateLimitExceededFlag = "ban-duration-for-websocket-rate-limit-exceeded"
+	RateLimitRequestPerSecondFlag                = "rate-limit-requests-per-second"
 )
 
 const (

protocol/integration/protocol_test.go

@@ -344,7 +344,7 @@ func createRpcProvider(t *testing.T, ctx context.Context, rpcProviderOptions rpc
 	chainTracker.StartAndServe(ctx)
 	reliabilityManager := reliabilitymanager.NewReliabilityManager(chainTracker, &mockProviderStateTracker, rpcProviderOptions.account.Addr.String(), chainRouter, chainParser)
 	mockReliabilityManager := NewMockReliabilityManager(reliabilityManager)
-	rpcProviderServer.ServeRPCRequests(ctx, rpcProviderEndpoint, chainParser, rws, providerSessionManager, mockReliabilityManager, rpcProviderOptions.account.SK, cache, chainRouter, &mockProviderStateTracker, rpcProviderOptions.account.Addr, rpcProviderOptions.lavaChainID, rpcprovider.DEFAULT_ALLOWED_MISSING_CU, nil, nil, nil, false)
+	rpcProviderServer.ServeRPCRequests(ctx, rpcProviderEndpoint, chainParser, rws, providerSessionManager, mockReliabilityManager, rpcProviderOptions.account.SK, cache, chainRouter, &mockProviderStateTracker, rpcProviderOptions.account.Addr, rpcProviderOptions.lavaChainID, rpcprovider.DEFAULT_ALLOWED_MISSING_CU, nil, nil, nil, false, nil)
 	listener := rpcprovider.NewProviderListener(ctx, rpcProviderEndpoint.NetworkAddress, "/health")
 	err = listener.RegisterReceiver(rpcProviderServer, rpcProviderEndpoint)
 	require.NoError(t, err)

protocol/metrics/metrics_provider.go

@@ -22,6 +22,7 @@ type ProviderMetrics struct {
 	totalRelaysServicedMetric *prometheus.CounterVec
 	totalErroredMetric        *prometheus.CounterVec
 	consumerQoSMetric         *prometheus.GaugeVec
+	loadRateMetric            *prometheus.GaugeVec
 }
 
 func (pm *ProviderMetrics) AddRelay(consumerAddress string, cu uint64, qos *pairingtypes.QualityOfServiceReport) {
@@ -49,6 +50,10 @@ func (pm *ProviderMetrics) AddRelay(consumerAddress string, cu uint64, qos *pair
 	}
 }
 
+func (pm *ProviderMetrics) SetLoadRate(loatRate float64) {
+	pm.loadRateMetric.WithLabelValues(pm.specID).Set(loatRate)
+}
+
 func (pm *ProviderMetrics) AddPayment(cu uint64) {
 	if pm == nil {
 		return
@@ -72,6 +77,7 @@ func NewProviderMetrics(specID, apiInterface string, totalCUServicedMetric *prom
 	totalRelaysServicedMetric *prometheus.CounterVec,
 	totalErroredMetric *prometheus.CounterVec,
 	consumerQoSMetric *prometheus.GaugeVec,
+	loadRateMetric *prometheus.GaugeVec,
 ) *ProviderMetrics {
 	pm := &ProviderMetrics{
 		specID:                    specID,
@@ -82,6 +88,7 @@ func NewProviderMetrics(specID, apiInterface string, totalCUServicedMetric *prom
 		totalRelaysServicedMetric: totalRelaysServicedMetric,
 		totalErroredMetric:        totalErroredMetric,
 		consumerQoSMetric:         consumerQoSMetric,
+		loadRateMetric:            loadRateMetric,
 	}
 	return pm
 }

protocol/metrics/metrics_provider_manager.go

@@ -41,6 +41,7 @@ type ProviderMetricsManager struct {
 	endpointsHealthChecksOk       uint64
 	relaysMonitors                map[string]*RelaysMonitor
 	relaysMonitorsLock            sync.RWMutex
+	loadRateMetric                *prometheus.GaugeVec
 }
 
 func NewProviderMetricsManager(networkAddress string) *ProviderMetricsManager {
@@ -107,6 +108,11 @@ func NewProviderMetricsManager(networkAddress string) *ProviderMetricsManager {
 		Help: "The total number of get latest block queries that succeeded by chainfetcher",
 	}, []string{"spec"})
 
+	loadRateMetric := prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "lava_provider_load_rate",
+		Help: "The load rate according to the load rate limit - Given Y simultaneous relay calls, a value of X  and will measure Y/X load rate.",
+	}, []string{"spec"})
+
 	fetchBlockSuccessMetric := prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: "lava_provider_fetch_block_success",
 		Help: "The total number of get specific block queries that succeeded by chainfetcher",
@@ -141,6 +147,7 @@ func NewProviderMetricsManager(networkAddress string) *ProviderMetricsManager {
 	prometheus.MustRegister(virtualEpochMetric)
 	prometheus.MustRegister(endpointsHealthChecksOkMetric)
 	prometheus.MustRegister(protocolVersionMetric)
+	prometheus.MustRegister(loadRateMetric)
 
 	providerMetricsManager := &ProviderMetricsManager{
 		providerMetrics:               map[string]*ProviderMetrics{},
@@ -161,6 +168,7 @@ func NewProviderMetricsManager(networkAddress string) *ProviderMetricsManager {
 		endpointsHealthChecksOk:       1,
 		protocolVersionMetric:         protocolVersionMetric,
 		relaysMonitors:                map[string]*RelaysMonitor{},
+		loadRateMetric:                loadRateMetric,
 	}
 
 	http.Handle("/metrics", promhttp.Handler())
@@ -209,7 +217,7 @@ func (pme *ProviderMetricsManager) AddProviderMetrics(specID, apiInterface strin
 	}
 
 	if pme.getProviderMetric(specID, apiInterface) == nil {
-		providerMetric := NewProviderMetrics(specID, apiInterface, pme.totalCUServicedMetric, pme.totalCUPaidMetric, pme.totalRelaysServicedMetric, pme.totalErroredMetric, pme.consumerQoSMetric)
+		providerMetric := NewProviderMetrics(specID, apiInterface, pme.totalCUServicedMetric, pme.totalCUPaidMetric, pme.totalRelaysServicedMetric, pme.totalErroredMetric, pme.consumerQoSMetric, pme.loadRateMetric)
 		pme.setProviderMetric(providerMetric)
 
 		endpoint := fmt.Sprintf("/metrics/%s/%s/health", specID, apiInterface)

protocol/rpcconsumer/rpcconsumer_server.go

@@ -1261,6 +1261,17 @@ func (rpccs *RPCConsumerServer) HandleDirectiveHeadersForMessage(chainMessage ch
 	chainMessage.SetForceCacheRefresh(ok)
 }
 
+func (rpccs *RPCConsumerServer) getMetadataFromRelayTrailer(metadataHeader string, relayResult *common.RelayResult) {
+	trailerValue := relayResult.ProviderTrailer.Get(metadataHeader)
+	if len(trailerValue) > 0 {
+		extensionMD := pairingtypes.Metadata{
+			Name:  metadataHeader,
+			Value: trailerValue[0],
+		}
+		relayResult.Reply.Metadata = append(relayResult.Reply.Metadata, extensionMD)
+	}
+}
+
 func (rpccs *RPCConsumerServer) appendHeadersToRelayResult(ctx context.Context, relayResult *common.RelayResult, protocolErrors uint64, relayProcessor *RelayProcessor, protocolMessage chainlib.ProtocolMessage, apiName string) {
 	if relayResult == nil {
 		return
@@ -1333,14 +1344,8 @@ func (rpccs *RPCConsumerServer) appendHeadersToRelayResult(ctx context.Context,
 	}
 
 	// fetch trailer information from the provider by using the provider trailer field.
-	providerNodeExtensions := relayResult.ProviderTrailer.Get(chainlib.RPCProviderNodeExtension)
-	if len(providerNodeExtensions) > 0 {
-		extensionMD := pairingtypes.Metadata{
-			Name:  chainlib.RPCProviderNodeExtension,
-			Value: providerNodeExtensions[0],
-		}
-		relayResult.Reply.Metadata = append(relayResult.Reply.Metadata, extensionMD)
-	}
+	rpccs.getMetadataFromRelayTrailer(chainlib.RPCProviderNodeExtension, relayResult)
+	rpccs.getMetadataFromRelayTrailer(chainlib.RpcProviderLoadRateHeader, relayResult)
 
 	directiveHeaders := protocolMessage.GetDirectiveHeaders()
 	_, debugRelays := directiveHeaders[common.LAVA_DEBUG_RELAY]

protocol/rpcprovider/provider_load_manager.go

@@ -0,0 +1,74 @@
+package rpcprovider
+
+import (
+	"context"
+	"strconv"
+	"sync/atomic"
+
+	"github.com/lavanet/lava/v3/protocol/chainlib"
+	grpc "google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
+)
+
+type ProviderLoadManager struct {
+	rateLimitThreshold      atomic.Uint64
+	activeRequestsPerSecond atomic.Uint64
+}
+
+func NewProviderLoadManager(rateLimitThreshold uint64) *ProviderLoadManager {
+	if rateLimitThreshold == 0 {
+		return nil
+	}
+	loadManager := &ProviderLoadManager{}
+
+	loadManager.rateLimitThreshold.Store(rateLimitThreshold)
+
+	return loadManager
+}
+
+func (loadManager *ProviderLoadManager) addRelayCall() {
+	if loadManager == nil {
+		return
+	}
+	loadManager.activeRequestsPerSecond.Add(1)
+}
+
+func (loadManager *ProviderLoadManager) subtractRelayCall() {
+	if loadManager == nil {
+		return
+	}
+	loadManager.activeRequestsPerSecond.Add(^uint64(0))
+}
+
+func (loadManager *ProviderLoadManager) getProviderLoad() float64 {
+	if loadManager == nil {
+		return 0
+	}
+	rateLimitThreshold := loadManager.rateLimitThreshold.Load()
+	if rateLimitThreshold == 0 {
+		return 0
+	}
+	activeRequests := loadManager.activeRequestsPerSecond.Load()
+	return float64(activeRequests) / float64(rateLimitThreshold)
+}
+
+func (loadManager *ProviderLoadManager) applyProviderLoadMetadataToContextTrailer(ctx context.Context) bool {
+	provideRelayLoad := loadManager.getProviderLoad()
+	if provideRelayLoad == 0 {
+		return false
+	}
+	formattedProviderLoad := strconv.FormatFloat(provideRelayLoad, 'f', -1, 64)
+
+	trailerMd := metadata.Pairs(chainlib.RpcProviderLoadRateHeader, formattedProviderLoad)
+	grpc.SetTrailer(ctx, trailerMd)
+	return true
+}
+
+func (loadManager *ProviderLoadManager) addAndSetRelayLoadToContextTrailer(ctx context.Context) bool {
+	loadManager.addRelayCall()
+	return loadManager.applyProviderLoadMetadataToContextTrailer(ctx)
+}
+
+func (loadManager *ProviderLoadManager) getActiveRequestsPerSecond() uint64 {
+	return loadManager.activeRequestsPerSecond.Load()
+}

protocol/rpcprovider/rpcprovider.go

@@ -110,6 +110,7 @@ type rpcProviderStartOptions struct {
 	healthCheckMetricsOptions *rpcProviderHealthCheckMetricsOptions
 	staticProvider            bool
 	staticSpecPath            string
+	relayLoadLimit            uint64
 }
 
 type rpcProviderHealthCheckMetricsOptions struct {
@@ -123,24 +124,26 @@ type RPCProvider struct {
 	rpcProviderListeners map[string]*ProviderListener
 	lock                 sync.Mutex
 	// all of the following members need to be concurrency proof
-	providerMetricsManager    *metrics.ProviderMetricsManager
-	rewardServer              *rewardserver.RewardServer
-	privKey                   *btcec.PrivateKey
-	lavaChainID               string
-	addr                      sdk.AccAddress
-	blockMemorySize           uint64
-	chainMutexes              map[string]*sync.Mutex
-	parallelConnections       uint
-	cache                     *performance.Cache
-	shardID                   uint // shardID is a flag that allows setting up multiple provider databases of the same chain
-	chainTrackers             *common.SafeSyncMap[string, *chaintracker.ChainTracker]
-	relaysMonitorAggregator   *metrics.RelaysMonitorAggregator
-	relaysHealthCheckEnabled  bool
-	relaysHealthCheckInterval time.Duration
-	grpcHealthCheckEndpoint   string
-	providerUniqueId          string
-	staticProvider            bool
-	staticSpecPath            string
+	providerMetricsManager       *metrics.ProviderMetricsManager
+	rewardServer                 *rewardserver.RewardServer
+	privKey                      *btcec.PrivateKey
+	lavaChainID                  string
+	addr                         sdk.AccAddress
+	blockMemorySize              uint64
+	chainMutexes                 map[string]*sync.Mutex
+	parallelConnections          uint
+	cache                        *performance.Cache
+	shardID                      uint // shardID is a flag that allows setting up multiple provider databases of the same chain
+	chainTrackers                *common.SafeSyncMap[string, *chaintracker.ChainTracker]
+	relaysMonitorAggregator      *metrics.RelaysMonitorAggregator
+	relaysHealthCheckEnabled     bool
+	relaysHealthCheckInterval    time.Duration
+	grpcHealthCheckEndpoint      string
+	providerUniqueId             string
+	staticProvider               bool
+	staticSpecPath               string
+	relayLoadLimit               uint64
+	providerLoadManagersPerChain map[string]*ProviderLoadManager
 }
 
 func (rpcp *RPCProvider) Start(options *rpcProviderStartOptions) (err error) {
@@ -165,6 +168,8 @@ func (rpcp *RPCProvider) Start(options *rpcProviderStartOptions) (err error) {
 	rpcp.grpcHealthCheckEndpoint = options.healthCheckMetricsOptions.grpcHealthCheckEndpoint
 	rpcp.staticProvider = options.staticProvider
 	rpcp.staticSpecPath = options.staticSpecPath
+	rpcp.relayLoadLimit = options.relayLoadLimit
+	rpcp.providerLoadManagersPerChain = make(map[string]*ProviderLoadManager)
 
 	// single state tracker
 	lavaChainFetcher := chainlib.NewLavaChainFetcher(ctx, options.clientCtx)
@@ -307,9 +312,7 @@ func (rpcp *RPCProvider) SetupProviderEndpoints(rpcProviderEndpoints []*lavasess
 	wg.Add(parallelJobs)
 	disabledEndpoints := make(chan *lavasession.RPCProviderEndpoint, parallelJobs)
 	// validate static spec configuration is used only on a single chain setup.
-	chainIds := make(map[string]struct{})
 	for _, rpcProviderEndpoint := range rpcProviderEndpoints {
-		chainIds[rpcProviderEndpoint.ChainID] = struct{}{}
 		setupEndpoint := func(rpcProviderEndpoint *lavasession.RPCProviderEndpoint, specValidator *SpecValidator) {
 			defer wg.Done()
 			err := rpcp.SetupEndpoint(context.Background(), rpcProviderEndpoint, specValidator)
@@ -404,6 +407,7 @@ func (rpcp *RPCProvider) SetupEndpoint(ctx context.Context, rpcProviderEndpoint
 			utils.Attribute{Key: "Chain", Value: rpcProviderEndpoint.ChainID},
 			utils.Attribute{Key: "apiInterface", Value: apiInterface})
 	}
+	var providerLoadManager *ProviderLoadManager
 
 	// in order to utilize shared resources between chains we need go routines with the same chain to wait for one another here
 	chainCommonSetup := func() error {
@@ -450,6 +454,14 @@ func (rpcp *RPCProvider) SetupEndpoint(ctx context.Context, rpcProviderEndpoint
 			chainTracker = chainTrackerLoaded
 			utils.LavaFormatDebug("reusing chain tracker", utils.Attribute{Key: "chain", Value: rpcProviderEndpoint.ChainID})
 		}
+
+		// create provider load manager per chain ID
+		var keyExists bool
+		providerLoadManager, keyExists = rpcp.providerLoadManagersPerChain[rpcProviderEndpoint.ChainID]
+		if !keyExists {
+			providerLoadManager = NewProviderLoadManager(rpcp.relayLoadLimit)
+			rpcp.providerLoadManagersPerChain[rpcProviderEndpoint.ChainID] = providerLoadManager
+		}
 		return nil
 	}
 	err = chainCommonSetup()
@@ -485,8 +497,7 @@ func (rpcp *RPCProvider) SetupEndpoint(ctx context.Context, rpcProviderEndpoint
 		utils.LavaFormatTrace("Creating provider node subscription manager", utils.LogAttr("rpcProviderEndpoint", rpcProviderEndpoint))
 		providerNodeSubscriptionManager = chainlib.NewProviderNodeSubscriptionManager(chainRouter, chainParser, rpcProviderServer, rpcp.privKey)
 	}
-
-	rpcProviderServer.ServeRPCRequests(ctx, rpcProviderEndpoint, chainParser, rpcp.rewardServer, providerSessionManager, reliabilityManager, rpcp.privKey, rpcp.cache, chainRouter, rpcp.providerStateTracker, rpcp.addr, rpcp.lavaChainID, DEFAULT_ALLOWED_MISSING_CU, providerMetrics, relaysMonitor, providerNodeSubscriptionManager, rpcp.staticProvider)
+	rpcProviderServer.ServeRPCRequests(ctx, rpcProviderEndpoint, chainParser, rpcp.rewardServer, providerSessionManager, reliabilityManager, rpcp.privKey, rpcp.cache, chainRouter, rpcp.providerStateTracker, rpcp.addr, rpcp.lavaChainID, DEFAULT_ALLOWED_MISSING_CU, providerMetrics, relaysMonitor, providerNodeSubscriptionManager, rpcp.staticProvider, providerLoadManager)
 	// set up grpc listener
 	var listener *ProviderListener
 	func() {
@@ -717,6 +728,7 @@ rpcprovider 127.0.0.1:3333 OSMOSIS tendermintrpc "wss://www.node-path.com:80,htt
 			if stickinessHeaderName != "" {
 				RPCProviderStickinessHeaderName = stickinessHeaderName
 			}
+			relayLoadLimit := viper.GetUint64(common.RateLimitRequestPerSecondFlag)
 			prometheusListenAddr := viper.GetString(metrics.MetricsListenFlagName)
 			rewardStoragePath := viper.GetString(rewardserver.RewardServerStorageFlagName)
 			rewardTTL := viper.GetDuration(rewardserver.RewardTTLFlagName)
@@ -754,6 +766,7 @@ rpcprovider 127.0.0.1:3333 OSMOSIS tendermintrpc "wss://www.node-path.com:80,htt
 				&rpcProviderHealthCheckMetricsOptions,
 				staticProvider,
 				offlineSpecPath,
+				relayLoadLimit,
 			}
 
 			rpcProvider := RPCProvider{}
@@ -790,7 +803,7 @@ rpcprovider 127.0.0.1:3333 OSMOSIS tendermintrpc "wss://www.node-path.com:80,htt
 	cmdRPCProvider.Flags().BoolVar(&chainlib.IgnoreSubscriptionNotConfiguredError, chainlib.IgnoreSubscriptionNotConfiguredErrorFlag, chainlib.IgnoreSubscriptionNotConfiguredError, "ignore webSocket node url not configured error, when subscription is enabled in spec")
 	cmdRPCProvider.Flags().IntVar(&numberOfRetriesAllowedOnNodeErrors, common.SetRelayCountOnNodeErrorFlag, 2, "set the number of retries attempt on node errors")
 	cmdRPCProvider.Flags().String(common.UseStaticSpecFlag, "", "load offline spec provided path to spec file, used to test specs before they are proposed on chain, example for spec with inheritance: --use-static-spec ./cookbook/specs/ibc.json,./cookbook/specs/tendermint.json,./cookbook/specs/cosmossdk.json,./cookbook/specs/ethermint.json,./cookbook/specs/ethereum.json,./cookbook/specs/evmos.json")
-
+	cmdRPCProvider.Flags().Uint64(common.RateLimitRequestPerSecondFlag, 0, "Measuring the load relative to this number for feedback - per second - per chain - default unlimited. Given Y simultaneous relay calls, a value of X  and will measure Y/X load rate.")
 	common.AddRollingLogConfig(cmdRPCProvider)
 	return cmdRPCProvider
 }

protocol/rpcprovider/rpcprovider_server.go

@@ -71,6 +71,7 @@ type RPCProviderServer struct {
 	providerUniqueId                string
 	StaticProvider                  bool
 	providerStateMachine            *ProviderStateMachine
+	providerLoadManager             *ProviderLoadManager
 }
 
 type ReliabilityManagerInf interface {
@@ -112,6 +113,7 @@ func (rpcps *RPCProviderServer) ServeRPCRequests(
 	relaysMonitor *metrics.RelaysMonitor,
 	providerNodeSubscriptionManager *chainlib.ProviderNodeSubscriptionManager,
 	staticProvider bool,
+	providerLoadManager *ProviderLoadManager,
 ) {
 	rpcps.cache = cache
 	rpcps.chainRouter = chainRouter
@@ -134,6 +136,7 @@ func (rpcps *RPCProviderServer) ServeRPCRequests(
 	rpcps.relaysMonitor = relaysMonitor
 	rpcps.providerNodeSubscriptionManager = providerNodeSubscriptionManager
 	rpcps.providerStateMachine = NewProviderStateMachine(rpcProviderEndpoint.ChainID, lavaprotocol.NewRelayRetriesManager(), chainRouter)
+	rpcps.providerLoadManager = providerLoadManager
 
 	rpcps.initRelaysMonitor(ctx)
 }
@@ -156,6 +159,11 @@ func (rpcps *RPCProviderServer) initRelaysMonitor(ctx context.Context) {
 	rpcps.relaysMonitor.Start(ctx)
 }
 
+func (rpcps *RPCProviderServer) setLoadMetric() {
+	loadRate := rpcps.providerLoadManager.getProviderLoad()
+	rpcps.metrics.SetLoadRate(loadRate)
+}
+
 func (rpcps *RPCProviderServer) craftChainMessage() (chainMessage chainlib.ChainMessage, err error) {
 	parsing, apiCollection, ok := rpcps.chainParser.GetParsingByTag(spectypes.FUNCTION_TAG_GET_BLOCKNUM)
 	if !ok {
@@ -180,7 +188,14 @@ func (rpcps *RPCProviderServer) craftChainMessage() (chainMessage chainlib.Chain
 
 // function used to handle relay requests from a consumer, it is called by a provider_listener by calling RegisterReceiver
 func (rpcps *RPCProviderServer) Relay(ctx context.Context, request *pairingtypes.RelayRequest) (*pairingtypes.RelayReply, error) {
-	grpc.SetTrailer(ctx, metadata.Pairs(chainlib.RpcProviderUniqueIdHeader, rpcps.providerUniqueId))
+	// count the number of simultaneous relay calls
+	isLoadRateSet := rpcps.providerLoadManager.addAndSetRelayLoadToContextTrailer(ctx)
+	defer func() { go rpcps.providerLoadManager.subtractRelayCall() }()
+	if isLoadRateSet {
+		go rpcps.setLoadMetric()
+	}
+	trailerMd := metadata.Pairs(chainlib.RpcProviderUniqueIdHeader, rpcps.providerUniqueId)
+	grpc.SetTrailer(ctx, trailerMd)
 	if request.RelayData == nil || request.RelaySession == nil {
 		return nil, utils.LavaFormatWarning("invalid relay request, internal fields are nil", nil)
 	}