test: fix and unit tests for query pending cctx within rate limit

changelog.md

@@ -82,6 +82,7 @@
 * [1985](https://github.com/zeta-chain/node/pull/1985) - improve fungible module coverage
 * [1992](https://github.com/zeta-chain/node/pull/1992) - remove setupKeeper from crosschain module
 * [2008](https://github.com/zeta-chain/node/pull/2008) - add test for connector bytecode update
+* [2060](https://github.com/zeta-chain/node/pull/2060) - add unit test for rate limiter query
 
 ### Fixes
 

docs/openapi/openapi.swagger.yaml

@@ -54013,6 +54013,9 @@ definitions:
       total_pending:
         type: string
         format: uint64
+      value_within_window:
+        type: string
+        format: uint64
       rate_limit_exceeded:
         type: boolean
   crosschainQueryMessagePassingProtocolFeeResponse:

proto/crosschain/query.proto

@@ -270,7 +270,8 @@ message QueryListPendingCctxWithinRateLimitRequest {
 message QueryListPendingCctxWithinRateLimitResponse {
   repeated CrossChainTx cross_chain_tx = 1;
   uint64 total_pending = 2;
-  bool rate_limit_exceeded = 3;
+  uint64 value_within_window = 3;
+  bool rate_limit_exceeded = 4;
 }
 
 message QueryLastZetaHeightRequest {}

typescript/crosschain/query_pb.d.ts

@@ -910,7 +910,12 @@ export declare class QueryListPendingCctxWithinRateLimitResponse extends Message
   totalPending: bigint;
 
   /**
-   * @generated from field: bool rate_limit_exceeded = 3;
+   * @generated from field: uint64 value_within_window = 3;
+   */
+  valueWithinWindow: bigint;
+
+  /**
+   * @generated from field: bool rate_limit_exceeded = 4;
    */
   rateLimitExceeded: boolean;
 

x/crosschain/keeper/grpc_query_cctx_rate_limit.go

@@ -2,6 +2,7 @@
 
 import (
 	"context"
+	"sort"
 	"strings"
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
@@ -68,20 +69,22 @@
 	}
 
 	// calculate the rate limiter sliding window left boundary (inclusive)
-	leftWindowBoundary := height - rateLimitFlags.Window
+	leftWindowBoundary := height - rateLimitFlags.Window + 1
 	if leftWindowBoundary < 0 {
 		leftWindowBoundary = 0
 	}
 
 	// get the conversion rates for all foreign coins
 	var gasCoinRates map[int64]sdk.Dec
 	var erc20CoinRates map[int64]map[string]sdk.Dec
-	var erc20Coins map[int64]map[string]fungibletypes.ForeignCoins
-	var rateLimitInZeta sdk.Dec
+	var foreignCoinMap map[int64]map[string]fungibletypes.ForeignCoins
+	var windowLimitInZeta sdk.Dec
+	var blockLimitInZeta sdk.Dec
 	if applyLimit {
 		gasCoinRates, erc20CoinRates = k.GetRateLimiterRates(ctx)
-		erc20Coins = k.fungibleKeeper.GetAllForeignERC20CoinMap(ctx)
-		rateLimitInZeta = sdk.NewDecFromBigInt(rateLimitFlags.Rate.BigInt())
+		foreignCoinMap = k.fungibleKeeper.GetAllForeignCoinMap(ctx)
+		windowLimitInZeta = sdk.NewDecFromBigInt(rateLimitFlags.Rate.BigInt())
+		blockLimitInZeta = windowLimitInZeta.Quo(sdk.NewDec(rateLimitFlags.Window))
 	}
 
 	// the criteria to stop adding cctxs to the rpc response
@@ -90,15 +93,46 @@
 		return uint32(len(cctxs)) >= limit
 	}
 
-	// query pending nonces for each foreign chain
+	// if a cctx falls within the rate limiter window
+	isCctxInWindow := func(cctx *types.CrossChainTx) bool {
+		// #nosec G701 checked positive
+		return cctx.InboundTxParams.InboundTxObservedExternalHeight >= uint64(leftWindowBoundary)
+	}
+
+	// query pending nonces for each foreign chain and get the lowest height of the pending cctxs
 	// Note: The pending nonces could change during the RPC call, so query them beforehand
+	lowestPendingCctxHeight := int64(0)
 	pendingNoncesMap := make(map[int64]*observertypes.PendingNonces)
 	for _, chain := range chains {
 		pendingNonces, found := k.GetObserverKeeper().GetPendingNonces(ctx, tss.TssPubkey, chain.ChainId)
 		if !found {
 			return nil, status.Error(codes.Internal, "pending nonces not found")
 		}
-		pendingNoncesMap[chain.ChainId] = &pendingNonces
+
+		// insert pending nonces and update lowest height
+		if pendingNonces.NonceLow < pendingNonces.NonceHigh {
+			pendingNoncesMap[chain.ChainId] = &pendingNonces
+			cctx, err := getCctxByChainIDAndNonce(k, ctx, tss.TssPubkey, chain.ChainId, pendingNonces.NonceLow)
+			if err != nil {
+				return nil, err
+			}
+			// #nosec G701 len always in range
+			cctxHeight := int64(cctx.InboundTxParams.InboundTxObservedExternalHeight)
+			if lowestPendingCctxHeight == 0 || cctxHeight < lowestPendingCctxHeight {
+				lowestPendingCctxHeight = cctxHeight
+			}
+		}
+	}
+
+	// invariant: for period of time > window, the average rate per block cannot exceed `blockLimitInZeta`
+	pendingCctxsLimitInZeta := windowLimitInZeta
+	if lowestPendingCctxHeight != 0 {
+		// `pendingCctxWindow` is the width of [lowestPendingCctxHeight, height] window
+		// if the window can be wider than the rate limit window, we should adjust the total limit proportionally
+		pendingCctxWindow := height - lowestPendingCctxHeight + 1
+		if pendingCctxWindow > rateLimitFlags.Window {
+			pendingCctxsLimitInZeta = blockLimitInZeta.Mul(sdk.NewDec(pendingCctxWindow))
+		}
 	}
 
 	// query backwards for potential missed pending cctxs for each foreign chain
@@ -114,23 +148,29 @@
 			endNonce = 0
 		}
 
+		// add the pending nonces to the total pending
+		// Note: the `totalPending` may not be accurate only if the rate limiter triggers early exit
+		// `totalPending` is now used for metrics only and it's okay to trade off accuracy for performance
+		// #nosec G701 always in range
+		totalPending += uint64(pendingNonces.NonceHigh - pendingNonces.NonceLow)
+
 		// query cctx by nonce backwards to the left boundary of the rate limit sliding window
 		for nonce := startNonce; nonce >= 0; nonce-- {
 			cctx, err := getCctxByChainIDAndNonce(k, ctx, tss.TssPubkey, chain.ChainId, nonce)
 			if err != nil {
 				return nil, err
 			}
+			inWindow := isCctxInWindow(cctx)
 
 			// We should at least go backwards by 1000 nonces to pick up missed pending cctxs
 			// We might go even further back if rate limiter is enabled and the endNonce hasn't hit the left window boundary yet
 			// There are two criteria to stop scanning backwards:
 			// criteria #1: we'll stop at the left window boundary if the `endNonce` hasn't hit it yet
-			// #nosec G701 always positive
-			if nonce < endNonce && cctx.InboundTxParams.InboundTxObservedExternalHeight < uint64(leftWindowBoundary) {
+			if nonce < endNonce && !inWindow {
 				break
 			}
 			// criteria #2: we should finish the RPC call if the rate limit is exceeded
-			if rateLimitExceeded(chain.ChainId, cctx, gasCoinRates, erc20CoinRates, erc20Coins, &totalCctxValueInZeta, rateLimitInZeta) {
+			if inWindow && rateLimitExceeded(chain.ChainId, cctx, gasCoinRates, erc20CoinRates, foreignCoinMap, &totalCctxValueInZeta, windowLimitInZeta) {
 				limitExceeded = true
 				break LoopBackwards
 			}
@@ -143,12 +183,6 @@
 				}
 			}
 		}
-
-		// add the pending nonces to the total pending
-		// Note: the `totalPending` may not be accurate only if the rate limiter triggers early exit
-		// `totalPending` is now used for metrics only and it's okay to trade off accuracy for performance
-		// #nosec G701 always in range
-		totalPending += uint64(pendingNonces.NonceHigh - pendingNonces.NonceLow)
 	}
 
 	// query forwards for pending cctxs for each foreign chain
@@ -167,54 +201,50 @@
 				break LoopForwards
 			}
 			// criteria #2: we should finish the RPC call if the rate limit is exceeded
-			if rateLimitExceeded(chain.ChainId, cctx, gasCoinRates, erc20CoinRates, erc20Coins, &totalCctxValueInZeta, rateLimitInZeta) {
+			if rateLimitExceeded(chain.ChainId, cctx, gasCoinRates, erc20CoinRates, foreignCoinMap, &totalCctxValueInZeta, pendingCctxsLimitInZeta) {
 				limitExceeded = true
 				break LoopForwards
 			}
 			cctxs = append(cctxs, cctx)
 		}
 	}
 
+	// sort the cctxs by chain ID and nonce (lower nonce holds higher priority for scheduling)
+	sort.Slice(cctxs, func(i, j int) bool {
+		if cctxs[i].GetCurrentOutTxParam().ReceiverChainId == cctxs[j].GetCurrentOutTxParam().ReceiverChainId {
+			return cctxs[i].GetCurrentOutTxParam().OutboundTxTssNonce < cctxs[j].GetCurrentOutTxParam().OutboundTxTssNonce
+		}
+		return cctxs[i].GetCurrentOutTxParam().ReceiverChainId < cctxs[j].GetCurrentOutTxParam().ReceiverChainId
+	})
+
 	return &types.QueryListPendingCctxWithinRateLimitResponse{
 		CrossChainTx:      cctxs,
 		TotalPending:      totalPending,
+		ValueWithinWindow: totalCctxValueInZeta.TruncateInt().Uint64(),
 		RateLimitExceeded: limitExceeded,
 	}, nil
 }
 
-// convertCctxValue converts the value of the cctx in ZETA using given conversion rates
-func convertCctxValue(
+// ConvertCctxValue converts the value of the cctx in ZETA using given conversion rates
+func ConvertCctxValue(
 	chainID int64,
 	cctx *types.CrossChainTx,
 	gasCoinRates map[int64]sdk.Dec,
 	erc20CoinRates map[int64]map[string]sdk.Dec,
-	erc20Coins map[int64]map[string]fungibletypes.ForeignCoins,
+	foreignCoinMap map[int64]map[string]fungibletypes.ForeignCoins,
 ) sdk.Dec {
 	var rate sdk.Dec
 	var decimals uint64
 	switch cctx.InboundTxParams.CoinType {
 	case coin.CoinType_Zeta:
 		// no conversion needed for ZETA
-		rate = sdk.NewDec(1)
+		amountCctx := sdk.NewDecFromBigInt(cctx.GetCurrentOutTxParam().Amount.BigInt())
+		return amountCctx.Quo(sdk.NewDec(10).Power(18))
 	case coin.CoinType_Gas:
 		rate = gasCoinRates[chainID]
 	case coin.CoinType_ERC20:
-		// get the ERC20 coin decimals
-		_, found := erc20Coins[chainID]
-		if !found {
-			// skip if no coin found for this chainID
-			return sdk.NewDec(0)
-		}
-		fCoin, found := erc20Coins[chainID][strings.ToLower(cctx.InboundTxParams.Asset)]
-		if !found {
-			// skip if no coin found for this Asset
-			return sdk.NewDec(0)
-		}
-		// #nosec G701 always in range
-		decimals = uint64(fCoin.Decimals)
-
 		// get the ERC20 coin rate
-		_, found = erc20CoinRates[chainID]
+		_, found := erc20CoinRates[chainID]
 		if !found {
 			// skip if no rate found for this chainID
 			return sdk.NewDec(0)
@@ -224,21 +254,35 @@
 		// skip CoinType_Cmd
 		return sdk.NewDec(0)
 	}
-
 	// should not happen, return 0 to skip if it happens
-	if rate.LTE(sdk.NewDec(0)) {
+	if rate.IsNil() || rate.LTE(sdk.NewDec(0)) {
+		return sdk.NewDec(0)
+	}
+
+	// get foreign coin decimals
+	foreignCoinFromChainMap, found := foreignCoinMap[chainID]
+	if !found {
+		// skip if no coin found for this chainID
+		return sdk.NewDec(0)
+	}
+	foreignCoin, found := foreignCoinFromChainMap[strings.ToLower(cctx.InboundTxParams.Asset)]
+	if !found {
+		// skip if no coin found for this Asset
 		return sdk.NewDec(0)
 	}
+	decimals = uint64(foreignCoin.Decimals)
 
-	// the reciprocal of `rate` is the amount of zrc20 needed to buy 1 ZETA
-	// for example, given rate = 0.8, the reciprocal is 1.25, which means 1.25 ZRC20 can buy 1 ZETA
-	// given decimals = 6, the `oneZeta` amount will be 1.25 * 10^6 = 1250000
-	oneZrc20 := sdk.NewDec(1).Power(decimals)
-	oneZeta := oneZrc20.Quo(rate)
+	// given decimals = 6, the `oneZrc20` amount will be 10^6 = 1000000
+	oneZrc20 := sdk.NewDec(10).Power(decimals)
 
 	// convert asset amount into ZETA
+	// step 1: convert the amount into ZRC20 integer amount
+	// step 2: convert the ZRC20 integer amount into decimal amount
+	// given amountCctx = 2000000, rate = 0.8, decimals = 6
+	// the amountZrc20 = 2000000 * 0.8 = 1600000, the amountZeta = 1600000 / 1000000 = 1.6
 	amountCctx := sdk.NewDecFromBigInt(cctx.GetCurrentOutTxParam().Amount.BigInt())
-	amountZeta := amountCctx.Quo(oneZeta)
+	amountZrc20 := amountCctx.Mul(rate)
+	amountZeta := amountZrc20.Quo(oneZrc20)
 	return amountZeta
 }
 
@@ -249,11 +293,11 @@
 	cctx *types.CrossChainTx,
 	gasCoinRates map[int64]sdk.Dec,
 	erc20CoinRates map[int64]map[string]sdk.Dec,
-	erc20Coins map[int64]map[string]fungibletypes.ForeignCoins,
+	foreignCoinMap map[int64]map[string]fungibletypes.ForeignCoins,
 	currentCctxValue *sdk.Dec,
 	rateLimitValue sdk.Dec,
 ) bool {
-	amountZeta := convertCctxValue(chainID, cctx, gasCoinRates, erc20CoinRates, erc20Coins)
+	amountZeta := ConvertCctxValue(chainID, cctx, gasCoinRates, erc20CoinRates, foreignCoinMap)
 	*currentCctxValue = currentCctxValue.Add(amountZeta)
 	return currentCctxValue.GT(rateLimitValue)
 }