diff --git a/sei-tendermint/config/autobahn.go b/sei-tendermint/config/autobahn.go index 4bd98509a5..78f68df76f 100644 --- a/sei-tendermint/config/autobahn.go +++ b/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) } diff --git a/sei-tendermint/internal/autobahn/producer/checktx_limit_test.go b/sei-tendermint/internal/autobahn/producer/checktx_limit_test.go new file mode 100644 index 0000000000..2b3b2de96d --- /dev/null +++ b/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()) +} diff --git a/sei-tendermint/internal/autobahn/producer/mempool.go b/sei-tendermint/internal/autobahn/producer/mempool.go index 750579851f..b5f21cd047 100644 --- a/sei-tendermint/internal/autobahn/producer/mempool.go +++ b/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() diff --git a/sei-tendermint/internal/autobahn/producer/metrics/metrics.gen.go b/sei-tendermint/internal/autobahn/producer/metrics/metrics.gen.go index 77c253497f..bdd94f1d2e 100644 --- a/sei-tendermint/internal/autobahn/producer/metrics/metrics.gen.go +++ b/sei-tendermint/internal/autobahn/producer/metrics/metrics.gen.go @@ -11,6 +11,7 @@ var Global = newMetrics() func init() { prometheus.MustRegister( + Global.checkTxWaitLatency, Global.checkTxLatency, Global.nonceLookupLatency, Global.capacityWaitLatency, @@ -23,11 +24,18 @@ func init() { func newMetrics() *metrics { return &metrics{ + checkTxWaitLatency: tmprometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: MetricsNamespace, + Subsystem: MetricsSubsystem, + Name: "check_tx_wait_latency", + Help: "Time an insert spent waiting for a CheckTx concurrency permit.", + Buckets: prometheus.ExponentialBuckets(0.00001, 2, 30), + }, nil), checkTxLatency: tmprometheus.NewHistogramVec(prometheus.HistogramOpts{ Namespace: MetricsNamespace, Subsystem: MetricsSubsystem, Name: "check_tx_latency", - Help: "Duration of the app CheckTx call made for each inserted tx.", + Help: "Duration of the app CheckTx call made for each inserted tx, excluding time spent waiting for a permit.", Buckets: prometheus.ExponentialBuckets(0.00001, 2, 30), }, nil), nonceLookupLatency: tmprometheus.NewHistogramVec(prometheus.HistogramOpts{ @@ -72,6 +80,10 @@ func newMetrics() *metrics { } } +func (m *metrics) checkTxWaitLatencyAt() *tmprometheus.Histogram { + return m.checkTxWaitLatency.WithLabelValues() +} + func (m *metrics) checkTxLatencyAt() *tmprometheus.Histogram { return m.checkTxLatency.WithLabelValues() } diff --git a/sei-tendermint/internal/autobahn/producer/metrics/metrics.go b/sei-tendermint/internal/autobahn/producer/metrics/metrics.go index 6ef3dfd7ea..d1caddd2fd 100644 --- a/sei-tendermint/internal/autobahn/producer/metrics/metrics.go +++ b/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()) } diff --git a/sei-tendermint/internal/autobahn/producer/state.go b/sei-tendermint/internal/autobahn/producer/state.go index 2ba6cb7883..529d20b5a1 100644 --- a/sei-tendermint/internal/autobahn/producer/state.go +++ b/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()), } } diff --git a/sei-tendermint/internal/rpc/core/mempool.go b/sei-tendermint/internal/rpc/core/mempool.go index 51e7c11d46..f57e0e1d88 100644 --- a/sei-tendermint/internal/rpc/core/mempool.go +++ b/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() diff --git a/sei-tendermint/node/setup.go b/sei-tendermint/node/setup.go index 7535ecacc7..2d571f4730 100644 --- a/sei-tendermint/node/setup.go +++ b/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