feat: PRT - add lava consumer blocked providers metrics

config/provider_examples/provider_example_cookbook/eth_provider_with_archive_debug_trace.yml

@@ -2,6 +2,8 @@
 # - archive
 # - debug
 # - trace
+# - trace + archive
+# - debug + archive
 
 # ** Replace the urls with your own node urls **
 
@@ -29,4 +31,20 @@ endpoints:
         - url: ws://my-eth-node.com/eth/archive/ws
           addons:
             - trace
+        - url: http://my-eth-node.com/eth/http/archive
+          addons:
+            - debug
+            - archive
+        - url: ws://my-eth-node.com/eth/archive/ws
+          addons:
+            - debug
+            - archive
+        - url: http://my-eth-node.com/eth/http/archive
+          addons:
+            - trace
+            - archive
+        - url: ws://my-eth-node.com/eth/archive/ws
+          addons:
+            - trace
+            - archive
 

go.mod

@@ -1,6 +1,6 @@
 module github.com/lavanet/lava/v4
 
-go 1.23
+go 1.23.3
 
 require (
 	github.com/99designs/keyring v1.2.1 // indirect

proto/lavanet/lava/pairing/genesis.proto

@@ -6,6 +6,7 @@ import "lavanet/lava/pairing/params.proto";
 import "lavanet/lava/pairing/epoch_cu.proto";
 import "lavanet/lava/fixationstore/fixation.proto";
 import "lavanet/lava/timerstore/timer.proto";
+import "lavanet/lava/pairing/reputation.proto";
 
 // this line is used by starport scaffolding # genesis/proto/import
 
@@ -18,17 +19,16 @@ message BadgeUsedCu {
 
 // GenesisState defines the pairing module's genesis state.
 message GenesisState {
+  reserved 2,3,4,7;
   Params params = 1 [(gogoproto.nullable) = false];
-  reserved 2;
-  reserved 3;
-  reserved 4;
   repeated BadgeUsedCu badgeUsedCuList = 5 [(gogoproto.nullable) = false];
   lavanet.lava.timerstore.GenesisState badgesTS = 6 [(gogoproto.nullable) = false];
-  lavanet.lava.fixationstore.GenesisState providerQosFS = 7 [(gogoproto.nullable) = false];
   repeated UniqueEpochSessionGenesis unique_epoch_sessions = 8 [(gogoproto.nullable) = false];
   repeated ProviderEpochCuGenesis provider_epoch_cus = 9 [(gogoproto.nullable) = false];
   repeated ProviderEpochComplainerCuGenesis provider_epoch_complained_cus = 10 [(gogoproto.nullable) = false];
   repeated ProviderConsumerEpochCuGenesis provider_consumer_epoch_cus = 11 [(gogoproto.nullable) = false];
+  repeated ReputationGenesis reputations = 12 [(gogoproto.nullable) = false];
+  lavanet.lava.fixationstore.GenesisState reputation_scores = 13 [(gogoproto.nullable) = false];
   // this line is used by starport scaffolding # genesis/proto/state
 }
 
@@ -60,4 +60,11 @@ message ProviderConsumerEpochCuGenesis {
   string project = 3;
   string chain_id = 4;
   ProviderConsumerEpochCu provider_consumer_epoch_cu = 5 [(gogoproto.nullable) = false];
+}
+
+message ReputationGenesis {
+  string chain_id = 1;
+  string cluster = 2;
+  string provider = 3;
+  Reputation reputation = 4 [(gogoproto.nullable) = false];
 }

proto/lavanet/lava/pairing/params.proto

@@ -23,6 +23,6 @@ message Params {
     (gogoproto.customtype) = "github.com/cosmos/cosmos-sdk/types.Dec",
     (gogoproto.nullable)   = false
   ];
-  int64 reputation_half_life_factor = 17 [(gogoproto.moretags) = "yaml:\"reputation_half_life_factor\""];
+  uint64 reputation_half_life_factor = 17 [(gogoproto.moretags) = "yaml:\"reputation_half_life_factor\""];
   uint64 reputation_relay_failure_cost = 18 [(gogoproto.moretags) = "yaml:\"reputation_relay_failure_cost\""];
 }

proto/lavanet/lava/pairing/query.proto

@@ -5,6 +5,7 @@ import "gogoproto/gogo.proto";
 import "google/api/annotations.proto";
 import "cosmos/base/query/v1beta1/pagination.proto";
 import "lavanet/lava/pairing/params.proto";
+import "lavanet/lava/pairing/reputation.proto";
 import "lavanet/lava/spec/spec.proto";
 
 
@@ -80,9 +81,19 @@ service Query {
 	}
 
 // Queries a for the aggregated CU of all ProviderEpochCu objects all the providers.
-rpc ProvidersEpochCu(QueryProvidersEpochCuRequest) returns (QueryProvidersEpochCuResponse) {
-	option (google.api.http).get = "/lavanet/lava/pairing/providers_epoch_cu";
-}
+	rpc ProvidersEpochCu(QueryProvidersEpochCuRequest) returns (QueryProvidersEpochCuResponse) {
+		option (google.api.http).get = "/lavanet/lava/pairing/providers_epoch_cu";
+	}
+
+// Queries a for a provider reputation.
+	rpc ProviderReputation(QueryProviderReputationRequest) returns (QueryProviderReputationResponse) {
+		option (google.api.http).get = "/lavanet/lava/pairing/provider_reputation/{provider}/{chainID}/{cluster}";
+	}
+
+// Queries a for a provider reputation's details (mainly for developers).
+	rpc ProviderReputationDetails(QueryProviderReputationDetailsRequest) returns (QueryProviderReputationDetailsResponse) {
+		option (google.api.http).get = "/lavanet/lava/pairing/provider_reputation_details/{address}/{chainID}/{cluster}";
+	}
 
 	// this line is used by starport scaffolding # 2
 
@@ -238,4 +249,39 @@ message QueryProvidersEpochCuResponse {
 message ProviderCuInfo {
 	string provider = 1;
 	uint64 cu = 2;
+}
+
+message QueryProviderReputationRequest {
+	string provider = 1;
+	string chainID = 2;
+	string cluster = 3;
+}
+
+message ReputationData {
+	uint64 rank = 1; // rank compared to other providers
+	uint64 providers = 2; // amount of providers with the same chainID+cluster
+	string overall_performance = 3; // overall performance metric which can be "good", "bad", or "low variance"
+	string chainID = 4;
+	string cluster = 5;
+}
+
+message QueryProviderReputationResponse {
+	repeated ReputationData data = 1 [(gogoproto.nullable) = false];
+}
+
+message QueryProviderReputationDetailsRequest {
+	string address = 1;
+	string chainID = 2;
+	string cluster = 3;
+}
+
+message ReputationDevData {
+	Reputation reputation = 1 [(gogoproto.nullable) = false];
+	ReputationPairingScore reputation_pairing_score = 2 [(gogoproto.nullable) = false];
+	string chainID = 4;
+	string cluster = 5;
+}
+
+message QueryProviderReputationDetailsResponse {
+	repeated ReputationDevData data = 1 [(gogoproto.nullable) = false];
 }

proto/lavanet/lava/pairing/relay.proto

@@ -95,16 +95,23 @@ message RelayReply {
 }
 
 message QualityOfServiceReport{
+    // Latency of provider answers in milliseconds, range 0-inf, lower is better
     string latency = 1 [
         (gogoproto.moretags) = "yaml:\"Latency\"",
         (gogoproto.customtype) = "github.com/cosmos/cosmos-sdk/types.Dec",
         (gogoproto.nullable)   = false
         ];
+
+    // Percentage of times the provider returned a non-error response, range 0-1, higher is better
     string availability  = 2 [
         (gogoproto.moretags) = "yaml:\"availability\"",
         (gogoproto.customtype) = "github.com/cosmos/cosmos-sdk/types.Dec",
         (gogoproto.nullable)   = false
         ];
+
+    // Amount of time the provider is not synced (have the latest block) in milliseconds, range 0-inf, lower is better.
+    // Example: in ETH we have 15sec block time. So sync = 15000 means that the provider is one block
+    // behind the actual latest block.
     string sync = 3 [
         (gogoproto.moretags) = "yaml:\"sync\"",
         (gogoproto.customtype) = "github.com/cosmos/cosmos-sdk/types.Dec",

proto/lavanet/lava/pairing/reputation.proto

@@ -0,0 +1,44 @@
+syntax = "proto3";
+package lavanet.lava.pairing;
+
+option go_package = "github.com/lavanet/lava/v4/x/pairing/types";
+import "gogoproto/gogo.proto";
+import "cosmos/base/v1beta1/coin.proto"; 
+
+// Frac is a fracture struct that helps calculating and updating weighted average on the go
+message Frac {
+	string num = 1 [(gogoproto.moretags) = "yaml:\"num\"", (gogoproto.customtype) = "github.com/cosmos/cosmos-sdk/types.Dec",
+        (gogoproto.nullable) = false];  // weighted average numerator (w1*s1+w2*s2+...)
+	string denom = 2 [(gogoproto.moretags) = "yaml:\"denom\"", (gogoproto.customtype) = "github.com/cosmos/cosmos-sdk/types.Dec",
+        (gogoproto.nullable) = false]; // weighted denominator (w1+w2+...)
+}
+
+// QosScore holds the QoS score from a QoS excellence report. The score and its variance are updated over time using a weighted average.
+// Currently, the weight is the amount of relays that are associated with the QoS report.
+message QosScore {
+	Frac score = 1 [(gogoproto.nullable) = false];
+	Frac variance = 2 [(gogoproto.nullable) = false];
+}
+
+// Reputation keeps the QosScore of a provider for a specific cluster for in the provider's geolocation. 
+// The store key is provider+chain+cluster.
+// The epoch_score is a QosScore object that is aggregated over an epoch. When an epoch ends, the "score" field is updated
+// with the epoch_score and the epoch_score is reset.
+// The time_last_updated is used to calculate the appropriate time decay upon update.
+// The creation_time is used to determine if the variance stabilization period has passed and score can be truncated.
+// The stake is used when converting the reputation QoS scores to repuatation pairing score.
+message Reputation {
+	QosScore score = 1 [(gogoproto.nullable) = false];
+	QosScore epoch_score = 2 [(gogoproto.nullable) = false];
+	int64 time_last_updated = 3;
+	int64 creation_time = 4;
+	cosmos.base.v1beta1.Coin stake = 5 [(gogoproto.nullable) = false]; 
+}
+
+// ReputationPairingScore holds the reputation pairing score used by the reputation pairing requirement. 
+// The score is ranged between [0.5-2]. It's kept in the reputations fixation store with a provider+chain+cluster key.
+message ReputationPairingScore {
+	string score = 1 [(gogoproto.moretags) = "yaml:\"score\"", (gogoproto.customtype) = "github.com/cosmos/cosmos-sdk/types.Dec",
+		(gogoproto.nullable) = false];
+}
+

protocol/chainlib/chain_router_test.go

@@ -1175,8 +1175,8 @@ func TestMain(m *testing.M) {
 	listener := createRPCServer()
 	for {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-		_, err := rpcclient.DialContext(ctx, listenerAddressHttp)
-		_, err2 := rpcclient.DialContext(ctx, listenerAddressWs)
+		_, err := rpcclient.DialContext(ctx, listenerAddressHttp, nil)
+		_, err2 := rpcclient.DialContext(ctx, listenerAddressWs, nil)
 		if err2 != nil {
 			utils.LavaFormatDebug("waiting for grpc server to launch")
 			continue

protocol/chainlib/chainproxy/connector.go

@@ -107,6 +107,18 @@ func (connector *Connector) numberOfUsedClients() int {
 	return int(atomic.LoadInt64(&connector.usedClients))
 }
 
+func (connector *Connector) getRpcClient(ctx context.Context, nodeUrl common.NodeUrl) (*rpcclient.Client, error) {
+	authPathNodeUrl := nodeUrl.AuthConfig.AddAuthPath(nodeUrl.Url)
+	// origin used for auth header in the websocket case
+	authHeaders := nodeUrl.GetAuthHeaders()
+	rpcClient, err := rpcclient.DialContext(ctx, authPathNodeUrl, authHeaders)
+	if err != nil {
+		return nil, err
+	}
+	nodeUrl.SetAuthHeaders(ctx, rpcClient.SetHeader)
+	return rpcClient, nil
+}
+
 func (connector *Connector) createConnection(ctx context.Context, nodeUrl common.NodeUrl, currentNumberOfConnections int) (*rpcclient.Client, error) {
 	var rpcClient *rpcclient.Client
 	var err error
@@ -124,21 +136,13 @@ func (connector *Connector) createConnection(ctx context.Context, nodeUrl common
 		}
 		timeout := common.AverageWorldLatency * (1 + time.Duration(numberOfConnectionAttempts))
 		nctx, cancel := nodeUrl.LowerContextTimeoutWithDuration(ctx, timeout)
-		// add auth path
-		authPathNodeUrl := nodeUrl.AuthConfig.AddAuthPath(nodeUrl.Url)
-		rpcClient, err = rpcclient.DialContext(nctx, authPathNodeUrl)
+		// get rpcClient
+		rpcClient, err = connector.getRpcClient(nctx, nodeUrl)
 		if err != nil {
-			utils.LavaFormatWarning("Could not connect to the node, retrying", err, []utils.Attribute{
-				{Key: "Current Number Of Connections", Value: currentNumberOfConnections},
-				{Key: "Network Address", Value: authPathNodeUrl},
-				{Key: "Number Of Attempts", Value: numberOfConnectionAttempts},
-				{Key: "timeout", Value: timeout},
-			}...)
 			cancel()
 			continue
 		}
 		cancel()
-		nodeUrl.SetAuthHeaders(ctx, rpcClient.SetHeader)
 		break
 	}
 
@@ -178,7 +182,8 @@ func (connector *Connector) increaseNumberOfClients(ctx context.Context, numberO
 	var err error
 	for connectionAttempt := 0; connectionAttempt < MaximumNumberOfParallelConnectionsAttempts; connectionAttempt++ {
 		nctx, cancel := connector.nodeUrl.LowerContextTimeoutWithDuration(ctx, common.AverageWorldLatency*2)
-		rpcClient, err = rpcclient.DialContext(nctx, connector.nodeUrl.Url)
+		// get rpcClient
+		rpcClient, err = connector.getRpcClient(nctx, connector.nodeUrl)
 		if err != nil {
 			utils.LavaFormatDebug(
 				"could no increase number of connections to the node jsonrpc connector, retrying",