Merge pull request #916 from onflow/mpeter/deduplicate-batched-txs

Deduplicate transactions on `BatchTxPool` prior to submission

Author: m-Peter

models/errors/errors.go

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

services/requester/batch_tx_pool.go

@@ -3,6 +3,7 @@ package requester
 import (
 	"context"
 	"encoding/hex"
+	"slices"
 	"sort"
 	"sync"
 	"time"
@@ -17,13 +18,15 @@ 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
 
 type pooledEvmTx struct {
 	txPayload cadence.String
+	txHash    gethCommon.Hash
 	nonce     uint64
 }
 
@@ -147,7 +150,11 @@ func (t *BatchTxPool) Add(
 		err = t.submitSingleTransaction(ctx, hexEncodedTx)
 	} else {
 		// Case 3. EOA activity found AND it was less than [X] seconds ago:
-		userTx := pooledEvmTx{txPayload: hexEncodedTx, nonce: tx.Nonce()}
+		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
+		}
 		t.pooledTxs[from] = append(t.pooledTxs[from], userTx)
 	}
 

tests/tx_batching_test.go

@@ -514,6 +514,59 @@ func Test_MultipleTransactionSubmissionsWithinNonRecentInterval(t *testing.T) {
 	)
 }
 
+func Test_MultipleTransactionSubmissionsWithDuplicates(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("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)
+
+	// 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)
+
+	// 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)
+		} else {
+			_, err := rpcTester.sendRawTx(dupSigned)
+			require.Error(t, err)
+			require.ErrorContains(t, err, "invalid: transaction already in pool")
+		}
+	}
+
+	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)