clean pending-forever cctxs and associated trackers

cmd/zetaclientd/start.go

@@ -237,7 +237,7 @@ func start(_ *cobra.Command, _ []string) error {
 	}
 
 	// CreateCoreObserver : Core observer wraps the zetacore bridge and adds the client and signer maps to it . This is the high level object used for CCTX interactions
-	mo1 := mc.NewCoreObserver(zetaBridge, signerMap, chainClientMap, metrics, tss, masterLogger, cfg, telemetryServer)
+	mo1 := mc.NewCoreObserver(zetaBridge, signerMap, chainClientMap, dbpath, metrics, tss, masterLogger, cfg, telemetryServer)
 	mo1.MonitorCore()
 
 	startLogger.Info().Msgf("awaiting the os.Interrupt, syscall.SIGTERM signals...")

docs/openapi/openapi.swagger.yaml

@@ -26646,6 +26646,37 @@ paths:
           type: boolean
       tags:
         - Query
+  /zeta-chain/crosschain/cctxPendingInNonceRange:
+    get:
+      summary: Queries a list of pending send items in nonce range.
+      operationId: Query_CctxAllPendingInNonceRange
+      responses:
+        "200":
+          description: A successful response.
+          schema:
+            $ref: '#/definitions/crosschainQueryAllCctxPendingInNonceRangeResponse'
+        default:
+          description: An unexpected error response.
+          schema:
+            $ref: '#/definitions/googlerpcStatus'
+      parameters:
+        - name: chain_id
+          in: query
+          required: false
+          type: string
+          format: int64
+        - name: nonce_low
+          in: query
+          required: false
+          type: string
+          format: uint64
+        - name: nonce_high
+          in: query
+          required: false
+          type: string
+          format: uint64
+      tags:
+        - Query
   /zeta-chain/crosschain/chainNonces:
     get:
       summary: Queries a list of chainNonces items.
@@ -50615,6 +50646,14 @@ definitions:
       tss:
         type: string
     title: store key is tss+chainid
+  crosschainQueryAllCctxPendingInNonceRangeResponse:
+    type: object
+    properties:
+      cross_chain_tx:
+        type: array
+        items:
+          type: object
+          $ref: '#/definitions/crosschainCrossChainTx'
   crosschainQueryAllCctxPendingResponse:
     type: object
     properties:

proto/crosschain/query.proto

@@ -127,6 +127,11 @@ service Query {
     option (google.api.http).get = "/zeta-chain/crosschain/cctxPending";
   }
 
+  // Queries a list of pending send items in nonce range.
+  rpc CctxAllPendingInNonceRange(QueryAllCctxPendingInNonceRangeRequest) returns (QueryAllCctxPendingInNonceRangeResponse) {
+    option (google.api.http).get = "/zeta-chain/crosschain/cctxPendingInNonceRange";
+  }
+
   // Queries a list of lastMetaHeight items.
   rpc LastZetaHeight(QueryLastZetaHeightRequest) returns (QueryLastZetaHeightResponse) {
     option (google.api.http).get = "/zeta-chain/crosschain/lastZetaHeight";
@@ -316,6 +321,16 @@ message QueryAllCctxPendingResponse {
   cosmos.base.query.v1beta1.PageResponse pagination = 2;
 }
 
+message QueryAllCctxPendingInNonceRangeRequest {
+  int64 chain_id = 1;
+  uint64 nonce_low = 2;
+  uint64 nonce_high = 3;
+}
+
+message QueryAllCctxPendingInNonceRangeResponse {
+  repeated CrossChainTx cross_chain_tx = 1;
+}
+
 message QueryLastZetaHeightRequest {}
 
 message QueryLastZetaHeightResponse {

x/crosschain/keeper/cctx_utils.go

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

x/crosschain/keeper/keeper_cross_chain_tx.go

@@ -215,6 +215,34 @@ func (k Keeper) CctxAllPending(c context.Context, req *types.QueryAllCctxPending
 	return &types.QueryAllCctxPendingResponse{CrossChainTx: sends}, nil
 }
 
+func (k Keeper) CctxAllPendingInNonceRange(c context.Context, req *types.QueryAllCctxPendingInNonceRangeRequest) (*types.QueryAllCctxPendingInNonceRangeResponse, error) {
+	if req == nil {
+		return nil, status.Error(codes.InvalidArgument, "invalid request")
+	}
+	ctx := sdk.UnwrapSDKContext(c)
+	tss, found := k.GetTSS(ctx)
+	if !found {
+		return nil, status.Error(codes.Internal, "tss not found")
+	}
+	cctxs := make([]*types.CrossChainTx, 0)
+
+	for i := req.NonceLow; i < req.NonceHigh; i++ {
+		res, found := k.GetNonceToCctx(ctx, tss.TssPubkey, req.ChainId, int64(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, res.CctxIndex)
+		if !found {
+			return nil, status.Error(codes.Internal, fmt.Sprintf("cctx not found: index %s", res.CctxIndex))
+		}
+		if IsPending(cctx) {
+			cctxs = append(cctxs, &cctx)
+		}
+	}
+
+	return &types.QueryAllCctxPendingInNonceRangeResponse{CrossChainTx: cctxs}, nil
+}
+
 func (k Keeper) CreateNewCCTX(ctx sdk.Context, msg *types.MsgVoteOnObservedInboundTx, index string, tssPubkey string, s types.CctxStatus, senderChain, receiverChain *common.Chain) types.CrossChainTx {
 	if msg.TxOrigin == "" {
 		msg.TxOrigin = msg.Sender