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

Author: m-Peter

services/requester/batch_tx_pool.go

@@ -124,6 +124,9 @@ func (t *BatchTxPool) Add(
 		return err
 	}
 
+	t.txMux.Lock()
+	defer t.txMux.Unlock()
+
 	eoaActivity, found := t.eoaActivityCache.Get(from)
 	nonce := tx.Nonce()
 
@@ -162,13 +165,11 @@ func (t *BatchTxPool) Add(
 		// If the EOA has pooled transactions, which are not yet processed,
 		// due to congestion or anything, make sure to include the current
 		// tx on that batch.
-		t.txMux.Lock()
 		hasBatch := len(t.pooledTxs[from]) > 0
 		if hasBatch {
 			userTx := pooledEvmTx{txPayload: hexEncodedTx, nonce: nonce}
 			t.pooledTxs[from] = append(t.pooledTxs[from], userTx)
 		}
-		t.txMux.Unlock()
 
 		// If it wasn't batched, submit individually
 		if !hasBatch {
@@ -177,10 +178,8 @@ func (t *BatchTxPool) Add(
 		}
 	} else {
 		// Case 3. EOA activity found AND it was less than [X] seconds ago:
-		t.txMux.Lock()
 		userTx := pooledEvmTx{txPayload: hexEncodedTx, nonce: nonce}
 		t.pooledTxs[from] = append(t.pooledTxs[from], userTx)
-		t.txMux.Unlock()
 	}
 
 	if err != nil {
@@ -285,43 +284,62 @@ func (t *BatchTxPool) submitSingleTransaction(
 	ctx context.Context,
 	hexEncodedTx cadence.String,
 ) error {
-	coinbaseAddress, err := cadence.NewString(t.config.Coinbase.Hex())
-	if err != nil {
-		return err
-	}
+	done := make(chan struct{})
+	var submitError error
 
-	script := replaceAddresses(runTxScript, t.config.FlowNetworkID)
-	flowTx, err := t.buildTransaction(
-		ctx,
-		t.getReferenceBlock(),
-		script,
-		cadence.NewArray([]cadence.Value{hexEncodedTx}),
-		coinbaseAddress,
-	)
-	if err != nil {
-		// If there was any error during the transaction build
-		// process, we record it as a dropped transaction.
-		t.collector.TransactionsDropped(1)
-		return err
-	}
+	// This method is called while holding the `t.txMux` lock,
+	// don't let it run for a long time, to avoid lock-contention
+	ctx, cancel := context.WithTimeout(ctx, time.Second*3)
+	defer cancel()
 
-	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
+	// build & submit transaction
+	go func() {
+		defer close(done)
+
+		coinbaseAddress, err := cadence.NewString(t.config.Coinbase.Hex())
+		if err != nil {
+			submitError = err
+			return
+		}
+
+		script := replaceAddresses(runTxScript, t.config.FlowNetworkID)
+		flowTx, err := t.buildTransaction(
+			ctx,
+			t.getReferenceBlock(),
+			script,
+			cadence.NewArray([]cadence.Value{hexEncodedTx}),
+			coinbaseAddress,
+		)
+		if err != nil {
+			// If there was any error during the transaction build
+			// process, we record it as a dropped transaction.
+			t.collector.TransactionsDropped(1)
+			submitError = err
+			return
+		}
+
+		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)
+			submitError = err
+			return
+		}
+	}()
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-done:
 	}
 
-	return nil
+	return submitError
 }
 
 func (t *BatchTxPool) updateEOAActivityMetadata(
 	from gethCommon.Address,
 	nonce uint64,
 ) {
-	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()

tests/tx_batching_test.go

@@ -515,7 +515,7 @@ func Test_MultipleTransactionSubmissionsWithinNonRecentInterval(t *testing.T) {
 }
 
 func Test_TransactionSubmissionWithPreviouslySubmittedTransactions(t *testing.T) {
-	_, cfg, stop := setupGatewayNode(t)
+	emu, cfg, stop := setupGatewayNode(t)
 	defer stop()
 
 	rpcTester := &rpcTest{
@@ -528,16 +528,28 @@ func Test_TransactionSubmissionWithPreviouslySubmittedTransactions(t *testing.T)
 	testAddr := common.HexToAddress("0x061B63D29332e4de81bD9F51A48609824CD113a8")
 	nonces := []uint64{0, 1, 2, 3, 2, 3, 4, 5}
 
+	g := errgroup.Group{}
+
+	startBlock, err := emu.GetLatestBlock()
+	require.NoError(t, err)
+
 	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(transferAmount), 23_500, eoaKey, nonce, &testAddr, nil)
-		require.NoError(t, err)
 
-		txHash, err := rpcTester.sendRawTx(signed)
-		require.NoError(t, err)
-		hashes = append(hashes, txHash)
+	for range 3 {
+		g.Go(func() error {
+			for _, nonce := range nonces {
+				signed, _, err := evmSign(big.NewInt(transferAmount), 23_500, eoaKey, nonce, &testAddr, nil)
+				require.NoError(t, err)
+
+				txHash, err := rpcTester.sendRawTx(signed)
+				require.NoError(t, err)
+				hashes = append(hashes, txHash)
+			}
+
+			return nil
+		})
 	}
 
 	expectedBalance := big.NewInt(6 * transferAmount)
@@ -550,6 +562,34 @@ func Test_TransactionSubmissionWithPreviouslySubmittedTransactions(t *testing.T)
 
 		return balance.Cmp(expectedBalance) == 0
 	}, time.Second*15, time.Second*1, "all transactions were not executed")
+
+	endBlock, err := emu.GetLatestBlock()
+	require.NoError(t, err)
+
+	blockEvents, err := emu.GetEventsForHeightRange(
+		"A.f8d6e0586b0a20c7.EVM.TransactionExecuted",
+		startBlock.Height+1,
+		endBlock.Height,
+	)
+
+	totalEVMEvents := 0
+	for _, blockEvent := range blockEvents {
+		totalEVMEvents += len(blockEvent.Events)
+	}
+	assert.Equal(t, 6, totalEVMEvents)
+
+	for i := startBlock.Height; i <= endBlock.Height; i++ {
+		block, err := emu.GetBlockByHeight(i)
+		require.NoError(t, err)
+
+		txResults, err := emu.GetTransactionResultsByBlockID(block.ID())
+		require.NoError(t, err)
+
+		for _, txResult := range txResults {
+			// Assert that we have no errors on the submitted transactions
+			require.Empty(t, txResult.ErrorMessage)
+		}
+	}
 }
 
 func setupGatewayNode(t *testing.T) (emulator.Emulator, config.Config, func()) {