refactor(mempool)!: match server/v2/cometbft and sdk mempool interface

baseapp/abci_utils.go

@@ -264,19 +264,25 @@ func (h *DefaultProposalHandler) PrepareProposalHandler() sdk.PrepareProposalHan
 
 		defer h.txSelector.Clear()
 
+		// decode transactions
+		decodedTxs := make([]sdk.Tx, len(req.Txs))
+		for i, txBz := range req.Txs {
+			tx, err := h.txVerifier.TxDecode(txBz)
+			if err != nil {
+				return nil, err
+			}
+
+			decodedTxs[i] = tx
+		}
+
 		// If the mempool is nil or NoOp we simply return the transactions
 		// requested from CometBFT, which, by default, should be in FIFO order.
 		//
 		// Note, we still need to ensure the transactions returned respect req.MaxTxBytes.
 		_, isNoOp := h.mempool.(mempool.NoOpMempool)
 		if h.mempool == nil || isNoOp {
-			for _, txBz := range req.Txs {
-				tx, err := h.txVerifier.TxDecode(txBz)
-				if err != nil {
-					return nil, err
-				}
-
-				stop := h.txSelector.SelectTxForProposal(ctx, uint64(req.MaxTxBytes), maxBlockGas, tx, txBz)
+			for _, tx := range decodedTxs {
+				stop := h.txSelector.SelectTxForProposal(ctx, uint64(req.MaxTxBytes), maxBlockGas, tx, tx.Bytes())
 				if stop {
 					break
 				}
@@ -291,7 +297,7 @@ func (h *DefaultProposalHandler) PrepareProposalHandler() sdk.PrepareProposalHan
 			selectedTxsNums int
 			invalidTxs      []sdk.Tx // invalid txs to be removed out of the loop to avoid dead lock
 		)
-		h.mempool.SelectBy(ctx, req.Txs, func(memTx sdk.Tx) bool {
+		h.mempool.SelectBy(ctx, decodedTxs, func(memTx sdk.Tx) bool {
 			unorderedTx, ok := memTx.(sdk.TxWithUnordered)
 			isUnordered := ok && unorderedTx.GetUnordered()
 			txSignersSeqs := make(map[string]uint64)

server/v2/cometbft/abci.go

@@ -471,9 +471,10 @@ func (c *Consensus[T]) FinalizeBlock(
 	}
 
 	// remove txs from the mempool
-	err = c.mempool.Remove(decodedTxs)
-	if err != nil {
-		return nil, fmt.Errorf("unable to remove txs: %w", err)
+	for _, tx := range decodedTxs {
+		if err = c.mempool.Remove(tx); err != nil {
+			return nil, fmt.Errorf("unable to remove txs: %w", err)
+		}
 	}
 
 	c.lastCommittedHeight.Store(req.Height)

server/v2/cometbft/config.go

@@ -2,6 +2,8 @@ package cometbft
 
 import (
 	cmtcfg "github.com/cometbft/cometbft/config"
+
+	"cosmossdk.io/server/v2/cometbft/mempool"
 )
 
 // Config is the configuration for the CometBFT application
@@ -20,6 +22,7 @@ func DefaultAppTomlConfig() *AppTomlConfig {
 		Transport:       "socket",
 		Trace:           false,
 		Standalone:      false,
+		Mempool:         mempool.DefaultConfig(),
 	}
 }
 
@@ -32,6 +35,9 @@ type AppTomlConfig struct {
 	Transport       string   `mapstructure:"transport" toml:"transport" comment:"transport defines the CometBFT RPC server transport protocol: socket, grpc"`
 	Trace           bool     `mapstructure:"trace" toml:"trace" comment:"trace enables the CometBFT RPC server to output trace information about its internal operations."`
 	Standalone      bool     `mapstructure:"standalone" toml:"standalone" comment:"standalone starts the application without the CometBFT node. The node should be started separately."`
+
+	// Sub configs
+	Mempool mempool.Config `mapstructure:"mempool" toml:"mempool" comment:"mempool defines the configuration for the SDK built-in app-side mempool implementations."`
 }
 
 // CfgOption is a function that allows to overwrite the default server configuration.

server/v2/cometbft/flags.go

@@ -51,10 +51,11 @@ func prefix(f string) string {
 
 // Server flags
 var (
-	Standalone     = prefix("standalone")
-	FlagAddress    = prefix("address")
-	FlagTransport  = prefix("transport")
-	FlagHaltHeight = prefix("halt-height")
-	FlagHaltTime   = prefix("halt-time")
-	FlagTrace      = prefix("trace")
+	Standalone        = prefix("standalone")
+	FlagAddress       = prefix("address")
+	FlagTransport     = prefix("transport")
+	FlagHaltHeight    = prefix("halt-height")
+	FlagHaltTime      = prefix("halt-time")
+	FlagTrace         = prefix("trace")
+	FlagMempoolMaxTxs = prefix("mempool.max-txs")
 )

server/v2/cometbft/handlers/defaults.go

@@ -79,7 +79,7 @@ func (h *DefaultProposalHandler[T]) PrepareHandler() PrepareHandler[T] {
 			// check again.
 			_, err := app.ValidateTx(ctx, memTx)
 			if err != nil {
-				err := h.mempool.Remove([]T{memTx})
+				err := h.mempool.Remove(memTx)
 				if err != nil && !errors.Is(err, mempool.ErrTxNotFound) {
 					return nil, err
 				}

server/v2/cometbft/internal/mock/mock_mempool.go

@@ -15,5 +15,6 @@ type MockMempool[T transaction.Tx] struct{}
 
 func (MockMempool[T]) Insert(context.Context, T) error                 { return nil }
 func (MockMempool[T]) Select(context.Context, []T) mempool.Iterator[T] { return nil }
+func (MockMempool[T]) SelectBy(context.Context, []T, func(T) bool)     {}
 func (MockMempool[T]) CountTx() int                                    { return 0 }
-func (MockMempool[T]) Remove([]T) error                                { return nil }
+func (MockMempool[T]) Remove(T) error                                  { return nil }

server/v2/cometbft/mempool/config.go

@@ -1,11 +1,16 @@
 package mempool
 
-// Config defines the configurations for the SDK built-in app-side mempool
-// implementations.
+var DefaultMaxTx = -1
+
+// Config defines the configurations for the SDK built-in app-side mempool implementations.
 type Config struct {
-	// MaxTxs defines the behavior of the mempool. A negative value indicates
-	// the mempool is disabled entirely, zero indicates that the mempool is
-	// unbounded in how many txs it may contain, and a positive value indicates
-	// the maximum amount of txs it may contain.
-	MaxTxs int `mapstructure:"max-txs"`
+	// MaxTxs defines the maximum number of transactions that can be in the mempool.
+	MaxTxs int `mapstructure:"max-txs" toml:"max-txs" comment:"max-txs defines the maximum number of transactions that can be in the mempool. A value of 0 indicates an unbounded mempool, a negative value disables the app-side mempool."`
+}
+
+// DefaultConfig returns a default configuration for the SDK built-in app-side mempool implementations.
+func DefaultConfig() Config {
+	return Config{
+		MaxTxs: DefaultMaxTx,
+	}
 }

server/v2/cometbft/mempool/doc.go

@@ -1,6 +1,2 @@
-/*
-The mempool package defines a few mempool services which can be used in conjunction with your consensus implementation
-
-*/
-
+// Package mempool defines a few mempool services which can be used in conjunction with your consensus implementation.
 package mempool

server/v2/cometbft/mempool/mempool.go

@@ -19,13 +19,18 @@ type Mempool[T transaction.Tx] interface {
 	Insert(context.Context, T) error
 
 	// Select returns an Iterator over the app-side mempool. If txs are specified,
-	// then they shall be incorporated into the Iterator. The Iterator must be
-	// closed by the caller.
+	// then they shall be incorporated into the Iterator. The Iterator is not thread-safe to use.
 	Select(context.Context, []T) Iterator[T]
 
+	// SelectBy use callback to iterate over the mempool, it's thread-safe to use.
+	SelectBy(context.Context, []T, func(T) bool)
+
+	// CountTx returns the number of transactions currently in the mempool.
+	CountTx() int
+
 	// Remove attempts to remove a transaction from the mempool, returning an error
 	// upon failure.
-	Remove([]T) error
+	Remove(T) error
 }
 
 // Iterator defines an app-side mempool iterator interface that is as minimal as

server/v2/cometbft/mempool/noop.go

@@ -4,8 +4,11 @@ import (
 	"context"
 
 	"cosmossdk.io/core/transaction"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
 )
 
+var _ Mempool[sdk.Tx] = (*NoOpMempool[sdk.Tx])(nil) // verify interface at compile time
 var _ Mempool[transaction.Tx] = (*NoOpMempool[transaction.Tx])(nil)
 
 // NoOpMempool defines a no-op mempool. Transactions are completely discarded and
@@ -16,7 +19,8 @@ var _ Mempool[transaction.Tx] = (*NoOpMempool[transaction.Tx])(nil)
 // is FIFO-ordered by default.
 type NoOpMempool[T transaction.Tx] struct{}
 
-func (NoOpMempool[T]) Insert(context.Context, T) error         { return nil }
-func (NoOpMempool[T]) Select(context.Context, []T) Iterator[T] { return nil }
-func (NoOpMempool[T]) CountTx() int                            { return 0 }
-func (NoOpMempool[T]) Remove([]T) error                        { return nil }
+func (NoOpMempool[T]) Insert(context.Context, T) error             { return nil }
+func (NoOpMempool[T]) Select(context.Context, []T) Iterator[T]     { return nil }
+func (NoOpMempool[T]) SelectBy(context.Context, []T, func(T) bool) {}
+func (NoOpMempool[T]) CountTx() int                                { return 0 }
+func (NoOpMempool[T]) Remove(T) error                              { return nil }

server/v2/cometbft/options.go

@@ -9,14 +9,15 @@ import (
 )
 
 // ServerOptions defines the options for the CometBFT server.
+// When an option takes a map[string]any, it can access the app.tom's cometbft section and the config.toml config.
 type ServerOptions[T transaction.Tx] struct {
-	Mempool                    mempool.Mempool[T]
 	PrepareProposalHandler     handlers.PrepareHandler[T]
 	ProcessProposalHandler     handlers.ProcessHandler[T]
 	VerifyVoteExtensionHandler handlers.VerifyVoteExtensionhandler
 	ExtendVoteHandler          handlers.ExtendVoteHandler
 
-	SnapshotOptions snapshots.SnapshotOptions
+	Mempool         func(cfg map[string]any) mempool.Mempool[T]
+	SnapshotOptions func(cfg map[string]any) snapshots.SnapshotOptions
 
 	AddrPeerFilter types.PeerFilter // filter peers by address and port
 	IdPeerFilter   types.PeerFilter // filter peers by node ID
@@ -26,12 +27,12 @@ type ServerOptions[T transaction.Tx] struct {
 // It defaults to a NoOpMempool and NoOp handlers.
 func DefaultServerOptions[T transaction.Tx]() ServerOptions[T] {
 	return ServerOptions[T]{
-		Mempool:                    mempool.NoOpMempool[T]{},
 		PrepareProposalHandler:     handlers.NoOpPrepareProposal[T](),
 		ProcessProposalHandler:     handlers.NoOpProcessProposal[T](),
 		VerifyVoteExtensionHandler: handlers.NoOpVerifyVoteExtensionHandler(),
 		ExtendVoteHandler:          handlers.NoOpExtendVote(),
-		SnapshotOptions:            snapshots.NewSnapshotOptions(0, 0),
+		Mempool:                    func(cfg map[string]any) mempool.Mempool[T] { return mempool.NoOpMempool[T]{} },
+		SnapshotOptions:            func(cfg map[string]any) snapshots.SnapshotOptions { return snapshots.NewSnapshotOptions(0, 0) },
 		AddrPeerFilter:             nil,
 		IdPeerFilter:               nil,
 	}

server/v2/cometbft/server.go

@@ -24,6 +24,7 @@ import (
 	"cosmossdk.io/log"
 	serverv2 "cosmossdk.io/server/v2"
 	cometlog "cosmossdk.io/server/v2/cometbft/log"
+	"cosmossdk.io/server/v2/cometbft/mempool"
 	"cosmossdk.io/server/v2/cometbft/types"
 	"cosmossdk.io/store/v2/snapshots"
 
@@ -105,7 +106,7 @@ func (s *CometBFTServer[T]) Init(appI serverv2.AppI[T], cfg map[string]any, logg
 		appI.Name(),
 		appI.GetConsensusAuthority(),
 		appI.GetAppManager(),
-		s.serverOptions.Mempool,
+		s.serverOptions.Mempool(cfg),
 		indexEvents,
 		appI.GetGPRCMethodsToMessageMap(),
 		store,
@@ -127,7 +128,7 @@ func (s *CometBFTServer[T]) Init(appI serverv2.AppI[T], cfg map[string]any, logg
 	if err != nil {
 		return err
 	}
-	consensus.snapshotManager = snapshots.NewManager(snapshotStore, s.serverOptions.SnapshotOptions, sc, ss, nil, s.logger)
+	consensus.snapshotManager = snapshots.NewManager(snapshotStore, s.serverOptions.SnapshotOptions(cfg), sc, ss, nil, s.logger)
 
 	s.Consensus = consensus
 
@@ -240,6 +241,7 @@ func (s *CometBFTServer[T]) StartCmdFlags() *pflag.FlagSet {
 	flags.Uint64(FlagHaltTime, 0, "Minimum block time (in Unix seconds) at which to gracefully halt the chain and shutdown the node")
 	flags.Bool(FlagTrace, false, "Provide full stack traces for errors in ABCI Log")
 	flags.Bool(Standalone, false, "Run app without CometBFT")
+	flags.Int(FlagMempoolMaxTxs, mempool.DefaultMaxTx, "Sets MaxTx value for the app-side mempool")
 
 	// add comet flags, we use an empty command to avoid duplicating CometBFT's AddNodeFlags.
 	// we can then merge the flag sets.

server/v2/commands.go

@@ -38,14 +38,14 @@ func AddCommands[T transaction.Tx](
 	rootCmd *cobra.Command,
 	newApp AppCreator[T],
 	logger log.Logger,
-	serverCfg ServerConfig,
+	globalServerCfg ServerConfig,
 	components ...ServerComponent[T],
 ) error {
 	if len(components) == 0 {
 		return errors.New("no components provided")
 	}
 
-	server := NewServer(logger, serverCfg, components...)
+	server := NewServer(logger, globalServerCfg, components...)
 	originalPersistentPreRunE := rootCmd.PersistentPreRunE
 	rootCmd.PersistentPreRunE = func(cmd *cobra.Command, args []string) error {
 		// set the default command outputs

simapp/v2/go.mod

@@ -35,6 +35,7 @@ require (
 	github.com/cosmos/cosmos-db v1.0.3-0.20240911104526-ddc3f09bfc22 // indirect
 	// this version is not used as it is always replaced by the latest Cosmos SDK version
 	github.com/cosmos/cosmos-sdk v0.53.0
+	github.com/spf13/cast v1.7.0 // indirect
 	github.com/spf13/cobra v1.8.1
 	github.com/spf13/pflag v1.0.5
 	github.com/spf13/viper v1.19.0
@@ -193,7 +194,6 @@ require (
 	github.com/sasha-s/go-deadlock v0.3.5 // indirect
 	github.com/sourcegraph/conc v0.3.0 // indirect
 	github.com/spf13/afero v1.11.0 // indirect
-	github.com/spf13/cast v1.7.0 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/supranational/blst v0.3.13 // indirect
 	github.com/syndtr/goleveldb v1.0.1-0.20220721030215-126854af5e6d // indirect

simapp/v2/simdv2/cmd/commands.go

@@ -74,7 +74,11 @@ func initRootCmd[T transaction.Tx](
 		newApp,
 		logger,
 		initServerConfig(),
-		cometbft.New(&genericTxDecoder[T]{txConfig}, cometbft.DefaultServerOptions[T]()),
+		cometbft.New(
+			&genericTxDecoder[T]{txConfig},
+			initCometOptions[T](),
+			initCometConfig(),
+		),
 		grpc.New[T](),
 		store.New[T](newApp),
 	); err != nil {

simapp/v2/simdv2/cmd/config.go

@@ -2,8 +2,13 @@ package cmd
 
 import (
 	"strings"
+	"time"
 
+	cmtcfg "github.com/cometbft/cometbft/config"
+
+	"cosmossdk.io/core/transaction"
 	serverv2 "cosmossdk.io/server/v2"
+	"cosmossdk.io/server/v2/cometbft"
 
 	clientconfig "github.com/cosmos/cosmos-sdk/client/config"
 	"github.com/cosmos/cosmos-sdk/crypto/keyring"
@@ -68,3 +73,34 @@ func initServerConfig() serverv2.ServerConfig {
 
 	return serverCfg
 }
+
+// initCometConfig helps to override default comet config template and configs.
+func initCometConfig() cometbft.CfgOption {
+	cfg := cmtcfg.DefaultConfig()
+
+	// display only warn logs by default except for p2p and state
+	cfg.LogLevel = "*:warn,p2p:info,state:info"
+	// increase block timeout
+	cfg.Consensus.TimeoutCommit = 5 * time.Second
+	// overwrite default pprof listen address
+	cfg.RPC.PprofListenAddress = "localhost:6060"
+
+	return cometbft.OverwriteDefaultConfigTomlConfig(cfg)
+}
+
+func initCometOptions[T transaction.Tx]() cometbft.ServerOptions[T] {
+	serverOptions := cometbft.DefaultServerOptions[T]()
+
+	// overwrite app mempool, using max-txs option
+	// serverOptions.Mempool = func(cfg map[string]any) mempool.Mempool[T] {
+	// 	if maxTxs := cast.ToInt(cfg[cometbft.FlagMempoolMaxTxs]); maxTxs >= 0 {
+	// 		return sdkmempool.NewSenderNonceMempool(
+	// 			sdkmempool.SenderNonceMaxTxOpt(maxTxs),
+	// 		)
+	// 	}
+
+	// 	return mempool.NoOpMempool[T]{}
+	// }
+
+	return serverOptions
+}

types/mempool/mempool.go

@@ -14,10 +14,10 @@ type Mempool interface {
 
 	// Select returns an Iterator over the app-side mempool. If txs are specified,
 	// then they shall be incorporated into the Iterator. The Iterator is not thread-safe to use.
-	Select(context.Context, [][]byte) Iterator
+	Select(context.Context, []sdk.Tx) Iterator
 
 	// SelectBy use callback to iterate over the mempool, it's thread-safe to use.
-	SelectBy(context.Context, [][]byte, func(sdk.Tx) bool)
+	SelectBy(context.Context, []sdk.Tx, func(sdk.Tx) bool)
 
 	// CountTx returns the number of transactions currently in the mempool.
 	CountTx() int

types/mempool/noop.go

@@ -17,7 +17,7 @@ var _ Mempool = (*NoOpMempool)(nil)
 type NoOpMempool struct{}
 
 func (NoOpMempool) Insert(context.Context, sdk.Tx) error                  { return nil }
-func (NoOpMempool) Select(context.Context, [][]byte) Iterator             { return nil }
-func (NoOpMempool) SelectBy(context.Context, [][]byte, func(sdk.Tx) bool) {}
+func (NoOpMempool) Select(context.Context, []sdk.Tx) Iterator             { return nil }
+func (NoOpMempool) SelectBy(context.Context, []sdk.Tx, func(sdk.Tx) bool) {}
 func (NoOpMempool) CountTx() int                                          { return 0 }
 func (NoOpMempool) Remove(sdk.Tx) error                                   { return nil }