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, 0)
 	listener := rpcprovider.NewProviderListener(ctx, rpcProviderEndpoint.NetworkAddress, "/health")
 	err = listener.RegisterReceiver(rpcProviderServer, rpcProviderEndpoint)
 	require.NoError(t, err)

protocol/rpcconsumer/rpcconsumer_server.go

@@ -1260,6 +1260,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
@@ -1332,14 +1343,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,49 @@
+package rpcprovider
+
+import (
+	"strconv"
+	"sync/atomic"
+)
+
+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)
+	loadManager.activeRequestsPerSecond.Store(0)
+
+	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() string {
+	if loadManager == nil {
+		return ""
+	}
+	loadedRateLimitThreshold := loadManager.rateLimitThreshold.Load()
+	if loadedRateLimitThreshold == 0 {
+		return "0"
+	}
+
+	return strconv.FormatUint(loadManager.activeRequestsPerSecond.Load()/loadedRateLimitThreshold, 10)
+}

protocol/rpcprovider/rpcprovider.go

@@ -110,6 +110,7 @@ type rpcProviderStartOptions struct {
 	healthCheckMetricsOptions *rpcProviderHealthCheckMetricsOptions
 	staticProvider            bool
 	staticSpecPath            string
+	relayLoadLimit            uint64
 }
 
 type rpcProviderHealthCheckMetricsOptions struct {
@@ -141,6 +142,7 @@ type RPCProvider struct {
 	providerUniqueId          string
 	staticProvider            bool
 	staticSpecPath            string
+	relayLoadLimit            uint64
 }
 
 func (rpcp *RPCProvider) Start(options *rpcProviderStartOptions) (err error) {
@@ -165,6 +167,7 @@ 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
 
 	// single state tracker
 	lavaChainFetcher := chainlib.NewLavaChainFetcher(ctx, options.clientCtx)
@@ -486,7 +489,7 @@ func (rpcp *RPCProvider) SetupEndpoint(ctx context.Context, 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, rpcp.relayLoadLimit)
 	// set up grpc listener
 	var listener *ProviderListener
 	func() {
@@ -717,6 +720,10 @@ rpcprovider 127.0.0.1:3333 OSMOSIS tendermintrpc "wss://www.node-path.com:80,htt
 			if stickinessHeaderName != "" {
 				RPCProviderStickinessHeaderName = stickinessHeaderName
 			}
+			relayLoadLimit, err := cmd.Flags().GetUint64(common.RateLimitRequestPerSecondFlag)
+			if err != nil {
+				utils.LavaFormatFatal("failed to read relay concurrent load limit flag", err)
+			}
 			prometheusListenAddr := viper.GetString(metrics.MetricsListenFlagName)
 			rewardStoragePath := viper.GetString(rewardserver.RewardServerStorageFlagName)
 			rewardTTL := viper.GetDuration(rewardserver.RewardTTLFlagName)
@@ -754,6 +761,7 @@ rpcprovider 127.0.0.1:3333 OSMOSIS tendermintrpc "wss://www.node-path.com:80,htt
 				&rpcProviderHealthCheckMetricsOptions,
 				staticProvider,
 				offlineSpecPath,
+				relayLoadLimit,
 			}
 
 			rpcProvider := RPCProvider{}
@@ -790,7 +798,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().Uint(common.RateLimitRequestPerSecondFlag, 0, "rate limit requests per second - per chain - default unlimited")
 	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,
+	relayLoadLimit uint64,
 ) {
 	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 = NewProviderLoadManager(relayLoadLimit)
 
 	rpcps.initRelaysMonitor(ctx)
 }
@@ -180,7 +183,12 @@ 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
+	rpcps.providerLoadManager.addRelayCall()
+	defer func() { go rpcps.providerLoadManager.subtractRelayCall() }()
+	provideRelayLoad := rpcps.providerLoadManager.getProviderLoad()
+	trailerMd := metadata.Pairs(chainlib.RpcProviderUniqueIdHeader, rpcps.providerUniqueId, chainlib.RpcProviderLoadRateHeader, provideRelayLoad)
+	grpc.SetTrailer(ctx, trailerMd)
 	if request.RelayData == nil || request.RelaySession == nil {
 		return nil, utils.LavaFormatWarning("invalid relay request, internal fields are nil", nil)
 	}