Support for syncing on mergemock

.gitignore

@@ -71,4 +71,4 @@ docker-compose.dev.yml
 libmdbx/build/*
 tests/testdata/*
 
-go.work
+go.work

cmd/integration/commands/stages.go

@@ -1128,7 +1128,7 @@ func newSync(ctx context.Context, db kv.RwDB, miningConfig *params.MiningConfig)
 
 	miningSync := stagedsync.New(
 		stagedsync.MiningStages(ctx,
-			stagedsync.StageMiningCreateBlockCfg(db, miner, *chainConfig, engine, nil, nil, tmpdir),
+			stagedsync.StageMiningCreateBlockCfg(db, miner, *chainConfig, engine, nil, nil, stagedsync.PresetHeaderFields{}, tmpdir),
 			stagedsync.StageMiningExecCfg(db, miner, events, *chainConfig, engine, &vm.Config{}, tmpdir),
 			stagedsync.StageHashStateCfg(db, tmpdir),
 			stagedsync.StageTrieCfg(db, false, true, tmpdir, getBlockReader(chainConfig)),

cmd/integration/commands/state_stages.go

@@ -308,7 +308,7 @@ func syncBySmallSteps(db kv.RwDB, miningConfig params.MiningConfig, ctx context.
 			miner.MiningConfig.ExtraData = nextBlock.Extra()
 			miningStages.MockExecFunc(stages.MiningCreateBlock, func(firstCycle bool, badBlockUnwind bool, s *stagedsync.StageState, u stagedsync.Unwinder, tx kv.RwTx) error {
 				err = stagedsync.SpawnMiningCreateBlockStage(s, tx,
-					stagedsync.StageMiningCreateBlockCfg(db, miner, *chainConfig, engine, nil, nil, tmpDir),
+					stagedsync.StageMiningCreateBlockCfg(db, miner, *chainConfig, engine, nil, nil, stagedsync.PresetHeaderFields{}, tmpDir),
 					quit)
 				if err != nil {
 					return err

cmd/rpcdaemon/commands/engine_api.go

@@ -3,11 +3,13 @@ package commands
 import (
 	"bytes"
 	"context"
+	"encoding/binary"
 	"fmt"
 	"math/big"
 
 	"github.com/holiman/uint256"
 	"github.com/ledgerwatch/erigon-lib/gointerfaces"
+	"github.com/ledgerwatch/erigon-lib/gointerfaces/remote"
 	types2 "github.com/ledgerwatch/erigon-lib/gointerfaces/types"
 	"github.com/ledgerwatch/erigon-lib/kv"
 	"github.com/ledgerwatch/erigon/cmd/rpcdaemon/services"
@@ -37,6 +39,13 @@ type ExecutionPayload struct {
 	Transactions  []hexutil.Bytes `json:"transactions"  gencodec:"required"`
 }
 
+// PayloadAttributes represent the attributes required to start assembling a payload
+type ForkChoiceState struct {
+	HeadHash           common.Hash `json:"headBlockHash"             gencodec:"required"`
+	SafeBlockHash      common.Hash `json:"safeBlockHash"             gencodec:"required"`
+	FinalizedBlockHash common.Hash `json:"finalizedBlockHash"        gencodec:"required"`
+}
+
 // PayloadAttributes represent the attributes required to start assembling a payload
 type PayloadAttributes struct {
 	Timestamp             hexutil.Uint64 `json:"timestamp"             gencodec:"required"`
@@ -46,9 +55,9 @@ type PayloadAttributes struct {
 
 // EngineAPI Beacon chain communication endpoint
 type EngineAPI interface {
-	ForkchoiceUpdatedV1(context.Context, struct{}, *PayloadAttributes) (map[string]interface{}, error)
+	ForkchoiceUpdatedV1(ctx context.Context, forkChoiceState *ForkChoiceState, payloadAttributes *PayloadAttributes) (map[string]interface{}, error)
 	ExecutePayloadV1(context.Context, *ExecutionPayload) (map[string]interface{}, error)
-	GetPayloadV1(ctx context.Context, payloadID hexutil.Uint64) (*ExecutionPayload, error)
+	GetPayloadV1(ctx context.Context, payloadID hexutil.Bytes) (*ExecutionPayload, error)
 	GetPayloadBodiesV1(ctx context.Context, blockHashes []rpc.BlockNumberOrHash) (map[common.Hash]ExecutionPayload, error)
 }
 
@@ -62,16 +71,42 @@ type EngineImpl struct {
 // ForkchoiceUpdatedV1 is executed only if we are running a beacon validator,
 // in erigon we do not use this for reorgs like go-ethereum does since we can do that in engine_executePayloadV1
 // if the payloadAttributes is different than null, we return
-func (e *EngineImpl) ForkchoiceUpdatedV1(_ context.Context, _ struct{}, payloadAttributes *PayloadAttributes) (map[string]interface{}, error) {
+func (e *EngineImpl) ForkchoiceUpdatedV1(ctx context.Context, forkChoiceState *ForkChoiceState, payloadAttributes *PayloadAttributes) (map[string]interface{}, error) {
 	// Unwinds can be made within engine_excutePayloadV1 so we can return success regardless
 	if payloadAttributes == nil {
 		return map[string]interface{}{
-			"status":    "SUCCESS",
-			"payloadId": nil,
+			"status": "SUCCESS",
 		}, nil
 	}
 	// Request for assembling payload
-	return nil, fmt.Errorf("invalid request")
+	reply, err := e.api.EngineForkchoiceUpdateV1(ctx, &remote.EngineForkChoiceUpdatedRequest{
+		Prepare: &remote.EnginePreparePayload{
+			Timestamp:    uint64(payloadAttributes.Timestamp),
+			Random:       gointerfaces.ConvertHashToH256(payloadAttributes.Random),
+			FeeRecipient: gointerfaces.ConvertAddressToH160(payloadAttributes.SuggestedFeeRecipient),
+		},
+		Forkchoice: &remote.EngineForkChoiceUpdated{
+			HeadBlockHash:      gointerfaces.ConvertHashToH256(forkChoiceState.HeadHash),
+			FinalizedBlockHash: gointerfaces.ConvertHashToH256(forkChoiceState.FinalizedBlockHash),
+			SafeBlockHash:      gointerfaces.ConvertHashToH256(forkChoiceState.SafeBlockHash),
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	// Process reply
+	if reply.Status == "SYNCING" {
+		return map[string]interface{}{
+			"status": reply.Status,
+		}, nil
+	}
+	encodedPayloadId := make([]byte, 8)
+	binary.BigEndian.PutUint64(encodedPayloadId, reply.PayloadId)
+	// Answer
+	return map[string]interface{}{
+		"status":    reply.Status,
+		"payloadId": hexutil.Bytes(encodedPayloadId),
+	}, nil
 }
 
 // ExecutePayloadV1 takes a block from the beacon chain and do either two of the following things
@@ -86,9 +121,6 @@ func (e *EngineImpl) ExecutePayloadV1(ctx context.Context, payload *ExecutionPay
 			return nil, fmt.Errorf("invalid request")
 		}
 	}
-	// Maximum length of extra is 32 bytes so we can use the hash datatype
-	extra := common.BytesToHash(payload.ExtraData)
-
 	log.Info("Received Payload from beacon-chain")
 
 	// Convert slice of hexutil.Bytes to a slice of slice of bytes
@@ -107,7 +139,7 @@ func (e *EngineImpl) ExecutePayloadV1(ctx context.Context, payload *ExecutionPay
 		GasLimit:      (uint64)(payload.GasLimit),
 		GasUsed:       (uint64)(payload.GasUsed),
 		Timestamp:     (uint64)(payload.Timestamp),
-		ExtraData:     gointerfaces.ConvertHashToH256(extra),
+		ExtraData:     payload.ExtraData,
 		BaseFeePerGas: gointerfaces.ConvertUint256IntToH256(baseFee),
 		BlockHash:     gointerfaces.ConvertHashToH256(payload.BlockHash),
 		Transactions:  transactions,
@@ -120,21 +152,21 @@ func (e *EngineImpl) ExecutePayloadV1(ctx context.Context, payload *ExecutionPay
 		var latestValidHash common.Hash = gointerfaces.ConvertH256ToHash(res.LatestValidHash)
 		return map[string]interface{}{
 			"status":          res.Status,
-			"latestValidHash": common.Bytes2Hex(latestValidHash.Bytes()),
+			"latestValidHash": latestValidHash,
 		}, nil
 	}
 	return map[string]interface{}{
 		"status": res.Status,
 	}, nil
 }
 
-func (e *EngineImpl) GetPayloadV1(ctx context.Context, payloadID hexutil.Uint64) (*ExecutionPayload, error) {
-	payload, err := e.api.EngineGetPayloadV1(ctx, (uint64)(payloadID))
+func (e *EngineImpl) GetPayloadV1(ctx context.Context, payloadID hexutil.Bytes) (*ExecutionPayload, error) {
+	decodedPayloadId := binary.BigEndian.Uint64(payloadID)
+	payload, err := e.api.EngineGetPayloadV1(ctx, decodedPayloadId)
 	if err != nil {
 		return nil, err
 	}
 	var bloom types.Bloom = gointerfaces.ConvertH2048ToBloom(payload.LogsBloom)
-	var extra common.Hash = gointerfaces.ConvertH256ToHash(payload.ExtraData)
 
 	var baseFee *big.Int
 	if payload.BaseFeePerGas != nil {
@@ -157,7 +189,7 @@ func (e *EngineImpl) GetPayloadV1(ctx context.Context, payloadID hexutil.Uint64)
 		GasLimit:      hexutil.Uint64(payload.GasLimit),
 		GasUsed:       hexutil.Uint64(payload.GasUsed),
 		Timestamp:     hexutil.Uint64(payload.Timestamp),
-		ExtraData:     extra[:],
+		ExtraData:     payload.ExtraData,
 		BaseFeePerGas: (*hexutil.Big)(baseFee),
 		BlockHash:     gointerfaces.ConvertH256ToHash(payload.BlockHash),
 		Transactions:  transactions,
@@ -176,20 +208,16 @@ func (e *EngineImpl) GetPayloadBodiesV1(ctx context.Context, blockHashes []rpc.B
 
 	for _, blockHash := range blockHashes {
 		hash := *blockHash.BlockHash
-
 		block, err := e.blockByHashWithSenders(tx, hash)
 		if err != nil {
 			return nil, err
 		}
-
 		if block == nil {
 			continue
 		}
 
 		var bloom types.Bloom = block.Bloom()
-
 		buf := bytes.NewBuffer(nil)
-
 		var encodedTransactions []hexutil.Bytes
 
 		for _, tx := range block.Transactions() {
@@ -199,7 +227,6 @@ func (e *EngineImpl) GetPayloadBodiesV1(ctx context.Context, blockHashes []rpc.B
 			if err != nil {
 				return nil, fmt.Errorf("broken tx rlp: %w", err)
 			}
-
 			encodedTransactions = append(encodedTransactions, common.CopyBytes(buf.Bytes()))
 		}
 
@@ -219,7 +246,6 @@ func (e *EngineImpl) GetPayloadBodiesV1(ctx context.Context, blockHashes []rpc.B
 			BlockHash:     block.Hash(),
 			Transactions:  encodedTransactions,
 		}
-
 	}
 	return blockHashToBody, nil
 }

cmd/rpcdaemon/rpcdaemontest/test_util.go

@@ -221,7 +221,7 @@ func CreateTestGrpcConn(t *testing.T, m *stages.MockSentry) (context.Context, *g
 	ethashApi := apis[1].Service.(*ethash.API)
 	server := grpc.NewServer()
 
-	remote.RegisterETHBACKENDServer(server, privateapi.NewEthBackendServer(ctx, nil, m.DB, m.Notifications.Events, snapshotsync.NewBlockReader(), nil, nil, nil, nil))
+	remote.RegisterETHBACKENDServer(server, privateapi.NewEthBackendServer(ctx, nil, m.DB, m.Notifications.Events, snapshotsync.NewBlockReader(), nil, nil, nil, nil, nil))
 	txpool.RegisterTxpoolServer(server, m.TxPoolGrpcServer)
 	txpool.RegisterMiningServer(server, privateapi.NewMiningServer(ctx, &IsMiningMock{}, ethashApi))
 	listener := bufconn.Listen(1024 * 1024)

cmd/rpcdaemon/services/eth_backend.go

@@ -34,6 +34,7 @@ type ApiBackend interface {
 	Subscribe(ctx context.Context, cb func(*remote.SubscribeReply)) error
 	BlockWithSenders(ctx context.Context, tx kv.Tx, hash common.Hash, blockHeight uint64) (block *types.Block, senders []common.Address, err error)
 	EngineExecutePayloadV1(ctx context.Context, payload *types2.ExecutionPayload) (*remote.EngineExecutePayloadReply, error)
+	EngineForkchoiceUpdateV1(ctx context.Context, request *remote.EngineForkChoiceUpdatedRequest) (*remote.EngineForkChoiceUpdatedReply, error)
 	EngineGetPayloadV1(ctx context.Context, payloadId uint64) (*types2.ExecutionPayload, error)
 	NodeInfo(ctx context.Context, limit uint32) ([]p2p.NodeInfo, error)
 }
@@ -164,6 +165,10 @@ func (back *RemoteBackend) EngineExecutePayloadV1(ctx context.Context, payload *
 	return back.remoteEthBackend.EngineExecutePayloadV1(ctx, payload)
 }
 
+func (back *RemoteBackend) EngineForkchoiceUpdateV1(ctx context.Context, request *remote.EngineForkChoiceUpdatedRequest) (*remote.EngineForkChoiceUpdatedReply, error) {
+	return back.remoteEthBackend.EngineForkChoiceUpdatedV1(ctx, request)
+}
+
 func (back *RemoteBackend) EngineGetPayloadV1(ctx context.Context, payloadId uint64) (res *types2.ExecutionPayload, err error) {
 	return back.remoteEthBackend.EngineGetPayloadV1(ctx, &remote.EngineGetPayloadRequest{
 		PayloadId: payloadId,

eth/backend.go

@@ -132,9 +132,9 @@ type Ethereum struct {
 	notifyMiningAboutNewTxs chan struct{}
 	// When we receive something here, it means that the beacon chain transitioned
 	// to proof-of-stake so we start reverse syncing from the header
-	reverseDownloadCh    chan types.Header
-	statusCh             chan privateapi.ExecutionStatus
-	waitingForPOSHeaders uint32 // atomic boolean flag
+	reverseDownloadCh     chan privateapi.PayloadMessage
+	statusCh              chan privateapi.ExecutionStatus
+	waitingForBeaconChain uint32 // atomic boolean flag
 }
 
 // New creates a new Ethereum object (including the
@@ -344,7 +344,7 @@ func New(stack *node.Node, config *ethconfig.Config, logger log.Logger) (*Ethere
 	miner := stagedsync.NewMiningState(&config.Miner)
 	backend.pendingBlocks = miner.PendingResultCh
 	backend.minedBlocks = miner.MiningResultCh
-	backend.reverseDownloadCh = make(chan types.Header)
+	backend.reverseDownloadCh = make(chan privateapi.PayloadMessage)
 	backend.statusCh = make(chan privateapi.ExecutionStatus)
 
 	var blockReader interfaces.FullBlockReader
@@ -393,7 +393,7 @@ func New(stack *node.Node, config *ethconfig.Config, logger log.Logger) (*Ethere
 
 	mining := stagedsync.New(
 		stagedsync.MiningStages(backend.sentryCtx,
-			stagedsync.StageMiningCreateBlockCfg(backend.chainDB, miner, *backend.chainConfig, backend.engine, backend.txPool2, backend.txPool2DB, tmpdir),
+			stagedsync.StageMiningCreateBlockCfg(backend.chainDB, miner, *backend.chainConfig, backend.engine, backend.txPool2, backend.txPool2DB, stagedsync.PresetHeaderFields{}, tmpdir),
 			stagedsync.StageMiningExecCfg(backend.chainDB, miner, backend.notifications.Events, *backend.chainConfig, backend.engine, &vm.Config{}, tmpdir),
 			stagedsync.StageHashStateCfg(backend.chainDB, tmpdir),
 			stagedsync.StageTrieCfg(backend.chainDB, false, true, tmpdir, blockReader),
@@ -404,9 +404,28 @@ func New(stack *node.Node, config *ethconfig.Config, logger log.Logger) (*Ethere
 	if casted, ok := backend.engine.(*ethash.Ethash); ok {
 		ethashApi = casted.APIs(nil)[1].Service.(*ethash.API)
 	}
-	atomic.StoreUint32(&backend.waitingForPOSHeaders, 0)
+	atomic.StoreUint32(&backend.waitingForBeaconChain, 0)
+	startAssembleFunc := func(timestamp uint64, random common.Hash) (types.Block, error) {
+		presets := stagedsync.PresetHeaderFields{
+			Timestamp: timestamp,
+			Random:    random,
+		}
+		proposing := stagedsync.New(
+			stagedsync.MiningStages(backend.sentryCtx,
+				stagedsync.StageMiningCreateBlockCfg(backend.chainDB, miner, *backend.chainConfig, backend.engine, backend.txPool2, backend.txPool2DB, presets, tmpdir),
+				stagedsync.StageMiningExecCfg(backend.chainDB, miner, backend.notifications.Events, *backend.chainConfig, backend.engine, &vm.Config{}, tmpdir),
+				stagedsync.StageHashStateCfg(backend.chainDB, tmpdir),
+				stagedsync.StageTrieCfg(backend.chainDB, false, true, tmpdir, blockReader),
+				stagedsync.StageMiningFinishCfg(backend.chainDB, *backend.chainConfig, backend.engine, miner, backend.miningSealingQuit),
+			), stagedsync.MiningUnwindOrder, stagedsync.MiningPruneOrder)
+		if err := stages2.MiningStep(ctx, backend.chainDB, proposing); err != nil {
+			return types.Block{}, err
+		}
+		block := <-miner.ProposeBlockCh
+		return *block, nil
+	}
 	ethBackendRPC := privateapi.NewEthBackendServer(ctx, backend, backend.chainDB, backend.notifications.Events,
-		blockReader, chainConfig, backend.reverseDownloadCh, backend.statusCh, &backend.waitingForPOSHeaders)
+		blockReader, chainConfig, backend.reverseDownloadCh, backend.statusCh, &backend.waitingForBeaconChain, startAssembleFunc)
 	miningRPC = privateapi.NewMiningServer(ctx, backend, ethashApi)
 	if stack.Config().PrivateApiAddr != "" {
 		var creds credentials.TransportCredentials
@@ -478,7 +497,7 @@ func New(stack *node.Node, config *ethconfig.Config, logger log.Logger) (*Ethere
 	backend.stagedSync, err = stages2.NewStagedSync(backend.sentryCtx, backend.logger, backend.chainDB,
 		stack.Config().P2P, *config, chainConfig.TerminalTotalDifficulty,
 		backend.sentryControlServer, tmpdir, backend.notifications.Accumulator,
-		backend.reverseDownloadCh, backend.statusCh, &backend.waitingForPOSHeaders,
+		backend.reverseDownloadCh, backend.statusCh, &backend.waitingForBeaconChain,
 		backend.downloaderClient)
 	if err != nil {
 		return nil, err
@@ -609,15 +628,19 @@ func (s *Ethereum) StartMining(ctx context.Context, db kv.RwDB, mining *stagedsy
 		defer debug.LogPanic()
 		defer close(s.waitForMiningStop)
 
-		mineEvery := time.NewTicker(3 * time.Second)
+		mineEvery := time.NewTicker(1 * time.Second)
 		defer mineEvery.Stop()
 
 		var works bool
 		var hasWork bool
 		errc := make(chan error, 1)
-
+		roTx, err := s.chainDB.BeginRo(ctx)
+		if err != nil {
+			log.Warn("mining", "err", err)
+			return
+		}
 		for {
-			mineEvery.Reset(3 * time.Second)
+			mineEvery.Reset(1 * time.Second)
 			select {
 			case <-s.notifyMiningAboutNewTxs:
 				hasWork = true
@@ -636,6 +659,18 @@ func (s *Ethereum) StartMining(ctx context.Context, db kv.RwDB, mining *stagedsy
 				return
 			}
 
+			headNumber := rawdb.ReadCurrentBlockNumber(roTx)
+			isTrans, err := rawdb.Transitioned(roTx, *headNumber, s.chainConfig.TerminalTotalDifficulty)
+			if err != nil {
+				log.Warn("mining", "err", err)
+				return
+			}
+
+			if isTrans {
+				log.Info("Validating on PoS engine")
+				return
+			}
+
 			if !works && hasWork {
 				works = true
 				go func() { errc <- stages2.MiningStep(ctx, db, mining) }()