Reject transactions that have already been submitted to the tx pool

api/utils.go

@@ -29,9 +29,6 @@ func resolveBlockTag(
 	if number, ok := blockNumberOrHash.Number(); ok {
 		height, err := resolveBlockNumber(number, blocksDB)
 		if err != nil {
-			logger.Error().Err(err).
-				Stringer("block_number", number).
-				Msg("failed to resolve block by number")
 			return 0, err
 		}
 		return height, nil
@@ -40,9 +37,6 @@ func resolveBlockTag(
 	if hash, ok := blockNumberOrHash.Hash(); ok {
 		height, err := blocksDB.GetHeightByID(hash)
 		if err != nil {
-			logger.Error().Err(err).
-				Stringer("block_hash", hash).
-				Msg("failed to resolve block by hash")
 			return 0, err
 		}
 		return height, nil

models/errors/errors.go

@@ -28,9 +28,8 @@ var (
 
 	// Transaction errors
 
-	ErrFailedTransaction    = errors.New("failed transaction")
-	ErrInvalidTransaction   = fmt.Errorf("%w: %w", ErrInvalid, ErrFailedTransaction)
-	ErrDuplicateTransaction = fmt.Errorf("%w: %s", ErrInvalid, "transaction already in pool")
+	ErrFailedTransaction  = errors.New("failed transaction")
+	ErrInvalidTransaction = fmt.Errorf("%w: %w", ErrInvalid, ErrFailedTransaction)
 
 	// Storage errors
 

services/requester/batch_tx_pool.go

@@ -18,39 +18,45 @@ import (
 	"github.com/onflow/flow-evm-gateway/config"
 	"github.com/onflow/flow-evm-gateway/metrics"
 	"github.com/onflow/flow-evm-gateway/models"
-	errs "github.com/onflow/flow-evm-gateway/models/errors"
 	"github.com/onflow/flow-evm-gateway/services/requester/keystore"
 )
 
-const eoaActivityCacheSize = 10_000
+const (
+	eoaActivityCacheSize     = 10_000
+	maxTrackedTxNoncesPerEOA = 30
+)
 
 type pooledEvmTx struct {
 	txPayload cadence.String
-	txHash    gethCommon.Hash
 	nonce     uint64
 }
 
-// BatchTxPool is a `TxPool` implementation that collects and groups
-// transactions based on their EOA signer, and submits them for execution
-// using a batch.
+type eoaActivityMetadata struct {
+	lastSubmission time.Time
+	txNonces       []uint64
+}
+
+// BatchTxPool is a `TxPool` implementation that groups incoming transactions
+// based on their EOA signer, and submits them for execution using a batch.
 //
 // The underlying Cadence EVM API used, is `EVM.batchRun`, instead of the
 // `EVM.run` used in `SingleTxPool`.
 //
 // The main advantage of this implementation over the `SingleTxPool`, is the
-// guarantee that transactions originated from the same EOA address, which
-// arrive in a short time interval (about the same as Flow's block production rate),
-// will be executed in the same order their arrived.
-// This helps to reduce the nonce mismatch errors which mainly occur from the
-// re-ordering of Cadence transactions that happens from Collection nodes.
+// guarantee that transactions originating from the same EOA address, which
+// arrive in a short time interval (configurable by the node operator),
+// will be executed in the same order they arrived.
+// This helps to reduce the execution errors which may occur from the
+// re-ordering of Cadence transactions that happens on Collection nodes.
 type BatchTxPool struct {
 	*SingleTxPool
-	pooledTxs   map[gethCommon.Address][]pooledEvmTx
-	txMux       sync.Mutex
-	eoaActivity *expirable.LRU[gethCommon.Address, time.Time]
+
+	pooledTxs        map[gethCommon.Address][]pooledEvmTx
+	txMux            sync.Mutex
+	eoaActivityCache *expirable.LRU[gethCommon.Address, eoaActivityMetadata]
 }
 
-var _ TxPool = &BatchTxPool{}
+var _ TxPool = (*BatchTxPool)(nil)
 
 func NewBatchTxPool(
 	ctx context.Context,
@@ -77,16 +83,16 @@ func NewBatchTxPool(
 		return nil, err
 	}
 
-	eoaActivity := expirable.NewLRU[gethCommon.Address, time.Time](
+	eoaActivityCache := expirable.NewLRU[gethCommon.Address, eoaActivityMetadata](
 		eoaActivityCacheSize,
 		nil,
 		config.EOAActivityCacheTTL,
 	)
 	batchPool := &BatchTxPool{
-		SingleTxPool: singleTxPool,
-		pooledTxs:    make(map[gethCommon.Address][]pooledEvmTx),
-		txMux:        sync.Mutex{},
-		eoaActivity:  eoaActivity,
+		SingleTxPool:     singleTxPool,
+		pooledTxs:        make(map[gethCommon.Address][]pooledEvmTx),
+		txMux:            sync.Mutex{},
+		eoaActivityCache: eoaActivityCache,
 	}
 
 	go batchPool.processPooledTransactions(ctx)
@@ -104,11 +110,6 @@ func (t *BatchTxPool) Add(
 ) error {
 	t.txPublisher.Publish(tx) // publish pending transaction event
 
-	// tx adding should be blocking, so we don't have races when
-	// pooled transactions are being processed in the background.
-	t.txMux.Lock()
-	defer t.txMux.Unlock()
-
 	from, err := models.DeriveTxSender(tx)
 	if err != nil {
 		return err
@@ -123,6 +124,29 @@ func (t *BatchTxPool) Add(
 		return err
 	}
 
+	t.txMux.Lock()
+
+	eoaActivity, found := t.eoaActivityCache.Get(from)
+	nonce := tx.Nonce()
+
+	// Skip transactions that have been already submitted,
+	// as they are *likely* to fail.
+	if found && slices.Contains(eoaActivity.txNonces, nonce) {
+		t.txMux.Unlock()
+		t.logger.Info().
+			Str("evm_tx", tx.Hash().Hex()).
+			Str("from", from.Hex()).
+			Uint64("nonce", nonce).
+			Msg("tx with same nonce has been already submitted")
+
+		return nil
+	}
+
+	t.updateEOAActivityMetadata(from, nonce)
+
+	// Determine action while holding lock
+	var shouldSubmitSingle bool
+
 	// Scenarios
 	// 1. EOA activity not found:
 	// => We send the transaction individually, without adding it
@@ -140,27 +164,50 @@ func (t *BatchTxPool) Add(
 	// For all 3 cases, we record the activity time for the next
 	// transactions that might come from the same EOA.
 	// [X] is equal to the configured `TxBatchInterval` duration.
-	lastActivityTime, found := t.eoaActivity.Get(from)
-
 	if !found {
 		// Case 1. EOA activity not found:
-		err = t.submitSingleTransaction(ctx, hexEncodedTx)
-	} else if time.Since(lastActivityTime) > t.config.TxBatchInterval {
+		shouldSubmitSingle = true
+	} else if time.Since(eoaActivity.lastSubmission) > t.config.TxBatchInterval {
 		// Case 2. EOA activity found AND it was more than [X] seconds ago:
-		err = t.submitSingleTransaction(ctx, hexEncodedTx)
+
+		// If the EOA has pooled transactions, which are not yet processed,
+		// due to congestion or anything, make sure to include the current
+		// tx on that batch.
+		shouldSubmitSingle = (len(t.pooledTxs[from]) == 0)
 	} else {
 		// Case 3. EOA activity found AND it was less than [X] seconds ago:
-		userTx := pooledEvmTx{txPayload: hexEncodedTx, txHash: tx.Hash(), nonce: tx.Nonce()}
-		// Prevent submission of duplicate transactions, based on their tx hash
-		if slices.Contains(t.pooledTxs[from], userTx) {
-			return errs.ErrDuplicateTransaction
-		}
+		shouldSubmitSingle = false
+	}
+
+	// Pool transaction in the batch
+	if !shouldSubmitSingle {
+		userTx := pooledEvmTx{txPayload: hexEncodedTx, nonce: nonce}
 		t.pooledTxs[from] = append(t.pooledTxs[from], userTx)
 	}
 
-	t.eoaActivity.Add(from, time.Now())
+	// Release lock before network I/O operation
+	t.txMux.Unlock()
+
+	// Submit single transaction without holding lock
+	if shouldSubmitSingle {
+		err = t.submitSingleTransaction(ctx, hexEncodedTx)
+	}
+
+	if err != nil {
+		// If there was an error during tx submission, remove the entry
+		// from the cache, to not block future requests with same nonce.
+		// Note: No need to acquire the `t.txMux` lock, `Remove` already
+		// has an internal lock.
+		t.eoaActivityCache.Remove(from)
+
+		t.logger.Error().Err(err).Msgf(
+			"failed to submit single Flow transaction for EOA: %s",
+			from.Hex(),
+		)
+		return err
+	}
 
-	return err
+	return nil
 }
 
 func (t *BatchTxPool) processPooledTransactions(ctx context.Context) {
@@ -188,10 +235,14 @@ func (t *BatchTxPool) processPooledTransactions(ctx context.Context) {
 				)
 				if err != nil {
 					t.logger.Error().Err(err).Msgf(
-						"failed to submit Flow transaction from BatchTxPool for EOA: %s",
+						"failed to submit batch Flow transaction for EOA: %s",
 						address.Hex(),
 					)
-					continue
+					// In case of any error, add the transactions back to the pool,
+					// as a retry mechanism.
+					t.txMux.Lock()
+					t.pooledTxs[address] = append(t.pooledTxs[address], pooledTxs...)
+					t.txMux.Unlock()
 				}
 			}
 		}
@@ -235,6 +286,9 @@ func (t *BatchTxPool) batchSubmitTransactionsForSameAddress(
 	}
 
 	if err := t.client.SendTransaction(ctx, *flowTx); err != nil {
+		// If there was any error while sending the transaction,
+		// we record all transactions as dropped.
+		t.collector.TransactionsDropped(len(hexEncodedTxs))
 		return err
 	}
 
@@ -245,29 +299,77 @@ func (t *BatchTxPool) submitSingleTransaction(
 	ctx context.Context,
 	hexEncodedTx cadence.String,
 ) error {
-	coinbaseAddress, err := cadence.NewString(t.config.Coinbase.Hex())
-	if err != nil {
-		return err
-	}
+	done := make(chan struct{})
+	var submitError error
+
+	// The 5-second timeout provides a 2-second buffer on top of ANs
+	// 3-second timeout for LN requests.
+	ctx, cancel := context.WithTimeout(ctx, time.Second*5)
+	defer cancel()
+
+	// Build & submit the transaction, in a separate goroutine. The AN calls
+	// do not respect the `context.WithTimeout` deadline, and can run for as
+	// long as is necessary for their completion.
+	// `context.WithTimeout` arranges for Done to be closed when the specified
+	// timeout elapses, and at that point we return an error to abort the
+	// transaction submission.
+	go func() {
+		defer close(done)
+
+		coinbaseAddress, err := cadence.NewString(t.config.Coinbase.Hex())
+		if err != nil {
+			submitError = err
+			return
+		}
 
-	script := replaceAddresses(runTxScript, t.config.FlowNetworkID)
-	flowTx, err := t.buildTransaction(
-		ctx,
-		t.getReferenceBlock(),
-		script,
-		cadence.NewArray([]cadence.Value{hexEncodedTx}),
-		coinbaseAddress,
-	)
-	if err != nil {
-		// If there was any error during the transaction build
-		// process, we record it as a dropped transaction.
-		t.collector.TransactionsDropped(1)
-		return err
+		script := replaceAddresses(runTxScript, t.config.FlowNetworkID)
+		flowTx, err := t.buildTransaction(
+			ctx,
+			t.getReferenceBlock(),
+			script,
+			cadence.NewArray([]cadence.Value{hexEncodedTx}),
+			coinbaseAddress,
+		)
+		if err != nil {
+			// If there was any error during the transaction build
+			// process, we record it as a dropped transaction.
+			t.collector.TransactionsDropped(1)
+			submitError = err
+			return
+		}
+
+		if err := t.client.SendTransaction(ctx, *flowTx); err != nil {
+			// If there was any error while sending the transaction,
+			// we record it as a dropped transaction.
+			t.collector.TransactionsDropped(1)
+			submitError = err
+			return
+		}
+	}()
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-done:
 	}
 
-	if err := t.client.SendTransaction(ctx, *flowTx); err != nil {
-		return err
+	return submitError
+}
+
+func (t *BatchTxPool) updateEOAActivityMetadata(
+	from gethCommon.Address,
+	nonce uint64,
+) {
+	// Update metadata for the last EOA activity only on successful add/submit.
+	eoaActivity, _ := t.eoaActivityCache.Get(from)
+	eoaActivity.lastSubmission = time.Now()
+	eoaActivity.txNonces = append(eoaActivity.txNonces, nonce)
+	// To avoid the slice of nonces from growing indefinitely,
+	// keep only the last `maxTrackedTxNoncesPerEOA` nonces.
+	if len(eoaActivity.txNonces) > maxTrackedTxNoncesPerEOA {
+		firstKeep := len(eoaActivity.txNonces) - maxTrackedTxNoncesPerEOA
+		eoaActivity.txNonces = eoaActivity.txNonces[firstKeep:]
 	}
 
-	return nil
+	t.eoaActivityCache.Add(from, eoaActivity)
 }

services/requester/requester.go

@@ -238,7 +238,7 @@ func (e *EVM) SendRawTransaction(ctx context.Context, data []byte) (common.Hash,
 	}
 
 	e.logger.Info().
-		Str("evm-id", tx.Hash().Hex()).
+		Str("evm_tx", tx.Hash().Hex()).
 		Str("to", to).
 		Str("from", from.Hex()).
 		Str("value", tx.Value().String()).