fix: prevent cosmos mempool proposal starvation under backlog

CHANGELOG.md

@@ -69,6 +69,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 large backlogs: carry rechecked snapshot across heights, prune committed txs via a per-signer nonce watermark, and serve last completed snapshot when recheck loop falls behind a proposal.
 
 ## v0.6.0
 

mempool/internal/heightsync/heightsync.go

@@ -116,6 +116,15 @@ type HeightSync[Store any] struct {
 	// fields of the Store itself
 	mu sync.RWMutex
 
+	// staleFallback makes GetStore return the current carried-forward Store
+	// (instead of nil) when it times out while still behind the target height.
+	// That Store is at a height <= target and, even mid-recheck, carry-forward
+	// keeps it a valid subset of validated txs. Only enable it for stores that
+	// stay free of state a committed block invalidated (see the cosmos pool's
+	// committed-nonce watermark), otherwise a stale Store could serve
+	// already-committed txs.
+	staleFallback bool
+
 	logger log.Logger
 }
 
@@ -134,9 +143,26 @@ func New[Store any](startHeight *big.Int, reset func(logger log.Logger) *Store,
 	return hs
 }
 
+// WithStaleFallback enables the stale fallback (see the staleFallback field)
+// and returns hs for chaining at construction.
+func (hs *HeightSync[Store]) WithStaleFallback() *HeightSync[Store] {
+	hs.staleFallback = true
+	return hs
+}
+
 // StartNewHeight resets the HeightSync for a new height, overwriting the
 // previous Store with a fresh Store via the reset fn.
 func (hs *HeightSync[Store]) StartNewHeight(height *big.Int) {
+	hs.StartNewHeightFrom(height, nil)
+}
+
+// StartNewHeightFrom starts a new height whose Store is derived from the
+// previous height's Store via carry (nil carry means a fresh Store, as in
+// StartNewHeight). Carrying lets a producer keep validated state across
+// heights, so a pass cancelled before completion does not discard everything
+// the previous height validated. carry runs while the HeightSync write lock is
+// held and must not call back into the HeightSync.
+func (hs *HeightSync[Store]) StartNewHeightFrom(height *big.Int, carry func(prev *Store) *Store) {
 	hs.mu.Lock()
 	defer hs.mu.Unlock()
 
@@ -146,9 +172,13 @@ func (hs *HeightSync[Store]) StartNewHeight(height *big.Int) {
 		panic(fmt.Errorf("height %s not ended before starting new height %s", hs.currentHeight.String(), height.String()))
 	}
 
-	// create new Store for this height
+	// create the Store for this height, either fresh or carried forward
 	hs.currentHeight = new(big.Int).Set(height)
-	hs.store = hs.reset(hs.logger)
+	if carry != nil {
+		hs.store = carry(hs.store)
+	} else {
+		hs.store = hs.reset(hs.logger)
+	}
 
 	// close old channel and create new one to wake up consumers
 	oldChan := hs.heightChanged
@@ -237,8 +267,7 @@ func (hs *HeightSync[Store]) GetStore(ctx context.Context, height *big.Int) *Sto
 		}
 
 		// current height is behind target, we cannot return the Store at the
-		// current height, so we must wait for the height to advance to the
-		// callers target height
+		// target height yet, so we wait for the height to advance.
 		heightChangedChan := hs.heightChanged
 		hs.mu.RUnlock()
 
@@ -250,10 +279,27 @@ func (hs *HeightSync[Store]) GetStore(ctx context.Context, height *big.Int) *Sto
 			// height changed, loop back to check if we've reached target
 			continue
 		case <-ctx.Done():
-			// caller is done waiting, but we still do not have a Store at the
-			// correct height to return, return nil instead
+			// The caller is done waiting and the height sync never reached the
+			// target height.
 			hsTimeout.Add(ctx, 1)
-			return nil
+			if !hs.staleFallback {
+				// caller opted out of stale results, return nil
+				return nil
+			}
+			// Rather than starve the caller (e.g. an empty block proposal),
+			// fall back to the current carried-forward Store: it is at a
+			// height <= target and, even mid-recheck, holds a valid subset of
+			// validated txs. Safe to serve as long as producers keep it free
+			// of state a since-committed block invalidated (the cosmos pool
+			// does this via its committed-nonce watermark).
+			hs.mu.RLock()
+			value := hs.store
+			// heights can skip past target while we waited, never serve future state
+			if hs.currentHeight.Cmp(height) > 0 {
+				value = nil
+			}
+			hs.mu.RUnlock()
+			return value
 		}
 	}
 }

mempool/internal/heightsync/heightsync_test.go

@@ -190,6 +190,87 @@ func TestStartNewHeightResetsValue(t *testing.T) {
 	require.Empty(t, result.get())
 }
 
+func TestStartNewHeightFromCarriesStore(t *testing.T) {
+	hv := heightsync.New(big.NewInt(1), newTestValue, log.NewNopLogger())
+
+	hv.StartNewHeight(big.NewInt(1))
+	hv.Do(func(s *testStore) { s.add("carried") })
+	hv.EndCurrentHeight()
+
+	// advance to height 2 carrying the previous store forward
+	hv.StartNewHeightFrom(big.NewInt(2), func(prev *testStore) *testStore {
+		return &testStore{items: prev.get()}
+	})
+	hv.Do(func(s *testStore) { s.add("fresh") })
+	hv.EndCurrentHeight()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+	defer cancel()
+
+	result := hv.GetStore(ctx, big.NewInt(2))
+	require.NotNil(t, result)
+	require.Equal(t, []string{"carried", "fresh"}, result.get())
+}
+
+// With the stale-fallback option, GetStore returns the current store (at a
+// height <= target) instead of nil when it times out behind the target height.
+func TestStaleFallbackReturnsLastStore(t *testing.T) {
+	hv := heightsync.New(big.NewInt(1), newTestValue, log.NewNopLogger()).WithStaleFallback()
+
+	hv.StartNewHeight(big.NewInt(1))
+	hv.Do(func(s *testStore) { s.add("h1") })
+	hv.EndCurrentHeight()
+
+	// request height 2 but never advance to it: times out on the height-behind path
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	value := hv.GetStore(ctx, big.NewInt(2))
+	require.NotNil(t, value)
+	require.Equal(t, []string{"h1"}, value.get())
+}
+
+// Without the option, the same height-behind timeout returns nil (the default).
+func TestNoStaleFallbackReturnsNil(t *testing.T) {
+	hv := heightsync.New(big.NewInt(1), newTestValue, log.NewNopLogger())
+
+	hv.StartNewHeight(big.NewInt(1))
+	hv.Do(func(s *testStore) { s.add("h1") })
+	hv.EndCurrentHeight()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	require.Nil(t, hv.GetStore(ctx, big.NewInt(2)))
+}
+
+// The stale fallback must never serve a store from a height past the target:
+// that is future state the target's proposal must not see, so GetStore
+// returns nil instead.
+func TestStaleFallbackDoesNotServePastTarget(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		hv := heightsync.New(big.NewInt(1), newTestValue, log.NewNopLogger()).WithStaleFallback()
+
+		// park a getter for height 2 on the height-behind wait
+		getCtx, cancelGet := context.WithCancel(context.Background())
+		defer cancelGet()
+		valueChan := make(chan *testStore)
+		go func() {
+			valueChan <- hv.GetStore(getCtx, big.NewInt(2))
+		}()
+		time.Sleep(1 * time.Second)
+
+		// Skip to height 3, cancelling the getter while carry holds the write
+		// lock: it wakes on the fallback path with the height past its target.
+		hv.StartNewHeightFrom(big.NewInt(3), func(prev *testStore) *testStore {
+			cancelGet()
+			return prev
+		})
+
+		require.Nil(t, <-valueChan, "fallback served a store from a height past the target")
+	})
+}
+
 func TestConcurrentDo(t *testing.T) {
 	hv := heightsync.New(big.NewInt(1), newTestValue, log.NewNopLogger())
 

mempool/mempool.go

@@ -200,11 +200,14 @@ func NewMempool(
 		panic("tx pool should contain only legacypool")
 	}
 
+	// Stale fallback: when recheck loop falls behind consensus, serve last completed
+	// snapshot rather than an empty proposal — since store's committed-nonce
+	// watermark keeps already-committed txs out.
 	heightSync := heightsync.New(
 		blockchain.CurrentBlock().Number,
 		NewCosmosTxStore,
 		logger.With("pool", "cosmos_recheck_mempool"),
-	)
+	).WithStaleFallback()
 
 	reservationHandle := reservationTracker.NewHandle(cosmosReserverHandlerID, reserver.WithRefCounter())
 
@@ -455,6 +458,10 @@ func (m *Mempool) removeCosmosTx(tx sdk.Tx, reason sdkmempool.RemoveReason) erro
 
 	if reason.Caller == sdkmempool.CallerRunTxFinalize {
 		m.recordNonceAdvances(tx)
+		// Prune committed tx from recheck snapshot synchronously. Snapshot is carried
+		// across heights, so without this a just-committed tx could be served
+		// into next proposal before async recheck pass drops it.
+		m.recheckCosmosPool.PruneCommitted(tx)
 	}
 
 	if err := m.recheckCosmosPool.Remove(tx); err != nil {
@@ -474,6 +481,9 @@ func (m *Mempool) removeEVMTx(tx sdk.Tx, msgEthereumTx *evmtypes.MsgEthereumTx,
 	if reason.Caller == sdkmempool.CallerRunTxFinalize {
 		_ = m.txTracker.IncludedInBlock(hash)
 		m.recordNonceAdvances(tx)
+		// an EVM tx consumes the same account sequence, so drop stale
+		// same-account cosmos txs from the snapshot too
+		m.recheckCosmosPool.PruneCommitted(tx)
 	}
 
 	if m.shouldRemoveFromEVMPool(hash, reason) {

mempool/recheck_pool.go

@@ -130,7 +130,7 @@ func NewRecheckMempool(
 		blockchain,
 		defaultCosmosPoolConfig,
 		maxTxs,
-		onTransactionReplace(reapList, signerExtractor, reserver, logger),
+		onTransactionReplace(reapList, recheckedTxs, signerExtractor, reserver, logger),
 	)
 
 	return &RecheckMempool{
@@ -429,7 +429,11 @@ func (m *RecheckMempool) runRecheck(done chan struct{}, newHead *ethtypes.Header
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
-	m.recheckedTxs.StartNewHeight(newHead.Number)
+	// Carry the validated set forward instead of resetting to an empty store,
+	// a pass cancelled by next block does not discard all progress and starve proposals.
+	// The pass prunes whatever became invalid, committed txs are kept out by
+	// the store's watermark (see CosmosTxStore.PruneCommitted).
+	m.recheckedTxs.StartNewHeightFrom(newHead.Number, (*CosmosTxStore).Clone)
 	defer m.recheckedTxs.EndCurrentHeight()
 
 	latestCtx, err := m.blockchain.GetLatestContext()
@@ -498,6 +502,10 @@ func (m *RecheckMempool) runRecheck(done chan struct{}, newHead *ethtypes.Header
 		}
 
 		removeTxs = append(removeTxs, txn)
+		// Drop from the snapshot at detection: the removal loop below is
+		// skipped on cancellation. ExtMempool removal still waits for the
+		// loop (reservations, multi-signer identification).
+		m.markTxRemoved(txn)
 
 		if keepFuturesOnError {
 			iter = iter.Next()
@@ -534,13 +542,30 @@ func (m *RecheckMempool) runRecheck(done chan struct{}, newHead *ethtypes.Header
 		}
 	}
 	txsRemoved = len(removeTxs)
+
+	// a completed pass makes watermarks recorded before it redundant
+	m.recheckedTxs.Do(func(store *CosmosTxStore) { store.AgeWatermarks() })
 }
 
 // markTxRechecked adds a tx into the height synced cosmos tx store.
 func (m *RecheckMempool) markTxRechecked(txn sdk.Tx) {
 	m.recheckedTxs.Do(func(store *CosmosTxStore) { store.AddTx(txn) })
 }
 
+// markTxRemoved drops a tx from the height synced cosmos tx store.
+func (m *RecheckMempool) markTxRemoved(txn sdk.Tx) {
+	m.recheckedTxs.Do(func(store *CosmosTxStore) { store.RemoveTx(txn) })
+}
+
+// PruneCommitted records that a block being finalized consumed tx's
+// signer/nonces and drops tx (and any lower-nonced sibling) from current snapshot.
+// It runs synchronously during FinalizeBlock so carried-forward store
+// can never feed an already-committed tx into a later proposal,
+// even before next recheck pass runs.
+func (m *RecheckMempool) PruneCommitted(txn sdk.Tx) {
+	m.recheckedTxs.Do(func(store *CosmosTxStore) { store.PruneCommitted(txn) })
+}
+
 // markTxInserted conservatively updates the current height snapshot for live inserts.
 // If the inserted tx replaces an existing tx, any other txs from the same sender with
 // a higher nonce is dropped and rebuilt by the next recheck.
@@ -621,16 +646,21 @@ func cosmosPoolConfig(
 
 func onTransactionReplace(
 	reapList *reaplist.ReapList,
+	recheckedTxs *heightsync.HeightSync[CosmosTxStore],
 	signerExtractor sdkmempool.SignerExtractionAdapter,
 	reserver *reserver.ReservationHandle,
 	logger log.Logger,
 ) func(oldTx, newTx sdk.Tx) {
-	return func(oldTx, _ sdk.Tx) {
+	return func(oldTx, newTx sdk.Tx) {
 		// tx is being replaced, we need to drop the tx that is going to be removed
 		// from the reap list. we assume that the tx doing the replacing has
 		// already been inserted into the reaplist via the insert.
 		reapList.DropCosmosTx(oldTx)
 
+		// drop the replaced tx from the snapshot when its signer set differs
+		// from the replacement's (see CosmosTxStore.InvalidateReplaced)
+		recheckedTxs.Do(func(store *CosmosTxStore) { store.InvalidateReplaced(oldTx, newTx) })
+
 		addrs, err := extractEVMAddresses(signerExtractor, oldTx)
 		if err != nil {
 			return