Reject transactions that have already been submitted to the tx pool

Author: m-Peter

services/requester/batch_tx_pool.go

@@ -3,6 +3,8 @@ package requester
 import (
 	"context"
 	"encoding/hex"
+	"fmt"
+	"slices"
 	"sort"
 	"sync"
 	"time"
@@ -17,6 +19,7 @@ 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"
 )
 
@@ -30,6 +33,11 @@ type pooledEvmTx struct {
 	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.
@@ -47,7 +55,7 @@ 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{}
@@ -73,7 +81,7 @@ func NewBatchTxPool(
 		keystore,
 	)
 
-	eoaActivity := expirable.NewLRU[gethCommon.Address, time.Time](
+	eoaActivity := expirable.NewLRU[gethCommon.Address, eoaActivityMetadata](
 		eoaActivityCacheSize,
 		nil,
 		eoaActivityCacheTTL,
@@ -109,7 +117,6 @@ func (t *BatchTxPool) Add(
 	if err != nil {
 		return err
 	}
-
 	txData, err := tx.MarshalBinary()
 	if err != nil {
 		return err
@@ -136,8 +143,21 @@ 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 || time.Since(lastActivityTime) > t.config.TxBatchInterval {
+	lastActivity, found := t.eoaActivity.Get(from)
+	txHash := tx.Hash()
+
+	// Reject transactions that have already been submitted,
+	// as they are *bound* to fail. Two transactions with
+	// identical hashes, are expected to have the exact same
+	// payload. If the first one was invalid, with a nonce
+	// mismatch, the second one is guaranteed to be invalid
+	// as well. No reason in wasting signing keys in this
+	// case.
+	if found && slices.Contains(lastActivity.txHashes, txHash) {
+		return fmt.Errorf("%w: a tx with hash %s has already been submitted", errs.ErrInvalid, txHash)
+	}
+
+	if !found || time.Since(lastActivity.submittedAt) > t.config.TxBatchInterval {
 		if err := t.submitSingleTransaction(ctx, hexEncodedTx); err != nil {
 			return err
 		}
@@ -146,7 +166,16 @@ func (t *BatchTxPool) Add(
 		t.pooledTxs[from] = append(t.pooledTxs[from], userTx)
 	}
 
-	t.eoaActivity.Add(from, time.Now())
+	// Update metadata for the last EOA activity
+	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) > 10 {
+		lastActivity.txHashes = lastActivity.txHashes[1:]
+	}
+
+	t.eoaActivity.Add(from, lastActivity)
 
 	return nil
 }

tests/tx_batching_test.go

@@ -513,6 +513,59 @@ func Test_MultipleTransactionSubmissionsWithinNonRecentInterval(t *testing.T) {
 	)
 }
 
+func Test_TransactionSubmissionWithPreviouslySubmittedTransactions(t *testing.T) {
+	_, cfg, stop := setupGatewayNode(t)
+	defer stop()
+
+	rpcTester := &rpcTest{
+		url: fmt.Sprintf("%s:%d", cfg.RPCHost, cfg.RPCPort),
+	}
+
+	eoaKey, err := crypto.HexToECDSA(eoaTestPrivateKey)
+	require.NoError(t, err)
+
+	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
+	for _, nonce := range nonces {
+		signed, _, err := evmSign(big.NewInt(1_000_000_000), 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.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())
+			if err != nil || rcp == nil || rcp.Status != 1 {
+				return false
+			}
+		}
+
+		return true
+	}, time.Second*15, time.Second*1, "all transactions were not executed")
+}
+
 func setupGatewayNode(t *testing.T) (emulator.Emulator, config.Config, func()) {
 	srv, err := startEmulator(true)
 	require.NoError(t, err)