beacon/light, cmd/blsync: add finalized exec root feature

beacon/light/api/api_server.go

@@ -48,6 +48,9 @@ func (s *ApiServer) Subscribe(eventCallback func(event request.Event)) {
 	}, func(head types.SignedHeader) {
 		log.Debug("New signed head received", "slot", head.Header.Slot, "blockRoot", head.Header.Hash(), "signerCount", head.Signature.SignerCount())
 		eventCallback(request.Event{Type: sync.EvNewSignedHead, Data: head})
+	}, func(head types.FinalityUpdate) {
+		log.Debug("New finality update received", "slot", head.Attested.Slot, "blockRoot", head.Attested.Hash(), "signerCount", head.Signature.SignerCount())
+		eventCallback(request.Event{Type: sync.EvNewFinalityUpdate, Data: head})
 	}, func(err error) {
 		log.Warn("Head event stream error", "err", err)
 	})

beacon/light/api/light_api.go

@@ -67,6 +67,12 @@ type jsonBeaconHeader struct {
 	Beacon types.Header `json:"beacon"`
 }
 
+type jsonHeaderWithExecProof struct {
+	Beacon          types.Header                    `json:"beacon"`
+	Execution       *capella.ExecutionPayloadHeader `json:"execution"`
+	ExecutionBranch merkle.Values                   `json:"execution_branch"`
+}
+
 // UnmarshalJSON unmarshals from JSON.
 func (u *CommitteeUpdate) UnmarshalJSON(input []byte) error {
 	var dec committeeUpdateJson
@@ -225,6 +231,55 @@ func decodeOptimisticHeadUpdate(enc []byte) (types.SignedHeader, error) {
 	}, nil
 }
 
+// GetFinalityUpdate fetches the latest available finality update.
+//
+// See data structure definition here:
+// https://github.com/ethereum/consensus-specs/blob/dev/specs/altair/light-client/sync-protocol.md#lightclientfinalityupdate
+func (api *BeaconLightApi) GetFinalityUpdate() (types.FinalityUpdate, error) {
+	resp, err := api.httpGet("/eth/v1/beacon/light_client/finality_update")
+	if err != nil {
+		return types.FinalityUpdate{}, err
+	}
+	return decodeFinalityUpdate(resp)
+}
+
+func decodeFinalityUpdate(enc []byte) (types.FinalityUpdate, error) {
+	var data struct {
+		Data struct {
+			Attested       jsonHeaderWithExecProof `json:"attested_header"`
+			Finalized      jsonHeaderWithExecProof `json:"finalized_header"`
+			FinalityBranch merkle.Values           `json:"finality_branch"`
+			Aggregate      types.SyncAggregate     `json:"sync_aggregate"`
+			SignatureSlot  common.Decimal          `json:"signature_slot"`
+		} `json:"data"`
+	}
+	if err := json.Unmarshal(enc, &data); err != nil {
+		return types.FinalityUpdate{}, err
+	}
+
+	if len(data.Data.Aggregate.Signers) != params.SyncCommitteeBitmaskSize {
+		return types.FinalityUpdate{}, errors.New("invalid sync_committee_bits length")
+	}
+	if len(data.Data.Aggregate.Signature) != params.BLSSignatureSize {
+		return types.FinalityUpdate{}, errors.New("invalid sync_committee_signature length")
+	}
+	return types.FinalityUpdate{
+		Attested: types.HeaderWithExecProof{
+			Header:        data.Data.Attested.Beacon,
+			PayloadHeader: data.Data.Attested.Execution,
+			PayloadBranch: data.Data.Attested.ExecutionBranch,
+		},
+		Finalized: types.HeaderWithExecProof{
+			Header:        data.Data.Finalized.Beacon,
+			PayloadHeader: data.Data.Finalized.Execution,
+			PayloadBranch: data.Data.Finalized.ExecutionBranch,
+		},
+		FinalityBranch: data.Data.FinalityBranch,
+		Signature:      data.Data.Aggregate,
+		SignatureSlot:  uint64(data.Data.SignatureSlot),
+	}, nil
+}
+
 // GetHead fetches and validates the beacon header with the given blockRoot.
 // If blockRoot is null hash then the latest head header is fetched.
 func (api *BeaconLightApi) GetHeader(blockRoot common.Hash) (types.Header, error) {
@@ -343,7 +398,7 @@ func decodeHeadEvent(enc []byte) (uint64, common.Hash, error) {
 // head updates and calls the specified callback functions when they are received.
 // The callbacks are also called for the current head and optimistic head at startup.
 // They are never called concurrently.
-func (api *BeaconLightApi) StartHeadListener(headFn func(slot uint64, blockRoot common.Hash), signedFn func(head types.SignedHeader), errFn func(err error)) func() {
+func (api *BeaconLightApi) StartHeadListener(headFn func(slot uint64, blockRoot common.Hash), signedFn func(head types.SignedHeader), finalityFn func(head types.FinalityUpdate), errFn func(err error)) func() {
 	closeCh := make(chan struct{})   // initiate closing the stream
 	closedCh := make(chan struct{})  // stream closed (or failed to create)
 	stoppedCh := make(chan struct{}) // sync loop stopped
@@ -354,7 +409,8 @@ func (api *BeaconLightApi) StartHeadListener(headFn func(slot uint64, blockRoot
 		// first actual event arrives; therefore we create the subscription in
 		// a separate goroutine while letting the main goroutine sync up to the
 		// current head
-		req, err := http.NewRequest("GET", api.url+"/eth/v1/events?topics=head&topics=light_client_optimistic_update", nil)
+		req, err := http.NewRequest("GET", api.url+
+			"/eth/v1/events?topics=head&topics=light_client_optimistic_update&topics=light_client_finality_update", nil)
 		if err != nil {
 			errFn(fmt.Errorf("Error creating event subscription request: %v", err))
 			return
@@ -381,6 +437,9 @@ func (api *BeaconLightApi) StartHeadListener(headFn func(slot uint64, blockRoot
 		if signedHead, err := api.GetOptimisticHeadUpdate(); err == nil {
 			signedFn(signedHead)
 		}
+		if finalityUpdate, err := api.GetFinalityUpdate(); err == nil {
+			finalityFn(finalityUpdate)
+		}
 		stream := <-streamCh
 		if stream == nil {
 			return
@@ -404,6 +463,12 @@ func (api *BeaconLightApi) StartHeadListener(headFn func(slot uint64, blockRoot
 					} else {
 						errFn(fmt.Errorf("Error decoding optimistic update event: %v", err))
 					}
+				case "light_client_finality_update":
+					if finalityUpdate, err := decodeFinalityUpdate([]byte(event.Data())); err == nil {
+						finalityFn(finalityUpdate)
+					} else {
+						errFn(fmt.Errorf("Error decoding finality update event: %v", err))
+					}
 				default:
 					errFn(fmt.Errorf("Unexpected event: %s", event.Event()))
 				}

beacon/light/head_tracker.go

@@ -33,7 +33,9 @@ type HeadTracker struct {
 	committeeChain    *CommitteeChain
 	minSignerCount    int
 	signedHead        types.SignedHeader
-	headSignerCount int
+	hasSignedHead     bool
+	finalityUpdate    types.FinalityUpdate
+	hasFinalityUpdate bool
 	prefetchHead      types.HeadInfo
 	changeCounter     uint64
 }
@@ -47,26 +49,55 @@ func NewHeadTracker(committeeChain *CommitteeChain, minSignerCount int) *HeadTra
 }
 
 // ValidatedHead returns the latest validated head.
-func (h *HeadTracker) ValidatedHead() types.SignedHeader {
+func (h *HeadTracker) ValidatedHead() (types.SignedHeader, bool) {
 	h.lock.RLock()
 	defer h.lock.RUnlock()
 
-	return h.signedHead
+	return h.signedHead, h.hasSignedHead
+}
+
+// ValidatedHead returns the latest validated head.
+func (h *HeadTracker) ValidatedFinality() (types.FinalityUpdate, bool) {
+	h.lock.RLock()
+	defer h.lock.RUnlock()
+
+	return h.finalityUpdate, h.hasFinalityUpdate
 }
 
 // Validate validates the given signed head. If the head is successfully validated
 // and it is better than the old validated head (higher slot or same slot and more
 // signers) then ValidatedHead is updated. The boolean return flag signals if
 // ValidatedHead has been changed.
-func (h *HeadTracker) Validate(head types.SignedHeader) (bool, error) {
+func (h *HeadTracker) ValidateHead(head types.SignedHeader) (bool, error) {
 	h.lock.Lock()
 	defer h.lock.Unlock()
 
+	replace, err := h.validate(head, h.signedHead)
+	if replace {
+		h.signedHead, h.hasSignedHead = head, true
+		h.changeCounter++
+	}
+	return replace, err
+}
+
+func (h *HeadTracker) ValidateFinality(update types.FinalityUpdate) (bool, error) {
+	h.lock.Lock()
+	defer h.lock.Unlock()
+
+	replace, err := h.validate(update.SignedHeader(), h.finalityUpdate.SignedHeader())
+	if replace {
+		h.finalityUpdate, h.hasFinalityUpdate = update, true
+		h.changeCounter++
+	}
+	return replace, err
+}
+
+func (h *HeadTracker) validate(head, oldHead types.SignedHeader) (bool, error) {
 	signerCount := head.Signature.SignerCount()
 	if signerCount < h.minSignerCount {
 		return false, errors.New("low signer count")
 	}
-	if head.Header.Slot < h.signedHead.Header.Slot || (head.Header.Slot == h.signedHead.Header.Slot && signerCount <= h.headSignerCount) {
+	if head.Header.Slot < oldHead.Header.Slot || (head.Header.Slot == oldHead.Header.Slot && signerCount <= oldHead.Signature.SignerCount()) {
 		return false, nil
 	}
 	sigOk, age, err := h.committeeChain.VerifySignedHeader(head)
@@ -82,8 +113,6 @@ func (h *HeadTracker) Validate(head types.SignedHeader) (bool, error) {
 	if !sigOk {
 		return false, errors.New("invalid header signature")
 	}
-	h.signedHead, h.headSignerCount = head, signerCount
-	h.changeCounter++
 	return true, nil
 }
 

beacon/light/sync/head_sync.go

@@ -22,7 +22,8 @@ import (
 )
 
 type headTracker interface {
-	Validate(head types.SignedHeader) (bool, error)
+	ValidateHead(head types.SignedHeader) (bool, error)
+	ValidateFinality(head types.FinalityUpdate) (bool, error)
 	SetPrefetchHead(head types.HeadInfo)
 }
 
@@ -37,6 +38,7 @@ type HeadSync struct {
 	nextSyncPeriod      uint64
 	chainInit           bool
 	unvalidatedHeads    map[request.Server]types.SignedHeader
+	unvalidatedFinality map[request.Server]types.FinalityUpdate
 	serverHeads         map[request.Server]types.HeadInfo
 	headServerCount     map[types.HeadInfo]headServerCount
 	headCounter         uint64
@@ -58,6 +60,7 @@ func NewHeadSync(headTracker headTracker, chain committeeChain) *HeadSync {
 		headTracker:         headTracker,
 		chain:               chain,
 		unvalidatedHeads:    make(map[request.Server]types.SignedHeader),
+		unvalidatedFinality: make(map[request.Server]types.FinalityUpdate),
 		serverHeads:         make(map[request.Server]types.HeadInfo),
 		headServerCount:     make(map[types.HeadInfo]headServerCount),
 	}
@@ -71,6 +74,8 @@ func (s *HeadSync) Process(events []request.Event) {
 			s.setServerHead(event.Server, event.Data.(types.HeadInfo))
 		case EvNewSignedHead:
 			s.newSignedHead(event.Server, event.Data.(types.SignedHeader))
+		case EvNewFinalityUpdate:
+			s.newFinalityUpdate(event.Server, event.Data.(types.FinalityUpdate))
 		case request.EvUnregistered:
 			s.setServerHead(event.Server, types.HeadInfo{})
 			delete(s.serverHeads, event.Server)
@@ -81,7 +86,7 @@ func (s *HeadSync) Process(events []request.Event) {
 	nextPeriod, chainInit := s.chain.NextSyncPeriod()
 	if nextPeriod != s.nextSyncPeriod || chainInit != s.chainInit {
 		s.nextSyncPeriod, s.chainInit = nextPeriod, chainInit
-		s.processUnvalidatedHeads()
+		s.processUnvalidated()
 	}
 }
 
@@ -96,21 +101,37 @@ func (s *HeadSync) newSignedHead(server request.Server, signedHead types.SignedH
 		s.unvalidatedHeads[server] = signedHead
 		return
 	}
-	s.headTracker.Validate(signedHead)
+	s.headTracker.ValidateHead(signedHead)
+}
+
+// newSignedHead handles received signed head; either validates it if the chain
+// is properly synced or stores it for further validation.
+func (s *HeadSync) newFinalityUpdate(server request.Server, finalityUpdate types.FinalityUpdate) {
+	if !s.chainInit || types.SyncPeriod(finalityUpdate.SignatureSlot) > s.nextSyncPeriod {
+		s.unvalidatedFinality[server] = finalityUpdate
+		return
+	}
+	s.headTracker.ValidateFinality(finalityUpdate)
 }
 
 // processUnvalidatedHeads iterates the list of unvalidated heads and validates
 // those which can be validated.
-func (s *HeadSync) processUnvalidatedHeads() {
+func (s *HeadSync) processUnvalidated() {
 	if !s.chainInit {
 		return
 	}
 	for server, signedHead := range s.unvalidatedHeads {
 		if types.SyncPeriod(signedHead.SignatureSlot) <= s.nextSyncPeriod {
-			s.headTracker.Validate(signedHead)
+			s.headTracker.ValidateHead(signedHead)
 			delete(s.unvalidatedHeads, server)
 		}
 	}
+	for server, finalityUpdate := range s.unvalidatedFinality {
+		if types.SyncPeriod(finalityUpdate.SignatureSlot) <= s.nextSyncPeriod {
+			s.headTracker.ValidateFinality(finalityUpdate)
+			delete(s.unvalidatedFinality, server)
+		}
+	}
 }
 
 // setServerHead processes non-validated server head announcements and updates

beacon/light/sync/test_helpers.go

@@ -229,11 +229,16 @@ type TestHeadTracker struct {
 	validated []types.SignedHeader
 }
 
-func (ht *TestHeadTracker) Validate(head types.SignedHeader) (bool, error) {
+func (ht *TestHeadTracker) ValidateHead(head types.SignedHeader) (bool, error) {
 	ht.validated = append(ht.validated, head)
 	return true, nil
 }
 
+//TODO add test case for finality
+func (ht *TestHeadTracker) ValidateFinality(head types.FinalityUpdate) (bool, error) {
+	return true, nil
+}
+
 func (ht *TestHeadTracker) ExpValidated(t *testing.T, tci int, expHeads []types.SignedHeader) {
 	for i, expHead := range expHeads {
 		if i >= len(ht.validated) {

beacon/light/sync/types.go

@@ -25,6 +25,7 @@ import (
 var (
 	EvNewHead           = &request.EventType{Name: "newHead"}           // data: types.HeadInfo
 	EvNewSignedHead     = &request.EventType{Name: "newSignedHead"}     // data: types.SignedHeader
+	EvNewFinalityUpdate = &request.EventType{Name: "newFinalityUpdate"} // data: types.FinalityUpdate
 )
 
 type (

beacon/params/params.go

@@ -41,4 +41,6 @@ const (
 	StateIndexNextSyncCommittee = 55
 	StateIndexExecPayload       = 56
 	StateIndexExecHead          = 908
+
+	BodyIndexExecPayload = 25
 )

beacon/types/light_sync.go

@@ -23,6 +23,8 @@ import (
 	"github.com/ethereum/go-ethereum/beacon/merkle"
 	"github.com/ethereum/go-ethereum/beacon/params"
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/protolambda/zrnt/eth2/beacon/capella"
+	"github.com/protolambda/ztyp/tree"
 )
 
 // HeadInfo represents an unvalidated new head announcement.
@@ -140,3 +142,42 @@ func (u UpdateScore) BetterThan(w UpdateScore) bool {
 	}
 	return u.SignerCount > w.SignerCount
 }
+
+type HeaderWithExecProof struct {
+	Header
+	PayloadHeader *capella.ExecutionPayloadHeader
+	PayloadBranch merkle.Values
+}
+
+func (h *HeaderWithExecProof) Validate() error {
+	payloadRoot := merkle.Value(h.PayloadHeader.HashTreeRoot(tree.GetHashFn()))
+	return merkle.VerifyProof(h.BodyRoot, params.BodyIndexExecPayload, h.PayloadBranch, payloadRoot)
+}
+
+type FinalityUpdate struct {
+	Attested, Finalized HeaderWithExecProof
+	FinalityBranch      merkle.Values
+	// Sync committee BLS signature aggregate
+	Signature SyncAggregate
+	// Slot in which the signature has been created (newer than Header.Slot,
+	// determines the signing sync committee)
+	SignatureSlot uint64
+}
+
+func (u *FinalityUpdate) SignedHeader() SignedHeader {
+	return SignedHeader{
+		Header:        u.Attested.Header,
+		Signature:     u.Signature,
+		SignatureSlot: u.SignatureSlot,
+	}
+}
+
+func (u *FinalityUpdate) Validate() error {
+	if err := u.Attested.Validate(); err != nil {
+		return err
+	}
+	if err := u.Finalized.Validate(); err != nil {
+		return err
+	}
+	return merkle.VerifyProof(u.Attested.StateRoot, params.StateIndexFinalBlock, u.FinalityBranch, merkle.Value(u.Finalized.Hash()))
+}

cmd/blsync/block_sync.go

@@ -34,12 +34,18 @@ type beaconBlockSync struct {
 	headTracker  headTracker
 
 	lastHeadBlock *capella.BeaconBlock
-	headBlockCh   chan *capella.BeaconBlock
+	headCh   chan headData
+}
+
+type headData struct {
+	block  *capella.BeaconBlock
+	update types.FinalityUpdate
 }
 
 type headTracker interface {
 	PrefetchHead() types.HeadInfo
-	ValidatedHead() types.SignedHeader
+	ValidatedHead() (types.SignedHeader, bool)
+	ValidatedFinality() (types.FinalityUpdate, bool)
 }
 
 // newBeaconBlockSync returns a new beaconBlockSync.
@@ -49,7 +55,7 @@ func newBeaconBlockSync(headTracker headTracker) *beaconBlockSync {
 		recentBlocks: lru.NewCache[common.Hash, *capella.BeaconBlock](10),
 		locked:       make(map[common.Hash]struct{}),
 		serverHeads:  make(map[request.Server]common.Hash),
-		headBlockCh:  make(chan *capella.BeaconBlock, 1),
+		headCh:  make(chan headData, 1),
 	}
 }
 
@@ -76,21 +82,29 @@ func (s *beaconBlockSync) Process(events []request.Event) {
 	}
 
 	// send validated head block
-	if vh := s.headTracker.ValidatedHead(); vh != (types.SignedHeader{}) {
+	head, ok := s.headTracker.ValidatedHead()
-		validatedHead := vh.Header.Hash()
+	if !ok {
-		if headBlock, ok := s.recentBlocks.Get(validatedHead); ok && headBlock != s.lastHeadBlock {
+		return
+	}
+	finality, ok := s.headTracker.ValidatedFinality()	//TODO fetch directly if subscription does not deliver
+	if !ok || head.Header.Epoch() != finality.Attested.Header.Epoch() {
+		return
+	}
+	validatedHead := head.Header.Hash()
+	headBlock, ok := s.recentBlocks.Get(validatedHead)
+	if !ok || headBlock == s.lastHeadBlock {
+		return
+	}
 	select {
-			case s.headBlockCh <- headBlock:
+	case s.headCh <- headData{block: headBlock, update: finality}:
 		s.lastHeadBlock = headBlock
 	default:
 	}
-		}
-	}
 }
 
 func (s *beaconBlockSync) MakeRequest(server request.Server) (request.Request, float32) {
 	// request validated head block if unavailable and not yet requested
-	if vh := s.headTracker.ValidatedHead(); vh != (types.SignedHeader{}) {
+	if vh, ok := s.headTracker.ValidatedHead(); ok {
 		validatedHead := vh.Header.Hash()
 		if _, ok := s.recentBlocks.Get(validatedHead); !ok {
 			if _, ok := s.locked[validatedHead]; !ok {

cmd/blsync/block_sync_test.go

@@ -45,12 +45,12 @@ func TestBlockSync(t *testing.T) {
 
 	expHeadBlock := func(tci int, expHead *capella.BeaconBlock) {
 		expInfo := blockHeadInfo(expHead)
-		var head *capella.BeaconBlock
+		var head headData
 		select {
-		case head = <-blockSync.headBlockCh:
+		case head = <-blockSync.headCh:
 		default:
 		}
-		headInfo := blockHeadInfo(head)
+		headInfo := blockHeadInfo(head.block)
 		if headInfo != expInfo {
 			t.Errorf("Wrong head block in test case #%d (expected {slot %d blockRoot %x}, got {slot %d blockRoot %x})", tci, expInfo.Slot, expInfo.BlockRoot, headInfo.Slot, headInfo.BlockRoot)
 		}
@@ -129,6 +129,15 @@ func (h *testHeadTracker) PrefetchHead() types.HeadInfo {
 	return h.prefetch
 }
 
-func (h *testHeadTracker) ValidatedHead() types.SignedHeader {
+func (h *testHeadTracker) ValidatedHead() (types.SignedHeader, bool) {
-	return h.validated
+	return h.validated, h.validated.Header != (types.Header{})
+}
+
+//TODO add test case for finality
+func (h *testHeadTracker) ValidatedFinality() (types.FinalityUpdate, bool) {
+	return types.FinalityUpdate{
+		Attested:      types.HeaderWithExecProof{Header: h.validated.Header},
+		Signature:     h.validated.Signature,
+		SignatureSlot: h.validated.SignatureSlot,
+	}, h.validated.Header != (types.Header{})
 }

cmd/blsync/engine_api.go

@@ -35,23 +35,24 @@ import (
 	"github.com/protolambda/ztyp/tree"
 )
 
-func updateEngineApi(client *rpc.Client, headBlockCh chan *capella.BeaconBlock) {
+func updateEngineApi(client *rpc.Client, headCh chan headData) {
-	for headBlock := range headBlockCh {
+	for headData := range headCh {
-		execBlock, err := getExecBlock(headBlock)
+		execBlock, err := getExecBlock(headData.block)
 		if err != nil {
 			log.Error("Error extracting execution block from validated beacon block", "error", err)
 			continue
 		}
 		execRoot := execBlock.Hash()
+		finalizedRoot := common.Hash(headData.update.Finalized.PayloadHeader.BlockHash)
 		if client == nil { // dry run, no engine API specified
-			log.Info("New execution block retrieved", "block number", execBlock.NumberU64(), "block hash", execRoot)
+			log.Info("New execution block retrieved", "block number", execBlock.NumberU64(), "block hash", execRoot, "finalized block hash", finalizedRoot)
 		} else {
 			if status, err := callNewPayloadV2(client, execBlock); err == nil {
 				log.Info("Successful NewPayload", "block number", execBlock.NumberU64(), "block hash", execRoot, "status", status)
 			} else {
 				log.Error("Failed NewPayload", "block number", execBlock.NumberU64(), "block hash", execRoot, "error", err)
 			}
-			if status, err := callForkchoiceUpdatedV1(client, execRoot, common.Hash{}); err == nil {
+			if status, err := callForkchoiceUpdatedV1(client, execRoot, finalizedRoot); err == nil {
 				log.Info("Successful ForkchoiceUpdated", "head", execRoot, "status", status)
 			} else {
 				log.Error("Failed ForkchoiceUpdated", "head", execRoot, "error", err)

cmd/blsync/main.go

@@ -126,7 +126,7 @@ func blsync(ctx *cli.Context) error {
 	scheduler.RegisterModule(forwardSync, "forwardSync")
 	scheduler.RegisterModule(headSync, "headSync")
 	scheduler.RegisterModule(beaconBlockSync, "beaconBlockSync")
-	go updateEngineApi(makeRPCClient(ctx), beaconBlockSync.headBlockCh)
+	go updateEngineApi(makeRPCClient(ctx), beaconBlockSync.headCh)
 	// start
 	scheduler.Start()
 	// register server(s)
@@ -137,6 +137,6 @@ func blsync(ctx *cli.Context) error {
 	// run until stopped
 	<-ctx.Done()
 	scheduler.Stop()
-	close(beaconBlockSync.headBlockCh)
+	close(beaconBlockSync.headCh)
 	return nil
 }