forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

cmd/geth: blsync integrated into geth

Browse source
This commit is contained in:
Zsolt Felfoldi 2024-02-01 06:08:58 +01:00 committed by Felix Lange
parent 8884782e4e
commit 130732e399
10 changed files with 369 additions and 204 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

145
cmd/blsync/block_sync.go → beacon/blsync/block_sync.go
View file

@ -14,15 +14,26 @@
// 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 main
package blsync
import (
"fmt"
"math/big"
"github.com/ethereum/go-ethereum/beacon/engine"
"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/common"
"github.com/ethereum/go-ethereum/common/lru"
ctypes "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/trie"
"github.com/holiman/uint256"
"github.com/protolambda/zrnt/eth2/beacon/capella"
"github.com/protolambda/zrnt/eth2/configs"
"github.com/protolambda/ztyp/tree"
)
// beaconBlockSync implements request.Module; it fetches the beacon blocks belonging
@ -33,13 +44,8 @@ type beaconBlockSync struct {
serverHeads map[request.Server]common.Hash
headTracker headTracker
lastHeadInfo types.HeadInfo
headCh chan headData
}
type headData struct {
block *capella.BeaconBlock
update types.FinalityUpdate
lastHeadInfo types.HeadInfo
chainHeadFeed *event.Feed
}
type headTracker interface {
@ -49,13 +55,13 @@ type headTracker interface {
}
// newBeaconBlockSync returns a new beaconBlockSync.
func newBeaconBlockSync(headTracker headTracker) *beaconBlockSync {
func newBeaconBlockSync(headTracker headTracker, chainHeadFeed *event.Feed) *beaconBlockSync {
return &beaconBlockSync{
headTracker: headTracker,
recentBlocks: lru.NewCache[common.Hash, *capella.BeaconBlock](10),
locked: make(map[common.Hash]request.ServerAndID),
serverHeads: make(map[request.Server]common.Hash),
headCh: make(chan headData, 1),
headTracker: headTracker,
chainHeadFeed: chainHeadFeed,
recentBlocks: lru.NewCache[common.Hash, *capella.BeaconBlock](10),
locked: make(map[common.Hash]request.ServerAndID),
serverHeads: make(map[request.Server]common.Hash),
}
}
@ -78,7 +84,7 @@ func (s *beaconBlockSync) Process(requester request.Requester, events []request.
delete(s.serverHeads, event.Server)
}
}
s.updateValidatedHead()
s.updateEventFeed()
// request validated head block if unavailable and not yet requested
if vh, ok := s.headTracker.ValidatedHead(); ok {
s.tryRequestBlock(requester, vh.Header.Hash(), false)
@ -89,31 +95,6 @@ func (s *beaconBlockSync) Process(requester request.Requester, events []request.
}
}
func (s *beaconBlockSync) updateValidatedHead() {
head, ok := s.headTracker.ValidatedHead()
if !ok {
return
}
finality, ok := s.headTracker.ValidatedFinality() //TODO fetch directly if subscription does not deliver
if !ok || head.Header.Epoch() != finality.Attested.Header.Epoch() {
return
}
validatedHead := head.Header.Hash()
headBlock, ok := s.recentBlocks.Get(validatedHead)
if !ok {
return
}
headInfo := blockHeadInfo(headBlock)
if headInfo == s.lastHeadInfo {
return
}
select {
case s.headCh <- headData{block: headBlock, update: finality}:
s.lastHeadInfo = headInfo
default:
}
}
func (s *beaconBlockSync) tryRequestBlock(requester request.Requester, blockRoot common.Hash, needSameHead bool) {
if _, ok := s.recentBlocks.Get(blockRoot); ok {
return
@ -130,9 +111,93 @@ func (s *beaconBlockSync) tryRequestBlock(requester request.Requester, blockRoot
return
}
}
func blockHeadInfo(block *capella.BeaconBlock) types.HeadInfo {
if block == nil {
return types.HeadInfo{}
}
return types.HeadInfo{Slot: uint64(block.Slot), BlockRoot: beaconBlockHash(block)}
}
// beaconBlockHash calculates the hash of a beacon block.
func beaconBlockHash(beaconBlock *capella.BeaconBlock) common.Hash {
return common.Hash(beaconBlock.HashTreeRoot(configs.Mainnet, tree.GetHashFn()))
}
// getExecBlock extracts the execution block from the beacon block's payload.
func getExecBlock(beaconBlock *capella.BeaconBlock) (*ctypes.Block, error) {
payload := &beaconBlock.Body.ExecutionPayload
txs := make([]*ctypes.Transaction, len(payload.Transactions))
for i, opaqueTx := range payload.Transactions {
var tx ctypes.Transaction
if err := tx.UnmarshalBinary(opaqueTx); err != nil {
return nil, fmt.Errorf("failed to parse tx %d: %v", i, err)
}
txs[i] = &tx
}
withdrawals := make([]*ctypes.Withdrawal, len(payload.Withdrawals))
for i, w := range payload.Withdrawals {
withdrawals[i] = &ctypes.Withdrawal{
Index: uint64(w.Index),
Validator: uint64(w.ValidatorIndex),
Address: common.Address(w.Address),
Amount: uint64(w.Amount),
}
}
wroot := ctypes.DeriveSha(ctypes.Withdrawals(withdrawals), trie.NewStackTrie(nil))
execHeader := &ctypes.Header{
ParentHash: common.Hash(payload.ParentHash),
UncleHash: ctypes.EmptyUncleHash,
Coinbase: common.Address(payload.FeeRecipient),
Root: common.Hash(payload.StateRoot),
TxHash: ctypes.DeriveSha(ctypes.Transactions(txs), trie.NewStackTrie(nil)),
ReceiptHash: common.Hash(payload.ReceiptsRoot),
Bloom: ctypes.Bloom(payload.LogsBloom),
Difficulty: common.Big0,
Number: new(big.Int).SetUint64(uint64(payload.BlockNumber)),
GasLimit: uint64(payload.GasLimit),
GasUsed: uint64(payload.GasUsed),
Time: uint64(payload.Timestamp),
Extra: []byte(payload.ExtraData),
MixDigest: common.Hash(payload.PrevRandao), // reused in merge
Nonce: ctypes.BlockNonce{}, // zero
BaseFee: (*uint256.Int)(&payload.BaseFeePerGas).ToBig(),
WithdrawalsHash: &wroot,
}
execBlock := ctypes.NewBlockWithHeader(execHeader).WithBody(txs, nil).WithWithdrawals(withdrawals)
if execBlockHash := execBlock.Hash(); execBlockHash != common.Hash(payload.BlockHash) {
return nil, fmt.Errorf("Sanity check failed, payload hash does not match (expected %x, got %x)", common.Hash(payload.BlockHash), execBlockHash)
}
return execBlock, nil
}
func (s *beaconBlockSync) updateEventFeed() {
head, ok := s.headTracker.ValidatedHead()
if !ok {
return
}
finality, ok := s.headTracker.ValidatedFinality() //TODO fetch directly if subscription does not deliver
if !ok || head.Header.Epoch() != finality.Attested.Header.Epoch() {
return
}
validatedHead := head.Header.Hash()
headBlock, ok := s.recentBlocks.Get(validatedHead)
if !ok {
return
}
headInfo := blockHeadInfo(headBlock)
if headInfo == s.lastHeadInfo {
return
}
s.lastHeadInfo = headInfo
// new head block and finality info available; extract executable data and send event to feed
execBlock, err := getExecBlock(headBlock)
if err != nil {
log.Error("Error extracting execution block from validated beacon block", "error", err)
return
}
s.chainHeadFeed.Send(types.ChainHeadEvent{
HeadBlock: engine.BlockToExecutableData(execBlock, nil, nil).ExecutionPayload,
Finalized: common.Hash(finality.Finalized.PayloadHeader.BlockHash),
})
}

2
cmd/blsync/block_sync_test.go → beacon/blsync/block_sync_test.go
View file

@ -14,7 +14,7 @@
// 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 main
package blsync
import (
"testing"

103
beacon/blsync/client.go Normal file
View file

@ -0,0 +1,103 @@
// Copyright 2024 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 blsync
import (
"strings"
"github.com/ethereum/go-ethereum/beacon/light"
"github.com/ethereum/go-ethereum/beacon/light/api"
"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/cmd/utils"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/ethdb/memorydb"
"github.com/ethereum/go-ethereum/event"
"github.com/urfave/cli/v2"
)
type Client struct {
scheduler *request.Scheduler
chainHeadFeed *event.Feed
urls []string
customHeader map[string]string
}
func NewClient(ctx *cli.Context) *Client {
if !ctx.IsSet(utils.BeaconApiFlag.Name) {
utils.Fatalf("Beacon node light client API URL not specified")
}
var (
chainConfig = makeChainConfig(ctx)
customHeader = make(map[string]string)
)
for _, s := range ctx.StringSlice(utils.BeaconApiHeaderFlag.Name) {
kv := strings.Split(s, ":")
if len(kv) != 2 {
utils.Fatalf("Invalid custom API header entry: %s", s)
}
customHeader[strings.TrimSpace(kv[0])] = strings.TrimSpace(kv[1])
}
// create data structures
var (
db = memorydb.New()
threshold = ctx.Int(utils.BeaconThresholdFlag.Name)
committeeChain = light.NewCommitteeChain(db, chainConfig.ChainConfig, threshold, !ctx.Bool(utils.BeaconNoFilterFlag.Name))
headTracker = light.NewHeadTracker(committeeChain, threshold)
)
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)
scheduler.RegisterTarget(headTracker)
scheduler.RegisterTarget(committeeChain)
scheduler.RegisterModule(checkpointInit, "checkpointInit")
scheduler.RegisterModule(forwardSync, "forwardSync")
scheduler.RegisterModule(headSync, "headSync")
scheduler.RegisterModule(beaconBlockSync, "beaconBlockSync")
return &Client{
scheduler: scheduler,
urls: ctx.StringSlice(utils.BeaconApiFlag.Name),
customHeader: customHeader,
chainHeadFeed: chainHeadFeed,
}
}
// 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) Start() {
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{}))
}
}
func (c *Client) Stop() {
c.scheduler.Stop()
}

31
cmd/blsync/config.go → beacon/blsync/config.go
View file

@ -14,18 +14,13 @@
// 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 main
package blsync
import (
"context"
"github.com/ethereum/go-ethereum/beacon/types"
"github.com/ethereum/go-ethereum/cmd/utils"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/rpc"
"github.com/urfave/cli/v2"
)
@ -116,27 +111,3 @@ func makeChainConfig(ctx *cli.Context) lightClientConfig {
}
return config
}
func makeRPCClient(ctx *cli.Context) *rpc.Client {
if !ctx.IsSet(utils.BlsyncApiFlag.Name) {
log.Warn("No engine API target specified, performing a dry run")
return nil
}
if !ctx.IsSet(utils.BlsyncJWTSecretFlag.Name) {
utils.Fatalf("JWT secret parameter missing") //TODO use default if datadir is specified
}
engineApiUrl, jwtFileName := ctx.String(utils.BlsyncApiFlag.Name), ctx.String(utils.BlsyncJWTSecretFlag.Name)
var jwtSecret [32]byte
if jwt, err := node.ObtainJWTSecret(jwtFileName); err == nil {
copy(jwtSecret[:], jwt)
} else {
utils.Fatalf("Error loading or generating JWT secret: %v", err)
}
auth := node.NewJWTAuth(jwtSecret)
cl, err := rpc.DialOptions(context.Background(), engineApiUrl, rpc.WithHTTPAuth(auth))
if err != nil {
utils.Fatalf("Could not create RPC client: %v", err)
}
return cl
}

9
beacon/types/light_sync.go
View file

@ -20,6 +20,7 @@ import (
"errors"
"fmt"
"github.com/ethereum/go-ethereum/beacon/engine"
"github.com/ethereum/go-ethereum/beacon/merkle"
"github.com/ethereum/go-ethereum/beacon/params"
"github.com/ethereum/go-ethereum/common"
@ -181,3 +182,11 @@ func (u *FinalityUpdate) Validate() error {
}
return merkle.VerifyProof(u.Attested.StateRoot, params.StateIndexFinalBlock, u.FinalityBranch, merkle.Value(u.Finalized.Hash()))
}
// ChainHeadEvent returns an authenticated execution payload associated with the
// latest accepted head of the beacon chain, along with the hash of the latest
// finalized execution block.
type ChainHeadEvent struct {
HeadBlock *engine.ExecutableData
Finalized common.Hash
}

90
cmd/blsync/engine_api.go
View file

@ -18,105 +18,39 @@ package main
import (
"context"
"fmt"
"math/big"
"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"
"github.com/ethereum/go-ethereum/trie"
"github.com/holiman/uint256"
"github.com/protolambda/zrnt/eth2/beacon/capella"
"github.com/protolambda/zrnt/eth2/configs"
"github.com/protolambda/ztyp/tree"
)
func updateEngineApi(client *rpc.Client, headCh chan headData) {
for headData := range headCh {
execBlock, err := getExecBlock(headData.block)
if err != nil {
log.Error("Error extracting execution block from validated beacon block", "error", err)
continue
}
execRoot := execBlock.Hash()
finalizedRoot := common.Hash(headData.update.Finalized.PayloadHeader.BlockHash)
func updateEngineApi(client *rpc.Client, headCh chan types.ChainHeadEvent) {
for event := range headCh {
if client == nil { // dry run, no engine API specified
log.Info("New execution block retrieved", "block number", execBlock.NumberU64(), "block hash", execRoot, "finalized block hash", finalizedRoot)
log.Info("New execution block retrieved", "block number", event.HeadBlock.Number, "block hash", event.HeadBlock.BlockHash, "finalized block hash", event.Finalized)
} else {
if status, err := callNewPayloadV2(client, execBlock); err == nil {
log.Info("Successful NewPayload", "block number", execBlock.NumberU64(), "block hash", execRoot, "status", status)
if status, err := callNewPayloadV2(client, event.HeadBlock); err == nil {
log.Info("Successful NewPayload", "block number", event.HeadBlock.Number, "block hash", event.HeadBlock.BlockHash, "status", status)
} else {
log.Error("Failed NewPayload", "block number", execBlock.NumberU64(), "block hash", execRoot, "error", err)
log.Error("Failed NewPayload", "block number", event.HeadBlock.Number, "block hash", event.HeadBlock.BlockHash, "error", err)
}
if status, err := callForkchoiceUpdatedV1(client, execRoot, finalizedRoot); err == nil {
log.Info("Successful ForkchoiceUpdated", "head", execRoot, "status", status)
if status, err := callForkchoiceUpdatedV1(client, event.HeadBlock.BlockHash, event.Finalized); err == nil {
log.Info("Successful ForkchoiceUpdated", "head", event.HeadBlock.BlockHash, "status", status)
} else {
log.Error("Failed ForkchoiceUpdated", "head", execRoot, "error", err)
log.Error("Failed ForkchoiceUpdated", "head", event.HeadBlock.BlockHash, "error", err)
}
}
}
}
// getExecBlock extracts the execution block from the beacon block's payload.
func getExecBlock(beaconBlock *capella.BeaconBlock) (*ctypes.Block, error) {
payload := &beaconBlock.Body.ExecutionPayload
txs := make([]*ctypes.Transaction, len(payload.Transactions))
for i, opaqueTx := range payload.Transactions {
var tx ctypes.Transaction
if err := tx.UnmarshalBinary(opaqueTx); err != nil {
return nil, fmt.Errorf("failed to parse tx %d: %v", i, err)
}
txs[i] = &tx
}
withdrawals := make([]*ctypes.Withdrawal, len(payload.Withdrawals))
for i, w := range payload.Withdrawals {
withdrawals[i] = &ctypes.Withdrawal{
Index: uint64(w.Index),
Validator: uint64(w.ValidatorIndex),
Address: common.Address(w.Address),
Amount: uint64(w.Amount),
}
}
wroot := ctypes.DeriveSha(ctypes.Withdrawals(withdrawals), trie.NewStackTrie(nil))
execHeader := &ctypes.Header{
ParentHash: common.Hash(payload.ParentHash),
UncleHash: ctypes.EmptyUncleHash,
Coinbase: common.Address(payload.FeeRecipient),
Root: common.Hash(payload.StateRoot),
TxHash: ctypes.DeriveSha(ctypes.Transactions(txs), trie.NewStackTrie(nil)),
ReceiptHash: common.Hash(payload.ReceiptsRoot),
Bloom: ctypes.Bloom(payload.LogsBloom),
Difficulty: common.Big0,
Number: new(big.Int).SetUint64(uint64(payload.BlockNumber)),
GasLimit: uint64(payload.GasLimit),
GasUsed: uint64(payload.GasUsed),
Time: uint64(payload.Timestamp),
Extra: []byte(payload.ExtraData),
MixDigest: common.Hash(payload.PrevRandao), // reused in merge
Nonce: ctypes.BlockNonce{}, // zero
BaseFee: (*uint256.Int)(&payload.BaseFeePerGas).ToBig(),
WithdrawalsHash: &wroot,
}
execBlock := ctypes.NewBlockWithHeader(execHeader).WithBody(txs, nil).WithWithdrawals(withdrawals)
if execBlockHash := execBlock.Hash(); execBlockHash != common.Hash(payload.BlockHash) {
return nil, fmt.Errorf("Sanity check failed, payload hash does not match (expected %x, got %x)", common.Hash(payload.BlockHash), execBlockHash)
}
return execBlock, nil
}
// beaconBlockHash calculates the hash of a beacon block.
func beaconBlockHash(beaconBlock *capella.BeaconBlock) common.Hash {
return common.Hash(beaconBlock.HashTreeRoot(configs.Mainnet, tree.GetHashFn()))
}
func callNewPayloadV2(client *rpc.Client, block *ctypes.Block) (string, error) {
func callNewPayloadV2(client *rpc.Client, execData *engine.ExecutableData) (string, error) {
var resp engine.PayloadStatusV1
ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
err := client.CallContext(ctx, &resp, "engine_newPayloadV2", *engine.BlockToExecutableData(block, nil, nil).ExecutionPayload)
err := client.CallContext(ctx, &resp, "engine_newPayloadV2", execData)
cancel()
return resp.Status, err
}

94
cmd/blsync/main.go
View file

@ -17,20 +17,18 @@
package main
import (
"context"
"fmt"
"io"
"os"
"strings"
"github.com/ethereum/go-ethereum/beacon/light"
"github.com/ethereum/go-ethereum/beacon/light/api"
"github.com/ethereum/go-ethereum/beacon/light/request"
"github.com/ethereum/go-ethereum/beacon/light/sync"
"github.com/ethereum/go-ethereum/beacon/blsync"
"github.com/ethereum/go-ethereum/beacon/types"
"github.com/ethereum/go-ethereum/cmd/utils"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/ethdb/memorydb"
"github.com/ethereum/go-ethereum/internal/flags"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/rpc"
"github.com/mattn/go-colorable"
"github.com/mattn/go-isatty"
"github.com/urfave/cli/v2"
@ -72,7 +70,7 @@ func main() {
verbosityFlag,
vmoduleFlag,
}
app.Action = blsync
app.Action = sync
if err := app.Run(os.Args); err != nil {
fmt.Fprintln(os.Stderr, err)
@ -80,7 +78,7 @@ func main() {
}
}
func blsync(ctx *cli.Context) error {
func sync(ctx *cli.Context) error {
usecolor := (isatty.IsTerminal(os.Stderr.Fd()) || isatty.IsCygwinTerminal(os.Stderr.Fd())) && os.Getenv("TERM") != "dumb"
output := io.Writer(os.Stderr)
if usecolor {
@ -89,53 +87,39 @@ func blsync(ctx *cli.Context) error {
verbosity := log.FromLegacyLevel(ctx.Int(verbosityFlag.Name))
log.SetDefault(log.NewLogger(log.NewTerminalHandlerWithLevel(output, verbosity, usecolor)))
if !ctx.IsSet(utils.BeaconApiFlag.Name) {
utils.Fatalf("Beacon node light client API URL not specified")
}
var (
chainConfig = makeChainConfig(ctx)
customHeader = make(map[string]string)
)
for _, s := range ctx.StringSlice(utils.BeaconApiHeaderFlag.Name) {
kv := strings.Split(s, ":")
if len(kv) != 2 {
utils.Fatalf("Invalid custom API header entry: %s", s)
}
customHeader[strings.TrimSpace(kv[0])] = strings.TrimSpace(kv[1])
}
// create data structures
var (
db = memorydb.New()
threshold = ctx.Int(utils.BeaconThresholdFlag.Name)
committeeChain = light.NewCommitteeChain(db, chainConfig.ChainConfig, threshold, !ctx.Bool(utils.BeaconNoFilterFlag.Name))
headTracker = light.NewHeadTracker(committeeChain, threshold)
)
headSync := sync.NewHeadSync(headTracker, committeeChain)
// set up scheduler and sync modules
scheduler := request.NewScheduler()
checkpointInit := sync.NewCheckpointInit(committeeChain, chainConfig.Checkpoint)
forwardSync := sync.NewForwardUpdateSync(committeeChain)
beaconBlockSync := newBeaconBlockSync(headTracker)
scheduler.RegisterTarget(headTracker)
scheduler.RegisterTarget(committeeChain)
scheduler.RegisterModule(checkpointInit, "checkpointInit")
scheduler.RegisterModule(forwardSync, "forwardSync")
scheduler.RegisterModule(headSync, "headSync")
scheduler.RegisterModule(beaconBlockSync, "beaconBlockSync")
go updateEngineApi(makeRPCClient(ctx), beaconBlockSync.headCh)
// start
scheduler.Start()
// register server(s)
for _, url := range ctx.StringSlice(utils.BeaconApiFlag.Name) {
beaconApi := api.NewBeaconLightApi(url, customHeader)
scheduler.RegisterServer(request.NewServer(api.NewApiServer(beaconApi), &mclock.System{}))
}
headCh := make(chan types.ChainHeadEvent, 1)
client := blsync.NewClient(ctx)
sub := client.SubscribeChainHeadEvent(headCh)
go updateEngineApi(makeRPCClient(ctx), headCh)
client.Start()
// run until stopped
<-ctx.Done()
scheduler.Stop()
close(beaconBlockSync.headCh)
client.Stop()
sub.Unsubscribe()
close(headCh)
return nil
}
func makeRPCClient(ctx *cli.Context) *rpc.Client {
if !ctx.IsSet(utils.BlsyncApiFlag.Name) {
log.Warn("No engine API target specified, performing a dry run")
return nil
}
if !ctx.IsSet(utils.BlsyncJWTSecretFlag.Name) {
utils.Fatalf("JWT secret parameter missing") //TODO use default if datadir is specified
}
engineApiUrl, jwtFileName := ctx.String(utils.BlsyncApiFlag.Name), ctx.String(utils.BlsyncJWTSecretFlag.Name)
var jwtSecret [32]byte
if jwt, err := node.ObtainJWTSecret(jwtFileName); err == nil {
copy(jwtSecret[:], jwt)
} else {
utils.Fatalf("Error loading or generating JWT secret: %v", err)
}
auth := node.NewJWTAuth(jwtSecret)
cl, err := rpc.DialOptions(context.Background(), engineApiUrl, rpc.WithHTTPAuth(auth))
if err != nil {
utils.Fatalf("Could not create RPC client: %v", err)
}
return cl
}

3
cmd/geth/config.go
View file

@ -31,6 +31,7 @@ import (
"github.com/ethereum/go-ethereum/accounts/keystore"
"github.com/ethereum/go-ethereum/accounts/scwallet"
"github.com/ethereum/go-ethereum/accounts/usbwallet"
"github.com/ethereum/go-ethereum/beacon/blsync"
"github.com/ethereum/go-ethereum/cmd/utils"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
@ -221,6 +222,8 @@ func makeFullNode(ctx *cli.Context) (*node.Node, ethapi.Backend) {
}
catalyst.RegisterSimulatedBeaconAPIs(stack, simBeacon)
stack.RegisterLifecycle(simBeacon)
} else if ctx.IsSet(utils.BeaconApiFlag.Name) {
stack.RegisterLifecycle(catalyst.NewBlsync(blsync.NewClient(ctx), eth))
} else {
err := catalyst.Register(stack, eth)
if err != nil {

8
cmd/geth/main.go
View file

@ -146,6 +146,14 @@ var (
configFileFlag,
utils.LogDebugFlag,
utils.LogBacktraceAtFlag,
utils.BeaconApiFlag,
utils.BeaconApiHeaderFlag,
utils.BeaconThresholdFlag,
utils.BeaconNoFilterFlag,
utils.BeaconConfigFlag,
utils.BeaconGenesisRootFlag,
utils.BeaconGenesisTimeFlag,
utils.BeaconCheckpointFlag,
}, utils.NetworkFlags, utils.DatabaseFlags)
rpcFlags = []cli.Flag{

88
eth/catalyst/blsync.go Normal file
View file

@ -0,0 +1,88 @@
// Copyright 2024 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 catalyst
import (
"github.com/ethereum/go-ethereum/beacon/engine"
"github.com/ethereum/go-ethereum/beacon/types"
"github.com/ethereum/go-ethereum/eth"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/log"
)
// Blsync tracks the head of the beacon chain through the beacon light client
// and drives the local node via ConsensusAPI.
type Blsync struct {
engine *ConsensusAPI
client Client
headCh chan types.ChainHeadEvent
headSub event.Subscription
quitCh chan struct{}
}
type Client interface {
SubscribeChainHeadEvent(ch chan<- types.ChainHeadEvent) event.Subscription
Start()
Stop()
}
// NewBlsync creates a new beacon light syncer.
func NewBlsync(client Client, eth *eth.Ethereum) *Blsync {
return &Blsync{
engine: newConsensusAPIWithoutHeartbeat(eth),
client: client,
headCh: make(chan types.ChainHeadEvent, 1),
quitCh: make(chan struct{}),
}
}
// Start starts underlying beacon light client and the sync logic for driving
// the local node.
func (b *Blsync) Start() error {
log.Info("Beacon light sync started")
b.headSub = b.client.SubscribeChainHeadEvent(b.headCh)
go b.client.Start()
for {
select {
case <-b.quitCh:
return nil
case head := <-b.headCh:
if _, err := b.engine.NewPayloadV2(*head.HeadBlock); err != nil {
log.Error("failed to send new payload", "err", err)
continue
}
update := engine.ForkchoiceStateV1{
HeadBlockHash: head.HeadBlock.BlockHash,
SafeBlockHash: head.Finalized, //TODO pass finalized or empty hash here?
FinalizedBlockHash: head.Finalized,
}
if _, err := b.engine.ForkchoiceUpdatedV1(update, nil); err != nil {
log.Error("failed to send forkchoice updated", "err", err)
continue
}
}
}
}
// Stop signals to the light client and syncer to exit.
func (b *Blsync) Stop() error {
b.client.Stop()
close(b.quitCh)
return nil
}