fix(mempool): mempool-owned branched context for admission + recheck

CHANGELOG.md

@@ -4,6 +4,9 @@
 
 *Jul 16, 2026*
 
+### Improvements
+
+* [#2159](https://github.com/crypto-org-chain/cronos/pull/2159) mempool-owned branched context for admission + recheck
 ### Chores
 
 * [#2157](https://github.com/crypto-org-chain/cronos/pull/2157) chore: repin cronos-store, cometbft v0.39, cosmos-sdk v0.54 forks.

app/app.go

@@ -350,6 +350,12 @@ type App struct {
 
 	senderCache *cache.SenderCache
 
+	// anteCache is the EVM ante's per-(sender, nonce) admission cache. Stored on
+	// App (not just local to setAnteHandler) so mempoolManager's eviction hook
+	// can share the same instance and delete a stale entry when the mempool
+	// evicts its tx without ever spending a RunTx on it (cascade/TTL eviction).
+	anteCache *cache.AnteCache
+
 	// unsafe to set for validator, used for testing
 	dummyCheckTx bool
 }
@@ -1202,6 +1208,16 @@ func New(
 			tmos.Exit(err.Error())
 		}
 
+		if app.mempoolManager != nil {
+			// Earliest correct point for the first mempoolState refresh: stores are
+			// now loaded, so the branch it takes sees committed state instead of an
+			// empty pre-load tree.
+			mu := app.mempoolManager.AdmissionMutex()
+			mu.Lock()
+			app.mempoolManager.RefreshMempoolStateLocked()
+			mu.Unlock()
+		}
+
 		if qmsVersion > 0 {
 			// it should not happens since we constraint the loaded iavl version to not exceed the versiondb version,
 			// still keep the check for safety.
@@ -1286,6 +1302,14 @@ func (app *App) setAnteHandler(txConfig client.TxConfig, mempoolMaxTxs int, blac
 		blockedMap[addr.String()] = struct{}{}
 	}
 	blockAddressDecorator := NewBlockAddressesDecorator(blockedMap, app.CronosKeeper.GetParams)
+	app.anteCache = cache.NewAnteCache(mempoolMaxTxs)
+	// mempoolManager (built earlier, inside the SetMempool baseAppOptions
+	// closure applied by NewBaseApp) is already set on app by this point, so
+	// wiring the eviction hook here can share app.anteCache with the ante
+	// options below rather than each holding a separate instance.
+	if app.mempoolManager != nil {
+		app.mempoolManager.SetEvictionHook(app.anteCache.Delete)
+	}
 	options := evmante.HandlerOptions{
 		AccountKeeper:          app.AccountKeeper,
 		BankKeeper:             app.BankKeeper,
@@ -1305,7 +1329,7 @@ func (app *App) setAnteHandler(txConfig client.TxConfig, mempoolMaxTxs int, blac
 		},
 		ExtraDecorators:   []sdk.AnteDecorator{blockAddressDecorator},
 		PendingTxListener: app.onPendingTx,
-		AnteCache:         cache.NewAnteCache(mempoolMaxTxs),
+		AnteCache:         app.anteCache,
 		SenderCache:       app.senderCache,
 	}
 
@@ -1680,11 +1704,23 @@ func (app *App) Commit() (*abci.ResponseCommit, error) {
 	}
 
 	resp, err := func() (*abci.ResponseCommit, error) {
-		// AppMempool.Lock() is a no-op; mu serializes checkState reset against concurrent admission.
+		// AppMempool.Lock() is a no-op; mu serializes BaseApp.Commit() and the
+		// mempoolState refresh against concurrent RunTx-based admission/recheck.
 		mu := app.mempoolManager.AdmissionMutex()
 		mu.Lock()
 		defer mu.Unlock()
-		return app.BaseApp.Commit()
+		resp, err := app.BaseApp.Commit()
+		if err == nil {
+			app.mempoolManager.RefreshMempoolStateLocked()
+		}
+		// On error, base is left pointing at the superseded store. Same for
+		// ApplySnapshotChunk: RestoreChunk streams straight into the live
+		// CommitMultiStore (snapshots.Manager.doRestoreSnapshot ->
+		// multistore.Restore) without ever calling RefreshMempoolStateLocked. A
+		// Commit error is effectively fatal and a state-syncing node isn't
+		// admitting or proposing, so nothing reads base until the next
+		// successful Commit refreshes it.
+		return resp, err
 	}()
 
 	if err == nil {

app/mempool/admission_test.go

@@ -251,7 +251,7 @@ func TestReapVsRecheckConcurrentRealTxs(t *testing.T) {
 	}
 	require.Equal(t, accounts, f.app.Mempool().CountTx())
 
-	// Admission bumped each sender's checkState nonce; an empty block resets it to
+	// Admission bumped each sender's nonce in base; an empty block resets it to
 	// committed (nonce 0) so recheck of these nonce-0 txs passes instead of
 	// failing stale.
 	_, err := f.app.FinalizeBlock(&abci.RequestFinalizeBlock{Height: 2, ProposerAddress: f.consAddress})

app/mempool/admitter.go

@@ -0,0 +1,129 @@
+package mempool
+
+import (
+	"fmt"
+
+	abci "github.com/cometbft/cometbft/abci/types"
+
+	errorsmod "cosmossdk.io/errors"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
+	sdkmempool "github.com/cosmos/cosmos-sdk/types/mempool"
+)
+
+// admitter is the admission half of the app mempool: peer-relayed InsertTx and
+// RPC CheckTx, both validated by RunTx against the shared branch.
+type admitter struct {
+	exec  *txExec
+	trace bool
+	// preVerify runs cheap verification lock-free before the tx admission mutex; set to nil for skip.
+	preVerify func([]byte) error
+}
+
+// admit is the shared admission path: preVerify + decode unlocked (bad txs skip
+// the mutex), then RunTx(ExecModeCheck) + cacheTx under it. Over-capacity maps to
+// CodeTypeRetry. tx stays nil when encCache is nil; BaseApp.RunTx accepts nil
+// sdk.Tx (uses txBytes).
+func (a *admitter) admit(txBytes []byte) (code uint32, codespace, log string) {
+	if a.preVerify != nil {
+		if err := a.preVerify(txBytes); err != nil {
+			cs, c, l := errorsmod.ABCIInfo(err, false)
+			return c, cs, l
+		}
+	}
+
+	var tx sdk.Tx
+	if a.exec.encCache != nil {
+		var err error
+		if tx, err = a.exec.decoder(txBytes); err != nil {
+			cs, c, l := errorsmod.ABCIInfo(sdkerrors.ErrTxDecode.Wrap(err.Error()), false)
+			return c, cs, l
+		}
+	}
+
+	a.exec.mu.Lock()
+	defer a.exec.mu.Unlock()
+
+	_, _, _, err := a.exec.runTxLocked(sdk.ExecModeCheck, txBytes, tx)
+	if err != nil {
+		if errorsmod.IsOf(err, sdkmempool.ErrMempoolTxMaxCapacity) {
+			return abci.CodeTypeRetry, "", "mempool is full"
+		}
+		cs, c, l := errorsmod.ABCIInfo(err, false)
+		return c, cs, l
+	}
+
+	a.cacheTx(tx, txBytes)
+	return abci.CodeTypeOK, "", ""
+}
+
+// cacheTx registers the already-decoded tx under its canonical bytes (raw
+// req.Tx bytes on encode error). No-op without a cache.
+func (a *admitter) cacheTx(tx sdk.Tx, raw []byte) {
+	if a.exec.encCache == nil {
+		return
+	}
+	bz := raw
+	if canonical, err := a.exec.txEncoder(tx); err == nil {
+		bz = canonical
+	}
+	a.exec.encCache.Set(tx, bz)
+}
+
+// insertTxHandler validates peer-relayed txs via RunTx(ExecModeCheck) before
+// admitting them.
+func (a *admitter) insertTxHandler() sdk.InsertTxHandler {
+	return func(req *abci.RequestInsertTx) (*abci.ResponseInsertTx, error) {
+		code, _, _ := a.admit(req.Tx)
+		return &abci.ResponseInsertTx{Code: code}, nil
+	}
+}
+
+// checkTxHandler runs RPC CheckTx. It calls the runner directly instead of the
+// runTx closure baseapp passes in (abci.go CheckTx), which hardcodes
+// txMultiStore = nil; the exec-mode mapping below mirrors BaseApp.CheckTx so
+// req.Type stays authoritative.
+func (a *admitter) checkTxHandler() sdk.CheckTxHandler {
+	return func(_ sdk.RunTx, req *abci.RequestCheckTx) (*abci.ResponseCheckTx, error) {
+		var mode sdk.ExecMode
+		switch req.Type {
+		case abci.CheckTxType_New:
+			mode = sdk.ExecModeCheck
+		case abci.CheckTxType_Recheck:
+			mode = sdk.ExecModeReCheck
+		default:
+			return nil, fmt.Errorf("unknown RequestCheckTx type: %s", req.Type)
+		}
+
+		// Decode before locking: proto unmarshal is CPU-intensive; decoder and
+		// DecodeCache have their own locks. Bad txs return without acquiring the mutex.
+		var tx sdk.Tx
+		if a.exec.encCache != nil {
+			var err error
+			if tx, err = a.exec.decoder(req.Tx); err != nil {
+				return sdkerrors.ResponseCheckTxWithEvents(sdkerrors.ErrTxDecode.Wrap(err.Error()), 0, 0, nil, a.trace), nil
+			}
+		}
+
+		a.exec.mu.Lock()
+		defer a.exec.mu.Unlock()
+
+		gasInfo, result, anteEvents, err := a.exec.runTxLocked(mode, req.Tx, tx)
+		if err != nil {
+			return sdkerrors.ResponseCheckTxWithEvents(err, gasInfo.GasWanted, gasInfo.GasUsed, anteEvents, a.trace), nil
+		}
+
+		a.cacheTx(tx, req.Tx)
+
+		// No MarkEventsToIndex (unlike default CheckTx): that flag only feeds
+		// the tx indexer on FinalizeBlock results, not CheckTx.
+		return &abci.ResponseCheckTx{
+			GasWanted: int64(gasInfo.GasWanted),
+			GasUsed:   int64(gasInfo.GasUsed),
+			Log:       result.Log,
+			Data:      result.Data,
+			Events:    result.Events,
+		}, nil
+	}
+}

app/mempool/exec.go

@@ -0,0 +1,61 @@
+package mempool
+
+import (
+	"sync"
+	"sync/atomic"
+
+	abci "github.com/cometbft/cometbft/abci/types"
+
+	"github.com/cosmos/cosmos-sdk/baseapp"
+	storetypes "github.com/cosmos/cosmos-sdk/store/v2/types"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+)
+
+type txRunner interface {
+	RunTx(mode sdk.ExecMode, txBytes []byte, tx sdk.Tx, txIndex int, txMultiStore storetypes.MultiStore, incarnationCache map[string]any) (sdk.GasInfo, *sdk.Result, []abci.Event, error)
+}
+
+var _ txRunner = (*baseapp.BaseApp)(nil)
+
+// txExec is what admission and recheck share: the RunTx entry point, the
+// branched state both run against, and the codecs both need.
+type txExec struct {
+	// mu guards state.base, the shared nonce authority for admission and recheck:
+	// RunTx is serialized through it, and App.Commit holds it across
+	// BaseApp.Commit() plus the post-Commit refresh so the swap never races a
+	// RunTx reader or the live memiavl tree mid-Commit. AppMempool.Lock() is a
+	// no-op, so mu also replaces the mempool lock BaseApp normally relies on.
+	// Held around RunTx and the cascade eviction that follows a proven nonce
+	// gap in the same chunk, so eviction stays atomic with admission; never
+	// held across the lock-free pool scan.
+	mu     sync.Mutex
+	runner txRunner
+	// state holds the CacheMultiStore branch RunTx uses in place of checkState.
+	// nil until the first refreshLocked call (after LoadLatestVersion in
+	// production, or never in the newManager() test constructor); store() returns
+	// nil in the meantime so RunTx's 5th arg falls back to checkState.
+	state *mempoolState
+	// gen counts state refreshes; a recheck pass abandons its remaining groups
+	// once gen advances, since they were selected against a superseded base.
+	gen       atomic.Uint64
+	encCache  *EncoderCache
+	txEncoder sdk.TxEncoder
+	decoder   sdk.TxDecoder
+}
+
+// runTxLocked runs tx against the mempool's own branch instead of checkState.
+// Precondition: the caller holds mu.
+func (e *txExec) runTxLocked(mode sdk.ExecMode, bz []byte, tx sdk.Tx) (sdk.GasInfo, *sdk.Result, []abci.Event, error) {
+	return e.runner.RunTx(mode, bz, tx, -1, e.state.store(), nil)
+}
+
+// refreshLocked rebranches off the freshly committed store and bumps gen,
+// canceling any recheck pass still validating against the superseded base.
+// Precondition: the caller holds mu.
+func (e *txExec) refreshLocked() {
+	if e.state == nil {
+		return
+	}
+	e.state.refreshLocked()
+	e.gen.Add(1)
+}