diff --git a/beacon/light/sync/head_sync.go b/beacon/light/sync/head_sync.go
index d0c19ea5d6..d6ab9e2849 100644
--- a/beacon/light/sync/head_sync.go
+++ b/beacon/light/sync/head_sync.go
@@ -29,15 +29,15 @@ type headTracker interface {
}
type HeadSync struct {
- headTracker headTracker
- chain committeeChain
- nextSyncPeriod uint64
- chainInit bool
- queuedHeads map[any][]types.SignedHeader
- serverHeads map[any]types.HeadInfo
- headServerCount map[types.HeadInfo]headServerCount
- headCounter uint64
- prefetchHead types.HeadInfo
+ headTracker headTracker
+ chain committeeChain
+ nextSyncPeriod uint64
+ chainInit bool
+ unvalidatedHeads map[any]types.SignedHeader
+ serverHeads map[any]types.HeadInfo
+ headServerCount map[types.HeadInfo]headServerCount
+ headCounter uint64
+ prefetchHead types.HeadInfo
}
type headServerCount struct {
@@ -47,12 +47,12 @@ type headServerCount struct {
func NewHeadSync(headTracker headTracker, chain committeeChain) *HeadSync {
s := &HeadSync{
- headTracker: headTracker,
- chain: chain,
- nextSyncPeriod: math.MaxUint64,
- queuedHeads: make(map[any][]types.SignedHeader),
- serverHeads: make(map[any]types.HeadInfo),
- headServerCount: make(map[types.HeadInfo]headServerCount),
+ headTracker: headTracker,
+ chain: chain,
+ nextSyncPeriod: math.MaxUint64,
+ unvalidatedHeads: make(map[any]types.SignedHeader),
+ serverHeads: make(map[any]types.HeadInfo),
+ headServerCount: make(map[types.HeadInfo]headServerCount),
}
return s
}
@@ -62,46 +62,43 @@ 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.processQueuedHeads()
+ s.processUnvalidatedHeadsHeads()
}
for _, event := range serverEvents {
switch event.Type {
case EvNewHead:
- trigger = trigger || s.setServerHead(event.Server, event.Data.(types.HeadInfo))
+ if s.setServerHead(event.Server, event.Data.(types.HeadInfo)) {
+ trigger = true
+ }
case EvNewSignedHead:
s.newSignedHead(event.Server, event.Data.(types.SignedHeader))
case request.EvUnregistered:
- trigger = trigger || s.setServerHead(event.Server, types.HeadInfo{})
+ if s.setServerHead(event.Server, types.HeadInfo{}) {
+ trigger = true
+ }
delete(s.serverHeads, event.Server)
- delete(s.queuedHeads, event.Server)
+ delete(s.unvalidatedHeads, event.Server)
}
}
return
}
func (s *HeadSync) newSignedHead(server any, signedHead types.SignedHeader) {
- if signedHead.Header.SyncPeriod() > s.nextSyncPeriod {
- s.queuedHeads[server] = append(s.queuedHeads[server], signedHead)
+ if !s.chainInit || signedHead.Header.SyncPeriod() > s.nextSyncPeriod {
+ s.unvalidatedHeads[server] = signedHead
return
}
s.headTracker.Validate(signedHead)
}
-func (s *HeadSync) processQueuedHeads() {
- for server, queued := range s.queuedHeads {
- j := len(queued)
- for i := len(queued) - 1; i >= 0; i-- {
- if signedHead := queued[i]; types.SyncPeriod(signedHead.SignatureSlot) <= s.nextSyncPeriod {
- s.headTracker.Validate(signedHead)
- } else {
- j--
- if j != i {
- queued[j] = queued[i]
- }
- }
- }
- if j != 0 {
- s.queuedHeads[server] = queued[j:]
+func (s *HeadSync) processUnvalidatedHeadsHeads() {
+ if !s.chainInit {
+ return
+ }
+ for server, signedHead := range s.unvalidatedHeads {
+ if types.SyncPeriod(signedHead.SignatureSlot) <= s.nextSyncPeriod {
+ s.headTracker.Validate(signedHead)
+ delete(s.unvalidatedHeads, server)
}
}
}
@@ -129,6 +126,9 @@ func (s *HeadSync) setServerHead(server any, head types.HeadInfo) bool {
}
h.serverCount++
s.headServerCount[head] = h
+ s.serverHeads[server] = head
+ } else {
+ delete(s.serverHeads, server)
}
var (
bestHead types.HeadInfo
diff --git a/beacon/light/sync/head_sync_test.go b/beacon/light/sync/head_sync_test.go
new file mode 100644
index 0000000000..07ae15edc7
--- /dev/null
+++ b/beacon/light/sync/head_sync_test.go
@@ -0,0 +1,110 @@
+// 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 sync
+
+import (
+ "testing"
+
+ "github.com/ethereum/go-ethereum/beacon/light/request"
+ "github.com/ethereum/go-ethereum/beacon/types"
+ "github.com/ethereum/go-ethereum/common"
+)
+
+var (
+ testServer1 = 1
+ testServer2 = 2
+ testServer3 = 3
+ testServer4 = 4
+
+ testHead0 = types.HeadInfo{}
+ testHead1 = types.HeadInfo{Slot: 123, BlockRoot: common.Hash{1}}
+ testHead2 = types.HeadInfo{Slot: 124, BlockRoot: common.Hash{2}}
+ 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}}}
+)
+
+func TestValidatedHead(t *testing.T) {
+ tracker := &testTracker{}
+ chain := &testCommitteeChain{}
+ ht := &testHeadTracker{}
+ headSync := NewHeadSync(ht, chain)
+
+ ht.expValidated(t, 1, nil)
+ 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})
+}
+
+func TestPrefetchHead(t *testing.T) {
+ tracker := &testTracker{}
+ chain := &testCommitteeChain{}
+ ht := &testHeadTracker{}
+ headSync := NewHeadSync(ht, chain)
+
+ ht.expPrefetch(t, 1, testHead0) // no servers registered
+ 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{
+ {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{
+ {Server: testServer1, Type: EvNewHead, Data: testHead2},
+ })
+ ht.expPrefetch(t, 4, testHead2) // s1: h2, s2: h2
+ 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{
+ {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{
+ {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{
+ {Server: testServer3, Type: request.EvUnregistered},
+ })
+ ht.expPrefetch(t, 8, testHead4) // s1: h2, s2: h3, s4: h4
+ 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{
+ {Server: testServer4, Type: request.EvUnregistered},
+ })
+ ht.expPrefetch(t, 10, testHead3) // s2: h3
+ headSync.Process(tracker, nil, []request.ServerEvent{
+ {Server: testServer2, Type: request.EvUnregistered},
+ })
+ ht.expPrefetch(t, 11, testHead0) // no servers registered
+}
diff --git a/beacon/light/sync/test_helpers.go b/beacon/light/sync/test_helpers.go
new file mode 100644
index 0000000000..fb4f435f4b
--- /dev/null
+++ b/beacon/light/sync/test_helpers.go
@@ -0,0 +1,88 @@
+// 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 sync
+
+import (
+ "testing"
+
+ "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
+}
+
+func (t *testTracker) InvalidResponse(id request.ServerAndId, desc string) {
+ return
+}
+
+type testCommitteeChain struct {
+ nsp uint64
+ init bool
+}
+
+func (t *testCommitteeChain) CheckpointInit(bootstrap types.BootstrapData) error {
+ return nil
+}
+
+func (t *testCommitteeChain) InsertUpdate(update *types.LightClientUpdate, nextCommittee *types.SerializedSyncCommittee) error {
+ return nil
+}
+
+func (t *testCommitteeChain) NextSyncPeriod() (uint64, bool) {
+ return t.nsp, t.init
+}
+
+type testHeadTracker struct {
+ phead types.HeadInfo
+ validated []types.SignedHeader
+}
+
+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) {
+ 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())
+ }
+ if ht.validated[i] != expHead {
+ vhead := ht.validated[i].Header
+ t.Errorf("Wrong validated head in test case #%d index #%d (expected {slot %d blockRoot %x}, got {slot %d blockRoot %x})", tci, i, expHead.Header.Slot, expHead.Header.Hash(), vhead.Slot, vhead.Hash())
+ }
+ }
+ for i := len(expHeads); i < len(ht.validated); i++ {
+ vhead := ht.validated[i].Header
+ t.Errorf("Unexpected validated head in test case #%d index #%d (expected none, got {slot %d blockRoot %x})", tci, i, vhead.Slot, vhead.Hash())
+ }
+ ht.validated = nil
+}
+
+func (ht *testHeadTracker) SetPrefetchHead(head types.HeadInfo) {
+ ht.phead = head
+}
+
+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)
+ }
+}