Merge branch 'mpeter/submitted-tx-validations' into mpeter/backport-submitted-tx-validations

Author: m-Peter

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

services/requester/batch_tx_pool.go

@@ -3,7 +3,6 @@ package requester
 import (
 	"context"
 	"encoding/hex"
-	"fmt"
 	"slices"
 	"sort"
 	"sync"
@@ -19,13 +18,12 @@ 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
-	maxTrackedTxNoncesPerEOA = 15
+	maxTrackedTxNoncesPerEOA = 30
 )
 
 type pooledEvmTx struct {
@@ -129,14 +127,15 @@ func (t *BatchTxPool) Add(
 	eoaActivity, found := t.eoaActivityCache.Get(from)
 	nonce := tx.Nonce()
 
-	// Reject transactions that have already been submitted,
+	// Skip transactions that have been already submitted,
 	// as they are *likely* to fail.
 	if found && slices.Contains(eoaActivity.txNonces, nonce) {
-		return fmt.Errorf(
-			"%w: a tx with nonce %d has already been submitted",
-			errs.ErrInvalid,
-			nonce,
-		)
+		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
 	}
 
 	// Scenarios
@@ -185,16 +184,25 @@ func (t *BatchTxPool) Add(
 	}
 
 	if err != nil {
+		t.logger.Error().Err(err).Msgf(
+			"failed to submit single Flow transaction for EOA: %s",
+			from.Hex(),
+		)
 		return err
 	}
 
+	t.txMux.Lock()
+	defer t.txMux.Unlock()
+
 	// 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,
-	// maintain only a handful of the last tx nonces.
+	// keep only the last `maxTrackedTxNoncesPerEOA` nonces.
 	if len(eoaActivity.txNonces) > maxTrackedTxNoncesPerEOA {
-		eoaActivity.txNonces = eoaActivity.txNonces[1:]
+		firstKeep := len(eoaActivity.txNonces) - maxTrackedTxNoncesPerEOA
+		eoaActivity.txNonces = eoaActivity.txNonces[firstKeep:]
 	}
 
 	t.eoaActivityCache.Add(from, eoaActivity)
@@ -227,7 +235,7 @@ 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
@@ -274,6 +282,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
 	}
 
@@ -305,6 +316,9 @@ func (t *BatchTxPool) submitSingleTransaction(
 	}
 
 	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)
 		return err
 	}
 

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()).

services/requester/single_tx_pool.go

@@ -95,6 +95,11 @@ func (t *SingleTxPool) Add(
 ) error {
 	t.txPublisher.Publish(tx) // publish pending transaction event
 
+	from, err := models.DeriveTxSender(tx)
+	if err != nil {
+		return err
+	}
+
 	txData, err := tx.MarshalBinary()
 	if err != nil {
 		return err
@@ -120,10 +125,21 @@ func (t *SingleTxPool) Add(
 		// If there was any error during the transaction build
 		// process, we record it as a dropped transaction.
 		t.collector.TransactionsDropped(1)
+		t.logger.Error().Err(err).Msgf(
+			"failed to build Flow transaction for EOA: %s",
+			from.Hex(),
+		)
 		return err
 	}
 
 	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)
+		t.logger.Error().Err(err).Msgf(
+			"failed to submit Flow transaction for EOA: %s",
+			from.Hex(),
+		)
 		return err
 	}
 
@@ -149,8 +165,8 @@ func (t *SingleTxPool) Add(
 				}
 
 				t.logger.Error().Err(res.Error).
-					Str("flow-id", flowTx.ID().String()).
-					Str("evm-id", tx.Hash().Hex()).
+					Str("flow_tx", flowTx.ID().String()).
+					Str("evm_tx", tx.Hash().Hex()).
 					Msg("flow transaction error")
 
 				// hide specific cause since it's an implementation issue

tests/tx_batching_test.go

@@ -528,42 +528,27 @@ func Test_TransactionSubmissionWithPreviouslySubmittedTransactions(t *testing.T)
 	testAddr := common.HexToAddress("0x061B63D29332e4de81bD9F51A48609824CD113a8")
 	nonces := []uint64{0, 1, 2, 3, 2, 3, 4, 5}
 
-	var errors []error
 	hashes := []common.Hash{}
 	// transfer some funds to the test address
+	transferAmount := int64(1_000_000_000)
 	for _, nonce := range nonces {
-		signed, _, err := evmSign(big.NewInt(1_000_000_000), 23_500, eoaKey, nonce, &testAddr, nil)
+		signed, _, err := evmSign(big.NewInt(transferAmount), 23_500, eoaKey, nonce, &testAddr, nil)
 		require.NoError(t, err)
 
 		txHash, err := rpcTester.sendRawTx(signed)
-		if err != nil {
-			errors = append(errors, err)
-		} else {
-			hashes = append(hashes, txHash)
-		}
+		require.NoError(t, err)
+		hashes = append(hashes, txHash)
 	}
 
-	require.Len(t, errors, 2)
-	assert.ErrorContains(
-		t,
-		errors[0],
-		"a tx with nonce 2 has already been submitted",
-	)
-	assert.ErrorContains(
-		t,
-		errors[1],
-		"a tx with nonce 3 has already been submitted",
-	)
+	expectedBalance := big.NewInt(6 * transferAmount)
 
 	assert.Eventually(t, func() bool {
-		for _, h := range hashes {
-			rcp, err := rpcTester.getReceipt(h.String())
-			if err != nil || rcp == nil || rcp.Status != 1 {
-				return false
-			}
+		balance, err := rpcTester.getBalance(testAddr)
+		if err != nil {
+			return false
 		}
 
-		return true
+		return balance.Cmp(expectedBalance) == 0
 	}, time.Second*15, time.Second*1, "all transactions were not executed")
 }