TxMempool rewrite (CON-305)

sei-cosmos/server/start.go

@@ -360,7 +360,7 @@ func startInProcess(
 		}
 		defer func() {
 			if tmNode.IsRunning() {
-				tmNode.Wait()
+				tmNode.Stop()
 			}
 		}()
 		// Add the tx service to the gRPC router. We only need to register this
@@ -464,8 +464,6 @@ func startInProcess(
 		}
 	}
 
-	// Defer cancelling as the last so that it is called first during unwinding.
-	defer cancel()
 	// wait for signal capture and gracefully return
 	return WaitForQuitSignals(goCtx, restartCh)
 }

sei-tendermint/config/config.go

@@ -13,6 +13,7 @@ import (
 
 	mempoolcfg "github.com/sei-protocol/sei-chain/sei-tendermint/internal/mempool"
 	tmos "github.com/sei-protocol/sei-chain/sei-tendermint/libs/os"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/types"
 )
 
@@ -860,15 +861,13 @@ type MempoolConfig struct {
 }
 
 func (cfg *MempoolConfig) ToMempoolConfig() *mempoolcfg.Config {
-	return &mempoolcfg.Config{
+	mcfg := &mempoolcfg.Config{
 		Size:                      cfg.Size,
 		MaxTxsBytes:               cfg.MaxTxsBytes,
 		CacheSize:                 cfg.CacheSize,
 		DuplicateTxsCacheSize:     cfg.DuplicateTxsCacheSize,
 		KeepInvalidTxsInCache:     cfg.KeepInvalidTxsInCache,
 		MaxTxBytes:                cfg.MaxTxBytes,
-		TTLDuration:               cfg.TTLDuration,
-		TTLNumBlocks:              cfg.TTLNumBlocks,
 		TxNotifyThreshold:         cfg.TxNotifyThreshold,
 		PendingSize:               cfg.PendingSize,
 		MaxPendingTxsBytes:        cfg.MaxPendingTxsBytes,
@@ -877,6 +876,13 @@ func (cfg *MempoolConfig) ToMempoolConfig() *mempoolcfg.Config {
 		DropUtilisationThreshold:  cfg.DropUtilisationThreshold,
 		DropPriorityReservoirSize: cfg.DropPriorityReservoirSize,
 	}
+	if cfg.TTLDuration != 0 {
+		mcfg.TTLDuration = utils.Some(cfg.TTLDuration)
+	}
+	if cfg.TTLNumBlocks != 0 {
+		mcfg.TTLNumBlocks = utils.Some(cfg.TTLNumBlocks)
+	}
+	return mcfg
 }
 
 // DefaultMempoolConfig returns a default configuration for the Tendermint mempool.
@@ -891,8 +897,8 @@ func DefaultMempoolConfig() *MempoolConfig {
 		KeepInvalidTxsInCache:        cfg.KeepInvalidTxsInCache,
 		MaxTxBytes:                   cfg.MaxTxBytes,
 		MaxBatchBytes:                0,
-		TTLDuration:                  cfg.TTLDuration,
-		TTLNumBlocks:                 cfg.TTLNumBlocks,
+		TTLDuration:                  cfg.TTLDuration.Or(0),
+		TTLNumBlocks:                 cfg.TTLNumBlocks.Or(0),
 		TxNotifyThreshold:            cfg.TxNotifyThreshold,
 		CheckTxErrorBlacklistEnabled: true,
 		CheckTxErrorThreshold:        50,

sei-tendermint/internal/autobahn/producer/state.go

@@ -66,30 +66,27 @@ func (s *State) makePayload(ctx context.Context) (*types.Payload, error) {
 	// Wait for transactions. We give up and produce an empty block if mempool is empty for
 	// cfg.BlockInterval.
 	_ = utils.WithTimeout(ctx, s.cfg.BlockInterval, func(ctx context.Context) error {
-		return s.txMempool.TxStore().WaitForTxs(ctx)
+		return s.txMempool.WaitForTxs(ctx)
 	})
 	// If the context has been cancelled though, we just fail.
 	if err := ctx.Err(); err != nil {
 		return nil, err
 	}
 
-	txs, gasEstimated := s.txMempool.PopTxs(mempool.ReapLimits{
+	txs, gasEstimated := s.txMempool.ReapTxs(mempool.ReapLimits{
 		MaxTxs:          utils.Some(min(types.MaxTxsPerBlock, s.cfg.maxTxsPerBlock())),
 		MaxBytes:        utils.Some(utils.Clamp[int64](types.MaxTxsBytesPerBlock)),
 		MaxGasWanted:    utils.Some(s.cfg.MaxGasPerBlockI64()),
 		MaxGasEstimated: utils.Some(s.cfg.MaxGasPerBlockI64()),
-	})
+	}, true)
 	payloadTxs := make([][]byte, 0, len(txs))
 	for _, tx := range txs {
 		payloadTxs = append(payloadTxs, tx)
 	}
 	payload, err := types.PayloadBuilder{
 		CreatedAt: time.Now(),
-		// TODO: ReapMaxTxsBytesMaxGas does not handle corner cases correctly rn, which actually
-		// can produce negative total gas. Fixing it right away might be backward incompatible afaict,
-		// so we leave it as is for now.
-		TotalGas: uint64(gasEstimated), // nolint:gosec
-		Txs:      payloadTxs,
+		TotalGas:  uint64(gasEstimated), // nolint:gosec // always non-negative
+		Txs:       payloadTxs,
 	}.Build()
 	// This should never happen: we construct the payload from correctly sized data.
 	if err != nil {

sei-tendermint/internal/consensus/mempool_test.go

@@ -148,7 +148,7 @@ func checkTxsRange(ctx context.Context, t *testing.T, cs *testState, start, end
 	for i := start; i < end; i++ {
 		txBytes := make([]byte, 8)
 		binary.BigEndian.PutUint64(txBytes, uint64(i))
-		res, err := cs.txMempool.CheckTx(ctx, txBytes, mempool.TxInfo{})
+		res, err := cs.txMempool.CheckTx(ctx, txBytes)
 		require.NoError(t, err, "error after checkTx")
 		require.Equal(t, code.CodeTypeOK, res.Code, "checkTx code is error, txBytes %X, index=%d", txBytes, i)
 	}
@@ -182,7 +182,7 @@ func TestMempoolTxConcurrentWithCommit(t *testing.T) {
 	for i := range int(numTxs) {
 		txBytes := make([]byte, 8)
 		binary.BigEndian.PutUint64(txBytes, uint64(i))
-		res, err := cs.txMempool.CheckTx(ctx, txBytes, mempool.TxInfo{})
+		res, err := cs.txMempool.CheckTx(ctx, txBytes)
 		require.NoError(t, err, "error after checkTx")
 		require.Equal(t, code.CodeTypeOK, res.Code, "checkTx code is error, txBytes %X, index=%d", txBytes, i)
 	}
@@ -234,7 +234,7 @@ func TestMempoolRmBadTx(t *testing.T) {
 		// 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
-		res, err := cs.txMempool.CheckTx(ctx, txBytes, mempool.TxInfo{})
+		res, err := cs.txMempool.CheckTx(ctx, txBytes)
 		if err != nil {
 			t.Errorf("error after CheckTx: %v", err)
 			return
@@ -247,7 +247,10 @@ func TestMempoolRmBadTx(t *testing.T) {
 
 		// check for the tx
 		for {
-			txs := cs.txMempool.ReapMaxBytesMaxGas(int64(len(txBytes)), utils.Max[int64](), utils.Max[int64]())
+			txs, _ := cs.txMempool.ReapTxs(
+				mempool.ReapLimits{MaxBytes: utils.Some(int64(len(txBytes)))},
+				false,
+			)
 			if len(txs) == 0 {
 				emptyMempoolCh <- struct{}{}
 				return

sei-tendermint/internal/consensus/reactor_test.go

@@ -152,7 +152,7 @@ func finalizeTx(
 	return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error {
 		for i, sub := range blocksSubs {
 			s.Spawn(func() error {
-				if _, err := states[i].txMempool.CheckTx(ctx, tx, mempool.TxInfo{}); err != nil {
+				if _, err := states[i].txMempool.CheckTx(ctx, tx); err != nil {
 					return fmt.Errorf("CheckTx(): %w", err)
 				}
 				for {
@@ -367,7 +367,6 @@ func TestReactorCreatesBlockWhenEmptyBlocksFalse(t *testing.T) {
 	_, err := states[1].txMempool.CheckTx(
 		ctx,
 		[]byte{1, 2, 3},
-		mempool.TxInfo{},
 	)
 	require.NoError(t, err)
 

sei-tendermint/internal/consensus/replay_test.go

@@ -154,7 +154,7 @@ func sendTxs(ctx context.Context, cs *testState) error {
 			return nil
 		}
 		tx := []byte{byte(i)}
-		if _, err := cs.txMempool.CheckTx(ctx, tx, mempool.TxInfo{}); err != nil {
+		if _, err := cs.txMempool.CheckTx(ctx, tx); err != nil {
 			return fmt.Errorf("cs.mempool.CheckTx(): %w", err)
 		}
 	}
@@ -271,8 +271,7 @@ type simulatorTestSuite struct {
 	Commits      []*types.Commit
 	CleanupFunc  cleanupFunc
 
-	Mempool *mempool.TxMempool
-	Evpool  sm.EvidencePool
+	Evpool sm.EvidencePool
 }
 
 const (
@@ -292,11 +291,8 @@ var modes = []uint{0, 1, 2, 3}
 func setupSimulator(ctx context.Context, t *testing.T) *simulatorTestSuite {
 	t.Helper()
 	cfg := configSetup(t)
-	proxyApp := kvstore.NewProxy()
-
 	sim := &simulatorTestSuite{
-		Mempool: newReplayTxMempool(proxyApp),
-		Evpool:  sm.EmptyEvidencePool{},
+		Evpool: sm.EmptyEvidencePool{},
 	}
 
 	nPeers := 7
@@ -389,7 +385,7 @@ func setupSimulator(ctx context.Context, t *testing.T) *simulatorTestSuite {
 	require.NoError(t, err)
 	valPubKey1ABCI := crypto.PubKeyToProto(newValidatorPubKey1)
 	newValidatorTx1 := kvstore.MakeValSetChangeTx(valPubKey1ABCI, testMinPower)
-	_, err = css[0].txMempool.CheckTx(ctx, newValidatorTx1, mempool.TxInfo{})
+	_, err = css[0].txMempool.CheckTx(ctx, newValidatorTx1)
 	assert.NoError(t, err)
 
 	css[0].signAddVotes(ctx, t, tmproto.PrecommitType, sim.Config.ChainID(),
@@ -408,7 +404,7 @@ func setupSimulator(ctx context.Context, t *testing.T) *simulatorTestSuite {
 	require.NoError(t, err)
 	updatePubKey1ABCI := crypto.PubKeyToProto(updateValidatorPubKey1)
 	updateValidatorTx1 := kvstore.MakeValSetChangeTx(updatePubKey1ABCI, 25)
-	_, err = css[0].txMempool.CheckTx(ctx, updateValidatorTx1, mempool.TxInfo{})
+	_, err = css[0].txMempool.CheckTx(ctx, updateValidatorTx1)
 	assert.NoError(t, err)
 	css[0].signAddVotes(ctx, t, tmproto.PrecommitType, sim.Config.ChainID(),
 		types.BlockID{Hash: rs.ProposalBlock.Hash(), PartSetHeader: rs.ProposalBlockParts.Header()},
@@ -426,14 +422,14 @@ func setupSimulator(ctx context.Context, t *testing.T) *simulatorTestSuite {
 	require.NoError(t, err)
 	newVal2ABCI := crypto.PubKeyToProto(newValidatorPubKey2)
 	newValidatorTx2 := kvstore.MakeValSetChangeTx(newVal2ABCI, testMinPower)
-	_, err = css[0].txMempool.CheckTx(ctx, newValidatorTx2, mempool.TxInfo{})
+	_, err = css[0].txMempool.CheckTx(ctx, newValidatorTx2)
 	assert.NoError(t, err)
 	pv, _ = css[nVals+2].privValidator.Get()
 	newValidatorPubKey3, err := pv.GetPubKey(ctx)
 	require.NoError(t, err)
 	newVal3ABCI := crypto.PubKeyToProto(newValidatorPubKey3)
 	newValidatorTx3 := kvstore.MakeValSetChangeTx(newVal3ABCI, testMinPower)
-	_, err = css[0].txMempool.CheckTx(ctx, newValidatorTx3, mempool.TxInfo{})
+	_, err = css[0].txMempool.CheckTx(ctx, newValidatorTx3)
 	assert.NoError(t, err)
 	css[0].signAddVotes(ctx, t, tmproto.PrecommitType, sim.Config.ChainID(),
 		types.BlockID{Hash: rs.ProposalBlock.Hash(), PartSetHeader: rs.ProposalBlockParts.Header()},
@@ -469,7 +465,7 @@ func setupSimulator(ctx context.Context, t *testing.T) *simulatorTestSuite {
 	ensureProposalFromCurrentLeader(height, round)
 	rs = css[0].GetRoundState()
 	removeValidatorTx2 := kvstore.MakeValSetChangeTx(newVal2ABCI, 0)
-	_, err = css[0].txMempool.CheckTx(ctx, removeValidatorTx2, mempool.TxInfo{})
+	_, err = css[0].txMempool.CheckTx(ctx, removeValidatorTx2)
 	assert.NoError(t, err)
 
 	for i := 0; i < nVals+1; i++ {
@@ -498,7 +494,7 @@ func setupSimulator(ctx context.Context, t *testing.T) *simulatorTestSuite {
 	rs = css[0].GetRoundState()
 
 	removeValidatorTx3 := kvstore.MakeValSetChangeTx(newVal3ABCI, 0)
-	_, err = css[0].txMempool.CheckTx(ctx, removeValidatorTx3, mempool.TxInfo{})
+	_, err = css[0].txMempool.CheckTx(ctx, removeValidatorTx3)
 	assert.NoError(t, err)
 	for i := 0; i < nVals+1; i++ {
 		if i == selfIndex {
@@ -655,11 +651,12 @@ func testHandshakeReplay(
 	store.commits = commits
 
 	state := genesisState.Copy()
+	replayMempool := newReplayTxMempool(kvstore.NewProxy())
 	// run the chain through state.ApplyBlock to build up the tendermint state
 	state = buildTMStateFromChain(
 		ctx,
 		t,
-		sim.Mempool,
+		replayMempool,
 		sim.Evpool,
 		stateStore,
 		state,
@@ -681,7 +678,7 @@ func testHandshakeReplay(
 		stateStore := sm.NewStore(stateDB1)
 		err := stateStore.Save(genesisState)
 		require.NoError(t, err)
-		buildAppStateFromChain(ctx, t, app, stateStore, sim.Mempool, sim.Evpool, genesisState, chain, eventBus, nBlocks, mode, store)
+		buildAppStateFromChain(ctx, t, app, stateStore, sim.Evpool, genesisState, chain, eventBus, nBlocks, mode, store)
 	}
 
 	// Prune block store if requested
@@ -759,7 +756,6 @@ func buildAppStateFromChain(
 	t *testing.T,
 	appClient *kvstore.Application,
 	stateStore sm.Store,
-	mempool *mempool.TxMempool,
 	evpool sm.EvidencePool,
 	state sm.State,
 	chain []*types.Block,
@@ -771,6 +767,7 @@ func buildAppStateFromChain(
 	t.Helper()
 	// start a new app without handshake, play nBlocks blocks
 	proxyApp := proxy.New(appClient, proxy.NopMetrics())
+	mempool := newReplayTxMempool(proxyApp)
 	state.Version.Consensus.App = kvstore.ProtocolVersion // simulate handshake, receive app version
 	_, err := appClient.InitChain(ctx, &abci.RequestInitChain{})
 	require.NoError(t, err)

sei-tendermint/internal/consensus/state.go

@@ -2279,7 +2279,7 @@ func (cs *State) buildProposalBlock(proposal *types.Proposal) *types.Block {
 	txs, missingTxs := cs.blockExec.SafeGetTxsByHashes(proposal.TxHashes)
 	if len(missingTxs) > 0 {
 		cs.metrics.ProposalMissingTxs.Set(float64(len(missingTxs)))
-		logger.Debug("Missing txs when trying to build block", "missing_txs", cs.blockExec.GetMissingTxs(proposal.TxHashes))
+		logger.Debug("Missing txs when trying to build block", "missing_txs", missingTxs)
 		return nil
 	}
 	block := cs.state.MakeBlock(proposal.Height, txs, proposal.LastCommit, proposal.Evidence, proposal.ProposerAddress)

sei-tendermint/internal/consensus/state_test.go

@@ -18,7 +18,6 @@ import (
 	"github.com/sei-protocol/sei-chain/sei-tendermint/crypto/ed25519"
 	cstypes "github.com/sei-protocol/sei-chain/sei-tendermint/internal/consensus/types"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/eventbus"
-	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/mempool"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/proxy"
 	tmpubsub "github.com/sei-protocol/sei-chain/sei-tendermint/internal/pubsub"
 	tmquery "github.com/sei-protocol/sei-chain/sei-tendermint/internal/pubsub/query"
@@ -2208,7 +2207,7 @@ func TestStartNextHeightCorrectlyAfterTimeout(t *testing.T) {
 
 	ensureNewBlockHeader(t, newBlockHeader, height, blockID.Hash)
 
-	_, err := cs1.txMempool.CheckTx(ctx, types.Tx("test-key=test-value"), mempool.TxInfo{})
+	_, err := cs1.txMempool.CheckTx(ctx, types.Tx("test-key=test-value"))
 	require.NoError(t, err, "failed to seed the mempool with a transaction")
 
 	ensureNewTimeout(t, timeoutProposeCh, height+1, 0)
@@ -2592,7 +2591,7 @@ func TestTryCreateProposalBlock_PartsMismatch(t *testing.T) {
 	incrementRound(vss[1:]...)
 	cs.startTestRound(ctx, height, round)
 
-	_, err := cs.txMempool.CheckTx(ctx, types.Tx("test-key=test-value"), mempool.TxInfo{})
+	_, err := cs.txMempool.CheckTx(ctx, types.Tx("test-key=test-value"))
 	require.NoError(t, err, "failed to seed the mempool with a transaction")
 
 	proposal, block := cs.decideProposal(ctx, t, vss[1], height, round)

sei-tendermint/internal/evidence/pool.go

@@ -534,7 +534,6 @@ func (evpool *Pool) removeEvidenceFromList(
 		ev := e.Value()
 		if _, ok := blockEvidenceMap[evMapKey(ev)]; ok {
 			evpool.evidenceList.Remove(e)
-			e.DetachPrev()
 		}
 	}
 }

sei-tendermint/internal/libs/clist/bench_test.go

@@ -2,24 +2,6 @@ package clist
 
 import "testing"
 
-func BenchmarkDetaching(b *testing.B) {
-	lst := New[int]()
-	for i := 0; i < b.N+1; i++ {
-		lst.PushBack(i)
-	}
-	start := lst.Front()
-	nxt := start.Next()
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		start.removed = true
-		start.detachNext()
-		start.DetachPrev()
-		tmp := nxt
-		nxt = nxt.Next()
-		start = tmp
-	}
-}
-
 // This is used to benchmark the time of RMutex.
 func BenchmarkRemoved(b *testing.B) {
 	lst := New[int]()