beacon/merkle: fixed unit test

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"
@@ -34,41 +32,25 @@ const (
 
 type SyncServer struct {
 	api         *BeaconLightApi
-	lock sync.RWMutex
-
 	unsubscribe func()
-	canRequestBootstrap          bool
-	firstUpdate, afterLastUpdate uint64
-	firstState                   uint64 //TODO ...
 }
 
 func NewSyncServer(api *BeaconLightApi) *SyncServer {
-	return &SyncServer{
+	return &SyncServer{api: api}
-		api:                 api,
-		canRequestBootstrap: true,
-	}
 }
 
 func (s *SyncServer) SubscribeHeads(newHead func(uint64, common.Hash), newSignedHead func(signedHead types.SignedHead)) {
-	s.lock.Lock()
+	s.unsubscribe = s.api.StartHeadListener(newHead, newSignedHead, func(err error) {
-	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) {
 		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 {
+func (s *SyncServer) RequestBeaconState(stateRoot common.Hash, format merkle.CompactProofFormat, response func(*merkle.MultiProof)) {
-	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)) {
 	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)
 		}
 	}()

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) {
+func (lc *LightChain) GetStateProof(slot uint64, stateRoot common.Hash) (merkle.MultiProof, error) {
-	if proof, ok := lc.stateProofCache.Get(slotAndHash{header.Slot, header.StateRoot}); ok {
+	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
 	}

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")
 		}

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)
+	CanSendTo(server *Server, moduleData *interface{}) (canSend bool, priority uint64)
-	SendTo(server *Server)
+	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
 		}
 	}

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)
 		}
 	}

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)

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{}),
 	}

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()

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(stateRoot common.Hash, format merkle.CompactProofFormat, response func(*merkle.MultiProof))
-	RequestBeaconState(slot uint64, 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) {
+func (r stateRequest) CanSendTo(server *request.Server, moduleData *interface{}) (canSend bool, priority uint64) {
-	if rs, ok := server.RequestServer.(beaconStateServer); !ok || r.header.Slot < rs.BeaconStateTail() {
+	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()

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) {
+func (r checkpointRequest) CanSendTo(server *request.Server, moduleData *interface{}) (canSend bool, priority uint64) {
-	if cs, ok := server.RequestServer.(checkpointInitServer); !ok || !cs.CanRequestBootstrap() {
+	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) {
+func (r updateRequest) CanSendTo(server *request.Server, moduleData *interface{}) (canSend bool, priority uint64) {
-	if us, ok := server.RequestServer.(updateServer); ok {
+	if _, ok := server.RequestServer.(updateServer); ok {
-		if updateRange := us.UpdateRange(); updateRange.Includes(r.first) {
+		firstUpdate, _ := (*moduleData).(uint64)
-			return true, updateRange.AfterLast
+		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()
+	headSlot, _ := server.LatestHead()
-	count := updateRange.AfterLast - r.first
+	afterLastUpdate := types.PeriodOfSlot(headSlot)
+	if afterLastUpdate <= r.first {
+		return
+	}
+	count := afterLastUpdate - r.first
 	if count > maxUpdateRequest {
 		count = maxUpdateRequest
 	}

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)

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() {