Reject transactions that have already been submitted to the tx pool

Author: m-Peter

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

@@ -3,6 +3,7 @@ package requester
 import (
 	"context"
 	"encoding/hex"
+	"fmt"
 	"slices"
 	"sort"
 	"sync"
@@ -22,14 +23,21 @@ import (
 	"github.com/onflow/flow-evm-gateway/services/requester/keystore"
 )
 
-const eoaActivityCacheSize = 10_000
+const (
+	eoaActivityCacheSize = 10_000
+	eoaActivityCacheTTL  = time.Second * 10
+)
 
 type pooledEvmTx struct {
 	txPayload cadence.String
-	txHash    gethCommon.Hash
 	nonce     uint64
 }
 
+type eoaActivityMetadata struct {
+	submittedAt time.Time
+	txHashes    []gethCommon.Hash
+}
+
 // BatchTxPool is a `TxPool` implementation that collects and groups
 // transactions based on their EOA signer, and submits them for execution
 // using a batch.
@@ -39,18 +47,19 @@ type pooledEvmTx struct {
 //
 // 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
+// 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 from Collection nodes.
 type BatchTxPool struct {
 	*SingleTxPool
+
 	pooledTxs   map[gethCommon.Address][]pooledEvmTx
 	txMux       sync.Mutex
-	eoaActivity *expirable.LRU[gethCommon.Address, time.Time]
+	eoaActivity *expirable.LRU[gethCommon.Address, eoaActivityMetadata]
 }
 
-var _ TxPool = &BatchTxPool{}
+var _ TxPool = (*BatchTxPool)(nil)
 
 func NewBatchTxPool(
 	ctx context.Context,
@@ -77,10 +86,10 @@ func NewBatchTxPool(
 		return nil, err
 	}
 
-	eoaActivity := expirable.NewLRU[gethCommon.Address, time.Time](
+	eoaActivity := expirable.NewLRU[gethCommon.Address, eoaActivityMetadata](
 		eoaActivityCacheSize,
 		nil,
-		config.EOAActivityCacheTTL,
+		eoaActivityCacheTTL,
 	)
 	batchPool := &BatchTxPool{
 		SingleTxPool: singleTxPool,
@@ -123,6 +132,21 @@ func (t *BatchTxPool) Add(
 		return err
 	}
 
+	lastActivity, found := t.eoaActivity.Get(from)
+	txHash := tx.Hash()
+
+	// Reject transactions that have already been submitted,
+	// as they are *likely* to fail. Two transactions with
+	// identical hashes, are expected to have the exact same
+	// payload.
+	if found && slices.Contains(lastActivity.txHashes, txHash) {
+		return fmt.Errorf(
+			"%w: a tx with hash %s has already been submitted",
+			errs.ErrInvalid,
+			txHash,
+		)
+	}
+
 	// Scenarios
 	// 1. EOA activity not found:
 	// => We send the transaction individually, without adding it
@@ -140,25 +164,30 @@ 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 {
+	} else if time.Since(lastActivity.submittedAt) > 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, 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
-		}
+		userTx := pooledEvmTx{txPayload: hexEncodedTx, nonce: tx.Nonce()}
 		t.pooledTxs[from] = append(t.pooledTxs[from], userTx)
 	}
 
-	t.eoaActivity.Add(from, time.Now())
+	// Update metadata for the last EOA activity only on successful add/submit.
+	if err == nil {
+		lastActivity.submittedAt = time.Now()
+		lastActivity.txHashes = append(lastActivity.txHashes, txHash)
+		// To avoid the slice of hashes from growing indefinitely,
+		// maintain only a handful of the last tx hashes.
+		if len(lastActivity.txHashes) > 15 {
+			lastActivity.txHashes = lastActivity.txHashes[1:]
+		}
+
+		t.eoaActivity.Add(from, lastActivity)
+	}
 
 	return err
 }

tests/tx_batching_test.go

@@ -332,7 +332,7 @@ func Test_MultipleTransactionSubmissionsWithinRecentInterval(t *testing.T) {
 	// activity of the EOA was X seconds ago, where:
 	// X = `cfg.TxBatchInterval`.
 	// For the E2E tests the `cfg.TxBatchInterval` is equal
-	// to 2 seconds.
+	// to 2.5 seconds.
 	for i := range uint64(2) {
 		signed, _, err := evmSign(
 			big.NewInt(500_000),
@@ -514,7 +514,7 @@ func Test_MultipleTransactionSubmissionsWithinNonRecentInterval(t *testing.T) {
 	)
 }
 
-func Test_MultipleTransactionSubmissionsWithDuplicates(t *testing.T) {
+func Test_TransactionSubmissionWithPreviouslySubmittedTransactions(t *testing.T) {
 	_, cfg, stop := setupGatewayNode(t)
 	defer stop()
 
@@ -525,36 +525,36 @@ func Test_MultipleTransactionSubmissionsWithDuplicates(t *testing.T) {
 	eoaKey, err := crypto.HexToECDSA(eoaTestPrivateKey)
 	require.NoError(t, err)
 
-	testAddr := common.HexToAddress("55253ed90B70b96C73092D8680915aaF50081194")
-	nonce := uint64(0)
-	hashes := make([]common.Hash, 0)
-
-	signed, _, err := evmSign(big.NewInt(10), 21000, eoaKey, nonce, &testAddr, nil)
-	require.NoError(t, err)
-
-	txHash, err := rpcTester.sendRawTx(signed)
-	require.NoError(t, err)
-	hashes = append(hashes, txHash)
+	testAddr := common.HexToAddress("0x061B63D29332e4de81bD9F51A48609824CD113a8")
+	nonces := []uint64{0, 1, 2, 3, 2, 3, 4, 5}
 
-	// Increment nonce for the duplicate test transactions that follow
-	nonce += 1
-	dupSigned, _, err := evmSign(big.NewInt(10), 15_000_000, eoaKey, nonce, &testAddr, nil)
-	require.NoError(t, err)
+	var errors []error
+	hashes := []common.Hash{}
+	// transfer some funds to the test address
+	for _, nonce := range nonces {
+		signed, _, err := evmSign(big.NewInt(1_000_000_000), 23_500, eoaKey, nonce, &testAddr, nil)
+		require.NoError(t, err)
 
-	// Submit 5 identical transactions to test duplicate detection:
-	// the first should succeed, the rest should be rejected as duplicates.
-	for i := range 5 {
-		if i == 0 {
-			txHash, err := rpcTester.sendRawTx(dupSigned)
-			require.NoError(t, err)
-			hashes = append(hashes, txHash)
+		txHash, err := rpcTester.sendRawTx(signed)
+		if err != nil {
+			errors = append(errors, err)
 		} else {
-			_, err := rpcTester.sendRawTx(dupSigned)
-			require.Error(t, err)
-			require.ErrorContains(t, err, "invalid: transaction already in pool")
+			hashes = append(hashes, txHash)
 		}
 	}
 
+	require.Len(t, errors, 2)
+	assert.ErrorContains(
+		t,
+		errors[0],
+		"a tx with hash 0x2bdf4aa4c3e273a624dddfdbde6614786b6a5329e246c531d3e0e9f92e79e04d has already been submitted",
+	)
+	assert.ErrorContains(
+		t,
+		errors[1],
+		"a tx with hash 0xb72e1f83861a63b5ad4b927295af07fa9546b01aac5dfce046a5fb20f9be9f2f has already been submitted",
+	)
+
 	assert.Eventually(t, func() bool {
 		for _, h := range hashes {
 			rcp, err := rpcTester.getReceipt(h.String())
@@ -606,7 +606,7 @@ func setupGatewayNode(t *testing.T) (emulator.Emulator, config.Config, func()) {
 		LogWriter:         testLogWriter(),
 		TxStateValidation: config.TxSealValidation,
 		TxBatchMode:       true,
-		TxBatchInterval:   time.Second * 2,
+		TxBatchInterval:   time.Millisecond * 2500, // 2.5 seconds, the same as mainnet
 	}
 
 	bootstrapDone := make(chan struct{})