beacon/light/sync: more unit tests and other changes

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 {
+	if err := checkpoint.Validate(); err != nil {
-		return nil, fmt.Errorf("invalid sync committee Merkle proof: %w", err)
+		return nil, fmt.Errorf("invalid checkpoint: %w", err)
+	}
+	if checkpoint.Header.Hash() != checkpointHash {
+		return nil, errors.New("wrong checkpoint hash")
 	}
 	return checkpoint, nil
 }

beacon/light/committee_chain.go

@@ -188,8 +188,7 @@ func (s *CommitteeChain) Reset() {
 	}
 }
 
-// CheckpointInit initializes a CommitteeChain based on a previously validated
+// CheckpointInit initializes a CommitteeChain based on a checkpoint.
-// 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()

beacon/light/request/request.go

@@ -23,12 +23,17 @@ import (
 )
 
 type (
+	Server      any
 	Request     any
 	Response    any
 	ID          uint64
-	ServerAndId struct {
+	ServerAndID struct {
-		Server any
+		Server Server
-		Id     ID
+		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)
 }

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
+// Module represents a mechanism which is typically responsible for downloading
-// passive data structure or a certain aspect of it. When registered to a Scheduler,
+// and updating a passive data structure.
-// it can be triggered either by server events, other modules or itself.
+// 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
+	// Process is a non-blocking function that is called on each Module whenever
-	// triggered. It can start network requests through the received Environment
+	// a processing round is triggered. It can start new requests through the
-	// and/or do other data processing tasks. If triggers are set up correctly,
+	// received Tracker, process events related to servers and previosly sent
-	// Process is eventually called whenever it might have something new to do
+	// requests and/or do other data processing tasks. Note that request events
-	// either because the data structures have been changed or because new servers
+	// are only passed to the module that made the given request while server
-	// became available or new requests became available at existing ones.
+	// 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)
+	// TryRequest iterates through currently available servers and selects the
-	InvalidResponse(id ServerAndId, desc string)
+	// 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
+	RequestWithID
-	Request            Request
 	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.
+// syncLoop calls all modules in the order of their registration.
-// A round of processing starts whenever there is at least one processable module.
+// A round of processing starts whenever the global trigger is fired. Triggers
-// Triggers triggered during a processing round do not affect the current round
+// fired during a processing round ensure that there is going to be a next round.
-// but 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
+// processModules runs an entire processing round, calling the process functions
-// with the appropriate Environment.
+// 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) {
+// addRequestEvent adds a request event to the sender module's Tracker, ensuring
-	sid := ServerAndId{Server: server, Id: id}
+// 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,
+			RequestWithID: RequestWithID{
-			Request:     pr.request,
+				ServerAndID: sid,
-			Response:    response,
+				Request:     pr.request,
-			Timeout:     timeout,
+			},
-			Finalized:   finalized,
+			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:

beacon/light/sync/head_sync.go

@@ -33,8 +33,8 @@ type HeadSync struct {
 	chain            committeeChain
 	nextSyncPeriod   uint64
 	chainInit        bool
-	unvalidatedHeads map[any]types.SignedHeader
+	unvalidatedHeads map[request.Server]types.SignedHeader
-	serverHeads      map[any]types.HeadInfo
+	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),
+		unvalidatedHeads: make(map[request.Server]types.SignedHeader),
-		serverHeads:      make(map[any]types.HeadInfo),
+		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) {
+func (s *HeadSync) newSignedHead(server request.Server, signedHead types.SignedHeader) (trigger bool) {
-	if !s.chainInit || signedHead.Header.SyncPeriod() > s.nextSyncPeriod {
+	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

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{}
+	tracker := &TestTracker{}
-	chain := &testCommitteeChain{}
+	chain := &TestCommitteeChain{}
-	ht := &testHeadTracker{}
+	ht := &TestHeadTracker{}
 	headSync := NewHeadSync(ht, chain)
 
-	ht.expValidated(t, 1, nil)
+	ht.ExpValidated(t, 1, nil)
-	headSync.Process(tracker, nil, []request.ServerEvent{
+	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)
+	ht.ExpValidated(t, 2, nil)
-	chain.init = true
+	chain.SetNextSyncPeriod(0)
-	headSync.Process(tracker, nil, nil)
+	ExpTrigger(t, 2, true, headSync.Process(tracker, nil, nil))
-	ht.expValidated(t, 3, []types.SignedHeader{testSHead1})
+	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{}
+	tracker := &TestTracker{}
-	chain := &testCommitteeChain{}
+	chain := &TestCommitteeChain{}
-	ht := &testHeadTracker{}
+	ht := &TestHeadTracker{}
 	headSync := NewHeadSync(ht, chain)
 
-	ht.expPrefetch(t, 1, testHead0) // no servers registered
+	ht.ExpPrefetch(t, 1, testHead0) // no servers registered
-	headSync.Process(tracker, nil, []request.ServerEvent{
+	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
+	ht.ExpPrefetch(t, 2, testHead1) // s1: h1
-	headSync.Process(tracker, nil, []request.ServerEvent{
+	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
+	ht.ExpPrefetch(t, 3, testHead2) // s1: h1, s2: h2
-	headSync.Process(tracker, nil, []request.ServerEvent{
+	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
+	ht.ExpPrefetch(t, 4, testHead2) // s1: h2, s2: h2
-	headSync.Process(tracker, nil, []request.ServerEvent{
+	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
+	ht.ExpPrefetch(t, 5, testHead2) // s1: h2, s2: h2, s3: h3
-	headSync.Process(tracker, nil, []request.ServerEvent{
+	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
+	ht.ExpPrefetch(t, 6, testHead2) // s1: h2, s2: h2, s3: h3, s4: h4
-	headSync.Process(tracker, nil, []request.ServerEvent{
+	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
+	ht.ExpPrefetch(t, 7, testHead3) // s1: h2, s2: h3, s3: h3, s4: h4
-	headSync.Process(tracker, nil, []request.ServerEvent{
+	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
+	ht.ExpPrefetch(t, 8, testHead4) // s1: h2, s2: h3, s4: h4
-	headSync.Process(tracker, nil, []request.ServerEvent{
+	ExpTrigger(t, 8, false, headSync.Process(tracker, nil, []request.ServerEvent{
 		{Server: testServer1, Type: request.EvUnregistered},
-	})
+	}))
-	ht.expPrefetch(t, 9, testHead4) // s2: h3, s4: h4
+	ht.ExpPrefetch(t, 9, testHead4) // s2: h3, s4: h4
-	headSync.Process(tracker, nil, []request.ServerEvent{
+	ExpTrigger(t, 9, true, headSync.Process(tracker, nil, []request.ServerEvent{
 		{Server: testServer4, Type: request.EvUnregistered},
-	})
+	}))
-	ht.expPrefetch(t, 10, testHead3) // s2: h3
+	ht.ExpPrefetch(t, 10, testHead3) // s2: h3
-	headSync.Process(tracker, nil, []request.ServerEvent{
+	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
 }

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{}
+type TestTracker struct {
-
+	servers   []request.Server
-func (t *testTracker) TryRequest(requestFn func(server any) (request.Request, float32)) (request.ServerAndId, request.Request) {
+	allowance map[request.Server]int
-	return request.ServerAndId{}, nil
+	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 {
+func ExpTrigger(t *testing.T, tci int, expTrigger, trigger bool) {
-	nsp  uint64
+	if trigger != expTrigger {
-	init bool
+		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)
 	}

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 {
+		}); ok {
-			req := request.(ReqUpdates)
+			req := request.Request.(ReqUpdates)
 			s.rangeLock.lock(req.FirstPeriod, req.Count, 1)
 		} else {
 			break

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 <http://www.gnu.org/licenses/>.
+
+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
+}

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 {
+func (c *BootstrapData) Validate() error {
-	if c.Header.Hash() != checkpointHash {
-		return errors.New("wrong checkpoint hash")
-	}
 	if c.CommitteeRoot != c.Committee.Root() {
 		return errors.New("wrong committee root")
 	}

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
+	serverHeads   map[request.Server]common.Hash
-	headTracker   *light.HeadTracker
+	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 header := s.headTracker.ValidatedHead().Header; header != (types.Header{}) {
-	if s.validatedHead == (types.Header{}) {
+		s.validatedHead = header.Hash()
-		return false
 	}
 
 	// 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 {
+		if event.Timeout || event.Finalized {
-			// unlock if timed out or returned with an invalid response without
+			// unlock if timed out or returned with an invalid response
-			// previously being unlocked by a timeout
 			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
 	}

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 <http://www.gnu.org/licenses/>.
+
+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
+}