refactor: PRT - QoS Manager refactor

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/lavaprotocol/finalizationconsensus/finalization_consensus_test.go

@@ -16,6 +16,7 @@ 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/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,9 +203,18 @@ 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) {
@@ -282,54 +292,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(singleConsumerSession.QoSManager.CalculateQoS(epoch, singleConsumerSession.SessionId, "", currentLatency, expectedLatency, expectedBH-latestServicedBlock, numOfProviders, 1))
+				require.Equal(t, uint64(1), singleConsumerSession.QoSManager.GetAnsweredRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, uint64(1), singleConsumerSession.QoSManager.GetTotalRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(1), singleConsumerSession.QoSManager.GetSyncScoreSum(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(1), singleConsumerSession.QoSManager.GetTotalSyncScore(epoch, singleConsumerSession.SessionId))
+
+				lastQoSReport := singleConsumerSession.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(singleConsumerSession.QoSManager.CalculateQoS(epoch, singleConsumerSession.SessionId, "", currentLatency, expectedLatency, expectedBH-latestServicedBlock, numOfProviders, 1))
+				require.Equal(t, uint64(2), singleConsumerSession.QoSManager.GetAnsweredRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, uint64(2), singleConsumerSession.QoSManager.GetTotalRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(2), singleConsumerSession.QoSManager.GetSyncScoreSum(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(2), singleConsumerSession.QoSManager.GetTotalSyncScore(epoch, singleConsumerSession.SessionId))
+
+				lastQoSReport = singleConsumerSession.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(singleConsumerSession.QoSManager.AddFailedRelay(epoch, singleConsumerSession.SessionId)) // this is how we add a failure
+				waitForDoneChan(singleConsumerSession.QoSManager.CalculateQoS(epoch, singleConsumerSession.SessionId, "", currentLatency, expectedLatency, expectedBH-latestServicedBlock, numOfProviders, 1))
+				require.Equal(t, uint64(3), singleConsumerSession.QoSManager.GetAnsweredRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, uint64(4), singleConsumerSession.QoSManager.GetTotalRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(3), singleConsumerSession.QoSManager.GetSyncScoreSum(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(3), singleConsumerSession.QoSManager.GetTotalSyncScore(epoch, singleConsumerSession.SessionId))
+
+				lastQoSReport = singleConsumerSession.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(singleConsumerSession.QoSManager.CalculateQoS(epoch, singleConsumerSession.SessionId, "", currentLatency, expectedLatency*2, expectedBH-latestServicedBlock, numOfProviders, 1))
+				require.Equal(t, uint64(4), singleConsumerSession.QoSManager.GetAnsweredRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, uint64(5), singleConsumerSession.QoSManager.GetTotalRelays(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(3), singleConsumerSession.QoSManager.GetSyncScoreSum(epoch, singleConsumerSession.SessionId))
+				require.Equal(t, int64(4), singleConsumerSession.QoSManager.GetTotalSyncScore(epoch, singleConsumerSession.SessionId))
+
+				lastQoSReport = singleConsumerSession.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 = singleConsumerSession.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 = singleConsumerSession.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],

protocol/lavaprotocol/request_builder.go

@@ -71,8 +71,8 @@ func ConstructRelaySession(lavaChainID string, relayRequestData *pairingtypes.Re
 		return nil
 	}
 
-	copiedQOS := copyQoSServiceReport(singleConsumerSession.QoSInfo.LastQoSReport)
-	copiedExcellenceQOS := copyQoSServiceReport(singleConsumerSession.QoSInfo.LastExcellenceQoSReport)
+	copiedQOS := copyQoSServiceReport(singleConsumerSession.QoSManager.GetLastQoSReport(uint64(epoch), singleConsumerSession.SessionId))
+	copiedReputation := copyQoSServiceReport(singleConsumerSession.QoSManager.GetLastReputationQoSReport(uint64(epoch), singleConsumerSession.SessionId)) // copy reputation report for the node
 
 	return &pairingtypes.RelaySession{
 		SpecId:                chainID,
@@ -87,7 +87,7 @@ func ConstructRelaySession(lavaChainID string, relayRequestData *pairingtypes.Re
 		LavaChainId:           lavaChainID,
 		Sig:                   nil,
 		Badge:                 nil,
-		QosExcellenceReport:   copiedExcellenceQOS,
+		QosExcellenceReport:   copiedReputation,
 	}
 }
 

protocol/lavaprotocol/response_builder_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/lavanet/lava/v4/protocol/lavaprotocol/finalizationverification"
 	"github.com/lavanet/lava/v4/protocol/lavasession"
+	"github.com/lavanet/lava/v4/protocol/qos"
 	"github.com/lavanet/lava/v4/utils/sigs"
 	pairingtypes "github.com/lavanet/lava/v4/x/pairing/types"
 	spectypes "github.com/lavanet/lava/v4/x/spec/types"
@@ -29,7 +30,7 @@ func TestSignAndExtractResponse(t *testing.T) {
 	singleConsumerSession := &lavasession.SingleConsumerSession{
 		CuSum:              20,
 		LatestRelayCu:      10, // set by GetSessions cuNeededForSession
-		QoSInfo:            lavasession.QoSReport{LastQoSReport: &pairingtypes.QualityOfServiceReport{}},
+		QoSManager:         qos.NewQoSManager(),
 		SessionId:          123,
 		Parent:             nil,
 		RelayNum:           1,
@@ -77,7 +78,7 @@ func TestSignAndExtractResponseLatest(t *testing.T) {
 	singleConsumerSession := &lavasession.SingleConsumerSession{
 		CuSum:              20,
 		LatestRelayCu:      10, // set by GetSessions cuNeededForSession
-		QoSInfo:            lavasession.QoSReport{LastQoSReport: &pairingtypes.QualityOfServiceReport{}},
+		QoSManager:         qos.NewQoSManager(),
 		SessionId:          123,
 		Parent:             nil,
 		RelayNum:           1,

protocol/lavaprotocol/reuqest_builder_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	"github.com/lavanet/lava/v4/protocol/lavasession"
+	"github.com/lavanet/lava/v4/protocol/qos"
 	"github.com/lavanet/lava/v4/utils/sigs"
 	pairingtypes "github.com/lavanet/lava/v4/x/pairing/types"
 	"github.com/stretchr/testify/require"
@@ -18,7 +19,7 @@ func TestSignAndExtract(t *testing.T) {
 	singleConsumerSession := &lavasession.SingleConsumerSession{
 		CuSum:              20,
 		LatestRelayCu:      10, // set by GetSessions cuNeededForSession
-		QoSInfo:            lavasession.QoSReport{LastQoSReport: &pairingtypes.QualityOfServiceReport{}},
+		QoSManager:         qos.NewQoSManager(),
 		SessionId:          123,
 		Parent:             nil,
 		RelayNum:           1,

protocol/lavasession/common.go

@@ -15,7 +15,6 @@ import (
 	sdkerrors "cosmossdk.io/errors"
 	"golang.org/x/exp/slices"
 
-	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/gogo/status"
 	"github.com/lavanet/lava/v4/protocol/chainlib/chainproxy"
 	"github.com/lavanet/lava/v4/utils"
@@ -48,14 +47,11 @@ const (
 	unixPrefix                                       = "unix:"
 )
 
-var AvailabilityPercentage sdk.Dec = sdk.NewDecWithPrec(1, 1) // TODO move to params pairing
 const (
-	PercentileToCalculateLatency = 0.9
-	MinProvidersForSync          = 0.6
-	OptimizerPerturbation        = 0.10
-	LatencyThresholdStatic       = 1 * time.Second
-	LatencyThresholdSlope        = 1 * time.Millisecond
-	StaleEpochDistance           = 3 // relays done 3 epochs back are ready to be rewarded
+	OptimizerPerturbation  = 0.10
+	LatencyThresholdStatic = 1 * time.Second
+	LatencyThresholdSlope  = 1 * time.Millisecond
+	StaleEpochDistance     = 3 // relays done 3 epochs back are ready to be rewarded
 
 )
 

protocol/lavasession/consumer_session_manager.go

@@ -563,7 +563,7 @@ func (csm *ConsumerSessionManager) GetSessions(ctx context.Context, cuNeededForS
 				sessionInfo.QoSSummeryResult = consumerSession.getQosComputedResultOrZero()
 				sessions[providerAddress] = sessionInfo
 
-				qosReport, _ := csm.providerOptimizer.GetExcellenceQoSReportForProvider(providerAddress)
+				qosReport, _ := csm.providerOptimizer.GetReputationReportForProvider(providerAddress)
 				if csm.rpcEndpoint.Geolocation != uint64(endpoint.endpoint.Geolocation) {
 					// rawQosReport is used only when building the relay payment message to be used to update
 					// the provider's reputation on-chain. If the consumer and provider don't share geolocation
@@ -936,7 +936,7 @@ func (csm *ConsumerSessionManager) OnSessionFailure(consumerSession *SingleConsu
 		consumerSession.BlockListed = true
 	}
 
-	consumerSession.QoSInfo.TotalRelays++
+	consumerSession.QoSManager.AddFailedRelay(consumerSession.epoch, consumerSession.SessionId)
 	consumerSession.ConsecutiveErrors = append(consumerSession.ConsecutiveErrors, errorReceived)
 	// copy consecutive errors for report.
 	errorsForConsumerSession := consumerSession.ConsecutiveErrors
@@ -1047,7 +1047,7 @@ func (csm *ConsumerSessionManager) OnSessionDone(
 	consumerSession.ConsecutiveErrors = []error{}
 	consumerSession.LatestBlock = latestServicedBlock // update latest serviced block
 	// calculate QoS
-	consumerSession.CalculateQoS(currentLatency, expectedLatency, expectedBH-latestServicedBlock, numOfProviders, int64(providersCount))
+	consumerSession.QoSManager.CalculateQoS(csm.atomicReadCurrentEpoch(), consumerSession.SessionId, consumerSession.Parent.PublicLavaAddress, currentLatency, expectedLatency, expectedBH-latestServicedBlock, numOfProviders, int64(providersCount))
 	if !isHangingApi {
 		// append relay data only for non hanging apis
 		go csm.providerOptimizer.AppendRelayData(consumerSession.Parent.PublicLavaAddress, currentLatency, specComputeUnits, uint64(latestServicedBlock))
@@ -1066,21 +1066,25 @@ func (csm *ConsumerSessionManager) updateMetricsManager(consumerSession *SingleC
 	info := csm.RPCEndpoint()
 	apiInterface := info.ApiInterface
 	chainId := info.ChainID
+
 	var lastQos *pairingtypes.QualityOfServiceReport
-	var lastQosExcellence *pairingtypes.QualityOfServiceReport
-	if consumerSession.QoSInfo.LastQoSReport != nil {
-		qos := *consumerSession.QoSInfo.LastQoSReport
+	lastQoSReport := consumerSession.QoSManager.GetLastQoSReport(csm.atomicReadCurrentEpoch(), consumerSession.SessionId)
+	if lastQoSReport != nil {
+		qos := *lastQoSReport
 		lastQos = &qos
 	}
-	if consumerSession.QoSInfo.LastExcellenceQoSReport != nil {
-		qosEx := *consumerSession.QoSInfo.LastExcellenceQoSReport
-		lastQosExcellence = &qosEx
+
+	var lastReputation *pairingtypes.QualityOfServiceReport
+	lastReputationReport := consumerSession.QoSManager.GetLastReputationQoSReport(csm.atomicReadCurrentEpoch(), consumerSession.SessionId)
+	if lastReputationReport != nil {
+		qosRep := *lastReputationReport
+		lastReputation = &qosRep
 	}
 	publicProviderAddress := consumerSession.Parent.PublicLavaAddress
 	publicProviderEndpoint := consumerSession.Parent.Endpoints[0].NetworkAddress
 
 	go func() {
-		csm.consumerMetricsManager.SetQOSMetrics(chainId, apiInterface, publicProviderAddress, publicProviderEndpoint, lastQos, lastQosExcellence, consumerSession.LatestBlock, consumerSession.RelayNum, relayLatency, sessionSuccessful)
+		csm.consumerMetricsManager.SetQOSMetrics(chainId, apiInterface, publicProviderAddress, publicProviderEndpoint, lastQos, lastReputation, consumerSession.LatestBlock, consumerSession.RelayNum, relayLatency, sessionSuccessful)
 	}()
 }