Track the latest finalized block header to avoid fetching it on every tx submission

Author: m-Peter

bootstrap/bootstrap.go

@@ -247,8 +247,9 @@ func (b *Bootstrap) StartAPIServer(ctx context.Context) error {
 
 	// create transaction pool
 	var txPool requester.TxPool
+	var err error
 	if b.config.TxBatchMode {
-		txPool = requester.NewBatchTxPool(
+		txPool, err = requester.NewBatchTxPool(
 			ctx,
 			b.client,
 			b.publishers.Transaction,
@@ -258,7 +259,8 @@ func (b *Bootstrap) StartAPIServer(ctx context.Context) error {
 			b.keystore,
 		)
 	} else {
-		txPool = requester.NewSingleTxPool(
+		txPool, err = requester.NewSingleTxPool(
+			ctx,
 			b.client,
 			b.publishers.Transaction,
 			b.logger,
@@ -267,6 +269,9 @@ func (b *Bootstrap) StartAPIServer(ctx context.Context) error {
 			b.keystore,
 		)
 	}
+	if err != nil {
+		return fmt.Errorf("failed to create transaction pool: %w", err)
+	}
 
 	evm, err := requester.NewEVM(
 		b.storages.Registers,

services/requester/batch_tx_pool.go

@@ -57,18 +57,22 @@ func NewBatchTxPool(
 	config config.Config,
 	collector metrics.Collector,
 	keystore *keystore.KeyStore,
-) *BatchTxPool {
+) (*BatchTxPool, error) {
 	// initialize the available keys metric since it is only updated when sending a tx
 	collector.AvailableSigningKeys(keystore.AvailableKeys())
 
-	singleTxPool := NewSingleTxPool(
+	singleTxPool, err := NewSingleTxPool(
+		ctx,
 		client,
 		transactionsPublisher,
 		logger,
 		config,
 		collector,
 		keystore,
 	)
+	if err != nil {
+		return nil, err
+	}
 
 	eoaActivity := expirable.NewLRU[gethCommon.Address, time.Time](
 		eoaActivityCacheSize,
@@ -84,7 +88,7 @@ func NewBatchTxPool(
 
 	go batchPool.processPooledTransactions(ctx)
 
-	return batchPool
+	return batchPool, nil
 }
 
 // Add adds the EVM transaction to the tx pool, grouped with the rest of the
@@ -161,10 +165,10 @@ func (t *BatchTxPool) processPooledTransactions(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			latestBlock, account, err := t.fetchFlowLatestBlockAndCOA(ctx)
+			account, err := t.client.GetAccount(ctx, t.config.COAAddress)
 			if err != nil {
 				t.logger.Error().Err(err).Msg(
-					"failed to get COA / latest Flow block on batch tx submission",
+					"failed to get COA on batch tx submission",
 				)
 				continue
 			}
@@ -180,7 +184,7 @@ func (t *BatchTxPool) processPooledTransactions(ctx context.Context) {
 			for address, pooledTxs := range txsGroupedByAddress {
 				err := t.batchSubmitTransactionsForSameAddress(
 					ctx,
-					latestBlock,
+					t.latestBlockHeader,
 					account,
 					pooledTxs,
 				)
@@ -198,7 +202,7 @@ func (t *BatchTxPool) processPooledTransactions(ctx context.Context) {
 
 func (t *BatchTxPool) batchSubmitTransactionsForSameAddress(
 	ctx context.Context,
-	latestBlock *flow.Block,
+	latestBlockHeader *flow.BlockHeader,
 	account *flow.Account,
 	pooledTxs []pooledEvmTx,
 ) error {
@@ -220,7 +224,7 @@ func (t *BatchTxPool) batchSubmitTransactionsForSameAddress(
 
 	script := replaceAddresses(batchRunTxScript, t.config.FlowNetworkID)
 	flowTx, err := t.buildTransaction(
-		latestBlock,
+		latestBlockHeader,
 		account,
 		script,
 		cadence.NewArray(hexEncodedTxs),
@@ -244,7 +248,7 @@ func (t *BatchTxPool) submitSingleTransaction(
 	ctx context.Context,
 	hexEncodedTx cadence.String,
 ) error {
-	latestBlock, account, err := t.fetchFlowLatestBlockAndCOA(ctx)
+	account, err := t.client.GetAccount(ctx, t.config.COAAddress)
 	if err != nil {
 		return err
 	}
@@ -256,7 +260,7 @@ func (t *BatchTxPool) submitSingleTransaction(
 
 	script := replaceAddresses(runTxScript, t.config.FlowNetworkID)
 	flowTx, err := t.buildTransaction(
-		latestBlock,
+		t.latestBlockHeader,
 		account,
 		script,
 		hexEncodedTx,

services/requester/single_tx_pool.go

@@ -13,50 +13,63 @@ import (
 	flowGo "github.com/onflow/flow-go/model/flow"
 	"github.com/rs/zerolog"
 	"github.com/sethvargo/go-retry"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/onflow/flow-evm-gateway/config"
 	"github.com/onflow/flow-evm-gateway/metrics"
 	"github.com/onflow/flow-evm-gateway/models"
 	"github.com/onflow/flow-evm-gateway/services/requester/keystore"
 )
 
+var blockHeaderUpdateFrequency = time.Second * 5
+
 // SingleTxPool is a simple implementation of the `TxPool` interface that submits
 // transactions as soon as they arrive, without any delays or batching strategies.
 type SingleTxPool struct {
-	logger      zerolog.Logger
-	client      *CrossSporkClient
-	pool        *sync.Map
-	txPublisher *models.Publisher[*gethTypes.Transaction]
-	config      config.Config
-	mux         sync.Mutex
-	keystore    *keystore.KeyStore
-	collector   metrics.Collector
+	logger            zerolog.Logger
+	client            *CrossSporkClient
+	pool              *sync.Map
+	txPublisher       *models.Publisher[*gethTypes.Transaction]
+	config            config.Config
+	mux               sync.Mutex
+	keystore          *keystore.KeyStore
+	collector         metrics.Collector
+	latestBlockHeader *flow.BlockHeader
 	// todo add methods to inspect transaction pool state
 }
 
 var _ TxPool = &SingleTxPool{}
 
 func NewSingleTxPool(
+	ctx context.Context,
 	client *CrossSporkClient,
 	transactionsPublisher *models.Publisher[*gethTypes.Transaction],
 	logger zerolog.Logger,
 	config config.Config,
 	collector metrics.Collector,
 	keystore *keystore.KeyStore,
-) *SingleTxPool {
+) (*SingleTxPool, error) {
+	latestBlockHeader, err := client.GetLatestBlockHeader(ctx, false)
+	if err != nil {
+		return nil, err
+	}
+
 	// initialize the available keys metric since it is only updated when sending a tx
 	collector.AvailableSigningKeys(keystore.AvailableKeys())
 
-	return &SingleTxPool{
-		logger:      logger.With().Str("component", "tx-pool").Logger(),
-		client:      client,
-		txPublisher: transactionsPublisher,
-		pool:        &sync.Map{},
-		config:      config,
-		collector:   collector,
-		keystore:    keystore,
+	singleTxPool := &SingleTxPool{
+		logger:            logger.With().Str("component", "tx-pool").Logger(),
+		client:            client,
+		txPublisher:       transactionsPublisher,
+		pool:              &sync.Map{},
+		config:            config,
+		collector:         collector,
+		keystore:          keystore,
+		latestBlockHeader: latestBlockHeader,
 	}
+
+	go singleTxPool.trackLatestBlockHeader(ctx)
+
+	return singleTxPool, nil
 }
 
 // Add creates a Cadence transaction that wraps the given EVM transaction in
@@ -93,14 +106,14 @@ func (t *SingleTxPool) Add(
 		return err
 	}
 
-	latestBlock, account, err := t.fetchFlowLatestBlockAndCOA(ctx)
+	account, err := t.client.GetAccount(ctx, t.config.COAAddress)
 	if err != nil {
 		return err
 	}
 
 	script := replaceAddresses(runTxScript, t.config.FlowNetworkID)
 	flowTx, err := t.buildTransaction(
-		latestBlock,
+		t.latestBlockHeader,
 		account,
 		script,
 		hexEncodedTx,
@@ -157,7 +170,7 @@ func (t *SingleTxPool) Add(
 // buildTransaction creates a Cadence transaction from the provided script,
 // with the given arguments and signs it with the configured COA account.
 func (t *SingleTxPool) buildTransaction(
-	latestBlock *flow.Block,
+	latestBlockHeader *flow.BlockHeader,
 	account *flow.Account,
 	script []byte,
 	args ...cadence.Value,
@@ -168,7 +181,7 @@ func (t *SingleTxPool) buildTransaction(
 
 	flowTx := flow.NewTransaction().
 		SetScript(script).
-		SetReferenceBlockID(latestBlock.ID).
+		SetReferenceBlockID(latestBlockHeader.ID).
 		SetComputeLimit(flowGo.DefaultMaxTransactionGasLimit)
 
 	for _, arg := range args {
@@ -193,37 +206,30 @@ func (t *SingleTxPool) buildTransaction(
 	}
 
 	// now that the transaction is prepared, store the transaction's metadata
-	accKey.SetLockMetadata(flowTx.ID(), latestBlock.Height)
+	accKey.SetLockMetadata(flowTx.ID(), latestBlockHeader.Height)
 
 	t.collector.OperatorBalance(account)
 
 	return flowTx, nil
 }
 
-func (t *SingleTxPool) fetchFlowLatestBlockAndCOA(ctx context.Context) (
-	*flow.Block,
-	*flow.Account,
-	error,
-) {
-	var (
-		g           = errgroup.Group{}
-		err1, err2  error
-		latestBlock *flow.Block
-		account     *flow.Account
-	)
+func (t *SingleTxPool) trackLatestBlockHeader(ctx context.Context) {
+	ticker := time.NewTicker(blockHeaderUpdateFrequency)
+	defer ticker.Stop()
 
-	// execute concurrently so we can speed up all the information we need for tx
-	g.Go(func() error {
-		latestBlock, err1 = t.client.GetLatestBlock(ctx, true)
-		return err1
-	})
-	g.Go(func() error {
-		account, err2 = t.client.GetAccount(ctx, t.config.COAAddress)
-		return err2
-	})
-	if err := g.Wait(); err != nil {
-		return nil, nil, err
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			blockHeader, err := t.client.GetLatestBlockHeader(ctx, false)
+			if err != nil {
+				t.logger.Error().Err(err).Msg(
+					"failed to update latest Flow block header",
+				)
+				continue
+			}
+			t.latestBlockHeader = blockHeader
+		}
 	}
-
-	return latestBlock, account, nil
 }

tests/helpers.go

@@ -90,7 +90,7 @@ func startEmulator(createTestAccounts bool) (*server.EmulatorServer, error) {
 		GenesisTokenSupply:     genesisToken,
 		WithContracts:          true,
 		Host:                   "localhost",
-		TransactionExpiry:      10,
+		TransactionExpiry:      flow.DefaultTransactionExpiry,
 		TransactionMaxGasLimit: flow.DefaultMaxTransactionGasLimit,
 		SetupEVMEnabled:        true,
 		SetupVMBridgeEnabled:   false,