feat: outbound cctx scheduling with rate limiter

changelog.md

@@ -54,6 +54,7 @@
 * [1954](https://github.com/zeta-chain/node/pull/1954) - add metric for concurrent keysigns
 * [1979](https://github.com/zeta-chain/node/pull/1979) - add script to import genesis data into an existing genesis file
 * [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
 * [2034](https://github.com/zeta-chain/node/pull/2034) - add support for zEVM message passing
 

cmd/zetae2e/stress.go

@@ -252,7 +252,7 @@ func EchoNetworkMetrics(runner *runner.E2ERunner) {
 		case <-ticker.C:
 			numTicks++
 			// Get all pending outbound transactions
-			cctxResp, err := runner.CctxClient.CctxListPending(context.Background(), &crosschaintypes.QueryListCctxPendingRequest{
+			cctxResp, err := runner.CctxClient.ListPendingCctx(context.Background(), &crosschaintypes.QueryListPendingCctxRequest{
 				ChainId: chainID.Int64(),
 			})
 			if err != nil {

docs/openapi/openapi.swagger.yaml

@@ -26582,32 +26582,6 @@ paths:
           type: string
       tags:
         - Query
-  /zeta-chain/crosschain/cctxPending:
-    get:
-      summary: Queries a list of pending cctxs.
-      operationId: Query_CctxListPending
-      responses:
-        "200":
-          description: A successful response.
-          schema:
-            $ref: '#/definitions/crosschainQueryListCctxPendingResponse'
-        default:
-          description: An unexpected error response.
-          schema:
-            $ref: '#/definitions/googlerpcStatus'
-      parameters:
-        - name: chain_id
-          in: query
-          required: false
-          type: string
-          format: int64
-        - name: limit
-          in: query
-          required: false
-          type: integer
-          format: int64
-      tags:
-        - Query
   /zeta-chain/crosschain/convertGasToZeta:
     get:
       operationId: Query_ConvertGasToZeta
@@ -27173,6 +27147,53 @@ paths:
           type: boolean
       tags:
         - Query
+  /zeta-chain/crosschain/pendingCctx:
+    get:
+      summary: Queries a list of pending cctxs.
+      operationId: Query_ListPendingCctx
+      responses:
+        "200":
+          description: A successful response.
+          schema:
+            $ref: '#/definitions/crosschainQueryListPendingCctxResponse'
+        default:
+          description: An unexpected error response.
+          schema:
+            $ref: '#/definitions/googlerpcStatus'
+      parameters:
+        - name: chain_id
+          in: query
+          required: false
+          type: string
+          format: int64
+        - name: limit
+          in: query
+          required: false
+          type: integer
+          format: int64
+      tags:
+        - Query
+  /zeta-chain/crosschain/pendingCctxWithinRateLimit:
+    get:
+      summary: Queries a list of pending cctxs within rate limit.
+      operationId: Query_ListPendingCctxWithinRateLimit
+      responses:
+        "200":
+          description: A successful response.
+          schema:
+            $ref: '#/definitions/crosschainQueryListPendingCctxWithinRateLimitResponse'
+        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/protocolFee:
     get:
       operationId: Query_ProtocolFee
@@ -53970,7 +53991,7 @@ definitions:
       Height:
         type: string
         format: int64
-  crosschainQueryListCctxPendingResponse:
+  crosschainQueryListPendingCctxResponse:
     type: object
     properties:
       CrossChainTx:
@@ -53981,6 +54002,19 @@ definitions:
       totalPending:
         type: string
         format: uint64
+  crosschainQueryListPendingCctxWithinRateLimitResponse:
+    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

@@ -95,8 +95,13 @@ service Query {
   }
 
   // Queries a list of pending cctxs.
-  rpc CctxListPending(QueryListCctxPendingRequest) returns (QueryListCctxPendingResponse) {
-    option (google.api.http).get = "/zeta-chain/crosschain/cctxPending";
+  rpc ListPendingCctx(QueryListPendingCctxRequest) returns (QueryListPendingCctxResponse) {
+    option (google.api.http).get = "/zeta-chain/crosschain/pendingCctx";
+  }
+
+  // Queries a list of pending cctxs within rate limit.
+  rpc ListPendingCctxWithinRateLimit(QueryListPendingCctxWithinRateLimitRequest) returns (QueryListPendingCctxWithinRateLimitResponse) {
+    option (google.api.http).get = "/zeta-chain/crosschain/pendingCctxWithinRateLimit";
   }
 
   rpc ZetaAccounting(QueryZetaAccountingRequest) returns (QueryZetaAccountingResponse) {
@@ -248,16 +253,26 @@ message QueryAllCctxResponse {
   cosmos.base.query.v1beta1.PageResponse pagination = 2;
 }
 
-message QueryListCctxPendingRequest {
+message QueryListPendingCctxRequest {
   int64 chain_id = 1;
   uint32 limit = 2;
 }
 
-message QueryListCctxPendingResponse {
+message QueryListPendingCctxResponse {
   repeated CrossChainTx CrossChainTx = 1;
   uint64 totalPending = 2;
 }
 
+message QueryListPendingCctxWithinRateLimitRequest {
+  uint32 limit = 1;
+}
+
+message QueryListPendingCctxWithinRateLimitResponse {
+  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

@@ -814,9 +814,9 @@ export declare class QueryAllCctxResponse extends Message<QueryAllCctxResponse>
 }
 
 /**
- * @generated from message zetachain.zetacore.crosschain.QueryListCctxPendingRequest
+ * @generated from message zetachain.zetacore.crosschain.QueryListPendingCctxRequest
  */
-export declare class QueryListCctxPendingRequest extends Message<QueryListCctxPendingRequest> {
+export declare class QueryListPendingCctxRequest extends Message<QueryListPendingCctxRequest> {
   /**
    * @generated from field: int64 chain_id = 1;
    */
@@ -827,25 +827,25 @@ export declare class QueryListCctxPendingRequest extends Message<QueryListCctxPe
    */
   limit: number;
 
-  constructor(data?: PartialMessage<QueryListCctxPendingRequest>);
+  constructor(data?: PartialMessage<QueryListPendingCctxRequest>);
 
   static readonly runtime: typeof proto3;
-  static readonly typeName = "zetachain.zetacore.crosschain.QueryListCctxPendingRequest";
+  static readonly typeName = "zetachain.zetacore.crosschain.QueryListPendingCctxRequest";
   static readonly fields: FieldList;
 
-  static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): QueryListCctxPendingRequest;
+  static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): QueryListPendingCctxRequest;
 
-  static fromJson(jsonValue: JsonValue, options?: Partial<JsonReadOptions>): QueryListCctxPendingRequest;
+  static fromJson(jsonValue: JsonValue, options?: Partial<JsonReadOptions>): QueryListPendingCctxRequest;
 
-  static fromJsonString(jsonString: string, options?: Partial<JsonReadOptions>): QueryListCctxPendingRequest;
+  static fromJsonString(jsonString: string, options?: Partial<JsonReadOptions>): QueryListPendingCctxRequest;
 
-  static equals(a: QueryListCctxPendingRequest | PlainMessage<QueryListCctxPendingRequest> | undefined, b: QueryListCctxPendingRequest | PlainMessage<QueryListCctxPendingRequest> | undefined): boolean;
+  static equals(a: QueryListPendingCctxRequest | PlainMessage<QueryListPendingCctxRequest> | undefined, b: QueryListPendingCctxRequest | PlainMessage<QueryListPendingCctxRequest> | undefined): boolean;
 }
 
 /**
- * @generated from message zetachain.zetacore.crosschain.QueryListCctxPendingResponse
+ * @generated from message zetachain.zetacore.crosschain.QueryListPendingCctxResponse
  */
-export declare class QueryListCctxPendingResponse extends Message<QueryListCctxPendingResponse> {
+export declare class QueryListPendingCctxResponse extends Message<QueryListPendingCctxResponse> {
   /**
    * @generated from field: repeated zetachain.zetacore.crosschain.CrossChainTx CrossChainTx = 1;
    */
@@ -856,19 +856,77 @@ export declare class QueryListCctxPendingResponse extends Message<QueryListCctxP
    */
   totalPending: bigint;
 
-  constructor(data?: PartialMessage<QueryListCctxPendingResponse>);
+  constructor(data?: PartialMessage<QueryListPendingCctxResponse>);
 
   static readonly runtime: typeof proto3;
-  static readonly typeName = "zetachain.zetacore.crosschain.QueryListCctxPendingResponse";
+  static readonly typeName = "zetachain.zetacore.crosschain.QueryListPendingCctxResponse";
   static readonly fields: FieldList;
 
-  static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): QueryListCctxPendingResponse;
+  static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): QueryListPendingCctxResponse;
 
-  static fromJson(jsonValue: JsonValue, options?: Partial<JsonReadOptions>): QueryListCctxPendingResponse;
+  static fromJson(jsonValue: JsonValue, options?: Partial<JsonReadOptions>): QueryListPendingCctxResponse;
 
-  static fromJsonString(jsonString: string, options?: Partial<JsonReadOptions>): QueryListCctxPendingResponse;
+  static fromJsonString(jsonString: string, options?: Partial<JsonReadOptions>): QueryListPendingCctxResponse;
 
-  static equals(a: QueryListCctxPendingResponse | PlainMessage<QueryListCctxPendingResponse> | undefined, b: QueryListCctxPendingResponse | PlainMessage<QueryListCctxPendingResponse> | undefined): boolean;
+  static equals(a: QueryListPendingCctxResponse | PlainMessage<QueryListPendingCctxResponse> | undefined, b: QueryListPendingCctxResponse | PlainMessage<QueryListPendingCctxResponse> | undefined): boolean;
+}
+
+/**
+ * @generated from message zetachain.zetacore.crosschain.QueryListPendingCctxWithinRateLimitRequest
+ */
+export declare class QueryListPendingCctxWithinRateLimitRequest extends Message<QueryListPendingCctxWithinRateLimitRequest> {
+  /**
+   * @generated from field: uint32 limit = 1;
+   */
+  limit: number;
+
+  constructor(data?: PartialMessage<QueryListPendingCctxWithinRateLimitRequest>);
+
+  static readonly runtime: typeof proto3;
+  static readonly typeName = "zetachain.zetacore.crosschain.QueryListPendingCctxWithinRateLimitRequest";
+  static readonly fields: FieldList;
+
+  static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): QueryListPendingCctxWithinRateLimitRequest;
+
+  static fromJson(jsonValue: JsonValue, options?: Partial<JsonReadOptions>): QueryListPendingCctxWithinRateLimitRequest;
+
+  static fromJsonString(jsonString: string, options?: Partial<JsonReadOptions>): QueryListPendingCctxWithinRateLimitRequest;
+
+  static equals(a: QueryListPendingCctxWithinRateLimitRequest | PlainMessage<QueryListPendingCctxWithinRateLimitRequest> | undefined, b: QueryListPendingCctxWithinRateLimitRequest | PlainMessage<QueryListPendingCctxWithinRateLimitRequest> | undefined): boolean;
+}
+
+/**
+ * @generated from message zetachain.zetacore.crosschain.QueryListPendingCctxWithinRateLimitResponse
+ */
+export declare class QueryListPendingCctxWithinRateLimitResponse extends Message<QueryListPendingCctxWithinRateLimitResponse> {
+  /**
+   * @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<QueryListPendingCctxWithinRateLimitResponse>);
+
+  static readonly runtime: typeof proto3;
+  static readonly typeName = "zetachain.zetacore.crosschain.QueryListPendingCctxWithinRateLimitResponse";
+  static readonly fields: FieldList;
+
+  static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): QueryListPendingCctxWithinRateLimitResponse;
+
+  static fromJson(jsonValue: JsonValue, options?: Partial<JsonReadOptions>): QueryListPendingCctxWithinRateLimitResponse;
+
+  static fromJsonString(jsonString: string, options?: Partial<JsonReadOptions>): QueryListPendingCctxWithinRateLimitResponse;
+
+  static equals(a: QueryListPendingCctxWithinRateLimitResponse | PlainMessage<QueryListPendingCctxWithinRateLimitResponse> | undefined, b: QueryListPendingCctxWithinRateLimitResponse | PlainMessage<QueryListPendingCctxWithinRateLimitResponse> | undefined): boolean;
 }
 
 /**

x/crosschain/client/cli/cli_cctx.go

@@ -69,13 +69,13 @@ func CmdPendingCctx() *cobra.Command {
 				return err
 			}
 
-			params := &types.QueryListCctxPendingRequest{
+			params := &types.QueryListPendingCctxRequest{
 				ChainId: chainID,
 				// #nosec G701 bit size verified
 				Limit: uint32(limit),
 			}
 
-			res, err := queryClient.CctxListPending(context.Background(), params)
+			res, err := queryClient.ListPendingCctx(context.Background(), params)
 			if err != nil {
 				return err
 			}

x/crosschain/keeper/abci.go

@@ -51,7 +51,7 @@ IterateChains:
 	for _, chain := range chains {
 		// support only external evm chains
 		if zetachains.IsEVMChain(chain.ChainId) && !zetachains.IsZetaChain(chain.ChainId) {
-			res, err := k.CctxListPending(sdk.UnwrapSDKContext(ctx), &types.QueryListCctxPendingRequest{
+			res, err := k.ListPendingCctx(sdk.UnwrapSDKContext(ctx), &types.QueryListPendingCctxRequest{
 				ChainId: chain.ChainId,
 				Limit:   gasPriceIncreaseFlags.MaxPendingCctxs,
 			})

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
 }