fix(dashboard_exporter): use new indexed events table instead of raw events

Author: invis-bitfly

backend/pkg/consapi/types/debug.go

@@ -56,12 +56,12 @@ type ElectraDeposit struct {
 }
 
 type ElectraConsolidation struct {
-	SourceValidatorIndex uint64 `json:"source_index,string" db:"source_index"`
-	TargetValidatorIndex uint64 `json:"target_index,string" db:"target_index"`
-	Amount               uint64 `db:"amount"`
+	SourcePubkey []byte `db:"source_pubkey"`
+	TargetPubkey []byte `db:"target_pubkey"`
+	Amount       uint64 `db:"amount"`
 }
 
 type ElectraExcessBalance struct {
-	ValidatorIndex uint64 `json:"validator_index,string" db:"validator_index"`
-	Amount         uint64 `json:"amount,string" db:"amount"`
+	ValidatorPubkey []byte `db:"validator_pubkey"`
+	Amount          uint64 `json:"amount,string" db:"amount"`
 }

backend/pkg/exporter/db/db.go

@@ -1859,11 +1859,10 @@ func ElectraGetEpochProcessedHashes(epoch uint64) ([][]byte, error) {
 		`, (utils.Config.ClConfig.SlotsPerEpoch*epoch)-1))
 	*/
 	// use goqu
-	q := goqu.Dialect("postgres").Select("block_root").
-		From("consensus_layer_events").
+	q := goqu.Dialect("postgres").Select("transition_block_hash").
+		From("consensus_layer_events_indexer_metadata").
 		Where(
-			goqu.I("event_name").Eq("EpochProcessedEvent"),
-			goqu.I("slot").Eq((utils.Config.ClConfig.SlotsPerEpoch*epoch)-1),
+			goqu.I("epoch").Eq(epoch - 1), // it is what it is
 		)
 	sql, args, err := q.Prepared(true).ToSQL()
 	if err != nil {
@@ -1881,18 +1880,17 @@ func ElectraGetProcessedDeposits(epoch uint64) ([]constypes.ElectraDeposit, erro
 	endSlot := (epoch+1)*utils.Config.ClConfig.SlotsPerEpoch - 1
 	var result []constypes.ElectraDeposit
 	q := goqu.Dialect("postgres").Select(
-		goqu.L("data->>'amount'").As("amount"),
-		goqu.L("data->>'pubkey'").As("pubkey"),
-		// SignatureValid field, assume true if not present
-		goqu.COALESCE(goqu.L("data->>'signature_valid'"), goqu.L("'true'")).As("signature_valid"),
+		goqu.L("amount").As("amount"),
+		goqu.L("pubkey").As("pubkey"),
+		goqu.L("true").As("signature_valid"),
 	).
-		From("consensus_layer_events").
+		From("blocks_deposit_requests_v2").
 		Where(
-			goqu.I("event_name").Eq("DepositProcessedEvent"),
-			goqu.I("slot").Gte(startSlot),
-			goqu.I("slot").Lt(endSlot),
+			goqu.I("status").Eq("completed"),
+			goqu.I("slot_processed").Gte(startSlot),
+			goqu.I("slot_processed").Lt(endSlot),
 		).
-		Order(goqu.I("slot").Asc(), goqu.I("event_index").Asc())
+		Order(goqu.I("slot_processed").Asc(), goqu.I("index_processed").Asc())
 	sql, args, err := q.Prepared(true).ToSQL()
 	if err != nil {
 		return nil, fmt.Errorf("error fetching electra deposits for epoch %v: %w", epoch, err)
@@ -1910,16 +1908,17 @@ func ElectraGetProcessedConsolidations(epoch uint64) ([]constypes.ElectraConsoli
 	endSlot := (epoch+1)*utils.Config.ClConfig.SlotsPerEpoch - 1
 	var consolidations []constypes.ElectraConsolidation
 	q := goqu.Dialect("postgres").Select(
-		goqu.L("data->>'amount'").As("amount"),
-		goqu.L("data->>'source_index'").As("source_index"),
-		goqu.L("data->>'target_index'").As("target_index"),
+		goqu.L("amount_consolidated").As("amount"),
+		goqu.L("source_pubkey").As("source_pubkey"),
+		goqu.L("target_pubkey").As("target_pubkey"),
 	).
-		From("consensus_layer_events").
+		From("blocks_consolidation_requests_v2").
 		Where(
-			goqu.I("event_name").Eq("ConsolidationProcessedEvent"),
-			goqu.I("slot").Gte(startSlot),
-			goqu.I("slot").Lt(endSlot),
-		)
+			goqu.I("status").Eq("completed"),
+			goqu.I("slot_processed").Gte(startSlot),
+			goqu.I("slot_processed").Lt(endSlot),
+		).
+		Order(goqu.I("slot_processed").Asc(), goqu.I("index_processed").Asc())
 	sql, args, err := q.Prepared(true).ToSQL()
 	if err != nil {
 		return nil, fmt.Errorf("error fetching electra consolidations for epoch %v: %w", epoch, err)
@@ -1936,15 +1935,16 @@ func ElectraGetRemovedExcessBalanceEvents(epoch uint64) ([]constypes.ElectraExce
 	endSlot := (epoch+1)*utils.Config.ClConfig.SlotsPerEpoch - 1
 	var excessBalanceEvents []constypes.ElectraExcessBalance
 	q := goqu.Dialect("postgres").Select(
-		goqu.L("data->>'validator_index'").As("validator_index"),
-		goqu.L("data->>'amount'").As("amount"),
+		goqu.L("validator_pubkey").As("validator_pubkey"),
+		goqu.L("amount").As("amount"),
 	).
-		From("consensus_layer_events").
+		From("blocks_removed_excess_balance_events").
 		Where(
-			goqu.I("event_name").Eq("RemovedExcessBalanceEvent"),
-			goqu.I("slot").Gte(startSlot),
-			goqu.I("slot").Lt(endSlot),
-		)
+			goqu.I("status").Eq("completed"),
+			goqu.I("slot_processed").Gte(startSlot),
+			goqu.I("slot_processed").Lt(endSlot),
+		).
+		Order(goqu.I("slot_processed").Asc(), goqu.I("index_processed").Asc())
 	sql, args, err := q.Prepared(true).ToSQL()
 	if err != nil {
 		return nil, fmt.Errorf("error fetching electra excess balance events for epoch %v: %w", epoch, err)

backend/pkg/exporter/modules/dashboard_data_process.go

@@ -573,12 +573,20 @@ func (d *dashboardData) processElectraConsolidations(data *MultiEpochData, tar *
 				return nil
 			}
 			for _, consolidation := range data.epochBasedData.electraConsolidations[epoch] {
-				(*tar)[tI].ConsolidationsIncomingAmount[uint64(tO)+consolidation.TargetValidatorIndex] += int64(consolidation.Amount)
-				(*tar)[tI].ConsolidationsIncomingCount[uint64(tO)+consolidation.TargetValidatorIndex]++
-				(*tar)[tI].ConsolidationsOutgoingAmount[uint64(tO)+consolidation.SourceValidatorIndex] += int64(consolidation.Amount)
-				(*tar)[tI].ConsolidationsOutgoingCount[uint64(tO)+consolidation.SourceValidatorIndex]++
+				targetIndex, ok := data.validatorBasedData.validatorIndices[string(consolidation.TargetPubkey)]
+				if !ok {
+					return fmt.Errorf("target validator %s not found in validator indices map", string(consolidation.TargetPubkey))
+				}
+				sourceIndex, ok := data.validatorBasedData.validatorIndices[string(consolidation.SourcePubkey)]
+				if !ok {
+					return fmt.Errorf("source validator %s not found in validator indices map", string(consolidation.SourcePubkey))
+				}
+				(*tar)[tI].ConsolidationsIncomingAmount[uint64(tO)+targetIndex] += int64(consolidation.Amount)
+				(*tar)[tI].ConsolidationsIncomingCount[uint64(tO)+targetIndex]++
+				(*tar)[tI].ConsolidationsOutgoingAmount[uint64(tO)+sourceIndex] += int64(consolidation.Amount)
+				(*tar)[tI].ConsolidationsOutgoingCount[uint64(tO)+sourceIndex]++
 				// source => target
-				d.log.Tracef("processed electra consolidation: %d => %d %d GWEI in epoch %d", consolidation.SourceValidatorIndex, consolidation.TargetValidatorIndex, consolidation.Amount, epoch)
+				d.log.Tracef("processed electra consolidation: %d => %d %d GWEI in epoch %d", consolidation.SourcePubkey, consolidation.TargetPubkey, consolidation.Amount, epoch)
 			}
 			return nil
 		})
@@ -603,9 +611,13 @@ func (d *dashboardData) processElectraRemovedExcessBalanceEvents(data *MultiEpoc
 				return nil
 			}
 			for _, event := range data.epochBasedData.electraRemovedExcessBalanceEvents[epoch] {
-				(*tar)[tI].WithdrawalsAmount[uint64(tO)+event.ValidatorIndex] += int64(event.Amount)
-				(*tar)[tI].WithdrawalsCount[uint64(tO)+event.ValidatorIndex]++
-				d.log.Tracef("processed electra removed excess balance event: %d %d GWEI in epoch %d", event.ValidatorIndex, event.Amount, epoch)
+				validatorIndex, ok := data.validatorBasedData.validatorIndices[string(event.ValidatorPubkey)]
+				if !ok {
+					return fmt.Errorf("validator %s not found in validator indices map", string(event.ValidatorPubkey))
+				}
+				(*tar)[tI].WithdrawalsAmount[uint64(tO)+validatorIndex] += int64(event.Amount)
+				(*tar)[tI].WithdrawalsCount[uint64(tO)+validatorIndex]++
+				d.log.Tracef("processed electra removed excess balance event: %d %d GWEI in epoch %d", event.ValidatorPubkey, event.Amount, epoch)
 			}
 			return nil
 		})