eth/catalyst: move block commit into its own go-routine to avoid deadlock

eth/catalyst/simulated_beacon_api.go

@@ -18,6 +18,7 @@ package catalyst
 
 import (
 	"context"
+	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -32,8 +33,9 @@ type api struct {
 
 func (a *api) loop() {
 	var (
-		newTxs = make(chan core.NewTxsEvent)
-		sub    = a.sim.eth.TxPool().SubscribeTransactions(newTxs, true)
+		newTxs   = make(chan core.NewTxsEvent)
+		sub      = a.sim.eth.TxPool().SubscribeTransactions(newTxs, true)
+		commitMu = sync.Mutex{}
 	)
 	defer sub.Unsubscribe()
 
@@ -42,12 +44,22 @@ func (a *api) loop() {
 		case <-a.sim.shutdownCh:
 			return
 		case w := <-a.sim.withdrawals.pending:
-			withdrawals := append(a.sim.withdrawals.gatherPending(9), w)
-			if err := a.sim.sealBlock(withdrawals, uint64(time.Now().Unix())); err != nil {
-				log.Warn("Error performing sealing work", "err", err)
-			}
+			go func() {
+				commitMu.Lock()
+				defer commitMu.Unlock()
+
+				withdrawals := append(a.sim.withdrawals.gatherPending(9), w)
+				if err := a.sim.sealBlock(withdrawals, uint64(time.Now().Unix())); err != nil {
+					log.Warn("Error performing sealing work", "err", err)
+				}
+			}()
 		case <-newTxs:
-			a.sim.Commit()
+			go func() {
+				commitMu.Lock()
+				defer commitMu.Unlock()
+
+				a.sim.Commit()
+			}()
 		}
 	}
 }

eth/catalyst/simulated_beacon_test.go

@@ -35,7 +35,7 @@ import (
 	"github.com/ethereum/go-ethereum/params"
 )
 
-func startSimulatedBeaconEthService(t *testing.T, genesis *core.Genesis) (*node.Node, *eth.Ethereum, *SimulatedBeacon) {
+func startSimulatedBeaconEthService(t *testing.T, genesis *core.Genesis, period uint64) (*node.Node, *eth.Ethereum, *SimulatedBeacon) {
 	t.Helper()
 
 	n, err := node.New(&node.Config{
@@ -55,7 +55,7 @@ func startSimulatedBeaconEthService(t *testing.T, genesis *core.Genesis) (*node.
 		t.Fatal("can't create eth service:", err)
 	}
 
-	simBeacon, err := NewSimulatedBeacon(1, ethservice)
+	simBeacon, err := NewSimulatedBeacon(period, ethservice)
 	if err != nil {
 		t.Fatal("can't create simulated beacon:", err)
 	}
@@ -87,7 +87,7 @@ func TestSimulatedBeaconSendWithdrawals(t *testing.T) {
 	// short period (1 second) for testing purposes
 	var gasLimit uint64 = 10_000_000
 	genesis := core.DeveloperGenesisBlock(gasLimit, &testAddr)
-	node, ethService, mock := startSimulatedBeaconEthService(t, genesis)
+	node, ethService, mock := startSimulatedBeaconEthService(t, genesis, 1)
 	_ = mock
 	defer node.Close()
 
@@ -140,3 +140,77 @@ func TestSimulatedBeaconSendWithdrawals(t *testing.T) {
 		}
 	}
 }
+
+// Tests that zero-period dev mode can handle a lot of simultaneous
+// transactions/withdrawals
+func TestOnDemandSpam(t *testing.T) {
+	var withdrawals []types.Withdrawal
+	txs := make(map[common.Hash]*types.Transaction)
+
+	var (
+		// testKey is a private key to use for funding a tester account.
+		testKey, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
+
+		// testAddr is the Ethereum address of the tester account.
+		testAddr = crypto.PubkeyToAddress(testKey.PublicKey)
+	)
+
+	// short period (1 second) for testing purposes
+	var gasLimit uint64 = 10_000_000
+	genesis := core.DeveloperGenesisBlock(gasLimit, &testAddr)
+	node, ethService, mock := startSimulatedBeaconEthService(t, genesis, 0)
+	_ = mock
+	defer node.Close()
+
+	api := &api{mock}
+	go api.loop()
+
+	chainHeadCh := make(chan core.ChainHeadEvent, 10)
+	subscription := ethService.BlockChain().SubscribeChainHeadEvent(chainHeadCh)
+	defer subscription.Unsubscribe()
+
+	// generate some withdrawals
+	for i := 0; i < 0; i++ {
+		withdrawals = append(withdrawals, types.Withdrawal{Index: uint64(i)})
+		if err := mock.withdrawals.add(&withdrawals[i]); err != nil {
+			t.Fatal("addWithdrawal failed", err)
+		}
+	}
+
+	// generate a bunch of transactions
+	signer := types.NewEIP155Signer(ethService.BlockChain().Config().ChainID)
+	for i := 0; i < 20000; i++ {
+		tx, err := types.SignTx(types.NewTransaction(uint64(i), common.Address{}, big.NewInt(1000), params.TxGas, big.NewInt(params.InitialBaseFee), nil), signer, testKey)
+		if err != nil {
+			t.Fatalf("error signing transaction, err=%v", err)
+		}
+		txs[tx.Hash()] = tx
+
+		if err := ethService.APIBackend.SendTx(context.Background(), tx); err != nil {
+			t.Fatal("SendTx failed", err)
+		}
+	}
+
+	includedTxs := make(map[common.Hash]struct{})
+	var includedWithdrawals []uint64
+
+	timer := time.NewTimer(20 * time.Second)
+	for {
+		select {
+		case evt := <-chainHeadCh:
+			for _, includedTx := range evt.Block.Transactions() {
+				includedTxs[includedTx.Hash()] = struct{}{}
+			}
+			for _, includedWithdrawal := range evt.Block.Withdrawals() {
+				includedWithdrawals = append(includedWithdrawals, includedWithdrawal.Index)
+			}
+
+			// ensure all withdrawals/txs included. this will take two blocks b/c number of withdrawals > 10
+			if len(includedTxs) == len(txs) && len(includedWithdrawals) == len(withdrawals) {
+				return
+			}
+		case <-timer.C:
+			t.Fatal("timed out without including all withdrawals/txs")
+		}
+	}
+}