[RelayMiner] Supplier rate limiting

config.yml

@@ -147,7 +147,8 @@ genesis:
         # Application module
         - address: pokt1rl3gjgzexmplmds3tq3r3yk84zlwdl6djzgsvm
           coins:
-            # TODO_MAINNET: Pass config.yml into ChatGPT to build a script that ensures the amounts line up
+            # TODO_MAINNET(@olshansk): Pass config.yml into ChatGPT to build a script
+            # that ensures the amounts line up
             - amount: "200000136" # MUST BE equal to the total of all app stakes below
               denom: upokt
         # Supplier module

pkg/relayer/miner/miner.go

@@ -89,9 +89,8 @@ func (mnr *miner) mapMineRelay(
 ) (_ either.Either[*relayer.MinedRelay], skip bool) {
 	relayBz, err := relay.Marshal()
 	if err != nil {
-		meteringError := mnr.relayMeter.SetNonApplicableRelayReward(ctx, relay.GetReq().GetMeta())
-		if meteringError != nil {
-			return either.Error[*relayer.MinedRelay](fmt.Errorf("failed to unclaim relay price: %w", meteringError)), false
+		if relayMeteringResult := mnr.unclaimRelayUPOKT(ctx, *relay); relayMeteringResult.IsError() {
+			return relayMeteringResult, false
 		}
 		return either.Error[*relayer.MinedRelay](err), false
 	}
@@ -100,18 +99,16 @@ func (mnr *miner) mapMineRelay(
 
 	relayDifficultyTargetHash, err := mnr.getServiceRelayDifficultyTargetHash(ctx, relay.Req)
 	if err != nil {
-		meteringError := mnr.relayMeter.SetNonApplicableRelayReward(ctx, relay.GetReq().GetMeta())
-		if meteringError != nil {
-			return either.Error[*relayer.MinedRelay](fmt.Errorf("failed to unclaim relay price: %w", meteringError)), false
+		if relayMeteringResult := mnr.unclaimRelayUPOKT(ctx, *relay); relayMeteringResult.IsError() {
+			return relayMeteringResult, true
 		}
-		return either.Error[*relayer.MinedRelay](err), false
+		return either.Error[*relayer.MinedRelay](err), true
 	}
 
 	// The relay IS NOT volume / reward applicable
 	if !protocol.IsRelayVolumeApplicable(relayHash, relayDifficultyTargetHash) {
-		meteringError := mnr.relayMeter.SetNonApplicableRelayReward(ctx, relay.GetReq().GetMeta())
-		if meteringError != nil {
-			return either.Error[*relayer.MinedRelay](fmt.Errorf("failed to unclaim relay price: %w", meteringError)), false
+		if eitherMeteringResult := mnr.unclaimRelayUPOKT(ctx, *relay); eitherMeteringResult.IsError() {
+			return eitherMeteringResult, true
 		}
 		return either.Success[*relayer.MinedRelay](nil), true
 	}
@@ -148,3 +145,13 @@ func (mnr *miner) getServiceRelayDifficultyTargetHash(ctx context.Context, req *
 
 	return serviceRelayDifficulty.GetTargetHash(), nil
 }
+
+// unclaimRelayUPOKT unclaims the relay UPOKT reward for the relay.
+// It returns an either.Error if the relay UPOKT reward could not be unclaimed.
+func (mnr *miner) unclaimRelayUPOKT(ctx context.Context, relay servicetypes.Relay) either.Either[*relayer.MinedRelay] {
+	if err := mnr.relayMeter.SetNonApplicableRelayReward(ctx, relay.GetReq().GetMeta()); err != nil {
+		return either.Error[*relayer.MinedRelay](err)
+	}
+
+	return either.Success[*relayer.MinedRelay](nil)
+}

pkg/relayer/miner/miner_test.go

@@ -50,9 +50,9 @@ func TestMiner_MinedRelays(t *testing.T) {
 
 	testqueryclients.SetServiceRelayDifficultyTargetHash(t, testSvcId, testRelayMiningTargetHash)
 	serviceQueryClientMock := testqueryclients.NewTestServiceQueryClient(t)
-	testRelayMeter := mockRelayMeter(t)
+	relayMeterMock := newMockRelayMeter(t)
 
-	deps := depinject.Supply(serviceQueryClientMock, testRelayMeter)
+	deps := depinject.Supply(serviceQueryClientMock, relayMeterMock)
 	mnr, err := miner.NewMiner(deps)
 	require.NoError(t, err)
 
@@ -158,8 +158,8 @@ func unmarshalHexMinedRelay(
 	}
 }
 
-// mockRelayMeter returns a mock RelayMeter that is used by the relay miner to claim and unclaim relays.
-func mockRelayMeter(t *testing.T) relayer.RelayMeter {
+// newMockRelayMeter returns a mock RelayMeter that is used by the relay miner to claim and unclaim relays.
+func newMockRelayMeter(t *testing.T) relayer.RelayMeter {
 	t.Helper()
 
 	ctrl := gomock.NewController(t)

pkg/relayer/proxy/errors.go

@@ -19,4 +19,5 @@ var (
 	ErrRelayerProxyMissingSupplierOperatorAddress = sdkerrors.Register(codespace, 11, "supplier operator address is missing")
 	ErrRelayerProxyUnknownSession                 = sdkerrors.Register(codespace, 12, "relayer proxy encountered unknown session")
 	ErrRelayerProxyRateLimited                    = sdkerrors.Register(codespace, 13, "offchain rate limit hit by relayer proxy")
+	ErrRelayerProxyUnclaimRelayPrice              = sdkerrors.Register(codespace, 14, "failed to unclaim relay price")
 )

pkg/relayer/proxy/proxy.go

@@ -81,7 +81,7 @@ type relayerProxy struct {
 
 	// relayMeter keeps track of the total amount of stake an onchhain Application
 	// will owe an onchain Supplier (backed by this RelayMiner) once the session settles.
-	// It also configures over servicing allowance.
+	// It also configures application over-servicing allowance.
 	relayMeter relayer.RelayMeter
 }
 

pkg/relayer/proxy/relay_meter.go

@@ -44,7 +44,10 @@ type sessionRelayMeter struct {
 	sessionHeader *sessiontypes.SessionHeader
 	// numOverServicedRelays is the number of relays that have been over-serviced
 	// by the relayer for the application.
-	numOverServicedRelays int
+	numOverServicedRelays uint64
+	// numOverservicedComputeUnits is the number of compute units that have been
+	// over-serviced by the relayer for the application.
+	numOverServicedComputeUnits uint64
 
 	// sharedParams, service and serviceRelayDifficulty are used to calculate the relay cost
 	// that increments the consumedAmount.
@@ -59,6 +62,8 @@ type sessionRelayMeter struct {
 // It ensures that no Application is over-serviced by the Supplier per session.
 // This is done by maintaining the max amount of stake the supplier can consume
 // per session and the amount of stake consumed by mined relays.
+// TODO_POST_MAINNET(@red-0ne): Consider making the relay meter a light client,
+// since it's already receiving all committed blocks and events.
 type ProxyRelayMeter struct {
 	// sessionToRelayMeterMap is a map of session IDs to their corresponding session relay meter.
 	// Only known applications (i.e. have sent at least one relay) have their stakes metered.
@@ -138,7 +143,7 @@ func (rmtr *ProxyRelayMeter) Start(ctx context.Context) error {
 // The relay reward is added optimistically, assuming that the relay will be volume / reward
 // applicable and the relay meter would remain up to date.
 func (rmtr *ProxyRelayMeter) AccumulateRelayReward(ctx context.Context, reqMeta servicetypes.RelayRequestMetadata) error {
-	// TODO_MAINNET: Locking the relay serving flow to ensure that the relay meter is updated
+	// TODO_MAINNET(@adshmh): Locking the relay serving flow to ensure that the relay meter is updated
 	// might be a bottleneck since ensureRequestAppMetrics is performing multiple
 	// sequential queries to the Pocket Network node.
 	// Re-evaluate when caching and invalidation is implemented.
@@ -166,7 +171,6 @@ func (rmtr *ProxyRelayMeter) AccumulateRelayReward(ctx context.Context, reqMeta
 	newConsumedCoin := appRelayMeter.consumedCoin.Add(relayCostCoin)
 
 	isAppOverServiced := appRelayMeter.maxCoin.IsLT(newConsumedCoin)
-
 	if !isAppOverServiced {
 		appRelayMeter.consumedCoin = newConsumedCoin
 		return nil
@@ -180,26 +184,23 @@ func (rmtr *ProxyRelayMeter) AccumulateRelayReward(ctx context.Context, reqMeta
 	// then return a rate limit error.
 	overServicingCoin := newConsumedCoin.Sub(appRelayMeter.maxCoin)
 
-	// In case Allowance is positive, add it to the maxCoin to allow no or limited over-servicing.
+	// In case Allowance is positive, add it to maxCoin to allow no or limited over-servicing.
 	if !allowUnlimitedOverServicing {
 		maxAllowedOverServicing := appRelayMeter.maxCoin.Add(rmtr.overServicingAllowanceCoins)
 		if maxAllowedOverServicing.IsLT(newConsumedCoin) {
 			return ErrRelayerProxyRateLimited.Wrapf(
-				"application has been rate limited, stake needed: %s, has: %s, ",
+				"application has been rate limited, stake needed: %s, has: %s",
 				newConsumedCoin.String(),
 				appRelayMeter.maxCoin.String(),
 			)
 		}
 	}
 
 	appRelayMeter.numOverServicedRelays++
-	numOverServicedRelays := appRelayMeter.numOverServicedRelays
-
-	// Exponential backoff: log only when numOverServicedRelays is a power of 2
-	shouldLog := (numOverServicedRelays & (numOverServicedRelays - 1)) == 0
+	appRelayMeter.numOverServicedComputeUnits += appRelayMeter.service.ComputeUnitsPerRelay
 
-	// Log the over-servicing warning.
-	if shouldLog {
+	// Exponential backoff, only log over-servicing when numOverServicedRelays is a power of 2
+	if shouldLogOverServicing(appRelayMeter.numOverServicedRelays) {
 		rmtr.logger.Warn().Msgf(
 			"overservicing enabled, application %q over-serviced %s",
 			appRelayMeter.app.GetAddress(),
@@ -230,11 +231,7 @@ func (rmtr *ProxyRelayMeter) SetNonApplicableRelayReward(ctx context.Context, re
 		sessionRelayMeter.serviceRelayDifficulty,
 	)
 	if err != nil {
-		return err
-	}
-
-	if sessionRelayMeter.numOverServicedRelays > 0 {
-		return nil
+		return ErrRelayerProxyUnclaimRelayPrice.Wrapf("%s", err)
 	}
 
 	// Decrease the consumed stake amount by relay cost.
@@ -437,3 +434,9 @@ func getAppStakePortionPayableToSessionSupplier(
 
 	return appStakePerSessionSupplierCoin
 }
+
+// shouldLogOverServicing returns true if the number of occurrences is a power of 2.
+// This is used to log the over-servicing warning with an exponential backoff.
+func shouldLogOverServicing(occurrence uint64) bool {
+	return (occurrence & (occurrence - 1)) == 0
+}

x/shared/types/session.go

@@ -176,3 +176,13 @@ func GetSettlementSessionEndHeight(sharedParams *Params, queryHeight int64) int6
 	return GetSessionEndToProofWindowCloseBlocks(sharedParams) +
 		GetSessionEndHeight(sharedParams, queryHeight) + 1
 }
+
+// GetNumPendingSessions returns the number of pending sessions (i.e. that have not
+// yet been settled) at the time of queryHeight.
+func GetNumPendingSessions(sharedParams *Params) int64 {
+	numBlocksPerSession := int64(sharedParams.GetNumBlocksPerSession())
+	pendingBlocks := GetSessionEndToProofWindowCloseBlocks(sharedParams)
+	// numBlocksPerSession - 1 is added to round up the integer division so that pending
+	// sessions are all the sessions that have their end height at least `pendingBlocks` old
+	return (pendingBlocks + numBlocksPerSession - 1) / numBlocksPerSession
+}

x/tokenomics/keeper/settle_pending_claims.go

@@ -203,6 +203,8 @@ func (k Keeper) SettlePendingClaims(ctx sdk.Context) (
 		appAddress := claim.GetSessionHeader().GetApplicationAddress()
 		applicationInitialStake := applicationInitialStakeMap[appAddress]
 
+		// TODO_MAINNET(@red-0ne): Add tests to ensure that a zero application stake
+		// is handled correctly.
 		if applicationInitialStake.IsZero() {
 			logger.Error(fmt.Sprintf("application %q has a zero initial stake", appAddress))
 

x/tokenomics/keeper/token_logic_modules.go

@@ -292,13 +292,17 @@ func (k Keeper) ProcessTokenLogicModules(
 
 	// Ensure the claim amount is within the limits set by Relay Mining.
 	// If not, update the settlement amount and emit relevant events.
+	// TODO_MAINNET(@red-0ne): Consider pulling this out of Keeper#ProcessTokenLogicModules
+	// and ensure claim amount limits are enforced before TLM processing.
 	actualSettlementCoin, err := k.ensureClaimAmountLimits(ctx, logger, &sharedParams, &application, &supplier, claimSettlementCoin, applicationInitialStake)
 	if err != nil {
 		return err
 	}
 	logger = logger.With("actual_settlement_upokt", actualSettlementCoin)
 	logger.Info(fmt.Sprintf("About to start processing TLMs for (%d) compute units, equal to (%s) claimed", numClaimComputeUnits, actualSettlementCoin))
 
+	// TODO_MAINNET(@red-0ne): Add tests to ensure that a zero settlement coin
+	// due to integer division rounding is handled correctly.
 	if actualSettlementCoin.Amount.IsZero() {
 		logger.Warn(fmt.Sprintf(
 			"actual settlement coin is zero, skipping TLM processing, application %q stake %s",
@@ -724,22 +728,19 @@ func (k Keeper) ensureClaimAmountLimits(
 	minRequiredAppStakeAmt := claimSettlementCoin.Amount.Add(globalInflationAmt)
 	totalClaimedCoin := sdk.NewCoin(volatile.DenomuPOKT, minRequiredAppStakeAmt)
 
-	numBlocksPerSession := int64(sharedParams.GetNumBlocksPerSession())
-	pendingBlocks := sharedtypes.GetSessionEndToProofWindowCloseBlocks(sharedParams)
-	// We are addding numBlocksPerSession - 1 to round up the integer division
-	// so that pending sessions are all the sessions that have their end height at least
-	// `pendingBlocks` old
-	pendingSessions := (pendingBlocks + numBlocksPerSession - 1) / numBlocksPerSession
+	// get the number of pending sessions that share the application stake at claim time
+	// This is used to calculate the maximum claimable amount for the supplier within a session.
+	numPendingSessions := sharedtypes.GetNumPendingSessions(sharedParams)
 
 	// The maximum any single supplier can claim is a fraction of the app's total stake
 	// divided by the number of suppliers per session.
 	// Re decentralization - This ensures the app biases towards using all suppliers in a session.
 	// Re costs - This is an easy way to split the stake evenly.
-	// TODO_FUTURE: See if there's a way to let the application to pick a single (the best) supplier
-	// in a session while maintaining a simple solution to implement this.
+	// TODO_FUTURE: See if there's a way to let the application prefer (the best)
+	// supplier(s) in a session while maintaining a simple solution to implement this.
 	maxClaimableAmt := appStake.Amount.
 		Quo(math.NewInt(sessionkeeper.NumSupplierPerSession)).
-		Quo(math.NewInt(pendingSessions))
+		Quo(math.NewInt(numPendingSessions))
 	maxClaimSettlementAmt := supplierAppStakeToMaxSettlementAmount(maxClaimableAmt)
 
 	// Check if the claimable amount is capped by the max claimable amount.