refactor: move deposits-aggregation from exporter to statistics and run only once per day

Author: guybrush

backend/cmd/statistics/main.go

@@ -1,6 +1,7 @@
 package statistics
 
 import (
+	"database/sql"
 	"errors"
 	"flag"
 	"fmt"
@@ -33,6 +34,7 @@ type options struct {
 	statisticsValidatorToggle bool
 	statisticsChartToggle     bool
 	statisticsGraffitiToggle  bool
+	statisticsDepositsToggle  bool
 	resetStatus               bool
 }
 
@@ -46,6 +48,7 @@ func Run() {
 	fs.BoolVar(&opt.statisticsValidatorToggle, "validators.enabled", false, "Toggle exporting validator statistics")
 	fs.BoolVar(&opt.statisticsChartToggle, "charts.enabled", false, "Toggle exporting chart series")
 	fs.BoolVar(&opt.statisticsGraffitiToggle, "graffiti.enabled", false, "Toggle exporting graffiti statistics")
+	fs.BoolVar(&opt.statisticsDepositsToggle, "deposits.enabled", false, "Toggle aggregating deposits")
 	fs.BoolVar(&opt.resetStatus, "validators.reset", false, "Export stats independent if they have already been exported previously")
 
 	versionFlag := fs.Bool("version", false, "Show version and exit")
@@ -349,6 +352,26 @@ func statisticsLoop(client rpc.Client) {
 			}
 		}
 
+		if opt.statisticsDepositsToggle {
+			lastDepositsAggregationTime := sql.NullTime{}
+			err := db.WriterDb.Get(&lastDepositsAggregationTime, "select time from eth1_deposits_aggregated_status order by time desc limit 1")
+			if err != nil && err != sql.ErrNoRows {
+				log.Error(err, fmt.Errorf("error aggregating deposits: getting lastDepositsAggregationTime:  %w", err), 0)
+			}
+			if !lastDepositsAggregationTime.Valid || time.Since(lastDepositsAggregationTime.Time) > time.Hour {
+				start := time.Now()
+				err := db.AggregateDeposits()
+				if err != nil {
+					log.Error(err, fmt.Errorf("error aggregating deposits: %w", err), 0)
+					loopError = err
+				}
+				log.InfoWithFields(log.Fields{
+					"lastDepositsAggregationTime": lastDepositsAggregationTime.Time,
+					"duration":                    time.Since(start),
+				}, "aggregated deposits")
+			}
+		}
+
 		if loopError == nil {
 			services.ReportStatus("statistics", "Running", nil)
 		} else {

backend/pkg/commons/db/migrations/postgres/20250107131603_eth1_deposits_aggregated_status.sql

@@ -0,0 +1,17 @@
+-- +goose Up
+-- +goose StatementBegin
+
+CREATE TABLE IF NOT EXISTS eth1_deposits_aggregated_status (
+	id SERIAL NOT NULL UNIQUE,
+	time TIMESTAMP WITHOUT TIME ZONE NOT NULL,
+	PRIMARY KEY (id)
+);
+
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+
+DROP TABLE IF EXISTS eth1_deposits_aggregated_status;
+
+-- +goose StatementEnd

backend/pkg/commons/db/statistics.go

@@ -1843,3 +1843,51 @@ func CheckIfDayIsFinalized(day uint64) error {
 
 	return nil
 }
+
+func AggregateDeposits() error {
+	start := time.Now()
+	defer func() {
+		metrics.TaskDuration.WithLabelValues("statistics_aggregate_eth1_deposits").Observe(time.Since(start).Seconds())
+	}()
+	_, err := WriterDb.Exec(`
+		INSERT INTO eth1_deposits_aggregated (from_address, amount, validcount, invalidcount, slashedcount, totalcount, activecount, pendingcount, voluntary_exit_count)
+		SELECT
+			eth1.from_address,
+			SUM(eth1.amount) as amount,
+			SUM(eth1.validcount) AS validcount,
+			SUM(eth1.invalidcount) AS invalidcount,
+			COUNT(CASE WHEN v.status = 'slashed' THEN 1 END) AS slashedcount,
+			COUNT(v.pubkey) AS totalcount,
+			COUNT(CASE WHEN v.status = 'active_online' OR v.status = 'active_offline' THEN 1 END) as activecount,
+			COUNT(CASE WHEN v.status = 'deposited' THEN 1 END) AS pendingcount,
+			COUNT(CASE WHEN v.status = 'exited' THEN 1 END) AS voluntary_exit_count
+		FROM (
+			SELECT
+				from_address,
+				publickey,
+				SUM(amount) AS amount,
+				COUNT(CASE WHEN valid_signature = 't' THEN 1 END) AS validcount,
+				COUNT(CASE WHEN valid_signature = 'f' THEN 1 END) AS invalidcount
+			FROM eth1_deposits
+			GROUP BY from_address, publickey
+		) eth1
+		LEFT JOIN (SELECT pubkey, status FROM validators) v ON v.pubkey = eth1.publickey
+		GROUP BY eth1.from_address
+		ON CONFLICT (from_address) DO UPDATE SET
+			amount               = excluded.amount,
+			validcount           = excluded.validcount,
+			invalidcount         = excluded.invalidcount,
+			slashedcount         = excluded.slashedcount,
+			totalcount           = excluded.totalcount,
+			activecount          = excluded.activecount,
+			pendingcount         = excluded.pendingcount,
+			voluntary_exit_count = excluded.voluntary_exit_count`)
+	if err != nil && err != sql.ErrNoRows {
+		return err
+	}
+	_, err = WriterDb.Exec(`INSERT INTO eth1_deposits_aggregated_status (id, time) VALUES (1, NOW())`)
+	if err != nil {
+		return err
+	}
+	return nil
+}

backend/pkg/exporter/modules/execution_deposits_exporter.go

@@ -26,7 +26,6 @@ import (
 	"github.com/gobitfly/beaconchain/pkg/commons/contracts/deposit_contract"
 	"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/rpc"
 	"github.com/gobitfly/beaconchain/pkg/commons/services"
 	"github.com/gobitfly/beaconchain/pkg/commons/types"
@@ -282,13 +281,6 @@ func (d *executionDepositsExporter) export() (err error) {
 
 	log.Debugf("updating cached deposits view took %v", time.Since(start))
 
-	if len(depositsToSave) > 0 {
-		err = d.aggregateDeposits()
-		if err != nil {
-			return err
-		}
-	}
-
 	return nil
 }
 
@@ -676,51 +668,6 @@ func (d *executionDepositsExporter) getDepositTraces(txsToTrace []string) (filte
 	return filteredTraces, nil
 }
 
-func (d *executionDepositsExporter) aggregateDeposits() error {
-	/// this could be a materialized view
-	start := time.Now()
-	defer func() {
-		metrics.TaskDuration.WithLabelValues("exporter_aggregate_eth1_deposits").Observe(time.Since(start).Seconds())
-	}()
-	_, err := db.WriterDb.Exec(`
-		INSERT INTO eth1_deposits_aggregated (from_address, amount, validcount, invalidcount, slashedcount, totalcount, activecount, pendingcount, voluntary_exit_count)
-		SELECT
-			eth1.from_address,
-			SUM(eth1.amount) as amount,
-			SUM(eth1.validcount) AS validcount,
-			SUM(eth1.invalidcount) AS invalidcount,
-			COUNT(CASE WHEN v.status = 'slashed' THEN 1 END) AS slashedcount,
-			COUNT(v.pubkey) AS totalcount,
-			COUNT(CASE WHEN v.status = 'active_online' OR v.status = 'active_offline' THEN 1 END) as activecount,
-			COUNT(CASE WHEN v.status = 'deposited' THEN 1 END) AS pendingcount,
-			COUNT(CASE WHEN v.status = 'exited' THEN 1 END) AS voluntary_exit_count
-		FROM (
-			SELECT
-				from_address,
-				publickey,
-				SUM(amount) AS amount,
-				COUNT(CASE WHEN valid_signature = 't' THEN 1 END) AS validcount,
-				COUNT(CASE WHEN valid_signature = 'f' THEN 1 END) AS invalidcount
-			FROM eth1_deposits
-			GROUP BY from_address, publickey
-		) eth1
-		LEFT JOIN (SELECT pubkey, status FROM validators) v ON v.pubkey = eth1.publickey
-		GROUP BY eth1.from_address
-		ON CONFLICT (from_address) DO UPDATE SET
-			amount               = excluded.amount,
-			validcount           = excluded.validcount,
-			invalidcount         = excluded.invalidcount,
-			slashedcount         = excluded.slashedcount,
-			totalcount           = excluded.totalcount,
-			activecount          = excluded.activecount,
-			pendingcount         = excluded.pendingcount,
-			voluntary_exit_count = excluded.voluntary_exit_count`)
-	if err != nil && err != sql.ErrNoRows {
-		return nil
-	}
-	return err
-}
-
 func (d *executionDepositsExporter) updateCachedView() error {
 	err := db.CacheQuery(`
 		SELECT