feat: outbound cctx scheduling with rate limiter

changelog.md

@@ -53,6 +53,7 @@
 * [1935](https://github.com/zeta-chain/node/pull/1935) - add an operational authority group
 * [1954](https://github.com/zeta-chain/node/pull/1954) - add metric for concurrent keysigns
 * [2006](https://github.com/zeta-chain/node/pull/2006) - add Amoy testnet static chain information
+* [2045](https://github.com/zeta-chain/node/pull/2046) - add grpc query with outbound rate limit for zetaclient to use
 * [2046](https://github.com/zeta-chain/node/pull/2046) - add state variable in crosschain for rate limiter flags
 
 ### Tests

docs/openapi/openapi.swagger.yaml

@@ -26608,6 +26608,27 @@ paths:
           format: int64
       tags:
         - Query
+  /zeta-chain/crosschain/cctxPendingWithRateLimit:
+    get:
+      summary: Queries a list of pending cctxs with rate limit.
+      operationId: Query_CctxListPendingWithinRateLimit
+      responses:
+        "200":
+          description: A successful response.
+          schema:
+            $ref: '#/definitions/crosschainQueryListCctxPendingWithRateLimitResponse'
+        default:
+          description: An unexpected error response.
+          schema:
+            $ref: '#/definitions/googlerpcStatus'
+      parameters:
+        - name: limit
+          in: query
+          required: false
+          type: integer
+          format: int64
+      tags:
+        - Query
   /zeta-chain/crosschain/convertGasToZeta:
     get:
       operationId: Query_ConvertGasToZeta
@@ -53981,6 +54002,19 @@ definitions:
       totalPending:
         type: string
         format: uint64
+  crosschainQueryListCctxPendingWithRateLimitResponse:
+    type: object
+    properties:
+      cross_chain_tx:
+        type: array
+        items:
+          type: object
+          $ref: '#/definitions/crosschainCrossChainTx'
+      total_pending:
+        type: string
+        format: uint64
+      rate_limit_exceeded:
+        type: boolean
   crosschainQueryMessagePassingProtocolFeeResponse:
     type: object
     properties:

proto/crosschain/query.proto

@@ -99,6 +99,11 @@ service Query {
     option (google.api.http).get = "/zeta-chain/crosschain/cctxPending";
   }
 
+  // Queries a list of pending cctxs with rate limit.
+  rpc CctxListPendingWithinRateLimit(QueryListCctxPendingWithRateLimitRequest) returns (QueryListCctxPendingWithRateLimitResponse) {
+    option (google.api.http).get = "/zeta-chain/crosschain/cctxPendingWithRateLimit";
+  }
+
   rpc ZetaAccounting(QueryZetaAccountingRequest) returns (QueryZetaAccountingResponse) {
     option (google.api.http).get = "/zeta-chain/crosschain/zetaAccounting";
   }
@@ -258,6 +263,16 @@ message QueryListCctxPendingResponse {
   uint64 totalPending = 2;
 }
 
+message QueryListCctxPendingWithRateLimitRequest {
+  uint32 limit = 2;
+}
+
+message QueryListCctxPendingWithRateLimitResponse {
+  repeated CrossChainTx cross_chain_tx = 1;
+  uint64 total_pending = 2;
+  bool rate_limit_exceeded = 3;
+}
+
 message QueryLastZetaHeightRequest {}
 
 message QueryLastZetaHeightResponse {

typescript/crosschain/query_pb.d.ts

@@ -871,6 +871,64 @@ export declare class QueryListCctxPendingResponse extends Message<QueryListCctxP
   static equals(a: QueryListCctxPendingResponse | PlainMessage<QueryListCctxPendingResponse> | undefined, b: QueryListCctxPendingResponse | PlainMessage<QueryListCctxPendingResponse> | undefined): boolean;
 }
 
+/**
+ * @generated from message zetachain.zetacore.crosschain.QueryListCctxPendingWithRateLimitRequest
+ */
+export declare class QueryListCctxPendingWithRateLimitRequest extends Message<QueryListCctxPendingWithRateLimitRequest> {
+  /**
+   * @generated from field: uint32 limit = 2;
+   */
+  limit: number;
+
+  constructor(data?: PartialMessage<QueryListCctxPendingWithRateLimitRequest>);
+
+  static readonly runtime: typeof proto3;
+  static readonly typeName = "zetachain.zetacore.crosschain.QueryListCctxPendingWithRateLimitRequest";
+  static readonly fields: FieldList;
+
+  static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): QueryListCctxPendingWithRateLimitRequest;
+
+  static fromJson(jsonValue: JsonValue, options?: Partial<JsonReadOptions>): QueryListCctxPendingWithRateLimitRequest;
+
+  static fromJsonString(jsonString: string, options?: Partial<JsonReadOptions>): QueryListCctxPendingWithRateLimitRequest;
+
+  static equals(a: QueryListCctxPendingWithRateLimitRequest | PlainMessage<QueryListCctxPendingWithRateLimitRequest> | undefined, b: QueryListCctxPendingWithRateLimitRequest | PlainMessage<QueryListCctxPendingWithRateLimitRequest> | undefined): boolean;
+}
+
+/**
+ * @generated from message zetachain.zetacore.crosschain.QueryListCctxPendingWithRateLimitResponse
+ */
+export declare class QueryListCctxPendingWithRateLimitResponse extends Message<QueryListCctxPendingWithRateLimitResponse> {
+  /**
+   * @generated from field: repeated zetachain.zetacore.crosschain.CrossChainTx cross_chain_tx = 1;
+   */
+  crossChainTx: CrossChainTx[];
+
+  /**
+   * @generated from field: uint64 total_pending = 2;
+   */
+  totalPending: bigint;
+
+  /**
+   * @generated from field: bool rate_limit_exceeded = 3;
+   */
+  rateLimitExceeded: boolean;
+
+  constructor(data?: PartialMessage<QueryListCctxPendingWithRateLimitResponse>);
+
+  static readonly runtime: typeof proto3;
+  static readonly typeName = "zetachain.zetacore.crosschain.QueryListCctxPendingWithRateLimitResponse";
+  static readonly fields: FieldList;
+
+  static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): QueryListCctxPendingWithRateLimitResponse;
+
+  static fromJson(jsonValue: JsonValue, options?: Partial<JsonReadOptions>): QueryListCctxPendingWithRateLimitResponse;
+
+  static fromJsonString(jsonString: string, options?: Partial<JsonReadOptions>): QueryListCctxPendingWithRateLimitResponse;
+
+  static equals(a: QueryListCctxPendingWithRateLimitResponse | PlainMessage<QueryListCctxPendingWithRateLimitResponse> | undefined, b: QueryListCctxPendingWithRateLimitResponse | PlainMessage<QueryListCctxPendingWithRateLimitResponse> | undefined): boolean;
+}
+
 /**
  * @generated from message zetachain.zetacore.crosschain.QueryLastZetaHeightRequest
  */

x/crosschain/keeper/cctx_utils.go

@@ -86,7 +86,7 @@ func (k Keeper) GetRevertGasLimit(ctx sdk.Context, cctx types.CrossChainTx) (uin
 	return 0, nil
 }
 
-func IsPending(cctx types.CrossChainTx) bool {
+func IsPending(cctx *types.CrossChainTx) bool {
 	// pending inbound is not considered a "pending" state because it has not reached consensus yet
 	return cctx.CctxStatus.Status == types.CctxStatus_PendingOutbound || cctx.CctxStatus.Status == types.CctxStatus_PendingRevert
 }

x/crosschain/keeper/cctx_utils_test.go

@@ -215,7 +215,7 @@ func Test_IsPending(t *testing.T) {
 	}
 	for _, tc := range tt {
 		t.Run(fmt.Sprintf("status %s", tc.status), func(t *testing.T) {
-			require.Equal(t, tc.expected, crosschainkeeper.IsPending(types.CrossChainTx{CctxStatus: &types.Status{Status: tc.status}}))
+			require.Equal(t, tc.expected, crosschainkeeper.IsPending(&types.CrossChainTx{CctxStatus: &types.Status{Status: tc.status}}))
 		})
 	}
 }

x/crosschain/keeper/foreign_coins.go

@@ -5,11 +5,11 @@ import (
 	fungibleModuleTypes "github.com/zeta-chain/zetacore/x/fungible/types"
 )
 
-func (k Keeper) GetAllForeignCoins(ctx sdk.Context) ([]fungibleModuleTypes.ForeignCoins, error) {
+func (k Keeper) GetAllForeignCoins(ctx sdk.Context) []fungibleModuleTypes.ForeignCoins {
 	chains := k.zetaObserverKeeper.GetSupportedChains(ctx)
 	var fCoins []fungibleModuleTypes.ForeignCoins
 	for _, chain := range chains {
 		fCoins = append(fCoins, k.fungibleKeeper.GetAllForeignCoinsForChain(ctx, chain.ChainId)...)
 	}
-	return fCoins, nil
+	return fCoins
 }

x/crosschain/keeper/foreign_coins_test.go

@@ -14,7 +14,6 @@ func TestKeeper_GetAllForeignCoins(t *testing.T) {
 	fc.ForeignChainId = 101
 	k.GetFungibleKeeper().SetForeignCoins(ctx, fc)
 
-	res, err := k.GetAllForeignCoins(ctx)
-	require.NoError(t, err)
+	res := k.GetAllForeignCoins(ctx)
 	require.Equal(t, 1, len(res))
 }

x/crosschain/keeper/grpc_query_cctx.go

@@ -8,6 +8,7 @@
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/cosmos/cosmos-sdk/types/query"
 	"github.com/zeta-chain/zetacore/x/crosschain/types"
+	observertypes "github.com/zeta-chain/zetacore/x/observer/types"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -78,42 +79,32 @@
 		return nil, status.Error(codes.Internal, "tss not found")
 	}
 	// #nosec G701 always in range
-	res, found := k.GetObserverKeeper().GetNonceToCctx(ctx, tss.TssPubkey, req.ChainID, int64(req.Nonce))
-	if !found {
-		return nil, status.Error(codes.Internal, fmt.Sprintf("nonceToCctx not found: nonce %d, chainid %d", req.Nonce, req.ChainID))
-	}
-	val, found := k.GetCrossChainTx(ctx, res.CctxIndex)
-	if !found {
-		return nil, status.Error(codes.Internal, fmt.Sprintf("cctx not found: index %s", res.CctxIndex))
+	cctx, err := getCctxByChainIDAndNonce(k, ctx, tss.TssPubkey, req.ChainID, int64(req.Nonce))
+	if err != nil {
+		return nil, err
 	}
 
-	return &types.QueryGetCctxResponse{CrossChainTx: &val}, nil
+	return &types.QueryGetCctxResponse{CrossChainTx: cctx}, nil
 }
 
 // CctxListPending returns a list of pending cctxs and the total number of pending cctxs
-// a limit for the number of cctxs to return can be specified
-// if no limit is specified, the default is MaxPendingCctxs
+// a limit for the number of cctxs to return can be specified or the default is MaxPendingCctxs
 func (k Keeper) CctxListPending(c context.Context, req *types.QueryListCctxPendingRequest) (*types.QueryListCctxPendingResponse, error) {
 	if req == nil {
 		return nil, status.Error(codes.InvalidArgument, "invalid request")
 	}
 
-	// check limit
-	// if no limit specified, default to MaxPendingCctxs
-	if req.Limit > MaxPendingCctxs {
-		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("limit exceeds max limit of %d", MaxPendingCctxs))
-	}
+	// use default MaxPendingCctxs if not specified or too high
 	limit := req.Limit
-	if limit == 0 {
+	if limit == 0 || limit > MaxPendingCctxs {
 		limit = MaxPendingCctxs
 	}
-
 	ctx := sdk.UnwrapSDKContext(c)
 
 	// query the nonces that are pending
 	tss, found := k.zetaObserverKeeper.GetTSS(ctx)
 	if !found {
-		return nil, status.Error(codes.Internal, "tss not found")
+		return nil, observertypes.ErrTssNotFound
 	}
 	pendingNonces, found := k.GetObserverKeeper().GetPendingNonces(ctx, tss.TssPubkey, req.ChainId)
 	if !found {
@@ -136,21 +127,16 @@
 		startNonce = 0
 	}
 	for i := startNonce; i < pendingNonces.NonceLow; i++ {
-		nonceToCctx, found := k.GetObserverKeeper().GetNonceToCctx(ctx, tss.TssPubkey, req.ChainId, i)
-		if !found {
-			return nil, status.Error(codes.Internal, fmt.Sprintf("nonceToCctx not found: nonce %d, chainid %d", i, req.ChainId))
-		}
-		cctx, found := k.GetCrossChainTx(ctx, nonceToCctx.CctxIndex)
-		if !found {
-			return nil, status.Error(codes.Internal, fmt.Sprintf("cctx not found: index %s", nonceToCctx.CctxIndex))
+		cctx, err := getCctxByChainIDAndNonce(k, ctx, tss.TssPubkey, req.ChainId, i)
+		if err != nil {
+			return nil, err
 		}
-		if cctx.CctxStatus.Status == types.CctxStatus_PendingOutbound || cctx.CctxStatus.Status == types.CctxStatus_PendingRevert {
-			totalPending++
 
-			// we check here if max cctxs is reached because we want to return the total pending cctxs
-			// even if we have reached the limit
+		// only take a `limit` number of pending cctxs as result but still count the total pending cctxs
+		if IsPending(cctx) {
+			totalPending++
 			if !maxCCTXsReached() {
-				cctxs = append(cctxs, &cctx)
+				cctxs = append(cctxs, cctx)
 			}
 		}
 	}
@@ -161,19 +147,28 @@
 
 	// now query the pending nonces that we know are pending
 	for i := pendingNonces.NonceLow; i < pendingNonces.NonceHigh && !maxCCTXsReached(); i++ {
-		nonceToCctx, found := k.GetObserverKeeper().GetNonceToCctx(ctx, tss.TssPubkey, req.ChainId, i)
-		if !found {
-			return nil, status.Error(codes.Internal, "nonceToCctx not found")
-		}
-		cctx, found := k.GetCrossChainTx(ctx, nonceToCctx.CctxIndex)
-		if !found {
-			return nil, status.Error(codes.Internal, "cctxIndex not found")
+		cctx, err := getCctxByChainIDAndNonce(k, ctx, tss.TssPubkey, req.ChainId, i)
+		if err != nil {
+			return nil, err
 		}
-		cctxs = append(cctxs, &cctx)
+		cctxs = append(cctxs, cctx)
 	}
 
 	return &types.QueryListCctxPendingResponse{
 		CrossChainTx: cctxs,
 		TotalPending: totalPending,
 	}, nil
 }
+
+// getCctxByChainIDAndNonce returns the cctx by chainID and nonce
+func getCctxByChainIDAndNonce(k Keeper, ctx sdk.Context, tssPubkey string, chainID int64, nonce int64) (*types.CrossChainTx, error) {
+	nonceToCctx, found := k.GetObserverKeeper().GetNonceToCctx(ctx, tssPubkey, chainID, nonce)
+	if !found {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("nonceToCctx not found: chainid %d, nonce %d", chainID, nonce))
+	}
+	cctx, found := k.GetCrossChainTx(ctx, nonceToCctx.CctxIndex)
+	if !found {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("cctx not found: index %s", nonceToCctx.CctxIndex))
+	}
+	return &cctx, nil
+}