From b9bf7d64a2546ebb8f07b48da211f68da294ee00 Mon Sep 17 00:00:00 2001 From: masih Date: Tue, 15 Sep 2026 13:49:52 +0000 Subject: [PATCH 1/2] FIFO admission queue with a waiter bound for blocked InsertTx calls --- .../internal/autobahn/producer/mempool.go | 190 +++++++++++++----- .../autobahn/producer/mempool_test.go | 179 +++++++++++++++++ .../internal/autobahn/producer/state.go | 14 +- .../p2p/giga_router_validator_test.go | 2 + .../internal/rpc/core/autobahn_env_test.go | 1 + sei-tendermint/node/setup.go | 1 + 6 files changed, 335 insertions(+), 52 deletions(-) diff --git a/sei-tendermint/internal/autobahn/producer/mempool.go b/sei-tendermint/internal/autobahn/producer/mempool.go index 13d4f1e359..333f7020de 100644 --- a/sei-tendermint/internal/autobahn/producer/mempool.go +++ b/sei-tendermint/internal/autobahn/producer/mempool.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "slices" "github.com/ethereum/go-ethereum/common" abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types" @@ -32,6 +33,24 @@ type blockSpec struct { // mempool is one produce session. State.mempool publishes it; nil means idle. type mempool struct { inner utils.Watch[*mempoolInner] + // pendingInserts is the number of InsertTx calls blocked in the admission queue. + pendingInserts utils.AtomicSend[uint64] +} + +// insertTicket is the place of one blocked InsertTx call in the admission queue. +// admitted is signalled when the ticket is at the head of the queue and the +// mempool has capacity, or when the mempool is closed. +type insertTicket struct { + admitted utils.AtomicSend[bool] +} + +func newInsertTicket() *insertTicket { + return &insertTicket{admitted: utils.NewAtomicSend(false)} +} + +func (t *insertTicket) wait(ctx context.Context) error { + _, err := t.admitted.Wait(ctx, func(admitted bool) bool { return admitted }) + return err } // mempoolInner is the lock-protected session state. The Watch value is fixed for @@ -46,6 +65,9 @@ type mempoolInner struct { nextBlock *blockSpec evmNonces map[common.Address]uint64 evmTxs map[common.Hash]tmtypes.Tx + // waiters is the FIFO admission queue of blocked InsertTx calls; only the head is + // ever signalled, so an update costs O(1) regardless of the queue length. + waiters []*insertTicket } func newMempoolInner(capacity uint64, lane types.LaneID, n types.BlockNumber) *mempoolInner { @@ -61,6 +83,48 @@ func newMempoolInner(capacity uint64, lane types.LaneID, n types.BlockNumber) *m } } +// close ends the session and releases every blocked InsertTx call. +func (m *mempoolInner) close(ctrl *utils.WatchCtrl) { + m.closed = true + for _, t := range m.waiters { + t.admitted.Store(true) + } + ctrl.Updated() +} + +// isHead reports whether ticket is the next call to be admitted: a call without a +// ticket is admitted only when nobody is queued ahead of it. +func (m *mempoolInner) isHead(ticket utils.Option[*insertTicket]) bool { + t, ok := ticket.Get() + if !ok { + return len(m.waiters) == 0 + } + return m.waiters[0] == t +} + +// signalHead wakes the oldest blocked InsertTx call if the mempool has capacity. +func (m *mempoolInner) signalHead() { + if len(m.waiters) > 0 && !m.IsFull() { + m.waiters[0].admitted.Store(true) + } +} + +func (mp *mempool) enqueue(m *mempoolInner) *insertTicket { + t := newInsertTicket() + m.waiters = append(m.waiters, t) + mp.pendingInserts.Store(uint64(len(m.waiters))) + return t +} + +// dequeue removes t from the admission queue and passes the turn to the next waiter. +func (mp *mempool) dequeue(m *mempoolInner, t *insertTicket) { + if i := slices.Index(m.waiters, t); i >= 0 { + m.waiters = slices.Delete(m.waiters, i, i+1) + mp.pendingInserts.Store(uint64(len(m.waiters))) + } + m.signalHead() +} + func (m *mempoolInner) IsFull() bool { return uint64(m.next-m.first) >= m.capacity && len(m.nextBlock.txs) > 0 } @@ -157,6 +221,7 @@ func (s *State) pruneMempool(mp *mempool, n types.BlockNumber) { // because local mempool is the only source of local lane blocks, // but we handle it gracefully anyway. m.next = max(m.next, n) + m.signalHead() } } @@ -165,7 +230,8 @@ func (s *State) TryInsertTx(ctx context.Context, tx tmtypes.Tx) (*abci.ResponseC return s.insertTx(ctx, tx, false) } -// InsertTx inserts tx to the mempool. Blocks if mempool is full. +// InsertTx inserts tx to the mempool. Blocks if mempool is full, admitting blocked +// calls in arrival order; returns errMempoolFull once Config.MaxPendingInserts calls are blocked. // The blocked calls are effectively the "unsequenced" part of the mempool. // After InsertTx returns, the sequence is already scheduled to be included in a lane. // TODO(gprusak): we might need some prioritization mechanism in case our node can handle more InsertTx calls/s @@ -244,63 +310,93 @@ func (s *State) insertTx(ctx context.Context, tx tmtypes.Tx, waitIfFull bool) (* return nil, errTooLarge } - for m, ctrl := range mp.inner.Lock() { - if m.closed { - return nil, ErrNotProducing - } - if m.IsFull() && !waitIfFull { - return nil, errMempoolFull - } - for m.IsFull() { - // mempool is constructed as a FIFO - we do not delay insertions of large txs (going over cap) - // in favor of waiting for smaller txs. This simple algorithm allows us to cap - // pending txs to size of a single block. We can refine this rule later if needed. - // NOTE: in case there are N concurrent InsertTx calls, this condition is reevaluated N times - // every time mempool is updated. Depending on proportion of N to the block size it might get too - // expensive. - if err := ctrl.Wait(ctx); err != nil { + // mempool is constructed as a FIFO - we do not delay insertions of large txs (going over cap) + // in favor of waiting for smaller txs. This simple algorithm allows us to cap + // pending txs to size of a single block. We can refine this rule later if needed. + // Blocked calls queue up in arrival order and only the head is woken when capacity + // frees up, so a mempool update costs O(1) regardless of the number of waiters. + ticket := utils.None[*insertTicket]() + for { + if t, ok := ticket.Get(); ok { + if err := t.wait(ctx); err != nil { + for m := range mp.inner.Lock() { + mp.dequeue(m, t) + } return nil, err } + } + for m, ctrl := range mp.inner.Lock() { if m.closed { + if t, ok := ticket.Get(); ok { + mp.dequeue(m, t) + } return nil, ErrNotProducing } - } - if resp.IsEVM { - addr := resp.EVMSenderAddress - nonce, ok := m.evmNonces[addr] - if !ok { - nonce = s.app.EvmNonce(addr) + if m.IsFull() && !waitIfFull { + return nil, errMempoolFull + } + if m.IsFull() || (waitIfFull && !m.isHead(ticket)) { + if t, ok := ticket.Get(); ok { + // A TryInsertTx may have filled the mempool since this ticket was signalled. + t.admitted.Store(false) + } else { + if uint64(len(m.waiters)) >= s.cfg.MaxPendingInserts { + return nil, errMempoolFull + } + ticket = utils.Some(mp.enqueue(m)) + } + continue } - if nonce != resp.EVMNonce { - return nil, fmt.Errorf("%w: got %v, want %v", errBadNonce, resp.EVMNonce, nonce) + err := s.appendTx(m, ctrl, tx, resp, gasWanted, gasEstimated) + if t, ok := ticket.Get(); ok { + mp.dequeue(m, t) } - m.evmNonces[addr] = nonce + 1 + if err != nil { + return nil, err + } + return resp.ResponseCheckTx, nil } - // If any limit would be exceeded, then construct a payload. - // Note that we use subtraction in a way avoiding arithmetic overflows. - ok := s.cfg.maxTxsPerBlock()-uint64(len(m.nextBlock.txs)) >= 1 - ok = ok && types.MaxTxsBytesPerBlock-m.nextBlock.sizeBytes >= uint64(len(tx)) - ok = ok && s.cfg.MaxGasWantedPerBlock-m.nextBlock.gasWanted >= gasWanted - ok = ok && s.cfg.MaxGasEstimatedPerBlock-m.nextBlock.gasEstimated >= gasEstimated + } +} + +// appendTx adds an admitted tx to the next lane block, sealing the current one first when the +// tx would exceed one of its limits. Must be called with the mempool locked and not full. +func (s *State) appendTx(m *mempoolInner, ctrl *utils.WatchCtrl, tx tmtypes.Tx, resp *abci.ResponseCheckTxV2, gasWanted, gasEstimated uint64) error { + if resp.IsEVM { + addr := resp.EVMSenderAddress + nonce, ok := m.evmNonces[addr] if !ok { - m.SealBlock() + nonce = s.app.EvmNonce(addr) } - if len(m.nextBlock.txs) == 0 { - // We notify that we start a new block. - ctrl.Updated() + if nonce != resp.EVMNonce { + return fmt.Errorf("%w: got %v, want %v", errBadNonce, resp.EVMNonce, nonce) } + m.evmNonces[addr] = nonce + 1 + } + // If any limit would be exceeded, then construct a payload. + // Note that we use subtraction in a way avoiding arithmetic overflows. + ok := s.cfg.maxTxsPerBlock()-uint64(len(m.nextBlock.txs)) >= 1 + ok = ok && types.MaxTxsBytesPerBlock-m.nextBlock.sizeBytes >= uint64(len(tx)) + ok = ok && s.cfg.MaxGasWantedPerBlock-m.nextBlock.gasWanted >= gasWanted + ok = ok && s.cfg.MaxGasEstimatedPerBlock-m.nextBlock.gasEstimated >= gasEstimated + if !ok { + m.SealBlock() + } + if len(m.nextBlock.txs) == 0 { + // We notify that we start a new block. + ctrl.Updated() + } - b := m.nextBlock - b.gasEstimated += utils.Clamp[uint64](gasEstimated) - b.gasWanted += utils.Clamp[uint64](resp.GasWanted) - b.sizeBytes += uint64(len(tx)) - b.txs = append(b.txs, tx) - if resp.IsEVM { - addr := resp.EVMSenderAddress - b.evmNonces[addr] = m.evmNonces[addr] - b.evmHashes = append(b.evmHashes, resp.EVMHash) - m.evmTxs[resp.EVMHash] = tx - } + b := m.nextBlock + b.gasEstimated += utils.Clamp[uint64](gasEstimated) + b.gasWanted += utils.Clamp[uint64](resp.GasWanted) + b.sizeBytes += uint64(len(tx)) + b.txs = append(b.txs, tx) + if resp.IsEVM { + addr := resp.EVMSenderAddress + b.evmNonces[addr] = m.evmNonces[addr] + b.evmHashes = append(b.evmHashes, resp.EVMHash) + m.evmTxs[resp.EVMHash] = tx } - return resp.ResponseCheckTx, nil + return nil } diff --git a/sei-tendermint/internal/autobahn/producer/mempool_test.go b/sei-tendermint/internal/autobahn/producer/mempool_test.go index 4da150ac16..54369f37e2 100644 --- a/sei-tendermint/internal/autobahn/producer/mempool_test.go +++ b/sei-tendermint/internal/autobahn/producer/mempool_test.go @@ -117,6 +117,7 @@ func (a *testApp) Cfg() *Config { MaxGasEstimatedPerBlock: 1000000, MaxTxsPerBlock: types.MaxTxsPerBlock, BlockInterval: time.Hour, + MaxPendingInserts: DefaultMaxPendingInserts, } } @@ -660,3 +661,181 @@ func TestInsertTx_WaitUnblocksOnLeave(t *testing.T) { t.Fatal("InsertTx did not unblock after leave") } } + +// fullTx returns a tx that alone fills a block, so every admitted one seals the previous block. +func (env *testEnv) fullTx(rng utils.Rng, app *testApp) *txSpec { + addr, nonce := app.NewAccount(rng) + tx := env.genTx(rng, addr, nonce) + tx.GasWanted = env.state.cfg.MaxGasWantedPerBlock + tx.GasEstimated = tx.GasWanted + return tx +} + +// fillMempool installs a session mempool and inserts txs until it is full. +func (env *testEnv) fillMempool(ctx context.Context, rng utils.Rng, app *testApp) (*mempool, error) { + env.alignLocalMempool() + mp := env.state.mempool.Load().OrPanic("aligned") + for range avail.BlocksPerLane + 1 { + if _, err := env.state.TryInsertTx(ctx, env.fullTx(rng, app).encode()); err != nil { + return nil, err + } + } + if _, err := env.state.TryInsertTx(ctx, env.fullTx(rng, app).encode()); !errors.Is(err, errMempoolFull) { + return nil, fmt.Errorf("TryInsertTx on full mempool: got %v, want errMempoolFull", err) + } + return mp, nil +} + +// freeOneBlock prunes the oldest lane block, making room for exactly one more sealed block. +func (env *testEnv) freeOneBlock(mp *mempool) { + var first types.BlockNumber + for m := range mp.inner.Lock() { + first = m.first + } + env.state.pruneMempool(mp, first+1) +} + +// waitPending blocks until at most n InsertTx calls are queued and returns the exact count. +func waitPending(ctx context.Context, mp *mempool, n uint64) (uint64, error) { + return mp.pendingInserts.Wait(ctx, func(got uint64) bool { return got <= n }) +} + +// spawnInserter spawns an InsertTx call and returns the first error different from want. +func (env *testEnv) spawnInserter(ctx context.Context, s scope.Scope, tx *txSpec, want error) { + s.Spawn(func() error { + _, err := env.state.InsertTx(ctx, tx.encode()) + if !errors.Is(err, want) { + return fmt.Errorf("InsertTx(): got %v, want %v", err, want) + } + return nil + }) +} + +// enqueueInserters spawns n blocked InsertTx calls one at a time, so the queue order is known. +func (env *testEnv) enqueueInserters(ctx context.Context, s scope.Scope, rng utils.Rng, app *testApp, mp *mempool, n int, want error) ([]*txSpec, error) { + txs := make([]*txSpec, 0, n) + pending := mp.pendingInserts.Load() + for range n { + tx := env.fullTx(rng, app) + txs = append(txs, tx) + env.spawnInserter(ctx, s, tx, want) + pending += 1 + if _, err := mp.pendingInserts.Wait(ctx, func(got uint64) bool { return got == pending }); err != nil { + return nil, err + } + } + return txs, nil +} + +// Blocked InsertTx calls are admitted in arrival order, one per freed block. +func TestInsertTx_FIFOAdmission(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + app := newTestApp() + env := newTestEnv(rng, app.Cfg(), app.Proxy()) + mp, err := env.fillMempool(ctx, rng, app) + require.NoError(t, err) + + const n = 5 + require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + txs, err := env.enqueueInserters(ctx, s, rng, app, mp, n, nil) + if err != nil { + return err + } + for i, tx := range txs { + env.freeOneBlock(mp) + got, err := waitPending(ctx, mp, uint64(n-i-1)) + if err != nil { + return err + } + if got != uint64(n-i-1) { + return fmt.Errorf("pending after freeing block %d: got %d, want %d", i, got, n-i-1) + } + if want := [][]byte{tx.encode()}; !slices.EqualFunc(env.state.UnconfirmedTxs(), want, slices.Equal) { + return fmt.Errorf("admitted tx %d out of order", i) + } + } + return nil + })) +} + +// A cancelled waiter leaves the queue without holding up the ones behind it. +func TestInsertTx_CancelledWaiterLeavesQueue(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + app := newTestApp() + env := newTestEnv(rng, app.Cfg(), app.Proxy()) + mp, err := env.fillMempool(ctx, rng, app) + require.NoError(t, err) + + require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + headCtx, cancelHead := context.WithCancel(ctx) + defer cancelHead() + env.spawnInserter(headCtx, s, env.fullTx(rng, app), context.Canceled) + if _, err := mp.pendingInserts.Wait(ctx, func(got uint64) bool { return got == 1 }); err != nil { + return err + } + txs, err := env.enqueueInserters(ctx, s, rng, app, mp, 2, nil) + if err != nil { + return err + } + cancelHead() + if _, err := waitPending(ctx, mp, 2); err != nil { + return err + } + for i, tx := range txs { + env.freeOneBlock(mp) + if _, err := waitPending(ctx, mp, uint64(1-i)); err != nil { + return err + } + if want := [][]byte{tx.encode()}; !slices.EqualFunc(env.state.UnconfirmedTxs(), want, slices.Equal) { + return fmt.Errorf("admitted tx %d out of order", i) + } + } + return nil + })) +} + +// Closing the mempool fails every queued InsertTx call with ErrNotProducing. +func TestInsertTx_WaitersReleasedOnClose(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + app := newTestApp() + env := newTestEnv(rng, app.Cfg(), app.Proxy()) + mp, err := env.fillMempool(ctx, rng, app) + require.NoError(t, err) + + require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + if _, err := env.enqueueInserters(ctx, s, rng, app, mp, 3, ErrNotProducing); err != nil { + return err + } + env.state.clearMempool() + return nil + })) +} + +// Once MaxPendingInserts calls are blocked, InsertTx fails immediately with errMempoolFull. +func TestInsertTx_PendingInsertsBounded(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + app := newTestApp() + cfg := app.Cfg() + cfg.MaxPendingInserts = 2 + env := newTestEnv(rng, cfg, app.Proxy()) + mp, err := env.fillMempool(ctx, rng, app) + require.NoError(t, err) + + require.NoError(t, utils.IgnoreCancel(scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + if _, err := env.enqueueInserters(ctx, s, rng, app, mp, 2, context.Canceled); err != nil { + return err + } + if _, err := env.state.InsertTx(ctx, env.fullTx(rng, app).encode()); !errors.Is(err, errMempoolFull) { + return fmt.Errorf("InsertTx over the bound: got %v, want errMempoolFull", err) + } + if got := mp.pendingInserts.Load(); got != 2 { + return fmt.Errorf("pending: got %d, want 2", got) + } + s.Cancel(context.Canceled) + return nil + }))) +} diff --git a/sei-tendermint/internal/autobahn/producer/state.go b/sei-tendermint/internal/autobahn/producer/state.go index a90c13f8a3..bef7177874 100644 --- a/sei-tendermint/internal/autobahn/producer/state.go +++ b/sei-tendermint/internal/autobahn/producer/state.go @@ -27,8 +27,14 @@ type Config struct { // benchmarks with stable throughput, in case execution performance degrades // when overloaded. MaxTxsPerSecond utils.Option[uint64] + // Max number of InsertTx calls blocked waiting for mempool capacity; + // further calls fail immediately with a mempool-full error. + MaxPendingInserts uint64 } +// DefaultMaxPendingInserts is the recommended Config.MaxPendingInserts. +const DefaultMaxPendingInserts uint64 = 4096 + const minTxGas = 21000 func (c *Config) maxTxsPerBlock() uint64 { @@ -58,7 +64,7 @@ func NewState(cfg *Config, consensus *consensus.State, app *proxy.Proxy) *State func (s *State) alignMempool(lane types.LaneID) (*mempool, types.BlockNumber) { n := s.consensus.Avail().NextBlock(lane) m := newMempoolInner(avail.BlocksPerLane, lane, n) - mp := &mempool{inner: utils.NewWatch(m)} + mp := &mempool{inner: utils.NewWatch(m), pendingInserts: utils.NewAtomicSend[uint64](0)} s.mempool.Store(utils.Some(mp)) return mp, n } @@ -71,8 +77,7 @@ func (s *State) clearMempool() { return } for m, ctrl := range mp.inner.Lock() { - m.closed = true - ctrl.Updated() + m.close(ctrl) } } @@ -110,8 +115,7 @@ func (s *State) runMempool(ctx context.Context, availState *avail.State, lane ty scope.SpawnBg(func() error { _ = availState.WaitUntilClosed(ctx, lane) for m, ctrl := range mp.inner.Lock() { - m.closed = true - ctrl.Updated() + m.close(ctrl) } return nil }) diff --git a/sei-tendermint/internal/p2p/giga_router_validator_test.go b/sei-tendermint/internal/p2p/giga_router_validator_test.go index 94ec3e790c..a7694e12ae 100644 --- a/sei-tendermint/internal/p2p/giga_router_validator_test.go +++ b/sei-tendermint/internal/p2p/giga_router_validator_test.go @@ -102,6 +102,7 @@ func TestGigaRouter_FinalizeBlocks(t *testing.T) { MaxTxsPerSecond: utils.None[uint64](), BlockInterval: 100 * time.Millisecond, AllowEmptyBlocks: false, + MaxPendingInserts: producer.DefaultMaxPendingInserts, }, }, cfg.nodeKey, dataState) require.NoError(t, err, "NewGigaValidatorRouter[%v]", i) @@ -272,6 +273,7 @@ func TestGigaRouter_EvmProxy(t *testing.T) { MaxTxsPerBlock: 1, MaxTxsPerSecond: utils.None[uint64](), BlockInterval: time.Second, + MaxPendingInserts: producer.DefaultMaxPendingInserts, }, }, nodeKeys[0], dataState) require.NoError(t, err) diff --git a/sei-tendermint/internal/rpc/core/autobahn_env_test.go b/sei-tendermint/internal/rpc/core/autobahn_env_test.go index e14ddbaf95..2dfc3f7cee 100644 --- a/sei-tendermint/internal/rpc/core/autobahn_env_test.go +++ b/sei-tendermint/internal/rpc/core/autobahn_env_test.go @@ -68,6 +68,7 @@ func newAutobahnBroadcastEnv(t *testing.T) *Environment { MaxTxsPerBlock: 1, MaxTxsPerSecond: utils.None[uint64](), BlockInterval: time.Second, + MaxPendingInserts: producer.DefaultMaxPendingInserts, }, }, nodeKey, dataState) require.NoError(t, err) diff --git a/sei-tendermint/node/setup.go b/sei-tendermint/node/setup.go index 724eb31b3e..7535ecacc7 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), + MaxPendingInserts: producer.DefaultMaxPendingInserts, }, }, nil } From 99d78dcb618d9fc65033d5db9524d94081fc3d37 Mon Sep 17 00:00:00 2001 From: masih Date: Tue, 15 Sep 2026 14:01:55 +0000 Subject: [PATCH 2/2] Default MaxPendingInserts at the accessor and scope InsertTx ordering godoc --- sei-tendermint/internal/autobahn/producer/mempool.go | 7 ++++--- sei-tendermint/internal/autobahn/producer/state.go | 10 +++++++++- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/sei-tendermint/internal/autobahn/producer/mempool.go b/sei-tendermint/internal/autobahn/producer/mempool.go index 333f7020de..111a639781 100644 --- a/sei-tendermint/internal/autobahn/producer/mempool.go +++ b/sei-tendermint/internal/autobahn/producer/mempool.go @@ -230,8 +230,9 @@ func (s *State) TryInsertTx(ctx context.Context, tx tmtypes.Tx) (*abci.ResponseC return s.insertTx(ctx, tx, false) } -// InsertTx inserts tx to the mempool. Blocks if mempool is full, admitting blocked -// calls in arrival order; returns errMempoolFull once Config.MaxPendingInserts calls are blocked. +// InsertTx inserts tx to the mempool. Blocks if mempool is full; blocked InsertTx calls are +// admitted in arrival order relative to each other, but TryInsertTx calls do not queue and may +// take freed capacity ahead of them. Returns errMempoolFull once Config.MaxPendingInserts calls are blocked. // The blocked calls are effectively the "unsequenced" part of the mempool. // After InsertTx returns, the sequence is already scheduled to be included in a lane. // TODO(gprusak): we might need some prioritization mechanism in case our node can handle more InsertTx calls/s @@ -340,7 +341,7 @@ func (s *State) insertTx(ctx context.Context, tx tmtypes.Tx, waitIfFull bool) (* // A TryInsertTx may have filled the mempool since this ticket was signalled. t.admitted.Store(false) } else { - if uint64(len(m.waiters)) >= s.cfg.MaxPendingInserts { + if uint64(len(m.waiters)) >= s.cfg.maxPendingInserts() { return nil, errMempoolFull } ticket = utils.Some(mp.enqueue(m)) diff --git a/sei-tendermint/internal/autobahn/producer/state.go b/sei-tendermint/internal/autobahn/producer/state.go index bef7177874..2ba6cb7883 100644 --- a/sei-tendermint/internal/autobahn/producer/state.go +++ b/sei-tendermint/internal/autobahn/producer/state.go @@ -29,10 +29,11 @@ type Config struct { MaxTxsPerSecond 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 } -// DefaultMaxPendingInserts is the recommended Config.MaxPendingInserts. +// DefaultMaxPendingInserts is the Config.MaxPendingInserts used when the field is 0. const DefaultMaxPendingInserts uint64 = 4096 const minTxGas = 21000 @@ -41,6 +42,13 @@ func (c *Config) maxTxsPerBlock() uint64 { return min(types.MaxTxsPerBlock, c.MaxTxsPerBlock) } +func (c *Config) maxPendingInserts() uint64 { + if c.MaxPendingInserts == 0 { + return DefaultMaxPendingInserts + } + return c.MaxPendingInserts +} + // State is the block producer state. type State struct { cfg *Config