Bound concurrent CheckTx on the producer ingest path

sei-tendermint/config/autobahn.go

@@ -3,6 +3,7 @@ package config
 import (
 	"errors"
 	"fmt"
+	"math"
 	"net/url"
 
 	"github.com/sei-protocol/sei-chain/sei-db/ledger_db/block/littblock"
@@ -73,6 +74,11 @@ type AutobahnFileConfig struct {
 	// fullnodes serving downstream block-sync are subject to the same
 	// cap). Absent ⇒ DefaultMaxInboundFullnodePeers. Some(0) ⇒ reject all.
 	MaxInboundFullnodePeers utils.Option[uint64] `json:"max_inbound_fullnode_peers,omitzero"`
+	// MaxConcurrentCheckTx caps the number of CheckTx calls the local mempool
+	// runs concurrently for incoming broadcast_tx requests, so that ingest
+	// cannot starve the consensus and data loops of CPU.
+	// Absent ⇒ half of GOMAXPROCS (at least 1).
+	MaxConcurrentCheckTx utils.Option[uint64] `json:"max_concurrent_check_tx,omitzero"`
 	// Whether validators proxy mempool EVM RPC requests to the validator
 	// handling a given shard of addresses.
 	// No-op on fullnodes: they do not have a local mempool, so EVM RPC
@@ -126,6 +132,9 @@ func (fc *AutobahnFileConfig) Validate() error {
 	if fc.DialInterval <= 0 {
 		return errors.New("dial_interval must be > 0")
 	}
+	if v, ok := fc.MaxConcurrentCheckTx.Get(); ok && (v == 0 || v > math.MaxInt32) {
+		return fmt.Errorf("max_concurrent_check_tx must be in 1..%d when set", math.MaxInt32)
+	}
 	if err := fc.BlockDB.Validate(); err != nil {
 		return fmt.Errorf("block_db: %w", err)
 	}

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

@@ -0,0 +1,174 @@
+package producer
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"runtime"
+	"testing"
+
+	abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/proxy"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/scope"
+)
+
+type gatedCheckTx struct {
+	open     bool
+	inflight int
+	maxSeen  int
+}
+
+// gatedApp is a testApp whose CheckTx blocks until the gate is opened,
+// tracking how many CheckTx calls are in flight at once.
+type gatedApp struct {
+	*testApp
+	gate utils.Watch[*gatedCheckTx]
+}
+
+func newGatedApp() *gatedApp {
+	return &gatedApp{
+		testApp: newTestApp(),
+		gate:    utils.NewWatch(&gatedCheckTx{}),
+	}
+}
+
+func (a *gatedApp) Proxy() *proxy.Proxy {
+	return proxy.New(a)
+}
+
+func (a *gatedApp) CheckTx(ctx context.Context, req *abci.RequestCheckTxV2) *abci.ResponseCheckTxV2 {
+	for g, ctrl := range a.gate.Lock() {
+		g.inflight += 1
+		g.maxSeen = max(g.maxSeen, g.inflight)
+		ctrl.Updated()
+		err := ctrl.WaitUntil(ctx, func() bool { return g.open })
+		g.inflight -= 1
+		ctrl.Updated()
+		if err != nil {
+			return &abci.ResponseCheckTxV2{ResponseCheckTx: &abci.ResponseCheckTx{Code: 1, Log: err.Error()}}
+		}
+	}
+	return a.testApp.CheckTx(ctx, req)
+}
+
+// waitInflight blocks until exactly n CheckTx calls are in flight.
+func (a *gatedApp) waitInflight(ctx context.Context, n int) error {
+	for g, ctrl := range a.gate.Lock() {
+		return ctrl.WaitUntil(ctx, func() bool { return g.inflight == n })
+	}
+	panic("unreachable")
+}
+
+func (a *gatedApp) openGate() {
+	for g, ctrl := range a.gate.Lock() {
+		g.open = true
+		ctrl.Updated()
+	}
+}
+
+func (a *gatedApp) maxInflight() int {
+	for g := range a.gate.Lock() {
+		return g.maxSeen
+	}
+	panic("unreachable")
+}
+
+// With limit N and M>N concurrent inserts, at most N CheckTx calls run at once
+// and every tx is still admitted.
+func TestInsertTx_BoundsConcurrentCheckTx(t *testing.T) {
+	for _, limit := range utils.Slice(1, 2, 3) {
+		t.Run(fmt.Sprint(limit), func(t *testing.T) {
+			ctx := t.Context()
+			rng := utils.TestRng()
+			app := newGatedApp()
+			cfg := app.Cfg()
+			cfg.MaxConcurrentCheckTx = utils.Some(uint64(limit)) //nolint:gosec // small test constant
+			env := newTestEnv(rng, cfg, app.Proxy())
+			env.alignLocalMempool()
+
+			const m = 6
+			txs := make([][]byte, 0, m)
+			for range m {
+				addr, nonce := app.NewAccount(rng)
+				txs = append(txs, env.genTx(rng, addr, nonce).encode())
+			}
+			require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error {
+				for _, tx := range txs {
+					s.Spawn(func() error {
+						_, err := env.state.InsertTx(ctx, tx)
+						return err
+					})
+				}
+				if err := app.waitInflight(ctx, limit); err != nil {
+					return err
+				}
+				app.openGate()
+				return nil
+			}))
+			require.Equal(t, limit, app.maxInflight())
+			require.Equal(t, m, len(env.state.UnconfirmedTxs()))
+		})
+	}
+}
+
+// Waiters cancelled while queued for a CheckTx permit return ctx error and
+// do not consume a permit, so later inserts still get through.
+func TestInsertTx_CancelledCheckTxWaiterReleasesPermit(t *testing.T) {
+	ctx := t.Context()
+	rng := utils.TestRng()
+	app := newGatedApp()
+	cfg := app.Cfg()
+	cfg.MaxConcurrentCheckTx = utils.Some[uint64](1)
+	env := newTestEnv(rng, cfg, app.Proxy())
+	env.alignLocalMempool()
+
+	newTx := func() []byte {
+		addr, nonce := app.NewAccount(rng)
+		return env.genTx(rng, addr, nonce).encode()
+	}
+	require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error {
+		holder := newTx()
+		s.Spawn(func() error {
+			_, err := env.state.InsertTx(ctx, holder)
+			return err
+		})
+		// The single permit is now held by holder, blocked in CheckTx.
+		if err := app.waitInflight(ctx, 1); err != nil {
+			return err
+		}
+		waitCtx, cancel := context.WithCancel(ctx)
+		if err := scope.Run(waitCtx, func(ctx context.Context, s scope.Scope) error {
+			for range 4 {
+				tx := newTx()
+				s.Spawn(func() error {
+					_, err := env.state.InsertTx(ctx, tx)
+					if !errors.Is(err, context.Canceled) {
+						return fmt.Errorf("InsertTx() = %v, want context.Canceled", err)
+					}
+					return nil
+				})
+			}
+			cancel()
+			return nil
+		}); err != nil {
+			return err
+		}
+		app.openGate()
+		// Once holder finishes, a fresh insert must acquire the permit.
+		_, err := env.state.InsertTx(ctx, newTx())
+		return err
+	}))
+	require.Equal(t, 1, app.maxInflight())
+	require.Equal(t, 2, len(env.state.UnconfirmedTxs()))
+}
+
+// An absent limit resolves to at least one permit so inserts proceed.
+func TestConfig_MaxConcurrentCheckTxDefault(t *testing.T) {
+	cfg := &Config{}
+	require.True(t, cfg.maxConcurrentCheckTx() >= 1)
+	require.Equal(t, max(1, runtime.GOMAXPROCS(0)/2), cfg.maxConcurrentCheckTx())
+	cfg.MaxConcurrentCheckTx = utils.Some[uint64](7)
+	require.Equal(t, 7, cfg.maxConcurrentCheckTx())
+}

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

@@ -270,14 +270,28 @@ func (s *State) getMempool(ctx context.Context) (*mempool, error) {
 	return mp, nil
 }
 
-// checkTx runs the app CheckTx for tx.
+// checkTx runs the app CheckTx for tx, holding one of cfg.MaxConcurrentCheckTx permits
+// for the duration of the call. Waiting for a permit is cancelled with ctx.
 func (s *State) checkTx(ctx context.Context, tx tmtypes.Tx) (*abci.ResponseCheckTxV2, error) {
+	release, err := s.acquireCheckTxPermit(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer release()
 	defer metrics.PhaseCheckTx.Enter()()
 	start := time.Now()
 	defer func() { metrics.ObserveCheckTx(time.Since(start)) }()
 	return s.app.CheckTxSafe(ctx, &abci.RequestCheckTxV2{Tx: tx})
 }
 
+// acquireCheckTxPermit takes one CheckTx permit, recording the wait as its own insert phase.
+func (s *State) acquireCheckTxPermit(ctx context.Context) (func(), error) {
+	defer metrics.PhaseCheckTxWait.Enter()()
+	start := time.Now()
+	defer func() { metrics.ObserveCheckTxWait(time.Since(start)) }()
+	return s.checkTxSem.Acquire(ctx)
+}
+
 // evmNonce reads the executed nonce of addr from the app.
 func (s *State) evmNonce(addr common.Address) uint64 {
 	start := time.Now()

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

@@ -11,7 +11,9 @@ const MetricsSubsystem = "internal_autobahn_producer"
 
 //go:generate go run github.com/sei-protocol/sei-chain/sei-tendermint/scripts/metricsgen -struct=metrics
 type metrics struct {
-	// Duration of the app CheckTx call made for each inserted tx.
+	// Time an insert spent waiting for a CheckTx concurrency permit.
+	checkTxWaitLatency prometheus.HistogramVec `metrics_buckets:"exp(0.00001, 2, 30)"`
+	// Duration of the app CheckTx call made for each inserted tx, excluding time spent waiting for a permit.
 	checkTxLatency prometheus.HistogramVec `metrics_buckets:"exp(0.00001, 2, 30)"`
 	// Duration of the app nonce lookup made for a sender not yet tracked by the mempool.
 	nonceLookupLatency prometheus.HistogramVec `metrics_buckets:"exp(0.00001, 2, 30)"`
@@ -31,6 +33,7 @@ type metrics struct {
 type Phase struct{ gauge *prometheus.GaugeInt }
 
 var (
+	PhaseCheckTxWait  = Phase{Global.inFlightAt("check_tx_wait")}
 	PhaseCheckTx      = Phase{Global.inFlightAt("check_tx")}
 	PhaseCapacityWait = Phase{Global.inFlightAt("capacity_wait")}
 	PhaseAdmit        = Phase{Global.inFlightAt("admit")}
@@ -58,6 +61,9 @@ var (
 // Observe counts one insert finishing with this outcome.
 func (r Result) Observe() { r.counter.Add(1) }
 
+// ObserveCheckTxWait records the time one insert waited for a CheckTx permit.
+func ObserveCheckTxWait(d time.Duration) { Global.checkTxWaitLatencyAt().Observe(d.Seconds()) }
+
 // ObserveCheckTx records the duration of one CheckTx call.
 func ObserveCheckTx(d time.Duration) { Global.checkTxLatencyAt().Observe(d.Seconds()) }
 

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

@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math"
+	"runtime"
 	"time"
 
 	"github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types"
@@ -27,12 +29,22 @@ type Config struct {
 	// benchmarks with stable throughput, in case execution performance degrades
 	// when overloaded.
 	MaxTxsPerSecond utils.Option[uint64]
+	// Max number of CheckTx calls executed concurrently by InsertTx/TryInsertTx.
+	// None means half of GOMAXPROCS, at least 1.
+	MaxConcurrentCheckTx utils.Option[uint64]
 	// Max number of InsertTx calls blocked waiting for mempool capacity;
 	// further calls fail immediately with a mempool-full error.
 	// 0 means DefaultMaxPendingInserts.
 	MaxPendingInserts uint64
 }
 
+func (c *Config) maxConcurrentCheckTx() int {
+	if v, ok := c.MaxConcurrentCheckTx.Get(); ok && v > 0 {
+		return int(min(v, math.MaxInt32))
+	}
+	return max(1, runtime.GOMAXPROCS(0)/2)
+}
+
 // DefaultMaxPendingInserts is the Config.MaxPendingInserts used when the field is 0.
 const DefaultMaxPendingInserts uint64 = 4096
 
@@ -55,16 +67,19 @@ type State struct {
 	app       *proxy.Proxy
 	mempool   utils.AtomicSend[utils.Option[*mempool]] // None when not producing
 	consensus *consensus.State
+	// checkTxSem bounds concurrent CheckTx calls on the insert path.
+	checkTxSem *utils.Semaphore
 }
 
 // NewState constructs a new block producer state.
 // Mempool starts None; alignMempool creates it for each produce session.
 func NewState(cfg *Config, consensus *consensus.State, app *proxy.Proxy) *State {
 	return &State{
-		cfg:       cfg,
-		app:       app,
-		mempool:   utils.NewAtomicSend(utils.None[*mempool]()),
-		consensus: consensus,
+		cfg:        cfg,
+		app:        app,
+		mempool:    utils.NewAtomicSend(utils.None[*mempool]()),
+		consensus:  consensus,
+		checkTxSem: utils.NewSemaphore(cfg.maxConcurrentCheckTx()),
 	}
 }
 

sei-tendermint/internal/rpc/core/mempool.go

@@ -64,7 +64,8 @@ func (env *Environment) BroadcastTxAsync(ctx context.Context, req *coretypes.Req
 		if !ok {
 			return nil, errors.New("autobahn fullnode has no local mempool; broadcast_tx_* must be sent to a validator")
 		}
-		go func() { _, _ = v.TryInsertTx(ctx, req.Tx) }()
+		// The request ctx is cancelled as soon as the handler returns; the insert must outlive it.
+		go func() { _, _ = v.TryInsertTx(context.WithoutCancel(ctx), req.Tx) }()
 		return &coretypes.ResultBroadcastTx{Hash: req.Tx.Hash().Bytes()}, nil
 	}
 	mp, err := env.requireMempool()

sei-tendermint/node/setup.go

@@ -270,6 +270,7 @@ func buildValidatorGigaConfig(
 			MaxTxsPerSecond:         fc.MaxTxsPerSecond,
 			AllowEmptyBlocks:        fc.AllowEmptyBlocks,
 			BlockInterval:           time.Duration(fc.BlockInterval),
+			MaxConcurrentCheckTx:    fc.MaxConcurrentCheckTx,
 			MaxPendingInserts:       producer.DefaultMaxPendingInserts,
 		},
 	}, nil