Sync beefy commitment on demand (#1342)

relayer/cmd/root.go

@@ -36,7 +36,6 @@ func init() {
 	rootCmd.AddCommand(storeBeaconStateCmd())
 	rootCmd.AddCommand(importBeaconStateCmd())
 	rootCmd.AddCommand(listBeaconStateCmd())
-	rootCmd.AddCommand(syncBeefyCommitmentCmd())
 }
 
 func Execute() {

relayer/cmd/run/beefy/command.go

@@ -21,6 +21,7 @@ var (
 	privateKey     string
 	privateKeyFile string
 	privateKeyID   string
+	onDemand       bool
 )
 
 func Command() *cobra.Command {
@@ -38,6 +39,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().BoolVarP(&onDemand, "on-demand", "", false, "Synchronize commitments on demand")
+
 	return cmd
 }
 
@@ -66,11 +69,6 @@ func run(_ *cobra.Command, _ []string) error {
 		return err
 	}
 
-	relay, err := beefy.NewRelay(&config, keypair)
-	if err != nil {
-		return err
-	}
-
 	ctx, cancel := context.WithCancel(context.Background())
 	eg, ctx := errgroup.WithContext(ctx)
 
@@ -90,11 +88,30 @@ func run(_ *cobra.Command, _ []string) error {
 		return nil
 	})
 
-	err = relay.Start(ctx, eg)
-	if err != nil {
-		logrus.WithError(err).Fatal("Unhandled error")
-		cancel()
-		return err
+	if !onDemand {
+		relay, err := beefy.NewRelay(&config, keypair)
+		if err != nil {
+			return err
+		}
+
+		err = relay.Start(ctx, eg)
+		if err != nil {
+			logrus.WithError(err).Fatal("Unhandled error")
+			cancel()
+			return err
+		}
+	} else {
+		relay, err := beefy.NewOnDemandRelay(&config, keypair)
+		if err != nil {
+			return err
+		}
+
+		err = relay.Start(ctx)
+		if err != nil {
+			logrus.WithError(err).Fatal("Unhandled error")
+			cancel()
+			return err
+		}
 	}
 
 	err = eg.Wait()

relayer/config/config.go

@@ -7,8 +7,7 @@ type PolkadotConfig struct {
 }
 
 type ParachainConfig struct {
-	Endpoint             string `mapstructure:"endpoint"`
-	MaxWatchedExtrinsics int64  `mapstructure:"maxWatchedExtrinsics"`
+	Endpoint string `mapstructure:"endpoint"`
 }
 
 type EthereumConfig struct {
@@ -27,9 +26,6 @@ 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")
-	}
 	return nil
 }
 

relayer/relays/beefy/config.go

@@ -7,12 +7,14 @@ import (
 )
 
 type Config struct {
-	Source SourceConfig `mapstructure:"source"`
-	Sink   SinkConfig   `mapstructure:"sink"`
+	Source       SourceConfig       `mapstructure:"source"`
+	Sink         SinkConfig         `mapstructure:"sink"`
+	OnDemandSync OnDemandSyncConfig `mapstructure:"on-demand-sync"`
 }
 
 type SourceConfig struct {
-	Polkadot config.PolkadotConfig `mapstructure:"polkadot"`
+	Polkadot  config.PolkadotConfig  `mapstructure:"polkadot"`
+	BridgeHub config.ParachainConfig `mapstructure:"bridge-hub"`
 }
 
 type SinkConfig struct {
@@ -23,6 +25,18 @@ type SinkConfig struct {
 
 type ContractsConfig struct {
 	BeefyClient string `mapstructure:"BeefyClient"`
+	Gateway     string `mapstructure:"Gateway"`
+}
+
+type OnDemandSyncConfig struct {
+	// ID of the AssetHub channel
+	AssetHubChannelID string `mapstructure:"asset-hub-channel-id"`
+	// Maximum number of tokens available to consume
+	MaxTokens uint64 `mapstructure:"max-tokens"`
+	// Number of tokens added each `RefillPeriod`
+	RefillAmount uint64 `mapstructure:"refill-amount"`
+	// Period between token refills
+	RefillPeriod uint64 `mapstructure:"refill-period"`
 }
 
 func (c Config) Validate() error {
@@ -40,5 +54,20 @@ func (c Config) Validate() error {
 	if c.Sink.Contracts.BeefyClient == "" {
 		return fmt.Errorf("sink contracts setting [BeefyClient] is not set")
 	}
+	if c.Sink.Contracts.Gateway == "" {
+		return fmt.Errorf("sink contracts setting [Gateway] is not set")
+	}
+	if c.OnDemandSync.AssetHubChannelID == "" {
+		return fmt.Errorf("`on-demand-sync.asset-hub-channel-id` not set")
+	}
+	if c.OnDemandSync.MaxTokens == 0 {
+		return fmt.Errorf("`on-demand-sync.max-tokens` not set")
+	}
+	if c.OnDemandSync.RefillAmount == 0 {
+		return fmt.Errorf("`on-demand-sync.refill-amount` not set")
+	}
+	if c.OnDemandSync.RefillPeriod == 0 {
+		return fmt.Errorf("`on-demand-sync.refill-period` not set")
+	}
 	return nil
 }

relayer/relays/beefy/main.go

@@ -80,83 +80,3 @@ func (relay *Relay) Start(ctx context.Context, eg *errgroup.Group) error {
 
 	return nil
 }
-
-func (relay *Relay) OneShotSync(ctx context.Context, blockNumber uint64) error {
-	// Initialize relaychainConn
-	err := relay.relaychainConn.Connect(ctx)
-	if err != nil {
-		return fmt.Errorf("create relaychain connection: %w", err)
-	}
-
-	// Initialize ethereumConn
-	err = relay.ethereumConn.Connect(ctx)
-	if err != nil {
-		return fmt.Errorf("create ethereum connection: %w", err)
-	}
-	err = relay.ethereumWriter.initialize(ctx)
-	if err != nil {
-		return fmt.Errorf("initialize EthereumWriter: %w", err)
-	}
-
-	state, err := relay.ethereumWriter.queryBeefyClientState(ctx)
-	if err != nil {
-		return fmt.Errorf("query beefy client state: %w", err)
-	}
-	// Ignore relay block already synced
-	if blockNumber <= state.LatestBeefyBlock {
-		log.WithFields(log.Fields{
-			"validatorSetID": state.CurrentValidatorSetID,
-			"beefyBlock":     state.LatestBeefyBlock,
-			"relayBlock":     blockNumber,
-		}).Info("Relay block already synced, just ignore")
-		return nil
-	}
-
-	// generate beefy update for that specific relay block
-	task, err := relay.polkadotListener.generateBeefyUpdate(blockNumber)
-	if err != nil {
-		return fmt.Errorf("fail to generate next beefy request: %w", err)
-	}
-
-	// Ignore commitment earlier than LatestBeefyBlock which is outdated
-	if task.SignedCommitment.Commitment.BlockNumber <= uint32(state.LatestBeefyBlock) {
-		log.WithFields(log.Fields{
-			"latestBeefyBlock":      state.LatestBeefyBlock,
-			"currentValidatorSetID": state.CurrentValidatorSetID,
-			"nextValidatorSetID":    state.NextValidatorSetID,
-			"blockNumberToSync":     task.SignedCommitment.Commitment.BlockNumber,
-		}).Info("Commitment outdated, just ignore")
-		return nil
-	}
-	if task.SignedCommitment.Commitment.ValidatorSetID > state.NextValidatorSetID {
-		log.WithFields(log.Fields{
-			"latestBeefyBlock":      state.LatestBeefyBlock,
-			"currentValidatorSetID": state.CurrentValidatorSetID,
-			"nextValidatorSetID":    state.NextValidatorSetID,
-			"validatorSetIDToSync":  task.SignedCommitment.Commitment.ValidatorSetID,
-		}).Warn("Task unexpected, wait for mandatory updates to catch up first")
-		return nil
-	}
-
-	// Submit the task
-	if task.SignedCommitment.Commitment.ValidatorSetID == state.CurrentValidatorSetID {
-		task.ValidatorsRoot = state.CurrentValidatorSetRoot
-	} else {
-		task.ValidatorsRoot = state.NextValidatorSetRoot
-	}
-	err = relay.ethereumWriter.submit(ctx, task)
-	if err != nil {
-		return fmt.Errorf("fail to submit beefy update: %w", err)
-	}
-
-	updatedState, err := relay.ethereumWriter.queryBeefyClientState(ctx)
-	if err != nil {
-		return fmt.Errorf("query beefy client state: %w", err)
-	}
-	log.WithFields(log.Fields{
-		"latestBeefyBlock":      updatedState.LatestBeefyBlock,
-		"currentValidatorSetID": updatedState.CurrentValidatorSetID,
-		"nextValidatorSetID":    updatedState.NextValidatorSetID,
-	}).Info("Sync beefy update success")
-	return nil
-}