Set a protocol-wide mempool gossip transaction size (CON-420)

sei-tendermint/config/config.go

@@ -884,8 +884,8 @@ type MempoolConfig struct {
 	// valid again in the future.
 	KeepInvalidTxsInCache bool `mapstructure:"keep-invalid-txs-in-cache"`
 
-	// Maximum size of a single transaction
-	// NOTE: the max size of a tx transmitted over the network is {max-tx-bytes}.
+	// Maximum size of a single transaction.
+	// XXX: Unused. Admission uses the protocol gossip limit. A different value logs a warning.
 	MaxTxBytes int `mapstructure:"max-tx-bytes"`
 
 	// Maximum size of a batch of transactions to send to a peer

sei-tendermint/config/toml.go

@@ -437,7 +437,7 @@ duplicate-txs-cache-size = "{{ .Mempool.DuplicateTxsCacheSize }}"
 keep-invalid-txs-in-cache = {{ .Mempool.KeepInvalidTxsInCache }}
 
 # Maximum size of a single transaction.
-# NOTE: the max size of a tx transmitted over the network is {max-tx-bytes}.
+# XXX: Unused. Admission uses the protocol gossip limit. A different value logs a warning.
 max-tx-bytes = {{ .Mempool.MaxTxBytes }}
 
 # Maximum size of a batch of transactions to send to a peer

sei-tendermint/internal/mempool/mempool.go

@@ -53,8 +53,8 @@ type Config struct {
 	// valid again in the future.
 	KeepInvalidTxsInCache bool
 
-	// Maximum size of a single transaction
-	// NOTE: the max size of a tx transmitted over the network is {max-tx-bytes}.
+	// Maximum size of a single transaction.
+	// XXX: Unused. Admission uses the protocol gossip limit. A different value logs a warning.
 	MaxTxBytes int
 
 	// time after which transaction is removed from mempool.
@@ -129,7 +129,7 @@ func DefaultConfig() *Config {
 		MaxTxsBytes:               1024 * 1024 * 1024, // 1GB
 		CacheSize:                 10000,
 		DuplicateTxsCacheSize:     100000,
-		MaxTxBytes:                1024 * 1024,                 // 1MB
+		MaxTxBytes:                types.MaxGossipTxBytes,
 		TTLDuration:               utils.Some(5 * time.Second), // prevent stale txs from filling mempool
 		TTLNumBlocks:              utils.Some(int64(10)),       // remove txs after 10 blocks
 		TxNotifyThreshold:         0,
@@ -209,6 +209,11 @@ func NewTxMempool(
 	app *proxy.Proxy,
 	txConstraintsFetcher TxConstraintsFetcher,
 ) *TxMempool {
+	if cfg.MaxTxBytes != types.MaxGossipTxBytes {
+		logger.Warn("mempool max-tx-bytes differs from the protocol gossip limit; admission uses the protocol limit",
+			"max-tx-bytes", cfg.MaxTxBytes,
+			"protocol", types.MaxGossipTxBytes)
+	}
 	txmp := &TxMempool{
 		config:               cfg,
 		app:                  app,
@@ -291,8 +296,8 @@ func (txmp *TxMempool) CheckTx(ctx context.Context, tx types.Tx) (*abci.Response
 	defer txmp.mtx.RUnlock()
 
 	// Early exit if tx is too large.
-	if txSize := len(tx); txSize > txmp.config.MaxTxBytes {
-		return nil, fmt.Errorf("%w: max size is %d, but got %d", ErrTxTooLarge, txmp.config.MaxTxBytes, txSize)
+	if txSize := len(tx); txSize > types.MaxGossipTxBytes {
+		return nil, fmt.Errorf("%w: max size is %d, but got %d", ErrTxTooLarge, types.MaxGossipTxBytes, txSize)
 	}
 	hTx := newHashedTx(tx)
 

sei-tendermint/internal/mempool/mempool_test.go

@@ -553,14 +553,14 @@ func TestTxMempool_CheckTxExceedsMaxSize(t *testing.T) {
 	txmp := setup(cfg, proxy.New(client), NopTxConstraintsFetcher)
 
 	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
-	tx := make([]byte, txmp.config.MaxTxBytes+1)
+	tx := make([]byte, types.MaxGossipTxBytes+1)
 	_, err := rng.Read(tx)
 	require.NoError(t, err)
 
 	_, err = txmp.CheckTx(ctx, tx)
 	require.Error(t, err)
 
-	tx = make([]byte, txmp.config.MaxTxBytes-1)
+	tx = make([]byte, types.MaxGossipTxBytes-1)
 	_, err = rng.Read(tx)
 	require.NoError(t, err)
 

sei-tendermint/internal/mempool/reactor/reactor.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"runtime/debug"
+	"sync"
 
 	"github.com/sei-protocol/sei-chain/sei-tendermint/config"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/libs/clist"
@@ -22,6 +23,15 @@ var (
 	logger = seilog.NewLogger("tendermint", "internal", "mempool")
 
 	_ service.Service = (*Reactor)(nil)
+
+	// mempoolRecvMessageCapacity is the encoded size of one MaxGossipTxBytes transaction.
+	mempoolRecvMessageCapacity = sync.OnceValue(func() int {
+		return (&pb.Message{
+			Sum: &pb.Message_Txs{
+				Txs: &pb.Txs{Txs: [][]byte{make([]byte, types.MaxGossipTxBytes)}},
+			},
+		}).Size()
+	})
 )
 
 const MempoolChannel p2p.ChannelID = 0x30
@@ -45,7 +55,7 @@ type Reactor struct {
 
 // NewReactor returns a reference to a new reactor.
 func NewReactor(cfg *config.MempoolConfig, txmp *mempool.TxMempool, router *p2p.Router) (*Reactor, error) {
-	channel, err := p2p.OpenChannel(router, GetChannelDescriptor(cfg))
+	channel, err := p2p.OpenChannel(router, GetChannelDescriptor())
 	if err != nil {
 		return nil, fmt.Errorf("router.OpenChannel(): %w", err)
 	}
@@ -65,19 +75,13 @@ func (r *Reactor) MarkReadyToStart() { r.readyToStart <- struct{}{} }
 
 // GetChannelDescriptor produces an instance of a descriptor for this package's
 // required channels.
-func GetChannelDescriptor(cfg *config.MempoolConfig) p2p.ChannelDescriptor[*pb.Message] {
-	largestTx := make([]byte, cfg.MaxTxBytes)
-	batchMsg := &pb.Message{
-		Sum: &pb.Message_Txs{
-			Txs: &pb.Txs{Txs: [][]byte{largestTx}},
-		},
-	}
-
+func GetChannelDescriptor() p2p.ChannelDescriptor[*pb.Message] {
 	return p2p.ChannelDescriptor[*pb.Message]{
 		ID:                  MempoolChannel,
 		MessageType:         new(pb.Message),
 		Priority:            5,
-		RecvMessageCapacity: batchMsg.Size(),
+		RecvMessageCapacity: mempoolRecvMessageCapacity(),
+		DiscardOversized:    true,
 		RecvBufferCapacity:  128,
 		Name:                "mempool",
 	}
@@ -239,11 +243,18 @@ func (r *Reactor) broadcastTxRoutine(ctx context.Context, peerID types.NodeID) {
 		}
 		for {
 			tx := next.Value()
-			r.channel.Send(&pb.Message{
-				Sum: &pb.Message_Txs{
-					Txs: &pb.Txs{Txs: [][]byte{tx}},
-				},
-			}, peerID)
+			if len(tx) > types.MaxGossipTxBytes {
+				logger.Debug("skipping gossip of tx above protocol size",
+					"tx", tx.Hash(),
+					"size", len(tx),
+					"peer", peerID)
+			} else {
+				r.channel.Send(&pb.Message{
+					Sum: &pb.Message_Txs{
+						Txs: &pb.Txs{Txs: [][]byte{tx}},
+					},
+				}, peerID)
+			}
 
 			next, err = next.NextWait(ctx)
 			if err != nil {