Merge pull request #977 from onflow/vishal/port-971-nonce-mempool-soft-finality

Port nonce-aware tx mempool (#971) to the soft-finality branch

Author: vishalchangrani

README.md

@@ -265,6 +265,12 @@ The application can be configured using the following flags at runtime:
 | `profiler-host`                | `localhost`      | Host for the pprof profiler                                                                |
 | `profiler-port`                | `6060`           | Port for the pprof profiler                                                                |
 | `tx-state-validation`          | `""`             | When set to `local-index` will validate EVM transaction state locally                      |
+| `tx-mempool-mode`              | `false`          | Enable the nonce-aware transaction mempool: expected-nonce transactions are submitted immediately, out-of-order transactions are held per-EOA until their nonce gap fills. Mutually exclusive with `tx-batch-mode` and requires `tx-state-validation=local-index`. |
+| `tx-collection-window`         | `300ms`          | Per-EOA sliding collection window for the transaction mempool. Resets on each arrival from the same EOA. Only applies when `tx-mempool-mode=true`.                        |
+| `tx-submission-spacing`        | `1200ms`         | 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. Must be >= `tx-collection-window`. Only applies when `tx-mempool-mode=true`. |
+| `tx-pool-ttl`                  | `30s`            | How long the transaction mempool holds an out-of-order transaction waiting for its nonce gap to fill, before submitting it anyway. Only applies when `tx-mempool-mode=true`.                                       |
+| `tx-max-batch-size`            | `5`              | Maximum number of EVM transactions per `EVM.batchRun` Cadence transaction in the transaction mempool. Only applies when `tx-mempool-mode=true`.                                                                        |
+| `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` disables the upper bound. A nonce below the indexed nonce is always rejected as nonce-too-low. Only applies when `tx-mempool-mode=true`. |
 
 
 # EVM Gateway Endpoints

bootstrap/bootstrap.go

@@ -269,11 +269,29 @@ 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 {
+		nonceProvider := requester.NewLocalNonceProvider(
+			b.config.FlowNetworkID,
+			b.storages.Registers,
+			b.storages.Blocks,
+			b.collector,
 		)
 		txPool, err = requester.NewBatchTxPool(
 			ctx,

cmd/run/cmd.go

@@ -230,6 +230,33 @@ func parseConfigFromFlags() error {
 	cfg.ExperimentalSoftFinalityEnabled = experimentalSoftFinalityEnabled
 	cfg.ExperimentalSealingVerificationEnabled = experimentalSealingVerificationEnabled
 
+	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
 }
 
@@ -299,6 +326,12 @@ func init() {
 	Cmd.Flags().BoolVar(&experimentalSoftFinalityEnabled, "experimental-soft-finality-enabled", false, "Sets whether the gateway should use the experimental soft finality feature. WARNING: This may result in incorrect results being returned in certain circumstances. Use only if you know what you are doing.")
 	Cmd.Flags().BoolVar(&experimentalSealingVerificationEnabled, "experimental-sealing-verification-enabled", true, "Sets whether the gateway should use the experimental soft finality sealing verification feature. WARNING: This may result in indexing halts if events do not match. Use only if you know what you are doing.")
 	Cmd.Flags().DurationVar(&cfg.EOAActivityCacheTTL, "eoa-activity-cache-ttl", time.Second*10, "[DEPRECATED] No longer has any effect. BatchTxPool now always pools every transaction before submission.")
+	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

@@ -133,6 +133,37 @@ type Config struct {
 	// parsing so existing deployments with --eoa-activity-cache-ttl set do not break.
 	// Deprecated: has no effect since BatchTxPool now always pools every transaction.
 	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

@@ -1,37 +1,63 @@
 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"
 )
 
-// NonceProvider returns the current nonce of the given EOA address.
-// The transaction mempool uses it to determine the expected next nonce.
+// 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 {
-	// GetNonce returns the current nonce of the given EOA address.
+	// 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.
-	GetNonce(address gethCommon.Address) (uint64, error)
+	GetNextNonce(address gethCommon.Address) (uint64, error)
 
-	// GetBlockView provides query capabilities over a specific state of the EVM chain.
-	GetBlockView() (*query.View, 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.
+// 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{}
@@ -40,20 +66,44 @@ 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,
 	}
 }
 
-func (p *LocalNonceProvider) GetBlockView() (*query.View, error) {
+// 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),
@@ -67,13 +117,19 @@ func (p *LocalNonceProvider) GetBlockView() (*query.View, error) {
 		return nil, err
 	}
 
+	p.mu.Lock()
+	p.cachedView = view
+	p.cachedHeight = height
+	p.mu.Unlock()
+
 	return view, nil
 }
 
-func (p *LocalNonceProvider) GetNonce(address gethCommon.Address) (uint64, error) {
+func (p *LocalNonceProvider) GetNextNonce(address gethCommon.Address) (uint64, error) {
 	view, err := p.GetBlockView()
 	if err != nil {
 		return 0, err
 	}
+
 	return view.GetNonce(address)
 }