Add nonce-aware transaction pool

bootstrap/bootstrap.go

@@ -245,7 +245,24 @@ func (b *Bootstrap) StartAPIServer(ctx context.Context) error {
 	// create transaction pool
 	var txPool requester.TxPool
 	var err error
-	if b.config.TxBatchMode {
+	if b.config.TxMemPoolMode {
+		nonceProvider := requester.NewLocalNonceProvider(
+			b.config.FlowNetworkID,
+			b.storages.Registers,
+			b.storages.Blocks,
+			b.collector,
+		)
+		txPool, err = requester.NewTxMemPool(
+			ctx,
+			b.client,
+			b.publishers.Transaction,
+			b.logger,
+			b.config,
+			b.collector,
+			b.keystore,
+			nonceProvider,
+		)
+	} else if b.config.TxBatchMode {
 		txPool, err = requester.NewBatchTxPool(
 			ctx,
 			b.client,

cmd/run/cmd.go

@@ -227,6 +227,33 @@ func parseConfigFromFlags() error {
 		return fmt.Errorf("tx-batch-mode should be enabled with tx-state-validation=local-index")
 	}
 
+	if cfg.TxMemPoolMode {
+		if cfg.TxBatchMode {
+			return fmt.Errorf("tx-mempool-mode and tx-batch-mode are mutually exclusive")
+		}
+		if cfg.TxStateValidation != config.LocalIndexValidation {
+			return fmt.Errorf("tx-mempool-mode requires tx-state-validation=local-index")
+		}
+		if cfg.TxCollectionWindow <= 0 {
+			return fmt.Errorf("tx-collection-window must be > 0 when tx-mempool-mode is enabled")
+		}
+		if cfg.TxSubmissionSpacing <= 0 {
+			return fmt.Errorf("tx-submission-spacing must be > 0 when tx-mempool-mode is enabled")
+		}
+		if cfg.TxCollectionWindow > cfg.TxSubmissionSpacing {
+			return fmt.Errorf(
+				"tx-collection-window (%s) must not exceed tx-submission-spacing (%s)",
+				cfg.TxCollectionWindow, cfg.TxSubmissionSpacing,
+			)
+		}
+		if cfg.TxPoolTTL <= 0 {
+			return fmt.Errorf("tx-pool-ttl must be > 0 when tx-mempool-mode is enabled")
+		}
+		if cfg.TxMaxBatchSize < 1 {
+			return fmt.Errorf("tx-max-batch-size must be >= 1 when tx-mempool-mode is enabled")
+		}
+	}
+
 	return nil
 }
 
@@ -291,6 +318,12 @@ func init() {
 	Cmd.Flags().BoolVar(&cfg.TxBatchMode, "tx-batch-mode", false, "Enable batch transaction submission, to avoid nonce mismatch issues for high-volume EOAs.")
 	Cmd.Flags().DurationVar(&cfg.TxBatchInterval, "tx-batch-interval", time.Millisecond*1200, "Time interval upon which to submit the transaction batches to the Flow network.")
 	Cmd.Flags().DurationVar(&cfg.EOAActivityCacheTTL, "eoa-activity-cache-ttl", time.Second*10, "Time interval used to track EOA activity. Tx send more frequently than this interval will be batched. Useful only when batch transaction submission is enabled.")
+	Cmd.Flags().BoolVar(&cfg.TxMemPoolMode, "tx-mempool-mode", false, "Enable the transaction mempool: expected-nonce transactions are submitted immediately, out-of-order transactions are held until their nonce gap fills. Mutually exclusive with --tx-batch-mode and requires --tx-state-validation=local-index.")
+	Cmd.Flags().DurationVar(&cfg.TxCollectionWindow, "tx-collection-window", 300*time.Millisecond, "Per-EOA sliding collection window for the transaction mempool. Resets on each arrival from the same EOA.")
+	Cmd.Flags().DurationVar(&cfg.TxSubmissionSpacing, "tx-submission-spacing", 1200*time.Millisecond, "Minimum gap between consecutive Cadence submissions for the same EOA in the transaction mempool; also serves as the flush deadline for a continuously-fed collection window. Recommended ~1.5x the block production rate.")
+	Cmd.Flags().DurationVar(&cfg.TxPoolTTL, "tx-pool-ttl", 30*time.Second, "How long the transaction mempool holds an out-of-order transaction waiting for its nonce gap to fill, before submitting it anyway.")
+	Cmd.Flags().IntVar(&cfg.TxMaxBatchSize, "tx-max-batch-size", 5, "Maximum number of EVM transactions per EVM.batchRun Cadence transaction in the transaction mempool.")
+	Cmd.Flags().Uint64Var(&cfg.TxMaxNonceGap, "tx-max-nonce-gap", 500, "How far ahead of an EOA's on-chain nonce the transaction mempool accepts a nonce; nonces beyond indexedNonce+gap are rejected as nonce-too-high. 0 means no upper bound. A nonce below the indexed nonce is always rejected as nonce-too-low regardless of this setting.")
 	Cmd.Flags().DurationVar(&cfg.RpcRequestTimeout, "rpc-request-timeout", time.Second*120, "Sets the maximum duration at which JSON-RPC requests should generate a response, before they timeout. The default is 120 seconds.")
 
 	err := Cmd.Flags().MarkDeprecated("init-cadence-height", "This flag is no longer necessary and will be removed in future version. The initial Cadence height is known for testnet/mainnet and this was only required for fresh deployments of EVM Gateway. Once the DB has been initialized, the latest index Cadence height will be used upon start-up.")

config/config.go

@@ -124,6 +124,37 @@ type Config struct {
 	// frequently than this interval will be batched.
 	// Useful only when batch transaction submission is enabled.
 	EOAActivityCacheTTL time.Duration
+	// TxMemPoolMode configures the gateway to use the transaction mempool:
+	// transactions carrying the expected next nonce (with nothing in flight)
+	// are submitted immediately, out-of-order transactions are held until their
+	// nonce gap fills, and consecutive Cadence submissions for the same EOA are
+	// spaced apart to avoid Collection Node re-ordering.
+	TxMemPoolMode bool
+	// TxCollectionWindow is the per-EOA sliding collection window used by the
+	// transaction mempool. The window resets on each new transaction arrival
+	// from the same EOA; when it elapses, the collected transactions are flushed.
+	TxCollectionWindow time.Duration
+	// TxSubmissionSpacing is the minimum gap between two consecutive Cadence
+	// transaction submissions for the same EOA (recommended ~1.5x the block
+	// production rate). It also serves as the flush deadline for a
+	// continuously-fed collection window, anchored at first enqueue.
+	TxSubmissionSpacing time.Duration
+	// TxPoolTTL is how long the transaction mempool holds an out-of-order
+	// transaction waiting for its nonce gap to fill. On expiry the transaction
+	// is submitted anyway, so the failure is observable instead of a silent drop.
+	TxPoolTTL time.Duration
+	// TxMaxBatchSize is the maximum number of EVM transactions submitted in a
+	// single EVM.batchRun Cadence transaction by the transaction mempool,
+	// bounded by the Cadence transaction computation limit.
+	TxMaxBatchSize int
+	// TxMaxNonceGap is how far ahead of an EOA's on-chain nonce the transaction
+	// mempool will accept a nonce. A transaction whose nonce exceeds
+	// indexedNonce + TxMaxNonceGap is rejected up front with ErrNonceTooHigh,
+	// giving the client immediate feedback instead of holding an unexecutable tx
+	// until TTL. 0 means no upper bound (any future nonce is accepted). This
+	// bounds only the upper end: a nonce below the indexed nonce is always
+	// rejected with ErrNonceTooLow regardless of this setting.
+	TxMaxNonceGap uint64
 	// RpcRequestTimeout is the maximum duration at which JSON-RPC requests should generate
 	// a response, before they timeout.
 	RpcRequestTimeout time.Duration

metrics/collector.go

@@ -100,6 +100,26 @@ var flowTotalSupply = prometheus.NewGauge(prometheus.GaugeOpts{
 	Help: "Total supply of FLOW tokens in EVM at a given time (in smallest unit, wei)",
 })
 
+var txPoolQueues = prometheus.NewGauge(prometheus.GaugeOpts{
+	Name: prefixedName("txpool_queues"),
+	Help: "Number of per-EOA queues currently held by the transaction mempool",
+})
+
+var txPoolQueuedTransactions = prometheus.NewGauge(prometheus.GaugeOpts{
+	Name: prefixedName("txpool_queued_transactions"),
+	Help: "Total number of transactions currently held across all queues in the transaction mempool",
+})
+
+var txPoolSubmissions = prometheus.NewCounterVec(prometheus.CounterOpts{
+	Name: prefixedName("txpool_submissions_total"),
+	Help: "Total EVM transaction batches the mempool submitted to Flow, by flush reason (fast-path, consecutive-prefix, ttl-expiry)",
+}, []string{"reason"})
+
+var txPoolNonceViewCache = prometheus.NewCounterVec(prometheus.CounterOpts{
+	Name: prefixedName("txpool_nonce_view_cache_total"),
+	Help: "Block-view cache accesses when reading EOA nonces (hit = reused the view built for the indexed height; miss = rebuilt it)",
+}, []string{"result"})
+
 var metrics = []prometheus.Collector{
 	apiErrors,
 	serverPanicsCounters,
@@ -118,6 +138,10 @@ var metrics = []prometheus.Collector{
 	transactionsDroppedCounter,
 	rateLimitedTransactionsCounter,
 	flowTotalSupply,
+	txPoolQueues,
+	txPoolQueuedTransactions,
+	txPoolSubmissions,
+	txPoolNonceViewCache,
 }
 
 type Collector interface {
@@ -137,6 +161,9 @@ type Collector interface {
 	TransactionsDropped(count int)
 	TransactionRateLimited()
 	FlowTotalSupply(totalSupply *big.Int)
+	TxPoolSize(queues int, queuedTransactions int)
+	TxPoolSubmission(reason string)
+	NonceViewCache(hit bool)
 }
 
 var _ Collector = &DefaultCollector{}
@@ -162,6 +189,10 @@ type DefaultCollector struct {
 	transactionsDroppedCounter     prometheus.Counter
 	rateLimitedTransactionsCounter prometheus.Counter
 	flowTotalSupply                prometheus.Gauge
+	txPoolQueues                   prometheus.Gauge
+	txPoolQueuedTransactions       prometheus.Gauge
+	txPoolSubmissions              *prometheus.CounterVec
+	txPoolNonceViewCache           *prometheus.CounterVec
 }
 
 func NewCollector(logger zerolog.Logger) Collector {
@@ -189,6 +220,10 @@ func NewCollector(logger zerolog.Logger) Collector {
 		transactionsDroppedCounter:     transactionsDroppedCounter,
 		rateLimitedTransactionsCounter: rateLimitedTransactionsCounter,
 		flowTotalSupply:                flowTotalSupply,
+		txPoolQueues:                   txPoolQueues,
+		txPoolQueuedTransactions:       txPoolQueuedTransactions,
+		txPoolSubmissions:              txPoolSubmissions,
+		txPoolNonceViewCache:           txPoolNonceViewCache,
 	}
 }
 
@@ -288,6 +323,23 @@ func (c *DefaultCollector) FlowTotalSupply(totalSupply *big.Int) {
 	c.flowTotalSupply.Set(floatTotalSupply)
 }
 
+func (c *DefaultCollector) TxPoolSize(queues int, queuedTransactions int) {
+	c.txPoolQueues.Set(float64(queues))
+	c.txPoolQueuedTransactions.Set(float64(queuedTransactions))
+}
+
+func (c *DefaultCollector) TxPoolSubmission(reason string) {
+	c.txPoolSubmissions.With(prometheus.Labels{"reason": reason}).Inc()
+}
+
+func (c *DefaultCollector) NonceViewCache(hit bool) {
+	result := "miss"
+	if hit {
+		result = "hit"
+	}
+	c.txPoolNonceViewCache.With(prometheus.Labels{"result": result}).Inc()
+}
+
 func prefixedName(name string) string {
 	return fmt.Sprintf("evm_gateway_%s", name)
 }

metrics/nop.go

@@ -27,3 +27,6 @@ func (c *nopCollector) RequestRateLimited(method string)           {}
 func (c *nopCollector) TransactionsDropped(count int)              {}
 func (c *nopCollector) TransactionRateLimited()                    {}
 func (c *nopCollector) FlowTotalSupply(totalSupply *big.Int)       {}
+func (c *nopCollector) TxPoolSize(queues int, queued int)          {}
+func (c *nopCollector) TxPoolSubmission(reason string)             {}
+func (c *nopCollector) NonceViewCache(hit bool)                    {}

models/errors/errors.go

@@ -32,6 +32,17 @@ var (
 	ErrFailedTransaction    = errors.New("failed transaction")
 	ErrInvalidTransaction   = fmt.Errorf("%w: %w", ErrInvalid, ErrFailedTransaction)
 	ErrDuplicateTransaction = fmt.Errorf("%w: %s", ErrInvalid, "transaction already in pool")
+	// ErrInFlightNonce is returned when a transaction carries a nonce that has
+	// already been submitted to the network and is awaiting execution. Letting
+	// it through would burn Flow fees on a guaranteed nonce-mismatch failure.
+	ErrInFlightNonce = fmt.Errorf("%w: %s", ErrInvalid, "transaction with the same nonce already submitted")
+	// ErrNonceTooLow is returned when a transaction's nonce is below the EOA's
+	// current on-chain nonce: it has already been used and can never execute.
+	ErrNonceTooLow = fmt.Errorf("%w: %s", ErrInvalid, "nonce too low")
+	// ErrNonceTooHigh is returned when a transaction's nonce is more than the
+	// configured maximum gap ahead of the EOA's on-chain nonce. Such a tx cannot
+	// execute until the gap fills, so it is rejected up front for fast feedback.
+	ErrNonceTooHigh = fmt.Errorf("%w: %s", ErrInvalid, "nonce too high")
 
 	// Storage errors
 

services/requester/nonce_provider.go

@@ -0,0 +1,135 @@
+package requester
+
+import (
+	"sync"
+
+	gethCommon "github.com/ethereum/go-ethereum/common"
+	"github.com/onflow/flow-go/fvm/evm"
+	"github.com/onflow/flow-go/fvm/evm/offchain/query"
+	flowGo "github.com/onflow/flow-go/model/flow"
+
+	"github.com/onflow/flow-evm-gateway/metrics"
+	"github.com/onflow/flow-evm-gateway/storage"
+	"github.com/onflow/flow-evm-gateway/storage/pebble"
+)
+
+// NonceView reads EOA nonces at a single, fixed EVM state (one built block
+// view). The mempool reads many EOAs' nonces from one view per flush tick
+// rather than rebuilding the (expensive) view per address. It is an interface
+// so tests can fake it without constructing a real query.View.
+type NonceView interface {
+	// GetNonce returns the EOA's account nonce (the next nonce to use) at this
+	// view's state. Named GetNonce — not GetNextNonce — because this interface is
+	// satisfied directly by flow-go's query.View, whose method is GetNonce.
+	GetNonce(address gethCommon.Address) (uint64, error)
+}
+
+// NonceProvider returns the next nonce of the given EOA address. The transaction
+// mempool uses it to determine the expected next nonce.
+type NonceProvider interface {
+	// GetNextNonce returns the account nonce of the given EOA — its transaction
+	// count, i.e. the next nonce the EOA should use (matches eth_getTransactionCount).
+	//
+	// A non-nil error represents an EXCEPTION, not an expected condition:
+	// the underlying read is a local state-index lookup that should not
+	// fail under normal operation. Callers must therefore treat an error
+	// as a hard failure (reject the transaction / abort the operation)
+	// rather than a routine, recoverable condition to swallow.
+	GetNextNonce(address gethCommon.Address) (uint64, error)
+
+	// GetBlockView returns a NonceView over the latest indexed EVM state. A
+	// non-nil error is an EXCEPTION, same contract as GetNextNonce.
+	GetBlockView() (NonceView, error)
+}
+
+// LocalNonceProvider reads the EOA nonce from the latest height of the
+// local state index. It caches the built block view and reuses it while the
+// indexed height is unchanged (see GetBlockView).
+type LocalNonceProvider struct {
+	chainID       flowGo.ChainID
+	registerStore *pebble.RegisterStorage
+	blocks        storage.BlockIndexer
+	collector     metrics.Collector
+
+	// mu guards only the cached-view slot below (its read and update), NOT the
+	// expensive view build in GetBlockView. Note the cached view itself is shared
+	// across reads; callers that read it concurrently must still serialize (the
+	// mempool does, via its queueMux).
+	mu           sync.Mutex
+	cachedView   NonceView
+	cachedHeight uint64
+}
+
+var _ NonceProvider = &LocalNonceProvider{}
+
+func NewLocalNonceProvider(
+	chainID flowGo.ChainID,
+	registerStore *pebble.RegisterStorage,
+	blocks storage.BlockIndexer,
+	collector metrics.Collector,
+) *LocalNonceProvider {
+	return &LocalNonceProvider{
+		chainID:       chainID,
+		registerStore: registerStore,
+		blocks:        blocks,
+		collector:     collector,
+	}
+}
+
+// GetBlockView returns a NonceView over the latest indexed EVM height. The view
+// is cached and reused while the indexed height is unchanged, so a burst of
+// reads within one block — many Add calls, or a collectDueBatches pass — builds
+// the (expensive) view only once. It is rebuilt when a new block is indexed.
+// Reuse is safe because an EOA's on-chain nonce cannot change without a new
+// block being indexed.
+func (p *LocalNonceProvider) GetBlockView() (NonceView, error) {
+	height, err := p.blocks.LatestEVMHeight()
+	if err != nil {
+		return nil, err
+	}
+
+	// Fast path: reuse the cached view for this indexed height. Only the cache
+	// read (and the update below) is locked — the expensive view build runs
+	// OUTSIDE the lock. A concurrent miss may build the view more than once for
+	// the same height, which is wasteful but correct (same height => same view)
+	// and does not occur under the mempool's serialized (queueMux) access.
+	p.mu.Lock()
+	if p.cachedView != nil && p.cachedHeight == height {
+		view := p.cachedView
+		p.mu.Unlock()
+		p.collector.NonceViewCache(true)
+		return view, nil
+	}
+	p.mu.Unlock()
+
+	p.collector.NonceViewCache(false)
+
+	viewProvider := query.NewViewProvider(
+		p.chainID,
+		evm.StorageAccountAddress(p.chainID),
+		p.registerStore,
+		NewOverridableBlocksProvider(p.blocks, p.chainID, nil),
+		blockGasLimit,
+	)
+
+	view, err := viewProvider.GetBlockView(height)
+	if err != nil {
+		return nil, err
+	}
+
+	p.mu.Lock()
+	p.cachedView = view
+	p.cachedHeight = height
+	p.mu.Unlock()
+
+	return view, nil
+}
+
+func (p *LocalNonceProvider) GetNextNonce(address gethCommon.Address) (uint64, error) {
+	view, err := p.GetBlockView()
+	if err != nil {
+		return 0, err
+	}
+
+	return view.GetNonce(address)
+}