forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

beacon/light/sync: all tests working with new Module interface

Browse source
This commit is contained in:
Zsolt Felfoldi 2024-01-17 05:38:53 +01:00 committed by Felix Lange
parent 7c173fc7a5
commit 426c1fb010
6 changed files with 205 additions and 225 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
beacon/light/sync/head_sync.go
View file

@ -17,8 +17,6 @@
package sync package sync
import ( import (
"math"
"github.com/ethereum/go-ethereum/beacon/light/request" "github.com/ethereum/go-ethereum/beacon/light/request"
"github.com/ethereum/go-ethereum/beacon/types" "github.com/ethereum/go-ethereum/beacon/types"
) )
@ -59,7 +57,6 @@ func NewHeadSync(headTracker headTracker, chain committeeChain) *HeadSync {
s := &HeadSync{ s := &HeadSync{
headTracker: headTracker, headTracker: headTracker,
chain: chain, chain: chain,
nextSyncPeriod: math.MaxUint64,
unvalidatedHeads: make(map[request.Server]types.SignedHeader), unvalidatedHeads: make(map[request.Server]types.SignedHeader),
serverHeads: make(map[request.Server]types.HeadInfo), serverHeads: make(map[request.Server]types.HeadInfo),
headServerCount: make(map[types.HeadInfo]headServerCount), headServerCount: make(map[types.HeadInfo]headServerCount),
@ -105,6 +102,9 @@ func (s *HeadSync) newSignedHead(server request.Server, signedHead types.SignedH
// processUnvalidatedHeads iterates the list of unvalidated heads and validates // processUnvalidatedHeads iterates the list of unvalidated heads and validates
// those which can be validated. // those which can be validated.
func (s *HeadSync) processUnvalidatedHeads() { func (s *HeadSync) processUnvalidatedHeads() {
if !s.chainInit {
return
}
for server, signedHead := range s.unvalidatedHeads { for server, signedHead := range s.unvalidatedHeads {
if types.SyncPeriod(signedHead.SignatureSlot) <= s.nextSyncPeriod { if types.SyncPeriod(signedHead.SignatureSlot) <= s.nextSyncPeriod {
s.headTracker.Validate(signedHead) s.headTracker.Validate(signedHead)

166
beacon/light/sync/head_sync_test.go
View file

@ -19,16 +19,15 @@ package sync
import ( import (
"testing" "testing"
"github.com/ethereum/go-ethereum/beacon/light/request"
"github.com/ethereum/go-ethereum/beacon/types" "github.com/ethereum/go-ethereum/beacon/types"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
) )
var ( var (
testServer1 = 1 testServer1 = &TestServer{ID: 1}
testServer2 = 2 testServer2 = &TestServer{ID: 2}
testServer3 = 3 testServer3 = &TestServer{ID: 3}
testServer4 = 4 testServer4 = &TestServer{ID: 4}
testHead0 = types.HeadInfo{} testHead0 = types.HeadInfo{}
testHead1 = types.HeadInfo{Slot: 123, BlockRoot: common.Hash{1}} testHead1 = types.HeadInfo{Slot: 123, BlockRoot: common.Hash{1}}
@ -47,93 +46,98 @@ func TestValidatedHead(t *testing.T) {
chain := &TestCommitteeChain{} chain := &TestCommitteeChain{}
ht := &TestHeadTracker{} ht := &TestHeadTracker{}
headSync := NewHeadSync(ht, chain) headSync := NewHeadSync(ht, chain)
ts := NewTestScheduler(t, headSync)
ht.ExpValidated(t, 0, nil)
ts.AddServer(testServer1, 1)
ts.ServerEvent(EvNewSignedHead, testServer1, testSHead1)
ts.Run(1, nil, nil)
ht.ExpValidated(t, 1, nil) ht.ExpValidated(t, 1, nil)
headSync.Process([]request.Event{
{Server: testServer1, Type: request.EvRegistered},
{Server: testServer1, Type: EvNewSignedHead, Data: testSHead1},
})
ht.ExpValidated(t, 2, nil)
chain.SetNextSyncPeriod(0) chain.SetNextSyncPeriod(0)
headSync.Process(nil) ts.Run(2, nil, nil)
ht.ExpValidated(t, 3, []types.SignedHeader{testSHead1}) ht.ExpValidated(t, 2, []types.SignedHeader{testSHead1})
chain.SetNextSyncPeriod(1) chain.SetNextSyncPeriod(1)
headSync.Process([]request.Event{ ts.ServerEvent(EvNewSignedHead, testServer1, testSHead2)
{Server: testServer1, Type: EvNewSignedHead, Data: testSHead2}, ts.AddServer(testServer2, 1)
{Server: testServer2, Type: request.EvRegistered}, ts.ServerEvent(EvNewSignedHead, testServer2, testSHead2)
{Server: testServer2, Type: EvNewSignedHead, Data: testSHead2}, ts.Run(3, nil, nil)
}) ht.ExpValidated(t, 3, []types.SignedHeader{testSHead2, testSHead2})
ht.ExpValidated(t, 4, []types.SignedHeader{testSHead2, testSHead2})
headSync.Process([]request.Event{ ts.ServerEvent(EvNewSignedHead, testServer1, testSHead3)
{Server: testServer1, Type: EvNewSignedHead, Data: testSHead3}, ts.AddServer(testServer3, 1)
{Server: testServer3, Type: request.EvRegistered}, ts.ServerEvent(EvNewSignedHead, testServer3, testSHead4)
{Server: testServer3, Type: EvNewSignedHead, Data: testSHead4}, ts.Run(4, nil, nil)
}) ht.ExpValidated(t, 4, nil)
ht.ExpValidated(t, 5, nil)
chain.SetNextSyncPeriod(2) chain.SetNextSyncPeriod(2)
headSync.Process(nil) ts.Run(5, nil, nil)
ht.ExpValidated(t, 6, []types.SignedHeader{testSHead3}) ht.ExpValidated(t, 5, []types.SignedHeader{testSHead3})
headSync.Process([]request.Event{
{Server: testServer3, Type: request.EvUnregistered}, ts.RemoveServer(testServer3)
}) ts.Run(6, nil, nil)
ht.ExpValidated(t, 7, nil) ht.ExpValidated(t, 6, nil)
chain.SetNextSyncPeriod(3) chain.SetNextSyncPeriod(3)
headSync.Process(nil) ts.Run(7, nil, nil)
ht.ExpValidated(t, 8, nil) ht.ExpValidated(t, 7, nil)
headSync.Process([]request.Event{
{Server: testServer2, Type: EvNewSignedHead, Data: testSHead4}, ts.ServerEvent(EvNewSignedHead, testServer2, testSHead4)
}) ts.Run(8, nil, nil)
ht.ExpValidated(t, 9, []types.SignedHeader{testSHead4}) ht.ExpValidated(t, 8, []types.SignedHeader{testSHead4})
} }
func TestPrefetchHead(t *testing.T) { func TestPrefetchHead(t *testing.T) {
chain := &TestCommitteeChain{} chain := &TestCommitteeChain{}
ht := &TestHeadTracker{} ht := &TestHeadTracker{}
headSync := NewHeadSync(ht, chain) headSync := NewHeadSync(ht, chain)
ts := NewTestScheduler(t, headSync)
ht.ExpPrefetch(t, 1, testHead0) // no servers registered ht.ExpPrefetch(t, 0, testHead0) // no servers registered
headSync.Process([]request.Event{
{Server: testServer1, Type: request.EvRegistered}, ts.AddServer(testServer1, 1)
{Server: testServer1, Type: EvNewHead, Data: testHead1}, ts.ServerEvent(EvNewHead, testServer1, testHead1)
}) ts.Run(1, nil, nil)
ht.ExpPrefetch(t, 2, testHead1) // s1: h1 ht.ExpPrefetch(t, 1, testHead1) // s1: h1
headSync.Process([]request.Event{
{Server: testServer2, Type: request.EvRegistered}, ts.AddServer(testServer2, 1)
{Server: testServer2, Type: EvNewHead, Data: testHead2}, ts.ServerEvent(EvNewHead, testServer2, testHead2)
}) ts.Run(2, nil, nil)
ht.ExpPrefetch(t, 3, testHead2) // s1: h1, s2: h2 ht.ExpPrefetch(t, 2, testHead2) // s1: h1, s2: h2
headSync.Process([]request.Event{
{Server: testServer1, Type: EvNewHead, Data: testHead2}, ts.ServerEvent(EvNewHead, testServer1, testHead2)
}) ts.Run(3, nil, nil)
ht.ExpPrefetch(t, 4, testHead2) // s1: h2, s2: h2 ht.ExpPrefetch(t, 3, testHead2) // s1: h2, s2: h2
headSync.Process([]request.Event{
{Server: testServer3, Type: request.EvRegistered}, ts.AddServer(testServer3, 1)
{Server: testServer3, Type: EvNewHead, Data: testHead3}, ts.ServerEvent(EvNewHead, testServer3, testHead3)
}) ts.Run(4, nil, nil)
ht.ExpPrefetch(t, 5, testHead2) // s1: h2, s2: h2, s3: h3 ht.ExpPrefetch(t, 4, testHead2) // s1: h2, s2: h2, s3: h3
headSync.Process([]request.Event{
{Server: testServer4, Type: request.EvRegistered}, ts.AddServer(testServer4, 1)
{Server: testServer4, Type: EvNewHead, Data: testHead4}, ts.ServerEvent(EvNewHead, testServer4, testHead4)
}) ts.Run(5, nil, nil)
ht.ExpPrefetch(t, 6, testHead2) // s1: h2, s2: h2, s3: h3, s4: h4 ht.ExpPrefetch(t, 5, testHead2) // s1: h2, s2: h2, s3: h3, s4: h4
headSync.Process([]request.Event{
{Server: testServer2, Type: EvNewHead, Data: testHead3}, ts.ServerEvent(EvNewHead, testServer2, testHead3)
}) ts.Run(6, nil, nil)
ht.ExpPrefetch(t, 7, testHead3) // s1: h2, s2: h3, s3: h3, s4: h4 ht.ExpPrefetch(t, 6, testHead3) // s1: h2, s2: h3, s3: h3, s4: h4
headSync.Process([]request.Event{
{Server: testServer3, Type: request.EvUnregistered}, ts.RemoveServer(testServer3)
}) ts.Run(7, nil, nil)
ht.ExpPrefetch(t, 8, testHead4) // s1: h2, s2: h3, s4: h4 ht.ExpPrefetch(t, 7, testHead4) // s1: h2, s2: h3, s4: h4
headSync.Process([]request.Event{
{Server: testServer1, Type: request.EvUnregistered}, ts.RemoveServer(testServer1)
}) ts.Run(8, nil, nil)
ht.ExpPrefetch(t, 9, testHead4) // s2: h3, s4: h4 ht.ExpPrefetch(t, 8, testHead4) // s2: h3, s4: h4
headSync.Process([]request.Event{
{Server: testServer4, Type: request.EvUnregistered}, ts.RemoveServer(testServer4)
}) ts.Run(9, nil, nil)
ht.ExpPrefetch(t, 10, testHead3) // s2: h3 ht.ExpPrefetch(t, 9, testHead3) // s2: h3
headSync.Process([]request.Event{
{Server: testServer2, Type: request.EvUnregistered}, ts.RemoveServer(testServer2)
}) ts.Run(10, nil, nil)
ht.ExpPrefetch(t, 11, testHead0) // no servers registered ht.ExpPrefetch(t, 10, testHead0) // no servers registered
} }

23
beacon/light/sync/test_helpers.go
View file

@ -62,14 +62,18 @@ func (ts *TestScheduler) Run(testIndex int, expServer request.Server, expReq req
return return
} }
if !ok { if !ok {
ts.t.Errorf("Missing request in test case #%d (expected none, got %v)", testIndex, expReqWithID) ts.t.Errorf("Missing request in test case #%d (expected %v, got none)", testIndex, expReqWithID)
return return
} }
if req != expReqWithID { if req != expReqWithID {
ts.t.Errorf("Wrong request in test case #%d (expected %v, got %v)", testIndex, req, expReqWithID) ts.t.Errorf("Wrong request in test case #%d (expected %v, got %v)", testIndex, expReqWithID, req)
} }
} }
func (ts *TestScheduler) Request(testIndex int) request.RequestWithID {
return ts.sent[testIndex]
}
func (ts *TestScheduler) ServerEvent(evType *request.EventType, server request.Server, data any) { func (ts *TestScheduler) ServerEvent(evType *request.EventType, server request.Server, data any) {
ts.events = append(ts.events, request.Event{ ts.events = append(ts.events, request.Event{
Type: evType, Type: evType,
@ -98,6 +102,20 @@ func (ts *TestScheduler) RequestEvent(evType *request.EventType, testIndex int,
func (ts *TestScheduler) AddServer(server request.Server, allowance int) { func (ts *TestScheduler) AddServer(server request.Server, allowance int) {
ts.servers = append(ts.servers, server) ts.servers = append(ts.servers, server)
ts.allowance[server] = allowance ts.allowance[server] = allowance
ts.ServerEvent(request.EvRegistered, server, nil)
}
func (ts *TestScheduler) RemoveServer(server request.Server) {
ts.servers = append(ts.servers, server)
for i, s := range ts.servers {
if s == server {
copy(ts.servers[i:len(ts.servers)-1], ts.servers[i+1:])
ts.servers = ts.servers[:len(ts.servers)-1]
break
}
}
delete(ts.allowance, server)
ts.ServerEvent(request.EvUnregistered, server, nil)
} }
func (ts *TestScheduler) AddAllowance(server request.Server, allowance int) { func (ts *TestScheduler) AddAllowance(server request.Server, allowance int) {
@ -129,6 +147,7 @@ func (ts *TestScheduler) tryRequest(testIndex int, requestFn func(server request
Request: bestReq, Request: bestReq,
} }
ts.sent[testIndex] = req ts.sent[testIndex] = req
ts.RequestEvent(request.EvRequest, testIndex, nil)
return req, true return req, true
} }

4
beacon/light/sync/update_sync.go
View file

@ -55,12 +55,12 @@ func NewCheckpointInit(chain committeeChain, checkpointHash common.Hash) *Checkp
func (s *CheckpointInit) Process(events []request.Event) { func (s *CheckpointInit) Process(events []request.Event) {
for _, event := range events { for _, event := range events {
if !event.IsRequestEvent() { if !event.IsRequestEvent() {
return continue
} }
sid, req, resp := event.RequestInfo() sid, req, resp := event.RequestInfo()
if event.Type == request.EvRequest { if event.Type == request.EvRequest {
s.locked = sid s.locked = sid
return continue
} }
if s.locked == sid { if s.locked == sid {
s.locked = request.ServerAndID{} s.locked = request.ServerAndID{}

229
beacon/light/sync/update_sync_test.go
View file

@ -33,41 +33,30 @@ func TestCheckpointInit(t *testing.T) {
ts.AddServer(testServer1, 1) ts.AddServer(testServer1, 1)
ts.AddServer(testServer2, 1) ts.AddServer(testServer2, 1)
chkInit.Process([]request.Event{
{Server: testServer1, Type: request.EvRegistered},
{Server: testServer2, Type: request.EvRegistered},
})
// expect bootstrap request to server 1 // expect bootstrap request to server 1
req1 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 1}, Request: ReqCheckpointData(checkpointHash)} ts.Run(1, testServer1, ReqCheckpointData(checkpointHash))
ts.ExpRequests(t, 1, []request.RequestWithID{req1})
// req1 times out; expect request to server 2 // server 1 times out; expect request to server 2
chkInit.Process([]request.Event{ ts.RequestEvent(request.EvTimeout, 1, nil)
TestReqEvent(request.EvTimeout, req1, nil), ts.Run(2, testServer2, ReqCheckpointData(checkpointHash))
})
req2 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 2}, Request: ReqCheckpointData(checkpointHash)} // invalid response to server 2; expect init state to still be false
ts.ExpRequests(t, 2, []request.RequestWithID{req2}) ts.RequestEvent(request.EvResponse, 2, &types.BootstrapData{Header: types.Header{Slot: 123456}})
// invalid response to req2; expect init state to still be false ts.Run(3, nil, nil)
wrongCheckpoint := &types.BootstrapData{Header: types.Header{Slot: 123456}}
chkInit.Process([]request.Event{
TestReqEvent(request.EvResponse, req2, wrongCheckpoint),
})
// req1 fails (hard timeout)
chkInit.Process([]request.Event{
TestReqEvent(request.EvFail, req1, nil),
})
chain.ExpInit(t, false) chain.ExpInit(t, false)
// server 3 is registered
// server 1 fails (hard timeout)
ts.RequestEvent(request.EvFail, 1, nil)
ts.Run(4, nil, nil)
chain.ExpInit(t, false)
// server 3 is registered; expect bootstrap request to server 3
ts.AddServer(testServer3, 1) ts.AddServer(testServer3, 1)
chkInit.Process([]request.Event{ ts.Run(5, testServer3, ReqCheckpointData(checkpointHash))
{Server: testServer3, Type: request.EvRegistered},
}) // valid response from server 3; expect chain to be initialized
// expect bootstrap request to server 3 ts.RequestEvent(request.EvResponse, 5, checkpoint)
req3 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer3, ID: 3}, Request: ReqCheckpointData(checkpointHash)} ts.Run(6, nil, nil)
ts.ExpRequests(t, 3, []request.RequestWithID{req3})
// valid response to req3; expect chain to be initialized
chkInit.Process([]request.Event{
TestReqEvent(request.EvResponse, req3, checkpoint),
})
chain.ExpInit(t, true) chain.ExpInit(t, true)
} }
@ -78,91 +67,73 @@ func TestUpdateSyncParallel(t *testing.T) {
ts := NewTestScheduler(t, updateSync) ts := NewTestScheduler(t, updateSync)
// add 2 servers, head at period 100; allow 3-3 parallel requests for each // add 2 servers, head at period 100; allow 3-3 parallel requests for each
ts.AddServer(testServer1, 3) ts.AddServer(testServer1, 3)
ts.ServerEvent(EvNewSignedHead, testServer1, types.SignedHeader{SignatureSlot: 0x2000*100 + 0x1000})
ts.AddServer(testServer2, 3) ts.AddServer(testServer2, 3)
ts.ServerEvent(EvNewSignedHead, testServer2, types.SignedHeader{SignatureSlot: 0x2000*100 + 0x1000})
updateSync.Process([]request.Event{
{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 // expect 6 requests to be sent
req1 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 1}, Request: ReqUpdates{FirstPeriod: 0, Count: 8}} ts.Run(1, testServer1, ReqUpdates{FirstPeriod: 0, Count: 8})
req2 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 2}, Request: ReqUpdates{FirstPeriod: 8, Count: 8}} ts.Run(2, testServer1, ReqUpdates{FirstPeriod: 8, Count: 8})
req3 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 3}, Request: ReqUpdates{FirstPeriod: 16, Count: 8}} ts.Run(3, testServer1, ReqUpdates{FirstPeriod: 16, Count: 8})
req4 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 4}, Request: ReqUpdates{FirstPeriod: 24, Count: 8}} ts.Run(4, testServer2, ReqUpdates{FirstPeriod: 24, Count: 8})
req5 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 5}, Request: ReqUpdates{FirstPeriod: 32, Count: 8}} ts.Run(5, testServer2, ReqUpdates{FirstPeriod: 32, Count: 8})
req6 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 6}, Request: ReqUpdates{FirstPeriod: 40, Count: 8}} ts.Run(6, testServer2, ReqUpdates{FirstPeriod: 40, Count: 8})
ts.ExpRequests(t, 1, []request.RequestWithID{req1, req2, req3, req4, req5, req6})
// valid response to request 1 // valid response to request 1; expect 8 periods synced and a new request started
ts.RequestEvent(request.EvResponse, 1, testRespUpdate(ts.Request(1)))
ts.AddAllowance(testServer1, 1) ts.AddAllowance(testServer1, 1)
updateSync.Process([]request.Event{ ts.Run(7, testServer1, ReqUpdates{FirstPeriod: 48, Count: 8})
TestReqEvent(request.EvResponse, req1, testRespUpdate(req1)),
})
// expect 8 periods synced and a new request started
chain.ExpNextSyncPeriod(t, 8) chain.ExpNextSyncPeriod(t, 8)
req7 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 7}, Request: ReqUpdates{FirstPeriod: 48, Count: 8}}
ts.ExpRequests(t, 2, []request.RequestWithID{req7})
// valid response to requests 4 and 5 // valid response to requests 4 and 5
ts.RequestEvent(request.EvResponse, 4, testRespUpdate(ts.Request(4)))
ts.RequestEvent(request.EvResponse, 5, testRespUpdate(ts.Request(5)))
ts.AddAllowance(testServer2, 2) ts.AddAllowance(testServer2, 2)
updateSync.Process([]request.Event{
TestReqEvent(request.EvResponse, req4, testRespUpdate(req4)),
TestReqEvent(request.EvResponse, req5, testRespUpdate(req5)),
})
// expect 2 more requests but no sync progress (responses 4 and 5 cannot be added before 2 and 3) // expect 2 more requests but no sync progress (responses 4 and 5 cannot be added before 2 and 3)
ts.Run(8, testServer2, ReqUpdates{FirstPeriod: 56, Count: 8})
ts.Run(9, testServer2, ReqUpdates{FirstPeriod: 64, Count: 8})
chain.ExpNextSyncPeriod(t, 8) 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}}
ts.ExpRequests(t, 3, []request.RequestWithID{req8, req9})
// soft timeout for requests 2 and 3 (server 1 is overloaded) // soft timeout for requests 2 and 3 (server 1 is overloaded)
updateSync.Process([]request.Event{ ts.RequestEvent(request.EvTimeout, 2, nil)
TestReqEvent(request.EvTimeout, req2, nil), ts.RequestEvent(request.EvTimeout, 3, nil)
TestReqEvent(request.EvTimeout, req3, nil),
})
// no allowance, no more requests // no allowance, no more requests
ts.ExpRequests(t, 4, nil) ts.Run(10, nil, nil)
// valid response to requests 6 and 8 and 9 // valid response to requests 6 and 8 and 9
ts.RequestEvent(request.EvResponse, 6, testRespUpdate(ts.Request(6)))
ts.RequestEvent(request.EvResponse, 8, testRespUpdate(ts.Request(8)))
ts.RequestEvent(request.EvResponse, 9, testRespUpdate(ts.Request(9)))
ts.AddAllowance(testServer2, 3) ts.AddAllowance(testServer2, 3)
updateSync.Process([]request.Event{
TestReqEvent(request.EvResponse, req6, testRespUpdate(req6)),
TestReqEvent(request.EvResponse, req8, testRespUpdate(req8)),
TestReqEvent(request.EvResponse, req9, testRespUpdate(req9)),
})
// server 2 can now resend requests 2 and 3 (timed out by server 1) and also send a new one // 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}} ts.Run(11, testServer2, ReqUpdates{FirstPeriod: 8, Count: 8})
req3r := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 11}, Request: ReqUpdates{FirstPeriod: 16, Count: 8}} ts.Run(12, testServer2, ReqUpdates{FirstPeriod: 16, Count: 8})
req10 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 12}, Request: ReqUpdates{FirstPeriod: 72, Count: 8}} ts.Run(13, testServer2, ReqUpdates{FirstPeriod: 72, Count: 8})
ts.ExpRequests(t, 5, []request.RequestWithID{req2r, req3r, req10})
// server 1 finally answers timed out request 2 // server 1 finally answers timed out request 2
ts.RequestEvent(request.EvResponse, 2, testRespUpdate(ts.Request(2)))
ts.AddAllowance(testServer1, 1) ts.AddAllowance(testServer1, 1)
updateSync.Process([]request.Event{
TestReqEvent(request.EvResponse, req2, testRespUpdate(req2)),
})
// expect sync progress and one new request // expect sync progress and one new request
ts.Run(14, testServer1, ReqUpdates{FirstPeriod: 80, Count: 8})
chain.ExpNextSyncPeriod(t, 16) chain.ExpNextSyncPeriod(t, 16)
req11 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 13}, Request: ReqUpdates{FirstPeriod: 80, Count: 8}}
ts.ExpRequests(t, 6, []request.RequestWithID{req11}) // server 2 answers requests 11 and 12 (resends of requests 2 and 3)
// server 2 answers re-sent requests 2 and 3 ts.RequestEvent(request.EvResponse, 11, testRespUpdate(ts.Request(11)))
ts.RequestEvent(request.EvResponse, 12, testRespUpdate(ts.Request(12)))
ts.AddAllowance(testServer2, 2) ts.AddAllowance(testServer2, 2)
updateSync.Process([]request.Event{ ts.Run(15, testServer2, ReqUpdates{FirstPeriod: 88, Count: 8})
TestReqEvent(request.EvResponse, req2r, testRespUpdate(req2r)), ts.Run(16, testServer2, ReqUpdates{FirstPeriod: 96, Count: 4})
TestReqEvent(request.EvResponse, req3r, testRespUpdate(req3r)),
})
// finally the gap is filled, update can process responses up to req6 // finally the gap is filled, update can process responses up to req6
chain.ExpNextSyncPeriod(t, 48) 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}}
ts.ExpRequests(t, 7, []request.RequestWithID{req12, req13})
// all remaining requests are answered // all remaining requests are answered
updateSync.Process([]request.Event{ ts.RequestEvent(request.EvResponse, 3, testRespUpdate(ts.Request(3)))
TestReqEvent(request.EvResponse, req3, testRespUpdate(req3)), ts.RequestEvent(request.EvResponse, 7, testRespUpdate(ts.Request(7)))
TestReqEvent(request.EvResponse, req7, testRespUpdate(req7)), ts.RequestEvent(request.EvResponse, 13, testRespUpdate(ts.Request(13)))
TestReqEvent(request.EvResponse, req10, testRespUpdate(req10)), ts.RequestEvent(request.EvResponse, 14, testRespUpdate(ts.Request(14)))
TestReqEvent(request.EvResponse, req11, testRespUpdate(req11)), ts.RequestEvent(request.EvResponse, 15, testRespUpdate(ts.Request(15)))
TestReqEvent(request.EvResponse, req12, testRespUpdate(req12)), ts.RequestEvent(request.EvResponse, 16, testRespUpdate(ts.Request(16)))
TestReqEvent(request.EvResponse, req13, testRespUpdate(req13)), ts.Run(17, nil, nil)
})
// expect chain to be fully synced // expect chain to be fully synced
chain.ExpNextSyncPeriod(t, 100) chain.ExpNextSyncPeriod(t, 100)
} }
@ -174,65 +145,53 @@ func TestUpdateSyncDifferentHeads(t *testing.T) {
ts := NewTestScheduler(t, updateSync) ts := NewTestScheduler(t, updateSync)
// add 3 servers with different announced head periods // add 3 servers with different announced head periods
ts.AddServer(testServer1, 1) ts.AddServer(testServer1, 1)
ts.ServerEvent(EvNewSignedHead, testServer1, types.SignedHeader{SignatureSlot: 0x2000*15 + 0x1000})
ts.AddServer(testServer2, 1) ts.AddServer(testServer2, 1)
ts.ServerEvent(EvNewSignedHead, testServer2, types.SignedHeader{SignatureSlot: 0x2000*16 + 0x1000})
ts.AddServer(testServer3, 1) ts.AddServer(testServer3, 1)
ts.ServerEvent(EvNewSignedHead, testServer3, types.SignedHeader{SignatureSlot: 0x2000*17 + 0x1000})
updateSync.Process([]request.Event{
{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 // expect request to the best announced head
req1 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer3, ID: 1}, Request: ReqUpdates{FirstPeriod: 10, Count: 7}} ts.Run(1, testServer3, ReqUpdates{FirstPeriod: 10, Count: 7})
ts.ExpRequests(t, 1, []request.RequestWithID{req1})
// request times out, expect request to the next best head // request times out, expect request to the next best head
updateSync.Process([]request.Event{ ts.RequestEvent(request.EvTimeout, 1, nil)
TestReqEvent(request.EvTimeout, req1, nil), ts.Run(2, testServer2, ReqUpdates{FirstPeriod: 10, Count: 6})
})
req2 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer2, ID: 2}, Request: ReqUpdates{FirstPeriod: 10, Count: 6}}
ts.ExpRequests(t, 2, []request.RequestWithID{req2})
// request times out, expect request to the last available server // request times out, expect request to the last available server
updateSync.Process([]request.Event{ ts.RequestEvent(request.EvTimeout, 2, nil)
TestReqEvent(request.EvTimeout, req2, nil), ts.Run(3, testServer1, ReqUpdates{FirstPeriod: 10, Count: 5})
})
req3 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer1, ID: 3}, Request: ReqUpdates{FirstPeriod: 10, Count: 5}}
ts.ExpRequests(t, 3, []request.RequestWithID{req3})
// valid response to request 3, expect chain synced to period 15 // valid response to request 3, expect chain synced to period 15
ts.RequestEvent(request.EvResponse, 3, testRespUpdate(ts.Request(3)))
ts.AddAllowance(testServer1, 1) ts.AddAllowance(testServer1, 1)
updateSync.Process([]request.Event{ ts.Run(4, nil, nil)
TestReqEvent(request.EvResponse, req3, testRespUpdate(req3)),
})
chain.ExpNextSyncPeriod(t, 15) chain.ExpNextSyncPeriod(t, 15)
// invalid response to request 1, server can only deliver updates up to period 15 despite announced head // 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}} req1x := ts.Request(1)
updateSync.Process([]request.Event{ req1x.Request = ReqUpdates{FirstPeriod: 10, Count: 5}
TestReqEvent(request.EvResponse, req1, testRespUpdate(req1x)), ts.RequestEvent(request.EvResponse, 1, testRespUpdate(req1x))
}) ts.Run(5, nil, nil)
// expect no progress of chain head // expect no progress of chain head
chain.ExpNextSyncPeriod(t, 15) chain.ExpNextSyncPeriod(t, 15)
// valid response to request 2, expect chain synced to period 16 // valid response to request 2, expect chain synced to period 16
ts.RequestEvent(request.EvResponse, 2, testRespUpdate(ts.Request(2)))
ts.AddAllowance(testServer2, 1) ts.AddAllowance(testServer2, 1)
updateSync.Process([]request.Event{ ts.Run(6, nil, nil)
TestReqEvent(request.EvResponse, req2, testRespUpdate(req2)),
})
chain.ExpNextSyncPeriod(t, 16) chain.ExpNextSyncPeriod(t, 16)
// a new server is registered with announced head period 17 // a new server is registered with announced head period 17
ts.AddServer(testServer4, 1) ts.AddServer(testServer4, 1)
updateSync.Process([]request.Event{ ts.ServerEvent(EvNewSignedHead, testServer4, types.SignedHeader{SignatureSlot: 0x2000*17 + 0x1000})
{Server: testServer4, Type: request.EvRegistered},
{Server: testServer4, Type: EvNewSignedHead, Data: types.SignedHeader{SignatureSlot: 0x2000*17 + 0x1000}},
})
// expect request to sync one more period // expect request to sync one more period
req4 := request.RequestWithID{ServerAndID: request.ServerAndID{Server: testServer4, ID: 4}, Request: ReqUpdates{FirstPeriod: 16, Count: 1}} ts.Run(7, testServer4, ReqUpdates{FirstPeriod: 16, Count: 1})
ts.ExpRequests(t, 4, []request.RequestWithID{req4})
// valid response, expect chain synced to period 17 // valid response, expect chain synced to period 17
ts.AddAllowance(testServer1, 1) ts.RequestEvent(request.EvResponse, 7, testRespUpdate(ts.Request(7)))
updateSync.Process([]request.Event{ ts.AddAllowance(testServer4, 1)
TestReqEvent(request.EvResponse, req4, testRespUpdate(req4)), ts.Run(8, nil, nil)
})
chain.ExpNextSyncPeriod(t, 17) chain.ExpNextSyncPeriod(t, 17)
} }

2
cmd/blsync/block_sync_test.go
View file

@ -56,8 +56,6 @@ func TestBlockSync(t *testing.T) {
} }
} }
ts.ServerEvent(request.EvRegistered, testServer1, nil)
ts.ServerEvent(request.EvRegistered, testServer2, nil)
// no block requests expected until head tracker knows about a head // no block requests expected until head tracker knows about a head
ts.Run(1, nil, nil) ts.Run(1, nil, nil)
expHeadBlock(1, nil) expHeadBlock(1, nil)