New Engine API semantics (Kiln v1)

cmd/integration/commands/stages.go

@@ -1144,7 +1144,7 @@ func newSync(ctx context.Context, db kv.RwDB, miningConfig *params.MiningConfig)
 
 	sync, err := stages2.NewStagedSync(context.Background(), logger, db, p2p.Config{}, cfg,
 		chainConfig.TerminalTotalDifficulty, sentryControlServer, tmpdir,
-		nil, nil, nil, nil,
+		nil, nil, nil, nil, nil,
 	)
 	if err != nil {
 		panic(err)

cmd/rpcdaemon/commands/engine_api.go

@@ -56,7 +56,7 @@ type PayloadAttributes struct {
 // EngineAPI Beacon chain communication endpoint
 type EngineAPI interface {
 	ForkchoiceUpdatedV1(ctx context.Context, forkChoiceState *ForkChoiceState, payloadAttributes *PayloadAttributes) (map[string]interface{}, error)
-	ExecutePayloadV1(context.Context, *ExecutionPayload) (map[string]interface{}, error)
+	NewPayloadV1(context.Context, *ExecutionPayload) (map[string]interface{}, error)
 	GetPayloadV1(ctx context.Context, payloadID hexutil.Bytes) (*ExecutionPayload, error)
 	GetPayloadBodiesV1(ctx context.Context, blockHashes []rpc.BlockNumberOrHash) (map[common.Hash]ExecutionPayload, error)
 }
@@ -68,51 +68,56 @@ type EngineImpl struct {
 	api services.ApiBackend
 }
 
-// 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 convertPayloadStatus(x *remote.EnginePayloadStatus) map[string]interface{} {
+	json := map[string]interface{}{
+		"status": x.Status.String(),
+	}
+	if x.LatestValidHash != nil {
+		json["latestValidHash"] = gointerfaces.ConvertH256ToHash(x.LatestValidHash)
+	}
+	if x.ValidationError != "" {
+		json["validationError"] = x.ValidationError
+	}
+
+	return json
+}
+
 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",
-		}, nil
+	var prepareParameters *remote.EnginePayloadAttributes
+	if payloadAttributes != nil {
+		prepareParameters = &remote.EnginePayloadAttributes{
+			Timestamp:             uint64(payloadAttributes.Timestamp),
+			Random:                gointerfaces.ConvertHashToH256(payloadAttributes.Random),
+			SuggestedFeeRecipient: gointerfaces.ConvertAddressToH160(payloadAttributes.SuggestedFeeRecipient),
+		}
 	}
-	// Request for assembling payload
-	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{
+	reply, err := e.api.EngineForkchoiceUpdatedV1(ctx, &remote.EngineForkChoiceUpdatedRequest{
+		ForkchoiceState: &remote.EngineForkChoiceState{
 			HeadBlockHash:      gointerfaces.ConvertHashToH256(forkChoiceState.HeadHash),
 			FinalizedBlockHash: gointerfaces.ConvertHashToH256(forkChoiceState.FinalizedBlockHash),
 			SafeBlockHash:      gointerfaces.ConvertHashToH256(forkChoiceState.SafeBlockHash),
 		},
+		PayloadAttributes: prepareParameters,
 	})
 	if err != nil {
 		return nil, err
 	}
-	// Process reply
-	if reply.Status == "SYNCING" {
-		return map[string]interface{}{
-			"status": reply.Status,
-		}, nil
+
+	json := map[string]interface{}{
+		"payloadStatus": convertPayloadStatus(reply.PayloadStatus),
+	}
+	if reply.PayloadId != 0 {
+		encodedPayloadId := make([]byte, 8)
+		binary.BigEndian.PutUint64(encodedPayloadId, reply.PayloadId)
+		json["payloadId"] = hexutil.Bytes(encodedPayloadId)
 	}
-	encodedPayloadId := make([]byte, 8)
-	binary.BigEndian.PutUint64(encodedPayloadId, reply.PayloadId)
-	// Answer
-	return map[string]interface{}{
-		"status":    reply.Status,
-		"payloadId": hexutil.Bytes(encodedPayloadId),
-	}, nil
+
+	return json, nil
 }
 
-// ExecutePayloadV1 takes a block from the beacon chain and do either two of the following things
-// - Stageloop the block just received if we have the payload's parent hash already
-// - Start the reverse sync process otherwise, and return "Syncing"
-func (e *EngineImpl) ExecutePayloadV1(ctx context.Context, payload *ExecutionPayload) (map[string]interface{}, error) {
+// NewPayloadV1 processes new payloads (blocks) from the beacon chain.
+// See https://github.com/ethereum/execution-apis/blob/main/src/engine/specification.md#engine_newpayloadv1
+func (e *EngineImpl) NewPayloadV1(ctx context.Context, payload *ExecutionPayload) (map[string]interface{}, error) {
 	var baseFee *uint256.Int
 	if payload.BaseFeePerGas != nil {
 		var overflow bool
@@ -128,7 +133,7 @@ func (e *EngineImpl) ExecutePayloadV1(ctx context.Context, payload *ExecutionPay
 	for i, transaction := range payload.Transactions {
 		transactions[i] = ([]byte)(transaction)
 	}
-	res, err := e.api.EngineExecutePayloadV1(ctx, &types2.ExecutionPayload{
+	res, err := e.api.EngineNewPayloadV1(ctx, &types2.ExecutionPayload{
 		ParentHash:    gointerfaces.ConvertHashToH256(payload.ParentHash),
 		Coinbase:      gointerfaces.ConvertAddressToH160(payload.FeeRecipient),
 		StateRoot:     gointerfaces.ConvertHashToH256(payload.StateRoot),
@@ -148,16 +153,7 @@ func (e *EngineImpl) ExecutePayloadV1(ctx context.Context, payload *ExecutionPay
 		return nil, err
 	}
 
-	if res.LatestValidHash != nil {
-		var latestValidHash common.Hash = gointerfaces.ConvertH256ToHash(res.LatestValidHash)
-		return map[string]interface{}{
-			"status":          res.Status,
-			"latestValidHash": latestValidHash,
-		}, nil
-	}
-	return map[string]interface{}{
-		"status": res.Status,
-	}, nil
+	return convertPayloadStatus(res), nil
 }
 
 func (e *EngineImpl) GetPayloadV1(ctx context.Context, payloadID hexutil.Bytes) (*ExecutionPayload, error) {

cmd/rpcdaemon/rpcdaemontest/test_util.go

@@ -7,10 +7,9 @@ import (
 	"net"
 	"testing"
 
-	"github.com/ledgerwatch/erigon-lib/gointerfaces/starknet"
-
 	"github.com/holiman/uint256"
 	"github.com/ledgerwatch/erigon-lib/gointerfaces/remote"
+	"github.com/ledgerwatch/erigon-lib/gointerfaces/starknet"
 	"github.com/ledgerwatch/erigon-lib/gointerfaces/txpool"
 	"github.com/ledgerwatch/erigon-lib/kv"
 	"github.com/ledgerwatch/erigon/accounts/abi/bind"
@@ -223,7 +222,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, nil, nil, false))
+	remote.RegisterETHBACKENDServer(server, privateapi.NewEthBackendServer(ctx, nil, m.DB, m.Notifications.Events, snapshotsync.NewBlockReader(), nil, nil, nil, nil, nil, nil, nil, false))
 	txpool.RegisterTxpoolServer(server, m.TxPoolGrpcServer)
 	txpool.RegisterMiningServer(server, privateapi.NewMiningServer(ctx, &IsMiningMock{}, ethashApi))
 	starknet.RegisterCAIROVMServer(server, &starknet.UnimplementedCAIROVMServer{})

cmd/rpcdaemon/services/eth_backend.go

@@ -33,8 +33,8 @@ type ApiBackend interface {
 	ClientVersion(ctx context.Context) (string, error)
 	Subscribe(ctx context.Context, cb func(*remote.SubscribeReply)) error
 	BlockWithSenders(ctx context.Context, tx kv.Getter, 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)
+	EngineNewPayloadV1(ctx context.Context, payload *types2.ExecutionPayload) (*remote.EnginePayloadStatus, error)
+	EngineForkchoiceUpdatedV1(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,11 +164,11 @@ func (back *RemoteBackend) BlockWithSenders(ctx context.Context, tx kv.Getter, h
 	return back.blockReader.BlockWithSenders(ctx, tx, hash, blockHeight)
 }
 
-func (back *RemoteBackend) EngineExecutePayloadV1(ctx context.Context, payload *types2.ExecutionPayload) (res *remote.EngineExecutePayloadReply, err error) {
-	return back.remoteEthBackend.EngineExecutePayloadV1(ctx, payload)
+func (back *RemoteBackend) EngineNewPayloadV1(ctx context.Context, payload *types2.ExecutionPayload) (res *remote.EnginePayloadStatus, err error) {
+	return back.remoteEthBackend.EngineNewPayloadV1(ctx, payload)
 }
 
-func (back *RemoteBackend) EngineForkchoiceUpdateV1(ctx context.Context, request *remote.EngineForkChoiceUpdatedRequest) (*remote.EngineForkChoiceUpdatedReply, error) {
+func (back *RemoteBackend) EngineForkchoiceUpdatedV1(ctx context.Context, request *remote.EngineForkChoiceUpdatedRequest) (*remote.EngineForkChoiceUpdatedReply, error) {
 	return back.remoteEthBackend.EngineForkChoiceUpdatedV1(ctx, request)
 }
 

eth/backend.go

@@ -136,8 +136,9 @@ type Ethereum struct {
 	txPool2GrpcServer       *txpool2.GrpcServer
 	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 privateapi.PayloadMessage
+	// to proof-of-stake so we start reverse syncing from the block
+	newPayloadCh          chan privateapi.PayloadMessage
+	forkChoiceCh          chan privateapi.ForkChoiceMessage
 	waitingForBeaconChain uint32 // atomic boolean flag
 
 	downloadProtocols *downloader.Protocols
@@ -372,7 +373,8 @@ func New(stack *node.Node, config *ethconfig.Config, txpoolCfg txpool2.Config, l
 	miner := stagedsync.NewMiningState(&config.Miner)
 	backend.pendingBlocks = miner.PendingResultCh
 	backend.minedBlocks = miner.MiningResultCh
-	backend.reverseDownloadCh = make(chan privateapi.PayloadMessage)
+	backend.newPayloadCh = make(chan privateapi.PayloadMessage)
+	backend.forkChoiceCh = make(chan privateapi.ForkChoiceMessage)
 
 	// proof-of-work mining
 	mining := stagedsync.New(
@@ -413,8 +415,8 @@ func New(stack *node.Node, config *ethconfig.Config, txpoolCfg txpool2.Config, l
 	atomic.StoreUint32(&backend.waitingForBeaconChain, 0)
 	// Initialize ethbackend
 	ethBackendRPC := privateapi.NewEthBackendServer(ctx, backend, backend.chainDB, backend.notifications.Events,
-		blockReader, chainConfig, backend.reverseDownloadCh, backend.sentryControlServer.Hd.ExecutionStatusCh, &backend.waitingForBeaconChain,
-		backend.sentryControlServer.Hd.SkipCycleHack, assembleBlockPOS, config.Miner.EnabledPOS)
+		blockReader, chainConfig, backend.newPayloadCh, backend.forkChoiceCh, backend.sentryControlServer.Hd.PayloadStatusCh,
+		&backend.waitingForBeaconChain, backend.sentryControlServer.Hd.SkipCycleHack, assembleBlockPOS, config.Miner.EnabledPOS)
 	miningRPC = privateapi.NewMiningServer(ctx, backend, ethashApi)
 	// If we enabled the proposer flag we initiates the block proposing thread
 	if config.Miner.EnabledPOS && chainConfig.TerminalTotalDifficulty != nil {
@@ -490,7 +492,7 @@ func New(stack *node.Node, config *ethconfig.Config, txpoolCfg txpool2.Config, l
 	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.waitingForBeaconChain,
+		backend.newPayloadCh, backend.forkChoiceCh, &backend.waitingForBeaconChain,
 		backend.downloaderClient)
 	if err != nil {
 		return nil, err