fix: refresh mempool caches every block

CHANGELOG.md

@@ -71,6 +71,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.
+- [\#1245](https://github.com/cosmos/evm/pull/1245) Refresh mempool caches every block.
 
 ## v0.6.0
 

evmd/mempool.go

@@ -70,10 +70,14 @@ func (app *EVMD) configureEVMMempool(appOpts servertypes.AppOptions, logger log.
 
 	app.SetMempool(mempool)
 
-	app.SetPrepareCheckStater(func(_ sdk.Context) {
-		if !mempool.HasEventBus() {
-			mempool.NotifyNewBlock()
-		}
+	app.SetPrepareCheckStater(func(ctx sdk.Context) {
+		// Notify every block, even with an event bus: its goroutine exits
+		// silently if CometBFT cancels the subscription, and both the pinned
+		// query context and the tx pool's statedb then outlive their IAVL
+		// version once pruning passes it. NotifyNewBlock emits at most one
+		// chain head event per height, so on a healthy node this call only
+		// refreshes the latest context.
+		mempool.NotifyNewBlockAt(ctx.BlockHeight())
 	})
 
 	return nil

mempool/blockchain.go

@@ -48,6 +48,12 @@ type Blockchain struct {
 	coinInfo           atomic.Pointer[evmtypes.EvmCoinInfo]
 
 	testingCommitMu sync.RWMutex
+
+	// notifyMu serializes NotifyNewBlock's drivers (PrepareCheckState and the
+	// event bus goroutine); atomic lastNotifiedHeight lets a duplicate height
+	// skip without the lock. It must not guard state read by CurrentBlock.
+	notifyMu           sync.Mutex
+	lastNotifiedHeight atomic.Int64
 }
 
 // NewBlockchain creates a new Blockchain instance that bridges Cosmos SDK state with Ethereum mempools.
@@ -173,14 +179,49 @@ func (b *Blockchain) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) even
 	return b.chainHeadFeed.Subscribe(ch)
 }
 
-// NotifyNewBlock sends a chain head event when a new block is finalized
-func (b *Blockchain) NotifyNewBlock() {
+// NotifyNewBlock refreshes the latest context and sends a chain head event,
+// returning the announced header (nil when nothing was sent). At most one
+// event fires per committed height, so multiple drivers are safe.
+func (b *Blockchain) NotifyNewBlock() *types.Header {
+	return b.NotifyNewBlockAt(0)
+}
+
+// NotifyNewBlockAt is NotifyNewBlock for drivers that know which height just
+// committed, so an already notified height returns before the refresh. The skip
+// is safe because both drivers name the same committed height -- PrepareCheckState
+// of block N runs after N commits and passes N, the height the event bus
+// goroutine's header event also carries. Height 0 means unknown.
+func (b *Blockchain) NotifyNewBlockAt(committedHeight int64) *types.Header {
+	// checked before the lock too — a loser must not wait out the winner's
+	// feed delivery just to skip; heights only grow, so a stale read can miss
+	// the skip but never take it wrongly
+	notified := func() bool {
+		return committedHeight > 0 && committedHeight <= b.lastNotifiedHeight.Load()
+	}
+	if notified() {
+		return nil
+	}
+
+	b.notifyMu.Lock()
+	defer b.notifyMu.Unlock()
+	if notified() {
+		return nil
+	}
+
 	latestCtx, err := b.newLatestContext()
 	if err != nil {
+		// Logged at error: a persistent failure here stops latestCtx advancing.
+		b.logger.Error("failed to refresh latest context", "error", err)
 		b.setLatestContext(sdk.Context{})
-		b.logger.Debug("failed to get latest context, notifying chain head", "error", err)
+		return nil
 	}
 	b.setLatestContext(latestCtx)
+
+	height := latestCtx.BlockHeight()
+	if height <= b.lastNotifiedHeight.Load() {
+		return nil // already notified for this height
+	}
+
 	header := b.CurrentBlock()
 	headerHash := header.Hash()
 
@@ -189,10 +230,12 @@ func (b *Blockchain) NotifyNewBlock() {
 		"block_hash", headerHash.Hex(),
 		"previous_hash", b.getPreviousHeaderHash().Hex())
 
+	b.lastNotifiedHeight.Store(height)
 	b.setPreviousHeaderHash(headerHash)
 	b.chainHeadFeed.Send(core.ChainHeadEvent{Header: header})
 
 	b.logger.Debug("chain head event sent to feed")
+	return header
 }
 
 // StateAt returns the StateDB object for a given block hash.

mempool/blockchain_test.go

@@ -1,12 +1,15 @@
 package mempool_test
 
 import (
+	"errors"
 	"math/big"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
@@ -130,3 +133,135 @@ func TestBlockchainRaceCondition(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, stateDB)
 }
+
+func TestNotifyNewBlockRefreshesContextEveryCall(t *testing.T) {
+	// ignore "already set": another test in the package may have configured it
+	_ = vmtypes.SetChainConfig(vmtypes.DefaultChainConfig(constants.EighteenDecimalsChainID))
+
+	var (
+		mu     sync.Mutex
+		height = int64(1)
+		mark   = "first"
+	)
+	set := func(h int64, m string) {
+		mu.Lock()
+		defer mu.Unlock()
+		height, mark = h, m
+	}
+	blockchain := newTestBlockchainWithGetter(t, func(int64, bool) (sdk.Context, error) {
+		mu.Lock()
+		defer mu.Unlock()
+		// BlockHeight() reads header.Height, so the height goes in the header.
+		return createMockContext().
+			WithBlockHeader(cmtproto.Header{Height: height, AppHash: []byte(mark)}), nil
+	})
+	events := make(chan core.ChainHeadEvent, 8)
+	sub := blockchain.SubscribeChainHeadEvent(events)
+	defer sub.Unsubscribe()
+
+	appHash := func() string {
+		ctx, err := blockchain.GetLatestContext()
+		require.NoError(t, err)
+		return string(ctx.BlockHeader().AppHash)
+	}
+
+	require.NotNil(t, blockchain.NotifyNewBlock(), "a new height should notify")
+	require.Len(t, events, 1)
+	require.Equal(t, "first", appHash())
+
+	// Same height, different context: no second event, but the pin still has
+	// to move -- this is the property the fix turns on.
+	set(1, "second")
+	require.Nil(t, blockchain.NotifyNewBlock(), "a repeated height must not notify")
+	require.Len(t, events, 1)
+	require.Equal(t, "second", appHash(), "the pin must refresh even when the event is skipped")
+
+	set(2, "third")
+	require.NotNil(t, blockchain.NotifyNewBlock(), "a later height should notify again")
+	require.Len(t, events, 2)
+	require.Equal(t, "third", appHash())
+}
+
+func TestNotifyNewBlockDedupsConcurrentDrivers(t *testing.T) {
+	// ignore "already set": another test in the package may have configured it
+	_ = vmtypes.SetChainConfig(vmtypes.DefaultChainConfig(constants.EighteenDecimalsChainID))
+
+	blockchain := newTestBlockchainWithGetter(t, func(int64, bool) (sdk.Context, error) {
+		return createMockContext().
+			WithBlockHeader(cmtproto.Header{Height: 1, AppHash: []byte("head")}), nil
+	})
+	events := make(chan core.ChainHeadEvent, 16)
+	sub := blockchain.SubscribeChainHeadEvent(events)
+	defer sub.Unsubscribe()
+
+	var (
+		wg       sync.WaitGroup
+		notified atomic.Int64
+	)
+	for range 8 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if blockchain.NotifyNewBlock() != nil {
+				notified.Add(1)
+			}
+		}()
+	}
+	wg.Wait()
+
+	require.Equal(t, int64(1), notified.Load(), "exactly one driver should notify a height")
+	require.Len(t, events, 1)
+}
+
+func TestNotifyNewBlockRecoversFromContextError(t *testing.T) {
+	var (
+		mu      sync.Mutex
+		failing = true
+	)
+	blockchain := newTestBlockchainWithGetter(t, func(int64, bool) (sdk.Context, error) {
+		mu.Lock()
+		defer mu.Unlock()
+		if failing {
+			return sdk.Context{}, errors.New("no context")
+		}
+		return createMockContext().
+			WithBlockHeader(cmtproto.Header{Height: 1, AppHash: []byte("healed")}), nil
+	})
+
+	require.Nil(t, blockchain.NotifyNewBlock(), "a failed refresh should not announce a head")
+
+	mu.Lock()
+	failing = false
+	mu.Unlock()
+
+	ctx, err := blockchain.GetLatestContext()
+	require.NoError(t, err)
+	require.Equal(t, "healed", string(ctx.BlockHeader().AppHash))
+}
+
+func TestNotifyNewBlockSkipsOnHeightHint(t *testing.T) {
+	var built, height atomic.Int64
+	height.Store(1)
+	blockchain := newTestBlockchainWithGetter(t, func(int64, bool) (sdk.Context, error) {
+		built.Add(1)
+		return createMockContext().
+			WithBlockHeader(cmtproto.Header{Height: height.Load(), AppHash: []byte("head")}), nil
+	})
+
+	require.NotNil(t, blockchain.NotifyNewBlockAt(1))
+	afterFirst := built.Load()
+
+	// A driver naming a height already notified must not build a context.
+	require.Nil(t, blockchain.NotifyNewBlockAt(1))
+	require.Equal(t, afterFirst, built.Load(), "the hint should short-circuit before newLatestContext")
+
+	// The backstop still works when the other driver has stopped: an unnotified
+	// height falls through and refreshes.
+	height.Store(2)
+	require.NotNil(t, blockchain.NotifyNewBlockAt(2))
+	require.Greater(t, built.Load(), afterFirst)
+
+	// A caller that cannot name its height is never skipped early.
+	height.Store(3)
+	require.NotNil(t, blockchain.NotifyNewBlock())
+}

mempool/mempool.go

@@ -532,10 +532,19 @@ func (m *Mempool) SetEventBus(eventBus *cmttypes.EventBus) {
 		panic(err)
 	}
 	m.eventBusWG.Go(func() {
+		// Nothing re-subscribes once this returns, but PrepareCheckState also
+		// drives NotifyNewBlock every block, so the caches keep advancing.
+		// Info level: this also fires on a clean Close or re-subscribe.
+		defer func() {
+			m.logger.Info("block header subscription ended", "reason", sub.Err())
+		}()
 		for {
 			select {
-			case <-sub.Out():
-				m.NotifyNewBlock()
+			case msg := <-sub.Out():
+				// The event names the block that just committed, a failed
+				// assertion leaves the zero height, which never skips.
+				ev, _ := msg.Data().(cmttypes.EventDataNewBlockHeader)
+				m.NotifyNewBlockAt(ev.Header.Height)
 			case <-sub.Canceled():
 				// Unsubscribe/Close cancels the subscription; Out() is never
 				// closed by CometBFT, so exit on cancellation to avoid leaking.
@@ -548,8 +557,17 @@ func (m *Mempool) SetEventBus(eventBus *cmttypes.EventBus) {
 // NotifyNewBlock manually notifies that there has been a new block produced
 // and it should update its internal data structures.
 func (m *Mempool) NotifyNewBlock() {
-	m.blockchain.NotifyNewBlock()
-	m.recheckCosmosPool.TriggerRecheck(m.blockchain.CurrentBlock())
+	m.NotifyNewBlockAt(0)
+}
+
+// NotifyNewBlockAt is NotifyNewBlock for drivers that know which height just
+// committed, see Blockchain.NotifyNewBlockAt.
+func (m *Mempool) NotifyNewBlockAt(committedHeight int64) {
+	// only recheck when the height advanced — a same-head trigger would cancel
+	// and restart the in-flight pass — and reuse the header it built
+	if header := m.blockchain.NotifyNewBlockAt(committedHeight); header != nil {
+		m.recheckCosmosPool.TriggerRecheck(header)
+	}
 }
 
 // HasEventBus returns true if the blockchain is configured to use an event bus for block notifications.