Merge pull request #852 from onflow/mpeter/batch-tx-pool-improvements

Improve performance of `BatchTxPool` for single-tx EOAs

Author: peterargue

cmd/run/cmd.go

@@ -294,4 +294,5 @@ func init() {
 	Cmd.Flags().DurationVar(&cfg.TxRequestLimitDuration, "tx-request-limit-duration", time.Second*3, "Time interval upon which to enforce transaction submission rate limiting.")
 	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.")
 }

config/config.go

@@ -107,4 +107,8 @@ type Config struct {
 	// TxBatchInterval is the time interval upon which to submit the transaction batches to the
 	// Flow network.
 	TxBatchInterval time.Duration
+	// EOAActivityCacheTTL is the 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.
+	EOAActivityCacheTTL time.Duration
 }

services/requester/batch_tx_pool.go

@@ -3,7 +3,6 @@ package requester
 import (
 	"context"
 	"encoding/hex"
-	"fmt"
 	"sort"
 	"sync"
 	"time"
@@ -21,16 +20,7 @@ import (
 	"github.com/onflow/flow-evm-gateway/services/requester/keystore"
 )
 
-const (
-	eoaActivityCacheSize = 10_000
-	// Buffer capacity of the channel used for individual transaction
-	// submission. Since `BatchTxPool.Add()` is called quite frequently,
-	// we don't want sending to the channel to block, until there's a
-	// receive ready. That's why we set a high buffer capacity, to allow
-	// `BatchTxPool.Add()` to do its job quickly and release any resources
-	// held, like locks etc.
-	txChanBufferSize = 1_000
-)
+const eoaActivityCacheSize = 10_000
 
 type pooledEvmTx struct {
 	txPayload cadence.String
@@ -53,13 +43,8 @@ type pooledEvmTx struct {
 type BatchTxPool struct {
 	*SingleTxPool
 	pooledTxs   map[gethCommon.Address][]pooledEvmTx
-	txChan      chan cadence.String
 	txMux       sync.Mutex
 	eoaActivity *expirable.LRU[gethCommon.Address, time.Time]
-
-	// Signal channel used to prevent blocking writes
-	// on `txChan` when the node is shutting down.
-	done chan struct{}
 }
 
 var _ TxPool = &BatchTxPool{}
@@ -88,19 +73,16 @@ func NewBatchTxPool(
 	eoaActivity := expirable.NewLRU[gethCommon.Address, time.Time](
 		eoaActivityCacheSize,
 		nil,
-		config.TxBatchInterval,
+		config.EOAActivityCacheTTL,
 	)
 	batchPool := &BatchTxPool{
 		SingleTxPool: singleTxPool,
 		pooledTxs:    make(map[gethCommon.Address][]pooledEvmTx),
-		txChan:       make(chan cadence.String, txChanBufferSize),
 		txMux:        sync.Mutex{},
 		eoaActivity:  eoaActivity,
-		done:         make(chan struct{}),
 	}
 
 	go batchPool.processPooledTransactions(ctx)
-	go batchPool.processIndividualTransactions(ctx)
 
 	return batchPool
 }
@@ -152,21 +134,22 @@ func (t *BatchTxPool) Add(
 	// transactions that might come from the same EOA.
 	// [X] is equal to the configured `TxBatchInterval` duration.
 	lastActivityTime, found := t.eoaActivity.Get(from)
-	if !found || time.Since(lastActivityTime) > t.config.TxBatchInterval {
-		select {
-		case <-t.done:
-			return fmt.Errorf("the server is shutting down")
-		default:
-			t.txChan <- hexEncodedTx
-		}
+
+	if !found {
+		// Case 1. EOA activity not found:
+		err = t.submitSingleTransaction(ctx, hexEncodedTx)
+	} else if time.Since(lastActivityTime) > t.config.TxBatchInterval {
+		// Case 2. EOA activity found AND it was more than [X] seconds ago:
+		err = t.submitSingleTransaction(ctx, hexEncodedTx)
 	} else {
+		// Case 3. EOA activity found AND it was less than [X] seconds ago:
 		userTx := pooledEvmTx{txPayload: hexEncodedTx, nonce: tx.Nonce()}
 		t.pooledTxs[from] = append(t.pooledTxs[from], userTx)
 	}
 
 	t.eoaActivity.Add(from, time.Now())
 
-	return nil
+	return err
 }
 
 func (t *BatchTxPool) processPooledTransactions(ctx context.Context) {
@@ -213,23 +196,6 @@ func (t *BatchTxPool) processPooledTransactions(ctx context.Context) {
 	}
 }
 
-func (t *BatchTxPool) processIndividualTransactions(ctx context.Context) {
-	for {
-		select {
-		case <-ctx.Done():
-			close(t.done)
-			return
-		case hexEncodedTx := <-t.txChan:
-			if err := t.submitSingleTransaction(ctx, hexEncodedTx); err != nil {
-				t.logger.Error().Err(err).Msg(
-					"failed to submit Flow transaction from BatchTxPool for single EOA tx",
-				)
-				continue
-			}
-		}
-	}
-}
-
 func (t *BatchTxPool) batchSubmitTransactionsForSameAddress(
 	ctx context.Context,
 	latestBlock *flow.Block,