beacon/light: request finality update explicitly when necessary

beacon/blsync/block_sync.go

@@ -139,7 +139,6 @@ func (s *beaconBlockSync) updateEventFeed() {
 			parent, ok := s.recentBlocks.Get(optimistic.Attested.ParentRoot)
 			if !ok || parent.Slot()/params.EpochLength == fe {
 				return // head is at first slot of next epoch, wait for finality update
-				//TODO: try to fetch finality update directly if subscription does not deliver
 			}
 		}
 	}

beacon/light/api/api_server.go

@@ -83,6 +83,9 @@ func (s *ApiServer) SendRequest(id request.ID, req request.Request) {
 		case sync.ReqBeaconBlock:
 			log.Debug("Beacon API: requesting block", "reqid", id, "hash", common.Hash(data))
 			resp, err = s.api.GetBeaconBlock(common.Hash(data))
+		case sync.ReqFinality:
+			log.Debug("Beacon API: requesting finality update")
+			resp, err = s.api.GetFinalityUpdate()
 		default:
 		}
 

beacon/light/api/light_api.go

@@ -548,7 +548,7 @@ func (api *BeaconLightApi) StartHeadListener(listener HeadEventListener) func()
 // established. It can only return nil when the context is canceled.
 func (api *BeaconLightApi) startEventStream(ctx context.Context, listener *HeadEventListener) *eventsource.Stream {
 	for retry := true; retry; retry = ctxSleep(ctx, 5*time.Second) {
-		path := "/eth/v1/events?topics=head&topics=light_client_optimistic_update&topics=light_client_finality_update"
+		path := "/eth/v1/events?topics=head&topics=light_client_finality_update&topics=light_client_optimistic_update"
 		log.Debug("Sending event subscription request")
 		req, err := http.NewRequestWithContext(ctx, "GET", api.url+path, nil)
 		if err != nil {

beacon/light/sync/head_sync.go

@@ -24,7 +24,8 @@ import (
 
 type headTracker interface {
 	ValidateOptimistic(update types.OptimisticUpdate) (bool, error)
-	ValidateFinality(update types.FinalityUpdate) (bool, error)
+	ValidateFinality(head types.FinalityUpdate) (bool, error)
+	ValidatedFinality() (types.FinalityUpdate, bool)
 	SetPrefetchHead(head types.HeadInfo)
 }
 
@@ -41,6 +42,7 @@ type HeadSync struct {
 	unvalidatedOptimistic map[request.Server]types.OptimisticUpdate
 	unvalidatedFinality   map[request.Server]types.FinalityUpdate
 	serverHeads           map[request.Server]types.HeadInfo
+	reqFinalityEpoch      map[request.Server]uint64 // next epoch to request finality update
 	headServerCount       map[types.HeadInfo]headServerCount
 	headCounter           uint64
 	prefetchHead          types.HeadInfo
@@ -64,6 +66,7 @@ func NewHeadSync(headTracker headTracker, chain committeeChain) *HeadSync {
 		unvalidatedFinality:   make(map[request.Server]types.FinalityUpdate),
 		serverHeads:           make(map[request.Server]types.HeadInfo),
 		headServerCount:       make(map[types.HeadInfo]headServerCount),
+		reqFinalityEpoch:      make(map[request.Server]uint64),
 	}
 	return s
 }
@@ -75,9 +78,22 @@ func (s *HeadSync) Process(requester request.Requester, events []request.Event)
 		case EvNewHead:
 			s.setServerHead(event.Server, event.Data.(types.HeadInfo))
 		case EvNewOptimisticUpdate:
-			s.newOptimisticUpdate(event.Server, event.Data.(types.OptimisticUpdate))
+			update := event.Data.(types.OptimisticUpdate)
+			s.newOptimisticUpdate(event.Server, update)
+			epoch := update.Attested.Epoch()
+			if epoch < s.reqFinalityEpoch[event.Server] {
+				continue
+			}
+			if finality, ok := s.headTracker.ValidatedFinality(); ok && finality.Attested.Header.Epoch() >= epoch {
+				continue
+			}
+			requester.Send(event.Server, ReqFinality{})
+			s.reqFinalityEpoch[event.Server] = epoch + 1
 		case EvNewFinalityUpdate:
 			s.newFinalityUpdate(event.Server, event.Data.(types.FinalityUpdate))
+		case request.EvResponse:
+			_, _, resp := event.RequestInfo()
+			s.newFinalityUpdate(event.Server, resp.(types.FinalityUpdate))
 		case request.EvUnregistered:
 			s.setServerHead(event.Server, types.HeadInfo{})
 			delete(s.serverHeads, event.Server)

beacon/light/sync/types.go

@@ -43,4 +43,5 @@ type (
 	}
 	ReqCheckpointData common.Hash
 	ReqBeaconBlock    common.Hash
+	ReqFinality       struct{}
 )