feat(exporter): add functionality for a no downtime migration of legacy balances from clickhouse to bigtable

misc(exporter): please linter, fix typo

misc(exporter): add clickhouse table migration file, switch from epoch to timestamp to enable future garbage collection

misc(exporter): add retries when migrating legacy balances

Author: peterbitfly

.gitignore

@@ -0,0 +1 @@
+.idea/

backend/cmd/misc/commands/migrate_legacy_balances.go

@@ -0,0 +1,89 @@
+package commands
+
+import (
+	"flag"
+	"fmt"
+	"time"
+
+	"github.com/gobitfly/beaconchain/cmd/misc/misctypes"
+	"github.com/gobitfly/beaconchain/pkg/commons/db"
+	"github.com/gobitfly/beaconchain/pkg/commons/log"
+	"github.com/gobitfly/beaconchain/pkg/commons/rpc"
+	"github.com/gobitfly/beaconchain/pkg/commons/types"
+)
+
+type MigrateLegacyBalancesCommand struct {
+	FlagSet *flag.FlagSet
+	Config  migrateLegacyBalancesCommandConfig
+}
+
+type migrateLegacyBalancesCommandConfig struct {
+	StartEpoch uint64
+	EndEpoch   uint64
+}
+
+func (s *MigrateLegacyBalancesCommand) ParseCommandOptions() {
+	s.FlagSet.Uint64Var(&s.Config.StartEpoch, "legacy-balances-start-epoch", 0, "Start epoch for the balance migration")
+	s.FlagSet.Uint64Var(&s.Config.EndEpoch, "legacy-balances-end-epoch", 0, "End epoch for the balance migration")
+}
+
+func (s *MigrateLegacyBalancesCommand) Requires() misctypes.Requires {
+	return misctypes.Requires{
+		ClickhouseDBs: true,
+		ClNode:        true,
+	}
+}
+
+func (s *MigrateLegacyBalancesCommand) Run(clClient *rpc.LighthouseClient) error {
+	if s.Config.EndEpoch == 0 {
+		s.showHelp()
+		return fmt.Errorf("end-epoch is required")
+	}
+
+	log.Infof("command: migrate-legacy-balances start-epoch=%d end-epoch=%d", s.Config.StartEpoch, s.Config.EndEpoch)
+
+	for epoch := s.Config.StartEpoch; epoch <= s.Config.EndEpoch; epoch++ {
+		for attempt := 1; attempt <= 5; attempt++ {
+			state, err := clClient.GetValidatorState(epoch)
+			if err != nil {
+				log.Error(err, fmt.Sprintf("error getting balances for epoch %d (attempt %d/5): %v", epoch, attempt, err), 0)
+				if attempt == 5 {
+					return fmt.Errorf("failed to get balances for epoch %d after %d attempts: %w", epoch, attempt, err)
+				}
+				time.Sleep(5 * time.Second)
+				continue
+			}
+
+			validators := make([]*types.Validator, 0, len(state.Data))
+			for _, validator := range state.Data {
+				validators = append(validators, &types.Validator{
+					Index:            validator.Index,
+					Balance:          validator.Balance,
+					EffectiveBalance: validator.Validator.EffectiveBalance,
+				})
+			}
+
+			log.Infof("writing epoch %d to clickhouse with %d validators", epoch, len(state.Data))
+			err = db.SaveLegacyValidatorBalancesToClickhouse(epoch, validators)
+			if err != nil {
+				log.Error(err, fmt.Sprintf("error saving legacy balances for epoch %d (attempt %d/5): %v", epoch, attempt, err), 0)
+				if attempt == 5 {
+					return fmt.Errorf("error saving legacy balances for epoch %d after %d attempts: %w", epoch, attempt, err)
+				}
+				time.Sleep(5 * time.Second)
+				continue
+			}
+
+			// success for this epoch
+			break
+		}
+	}
+	return nil
+}
+
+func (s *MigrateLegacyBalancesCommand) showHelp() {
+	log.Infof("Usage: migrate_legacy_balances [options]")
+	log.Infof("Options:")
+	log.Infof("  --legacy-balances-start-epoch int\tStart epoch for the balance migration")
+	log.Infof("  --legacy-balances-end-epoch int\tEnd epoch for the balance migration")
+}

backend/cmd/misc/main.go

@@ -87,7 +87,8 @@ var REQUIRES_LIST = map[string]misctypes.Requires{
 	"clear-raw-bigtable": {
 		RawBigtable: true,
 	},
-	"app-bundle": (&commands.AppBundleCommand{}).Requires(),
+	"app-bundle":              (&commands.AppBundleCommand{}).Requires(),
+	"migrate-legacy-balances": (&commands.MigrateLegacyBalancesCommand{}).Requires(),
 	"update-highest-active-validatorindex": {
 		Bigtable: true,
 		ClNode:   true,
@@ -105,8 +106,12 @@ func Run() {
 		FlagSet: fs,
 	}
 
+	migrateLegacyBalancesCommand := commands.MigrateLegacyBalancesCommand{
+		FlagSet: fs,
+	}
+
 	configPath := fs.String("config", "config/default.config.yml", "Path to the config file")
-	fs.StringVar(&opts.Command, "command", "", "command to run, available: updateAPIKey, applyDbSchema, initBigtableSchema, epoch-export, debug-rewards, debug-blocks, clear-bigtable, clear-raw-bigtable, index-old-eth1-blocks, update-aggregation-bits, historic-prices-export, index-missing-blocks, export-epoch-missed-slots, migrate-last-attestation-slot-bigtable, export-genesis-validators, update-block-finalization-sequentially, nameValidatorsByRanges, export-stats-totals, export-sync-committee-periods, export-sync-committee-validator-stats, partition-validator-stats, migrate-app-purchases, collect-notifications, collect-user-db-notifications, verify-fcm-tokens, app-bundle, update-highest-active-validatorindex")
+	fs.StringVar(&opts.Command, "command", "", "command to run, available: updateAPIKey, applyDbSchema, initBigtableSchema, epoch-export, debug-rewards, debug-blocks, clear-bigtable, clear-raw-bigtable, index-old-eth1-blocks, update-aggregation-bits, historic-prices-export, index-missing-blocks, export-epoch-missed-slots, migrate-last-attestation-slot-bigtable, export-genesis-validators, update-block-finalization-sequentially, nameValidatorsByRanges, export-stats-totals, export-sync-committee-periods, export-sync-committee-validator-stats, partition-validator-stats, migrate-app-purchases, collect-notifications, collect-user-db-notifications, verify-fcm-tokens, app-bundle, update-highest-active-validatorindex, migrate-legacy-balances")
 	fs.Uint64Var(&opts.StartEpoch, "start-epoch", 0, "start epoch")
 	fs.Uint64Var(&opts.EndEpoch, "end-epoch", 0, "end epoch")
 	fs.Uint64Var(&opts.User, "user", 0, "user id")
@@ -133,6 +138,8 @@ func Run() {
 
 	statsPartitionCommand.ParseCommandOptions()
 	appBundleCommand.ParseCommandOptions()
+	migrateLegacyBalancesCommand.ParseCommandOptions()
+
 	_ = fs.Parse(os.Args[2:])
 
 	if *versionFlag {
@@ -221,6 +228,9 @@ func Run() {
 		db.ClickHouseWriter, db.ClickHouseReader = db.MustInitDB(&cfg.ClickHouse.WriterDatabase, &cfg.ClickHouse.ReaderDatabase, "clickhouse", "clickhouse")
 		defer db.ClickHouseReader.Close()
 		defer db.ClickHouseWriter.Close()
+
+		db.ClickHouseNativeWriter = db.MustInitClickhouseNative(&cfg.ClickHouse.WriterDatabase)
+		defer db.ClickHouseNativeWriter.Close()
 	}
 
 	// Initialize the persistent redis client
@@ -487,6 +497,8 @@ func Run() {
 	case "app-bundle":
 		appBundleCommand.Config.DryRun = opts.DryRun
 		err = appBundleCommand.Run()
+	case "migrate-legacy-balances":
+		err = migrateLegacyBalancesCommand.Run(rpcClient)
 	case "fix-ens":
 		err = fixEns(erigonClient)
 	case "fix-ens-addresses":

backend/pkg/commons/db/clickhouse_legacy_balances.go

@@ -0,0 +1,54 @@
+package db
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/gobitfly/beaconchain/pkg/commons/types"
+	"github.com/gobitfly/beaconchain/pkg/commons/utils"
+	"github.com/google/uuid"
+)
+
+type LegacyValidatorBalanceTable struct {
+	ValidatorIndex   []uint32     `db:"validator_index"`
+	EpochTimestamp   []*time.Time `db:"epoch_timestamp"`
+	Balance          []uint64     `db:"balance"`
+	EffectiveBalance []uint64     `db:"effective_balance"`
+}
+
+func (c LegacyValidatorBalanceTable) Get(str string) any {
+	switch str {
+	case "validator_index":
+		return c.ValidatorIndex
+	case "epoch_timestamp":
+		return c.EpochTimestamp
+	case "balance":
+		return c.Balance
+	case "effective_balance":
+		return c.EffectiveBalance
+	default:
+		return nil
+	}
+}
+func (c LegacyValidatorBalanceTable) Extend(cOther UltraFastClickhouseStruct) error {
+	return nil
+}
+
+func SaveLegacyValidatorBalancesToClickhouse(epoch uint64, state []*types.Validator) error {
+	epochTs := utils.EpochToTime(epoch)
+
+	epochData := LegacyValidatorBalanceTable{}
+	for _, validator := range state {
+		epochData.ValidatorIndex = append(epochData.ValidatorIndex, uint32(validator.Index))
+		epochData.EpochTimestamp = append(epochData.EpochTimestamp, &epochTs)
+		epochData.Balance = append(epochData.Balance, validator.Balance)
+		epochData.EffectiveBalance = append(epochData.EffectiveBalance, validator.EffectiveBalance)
+	}
+
+	err := UltraFastDumpToClickhouse(epochData, "legacy_validator_epoch_balances", uuid.New().String())
+	if err != nil {
+		return fmt.Errorf("error writing legacy validator epoch balances to clickhouse: %w", err)
+	}
+
+	return nil
+}

backend/pkg/commons/db/migrations/clickhouse/20251007053053_add_legacy_balances_table.sql

@@ -0,0 +1,16 @@
+-- +goose Up
+-- +goose StatementBegin
+CREATE TABLE IF NOT EXISTS legacy_validator_epoch_balances (
+   `validator_index` UInt32,
+   `epoch_timestamp` DateTime,
+   `balance` UInt64,
+   `effective_balance` UInt64
+)
+ENGINE = ReplacingMergeTree
+ORDER BY (validator_index, epoch_timestamp);
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+DROP TABLE IF EXISTS legacy_validator_epoch_balances;
+-- +goose StatementEnd

backend/pkg/exporter/modules/slot_exporter.go

@@ -502,6 +502,15 @@ func ExportSlot(client rpc.Client, slot uint64, isHeadEpoch bool, tx *sqlx.Tx) e
 			return nil
 		})
 
+		// save the validator balances to clickhouse
+		g.Go(func() error {
+			err := db.SaveLegacyValidatorBalancesToClickhouse(epoch, block.Validators)
+			if err != nil {
+				return fmt.Errorf("error exporting validator balances to clickhouse for slot %v: %w", block.Slot, err)
+			}
+			return nil
+		})
+
 		// if we are exporting the head epoch, update the validator db table
 		if isHeadEpoch {
 			// this function sets exports the validator status into the db