Deduplicate transactions on BatchTxPool prior to submission

Author: m-Peter

services/requester/batch_tx_pool.go

@@ -3,6 +3,7 @@ package requester
 import (
 	"context"
 	"encoding/hex"
+	"slices"
 	"sort"
 	"sync"
 	"time"
@@ -24,6 +25,7 @@ const eoaActivityCacheSize = 10_000
 
 type pooledEvmTx struct {
 	txPayload cadence.String
+	txHash    gethCommon.Hash
 	nonce     uint64
 }
 
@@ -143,7 +145,7 @@ 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()}
 		t.pooledTxs[from] = append(t.pooledTxs[from], userTx)
 	}
 
@@ -207,10 +209,22 @@ func (t *BatchTxPool) batchSubmitTransactionsForSameAddress(
 	sort.Slice(pooledTxs, func(i, j int) bool {
 		return pooledTxs[i].nonce < pooledTxs[j].nonce
 	})
+	// Filter out duplicate transactions, based on their tx hash
+	slices.CompactFunc(pooledTxs, func(a pooledEvmTx, b pooledEvmTx) bool {
+		if a.txHash == b.txHash {
+			return true
+		}
+		return false
+	})
 
-	hexEncodedTxs := make([]cadence.Value, len(pooledTxs))
-	for i, txPayload := range pooledTxs {
-		hexEncodedTxs[i] = txPayload.txPayload
+	hexEncodedTxs := make([]cadence.Value, 0)
+	for _, txPayload := range pooledTxs {
+		// The `slices.CompactFunc` still reserves the duplicate entries
+		// as zero values in the array, so we need to avoid including those.
+		if txPayload.txHash == (gethCommon.Hash{}) {
+			continue
+		}
+		hexEncodedTxs = append(hexEncodedTxs, txPayload.txPayload)
 	}
 
 	coinbaseAddress, err := cadence.NewString(t.config.Coinbase.Hex())

tests/tx_batching_test.go

@@ -514,6 +514,53 @@ 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)
+	nonce += 1
+
+	txHash, err := rpcTester.sendRawTx(signed)
+	require.NoError(t, err)
+	hashes = append(hashes, txHash)
+
+	for range 5 {
+		// All these transactions are duplicates, since we don't change any
+		// of the payload data. These will end up having the same tx hash
+		// as well.
+		signed, _, err := evmSign(big.NewInt(10), 15_000_000, eoaKey, nonce, &testAddr, nil)
+		require.NoError(t, err)
+
+		txHash, err := rpcTester.sendRawTx(signed)
+		require.NoError(t, err)
+		hashes = append(hashes, txHash)
+	}
+
+	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)