feat: PRT - Degrade availability on node errors

protocol/chainlib/consumer_ws_subscription_manager_test.go

@@ -10,20 +10,20 @@ import (
 	"testing"
 	"time"
 
-	gomock "github.com/golang/mock/gomock"
 	"github.com/lavanet/lava/v4/protocol/chainlib/extensionslib"
 	"github.com/lavanet/lava/v4/protocol/common"
 	"github.com/lavanet/lava/v4/protocol/lavaprotocol"
 	"github.com/lavanet/lava/v4/protocol/lavasession"
 	"github.com/lavanet/lava/v4/protocol/metrics"
 	"github.com/lavanet/lava/v4/protocol/provideroptimizer"
+	"github.com/lavanet/lava/v4/protocol/qos"
 	"github.com/lavanet/lava/v4/utils"
 	"github.com/lavanet/lava/v4/utils/rand"
 	pairingtypes "github.com/lavanet/lava/v4/x/pairing/types"
 	spectypes "github.com/lavanet/lava/v4/x/spec/types"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	gomockuber "go.uber.org/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 const (
@@ -448,7 +448,7 @@ func TestConsumerWSSubscriptionManager(t *testing.T) {
 			relaySender := NewMockRelaySender(ctrl)
 			relaySender.
 				EXPECT().
-				SendParsedRelay(gomock.Any(), gomock.Any(), gomockuber.Cond(func(x any) bool {
+				SendParsedRelay(gomock.Any(), gomock.Any(), gomock.Cond(func(x any) bool {
 					protocolMsg, ok := x.(ProtocolMessage)
 					require.True(t, ok)
 					require.NotNil(t, protocolMsg)
@@ -479,7 +479,7 @@ func TestConsumerWSSubscriptionManager(t *testing.T) {
 
 			relaySender.
 				EXPECT().
-				ParseRelay(gomock.Any(), gomock.Any(), gomockuber.Cond(func(x any) bool {
+				ParseRelay(gomock.Any(), gomock.Any(), gomock.Cond(func(x any) bool {
 					reqData, ok := x.(string)
 					require.True(t, ok)
 					areEqual := reqData == string(play.unsubscribeMessage1)
@@ -490,7 +490,7 @@ func TestConsumerWSSubscriptionManager(t *testing.T) {
 
 			relaySender.
 				EXPECT().
-				ParseRelay(gomock.Any(), gomock.Any(), gomockuber.Cond(func(x any) bool {
+				ParseRelay(gomock.Any(), gomock.Any(), gomock.Cond(func(x any) bool {
 					reqData, ok := x.(string)
 					require.True(t, ok)
 					areEqual := reqData == string(play.subscriptionRequestData1)
@@ -595,7 +595,7 @@ func TestConsumerWSSubscriptionManager(t *testing.T) {
 			subscribeProtocolMessage2 := NewProtocolMessage(subscribeChainMessage2, nil, nil, dapp2, ts.Consumer.Addr.String())
 			relaySender.
 				EXPECT().
-				ParseRelay(gomock.Any(), gomock.Any(), gomockuber.Cond(func(x any) bool {
+				ParseRelay(gomock.Any(), gomock.Any(), gomock.Cond(func(x any) bool {
 					reqData, ok := x.(string)
 					require.True(t, ok)
 					areEqual := reqData == string(play.unsubscribeMessage2)
@@ -606,7 +606,7 @@ func TestConsumerWSSubscriptionManager(t *testing.T) {
 
 			relaySender.
 				EXPECT().
-				ParseRelay(gomock.Any(), gomock.Any(), gomockuber.Cond(func(x any) bool {
+				ParseRelay(gomock.Any(), gomock.Any(), gomock.Cond(func(x any) bool {
 					reqData, ok := x.(string)
 					require.True(t, ok)
 					areEqual := reqData == string(play.subscriptionRequestData2)
@@ -725,8 +725,8 @@ func CreateConsumerSessionManager(chainID, apiInterface, consumerPublicAddress s
 	baseLatency := common.AverageWorldLatency / 2 // we want performance to be half our timeout or better
 	return lavasession.NewConsumerSessionManager(
 		&lavasession.RPCEndpoint{NetworkAddress: "stub", ChainID: chainID, ApiInterface: apiInterface, TLSEnabled: false, HealthCheckPath: "/", Geolocation: 0},
-		provideroptimizer.NewProviderOptimizer(provideroptimizer.STRATEGY_BALANCED, 0, baseLatency, 1, nil, "dontcare"),
 		nil, nil, consumerPublicAddress,
 		lavasession.NewActiveSubscriptionProvidersStorage(),
+		qos.NewQoSManager(provideroptimizer.NewProviderOptimizer(provideroptimizer.STRATEGY_BALANCED, 0, baseLatency, 1, nil, "dontcare")),
 	)
 }

protocol/chainlib/protocol_message.go

@@ -8,6 +8,7 @@ import (
 	"github.com/lavanet/lava/v4/utils"
 	pairingtypes "github.com/lavanet/lava/v4/x/pairing/types"
 	"github.com/lavanet/lava/v4/x/spec/types"
+	spectypes "github.com/lavanet/lava/v4/x/spec/types"
 )
 
 type UserData struct {
@@ -94,6 +95,11 @@ func (bpm *BaseProtocolMessage) GetBlockedProviders() []string {
 	return nil
 }
 
+func (bpm *BaseProtocolMessage) IsDefaultApi() bool {
+	api := bpm.GetApi()
+	return AllowMissingApisByDefault && strings.HasPrefix(api.Name, "Default-") && api.BlockParsing.ParserFunc == spectypes.PARSER_FUNC_EMPTY
+}
+
 func NewProtocolMessage(chainMessage ChainMessage, directiveHeaders map[string]string, relayRequestData *pairingtypes.RelayPrivateData, dappId, consumerIp string) ProtocolMessage {
 	return &BaseProtocolMessage{
 		ChainMessage:     chainMessage,
@@ -111,4 +117,5 @@ type ProtocolMessage interface {
 	GetBlockedProviders() []string
 	GetUserData() common.UserData
 	UpdateEarliestAndValidateExtensionRules(extensionParser *extensionslib.ExtensionParser, earliestBlockHashRequested int64, addon string, seenBlock int64) bool
+	IsDefaultApi() bool
 }

protocol/chainlib/protocol_message_test.go

@@ -0,0 +1,74 @@
+package chainlib
+
+import (
+	"testing"
+
+	"github.com/lavanet/lava/v4/x/spec/types"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsDefaultApi(t *testing.T) {
+	t.Run("default api", func(t *testing.T) {
+		AllowMissingApisByDefault = true
+		baseProtocolMessage := &BaseProtocolMessage{
+			ChainMessage: &baseChainMessageContainer{
+				api: &types.Api{
+					Name: "Default-API",
+					BlockParsing: types.BlockParser{
+						ParserFunc: types.PARSER_FUNC_EMPTY,
+					},
+				},
+			},
+		}
+
+		assert.True(t, baseProtocolMessage.IsDefaultApi())
+	})
+
+	t.Run("non-default api - AllowMissingApisByDefault is false", func(t *testing.T) {
+		AllowMissingApisByDefault = false
+		baseProtocolMessage := &BaseProtocolMessage{
+			ChainMessage: &baseChainMessageContainer{
+				api: &types.Api{
+					Name: "Default-API",
+					BlockParsing: types.BlockParser{
+						ParserFunc: types.PARSER_FUNC_EMPTY,
+					},
+				},
+			},
+		}
+
+		assert.False(t, baseProtocolMessage.IsDefaultApi())
+	})
+
+	t.Run("non-default api - Name is not Default-*", func(t *testing.T) {
+		AllowMissingApisByDefault = true
+		baseProtocolMessage := &BaseProtocolMessage{
+			ChainMessage: &baseChainMessageContainer{
+				api: &types.Api{
+					Name: "API",
+					BlockParsing: types.BlockParser{
+						ParserFunc: types.PARSER_FUNC_EMPTY,
+					},
+				},
+			},
+		}
+
+		assert.False(t, baseProtocolMessage.IsDefaultApi())
+	})
+
+	t.Run("non-default api - BlockParsing is not empty", func(t *testing.T) {
+		AllowMissingApisByDefault = true
+		baseProtocolMessage := &BaseProtocolMessage{
+			ChainMessage: &baseChainMessageContainer{
+				api: &types.Api{
+					Name: "Default-API",
+					BlockParsing: types.BlockParser{
+						ParserFunc: types.PARSER_FUNC_DEFAULT,
+					},
+				},
+			},
+		}
+
+		assert.False(t, baseProtocolMessage.IsDefaultApi())
+	})
+}

protocol/common/endpoints.go

@@ -251,9 +251,9 @@ type ConflictHandlerInterface interface {
 }
 
 type ProviderInfo struct {
-	ProviderAddress              string
-	ProviderQoSExcellenceSummery sdk.Dec // the number represents the average qos for this provider session
-	ProviderStake                sdk.Coin
+	ProviderAddress           string
+	ProviderReputationSummery sdk.Dec // the number represents the average qos for this provider session
+	ProviderStake             sdk.Coin
 }
 
 type RelayResult struct {

protocol/integration/protocol_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/lavanet/lava/v4/protocol/metrics"
 	"github.com/lavanet/lava/v4/protocol/performance"
 	"github.com/lavanet/lava/v4/protocol/provideroptimizer"
+	"github.com/lavanet/lava/v4/protocol/qos"
 	"github.com/lavanet/lava/v4/protocol/rpcconsumer"
 	"github.com/lavanet/lava/v4/protocol/rpcprovider"
 	"github.com/lavanet/lava/v4/protocol/rpcprovider/reliabilitymanager"
@@ -234,7 +235,8 @@ func createRpcConsumer(t *testing.T, ctx context.Context, rpcConsumerOptions rpc
 	_, averageBlockTime, _, _ := chainParser.ChainBlockStats()
 	baseLatency := common.AverageWorldLatency / 2
 	optimizer := provideroptimizer.NewProviderOptimizer(provideroptimizer.STRATEGY_BALANCED, averageBlockTime, baseLatency, 2, nil, "dontcare")
-	consumerSessionManager := lavasession.NewConsumerSessionManager(rpcEndpoint, optimizer, nil, nil, "test", lavasession.NewActiveSubscriptionProvidersStorage())
+	qosManager := qos.NewQoSManager(optimizer)
+	consumerSessionManager := lavasession.NewConsumerSessionManager(rpcEndpoint, nil, nil, "test", lavasession.NewActiveSubscriptionProvidersStorage(), qosManager)
 	consumerSessionManager.UpdateAllProviders(rpcConsumerOptions.epoch, rpcConsumerOptions.pairingList)
 
 	// Just setting the providers available extensions and policy so the consumer is aware of them
@@ -279,7 +281,7 @@ func createRpcConsumer(t *testing.T, ctx context.Context, rpcConsumerOptions rpc
 	consumerCmdFlags := common.ConsumerCmdFlags{}
 	rpcconsumerLogs, err := metrics.NewRPCConsumerLogs(nil, nil, nil)
 	require.NoError(t, err)
-	err = rpcConsumerServer.ServeRPCRequests(ctx, rpcEndpoint, consumerStateTracker, chainParser, finalizationConsensus, consumerSessionManager, rpcConsumerOptions.requiredResponses, rpcConsumerOptions.account.SK, rpcConsumerOptions.lavaChainID, cache, rpcconsumerLogs, rpcConsumerOptions.account.Addr, consumerConsistency, nil, consumerCmdFlags, false, nil, nil, nil)
+	err = rpcConsumerServer.ServeRPCRequests(ctx, rpcEndpoint, consumerStateTracker, chainParser, finalizationConsensus, consumerSessionManager, rpcConsumerOptions.requiredResponses, rpcConsumerOptions.account.SK, rpcConsumerOptions.lavaChainID, cache, rpcconsumerLogs, rpcConsumerOptions.account.Addr, consumerConsistency, nil, consumerCmdFlags, false, nil, nil, nil, consumerSessionManager)
 	require.NoError(t, err)
 
 	// wait for consumer to finish initialization

protocol/lavaprotocol/finalizationconsensus/finalization_consensus_test.go

@@ -16,6 +16,8 @@ import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/lavanet/lava/v4/protocol/chainlib"
 	"github.com/lavanet/lava/v4/protocol/lavasession"
+	"github.com/lavanet/lava/v4/protocol/provideroptimizer"
+	"github.com/lavanet/lava/v4/protocol/qos"
 	pairingtypes "github.com/lavanet/lava/v4/x/pairing/types"
 	spectypes "github.com/lavanet/lava/v4/x/spec/types"
 	"github.com/stretchr/testify/require"
@@ -202,13 +204,23 @@ func TestConsensusHashesInsertion(t *testing.T) {
 
 func TestQoS(t *testing.T) {
 	decToSet, _ := sdk.NewDecFromStr("0.05") // test values fit 0.05 Availability requirements
-	lavasession.AvailabilityPercentage = decToSet
+	qos.AvailabilityPercentage = decToSet
 	rand.InitRandomSeed()
 	chainsToTest := []string{"APT1", "LAV1", "ETH1"}
+
+	waitForDoneChan := func(doneChan <-chan struct{}) {
+		select {
+		case <-doneChan:
+		case <-time.After(5 * time.Second):
+			t.Fatal("timeout waiting for qos calculation to finish")
+		}
+	}
+
 	for i := 0; i < 10; i++ {
 		for _, chainID := range chainsToTest {
 			t.Run(chainID, func(t *testing.T) {
 				ctx := context.Background()
+				qosManager := qos.NewQoSManager(&provideroptimizer.ProviderOptimizer{})
 				chainParser, _, _, closeServer, _, err := chainlib.CreateChainLibMocks(ctx, chainID, "0", func(http.ResponseWriter, *http.Request) {}, nil, "../../../", nil)
 				if closeServer != nil {
 					defer closeServer()
@@ -282,54 +294,65 @@ func TestQoS(t *testing.T) {
 				currentLatency := time.Millisecond
 				expectedLatency := time.Millisecond
 				latestServicedBlock := expectedBH
-				singleConsumerSession.CalculateQoS(currentLatency, expectedLatency, expectedBH-latestServicedBlock, numOfProviders, 1)
-				require.Equal(t, uint64(1), singleConsumerSession.QoSInfo.AnsweredRelays)
-				require.Equal(t, uint64(1), singleConsumerSession.QoSInfo.TotalRelays)
-				require.Equal(t, int64(1), singleConsumerSession.QoSInfo.SyncScoreSum)
-				require.Equal(t, int64(1), singleConsumerSession.QoSInfo.TotalSyncScore)
-				require.Equal(t, sdk.OneDec(), singleConsumerSession.QoSInfo.LastQoSReport.Availability)
-				require.Equal(t, sdk.OneDec(), singleConsumerSession.QoSInfo.LastQoSReport.Sync)
-				require.Equal(t, sdk.OneDec(), singleConsumerSession.QoSInfo.LastQoSReport.Latency)
+				waitForDoneChan(qosManager.CalculateQoS(epoch, singleConsumerSession.SessionId, "", currentLatency, expectedLatency, expectedBH-latestServicedBlock, numOfProviders, 1))
+				require.Equal(t, uint64(1), qosManager.GetAnsweredRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, uint64(1), qosManager.GetTotalRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(1), qosManager.GetSyncScoreSum(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(1), qosManager.GetTotalSyncScore(epoch, singleConsumerSession.SessionId))
+
+				lastQoSReport := qosManager.GetLastQoSReport(epoch, singleConsumerSession.SessionId)
+				require.Equal(t, sdk.OneDec(), lastQoSReport.Availability)
+				require.Equal(t, sdk.OneDec(), lastQoSReport.Sync)
+				require.Equal(t, sdk.OneDec(), lastQoSReport.Latency)
 
 				latestServicedBlock = expectedBH + 1
-				singleConsumerSession.CalculateQoS(currentLatency, expectedLatency, expectedBH-latestServicedBlock, numOfProviders, 1)
-				require.Equal(t, uint64(2), singleConsumerSession.QoSInfo.AnsweredRelays)
-				require.Equal(t, uint64(2), singleConsumerSession.QoSInfo.TotalRelays)
-				require.Equal(t, int64(2), singleConsumerSession.QoSInfo.SyncScoreSum)
-				require.Equal(t, int64(2), singleConsumerSession.QoSInfo.TotalSyncScore)
-				require.Equal(t, sdk.OneDec(), singleConsumerSession.QoSInfo.LastQoSReport.Availability)
-				require.Equal(t, sdk.OneDec(), singleConsumerSession.QoSInfo.LastQoSReport.Sync)
-				require.Equal(t, sdk.OneDec(), singleConsumerSession.QoSInfo.LastQoSReport.Latency)
-
-				singleConsumerSession.QoSInfo.TotalRelays++ // this is how we add a failure
-				singleConsumerSession.CalculateQoS(currentLatency, expectedLatency, expectedBH-latestServicedBlock, numOfProviders, 1)
-				require.Equal(t, uint64(3), singleConsumerSession.QoSInfo.AnsweredRelays)
-				require.Equal(t, uint64(4), singleConsumerSession.QoSInfo.TotalRelays)
-				require.Equal(t, int64(3), singleConsumerSession.QoSInfo.SyncScoreSum)
-				require.Equal(t, int64(3), singleConsumerSession.QoSInfo.TotalSyncScore)
-
-				require.Equal(t, sdk.ZeroDec(), singleConsumerSession.QoSInfo.LastQoSReport.Availability) // because availability below 95% is 0
-				require.Equal(t, sdk.OneDec(), singleConsumerSession.QoSInfo.LastQoSReport.Sync)
-				require.Equal(t, sdk.OneDec(), singleConsumerSession.QoSInfo.LastQoSReport.Latency)
+				waitForDoneChan(qosManager.CalculateQoS(epoch, singleConsumerSession.SessionId, "", currentLatency, expectedLatency, expectedBH-latestServicedBlock, numOfProviders, 1))
+				require.Equal(t, uint64(2), qosManager.GetAnsweredRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, uint64(2), qosManager.GetTotalRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(2), qosManager.GetSyncScoreSum(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(2), qosManager.GetTotalSyncScore(epoch, singleConsumerSession.SessionId))
+
+				lastQoSReport = qosManager.GetLastQoSReport(epoch, singleConsumerSession.SessionId)
+				require.Equal(t, sdk.OneDec(), lastQoSReport.Availability)
+				require.Equal(t, sdk.OneDec(), lastQoSReport.Sync)
+				require.Equal(t, sdk.OneDec(), lastQoSReport.Latency)
+
+				waitForDoneChan(qosManager.AddFailedRelay(epoch, singleConsumerSession.SessionId)) // this is how we add a failure
+				waitForDoneChan(qosManager.CalculateQoS(epoch, singleConsumerSession.SessionId, "", currentLatency, expectedLatency, expectedBH-latestServicedBlock, numOfProviders, 1))
+				require.Equal(t, uint64(3), qosManager.GetAnsweredRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, uint64(4), qosManager.GetTotalRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(3), qosManager.GetSyncScoreSum(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(3), qosManager.GetTotalSyncScore(epoch, singleConsumerSession.SessionId))
+
+				lastQoSReport = qosManager.GetLastQoSReport(epoch, singleConsumerSession.SessionId)
+				require.Equal(t, sdk.ZeroDec(), lastQoSReport.Availability) // because availability below 95% is 0
+				require.Equal(t, sdk.OneDec(), lastQoSReport.Sync)
+				require.Equal(t, sdk.OneDec(), lastQoSReport.Latency)
 
 				latestServicedBlock = expectedBH - 1 // is one block below threshold
-				singleConsumerSession.CalculateQoS(currentLatency, expectedLatency*2, expectedBH-latestServicedBlock, numOfProviders, 1)
-				require.Equal(t, uint64(4), singleConsumerSession.QoSInfo.AnsweredRelays)
-				require.Equal(t, uint64(5), singleConsumerSession.QoSInfo.TotalRelays)
-				require.Equal(t, int64(3), singleConsumerSession.QoSInfo.SyncScoreSum)
-				require.Equal(t, int64(4), singleConsumerSession.QoSInfo.TotalSyncScore)
-
-				require.Equal(t, sdk.ZeroDec(), singleConsumerSession.QoSInfo.LastQoSReport.Availability) // because availability below 95% is 0
-				require.Equal(t, sdk.MustNewDecFromStr("0.75"), singleConsumerSession.QoSInfo.LastQoSReport.Sync)
-				require.Equal(t, sdk.OneDec(), singleConsumerSession.QoSInfo.LastQoSReport.Latency)
+				waitForDoneChan(qosManager.CalculateQoS(epoch, singleConsumerSession.SessionId, "", currentLatency, expectedLatency*2, expectedBH-latestServicedBlock, numOfProviders, 1))
+				require.Equal(t, uint64(4), qosManager.GetAnsweredRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, uint64(5), qosManager.GetTotalRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(3), qosManager.GetSyncScoreSum(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(4), qosManager.GetTotalSyncScore(epoch, singleConsumerSession.SessionId))
+
+				lastQoSReport = qosManager.GetLastQoSReport(epoch, singleConsumerSession.SessionId)
+				require.Equal(t, sdk.ZeroDec(), lastQoSReport.Availability) // because availability below 95% is 0
+				require.Equal(t, sdk.MustNewDecFromStr("0.75"), lastQoSReport.Sync)
+				require.Equal(t, sdk.OneDec(), lastQoSReport.Latency)
+
 				latestServicedBlock = expectedBH + 1
 				// add in a loop so availability goes above 95%
+				doneChan := make(<-chan struct{})
 				for i := 5; i < 100; i++ {
-					singleConsumerSession.CalculateQoS(currentLatency, expectedLatency*2, expectedBH-latestServicedBlock, numOfProviders, 1)
+					doneChan = qosManager.CalculateQoS(epoch, singleConsumerSession.SessionId, "", currentLatency, expectedLatency*2, expectedBH-latestServicedBlock, numOfProviders, 1)
 				}
-				require.Equal(t, sdk.MustNewDecFromStr("0.8"), singleConsumerSession.QoSInfo.LastQoSReport.Availability) // because availability below 95% is 0
-				require.Equal(t, sdk.MustNewDecFromStr("0.989898989898989898"), singleConsumerSession.QoSInfo.LastQoSReport.Sync)
-				require.Equal(t, sdk.OneDec(), singleConsumerSession.QoSInfo.LastQoSReport.Latency)
+				waitForDoneChan(doneChan)
+
+				lastQoSReport = qosManager.GetLastQoSReport(epoch, singleConsumerSession.SessionId)
+				require.Equal(t, sdk.MustNewDecFromStr("0.8"), lastQoSReport.Availability) // because availability below 95% is 0
+				require.Equal(t, sdk.MustNewDecFromStr("0.989898989898989898"), lastQoSReport.Sync)
+				require.Equal(t, sdk.OneDec(), lastQoSReport.Latency)
 
 				finalizationInsertionsSpreadBlocks := []finalizationTestInsertion{
 					finalizationInsertionForProviders(chainID, epoch, 200, 0, 1, true, "", blocksInFinalizationProof, blockDistanceForFinalizedData)[0],