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

eth/catalyst/api.go

@@ -184,7 +184,7 @@ func (api *ConsensusAPI) ForkchoiceUpdatedV1(update engine.ForkchoiceStateV1, pa
 			return engine.STATUS_INVALID, engine.InvalidParams.With(errors.New("forkChoiceUpdateV1 called post-shanghai"))
 		}
 	}
-	return api.forkchoiceUpdated(update, payloadAttributes, engine.PayloadV1, false)
+	return api.forkchoiceUpdated(update, payloadAttributes, engine.PayloadV1)
 }
 
 // ForkchoiceUpdatedV2 is equivalent to V1 with the addition of withdrawals in the payload
@@ -207,7 +207,7 @@ func (api *ConsensusAPI) ForkchoiceUpdatedV2(update engine.ForkchoiceStateV1, pa
 			return engine.STATUS_INVALID, engine.UnsupportedFork.With(errors.New("forkchoiceUpdatedV2 must only be called with paris and shanghai payloads"))
 		}
 	}
-	return api.forkchoiceUpdated(update, params, engine.PayloadV2, false)
+	return api.forkchoiceUpdated(update, params, engine.PayloadV2)
 }
 
 // ForkchoiceUpdatedV3 is equivalent to V2 with the addition of parent beacon block root
@@ -228,10 +228,10 @@ func (api *ConsensusAPI) ForkchoiceUpdatedV3(update engine.ForkchoiceStateV1, pa
 	// hash, even if params are wrong. To do this we need to split up
 	// forkchoiceUpdate into a function that only updates the head and then a
 	// function that kicks off block construction.
-	return api.forkchoiceUpdated(update, params, engine.PayloadV3, false)
+	return api.forkchoiceUpdated(update, params, engine.PayloadV3)
 }
 
-func (api *ConsensusAPI) forkchoiceUpdated(update engine.ForkchoiceStateV1, payloadAttributes *engine.PayloadAttributes, payloadVersion engine.PayloadVersion, simulatorMode bool) (engine.ForkChoiceResponse, error) {
+func (api *ConsensusAPI) forkchoiceUpdated(update engine.ForkchoiceStateV1, payloadAttributes *engine.PayloadAttributes, payloadVersion engine.PayloadVersion) (engine.ForkChoiceResponse, error) {
 	api.forkchoiceLock.Lock()
 	defer api.forkchoiceLock.Unlock()
 
@@ -374,19 +374,7 @@ func (api *ConsensusAPI) forkchoiceUpdated(update engine.ForkchoiceStateV1, payl
 		if api.localBlocks.has(id) {
 			return valid(&id), nil
 		}
-		// If the beacon chain is ran by a simulator, then transaction insertion,
-		// block insertion and block production will happen without any timing
-		// delay between them. This will cause flaky simulator executions due to
-		// the transaction pool running its internal reset operation on a back-
-		// ground thread. To avoid the racey behavior - in simulator mode - the
-		// pool will be explicitly blocked on its reset before continuing to the
-		// block production below.
-		if simulatorMode {
-			if err := api.eth.TxPool().Sync(); err != nil {
-				log.Error("Failed to sync transaction pool", "err", err)
-				return valid(nil), engine.InvalidPayloadAttributes.With(err)
-			}
-		}
+
 		payload, err := api.eth.Miner().BuildPayload(args)
 		if err != nil {
 			log.Error("Failed to build payload", "err", err)

eth/catalyst/simulated_beacon.go

@@ -164,7 +164,7 @@ func (c *SimulatedBeacon) sealBlock(withdrawals []*types.Withdrawal, timestamp u
 		Withdrawals:           withdrawals,
 		Random:                random,
 		BeaconRoot:            &common.Hash{},
-	}, engine.PayloadV3, true)
+	}, engine.PayloadV3)
 	if err != nil {
 		return err
 	}

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,36 @@ 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()
+				// When the beacon chain is ran by a simulator, then transaction insertion,
+				// block insertion and block production will happen without any timing
+				// delay between them. This will cause flaky simulator executions due to
+				// the transaction pool running its internal reset operation on a back-
+				// ground thread. To avoid the racey behavior - in simulator mode - the
+				// pool will be explicitly blocked on its reset before continuing to the
+				// block production below.
+				if err := a.sim.eth.TxPool().Sync(); err != nil {
+					log.Error("Failed to sync transaction pool", "err", err)
+					return
+				}
+				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()
+
+				if err := a.sim.eth.TxPool().Sync(); err != nil {
+					log.Error("Failed to sync transaction pool", "err", err)
+					return
+				}
+				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,78 @@ 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)
+	defer timer.Stop()
+	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")
+		}
+	}
+}