miner: discard interrupted blocks (ethereum#24638)

During mining, when a new head arrives and interrupts the block building, the block being built should not be commited (but discarded). Committing the interrupted block introduces unnecessary delay, and possibly causes miner to mine on the previous head, which could result in higher uncle rate.

eth/catalyst/api.go

@@ -28,7 +28,6 @@ import (
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/core/beacon"
 	"github.com/ethereum/go-ethereum/core/rawdb"
-	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/node"
@@ -356,11 +355,3 @@ func (api *ConsensusAPI) assembleBlock(parentHash common.Hash, params *beacon.Pa
 	}
 	return beacon.BlockToExecutableData(block), nil
 }
-
-// Used in tests to add a the list of transactions from a block to the tx pool.
-func (api *ConsensusAPI) insertTransactions(txs types.Transactions) error {
-	for _, tx := range txs {
-		api.eth.TxPool().AddLocal(tx)
-	}
-	return nil
-}

eth/catalyst/api_test.go

@@ -108,7 +108,7 @@ func TestEth2AssembleBlockWithAnotherBlocksTxs(t *testing.T) {
 	api := NewConsensusAPI(ethservice)
 
 	// Put the 10th block's tx in the pool and produce a new block
-	api.insertTransactions(blocks[9].Transactions())
+	api.eth.TxPool().AddRemotesSync(blocks[9].Transactions())
 	blockParams := beacon.PayloadAttributesV1{
 		Timestamp: blocks[8].Time() + 5,
 	}

miner/worker.go

@@ -78,6 +78,11 @@ const (
 	staleThreshold = 7
 )
 
+var (
+	errBlockInterruptedByNewHead  = errors.New("new head arrived while building block")
+	errBlockInterruptedByRecommit = errors.New("recommit interrupt while building block")
+)
+
 // environment is the worker's current environment and holds all
 // information of the sealing block generation.
 type environment struct {
@@ -913,7 +918,7 @@ func (w *worker) commitTransaction(env *environment, tx *types.Transaction) ([]*
 	return receipt.Logs, nil
 }
 
-func (w *worker) commitTransactions(env *environment, txs *types.TransactionsByPriceAndNonce, interrupt *int32, tstart *time.Time, committedTxs map[common.Hash]*types.Transaction) bool {
+func (w *worker) commitTransactions(env *environment, txs *types.TransactionsByPriceAndNonce, interrupt *int32, tstart *time.Time, committedTxs map[common.Hash]*types.Transaction) error {
 	gasLimit := env.header.GasLimit
 	if env.gasPool == nil {
 		env.gasPool = new(core.GasPool).AddGas(gasLimit)
@@ -938,8 +943,9 @@ func (w *worker) commitTransactions(env *environment, txs *types.TransactionsByP
 					ratio: ratio,
 					inc:   true,
 				}
+				return errBlockInterruptedByRecommit
 			}
-			return atomic.LoadInt32(interrupt) == commitInterruptNewHead
+			return errBlockInterruptedByNewHead
 		}
 		// If we don't have enough gas for any further transactions then we're done
 		if env.gasPool.Gas() < params.TxGas {
@@ -1035,10 +1041,10 @@ func (w *worker) commitTransactions(env *environment, txs *types.TransactionsByP
 	if interrupt != nil {
 		w.resubmitAdjustCh <- &intervalAdjust{inc: false}
 	}
-	return false
+	return nil
 }
 
-func (w *worker) commitTransactionsSimple(env *environment, txs *TxOrderer, interrupt *int32, tstart *time.Time) bool {
+func (w *worker) commitTransactionsSimple(env *environment, txs *TxOrderer, interrupt *int32, tstart *time.Time) error {
 	gasLimit := env.header.GasLimit
 	if env.gasPool == nil {
 		env.gasPool = new(core.GasPool).AddGas(gasLimit)
@@ -1066,8 +1072,9 @@ func (w *worker) commitTransactionsSimple(env *environment, txs *TxOrderer, inte
 					ratio: ratio,
 					inc:   true,
 				}
+				return errBlockInterruptedByRecommit
 			}
-			return atomic.LoadInt32(interrupt) == commitInterruptNewHead
+			return errBlockInterruptedByNewHead
 		}
 		// If we don't have enough gas for any further transactions then we're done
 		if env.gasPool.Gas() < params.TxGas {
@@ -1158,7 +1165,7 @@ func (w *worker) commitTransactionsSimple(env *environment, txs *TxOrderer, inte
 	if interrupt != nil {
 		w.resubmitAdjustCh <- &intervalAdjust{inc: false}
 	}
-	return false
+	return nil
 }
 
 // collects ancestors' block times for possible throttling
@@ -1249,12 +1256,12 @@ func (w *worker) commitTransactionsEx(env *environment, interrupt *int32, tstart
 			}
 
 			txs := types.NewTransactionsByPriceAndNonce(env.signer, pending, env.header.BaseFee)
-			if w.commitTransactions(env, txs, interrupt, &tstart, committedTxs) {
+			if err := w.commitTransactions(env, txs, interrupt, &tstart, committedTxs); err != nil {
 				return true
 			}
 		} else {
 			txs := NewTxOrderer(pending, committedTxs)
-			if w.commitTransactionsSimple(env, txs, interrupt, &tstart) {
+			if err := w.commitTransactionsSimple(env, txs, interrupt, &tstart); err != nil {
 				return true
 			}
 		}
@@ -1392,7 +1399,7 @@ func (w *worker) prepareWork(genParams *generateParams) (*environment, error) {
 // fillTransactions retrieves the pending transactions from the txpool and fills them
 // into the given sealing block. The transaction selection and ordering strategy can
 // be customized with the plugin in the future.
-func (w *worker) fillTransactions(interrupt *int32, env *environment) {
+func (w *worker) fillTransactions(interrupt *int32, env *environment) error {
 	// Split the pending transactions into locals and remotes
 	// Fill the block with all available pending transactions.
 	pending := w.eth.TxPool().Pending(true)
@@ -1405,16 +1412,19 @@ func (w *worker) fillTransactions(interrupt *int32, env *environment) {
 	}
 	if len(localTxs) > 0 {
 		txs := types.NewTransactionsByPriceAndNonce(env.signer, localTxs, env.header.BaseFee)
-		if w.commitTransactions(env, txs, interrupt, nil, nil) {
-			return
+
+		if err := w.commitTransactions(env, txs, interrupt, nil, nil); err != nil {
+			return err
 		}
 	}
 	if len(remoteTxs) > 0 {
 		txs := types.NewTransactionsByPriceAndNonce(env.signer, remoteTxs, env.header.BaseFee)
-		if w.commitTransactions(env, txs, interrupt, nil, nil) {
-			return
+
+		if err := w.commitTransactions(env, txs, interrupt, nil, nil); err != nil {
+			return err
 		}
 	}
+	return nil
 }
 
 // refreshPending reinitialize pending state
@@ -1476,6 +1486,7 @@ func (w *worker) generateWork(params *generateParams) (*types.Block, error) {
 	defer work.discard()
 
 	w.fillTransactions(nil, work)
+
 	return w.engine.FinalizeAndAssemble(w.chain, work.header, work.state, work.txs, work.unclelist(), work.receipts)
 }
 
@@ -1631,7 +1642,12 @@ func (w *worker) commitWork(interrupt *int32, noempty bool, timestamp int64) {
 		w.commit(work.copy(), nil, false, start)
 	}
 	// Fill pending transactions from the txpool
-	w.fillTransactions(interrupt, work)
+	err = w.fillTransactions(interrupt, work)
+	if errors.Is(err, errBlockInterruptedByNewHead) {
+		work.discard()
+		return
+	}
+
 	w.commit(work.copy(), w.fullTaskHook, true, start)
 
 	// Swap out the old work with the new one, terminating any leftover