Skip to content

Commit

Permalink
cmd, eth: switch the dev synctarget to hash from block (#28209)
Browse files Browse the repository at this point in the history
* cmd, eth: switch the dev synctarget to hash from block

* cmd/utils, eth/catalyst: terminate node wyen synctarget reached
  • Loading branch information
karalabe authored Sep 29, 2023
1 parent a408e37 commit 7b6ff52
Show file tree
Hide file tree
Showing 6 changed files with 126 additions and 89 deletions.
13 changes: 8 additions & 5 deletions cmd/geth/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@ import (
"github.com/ethereum/go-ethereum/accounts/scwallet"
"github.com/ethereum/go-ethereum/accounts/usbwallet"
"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/eth/catalyst"
"github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/eth/ethconfig"
Expand Down Expand Up @@ -199,17 +201,18 @@ func makeFullNode(ctx *cli.Context) (*node.Node, ethapi.Backend) {
if ctx.IsSet(utils.GraphQLEnabledFlag.Name) {
utils.RegisterGraphQLService(stack, backend, filterSystem, &cfg.Node)
}

// Add the Ethereum Stats daemon if requested.
if cfg.Ethstats.URL != "" {
utils.RegisterEthStatsService(stack, backend, cfg.Ethstats.URL)
}

// Configure full-sync tester service if requested
if ctx.IsSet(utils.SyncTargetFlag.Name) && cfg.Eth.SyncMode == downloader.FullSync {
utils.RegisterFullSyncTester(stack, eth, ctx.Path(utils.SyncTargetFlag.Name))
if ctx.IsSet(utils.SyncTargetFlag.Name) {
hex := hexutil.MustDecode(ctx.String(utils.SyncTargetFlag.Name))
if len(hex) != common.HashLength {
utils.Fatalf("invalid sync target length: have %d, want %d", len(hex), common.HashLength)
}
utils.RegisterFullSyncTester(stack, eth, common.BytesToHash(hex))
}

// Start the dev mode if requested, or launch the engine API for
// interacting with external consensus client.
if ctx.IsSet(utils.DeveloperFlag.Name) {
Expand Down
30 changes: 8 additions & 22 deletions cmd/utils/flags.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@
package utils

import (
"bytes"
"context"
"crypto/ecdsa"
"encoding/hex"
Expand All @@ -39,11 +38,9 @@ import (
"github.com/ethereum/go-ethereum/accounts/keystore"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/fdlimit"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/core/txpool/legacypool"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/core/vm"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/crypto/kzg4844"
Expand Down Expand Up @@ -72,7 +69,6 @@ import (
"github.com/ethereum/go-ethereum/p2p/nat"
"github.com/ethereum/go-ethereum/p2p/netutil"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/trie"
"github.com/ethereum/go-ethereum/trie/triedb/hashdb"
Expand Down Expand Up @@ -595,9 +591,9 @@ var (
}

// MISC settings
SyncTargetFlag = &cli.PathFlag{
SyncTargetFlag = &cli.StringFlag{
Name: "synctarget",
Usage: `File for containing the hex-encoded block-rlp as sync target(dev feature)`,
Usage: `Hash of the block to full sync to (dev testing feature)`,
TakesFile: true,
Category: flags.MiscCategory,
}
Expand Down Expand Up @@ -1691,7 +1687,9 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
log.Debug("Sanitizing Go's GC trigger", "percent", int(gogc))
godebug.SetGCPercent(int(gogc))

if ctx.IsSet(SyncModeFlag.Name) {
if ctx.IsSet(SyncTargetFlag.Name) {
cfg.SyncMode = downloader.FullSync // dev sync target forces full sync
} else if ctx.IsSet(SyncModeFlag.Name) {
cfg.SyncMode = *flags.GlobalTextMarshaler(ctx, SyncModeFlag.Name).(*downloader.SyncMode)
}
if ctx.IsSet(NetworkIdFlag.Name) {
Expand Down Expand Up @@ -1976,21 +1974,9 @@ func RegisterFilterAPI(stack *node.Node, backend ethapi.Backend, ethcfg *ethconf
}

// RegisterFullSyncTester adds the full-sync tester service into node.
func RegisterFullSyncTester(stack *node.Node, eth *eth.Ethereum, path string) {
blob, err := os.ReadFile(path)
if err != nil {
Fatalf("Failed to read block file: %v", err)
}
rlpBlob, err := hexutil.Decode(string(bytes.TrimRight(blob, "\r\n")))
if err != nil {
Fatalf("Failed to decode block blob: %v", err)
}
var block types.Block
if err := rlp.DecodeBytes(rlpBlob, &block); err != nil {
Fatalf("Failed to decode block: %v", err)
}
catalyst.RegisterFullSyncTester(stack, eth, &block)
log.Info("Registered full-sync tester", "number", block.NumberU64(), "hash", block.Hash())
func RegisterFullSyncTester(stack *node.Node, eth *eth.Ethereum, target common.Hash) {
catalyst.RegisterFullSyncTester(stack, eth, target)
log.Info("Registered full-sync tester", "hash", target)
}

func SetupMetrics(ctx *cli.Context) {
Expand Down
54 changes: 27 additions & 27 deletions eth/catalyst/tester.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,31 +20,35 @@ import (
"sync"
"time"

"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/eth"
"github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
)

// FullSyncTester is an auxiliary service that allows Geth to perform full sync
// alone without consensus-layer attached. Users must specify a valid block as
// the sync target. This tester can be applied to different networks, no matter
// it's pre-merge or post-merge, but only for full-sync.
// alone without consensus-layer attached. Users must specify a valid block hash
// as the sync target.
//
// This tester can be applied to different networks, no matter it's pre-merge or
// post-merge, but only for full-sync.
type FullSyncTester struct {
api *ConsensusAPI
block *types.Block
closed chan struct{}
wg sync.WaitGroup
stack *node.Node
backend *eth.Ethereum
target common.Hash
closed chan struct{}
wg sync.WaitGroup
}

// RegisterFullSyncTester registers the full-sync tester service into the node
// stack for launching and stopping the service controlled by node.
func RegisterFullSyncTester(stack *node.Node, backend *eth.Ethereum, block *types.Block) (*FullSyncTester, error) {
func RegisterFullSyncTester(stack *node.Node, backend *eth.Ethereum, target common.Hash) (*FullSyncTester, error) {
cl := &FullSyncTester{
api: newConsensusAPIWithoutHeartbeat(backend),
block: block,
closed: make(chan struct{}),
stack: stack,
backend: backend,
target: target,
closed: make(chan struct{}),
}
stack.RegisterLifecycle(cl)
return cl, nil
Expand All @@ -56,29 +60,25 @@ func (tester *FullSyncTester) Start() error {
go func() {
defer tester.wg.Done()

// Trigger beacon sync with the provided block hash as trusted
// chain head.
err := tester.backend.Downloader().BeaconDevSync(downloader.FullSync, tester.target, tester.closed)
if err != nil {
log.Info("Failed to trigger beacon sync", "err", err)
}

ticker := time.NewTicker(time.Second * 5)
defer ticker.Stop()

for {
select {
case <-ticker.C:
// Don't bother downloader in case it's already syncing.
if tester.api.eth.Downloader().Synchronising() {
continue
}
// Short circuit in case the target block is already stored
// locally. TODO(somehow terminate the node stack if target
// is reached).
if tester.api.eth.BlockChain().HasBlock(tester.block.Hash(), tester.block.NumberU64()) {
log.Info("Full-sync target reached", "number", tester.block.NumberU64(), "hash", tester.block.Hash())
// Stop in case the target block is already stored locally.
if block := tester.backend.BlockChain().GetBlockByHash(tester.target); block != nil {
log.Info("Full-sync target reached", "number", block.NumberU64(), "hash", block.Hash())
go tester.stack.Close() // async since we need to close ourselves
return
}
// Trigger beacon sync with the provided block header as
// trusted chain head.
err := tester.api.eth.Downloader().BeaconSync(downloader.FullSync, tester.block.Header(), tester.block.Header())
if err != nil {
log.Info("Failed to beacon sync", "err", err)
}

case <-tester.closed:
return
Expand Down
81 changes: 81 additions & 0 deletions eth/downloader/beacondevsync.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
// Copyright 2023 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 downloader

import (
"errors"
"time"

"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
)

// BeaconDevSync is a development helper to test synchronization by providing
// a block hash instead of header to run the beacon sync against.
//
// The method will reach out to the network to retrieve the header of the sync
// target instead of receiving it from the consensus node.
//
// Note, this must not be used in live code. If the forkchcoice endpoint where
// to use this instead of giving us the payload first, then essentially nobody
// in the network would have the block yet that we'd attempt to retrieve.
func (d *Downloader) BeaconDevSync(mode SyncMode, hash common.Hash, stop chan struct{}) error {
// Be very loud that this code should not be used in a live node
log.Warn("----------------------------------")
log.Warn("Beacon syncing with hash as target", "hash", hash)
log.Warn("This is unhealthy for a live node!")
log.Warn("----------------------------------")

log.Info("Waiting for peers to retrieve sync target")
for {
// If the node is going down, unblock
select {
case <-stop:
return errors.New("stop requested")
default:
}
// Pick a random peer to sync from and keep retrying if none are yet
// available due to fresh startup
d.peers.lock.RLock()
var peer *peerConnection
for _, peer = range d.peers.peers {
break
}
d.peers.lock.RUnlock()

if peer == nil {
time.Sleep(time.Second)
continue
}
// Found a peer, attempt to retrieve the header whilst blocking and
// retry if it fails for whatever reason
log.Info("Attempting to retrieve sync target", "peer", peer.id)
headers, metas, err := d.fetchHeadersByHash(peer, hash, 1, 0, false)
if err != nil || len(headers) != 1 {
log.Warn("Failed to fetch sync target", "headers", len(headers), "err", err)
time.Sleep(time.Second)
continue
}
// Head header retrieved, if the hash matches, start the actual sync
if metas[0] != hash {
log.Error("Received invalid sync target", "want", hash, "have", metas[0])
time.Sleep(time.Second)
continue
}
return d.BeaconSync(mode, headers[0], headers[0])
}
}
10 changes: 0 additions & 10 deletions eth/downloader/downloader.go
Original file line number Diff line number Diff line change
Expand Up @@ -286,11 +286,6 @@ func (d *Downloader) Progress() ethereum.SyncProgress {
}
}

// Synchronising returns whether the downloader is currently retrieving blocks.
func (d *Downloader) Synchronising() bool {
return d.synchronising.Load()
}

// RegisterPeer injects a new download peer into the set of block source to be
// used for fetching hashes and blocks from.
func (d *Downloader) RegisterPeer(id string, version uint, peer Peer) error {
Expand All @@ -309,11 +304,6 @@ func (d *Downloader) RegisterPeer(id string, version uint, peer Peer) error {
return nil
}

// RegisterLightPeer injects a light client peer, wrapping it so it appears as a regular peer.
func (d *Downloader) RegisterLightPeer(id string, version uint, peer LightPeer) error {
return d.RegisterPeer(id, version, &lightPeerWrapper{peer})
}

// UnregisterPeer remove a peer from the known list, preventing any action from
// the specified peer. An effort is also made to return any pending fetches into
// the queue.
Expand Down
27 changes: 2 additions & 25 deletions eth/downloader/peer.go
Original file line number Diff line number Diff line change
Expand Up @@ -55,39 +55,16 @@ type peerConnection struct {
lock sync.RWMutex
}

// LightPeer encapsulates the methods required to synchronise with a remote light peer.
type LightPeer interface {
// Peer encapsulates the methods required to synchronise with a remote full peer.
type Peer interface {
Head() (common.Hash, *big.Int)
RequestHeadersByHash(common.Hash, int, int, bool, chan *eth.Response) (*eth.Request, error)
RequestHeadersByNumber(uint64, int, int, bool, chan *eth.Response) (*eth.Request, error)
}

// Peer encapsulates the methods required to synchronise with a remote full peer.
type Peer interface {
LightPeer
RequestBodies([]common.Hash, chan *eth.Response) (*eth.Request, error)
RequestReceipts([]common.Hash, chan *eth.Response) (*eth.Request, error)
}

// lightPeerWrapper wraps a LightPeer struct, stubbing out the Peer-only methods.
type lightPeerWrapper struct {
peer LightPeer
}

func (w *lightPeerWrapper) Head() (common.Hash, *big.Int) { return w.peer.Head() }
func (w *lightPeerWrapper) RequestHeadersByHash(h common.Hash, amount int, skip int, reverse bool, sink chan *eth.Response) (*eth.Request, error) {
return w.peer.RequestHeadersByHash(h, amount, skip, reverse, sink)
}
func (w *lightPeerWrapper) RequestHeadersByNumber(i uint64, amount int, skip int, reverse bool, sink chan *eth.Response) (*eth.Request, error) {
return w.peer.RequestHeadersByNumber(i, amount, skip, reverse, sink)
}
func (w *lightPeerWrapper) RequestBodies([]common.Hash, chan *eth.Response) (*eth.Request, error) {
panic("RequestBodies not supported in light client mode sync")
}
func (w *lightPeerWrapper) RequestReceipts([]common.Hash, chan *eth.Response) (*eth.Request, error) {
panic("RequestReceipts not supported in light client mode sync")
}

// newPeerConnection creates a new downloader peer.
func newPeerConnection(id string, version uint, peer Peer, logger log.Logger) *peerConnection {
return &peerConnection{
Expand Down

0 comments on commit 7b6ff52

Please sign in to comment.