Made autobahn producer use TxMempool (#3224)

Also
* disabled bunch of reactors unused in autobahn mode.
* fixed race conditions in mempool processPeerUpdates

Author: pompon0

sei-tendermint/config/config.go

@@ -11,6 +11,7 @@ import (
 	"strings"
 	"time"
 
+	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/types"
 )
@@ -857,38 +858,60 @@ type MempoolConfig struct {
 	DropPriorityReservoirSize int `mapstructure:"drop-priority-reservoir-size"`
 }
 
+func (cfg *MempoolConfig) ToMempoolConfig() *mempoolcfg.Config {
+	return &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,
+		RemoveExpiredTxsFromQueue: cfg.RemoveExpiredTxsFromQueue,
+		DropPriorityThreshold:     cfg.DropPriorityThreshold,
+		DropUtilisationThreshold:  cfg.DropUtilisationThreshold,
+		DropPriorityReservoirSize: cfg.DropPriorityReservoirSize,
+	}
+}
+
 // DefaultMempoolConfig returns a default configuration for the Tendermint mempool.
 func DefaultMempoolConfig() *MempoolConfig {
+	cfg := mempoolcfg.DefaultConfig()
 	return &MempoolConfig{
-		Broadcast: true,
-		// Each signature verification takes .5ms, Size reduced until we implement
-		// ABCI Recheck
-		Size:                         5000,
-		MaxTxsBytes:                  1024 * 1024 * 1024, // 1GB
-		CacheSize:                    10000,
-		DuplicateTxsCacheSize:        100000,
-		MaxTxBytes:                   1024 * 1024,     // 1MB
-		TTLDuration:                  5 * time.Second, // prevent stale txs from filling mempool
-		TTLNumBlocks:                 10,              // remove txs after 10 blocks
-		TxNotifyThreshold:            0,
+		Broadcast:                    true,
+		Size:                         cfg.Size,
+		MaxTxsBytes:                  cfg.MaxTxsBytes,
+		CacheSize:                    cfg.CacheSize,
+		DuplicateTxsCacheSize:        cfg.DuplicateTxsCacheSize,
+		KeepInvalidTxsInCache:        cfg.KeepInvalidTxsInCache,
+		MaxTxBytes:                   cfg.MaxTxBytes,
+		MaxBatchBytes:                0,
+		TTLDuration:                  cfg.TTLDuration,
+		TTLNumBlocks:                 cfg.TTLNumBlocks,
+		TxNotifyThreshold:            cfg.TxNotifyThreshold,
 		CheckTxErrorBlacklistEnabled: true,
 		CheckTxErrorThreshold:        50,
-		PendingSize:                  5000,
-		MaxPendingTxsBytes:           1024 * 1024 * 1024, // 1GB
-		PendingTTLDuration:           0 * time.Second,
+		PendingSize:                  cfg.PendingSize,
+		MaxPendingTxsBytes:           cfg.MaxPendingTxsBytes,
+		PendingTTLDuration:           0,
 		PendingTTLNumBlocks:          0,
-		RemoveExpiredTxsFromQueue:    true,
-		DropPriorityThreshold:        0.1,
-		DropUtilisationThreshold:     1.0,
-		DropPriorityReservoirSize:    10_240,
+		RemoveExpiredTxsFromQueue:    cfg.RemoveExpiredTxsFromQueue,
+		DropPriorityThreshold:        cfg.DropPriorityThreshold,
+		DropUtilisationThreshold:     cfg.DropUtilisationThreshold,
+		DropPriorityReservoirSize:    cfg.DropPriorityReservoirSize,
 	}
 }
 
 // TestMempoolConfig returns a configuration for testing the Tendermint mempool
 func TestMempoolConfig() *MempoolConfig {
 	cfg := DefaultMempoolConfig()
-	cfg.CacheSize = 1000
-	cfg.DropUtilisationThreshold = 0.0
+	testCfg := mempoolcfg.TestConfig()
+	cfg.CacheSize = testCfg.CacheSize
+	cfg.DropUtilisationThreshold = testCfg.DropUtilisationThreshold
 	return cfg
 }
 

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

@@ -6,8 +6,8 @@ import (
 	"time"
 
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/consensus"
-	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/pb"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/types"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/mempool"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/scope"
 	"golang.org/x/time/rate"
@@ -32,19 +32,18 @@ func (c *Config) maxTxsPerBlock() uint64 {
 
 // State is the block producer state.
 type State struct {
-	cfg *Config
-	// channel of transactions to build the next block from.
-	mempool chan *pb.Transaction
+	cfg       *Config
+	txMempool *mempool.TxMempool
 	// consensus state to which published blocks will be reported.
 	consensus *consensus.State
 }
 
 // NewState constructs a new block producer state.
 // Returns an error if the current node is NOT a producer.
-func NewState(cfg *Config, consensus *consensus.State) *State {
+func NewState(cfg *Config, txMempool *mempool.TxMempool, consensus *consensus.State) *State {
 	return &State{
 		cfg:       cfg,
-		mempool:   make(chan *pb.Transaction, cfg.MempoolSize),
+		txMempool: txMempool,
 		consensus: consensus,
 	}
 }
@@ -54,22 +53,37 @@ func NewState(cfg *Config, consensus *consensus.State) *State {
 func (s *State) makePayload(ctx context.Context) *types.Payload {
 	ctx, cancel := context.WithTimeout(ctx, s.cfg.BlockInterval)
 	defer cancel()
-	maxTxs := s.cfg.maxTxsPerBlock()
-	var totalGas uint64
-	var txs [][]byte
-	for totalGas < s.cfg.MaxGasPerBlock && uint64(len(txs)) < maxTxs {
-		tx, err := utils.Recv(ctx, s.mempool)
-		if err != nil {
-			break
+
+	if s.txMempool.NumTxsNotPending() == 0 {
+		select {
+		case <-ctx.Done():
+		case <-s.txMempool.TxsAvailable():
 		}
-		txs = append(txs, tx.Payload)
-		totalGas += tx.GasUsed
 	}
-	return types.PayloadBuilder{
+
+	txs, gasEstimated := s.txMempool.PopTxs(mempool.ReapLimits{
+		MaxTxs:          utils.Some(min(types.MaxTxsPerBlock, s.cfg.maxTxsPerBlock())),
+		MaxBytes:        utils.Some(utils.Clamp[int64](types.MaxTxsBytesPerBlock)),
+		MaxGasWanted:    utils.Some(utils.Clamp[int64](s.cfg.MaxGasPerBlock)),
+		MaxGasEstimated: utils.Some(utils.Clamp[int64](s.cfg.MaxGasPerBlock)),
+	})
+	payloadTxs := make([][]byte, 0, len(txs))
+	for _, tx := range txs {
+		payloadTxs = append(payloadTxs, tx)
+	}
+	payload, err := types.PayloadBuilder{
 		CreatedAt: time.Now(),
-		TotalGas:  totalGas,
-		Txs:       txs,
+		// 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,
 	}.Build()
+	// This should never happen: we construct the payload from correctly sized data.
+	if err != nil {
+		panic(fmt.Errorf("PayloadBuilder{}.Build(): %w", err))
+	}
+	return payload
 }
 
 // nextPayload constructs the payload for the next block.
@@ -86,11 +100,6 @@ func (s *State) nextPayload(ctx context.Context) (*types.Payload, error) {
 	}
 }
 
-// PushToMempool pushes the transaction to the mempool.
-func (s *State) PushToMempool(ctx context.Context, tx *pb.Transaction) error {
-	return utils.Send(ctx, s.mempool, tx)
-}
-
 // Run runs the background tasks of the producer state.
 func (s *State) Run(ctx context.Context) error {
 	return scope.Run(ctx, func(ctx context.Context, scope scope.Scope) error {

sei-tendermint/internal/autobahn/types/block.go

@@ -72,6 +72,30 @@ func (h *BlockHeader) Verify(c *Committee) error {
 	return nil
 }
 
+const standardTxBytes uint64 = 1024
+
+// Maximum number of transactions in a block.
+const MaxTxsPerBlock uint64 = 2000
+
+// Maximum total size of all the transactions.
+// It can be split arbitrarily across transactions (1 large, 2000 small ones, etc.)
+// up to MaxTxsPerBlock limit.
+const MaxTxsBytesPerBlock = MaxTxsPerBlock * standardTxBytes
+
+// Upper bound on the block proto encoding.
+var MaxBlockProtoSize = func() uint64 {
+	// Payload.Txs represents the variable part of the Block size.
+	// Proto size is maximized if we distribute data evenly across transactions.
+	tx := make([]byte, standardTxBytes)
+	txs := make([][]byte, MaxTxsPerBlock)
+	for i := range txs {
+		txs[i] = tx
+	}
+	// Crude estimate of all other fields.
+	const otherFields = 100 * 1024
+	return otherFields + uint64(protoutils.Size(&pb.Block{Payload: &pb.Payload{Txs: txs}}))
+}()
+
 // Block .
 type Block struct {
 	utils.ReadOnly
@@ -149,7 +173,19 @@ type Payload struct {
 }
 
 // Build builds the Payload.
-func (b PayloadBuilder) Build() *Payload { return &Payload{p: b} }
+func (b PayloadBuilder) Build() (*Payload, error) {
+	if uint64(len(b.Txs)) > MaxTxsPerBlock {
+		return nil, fmt.Errorf("too many transactions")
+	}
+	total := uint64(0)
+	for _, tx := range b.Txs {
+		total += uint64(len(tx))
+	}
+	if total > MaxTxsBytesPerBlock {
+		return nil, fmt.Errorf("total txs bytes too large")
+	}
+	return &Payload{p: b}, nil
+}
 
 // ToBuilder converts the Payload to a PayloadBuilder.
 func (p *Payload) ToBuilder() PayloadBuilder { return p.p }
@@ -245,7 +281,7 @@ var PayloadConv = protoutils.Conv[*Payload, *pb.Payload]{
 			Coinbase:  p.Coinbase,
 			Basefee:   *p.Basefee,
 			Txs:       p.Txs,
-		}.Build(), nil
+		}.Build()
 	},
 }
 

sei-tendermint/internal/autobahn/types/testonly.go

@@ -90,14 +90,14 @@ func GenBlockHeader(rng utils.Rng) *BlockHeader {
 
 // GenPayload generates a random Payload.
 func GenPayload(rng utils.Rng) *Payload {
-	return PayloadBuilder{
+	return utils.OrPanic1(PayloadBuilder{
 		CreatedAt: utils.GenTimestamp(rng),
 		TotalGas:  rng.Uint64(),
 		EdgeCount: rng.Int63(),
 		Coinbase:  utils.GenBytes(rng, 10),
 		Basefee:   rng.Int63(),
 		Txs:       utils.GenSlice(rng, func(rng utils.Rng) []byte { return utils.GenBytes(rng, 10) }),
-	}.Build()
+	}.Build())
 }
 
 // GenBlock generates a random Block.

sei-tendermint/internal/blocksync/reactor_test.go

@@ -93,7 +93,7 @@ func makeReactor(
 	state, err := sm.MakeGenesisState(genDoc)
 	require.NoError(t, err)
 	require.NoError(t, stateStore.Save(state))
-	mp := mempool.NewTxMempool(config.TestMempoolConfig(), app, mempool.NopMetrics(), mempool.NopTxConstraintsFetcher)
+	mp := mempool.NewTxMempool(mempool.TestConfig(), app, mempool.NopMetrics(), mempool.NopTxConstraintsFetcher)
 	eventbus := eventbus.NewDefault()
 	require.NoError(t, eventbus.Start(ctx))
 

sei-tendermint/internal/consensus/common_test.go

@@ -465,7 +465,7 @@ func newStateWithConfigAndBlockStore(
 	// Make Mempool
 
 	mempool := mempool.NewTxMempool(
-		thisConfig.Mempool,
+		thisConfig.Mempool.ToMempoolConfig(),
 		proxyAppConnMem,
 		mempool.NopMetrics(),
 		mempool.NopTxConstraintsFetcher,

sei-tendermint/internal/consensus/mempool_test.go

@@ -20,6 +20,7 @@ import (
 	sm "github.com/sei-protocol/sei-chain/sei-tendermint/internal/state"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/store"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/test/factory"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/types"
 )
 
@@ -247,7 +248,7 @@ func TestMempoolRmBadTx(t *testing.T) {
 
 		// check for the tx
 		for {
-			txs := cs.txMempool.ReapMaxBytesMaxGas(int64(len(txBytes)), -1, -1)
+			txs := cs.txMempool.ReapMaxBytesMaxGas(int64(len(txBytes)), utils.Max[int64](), utils.Max[int64]())
 			if len(txs) == 0 {
 				emptyMempoolCh <- struct{}{}
 				return

sei-tendermint/internal/consensus/reactor_test.go

@@ -273,7 +273,7 @@ func TestReactorWithEvidence(t *testing.T) {
 		proxyAppConnCon := app
 
 		mempool := mempool.NewTxMempool(
-			thisConfig.Mempool,
+			thisConfig.Mempool.ToMempoolConfig(),
 			proxyAppConnMem,
 			mempool.NopMetrics(),
 			mempool.NopTxConstraintsFetcher,

sei-tendermint/internal/consensus/replay.go

@@ -135,7 +135,7 @@ func NewHandshaker(
 }
 
 func newReplayTxMempool(appClient abci.Application) *mempool.TxMempool {
-	return mempool.NewTxMempool(config.DefaultMempoolConfig(), appClient, mempool.NopMetrics(), mempool.NopTxConstraintsFetcher)
+	return mempool.NewTxMempool(config.DefaultMempoolConfig().ToMempoolConfig(), appClient, mempool.NopMetrics(), mempool.NopTxConstraintsFetcher)
 }
 
 // NBlocks returns the number of blocks applied to the state.