Parachain Relayer V2

relayer/chain/parachain/writer.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/snowfork/go-substrate-rpc-client/v4/rpc/author"
 	"github.com/snowfork/go-substrate-rpc-client/v4/types"
@@ -48,6 +49,10 @@ func NewParachainWriter(
 }
 
 func (wr *ParachainWriter) Start(ctx context.Context, eg *errgroup.Group) error {
+	err := wr.conn.ConnectWithHeartBeat(ctx, 30*time.Second)
+	if err != nil {
+		return err
+	}
 	nonce, err := wr.queryAccountNonce()
 	if err != nil {
 		return err

relayer/cmd/generate_beacon_data.go

@@ -556,7 +556,7 @@ func getEthereumEvent(ctx context.Context, gatewayContract *contracts.Gateway, c
 	for event == nil {
 		log.Info("looking for Ethereum event")
 
-		iter, err := gatewayContract.FilterOutboundMessageAccepted(&opts, [][32]byte{channelID}, [][32]byte{})
+		iter, err := gatewayContract.FilterOutboundMessageAccepted(&opts)
 		if err != nil {
 			return nil, err
 		}

relayer/cmd/run/parachain/command.go

@@ -14,17 +14,19 @@ import (
 	"github.com/mitchellh/mapstructure"
 	"github.com/sirupsen/logrus"
 	"github.com/snowfork/snowbridge/relayer/chain/ethereum"
+	para "github.com/snowfork/snowbridge/relayer/chain/parachain"
 	"github.com/snowfork/snowbridge/relayer/relays/parachain"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 	"golang.org/x/sync/errgroup"
 )
 
 var (
-	configFile     string
-	privateKey     string
-	privateKeyFile string
-	privateKeyID   string
+	configFile          string
+	privateKey          string
+	privateKeyFile      string
+	privateKeyID        string
+	parachainPrivateKey string
 )
 
 func Command() *cobra.Command {
@@ -42,6 +44,8 @@ func Command() *cobra.Command {
 	cmd.Flags().StringVar(&privateKeyFile, "ethereum.private-key-file", "", "The file from which to read the private key")
 	cmd.Flags().StringVar(&privateKeyID, "ethereum.private-key-id", "", "The secret id to lookup the private key in AWS Secrets Manager")
 
+	cmd.Flags().StringVar(&parachainPrivateKey, "substrate.private-key", "", "substrate private key")
+
 	return cmd
 }
 
@@ -70,7 +74,12 @@ func run(_ *cobra.Command, _ []string) error {
 		return err
 	}
 
-	relay, err := parachain.NewRelay(&config, keypair)
+	keypair2, err := para.ResolvePrivateKey(parachainPrivateKey, "", "")
+	if err != nil {
+		return err
+	}
+
+	relay, err := parachain.NewRelay(&config, keypair, keypair2)
 	if err != nil {
 		return err
 	}

relayer/generate.go

@@ -1,4 +1,4 @@
 //go:generate bash -c "jq .abi ../contracts/out/BeefyClient.sol/BeefyClient.json | abigen --abi - --type BeefyClient --pkg contracts --out contracts/beefy_client.go"
-//go:generate bash -c "jq .abi ../contracts/out/IGateway.sol/IGateway.json | abigen --abi - --type Gateway --pkg contracts --out contracts/gateway.go"
+//go:generate bash -c "jq .abi ../contracts/out/IGateway.sol/IGatewayV2.json | abigen --abi - --type Gateway --pkg contracts --out contracts/gateway.go"
 
 package main

relayer/relays/execution/main.go

@@ -222,16 +222,9 @@ func (r *Relay) fetchLatestParachainNonce() (uint64, error) {
 	return paraNonce, nil
 }
 
+// Todo: nonce deprecated
 func (r *Relay) fetchEthereumNonce(ctx context.Context) (uint64, error) {
-	opts := bind.CallOpts{
-		Context: ctx,
-	}
-	_, ethOutboundNonce, err := r.gatewayContract.ChannelNoncesOf(&opts, r.config.Source.ChannelID)
-	if err != nil {
-		return 0, fmt.Errorf("fetch Gateway.ChannelNoncesOf(%v): %w", r.config.Source.ChannelID, err)
-	}
-
-	return ethOutboundNonce, nil
+	return 0, nil
 }
 
 const BlocksPerQuery = 4096
@@ -286,7 +279,7 @@ func (r *Relay) findEvents(
 }
 
 func (r *Relay) findEventsWithFilter(opts *bind.FilterOpts, channelID [32]byte, start uint64) (bool, []*contracts.GatewayOutboundMessageAccepted, error) {
-	iter, err := r.gatewayContract.FilterOutboundMessageAccepted(opts, [][32]byte{channelID}, [][32]byte{})
+	iter, err := r.gatewayContract.FilterOutboundMessageAccepted(opts)
 	if err != nil {
 		return false, nil, err
 	}
@@ -409,7 +402,6 @@ func (r *Relay) doSubmit(ctx context.Context, ev *contracts.GatewayOutboundMessa
 		"blockNumber": ev.Raw.BlockNumber,
 		"txHash":      ev.Raw.TxHash.Hex(),
 		"txIndex":     ev.Raw.TxIndex,
-		"channelID":   types.H256(ev.ChannelID).Hex(),
 	})
 
 	nextBlockNumber := new(big.Int).SetUint64(ev.Raw.BlockNumber + 1)

relayer/relays/parachain/beefy-listener.go

@@ -160,7 +160,7 @@ func (li *BeefyListener) doScan(ctx context.Context, beefyBlockNumber uint64) er
 	}
 	for _, task := range tasks {
 		paraNonce := (*task.MessageProofs)[0].Message.Nonce
-		waitingPeriod := (paraNonce + li.scheduleConfig.TotalRelayerCount - li.scheduleConfig.ID) % li.scheduleConfig.TotalRelayerCount
+		waitingPeriod := (uint64(paraNonce) + li.scheduleConfig.TotalRelayerCount - li.scheduleConfig.ID) % li.scheduleConfig.TotalRelayerCount
 		err = li.waitAndSend(ctx, task, waitingPeriod)
 		if err != nil {
 			return fmt.Errorf("wait task for nonce %d: %w", paraNonce, err)
@@ -326,11 +326,11 @@ func (li *BeefyListener) waitAndSend(ctx context.Context, task *Task, waitingPer
 	var cnt uint64
 	var err error
 	for {
-		ethInboundNonce, err := li.scanner.findLatestNonce(ctx)
+		isRelayed, err := li.scanner.isNonceRelayed(ctx, uint64(paraNonce))
 		if err != nil {
 			return err
 		}
-		if ethInboundNonce >= paraNonce {
+		if isRelayed {
 			log.Info(fmt.Sprintf("nonce %d picked up by another relayer, just skip", paraNonce))
 			return nil
 		}

relayer/relays/parachain/config.go

@@ -5,20 +5,22 @@ import (
 	"fmt"
 
 	"github.com/snowfork/snowbridge/relayer/config"
+	beaconconf "github.com/snowfork/snowbridge/relayer/relays/beacon/config"
 )
 
 type Config struct {
-	Source   SourceConfig   `mapstructure:"source"`
-	Sink     SinkConfig     `mapstructure:"sink"`
-	Schedule ScheduleConfig `mapstructure:"schedule"`
+	Source        SourceConfig   `mapstructure:"source"`
+	Sink          SinkConfig     `mapstructure:"sink"`
+	Schedule      ScheduleConfig `mapstructure:"schedule"`
+	RewardAddress string         `mapstructure:"reward-address"`
 }
 
 type SourceConfig struct {
-	Polkadot  config.PolkadotConfig  `mapstructure:"polkadot"`
-	Parachain config.ParachainConfig `mapstructure:"parachain"`
-	Ethereum  config.EthereumConfig  `mapstructure:"ethereum"`
-	Contracts SourceContractsConfig  `mapstructure:"contracts"`
-	ChannelID ChannelID              `mapstructure:"channel-id"`
+	Polkadot  config.PolkadotConfig   `mapstructure:"polkadot"`
+	Parachain config.ParachainConfig  `mapstructure:"parachain"`
+	Ethereum  config.EthereumConfig   `mapstructure:"ethereum"`
+	Contracts SourceContractsConfig   `mapstructure:"contracts"`
+	Beacon    beaconconf.BeaconConfig `mapstructure:"beacon"`
 }
 
 type SourceContractsConfig struct {
@@ -76,9 +78,6 @@ func (c Config) Validate() error {
 	if c.Source.Contracts.Gateway == "" {
 		return fmt.Errorf("source contracts setting [Gateway] is not set")
 	}
-	if c.Source.ChannelID == [32]byte{} {
-		return fmt.Errorf("source setting [channel-id] is not set")
-	}
 
 	// Sink
 	err = c.Sink.Ethereum.Validate()
@@ -95,5 +94,9 @@ func (c Config) Validate() error {
 		return fmt.Errorf("relay config: %w", err)
 	}
 
+	if c.RewardAddress == "" {
+		return fmt.Errorf("reward address is not set")
+	}
+
 	return nil
 }

relayer/relays/parachain/digest_item.go

@@ -8,8 +8,8 @@ func ExtractCommitmentFromDigest(digest types.Digest) (*types.H256, error) {
 	for _, digestItem := range digest {
 		if digestItem.IsOther {
 			digestItemRawBytes := digestItem.AsOther
-			// Prefix 0 reserved for snowbridge
-			if digestItemRawBytes[0] == 0 {
+			// Prefix 0 reserved for snowbridge V2
+			if digestItemRawBytes[0] == 1 {
 				var commitment types.H256
 				err := types.DecodeFromBytes(digestItemRawBytes[1:], &commitment)
 				if err != nil {

relayer/relays/parachain/ethereum-writer.go

@@ -17,30 +17,34 @@ import (
 	"github.com/snowfork/snowbridge/relayer/chain/ethereum"
 	"github.com/snowfork/snowbridge/relayer/contracts"
 	"github.com/snowfork/snowbridge/relayer/crypto/keccak"
+	"github.com/snowfork/snowbridge/relayer/relays/util"
 
 	gsrpcTypes "github.com/snowfork/go-substrate-rpc-client/v4/types"
 
 	log "github.com/sirupsen/logrus"
 )
 
 type EthereumWriter struct {
-	config     *SinkConfig
-	conn       *ethereum.Connection
-	gateway    *contracts.Gateway
-	tasks      <-chan *Task
-	gatewayABI abi.ABI
+	config      *SinkConfig
+	conn        *ethereum.Connection
+	gateway     *contracts.Gateway
+	tasks       <-chan *Task
+	gatewayABI  abi.ABI
+	relayConfig *Config
 }
 
 func NewEthereumWriter(
 	config *SinkConfig,
 	conn *ethereum.Connection,
 	tasks <-chan *Task,
+	relayConfig *Config,
 ) (*EthereumWriter, error) {
 	return &EthereumWriter{
-		config:  config,
-		conn:    conn,
-		gateway: nil,
-		tasks:   tasks,
+		config:      config,
+		conn:        conn,
+		gateway:     nil,
+		tasks:       tasks,
+		relayConfig: relayConfig,
 	}, nil
 }
 
@@ -143,8 +147,13 @@ func (wr *EthereumWriter) WriteChannel(
 		LeafProofOrder: new(big.Int).SetUint64(proof.MMRProof.MerkleProofOrder),
 	}
 
-	tx, err := wr.gateway.SubmitV1(
-		options, message, commitmentProof.Proof.InnerHashes, verificationProof,
+	rewardAddress, err := util.HexStringTo32Bytes(wr.relayConfig.RewardAddress)
+	if err != nil {
+		return fmt.Errorf("convert to reward address: %w", err)
+	}
+
+	tx, err := wr.gateway.V2Submit(
+		options, message, commitmentProof.Proof.InnerHashes, verificationProof, rewardAddress,
 	)
 	if err != nil {
 		return fmt.Errorf("send transaction Gateway.submit: %w", err)
@@ -183,9 +192,8 @@ func (wr *EthereumWriter) WriteChannel(
 				return fmt.Errorf("unpack event log: %w", err)
 			}
 			log.WithFields(log.Fields{
-				"channelID": Hex(holder.ChannelID[:]),
-				"nonce":     holder.Nonce,
-				"success":   holder.Success,
+				"nonce":   holder.Nonce,
+				"success": holder.Success,
 			}).Info("Message dispatched")
 		}
 	}

relayer/relays/parachain/logger.go

@@ -43,10 +43,9 @@ func (wr *EthereumWriter) logFieldsForSubmission(
 
 	params := log.Fields{
 		"message": log.Fields{
-			"channelID": Hex(message.ChannelID[:]),
-			"nonce":     message.Nonce,
-			"command":   message.Command,
-			"params":    Hex(message.Params),
+			"nonce":    message.Nonce,
+			"commands": message.Commands,
+			"origin":   Hex(message.Origin[:]),
 		},
 		"messageProof": messageProofHexes,
 		"proof": log.Fields{

relayer/relays/parachain/main.go

@@ -11,6 +11,12 @@ import (
 	"github.com/snowfork/snowbridge/relayer/chain/parachain"
 	"github.com/snowfork/snowbridge/relayer/chain/relaychain"
 	"github.com/snowfork/snowbridge/relayer/crypto/secp256k1"
+	"github.com/snowfork/snowbridge/relayer/crypto/sr25519"
+
+	"github.com/snowfork/snowbridge/relayer/relays/beacon/header"
+	"github.com/snowfork/snowbridge/relayer/relays/beacon/header/syncer/api"
+	"github.com/snowfork/snowbridge/relayer/relays/beacon/protocol"
+	"github.com/snowfork/snowbridge/relayer/relays/beacon/store"
 
 	log "github.com/sirupsen/logrus"
 )
@@ -23,9 +29,12 @@ type Relay struct {
 	ethereumConnBeefy     *ethereum.Connection
 	ethereumChannelWriter *EthereumWriter
 	beefyListener         *BeefyListener
+	parachainWriter       *parachain.ParachainWriter
+	beaconHeader          *header.Header
+	headerCache           *ethereum.HeaderCache
 }
 
-func NewRelay(config *Config, keypair *secp256k1.Keypair) (*Relay, error) {
+func NewRelay(config *Config, keypair *secp256k1.Keypair, keypair2 *sr25519.Keypair) (*Relay, error) {
 	log.Info("Creating worker")
 
 	parachainConn := parachain.NewConnection(config.Source.Parachain.Endpoint, nil)
@@ -41,6 +50,7 @@ func NewRelay(config *Config, keypair *secp256k1.Keypair) (*Relay, error) {
 		&config.Sink,
 		ethereumConnWriter,
 		tasks,
+		config,
 	)
 	if err != nil {
 		return nil, err
@@ -55,6 +65,30 @@ func NewRelay(config *Config, keypair *secp256k1.Keypair) (*Relay, error) {
 		tasks,
 	)
 
+	parachainWriterConn := parachain.NewConnection(config.Source.Parachain.Endpoint, keypair2.AsKeyringPair())
+
+	parachainWriter := parachain.NewParachainWriter(
+		parachainWriterConn,
+		8,
+	)
+	headerCache, err := ethereum.NewHeaderBlockCache(
+		&ethereum.DefaultBlockLoader{Conn: ethereumConnWriter},
+	)
+	if err != nil {
+		return nil, err
+	}
+	p := protocol.New(config.Source.Beacon.Spec, 20)
+	store := store.New(config.Source.Beacon.DataStore.Location, config.Source.Beacon.DataStore.MaxEntries, *p)
+	store.Connect()
+	beaconAPI := api.NewBeaconClient(config.Source.Beacon.Endpoint, config.Source.Beacon.StateEndpoint)
+	beaconHeader := header.New(
+		parachainWriter,
+		beaconAPI,
+		config.Source.Beacon.Spec,
+		&store,
+		p,
+		0, // setting is not used in the execution relay
+	)
 	return &Relay{
 		config:                config,
 		parachainConn:         parachainConn,
@@ -63,6 +97,9 @@ func NewRelay(config *Config, keypair *secp256k1.Keypair) (*Relay, error) {
 		ethereumConnBeefy:     ethereumConnBeefy,
 		ethereumChannelWriter: ethereumChannelWriter,
 		beefyListener:         beefyListener,
+		parachainWriter:       parachainWriter,
+		beaconHeader:          &beaconHeader,
+		headerCache:           headerCache,
 	}, nil
 }
 
@@ -99,6 +136,16 @@ func (relay *Relay) Start(ctx context.Context, eg *errgroup.Group) error {
 		return err
 	}
 
+	err = relay.parachainWriter.Start(ctx, eg)
+	if err != nil {
+		return err
+	}
+
+	err = relay.startDeliverProof(ctx, eg)
+	if err != nil {
+		return err
+	}
+
 	log.Info("Current relay's ID:", relay.config.Schedule.ID)
 
 	return nil