feat! content addressable transaction pool

config/config.go

@@ -28,6 +28,7 @@ const (
 	// Default is v0.
 	MempoolV0 = "v0"
 	MempoolV1 = "v1"
+	MempoolV2 = "v2"
 )
 
 // NOTE: Most of the structs & relevant comments + the
@@ -687,9 +688,7 @@ type MempoolConfig struct {
 	// Mempool version to use:
 	//  1) "v0" - (default) FIFO mempool.
 	//  2) "v1" - prioritized mempool.
-	// WARNING: There's a known memory leak with the prioritized mempool
-	// that the team are working on. Read more here:
-	// https://github.com/tendermint/tendermint/issues/8775
+	//  3) "v2" - content addressable transaction pool
 	Version   string `mapstructure:"version"`
 	RootDir   string `mapstructure:"home"`
 	Recheck   bool   `mapstructure:"recheck"`
@@ -735,7 +734,7 @@ type MempoolConfig struct {
 // DefaultMempoolConfig returns a default configuration for the Tendermint mempool
 func DefaultMempoolConfig() *MempoolConfig {
 	return &MempoolConfig{
-		Version:   MempoolV0,
+		Version:   MempoolV2,
 		Recheck:   true,
 		Broadcast: true,
 		WalPath:   "",

config/toml.go

@@ -345,6 +345,7 @@ dial_timeout = "{{ .P2P.DialTimeout }}"
 # Mempool version to use:
 #   1) "v0" - (default) FIFO mempool.
 #   2) "v1" - prioritized mempool.
+#   3) "v2" - content addressable transaction pool
 version = "{{ .Mempool.Version }}"
 
 recheck = {{ .Mempool.Recheck }}

consensus/byzantine_test.go

@@ -23,6 +23,7 @@ import (
 	mempl "github.com/tendermint/tendermint/mempool"
 
 	cfg "github.com/tendermint/tendermint/config"
+	mempoolv2 "github.com/tendermint/tendermint/mempool/cat"
 	mempoolv0 "github.com/tendermint/tendermint/mempool/v0"
 	mempoolv1 "github.com/tendermint/tendermint/mempool/v1"
 	"github.com/tendermint/tendermint/p2p"
@@ -88,6 +89,15 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 				mempoolv1.WithPreCheck(sm.TxPreCheck(state)),
 				mempoolv1.WithPostCheck(sm.TxPostCheck(state)),
 			)
+		case cfg.MempoolV2:
+			mempool = mempoolv2.NewTxPool(
+				logger,
+				config.Mempool,
+				proxyAppConnConMem,
+				state.LastBlockHeight,
+				mempoolv2.WithPreCheck(sm.TxPreCheck(state)),
+				mempoolv2.WithPostCheck(sm.TxPostCheck(state)),
+			)
 		}
 
 		if thisConfig.Consensus.WaitForTxs() {

consensus/common_test.go

@@ -29,6 +29,7 @@ import (
 	tmpubsub "github.com/tendermint/tendermint/libs/pubsub"
 	tmsync "github.com/tendermint/tendermint/libs/sync"
 	mempl "github.com/tendermint/tendermint/mempool"
+	mempoolv2 "github.com/tendermint/tendermint/mempool/cat"
 	mempoolv0 "github.com/tendermint/tendermint/mempool/v0"
 	mempoolv1 "github.com/tendermint/tendermint/mempool/v1"
 	"github.com/tendermint/tendermint/p2p"
@@ -418,6 +419,17 @@ func newStateWithConfigAndBlockStore(
 			mempoolv1.WithPreCheck(sm.TxPreCheck(state)),
 			mempoolv1.WithPostCheck(sm.TxPostCheck(state)),
 		)
+	case cfg.MempoolV2:
+		logger := consensusLogger()
+		mempool = mempoolv2.NewTxPool(
+			logger,
+			config.Mempool,
+			proxyAppConnConMem,
+			state.LastBlockHeight,
+			mempoolv2.WithMetrics(memplMetrics),
+			mempoolv2.WithPreCheck(sm.TxPreCheck(state)),
+			mempoolv2.WithPostCheck(sm.TxPostCheck(state)),
+		)
 	}
 	if thisConfig.Consensus.WaitForTxs() {
 		mempool.EnableTxsAvailable()
@@ -706,7 +718,10 @@ func consensusLogger() log.Logger {
 	return log.TestingLoggerWithColorFn(func(keyvals ...interface{}) term.FgBgColor {
 		for i := 0; i < len(keyvals)-1; i += 2 {
 			if keyvals[i] == "validator" {
-				return term.FgBgColor{Fg: term.Color(uint8(keyvals[i+1].(int) + 1))}
+				index, ok := keyvals[i+1].(int)
+				if ok {
+					return term.FgBgColor{Fg: term.Color(uint8(index + 1))}
+				}
 			}
 		}
 		return term.FgBgColor{}

consensus/mempool_test.go

@@ -154,22 +154,12 @@ func TestMempoolRmBadTx(t *testing.T) {
 	assert.True(t, len(resCommit.Data) > 0)
 
 	emptyMempoolCh := make(chan struct{})
-	checkTxRespCh := make(chan struct{})
 	go func() {
 		// Try to send the tx through the mempool.
 		// CheckTx should not err, but the app should return a bad abci code
 		// and the tx should get removed from the pool
-		err := assertMempool(cs.txNotifier).CheckTx(txBytes, func(r *abci.Response) {
-			if r.GetCheckTx().Code != code.CodeTypeBadNonce {
-				t.Errorf("expected checktx to return bad nonce, got %v", r)
-				return
-			}
-			checkTxRespCh <- struct{}{}
-		}, mempl.TxInfo{})
-		if err != nil {
-			t.Errorf("error after CheckTx: %v", err)
-			return
-		}
+		err := assertMempool(cs.txNotifier).CheckTx(txBytes, nil, mempl.TxInfo{})
+		require.Error(t, err)
 
 		// check for the tx
 		for {
@@ -182,18 +172,8 @@ func TestMempoolRmBadTx(t *testing.T) {
 		}
 	}()
 
-	// Wait until the tx returns
-	ticker := time.After(time.Second * 5)
-	select {
-	case <-checkTxRespCh:
-		// success
-	case <-ticker:
-		t.Errorf("timed out waiting for tx to return")
-		return
-	}
-
 	// Wait until the tx is removed
-	ticker = time.After(time.Second * 5)
+	ticker := time.After(time.Second * 5)
 	select {
 	case <-emptyMempoolCh:
 		// success
@@ -232,10 +212,20 @@ func (app *CounterApplication) DeliverTx(req abci.RequestDeliverTx) abci.Respons
 
 func (app *CounterApplication) CheckTx(req abci.RequestCheckTx) abci.ResponseCheckTx {
 	txValue := txAsUint64(req.Tx)
+	if req.Type == abci.CheckTxType_Recheck {
+		if txValue >= uint64(app.txCount) {
+			return abci.ResponseCheckTx{Code: code.CodeTypeOK}
+		}
+		return abci.ResponseCheckTx{
+			Code: code.CodeTypeBadNonce,
+			Log:  fmt.Sprintf("Invalid nonce. Expected %v, got %v", app.mempoolTxCount, txValue),
+		}
+	}
 	if txValue != uint64(app.mempoolTxCount) {
 		return abci.ResponseCheckTx{
 			Code: code.CodeTypeBadNonce,
-			Log:  fmt.Sprintf("Invalid nonce. Expected %v, got %v", app.mempoolTxCount, txValue)}
+			Log:  fmt.Sprintf("Invalid nonce. Expected %v, got %v", app.mempoolTxCount, txValue),
+		}
 	}
 	app.mempoolTxCount++
 	return abci.ResponseCheckTx{Code: code.CodeTypeOK}
@@ -248,7 +238,9 @@ func txAsUint64(tx []byte) uint64 {
 }
 
 func (app *CounterApplication) Commit() abci.ResponseCommit {
-	app.mempoolTxCount = app.txCount
+	if app.mempoolTxCount < app.txCount {
+		app.mempoolTxCount = app.txCount
+	}
 	if app.txCount == 0 {
 		return abci.ResponseCommit{}
 	}

consensus/reactor_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/tendermint/tendermint/libs/log"
 	tmsync "github.com/tendermint/tendermint/libs/sync"
 	mempl "github.com/tendermint/tendermint/mempool"
+	mempoolv2 "github.com/tendermint/tendermint/mempool/cat"
 	mempoolv0 "github.com/tendermint/tendermint/mempool/v0"
 	mempoolv1 "github.com/tendermint/tendermint/mempool/v1"
 	"github.com/tendermint/tendermint/p2p"
@@ -184,6 +185,16 @@ func TestReactorWithEvidence(t *testing.T) {
 				mempoolv1.WithPreCheck(sm.TxPreCheck(state)),
 				mempoolv1.WithPostCheck(sm.TxPostCheck(state)),
 			)
+		case cfg.MempoolV2:
+			mempool = mempoolv2.NewTxPool(
+				logger,
+				config.Mempool,
+				proxyAppConnConMem,
+				state.LastBlockHeight,
+				mempoolv2.WithMetrics(memplMetrics),
+				mempoolv2.WithPreCheck(sm.TxPreCheck(state)),
+				mempoolv2.WithPostCheck(sm.TxPostCheck(state)),
+			)
 		}
 		if thisConfig.Consensus.WaitForTxs() {
 			mempool.EnableTxsAvailable()

light/rpc/client.go

@@ -131,8 +131,8 @@ func (c *Client) ABCIQuery(ctx context.Context, path string, data tmbytes.HexByt
 
 // ABCIQueryWithOptions returns an error if opts.Prove is false.
 func (c *Client) ABCIQueryWithOptions(ctx context.Context, path string, data tmbytes.HexBytes,
-	opts rpcclient.ABCIQueryOptions) (*ctypes.ResultABCIQuery, error) {
-
+	opts rpcclient.ABCIQueryOptions,
+) (*ctypes.ResultABCIQuery, error) {
 	// always request the proof
 	opts.Prove = true
 
@@ -516,7 +516,6 @@ func (c *Client) Validators(
 	height *int64,
 	pagePtr, perPagePtr *int,
 ) (*ctypes.ResultValidators, error) {
-
 	// Update the light client if we're behind and retrieve the light block at the
 	// requested height or at the latest height if no height is provided.
 	l, err := c.updateLightClientIfNeededTo(ctx, height)
@@ -538,15 +537,17 @@ func (c *Client) Validators(
 		BlockHeight: l.Height,
 		Validators:  v,
 		Count:       len(v),
-		Total:       totalCount}, nil
+		Total:       totalCount,
+	}, nil
 }
 
 func (c *Client) BroadcastEvidence(ctx context.Context, ev types.Evidence) (*ctypes.ResultBroadcastEvidence, error) {
 	return c.next.BroadcastEvidence(ctx, ev)
 }
 
 func (c *Client) Subscribe(ctx context.Context, subscriber, query string,
-	outCapacity ...int) (out <-chan ctypes.ResultEvent, err error) {
+	outCapacity ...int,
+) (out <-chan ctypes.ResultEvent, err error) {
 	return c.next.Subscribe(ctx, subscriber, query, outCapacity...)
 }