fix(cronos): bound concurrent ReplayBlock queries and honor the request context

CHANGELOG.md

@@ -12,6 +12,7 @@
 
 ### Bug fixes
 
+* [#2179](https://github.com/crypto-org-chain/cronos/pull/2179) fix(cronos): bound concurrent `ReplayBlock` queries and honor the request context.
 * [#2176](https://github.com/crypto-org-chain/cronos/pull/2176) fix(app): retry block list decryption instead of caching the blob before it is applied.
 * [#2155](https://github.com/crypto-org-chain/cronos/pull/2155) fix(mempool): size tx-cache-size and max-tx-bytes from mempool config directly.
 * [#2169](https://github.com/crypto-org-chain/cronos/pull/2169) fix(cronos): add safe multiply int check during voucher conversion to EVM coins.
@@ -23,7 +24,6 @@
 * [#2180](https://github.com/crypto-org-chain/cronos/pull/2180) chore: bump golang.org/x/text to v0.39.0.
 * [#2157](https://github.com/crypto-org-chain/cronos/pull/2157) chore: repin cronos-store, cometbft v0.39, cosmos-sdk v0.54 forks.
 
-
 *Jul 16, 2026*
 
 ## v1.8.0-alpha

x/cronos/keeper/export_test.go

@@ -0,0 +1,19 @@
+package keeper
+
+import "sync/atomic"
+
+var ReplayBlockSemaphore = replayBlockSem
+
+const ReplayBlockConcurrencyLimit = replayBlockConcurrency
+
+const ReplayBlockMaxQueued = replayBlockMaxQueued
+
+// SetReplayBlockQueued sets the waiter counter directly.
+func SetReplayBlockQueued(n int32) {
+	atomic.StoreInt32(&replayBlockQueued, n)
+}
+
+// ResetReplayBlockQueued clears the waiter counter.
+func ResetReplayBlockQueued() {
+	SetReplayBlockQueued(0)
+}

x/cronos/keeper/grpc_query.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"math/big"
+	"sync/atomic"
 
 	"github.com/crypto-org-chain/cronos/x/cronos/types"
 	"github.com/ethereum/go-ethereum/common"
@@ -25,8 +26,21 @@ const (
 	// ReplayBlockGasCap caps per-message EVM gas in a ReplayBlock query.
 	// Since historical blocks may have used different limits, we use a fixed upper bound value.
 	ReplayBlockGasCap = 60_000_000
+
+	// replayBlockConcurrency bounds how many ReplayBlock queries may run their
+	// EVM replay loop at once.
+	replayBlockConcurrency = 4
+
+	// replayBlockMaxQueued bounds how many callers may wait for a free slot.
+	replayBlockMaxQueued = 4 * replayBlockConcurrency
 )
 
+// replayBlockSem limits concurrent ReplayBlock executions across all calls to this process.
+var replayBlockSem = make(chan struct{}, replayBlockConcurrency)
+
+// replayBlockQueued counts callers currently running or waiting for a slot.
+var replayBlockQueued int32
+
 var _ types.QueryServer = Keeper{}
 
 // ContractByDenom query contract by denom, returns both external contract and auto deployed contract
@@ -67,6 +81,22 @@ func (k Keeper) ReplayBlock(goCtx context.Context, req *types.ReplayBlockRequest
 			"too many messages in ReplayBlock request: %d (max %d)", len(req.Msgs), MaxReplayBlockMsgs)
 	}
 
+	// Reject once too many callers are already running or queued.
+	if atomic.AddInt32(&replayBlockQueued, 1) > replayBlockMaxQueued {
+		atomic.AddInt32(&replayBlockQueued, -1)
+		return nil, status.Error(codes.ResourceExhausted, "too many concurrent ReplayBlock queries")
+	}
+	defer atomic.AddInt32(&replayBlockQueued, -1)
+
+	// Wait for a free execution slot; a client disconnect frees the caller
+	// without consuming a slot.
+	select {
+	case replayBlockSem <- struct{}{}:
+		defer func() { <-replayBlockSem }()
+	case <-goCtx.Done():
+		return nil, status.FromContextError(goCtx.Err()).Err()
+	}
+
 	rsps := make([]*evmtypes.MsgEthereumTxResponse, 0, len(req.Msgs))
 
 	// prepare the block context, the multistore version should be setup already in grpc query context.
@@ -110,6 +140,11 @@ func (k Keeper) ReplayBlock(goCtx context.Context, req *types.ReplayBlockRequest
 
 	// we assume the message executions are successful, they are filtered in json-rpc api
 	for _, msg := range req.Msgs {
+		// abort if the caller is already gone
+		if err := ctx.Err(); err != nil {
+			return nil, status.FromContextError(err).Err()
+		}
+
 		// deduct fee
 		// populate the `From` field
 		if _, err := msg.GetSenderLegacy(ethtypes.LatestSignerForChainID(chainID)); err != nil {

x/cronos/keeper/grpc_query_replay_concurrency_test.go

@@ -0,0 +1,76 @@
+package keeper_test
+
+import (
+	"context"
+	"math/big"
+	"time"
+
+	cronoskeeper "github.com/crypto-org-chain/cronos/x/cronos/keeper"
+	"github.com/crypto-org-chain/cronos/x/cronos/types"
+	evmtypes "github.com/evmos/ethermint/x/evm/types"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func (suite *KeeperTestSuite) TestReplayBlockConcurrencyLimit() {
+	req := &types.ReplayBlockRequest{
+		BlockNumber: 1,
+		BlockTime:   suite.ctx.BlockTime(),
+	}
+
+	for i := 0; i < cronoskeeper.ReplayBlockConcurrencyLimit; i++ {
+		cronoskeeper.ReplayBlockSemaphore <- struct{}{}
+	}
+
+	callCtx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+	defer cancel()
+
+	_, err := suite.app.CronosKeeper.ReplayBlock(suite.ctx.WithContext(callCtx), req)
+
+	for i := 0; i < cronoskeeper.ReplayBlockConcurrencyLimit; i++ {
+		<-cronoskeeper.ReplayBlockSemaphore
+	}
+	cronoskeeper.ResetReplayBlockQueued()
+
+	suite.Require().Error(err)
+	suite.Require().Equal(codes.DeadlineExceeded, status.Code(err))
+
+	_, err = suite.app.CronosKeeper.ReplayBlock(suite.ctx, req)
+	suite.Require().NoError(err)
+}
+
+func (suite *KeeperTestSuite) TestReplayBlockRejectsWhenQueueFull() {
+	defer cronoskeeper.ResetReplayBlockQueued()
+
+	req := &types.ReplayBlockRequest{
+		BlockNumber: 1,
+		BlockTime:   suite.ctx.BlockTime(),
+	}
+
+	cronoskeeper.SetReplayBlockQueued(cronoskeeper.ReplayBlockMaxQueued)
+	_, err := suite.app.CronosKeeper.ReplayBlock(suite.ctx, req)
+	suite.Require().Error(err)
+	suite.Require().Equal(codes.ResourceExhausted, status.Code(err))
+
+	cronoskeeper.SetReplayBlockQueued(cronoskeeper.ReplayBlockMaxQueued - 1)
+	_, err = suite.app.CronosKeeper.ReplayBlock(suite.ctx, req)
+	suite.Require().NoError(err)
+}
+
+func (suite *KeeperTestSuite) TestReplayBlockAbortsOnCancelledContext() {
+	newMsg := func(gas uint64) *evmtypes.MsgEthereumTx {
+		return evmtypes.NewTx(big.NewInt(1), 0, &suite.address, big.NewInt(0), gas, big.NewInt(1), nil, nil, nil, nil)
+	}
+	req := &types.ReplayBlockRequest{
+		Msgs:        []*evmtypes.MsgEthereumTx{newMsg(1000), newMsg(1000), newMsg(1000)},
+		BlockNumber: 1,
+		BlockTime:   suite.ctx.BlockTime(),
+	}
+
+	callCtx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	_, err := suite.app.CronosKeeper.ReplayBlock(suite.ctx.WithContext(callCtx), req)
+	suite.Require().Error(err)
+	suite.Require().Equal(codes.Canceled, status.Code(err))
+}