Staking clients

README.md

@@ -120,6 +120,8 @@ The configuration is read from `toml` file.
 The settings for `[db]`, `[logger]` are the same as for the indexer above.
 Specific settings are listed below.
 
+**Note:** We recommend that the user accessing the database is not the same as for the indexer. The user for the services should only have read permissions enabled!
+
 Config file can be specified using the command line parameter `--config`, e.g., `./services --config config.local.toml`. The default config file name is `config.toml`.
 
 ```toml

config.toml

@@ -52,22 +52,18 @@ period = "90s"
 uptime_threshold = 0.8
 
 [epochs]
-# start of the voting/mirroring epoch
-start = "2021-08-01T00:00:00Z"
-period = "180s"
 # first epoch to be voted for (do not leav this empty since the first epoch is well back in time
 first = 1111
 
 [voting_cronjob]
 enabled = false
 timeout = "10s"
-# voting contract address
-contract_address = "0x0000000"
 
 [mirroring_cronjob]
 enabled = false
 timeout = "10s"
-# mirroring contract address
-contract_address = "0x0000000"
-# address binder contract address
-address_binder_contract_address = "0x0000000"
+
+[contract_addresses]
+address_binder = "0x0000000"
+mirroring = "0x0000000"
+voting = "0x0000000"

config/config.go

@@ -1,15 +1,13 @@
 package config
 
 import (
-	"flag"
-	"flare-indexer/utils"
 	"fmt"
 	"log"
 	"os"
 	"strings"
-	"time"
 
 	"github.com/BurntSushi/toml"
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/kelseyhightower/envconfig"
 )
 
@@ -70,9 +68,11 @@ func (cfg ChainConfig) GetPrivateKey() (string, error) {
 }
 
 type EpochConfig struct {
-	Period time.Duration   `toml:"period" envconfig:"EPOCH_PERIOD"`
-	Start  utils.Timestamp `toml:"start" envconfig:"EPOCH_TIME"`
-	First  int64           `toml:"first" envconfig:"EPOCH_FIRST"`
+	First int64 `toml:"first" envconfig:"EPOCH_FIRST"`
+}
+
+type ContractAddresses struct {
+	Voting common.Address `toml:"voting" envconfig:"VOTING_CONTRACT_ADDRESS"`
 }
 
 func ParseConfigFile(cfg interface{}, fileName string, allowMissing bool) error {
@@ -99,9 +99,3 @@ func ReadEnv(cfg interface{}) error {
 	}
 	return nil
 }
-
-func ConfigFileName() string {
-	cfgFlag := flag.String("config", CONFIG_FILE, "Configuration file (toml format)")
-	flag.Parse()
-	return *cfgFlag
-}

database/queries.go

@@ -65,3 +65,7 @@ func PersistUptimeAggregations(db *gorm.DB, aggregations []*UptimeAggregation) e
 	}
 	return db.Create(aggregations).Error
 }
+
+func DeleteUptimesBefore(db *gorm.DB, timestamp time.Time) error {
+	return db.Where("timestamp < ?", timestamp).Delete(&UptimeCronjob{}).Error
+}

indexer/config/config.go

@@ -8,16 +8,17 @@ import (
 )
 
 type Config struct {
-	DB            config.DBConfig     `toml:"db"`
-	Logger        config.LoggerConfig `toml:"logger"`
-	Chain         config.ChainConfig  `toml:"chain"`
-	Metrics       MetricsConfig       `toml:"metrics"`
-	XChainIndexer IndexerConfig       `toml:"x_chain_indexer"`
-	PChainIndexer IndexerConfig       `toml:"p_chain_indexer"`
-	UptimeCronjob UptimeConfig        `toml:"uptime_cronjob"`
-	Mirror        MirrorConfig        `toml:"mirroring_cronjob"`
-	VotingCronjob VotingConfig        `toml:"voting_cronjob"`
-	Epochs        config.EpochConfig  `toml:"epochs"`
+	DB                config.DBConfig     `toml:"db"`
+	Logger            config.LoggerConfig `toml:"logger"`
+	Chain             config.ChainConfig  `toml:"chain"`
+	Metrics           MetricsConfig       `toml:"metrics"`
+	XChainIndexer     IndexerConfig       `toml:"x_chain_indexer"`
+	PChainIndexer     IndexerConfig       `toml:"p_chain_indexer"`
+	UptimeCronjob     UptimeConfig        `toml:"uptime_cronjob"`
+	Mirror            MirrorConfig        `toml:"mirroring_cronjob"`
+	VotingCronjob     VotingConfig        `toml:"voting_cronjob"`
+	Epochs            config.EpochConfig  `toml:"epochs"`
+	ContractAddresses ContractAddresses   `toml:"contract_addresses"`
 }
 
 type MetricsConfig struct {
@@ -40,20 +41,24 @@ type CronjobConfig struct {
 
 type MirrorConfig struct {
 	CronjobConfig
-	MirroringContract     common.Address `toml:"contract_address" envconfig:"MIRRORING_CONTRACT_ADDRESS"`
-	AddressBinderContract common.Address `toml:"address_binder_contract_address" envconfig:"ADDRESS_BINDER_CONTRACT_ADDRESS"`
 }
 
 type VotingConfig struct {
 	CronjobConfig
-	ContractAddress common.Address `toml:"contract_address" envconfig:"VOTING_CONTRACT_ADDRESS"`
 }
 
 type UptimeConfig struct {
 	CronjobConfig
 	config.EpochConfig
-	EnableVoting    bool    `toml:"enable_voting"`
-	UptimeThreshold float64 `toml:"uptime_threshold"`
+	EnableVoting                   bool    `toml:"enable_voting"`
+	UptimeThreshold                float64 `toml:"uptime_threshold"`
+	DeleteOldUptimesEpochThreshold int64   `toml:"delete_old_uptimes_epoch_threshold"`
+}
+
+type ContractAddresses struct {
+	config.ContractAddresses
+	AddressBinder common.Address `toml:"address_binder" envconfig:"ADDRESS_BINDER_CONTRACT_ADDRESS"`
+	Mirroring     common.Address `toml:"mirroring" envconfig:"MIRRORING_CONTRACT_ADDRESS"`
 }
 
 func newConfig() *Config {
@@ -79,9 +84,6 @@ func newConfig() *Config {
 		Chain: config.ChainConfig{
 			NodeURL: "http://localhost:9650/",
 		},
-		Epochs: config.EpochConfig{
-			Period: 90 * time.Second,
-		},
 	}
 }
 
@@ -93,8 +95,7 @@ func (c Config) ChainConfig() config.ChainConfig {
 	return c.Chain
 }
 
-func BuildConfig() (*Config, error) {
-	cfgFileName := config.ConfigFileName()
+func BuildConfig(cfgFileName string) (*Config, error) {
 	cfg := newConfig()
 	err := config.ParseConfigFile(cfg, cfgFileName, false)
 	if err != nil {

indexer/context/context.go

@@ -1,6 +1,7 @@
 package context
 
 import (
+	"flag"
 	globalConfig "flare-indexer/config"
 	"flare-indexer/database"
 	"flare-indexer/indexer/config"
@@ -11,30 +12,62 @@ import (
 type IndexerContext interface {
 	Config() *config.Config
 	DB() *gorm.DB
+	Flags() *IndexerFlags
+}
+
+type IndexerFlags struct {
+	ConfigFileName string
+
+	// Set start epoch for voting cronjob to this value, overrides config and database value,
+	// valid value is > 0
+	ResetVotingCronjob int64
+
+	// Set start epoch for mirroring cronjob to this value, overrides config and database value,
+	// valid value is > 0
+	ResetMirrorCronjob int64
 }
 
 type indexerContext struct {
 	config *config.Config
 	db     *gorm.DB
+	flags  *IndexerFlags
 }
 
 func BuildContext() (IndexerContext, error) {
-	ctx := indexerContext{}
-
-	cfg, err := config.BuildConfig()
+	flags := parseIndexerFlags()
+	cfg, err := config.BuildConfig(flags.ConfigFileName)
 	if err != nil {
 		return nil, err
 	}
-	ctx.config = cfg
 	globalConfig.GlobalConfigCallback.Call(cfg)
 
-	ctx.db, err = database.ConnectAndInitialize(&cfg.DB)
+	db, err := database.ConnectAndInitialize(&cfg.DB)
 	if err != nil {
 		return nil, err
 	}
-	return &ctx, nil
+
+	return &indexerContext{
+		config: cfg,
+		db:     db,
+		flags:  flags,
+	}, nil
 }
 
 func (c *indexerContext) Config() *config.Config { return c.config }
 
 func (c *indexerContext) DB() *gorm.DB { return c.db }
+
+func (c *indexerContext) Flags() *IndexerFlags { return c.flags }
+
+func parseIndexerFlags() *IndexerFlags {
+	cfgFlag := flag.String("config", globalConfig.CONFIG_FILE, "Configuration file (toml format)")
+	resetVotingFlag := flag.Int64("reset-voting", 0, "Set start epoch for voting cronjob to this value, overrides config and database value, valid values are > 0")
+	resetMirrorFlag := flag.Int64("reset-mirroring", 0, "Set start epoch for mirroring cronjob to this value, overrides config and database value, valid values are > 0")
+	flag.Parse()
+
+	return &IndexerFlags{
+		ConfigFileName:     *cfgFlag,
+		ResetVotingCronjob: *resetVotingFlag,
+		ResetMirrorCronjob: *resetMirrorFlag,
+	}
+}

indexer/cronjob/cronjob.go

@@ -1,7 +1,6 @@
 package cronjob
 
 import (
-	globalConfig "flare-indexer/config"
 	"flare-indexer/database"
 	"flare-indexer/indexer/config"
 	"flare-indexer/logger"
@@ -58,11 +57,11 @@ type epochRange struct {
 	end   int64
 }
 
-func newEpochCronjob(cronjobCfg *config.CronjobConfig, epochCfg *globalConfig.EpochConfig) epochCronjob {
+func newEpochCronjob(cronjobCfg *config.CronjobConfig, epochs staking.EpochInfo) epochCronjob {
 	return epochCronjob{
 		enabled:   cronjobCfg.Enabled,
 		timeout:   cronjobCfg.Timeout,
-		epochs:    staking.NewEpochInfo(epochCfg),
+		epochs:    epochs,
 		batchSize: cronjobCfg.BatchSize,
 		delay:     cronjobCfg.Delay,
 	}

indexer/cronjob/mirror.go

@@ -29,7 +29,7 @@ type mirrorCronJob struct {
 
 type mirrorDB interface {
 	FetchState(name string) (database.State, error)
-	UpdateJobState(epoch int64) error
+	UpdateJobState(epoch int64, force bool) error
 	GetPChainTxsForEpoch(start, end time.Time) ([]database.PChainTxData, error)
 	GetPChainTx(txID string, address string) (*database.PChainTxData, error)
 }
@@ -42,6 +42,7 @@ type mirrorContracts interface {
 	) error
 	IsAddressRegistered(address string) (bool, error)
 	RegisterPublicKey(publicKey crypto.PublicKey) error
+	EpochConfig() (time.Time, time.Duration, error)
 }
 
 func NewMirrorCronjob(ctx indexerctx.IndexerContext) (Cronjob, error) {
@@ -56,11 +57,22 @@ func NewMirrorCronjob(ctx indexerctx.IndexerContext) (Cronjob, error) {
 		return nil, err
 	}
 
-	return &mirrorCronJob{
-		epochCronjob: newEpochCronjob(&cfg.Mirror.CronjobConfig, &cfg.Epochs),
+	start, period, err := contracts.EpochConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	epochs := staking.NewEpochInfo(&cfg.Epochs, start, period)
+
+	mc := &mirrorCronJob{
+		epochCronjob: newEpochCronjob(&cfg.Mirror.CronjobConfig, epochs),
 		db:           NewMirrorDBGorm(ctx.DB()),
 		contracts:    contracts,
-	}, nil
+	}
+
+	err = mc.reset(ctx.Flags().ResetMirrorCronjob)
+
+	return mc, err
 }
 
 func (c *mirrorCronJob) Name() string {
@@ -108,7 +120,7 @@ func (c *mirrorCronJob) Call() error {
 
 	logger.Debug("successfully mirrored epochs %d-%d", epochRange.start, epochRange.end)
 
-	if err := c.db.UpdateJobState(epochRange.end); err != nil {
+	if err := c.db.UpdateJobState(epochRange.end+1, false); err != nil {
 		return err
 	}
 
@@ -323,3 +335,17 @@ func (c *mirrorCronJob) registerAddress(txID string, address string) error {
 	}
 	return nil
 }
+
+func (c *mirrorCronJob) reset(firstEpoch int64) error {
+	if firstEpoch <= 0 {
+		return nil
+	}
+
+	logger.Info("Resetting mirroring cronjob state to epoch %d", firstEpoch)
+	err := c.db.UpdateJobState(firstEpoch, true)
+	if err != nil {
+		return err
+	}
+	c.epochs.First = firstEpoch
+	return nil
+}