diff --git a/CHANGELOG.md b/CHANGELOG.md index af0680f97..d1fe7f98f 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/evmd/mempool.go b/evmd/mempool.go index 55986ce2a..53fe8efb6 100644 --- a/evmd/mempool.go +++ b/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 diff --git a/mempool/blockchain.go b/mempool/blockchain.go index 2c49042d5..225f4e60f 100644 --- a/mempool/blockchain.go +++ b/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. diff --git a/mempool/blockchain_test.go b/mempool/blockchain_test.go index ffc7f25d7..81930d54f 100644 --- a/mempool/blockchain_test.go +++ b/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()) +} diff --git a/mempool/mempool.go b/mempool/mempool.go index 180127ab2..06e533e5f 100644 --- a/mempool/mempool.go +++ b/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.