fix: notify new block for mempool in time to avoid insufficient funds error (cosmos#471)

* fix: avoid insufficient funds error when fund and send tx within same block

get SpendableCoin from sdk ctx to reflect uncommitted balance change

* wrong hash

* fix test

* cleanup doc

* notify

* revert

* cleanup

* doc

* unsubscribe

* lint

* cleanup

* align shutdown timeout to avoid stuck on closing JSON-RPC

* cleanup

* Apply suggestions from code review

* Apply suggestions from code review

* lint

* lint

---------

Co-authored-by: Alex | Interchain Labs <[email protected]>

Author: mmsqe

CHANGELOG.md

@@ -6,10 +6,12 @@
 
 ### BUG FIXES
 
+- [\#471](https://github.com/cosmos/evm/pull/471) Notify new block for mempool in time.
+
 ### IMPROVEMENTS
 
+- [\#467](https://github.com/cosmos/evm/pull/467) Replace GlobalEVMMempool by passing to JSONRPC on initiate.
 - [\#352](https://github.com/cosmos/evm/pull/352) Remove the creation of a Geth EVM instance, stateDB during the AnteHandler balance check.
-- [\#467](https://github.com/cosmos/evm/pull/467) Ensure SetGlobalEVMMempool is thread-safe and only sets global mempool instance once.
 
 ### FEATURES
 

evmd/ante/evm_handler.go

@@ -18,6 +18,6 @@ func newMonoEVMAnteHandler(options ante.HandlerOptions) sdk.AnteHandler {
 		),
 		ante.NewTxListenerDecorator(options.PendingTxListener),
 	}
-	
+
 	return sdk.ChainAnteDecorators(decorators...)
 }

evmd/app.go

@@ -2,6 +2,7 @@ package evmd
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -1145,6 +1146,22 @@ func (app *EVMD) SetClientCtx(clientCtx client.Context) {
 	app.clientCtx = clientCtx
 }
 
+// Close unsubscribes from the CometBFT event bus (if set) and closes the underlying BaseApp.
+func (app *EVMD) Close() error {
+	var err error
+	if m, ok := app.GetMempool().(*evmmempool.ExperimentalEVMMempool); ok {
+		err = m.Close()
+	}
+	err = errors.Join(err, app.BaseApp.Close())
+	msg := "Application gracefully shutdown"
+	if err == nil {
+		app.Logger().Info(msg)
+	} else {
+		app.Logger().Error(msg, "error", err)
+	}
+	return err
+}
+
 // AutoCliOpts returns the autocli options for the app.
 func (app *EVMD) AutoCliOpts() autocli.AppOptions {
 	modules := make(map[string]appmodule.AppModule, 0)

mempool/README.md

@@ -82,11 +82,6 @@ if evmtypes.GetChainConfig() != nil {
     )
     app.EVMMempool = evmMempool
 
-    // Set the global mempool for RPC access
-    if err := evmmempool.SetGlobalEVMMempool(evmMempool); err != nil {
-        panic(err)
-    }
-    
     // Replace BaseApp mempool
     app.SetMempool(evmMempool)
 
@@ -103,6 +98,23 @@ if evmtypes.GetChainConfig() != nil {
     )
     app.SetPrepareProposal(abciProposalHandler.PrepareProposalHandler())
 }
+
+// Close unsubscribes from the CometBFT event bus (if set) and closes the underlying BaseApp.
+func (app *EVMD) Close() error {
+	var err error
+	if m, ok := app.GetMempool().(*evmmempool.ExperimentalEVMMempool); ok {
+		err = m.Close()
+	}
+	err = errors.Join(err, app.BaseApp.Close())
+	msg := "Application gracefully shutdown"
+	if err == nil {
+		app.Logger().Info(msg)
+	} else {
+		app.Logger().Error(msg, "error", err)
+	}
+	return err
+}
+
 ```
 
 ### Configuration Options
@@ -211,7 +223,7 @@ ERROR unable to publish transaction nonce=40 expected=12: invalid sequence
 ERROR unable to publish transaction nonce=41 expected=12: invalid sequence
 ```
 
-**Real-World Testing**: The [`tests/systemtests/Counter/script/SimpleSends.s.sol`](../../tests/systemtests/Counter/script/SimpleSends.s.sol) script demonstrates typical Ethereum tooling behavior - it sends 10 sequential transactions in a batch, which naturally arrive out of order and create nonce gaps. With the default Cosmos mempool, this script would fail with sequence errors. With the EVM mempool, all transactions are queued locally and promoted as gaps are filled, allowing the script to succeed.
+**Real-World Testing**: The [`tests/systemtests/Counter/script/SimpleSends.s.sol`](../tests/systemtests/Counter/script/SimpleSends.s.sol) script demonstrates typical Ethereum tooling behavior - it sends 10 sequential transactions in a batch, which naturally arrive out of order and create nonce gaps. With the default Cosmos mempool, this script would fail with sequence errors. With the EVM mempool, all transactions are queued locally and promoted as gaps are filled, allowing the script to succeed.
 
 ### Design Principles
 

mempool/iterator.go

@@ -340,17 +340,18 @@ func (i *EVMMempoolIterator) convertEVMToSDKTx(nextEVMTx *txpool.LazyTransaction
 	}
 
 	msgEthereumTx := &msgtypes.MsgEthereumTx{}
+	hash := nextEVMTx.Tx.Hash()
 	if err := msgEthereumTx.FromSignedEthereumTx(nextEVMTx.Tx, ethtypes.LatestSignerForChainID(i.chainID)); err != nil {
-		i.logger.Error("failed to convert signed Ethereum transaction", "error", err, "tx_hash", nextEVMTx.Tx.Hash().Hex())
+		i.logger.Error("failed to convert signed Ethereum transaction", "error", err, "tx_hash", hash)
 		return nil // Return nil for invalid tx instead of panicking
 	}
 
 	cosmosTx, err := msgEthereumTx.BuildTx(i.txConfig.NewTxBuilder(), i.bondDenom)
 	if err != nil {
-		i.logger.Error("failed to build Cosmos transaction from EVM transaction", "error", err, "tx_hash", nextEVMTx.Tx.Hash().Hex())
+		i.logger.Error("failed to build Cosmos transaction from EVM transaction", "error", err, "tx_hash", hash)
 		return nil
 	}
 
-	i.logger.Debug("successfully converted EVM transaction to Cosmos transaction", "tx_hash", nextEVMTx.Tx.Hash().Hex())
+	i.logger.Debug("successfully converted EVM transaction to Cosmos transaction", "tx_hash", hash)
 	return cosmosTx
 }

mempool/mempool.go

@@ -9,13 +9,15 @@ import (
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
 	"github.com/holiman/uint256"
 
+	cmttypes "github.com/cometbft/cometbft/types"
+
 	"github.com/cosmos/evm/mempool/miner"
 	"github.com/cosmos/evm/mempool/txpool"
 	"github.com/cosmos/evm/mempool/txpool/legacypool"
+	"github.com/cosmos/evm/rpc/stream"
 	"github.com/cosmos/evm/x/precisebank/types"
 	evmtypes "github.com/cosmos/evm/x/vm/types"
 
-	errorsmod "cosmossdk.io/errors"
 	"cosmossdk.io/log"
 	"cosmossdk.io/math"
 
@@ -27,6 +29,8 @@ import (
 
 var _ sdkmempool.ExtMempool = &ExperimentalEVMMempool{}
 
+const SubscriberName = "evm"
+
 type (
 	// ExperimentalEVMMempool is a unified mempool that manages both EVM and Cosmos SDK transactions.
 	// It provides a single interface for transaction insertion, selection, and removal while
@@ -54,6 +58,8 @@ type (
 
 		/** Concurrency **/
 		mtx sync.Mutex
+
+		eventBus *cmttypes.EventBus
 	}
 )
 
@@ -202,22 +208,18 @@ func (m *ExperimentalEVMMempool) Insert(goCtx context.Context, tx sdk.Tx) error
 	blockHeight := ctx.BlockHeight()
 
 	m.logger.Debug("inserting transaction into mempool", "block_height", blockHeight)
-
-	if blockHeight < 2 {
-		return errorsmod.Wrap(sdkerrors.ErrInvalidHeight, "Mempool is not ready. Please wait for block 1 to finalize.")
-	}
-
 	ethMsg, err := m.getEVMMessage(tx)
 	if err == nil {
 		// Insert into EVM pool
-		m.logger.Debug("inserting EVM transaction", "tx_hash", ethMsg.Hash)
+		hash := ethMsg.Hash()
+		m.logger.Debug("inserting EVM transaction", "tx_hash", hash)
 		ethTxs := []*ethtypes.Transaction{ethMsg.AsTransaction()}
 		errs := m.txPool.Add(ethTxs, true)
 		if len(errs) > 0 && errs[0] != nil {
-			m.logger.Error("failed to insert EVM transaction", "error", errs[0], "tx_hash", ethMsg.Hash)
+			m.logger.Error("failed to insert EVM transaction", "error", errs[0], "tx_hash", hash)
 			return errs[0]
 		}
-		m.logger.Debug("EVM transaction inserted successfully", "tx_hash", ethMsg.Hash)
+		m.logger.Debug("EVM transaction inserted successfully", "tx_hash", hash)
 		return nil
 	}
 
@@ -300,11 +302,12 @@ func (m *ExperimentalEVMMempool) Remove(tx sdk.Tx) error {
 		// We should not do this with EVM transactions because removing them causes the subsequent ones to
 		// be dequeued as temporarily invalid, only to be requeued a block later.
 		// The EVM mempool handles removal based on account nonce automatically.
+		hash := msg.Hash()
 		if m.shouldRemoveFromEVMPool(tx) {
-			m.logger.Debug("manually removing EVM transaction", "tx_hash", msg.Hash())
-			m.legacyTxPool.RemoveTx(msg.Hash(), false, true)
+			m.logger.Debug("manually removing EVM transaction", "tx_hash", hash)
+			m.legacyTxPool.RemoveTx(hash, false, true)
 		} else {
-			m.logger.Debug("skipping manual removal of EVM transaction, leaving to mempool to handle", "tx_hash", msg.Hash)
+			m.logger.Debug("skipping manual removal of EVM transaction, leaving to mempool to handle", "tx_hash", hash)
 		}
 		return nil
 	}
@@ -372,6 +375,31 @@ func (m *ExperimentalEVMMempool) SelectBy(goCtx context.Context, i [][]byte, f f
 	}
 }
 
+// SetEventBus sets CometBFT event bus to listen for new block header event.
+func (m *ExperimentalEVMMempool) SetEventBus(eventBus *cmttypes.EventBus) {
+	if m.eventBus != nil {
+		m.eventBus.Unsubscribe(context.Background(), SubscriberName, stream.NewBlockHeaderEvents) //nolint: errcheck
+	}
+	m.eventBus = eventBus
+	sub, err := eventBus.Subscribe(context.Background(), SubscriberName, stream.NewBlockHeaderEvents)
+	if err != nil {
+		panic(err)
+	}
+	go func() {
+		for range sub.Out() {
+			m.GetBlockchain().NotifyNewBlock()
+		}
+	}()
+}
+
+// Close unsubscribes from the CometBFT event bus.
+func (m *ExperimentalEVMMempool) Close() error {
+	if m.eventBus != nil {
+		return m.eventBus.Unsubscribe(context.Background(), SubscriberName, stream.NewBlockHeaderEvents)
+	}
+	return nil
+}
+
 // getEVMMessage validates that the transaction contains exactly one message and returns it if it's an EVM message.
 // Returns an error if the transaction has no messages, multiple messages, or the single message is not an EVM transaction.
 func (m *ExperimentalEVMMempool) getEVMMessage(tx sdk.Tx) (*evmtypes.MsgEthereumTx, error) {

rpc/backend/sign_tx.go

@@ -13,6 +13,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/signer/core/apitypes"
 
+	"github.com/cosmos/evm/mempool"
 	evmtypes "github.com/cosmos/evm/x/vm/types"
 
 	errorsmod "cosmossdk.io/errors"
@@ -109,7 +110,7 @@ func (b *Backend) SendTransaction(args evmtypes.TransactionArgs) (common.Hash, e
 	}
 	if err != nil {
 		// Check if this is a nonce gap error that was successfully queued
-		if strings.Contains(err.Error(), "tx nonce is higher than account nonce") {
+		if strings.Contains(err.Error(), mempool.ErrNonceGap.Error()) {
 			// Transaction was successfully queued due to nonce gap, return success to client
 			b.Logger.Debug("transaction queued due to nonce gap", "hash", txHash.Hex())
 			return txHash, nil

rpc/stream/rpc.go

@@ -40,8 +40,9 @@ var (
 		cmttypes.EventTx,
 		sdk.EventTypeMessage,
 		sdk.AttributeKeyModule, evmtypes.ModuleName)).String()
-	blockEvents  = cmttypes.QueryForEvent(cmttypes.EventNewBlock).String()
-	evmTxHashKey = fmt.Sprintf("%s.%s", evmtypes.TypeMsgEthereumTx, evmtypes.AttributeKeyEthereumTxHash)
+	blockEvents          = cmttypes.QueryForEvent(cmttypes.EventNewBlock).String()
+	evmTxHashKey         = fmt.Sprintf("%s.%s", evmtypes.TypeMsgEthereumTx, evmtypes.AttributeKeyEthereumTxHash)
+	NewBlockHeaderEvents = cmtquery.MustCompile(fmt.Sprintf("%s='%s'", cmttypes.EventTypeKey, cmttypes.EventNewBlockHeader))
 )
 
 type RPCHeader struct {

server/json_rpc.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log/slog"
 	"net/http"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	ethrpc "github.com/ethereum/go-ethereum/rpc"
@@ -24,6 +25,8 @@ import (
 	"github.com/cosmos/cosmos-sdk/server"
 )
 
+const shutdownTimeout = 5 * time.Second
+
 type AppWithPendingTxStream interface {
 	RegisterPendingTxListener(listener func(common.Hash))
 }
@@ -109,7 +112,9 @@ func StartJSONRPC(
 			// The calling process canceled or closed the provided context, so we must
 			// gracefully stop the JSON-RPC server.
 			logger.Info("stopping JSON-RPC server...", "address", config.JSONRPC.Address)
-			if err := httpSrv.Shutdown(context.Background()); err != nil {
+			ctxShutdown, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
+			defer cancel()
+			if err := httpSrv.Shutdown(ctxShutdown); err != nil {
 				logger.Error("failed to shutdown JSON-RPC server", "error", err.Error())
 			}
 			return nil

server/start.go

@@ -432,6 +432,9 @@ func startInProcess(svrCtx *server.Context, clientCtx client.Context, opts Start
 			return err
 		}
 
+		if m, ok := evmApp.GetMempool().(*evmmempool.ExperimentalEVMMempool); ok {
+			m.SetEventBus(tmNode.EventBus())
+		}
 		defer func() {
 			if tmNode.IsRunning() {
 				_ = tmNode.Stop()