From 7c613fe752376925307f609c733c03b8f9a26b92 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Tue, 25 Apr 2023 11:51:13 +0200 Subject: [PATCH] beacon/merkle: fixed unit test --- beacon/light/api/sync_server.go | 63 +++----------------------- beacon/light/light_chain.go | 6 +-- beacon/light/light_chain_test.go | 2 +- beacon/light/request/environment.go | 11 +++-- beacon/light/request/scheduler.go | 5 ++ beacon/light/request/scheduler_test.go | 4 +- beacon/light/request/server.go | 3 ++ beacon/light/sync/header_sync.go | 4 +- beacon/light/sync/state_sync.go | 18 ++++---- beacon/light/sync/update_sync.go | 37 +++++++++------ beacon/merkle/binary_merkle_test.go | 6 +-- cmd/blsync/block_sync.go | 27 ++++++----- 12 files changed, 80 insertions(+), 106 deletions(-) diff --git a/beacon/light/api/sync_server.go b/beacon/light/api/sync_server.go index 47eb689e7f..7a24e6e49e 100755 --- a/beacon/light/api/sync_server.go +++ b/beacon/light/api/sync_server.go @@ -17,8 +17,6 @@ package api import ( - "sync" - "github.com/ethereum/go-ethereum/beacon/light" "github.com/ethereum/go-ethereum/beacon/light/types" "github.com/ethereum/go-ethereum/beacon/merkle" @@ -33,42 +31,26 @@ const ( ) type SyncServer struct { - api *BeaconLightApi - lock sync.RWMutex - - unsubscribe func() - canRequestBootstrap bool - firstUpdate, afterLastUpdate uint64 - firstState uint64 //TODO ... + api *BeaconLightApi + unsubscribe func() } func NewSyncServer(api *BeaconLightApi) *SyncServer { - return &SyncServer{ - api: api, - canRequestBootstrap: true, - } + return &SyncServer{api: api} } func (s *SyncServer) SubscribeHeads(newHead func(uint64, common.Hash), newSignedHead func(signedHead types.SignedHead)) { - s.lock.Lock() - s.unsubscribe = s.api.StartHeadListener(newHead, func(signedHead types.SignedHead) { - s.lock.Lock() - s.afterLastUpdate = types.PeriodOfSlot(signedHead.Header.Slot + 256) - s.lock.Unlock() - newSignedHead(signedHead) - }, func(err error) { + s.unsubscribe = s.api.StartHeadListener(newHead, newSignedHead, func(err error) { log.Warn("Head event stream error", "err", err) }) - s.lock.Unlock() } +// Note: UnsubscribeHeads should not be called concurrently with SubscribeHeads func (s *SyncServer) UnsubscribeHeads() { - s.lock.Lock() if s.unsubscribe != nil { s.unsubscribe() s.unsubscribe = nil } - s.lock.Unlock() } func (s *SyncServer) DelayUntil() mclock.AbsTime { return 0 } //TODO @@ -77,37 +59,16 @@ func (s *SyncServer) Fail(desc string) { log.Warn("API endpoint failure", "URL", s.api.url, "error", desc) } -func (s *SyncServer) CanRequestBootstrap() bool { - s.lock.RLock() - defer s.lock.RUnlock() - - return s.canRequestBootstrap -} - func (s *SyncServer) RequestBootstrap(checkpointHash common.Hash, response func(*light.CheckpointData)) { go func() { if checkpoint, err := s.api.GetCheckpointData(checkpointHash); err == nil { response(checkpoint) } else { - s.lock.Lock() - s.canRequestBootstrap = false - s.lock.Unlock() response(nil) } }() } -func (s *SyncServer) UpdateRange() types.PeriodRange { - s.lock.RLock() - defer s.lock.RUnlock() - - r := types.PeriodRange{First: s.firstUpdate, AfterLast: s.afterLastUpdate} - if !r.IsEmpty() { - return r - } - return types.PeriodRange{} -} - func (s *SyncServer) RequestUpdates(first, count uint64, response func([]*types.LightClientUpdate, []*types.SerializedCommittee)) { go func() { if updates, committees, err := s.api.GetBestUpdatesAndCommittees(first, count); err == nil { @@ -138,23 +99,11 @@ func (s *SyncServer) RequestBeaconHeader(blockRoot common.Hash, response func(*t }() } -func (s *SyncServer) BeaconStateTail() uint64 { - s.lock.RLock() - defer s.lock.RUnlock() - - return s.firstState -} - -func (s *SyncServer) RequestBeaconState(slot uint64, stateRoot common.Hash, format merkle.CompactProofFormat, response func(*merkle.MultiProof)) { +func (s *SyncServer) RequestBeaconState(stateRoot common.Hash, format merkle.CompactProofFormat, response func(*merkle.MultiProof)) { go func() { if proof, err := s.api.GetStateProof(stateRoot, format); err == nil { response(&proof) } else { - s.lock.Lock() - if slot >= s.firstState { - s.firstState = slot + 1 - } - s.lock.Unlock() response(nil) } }() diff --git a/beacon/light/light_chain.go b/beacon/light/light_chain.go index 3910dd26a8..63b1ed7436 100644 --- a/beacon/light/light_chain.go +++ b/beacon/light/light_chain.go @@ -555,11 +555,11 @@ func (lc *LightChain) HasStateProof(header types.Header) bool { } // GetStateProof returns the state proof belonging to the given header. -func (lc *LightChain) GetStateProof(header types.Header) (merkle.MultiProof, error) { - if proof, ok := lc.stateProofCache.Get(slotAndHash{header.Slot, header.StateRoot}); ok { +func (lc *LightChain) GetStateProof(slot uint64, stateRoot common.Hash) (merkle.MultiProof, error) { + if proof, ok := lc.stateProofCache.Get(slotAndHash{slot, stateRoot}); ok { return proof, nil } - proofEnc, err := lc.db.Get(getStateProofKey(header.Slot, header.StateRoot)) + proofEnc, err := lc.db.Get(getStateProofKey(slot, stateRoot)) if err != nil { return merkle.MultiProof{}, ErrNotFound } diff --git a/beacon/light/light_chain_test.go b/beacon/light/light_chain_test.go index 6c3c182ba9..9de11608e6 100644 --- a/beacon/light/light_chain_test.go +++ b/beacon/light/light_chain_test.go @@ -383,7 +383,7 @@ func (c *lightChainTest) checkStateProof(header types.Header, expFound bool) { if found := c.chain.HasStateProof(header); found != expFound { c.t.Errorf("Incorrect result from HasStateProof (expected %v, got %v)", expFound, found) } - if proof, err := c.chain.GetStateProof(header); err == nil { + if proof, err := c.chain.GetStateProof(header.Slot, header.StateRoot); err == nil { if !expFound { c.t.Errorf("Unexpected state proof found by GetStateProof") } diff --git a/beacon/light/request/environment.go b/beacon/light/request/environment.go index 7caad2bad7..e85b82907b 100644 --- a/beacon/light/request/environment.go +++ b/beacon/light/request/environment.go @@ -20,8 +20,8 @@ package request // conditions under which it can be sent to a certain server while SendTo sends it // to the selected one. Both should be non-blocking and are never called concurrently. type request interface { - CanSendTo(server *Server) (canSend bool, priority uint64) - SendTo(server *Server) + CanSendTo(server *Server, moduleData *interface{}) (canSend bool, priority uint64) + SendTo(server *Server, moduleData *interface{}) } // Environment allows modules to start network requests when triggered. It is @@ -33,6 +33,7 @@ type request interface { type Environment struct { *HeadTracker scheduler *Scheduler + module Module allServers []*Server canRequestNow map[*Server]struct{} } @@ -49,7 +50,7 @@ func (s *Environment) TryRequest(req request) bool { delete(s.canRequestNow, server) continue } - canSend, requestPriority := req.CanSendTo(server) + canSend, requestPriority := req.CanSendTo(server, server.moduleData[s.module]) if !canSend || requestPriority < maxRequestPriority || (requestPriority == maxRequestPriority && serverPriority <= maxServerPriority) { continue @@ -58,7 +59,7 @@ func (s *Environment) TryRequest(req request) bool { bestServer = server } if bestServer != nil { - req.SendTo(bestServer) + req.SendTo(bestServer, bestServer.moduleData[s.module]) return true } return false @@ -80,7 +81,7 @@ func (s *Environment) CanRequestNow() bool { // at the moment) that could serve the given request. func (s *Environment) CanRequestLater(req request) bool { for _, server := range s.allServers { - if canSend, _ := req.CanSendTo(server); canSend { + if canSend, _ := req.CanSendTo(server, server.moduleData[s.module]); canSend { return true } } diff --git a/beacon/light/request/scheduler.go b/beacon/light/request/scheduler.go index 775142e523..475249c19d 100644 --- a/beacon/light/request/scheduler.go +++ b/beacon/light/request/scheduler.go @@ -156,6 +156,9 @@ func (s *Scheduler) RegisterServer(requestServer RequestServer) { defer s.lock.Unlock() server := s.newServer(requestServer) + for _, module := range s.modules { + server.moduleData[module] = new(interface{}) + } s.servers = append(s.servers, server) s.headTracker.registerServer(server) s.triggerServer(server) @@ -260,8 +263,10 @@ func (s *Scheduler) processModules(trModules map[Module]struct{}, trServers map[ for _, module := range s.modules { if _, ok := trModules[module]; ok { + mtEnv.module = module module.Process(&mtEnv) } else if len(stEnv.canRequestNow) > 0 { + stEnv.module = module module.Process(&stEnv) } } diff --git a/beacon/light/request/scheduler_test.go b/beacon/light/request/scheduler_test.go index a7f0d924b2..8f707c0a2f 100644 --- a/beacon/light/request/scheduler_test.go +++ b/beacon/light/request/scheduler_test.go @@ -201,11 +201,11 @@ type testRequest struct { returnFns []func() } -func (r *testRequest) CanSendTo(server *Server) (canSend bool, priority uint64) { +func (r *testRequest) CanSendTo(server *Server, moduleData *interface{}) (canSend bool, priority uint64) { return r.reqLock.CanRequest(), 0 } -func (r *testRequest) SendTo(server *Server) { +func (r *testRequest) SendTo(server *Server, moduleData *interface{}) { reqId := r.reqLock.Send(server) r.returnFns = append(r.returnFns, func() { r.reqLock.Returned(server, reqId) diff --git a/beacon/light/request/server.go b/beacon/light/request/server.go index eb185a5567..88e4cbc29a 100644 --- a/beacon/light/request/server.go +++ b/beacon/light/request/server.go @@ -46,6 +46,8 @@ type Server struct { latestHeadHash common.Hash unregistered bool // accessed under HeadTracker.prefetchLock + moduleData map[Module]*interface{} + lock sync.Mutex sent map[uint64]chan struct{} // closed when returned; nil when timed out timeoutCount int @@ -61,6 +63,7 @@ func (s *Scheduler) newServer(server RequestServer) *Server { return &Server{ RequestServer: server, scheduler: s, + moduleData: make(map[Module]*interface{}), sent: make(map[uint64]chan struct{}), stopCh: make(chan struct{}), } diff --git a/beacon/light/sync/header_sync.go b/beacon/light/sync/header_sync.go index 923f7c4036..b5444246ff 100644 --- a/beacon/light/sync/header_sync.go +++ b/beacon/light/sync/header_sync.go @@ -155,7 +155,7 @@ type headerRequest struct { prefetch bool } -func (r headerRequest) CanSendTo(server *request.Server) (canSend bool, priority uint64) { +func (r headerRequest) CanSendTo(server *request.Server, moduleData *interface{}) (canSend bool, priority uint64) { if _, ok := server.RequestServer.(beaconHeaderServer); !ok { return false, 0 } @@ -166,7 +166,7 @@ func (r headerRequest) CanSendTo(server *request.Server) (canSend bool, priority return r.blockRoot == headRoot, 0 } -func (r headerRequest) SendTo(server *request.Server) { +func (r headerRequest) SendTo(server *request.Server, moduleData *interface{}) { reqId := r.reqLock.Send(server, r.blockRoot) server.RequestServer.(beaconHeaderServer).RequestBeaconHeader(r.blockRoot, func(header *types.Header) { r.lock.Lock() diff --git a/beacon/light/sync/state_sync.go b/beacon/light/sync/state_sync.go index 92653b5b51..d91188257a 100644 --- a/beacon/light/sync/state_sync.go +++ b/beacon/light/sync/state_sync.go @@ -23,7 +23,6 @@ import ( "github.com/ethereum/go-ethereum/beacon/light" "github.com/ethereum/go-ethereum/beacon/light/request" - "github.com/ethereum/go-ethereum/beacon/light/types" "github.com/ethereum/go-ethereum/beacon/merkle" "github.com/ethereum/go-ethereum/common" @@ -32,8 +31,7 @@ import ( type beaconStateServer interface { request.RequestServer - BeaconStateTail() uint64 - RequestBeaconState(slot uint64, stateRoot common.Hash, format merkle.CompactProofFormat, response func(*merkle.MultiProof)) + RequestBeaconState(stateRoot common.Hash, format merkle.CompactProofFormat, response func(*merkle.MultiProof)) } type StateSync struct { @@ -192,8 +190,9 @@ type stateRequest struct { prefetch bool } -func (r stateRequest) CanSendTo(server *request.Server) (canSend bool, priority uint64) { - if rs, ok := server.RequestServer.(beaconStateServer); !ok || r.header.Slot < rs.BeaconStateTail() { +func (r stateRequest) CanSendTo(server *request.Server, moduleData *interface{}) (canSend bool, priority uint64) { + stateTail, _ := (*moduleData).(uint64) + if _, ok := server.RequestServer.(beaconStateServer); !ok || r.header.Slot < stateTail { return false, 0 } if !r.prefetch { @@ -203,15 +202,18 @@ func (r stateRequest) CanSendTo(server *request.Server) (canSend bool, priority return r.header.Hash() == headRoot, 0 } -func (r stateRequest) SendTo(server *request.Server) { +func (r stateRequest) SendTo(server *request.Server, moduleData *interface{}) { reqId := r.reqLock.Send(server, r.header.StateRoot) - server.RequestServer.(beaconStateServer).RequestBeaconState(r.header.Slot, r.header.StateRoot, r.syncProofFormat, func(proof *merkle.MultiProof) { + server.RequestServer.(beaconStateServer).RequestBeaconState(r.header.StateRoot, r.syncProofFormat, func(proof *merkle.MultiProof) { r.lock.Lock() defer r.lock.Unlock() r.reqLock.Returned(server, reqId, r.header.StateRoot) if proof == nil { - //server.Fail("error retrieving beacon state proof") + stateTail, _ := (*moduleData).(uint64) + if r.header.Slot >= stateTail { + (*moduleData) = r.header.Slot + 1 + } return } oldStateHead, _, _ := r.chain.StateProofRange() diff --git a/beacon/light/sync/update_sync.go b/beacon/light/sync/update_sync.go index f935156af0..7b05634a0c 100644 --- a/beacon/light/sync/update_sync.go +++ b/beacon/light/sync/update_sync.go @@ -30,7 +30,6 @@ const maxUpdateRequest = 8 type checkpointInitServer interface { request.RequestServer - CanRequestBootstrap() bool RequestBootstrap(checkpointHash common.Hash, response func(*light.CheckpointData)) } @@ -89,14 +88,15 @@ type checkpointRequest struct { checkpointHash common.Hash } -func (r checkpointRequest) CanSendTo(server *request.Server) (canSend bool, priority uint64) { - if cs, ok := server.RequestServer.(checkpointInitServer); !ok || !cs.CanRequestBootstrap() { +func (r checkpointRequest) CanSendTo(server *request.Server, moduleData *interface{}) (canSend bool, priority uint64) { + if _, ok := server.RequestServer.(checkpointInitServer); !ok || (*moduleData) != nil { + // if moduleData is not nil then the request has failed once already return false, 0 } return true, 0 } -func (r checkpointRequest) SendTo(server *request.Server) { +func (r checkpointRequest) SendTo(server *request.Server, moduleData *interface{}) { reqId := r.reqLock.Send(server) server.RequestServer.(checkpointInitServer).RequestBootstrap(r.checkpointHash, func(checkpoint *light.CheckpointData) { r.lock.Lock() @@ -104,6 +104,7 @@ func (r checkpointRequest) SendTo(server *request.Server) { r.reqLock.Returned(server, reqId) if checkpoint == nil || !checkpoint.Validate() { + (*moduleData) = struct{}{} server.Fail("error retrieving checkpoint data") return } @@ -116,7 +117,6 @@ func (r checkpointRequest) SendTo(server *request.Server) { type updateServer interface { request.RequestServer - UpdateRange() types.PeriodRange RequestUpdates(first, count uint64, response func([]*types.LightClientUpdate, []*types.SerializedCommittee)) } @@ -135,9 +135,11 @@ func NewForwardUpdateSync(chain *light.CommitteeChain) *ForwardUpdateSync { // SetupModuleTriggers implements request.Module func (s *ForwardUpdateSync) SetupModuleTriggers(trigger func(id string, subscribe bool) *request.ModuleTrigger) { s.reqLock.Trigger = trigger("forwardUpdateSync", true) - // committeeChainInit signals that the committee chain is initialized (has fixed committee roots) and the first update request can be constructed. + // committeeChainInit signals that the committee chain is initialized (has + // fixed committee roots) and the first update request can be constructed. trigger("committeeChainInit", true) - // validatedHead ensures that the UpdateRange of each server is re-checked as new heads appear and new updates are synced as they become available. + // validatedHead ensures that the UpdateRange of each server is re-checked + // as new heads appear and new updates are synced as they become available. trigger("validatedHead", true) // newUpdate is triggered when a new update is successfully added to the committee chain s.newUpdateTrigger = trigger("newUpdate", true) @@ -166,19 +168,26 @@ type updateRequest struct { first uint64 } -func (r updateRequest) CanSendTo(server *request.Server) (canSend bool, priority uint64) { - if us, ok := server.RequestServer.(updateServer); ok { - if updateRange := us.UpdateRange(); updateRange.Includes(r.first) { - return true, updateRange.AfterLast +func (r updateRequest) CanSendTo(server *request.Server, moduleData *interface{}) (canSend bool, priority uint64) { + if _, ok := server.RequestServer.(updateServer); ok { + firstUpdate, _ := (*moduleData).(uint64) + headSlot, _ := server.LatestHead() + afterLastUpdate := types.PeriodOfSlot(headSlot) + if r.first >= firstUpdate && r.first < afterLastUpdate { + return true, afterLastUpdate } } return false, 0 } -func (r updateRequest) SendTo(server *request.Server) { +func (r updateRequest) SendTo(server *request.Server, moduleData *interface{}) { us := server.RequestServer.(updateServer) - updateRange := us.UpdateRange() - count := updateRange.AfterLast - r.first + headSlot, _ := server.LatestHead() + afterLastUpdate := types.PeriodOfSlot(headSlot) + if afterLastUpdate <= r.first { + return + } + count := afterLastUpdate - r.first if count > maxUpdateRequest { count = maxUpdateRequest } diff --git a/beacon/merkle/binary_merkle_test.go b/beacon/merkle/binary_merkle_test.go index 9dab56ba32..4c39717b3f 100644 --- a/beacon/merkle/binary_merkle_test.go +++ b/beacon/merkle/binary_merkle_test.go @@ -157,7 +157,7 @@ func TestMultiProof(t *testing.T) { readers := make([]ProofReader, len(indexList)) for i, index := range indexList { var mp MultiProof - mp.Format = NewIndexMapFormat().AddLeaf(index, nil) + mp.Format = EncodeCompactProofFormat(NewIndexMapFormat().AddLeaf(index, nil)) writer := NewMultiProofWriter(mp.Format, &mp.Values, nil) testTraverseProof(t, testProofReader, writer, true) readers[i] = mp.Reader(nil) @@ -170,7 +170,7 @@ func TestMultiProof(t *testing.T) { for i := 0; i < mpCount; i++ { format.AddLeaf(indexList[i], nil) } - mp.Format = format + mp.Format = EncodeCompactProofFormat(format) expSuccess := rand.Intn(2) == 0 if !expSuccess { // add an index that should not be available in the merged reader, expect the traversal to fail @@ -183,7 +183,7 @@ func TestMultiProof(t *testing.T) { mps := make([]MultiProof, mpwCount) writers := make([]ProofWriter, mpwCount) for i := range mps { - mps[i].Format = NewIndexMapFormat().AddLeaf(indexList[i], nil) + mps[i].Format = EncodeCompactProofFormat(NewIndexMapFormat().AddLeaf(indexList[i], nil)) writers[i] = NewMultiProofWriter(mps[i].Format, &mps[i].Values, nil) } reader := mp.Reader(nil) diff --git a/cmd/blsync/block_sync.go b/cmd/blsync/block_sync.go index 6bc807fd29..146efe9f7c 100755 --- a/cmd/blsync/block_sync.go +++ b/cmd/blsync/block_sync.go @@ -127,7 +127,7 @@ type blockRequest struct { prefetch bool } -func (r blockRequest) CanSendTo(server *request.Server) (canSend bool, priority uint64) { +func (r blockRequest) CanSendTo(server *request.Server, moduleData *interface{}) (canSend bool, priority uint64) { if _, ok := server.RequestServer.(beaconBlockServer); !ok { return false, 0 } @@ -138,7 +138,7 @@ func (r blockRequest) CanSendTo(server *request.Server) (canSend bool, priority return r.blockRoot == headRoot, 0 } -func (r blockRequest) SendTo(server *request.Server) { +func (r blockRequest) SendTo(server *request.Server, moduleData *interface{}) { reqId := r.reqLock.Send(server, r.blockRoot) server.RequestServer.(beaconBlockServer).RequestBeaconBlock(r.blockRoot, func(block *capella.BeaconBlock) { r.lock.Lock() @@ -223,6 +223,7 @@ type engineApiUpdater struct { chain *light.LightChain updating bool selfTrigger *request.ModuleTrigger + tailTarget uint64 } // SetupModuleTriggers implements request.Module @@ -235,7 +236,11 @@ func (s *engineApiUpdater) SetupModuleTriggers(trigger func(id string, subscribe // Process implements request.Module func (s *engineApiUpdater) Process(env *request.Environment) { s.lock.Lock() - defer s.lock.Unlock() + defer func() { + s.headerSync.SetTailTarget(s.tailTarget) + s.chain.Prune(s.tailTarget, true) + s.lock.Unlock() + }() if s.updating { return @@ -248,25 +253,25 @@ func (s *engineApiUpdater) Process(env *request.Environment) { if headRoot == s.lastHead { return } - if headBlock.Slot > reverseSyncHeaders { - s.headerSync.SetTailTarget(uint64(headBlock.Slot) - reverseSyncHeaders) - } else { - s.headerSync.SetTailTarget(0) - } head, err := s.chain.GetHeaderByHash(headRoot) if err != nil { return } + if uint64(headBlock.Slot) > s.tailTarget+reverseSyncHeaders { + s.tailTarget = uint64(headBlock.Slot) - reverseSyncHeaders + } var finalizedExecRoot common.Hash - if state, err := s.chain.GetStateProof(head); err == nil { + if state, err := s.chain.GetStateProof(head.Slot, head.StateRoot); err == nil { finalizedRoot := common.Hash(state.Values[finalizedBlockIndex]) if finalized, err := s.chain.GetHeaderByHash(finalizedRoot); err == nil { - if finalizedState, err := s.chain.GetStateProof(finalized); err == nil { + if finalizedState, err := s.chain.GetStateProof(finalized.Slot, finalized.StateRoot); err == nil { finalizedExecRoot = common.Hash(finalizedState.Values[execBlockIndex]) } - s.chain.Prune(finalized.Slot, true) + if finalized.Slot > s.tailTarget { + s.tailTarget = finalized.Slot + } } } else { if s.stateSync.HeadSyncPossible() {