feat: v0.34.x Prioritized Mempool (#8695)

* Updated mocks

* add reactor tests

* add v1 reactor tests

* Fix fuzz test for priority mempool

* e2e adapted to mempool v1; prio pool is default now

* Reverted default mempool to be fifo

* Changed buf version

* Added priority mempool to ci testnet

* Fixed linter

* Updated makefile

* Aligned makefile changes to v0.34.x

* Added go install for proto

* Add log message to warn about prioritized mempool bug

Signed-off-by: Thane Thomson <connect@thanethomson.com>

* Changelog message

Co-authored-by: Jasmina Malicevic <jasmina.dustinac@gmail.com>
Co-authored-by: Callum Waters <cmwaters19@gmail.com>
Co-authored-by: Sam Kleinman <garen@tychoish.com>
Co-authored-by: Thane Thomson <connect@thanethomson.com>

test/e2e/generator/generate.go

@@ -32,6 +32,7 @@ var (
 	// FIXME: v2 disabled due to flake
 	nodeFastSyncs         = uniformChoice{"v0"} // "v2"
 	nodeStateSyncs        = uniformChoice{false, true}
+	nodeMempools          = uniformChoice{"v0", "v1"}
 	nodePersistIntervals  = uniformChoice{0, 1, 5}
 	nodeSnapshotIntervals = uniformChoice{0, 3}
 	nodeRetainBlocks      = uniformChoice{0, 1, 5}
@@ -210,6 +211,7 @@ func generateNode(
 		Database:         nodeDatabases.Choose(r).(string),
 		PrivvalProtocol:  nodePrivvalProtocols.Choose(r).(string),
 		FastSync:         nodeFastSyncs.Choose(r).(string),
+		Mempool:          nodeMempools.Choose(r).(string),
 		StateSync:        nodeStateSyncs.Choose(r).(bool) && startAt > 0,
 		PersistInterval:  ptrUint64(uint64(nodePersistIntervals.Choose(r).(int))),
 		SnapshotInterval: uint64(nodeSnapshotIntervals.Choose(r).(int)),

test/e2e/networks/ci.toml

@@ -68,6 +68,7 @@ start_at = 1005 # Becomes part of the validator set at 1010
 seeds = ["seed02"]
 database = "cleveldb"
 fast_sync = "v0"
+mempool_version = "v1"
 # FIXME: should be grpc, disabled due to https://github.com/tendermint/tendermint/issues/5439
 #abci_protocol = "grpc"
 privval_protocol = "tcp"

test/e2e/pkg/manifest.go

@@ -92,6 +92,10 @@ type ManifestNode struct {
 	// Defaults to disabled.
 	FastSync string `toml:"fast_sync"`
 
+	// Mempool specifies which version of mempool to use. Either "v0" or "v1"
+	// This defaults to v0.
+	Mempool string `toml:"mempool_version"`
+
 	// StateSync enables state sync. The runner automatically configures trusted
 	// block hashes and RPC servers. At least one node in the network must have
 	// SnapshotInterval set to non-zero, and the state syncing node must have

test/e2e/pkg/testnet.go

@@ -75,6 +75,7 @@ type Node struct {
 	StartAt          int64
 	FastSync         string
 	StateSync        bool
+	Mempool          string
 	Database         string
 	ABCIProtocol     Protocol
 	PrivvalProtocol  Protocol
@@ -157,6 +158,7 @@ func LoadTestnet(file string) (*Testnet, error) {
 			PrivvalProtocol:  ProtocolFile,
 			StartAt:          nodeManifest.StartAt,
 			FastSync:         nodeManifest.FastSync,
+			Mempool:          nodeManifest.Mempool,
 			StateSync:        nodeManifest.StateSync,
 			PersistInterval:  1,
 			SnapshotInterval: nodeManifest.SnapshotInterval,
@@ -309,6 +311,12 @@ func (n Node) Validate(testnet Testnet) error {
 	case "", "v0", "v1", "v2":
 	default:
 		return fmt.Errorf("invalid fast sync setting %q", n.FastSync)
+
+	}
+	switch n.Mempool {
+	case "", "v0", "v1":
+	default:
+		return fmt.Errorf("invalid mempool version %q", n.Mempool)
 	}
 	switch n.Database {
 	case "goleveldb", "cleveldb", "boltdb", "rocksdb", "badgerdb":

test/e2e/runner/setup.go

@@ -284,6 +284,9 @@ func MakeConfig(node *e2e.Node) (*config.Config, error) {
 	default:
 		return nil, fmt.Errorf("unexpected mode %q", node.Mode)
 	}
+	if node.Mempool != "" {
+		cfg.Mempool.Version = node.Mempool
+	}
 
 	if node.FastSync == "" {
 		cfg.FastSyncMode = false

test/fuzz/mempool/v0/checktx.go

@@ -1,9 +1,10 @@
-package checktx
+package v0
 
 import (
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	"github.com/tendermint/tendermint/config"
 	mempl "github.com/tendermint/tendermint/mempool"
+	mempoolv0 "github.com/tendermint/tendermint/mempool/v0"
 	"github.com/tendermint/tendermint/proxy"
 )
 
@@ -20,8 +21,7 @@ func init() {
 
 	cfg := config.DefaultMempoolConfig()
 	cfg.Broadcast = false
-
-	mempool = mempl.NewCListMempool(cfg, appConnMem, 0)
+	mempool = mempoolv0.NewCListMempool(cfg, appConnMem, 0)
 }
 
 func Fuzz(data []byte) int {

test/fuzz/mempool/v0/fuzz_test.go

@@ -0,0 +1,33 @@
+package v0_test
+
+import (
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	mempoolv0 "github.com/tendermint/tendermint/test/fuzz/mempool/v0"
+)
+
+const testdataCasesDir = "testdata/cases"
+
+func TestMempoolTestdataCases(t *testing.T) {
+	entries, err := os.ReadDir(testdataCasesDir)
+	require.NoError(t, err)
+
+	for _, e := range entries {
+		entry := e
+		t.Run(entry.Name(), func(t *testing.T) {
+			defer func() {
+				r := recover()
+				require.Nilf(t, r, "testdata/cases test panic")
+			}()
+			f, err := os.Open(filepath.Join(testdataCasesDir, entry.Name()))
+			require.NoError(t, err)
+			input, err := ioutil.ReadAll(f)
+			require.NoError(t, err)
+			mempoolv0.Fuzz(input)
+		})
+	}
+}

test/fuzz/mempool/v1/checktx.go

@@ -0,0 +1,37 @@
+package v1
+
+import (
+	"github.com/tendermint/tendermint/abci/example/kvstore"
+	"github.com/tendermint/tendermint/config"
+	"github.com/tendermint/tendermint/libs/log"
+	mempl "github.com/tendermint/tendermint/mempool"
+	"github.com/tendermint/tendermint/proxy"
+
+	mempoolv1 "github.com/tendermint/tendermint/mempool/v1"
+)
+
+var mempool mempl.Mempool
+
+func init() {
+	app := kvstore.NewApplication()
+	cc := proxy.NewLocalClientCreator(app)
+	appConnMem, _ := cc.NewABCIClient()
+	err := appConnMem.Start()
+	if err != nil {
+		panic(err)
+	}
+	cfg := config.DefaultMempoolConfig()
+	cfg.Broadcast = false
+	log := log.NewNopLogger()
+	mempool = mempoolv1.NewTxMempool(log, cfg, appConnMem, 0)
+}
+
+func Fuzz(data []byte) int {
+
+	err := mempool.CheckTx(data, nil, mempl.TxInfo{})
+	if err != nil {
+		return 0
+	}
+
+	return 1
+}

test/fuzz/mempool/v1/fuzz_test.go

@@ -0,0 +1,33 @@
+package v1_test
+
+import (
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	mempoolv1 "github.com/tendermint/tendermint/test/fuzz/mempool/v1"
+)
+
+const testdataCasesDir = "testdata/cases"
+
+func TestMempoolTestdataCases(t *testing.T) {
+	entries, err := os.ReadDir(testdataCasesDir)
+	require.NoError(t, err)
+
+	for _, e := range entries {
+		entry := e
+		t.Run(entry.Name(), func(t *testing.T) {
+			defer func() {
+				r := recover()
+				require.Nilf(t, r, "testdata/cases test panic")
+			}()
+			f, err := os.Open(filepath.Join(testdataCasesDir, entry.Name()))
+			require.NoError(t, err)
+			input, err := ioutil.ReadAll(f)
+			require.NoError(t, err)
+			mempoolv1.Fuzz(input)
+		})
+	}
+}

test/maverick/consensus/replay_stubs.go

@@ -15,12 +15,14 @@ type emptyMempool struct{}
 
 var _ mempl.Mempool = emptyMempool{}
 
-func (emptyMempool) Lock()     {}
-func (emptyMempool) Unlock()   {}
-func (emptyMempool) Size() int { return 0 }
+func (emptyMempool) Lock()            {}
+func (emptyMempool) Unlock()          {}
+func (emptyMempool) Size() int        { return 0 }
+func (emptyMempool) SizeBytes() int64 { return 0 }
 func (emptyMempool) CheckTx(_ types.Tx, _ func(*abci.Response), _ mempl.TxInfo) error {
 	return nil
 }
+func (emptyMempool) RemoveTxByKey(txKey types.TxKey) error   { return nil }
 func (emptyMempool) ReapMaxBytesMaxGas(_, _ int64) types.Txs { return types.Txs{} }
 func (emptyMempool) ReapMaxTxs(n int) types.Txs              { return types.Txs{} }
 func (emptyMempool) Update(

test/maverick/node/node.go

@@ -32,6 +32,8 @@ import (
 	"github.com/tendermint/tendermint/libs/service"
 	"github.com/tendermint/tendermint/light"
 	mempl "github.com/tendermint/tendermint/mempool"
+	mempoolv0 "github.com/tendermint/tendermint/mempool/v0"
+	mempoolv1 "github.com/tendermint/tendermint/mempool/v1"
 	"github.com/tendermint/tendermint/p2p"
 	"github.com/tendermint/tendermint/p2p/pex"
 	"github.com/tendermint/tendermint/privval"
@@ -238,7 +240,7 @@ type Node struct {
 	stateStore        sm.Store
 	blockStore        *store.BlockStore // store the blockchain to disk
 	bcReactor         p2p.Reactor       // for fast-syncing
-	mempoolReactor    *mempl.Reactor    // for gossipping transactions
+	mempoolReactor    p2p.Reactor       // for gossipping transactions
 	mempool           mempl.Mempool
 	stateSync         bool                    // whether the node should state sync on startup
 	stateSyncReactor  *statesync.Reactor      // for hosting and restoring state sync snapshots
@@ -378,24 +380,56 @@ func onlyValidatorIsUs(state sm.State, pubKey crypto.PubKey) bool {
 }
 
 func createMempoolAndMempoolReactor(config *cfg.Config, proxyApp proxy.AppConns,
-	state sm.State, memplMetrics *mempl.Metrics, logger log.Logger) (*mempl.Reactor, *mempl.CListMempool) {
+	state sm.State, memplMetrics *mempl.Metrics, logger log.Logger) (p2p.Reactor, mempl.Mempool) {
 
-	mempool := mempl.NewCListMempool(
-		config.Mempool,
-		proxyApp.Mempool(),
-		state.LastBlockHeight,
-		mempl.WithMetrics(memplMetrics),
-		mempl.WithPreCheck(sm.TxPreCheck(state)),
-		mempl.WithPostCheck(sm.TxPostCheck(state)),
-	)
-	mempoolLogger := logger.With("module", "mempool")
-	mempoolReactor := mempl.NewReactor(config.Mempool, mempool)
-	mempoolReactor.SetLogger(mempoolLogger)
+	switch config.Mempool.Version {
+	case cfg.MempoolV1:
+		mp := mempoolv1.NewTxMempool(
+			logger,
+			config.Mempool,
+			proxyApp.Mempool(),
+			state.LastBlockHeight,
+			mempoolv1.WithMetrics(memplMetrics),
+			mempoolv1.WithPreCheck(sm.TxPreCheck(state)),
+			mempoolv1.WithPostCheck(sm.TxPostCheck(state)),
+		)
 
-	if config.Consensus.WaitForTxs() {
-		mempool.EnableTxsAvailable()
+		reactor := mempoolv1.NewReactor(
+			config.Mempool,
+			mp,
+		)
+		if config.Consensus.WaitForTxs() {
+			mp.EnableTxsAvailable()
+		}
+
+		return reactor, mp
+
+	case cfg.MempoolV0:
+		mp := mempoolv0.NewCListMempool(
+			config.Mempool,
+			proxyApp.Mempool(),
+			state.LastBlockHeight,
+			mempoolv0.WithMetrics(memplMetrics),
+			mempoolv0.WithPreCheck(sm.TxPreCheck(state)),
+			mempoolv0.WithPostCheck(sm.TxPostCheck(state)),
+		)
+
+		mp.SetLogger(logger)
+		mp.SetLogger(logger)
+
+		reactor := mempoolv0.NewReactor(
+			config.Mempool,
+			mp,
+		)
+		if config.Consensus.WaitForTxs() {
+			mp.EnableTxsAvailable()
+		}
+
+		return reactor, mp
+
+	default:
+		return nil, nil
 	}
-	return mempoolReactor, mempool
 }
 
 func createEvidenceReactor(config *cfg.Config, dbProvider DBProvider,
@@ -441,7 +475,7 @@ func createConsensusReactor(config *cfg.Config,
 	state sm.State,
 	blockExec *sm.BlockExecutor,
 	blockStore sm.BlockStore,
-	mempool *mempl.CListMempool,
+	mempool mempl.Mempool,
 	evidencePool *evidence.Pool,
 	privValidator types.PrivValidator,
 	csMetrics *consensus.Metrics,
@@ -545,7 +579,7 @@ func createSwitch(config *cfg.Config,
 	transport p2p.Transport,
 	p2pMetrics *p2p.Metrics,
 	peerFilters []p2p.PeerFilterFunc,
-	mempoolReactor *mempl.Reactor,
+	mempoolReactor p2p.Reactor,
 	bcReactor p2p.Reactor,
 	stateSyncReactor *statesync.Reactor,
 	consensusReactor *cs.Reactor,
@@ -947,13 +981,6 @@ func (n *Node) OnStart() error {
 
 	n.isListening = true
 
-	if n.config.Mempool.WalEnabled() {
-		err = n.mempool.InitWAL()
-		if err != nil {
-			return fmt.Errorf("init mempool WAL: %w", err)
-		}
-	}
-
 	// Start the switch (the P2P server).
 	err = n.sw.Start()
 	if err != nil {
@@ -1001,11 +1028,6 @@ func (n *Node) OnStop() {
 		n.Logger.Error("Error closing switch", "err", err)
 	}
 
-	// stop mempool WAL
-	if n.config.Mempool.WalEnabled() {
-		n.mempool.CloseWAL()
-	}
-
 	if err := n.transport.Close(); err != nil {
 		n.Logger.Error("Error closing transport", "err", err)
 	}
@@ -1224,7 +1246,7 @@ func (n *Node) ConsensusReactor() *cs.Reactor {
 }
 
 // MempoolReactor returns the Node's mempool reactor.
-func (n *Node) MempoolReactor() *mempl.Reactor {
+func (n *Node) MempoolReactor() p2p.Reactor {
 	return n.mempoolReactor
 }