feat(dashboard_exporter)!: backfill for electra fork epoch events

Author: invis-bitfly

backend/pkg/commons/db/migrations/clickhouse/20250520110413_add_backfill_support_for_electra_fork_epoch.sql

@@ -0,0 +1,101 @@
+-- +goose Up
+-- +goose StatementBegin
+ CREATE TABLE IF NOT EXISTS _insert_sink_backfill_electra_fork_epoch_events
+(
+    `epoch_timestamp` DateTime COMMENT 'timestamp of the epoch',
+    `validator_index` UInt64 COMMENT 'validator index',
+    `withdrawals_amount` Int64 COMMENT 'sum of withdrawals_amount',
+    `withdrawals_count` Int64 COMMENT 'sum of withdrawals_count',
+    -- we will need something to fix the roi as well
+    -- roi_dividend is `balance_end - deposits_amount + withdrawals_amount - consolidations_incoming_amount + consolidations_outgoing_amount`
+    -- so to fix it, subtract the withdrawals_amount from the roi_divisor
+)
+ENGINE = ReplacingMergeTree() -- we need it persistent because we will use it to update the epoch table through the dictionary
+ORDER BY (validator_index, epoch_timestamp)
+SETTINGS index_granularity=128, non_replicated_deduplication_window = 2048, replicated_deduplication_window = 2048;
+-- +goose StatementEnd
+-- +goose StatementBegin
+-- connect the sink to the _final_validator_dashboard_roi_hourly table
+CREATE MATERIALIZED VIEW IF NOT EXISTS _mv_backfill_final_validator_dashboard_roi_hourly_electra_fork TO _final_validator_dashboard_roi_hourly
+(
+    `validator_index` UInt64,
+    `t` DateTime,
+    `roi_dividend` Int128,
+    `roi_divisor` Int128
+)
+AS SELECT
+    validator_index AS validator_index,
+    toStartOfHour(foo.epoch_timestamp) AS t,
+    SUM(CAST(foo.withdrawals_amount, 'Int128')) AS roi_dividend, -- i.e, we add the withdrawal amount to the roi_dividend
+    0 AS roi_divisor -- i.e, we dont modify the divisor
+FROM _insert_sink_backfill_electra_fork_epoch_events AS foo
+GROUP BY
+    t,
+    validator_index
+-- once it is in the hourly, it will be automatically propagated to the daily, weekly and monthly tables by the existing materialized views
+-- +goose StatementEnd
+-- +goose StatementBegin
+-- connect the sink to the _final_validator_dashboard_data_hourly table
+CREATE MATERIALIZED VIEW IF NOT EXISTS _mv_backfill_final_validator_dashboard_data_hourly_electra_fork TO _final_validator_dashboard_data_hourly
+(
+    `validator_index` UInt64,
+    `t` DateTime,
+    `withdrawals_amount` Int64,
+    `withdrawals_count` Int64
+) 
+AS SELECT
+    validator_index AS validator_index,
+    toStartOfHour(foo.epoch_timestamp) AS t,
+    SUM(foo.withdrawals_amount) AS withdrawals_amount, -- i.e we add the withdrawal amount to the withdrawal amount
+    SUM(foo.withdrawals_count) AS withdrawals_count -- i.e we add the withdrawal count to the withdrawal count
+FROM _insert_sink_backfill_electra_fork_epoch_events AS foo
+GROUP BY
+    t,
+    validator_index
+-- +goose StatementEnd
+-- +goose StatementBegin
+SELECT 'the migration will now attempt to create a dictionary. this will only work if the migration is run as the default (root) user.';
+-- +goose StatementEnd
+-- +goose StatementBegin
+CREATE DICTIONARY IF NOT EXISTS _dict_backfill_electra_fork_epoch_events (validator_index Int64, epoch_timestamp DateTime, withdrawals_amount Int64, withdrawals_count Int64)
+PRIMARY KEY epoch_timestamp, validator_index SOURCE(CLICKHOUSE(TABLE _insert_sink_backfill_electra_fork_epoch_events DB currentDatabase()))
+LAYOUT(complex_key_direct()); -- we dont use complex_key_cache here because that would require us to make sure that the cache is up to date when we trigger the mutation. it should only be at most 200k keys anyways
+-- +goose StatementEnd
+-- +goose StatementBegin
+-- kickstart the backfill with the latest exported epoch
+INSERT INTO _exporter_backfill_metadata 
+    SELECT
+        epoch,
+        'electra_fork_epoch_events' as backfill_name,
+        transfer_batch_id as backfill_batch_id,
+        NULL AS successful_backfill
+    FROM
+    (
+        SELECT *
+        FROM _exporter_metadata
+        WHERE successful_transfer IS NOT NULL
+        ORDER BY epoch DESC
+        LIMIT 1
+    )
+-- +goose StatementEnd
+-- +goose StatementBegin
+SELECT 'successfully created the dictionary. now start the dashboard exporter again. once the backfill is done (i.e it shows up as 0 epochs left in the grafana dashboard), run the next migration version to update the epoch table.';
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+DROP VIEW IF EXISTS _mv_backfill_final_validator_dashboard_roi_hourly_electra_fork;
+-- +goose StatementEnd
+-- +goose StatementBegin
+DROP VIEW IF EXISTS _mv_backfill_final_validator_dashboard_data_hourly_electra_fork;
+-- +goose StatementEnd
+-- +goose StatementBegin
+ALTER TABLE _exporter_backfill_metadata 
+    DELETE WHERE backfill_name = 'electra_fork_epoch_events';
+-- +goose StatementEnd
+-- +goose StatementBegin
+DROP DICTIONARY IF EXISTS _dict_backfill_electra_fork_epoch_events;
+-- +goose StatementEnd
+-- +goose StatementBegin
+DROP TABLE IF EXISTS _insert_sink_backfill_electra_fork_epoch_events;
+-- +goose StatementEnd

backend/pkg/exporter/db/db.go

@@ -1791,7 +1791,10 @@ func GetPendingBackfillEpochs(t BackfillType, limit int64) ([]BackfillMetadata,
 		return nil, fmt.Errorf("error fetching min assigned epoch: %w", err)
 	}
 	if minAssignedEpoch == -1 {
-		return nil, fmt.Errorf("no epochs found with transfer batch id")
+		// this isn't necessarily an error, because on a fresh export this backfill never gets assigned
+		// so log a debug message and return nil
+		log.Debugf("no epochs found with transfer batch id %s. this is expected if the export was started after the backfill was implemented", t)
+		return nil, nil
 	}
 	if minAssignedEpoch == 0 {
 		// this means we have assigned epochs till the first one, no more pending epochs
@@ -1983,6 +1986,7 @@ func (b BackfillType) Value() (driver.Value, error) {
 }
 
 const (
-	BackfillTypeRoi      BackfillType = "roi"
-	BackfillTypeEBLookup BackfillType = "eb_lookup"
+	BackfillTypeRoi                    BackfillType = "roi"
+	BackfillTypeEBLookup               BackfillType = "eb_lookup"
+	BackfillTypeElectraForkEpochEvents BackfillType = "electra_fork_epoch_events"
 )

backend/pkg/exporter/modules/dashboard_data_backfills.go

@@ -4,11 +4,12 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/gobitfly/beaconchain/pkg/commons/db"
 	"github.com/gobitfly/beaconchain/pkg/commons/log"
 	"github.com/gobitfly/beaconchain/pkg/commons/metrics"
 	"github.com/gobitfly/beaconchain/pkg/commons/utils"
-	"github.com/gobitfly/beaconchain/pkg/exporter/db"
 	edb "github.com/gobitfly/beaconchain/pkg/exporter/db"
+	"github.com/gobitfly/beaconchain/pkg/exporter/types"
 	"github.com/google/uuid"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
@@ -20,6 +21,7 @@ func (d *dashboardData) roiBackfillTask() {
 	jobs := []edb.BackfillType{
 		edb.BackfillTypeRoi,
 		edb.BackfillTypeEBLookup,
+		edb.BackfillTypeElectraForkEpochEvents,
 	}
 	for _, backfillType := range jobs {
 		// fork for every backfill we have to do
@@ -52,7 +54,7 @@ func (d *dashboardData) roiBackfillTask() {
 		}()
 	}
 }
-func (d *dashboardData) handleIncompleteBackfills(t db.BackfillType) error {
+func (d *dashboardData) handleIncompleteBackfills(t edb.BackfillType) error {
 	incomplete, err := edb.GetIncompleteBackfillEpochs(t)
 	if err != nil {
 		return errors.Wrap(err, "failed to get incomplete backfill epochs")
@@ -70,7 +72,7 @@ func (d *dashboardData) handleIncompleteBackfills(t db.BackfillType) error {
 	return nil
 }
 
-func (d *dashboardData) backfillEpochs(t db.BackfillType, epochs []edb.BackfillMetadata) error {
+func (d *dashboardData) backfillEpochs(t edb.BackfillType, epochs []edb.BackfillMetadata) error {
 	now := time.Now()
 	defer func() {
 		metrics.TaskDuration.WithLabelValues(fmt.Sprintf("dashboard_data_exporter_backfill_%s_overall", t)).Observe(time.Since(now).Seconds())
@@ -97,6 +99,8 @@ func (d *dashboardData) backfillEpochs(t db.BackfillType, epochs []edb.BackfillM
 				err = edb.BackfillRoi(epochs)
 			case edb.BackfillTypeEBLookup:
 				err = edb.BackfillEBLookup(epochs)
+			case edb.BackfillTypeElectraForkEpochEvents:
+				err = d.BackfillElectraForkEpochEvents(epochs)
 			default:
 				return fmt.Errorf("unknown backfill type %s", t)
 			}
@@ -108,6 +112,12 @@ func (d *dashboardData) backfillEpochs(t db.BackfillType, epochs []edb.BackfillM
 
 			now := time.Now()
 			for i := range epochs {
+				if epochs[i].BackfillName != t {
+					return fmt.Errorf("backfill name %s does not match backfill type %s", epochs[i].BackfillName, t)
+				}
+				if epochs[i].BackfillBatchId == nil {
+					return fmt.Errorf("backfill batch id is nil for epoch %v", epochs[i])
+				}
 				epochs[i].SuccessfulBackfill = &now
 			}
 			err = edb.PushBackfillMetadata(epochs)
@@ -125,13 +135,13 @@ func (d *dashboardData) backfillEpochs(t db.BackfillType, epochs []edb.BackfillM
 	return nil
 }
 
-func (d *dashboardData) handlePendingBackfills(t db.BackfillType) error {
+func (d *dashboardData) handlePendingBackfills(t edb.BackfillType) error {
 	pending, err := edb.GetPendingBackfillEpochs(t, utils.Config.DashboardExporter.BackfillAtOnce*utils.Config.DashboardExporter.BackfillInParallel)
 	if err != nil {
 		return errors.Wrap(err, "failed to get pending backfill epochs")
 	}
 	if len(pending) == 0 {
-		d.log.Debugf("handlePendingBackfills, no pending backfill epochs")
+		d.log.Debugf("handlePendingBackfills, no pending backfill epochs") // this log messages, and basically all others, should include the backfill type
 		return nil
 	}
 	d.log.InfoWithFields(log.Fields{"backfillType": t, "pendingCount": len(pending)}, "handlePendingBackfills, found pending backfill epochs")
@@ -157,3 +167,77 @@ func (d *dashboardData) handlePendingBackfills(t db.BackfillType) error {
 	}
 	return nil
 }
+
+func (d *dashboardData) BackfillElectraForkEpochEvents(epochs []edb.BackfillMetadata) (err error) {
+	metricPrefix := string("dashboard_data_exporter_backfill_" + edb.BackfillTypeElectraForkEpochEvents)
+	start := time.Now()
+	defer func() {
+		metrics.TaskDuration.WithLabelValues(metricPrefix + "_batch").Observe(time.Since(start).Seconds())
+	}()
+	// filter out epoch that arent the fork epoch
+	forkEpoch := utils.Config.Chain.ClConfig.ElectraForkEpoch
+	var forkEpochs []edb.BackfillMetadata
+	for _, e := range epochs {
+		if e.Epoch == forkEpoch {
+			forkEpochs = append(forkEpochs, e)
+		}
+	}
+	if len(forkEpochs) == 0 {
+		log.Debugf("no epochs to backfill for %s", edb.BackfillTypeElectraForkEpochEvents)
+		return nil
+	}
+	if len(forkEpochs) > 1 {
+		return fmt.Errorf("more than one epoch to backfill for %s", edb.BackfillTypeElectraForkEpochEvents)
+	}
+	backfillBatchID := epochs[0].BackfillBatchId
+	epoch := forkEpochs[0].Epoch
+
+	// reuse the existing stuff to do this, tho obv could be leaner
+	nodeData := NewMultiEpochData(1)
+	// prefill epochBasedData.epochs
+	nodeData.epochBasedData.epochs = []uint64{forkEpochs[0].Epoch}
+	eg := &errgroup.Group{}
+	eg.Go(func() error {
+		return d.fetchEpochValidatorStates(epoch, epoch, &nodeData)
+	})
+	eg.Go(func() error {
+		return d.fetchElectraRemovedExcessBalances(epoch, epoch, &nodeData)
+	})
+	if err := eg.Wait(); err != nil {
+		return fmt.Errorf("failed to fetch validator states for epoch %d: %w", epoch, err)
+	}
+
+	// prepare the target data. i hate this as much as you do, but i dont have the time to refactor this rn
+	// was never really intended to be used like this anyway
+	processedData := make([]types.VDBDataEpochColumns, 1)
+	nodeData.epochBasedData.tarIndices = []int{0}
+	nodeData.epochBasedData.tarOffsets = []int{0}
+	valiCount := len(nodeData.epochBasedData.validatorStates[int64(epoch)].Data)
+	processedData[0], err = types.NewVDBDataEpochColumns(valiCount)
+	if err != nil {
+		return fmt.Errorf("failed to create new VDBDataEpochColumns: %w", err)
+	}
+
+	processedData[0].EpochsContained = []uint64{epoch}
+	eg = &errgroup.Group{} // docs say we should not reuse a group for different tasks
+	eg.Go(func() error {
+		// sets epochTimestamp and validatorIndex
+		return d.processValidatorStates(&nodeData, &processedData)
+	})
+	eg.Go(func() error {
+		// sets withdrawalAmount and withdrawalCount
+		return d.processElectraRemovedExcessBalanceEvents(&nodeData, &processedData)
+	})
+	if err := eg.Wait(); err != nil {
+		return fmt.Errorf("failed to process electra removed excess balance events: %w", err)
+	}
+
+	// now write it to the insert sink
+	err = db.UltraFastDumpToClickhouse(&processedData[0], "_insert_sink_backfill_electra_fork_epoch_events", backfillBatchID.String())
+	if err != nil {
+		d.log.Error(err, "failed to insert epochs", 0, log.Fields{"epochs": forkEpochs})
+		return errors.Wrap(err, "failed to insert epochs")
+	}
+
+	return nil
+}