From 84dd61932aea22818f6bba63fe5832393b25e2cc Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Thu, 14 Mar 2024 22:49:26 +0100 Subject: [PATCH] beacon/blsync: new engine API client --- beacon/blsync/block_sync.go | 17 ++-- beacon/blsync/block_sync_test.go | 6 +- beacon/blsync/client.go | 39 +++++---- beacon/blsync/engineapi.go | 132 +++++++++++++++++++++++++++++++ beacon/types/config.go | 6 +- 5 files changed, 171 insertions(+), 29 deletions(-) create mode 100644 beacon/blsync/engineapi.go diff --git a/beacon/blsync/block_sync.go b/beacon/blsync/block_sync.go index c15eb3f0cb..62f581b311 100755 --- a/beacon/blsync/block_sync.go +++ b/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 { diff --git a/beacon/blsync/block_sync_test.go b/beacon/blsync/block_sync_test.go index 554f39999e..6db4f0652a 100644 --- a/beacon/blsync/block_sync_test.go +++ b/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) diff --git a/beacon/blsync/client.go b/beacon/blsync/client.go index 1bfbb13160..c81f18b85e 100644 --- a/beacon/blsync/client.go +++ b/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() } diff --git a/beacon/blsync/engineapi.go b/beacon/blsync/engineapi.go new file mode 100644 index 0000000000..10044ab6fb --- /dev/null +++ b/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 +} diff --git a/beacon/types/config.go b/beacon/types/config.go index eb3f061c7d..a52da5212e 100644 --- a/beacon/types/config.go +++ b/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.