cmd, eth: switch the dev synctarget to hash from block (#28209)

* cmd, eth: switch the dev synctarget to hash from block

* cmd/utils, eth/catalyst: terminate node wyen synctarget reached
This commit is contained in:
Péter Szilágyi 2023-09-29 22:11:15 +03:00 committed by GitHub
parent a408e37fa1
commit 7b6ff527d5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 126 additions and 89 deletions

View File

@ -32,6 +32,8 @@ import (
"github.com/ethereum/go-ethereum/accounts/scwallet" "github.com/ethereum/go-ethereum/accounts/scwallet"
"github.com/ethereum/go-ethereum/accounts/usbwallet" "github.com/ethereum/go-ethereum/accounts/usbwallet"
"github.com/ethereum/go-ethereum/cmd/utils" "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/catalyst"
"github.com/ethereum/go-ethereum/eth/downloader" "github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/eth/ethconfig" "github.com/ethereum/go-ethereum/eth/ethconfig"
@ -199,17 +201,18 @@ func makeFullNode(ctx *cli.Context) (*node.Node, ethapi.Backend) {
if ctx.IsSet(utils.GraphQLEnabledFlag.Name) { if ctx.IsSet(utils.GraphQLEnabledFlag.Name) {
utils.RegisterGraphQLService(stack, backend, filterSystem, &cfg.Node) utils.RegisterGraphQLService(stack, backend, filterSystem, &cfg.Node)
} }
// Add the Ethereum Stats daemon if requested. // Add the Ethereum Stats daemon if requested.
if cfg.Ethstats.URL != "" { if cfg.Ethstats.URL != "" {
utils.RegisterEthStatsService(stack, backend, cfg.Ethstats.URL) utils.RegisterEthStatsService(stack, backend, cfg.Ethstats.URL)
} }
// Configure full-sync tester service if requested // Configure full-sync tester service if requested
if ctx.IsSet(utils.SyncTargetFlag.Name) && cfg.Eth.SyncMode == downloader.FullSync { if ctx.IsSet(utils.SyncTargetFlag.Name) {
utils.RegisterFullSyncTester(stack, eth, ctx.Path(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 // Start the dev mode if requested, or launch the engine API for
// interacting with external consensus client. // interacting with external consensus client.
if ctx.IsSet(utils.DeveloperFlag.Name) { if ctx.IsSet(utils.DeveloperFlag.Name) {

View File

@ -18,7 +18,6 @@
package utils package utils
import ( import (
"bytes"
"context" "context"
"crypto/ecdsa" "crypto/ecdsa"
"encoding/hex" "encoding/hex"
@ -39,11 +38,9 @@ import (
"github.com/ethereum/go-ethereum/accounts/keystore" "github.com/ethereum/go-ethereum/accounts/keystore"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/fdlimit" "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"
"github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/core/txpool/legacypool" "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/core/vm"
"github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/crypto/kzg4844" "github.com/ethereum/go-ethereum/crypto/kzg4844"
@ -72,7 +69,6 @@ import (
"github.com/ethereum/go-ethereum/p2p/nat" "github.com/ethereum/go-ethereum/p2p/nat"
"github.com/ethereum/go-ethereum/p2p/netutil" "github.com/ethereum/go-ethereum/p2p/netutil"
"github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/trie" "github.com/ethereum/go-ethereum/trie"
"github.com/ethereum/go-ethereum/trie/triedb/hashdb" "github.com/ethereum/go-ethereum/trie/triedb/hashdb"
@ -595,9 +591,9 @@ var (
} }
// MISC settings // MISC settings
SyncTargetFlag = &cli.PathFlag{ SyncTargetFlag = &cli.StringFlag{
Name: "synctarget", 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, TakesFile: true,
Category: flags.MiscCategory, Category: flags.MiscCategory,
} }
@ -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)) log.Debug("Sanitizing Go's GC trigger", "percent", int(gogc))
godebug.SetGCPercent(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) cfg.SyncMode = *flags.GlobalTextMarshaler(ctx, SyncModeFlag.Name).(*downloader.SyncMode)
} }
if ctx.IsSet(NetworkIdFlag.Name) { if ctx.IsSet(NetworkIdFlag.Name) {
@ -1976,21 +1974,9 @@ func RegisterFilterAPI(stack *node.Node, backend ethapi.Backend, ethcfg *ethconf
} }
// RegisterFullSyncTester adds the full-sync tester service into node. // RegisterFullSyncTester adds the full-sync tester service into node.
func RegisterFullSyncTester(stack *node.Node, eth *eth.Ethereum, path string) { func RegisterFullSyncTester(stack *node.Node, eth *eth.Ethereum, target common.Hash) {
blob, err := os.ReadFile(path) catalyst.RegisterFullSyncTester(stack, eth, target)
if err != nil { log.Info("Registered full-sync tester", "hash", target)
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 SetupMetrics(ctx *cli.Context) { func SetupMetrics(ctx *cli.Context) {

View File

@ -20,7 +20,7 @@ import (
"sync" "sync"
"time" "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"
"github.com/ethereum/go-ethereum/eth/downloader" "github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
@ -28,22 +28,26 @@ import (
) )
// FullSyncTester is an auxiliary service that allows Geth to perform full sync // FullSyncTester is an auxiliary service that allows Geth to perform full sync
// alone without consensus-layer attached. Users must specify a valid block as // alone without consensus-layer attached. Users must specify a valid block hash
// the sync target. This tester can be applied to different networks, no matter // as the sync target.
// it's pre-merge or post-merge, but only for full-sync. //
// 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 { type FullSyncTester struct {
api *ConsensusAPI stack *node.Node
block *types.Block backend *eth.Ethereum
target common.Hash
closed chan struct{} closed chan struct{}
wg sync.WaitGroup wg sync.WaitGroup
} }
// RegisterFullSyncTester registers the full-sync tester service into the node // RegisterFullSyncTester registers the full-sync tester service into the node
// stack for launching and stopping the service controlled by 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{ cl := &FullSyncTester{
api: newConsensusAPIWithoutHeartbeat(backend), stack: stack,
block: block, backend: backend,
target: target,
closed: make(chan struct{}), closed: make(chan struct{}),
} }
stack.RegisterLifecycle(cl) stack.RegisterLifecycle(cl)
@ -56,29 +60,25 @@ func (tester *FullSyncTester) Start() error {
go func() { go func() {
defer tester.wg.Done() 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) ticker := time.NewTicker(time.Second * 5)
defer ticker.Stop() defer ticker.Stop()
for { for {
select { select {
case <-ticker.C: case <-ticker.C:
// Don't bother downloader in case it's already syncing. // Stop in case the target block is already stored locally.
if tester.api.eth.Downloader().Synchronising() { if block := tester.backend.BlockChain().GetBlockByHash(tester.target); block != nil {
continue log.Info("Full-sync target reached", "number", block.NumberU64(), "hash", block.Hash())
} go tester.stack.Close() // async since we need to close ourselves
// 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())
return 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: case <-tester.closed:
return return

View File

@ -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])
}
}

View File

@ -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 // RegisterPeer injects a new download peer into the set of block source to be
// used for fetching hashes and blocks from. // used for fetching hashes and blocks from.
func (d *Downloader) RegisterPeer(id string, version uint, peer Peer) error { func (d *Downloader) RegisterPeer(id string, version uint, peer Peer) error {
@ -309,11 +304,6 @@ func (d *Downloader) RegisterPeer(id string, version uint, peer Peer) error {
return nil 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 // 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 specified peer. An effort is also made to return any pending fetches into
// the queue. // the queue.

View File

@ -55,39 +55,16 @@ type peerConnection struct {
lock sync.RWMutex lock sync.RWMutex
} }
// LightPeer encapsulates the methods required to synchronise with a remote light peer. // Peer encapsulates the methods required to synchronise with a remote full peer.
type LightPeer interface { type Peer interface {
Head() (common.Hash, *big.Int) Head() (common.Hash, *big.Int)
RequestHeadersByHash(common.Hash, int, int, bool, chan *eth.Response) (*eth.Request, error) RequestHeadersByHash(common.Hash, int, int, bool, chan *eth.Response) (*eth.Request, error)
RequestHeadersByNumber(uint64, 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) RequestBodies([]common.Hash, chan *eth.Response) (*eth.Request, error)
RequestReceipts([]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. // newPeerConnection creates a new downloader peer.
func newPeerConnection(id string, version uint, peer Peer, logger log.Logger) *peerConnection { func newPeerConnection(id string, version uint, peer Peer, logger log.Logger) *peerConnection {
return &peerConnection{ return &peerConnection{