eth, les: add new config field SyncFromCheckpoint (#22123)

This PR introduces a new config field SyncFromCheckpoint for light client.

In some special scenarios, it's required to start synchronization from some
arbitrary checkpoint or even from the scratch. So this PR offers this
flexibility to users so that the synchronization start point can be configured.

There are two relevant configs: SyncFromCheckpoint and Checkpoint.

- If the SyncFromCheckpoint is true, the light client will try to sync from the
  specified checkpoint.

- If the Checkpoint is not configured, then the light client will sync from the
  scratch(from the latest header if the database is not empty)

Additional notes: these two configs are not visible in the CLI flags but only
accessable in the config file.

Example Usage:

[Eth]
SyncFromCheckpoint = true

[Eth.Checkpoint]
SectionIndex = 100
SectionHead = "0xabc"
CHTRoot = "0xabc"
BloomRoot = "0xabc"

PS. Historical checkpoint can be retrieved from the synced full node or light
client via les_getCheckpoint API.
This commit is contained in:
gary rong 2021-01-19 17:52:45 +08:00 committed by GitHub
parent 24c1e3053b
commit 45cb1a580a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 209 additions and 35 deletions

View File

@ -127,11 +127,12 @@ type Config struct {
Whitelist map[uint64]common.Hash `toml:"-"` Whitelist map[uint64]common.Hash `toml:"-"`
// Light client options // Light client options
LightServ int `toml:",omitempty"` // Maximum percentage of time allowed for serving LES requests LightServ int `toml:",omitempty"` // Maximum percentage of time allowed for serving LES requests
LightIngress int `toml:",omitempty"` // Incoming bandwidth limit for light servers LightIngress int `toml:",omitempty"` // Incoming bandwidth limit for light servers
LightEgress int `toml:",omitempty"` // Outgoing bandwidth limit for light servers LightEgress int `toml:",omitempty"` // Outgoing bandwidth limit for light servers
LightPeers int `toml:",omitempty"` // Maximum number of LES client peers LightPeers int `toml:",omitempty"` // Maximum number of LES client peers
LightNoPrune bool `toml:",omitempty"` // Whether to disable light chain pruning LightNoPrune bool `toml:",omitempty"` // Whether to disable light chain pruning
SyncFromCheckpoint bool `toml:",omitempty"` // Whether to sync the header chain from the configured checkpoint
// Ultra Light client options // Ultra Light client options
UltraLightServers []string `toml:",omitempty"` // List of trusted ultra light servers UltraLightServers []string `toml:",omitempty"` // List of trusted ultra light servers

View File

@ -21,6 +21,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
NetworkId uint64 NetworkId uint64
SyncMode downloader.SyncMode SyncMode downloader.SyncMode
EthDiscoveryURLs []string EthDiscoveryURLs []string
SnapDiscoveryURLs []string
NoPruning bool NoPruning bool
NoPrefetch bool NoPrefetch bool
TxLookupLimit uint64 `toml:",omitempty"` TxLookupLimit uint64 `toml:",omitempty"`
@ -30,6 +31,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
LightEgress int `toml:",omitempty"` LightEgress int `toml:",omitempty"`
LightPeers int `toml:",omitempty"` LightPeers int `toml:",omitempty"`
LightNoPrune bool `toml:",omitempty"` LightNoPrune bool `toml:",omitempty"`
SyncFromCheckpoint bool `toml:",omitempty"`
UltraLightServers []string `toml:",omitempty"` UltraLightServers []string `toml:",omitempty"`
UltraLightFraction int `toml:",omitempty"` UltraLightFraction int `toml:",omitempty"`
UltraLightOnlyAnnounce bool `toml:",omitempty"` UltraLightOnlyAnnounce bool `toml:",omitempty"`
@ -62,6 +64,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
enc.NetworkId = c.NetworkId enc.NetworkId = c.NetworkId
enc.SyncMode = c.SyncMode enc.SyncMode = c.SyncMode
enc.EthDiscoveryURLs = c.EthDiscoveryURLs enc.EthDiscoveryURLs = c.EthDiscoveryURLs
enc.SnapDiscoveryURLs = c.SnapDiscoveryURLs
enc.NoPruning = c.NoPruning enc.NoPruning = c.NoPruning
enc.NoPrefetch = c.NoPrefetch enc.NoPrefetch = c.NoPrefetch
enc.TxLookupLimit = c.TxLookupLimit enc.TxLookupLimit = c.TxLookupLimit
@ -71,6 +74,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
enc.LightEgress = c.LightEgress enc.LightEgress = c.LightEgress
enc.LightPeers = c.LightPeers enc.LightPeers = c.LightPeers
enc.LightNoPrune = c.LightNoPrune enc.LightNoPrune = c.LightNoPrune
enc.SyncFromCheckpoint = c.SyncFromCheckpoint
enc.UltraLightServers = c.UltraLightServers enc.UltraLightServers = c.UltraLightServers
enc.UltraLightFraction = c.UltraLightFraction enc.UltraLightFraction = c.UltraLightFraction
enc.UltraLightOnlyAnnounce = c.UltraLightOnlyAnnounce enc.UltraLightOnlyAnnounce = c.UltraLightOnlyAnnounce
@ -107,6 +111,7 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
NetworkId *uint64 NetworkId *uint64
SyncMode *downloader.SyncMode SyncMode *downloader.SyncMode
EthDiscoveryURLs []string EthDiscoveryURLs []string
SnapDiscoveryURLs []string
NoPruning *bool NoPruning *bool
NoPrefetch *bool NoPrefetch *bool
TxLookupLimit *uint64 `toml:",omitempty"` TxLookupLimit *uint64 `toml:",omitempty"`
@ -116,6 +121,7 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
LightEgress *int `toml:",omitempty"` LightEgress *int `toml:",omitempty"`
LightPeers *int `toml:",omitempty"` LightPeers *int `toml:",omitempty"`
LightNoPrune *bool `toml:",omitempty"` LightNoPrune *bool `toml:",omitempty"`
SyncFromCheckpoint *bool `toml:",omitempty"`
UltraLightServers []string `toml:",omitempty"` UltraLightServers []string `toml:",omitempty"`
UltraLightFraction *int `toml:",omitempty"` UltraLightFraction *int `toml:",omitempty"`
UltraLightOnlyAnnounce *bool `toml:",omitempty"` UltraLightOnlyAnnounce *bool `toml:",omitempty"`
@ -159,6 +165,9 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
if dec.EthDiscoveryURLs != nil { if dec.EthDiscoveryURLs != nil {
c.EthDiscoveryURLs = dec.EthDiscoveryURLs c.EthDiscoveryURLs = dec.EthDiscoveryURLs
} }
if dec.SnapDiscoveryURLs != nil {
c.SnapDiscoveryURLs = dec.SnapDiscoveryURLs
}
if dec.NoPruning != nil { if dec.NoPruning != nil {
c.NoPruning = *dec.NoPruning c.NoPruning = *dec.NoPruning
} }
@ -186,6 +195,9 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
if dec.LightNoPrune != nil { if dec.LightNoPrune != nil {
c.LightNoPrune = *dec.LightNoPrune c.LightNoPrune = *dec.LightNoPrune
} }
if dec.SyncFromCheckpoint != nil {
c.SyncFromCheckpoint = *dec.SyncFromCheckpoint
}
if dec.UltraLightServers != nil { if dec.UltraLightServers != nil {
c.UltraLightServers = dec.UltraLightServers c.UltraLightServers = dec.UltraLightServers
} }

View File

@ -44,9 +44,12 @@ type clientHandler struct {
downloader *downloader.Downloader downloader *downloader.Downloader
backend *LightEthereum backend *LightEthereum
closeCh chan struct{} closeCh chan struct{}
wg sync.WaitGroup // WaitGroup used to track all connected peers. wg sync.WaitGroup // WaitGroup used to track all connected peers.
syncDone func() // Test hooks when syncing is done.
// Hooks used in the testing
syncStart func(header *types.Header) // Hook called when the syncing is started
syncEnd func(header *types.Header) // Hook called when the syncing is done
} }
func newClientHandler(ulcServers []string, ulcFraction int, checkpoint *params.TrustedCheckpoint, backend *LightEthereum) *clientHandler { func newClientHandler(ulcServers []string, ulcFraction int, checkpoint *params.TrustedCheckpoint, backend *LightEthereum) *clientHandler {

View File

@ -157,17 +157,17 @@ func (c *lesCommons) setupOracle(node *node.Node, genesis common.Hash, ethconfig
config = params.CheckpointOracles[genesis] config = params.CheckpointOracles[genesis]
} }
if config == nil { if config == nil {
log.Info("Checkpoint registrar is not enabled") log.Info("Checkpoint oracle is not enabled")
return nil return nil
} }
if config.Address == (common.Address{}) || uint64(len(config.Signers)) < config.Threshold { if config.Address == (common.Address{}) || uint64(len(config.Signers)) < config.Threshold {
log.Warn("Invalid checkpoint registrar config") log.Warn("Invalid checkpoint oracle config")
return nil return nil
} }
oracle := checkpointoracle.New(config, c.localCheckpoint) oracle := checkpointoracle.New(config, c.localCheckpoint)
rpcClient, _ := node.Attach() rpcClient, _ := node.Attach()
client := ethclient.NewClient(rpcClient) client := ethclient.NewClient(rpcClient)
oracle.Start(client) oracle.Start(client)
log.Info("Configured checkpoint registrar", "address", config.Address, "signers", len(config.Signers), "threshold", config.Threshold) log.Info("Configured checkpoint oracle", "address", config.Address, "signers", len(config.Signers), "threshold", config.Threshold)
return oracle return oracle
} }

View File

@ -26,6 +26,7 @@ import (
"github.com/ethereum/go-ethereum/eth/downloader" "github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/light"
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/params"
) )
var errInvalidCheckpoint = errors.New("invalid advertised checkpoint") var errInvalidCheckpoint = errors.New("invalid advertised checkpoint")
@ -98,22 +99,33 @@ func (h *clientHandler) synchronise(peer *serverPeer) {
if currentTd != nil && peer.Td().Cmp(currentTd) < 0 { if currentTd != nil && peer.Td().Cmp(currentTd) < 0 {
return return
} }
// Recap the checkpoint. // Recap the checkpoint. The light client may be connected to several different
// // versions of the server.
// The light client may be connected to several different versions of the server. // (1) Old version server which can not provide stable checkpoint in the
// (1) Old version server which can not provide stable checkpoint in the handshake packet. // handshake packet.
// => Use hardcoded checkpoint or empty checkpoint // => Use local checkpoint or empty checkpoint
// (2) New version server but simple checkpoint syncing is not enabled(e.g. mainnet, new testnet or private network) // (2) New version server but simple checkpoint syncing is not enabled
// => Use hardcoded checkpoint or empty checkpoint // (e.g. mainnet, new testnet or private network)
// (3) New version server but the provided stable checkpoint is even lower than the hardcoded one. // => Use local checkpoint or empty checkpoint
// => Use hardcoded checkpoint // (3) New version server but the provided stable checkpoint is even lower
// than the local one.
// => Use local checkpoint
// (4) New version server with valid and higher stable checkpoint // (4) New version server with valid and higher stable checkpoint
// => Use provided checkpoint // => Use provided checkpoint
var checkpoint = &peer.checkpoint var (
var hardcoded bool local bool
checkpoint = &peer.checkpoint
)
if h.checkpoint != nil && h.checkpoint.SectionIndex >= peer.checkpoint.SectionIndex { if h.checkpoint != nil && h.checkpoint.SectionIndex >= peer.checkpoint.SectionIndex {
checkpoint = h.checkpoint // Use the hardcoded one. local, checkpoint = true, h.checkpoint
hardcoded = true }
// Replace the checkpoint with locally configured one If it's required by
// users. Nil checkpoint means synchronization from the scratch.
if h.backend.config.SyncFromCheckpoint {
local, checkpoint = true, h.backend.config.Checkpoint
if h.backend.config.Checkpoint == nil {
checkpoint = &params.TrustedCheckpoint{}
}
} }
// Determine whether we should run checkpoint syncing or normal light syncing. // Determine whether we should run checkpoint syncing or normal light syncing.
// //
@ -121,7 +133,7 @@ func (h *clientHandler) synchronise(peer *serverPeer) {
// //
// 1. The checkpoint is empty // 1. The checkpoint is empty
// 2. The latest head block of the local chain is above the checkpoint. // 2. The latest head block of the local chain is above the checkpoint.
// 3. The checkpoint is hardcoded(recap with local hardcoded checkpoint) // 3. The checkpoint is local(replaced with local checkpoint)
// 4. For some networks the checkpoint syncing is not activated. // 4. For some networks the checkpoint syncing is not activated.
mode := checkpointSync mode := checkpointSync
switch { switch {
@ -131,7 +143,7 @@ func (h *clientHandler) synchronise(peer *serverPeer) {
case latest.Number.Uint64() >= (checkpoint.SectionIndex+1)*h.backend.iConfig.ChtSize-1: case latest.Number.Uint64() >= (checkpoint.SectionIndex+1)*h.backend.iConfig.ChtSize-1:
mode = lightSync mode = lightSync
log.Debug("Disable checkpoint syncing", "reason", "local chain beyond the checkpoint") log.Debug("Disable checkpoint syncing", "reason", "local chain beyond the checkpoint")
case hardcoded: case local:
mode = legacyCheckpointSync mode = legacyCheckpointSync
log.Debug("Disable checkpoint syncing", "reason", "checkpoint is hardcoded") log.Debug("Disable checkpoint syncing", "reason", "checkpoint is hardcoded")
case h.backend.oracle == nil || !h.backend.oracle.IsRunning(): case h.backend.oracle == nil || !h.backend.oracle.IsRunning():
@ -143,12 +155,14 @@ func (h *clientHandler) synchronise(peer *serverPeer) {
} }
log.Debug("Disable checkpoint syncing", "reason", "checkpoint syncing is not activated") log.Debug("Disable checkpoint syncing", "reason", "checkpoint syncing is not activated")
} }
// Notify testing framework if syncing has completed(for testing purpose). // Notify testing framework if syncing has completed(for testing purpose).
defer func() { defer func() {
if h.syncDone != nil { if h.syncEnd != nil {
h.syncDone() h.syncEnd(h.backend.blockchain.CurrentHeader())
} }
}() }()
start := time.Now() start := time.Now()
if mode == checkpointSync || mode == legacyCheckpointSync { if mode == checkpointSync || mode == legacyCheckpointSync {
// Validate the advertised checkpoint // Validate the advertised checkpoint
@ -177,6 +191,10 @@ func (h *clientHandler) synchronise(peer *serverPeer) {
return return
} }
} }
if h.syncStart != nil {
h.syncStart(h.backend.blockchain.CurrentHeader())
}
// Fetch the remaining block headers based on the current chain header. // Fetch the remaining block headers based on the current chain header.
if err := h.downloader.Synchronise(peer.id, peer.Head(), peer.Td(), downloader.LightSync); err != nil { if err := h.downloader.Synchronise(peer.id, peer.Head(), peer.Td(), downloader.LightSync); err != nil {
log.Debug("Synchronise failed", "reason", err) log.Debug("Synchronise failed", "reason", err)

View File

@ -24,6 +24,7 @@ import (
"github.com/ethereum/go-ethereum/accounts/abi/bind" "github.com/ethereum/go-ethereum/accounts/abi/bind"
"github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/light"
"github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/params"
@ -53,7 +54,7 @@ func testCheckpointSyncing(t *testing.T, protocol int, syncMode int) {
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
} }
} }
// Generate 128+1 blocks (totally 1 CHT sections) // Generate 128+1 blocks (totally 1 CHT section)
server, client, tearDown := newClientServerEnv(t, int(config.ChtSize+config.ChtConfirms), protocol, waitIndexers, nil, 0, false, false, true) server, client, tearDown := newClientServerEnv(t, int(config.ChtSize+config.ChtConfirms), protocol, waitIndexers, nil, 0, false, false, true)
defer tearDown() defer tearDown()
@ -100,8 +101,7 @@ func testCheckpointSyncing(t *testing.T, protocol int, syncMode int) {
} }
done := make(chan error) done := make(chan error)
client.handler.syncDone = func() { client.handler.syncEnd = func(header *types.Header) {
header := client.handler.backend.blockchain.CurrentHeader()
if header.Number.Uint64() == expected { if header.Number.Uint64() == expected {
done <- nil done <- nil
} else { } else {
@ -144,7 +144,7 @@ func testMissOracleBackend(t *testing.T, hasCheckpoint bool) {
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
} }
} }
// Generate 512+4 blocks (totally 1 CHT sections) // Generate 128+1 blocks (totally 1 CHT section)
server, client, tearDown := newClientServerEnv(t, int(config.ChtSize+config.ChtConfirms), 3, waitIndexers, nil, 0, false, false, true) server, client, tearDown := newClientServerEnv(t, int(config.ChtSize+config.ChtConfirms), 3, waitIndexers, nil, 0, false, false, true)
defer tearDown() defer tearDown()
@ -198,8 +198,7 @@ func testMissOracleBackend(t *testing.T, hasCheckpoint bool) {
} }
done := make(chan error) done := make(chan error)
client.handler.syncDone = func() { client.handler.syncEnd = func(header *types.Header) {
header := client.handler.backend.blockchain.CurrentHeader()
if header.Number.Uint64() == expected { if header.Number.Uint64() == expected {
done <- nil done <- nil
} else { } else {
@ -220,3 +219,144 @@ func testMissOracleBackend(t *testing.T, hasCheckpoint bool) {
t.Error("checkpoint syncing timeout") t.Error("checkpoint syncing timeout")
} }
} }
func TestSyncFromConfiguredCheckpoint(t *testing.T) {
config := light.TestServerIndexerConfig
waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) {
for {
cs, _, _ := cIndexer.Sections()
bts, _, _ := btIndexer.Sections()
if cs >= 2 && bts >= 2 {
break
}
time.Sleep(10 * time.Millisecond)
}
}
// Generate 256+1 blocks (totally 2 CHT sections)
server, client, tearDown := newClientServerEnv(t, int(2*config.ChtSize+config.ChtConfirms), 3, waitIndexers, nil, 0, false, false, true)
defer tearDown()
// Configure the local checkpoint(the first section)
head := server.handler.blockchain.GetHeaderByNumber(config.ChtSize - 1).Hash()
cp := &params.TrustedCheckpoint{
SectionIndex: 0,
SectionHead: head,
CHTRoot: light.GetChtRoot(server.db, 0, head),
BloomRoot: light.GetBloomTrieRoot(server.db, 0, head),
}
client.handler.backend.config.SyncFromCheckpoint = true
client.handler.backend.config.Checkpoint = cp
client.handler.checkpoint = cp
client.handler.backend.blockchain.AddTrustedCheckpoint(cp)
var (
start = make(chan error, 1)
end = make(chan error, 1)
expectStart = config.ChtSize - 1
expectEnd = 2*config.ChtSize + config.ChtConfirms
)
client.handler.syncStart = func(header *types.Header) {
if header.Number.Uint64() == expectStart {
start <- nil
} else {
start <- fmt.Errorf("blockchain length mismatch, want %d, got %d", expectStart, header.Number)
}
}
client.handler.syncEnd = func(header *types.Header) {
if header.Number.Uint64() == expectEnd {
end <- nil
} else {
end <- fmt.Errorf("blockchain length mismatch, want %d, got %d", expectEnd, header.Number)
}
}
// Create connected peer pair.
if _, _, err := newTestPeerPair("peer", 2, server.handler, client.handler); err != nil {
t.Fatalf("Failed to connect testing peers %v", err)
}
select {
case err := <-start:
if err != nil {
t.Error("sync failed", err)
}
return
case <-time.NewTimer(10 * time.Second).C:
t.Error("checkpoint syncing timeout")
}
select {
case err := <-end:
if err != nil {
t.Error("sync failed", err)
}
return
case <-time.NewTimer(10 * time.Second).C:
t.Error("checkpoint syncing timeout")
}
}
func TestSyncAll(t *testing.T) {
config := light.TestServerIndexerConfig
waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) {
for {
cs, _, _ := cIndexer.Sections()
bts, _, _ := btIndexer.Sections()
if cs >= 2 && bts >= 2 {
break
}
time.Sleep(10 * time.Millisecond)
}
}
// Generate 256+1 blocks (totally 2 CHT sections)
server, client, tearDown := newClientServerEnv(t, int(2*config.ChtSize+config.ChtConfirms), 3, waitIndexers, nil, 0, false, false, true)
defer tearDown()
client.handler.backend.config.SyncFromCheckpoint = true
var (
start = make(chan error, 1)
end = make(chan error, 1)
expectStart = uint64(0)
expectEnd = 2*config.ChtSize + config.ChtConfirms
)
client.handler.syncStart = func(header *types.Header) {
if header.Number.Uint64() == expectStart {
start <- nil
} else {
start <- fmt.Errorf("blockchain length mismatch, want %d, got %d", expectStart, header.Number)
}
}
client.handler.syncEnd = func(header *types.Header) {
if header.Number.Uint64() == expectEnd {
end <- nil
} else {
end <- fmt.Errorf("blockchain length mismatch, want %d, got %d", expectEnd, header.Number)
}
}
// Create connected peer pair.
if _, _, err := newTestPeerPair("peer", 2, server.handler, client.handler); err != nil {
t.Fatalf("Failed to connect testing peers %v", err)
}
select {
case err := <-start:
if err != nil {
t.Error("sync failed", err)
}
return
case <-time.NewTimer(10 * time.Second).C:
t.Error("checkpoint syncing timeout")
}
select {
case err := <-end:
if err != nil {
t.Error("sync failed", err)
}
return
case <-time.NewTimer(10 * time.Second).C:
t.Error("checkpoint syncing timeout")
}
}

View File

@ -541,7 +541,7 @@ func newClientServerEnv(t *testing.T, blocks int, protocol int, callback indexer
) )
if connect { if connect {
done := make(chan struct{}) done := make(chan struct{})
client.syncDone = func() { close(done) } client.syncEnd = func(_ *types.Header) { close(done) }
cpeer, speer, err = newTestPeerPair("peer", protocol, server, client) cpeer, speer, err = newTestPeerPair("peer", protocol, server, client)
if err != nil { if err != nil {
t.Fatalf("Failed to connect testing peers %v", err) t.Fatalf("Failed to connect testing peers %v", err)