diff --git a/beacon/light/api/light_api.go b/beacon/light/api/light_api.go
index 10be896028..f5ed73ad89 100755
--- a/beacon/light/api/light_api.go
+++ b/beacon/light/api/light_api.go
@@ -296,8 +296,11 @@ func (api *BeaconLightApi) GetCheckpointData(checkpointHash common.Hash) (*types
CommitteeRoot: data.Data.Committee.Root(),
Committee: data.Data.Committee,
}
- if err := checkpoint.Validate(checkpointHash); err != nil {
- return nil, fmt.Errorf("invalid sync committee Merkle proof: %w", err)
+ if err := checkpoint.Validate(); err != nil {
+ return nil, fmt.Errorf("invalid checkpoint: %w", err)
+ }
+ if checkpoint.Header.Hash() != checkpointHash {
+ return nil, errors.New("wrong checkpoint hash")
}
return checkpoint, nil
}
diff --git a/beacon/light/committee_chain.go b/beacon/light/committee_chain.go
index 93fb6b669b..a090e298d2 100644
--- a/beacon/light/committee_chain.go
+++ b/beacon/light/committee_chain.go
@@ -188,8 +188,7 @@ func (s *CommitteeChain) Reset() {
}
}
-// CheckpointInit initializes a CommitteeChain based on a previously validated
-// checkpoint.
+// CheckpointInit initializes a CommitteeChain based on a checkpoint.
// Note: if the chain is already initialized and the committees proven by the
// checkpoint do match the existing chain then the chain is retained and the
// new checkpoint becomes fixed.
@@ -197,6 +196,9 @@ func (s *CommitteeChain) CheckpointInit(bootstrap types.BootstrapData) error {
s.chainmu.Lock()
defer s.chainmu.Unlock()
+ if err := bootstrap.Validate(); err != nil {
+ return err
+ }
period := bootstrap.Header.SyncPeriod()
if err := s.deleteFixedCommitteeRootsFrom(period + 2); err != nil {
s.Reset()
diff --git a/beacon/light/request/request.go b/beacon/light/request/request.go
index 70a7e4ba04..36128b468f 100644
--- a/beacon/light/request/request.go
+++ b/beacon/light/request/request.go
@@ -23,12 +23,17 @@ import (
)
type (
+ Server any
Request any
Response any
ID uint64
- ServerAndId struct {
- Server any
- Id ID
+ ServerAndID struct {
+ Server Server
+ ID ID
+ }
+ RequestWithID struct {
+ ServerAndID
+ Request Request
}
)
@@ -40,7 +45,7 @@ type tracker struct {
requestEvents []RequestEvent
}
-func (p *tracker) TryRequest(requestFn func(server any) (Request, float32)) (ServerAndId, Request) {
+func (p *tracker) TryRequest(requestFn func(server Server) (Request, float32)) (RequestWithID, bool) {
var (
maxServerPriority, maxRequestPriority float32
bestServer server
@@ -69,13 +74,13 @@ func (p *tracker) TryRequest(requestFn func(server any) (Request, float32)) (Ser
}
log.Debug("Request attempt", "serverCount", serverCount, "removedServers", removed, "requestCandidates", candidates)
if bestServer == nil {
- return ServerAndId{}, nil
+ return RequestWithID{}, false
}
- id := ServerAndId{Server: bestServer, Id: bestServer.sendRequest(bestRequest)}
+ id := ServerAndID{Server: bestServer, ID: bestServer.sendRequest(bestRequest)}
p.scheduler.pending[id] = pendingRequest{request: bestRequest, module: p.module}
- return id, bestRequest
+ return RequestWithID{ServerAndID: id, Request: bestRequest}, true
}
-func (p *tracker) InvalidResponse(id ServerAndId, desc string) {
+func (p *tracker) InvalidResponse(id ServerAndID, desc string) {
id.Server.(server).fail(desc)
}
diff --git a/beacon/light/request/scheduler.go b/beacon/light/request/scheduler.go
index 00cecf6f13..3cc7552a96 100644
--- a/beacon/light/request/scheduler.go
+++ b/beacon/light/request/scheduler.go
@@ -23,16 +23,27 @@ import (
"github.com/ethereum/go-ethereum/log"
)
-// Module represents an update mechanism which is typically responsible for a
-// passive data structure or a certain aspect of it. When registered to a Scheduler,
-// it can be triggered either by server events, other modules or itself.
+// Module represents a mechanism which is typically responsible for downloading
+// and updating a passive data structure.
+// Modules can start network requests through Tracker and receive request events
+// related to the sent requests that can signal a response, a failure or a timeout.
+// They also receive server-related events. Note that they do not directly interact
+// with servers but may keep track of certain parameters of registered servers,
+// based on the received server events. These server parameters may affect the
+// possible range of requests to be sent to a given server.
+// Modules are called by Scheduler whenever a global trigger is fired. All request
+// and server events fire the trigger. Modules themselves can also self-trigger,
+// ensuring an immediate next processing round after the target data structure has
+// been changed in a way that could make further actions possible either by the
+// same or another Module.
type Module interface {
- // Process is a non-blocking function that is called whenever the module is
- // triggered. It can start network requests through the received Environment
- // and/or do other data processing tasks. If triggers are set up correctly,
- // Process is eventually called whenever it might have something new to do
- // either because the data structures have been changed or because new servers
- // became available or new requests became available at existing ones.
+ // Process is a non-blocking function that is called on each Module whenever
+ // a processing round is triggered. It can start new requests through the
+ // received Tracker, process events related to servers and previosly sent
+ // requests and/or do other data processing tasks. Note that request events
+ // are only passed to the module that made the given request while server
+ // events are passed to every module. Process can also trigger a next
+ // processing round by returning true.
//
// Note: Process functions of different modules are never called concurrently;
// they are called by Scheduler in the same order of priority as they were
@@ -40,9 +51,27 @@ type Module interface {
Process(Tracker, []RequestEvent, []ServerEvent) bool
}
+// Tracker allows Modules to start requests and provide feedback about responses
+// that were found to be invalid during processing.
type Tracker interface {
- TryRequest(requestFn func(server any) (Request, float32)) (ServerAndId, Request)
- InvalidResponse(id ServerAndId, desc string)
+ // TryRequest iterates through currently available servers and selects the
+ // best server and request to send. The caller provides a callback function
+ // that generates a request candidate for each available server. Note that
+ // the module may keep track of relevant server specific info, such as assumed
+ // available range of data to request, and therefore it may generate different
+ // request candidates for different servers. The callback also returns a
+ // priority value. TryRequest selects the request candidate with the highest
+ // priority value. If multiple candidates belonging to multiple servers have
+ // the same highest priority then it selects based on server priority.
+ // If a request candidate and a server has been selected, the request is sent
+ // and also returned along with the target server and request ID.
+ TryRequest(requestFn func(server Server) (Request, float32)) (RequestWithID, bool)
+ // InvalidResponse signals that the given response was invalid. Note that
+ // certain responses can only be judged by modules, in the context of existing,
+ // partially synced data structures. Giving this signal results in blocking
+ // the given server for a certain amount of time, ensuring that the same
+ // request will not be instantly sent again to the same server.
+ InvalidResponse(id ServerAndID, desc string)
}
// Scheduler is a modular network data retrieval framework that coordinates multiple
@@ -56,7 +85,7 @@ type Scheduler struct {
names map[Module]string
trackers map[Module]*tracker
servers map[server]struct{}
- pending map[ServerAndId]pendingRequest
+ pending map[ServerAndID]pendingRequest
serverEvents []ServerEvent
stopCh chan chan struct{}
@@ -65,19 +94,32 @@ type Scheduler struct {
// testTimerResults []bool // true is appended when simulated timer is processed; false when stopped
}
+// ServerEvent represents a server-related event. These events are passed to all
+// modules. Scheduler generates an EvRegister and EvUnregister event for each
+// server when added or removed. Other, application-specific server events may
+// be emitted by the servers themselves and are also passed to the modules.
type ServerEvent struct {
- Server any
+ Server Server
Type string
- Data any
+ Data any // data type defined by application-specific events
}
+// RequestEvent represents a request-related event. These events are passed to
+// the module that sent the given request. A Finalized event means either a
+// response or a hard timeout (depending on whether there is also a Response),
+// after which no further event related to the same request is emitted.
+// Once a module has successfully sent a request, Scheduler guarantees that it
+// receives exactly one Finalized event related to it. If a request reaches a
+// soft timeout, a Timeout event is emitted that is not Finalized yet. In this
+// case later the Finalized event will also have its Timeout flag set.
type RequestEvent struct {
- ServerAndId
- Request Request
+ RequestWithID
Response Response
Timeout, Finalized bool
}
+// pendingRequest keeps track of sent and not finalized requests and their sender
+// modules and whether a soft timeout has already happened.
type pendingRequest struct {
request Request
module Module
@@ -91,7 +133,7 @@ func NewScheduler(clock mclock.Clock) *Scheduler {
servers: make(map[server]struct{}),
names: make(map[Module]string),
trackers: make(map[Module]*tracker),
- pending: make(map[ServerAndId]pendingRequest),
+ pending: make(map[ServerAndID]pendingRequest),
stopCh: make(chan chan struct{}),
// Note: testWaitCh should not have capacity in order to ensure
// that after a trigger happens testWaitCh will block until the resulting
@@ -170,10 +212,9 @@ func (s *Scheduler) Stop() {
<-stop
}
-// syncLoop calls all processable modules in the order of their registration.
-// A round of processing starts whenever there is at least one processable module.
-// Triggers triggered during a processing round do not affect the current round
-// but ensure that there is going to be a next round.
+// syncLoop calls all modules in the order of their registration.
+// A round of processing starts whenever the global trigger is fired. Triggers
+// fired during a processing round ensure that there is going to be a next round.
func (s *Scheduler) syncLoop() {
for {
s.processModules()
@@ -191,8 +232,8 @@ func (s *Scheduler) syncLoop() {
}
}
-// processModules runs an entire processing round, calling processable modules
-// with the appropriate Environment.
+// processModules runs an entire processing round, calling the process functions
+// of all modules, passing all relevant events.
func (s *Scheduler) processModules() {
s.lock.Lock()
servers := make(serverSet)
@@ -237,6 +278,8 @@ func (s *Scheduler) processModules() {
}
}
+// Trigger starts a new processing round. If fired during processing, it ensures
+// another full round of processing all modules.
func (s *Scheduler) Trigger() {
select {
case s.triggerCh <- struct{}{}:
@@ -244,17 +287,21 @@ func (s *Scheduler) Trigger() {
}
}
-func (s *Scheduler) addRequestEvent(server any, id ID, response Response, timeout, finalized bool) {
- sid := ServerAndId{Server: server, Id: id}
+// addRequestEvent adds a request event to the sender module's Tracker, ensuring
+// that the module receives it in the next processing round.
+func (s *Scheduler) addRequestEvent(server Server, id ID, response Response, timeout, finalized bool) {
+ sid := ServerAndID{Server: server, ID: id}
if pr, ok := s.pending[sid]; ok {
tracker := s.trackers[pr.module]
timeout = timeout || pr.timeout
tracker.requestEvents = append(tracker.requestEvents, RequestEvent{
- ServerAndId: sid,
- Request: pr.request,
- Response: response,
- Timeout: timeout,
- Finalized: finalized,
+ RequestWithID: RequestWithID{
+ ServerAndID: sid,
+ Request: pr.request,
+ },
+ Response: response,
+ Timeout: timeout,
+ Finalized: finalized,
})
if timeout && !finalized {
pr.timeout = true
@@ -265,11 +312,18 @@ func (s *Scheduler) addRequestEvent(server any, id ID, response Response, timeou
}
}
-func (s *Scheduler) addServerEvent(server any, event Event) {
+// addServerEvent adds a server event to the global server event list, ensuring
+// that all modules receive it in the next processing round.
+func (s *Scheduler) addServerEvent(server Server, event Event) {
s.serverEvents = append(s.serverEvents, ServerEvent{Server: server, Type: event.Type, Data: event.Data})
}
-func (s *Scheduler) handleEvent(server any, event Event) {
+// handleEvent processes an Event and adds it either as a request event or a
+// server event, depending on its type. In case of an EvUnregistered server event
+// it also closes all pending requests to the given server by emitting a failed
+// request event (Finalized without Response), ensuring that all requests get
+// finalized and thereby allowing the module logic to be safe and simple.
+func (s *Scheduler) handleEvent(server Server, event Event) {
s.Trigger()
switch event.Type {
case EvResponse:
@@ -284,7 +338,7 @@ func (s *Scheduler) handleEvent(server any, event Event) {
if id.Server != server {
continue
}
- s.addRequestEvent(server, id.Id, nil, false, true)
+ s.addRequestEvent(server, id.ID, nil, false, true)
}
s.addServerEvent(server, event)
default:
diff --git a/beacon/light/sync/head_sync.go b/beacon/light/sync/head_sync.go
index d6ab9e2849..b0988f9006 100644
--- a/beacon/light/sync/head_sync.go
+++ b/beacon/light/sync/head_sync.go
@@ -33,8 +33,8 @@ type HeadSync struct {
chain committeeChain
nextSyncPeriod uint64
chainInit bool
- unvalidatedHeads map[any]types.SignedHeader
- serverHeads map[any]types.HeadInfo
+ unvalidatedHeads map[request.Server]types.SignedHeader
+ serverHeads map[request.Server]types.HeadInfo
headServerCount map[types.HeadInfo]headServerCount
headCounter uint64
prefetchHead types.HeadInfo
@@ -50,8 +50,8 @@ func NewHeadSync(headTracker headTracker, chain committeeChain) *HeadSync {
headTracker: headTracker,
chain: chain,
nextSyncPeriod: math.MaxUint64,
- unvalidatedHeads: make(map[any]types.SignedHeader),
- serverHeads: make(map[any]types.HeadInfo),
+ unvalidatedHeads: make(map[request.Server]types.SignedHeader),
+ serverHeads: make(map[request.Server]types.HeadInfo),
headServerCount: make(map[types.HeadInfo]headServerCount),
}
return s
@@ -62,7 +62,7 @@ func (s *HeadSync) Process(tracker request.Tracker, requestEvents []request.Requ
nextPeriod, chainInit := s.chain.NextSyncPeriod()
if nextPeriod != s.nextSyncPeriod || chainInit != s.chainInit {
s.nextSyncPeriod, s.chainInit = nextPeriod, chainInit
- s.processUnvalidatedHeadsHeads()
+ trigger = s.processUnvalidatedHeadsHeads()
}
for _, event := range serverEvents {
switch event.Type {
@@ -71,7 +71,9 @@ func (s *HeadSync) Process(tracker request.Tracker, requestEvents []request.Requ
trigger = true
}
case EvNewSignedHead:
- s.newSignedHead(event.Server, event.Data.(types.SignedHeader))
+ if s.newSignedHead(event.Server, event.Data.(types.SignedHeader)) {
+ trigger = true
+ }
case request.EvUnregistered:
if s.setServerHead(event.Server, types.HeadInfo{}) {
trigger = true
@@ -83,30 +85,34 @@ func (s *HeadSync) Process(tracker request.Tracker, requestEvents []request.Requ
return
}
-func (s *HeadSync) newSignedHead(server any, signedHead types.SignedHeader) {
- if !s.chainInit || signedHead.Header.SyncPeriod() > s.nextSyncPeriod {
+func (s *HeadSync) newSignedHead(server request.Server, signedHead types.SignedHeader) (trigger bool) {
+ if !s.chainInit || types.SyncPeriod(signedHead.SignatureSlot) > s.nextSyncPeriod {
s.unvalidatedHeads[server] = signedHead
- return
+ return false
}
- s.headTracker.Validate(signedHead)
+ updated, _ := s.headTracker.Validate(signedHead)
+ return updated
}
-func (s *HeadSync) processUnvalidatedHeadsHeads() {
+func (s *HeadSync) processUnvalidatedHeadsHeads() (trigger bool) {
if !s.chainInit {
- return
+ return false
}
for server, signedHead := range s.unvalidatedHeads {
if types.SyncPeriod(signedHead.SignatureSlot) <= s.nextSyncPeriod {
- s.headTracker.Validate(signedHead)
+ if updated, _ := s.headTracker.Validate(signedHead); updated {
+ trigger = true
+ }
delete(s.unvalidatedHeads, server)
}
}
+ return
}
// setServerHead processes non-validated server head announcements and updates
// the prefetch head if necessary.
//TODO report server failure if a server announces many heads that do not become validated soon.
-func (s *HeadSync) setServerHead(server any, head types.HeadInfo) bool {
+func (s *HeadSync) setServerHead(server request.Server, head types.HeadInfo) bool {
if oldHead, ok := s.serverHeads[server]; ok {
if head == oldHead {
return false
diff --git a/beacon/light/sync/head_sync_test.go b/beacon/light/sync/head_sync_test.go
index 07ae15edc7..ec2e553514 100644
--- a/beacon/light/sync/head_sync_test.go
+++ b/beacon/light/sync/head_sync_test.go
@@ -36,75 +36,106 @@ var (
testHead3 = types.HeadInfo{Slot: 124, BlockRoot: common.Hash{3}}
testHead4 = types.HeadInfo{Slot: 125, BlockRoot: common.Hash{4}}
- testSHead1 = types.SignedHeader{Header: types.Header{Slot: 123, StateRoot: common.Hash{1}}}
+ testSHead1 = types.SignedHeader{SignatureSlot: 0x0124, Header: types.Header{Slot: 0x0123, StateRoot: common.Hash{1}}}
+ testSHead2 = types.SignedHeader{SignatureSlot: 0x2010, Header: types.Header{Slot: 0x200e, StateRoot: common.Hash{2}}}
+ // testSHead3 is at the end of period 1 but signed in period 2
+ testSHead3 = types.SignedHeader{SignatureSlot: 0x4000, Header: types.Header{Slot: 0x3fff, StateRoot: common.Hash{3}}}
+ testSHead4 = types.SignedHeader{SignatureSlot: 0x6444, Header: types.Header{Slot: 0x6443, StateRoot: common.Hash{4}}}
)
func TestValidatedHead(t *testing.T) {
- tracker := &testTracker{}
- chain := &testCommitteeChain{}
- ht := &testHeadTracker{}
+ tracker := &TestTracker{}
+ chain := &TestCommitteeChain{}
+ ht := &TestHeadTracker{}
headSync := NewHeadSync(ht, chain)
- ht.expValidated(t, 1, nil)
- headSync.Process(tracker, nil, []request.ServerEvent{
+ ht.ExpValidated(t, 1, nil)
+ ExpTrigger(t, 1, false, headSync.Process(tracker, nil, []request.ServerEvent{
{Server: testServer1, Type: request.EvRegistered},
{Server: testServer1, Type: EvNewSignedHead, Data: testSHead1},
- })
- ht.expValidated(t, 2, nil)
- chain.init = true
- headSync.Process(tracker, nil, nil)
- ht.expValidated(t, 3, []types.SignedHeader{testSHead1})
+ }))
+ ht.ExpValidated(t, 2, nil)
+ chain.SetNextSyncPeriod(0)
+ ExpTrigger(t, 2, true, headSync.Process(tracker, nil, nil))
+ ht.ExpValidated(t, 3, []types.SignedHeader{testSHead1})
+ chain.SetNextSyncPeriod(1)
+ ExpTrigger(t, 3, true, headSync.Process(tracker, nil, []request.ServerEvent{
+ {Server: testServer1, Type: EvNewSignedHead, Data: testSHead2},
+ {Server: testServer2, Type: request.EvRegistered},
+ {Server: testServer2, Type: EvNewSignedHead, Data: testSHead2},
+ }))
+ ht.ExpValidated(t, 4, []types.SignedHeader{testSHead2, testSHead2})
+ ExpTrigger(t, 4, false, headSync.Process(tracker, nil, []request.ServerEvent{
+ {Server: testServer1, Type: EvNewSignedHead, Data: testSHead3},
+ {Server: testServer3, Type: request.EvRegistered},
+ {Server: testServer3, Type: EvNewSignedHead, Data: testSHead4},
+ }))
+ ht.ExpValidated(t, 5, nil)
+ chain.SetNextSyncPeriod(2)
+ ExpTrigger(t, 5, true, headSync.Process(tracker, nil, nil))
+ ht.ExpValidated(t, 6, []types.SignedHeader{testSHead3})
+ ExpTrigger(t, 6, false, headSync.Process(tracker, nil, []request.ServerEvent{
+ {Server: testServer3, Type: request.EvUnregistered},
+ }))
+ ht.ExpValidated(t, 7, nil)
+ chain.SetNextSyncPeriod(3)
+ ExpTrigger(t, 7, false, headSync.Process(tracker, nil, nil))
+ ht.ExpValidated(t, 8, nil)
+ ExpTrigger(t, 8, true, headSync.Process(tracker, nil, []request.ServerEvent{
+ {Server: testServer2, Type: EvNewSignedHead, Data: testSHead4},
+ }))
+ ht.ExpValidated(t, 9, []types.SignedHeader{testSHead4})
}
func TestPrefetchHead(t *testing.T) {
- tracker := &testTracker{}
- chain := &testCommitteeChain{}
- ht := &testHeadTracker{}
+ tracker := &TestTracker{}
+ chain := &TestCommitteeChain{}
+ ht := &TestHeadTracker{}
headSync := NewHeadSync(ht, chain)
- ht.expPrefetch(t, 1, testHead0) // no servers registered
- headSync.Process(tracker, nil, []request.ServerEvent{
+ ht.ExpPrefetch(t, 1, testHead0) // no servers registered
+ ExpTrigger(t, 1, true, headSync.Process(tracker, nil, []request.ServerEvent{
{Server: testServer1, Type: request.EvRegistered},
{Server: testServer1, Type: EvNewHead, Data: testHead1},
- })
- ht.expPrefetch(t, 2, testHead1) // s1: h1
- headSync.Process(tracker, nil, []request.ServerEvent{
+ }))
+ ht.ExpPrefetch(t, 2, testHead1) // s1: h1
+ ExpTrigger(t, 2, true, headSync.Process(tracker, nil, []request.ServerEvent{
{Server: testServer2, Type: request.EvRegistered},
{Server: testServer2, Type: EvNewHead, Data: testHead2},
- })
- ht.expPrefetch(t, 3, testHead2) // s1: h1, s2: h2
- headSync.Process(tracker, nil, []request.ServerEvent{
+ }))
+ ht.ExpPrefetch(t, 3, testHead2) // s1: h1, s2: h2
+ ExpTrigger(t, 3, false, headSync.Process(tracker, nil, []request.ServerEvent{
{Server: testServer1, Type: EvNewHead, Data: testHead2},
- })
- ht.expPrefetch(t, 4, testHead2) // s1: h2, s2: h2
- headSync.Process(tracker, nil, []request.ServerEvent{
+ }))
+ ht.ExpPrefetch(t, 4, testHead2) // s1: h2, s2: h2
+ ExpTrigger(t, 4, false, headSync.Process(tracker, nil, []request.ServerEvent{
{Server: testServer3, Type: request.EvRegistered},
{Server: testServer3, Type: EvNewHead, Data: testHead3},
- })
- ht.expPrefetch(t, 5, testHead2) // s1: h2, s2: h2, s3: h3
- headSync.Process(tracker, nil, []request.ServerEvent{
+ }))
+ ht.ExpPrefetch(t, 5, testHead2) // s1: h2, s2: h2, s3: h3
+ ExpTrigger(t, 5, false, headSync.Process(tracker, nil, []request.ServerEvent{
{Server: testServer4, Type: request.EvRegistered},
{Server: testServer4, Type: EvNewHead, Data: testHead4},
- })
- ht.expPrefetch(t, 6, testHead2) // s1: h2, s2: h2, s3: h3, s4: h4
- headSync.Process(tracker, nil, []request.ServerEvent{
+ }))
+ ht.ExpPrefetch(t, 6, testHead2) // s1: h2, s2: h2, s3: h3, s4: h4
+ ExpTrigger(t, 6, true, headSync.Process(tracker, nil, []request.ServerEvent{
{Server: testServer2, Type: EvNewHead, Data: testHead3},
- })
- ht.expPrefetch(t, 7, testHead3) // s1: h2, s2: h3, s3: h3, s4: h4
- headSync.Process(tracker, nil, []request.ServerEvent{
+ }))
+ ht.ExpPrefetch(t, 7, testHead3) // s1: h2, s2: h3, s3: h3, s4: h4
+ ExpTrigger(t, 7, true, headSync.Process(tracker, nil, []request.ServerEvent{
{Server: testServer3, Type: request.EvUnregistered},
- })
- ht.expPrefetch(t, 8, testHead4) // s1: h2, s2: h3, s4: h4
- headSync.Process(tracker, nil, []request.ServerEvent{
+ }))
+ ht.ExpPrefetch(t, 8, testHead4) // s1: h2, s2: h3, s4: h4
+ ExpTrigger(t, 8, false, headSync.Process(tracker, nil, []request.ServerEvent{
{Server: testServer1, Type: request.EvUnregistered},
- })
- ht.expPrefetch(t, 9, testHead4) // s2: h3, s4: h4
- headSync.Process(tracker, nil, []request.ServerEvent{
+ }))
+ ht.ExpPrefetch(t, 9, testHead4) // s2: h3, s4: h4
+ ExpTrigger(t, 9, true, headSync.Process(tracker, nil, []request.ServerEvent{
{Server: testServer4, Type: request.EvUnregistered},
- })
- ht.expPrefetch(t, 10, testHead3) // s2: h3
- headSync.Process(tracker, nil, []request.ServerEvent{
+ }))
+ ht.ExpPrefetch(t, 10, testHead3) // s2: h3
+ ExpTrigger(t, 10, true, headSync.Process(tracker, nil, []request.ServerEvent{
{Server: testServer2, Type: request.EvUnregistered},
- })
- ht.expPrefetch(t, 11, testHead0) // no servers registered
+ }))
+ ht.ExpPrefetch(t, 11, testHead0) // no servers registered
}
diff --git a/beacon/light/sync/test_helpers.go b/beacon/light/sync/test_helpers.go
index fb4f435f4b..efd2f866be 100644
--- a/beacon/light/sync/test_helpers.go
+++ b/beacon/light/sync/test_helpers.go
@@ -19,51 +19,141 @@ package sync
import (
"testing"
+ "github.com/ethereum/go-ethereum/beacon/light"
"github.com/ethereum/go-ethereum/beacon/light/request"
"github.com/ethereum/go-ethereum/beacon/types"
)
-type testTracker struct{}
-
-func (t *testTracker) TryRequest(requestFn func(server any) (request.Request, float32)) (request.ServerAndId, request.Request) {
- return request.ServerAndId{}, nil
+type TestTracker struct {
+ servers []request.Server
+ allowance map[request.Server]int
+ sent []request.RequestWithID
+ lastId request.ID
}
-func (t *testTracker) InvalidResponse(id request.ServerAndId, desc string) {
+func (tt *TestTracker) AddServer(server request.Server, allowance int) {
+ tt.servers = append(tt.servers, server)
+ if tt.allowance == nil {
+ tt.allowance = make(map[request.Server]int)
+ }
+ tt.allowance[server] = allowance
+}
+
+func (tt *TestTracker) AddAllowance(server request.Server, allowance int) {
+ tt.allowance[server] += allowance
+}
+
+func (tt *TestTracker) TryRequest(requestFn func(server request.Server) (request.Request, float32)) (request.RequestWithID, bool) {
+ var (
+ bestServer request.Server
+ bestReq request.Request
+ bestPri float32
+ )
+ for _, server := range tt.servers {
+ if tt.allowance[server] == 0 {
+ continue
+ }
+ req, pri := requestFn(server)
+ if req != nil && (bestReq == nil || pri > bestPri) {
+ bestServer, bestReq, bestPri = server, req, pri
+ }
+ }
+ if bestServer == nil {
+ return request.RequestWithID{}, false
+ }
+ tt.allowance[bestServer]--
+ tt.lastId++
+ req := request.RequestWithID{
+ ServerAndID: request.ServerAndID{Server: bestServer, ID: tt.lastId},
+ Request: bestReq,
+ }
+ tt.sent = append(tt.sent, req)
+ return req, true
+}
+
+func (tt *TestTracker) ExpRequests(t *testing.T, tci int, expSent []request.RequestWithID) {
+ for i, expReq := range expSent {
+ if i >= len(tt.sent) {
+ t.Errorf("Missing sent request in test case #%d index #%d (expected %v, got none)", tci, i, expReq)
+ continue
+ }
+ if tt.sent[i] != expReq {
+ t.Errorf("Wrong sent request in test case #%d index #%d (expected %v, got %v)", tci, i, expReq, tt.sent[i])
+ }
+ }
+ for i := len(expSent); i < len(tt.sent); i++ {
+ t.Errorf("Unexpected sent request in test case #%d index #%d (expected none, got %v)", tci, i, tt.sent[i])
+ }
+ tt.sent = nil
+}
+
+func (tt *TestTracker) InvalidResponse(id request.ServerAndID, desc string) {
return
}
-type testCommitteeChain struct {
- nsp uint64
- init bool
+func ExpTrigger(t *testing.T, tci int, expTrigger, trigger bool) {
+ if trigger != expTrigger {
+ t.Errorf("Invalid process trigger output in test case #%d (expected %v, got %v)", tci, expTrigger, trigger)
+ }
}
-func (t *testCommitteeChain) CheckpointInit(bootstrap types.BootstrapData) error {
+type TestCommitteeChain struct {
+ fsp, nsp uint64
+ init bool
+}
+
+func (t *TestCommitteeChain) CheckpointInit(bootstrap types.BootstrapData) error {
+ t.fsp, t.nsp, t.init = bootstrap.Header.SyncPeriod(), bootstrap.Header.SyncPeriod()+2, true
return nil
}
-func (t *testCommitteeChain) InsertUpdate(update *types.LightClientUpdate, nextCommittee *types.SerializedSyncCommittee) error {
+func (t *TestCommitteeChain) InsertUpdate(update *types.LightClientUpdate, nextCommittee *types.SerializedSyncCommittee) error {
+ period := update.AttestedHeader.Header.SyncPeriod()
+ if period < t.fsp || period > t.nsp || !t.init {
+ return light.ErrInvalidPeriod
+ }
+ if period == t.nsp {
+ t.nsp++
+ }
return nil
}
-func (t *testCommitteeChain) NextSyncPeriod() (uint64, bool) {
+func (t *TestCommitteeChain) NextSyncPeriod() (uint64, bool) {
return t.nsp, t.init
}
-type testHeadTracker struct {
+func (tc *TestCommitteeChain) ExpInit(t *testing.T, ExpInit bool) {
+ if tc.init != ExpInit {
+ t.Errorf("Incorrect init flag (expected %v, got %v)", ExpInit, tc.init)
+ }
+}
+
+func (t *TestCommitteeChain) SetNextSyncPeriod(nsp uint64) {
+ t.init, t.nsp = true, nsp
+}
+
+func (tc *TestCommitteeChain) ExpNextSyncPeriod(t *testing.T, expNsp uint64) {
+ tc.ExpInit(t, true)
+ if tc.nsp != expNsp {
+ t.Errorf("Incorrect NextSyncPeriod (expected %d, got %d)", expNsp, tc.nsp)
+ }
+}
+
+type TestHeadTracker struct {
phead types.HeadInfo
validated []types.SignedHeader
}
-func (ht *testHeadTracker) Validate(head types.SignedHeader) (bool, error) {
+func (ht *TestHeadTracker) Validate(head types.SignedHeader) (bool, error) {
ht.validated = append(ht.validated, head)
return true, nil
}
-func (ht *testHeadTracker) expValidated(t *testing.T, tci int, expHeads []types.SignedHeader) {
+func (ht *TestHeadTracker) ExpValidated(t *testing.T, tci int, expHeads []types.SignedHeader) {
for i, expHead := range expHeads {
if i >= len(ht.validated) {
t.Errorf("Missing validated head in test case #%d index #%d (expected {slot %d blockRoot %x}, got none)", tci, i, expHead.Header.Slot, expHead.Header.Hash())
+ continue
}
if ht.validated[i] != expHead {
vhead := ht.validated[i].Header
@@ -77,11 +167,11 @@ func (ht *testHeadTracker) expValidated(t *testing.T, tci int, expHeads []types.
ht.validated = nil
}
-func (ht *testHeadTracker) SetPrefetchHead(head types.HeadInfo) {
+func (ht *TestHeadTracker) SetPrefetchHead(head types.HeadInfo) {
ht.phead = head
}
-func (ht *testHeadTracker) expPrefetch(t *testing.T, tci int, exp types.HeadInfo) {
+func (ht *TestHeadTracker) ExpPrefetch(t *testing.T, tci int, exp types.HeadInfo) {
if ht.phead != exp {
t.Errorf("Wrong prefetch head in test case #%d (expected {slot %d blockRoot %x}, got {slot %d blockRoot %x})", tci, exp.Slot, exp.BlockRoot, ht.phead.Slot, ht.phead.BlockRoot)
}
diff --git a/beacon/light/sync/update_sync.go b/beacon/light/sync/update_sync.go
index 50378b2bad..5f626f2f8b 100644
--- a/beacon/light/sync/update_sync.go
+++ b/beacon/light/sync/update_sync.go
@@ -58,18 +58,18 @@ func (s *CheckpointInit) Process(tracker request.Tracker, requestEvents []reques
s.pending = false
}
if event.Response != nil {
- if checkpoint, ok := event.Response.(*types.BootstrapData); ok && checkpoint.Validate(common.Hash(event.Request.(ReqCheckpointData))) == nil {
+ if checkpoint, ok := event.Response.(*types.BootstrapData); ok && checkpoint.Header.Hash() == common.Hash(event.Request.(ReqCheckpointData)) {
s.chain.CheckpointInit(*checkpoint) //TODO
s.initialized = true
return true
}
- tracker.InvalidResponse(event.ServerAndId, "invalid checkpoint data")
+ tracker.InvalidResponse(event.ServerAndID, "invalid checkpoint data")
}
}
if !s.pending {
- if _, request := tracker.TryRequest(func(server any) (request.Request, float32) {
+ if _, ok := tracker.TryRequest(func(server request.Server) (request.Request, float32) {
return ReqCheckpointData(s.checkpointHash), 0
- }); request != nil {
+ }); ok {
s.pending = true
}
}
@@ -77,17 +77,17 @@ func (s *CheckpointInit) Process(tracker request.Tracker, requestEvents []reques
}
type ForwardUpdateSync struct {
- chain *light.CommitteeChain
+ chain committeeChain
rangeLock rangeLock
processQueue []request.RequestEvent
- nextSyncPeriod map[any]uint64
+ nextSyncPeriod map[request.Server]uint64
}
-func NewForwardUpdateSync(chain *light.CommitteeChain) *ForwardUpdateSync {
+func NewForwardUpdateSync(chain committeeChain) *ForwardUpdateSync {
return &ForwardUpdateSync{
chain: chain,
rangeLock: make(rangeLock),
- nextSyncPeriod: make(map[any]uint64),
+ nextSyncPeriod: make(map[request.Server]uint64),
}
}
@@ -157,7 +157,7 @@ func (s *ForwardUpdateSync) processResponse(tracker request.Tracker, event reque
return
}
if err == light.ErrInvalidUpdate || err == light.ErrWrongCommitteeRoot || err == light.ErrCannotReorg {
- tracker.InvalidResponse(event.ServerAndId, "invalid update received")
+ tracker.InvalidResponse(event.ServerAndID, "invalid update received")
} else {
log.Error("Unexpected InsertUpdate error", "error", err)
}
@@ -181,7 +181,7 @@ func (s *ForwardUpdateSync) Process(tracker request.Tracker, requestEvents []req
// iterate events and add responses to process queue
for _, event := range requestEvents {
if event.Response != nil && !s.verifyRange(event) {
- tracker.InvalidResponse(event.ServerAndId, "invalid update range")
+ tracker.InvalidResponse(event.ServerAndID, "invalid update range")
event.Response = nil
}
req := event.Request.(ReqUpdates)
@@ -220,7 +220,7 @@ func (s *ForwardUpdateSync) Process(tracker request.Tracker, requestEvents []req
switch event.Type {
case EvNewSignedHead:
signedHead := event.Data.(types.SignedHeader)
- s.nextSyncPeriod[event.Server] = types.SyncPeriod(signedHead.Header.Slot + 256)
+ s.nextSyncPeriod[event.Server] = types.SyncPeriod(signedHead.SignatureSlot + 256)
case request.EvUnregistered:
delete(s.nextSyncPeriod, event.Server)
}
@@ -233,7 +233,7 @@ func (s *ForwardUpdateSync) Process(tracker request.Tracker, requestEvents []req
}
for {
firstPeriod, maxCount := s.rangeLock.firstUnlocked(startPeriod, maxUpdateRequest)
- if _, request := tracker.TryRequest(func(server any) (request.Request, float32) {
+ if request, ok := tracker.TryRequest(func(server request.Server) (request.Request, float32) {
nextPeriod := s.nextSyncPeriod[server]
if nextPeriod <= firstPeriod {
return nil, 0
@@ -243,8 +243,8 @@ func (s *ForwardUpdateSync) Process(tracker request.Tracker, requestEvents []req
count = nextPeriod - firstPeriod
}
return ReqUpdates{FirstPeriod: firstPeriod, Count: count}, float32(count)
- }); request != nil {
- req := request.(ReqUpdates)
+ }); ok {
+ req := request.Request.(ReqUpdates)
s.rangeLock.lock(req.FirstPeriod, req.Count, 1)
} else {
break
diff --git a/beacon/light/sync/update_sync_test.go b/beacon/light/sync/update_sync_test.go
new file mode 100644
index 0000000000..9b5b4be018
--- /dev/null
+++ b/beacon/light/sync/update_sync_test.go
@@ -0,0 +1,245 @@
+// 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 .
+
+package sync
+
+import (
+ "testing"
+
+ "github.com/ethereum/go-ethereum/beacon/light/request"
+ "github.com/ethereum/go-ethereum/beacon/types"
+)
+
+func TestCheckpointInit(t *testing.T) {
+ tracker := &TestTracker{}
+ // add 2 servers
+ tracker.AddServer(testServer1, 1)
+ tracker.AddServer(testServer2, 1)
+ chain := &TestCommitteeChain{}
+ checkpoint := &types.BootstrapData{Header: types.Header{Slot: 0x2000*4 + 0x1000}} // period 4
+ checkpointHash := checkpoint.Header.Hash()
+ chkInit := NewCheckpointInit(chain, checkpointHash)
+ ExpTrigger(t, 1, false, chkInit.Process(tracker, nil, []request.ServerEvent{
+ {Server: testServer1, Type: request.EvRegistered},
+ {Server: testServer2, Type: request.EvRegistered},
+ }))
+ // expect bootstrap request to server 1
+ req1 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 1}, Request: ReqCheckpointData(checkpointHash)}
+ tracker.ExpRequests(t, 1, []request.RequestWithID{req1})
+ // request times out; expect request to server 2
+ ExpTrigger(t, 2, false, chkInit.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req1, Timeout: true},
+ }, nil))
+ req2 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 2}, Request: ReqCheckpointData(checkpointHash)}
+ tracker.ExpRequests(t, 2, []request.RequestWithID{req2})
+ // invalid response to req2; expect init state to still be false
+ wrongCheckpoint := &types.BootstrapData{Header: types.Header{Slot: 123456}}
+ ExpTrigger(t, 3, false, chkInit.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req2, Response: wrongCheckpoint, Finalized: true},
+ }, nil))
+ chain.ExpInit(t, false)
+ // server 3 is registered
+ tracker.AddServer(testServer3, 1)
+ ExpTrigger(t, 4, false, chkInit.Process(tracker, nil, []request.ServerEvent{
+ {Server: testServer3, Type: request.EvRegistered},
+ }))
+ // expect bootstrap request to server 3
+ req3 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer3, ID: 3}, Request: ReqCheckpointData(checkpointHash)}
+ tracker.ExpRequests(t, 3, []request.RequestWithID{req3})
+ // valid response to req3; expect chain to be initialized
+ ExpTrigger(t, 5, true, chkInit.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req3, Response: checkpoint, Finalized: true},
+ }, nil))
+ chain.ExpInit(t, true)
+}
+
+func TestUpdateSyncParallel(t *testing.T) {
+ tracker := &TestTracker{}
+ // add 2 servers, head at period 100; allow 3-3 parallel requests for each
+ tracker.AddServer(testServer1, 3)
+ tracker.AddServer(testServer2, 3)
+ chain := &TestCommitteeChain{}
+ chain.SetNextSyncPeriod(0)
+ updateSync := NewForwardUpdateSync(chain)
+ ExpTrigger(t, 1, false, updateSync.Process(tracker, nil, []request.ServerEvent{
+ {Server: testServer1, Type: request.EvRegistered},
+ {Server: testServer1, Type: EvNewSignedHead, Data: types.SignedHeader{SignatureSlot: 0x2000*100 + 0x1000}},
+ {Server: testServer2, Type: request.EvRegistered},
+ {Server: testServer2, Type: EvNewSignedHead, Data: types.SignedHeader{SignatureSlot: 0x2000*100 + 0x1000}},
+ }))
+ // expect 6 requests to be sent
+ req1 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 1}, Request: ReqUpdates{FirstPeriod: 0, Count: 8}}
+ req2 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 2}, Request: ReqUpdates{FirstPeriod: 8, Count: 8}}
+ req3 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 3}, Request: ReqUpdates{FirstPeriod: 16, Count: 8}}
+ req4 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 4}, Request: ReqUpdates{FirstPeriod: 24, Count: 8}}
+ req5 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 5}, Request: ReqUpdates{FirstPeriod: 32, Count: 8}}
+ req6 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 6}, Request: ReqUpdates{FirstPeriod: 40, Count: 8}}
+ tracker.ExpRequests(t, 1, []request.RequestWithID{req1, req2, req3, req4, req5, req6})
+ // valid response to request 1
+ tracker.AddAllowance(testServer1, 1)
+ ExpTrigger(t, 2, true, updateSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req1, Response: testRespUpdate(req1), Finalized: true},
+ }, nil))
+ // expect 8 periods synced and a new request started
+ chain.ExpNextSyncPeriod(t, 8)
+ req7 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 7}, Request: ReqUpdates{FirstPeriod: 48, Count: 8}}
+ tracker.ExpRequests(t, 2, []request.RequestWithID{req7})
+ // valid response to requests 4 and 5
+ tracker.AddAllowance(testServer2, 2)
+ ExpTrigger(t, 3, false, updateSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req4, Response: testRespUpdate(req4), Finalized: true},
+ request.RequestEvent{RequestWithID: req5, Response: testRespUpdate(req5), Finalized: true},
+ }, nil))
+ // expect 2 more requests but no sync progress (responses 4 and 5 cannot be added before 2 and 3)
+ chain.ExpNextSyncPeriod(t, 8)
+ req8 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 8}, Request: ReqUpdates{FirstPeriod: 56, Count: 8}}
+ req9 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 9}, Request: ReqUpdates{FirstPeriod: 64, Count: 8}}
+ tracker.ExpRequests(t, 3, []request.RequestWithID{req8, req9})
+ // soft timeout for requests 2 and 3 (server 1 is overloaded)
+ ExpTrigger(t, 4, false, updateSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req2, Timeout: true},
+ request.RequestEvent{RequestWithID: req3, Timeout: true},
+ }, nil))
+ // no allowance, no more requests
+ tracker.ExpRequests(t, 4, nil)
+ // valid response to requests 6 and 8 and 9
+ tracker.AddAllowance(testServer2, 3)
+ ExpTrigger(t, 5, false, updateSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req6, Response: testRespUpdate(req6), Finalized: true},
+ request.RequestEvent{RequestWithID: req8, Response: testRespUpdate(req8), Finalized: true},
+ request.RequestEvent{RequestWithID: req9, Response: testRespUpdate(req9), Finalized: true},
+ }, nil))
+ // server 2 can now resend requests 2 and 3 (timed out by server 1) and also send a new one
+ req2r := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 10}, Request: ReqUpdates{FirstPeriod: 8, Count: 8}}
+ req3r := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 11}, Request: ReqUpdates{FirstPeriod: 16, Count: 8}}
+ req10 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 12}, Request: ReqUpdates{FirstPeriod: 72, Count: 8}}
+ tracker.ExpRequests(t, 5, []request.RequestWithID{req2r, req3r, req10})
+ // server 1 finally answers timed out request 2
+ tracker.AddAllowance(testServer1, 1)
+ ExpTrigger(t, 6, true, updateSync.Process(tracker, []request.RequestEvent{
+ // note that Timeout flag has to be true once the request timed out, even if answered later
+ request.RequestEvent{RequestWithID: req2, Response: testRespUpdate(req2), Timeout: true, Finalized: true},
+ }, nil))
+ // expect sync progress and one new request
+ chain.ExpNextSyncPeriod(t, 16)
+ req11 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 13}, Request: ReqUpdates{FirstPeriod: 80, Count: 8}}
+ tracker.ExpRequests(t, 6, []request.RequestWithID{req11})
+ // server 2 answers re-sent requests 2 and 3
+ tracker.AddAllowance(testServer2, 2)
+ ExpTrigger(t, 7, true, updateSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req2r, Response: testRespUpdate(req2r), Finalized: true},
+ request.RequestEvent{RequestWithID: req3r, Response: testRespUpdate(req3r), Finalized: true},
+ }, nil))
+ // finally the gap is filled, update can process responses up to req6
+ chain.ExpNextSyncPeriod(t, 48)
+ // expect 2 new requests from server 2 (now the available range is covered)
+ req12 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 14}, Request: ReqUpdates{FirstPeriod: 88, Count: 8}}
+ req13 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 15}, Request: ReqUpdates{FirstPeriod: 96, Count: 4}}
+ tracker.ExpRequests(t, 7, []request.RequestWithID{req12, req13})
+ // all remaining requests are answered
+ ExpTrigger(t, 8, true, updateSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req3, Response: testRespUpdate(req3), Timeout: true, Finalized: true},
+ request.RequestEvent{RequestWithID: req7, Response: testRespUpdate(req7), Finalized: true},
+ request.RequestEvent{RequestWithID: req10, Response: testRespUpdate(req10), Finalized: true},
+ request.RequestEvent{RequestWithID: req11, Response: testRespUpdate(req11), Finalized: true},
+ request.RequestEvent{RequestWithID: req12, Response: testRespUpdate(req12), Finalized: true},
+ request.RequestEvent{RequestWithID: req13, Response: testRespUpdate(req13), Finalized: true},
+ }, nil))
+ // expect chain to be fully synced
+ chain.ExpNextSyncPeriod(t, 100)
+}
+
+func TestUpdateSyncDifferentHeads(t *testing.T) {
+ tracker := &TestTracker{}
+ // add 3 servers with different announced head periods
+ tracker.AddServer(testServer1, 1)
+ tracker.AddServer(testServer2, 1)
+ tracker.AddServer(testServer3, 1)
+ chain := &TestCommitteeChain{}
+ chain.SetNextSyncPeriod(10)
+ updateSync := NewForwardUpdateSync(chain)
+ ExpTrigger(t, 1, false, updateSync.Process(tracker, nil, []request.ServerEvent{
+ {Server: testServer1, Type: request.EvRegistered},
+ {Server: testServer1, Type: EvNewSignedHead, Data: types.SignedHeader{SignatureSlot: 0x2000*15 + 0x1000}},
+ {Server: testServer2, Type: request.EvRegistered},
+ {Server: testServer2, Type: EvNewSignedHead, Data: types.SignedHeader{SignatureSlot: 0x2000*16 + 0x1000}},
+ {Server: testServer3, Type: request.EvRegistered},
+ {Server: testServer3, Type: EvNewSignedHead, Data: types.SignedHeader{SignatureSlot: 0x2000*17 + 0x1000}},
+ }))
+ // expect request to the best announced head
+ req1 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer3, ID: 1}, Request: ReqUpdates{FirstPeriod: 10, Count: 7}}
+ tracker.ExpRequests(t, 1, []request.RequestWithID{req1})
+ // request times out, expect request to the next best head
+ ExpTrigger(t, 2, false, updateSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req1, Timeout: true},
+ }, nil))
+ req2 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 2}, Request: ReqUpdates{FirstPeriod: 10, Count: 6}}
+ tracker.ExpRequests(t, 2, []request.RequestWithID{req2})
+ // request times out, expect request to the last available server
+ ExpTrigger(t, 3, false, updateSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req2, Timeout: true},
+ }, nil))
+ req3 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 3}, Request: ReqUpdates{FirstPeriod: 10, Count: 5}}
+ tracker.ExpRequests(t, 3, []request.RequestWithID{req3})
+ // valid response to request 3, expect chain synced to period 15
+ tracker.AddAllowance(testServer1, 1)
+ ExpTrigger(t, 4, true, updateSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req3, Response: testRespUpdate(req3), Finalized: true},
+ }, nil))
+ chain.ExpNextSyncPeriod(t, 15)
+ // invalid response to request 1, server can only deliver updates up to period 15 despite announced head
+ req1x := request.RequestWithID{ServerAndID: req1.ServerAndID, Request: ReqUpdates{FirstPeriod: 10, Count: 5}}
+ ExpTrigger(t, 5, false, updateSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req1, Response: testRespUpdate(req1x), Timeout: true, Finalized: true},
+ }, nil))
+ // expect no progress of chain head
+ chain.ExpNextSyncPeriod(t, 15)
+ // valid response to request 2, expect chain synced to period 16
+ tracker.AddAllowance(testServer2, 1)
+ ExpTrigger(t, 6, true, updateSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req2, Response: testRespUpdate(req2), Timeout: true, Finalized: true},
+ }, nil))
+ chain.ExpNextSyncPeriod(t, 16)
+ // a new server is registered with announced head period 17
+ tracker.AddServer(testServer4, 1)
+ ExpTrigger(t, 7, false, updateSync.Process(tracker, nil, []request.ServerEvent{
+ {Server: testServer4, Type: request.EvRegistered},
+ {Server: testServer4, Type: EvNewSignedHead, Data: types.SignedHeader{SignatureSlot: 0x2000*17 + 0x1000}},
+ }))
+ // expect request to sync one more period
+ req4 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer4, ID: 4}, Request: ReqUpdates{FirstPeriod: 16, Count: 1}}
+ tracker.ExpRequests(t, 4, []request.RequestWithID{req4})
+ // valid response, expect chain synced to period 17
+ tracker.AddAllowance(testServer1, 1)
+ ExpTrigger(t, 8, true, updateSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req4, Response: testRespUpdate(req4), Finalized: true},
+ }, nil))
+ chain.ExpNextSyncPeriod(t, 17)
+}
+
+func testRespUpdate(request request.RequestWithID) request.Response {
+ var resp RespUpdates
+ req := request.Request.(ReqUpdates)
+ resp.Updates = make([]*types.LightClientUpdate, int(req.Count))
+ resp.Committees = make([]*types.SerializedSyncCommittee, int(req.Count))
+ period := req.FirstPeriod
+ for i := range resp.Updates {
+ resp.Updates[i] = &types.LightClientUpdate{AttestedHeader: types.SignedHeader{Header: types.Header{Slot: 0x2000*period + 0x1000}}}
+ resp.Committees[i] = new(types.SerializedSyncCommittee)
+ period++
+ }
+ return resp
+}
diff --git a/beacon/types/light_sync.go b/beacon/types/light_sync.go
index 1198754058..cd1a702f17 100644
--- a/beacon/types/light_sync.go
+++ b/beacon/types/light_sync.go
@@ -42,10 +42,7 @@ type BootstrapData struct {
}
// Validate verifies the proof included in BootstrapData.
-func (c *BootstrapData) Validate(checkpointHash common.Hash) error {
- if c.Header.Hash() != checkpointHash {
- return errors.New("wrong checkpoint hash")
- }
+func (c *BootstrapData) Validate() error {
if c.CommitteeRoot != c.Committee.Root() {
return errors.New("wrong committee root")
}
diff --git a/cmd/blsync/block_sync.go b/cmd/blsync/block_sync.go
index 19d585e95e..8ee03afd1d 100755
--- a/cmd/blsync/block_sync.go
+++ b/cmd/blsync/block_sync.go
@@ -21,7 +21,6 @@ import (
"math/big"
"sync/atomic"
- "github.com/ethereum/go-ethereum/beacon/light"
"github.com/ethereum/go-ethereum/beacon/light/request"
"github.com/ethereum/go-ethereum/beacon/light/sync"
"github.com/ethereum/go-ethereum/beacon/types"
@@ -38,30 +37,32 @@ import (
"github.com/protolambda/ztyp/tree"
)
-const reverseSyncHeaders = 128
-
type beaconBlockSync struct {
recentBlocks *lru.Cache[common.Hash, *capella.BeaconBlock]
- validatedHead types.Header
+ validatedHead common.Hash
pending map[common.Hash]struct{}
- serverHeads map[any]common.Hash
- headTracker *light.HeadTracker
+ serverHeads map[request.Server]common.Hash
+ headTracker headTracker
}
-func newBeaconBlockSyncer(headTracker *light.HeadTracker) *beaconBlockSync {
+type headTracker interface {
+ PrefetchHead() types.HeadInfo
+ ValidatedHead() types.SignedHeader
+}
+
+func newBeaconBlockSyncer(headTracker headTracker) *beaconBlockSync {
return &beaconBlockSync{
headTracker: headTracker,
recentBlocks: lru.NewCache[common.Hash, *capella.BeaconBlock](10),
pending: make(map[common.Hash]struct{}),
- serverHeads: make(map[any]common.Hash),
+ serverHeads: make(map[request.Server]common.Hash),
}
}
// Process implements request.Module
func (s *beaconBlockSync) Process(tracker request.Tracker, requestEvents []request.RequestEvent, serverEvents []request.ServerEvent) (trigger bool) {
- s.validatedHead = s.headTracker.ValidatedHead().Header
- if s.validatedHead == (types.Header{}) {
- return false
+ if header := s.headTracker.ValidatedHead().Header; header != (types.Header{}) {
+ s.validatedHead = header.Hash()
}
// iterate events and add valid responses to recentBlocks
@@ -70,13 +71,12 @@ func (s *beaconBlockSync) Process(tracker request.Tracker, requestEvents []reque
if event.Response != nil {
block := event.Response.(*capella.BeaconBlock)
s.recentBlocks.Add(blockRoot, block)
- if blockRoot == s.validatedHead.Hash() {
+ if blockRoot == s.validatedHead {
trigger = true
}
}
- if event.Timeout != event.Finalized {
- // unlock if timed out or returned with an invalid response without
- // previously being unlocked by a timeout
+ if event.Timeout || event.Finalized {
+ // unlock if timed out or returned with an invalid response
delete(s.pending, blockRoot)
}
}
@@ -92,7 +92,9 @@ func (s *beaconBlockSync) Process(tracker request.Tracker, requestEvents []reque
}
// start new requests if necessary
- s.tryRequestBlock(tracker, s.validatedHead.Hash(), false)
+ if s.validatedHead != (common.Hash{}) {
+ s.tryRequestBlock(tracker, s.validatedHead, false)
+ }
if prefetchHead := s.headTracker.PrefetchHead().BlockRoot; prefetchHead != (common.Hash{}) {
s.tryRequestBlock(tracker, prefetchHead, true)
}
@@ -101,7 +103,7 @@ func (s *beaconBlockSync) Process(tracker request.Tracker, requestEvents []reque
// belongs to validatedHead (or nil)
func (s *beaconBlockSync) getHeadBlock() *capella.BeaconBlock {
- block, _ := s.recentBlocks.Get(s.validatedHead.Hash())
+ block, _ := s.recentBlocks.Get(s.validatedHead)
return block
}
@@ -112,14 +114,14 @@ func (s *beaconBlockSync) tryRequestBlock(tracker request.Tracker, blockRoot com
if _, ok := s.pending[blockRoot]; ok {
return
}
- if _, request := tracker.TryRequest(func(server any) (request.Request, float32) {
+ if _, ok := tracker.TryRequest(func(server request.Server) (request.Request, float32) {
if prefetch && s.serverHeads[server] != blockRoot {
// when requesting a not yet validated head, request it from someone
// who has announced it already
return nil, 0
}
return sync.ReqBeaconBlock(blockRoot), 0
- }); request != nil {
+ }); ok {
s.pending[blockRoot] = struct{}{}
}
}
@@ -170,6 +172,10 @@ func getExecBlock(beaconBlock *capella.BeaconBlock) (*ctypes.Block, error) {
return execBlock, nil
}
+func beaconBlockHash(beaconBlock *capella.BeaconBlock) common.Hash {
+ return common.Hash(beaconBlock.HashTreeRoot(configs.Mainnet, tree.GetHashFn()))
+}
+
type engineApiUpdater struct {
client *rpc.Client
trigger func()
@@ -187,7 +193,7 @@ func (s *engineApiUpdater) Process(tracker request.Tracker, requestEvents []requ
if headBlock == nil {
return false
}
- headRoot := common.Hash(headBlock.HashTreeRoot(configs.Mainnet, tree.GetHashFn()))
+ headRoot := beaconBlockHash(headBlock)
if headRoot == s.lastHead {
return false
}
diff --git a/cmd/blsync/block_sync_test.go b/cmd/blsync/block_sync_test.go
new file mode 100644
index 0000000000..ab82fd4076
--- /dev/null
+++ b/cmd/blsync/block_sync_test.go
@@ -0,0 +1,140 @@
+// 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 .
+
+package main
+
+import (
+ "testing"
+
+ "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/protolambda/zrnt/eth2/beacon/capella"
+ "github.com/protolambda/zrnt/eth2/configs"
+ "github.com/protolambda/ztyp/tree"
+)
+
+var (
+ testServer1 = 1
+ testServer2 = 2
+
+ testBlock1 = &capella.BeaconBlock{Slot: 123}
+ testBlock2 = &capella.BeaconBlock{Slot: 124}
+)
+
+func TestBlockSync(t *testing.T) {
+ tracker := &sync.TestTracker{}
+ tracker.AddServer(testServer1, 1)
+ tracker.AddServer(testServer2, 1)
+ ht := &testHeadTracker{}
+ blockSync := newBeaconBlockSyncer(ht)
+
+ expHeadBlock := func(tci int, expHead *capella.BeaconBlock) {
+ expInfo := blockHeadInfo(expHead)
+ headInfo := blockHeadInfo(blockSync.getHeadBlock())
+ if headInfo != expInfo {
+ t.Errorf("Wrong head block in test case #%d (expected {slot %d blockRoot %x}, got {slot %d blockRoot %x})", tci, expInfo.Slot, expInfo.BlockRoot, headInfo.Slot, headInfo.BlockRoot)
+ }
+ }
+
+ sync.ExpTrigger(t, 1, false, blockSync.Process(tracker, nil, []request.ServerEvent{
+ {Server: testServer1, Type: request.EvRegistered},
+ {Server: testServer2, Type: request.EvRegistered},
+ }))
+ // no block requests expected until head tracker knows about a head
+ tracker.ExpRequests(t, 1, nil)
+ expHeadBlock(1, nil)
+ // set block 1 as prefetch head, announced by server 2
+ head1 := blockHeadInfo(testBlock1)
+ ht.prefetch = head1
+ sync.ExpTrigger(t, 2, false, blockSync.Process(tracker, nil, []request.ServerEvent{
+ {Server: testServer2, Type: sync.EvNewHead, Data: head1},
+ }))
+ // expect request to server 2 which has announced the head
+ req1 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 1}, Request: sync.ReqBeaconBlock(head1.BlockRoot)}
+ tracker.ExpRequests(t, 2, []request.RequestWithID{req1})
+ // valid response
+ tracker.AddAllowance(testServer2, 1)
+ sync.ExpTrigger(t, 3, false, blockSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req1, Response: testBlock1, Finalized: true},
+ }, nil))
+ // head block still not expected as the fetched block is not the validated head yet
+ expHeadBlock(2, nil)
+ // set as validated head, expect no further requests but block 1 set as head block
+ ht.validated.Header = blockHeader(testBlock1)
+ sync.ExpTrigger(t, 4, false, blockSync.Process(tracker, nil, nil))
+ tracker.ExpRequests(t, 3, nil)
+ expHeadBlock(3, testBlock1)
+
+ // set block 2 as prefetch head, announced by server 1
+ head2 := blockHeadInfo(testBlock2)
+ ht.prefetch = head2
+ sync.ExpTrigger(t, 5, false, blockSync.Process(tracker, nil, []request.ServerEvent{
+ {Server: testServer1, Type: sync.EvNewHead, Data: head2},
+ }))
+ // expect request to server 1
+ req2 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 2}, Request: sync.ReqBeaconBlock(head2.BlockRoot)}
+ tracker.ExpRequests(t, 4, []request.RequestWithID{req2})
+ // req2 times out but no further requests expected because server 2 has not announced it
+ sync.ExpTrigger(t, 6, false, blockSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req2, Timeout: true},
+ }, nil))
+ tracker.ExpRequests(t, 5, nil)
+ // set as validated head before retrieving block; now it's assumed to be available from server 2 too
+ ht.validated.Header = blockHeader(testBlock2)
+ sync.ExpTrigger(t, 7, false, blockSync.Process(tracker, nil, nil))
+ // now head block is unavailable again
+ expHeadBlock(4, nil)
+ // expect req2 retry to server 2
+ req2r := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 3}, Request: sync.ReqBeaconBlock(head2.BlockRoot)}
+ tracker.ExpRequests(t, 6, []request.RequestWithID{req2r})
+ // valid response, now head block should be block 2 immediately as it is already validated
+ sync.ExpTrigger(t, 8, true, blockSync.Process(tracker, []request.RequestEvent{
+ request.RequestEvent{RequestWithID: req2r, Response: testBlock2, Finalized: true},
+ }, nil))
+ expHeadBlock(5, testBlock2)
+}
+
+func blockHeadInfo(block *capella.BeaconBlock) types.HeadInfo {
+ if block == nil {
+ return types.HeadInfo{}
+ }
+ return types.HeadInfo{Slot: uint64(block.Slot), BlockRoot: beaconBlockHash(block)}
+}
+
+func blockHeader(block *capella.BeaconBlock) types.Header {
+ return types.Header{
+ Slot: uint64(block.Slot),
+ ProposerIndex: uint64(block.ProposerIndex),
+ ParentRoot: common.Hash(block.ParentRoot),
+ StateRoot: common.Hash(block.StateRoot),
+ BodyRoot: common.Hash(block.Body.HashTreeRoot(configs.Mainnet, tree.GetHashFn())),
+ }
+}
+
+type testHeadTracker struct {
+ prefetch types.HeadInfo
+ validated types.SignedHeader
+}
+
+func (h *testHeadTracker) PrefetchHead() types.HeadInfo {
+ return h.prefetch
+}
+
+func (h *testHeadTracker) ValidatedHead() types.SignedHeader {
+ return h.validated
+}