Sync execution update on demand

.gitignore

@@ -30,3 +30,5 @@ contracts/beefy-state.json
 
 go/
 gocache/
+go.work*
+control/target/

relayer/chain/ethereum/header.go

@@ -5,11 +5,12 @@ package ethereum
 
 import (
 	"fmt"
+	"math/big"
+
 	etypes "github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/snowfork/go-substrate-rpc-client/v4/scale"
 	types "github.com/snowfork/go-substrate-rpc-client/v4/types"
-	"math/big"
 )
 
 type HeaderID struct {

relayer/chain/parachain/writer.go

@@ -385,3 +385,24 @@ func (wr *ParachainWriter) getNumberFromParachain(pallet, storage string) (uint6
 
 	return uint64(number), nil
 }
+
+func (wr *ParachainWriter) GetCompactExecutionHeaderStateByBlockHash(blockHash types.H256) (state.CompactExecutionHeaderState, error) {
+	var headerState state.CompactExecutionHeaderState
+	key, err := types.CreateStorageKey(wr.conn.Metadata(), "EthereumBeaconClient", "ExecutionHeaders", blockHash[:], nil)
+	if err != nil {
+		return headerState, fmt.Errorf("create storage key for ExecutionHeaders: %w", err)
+	}
+
+	var compactExecutionHeader scale.CompactExecutionHeader
+	_, err = wr.conn.API().RPC.State.GetStorageLatest(key, &compactExecutionHeader)
+	if err != nil {
+		return headerState, fmt.Errorf("get storage for ExecutionHeaders (err): %w", err)
+	}
+	headerState = state.CompactExecutionHeaderState{
+		ParentHash:   common.Hash(compactExecutionHeader.ParentHash),
+		BlockNumber:  uint64(compactExecutionHeader.BlockNumber.Int64()),
+		StateRoot:    common.Hash(compactExecutionHeader.StateRoot),
+		ReceiptsRoot: common.Hash(compactExecutionHeader.ReceiptsRoot),
+	}
+	return headerState, nil
+}

relayer/cmd/generate_beacon_data.go

@@ -53,7 +53,7 @@ func generateBeaconCheckpointCmd() *cobra.Command {
 	}
 
 	cmd.Flags().String("url", "http://127.0.0.1:9596", "Beacon URL")
-	cmd.Flags().Bool("export_json", false, "Export Json")
+	cmd.Flags().Bool("export-json", false, "Export Json")
 
 	return cmd
 }
@@ -115,9 +115,9 @@ func generateBeaconCheckpoint(cmd *cobra.Command, _ []string) error {
 		if err != nil {
 			return fmt.Errorf("get initial sync: %w", err)
 		}
-		exportJson, err := cmd.Flags().GetBool("export_json")
+		exportJson, err := cmd.Flags().GetBool("export-json")
 		if err != nil {
-			return err
+			return fmt.Errorf("get initial sync: %w", err)
 		}
 		if exportJson {
 			initialSync := checkPointScale.ToJSON()
@@ -282,7 +282,7 @@ func generateBeaconTestFixture(cmd *cobra.Command, _ []string) error {
 			BlockRootsTree:     finalizedUpdateAfterMessage.BlockRootsTree,
 			Slot:               uint64(finalizedUpdateAfterMessage.Payload.FinalizedHeader.Slot),
 		}
-		headerUpdateScale, err := s.GetNextHeaderUpdateBySlotWithCheckpoint(beaconBlockSlot, &checkPoint)
+		headerUpdateScale, err := s.GetHeaderUpdateBySlotWithCheckpoint(beaconBlockSlot, &checkPoint)
 		if err != nil {
 			return fmt.Errorf("get header update: %w", err)
 		}

relayer/relays/beacon/header/header.go

@@ -6,11 +6,12 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/snowfork/go-substrate-rpc-client/v4/types"
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/snowfork/snowbridge/relayer/relays/beacon/config"
 	"github.com/snowfork/snowbridge/relayer/relays/beacon/header/syncer/scale"
 
 	log "github.com/sirupsen/logrus"
+	"github.com/snowfork/go-substrate-rpc-client/v4/types"
 	"github.com/snowfork/snowbridge/relayer/chain/parachain"
 	"github.com/snowfork/snowbridge/relayer/relays/beacon/cache"
 	"github.com/snowfork/snowbridge/relayer/relays/beacon/header/syncer"
@@ -186,11 +187,6 @@ func (h *Header) SyncFinalizedHeader(ctx context.Context) error {
 }
 
 func (h *Header) SyncHeaders(ctx context.Context) error {
-	err := h.SyncExecutionHeaders(ctx)
-	if err != nil {
-		return err
-	}
-
 	hasChanged, err := h.syncer.HasFinalizedHeaderChanged(h.cache.Finalized.LastSyncedHash)
 	if err != nil {
 		return err
@@ -208,79 +204,12 @@ func (h *Header) SyncHeaders(ctx context.Context) error {
 	return nil
 }
 
-func (h *Header) SyncExecutionHeaders(ctx context.Context) error {
-	fromSlot := h.cache.LastSyncedExecutionSlot
-	// SyncExecutionHeaders at least from initial checkpoint
-	if fromSlot <= h.cache.InitialCheckpointSlot {
-		fromSlot = h.cache.InitialCheckpointSlot
-	}
-	toSlot := h.cache.Finalized.LastSyncedSlot
-	if fromSlot >= toSlot {
-		log.WithFields(log.Fields{
-			"fromSlot": fromSlot,
-			"toSlot":   toSlot,
-		}).Info("execution headers sync up to date with last finalized header")
-		return nil
-	}
-	log.WithFields(log.Fields{
-		"fromSlot":   fromSlot,
-		"fromEpoch":  h.syncer.ComputeEpochAtSlot(fromSlot),
-		"toSlot":     toSlot,
-		"toEpoch":    h.syncer.ComputeEpochAtSlot(toSlot),
-		"totalSlots": toSlot - fromSlot,
-	}).Info("starting to back-fill headers")
-
-	var headersToSync []scale.HeaderUpdatePayload
-
-	// start syncing at next block after last synced block
-	currentSlot := fromSlot
-	headerUpdate, err := h.getNextHeaderUpdateBySlot(currentSlot)
-	if err != nil {
-		return fmt.Errorf("get next header update by slot with ancestry proof: %w", err)
-	}
-	currentSlot = uint64(headerUpdate.Header.Slot)
-
-	for currentSlot <= toSlot {
-		log.WithFields(log.Fields{
-			"currentSlot": currentSlot,
-		}).Info("fetching next header at slot")
-
-		var nextHeaderUpdate scale.HeaderUpdatePayload
-		if currentSlot >= toSlot {
-			// Just construct an empty update so to break the loop
-			nextHeaderUpdate = scale.HeaderUpdatePayload{Header: scale.BeaconHeader{Slot: types.U64(toSlot + 1)}}
-		} else {
-			// To get the sync witness for the current synced header. This header
-			// will be used as the next update.
-			nextHeaderUpdate, err = h.getNextHeaderUpdateBySlot(currentSlot)
-			if err != nil {
-				return fmt.Errorf("get next header update by slot with ancestry proof: %w", err)
-			}
-		}
-
-		headersToSync = append(headersToSync, headerUpdate)
-		// last slot to be synced, sync headers
-		if currentSlot >= toSlot {
-			err = h.batchSyncHeaders(ctx, headersToSync)
-			if err != nil {
-				return fmt.Errorf("batch sync headers failed: %w", err)
-			}
-		}
-		headerUpdate = nextHeaderUpdate
-		currentSlot = uint64(headerUpdate.Header.Slot)
-	}
-	// waiting for all batch calls to be executed on chain
-	err = h.waitingForBatchCallFinished(toSlot)
-	if err != nil {
-		return err
-	}
-	h.cache.SetLastSyncedExecutionSlot(toSlot)
-	return nil
-}
-
 func (h *Header) syncLaggingSyncCommitteePeriods(ctx context.Context, latestSyncedPeriod, currentSyncPeriod uint64) error {
-	// sync for the next period
-	periodsToSync := []uint64{latestSyncedPeriod + 1}
+	// sync for all missing periods
+	periodsToSync := []uint64{}
+	for i := latestSyncedPeriod + 1; i <= currentSyncPeriod; i++ {
+		periodsToSync = append(periodsToSync, i)
+	}
 
 	// For initialPeriod special handling here to sync it again for nextSyncCommittee which is not included in InitCheckpoint
 	if h.isInitialSyncPeriod() {
@@ -353,6 +282,13 @@ func (h *Header) populateClosestCheckpoint(slot uint64) (cache.Proof, error) {
 		checkpointSlot := checkpoint.Slot
 		if checkpointSlot == 0 {
 			checkpointSlot = h.syncer.CalculateNextCheckpointSlot(slot)
+			lastFinalizedHeaderState, err := h.writer.GetLastFinalizedHeaderState()
+			if err != nil {
+				return cache.Proof{}, fmt.Errorf("fetch parachain last finalized header state: %w", err)
+			}
+			if checkpointSlot > lastFinalizedHeaderState.BeaconSlot {
+				checkpointSlot = lastFinalizedHeaderState.BeaconSlot
+			}
 			log.WithFields(log.Fields{"calculatedCheckpointSlot": checkpointSlot}).Info("checkpoint slot not available, try with slot in next sync period instead")
 		}
 		err := h.populateFinalizedCheckpoint(checkpointSlot)
@@ -377,8 +313,7 @@ func (h *Header) populateClosestCheckpoint(slot uint64) (cache.Proof, error) {
 	return checkpoint, nil
 }
 
-func (h *Header) getNextHeaderUpdateBySlot(slot uint64) (scale.HeaderUpdatePayload, error) {
-	slot = slot + 1
+func (h *Header) getHeaderUpdateBySlot(slot uint64) (scale.HeaderUpdatePayload, error) {
 	header, err := h.syncer.FindBeaconHeaderWithBlockIncluded(slot)
 	if err != nil {
 		return scale.HeaderUpdatePayload{}, fmt.Errorf("get next beacon header with block included: %w", err)
@@ -394,14 +329,33 @@ func (h *Header) getNextHeaderUpdateBySlot(slot uint64) (scale.HeaderUpdatePaylo
 	return h.syncer.GetHeaderUpdate(blockRoot, &checkpoint)
 }
 
-func (h *Header) batchSyncHeaders(ctx context.Context, headerUpdates []scale.HeaderUpdatePayload) error {
-	headerUpdatesInf := make([]interface{}, len(headerUpdates))
-	for i, v := range headerUpdates {
-		headerUpdatesInf[i] = v
+func (h *Header) SyncExecutionHeader(ctx context.Context, blockRoot common.Hash) error {
+	header, err := h.syncer.Client.GetHeader(blockRoot)
+	if err != nil {
+		return fmt.Errorf("get beacon header by blockRoot: %w", err)
 	}
-	err := h.writer.BatchCall(ctx, "EthereumBeaconClient.submit_execution_header", headerUpdatesInf)
+	headerUpdate, err := h.getHeaderUpdateBySlot(header.Slot)
 	if err != nil {
-		return err
+		return fmt.Errorf("get header update by slot with ancestry proof: %w", err)
+	}
+	err = h.writer.WriteToParachainAndWatch(ctx, "EthereumBeaconClient.submit_execution_header", headerUpdate)
+	if err != nil {
+		return fmt.Errorf("submit_execution_header: %w", err)
+	}
+	var blockHash types.H256
+	if headerUpdate.ExecutionHeader.Deneb != nil {
+		blockHash = headerUpdate.ExecutionHeader.Deneb.BlockHash
+	} else if headerUpdate.ExecutionHeader.Capella != nil {
+		blockHash = headerUpdate.ExecutionHeader.Capella.BlockHash
+	} else {
+		return fmt.Errorf("invalid blockHash in headerUpdate")
+	}
+	compactExecutionHeaderState, err := h.writer.GetCompactExecutionHeaderStateByBlockHash(blockHash)
+	if err != nil {
+		return fmt.Errorf("get compactExecutionHeaderState by blockHash: %w", err)
+	}
+	if compactExecutionHeaderState.BlockNumber == 0 {
+		return fmt.Errorf("invalid compactExecutionHeaderState")
 	}
 	return nil
 }
@@ -411,24 +365,3 @@ func (h *Header) isInitialSyncPeriod() bool {
 	lastFinalizedPeriod := h.syncer.ComputeSyncPeriodAtSlot(h.cache.Finalized.LastSyncedSlot)
 	return initialPeriod == lastFinalizedPeriod
 }
-
-func (h *Header) waitingForBatchCallFinished(toSlot uint64) error {
-	batchCallFinished := false
-	cnt := 0
-	for cnt <= 12 {
-		executionHeaderState, err := h.writer.GetLastExecutionHeaderState()
-		if err != nil {
-			return fmt.Errorf("fetch last execution hash: %w", err)
-		}
-		if executionHeaderState.BeaconSlot == toSlot {
-			batchCallFinished = true
-			break
-		}
-		time.Sleep(6 * time.Second)
-		cnt++
-	}
-	if !batchCallFinished {
-		return ErrExecutionHeaderNotImported
-	}
-	return nil
-}

relayer/relays/beacon/header/syncer/api/api.go

@@ -14,17 +14,14 @@ import (
 	"github.com/snowfork/snowbridge/relayer/relays/util"
 )
 
-const (
-	ConstructRequestErrorMessage = "construct header request"
-	DoHTTPRequestErrorMessage    = "do http request"
-	HTTPStatusNotOKErrorMessage  = "http status not ok"
-	ReadResponseBodyErrorMessage = "read response body"
-	UnmarshalBodyErrorMessage    = "unmarshal body"
-)
-
 var (
 	ErrNotFound                        = errors.New("not found")
 	ErrSyncCommitteeUpdateNotAvailable = errors.New("no sync committee update available")
+	ConstructRequestErrorMessage       = "construct header request"
+	DoHTTPRequestErrorMessage          = "do http request"
+	HTTPStatusNotOKErrorMessage        = "http status not ok"
+	ReadResponseBodyErrorMessage       = "read response body"
+	UnmarshalBodyErrorMessage          = "unmarshal body"
 )
 
 type BeaconClient struct {
@@ -42,30 +39,30 @@ func NewBeaconClient(endpoint string, slotsInEpoch uint64) *BeaconClient {
 }
 
 func (b *BeaconClient) GetBootstrap(blockRoot common.Hash) (BootstrapResponse, error) {
+	var response BootstrapResponse
 	req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("%s/eth/v1/beacon/light_client/bootstrap/%s", b.endpoint, blockRoot), nil)
 	if err != nil {
-		return BootstrapResponse{}, fmt.Errorf("%s: %w", ConstructRequestErrorMessage, err)
+		return response, fmt.Errorf("%s: %w", ConstructRequestErrorMessage, err)
 	}
 
 	req.Header.Set("accept", "application/json")
 	res, err := b.httpClient.Do(req)
 	if err != nil {
-		return BootstrapResponse{}, fmt.Errorf("%s: %w", DoHTTPRequestErrorMessage, err)
+		return response, fmt.Errorf("%s: %w", DoHTTPRequestErrorMessage, err)
 	}
 
 	if res.StatusCode != http.StatusOK {
-		return BootstrapResponse{}, fmt.Errorf("%s: %d", HTTPStatusNotOKErrorMessage, res.StatusCode)
+		return response, fmt.Errorf("%s: %d", HTTPStatusNotOKErrorMessage, res.StatusCode)
 	}
 
 	bodyBytes, err := io.ReadAll(res.Body)
 	if err != nil {
-		return BootstrapResponse{}, fmt.Errorf("%s: %w", ReadResponseBodyErrorMessage, err)
+		return response, fmt.Errorf("%s: %w", ReadResponseBodyErrorMessage, err)
 	}
 
-	var response BootstrapResponse
 	err = json.Unmarshal(bodyBytes, &response)
 	if err != nil {
-		return BootstrapResponse{}, fmt.Errorf("%s: %w", UnmarshalBodyErrorMessage, err)
+		return response, fmt.Errorf("%s: %w", UnmarshalBodyErrorMessage, err)
 	}
 
 	return response, nil