miner, eth: implement recommit mechanism for payload building (#25836)

* miner, eth: implement recommit for payload building

* miner: address comments from marius

miner/miner.go

@@ -251,26 +251,7 @@ func (miner *Miner) SubscribePendingLogs(ch chan<- []*types.Log) event.Subscript
 	return miner.worker.pendingLogsFeed.Subscribe(ch)
 }
 
-// GetSealingBlockAsync requests to generate a sealing block according to the
-// given parameters. Regardless of whether the generation is successful or not,
-// there is always a result that will be returned through the result channel.
-// The difference is that if the execution fails, the returned result is nil
-// and the concrete error is dropped silently.
-func (miner *Miner) GetSealingBlockAsync(parent common.Hash, timestamp uint64, coinbase common.Address, random common.Hash, noTxs bool) (chan *types.Block, error) {
-	resCh, _, err := miner.worker.getSealingBlock(parent, timestamp, coinbase, random, noTxs)
-	if err != nil {
-		return nil, err
-	}
-	return resCh, nil
-}
-
-// GetSealingBlockSync creates a sealing block according to the given parameters.
-// If the generation is failed or the underlying work is already closed, an error
-// will be returned.
-func (miner *Miner) GetSealingBlockSync(parent common.Hash, timestamp uint64, coinbase common.Address, random common.Hash, noTxs bool) (*types.Block, error) {
-	resCh, errCh, err := miner.worker.getSealingBlock(parent, timestamp, coinbase, random, noTxs)
-	if err != nil {
-		return nil, err
-	}
-	return <-resCh, <-errCh
+// BuildPayload builds the payload according to the provided parameters.
+func (miner *Miner) BuildPayload(args *BuildPayloadArgs) (*Payload, error) {
+	return miner.worker.buildPayload(args)
 }

miner/payload_building.go

@@ -0,0 +1,168 @@
+// Copyright 2022 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>
+
+package miner
+
+import (
+	"math/big"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/beacon"
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+// BuildPayloadArgs contains the provided parameters for building payload.
+// Check engine-api specification for more details.
+// https://github.com/ethereum/execution-apis/blob/main/src/engine/specification.md#payloadattributesv1
+type BuildPayloadArgs struct {
+	Parent       common.Hash    // The parent block to build payload on top
+	Timestamp    uint64         // The provided timestamp of generated payload
+	FeeRecipient common.Address // The provided recipient address for collecting transaction fee
+	Random       common.Hash    // The provided randomness value
+}
+
+// Payload wraps the built payload(block waiting for sealing). According to the
+// engine-api specification, EL should build the initial version of the payload
+// which has an empty transaction set and then keep update it in order to maximize
+// the revenue. Therefore, the empty-block here is always available and full-block
+// will be set/updated afterwards.
+type Payload struct {
+	empty    *types.Block
+	full     *types.Block
+	fullFees *big.Int
+	stop     chan struct{}
+	lock     *sync.Mutex
+	cond     *sync.Cond
+}
+
+// newPayload initializes the payload object.
+func newPayload(empty *types.Block) *Payload {
+	lock := new(sync.Mutex)
+	return &Payload{
+		empty: empty,
+		stop:  make(chan struct{}),
+		lock:  lock,
+		cond:  sync.NewCond(lock),
+	}
+}
+
+// update updates the full-block with latest built version.
+func (payload *Payload) update(block *types.Block, fees *big.Int) {
+	payload.lock.Lock()
+	defer payload.lock.Unlock()
+
+	select {
+	case <-payload.stop:
+		return // reject stale update
+	default:
+	}
+	// Ensure the newly provided full block has a higher transaction fee.
+	// In post-merge stage, there is no uncle reward anymore and transaction
+	// fee(apart from the mev revenue) is the only indicator for comparison.
+	if payload.full == nil || fees.Cmp(payload.fullFees) > 0 {
+		payload.full = block
+		payload.fullFees = fees
+	}
+	payload.cond.Broadcast() // fire signal for notifying full block
+}
+
+// Resolve returns the latest built payload and also terminates the background
+// thread for updating payload. It's safe to be called multiple times.
+func (payload *Payload) Resolve() *beacon.ExecutableDataV1 {
+	payload.lock.Lock()
+	defer payload.lock.Unlock()
+
+	select {
+	case <-payload.stop:
+	default:
+		close(payload.stop)
+	}
+	if payload.full != nil {
+		return beacon.BlockToExecutableData(payload.full)
+	}
+	return beacon.BlockToExecutableData(payload.empty)
+}
+
+// ResolveEmpty is basically identical to Resolve, but it expects empty block only.
+// It's only used in tests.
+func (payload *Payload) ResolveEmpty() *beacon.ExecutableDataV1 {
+	payload.lock.Lock()
+	defer payload.lock.Unlock()
+
+	return beacon.BlockToExecutableData(payload.empty)
+}
+
+// ResolveFull is basically identical to Resolve, but it expects full block only.
+// It's only used in tests.
+func (payload *Payload) ResolveFull() *beacon.ExecutableDataV1 {
+	payload.lock.Lock()
+	defer payload.lock.Unlock()
+
+	if payload.full == nil {
+		select {
+		case <-payload.stop:
+			return nil
+		default:
+		}
+		payload.cond.Wait()
+	}
+	return beacon.BlockToExecutableData(payload.full)
+}
+
+// buildPayload builds the payload according to the provided parameters.
+func (w *worker) buildPayload(args *BuildPayloadArgs) (*Payload, error) {
+	// Build the initial version with no transaction included. It should be fast
+	// enough to run. The empty payload can at least make sure there is something
+	// to deliver for not missing slot.
+	empty, _, err := w.getSealingBlock(args.Parent, args.Timestamp, args.FeeRecipient, args.Random, true)
+	if err != nil {
+		return nil, err
+	}
+	// Construct a payload object for return.
+	payload := newPayload(empty)
+
+	// Spin up a routine for updating the payload in background. This strategy
+	// can maximum the revenue for including transactions with highest fee.
+	go func() {
+		// Setup the timer for re-building the payload. The initial clock is kept
+		// for triggering process immediately.
+		timer := time.NewTimer(0)
+		defer timer.Stop()
+
+		// Setup the timer for terminating the process if SECONDS_PER_SLOT (12s in
+		// the Mainnet configuration) have passed since the point in time identified
+		// by the timestamp parameter.
+		endTimer := time.NewTimer(time.Second * 12)
+
+		for {
+			select {
+			case <-timer.C:
+				block, fees, err := w.getSealingBlock(args.Parent, args.Timestamp, args.FeeRecipient, args.Random, false)
+				if err == nil {
+					payload.update(block, fees)
+				}
+				timer.Reset(w.recommit)
+			case <-payload.stop:
+				return
+			case <-endTimer.C:
+				return
+			}
+		}
+	}()
+	return payload, nil
+}

miner/payload_building_test.go

@@ -0,0 +1,80 @@
+// Copyright 2022 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>
+
+package miner
+
+import (
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/consensus/ethash"
+	"github.com/ethereum/go-ethereum/core/beacon"
+	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/params"
+)
+
+func TestBuildPayload(t *testing.T) {
+	var (
+		db        = rawdb.NewMemoryDatabase()
+		recipient = common.HexToAddress("0xdeadbeef")
+	)
+	w, b := newTestWorker(t, params.TestChainConfig, ethash.NewFaker(), db, 0)
+	defer w.close()
+
+	timestamp := uint64(time.Now().Unix())
+	args := &BuildPayloadArgs{
+		Parent:       b.chain.CurrentBlock().Hash(),
+		Timestamp:    timestamp,
+		Random:       common.Hash{},
+		FeeRecipient: recipient,
+	}
+	payload, err := w.buildPayload(args)
+	if err != nil {
+		t.Fatalf("Failed to build payload %v", err)
+	}
+	verify := func(data *beacon.ExecutableDataV1, txs int) {
+		if data.ParentHash != b.chain.CurrentBlock().Hash() {
+			t.Fatal("Unexpect parent hash")
+		}
+		if data.Random != (common.Hash{}) {
+			t.Fatal("Unexpect random value")
+		}
+		if data.Timestamp != timestamp {
+			t.Fatal("Unexpect timestamp")
+		}
+		if data.FeeRecipient != recipient {
+			t.Fatal("Unexpect fee recipient")
+		}
+		if len(data.Transactions) != txs {
+			t.Fatal("Unexpect transaction set")
+		}
+	}
+	empty := payload.ResolveEmpty()
+	verify(empty, 0)
+
+	full := payload.ResolveFull()
+	verify(full, len(pendingTxs))
+
+	// Ensure resolve can be called multiple times and the
+	// result should be unchanged
+	dataOne := payload.Resolve()
+	dataTwo := payload.Resolve()
+	if !reflect.DeepEqual(dataOne, dataTwo) {
+		t.Fatal("Unexpected payload data")
+	}
+}

miner/stress/beacon/main.go

@@ -81,8 +81,8 @@ var (
 	transitionDifficulty = new(big.Int).Mul(big.NewInt(20), params.MinimumDifficulty)
 
 	// blockInterval is the time interval for creating a new eth2 block
-	blockInterval    = time.Second * 3
 	blockIntervalInt = 3
+	blockInterval    = time.Second * time.Duration(blockIntervalInt)
 
 	// finalizationDist is the block distance for finalizing block
 	finalizationDist = 10
@@ -164,6 +164,7 @@ func (n *ethNode) assembleBlock(parentHash common.Hash, parentTimestamp uint64)
 	if err != nil {
 		return nil, err
 	}
+	time.Sleep(time.Second * 5) // give enough time for block creation
 	return n.api.GetPayloadV1(*payload.PayloadID)
 }
 
@@ -316,17 +317,14 @@ func (mgr *nodeManager) run() {
 		}
 		nodes := mgr.getNodes(eth2MiningNode)
 		nodes = append(nodes, mgr.getNodes(eth2NormalNode)...)
-		nodes = append(nodes, mgr.getNodes(eth2LightClient)...)
+		//nodes = append(nodes, mgr.getNodes(eth2LightClient)...)
 		for _, node := range nodes {
 			fcState := beacon.ForkchoiceStateV1{
-				HeadBlockHash:      oldest.Hash(),
-				SafeBlockHash:      common.Hash{},
+				HeadBlockHash:      parentBlock.Hash(),
+				SafeBlockHash:      oldest.Hash(),
 				FinalizedBlockHash: oldest.Hash(),
 			}
-			// TODO(rjl493456442) finalization doesn't work properly, FIX IT
-			_ = fcState
-			_ = node
-			//node.api.ForkchoiceUpdatedV1(fcState, nil)
+			node.api.ForkchoiceUpdatedV1(fcState, nil)
 		}
 		log.Info("Finalised eth2 block", "number", oldest.NumberU64(), "hash", oldest.Hash())
 		waitFinalise = waitFinalise[1:]
@@ -423,7 +421,7 @@ func main() {
 		node := nodes[index%len(nodes)]
 
 		// Create a self transaction and inject into the pool
-		tx, err := types.SignTx(types.NewTransaction(nonces[index], crypto.PubkeyToAddress(faucets[index].PublicKey), new(big.Int), 21000, big.NewInt(100000000000+rand.Int63n(65536)), nil), types.HomesteadSigner{}, faucets[index])
+		tx, err := types.SignTx(types.NewTransaction(nonces[index], crypto.PubkeyToAddress(faucets[index].PublicKey), new(big.Int), 21000, big.NewInt(10_000_000_000+rand.Int63n(6_553_600_000)), nil), types.HomesteadSigner{}, faucets[index])
 		if err != nil {
 			panic(err)
 		}
@@ -492,7 +490,7 @@ func makeFullNode(genesis *core.Genesis) (*node.Node, *eth.Ethereum, *ethcatalys
 			GasFloor: genesis.GasLimit * 9 / 10,
 			GasCeil:  genesis.GasLimit * 11 / 10,
 			GasPrice: big.NewInt(1),
-			Recommit: 10 * time.Second, // Disable the recommit
+			Recommit: 1 * time.Second,
 		},
 		LightServ:        100,
 		LightPeers:       10,

miner/worker.go

@@ -169,11 +169,17 @@ type newWorkReq struct {
 	timestamp int64
 }
 
+// newPayloadResult represents a result struct corresponds to payload generation.
+type newPayloadResult struct {
+	err   error
+	block *types.Block
+	fees  *big.Int
+}
+
 // getWorkReq represents a request for getting a new sealing work with provided parameters.
 type getWorkReq struct {
 	params *generateParams
-	result chan *types.Block // non-blocking channel
-	err    chan error
+	result chan *newPayloadResult // non-blocking channel
 }
 
 // intervalAdjust represents a resubmitting interval adjustment.
@@ -250,6 +256,10 @@ type worker struct {
 	// in case there are some computation expensive transactions in txpool.
 	newpayloadTimeout time.Duration
 
+	// recommit is the time interval to re-create sealing work or to re-build
+	// payload in proof-of-stake stage.
+	recommit time.Duration
+
 	// External functions
 	isLocalBlock func(header *types.Header) bool // Function used to determine whether the specified block is mined by local miner.
 
@@ -297,6 +307,8 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus
 		log.Warn("Sanitizing miner recommit interval", "provided", recommit, "updated", minRecommitInterval)
 		recommit = minRecommitInterval
 	}
+	worker.recommit = recommit
+
 	// Sanitize the timeout config for creating payload.
 	newpayloadTimeout := worker.config.NewPayloadTimeout
 	if newpayloadTimeout == 0 {
@@ -553,13 +565,11 @@ func (w *worker) mainLoop() {
 			w.commitWork(req.interrupt, req.noempty, req.timestamp)
 
 		case req := <-w.getWorkCh:
-			block, err := w.generateWork(req.params)
-			if err != nil {
-				req.err <- err
-				req.result <- nil
-			} else {
-				req.err <- nil
-				req.result <- block
+			block, fees, err := w.generateWork(req.params)
+			req.result <- &newPayloadResult{
+				err:   err,
+				block: block,
+				fees:  fees,
 			}
 		case ev := <-w.chainSideCh:
 			// Short circuit for duplicate side blocks
@@ -1071,10 +1081,10 @@ func (w *worker) fillTransactions(interrupt *int32, env *environment) error {
 }
 
 // generateWork generates a sealing block based on the given parameters.
-func (w *worker) generateWork(params *generateParams) (*types.Block, error) {
+func (w *worker) generateWork(params *generateParams) (*types.Block, *big.Int, error) {
 	work, err := w.prepareWork(params)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	defer work.discard()
 
@@ -1090,7 +1100,11 @@ func (w *worker) generateWork(params *generateParams) (*types.Block, error) {
 			log.Warn("Block building is interrupted", "allowance", common.PrettyDuration(w.newpayloadTimeout))
 		}
 	}
-	return w.engine.FinalizeAndAssemble(w.chain, work.header, work.state, work.txs, work.unclelist(), work.receipts)
+	block, err := w.engine.FinalizeAndAssemble(w.chain, work.header, work.state, work.txs, work.unclelist(), work.receipts)
+	if err != nil {
+		return nil, nil, err
+	}
+	return block, totalFees(block, work.receipts), nil
 }
 
 // commitWork generates several new sealing tasks based on the parent block
@@ -1180,9 +1194,12 @@ func (w *worker) commit(env *environment, interval func(), update bool, start ti
 			select {
 			case w.taskCh <- &task{receipts: env.receipts, state: env.state, block: block, createdAt: time.Now()}:
 				w.unconfirmed.Shift(block.NumberU64() - 1)
+
+				fees := totalFees(block, env.receipts)
+				feesInEther := new(big.Float).Quo(new(big.Float).SetInt(fees), new(big.Float).SetInt(big.NewInt(params.Ether)))
 				log.Info("Commit new sealing work", "number", block.Number(), "sealhash", w.engine.SealHash(block.Header()),
 					"uncles", len(env.uncles), "txs", env.tcount,
-					"gas", block.GasUsed(), "fees", totalFees(block, env.receipts),
+					"gas", block.GasUsed(), "fees", feesInEther,
 					"elapsed", common.PrettyDuration(time.Since(start)))
 
 			case <-w.exitCh:
@@ -1199,11 +1216,7 @@ func (w *worker) commit(env *environment, interval func(), update bool, start ti
 // getSealingBlock generates the sealing block based on the given parameters.
 // The generation result will be passed back via the given channel no matter
 // the generation itself succeeds or not.
-func (w *worker) getSealingBlock(parent common.Hash, timestamp uint64, coinbase common.Address, random common.Hash, noTxs bool) (chan *types.Block, chan error, error) {
-	var (
-		resCh = make(chan *types.Block, 1)
-		errCh = make(chan error, 1)
-	)
+func (w *worker) getSealingBlock(parent common.Hash, timestamp uint64, coinbase common.Address, random common.Hash, noTxs bool) (*types.Block, *big.Int, error) {
 	req := &getWorkReq{
 		params: &generateParams{
 			timestamp:  timestamp,
@@ -1215,12 +1228,15 @@ func (w *worker) getSealingBlock(parent common.Hash, timestamp uint64, coinbase
 			noExtra:    true,
 			noTxs:      noTxs,
 		},
-		result: resCh,
-		err:    errCh,
+		result: make(chan *newPayloadResult, 1),
 	}
 	select {
 	case w.getWorkCh <- req:
-		return resCh, errCh, nil
+		result := <-req.result
+		if result.err != nil {
+			return nil, nil, result.err
+		}
+		return result.block, result.fees, nil
 	case <-w.exitCh:
 		return nil, nil, errors.New("miner closed")
 	}
@@ -1251,14 +1267,14 @@ func (w *worker) postSideBlock(event core.ChainSideEvent) {
 	}
 }
 
-// totalFees computes total consumed miner fees in ETH. Block transactions and receipts have to have the same order.
-func totalFees(block *types.Block, receipts []*types.Receipt) *big.Float {
+// totalFees computes total consumed miner fees in Wei. Block transactions and receipts have to have the same order.
+func totalFees(block *types.Block, receipts []*types.Receipt) *big.Int {
 	feesWei := new(big.Int)
 	for i, tx := range block.Transactions() {
 		minerFee, _ := tx.EffectiveGasTip(block.BaseFee())
 		feesWei.Add(feesWei, new(big.Int).Mul(new(big.Int).SetUint64(receipts[i].GasUsed), minerFee))
 	}
-	return new(big.Float).Quo(new(big.Float).SetInt(feesWei), new(big.Float).SetInt(big.NewInt(params.Ether)))
+	return feesWei
 }
 
 // signalToErr converts the interruption signal to a concrete error type for return.

miner/worker_test.go

@@ -634,9 +634,7 @@ func testGetSealingWork(t *testing.T, chainConfig *params.ChainConfig, engine co
 
 	// This API should work even when the automatic sealing is not enabled
 	for _, c := range cases {
-		resChan, errChan, _ := w.getSealingBlock(c.parent, timestamp, c.coinbase, c.random, false)
-		block := <-resChan
-		err := <-errChan
+		block, _, err := w.getSealingBlock(c.parent, timestamp, c.coinbase, c.random, false)
 		if c.expectErr {
 			if err == nil {
 				t.Error("Expect error but get nil")
@@ -652,9 +650,7 @@ func testGetSealingWork(t *testing.T, chainConfig *params.ChainConfig, engine co
 	// This API should work even when the automatic sealing is enabled
 	w.start()
 	for _, c := range cases {
-		resChan, errChan, _ := w.getSealingBlock(c.parent, timestamp, c.coinbase, c.random, false)
-		block := <-resChan
-		err := <-errChan
+		block, _, err := w.getSealingBlock(c.parent, timestamp, c.coinbase, c.random, false)
 		if c.expectErr {
 			if err == nil {
 				t.Error("Expect error but get nil")