diff --git a/beacon/light/request/scheduler.go b/beacon/light/request/scheduler.go index 498313eb18..173fce9f1a 100644 --- a/beacon/light/request/scheduler.go +++ b/beacon/light/request/scheduler.go @@ -17,7 +17,6 @@ package request import ( - "math" "sync" "github.com/ethereum/go-ethereum/log" @@ -45,19 +44,13 @@ type Module interface { // Note: Process functions of different modules are never called concurrently; // they are called by Scheduler in the same order of priority as they were // registered in. - Process([]Event) - // MakeRequest generates a request candidate based on the state of the target - // structure(s) and the internal state of the module. This candidate is - // typically the next obtainable item (or range of items) of the target - // structure that is assumed to be available at the given server and has not - // been requested yet (or has been requested but already timed out and should - // be resent). - // MakeRequest is always called after Process. Note that it is the Scheduler's - // job to select the best possible requests and actually send them. If a - // request has been sent, the module is notified through an EvRequest event - // which also immediately triggers a next processing round, allowing modules - // to send more requests if possible and necessary. - MakeRequest(Server) (Request, float32) + Process(Requester, []Event) +} + +type Requester interface { + CanSendTo() []Server + Send(Server, Request) ID + Fail(Server, string) } // Scheduler is a modular network data retrieval framework that coordinates multiple @@ -72,7 +65,10 @@ type Scheduler struct { servers map[server]struct{} targets map[targetData]uint64 - pending map[ServerAndID]pendingRequest + requesterLock sync.RWMutex + serverOrder []server + pending map[ServerAndID]pendingRequest + // eventLock guards access to the events list. Note that eventLock can be // locked either while lock is locked or unlocked but lock cannot be locked // while eventLock is locked. @@ -87,15 +83,12 @@ type Scheduler struct { } type ( - // Server identifies a server without allowing any direct interaction except - // for reporting a server side failure. + // Server identifies a server without allowing any direct interaction. // Note: server interface is used by Scheduler and Tracker but not used by // the modules that do not interact with them directly. // In order to make module testing easier, Server interface is used in // events and modules. - Server interface { - Fail(desc string) - } + Server any Request any Response any ID uint64 @@ -238,91 +231,16 @@ func (s *Scheduler) targetChanged() (changed bool) { // requests are generated and sent if necessary and possible. func (s *Scheduler) processRound() { for { - filteredEvents := s.filterEvents() log.Debug("Processing modules") + filteredEvents := s.filterEvents() for _, module := range s.modules { log.Debug("Processing module", "name", s.names[module], "events", len(filteredEvents[module])) - module.Process(filteredEvents[module]) + module.Process(requester{s, module}, filteredEvents[module]) } if !s.targetChanged() { break } } - s.sendRequests() -} - -// sendRequests lets each module generate a request if necessary and sends it to -// a suitable server if possible. -// Note that if a request is sent, an EvRequest event will immediately trigger a -// next processing round, thereby allowing modules to create any number of requests -// in any suitable moment as long as there is a server that can accept them. -func (s *Scheduler) sendRequests() { - servers := make(map[server]struct{}) - for server := range s.servers { - if ok, _ := server.canRequestNow(); ok { - servers[server] = struct{}{} - } - } - log.Debug("Generating request candidates", "servers", len(servers)) - - for _, module := range s.modules { - if len(servers) == 0 { - return - } - if s.tryRequest(module, servers) { - log.Debug("Sent request", "module", s.names[module]) - } - } -} - -// tryRequest tries to generate request candidates for a given module and a given -// set of servers, then selects the best candidate if there is one and sends the -// request to the server it was generated for. -// The candidates are primarily ranked based on "request priority", a number that -// Module.MakeRequest has returned along with the request candidate. This ranking -// may or may not be used depending on the type of the request, identical requests -// typically have the same priority while multiple item requests may have a -// priority based on the number of items requested. -// If there are multiple candidates with identical request priority then they are -// ranked based on "server priority" which is determined by the server. This value -// is typically higher is the server is expected to respond quicker or with a -// higher chance (typically a lower number of pending requests). -// Note that tryRequest can also remove items from the set of available servers -// if they are no longer able to accept requests in the current processing round. -func (s *Scheduler) tryRequest(module Module, servers map[server]struct{}) bool { - var ( - maxServerPriority, maxRequestPriority float32 - bestServer server - bestRequest Request - ) - maxServerPriority, maxRequestPriority = -math.MaxFloat32, -math.MaxFloat32 - serverCount := len(servers) - var removed, candidates int - for server := range servers { - canRequest, serverPriority := server.canRequestNow() - if !canRequest { - delete(servers, server) - removed++ - continue - } - request, requestPriority := module.MakeRequest(server) - if request != nil { - candidates++ - } - if request == nil || requestPriority < maxRequestPriority || - (requestPriority == maxRequestPriority && serverPriority <= maxServerPriority) { - continue - } - maxServerPriority, maxRequestPriority = serverPriority, requestPriority - bestServer, bestRequest = server, request - } - log.Debug("Request attempt", "serverCount", serverCount, "removedServers", removed, "requestCandidates", candidates) - if bestServer == nil { - return false - } - sid := ServerAndID{Server: bestServer, ID: bestServer.sendRequest(bestRequest)} - s.pending[sid] = pendingRequest{request: bestRequest, module: module} - return true } // Trigger starts a new processing round. If fired during processing, it ensures @@ -358,6 +276,9 @@ func (s *Scheduler) filterEvents() map[Module][]Event { s.events = nil s.eventLock.Unlock() + s.requesterLock.Lock() + defer s.requesterLock.Unlock() + filteredEvents := make(map[Module][]Event) for _, event := range events { server := event.Server.(server) @@ -379,9 +300,19 @@ func (s *Scheduler) filterEvents() map[Module][]Event { switch event.Type { case EvRegistered: s.servers[server] = struct{}{} + s.serverOrder = append(s.serverOrder, nil) + copy(s.serverOrder[1:], s.serverOrder[:len(s.serverOrder)-1]) + s.serverOrder[0] = server case EvUnregistered: s.closePending(event.Server, filteredEvents) delete(s.servers, server) + for i, srv := range s.serverOrder { + if srv == server { + copy(s.serverOrder[i:len(s.serverOrder)-1], s.serverOrder[i+1:]) + s.serverOrder = s.serverOrder[:len(s.serverOrder)-1] + break + } + } } for _, module := range s.modules { filteredEvents[module] = append(filteredEvents[module], event) @@ -408,3 +339,44 @@ func (s *Scheduler) closePending(server Server, filteredEvents map[Module][]Even } } } + +type requester struct { + *Scheduler + module Module +} + +func (s requester) CanSendTo() []Server { + s.requesterLock.RLock() + defer s.requesterLock.RUnlock() + + list := make([]Server, 0, len(s.serverOrder)) + for _, server := range s.serverOrder { + if server.canRequestNow() { + list = append(list, server) + } + } + return list +} + +func (s requester) Send(srv Server, req Request) ID { + s.requesterLock.Lock() + defer s.requesterLock.Unlock() + + server := srv.(server) + id := server.sendRequest(req) + sid := ServerAndID{Server: srv, ID: id} + s.pending[sid] = pendingRequest{request: req, module: s.module} + for i, ss := range s.serverOrder { + if ss == server { + copy(s.serverOrder[i:len(s.serverOrder)-1], s.serverOrder[i+1:]) + s.serverOrder[len(s.serverOrder)-1] = server + return id + } + } + log.Error("Target server not found in ordered list of registered servers") + return id +} + +func (s requester) Fail(srv Server, desc string) { + srv.(server).fail(desc) +} diff --git a/beacon/light/request/server.go b/beacon/light/request/server.go index 3fc2c5fc32..16bfc2ceea 100644 --- a/beacon/light/request/server.go +++ b/beacon/light/request/server.go @@ -18,7 +18,6 @@ package request import ( "math" - "math/rand" "sync" "time" @@ -28,7 +27,6 @@ import ( var ( // request events - EvRequest = &EventType{Name: "request", requestEvent: true} // data: RequestResponse; sent by Scheduler EvResponse = &EventType{Name: "response", requestEvent: true} // data: RequestResponse; sent by requestServer EvFail = &EventType{Name: "fail", requestEvent: true} // data: RequestResponse; sent by requestServer EvTimeout = &EventType{Name: "timeout", requestEvent: true} // data: RequestResponse; sent by serverWithTimeout @@ -70,9 +68,9 @@ type requestServer interface { // new requests based on timeouts and response failures. type server interface { subscribe(eventCallback func(Event)) - canRequestNow() (bool, float32) + canRequestNow() bool sendRequest(Request) ID - Fail(string) + fail(string) unsubscribe() } @@ -121,7 +119,7 @@ type RequestResponse struct { Response Response } -// serverWithTimeout wraps a requestServer and introduces two new request event +//TODO serverWithTimeout wraps a requestServer and introduces two new request event // types: EvRequest and EvTimeout. Whenever a request is successfully sent, an // EvRequest event is emitted first. The request's lifecycle is concluded if // EvResponse or EvFail emitted by the parent requestServer. If this does not @@ -161,9 +159,7 @@ func (s *serverWithTimeout) sendRequest(request Request) (reqId ID) { s.lock.Lock() s.lastID++ id := s.lastID - reqData := RequestResponse{ID: id, Request: request} - s.childEventCb(Event{Type: EvRequest, Data: reqData}) - s.startTimeout(reqData) + s.startTimeout(RequestResponse{ID: id, Request: request}) s.lock.Unlock() s.parent.SendRequest(id, request) return id @@ -293,12 +289,12 @@ func (s *serverWithLimits) eventCallback(event Event) { s.parallelLimit += parallelAdjustUp } s.pendingCount-- - if canRequest, _ := s.canRequest(); canRequest { + if s.canRequest() { sendCanRequestAgain = s.sendEvent s.sendEvent = false } if event.Type == EvFail { - s.fail("failed request") + s.failLocked("failed request") } } childEventCb := s.childEventCb @@ -331,14 +327,14 @@ func (s *serverWithLimits) unsubscribe() { } // canRequest checks whether a new request can be started. -func (s *serverWithLimits) canRequest() (bool, float32) { +func (s *serverWithLimits) canRequest() bool { if s.delayTimer != nil || s.pendingCount >= int(s.parallelLimit) || s.timeoutCount > 0 { - return false, 0 + return false } if s.parallelLimit < minParallelLimit { s.parallelLimit = minParallelLimit } - return true, -(float32(s.pendingCount) + rand.Float32()) / s.parallelLimit + return true } // canRequestNow checks whether a new request can be started, according to the @@ -349,10 +345,10 @@ func (s *serverWithLimits) canRequest() (bool, float32) { // set of servers. // If it returns false then it is guaranteed that an EvCanRequestAgain will be // sent whenever the server becomes available for requesting again. -func (s *serverWithLimits) canRequestNow() (bool, float32) { +func (s *serverWithLimits) canRequestNow() bool { var sendCanRequestAgain bool s.lock.Lock() - canRequest, priority := s.canRequest() + canRequest := s.canRequest() if canRequest { sendCanRequestAgain = s.sendEvent s.sendEvent = false @@ -362,7 +358,7 @@ func (s *serverWithLimits) canRequestNow() (bool, float32) { if sendCanRequestAgain { childEventCb(Event{Type: EvCanRequestAgain}) } - return canRequest, priority + return canRequest } // delay sets the delay timer to the given duration, disabling new requests for @@ -384,7 +380,7 @@ func (s *serverWithLimits) delay(delay time.Duration) { s.lock.Lock() if s.delayTimer != nil && s.delayCounter == delayCounter { // do nothing if there is a new timer now s.delayTimer = nil - if canRequest, _ := s.canRequest(); canRequest { + if s.canRequest() { sendCanRequestAgain = s.sendEvent s.sendEvent = false } @@ -399,15 +395,15 @@ func (s *serverWithLimits) delay(delay time.Duration) { // fail reports that a response from the server was found invalid by the processing // Module, disabling new requests for a dynamically adjused time period. -func (s *serverWithLimits) Fail(desc string) { +func (s *serverWithLimits) fail(desc string) { s.lock.Lock() defer s.lock.Unlock() - s.fail(desc) + s.failLocked(desc) } -// fail calculates the dynamic failure delay and applies it. -func (s *serverWithLimits) fail(desc string) { +// failLocked calculates the dynamic failure delay and applies it. +func (s *serverWithLimits) failLocked(desc string) { log.Debug("Server error", "description", desc) s.failureDelay *= 2 now := s.clock.Now() diff --git a/beacon/light/sync/head_sync.go b/beacon/light/sync/head_sync.go index f97a20d17d..1474347063 100644 --- a/beacon/light/sync/head_sync.go +++ b/beacon/light/sync/head_sync.go @@ -67,7 +67,7 @@ func NewHeadSync(headTracker headTracker, chain committeeChain) *HeadSync { return s } -func (s *HeadSync) Process(events []request.Event) { +func (s *HeadSync) Process(requester request.Requester, events []request.Event) { for _, event := range events { switch event.Type { case EvNewHead: @@ -90,10 +90,6 @@ func (s *HeadSync) Process(events []request.Event) { } } -func (s *HeadSync) MakeRequest(server request.Server) (request.Request, float32) { - return nil, 0 -} - // newSignedHead handles received signed head; either validates it if the chain // is properly synced or stores it for further validation. func (s *HeadSync) newSignedHead(server request.Server, signedHead types.SignedHeader) { diff --git a/beacon/light/sync/head_sync_test.go b/beacon/light/sync/head_sync_test.go index 0451457f3d..12faad6292 100644 --- a/beacon/light/sync/head_sync_test.go +++ b/beacon/light/sync/head_sync_test.go @@ -24,10 +24,10 @@ import ( ) var ( - testServer1 = &TestServer{ID: 1} - testServer2 = &TestServer{ID: 2} - testServer3 = &TestServer{ID: 3} - testServer4 = &TestServer{ID: 4} + testServer1 = "testServer1" + testServer2 = "testServer2" + testServer3 = "testServer3" + testServer4 = "testServer4" testHead0 = types.HeadInfo{} testHead1 = types.HeadInfo{Slot: 123, BlockRoot: common.Hash{1}} @@ -52,12 +52,12 @@ func TestValidatedHead(t *testing.T) { ts.AddServer(testServer1, 1) ts.ServerEvent(EvNewSignedHead, testServer1, testSHead1) - ts.Run(1, nil, nil) + ts.Run(1) // announced head should be queued because of uninitialized chain ht.ExpValidated(t, 1, nil) chain.SetNextSyncPeriod(0) // initialize chain - ts.Run(2, nil, nil) + ts.Run(2) // expect previously queued head to be validated ht.ExpValidated(t, 2, []types.SignedHeader{testSHead1}) @@ -65,34 +65,34 @@ func TestValidatedHead(t *testing.T) { ts.ServerEvent(EvNewSignedHead, testServer1, testSHead2) ts.AddServer(testServer2, 1) ts.ServerEvent(EvNewSignedHead, testServer2, testSHead2) - ts.Run(3, nil, nil) + ts.Run(3) // expect both head announcements to be validated instantly ht.ExpValidated(t, 3, []types.SignedHeader{testSHead2, testSHead2}) ts.ServerEvent(EvNewSignedHead, testServer1, testSHead3) ts.AddServer(testServer3, 1) ts.ServerEvent(EvNewSignedHead, testServer3, testSHead4) - ts.Run(4, nil, nil) + ts.Run(4) // future period annonced heads should be queued ht.ExpValidated(t, 4, nil) chain.SetNextSyncPeriod(2) - ts.Run(5, nil, nil) + ts.Run(5) // testSHead3 can be validated now but not testSHead4 ht.ExpValidated(t, 5, []types.SignedHeader{testSHead3}) // server 3 disconnected without proving period 3, its announced head should be dropped ts.RemoveServer(testServer3) - ts.Run(6, nil, nil) + ts.Run(6) ht.ExpValidated(t, 6, nil) chain.SetNextSyncPeriod(3) - ts.Run(7, nil, nil) + ts.Run(7) // testSHead4 could be validated now but it's not queued by any registered server ht.ExpValidated(t, 7, nil) ts.ServerEvent(EvNewSignedHead, testServer2, testSHead4) - ts.Run(8, nil, nil) + ts.Run(8) // now testSHead4 should be validated ht.ExpValidated(t, 8, []types.SignedHeader{testSHead4}) } @@ -107,45 +107,45 @@ func TestPrefetchHead(t *testing.T) { ts.AddServer(testServer1, 1) ts.ServerEvent(EvNewHead, testServer1, testHead1) - ts.Run(1, nil, nil) + ts.Run(1) ht.ExpPrefetch(t, 1, testHead1) // s1: h1 ts.AddServer(testServer2, 1) ts.ServerEvent(EvNewHead, testServer2, testHead2) - ts.Run(2, nil, nil) + ts.Run(2) ht.ExpPrefetch(t, 2, testHead2) // s1: h1, s2: h2 ts.ServerEvent(EvNewHead, testServer1, testHead2) - ts.Run(3, nil, nil) + ts.Run(3) ht.ExpPrefetch(t, 3, testHead2) // s1: h2, s2: h2 ts.AddServer(testServer3, 1) ts.ServerEvent(EvNewHead, testServer3, testHead3) - ts.Run(4, nil, nil) + ts.Run(4) ht.ExpPrefetch(t, 4, testHead2) // s1: h2, s2: h2, s3: h3 ts.AddServer(testServer4, 1) ts.ServerEvent(EvNewHead, testServer4, testHead4) - ts.Run(5, nil, nil) + ts.Run(5) ht.ExpPrefetch(t, 5, testHead2) // s1: h2, s2: h2, s3: h3, s4: h4 ts.ServerEvent(EvNewHead, testServer2, testHead3) - ts.Run(6, nil, nil) + ts.Run(6) ht.ExpPrefetch(t, 6, testHead3) // s1: h2, s2: h3, s3: h3, s4: h4 ts.RemoveServer(testServer3) - ts.Run(7, nil, nil) + ts.Run(7) ht.ExpPrefetch(t, 7, testHead4) // s1: h2, s2: h3, s4: h4 ts.RemoveServer(testServer1) - ts.Run(8, nil, nil) + ts.Run(8) ht.ExpPrefetch(t, 8, testHead4) // s2: h3, s4: h4 ts.RemoveServer(testServer4) - ts.Run(9, nil, nil) + ts.Run(9) ht.ExpPrefetch(t, 9, testHead3) // s2: h3 ts.RemoveServer(testServer2) - ts.Run(10, nil, nil) + ts.Run(10) ht.ExpPrefetch(t, 10, testHead0) // no servers registered } diff --git a/beacon/light/sync/test_helpers.go b/beacon/light/sync/test_helpers.go index fd218a05ab..688d00b26f 100644 --- a/beacon/light/sync/test_helpers.go +++ b/beacon/light/sync/test_helpers.go @@ -17,6 +17,7 @@ package sync import ( + "reflect" "testing" "github.com/ethereum/go-ethereum/beacon/light" @@ -24,15 +25,6 @@ import ( "github.com/ethereum/go-ethereum/beacon/types" ) -type TestServer struct { - ts *TestScheduler - ID int -} - -func (s *TestServer) Fail(desc string) { - s.ts.serverFail(s) -} - type requestWithID struct { sid request.ServerAndID request request.Request @@ -44,7 +36,7 @@ type TestScheduler struct { events []request.Event servers []request.Server allowance map[request.Server]int - sent map[int]requestWithID + sent map[int][]requestWithID testIndex int expFail map[request.Server]int // expected Server.Fail calls during next Run lastId request.ID @@ -56,13 +48,26 @@ func NewTestScheduler(t *testing.T, module request.Module) *TestScheduler { module: module, allowance: make(map[request.Server]int), expFail: make(map[request.Server]int), - sent: make(map[int]requestWithID), + sent: make(map[int][]requestWithID), } } -func (ts *TestScheduler) Run(testIndex int, expServer request.Server, expReq request.Request) { +func (ts *TestScheduler) Run(testIndex int, exp ...any) { + expReqs := make([]requestWithID, len(exp)/2) + id := ts.lastId + for i := range expReqs { + id++ + expReqs[i] = requestWithID{ + sid: request.ServerAndID{Server: exp[i*2].(request.Server), ID: id}, + request: exp[i*2+1].(request.Request), + } + } + if len(expReqs) == 0 { + expReqs = nil + } + ts.testIndex = testIndex - ts.module.Process(ts.events) + ts.module.Process(ts, ts.events) ts.events = nil for server, count := range ts.expFail { @@ -70,31 +75,47 @@ func (ts *TestScheduler) Run(testIndex int, expServer request.Server, expReq req if count == 0 { continue } - ts.t.Errorf("Missing %d Server.Fail(s) from server %d in test case #%d", count, server.(*TestServer).ID, testIndex) + ts.t.Errorf("Missing %d Server.Fail(s) from server %s in test case #%d", count, server.(string), testIndex) } - expReqWithID := requestWithID{ - sid: request.ServerAndID{Server: expServer, ID: ts.lastId + 1}, - request: expReq, - } - req, ok := ts.tryRequest(testIndex, ts.module.MakeRequest) - if expReq == nil { - if ok { - ts.t.Errorf("Unexpected request in test case #%d (expected none, got %v)", testIndex, req) - } - return - } - if !ok { - ts.t.Errorf("Missing request in test case #%d (expected %v, got none)", testIndex, expReqWithID) - return - } - if req != expReqWithID { - ts.t.Errorf("Wrong request in test case #%d (expected %v, got %v)", testIndex, expReqWithID, req) + if !reflect.DeepEqual(ts.sent[testIndex], expReqs) { + ts.t.Errorf("Wrong sent requests in test case #%d (expected %v, got %v)", testIndex, expReqs, ts.sent[testIndex]) } } -func (ts *TestScheduler) Request(testIndex int) request.Request { - return ts.sent[testIndex].request +func (ts *TestScheduler) CanSendTo() (cs []request.Server) { + for _, server := range ts.servers { + if ts.allowance[server] > 0 { + cs = append(cs, server) + } + } + return +} + +func (ts *TestScheduler) Send(server request.Server, req request.Request) request.ID { + ts.lastId++ + ts.sent[ts.testIndex] = append(ts.sent[ts.testIndex], requestWithID{ + sid: request.ServerAndID{Server: server, ID: ts.lastId}, + request: req, + }) + ts.allowance[server]-- + return ts.lastId +} + +func (ts *TestScheduler) Fail(server request.Server, desc string) { + if ts.expFail[server] == 0 { + ts.t.Errorf("Unexpected Fail from server %s in test case #%d: %s", server.(string), ts.testIndex, desc) + return + } + ts.expFail[server]-- +} + +func (ts *TestScheduler) Request(testIndex, reqIndex int) requestWithID { + if len(ts.sent[testIndex]) < reqIndex { + ts.t.Errorf("Missing request from test case %d index %d", testIndex, reqIndex) + return requestWithID{} + } + return ts.sent[testIndex][reqIndex-1] } func (ts *TestScheduler) ServerEvent(evType *request.EventType, server request.Server, data any) { @@ -105,10 +126,8 @@ func (ts *TestScheduler) ServerEvent(evType *request.EventType, server request.S }) } -func (ts *TestScheduler) RequestEvent(evType *request.EventType, testIndex int, resp request.Response) { - req, ok := ts.sent[testIndex] - if !ok { - ts.t.Errorf("Missing request from test case %v", testIndex) +func (ts *TestScheduler) RequestEvent(evType *request.EventType, req requestWithID, resp request.Response) { + if req.request == nil { return } ts.events = append(ts.events, request.Event{ @@ -123,7 +142,6 @@ func (ts *TestScheduler) RequestEvent(evType *request.EventType, testIndex int, } func (ts *TestScheduler) AddServer(server request.Server, allowance int) { - server.(*TestServer).ts = ts ts.servers = append(ts.servers, server) ts.allowance[server] = allowance ts.ServerEvent(request.EvRegistered, server, nil) @@ -150,43 +168,6 @@ func (ts *TestScheduler) ExpFail(server request.Server) { ts.expFail[server]++ } -func (ts *TestScheduler) serverFail(server request.Server) { - if ts.expFail[server] == 0 { - ts.t.Errorf("Unexpected Server.Fail from server %d in test case #%d", server.(*TestServer).ID, ts.testIndex) - return - } - ts.expFail[server]-- -} - -func (ts *TestScheduler) tryRequest(testIndex int, requestFn func(server request.Server) (request.Request, float32)) (requestWithID, bool) { - var ( - bestServer request.Server - bestReq request.Request - bestPri float32 - ) - for _, server := range ts.servers { - if ts.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 requestWithID{}, false - } - ts.allowance[bestServer]-- - ts.lastId++ - req := requestWithID{ - sid: request.ServerAndID{Server: bestServer, ID: ts.lastId}, - request: bestReq, - } - ts.sent[testIndex] = req - ts.RequestEvent(request.EvRequest, testIndex, nil) - return req, true -} - type TestCommitteeChain struct { fsp, nsp uint64 init bool diff --git a/beacon/light/sync/update_sync.go b/beacon/light/sync/update_sync.go index d59d6097fe..5513f9becd 100644 --- a/beacon/light/sync/update_sync.go +++ b/beacon/light/sync/update_sync.go @@ -52,16 +52,12 @@ func NewCheckpointInit(chain committeeChain, checkpointHash common.Hash) *Checkp } } -func (s *CheckpointInit) Process(events []request.Event) { +func (s *CheckpointInit) Process(requester request.Requester, events []request.Event) { for _, event := range events { if !event.IsRequestEvent() { continue } sid, req, resp := event.RequestInfo() - if event.Type == request.EvRequest { - s.locked = sid - continue - } if s.locked == sid { s.locked = request.ServerAndID{} } @@ -71,16 +67,21 @@ func (s *CheckpointInit) Process(events []request.Event) { s.initialized = true return } - event.Server.Fail("invalid checkpoint data") + + requester.Fail(event.Server, "invalid checkpoint data") } } -} - -func (s *CheckpointInit) MakeRequest(server request.Server) (request.Request, float32) { + // start a request if possible if s.initialized || s.locked != (request.ServerAndID{}) { - return nil, 0 + return } - return ReqCheckpointData(s.checkpointHash), 0 + cs := requester.CanSendTo() + if len(cs) == 0 { + return + } + server := cs[0] + id := requester.Send(server, ReqCheckpointData(s.checkpointHash)) + s.locked = request.ServerAndID{Server: server, ID: id} } // ForwardUpdateSync implements request.Module; it fetches updates between the @@ -194,12 +195,9 @@ func (u updateResponseList) Less(i, j int) bool { return u[i].request.FirstPeriod < u[j].request.FirstPeriod } -func (s *ForwardUpdateSync) Process(events []request.Event) { +func (s *ForwardUpdateSync) Process(requester request.Requester, events []request.Event) { for _, event := range events { switch event.Type { - case request.EvRequest: - sid, req, _ := event.RequestInfo() - s.lockRange(sid, req.(ReqUpdates)) case request.EvResponse, request.EvFail, request.EvTimeout: sid, rq, rs := event.RequestInfo() req := rq.(ReqUpdates) @@ -212,7 +210,7 @@ func (s *ForwardUpdateSync) Process(events []request.Event) { s.lockRange(sid, req) queued = true } else { - event.Server.Fail("invalid update range") + requester.Fail(event.Server, "invalid update range") } } if !queued { @@ -230,8 +228,8 @@ func (s *ForwardUpdateSync) Process(events []request.Event) { sort.Sort(updateResponseList(s.processQueue)) for s.processQueue != nil { u := s.processQueue[0] - if !s.processResponse(u) { - return + if !s.processResponse(requester, u) { + break } s.unlockRange(u.sid, u.request) s.processQueue = s.processQueue[1:] @@ -239,11 +237,43 @@ func (s *ForwardUpdateSync) Process(events []request.Event) { s.processQueue = nil } } + + // start new requests if possible + startPeriod, chainInit := s.chain.NextSyncPeriod() + if !chainInit { + return + } + for { + firstPeriod, maxCount := s.rangeLock.firstUnlocked(startPeriod, maxUpdateRequest) + var ( + sendTo request.Server + bestCount uint64 + ) + for _, server := range requester.CanSendTo() { + nextPeriod := s.nextSyncPeriod[server] + if nextPeriod <= firstPeriod { + continue + } + count := maxCount + if nextPeriod < firstPeriod+maxCount { + count = nextPeriod - firstPeriod + } + if count > bestCount { + sendTo, bestCount = server, count + } + } + if sendTo == nil { + return + } + req := ReqUpdates{FirstPeriod: firstPeriod, Count: bestCount} + id := requester.Send(sendTo, req) + s.lockRange(request.ServerAndID{Server: sendTo, ID: id}, req) + } } // processResponse adds the fetched updates and committees to the committee chain. // Returns true in case of full or partial success. -func (s *ForwardUpdateSync) processResponse(u updateResponse) (success bool) { +func (s *ForwardUpdateSync) processResponse(requester request.Requester, u updateResponse) (success bool) { for i, update := range u.response.Updates { if err := s.chain.InsertUpdate(update, u.response.Committees[i]); err != nil { if err == light.ErrInvalidPeriod { @@ -252,7 +282,7 @@ func (s *ForwardUpdateSync) processResponse(u updateResponse) (success bool) { return } if err == light.ErrInvalidUpdate || err == light.ErrWrongCommitteeRoot || err == light.ErrCannotReorg { - u.sid.Server.Fail("invalid update received") + requester.Fail(u.sid.Server, "invalid update received") } else { log.Error("Unexpected InsertUpdate error", "error", err) } @@ -262,20 +292,3 @@ func (s *ForwardUpdateSync) processResponse(u updateResponse) (success bool) { } return } - -func (s *ForwardUpdateSync) MakeRequest(server request.Server) (request.Request, float32) { - startPeriod, chainInit := s.chain.NextSyncPeriod() - if !chainInit { - return nil, 0 - } - firstPeriod, maxCount := s.rangeLock.firstUnlocked(startPeriod, maxUpdateRequest) - nextPeriod := s.nextSyncPeriod[server] - if nextPeriod <= firstPeriod { - return nil, 0 - } - count := maxCount - if nextPeriod < firstPeriod+maxCount { - count = nextPeriod - firstPeriod - } - return ReqUpdates{FirstPeriod: firstPeriod, Count: count}, float32(count) -} diff --git a/beacon/light/sync/update_sync_test.go b/beacon/light/sync/update_sync_test.go index 1f132829c0..07498b8366 100644 --- a/beacon/light/sync/update_sync_test.go +++ b/beacon/light/sync/update_sync_test.go @@ -37,18 +37,18 @@ func TestCheckpointInit(t *testing.T) { ts.Run(1, testServer1, ReqCheckpointData(checkpointHash)) // server 1 times out; expect request to server 2 - ts.RequestEvent(request.EvTimeout, 1, nil) + ts.RequestEvent(request.EvTimeout, ts.Request(1, 1), nil) ts.Run(2, testServer2, ReqCheckpointData(checkpointHash)) // invalid response from server 2; expect init state to still be false - ts.RequestEvent(request.EvResponse, 2, &types.BootstrapData{Header: types.Header{Slot: 123456}}) + ts.RequestEvent(request.EvResponse, ts.Request(2, 1), &types.BootstrapData{Header: types.Header{Slot: 123456}}) ts.ExpFail(testServer2) - ts.Run(3, nil, nil) + ts.Run(3) chain.ExpInit(t, false) // server 1 fails (hard timeout) - ts.RequestEvent(request.EvFail, 1, nil) - ts.Run(4, nil, nil) + ts.RequestEvent(request.EvFail, ts.Request(1, 1), nil) + ts.Run(4) chain.ExpInit(t, false) // server 3 is registered; expect bootstrap request to server 3 @@ -56,8 +56,8 @@ func TestCheckpointInit(t *testing.T) { ts.Run(5, testServer3, ReqCheckpointData(checkpointHash)) // valid response from server 3; expect chain to be initialized - ts.RequestEvent(request.EvResponse, 5, checkpoint) - ts.Run(6, nil, nil) + ts.RequestEvent(request.EvResponse, ts.Request(5, 1), checkpoint) + ts.Run(6) chain.ExpInit(t, true) } @@ -73,68 +73,72 @@ func TestUpdateSyncParallel(t *testing.T) { ts.ServerEvent(EvNewSignedHead, testServer2, types.SignedHeader{SignatureSlot: 0x2000*100 + 0x1000}) // expect 6 requests to be sent - ts.Run(1, testServer1, ReqUpdates{FirstPeriod: 0, Count: 8}) - ts.Run(2, testServer1, ReqUpdates{FirstPeriod: 8, Count: 8}) - ts.Run(3, testServer1, ReqUpdates{FirstPeriod: 16, Count: 8}) - ts.Run(4, testServer2, ReqUpdates{FirstPeriod: 24, Count: 8}) - ts.Run(5, testServer2, ReqUpdates{FirstPeriod: 32, Count: 8}) - ts.Run(6, testServer2, ReqUpdates{FirstPeriod: 40, Count: 8}) + ts.Run(1, + testServer1, ReqUpdates{FirstPeriod: 0, Count: 8}, + testServer1, ReqUpdates{FirstPeriod: 8, Count: 8}, + testServer1, ReqUpdates{FirstPeriod: 16, Count: 8}, + testServer2, ReqUpdates{FirstPeriod: 24, Count: 8}, + testServer2, ReqUpdates{FirstPeriod: 32, Count: 8}, + testServer2, ReqUpdates{FirstPeriod: 40, Count: 8}) // valid response to request 1; expect 8 periods synced and a new request started - ts.RequestEvent(request.EvResponse, 1, testRespUpdate(ts.Request(1))) + ts.RequestEvent(request.EvResponse, ts.Request(1, 1), testRespUpdate(ts.Request(1, 1))) ts.AddAllowance(testServer1, 1) ts.Run(7, testServer1, ReqUpdates{FirstPeriod: 48, Count: 8}) chain.ExpNextSyncPeriod(t, 8) // 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.RequestEvent(request.EvResponse, ts.Request(1, 4), testRespUpdate(ts.Request(1, 4))) + ts.RequestEvent(request.EvResponse, ts.Request(1, 5), testRespUpdate(ts.Request(1, 5))) ts.AddAllowance(testServer2, 2) // 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}) + ts.Run(8, + testServer2, ReqUpdates{FirstPeriod: 56, Count: 8}, + testServer2, ReqUpdates{FirstPeriod: 64, Count: 8}) chain.ExpNextSyncPeriod(t, 8) // soft timeout for requests 2 and 3 (server 1 is overloaded) - ts.RequestEvent(request.EvTimeout, 2, nil) - ts.RequestEvent(request.EvTimeout, 3, nil) + ts.RequestEvent(request.EvTimeout, ts.Request(1, 2), nil) + ts.RequestEvent(request.EvTimeout, ts.Request(1, 3), nil) // no allowance, no more requests - ts.Run(10, nil, nil) + ts.Run(10) // 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.RequestEvent(request.EvResponse, ts.Request(1, 6), testRespUpdate(ts.Request(1, 6))) + ts.RequestEvent(request.EvResponse, ts.Request(8, 1), testRespUpdate(ts.Request(8, 1))) + ts.RequestEvent(request.EvResponse, ts.Request(8, 2), testRespUpdate(ts.Request(8, 2))) ts.AddAllowance(testServer2, 3) // server 2 can now resend requests 2 and 3 (timed out by server 1) and also send a new one - ts.Run(11, testServer2, ReqUpdates{FirstPeriod: 8, Count: 8}) - ts.Run(12, testServer2, ReqUpdates{FirstPeriod: 16, Count: 8}) - ts.Run(13, testServer2, ReqUpdates{FirstPeriod: 72, Count: 8}) + ts.Run(11, + testServer2, ReqUpdates{FirstPeriod: 8, Count: 8}, + testServer2, ReqUpdates{FirstPeriod: 16, Count: 8}, + testServer2, ReqUpdates{FirstPeriod: 72, Count: 8}) // server 1 finally answers timed out request 2 - ts.RequestEvent(request.EvResponse, 2, testRespUpdate(ts.Request(2))) + ts.RequestEvent(request.EvResponse, ts.Request(1, 2), testRespUpdate(ts.Request(1, 2))) ts.AddAllowance(testServer1, 1) // expect sync progress and one new request ts.Run(14, testServer1, ReqUpdates{FirstPeriod: 80, Count: 8}) chain.ExpNextSyncPeriod(t, 16) // server 2 answers requests 11 and 12 (resends of requests 2 and 3) - ts.RequestEvent(request.EvResponse, 11, testRespUpdate(ts.Request(11))) - ts.RequestEvent(request.EvResponse, 12, testRespUpdate(ts.Request(12))) + ts.RequestEvent(request.EvResponse, ts.Request(11, 1), testRespUpdate(ts.Request(11, 1))) + ts.RequestEvent(request.EvResponse, ts.Request(11, 2), testRespUpdate(ts.Request(11, 2))) ts.AddAllowance(testServer2, 2) - ts.Run(15, testServer2, ReqUpdates{FirstPeriod: 88, Count: 8}) - ts.Run(16, testServer2, ReqUpdates{FirstPeriod: 96, Count: 4}) + ts.Run(15, + testServer2, ReqUpdates{FirstPeriod: 88, Count: 8}, + testServer2, ReqUpdates{FirstPeriod: 96, Count: 4}) // finally the gap is filled, update can process responses up to req6 chain.ExpNextSyncPeriod(t, 48) // all remaining requests are answered - ts.RequestEvent(request.EvResponse, 3, testRespUpdate(ts.Request(3))) - ts.RequestEvent(request.EvResponse, 7, testRespUpdate(ts.Request(7))) - ts.RequestEvent(request.EvResponse, 13, testRespUpdate(ts.Request(13))) - ts.RequestEvent(request.EvResponse, 14, testRespUpdate(ts.Request(14))) - ts.RequestEvent(request.EvResponse, 15, testRespUpdate(ts.Request(15))) - ts.RequestEvent(request.EvResponse, 16, testRespUpdate(ts.Request(16))) - ts.Run(17, nil, nil) + ts.RequestEvent(request.EvResponse, ts.Request(1, 3), testRespUpdate(ts.Request(1, 3))) + ts.RequestEvent(request.EvResponse, ts.Request(7, 1), testRespUpdate(ts.Request(7, 1))) + ts.RequestEvent(request.EvResponse, ts.Request(11, 3), testRespUpdate(ts.Request(11, 3))) + ts.RequestEvent(request.EvResponse, ts.Request(14, 1), testRespUpdate(ts.Request(14, 1))) + ts.RequestEvent(request.EvResponse, ts.Request(15, 1), testRespUpdate(ts.Request(15, 1))) + ts.RequestEvent(request.EvResponse, ts.Request(15, 2), testRespUpdate(ts.Request(15, 2))) + ts.Run(17) // expect chain to be fully synced chain.ExpNextSyncPeriod(t, 100) } @@ -156,30 +160,32 @@ func TestUpdateSyncDifferentHeads(t *testing.T) { ts.Run(1, testServer3, ReqUpdates{FirstPeriod: 10, Count: 7}) // request times out, expect request to the next best head - ts.RequestEvent(request.EvTimeout, 1, nil) + ts.RequestEvent(request.EvTimeout, ts.Request(1, 1), nil) ts.Run(2, testServer2, ReqUpdates{FirstPeriod: 10, Count: 6}) // request times out, expect request to the last available server - ts.RequestEvent(request.EvTimeout, 2, nil) + ts.RequestEvent(request.EvTimeout, ts.Request(2, 1), nil) ts.Run(3, testServer1, ReqUpdates{FirstPeriod: 10, Count: 5}) // valid response to request 3, expect chain synced to period 15 - ts.RequestEvent(request.EvResponse, 3, testRespUpdate(ts.Request(3))) + ts.RequestEvent(request.EvResponse, ts.Request(3, 1), testRespUpdate(ts.Request(3, 1))) ts.AddAllowance(testServer1, 1) - ts.Run(4, nil, nil) + ts.Run(4) chain.ExpNextSyncPeriod(t, 15) // invalid response to request 1, server can only deliver updates up to period 15 despite announced head - ts.RequestEvent(request.EvResponse, 1, testRespUpdate(ReqUpdates{FirstPeriod: 10, Count: 5})) + truncated := ts.Request(1, 1) + truncated.request = ReqUpdates{FirstPeriod: 10, Count: 5} + ts.RequestEvent(request.EvResponse, ts.Request(1, 1), testRespUpdate(truncated)) ts.ExpFail(testServer3) - ts.Run(5, nil, nil) + ts.Run(5) // expect no progress of chain head chain.ExpNextSyncPeriod(t, 15) // valid response to request 2, expect chain synced to period 16 - ts.RequestEvent(request.EvResponse, 2, testRespUpdate(ts.Request(2))) + ts.RequestEvent(request.EvResponse, ts.Request(2, 1), testRespUpdate(ts.Request(2, 1))) ts.AddAllowance(testServer2, 1) - ts.Run(6, nil, nil) + ts.Run(6) chain.ExpNextSyncPeriod(t, 16) // a new server is registered with announced head period 17 @@ -189,15 +195,18 @@ func TestUpdateSyncDifferentHeads(t *testing.T) { ts.Run(7, testServer4, ReqUpdates{FirstPeriod: 16, Count: 1}) // valid response, expect chain synced to period 17 - ts.RequestEvent(request.EvResponse, 7, testRespUpdate(ts.Request(7))) + ts.RequestEvent(request.EvResponse, ts.Request(7, 1), testRespUpdate(ts.Request(7, 1))) ts.AddAllowance(testServer4, 1) - ts.Run(8, nil, nil) + ts.Run(8) chain.ExpNextSyncPeriod(t, 17) } -func testRespUpdate(request request.Request) request.Response { +func testRespUpdate(request requestWithID) request.Response { var resp RespUpdates - req := request.(ReqUpdates) + if request.request == nil { + return resp + } + req := request.request.(ReqUpdates) resp.Updates = make([]*types.LightClientUpdate, int(req.Count)) resp.Committees = make([]*types.SerializedSyncCommittee, int(req.Count)) period := req.FirstPeriod diff --git a/cmd/blsync/block_sync.go b/cmd/blsync/block_sync.go index e665e509e0..135fd4dfd6 100755 --- a/cmd/blsync/block_sync.go +++ b/cmd/blsync/block_sync.go @@ -29,7 +29,7 @@ import ( // to the validated and prefetch heads. type beaconBlockSync struct { recentBlocks *lru.Cache[common.Hash, *capella.BeaconBlock] - locked map[common.Hash]struct{} + locked map[common.Hash]request.ServerAndID serverHeads map[request.Server]common.Hash headTracker headTracker @@ -53,34 +53,42 @@ func newBeaconBlockSync(headTracker headTracker) *beaconBlockSync { return &beaconBlockSync{ headTracker: headTracker, recentBlocks: lru.NewCache[common.Hash, *capella.BeaconBlock](10), - locked: make(map[common.Hash]struct{}), + locked: make(map[common.Hash]request.ServerAndID), serverHeads: make(map[request.Server]common.Hash), headCh: make(chan headData, 1), } } -func (s *beaconBlockSync) Process(events []request.Event) { +func (s *beaconBlockSync) Process(requester request.Requester, events []request.Event) { for _, event := range events { switch event.Type { - case request.EvRequest: - _, req, _ := event.RequestInfo() - blockRoot := common.Hash(req.(sync.ReqBeaconBlock)) - s.locked[blockRoot] = struct{}{} case request.EvResponse, request.EvFail, request.EvTimeout: - _, req, resp := event.RequestInfo() + sid, req, resp := event.RequestInfo() blockRoot := common.Hash(req.(sync.ReqBeaconBlock)) if resp != nil { s.recentBlocks.Add(blockRoot, resp.(*capella.BeaconBlock)) } - delete(s.locked, blockRoot) + if s.locked[blockRoot] == sid { + delete(s.locked, blockRoot) + } case sync.EvNewHead: s.serverHeads[event.Server] = event.Data.(types.HeadInfo).BlockRoot case request.EvUnregistered: delete(s.serverHeads, event.Server) } } + s.updateValidatedHead() + // request validated head block if unavailable and not yet requested + if vh, ok := s.headTracker.ValidatedHead(); ok { + s.tryRequestBlock(requester, vh.Header.Hash(), false) + } + // request prefetch head if the given server has announced it + if prefetchHead := s.headTracker.PrefetchHead().BlockRoot; prefetchHead != (common.Hash{}) { + s.tryRequestBlock(requester, prefetchHead, true) + } +} - // send validated head block +func (s *beaconBlockSync) updateValidatedHead() { head, ok := s.headTracker.ValidatedHead() if !ok { return @@ -105,27 +113,22 @@ func (s *beaconBlockSync) Process(events []request.Event) { } } -func (s *beaconBlockSync) MakeRequest(server request.Server) (request.Request, float32) { - // request validated head block if unavailable and not yet requested - if vh, ok := s.headTracker.ValidatedHead(); ok { - validatedHead := vh.Header.Hash() - if _, ok := s.recentBlocks.Get(validatedHead); !ok { - if _, ok := s.locked[validatedHead]; !ok { - return sync.ReqBeaconBlock(validatedHead), 1 - } - } +func (s *beaconBlockSync) tryRequestBlock(requester request.Requester, blockRoot common.Hash, needSameHead bool) { + if _, ok := s.recentBlocks.Get(blockRoot); ok { + return } - // request prefetch head if the given server has announced it - if prefetchHead := s.headTracker.PrefetchHead().BlockRoot; prefetchHead != (common.Hash{}) && prefetchHead == s.serverHeads[server] { - if _, ok := s.recentBlocks.Get(prefetchHead); !ok { - if _, ok := s.locked[prefetchHead]; !ok { - return sync.ReqBeaconBlock(prefetchHead), 0 - } - } + if _, ok := s.locked[blockRoot]; ok { + return + } + for _, server := range requester.CanSendTo() { + if needSameHead && (s.serverHeads[server] != blockRoot) { + continue + } + id := requester.Send(server, sync.ReqBeaconBlock(blockRoot)) + s.locked[blockRoot] = request.ServerAndID{Server: server, ID: id} + return } - return nil, 0 } - func blockHeadInfo(block *capella.BeaconBlock) types.HeadInfo { if block == nil { return types.HeadInfo{} diff --git a/cmd/blsync/block_sync_test.go b/cmd/blsync/block_sync_test.go index 93526f11d0..e6cfc3d185 100644 --- a/cmd/blsync/block_sync_test.go +++ b/cmd/blsync/block_sync_test.go @@ -29,8 +29,8 @@ import ( ) var ( - testServer1 = &sync.TestServer{ID: 1} - testServer2 = &sync.TestServer{ID: 2} + testServer1 = "testServer1" + testServer2 = "testServer2" testBlock1 = &capella.BeaconBlock{Slot: 123} testBlock2 = &capella.BeaconBlock{Slot: 124} @@ -57,7 +57,7 @@ func TestBlockSync(t *testing.T) { } // no block requests expected until head tracker knows about a head - ts.Run(1, nil, nil) + ts.Run(1) expHeadBlock(1, nil) // set block 1 as prefetch head, announced by server 2 @@ -68,15 +68,15 @@ func TestBlockSync(t *testing.T) { ts.Run(2, testServer2, sync.ReqBeaconBlock(head1.BlockRoot)) // valid response - ts.RequestEvent(request.EvResponse, 2, testBlock1) + ts.RequestEvent(request.EvResponse, ts.Request(2, 1), testBlock1) ts.AddAllowance(testServer2, 1) - ts.Run(3, nil, nil) + ts.Run(3) // head block still not expected as the fetched block is not the validated head yet expHeadBlock(3, nil) // set as validated head, expect no further requests but block 1 set as head block ht.validated.Header = blockHeader(testBlock1) - ts.Run(4, nil, nil) + ts.Run(4) expHeadBlock(4, testBlock1) // set block 2 as prefetch head, announced by server 1 @@ -87,8 +87,8 @@ func TestBlockSync(t *testing.T) { ts.Run(5, testServer1, sync.ReqBeaconBlock(head2.BlockRoot)) // req2 fails, no further requests expected because server 2 has not announced it - ts.RequestEvent(request.EvFail, 5, nil) - ts.Run(6, nil, nil) + ts.RequestEvent(request.EvFail, ts.Request(5, 1), nil) + ts.Run(6) // set as validated head before retrieving block; now it's assumed to be available from server 2 too ht.validated.Header = blockHeader(testBlock2) @@ -98,8 +98,8 @@ func TestBlockSync(t *testing.T) { expHeadBlock(4, nil) // valid response, now head block should be block 2 immediately as it is already validated - ts.RequestEvent(request.EvResponse, 7, testBlock2) - ts.Run(8, nil, nil) + ts.RequestEvent(request.EvResponse, ts.Request(7, 1), testBlock2) + ts.Run(8) expHeadBlock(5, testBlock2) }