beacon/blsync: new engine API client

beacon/blsync/block_sync.go

@@ -35,7 +35,7 @@ type beaconBlockSync struct {
 	headTracker  headTracker
 
 	lastHeadInfo  types.HeadInfo
-	chainHeadFeed *event.Feed
+	chainHeadFeed event.FeedOf[types.ChainHeadEvent]
 }
 
 type headTracker interface {
@@ -45,16 +45,19 @@ type headTracker interface {
 }
 
 // newBeaconBlockSync returns a new beaconBlockSync.
-func newBeaconBlockSync(headTracker headTracker, chainHeadFeed *event.Feed) *beaconBlockSync {
+func newBeaconBlockSync(headTracker headTracker) *beaconBlockSync {
 	return &beaconBlockSync{
-		headTracker:   headTracker,
-		chainHeadFeed: chainHeadFeed,
-		recentBlocks:  lru.NewCache[common.Hash, *types.BeaconBlock](10),
-		locked:        make(map[common.Hash]request.ServerAndID),
-		serverHeads:   make(map[request.Server]common.Hash),
+		headTracker:  headTracker,
+		recentBlocks: lru.NewCache[common.Hash, *types.BeaconBlock](10),
+		locked:       make(map[common.Hash]request.ServerAndID),
+		serverHeads:  make(map[request.Server]common.Hash),
 	}
 }
 
+func (s *beaconBlockSync) SubscribeChainHead(ch chan<- types.ChainHeadEvent) event.Subscription {
+	return s.chainHeadFeed.Subscribe(ch)
+}
+
 // Process implements request.Module.
 func (s *beaconBlockSync) Process(requester request.Requester, events []request.Event) {
 	for _, event := range events {

beacon/blsync/block_sync_test.go

@@ -22,7 +22,6 @@ import (
 	"github.com/ethereum/go-ethereum/beacon/light/request"
 	"github.com/ethereum/go-ethereum/beacon/light/sync"
 	"github.com/ethereum/go-ethereum/beacon/types"
-	"github.com/ethereum/go-ethereum/event"
 	"github.com/protolambda/zrnt/eth2/beacon/deneb"
 )
 
@@ -46,10 +45,9 @@ var (
 
 func TestBlockSync(t *testing.T) {
 	ht := &testHeadTracker{}
-	eventFeed := new(event.Feed)
-	blockSync := newBeaconBlockSync(ht, eventFeed)
+	blockSync := newBeaconBlockSync(ht)
 	headCh := make(chan types.ChainHeadEvent, 16)
-	eventFeed.Subscribe(headCh)
+	blockSync.SubscribeChainHead(headCh)
 	ts := sync.NewTestScheduler(t, blockSync)
 	ts.AddServer(testServer1, 1)
 	ts.AddServer(testServer2, 1)

beacon/blsync/client.go

@@ -28,14 +28,20 @@ import (
 	"github.com/ethereum/go-ethereum/common/mclock"
 	"github.com/ethereum/go-ethereum/ethdb/memorydb"
 	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/urfave/cli/v2"
 )
 
 type Client struct {
-	scheduler     *request.Scheduler
-	chainHeadFeed *event.Feed
-	urls          []string
-	customHeader  map[string]string
+	urls         []string
+	customHeader map[string]string
+	chainConfig  *lightClientConfig
+	scheduler    *request.Scheduler
+	blockSync    *beaconBlockSync
+	engineRPC    *rpc.Client
+
+	chainHeadSub event.Subscription
+	engineClient *engineClient
 }
 
 func NewClient(ctx *cli.Context) *Client {
@@ -53,6 +59,7 @@ func NewClient(ctx *cli.Context) *Client {
 		}
 		customHeader[strings.TrimSpace(kv[0])] = strings.TrimSpace(kv[1])
 	}
+
 	// create data structures
 	var (
 		db             = memorydb.New()
@@ -63,11 +70,10 @@ func NewClient(ctx *cli.Context) *Client {
 	headSync := sync.NewHeadSync(headTracker, committeeChain)
 
 	// set up scheduler and sync modules
-	chainHeadFeed := new(event.Feed)
 	scheduler := request.NewScheduler()
 	checkpointInit := sync.NewCheckpointInit(committeeChain, chainConfig.Checkpoint)
 	forwardSync := sync.NewForwardUpdateSync(committeeChain)
-	beaconBlockSync := newBeaconBlockSync(headTracker, chainHeadFeed)
+	beaconBlockSync := newBeaconBlockSync(headTracker)
 	scheduler.RegisterTarget(headTracker)
 	scheduler.RegisterTarget(committeeChain)
 	scheduler.RegisterModule(checkpointInit, "checkpointInit")
@@ -76,22 +82,23 @@ func NewClient(ctx *cli.Context) *Client {
 	scheduler.RegisterModule(beaconBlockSync, "beaconBlockSync")
 
 	return &Client{
-		scheduler:     scheduler,
-		urls:          ctx.StringSlice(utils.BeaconApiFlag.Name),
-		customHeader:  customHeader,
-		chainHeadFeed: chainHeadFeed,
+		scheduler:    scheduler,
+		urls:         ctx.StringSlice(utils.BeaconApiFlag.Name),
+		customHeader: customHeader,
+		blockSync:    beaconBlockSync,
 	}
 }
 
-// SubscribeChainHeadEvent allows callers to subscribe a provided channel to new
-// head updates.
-func (c *Client) SubscribeChainHeadEvent(ch chan<- types.ChainHeadEvent) event.Subscription {
-	return c.chainHeadFeed.Subscribe(ch)
+func (c *Client) SetEngineRPC(engine *rpc.Client) {
+	c.engineRPC = engine
 }
 
 func (c *Client) Start() {
+	headCh := make(chan types.ChainHeadEvent, 16)
+	c.blockSync.SubscribeChainHead(headCh)
+	c.engineClient = startEngineClient(c.chainConfig, c.engineRPC, headCh)
+
 	c.scheduler.Start()
-	// register server(s)
 	for _, url := range c.urls {
 		beaconApi := api.NewBeaconLightApi(url, c.customHeader)
 		c.scheduler.RegisterServer(request.NewServer(api.NewApiServer(beaconApi), &mclock.System{}))
@@ -99,5 +106,7 @@ func (c *Client) Start() {
 }
 
 func (c *Client) Stop() {
+	c.engineClient.stop()
+	c.chainHeadSub.Unsubscribe()
 	c.scheduler.Stop()
 }

beacon/blsync/engineapi.go

@@ -0,0 +1,132 @@
+package blsync
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/beacon/engine"
+	"github.com/ethereum/go-ethereum/beacon/types"
+	"github.com/ethereum/go-ethereum/common"
+	ctypes "github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+type engineClient struct {
+	config     *lightClientConfig
+	rpc        *rpc.Client
+	rootCtx    context.Context
+	cancelRoot context.CancelFunc
+	wg         sync.WaitGroup
+}
+
+func startEngineClient(config *lightClientConfig, rpc *rpc.Client, headCh <-chan types.ChainHeadEvent) *engineClient {
+	ctx, cancel := context.WithCancel(context.Background())
+	ec := &engineClient{
+		config:     config,
+		rpc:        rpc,
+		rootCtx:    ctx,
+		cancelRoot: cancel,
+	}
+	ec.wg.Add(1)
+	go ec.updateLoop(headCh)
+	return ec
+}
+
+func (ec *engineClient) stop() {
+	ec.cancelRoot()
+	ec.wg.Wait()
+}
+
+func (ec *engineClient) updateLoop(headCh <-chan types.ChainHeadEvent) {
+	defer ec.wg.Done()
+
+	for {
+		select {
+		case <-ec.rootCtx.Done():
+			return
+
+		case event := <-headCh:
+			if ec.rpc == nil { // dry run, no engine API specified
+				log.Info("New execution block retrieved", "number", event.Block.NumberU64(), "hash", event.Block.Hash(), "finalized", event.Finalized)
+				continue
+			}
+
+			fork := ec.config.ForkAtEpoch(event.BeaconHead.Epoch())
+			forkName := strings.ToLower(fork.Name)
+
+			if status, err := ec.callNewPayload(forkName, event); err == nil {
+				log.Info("Successful NewPayload", "number", event.Block.NumberU64(), "hash", event.Block.Hash(), "status", status)
+			} else {
+				log.Error("Failed NewPayload", "number", event.Block.NumberU64(), "hash", event.Block.Hash(), "error", err)
+			}
+
+			if status, err := ec.callForkchoiceUpdated(forkName, event); err == nil {
+				log.Info("Successful ForkchoiceUpdated", "head", event.Block.Hash(), "status", status)
+			} else {
+				log.Error("Failed ForkchoiceUpdated", "head", event.Block.Hash(), "error", err)
+			}
+		}
+	}
+}
+
+func (ec *engineClient) callNewPayload(fork string, event types.ChainHeadEvent) (string, error) {
+	execData := engine.BlockToExecutableData(event.Block, nil, nil).ExecutionPayload
+
+	var (
+		method string
+		params = []any{execData}
+	)
+	switch fork {
+	case "deneb":
+		method = "engine_newPayloadV3"
+		parentBeaconRoot := event.BeaconHead.ParentRoot
+		blobHashes := collectBlobHashes(event.Block)
+		params = append(params, blobHashes, parentBeaconRoot)
+	case "capella":
+		method = "engine_newPayloadV2"
+	default:
+		method = "engine_newPayloadV1"
+	}
+
+	ctx, cancel := context.WithTimeout(ec.rootCtx, time.Second*5)
+	defer cancel()
+	var resp engine.PayloadStatusV1
+	err := ec.rpc.CallContext(ctx, &resp, method, params...)
+	return resp.Status, err
+}
+
+func collectBlobHashes(b *ctypes.Block) (list []common.Hash) {
+	for _, tx := range b.Transactions() {
+		for _, h := range tx.BlobHashes() {
+			list = append(list, h)
+		}
+	}
+	return list
+}
+
+func (ec *engineClient) callForkchoiceUpdated(fork string, event types.ChainHeadEvent) (string, error) {
+	update := engine.ForkchoiceStateV1{
+		HeadBlockHash:      event.Block.Hash(),
+		SafeBlockHash:      event.Finalized,
+		FinalizedBlockHash: event.Finalized,
+	}
+
+	var method string
+	switch fork {
+	case "deneb":
+		method = "engine_forkchoiceUpdatedV3"
+	case "capella":
+		method = "engine_forkchoiceUpdatedV2"
+	default:
+		method = "engine_forkchoiceUpdatedV1"
+	}
+
+	ctx, cancel := context.WithTimeout(ec.rootCtx, time.Second*5)
+	defer cancel()
+	var resp engine.ForkChoiceResponse
+	err := ec.rpc.CallContext(ctx, &resp, method, update, nil)
+	return resp.PayloadStatus.Status, err
+}

beacon/types/config.go

@@ -111,13 +111,13 @@ type ChainConfig struct {
 }
 
 // ForkAtEpoch returns the latest active fork at the given epoch.
-func (c *ChainConfig) ForkAtEpoch(epoch uint64) *Fork {
+func (c *ChainConfig) ForkAtEpoch(epoch uint64) Fork {
 	for i := len(c.Forks) - 1; i >= 0; i-- {
 		if c.Forks[i].Epoch <= epoch {
-			return c.Forks[i]
+			return *c.Forks[i]
 		}
 	}
-	return nil
+	return Fork{}
 }
 
 // AddFork adds a new item to the list of forks.