Add Para threads to MMR Root

.github/workflows/release-relayer.yml

@@ -4,7 +4,6 @@ on:
   push:
     branches:
       - main
-      - release-v1.0.0
   workflow_dispatch:
 
 env:

relayer/chain/relaychain/connection.go

@@ -6,6 +6,7 @@ package relaychain
 import (
 	"context"
 	"fmt"
+	"sort"
 
 	gsrpc "github.com/snowfork/go-substrate-rpc-client/v4"
 	"github.com/snowfork/go-substrate-rpc-client/v4/types"
@@ -130,37 +131,6 @@ type ParaHead struct {
 	Data   types.Bytes
 }
 
-// Fetches heads for each parachain Id filtering out para threads.
-func (conn *Connection) FetchParachainHeads(relayChainBlockHash types.Hash) ([]ParaHead, error) {
-	// Fetch para heads
-	paraHeads, err := conn.fetchParaHeads(relayChainBlockHash)
-	if err != nil {
-		log.WithError(err).Error("Cannot fetch para heads.")
-		return nil, err
-	}
-
-	// fetch ids of parachains (not including parathreads)
-	var parachainIDs []uint32
-	parachainsKey, err := types.CreateStorageKey(conn.Metadata(), "Paras", "Parachains", nil, nil)
-	if err != nil {
-		return nil, err
-	}
-
-	_, err = conn.API().RPC.State.GetStorage(parachainsKey, &parachainIDs, relayChainBlockHash)
-	if err != nil {
-		return nil, err
-	}
-
-	// filter out parathreads
-	var parachainHeads []ParaHead
-	for _, v := range parachainIDs {
-		if head, ok := paraHeads[v]; ok {
-			parachainHeads = append(parachainHeads, head)
-		}
-	}
-	return parachainHeads, nil
-}
-
 func (co *Connection) FetchParachainHead(relayBlockhash types.Hash, paraID uint32, header *types.Header) (bool, error) {
 	encodedParaID, err := types.EncodeToBytes(paraID)
 	if err != nil {
@@ -272,7 +242,8 @@ func (co *Connection) fetchKeys(keyPrefix []byte, blockHash types.Hash) ([]types
 //	Key: hash_twox_128("Paras") + hash_twox_128("Heads") + hash_twox_64(ParaId) + Encode(ParaId)
 const ParaIDOffset = 16 + 16 + 8
 
-func (co *Connection) fetchParaHeads(blockHash types.Hash) (map[uint32]ParaHead, error) {
+// Fetch heads for all Paras. Included are parachains and parathreads.
+func (co *Connection) FetchParasHeads(blockHash types.Hash) ([]ParaHead, error) {
 	keyPrefix := types.CreateStorageKeyPrefix("Paras", "Heads")
 	keys, err := co.fetchKeys(keyPrefix, blockHash)
 	if err != nil {
@@ -292,7 +263,7 @@ func (co *Connection) fetchParaHeads(blockHash types.Hash) (map[uint32]ParaHead,
 		return nil, err
 	}
 
-	heads := make(map[uint32]ParaHead)
+	heads := make([]ParaHead, 0, 32)
 	for _, changeSet := range changeSets {
 		for _, change := range changeSet.Changes {
 			if change.StorageData.IsNone() {
@@ -313,12 +284,47 @@ func (co *Connection) fetchParaHeads(blockHash types.Hash) (map[uint32]ParaHead,
 				return nil, err
 			}
 
-			heads[paraID] = ParaHead{
+			heads = append(heads, ParaHead{
 				ParaID: paraID,
 				Data:   headData,
-			}
+			})
 		}
 	}
 
+	sort.SliceStable(heads, func(i int, j int) bool {
+		return heads[i].ParaID < heads[j].ParaID
+	})
+
+	return heads, nil
+}
+
+// Filters para heads to parachains only.
+func (conn *Connection) FilterParachainHeads(paraHeads []ParaHead, relayChainBlockHash types.Hash) ([]ParaHead, error) {
+
+	// fetch ids of parachains (not including parathreads)
+	var parachainIDs []uint32
+	parachainsKey, err := types.CreateStorageKey(conn.Metadata(), "Paras", "Parachains", nil, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = conn.API().RPC.State.GetStorage(parachainsKey, &parachainIDs, relayChainBlockHash)
+	if err != nil {
+		return nil, err
+	}
+
+	// create a set of parachains
+	parachains := make(map[uint32]struct{}, len(paraHeads))
+	for _, parachain := range parachainIDs {
+		parachains[parachain] = struct{}{}
+	}
+
+	// filter to return parachains
+	heads := make([]ParaHead, 0, len(paraHeads))
+	for _, head := range paraHeads {
+		if _, ok := parachains[head.ParaID]; ok {
+			heads = append(heads, head)
+		}
+	}
 	return heads, nil
 }

relayer/cmd/generate_beacon_data.go

@@ -132,7 +132,7 @@ func generateBeaconCheckpoint(cmd *cobra.Command, _ []string) error {
 			return err
 		}
 
-		p := protocol.New(conf.Source.Beacon.Spec)
+		p := protocol.New(conf.Source.Beacon.Spec, conf.Sink.Parachain.HeaderRedundancy)
 		store := store.New(conf.Source.Beacon.DataStore.Location, conf.Source.Beacon.DataStore.MaxEntries, *p)
 		store.Connect()
 		defer store.Close()
@@ -193,7 +193,7 @@ func generateBeaconTestFixture(cmd *cobra.Command, _ []string) error {
 			return err
 		}
 
-		p := protocol.New(conf.Source.Beacon.Spec)
+		p := protocol.New(conf.Source.Beacon.Spec, conf.Sink.Parachain.HeaderRedundancy)
 
 		store := store.New(conf.Source.Beacon.DataStore.Location, conf.Source.Beacon.DataStore.MaxEntries, *p)
 		err = store.Connect()
@@ -504,7 +504,7 @@ func generateExecutionUpdate(cmd *cobra.Command, _ []string) error {
 		}
 		log.WithFields(log.Fields{"endpoint": conf.Source.Beacon.Endpoint}).Info("connecting to beacon API")
 
-		p := protocol.New(conf.Source.Beacon.Spec)
+		p := protocol.New(conf.Source.Beacon.Spec, conf.Sink.Parachain.HeaderRedundancy)
 
 		store := store.New(conf.Source.Beacon.DataStore.Location, conf.Source.Beacon.DataStore.MaxEntries, *p)
 		store.Connect()
@@ -695,7 +695,7 @@ func generateInboundFixture(cmd *cobra.Command, _ []string) error {
 			return err
 		}
 
-		p := protocol.New(beaconConf.Source.Beacon.Spec)
+		p := protocol.New(beaconConf.Source.Beacon.Spec, beaconConf.Sink.Parachain.HeaderRedundancy)
 
 		store := store.New(beaconConf.Source.Beacon.DataStore.Location, beaconConf.Source.Beacon.DataStore.MaxEntries, *p)
 		store.Connect()

relayer/cmd/import_beacon_state.go

@@ -71,7 +71,7 @@ func importBeaconState(cmd *cobra.Command, _ []string) error {
 		return fmt.Errorf("open finalized state file: %w", err)
 	}
 
-	p := protocol.New(conf.Source.Beacon.Spec)
+	p := protocol.New(conf.Source.Beacon.Spec, conf.Sink.Parachain.HeaderRedundancy)
 	store := store.New(conf.Source.Beacon.DataStore.Location, conf.Source.Beacon.DataStore.MaxEntries, *p)
 	beaconClient := api.NewBeaconClient(conf.Source.Beacon.Endpoint, conf.Source.Beacon.StateEndpoint)
 	syncer := syncer.New(beaconClient, &store, p)

relayer/cmd/import_execution_header.go

@@ -109,7 +109,7 @@ func importExecutionHeaderFn(cmd *cobra.Command, _ []string) error {
 
 		log.WithField("hash", beaconHeader).Info("will be syncing execution header for beacon hash")
 
-		p := protocol.New(conf.Source.Beacon.Spec)
+		p := protocol.New(conf.Source.Beacon.Spec, conf.Sink.Parachain.HeaderRedundancy)
 		store := store.New(conf.Source.Beacon.DataStore.Location, conf.Source.Beacon.DataStore.MaxEntries, *p)
 		store.Connect()
 		defer store.Close()

relayer/cmd/list_beacon_states.go

@@ -49,7 +49,7 @@ func listBeaconState(cmd *cobra.Command, _ []string) error {
 		return err
 	}
 
-	p := protocol.New(conf.Source.Beacon.Spec)
+	p := protocol.New(conf.Source.Beacon.Spec, conf.Sink.Parachain.HeaderRedundancy)
 	store := store.New(conf.Source.Beacon.DataStore.Location, conf.Source.Beacon.DataStore.MaxEntries, *p)
 
 	err = store.Connect()

relayer/cmd/parachain_head_proof.go

@@ -92,7 +92,7 @@ func ParachainHeadProofFn(cmd *cobra.Command, _ []string) error {
 		return err
 	}
 
-	paraHeadsAsSlice, err := conn.FetchParachainHeads(relayChainBlockHash)
+	paraHeadsAsSlice, err := conn.FetchParasHeads(relayChainBlockHash)
 	if err != nil {
 		log.WithError(err).Error("Cannot fetch parachain headers")
 		return err

relayer/cmd/store_beacon_state.go

@@ -49,7 +49,7 @@ func storeBeaconState(cmd *cobra.Command, _ []string) error {
 		return err
 	}
 
-	p := protocol.New(conf.Source.Beacon.Spec)
+	p := protocol.New(conf.Source.Beacon.Spec, conf.Sink.Parachain.HeaderRedundancy)
 	store := store.New(conf.Source.Beacon.DataStore.Location, conf.Source.Beacon.DataStore.MaxEntries, *p)
 	beaconClient := api.NewBeaconClient(conf.Source.Beacon.Endpoint, conf.Source.Beacon.StateEndpoint)
 	syncer := syncer.New(beaconClient, &store, p)

relayer/relays/beacon/config/config.go

@@ -3,7 +3,6 @@ package config
 import (
 	"errors"
 	"fmt"
-	"github.com/snowfork/snowbridge/relayer/config"
 )
 
 type Config struct {
@@ -35,8 +34,16 @@ type BeaconConfig struct {
 }
 
 type SinkConfig struct {
-	Parachain          config.ParachainConfig `mapstructure:"parachain"`
-	UpdateSlotInterval uint64                 `mapstructure:"updateSlotInterval"`
+	Parachain          ParachainConfig `mapstructure:"parachain"`
+	UpdateSlotInterval uint64          `mapstructure:"updateSlotInterval"`
+}
+
+type ParachainConfig struct {
+	Endpoint             string `mapstructure:"endpoint"`
+	MaxWatchedExtrinsics int64  `mapstructure:"maxWatchedExtrinsics"`
+	// The max number of header in the FinalizedBeaconStateBuffer on-chain.
+	// https://github.com/paritytech/polkadot-sdk/blob/master/bridges/snowbridge/pallets/ethereum-client/src/types.rs#L23
+	HeaderRedundancy uint64 `mapstructure:"headerRedundancy"`
 }
 
 func (c Config) Validate() error {
@@ -81,3 +88,16 @@ func (b BeaconConfig) Validate() error {
 	}
 	return nil
 }
+
+func (p ParachainConfig) Validate() error {
+	if p.Endpoint == "" {
+		return errors.New("[endpoint] is not set")
+	}
+	if p.MaxWatchedExtrinsics == 0 {
+		return errors.New("[maxWatchedExtrinsics] is not set")
+	}
+	if p.HeaderRedundancy == 0 {
+		return errors.New("[HeaderRedundancy] is not set")
+	}
+	return nil
+}

relayer/relays/beacon/header/header.go

@@ -523,32 +523,32 @@ func (h *Header) findLatestCheckPoint(slot uint64) (state.FinalizedHeader, error
 		return beaconState, fmt.Errorf("GetLastFinalizedStateIndex error: %w", err)
 	}
 	startIndex := uint64(lastIndex)
-	endIndex := uint64(0)
+	endIndex := startIndex + 1
 
 	syncCommitteePeriod := h.protocol.Settings.SlotsInEpoch * h.protocol.Settings.EpochsPerSyncCommitteePeriod
-	slotPeriodIndex := slot / syncCommitteePeriod
-
-	for index := startIndex; index >= endIndex; index-- {
+	totalStates := syncCommitteePeriod * h.protocol.HeaderRedundancy // Total size of the circular buffer,
+	// https://github.com/paritytech/polkadot-sdk/blob/master/bridges/snowbridge/pallets/ethereum-client/src/lib.rs#L75
+	for index := startIndex; index != endIndex; index = (index - 1 + totalStates) % totalStates {
 		beaconRoot, err := h.writer.GetFinalizedBeaconRootByIndex(uint32(index))
 		if err != nil {
 			return beaconState, fmt.Errorf("GetFinalizedBeaconRootByIndex %d, error: %w", index, err)
 		}
 		beaconState, err = h.writer.GetFinalizedHeaderStateByBlockRoot(beaconRoot)
 		if err != nil {
-			return beaconState, fmt.Errorf("GetFinalizedHeaderStateByBlockRoot %s, error: %w", beaconRoot.Hex(), err)
+			// As soon as it can't find a block root, it means the circular wrap around array is empty.
+			log.WithFields(log.Fields{"index": index, "blockRoot": beaconRoot.Hex()}).WithError(err).Info("searching for checkpoint on-chain failed")
+			break
 		}
-		statePeriodIndex := beaconState.BeaconSlot / syncCommitteePeriod
+
 		if beaconState.BeaconSlot < slot {
+			log.WithFields(log.Fields{"index": index, "blockRoot": beaconRoot.Hex()}).WithError(err).Debug("unable to find a relevant on-chain header")
 			break
 		}
 		// Found the beaconState
-		if beaconState.BeaconSlot > slot && beaconState.BeaconSlot < slot+syncCommitteePeriod && slotPeriodIndex == statePeriodIndex {
-			break
+		if beaconState.BeaconSlot > slot && beaconState.BeaconSlot < slot+syncCommitteePeriod {
+			return beaconState, nil
 		}
 	}
-	if beaconState.BeaconSlot > slot && beaconState.BeaconSlot < slot+syncCommitteePeriod {
-		return beaconState, nil
-	}
 
 	return beaconState, fmt.Errorf("no checkpoint on chain for slot %d", slot)
 }