fix: prevent cosmos mempool proposal starvation under backlog

CHANGELOG.md

@@ -73,6 +73,7 @@
 - [\#1050](https://github.com/cosmos/evm/pull/1050) Align precompile gas calculation with expected EVM gas semantics.
 - [\#1107](https://github.com/cosmos/evm/pull/1107) Skip StateDB commit error transactions during receipt conversion to prevent `invalid message index` errors in block RPCs.
 - [\#1216](https://github.com/cosmos/evm/pull/1216) Fix blocking on mempool event bus unsubscribe.
+- [\#1227](https://github.com/cosmos/evm/issues/1227) Fix cosmos-pool proposal starvation under backlogs: carry the rechecked snapshot across heights, serve it when recheck lags, and re-verify proposal txs not validated at the proposal base.
 - [\#1244](https://github.com/cosmos/evm/pull/1244) Avoid node crash from panics in the mempool insert queue.
 
 ## v0.6.0

evmd/app.go

@@ -132,6 +132,13 @@ func init() {
 	// manually update the power reduction by replacing micro (u) -> atto (a) evmos
 	sdk.DefaultPowerReduction = utils.AttoPowerReduction
 
+	// Pin config-registry scope to skip per-call syscalls, see cosmos-sdk#26729
+	if os.Getenv(sdk.EnvConfigScope) == "" {
+		if err := os.Setenv(sdk.EnvConfigScope, appName); err != nil {
+			panic(err)
+		}
+	}
+
 	defaultNodeHome = evmconfig.MustGetDefaultNodeHome()
 }
 

evmd/mempool.go

@@ -1,6 +1,8 @@
 package evmd
 
 import (
+	abci "github.com/cometbft/cometbft/abci/types"
+
 	evmmempool "github.com/cosmos/evm/mempool"
 	"github.com/cosmos/evm/server"
 	evmtypes "github.com/cosmos/evm/x/vm/types"
@@ -53,15 +55,24 @@ func (app *EVMD) configureEVMMempool(appOpts servertypes.AppOptions, logger log.
 
 	app.EVMMempool = mempool
 
-	// create ABCI handlers
-	proposalHandler := baseapp.NewDefaultProposalHandler(mempool, NewNoCheckProposalTxVerifier(app.BaseApp))
+	// Re-run ante for any selected tx the mempool cannot prove was validated
+	// at the height this proposal builds on (see SnapshotVerifiedTxVerifier).
+	// The base comes from the ABCI request, not the notify-driven pin, which
+	// can lag a beat behind the last commit.
+	verifier := NewSnapshotVerifiedTxVerifier(app.BaseApp, mempool, logger)
+	proposalHandler := baseapp.NewDefaultProposalHandler(mempool, verifier)
+	defaultPrepareProposal := proposalHandler.PrepareProposalHandler()
+	prepareProposalHandler := func(ctx sdk.Context, req *abci.RequestPrepareProposal) (*abci.ResponsePrepareProposal, error) {
+		verifier.SetProposalBase(req.Height - 1)
+		return defaultPrepareProposal(ctx, req)
+	}
 
 	insertTxHandler := mempool.NewInsertTxHandler(app.TxDecode)
 	reapTxsHandler := mempool.NewReapTxsHandler()
 	checkTxHandler := mempool.NewCheckTxHandler(app.TxDecode, checkTxTimeout)
 
 	// set handlers and the mempool
-	app.SetPrepareProposal(proposalHandler.PrepareProposalHandler())
+	app.SetPrepareProposal(prepareProposalHandler)
 	app.SetProcessProposal(proposalHandler.ProcessProposalHandler())
 	app.SetInsertTxHandler(insertTxHandler)
 	app.SetReapTxsHandler(reapTxsHandler)

evmd/tx_verifier.go

@@ -1,27 +1,107 @@
 package evmd
 
 import (
+	"sync"
+	"sync/atomic"
+
+	evmtypes "github.com/cosmos/evm/x/vm/types"
+
+	"cosmossdk.io/log/v2"
+
 	"github.com/cosmos/cosmos-sdk/baseapp"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 )
 
-var _ baseapp.ProposalTxVerifier = &NoCheckProposalTxVerifier{}
+var _ baseapp.ProposalTxVerifier = &SnapshotVerifiedTxVerifier{}
 
-type NoCheckProposalTxVerifier struct {
-	*baseapp.BaseApp
+// ProposalSnapshot reports whether the mempool validated a proposal candidate
+// at the height the proposal builds on (see Mempool.ProposalTxValidatedAt).
+type ProposalSnapshot interface {
+	ProposalTxValidatedAt(tx sdk.Tx, base uint64) bool
 }
 
-func NewNoCheckProposalTxVerifier(b *baseapp.BaseApp) *NoCheckProposalTxVerifier {
-	return &NoCheckProposalTxVerifier{BaseApp: b}
+// SnapshotVerifiedTxVerifier re-runs ante over a proposal candidate only when
+// the mempool cannot show it was validated at the proposal's base height, set
+// per proposal from the ABCI request so a lagging pin fails closed.
+//
+// Encoding skips the ante effects (sequence bump, fee deduction) that a later
+// stale tx of the same signer needs, so encoded cosmos txs are queued and
+// verified ahead of the next stale one. Proposals without stale txs pay
+// nothing; a mixed one pays once per encoded tx, bounded by block capacity.
+type SnapshotVerifiedTxVerifier struct {
+	// ProposalTxVerifier runs full verification and the tx codec; evmd wires the BaseApp.
+	baseapp.ProposalTxVerifier
+	snapshot ProposalSnapshot
+	logger   log.Logger
+
+	// proposalBase is the height the in-flight proposal builds on
+	// (req.Height - 1). Zero means unknown and re-verifies everything.
+	proposalBase atomic.Int64
+
+	// encoded holds cosmos txs encoded without verification, in selection
+	// order, until their ante effects reach the prepare-proposal state.
+	mu      sync.Mutex
+	encoded []sdk.Tx
 }
 
-// PrepareProposalVerifyTx overrides the typical tx verification done in
-// BaseApp's PrepareProposalHandler. The default PrepareProposalVerifyTx
-// implementation encodes the tx to bytes, then calls runTx in 'checktx' mode,
-// executing all antehandlers.
-//
-// We now override the implementation to only verify that the tx can be encoded
-// to bytes, since we will guarantee that all txs selected are valid elsewhere.
-func (txv *NoCheckProposalTxVerifier) PrepareProposalVerifyTx(tx sdk.Tx) ([]byte, error) {
-	return txv.TxEncode(tx)
+func NewSnapshotVerifiedTxVerifier(base baseapp.ProposalTxVerifier, snapshot ProposalSnapshot, logger log.Logger) *SnapshotVerifiedTxVerifier {
+	return &SnapshotVerifiedTxVerifier{
+		ProposalTxVerifier: base,
+		snapshot:           snapshot,
+		logger:             logger.With(log.ModuleKey, "SnapshotVerifiedTxVerifier"),
+	}
+}
+
+// SetProposalBase records the height the next proposal builds on. Its queue
+// starts empty: the prepare-proposal state is rebuilt per proposal.
+func (txv *SnapshotVerifiedTxVerifier) SetProposalBase(height int64) {
+	txv.proposalBase.Store(height)
+	txv.mu.Lock()
+	txv.encoded = nil
+	txv.mu.Unlock()
+}
+
+// PrepareProposalVerifyTx encodes txs validated at the proposal's base height,
+// and fully verifies stale or unknown ones after landing the ante effects of
+// the txs encoded before them.
+func (txv *SnapshotVerifiedTxVerifier) PrepareProposalVerifyTx(tx sdk.Tx) ([]byte, error) {
+	if base := txv.proposalBase.Load(); base > 0 && txv.snapshot.ProposalTxValidatedAt(tx, uint64(base)) {
+		bz, err := txv.TxEncode(tx)
+		if err == nil && !isEVMTx(tx) {
+			txv.mu.Lock()
+			txv.encoded = append(txv.encoded, tx)
+			txv.mu.Unlock()
+		}
+		return bz, err
+	}
+	txv.verifyEncoded()
+	return txv.ProposalTxVerifier.PrepareProposalVerifyTx(tx)
+}
+
+// verifyEncoded replays the queued txs through full verification so their ante
+// effects reach the prepare-proposal state. They passed at base, so a failure
+// is only logged: the stale tx behind it then fails on its own and is skipped.
+func (txv *SnapshotVerifiedTxVerifier) verifyEncoded() {
+	txv.mu.Lock()
+	queued := txv.encoded
+	txv.encoded = nil
+	txv.mu.Unlock()
+
+	for _, tx := range queued {
+		if _, err := txv.ProposalTxVerifier.PrepareProposalVerifyTx(tx); err != nil {
+			txv.logger.Warn("encoded proposal tx failed verification at the proposal base", "err", err)
+		}
+	}
+}
+
+// isEVMTx reports whether tx carries a single MsgEthereumTx. Such txs are
+// never queued: the reserver keeps an address in one pool at a time, so they
+// never precede a stale cosmos tx of the same signer.
+func isEVMTx(tx sdk.Tx) bool {
+	msgs := tx.GetMsgs()
+	if len(msgs) != 1 {
+		return false
+	}
+	_, ok := msgs[0].(*evmtypes.MsgEthereumTx)
+	return ok
 }

evmd/tx_verifier_test.go

@@ -0,0 +1,148 @@
+package evmd
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	protov2 "google.golang.org/protobuf/proto"
+
+	evmtypes "github.com/cosmos/evm/x/vm/types"
+
+	"cosmossdk.io/log/v2"
+
+	"github.com/cosmos/cosmos-sdk/baseapp"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+)
+
+// stubTx is a minimal sdk.Tx whose identity is its pointer.
+type stubTx struct {
+	msgs []sdk.Msg
+}
+
+func (s *stubTx) GetMsgs() []sdk.Msg                    { return s.msgs }
+func (s *stubTx) GetMsgsV2() ([]protov2.Message, error) { return nil, nil }
+
+// recordingVerifier records which txs reached full verification and which
+// were only encoded.
+type recordingVerifier struct {
+	baseapp.ProposalTxVerifier // unused methods
+	verified                   []sdk.Tx
+	encoded                    []sdk.Tx
+}
+
+func (v *recordingVerifier) PrepareProposalVerifyTx(tx sdk.Tx) ([]byte, error) {
+	v.verified = append(v.verified, tx)
+	return []byte("verified"), nil
+}
+
+func (v *recordingVerifier) TxEncode(tx sdk.Tx) ([]byte, error) {
+	v.encoded = append(v.encoded, tx)
+	return []byte("encoded"), nil
+}
+
+// stampedSnapshot maps txs to the height they were validated at.
+type stampedSnapshot map[sdk.Tx]uint64
+
+func (s stampedSnapshot) ProposalTxValidatedAt(tx sdk.Tx, base uint64) bool {
+	height, ok := s[tx]
+	return ok && height == base
+}
+
+func TestSnapshotVerifiedTxVerifier(t *testing.T) {
+	const base = int64(10)
+
+	var (
+		fresh1 = &stubTx{}
+		fresh2 = &stubTx{}
+		stale1 = &stubTx{}
+		stale2 = &stubTx{}
+		evm    = &stubTx{msgs: []sdk.Msg{&evmtypes.MsgEthereumTx{}}}
+	)
+	snapshot := stampedSnapshot{
+		fresh1: uint64(base),
+		fresh2: uint64(base),
+		stale1: uint64(base) - 1,
+		stale2: uint64(base) - 1,
+		evm:    uint64(base),
+	}
+
+	testCases := []struct {
+		name         string
+		base         int64
+		txs          []sdk.Tx
+		wantVerified []sdk.Tx
+		wantEncoded  []sdk.Tx
+	}{
+		{
+			name:        "txs validated at base are encoded, not verified",
+			base:        base,
+			txs:         []sdk.Tx{fresh1, fresh2},
+			wantEncoded: []sdk.Tx{fresh1, fresh2},
+		},
+		{
+			name:         "encoded txs are verified ahead of the stale tx that follows",
+			base:         base,
+			txs:          []sdk.Tx{fresh1, fresh2, stale1},
+			wantVerified: []sdk.Tx{fresh1, fresh2, stale1},
+			wantEncoded:  []sdk.Tx{fresh1, fresh2},
+		},
+		{
+			name:         "each encoded tx is replayed once",
+			base:         base,
+			txs:          []sdk.Tx{fresh1, stale1, fresh2, stale2},
+			wantVerified: []sdk.Tx{fresh1, stale1, fresh2, stale2},
+			wantEncoded:  []sdk.Tx{fresh1, fresh2},
+		},
+		{
+			name:         "an unknown base verifies everything",
+			base:         0,
+			txs:          []sdk.Tx{fresh1, stale1},
+			wantVerified: []sdk.Tx{fresh1, stale1},
+		},
+		{
+			name:         "evm txs are encoded but never replayed",
+			base:         base,
+			txs:          []sdk.Tx{evm, stale1},
+			wantVerified: []sdk.Tx{stale1},
+			wantEncoded:  []sdk.Tx{evm},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			recorder := &recordingVerifier{}
+			verifier := NewSnapshotVerifiedTxVerifier(recorder, snapshot, log.NewNopLogger())
+			verifier.SetProposalBase(tc.base)
+
+			for _, tx := range tc.txs {
+				_, err := verifier.PrepareProposalVerifyTx(tx)
+				require.NoError(t, err)
+			}
+
+			require.Equal(t, tc.wantVerified, recorder.verified)
+			require.Equal(t, tc.wantEncoded, recorder.encoded)
+		})
+	}
+}
+
+// A new proposal rebuilds the prepare-proposal state, so txs encoded for the
+// previous proposal must not be replayed into it.
+func TestSnapshotVerifiedTxVerifier_NewProposalDropsQueue(t *testing.T) {
+	const base = int64(10)
+	fresh := &stubTx{}
+	stale := &stubTx{}
+	snapshot := stampedSnapshot{fresh: uint64(base), stale: uint64(base) - 1}
+
+	recorder := &recordingVerifier{}
+	verifier := NewSnapshotVerifiedTxVerifier(recorder, snapshot, log.NewNopLogger())
+
+	verifier.SetProposalBase(base)
+	_, err := verifier.PrepareProposalVerifyTx(fresh)
+	require.NoError(t, err)
+	require.Empty(t, recorder.verified)
+
+	verifier.SetProposalBase(base)
+	_, err = verifier.PrepareProposalVerifyTx(stale)
+	require.NoError(t, err)
+	require.Equal(t, []sdk.Tx{stale}, recorder.verified, "previous proposal's queue must not be replayed")
+}