added total value in rate limiter window for monitoring purpose

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

@@ -194,6 +194,7 @@ LoopForwards:
 	return &types.QueryListPendingCctxWithinRateLimitResponse{
 		CrossChainTx:      cctxs,
 		TotalPending:      totalPending,
+		ValueWithinWindow: totalCctxValueInZeta.TruncateInt().Uint64(),
 		RateLimitExceeded: limitExceeded,
 	}, nil
 }

x/crosschain/keeper/grpc_query_cctx_rate_limit_test.go

@@ -406,11 +406,13 @@ func TestKeeper_ListPendingCctxWithinRateLimit(t *testing.T) {
 		require.EqualValues(t, cctxsBTC, res.CrossChainTx[200:400])
 		require.EqualValues(t, uint64(400), res.TotalPending)
 		require.False(t, res.RateLimitExceeded)
+		require.Equal(t, uint64(1500), res.ValueWithinWindow) // 500 * (2.5 + 0.5)
 	})
 	t.Run("Set rate to a lower value (< 1200) to early break the LoopBackwards with criteria #2", func(t *testing.T) {
 		k, ctx, cctxsETH, cctxsBTC, rlFlags := createKeeperForRateLimiterTest(t)
 
-		rlFlags.Rate = math.NewUint(1000) // 1000 ZETA
+		rate := uint64(1000) // 1000 ZETA
+		rlFlags.Rate = math.NewUint(rate)
 		k.SetRateLimiterFlags(ctx, rlFlags)
 
 		res, err := k.ListPendingCctxWithinRateLimit(ctx, &types.QueryListPendingCctxWithinRateLimitRequest{})
@@ -420,6 +422,7 @@ func TestKeeper_ListPendingCctxWithinRateLimit(t *testing.T) {
 		require.EqualValues(t, cctxsBTC[:100], res.CrossChainTx[100:200])
 		require.EqualValues(t, uint64(400), res.TotalPending)
 		require.True(t, res.RateLimitExceeded)
+		require.True(t, res.ValueWithinWindow >= rate)
 	})
 	t.Run("Set high rate and big window to early to break inner loop with the criteria #1", func(t *testing.T) {
 		k, ctx, cctxsETH, cctxsBTC, rlFlags := createKeeperForRateLimiterTest(t)
@@ -436,6 +439,7 @@ func TestKeeper_ListPendingCctxWithinRateLimit(t *testing.T) {
 		require.EqualValues(t, cctxsBTC, res.CrossChainTx[200:400])
 		require.EqualValues(t, uint64(400), res.TotalPending)
 		require.False(t, res.RateLimitExceeded)
+		require.EqualValues(t, uint64(3450), res.ValueWithinWindow) // 1150 * (2.5 + 0.5)
 	})
 	t.Run("Set lower request limit to early break the LoopForwards loop", func(t *testing.T) {
 		k, ctx, cctxsETH, cctxsBTC, _ := createKeeperForRateLimiterTest(t)
@@ -447,6 +451,7 @@ func TestKeeper_ListPendingCctxWithinRateLimit(t *testing.T) {
 		require.EqualValues(t, cctxsBTC, res.CrossChainTx[100:300])
 		require.EqualValues(t, uint64(400), res.TotalPending)
 		require.False(t, res.RateLimitExceeded)
+		require.EqualValues(t, uint64(1250), res.ValueWithinWindow) // 500 * 0.5 + 400 * 2.5
 	})
 	t.Run("Set rate to middle value (1200 < rate < 1500) to early break the LoopForwards loop with criteria #2", func(t *testing.T) {
 		k, ctx, cctxsETH, cctxsBTC, rlFlags := createKeeperForRateLimiterTest(t)
@@ -462,5 +467,6 @@ func TestKeeper_ListPendingCctxWithinRateLimit(t *testing.T) {
 		require.EqualValues(t, cctxsBTC, res.CrossChainTx[120:320])
 		require.EqualValues(t, uint64(400), res.TotalPending)
 		require.True(t, res.RateLimitExceeded)
+		require.True(t, res.ValueWithinWindow >= 1300)
 	})
 }

zetaclient/interfaces/interfaces.go

@@ -98,7 +98,7 @@ type ZetaCoreBridger interface {
 	GetZetaBlockHeight() (int64, error)
 	GetLastBlockHeightByChain(chain chains.Chain) (*crosschaintypes.LastBlockHeight, error)
 	ListPendingCctx(chainID int64) ([]*crosschaintypes.CrossChainTx, uint64, error)
-	ListPendingCctxWithinRatelimit() ([]*crosschaintypes.CrossChainTx, uint64, bool, error)
+	ListPendingCctxWithinRatelimit() ([]*crosschaintypes.CrossChainTx, uint64, uint64, bool, error)
 	GetPendingNoncesByChain(chainID int64) (observertypes.PendingNonces, error)
 	GetCctxByNonce(chainID int64, nonce uint64) (*crosschaintypes.CrossChainTx, error)
 	GetOutTxTracker(chain chains.Chain, nonce uint64) (*crosschaintypes.OutTxTracker, error)

zetaclient/testutils/stub/core_bridge.go

@@ -121,11 +121,11 @@ func (z *MockZetaCoreBridge) ListPendingCctx(_ int64) ([]*cctxtypes.CrossChainTx
 	return []*cctxtypes.CrossChainTx{}, 0, nil
 }
 
-func (z *MockZetaCoreBridge) ListPendingCctxWithinRatelimit() ([]*cctxtypes.CrossChainTx, uint64, bool, error) {
+func (z *MockZetaCoreBridge) ListPendingCctxWithinRatelimit() ([]*cctxtypes.CrossChainTx, uint64, uint64, bool, error) {
 	if z.paused {
-		return nil, 0, false, errors.New(ErrMsgPaused)
+		return nil, 0, 0, false, errors.New(ErrMsgPaused)
 	}
-	return []*cctxtypes.CrossChainTx{}, 0, false, nil
+	return []*cctxtypes.CrossChainTx{}, 0, 0, false, nil
 }
 
 func (z *MockZetaCoreBridge) GetPendingNoncesByChain(_ int64) (observerTypes.PendingNonces, error) {

zetaclient/zetabridge/query.go

@@ -139,7 +139,7 @@ func (b *ZetaCoreBridge) ListPendingCctx(chainID int64) ([]*types.CrossChainTx,
 // ListPendingCctxWithinRatelimit returns a list of pending cctxs that do not exceed the outbound rate limit
 //   - The max size of the list is crosschainkeeper.MaxPendingCctxs
 //   - The returned `rateLimitExceeded` flag indicates if the rate limit is exceeded or not
-func (b *ZetaCoreBridge) ListPendingCctxWithinRatelimit() ([]*types.CrossChainTx, uint64, bool, error) {
+func (b *ZetaCoreBridge) ListPendingCctxWithinRatelimit() ([]*types.CrossChainTx, uint64, uint64, bool, error) {
 	client := types.NewQueryClient(b.grpcConn)
 	maxSizeOption := grpc.MaxCallRecvMsgSize(32 * 1024 * 1024)
 	resp, err := client.ListPendingCctxWithinRateLimit(
@@ -148,9 +148,9 @@ func (b *ZetaCoreBridge) ListPendingCctxWithinRatelimit() ([]*types.CrossChainTx
 		maxSizeOption,
 	)
 	if err != nil {
-		return nil, 0, false, err
+		return nil, 0, 0, false, err
 	}
-	return resp.CrossChainTx, resp.TotalPending, resp.RateLimitExceeded, nil
+	return resp.CrossChainTx, resp.TotalPending, resp.ValueWithinWindow, resp.RateLimitExceeded, nil
 }
 
 func (b *ZetaCoreBridge) GetAbortedZetaAmount() (string, error) {

zetaclient/zetacore_observer.go

@@ -134,10 +134,14 @@ func (co *CoreObserver) startCctxScheduler(appContext *appcontext.AppContext) {
 					metrics.HotKeyBurnRate.Set(float64(co.ts.HotKeyBurnRate.GetBurnRate().Int64()))
 
 					// query pending cctxs across all foreign chains with rate limit
-					cctxMap, err := co.getAllPendingCctxWithRatelimit()
+					cctxMap, valueWithinWindow, err := co.getAllPendingCctxWithRatelimit()
 					if err != nil {
 						co.logger.ZetaChainWatcher.Error().Err(err).Msgf("startCctxScheduler: queryPendingCctxWithRatelimit failed")
 					}
+					// print value within rate limiter window every minute
+					if bn%10 == 0 {
+						co.logger.ZetaChainWatcher.Debug().Msgf("startCctxScheduler: value within rate limiter window is %d ZETA", valueWithinWindow)
+					}
 
 					// schedule keysign for pending cctxs on each chain
 					coreContext := appContext.ZetaCoreContext()
@@ -190,10 +194,10 @@ func (co *CoreObserver) startCctxScheduler(appContext *appcontext.AppContext) {
 }
 
 // getAllPendingCctxWithRatelimit get pending cctxs across all foreign chains with rate limit
-func (co *CoreObserver) getAllPendingCctxWithRatelimit() (map[int64][]*types.CrossChainTx, error) {
-	cctxList, totalPending, rateLimitExceeded, err := co.bridge.ListPendingCctxWithinRatelimit()
+func (co *CoreObserver) getAllPendingCctxWithRatelimit() (map[int64][]*types.CrossChainTx, uint64, error) {
+	cctxList, totalPending, valueWithinWindow, rateLimitExceeded, err := co.bridge.ListPendingCctxWithinRatelimit()
 	if err != nil {
-		return nil, err
+		return nil, 0, err
 	}
 	if rateLimitExceeded {
 		co.logger.ZetaChainWatcher.Warn().Msgf("rate limit exceeded, fetched %d cctxs out of %d", len(cctxList), totalPending)
@@ -209,7 +213,7 @@ func (co *CoreObserver) getAllPendingCctxWithRatelimit() (map[int64][]*types.Cro
 		cctxMap[chainID] = append(cctxMap[chainID], cctx)
 	}
 
-	return cctxMap, nil
+	return cctxMap, valueWithinWindow, nil
 }
 
 // scheduleCctxEVM schedules evm outtx keysign on each ZetaChain block (the ticker)