diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f829b12c3..f01b58a434 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/x/cronos/keeper/export_test.go b/x/cronos/keeper/export_test.go new file mode 100644 index 0000000000..24b722c5b7 --- /dev/null +++ b/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) +} diff --git a/x/cronos/keeper/grpc_query.go b/x/cronos/keeper/grpc_query.go index 43f276c894..892fe2d72f 100644 --- a/x/cronos/keeper/grpc_query.go +++ b/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 { diff --git a/x/cronos/keeper/grpc_query_replay_concurrency_test.go b/x/cronos/keeper/grpc_query_replay_concurrency_test.go new file mode 100644 index 0000000000..0b83197d9f --- /dev/null +++ b/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)) +}