feat(mempool): proxy app mempool implementation

baseapp/abci.go

@@ -25,6 +25,7 @@ import (
 	"github.com/cosmos/cosmos-sdk/telemetry"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
+	"github.com/cosmos/cosmos-sdk/types/mempool"
 )
 
 // Supported ABCI Query prefixes and paths
@@ -965,6 +966,10 @@ func (app *BaseApp) Commit() (*abci.ResponseCommit, error) {
 		app.prepareCheckStater(app.checkState.Context())
 	}
 
+	if promoter, ok := app.mempool.(mempool.QueuePromoter); ok {
+		promoter.PromoteQueued(app.checkState.Context())
+	}
+
 	// The SnapshotIfApplicable method will create the snapshot by starting the goroutine
 	app.snapshotManager.SnapshotIfApplicable(header.Height)
 

baseapp/baseapp.go

@@ -10,6 +10,7 @@ import (
 	"github.com/cockroachdb/errors"
 	abci "github.com/cometbft/cometbft/abci/types"
 	"github.com/cometbft/cometbft/crypto/tmhash"
+	cmtmempool "github.com/cometbft/cometbft/mempool"
 	cmtproto "github.com/cometbft/cometbft/proto/tendermint/types"
 	dbm "github.com/cosmos/cosmos-db"
 	"github.com/cosmos/gogoproto/proto"
@@ -417,6 +418,19 @@ func (app *BaseApp) Mempool() mempool.Mempool {
 	return app.mempool
 }
 
+// ConnectMempoolEvents implements servertypes.MempoolEventConnector.
+func (app *BaseApp) ConnectMempoolEvents(eventCh chan cmtmempool.AppMempoolEvent) {
+	if receiver, ok := app.mempool.(mempool.EventReceiver); ok {
+		receiver.SetEventCh(eventCh)
+	}
+}
+
+// GetContextForSimulate returns a non-mutating context derived from the latest
+// committed state, suitable for transaction simulation (e.g. mempool cleanup).
+func (app *BaseApp) GetContextForSimulate(txBytes []byte) sdk.Context {
+	return app.getContextForTx(execModeSimulate, txBytes)
+}
+
 // Init initializes the app. It seals the app, preventing any
 // further modifications. In addition, it validates the app against
 // the earlier provided settings. Returns an error if validation fails.
@@ -979,7 +993,8 @@ func (app *BaseApp) runTx(mode execMode, txBytes []byte) (gInfo sdk.GasInfo, res
 
 	switch mode {
 	case execModeCheck:
-		if err := app.mempool.Insert(mempoolCtx, tx); err != nil {
+		insertCtx := context.WithValue(mempoolCtx, mempool.TxBytesContextKey{}, txBytes)
+		if err := app.mempool.Insert(insertCtx, tx); err != nil {
 			return gInfo, nil, anteEvents, err
 		}
 

server/start.go

@@ -16,6 +16,7 @@ import (
 	cmtcmd "github.com/cometbft/cometbft/cmd/cometbft/commands"
 	cmtcfg "github.com/cometbft/cometbft/config"
 	cmtjson "github.com/cometbft/cometbft/libs/json"
+	cmtmempool "github.com/cometbft/cometbft/mempool"
 	"github.com/cometbft/cometbft/node"
 	"github.com/cometbft/cometbft/p2p"
 	pvm "github.com/cometbft/cometbft/privval"
@@ -327,6 +328,8 @@ func startInProcess(svrCtx *Context, svrCfg serverconfig.Config, clientCtx clien
 		}
 		defer cleanupFn()
 
+		connectMempoolEvents(tmNode.Mempool(), app)
+
 		// Add the tx service to the gRPC router. We only need to register this
 		// service if API or gRPC is enabled, and avoid doing so in the general
 		// case, because it spawns a new local CometBFT RPC client.
@@ -403,6 +406,20 @@ func startCmtNode(
 	return tmNode, cleanupFn, nil
 }
 
+// connectMempoolEvents wires the CometBFT mempool event channel to the
+// application if both sides support it.
+func connectMempoolEvents(mp cmtmempool.Mempool, app types.Application) {
+	ep, ok := mp.(cmtmempool.EventProvider)
+	if !ok {
+		return
+	}
+	connector, ok := app.(types.MempoolEventConnector)
+	if !ok {
+		return
+	}
+	connector.ConnectMempoolEvents(ep.AppEventCh())
+}
+
 func getAndValidateConfig(svrCtx *Context) (serverconfig.Config, error) {
 	config, err := serverconfig.GetConfig(svrCtx.Viper)
 	if err != nil {

server/types/app.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"io"
 
+	cmtmempool "github.com/cometbft/cometbft/mempool"
 	cmtproto "github.com/cometbft/cometbft/proto/tendermint/types"
 	cmttypes "github.com/cometbft/cometbft/types"
 	dbm "github.com/cosmos/cosmos-db"
@@ -84,6 +85,14 @@ type (
 		ConsensusParams cmtproto.ConsensusParams
 	}
 
+	// MempoolEventConnector is optionally implemented by applications that
+	// receive CometBFT  ProxyMempool event channel after node startup.
+	// The app uses this channel to push EventTxInserted and EventTxRemoved
+	// events so the CometBFT reactor can manage gossip and tx caching.
+	MempoolEventConnector interface {
+		ConnectMempoolEvents(eventCh chan cmtmempool.AppMempoolEvent)
+	}
+
 	// AppExporter is a function that dumps all app state to
 	// JSON-serializable structure and returns the current validator set.
 	AppExporter func(

simapp/app.go

@@ -32,6 +32,7 @@ import (
 	upgradekeeper "cosmossdk.io/x/upgrade/keeper"
 	upgradetypes "cosmossdk.io/x/upgrade/types"
 	abci "github.com/cometbft/cometbft/abci/types"
+
 	dbm "github.com/cosmos/cosmos-db"
 	"github.com/cosmos/gogoproto/proto"
 	"github.com/spf13/cast"
@@ -53,6 +54,7 @@ import (
 	"github.com/cosmos/cosmos-sdk/std"
 	testdata_pulsar "github.com/cosmos/cosmos-sdk/testutil/testdata/testpb"
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	sdkmempool "github.com/cosmos/cosmos-sdk/types/mempool"
 	"github.com/cosmos/cosmos-sdk/types/module"
 	"github.com/cosmos/cosmos-sdk/types/msgservice"
 	sigtypes "github.com/cosmos/cosmos-sdk/types/tx/signing"
@@ -249,6 +251,18 @@ func NewSimApp(
 	}
 	baseAppOptions = append(baseAppOptions, voteExtOp, baseapp.SetOptimisticExecution())
 
+	// create the ProxyAppMempool and set it as a BaseApp option so it is used
+	// before the default proposal handler is created in NewBaseApp.
+	proxyAppMempool := sdkmempool.NewProxyAppMempool(
+		sdkmempool.ProxyAppMempoolConfig{
+			MaxTxsPerSender: 16,
+			MaxTotalTxs:     500,
+		},
+		nil,
+		txConfig.TxEncoder(),
+	)
+	baseAppOptions = append(baseAppOptions, baseapp.SetMempool(proxyAppMempool))
+
 	bApp := baseapp.NewBaseApp(appName, logger, db, txConfig.TxDecoder(), baseAppOptions...)
 	bApp.SetCommitMultiStoreTracer(traceStore)
 	bApp.SetVersion(version.Version)
@@ -288,6 +302,9 @@ func NewSimApp(
 	// add keepers
 	app.AccountKeeper = authkeeper.NewAccountKeeper(appCodec, runtime.NewKVStoreService(keys[authtypes.StoreKey]), authtypes.ProtoBaseAccount, maccPerms, authcodec.NewBech32Codec(sdk.Bech32MainPrefix), sdk.Bech32MainPrefix, authtypes.NewModuleAddress(govtypes.ModuleName).String())
 
+	// wire the AccountKeeper
+	proxyAppMempool.SetAccountKeeper(app.AccountKeeper)
+
 	app.BankKeeper = bankkeeper.NewBaseKeeper(
 		appCodec,
 		runtime.NewKVStoreService(keys[banktypes.StoreKey]),
@@ -565,7 +582,8 @@ func NewSimApp(
 }
 
 func (app *SimApp) setAnteHandler(txConfig client.TxConfig) {
-	anteHandler, err := NewAnteHandler(
+	// Full ante handler for PrepareProposal/ProcessProposal/FinalizeBlock
+	fullHandler, err := NewAnteHandler(
 		HandlerOptions{
 			ante.HandlerOptions{
 				AccountKeeper:   app.AccountKeeper,
@@ -581,8 +599,17 @@ func (app *SimApp) setAnteHandler(txConfig client.TxConfig) {
 		panic(err)
 	}
 
-	// Set the AnteHandler for the app
-	app.SetAnteHandler(anteHandler)
+	minimalHandler, err := ante.NewMinimalAnteHandler(ante.MinimalHandlerOptions{
+		AccountKeeper:   app.AccountKeeper,
+		SignModeHandler: txConfig.SignModeHandler(),
+		SigGasConsumer:  ante.DefaultSigVerificationGasConsumer,
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	dualHandler := ante.NewDualAnteHandler(minimalHandler, fullHandler)
+	app.SetAnteHandler(dualHandler)
 }
 
 func (app *SimApp) setPostHandler() {